feat(judge): add Smart Approvals (auto-approve trusted judge verdicts)

Opt-in judge.smart_approvals (default off): when the intent-validation
LLM judge returns a high-confidence "approve" verdict, the tool batch is
approved automatically with no operator prompt. review/deny recommendations,
low confidence, judge errors (llm_fallback), and a deterministic heuristic
deny/critical finding all still require a human. Requires judge.enabled.

- Batch-atomic: a parallel tool batch auto-approves only if every call
  qualifies; one non-qualifying call holds the whole batch for a human.
- Gate: tier==llm + recommendation==approve + confidence >=
  judge.confidence_threshold (default raised 0.7 -> 0.95), with a floor
  that never clears an explicit heuristic deny/critical verdict.
- approve_tools waits for the async LLM verdicts, finalises the audit
  trail (AutoApproveReason.smart_approval), and re-emits verdicts after
  the card so the live chip updates; the auto-approved row renders the
  LLM verdict rather than the cautious heuristic carry-over.
- judge: always deliver exactly one verdict per call (fallback on error);
  reject non-finite confidence so NaN can't clear the bar.
- Drop verdicts from a superseded judge generation so a reused call_id
  from a prior turn's still-running daemon can't satisfy the gate's wait.

Config plumbed through the server/console/CLI builders and the live
_judge_cfg; admin Judge tab renders the toggle. Docs + example config
updated. ~35 tests covering the gate matrix, batch-atomicity, the
heuristic floor, audit stamping, the streaming re-emit, NaN/duplicate-id
defenses, and the cross-turn generation guard.

docs/judge.md

@@ -37,13 +37,27 @@ model = ""                    # empty = same as session model
 provider = ""                 # empty = same as session provider
 base_url = ""
 api_key = ""
-confidence_threshold = 0.7   # reserved for v2 smart approvals (not used in v1)
+smart_approvals = false       # auto-approve high-confidence "approve" LLM verdicts (opt-in)
+confidence_threshold = 0.95   # Smart Approvals auto-approve bar (LLM recommendation=approve)
 max_context_ratio = 0.5       # max % of judge context window for history
 timeout = 60.0                # seconds (generous for local models)
 read_only_tools = true        # judge can use read_file/list_directory
 cancel_on_approval = false    # stop judging remaining tool calls once user decides
 ```
 
+### Smart Approvals
+
+With `smart_approvals = true` (off by default) a tool call is approved
+automatically — no operator prompt — when the intent judge's **LLM** verdict
+recommends `approve` with confidence at or above `confidence_threshold`. Every
+other outcome still reaches a human: `review` / `deny` recommendations,
+confidence below the threshold, judge errors or timeouts (`llm_fallback`), and
+any call the deterministic heuristic rules flagged `deny` / `critical` (the LLM
+may raise the heuristic's assessment but never lower it). Requires the judge to
+be enabled; auto-approved calls are tagged `smart_approval` in the dashboard and
+audit trail. Smart Approvals applies to the web and coordinator surfaces, not
+the interactive CLI.
+
 All fields are optional. The judge is enabled by default; use `enabled = false`
 (or `--no-judge` on the command line) to disable it.
 
@@ -54,9 +68,12 @@ All fields are optional. The judge is enabled by default; use `enabled = false`
 --judge-model MODEL            Model for judge
 --judge-provider PROVIDER      Provider for judge
 --judge-timeout SECONDS        LLM judge timeout (default: 60)
---judge-confidence FLOAT       Confidence threshold (default: 0.7)
+--judge-confidence FLOAT       Confidence threshold, 0-1 (default: 0.95)
 ```
 
+(Smart Approvals is configured via `[judge] smart_approvals` / the admin Judge
+settings, not a CLI flag — the interactive CLI prompts for approval directly.)
+
 CLI flags override `config.toml` values.
 
 ---

tests/test_judge.py

@@ -216,6 +216,47 @@ class TestErrorHandling:
         assert len(callback_results) == 1
         assert callback_results[0].tier == "llm_fallback"
 
+    def test_evaluate_single_raise_delivers_fallback(self):
+        """If ``_evaluate_single`` *raises* (not just returns None), the
+        daemon still delivers exactly one fallback verdict for that item.
+        Smart Approvals waits on the full verdict set before gating, so a
+        silently-skipped item would otherwise block that wait until its
+        timeout."""
+        judge = _make_judge()
+        judge._evaluate_single = MagicMock(  # type: ignore[method-assign]
+            side_effect=RuntimeError("boom")
+        )
+        callback_results: list[IntentVerdict] = []
+        judge.evaluate(
+            [_make_item()],
+            [{"role": "user", "content": "test"}],
+            callback_results.append,
+        )
+        time.sleep(0.5)
+        assert len(callback_results) == 1
+        assert callback_results[0].tier == "llm_fallback"
+
+    def test_executor_poison_delivers_fallback(self):
+        """An _ExecutorPoisonedError (a judge-call timeout poisoning the
+        single-worker executor) restarts the executor AND still delivers one
+        fallback for the interrupted item — the twin of the generic-exception
+        path, and load-bearing for Smart Approvals' batch-completeness wait."""
+        from turnstone.core.judge import _ExecutorPoisonedError
+
+        judge = _make_judge()
+        judge._evaluate_single = MagicMock(  # type: ignore[method-assign]
+            side_effect=_ExecutorPoisonedError()
+        )
+        callback_results: list[IntentVerdict] = []
+        judge.evaluate(
+            [_make_item()],
+            [{"role": "user", "content": "test"}],
+            callback_results.append,
+        )
+        time.sleep(0.5)
+        assert len(callback_results) == 1
+        assert callback_results[0].tier == "llm_fallback"
+
     def test_empty_content_returns_none(self):
         """Provider returns empty content, no tool calls."""
         provider = _make_mock_provider(response_content="")

tests/test_session.py

@@ -679,6 +679,36 @@ class TestTaskExec:
         assert fa["skill"] == ""
         assert fa["prompt"] == "do x"
 
+    def test_evaluate_intent_drops_superseded_generation_verdict(self, tmp_db, monkeypatch) -> None:
+        """A prior turn's judge daemon (still running because
+        cancel_on_approval defaults False) must NOT deliver verdicts once a
+        newer turn has superseded it — otherwise a model that reuses a
+        call_id across turns could ride a stale ``approve`` into a wrongful
+        Smart Approval of a different call."""
+        session = _make_session()
+        session.ui.on_intent_verdict = MagicMock()
+        fake_verdict = MagicMock()
+        fake_verdict.to_dict.return_value = {"verdict_id": "v0", "call_id": "c1", "tier": "llm"}
+        captured: list[Any] = []
+        fake_judge = MagicMock()
+        fake_judge.evaluate.side_effect = lambda items, *_a, **kw: (
+            captured.append(kw.get("callback")) or [fake_verdict] * len(items)
+        )
+        monkeypatch.setattr(session, "_ensure_judge", lambda: fake_judge)
+
+        item = {"call_id": "c1", "func_name": "bash", "needs_approval": True, "command": "ls"}
+        session._evaluate_intent([dict(item)])  # generation A
+        session._evaluate_intent([dict(item)])  # generation B supersedes A
+        callback_a, callback_b = captured[0], captured[1]
+
+        # A's late verdict (the superseded daemon) is dropped.
+        callback_a(fake_verdict)
+        session.ui.on_intent_verdict.assert_not_called()
+
+        # B's verdict (the current generation) is delivered normally.
+        callback_b(fake_verdict)
+        session.ui.on_intent_verdict.assert_called_once()
+
 
 # ---------------------------------------------------------------------------
 # Per-call model override on plan_agent / task_agent

tests/test_session_ui_base.py

