feat(durability): productionize DBOS — entrypoint wiring, resume proof, validated procedure (#154)

backend/agents/document.py

@@ -2,7 +2,8 @@
 
 Coordinates classification, summary, concept extraction, and (when
 applicable) syllabus extraction, then merges results into the user's
-course graph by calling `apply_concepts_to_graph` directly.
+course graph via `_step_apply_graph`, a durable step that calls
+`apply_concepts_to_graph`.
 
 There is no orchestrator agent here: the graph merge is a deterministic
 function call with concept names already produced by the workers, so
@@ -13,7 +14,8 @@
  asyncio.gather.
 - Classification runs first because it gates whether syllabus
  extraction runs at all.
-- The graph update is a direct async function call after workers complete.
+- The graph update runs as a durable step (`_step_apply_graph`) after
+ workers complete, so a DBOS resume does not re-run the merge.
 
 Failure contract (ADR 0024 — this pipeline is the ONLY upload pipeline;
 ADR 0001's legacy fallback was retired in #151b):
@@ -28,12 +30,19 @@
  from bare exceptions in logs.
 
 Internal API: the `_step_*` functions defined below are wrapped with
-@durable_step and are meant to be called ONLY from `_run_workers`,
-which is itself reached only via `process_document` (the
-@durable_workflow). Calling a `_step_*` outside the workflow is
-undefined behavior under DBOS — depending on version, it may no-op
-silently, raise, or warn. Don't import them from routes or other
-modules.
+@durable_step and are meant to be called ONLY from within the
+@durable_workflow-decorated `process_document` — either directly
+(`_step_apply_graph`) or via `_run_workers` (the other four). For the
+pinned dbos==2.28.0, calling a `_step_*` outside a workflow context is
+NOT undefined: `dbos/_core.py::decorate_step`'s wrapper checks
+`ctx.is_workflow()` and, when there's no ambient workflow context, falls
+straight through to `return func(*args, **kwargs)` — the plain function
+runs for real, synchronously, with no DBOS registry lookup and no error
+(verified at `dbos/_core.py:2126-2152`). That's well-defined but still
+wrong to rely on here: the call would execute but get NONE of DBOS's
+checkpoint/resume behavior, silently losing durability for that call.
+Don't import them from routes or other modules; call only from
+`_run_workers`/`process_document`.
 """
 
 from __future__ import annotations
@@ -119,6 +128,34 @@ async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
  return result.output
 
 
+@durable_step
+async def _step_apply_graph(
+ user_id: str, course_id: str | None, concept_names: list[str],
+) -> int:
+ """Merge extracted concepts into the graph as a durable step, so a
+ DBOS resume does not re-run the merge (it's the pipeline's one
+ real-Supabase side effect besides persistence, which happens outside
+ process_document entirely — see routes/documents.py).
+
+ Calls `apply_concepts_to_graph` by its bare (module-global) name
+ rather than binding it to a local/default-arg at decoration time —
+ that matters because a plain global reference inside a function body
+ is looked up fresh from the function's `__globals__` (this module's
+ namespace) on EVERY call, not captured once when the function is
+ defined. That is what lets tests/test_dbos_resume.py's
+ `document_module.apply_concepts_to_graph = AsyncMock(...)`
+ monkeypatch take effect here: it reassigns the SAME name in this
+ module's namespace that this call resolves at call time. (A default
+ argument like `apply_fn=apply_concepts_to_graph` would instead freeze
+ in the ORIGINAL function object at decoration time and silently not
+ observe the monkeypatch.) routes/documents.py's streaming route
+ imports and calls the same `apply_concepts_to_graph` independently,
+ inline and non-durable per ADR 0011's streaming-route asymmetry —
+ a separate binding this monkeypatch doesn't touch, and doesn't need to.
+ """
+ return await apply_concepts_to_graph(user_id, course_id, concept_names)
+
+
 async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
  """Run classification first, then fan out the other workers in parallel.
 
@@ -152,22 +189,22 @@ async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
 
 @durable_workflow
 async def process_document(text: str, deps: SaplingDeps) -> DocumentProcessingResult:
- """Run workers in parallel, then merge concepts into the graph directly.
+ """Run workers in parallel, then merge concepts into the graph via the
+ durable `_step_apply_graph`.
 
  DocumentProcessingResult is composed deterministically here from worker
- outputs. The graph merge is a plain async function call — no orchestrator
- agent — because it has no decisions to make beyond passing the
- already-extracted concept names through.
+ outputs. The graph merge has no orchestrator agent — no decisions to
+ make beyond passing the already-extracted concept names through — but
+ IS wrapped as a checkpointed step (`_step_apply_graph`) so a DBOS
+ resume of this workflow doesn't repeat the merge.
 
  Wrapped in `@durable_workflow` from services.durable: a no-op when
  DBOS_ENABLED is unset (the default), a real DBOS workflow when the
  operator opts in. See ADR 0011 for the activation procedure.
  """
  workers = await _run_workers(text, deps)
  concept_names = [c.name for c in workers.concepts.concepts]
- merged = await apply_concepts_to_graph(
- deps.user_id, deps.course_id, concept_names,
- )
+ merged = await _step_apply_graph(deps.user_id, deps.course_id, concept_names)
  return DocumentProcessingResult(
  classification=workers.classification,
  summary=workers.summary,

backend/main.py

@@ -29,6 +29,7 @@
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
 from services.storage_service import ALLOWED_CONTENT_TYPES, ensure_bucket_exists
+from services.durable import init_dbos, shutdown_dbos
 
 try:
  from recost.frameworks.fastapi import RecostMiddleware
@@ -85,10 +86,15 @@ async def _lifespan(_app: FastAPI):
  # usage + event rows flush off the request path.
  from services import events_service
  events_service.start_worker()
+ # ADR 0011 / #154: construct + launch DBOS when DBOS_ENABLED=true; no-op
+ # passthrough otherwise. Fails loudly (raises) if the operator opted in
+ # and launch fails — see services/durable.py::init_dbos.
+ init_dbos()
  yield
  # Stop the drain thread and flush anything still queued so the last batch
  # of usage rows isn't lost on shutdown.
  events_service.shutdown()
+ shutdown_dbos()
 
 
 def _drop_request_arguments(_request, _attributes):

backend/requirements-durable.txt

@@ -0,0 +1,11 @@
+# Opt-in durable-execution extra (ADR 0011 / #154).
+#
+# Installs the `dbos` package so backend/services/durable.py's @workflow /
+# @step decorators go real when an operator sets DBOS_ENABLED=true (with
+# DBOS_DATABASE_URL pointing at a Postgres instance). Never in
+# requirements.txt or requirements.lock: durability is opt-in and the
+# default (prod-default and hermetic-test) path must keep working with
+# `dbos` NOT installed — see backend/tests/test_durable_shim.py.
+#
+# Install with: pip install -r requirements-durable.txt
+dbos>=2.28,<3

backend/routes/documents.py

@@ -41,6 +41,7 @@
 from services.achievement_service import check_achievements
 from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
+from services.durable import workflow_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
 from agents.classifier import classifier_agent
@@ -606,8 +607,13 @@ async def upload_document_sync(
  # both branches surface a retry-friendly 502: guardrail trips (budget /
  # degenerate output) log at WARNING, anything else is a bug and logs the
  # full exception.
+ # Scope the DBOS workflow id to user_id + request_id, not request_id
+ # alone: X-Request-ID is client-supplied, so an unscoped id would let
+ # one user's replay attach to another user's in-flight/completed
+ # workflow (state poisoning). No-op (nullcontext) when DBOS is off.
  try:
- result: DocumentProcessingResult = await process_document(extracted_text, deps)
+ with workflow_id(f"doc:{user_id}:{request_id}"):
+ result: DocumentProcessingResult = await process_document(extracted_text, deps)
  except (UsageLimitExceeded, UnexpectedModelBehavior) as e:
  logger.warning(
  "Agent guardrails tripped for '%s'; returning 502",

backend/agents/document.py

@@ -2,7 +2,8 @@
 
 Coordinates classification, summary, concept extraction, and (when
 applicable) syllabus extraction, then merges results into the user's
-course graph by calling `apply_concepts_to_graph` directly.
+course graph via `_step_apply_graph`, a durable step that calls
+`apply_concepts_to_graph`.
 
 There is no orchestrator agent here: the graph merge is a deterministic
 function call with concept names already produced by the workers, so
@@ -13,7 +14,8 @@
  asyncio.gather.
 - Classification runs first because it gates whether syllabus
  extraction runs at all.
-- The graph update is a direct async function call after workers complete.
+- The graph update runs as a durable step (`_step_apply_graph`) after
+ workers complete, so a DBOS resume does not re-run the merge.
 
 Failure contract (ADR 0024 — this pipeline is the ONLY upload pipeline;
 ADR 0001's legacy fallback was retired in #151b):
@@ -28,12 +30,19 @@
  from bare exceptions in logs.
 
 Internal API: the `_step_*` functions defined below are wrapped with
-@durable_step and are meant to be called ONLY from `_run_workers`,
-which is itself reached only via `process_document` (the
-@durable_workflow). Calling a `_step_*` outside the workflow is
-undefined behavior under DBOS — depending on version, it may no-op
-silently, raise, or warn. Don't import them from routes or other
-modules.
+@durable_step and are meant to be called ONLY from within the
+@durable_workflow-decorated `process_document` — either directly
+(`_step_apply_graph`) or via `_run_workers` (the other four). For the
+pinned dbos==2.28.0, calling a `_step_*` outside a workflow context is
+NOT undefined: `dbos/_core.py::decorate_step`'s wrapper checks
+`ctx.is_workflow()` and, when there's no ambient workflow context, falls
+straight through to `return func(*args, **kwargs)` — the plain function
+runs for real, synchronously, with no DBOS registry lookup and no error
+(verified at `dbos/_core.py:2126-2152`). That's well-defined but still
+wrong to rely on here: the call would execute but get NONE of DBOS's
+checkpoint/resume behavior, silently losing durability for that call.
+Don't import them from routes or other modules; call only from
+`_run_workers`/`process_document`.
 """
 
 from __future__ import annotations
