fix(judge): UPSERT intent_verdicts so llm_fallback upgrades land

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

@@ -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
 # ---------------------------------------------------------------------------

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"

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", ""),

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

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.
 

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