From 18c330142842c1a2512649a72c3a0e09c917ab59 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Sun, 5 Jul 2026 01:16:22 -0700 Subject: [PATCH] feat(api): cycle-routed approval resolution across server, console, schemas, SDKs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /approve accepts cycle_id / call_id selectors and 409s on stale selectors with the current cycle's ids so clients re-render instead of silently resolving an unrelated batch. Selector-less bodies pin the resolve to the cycle the lookup returned (not "whichever is oldest by the time the resolve runs"), and Approve+Always names apply only after the pinned cycle actually resolved — the auto-approve whitelist can no longer describe a different batch than the one that resolved. approve_request carries cycle_id; approval_resolved carries cycle_id + call_ids; SSE reconnect replays every live cycle's card. The console collector, coordinator UI fan-out, and both SDKs (Python + TS) thread the cycle correlation through. BREAKING (1.7): the singular pending_approval_detail field is removed from dashboard rows, workstream detail, and node snapshots — replaced by the pending_approval_details list (one entry per live cycle, each carrying its cycle_id). stable/1.6 keeps the old shape. --- sdk/typescript/src/events.ts | 10 + sdk/typescript/src/server.ts | 9 + turnstone/api/server_schemas.py | 62 ++++-- turnstone/console/collector.py | 9 + turnstone/console/coordinator_adapter.py | 5 + turnstone/console/coordinator_ui.py | 16 +- turnstone/console/server.py | 56 ++--- turnstone/core/session_routes.py | 266 +++++++++++++++-------- turnstone/sdk/console.py | 22 +- turnstone/sdk/events.py | 19 ++ turnstone/sdk/server.py | 25 ++- turnstone/server.py | 40 +++- 12 files changed, 388 insertions(+), 151 deletions(-) diff --git a/sdk/typescript/src/events.ts b/sdk/typescript/src/events.ts index dbb65b89..7de34ae5 100644 --- a/sdk/typescript/src/events.ts +++ b/sdk/typescript/src/events.ts @@ -75,15 +75,25 @@ export interface ToolInfoEvent { items: Array>; } +/** One approval CYCLE awaiting the operator. Several can be outstanding + * at once (parallel task agents each gate their own tool calls) — key + * prompt UI by `cycle_id` and echo it back on the approve POST. */ export interface ApproveRequestEvent { type: "approve_request"; + cycle_id: string; items: Array>; + judge_pending?: boolean; } +/** A specific approval cycle resolved; `cycle_id`/`call_ids` identify + * which prompt to dismiss. */ export interface ApprovalResolvedEvent { type: "approval_resolved"; approved: boolean; feedback: string; + always?: boolean; + cycle_id: string; + call_ids: string[]; } export interface ToolResultEvent { diff --git a/sdk/typescript/src/server.ts b/sdk/typescript/src/server.ts index fd49e39c..8d96240c 100644 --- a/sdk/typescript/src/server.ts +++ b/sdk/typescript/src/server.ts @@ -166,6 +166,13 @@ export class TurnstoneServer extends BaseClient { approved?: boolean; feedback?: string | null; always?: boolean; + /** Resolve exactly this approval cycle (from ApproveRequestEvent.cycle_id). + * Omitting it resolves the OLDEST live cycle — ambiguous when parallel + * task agents have several prompts outstanding, so pass it whenever the + * triggering event is known. */ + cycleId?: string; + /** Alternative selector: any call_id inside the target cycle. */ + callId?: string; }): Promise { return this.request( "POST", @@ -175,6 +182,8 @@ export class TurnstoneServer extends BaseClient { approved: opts.approved ?? true, feedback: opts.feedback, always: opts.always, + cycle_id: opts.cycleId, + call_id: opts.callId, }, }, ); diff --git a/turnstone/api/server_schemas.py b/turnstone/api/server_schemas.py index f1ec2847..f9053698 100644 --- a/turnstone/api/server_schemas.py +++ b/turnstone/api/server_schemas.py @@ -339,13 +339,24 @@ class RecentAutoApproval(BaseModel): class PendingApprovalDetail(BaseModel): """Inline approval payload merged into ``DashboardWorkstream``. - Set when a workstream's ``approve_tools`` is parked on - ``_approval_event``; ``None`` (omitted) otherwise. Cross-tenant + One entry per live approval CYCLE — a gate thread parked in + ``approve_tools`` awaiting the operator. Parallel task agents run + concurrent gates, so a workstream can have several of these at + once (``pending_approval_details``, oldest first). Cross-tenant exposure here follows the same trusted-team posture as ``activity`` / ``tokens`` — see ``server.py``'s ``dashboard`` handler comment. """ + cycle_id: str = Field( + default="", + description=( + "Identity of this approval cycle. Echo it back on " + "``POST /v1/api/workstreams/{ws_id}/approve`` to resolve " + "exactly this round — required for correctness when " + "several cycles are live (parallel task agents)." + ), + ) call_id: str = Field( default="", description=( @@ -388,16 +399,21 @@ class DashboardWorkstream(BaseModel): parent_ws_id: str | None = None user_id: str = "" project_id: str | None = None - pending_approval_detail: PendingApprovalDetail | None = Field( - default=None, + pending_approval_details: list[PendingApprovalDetail] = Field( + default_factory=list, description=( "Inline approval payload for the coordinator children-tree " - "UI. Carries the merged ``_pending_approval`` items list + " - "per-call_id LLM verdict cache so a coord can render " - "approve/deny buttons + judge pill without a separate " - "per-child round-trip. ``None`` when no approval is pending. " - "Also surfaced (verbatim) on ``GET /v1/api/cluster/ws/live`` " - "via the ``_CLUSTER_WS_LIVE_KEYS`` projection." + "UI: EVERY live approval cycle, oldest first — parallel " + "task agents gate concurrently, so a workstream can hold " + "several prompts at once. Each entry carries the cycle's " + "items + per-call_id LLM verdict cache so a coord can " + "render approve/deny buttons + judge pill without a " + "separate per-child round-trip; resolve each with its " + "``cycle_id``. Empty when no approval is pending. Also " + "surfaced (verbatim) on ``GET /v1/api/cluster/ws/live`` " + "via the ``_CLUSTER_WS_LIVE_KEYS`` projection. Replaces " + "1.6's ``pending_approval_detail`` single-object field " + "(breaking, 1.7)." ), ) recent_auto_approvals: list[RecentAutoApproval] = Field( @@ -482,21 +498,25 @@ class WorkstreamDetailResponse(BaseModel): pending_approval: bool = Field( default=False, description=( - "True when the workstream is parked on ``_approval_event`` " - "awaiting an operator approve/deny. Mirrors the same field " - "on ``DashboardWorkstream`` / cluster live projections so a " - "freshly-loaded chat tab can render the inline approval gate " - "from the detail snapshot before SSE replay arrives." + "True when at least one approval cycle is live (a gate " + "thread parked awaiting an operator approve/deny). Mirrors " + "the same field on ``DashboardWorkstream`` / cluster live " + "projections so a freshly-loaded chat tab can render the " + "inline approval gate from the detail snapshot before SSE " + "replay arrives." ), ) - pending_approval_detail: PendingApprovalDetail | None = Field( - default=None, + pending_approval_details: list[PendingApprovalDetail] = Field( + default_factory=list, description=( - "Inline approval payload — same shape as ``DashboardWorkstream" - ".pending_approval_detail``. ``None`` when no approval is " - "pending. Lets a reload paint the action row + judge " + "Inline approval payloads, one per live cycle, oldest " + "first — same shape as ``DashboardWorkstream" + ".pending_approval_details``. Empty when no approval is " + "pending. Lets a reload paint every action row + judge " "verdicts immediately instead of relying on the SSE " - "approve_request replay timing window." + "approve_request replay timing window. Replaces 1.6's " + "``pending_approval_detail`` single-object field " + "(breaking, 1.7)." ), ) diff --git a/turnstone/console/collector.py b/turnstone/console/collector.py index 4cf903c5..7a3330b2 100644 --- a/turnstone/console/collector.py +++ b/turnstone/console/collector.py @@ -733,6 +733,9 @@ class ClusterCollector: # coordinator's tree UI can clear the pending-approval # pill the moment the user decides, rather than waiting # for the subsequent state-change piggyback. + # ``cycle_id`` / ``call_ids`` name WHICH cycle resolved — + # a child running parallel task agents can have several + # approval blocks live at once. ws_id = data.get("ws_id", "") if ws_id: pending_events.append( @@ -743,6 +746,8 @@ class ClusterCollector: "approved": bool(data.get("approved", False)), "feedback": data.get("feedback", "") or "", "always": bool(data.get("always", False)), + "cycle_id": data.get("cycle_id", "") or "", + "call_ids": data.get("call_ids") or [], } ) @@ -1391,6 +1396,8 @@ class ClusterCollector: approved: bool, feedback: str = "", always: bool = False, + cycle_id: str = "", + call_ids: list[str] | None = None, ) -> None: """Fan an ``approval_resolved`` decision for a console-pseudo-node ws. @@ -1407,6 +1414,8 @@ class ClusterCollector: "approved": approved, "feedback": feedback, "always": always, + "cycle_id": cycle_id, + "call_ids": call_ids or [], } ) diff --git a/turnstone/console/coordinator_adapter.py b/turnstone/console/coordinator_adapter.py index b900dd15..620c1d46 100644 --- a/turnstone/console/coordinator_adapter.py +++ b/turnstone/console/coordinator_adapter.py @@ -670,6 +670,11 @@ class CoordinatorAdapter: "approved": bool(event.get("approved", False)), "feedback": event.get("feedback", "") or "", "always": bool(event.get("always", False)), + # Which cycle resolved — the tree row can hold several + # approval blocks when the child runs parallel task + # agents; empty (legacy node) clears them all. + "cycle_id": event.get("cycle_id", "") or "", + "call_ids": event.get("call_ids") or [], } else: # approve_request # Push path for the initial approval items — diff --git a/turnstone/console/coordinator_ui.py b/turnstone/console/coordinator_ui.py index a8442830..5d013434 100644 --- a/turnstone/console/coordinator_ui.py +++ b/turnstone/console/coordinator_ui.py @@ -5,8 +5,8 @@ Mirrors ``turnstone.server.WebUI`` but scoped to the console's needs: - Per-session SSE listener fan-out (inherited from :class:`SessionUIBase` — same ``threading.Lock`` + queue list pattern WebUI uses). -- ``threading.Event`` + ``_approval_result`` for blocking the worker - thread until a console endpoint delivers the decision (inherited). +- Per-cycle ``ApprovalCycle`` registry for blocking each gate thread + until a console endpoint delivers its decision (inherited). - Per-ws metric tracking + turn-content accumulator + activity bookkeeping (inherited from :class:`SessionUIBase` post the rich ``ws_state`` payload lift). Coord populates the same @@ -196,6 +196,8 @@ class ConsoleCoordinatorUI(SessionUIBase): feedback: str | None = None, *, always: bool = False, + cycle_id: str = "", + call_ids: tuple[str, ...] = (), ) -> None: """Fan an ``approval_resolved`` decision to the cluster collector. @@ -213,6 +215,8 @@ class ConsoleCoordinatorUI(SessionUIBase): approved=approved, feedback=feedback or "", always=always, + cycle_id=cycle_id, + call_ids=list(call_ids), ) except Exception: log.debug( @@ -317,8 +321,12 @@ class ConsoleCoordinatorUI(SessionUIBase): return fire_judge_verdict_metric(cm, verdict, "heuristic") - def on_intent_verdict(self, verdict: dict[str, Any]) -> None: - super().on_intent_verdict(verdict) + def on_intent_verdict( + self, + verdict: dict[str, Any], + judge_event: object | None = None, + ) -> None: + super().on_intent_verdict(verdict, judge_event) cm = ConsoleCoordinatorUI._console_metrics if cm is None: return diff --git a/turnstone/console/server.py b/turnstone/console/server.py index f065b081..0a201395 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -1047,12 +1047,14 @@ _CLUSTER_WS_LIVE_KEYS = ( "model_alias", "title", "name", - # Carries the inline approve/deny payload (items + judge_verdict) - # so coord live-bulk callers can render row-level UI without a - # per-child round-trip. ``None`` when no approval is pending. - # Cross-tenant exposure follows the trusted-team posture documented - # on ``SessionUIBase.serialize_pending_approval_detail``. - "pending_approval_detail", + # Carries the inline approve/deny payloads (one per live cycle, + # items + judge_verdict each) so coord live-bulk callers can render + # row-level UI without a per-child round-trip. ``[]`` when no + # approval is pending; several entries when parallel task agents + # gate concurrently. Cross-tenant exposure follows the trusted-team + # posture documented on + # ``SessionUIBase.serialize_pending_approval_details``. + "pending_approval_details", # Ring buffer of the child's recent auto-approves (last 10) for # the coord-tree's "auto-approved by skill X" pill. Without this # the operator has no surface to see WHICH tool calls bypassed @@ -1170,16 +1172,16 @@ def _coordinator_live_snapshot(ws: Any) -> dict[str, Any]: val = getattr(obj, name, "") if obj else "" return val if isinstance(val, str) else "" - # Coord rows synthesize the same ``pending_approval_detail`` shape + # Coord rows synthesize the same ``pending_approval_details`` shape # the node-side dashboard produces — single source of truth via - # ``SessionUIBase.serialize_pending_approval_detail``. The console + # ``SessionUIBase.serialize_pending_approval_details``. The console # coord LLM judge isn't wired today (``coordinator_ui.py:138`` # hardcodes ``judge_pending=False``), so ``judge_verdict`` will # always be ``None`` for these rows; the coord-self stretch in # the plan covers that follow-up. ``ui`` may be ``None`` in # transient states (newly-created ws before activation); every # active coord UI is a ``SessionUIBase`` and supports the method. - pending_approval_detail = ui.serialize_pending_approval_detail() if ui is not None else None + pending_approval_details = ui.serialize_pending_approval_details() if ui is not None else [] recent_auto_approvals = ui.serialize_recent_auto_approvals() if ui is not None else [] return { @@ -1194,7 +1196,7 @@ def _coordinator_live_snapshot(ws: Any) -> dict[str, Any]: "title": "", "name": getattr(ws, "name", "") or "", "pending_approval": pending_approval, - "pending_approval_detail": pending_approval_detail, + "pending_approval_details": pending_approval_details, "recent_auto_approvals": recent_auto_approvals, } @@ -1286,15 +1288,15 @@ async def _fetch_live_block( # ``activity_state="approval"`` is set inside approve_tools # AFTER the state transition fires, so a bulk fetch that # races with that window can see state=attention and - # activity_state="" simultaneously. A non-null - # ``pending_approval_detail`` is also a definitive signal - # (the serializer only emits non-None when ``_pending_approval`` - # is set on the UI). Any of the three flips this true; the - # frontend reducer mirrors the same disjunction. + # activity_state="" simultaneously. A non-empty + # ``pending_approval_details`` is also a definitive signal + # (the serializer emits entries only for live cycles). Any + # of the three flips this true; the frontend reducer + # mirrors the same disjunction. live["pending_approval"] = ( live.get("activity_state") == "approval" or entry.get("state") == "attention" - or live.get("pending_approval_detail") is not None + or bool(live.get("pending_approval_details")) ) return live return None @@ -3561,16 +3563,18 @@ def _coord_events_replay( """ yield from session_replay_preamble(ws.session, ui) - pending_approval = getattr(ui, "_pending_approval", None) - if pending_approval is not None: - yield pending_approval - # Cached LLM verdicts that fired since the approval prompt - # — without this replay, a reconnecting / refreshing tab - # sees the approve_request prompt but no judge chip, and - # since intent_verdict only fires once per call_id (no - # push to a late subscriber), the chip would never appear - # until the operator re-invokes the action. Mirrors the - # interactive path at ``turnstone/server.py:875-878``. + # EVERY live approval cycle replays (parallel task agents can have + # several outstanding), each card followed once by the cached LLM + # verdicts — without this replay, a reconnecting / refreshing tab + # sees the approve_request prompts but no judge chips, and since + # intent_verdict only fires once per call_id (no push to a late + # subscriber), the chips would never appear until the operator + # re-invokes the action. Mirrors the interactive path in + # ``turnstone/server.py`` (fresh-connect replay). + cards_fn = getattr(ui, "pending_approval_cards", None) + pending_cards = cards_fn() if callable(cards_fn) else [] + if pending_cards: + yield from pending_cards llm_verdicts = getattr(ui, "_llm_verdicts", None) ws_lock = getattr(ui, "_ws_lock", None) if llm_verdicts and ws_lock is not None: diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 97c4a72b..edd44f5a 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -708,12 +708,15 @@ def make_approve_handler( ) -> Handler: """Lifted body for ``POST {prefix}/{ws_id}/approve``. - Resolves a pending tool approval on the workstream's UI. Both - kinds expose the same approve / feedback / always body shape and - the same ``ui.resolve_approval(approved, feedback)`` mechanic; - differences are auth scope, manager lookup, and the - ``__budget_override__`` filter (interactive-only — coord workstreams - don't have the budget-override pseudo-tool). + Resolves ONE pending approval cycle on the workstream's UI. Both + kinds expose the same approve / feedback / always / call_id / + cycle_id body shape and the same cycle-routed + ``ui.resolve_approval(...)`` mechanic; differences are auth scope, + manager lookup, and the ``__budget_override__`` filter + (interactive-only — coord workstreams don't have the + budget-override pseudo-tool). With parallel task agents a + workstream can hold several cycles; a body without a selector + resolves the oldest. ``accepted_permissions`` is OR-checked via :func:`require_any_permission` only when ``cfg.permission_gate`` is ``None`` — i.e. for the @@ -766,49 +769,140 @@ def make_approve_handler( {"error": "session UI does not support approval"}, status_code=409, ) - # ``_pending_approval`` and ``auto_approve_tools`` aren't on the - # ``SessionUI`` Protocol — both interactive ``WebUI`` and - # ``ConsoleCoordinatorUI`` add them, but a kind-agnostic body - # has to look them up dynamically. The CLI ``CliUI`` wouldn't - # have either, so accessing through ``getattr`` is also safer. - pending = getattr(ui, "_pending_approval", None) auto_approve_tools = getattr(ui, "auto_approve_tools", None) - # call_id guard — when the body sends a call_id, it must - # match one of the currently-pending items. Stops a stale - # row in the coordinator's children tree (where the operator - # clicked approve on call A) from silently resolving an - # unrelated call B that took A's place after the row was - # rendered. Empty/missing call_id preserves backward-compat - # for clients (CLI, channel adapters) that don't track it. + # Cycle routing — with parallel task agents a workstream can + # have SEVERAL approval cycles live at once, each its own + # prompt. A decision addresses exactly one: + # - ``cycle_id`` (new clients) selects it directly; + # - ``call_id`` (coord tree rows, channel adapters) selects + # the cycle containing that call — and doubles as the + # legacy stale-guard: a click on a row whose round was + # already replaced 409s instead of silently resolving an + # unrelated batch; + # - neither (CLI wrappers, old tabs) → the OLDEST live cycle, + # matching the order the prompts were issued. body_call_id_raw = body.get("call_id", "") body_call_id = body_call_id_raw.strip() if isinstance(body_call_id_raw, str) else "" - if body_call_id: - if pending is None: - return JSONResponse( - {"error": "no pending approval", "current_call_id": None}, - status_code=409, - ) - pending_items = pending.get("items") or [] - pending_call_ids = { - item.get("call_id", "") for item in pending_items if item.get("call_id") - } - if body_call_id not in pending_call_ids: - # Primary = first non-empty call_id in list order, matching - # ``serialize_pending_approval_detail``. The two definitions - # must agree so the UI can re-render against the same - # identifier the server reports as current. + body_cycle_id_raw = body.get("cycle_id", "") + body_cycle_id = body_cycle_id_raw.strip() if isinstance(body_cycle_id_raw, str) else "" + find_cycle = getattr(ui, "find_approval_cycle", None) + target_card: dict[str, Any] | None = None + pinned_cycle_id: str | None = None + if find_cycle is not None: + target_card = find_cycle(cycle_id=body_cycle_id or None, call_id=body_call_id or None) + if target_card is None and (body_cycle_id or body_call_id): + # Selector given but nothing matched: the round was + # resolved/replaced after this client rendered it. + # Report the CURRENT oldest cycle (first entry of + # ``serialize_pending_approval_details``) so the client + # can re-render against what the server thinks is live. + current = find_cycle() + current_items = (current or {}).get("items") or [] primary = next( - (item.get("call_id", "") for item in pending_items if item.get("call_id")), + (item.get("call_id", "") for item in current_items if item.get("call_id")), None, ) return JSONResponse( - {"error": "stale call_id", "current_call_id": primary}, + { + "error": ("stale call_id" if body_call_id else "stale cycle_id"), + "current_call_id": primary, + "current_cycle_id": (current or {}).get("cycle_id"), + }, status_code=409, ) - if always and approved and pending and auto_approve_tools is not None: + # Pin the resolution to the exact cycle the lookup returned. + # For selector-less bodies the lookup and the resolve would + # otherwise EACH independently pick "the oldest" — a cycle + # resolving in the gap (gate timeout, peer tab, smart + # approval) silently retargets the resolve at the next + # cycle while the always-names below were collected from + # the previous one, whitelisting a batch the operator never + # looked at. + pinned_cycle_id = (target_card or {}).get("cycle_id") or None + else: + # Legacy/stub UI (tests, external SessionUI impls): fall back + # to the single-slot view for the always-names read below. + target_card = getattr(ui, "_pending_approval", None) + if body_call_id: + if target_card is None: + return JSONResponse( + {"error": "no pending approval", "current_call_id": None}, + status_code=409, + ) + legacy_ids = { + item.get("call_id", "") + for item in target_card.get("items") or [] + if item.get("call_id") + } + if body_call_id not in legacy_ids: + primary = next(iter(sorted(legacy_ids)), None) + return JSONResponse( + {"error": "stale call_id", "current_call_id": primary}, + status_code=409, + ) + # Resolve FIRST, then whitelist: the "Approve + Always" names + # must describe the cycle that actually resolved. On the cycle + # path the resolve is pinned to the lookup's cycle_id, so the + # only race left is that cycle resolving in the gap — then + # ``resolved_cycle`` comes back ``None`` and the whitelist below + # is skipped (approving a card someone else already resolved + # must not grow the auto-approve set). ``always`` still rides + # the ``approval_resolved`` SSE event so peer tabs that didn't + # click can render the right status pill ("✓ approved · always" + # vs plain "✓ approved") without a side-channel broadcast. + try: + if find_cycle is not None: + if pinned_cycle_id is not None: + resolved_cycle = ui.resolve_approval( + approved, + feedback, + always=always, + cycle_id=pinned_cycle_id, + ) + elif body_call_id or body_cycle_id: + # Lookup matched a card that carries no cycle_id + # (custom registrations outside ``approve_tools``): + # honor the client's own selector. + resolved_cycle = ui.resolve_approval( + approved, + feedback, + always=always, + call_id=body_call_id or None, + cycle_id=body_cycle_id or None, + ) + else: + # No selector AND nothing pending at lookup time: + # resolve nothing rather than racing a cycle that + # registered in the gap — the client can't have + # been looking at it. + resolved_cycle = None + else: + resolved_cycle = ui.resolve_approval( + approved, + feedback, + always=always, + call_id=body_call_id or None, + cycle_id=body_cycle_id or None, + ) + except TypeError: + # Pre-cycle SessionUI impls (external/custom) without the + # selector kwargs. + resolved_cycle = ui.resolve_approval(approved, feedback, always=always) + if ( + always + and approved + and target_card + and auto_approve_tools is not None + # Cycle-registry UIs: whitelist only when OUR resolve landed + # on the pinned cycle (non-None return). Stub/legacy UIs + # (no registry) keep the unconditional legacy behavior — + # their resolve's return value carries no cycle contract to + # gate on. + and (find_cycle is None or resolved_cycle is not None) + ): tool_names: set[str] = { it.get("approval_label", "") or it.get("func_name", "") - for it in pending.get("items", []) + for it in target_card.get("items", []) if it.get("needs_approval") and it.get("func_name") and not it.get("error") } tool_names.discard("") @@ -828,13 +922,7 @@ def make_approve_handler( if source_map is not None: for t in tool_names: source_map[t] = AutoApproveReason.ALWAYS - # Forward ``always`` so the resulting ``approval_resolved`` SSE - # event carries the intent — peer tabs that didn't click but - # are subscribed to the same workstream can render the right - # status pill ("✓ approved · always" vs plain "✓ approved") - # without needing a side-channel broadcast. - ui.resolve_approval(approved, feedback, always=always) - return JSONResponse({"status": "ok"}) + return JSONResponse({"status": "ok", "cycle_id": resolved_cycle}) return approve @@ -1154,13 +1242,13 @@ def make_cancel_handler( ``coord_mgr.cancel`` which silently no-op'd on a placeholder; the lifted body 400s for parity with interactive's existing "No session" branch. - - ``resolve_approval`` is **gated on ``ui._pending_approval is not None``** - because :meth:`SessionUIBase.resolve_approval` is *not* - idempotent — it always broadcasts ``approval_resolved`` and - overwrites ``_approval_result``. Without the gate, every idle - cancel would leak a stale resolution event to SSE listeners. - The gate preserves the recovery semantics for the genuine - stuck case while skipping the broadcast on idle cancels. + - Pending approvals are denied via ``resolve_all_approvals`` — + cancel addresses the workstream, so EVERY live cycle (parallel + task agents can park several gates at once) wakes with its own + denied result. The sweep is a no-op when nothing is pending + (no stale ``approval_resolved`` broadcast on idle cancels); + legacy/stub UIs without it fall back to the old single-slot + ``resolve_approval`` gated on ``_pending_approval``. """ async def cancel(request: Request) -> Response: @@ -1218,29 +1306,34 @@ def make_cancel_handler( dropped = {} # Always set the cooperative cancel flag — cheap, no harm if - # nothing's running. resolve_approval is gated on its - # ``_pending_approval`` slot: pre-lift coord called it - # unconditionally via ``mgr.cancel`` (which is - # recovery-friendly: a stuck approval-pending state from a - # crashed worker can still be cleared), but ``resolve_approval`` - # is NOT idempotent — calling it with no pending approval - # broadcasts a stale ``approval_resolved`` SSE event and - # overwrites ``_approval_result``. Gating on the pending slot - # preserves the recovery semantics for the actual stuck case - # while skipping the broadcast on idle cancels. + # nothing's running. Cancel addresses the WORKSTREAM, so it + # denies EVERY live approval cycle: with parallel task agents + # several gate threads can be parked at once and each must wake + # with its own (denied) result. ``resolve_all_approvals`` is a + # no-op with no live cycles (returns 0, no SSE broadcast), so + # idle cancels stay silent — the same property the old + # pending-slot gate provided; the recovery semantics for a + # stuck approval-pending state are preserved because a stuck + # cycle IS a live cycle. Legacy/stub UIs without the sweep + # keep the old single-slot fallback. try: session.cancel() except Exception: log.debug("ws.cancel.session_failed ws=%s", ws_id[:8], exc_info=True) - if hasattr(ui, "resolve_approval") and getattr(ui, "_pending_approval", None) is not None: - try: + try: + if hasattr(ui, "resolve_all_approvals"): + ui.resolve_all_approvals(False, "Cancelled by user") + elif ( + hasattr(ui, "resolve_approval") + and getattr(ui, "_pending_approval", None) is not None + ): ui.resolve_approval(False, "Cancelled by user") - except Exception: - log.debug( - "ws.cancel.resolve_approval_failed ws=%s", - ws_id[:8], - exc_info=True, - ) + except Exception: + log.debug( + "ws.cancel.resolve_approval_failed ws=%s", + ws_id[:8], + exc_info=True, + ) # The remaining steps only matter when a worker is actually # running: force-recovery has nothing to recover otherwise, @@ -3630,9 +3723,10 @@ def make_detail_handler(cfg: SessionEndpointConfig) -> Handler: if not ws_id: return JSONResponse({"error": "ws_id is required"}, status_code=400) - # Cross-tenant gate. PR 447 added ``pending_approval_detail`` - # to the response (tool previews, function arguments, LLM - # judge reasoning) — a richer payload than the pre-PR + # Cross-tenant gate. PR 447 added the inline approval payload + # (now ``pending_approval_details``) to the response (tool + # previews, function arguments, LLM judge reasoning) — a + # richer payload than the pre-PR # ``{ws_id, name, state, user_id, kind}`` tuple. Coord wires # ``tenant_check=None`` (the cluster-wide ``admin.coordinator`` # permission_gate covers it); interactive wires @@ -3690,26 +3784,28 @@ def make_detail_handler(cfg: SessionEndpointConfig) -> Handler: # paint the inline approval gate from this single response # instead of waiting for the SSE approve_request replay (which # introduces a brief --running flash on reload). Both keys - # (``pending_approval`` + ``pending_approval_detail``) are + # (``pending_approval`` + ``pending_approval_details``) are # always present in the response: a UI that doesn't expose - # ``serialize_pending_approval_detail`` (CLI / channel - # adapters) reports ``False`` / ``null`` for them. The + # ``serialize_pending_approval_details`` (CLI / channel + # adapters) reports ``False`` / ``[]`` for them. The # ``_pending_approval`` lookup is asserted as ``dict`` (its - # only real production shape — see - # ``SessionUIBase._pending_approval``) so a MagicMock-based - # unit test or other non-dict sentinel doesn't trip the path. + # only real production shape — the oldest-cycle view kept by + # ``SessionUIBase``) so a MagicMock-based unit test or other + # non-dict sentinel doesn't trip the path. pending_approval = False - pending_approval_detail: dict[str, Any] | None = None + pending_approval_details: list[dict[str, Any]] = [] ui = ws.ui pending_raw = getattr(ui, "_pending_approval", None) if ui is not None else None if isinstance(pending_raw, dict): pending_approval = True - serializer = getattr(ui, "serialize_pending_approval_detail", None) + # Full per-cycle list — parallel task agents can have + # several prompts live; the reload path paints them all. + serializer = getattr(ui, "serialize_pending_approval_details", None) if callable(serializer): try: - serialized = serializer() - if isinstance(serialized, dict) or serialized is None: - pending_approval_detail = serialized + maybe_list = serializer() + if isinstance(maybe_list, list): + pending_approval_details = maybe_list except Exception: # Defensive: a malformed verdict object inside the # serializer shouldn't fail the entire detail @@ -3730,7 +3826,7 @@ def make_detail_handler(cfg: SessionEndpointConfig) -> Handler: "user_id": ws.user_id, "kind": ws.kind, "pending_approval": pending_approval, - "pending_approval_detail": pending_approval_detail, + "pending_approval_details": pending_approval_details, } ) diff --git a/turnstone/sdk/console.py b/turnstone/sdk/console.py index 5647d54c..aa9d0e20 100644 --- a/turnstone/sdk/console.py +++ b/turnstone/sdk/console.py @@ -421,13 +421,24 @@ class AsyncTurnstoneConsole(_BaseClient): approved: bool = True, feedback: str = "", always: bool = False, + cycle_id: str = "", + call_id: str = "", ) -> dict[str, Any]: - """Approve or reject a pending tool call via the routing proxy.""" + """Approve or reject a pending tool call via the routing proxy. + + ``cycle_id`` / ``call_id`` select the approval cycle when the + workstream has several live (parallel task agents); omitting + both resolves the oldest. + """ body: dict[str, Any] = {"approved": approved} if feedback: body["feedback"] = feedback if always: body["always"] = True + if cycle_id: + body["cycle_id"] = cycle_id + if call_id: + body["call_id"] = call_id return await self._request( "POST", f"/v1/api/route/workstreams/{ws_id}/approve", json_body=body ) @@ -1357,10 +1368,17 @@ class TurnstoneConsole: approved: bool = True, feedback: str = "", always: bool = False, + cycle_id: str = "", + call_id: str = "", ) -> dict[str, Any]: return self._runner.run( self._async.route_approve( - ws_id=ws_id, approved=approved, feedback=feedback, always=always + ws_id=ws_id, + approved=approved, + feedback=feedback, + always=always, + cycle_id=cycle_id, + call_id=call_id, ) ) diff --git a/turnstone/sdk/events.py b/turnstone/sdk/events.py index ed6e5640..216a65b6 100644 --- a/turnstone/sdk/events.py +++ b/turnstone/sdk/events.py @@ -157,16 +157,35 @@ class ToolInfoEvent(ServerEvent): @dataclass class ApproveRequestEvent(ServerEvent): + """One approval CYCLE awaiting the operator. + + ``cycle_id`` names the round: echo it back on the approve POST to + resolve exactly this batch. Several of these can be outstanding at + once — parallel task agents each gate their own tool calls — so + clients must key prompt UI by ``cycle_id``, not assume a singleton. + """ + type: str = "approve_request" + cycle_id: str = "" items: list[dict[str, Any]] = field(default_factory=list) judge_pending: bool = False @dataclass class ApprovalResolvedEvent(ServerEvent): + """A specific approval cycle resolved (by any connected client). + + ``cycle_id`` / ``call_ids`` identify WHICH prompt to dismiss — + with concurrent cycles a bare "something resolved" would dismiss + the wrong card. + """ + type: str = "approval_resolved" approved: bool = False feedback: str = "" + always: bool = False + cycle_id: str = "" + call_ids: list[str] = field(default_factory=list) @dataclass diff --git a/turnstone/sdk/server.py b/turnstone/sdk/server.py index e4222350..cb41f40e 100644 --- a/turnstone/sdk/server.py +++ b/turnstone/sdk/server.py @@ -276,12 +276,26 @@ class AsyncTurnstoneServer(_BaseClient): approved: bool = True, feedback: str | None = None, always: bool = False, + cycle_id: str | None = None, + call_id: str | None = None, ) -> StatusResponse: + """Resolve one approval cycle. + + ``cycle_id`` (from the ``approve_request`` event) or ``call_id`` + (any member call) selects the cycle; omitting both resolves the + OLDEST live one — ambiguous when parallel task agents have + several prompts outstanding, so pass a selector whenever the + triggering event is known. + """ body: dict[str, Any] = {"approved": approved} if feedback is not None: body["feedback"] = feedback if always: body["always"] = True + if cycle_id: + body["cycle_id"] = cycle_id + if call_id: + body["call_id"] = call_id return await self._request( "POST", f"/v1/api/workstreams/{ws_id}/approve", @@ -684,9 +698,18 @@ class TurnstoneServer: approved: bool = True, feedback: str | None = None, always: bool = False, + cycle_id: str | None = None, + call_id: str | None = None, ) -> StatusResponse: return self._runner.run( - self._async.approve(ws_id=ws_id, approved=approved, feedback=feedback, always=always) + self._async.approve( + ws_id=ws_id, + approved=approved, + feedback=feedback, + always=always, + cycle_id=cycle_id, + call_id=call_id, + ) ) def command(self, *, ws_id: str, command: str) -> StatusResponse: diff --git a/turnstone/server.py b/turnstone/server.py index fadb8754..b83479fb 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -256,12 +256,16 @@ class WebUI(SessionUIBase): feedback: str | None = None, *, always: bool = False, + cycle_id: str = "", + call_ids: tuple[str, ...] = (), ) -> None: """Send an ``approval_resolved`` decision to the global SSE channel. Clears the parent's pending-approval pill in lockstep with the actual decision rather than waiting for the next - state-change piggyback. + state-change piggyback. ``cycle_id`` / ``call_ids`` identify + WHICH cycle resolved so the coord tree clears the right block + when several are live (parallel task agents). """ if WebUI._global_queue is not None: with contextlib.suppress(queue.Full): @@ -272,6 +276,8 @@ class WebUI(SessionUIBase): "approved": approved, "feedback": feedback or "", "always": bool(always), + "cycle_id": cycle_id, + "call_ids": list(call_ids), } ) @@ -402,11 +408,15 @@ class WebUI(SessionUIBase): {"type": "ws_rename", "ws_id": self.ws_id, "name": name} ) - def on_intent_verdict(self, verdict: dict[str, Any]) -> None: + def on_intent_verdict( + self, + verdict: dict[str, Any], + judge_event: object | None = None, + ) -> None: """Extend :meth:`SessionUIBase.on_intent_verdict` with a node-level prometheus metric update. """ - super().on_intent_verdict(verdict) + super().on_intent_verdict(verdict, judge_event) fire_judge_verdict_metric(_metrics, verdict, "llm") # ``on_output_warning`` inherited from :class:`SessionUIBase`. @@ -746,9 +756,15 @@ def _interactive_events_replay( # Pending approval re-injection (so a reconnecting tab sees the # prompt) + cached LLM verdicts received since the prompt fired. - pending_approval = getattr(ui, "_pending_approval", None) - if pending_approval is not None: - yield pending_approval + # EVERY live cycle replays — parallel task agents can have several + # prompts outstanding, and a tab that repaints only the newest + # leaves the others unanswerable. Cards first (oldest-first), then + # the verdict cache once: clients route ``intent_verdict`` by + # call_id, so ordering across cards is irrelevant as long as every + # card exists before its verdicts. + pending_cards = ui.pending_approval_cards() if hasattr(ui, "pending_approval_cards") else [] + if pending_cards: + yield from pending_cards with ui._ws_lock: cached_verdicts = list(ui._llm_verdicts.values()) for v in cached_verdicts: @@ -881,16 +897,16 @@ def _build_node_snapshot(app_state: Any) -> dict[str, Any]: title = "" if ws.session: title = get_workstream_display_name(ws.session.ws_id) or "" - # ``pending_approval_detail`` mirrors the dashboard handler's + # ``pending_approval_details`` mirrors the dashboard handler's # projection so the console collector's reconnect-via-snapshot # path (``_reconcile_node``) can carry the rich approval payload # across reconnects — without it, a child sitting in approval- # pending across a console restart or network blip would render # with no buttons until the next state change. Same data, same # ``read`` scope as ``/v1/api/dashboard``. - approval_detail: dict[str, Any] | None = None - if ui is not None and hasattr(ui, "serialize_pending_approval_detail"): - approval_detail = ui.serialize_pending_approval_detail() + approval_details: list[dict[str, Any]] = [] + if ui is not None and hasattr(ui, "serialize_pending_approval_details"): + approval_details = ui.serialize_pending_approval_details() ws_list.append( { "id": ws.id, @@ -909,7 +925,7 @@ def _build_node_snapshot(app_state: Any) -> dict[str, Any]: "user_id": ws.user_id, "project_id": ws.project_id, "persona": ws.persona, - "pending_approval_detail": approval_detail, + "pending_approval_details": approval_details, } ) return { @@ -1132,7 +1148,7 @@ async def dashboard(request: Request) -> JSONResponse: "user_id": ws.user_id, "project_id": ws.project_id, "persona": ws.persona, - "pending_approval_detail": ui.serialize_pending_approval_detail(), + "pending_approval_details": ui.serialize_pending_approval_details(), # Per-ws ring buffer of recent auto-approves (last 10). # Lets the coord-tree render a "recently auto-approved # by skill X" pill without a per-child round-trip — the