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