From b7b4dcc0dfeef7775bc93284452047bfc6fe596b Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Mon, 18 May 2026 20:29:21 -0700 Subject: [PATCH] fix(judge): UPSERT intent_verdicts so llm_fallback upgrades land MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Async LLM-tier "llm_fallback" verdicts (judge.py:1073, judge.py:1131 via _deliver_fallbacks) deliberately reuse the heuristic verdict's ``verdict_id`` so the row gets "upgraded in place" from heuristic → llm_fallback when the LLM judge times out, is cancelled, or returns no content. The consumer ``_persist_intent_verdict`` was doing a plain INSERT via ``create_intent_verdict``, hitting the ``intent_verdicts_pkey`` constraint on every llm_fallback delivery. Postgres logged the duplicate-key error; the application try/except swallowed it at log.debug — so the row never actually got upgraded and the LLM judge's annotation ("(LLM judge did not return a verdict)") was lost. The collision rate exploded on stable/1.5 smoke tests because PR #527 (just merged) added two new heuristic-INSERT paths in the auto-approve early-return branches of ``approve_tools`` — previously those branches dropped heuristic verdicts on the floor, leaving no row for the fallback to collide with. Fix: - New ``upsert_intent_verdict`` method on the storage protocol + sqlite + postgres impls, using dialect-specific ``insert(...).on_conflict_do_update(index_elements=["verdict_id"], set_={...})``. Set_ clause updates ONLY the three fields that genuinely change between heuristic and llm_fallback: ``tier``, ``reasoning``, ``judge_model``. - Every other column is excluded from set_: identity columns (verdict_id, ws_id, call_id, func_name, func_args), carried- verbatim columns (intent_summary, risk_level, confidence, recommendation, evidence, latency_ms), and ``user_decision``. - ``user_decision`` exclusion is load-bearing: ``IntentVerdict .to_dict()`` doesn't project it, so a fallback verdict reaching ``_persist_intent_verdict`` carries the kwarg's ``"pending"`` default. If the operator already resolved the approval between heuristic INSERT and fallback delivery, the row's user_decision has been stamped to ``"approved"``/``"denied"``/``"timeout"`` (or an auto-approve reason at heuristic-INSERT time per PR #527). Including ``user_decision`` in set_ would silently clobber that back to ``"pending"``. - ``_persist_intent_verdict`` switched from ``create_*`` to ``upsert_*``. Bulk path ``create_intent_verdicts_bulk`` stays as plain INSERT — every heuristic ``verdict_id`` is freshly minted in ``judge.evaluate`` so in-turn dups can't happen. The inverse race (daemon-judge verdict lands BEFORE the bulk write) IS reachable today but its observable behavior is unchanged by the per-row UPSERT switch; documented at the bulk site for a future hardening pass. Test coverage: - TestIntentVerdictUpsert × 4 — fresh-id insert, conflict-upgrade, user_decision preservation across heuristic→approved→fallback, identity + carried-field preservation. - Existing tests in test_session_ui_base.py updated to mock the new upsert method instead of create_intent_verdict. --- tests/test_judge_storage.py | 117 ++++++++++++++++++++++++++ tests/test_session_ui_base.py | 10 +-- turnstone/core/session_ui_base.py | 41 +++++++-- turnstone/core/storage/_postgresql.py | 55 ++++++++++++ turnstone/core/storage/_protocol.py | 57 +++++++++++++ turnstone/core/storage/_sqlite.py | 55 ++++++++++++ 6 files changed, 322 insertions(+), 13 deletions(-) diff --git a/tests/test_judge_storage.py b/tests/test_judge_storage.py index a513b3be..ff6f58b5 100644 --- a/tests/test_judge_storage.py +++ b/tests/test_judge_storage.py @@ -118,6 +118,123 @@ class TestIntentVerdictCRUD: assert ok is False +class TestIntentVerdictUpsert: + """``upsert_intent_verdict`` — the LLM-tier-aware persistence path. + + Backs the heuristic → llm_fallback "upgrade in place" pattern. + The async judge's fallback verdicts deliberately reuse the + heuristic ``verdict_id``; a plain INSERT would collide on the + PK and the upgrade would be lost to a silently-swallowed + exception (Postgres logged ``intent_verdicts_pkey`` violations + for every fallback delivery on stable/1.5 smoke tests). + """ + + def test_upsert_on_fresh_id_inserts(self, db): + """No conflict — behaves like a regular INSERT.""" + db.upsert_intent_verdict(**_make_verdict_kwargs()) + v = db.get_intent_verdict("v_001") + assert v is not None + assert v["tier"] == "heuristic" + assert v["user_decision"] == "pending" + + def test_upsert_on_conflict_upgrades_tier_reasoning_judge_model(self, db): + """On PK conflict: tier, reasoning, judge_model update — every + other field is preserved. Mirrors what the judge emits when + promoting heuristic → llm_fallback.""" + db.upsert_intent_verdict( + **_make_verdict_kwargs( + tier="heuristic", + reasoning="initial heuristic reasoning", + judge_model="", + ) + ) + db.upsert_intent_verdict( + **_make_verdict_kwargs( + tier="llm_fallback", + reasoning="initial heuristic reasoning (LLM judge did not return a verdict)", + judge_model="gpt-5-judge", + ) + ) + v = db.get_intent_verdict("v_001") + assert v is not None + # The three fields that should change. + assert v["tier"] == "llm_fallback" + assert "LLM judge did not return" in v["reasoning"] + assert v["judge_model"] == "gpt-5-judge" + + def test_upsert_on_conflict_preserves_user_decision(self, db): + """LOAD-BEARING: a manually-resolved approval (user_decision= + ``"approved"``) or auto-approve-stamped row (user_decision= + ``"policy"``/``"blanket"``/etc.) must NOT be clobbered back to + ``"pending"`` when the late LLM-fallback verdict lands. + ``IntentVerdict.to_dict()`` doesn't project user_decision, so + the upsert's defaulted ``"pending"`` would silently overwrite + the real value if user_decision were in the on-conflict + SET clause.""" + db.upsert_intent_verdict(**_make_verdict_kwargs()) + ok = db.update_intent_verdict("v_001", user_decision="approved") + assert ok is True + # Simulate the late LLM-fallback delivery — same verdict_id, + # default user_decision (the IntentVerdict.to_dict() shape). + db.upsert_intent_verdict( + **_make_verdict_kwargs( + tier="llm_fallback", + reasoning="extended (LLM judge did not return a verdict)", + judge_model="gpt-5-judge", + ) + ) + v = db.get_intent_verdict("v_001") + assert v is not None + assert v["user_decision"] == "approved" # NOT clobbered to "pending" + assert v["tier"] == "llm_fallback" # but the upgrade did land + + def test_upsert_on_conflict_preserves_identity_and_carried_fields(self, db): + """Identity columns (ws_id, call_id, func_name, func_args) and + carried-verbatim columns (intent_summary, risk_level, + confidence, recommendation, evidence, latency_ms) are + excluded from the on-conflict SET — verify they aren't + changed even when the second upsert passes different values + (defensive against a future judge bug that ships divergent + carried fields).""" + db.upsert_intent_verdict(**_make_verdict_kwargs()) + db.upsert_intent_verdict( + **_make_verdict_kwargs( + # Same verdict_id (conflict trigger), divergent everything else. + ws_id="ws-different", + call_id="tc_different", + func_name="bash_v2", + func_args='{"command":"rm -rf /"}', + intent_summary="totally different summary", + risk_level="critical", + confidence=0.0, + recommendation="deny", + evidence='["dangerous"]', + latency_ms=99999, + # The three fields that DO update. + tier="llm_fallback", + reasoning="upgraded reasoning", + judge_model="judge-v2", + ) + ) + v = db.get_intent_verdict("v_001") + assert v is not None + # All preserved from the first upsert (identity + carried). + assert v["ws_id"] == "ws-abc" + assert v["call_id"] == "tc_001" + assert v["func_name"] == "bash" + assert v["func_args"] == '{"command":"echo hello"}' + assert v["intent_summary"] == "Echo a greeting to stdout" + assert v["risk_level"] == "low" + assert v["confidence"] == 0.85 + assert v["recommendation"] == "approve" + assert v["evidence"] == '["The command only prints text."]' + assert v["latency_ms"] == 2 + # Only the three updated. + assert v["tier"] == "llm_fallback" + assert v["reasoning"] == "upgraded reasoning" + assert v["judge_model"] == "judge-v2" + + # --------------------------------------------------------------------------- # Bulk insert # --------------------------------------------------------------------------- diff --git a/tests/test_session_ui_base.py b/tests/test_session_ui_base.py index f90d87b7..0d14ee67 100644 --- a/tests/test_session_ui_base.py +++ b/tests/test_session_ui_base.py @@ -167,8 +167,8 @@ def test_on_intent_verdict_persists_verdict_row() -> None: } with _patch_get_storage(storage): ui.on_intent_verdict(verdict) - storage.create_intent_verdict.assert_called_once() - kwargs = storage.create_intent_verdict.call_args.kwargs + storage.upsert_intent_verdict.assert_called_once() + kwargs = storage.upsert_intent_verdict.call_args.kwargs assert kwargs["verdict_id"] == "v1" assert kwargs["ws_id"] == "ws-1" assert kwargs["call_id"] == "c1" @@ -426,8 +426,8 @@ def test_on_intent_verdict_consumes_auto_approve_reason() -> None: ui._auto_approve_reasons["c-x"] = ("auto_approve_tools", 0.0) with _patch_get_storage(storage): ui.on_intent_verdict({"verdict_id": "v-x", "call_id": "c-x"}) - storage.create_intent_verdict.assert_called_once() - kwargs = storage.create_intent_verdict.call_args.kwargs + storage.upsert_intent_verdict.assert_called_once() + kwargs = storage.upsert_intent_verdict.call_args.kwargs assert kwargs["user_decision"] == "auto_approve_tools" # Consumed on read so the same call_id can't double-stamp later. assert "c-x" not in ui._auto_approve_reasons @@ -465,7 +465,7 @@ def test_on_intent_verdict_auto_reason_survives_resolve_cycle() -> None: # The auto verdict's INSERT carried the policy reason. insert_calls = { c.kwargs["verdict_id"]: c.kwargs["user_decision"] - for c in storage.create_intent_verdict.call_args_list + for c in storage.upsert_intent_verdict.call_args_list } assert insert_calls["v-auto"] == "policy" assert insert_calls["v-pending"] == "pending" diff --git a/turnstone/core/session_ui_base.py b/turnstone/core/session_ui_base.py index 31480be0..c36408e4 100644 --- a/turnstone/core/session_ui_base.py +++ b/turnstone/core/session_ui_base.py @@ -909,6 +909,25 @@ class SessionUIBase: } for v in verdicts ] + # Plain INSERT (not UPSERT) at the bulk site. The race + # where a daemon-judge verdict lands BEFORE this bulk + # write IS reachable today: ``_evaluate_intent`` + # (session.py) spawns the daemon thread before + # ``approve_tools`` is called, and the daemon's first + # emission (heuristic-only short batch, fast LLM response, + # or cancel-event ``_deliver_fallbacks`` from judge.py) + # can fire ``_persist_intent_verdict`` before this bulk + # INSERT runs. Outcome of that race is unchanged by the + # per-row UPSERT switch: the bulk INSERT statement aborts + # on PK collision regardless of whether the colliding row + # was planted by INSERT or UPSERT, and the wrapping + # ``try/except`` swallows it. Race A (daemon fires AFTER + # bulk) IS improved by the fix: heuristic→llm_fallback + # upgrade-in-place now lands. Future bulk-side hardening + # (``ON CONFLICT DO NOTHING``) would preserve the OTHER + # rows in the batch when one collides, but would keep the + # daemon's ``tier`` ("llm"/"llm_fallback") for the + # colliding row instead of the bulk's heuristic stamp. storage.create_intent_verdicts_bulk(rows) except Exception: log.debug("Failed to bulk-persist intent verdicts", exc_info=True) @@ -919,15 +938,21 @@ class SessionUIBase: *, default_tier: str = "llm", ) -> None: - """Persist an intent-judge verdict row. + """Persist an intent-judge verdict row via UPSERT. - Used by both the async LLM-tier path (``on_intent_verdict``, - default tier ``"llm"``) and the synchronous heuristic-tier - path (``approve_tools``, caller passes ``default_tier="heuristic"``). + Used by the async LLM-tier path (``on_intent_verdict``, + default tier ``"llm"``). Routes through ``upsert_intent_verdict`` + because ``tier="llm_fallback"`` verdicts deliberately reuse the + heuristic verdict's ``verdict_id`` (see ``judge.py`` — + ``_deliver_fallbacks`` and the in-loop fallback path) + so the row gets "upgraded in place" from heuristic → + llm_fallback. A plain INSERT would collide on the PK and the + upgrade would be lost to a silently-swallowed exception. ``default_tier`` only matters when the verdict dict doesn't - already carry a ``tier`` key — both real producers always set it, - but the fallback is the right call-site label so a malformed - verdict still lands on the correct row classification. + already carry a ``tier`` key — both real producers always set + it, but the fallback is the right call-site label so a + malformed verdict still lands on the correct row + classification. """ try: from turnstone.core.storage._registry import get_storage @@ -935,7 +960,7 @@ class SessionUIBase: storage = get_storage() if storage is None: return - storage.create_intent_verdict( + storage.upsert_intent_verdict( verdict_id=verdict.get("verdict_id", ""), ws_id=self.ws_id, call_id=verdict.get("call_id", ""), diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py index 82728bd7..c1b9c9ee 100644 --- a/turnstone/core/storage/_postgresql.py +++ b/turnstone/core/storage/_postgresql.py @@ -3389,6 +3389,61 @@ class PostgreSQLBackend: ) conn.commit() + def upsert_intent_verdict( + self, + verdict_id: str, + ws_id: str, + call_id: str, + func_name: str, + func_args: str, + intent_summary: str, + risk_level: str, + confidence: float, + recommendation: str, + reasoning: str, + evidence: str, + tier: str, + judge_model: str, + latency_ms: int, + user_decision: str = "pending", + ) -> None: + from sqlalchemy.dialects.postgresql import insert as pg_insert + + now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + stmt = pg_insert(intent_verdicts).values( + verdict_id=verdict_id, + ws_id=ws_id, + call_id=call_id, + func_name=func_name, + func_args=func_args, + intent_summary=intent_summary, + risk_level=risk_level, + confidence=confidence, + recommendation=recommendation, + reasoning=reasoning, + evidence=evidence, + tier=tier, + judge_model=judge_model, + latency_ms=latency_ms, + user_decision=user_decision, + created=now, + ) + # On verdict_id conflict, update only the three fields that + # genuinely change between heuristic and llm_fallback. See the + # protocol docstring for the full exclusion rationale — + # ``user_decision`` exclusion in particular is load-bearing. + stmt = stmt.on_conflict_do_update( + index_elements=[intent_verdicts.c.verdict_id], + set_={ + "tier": tier, + "reasoning": reasoning, + "judge_model": judge_model, + }, + ) + with self._conn() as conn: + conn.execute(stmt) + conn.commit() + def create_intent_verdicts_bulk(self, verdicts: list[dict[str, Any]]) -> None: if not verdicts: return diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py index 6c59f4c8..238865bb 100644 --- a/turnstone/core/storage/_protocol.py +++ b/turnstone/core/storage/_protocol.py @@ -1623,6 +1623,63 @@ class StorageBackend(Protocol): """ ... + def upsert_intent_verdict( + self, + verdict_id: str, + ws_id: str, + call_id: str, + func_name: str, + func_args: str, + intent_summary: str, + risk_level: str, + confidence: float, + recommendation: str, + reasoning: str, + evidence: str, + tier: str, + judge_model: str, + latency_ms: int, + user_decision: str = "pending", + ) -> None: + """INSERT a verdict row, or UPDATE the judge-output fields on conflict. + + Async LLM judge verdicts with ``tier="llm_fallback"`` deliberately + reuse the heuristic verdict's ``verdict_id`` so the row gets + "upgraded in place" from heuristic → fallback when the LLM tier + doesn't return a real verdict (timeout / cancelled / no-content). + A plain INSERT collides on ``intent_verdicts_pkey``; this method + ``ON CONFLICT (verdict_id) DO UPDATE`` updates only the columns + that genuinely change between the two tiers: + + - ``tier`` (the upgrade itself) + - ``reasoning`` (gets " (LLM judge did not return a verdict)" appended) + - ``judge_model`` (heuristic carries "", fallback carries the model) + + Every other column is EXCLUDED from the on-conflict SET clause: + + - Identity columns (``verdict_id``, ``ws_id``, ``call_id``, + ``func_name``, ``func_args``) — already the same row. + - Carried-verbatim columns (``intent_summary``, ``risk_level``, + ``confidence``, ``recommendation``, ``evidence``, ``latency_ms``) — + the fallback copies them from the heuristic verdict; updating + would be a no-op. + - ``user_decision`` — LOAD-BEARING exclusion. ``IntentVerdict.to_dict()`` + doesn't project it, so a fallback verdict reaching this layer + defaults the kwarg to ``"pending"``. If the operator already + resolved the approval between heuristic INSERT and fallback + fire, the row's ``user_decision`` was already updated to + ``"approved"``/``"denied"``/``"timeout"`` (or stamped to an + auto-approve reason at heuristic-INSERT time). Clobbering it + back to ``"pending"`` would undo that. + - ``created`` — preserve the original timestamp. + + Used by :meth:`SessionUIBase._persist_intent_verdict` for every + async LLM-tier delivery; the synchronous heuristic-bulk path + (:meth:`create_intent_verdicts_bulk`) stays as plain INSERT + since each heuristic UUID is freshly generated per turn. + """ + ... + def create_intent_verdicts_bulk(self, verdicts: list[dict[str, Any]]) -> None: """Insert many intent_verdict rows in one transaction. diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py index c5e33892..46aee814 100644 --- a/turnstone/core/storage/_sqlite.py +++ b/turnstone/core/storage/_sqlite.py @@ -3551,6 +3551,61 @@ class SQLiteBackend: ) conn.commit() + def upsert_intent_verdict( + self, + verdict_id: str, + ws_id: str, + call_id: str, + func_name: str, + func_args: str, + intent_summary: str, + risk_level: str, + confidence: float, + recommendation: str, + reasoning: str, + evidence: str, + tier: str, + judge_model: str, + latency_ms: int, + user_decision: str = "pending", + ) -> None: + from sqlalchemy.dialects.sqlite import insert as sqlite_insert + + now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + stmt = sqlite_insert(intent_verdicts).values( + verdict_id=verdict_id, + ws_id=ws_id, + call_id=call_id, + func_name=func_name, + func_args=func_args, + intent_summary=intent_summary, + risk_level=risk_level, + confidence=confidence, + recommendation=recommendation, + reasoning=reasoning, + evidence=evidence, + tier=tier, + judge_model=judge_model, + latency_ms=latency_ms, + user_decision=user_decision, + created=now, + ) + # On verdict_id conflict, update only the three fields that + # genuinely change between heuristic and llm_fallback. See the + # protocol docstring for the full exclusion rationale — + # ``user_decision`` exclusion in particular is load-bearing. + stmt = stmt.on_conflict_do_update( + index_elements=["verdict_id"], + set_={ + "tier": tier, + "reasoning": reasoning, + "judge_model": judge_model, + }, + ) + with self._conn() as conn: + conn.execute(stmt) + conn.commit() + def create_intent_verdicts_bulk(self, verdicts: list[dict[str, Any]]) -> None: if not verdicts: return