@@ -1526,3 +1526,522 @@ def test_concurrent_writer_and_register_with_snapshot_no_loss_no_dup() -> None:
     assert reconstructed == expected, (
         f"reconstruction mismatch: len(rec)={len(reconstructed)}, len(exp)={len(expected)}"
     )
+
+
+# ---------------------------------------------------------------------------
+# Smart Approvals (judge.smart_approvals)
+# ---------------------------------------------------------------------------
+
+
+class _SeedingUI(_ConcreteUI):
+    """Re-delivers seeded LLM verdicts right after the approval-cycle
+    reset clears the cache — simulates the async judge daemon delivering
+    them via ``on_intent_verdict`` during the Smart Approvals wait, which
+    is the only point at which they can land and survive the reset."""
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.seed_verdicts: list[dict[str, Any]] = []
+
+    def _reset_approval_cycle(self) -> None:
+        super()._reset_approval_cycle()
+        for verdict in self.seed_verdicts:
+            self.on_intent_verdict(dict(verdict))
+
+
+def _patch_policies(verdicts: dict[str, str]):  # type: ignore[no-untyped-def]
+    """Neutralise the admin tool-policy stage so approve_tools tests
+    isolate the Smart Approvals gate."""
+    return patch(
+        "turnstone.core.policy.evaluate_tool_policies_batch",
+        return_value=verdicts,
+    )
+
+
+def _drain(lq: queue.Queue[Any]) -> list[dict[str, Any]]:
+    """Drain all currently-queued events off a listener queue."""
+    out: list[dict[str, Any]] = []
+    while True:
+        try:
+            out.append(lq.get_nowait())
+        except queue.Empty:
+            return out
+
+
+def _smart_ui() -> _ConcreteUI:
+    ui = _make_ui()
+    ui.smart_approvals_enabled = True
+    ui.smart_approval_threshold = 0.95
+    ui.smart_approval_wait_seconds = 1.0
+    return ui
+
+
+def _pending_item(call_id: str, func_name: str = "bash") -> dict[str, Any]:
+    """A still-pending tool call carrying a heuristic verdict, matching
+    what ``ChatSession._evaluate_intent`` attaches before the gate."""
+    return {
+        "call_id": call_id,
+        "func_name": func_name,
+        "approval_label": func_name,
+        "header": f"Tool: {func_name}",
+        "preview": "",
+        "needs_approval": True,
+        "_heuristic_verdict": {
+            "verdict_id": f"h-{call_id}",
+            "call_id": call_id,
+            "func_name": func_name,
+            "risk_level": "medium",
+            "confidence": 0.5,
+            "recommendation": "review",
+        },
+    }
+
+
+def _llm_verdict(
+    call_id: str,
+    *,
+    recommendation: str = "approve",
+    confidence: float = 0.99,
+    tier: str = "llm",
+) -> dict[str, Any]:
+    return {
+        "verdict_id": f"v-{call_id}",
+        "call_id": call_id,
+        "func_name": "bash",
+        "risk_level": "low",
+        "confidence": confidence,
+        "recommendation": recommendation,
+        "tier": tier,
+        "intent_summary": "",
+        "reasoning": "",
+        "evidence": [],
+    }
+
+
+def test_smart_approval_clears_high_confidence_llm_approve() -> None:
+    ui = _smart_ui()
+    item = _pending_item("c1")
+    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=0.99)
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([item])
+    assert remaining == []  # nothing left for a human
+    assert item["needs_approval"] is False
+    assert item["auto_approved"] is True
+    assert item["auto_approve_reason"] == "smart_approval"
+
+
+def test_smart_approval_clears_at_exact_threshold() -> None:
+    """``confidence >= threshold`` — the boundary value auto-approves."""
+    ui = _smart_ui()
+    item = _pending_item("c1")
+    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=0.95)
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([item])
+    assert remaining == []
+    assert item["auto_approved"] is True
+
+
+def test_smart_approval_holds_just_below_threshold() -> None:
+    ui = _smart_ui()
+    item = _pending_item("c1")
+    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=0.94)
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([item])
+    assert remaining == [item]
+    assert item.get("auto_approved") is not True
+    assert item["needs_approval"] is True
+
+
+def test_smart_approval_holds_review_and_deny() -> None:
+    """Only ``approve`` auto-approves; ``review`` / ``deny`` reach a human
+    no matter how confident the judge is."""
+    ui = _smart_ui()
+    for rec in ("review", "deny"):
+        item = _pending_item("c1")
+        ui._llm_verdicts = {"c1": _llm_verdict("c1", recommendation=rec, confidence=1.0)}
+        with _patch_get_storage(MagicMock()):
+            remaining = ui._apply_smart_approvals([item])
+        assert remaining == [item], rec
+        assert item.get("auto_approved") is not True, rec
+
+
+def test_smart_approval_holds_llm_fallback_even_if_approve() -> None:
+    """A ``llm_fallback`` verdict means the LLM stage timed out / errored
+    and the row is the heuristic carry-over.  Even if it reads ``approve``
+    at full confidence it must reach a human — errors require attention."""
+    ui = _smart_ui()
+    item = _pending_item("c1")
+    ui._llm_verdicts["c1"] = _llm_verdict(
+        "c1", recommendation="approve", confidence=1.0, tier="llm_fallback"
+    )
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([item])
+    assert remaining == [item]
+    assert item.get("auto_approved") is not True
+
+
+def test_smart_approval_holds_when_no_verdict_arrives() -> None:
+    """Wait budget elapses with no verdict cached → fail closed to the
+    human gate."""
+    ui = _smart_ui()
+    ui.smart_approval_wait_seconds = 0.05  # nothing will be delivered
+    item = _pending_item("c1")
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([item])
+    assert remaining == [item]
+    assert item["needs_approval"] is True
+
+
+def test_smart_approval_batch_atomic_holds_whole_batch_on_one_failure() -> None:
+    """Batch-atomic: a single non-qualifying call (here a review) in a
+    parallel batch holds the ENTIRE batch for a human — including the call
+    that individually qualified.  Parallel calls are one unit of intent."""
+    ui = _smart_ui()
+    a = _pending_item("c1")
+    b = _pending_item("c2")
+    ui._llm_verdicts = {
+        "c1": _llm_verdict("c1", recommendation="approve", confidence=0.99),
+        "c2": _llm_verdict("c2", recommendation="review", confidence=0.99),
+    }
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([a, b])
+    assert remaining == [a, b]  # NONE auto-approved
+    assert a.get("auto_approved") is not True
+    assert b.get("auto_approved") is not True
+
+
+def test_smart_approval_approves_full_batch_when_all_qualify() -> None:
+    """When every call in a parallel batch qualifies, the whole batch is
+    auto-approved and nothing is left for a human."""
+    ui = _smart_ui()
+    a = _pending_item("c1")
+    b = _pending_item("c2")
+    ui._llm_verdicts = {
+        "c1": _llm_verdict("c1", recommendation="approve", confidence=0.99),
+        "c2": _llm_verdict("c2", recommendation="approve", confidence=0.96),
+    }
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([a, b])
+    assert remaining == []
+    assert a["auto_approved"] is True and b["auto_approved"] is True
+    assert a["needs_approval"] is False and b["needs_approval"] is False
+
+
+def test_smart_approved_item_serializes_llm_verdict_not_heuristic() -> None:
+    """The auto-approved tool row must carry the driving LLM verdict
+    (llm/approve) as judge_verdict so the UI doesn't render a contradictory
+    heuristic 'review/medium' chip beside the SMART_APPROVAL pill."""
+    ui = _smart_ui()
+    item = _pending_item("c1")  # heuristic verdict is review / medium
+    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=0.99)
+    with _patch_get_storage(MagicMock()):
+        ui._apply_smart_approvals([item])
+    serialized = _ConcreteUI._serialize_approval_items([item])[0]
+    assert serialized["auto_approved"] is True
+    assert serialized["auto_approve_reason"] == "smart_approval"
+    judge_verdict = serialized["judge_verdict"]
+    assert judge_verdict["tier"] == "llm"
+    assert judge_verdict["recommendation"] == "approve"
+    # Heuristic still carried, but judge_verdict is what the row renders.
+    assert serialized["heuristic_verdict"]["recommendation"] == "review"
+
+
+def test_smart_approval_holds_batch_when_one_call_has_no_verdict() -> None:
+    """A parallel batch where one call never gets a verdict (timeout) holds
+    the whole batch, even though its sibling qualified."""
+    ui = _smart_ui()
+    ui.smart_approval_wait_seconds = 0.05
+    a = _pending_item("c1")
+    b = _pending_item("c2")
+    ui._llm_verdicts = {"c1": _llm_verdict("c1", recommendation="approve", confidence=0.99)}
+    # c2 has no verdict — the wait times out and the batch is held.
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([a, b])
+    assert remaining == [a, b]
+    assert a.get("auto_approved") is not True
+
+
+def test_smart_approval_skips_budget_override_pseudo_tool() -> None:
+    """The synthetic ``__budget_override__`` must always reach a human,
+    never smart-approved."""
+    ui = _smart_ui()
+    item = _pending_item("c1", func_name="__budget_override__")
+    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=1.0)
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([item])
+    assert remaining == [item]
+    assert item.get("auto_approved") is not True
+
+
+def test_smart_approval_stamps_verdict_user_decision() -> None:
+    """The LLM verdict arrived during the wait (parked in
+    ``_pending_verdicts`` as pending); the smart stage pulls it out so a
+    sibling's resolve can't re-stamp it, and records ``smart_approval`` on
+    both the cached dict and the persisted row."""
+    storage = MagicMock()
+    ui = _smart_ui()
+    item = _pending_item("c1")
+    verdict = _llm_verdict("c1", recommendation="approve", confidence=0.99)
+    ui._llm_verdicts["c1"] = verdict
+    ui._pending_verdicts = [verdict]  # as on_intent_verdict would have parked it
+    with _patch_get_storage(storage):
+        ui._apply_smart_approvals([item])
+    assert ui._pending_verdicts == []
+    assert ui._llm_verdicts["c1"]["user_decision"] == "smart_approval"
+    storage.update_intent_verdict.assert_called_once_with("v-c1", user_decision="smart_approval")
+
+
+def test_approve_tools_smart_approves_whole_batch_without_prompt() -> None:
+    """End-to-end through approve_tools: the verdict is delivered after
+    the cache reset (via _SeedingUI), the gate auto-approves, and the
+    function returns approved without ever emitting an approval prompt."""
+    storage = MagicMock()
+    ui = _SeedingUI(ws_id="ws-1", user_id="u1")
+    ui.smart_approvals_enabled = True
+    ui.smart_approval_threshold = 0.95
+    ui.smart_approval_wait_seconds = 1.0
+    item = _pending_item("c1")
+    ui.seed_verdicts = [_llm_verdict("c1", recommendation="approve", confidence=0.99)]
+    lq = ui._register_listener()
+    with _patch_get_storage(storage), _patch_policies({}):
+        approved, feedback = ui.approve_tools([item])
+    assert approved is True
+    assert feedback is None
+    assert item["auto_approved"] is True
+    assert item["auto_approve_reason"] == "smart_approval"
+    assert item["needs_approval"] is False
+    assert ui._pending_approval is None  # operator was never prompted
+    assert ui._pending_verdicts == []  # smart verdict pulled out + stamped
+    # No approval prompt was fanned out to listeners.
+    events = []
+    while True:
+        try:
+            events.append(lq.get_nowait()["type"])
+        except queue.Empty:
+            break
+    assert "approve_request" not in events
+
+
+def test_approve_tools_skips_smart_stage_when_disabled() -> None:
+    """With Smart Approvals off (the default), a confident approve verdict
+    does NOT bypass the human — approve_tools blocks on the prompt as
+    before."""
+    storage = MagicMock()
+    ui = _SeedingUI(ws_id="ws-1", user_id="u1")
+    ui.smart_approvals_enabled = False
+    ui.smart_approval_wait_seconds = 1.0
+    item = _pending_item("c1")
+    ui.seed_verdicts = [_llm_verdict("c1", recommendation="approve", confidence=0.99)]
+    timer = threading.Timer(0.05, lambda: ui.resolve_approval(True, "ok"))
+    timer.start()
+    try:
+        with _patch_get_storage(storage), _patch_policies({}):
+            approved, _feedback = ui.approve_tools([item])
+    finally:
+        timer.cancel()
+    assert approved is True  # the human approved, not the judge
+    assert item.get("auto_approve_reason") != "smart_approval"
+    assert item.get("auto_approved") is not True
+
+
+def test_await_llm_verdicts_returns_when_verdict_delivered() -> None:
+    """The wait wakes as soon as the last needed verdict lands, well
+    before the budget elapses."""
+    ui = _smart_ui()
+
+    def _deliver() -> None:
+        with _patch_get_storage(MagicMock()):
+            ui.on_intent_verdict(_llm_verdict("c1"))
+
+    timer = threading.Timer(0.02, _deliver)
+    timer.start()
+    try:
+        # Generous budget; should return on the notify, not the timeout.
+        ui._await_llm_verdicts({"c1"}, 5.0)
+    finally:
+        timer.cancel()
+    assert "c1" in ui._llm_verdicts
+
+
+def test_smart_approval_respects_heuristic_deny_floor() -> None:
+    """A high-confidence LLM ``approve`` must NOT override a deterministic
+    heuristic ``deny`` — the LLM may escalate the heuristic but never lower
+    it.  The call reaches a human."""
+    ui = _smart_ui()
+    item = _pending_item("c1")
+    item["_heuristic_verdict"]["recommendation"] = "deny"
+    item["_heuristic_verdict"]["risk_level"] = "critical"
+    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=1.0)
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([item])
+    assert remaining == [item]
+    assert item.get("auto_approved") is not True
+    assert item["needs_approval"] is True
+
+
+def test_smart_approval_respects_heuristic_critical_floor() -> None:
+    """A heuristic ``critical`` risk_level blocks smart approval even when
+    the heuristic recommendation itself isn't ``deny``."""
+    ui = _smart_ui()
+    item = _pending_item("c1")
+    item["_heuristic_verdict"]["recommendation"] = "review"
+    item["_heuristic_verdict"]["risk_level"] = "critical"
+    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=1.0)
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([item])
+    assert remaining == [item]
+    assert item.get("auto_approved") is not True
+
+
+def test_smart_approval_skips_oversized_batch() -> None:
+    """A batch with more calls than the FIFO verdict-cache cap can't be
+    reliably awaited (older verdicts evict before the wait sees them all),
+    so the whole batch reaches a human rather than stalling on the wait."""
+    ui = _smart_ui()
+    n = ui._LLM_VERDICT_CACHE_MAX + 1
+    items = [_pending_item(f"c{i}") for i in range(n)]
+    for i in range(n):
+        ui._llm_verdicts[f"c{i}"] = _llm_verdict(f"c{i}", recommendation="approve", confidence=1.0)
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals(items)
+    assert remaining == items  # none auto-approved
+    assert all(it.get("auto_approved") is not True for it in items)
+
+
+def test_replay_pending_verdicts_reemits_cached_verdicts() -> None:
+    """The streaming-fix helper re-fans-out each pending call's cached LLM
+    verdict as an intent_verdict event."""
+    ui = _smart_ui()
+    item = _pending_item("c1")
+    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="review", confidence=0.9)
+    lq = ui._register_listener()
+    ui._replay_pending_verdicts([item])
+    intent_events = []
+    while True:
+        try:
+            ev = lq.get_nowait()
+        except queue.Empty:
+            break
+        if ev.get("type") == "intent_verdict":
+            intent_events.append(ev)
+    assert len(intent_events) == 1
+    assert intent_events[0]["call_id"] == "c1"
+    assert intent_events[0]["recommendation"] == "review"
+
+
+def test_approve_tools_reemits_verdict_after_card_on_held_batch() -> None:
+    """Streaming regression fix: when Smart Approvals holds a batch (e.g. a
+    review verdict), the approve_request card is FOLLOWED by a re-emitted
+    intent_verdict so the live chip updates without a browser reload."""
+    storage = MagicMock()
+    ui = _SeedingUI(ws_id="ws-1", user_id="u1")
+    ui.smart_approvals_enabled = True
+    ui.smart_approval_threshold = 0.95
+    ui.smart_approval_wait_seconds = 1.0
+    item = _pending_item("c1")
+    ui.seed_verdicts = [_llm_verdict("c1", recommendation="review", confidence=0.99)]
+    lq = ui._register_listener()
+    timer = threading.Timer(0.1, lambda: ui.resolve_approval(False, "no"))
+    timer.start()
+    try:
+        with _patch_get_storage(storage), _patch_policies({}):
+            ui.approve_tools([item])
+    finally:
+        timer.cancel()
+    events = []
+    while True:
+        try:
+            events.append(lq.get_nowait())
+        except queue.Empty:
+            break
+    types = [e.get("type") for e in events]
+    assert "approve_request" in types
+    # An intent_verdict is re-emitted AFTER the card (the live chip update).
+    ar = types.index("approve_request")
+    assert "intent_verdict" in types[ar + 1 :]
+    # The wait already collected the verdict, so the card must not claim the
+    # judge is still working — no spurious "judge pending" spinner / poll.
+    assert events[ar].get("judge_pending") is False
+
+
+def test_judge_pending_true_when_llm_verdict_not_yet_cached() -> None:
+    """Normal async flow (Smart Approvals off): a judged call whose LLM
+    verdict hasn't arrived yet → approve_request reports judge_pending=True."""
+    ui = _make_ui()  # smart_approvals_enabled defaults False
+    item = _pending_item("c1")  # carries _heuristic_verdict, no cached LLM verdict
+    lq = ui._register_listener()
+    timer = threading.Timer(0.1, lambda: ui.resolve_approval(True, "ok"))
+    timer.start()
+    try:
+        with _patch_get_storage(MagicMock()), _patch_policies({}):
+            ui.approve_tools([item])
+    finally:
+        timer.cancel()
+    reqs = [e for e in _drain(lq) if e.get("type") == "approve_request"]
+    assert reqs and reqs[0]["judge_pending"] is True
+
+
+def test_auto_approve_reason_vocabulary_matches_js() -> None:
+    """AutoApproveReason.ALL must stay in lockstep with the JS
+    KNOWN_AUTO_APPROVE_REASONS set — a server-sent reason missing from the JS
+    set degrades to the 'unknown' pill on the coordinator tree."""
+    import re
+    from pathlib import Path
+
+    from turnstone.core.session_ui_base import AutoApproveReason
+
+    js = Path(__file__).resolve().parents[1] / "turnstone/console/static/coordinator/coordinator.js"
+    m = re.search(
+        r"KNOWN_AUTO_APPROVE_REASONS\s*=\s*new Set\(\s*\[(.*?)\]",
+        js.read_text(),
+        re.S,
+    )
+    assert m, "KNOWN_AUTO_APPROVE_REASONS set not found in coordinator.js"
+    js_reasons = set(re.findall(r'"([^"]+)"', m.group(1)))
+    assert js_reasons == AutoApproveReason.ALL
+
+
+def test_verdict_confidence_rejects_non_finite() -> None:
+    """NaN/inf confidence is treated as malformed (0.0), not clamped to 1.0."""
+    assert _ConcreteUI._verdict_confidence({"confidence": float("nan")}) == 0.0
+    assert _ConcreteUI._verdict_confidence({"confidence": float("inf")}) == 0.0
+    assert _ConcreteUI._verdict_confidence({"confidence": 0.97}) == 0.97
+
+
+def test_smart_approval_holds_nan_confidence() -> None:
+    """A NaN confidence (json.loads accepts NaN) must NOT clear the
+    auto-approve bar even with recommendation=approve."""
+    ui = _smart_ui()
+    item = _pending_item("c1")
+    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=float("nan"))
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([item])
+    assert remaining == [item]
+    assert item.get("auto_approved") is not True
+
+
+def test_smart_approval_holds_batch_with_duplicate_call_ids() -> None:
+    """Two pending calls sharing a call_id (some local models emit duplicate
+    non-empty ids) must not both be cleared by the single shared verdict —
+    hold the whole batch."""
+    ui = _smart_ui()
+    a = _pending_item("dup")
+    b = _pending_item("dup")  # same call_id, distinct call
+    ui._llm_verdicts["dup"] = _llm_verdict("dup", recommendation="approve", confidence=0.99)
+    with _patch_get_storage(MagicMock()):
+        remaining = ui._apply_smart_approvals([a, b])
+    assert remaining == [a, b]
+    assert a.get("auto_approved") is not True
+
+
+def test_on_intent_verdict_skips_append_for_already_finalized_verdict() -> None:
+    """Guards the audit-corruption race: a verdict already stamped with a
+    final user_decision (e.g. ``_finalize_smart_verdicts`` ran between this
+    verdict's notify and its append) is NOT re-parked in _pending_verdicts,
+    so a later round's resolve_approval can't overwrite its audit row."""
+    ui = _make_ui()
+    verdict = {"verdict_id": "v1", "call_id": "c1", "user_decision": "smart_approval"}
+    with _patch_get_storage(MagicMock()):
+        ui.on_intent_verdict(verdict)
+    assert ui._pending_verdicts == []
+    assert ui._llm_verdicts["c1"]["user_decision"] == "smart_approval"

