diff --git a/tests/test_console.py b/tests/test_console.py index fa05f984..2a13f1ea 100644 --- a/tests/test_console.py +++ b/tests/test_console.py @@ -314,6 +314,48 @@ 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.""" + c = _make_collector() + c._nodes["node-a"] = NodeSnapshot( + node_id="node-a", + server_url="http://a:8080", + workstreams={"ws1": {"id": "ws1", "name": "same", "state": "idle"}}, + ) + 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", + { + "type": "node_snapshot", + "node_id": "node-a", + "workstreams": [ + { + "id": "ws1", + "name": "same", + "state": "running", + "activity_state": "approval", + "pending_approval_detail": detail, + } + ], + "health": {}, + "aggregate": {}, + }, + ) + + event = q.get_nowait() + assert event["type"] == "cluster_state" + assert event["pending_approval_detail"] == detail + def test_apply_snapshot_skips_empty_id_workstream(self): c = _make_collector() c._nodes["node-a"] = NodeSnapshot(node_id="node-a", server_url="http://a:8080") @@ -359,6 +401,40 @@ 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.""" + 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 = { + "items": [{"call_id": "c1", "header": "tool x"}], + "judge_pending": False, + } + c._apply_delta( + "node-a", + { + "type": "ws_state", + "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 + def test_apply_delta_ws_created(self): c = _make_collector() c._nodes["node-a"] = NodeSnapshot(node_id="node-a", server_url="http://a:8080") diff --git a/tests/test_coordinator_adapter.py b/tests/test_coordinator_adapter.py index edd88927..c1decbf3 100644 --- a/tests/test_coordinator_adapter.py +++ b/tests/test_coordinator_adapter.py @@ -558,3 +558,54 @@ 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).""" + 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._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 + + 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".""" + adapter, recorder, _ = self._setup() + with adapter._children_lock: + adapter._merge_child_ids_locked("coord-a", ["child-a1"]) + adapter._dispatch_child_event( + { + "type": "cluster_state", + "ws_id": "child-a1", + "state": "running", + "activity_state": "tool", + } + ) + assert len(recorder.enqueued) == 1 + payload = recorder.enqueued[0] + assert "pending_approval_detail" in payload + assert payload["pending_approval_detail"] is None diff --git a/tests/test_coordinator_page.py b/tests/test_coordinator_page.py index cf0c2d39..1bcc27a4 100644 --- a/tests/test_coordinator_page.py +++ b/tests/test_coordinator_page.py @@ -152,3 +152,95 @@ def test_coordinator_js_exposes_inline_approval_helpers(): # any prior denial. bug-1 / bug-3 from the second /review pass. assert "Denied by user" in body 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. + + 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 + codebase has no JS test framework, so locking the guard's shape + here is the next-best thing to a behavioral test.""" + import re + from pathlib import Path + + coord_js = Path(__file__).resolve().parent.parent / ( + "turnstone/console/static/coordinator/coordinator.js" + ) + 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). + assert "enteredApproval" not in body + assert "leftApproval" not in body + + # 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. + assert re.search( + r"sseUpdatedAt:\s*Date\.now\(\)", + body, + ), "handleChildState must write sseUpdatedAt: Date.now() onto liveBadgeCache entries" + + # flushLiveFetches' merge guard structure: SSE-set pending_approval + # / _detail wins over a stale bulk-poll snapshot when (live) AND + # (prev exists) AND (prev.sseUpdatedAt set) AND (within window) + # AND (prev.live exists). Inverting the comparison or dropping + # any of these guards reopens the clobber bug. + merge_guard = re.search( + r"if\s*\(\s*live\s*&&\s*prev\s*&&\s*prev\.sseUpdatedAt\s*&&\s*" + r"now\s*-\s*prev\.sseUpdatedAt\s*<\s*SSE_AUTHORITATIVE_MS\s*&&\s*" + r"prev\.live\s*\)", + body, + ) + assert merge_guard is not None, ( + "flushLiveFetches merge guard must be the conjunction " + "(live && prev && prev.sseUpdatedAt && now - prev.sseUpdatedAt < " + "SSE_AUTHORITATIVE_MS && prev.live). An inverted comparison or " + "missing prev.live check would let a stale bulk-poll clobber a " + "fresh SSE-set approval." + ) + + # The merge body must preserve BOTH pending_approval and + # pending_approval_detail from prev — preserving only one would + # render a row with a phantom badge but no buttons (or vice versa). + merge_body = re.search( + r"mergedLive\s*=\s*Object\.assign\(\s*\{\}\s*,\s*live\s*,\s*\{" + r"[^}]*pending_approval:\s*prev\.live\.pending_approval[^}]*" + r"pending_approval_detail:\s*prev\.live\.pending_approval_detail", + body, + ) + assert merge_body is not None, ( + "Merge body must preserve both pending_approval AND " + "pending_approval_detail from prev.live — preserving only one " + "creates a half-rendered approval row." + ) + + # flushLiveFetches must forward sseUpdatedAt onto the new cache + # entry so the SSE-source tag survives the bulk-poll write back — + # without this, every bulk-poll resets the window and the next + # late-arriving poll silently clobbers. + assert re.search( + r"sseUpdatedAt:\s*prev\s*\?\s*prev\.sseUpdatedAt", + body, + ), ( + "flushLiveFetches must forward prev.sseUpdatedAt onto the new " + "cache entry (preserving the SSE-source window across bulk-poll " + "cycles) — without this, the second bulk-poll after an SSE " + "transition silently clobbers." + ) diff --git a/tests/test_webui_content.py b/tests/test_webui_content.py index a8965db4..b426f1fe 100644 --- a/tests/test_webui_content.py +++ b/tests/test_webui_content.py @@ -155,3 +155,87 @@ class TestContentAccumulation: assert len(idle_events) == 1 # Content should be capped, not contain everything 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.""" + + 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") + + events = _drain_global() + running_events = [e for e in events if e.get("state") == "running"] + 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).""" + 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": [ + { + "call_id": "c1", + "header": "tool x", + "func_args": "{}", + "intent_summary": "do x", + "needs_approval": True, + } + ], + "judge_pending": False, + } + ui._broadcast_state("attention") + + 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" + + 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.""" + ui = _make_ui() + ui._pending_approval = { + "type": "approve_request", + "items": [{"call_id": "c1", "header": "x"}], + "judge_pending": False, + } + ui._broadcast_state("attention") + _drain_global() # discard the with-detail event + + ui._pending_approval = None + ui._broadcast_state("running") + events = _drain_global() + running = [e for e in events if e.get("state") == "running"] + assert len(running) == 1 + assert "pending_approval_detail" not in running[0] diff --git a/turnstone/console/collector.py b/turnstone/console/collector.py index ab71e932..512e2798 100644 --- a/turnstone/console/collector.py +++ b/turnstone/console/collector.py @@ -498,6 +498,7 @@ 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", "") @@ -557,6 +558,18 @@ 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", @@ -568,6 +581,7 @@ 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"), } ) diff --git a/turnstone/console/coordinator_adapter.py b/turnstone/console/coordinator_adapter.py index 952df337..204f6d9f 100644 --- a/turnstone/console/coordinator_adapter.py +++ b/turnstone/console/coordinator_adapter.py @@ -685,11 +685,13 @@ class CoordinatorAdapter: "tokens": event.get("tokens", 0), "node_id": event.get("node_id", ""), # activity_state lets the JS detect approval-state - # transitions and fire urgent live-bulk fetches so - # inline approve/deny buttons render in lockstep - # with the child entering attention (instead of - # waiting up to 5s for the next TTL window). + # 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": event.get("activity_state", ""), + "pending_approval_detail": event.get("pending_approval_detail"), } elif etype == "ws_closed": child_event = { diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js index 729a457a..27cec842 100644 --- a/turnstone/console/static/coordinator/coordinator.js +++ b/turnstone/console/static/coordinator/coordinator.js @@ -2085,6 +2085,14 @@ const TERMINAL_CHILD_STATES = new Set(["closed", "deleted"]); const LIVE_BADGE_TTL_MS = 5000; const LIVE_BADGE_DEBOUNCE_MS = 250; + // After handleChildState mutates liveBadgeCache from a child_ws_state + // SSE event, a bulk-poll landing within this window must NOT + // overwrite the SSE-supplied pending_approval / _detail fields with + // its own (potentially stale) snapshot — the upstream node + // /dashboard cache has its own ~2s TTL so a poll right after a + // transition can carry pre-transition state. 3s covers the worst + // case (upstream TTL + console TTL minus a margin). + const SSE_AUTHORITATIVE_MS = 3000; // Debounce window for /tasks refreshes triggered by ``tasks`` // tool_result SSE events. Without it, a model that runs // ``add → list`` (or any back-to-back mutation pair) double-fetches @@ -2192,9 +2200,11 @@ // 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 yet (urgent live-bulk fetch is in flight); the row gets - // re-rendered when it lands. + // 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) { const detail = cached.live.pending_approval_detail; const block = renderApprovalBlock(child, detail); @@ -3066,12 +3076,35 @@ ? results[id] : null; const wasDenied = denied.indexOf(id) !== -1; + const prev = liveBadgeCache.get(id); + // SSE-set pending_approval / _detail wins over a stale + // bulk-poll snapshot for SSE_AUTHORITATIVE_MS after the + // SSE update. Without this guard, a poll landing right + // after a child_ws_state transition can clobber freshly- + // mutated approval state with pre-transition data from + // the upstream /dashboard cache (which has its own ~2s + // TTL). Other fields (tokens, context_ratio) still track + // the bulk response — only the approval surface is gated. + let mergedLive = live; + if ( + live && + prev && + prev.sseUpdatedAt && + now - prev.sseUpdatedAt < SSE_AUTHORITATIVE_MS && + prev.live + ) { + mergedLive = Object.assign({}, live, { + pending_approval: prev.live.pending_approval, + pending_approval_detail: prev.live.pending_approval_detail, + }); + } liveBadgeCache.set(id, { - live: live, + live: mergedLive, fetched: now, // Denied ids are permission/identity misses — mark permanent // so SSE state ticks on those rows don't retry every window. permanent: wasDenied, + sseUpdatedAt: prev ? prev.sseUpdatedAt || 0 : 0, }); const row = childrenTreeEl.querySelector( '.ch-row[data-ws-id="' + cssEscape(id) + '"]', @@ -3092,10 +3125,12 @@ const isPermanent = e && /HTTP 403/.test(e.message || ""); const now = Date.now(); ids.forEach((id) => { + const prev = liveBadgeCache.get(id); liveBadgeCache.set(id, { live: null, fetched: now, permanent: isPermanent, + sseUpdatedAt: prev ? prev.sseUpdatedAt || 0 : 0, }); }); if (!isPermanent) console.warn("flushLiveFetches failed", e); @@ -3135,7 +3170,6 @@ ws_id: childId, name: "", }; - const prevActivity = existing.activity_state || ""; existing.state = ev.state || existing.state; existing.activity_state = typeof ev.activity_state === "string" @@ -3144,32 +3178,57 @@ if (ev.node_id) existing.node_id = ev.node_id; childrenState.set(childId, existing); _touchChild(childId); - renderChildren(); - // Do NOT invalidateLiveBadge on routine state ticks — that defeats - // the 5s TTL cache and devolves rate-limiting to the 250ms - // debouncer, hitting cluster_ws_detail ~4 req/s per chatty child. - // The TTL check in scheduleLiveFetch will refresh the badge on its - // own schedule; identity-changing events (created/rename/closed) - // still invalidate below. - // - // Two activity_state transitions warrant an *urgent* (TTL-bypassing) - // fetch so the row carries pending_approval_detail in lockstep with - // the child's true state: - // - "" / "tool" / "thinking" → "approval" (need rich payload now - // so the inline approve/deny buttons can render) - // - "approval" → anything else (need to drop the - // stale payload so the buttons disappear; without this the - // 5s TTL leaves stale buttons on a row whose approval was - // resolved elsewhere — e.g. the child's own UI tab) - const enteredApproval = - existing.activity_state === "approval" && prevActivity !== "approval"; - const leftApproval = - prevActivity === "approval" && existing.activity_state !== "approval"; - if (enteredApproval || leftApproval) { - scheduleLiveFetch(childId, { urgent: true }); + // pending_approval_detail rides on every ws_state event (see + // turnstone/server.py WebUI._broadcast_state) so 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; + 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 { - scheduleLiveFetch(childId); + nextDetail = null; } + const nextLive = Object.assign({}, (cached && cached.live) || {}, { + pending_approval: pendingApproval, + pending_approval_detail: nextDetail, + }); + liveBadgeCache.set(childId, { + live: nextLive, + // Preserve prior bulk-poll fetched timestamp so a fresh SSE + // tick doesn't artificially extend the 5s TTL gate in + // scheduleLiveFetch — the bulk-poll still drives slower- + // moving fields (tokens, context_ratio) on its own schedule. + fetched: cached ? cached.fetched : 0, + permanent: !!(cached && cached.permanent), + sseUpdatedAt: Date.now(), + }); + renderChildren(); + // Do NOT invalidateLiveBadge on routine state ticks — that + // defeats the 5s TTL cache and devolves rate-limiting to the + // 250ms debouncer. The TTL check in scheduleLiveFetch handles + // refresh cadence for slower-moving fields; identity-changing + // events (created/rename/closed) still invalidate below. + scheduleLiveFetch(childId); } function handleChildClosed(ev) { diff --git a/turnstone/server.py b/turnstone/server.py index 0b14159c..39d41121 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -194,6 +194,18 @@ 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 try: WebUI._global_queue.put_nowait(event) except queue.Full: @@ -823,6 +835,16 @@ def _build_node_snapshot(app_state: Any) -> dict[str, Any]: title = "" if ws.session: title = get_workstream_display_name(ws.session.ws_id) or "" + # ``pending_approval_detail`` mirrors the dashboard handler's + # projection so the console collector's reconnect-via-snapshot + # path (``_reconcile_node``) can carry the rich approval payload + # across reconnects — without it, a child sitting in approval- + # pending across a console restart or network blip would render + # with no buttons until the next state change. Same data, same + # ``read`` scope as ``/v1/api/dashboard``. + approval_detail: dict[str, Any] | None = None + if ui is not None and hasattr(ui, "serialize_pending_approval_detail"): + approval_detail = ui.serialize_pending_approval_detail() ws_list.append( { "id": ws.id, @@ -839,6 +861,7 @@ def _build_node_snapshot(app_state: Any) -> dict[str, Any]: "kind": ws.kind, "parent_ws_id": ws.parent_ws_id, "user_id": ws.user_id, + "pending_approval_detail": approval_detail, } ) return {