@@ -119,6 +128,34 @@ async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
  return result.output
 
 
+@durable_step
+async def _step_apply_graph(
+ user_id: str, course_id: str | None, concept_names: list[str],
+) -> int:
+ """Merge extracted concepts into the graph as a durable step, so a
+ DBOS resume does not re-run the merge (it's the pipeline's one
+ real-Supabase side effect besides persistence, which happens outside
+ process_document entirely — see routes/documents.py).
+
+ Calls `apply_concepts_to_graph` by its bare (module-global) name
+ rather than binding it to a local/default-arg at decoration time —
+ that matters because a plain global reference inside a function body
+ is looked up fresh from the function's `__globals__` (this module's
+ namespace) on EVERY call, not captured once when the function is
+ defined. That is what lets tests/test_dbos_resume.py's
+ `document_module.apply_concepts_to_graph = AsyncMock(...)`
+ monkeypatch take effect here: it reassigns the SAME name in this
+ module's namespace that this call resolves at call time. (A default
+ argument like `apply_fn=apply_concepts_to_graph` would instead freeze
+ in the ORIGINAL function object at decoration time and silently not
+ observe the monkeypatch.) routes/documents.py's streaming route
+ imports and calls the same `apply_concepts_to_graph` independently,
+ inline and non-durable per ADR 0011's streaming-route asymmetry —
+ a separate binding this monkeypatch doesn't touch, and doesn't need to.
+ """
+ return await apply_concepts_to_graph(user_id, course_id, concept_names)
+
+
 async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
  """Run classification first, then fan out the other workers in parallel.
 
@@ -152,22 +189,22 @@ async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
 
 @durable_workflow
 async def process_document(text: str, deps: SaplingDeps) -> DocumentProcessingResult:
- """Run workers in parallel, then merge concepts into the graph directly.
+ """Run workers in parallel, then merge concepts into the graph via the
+ durable `_step_apply_graph`.
 
  DocumentProcessingResult is composed deterministically here from worker
- outputs. The graph merge is a plain async function call — no orchestrator
- agent — because it has no decisions to make beyond passing the
- already-extracted concept names through.
+ outputs. The graph merge has no orchestrator agent — no decisions to
+ make beyond passing the already-extracted concept names through — but
+ IS wrapped as a checkpointed step (`_step_apply_graph`) so a DBOS
+ resume of this workflow doesn't repeat the merge.
 
  Wrapped in `@durable_workflow` from services.durable: a no-op when
  DBOS_ENABLED is unset (the default), a real DBOS workflow when the
  operator opts in. See ADR 0011 for the activation procedure.
  """
  workers = await _run_workers(text, deps)
  concept_names = [c.name for c in workers.concepts.concepts]