tests/test_settings_registry.py

@@ -46,6 +46,27 @@ class TestValidateKey:
         assert validate_key("audio.stt_prompt").default == ""
         assert validate_key("audio.tts_voice").default == "alloy"
 
+    def test_smart_approvals_setting_registered(self):
+        defn = validate_key("judge.smart_approvals")
+        assert defn.key == "judge.smart_approvals"
+        assert defn.type == "bool"
+        assert defn.default is False  # opt-in: off by default
+        assert defn.section == "judge"
+        assert "judge.smart_approvals" in SETTINGS
+
+    def test_confidence_threshold_is_smart_approval_bar(self):
+        """Default bumped to the Smart Approvals auto-approve bar (0.95),
+        still clamped to [0, 1]."""
+        defn = validate_key("judge.confidence_threshold")
+        assert defn.type == "float"
+        assert defn.default == 0.95
+        assert defn.min_value == 0.0
+        assert defn.max_value == 1.0
+
+    def test_smart_approvals_bool_coercion(self):
+        assert validate_value("judge.smart_approvals", "true") is True
+        assert validate_value("judge.smart_approvals", "false") is False
+
 
 # ---------------------------------------------------------------------------
 # validate_value — type coercion

turnstone.example.toml

@@ -107,7 +107,8 @@
 
 [judge]
 # enabled = true               # Enable intent validation
