refactor(core): unify approve_tools across both kinds (#436)

* refactor(core): unify approve_tools across kinds + judge visibility + perf

Lift WebUI.approve_tools to SessionUIBase so both interactive and
coordinator workstreams run the same body. The shared body now owns
tool-policy gating, per-tool auto-approve, blanket carve-out for
__budget_override__, activity tagging, heuristic-verdict persistence,
and the approve_request/approval_event blocking pattern. Subclass
hooks layer kind-specific surfaces on top.

This closes the drift the LLM-judge audit flagged on coord — the
judge (heuristic + LLM tier) now sees actual tool args for every
coord tool call instead of empty func_args. spawn_batch projects
the full children list so a malicious mid-batch entry is no longer
hidden.

= Unification core =
- SessionUIBase.approve_tools: lifted body covering policy / per-tool
  auto-approve / blanket / activity tagging / heuristic-verdict
  persistence / approval gate
- _APPROVAL_WAIT_TIMEOUT class constant + _record_judge_metric hook
- WebUI.approve_tools deleted; _record_judge_metric override fires
  per-node MetricsCollector.record_judge_verdict
- ConsoleCoordinatorUI.approve_tools deleted; _record_judge_metric
  + on_intent_verdict overrides fire ConsoleMetrics.record_judge_verdict
- ConsoleMetrics.record_judge_verdict + turnstone_judge_verdicts_total
  in /metrics text output (cluster PromQL rolls coord+interactive up
  uniformly)
- _console_metrics class attribute wired in console lifespan
- Frontend: coord SSE event tools_auto_approved -> tool_info for parity

= Judge args visibility =
- _evaluate_intent populates func_args for all coord tools that hit
  approval (spawn_workstream / spawn_batch / send_to_workstream /
  close_workstream / close_all_children / cancel_workstream /
  delete_workstream / task_list)
- spawn_batch projects every child's skill / initial_message[:200] /
  target_node so the judge sees the full fan-out (was first child only)
- fire_judge_verdict_metric helper collapses 4 sites of identical
  record_judge_verdict shape across WebUI + ConsoleCoordinatorUI

= Hardening =
- __budget_override__ carve-out reads from pre-filter items list, not
  post-filter pending; policy block skips matching the synthetic
  name entirely so a wildcard `*: allow` cannot strip the override
  before the gate sees it
- _persist_intent_verdict default_tier parameter so heuristic + llm
  paths share the storage write helper

= Performance =
- TTL cache on list_tool_policies in turnstone/core/policy.py
  (60s, keyed by org_id, lock-free hits)
- Storage-layer invalidation: create/update/delete_tool_policy on
  both SQLite and PostgreSQL backends call invalidate_policy_cache
  (covers admin-API path + direct test fixtures + any future caller)
- Admin-API handlers also call invalidate_policy_cache as
  defense-in-depth
- storage.create_intent_verdicts_bulk on both backends: one
  multi-row INSERT + one commit instead of N round-trips. approve_tools
  switches to the bulk path so a fan-out turn no longer pays N x commit
  before the approval prompt enqueues
- _persist_intent_verdicts_bulk helper on SessionUIBase

= Test coverage =
- tests/test_coord_ui_approve_tools.py (NEW, 17 cases): inheritance
  regression, tool-policy deny/allow/mixed on coord, heuristic verdict
  persistence (bulk path), activity tagging on auto-approve and pending,
  judge_pending dynamic flag (true + false), event-name parity,
  per-tool auto-approve, __budget_override__ carve-out under blanket
  + wildcard policy, _record_judge_metric wired/unwired, on_intent_verdict
  llm-tier metric
- tests/test_console_metrics.py: 3 cases for the new
  record_judge_verdict counter
- tests/test_judge_storage.py: 3 cases for create_intent_verdicts_bulk
- tests/test_coordinator_tools.py: 3 cases pinning the spawn_batch
  full-children projection (truncation, mid-batch visibility, empty
  defensive)
- tests/conftest.py: autouse _clear_policy_cache fixture so the
  process-level cache doesn't leak between tests with distinct storage
  instances

= Drift fixes (review feedback) =
- Refresh stale "no-op on coord" comments now that coord overrides
  the hook
- WebUI.on_plan_review timeout uses self._APPROVAL_WAIT_TIMEOUT
  instead of literal 3600
- Drop redundant bool() wrapper around any() in judge_pending
- Rephrase broken docstring grammar in _coord_spawn_metrics
- Hoist redundant get_storage import out of approve_tools per-item loop
  (folded into _persist_intent_verdicts_bulk helper)

= Validation =
- pytest -m "not live": 4679 passed, 3 deselected
- ruff check + ruff format: clean
- mypy: no issues in 175 source files

* fix(approval): apply Copilot feedback on PR #436

- Policy-cache invalidation now drops both the org-scoped slot AND the
  default ``""`` slot on ``create_tool_policy`` for both SQLite and
  PostgreSQL backends. ``list_tool_policies("")`` returns rows from
  every org_id, and the production evaluators (SessionUIBase.approve_tools
  / cli.py) read with the default ``org_id=""``, so an org-scoped insert
  that only invalidated its own slot would leave the default cache slot
  stale until the TTL window expired.
- Cap ``reason`` to 200 chars in ``_evaluate_intent`` for ``close_workstream``
  and ``close_all_children`` — both fields are LLM/user-provided and the
  preparer doesn't size-limit them, so an unbounded reason could bloat
  the persisted verdict row's func_args. Matches the cap applied to other
  free-form coord tool fields (initial_message, message, title).
- Refresh ``_PolicyCache`` docstring: it claimed lock-free reads on
  cache hit but ``get()`` always acquires ``self._lock``. Updated to
  reflect that the lock is held briefly to copy the policies reference.

Validation: targeted suite 201/201, ruff + mypy clean.

tests/conftest.py

@@ -95,3 +95,21 @@ def mock_openai_client():
     client = MagicMock()
     client.models.list.return_value.data = [MagicMock(id="test-model")]
     return client
+
+
+@pytest.fixture(autouse=True)
+def _clear_policy_cache():
+    """Drop the in-process tool-policy cache between tests.
+
+    The cache is keyed by org_id (default ``""``), so without this
+    autouse hook a policy created in test A would leak into test B's
+    ``evaluate_tool_policy`` call — distinct storage instances, same
+    cache slot. Production singleton storage doesn't see the leak
+    because there's only one storage instance for the process lifetime;
+    the test isolation requirement is what motivates the autouse.
+    """
+    from turnstone.core.policy import invalidate_policy_cache
+
+    invalidate_policy_cache()
+    yield
+    invalidate_policy_cache()

tests/test_console_metrics.py

@@ -37,6 +37,33 @@ class TestRecordRoute:
         assert "turnstone_router_request_duration_seconds_sum" in text
 
 
+class TestRecordJudgeVerdict:
+    """Coord-side intent-judge verdict counter."""
+
+    def test_single_verdict(self) -> None:
+        m = ConsoleMetrics()
+        m.record_judge_verdict("heuristic", "high", 12)
+
+        text = m.generate_text()
+        assert 'turnstone_judge_verdicts_total{tier="heuristic",risk_level="high"} 1' in text
+
+    def test_aggregates_by_tier_and_risk(self) -> None:
+        m = ConsoleMetrics()
+        m.record_judge_verdict("heuristic", "low", 5)
+        m.record_judge_verdict("heuristic", "low", 7)
+        m.record_judge_verdict("llm", "high", 250)
+
+        text = m.generate_text()
+        assert 'turnstone_judge_verdicts_total{tier="heuristic",risk_level="low"} 2' in text
+        assert 'turnstone_judge_verdicts_total{tier="llm",risk_level="high"} 1' in text
+
+    def test_section_omitted_when_empty(self) -> None:
+        """No verdicts recorded → don't emit the empty header block."""
+        m = ConsoleMetrics()
+        text = m.generate_text()
+        assert "turnstone_judge_verdicts_total" not in text
+
+
 class TestRouterInfo:
     """Live-membership gauge + refresh counter."""
 

tests/test_coord_ui_approve_tools.py

@@ -0,0 +1,466 @@
+"""Tests for the unified ``approve_tools`` body, viewed from the coord side.
+
+The body itself is exercised by ``test_webui_auto_approve_visibility``;
+this file pins down the coord-specific contracts that lifting the body
+to ``SessionUIBase`` automatically enables:
+
+- Tool-policy gating now applies to coord tool calls (was interactive-only).
+- Heuristic verdicts persist on coord (was interactive-only).
+- The activity tag fields populate on coord during pending approval.
+- ``judge_pending`` is dynamic on the coord ``approve_request``
+  (was hardcoded ``False``).
+- The auto-approve fall-through emits ``tool_info`` (was
+  ``tools_auto_approved``).
+- ``_record_judge_metric`` is a no-op on coord (no Prometheus on console).
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from turnstone.console.coordinator_ui import ConsoleCoordinatorUI
+
+
+def _make_items(*specs: tuple[str, str], needs_approval: bool = True) -> list[dict[str, Any]]:
+    return [
+        {
+            "call_id": call_id,
+            "header": f"Tool: {func}",
+            "preview": "preview text",
+            "func_name": func,
+            "approval_label": func,
+            "needs_approval": needs_approval,
+        }
+        for call_id, func in specs
+    ]
+
+
+def _patch_storage(storage: Any):
+    return patch("turnstone.core.storage._registry.get_storage", return_value=storage)
+
+
+def _patch_policies(verdicts: dict[str, str]):
+    return patch(
+        "turnstone.core.policy.evaluate_tool_policies_batch",
+        return_value=verdicts,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Inheritance regression — the unification itself
+# ---------------------------------------------------------------------------
+
+
+def test_coord_inherits_approve_tools_from_base() -> None:
+    """``ConsoleCoordinatorUI`` must NOT define its own ``approve_tools``;
+    the shared body lives on :class:`SessionUIBase`. A future drift —
+    adding a coord-only override — is exactly the kind of bug this
+    unification is meant to prevent, so guard it explicitly."""
+    assert "approve_tools" not in ConsoleCoordinatorUI.__dict__, (
+        "ConsoleCoordinatorUI shouldn't redefine approve_tools — "
+        "the shared body on SessionUIBase covers both kinds."
+    )
+    assert ConsoleCoordinatorUI.approve_tools.__qualname__ == "SessionUIBase.approve_tools"
+
+
+# ---------------------------------------------------------------------------
+# Tool-policy gating now applies to coord
+# ---------------------------------------------------------------------------
+
+
+def test_coord_tool_policy_deny_blocks_coord_tool() -> None:
+    """Admin-defined ``deny`` policies now fire on coord tool calls.
+    Pre-lift this was interactive-only; an admin who wanted to block
+    e.g. ``delete_workstream`` on the coord couldn't."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    items = _make_items(("c1", "delete_workstream"))
+
+    storage = MagicMock()
+    with _patch_storage(storage), _patch_policies({"delete_workstream": "deny"}):
+        approved, err = ui.approve_tools(items)
+
+    assert approved is False
+    assert err == "Blocked by tool policy"
+    assert items[0].get("denied") is True
+
+
+def test_coord_tool_policy_allow_tags_with_policy_source() -> None:
+    """Admin ``allow`` rule auto-approves the item with
+    ``AutoApproveReason.POLICY``. This was a no-op on coord pre-lift."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    items = _make_items(("c1", "spawn_workstream"))
+
+    storage = MagicMock()
+    with _patch_storage(storage), _patch_policies({"spawn_workstream": "allow"}):
+        approved, _err = ui.approve_tools(items)
+
+    assert approved is True
+    snapshot = ui.serialize_recent_auto_approvals()
+    assert len(snapshot) == 1
+    assert snapshot[0]["func_name"] == "spawn_workstream"
+    assert snapshot[0]["auto_approve_reason"] == "policy"
+
+
+def test_coord_tool_policy_mixed_allow_deny_records_allowed_sibling() -> None:
+    """Same ``mixed-policy`` audit-leak fix that
+    ``test_webui_auto_approve_visibility`` validates for interactive,
+    now auto-applies to coord via the lifted body."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    items = _make_items(("c1", "delete_workstream"), ("c2", "list_workstreams"))
+
+    storage = MagicMock()
+    with (
+        _patch_storage(storage),
+        _patch_policies({"delete_workstream": "deny", "list_workstreams": "allow"}),
+    ):
+        approved, _err = ui.approve_tools(items)
+
+    assert approved is False
+    snapshot = ui.serialize_recent_auto_approvals()
+    assert len(snapshot) == 1
+    assert snapshot[0]["func_name"] == "list_workstreams"
+    assert snapshot[0]["auto_approve_reason"] == "policy"
+
+
+# ---------------------------------------------------------------------------
+# Heuristic-verdict persistence + metric hook
+# ---------------------------------------------------------------------------
+
+
+def test_coord_heuristic_verdict_persists_to_storage() -> None:
+    """Heuristic verdicts attached to items now flow through to
+    ``storage.create_intent_verdicts_bulk`` on coord. Pre-lift coord
+    silently dropped them; only LLM-tier verdicts (from the daemon
+    judge thread via ``on_intent_verdict``) reached storage. Post
+    perf-2 the path uses bulk INSERT so a fan-out turn pays one commit
+    instead of N."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    hv = {
+        "verdict_id": "v1",
+        "call_id": "c1",
+        "func_name": "spawn_workstream",
+        "tier": "heuristic",
+        "risk_level": "high",
+        "confidence": 0.75,
+        "recommendation": "review",
+        "reasoning": "spawning child with bash skill",
+        "evidence": ["bash"],
+        "latency_ms": 12,
+    }
+    items = _make_items(("c1", "spawn_workstream"))
+    items[0]["_heuristic_verdict"] = hv
+
+    storage = MagicMock()
+    timer = threading.Timer(0.05, lambda: ui.resolve_approval(False))
+    timer.start()
+    try:
+        with _patch_storage(storage):
+            ui.approve_tools(items)
+    finally:
+        timer.cancel()
+
+    storage.create_intent_verdicts_bulk.assert_called_once()
+    rows = storage.create_intent_verdicts_bulk.call_args.args[0]
+    assert len(rows) == 1
+    assert rows[0]["verdict_id"] == "v1"
+    assert rows[0]["tier"] == "heuristic"
+    assert rows[0]["ws_id"] == "coord-1"
+
+
+def test_coord_record_judge_metric_fires_console_metrics() -> None:
+    """``_record_judge_metric`` increments the console's
+    ``ConsoleMetrics`` judge counter when the class attribute is wired,
+    so coord verdicts surface on the console's /metrics endpoint
+    alongside the per-node series."""
+    from turnstone.console.metrics import ConsoleMetrics
+
+    cm = ConsoleMetrics()
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    try:
+        ConsoleCoordinatorUI._console_metrics = cm
+        ui._record_judge_metric({"tier": "heuristic", "risk_level": "high", "latency_ms": 12})
+    finally:
+        ConsoleCoordinatorUI._console_metrics = None
+
+    text = cm.generate_text()
+    assert 'turnstone_judge_verdicts_total{tier="heuristic",risk_level="high"} 1' in text
+
+
+def test_coord_record_judge_metric_safe_when_unwired() -> None:
+    """No /metrics instance set → silent no-op. Test fixtures that
+    don't spin up a full console app must not crash on judge
+    verdicts during the shared ``approve_tools`` body."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    # Sanity: class attribute is None at module import time outside
+    # the lifespan — exactly the test-fixture state.
+    assert ConsoleCoordinatorUI._console_metrics is None
+    # Should not raise.
+    ui._record_judge_metric({"tier": "heuristic", "risk_level": "low"})
+
+
+def test_coord_on_intent_verdict_fires_metric_for_llm_tier() -> None:
+    """Async LLM verdicts from the daemon judge thread land at
+    ``on_intent_verdict``. Coord overrides it to fire the same
+    ``record_judge_verdict`` call WebUI does — different tier label,
+    same cluster-wide histogram."""
+    from turnstone.console.metrics import ConsoleMetrics
+
+    cm = ConsoleMetrics()
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    try:
+        ConsoleCoordinatorUI._console_metrics = cm
+        with _patch_storage(MagicMock()):
+            ui.on_intent_verdict(
+                {
+                    "verdict_id": "v1",
+                    "call_id": "c1",
+                    "tier": "llm",
+                    "risk_level": "medium",
+                    "latency_ms": 250,
+                }
+            )
+    finally:
+        ConsoleCoordinatorUI._console_metrics = None
+
+    text = cm.generate_text()
+    assert 'turnstone_judge_verdicts_total{tier="llm",risk_level="medium"} 1' in text
+
+
+# ---------------------------------------------------------------------------
+# Activity tagging during pending approval
+# ---------------------------------------------------------------------------
+
+
+def test_coord_pending_approval_sets_activity_tag() -> None:
+    """The shared body tags ``_ws_current_activity`` /
+    ``_ws_activity_state`` so the cluster collector's coord-row
+    snapshot reflects the approval wait. Pre-lift coord left these
+    fields empty during pending approval."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    items = _make_items(("c1", "delete_workstream"))
+
+    captured: dict[str, str] = {}
+
+    def _capture_activity() -> None:
+        captured["activity"] = ui._ws_current_activity
+        captured["state"] = ui._ws_activity_state
+        ui.resolve_approval(False)
+
+    timer = threading.Timer(0.05, _capture_activity)
+    timer.start()
+    try:
+        with _patch_storage(MagicMock()):
+            ui.approve_tools(items)
+    finally:
+        timer.cancel()
+
+    assert "Awaiting approval" in captured["activity"]
+    assert "delete_workstream" in captured["activity"]
+    assert captured["state"] == "approval"
+
+
+def test_coord_auto_approve_sets_tool_activity_tag() -> None:
+    """Blanket auto-approve flips activity to the ``⚙ {tool}: {preview}``
+    shape WebUI has used; coord row now mirrors it."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    ui.auto_approve = True
+    items = _make_items(("c1", "spawn_workstream"))
+
+    with _patch_storage(MagicMock()):
+        approved, _err = ui.approve_tools(items)
+
+    assert approved is True
+    assert "spawn_workstream" in ui._ws_current_activity
+    assert ui._ws_activity_state == "tool"
+
+
+# ---------------------------------------------------------------------------
+# judge_pending flag + event-name parity
+# ---------------------------------------------------------------------------
+
+
+def test_coord_judge_pending_flag_dynamic_when_heuristic_present() -> None:
+    """Pre-lift coord hardcoded ``judge_pending=False`` on every
+    ``approve_request``; the unified body computes the bool from the
+    items, matching WebUI."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    items = _make_items(("c1", "spawn_workstream"))
+    items[0]["_heuristic_verdict"] = {"verdict_id": "v1", "tier": "heuristic"}
+
+    captured_events: list[dict[str, Any]] = []
+    ui._enqueue = captured_events.append  # type: ignore[method-assign]
+
+    timer = threading.Timer(0.05, lambda: ui.resolve_approval(False))
+    timer.start()
+    try:
+        with _patch_storage(MagicMock()):
+            ui.approve_tools(items)
+    finally:
+        timer.cancel()
+
+    approve_requests = [e for e in captured_events if e.get("type") == "approve_request"]
+    assert len(approve_requests) == 1
+    assert approve_requests[0]["judge_pending"] is True
+
+
+def test_coord_blanket_auto_approve_emits_tool_info() -> None:
+    """Event-name parity: the auto-approve fall-through emits
+    ``tool_info`` for both kinds. Pre-lift coord emitted
+    ``tools_auto_approved`` — the rename happens implicitly via
+    inheritance."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    ui.auto_approve = True
+    items = _make_items(("c1", "spawn_workstream"))
+
+    captured_events: list[dict[str, Any]] = []
+    ui._enqueue = captured_events.append  # type: ignore[method-assign]
+
+    with _patch_storage(MagicMock()):
+        ui.approve_tools(items)
+
+    types = [e.get("type") for e in captured_events]
+    assert "tool_info" in types
+    assert "tools_auto_approved" not in types
+
+
+def test_coord_judge_pending_false_when_no_heuristic_verdict() -> None:
+    """Counterpart to ``test_coord_judge_pending_flag_dynamic_when_heuristic_present``:
+    items with no ``_heuristic_verdict`` produce ``approve_request`` with
+    ``judge_pending=False``. Without this case pinned, a regression that
+    hardcodes ``judge_pending=True`` (the inverse of the pre-lift coord
+    bug) would slip through unnoticed."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    items = _make_items(("c1", "spawn_workstream"))
+    # Deliberately no _heuristic_verdict on any item.
+
+    captured_events: list[dict[str, Any]] = []
+    ui._enqueue = captured_events.append  # type: ignore[method-assign]
+
+    timer = threading.Timer(0.05, lambda: ui.resolve_approval(False))
+    timer.start()
+    try:
+        with _patch_storage(MagicMock()):
+            ui.approve_tools(items)
+    finally:
+        timer.cancel()
+
+    approve_requests = [e for e in captured_events if e.get("type") == "approve_request"]
+    assert len(approve_requests) == 1
+    assert approve_requests[0]["judge_pending"] is False
+
+
+# ---------------------------------------------------------------------------
+# Per-tool auto-approve via auto_approve_tools (set membership)
+# ---------------------------------------------------------------------------
+
+
+def test_coord_per_tool_auto_approve_tags_with_source() -> None:
+    """When a coord tool name lands in ``auto_approve_tools`` (e.g. via a
+    skill template's ``allowed_tools``), the lifted body short-circuits
+    the prompt and tags the item with ``AutoApproveReason.AUTO_APPROVE_TOOLS``
+    (or the per-tool source from ``_auto_approve_tools_source``).
+    Mirrors the WebUI test ``test_auto_approve_tools_skill_source_renders_as_skill``
+    on the coord side so the unified body gains parity coverage."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    ui.auto_approve_tools = {"spawn_workstream"}
+    ui._auto_approve_tools_source = {"spawn_workstream": "skill"}
+    items = _make_items(("c1", "spawn_workstream"))
+
+    storage = MagicMock()
+    with _patch_storage(storage):
+        approved, _err = ui.approve_tools(items)
+
+    assert approved is True
+    snapshot = ui.serialize_recent_auto_approvals()
+    assert len(snapshot) == 1
+    assert snapshot[0]["func_name"] == "spawn_workstream"
+    assert snapshot[0]["auto_approve_reason"] == "skill"
+
+
+# ---------------------------------------------------------------------------
+# __budget_override__ carve-out — sec-2 hardening
+# ---------------------------------------------------------------------------
+
+
+def test_coord_budget_override_prompts_even_under_blanket_auto_approve() -> None:
+    """The carve-out promises ``__budget_override__`` always prompts the
+    operator. Pin that behavior on the coord side so a future regression
+    of the post-filter / pre-filter check (sec-2) gets caught.
+
+    ``__budget_override__`` is interactive-only today (coord workstreams
+    don't have token budgets), but the synthetic item can be threaded
+    through ``approve_tools`` directly the same way ``ChatSession.send``
+    does on the interactive side. The carve-out fires uniformly across
+    both kinds."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    ui.auto_approve = True  # blanket flag — should NOT bypass the carve-out
+    items = [
+        {
+            "call_id": "c1",
+            "header": "Token budget exhausted",
+            "preview": "Token budget (200,000) exhausted. Approve to continue.",
+            "func_name": "__budget_override__",
+            "approval_label": "__budget_override__",
+            "needs_approval": True,
+        }
+    ]
+
+    captured_events: list[dict[str, Any]] = []
+    ui._enqueue = captured_events.append  # type: ignore[method-assign]
+
+    timer = threading.Timer(0.05, lambda: ui.resolve_approval(True))
+    timer.start()
+    try:
+        with _patch_storage(MagicMock()):
+            approved, _err = ui.approve_tools(items)
+    finally:
+        timer.cancel()
+
+    assert approved is True
+    # The carve-out forces the prompt path, NOT the auto-approve fall-through.
+    types = [e.get("type") for e in captured_events]
+    assert "approve_request" in types, (
+        "Budget override must produce an approve_request even under blanket auto_approve"
+    )
+    assert "tool_info" not in types, (
+        "Auto-approve fall-through must not fire when a budget override is present"
+    )
+
+
+def test_coord_budget_override_survives_wildcard_allow_policy() -> None:
+    """A wildcard ``*: allow`` policy must not strip ``__budget_override__``
+    from the gate. Pre-sec-2, the policy block could mark the item
+    ``needs_approval=False`` and remove it from ``pending``, after which
+    the carve-out (which read ``pending``) would see no override and
+    blanket auto-approve would silently fire. Post-fix the carve-out
+    reads from the pre-filter ``items`` list AND the policy block skips
+    matching the synthetic name entirely."""
+    ui = ConsoleCoordinatorUI(ws_id="coord-1", user_id="u1")
+    ui.auto_approve = True
+    items = [
+        {
+            "call_id": "c1",
+            "header": "Token budget exhausted",
+            "preview": "Token budget exhausted. Approve to continue.",
+            "func_name": "__budget_override__",
+            "approval_label": "__budget_override__",
+            "needs_approval": True,
+        }
+    ]
+
+    captured_events: list[dict[str, Any]] = []
+    ui._enqueue = captured_events.append  # type: ignore[method-assign]
+
+    timer = threading.Timer(0.05, lambda: ui.resolve_approval(True))
+    timer.start()
+    try:
+        with _patch_storage(MagicMock()), _patch_policies({"__budget_override__": "allow"}):
+            approved, _err = ui.approve_tools(items)
+    finally:
+        timer.cancel()
+
+    assert approved is True
+    types = [e.get("type") for e in captured_events]
+    assert "approve_request" in types, "Wildcard allow must not strip the budget-override prompt"

tests/test_coordinator_tools.py

@@ -1150,6 +1150,107 @@ def test_spawn_batch_exec_emits_batch_started_and_ended(coord_session):
     assert events[-1]["denied"] == 0
 
 
+# ---------------------------------------------------------------------------
+# spawn_batch — _evaluate_intent func_args projection (sec-3 follow-up)
+# ---------------------------------------------------------------------------
+#
+# The judge (heuristic + LLM) reads ``item["func_args"]`` to reason about
+# what the coordinator is about to do. Pre-fix, spawn_batch projected only
+# the FIRST child's skill + initial_message — a malicious mid-batch entry
+# was invisible to both tiers. These tests pin the full-children projection.
+
+
+def _stub_judge_for_evaluate_intent(monkeypatch, sess):
+    """Stub _ensure_judge so _evaluate_intent's setup loop runs.
+
+    The actual judge.evaluate() is mocked to return one verdict per item
+    so the heuristic-attach loop doesn't IndexError. Tests assert on the
+    func_args populated BEFORE judge.evaluate is invoked.
+    """
+    fake_verdict = MagicMock()
+    fake_verdict.to_dict.return_value = {"verdict_id": "v0", "tier": "heuristic"}
+    fake_judge = MagicMock()
+    # judge.evaluate(items, messages, callback=, cancel_event=) → list[verdict]
+    fake_judge.evaluate.side_effect = lambda items, *_args, **_kw: [fake_verdict] * len(items)
+    monkeypatch.setattr(sess, "_ensure_judge", lambda: fake_judge)
+    return fake_judge
+
+
+def test_spawn_batch_evaluate_intent_projects_all_children(coord_session, monkeypatch):
+    sess, _coord, _ui = coord_session
+    _stub_judge_for_evaluate_intent(monkeypatch, sess)
+    item = sess._prepare_tool(
+        _tc(
+            "spawn_batch",
+            {
+                "children": [
+                    {"initial_message": "audit auth.py for CSRF", "skill": "engineer"},
+                    {"initial_message": "rm -rf the docs tree", "skill": "bash-runner"},
+                    {
+                        "initial_message": "compare FastAPI vs Starlette",
+                        "skill": "researcher",
+                        "target_node": "node-7",
+                    },
+                ]
+            },
+        )
+    )
+    sess._evaluate_intent([item])
+
+    fa = item["func_args"]
+    assert fa["child_count"] == 3
+    children = fa["children"]
+    assert len(children) == 3
+    assert children[0]["skill"] == "engineer"
+    assert children[0]["initial_message"] == "audit auth.py for CSRF"
+    assert children[0]["target_node"] == ""
+    # Mid-batch entry is fully visible — the bug this fix exists to close.
+    assert children[1]["skill"] == "bash-runner"
+    assert children[1]["initial_message"] == "rm -rf the docs tree"
+    assert children[2]["skill"] == "researcher"
+    assert children[2]["target_node"] == "node-7"
+
+
+def test_spawn_batch_evaluate_intent_truncates_long_messages(coord_session, monkeypatch):
+    sess, _coord, _ui = coord_session
+    _stub_judge_for_evaluate_intent(monkeypatch, sess)
+    long_msg = "x" * 500
+    item = sess._prepare_tool(
+        _tc("spawn_batch", {"children": [{"initial_message": long_msg, "skill": "researcher"}]})
+    )
+    sess._evaluate_intent([item])
+
+    children = item["func_args"]["children"]
+    assert len(children) == 1
+    # Cap is 200 chars — same shape every other coord-tool projection uses.
+    assert len(children[0]["initial_message"]) == 200
+    assert children[0]["initial_message"] == "x" * 200
+
+
+def test_spawn_batch_evaluate_intent_handles_empty_children_defensively(coord_session, monkeypatch):
+    """``_prepare_spawn_batch`` rejects an empty children list before this
+    code runs, so we shouldn't reach _evaluate_intent with one in
+    practice — but if a future caller bypasses the preparer the
+    projection must still produce a valid dict. Pinning the defensive
+    shape so the JSON-serialised verdict row stays well-formed."""
+    sess, _coord, _ui = coord_session
+    _stub_judge_for_evaluate_intent(monkeypatch, sess)
+    # Synthesise an item directly — bypassing _prepare_tool, since the
+    # preparer's empty-list rejection would prevent us reaching here.
+    fake_item = {
+        "call_id": "call-empty",
+        "func_name": "spawn_batch",
+        "needs_approval": True,
+        "approval_label": "spawn_batch",
+        "children": [],
+    }
+    sess._evaluate_intent([fake_item])
+
+    fa = fake_item["func_args"]
+    assert fa["child_count"] == 0
+    assert fa["children"] == []
+
+
 # ---------------------------------------------------------------------------
 # close_all_children
 # ---------------------------------------------------------------------------

tests/test_judge_storage.py

@@ -114,6 +114,59 @@ class TestIntentVerdictCRUD:
         assert ok is False
 
 
+# ---------------------------------------------------------------------------
+# Bulk insert
+# ---------------------------------------------------------------------------
+
+
+class TestIntentVerdictBulkInsert:
+    """Coverage for ``create_intent_verdicts_bulk`` — backs the
+    ``approve_tools`` per-turn heuristic-verdict persistence path so a
+    fan-out turn pays one commit instead of N.
+    """
+
+    def test_bulk_insert_creates_all_rows(self, db):
+        db.create_intent_verdicts_bulk(
+            [
+                _make_verdict_kwargs(verdict_id="b1", call_id="c1"),
+                _make_verdict_kwargs(verdict_id="b2", call_id="c2"),
+                _make_verdict_kwargs(verdict_id="b3", call_id="c3"),
+            ]
+        )
+        for vid in ("b1", "b2", "b3"):
+            v = db.get_intent_verdict(vid)
+            assert v is not None
+            assert v["verdict_id"] == vid
+
+    def test_bulk_insert_empty_list_is_noop(self, db):
+        # Must not raise and must not commit a phantom row.
+        db.create_intent_verdicts_bulk([])
+        assert db.list_intent_verdicts() == []
+
+    def test_bulk_insert_preserves_distinct_field_values(self, db):
+        db.create_intent_verdicts_bulk(
+            [
+                _make_verdict_kwargs(
+                    verdict_id="b1",
+                    risk_level="low",
+                    tier="heuristic",
+                    confidence=0.4,
+                ),
+                _make_verdict_kwargs(
+                    verdict_id="b2",
+                    risk_level="high",
+                    tier="llm",
+                    confidence=0.95,
+                ),
+            ]
+        )
+        v1 = db.get_intent_verdict("b1")
+        v2 = db.get_intent_verdict("b2")
+        assert v1 is not None and v2 is not None
+        assert v1["risk_level"] == "low" and v1["tier"] == "heuristic"
+        assert v2["risk_level"] == "high" and v2["tier"] == "llm"
+
+
 # ---------------------------------------------------------------------------
 # List queries
 # ---------------------------------------------------------------------------

turnstone/console/coordinator_ui.py

@@ -20,8 +20,12 @@ Mirrors ``turnstone.server.WebUI`` but scoped to the console's needs:
   broadcasts route through the cluster collector instead
   (``coord_adapter.emit_state`` for state changes;
   :meth:`_broadcast_activity` override for live activity ticks).
-- No per-node Prometheus metrics — the console has no /metrics
-  endpoint. WebUI's ``_metrics.record_*`` calls don't apply here.
+- Console-side Prometheus metrics — the console exposes ``/metrics``
+  backed by :class:`ConsoleMetrics` (lighter than the per-node
+  :class:`MetricsCollector` but the judge-verdict counter is parity
+  shape so a cluster-wide PromQL query rolls up coord + interactive
+  uniformly). Wired here via the ``_console_metrics`` class attribute
+  set at console startup.
 
 Contract: this class must conform to :class:`turnstone.core.session.SessionUI`.
 """
@@ -31,21 +35,16 @@ from __future__ import annotations
 from typing import TYPE_CHECKING, Any
 
 from turnstone.core.log import get_logger
-from turnstone.core.session_ui_base import AutoApproveReason, SessionUIBase
+from turnstone.core.session_ui_base import SessionUIBase, fire_judge_verdict_metric
 from turnstone.core.workstream import WorkstreamState
 
 if TYPE_CHECKING:
     from turnstone.console.collector import ClusterCollector
+    from turnstone.console.metrics import ConsoleMetrics
     from turnstone.core.session_manager import SessionManager
 
 log = get_logger(__name__)
 
-# Hard cap on how long a worker thread blocks waiting for an approval /
-# plan-review decision.  Exported as a constant so both blocking paths
-# stay in lockstep and a future `coordinator.approval_timeout_seconds`
-# setting can swap the literal.
-_APPROVAL_WAIT_TIMEOUT = 3600
-
 
 class ConsoleCoordinatorUI(SessionUIBase):
     """SessionUI for a single coordinator session in the console.
@@ -68,6 +67,13 @@ class ConsoleCoordinatorUI(SessionUIBase):
     # with CoordinatorManager; this replaces it without reviving the
     # closure-per-install pattern).
     _collector: ClusterCollector | None = None
+    # Shared reference to the console's :class:`ConsoleMetrics`
+    # instance. Set at console startup so ``_record_judge_metric`` and
+    # ``on_intent_verdict`` can fire ``turnstone_judge_verdicts_total``
+    # the same way the per-node ``WebUI`` does. ``None`` until the
+    # lifespan wires it (and during tests that don't spin up the full
+    # console app).
+    _console_metrics: ConsoleMetrics | None = None
 
     # ------------------------------------------------------------------
     # SessionUI protocol — streaming
@@ -87,99 +93,16 @@ class ConsoleCoordinatorUI(SessionUIBase):
 
     # ------------------------------------------------------------------
     # SessionUI protocol — approvals
+    #
+    # ``approve_tools`` / ``resolve_approval`` / ``resolve_plan`` are
+    # inherited from :class:`SessionUIBase`. The shared body covers
+    # tool-policy gating, per-tool auto-approve, blanket auto-approve,
+    # heuristic-verdict persistence, and activity tagging the same way
+    # interactive sessions get them. ``__budget_override__`` is
+    # interactive-only today; the carve-out in the shared body is a
+    # no-op on coord (coord workstreams don't have token budgets).
     # ------------------------------------------------------------------
 
-    def approve_tools(self, items: list[dict[str, Any]]) -> tuple[bool, str | None]:
-        self._reset_approval_cycle()
-        pending = [it for it in items if it.get("needs_approval") and not it.get("error")]
-
-        if not pending:
-            # Nothing to approve; broadcast tool info anyway so the UI
-            # can render the tool preview.
-            if items:
-                self._enqueue(
-                    {
-                        "type": "tools_auto_approved",
-                        "items": self._serialize_approval_items(items),
-                    }
-                )
-            return True, None
-
-        # Per-tool auto-approve: 'Always approve this tool' adds the
-        # tool name to ``auto_approve_tools``.  This must short-circuit
-        # independently of the blanket ``auto_approve`` flag — matches
-        # the WebUI two-tier contract (turnstone/server.py).
-        if self.auto_approve_tools:
-            # Match WebUI's set-membership key: ``approval_label or
-            # func_name``.  ``auto_approve_tools`` is populated by
-            # both the skill template (bare func_name) and the
-            # "Approve + Always" handler (which tags via approval_label
-            # at session_routes.py).  Pre-fix the coord UI used only
-            # func_name, so an Always-added tool whose approval_label
-            # differs from func_name (e.g. ``skill__name``,
-            # ``mcp_resource__uri``) wouldn't match here and the
-            # operator would get prompted again on the coord page.
-            pending_names = {
-                it.get("approval_label", "") or it.get("func_name", "")
-                for it in pending
-                if it.get("func_name")
-            }
-            if pending_names and pending_names.issubset(self.auto_approve_tools):
-                # Tag for /dashboard visibility — matches WebUI shape so
-                # the coord-tree pill renders the same source string
-                # (``skill`` / ``always`` / generic) for both kinds.
-                self._tag_auto_approved(
-                    pending,
-                    AutoApproveReason.AUTO_APPROVE_TOOLS,
-                    source_map=self._auto_approve_tools_source,
-                )
-                self._record_auto_approves(items)
-                self._enqueue(
-                    {
-                        "type": "tools_auto_approved",
-                        "items": self._serialize_approval_items(items),
-                    }
-                )
-                return True, None
-
-        # Blanket auto-approve (set e.g. during scripted
-        # restart-rehydration) — also matches WebUI semantics.
-        if self.auto_approve:
-            self._tag_auto_approved(pending, AutoApproveReason.BLANKET)
-            self._record_auto_approves(items)
-            self._enqueue(
-                {
-                    "type": "tools_auto_approved",
-                    "items": self._serialize_approval_items(items),
-                }
-            )
-            return True, None
-
-        self._approval_event.clear()
-        self._pending_approval = {
-            "type": "approve_request",
-            "items": self._serialize_approval_items(items),
-            "judge_pending": False,
-        }
-        self._enqueue(self._pending_approval)
-        if not self._approval_event.wait(timeout=_APPROVAL_WAIT_TIMEOUT):
-            log.warning("coord_ui.approval_timeout ws=%s", self.ws_id)
-            self.resolve_approval(False, "Approval timed out after 1 hour")
-        self._pending_approval = None
-        approved, feedback = self._approval_result
-
-        if not approved:
-            denial_msg = "Denied by user"
-            if feedback:
-                denial_msg += f": {feedback}"
-            for item in pending:
-                item["denied"] = True
-                item["denial_msg"] = denial_msg
-
-        return approved, feedback
-
-    # ``resolve_approval`` inherited from :class:`SessionUIBase`.
-
     def on_plan_review(self, content: str) -> str:
         # Coordinator sessions don't fire plan_agent (AGENT_TOOLS is []
         # for coordinator kind) so this path shouldn't normally run.
@@ -187,14 +110,12 @@ class ConsoleCoordinatorUI(SessionUIBase):
         self._plan_event.clear()
         self._pending_plan_review = {"type": "plan_review", "content": content}
         self._enqueue(self._pending_plan_review)
-        if not self._plan_event.wait(timeout=_APPROVAL_WAIT_TIMEOUT):
+        if not self._plan_event.wait(timeout=self._APPROVAL_WAIT_TIMEOUT):
             log.warning("coord_ui.plan_review_timeout ws=%s", self.ws_id)
             self.resolve_plan("reject")
         self._pending_plan_review = None
         return self._plan_result
 
-    # ``resolve_plan`` inherited from :class:`SessionUIBase`.
-
     # ------------------------------------------------------------------
     # SessionUI protocol — broadcast hook + state change + rename
     # ------------------------------------------------------------------
@@ -299,7 +220,37 @@ class ConsoleCoordinatorUI(SessionUIBase):
                     exc_info=True,
                 )
 
-    # ``on_intent_verdict`` and ``on_output_warning`` inherited from
-    # :class:`SessionUIBase`. Coordinator sessions now persist verdicts
-    # and output assessments to storage alongside the interactive path
-    # (the "skip the persistence" deferral note has been retired).
+    # ``on_output_warning`` inherited from :class:`SessionUIBase`.
+    # Coordinator sessions persist verdicts and output assessments to
+    # storage alongside the interactive path.
+
+    # ------------------------------------------------------------------
+    # Prometheus metric hooks — fire ``turnstone_judge_verdicts_total``
+    # against the console's :class:`ConsoleMetrics` instance so the
+    # console's /metrics endpoint surfaces coord verdicts the same way
+    # the per-node /metrics surfaces interactive ones. Two call sites:
+    #
+    # - :meth:`_record_judge_metric` — heuristic tier, fired from the
+    #   shared ``approve_tools`` body during the synchronous
+    #   approval gate.
+    # - :meth:`on_intent_verdict` — LLM tier, fired by the daemon
+    #   judge thread asynchronously.
+    #
+    # Both end up at ``record_judge_verdict(tier, risk, latency_ms)`` —
+    # mirrors ``WebUI``'s pattern at ``server.py``. ``None`` guard
+    # covers the test-fixture case where the console lifespan didn't
+    # wire the class attribute.
+    # ------------------------------------------------------------------
+
+    def _record_judge_metric(self, verdict: dict[str, Any]) -> None:
+        cm = ConsoleCoordinatorUI._console_metrics
+        if cm is None:
+            return
+        fire_judge_verdict_metric(cm, verdict, "heuristic")
+
+    def on_intent_verdict(self, verdict: dict[str, Any]) -> None:
+        super().on_intent_verdict(verdict)
+        cm = ConsoleCoordinatorUI._console_metrics
+        if cm is None:
+            return
+        fire_judge_verdict_metric(cm, verdict, "llm")

turnstone/console/metrics.py

@@ -21,6 +21,11 @@ class ConsoleMetrics:
         self._router_duration_count: dict[str, int] = defaultdict(int)
         self._router_membership: int = 0
         self._router_refresh_count: int = 0
+        # Judge verdicts on coord workstreams — keyed by (tier, risk_level)
+        # so the dashboard can split heuristic vs llm verdicts and the
+        # alerting rules can fire on a coord-side risk distribution
+        # shift the same way they do on per-node interactive metrics.
+        self._judge_verdicts: dict[tuple[str, str], int] = defaultdict(int)
         self._start_time: float = time.monotonic()
 
     def record_route(self, method: str, status: int, duration: float) -> None:
@@ -37,6 +42,18 @@ class ConsoleMetrics:
             self._router_membership = membership
             self._router_refresh_count = refresh_count
 
+    def record_judge_verdict(self, tier: str, risk_level: str, latency_ms: int) -> None:
+        """Record an intent-judge verdict on a coord workstream.
+
+        Mirrors the per-node ``MetricsCollector.record_judge_verdict``
+        in ``core/metrics.py``. Latency is currently aggregated only as
+        a counter increment; promote to a histogram if/when the
+        operator dashboard needs distribution shape.
+        """
+        del latency_ms  # parity with core/metrics.py shape; not tracked yet
+        with self._lock:
+            self._judge_verdicts[(tier, risk_level)] += 1
+
     def generate_text(self) -> str:
         """Return Prometheus text exposition format (v0.0.4)."""
         lines: list[str] = []
@@ -47,6 +64,7 @@ class ConsoleMetrics:
             duration_count = dict(self._router_duration_count)
             router_membership = self._router_membership
             router_refresh_count = self._router_refresh_count
+            judge_verdicts = dict(self._judge_verdicts)
 
         # turnstone_router_requests_total
         lines.append("# HELP turnstone_router_requests_total Console-routed requests")
@@ -83,6 +101,17 @@ class ConsoleMetrics:
         lines.append("# TYPE turnstone_router_refresh_total counter")
         lines.append(f"turnstone_router_refresh_total {router_refresh_count}")
 
+        # turnstone_judge_verdicts_total — coord-side intent-judge
+        # verdicts. Same metric name as the per-node series so a
+        # cluster-wide dashboard query rolls them up uniformly.
+        if judge_verdicts:
+            lines.append("# HELP turnstone_judge_verdicts_total Total intent validation verdicts")
+            lines.append("# TYPE turnstone_judge_verdicts_total counter")
+            for (tier, risk), cnt in sorted(judge_verdicts.items()):
+                lines.append(
+                    f'turnstone_judge_verdicts_total{{tier="{tier}",risk_level="{risk}"}} {cnt}'
+                )
+
         lines.append("")  # trailing newline
         return "\n".join(lines)
 

turnstone/console/server.py

@@ -2747,10 +2747,11 @@ def _coord_spawn_metrics(_request: Request, ui: Any) -> None:
     Wired onto :attr:`SessionEndpointConfig.spawn_metrics`. Increments
     ``_ws_messages`` and resets ``_ws_turn_tool_calls`` so the rich
     ``ws_state`` cluster broadcast renders the same per-turn shape
-    coord rows on the dashboard need. Coord doesn't have a Prometheus
-    endpoint to feed (the console isn't a node), so the
-    ``_metrics.record_message_sent()`` call interactive's analog
-    fires is omitted.
+    coord rows on the dashboard need. Console-side Prometheus runs
+    through :class:`ConsoleMetrics` (lighter than the per-node collector
+    — judge verdicts and routing/membership only); the interactive
+    analog ``_metrics.record_message_sent()`` has no console counterpart
+    yet, so this hook only owns the per-UI counter writes.
     """
     if (
         hasattr(ui, "_ws_lock")
@@ -3923,10 +3924,13 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]:
                 coord_adapter.attach(coord_mgr)
                 app.state.coord_state_writer = coord_state_writer
                 # Shared refs so ConsoleCoordinatorUI.on_state_change
-                # flows state transitions through the unified manager
-                # and on_rename fans out to the cluster dashboard.
+                # flows state transitions through the unified manager,
+                # on_rename fans out to the cluster dashboard, and
+                # _record_judge_metric / on_intent_verdict feed the
+                # console's /metrics endpoint with coord verdicts.
                 ConsoleCoordinatorUI._coord_mgr = coord_mgr
                 ConsoleCoordinatorUI._collector = app.state.collector
+                ConsoleCoordinatorUI._console_metrics = app.state.console_metrics
                 app.state.coord_mgr = coord_mgr
                 app.state.coord_adapter = coord_adapter
                 # Wire the cluster-event subscription so the coordinator's
@@ -3981,6 +3985,7 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]:
 
         ConsoleCoordinatorUI._coord_mgr = None
         ConsoleCoordinatorUI._collector = None
+        ConsoleCoordinatorUI._console_metrics = None
     except Exception:
         log.debug("console.coord_ui_refs_reset_failed", exc_info=True)
     await app.state.proxy_sse_client.aclose()
@@ -5350,6 +5355,11 @@ async def admin_create_policy(request: Request) -> JSONResponse:
         enabled=enabled,
         created_by=audit_uid,
     )