- merged = await apply_concepts_to_graph(
- deps.user_id, deps.course_id, concept_names,
- )
+ merged = await _step_apply_graph(deps.user_id, deps.course_id, concept_names)
  return DocumentProcessingResult(
  classification=workers.classification,
  summary=workers.summary,

backend/main.py

@@ -29,6 +29,7 @@
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
 from services.storage_service import ALLOWED_CONTENT_TYPES, ensure_bucket_exists
+from services.durable import init_dbos, shutdown_dbos
 
 try:
  from recost.frameworks.fastapi import RecostMiddleware
@@ -85,10 +86,15 @@ async def _lifespan(_app: FastAPI):
  # usage + event rows flush off the request path.
  from services import events_service
  events_service.start_worker()
+ # ADR 0011 / #154: construct + launch DBOS when DBOS_ENABLED=true; no-op
+ # passthrough otherwise. Fails loudly (raises) if the operator opted in
+ # and launch fails — see services/durable.py::init_dbos.
+ init_dbos()
  yield
  # Stop the drain thread and flush anything still queued so the last batch
  # of usage rows isn't lost on shutdown.
  events_service.shutdown()
+ shutdown_dbos()
 
 
 def _drop_request_arguments(_request, _attributes):

backend/requirements-durable.txt

@@ -0,0 +1,11 @@
+# Opt-in durable-execution extra (ADR 0011 / #154).
+#
+# Installs the `dbos` package so backend/services/durable.py's @workflow /
+# @step decorators go real when an operator sets DBOS_ENABLED=true (with
+# DBOS_DATABASE_URL pointing at a Postgres instance). Never in
+# requirements.txt or requirements.lock: durability is opt-in and the
+# default (prod-default and hermetic-test) path must keep working with
+# `dbos` NOT installed — see backend/tests/test_durable_shim.py.
+#
+# Install with: pip install -r requirements-durable.txt
+dbos>=2.28,<3

backend/routes/documents.py

@@ -41,6 +41,7 @@
 from services.achievement_service import check_achievements
 from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
+from services.durable import workflow_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
 from agents.classifier import classifier_agent
@@ -606,8 +607,13 @@ async def upload_document_sync(
  # both branches surface a retry-friendly 502: guardrail trips (budget /
  # degenerate output) log at WARNING, anything else is a bug and logs the
  # full exception.
+ # Scope the DBOS workflow id to user_id + request_id, not request_id
+ # alone: X-Request-ID is client-supplied, so an unscoped id would let
+ # one user's replay attach to another user's in-flight/completed
+ # workflow (state poisoning). No-op (nullcontext) when DBOS is off.
  try:
- result: DocumentProcessingResult = await process_document(extracted_text, deps)
+ with workflow_id(f"doc:{user_id}:{request_id}"):
+ result: DocumentProcessingResult = await process_document(extracted_text, deps)
  except (UsageLimitExceeded, UnexpectedModelBehavior) as e:
  logger.warning(
  "Agent guardrails tripped for '%s'; returning 502",

backend/agents/document.py

@@ -2,7 +2,8 @@
 
 Coordinates classification, summary, concept extraction, and (when
 applicable) syllabus extraction, then merges results into the user's
-course graph by calling `apply_concepts_to_graph` directly.
+course graph via `_step_apply_graph`, a durable step that calls
+`apply_concepts_to_graph`.
 
 There is no orchestrator agent here: the graph merge is a deterministic
 function call with concept names already produced by the workers, so
@@ -13,7 +14,8 @@
  asyncio.gather.
 - Classification runs first because it gates whether syllabus
  extraction runs at all.
-- The graph update is a direct async function call after workers complete.
+- The graph update runs as a durable step (`_step_apply_graph`) after
+ workers complete, so a DBOS resume does not re-run the merge.
 
 Failure contract (ADR 0024 — this pipeline is the ONLY upload pipeline;
 ADR 0001's legacy fallback was retired in #151b):
@@ -28,12 +30,19 @@
  from bare exceptions in logs.
 
 Internal API: the `_step_*` functions defined below are wrapped with
-@durable_step and are meant to be called ONLY from `_run_workers`,
-which is itself reached only via `process_document` (the
-@durable_workflow). Calling a `_step_*` outside the workflow is
-undefined behavior under DBOS — depending on version, it may no-op
-silently, raise, or warn. Don't import them from routes or other
-modules.
+@durable_step and are meant to be called ONLY from within the
+@durable_workflow-decorated `process_document` — either directly
+(`_step_apply_graph`) or via `_run_workers` (the other four). For the
+pinned dbos==2.28.0, calling a `_step_*` outside a workflow context is
+NOT undefined: `dbos/_core.py::decorate_step`'s wrapper checks
+`ctx.is_workflow()` and, when there's no ambient workflow context, falls
+straight through to `return func(*args, **kwargs)` — the plain function
+runs for real, synchronously, with no DBOS registry lookup and no error
+(verified at `dbos/_core.py:2126-2152`). That's well-defined but still
+wrong to rely on here: the call would execute but get NONE of DBOS's
+checkpoint/resume behavior, silently losing durability for that call.
+Don't import them from routes or other modules; call only from
+`_run_workers`/`process_document`.
 """
 
 from __future__ import annotations
@@ -119,6 +128,34 @@ async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
  return result.output
 
 
+@durable_step
+async def _step_apply_graph(
+ user_id: str, course_id: str | None, concept_names: list[str],
+) -> int:
+ """Merge extracted concepts into the graph as a durable step, so a
+ DBOS resume does not re-run the merge (it's the pipeline's one
+ real-Supabase side effect besides persistence, which happens outside
+ process_document entirely — see routes/documents.py).
+
+ Calls `apply_concepts_to_graph` by its bare (module-global) name
+ rather than binding it to a local/default-arg at decoration time —
+ that matters because a plain global reference inside a function body
+ is looked up fresh from the function's `__globals__` (this module's
+ namespace) on EVERY call, not captured once when the function is
+ defined. That is what lets tests/test_dbos_resume.py's
+ `document_module.apply_concepts_to_graph = AsyncMock(...)`
+ monkeypatch take effect here: it reassigns the SAME name in this
+ module's namespace that this call resolves at call time. (A default
+ argument like `apply_fn=apply_concepts_to_graph` would instead freeze
+ in the ORIGINAL function object at decoration time and silently not
+ observe the monkeypatch.) routes/documents.py's streaming route
+ imports and calls the same `apply_concepts_to_graph` independently,
+ inline and non-durable per ADR 0011's streaming-route asymmetry —
+ a separate binding this monkeypatch doesn't touch, and doesn't need to.
+ """
+ return await apply_concepts_to_graph(user_id, course_id, concept_names)
+
+
 async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
  """Run classification first, then fan out the other workers in parallel.
 
@@ -152,22 +189,22 @@ async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
 
 @durable_workflow
 async def process_document(text: str, deps: SaplingDeps) -> DocumentProcessingResult:
- """Run workers in parallel, then merge concepts into the graph directly.
+ """Run workers in parallel, then merge concepts into the graph via the
+ durable `_step_apply_graph`.
 
  DocumentProcessingResult is composed deterministically here from worker
- outputs. The graph merge is a plain async function call — no orchestrator
- agent — because it has no decisions to make beyond passing the
- already-extracted concept names through.
+ outputs. The graph merge has no orchestrator agent — no decisions to
+ make beyond passing the already-extracted concept names through — but
+ IS wrapped as a checkpointed step (`_step_apply_graph`) so a DBOS
+ resume of this workflow doesn't repeat the merge.
 
  Wrapped in `@durable_workflow` from services.durable: a no-op when
  DBOS_ENABLED is unset (the default), a real DBOS workflow when the
  operator opts in. See ADR 0011 for the activation procedure.
  """
  workers = await _run_workers(text, deps)
  concept_names = [c.name for c in workers.concepts.concepts]
- merged = await apply_concepts_to_graph(
- deps.user_id, deps.course_id, concept_names,
- )
+ merged = await _step_apply_graph(deps.user_id, deps.course_id, concept_names)
  return DocumentProcessingResult(
  classification=workers.classification,
  summary=workers.summary,

backend/main.py

@@ -29,6 +29,7 @@
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
 from services.storage_service import ALLOWED_CONTENT_TYPES, ensure_bucket_exists
+from services.durable import init_dbos, shutdown_dbos
 
 try:
  from recost.frameworks.fastapi import RecostMiddleware
@@ -85,10 +86,15 @@ async def _lifespan(_app: FastAPI):
  # usage + event rows flush off the request path.
  from services import events_service
  events_service.start_worker()
+ # ADR 0011 / #154: construct + launch DBOS when DBOS_ENABLED=true; no-op
+ # passthrough otherwise. Fails loudly (raises) if the operator opted in
+ # and launch fails — see services/durable.py::init_dbos.
+ init_dbos()
  yield
  # Stop the drain thread and flush anything still queued so the last batch
  # of usage rows isn't lost on shutdown.
  events_service.shutdown()
+ shutdown_dbos()
 
 
 def _drop_request_arguments(_request, _attributes):

backend/requirements-durable.txt

@@ -0,0 +1,11 @@
+# Opt-in durable-execution extra (ADR 0011 / #154).
+#
+# Installs the `dbos` package so backend/services/durable.py's @workflow /
+# @step decorators go real when an operator sets DBOS_ENABLED=true (with
+# DBOS_DATABASE_URL pointing at a Postgres instance). Never in
+# requirements.txt or requirements.lock: durability is opt-in and the
+# default (prod-default and hermetic-test) path must keep working with
+# `dbos` NOT installed — see backend/tests/test_durable_shim.py.
+#
+# Install with: pip install -r requirements-durable.txt
+dbos>=2.28,<3

backend/routes/documents.py

@@ -41,6 +41,7 @@
 from services.achievement_service import check_achievements
 from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
+from services.durable import workflow_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
 from agents.classifier import classifier_agent
@@ -606,8 +607,13 @@ async def upload_document_sync(
  # both branches surface a retry-friendly 502: guardrail trips (budget /
  # degenerate output) log at WARNING, anything else is a bug and logs the
  # full exception.
+ # Scope the DBOS workflow id to user_id + request_id, not request_id
+ # alone: X-Request-ID is client-supplied, so an unscoped id would let
+ # one user's replay attach to another user's in-flight/completed
+ # workflow (state poisoning). No-op (nullcontext) when DBOS is off.
  try:
- result: DocumentProcessingResult = await process_document(extracted_text, deps)
+ with workflow_id(f"doc:{user_id}:{request_id}"):
+ result: DocumentProcessingResult = await process_document(extracted_text, deps)
  except (UsageLimitExceeded, UnexpectedModelBehavior) as e:
  logger.warning(
  "Agent guardrails tripped for '%s'; returning 502",

backend/agents/document.py

@@ -2,7 +2,8 @@
 
 Coordinates classification, summary, concept extraction, and (when
 applicable) syllabus extraction, then merges results into the user's
-course graph by calling `apply_concepts_to_graph` directly.
+course graph via `_step_apply_graph`, a durable step that calls
+`apply_concepts_to_graph`.
 
 There is no orchestrator agent here: the graph merge is a deterministic
 function call with concept names already produced by the workers, so
@@ -13,7 +14,8 @@
  asyncio.gather.
 - Classification runs first because it gates whether syllabus
  extraction runs at all.
-- The graph update is a direct async function call after workers complete.
+- The graph update runs as a durable step (`_step_apply_graph`) after
+ workers complete, so a DBOS resume does not re-run the merge.
 
 Failure contract (ADR 0024 — this pipeline is the ONLY upload pipeline;
 ADR 0001's legacy fallback was retired in #151b):
@@ -28,12 +30,19 @@
  from bare exceptions in logs.
 
 Internal API: the `_step_*` functions defined below are wrapped with
-@durable_step and are meant to be called ONLY from `_run_workers`,
-which is itself reached only via `process_document` (the
-@durable_workflow). Calling a `_step_*` outside the workflow is
-undefined behavior under DBOS — depending on version, it may no-op
-silently, raise, or warn. Don't import them from routes or other
-modules.
+@durable_step and are meant to be called ONLY from within the
+@durable_workflow-decorated `process_document` — either directly
+(`_step_apply_graph`) or via `_run_workers` (the other four). For the
+pinned dbos==2.28.0, calling a `_step_*` outside a workflow context is
+NOT undefined: `dbos/_core.py::decorate_step`'s wrapper checks
+`ctx.is_workflow()` and, when there's no ambient workflow context, falls
+straight through to `return func(*args, **kwargs)` — the plain function
+runs for real, synchronously, with no DBOS registry lookup and no error
+(verified at `dbos/_core.py:2126-2152`). That's well-defined but still
+wrong to rely on here: the call would execute but get NONE of DBOS's
+checkpoint/resume behavior, silently losing durability for that call.
+Don't import them from routes or other modules; call only from
+`_run_workers`/`process_document`.
 """
 
 from __future__ import annotations
@@ -119,6 +128,34 @@ async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
  return result.output
 
 
+@durable_step
+async def _step_apply_graph(
+ user_id: str, course_id: str | None, concept_names: list[str],
+) -> int:
+ """Merge extracted concepts into the graph as a durable step, so a
+ DBOS resume does not re-run the merge (it's the pipeline's one
+ real-Supabase side effect besides persistence, which happens outside
+ process_document entirely — see routes/documents.py).
+
+ Calls `apply_concepts_to_graph` by its bare (module-global) name
+ rather than binding it to a local/default-arg at decoration time —
+ that matters because a plain global reference inside a function body
+ is looked up fresh from the function's `__globals__` (this module's
+ namespace) on EVERY call, not captured once when the function is
+ defined. That is what lets tests/test_dbos_resume.py's
+ `document_module.apply_concepts_to_graph = AsyncMock(...)`
+ monkeypatch take effect here: it reassigns the SAME name in this
+ module's namespace that this call resolves at call time. (A default
+ argument like `apply_fn=apply_concepts_to_graph` would instead freeze
+ in the ORIGINAL function object at decoration time and silently not
+ observe the monkeypatch.) routes/documents.py's streaming route
+ imports and calls the same `apply_concepts_to_graph` independently,
+ inline and non-durable per ADR 0011's streaming-route asymmetry —
+ a separate binding this monkeypatch doesn't touch, and doesn't need to.
+ """
+ return await apply_concepts_to_graph(user_id, course_id, concept_names)
+
+
 async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
  """Run classification first, then fan out the other workers in parallel.
 
@@ -152,22 +189,22 @@ async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
 
 @durable_workflow
 async def process_document(text: str, deps: SaplingDeps) -> DocumentProcessingResult:
- """Run workers in parallel, then merge concepts into the graph directly.
+ """Run workers in parallel, then merge concepts into the graph via the
+ durable `_step_apply_graph`.
 
  DocumentProcessingResult is composed deterministically here from worker
- outputs. The graph merge is a plain async function call — no orchestrator
- agent — because it has no decisions to make beyond passing the
- already-extracted concept names through.
+ outputs. The graph merge has no orchestrator agent — no decisions to
+ make beyond passing the already-extracted concept names through — but
+ IS wrapped as a checkpointed step (`_step_apply_graph`) so a DBOS
+ resume of this workflow doesn't repeat the merge.
 
  Wrapped in `@durable_workflow` from services.durable: a no-op when
  DBOS_ENABLED is unset (the default), a real DBOS workflow when the
  operator opts in. See ADR 0011 for the activation procedure.
  """
  workers = await _run_workers(text, deps)
  concept_names = [c.name for c in workers.concepts.concepts]
- merged = await apply_concepts_to_graph(
- deps.user_id, deps.course_id, concept_names,
- )
+ merged = await _step_apply_graph(deps.user_id, deps.course_id, concept_names)
  return DocumentProcessingResult(
  classification=workers.classification,
  summary=workers.summary,

backend/main.py

@@ -29,6 +29,7 @@
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
 from services.storage_service import ALLOWED_CONTENT_TYPES, ensure_bucket_exists
+from services.durable import init_dbos, shutdown_dbos
 
 try:
  from recost.frameworks.fastapi import RecostMiddleware
@@ -85,10 +86,15 @@ async def _lifespan(_app: FastAPI):
  # usage + event rows flush off the request path.
  from services import events_service
  events_service.start_worker()
+ # ADR 0011 / #154: construct + launch DBOS when DBOS_ENABLED=true; no-op
+ # passthrough otherwise. Fails loudly (raises) if the operator opted in
+ # and launch fails — see services/durable.py::init_dbos.
+ init_dbos()
  yield
  # Stop the drain thread and flush anything still queued so the last batch
  # of usage rows isn't lost on shutdown.
  events_service.shutdown()
+ shutdown_dbos()
 
 
 def _drop_request_arguments(_request, _attributes):

backend/requirements-durable.txt

@@ -0,0 +1,11 @@
+# Opt-in durable-execution extra (ADR 0011 / #154).
+#
+# Installs the `dbos` package so backend/services/durable.py's @workflow /
+# @step decorators go real when an operator sets DBOS_ENABLED=true (with
+# DBOS_DATABASE_URL pointing at a Postgres instance). Never in
+# requirements.txt or requirements.lock: durability is opt-in and the
+# default (prod-default and hermetic-test) path must keep working with
+# `dbos` NOT installed — see backend/tests/test_durable_shim.py.
+#
+# Install with: pip install -r requirements-durable.txt
+dbos>=2.28,<3

backend/routes/documents.py

@@ -41,6 +41,7 @@
 from services.achievement_service import check_achievements
 from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
+from services.durable import workflow_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
 from agents.classifier import classifier_agent
@@ -606,8 +607,13 @@ async def upload_document_sync(
  # both branches surface a retry-friendly 502: guardrail trips (budget /
  # degenerate output) log at WARNING, anything else is a bug and logs the
  # full exception.
+ # Scope the DBOS workflow id to user_id + request_id, not request_id
+ # alone: X-Request-ID is client-supplied, so an unscoped id would let
+ # one user's replay attach to another user's in-flight/completed
+ # workflow (state poisoning). No-op (nullcontext) when DBOS is off.
  try:
- result: DocumentProcessingResult = await process_document(extracted_text, deps)
+ with workflow_id(f"doc:{user_id}:{request_id}"):
+ result: DocumentProcessingResult = await process_document(extracted_text, deps)
  except (UsageLimitExceeded, UnexpectedModelBehavior) as e:
  logger.warning(
  "Agent guardrails tripped for '%s'; returning 502",

backend/agents/document.py

@@ -2,7 +2,8 @@
 
 Coordinates classification, summary, concept extraction, and (when
 applicable) syllabus extraction, then merges results into the user's
-course graph by calling `apply_concepts_to_graph` directly.
+course graph via `_step_apply_graph`, a durable step that calls
+`apply_concepts_to_graph`.
 
 There is no orchestrator agent here: the graph merge is a deterministic
 function call with concept names already produced by the workers, so
@@ -13,7 +14,8 @@
  asyncio.gather.
 - Classification runs first because it gates whether syllabus
  extraction runs at all.
-- The graph update is a direct async function call after workers complete.
+- The graph update runs as a durable step (`_step_apply_graph`) after
+ workers complete, so a DBOS resume does not re-run the merge.
 
 Failure contract (ADR 0024 — this pipeline is the ONLY upload pipeline;
 ADR 0001's legacy fallback was retired in #151b):
@@ -28,12 +30,19 @@
  from bare exceptions in logs.
 
 Internal API: the `_step_*` functions defined below are wrapped with
-@durable_step and are meant to be called ONLY from `_run_workers`,
-which is itself reached only via `process_document` (the
-@durable_workflow). Calling a `_step_*` outside the workflow is
-undefined behavior under DBOS — depending on version, it may no-op
-silently, raise, or warn. Don't import them from routes or other
-modules.
+@durable_step and are meant to be called ONLY from within the
+@durable_workflow-decorated `process_document` — either directly
+(`_step_apply_graph`) or via `_run_workers` (the other four). For the
+pinned dbos==2.28.0, calling a `_step_*` outside a workflow context is
+NOT undefined: `dbos/_core.py::decorate_step`'s wrapper checks
+`ctx.is_workflow()` and, when there's no ambient workflow context, falls
+straight through to `return func(*args, **kwargs)` — the plain function
+runs for real, synchronously, with no DBOS registry lookup and no error
+(verified at `dbos/_core.py:2126-2152`). That's well-defined but still
+wrong to rely on here: the call would execute but get NONE of DBOS's
+checkpoint/resume behavior, silently losing durability for that call.
+Don't import them from routes or other modules; call only from
+`_run_workers`/`process_document`.
 """
 
 from __future__ import annotations
@@ -119,6 +128,34 @@ async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
  return result.output
 
 
+@durable_step
+async def _step_apply_graph(
+ user_id: str, course_id: str | None, concept_names: list[str],
+) -> int:
+ """Merge extracted concepts into the graph as a durable step, so a
+ DBOS resume does not re-run the merge (it's the pipeline's one
+ real-Supabase side effect besides persistence, which happens outside
+ process_document entirely — see routes/documents.py).
+
+ Calls `apply_concepts_to_graph` by its bare (module-global) name
+ rather than binding it to a local/default-arg at decoration time —
+ that matters because a plain global reference inside a function body
+ is looked up fresh from the function's `__globals__` (this module's
+ namespace) on EVERY call, not captured once when the function is
+ defined. That is what lets tests/test_dbos_resume.py's
+ `document_module.apply_concepts_to_graph = AsyncMock(...)`
+ monkeypatch take effect here: it reassigns the SAME name in this
+ module's namespace that this call resolves at call time. (A default
+ argument like `apply_fn=apply_concepts_to_graph` would instead freeze
+ in the ORIGINAL function object at decoration time and silently not
+ observe the monkeypatch.) routes/documents.py's streaming route
+ imports and calls the same `apply_concepts_to_graph` independently,
+ inline and non-durable per ADR 0011's streaming-route asymmetry —
+ a separate binding this monkeypatch doesn't touch, and doesn't need to.
+ """
+ return await apply_concepts_to_graph(user_id, course_id, concept_names)
+
+
 async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
  """Run classification first, then fan out the other workers in parallel.
 
@@ -152,22 +189,22 @@ async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
 
 @durable_workflow
 async def process_document(text: str, deps: SaplingDeps) -> DocumentProcessingResult:
- """Run workers in parallel, then merge concepts into the graph directly.
+ """Run workers in parallel, then merge concepts into the graph via the
+ durable `_step_apply_graph`.
 
  DocumentProcessingResult is composed deterministically here from worker
- outputs. The graph merge is a plain async function call — no orchestrator
- agent — because it has no decisions to make beyond passing the
- already-extracted concept names through.
+ outputs. The graph merge has no orchestrator agent — no decisions to
+ make beyond passing the already-extracted concept names through — but
+ IS wrapped as a checkpointed step (`_step_apply_graph`) so a DBOS
+ resume of this workflow doesn't repeat the merge.
 
  Wrapped in `@durable_workflow` from services.durable: a no-op when
  DBOS_ENABLED is unset (the default), a real DBOS workflow when the
  operator opts in. See ADR 0011 for the activation procedure.
  """
  workers = await _run_workers(text, deps)
  concept_names = [c.name for c in workers.concepts.concepts]
- merged = await apply_concepts_to_graph(
- deps.user_id, deps.course_id, concept_names,
- )
+ merged = await _step_apply_graph(deps.user_id, deps.course_id, concept_names)
  return DocumentProcessingResult(
  classification=workers.classification,
  summary=workers.summary,

backend/main.py

@@ -29,6 +29,7 @@
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
 from services.storage_service import ALLOWED_CONTENT_TYPES, ensure_bucket_exists
+from services.durable import init_dbos, shutdown_dbos
 
 try:
  from recost.frameworks.fastapi import RecostMiddleware
@@ -85,10 +86,15 @@ async def _lifespan(_app: FastAPI):
  # usage + event rows flush off the request path.
  from services import events_service
  events_service.start_worker()
+ # ADR 0011 / #154: construct + launch DBOS when DBOS_ENABLED=true; no-op
+ # passthrough otherwise. Fails loudly (raises) if the operator opted in
+ # and launch fails — see services/durable.py::init_dbos.
+ init_dbos()
  yield
  # Stop the drain thread and flush anything still queued so the last batch
  # of usage rows isn't lost on shutdown.
  events_service.shutdown()
+ shutdown_dbos()
 
 
 def _drop_request_arguments(_request, _attributes):

backend/requirements-durable.txt

@@ -0,0 +1,11 @@
+# Opt-in durable-execution extra (ADR 0011 / #154).
+#
+# Installs the `dbos` package so backend/services/durable.py's @workflow /
+# @step decorators go real when an operator sets DBOS_ENABLED=true (with
+# DBOS_DATABASE_URL pointing at a Postgres instance). Never in
+# requirements.txt or requirements.lock: durability is opt-in and the
+# default (prod-default and hermetic-test) path must keep working with
+# `dbos` NOT installed — see backend/tests/test_durable_shim.py.
+#
+# Install with: pip install -r requirements-durable.txt
+dbos>=2.28,<3

backend/routes/documents.py

@@ -41,6 +41,7 @@
 from services.achievement_service import check_achievements
 from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
+from services.durable import workflow_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
 from agents.classifier import classifier_agent
@@ -606,8 +607,13 @@ async def upload_document_sync(
  # both branches surface a retry-friendly 502: guardrail trips (budget /
  # degenerate output) log at WARNING, anything else is a bug and logs the
  # full exception.
+ # Scope the DBOS workflow id to user_id + request_id, not request_id
+ # alone: X-Request-ID is client-supplied, so an unscoped id would let
+ # one user's replay attach to another user's in-flight/completed
+ # workflow (state poisoning). No-op (nullcontext) when DBOS is off.
  try:
- result: DocumentProcessingResult = await process_document(extracted_text, deps)
+ with workflow_id(f"doc:{user_id}:{request_id}"):
+ result: DocumentProcessingResult = await process_document(extracted_text, deps)
  except (UsageLimitExceeded, UnexpectedModelBehavior) as e:
  logger.warning(
  "Agent guardrails tripped for '%s'; returning 502",

backend/agents/document.py

@@ -2,7 +2,8 @@
 
 Coordinates classification, summary, concept extraction, and (when
 applicable) syllabus extraction, then merges results into the user's
-course graph by calling `apply_concepts_to_graph` directly.
+course graph via `_step_apply_graph`, a durable step that calls
+`apply_concepts_to_graph`.
 
 There is no orchestrator agent here: the graph merge is a deterministic
 function call with concept names already produced by the workers, so
@@ -13,7 +14,8 @@
  asyncio.gather.
 - Classification runs first because it gates whether syllabus
  extraction runs at all.
-- The graph update is a direct async function call after workers complete.
+- The graph update runs as a durable step (`_step_apply_graph`) after
+ workers complete, so a DBOS resume does not re-run the merge.
 
 Failure contract (ADR 0024 — this pipeline is the ONLY upload pipeline;
 ADR 0001's legacy fallback was retired in #151b):
@@ -28,12 +30,19 @@
  from bare exceptions in logs.
 
 Internal API: the `_step_*` functions defined below are wrapped with
-@durable_step and are meant to be called ONLY from `_run_workers`,
-which is itself reached only via `process_document` (the
-@durable_workflow). Calling a `_step_*` outside the workflow is
-undefined behavior under DBOS — depending on version, it may no-op
-silently, raise, or warn. Don't import them from routes or other
-modules.
+@durable_step and are meant to be called ONLY from within the
+@durable_workflow-decorated `process_document` — either directly
+(`_step_apply_graph`) or via `_run_workers` (the other four). For the
+pinned dbos==2.28.0, calling a `_step_*` outside a workflow context is
+NOT undefined: `dbos/_core.py::decorate_step`'s wrapper checks
+`ctx.is_workflow()` and, when there's no ambient workflow context, falls
+straight through to `return func(*args, **kwargs)` — the plain function
+runs for real, synchronously, with no DBOS registry lookup and no error
+(verified at `dbos/_core.py:2126-2152`). That's well-defined but still
+wrong to rely on here: the call would execute but get NONE of DBOS's
+checkpoint/resume behavior, silently losing durability for that call.
+Don't import them from routes or other modules; call only from
+`_run_workers`/`process_document`.
 """
 
 from __future__ import annotations
@@ -119,6 +128,34 @@ async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
  return result.output
 
 
+@durable_step
+async def _step_apply_graph(
+ user_id: str, course_id: str | None, concept_names: list[str],
+) -> int:
+ """Merge extracted concepts into the graph as a durable step, so a
+ DBOS resume does not re-run the merge (it's the pipeline's one
+ real-Supabase side effect besides persistence, which happens outside
+ process_document entirely — see routes/documents.py).
+
+ Calls `apply_concepts_to_graph` by its bare (module-global) name
+ rather than binding it to a local/default-arg at decoration time —
+ that matters because a plain global reference inside a function body
+ is looked up fresh from the function's `__globals__` (this module's
+ namespace) on EVERY call, not captured once when the function is
+ defined. That is what lets tests/test_dbos_resume.py's
+ `document_module.apply_concepts_to_graph = AsyncMock(...)`
+ monkeypatch take effect here: it reassigns the SAME name in this
+ module's namespace that this call resolves at call time. (A default
+ argument like `apply_fn=apply_concepts_to_graph` would instead freeze
+ in the ORIGINAL function object at decoration time and silently not
+ observe the monkeypatch.) routes/documents.py's streaming route
+ imports and calls the same `apply_concepts_to_graph` independently,
+ inline and non-durable per ADR 0011's streaming-route asymmetry —
+ a separate binding this monkeypatch doesn't touch, and doesn't need to.
+ """
+ return await apply_concepts_to_graph(user_id, course_id, concept_names)
+
+
 async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
  """Run classification first, then fan out the other workers in parallel.
 
@@ -152,22 +189,22 @@ async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
 
 @durable_workflow
 async def process_document(text: str, deps: SaplingDeps) -> DocumentProcessingResult:
- """Run workers in parallel, then merge concepts into the graph directly.
+ """Run workers in parallel, then merge concepts into the graph via the
+ durable `_step_apply_graph`.
 
  DocumentProcessingResult is composed deterministically here from worker
- outputs. The graph merge is a plain async function call — no orchestrator
- agent — because it has no decisions to make beyond passing the
- already-extracted concept names through.
+ outputs. The graph merge has no orchestrator agent — no decisions to
+ make beyond passing the already-extracted concept names through — but
+ IS wrapped as a checkpointed step (`_step_apply_graph`) so a DBOS
+ resume of this workflow doesn't repeat the merge.
 
  Wrapped in `@durable_workflow` from services.durable: a no-op when
  DBOS_ENABLED is unset (the default), a real DBOS workflow when the
  operator opts in. See ADR 0011 for the activation procedure.
  """
  workers = await _run_workers(text, deps)
  concept_names = [c.name for c in workers.concepts.concepts]
- merged = await apply_concepts_to_graph(
- deps.user_id, deps.course_id, concept_names,
- )
+ merged = await _step_apply_graph(deps.user_id, deps.course_id, concept_names)
  return DocumentProcessingResult(
  classification=workers.classification,
  summary=workers.summary,

backend/main.py

@@ -29,6 +29,7 @@
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
 from services.storage_service import ALLOWED_CONTENT_TYPES, ensure_bucket_exists
+from services.durable import init_dbos, shutdown_dbos
 
 try:
  from recost.frameworks.fastapi import RecostMiddleware
@@ -85,10 +86,15 @@ async def _lifespan(_app: FastAPI):
  # usage + event rows flush off the request path.
  from services import events_service
  events_service.start_worker()
+ # ADR 0011 / #154: construct + launch DBOS when DBOS_ENABLED=true; no-op
+ # passthrough otherwise. Fails loudly (raises) if the operator opted in
+ # and launch fails — see services/durable.py::init_dbos.
+ init_dbos()
  yield
  # Stop the drain thread and flush anything still queued so the last batch
  # of usage rows isn't lost on shutdown.
  events_service.shutdown()
+ shutdown_dbos()
 
 
 def _drop_request_arguments(_request, _attributes):

backend/requirements-durable.txt

@@ -0,0 +1,11 @@
+# Opt-in durable-execution extra (ADR 0011 / #154).
+#
+# Installs the `dbos` package so backend/services/durable.py's @workflow /
+# @step decorators go real when an operator sets DBOS_ENABLED=true (with
+# DBOS_DATABASE_URL pointing at a Postgres instance). Never in
+# requirements.txt or requirements.lock: durability is opt-in and the
+# default (prod-default and hermetic-test) path must keep working with
+# `dbos` NOT installed — see backend/tests/test_durable_shim.py.
+#
+# Install with: pip install -r requirements-durable.txt
+dbos>=2.28,<3

backend/routes/documents.py

@@ -41,6 +41,7 @@
 from services.achievement_service import check_achievements
 from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
+from services.durable import workflow_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
 from agents.classifier import classifier_agent
@@ -606,8 +607,13 @@ async def upload_document_sync(
  # both branches surface a retry-friendly 502: guardrail trips (budget /
  # degenerate output) log at WARNING, anything else is a bug and logs the
  # full exception.
+ # Scope the DBOS workflow id to user_id + request_id, not request_id
+ # alone: X-Request-ID is client-supplied, so an unscoped id would let
+ # one user's replay attach to another user's in-flight/completed
+ # workflow (state poisoning). No-op (nullcontext) when DBOS is off.
  try:
- result: DocumentProcessingResult = await process_document(extracted_text, deps)
+ with workflow_id(f"doc:{user_id}:{request_id}"):
+ result: DocumentProcessingResult = await process_document(extracted_text, deps)
  except (UsageLimitExceeded, UnexpectedModelBehavior) as e:
  logger.warning(
  "Agent guardrails tripped for '%s'; returning 502",

backend/agents/document.py

@@ -2,7 +2,8 @@
 
 Coordinates classification, summary, concept extraction, and (when
 applicable) syllabus extraction, then merges results into the user's
-course graph by calling `apply_concepts_to_graph` directly.
+course graph via `_step_apply_graph`, a durable step that calls
+`apply_concepts_to_graph`.
 
 There is no orchestrator agent here: the graph merge is a deterministic
 function call with concept names already produced by the workers, so
@@ -13,7 +14,8 @@
  asyncio.gather.
 - Classification runs first because it gates whether syllabus
  extraction runs at all.
-- The graph update is a direct async function call after workers complete.
+- The graph update runs as a durable step (`_step_apply_graph`) after
+ workers complete, so a DBOS resume does not re-run the merge.
 
 Failure contract (ADR 0024 — this pipeline is the ONLY upload pipeline;
 ADR 0001's legacy fallback was retired in #151b):
@@ -28,12 +30,19 @@
  from bare exceptions in logs.
 
 Internal API: the `_step_*` functions defined below are wrapped with
-@durable_step and are meant to be called ONLY from `_run_workers`,
-which is itself reached only via `process_document` (the
-@durable_workflow). Calling a `_step_*` outside the workflow is
-undefined behavior under DBOS — depending on version, it may no-op
-silently, raise, or warn. Don't import them from routes or other
-modules.
+@durable_step and are meant to be called ONLY from within the
+@durable_workflow-decorated `process_document` — either directly
+(`_step_apply_graph`) or via `_run_workers` (the other four). For the
+pinned dbos==2.28.0, calling a `_step_*` outside a workflow context is
+NOT undefined: `dbos/_core.py::decorate_step`'s wrapper checks
+`ctx.is_workflow()` and, when there's no ambient workflow context, falls
+straight through to `return func(*args, **kwargs)` — the plain function
+runs for real, synchronously, with no DBOS registry lookup and no error
+(verified at `dbos/_core.py:2126-2152`). That's well-defined but still
+wrong to rely on here: the call would execute but get NONE of DBOS's
+checkpoint/resume behavior, silently losing durability for that call.
+Don't import them from routes or other modules; call only from
+`_run_workers`/`process_document`.
 """
 
 from __future__ import annotations
@@ -119,6 +128,34 @@ async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
  return result.output
 
 
+@durable_step
+async def _step_apply_graph(
+ user_id: str, course_id: str | None, concept_names: list[str],
+) -> int:
+ """Merge extracted concepts into the graph as a durable step, so a
+ DBOS resume does not re-run the merge (it's the pipeline's one
+ real-Supabase side effect besides persistence, which happens outside
+ process_document entirely — see routes/documents.py).
+
+ Calls `apply_concepts_to_graph` by its bare (module-global) name
+ rather than binding it to a local/default-arg at decoration time —
+ that matters because a plain global reference inside a function body
+ is looked up fresh from the function's `__globals__` (this module's
+ namespace) on EVERY call, not captured once when the function is
+ defined. That is what lets tests/test_dbos_resume.py's
+ `document_module.apply_concepts_to_graph = AsyncMock(...)`
+ monkeypatch take effect here: it reassigns the SAME name in this
+ module's namespace that this call resolves at call time. (A default
+ argument like `apply_fn=apply_concepts_to_graph` would instead freeze
+ in the ORIGINAL function object at decoration time and silently not
+ observe the monkeypatch.) routes/documents.py's streaming route
+ imports and calls the same `apply_concepts_to_graph` independently,
+ inline and non-durable per ADR 0011's streaming-route asymmetry —
+ a separate binding this monkeypatch doesn't touch, and doesn't need to.
+ """
+ return await apply_concepts_to_graph(user_id, course_id, concept_names)
+
+
 async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
  """Run classification first, then fan out the other workers in parallel.
 
@@ -152,22 +189,22 @@ async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
 
 @durable_workflow
 async def process_document(text: str, deps: SaplingDeps) -> DocumentProcessingResult:
- """Run workers in parallel, then merge concepts into the graph directly.
+ """Run workers in parallel, then merge concepts into the graph via the
+ durable `_step_apply_graph`.
 
  DocumentProcessingResult is composed deterministically here from worker
- outputs. The graph merge is a plain async function call — no orchestrator
- agent — because it has no decisions to make beyond passing the
- already-extracted concept names through.
+ outputs. The graph merge has no orchestrator agent — no decisions to
+ make beyond passing the already-extracted concept names through — but
+ IS wrapped as a checkpointed step (`_step_apply_graph`) so a DBOS
+ resume of this workflow doesn't repeat the merge.
 
  Wrapped in `@durable_workflow` from services.durable: a no-op when
  DBOS_ENABLED is unset (the default), a real DBOS workflow when the
  operator opts in. See ADR 0011 for the activation procedure.
  """
  workers = await _run_workers(text, deps)
  concept_names = [c.name for c in workers.concepts.concepts]
- merged = await apply_concepts_to_graph(
- deps.user_id, deps.course_id, concept_names,
- )
+ merged = await _step_apply_graph(deps.user_id, deps.course_id, concept_names)
  return DocumentProcessingResult(
  classification=workers.classification,
  summary=workers.summary,

backend/main.py

@@ -29,6 +29,7 @@
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
 from services.storage_service import ALLOWED_CONTENT_TYPES, ensure_bucket_exists
+from services.durable import init_dbos, shutdown_dbos
 
 try:
  from recost.frameworks.fastapi import RecostMiddleware
@@ -85,10 +86,15 @@ async def _lifespan(_app: FastAPI):
  # usage + event rows flush off the request path.
  from services import events_service
  events_service.start_worker()
+ # ADR 0011 / #154: construct + launch DBOS when DBOS_ENABLED=true; no-op
+ # passthrough otherwise. Fails loudly (raises) if the operator opted in
+ # and launch fails — see services/durable.py::init_dbos.
+ init_dbos()
  yield
  # Stop the drain thread and flush anything still queued so the last batch
  # of usage rows isn't lost on shutdown.
  events_service.shutdown()
+ shutdown_dbos()
 
 
 def _drop_request_arguments(_request, _attributes):

backend/requirements-durable.txt

@@ -0,0 +1,11 @@
+# Opt-in durable-execution extra (ADR 0011 / #154).
+#
+# Installs the `dbos` package so backend/services/durable.py's @workflow /
+# @step decorators go real when an operator sets DBOS_ENABLED=true (with
+# DBOS_DATABASE_URL pointing at a Postgres instance). Never in
+# requirements.txt or requirements.lock: durability is opt-in and the
+# default (prod-default and hermetic-test) path must keep working with
+# `dbos` NOT installed — see backend/tests/test_durable_shim.py.
+#
+# Install with: pip install -r requirements-durable.txt
+dbos>=2.28,<3

backend/routes/documents.py

@@ -41,6 +41,7 @@
 from services.achievement_service import check_achievements
 from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
+from services.durable import workflow_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
 from agents.classifier import classifier_agent
@@ -606,8 +607,13 @@ async def upload_document_sync(
  # both branches surface a retry-friendly 502: guardrail trips (budget /
  # degenerate output) log at WARNING, anything else is a bug and logs the
  # full exception.
+ # Scope the DBOS workflow id to user_id + request_id, not request_id
+ # alone: X-Request-ID is client-supplied, so an unscoped id would let
+ # one user's replay attach to another user's in-flight/completed
+ # workflow (state poisoning). No-op (nullcontext) when DBOS is off.
  try:
- result: DocumentProcessingResult = await process_document(extracted_text, deps)
+ with workflow_id(f"doc:{user_id}:{request_id}"):
+ result: DocumentProcessingResult = await process_document(extracted_text, deps)
  except (UsageLimitExceeded, UnexpectedModelBehavior) as e:
  logger.warning(
  "Agent guardrails tripped for '%s'; returning 502",

backend/agents/document.py

@@ -2,7 +2,8 @@
 
 Coordinates classification, summary, concept extraction, and (when
 applicable) syllabus extraction, then merges results into the user's
-course graph by calling `apply_concepts_to_graph` directly.
+course graph via `_step_apply_graph`, a durable step that calls
+`apply_concepts_to_graph`.
 
 There is no orchestrator agent here: the graph merge is a deterministic
 function call with concept names already produced by the workers, so
@@ -13,7 +14,8 @@
  asyncio.gather.
 - Classification runs first because it gates whether syllabus
  extraction runs at all.
-- The graph update is a direct async function call after workers complete.
+- The graph update runs as a durable step (`_step_apply_graph`) after
+ workers complete, so a DBOS resume does not re-run the merge.
 
 Failure contract (ADR 0024 — this pipeline is the ONLY upload pipeline;
 ADR 0001's legacy fallback was retired in #151b):
@@ -28,12 +30,19 @@
  from bare exceptions in logs.
 
 Internal API: the `_step_*` functions defined below are wrapped with
-@durable_step and are meant to be called ONLY from `_run_workers`,
-which is itself reached only via `process_document` (the
-@durable_workflow). Calling a `_step_*` outside the workflow is
-undefined behavior under DBOS — depending on version, it may no-op
-silently, raise, or warn. Don't import them from routes or other
-modules.
+@durable_step and are meant to be called ONLY from within the
+@durable_workflow-decorated `process_document` — either directly
+(`_step_apply_graph`) or via `_run_workers` (the other four). For the
+pinned dbos==2.28.0, calling a `_step_*` outside a workflow context is
+NOT undefined: `dbos/_core.py::decorate_step`'s wrapper checks
+`ctx.is_workflow()` and, when there's no ambient workflow context, falls
+straight through to `return func(*args, **kwargs)` — the plain function
+runs for real, synchronously, with no DBOS registry lookup and no error
+(verified at `dbos/_core.py:2126-2152`). That's well-defined but still
+wrong to rely on here: the call would execute but get NONE of DBOS's
+checkpoint/resume behavior, silently losing durability for that call.
+Don't import them from routes or other modules; call only from
+`_run_workers`/`process_document`.
 """
 
 from __future__ import annotations
@@ -119,6 +128,34 @@ async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
  return result.output
 
 
+@durable_step
+async def _step_apply_graph(
+ user_id: str, course_id: str | None, concept_names: list[str],
+) -> int:
+ """Merge extracted concepts into the graph as a durable step, so a
+ DBOS resume does not re-run the merge (it's the pipeline's one
+ real-Supabase side effect besides persistence, which happens outside
+ process_document entirely — see routes/documents.py).
+
+ Calls `apply_concepts_to_graph` by its bare (module-global) name
+ rather than binding it to a local/default-arg at decoration time —
+ that matters because a plain global reference inside a function body
+ is looked up fresh from the function's `__globals__` (this module's
+ namespace) on EVERY call, not captured once when the function is
+ defined. That is what lets tests/test_dbos_resume.py's
+ `document_module.apply_concepts_to_graph = AsyncMock(...)`
+ monkeypatch take effect here: it reassigns the SAME name in this
+ module's namespace that this call resolves at call time. (A default
+ argument like `apply_fn=apply_concepts_to_graph` would instead freeze
+ in the ORIGINAL function object at decoration time and silently not
+ observe the monkeypatch.) routes/documents.py's streaming route
+ imports and calls the same `apply_concepts_to_graph` independently,
+ inline and non-durable per ADR 0011's streaming-route asymmetry —
+ a separate binding this monkeypatch doesn't touch, and doesn't need to.
+ """
+ return await apply_concepts_to_graph(user_id, course_id, concept_names)
+
+
 async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
  """Run classification first, then fan out the other workers in parallel.
 
@@ -152,22 +189,22 @@ async def _run_workers(text: str, deps: SaplingDeps) -> _WorkerResults:
 
 @durable_workflow
 async def process_document(text: str, deps: SaplingDeps) -> DocumentProcessingResult:
- """Run workers in parallel, then merge concepts into the graph directly.
+ """Run workers in parallel, then merge concepts into the graph via the
+ durable `_step_apply_graph`.
 
  DocumentProcessingResult is composed deterministically here from worker
- outputs. The graph merge is a plain async function call — no orchestrator
- agent — because it has no decisions to make beyond passing the
- already-extracted concept names through.
+ outputs. The graph merge has no orchestrator agent — no decisions to
+ make beyond passing the already-extracted concept names through — but
+ IS wrapped as a checkpointed step (`_step_apply_graph`) so a DBOS
+ resume of this workflow doesn't repeat the merge.
 
  Wrapped in `@durable_workflow` from services.durable: a no-op when
  DBOS_ENABLED is unset (the default), a real DBOS workflow when the
  operator opts in. See ADR 0011 for the activation procedure.
  """
  workers = await _run_workers(text, deps)
  concept_names = [c.name for c in workers.concepts.concepts]
- merged = await apply_concepts_to_graph(
- deps.user_id, deps.course_id, concept_names,
- )
+ merged = await _step_apply_graph(deps.user_id, deps.course_id, concept_names)
  return DocumentProcessingResult(
  classification=workers.classification,
  summary=workers.summary,

backend/main.py

@@ -29,6 +29,7 @@
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
 from services.storage_service import ALLOWED_CONTENT_TYPES, ensure_bucket_exists
+from services.durable import init_dbos, shutdown_dbos
 
 try:
  from recost.frameworks.fastapi import RecostMiddleware
@@ -85,10 +86,15 @@ async def _lifespan(_app: FastAPI):
  # usage + event rows flush off the request path.
  from services import events_service
  events_service.start_worker()
+ # ADR 0011 / #154: construct + launch DBOS when DBOS_ENABLED=true; no-op
+ # passthrough otherwise. Fails loudly (raises) if the operator opted in
+ # and launch fails — see services/durable.py::init_dbos.
+ init_dbos()
  yield
  # Stop the drain thread and flush anything still queued so the last batch
  # of usage rows isn't lost on shutdown.
  events_service.shutdown()
+ shutdown_dbos()
 
 
 def _drop_request_arguments(_request, _attributes):

backend/requirements-durable.txt

@@ -0,0 +1,11 @@
+# Opt-in durable-execution extra (ADR 0011 / #154).
+#
+# Installs the `dbos` package so backend/services/durable.py's @workflow /
+# @step decorators go real when an operator sets DBOS_ENABLED=true (with
+# DBOS_DATABASE_URL pointing at a Postgres instance). Never in
+# requirements.txt or requirements.lock: durability is opt-in and the
+# default (prod-default and hermetic-test) path must keep working with
+# `dbos` NOT installed — see backend/tests/test_durable_shim.py.
+#
+# Install with: pip install -r requirements-durable.txt
+dbos>=2.28,<3

backend/routes/documents.py

@@ -41,6 +41,7 @@
 from services.achievement_service import check_achievements
 from services.agent_events import SSE_CACHE_CONTROL, SaplingEvent, sapling_event_to_sse
 from services.request_context import current_request_id
+from services.durable import workflow_id
 from agents import WORKER_LIMITS
 from agents._providers import model_mode
 from agents.classifier import classifier_agent
@@ -606,8 +607,13 @@ async def upload_document_sync(
  # both branches surface a retry-friendly 502: guardrail trips (budget /
  # degenerate output) log at WARNING, anything else is a bug and logs the
  # full exception.
+ # Scope the DBOS workflow id to user_id + request_id, not request_id
+ # alone: X-Request-ID is client-supplied, so an unscoped id would let
+ # one user's replay attach to another user's in-flight/completed
+ # workflow (state poisoning). No-op (nullcontext) when DBOS is off.
  try:
- result: DocumentProcessingResult = await process_document(extracted_text, deps)
+ with workflow_id(f"doc:{user_id}:{request_id}"):
+ result: DocumentProcessingResult = await process_document(extracted_text, deps)
  except (UsageLimitExceeded, UnexpectedModelBehavior) as e:
  logger.warning(
  "Agent guardrails tripped for '%s'; returning 502",