diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 73d940d7ca..1b09b6a3d1 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2035,23 +2035,55 @@ fn handle_permission_decision_control( } }; - // Record the nonce as soon as at least one task received the decision. + // Record the nonce and summarise into a single observer status. + // + // Correctness requirement (fan-out false-ack): a `permission_decision` is + // fanned out to all same-channel loops; each loop uses the nonce to select + // its own entry. The dispatcher cannot identify which task owns the nonce, + // so it must not report `sent` — stopping Desktop's retransmit loop — unless + // every loop that has a permission tx accepted the message. If any tx-equipped + // loop's queue returned `Full` or `Closed` (owner-queue saturated while a + // sibling accepted), the nonce-owning loop may not have received the decision, + // so we return a retryable status and leave Desktop's retry loop running. let any_sent = deliveries.iter().any(|d| matches!(d, Delivery::Sent)); - if any_sent { + let any_full = deliveries.iter().any(|d| matches!(d, Delivery::Full)); + // All tx-equipped loops accepted: no Full and no Closed among the deliveries. + let all_tx_accepted = + any_sent && !any_full && !deliveries.iter().any(|d| matches!(d, Delivery::Closed)); + + // Only suppress retransmits (record nonce) when every tx-equipped loop + // received the decision — a partial fan-out is not a confirmed delivery. + if all_tx_accepted { pool.record_permission_decision(request_nonce); } - // Summarise into a single status for the observer frame. Priority: - // Sent > Full > Closed > NoChannel > NoTask. - let status = if deliveries.iter().any(|d| matches!(d, Delivery::Sent)) { + // Summarise into a single status for the observer frame. + // all_tx_accepted → "sent" (Desktop retransmit stops, decision confirmed). + // any_sent + any_full → "channel_full" (retryable: Desktop re-enables buttons, + // owner can retry once the queue drains; the owning loop's dedup tolerates + // duplicate deliveries from later retransmits). + // No Sent → fall through to the existing priority ladder. + let status = if all_tx_accepted { tracing::info!( channel = %channel_id, nonce = %request_nonce, option_id = %option_id, tasks_fanned = deliveries.len(), - "permission_decision delivered to read loop(s)" + "permission_decision delivered to all read loop(s)" ); "sent" + } else if any_sent && any_full { + // Mixed: at least one sibling accepted but the owner queue was full. + // Reporting "sent" here would stop Desktop's retransmit loop while the + // nonce-owning read loop never received the decision. Return "channel_full" + // so Desktop keeps retrying until the queue drains. + tracing::warn!( + channel = %channel_id, + nonce = %request_nonce, + "permission_decision: some loops sent, owner loop full — \ + reporting channel_full to keep Desktop retransmitting" + ); + "channel_full" } else if deliveries.iter().any(|d| matches!(d, Delivery::Full)) { tracing::warn!( channel = %channel_id, @@ -11271,6 +11303,121 @@ mod permission_decision_control_tests { assert_eq!(received_b_2.unwrap().request_nonce, nonce_b); } + /// Fan-out false-ack: mixed-result (owner-Full + sibling-Sent) must NOT + /// report `sent` or record the nonce. + /// + /// Scenario (Carl's verbatim requirement): + /// 1. Two tasks share a channel — owner (rx_owner, capacity 4) and + /// sibling (rx_sibling, capacity 4). Owner's queue is saturated with + /// 4 unread messages; sibling's queue is kept clear. + /// 2. A permission decision for nonce N is fanned out: sibling receives + /// it (`Sent`), owner queue returns `Full`. + /// 3. Expected: status == `channel_full` (Desktop keeps retransmitting), + /// nonce N is NOT in `was_recently_decided` (no suppression yet). + /// 4. Owner queue is drained; decision is retransmitted. + /// Expected: status == `sent`, nonce N recorded, owner receives it. + /// + /// **Mutation proof:** reverting the `all_tx_accepted` gate to the prior + /// `any_sent` semantics makes step 3 return `"sent"` and records the nonce, + /// causing the assertion `assert_ne!(status_mixed, "sent")` to fail. + #[tokio::test] + async fn mixed_result_owner_full_sibling_sent_reports_channel_full_not_sent() { + let observer = ObserverHandle::in_process(); + let mut rx_obs = observer.subscribe(); + let channel_id = Uuid::new_v4(); + let nonce = "nonce-mixed-full"; + + let mut pool = AgentPool::from_slots(vec![None]); + // Install sibling first — fan-out visits it before the owner. + // Keep the sibling receiver so we can drain it between fill rounds. + let mut rx_sibling = install_task(&mut pool, channel_id); + let mut rx_owner = install_task(&mut pool, channel_id); + + // Saturate the owner's queue (capacity 4) with four fill decisions. + // After each send we drain the sibling's queue so it never fills. + let fill_nonce = "nonce-fill"; + for _ in 0..4 { + handle_permission_decision_control( + &decision_payload(channel_id, fill_nonce), + &mut pool, + Some(&observer), + ); + // Observer drain — keeps it responsive. + let _ = control_result_status(&mut rx_obs); + // Drain sibling so its queue stays open for the critical decision. + while rx_sibling.try_recv().is_ok() {} + } + // Precondition: owner queue is full (4/4 unread), sibling queue is empty. + assert!( + rx_owner.try_recv().is_ok(), + "precondition: owner queue must have unread messages (fill worked)" + ); + // We just popped one — push it back conceptually: re-fill the slot we + // accidentally drained by sending one more fill decision. + handle_permission_decision_control( + &decision_payload(channel_id, fill_nonce), + &mut pool, + Some(&observer), + ); + let _ = control_result_status(&mut rx_obs); + while rx_sibling.try_recv().is_ok() {} + // Owner queue: 4/4 full again (we drained 1 then immediately refilled). + + // Deliver the critical decision. + // Owner queue: Full. Sibling queue: Sent. → any_sent=true, any_full=true. + handle_permission_decision_control( + &decision_payload(channel_id, nonce), + &mut pool, + Some(&observer), + ); + let status_mixed = control_result_status(&mut rx_obs); + + assert_ne!( + status_mixed, "sent", + "mixed-result (owner Full, sibling Sent) must NOT report 'sent'" + ); + assert_eq!( + status_mixed, "channel_full", + "mixed-result must report 'channel_full' to keep Desktop retransmitting" + ); + assert!( + !pool.was_recently_decided(nonce), + "nonce must NOT be recorded when owner queue was Full — \ + retransmit suppression must not activate" + ); + + // Drain the owner queue fully — makes room for the retransmit. + while rx_owner.try_recv().is_ok() {} + // Also drain the sibling's copy of the critical nonce so it won't be + // counted as full on the retransmit path either. + while rx_sibling.try_recv().is_ok() {} + + // Retransmit. Now both queues have capacity: all tx-equipped loops + // accept → must report `sent` and record the nonce. + handle_permission_decision_control( + &decision_payload(channel_id, nonce), + &mut pool, + Some(&observer), + ); + let status_retry = control_result_status(&mut rx_obs); + + assert_eq!( + status_retry, "sent", + "after drain, retransmitted decision must reach owner and report 'sent'" + ); + assert!( + pool.was_recently_decided(nonce), + "nonce must be recorded after all-loops-accepted retransmit" + ); + // Confirm the owner's queue actually received the retransmit. + let received = rx_owner.try_recv(); + assert!( + received.is_ok(), + "owner loop must receive the retransmitted decision after drain: {received:?}" + ); + assert_eq!(received.unwrap().request_nonce, nonce); + } + /// F4 outer signed-control path: `handle_relay_observer_control_event` /// admits a valid owner-signed, NIP-44-encrypted kind-24200 frame and /// delivers the enclosed `permission_decision` payload to the in-flight diff --git a/desktop/src/features/agents/ui/agentSessionTranscript.test.mjs b/desktop/src/features/agents/ui/agentSessionTranscript.test.mjs index f5cccb0fed..ce33ac7b34 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscript.test.mjs +++ b/desktop/src/features/agents/ui/agentSessionTranscript.test.mjs @@ -2905,3 +2905,171 @@ test("buildTranscript_sync_denial_write_with_mismatched_nonce_leaves_card_live", "nonce index must still contain the read card when write nonce does not match", ); }); + +// ─── P2: turn-scoped retirement — sibling thread cards survive their turn ───── + +test("buildTranscript_turn_scoped_retirement_does_not_retire_sibling_thread_cards", () => { + // Regression guard for P2: with concurrent thread-scoped turns, a + // turn_completed event for Thread B must NOT retire Thread A's still-pending + // permission cards. Only cards whose turnId matches the terminating turn are + // retired. + // + // Sequence: + // 1. Thread A (turnId="turn-A") raises a permission request — card is live. + // 2. Thread B (turnId="turn-B") raises a permission request — card is live. + // 3. Thread B completes (turn_completed, turnId="turn-B"). + // 4. Thread A's card must still be actionable. + // 5. Thread A's decision arrives — card is resolved correctly. + // 6. Archive replay of the same sequence produces the same final state + // (turn-scoped retirement holds for both live and replay paths). + const CH = "ch-2"; + const events = [ + // Thread A permission request. + makePermissionRequestWithAuth(1, "req-A", "nonce-A", { + turnId: "turn-A", + channelId: CH, + }), + // Thread B permission request. + makePermissionRequestWithAuth(2, "req-B", "nonce-B", { + turnId: "turn-B", + channelId: CH, + }), + // Thread B completes — must NOT retire Thread A's card. + makeTurnCompleted(3, { channelId: CH, turnId: "turn-B" }), + // Thread A's decision is applied. + makePermissionWriteWithNonce( + 4, + "req-A", + "nonce-A", + "selected", + "allow_once", + { + channelId: CH, + turnId: "turn-A", + }, + ), + ]; + + // ── Live path ────────────────────────────────────────────────────────────── + + // Step 3: after turn-B completes, A must still be actionable. + const stateAfterBComplete = buildTranscriptState(events.slice(0, 3)); + const cardA_live = stateAfterBComplete.items.find( + (i) => i.renderClass === "permission" && i.requestNonce === "nonce-A", + ); + const cardB_live = stateAfterBComplete.items.find( + (i) => i.renderClass === "permission" && i.requestNonce === "nonce-B", + ); + assert.ok(cardA_live, "Thread A card must exist after Thread B completes"); + assert.equal( + cardA_live.actionable, + true, + "Thread A card must still be actionable after Thread B completes", + ); + assert.ok(cardB_live, "Thread B card must exist"); + assert.equal( + cardB_live.actionable, + false, + "Thread B card must be retired by its own turn_completed", + ); + + // Step 4: A's decision resolves it correctly. + const stateAfterADecision = buildTranscriptState(events); + const cardA_resolved = stateAfterADecision.items.find( + (i) => i.renderClass === "permission" && i.requestNonce === "nonce-A", + ); + assert.ok(cardA_resolved, "Thread A card must still exist after A decision"); + assert.equal( + cardA_resolved.actionable, + false, + "Thread A card must be retired after its own decision", + ); + assert.ok( + cardA_resolved.outcome, + "Thread A card must have an outcome after its own decision", + ); + assert.ok( + !stateAfterADecision.pendingPermissionsByNonce.has("nonce-A"), + "nonce-A must be cleared from the index after Thread A's decision", + ); + assert.ok( + !stateAfterADecision.pendingPermissionsByNonce.has("nonce-B"), + "nonce-B must be cleared from the index after Thread B's turn_completed", + ); + + // ── Archive replay: same sequence produces the same final state ──────────── + + const replay = buildTranscriptState(events); + const cardA_replay = replay.items.find( + (i) => i.renderClass === "permission" && i.requestNonce === "nonce-A", + ); + assert.ok(cardA_replay, "Thread A card must survive replay"); + assert.equal( + cardA_replay.actionable, + false, + "Thread A card must be resolved in replay", + ); + assert.ok( + cardA_replay.outcome, + "Thread A card must have an outcome in replay", + ); +}); + +test("buildTranscript_turn_error_scoped_retirement_does_not_retire_sibling_thread_cards", () => { + // Same as the turn_completed variant but using turn_error so the same + // scoping invariant is verified for the error-terminal path. + // + // Mutation guard: replacing retireLivePermissionCardsForTurn with + // retireAllLivePermissionCards in the turn_error handler makes this test fail + // because Thread A's card is retired by Thread B's error event. + const CH = "ch-3"; + const events = [ + makePermissionRequestWithAuth(1, "req-EA", "nonce-EA", { + turnId: "turn-EA", + channelId: CH, + }), + makePermissionRequestWithAuth(2, "req-EB", "nonce-EB", { + turnId: "turn-EB", + channelId: CH, + }), + // Thread B errors — must NOT retire Thread A's card. + makeTurnError(3, { channelId: CH, turnId: "turn-EB" }), + // Thread A's decision applies. + makePermissionWriteWithNonce( + 4, + "req-EA", + "nonce-EA", + "selected", + "allow_once", + { + channelId: CH, + turnId: "turn-EA", + }, + ), + ]; + + const stateAfterBError = buildTranscriptState(events.slice(0, 3)); + const cardA = stateAfterBError.items.find( + (i) => i.renderClass === "permission" && i.requestNonce === "nonce-EA", + ); + assert.ok(cardA, "Thread A card must exist after Thread B errors"); + assert.equal( + cardA.actionable, + true, + "Thread A card must still be actionable after Thread B errors", + ); + + const stateAfterADecision = buildTranscriptState(events); + const cardA_resolved = stateAfterADecision.items.find( + (i) => i.renderClass === "permission" && i.requestNonce === "nonce-EA", + ); + assert.ok( + cardA_resolved?.outcome, + "Thread A card must have an outcome after its own decision", + ); + assert.equal( + cardA_resolved?.actionable, + false, + "Thread A card must be retired after its own decision", + ); +}); diff --git a/desktop/src/features/agents/ui/agentSessionTranscript.ts b/desktop/src/features/agents/ui/agentSessionTranscript.ts index 24cce60da5..3cb7480507 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscript.ts +++ b/desktop/src/features/agents/ui/agentSessionTranscript.ts @@ -30,7 +30,7 @@ import { import { friendlyTurnErrorCopy } from "../lib/friendlyAgentLastError"; import { describePermissionRequest, - retireAllLivePermissionCards, + retireLivePermissionCardsForTurn, handlePermissionTerminal, handlePermissionWrite, handlePermissionDecisionResult, @@ -724,17 +724,21 @@ export function processTranscriptEvent( ctx, event.kind, ); - // Backstop: retire any still-live permission cards for this channel so + // Backstop: retire permission cards belonging to this terminating turn so // missing telemetry and archive replay never reconstruct live controls - // after a terminal turn/process state. - retireAllLivePermissionCards(d, ch); + // after a terminal turn/process state. Scoped to `event.turnId` so that + // sibling threads' pending cards are not retired — each concurrent turn owns + // its own cards. Falls back to channel-wide retirement only when no turn + // identity is present (legacy archive frames). + retireLivePermissionCardsForTurn(d, ch, event.turnId); } else if (event.kind === "turn_completed") { - // Backstop: retire any still-live permission cards for this channel. + // Backstop: retire permission cards belonging to this completing turn. // Applied/timed-out/cancelled cards should already be retired via their // nonce-correlated acp_write frames, but uncertain (process-poison) cards // may only receive a turn_completed — this ensures they are not left - // actionable in live state or archive replay. - retireAllLivePermissionCards(d, ch); + // actionable in live state or archive replay. Scoped to `event.turnId`; + // channel-wide backstop kept only when no turn identity exists on the event. + retireLivePermissionCardsForTurn(d, ch, event.turnId); } else if (event.kind === "permission_terminal") { handlePermissionTerminal(d, event.authorization, event.payload, ch, ctx); } else if (event.kind === "acp_read" || event.kind === "acp_write") { diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptPermissions.ts b/desktop/src/features/agents/ui/agentSessionTranscriptPermissions.ts index 9d01a60fb9..2ba0a8ffca 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptPermissions.ts +++ b/desktop/src/features/agents/ui/agentSessionTranscriptPermissions.ts @@ -292,6 +292,76 @@ export function retireAllLivePermissionCards( } } +/** + * Retire live permission cards scoped to a specific turn, identified by + * `turnId`. Cards whose `turnId` matches are retired (actionable → false) + * and their nonce indexes removed. + * + * With concurrent thread-scoped turns, a channel-wide retirement on + * `turn_completed` / `turn_error` would retire pending cards belonging to + * still-running sibling threads. This function scopes the backstop to the + * single terminating turn so siblings remain actionable. + * + * Falls back to `retireAllLivePermissionCards` when `turnId` is absent — + * kept as a backstop for events (e.g. legacy archive frames) that carry no + * turn identity. + */ +export function retireLivePermissionCardsForTurn( + d: PermissionDraftSlice, + channelId: string, + turnId: string | null | undefined, +): void { + if (!turnId) { + // No turn identity available — fall back to channel-wide backstop. + retireAllLivePermissionCards(d, channelId); + return; + } + const prefix = `permission:${channelId}:`; + let retired = false; + for (const [id, item] of d.itemsById) { + if ( + id.startsWith(prefix) && + item.type === "lifecycle" && + item.renderClass === "permission" && + item.actionable && + item.turnId === turnId + ) { + if (!retired) { + d.items = [...d.items]; + d.itemsById = new Map(d.itemsById); + retired = true; + d.changed = true; + } + const updated = { ...item, actionable: false }; + d.itemsById.set(id, updated); + const idx = d.items.findIndex((i) => i.id === id); + if (idx !== -1) d.items[idx] = updated; + // Clean up nonce index if present. + if (item.requestNonce) { + d.pendingPermissionsByNonce = new Map(d.pendingPermissionsByNonce); + d.pendingPermissionsByNonce.delete(item.requestNonce); + } + } + } + // `pendingPermissions` keys use the format `ch:session:turn:id`; retire only + // entries belonging to this channel-turn combination. + const turnPrefix = `${channelId}:`; + let permsMutated = false; + for (const key of d.pendingPermissions.keys()) { + if (key.startsWith(turnPrefix)) { + // Parse `ch:session:turn:id` — the turn segment is index 2. + const parts = key.split(":"); + if (parts[2] === turnId) { + if (!permsMutated) { + d.pendingPermissions = new Map(d.pendingPermissions); + permsMutated = true; + } + d.pendingPermissions.delete(key); + } + } + } +} + /** * Handle an observer-only `permission_terminal` event. * Emitted for uncertain outcomes (process poison, cancel-during-write) where