-# confidence_threshold = 0.7   # Minimum confidence for heuristic verdicts
+# smart_approvals = false      # Auto-approve high-confidence "approve" LLM verdicts (opt-in)
+# confidence_threshold = 0.95  # Smart Approvals auto-approve bar (LLM recommendation=approve)
 # output_guard = true          # Scan tool output for security signals
 # redact_secrets = true        # Redact detected credentials in output
 

turnstone/api/server_schemas.py

@@ -307,8 +307,10 @@ class RecentAutoApproval(BaseModel):
             "Source that fired the bypass.  ``skill`` (skill template's "
             "``allowed_tools``), ``always`` (user 'Approve + Always' "
             "click), ``policy`` (admin tool-policy ``allow`` rule), "
-            "``blanket`` (workstream-level ``auto_approve=True``), or "
-            "``auto_approve_tools`` (legacy / unknown writer)."
+            "``blanket`` (workstream-level ``auto_approve=True``), "
+            "``smart_approval`` (Smart Approvals: high-confidence LLM "
+            "judge ``approve`` verdict), or ``auto_approve_tools`` "
+            "(legacy / unknown writer)."
         ),
     )
     ts: float = Field(

turnstone/cli.py

@@ -1070,8 +1070,8 @@ def main() -> None:
         "--judge-confidence",
         dest="judge_confidence",
         type=float,
-        default=0.7,
-        help="Confidence threshold for judge (default: 0.7)",
+        default=0.95,
+        help="Judge verdict confidence threshold, 0-1 (default: 0.95)",
     )
     from turnstone.core.config import add_config_arg, apply_config
 

