diff --git a/tests/_replay_helpers.py b/tests/_replay_helpers.py index d315811b..171100cf 100644 --- a/tests/_replay_helpers.py +++ b/tests/_replay_helpers.py @@ -51,6 +51,12 @@ def make_replay_mocks( ui._ws_messages = 0 for key, value in ui_overrides.items(): setattr(ui, key, value) + # Both replay paths read cycle cards via ``pending_approval_cards()`` + # (one card per concurrent approval cycle). Model it from the + # single-slot ``_pending_approval`` override so tests keep seeding + # the one field; a bare MagicMock here would iterate empty and + # silently drop the approve_request from the replay. + ui.pending_approval_cards = lambda: [ui._pending_approval] if ui._pending_approval else [] ws = MagicMock() ws.session = session request = MagicMock() diff --git a/tests/test_channel_discord.py b/tests/test_channel_discord.py index 3ce4f0b1..a20a3f6a 100644 --- a/tests/test_channel_discord.py +++ b/tests/test_channel_discord.py @@ -41,6 +41,11 @@ def _bind_ws_event_handlers(bot, cls): attr = getattr(cls, name) if callable(attr): setattr(bot, name, attr.__get__(bot, cls)) + # ``_handle_stream_end`` delegates the all-cycles sweep to + # ``_pop_ws_approvals``; bind the real method too so dispatcher + # tests observe the pop instead of a spec'd AsyncMock no-op. + if hasattr(cls, "_pop_ws_approvals"): + bot._pop_ws_approvals = cls._pop_ws_approvals.__get__(bot, cls) def _make_message(*, bot=False, guild=True, content="hello", channel=None, reference=None): @@ -537,7 +542,7 @@ class TestApprovalVerdictDisplay: }, } ] - event = ApproveRequestEvent(ws_id="ws-1", items=items) + event = ApproveRequestEvent(ws_id="ws-1", cycle_id="cyc-1", items=items) _run(bot._on_ws_event("ws-1", thread, event)) # thread.send was called with an embed containing a verdict field @@ -551,8 +556,8 @@ class TestApprovalVerdictDisplay: assert "HIGH" in field.value assert "85%" in field.value - # Pending approval message tracked - assert "ws-1" in bot._pending_approval_msgs + # Pending approval message tracked under (ws_id, cycle_id). + assert ("ws-1", "cyc-1") in bot._pending_approval_msgs def test_approval_without_verdict(self): """ApproveRequestEvent items without verdict still work normally.""" @@ -585,10 +590,11 @@ class TestApprovalVerdictDisplay: embed = MagicMock() msg.embeds = [embed] msg.edit = AsyncMock() - bot._pending_approval_msgs["ws-1"] = msg + bot._pending_approval_msgs[("ws-1", "cyc-1")] = (msg, frozenset({"c-1"})) event = IntentVerdictEvent( ws_id="ws-1", + call_id="c-1", func_name="bash", risk_level="high", recommendation="deny", @@ -628,7 +634,10 @@ class TestApprovalVerdictDisplay: bot._streaming = {} bot._thinking_msgs = {} bot._tool_info_msgs = {} - bot._pending_approval_msgs = {"ws-1": MagicMock()} + bot._pending_approval_msgs = { + ("ws-1", "cyc-1"): (MagicMock(), frozenset()), + ("ws-1", "cyc-2"): (MagicMock(), frozenset()), + } bot._notify_reply_channels = {} _bind_ws_event_handlers(bot, TurnstoneBot) @@ -636,7 +645,8 @@ class TestApprovalVerdictDisplay: event = StreamEndEvent(ws_id="ws-1") _run(bot._on_ws_event("ws-1", thread, event)) - assert "ws-1" not in bot._pending_approval_msgs + # ALL of the ws's cycles are swept, not just one entry. + assert not bot._pending_approval_msgs class TestStreamEndBehavior: @@ -1657,19 +1667,21 @@ class TestApprovalResolved: bot = self._make_bot() thread = AsyncMock() - # Set up a pending approval message with components. + # Set up a pending approval message with components. The event + # below carries no cycle_id (pre-multi-cycle server) — the + # legacy fallback clears the ws's single tracked entry. approval_msg = MagicMock() approval_msg.embeds = [MagicMock()] approval_msg.components = [] approval_msg.edit = AsyncMock() - bot._pending_approval_msgs["ws-1"] = approval_msg + bot._pending_approval_msgs[("ws-1", "cyc-1")] = (approval_msg, frozenset()) event = ApprovalResolvedEvent(ws_id="ws-1", approved=False, feedback="timeout") _run(bot._on_ws_event("ws-1", thread, event)) approval_msg.edit.assert_awaited_once() # Pending approval message should be removed. - assert "ws-1" not in bot._pending_approval_msgs + assert not bot._pending_approval_msgs def test_disables_buttons_on_approved(self): from turnstone.sdk.events import ApprovalResolvedEvent @@ -1681,9 +1693,11 @@ class TestApprovalResolved: approval_msg.embeds = [MagicMock()] approval_msg.components = [] approval_msg.edit = AsyncMock() - bot._pending_approval_msgs["ws-1"] = approval_msg + bot._pending_approval_msgs[("ws-1", "cyc-1")] = (approval_msg, frozenset()) - event = ApprovalResolvedEvent(ws_id="ws-1", approved=True) + # Cycle-routed resolution: the event's cycle_id selects exactly + # this tracked message. + event = ApprovalResolvedEvent(ws_id="ws-1", approved=True, cycle_id="cyc-1") _run(bot._on_ws_event("ws-1", thread, event)) approval_msg.edit.assert_awaited_once() diff --git a/tests/test_channel_routing.py b/tests/test_channel_routing.py index c1986c3f..5d569552 100644 --- a/tests/test_channel_routing.py +++ b/tests/test_channel_routing.py @@ -87,7 +87,7 @@ class TestSendApproval: monkeypatch.setattr(router._server, "approve", mock_approve) await router.send_approval("ws-1", "corr-abc", approved=True, feedback="ok") mock_approve.assert_awaited_once_with( - ws_id="ws-1", approved=True, feedback="ok", always=False + ws_id="ws-1", approved=True, feedback="ok", always=False, cycle_id="corr-abc" ) @pytest.mark.anyio @@ -99,7 +99,7 @@ class TestSendApproval: monkeypatch.setattr(router._server, "approve", mock_approve) await router.send_approval("ws-1", "corr-abc", approved=False) mock_approve.assert_awaited_once_with( - ws_id="ws-1", approved=False, feedback=None, always=False + ws_id="ws-1", approved=False, feedback=None, always=False, cycle_id="corr-abc" ) @pytest.mark.anyio @@ -110,7 +110,9 @@ class TestSendApproval: mock_approve = AsyncMock() monkeypatch.setattr(console_router._console, "route_approve", mock_approve) await console_router.send_approval("ws-1", "corr-abc", approved=True, always=True) - mock_approve.assert_awaited_once_with(ws_id="ws-1", approved=True, feedback="", always=True) + mock_approve.assert_awaited_once_with( + ws_id="ws-1", approved=True, feedback="", always=True, cycle_id="corr-abc" + ) class TestDeleteRoute: diff --git a/tests/test_channel_slack.py b/tests/test_channel_slack.py index 120ebd9c..a01cf07f 100644 --- a/tests/test_channel_slack.py +++ b/tests/test_channel_slack.py @@ -576,10 +576,11 @@ class TestApprovalOwnership: bot, router, client = _make_bot() ws_id = "ws-1" - bot._pending_approval[ws_id] = PendingApproval( # type: ignore[attr-defined] + bot._pending_approval[(ws_id, "corr-1")] = PendingApproval( # type: ignore[attr-defined] channel="C01SAPU5414", message_ts="111.222", owner_user_id="U_OWNER", + cycle_id="corr-1", ) body = { @@ -598,10 +599,11 @@ class TestApprovalOwnership: bot, router, client = _make_bot() ws_id = "ws-1" - bot._pending_approval[ws_id] = PendingApproval( # type: ignore[attr-defined] + bot._pending_approval[(ws_id, "corr-1")] = PendingApproval( # type: ignore[attr-defined] channel="C01SAPU5414", message_ts="111.222", owner_user_id="U_OWNER", + cycle_id="corr-1", ) body = { @@ -620,10 +622,11 @@ class TestApprovalOwnership: bot, router, client = _make_bot() ws_id = "ws-1" - bot._pending_approval[ws_id] = PendingApproval( # type: ignore[attr-defined] + bot._pending_approval[(ws_id, "corr-1")] = PendingApproval( # type: ignore[attr-defined] channel="C01SAPU5414", message_ts="111.222", owner_user_id="U_OWNER", + cycle_id="corr-1", ) body = { @@ -776,7 +779,9 @@ class TestWsEventDispatch: bot, client = self._make_ws_bot() event = ApproveRequestEvent( - ws_id="ws-1", items=[{"func_name": "bash", "needs_approval": True}] + ws_id="ws-1", + cycle_id="cyc-1", + items=[{"call_id": "c-1", "func_name": "bash", "needs_approval": True}], ) route = SlackRoute(channel="C1", user_id="U12345", thread_ts="123.456") _run(bot._on_ws_event("ws-1", route, event)) # type: ignore[attr-defined] @@ -784,8 +789,12 @@ class TestWsEventDispatch: client.chat_postMessage.assert_awaited_once() call_kwargs = client.chat_postMessage.call_args[1] assert "blocks" in call_kwargs - assert "ws-1" in bot._pending_approval # type: ignore[attr-defined] - assert bot._pending_approval["ws-1"].owner_user_id == "U12345" # type: ignore[attr-defined] + # Tracked under (ws_id, cycle_id) so concurrent cycles each get + # their own Slack message. + entry = bot._pending_approval[("ws-1", "cyc-1")] # type: ignore[attr-defined] + assert entry.owner_user_id == "U12345" + assert entry.cycle_id == "cyc-1" + assert entry.call_ids == frozenset({"c-1"}) def test_intent_verdict_updates_approval_message(self) -> None: from turnstone.channels.slack.bot import PendingApproval @@ -797,14 +806,17 @@ class TestWsEventDispatch: return_value={"ok": True, "messages": [{"blocks": []}]} ) - bot._pending_approval["ws-1"] = PendingApproval( # type: ignore[attr-defined] + bot._pending_approval[("ws-1", "cyc-1")] = PendingApproval( # type: ignore[attr-defined] channel="C1", message_ts="999.000", owner_user_id="U12345", + cycle_id="cyc-1", + call_ids=frozenset({"c-1"}), ) event = IntentVerdictEvent( ws_id="ws-1", + call_id="c-1", func_name="bash", risk_level="high", confidence=0.9, @@ -821,17 +833,20 @@ class TestWsEventDispatch: from turnstone.sdk.events import ApprovalResolvedEvent bot, client = self._make_ws_bot() - bot._pending_approval["ws-1"] = PendingApproval( # type: ignore[attr-defined] + bot._pending_approval[("ws-1", "cyc-9")] = PendingApproval( # type: ignore[attr-defined] channel="C1", message_ts="999.000", owner_user_id="U12345", + cycle_id="cyc-9", ) + # Event WITHOUT a cycle_id (pre-multi-cycle server): the legacy + # fallback clears the ws's single tracked entry, as before. event = ApprovalResolvedEvent(ws_id="ws-1", approved=True) route = SlackRoute(channel="C1", user_id="U12345", thread_ts="123.456") _run(bot._on_ws_event("ws-1", route, event)) # type: ignore[attr-defined] - assert "ws-1" not in bot._pending_approval # type: ignore[attr-defined] + assert not bot._pending_approval # type: ignore[attr-defined] client.chat_update.assert_awaited_once() def test_link_prefix_does_not_hijack_regular_prompt(self) -> None: diff --git a/tests/test_coord_ui_approve_tools.py b/tests/test_coord_ui_approve_tools.py index 5b37db29..e491a4a7 100644 --- a/tests/test_coord_ui_approve_tools.py +++ b/tests/test_coord_ui_approve_tools.py @@ -525,12 +525,16 @@ class TestBroadcastApprovalResolved: collector = MagicMock() ConsoleCoordinatorUI._collector = collector try: - ui._broadcast_approval_resolved(True, "lgtm", always=True) + ui._broadcast_approval_resolved( + True, "lgtm", always=True, cycle_id="cyc-1", call_ids=("c-1", "c-2") + ) collector.emit_console_ws_approval_resolved.assert_called_once_with( "coord-a", approved=True, feedback="lgtm", always=True, + cycle_id="cyc-1", + call_ids=["c-1", "c-2"], ) finally: ConsoleCoordinatorUI._collector = None @@ -546,6 +550,8 @@ class TestBroadcastApprovalResolved: approved=False, feedback="", always=False, + cycle_id="", + call_ids=[], ) finally: ConsoleCoordinatorUI._collector = None diff --git a/tests/test_coordinator_adapter.py b/tests/test_coordinator_adapter.py index 74c4e6f0..6a0fb502 100644 --- a/tests/test_coordinator_adapter.py +++ b/tests/test_coordinator_adapter.py @@ -195,6 +195,21 @@ def test_emit_tolerates_collector_exception() -> None: # --------------------------------------------------------------------------- +def test_cleanup_ui_sweeps_all_approval_cycles_on_registry_uis() -> None: + """The real ConsoleCoordinatorUI carries the approval-cycle + registry: cleanup denies + wakes EVERY parked gate via + ``resolve_all_approvals`` (parallel task agents can hold several), + not the pre-cycle single-slot kick.""" + adapter, _ = _make_adapter() + ws = _make_ws() + ws.ui.resolve_all_approvals = MagicMock(return_value=2) # type: ignore[attr-defined] + adapter.cleanup_ui(ws) + ws.ui.resolve_all_approvals.assert_called_once_with( # type: ignore[attr-defined] + False, "Workstream closed" + ) + assert ws.ui._fg_event.is_set() # type: ignore[attr-defined] + + def test_cleanup_ui_unblocks_events_and_broadcasts_to_listeners() -> None: adapter, _ = _make_adapter() ws = _make_ws() diff --git a/tests/test_coordinator_endpoints.py b/tests/test_coordinator_endpoints.py index fa5c3484..860d6124 100644 --- a/tests/test_coordinator_endpoints.py +++ b/tests/test_coordinator_endpoints.py @@ -1103,18 +1103,7 @@ def test_approve_resolves_ui_event(storage): mgr = _build_mgr(storage) ws = mgr.create(user_id="user-1") assert isinstance(ws.ui, ConsoleCoordinatorUI) - ws.ui._pending_approval = { - "type": "approve_request", - "items": [ - { - "call_id": "c-1", - "func_name": "spawn_workstream", - "approval_label": "spawn_workstream", - "needs_approval": True, - } - ], - } - ws.ui._approval_event.clear() + cycle = _seed_pending(ws, "c-1") client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) resp = client.post( f"/v1/api/workstreams/{ws.id}/approve", @@ -1122,34 +1111,46 @@ def test_approve_resolves_ui_event(storage): headers=_COORD_HEADERS, ) assert resp.status_code == 200 - assert ws.ui._approval_event.is_set() - assert ws.ui._approval_result == (True, None) + assert resp.json()["cycle_id"] == cycle.cycle_id + assert cycle.event.is_set() + assert cycle.result == (True, None) assert "spawn_workstream" in ws.ui.auto_approve_tools -def _seed_pending(ws, *call_ids: str) -> None: - ws.ui._pending_approval = { +def _seed_pending(ws, *call_ids: str, func_name: str = "spawn_workstream"): + """Register a live ApprovalCycle on the coord UI the way its + ``approve_tools`` gate does, returning the cycle for direct + event/result assertions (the pre-cycle singleton + ``_approval_event`` / ``_approval_result`` slots are gone).""" + from turnstone.core.session_ui_base import ApprovalCycle + + items = [ + { + "call_id": cid, + "func_name": func_name, + "approval_label": func_name, + "needs_approval": True, + } + for cid in call_ids + ] + card = { "type": "approve_request", - "items": [ - { - "call_id": cid, - "func_name": "spawn_workstream", - "approval_label": "spawn_workstream", - "needs_approval": True, - } - for cid in call_ids - ], + "cycle_id": f"cyc-{'-'.join(call_ids)}", + "items": ws.ui._serialize_approval_items(items), + "judge_pending": False, } - ws.ui._approval_event.clear() + cycle = ApprovalCycle(items, card, None) + ws.ui._register_approval_cycle(cycle) + return cycle def test_approve_409_on_stale_call_id(storage): """Body call_id doesn't match any pending item → 409 with the - current primary call_id so the UI can re-render against the - new round.""" + current primary call_id + cycle_id so the UI can re-render + against the new round.""" mgr = _build_mgr(storage) ws = mgr.create(user_id="user-1") - _seed_pending(ws, "c-current") + cycle = _seed_pending(ws, "c-current") client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) resp = client.post( f"/v1/api/workstreams/{ws.id}/approve", @@ -1160,17 +1161,17 @@ def test_approve_409_on_stale_call_id(storage): body = resp.json() assert body["error"] == "stale call_id" assert body["current_call_id"] == "c-current" - # Approval event must NOT be set — no resolve_approval ran. - assert not ws.ui._approval_event.is_set() + assert body["current_cycle_id"] == cycle.cycle_id + # The live cycle must NOT have been resolved. + assert not cycle.event.is_set() def test_approve_409_when_no_pending_and_call_id_sent(storage): - """Body sends a call_id but the UI has no pending approval — - 409 with current_call_id=None so the UI knows to clear the row.""" + """Body sends a call_id but the UI has no live cycle — 409 with + current_call_id=None so the UI knows to clear the row.""" mgr = _build_mgr(storage) ws = mgr.create(user_id="user-1") - # No _pending_approval seeded → ui._pending_approval is None. - ws.ui._approval_event.clear() + # No cycle registered. client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) resp = client.post( f"/v1/api/workstreams/{ws.id}/approve", @@ -1179,18 +1180,18 @@ def test_approve_409_when_no_pending_and_call_id_sent(storage): ) assert resp.status_code == 409 body = resp.json() - assert body["error"] == "no pending approval" + assert body["error"] == "stale call_id" assert body["current_call_id"] is None - assert not ws.ui._approval_event.is_set() + assert body["current_cycle_id"] is None def test_approve_no_call_id_preserves_backward_compat(storage): """Existing clients (CLI, channel adapters) that omit call_id - must still resolve approvals — the guard only kicks in when - call_id is present in the body.""" + must still resolve approvals — a selector-less body lands on the + oldest live cycle.""" mgr = _build_mgr(storage) ws = mgr.create(user_id="user-1") - _seed_pending(ws, "c-1") + cycle = _seed_pending(ws, "c-1") client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) resp = client.post( f"/v1/api/workstreams/{ws.id}/approve", @@ -1198,18 +1199,18 @@ def test_approve_no_call_id_preserves_backward_compat(storage): headers=_COORD_HEADERS, ) assert resp.status_code == 200 - assert ws.ui._approval_event.is_set() + assert resp.json()["cycle_id"] == cycle.cycle_id + assert cycle.event.is_set() -def test_approve_no_call_id_no_pending_falls_through(storage): - """Legacy clients (no call_id) calling approve when pending is - None hit the existing resolve_approval no-op path — the new - guard must not change that behavior. Regression guard for the - legacy code path that the call_id check intentionally bypasses.""" +def test_approve_no_call_id_no_pending_resolves_nothing(storage): + """Legacy clients (no call_id) calling approve with no live cycle: + 200 with ``cycle_id: null`` — the handler resolves NOTHING rather + than racing a cycle that registers between its lookup and its + resolve (the client can't have been looking at one).""" mgr = _build_mgr(storage) ws = mgr.create(user_id="user-1") - ws.ui._approval_event.clear() - # No _pending_approval seeded. + # No cycle registered. client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) resp = client.post( f"/v1/api/workstreams/{ws.id}/approve", @@ -1217,7 +1218,7 @@ def test_approve_no_call_id_no_pending_falls_through(storage): headers=_COORD_HEADERS, ) assert resp.status_code == 200 - assert ws.ui._approval_event.is_set() + assert resp.json()["cycle_id"] is None def test_approve_call_id_matches_any_item_in_multi_envelope(storage): @@ -1226,7 +1227,7 @@ def test_approve_call_id_matches_any_item_in_multi_envelope(storage): one-boolean semantics of resolve_approval.""" mgr = _build_mgr(storage) ws = mgr.create(user_id="user-1") - _seed_pending(ws, "c-1", "c-2", "c-3") + cycle = _seed_pending(ws, "c-1", "c-2", "c-3") client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) resp = client.post( f"/v1/api/workstreams/{ws.id}/approve", @@ -1234,7 +1235,61 @@ def test_approve_call_id_matches_any_item_in_multi_envelope(storage): headers=_COORD_HEADERS, ) assert resp.status_code == 200 - assert ws.ui._approval_event.is_set() + assert cycle.event.is_set() + + +def test_selectorless_always_whitelists_only_the_resolved_oldest_cycle(storage): + """sweep-3 regression: with several live cycles, a selector-less + "Approve + Always" must whitelist the tools of the cycle it + actually resolved (the oldest) — not a sibling's.""" + mgr = _build_mgr(storage) + ws = mgr.create(user_id="user-1") + oldest = _seed_pending(ws, "a-1", func_name="spawn_workstream") + newer = _seed_pending(ws, "b-1", func_name="send_message") + client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) + resp = client.post( + f"/v1/api/workstreams/{ws.id}/approve", + json={"approved": True, "always": True}, # no selector + headers=_COORD_HEADERS, + ) + assert resp.status_code == 200 + assert resp.json()["cycle_id"] == oldest.cycle_id + assert oldest.event.is_set() + assert not newer.event.is_set() + assert "spawn_workstream" in ws.ui.auto_approve_tools + assert "send_message" not in ws.ui.auto_approve_tools + + +def test_approve_always_skips_whitelist_when_pinned_cycle_lost_the_race(storage): + """sweep-3 regression: the handler collects always-names from the + cycle its lookup pinned; if that cycle is resolved by someone else + (gate timeout, peer tab) between lookup and resolve, the whitelist + must NOT grow — approving a card that already resolved must not + auto-approve anything.""" + mgr = _build_mgr(storage) + ws = mgr.create(user_id="user-1") + _seed_pending(ws, "a-1", func_name="spawn_workstream") + ui = ws.ui + real_find = ui.find_approval_cycle + + def racing_find(**kwargs): + card = real_find(**kwargs) + if card is not None: + # A concurrent resolver wins the gap between the handler's + # lookup and its (pinned) resolve. + ui.resolve_approval(False, "raced", cycle_id=card["cycle_id"]) + return card + + ui.find_approval_cycle = racing_find + client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) + resp = client.post( + f"/v1/api/workstreams/{ws.id}/approve", + json={"approved": True, "always": True}, + headers=_COORD_HEADERS, + ) + assert resp.status_code == 200 + assert resp.json()["cycle_id"] is None + assert "spawn_workstream" not in ws.ui.auto_approve_tools # --------------------------------------------------------------------------- @@ -1521,15 +1576,19 @@ def test_export_404_when_kind_interactive(storage): def test_cancel_resolves_pending_approval(storage): + """Cancel addresses the workstream, not one batch — EVERY live + cycle resolves (parallel task agents can hold several gates).""" mgr = _build_mgr(storage) ws = mgr.create(user_id="user-1") assert isinstance(ws.ui, ConsoleCoordinatorUI) - ws.ui._pending_approval = {"type": "approve_request", "items": []} - ws.ui._approval_event.clear() + first = _seed_pending(ws, "c-1") + second = _seed_pending(ws, "c-2") client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) resp = client.post(f"/v1/api/workstreams/{ws.id}/cancel", headers=_COORD_HEADERS) assert resp.status_code == 200 - assert ws.ui._approval_event.is_set() + assert first.event.is_set() + assert second.event.is_set() + assert first.result == (False, "Cancelled by user") def test_cancel_response_always_includes_dropped_key(storage): @@ -2399,6 +2458,7 @@ def test_cluster_inspect_node_backed_pending_approval_detail_passes_through(stor ws_id = "f0" * 16 _seed_node_workstream(storage, ws_id=ws_id, node_id="node-a") detail = { + "cycle_id": "cyc-bash", "call_id": "c-bash", "judge_pending": False, "items": [ @@ -2427,7 +2487,7 @@ def test_cluster_inspect_node_backed_pending_approval_detail_passes_through(stor "activity_state": "approval", "activity": "awaiting approval", "tokens": 100, - "pending_approval_detail": detail, + "pending_approval_details": [detail], } ] } @@ -2438,7 +2498,7 @@ def test_cluster_inspect_node_backed_pending_approval_detail_passes_through(stor assert resp.status_code == 200 live = resp.json()["live"] assert live["pending_approval"] is True # derived bool, existing behavior - assert live["pending_approval_detail"] == detail # full payload, new behavior + assert live["pending_approval_details"] == [detail] # full payload passthrough def test_cluster_inspect_node_backed_pending_approval_synthesized(storage): diff --git a/tests/test_coordinator_page.py b/tests/test_coordinator_page.py index 6ff919dc..4f202a72 100644 --- a/tests/test_coordinator_page.py +++ b/tests/test_coordinator_page.py @@ -313,17 +313,17 @@ def test_coordinator_js_handle_child_state_no_longer_reads_sse_pending_approval_ ) # The merge body must preserve BOTH pending_approval and - # pending_approval_detail from prev — preserving only one would + # pending_approval_details 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", + r"pending_approval_details:\s*prev\.live\.pending_approval_details", body, ) assert merge_body is not None, ( "Merge body must preserve both pending_approval AND " - "pending_approval_detail from prev.live — preserving only one " + "pending_approval_details from prev.live — preserving only one " "creates a half-rendered approval row." ) diff --git a/tests/test_phase6_endpoints.py b/tests/test_phase6_endpoints.py index 162908f6..f41d26d0 100644 --- a/tests/test_phase6_endpoints.py +++ b/tests/test_phase6_endpoints.py @@ -228,34 +228,42 @@ def test_bulk_live_coordinator_row_uses_manager_snapshot(storage): live = body["results"][ws.id] assert live is not None assert "pending_approval" in live - # New field always present on the wire — None when no approval - # is pending so the JS can `key in row` without surprise. - assert "pending_approval_detail" in live - assert live["pending_approval_detail"] is None + # The details list is always present on the wire — empty when no + # approval is pending so the JS can `key in row` without surprise. + # Replaces 1.6's singular ``pending_approval_detail`` null + # (breaking, 1.7). + assert "pending_approval_details" in live + assert live["pending_approval_details"] == [] -def test_bulk_live_coordinator_row_includes_pending_approval_detail(storage): - """When _pending_approval is set on a coord UI, the live block - surfaces the merged items + judge_verdict payload through the - coord-pseudo-node path. End-to-end equivalent of the dashboard - test in test_server_authz, but for the console live-bulk - endpoint that the coord tree UI actually consumes.""" +def test_bulk_live_coordinator_row_includes_pending_approval_details(storage): + """When an approval cycle is live on a coord UI, the live block + surfaces one detail entry per cycle with merged items + + judge_verdict through the coord-pseudo-node path. End-to-end + equivalent of the dashboard test in test_server_authz, but for + the console live-bulk endpoint that the coord tree UI actually + consumes.""" + from turnstone.core.session_ui_base import ApprovalCycle + mgr = _build_mgr(storage) ws = mgr.create(user_id="user-1") - ws.ui._pending_approval = { + items = [ + { + "call_id": "c-99", + "header": "spawn_workstream", + "preview": "{...}", + "func_name": "spawn_workstream", + "approval_label": "spawn_workstream", + "needs_approval": True, + } + ] + card = { "type": "approve_request", - "items": [ - { - "call_id": "c-99", - "header": "spawn_workstream", - "preview": "{...}", - "func_name": "spawn_workstream", - "approval_label": "spawn_workstream", - "needs_approval": True, - } - ], + "cycle_id": "cyc-99", + "items": ws.ui._serialize_approval_items(items), "judge_pending": False, } + ws.ui._register_approval_cycle(ApprovalCycle(items, card, None)) ws.ui._llm_verdicts["c-99"] = { "recommendation": "approve", "risk_level": "low", @@ -269,8 +277,10 @@ def test_bulk_live_coordinator_row_includes_pending_approval_detail(storage): assert resp.status_code == 200 live = resp.json()["results"][ws.id] assert live["pending_approval"] is True # boolean derived flag - detail = live["pending_approval_detail"] - assert detail is not None + details = live["pending_approval_details"] + assert len(details) == 1 + detail = details[0] + assert detail["cycle_id"] == "cyc-99" assert detail["call_id"] == "c-99" assert detail["items"][0]["func_name"] == "spawn_workstream" assert detail["items"][0]["judge_verdict"]["recommendation"] == "approve" diff --git a/tests/test_server_authz.py b/tests/test_server_authz.py index caf60797..5381aad3 100644 --- a/tests/test_server_authz.py +++ b/tests/test_server_authz.py @@ -88,18 +88,19 @@ class _FakeUI: self._ws_turn_tool_calls = 0 self._llm_verdicts: dict[str, dict[str, Any]] = {} - def serialize_pending_approval_detail(self) -> dict[str, Any] | None: - # Mirrors SessionUIBase.serialize_pending_approval_detail — + def serialize_pending_approval_details(self) -> list[dict[str, Any]]: + # Mirrors SessionUIBase.serialize_pending_approval_details — # the fake is monkeypatched in for ``WebUI`` and the dashboard # handler reads this method during projection. Real subclasses - # inherit from ``SessionUIBase``; the fake replicates the - # shape directly to stay decoupled. + # iterate their approval-cycle registry (one entry per live + # cycle); the fake models a single slot, so the list carries + # zero or one entries. pending = self._pending_approval if pending is None: - return None + return [] items = pending.get("items") or [] if not items: - return None + return [] call_ids = [item.get("call_id", "") for item in items] # Match the real impl's pattern (session_ui_base.py): snapshot # references under the lock, copy after release. Writers only @@ -133,11 +134,14 @@ class _FakeUI: # fake here keeps test-vs-prod behavioural drift from # masking a real-shape regression. primary = next((cid for cid in call_ids if cid), "") - return { - "call_id": primary, - "judge_pending": bool(pending.get("judge_pending", False)), - "items": serialized, - } + return [ + { + "cycle_id": pending.get("cycle_id", ""), + "call_id": primary, + "judge_pending": bool(pending.get("judge_pending", False)), + "items": serialized, + } + ] def serialize_recent_auto_approvals(self) -> list[dict[str, Any]]: # Empty buffer for tests that don't exercise the auto-approve @@ -671,28 +675,35 @@ class TestDashboardTrustedTeamVisibility: owners = {w["user_id"] for w in data["workstreams"]} assert {"user-a", "user-b"}.issubset(owners) - def test_dashboard_pending_approval_detail_default_none(self, app_client): - """No pending approval → field is explicitly null on the wire so - consumers can distinguish "not present" from "absent key".""" + def test_dashboard_pending_approval_details_default_empty(self, app_client): + """No pending approval → the list field is explicitly empty on + the wire so consumers can distinguish "nothing pending" from + "absent key". Replaces 1.6's singular ``pending_approval_detail`` + null (breaking, 1.7).""" client, _mgr = app_client client.post("/v1/api/workstreams/new", json={"name": "a"}, headers=_auth("user-a")) resp = client.get("/v1/api/dashboard", headers=_auth("user-a")) assert resp.status_code == 200 rows = resp.json()["workstreams"] assert len(rows) == 1 - assert "pending_approval_detail" in rows[0] - assert rows[0]["pending_approval_detail"] is None + assert "pending_approval_details" in rows[0] + assert rows[0]["pending_approval_details"] == [] + # The 1.6 singular field is GONE, not null — a consumer still + # reading it should break loudly, not read None forever. + assert "pending_approval_detail" not in rows[0] - def test_dashboard_pending_approval_detail_merges_judge_verdict(self, app_client): + def test_dashboard_pending_approval_details_merge_judge_verdict(self, app_client): """When _pending_approval is set on a ws's UI, /dashboard - embeds the merged items + judge_verdict so coord live-bulk - callers can render inline approve/deny buttons.""" + embeds one detail entry per live cycle with merged items + + judge_verdict so coord live-bulk callers can render inline + approve/deny buttons.""" client, mgr = app_client client.post("/v1/api/workstreams/new", json={"name": "a"}, headers=_auth("user-a")) ws_id = next(iter(mgr.list_all())).id ui = mgr.get(ws_id).ui ui._pending_approval = { "type": "approve_request", + "cycle_id": "cyc-1", "items": [ { "call_id": "c-1", @@ -714,8 +725,10 @@ class TestDashboardTrustedTeamVisibility: resp = client.get("/v1/api/dashboard", headers=_auth("user-a")) assert resp.status_code == 200 row = next(w for w in resp.json()["workstreams"] if w["ws_id"] == ws_id) - detail = row["pending_approval_detail"] - assert detail is not None + details = row["pending_approval_details"] + assert len(details) == 1 + detail = details[0] + assert detail["cycle_id"] == "cyc-1" assert detail["call_id"] == "c-1" assert detail["judge_pending"] is False item = detail["items"][0] diff --git a/tests/test_session.py b/tests/test_session.py index e7e29278..23d82c35 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -798,6 +798,89 @@ class TestTaskExec: captured[0](fake_verdict) # must not raise session.ui.on_intent_verdict.assert_not_called() + def test_evaluate_intent_agent_gate_owns_generation_off_the_main_slot( + self, tmp_db, monkeypatch + ) -> None: + """Sub-agent gates run the SAME judge pipeline as the main loop + but as their OWN generation (release blocker #1: task_agent + calls used to reach the gate judge-blind). The main-loop + supersede slot stays untouched — with parallel task agents, + publishing into it would make every sibling's verdicts look + stale to the previous sibling's callback — while the generation + is stamped on the items for the UI's origin checks, registered + for ``close()``'s sweep, delivered alongside the verdict, and + grounded on the SUB-AGENT's trajectory (its task prompt is the + delegation contract), not the parent conversation.""" + import threading + + from turnstone.core.session_ui_base import SessionUIBase + from turnstone.core.trajectory import turns_from_dicts + + class _GateUI(SessionUIBase): + pass + + session = _make_session() + ui = _GateUI(ws_id="ws-gate", user_id="u1") + ui.on_intent_verdict = MagicMock() # shadow: capture delivery kwargs + session.ui = ui + + captured: dict[str, Any] = {} + fake_verdict = MagicMock() + fake_verdict.to_dict.return_value = {"verdict_id": "v0", "call_id": "c1", "tier": "llm"} + fake_judge = MagicMock() + + def _eval(items, convo, **kw): + captured["convo"] = convo + captured["callback"] = kw.get("callback") + captured["cancel_event"] = kw.get("cancel_event") + captured["done"] = kw.get("done_callback") + return [fake_verdict] * len(items) + + fake_judge.evaluate.side_effect = _eval + monkeypatch.setattr(session, "_ensure_judge", lambda: fake_judge) + + main_slot = threading.Event() + session._judge_cancel_event = main_slot + agent_turns = turns_from_dicts([{"role": "user", "content": "Task: reindex the docs tree"}]) + item = {"call_id": "c1", "func_name": "bash", "needs_approval": True, "command": "ls"} + + ev = session._evaluate_intent([item], conversation=agent_turns, agent_gate=True) + + assert ev is not None and ev is not main_slot + # Main-loop slot untouched by the sub-agent spawn. + assert session._judge_cancel_event is main_slot + # Generation stamped for the UI's origin checks + close() sweep, + # and handed to the daemon as its cancel event. + assert item["_judge_event"] is ev + assert ev in session._judge_cancel_events + assert captured["cancel_event"] is ev + # Judge grounded on the sub-agent trajectory, not session.messages. + assert any("reindex the docs tree" in str(m) for m in captured["convo"]) + # Delivery rides the generation into the UI. + captured["callback"](fake_verdict) + assert ui.on_intent_verdict.call_args.kwargs.get("judge_event") is ev + # Daemon completion keeps the close()-sweep set exact. + captured["done"]() + assert ev not in session._judge_cancel_events + + def test_close_fires_agent_gate_judge_generations(self, tmp_db, monkeypatch) -> None: + """``close()`` aborts EVERY in-flight judge daemon — including + sub-agent generations that never touched the main slot — so a + torn-down session can't leave daemons running against a dead + UI.""" + session = _make_session() + fake_verdict = MagicMock() + fake_verdict.to_dict.return_value = {"verdict_id": "v0", "call_id": "c1", "tier": "llm"} + fake_judge = MagicMock() + fake_judge.evaluate.side_effect = lambda items, *_a, **_kw: [fake_verdict] * len(items) + monkeypatch.setattr(session, "_ensure_judge", lambda: fake_judge) + + item = {"call_id": "c1", "func_name": "bash", "needs_approval": True, "command": "ls"} + ev = session._evaluate_intent([item], conversation=[], agent_gate=True) + assert ev is not None and not ev.is_set() + session.close() + assert ev.is_set() + def _drive_gate(self, session, monkeypatch, *, cancel_on_approval: bool): """Run one needs_approval bash item through ``_execute_tools`` with a stubbed judge + approval gate; return the cancel event the judge diff --git a/tests/test_session_ui_base.py b/tests/test_session_ui_base.py index 6779dc2f..86edcee7 100644 --- a/tests/test_session_ui_base.py +++ b/tests/test_session_ui_base.py @@ -13,8 +13,11 @@ in its own test files. from __future__ import annotations +import contextlib import queue import threading +import time +from collections.abc import Callable, Iterator from typing import Any from unittest.mock import MagicMock, patch @@ -93,22 +96,67 @@ def test_enqueue_tolerates_full_listener_queue() -> None: # --------------------------------------------------------------------------- -def test_resolve_approval_sets_result_and_unblocks_event() -> None: +def _register_cycle( + ui: SessionUIBase, + call_ids: list[str], + *, + judge_event: object | None = None, +) -> Any: + """Register a live ApprovalCycle the way ``approve_tools`` does. + + Builds the card from wire-shaped items so serializer tests see the + exact production shape, registers under the lock, and returns the + cycle for direct assertions. + """ + from turnstone.core.session_ui_base import ApprovalCycle + + items = [ + {"call_id": cid, "func_name": "bash", "approval_label": "bash", "needs_approval": True} + for cid in call_ids + ] + card: dict[str, Any] = { + "type": "approve_request", + "cycle_id": f"cycle-{'-'.join(call_ids)}", + "items": ui._serialize_approval_items(items), + "judge_pending": False, + } + cycle = ApprovalCycle(items, card, judge_event) + ui._register_approval_cycle(cycle) + return cycle + + +def test_resolve_approval_sets_result_and_unblocks_cycle_event() -> None: ui = _make_ui() - ui._approval_event.clear() - ui.resolve_approval(True, "looks good") - assert ui._approval_result == (True, "looks good") - assert ui._approval_event.is_set() + cycle = _register_cycle(ui, ["c1"]) + resolved = ui.resolve_approval(True, "looks good") + assert resolved == cycle.cycle_id + assert cycle.result == (True, "looks good") + assert cycle.resolved is True + assert cycle.event.is_set() + + +def test_resolve_approval_without_live_cycle_is_a_noop() -> None: + """No live cycle → nothing to resolve: returns None and broadcasts + nothing (the old singleton overwrote a shared result slot and + leaked a stale ``approval_resolved`` event on idle cancels).""" + ui = _make_ui() + lq = ui._register_listener() + assert ui.resolve_approval(True, "nobody asked") is None + assert lq.empty() def test_resolve_approval_broadcasts_approval_resolved() -> None: ui = _make_ui() + cycle = _register_cycle(ui, ["c1"]) lq = ui._register_listener() ui.resolve_approval(False, "nope") event = lq.get_nowait() assert event["type"] == "approval_resolved" assert event["approved"] is False assert event["feedback"] == "nope" + # Cycle identity rides the event so clients dismiss the RIGHT card. + assert event["cycle_id"] == cycle.cycle_id + assert event["call_ids"] == ["c1"] # --------------------------------------------------------------------------- @@ -156,26 +204,54 @@ def test_on_intent_verdict_persists_verdict_row() -> None: assert kwargs["call_id"] == "c1" -def test_on_intent_verdict_queues_pending_when_decision_unset() -> None: +def test_on_intent_verdict_parks_on_owning_cycle_when_undecided() -> None: + ui = _make_ui() + cycle = _register_cycle(ui, ["c1"]) + with _patch_get_storage(MagicMock()): + ui.on_intent_verdict({"verdict_id": "v1", "call_id": "c1"}) + assert cycle.pending_verdicts == [{"verdict_id": "v1", "call_id": "c1"}] + + +def test_on_intent_verdict_without_owner_is_cache_only() -> None: + """No live cycle owns the call (pre-cycle Smart-Approvals arrival): + the verdict caches for the wait/replay but parks nowhere — the + gate's registration sweep adopts it when the cycle is created.""" ui = _make_ui() with _patch_get_storage(MagicMock()): ui.on_intent_verdict({"verdict_id": "v1", "call_id": "c1"}) - assert ui._pending_verdicts == [{"verdict_id": "v1", "call_id": "c1"}] + assert ui._llm_verdicts["c1"]["verdict_id"] == "v1" + assert ui._approval_cycles == {} def test_on_intent_verdict_stamps_immediately_when_decision_already_set() -> None: - """Late-arriving verdict (after approval resolved) gets - user_decision stamped immediately instead of queued.""" + """Late-arriving verdict (after its round resolved) gets + user_decision stamped from the per-call decision map instead of + parked — the run-to-completion daemon can deliver seconds after + the gate closed.""" storage = MagicMock() ui = _make_ui() - ui._last_verdict_decision = "approved" + ui._recent_decisions["c-late"] = ("approved", None) with _patch_get_storage(storage): ui.on_intent_verdict({"verdict_id": "v-late", "call_id": "c-late"}) - # Not queued — decision was already set. - assert ui._pending_verdicts == [] storage.update_intent_verdict.assert_called_once_with("v-late", user_decision="approved") +def test_on_intent_verdict_late_stamp_is_per_call_not_global() -> None: + """Concurrent-cycles regression: sibling B's late verdict must NOT + inherit sibling A's decision — the old single ``_last_verdict_decision`` + string stamped every late verdict with whichever round resolved last.""" + storage = MagicMock() + ui = _make_ui() + _register_cycle(ui, ["a1"]) + _register_cycle(ui, ["b1"]) + with _patch_get_storage(storage): + ui.resolve_approval(True, None, call_id="a1") # approve A only + ui.on_intent_verdict({"verdict_id": "v-b", "call_id": "b1"}) + # B's verdict is parked on B's still-open cycle, unstamped. + for call in storage.update_intent_verdict.call_args_list: + assert call.args[0] != "v-b", "sibling A's decision leaked onto B's verdict" + + def test_on_superseded_intent_verdict_persists_without_live_surfaces() -> None: """The persist-only audit hook for verdicts that landed after a newer turn replaced their judge generation: the row reaches storage with @@ -201,7 +277,6 @@ def test_on_superseded_intent_verdict_persists_without_live_surfaces() -> None: assert kwargs["user_decision"] == "superseded" assert lq.empty() # no SSE delivery assert "c-late" not in ui._llm_verdicts # no replay-cache write - assert ui._pending_verdicts == [] # no decision-stamp park assert "user_decision" not in verdict # caller's dict not mutated @@ -221,54 +296,64 @@ def test_llm_verdict_cache_evicts_oldest_at_cap() -> None: # --------------------------------------------------------------------------- -# Approval cycle reset — the bug-1 regression +# Per-round verdict purge — the bug-1 regression, scoped for concurrency # --------------------------------------------------------------------------- -def test_reset_approval_cycle_clears_decision_and_cache() -> None: +def test_purge_round_verdicts_is_scoped_to_the_entering_batch() -> None: + """Successor of the whole-cache reset: entering a gate evicts stale + verdict state for ITS call_ids only — a concurrent sibling cycle's + cached verdicts must survive (the old full clear wiped them + mid-wait and sent qualifying batches to a human).""" ui = _make_ui() - ui._last_verdict_decision = "approved" + ui._recent_decisions["c-stale"] = ("approved", None) ui._llm_verdicts["c-stale"] = {"verdict_id": "stale"} - ui._reset_approval_cycle() - assert ui._last_verdict_decision == "" - assert ui._llm_verdicts == {} + ui._verdict_origins["c-stale"] = 123 + ui._llm_verdicts["c-sibling"] = {"verdict_id": "sibling"} + ui._purge_round_verdicts({"c-stale"}) + assert "c-stale" not in ui._llm_verdicts + assert "c-stale" not in ui._verdict_origins + assert "c-stale" not in ui._recent_decisions + # The concurrent sibling's verdict is untouched. + assert ui._llm_verdicts["c-sibling"]["verdict_id"] == "sibling" def test_late_verdict_in_new_round_not_stamped_with_prior_decision() -> None: - """Regression test for the ultrareview bug-1 finding. + """Regression test for the ultrareview bug-1 finding, cycle-scoped. - Round 1: approve → _last_verdict_decision = "approved". - Round 2 begins: caller calls _reset_approval_cycle(). + Round 1 (call c1): approve → decision recorded for c1 only. + Round 2 (call c2, new cycle) begins. A verdict fires mid-round 2: must NOT inherit "approved" from - round 1. Must land in _pending_verdicts waiting for this round's - resolution. + round 1. Must park on round 2's cycle awaiting ITS resolution. """ storage = MagicMock() ui = _make_ui() - # Simulate round 1 completion. + # Round 1 completion. + _register_cycle(ui, ["c1"]) with _patch_get_storage(storage): ui.on_intent_verdict({"verdict_id": "v1", "call_id": "c1"}) ui.resolve_approval(True, None) - assert ui._last_verdict_decision == "approved" - # Round 2 begins — subclass approve_tools calls this at entry. - ui._reset_approval_cycle() + assert ui._recent_decisions.get("c1") == ("approved", None) + # Round 2 begins — approve_tools purges the entering ids and + # registers a fresh cycle. + ui._purge_round_verdicts({"c2"}) + cycle2 = _register_cycle(ui, ["c2"]) # Late judge fires during round 2 BEFORE the user decides. with _patch_get_storage(storage): ui.on_intent_verdict({"verdict_id": "v2", "call_id": "c2"}) - # The new verdict must be pending (awaiting this round's decision), + # The new verdict parks on round 2's cycle (awaiting its decision), # NOT already stamped with round 1's "approved". - assert ui._pending_verdicts == [{"verdict_id": "v2", "call_id": "c2"}] - # update_intent_verdict was only called ONCE: for v1 when round 1 - # resolved. v2 should NOT have been stamped. + assert cycle2.pending_verdicts == [{"verdict_id": "v2", "call_id": "c2"}] for call in storage.update_intent_verdict.call_args_list: assert call.args[0] != "v2", "late verdict was stamped with prior round's decision" -def test_both_subclasses_call_reset_from_approve_tools() -> None: - """Regression for bug-1: the real subclass ``approve_tools`` - methods must invoke ``_reset_approval_cycle`` at entry. Without - this, coord sessions that already resolved a prior approval stamp - the next round's late verdicts with the stale decision. +def test_both_subclasses_purge_round_state_from_approve_tools() -> None: + """Regression for bug-1, scoped: the real subclass ``approve_tools`` + bodies must purge the ENTERING batch's stale verdict state at entry + — a provider that reuses call_ids across turns must not have round + 1's cached ``approve`` pre-satisfy round 2's Smart-Approvals wait. + A concurrent sibling's cache entry survives. """ import turnstone.server from turnstone.console.coordinator_ui import ConsoleCoordinatorUI @@ -277,39 +362,47 @@ def test_both_subclasses_call_reset_from_approve_tools() -> None: for cls in (webui, ConsoleCoordinatorUI): ui = cls(ws_id="ws-x", user_id="u1") - # Stage state as if a prior approval round already finished. - ui._last_verdict_decision = "approved" - ui._llm_verdicts["stale"] = {"verdict_id": "stale"} - # Entering approve_tools for a new round — the reset must fire. - # Pass items with needs_approval=False so approve_tools returns - # without blocking on user input. + # Stage state as if a prior round already finished on the SAME + # call_id this round reuses, plus an unrelated sibling entry. + ui._recent_decisions["c-reused"] = ("approved", None) + ui._llm_verdicts["c-reused"] = {"verdict_id": "stale"} + ui._llm_verdicts["c-sibling"] = {"verdict_id": "sibling"} + # Entering approve_tools for the new round — the scoped purge + # must fire for c-reused. needs_approval=False so the gate + # returns without blocking on user input. with _patch_get_storage(MagicMock()): - ui.approve_tools([{"func_name": "ls", "needs_approval": False}]) - assert ui._last_verdict_decision == "", ( - f"{cls.__name__}.approve_tools did not call _reset_approval_cycle " - "— next round's verdicts would inherit the prior decision" + ui.approve_tools([{"call_id": "c-reused", "func_name": "ls", "needs_approval": False}]) + assert "c-reused" not in ui._llm_verdicts, ( + f"{cls.__name__}.approve_tools did not purge the entering batch's " + "stale cached verdict — round 2 could ride round 1's approve" ) - assert ui._llm_verdicts == {}, ( - f"{cls.__name__}.approve_tools did not clear the LLM verdict cache" + assert "c-reused" not in ui._recent_decisions, ( + f"{cls.__name__}.approve_tools did not purge the entering batch's " + "stale decision — this round's verdicts would inherit it" + ) + assert ui._llm_verdicts.get("c-sibling") == {"verdict_id": "sibling"}, ( + f"{cls.__name__}.approve_tools wiped a concurrent sibling's verdict" ) def test_on_intent_verdict_decision_check_and_queue_are_atomic() -> None: """Regression for the on_intent_verdict ↔ resolve_approval race. - Prior implementation acquired ``_ws_lock`` twice: once to read - ``_last_verdict_decision``, once to append to - ``_pending_verdicts``. Between those two acquisitions - ``resolve_approval`` could swap-and-clear the pending list and - set the decision — our verdict then got appended to the fresh - list and stamped with the NEXT round's decision. + The owner-check, the park, and the fallback decision-read must + happen under a SINGLE lock acquisition: ``resolve_approval`` marks + the cycle resolved and records the per-call decision atomically + under the same lock, so exactly one side wins — the verdict is + either parked pre-decision (resolve stamps it from the cycle's + ``pending_verdicts``) or stamped post-decision (from + ``_recent_decisions``). A check-then-release-then-park pattern + reopens the window where a verdict lands unparked AND unstamped — + an audit row stuck at "pending" forever. - Fix: decision check + append happen under a single lock - acquisition. This test counts lock acquisitions during one - ``on_intent_verdict`` and fails if the release-then-reacquire - pattern returns. + This test counts lock acquisitions during one ``on_intent_verdict`` + and fails if the release-then-reacquire pattern returns. """ ui = _make_ui() + _register_cycle(ui, ["c1"]) acquire_count = 0 original_lock = ui._ws_lock @@ -335,32 +428,34 @@ def test_on_intent_verdict_decision_check_and_queue_are_atomic() -> None: with _patch_get_storage(MagicMock()): ui.on_intent_verdict({"verdict_id": "v1", "call_id": "c1"}) # Two acquisitions: one for the cache write (call_id is truthy), - # one for decision-check + pending-append. Before the fix there - # were three, with a window resolve_approval could slip into. + # one for owner-check + park-or-stamp. A third acquisition means + # the release-then-reacquire window is back. assert acquire_count == 2, ( f"on_intent_verdict acquired _ws_lock {acquire_count} times; " - "decision-check + pending-append must happen under ONE acquisition " + "owner-check + park-or-stamp must happen under ONE acquisition " "to avoid a race with resolve_approval" ) def test_resolve_approval_stamps_all_pending_verdicts() -> None: - """Normal path: multiple verdicts queued during the round, all get + """Normal path: multiple verdicts parked on the round's cycle, all stamped with the user's decision on resolve.""" storage = MagicMock() ui = _make_ui() + cycle = _register_cycle(ui, ["c1", "c2"]) with _patch_get_storage(storage): ui.on_intent_verdict({"verdict_id": "v1", "call_id": "c1"}) ui.on_intent_verdict({"verdict_id": "v2", "call_id": "c2"}) - assert len(ui._pending_verdicts) == 2 + assert len(cycle.pending_verdicts) == 2 with _patch_get_storage(storage): ui.resolve_approval(False, "too risky") # Both verdicts get stamped. stamped_ids = {c.args[0] for c in storage.update_intent_verdict.call_args_list} assert stamped_ids == {"v1", "v2"} - # Pending list cleared after resolve. - assert ui._pending_verdicts == [] - assert ui._last_verdict_decision == "denied" + # Cycle's park cleared after resolve; decisions recorded per call. + assert cycle.pending_verdicts == [] + assert ui._recent_decisions.get("c1") == ("denied", None) + assert ui._recent_decisions.get("c2") == ("denied", None) # --------------------------------------------------------------------------- @@ -377,12 +472,13 @@ def test_resolve_approval_timeout_kwarg_writes_timeout_value() -> None: string used to carry this distinction but the column alone could not.""" storage = MagicMock() ui = _make_ui() + _register_cycle(ui, ["c1"]) with _patch_get_storage(storage): ui.on_intent_verdict({"verdict_id": "v1", "call_id": "c1"}) with _patch_get_storage(storage): ui.resolve_approval(False, "expired", timeout=True) storage.update_intent_verdict.assert_any_call("v1", user_decision="timeout") - assert ui._last_verdict_decision == "timeout" + assert ui._recent_decisions.get("c1") == ("timeout", None) def test_resolve_approval_timeout_with_approved_raises() -> None: @@ -442,11 +538,6 @@ def test_on_intent_verdict_consumes_auto_approve_reason() -> None: assert kwargs["user_decision"] == "auto_approve_tools" # Consumed on read so the same call_id can't double-stamp later. assert "c-x" not in ui._auto_approve_reasons - # Auto-stamped verdicts must NOT join _pending_verdicts — the - # row's final decision is already set; appending would let a - # later resolve_approval overwrite the auto-reason with the - # manual decision (real audit-trail clobber bug). - assert ui._pending_verdicts == [] def test_on_intent_verdict_auto_reason_survives_resolve_cycle() -> None: @@ -459,6 +550,7 @@ def test_on_intent_verdict_auto_reason_survives_resolve_cycle() -> None: ``"approved"``/``"denied"`` by the resolve path.""" storage = MagicMock() ui = _make_ui() + _register_cycle(ui, ["c-pending"]) ui._auto_approve_reasons["c-auto"] = ("policy", 0.0) with _patch_get_storage(storage): # LLM verdict fires for the auto-approved sibling. @@ -627,39 +719,54 @@ def test_record_output_assessment_defaults_to_heuristic_tier() -> None: # --------------------------------------------------------------------------- -# serialize_pending_approval_detail — dashboard projection +# serialize_pending_approval_details — dashboard projection (per cycle) # --------------------------------------------------------------------------- -def test_serialize_pending_approval_detail_returns_none_when_unset() -> None: +def _register_card_cycle(ui: SessionUIBase, card: dict[str, Any]) -> Any: + """Register a cycle from a raw approve_request card (shape-exact tests).""" + from turnstone.core.session_ui_base import ApprovalCycle + + cycle = ApprovalCycle(list(card.get("items") or []), card, None) + ui._register_approval_cycle(cycle) + return cycle + + +def test_serialize_pending_approval_details_empty_when_no_cycles() -> None: ui = _make_ui() - assert ui.serialize_pending_approval_detail() is None + assert ui.serialize_pending_approval_details() == [] -def test_serialize_pending_approval_detail_returns_none_when_items_empty() -> None: +def test_serialize_pending_approval_details_skips_empty_items_card() -> None: ui = _make_ui() - ui._pending_approval = {"type": "approve_request", "items": [], "judge_pending": False} - assert ui.serialize_pending_approval_detail() is None + _register_card_cycle( + ui, {"type": "approve_request", "cycle_id": "cy-0", "items": [], "judge_pending": False} + ) + assert ui.serialize_pending_approval_details() == [] -def test_serialize_pending_approval_detail_merges_judge_verdict() -> None: +def test_serialize_pending_approval_details_merges_judge_verdict() -> None: ui = _make_ui() - ui._pending_approval = { - "type": "approve_request", - "items": [ - { - "call_id": "c-1", - "header": "bash", - "preview": "$ ls", - "func_name": "bash", - "approval_label": "bash", - "needs_approval": True, - "error": None, - "verdict": {"recommendation": "review", "tier": "heuristic"}, - } - ], - "judge_pending": True, - } + _register_card_cycle( + ui, + { + "type": "approve_request", + "cycle_id": "cy-1", + "items": [ + { + "call_id": "c-1", + "header": "bash", + "preview": "$ ls", + "func_name": "bash", + "approval_label": "bash", + "needs_approval": True, + "error": None, + "heuristic_verdict": {"recommendation": "review", "tier": "heuristic"}, + } + ], + "judge_pending": True, + }, + ) ui._llm_verdicts["c-1"] = { "verdict_id": "v-1", "call_id": "c-1", @@ -667,8 +774,10 @@ def test_serialize_pending_approval_detail_merges_judge_verdict() -> None: "recommendation": "deny", "tier": "llm", } - detail = ui.serialize_pending_approval_detail() - assert detail is not None + details = ui.serialize_pending_approval_details() + assert len(details) == 1 + detail = details[0] + assert detail["cycle_id"] == "cy-1" assert detail["call_id"] == "c-1" assert detail["judge_pending"] is True assert len(detail["items"]) == 1 @@ -681,63 +790,100 @@ def test_serialize_pending_approval_detail_merges_judge_verdict() -> None: assert item["judge_verdict"]["risk_level"] == "high" -def test_serialize_pending_approval_detail_judge_verdict_none_when_missing() -> None: +def test_serialize_pending_approval_details_judge_verdict_none_when_missing() -> None: """No cached verdict for the call_id → judge_verdict is None, not absent or some sentinel.""" ui = _make_ui() - ui._pending_approval = { - "type": "approve_request", - "items": [{"call_id": "c-1", "func_name": "ls", "needs_approval": True}], - "judge_pending": True, - } - detail = ui.serialize_pending_approval_detail() - assert detail is not None - assert detail["items"][0]["judge_verdict"] is None - assert detail["items"][0]["heuristic_verdict"] is None + _register_card_cycle( + ui, + { + "type": "approve_request", + "cycle_id": "cy-1", + "items": [{"call_id": "c-1", "func_name": "ls", "needs_approval": True}], + "judge_pending": True, + }, + ) + details = ui.serialize_pending_approval_details() + assert details[0]["items"][0]["judge_verdict"] is None + assert details[0]["items"][0]["heuristic_verdict"] is None -def test_serialize_pending_approval_detail_multi_item() -> None: +def test_serialize_pending_approval_details_multi_item() -> None: ui = _make_ui() - ui._pending_approval = { - "type": "approve_request", - "items": [ - {"call_id": "c-1", "func_name": "bash", "needs_approval": True}, - {"call_id": "c-2", "func_name": "mcp__sf__query", "needs_approval": True}, - ], - "judge_pending": False, - } + _register_card_cycle( + ui, + { + "type": "approve_request", + "cycle_id": "cy-1", + "items": [ + {"call_id": "c-1", "func_name": "bash", "needs_approval": True}, + {"call_id": "c-2", "func_name": "mcp__sf__query", "needs_approval": True}, + ], + "judge_pending": False, + }, + ) ui._llm_verdicts["c-2"] = {"recommendation": "deny", "risk_level": "crit"} - detail = ui.serialize_pending_approval_detail() - assert detail is not None + details = ui.serialize_pending_approval_details() + detail = details[0] assert detail["call_id"] == "c-1" # primary = first item assert len(detail["items"]) == 2 assert detail["items"][0]["judge_verdict"] is None assert detail["items"][1]["judge_verdict"]["recommendation"] == "deny" -def test_serialize_pending_approval_detail_tool_policy_denied_passthrough() -> None: +def test_serialize_pending_approval_details_one_entry_per_live_cycle() -> None: + """Parallel task agents: every live cycle serializes, oldest first, + each addressable by its cycle_id — the single-slot serializer only + ever showed the one card the last writer left behind.""" + ui = _make_ui() + _register_card_cycle( + ui, + { + "type": "approve_request", + "cycle_id": "cy-old", + "items": [{"call_id": "a-1", "func_name": "bash", "needs_approval": True}], + "judge_pending": False, + }, + ) + _register_card_cycle( + ui, + { + "type": "approve_request", + "cycle_id": "cy-new", + "items": [{"call_id": "b-1", "func_name": "write_file", "needs_approval": True}], + "judge_pending": False, + }, + ) + details = ui.serialize_pending_approval_details() + assert [d["cycle_id"] for d in details] == ["cy-old", "cy-new"] + assert [d["call_id"] for d in details] == ["a-1", "b-1"] + + +def test_serialize_pending_approval_details_tool_policy_denied_passthrough() -> None: """A tool-policy-denied item carries error + needs_approval=False after WebUI.approve_tools mutates the items list. The serializer must round-trip both fields so the JS can detect the POLICY-BLOCKED matrix row and render the banner instead of approve/deny buttons.""" ui = _make_ui() - ui._pending_approval = { - "type": "approve_request", - "items": [ - { - "call_id": "c-1", - "func_name": "rm_rf", - "approval_label": "rm_rf", - "needs_approval": False, - "error": "Blocked by tool policy (pattern match for 'rm_rf')", - } - ], - "judge_pending": False, - } - detail = ui.serialize_pending_approval_detail() - assert detail is not None - item = detail["items"][0] + _register_card_cycle( + ui, + { + "type": "approve_request", + "cycle_id": "cy-1", + "items": [ + { + "call_id": "c-1", + "func_name": "rm_rf", + "approval_label": "rm_rf", + "needs_approval": False, + "error": "Blocked by tool policy (pattern match for 'rm_rf')", + } + ], + "judge_pending": False, + }, + ) + item = ui.serialize_pending_approval_details()[0]["items"][0] # Both fields are the JS detection keys for the POLICY-BLOCKED # branch in renderApprovalBlock — drift here silently regresses # to a buttoned approve UI on a server-blocked call. @@ -745,26 +891,29 @@ def test_serialize_pending_approval_detail_tool_policy_denied_passthrough() -> N assert item["error"] == "Blocked by tool policy (pattern match for 'rm_rf')" -def test_serialize_pending_approval_detail_judge_unavailable_path() -> None: +def test_serialize_pending_approval_details_judge_unavailable_path() -> None: """No judge_verdict + no heuristic_verdict + judge_pending=False is the (judge unavailable) matrix row — the JS detects it via !verdict && !judgePending && !policyBlocked. Verify the serialized payload preserves the absence of all three signals.""" ui = _make_ui() - ui._pending_approval = { - "type": "approve_request", - "items": [ - { - "call_id": "c-1", - "func_name": "bash", - "approval_label": "bash", - "needs_approval": True, - } - ], - "judge_pending": False, - } - detail = ui.serialize_pending_approval_detail() - assert detail is not None + _register_card_cycle( + ui, + { + "type": "approve_request", + "cycle_id": "cy-1", + "items": [ + { + "call_id": "c-1", + "func_name": "bash", + "approval_label": "bash", + "needs_approval": True, + } + ], + "judge_pending": False, + }, + ) + detail = ui.serialize_pending_approval_details()[0] assert detail["judge_pending"] is False item = detail["items"][0] assert item["judge_verdict"] is None @@ -773,18 +922,21 @@ def test_serialize_pending_approval_detail_judge_unavailable_path() -> None: assert item["error"] is None -def test_serialize_pending_approval_detail_returned_dict_is_decoupled() -> None: +def test_serialize_pending_approval_details_returned_dict_is_decoupled() -> None: """Mutating the returned dict must not corrupt the cached verdict, which other consumers may still read.""" ui = _make_ui() - ui._pending_approval = { - "type": "approve_request", - "items": [{"call_id": "c-1", "func_name": "bash", "needs_approval": True}], - "judge_pending": False, - } + _register_card_cycle( + ui, + { + "type": "approve_request", + "cycle_id": "cy-1", + "items": [{"call_id": "c-1", "func_name": "bash", "needs_approval": True}], + "judge_pending": False, + }, + ) ui._llm_verdicts["c-1"] = {"recommendation": "approve"} - detail = ui.serialize_pending_approval_detail() - assert detail is not None + detail = ui.serialize_pending_approval_details()[0] detail["items"][0]["judge_verdict"]["recommendation"] = "MUTATED" assert ui._llm_verdicts["c-1"]["recommendation"] == "approve" @@ -1535,19 +1687,27 @@ def test_concurrent_writer_and_register_with_snapshot_no_loss_no_dup() -> None: class _SeedingUI(_ConcreteUI): - """Re-delivers seeded LLM verdicts right after the approval-cycle - reset clears the cache — simulates the async judge daemon delivering - them via ``on_intent_verdict`` during the Smart Approvals wait, which - is the only point at which they can land and survive the reset.""" + """Re-delivers seeded LLM verdicts right after the per-round purge + evicts the entering batch's ids — simulates the async judge daemon + delivering them via ``on_intent_verdict`` during the Smart Approvals + wait, which is the only point at which they can land and survive + the purge. ``seed_judge_event`` optionally tags the deliveries with + a generation, for window tests that need a STALE-generation arrival + between the purge and the cycle registration.""" def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) self.seed_verdicts: list[dict[str, Any]] = [] + self.seed_judge_event: threading.Event | None = None - def _reset_approval_cycle(self) -> None: - super()._reset_approval_cycle() + def _purge_round_verdicts( + self, + call_ids: set[str], + keep_origin: threading.Event | None = None, + ) -> None: + super()._purge_round_verdicts(call_ids, keep_origin=keep_origin) for verdict in self.seed_verdicts: - self.on_intent_verdict(dict(verdict)) + self.on_intent_verdict(dict(verdict), judge_event=self.seed_judge_event) def _patch_policies(verdicts: dict[str, str]): # type: ignore[no-untyped-def] @@ -1775,19 +1935,17 @@ def test_smart_approval_skips_budget_override_pseudo_tool() -> None: def test_smart_approval_stamps_verdict_user_decision() -> None: - """The LLM verdict arrived during the wait (parked in - ``_pending_verdicts`` as pending); the smart stage pulls it out so a - sibling's resolve can't re-stamp it, and records ``smart_approval`` on - both the cached dict and the persisted row.""" + """The LLM verdict arrived during the wait (cache-only — no cycle + exists yet at that point); the smart stage stamps ``smart_approval`` + on both the cached dict and the persisted row, and the non-"pending" + stamp keeps every later cycle-registration sweep away from it.""" storage = MagicMock() ui = _smart_ui() item = _pending_item("c1") verdict = _llm_verdict("c1", recommendation="approve", confidence=0.99) ui._llm_verdicts["c1"] = verdict - ui._pending_verdicts = [verdict] # as on_intent_verdict would have parked it with _patch_get_storage(storage): ui._apply_smart_approvals([item]) - assert ui._pending_verdicts == [] assert ui._llm_verdicts["c1"]["user_decision"] == "smart_approval" storage.update_intent_verdict.assert_called_once_with("v-c1", user_decision="smart_approval") @@ -1812,7 +1970,7 @@ def test_approve_tools_smart_approves_whole_batch_without_prompt() -> None: assert item["auto_approve_reason"] == "smart_approval" assert item["needs_approval"] is False assert ui._pending_approval is None # operator was never prompted - assert ui._pending_verdicts == [] # smart verdict pulled out + stamped + assert ui._approval_cycles == {} # no cycle was ever registered # No approval prompt was fanned out to listeners. events = [] while True: @@ -2035,16 +2193,17 @@ def test_smart_approval_holds_batch_with_duplicate_call_ids() -> None: assert a.get("auto_approved") is not True -def test_on_intent_verdict_skips_append_for_already_finalized_verdict() -> None: +def test_on_intent_verdict_skips_park_for_already_finalized_verdict() -> None: """Guards the audit-corruption race: a verdict already stamped with a final user_decision (e.g. ``_finalize_smart_verdicts`` ran between this - verdict's notify and its append) is NOT re-parked in _pending_verdicts, - so a later round's resolve_approval can't overwrite its audit row.""" + verdict's notify and its park) is NOT parked on its owning cycle, + so that cycle's resolve_approval can't overwrite the audit row.""" ui = _make_ui() + cycle = _register_cycle(ui, ["c1"]) verdict = {"verdict_id": "v1", "call_id": "c1", "user_decision": "smart_approval"} with _patch_get_storage(MagicMock()): ui.on_intent_verdict(verdict) - assert ui._pending_verdicts == [] + assert cycle.pending_verdicts == [] assert ui._llm_verdicts["c1"]["user_decision"] == "smart_approval" @@ -2264,3 +2423,395 @@ class TestAgentTrajectoryStash: assert ui.get_agent_trajectory("t0") is None assert ui.get_agent_trajectory("t2") is None assert ui.get_agent_trajectory(f"t{_AGENT_TRAJECTORY_CAP + 2}") is not None + + +# --------------------------------------------------------------------------- +# Concurrent approval cycles — parallel task agents each run their own gate. +# Regression matrix for the two 1.7 release blockers: cross-approval (one +# click resolving every parked gate) and the lost-wakeup hang (a sibling's +# gate entry eating a just-fired resolution). +# --------------------------------------------------------------------------- + + +_Spawn = Callable[[dict[str, Any]], tuple[threading.Thread, dict[str, Any]]] + + +@contextlib.contextmanager +def _gate_harness(ui: SessionUIBase) -> Iterator[_Spawn]: + """ONE storage/policy patch pair + spawn + guaranteed teardown for + concurrent ``approve_tools`` gates. + + The patches are applied ONCE, on the calling thread, and cover every + spawned gate thread: ``mock.patch`` start/stop of the SAME target + from concurrent threads corrupts the patcher's restore stack — the + second stop can reinstall the first thread's mock as the "original", + leaking it into every later test in the process. + + Teardown keeps sweeping ``resolve_all_approvals`` until every gate + thread has exited — a gate that registers its cycle after a single + sweep would otherwise park for the full approval timeout and trip + the conftest thread-leak guard. + """ + threads: list[threading.Thread] = [] + + def spawn(item: dict[str, Any]) -> tuple[threading.Thread, dict[str, Any]]: + box: dict[str, Any] = {} + + def _run() -> None: + approved, feedback = ui.approve_tools([item]) + box["approved"] = approved + box["feedback"] = feedback + + t = threading.Thread(target=_run, daemon=True) + t.start() + threads.append(t) + return t, box + + with _patch_get_storage(MagicMock()), _patch_policies({}): + try: + yield spawn + finally: + stop = time.monotonic() + 5.0 + while any(t.is_alive() for t in threads) and time.monotonic() < stop: + ui.resolve_all_approvals(False, "test teardown") + time.sleep(0.01) + for t in threads: + t.join(timeout=1.0) + + +def _wait_for_cycles(ui: SessionUIBase, count: int, deadline: float = 5.0) -> None: + stop = time.monotonic() + deadline + while time.monotonic() < stop: + with ui._ws_lock: + if len(ui._approval_cycles) >= count: + return + time.sleep(0.005) + raise AssertionError(f"never saw {count} live cycles") + + +def test_concurrent_gates_resolve_independently() -> None: + """THE cross-approval regression: two parallel gates, two separate + decisions. Approving A's cycle must not wake B, and B's later + denial must reach B's thread — one click can no longer resolve + every parked batch with the same verdict.""" + ui = _make_ui() + with _gate_harness(ui) as spawn: + ta, box_a = spawn(_pending_item("a-1")) + tb, box_b = spawn(_pending_item("b-1")) + _wait_for_cycles(ui, 2) + assert ui.resolve_approval(True, "run it", call_id="a-1") is not None + ta.join(timeout=5.0) + assert not ta.is_alive(), "A's gate did not wake on its own resolution" + # B is still parked — A's approval must NOT have leaked to it. + assert tb.is_alive(), "resolving A also unblocked B (cross-approval)" + assert box_a == {"approved": True, "feedback": "run it"} + assert ui.resolve_approval(False, "not this one", call_id="b-1") is not None + tb.join(timeout=5.0) + assert not tb.is_alive() + assert box_b["approved"] is False + assert box_b["feedback"] == "not this one" + + +def test_sibling_gate_entry_cannot_eat_a_resolution() -> None: + """THE lost-wakeup regression: under the singleton event, sibling B + entering the gate ran ``event.clear()`` and could erase A's + just-fired resolution — A then parked for the full 3600s timeout + ("approval dialog stuck"). Per-cycle events make the interleaving + structurally impossible: A's resolution lands on A's OWN event, so + B's registration can't touch it.""" + ui = _make_ui() + with _gate_harness(ui) as spawn: + ta, box_a = spawn(_pending_item("a-1")) + _wait_for_cycles(ui, 1) + # Resolve A and IMMEDIATELY register sibling B — the old code's + # clear() window. A must still return promptly. + ui.resolve_approval(True, None, call_id="a-1") + spawn(_pending_item("b-1")) + ta.join(timeout=5.0) + assert not ta.is_alive(), ( + "A's gate lost its wakeup when sibling B entered — the singleton-event race is back" + ) + assert box_a["approved"] is True + + +def test_selectorless_resolve_hits_oldest_cycle() -> None: + """Legacy clients (CLI wrappers, channel adapters, old tabs) send no + selector — the decision lands on the OLDEST live cycle, matching + the order the prompts were issued.""" + ui = _make_ui() + with _gate_harness(ui) as spawn: + ta, box_a = spawn(_pending_item("a-1")) + _wait_for_cycles(ui, 1) + tb, _box_b = spawn(_pending_item("b-1")) + _wait_for_cycles(ui, 2) + ui.resolve_approval(True, "first in, first out") + ta.join(timeout=5.0) + assert not ta.is_alive(), "selector-less resolve missed the oldest cycle" + assert box_a["approved"] is True + assert tb.is_alive(), "selector-less resolve hit more than one cycle" + + +def test_resolve_all_approvals_wakes_every_gate() -> None: + """The cancel/close sweep: every parked gate wakes with its own + denied result.""" + ui = _make_ui() + with _gate_harness(ui) as spawn: + ta, box_a = spawn(_pending_item("a-1")) + tb, box_b = spawn(_pending_item("b-1")) + _wait_for_cycles(ui, 2) + assert ui.resolve_all_approvals(False, "Cancelled by user") == 2 + ta.join(timeout=5.0) + tb.join(timeout=5.0) + assert box_a["approved"] is False + assert box_b["approved"] is False + assert "Cancelled by user" in (box_a["feedback"] or "") + + +def test_resolve_all_approvals_noop_when_idle() -> None: + """Idle cancels stay silent — no stale approval_resolved broadcast.""" + ui = _make_ui() + lq = ui._register_listener() + assert ui.resolve_all_approvals(False, "Cancelled by user") == 0 + assert lq.empty() + + +def test_double_resolution_is_a_guarded_noop() -> None: + """A second decision racing the first (two tabs, or timeout racing a + click) must not re-resolve, re-broadcast, or clobber the recorded + result.""" + ui = _make_ui() + with _gate_harness(ui) as spawn: + ta, box_a = spawn(_pending_item("a-1")) + _wait_for_cycles(ui, 1) + first = ui.resolve_approval(True, "yes", call_id="a-1") + second = ui.resolve_approval(False, "no", call_id="a-1") + assert first is not None + assert second is None + ta.join(timeout=5.0) + assert box_a == {"approved": True, "feedback": "yes"} + + +def test_pending_cards_and_legacy_view_track_cycles() -> None: + """``pending_approval_cards`` lists every live cycle's card (SSE + replay repaints them all); the legacy ``_pending_approval`` view + tracks the OLDEST for boolean-ish consumers and rolls forward as + cycles resolve.""" + ui = _make_ui() + with _gate_harness(ui) as spawn: + ta, _box_a = spawn(_pending_item("a-1")) + _wait_for_cycles(ui, 1) + spawn(_pending_item("b-1")) + _wait_for_cycles(ui, 2) + cards = ui.pending_approval_cards() + assert [c["items"][0]["call_id"] for c in cards] == ["a-1", "b-1"] + assert ui._pending_approval is not None + assert ui._pending_approval["items"][0]["call_id"] == "a-1" + ui.resolve_approval(True, None, call_id="a-1") + ta.join(timeout=5.0) + # View rolls forward to the surviving cycle. + assert ui._pending_approval is not None + assert ui._pending_approval["items"][0]["call_id"] == "b-1" + + +def test_stale_generation_verdict_cannot_touch_live_cycle() -> None: + """A prior turn's run-to-completion daemon delivering a reused + call_id must not satisfy the NEW cycle's wait: the delivery's + generation (its cancel event) is identity-checked against the + owning cycle's — mismatch persists for audit only, with no cache + write, no SSE, no park.""" + storage = MagicMock() + ui = _make_ui() + fresh_gen = threading.Event() + stale_gen = threading.Event() + cycle = _register_cycle(ui, ["c-reused"], judge_event=fresh_gen) + lq = ui._register_listener() + with _patch_get_storage(storage): + ui.on_intent_verdict( + {"verdict_id": "v-stale", "call_id": "c-reused", "tier": "llm"}, + judge_event=stale_gen, + ) + assert "c-reused" not in ui._llm_verdicts + assert cycle.pending_verdicts == [] + assert lq.empty() + kwargs = storage.upsert_intent_verdict.call_args.kwargs + assert kwargs["user_decision"] == "superseded" + # The cycle's OWN generation delivers normally. + with _patch_get_storage(storage): + ui.on_intent_verdict( + {"verdict_id": "v-fresh", "call_id": "c-reused", "tier": "llm"}, + judge_event=fresh_gen, + ) + assert ui._llm_verdicts["c-reused"]["verdict_id"] == "v-fresh" + assert cycle.pending_verdicts and cycle.pending_verdicts[0]["verdict_id"] == "v-fresh" + + +def test_smart_approval_rejects_stale_origin_verdict() -> None: + """Smart-Approvals qualification requires the cached verdict to have + been delivered by THIS batch's judge generation — a cached approve + of unknown/stale origin sends the batch to a human.""" + ui = _smart_ui() + ui.smart_approval_wait_seconds = 0.05 + fresh_gen = threading.Event() + item = _pending_item("c1") + item["_judge_event"] = fresh_gen + ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=0.99) + ui._verdict_origins["c1"] = id(object()) # a different generation delivered it + with _patch_get_storage(MagicMock()): + remaining = ui._apply_smart_approvals([item]) + assert remaining == [item] + assert item.get("auto_approved") is not True + # Same verdict with the RIGHT origin qualifies. + ui._verdict_origins["c1"] = id(fresh_gen) + with _patch_get_storage(MagicMock()): + remaining = ui._apply_smart_approvals([item]) + assert remaining == [] + assert item["auto_approved"] is True + + +def test_concurrent_smart_gate_and_human_gate() -> None: + """A smart-qualifying batch auto-approves while a sibling batch is + parked on a human — the sibling's cycle survives untouched (the old + whole-cache reset at gate entry wiped its verdicts mid-wait).""" + ui = _make_ui() + ui.smart_approvals_enabled = True + ui.smart_approval_threshold = 0.95 + ui.smart_approval_wait_seconds = 1.0 + # Regression guard: if the smart batch ever falls through to the + # human gate (it runs on THIS thread), fail in seconds instead of + # hanging the suite for the full approval timeout. + ui._APPROVAL_WAIT_TIMEOUT = 10.0 + with _gate_harness(ui) as spawn: + # Human-gated sibling parks first. + ta, box_a = spawn(_pending_item("a-1")) + _wait_for_cycles(ui, 1) + # Smart-qualifying batch flows straight through on this thread — + # its verdict was delivered by its OWN generation before the + # gate was entered, so the entry purge must spare it. + gen = threading.Event() + item = _pending_item("s-1") + item["_judge_event"] = gen + ui._llm_verdicts["s-1"] = _llm_verdict("s-1", recommendation="approve", confidence=0.99) + ui._verdict_origins["s-1"] = id(gen) + approved, _ = ui.approve_tools([item]) + assert approved is True + assert item["auto_approved"] is True + # Sibling still parked, its cycle + verdict path intact. + assert ta.is_alive() + ui.resolve_approval(True, "ok", call_id="a-1") + ta.join(timeout=5.0) + assert box_a["approved"] is True + + +def test_purge_round_verdicts_keeps_entry_from_the_entering_generation() -> None: + """``keep_origin``: a verdict the entering batch's OWN judge spawn + already delivered survives the entry purge. The judge daemon is + spawned before the gate is entered, so a fast judge can beat the + gate to the cache — evicting its verdict as if it were a prior + round's leftover stalled the Smart-Approvals wait to its full + budget and sent an already-cleared batch to a human. Foreign + generations and prior-round decisions still purge.""" + ui = _make_ui() + gen = threading.Event() + other_gen = threading.Event() + ui._llm_verdicts["c-own"] = {"verdict_id": "own"} + ui._verdict_origins["c-own"] = id(gen) + ui._llm_verdicts["c-foreign"] = {"verdict_id": "foreign"} + ui._verdict_origins["c-foreign"] = id(other_gen) + ui._recent_decisions["c-own"] = ("approved", None) + ui._purge_round_verdicts({"c-own", "c-foreign"}, keep_origin=gen) + assert ui._llm_verdicts.get("c-own") == {"verdict_id": "own"} + assert ui._verdict_origins.get("c-own") == id(gen) + assert "c-foreign" not in ui._llm_verdicts + assert "c-foreign" not in ui._verdict_origins + # Decisions never survive: this round has not been decided yet. + assert "c-own" not in ui._recent_decisions + + +def test_smart_gate_uses_verdict_delivered_before_gate_entry() -> None: + """Production shape of the generation-aware purge: judge spawned + before the gate, verdict delivered before ``approve_tools`` runs. + The entry purge spares the same-generation verdict, so the smart + wait sees it immediately and the batch auto-approves without a + human prompt or a full-budget stall.""" + ui = _smart_ui() + ui.smart_approval_wait_seconds = 3.0 + # Regression guard: a purged verdict sends this batch to the human + # gate on THIS thread — bound the park so the test fails instead of + # hanging the suite. + ui._APPROVAL_WAIT_TIMEOUT = 1.0 + gen = threading.Event() + item = _pending_item("s-1") + item["_judge_event"] = gen + with _patch_get_storage(MagicMock()), _patch_policies({}): + # The "fast judge": delivery lands before the gate is entered. + ui.on_intent_verdict(_llm_verdict("s-1"), judge_event=gen) + approved, _feedback = ui.approve_tools([item]) + assert approved is True, "entry purge evicted this batch's own pre-delivered verdict" + assert item["auto_approved"] is True + + +def test_registration_evicts_stale_generation_window_arrival() -> None: + """A STALE generation delivering into the purge→register window + (the entry purge can't see arrivals that land during the policy + round-trip or the smart wait) must not blank the card's "judge + analysing" cue, be adopted into ``pending_verdicts`` for + decision-stamping, or linger in the replay cache once the cycle + registers.""" + ui = _SeedingUI(ws_id="ws-1", user_id="u1") + stale_gen = threading.Event() + fresh_gen = threading.Event() + # _SeedingUI re-delivers right after the entry purge — inside the + # purge→register window — tagged with the STALE generation. + ui.seed_verdicts = [_llm_verdict("w-1")] + ui.seed_judge_event = stale_gen + item = _pending_item("w-1") + item["_judge_event"] = fresh_gen + with _gate_harness(ui) as spawn: + _t, box = spawn(item) + _wait_for_cycles(ui, 1) + with ui._ws_lock: + cycle = next(iter(ui._approval_cycles.values())) + assert "w-1" not in ui._llm_verdicts, "stale window arrival survived registration" + assert "w-1" not in ui._verdict_origins + assert cycle.card["judge_pending"] is True, "stale verdict blanked the judge cue" + assert [v["verdict_id"] for v in cycle.pending_verdicts] == ["h-w-1"], ( + "stale window arrival was adopted for decision-stamping" + ) + ui.resolve_approval(True, None, call_id="w-1") + assert box["approved"] is True + + +def test_late_stale_generation_verdict_stamps_superseded() -> None: + """A late verdict from generation A delivering AFTER its round + resolved — and after a reused call_id's round from generation B + also resolved — must not steal B's recorded decision. Recorded + decisions are generation-tagged: a mismatched late delivery stamps + ``superseded`` (same vocabulary as the superseded persist path); a + same-generation late delivery still stamps the real decision.""" + storage = MagicMock() + ui = _make_ui() + gen_a = threading.Event() + gen_b = threading.Event() + # Round B (reusing the call_id generation A once judged) resolves + # and its gate unregisters the cycle — the decision survives only + # in ``_recent_decisions``, tagged with B's generation. + cycle_b = _register_cycle(ui, ["c-reuse"], judge_event=gen_b) + with _patch_get_storage(storage): + ui.resolve_approval(True, None, call_id="c-reuse") + ui._unregister_approval_cycle(cycle_b) + assert ui._recent_decisions["c-reuse"] == ("approved", gen_b) + # Generation A's run-to-completion daemon delivers late — no live + # owner, and the decision on file belongs to B. + with _patch_get_storage(storage): + ui.on_intent_verdict( + {"verdict_id": "v-stale-late", "call_id": "c-reuse", "tier": "llm"}, + judge_event=gen_a, + ) + storage.update_intent_verdict.assert_any_call("v-stale-late", user_decision="superseded") + # B's own late delivery still stamps B's real decision. + with _patch_get_storage(storage): + ui.on_intent_verdict( + {"verdict_id": "v-b-late", "call_id": "c-reuse", "tier": "llm"}, + judge_event=gen_b, + ) + storage.update_intent_verdict.assert_any_call("v-b-late", user_decision="approved") diff --git a/tests/test_workstream_endpoints.py b/tests/test_workstream_endpoints.py index c841606c..1b795ee4 100644 --- a/tests/test_workstream_endpoints.py +++ b/tests/test_workstream_endpoints.py @@ -1598,7 +1598,7 @@ class TestDetailInteractive: "user_id": "test-user", "kind": "interactive", "pending_approval": False, - "pending_approval_detail": None, + "pending_approval_details": [], } def test_pending_approval_fields_propagate_from_ui(self): @@ -1631,23 +1631,26 @@ class TestDetailInteractive: ], "judge_pending": True, } - loaded_ws.ui.serialize_pending_approval_detail = MagicMock( - return_value={ - "call_id": "c-1", - "judge_pending": True, - "items": [ - { - "call_id": "c-1", - "func_name": "spawn_workstream", - "needs_approval": True, - "heuristic_verdict": { - "recommendation": "approve", - "risk_level": "low", - "confidence": 0.9, - }, - } - ], - } + loaded_ws.ui.serialize_pending_approval_details = MagicMock( + return_value=[ + { + "cycle_id": "cyc-1", + "call_id": "c-1", + "judge_pending": True, + "items": [ + { + "call_id": "c-1", + "func_name": "spawn_workstream", + "needs_approval": True, + "heuristic_verdict": { + "recommendation": "approve", + "risk_level": "low", + "confidence": 0.9, + }, + } + ], + } + ] ) mock_mgr = MagicMock() mock_mgr.get.return_value = loaded_ws @@ -1657,9 +1660,12 @@ class TestDetailInteractive: assert r.status_code == 200 body = r.json() assert body["pending_approval"] is True - assert body["pending_approval_detail"]["call_id"] == "c-1" - assert body["pending_approval_detail"]["judge_pending"] is True - items = body["pending_approval_detail"]["items"] + details = body["pending_approval_details"] + assert len(details) == 1 + assert details[0]["cycle_id"] == "cyc-1" + assert details[0]["call_id"] == "c-1" + assert details[0]["judge_pending"] is True + items = details[0]["items"] assert len(items) == 1 assert items[0]["func_name"] == "spawn_workstream" assert items[0]["needs_approval"] is True @@ -1680,7 +1686,7 @@ class TestDetailInteractive: loaded_ws.user_id = "test-user" loaded_ws.kind = "coordinator" loaded_ws.ui._pending_approval = {"items": []} - loaded_ws.ui.serialize_pending_approval_detail = MagicMock( + loaded_ws.ui.serialize_pending_approval_details = MagicMock( side_effect=RuntimeError("verdict object is malformed"), ) mock_mgr = MagicMock() @@ -1691,7 +1697,7 @@ class TestDetailInteractive: assert r.status_code == 200 body = r.json() assert body["pending_approval"] is True - assert body["pending_approval_detail"] is None + assert body["pending_approval_details"] == [] def test_lazy_rehydrates_on_miss(self): """``mgr.get`` miss → ``mgr.open`` rehydrate. Same flow as coord; @@ -1801,7 +1807,7 @@ class TestTenantCheckOnReadEndpoints: # Sensitive fields the PR added must not surface for a # non-owning caller. assert "name" not in body - assert "pending_approval_detail" not in body + assert "pending_approval_details" not in body assert "user_id" not in body # And mgr.get was NEVER consulted — the gate fires first. mock_mgr.get.assert_not_called() @@ -1832,7 +1838,7 @@ class TestTenantCheckOnReadEndpoints: body = r.json() assert body["ws_id"] == ws_id assert body["pending_approval"] is False - assert body["pending_approval_detail"] is None + assert body["pending_approval_details"] == [] def test_history_404s_when_tenant_check_rejects(self, _inject_storage): """A non-owning interactive caller reading another user's ws_id