fix(acp): scope permission-card retirement to terminating turn; fix fan-out false-ack

P1 (fan-out false-ack, lib.rs): The prior fix correctly gated nonce
recording and 'sent' status on all_tx_accepted (every tx-equipped loop
accepted). Adds the required regression test: saturate the owner queue
while leaving a sibling queue open, deliver a decision -> status must be
'channel_full' not 'sent', nonce not recorded; drain owner queue, retransmit
-> status 'sent', nonce recorded, owner loop receives the message.

P2 (channel-wide card retirement, agentSessionTranscript.ts): Replace
retireAllLivePermissionCards with retireLivePermissionCardsForTurn at both
turn_completed and turn_error/agent_panic call sites. With concurrent
thread-scoped turns, channel-wide retirement on turn B completing would
retire thread A's still-pending cards and delete their nonce indexes, making
a later A decision drop on an unknown nonce. Turn-scoped retirement confines
the backstop to cards whose turnId matches the terminating turn; falls back
to channel-wide only when no turn identity is present (legacy archive frames).

retireLivePermissionCardsForTurn added to agentSessionTranscriptPermissions.ts
with the same copy-on-write semantics as retireAllLivePermissionCards, scoping
both the itemsById scan (item.turnId === turnId) and the pendingPermissions
cleanup (ch:session:turn:id key format, parts[2] matches turnId).

Regression tests added (agentSessionTranscript.test.mjs):
- buildTranscript_turn_scoped_retirement_does_not_retire_sibling_thread_cards:
  thread A pending -> thread B turn_completed -> A still actionable -> A
  decision resolves correctly, including archive replay.
- buildTranscript_turn_error_scoped_retirement_does_not_retire_sibling_thread_cards:
  same invariant for the turn_error terminal path.

Mutation table:
- P1: restore any_sent semantics -> mixed-result test fails at 'sent'!=
  'channel_full' assertion.
- P2: restore retireAllLivePermissionCards -> both sibling-isolation tests
  fail because A's card is retired by B's terminal event.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
2026-09-01 19:23:57 -04:00
parent fbeb9fafb3
commit 5c8ec7a2ec
4 changed files with 402 additions and 13 deletions

View File

@@ -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

View File

@@ -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",
);
});

View File

@@ -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") {

View File

@@ -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