turnstone/console/session_factory.py

@@ -63,6 +63,7 @@ def build_console_session_factory(
         return JudgeConfig(
             enabled=config_store.get("judge.enabled"),
             model=config_store.get("judge.model"),
+            smart_approvals=config_store.get("judge.smart_approvals"),
             confidence_threshold=config_store.get("judge.confidence_threshold"),
             max_context_ratio=config_store.get("judge.max_context_ratio"),
             timeout=config_store.get("judge.timeout"),

turnstone/console/static/coordinator/coordinator.js

@@ -230,6 +230,7 @@
     "policy",
     "blanket",
     "auto_approve_tools",
+    "smart_approval",
   ]);
   const UNKNOWN_AUTO_APPROVE_REASON = "unknown";
 

turnstone/core/judge.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import fnmatch
 import json
+import math
 import os
 import re
 import threading
@@ -82,7 +83,8 @@ class JudgeConfig:
 
     enabled: bool = True
     model: str = ""  # empty = use session model
-    confidence_threshold: float = 0.7
+    smart_approvals: bool = False  # auto-approve high-confidence "approve" LLM verdicts
+    confidence_threshold: float = 0.95  # Smart Approvals auto-approve bar (recommendation=approve)
     max_context_ratio: float = 0.5
     timeout: float = 60.0  # per-turn timeout in seconds (see class docstring)
     read_only_tools: bool = True
@@ -1109,11 +1111,20 @@ class IntentJudge:
                 except _ExecutorPoisonedError:
                     executor.shutdown(wait=False, cancel_futures=True)
                     executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="judge-api")
+                    # Deliver a fallback for the interrupted item so every
+                    # call still gets exactly one verdict. Smart Approvals
+                    # waits on the full set before gating; a silently-
+                    # skipped item would otherwise block that wait until
+                    # its timeout (and the advisory UI would miss a chip).
+                    self._deliver_fallbacks(
+                        [item], [h_verdict], callback, "judge executor restarted"
+                    )
                 except Exception:
                     log.exception(
                         "Judge evaluation failed for %s",
                         item.get("func_name", "?"),
                     )
+                    self._deliver_fallbacks([item], [h_verdict], callback, "judge evaluation error")
         finally:
             executor.shutdown(wait=False, cancel_futures=True)
             try:
@@ -1573,8 +1584,13 @@ class IntentJudge:
 
         confidence = 0.5
         try:
-            confidence = float(data.get("confidence", 0.5))
-            confidence = max(0.0, min(1.0, confidence))
+            parsed = float(data.get("confidence", 0.5))
+            # Reject non-finite (NaN/inf): json.loads accepts NaN, and a NaN
+            # would survive max/min clamping as 1.0 (comparisons with NaN are
+            # all False) — keep the cautious 0.5 default instead so a NaN can't
+            # masquerade as maximum confidence downstream (e.g. Smart Approvals).
+            if math.isfinite(parsed):
+                confidence = max(0.0, min(1.0, parsed))
         except (ValueError, TypeError):
             pass  # keeps default 0.5
 

turnstone/core/session.py