+    # Drop the cached policy snapshot so the next ``approve_tools`` read
+    # picks up this rule without waiting for the TTL window to expire.
+    from turnstone.core.policy import invalidate_policy_cache
+
+    invalidate_policy_cache(org_id)
 
     record_audit(
         storage,
@@ -5408,6 +5418,12 @@ async def admin_update_policy(request: Request) -> JSONResponse:
         updates["enabled"] = bool(body["enabled"])
 
     storage.update_tool_policy(policy_id, **updates)
+    # Drop the cached policy snapshot so the next ``approve_tools`` read
+    # picks up this update without waiting for the TTL window. Use the
+    # existing row's org_id so the right slot is invalidated.
+    from turnstone.core.policy import invalidate_policy_cache
+
+    invalidate_policy_cache(existing.get("org_id", "") if isinstance(existing, dict) else None)
 
     audit_uid, ip = _audit_context(request)
     record_audit(storage, audit_uid, "policy.update", "policy", policy_id, updates, ip)
@@ -5435,6 +5451,11 @@ async def admin_delete_policy(request: Request) -> JSONResponse:
         return JSONResponse({"error": "Policy not found"}, status_code=404)
 
     storage.delete_tool_policy(policy_id)
+    # Drop the cached policy snapshot so the next ``approve_tools`` read
+    # stops applying the deleted rule. Use the existing row's org_id.
+    from turnstone.core.policy import invalidate_policy_cache
+
+    invalidate_policy_cache(existing.get("org_id", "") if isinstance(existing, dict) else None)
 
     audit_uid, ip = _audit_context(request)
     record_audit(
@@ -10116,9 +10137,10 @@ def create_app(
         # cluster broadcast (PR #420) reads ``_ws_messages`` and
         # resets ``_ws_turn_tool_calls`` per turn so coord rows render
         # the same activity / per-turn counts interactive rows do.
-        # Coord doesn't fire Prometheus metrics (the console isn't a
-        # node and has no /metrics endpoint), but the per-UI counter
-        # writes match interactive's pattern.
+        # Judge verdicts on coord feed the console's /metrics endpoint
+        # via :class:`ConsoleCoordinatorUI._record_judge_metric` /
+        # ``on_intent_verdict``; this hook only owns the per-UI counter
+        # writes that match interactive's pattern.
         spawn_metrics=_coord_spawn_metrics,
         emit_message_queued=True,
         events_replay=_coord_events_replay,

turnstone/console/static/coordinator/coordinator.js

@@ -908,7 +908,10 @@
           { label: "queued" },
         );
         break;
-      case "tools_auto_approved":
+      case "tool_info":
+        // Renamed from ``tools_auto_approved`` when ``approve_tools``
+        // unified onto SessionUIBase — the shared body emits ``tool_info``
+        // for both kinds, matching the interactive payload name.
         (ev.items || []).forEach((it) => appendToolCall(it));
         break;
       // Child-workstream fan-out routed through the coordinator's own

turnstone/core/policy.py

@@ -2,12 +2,21 @@
 
 Evaluates tool calls against admin-defined policies to determine whether
 a tool should be auto-allowed, denied, or require human approval.
+
+Policies are read through a small in-process TTL cache so the per-turn
+``approve_tools`` path doesn't hit storage on every assistant turn —
+admin-edited policies propagate to new lookups within ``_POLICY_CACHE_TTL``
+seconds. Mutation handlers (``storage.create_tool_policy`` /
+``update_tool_policy`` / ``delete_tool_policy``) call
+:func:`invalidate_policy_cache` for synchronous propagation.
 """
 
 from __future__ import annotations
 
 import fnmatch
-from typing import TYPE_CHECKING
+import threading
+import time
+from typing import TYPE_CHECKING, Any
 
 from turnstone.core.log import get_logger
 
@@ -17,6 +26,70 @@ if TYPE_CHECKING:
 log = get_logger(__name__)
 
 
+# Cache window for ``storage.list_tool_policies`` reads. Admin-edited
+# config — 60s is short enough that a one-off edit lands quickly without
+# manual invalidation, and long enough that a tool-heavy autonomous turn
+# (coord + interactive) doesn't hit storage on every assistant turn.
+_POLICY_CACHE_TTL: float = 60.0
+
+
+class _PolicyCache:
+    """TTL-keyed snapshot of ``storage.list_tool_policies`` per org_id.
+
+    Reads check the cache under ``self._lock`` (briefly held just long
+    enough to copy the policies reference and TTL stamp); on miss the
+    caller fetches outside the lock and writes back under it.
+    Concurrent misses on the same org_id can produce two SELECTs but
+    only one cache slot — last-writer wins, both writers see the same
+    data within a tight window so the benign double-fetch is acceptable.
+    """
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._entries: dict[str, tuple[float, list[dict[str, Any]]]] = {}
+
+    def get(
+        self,
+        storage: StorageBackend,
+        org_id: str,
+    ) -> list[dict[str, Any]] | None:
+        now = time.monotonic()
+        with self._lock:
+            entry = self._entries.get(org_id)
+            if entry is not None:
+                ts, policies = entry
+                if now - ts < _POLICY_CACHE_TTL:
+                    return policies
+        try:
+            policies = storage.list_tool_policies(org_id=org_id)
+        except Exception:
+            log.warning("Failed to load tool policies", exc_info=True)
+            return None
+        with self._lock:
+            self._entries[org_id] = (time.monotonic(), policies)
+        return policies
+
+    def invalidate(self, org_id: str | None = None) -> None:
+        with self._lock:
+            if org_id is None:
+                self._entries.clear()
+            else:
+                self._entries.pop(org_id, None)
+
+
+_cache = _PolicyCache()
+
+
+def invalidate_policy_cache(org_id: str | None = None) -> None:
+    """Drop the cached policy snapshot for ``org_id`` (or all orgs).
+
+    Call after every ``create_tool_policy`` / ``update_tool_policy`` /
+    ``delete_tool_policy`` so the next ``evaluate_*`` reads fresh data.
+    Pass ``None`` for global invalidation (e.g. test teardown).
+    """
+    _cache.invalidate(org_id)
+
+
 def evaluate_tool_policy(
     storage: StorageBackend,
     tool_name: str,
@@ -31,10 +104,8 @@ def evaluate_tool_policy(
     or ``None`` if no policy matches (caller should fall through to the
     default approval behaviour).
     """
-    try:
-        policies = storage.list_tool_policies(org_id=org_id)
-    except Exception:
-        log.warning("Failed to load tool policies", exc_info=True)
+    policies = _cache.get(storage, org_id)
+    if policies is None:
         return None
 
     for policy in policies:
@@ -56,14 +127,12 @@ def evaluate_tool_policies_batch(
     tool_names: list[str],
     org_id: str = "",
 ) -> dict[str, str | None]:
-    """Evaluate policies for multiple tools at once (single DB query).
+    """Evaluate policies for multiple tools at once (single cached read).
 
     Returns a dict mapping each tool name to its policy result.
     """
-    try:
-        policies = storage.list_tool_policies(org_id=org_id)
-    except Exception:
-        log.warning("Failed to load tool policies", exc_info=True)
+    policies = _cache.get(storage, org_id)
+    if policies is None:
         return {name: None for name in tool_names}
 
     results: dict[str, str | None] = {}

turnstone/core/session.py

@@ -3353,6 +3353,65 @@ class ChatSession:
                 it["func_args"] = {"prompt": it.get("prompt", "")[:200]}
             elif name == "plan_agent":
                 it["func_args"] = {"goal": it.get("prompt", "")[:200]}
+            # Coordinator tool args — only the ``needs_approval=True`` set
+            # reaches this point (read-only inspect / list_* / wait
+            # tools are filtered above), so this matches the auditable
+            # surface 1:1. Free-form fields capped to keep the verdict
+            # row size bounded.
+            elif name == "spawn_workstream":
+                it["func_args"] = {
+                    "skill": it.get("skill", ""),
+                    "initial_message": it.get("initial_message", "")[:200],
+                    "target_node": it.get("target_node", ""),
+                    "name": it.get("name", ""),
+                    "model": it.get("model", ""),
+                }
+            elif name == "spawn_batch":
+                # Project every child so the judge sees the full fan-out.
+                # First-child-only projection (the prior shape) hid a
+                # malicious mid-batch entry from both heuristic and LLM
+                # tiers. Tool schema caps ``children`` at 10, so worst
+                # case is ~3 KiB of JSON in the verdict row — comparable
+                # to the existing ``reasoning`` / ``evidence`` fields.
+                # ``name`` (cosmetic) and ``model`` (registry alias)
+                # skipped to keep the payload lean; risk-relevant fields
+                # are skill, initial_message, target_node.
+                children = it.get("children") or []
+                it["func_args"] = {
+                    "child_count": len(children),
+                    "children": [
+                        {
+                            "skill": c.get("skill", "") if isinstance(c, dict) else "",
+                            "initial_message": (
+                                c.get("initial_message", "")[:200] if isinstance(c, dict) else ""
+                            ),
+                            "target_node": (
+                                c.get("target_node", "") if isinstance(c, dict) else ""
+                            ),
+                        }
+                        for c in children
+                    ],
+                }
+            elif name == "send_to_workstream":
+                it["func_args"] = {
+                    "ws_id": it.get("ws_id", ""),
+                    "message": it.get("message", "")[:200],
+                }
+            elif name == "close_workstream":
+                it["func_args"] = {
+                    "ws_id": it.get("ws_id", ""),
+                    "reason": it.get("reason", "")[:200],
+                }
+            elif name == "close_all_children":
+                it["func_args"] = {"reason": it.get("reason", "")[:200]}
+            elif name in ("cancel_workstream", "delete_workstream"):
+                it["func_args"] = {"ws_id": it.get("ws_id", "")}
+            elif name == "task_list":
+                it["func_args"] = {
+                    "action": it.get("action", ""),
+                    "task_id": it.get("task_id", ""),
+                    "title": it.get("title", "")[:100],
+                }
             elif it.get("mcp_args"):
                 it["func_args"] = it["mcp_args"]
 

turnstone/core/session_ui_base.py

@@ -51,6 +51,32 @@ _DEFAULT_LISTENER_QUEUE_MAX = 500
 _MAX_TURN_CONTENT_CHARS = 256 * 1024
 
 
+def fire_judge_verdict_metric(
+    metrics: Any,
+    verdict: dict[str, Any],
+    default_tier: str,
+) -> None:
+    """Fire ``record_judge_verdict`` on the given Prometheus collector.
+
+    Both :class:`turnstone.server.WebUI` (per-node ``MetricsCollector``)
+    and :class:`turnstone.console.coordinator_ui.ConsoleCoordinatorUI`
+    (console-side :class:`ConsoleMetrics`) route their hook overrides
+    through this helper. Pins the ``(tier, risk_level, latency_ms)``
+    extraction shape so a future signature change to
+    ``record_judge_verdict`` lands in one place instead of four.
+
+    ``default_tier`` is the call-site label (``"heuristic"`` or
+    ``"llm"``) used only when the verdict dict doesn't already carry
+    a ``tier`` key — both real producers always set it, but the
+    fallback keeps a malformed verdict on the right histogram bucket.
+    """
+    metrics.record_judge_verdict(
+        verdict.get("tier", default_tier),
+        verdict.get("risk_level", "medium"),
+        verdict.get("latency_ms", 0),
+    )
+
+
 class AutoApproveReason:
     """Source vocabulary for ``auto_approve_reason`` annotations.
 
@@ -327,6 +353,235 @@ class SessionUIBase:
         self._enqueue({"type": "plan_resolved", "feedback": feedback})
         self._plan_event.set()
 
+    def approve_tools(self, items: list[dict[str, Any]]) -> tuple[bool, str | None]:
+        """Two-phase approval gate for a batch of tool calls.
+
+        Shared body for both interactive (:class:`turnstone.server.WebUI`)
+        and coordinator (:class:`turnstone.console.coordinator_ui.ConsoleCoordinatorUI`)
+        sessions. Order of resolution:
+
+        1. Reset the per-round verdict cache so late LLM verdicts from
+           the previous round can't leak onto this one.
+        2. Evaluate admin-defined tool policies (deny short-circuits;
+           allow tags items as auto-approved with ``AutoApproveReason.POLICY``).
+        3. Per-tool auto-approve via ``self.auto_approve_tools`` (skill
+           ``allowed_tools`` and operator "Approve + Always").
+        4. Budget-override carve-out + blanket ``self.auto_approve``.
+           Synthetic ``__budget_override__`` items always prompt.
+        5. Activity tagging + ``_broadcast_activity`` so the dashboard
+           reflects approval state.
+        6. Heuristic verdict persistence (one row per ``_heuristic_verdict``
+           item) + ``_record_judge_metric`` hook (subclass-overridden to
+           feed the node's or console's Prometheus collector).
+        7. Emit the ``approve_request`` and block on ``_approval_event``
+           up to ``_APPROVAL_WAIT_TIMEOUT``.
+
+        ``__budget_override__`` is interactive-only today (coord
+        workstreams don't have token budgets), but the carve-out check
+        is cheap (``any(...)`` over pending) and is a no-op on coord;
+        kept unconditional so a future coord-skill path picks it up
+        for free.
+        """
+        self._reset_approval_cycle()
+        pending = [it for it in items if it.get("needs_approval") and not it.get("error")]
+
+        # ``__budget_override__`` is a synthetic UI-only pseudo-tool injected
+        # by ChatSession.send when a skill's token budget is exhausted; its
+        # whole purpose is to force an operator prompt before the next turn
+        # spends past the cap. Read from the pre-filter ``items`` list (not
+        # ``pending``) so a wildcard ``*: allow`` policy or a stray entry in
+        # ``auto_approve_tools`` cannot strip the override from ``pending``
+        # before the carve-out gate at the auto-approve fall-through can see
+        # it. Same intent gates the policy block above.
+        has_budget_override = any(it.get("func_name") == "__budget_override__" for it in items)
+
+        # -- Tool policy evaluation -----------------------------------------------
+        # Check admin-defined tool policies before the auto_approve check.
+        # ``__budget_override__`` is excluded from policy matching: it is a
+        # synthetic UI-only pseudo-tool that exists specifically to force an
+        # operator prompt when a skill's token budget is exhausted, so a
+        # wildcard ``*: allow`` policy must never auto-approve it. Same
+        # rationale gates the carve-out check below at line 470.
+        if pending:
+            try:
+                from turnstone.core.policy import evaluate_tool_policies_batch
+                from turnstone.core.storage._registry import get_storage
+
+                storage = get_storage()
+                if storage is not None:
+                    tool_names = [
+                        it.get("approval_label", "") or it.get("func_name", "")
+                        for it in pending
+                        if it.get("func_name") and it.get("func_name") != "__budget_override__"
+                    ]
+                    if tool_names:
+                        verdicts = evaluate_tool_policies_batch(storage, tool_names)
+                        still_pending = []
+                        for it in pending:
+                            policy_name = it.get("approval_label", "") or it.get("func_name", "")
+                            # Synthetic budget-override item bypasses policy
+                            # matching entirely — falls through to the carve-out
+                            # gate so an operator always sees the prompt.
+                            if it.get("func_name") == "__budget_override__":
+                                still_pending.append(it)
+                                continue
+                            verdict = verdicts.get(policy_name)
+                            if verdict == "deny":
+                                it["denied"] = True
+                                it["denial_msg"] = (
+                                    f"Blocked by tool policy (pattern match for '{policy_name}')"
+                                )
+                            elif verdict == "allow":
+                                # Admin-defined ``allow`` rule fires the
+                                # auto-approve gate without any UI prompt.
+                                # Tag for /dashboard visibility so the
+                                # operator can see which calls bypassed
+                                # the prompt and why.
+                                it["needs_approval"] = False
+                                self._tag_auto_approved([it], AutoApproveReason.POLICY)
+                            else:
+                                still_pending.append(it)
+                        # If all were resolved by policy, check if any were denied
+                        if not still_pending:
+                            any_denied = any(it.get("denied") for it in items)
+                            if any_denied:
+                                # Record the policy-allowed siblings before
+                                # the early return — the fall-through
+                                # branch never runs on this path, so without
+                                # this the policy bypass is invisible to
+                                # /dashboard + audit.
+                                self._record_auto_approves(items)
+                                self._enqueue(
+                                    {
+                                        "type": "tool_info",
+                                        "items": self._serialize_approval_items(items),
+                                    }
+                                )
+                                return False, "Blocked by tool policy"
+                        pending = still_pending
+            except Exception:
+                log.debug("Tool policy evaluation failed", exc_info=True)
+        # -- End tool policy evaluation -------------------------------------------
+
+        # Per-tool auto-approve check (from workstream template or interactive "Always").
+        # Suppressed when a budget-override item is present so the carve-out
+        # at the next gate stays effective even if ``__budget_override__`` ever
+        # lands in ``auto_approve_tools`` (defensive — listings filter it out
+        # today, but the worker can be configured by a skill template).
+        if pending and self.auto_approve_tools and not has_budget_override:
+            pending_names = {
+                it.get("approval_label", "") or it.get("func_name", "")
+                for it in pending
+                if it.get("func_name")
+            }
+            if pending_names and pending_names.issubset(self.auto_approve_tools):
+                # Tag each formerly-pending item with the per-tool source
+                # recorded when ``auto_approve_tools`` was populated:
+                # ``skill`` (skill template's ``allowed_tools``) /
+                # ``always`` (user "Approve + Always" click) / fallback
+                # ``auto_approve_tools`` for legacy or unknown writers.
+                # Visibility for the skill-vs-explicit conflation
+                # flagged on the coord tree dashboard.
+                self._tag_auto_approved(
+                    pending,
+                    AutoApproveReason.AUTO_APPROVE_TOOLS,
+                    source_map=self._auto_approve_tools_source,
+                )
+                pending = []
+
+        # Budget override requires explicit approval — never auto-approved by
+        # blanket auto_approve (tool policies can still allow it explicitly,
+        # but the policy block above carves out ``__budget_override__`` so
+        # that path is unreachable too). ``has_budget_override`` was computed
+        # from the pre-filter ``items`` list at the top of the function so a
+        # policy/auto-approve pass that drained the override from ``pending``
+        # cannot disarm this gate.
+        blanket_active = self.auto_approve and not has_budget_override
+        if not pending or blanket_active:
+            if blanket_active and pending:
+                # Blanket flag drained the rest of pending — tag so the
+                # dashboard can distinguish from
+                # ``auto_approve_tools`` / ``policy``.  No need to
+                # clear ``pending`` here: the function returns inside
+                # this block without reading it again.
+                self._tag_auto_approved(pending, AutoApproveReason.BLANKET)
+            # Track auto-approved tool activity
+            first = items[0] if items else {}
+            label = first.get("func_name", "")
+            preview = first.get("preview", "")[:80]
+            with self._ws_lock:
+                self._ws_current_activity = f"⚙ {label}: {preview}" if label else ""
+                self._ws_activity_state = "tool" if label else ""
+            self._broadcast_activity()
+            self._record_auto_approves(items)
+            self._enqueue({"type": "tool_info", "items": self._serialize_approval_items(items)})
+            return True, None
+
+        # Track pending approval activity
+        first_pending = pending[0]
+        label = first_pending.get("func_name", "")
+        preview = first_pending.get("preview", "")[:60]
+        with self._ws_lock:
+            self._ws_current_activity = f"⏳ Awaiting approval: {label} — {preview}"
+            self._ws_activity_state = "approval"
+        self._broadcast_activity()
+
+        # Persist heuristic verdicts and track for user_decision update.
+        # Build list locally, then assign under lock to avoid racing with
+        # the judge daemon thread's on_intent_verdict() appends. Storage
+        # write goes through the bulk path so a tool-heavy turn pays one
+        # commit instead of N (was visible as time-to-render-prompt
+        # latency for fan-out turns); the per-item Prometheus call stays
+        # in the loop because it's a lock+increment, not a DB round-trip.
+        heuristic_verdicts: list[dict[str, Any]] = []
+        for item in items:
+            hv = item.get("_heuristic_verdict")
+            if hv:
+                heuristic_verdicts.append(hv)
+                # Subclass-overridden Prometheus surface: WebUI feeds
+                # the per-node /metrics endpoint, ConsoleCoordinatorUI
+                # feeds the console's /metrics endpoint via ConsoleMetrics.
+                self._record_judge_metric(hv)
+        self._persist_intent_verdicts_bulk(heuristic_verdicts, default_tier="heuristic")
+
+        with self._ws_lock:
+            self._pending_verdicts = heuristic_verdicts
+
+        # Record any items the policy block already auto-approved
+        # before falling through to the prompt — without this the
+        # mixed-policy-then-prompt path leaves the policy bypass
+        # invisible to /dashboard (the auto-approve fall-through never
+        # runs since pending is non-empty + blanket inactive).
+        # 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)
+        self._approval_event.clear()
+        self._pending_approval = {
+            "type": "approve_request",
+            "items": self._serialize_approval_items(items),
+            "judge_pending": judge_pending,
+        }
+        self._enqueue(self._pending_approval)
+        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.
+            log.warning("Approval timed out for ws_id=%s", self.ws_id)
+            self.resolve_approval(False, "Approval timed out after 1 hour")
+        self._pending_approval = None
+        approved, feedback = self._approval_result
+
+        if not approved:
+            denial_msg = "Denied by user"
+            if feedback:
+                denial_msg += f": {feedback}"
+            for item in pending:
+                item["denied"] = True
+                item["denial_msg"] = denial_msg
+
+        return approved, feedback
+
     # ------------------------------------------------------------------
     # Intent-judge + output-guard plumbing
     # ------------------------------------------------------------------
@@ -335,6 +590,13 @@ class SessionUIBase:
     # can't grow unbounded. FIFO eviction on insert.
     _LLM_VERDICT_CACHE_MAX = 50
 
+    # Hard cap on how long a worker thread blocks waiting for an
+    # approval / plan-review decision. Subclasses' ``approve_tools`` and
+    # ``on_plan_review`` reference this rather than the literal so a
+    # future ``settings.approval_timeout_seconds`` knob can swap it in
+    # one place.
+    _APPROVAL_WAIT_TIMEOUT = 3600
+
     def on_intent_verdict(self, verdict: dict[str, Any]) -> None:
         """Deliver an LLM intent-judge verdict to the frontend + persist.
 
@@ -373,7 +635,90 @@ class SessionUIBase:
         if decision:
             self._persist_verdict_decisions([verdict], decision)
 
-    def _persist_intent_verdict(self, verdict: dict[str, Any]) -> None:
+    def _record_judge_metric(self, verdict: dict[str, Any]) -> None:
+        """Extension point for transport-specific Prometheus metrics.
+
+        ``approve_tools`` calls this for each persisted heuristic
+        verdict. Subclasses override to fan the verdict into their
+        own metrics collector:
+
+        - ``WebUI`` writes to the per-node ``MetricsCollector`` so the
+          node's /metrics endpoint surfaces ``turnstone_judge_verdicts_total``.
+        - ``ConsoleCoordinatorUI`` writes to ``ConsoleMetrics`` so the
+          console's /metrics endpoint surfaces the same metric name —
+          a cluster-wide PromQL query rolls coord and interactive
+          verdicts up uniformly.
+
+        Default no-op covers test fixtures and any future SessionUI
+        impl that doesn't expose a /metrics surface. Mirrors the
+        pattern used for ``_broadcast_state`` / ``_broadcast_activity``.
+        """
+        del verdict  # default impl: no metrics surface
+
+    def _persist_intent_verdicts_bulk(
+        self,
+        verdicts: list[dict[str, Any]],
+        *,
+        default_tier: str = "heuristic",
+    ) -> None:
+        """Bulk-insert a list of intent-judge verdicts in one transaction.
+
+        Used by ``approve_tools`` so the per-turn heuristic-verdict
+        persistence doesn't block on N×commit before the approval UI
+        enqueues. Each verdict dict mirrors the keyword args of
+        :meth:`_persist_intent_verdict`; ``ws_id`` is stamped from
+        ``self.ws_id`` and ``evidence`` is JSON-encoded so the row
+        shape matches the per-row path. Storage failure is best-effort
+        (logged at debug) — the verdict cache and UI dispatch run
+        independently of the DB write.
+        """
+        if not verdicts:
+            return
+        try:
+            from turnstone.core.storage._registry import get_storage
+
+            storage = get_storage()
+            if storage is None:
+                return
+            rows = [
+                {
+                    "verdict_id": v.get("verdict_id", ""),
+                    "ws_id": self.ws_id,
+                    "call_id": v.get("call_id", ""),
+                    "func_name": v.get("func_name", ""),
+                    "func_args": v.get("func_args", ""),
+                    "intent_summary": v.get("intent_summary", ""),
+                    "risk_level": v.get("risk_level", "medium"),
+                    "confidence": v.get("confidence", 0.5),
+                    "recommendation": v.get("recommendation", "review"),
+                    "reasoning": v.get("reasoning", ""),
+                    "evidence": json.dumps(v.get("evidence", [])),
+                    "tier": v.get("tier", default_tier),
+                    "judge_model": v.get("judge_model", ""),
+                    "latency_ms": v.get("latency_ms", 0),
+                }
+                for v in verdicts
+            ]
+            storage.create_intent_verdicts_bulk(rows)
+        except Exception:
+            log.debug("Failed to bulk-persist intent verdicts", exc_info=True)
+
+    def _persist_intent_verdict(
+        self,
+        verdict: dict[str, Any],
+        *,
+        default_tier: str = "llm",
+    ) -> None:
+        """Persist an intent-judge verdict row.
+
+        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"``).
+        ``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.
+        """
         try:
             from turnstone.core.storage._registry import get_storage
 
@@ -392,12 +737,12 @@ class SessionUIBase:
                 recommendation=verdict.get("recommendation", "review"),
                 reasoning=verdict.get("reasoning", ""),
                 evidence=json.dumps(verdict.get("evidence", [])),
-                tier=verdict.get("tier", "llm"),
+                tier=verdict.get("tier", default_tier),
                 judge_model=verdict.get("judge_model", ""),
                 latency_ms=verdict.get("latency_ms", 0),
             )
         except Exception:
-            log.debug("Failed to persist LLM verdict", exc_info=True)
+            log.debug("Failed to persist intent verdict", exc_info=True)
 
     def serialize_pending_approval_detail(self) -> dict[str, Any] | None:
         """Build the inline approval payload for dashboard projection.

turnstone/core/storage/_postgresql.py

@@ -2121,6 +2121,18 @@ class PostgreSQLBackend:
                 },
             )
             conn.commit()
