From 4a452d82571c07afd83a8d681db01e28de8c08c1 Mon Sep 17 00:00:00 2001 From: Michael Sitarzewski Date: Sun, 21 Jun 2026 22:21:49 -0500 Subject: [PATCH] feat(api): persist REST /api/ask via the shared incremental path POST /api/ask now persists the full consensus debate (all rounds, contributions, citations, decision, overview, follow-ups, usage) through the same IncrementalPersister used by the CLI and WebSocket, replacing the lite path that saved only the final decision. A crash mid-request now leaves a real partial thread instead of nothing. _run_consensus gains an additive on_thread_created callback so REST can surface the thread ID the persister creates up front, without changing the 8-tuple return (no impact on the other callers). The now-dead _persist_result helper is removed. 1663 Python tests pass, mypy + ruff clean. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01EkrekgzMAQko92UkjnXhHL --- src/duh/api/routes/ask.py | 59 ++++-------------------- src/duh/cli/app.py | 5 ++ tests/unit/test_api_ask.py | 93 +++++++++++++++++--------------------- 3 files changed, 55 insertions(+), 102 deletions(-) diff --git a/src/duh/api/routes/ask.py b/src/duh/api/routes/ask.py index 26416bf..62cf170 100644 --- a/src/duh/api/routes/ask.py +++ b/src/duh/api/routes/ask.py @@ -111,6 +111,11 @@ async def _handle_consensus( # type: ignore[no-untyped-def] from duh.cli.app import _run_consensus use_native_search = config.tools.enabled and config.tools.web_search.native + + # Persist the full debate via the shared incremental path (same as CLI/WS), + # capturing the thread ID it creates up front. Replaces the old lite path + # that only saved the final decision. + created: dict[str, str] = {} ( decision, confidence, @@ -129,26 +134,10 @@ async def _handle_consensus( # type: ignore[no-untyped-def] proposer_override=body.proposer, challengers_override=body.challengers, web_search=use_native_search, + db_factory=db_factory, + on_thread_created=lambda tid: created.__setitem__("id", tid), ) - - thread_id: str | None = None - if db_factory is not None: - try: - thread_id = await _persist_result( - db_factory, - body.question, - decision, - confidence, - dissent, - rigor=rigor, - usage={ - "input_tokens": pm.total_input_tokens, - "output_tokens": pm.total_output_tokens, - "cost_usd": cost, - }, - ) - except Exception: - logger.exception("Failed to persist consensus thread") + thread_id: str | None = created.get("id") return AskResponse( decision=decision, @@ -244,38 +233,6 @@ async def _handle_decompose(body: AskRequest, config, pm) -> AskResponse: # typ ) -async def _persist_result( - db_factory: object, - question: str, - decision: str, - confidence: float, - dissent: str | None, - *, - rigor: float = 0.0, - usage: dict[str, float] | None = None, -) -> str: - """Persist a consensus result to the database. - - Returns the new thread ID. - """ - import json as _json - - from duh.memory.repository import MemoryRepository - - async with db_factory() as session: # type: ignore[operator] - repo = MemoryRepository(session) - thread = await repo.create_thread(question) - thread.status = "complete" - if usage: - thread.usage_json = _json.dumps(usage) - turn = await repo.create_turn(thread.id, 1, "COMMIT") - await repo.save_decision( - turn.id, thread.id, decision, confidence, rigor=rigor, dissent=dissent - ) - await session.commit() - return str(thread.id) - - @router.post("/refine", response_model=RefineResponse) async def refine(body: RefineRequest, request: Request) -> RefineResponse: """Analyze a question for ambiguity and suggest clarifications.""" diff --git a/src/duh/cli/app.py b/src/duh/cli/app.py index 1f9fce7..92672a3 100644 --- a/src/duh/cli/app.py +++ b/src/duh/cli/app.py @@ -21,6 +21,8 @@ from duh.core.errors import ConfigError, DuhError if TYPE_CHECKING: + from collections.abc import Callable + from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker from duh.cli.display import ConsensusDisplay @@ -263,6 +265,7 @@ async def _run_consensus( challenger_count: int | None = None, web_search: bool = False, db_factory: async_sessionmaker[AsyncSession] | None = None, + on_thread_created: Callable[[str], None] | None = None, ) -> tuple[ str, float, @@ -313,6 +316,8 @@ async def _run_consensus( persister = IncrementalPersister(db_factory, question) try: ctx.thread_id = await persister.start() + if on_thread_created is not None: + on_thread_created(ctx.thread_id) except Exception: import logging as _logging diff --git a/tests/unit/test_api_ask.py b/tests/unit/test_api_ask.py index d42e1c5..9b53982 100644 --- a/tests/unit/test_api_ask.py +++ b/tests/unit/test_api_ask.py @@ -247,54 +247,45 @@ async def test_decompose_protocol(self) -> None: mock_fn.assert_called_once() -# ── TestPersistResult ───────────────────────────────────────── - - -class TestPersistResult: - async def _factory(self) -> async_sessionmaker: - engine = create_async_engine("sqlite+aiosqlite://") - async with engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - return async_sessionmaker(engine, expire_on_commit=False) - - async def test_persists_usage_json(self) -> None: - """_persist_result records run-level usage on the thread.""" - from duh.api.routes.ask import _persist_result - from duh.memory.repository import MemoryRepository - - factory = await self._factory() - tid = await _persist_result( - factory, - "Which database?", - "Use PostgreSQL", - 0.9, - None, - rigor=1.0, - usage={"input_tokens": 6257, "output_tokens": 2945, "cost_usd": 0.037}, - ) - - async with factory() as session: - repo = MemoryRepository(session) - thread = await repo.get_thread(tid) - assert thread is not None - assert thread.usage_json is not None - import json as _json - - stored = _json.loads(thread.usage_json) - assert stored["input_tokens"] == 6257 - assert stored["output_tokens"] == 2945 - assert stored["cost_usd"] == 0.037 - - async def test_no_usage_leaves_column_null(self) -> None: - """Omitting usage leaves usage_json unset (backward compatible).""" - from duh.api.routes.ask import _persist_result - from duh.memory.repository import MemoryRepository - - factory = await self._factory() - tid = await _persist_result(factory, "Q", "A", 0.8, None) - - async with factory() as session: - repo = MemoryRepository(session) - thread = await repo.get_thread(tid) - assert thread is not None - assert thread.usage_json is None +# ── TestUnifiedPersistence ──────────────────────────────────── + + +class TestUnifiedPersistence: + """REST persists via the shared incremental path (not a lite path).""" + + async def test_forwards_db_factory_and_returns_thread_id(self) -> None: + """REST hands its db_factory to _run_consensus and returns the + thread ID the incremental persister creates up front.""" + client, _ = await _make_app() + seen: dict[str, object] = {} + + async def fake_run_consensus(question, config, pm, *args, **kwargs): # type: ignore[no-untyped-def] + seen["db_factory"] = kwargs.get("db_factory") + cb = kwargs.get("on_thread_created") + if cb is not None: + cb("thread-xyz") + return ("Decision", 0.9, 1.0, None, 0.01, None, [], []) + + with patch("duh.cli.app._run_consensus", side_effect=fake_run_consensus): + resp = client.post("/api/ask", json={"question": "Which DB?"}) + + assert resp.status_code == 200 + # The factory was forwarded so persistence happens inside the run. + assert seen["db_factory"] is not None + # The thread ID surfaced via the on_thread_created callback. + assert resp.json()["thread_id"] == "thread-xyz" + + async def test_thread_id_null_when_no_thread_created(self) -> None: + """If the persister never fires its callback (e.g. start() failed), + the response reports a null thread_id rather than erroring.""" + client, _ = await _make_app() + + async def fake_run_consensus(question, config, pm, *args, **kwargs): # type: ignore[no-untyped-def] + # Simulate persistence not producing a thread: callback not called. + return ("Decision", 0.9, 1.0, None, 0.01, None, [], []) + + with patch("duh.cli.app._run_consensus", side_effect=fake_run_consensus): + resp = client.post("/api/ask", json={"question": "Which DB?"}) + + assert resp.status_code == 200 + assert resp.json()["thread_id"] is None