diff --git a/tests/_coord_test_helpers.py b/tests/_coord_test_helpers.py index 3d5b3725..f3d3fe1c 100644 --- a/tests/_coord_test_helpers.py +++ b/tests/_coord_test_helpers.py @@ -35,11 +35,10 @@ def _seed_children( The production path populates the registry via the cluster-event fan-out thread observing ``ws_created`` events. These tests just need a known-children set for the endpoint handlers to iterate — - inject directly under ``_children_lock`` rather than spinning up - the collector + fan-out plumbing. + inject directly via the registry's bulk-merge surface rather than + spinning up the collector + fan-out plumbing. """ - with adapter._children_lock: - adapter._merge_child_ids_locked(coord_ws_id, child_ws_ids) + adapter._registry.merge_children(coord_ws_id, child_ws_ids) class _AuthMiddleware(BaseHTTPMiddleware): diff --git a/tests/test_child_source.py b/tests/test_child_source.py new file mode 100644 index 00000000..603296ba --- /dev/null +++ b/tests/test_child_source.py @@ -0,0 +1,270 @@ +"""Unit tests for :mod:`turnstone.core.child_source`. + +Covers both strategies in isolation against fakes — no live collector, +no live SessionManager. Adapter-level integration coverage continues to +live in ``test_coordinator_adapter.py``. +""" + +from __future__ import annotations + +import contextlib +import time +from typing import TYPE_CHECKING, Any + +from turnstone.core.child_source import ClusterChildSource, SameNodeChildSource +from turnstone.core.children_registry import ChildrenRegistry +from turnstone.core.workstream import WorkstreamState + +if TYPE_CHECKING: + import queue + + +# --------------------------------------------------------------------------- +# SameNodeChildSource +# --------------------------------------------------------------------------- + + +class _FakeManager: + """Minimal SessionManager stand-in implementing the subscribe API.""" + + def __init__(self) -> None: + self.subscribers: list[Any] = [] + + def subscribe_to_state(self, callback: Any) -> None: + self.subscribers.append(callback) + + def unsubscribe_from_state(self, callback: Any) -> None: + with contextlib.suppress(ValueError): + self.subscribers.remove(callback) + + def fire(self, ws_id: str, state: WorkstreamState) -> None: + for cb in self.subscribers: + cb(ws_id, state) + + +class TestSameNodeChildSource: + def test_start_subscribes_to_manager(self) -> None: + mgr = _FakeManager() + registry = ChildrenRegistry() + src = SameNodeChildSource(mgr, registry) + sink_calls: list[dict[str, Any]] = [] + src.start(sink=sink_calls.append) + assert len(mgr.subscribers) == 1 + + def test_state_change_for_known_child_pushes_to_sink(self) -> None: + mgr = _FakeManager() + registry = ChildrenRegistry() + registry.install("p1", object()) + registry.add_child("p1", "c1") + src = SameNodeChildSource(mgr, registry) + sink_calls: list[dict[str, Any]] = [] + src.start(sink=sink_calls.append) + + mgr.fire("c1", WorkstreamState.RUNNING) + + assert len(sink_calls) == 1 + ev = sink_calls[0] + assert ev["type"] == "cluster_state" + assert ev["ws_id"] == "c1" + assert ev["state"] == "running" + + def test_state_change_for_unknown_workstream_is_dropped(self) -> None: + mgr = _FakeManager() + registry = ChildrenRegistry() + src = SameNodeChildSource(mgr, registry) + sink_calls: list[dict[str, Any]] = [] + src.start(sink=sink_calls.append) + + # No registry entry — pre-filter drops the event without + # invoking the sink. + mgr.fire("ws-unknown", WorkstreamState.IDLE) + assert sink_calls == [] + + def test_shutdown_unsubscribes(self) -> None: + mgr = _FakeManager() + registry = ChildrenRegistry() + src = SameNodeChildSource(mgr, registry) + src.start(sink=lambda ev: None) + assert len(mgr.subscribers) == 1 + src.shutdown() + assert mgr.subscribers == [] + + def test_start_is_idempotent(self) -> None: + mgr = _FakeManager() + registry = ChildrenRegistry() + src = SameNodeChildSource(mgr, registry) + src.start(sink=lambda ev: None) + src.start(sink=lambda ev: None) + # Second start is a no-op; only one subscription. + assert len(mgr.subscribers) == 1 + + def test_sink_exception_does_not_propagate(self) -> None: + mgr = _FakeManager() + registry = ChildrenRegistry() + registry.install("p1", object()) + registry.add_child("p1", "c1") + src = SameNodeChildSource(mgr, registry) + + def bad_sink(ev: dict[str, Any]) -> None: + raise RuntimeError("sink boom") + + src.start(sink=bad_sink) + # Should not raise — the strategy catches sink failures and logs. + mgr.fire("c1", WorkstreamState.RUNNING) + + +# --------------------------------------------------------------------------- +# ClusterChildSource +# --------------------------------------------------------------------------- + + +class _FakeCollector: + """Minimal ClusterCollector stand-in providing the listener API.""" + + def __init__(self, snapshot: dict[str, Any] | None = None) -> None: + self._snapshot = snapshot or {"nodes": []} + self.queues: list[queue.Queue[dict[str, Any]]] = [] + self.unregistered: list[queue.Queue[dict[str, Any]]] = [] + + def get_snapshot_and_register(self, q: queue.Queue[dict[str, Any]]) -> dict[str, Any]: + self.queues.append(q) + return self._snapshot + + def unregister_listener(self, q: queue.Queue[dict[str, Any]]) -> None: + self.unregistered.append(q) + + def emit(self, event: dict[str, Any]) -> None: + """Push an event to all registered listener queues.""" + for q in self.queues: + q.put(event) + + +class TestClusterChildSource: + def test_start_subscribes_to_collector(self) -> None: + coll = _FakeCollector() + registry = ChildrenRegistry() + src = ClusterChildSource( + collector=coll, + registry=registry, + parents_provider=list, + ) + try: + src.start(sink=lambda ev: None) + assert len(coll.queues) == 1 + finally: + src.shutdown() + + def test_start_primes_registry_from_snapshot(self) -> None: + snapshot = { + "nodes": [ + { + "workstreams": [ + {"id": "c1", "parent_ws_id": "p1"}, + {"id": "c2", "parent_ws_id": "p1"}, + # Unknown parent — dropped + {"id": "x", "parent_ws_id": "p-unknown"}, + ], + }, + ], + } + coll = _FakeCollector(snapshot) + registry = ChildrenRegistry() + registry.install("p1", object()) + src = ClusterChildSource( + collector=coll, + registry=registry, + parents_provider=lambda: ["p1"], + ) + try: + src.start(sink=lambda ev: None) + assert set(registry.children_of("p1")) == {"c1", "c2"} + assert registry.parent_for("x") is None + finally: + src.shutdown() + + def test_event_dispatched_to_sink(self) -> None: + coll = _FakeCollector() + registry = ChildrenRegistry() + src = ClusterChildSource( + collector=coll, + registry=registry, + parents_provider=list, + ) + sink_calls: list[dict[str, Any]] = [] + + try: + src.start(sink=sink_calls.append) + coll.emit({"type": "cluster_state", "ws_id": "c1", "state": "running"}) + # Daemon thread loop has 1.0s queue timeout; poll briefly. + for _ in range(20): + if sink_calls: + break + time.sleep(0.05) + assert len(sink_calls) == 1 + assert sink_calls[0]["ws_id"] == "c1" + finally: + src.shutdown() + + def test_shutdown_unregisters_and_joins_thread(self) -> None: + coll = _FakeCollector() + registry = ChildrenRegistry() + src = ClusterChildSource( + collector=coll, + registry=registry, + parents_provider=list, + ) + src.start(sink=lambda ev: None) + src.shutdown() + assert coll.unregistered == coll.queues + # Second shutdown is a no-op (idempotent). + src.shutdown() + + def test_start_is_idempotent(self) -> None: + coll = _FakeCollector() + registry = ChildrenRegistry() + src = ClusterChildSource( + collector=coll, + registry=registry, + parents_provider=list, + ) + try: + src.start(sink=lambda ev: None) + src.start(sink=lambda ev: None) + assert len(coll.queues) == 1 + finally: + src.shutdown() + + def test_sink_exception_does_not_kill_thread(self) -> None: + coll = _FakeCollector() + registry = ChildrenRegistry() + src = ClusterChildSource( + collector=coll, + registry=registry, + parents_provider=list, + ) + survived_calls: list[dict[str, Any]] = [] + call_count = [0] + + def flaky_sink(ev: dict[str, Any]) -> None: + call_count[0] += 1 + if call_count[0] == 1: + raise RuntimeError("first one boom") + survived_calls.append(ev) + + try: + src.start(sink=flaky_sink) + coll.emit({"type": "cluster_state", "ws_id": "c1", "state": "x"}) + coll.emit({"type": "cluster_state", "ws_id": "c2", "state": "y"}) + for _ in range(40): + if survived_calls: + break + time.sleep(0.05) + assert len(survived_calls) == 1 + assert survived_calls[0]["ws_id"] == "c2" + finally: + src.shutdown() + + +# Multi-subscriber observer tests for ``SessionManager.subscribe_to_state`` +# / ``unsubscribe_from_state`` live in ``test_session_manager.py`` where +# the proper FakeAdapter / FakeStorage construction helpers already exist. diff --git a/tests/test_children_registry.py b/tests/test_children_registry.py new file mode 100644 index 00000000..72a54c91 --- /dev/null +++ b/tests/test_children_registry.py @@ -0,0 +1,239 @@ +"""Unit tests for :class:`turnstone.core.children_registry.ChildrenRegistry`. + +The registry was lifted from ``CoordinatorAdapter`` in Stage 3 Step 1. +Adapter-level coverage for the integrated behavior already lives in +``test_coordinator_adapter.py``; this file pins the data structure +invariants in isolation so the registry can be reused by future +``ChildSource`` strategies (Step 2) without re-deriving the behavior +from the adapter test surface. +""" + +from __future__ import annotations + +import threading + +import pytest + +from turnstone.core.children_registry import ChildrenRegistry + + +class _Sentinel: + """Lightweight UI stand-in; identity-comparable, no behavior.""" + + +@pytest.fixture +def registry() -> ChildrenRegistry: + return ChildrenRegistry() + + +# --------------------------------------------------------------------------- +# install / uninstall +# --------------------------------------------------------------------------- + + +class TestInstallUninstall: + def test_install_seeds_empty_child_set_and_presence(self, registry: ChildrenRegistry) -> None: + ui = _Sentinel() + registry.install("p1", ui) + assert registry.children_of("p1") == [] + assert registry.ui_for("p1") is ui + assert registry.parents() == ["p1"] + + def test_install_is_idempotent_repoints_ui_keeps_children( + self, registry: ChildrenRegistry + ) -> None: + ui_a = _Sentinel() + ui_b = _Sentinel() + registry.install("p1", ui_a) + registry.merge_children("p1", ["c1", "c2"]) + registry.install("p1", ui_b) + assert registry.ui_for("p1") is ui_b + assert set(registry.children_of("p1")) == {"c1", "c2"} + + def test_uninstall_clears_forward_reverse_and_presence( + self, registry: ChildrenRegistry + ) -> None: + ui = _Sentinel() + registry.install("p1", ui) + registry.merge_children("p1", ["c1", "c2"]) + registry.uninstall("p1") + assert registry.children_of("p1") == [] + assert registry.ui_for("p1") is None + assert registry.parents() == [] + assert registry.parent_for("c1") is None + assert registry.parent_for("c2") is None + + def test_uninstall_unknown_parent_is_noop(self, registry: ChildrenRegistry) -> None: + registry.uninstall("never-installed") # must not raise + + def test_uninstall_does_not_clobber_other_parents(self, registry: ChildrenRegistry) -> None: + registry.install("p1", _Sentinel()) + registry.install("p2", _Sentinel()) + registry.merge_children("p1", ["c1"]) + registry.merge_children("p2", ["c2"]) + registry.uninstall("p1") + assert registry.parent_for("c1") is None + assert registry.parent_for("c2") == "p2" + assert registry.parents() == ["p2"] + + +# --------------------------------------------------------------------------- +# add_child — atomic check-and-route +# --------------------------------------------------------------------------- + + +class TestAddChild: + def test_add_child_returns_ui_on_success(self, registry: ChildrenRegistry) -> None: + ui = _Sentinel() + registry.install("p1", ui) + assert registry.add_child("p1", "c1") is ui + assert registry.parent_for("c1") == "p1" + assert registry.children_of("p1") == ["c1"] + + def test_add_child_returns_none_when_parent_not_installed( + self, registry: ChildrenRegistry + ) -> None: + assert registry.add_child("absent", "c1") is None + assert registry.parent_for("c1") is None + + def test_add_child_returns_none_on_duplicate(self, registry: ChildrenRegistry) -> None: + ui = _Sentinel() + registry.install("p1", ui) + assert registry.add_child("p1", "c1") is ui + # second add for same child returns None — caller must not + # double-dispatch. + assert registry.add_child("p1", "c1") is None + assert registry.children_of("p1") == ["c1"] + + +# --------------------------------------------------------------------------- +# merge_children — bulk seeding +# --------------------------------------------------------------------------- + + +class TestMergeChildren: + def test_merge_seeds_forward_and_reverse(self, registry: ChildrenRegistry) -> None: + registry.merge_children("p1", ["c1", "c2", "c3"]) + assert set(registry.children_of("p1")) == {"c1", "c2", "c3"} + for cid in ("c1", "c2", "c3"): + assert registry.parent_for(cid) == "p1" + + def test_merge_is_idempotent(self, registry: ChildrenRegistry) -> None: + registry.merge_children("p1", ["c1"]) + registry.merge_children("p1", ["c1"]) + assert registry.children_of("p1") == ["c1"] + + def test_merge_skips_empty_or_falsy_ids(self, registry: ChildrenRegistry) -> None: + registry.merge_children("p1", ["", "c1", "", "c2"]) + assert set(registry.children_of("p1")) == {"c1", "c2"} + + def test_merge_does_not_require_install(self, registry: ChildrenRegistry) -> None: + # Snapshot-priming may run before the parent's install fires — + # the merge still seeds the forward set so the install picks + # the children up. (Storage-seeded rebuild relies on this.) + registry.merge_children("p1", ["c1"]) + assert registry.children_of("p1") == ["c1"] + # ui_for is still None because install hasn't run + assert registry.ui_for("p1") is None + + +# --------------------------------------------------------------------------- +# Lookups — return copies, not live refs +# --------------------------------------------------------------------------- + + +class TestLookups: + def test_children_of_returns_copy(self, registry: ChildrenRegistry) -> None: + registry.install("p1", _Sentinel()) + registry.merge_children("p1", ["c1", "c2"]) + snap = registry.children_of("p1") + snap.append("c3-injected") + assert "c3-injected" not in registry.children_of("p1") + + def test_children_of_unknown_parent_returns_empty(self, registry: ChildrenRegistry) -> None: + assert registry.children_of("absent") == [] + + def test_parent_for_unknown_child_returns_none(self, registry: ChildrenRegistry) -> None: + assert registry.parent_for("absent") is None + + def test_parents_returns_copy(self, registry: ChildrenRegistry) -> None: + registry.install("p1", _Sentinel()) + snap = registry.parents() + snap.append("p2-injected") + assert "p2-injected" not in registry.parents() + + +# --------------------------------------------------------------------------- +# Concurrency — concurrent add_child must not exceed the unique-set +# invariant or leave a half-installed reverse-index entry. +# --------------------------------------------------------------------------- + + +class TestConcurrency: + def test_concurrent_add_child_returns_ui_exactly_once_per_unique( + self, registry: ChildrenRegistry + ) -> None: + ui = _Sentinel() + registry.install("p1", ui) + results: list[object] = [] + results_lock = threading.Lock() + + def attempt_add(child_id: str) -> None: + r = registry.add_child("p1", child_id) + with results_lock: + results.append(r) + + threads = [threading.Thread(target=attempt_add, args=("c1",)) for _ in range(20)] + for t in threads: + t.start() + for t in threads: + t.join() + + # Exactly one thread sees the UI; the remaining 19 see None + # (duplicate). The forward + reverse indexes carry exactly one + # entry for c1. + successes = [r for r in results if r is ui] + nones = [r for r in results if r is None] + assert len(successes) == 1 + assert len(nones) == 19 + assert registry.children_of("p1") == ["c1"] + assert registry.parent_for("c1") == "p1" + + def test_concurrent_install_and_add_child_no_resurrect( + self, registry: ChildrenRegistry + ) -> None: + # add_child racing with uninstall: either lands first (registry + # populated) or the parent is gone (returns None). Must NOT + # leave a forward-set entry without presence — that would be + # the "resurrected after close" leak the locked dispatch path + # was guarding against. + ui = _Sentinel() + registry.install("p1", ui) + + outcomes: list[object] = [] + + def adder() -> None: + outcomes.append(registry.add_child("p1", "c1")) + + def uninstaller() -> None: + registry.uninstall("p1") + + threads = [ + threading.Thread(target=adder), + threading.Thread(target=uninstaller), + ] + for t in threads: + t.start() + for t in threads: + t.join() + + # If add_child landed first: c1 is in the forward set, then + # uninstall clears everything. End state: nothing. + # If uninstall landed first: add_child sees no presence, + # returns None, no entry added. End state: nothing. + # Either way, the leak invariant holds: child set is empty or + # parent is gone, never "child set populated but no presence". + children = registry.children_of("p1") + ui_present = registry.ui_for("p1") is not None + if children: + assert ui_present, "registry leaked: children set without presence" diff --git a/tests/test_console.py b/tests/test_console.py index 2a13f1ea..686b7442 100644 --- a/tests/test_console.py +++ b/tests/test_console.py @@ -314,12 +314,14 @@ class TestCollectorSnapshot: assert event["ws_id"] == "ws1" assert event["state"] == "running" - def test_apply_snapshot_state_change_forwards_pending_approval_detail(self): - """Reconnect-via-snapshot is the resync path after every console - restart or network blip. Without forwarding the field here, - a child sitting in approval-pending across the gap renders as - ``activity_state=approval`` with no buttons until the next - state change — broken UX during the most common re-sync event.""" + def test_apply_snapshot_state_change_does_not_carry_pending_approval_detail(self): + """Stage 3 cleanup — the snapshot-resync cluster_state event no + longer piggybacks ``pending_approval_detail`` (the field is + gone from cluster_state entirely). On reconnect the browser's + bulk fetch — triggered by the ``activity_state="approval"`` + transition in the reducer — pulls the items directly from + ``ui.serialize_pending_approval_detail()`` via the dashboard + endpoint.""" c = _make_collector() c._nodes["node-a"] = NodeSnapshot( node_id="node-a", @@ -329,10 +331,6 @@ class TestCollectorSnapshot: q: queue.Queue[dict] = queue.Queue() c.register_listener(q) - detail = { - "items": [{"call_id": "c1", "header": "tool x"}], - "judge_pending": False, - } c._apply_snapshot( "node-a", { @@ -344,7 +342,6 @@ class TestCollectorSnapshot: "name": "same", "state": "running", "activity_state": "approval", - "pending_approval_detail": detail, } ], "health": {}, @@ -354,7 +351,8 @@ class TestCollectorSnapshot: event = q.get_nowait() assert event["type"] == "cluster_state" - assert event["pending_approval_detail"] == detail + assert event["activity_state"] == "approval" + assert "pending_approval_detail" not in event def test_apply_snapshot_skips_empty_id_workstream(self): c = _make_collector() @@ -401,12 +399,12 @@ class TestCollectorDelta: # Verify in-memory state was updated assert c._nodes["node-a"].workstreams["ws1"]["state"] == "running" - def test_apply_delta_ws_state_forwards_pending_approval_detail(self): - """The rich approval payload now travels on the cluster bus so - coord tabs can render inline approve/deny buttons in lockstep - with the activity_state transition. Collector must forward - the field verbatim — the adapter does the child-routing on - top, but the bus carries the data.""" + def test_apply_delta_ws_state_does_not_carry_pending_approval_detail(self): + """Stage 3 cleanup — ``cluster_state`` no longer carries the + ``pending_approval_detail`` piggyback. Approval items now arrive + via bulk fetch on activity_state transition; verdicts via the + explicit ``intent_verdict`` event class. Symmetric event flow, + no piggyback to dedupe against.""" c = _make_collector() c._nodes["node-a"] = NodeSnapshot( node_id="node-a", @@ -416,10 +414,6 @@ class TestCollectorDelta: q: queue.Queue[dict] = queue.Queue() c.register_listener(q) - detail = { - "items": [{"call_id": "c1", "header": "tool x"}], - "judge_pending": False, - } c._apply_delta( "node-a", { @@ -427,13 +421,13 @@ class TestCollectorDelta: "ws_id": "ws1", "state": "running", "activity_state": "approval", - "pending_approval_detail": detail, }, ) event = q.get_nowait() assert event["type"] == "cluster_state" - assert event["pending_approval_detail"] == detail + assert event["activity_state"] == "approval" + assert "pending_approval_detail" not in event def test_apply_delta_ws_created(self): c = _make_collector() @@ -480,6 +474,139 @@ class TestCollectorDelta: assert event["name"] == "new-name" assert c._nodes["node-a"].workstreams["ws1"]["name"] == "new-name" + def test_apply_delta_intent_verdict_forwards_verbatim(self): + """Stage 3 Step 5 — node-emitted intent_verdict events flow + through _apply_delta to cluster fan-out so coord adapters can + re-emit as child_ws_intent_verdict on the parent's SSE.""" + c = _make_collector() + c._nodes["node-a"] = NodeSnapshot( + node_id="node-a", + server_url="http://a:8080", + workstreams={"ws1": {"id": "ws1", "name": "test", "state": "idle"}}, + ) + q: queue.Queue[dict] = queue.Queue() + c.register_listener(q) + + verdict = { + "call_id": "c1", + "risk_level": "low", + "confidence": 0.9, + "recommendation": "approve", + } + c._apply_delta( + "node-a", + {"type": "intent_verdict", "ws_id": "ws1", "verdict": verdict}, + ) + + event = q.get_nowait() + assert event["type"] == "intent_verdict" + assert event["ws_id"] == "ws1" + assert event["node_id"] == "node-a" + assert event["verdict"] == verdict + + def test_apply_delta_intent_verdict_drops_when_ws_id_missing(self): + c = _make_collector() + c._nodes["node-a"] = NodeSnapshot(node_id="node-a", server_url="http://a:8080") + q: queue.Queue[dict] = queue.Queue() + c.register_listener(q) + + c._apply_delta("node-a", {"type": "intent_verdict", "verdict": {}}) + + assert q.empty() + + def test_apply_delta_approval_resolved_forwards_verbatim(self): + """Stage 3 Step 5 — paired with intent_verdict; clears the + coord tree's pending-approval pill in lockstep with the + actual decision rather than waiting for the state-change + piggyback.""" + c = _make_collector() + c._nodes["node-a"] = NodeSnapshot( + node_id="node-a", + server_url="http://a:8080", + workstreams={"ws1": {"id": "ws1", "name": "test", "state": "idle"}}, + ) + q: queue.Queue[dict] = queue.Queue() + c.register_listener(q) + + c._apply_delta( + "node-a", + { + "type": "approval_resolved", + "ws_id": "ws1", + "approved": True, + "feedback": "lgtm", + "always": False, + }, + ) + + event = q.get_nowait() + assert event["type"] == "approval_resolved" + assert event["ws_id"] == "ws1" + assert event["node_id"] == "node-a" + assert event["approved"] is True + assert event["feedback"] == "lgtm" + assert event["always"] is False + + def test_apply_delta_approve_request_forwards_detail(self): + """Push path for the initial approval items — eliminates the + bulk-fetch race that left the coord row stuck on a loading + placeholder when the bulk fetch landed in the gap between + _emit_state(ATTENTION) and approve_tools setting _pending_approval.""" + c = _make_collector() + c._nodes["node-a"] = NodeSnapshot( + node_id="node-a", + server_url="http://a:8080", + workstreams={"ws1": {"id": "ws1", "name": "test", "state": "idle"}}, + ) + q: queue.Queue[dict] = queue.Queue() + c.register_listener(q) + + detail = { + "type": "approve_request", + "items": [{"call_id": "c1", "header": "tool x"}], + "judge_pending": True, + } + c._apply_delta( + "node-a", + {"type": "approve_request", "ws_id": "ws1", "detail": detail}, + ) + + event = q.get_nowait() + assert event["type"] == "approve_request" + assert event["ws_id"] == "ws1" + assert event["node_id"] == "node-a" + assert event["detail"] == detail + + def test_apply_delta_approve_request_drops_when_ws_id_missing(self): + c = _make_collector() + c._nodes["node-a"] = NodeSnapshot(node_id="node-a", server_url="http://a:8080") + q: queue.Queue[dict] = queue.Queue() + c.register_listener(q) + + c._apply_delta("node-a", {"type": "approve_request", "detail": {}}) + + assert q.empty() + + def test_apply_delta_approval_resolved_coerces_missing_fields(self): + """Defensive: ``approved`` / ``always`` / ``feedback`` may be + omitted by older nodes mid-rolling-upgrade; collector coerces + to safe defaults.""" + c = _make_collector() + c._nodes["node-a"] = NodeSnapshot( + node_id="node-a", + server_url="http://a:8080", + workstreams={"ws1": {"id": "ws1", "name": "test", "state": "idle"}}, + ) + q: queue.Queue[dict] = queue.Queue() + c.register_listener(q) + + c._apply_delta("node-a", {"type": "approval_resolved", "ws_id": "ws1"}) + + event = q.get_nowait() + assert event["approved"] is False + assert event["feedback"] == "" + assert event["always"] is False + def test_apply_delta_health_changed(self): c = _make_collector() c._nodes["node-a"] = NodeSnapshot( diff --git a/tests/test_coord_ui_approve_tools.py b/tests/test_coord_ui_approve_tools.py index 57ce9c34..e62635e4 100644 --- a/tests/test_coord_ui_approve_tools.py +++ b/tests/test_coord_ui_approve_tools.py @@ -464,3 +464,148 @@ def test_coord_budget_override_survives_wildcard_allow_policy() -> None: 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" + + +# --------------------------------------------------------------------------- +# Cluster-bus broadcast hooks — _broadcast_intent_verdict / _approval_resolved +# --------------------------------------------------------------------------- + + +class TestBroadcastIntentVerdict: + """``ConsoleCoordinatorUI._broadcast_intent_verdict`` overrides the + no-op base hook to push the verdict onto the cluster bus via + ``ClusterCollector.emit_console_ws_intent_verdict``. The far more + common path is the per-node ``WebUI`` override (covered in + test_webui_content.py); this lights up the rare coord-self path + (a coord that runs its own LLM judge). + """ + + def test_calls_collector_emit_with_ws_id_and_verdict(self) -> None: + ui = ConsoleCoordinatorUI(ws_id="coord-a", user_id="u1") + collector = MagicMock() + ConsoleCoordinatorUI._collector = collector + try: + verdict = { + "call_id": "c1", + "risk_level": "high", + "confidence": 0.91, + } + ui._broadcast_intent_verdict(verdict) + collector.emit_console_ws_intent_verdict.assert_called_once_with( + "coord-a", + verdict, + ) + finally: + ConsoleCoordinatorUI._collector = None + + def test_no_op_when_collector_unset(self) -> None: + ui = ConsoleCoordinatorUI(ws_id="coord-a", user_id="u1") + ConsoleCoordinatorUI._collector = None + # Doesn't raise. + ui._broadcast_intent_verdict({"call_id": "c1"}) + + def test_collector_exception_swallowed(self) -> None: + ui = ConsoleCoordinatorUI(ws_id="coord-a", user_id="u1") + collector = MagicMock() + collector.emit_console_ws_intent_verdict.side_effect = RuntimeError("boom") + ConsoleCoordinatorUI._collector = collector + try: + # Doesn't raise — collector failures are observational only. + ui._broadcast_intent_verdict({"call_id": "c1"}) + finally: + ConsoleCoordinatorUI._collector = None + + +class TestBroadcastApprovalResolved: + """``ConsoleCoordinatorUI._broadcast_approval_resolved`` overrides + the base hook to push the resolution onto the cluster bus via + ``ClusterCollector.emit_console_ws_approval_resolved``.""" + + def test_calls_collector_with_decision_fields(self) -> None: + ui = ConsoleCoordinatorUI(ws_id="coord-a", user_id="u1") + collector = MagicMock() + ConsoleCoordinatorUI._collector = collector + try: + ui._broadcast_approval_resolved(True, "lgtm", always=True) + collector.emit_console_ws_approval_resolved.assert_called_once_with( + "coord-a", + approved=True, + feedback="lgtm", + always=True, + ) + finally: + ConsoleCoordinatorUI._collector = None + + def test_normalises_none_feedback_to_empty_string(self) -> None: + ui = ConsoleCoordinatorUI(ws_id="coord-a", user_id="u1") + collector = MagicMock() + ConsoleCoordinatorUI._collector = collector + try: + ui._broadcast_approval_resolved(False, None) + collector.emit_console_ws_approval_resolved.assert_called_once_with( + "coord-a", + approved=False, + feedback="", + always=False, + ) + finally: + ConsoleCoordinatorUI._collector = None + + def test_no_op_when_collector_unset(self) -> None: + ui = ConsoleCoordinatorUI(ws_id="coord-a", user_id="u1") + ConsoleCoordinatorUI._collector = None + # Doesn't raise. + ui._broadcast_approval_resolved(True, None) + + def test_collector_exception_swallowed(self) -> None: + ui = ConsoleCoordinatorUI(ws_id="coord-a", user_id="u1") + collector = MagicMock() + collector.emit_console_ws_approval_resolved.side_effect = RuntimeError("boom") + ConsoleCoordinatorUI._collector = collector + try: + # Doesn't raise. + ui._broadcast_approval_resolved(True, "ok") + finally: + ConsoleCoordinatorUI._collector = None + + +class TestBroadcastApproveRequest: + """Coord-side override for the approve_request push. Same rationale + as the WebUI override — the coord-self path is rare today, but + parity keeps the override symmetric with the rest of the broadcast + family.""" + + def test_calls_collector_emit_with_ws_id_and_detail(self) -> None: + ui = ConsoleCoordinatorUI(ws_id="coord-a", user_id="u1") + collector = MagicMock() + ConsoleCoordinatorUI._collector = collector + try: + detail = { + "type": "approve_request", + "items": [{"call_id": "c1", "header": "tool x"}], + "judge_pending": True, + } + ui._broadcast_approve_request(detail) + collector.emit_console_ws_approve_request.assert_called_once_with( + "coord-a", + detail, + ) + finally: + ConsoleCoordinatorUI._collector = None + + def test_no_op_when_collector_unset(self) -> None: + ui = ConsoleCoordinatorUI(ws_id="coord-a", user_id="u1") + ConsoleCoordinatorUI._collector = None + # Doesn't raise. + ui._broadcast_approve_request({"items": []}) + + def test_collector_exception_swallowed(self) -> None: + ui = ConsoleCoordinatorUI(ws_id="coord-a", user_id="u1") + collector = MagicMock() + collector.emit_console_ws_approve_request.side_effect = RuntimeError("boom") + ConsoleCoordinatorUI._collector = collector + try: + # Doesn't raise. + ui._broadcast_approve_request({"items": []}) + finally: + ConsoleCoordinatorUI._collector = None diff --git a/tests/test_coordinator_adapter.py b/tests/test_coordinator_adapter.py index c1decbf3..47328ce4 100644 --- a/tests/test_coordinator_adapter.py +++ b/tests/test_coordinator_adapter.py @@ -378,13 +378,21 @@ class TestCoordinatorAdapterWorkerDispatch: class TestCoordinatorAdapterChildrenRegistry: - def test_emit_created_seeds_empty_children_set(self) -> None: + """Adapter-level integration with :class:`ChildrenRegistry`. + + Pure-registry invariants (forward/reverse consistency, idempotent + merge, locking) live in ``test_children_registry.py``. These + tests cover the adapter's wiring: that ``emit_*`` paths drive the + registry correctly and that the snapshot-priming bridge between + a collector snapshot and the registry preserves merge semantics. + """ + + def test_emit_created_installs_parent(self) -> None: adapter, _ = _make_adapter() ws = _make_ws() adapter.emit_created(ws) - assert ws.id in adapter._children - assert adapter._children[ws.id] == set() - assert adapter._active_coords[ws.id] is ws.ui + assert adapter._registry.children_of(ws.id) == [] + assert adapter._registry.ui_for(ws.id) is ws.ui def test_emit_rehydrated_calls_rebuild(self) -> None: adapter, _ = _make_adapter() @@ -398,42 +406,36 @@ class TestCoordinatorAdapterChildrenRegistry: adapter.emit_rehydrated(ws) assert calls == [ws.id] - def test_emit_closed_clears_forward_and_reverse_indexes(self) -> None: + def test_emit_closed_uninstalls_parent_and_clears_children(self) -> None: adapter, _ = _make_adapter() - with adapter._children_lock: - adapter._merge_child_ids_locked("coord-a", ["child-a1", "child-a2"]) - adapter._merge_child_ids_locked("coord-b", ["child-b1"]) - adapter._active_coords["coord-a"] = object() - adapter._active_coords["coord-b"] = object() + adapter._registry.install("coord-a", object()) + adapter._registry.install("coord-b", object()) + adapter._registry.merge_children("coord-a", ["child-a1", "child-a2"]) + adapter._registry.merge_children("coord-b", ["child-b1"]) adapter.emit_closed("coord-a") - assert "coord-a" not in adapter._children - assert "coord-a" not in adapter._active_coords - assert "child-a1" not in adapter._child_to_coord - assert "child-a2" not in adapter._child_to_coord + assert adapter._registry.ui_for("coord-a") is None + assert adapter._registry.children_of("coord-a") == [] + assert adapter._registry.parent_for("child-a1") is None + assert adapter._registry.parent_for("child-a2") is None # coord-b untouched - assert adapter._child_to_coord["child-b1"] == "coord-b" - assert "coord-b" in adapter._children - - def test_merge_child_ids_locked_is_idempotent(self) -> None: - adapter, _ = _make_adapter() - with adapter._children_lock: - adapter._merge_child_ids_locked("coord-a", ["child-1"]) - adapter._merge_child_ids_locked("coord-a", ["child-1"]) - assert adapter._children["coord-a"] == {"child-1"} - assert adapter._child_to_coord == {"child-1": "coord-a"} + assert adapter._registry.parent_for("child-b1") == "coord-b" + assert adapter._registry.ui_for("coord-b") is not None def test_prime_children_from_snapshot_merges_without_overwriting(self) -> None: + # Snapshot priming now lives on ClusterChildSource (production + # path). The adapter no longer carries its own duplicate copy. + from turnstone.core.child_source import ClusterChildSource + adapter, _ = _make_adapter() - # Seed one in-memory coord + one existing child - coord_ws = _make_ws() - coord_ws.id = "coord-a" - mgr = MagicMock() - mgr.list_all.return_value = [coord_ws] - adapter.attach(mgr) - with adapter._children_lock: - adapter._merge_child_ids_locked("coord-a", ["child-a1"]) + adapter._registry.merge_children("coord-a", ["child-a1"]) + + source = ClusterChildSource( + collector=MagicMock(), + registry=adapter._registry, + parents_provider=lambda: ["coord-a"], + ) snapshot = { "nodes": [ @@ -448,10 +450,13 @@ class TestCoordinatorAdapterChildrenRegistry: }, ], } - adapter._prime_children_from_snapshot(snapshot) - assert adapter._children["coord-a"] == {"child-a1", "child-a2"} - assert adapter._child_to_coord["child-a2"] == "coord-a" - assert "child-x" not in adapter._child_to_coord + source._prime_from_snapshot(snapshot) + assert set(adapter._registry.children_of("coord-a")) == { + "child-a1", + "child-a2", + } + assert adapter._registry.parent_for("child-a2") == "coord-a" + assert adapter._registry.parent_for("child-x") is None # --------------------------------------------------------------------------- @@ -478,9 +483,7 @@ class TestCoordinatorAdapterDispatchChildEvent: coord_ws.id = coord_id recorder = _UIRecorder() coord_ws.ui = recorder # type: ignore[assignment] - with adapter._children_lock: - adapter._children.setdefault(coord_id, set()) - adapter._active_coords[coord_id] = recorder + adapter._registry.install(coord_id, recorder) adapter.attach(_StubManager(coord_ws)) # type: ignore[arg-type] return adapter, recorder, coord_ws @@ -510,12 +513,11 @@ class TestCoordinatorAdapterDispatchChildEvent: assert payload["child_ws_id"] == "child-a1" assert payload["parent_ws_id"] == "coord-a" # Reverse index updated for subsequent cluster_state events. - assert adapter._child_to_coord["child-a1"] == "coord-a" + assert adapter._registry.parent_for("child-a1") == "coord-a" def test_dispatch_cluster_state_routes_via_reverse_index(self) -> None: adapter, recorder, _ = self._setup() - with adapter._children_lock: - adapter._merge_child_ids_locked("coord-a", ["child-a1"]) + adapter._registry.merge_children("coord-a", ["child-a1"]) adapter._dispatch_child_event( { "type": "cluster_state", @@ -533,8 +535,7 @@ class TestCoordinatorAdapterDispatchChildEvent: def test_dispatch_ws_closed_routes_to_parent_coord(self) -> None: adapter, recorder, _ = self._setup() - with adapter._children_lock: - adapter._merge_child_ids_locked("coord-a", ["child-a1"]) + adapter._registry.merge_children("coord-a", ["child-a1"]) adapter._dispatch_child_event( {"type": "ws_closed", "ws_id": "child-a1", "reason": "evicted"} ) @@ -548,8 +549,7 @@ class TestCoordinatorAdapterDispatchChildEvent: """perf-6: _enqueue_on_ui mutates the payload dict in place with the coord's ws_id so the browser can discriminate child events.""" adapter, recorder, _ = self._setup() - with adapter._children_lock: - adapter._merge_child_ids_locked("coord-a", ["child-a1"]) + adapter._registry.merge_children("coord-a", ["child-a1"]) adapter._dispatch_child_event( { "type": "cluster_state", @@ -559,53 +559,160 @@ class TestCoordinatorAdapterDispatchChildEvent: ) assert recorder.enqueued[0]["ws_id"] == "coord-a" - def test_dispatch_cluster_state_forwards_pending_approval_detail(self) -> None: - """The rich approval payload now rides on child_ws_state directly so - the browser can mutate liveBadgeCache without a separate live-bulk - fetch. Drift here means the inline approve/deny buttons would - regress to chasing the dashboard cache (the load-storm pattern - Shape A is unwinding).""" + def test_dispatch_cluster_state_does_not_carry_pending_approval_detail( + self, + ) -> None: + """Stage 3 cleanup — the ``pending_approval_detail`` piggyback + on ``cluster_state`` is gone. Approval items now arrive via + bulk fetch (triggered by ``activity_state="approval"`` in the + browser); verdicts via ``child_ws_intent_verdict``; resolution + via ``child_ws_approval_resolved``. The state event carries + only state + activity_state — no detail field.""" adapter, recorder, _ = self._setup() - with adapter._children_lock: - adapter._merge_child_ids_locked("coord-a", ["child-a1"]) - detail = { - "items": [{"call_id": "c1", "header": "tool x"}], - "judge_pending": False, - } + adapter._registry.merge_children("coord-a", ["child-a1"]) adapter._dispatch_child_event( { "type": "cluster_state", "ws_id": "child-a1", "state": "running", "activity_state": "approval", - "pending_approval_detail": detail, } ) assert len(recorder.enqueued) == 1 payload = recorder.enqueued[0] assert payload["type"] == "child_ws_state" assert payload["activity_state"] == "approval" - assert payload["pending_approval_detail"] == detail + assert "pending_approval_detail" not in payload - def test_dispatch_cluster_state_pending_approval_detail_none_passes_through( - self, - ) -> None: - """Missing pending_approval_detail (no approval pending, or pre-fix - node mid-rolling-upgrade) must forward as None — not raise, not - omit — so the browser's handleChildState treats it as "no SSE- - supplied detail, fall back to cached value".""" + def test_dispatch_intent_verdict_emits_child_ws_intent_verdict(self) -> None: + """Stage 3 Step 6 — explicit verdict events are re-emitted as + child_ws_intent_verdict on the parent's SSE so the tree UI + renders the risk pill without polling.""" adapter, recorder, _ = self._setup() - with adapter._children_lock: - adapter._merge_child_ids_locked("coord-a", ["child-a1"]) + adapter._registry.merge_children("coord-a", ["child-a1"]) + verdict = { + "call_id": "c1", + "risk_level": "low", + "confidence": 0.92, + "recommendation": "approve", + } adapter._dispatch_child_event( { - "type": "cluster_state", + "type": "intent_verdict", "ws_id": "child-a1", - "state": "running", - "activity_state": "tool", + "node_id": "node-1", + "verdict": verdict, } ) assert len(recorder.enqueued) == 1 payload = recorder.enqueued[0] - assert "pending_approval_detail" in payload - assert payload["pending_approval_detail"] is None + assert payload["type"] == "child_ws_intent_verdict" + assert payload["child_ws_id"] == "child-a1" + assert payload["parent_ws_id"] == "coord-a" + assert payload["node_id"] == "node-1" + assert payload["verdict"] == verdict + + def test_dispatch_intent_verdict_unknown_child_drops(self) -> None: + adapter, recorder, _ = self._setup() + adapter._dispatch_child_event( + { + "type": "intent_verdict", + "ws_id": "ws-orphan", + "verdict": {"call_id": "c1"}, + } + ) + assert recorder.enqueued == [] + + def test_dispatch_approval_resolved_emits_child_ws_approval_resolved( + self, + ) -> None: + """Stage 3 Step 6 — paired with intent_verdict; clears the + pending-approval pill on the parent's tree UI in lockstep + with the actual decision.""" + adapter, recorder, _ = self._setup() + adapter._registry.merge_children("coord-a", ["child-a1"]) + adapter._dispatch_child_event( + { + "type": "approval_resolved", + "ws_id": "child-a1", + "node_id": "node-1", + "approved": True, + "feedback": "lgtm", + "always": False, + } + ) + assert len(recorder.enqueued) == 1 + payload = recorder.enqueued[0] + assert payload["type"] == "child_ws_approval_resolved" + assert payload["child_ws_id"] == "child-a1" + assert payload["parent_ws_id"] == "coord-a" + assert payload["approved"] is True + assert payload["feedback"] == "lgtm" + assert payload["always"] is False + + def test_dispatch_approval_resolved_coerces_missing_fields(self) -> None: + """Older nodes mid-rolling-upgrade may omit approved / always / + feedback; dispatch coerces to safe defaults.""" + adapter, recorder, _ = self._setup() + adapter._registry.merge_children("coord-a", ["child-a1"]) + adapter._dispatch_child_event({"type": "approval_resolved", "ws_id": "child-a1"}) + assert len(recorder.enqueued) == 1 + payload = recorder.enqueued[0] + assert payload["approved"] is False + assert payload["feedback"] == "" + assert payload["always"] is False + + def test_dispatch_approval_resolved_unknown_child_drops(self) -> None: + """Symmetric to the intent_verdict drop test — events for + ws_ids the registry doesn't know about silently drop instead + of fanning out to a parent that has no business seeing them.""" + adapter, recorder, _ = self._setup() + adapter._dispatch_child_event( + { + "type": "approval_resolved", + "ws_id": "ws-orphan", + "approved": True, + }, + ) + assert recorder.enqueued == [] + + def test_dispatch_approve_request_emits_child_ws_approve_request( + self, + ) -> None: + """Push path for the initial approval items — eliminates the + bulk-fetch race that left the coord row stuck on a loading + placeholder when the bulk fetch landed in the gap between + _emit_state(ATTENTION) and approve_tools setting _pending_approval.""" + adapter, recorder, _ = self._setup() + adapter._registry.merge_children("coord-a", ["child-a1"]) + detail = { + "type": "approve_request", + "items": [{"call_id": "c1", "header": "tool x"}], + "judge_pending": True, + } + adapter._dispatch_child_event( + { + "type": "approve_request", + "ws_id": "child-a1", + "node_id": "node-1", + "detail": detail, + }, + ) + assert len(recorder.enqueued) == 1 + payload = recorder.enqueued[0] + assert payload["type"] == "child_ws_approve_request" + assert payload["child_ws_id"] == "child-a1" + assert payload["parent_ws_id"] == "coord-a" + assert payload["node_id"] == "node-1" + assert payload["detail"] == detail + + def test_dispatch_approve_request_unknown_child_drops(self) -> None: + adapter, recorder, _ = self._setup() + adapter._dispatch_child_event( + { + "type": "approve_request", + "ws_id": "ws-orphan", + "detail": {"items": []}, + }, + ) + assert recorder.enqueued == [] diff --git a/tests/test_coordinator_page.py b/tests/test_coordinator_page.py index 1bcc27a4..4b0f0d87 100644 --- a/tests/test_coordinator_page.py +++ b/tests/test_coordinator_page.py @@ -79,9 +79,10 @@ def test_coordinator_js_exposes_inline_approval_helpers(): assert "function submitChildApproval" in body or "submitChildApproval(" in body # The shared approve POST helper (parameterized for child ws_ids) assert "function approveWorkstream" in body or "approveWorkstream(" in body - # The urgent live-bulk fetch option that fires on activity_state - # transitions in/out of "approval" - assert "{ urgent: true }" in body or "urgent: true" in body + # The 409 stale-call_id retry path uses invalidateLiveBadge + + # scheduleLiveFetch (Stage 3 cleanup removed the urgent flag — + # cache invalidation makes the TTL gate fall through naturally). + assert "invalidateLiveBadge(targetWsId)" in body # Server-side payload field — drift here means the JS reads stale keys assert "pending_approval_detail" in body # Reconnect parity (chunk 4): the SSE re-open handler must drop @@ -90,10 +91,10 @@ def test_coordinator_js_exposes_inline_approval_helpers(): # can't render zombie approve/deny buttons on a row whose # approval was resolved during the gap. The implementation # iterates the cache and deletes only !permanent entries — - # asserting the literal Map iteration form keeps a refactor - # back to liveBadgeCache.clear() (which would re-pay 403s on - # every reconnect for denied ids) from sneaking in. - assert "liveBadgeCache.delete" in body + # asserting the literal helper call keeps a refactor back to + # _liveBadgeCacheClear() (which would re-pay 403s on every + # reconnect for denied ids) from sneaking in. + assert "_liveBadgeCacheDelete" in body # Edge-case matrix sentinel labels — POLICY-BLOCKED renders when # an item has error set + needs_approval=False (server-side # tool policy already blocked the call); "(judge unavailable)" @@ -113,15 +114,19 @@ def test_coordinator_js_exposes_inline_approval_helpers(): # coord-self ws_id (the coord lives on the console process). # Children live on cluster nodes and 404 without the prefix. assert "/v1/api/route/workstreams/" in body - # Late-judge polling — the LLM judge runs async on the child - # node and never pushes a signal that reaches the coord, so - # the row's pending_approval_detail with judge_pending=true - # would freeze on heuristic verdicts forever without this - # poll loop. The poller is GLOBAL (not per-row) so off-screen - # rows still refresh — a per-row poller's scheduleLiveFetch - # call short-circuits on non-visible rows, leaving them stuck. - assert "_maybeStartJudgePoll" in body - assert "_judgePollTick" in body + # Late-arriving LLM judge verdicts — Stage 3 Step 5 promoted + # ``intent_verdict`` and ``approval_resolved`` to first-class + # cluster-bus event types, so the coord adapter dispatches them + # as ``child_ws_intent_verdict`` / ``child_ws_approval_resolved`` + # on the parent's SSE stream. The browser handlers write + # directly to liveBadgeCache (bypassing scheduleLiveFetch's + # visibility gate cleanly) so off-screen rows pick up verdicts + # without polling. Replaced the old ``_judgePollTick`` 90-second + # global poll loop and its visibility-gate-bypass workaround. + assert "handleChildIntentVerdict" in body + assert "handleChildApprovalResolved" in body + assert "child_ws_intent_verdict" in body + assert "child_ws_approval_resolved" in body # Reload parity for the coord-self approval gate: init() must # consume the authoritative GET /workstreams snapshot's # pending_approval_detail so a freshly opened tab can render @@ -154,18 +159,19 @@ def test_coordinator_js_exposes_inline_approval_helpers(): assert "callOutcomes" in body -def test_coordinator_js_handle_child_state_reads_sse_pending_approval_detail(): - """Lock the Shape A behavior change: child_ws_state SSE events now - carry ``pending_approval_detail`` directly so the browser mutates - ``liveBadgeCache`` without firing an urgent live-bulk fetch on - every activity_state transition into/out of approval. A refactor - that re-introduces the urgent-fetch path on routine transitions - (or drops the SSE-source merge guard in flushLiveFetches) would - re-open the load-storm pattern this PR is fixing. +def test_coordinator_js_handle_child_state_no_longer_reads_sse_pending_approval_detail(): + """Stage 3 cleanup — ``pending_approval_detail`` is no longer + piggybacked on child_ws_state events. Approval items now arrive + via bulk fetch on the activity_state="approval" transition; + verdicts via the explicit ``child_ws_intent_verdict`` event class; + resolution via ``child_ws_approval_resolved``. A refactor that + re-introduces the piggyback would silently re-open the + duplicate-path race the dedicated event classes were added to + eliminate. Structural assertions (regex against multi-line source) — symbol- presence alone wouldn't catch a guard that keeps the names but - inverts the comparison or drops the ``prev.live`` check. This + inverts the comparison or drops the ``prev.live`` check. This codebase has no JS test framework, so locking the guard's shape here is the next-best thing to a behavioral test.""" import re @@ -176,26 +182,46 @@ def test_coordinator_js_handle_child_state_reads_sse_pending_approval_detail(): ) body = coord_js.read_text(encoding="utf-8") - # handleChildState now reads the SSE-supplied detail. - assert "ev.pending_approval_detail" in body - # The pre-fix urgent-fetch on activity_state transitions is - # gone (the 409 retry path keeps its own ``{ urgent: true }`` - # for stale-call_id refresh — that's a different scenario). + # The piggyback read is gone from handleChildState. (The string + # may still appear elsewhere — e.g. handleChildIntentVerdict + # reading from cache, or comments — but never as ``ev.pending_approval_detail``.) + assert "ev.pending_approval_detail" not in body + # The pre-fix urgent-fetch on activity_state transitions is gone. assert "enteredApproval" not in body assert "leftApproval" not in body + # ``pendingApproval`` flag derivation must check BOTH state and + # activity_state. The worker thread can fire the state transition + # to "attention" before approve_tools updates activity_state, so + # checking only activity_state misses children that legitimately + # need approval. Pin the disjunction so the regression doesn't + # silently re-introduce. + assert re.search( + r'existing\.state\s*===\s*"attention"\s*\|\|\s*' + r'existing\.activity_state\s*===\s*"approval"', + body, + ), ( + "handleChildState must derive pendingApproval from " + "(state==='attention' || activity_state==='approval')" + ) # SSE-authoritative window constant is defined and used. assert re.search(r"\bconst\s+SSE_AUTHORITATIVE_MS\s*=\s*\d+", body), ( "SSE_AUTHORITATIVE_MS constant must be defined as a numeric literal" ) - # handleChildState writes sseUpdatedAt = Date.now() into the cache - # entry it sets. This is the SSE-source tag; without it, the - # merge guard in flushLiveFetches has nothing to gate on. + # SSE writers tag entries with sseUpdatedAt: Date.now() so the + # merge guard in flushLiveFetches preserves them against stale + # bulk-fetch responses. handleChildState only stamps when it + # AUTHORITATIVELY clears the detail (off-approval transition); + # writers that stamp unconditionally are intent_verdict (verdict + # stamp), approval_resolved (clear), and the optimistic-clear + # path in submitChildApproval. Pinning the literal Date.now() + # call keeps a refactor that drops the SSE-source tag entirely + # from sneaking in. assert re.search( r"sseUpdatedAt:\s*Date\.now\(\)", body, - ), "handleChildState must write sseUpdatedAt: Date.now() onto liveBadgeCache entries" + ), "Critical SSE writers must stamp sseUpdatedAt: Date.now()" # flushLiveFetches' merge guard structure: SSE-set pending_approval # / _detail wins over a stale bulk-poll snapshot when (live) AND diff --git a/tests/test_session_manager.py b/tests/test_session_manager.py index 656f5e7d..ac9772e3 100644 --- a/tests/test_session_manager.py +++ b/tests/test_session_manager.py @@ -19,7 +19,10 @@ import threading import time from dataclasses import dataclass from datetime import UTC, datetime -from typing import Any +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from collections.abc import Callable from unittest.mock import MagicMock import pytest @@ -1368,3 +1371,127 @@ class TestSessionManagerWithStateWriter: assert "running" not in ws_writes, ( f"set_state after close enqueued through buffer: {ws_writes}" ) + + +# --------------------------------------------------------------------------- +# Multi-subscriber observer — subscribe_to_state / unsubscribe_from_state +# --------------------------------------------------------------------------- + + +class TestStateSubscribers: + """Multi-subscriber observer for ``set_state``. + + Used by the CLI's background-attention notifier and by + ``SameNodeChildSource``. Subscribe / unsubscribe must be safe under + concurrent dispatch, and dispatch must not skip / repeat callbacks + when subscribers register or unregister mid-iteration. + """ + + def test_subscribe_fires_on_set_state(self) -> None: + mgr, _, _ = _make_manager() + ws = mgr.create(user_id="u1", name="ws", skill=None) + events: list[tuple[str, str]] = [] + + def cb(ws_id: str, state: WorkstreamState) -> None: + events.append((ws_id, state.value)) + + mgr.subscribe_to_state(cb) + mgr.set_state(ws.id, WorkstreamState.RUNNING) + assert events == [(ws.id, "running")] + + def test_unsubscribe_stops_firing(self) -> None: + mgr, _, _ = _make_manager() + ws = mgr.create(user_id="u1", name="ws", skill=None) + events: list[str] = [] + + def cb(_ws_id: str, state: WorkstreamState) -> None: + events.append(state.value) + + mgr.subscribe_to_state(cb) + mgr.unsubscribe_from_state(cb) + mgr.set_state(ws.id, WorkstreamState.RUNNING) + assert events == [] + + def test_unsubscribe_unknown_is_noop(self) -> None: + mgr, _, _ = _make_manager() + # Doesn't raise. + mgr.unsubscribe_from_state(lambda *_: None) + + def test_multiple_subscribers_fire_in_registration_order(self) -> None: + mgr, _, _ = _make_manager() + ws = mgr.create(user_id="u1", name="ws", skill=None) + order: list[int] = [] + + def make(i: int) -> Callable[[str, WorkstreamState], None]: + def cb(_ws_id: str, _state: WorkstreamState) -> None: + order.append(i) + + return cb + + mgr.subscribe_to_state(make(1)) + mgr.subscribe_to_state(make(2)) + mgr.subscribe_to_state(make(3)) + mgr.set_state(ws.id, WorkstreamState.IDLE) + assert order == [1, 2, 3] + + def test_subscriber_exception_does_not_block_others(self) -> None: + mgr, _, _ = _make_manager() + ws = mgr.create(user_id="u1", name="ws", skill=None) + survived: list[str] = [] + + def boom(*_: Any) -> None: + raise RuntimeError("subscriber crash") + + def good(_ws_id: str, state: WorkstreamState) -> None: + survived.append(state.value) + + mgr.subscribe_to_state(boom) + mgr.subscribe_to_state(good) + mgr.set_state(ws.id, WorkstreamState.RUNNING) + assert survived == ["running"] + + @pytest.mark.parametrize("n_threads", [10, 50]) + def test_concurrent_subscribe(self, n_threads: int) -> None: + """Subscribe from many threads; all callbacks land in the list. + + Validates the lock around mutation — without it the underlying + list.append could lose entries under contention. + """ + mgr, _, _ = _make_manager() + callbacks = [lambda *_, i=i: None for i in range(n_threads)] + threads = [threading.Thread(target=mgr.subscribe_to_state, args=(cb,)) for cb in callbacks] + for t in threads: + t.start() + for t in threads: + t.join() + # Snapshot under the lock to read the count safely. + with mgr._state_subscribers_lock: + assert len(mgr._state_subscribers) == n_threads + + def test_subscribe_during_dispatch_does_not_corrupt_iteration(self) -> None: + """A subscriber that calls subscribe_to_state during its own + callback must not affect the in-flight dispatch (snapshot + isolation). This is the bug-1 invariant: mutation during + iteration can't shift the iterator's index because dispatch + iterates a snapshot, not the live list. + """ + mgr, _, _ = _make_manager() + ws = mgr.create(user_id="u1", name="ws", skill=None) + fired: list[str] = [] + + def late(_ws_id: str, state: WorkstreamState) -> None: + fired.append("late:" + state.value) + + def first(_ws_id: str, state: WorkstreamState) -> None: + fired.append("first:" + state.value) + mgr.subscribe_to_state(late) # mid-dispatch addition + + mgr.subscribe_to_state(first) + mgr.set_state(ws.id, WorkstreamState.RUNNING) + # ``late`` was added during dispatch but the snapshot was + # already frozen — so it doesn't fire on this round. + assert fired == ["first:running"] + # Next round it does fire, in registration order after first. + fired.clear() + mgr.set_state(ws.id, WorkstreamState.IDLE) + assert fired == ["first:idle", "late:idle"] diff --git a/tests/test_webui_content.py b/tests/test_webui_content.py index b426f1fe..372e57aa 100644 --- a/tests/test_webui_content.py +++ b/tests/test_webui_content.py @@ -157,20 +157,17 @@ class TestContentAccumulation: assert len(idle_events[0]["content"]) <= _MAX_TURN_CONTENT_CHARS + 1024 -class TestPendingApprovalDetailGate: - """The Shape A SSE plumbing carries ``pending_approval_detail`` on the - ``ws_state`` event so the coord tree UI can render inline approve/deny - buttons in lockstep with the activity_state transition. The gate - (``if self._pending_approval is not None``) keeps the per-broadcast - serializer cost off the common no-approval-pending path — these tests - lock both branches down.""" +class TestPendingApprovalDetailNotPiggybacked: + """Stage 3 cleanup — ``pending_approval_detail`` is no longer + piggybacked on ``ws_state`` events. Approval items now arrive via + bulk fetch when the coord tree's reducer sees the + ``activity_state="approval"`` transition; verdicts via the explicit + ``intent_verdict`` event class; resolution via + ``approval_resolved``. These tests lock the no-piggyback contract + down so a future regression doesn't silently re-introduce the + duplicated path.""" def test_state_broadcast_omits_field_when_no_approval_pending(self): - """Common case: no approval pending → field absent from event so the - per-broadcast verdict-cache deepcopy in - ``serialize_pending_approval_detail`` never runs. A regression - that drops the gate would silently 10x the cost of every state - broadcast in the steady state.""" ui = _make_ui() assert ui._pending_approval is None ui._broadcast_state("running") @@ -180,17 +177,12 @@ class TestPendingApprovalDetailGate: assert len(running_events) == 1 assert "pending_approval_detail" not in running_events[0] - def test_state_broadcast_includes_field_when_approval_pending(self): - """When an approval is pending the broadcast must carry the rich - payload — the coord tree UI reads it directly to render inline - approve/deny buttons. Without this, a coord browser would have - to chase a separate ``cluster/ws/live`` fetch on every - activity_state transition (the load-storm pattern Shape A is - unwinding).""" + def test_state_broadcast_omits_field_even_when_approval_pending(self): + """The piggyback is gone: even when ``_pending_approval`` is set, + the state broadcast must NOT carry ``pending_approval_detail``. + The browser triggers a bulk fetch off the + ``activity_state="approval"`` transition to get the items.""" ui = _make_ui() - # Mirror the shape ``pause_for_approval`` writes (session_ui_base - # lines 576-580) — items with call_id + header is the minimum - # the serializer needs to project. ui._pending_approval = { "type": "approve_request", "items": [ @@ -209,21 +201,9 @@ class TestPendingApprovalDetailGate: events = _drain_global() attn = [e for e in events if e.get("state") == "attention"] assert len(attn) == 1 - # Field present and structurally sound — the serializer's - # full shape is covered by tests/test_session_ui_base.py; - # here we only need to confirm the gate fires and the - # serializer's output is what lands on the event. - assert "pending_approval_detail" in attn[0] - detail = attn[0]["pending_approval_detail"] - assert detail is not None - assert detail.get("items") - assert detail["items"][0]["call_id"] == "c1" + assert "pending_approval_detail" not in attn[0] - def test_field_cleared_after_approval_resolves(self): - """Once ``_pending_approval`` is cleared, subsequent state - broadcasts must drop the field again — without this, the - browser would render stale approve/deny buttons until the - next bulk-poll TTL window expired.""" + def test_field_stays_absent_after_approval_resolves(self): ui = _make_ui() ui._pending_approval = { "type": "approve_request", @@ -231,7 +211,7 @@ class TestPendingApprovalDetailGate: "judge_pending": False, } ui._broadcast_state("attention") - _drain_global() # discard the with-detail event + _drain_global() ui._pending_approval = None ui._broadcast_state("running") @@ -239,3 +219,115 @@ class TestPendingApprovalDetailGate: running = [e for e in events if e.get("state") == "running"] assert len(running) == 1 assert "pending_approval_detail" not in running[0] + + +class TestBroadcastIntentVerdict: + """Producer-side coverage for ``WebUI._broadcast_intent_verdict``. + + The collector-side test (``test_apply_delta_intent_verdict_*`` in + test_console.py) covers consumption; this pins the event shape the + producer puts on the global queue. A field rename or missed key + here would slip past the consumer test because the consumer reads + via ``data.get(...)``. + """ + + def test_pushes_intent_verdict_event_to_global_queue(self): + ui = _make_ui() + verdict = { + "call_id": "c1", + "risk_level": "low", + "confidence": 0.92, + "recommendation": "approve", + "reasoning": "tool reads only", + } + ui._broadcast_intent_verdict(verdict) + + events = _drain_global() + assert len(events) == 1 + ev = events[0] + assert ev["type"] == "intent_verdict" + assert ev["ws_id"] == "ws-test" + assert ev["verdict"] == verdict + + def test_no_op_when_global_queue_unset(self): + WebUI._global_queue = None + ui = _make_ui() + # Doesn't raise. + ui._broadcast_intent_verdict({"call_id": "c1"}) + + def test_queue_full_swallowed(self): + # Force a tiny queue then fill it so the next put_nowait + # raises queue.Full — the broadcast must absorb it without + # propagating (matches _broadcast_state's queue.Full handling). + WebUI._global_queue = queue.Queue(maxsize=1) + WebUI._global_queue.put_nowait({"sentinel": True}) + ui = _make_ui() + # Doesn't raise. + ui._broadcast_intent_verdict({"call_id": "c1"}) + + +class TestBroadcastApprovalResolved: + """Producer-side coverage for ``WebUI._broadcast_approval_resolved``.""" + + def test_pushes_approval_resolved_event_to_global_queue(self): + ui = _make_ui() + ui._broadcast_approval_resolved(True, "lgtm", always=False) + + events = _drain_global() + assert len(events) == 1 + ev = events[0] + assert ev["type"] == "approval_resolved" + assert ev["ws_id"] == "ws-test" + assert ev["approved"] is True + assert ev["feedback"] == "lgtm" + assert ev["always"] is False + + def test_normalises_none_feedback_to_empty_string(self): + ui = _make_ui() + ui._broadcast_approval_resolved(False, None) + + events = _drain_global() + assert events[0]["feedback"] == "" + assert events[0]["approved"] is False + assert events[0]["always"] is False + + def test_always_kwarg_propagates(self): + ui = _make_ui() + ui._broadcast_approval_resolved(True, "ok", always=True) + events = _drain_global() + assert events[0]["always"] is True + + def test_no_op_when_global_queue_unset(self): + WebUI._global_queue = None + ui = _make_ui() + # Doesn't raise. + ui._broadcast_approval_resolved(True, None) + + +class TestBroadcastApproveRequest: + """Producer-side coverage for ``WebUI._broadcast_approve_request`` — + push path for the initial approval items so a coord parent's tree + UI can render the inline approve/deny block immediately without + waiting for a bulk-fetch round-trip.""" + + def test_pushes_approve_request_event_to_global_queue(self): + ui = _make_ui() + detail = { + "type": "approve_request", + "items": [{"call_id": "c1", "header": "tool x"}], + "judge_pending": True, + } + ui._broadcast_approve_request(detail) + + events = _drain_global() + assert len(events) == 1 + ev = events[0] + assert ev["type"] == "approve_request" + assert ev["ws_id"] == "ws-test" + assert ev["detail"] == detail + + def test_no_op_when_global_queue_unset(self): + WebUI._global_queue = None + ui = _make_ui() + # Doesn't raise. + ui._broadcast_approve_request({"items": []}) diff --git a/turnstone/cli.py b/turnstone/cli.py index 702c901f..a8107ca8 100644 --- a/turnstone/cli.py +++ b/turnstone/cli.py @@ -1242,7 +1242,7 @@ def main() -> None: ) sys.stderr.flush() - manager._on_state_change = _bg_attention_notify + manager.subscribe_to_state(_bg_attention_notify) # Print banner print(f"\n{bold('Chat')} with {cyan(model)}") diff --git a/turnstone/console/collector.py b/turnstone/console/collector.py index 512e2798..e9e319ea 100644 --- a/turnstone/console/collector.py +++ b/turnstone/console/collector.py @@ -12,16 +12,22 @@ import asyncio import contextlib import json import logging -import queue import random import threading import time from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any +if TYPE_CHECKING: + import queue + import httpx import httpx_sse +from turnstone.core.session_ui_base import ( + _CRITICAL_EVENT_TYPES, + _put_with_priority, +) from turnstone.core.workstream import WorkstreamKind if TYPE_CHECKING: @@ -157,11 +163,19 @@ class ClusterCollector: log.info("ClusterCollector stopped") def _fanout(self, event: dict[str, Any]) -> None: - """Copy an event to all registered SSE listener queues.""" + """Copy an event to all registered SSE listener queues. + + Critical event types (verdicts, approval resolutions, child + lifecycle) evict one oldest queue entry to make room rather + than dropping themselves on a full queue. Best-effort events + (state ticks, activity) drop as before. See + :data:`turnstone.core.session_ui_base._CRITICAL_EVENT_TYPES` + for the canonical list. + """ + critical = event.get("type") in _CRITICAL_EVENT_TYPES with self._listeners_lock: for q in self._listeners: - with contextlib.suppress(queue.Full): - q.put_nowait(event) + _put_with_priority(q, event, critical=critical) # -- auth helpers -------------------------------------------------------- @@ -498,7 +512,6 @@ class ClusterCollector: "kind": WorkstreamKind.from_raw(new_w.get("kind")), "parent_ws_id": new_w.get("parent_ws_id"), "activity_state": new_w.get("activity_state", ""), - "pending_approval_detail": new_w.get("pending_approval_detail"), } ) old_name = old_ws.get("title", "") or old_ws.get("name", "") @@ -558,18 +571,6 @@ class ClusterCollector: ws["kind"] = data["kind"] if "parent_ws_id" in data: ws["parent_ws_id"] = data["parent_ws_id"] - # ``pending_approval_detail`` overwrites (no - # ``ws.get`` fallback): the node's broadcast gate - # on ``_pending_approval is not None`` means the - # field is absent from ``data`` exactly when no - # approval is pending — falling back to the cached - # value would resurrect a stale detail after the - # approval resolved. Without this assignment the - # cached ``node.workstreams`` dict served by - # ``get_node_detail`` / ``get_snapshot`` between - # reconciliations would render stale approve/deny - # buttons on closed approvals. - ws["pending_approval_detail"] = data.get("pending_approval_detail") pending_events.append( { "type": "cluster_state", @@ -581,7 +582,6 @@ class ClusterCollector: "kind": WorkstreamKind.from_raw(ws.get("kind")), "parent_ws_id": ws.get("parent_ws_id"), "activity_state": ws.get("activity_state", ""), - "pending_approval_detail": data.get("pending_approval_detail"), } ) @@ -648,6 +648,60 @@ class ClusterCollector: ws["name"] = name pending_events.append({"type": "ws_rename", "ws_id": ws_id, "name": name}) + elif etype == "intent_verdict": + # Pure pass-through for LLM judge verdicts. The node's + # WebUI._broadcast_intent_verdict pushes onto the + # global queue (Stage 3 Step 5); the cluster fan-out + # forwards to subscribers (e.g. CoordinatorAdapter, + # which dispatches to the parent's coord SSE as + # ``child_ws_intent_verdict``). + ws_id = data.get("ws_id", "") + if ws_id: + pending_events.append( + { + "type": "intent_verdict", + "ws_id": ws_id, + "node_id": node_id, + "verdict": data.get("verdict") or {}, + } + ) + + elif etype == "approval_resolved": + # Pure pass-through for approve/deny resolutions. Pairs + # with ``intent_verdict`` above so the parent + # coordinator's tree UI can clear the pending-approval + # pill the moment the user decides, rather than waiting + # for the subsequent state-change piggyback. + ws_id = data.get("ws_id", "") + if ws_id: + pending_events.append( + { + "type": "approval_resolved", + "ws_id": ws_id, + "node_id": node_id, + "approved": bool(data.get("approved", False)), + "feedback": data.get("feedback", "") or "", + "always": bool(data.get("always", False)), + } + ) + + elif etype == "approve_request": + # Push path for the initial approval items. Eliminates + # the bulk-fetch race that otherwise leaves the coord + # tree stuck on a loading placeholder when the bulk + # fetch lands in the gap between the state transition + # to ATTENTION and ``_pending_approval`` being set. + ws_id = data.get("ws_id", "") + if ws_id: + pending_events.append( + { + "type": "approve_request", + "ws_id": ws_id, + "node_id": node_id, + "detail": data.get("detail") or {}, + } + ) + elif etype == "health_changed": # Update the health dict's backend status in-place bstatus = data.get("backend_status", "") @@ -1217,3 +1271,70 @@ class ClusterCollector: return entry["name"] = name self._fanout({"type": "ws_rename", "ws_id": ws_id, "name": name}) + + def emit_console_ws_intent_verdict(self, ws_id: str, verdict: dict[str, Any]) -> None: + """Fan an LLM intent-judge verdict for a console-pseudo-node ws. + + Gives coord-spawned approval flows a first-class cluster-bus + event so the parent's tree UI can render the risk pill + + verdict result without polling. + ``CoordinatorAdapter._dispatch_child_event`` re-emits these as + ``child_ws_intent_verdict`` for the parent coordinator's SSE + stream. The dispatch path filters by registry membership; + downstream subscribers tolerate verdicts for ws_ids they don't + own (silently drop), so we skip the membership pre-check that + would otherwise add a lock acquisition per emit on a path + that fires once per tool-call during heuristic+LLM judging. + """ + self._fanout( + { + "type": "intent_verdict", + "ws_id": ws_id, + "node_id": self.CONSOLE_PSEUDO_NODE_ID, + "verdict": verdict, + } + ) + + def emit_console_ws_approval_resolved( + self, + ws_id: str, + *, + approved: bool, + feedback: str = "", + always: bool = False, + ) -> None: + """Fan an ``approval_resolved`` decision for a console-pseudo-node ws. + + Paired with :meth:`emit_console_ws_intent_verdict` so the + coord tree UI clears the pending-approval pill in lockstep + with the actual decision. Same lock-skip rationale as the + intent-verdict emit above. + """ + self._fanout( + { + "type": "approval_resolved", + "ws_id": ws_id, + "node_id": self.CONSOLE_PSEUDO_NODE_ID, + "approved": approved, + "feedback": feedback, + "always": always, + } + ) + + def emit_console_ws_approve_request(self, ws_id: str, detail: dict[str, Any]) -> None: + """Fan an ``approve_request`` payload for a console-pseudo-node ws. + + Push path for the initial approval items so a coord parent's + tree UI can render the inline approve/deny block immediately + without a bulk-fetch round-trip. ``CoordinatorAdapter._dispatch_child_event`` + re-emits as ``child_ws_approve_request`` for the parent + coordinator's SSE stream. + """ + self._fanout( + { + "type": "approve_request", + "ws_id": ws_id, + "node_id": self.CONSOLE_PSEUDO_NODE_ID, + "detail": detail, + } + ) diff --git a/turnstone/console/coordinator_adapter.py b/turnstone/console/coordinator_adapter.py index 204f6d9f..7604daa0 100644 --- a/turnstone/console/coordinator_adapter.py +++ b/turnstone/console/coordinator_adapter.py @@ -15,17 +15,17 @@ for the storage-seeded children rebuild. from __future__ import annotations -import queue -import threading from typing import TYPE_CHECKING, Any from turnstone.core import session_worker from turnstone.core.adapters._ui_cleanup import cleanup_session_ui +from turnstone.core.child_source import ClusterChildSource +from turnstone.core.children_registry import ChildrenRegistry from turnstone.core.log import get_logger from turnstone.core.workstream import Workstream, WorkstreamKind, WorkstreamState if TYPE_CHECKING: - from collections.abc import Callable, Iterable + from collections.abc import Callable from turnstone.console.collector import ClusterCollector from turnstone.console.coordinator_ui import ConsoleCoordinatorUI @@ -56,38 +56,25 @@ class CoordinatorAdapter: # the manager to the adapter's ``__init__`` — break the cycle # with a setter called from console startup. self._manager: SessionManager | None = None - # Per-coordinator known-child ws_id set. Populated lazily on - # create/open from storage and updated live as the cluster fan-out - # thread sees ws_created events with matching parent_ws_id. - # Closed / deleted children stay in the registry so the tree UI - # can keep rendering them grayed out; the authoritative render - # path reads storage for state. Bounded by eventual coordinator - # close/eviction. - self._children: dict[str, set[str]] = {} - # Reverse index for O(1) child → coord lookup on every cluster - # event. Without this, every cluster event incurs a linear scan - # over every coordinator's child set while holding the fan-out - # lock — a hot-path tax that scales with both active coordinators - # and their retained-history depth. - self._child_to_coord: dict[str, str] = {} - self._children_lock = threading.Lock() - # Cluster-event fan-out: subscribes to the ClusterCollector's - # listener channel, filters by known child ws_ids, and re-emits - # child_ws_* events on the matching coordinator's UI. Configured - # lazily via ``start_child_event_fanout(collector)`` from the - # console lifespan once both the manager and collector exist. - self._collector_queue: queue.Queue[dict[str, Any]] | None = None - self._fanout_thread: threading.Thread | None = None - self._fanout_stop = threading.Event() - # Coord-ws-id → UI map for the fan-out dispatch path. Read - # and written under ``self._children_lock`` alongside the - # forward/reverse child maps. (Previously the value was - # ``(user_id, ui)`` with a copy-on-write dict swap so the - # dispatch could read it lock-free — but _dispatch_child_event - # already re-validates the parent under _children_lock anyway, - # so the lock-free snapshot was premature. The user_id half is - # also dead after a46dab1 dropped row-level ownership gates.) - self._active_coords: dict[str, Any] = {} + # Children registry — universal parent → children + reverse + # lookup primitive. Lifted from inline data on this adapter to + # ``turnstone.core.children_registry.ChildrenRegistry`` (Stage 3 + # Step 1) so the same primitive can serve interactive + # workstreams when they gain spawn capability. The dispatch + # path (`_dispatch_child_event`) calls into the registry + # atomically; everything else delegates through the legacy + # method names which still exist as thin shims for the + # cluster-routing + cleanup callers. + self._registry = ChildrenRegistry() + # Cross-node child events arrive via ``ClusterChildSource`` + # (Stage 3 Step 2): a strategy that subscribes to the + # collector's listener channel and runs a daemon thread that + # drains the queue, pushing each event to the sink. The sink + # is :meth:`_dispatch_child_event` so the existing translation + # logic (cluster_state → child_ws_state, etc.) stays in one + # place. Constructed lazily by ``start_child_event_fanout`` so + # the collector reference is available. + self._child_source: ClusterChildSource | None = None def attach(self, manager: SessionManager) -> None: """Late-bind the owning :class:`SessionManager`. @@ -109,7 +96,7 @@ class CoordinatorAdapter: # needs the empty forward/presence entries so # ``_dispatch_child_event`` recognises this coordinator when # its first child is spawned. - self._install_coord_registry(ws) + self._registry.install(ws.id, ws.ui) self._fanout_console_ws_created(ws) def emit_rehydrated(self, ws: Workstream) -> None: @@ -117,21 +104,10 @@ class CoordinatorAdapter: # it from storage after the registry seed so a ``ws_created`` # for an already-spawned child that fires mid-rebuild merges # cleanly. - self._install_coord_registry(ws) + self._registry.install(ws.id, ws.ui) self._rebuild_children_registry(ws.id) self._fanout_console_ws_created(ws) - def _install_coord_registry(self, ws: Workstream) -> None: - """Seed the children registry + presence map for ``ws``. - - Shared by ``emit_created`` and ``emit_rehydrated`` — the - difference between the two is purely whether we then rebuild - from storage. - """ - with self._children_lock: - self._children.setdefault(ws.id, set()) - self._active_coords[ws.id] = ws.ui - def _fanout_console_ws_created(self, ws: Workstream) -> None: try: self._collector.emit_console_ws_created( @@ -197,14 +173,11 @@ class CoordinatorAdapter: # toast only fire for real-node (interactive) ws_closed events. del reason, name # Drop the coordinator's children-registry entries AND its - # presence slot. Mirrors the eviction/close paths from the old - # CoordinatorManager (which did the same under _children_lock - # + _lock respectively). A plain _children.pop without clearing - # the reverse index would leak every evicted coordinator's - # child→parent pointers forever. - with self._children_lock: - self._pop_coord_registry_locked(ws_id) - self._active_coords.pop(ws_id, None) + # presence slot. A plain pop without clearing the reverse + # index would leak every evicted coordinator's child→parent + # pointers forever — :meth:`ChildrenRegistry.uninstall` + # handles forward set + reverse entries + presence atomically. + self._registry.uninstall(ws_id) try: self._collector.emit_console_ws_closed(ws_id) except Exception: @@ -356,25 +329,6 @@ class CoordinatorAdapter: # Children registry # ------------------------------------------------------------------ - def _merge_child_ids_locked(self, coord_ws_id: str, child_ids: Iterable[str]) -> None: - """Merge ``child_ids`` into ``coord_ws_id``'s forward + reverse maps. - - Caller MUST hold ``self._children_lock``. Idempotent — re-adding - an existing child is a no-op (the reverse-index pointer is - already correct). Empty / falsy entries in ``child_ids`` are - skipped. - - Sole write-path for bulk registry updates so - ``_rebuild_children_registry`` (storage-seeded) and - ``_prime_children_from_snapshot`` (collector-seeded) agree on - ordering and reverse-index invariants. - """ - existing = self._children.setdefault(coord_ws_id, set()) - for cid in child_ids: - if cid and cid not in existing: - existing.add(cid) - self._child_to_coord[cid] = coord_ws_id - def _rebuild_children_registry(self, coord_ws_id: str) -> None: """Populate ``self._children[coord_ws_id]`` from storage. @@ -445,174 +399,88 @@ class CoordinatorAdapter: if not child_id: continue child_ids.append(child_id) - with self._children_lock: - self._merge_child_ids_locked(coord_ws_id, child_ids) - - def _coord_for_child(self, child_ws_id: str) -> str | None: - """Reverse-lookup: which coordinator owns this child ws_id? - - O(1) via the ``_child_to_coord`` reverse index. Cluster events - fire on every token tick across the cluster; a linear scan here - turned into a hot-path tax as the retained-history set grew. - """ - with self._children_lock: - return self._child_to_coord.get(child_ws_id) + self._registry.merge_children(coord_ws_id, child_ids) def children_snapshot(self, coord_ws_id: str) -> list[str]: """Return a snapshot of the coordinator's direct child ws_ids. - Used by ``stop_cascade`` to iterate children without holding the - registry lock during the per-child HTTP dispatch. A mutation - racing with the snapshot (child spawned mid-cascade) either - lands before the snapshot and gets cancelled, or lands after - and is out of scope for this batch — both outcomes are safe. - Returns an empty list for unknown coordinators. + Used by ``stop_cascade`` to iterate children without holding + the registry lock during the per-child HTTP dispatch. A + mutation racing with the snapshot (child spawned mid-cascade) + either lands before (cancelled) or after (out of scope for + this batch) — both safe. Returns an empty list for unknown + coordinators. """ - with self._children_lock: - child_set = self._children.get(coord_ws_id) - return list(child_set) if child_set else [] - - def _pop_coord_registry_locked(self, coord_ws_id: str) -> None: - """Remove a coordinator's forward set + reverse-index entries. - - Caller MUST hold ``self._children_lock``. Used by close / - eviction paths so stale coordinators don't leak registry - entries. No-op if the coordinator is unknown. - """ - child_set = self._children.pop(coord_ws_id, None) - if child_set is None: - return - for cid in child_set: - # Defensive: only clear the reverse entry if it still points - # at THIS coordinator. If a child has since been reassigned - # (unusual but possible on schema changes), we don't want to - # orphan the new owner's entry. - if self._child_to_coord.get(cid) == coord_ws_id: - self._child_to_coord.pop(cid, None) + return self._registry.children_of(coord_ws_id) # ------------------------------------------------------------------ # Cluster-event fan-out thread # ------------------------------------------------------------------ def start_child_event_fanout(self, collector: ClusterCollector) -> None: - """Subscribe to cluster events and start the filter + re-emit thread. + """Subscribe to cluster events via :class:`ClusterChildSource`. - Idempotent — calling twice is a no-op (already-started fan-out - thread stays). Called once from the console lifespan after both - the collector and the session manager are constructed. + Idempotent — already-started ChildSource stays. Called once + from the console lifespan after both the collector and the + session manager are constructed. """ - if self._fanout_thread is not None and self._fanout_thread.is_alive(): + if self._child_source is not None: return self._collector = collector - self._collector_queue = queue.Queue(maxsize=1000) - # Ensure the "console" pseudo-node exists in the snapshot map so - # emit_console_ws_* calls from create / close / open land on a - # real node entry the snapshot will surface. + # Coord-specific transport setup: the "console" pseudo-node + # must exist in the snapshot map BEFORE any + # ``emit_console_ws_*`` calls (and before the snapshot is + # taken inside ``ChildSource.start``) so those emits land on + # a real node entry the snapshot surfaces. collector.ensure_console_pseudo_node() - # Register with the collector — use the existing listener channel - # the browser SSE fan-out uses; the collector treats our queue as - # just another subscriber. - snapshot = collector.get_snapshot_and_register(self._collector_queue) - # Prime the child registry from the snapshot so a coordinator - # that opens right after a console restart sees already-live - # children without waiting for the next ``ws_state`` tick to - # discover them via the fan-out path. - self._prime_children_from_snapshot(snapshot) - # Seed the pseudo-node with any coordinators already loaded in - # memory when the collector binds. Prevents a race where early - # creates happened before the collector was wired up and their - # rows never showed on the snapshot. - mgr = self._manager - if mgr is not None: - for ws in mgr.list_all(): - try: - collector.emit_console_ws_created( - ws.id, - name=ws.name, - user_id=ws.user_id or "", - kind=WorkstreamKind.COORDINATOR.value, - state=ws.state.value, - parent_ws_id=None, - ) - except Exception: - log.debug( - "coord_adapter.collector_seed_failed ws=%s", - ws.id[:8], - exc_info=True, - ) - self._fanout_stop.clear() - t = threading.Thread( - target=self._fanout_loop, - name="coord-adapter-child-fanout", - daemon=True, - ) - self._fanout_thread = t - t.start() - - def _prime_children_from_snapshot(self, snapshot: dict[str, Any]) -> None: - """Populate ``_children`` + ``_child_to_coord`` from a collector snapshot. - - The snapshot's per-node workstreams carry ``parent_ws_id``. For - every workstream whose parent is an in-memory coordinator, - record the child so the fan-out filter sees it immediately. - """ - nodes = snapshot.get("nodes", []) if isinstance(snapshot, dict) else [] - if not nodes: - return mgr = self._manager if mgr is None: raise RuntimeError( "CoordinatorAdapter: manager not attached — call attach(mgr) after construction" ) - by_parent: dict[str, list[str]] = {} - known = {ws.id for ws in mgr.list_all()} - for node in nodes: - for entry in node.get("workstreams", []) or []: - parent = entry.get("parent_ws_id") or "" - child_id = entry.get("id") or "" - if not parent or not child_id or parent not in known: - continue - by_parent.setdefault(parent, []).append(child_id) - if not by_parent: - return - with self._children_lock: - for parent, kids in by_parent.items(): - self._merge_child_ids_locked(parent, kids) + # Build + start the strategy. The sink is + # :meth:`_dispatch_child_event` so cluster events flow through + # the same translation path that synthesises ``child_ws_*`` + # payloads for the parent's UI. + source = ClusterChildSource( + collector=collector, + registry=self._registry, + parents_provider=lambda: [ws.id for ws in mgr.list_all()], + ) + source.start(sink=self._dispatch_child_event) + self._child_source = source + # Seed the pseudo-node with any coordinators already loaded in + # memory when the collector binds. Prevents a race where early + # creates happened before the collector was wired up and their + # rows never showed on the snapshot. (Coord-specific — interactive + # has no analogous pseudo-node.) + for ws in mgr.list_all(): + try: + collector.emit_console_ws_created( + ws.id, + name=ws.name, + user_id=ws.user_id or "", + kind=WorkstreamKind.COORDINATOR.value, + state=ws.state.value, + parent_ws_id=None, + ) + except Exception: + log.debug( + "coord_adapter.collector_seed_failed ws=%s", + ws.id[:8], + exc_info=True, + ) def shutdown(self) -> None: - """Stop the fan-out thread and unregister from the collector. + """Stop the ChildSource and unregister from the collector. Safe to call multiple times; idempotent. Invoked from the console lifespan teardown so SSE listener queues don't leak. """ - self._fanout_stop.set() - t = self._fanout_thread - q = self._collector_queue - coll = self._collector - self._fanout_thread = None - self._collector_queue = None - if coll is not None and q is not None: - try: - coll.unregister_listener(q) - except Exception: - log.debug("coord_adapter.unregister_listener_failed", exc_info=True) - if t is not None: - t.join(timeout=2.0) - - def _fanout_loop(self) -> None: - """Drain collector events, filter by known children, dispatch.""" - q = self._collector_queue - if q is None: - return - while not self._fanout_stop.is_set(): - try: - event = q.get(timeout=1.0) - except queue.Empty: - continue - try: - self._dispatch_child_event(event) - except Exception: - log.debug("coord_adapter.fanout.dispatch_failed", exc_info=True) + src = self._child_source + self._child_source = None + if src is not None: + src.shutdown() def _dispatch_child_event(self, event: dict[str, Any]) -> None: """Match a cluster event to a coordinator and re-emit on its UI. @@ -622,10 +490,12 @@ class CoordinatorAdapter: - ``ws_created`` with ``parent_ws_id`` matching an in-memory coordinator → add to registry + re-emit as ``child_ws_created``. - - ``cluster_state`` / ``ws_closed`` / ``ws_rename`` whose - ``ws_id`` is in any coordinator's known-children registry → - re-emit as ``child_ws_state`` / ``child_ws_closed`` / - ``child_ws_rename``. + - ``cluster_state`` / ``ws_closed`` / ``ws_rename`` / + ``intent_verdict`` / ``approval_resolved`` whose ``ws_id`` + is in any coordinator's known-children registry → re-emit as + ``child_ws_state`` / ``child_ws_closed`` / + ``child_ws_rename`` / ``child_ws_intent_verdict`` / + ``child_ws_approval_resolved``. Events for ws_ids we don't own silently drop — the filter lives on the server so each coordinator's SSE stream stays small. @@ -639,23 +509,17 @@ class CoordinatorAdapter: parent = event.get("parent_ws_id") or "" if not parent: return - # Presence check + registry mutation under the same lock: - # a concurrent close()/eviction can pop the entry between - # the check and the mutation, after which a bare setdefault - # would resurrect the entry — leaking the registry key and - # enqueuing onto the closed coordinator's UI. Trusted-team - # posture (#400 / a46dab1) means no per-event tenant gate - # here; scope-level auth at the SSE endpoint is the only - # boundary. - with self._children_lock: - coord_ui = self._active_coords.get(parent) - if coord_ui is None: - return - existing = self._children.setdefault(parent, set()) - if ws_id in existing: - return - existing.add(ws_id) - self._child_to_coord[ws_id] = parent + # Atomic check-and-route under the registry's lock: a + # concurrent close()/eviction can pop the parent's entry + # between the presence check and the mutation, so the two + # must happen together. ``add_child`` returns the parent's + # UI on success or None on (a) parent not installed, or + # (b) duplicate child. Trusted-team posture (#400 / a46dab1) + # means no per-event tenant gate here; scope-level auth at + # the SSE endpoint is the only boundary. + coord_ui = self._registry.add_child(parent, ws_id) + if coord_ui is None: + return payload = { "type": "child_ws_created", "ws_id": ws_id, @@ -668,8 +532,15 @@ class CoordinatorAdapter: _enqueue_on_ui(coord_ui, parent, payload) return - if etype in ("cluster_state", "ws_closed", "ws_rename"): - coord_id = self._coord_for_child(ws_id) + if etype in ( + "cluster_state", + "ws_closed", + "ws_rename", + "intent_verdict", + "approval_resolved", + "approve_request", + ): + coord_id = self._registry.parent_for(ws_id) if coord_id is None: return mgr = self._manager @@ -684,14 +555,13 @@ class CoordinatorAdapter: "state": event.get("state", ""), "tokens": event.get("tokens", 0), "node_id": event.get("node_id", ""), - # activity_state lets the JS detect approval-state - # transitions; pending_approval_detail rides on - # the same event so the browser can mutate - # liveBadgeCache directly and render inline - # approve/deny buttons in lockstep with the - # transition, no separate dashboard refetch. + # ``activity_state`` lets the JS detect approval + # transitions and trigger a bulk fetch for the + # initial detail. The detail itself no longer + # piggybacks here (Stage 3 cleanup) — verdicts + # arrive via the explicit ``intent_verdict`` event + # class and resolution via ``approval_resolved``. "activity_state": event.get("activity_state", ""), - "pending_approval_detail": event.get("pending_approval_detail"), } elif etype == "ws_closed": child_event = { @@ -700,13 +570,52 @@ class CoordinatorAdapter: "parent_ws_id": coord_id, "reason": event.get("reason", ""), } - else: # ws_rename + elif etype == "ws_rename": child_event = { "type": "child_ws_rename", "child_ws_id": ws_id, "parent_ws_id": coord_id, "name": event.get("name", ""), } + elif etype == "intent_verdict": + # Per-coord re-emit of an explicit verdict event so + # the tree UI can render the risk pill + verdict + # result without polling. The corresponding + # ``cluster_state`` event no longer carries the + # detail (the piggyback was removed end-to-end); + # the bulk fetch on initial approval entry plus this + # explicit event class are the only carriers. + child_event = { + "type": "child_ws_intent_verdict", + "child_ws_id": ws_id, + "parent_ws_id": coord_id, + "node_id": event.get("node_id", ""), + "verdict": event.get("verdict") or {}, + } + elif etype == "approval_resolved": + child_event = { + "type": "child_ws_approval_resolved", + "child_ws_id": ws_id, + "parent_ws_id": coord_id, + "node_id": event.get("node_id", ""), + "approved": bool(event.get("approved", False)), + "feedback": event.get("feedback", "") or "", + "always": bool(event.get("always", False)), + } + else: # approve_request + # Push path for the initial approval items — + # eliminates the bulk-fetch race that previously left + # the coord row stuck on a loading placeholder when + # the bulk fetch landed in the gap between the state + # transition to ATTENTION and ``_pending_approval`` + # being set inside ``approve_tools``. + child_event = { + "type": "child_ws_approve_request", + "child_ws_id": ws_id, + "parent_ws_id": coord_id, + "node_id": event.get("node_id", ""), + "detail": event.get("detail") or {}, + } _enqueue_on_ui(owning_ws.ui, coord_id, child_event) diff --git a/turnstone/console/coordinator_ui.py b/turnstone/console/coordinator_ui.py index 44f1496d..11164b60 100644 --- a/turnstone/console/coordinator_ui.py +++ b/turnstone/console/coordinator_ui.py @@ -181,6 +181,81 @@ class ConsoleCoordinatorUI(SessionUIBase): with self._ws_lock: self._last_broadcast_activity = current + def _broadcast_intent_verdict(self, verdict: dict[str, Any]) -> None: + """Fan an LLM intent-judge verdict to the cluster collector. + + Stage 3 Step 5 — overrides the no-op base hook so a coord + workstream that produces its own verdict (rare — coord agents + don't typically run the LLM judge) gets a first-class + cluster-bus event for the parent's tree UI. The far more + common path is a CHILD workstream firing its verdict on a + real node; that path goes through ``WebUI._broadcast_intent_verdict`` + → global SSE → collector ``_apply_delta``. + """ + collector = ConsoleCoordinatorUI._collector + if collector is None: + return + try: + collector.emit_console_ws_intent_verdict(self.ws_id, verdict) + except Exception: + log.debug( + "coord_ui.intent_verdict_fanout_failed ws=%s", + self.ws_id, + exc_info=True, + ) + + def _broadcast_approval_resolved( + self, + approved: bool, + feedback: str | None = None, + *, + always: bool = False, + ) -> None: + """Fan an ``approval_resolved`` decision to the cluster collector. + + Stage 3 Step 5 — paired with :meth:`_broadcast_intent_verdict`. + Same rationale: coord-direct approvals are rare; the typical + path is a child workstream resolving on its node, with that + node's WebUI broadcasting through the global queue. + """ + collector = ConsoleCoordinatorUI._collector + if collector is None: + return + try: + collector.emit_console_ws_approval_resolved( + self.ws_id, + approved=approved, + feedback=feedback or "", + always=always, + ) + except Exception: + log.debug( + "coord_ui.approval_resolved_fanout_failed ws=%s", + self.ws_id, + exc_info=True, + ) + + def _broadcast_approve_request(self, detail: dict[str, Any]) -> None: + """Fan an ``approve_request`` payload to the cluster collector. + + Push path for the initial approval items. Same rationale as + the other two broadcast hooks: coord-self approvals are rare + (the LLM judge isn't wired on the console coord today), but + the override exists for symmetry and lights up the same path + a future coord-self gate would use. + """ + collector = ConsoleCoordinatorUI._collector + if collector is None: + return + try: + collector.emit_console_ws_approve_request(self.ws_id, detail) + except Exception: + log.debug( + "coord_ui.approve_request_fanout_failed ws=%s", + self.ws_id, + exc_info=True, + ) + def on_state_change(self, state: str) -> None: # Flow state transitions through the unified SessionManager so # the storage write + adapter emit_state fan-out stay in lockstep diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 00c8d5ab..308087f5 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -846,8 +846,20 @@ async def _fetch_live_block( live = {k: entry.get(k) for k in _CLUSTER_WS_LIVE_KEYS if k in entry} # Derived field — kept in lockstep with # _coordinator_live_snapshot so both origins produce the - # same keys. - live["pending_approval"] = live.get("activity_state") == "approval" + # same keys. ``state="attention"`` is the canonical signal; + # ``activity_state="approval"`` is set inside approve_tools + # AFTER the state transition fires, so a bulk fetch that + # races with that window can see state=attention and + # activity_state="" simultaneously. A non-null + # ``pending_approval_detail`` is also a definitive signal + # (the serializer only emits non-None when ``_pending_approval`` + # is set on the UI). Any of the three flips this true; the + # frontend reducer mirrors the same disjunction. + live["pending_approval"] = ( + live.get("activity_state") == "approval" + or entry.get("state") == "attention" + or live.get("pending_approval_detail") is not None + ) return live return None diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js index c3560503..c287d108 100644 --- a/turnstone/console/static/coordinator/coordinator.js +++ b/turnstone/console/static/coordinator/coordinator.js @@ -1730,7 +1730,7 @@ // state) so a user lacking admin.cluster.inspect doesn't // pay one 403 per denied id on every reconnect. for (const [id, c] of liveBadgeCache) { - if (!c || !c.permanent) liveBadgeCache.delete(id); + if (!c || !c.permanent) _liveBadgeCacheDelete(id); } } }; @@ -2038,6 +2038,21 @@ case "child_ws_rename": handleChildRename(ev); break; + // Stage 3 Step 7 — explicit verdict + approval-resolved events + // arrive without piggybacking on cluster_state, so verdicts that + // land while a child is in `attention` (no state transition) + // reach the parent's tree UI promptly. Reducer writes + // liveBadgeCache directly (bypassing scheduleLiveFetch) so + // off-screen rows update too. + case "child_ws_intent_verdict": + handleChildIntentVerdict(ev); + break; + case "child_ws_approval_resolved": + handleChildApprovalResolved(ev); + break; + case "child_ws_approve_request": + handleChildApproveRequest(ev); + break; // wait_for_workstream observability (#14) — the worker thread // can block up to 600s inside the tool; these events drive a // sidebar indicator so operators see the coordinator is alive. @@ -2050,12 +2065,40 @@ case "wait_ended": handleWaitEnded(ev); break; + // Server-side SSE queue depth — debug aid surfaced in the + // status bar so operators can spot a backed-up tab (slow + // consumer / browser throttling) without devtools. + case "_queue_stats": + handleQueueStats(ev); + break; default: // Unknown event type — ignore silently. break; } } + // ------------------------------------------------------------------ + // SSE queue depth indicator + // ------------------------------------------------------------------ + // + // The server emits a periodic ``_queue_stats`` event carrying the + // per-tab listener queue depth. Surface in the coord status bar + // with color escalation (default → warn at >50% → danger at >80%) + // so a backed-up consumer is glanceable. The element starts + // hidden; first sample reveals it. + function handleQueueStats(ev) { + const el = document.getElementById("coord-sb-sse-queue"); + if (!el) return; + const depth = typeof ev.depth === "number" ? ev.depth : 0; + const max = typeof ev.max === "number" && ev.max > 0 ? ev.max : 500; + el.hidden = false; + el.textContent = "queue " + depth + "/" + max; + el.classList.remove("warn", "danger"); + const ratio = depth / max; + if (ratio > 0.8) el.classList.add("danger"); + else if (ratio > 0.5) el.classList.add("warn"); + } + // ------------------------------------------------------------------ // wait_for_workstream progress indicator (#14) // ------------------------------------------------------------------ @@ -2154,6 +2197,37 @@ const childrenState = new Map(); // ws_id -> {live: , fetched: } for the 5s TTL live-badge cache. const liveBadgeCache = new Map(); + // Incrementally-maintained set of ws_ids currently flagged + // ``pending_approval`` in the cache. Updated at every cache mutation + // via ``_liveBadgeCacheSet`` / ``_liveBadgeCacheDelete`` / + // ``_liveBadgeCacheClear`` below so the sidebar pending-count read + // is O(1) instead of an O(N) walk over the cache on every render + // (caught by /review perf-1). + const pendingApprovalIds = new Set(); + // The helpers reach into the Map's prototype directly so a future + // edit can't accidentally rewrite the body's ``liveBadgeCache.set`` + // into the helper name and create infinite recursion (already + // happened once when bulk-replacing call sites — caught at runtime + // as "InternalError: too much recursion"). + const _mapSet = Map.prototype.set; + const _mapDelete = Map.prototype.delete; + const _mapClear = Map.prototype.clear; + function _liveBadgeCacheSet(id, entry) { + _mapSet.call(liveBadgeCache, id, entry); + if (entry && entry.live && entry.live.pending_approval) { + pendingApprovalIds.add(id); + } else { + pendingApprovalIds.delete(id); + } + } + function _liveBadgeCacheDelete(id) { + _mapDelete.call(liveBadgeCache, id); + pendingApprovalIds.delete(id); + } + function _liveBadgeCacheClear() { + _mapClear.call(liveBadgeCache); + pendingApprovalIds.clear(); + } // ws_ids currently visible in the viewport — only these trigger // live-fetch on SSE state changes. Populated by an // IntersectionObserver attached to each rendered .ch-row so a @@ -2280,23 +2354,22 @@ } } row.appendChild(meta); - // Inline approve/deny block \u2014 shown only when the live block - // carries pending_approval_detail (the rich payload added by the - // server-side dashboard projection). A "\u2691 approval" badge alone - // means the child is in attention state but the rich detail - // hasn't arrived on the cache yet \u2014 a rare cross-version race - // (e.g. a node mid-rolling-upgrade emitted ws_state without - // pending_approval_detail before this PR landed). The next SSE - // tick or the 5s TTL bulk-poll catches up and re-renders. - if (cached && cached.live && cached.live.pending_approval_detail) { + // Inline approve/deny block \u2014 the detail arrives via the bulk + // fetch triggered by handleChildState off the activity_state="approval" + // transition. Verdicts that land later arrive via + // child_ws_intent_verdict and update the cached detail in place; + // resolution arrives via child_ws_approval_resolved and clears it. + // While the bulk fetch is in flight (~250ms debounce + ~100ms HTTP) + // we render a loading placeholder so the row keeps its height + // stable AND so a screen reader has a labelled region to land on + // \u2014 without it the operator sees a "demand for action" badge + // with no actionable content. + if (cached && cached.live && cached.live.pending_approval) { const detail = cached.live.pending_approval_detail; - const block = renderApprovalBlock(child, detail); + const block = detail + ? renderApprovalBlock(child, detail) + : renderApprovalPlaceholder(child); if (block) row.appendChild(block); - // Late-arriving LLM judge: see comment above - // _maybeStartJudgePoll for the why. Hooks into a single - // global poller (not per-row) so off-screen rows still - // refresh and one bulk request covers every pending row. - _maybeStartJudgePoll(); } // Recent auto-approves — tools that bypassed the operator gate // (skill ``allowed_tools`` allowlist / blanket / admin policy / @@ -2370,78 +2443,6 @@ return pill; } - // Late-judge polling — single global timer driving a bulk - // re-fetch of every row whose `pending_approval_detail.judge_pending` - // is true and not every item has a `judge_verdict` yet. Necessary - // because the LLM judge runs async on the child node and never - // pushes a signal that reaches the coord; without polling the - // operator would stare at a "heuristic"-tier pill forever. - // - // Why a global poller instead of per-row: - // 1. Per-row would invalidate the cache and call scheduleLiveFetch - // for off-screen rows, but scheduleLiveFetch returns early on - // non-visible ids — leaving the cache empty AND no fetch in - // flight. Result: off-screen rows stuck on stale heuristic. - // 2. A single bulk request covers every pending row in one - // round-trip; the per-row design would issue N microtask- - // batched fetches that share the bulk path anyway. - const JUDGE_POLL_INTERVAL_MS = 2000; - // Cap by total wall-clock time, not attempts. Real LLM judges - // (esp. with reasoning effort) can exceed 30s — a 6-attempt / - // 12s cap was prematurely giving up. - const JUDGE_POLL_MAX_DURATION_MS = 90_000; - let judgePollTimer = null; - let judgePollStartedAt = 0; - - function _maybeStartJudgePoll() { - if (judgePollTimer !== null) return; // already polling - judgePollStartedAt = Date.now(); - judgePollTimer = setTimeout(_judgePollTick, JUDGE_POLL_INTERVAL_MS); - } - - function _judgePollTick() { - judgePollTimer = null; - if (Date.now() - judgePollStartedAt > JUDGE_POLL_MAX_DURATION_MS) { - // Failed / timed-out judge — give up so we don't poll forever. - // Operator can hit the Refresh button to force a fresh fetch. - return; - } - // Walk the full childrenState — not just visible — so off-screen - // rows still get refreshed. scheduleLiveFetch's visibility gate - // exists to keep idle rows from burning round-trips; here we - // explicitly want every pending-judge row in the next bulk - // regardless of viewport. - let stillPending = false; - for (const [wsId, entry] of childrenState) { - if (TERMINAL_CHILD_STATES.has(entry.state)) continue; - const cached = liveBadgeCache.get(wsId); - if (!cached || !cached.live) continue; - const detail = cached.live.pending_approval_detail; - if (!detail || !detail.judge_pending) continue; - const items = Array.isArray(detail.items) ? detail.items : []; - const allHaveJudge = - items.length > 0 && items.every((it) => it.judge_verdict); - if (allHaveJudge) continue; - // Bypass scheduleLiveFetch entirely (skips visibility + TTL - // gates) by adding to pendingLiveIds directly. flushLiveFetches - // will batch every pending row into one request. - if (WS_ID_RE.test(wsId)) { - pendingLiveIds.add(wsId); - stillPending = true; - } - } - if (!stillPending) return; // every verdict landed — done - // Cancel any debounce that's still pending so our flush runs - // now instead of waiting for it. Then flush directly so the - // bulk request fires before the next tick re-arms. - if (liveBadgeFlushTimer !== null) { - clearTimeout(liveBadgeFlushTimer); - liveBadgeFlushTimer = null; - } - flushLiveFetches(); - judgePollTimer = setTimeout(_judgePollTick, JUDGE_POLL_INTERVAL_MS); - } - // Build the inline approval block: severity pill, intent summary + // judge reasoning, and approve/deny buttons. Returns a DOM node or // null if the detail is unusable (defensive \u2014 server is supposed to @@ -2534,6 +2535,37 @@ return sub; } + // Loading-state placeholder rendered while the bulk fetch is + // in-flight. Same outer ``.approval-block`` class so the row's + // height stays stable when the real content swaps in (no shove of + // sibling rows mid-mouse-movement). Carries an aria-label so a + // screen reader announces the pending approval; the actual + // assertive ``announceApproval`` call in handleChildState fires + // once on the rising edge so the operator hears the demand even + // before the bulk fetch lands. + function renderApprovalPlaceholder(child) { + const block = document.createElement("div"); + block.className = "approval-block approval-block-loading"; + block.setAttribute("role", "region"); + block.setAttribute( + "aria-label", + "Approval required for " + + (child.name || child.ws_id || "child") + + " — loading details", + ); + const header = document.createElement("div"); + header.className = "approval-header"; + const pill = document.createElement("span"); + pill.className = "approval-pill approval-pill-pending"; + const spin = document.createElement("span"); + spin.className = "approval-loading-spin"; + pill.appendChild(spin); + pill.appendChild(document.createTextNode(" loading…")); + header.appendChild(pill); + block.appendChild(header); + return block; + } + function renderApprovalBlock(child, detail) { if (!detail || !Array.isArray(detail.items) || detail.items.length === 0) { return null; @@ -2562,6 +2594,16 @@ const block = document.createElement("div"); block.className = "approval-block"; + block.setAttribute("role", "region"); + block.setAttribute( + "aria-label", + "Approval required for " + + (child.name || child.ws_id || "child") + + " — " + + (primary && primary.header + ? primary.header + : items.length + " tool calls"), + ); // Header line: pill + tool name(s) + tier:model const header = document.createElement("div"); @@ -2603,6 +2645,22 @@ const conf = verdict.confidence; const confStr = typeof conf === "number" ? " " + conf.toFixed(2) : ""; pill.textContent = (verdict.risk_level || "").toUpperCase() + confStr; + // SR-friendly label: spell out the risk level + optional + // confidence so a screen reader doesn't read "LOW 0.85" as + // a string of letters and a number. Visual text stays compact. + const riskWord = + riskCls === "crit" + ? "critical" + : riskCls === "high" + ? "high" + : riskCls === "med" + ? "medium" + : "low"; + const confLabel = + typeof conf === "number" + ? ", confidence " + Math.round(conf * 100) + "%" + : ""; + pill.setAttribute("aria-label", riskWord + " risk" + confLabel); } header.appendChild(pill); @@ -2803,17 +2861,33 @@ // Stale call_id \u2014 server has rolled to a new round, or // (more commonly) the approval was already resolved on // another channel and this click raced. Keep both buttons - // disabled until the urgent refresh lands and re-renders - // the row: the row is about to be replaced wholesale, so - // the disabled DOM is dropped along with it. Re-enabling - // here was the bug \u2014 it opened a window where rapid clicks - // hit the same already-resolved approval, each producing a - // fresh 409, looping until the live-bulk eventually cleared - // the row. On the rare path where the urgent refresh fails - // entirely, the operator can hit the Refresh button on the - // children panel to force a full reload. + // disabled until the refresh lands and re-renders the row: + // the row is about to be replaced wholesale, so the disabled + // DOM is dropped along with it. Re-enabling here was the bug + // \u2014 it opened a window where rapid clicks hit the same + // already-resolved approval, each producing a fresh 409, + // looping until the live-bulk eventually cleared the row. + // On the rare path where the refresh fails entirely, the + // operator can hit the Refresh button on the children panel + // to force a full reload. invalidateLiveBadge clears the + // cached entry so the next scheduleLiveFetch falls through + // the TTL gate (no cached entry to compare against); the + // standard 250ms debounce batches with any other in-flight + // pending ids. + // Inline note so the operator's "did my click work?" question + // gets answered without a noisy toast. Stays in the row until + // the refresh replaces the whole approval block. + const block = denyBtn.closest(".approval-block"); + if (block && !block.querySelector(".approval-stale-note")) { + const note = document.createElement("div"); + note.className = "approval-stale-note"; + note.setAttribute("role", "status"); + note.textContent = + "\u21bb already resolved elsewhere \u2014 refreshing\u2026"; + block.appendChild(note); + } invalidateLiveBadge(targetWsId); - scheduleLiveFetch(targetWsId, { urgent: true }); + scheduleLiveFetch(targetWsId); // Quiet console-warn for diagnostics; no toast \u2014 the // disappearing buttons / fresh row IS the operator-facing // signal, and a toast on every rapid-click 409 would just @@ -2826,14 +2900,30 @@ } // Optimistic clear \u2014 the next child_ws_state event will arrive // shortly and trigger a real refresh, but clearing locally - // makes the buttons disappear immediately on click. + // makes the buttons disappear immediately on click. The + // ``sseUpdatedAt`` bump is load-bearing: ``flushLiveFetches``'s + // merge guard preserves cleared pending_approval / _detail + // against an in-flight bulk fetch only while + // ``now - prev.sseUpdatedAt < SSE_AUTHORITATIVE_MS`` \u2014 without + // bumping, a bulk fetch landing in the gap reverts the + // cleared state from the upstream cache and the approve/deny + // pill flickers back. (Caught by /review bug-4.) const cached = liveBadgeCache.get(targetWsId); if (cached && cached.live) { cached.live = Object.assign({}, cached.live, { pending_approval: false, pending_approval_detail: null, }); - liveBadgeCache.set(targetWsId, cached); + cached.sseUpdatedAt = Date.now(); + _liveBadgeCacheSet(targetWsId, cached); + } + // Also clear the activity_state mirror so the row's "⚑ approval" + // badge in .meta disappears immediately too — without this, the + // row shows the badge with no buttons for ~50-150ms until the + // child_ws_approval_resolved push or next state event lands. + const childState = childrenState.get(targetWsId); + if (childState && childState.activity_state === "approval") { + childState.activity_state = ""; } renderChildren(); } catch (e) { @@ -2902,7 +2992,49 @@ return _childObserver; } + // Focus preservation helpers shared by _renderChildrenNow and + // _updateChildRow. ``replaceChildren()`` / ``replaceWith()`` blow + // away the focused element silently — without restore, the operator + // gets bounced to mid-Tab whenever any state event fires for + // a row in the children tree. + function _captureRowFocusKey(scopeEl) { + const active = document.activeElement; + if (!active || !scopeEl || !scopeEl.contains(active)) return null; + const row = active.closest(".ch-row"); + if (!row || !row.dataset.wsId) return null; + return { + wsId: row.dataset.wsId, + marker: active.className || active.tagName, + }; + } + + function _restoreRowFocus(scopeEl, focusKey) { + if (!focusKey || !scopeEl) return; + const sel = '.ch-row[data-ws-id="' + cssEscape(focusKey.wsId) + '"]'; + const row = scopeEl.matches(sel) ? scopeEl : scopeEl.querySelector(sel); + if (!row) return; + let target = null; + if (focusKey.marker) { + // CSS.escape can't safely round-trip a class list with spaces, + // so we walk focusables and string-compare. A future refactor + // could swap to a stable ``data-focus-key`` attribute on each + // focusable element to dodge class-string identity entirely + // (see /review bug-2 — currently low risk because all + // focusables in renderChildRow / renderApprovalBlock carry + // single-class names). + const candidates = row.querySelectorAll("button, [tabindex], a, summary"); + for (const el of candidates) { + if ((el.className || el.tagName) === focusKey.marker) { + target = el; + break; + } + } + } + if (target) target.focus({ preventScroll: true }); + } + function _renderChildrenNow() { + const focusKey = _captureRowFocusKey(childrenTreeEl); childrenTreeEl.setAttribute("aria-busy", "false"); const rows = Array.from(childrenState.values()); // Sort: non-terminal states first, then by name. @@ -2934,7 +3066,42 @@ else visibleChildIds.add(r.ws_id); // fallback: treat all visible }); } - childrenCountEl.textContent = rows.length ? "(" + rows.length + ")" : ""; + // Sidebar count: total + pending-approval annotation. The + // pending count is maintained incrementally on cache mutations + // (see ``pendingApprovalIds`` near the cache definition) so this + // is O(1) per render rather than an O(N) walk over the cache. + const pending = pendingApprovalIds.size; + childrenCountEl.textContent = rows.length + ? "(" + + rows.length + + (pending > 0 ? " · " + pending + " pending" : "") + + ")" + : ""; + _restoreRowFocus(childrenTreeEl, focusKey); + } + + // Targeted single-row update — used by handlers that only touch + // one row's state (verdict landing, approval resolved). Avoids + // the tree-wide rebuild ``_renderChildrenNow`` does so a 200-child + // tree doesn't re-paint 199 unaffected rows on every verdict + // arrival. Falls back to the full render if the row isn't in the + // DOM yet (first-time render). Preserves keyboard focus across + // the row swap — the same invariant ``_renderChildrenNow`` keeps + // for tree-wide rebuilds (caught by /review bug-3). + function _updateChildRow(childId) { + const sel = '.ch-row[data-ws-id="' + cssEscape(childId) + '"]'; + const row = childrenTreeEl.querySelector(sel); + const entry = childrenState.get(childId); + if (row && entry) { + const focusKey = _captureRowFocusKey(row); + const replacement = renderChildRow(entry); + row.replaceWith(replacement); + const obs = _getChildObserver(); + if (obs) obs.observe(replacement); + _restoreRowFocus(replacement, focusKey); + } else { + renderChildren(); + } } function renderTaskRow(task) { @@ -3065,15 +3232,9 @@ const LIVE_BADGE_BULK_FLUSH_MS = LIVE_BADGE_DEBOUNCE_MS; const pendingLiveIds = new Set(); let liveBadgeFlushTimer = null; - // Urgent-flush coalesce flag — N urgent calls in the same JS tick - // would otherwise issue N single-id bulk fetches (bulk endpoint - // accepts up to LIVE_BADGE_BULK_CAP ids per request). queueMicrotask - // batches them into one request that drains pendingLiveIds. - let urgentFlushScheduled = false; - function scheduleLiveFetch(childWsId, opts) { + function scheduleLiveFetch(childWsId) { if (!childWsId) return; - const urgent = !!(opts && opts.urgent); // Skip terminal-state children entirely — their live block will // never change again; fetching just burns a round-trip and caches // a stale value. Renderer already styles closed/deleted rows. @@ -3093,40 +3254,16 @@ // change on any child triggers a fresh fetch → retry storm for // users who'll never have permission mid-session. if (cached.permanent) return; - // Urgent fetches bypass the TTL — used when a child enters - // approval state and the row needs the rich pending_approval_detail - // payload (call_id, items, judge_verdict) to render inline buttons. - // Waiting for the next 5s TTL window would leave the operator - // staring at "⚑ approval" with no way to act. - if (!urgent && Date.now() - cached.fetched < LIVE_BADGE_TTL_MS) return; + // TTL gate — slower-moving fields (tokens, context_ratio) refresh + // on this cadence. Approval state is event-driven (intent_verdict + // / approval_resolved push, plus the bulk fetch on initial + // approval entry); callers wanting a forced re-fetch + // (e.g. 409 stale-call_id retry) call invalidateLiveBadge first + // so the cache is empty and this gate falls through. + if (Date.now() - cached.fetched < LIVE_BADGE_TTL_MS) return; } if (!WS_ID_RE.test(childWsId)) return; pendingLiveIds.add(childWsId); - // Urgent: cancel the pending debounce and schedule a flush on - // the next microtask so N urgent calls in the same tick coalesce - // into one bulk request. Without the microtask hop, each urgent - // caller would drain pendingLiveIds with a single id and fire a - // separate fetch — defeating the bulk endpoint that accepts up - // to LIVE_BADGE_BULK_CAP ids per request. - if (urgent) { - if (liveBadgeFlushTimer !== null) { - clearTimeout(liveBadgeFlushTimer); - liveBadgeFlushTimer = null; - } - if (!urgentFlushScheduled) { - urgentFlushScheduled = true; - const flush = () => { - urgentFlushScheduled = false; - flushLiveFetches(); - }; - if (typeof queueMicrotask === "function") { - queueMicrotask(flush); - } else { - setTimeout(flush, 0); - } - } - return; - } if (liveBadgeFlushTimer !== null) return; liveBadgeFlushTimer = setTimeout(() => { liveBadgeFlushTimer = null; @@ -3181,7 +3318,7 @@ pending_approval_detail: prev.live.pending_approval_detail, }); } - liveBadgeCache.set(id, { + _liveBadgeCacheSet(id, { live: mergedLive, fetched: now, // Denied ids are permission/identity misses — mark permanent @@ -3209,7 +3346,7 @@ const now = Date.now(); ids.forEach((id) => { const prev = liveBadgeCache.get(id); - liveBadgeCache.set(id, { + _liveBadgeCacheSet(id, { live: null, fetched: now, permanent: isPermanent, @@ -3221,7 +3358,7 @@ } function invalidateLiveBadge(childWsId) { - liveBadgeCache.delete(childWsId); + _liveBadgeCacheDelete(childWsId); } // --- SSE handlers for child_ws_* events ---------------------------- @@ -3261,44 +3398,54 @@ if (ev.node_id) existing.node_id = ev.node_id; childrenState.set(childId, existing); _touchChild(childId); - // pending_approval_detail rides on the ws_state event when an - // approval is pending (see turnstone/server.py - // WebUI._broadcast_state — gated on ``_pending_approval is not - // None``, so absent on the steady state and possibly null on a - // node mid-rolling-upgrade). When present we mutate - // liveBadgeCache directly here — inline approve/deny buttons - // render in lockstep with the activity_state transition without - // a separate live-bulk fetch. The cache entry is tagged - // ``sseUpdatedAt`` so a bulk-poll landing within - // SSE_AUTHORITATIVE_MS preserves the SSE-supplied fields - // (the upstream /dashboard cache's ~2s TTL would otherwise - // clobber a fresh transition with pre-transition state). - const pendingApproval = existing.activity_state === "approval"; - const evDetail = - ev.pending_approval_detail !== undefined - ? ev.pending_approval_detail - : null; + // Track ``pending_approval`` flag from BOTH state and + // activity_state. ``state="attention"`` is the canonical signal + // that the workstream needs operator attention; ``activity_state + // ="approval"`` is the secondary signal set by approve_tools. + // We trust either: in practice the worker thread can fire the + // state transition before approve_tools has updated activity_state + // (the two writes happen on different lines under different + // locks), so a state-attention event with empty activity_state + // is a real and common case — checking only activity_state + // misses 30+ children all sitting in attention. (Caught manual + // testing: 30 children all state=attention rendered with no + // approval blocks because pendingApproval was always false.) + const pendingApproval = + existing.state === "attention" || existing.activity_state === "approval"; const cached = liveBadgeCache.get(childId); - // When pending, prefer SSE-supplied detail. If SSE didn't - // carry it (rare race; e.g. a node mid-rolling-upgrade), keep - // any existing cached detail rather than blanking the row — - // the next bulk-poll catches up. When not pending, hard-clear. - let nextDetail; - if (pendingApproval) { - nextDetail = - evDetail !== null - ? evDetail - : cached && cached.live - ? cached.live.pending_approval_detail - : null; - } else { - nextDetail = null; + const cachedLive = (cached && cached.live) || {}; + // Rising-edge detection BEFORE we mutate the cache. The chat-pane + // tool batches already announce assertively + // (renderApprovalDock / appendToolBatch); the children-tree was + // silent for SR users — fixing that here so a blind operator + // hears the demand for action. + const wasPendingApproval = cachedLive.pending_approval === true; + if (pendingApproval && !wasPendingApproval) { + _announceAssertive( + "Approval required: " + (existing.name || childId.slice(0, 8)), + ); } - const nextLive = Object.assign({}, (cached && cached.live) || {}, { + const nextLive = Object.assign({}, cachedLive, { pending_approval: pendingApproval, - pending_approval_detail: nextDetail, }); - liveBadgeCache.set(childId, { + // Off-approval transition is the ONLY case here that + // authoritatively writes a value the bulk fetch must not + // resurrect (a stale bulk-fetch landing within + // SSE_AUTHORITATIVE_MS would otherwise re-render the cleared + // approval block). Setting ``pending_approval=true`` does NOT + // claim cache authority — the bulk fetch is the source of + // truth for ``pending_approval_detail``, and bumping + // sseUpdatedAt here makes the merge guard in flushLiveFetches + // preserve our stale (often null) detail over the bulk + // fetch's actual data, leaving the row stuck on the loading + // placeholder. (Caught when the screenshot showed buttons + // briefly then loading replaced them.) + const detailClearedAuthoritatively = + !pendingApproval && cachedLive.pending_approval === true; + if (!pendingApproval) { + nextLive.pending_approval_detail = null; + } + _liveBadgeCacheSet(childId, { live: nextLive, // Preserve prior bulk-poll fetched timestamp so a fresh SSE // tick doesn't artificially extend the 5s TTL gate in @@ -3306,7 +3453,11 @@ // moving fields (tokens, context_ratio) on its own schedule. fetched: cached ? cached.fetched : 0, permanent: !!(cached && cached.permanent), - sseUpdatedAt: Date.now(), + sseUpdatedAt: detailClearedAuthoritatively + ? Date.now() + : cached + ? cached.sseUpdatedAt || 0 + : 0, }); renderChildren(); // Do NOT invalidateLiveBadge on routine state ticks — that @@ -3344,6 +3495,104 @@ renderChildren(); } + // Stage 3 Step 7 — verdict + approval reducer. + // + // Both write directly to liveBadgeCache and skip scheduleLiveFetch, + // which sidesteps the visibility gate in that path so off-screen + // rows pick up the new cache value the moment they scroll back in. + // Both are idempotent — the bulk-fetch (single source of truth for + // the items list) and the explicit intent_verdict event can deliver + // overlapping data; receiving twice re-stamps the same value. + // + // Both call ``_updateChildRow`` (targeted single-row swap) instead + // of ``renderChildren`` so a verdict landing on row 3 doesn't rebuild + // every other row in a 200-child sidebar — which would also blow + // away keyboard focus on whatever row the operator was tabbing to. + + function handleChildIntentVerdict(ev) { + const childId = ev.child_ws_id || ev.ws_id; + if (!childId) return; + const verdict = ev.verdict || {}; + const callId = verdict.call_id || ""; + if (!callId) return; + const cached = liveBadgeCache.get(childId); + const cachedLive = (cached && cached.live) || {}; + const detail = cachedLive.pending_approval_detail; + if (!detail) { + // No pending_approval_detail to stamp the verdict onto. The + // verdict is still durable in storage; the next bulk fetch + // will hydrate the detail and include the verdict via the + // existing serialize path. + return; + } + // Stamp on the matching item (UI render reads judge_verdict per + // item) AND on the by-call_id map (matches the + // serialize_pending_approval_detail shape). + const items = Array.isArray(detail.items) ? detail.items : []; + for (const item of items) { + if (item && item.call_id === callId) { + item.judge_verdict = verdict; + break; + } + } + if (!detail.llm_verdicts) detail.llm_verdicts = {}; + detail.llm_verdicts[callId] = verdict; + // judge_pending flips false once every item has a verdict — + // matches the server-side serializer's logic. + if (items.length > 0) { + detail.judge_pending = !items.every((it) => it && it.judge_verdict); + } + _liveBadgeCacheSet(childId, { + live: cachedLive, + fetched: cached ? cached.fetched : 0, + permanent: !!(cached && cached.permanent), + sseUpdatedAt: Date.now(), + }); + _updateChildRow(childId); + } + + function handleChildApprovalResolved(ev) { + const childId = ev.child_ws_id || ev.ws_id; + if (!childId) return; + const cached = liveBadgeCache.get(childId); + const cachedLive = (cached && cached.live) || {}; + cachedLive.pending_approval = false; + cachedLive.pending_approval_detail = null; + _liveBadgeCacheSet(childId, { + live: cachedLive, + fetched: cached ? cached.fetched : 0, + permanent: !!(cached && cached.permanent), + sseUpdatedAt: Date.now(), + }); + _updateChildRow(childId); + } + + // Push path for the initial approval items — eliminates the + // bulk-fetch race that previously left rows stuck on a loading + // placeholder when the bulk fetch landed in the gap between the + // state transition to ATTENTION and ``_pending_approval`` being + // set inside ``approve_tools``. Stamps the items into the cache + // directly; the bulk fetch remains as a reconnect / refresh + // fallback. Idempotent — receiving the same approve_request twice + // re-stamps the same detail. + function handleChildApproveRequest(ev) { + const childId = ev.child_ws_id || ev.ws_id; + if (!childId) return; + const detail = ev.detail || null; + if (!detail) return; + const cached = liveBadgeCache.get(childId); + const cachedLive = (cached && cached.live) || {}; + cachedLive.pending_approval = true; + cachedLive.pending_approval_detail = detail; + _liveBadgeCacheSet(childId, { + live: cachedLive, + fetched: cached ? cached.fetched : 0, + permanent: !!(cached && cached.permanent), + sseUpdatedAt: Date.now(), + }); + _updateChildRow(childId); + } + // Periodic sweep of stale terminal rows. Operator tabs left open all // day would otherwise accumulate entries for every child the // coordinator ever spawned — rows the user can still see (state != @@ -3359,7 +3608,7 @@ if (terminal && now - lastSeen > CHILDREN_TERMINAL_GRACE_MS) { childrenState.delete(id); childrenLastSeen.delete(id); - liveBadgeCache.delete(id); + _liveBadgeCacheDelete(id); visibleChildIds.delete(id); removed += 1; } @@ -3375,7 +3624,7 @@ const id = byAge[i][0]; childrenState.delete(id); childrenLastSeen.delete(id); - liveBadgeCache.delete(id); + _liveBadgeCacheDelete(id); visibleChildIds.delete(id); removed += 1; } @@ -3388,7 +3637,7 @@ if (childrenRefreshBtn) { childrenRefreshBtn.addEventListener("click", () => { - liveBadgeCache.clear(); + _liveBadgeCacheClear(); // Explicit refresh wipes SSE-discovered rows the server no // longer knows about — the operator asked for a clean snapshot. loadChildren({ replace: true }); diff --git a/turnstone/console/static/coordinator/index.html b/turnstone/console/static/coordinator/index.html index ae4988e8..72028162 100644 --- a/turnstone/console/static/coordinator/index.html +++ b/turnstone/console/static/coordinator/index.html @@ -181,6 +181,42 @@ font-family: var(--font-mono); font-size: 11px; line-height: 1.4; + /* Smooth the auto-expand on judge-verdict landing so the eye can + track sibling rows shifting down rather than them snapping. */ + transition: max-height 200ms ease-out; + max-height: 80vh; + overflow: hidden; + } + /* Loading-state placeholder rendered while the bulk fetch is + in-flight. Same outer container as the real block so the row + height is stable when the real content swaps in. */ + .ch-row .approval-block-loading { + color: var(--ink-3); + } + .ch-row .approval-loading-spin { + display: inline-block; + width: 8px; + height: 8px; + border-radius: 50%; + border: 1.5px solid var(--accent); + border-top-color: transparent; + margin-right: 5px; + animation: ts-spin 0.9s linear infinite; + vertical-align: middle; + } + /* Inline status note shown when the operator clicks Approve/Deny + but the call_id is stale (already resolved on another channel). + Quieter than a toast — the row is about to be replaced wholesale + by the refresh, so this only needs to bridge ~350ms. */ + .ch-row .approval-stale-note { + font-size: 11px; + color: var(--ink-3); + font-style: italic; + margin-top: 4px; + } + @media (prefers-reduced-motion: reduce) { + .ch-row .approval-block { transition: none; } + .ch-row .approval-loading-spin { animation: none; } } /* Auto-approved pill — same 22px indent as .approval-block / .meta so a child row showing both stacks cleanly. Lower visual weight @@ -442,6 +478,22 @@ link feed.css (no .feed-item usage). */ @keyframes ts-spin { to { transform: rotate(360deg); } } + /* SSE queue depth indicator — populated by the periodic + ``_queue_stats`` event in the events SSE handler. Color escalates + so a backed-up queue (slow consumer / browser throttling + background tabs) is glanceable from the status bar. */ + .ws-status-bar .ws-sb-sse-queue { + color: var(--ink-3); + font-size: 11px; + } + .ws-status-bar .ws-sb-sse-queue.warn { + color: var(--warn); + } + .ws-status-bar .ws-sb-sse-queue.danger { + color: var(--err); + font-weight: 600; + } + /* Sidebar mobile toggle (desktop hides; mobile shows via media query below). */ #coord-sidebar-toggle { @@ -542,6 +594,12 @@ 0 / — 0 tools turn 0 + + diff --git a/turnstone/core/child_source.py b/turnstone/core/child_source.py new file mode 100644 index 00000000..eac64cfd --- /dev/null +++ b/turnstone/core/child_source.py @@ -0,0 +1,264 @@ +"""Strategy interface for delivering child workstream lifecycle events. + +Two implementations bind to the unified :class:`ChildrenRegistry`: + +- :class:`SameNodeChildSource` — subscribes to a :class:`SessionManager`'s + state-change callbacks. In-process, no transport. For interactive + workstreams that spawn children locally (no cluster routing). +- :class:`ClusterChildSource` — subscribes to a :class:`ClusterCollector`'s + listener channel and runs a daemon thread that drains the queue and + pushes events to the sink. For coordinator workstreams whose + children are routed across the cluster by hash bucket. + +The strategy doesn't translate events into UI-shaped payloads; that's +the sink's job. This split keeps the strategy generic across kinds and +lets the consumer (e.g. ``CoordinatorAdapter._dispatch_child_event``) +own the per-kind translation. + +Sink signature: ``Callable[[dict[str, Any]], None]``. The sink is +responsible for filtering by registry membership; strategies push raw +events without registry-side filtering so the sink can decide whether +to act based on its own state. +""" + +from __future__ import annotations + +import queue +import threading +from typing import TYPE_CHECKING, Any, Protocol + +from turnstone.core.log import get_logger + +if TYPE_CHECKING: + from collections.abc import Callable, Iterable + + from turnstone.core.children_registry import ChildrenRegistry + from turnstone.core.workstream import WorkstreamState + +log = get_logger(__name__) + + +class ChildSource(Protocol): + """Subscription strategy for child workstream lifecycle events.""" + + def start(self, sink: Callable[[dict[str, Any]], None]) -> None: + """Begin delivering events to ``sink``. Idempotent.""" + + def shutdown(self) -> None: + """Stop the strategy. Idempotent; safe to call multiple times.""" + + +class _CollectorProtocol(Protocol): + """Subset of :class:`ClusterCollector` that :class:`ClusterChildSource` consumes. + + Defined here (not imported) to keep ``turnstone/core/`` free of + ``turnstone/console/`` imports AND to let test fakes satisfy the + type signature without subclassing the real collector. + """ + + def get_snapshot_and_register(self, q: queue.Queue[dict[str, Any]]) -> dict[str, Any]: ... + + def unregister_listener(self, q: queue.Queue[dict[str, Any]]) -> None: ... + + +class _ManagerProtocol(Protocol): + """Subset of :class:`SessionManager` that :class:`SameNodeChildSource` consumes. + + Lets test fakes participate in the strategy's typed surface + without forcing a full SessionManager construction. + """ + + def subscribe_to_state(self, callback: Callable[[str, WorkstreamState], None]) -> None: ... + + def unsubscribe_from_state(self, callback: Callable[[str, WorkstreamState], None]) -> None: ... + + +class SameNodeChildSource: + """In-process child events via :class:`SessionManager` state observer. + + Subscribes to the manager's state-change callbacks (registered via + :meth:`SessionManager.subscribe_to_state`). For each transition on + a workstream that's a known child (per the + :class:`ChildrenRegistry` reverse index), synthesises a + cluster-state-shaped event and pushes it to the sink. + + Used by interactive workstreams when they gain spawn capability — + children live on the same node as the parent, so no cluster routing + is needed and event fan-out is in-process. + """ + + def __init__( + self, + manager: _ManagerProtocol, + registry: ChildrenRegistry, + ) -> None: + self._manager = manager + self._registry = registry + self._sink: Callable[[dict[str, Any]], None] | None = None + self._callback: Callable[[str, WorkstreamState], None] | None = None + + def start(self, sink: Callable[[dict[str, Any]], None]) -> None: + if self._callback is not None: + return # idempotent — already started + self._sink = sink + + def _on_state(ws_id: str, state: WorkstreamState) -> None: + sink_fn = self._sink + if sink_fn is None: + return + # Cheap pre-filter: skip dispatch for transitions on + # workstreams that aren't children of any in-memory parent. + # ``has_children`` is a lock-free dict-truthiness read; the + # ``parent_for`` call below would otherwise acquire the + # registry lock on every state change even when no + # children exist (the steady state for an interactive + # manager). The cluster strategy can't pre-filter because + # the collector queue carries all events; here we have the + # information to skip the synthesis entirely. + if not self._registry.has_children(): + return + if self._registry.parent_for(ws_id) is None: + return + # ``pending_approval_detail`` deliberately omitted — the + # field was removed from cluster_state end-to-end in the + # Stage 3 cleanup pass. Approval items arrive via bulk + # fetch; verdicts via the explicit intent_verdict event; + # resolution via approval_resolved. + event = { + "type": "cluster_state", + "ws_id": ws_id, + "state": state.value, + "node_id": "", + "tokens": 0, + "activity_state": "", + } + try: + sink_fn(event) + except Exception: + log.debug("same_node_child_source.sink_failed", exc_info=True) + + self._callback = _on_state + self._manager.subscribe_to_state(_on_state) + + def shutdown(self) -> None: + cb = self._callback + if cb is None: + return + try: + self._manager.unsubscribe_from_state(cb) + except Exception: + log.debug("same_node_child_source.unsubscribe_failed", exc_info=True) + self._callback = None + self._sink = None + + +class ClusterChildSource: + """Cross-node child events via :class:`ClusterCollector` subscription. + + Refactor of the existing fan-out machinery from + ``CoordinatorAdapter`` (was ``_collector_queue`` + + ``_fanout_thread`` + ``_fanout_loop``). Subscribes as a listener on + the collector's broadcast channel and runs a daemon thread that + drains the queue, pushing each event to the sink. + + On :meth:`start`, also primes the registry from the collector's + snapshot so a parent that re-installs after a console restart sees + its already-live children without waiting for the next state tick. + The ``parents_provider`` callback returns the set of in-memory + parent ws_ids for snapshot filtering — only children whose parent + is currently installed get merged. + """ + + def __init__( + self, + collector: _CollectorProtocol, + registry: ChildrenRegistry, + *, + parents_provider: Callable[[], Iterable[str]], + ) -> None: + self._collector = collector + self._registry = registry + self._parents_provider = parents_provider + self._sink: Callable[[dict[str, Any]], None] | None = None + self._queue: queue.Queue[dict[str, Any]] | None = None + self._thread: threading.Thread | None = None + self._stop = threading.Event() + + def start(self, sink: Callable[[dict[str, Any]], None]) -> None: + if self._thread is not None and self._thread.is_alive(): + return # idempotent — already started + self._sink = sink + self._queue = queue.Queue(maxsize=1000) + snapshot = self._collector.get_snapshot_and_register(self._queue) + self._prime_from_snapshot(snapshot) + self._stop.clear() + t = threading.Thread( + target=self._loop, + name="cluster-child-source", + daemon=True, + ) + self._thread = t + t.start() + + def shutdown(self) -> None: + self._stop.set() + t = self._thread + q = self._queue + coll = self._collector + self._thread = None + self._queue = None + if coll is not None and q is not None: + try: + coll.unregister_listener(q) + except Exception: + log.debug( + "cluster_child_source.unregister_listener_failed", + exc_info=True, + ) + if t is not None: + t.join(timeout=2.0) + self._sink = None + + def _loop(self) -> None: + q = self._queue + if q is None: + return + while not self._stop.is_set(): + try: + event = q.get(timeout=1.0) + except queue.Empty: + continue + sink = self._sink + if sink is None: + continue + try: + sink(event) + except Exception: + log.debug("cluster_child_source.dispatch_failed", exc_info=True) + + def _prime_from_snapshot(self, snapshot: dict[str, Any]) -> None: + """Populate the registry from a collector snapshot. + + For every workstream in the snapshot whose ``parent_ws_id`` + names a currently-installed parent (per ``parents_provider``), + merge it into the registry. Caller-installed parents that + appear in the snapshot are seeded; unknown parents are skipped + — they'll be picked up by the live fan-out path once their + ``ws_created`` event arrives. + """ + nodes = snapshot.get("nodes", []) if isinstance(snapshot, dict) else [] + if not nodes: + return + known_parents = set(self._parents_provider()) + if not known_parents: + return + by_parent: dict[str, list[str]] = {} + for node in nodes: + for entry in node.get("workstreams", []) or []: + parent = entry.get("parent_ws_id") or "" + child_id = entry.get("id") or "" + if not parent or not child_id or parent not in known_parents: + continue + by_parent.setdefault(parent, []).append(child_id) + for parent, kids in by_parent.items(): + self._registry.merge_children(parent, kids) diff --git a/turnstone/core/children_registry.py b/turnstone/core/children_registry.py new file mode 100644 index 00000000..0a5f7ff7 --- /dev/null +++ b/turnstone/core/children_registry.py @@ -0,0 +1,177 @@ +"""Universal parent → children registry for SessionManager. + +Pure data + lookups; no IO, no transport. Lifted from +:class:`turnstone.console.coordinator_adapter.CoordinatorAdapter` where +it lived bound to the coordinator kind. The lift is what lets the +``ChildSource`` strategies (Step 2) plug into a single shared primitive +regardless of whether children are local (interactive) or cluster-routed +(coordinator). + +Storage rebuild and snapshot priming happen in the caller — typically +the ``ChildSource`` implementation that owns the relevant transport. +The registry exposes :meth:`merge_children` for bulk seeding so callers +that compute child id lists from any source can feed them in without +the registry needing to know about storage shapes or collector +snapshots. + +Threading: every public method is internally locked. Helpers suffixed +``_locked`` require the caller to already hold :attr:`_lock`. +""" + +from __future__ import annotations + +import threading +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from collections.abc import Iterable + + +class ChildrenRegistry: + """Tracks parent → children + reverse lookup for in-memory parents. + + The forward index (``_children``) is parent_ws_id → set of child ws_ids. + The reverse index (``_child_to_parent``) is child_ws_id → parent_ws_id. + The presence map (``_active``) is parent_ws_id → UI ref, used by the + dispatch path to atomically check-and-route in one lock acquisition. + + Closed / deleted children stay in the registry until their owning + parent is uninstalled — the tree UI keeps rendering them grayed out; + state authority lives in storage, not here. + """ + + def __init__(self) -> None: + self._children: dict[str, set[str]] = {} + self._child_to_parent: dict[str, str] = {} + self._lock = threading.Lock() + self._active: dict[str, Any] = {} + + # ------------------------------------------------------------------ + # Lifecycle — install / uninstall a parent + # ------------------------------------------------------------------ + + def install(self, parent_ws_id: str, ui: Any) -> None: + """Seed the forward set + presence map for a new parent. + + Idempotent — re-installing re-points the UI but leaves the + existing child set intact. Mirrors the original + ``_install_coord_registry`` semantics so a coordinator that + rehydrates after a crash doesn't lose its known-children. + """ + with self._lock: + self._children.setdefault(parent_ws_id, set()) + self._active[parent_ws_id] = ui + + def uninstall(self, parent_ws_id: str) -> None: + """Drop a parent: forward set, reverse-index entries, presence. + + No-op if the parent is unknown. Used by close / eviction paths. + """ + with self._lock: + self._uninstall_locked(parent_ws_id) + + # ------------------------------------------------------------------ + # Mutation — register children under a parent + # ------------------------------------------------------------------ + + def add_child(self, parent_ws_id: str, child_ws_id: str) -> Any | None: + """Register a child under a parent. Returns parent's UI or None. + + Returns the parent's UI on success (so the dispatch path can + atomically check-and-route in one lock acquisition). Returns + ``None`` if the parent isn't installed (concurrent close / + eviction) or if the child is already registered (duplicate + ws_created from the cluster fan-out). + """ + with self._lock: + ui = self._active.get(parent_ws_id) + if ui is None: + return None + existing = self._children.setdefault(parent_ws_id, set()) + if child_ws_id in existing: + return None + existing.add(child_ws_id) + self._child_to_parent[child_ws_id] = parent_ws_id + return ui + + def merge_children(self, parent_ws_id: str, child_ws_ids: Iterable[str]) -> None: + """Bulk-merge child_ids under a parent. Idempotent. + + Sole bulk write-path — used by both storage-seeded rebuilds and + snapshot-seeded priming so reverse-index ordering invariants + hold regardless of which seed source races first. + """ + with self._lock: + self._merge_locked(parent_ws_id, child_ws_ids) + + # ------------------------------------------------------------------ + # Lookups + # ------------------------------------------------------------------ + + def parent_for(self, child_ws_id: str) -> str | None: + """Reverse lookup: which parent owns this child? O(1).""" + with self._lock: + return self._child_to_parent.get(child_ws_id) + + def has_children(self) -> bool: + """Lock-free fast path: any child registered under any parent? + + Reads ``bool(self._child_to_parent)`` without taking the lock. + Dict-truthiness is a single GIL-atomic read, so callers on + the hot state-broadcast path can short-circuit without paying + the lock acquisition when the registry is empty (the steady + state for an interactive manager today). The answer is best- + effort — if a child is added concurrently with the read the + caller may falsely return ``False``, but the next state event + will pick up the change correctly. + """ + return bool(self._child_to_parent) + + def children_of(self, parent_ws_id: str) -> list[str]: + """Snapshot copy of the parent's child ws_ids. + + Returned list is a copy so callers can iterate without holding + the registry lock during per-child work. A mutation racing with + the snapshot either lands before (included) or after (excluded) + — both outcomes are safe for cascade-style dispatch. + """ + with self._lock: + child_set = self._children.get(parent_ws_id) + return list(child_set) if child_set else [] + + def ui_for(self, parent_ws_id: str) -> Any | None: + """Look up the UI registered for a parent.""" + with self._lock: + return self._active.get(parent_ws_id) + + def parents(self) -> list[str]: + """Snapshot copy of installed parent ws_ids.""" + with self._lock: + return list(self._active) + + # ------------------------------------------------------------------ + # Locked helpers — caller must hold ``self._lock`` + # ------------------------------------------------------------------ + + def _merge_locked(self, parent_ws_id: str, child_ws_ids: Iterable[str]) -> None: + """Idempotent merge under caller's lock. Empty/falsy ids skipped.""" + existing = self._children.setdefault(parent_ws_id, set()) + for cid in child_ws_ids: + if cid and cid not in existing: + existing.add(cid) + self._child_to_parent[cid] = parent_ws_id + + def _uninstall_locked(self, parent_ws_id: str) -> None: + """Pop forward set + presence + own reverse-index entries. + + Defensive: only clears reverse entries that still point at + ``parent_ws_id``. Schema-shaped reassignments (rare but + possible) shouldn't orphan the new owner's entry. + """ + child_set = self._children.pop(parent_ws_id, None) + self._active.pop(parent_ws_id, None) + if child_set is None: + return + for cid in child_set: + if self._child_to_parent.get(cid) == parent_ws_id: + self._child_to_parent.pop(cid, None) diff --git a/turnstone/core/session_manager.py b/turnstone/core/session_manager.py index b227fdf8..a5808fe9 100644 --- a/turnstone/core/session_manager.py +++ b/turnstone/core/session_manager.py @@ -214,13 +214,21 @@ class SessionManager: # manager never reads them. self._active_id: str | None = None self._eviction_count: int = 0 - # Optional state-change observer. The CLI sets this to a - # callback that prints a background-attention notification - # when a non-focused workstream transitions to ATTENTION. - # Web/coord paths use the event_emitter's emit_state for their - # own fan-out; this is a second, manager-level hook for callers - # that don't consume SSE. - self._on_state_change: Callable[[str, WorkstreamState], None] | None = None + # State-change subscribers. Multi-subscriber to support the + # CLI's background-attention notification AND the in-process + # ``SameNodeChildSource`` strategy that delivers child + # workstream state changes to a parent's UI without going + # through the cluster bus. Each callback fires under + # exception-suppression so one failing subscriber doesn't + # block the others. Subscribers register via + # :meth:`subscribe_to_state`. ``_state_subscribers_lock`` + # guards mutation + snapshot — set_state copies the list + # under the lock then iterates the snapshot unlocked so a + # slow subscriber doesn't block subscribe/unsubscribe (and + # so concurrent subscribe/unsubscribe during a state event + # can't shift the iterator's index — caught by /review bug-1). + self._state_subscribers: list[Callable[[str, WorkstreamState], None]] = [] + self._state_subscribers_lock = threading.Lock() # ------------------------------------------------------------------ # Properties @@ -816,9 +824,36 @@ class SessionManager: log.debug("session_mgr.state_update_failed ws=%s", ws_id[:8], exc_info=True) if self._event_emitter is not None: self._event_emitter.emit_state(ws, state) - if self._on_state_change is not None: + # Snapshot under the subscribers lock so concurrent + # subscribe / unsubscribe can't shift the iterator's index + # mid-dispatch (skipping or repeating callbacks). Iterate + # the snapshot WITHOUT the lock so a slow callback doesn't + # block subscribe / unsubscribe. + with self._state_subscribers_lock: + subscribers = list(self._state_subscribers) + for callback in subscribers: with contextlib.suppress(Exception): - self._on_state_change(ws_id, state) + callback(ws_id, state) + + # ------------------------------------------------------------------ + # State-change subscription + # ------------------------------------------------------------------ + + def subscribe_to_state(self, callback: Callable[[str, WorkstreamState], None]) -> None: + """Register ``callback`` to fire on every workstream state change. + + Multiple subscribers are supported and fire in registration order. + Each callback is wrapped in exception-suppression so a failing + subscriber doesn't block the others. Use + :meth:`unsubscribe_from_state` to remove. + """ + with self._state_subscribers_lock: + self._state_subscribers.append(callback) + + def unsubscribe_from_state(self, callback: Callable[[str, WorkstreamState], None]) -> None: + """Remove a previously-registered state-change callback. No-op if absent.""" + with self._state_subscribers_lock, contextlib.suppress(ValueError): + self._state_subscribers.remove(callback) def cancel(self, ws_id: str) -> bool: """Cancel in-flight generation and unblock any pending approval / plan. diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index d5ec3b7d..062f3eb7 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -1377,9 +1377,14 @@ def make_events_handler(cfg: SessionEndpointConfig) -> Handler: import asyncio import json import queue + import time from sse_starlette import EventSourceResponse + from turnstone.core.session_ui_base import ( + _DEFAULT_LISTENER_QUEUE_MAX, + ) + if cfg.permission_gate is not None: err = cfg.permission_gate(request) if err is not None: @@ -1474,9 +1479,32 @@ def make_events_handler(cfg: SessionEndpointConfig) -> Handler: # cancel-detection latency the timeout would otherwise # gate; shortening to 1s 5x'd the wakeup rate without # any client-observable benefit). + # + # Periodic ``_queue_stats`` emission carries the + # per-tab listener queue depth so the browser can + # surface backed-up SSE consumers in the status bar + # (slow tab / browser throttling / network stall + # would otherwise be invisible until events visibly + # stop arriving). 2s cadence is observational — the + # JSON payload is tiny and won't measurably grow the + # queue itself. + queue_stats_interval_s = 2.0 + last_queue_stats_at = 0.0 while True: if await request.is_disconnected(): return + now_mono = time.monotonic() + if now_mono - last_queue_stats_at >= queue_stats_interval_s: + last_queue_stats_at = now_mono + yield { + "data": json.dumps( + { + "type": "_queue_stats", + "depth": client_queue.qsize(), + "max": _DEFAULT_LISTENER_QUEUE_MAX, + } + ) + } try: event = await loop.run_in_executor( live_executor, diff --git a/turnstone/core/session_ui_base.py b/turnstone/core/session_ui_base.py index d277bae2..c7144ae1 100644 --- a/turnstone/core/session_ui_base.py +++ b/turnstone/core/session_ui_base.py @@ -41,6 +41,56 @@ log = get_logger(__name__) # from bloating memory. _DEFAULT_LISTENER_QUEUE_MAX = 500 +# Event types whose delivery is load-bearing for correctness — not +# observational fluff. When a listener queue is full, ``_enqueue`` +# evicts the oldest event to make room for these rather than dropping +# them on the floor (the default ``queue.Full`` swallow). Token-stream +# / status / activity events drop equally as today; an approval +# request or verdict landing during a chatty mid-generation burst +# would otherwise be silently lost on a backed-up tab. +_CRITICAL_EVENT_TYPES = frozenset( + { + "intent_verdict", + "intent_pending", + "approval_resolved", + "approve_request", + "plan_review", + "ws_closed", + "child_ws_intent_verdict", + "child_ws_approval_resolved", + "child_ws_closed", + "child_ws_created", + } +) + + +def _put_with_priority( + q: queue.Queue[dict[str, Any]], + data: dict[str, Any], + *, + critical: bool, +) -> None: + """Put ``data`` onto ``q`` with selective drop-on-full semantics. + + Critical events evict one oldest item from ``q`` if it's full, + then retry the put. Best-effort events drop themselves on full + (matches the historical ``contextlib.suppress(queue.Full)`` + behaviour). The eviction is bounded — one drop is enough since + the queue was already at capacity before we tried. + """ + if critical: + try: + q.put_nowait(data) + except queue.Full: + with contextlib.suppress(queue.Empty): + q.get_nowait() + with contextlib.suppress(queue.Full): + q.put_nowait(data) + else: + with contextlib.suppress(queue.Full): + q.put_nowait(data) + + # Cap on the per-turn assistant content accumulator. The accumulator # is piggybacked onto the ``ws_state:idle`` broadcast payload so the # cluster collector / dashboard can render the freshly-emitted assistant @@ -246,14 +296,22 @@ class SessionUIBase: browser can validate it belongs to the pane's current workstream. Shallow-copies on stamp to avoid mutating a caller-owned dict. + + Critical event types (see :data:`_CRITICAL_EVENT_TYPES`) + evict one oldest item from a full queue rather than dropping + themselves — a chatty mid-generation burst can fill the + 500-slot per-tab queue, and silently dropping an + ``approve_request`` or ``intent_verdict`` because we couldn't + squeeze past a backlog of token-stream events would be a real + UX regression. Best-effort events drop on full as before. """ if "ws_id" not in data: data = {**data, "ws_id": self.ws_id} + critical = data.get("type") in _CRITICAL_EVENT_TYPES with self._listeners_lock: snapshot = list(self._listeners) for lq in snapshot: - with contextlib.suppress(queue.Full): - lq.put_nowait(data) + _put_with_priority(lq, data, critical=critical) def _register_listener( self, maxsize: int = _DEFAULT_LISTENER_QUEUE_MAX @@ -327,6 +385,11 @@ class SessionUIBase: "always": bool(always), } ) + # Kind-specific cross-stream broadcast — ConsoleCoordinatorUI + # overrides to push onto the cluster bus so a coord parent's + # tree UI clears the pending-approval pill in lockstep with + # the actual decision. Stage 3 Step 4. + self._broadcast_approval_resolved(approved, feedback, always=always) self._approval_event.set() @staticmethod @@ -579,6 +642,20 @@ class SessionUIBase: "judge_pending": judge_pending, } self._enqueue(self._pending_approval) + # Cross-stream broadcast — push the items via the cluster bus + # so a coord parent's tree UI can render the inline approve/deny + # block without waiting for a bulk fetch. Without this, the + # bulk fetch races with this assignment: the state transition + # to ATTENTION fires upstream BEFORE approve_tools runs (see + # session.py:_emit_state("attention") preceding ui.approve_tools), + # so a bulk fetch landing in the ~50-200ms window between + # _emit_state and this point sees ``_pending_approval=None`` + # and returns ``pending_approval_detail: null``. The 5s TTL + # then locks the coord row on a "loading" placeholder until + # the next state event triggers a refresh — which never comes + # while parked on _approval_event.wait. The push path + # eliminates the race. + self._broadcast_approve_request(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. @@ -632,6 +709,12 @@ class SessionUIBase: del self._llm_verdicts[oldest_key] self._llm_verdicts[call_id] = verdict self._enqueue({"type": "intent_verdict", **verdict}) + # Kind-specific cross-stream broadcast — ConsoleCoordinatorUI + # overrides to push onto the cluster bus so a coord parent's + # tree UI sees the verdict without polling. Default is no-op + # (the per-ws ``_enqueue`` above already covers WebUI's own + # SSE listeners). Stage 3 Step 4. + self._broadcast_intent_verdict(verdict) self._persist_intent_verdict(verdict) # Decision check + either queue or flag-for-persist happen # under ONE lock acquisition so resolve_approval can't swap- @@ -1351,6 +1434,42 @@ class SessionUIBase: Default: no-op. Subclasses override. """ + def _broadcast_intent_verdict(self, verdict: dict[str, Any]) -> None: # noqa: ARG002 — hook stub + """Fan an LLM intent-judge verdict out to the kind's transport. + + Default: no-op. ``ConsoleCoordinatorUI`` overrides to push a + ``intent_verdict`` event onto the cluster bus so the parent + coordinator's tree UI can render the risk pill + verdict + result without polling. Stage 3 Step 4: hook only — the + cluster-bus event class lands in Step 5. + """ + + def _broadcast_approval_resolved( + self, + approved: bool, # noqa: ARG002 — hook stub + feedback: str | None = None, # noqa: ARG002 — hook stub + *, + always: bool = False, # noqa: ARG002 — hook stub + ) -> None: + """Fan an ``approval_resolved`` decision out to the kind's transport. + + Default: no-op. ``ConsoleCoordinatorUI`` overrides to push to + the cluster bus so the parent coordinator's tree UI can clear + the pending-approval pill in sync with the actual decision. + """ + + def _broadcast_approve_request(self, detail: dict[str, Any]) -> None: # noqa: ARG002 — hook stub + """Fan an ``approve_request`` payload out to the kind's transport. + + Default: no-op. ``WebUI`` and ``ConsoleCoordinatorUI`` override + to push the items list (the same dict that landed in + ``_pending_approval``) onto their respective transports. The + push path eliminates the bulk-fetch race that otherwise + leaves coord rows stuck on a loading placeholder when the + bulk fetch lands in the gap between the state transition to + ATTENTION and ``_pending_approval`` being set. + """ + # ------------------------------------------------------------------ # State-broadcast snapshot helper # ------------------------------------------------------------------ diff --git a/turnstone/server.py b/turnstone/server.py index d14ce5ee..0325b2b7 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -81,6 +81,7 @@ from turnstone.core.session_routes import ( from turnstone.core.session_ui_base import ( AutoApproveReason, SessionUIBase, + _put_with_priority, fire_judge_verdict_metric, ) from turnstone.core.tools import TOOLS # noqa: F401 — available for introspection @@ -194,18 +195,14 @@ class WebUI(SessionUIBase): } if state == "idle": event["content"] = payload["content"] - # Coord tree-UI renders inline approve/deny buttons off - # ``pending_approval_detail``; carrying it on the - # state-change broadcast lets the cluster bus update those - # buttons in lockstep with ``activity_state`` instead of - # forcing the browser to chase a separate dashboard fetch. - # Gated on existence so we don't pay the serializer's - # per-broadcast verdict-cache deepcopy on the common - # no-approval-pending path. - if self._pending_approval is not None: - detail = self.serialize_pending_approval_detail() - if detail is not None: - event["pending_approval_detail"] = detail + # ``pending_approval_detail`` is NO LONGER piggybacked on + # state-change events (Stage 3 cleanup). Symmetric event + # flow now: initial approval items arrive via bulk fetch + # triggered by the ``activity_state="approval"`` transition, + # individual verdicts via the explicit + # ``intent_verdict`` event class, and resolution via + # ``approval_resolved``. Reducer no longer has to dedupe + # the piggyback path against the explicit one. try: WebUI._global_queue.put_nowait(event) except queue.Full: @@ -230,6 +227,82 @@ class WebUI(SessionUIBase): } ) + def _broadcast_intent_verdict(self, verdict: dict[str, Any]) -> None: + """Send an LLM intent-judge verdict to the global SSE channel. + + Stage 3 Step 5 — the cluster collector's ``_apply_delta`` + forwards this verbatim to the cluster bus, where coord + adapters dispatch it as ``child_ws_intent_verdict`` for the + owning parent's tree UI. Unlike the existing + ``pending_approval_detail`` piggyback on ``ws_state``, this + fires WHENEVER a verdict lands — including the common case + where the judge daemon writes during ``attention`` with no + state transition to ride along on. + """ + if WebUI._global_queue is not None: + # Critical event — evict an oldest item if the global + # queue is full rather than dropping the verdict. + _put_with_priority( + WebUI._global_queue, + { + "type": "intent_verdict", + "ws_id": self.ws_id, + "verdict": verdict, + }, + critical=True, + ) + + def _broadcast_approval_resolved( + self, + approved: bool, + feedback: str | None = None, + *, + always: bool = False, + ) -> None: + """Send an ``approval_resolved`` decision to the global SSE channel. + + Clears the parent's pending-approval pill in lockstep with + the actual decision rather than waiting for the next + state-change piggyback. + """ + if WebUI._global_queue is not None: + # Critical event — evict an oldest item if the global + # queue is full rather than dropping the resolution. + _put_with_priority( + WebUI._global_queue, + { + "type": "approval_resolved", + "ws_id": self.ws_id, + "approved": approved, + "feedback": feedback or "", + "always": bool(always), + }, + critical=True, + ) + + def _broadcast_approve_request(self, detail: dict[str, Any]) -> None: + """Send an ``approve_request`` payload to the global SSE channel. + + Push path for the initial approval items so a coord parent's + tree UI can render the inline approve/deny block immediately + without waiting for a bulk-fetch round-trip. The bulk fetch + races with ``_pending_approval`` being set inside + ``approve_tools`` (the state transition to ATTENTION fires + upstream first); the push path eliminates that race entirely. + """ + if WebUI._global_queue is not None: + # Critical event — evict an oldest item if the global + # queue is full rather than dropping the items. + _put_with_priority( + WebUI._global_queue, + { + "type": "approve_request", + "ws_id": self.ws_id, + "detail": detail, + }, + critical=True, + ) + # --- SessionUI protocol --- # # ``on_thinking_start`` / ``on_thinking_stop`` / ``on_reasoning_token``