+        # Drop both the org-specific slot AND the default ``""`` slot.
+        # ``list_tool_policies("")`` returns rows for every org_id (no
+        # WHERE filter when org_id is falsy), and the default
+        # evaluators (``SessionUIBase.approve_tools`` / ``cli.py``) use
+        # ``org_id=""``, so an org-scoped insert that only invalidated
+        # the org slot would leave the default slot serving stale data
+        # until the TTL window expired.
+        from turnstone.core.policy import invalidate_policy_cache
+
+        invalidate_policy_cache(org_id)
+        if org_id != "":
+            invalidate_policy_cache("")
 
     def get_tool_policy(self, policy_id: str) -> dict[str, Any] | None:
         with self._conn() as conn:
@@ -2154,7 +2166,12 @@ class PostgreSQLBackend:
                 .values(**fields)
             )
             conn.commit()
-            return result.rowcount > 0
+            updated = result.rowcount > 0
+        if updated:
+            from turnstone.core.policy import invalidate_policy_cache
+
+            invalidate_policy_cache()
+        return updated
 
     def delete_tool_policy(self, policy_id: str) -> bool:
         with self._conn() as conn:
@@ -2162,7 +2179,12 @@ class PostgreSQLBackend:
                 sa.delete(tool_policies).where(tool_policies.c.policy_id == policy_id)
             )
             conn.commit()
