diff --git a/backend/agents/document.py b/backend/agents/document.py index 6fad8f0b..c6853907 100644 --- a/backend/agents/document.py +++ b/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,12 +189,14 @@ 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 @@ -165,9 +204,7 @@ async def process_document(text: str, deps: SaplingDeps) -> DocumentProcessingRe """ 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, diff --git a/backend/main.py b/backend/main.py index 8c7fc61f..03206bc5 100644 --- a/backend/main.py +++ b/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): diff --git a/backend/requirements-durable.txt b/backend/requirements-durable.txt new file mode 100644 index 00000000..e7487ebf --- /dev/null +++ b/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 diff --git a/backend/routes/documents.py b/backend/routes/documents.py index 7ee44490..4693f4c2 100644 --- a/backend/routes/documents.py +++ b/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", diff --git a/backend/services/durable.py b/backend/services/durable.py index e69b9962..35fc46f7 100644 --- a/backend/services/durable.py +++ b/backend/services/durable.py @@ -1,22 +1,79 @@ -"""Optional durable-execution shim. +"""Optional durable-execution shim (ADR 0011 / #154). -When `DBOS_ENABLED=true` AND the `dbos` package is importable AND -`DBOS_DATABASE_URL` is set in the env, this module exposes real DBOS -workflow + step decorators so an in-flight upload can survive a worker -crash and resume from the last checkpoint. +When ALL of the following hold: -When any of those preconditions fail (the default state in this repo), -the decorators degrade to identity passthroughs that don't add anything -to the wrapped function. Code is callable in both modes; the only -difference is durability. + - DBOS_ENABLED=true + - the `dbos` package is importable (backend/requirements-durable.txt, + installed separately from requirements.txt/requirements.lock) + - DBOS_DATABASE_URL is a non-empty Postgres connection string -This lets us land the integration code in main, document the path in -ADR 0011, and let operators flip the flag once their DBOS Postgres -schema is provisioned — without making `dbos` a hard import. +this module exposes real DBOS workflow + step decorators, and +`workflow_id()` (below) lets a caller pin an invocation to a specific DBOS +workflow id. The product-level crash semantic this enables: a CLIENT RETRY +of the same logical operation (same idempotency key, e.g. `/upload/sync`'s +`X-Request-ID`) attaches to the SAME workflow instead of starting a new +one, resuming at the last completed step instead of re-running every +agent call. Nothing resumes for the ORIGINAL caller — their HTTP +connection is already gone once the worker crashes — but DBOS's own +background auto-recovery (`init_dbos()` -> `DBOS.launch()`, see below) +independently completes an abandoned in-flight workflow even without a +retry, and a later same-id retry receives THAT recorded result instead of +re-running the pipeline. + +When any precondition fails (the default state in this repo — the flag is +off), the decorators degrade to identity passthroughs that don't add +anything to the wrapped function. Code is callable in both modes; the only +difference is durability. If the flag is on but the `dbos` import fails, or +DBOS_DATABASE_URL is missing, we log a warning and degrade to passthrough +rather than crash at import time — this module is imported by +agents/document.py, which is imported by main.py, so an import-time raise +here would take the whole app down before validate_config() even runs. + +Construction + launch is a SEPARATE step from decoration, deliberately: +`init_dbos()` (called from main.py's `_lifespan`, after the #174 secrets +validation) constructs the `DBOS` singleton and calls `DBOS.launch()`. That +split is required by import order, not just style — `agents.document` +applies `@workflow`/`@step` to `process_document`/`_step_*` at IMPORT time, +which happens well before the FastAPI lifespan runs `init_dbos()`. Verified +against the installed dbos==2.28.0 (backend/requirements-durable.txt) that +this decorate-before-construct-before-launch order is exactly the supported +pattern: `DBOS.workflow()`/`DBOS.step()` register onto a lazily-created +global `DBOSRegistry` (`dbos/_dbos.py::_get_or_create_dbos_registry`, no +`DBOS` instance required), and `DBOS.__init__` picks up that SAME registry +in `self._registry = _get_or_create_dbos_registry()` (`dbos/_dbos.py:417`) +whenever the instance is later constructed — so registrations recorded +before construction are not lost. If a future dbos major version drops this +registry indirection, decorating agents.document at import time would need +to move to something init_dbos() runs directly; the fact that our decorator +capture (below) and DBOS() construction (in init_dbos) are already two +separate steps rather than one makes that migration a local change, not a +redesign. + +Fail-loud contract for init_dbos(): an explicit DBOS_ENABLED=true opt-in +must not degrade silently. The failure mode that matters most: if a +precondition above (import, DBOS_DATABASE_URL) already failed, the +decorators are ALREADY identity passthroughs by the time init_dbos() runs +(this module logged one WARNING at import time, nothing more) — +`@workflow`/`@step` code then runs and SUCCEEDS with zero durability, no +further signal, on routes that look durable in the source (X-Request-ID +idempotency, the decorations are present). init_dbos() now RAISES in that +case too, not just on a construct/launch failure — same posture as #174's +validate_config(). See init_dbos()'s own docstring for exactly which raises +when. + +(This is a DIFFERENT failure mode from decorators going real with `DBOS()` +never constructed anywhere — that already fails LOUD on its own: +dbos/_core.py's `workflow_wrapper` raises `DBOSException("... invoked +before DBOS initialized")` on every such call, deterministically, verified +against dbos==2.28.0 at `dbos/_core.py:1369-1372`. That was the actual +state of every decorated call before #154 shipped this file's init_dbos() +— a deterministic exception (a 502 via routes/documents.py's exception +handling) on every call, not a silent no-op. See ADR 0011's #154 update.) """ from __future__ import annotations +import contextlib import logging import os from functools import wraps @@ -28,25 +85,43 @@ _ENABLED = os.getenv("DBOS_ENABLED", "false").lower() == "true" +_DATABASE_URL = os.getenv("DBOS_DATABASE_URL", "").strip() _HAS_DBOS = False _dbos_workflow = None _dbos_step = None +# Set when `dbos` fails to import with DBOS_ENABLED=true. init_dbos() cites +# this in its RuntimeError so an operator sees WHY activation failed, not +# just THAT it did (Finding B / #154 review round). +_IMPORT_ERROR: str | None = None if _ENABLED: - try: - from dbos import DBOS # type: ignore[import-not-found] - # DBOS init must be done by the application entrypoint; we just - # capture the decorators here and trust that DBOS() was called - # in main.py BEFORE any decorated function is invoked. - _dbos_workflow = DBOS.workflow - _dbos_step = DBOS.step - _HAS_DBOS = True - except Exception as e: # ImportError or DBOS init failure + if not _DATABASE_URL: logger.warning( - "DBOS_ENABLED=true but DBOS could not be loaded (%s). " - "Durable decorators will degrade to no-ops.", - e, + "DBOS_ENABLED=true but DBOS_DATABASE_URL is not set. Durable " + "decorators will degrade to passthroughs until a database URL " + "is provided (see docs/decisions/0011-durable-execution-dbos.md). " + "init_dbos() will raise at startup until this is fixed." ) + else: + try: + from dbos import DBOS # type: ignore[import-not-found] + + # Only capture the decorator factories here. Constructing the + # DBOS() singleton and calling DBOS.launch() happens later, in + # init_dbos() — see the module docstring for why that split is + # required (agents.document decorates at import time, well + # before main.py's lifespan runs). + _dbos_workflow = DBOS.workflow + _dbos_step = DBOS.step + _HAS_DBOS = True + except Exception as e: # ImportError or anything else at import + _IMPORT_ERROR = str(e) + logger.warning( + "DBOS_ENABLED=true but DBOS could not be imported (%s). " + "Durable decorators will degrade to passthroughs. " + "init_dbos() will raise at startup until this is fixed.", + e, + ) def is_durable() -> bool: @@ -84,3 +159,183 @@ async def passthrough(*args: Any, **kwargs: Any) -> Any: return await fn(*args, **kwargs) return passthrough # type: ignore[return-value] + + +def workflow_id(wfid: str) -> "contextlib.AbstractContextManager[None]": + """Pin `wfid` as the DBOS workflow id for the next workflow invocation + started inside this `with` block (only the FIRST one started inside + the block gets it — see `SetWorkflowID`'s own docstring). + + Why this matters: a caller that re-enters this context with the SAME + `wfid` (e.g. a client retry presenting the same idempotency key) + attaches to the SAME workflow row instead of starting a new one, and + does not re-execute already-checkpointed `@step` calls inside it. + Verified against the installed dbos==2.28.0: + + - `from dbos import SetWorkflowID` — re-exported at the top of + `dbos/__init__.py` (`dbos/__init__.py:8`), defined at + `dbos/_context.py:454`. A plain SYNC context manager (`__enter__`/ + `__exit__`, no `async with` needed): `__enter__` sets + `ctx.id_assigned_for_next_workflow = wfid` on the ambient + DBOSContext (`dbos/_context.py:471-485`), which + `workflow_wrapper` (`dbos/_core.py`) reads when starting the next + `@workflow`-decorated call. + - Same-id reattach: `workflow_wrapper` inserts/updates the workflow's + row keyed on `workflow_uuid` via an upsert + (`dbos/_sys_db.py::_insert_workflow_status`, `dbos/_sys_db.py:718`, + ON CONFLICT DO UPDATE). A plain (non-recovery, non-dequeue) call + whose `workflow_uuid` already has a row does NOT re-run the + workflow body — `should_execute` stays True only for the original + inserting call (or a recovery/dequeue request); every other direct + call instead gets `_deferred_workflow_result`, which awaits the + existing workflow's recorded result (`dbos/_core.py:1452-1459`; + the owner-mismatch check is `dbos/_sys_db.py:875-880`). This holds + for BOTH a completed (SUCCESS) row — the recorded output returns + immediately, no re-execution — and a still-PENDING one, which + blocks until SOME execution finishes it (typically + `DBOS.launch()`'s own background auto-recovery of PENDING rows for + this executor — see `init_dbos()`) and records a result. Either + way, already-checkpointed `@step` calls inside are never re-run. + + No-op (`contextlib.nullcontext()`) when `is_durable()` is False, so + callers don't need to branch on the flag themselves. + """ + if is_durable(): + from dbos import SetWorkflowID # dbos/_context.py:454; dbos/__init__.py:8 + + return SetWorkflowID(wfid) + return contextlib.nullcontext() + + +def init_dbos() -> bool: + """Construct and launch the DBOS runtime, once, from main.py's + `_lifespan` (after the #174 secrets validation, before `yield`). + + Three outcomes: + + - DBOS_ENABLED unset/false (the default): no-op, returns False. Logs + one INFO line. `@workflow`/`@step` stay identity passthroughs. + - DBOS_ENABLED=true but `is_durable()` is False — a precondition + failed at import time (`dbos` not importable, or DBOS_DATABASE_URL + missing; this module already logged one WARNING and degraded the + decorators to passthrough): RAISES `RuntimeError` naming exactly + which precondition failed. Without this, an explicit opt-in would + silently run with zero durability — routes look durable in the + source (the `@workflow`/`@step` decorations are present) while every + request quietly loses crash-resume, with no signal beyond the one + WARNING line at boot. Same posture as #174's validate_config(). + - DBOS_ENABLED=true and `is_durable()` is True: constructs `DBOS` and + calls `DBOS.launch()` (below). ANY exception here is logged and + RE-RAISED too — same fail-loud posture, for the same reason. + + (Both RAISE paths above are silent-degradation guards new in this + update. Neither is the same failure mode as a `@workflow`/`@step` + call reaching a registry whose `DBOS()` was never constructed at all + — that already fails loud on its own: dbos/_core.py's + `workflow_wrapper` raises `DBOSException("... invoked before DBOS + initialized")` (`dbos/_core.py:1369-1372`, verified against + dbos==2.28.0) the instant such a call is made, surfaced as a 502 via + routes/documents.py's exception handling. That was the actual state of + every decorated call before #154 shipped this function — a + deterministic exception on every call, not a no-op; see ADR 0011's + #154 update.) + + When durable, this is the ONE place the `DBOS` singleton is + constructed and `DBOS.launch()` is called: + + - Config: verified against dbos==2.28.0's `DBOSConfig` TypedDict + (`dbos/_dbos_config.py`). `name` is the only required key + (`translate_dbos_config_to_config_file` raises + `DBOSInitializationError` without it). We set `system_database_url` + to DBOS_DATABASE_URL — that is the checkpoint store our + `@workflow`/`@step` decorators read/write. We deliberately do NOT + set `database_url` (DEPRECATED — of the two, that is the ONLY one + `DBOSConfig`'s own docstring marks that way) or + `application_database_url` (current, NOT deprecated, but + provisions a SEPARATE "application database" used only by + `@DBOS.transaction`-decorated functions, which Sapling doesn't + use): `dbos/_dbos_config.py::get_system_database_url` reads + `system_database_url` first and only falls back to deriving a + sys-db name from `database_url` when `system_database_url` is + absent — so passing ours directly is both sufficient and the + documented default path). + - Migrations: `DBOS.launch()` runs the system-database migrations + ITSELF (`dbos/_dbos.py::DBOS._launch` calls + `self._sys_db.run_migrations()` before doing anything else) — no + separate `dbos migrate` CLI step is required for normal operation. + (`dbos.run_dbos_database_migrations()` exists as a standalone + helper for the case where the app's DB role lacks DDL grants and + migrations must be pre-provisioned by a different role — not our + setup; Supabase-adjacent Postgres roles here have DDL rights.) + - Recovery: `DBOS.launch()` ALSO auto-recovers PENDING workflows for + this executor on startup (`DBOS._launch` queries + `self._sys_db.get_pending_workflows(...)` and resumes each one on + a background thread) — no explicit recovery call is needed on the + happy path. See tests/test_dbos_resume.py for a crash/resume proof + against a real Postgres. + + ANY exception while constructing/launching (below) is logged and + RE-RAISED — the second of the two RAISE paths described above. + """ + if not _ENABLED: + logger.info("durable execution off — passthrough decorators") + return False + + if not is_durable(): + # DBOS_ENABLED=true but a precondition already failed at import + # time (see the module-level check above) — name exactly which one, + # rather than degrading to the flag-off no-op. Fixes the silent- + # degrade gap: previously this returned False here too, so an + # explicit opt-in with a typo'd/missing DBOS_DATABASE_URL (or an + # uninstalled `dbos`) would boot and serve requests looking durable + # in the source while running with zero durability. + if not _DATABASE_URL: + raise RuntimeError( + "DBOS_ENABLED=true but DBOS_DATABASE_URL is not set. Set " + "DBOS_DATABASE_URL to a Postgres connection string, or " + "unset DBOS_ENABLED to run without durability (see " + "docs/decisions/0011-durable-execution-dbos.md)." + ) + raise RuntimeError( + f"DBOS_ENABLED=true but the `dbos` package could not be " + f"imported ({_IMPORT_ERROR}). Install " + f"backend/requirements-durable.txt, or unset DBOS_ENABLED to " + f"run without durability." + ) + + try: + from dbos import DBOS, DBOSConfig # type: ignore[import-not-found] + + config: DBOSConfig = { + "name": "sapling", + "system_database_url": _DATABASE_URL, + } + DBOS(config=config) + DBOS.launch() + except Exception: + logger.exception( + "DBOS_ENABLED=true but DBOS failed to construct/launch. Failing " + "startup loudly rather than serving requests with un-launched " + "durable decorators." + ) + raise + + logger.info("durable execution ACTIVE (DBOS launched)") + return True + + +def shutdown_dbos() -> None: + """Tear down the DBOS runtime from main.py's `_lifespan` shutdown path. + + No-op when not durable. Never raises — shutdown must not be the thing + that turns a clean deploy rollover into a crash loop; any failure here + is logged instead. + """ + if not is_durable(): + return + try: + from dbos import DBOS # type: ignore[import-not-found] + + DBOS.destroy() + except Exception: + logger.exception("Error shutting down DBOS; continuing shutdown.") diff --git a/backend/tests/test_dbos_resume.py b/backend/tests/test_dbos_resume.py new file mode 100644 index 00000000..bc929c27 --- /dev/null +++ b/backend/tests/test_dbos_resume.py @@ -0,0 +1,525 @@ +"""Opt-in DBOS crash/resume proof (ADR 0011 / #154). + +Skipped by default: needs the real `dbos` package (backend/requirements- +durable.txt, never requirements.txt/requirements.lock) AND a live Postgres +for DBOS's own system database. Run explicitly with: + + RUN_DBOS_RESUME=1 DBOS_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:54322/postgres \ + venv/bin/python -m pytest tests/test_dbos_resume.py -q + +(defaults to the local Supabase stack's Postgres -- see docs/local-supabase.md +-- but any reachable Postgres works; DBOS provisions its own `dbos` schema in +whatever database the URL points at, alongside Supabase's own schemas). + +Both tests use a SUBPROCESS per phase rather than reloading services.durable +in-process. That is not just style: `agents.document` applies +`@durable_workflow` / `@durable_step` to `process_document` / `_step_*` at +IMPORT time (see agents/document.py's docstring), and a pytest session has +almost certainly already imported `agents.document` (transitively, via +`routes.documents` -> `main`, which `test_documents_routes.py` and others +import at collection time) BEFORE this file's tests run -- using whatever +`services.durable.workflow`/`step` looked like at THAT import (passthrough, +since the default test env has DBOS_ENABLED unset). Reloading +`services.durable` later would not retroactively change the ALREADY-BOUND +decorators on `agents.document.process_document`. A fresh subprocess, with +DBOS_ENABLED=true set before the interpreter even starts, is the only way to +get a REAL DBOS-backed `process_document` / minimal workflow to test against. + +Helper scripts below are plain `#`-commented (not docstring'd) deliberately: +they are string constants embedded in THIS file, written out verbatim to a +tmp_path file and run by a fresh interpreter -- a triple-quoted docstring +inside a triple-quoted Python string constant is a real nesting headache +(first attempt at this file tripped over exactly that), and a `#` comment +sidesteps it entirely. +""" + +from __future__ import annotations + +import importlib.util +import json +import os +import subprocess +import sys +import uuid +from pathlib import Path + +import pytest + +_BACKEND_DIR = Path(__file__).resolve().parents[1] + +pytestmark = [ + pytest.mark.skipif( + os.getenv("RUN_DBOS_RESUME") != "1", + reason="opt-in: needs dbos + local Postgres (RUN_DBOS_RESUME=1)", + ), + pytest.mark.skipif( + importlib.util.find_spec("dbos") is None, + reason="dbos not installed (see backend/requirements-durable.txt)", + ), +] + +_DEFAULT_DBOS_DATABASE_URL = "postgresql://postgres:postgres@127.0.0.1:54322/postgres" + + +def _dbos_database_url() -> str: + return os.getenv("DBOS_DATABASE_URL") or _DEFAULT_DBOS_DATABASE_URL + + +def _subprocess_env(**extra: str) -> dict: + env = dict(os.environ) + env["PYTHONPATH"] = "." + env["DBOS_ENABLED"] = "true" + env["DBOS_DATABASE_URL"] = _dbos_database_url() + env.update(extra) + return env + + +def _run_helper(script_path: Path, *args: str, env: dict) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, str(script_path), *args], + cwd=str(_BACKEND_DIR), + env=env, + capture_output=True, + text=True, + timeout=60, + ) + + +# -- 1. Minimal 2-step workflow: crash mid-step-2, resume, assert no re-run -- +# +# A 2-step DBOS workflow used to prove OUR shim wires real DBOS crash-resume, +# not just that dbos itself works. Run twice as two SEPARATE processes +# against the same DBOS_DATABASE_URL + workflow id: +# +# phase crash -- runs step1 (checkpointed), then step2 marks a marker +# file and os._exit(42) BEFORE returning (simulates a +# worker crash mid-workflow, leaving the row PENDING). +# phase resume -- a fresh process; DBOS.launch()'s startup recovery +# re-runs ONLY step2 (step1 is already checkpointed +# SUCCESS) and waits for completion via +# DBOS.retrieve_workflow(...).get_result(). + +_MINIMAL_WORKFLOW_HELPER = r""" +import asyncio +import os +import sys + +import services.durable as durable + +phase = sys.argv[1] +workflow_id = sys.argv[2] +step1_counter = sys.argv[3] +step2_counter = sys.argv[4] +marker_path = sys.argv[5] + +assert durable.is_durable(), "DBOS_ENABLED=true but the shim did not activate" + + +@durable.step +async def step1(): + with open(step1_counter, "a") as f: + print("x", file=f) + + +@durable.step +async def step2(): + if not os.path.exists(marker_path): + open(marker_path, "w").close() + os._exit(42) # simulate a hard crash mid-workflow, before checkpointing + with open(step2_counter, "a") as f: + print("x", file=f) + return "done" + + +@durable.workflow +async def minimal_workflow(): + await step1() + return await step2() + + +durable.init_dbos() + +if phase == "crash": + from dbos import SetWorkflowID + + with SetWorkflowID(workflow_id): + asyncio.run(minimal_workflow()) + print("RESULT:unreachable") # os._exit(42) fires before this +elif phase == "resume": + from dbos import DBOS + + handle = DBOS.retrieve_workflow(workflow_id) + result = handle.get_result(polling_interval_sec=0.2) + print("RESULT:" + str(result)) +else: + raise SystemExit("unknown phase " + repr(phase)) +""" + + +def test_shim_resume_minimal_workflow(tmp_path): + """Crash mid-workflow, resume in a fresh process, and assert the + resume-at-last-completed-step contract: step1 (already checkpointed + before the crash) does NOT re-run, and step2 runs exactly once to + completion on resume. + """ + script = tmp_path / "minimal_workflow_helper.py" + script.write_text(_MINIMAL_WORKFLOW_HELPER) + + workflow_id = "dbos-resume-test-" + str(uuid.uuid4()) + step1_counter = tmp_path / "step1_counter.txt" + step2_counter = tmp_path / "step2_counter.txt" + marker = tmp_path / "step2_marker.txt" + + env = _subprocess_env() + args = (workflow_id, str(step1_counter), str(step2_counter), str(marker)) + + crash = _run_helper(script, "crash", *args, env=env) + assert crash.returncode == 42, ( + "expected the crash phase to os._exit(42); stdout=" + repr(crash.stdout) + + " stderr=" + repr(crash.stderr) + ) + + resume = _run_helper(script, "resume", *args, env=env) + assert resume.returncode == 0, ( + "expected the resume phase to complete cleanly; stdout=" + repr(resume.stdout) + + " stderr=" + repr(resume.stderr) + ) + assert "RESULT:done" in resume.stdout + + # The resume-at-last-completed-step contract: step1 ran exactly once + # (during the crash phase) -- DBOS must NOT re-run an already- + # checkpointed step. step2 also ran exactly once to completion (its + # first, crashing invocation never returned, so it never appended to + # the counter file -- only the resumed invocation did). + assert step1_counter.read_text().splitlines() == ["x"] + assert step2_counter.read_text().splitlines() == ["x"] + + +# -- 2. process_document under DBOS produces the same shape as DBOS-off ----- +# +# What this proves: `agents.document.process_document`, decorated for REAL +# under DBOS (not passthrough), still returns the exact same deterministic +# function-mode output as the DBOS-off hermetic suite asserts elsewhere +# (tests/test_e2e_function_handlers.py) -- i.e. wrapping it in +# `@durable_workflow`/`@durable_step` changes durability, not behavior. +# +# What this does NOT prove: this stubs `apply_concepts_to_graph` (the one +# real-Supabase side effect inside process_document) rather than running +# against the live local stack + seeded course/user rows -- wiring that +# full stack through a subprocess was disproportionate to what this test is +# for (see the module docstring). The HTTP-route-level, real-DB behavior +# under DBOS-off is already covered by tests/test_documents_routes.py and +# the Chapter 1 Playwright upload journey; this test's job is narrower and +# DBOS-specific: prove the workflow is genuinely DBOS-backed +# (`durable.is_durable()` True in the subprocess) AND its output is +# unchanged. +# +# Env (DBOS_ENABLED, DBOS_DATABASE_URL, SAPLING_MODEL_MODE=function, +# SAPLING_FUNCTION_HANDLERS=agents.function_handlers_e2e) is set before this +# helper's interpreter starts, so agents.document decorates +# process_document/_step_* as REAL DBOS workflow/steps at import time below. + +_PROCESS_DOCUMENT_HELPER = r""" +import asyncio +import json +from unittest.mock import AsyncMock + +import services.durable as durable + +assert durable.is_durable(), "DBOS_ENABLED=true but the shim did not activate" + +import agents.document as document_module +from agents.deps import SaplingDeps + +# The only real-Supabase side effect inside process_document -- stubbed so +# this test needs nothing but DBOS's own Postgres (see module comment above). +document_module.apply_concepts_to_graph = AsyncMock(return_value=0) + +durable.init_dbos() + +deps = SaplingDeps( + user_id="dbos-resume-test-user", + course_id="dbos-resume-test-course", + supabase=None, + request_id="dbos-resume-test", +) +result = asyncio.run(document_module.process_document( + "Deterministic fixture text for the DBOS pipeline-parity test.", deps, +)) + +print("RESULT_JSON:" + json.dumps({ + "category": result.classification.category, + "abstract": result.summary.abstract, + "concepts": sorted(c.name for c in result.concepts.concepts), +})) +""" + + +def test_pipeline_identical_with_dbos_on(tmp_path): + """process_document, decorated for real under DBOS, returns the same + classification/summary/concepts shape as the function-mode constants + (agents/function_handlers_e2e.py) -- the same constants the DBOS-off + hermetic suite pins. See the section comment above for exactly what + this does and does not prove. + """ + from agents.function_handlers_e2e import E2E_DOC_ABSTRACT, E2E_DOC_CATEGORY, E2E_DOC_CONCEPTS + + script = tmp_path / "process_document_helper.py" + script.write_text(_PROCESS_DOCUMENT_HELPER) + + env = _subprocess_env( + SAPLING_MODEL_MODE="function", + SAPLING_FUNCTION_HANDLERS="agents.function_handlers_e2e", + ) + proc = _run_helper(script, env=env) + assert proc.returncode == 0, ( + "expected process_document to complete under DBOS; stdout=" + repr(proc.stdout) + + " stderr=" + repr(proc.stderr) + ) + + [result_line] = [ + line for line in proc.stdout.splitlines() if line.startswith("RESULT_JSON:") + ] + payload = json.loads(result_line[len("RESULT_JSON:"):]) + + assert payload["category"] == E2E_DOC_CATEGORY + assert payload["abstract"] == E2E_DOC_ABSTRACT + assert payload["concepts"] == sorted(name for name, _desc, _imp in E2E_DOC_CONCEPTS) + + +# -- 3. Real pipeline: crash mid-flight, retry with the SAME workflow_id, -- +# and prove DBOS resumes rather than re-runs (the toy-workflow gap) ----- +# +# Sections 1 and 2 above prove two separate things in isolation: a raw DBOS +# workflow resumes correctly (1), and process_document runs for real under +# DBOS with unchanged output (2). Neither proves the actual PRODUCT +# behavior: that a client retry of a crashed upload (the same +# X-Request-ID, per routes/documents.py's `/upload/sync`) attaches to the +# SAME workflow and skips already-completed steps, rather than starting an +# unrelated new workflow that re-runs everything. This test is that proof, +# using services.durable.workflow_id(...) the same way the route does +# (`/upload/sync` wraps process_document in +# `workflow_id(f"doc:{user_id}:{request_id}")`). +# +# Phase 1 ("crash"): a TEST-LOCAL SAPLING_FUNCTION_HANDLERS module scripts +# the classifier/summary/concepts tasks process_document reaches (no +# syllabus -- the classifier handler always answers non-syllabus). The +# classify handler appends a line to a counter file on every invocation. +# The summary handler simulates a worker crash: on the very first call +# ever (a marker file doesn't exist yet), it creates the marker and calls +# os._exit(42) BEFORE returning -- exactly like section 1's step2, this +# skips checkpointing that step, leaving the workflow row PENDING. +# Classification completes and checkpoints BEFORE summary/concepts even +# start (agents.document._run_workers awaits it first), so it must NOT +# re-run in phase 2. +# +# Phase 2 ("resume"): a fresh process, same workflow id, same handlers +# module (the marker file now exists, so summary succeeds this time). +# `durable.init_dbos()`'s own launch-time auto-recovery picks up the +# still-PENDING workflow from phase 1 and resumes it on a background +# thread; this phase's own `process_document(...)` call (wrapped in the +# SAME `workflow_id(...)`) does NOT re-run the body itself -- a plain +# re-invocation of an existing workflow_uuid attaches to it instead of +# restarting it (see workflow_id()'s docstring in services/durable.py) -- +# it waits for and returns whichever execution (the background recovery) +# actually finishes the workflow. The assertion that matters doesn't +# depend on that internal detail: the classify counter file has EXACTLY +# ONE line after both phases combined. + +_PIPELINE_RESUME_HANDLERS_MODULE = r""" +# Test-local SAPLING_FUNCTION_HANDLERS module for +# test_pipeline_crash_resume_via_workflow_id (tests/test_dbos_resume.py). +# Registers handlers for the three document-pipeline tasks process_document +# reaches when classification is a non-syllabus category (classifier, +# summary, concepts -- syllabus never runs). Shapes mirror +# agents/function_handlers_e2e.py's document-pipeline handlers. + +import os + +from pydantic_ai.messages import ModelResponse, ToolCallPart + +from agents._providers import register_function_handler + +_CLASSIFY_COUNTER = os.environ["DBOS_RESUME_CLASSIFY_COUNTER"] +_SUMMARY_MARKER = os.environ["DBOS_RESUME_SUMMARY_MARKER"] + + +def _structured(args): + def handler(messages, info): + return ModelResponse( + parts=[ToolCallPart(tool_name=info.output_tools[0].name, args=args)] + ) + + return handler + + +def _classify_handler(messages, info): + # Appended to on EVERY invocation -- the test asserts this file has + # exactly one line at the end, proving the already-checkpointed + # classify step was NOT re-run on the phase-2 retry. + with open(_CLASSIFY_COUNTER, "a") as f: + print("x", file=f) + return _structured({ + "category": "lecture_notes", + "is_syllabus": False, + "confidence": 0.95, + "rationale": "dbos resume-test fixture classification.", + })(messages, info) + + +def _summary_handler(messages, info): + # Simulates a worker crash mid-pipeline: on the FIRST invocation ever + # (marker file absent), write the marker and hard-exit before + # returning -- os._exit skips checkpointing this step's result, + # leaving the workflow row PENDING. On any later invocation (phase 2's + # resumed run), the marker exists, so this returns normally. + if not os.path.exists(_SUMMARY_MARKER): + open(_SUMMARY_MARKER, "w").close() + os._exit(42) + return _structured({ + "headline": "DBOS resume-test headline.", + "abstract": "DBOS resume-test abstract for the pipeline crash/resume proof.", + "key_points": [ + "Resume point one.", + "Resume point two.", + "Resume point three.", + ], + })(messages, info) + + +register_function_handler("classifier", _classify_handler) +register_function_handler("summary", _summary_handler) +register_function_handler( + "concepts", + _structured({ + "concepts": [ + { + "name": "Resume Concept", + "description": "dbos resume-test fixture concept.", + "importance": 0.5, + }, + ], + }), +) +""" + + +_PIPELINE_RESUME_HELPER = r""" +import asyncio +import json +import sys +from unittest.mock import AsyncMock + +import services.durable as durable + +assert durable.is_durable(), "DBOS_ENABLED=true but the shim did not activate" + +import agents.document as document_module +from agents.deps import SaplingDeps + +workflow_id = sys.argv[1] +graph_counter = sys.argv[2] + + +async def _fake_apply_concepts_to_graph(user_id, course_id, concept_names): + # File-backed, not just an AsyncMock call count: this test spans TWO + # separate subprocesses, and an in-memory mock count would not survive + # across them. + with open(graph_counter, "a") as f: + print("x", file=f) + return 0 + + +# The one real-Supabase side effect inside process_document -- stubbed the +# same way test_pipeline_identical_with_dbos_on above does, but with a +# file-backed side_effect (see _fake_apply_concepts_to_graph) instead of +# just an AsyncMock call count. +document_module.apply_concepts_to_graph = AsyncMock( + side_effect=_fake_apply_concepts_to_graph +) + +durable.init_dbos() + +deps = SaplingDeps( + user_id="dbos-resume-pipeline-user", + course_id="dbos-resume-pipeline-course", + supabase=None, + request_id="dbos-resume-pipeline", +) + +with durable.workflow_id(workflow_id): + result = asyncio.run(document_module.process_document( + "Deterministic fixture text for the DBOS pipeline crash/resume test.", + deps, + )) + +print("RESULT_JSON:" + json.dumps({ + "category": result.classification.category, + "graph_updated": result.graph_updated, +})) +""" + + +def test_pipeline_crash_resume_via_workflow_id(tmp_path): + """The real-pipeline resume proof via `workflow_id` -- the gap sections + 1 and 2 above leave open (see the module comment above section 1). + + Phase 1 crashes process_document mid-pipeline (inside the summary + step) via a scripted function-mode handler, inside + `durable.workflow_id(wfid)`. Phase 2 re-invokes process_document with + the SAME wfid and the same handlers module (the crash marker now + exists, so summary succeeds). The proof: the classify counter file has + EXACTLY ONE line after both phases -- the classify step, already + checkpointed before the crash, was not re-run on the phase-2 retry, + which is the resume-at-last-completed-step contract on the REAL + upload pipeline (not a toy workflow). + """ + handlers_module = tmp_path / "dbos_resume_pipeline_handlers.py" + handlers_module.write_text(_PIPELINE_RESUME_HANDLERS_MODULE) + + script = tmp_path / "pipeline_resume_helper.py" + script.write_text(_PIPELINE_RESUME_HELPER) + + workflow_id = "pipeline-resume-" + str(uuid.uuid4()) + classify_counter = tmp_path / "classify_counter.txt" + summary_marker = tmp_path / "summary_marker.txt" + graph_counter = tmp_path / "graph_counter.txt" + + env = _subprocess_env( + SAPLING_MODEL_MODE="function", + SAPLING_FUNCTION_HANDLERS="dbos_resume_pipeline_handlers", + PYTHONPATH=str(tmp_path) + os.pathsep + ".", + DBOS_RESUME_CLASSIFY_COUNTER=str(classify_counter), + DBOS_RESUME_SUMMARY_MARKER=str(summary_marker), + ) + args = (workflow_id, str(graph_counter)) + + crash = _run_helper(script, *args, env=env) + assert crash.returncode == 42, ( + "expected the crash phase to os._exit(42); stdout=" + repr(crash.stdout) + + " stderr=" + repr(crash.stderr) + ) + + resume = _run_helper(script, *args, env=env) + assert resume.returncode == 0, ( + "expected the resume phase to complete cleanly; stdout=" + repr(resume.stdout) + + " stderr=" + repr(resume.stderr) + ) + + # The resume-at-last-completed-step contract on the REAL pipeline: + # classify ran exactly once (during the crash phase) -- DBOS must NOT + # re-run an already-checkpointed step on the phase-2 same-id retry. + assert classify_counter.read_text().splitlines() == ["x"] + + # The _step_apply_graph checkpoint claim: the graph merge runs exactly + # once total, on whichever execution actually completes the workflow + # (phase 1 crashes before reaching it). File-backed so the count is + # correct across the two subprocesses. + assert graph_counter.read_text().splitlines() == ["x"] + + [result_line] = [ + line for line in resume.stdout.splitlines() if line.startswith("RESULT_JSON:") + ] + payload = json.loads(result_line[len("RESULT_JSON:"):]) + assert payload["category"] == "lecture_notes" + assert payload["graph_updated"] is False # _fake_apply_concepts_to_graph returns 0 diff --git a/backend/tests/test_documents_routes.py b/backend/tests/test_documents_routes.py index 7bf204e0..b8600c6c 100644 --- a/backend/tests/test_documents_routes.py +++ b/backend/tests/test_documents_routes.py @@ -1157,6 +1157,11 @@ def test_sync_replay_returns_same_doc_without_reprocessing(self): proc.assert_not_called() def test_streaming_replay_emits_done_without_reprocessing(self): + """#154/#132: a crash after the streaming `result` event (or any + client retry with the same X-Request-ID) must leave exactly ONE + consistent document -- the idempotent-replay short-circuit returns + the already-persisted row without re-running any agent AND without + writing a second `documents` row.""" existing = { "id": "doc-existing-stream", "user_id": "u1", @@ -1185,6 +1190,9 @@ def test_streaming_replay_emits_done_without_reprocessing(self): ) as r: assert r.status_code == 200 body = r.read() + # Zero new writes on replay: the idempotency short-circuit + # returns before _persist_document (or any other insert) runs. + t.return_value.insert.assert_not_called() # Orchestrator's classifier must not have been called on the replay. cls_run.assert_not_called() events = _parse_sse_stream(body) diff --git a/backend/tests/test_durable_shim.py b/backend/tests/test_durable_shim.py new file mode 100644 index 00000000..51704b49 --- /dev/null +++ b/backend/tests/test_durable_shim.py @@ -0,0 +1,297 @@ +"""Hermetic coverage for services/durable.py (ADR 0011 / #154). + +Runs WITHOUT the `dbos` package installed (it lives only in +requirements-durable.txt, never requirements.txt/requirements.lock — see +that file's header). Each test reloads `services.durable` under a +monkeypatched env + a fake `sys.modules['dbos']`, so it can flip through +every precondition combination in one process without the real package. + +The `durable_module` fixture guarantees the shared `services.durable` +module object is back in its pristine, flag-off, passthrough state before +the next test runs, regardless of what a given test set — other modules +(agents.document, and anything importing it) hold their OWN references to +whatever `workflow`/`step` looked like at THEIR import time, so this reload +only affects code that reads `services.durable.` fresh; but the +module's global state must not leak across tests in this file, or into any +other test file that imports `services.durable` directly. +""" + +from __future__ import annotations + +import asyncio +import importlib +import logging +import os +import sys +import types +from functools import wraps + +import pytest + +import services.durable as durable + +_ENV_KEYS = ("DBOS_ENABLED", "DBOS_DATABASE_URL") + + +@pytest.fixture +def durable_module(): + """Yield the live `services.durable` module; restore env, sys.modules, + and the module's own state to pristine passthrough in a finally block + so a test that blows up mid-way still can't leak into the rest of the + suite (which imports agents.document -> services.durable at collection + time).""" + saved_env = {k: os.environ.get(k) for k in _ENV_KEYS} + had_dbos = "dbos" in sys.modules + saved_dbos = sys.modules.get("dbos") + try: + yield durable + finally: + for key, value in saved_env.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + if had_dbos: + sys.modules["dbos"] = saved_dbos + else: + sys.modules.pop("dbos", None) + importlib.reload(durable) + + +def _make_stub_dbos_class(*, launch_raises: bool = False): + """A fresh fake `DBOS` class per call (class-level counters, so reusing + one across tests would leak call counts between them).""" + + class _StubDBOS: + construct_count = 0 + launch_count = 0 + destroy_count = 0 + constructed_with: dict | None = None + workflow_calls: list = [] + step_calls: list = [] + + def __init__(self, *, config): + type(self).construct_count += 1 + type(self).constructed_with = dict(config) + + @classmethod + def workflow(cls, *args, **kwargs): + def decorator(fn): + cls.workflow_calls.append(fn) + + @wraps(fn) + async def wrapper(*a, **k): + return await fn(*a, **k) + + return wrapper + + return decorator + + @classmethod + def step(cls, *args, **kwargs): + def decorator(fn): + cls.step_calls.append(fn) + + @wraps(fn) + async def wrapper(*a, **k): + return await fn(*a, **k) + + return wrapper + + return decorator + + @classmethod + def launch(cls): + cls.launch_count += 1 + if launch_raises: + raise RuntimeError("stub launch failure") + + @classmethod + def destroy(cls): + cls.destroy_count += 1 + + return _StubDBOS + + +def _install_stub_dbos(*, launch_raises: bool = False): + """Install a fake `dbos` module into sys.modules and return + (fake_module, stub_DBOS_class).""" + stub_cls = _make_stub_dbos_class(launch_raises=launch_raises) + fake_module = types.ModuleType("dbos") + fake_module.DBOS = stub_cls + fake_module.DBOSConfig = dict # only used as a type annotation at runtime + sys.modules["dbos"] = fake_module + return fake_module, stub_cls + + +def _install_broken_dbos(): + """Force `from dbos import DBOS` to raise ImportError on reload, the + same failure mode as `dbos` genuinely not being installed.""" + sys.modules["dbos"] = None + + +# ── Default (flag off): passthrough ───────────────────────────────────────── + +def test_default_flag_off_is_durable_false(durable_module): + os.environ.pop("DBOS_ENABLED", None) + os.environ.pop("DBOS_DATABASE_URL", None) + sys.modules.pop("dbos", None) + d = importlib.reload(durable_module) + assert d.is_durable() is False + + +def test_passthrough_workflow_preserves_return_value_and_args(durable_module): + os.environ.pop("DBOS_ENABLED", None) + d = importlib.reload(durable_module) + + calls = [] + + @d.workflow + async def wf(a, b, *, c=1): + calls.append((a, b, c)) + return a + b + c + + result = asyncio.run(wf(1, 2, c=3)) + assert result == 6 + assert calls == [(1, 2, 3)] + + +def test_passthrough_step_preserves_return_value_and_args(durable_module): + os.environ.pop("DBOS_ENABLED", None) + d = importlib.reload(durable_module) + + calls = [] + + @d.step + async def st(a, b, *, c=1): + calls.append((a, b, c)) + return a + b + c + + result = asyncio.run(st(4, 5, c=6)) + assert result == 15 + assert calls == [(4, 5, 6)] + + +def test_passthrough_workflow_propagates_exceptions(durable_module): + os.environ.pop("DBOS_ENABLED", None) + d = importlib.reload(durable_module) + + @d.workflow + async def wf(): + raise ValueError("boom") + + with pytest.raises(ValueError, match="boom"): + asyncio.run(wf()) + + +def test_passthrough_step_propagates_exceptions(durable_module): + os.environ.pop("DBOS_ENABLED", None) + d = importlib.reload(durable_module) + + @d.step + async def st(): + raise ValueError("boom") + + with pytest.raises(ValueError, match="boom"): + asyncio.run(st()) + + +def test_init_dbos_noop_when_flag_off(durable_module): + os.environ.pop("DBOS_ENABLED", None) + d = importlib.reload(durable_module) + assert d.init_dbos() is False + + +def test_shutdown_dbos_noop_when_flag_off(durable_module): + os.environ.pop("DBOS_ENABLED", None) + d = importlib.reload(durable_module) + d.shutdown_dbos() # must not raise + + +# ── DBOS_ENABLED=true but `dbos` fails to import ──────────────────────────── + +def test_enabled_but_dbos_import_fails_warns_then_init_dbos_raises(durable_module, caplog): + os.environ["DBOS_ENABLED"] = "true" + os.environ["DBOS_DATABASE_URL"] = "postgresql://x/y" + _install_broken_dbos() + with caplog.at_level(logging.WARNING, logger="services.durable"): + d = importlib.reload(durable_module) + # Import-time behavior is unchanged: warn + degrade decorators to + # passthrough (an import-time raise would take the whole app down + # before validate_config() runs — see the module docstring). + assert d.is_durable() is False + assert any("could not be imported" in r.message for r in caplog.records) + + # init_dbos() behavior changed (#154 review round, Finding B): this + # precondition failure now fails loud at startup instead of returning + # False and silently serving requests with zero durability. + with pytest.raises(RuntimeError, match="could not be imported"): + d.init_dbos() + + +# ── DBOS_ENABLED=true + working stub + DBOS_DATABASE_URL set: real mode ──── + +def test_enabled_with_working_stub_and_url_activates_durable(durable_module): + os.environ["DBOS_ENABLED"] = "true" + os.environ["DBOS_DATABASE_URL"] = "postgresql://u:p@localhost:5432/dbosdb" + _fake_module, stub_cls = _install_stub_dbos() + d = importlib.reload(durable_module) + assert d.is_durable() is True + + @d.workflow + async def wf(x): + return x + + @d.step + async def st(x): + return x + + # Decorators delegated to the stub instead of falling through to the + # identity passthrough. + assert len(stub_cls.workflow_calls) == 1 + assert len(stub_cls.step_calls) == 1 + assert asyncio.run(wf("hi")) == "hi" + assert asyncio.run(st("hi")) == "hi" + + assert d.init_dbos() is True + assert stub_cls.construct_count == 1 + assert stub_cls.launch_count == 1 + assert stub_cls.constructed_with == { + "name": "sapling", + "system_database_url": "postgresql://u:p@localhost:5432/dbosdb", + } + + d.shutdown_dbos() + assert stub_cls.destroy_count == 1 + + +# ── DBOS_ENABLED=true + working stub but NO DBOS_DATABASE_URL ─────────────── + +def test_enabled_without_database_url_warns_then_init_dbos_raises(durable_module, caplog): + os.environ["DBOS_ENABLED"] = "true" + os.environ.pop("DBOS_DATABASE_URL", None) + _install_stub_dbos() + with caplog.at_level(logging.WARNING, logger="services.durable"): + d = importlib.reload(durable_module) + # Import-time behavior is unchanged: warn + degrade to passthrough. + assert d.is_durable() is False + assert any("DBOS_DATABASE_URL" in r.message for r in caplog.records) + + # init_dbos() behavior changed (#154 review round, Finding B): raises + # instead of returning False, naming the missing precondition. + with pytest.raises(RuntimeError, match="DBOS_DATABASE_URL"): + d.init_dbos() + + +# ── DBOS_ENABLED=true + stub whose launch() raises: fail-loud contract ───── + +def test_init_dbos_raises_when_launch_fails(durable_module): + os.environ["DBOS_ENABLED"] = "true" + os.environ["DBOS_DATABASE_URL"] = "postgresql://u:p@localhost:5432/dbosdb" + _fake_module, stub_cls = _install_stub_dbos(launch_raises=True) + d = importlib.reload(durable_module) + assert d.is_durable() is True + + with pytest.raises(RuntimeError, match="stub launch failure"): + d.init_dbos() + assert stub_cls.launch_count == 1 diff --git a/docs/architecture.md b/docs/architecture.md index f31e336b..3e8b5fad 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -8,7 +8,7 @@ A single FastAPI app (`backend/main.py:87`) mounts every router under `/api/=2.28,<3`, + pinned to the version this update's verification ran against). Never + added to `requirements.txt`/`requirements.lock` — durability stays + opt-in and the hermetic suite runs with `dbos` NOT installed. +4. **Test coverage, previously zero, now covers both modes and resume:** + - `backend/tests/test_durable_shim.py` — hermetic, runs WITHOUT `dbos` + installed. Reloads the shim under every precondition combination + (flag off, flag on + import failure, flag on + no URL, flag on + + working stub, stub whose `launch()` raises) and asserts passthrough + byte-compatibility (return values, args/kwargs, exceptions) plus the + fail-loud contract. + - `backend/tests/test_dbos_resume.py` — opt-in + (`RUN_DBOS_RESUME=1`, skipped otherwise; also skips if `dbos` isn't + installed). A real subprocess crash/resume proof against a live + Postgres: `test_shim_resume_minimal_workflow` crashes a 2-step + workflow mid-step-2 (`os._exit(42)`), resumes it in a fresh process, + and asserts the already-checkpointed step did NOT re-run while the + crashed step ran to completion exactly once. + `test_pipeline_identical_with_dbos_on` runs + `agents.document.process_document` for real under DBOS + (`SAPLING_MODEL_MODE=function`) and asserts its output matches the + function-mode constants byte-for-byte — proof that wrapping the + pipeline in `@durable_workflow`/`@durable_step` changes durability, + not behavior. `test_pipeline_crash_resume_via_workflow_id` is the + REAL-pipeline resume proof the other two don't cover (see point 6 + below): it crashes `process_document` mid-pipeline via a scripted + function-mode handler, then re-invokes it in a second process wrapped + in the SAME `services.durable.workflow_id(...)`, and asserts the + already-checkpointed classify step ran exactly once total across both + processes. The first two subprocess tests were run against a real + `dbos==2.28.0` + a throwaway Postgres during the original #154 update + and passed; the third was added in this update's review round and is + verified by source-reading only here (see its own docstring for the + dbos facts cited) — it needs the same live-Postgres run to confirm. + See each file's docstrings for exactly what it does and does not + prove. +5. **Activation procedure, corrected against the installed + `dbos==2.28.0`** (verified directly against the package source in a + scratch venv; see `backend/services/durable.py::init_dbos`'s docstring + for the file/line citations): + + a. `pip install -r backend/requirements-durable.txt`. + b. Provision a reachable Postgres for DBOS's own system database (the + SAME Postgres instance Supabase/`SUPABASE_DB_URL` uses is fine — DBOS + creates its own `dbos` schema there; see "System schema" below). + c. Set `DBOS_ENABLED=true` and `DBOS_DATABASE_URL=postgres://...`. + d. Restart the FastAPI workers. `init_dbos()` runs in `_lifespan`; + ANY failure constructing/launching DBOS makes the app FAIL TO START + (same fail-loud posture as #174's `validate_config()`) rather than + boot into a silently non-durable state. + e. ~~Run DBOS migrations (`dbos migrate`)~~ — **this step from the + 2026-05-04 procedure below is STALE and is corrected here**: + `DBOS.launch()` runs the system-database migrations ITSELF + (verified in `dbos/_dbos.py::DBOS._launch`, which calls + `self._sys_db.run_migrations()` before anything else happens). No + separate `dbos migrate` step is needed for Sapling's setup, where the + app's DB role has ordinary DDL rights. (`dbos.run_dbos_database_ + migrations()` exists as a standalone helper for the case where it + doesn't — not exercised here.) + f. Confirm activation: `services.durable.is_durable()` returns `True`; + `init_dbos()` logs `"durable execution ACTIVE (DBOS launched)"` at + startup (or `"durable execution off — passthrough decorators"` in the + default, off, mode) — both plain `logger.info` calls, so both are + captured by #119's Logfire app-logging instrumentation with no + further wiring. +6. **`workflow_id` wiring — the review-round fix that makes resume real for + the product, not just the mechanism.** Everything above (entrypoint, + migrations, background recovery) made DBOS's crash/resume machinery + reachable, but nothing tied a specific upload attempt's workflow to a + RETRY of that same attempt — every invocation of `process_document` got + an auto-generated workflow id, so a client retry (same `X-Request-ID`) + started an unrelated new workflow and gained nothing from whatever the + crashed attempt had already checkpointed. Fixed: + - `services.durable.workflow_id(wfid)` — a context manager, `SetWorkflowID(wfid)` + when durable else `contextlib.nullcontext()` — pins the DBOS workflow + id for the invocation started inside it (see its own docstring for the + dbos==2.28.0 facts verified: `SetWorkflowID`'s import path and + same-id-reattach semantics for both PENDING and SUCCESS rows). + - `routes/documents.py`'s `/upload/sync` wraps its `process_document` + call in `workflow_id(f"doc:{user_id}:{request_id}")` — scoped to + `user_id` + `request_id`, not `request_id` alone, because + `X-Request-ID` is client-supplied and an unscoped id would let one + user's replay attach to another user's workflow. + - `agents/document.py`'s graph merge is now its own durable step + (`_step_apply_graph`), not a bare call inside `process_document` — so + a resumed workflow doesn't repeat the merge either. + + **The corrected product-level crash semantic** (replacing any earlier + text in this module/ADR implying an in-flight upload transparently + survives a crash on its own): nothing resumes for the ORIGINAL caller — + their HTTP connection to the crashed worker is already gone. What + changed is what a CLIENT RETRY (same `X-Request-ID`) gets: the route's + existing idempotency cache (`_existing_doc_by_request_id`, ADR 0009) + still serves an already-PERSISTED document instantly; for an upload that + crashed before persistence, the retry now ATTACHES to the SAME DBOS + workflow instead of starting a fresh one, and DBOS resumes it at the + last completed step rather than re-running every agent call. If no + retry ever arrives, `DBOS.launch()`'s background auto-recovery (point 5 + above) still completes the abandoned workflow on its own — a LATER + same-id retry then receives that already-recorded result. + + **Picklability constraint, newly load-bearing.** DBOS's default + serializer records workflow inputs with `pickle` + (`dbos/_serialization.py::DefaultSerializer.serialize`) — so every + argument to a `@durable_workflow`/`@durable_step` call must stay + picklable once DBOS is on. `process_document(text, deps)` passes a + `SaplingDeps` built with `supabase=None` and otherwise plain + str/None/list fields (`routes/documents.py`'s `/upload/sync` handler) — + that is precisely what keeps it recordable today. A future field on + `SaplingDeps` (or a new argument to a `_step_*`) that holds something + unpicklable (a live client, a lock, a generator) would break ONLY the + flag-on path, silently, at call time — worth remembering before adding + fields to deps used on this path. + +**Resume monitoring.** Beyond the startup INFO log above, `DBOS.launch()` +logs its own recovery activity (`"Recovering N workflows from application +version ..."` / `"No workflows to recover..."`) through the same app- +logging path, so Logfire has it automatically. For a direct look, DBOS's +system database is its own schema (defaults to `dbos`, confirmed from the +installed package's migration DDL in `dbos/_migration.py`) with a +`workflow_status` table (`workflow_uuid`, `status`, `name`, `executor_id`, +`recovery_attempts`, `created_at`/`updated_at` as epoch-ms bigints, …). A +sample query to list resumed/pending workflows: + +```sql +select workflow_uuid, name, status, executor_id, recovery_attempts, + to_timestamp(created_at / 1000.0) as created_at, + to_timestamp(updated_at / 1000.0) as updated_at +from dbos.workflow_status +where status in ('PENDING', 'ENQUEUED') + or recovery_attempts > 0 +order by created_at desc; +``` + +`dbos.*` is entirely DBOS-managed — created and migrated by +`DBOS.launch()` itself — and is deliberately OUTSIDE +`backend/db/migrations/`; it is not a Sapling schema, and `python -m +db.migrate` never touches it. + +**Streaming-route asymmetry, reaffirmed.** Still intentional (unchanged +from the 2026-05-04 text below): the streaming `POST /api/documents/upload` +route stays non-durable, with X-Request-ID replay as its crash semantic. +That semantic is now more tightly pinned than when this ADR was first +written: PR #464 closed the "#132 remainder" — one `result` SSE event per +streamed upload, ever, with post-result persistence failures always +terminating in `error:failed` + `status:done` rather than a legacy-fallback +second run (routes/documents.py's post-roll comment block, "#154 builds on +this structure — keep persistence in the post-roll, after the result +event"). `tests/test_documents_routes.py::TestUploadIdempotency:: +test_streaming_replay_emits_done_without_reprocessing` now asserts (not +just implies) BOTH halves of the #132/#154 acceptance criterion — a crash +after the `result` event leaves exactly one consistent document — directly: +zero agent invocations (`classifier_agent.run` unmocked-but-spied, +`assert_not_called()`) and zero new `documents` inserts +(`t.return_value.insert.assert_not_called()`) on an X-Request-ID replay. + +**Still deferred, unchanged:** production incident-driven validation (this +ADR's original "when to revisit" trigger #1 — a real lost mid-flight +upload — hasn't happened, so `DBOS_ENABLED` stays off in every deployed +environment); DBOS Conductor / multi-executor recovery (Sapling runs a +single executor, `executor_id="local"`, so cross-executor recovery +semantics are unexercised by the resume test above); and coexistence with +ADR 0010's two-phase upload if that ever ships (still an open sequencing +question — see that section, unchanged, further down). + ## Update (2026-05-04) `backend/services/durable.py` shipped with `@workflow` and `@step`