@@ -1246,6 +1246,7 @@ class ChatSession:
         return JudgeConfig(
             enabled=cs.get("judge.enabled"),
             model=jc.model,
+            smart_approvals=cs.get("judge.smart_approvals"),
             confidence_threshold=cs.get("judge.confidence_threshold"),
             max_context_ratio=cs.get("judge.max_context_ratio"),
             timeout=cs.get("judge.timeout"),
@@ -5201,14 +5202,36 @@ class ChatSession:
             elif it.get("mcp_args"):
                 it["func_args"] = it["mcp_args"]
 
+        # Publish this judge generation as the session's current cancel event
+        # BEFORE spawning the daemon, so the callback can detect when a later
+        # turn has superseded it (this turn always runs before its own
+        # approve_tools, so the assignment is in place before any verdict can
+        # land).  ``_execute_tools`` re-asserts the same value and handles the
+        # judge-disabled (None) case.
+        cancel_event = threading.Event()
+        self._judge_cancel_event = cancel_event
+
         def _on_verdict(verdict: object) -> None:
-            """Callback from the daemon judge thread."""
+            """Callback from the daemon judge thread.
+
+            Drop the verdict when a newer turn has replaced this judge
+            generation.  With ``cancel_on_approval=False`` (the default) the
+            prior turn's daemon runs to completion and would otherwise write a
+            stale verdict — keyed only by ``call_id`` — into the freshly-reset
+            ``_llm_verdicts`` cache; a model that reuses a ``call_id`` across
+            turns could then ride that stale ``approve`` to a wrongful Smart
+            Approval of a *different* call.  Identity-comparing the live
+            generation closes that without affecting same-turn late delivery
+            (``cancel_on_approval=False`` still streams this turn's verdicts,
+            since the session event still points at this ``cancel_event``).
+            """
+            if self._judge_cancel_event is not cancel_event:
+                return
             try:
                 self.ui.on_intent_verdict(verdict.to_dict())  # type: ignore[attr-defined]
             except Exception:
                 log.debug("judge.verdict_delivery_failed", exc_info=True)
 
-        cancel_event = threading.Event()
         heuristic_verdicts = judge.evaluate(
             pending,
             list(self.messages),  # snapshot — daemon thread must not see mutations
@@ -5824,6 +5847,23 @@ class ChatSession:
         judge_cancel = self._evaluate_intent(items)
         self._judge_cancel_event = judge_cancel  # track for close()
 
+        # Push the live Smart Approvals config onto the UI just before the
+        # gate so a hot-reloaded ``judge.*`` change takes effect on this
+        # batch.  Only SessionUIBase carries the smart-approval gate (the
+        # CLI / eval UIs have their own ``approve_tools``); the isinstance
+        # check both skips those and narrows the type for the attribute
+        # writes.  ``approve_tools`` acts on these only when the judge is
+        # enabled AND ``judge.smart_approvals`` is on, so the feature stays
+        # inert (human-gated, as today) unless explicitly turned on.
+        from turnstone.core.session_ui_base import SessionUIBase
+
+        if isinstance(self.ui, SessionUIBase):
+            jc = self._judge_cfg
+            self.ui.smart_approvals_enabled = bool(jc and jc.enabled and jc.smart_approvals)
+            if jc is not None:
+                self.ui.smart_approval_threshold = jc.confidence_threshold
+                self.ui.smart_approval_wait_seconds = jc.timeout
+
         # Phase 2: approve via UI
         self._emit_state("attention")
         try:

turnstone/core/session_ui_base.py

@@ -27,6 +27,7 @@ import collections
 import contextlib
 import copy
 import json
+import math
 import os
 import queue
 import threading
@@ -151,7 +152,7 @@ class AutoApproveReason:
     ``.value`` dance at every emit site, and no surprise behaviour
     if a consumer compares against the literal).
 
-    The five reasons reflect the disjoint set of paths that bypass
+    The six reasons reflect the disjoint set of paths that bypass
     the operator approval gate:
 
     - :attr:`SKILL` — workstream's skill template populated
@@ -173,6 +174,12 @@ class AutoApproveReason:
       populated (legacy / pre-source-tracking instances).  Visible
       as a generic pill rather than a misleading ``skill`` /
       ``always`` claim.
+    - :attr:`SMART_APPROVAL` — Smart Approvals (``judge.smart_approvals``)
+      auto-approved the call because the intent judge's LLM verdict
+      recommended ``approve`` with confidence at or above
+      ``judge.confidence_threshold``.  Distinct pill so an operator
+      can see the judge — not a policy or a prior "Always" click —
+      cleared this call.
     """
 
     SKILL = "skill"
@@ -180,8 +187,11 @@ class AutoApproveReason:
     POLICY = "policy"
     BLANKET = "blanket"
     AUTO_APPROVE_TOOLS = "auto_approve_tools"
+    SMART_APPROVAL = "smart_approval"
 
-    ALL: frozenset[str] = frozenset({SKILL, ALWAYS, POLICY, BLANKET, AUTO_APPROVE_TOOLS})
+    ALL: frozenset[str] = frozenset(
+        {SKILL, ALWAYS, POLICY, BLANKET, AUTO_APPROVE_TOOLS, SMART_APPROVAL}
+    )
 
 
 class SessionUIBase:
@@ -244,6 +254,21 @@ class SessionUIBase:
         self._pending_plan_review: dict[str, Any] | None = None
         self.auto_approve = False
         self.auto_approve_tools: set[str] = set()
+        # Smart Approvals (``judge.smart_approvals``): when enabled, a
+        # tool call whose LLM intent verdict recommends ``approve`` with
+        # confidence ≥ ``smart_approval_threshold`` is auto-approved
+        # without an operator prompt.  ChatSession pushes these three
+        # values onto the UI from the live judge config each turn (just
+        # before ``approve_tools``) so a hot-reloaded settings change
+        # takes effect on the next batch.  Defaults keep the feature off
+        # for any UI the session doesn't configure (eval, fixtures).
+        self.smart_approvals_enabled = False
+        self.smart_approval_threshold = 0.95
+        # How long ``approve_tools`` waits for the async LLM verdict
+        # before falling back to a human prompt (fail-closed).  Bounded
+        # by the judge timeout; the wait returns early the moment every
+        # pending call has a verdict.
+        self.smart_approval_wait_seconds = 0.0
         # Per-tool source for ``auto_approve_tools`` membership.  Two
         # writers populate the set with semantically different intent:
         #
@@ -356,6 +381,12 @@ class SessionUIBase:
         # Verdict cache for SSE reconnect replay (tab switching
         # shouldn't lose the judge's final call on a just-run tool).
         self._llm_verdicts: dict[str, dict[str, Any]] = {}
+        # Signalled by ``on_intent_verdict`` whenever an LLM verdict
+        # lands in ``_llm_verdicts``; Smart Approvals (``approve_tools``)
+        # waits on it for the verdicts of the calls it's about to gate.
+        # Shares ``_ws_lock`` so the wait + the cache write are one
+        # critical section (no separate lock to order).
+        self._verdict_cond = threading.Condition(self._ws_lock)
         # Re-populate the recent-auto-approve ring buffer from the
         # audit log so the dashboard pill survives UI rebuilds —
         # saved-workstream rehydrate / coord→node click-through /
@@ -896,6 +927,23 @@ class SessionUIBase:
         # policy/auto-approve pass that drained the override from ``pending``
         # cannot disarm this gate.
         blanket_active = self.auto_approve and not has_budget_override
+
+        # -- Smart Approvals (judge.smart_approvals) -----------------------------
+        # Last automatic gate before the human prompt, after the explicit
+        # operator-configured ones (policy / "Always" / blanket): wait
+        # briefly for the async LLM intent verdict and auto-approve every
+        # still-pending call the judge cleared with a high-confidence
+        # ``approve``.  Skipped under blanket auto-approve (everything is
+        # approved already) and when a ``__budget_override__`` pseudo-tool
+        # is present (it must always reach a human).
+        if (
+            pending
+            and self.smart_approvals_enabled
+            and not blanket_active
+            and not has_budget_override
+        ):
+            pending = self._apply_smart_approvals(pending)
+
         if not pending or blanket_active:
             if blanket_active and pending:
                 # Blanket flag drained the rest of pending — tag so the
@@ -964,7 +1012,19 @@ class SessionUIBase:
         self._persist_intent_verdicts_bulk(heuristic_verdicts, default_tier="heuristic")
 
         with self._ws_lock:
-            self._pending_verdicts = pending_verdicts
+            # Normally a plain assignment: in the async flow the LLM judge
+            # is slower than this setup so ``_pending_verdicts`` is still
+            # the empty list ``_reset_approval_cycle`` left it.  But Smart
+            # Approvals deliberately waits for verdicts upstream, so by here
+            # ``on_intent_verdict`` has already parked the LLM verdicts of
+            # the human-pending siblings; carry them across the reassignment
+            # so ``resolve_approval`` can stamp their ``user_decision``
+            # (without this they'd be dropped and stuck at ``"pending"``).
+            # Smart-approved calls were pulled out + stamped already by
+            # ``_finalize_smart_verdicts``, so they don't reappear here.
+            pending_cids = {hv.get("call_id") for hv in pending_verdicts}
+            early_llm = [v for v in self._pending_verdicts if v.get("call_id") in pending_cids]
+            self._pending_verdicts = pending_verdicts + early_llm
 
         # Record any items the policy block already auto-approved
         # before falling through to the prompt — without this the
@@ -974,8 +1034,17 @@ class SessionUIBase:
         # No-op when no items are auto-approve-tagged.
         self._record_auto_approves(items)
 
-        # Send approval request and block
-        judge_pending = any(it.get("_heuristic_verdict") for it in items)
+        # Send approval request and block.  ``judge_pending`` tells the UI
+        # whether to expect LLM verdicts still in flight: true only when a
+        # judged item does NOT yet have its LLM verdict cached.  Under Smart
+        # Approvals the gate already waited for every verdict, so they are
+        # present and this is false (no spurious "judge working" spinner /
+        # poll); in the normal async flow they haven't arrived yet → true.
+        with self._ws_lock:
+            judge_pending = any(
+                it.get("_heuristic_verdict") and it.get("call_id", "") not in self._llm_verdicts
+                for it in items
+            )
         self._approval_event.clear()
         self._pending_approval = {
             "type": "approve_request",
@@ -997,6 +1066,16 @@ class SessionUIBase:
         # while parked on _approval_event.wait. The push path
         # eliminates the race.
         self._broadcast_approve_request(self._pending_approval)
+        # Smart Approvals waited for the LLM verdicts BEFORE this card was
+        # built, so on_intent_verdict already fanned out their
+        # ``intent_verdict`` events while no card existed — a live client
+        # dropped them and the chip would stay on the heuristic value until
+        # a reload re-merged the cache.  Re-emit them now, after the card,
+        # to restore the normal approve_request → intent_verdict ordering so
+        # the live chip updates.  No-op in the normal async flow (cache is
+        # empty here) and when the feature is off.
+        if self.smart_approvals_enabled:
+            self._replay_pending_verdicts(items)
         if not self._approval_event.wait(timeout=self._APPROVAL_WAIT_TIMEOUT):
             # Approval timed out (e.g., user disconnected). Deny via
             # resolve_approval so verdicts and state are updated consistently.
@@ -1021,6 +1100,186 @@ class SessionUIBase:
 
         return approved, feedback
 
+    # ------------------------------------------------------------------
+    # Smart Approvals (judge.smart_approvals)
+    # ------------------------------------------------------------------
+
+    def _apply_smart_approvals(self, pending: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        """Auto-approve a tool batch the LLM judge cleared confidently.
+
+        **Batch-atomic.**  Waits (bounded by ``smart_approval_wait_seconds``)
+        for the async LLM verdicts of EVERY pending call, then auto-approves
+        the whole batch only if EVERY call qualifies; a single non-qualifying
+        call sends the entire batch to a human.  Parallel tool calls are one
+        unit of intent — approving the safe-looking members while a sibling
+        is held would let a multi-step action through piecemeal — so it is
+        all-or-nothing.
+
+        A call qualifies when its LLM verdict is a *completed* ``approve`` —
+        tier ``"llm"`` (NOT the ``"llm_fallback"`` error tier) at confidence
+        ≥ ``smart_approval_threshold`` — AND the deterministic heuristic did
+        not *explicitly* flag it ``deny`` / ``critical``.  That heuristic
+        floor only blocks the explicit danger verdicts: the heuristic
+        DEFAULT for an unmatched tool is ``review``, and upgrading ``review``
+        → ``approve`` on a confident LLM verdict is exactly this feature's
+        job, so ``review`` is not a floor.  But a call the pattern rules
+        matched as ``deny`` / ``critical`` (e.g. ``rm -rf /``) is never
+        cleared by the (promptable) LLM — those always reach a human.
+
+        Returns ``[]`` when the whole batch is auto-approved, or *pending*
+        unchanged when anything is uncertain (review/deny/low-confidence/
+        error/timeout/heuristic-danger/no-verdict) — fails closed.
+        """
+        # Only calls the judge actually evaluated carry a heuristic verdict;
+        # the ``__budget_override__`` pseudo-tool is never smart-approved, so
+        # its presence makes ``candidates`` smaller than ``pending`` and the
+        # batch-completeness check below holds the whole batch for a human.
+        candidates = [
+            it
+            for it in pending
+            if it.get("_heuristic_verdict") and it.get("func_name") != "__budget_override__"
+        ]
+        needed = {it.get("call_id", "") for it in candidates if it.get("call_id")}
+        if not needed:
+            return pending
+        # The whole batch must be eligible before we pay the verdict wait:
+        #   - every pending call must be a judged candidate — an unjudged
+        #     sibling or the ``__budget_override__`` pseudo-tool makes
+        #     candidates < pending, AND
+        #   - call_ids must be unique — some local models emit duplicate
+        #     non-empty tool-call ids (``_ensure_tool_call_ids`` only fills
+        #     MISSING ones), which collapse in the ``needed`` set and would let
+        #     one verdict clear two distinct calls (their args differ).
+        # Either mismatch → hold the whole batch for a human.  Checked
+        # pre-wait so an ineligible batch never pays the (up-to-timeout) wait.
+        if len(candidates) != len(pending) or len(needed) != len(candidates):
+            return pending
+        # The per-round verdict cache is FIFO-capped; a batch with more calls
+        # than the cap can't hold every verdict at once, so the wait could
+        # never see them all and would stall to its full budget.  Hold such
+        # (pathological) batches for a human.
+        if len(needed) > self._LLM_VERDICT_CACHE_MAX:
+            log.info("judge.smart_approval.batch_too_large", ws_id=self.ws_id, count=len(needed))
+            return pending
+        self._await_llm_verdicts(needed, self.smart_approval_wait_seconds)
+
+        threshold = self.smart_approval_threshold
+        qualified: dict[str, dict[str, Any]] = {}
+        with self._ws_lock:
+            for it in candidates:
+                cid = it.get("call_id", "")
+                v = self._llm_verdicts.get(cid)
+                # Require a COMPLETED LLM verdict — tier "llm", not the
+                # "llm_fallback" error carry-over ("heuristic" never lands in
+                # this cache) — recommending "approve" at/above threshold.
+                if (
+                    v is None
+                    or v.get("tier") != "llm"
+                    or v.get("recommendation") != "approve"
+                    or self._verdict_confidence(v) < threshold
+                ):
+                    return pending  # one fails → none of the batch auto-approves
+                hv = it.get("_heuristic_verdict") or {}
+                if hv.get("recommendation") == "deny" or hv.get("risk_level") == "critical":
+                    return pending  # explicit deterministic danger flag → human
+                qualified[cid] = v
+
+        # Whole batch qualified.  Clear the gate flag (mirrors the policy
+        # ``allow`` branch) so each call is treated as resolved: the coord
+        # pill renders (``auto_approved && !needs_approval``) and the denial
+        # sweep in ChatSession._execute_tools leaves them to execute.  Attach
+        # the driving LLM verdict so the auto-approved tool row renders it
+        # (llm tier, approve) instead of the cautious heuristic carry-over,
+        # which would read contradictorily beside the SMART_APPROVAL pill.
+        for it in candidates:
+            it["needs_approval"] = False
+            it["_llm_verdict"] = qualified.get(it.get("call_id", ""))
+        self._tag_auto_approved(candidates, AutoApproveReason.SMART_APPROVAL)
+        self._finalize_smart_verdicts(needed)
+        log.info("judge.smart_approval", ws_id=self.ws_id, approved=len(candidates))
+        return []
+
+    @staticmethod
+    def _verdict_confidence(verdict: dict[str, Any]) -> float:
+        """Verdict confidence clamped to ``[0.0, 1.0]``; ``0.0`` if malformed.
+
+        Rejects non-finite values explicitly: ``min``/``max`` would let a NaN
+        through as ``1.0`` (every comparison with NaN is False), and
+        ``json.loads`` accepts ``NaN`` by default — so a verdict reporting
+        ``"confidence": NaN`` could otherwise clear the auto-approve bar.
+        """
+        try:
+            confidence = float(verdict.get("confidence", 0.0))
+        except (TypeError, ValueError):
+            return 0.0
+        if not math.isfinite(confidence):
+            return 0.0
+        return max(0.0, min(1.0, confidence))
+
+    def _await_llm_verdicts(self, needed: set[str], budget_seconds: float) -> None:
+        """Block until every call_id in *needed* has an LLM verdict cached.
+
+        Returns the instant the last verdict lands; otherwise gives up
+        after *budget_seconds* and leaves the missing calls for the human
+        gate (fail-closed).  ``on_intent_verdict`` notifies
+        ``_verdict_cond`` on every cache write, and the judge delivers
+        exactly one verdict (LLM or ``llm_fallback``) per call, so the
+        common case is an early return at real judge latency rather than a
+        full-budget wait.
+        """
+        if budget_seconds <= 0 or not needed:
+            return
+        deadline = time.monotonic() + budget_seconds
+        with self._verdict_cond:
+            while not needed.issubset(self._llm_verdicts.keys()):
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    return
+                self._verdict_cond.wait(timeout=remaining)
+
+    def _finalize_smart_verdicts(self, smart_ids: set[str]) -> None:
+        """Stamp ``smart_approval`` on the LLM verdicts of auto-approved calls.
+
+        The verdicts arrived during ``_await_llm_verdicts`` — before the
+        call was tagged auto-approved — so ``on_intent_verdict`` parked
+        them in ``_pending_verdicts`` with ``user_decision="pending"``.
+        Pull them out (so a human-pending sibling's ``resolve_approval``
+        can't overwrite the reason), stamp the cached dict in place (the
+        reconnect-replay payload reflects the final decision), and UPDATE
+        the persisted rows.  The matching heuristic verdict is stamped by
+        ``approve_tools``'s own persistence path via the ``auto_approved``
+        tag set just before this call.
+        """
+        stamped: list[dict[str, Any]] = []
+        with self._ws_lock:
+            for cid in smart_ids:
+                v = self._llm_verdicts.get(cid)
+                if v is not None:
+                    v["user_decision"] = AutoApproveReason.SMART_APPROVAL
+                    stamped.append(v)
+            if self._pending_verdicts:
+                self._pending_verdicts = [
+                    v for v in self._pending_verdicts if v.get("call_id") not in smart_ids
+                ]
+        if stamped:
+            self._persist_verdict_decisions(stamped, AutoApproveReason.SMART_APPROVAL)
+
+    def _replay_pending_verdicts(self, items: list[dict[str, Any]]) -> None:
+        """Re-emit already-cached LLM verdicts for the human-pending calls.
+
+        Mirrors the reconnect-replay re-injection: after the
+        ``approve_request`` card exists, re-send each pending call's cached
+        ``intent_verdict`` so a live client (which dropped the events the
+        Smart Approvals wait fanned out before the card) applies them to the
+        chip.  Snapshots under the lock, fans out without it.
+        """
+        cids = {it.get("call_id", "") for it in items if it.get("call_id")}
+        with self._ws_lock:
+            verdicts = [dict(self._llm_verdicts[c]) for c in cids if c in self._llm_verdicts]
+        for verdict in verdicts:
+            self._enqueue({"type": "intent_verdict", **verdict})
+            self._broadcast_intent_verdict(verdict)
+
     # ------------------------------------------------------------------
     # Intent-judge + output-guard plumbing
     # ------------------------------------------------------------------
@@ -1064,6 +1323,11 @@ class SessionUIBase:
                     oldest_key = next(iter(self._llm_verdicts))
                     del self._llm_verdicts[oldest_key]
                 self._llm_verdicts[call_id] = verdict
+                # Wake any Smart Approvals wait parked on this call's
+                # verdict (``_verdict_cond`` shares ``_ws_lock``, so the
+                # notify is valid here and the waiter re-checks its
+                # call-id set on wake).
+                self._verdict_cond.notify_all()
                 # Pop (not get) — once consumed the entry isn't useful
                 # again; TTL pruning at the writer side keeps the
                 # never-consumed case bounded too.
@@ -1103,7 +1367,19 @@ class SessionUIBase:
             return
         with self._ws_lock:
             decision = self._last_verdict_decision
-            if not decision:
+            # Skip the append when the verdict already carries a final
+            # ``user_decision``.  Smart Approvals' ``_finalize_smart_verdicts``
+            # runs on the worker thread the moment ``_await_llm_verdicts``
+            # wakes — which is this method's ``notify_all`` (above), fired
+            # BEFORE this append and with an unlocked ``_persist_intent_verdict``
+            # in between.  So finalize can stamp ``smart_approval`` on this
+            # same cached dict and clear ``_pending_verdicts`` before we get
+            # here; without this guard we'd re-park the already-final verdict,
+            # and the NEXT round's ``resolve_approval`` would overwrite its
+            # audit row with approved/denied/timeout.  (The reverse ordering —
+            # append before finalize — is handled by finalize's own
+            # ``_pending_verdicts`` filter.)
+            if not decision and verdict.get("user_decision", "pending") == "pending":
                 self._pending_verdicts.append(verdict)
         if decision:
             self._persist_verdict_decisions([verdict], decision)
@@ -1381,6 +1657,13 @@ class SessionUIBase:
             }
             if "_heuristic_verdict" in it:
                 entry["heuristic_verdict"] = it["_heuristic_verdict"]
+            if it.get("_llm_verdict"):
+                # The completed LLM verdict that drove a Smart Approval.
+                # Sent so the auto-approved tool row renders the llm-tier
+                # approve/risk that actually cleared the call, instead of the
+                # cautious heuristic carry-over — which reads contradictorily
+                # next to the SMART_APPROVAL pill (e.g. "review/medium" + ✓).
+                entry["judge_verdict"] = it["_llm_verdict"]
             if it.get("auto_approved"):
                 entry["auto_approved"] = True
                 entry["auto_approve_reason"] = it.get("auto_approve_reason", "")

turnstone/core/settings_registry.py

@@ -475,17 +475,32 @@ def _build_registry() -> dict[str, SettingDef]:
             "registered aliases inherit the session model and log a warning — "
             "register the model in the Models tab and reference it by alias.",
         ),
+        SettingDef(
+            "judge.smart_approvals",
+            "bool",
+            False,
+            "Enable Smart Approvals",
+            "judge",
+            help="When enabled, a tool call is approved automatically \u2014 without waiting for a "
+            "human \u2014 if the intent-validation judge's LLM verdict recommends 'approve' with a "
+            "confidence at or above judge.confidence_threshold. 'review' and 'deny' "
+            "recommendations, low-confidence verdicts, and judge errors (timeouts / fallbacks) "
+            "always still require human approval. Requires judge.enabled and a working LLM judge. "
+            "Disabled by default \u2014 opt in once you trust the judge on your workload.",
+        ),
         SettingDef(
             "judge.confidence_threshold",
             "float",
-            0.7,
-            "Min confidence for judge verdict",
+            0.95,
+            "Min judge confidence for Smart Approvals",
             "judge",
             min_value=0.0,
             max_value=1.0,
             help="The judge reports how confident it is in its safety assessment (0\u20131). "
-            "Verdicts below this threshold are flagged as low-confidence. Future versions "
-            "can use this for auto-approval of high-confidence safe verdicts.",
+            "When Smart Approvals (judge.smart_approvals) is enabled, a tool call is "
+            "auto-approved only if the LLM verdict recommends 'approve' with confidence at or "
+            "above this value. Lower it to auto-approve more aggressively; raise it toward 1.0 "
+            "to require near-certainty. Has no effect while judge.smart_approvals is off.",
         ),
         SettingDef(
             "judge.max_context_ratio",

turnstone/server.py

@@ -4389,6 +4389,7 @@ def main() -> None:
         return JudgeConfig(
             enabled=config_store.get("judge.enabled"),
             model=config_store.get("judge.model"),
+            smart_approvals=config_store.get("judge.smart_approvals"),
             confidence_threshold=config_store.get("judge.confidence_threshold"),
             max_context_ratio=config_store.get("judge.max_context_ratio"),
             timeout=config_store.get("judge.timeout"),

turnstone/ui/static/app.js

@@ -2290,14 +2290,18 @@ class Pane {
 
     items.forEach((item) => {
       block.appendChild(buildToolDiv(item));
-      // Render verdict badge if present.  Server emits the heuristic
-      // verdict under ``heuristic_verdict`` (matches the api/server_schemas
-      // PendingApprovalItem shape).  Falls back to the legacy ``verdict``
-      // key in case a stale SSE payload arrives mid-deploy.
-      const heuristic = item.heuristic_verdict || item.verdict;
-      if (heuristic) {
-        block.appendChild(renderVerdictBadge(heuristic, judgePending));
-        const rec = heuristic.recommendation || "review";
+      // Render verdict badge if present.  Prefer the completed LLM verdict
+      // (``judge_verdict``, set by the server on a Smart-Approved item) so an
+      // auto-approved row shows the llm/approve verdict that cleared it rather
+      // than the cautious heuristic carry-over.  Otherwise the heuristic
+      // verdict (``heuristic_verdict``, matching the api/server_schemas
+      // PendingApprovalItem shape); ``verdict`` is the legacy fallback for a
+      // stale mid-deploy SSE payload.
+      const verdict =
+        item.judge_verdict || item.heuristic_verdict || item.verdict;
+      if (verdict) {
+        block.appendChild(renderVerdictBadge(verdict, judgePending));
+        const rec = verdict.recommendation || "review";
         if (
           !glowRec ||
           rec === "deny" ||