Persist REST /api/ask via the shared incremental path

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."""

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
 

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