-            return result.rowcount > 0
+            deleted = result.rowcount > 0
+        if deleted:
+            from turnstone.core.policy import invalidate_policy_cache
+
+            invalidate_policy_cache()
+        return deleted
 
     # -- Prompt templates ------------------------------------------------------
 
@@ -2966,6 +2988,34 @@ class PostgreSQLBackend:
             )
             conn.commit()
 
+    def create_intent_verdicts_bulk(self, verdicts: list[dict[str, Any]]) -> None:
+        if not verdicts:
+            return
+        now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
+        rows = [
+            {
+                "verdict_id": v.get("verdict_id", ""),
+                "ws_id": v.get("ws_id", ""),
+                "call_id": v.get("call_id", ""),
+                "func_name": v.get("func_name", ""),
+                "func_args": v.get("func_args", ""),
+                "intent_summary": v.get("intent_summary", ""),
+                "risk_level": v.get("risk_level", "medium"),
+                "confidence": v.get("confidence", 0.5),
+                "recommendation": v.get("recommendation", "review"),
+                "reasoning": v.get("reasoning", ""),
+                "evidence": v.get("evidence", ""),
+                "tier": v.get("tier", "heuristic"),
+                "judge_model": v.get("judge_model", ""),
+                "latency_ms": v.get("latency_ms", 0),
+                "created": now,
+            }
+            for v in verdicts
+        ]
+        with self._conn() as conn:
+            conn.execute(sa.insert(intent_verdicts), rows)
+            conn.commit()
+
     def get_intent_verdict(self, verdict_id: str) -> dict[str, Any] | None:
         with self._conn() as conn:
             row = conn.execute(

turnstone/core/storage/_protocol.py

@@ -1258,6 +1258,20 @@ class StorageBackend(Protocol):
         """Record an intent validation verdict."""
         ...
 
+    def create_intent_verdicts_bulk(self, verdicts: list[dict[str, Any]]) -> None:
+        """Insert many intent_verdict rows in one transaction.
+
+        Each dict mirrors :meth:`create_intent_verdict`'s keyword args
+        (``verdict_id`` / ``ws_id`` / ``call_id`` / ``func_name`` /
+        ``func_args`` / ``intent_summary`` / ``risk_level`` /
+        ``confidence`` / ``recommendation`` / ``reasoning`` / ``evidence`` /
+        ``tier`` / ``judge_model`` / ``latency_ms``). Used by the
+        synchronous heuristic-verdict persistence loop in
+        ``approve_tools`` so a tool-heavy turn doesn't pay N×commit
+        latency before the approval prompt renders.
+        """
+        ...
+
     def get_intent_verdict(self, verdict_id: str) -> dict[str, Any] | None:
         """Return intent verdict dict or None."""
         ...

turnstone/core/storage/_sqlite.py

@@ -2224,6 +2224,18 @@ class SQLiteBackend:
                 },
             )
             conn.commit()
