diff --git a/tests/conftest.py b/tests/conftest.py index 2fdf5d67..987fe246 100644 --- a/tests/conftest.py +++ b/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() diff --git a/tests/test_console_metrics.py b/tests/test_console_metrics.py index 5cbc964f..4201e6e9 100644 --- a/tests/test_console_metrics.py +++ b/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.""" diff --git a/tests/test_coord_ui_approve_tools.py b/tests/test_coord_ui_approve_tools.py new file mode 100644 index 00000000..57ce9c34 --- /dev/null +++ b/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" diff --git a/tests/test_coordinator_tools.py b/tests/test_coordinator_tools.py index 9c4f9f3f..15b85026 100644 --- a/tests/test_coordinator_tools.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/tests/test_judge_storage.py b/tests/test_judge_storage.py index fd330622..38a191f4 100644 --- a/tests/test_judge_storage.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/turnstone/console/coordinator_ui.py b/turnstone/console/coordinator_ui.py index d85c63db..44f1496d 100644 --- a/turnstone/console/coordinator_ui.py +++ b/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") diff --git a/turnstone/console/metrics.py b/turnstone/console/metrics.py index 560ae0bb..c1d17737 100644 --- a/turnstone/console/metrics.py +++ b/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) diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 911f7d13..271eaaa7 100644 --- a/turnstone/console/server.py +++ b/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, diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js index f61e4755..6fd2da3a 100644 --- a/turnstone/console/static/coordinator/coordinator.js +++ b/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 diff --git a/turnstone/core/policy.py b/turnstone/core/policy.py index 17c5cb13..3083ee13 100644 --- a/turnstone/core/policy.py +++ b/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] = {} diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 5933ddb7..3c8d24a9 100644 --- a/turnstone/core/session.py +++ b/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"] diff --git a/turnstone/core/session_ui_base.py b/turnstone/core/session_ui_base.py index b1c62310..930246f4 100644 --- a/turnstone/core/session_ui_base.py +++ b/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. diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py index 7ef2861c..08ed4983 100644 --- a/turnstone/core/storage/_postgresql.py +++ b/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( diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py index 576d564f..6f83ae51 100644 --- a/turnstone/core/storage/_protocol.py +++ b/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.""" ... diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py index 97854db1..33715fe8 100644 --- a/turnstone/core/storage/_sqlite.py +++ b/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( diff --git a/turnstone/server.py b/turnstone/server.py index 28932f8e..0bd20203 100644 --- a/turnstone/server.py +++ b/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`.