ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-12 23:12:23 -06:00
Code Issues Packages Projects Releases Wiki Activity

test(approvals): concurrency matrix + suite migration to the cycle model

Browse Source 
New regression matrix for the release blockers: cross-approval
independence, lost-wakeup at gate entry, FIFO selector-less
resolution, resolve-all sweep, double-resolution no-op, cards/legacy
view tracking, and the generation-exactness set — stale delivery
rejection, Smart-Approvals origin check, purge keep_origin, the
purge-to-register window eviction, late cross-generation "superseded"
stamping, concurrent smart+human gates, and the pre-delivered-verdict
fast path. Plus sub-agent judge wiring (agent_gate off the main
slot, close() firing all generations) and endpoint tests for cycle
pinning and the Approve+Always race guard.

Gate threads run under one shared mock-patch harness — mock.patch
start/stop of the same target from concurrent threads corrupts the
patcher's restore stack — with a sweep-until-dead teardown so the
conftest leak guard can't trip. Existing suites migrate off the
singleton fields to cycle assertions and the
pending_approval_details wire shape.
This commit is contained in:
Patrick Buckley
2026-07-05 01:16:37 -07:00
parent 6c94514106
commit 7f0e0406b3
13 changed files with 1110 additions and 329 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/_replay_helpers.py
+6
View File
@@ -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()
tests/test_channel_discord.py
+25 -11
View File
@@ -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()
tests/test_channel_routing.py
+5 -3
View File
@@ -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:
tests/test_channel_slack.py
+24 -9
View File
@@ -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:
tests/test_coord_ui_approve_tools.py
+7 -1
View File
@@ -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
tests/test_coordinator_adapter.py
+15
View File
@@ -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()
tests/test_coordinator_endpoints.py
+116 -56
View File
@@ -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):
tests/test_coordinator_page.py
+3 -3
View File
@@ -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."
)
tests/test_phase6_endpoints.py
+33 -23
View File
@@ -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"
tests/test_server_authz.py
+34 -21
View File
@@ -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]
tests/test_session.py
+83
View File
@@ -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
tests/test_session_ui_base.py
+728 -177
View File
File diff suppressed because it is too large Load Diff
tests/test_workstream_endpoints.py
+31 -25
View File
@@ -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