+        # Drop both the org-specific slot AND the default ``""`` slot.
+        # ``list_tool_policies("")`` returns rows for every org_id (no
+        # WHERE filter when org_id is falsy), and the default
+        # evaluators (``SessionUIBase.approve_tools`` / ``cli.py``) use
+        # ``org_id=""``, so an org-scoped insert that only invalidated
+        # the org slot would leave the default slot serving stale data
+        # until the TTL window expired.
+        from turnstone.core.policy import invalidate_policy_cache
+
+        invalidate_policy_cache(org_id)
+        if org_id != "":
+            invalidate_policy_cache("")
 
     def get_tool_policy(self, policy_id: str) -> dict[str, Any] | None:
         with self._conn() as conn:
@@ -2257,7 +2269,15 @@ class SQLiteBackend:
                 .values(**fields)
             )
             conn.commit()
-            return result.rowcount > 0
+            updated = result.rowcount > 0
+        if updated:
+            # Invalidate every org slot — the update doesn't expose
+            # the row's org_id without a re-read, and policy mutations
+            # are admin-rate so a global drop is fine.
+            from turnstone.core.policy import invalidate_policy_cache
+
+            invalidate_policy_cache()
+        return updated
 
     def delete_tool_policy(self, policy_id: str) -> bool:
         with self._conn() as conn:
@@ -2265,7 +2285,12 @@ class SQLiteBackend:
                 sa.delete(tool_policies).where(tool_policies.c.policy_id == policy_id)
             )
             conn.commit()
-            return result.rowcount > 0
+            deleted = result.rowcount > 0
+        if deleted:
+            from turnstone.core.policy import invalidate_policy_cache
+
+            invalidate_policy_cache()
+        return deleted
 
     # -- Prompt templates ------------------------------------------------------
 
@@ -3072,6 +3097,34 @@ class SQLiteBackend:
             )
             conn.commit()
 
+    def create_intent_verdicts_bulk(self, verdicts: list[dict[str, Any]]) -> None:
+        if not verdicts:
+            return
+        now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
+        rows = [
+            {
+                "verdict_id": v.get("verdict_id", ""),
+                "ws_id": v.get("ws_id", ""),
+                "call_id": v.get("call_id", ""),
+                "func_name": v.get("func_name", ""),
+                "func_args": v.get("func_args", ""),
+                "intent_summary": v.get("intent_summary", ""),
+                "risk_level": v.get("risk_level", "medium"),
+                "confidence": v.get("confidence", 0.5),
+                "recommendation": v.get("recommendation", "review"),
+                "reasoning": v.get("reasoning", ""),
+                "evidence": v.get("evidence", ""),
+                "tier": v.get("tier", "heuristic"),
+                "judge_model": v.get("judge_model", ""),
+                "latency_ms": v.get("latency_ms", 0),
+                "created": now,
+            }
+            for v in verdicts
+        ]
+        with self._conn() as conn:
+            conn.execute(sa.insert(intent_verdicts), rows)
+            conn.commit()
+
     def get_intent_verdict(self, verdict_id: str) -> dict[str, Any] | None:
         with self._conn() as conn:
             row = conn.execute(

turnstone/server.py

@@ -77,7 +77,11 @@ from turnstone.core.session_routes import (
     make_send_handler,
     register_session_routes,
 )
-from turnstone.core.session_ui_base import AutoApproveReason, SessionUIBase
+from turnstone.core.session_ui_base import (
+    AutoApproveReason,
+    SessionUIBase,
+    fire_judge_verdict_metric,
+)
 from turnstone.core.tools import TOOLS  # noqa: F401 — available for introspection
 from turnstone.core.web_helpers import version_html as _version_html
 from turnstone.core.workstream import (
@@ -222,196 +226,23 @@ class WebUI(SessionUIBase):
     # overridden below to layer Prometheus ``_metrics.record_*`` calls
     # (node-only) on top of the shared per-ws metric writes.
 
-    def approve_tools(self, items: list[dict[str, Any]]) -> tuple[bool, str | None]:
-        self._reset_approval_cycle()
-        pending = [it for it in items if it.get("needs_approval") and not it.get("error")]
+    # ``approve_tools`` is inherited from :class:`SessionUIBase`. The
+    # node-level prometheus metric for heuristic verdicts is layered via
+    # the ``_record_judge_metric`` hook below so the lifted body stays
+    # transport-agnostic.
 
-        # -- Tool policy evaluation -----------------------------------------------
-        # Check admin-defined tool policies before the auto_approve check.
-        if pending:
-            try:
-                from turnstone.core.policy import evaluate_tool_policies_batch
-                from turnstone.core.storage._registry import get_storage
+    def _record_judge_metric(self, verdict: dict[str, Any]) -> None:
+        """Layer the per-node Prometheus metric on top of the shared body.
 
-                storage = get_storage()
-                if storage is not None:
-                    tool_names = [
-                        it.get("approval_label", "") or it.get("func_name", "")
-                        for it in pending
-                        if it.get("func_name")
-                    ]
-                    if tool_names:
-                        verdicts = evaluate_tool_policies_batch(storage, tool_names)
-                        still_pending = []
-                        for it in pending:
-                            policy_name = it.get("approval_label", "") or it.get("func_name", "")
-                            verdict = verdicts.get(policy_name)
-                            if verdict == "deny":
-                                it["denied"] = True
-                                it["denial_msg"] = (
-                                    f"Blocked by tool policy (pattern match for '{policy_name}')"
-                                )
-                            elif verdict == "allow":
-                                # Admin-defined ``allow`` rule fires the
-                                # auto-approve gate without any UI prompt.
-                                # Tag for /dashboard visibility so the
-                                # operator can see which calls bypassed
-                                # the prompt and why.
-                                it["needs_approval"] = False
-                                self._tag_auto_approved([it], AutoApproveReason.POLICY)
-                            else:
-                                still_pending.append(it)
-                        # If all were resolved by policy, check if any were denied
-                        if not still_pending:
-                            any_denied = any(it.get("denied") for it in items)
-                            if any_denied:
-                                # Record the policy-allowed siblings before
-                                # the early return — the line-325 fall-
-                                # through branch never runs on this path,
-                                # so without this the policy bypass is
-                                # invisible to /dashboard + audit.
-                                self._record_auto_approves(items)
-                                self._enqueue(
-                                    {
-                                        "type": "tool_info",
-                                        "items": self._serialize_approval_items(items),
-                                    }
-                                )
-                                return False, "Blocked by tool policy"
-                        pending = still_pending
-            except Exception:
-                log.debug("Tool policy evaluation failed", exc_info=True)
-        # -- End tool policy evaluation -------------------------------------------
-
-        # Per-tool auto-approve check (from workstream template or interactive "Always")
-        if pending and self.auto_approve_tools:
-            pending_names = {
-                it.get("approval_label", "") or it.get("func_name", "")
-                for it in pending
-                if it.get("func_name")
-            }
-            if pending_names and pending_names.issubset(self.auto_approve_tools):
-                # Tag each formerly-pending item with the per-tool source
-                # recorded when ``auto_approve_tools`` was populated:
-                # ``skill`` (skill template's ``allowed_tools``) /
-                # ``always`` (user "Approve + Always" click) / fallback
-                # ``auto_approve_tools`` for legacy or unknown writers.
-                # Visibility for the skill-vs-explicit conflation
-                # flagged on the coord tree dashboard.
-                self._tag_auto_approved(
-                    pending,
-                    AutoApproveReason.AUTO_APPROVE_TOOLS,
-                    source_map=self._auto_approve_tools_source,
-                )
-                pending = []
-
-        # Budget override requires explicit approval — never auto-approved by
-        # blanket auto_approve (tool policies can still allow it explicitly).
-        has_budget_override = any(it.get("func_name") == "__budget_override__" for it in pending)
-        blanket_active = self.auto_approve and not has_budget_override
-        if not pending or blanket_active:
-            if blanket_active and pending:
-                # Blanket flag drained the rest of pending \u2014 tag so the
-                # dashboard can distinguish from
-                # ``auto_approve_tools`` / ``policy``.  No need to
-                # clear ``pending`` here: the function returns inside
-                # this block without reading it again.
-                self._tag_auto_approved(pending, AutoApproveReason.BLANKET)
-            # Track auto-approved tool activity
-            first = items[0] if items else {}
-            label = first.get("func_name", "")
-            preview = first.get("preview", "")[:80]
-            with self._ws_lock:
-                self._ws_current_activity = f"\u2699 {label}: {preview}" if label else ""
-                self._ws_activity_state = "tool" if label else ""
-            self._broadcast_activity()
-            self._record_auto_approves(items)
-            self._enqueue({"type": "tool_info", "items": self._serialize_approval_items(items)})
-            return True, None
-
-        # Track pending approval activity
-        first_pending = pending[0]
-        label = first_pending.get("func_name", "")
-        preview = first_pending.get("preview", "")[:60]
-        with self._ws_lock:
-            self._ws_current_activity = f"\u23f3 Awaiting approval: {label} \u2014 {preview}"
-            self._ws_activity_state = "approval"
-        self._broadcast_activity()
-
-        # Persist heuristic verdicts and track for user_decision update.
-        # Build list locally, then assign under lock to avoid racing with
-        # the judge daemon thread's on_intent_verdict() appends.
-        heuristic_verdicts: list[dict[str, Any]] = []
-        for item in items:
-            hv = item.get("_heuristic_verdict")
-            if hv:
-                heuristic_verdicts.append(hv)
-                try:
-                    from turnstone.core.storage._registry import get_storage
-
-                    storage = get_storage()
-                    if storage is not None:
-                        storage.create_intent_verdict(
-                            verdict_id=hv.get("verdict_id", ""),
-                            ws_id=self.ws_id,
-                            call_id=hv.get("call_id", ""),
-                            func_name=hv.get("func_name", ""),
-                            func_args=hv.get("func_args", ""),
-                            intent_summary=hv.get("intent_summary", ""),
-                            risk_level=hv.get("risk_level", "medium"),
-                            confidence=hv.get("confidence", 0.5),
-                            recommendation=hv.get("recommendation", "review"),
-                            reasoning=hv.get("reasoning", ""),
-                            evidence=json.dumps(hv.get("evidence", [])),
-                            tier=hv.get("tier", "heuristic"),
-                            judge_model=hv.get("judge_model", ""),
-                            latency_ms=hv.get("latency_ms", 0),
-                        )
-                except Exception:
-                    log.debug("Failed to persist heuristic verdict", exc_info=True)
-                _metrics.record_judge_verdict(
-                    hv.get("tier", "heuristic"),
-                    hv.get("risk_level", "medium"),
-                    hv.get("latency_ms", 0),
-                )
-
-        with self._ws_lock:
-            self._pending_verdicts = heuristic_verdicts
-
-        # Record any items the policy block already auto-approved
-        # before falling through to the prompt — without this the
-        # mixed-policy-then-prompt path leaves the policy bypass
-        # invisible to /dashboard (the line-325 fall-through never
-        # runs since pending is non-empty + blanket inactive).
-        # No-op when no items are auto-approve-tagged.
-        self._record_auto_approves(items)
-
-        # Send approval request and block
-        judge_pending = bool(any(it.get("_heuristic_verdict") for it in items))
-        self._approval_event.clear()
-        self._pending_approval = {
-            "type": "approve_request",
-            "items": self._serialize_approval_items(items),
-            "judge_pending": judge_pending,
-        }
-        self._enqueue(self._pending_approval)
-        if not self._approval_event.wait(timeout=3600):
-            # Approval timed out (e.g., user disconnected). Deny via
-            # resolve_approval so verdicts and state are updated consistently.
-            log.warning("Approval timed out for ws_id=%s", self.ws_id)
-            self.resolve_approval(False, "Approval timed out after 1 hour")
-        self._pending_approval = None
-        approved, feedback = self._approval_result
-
-        if not approved:
-            denial_msg = "Denied by user"
-            if feedback:
-                denial_msg += f": {feedback}"
-            for item in pending:
-                item["denied"] = True
-                item["denial_msg"] = denial_msg
-
-        return approved, feedback
+        ``SessionUIBase.approve_tools`` calls this for each persisted
+        heuristic verdict; ``ConsoleCoordinatorUI`` overrides the same
+        hook to feed the console's ``ConsoleMetrics`` — same metric
+        name, so a cluster-wide PromQL query rolls coord and
+        interactive verdicts up uniformly. The LLM-tier counterpart
+        lives in ``on_intent_verdict`` below — same metric, different
+        tier label, same ``record_judge_verdict`` call.
+        """
+        fire_judge_verdict_metric(_metrics, verdict, "heuristic")
 
     def on_tool_result(
         self,
@@ -448,7 +279,7 @@ class WebUI(SessionUIBase):
         self._plan_event.clear()
         self._pending_plan_review = {"type": "plan_review", "content": content}
         self._enqueue(self._pending_plan_review)
-        if not self._plan_event.wait(timeout=3600):
+        if not self._plan_event.wait(timeout=self._APPROVAL_WAIT_TIMEOUT):
             log.warning("Plan review timed out for ws_id=%s", self.ws_id)
             self._plan_result = ""
         self._pending_plan_review = None
@@ -486,11 +317,7 @@ class WebUI(SessionUIBase):
         node-level prometheus metric update.
         """
         super().on_intent_verdict(verdict)
-        _metrics.record_judge_verdict(
-            verdict.get("tier", "llm"),
-            verdict.get("risk_level", "medium"),
-            verdict.get("latency_ms", 0),
-        )
+        fire_judge_verdict_metric(_metrics, verdict, "llm")
 
     # ``on_output_warning`` inherited from :class:`SessionUIBase`.