From b555858b7fee3c865bce43f41f20f2c8c4ff356d Mon Sep 17 00:00:00 2001 From: QuicksilverSlick Date: Fri, 11 Sep 2026 10:47:05 -0500 Subject: [PATCH 1/2] feat(buzz-acp): park guest permission requests until an approver answers Slice A of the approval loop. A guest turn can now park a session/request_permission instead of refusing it: nothing is written to the agent until a verdict arrives on that request's own oneshot, or the wait expires to reject_once. Nothing routes requests to the owner yet (slice B), and shared agents still run in bypassPermissions (slice C), so behaviour in production is unchanged. - approval_gate.rs (new, pure): the parking rules and limits (600 s wait, 4 parked per connection, 6 asks per turn, a 30 s minimum window, a 10 s margin inside the hard deadline); RequestDigest, which binds each verdict to its request; DenyReason; AllowVia, with standing grants refunded against the per-turn cap; and TurnClock. While a request waits, the idle deadline is suspended and the hard deadline is credited for the wait, up to 3 timeouts. - pending_permissions.rs: parked asks, retirement, and a lazy verdict poll. - acp.rs: the read loop polls verdicts first under `biased`, expires overdue asks (closing the channel before writing the refusal), and answers every held or owed request on exit and on cancel. - Review fixes, each with a test that fails without it: - the prompt stops counting as in flight once its result or error is read, before the exit drain; - a new prompt first answers ids an earlier turn left owed; - the dropped-future cancel path resets authority to Guest; - an expired ask is closed before its refusal is written. Windows: 941 pass. The 32 failures are the 29 known timing failures plus the 3 wire tests that need Unix bash. Co-Authored-By: Claude Opus 5 --- crates/buzz-acp/src/acp.rs | 2029 ++++++++++++++++++-- crates/buzz-acp/src/approval_gate.rs | 1062 ++++++++++ crates/buzz-acp/src/lib.rs | 1 + crates/buzz-acp/src/pending_permissions.rs | 409 +++- crates/buzz-acp/src/pool.rs | 3 + 5 files changed, 3344 insertions(+), 160 deletions(-) create mode 100644 crates/buzz-acp/src/approval_gate.rs diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 0799b173be8..75bd8bb3a32 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -13,7 +13,13 @@ use tokio::io::AsyncWriteExt; use tokio::process::{Child, ChildStdin, ChildStdout}; use tokio_util::codec::{FramedRead, LinesCodec, LinesCodecError}; +use crate::approval_gate::{ + admit, request_digest, resolve, tool_label, Admission, ApprovalAsk, ApprovalEvent, + ApprovalLink, ApprovalSettled, ApprovalVerdict, Fire, ParkGate, ParkId, RefusalReason, + Settlement, TurnClock, VerdictOutcome, +}; use crate::observer::{ObserverContext, ObserverHandle}; +use crate::pending_permissions::ParkedAsk; use crate::usage::{ PromptResponseUsage, StandardAdapterKind, StandardUsageTracker, TurnUsage, UsageTracker, }; @@ -179,6 +185,12 @@ pub struct AcpClient { /// has no idea whether a tool call traces back to the owner or to someone /// on the allowlist. Defaults to `Guest`, so a path that forgets to set it /// is treated as untrusted rather than trusted. + /// + /// Reset to `Guest` when the prompt returns, and by a cancel that takes + /// over a turn whose prompt future was dropped before it could return. + /// Only the prompt loop reads it today, but an owner turn's trust left + /// lying on the connection is one careless read away from approving a + /// later guest's request. current_turn_authority: TurnAuthority, /// Permission requests refused during the turn in flight. /// @@ -501,6 +513,10 @@ pub struct PermissionRefusal { /// Tool title or kind as the agent reported it, for the person reading the /// notice. `(untitled)` when the agent sent neither. pub tool: String, + /// Why it was refused. A request refused before anyone was asked reads + /// differently to the person who made it than one an approver declined, + /// and this is the only thing that tells the two apart. + pub reason: RefusalReason, } impl AcpClient { @@ -850,6 +866,7 @@ impl AcpClient { idle_timeout: std::time::Duration, max_duration: std::time::Duration, authority: TurnAuthority, + approvals: Option, ) -> Result { self.session_prompt_blocks_with_idle_timeout( session_id, @@ -857,6 +874,7 @@ impl AcpClient { idle_timeout, max_duration, authority, + approvals, ) .await } @@ -874,12 +892,19 @@ impl AcpClient { idle_timeout: std::time::Duration, max_duration: std::time::Duration, authority: TurnAuthority, + approvals: Option, ) -> Result { // A required parameter rather than a setter: every dispatch has to // decide, and the compiler says so. A setter would let a new call site // inherit whatever the previous turn left behind, which for a shared // agent means one collaborator's turn running under the authority of // whoever prompted last. + // + // `approvals` is required for the same reason, and for one more: a + // link installed on the connection would be taken by whichever read + // loop ran first. A new session's `initial_message` runs its own + // prompt before the turn's, so that turn could never park. `None` + // means no request can wait on a person. self.current_turn_authority = authority; // Cleared per turn: a refusal from a previous turn reported against // this one would name the wrong request to the wrong person. @@ -905,22 +930,46 @@ impl AcpClient { "params": params, }); - tracing::debug!(target: "acp::wire", "→ {}", &serde_json::to_string(&msg).unwrap_or_default()); - if let Err(e) = self.write_ndjson(&msg).await { + // Answer whatever an earlier turn left owed, `cancelled`, before this + // prompt goes out. A prompt future dropped in its exit drain, after + // its response arrived, leaves no prompt in flight, so the pool runs no + // cancel and the ids that drain had not reached stay owed. Left owed + // into this turn, a new request that reuses one of those ids would be + // ignored as a retransmit, and the agent would wait on it forever. + // + // After `last_prompt_id` is set, never before. These writes can yield, + // and `pool.rs` relies on this call setting it before its first yield + // point (see its Race 1 branch): a control signal that won the + // `select!` here would otherwise find no prompt in flight and report + // this turn completed when its prompt was never sent. + let sent = match self.answer_owed_cancelled().await { + Ok(()) => { + tracing::debug!(target: "acp::wire", "→ {}", &serde_json::to_string(&msg).unwrap_or_default()); + self.write_ndjson(&msg).await + } + Err(e) => Err(e), + }; + if let Err(e) = sent { self.last_prompt_id = None; self.current_hard_deadline = None; + self.current_turn_authority = TurnAuthority::Guest; return Err(e); } let result = self - .read_until_response_with_idle_timeout( + .read_until_response_gated( session_id, id, idle_timeout, hard_deadline, max_duration, + approvals, ) .await; + // The turn is over, however it ended: nothing on this connection may + // be decided with its authority again. The next prompt sets its own, + // and a cancel drain that follows a timeout decides as a guest. + self.current_turn_authority = TurnAuthority::Guest; // On timeout errors, leave current_hard_deadline set so cancel_with_cleanup // can inherit the remaining budget. Clear it on all other outcomes. @@ -1048,8 +1097,8 @@ impl AcpClient { /// Cancel a turn cleanly, handling any pending permission request first. /// /// Steps: - /// 1. If there is a pending `session/request_permission` that hasn't been - /// responded to yet, respond with `outcome: "cancelled"`. + /// 1. Answer every `session/request_permission` still owed a response, + /// parked ones included, with `outcome: "cancelled"`. /// 2. Send `session/cancel` notification (no id). /// 3. Continue reading until the `session/prompt` response arrives with `stopReason: "cancelled"`. /// @@ -1125,6 +1174,12 @@ impl AcpClient { let prompt_id = self.last_prompt_id.take().ok_or_else(|| { AcpError::Protocol("cancel_with_cleanup called with no in-flight prompt".into()) })?; + // The turn this cancels is over from here. When `pool.rs` drops the + // prompt future, the reset at the end of the prompt call never runs, so + // without this the drain below would decide what the agent asks while + // it winds down with the dead turn's authority: after an owner's turn, + // it would approve it. + self.current_turn_authority = TurnAuthority::Guest; // Step 1: respond to any pending permission request with "cancelled", // but only if we haven't already responded (guards against double-response race). @@ -1136,15 +1191,12 @@ impl AcpClient { // write fails here the pipe is almost certainly dead, but the remaining // ids stay recorded rather than being silently discarded by an early // return. - while let Some(perm_id) = self.pending_permissions.next_owed().cloned() { - let response = permission_response_cancelled(&perm_id); - self.write_ndjson(&response).await?; - self.pending_permissions.forget(&perm_id); - tracing::debug!( - target: "acp::cancel", - "responded cancelled to pending permission id={perm_id}" - ); - } + // + // Parked requests are answered here too: this is the only cleanup + // that runs after `pool.rs` drops a prompt future mid-wait. Their + // verdict receivers are retired before the first write, so the main + // loop's `send` already fails even if a write below does too. + self.answer_owed_cancelled().await?; // Step 2: send session/cancel notification (no id) self.session_cancel(session_id).await?; @@ -1157,6 +1209,9 @@ impl AcpClient { let remaining = hard_deadline .checked_duration_since(tokio::time::Instant::now()) .unwrap_or_default(); + // The ungated loop: a cancel drain is not a turn and has nobody to + // ask, so a request raised while the agent winds down is answered at + // once and can never page a person. let result = self .read_until_response_with_idle_timeout( session_id, @@ -1290,16 +1345,32 @@ impl AcpClient { /// /// While waiting, handles: /// - `session/update` notifications → logged via tracing - /// - `session/request_permission` requests → auto-approved with `allow_once` + /// - `session/request_permission` requests → decided as a guest and never + /// parked: no turn is running, so there is nobody to ask /// - Any other messages → debug-logged and ignored; if they carry an `id` /// (i.e. they are requests, not notifications), a JSON-RPC -32601 error is sent. /// /// Compares the incoming `id` field as a `serde_json::Value` against /// `json!(expected_id)` so that both numeric and string IDs work correctly. + /// + /// When the loop fails, every request it leaves owed is answered before + /// returning. Its `?` exits (a request with no usable option or no options + /// at all, a failed `-32601` write) would otherwise strand the id: nothing + /// else answers it until the next prompt's loop exits or a cancel runs, + /// and the agent may be blocked on it until then. async fn read_until_response( &mut self, expected_id: u64, ) -> Result { + let result = self.read_setup_loop(expected_id).await; + if result.is_err() { + self.answer_owed_on_exit(None).await; + } + result + } + + /// The body of [`read_until_response`](Self::read_until_response). + async fn read_setup_loop(&mut self, expected_id: u64) -> Result { loop { // LinesCodec::new_with_max_length enforces MAX_LINE_SIZE at the // read level — the buffer never grows beyond the limit, preventing @@ -1368,7 +1439,9 @@ impl AcpClient { "session/request_permission" => { // No turn is running here (session/new, set_config_option), so decide as // a guest -- never with whatever authority the last turn left behind. - self.handle_permission_request(&msg, TurnAuthority::Guest) + // And with no `ParkContext`: an RPC with no turn behind it has + // nobody to ask, so it must never wait on one. + self.handle_permission_request(&msg, TurnAuthority::Guest, None) .await?; } other => { @@ -1410,8 +1483,9 @@ impl AcpClient { /// from the original turn rather than starting a fresh timer. /// /// While reading, the loop interleaves goose-native non-cancelling steer - /// requests via `tokio::select!`. The select uses `biased` for - /// reader-first throughput, with a pre-select deadline check at the top + /// requests via `tokio::select!`. The select uses `biased`: a parked + /// request's verdict first, so a busy agent cannot starve an approval, + /// then the reader for throughput, with a pre-select deadline check at the top /// of every loop iteration so a continuously-ready reader arm cannot /// starve the hard deadline (Max's review gate). The steer arm is /// guarded by `pending_steer.is_none()` so at most one steer is in @@ -1423,6 +1497,9 @@ impl AcpClient { /// write time without needing access to outer state. See /// [`crate::pool::SteerRequest`] for why params are built here and not /// in the main loop. + /// + /// Cannot park: it passes no approval link, so a request that needs an + /// approver is refused at once. The cancel re-entry goes through here. async fn read_until_response_with_idle_timeout( &mut self, session_id: &str, @@ -1430,6 +1507,83 @@ impl AcpClient { idle_timeout: std::time::Duration, hard_deadline: tokio::time::Instant, max_duration: std::time::Duration, + ) -> Result { + self.read_until_response_gated( + session_id, + expected_id, + idle_timeout, + hard_deadline, + max_duration, + None, + ) + .await + } + + /// The prompt read loop, able to park requests when given a link, and + /// one drain that covers every way out of it. + /// + /// The drain wraps the loop rather than sitting at each `return` because + /// the loop has many exits, `?` among them, and a request left owed on any + /// one of them blocks the agent. The only exit it cannot cover is the + /// future being dropped mid-poll. Dropped before the prompt's response + /// arrived, `cancel_with_cleanup` answers what it left from the + /// connection's registry; dropped after, in this drain, the next prompt + /// does. + async fn read_until_response_gated( + &mut self, + session_id: &str, + expected_id: u64, + idle_timeout: std::time::Duration, + hard_deadline: tokio::time::Instant, + max_duration: std::time::Duration, + approvals: Option, + ) -> Result { + let mut gate = approvals.map(ParkGate::new); + let result = self + .read_loop( + session_id, + expected_id, + idle_timeout, + hard_deadline, + max_duration, + &mut gate, + ) + .await; + // The prompt response has been consumed, so the prompt is no longer in + // flight, even though the drain below may still await writes. Say so + // before draining: if the pool drops this future mid-drain, it must not + // run a cancel that waits out its grace for a response that already + // arrived, and then re-run a turn that completed. + // + // A JSON-RPC error answering the prompt consumes that response as + // surely as a result does, and nothing else in the loop returns + // `AgentError`. Every other error (a timeout, a broken or oversized + // stream) leaves the response unread: the prompt stays in flight, and + // the pool's cancel drains it. + // + // Cleared here, a drop mid-drain takes the pool's "turn completed + // before the control signal" branch, which runs no cleanup. Whatever + // the drain had not answered stays owed, and the next prompt answers + // it before it is sent. + if matches!(result, Ok(_) | Err(AcpError::AgentError { .. })) + && self.last_prompt_id == Some(expected_id) + { + self.last_prompt_id = None; + } + self.answer_owed_on_exit(gate.as_ref()).await; + result + } + + /// The body of the prompt read loop. `gate` is `Some` only for a turn + /// that may park requests. + async fn read_loop( + &mut self, + session_id: &str, + expected_id: u64, + idle_timeout: std::time::Duration, + hard_deadline: tokio::time::Instant, + max_duration: std::time::Duration, + gate: &mut Option, ) -> Result { use tokio::time::Instant; @@ -1452,20 +1606,41 @@ impl AcpClient { tokio::sync::oneshot::Sender, )> = None; - let now = Instant::now(); - let mut idle_deadline = now + idle_timeout; - let mut hard_deadline = hard_deadline; - let mut last_activity_at = now; + let credit_cap = gate + .as_ref() + .map_or(std::time::Duration::ZERO, |gate| gate.limits().credit_cap); + let mut clock = TurnClock::new(Instant::now(), idle_timeout, hard_deadline, credit_cap); + let entry_hard = hard_deadline; + + // Nothing should be parked on entry: every exit drains, and a dropped + // future is cleaned up by cancel_with_cleanup. A cancel that failed + // before answering could leave one behind, though, and a stale parked + // state would suspend idle for a wait nobody is having. Retire it; the + // id stays owed, and this loop's exit drain answers it. + let stale = self.pending_permissions.retire_all_parked(); + if !stale.is_empty() { + tracing::error!( + target: "acp::permission", + count = stale.len(), + "parked permission requests outlived their turn; retired unanswered" + ); + } loop { // Determine which deadline fires first BEFORE sleeping — this is // the classification we'll use on timeout, immune to scheduler jitter. - let idle_fires_first = idle_deadline < hard_deadline; - let next_deadline = if idle_fires_first { - idle_deadline - } else { - hard_deadline - }; + let now = Instant::now(); + clock.sync(now, self.pending_permissions.parked_count() > 0); + // Publish a credited or renewed hard deadline, so a cancel after an + // idle timeout inherits it. Only when it has moved from the entry + // value, so a loop whose deadline never moves (the cancel re-entry + // above all) leaves the field exactly as it found it. + let effective_hard = clock.effective_hard(now); + if effective_hard != entry_hard { + self.current_hard_deadline = Some(effective_hard); + } + let (next_deadline, fire) = + clock.next_deadline(now, self.pending_permissions.earliest_expiry()); // Pre-select deadline check — required by Max's review. Under // `biased`, a continuously-ready reader arm wins every poll and @@ -1474,28 +1649,47 @@ impl AcpClient { // producing output (see `acp.rs:608` for why the hard deadline // exists). Check the classified deadline here so a steady- // stream agent is still bounded. - if Instant::now() >= next_deadline { - if let Some((_, _, ack_tx)) = pending_steer.take() { - // Prompt is timing out — release the withheld event via - // PromptCompletedNeutral (no fallback signal: there is - // no in-flight turn to signal once we return, and - // normal dispatch handles redelivery). - let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral); - } - if idle_fires_first { - tracing::warn!("idle timeout ({idle_timeout:?}) — no agent activity"); - return Err(AcpError::IdleTimeout(idle_timeout)); - } else { - let silence = Instant::now().saturating_duration_since(last_activity_at); - tracing::warn!("hard turn timeout exceeded (silence {silence:?})"); - return Err(AcpError::HardTimeout { silence }); + if now >= next_deadline { + match fire { + // An expired ask is refused and the turn carries on. Settled + // only here, so expiries have exactly one place to happen. + Fire::Approval => { + self.expire_due_approvals(now, gate).await?; + continue; + } + Fire::Idle | Fire::Hard => { + if let Some((_, _, ack_tx)) = pending_steer.take() { + // Prompt is timing out — release the withheld event via + // PromptCompletedNeutral (no fallback signal: there is + // no in-flight turn to signal once we return, and + // normal dispatch handles redelivery). + let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral); + } + if fire == Fire::Idle { + tracing::warn!("idle timeout ({idle_timeout:?}) — no agent activity"); + return Err(AcpError::IdleTimeout(idle_timeout)); + } + let silence = clock.silence(now); + tracing::warn!("hard turn timeout exceeded (silence {silence:?})"); + return Err(AcpError::HardTimeout { silence }); + } } } + let parked_any = self.pending_permissions.parked_count() > 0; + // LinesCodec::new_with_max_length enforces MAX_LINE_SIZE at the // read level — the buffer never grows beyond the limit. let read_result = tokio::select! { biased; + // Ahead of the reader, so an agent that streams output without + // pause cannot keep an approver's verdict from being applied. + // Disabled while nothing is parked; `next_verdict` is lazy, so + // building it for a disabled branch touches no receiver. + (park_id, verdict) = self.pending_permissions.next_verdict(), if parked_any => { + self.settle_parked(park_id, SettleSource::Arm(verdict), gate).await?; + None + } read_result = self.reader.next() => Some(read_result), // Steer arm: gated off whenever a steer write is already in // flight so we don't stack two writes against the same @@ -1613,16 +1807,23 @@ impl AcpClient { // would catch this anyway, but firing the deadline arm // here makes the wakeup immediate (no extra reader poll // round-trip when stdout is idle). - if let Some((_, _, ack_tx)) = pending_steer.take() { - let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral); - } - if idle_fires_first { - tracing::warn!("idle timeout ({idle_timeout:?}) — no agent activity"); - return Err(AcpError::IdleTimeout(idle_timeout)); - } else { - let silence = Instant::now().saturating_duration_since(last_activity_at); - tracing::warn!("hard turn timeout exceeded (silence {silence:?})"); - return Err(AcpError::HardTimeout { silence }); + // + // An approval expiry is left to that check, so expiries + // are settled in one place only. + match fire { + Fire::Approval => None, + Fire::Idle | Fire::Hard => { + if let Some((_, _, ack_tx)) = pending_steer.take() { + let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral); + } + if fire == Fire::Idle { + tracing::warn!("idle timeout ({idle_timeout:?}) — no agent activity"); + return Err(AcpError::IdleTimeout(idle_timeout)); + } + let silence = clock.silence(Instant::now()); + tracing::warn!("hard turn timeout exceeded (silence {silence:?})"); + return Err(AcpError::HardTimeout { silence }); + } } } }; @@ -1683,9 +1884,7 @@ impl AcpClient { }; self.observe("acp_read", msg.clone()); - let activity_now = Instant::now(); - idle_deadline = activity_now + idle_timeout; - last_activity_at = activity_now; + clock.on_activity(Instant::now()); // Steer response routing must come BEFORE the prompt // response check: a steer response is a regular @@ -1757,10 +1956,11 @@ impl AcpClient { } Some(_) => { let renew_now = Instant::now(); - let new_deadline = renew_now + max_duration; - if new_deadline > hard_deadline { - hard_deadline = new_deadline; - self.current_hard_deadline = Some(new_deadline); + if clock + .renew_hard(renew_now, renew_now + max_duration) + { + self.current_hard_deadline = + Some(clock.effective_hard(renew_now)); tracing::info!( "steer success: renewed hard deadline ({max_duration:?} from now)" ); @@ -1820,9 +2020,7 @@ impl AcpClient { match method { "session/update" => { if self.handle_session_update(&msg) { - let activity_now = Instant::now(); - idle_deadline = activity_now + idle_timeout; - last_activity_at = activity_now; + clock.on_activity(Instant::now()); tracing::debug!("idle clock reset: tool call started"); } } @@ -1831,7 +2029,13 @@ impl AcpClient { } "session/request_permission" => { let authority = self.current_turn_authority; - self.handle_permission_request(&msg, authority).await?; + let park = gate.as_mut().map(|gate| ParkContext { + gate, + clock: &clock, + session_id, + }); + self.handle_permission_request(&msg, authority, park) + .await?; } other => { // If the unknown message has an id, it's a request expecting a reply. @@ -2047,15 +2251,6 @@ impl AcpClient { } } - /// Auto-approve a `session/request_permission` request from the agent. - /// - /// Finds the option with `kind == "allow_once"` and responds with its `optionId`. - /// If no `allow_once` option exists, falls back to `reject_once`. - /// - /// **Critical:** Never hardcode `optionId` — always find it dynamically by `kind`. - /// - /// The request `id` is stored as `serde_json::Value` to support both numeric - /// and string IDs per JSON-RPC 2.0. /// Take the refusals recorded during the turn that just finished. /// /// Draining rather than borrowing, so the caller cannot post the same @@ -2065,14 +2260,27 @@ impl AcpClient { std::mem::take(&mut self.permission_refusals) } + /// Answer, refuse, or park a `session/request_permission` from the agent. + /// + /// Decides first, then acts. [`decide_permission`] says what the turn's + /// authority allows, and only a request that needs an approver can park. + /// Option ids are always the agent's own, found by `kind`, and the request + /// `id` stays a `serde_json::Value` because JSON-RPC 2.0 allows both + /// numeric and string ids. + /// /// `authority` is passed in, not read off `self`. The setup loop decides as /// `Guest` because no turn is running, and the prompt loop passes the turn's /// current authority. Reading the field here once let the setup loop decide /// with whatever authority the previous turn happened to leave behind. + /// + /// `park` is the turn's capability to wait on a person, and only the gated + /// prompt loop builds one. Without it, a request that needs an approver is + /// refused exactly as a guest's request always was. async fn handle_permission_request( &mut self, msg: &serde_json::Value, authority: TurnAuthority, + park: Option>, ) -> Result<(), AcpError> { // Extract id as a Value — JSON-RPC 2.0 allows both numeric and string IDs. let id = msg @@ -2080,6 +2288,18 @@ impl AcpClient { .cloned() .ok_or_else(|| AcpError::Protocol("permission request missing id".into()))?; + // An id that is still owed is a retransmit. Answering it would send + // two responses to one id, and parking it would ask twice about one + // request; the answer already owed covers both copies. + if self.pending_permissions.contains(&id) { + tracing::warn!( + target: "acp::permission", + id = %id, + "ignoring a repeated permission request that is still owed an answer" + ); + return Ok(()); + } + // Record the id so cancel_with_cleanup can answer it. Pushed rather // than assigned: a concurrent sibling request must not evict this one. self.pending_permissions.record(id.clone()); @@ -2094,45 +2314,195 @@ impl AcpClient { options.len() ); - let decision = decide_permission(authority, options); + let tool = tool_label(&msg["params"]["toolCall"]); - let response = match &decision { + match decide_permission(authority, options) { PermissionDecision::Approve(option_id) => { tracing::info!( target: "acp::permission", "auto-approving permission id={id} with allow_once optionId={option_id:?}" ); - permission_response_selected(&id, option_id) + self.answer_permission(&id, permission_response_selected(&id, &option_id)) + .await } PermissionDecision::Refuse(option_id) => { - // warn, not debug: a refusal changes what the agent did, and on - // a guest turn it is the whole point of the gate. At debug this - // is invisible in practice, and "the agent just didn't do it" - // is indistinguishable from a bug. - let tool = msg["params"]["toolCall"]["title"] - .as_str() - .or_else(|| msg["params"]["toolCall"]["kind"].as_str()) - .unwrap_or("(untitled)"); - tracing::warn!( - target: "acp::permission", - id = %id, - tool = %tool, - authority = ?authority, - "refusing permission request" - ); - self.permission_refusals.push(PermissionRefusal { - tool: tool.to_string(), - }); - permission_response_selected(&id, option_id) + self.refuse_permission( + &id, + &option_id, + tool, + RefusalReason::NoAllowOffered, + authority, + ) + .await } - PermissionDecision::NoUsableOption => { - return Err(AcpError::Protocol( - "no suitable permission option found (neither allow_once nor reject_once)" - .into(), - )); + PermissionDecision::NoUsableOption => Err(AcpError::Protocol( + "no suitable permission option found (neither allow_once nor reject_once)".into(), + )), + PermissionDecision::NeedsApprover { allow, reject } => { + let reason = match park { + // Parked: nothing is written now. The id stays owed until a + // verdict, an expiry, or the loop's exit answers it. + Some(ctx) => match self.park_permission(&id, msg, &tool, allow, &reject, ctx) { + Ok(()) => return Ok(()), + Err(reason) => reason, + }, + None => RefusalReason::NoApprovalPath, + }; + self.refuse_permission(&id, &reject, tool, reason, authority) + .await } + } + } + + /// Park a request that needs an approver, and hand it to the main loop. + /// + /// Synchronous on purpose: parking writes nothing to the agent and the + /// hand-off is an unbounded send, so no await can fall between recording + /// the parked state and the main loop hearing about it. + /// + /// `Err` is the reason to refuse instead; the caller writes the refusal. + fn park_permission( + &mut self, + id: &serde_json::Value, + msg: &serde_json::Value, + tool: &str, + allow: String, + reject: &str, + ctx: ParkContext<'_>, + ) -> Result<(), RefusalReason> { + let now = tokio::time::Instant::now(); + let limits = *ctx.gate.limits(); + // A request that names no session is taken to be for this one. One + // that names another is refused, never parked against this turn. + let session_matches = msg["params"]["sessionId"] + .as_str() + .is_none_or(|sid| sid == ctx.session_id); + let expires_at = match admit( + self.pending_permissions.parked_count(), + ctx.gate.asks_charged(), + session_matches, + ctx.clock.ask_expiry(now, &limits), + &limits, + ) { + Admission::Park { expires_at } => expires_at, + Admission::Refuse(reason) => return Err(reason), }; + let tool_call = &msg["params"]["toolCall"]; + let digest = request_digest(ctx.session_id, tool_call); + let (verdict_tx, verdict_rx) = tokio::sync::oneshot::channel(); + let park_id = ParkId::mint(); + // Parked before the ask goes out, so a verdict that comes straight + // back always finds its receiver in place. + let parked = self.pending_permissions.park( + id, + ParkedAsk { + park_id, + allow_option_id: allow, + reject_option_id: reject.to_string(), + digest, + tool: tool.to_string(), + parked_at: now, + expires_at, + verdict_rx, + }, + ); + if let Err(rejected) = parked { + // The id was recorded unparked just before this call, so this + // cannot happen. Refused rather than trusted if it ever does. + tracing::error!( + target: "acp::permission", + id = %id, + ?rejected, + "could not park a permission request" + ); + return Err(RefusalReason::NoApprovalPath); + } + let sent = ctx.gate.send(ApprovalEvent::Ask(ApprovalAsk { + park_id, + turn_id: ctx.gate.turn_id().to_string(), + session_id: ctx.session_id.to_string(), + digest, + tool_call: tool_call.clone(), + parked_at: now, + expires_at, + verdict_tx, + })); + if !sent { + // The main loop is gone, so nobody will ever see this ask. Refuse + // now instead of holding the agent until the ask expires. + let _ = self.pending_permissions.take_parked(park_id); + tracing::warn!( + target: "acp::permission", + id = %id, + tool = %tool, + "no approval path: the main loop is not listening" + ); + return Err(RefusalReason::NoApprovalPath); + } + ctx.gate.charge(); + + let expires_in = expires_at.saturating_duration_since(now); + tracing::info!( + target: "acp::permission", + id = %id, + park_id = park_id.0, + turn_id = %ctx.gate.turn_id(), + tool = %tool, + digest = %digest.to_hex(), + "parked permission request for an approver (expires in {expires_in:?})" + ); + self.observe( + "approval_parked", + serde_json::json!({ + "parkId": park_id.0, + "turnId": ctx.gate.turn_id(), + "digest": digest.to_hex(), + "tool": tool, + "kind": tool_call.get("kind"), + "expiresInMs": u64::try_from(expires_in.as_millis()).unwrap_or(u64::MAX), + }), + ); + Ok(()) + } + + /// Refuse a request with the agent's own `reject_once`, recording why so + /// the person who asked can be told. + async fn refuse_permission( + &mut self, + id: &serde_json::Value, + reject_option_id: &str, + tool: String, + reason: RefusalReason, + authority: TurnAuthority, + ) -> Result<(), AcpError> { + // warn, not debug: a refusal changes what the agent did, and on + // a guest turn it is the whole point of the gate. At debug this + // is invisible in practice, and "the agent just didn't do it" + // is indistinguishable from a bug. + tracing::warn!( + target: "acp::permission", + id = %id, + tool = %tool, + authority = ?authority, + reason = ?reason, + "refusing permission request" + ); + self.permission_refusals + .push(PermissionRefusal { tool, reason }); + self.answer_permission(id, permission_response_selected(id, reject_option_id)) + .await + } + + /// Write one answer to a permission request, then forget its id. + /// + /// Every answer goes through here, parked or not, so the order below holds + /// on every path. + async fn answer_permission( + &mut self, + id: &serde_json::Value, + response: serde_json::Value, + ) -> Result<(), AcpError> { // Write the response first, then forget the id. // // Order is load-bearing and was a deadlock once. Clearing before the @@ -2146,10 +2516,240 @@ impl AcpClient { // Only THIS id is removed. A sibling request still in flight keeps its // own entry and its own claim on an answer. self.write_ndjson(&response).await?; - self.pending_permissions.forget(&id); + self.pending_permissions.forget(id); + Ok(()) + } + + /// Answer one parked request, from its verdict or from its expiry. + async fn settle_parked( + &mut self, + park_id: ParkId, + source: SettleSource, + gate: &mut Option, + ) -> Result<(), AcpError> { + use tokio::sync::oneshot::error::TryRecvError; + + // First, and before any await: the verdict arm has just seen this + // receiver complete, and a tokio oneshot receiver panics if it is + // polled again. Out of the registry is out of the arm. + let Some((rpc_id, mut ask)) = self.pending_permissions.take_parked(park_id) else { + return Ok(()); + }; + let outcome = match source { + SettleSource::Arm(Ok(verdict)) => VerdictOutcome::Delivered(verdict), + SettleSource::Arm(Err(_)) => VerdictOutcome::Closed, + // A verdict that landed by the expiry instant still counts: the + // approver answered in time, and the loop only just noticed. + SettleSource::Expiry => match ask.verdict_rx.try_recv() { + Ok(verdict) => VerdictOutcome::Delivered(verdict), + Err(TryRecvError::Closed) => VerdictOutcome::Closed, + Err(TryRecvError::Empty) => { + // Close, then look once more. A verdict sent in between + // is honoured, and any send after the close fails, so the + // main loop is never told a verdict landed on a request + // that is about to be refused as expired. + ask.verdict_rx.close(); + match ask.verdict_rx.try_recv() { + Ok(verdict) => VerdictOutcome::Delivered(verdict), + Err(_) => VerdictOutcome::Expired, + } + } + }, + }; + // Gone before the write below can await. Alive during that await, with + // no loop polling it, it would let the main loop's `send` succeed for a + // request this is about to answer another way. + drop(ask.verdict_rx); + + let resolution = resolve(&ask.digest, outcome); + let option_id = if resolution.allow { + &ask.allow_option_id + } else { + &ask.reject_option_id + }; + self.answer_permission(&rpc_id, permission_response_selected(&rpc_id, option_id)) + .await?; + if let Some(reason) = resolution.refusal { + self.permission_refusals.push(PermissionRefusal { + tool: ask.tool.clone(), + reason, + }); + } + let waited = tokio::time::Instant::now().saturating_duration_since(ask.parked_at); + if let Some(gate) = gate.as_mut() { + if resolution.refund_ask { + gate.refund(); + } + // After the write, never before: `Allowed` must mean the agent was + // told. If this send fails the main loop is gone, and there is + // nobody left to tell. + gate.send(ApprovalEvent::Settled(ApprovalSettled { + park_id, + turn_id: gate.turn_id().to_string(), + settlement: resolution.settlement, + waited, + })); + } + self.report_settled( + &rpc_id, + park_id, + gate.as_ref().map(ParkGate::turn_id), + &ask.tool, + resolution.settlement, + waited, + ); + Ok(()) + } + + /// Refuse every parked request whose expiry has passed. The turn carries on. + async fn expire_due_approvals( + &mut self, + now: tokio::time::Instant, + gate: &mut Option, + ) -> Result<(), AcpError> { + while let Some(park_id) = self.pending_permissions.first_expired(now) { + self.settle_parked(park_id, SettleSource::Expiry, gate) + .await?; + } + Ok(()) + } + + /// Answer every request a read loop is leaving owed, whichever way it left. + /// + /// Parked requests get the agent's own `reject_once`, and the main loop a + /// `Settled`. Anything else still owed (a failed write, a request with no + /// usable option) gets `cancelled`. Stops at the first failed write: the + /// pipe is almost certainly dead, and whatever is left stays owed for + /// `cancel_with_cleanup`. + async fn answer_owed_on_exit(&mut self, gate: Option<&ParkGate>) { + // Retire every parked request before writing to any of them. Each write + // can await for up to its timeout, and no receiver may outlive the loop + // that was polling it while that happens. + for ask in self.pending_permissions.retire_all_parked() { + let resolution = resolve(&ask.digest, VerdictOutcome::TurnEnded); + let response = permission_response_selected(&ask.rpc_id, &ask.reject_option_id); + if let Err(e) = self.answer_permission(&ask.rpc_id, response).await { + tracing::warn!( + target: "acp::permission", + "exit drain stopped at a failed write ({e}); the rest stay owed" + ); + return; + } + if let Some(reason) = resolution.refusal { + self.permission_refusals.push(PermissionRefusal { + tool: ask.tool.clone(), + reason, + }); + } + let waited = tokio::time::Instant::now().saturating_duration_since(ask.parked_at); + if let Some(gate) = gate { + gate.send(ApprovalEvent::Settled(ApprovalSettled { + park_id: ask.park_id, + turn_id: gate.turn_id().to_string(), + settlement: resolution.settlement, + waited, + })); + } + self.report_settled( + &ask.rpc_id, + ask.park_id, + gate.map(ParkGate::turn_id), + &ask.tool, + resolution.settlement, + waited, + ); + } + if let Err(e) = self.answer_owed_cancelled().await { + tracing::warn!( + target: "acp::permission", + "exit drain stopped at a failed write ({e}); the rest stay owed" + ); + } + } + + /// Answer `cancelled` to every request still owed, oldest first. + /// + /// Anything still parked is retired first, which drops its verdict + /// receiver before the first write can await. Stops at, and returns, the + /// first failed write; the ids it did not reach stay owed. + async fn answer_owed_cancelled(&mut self) -> Result<(), AcpError> { + for ask in self.pending_permissions.retire_all_parked() { + tracing::info!( + target: "acp::permission", + id = %ask.rpc_id, + park_id = ask.park_id.0, + tool = %ask.tool, + "answering a parked permission request cancelled" + ); + } + while let Some(id) = self.pending_permissions.next_owed().cloned() { + self.answer_permission(&id, permission_response_cancelled(&id)) + .await?; + tracing::debug!( + target: "acp::permission", + "responded cancelled to pending permission id={id}" + ); + } Ok(()) } + /// Log and observe how a parked request ended. + fn report_settled( + &self, + rpc_id: &serde_json::Value, + park_id: ParkId, + turn_id: Option<&str>, + tool: &str, + settlement: Settlement, + waited: std::time::Duration, + ) { + let name = settlement.as_str(); + match settlement { + Settlement::Allowed { .. } | Settlement::Denied(_) | Settlement::TurnEnded => { + tracing::info!( + target: "acp::permission", + id = %rpc_id, + park_id = park_id.0, + tool = %tool, + ?waited, + "parked permission request settled: {name}" + ); + } + Settlement::Expired | Settlement::VerdictChannelClosed => { + tracing::warn!( + target: "acp::permission", + id = %rpc_id, + park_id = park_id.0, + tool = %tool, + ?waited, + "parked permission request refused unanswered: {name}" + ); + } + // The main loop answered one request with another's verdict. It was + // refused, so nothing ran, but the binding between asks and + // verdicts is broken somewhere. + Settlement::DigestMismatch => { + tracing::error!( + target: "acp::permission", + id = %rpc_id, + park_id = park_id.0, + tool = %tool, + "verdict digest did not match the parked request; refused" + ); + } + } + let mut payload = serde_json::json!({ + "parkId": park_id.0, + "turnId": turn_id, + "settlement": name, + "waitedMs": u64::try_from(waited.as_millis()).unwrap_or(u64::MAX), + }); + if let Settlement::Allowed { via } = settlement { + payload["via"] = serde_json::json!(via.as_str()); + } + self.observe("approval_settled", payload); + } + /// Parse a completed prompt response and retain its optional per-turn usage. fn parse_prompt_response( &mut self, @@ -2254,6 +2854,30 @@ pub(crate) enum PermissionDecision { Refuse(String), /// The agent offered nothing usable for the decision we reached. NoUsableOption, + /// Only someone with authority can allow this. `allow` is written if they + /// do; `reject` if they do not, or if nobody can be asked. + NeedsApprover { allow: String, reject: String }, +} + +/// What a parked request is being settled from. +enum SettleSource { + /// The verdict arm saw its channel complete: a verdict, or a dropped + /// sender. + Arm(Result), + /// Its expiry passed. + Expiry, +} + +/// What the prompt loop lends the permission handler so a request can park. +/// +/// Built per request from the loop's own locals and gone when the handler +/// returns, so the capability can never outlive the turn that holds it. +struct ParkContext<'a> { + gate: &'a mut ParkGate, + clock: &'a TurnClock, + /// The read loop's own session id: what the digest binds, and what a + /// request's `sessionId` must match. + session_id: &'a str, } /// Decide a permission request from the turn's authority and the offered options. @@ -2281,12 +2905,17 @@ pub(crate) fn decide_permission( // approval. The agent holds an unrestricted shell, so the tool name // says nothing useful about reach; who is asking is the axis that can // actually be enforced. - TurnAuthority::Guest => match by_kind("reject_once") { - Some(id) => PermissionDecision::Refuse(id), + TurnAuthority::Guest => match (by_kind("allow_once"), by_kind("reject_once")) { + // Never approved here. Only someone with authority can allow a + // guest's request, so it waits for one if the turn can reach one, + // and is refused if it cannot. + (Some(allow), Some(reject)) => PermissionDecision::NeedsApprover { allow, reject }, + // Nothing to approve, so nobody to ask. + (None, Some(reject)) => PermissionDecision::Refuse(reject), // Falling back to approval here would invert the rule, so // report that nothing usable was offered and let the caller fail // the turn. - None => PermissionDecision::NoUsableOption, + (_, None) => PermissionDecision::NoUsableOption, }, TurnAuthority::Owner => match by_kind("allow_once") { Some(id) => PermissionDecision::Approve(id), @@ -2735,21 +3364,37 @@ mod tests { } #[test] - fn a_guest_turn_is_refused_even_when_approval_is_offered() { + fn a_guest_turn_needs_an_approver_when_both_options_are_offered() { // The regression this exists to prevent: before the gate, every // request was auto-approved in every mode, so anyone on the allowlist - // could drive tool calls on the owner's machine with nobody asked. + // could drive tool calls on the owner's machine with nobody asked. An + // offered allow_once means someone may be asked, never that the + // guest's turn may take it. Both ids are the agent's own, by kind. let options = vec![ opt("allow_once", "allow-42"), opt("reject_once", "reject-7"), ]; assert_eq!( decide_permission(TurnAuthority::Guest, &options), - PermissionDecision::Refuse("reject-7".to_string()), + PermissionDecision::NeedsApprover { + allow: "allow-42".to_string(), + reject: "reject-7".to_string(), + }, "an available allow_once must not tempt a guest turn into approval" ); } + #[test] + fn a_guest_turn_offering_only_reject_is_refused_without_asking() { + // With no allow_once there is nothing an approver could say yes to, + // so paging one would only waste their time. + let options = vec![opt("reject_once", "reject-7")]; + assert_eq!( + decide_permission(TurnAuthority::Guest, &options), + PermissionDecision::Refuse("reject-7".to_string()) + ); + } + #[test] fn a_guest_turn_with_no_reject_option_reports_no_usable_option() { // Falling back to the offered allow_once would invert the rule @@ -2781,50 +3426,1139 @@ mod tests { } } - use super::*; - - #[test] - fn stop_reason_parses_all_known_values() { - assert_eq!(StopReason::from_str("end_turn"), Some(StopReason::EndTurn)); - assert_eq!( - StopReason::from_str("cancelled"), - Some(StopReason::Cancelled) - ); - assert_eq!( - StopReason::from_str("max_tokens"), - Some(StopReason::MaxTokens) - ); - assert_eq!( - StopReason::from_str("max_turn_requests"), - Some(StopReason::MaxTurnRequests) - ); - assert_eq!(StopReason::from_str("refusal"), Some(StopReason::Refusal)); + // ── Parking a guest's request ──────────────────────────────────────────── + // + // `cat` stands in for the agent in these: every line the harness writes + // comes straight back on the reader, so a test can see exactly what was + // answered without a bash script, and they run on every platform. + + use crate::approval_gate::{AllowVia, ApprovalDecision, GateLimits, RequestDigest}; + + const PARK_SESSION: &str = "sess-test"; + + fn approval_link( + limits: GateLimits, + ) -> ( + ApprovalLink, + tokio::sync::mpsc::UnboundedReceiver, + ) { + let (events_tx, events_rx) = tokio::sync::mpsc::unbounded_channel(); + let link = ApprovalLink { + events_tx, + turn_id: "turn-1".into(), + limits, + }; + (link, events_rx) } - #[test] - fn stop_reason_returns_none_for_unknown() { - assert_eq!(StopReason::from_str("unknown_value"), None); - assert_eq!(StopReason::from_str(""), None); - assert_eq!(StopReason::from_str("endturn"), None); // no camelCase — still unknown + fn probe_request_with_id(id: u64) -> serde_json::Value { + let mut request: serde_json::Value = + serde_json::from_str(PROBE_PERMISSION_REQUEST).expect("probe is JSON"); + request["id"] = serde_json::json!(id); + request } - #[test] - fn stop_reason_is_case_insensitive() { - // Agents may send uppercase or mixed-case variants — all should parse correctly. - assert_eq!(StopReason::from_str("END_TURN"), Some(StopReason::EndTurn)); - assert_eq!( - StopReason::from_str("CANCELLED"), - Some(StopReason::Cancelled) - ); - assert_eq!( - StopReason::from_str("Max_Tokens"), - Some(StopReason::MaxTokens) - ); - assert_eq!( - StopReason::from_str("MAX_TURN_REQUESTS"), - Some(StopReason::MaxTurnRequests) - ); - assert_eq!(StopReason::from_str("Refusal"), Some(StopReason::Refusal)); + fn probe_tool_call() -> serde_json::Value { + probe_request_with_id(7)["params"]["toolCall"].clone() + } + + /// A clock with plenty of turn left, so a fresh ask gets its full window. + fn roomy_clock(limits: &GateLimits) -> TurnClock { + let now = tokio::time::Instant::now(); + TurnClock::new( + now, + std::time::Duration::from_secs(60), + now + std::time::Duration::from_secs(3600), + limits.credit_cap, + ) + } + + /// The next line the harness wrote, as `cat` echoed it back. + async fn next_written(client: &mut AcpClient) -> serde_json::Value { + let line = tokio::time::timeout(std::time::Duration::from_secs(5), client.reader.next()) + .await + .expect("the harness should have written an answer") + .expect("cat closed its stdout") + .expect("cat's line was readable"); + serde_json::from_str(&line).expect("the harness writes JSON") + } + + /// Whether the harness wrote nothing within a short window. + async fn nothing_written(client: &mut AcpClient) -> bool { + tokio::time::timeout(std::time::Duration::from_millis(300), client.reader.next()) + .await + .is_err() + } + + /// Park the probe request as the handler would, and hand back what the + /// main loop would hold. + fn park_directly( + client: &mut AcpClient, + id: u64, + expires_in: std::time::Duration, + ) -> (tokio::sync::oneshot::Sender, RequestDigest) { + let rpc_id = serde_json::json!(id); + client.pending_permissions.record(rpc_id.clone()); + let (verdict_tx, verdict_rx) = tokio::sync::oneshot::channel(); + let now = tokio::time::Instant::now(); + let digest = request_digest(PARK_SESSION, &probe_tool_call()); + client + .pending_permissions + .park( + &rpc_id, + ParkedAsk { + park_id: ParkId::mint(), + allow_option_id: "allow-1".into(), + reject_option_id: "reject-1".into(), + digest, + tool: "Bash".into(), + parked_at: now, + expires_at: now + expires_in, + verdict_rx, + }, + ) + .expect("recorded just above"); + (verdict_tx, digest) + } + + fn refusal(reason: RefusalReason) -> PermissionRefusal { + PermissionRefusal { + tool: "Bash".into(), + reason, + } + } + + #[tokio::test] + async fn a_guest_request_parks_without_answering_and_asks_once() { + let mut client = spawn_inert_client().await; + let limits = GateLimits::default(); + let (link, mut events) = approval_link(limits); + let mut gate = ParkGate::new(link); + let clock = roomy_clock(&limits); + let request = probe_request_with_id(7); + + client + .handle_permission_request( + &request, + TurnAuthority::Guest, + Some(ParkContext { + gate: &mut gate, + clock: &clock, + session_id: PARK_SESSION, + }), + ) + .await + .expect("parking writes nothing, so it cannot fail"); + + assert!( + nothing_written(&mut client).await, + "a parked request is not answered yet" + ); + assert_eq!( + client.pending_permissions.next_owed(), + Some(&serde_json::json!(7)), + "a parked request is still owed, so a cancel would answer it" + ); + assert_eq!(gate.asks_charged(), 1); + let ask = match events.try_recv() { + Ok(ApprovalEvent::Ask(ask)) => ask, + other => panic!("expected one ask, got {other:?}"), + }; + assert_eq!( + ask.digest, + request_digest(PARK_SESSION, &probe_tool_call()), + "the ask is bound to exactly what the agent asked" + ); + assert_eq!(ask.turn_id, "turn-1"); + assert_eq!(ask.session_id, PARK_SESSION); + assert_eq!( + ask.tool_call, + probe_tool_call(), + "the renderer gets the tool call verbatim" + ); + assert_eq!(ask.expires_at - ask.parked_at, limits.approval_timeout); + assert_eq!( + client.pending_permissions.first_expired(ask.expires_at), + Some(ask.park_id), + "the ask names the request that is parked" + ); + assert!(!ask.verdict_tx.is_closed()); + + // A retransmit of an id still owed is neither answered nor asked twice. + client + .handle_permission_request( + &request, + TurnAuthority::Guest, + Some(ParkContext { + gate: &mut gate, + clock: &clock, + session_id: PARK_SESSION, + }), + ) + .await + .expect("a retransmit is ignored"); + assert!(nothing_written(&mut client).await); + assert!(events.try_recv().is_err(), "one request, one ask"); + assert_eq!(gate.asks_charged(), 1); + assert!(client.take_permission_refusals().is_empty()); + } + + #[tokio::test] + async fn a_guest_request_is_refused_at_once_when_nobody_is_listening() { + let mut client = spawn_inert_client().await; + let limits = GateLimits::default(); + let (link, events) = approval_link(limits); + drop(events); + let mut gate = ParkGate::new(link); + let clock = roomy_clock(&limits); + + client + .handle_permission_request( + &probe_request_with_id(7), + TurnAuthority::Guest, + Some(ParkContext { + gate: &mut gate, + clock: &clock, + session_id: PARK_SESSION, + }), + ) + .await + .expect("the refusal is written"); + + let answer = next_written(&mut client).await; + assert_eq!( + answer["result"]["outcome"]["optionId"], "reject-1", + "an ask nobody can see must not hold the agent until it expires" + ); + assert_eq!(client.pending_permissions.parked_count(), 0); + assert!(client.pending_permissions.next_owed().is_none()); + assert_eq!( + gate.asks_charged(), + 0, + "an ask that never went out costs nothing" + ); + assert_eq!( + client.take_permission_refusals(), + vec![refusal(RefusalReason::NoApprovalPath)] + ); + } + + #[tokio::test] + async fn a_request_for_another_session_is_refused_not_parked() { + let mut client = spawn_inert_client().await; + let limits = GateLimits::default(); + let (link, mut events) = approval_link(limits); + let mut gate = ParkGate::new(link); + let clock = roomy_clock(&limits); + let mut request = probe_request_with_id(7); + request["params"]["sessionId"] = serde_json::json!("another-session"); + + client + .handle_permission_request( + &request, + TurnAuthority::Guest, + Some(ParkContext { + gate: &mut gate, + clock: &clock, + session_id: PARK_SESSION, + }), + ) + .await + .expect("the refusal is written"); + + let answer = next_written(&mut client).await; + assert_eq!(answer["result"]["outcome"]["optionId"], "reject-1"); + assert!(events.try_recv().is_err(), "nobody is asked about it"); + assert_eq!( + client.take_permission_refusals(), + vec![refusal(RefusalReason::SessionMismatch)] + ); + } + + /// The approver answered in time; the loop was merely late to look. + #[tokio::test] + async fn a_verdict_delivered_by_the_expiry_instant_is_honoured() { + let mut client = spawn_inert_client().await; + let (link, mut events) = approval_link(GateLimits::default()); + let mut gate = Some(ParkGate::new(link)); + let window = std::time::Duration::from_secs(600); + let (verdict_tx, digest) = park_directly(&mut client, 7, window); + verdict_tx + .send(ApprovalVerdict { + digest, + decision: ApprovalDecision::Allow { + via: AllowVia::Approver, + }, + }) + .expect("the request is parked"); + + client + .expire_due_approvals(tokio::time::Instant::now() + window, &mut gate) + .await + .expect("the answer is written"); + + let answer = next_written(&mut client).await; + assert_eq!(answer["result"]["outcome"]["optionId"], "allow-1"); + match events.try_recv() { + Ok(ApprovalEvent::Settled(settled)) => assert_eq!( + settled.settlement, + Settlement::Allowed { + via: AllowVia::Approver + } + ), + other => panic!("expected settled, got {other:?}"), + } + assert!(client.take_permission_refusals().is_empty()); + assert!(client.pending_permissions.next_owed().is_none()); + } + + #[tokio::test] + async fn an_unanswered_ask_expires_to_reject_and_closes_its_channel() { + let mut client = spawn_inert_client().await; + let (link, mut events) = approval_link(GateLimits::default()); + let mut gate = Some(ParkGate::new(link)); + let window = std::time::Duration::from_secs(600); + let (verdict_tx, _) = park_directly(&mut client, 7, window); + + client + .expire_due_approvals(tokio::time::Instant::now() + window, &mut gate) + .await + .expect("the refusal is written"); + + let answer = next_written(&mut client).await; + assert_eq!(answer["result"]["outcome"]["optionId"], "reject-1"); + match events.try_recv() { + Ok(ApprovalEvent::Settled(settled)) => { + assert_eq!(settled.settlement, Settlement::Expired) + } + other => panic!("expected settled, got {other:?}"), + } + assert!( + verdict_tx.is_closed(), + "a yes that arrives now must be told it is too late" + ); + assert_eq!( + client.take_permission_refusals(), + vec![refusal(RefusalReason::Expired)] + ); + } + + #[tokio::test] + async fn a_failed_drain_write_stops_and_leaves_ids_owed_for_cancel() { + let mut client = spawn_inert_client().await; + // The agent is gone, so every write fails. + client.shutdown().await; + let mut senders = Vec::new(); + for id in 1..=3 { + let (verdict_tx, _) = + park_directly(&mut client, id, std::time::Duration::from_secs(600)); + senders.push(verdict_tx); + } + + client.answer_owed_on_exit(None).await; + + assert_eq!( + client.pending_permissions.parked_count(), + 0, + "every request is retired even though no write got through" + ); + assert!(senders.iter().all(|tx| tx.is_closed())); + assert_eq!( + client.pending_permissions.next_owed(), + Some(&serde_json::json!(1)), + "the ids stay owed for cancel_with_cleanup" + ); + } + + /// Cancel is the one cleanup after `pool.rs` drops a prompt future + /// mid-wait. It retires parked requests before its first write, so even + /// when that write fails the main loop learns the ask is dead instead of + /// being told, until the ask expires, that its verdicts are delivered. + #[tokio::test] + async fn cancel_retires_parked_receivers_before_a_write_can_fail() { + let mut client = spawn_inert_client().await; + client.shutdown().await; + let (verdict_tx, _) = park_directly(&mut client, 7, std::time::Duration::from_secs(600)); + client.last_prompt_id = Some(1); + + let result = client + .cancel_with_cleanup_grace(PARK_SESSION, std::time::Duration::from_secs(1)) + .await; + + assert!( + result.is_err(), + "the pipe is dead, so the cancel cannot finish" + ); + assert!(verdict_tx.is_closed()); + assert_eq!(client.pending_permissions.parked_count(), 0); + assert_eq!( + client.pending_permissions.next_owed(), + Some(&serde_json::json!(7)), + "nothing reached the agent, so the id is still owed" + ); + } + + /// `cat` echoes the prompt back as a request the harness does not know, + /// the harness answers `-32601`, and `cat` echoes that back as the + /// prompt's error response: a prompt that ends at once, with no script. + #[tokio::test] + async fn a_finished_prompt_leaves_the_connection_untrusted() { + let mut client = spawn_inert_client().await; + let result = client + .session_prompt_with_idle_timeout( + PARK_SESSION, + "hello", + std::time::Duration::from_secs(5), + std::time::Duration::from_secs(10), + TurnAuthority::Owner, + None, + ) + .await; + assert!( + matches!(result, Err(AcpError::AgentError { code: -32601, .. })), + "expected the echoed -32601, got {result:?}" + ); + assert_eq!( + client.current_turn_authority, + TurnAuthority::Guest, + "an owner turn's trust must not outlive the turn" + ); + } + + // ── Around a write that has not finished (cat and sleep, every platform) ── + // + // Some orderings only show while an answer is still on its way to the + // agent: what the harness has already let go of by then, and what it still + // holds. The wire cannot show that, so these tests hold a write open on + // purpose, and look at the harness, or drop it, while it waits. + + /// The prompt's id in tests that answer the prompt by hand. + const PROMPT_ID: u64 = 41; + + fn prompt_response(stop_reason: &str) -> serde_json::Value { + serde_json::json!({ + "jsonrpc": "2.0", + "id": PROMPT_ID, + "result": {"stopReason": stop_reason}, + }) + } + + /// A child whose stdin will not take another byte. + /// + /// `sleep` never reads stdin, so the pipe is filled here until a write + /// stalls, and every write after that waits for as long as the child lives. + /// Swap its `stdin` into a client under test to hold that client's next + /// write open. + /// + /// End the test with `shutdown` on it. On Windows a stuck write occupies a + /// blocking thread until the pipe breaks, and the runtime waits for that + /// thread before the test can finish. + async fn stalled_stdin() -> AcpClient { + use tokio::io::AsyncWriteExt; + let mut stalled = AcpClient::spawn("sleep", &["3600".to_string()], &[], false) + .await + .expect("spawn sleep as an agent that never reads"); + let chunk = vec![b'x'; 64 * 1024]; + // `write`, not `write_all`, so a stalled attempt leaves nothing + // half-written. A whole second without progress means the pipe is + // full, and nothing will ever empty it. + loop { + let attempt = stalled.stdin.write(&chunk); + match tokio::time::timeout(std::time::Duration::from_secs(1), attempt).await { + Ok(Ok(_)) => {} + Ok(Err(e)) => panic!("filling the stalled pipe failed: {e}"), + Err(_) => return stalled, + } + } + } + + /// Expiry closes the ask before its refusal is written, not after. The + /// write can take up to its timeout, and a receiver alive that long, with + /// no loop polling it, would let the main loop's `send` succeed for a + /// request that is being refused. + #[tokio::test] + async fn an_expired_ask_is_closed_while_its_refusal_is_still_being_written() { + let mut client = spawn_inert_client().await; + let (link, _events) = approval_link(GateLimits::default()); + let mut gate = Some(ParkGate::new(link)); + let window = std::time::Duration::from_secs(600); + let (verdict_tx, digest) = park_directly(&mut client, 7, window); + let mut stalled = stalled_stdin().await; + std::mem::swap(&mut client.stdin, &mut stalled.stdin); + + let mut expiry = + Box::pin(client.expire_due_approvals(tokio::time::Instant::now() + window, &mut gate)); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(500), &mut expiry) + .await + .is_err(), + "the refusal should be stuck in its write" + ); + assert!( + verdict_tx.is_closed(), + "the ask must be closed before its refusal is written" + ); + let late = ApprovalVerdict { + digest, + decision: ApprovalDecision::Allow { + via: AllowVia::Approver, + }, + }; + assert!( + verdict_tx.send(late).is_err(), + "a yes sent now must be told it is too late, not taken" + ); + drop(expiry); + stalled.shutdown().await; + } + + /// Run the gated read loop to the prompt's `response` with a guest's + /// request parked, and drop it where `pool.rs`'s `select!` can: in the exit + /// drain, with that request's refusal stuck in its write. + /// + /// `cat` plays the agent, echoing the request and then the response + /// written to it here; the stalled stdin swapped in afterwards holds the + /// drain's first write. Returns the client, and the stalled child for the + /// test to shut down. + async fn drop_in_the_exit_drain(response: serde_json::Value) -> (AcpClient, AcpClient) { + let mut client = spawn_inert_client().await; + for line in [probe_request_with_id(7), response] { + client + .write_ndjson(&line) + .await + .expect("cat takes the line"); + } + let mut stalled = stalled_stdin().await; + std::mem::swap(&mut client.stdin, &mut stalled.stdin); + client.last_prompt_id = Some(PROMPT_ID); + + let (link, mut events) = approval_link(GateLimits::default()); + let max = std::time::Duration::from_secs(3600); + let mut turn = Box::pin(client.read_until_response_gated( + PARK_SESSION, + PROMPT_ID, + std::time::Duration::from_secs(30), + tokio::time::Instant::now() + max, + max, + Some(link), + )); + // The drain retires the parked request, closing its ask, before its + // first write. A closed ask on a turn still pending therefore means the + // turn is in that write. Polled in steps, so a slow `cat` only makes + // this wait longer. + let mut ask = None; + let give_up = tokio::time::Instant::now() + std::time::Duration::from_secs(10); + loop { + if let Ok(result) = + tokio::time::timeout(std::time::Duration::from_millis(50), &mut turn).await + { + panic!("the turn should be stuck in its exit drain, but it ended: {result:?}"); + } + if ask.is_none() { + ask = match events.try_recv() { + Ok(ApprovalEvent::Ask(ask)) => Some(ask), + Ok(other) => panic!("expected the request's ask, got {other:?}"), + Err(_) => None, + }; + } + if ask.as_ref().is_some_and(|ask| ask.verdict_tx.is_closed()) { + break; + } + assert!( + tokio::time::Instant::now() < give_up, + "the turn never reached its exit drain" + ); + } + drop(turn); + (client, stalled) + } + + /// `has_in_flight_prompt` is all the pool asks after dropping a turn. Were + /// it still true once the response had been read, the pool would cancel, + /// wait out its grace for a response that is not coming, and re-run a turn + /// that completed. + #[tokio::test] + async fn a_prompt_whose_result_was_read_is_not_in_flight_during_the_drain() { + let (client, mut stalled) = drop_in_the_exit_drain(prompt_response("end_turn")).await; + assert!( + !client.has_in_flight_prompt(), + "the response was read, so the prompt is no longer in flight" + ); + stalled.shutdown().await; + } + + /// A JSON-RPC error answering the prompt is its response too: read, and + /// not coming again. + #[tokio::test] + async fn a_prompt_whose_error_was_read_is_not_in_flight_during_the_drain() { + let error = serde_json::json!({ + "jsonrpc": "2.0", + "id": PROMPT_ID, + "error": {"code": -32000, "message": "the agent failed the turn"}, + }); + let (client, mut stalled) = drop_in_the_exit_drain(error).await; + assert!( + !client.has_in_flight_prompt(), + "the error was the response, so the prompt is no longer in flight" + ); + stalled.shutdown().await; + } + + /// The other side of the two above: a prompt that timed out still has its + /// response coming, so it stays in flight for the pool's cancel to drain. + #[tokio::test] + async fn a_prompt_that_timed_out_stays_in_flight_for_the_cancel() { + let mut client = spawn_inert_client().await; + client.last_prompt_id = Some(PROMPT_ID); + let max = std::time::Duration::from_secs(3600); + let result = client + .read_until_response_gated( + PARK_SESSION, + PROMPT_ID, + std::time::Duration::from_millis(200), + tokio::time::Instant::now() + max, + max, + None, + ) + .await; + assert!( + matches!(result, Err(AcpError::IdleTimeout(_))), + "expected an idle timeout, got {result:?}" + ); + assert_eq!( + client.last_prompt_id, + Some(PROMPT_ID), + "the response is still coming, and the cancel must drain it" + ); + } + + /// A turn dropped in its exit drain leaves the ids that drain had not + /// reached still owed, and no prompt in flight for a cancel to find. The + /// next prompt answers them before it is written. + #[tokio::test] + async fn a_new_prompt_first_answers_what_an_earlier_turn_left_owed() { + // Two cats: `client` reads the first and writes to the second, so the + // test reads, in order, exactly what the harness wrote. + let mut client = spawn_inert_client().await; + let mut wire = spawn_inert_client().await; + // Queued before the swap, so the prompt ends as soon as it is sent. A + // fresh client's first request, the prompt, is id 0. + let end_turn = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "result": {"stopReason": "end_turn"}, + }); + client + .write_ndjson(&end_turn) + .await + .expect("cat takes the response"); + std::mem::swap(&mut client.stdin, &mut wire.stdin); + client.pending_permissions.record(serde_json::json!(7)); + + let result = client + .session_prompt_with_idle_timeout( + PARK_SESSION, + "hello", + std::time::Duration::from_secs(5), + std::time::Duration::from_secs(10), + TurnAuthority::Guest, + None, + ) + .await; + assert!(matches!(result, Ok(StopReason::EndTurn)), "got {result:?}"); + + let first = next_written(&mut wire).await; + assert_eq!( + first["id"], 7, + "the stale id must be answered before the prompt, got {first}" + ); + assert_eq!(first["result"]["outcome"]["outcome"], "cancelled"); + let second = next_written(&mut wire).await; + assert_eq!(second["method"], "session/prompt", "got {second}"); + assert!(client.pending_permissions.is_empty()); + } + + /// `pool.rs` drops the prompt future before it cancels, so the reset at the + /// end of the prompt call never runs. The cancel drain must still decide + /// what the agent asks while it winds down as nobody's turn, not as the + /// dead one's. + #[tokio::test] + async fn a_cancel_after_a_dropped_owner_turn_decides_as_a_guest() { + let mut client = spawn_inert_client().await; + // What the agent says while it winds down, queued for `cat` to echo: a + // request, then the cancelled prompt's response. + for line in [probe_request_with_id(7), prompt_response("cancelled")] { + client + .write_ndjson(&line) + .await + .expect("cat takes the line"); + } + // As a dropped owner turn leaves them. + client.current_turn_authority = TurnAuthority::Owner; + client.last_prompt_id = Some(PROMPT_ID); + + let result = client + .cancel_with_cleanup_grace(PARK_SESSION, std::time::Duration::from_secs(5)) + .await; + assert!( + matches!(result, Ok(StopReason::Cancelled)), + "got {result:?}" + ); + assert_eq!(client.current_turn_authority, TurnAuthority::Guest); + + // The drain's answer comes back through `cat`, behind `session/cancel`. + let answer = loop { + let line = next_written(&mut client).await; + if line["id"] == 7 { + break line; + } + }; + assert_eq!( + answer["result"]["outcome"]["optionId"], "reject-1", + "a request raised during a cancel is decided as a guest's" + ); + } + + // ── Parking on the real read loop (bash fake agents, Linux CI) ─────────── + + /// A fresh client's first request, the prompt, is always id 0. + #[cfg(unix)] + const END_FIRST_PROMPT: &str = r#"{"jsonrpc":"2.0","id":0,"result":{"stopReason":"end_turn"}}"#; + + /// The same request as the probe, with no `options`, which the handler + /// can only fail with `?`. + #[cfg(unix)] + const OPTIONLESS_PERMISSION_REQUEST: &str = r#"{"jsonrpc":"2.0","id":8,"method":"session/request_permission","params":{"sessionId":"sess-test","toolCall":{"toolCallId":"t2","title":"Bash","kind":"execute"}}}"#; + + /// The replies a fake agent captured, waiting until it has written `n`. + /// A loop that answers on its way out can return before the agent has + /// read the answer. + #[cfg(unix)] + async fn awaited_replies(path: &std::path::Path, n: usize) -> Vec { + for _ in 0..100 { + if let Ok(raw) = std::fs::read_to_string(path) { + let lines: Vec<&str> = raw.lines().filter(|l| !l.is_empty()).collect(); + if lines.len() >= n { + return lines + .iter() + .map(|l| serde_json::from_str(l).expect("the harness writes JSON")) + .collect(); + } + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + panic!("the fake agent never captured {n} replies"); + } + + /// Prompt, get asked, capture the answer, then end the turn. + #[cfg(unix)] + fn ask_then_end_script(capture: &std::path::Path) -> String { + format!( + "read -r _prompt; printf '%s\\n' '{req}'; read -r reply; \ + printf '%s' \"$reply\" > '{capture}'; printf '%s\\n' '{done}'; sleep 5", + req = PROBE_PERMISSION_REQUEST, + capture = bash_path(capture), + done = END_FIRST_PROMPT, + ) + } + + #[cfg(unix)] + #[tokio::test] + async fn guest_permission_parks_and_an_allow_verdict_writes_allow_once() { + let capture = probe_capture_path("park-allow"); + let mut client = spawn_script(&ask_then_end_script(&capture)).await; + let (link, mut events) = approval_link(GateLimits::default()); + let approver = tokio::spawn(async move { + let ask = match events.recv().await { + Some(ApprovalEvent::Ask(ask)) => ask, + other => panic!("expected an ask, got {other:?}"), + }; + assert_eq!( + ask.digest, + request_digest(PARK_SESSION, &probe_tool_call()), + "the ask carries the digest of what the agent asked" + ); + ask.verdict_tx + .send(ApprovalVerdict { + digest: ask.digest, + decision: ApprovalDecision::Allow { + via: AllowVia::Approver, + }, + }) + .expect("the read loop is waiting on this verdict"); + match events.recv().await { + Some(ApprovalEvent::Settled(settled)) => settled, + other => panic!("expected settled, got {other:?}"), + } + }); + + let stop = client + .session_prompt_with_idle_timeout( + PARK_SESSION, + "hello", + std::time::Duration::from_secs(4), + std::time::Duration::from_secs(12), + TurnAuthority::Guest, + Some(link), + ) + .await + .expect("the turn ends normally"); + + assert_eq!(stop, StopReason::EndTurn); + assert_eq!(captured_answer(&capture), "allow-1"); + let settled = approver.await.expect("approver task"); + assert_eq!( + settled.settlement, + Settlement::Allowed { + via: AllowVia::Approver + } + ); + assert!(client.take_permission_refusals().is_empty()); + let _ = std::fs::remove_file(&capture); + } + + /// With no link, a guest's request is refused exactly as it was before + /// parking existed. + #[cfg(unix)] + #[tokio::test] + async fn guest_permission_without_a_link_is_refused_inline() { + let capture = probe_capture_path("park-nolink"); + let mut client = spawn_script(&ask_then_end_script(&capture)).await; + + let stop = client + .session_prompt_with_idle_timeout( + PARK_SESSION, + "hello", + std::time::Duration::from_secs(4), + std::time::Duration::from_secs(12), + TurnAuthority::Guest, + None, + ) + .await + .expect("the turn ends normally"); + + assert_eq!(stop, StopReason::EndTurn); + assert_eq!(captured_answer(&capture), "reject-1"); + assert_eq!( + client.take_permission_refusals(), + vec![refusal(RefusalReason::NoApprovalPath)] + ); + let _ = std::fs::remove_file(&capture); + } + + /// An ask nobody answers is refused at its expiry and the turn carries on. + /// The idle timeout is shorter than the window, so this also shows idle + /// is suspended while a person is being asked. + #[cfg(unix)] + #[tokio::test] + async fn an_unanswered_ask_expires_to_reject_and_the_turn_continues() { + let capture = probe_capture_path("park-expire"); + let mut client = spawn_script(&ask_then_end_script(&capture)).await; + let limits = GateLimits { + approval_timeout: std::time::Duration::from_secs(1), + min_ask_window: std::time::Duration::from_millis(100), + hard_margin: std::time::Duration::from_millis(100), + ..GateLimits::default() + }; + let (link, mut events) = approval_link(limits); + let started = tokio::time::Instant::now(); + + let stop = client + .session_prompt_with_idle_timeout( + PARK_SESSION, + "hello", + std::time::Duration::from_millis(300), + std::time::Duration::from_secs(12), + TurnAuthority::Guest, + Some(link), + ) + .await + .expect("an expiry refuses the request; it does not end the turn"); + + assert_eq!(stop, StopReason::EndTurn); + assert!(started.elapsed() >= limits.approval_timeout); + assert_eq!(captured_answer(&capture), "reject-1"); + let _ask = match events.try_recv() { + Ok(ApprovalEvent::Ask(ask)) => ask, + other => panic!("expected an ask, got {other:?}"), + }; + match events.try_recv() { + Ok(ApprovalEvent::Settled(settled)) => { + assert_eq!(settled.settlement, Settlement::Expired) + } + other => panic!("expected settled, got {other:?}"), + } + assert_eq!( + client.take_permission_refusals(), + vec![refusal(RefusalReason::Expired)] + ); + let _ = std::fs::remove_file(&capture); + } + + /// The prompt response arrives while a request is still parked. The exit + /// drain refuses it before the call returns. The in-flight check below + /// runs after the call has returned, where the prompt is cleared either + /// way; that a drop mid-drain finds no prompt in flight is pinned by + /// `a_prompt_whose_result_was_read_is_not_in_flight_during_the_drain`. + #[cfg(unix)] + #[tokio::test] + async fn prompt_result_while_parked_answers_reject_before_returning() { + let capture = probe_capture_path("park-exit"); + let script = format!( + "read -r _prompt; printf '%s\\n' '{req}'; printf '%s\\n' '{done}'; \ + read -r reply; printf '%s\\n' \"$reply\" > '{capture}'; sleep 5", + req = PROBE_PERMISSION_REQUEST, + done = END_FIRST_PROMPT, + capture = bash_path(&capture), + ); + let mut client = spawn_script(&script).await; + let (link, mut events) = approval_link(GateLimits::default()); + + let stop = client + .session_prompt_with_idle_timeout( + PARK_SESSION, + "hello", + std::time::Duration::from_secs(4), + std::time::Duration::from_secs(12), + TurnAuthority::Guest, + Some(link), + ) + .await + .expect("the turn ends normally"); + + assert_eq!(stop, StopReason::EndTurn); + assert!(!client.has_in_flight_prompt()); + let ask = match events.try_recv() { + Ok(ApprovalEvent::Ask(ask)) => ask, + other => panic!("expected an ask, got {other:?}"), + }; + match events.try_recv() { + Ok(ApprovalEvent::Settled(settled)) => { + assert_eq!(settled.settlement, Settlement::TurnEnded); + assert_eq!(settled.park_id, ask.park_id); + } + other => panic!("expected settled, got {other:?}"), + } + assert!( + ask.verdict_tx.is_closed(), + "a verdict sent now must be told it is too late" + ); + let replies = awaited_replies(&capture, 1).await; + assert_eq!(replies[0]["result"]["outcome"]["optionId"], "reject-1"); + assert_eq!( + client.take_permission_refusals(), + vec![refusal(RefusalReason::TurnEnded)] + ); + let _ = std::fs::remove_file(&capture); + } + + /// A request with no options fails the handler with `?`. The drain still + /// refuses the parked request and answers the broken one `cancelled`. + #[cfg(unix)] + #[tokio::test] + async fn a_question_mark_exit_answers_the_parked_request() { + let capture = probe_capture_path("park-qmark"); + let script = format!( + "read -r _prompt; printf '%s\\n' '{req}'; printf '%s\\n' '{broken}'; \ + read -r first; read -r second; \ + printf '%s\\n%s\\n' \"$first\" \"$second\" > '{capture}'; sleep 5", + req = PROBE_PERMISSION_REQUEST, + broken = OPTIONLESS_PERMISSION_REQUEST, + capture = bash_path(&capture), + ); + let mut client = spawn_script(&script).await; + let (link, mut events) = approval_link(GateLimits::default()); + + let result = client + .session_prompt_with_idle_timeout( + PARK_SESSION, + "hello", + std::time::Duration::from_secs(4), + std::time::Duration::from_secs(12), + TurnAuthority::Guest, + Some(link), + ) + .await; + + assert!( + matches!(result, Err(AcpError::Protocol(_))), + "got {result:?}" + ); + let replies = awaited_replies(&capture, 2).await; + assert_eq!(replies[0]["id"], 7); + assert_eq!(replies[0]["result"]["outcome"]["optionId"], "reject-1"); + assert_eq!(replies[1]["id"], 8); + assert_eq!(replies[1]["result"]["outcome"]["outcome"], "cancelled"); + let ask = match events.try_recv() { + Ok(ApprovalEvent::Ask(ask)) => ask, + other => panic!("expected an ask, got {other:?}"), + }; + assert!(ask.verdict_tx.is_closed()); + let _ = std::fs::remove_file(&capture); + } + + /// The `pool.rs` control arm drops the prompt future while a request is + /// parked. Only `cancel_with_cleanup_grace` runs after that, and it must + /// answer the parked request before it sends `session/cancel`. + #[cfg(unix)] + #[tokio::test] + async fn dropped_prompt_future_is_answered_cancelled_by_cancel_with_cleanup_grace() { + let capture = probe_capture_path("park-drop"); + let script = format!( + "read -r _prompt; printf '%s\\n' '{req}'; read -r reply; \ + printf '%s\\n' \"$reply\" > '{capture}'; read -r _cancel; \ + printf '%s\\n' '{cancelled}'; sleep 5", + req = PROBE_PERMISSION_REQUEST, + capture = bash_path(&capture), + cancelled = r#"{"jsonrpc":"2.0","id":0,"result":{"stopReason":"cancelled"}}"#, + ); + let mut client = spawn_script(&script).await; + let (link, mut events) = approval_link(GateLimits::default()); + + let ask = { + let prompt = client.session_prompt_with_idle_timeout( + PARK_SESSION, + "hello", + std::time::Duration::from_secs(4), + std::time::Duration::from_secs(12), + TurnAuthority::Guest, + Some(link), + ); + tokio::select! { + biased; + event = events.recv() => match event { + Some(ApprovalEvent::Ask(ask)) => ask, + other => panic!("expected an ask, got {other:?}"), + }, + result = prompt => panic!("the prompt must still be parked, got {result:?}"), + } + }; + + let stop = client + .cancel_with_cleanup_grace(PARK_SESSION, std::time::Duration::from_secs(5)) + .await + .expect("the agent acknowledges the cancel"); + + assert_eq!(stop, StopReason::Cancelled); + let replies = awaited_replies(&capture, 1).await; + assert_eq!(replies[0]["id"], 7); + assert_eq!(replies[0]["result"]["outcome"]["outcome"], "cancelled"); + assert!( + ask.verdict_tx.is_closed(), + "the main loop must see that the ask is dead" + ); + let _ = std::fs::remove_file(&capture); + } + + /// The setup loop has no turn behind it and so nobody to ask: a guest's + /// request there is refused inline and can never leave an untimed RPC + /// waiting on a person. + #[cfg(unix)] + #[tokio::test] + async fn setup_rpc_permission_request_never_parks() { + let capture = probe_capture_path("park-setup"); + let script = format!( + "printf '%s\\n' '{req}'; read -r reply; printf '%s' \"$reply\" > '{capture}'; \ + printf '%s\\n' '{done}'; sleep 5", + req = PROBE_PERMISSION_REQUEST, + capture = bash_path(&capture), + done = r#"{"jsonrpc":"2.0","id":5,"result":{"ok":true}}"#, + ); + let mut client = spawn_script(&script).await; + + let result = tokio::time::timeout( + std::time::Duration::from_secs(10), + client.read_until_response(5), + ) + .await + .expect("the setup loop must finish"); + + assert!(result.is_ok(), "setup should complete, got {result:?}"); + assert_eq!(captured_answer(&capture), "reject-1"); + assert_eq!(client.pending_permissions.parked_count(), 0); + assert_eq!( + client.take_permission_refusals(), + vec![refusal(RefusalReason::NoApprovalPath)] + ); + let _ = std::fs::remove_file(&capture); + } + + /// A setup RPC whose loop fails on a broken request answers that request + /// on the way out, rather than leaving the agent blocked on it. + #[cfg(unix)] + #[tokio::test] + async fn a_setup_loop_error_answers_the_request_it_left_owed() { + let capture = probe_capture_path("park-setup-err"); + let script = format!( + "printf '%s\\n' '{broken}'; read -r reply; printf '%s\\n' \"$reply\" > '{capture}'; \ + sleep 5", + broken = OPTIONLESS_PERMISSION_REQUEST, + capture = bash_path(&capture), + ); + let mut client = spawn_script(&script).await; + + let result = tokio::time::timeout( + std::time::Duration::from_secs(10), + client.read_until_response(5), + ) + .await + .expect("the setup loop must finish"); + + assert!( + matches!(result, Err(AcpError::Protocol(_))), + "got {result:?}" + ); + let replies = awaited_replies(&capture, 1).await; + assert_eq!(replies[0]["id"], 8); + assert_eq!(replies[0]["result"]["outcome"]["outcome"], "cancelled"); + assert!(client.pending_permissions.next_owed().is_none()); + let _ = std::fs::remove_file(&capture); + } + + use super::*; + + #[test] + fn stop_reason_parses_all_known_values() { + assert_eq!(StopReason::from_str("end_turn"), Some(StopReason::EndTurn)); + assert_eq!( + StopReason::from_str("cancelled"), + Some(StopReason::Cancelled) + ); + assert_eq!( + StopReason::from_str("max_tokens"), + Some(StopReason::MaxTokens) + ); + assert_eq!( + StopReason::from_str("max_turn_requests"), + Some(StopReason::MaxTurnRequests) + ); + assert_eq!(StopReason::from_str("refusal"), Some(StopReason::Refusal)); + } + + #[test] + fn stop_reason_returns_none_for_unknown() { + assert_eq!(StopReason::from_str("unknown_value"), None); + assert_eq!(StopReason::from_str(""), None); + assert_eq!(StopReason::from_str("endturn"), None); // no camelCase — still unknown + } + + #[test] + fn stop_reason_is_case_insensitive() { + // Agents may send uppercase or mixed-case variants — all should parse correctly. + assert_eq!(StopReason::from_str("END_TURN"), Some(StopReason::EndTurn)); + assert_eq!( + StopReason::from_str("CANCELLED"), + Some(StopReason::Cancelled) + ); + assert_eq!( + StopReason::from_str("Max_Tokens"), + Some(StopReason::MaxTokens) + ); + assert_eq!( + StopReason::from_str("MAX_TURN_REQUESTS"), + Some(StopReason::MaxTurnRequests) + ); + assert_eq!(StopReason::from_str("Refusal"), Some(StopReason::Refusal)); } #[test] @@ -5004,6 +6738,7 @@ mod tests { std::time::Duration::from_secs(5), // A harness-driven wire test, not a channel turn: no inbound author. TurnAuthority::Owner, + None, ) .await .expect("wire prompt"); diff --git a/crates/buzz-acp/src/approval_gate.rs b/crates/buzz-acp/src/approval_gate.rs new file mode 100644 index 00000000000..2db7521c84d --- /dev/null +++ b/crates/buzz-acp/src/approval_gate.rs @@ -0,0 +1,1062 @@ +//! The pure half of parking a permission request until an approver answers. +//! +//! # Why this exists +//! +//! A guest turn used to refuse every permission request outright. Many of +//! those are requests the owner would allow, so a guest turn can now *park* a +//! request instead: write nothing to the agent yet, hand the request to the +//! main loop, and wait for a verdict. This module holds every rule that +//! decides how that goes (whether a request may park, how long it may wait, +//! which verdict counts, and how waiting bends the turn's deadlines) and none +//! of the I/O. +//! +//! It is its own module for two reasons. +//! +//! **Testable everywhere.** The read loop can only be exercised against a +//! spawned agent, and those tests shell out to bash, which the Windows build +//! cannot run. Kept pure, every rule here is a plain unit test on every +//! platform. +//! +//! **One owner of the contract.** The read loop and the main loop meet at the +//! types below: an [`ApprovalEvent`] goes out, an [`ApprovalVerdict`] comes +//! back, and a [`RequestDigest`] binds the two. Both sides import them from +//! here and nowhere else. Two digest implementations that disagreed by a +//! single byte would refuse every verdict as a mismatch, and two +//! [`DenyReason`]s would need a translation that could silently drop a case. +//! +//! # What makes a verdict trustworthy +//! +//! - It arrives only on the oneshot created for that one request. Neither the +//! agent nor a guest can write to it, and it cannot reach another request. +//! - It is applied only when its digest matches the parked request's. The +//! main loop binds its ticket to that digest, so a verdict cross-wired to +//! the wrong request is refused instead of trusted. +//! - Every wait is bounded. An unanswered request expires to `reject_once`, +//! and parking is capped per connection and per turn. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use sha2::{Digest, Sha256}; +use tokio::time::Instant; + +/// How long a parked request waits for an approver before it is refused. +/// +/// Long enough for the owner to see a notification and answer from a phone, +/// short enough that an abandoned ask does not hold a turn open for long. +pub(crate) const DEFAULT_APPROVAL_TIMEOUT: Duration = Duration::from_secs(600); + +/// Most requests one connection may have parked at once. +/// +/// One assistant message can carry several tool calls, so more than one can +/// legitimately be waiting. Past a handful, the agent is asking faster than a +/// person can answer, and refusing is kinder than stacking up cards. +pub(crate) const MAX_PARKED_PER_CONNECTION: usize = 4; + +/// Most asks one turn may put in front of a person. +/// +/// Counts only asks that paged someone. A request a standing grant answers is +/// refunded, or a one-hour grant would stop working after six calls. +pub(crate) const MAX_ASKS_PER_TURN: u8 = 6; + +/// How far inside the turn's hard deadline an ask must expire. +/// +/// An expiry answers the agent `reject_once` and lets the turn carry on. The +/// margin leaves room for that answer to land before the hard deadline ends +/// the turn outright. +pub(crate) const APPROVAL_HARD_MARGIN: Duration = Duration::from_secs(10); + +/// The shortest window worth asking in. +/// +/// Nobody can read and answer a card faster, so a shorter window is refused up +/// front rather than posted and left to expire. +pub(crate) const MIN_ASK_WINDOW: Duration = Duration::from_secs(30); + +/// How many approval timeouts of waiting one turn may be credited in total. +/// +/// Waiting extends the turn's hard deadline (see [`TurnClock`]). The cap stops +/// a guest from stretching one turn's wall clock without limit by asking over +/// and over. +pub(crate) const CREDIT_CAP_TIMEOUTS: u32 = 3; + +/// Domain separator for [`request_digest`], so a digest of a permission +/// request can never equal a SHA-256 taken over the same bytes for another +/// purpose. +const DIGEST_DOMAIN: &[u8] = b"buzz-acp/permission-digest/v1\0"; + +/// The caps and windows one turn's parking runs under. +/// +/// A value rather than the constants themselves so tests can shrink the +/// windows. Production always builds it with [`GateLimits::production`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct GateLimits { + pub(crate) approval_timeout: Duration, + pub(crate) max_parked: usize, + pub(crate) max_asks_per_turn: u8, + pub(crate) hard_margin: Duration, + pub(crate) min_ask_window: Duration, + pub(crate) credit_cap: Duration, +} + +impl GateLimits { + /// The production limits for a given approval timeout. + /// + /// The credit cap follows the timeout, so a longer window still buys the + /// same number of full waits per turn. + pub(crate) fn production(approval_timeout: Duration) -> Self { + Self { + approval_timeout, + max_parked: MAX_PARKED_PER_CONNECTION, + max_asks_per_turn: MAX_ASKS_PER_TURN, + hard_margin: APPROVAL_HARD_MARGIN, + min_ask_window: MIN_ASK_WINDOW, + credit_cap: approval_timeout.saturating_mul(CREDIT_CAP_TIMEOUTS), + } + } +} + +impl Default for GateLimits { + fn default() -> Self { + Self::production(DEFAULT_APPROVAL_TIMEOUT) + } +} + +/// SHA-256 over one permission request, binding a verdict to exactly the +/// request the approver was shown. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub(crate) struct RequestDigest(pub(crate) [u8; 32]); + +impl RequestDigest { + /// Hex form, for logs and the observer feed. + pub(crate) fn to_hex(self) -> String { + hex::encode(self.0) + } +} + +/// Names one parked request for the life of the process. +/// +/// Minted from a process-wide counter and never reused, so a late +/// [`ApprovalSettled`] can never be mistaken for a newer request's. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub(crate) struct ParkId(pub(crate) u64); + +static NEXT_PARK_ID: AtomicU64 = AtomicU64::new(1); + +impl ParkId { + pub(crate) fn mint() -> Self { + ParkId(NEXT_PARK_ID.fetch_add(1, Ordering::Relaxed)) + } +} + +/// Who allowed a request. +/// +/// Carried through to the settlement because a standing grant's allow is +/// refunded against the turn's ask budget and a person's is not. +// `expect`, not `allow`: the day the main loop's approval route builds or +// reads these, the expectation fails the build until the attribute goes. +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "the main loop's approval route builds these; until it lands only tests do" + ) +)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum AllowVia { + /// A person answered this request. + Approver, + /// A grant the owner gave earlier covered it, and nobody was paged. + StandingGrant, +} + +impl AllowVia { + /// The name the observer feed reports. + pub(crate) fn as_str(&self) -> &'static str { + match self { + AllowVia::Approver => "approver", + AllowVia::StandingGrant => "standing_grant", + } + } +} + +/// Why the main loop refused a request. +/// +/// Owned here with the rest of the contract. The main loop adds its reasons +/// to this enum rather than keeping its own, so no reason is ever lost in a +/// translation between two sets. +#[expect( + dead_code, + reason = "the main loop's approval route builds these; tests build only some" +)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum DenyReason { + /// An approver said no. + Approver, + /// Nobody holds the authority to approve for this agent. + NoApprover, + /// The ask could not be delivered to the approver. + Undeliverable, + /// The request could not be shown faithfully, so nobody can be asked to + /// approve it. + Unrenderable, + /// The answer could not be tied to exactly one request. + Ambiguous, + /// The harness is shutting down. + Shutdown, +} + +/// The main loop's answer to one ask. +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "the main loop's approval route builds these; until it lands only tests do" + ) +)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ApprovalDecision { + Allow { via: AllowVia }, + Deny(DenyReason), +} + +/// A verdict on its way back to the read loop. +/// +/// Carries the digest the main loop bound its ticket to. The read loop +/// compares it with the parked request's own before acting, so a verdict can +/// only ever allow the request it was issued for. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct ApprovalVerdict { + pub(crate) digest: RequestDigest, + pub(crate) decision: ApprovalDecision, +} + +/// How a parked request ended. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Settlement { + /// `allow_once` was written to the agent. + Allowed { via: AllowVia }, + /// The main loop refused it, and `reject_once` was written. + Denied(DenyReason), + /// Nobody answered in time, and `reject_once` was written. + Expired, + /// The main loop dropped the verdict sender without answering. Refused at + /// once rather than left to wait out its expiry. + VerdictChannelClosed, + /// The verdict carried another request's digest. Refused, and a bug. + DigestMismatch, + /// The turn ended while it waited, and `reject_once` was written. + TurnEnded, +} + +impl Settlement { + /// The name the observer feed reports. + pub(crate) fn as_str(&self) -> &'static str { + match self { + Settlement::Allowed { .. } => "allowed", + Settlement::Denied(_) => "denied", + Settlement::Expired => "expired", + Settlement::VerdictChannelClosed => "verdict_channel_closed", + Settlement::DigestMismatch => "digest_mismatch", + Settlement::TurnEnded => "turn_ended", + } + } +} + +/// Why a permission request was refused, kept so the layer above can say so. +/// +/// The first six are decided before anything is parked. The rest are the ways +/// a parked request can end without being allowed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum RefusalReason { + /// The agent offered no `allow_once`, so there was nothing to approve. + NoAllowOffered, + /// The request needed an approver, and this turn had no way to reach one. + NoApprovalPath, + /// The request named a session other than the one this turn is running. + SessionMismatch, + /// This connection already had the most requests it may park. + TooManyParked, + /// This turn had already asked a person the most times it may. + AskBudgetExhausted, + /// Too little of the turn was left to give a person time to answer. + NoTimeBudget, + Denied(DenyReason), + Expired, + VerdictChannelClosed, + DigestMismatch, + TurnEnded, +} + +/// One parked request, handed to the main loop to find an approver. +/// +/// Sent exactly once per parked request. The read loop keeps the receiving +/// end of `verdict_tx`, so whatever the main loop sends on it can only ever +/// reach this request. +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "the main loop's approval route reads these; until it lands only tests do" + ) +)] +#[derive(Debug)] +pub(crate) struct ApprovalAsk { + /// Pairs this ask with its later [`ApprovalSettled`]. + pub(crate) park_id: ParkId, + /// The turn that parked it, so the main loop can drop the turn's asks + /// when the turn ends. + pub(crate) turn_id: String, + pub(crate) session_id: String, + /// What the main loop binds its ticket to, and echoes in the verdict. + pub(crate) digest: RequestDigest, + /// `params.toolCall` exactly as the agent sent it, to render the card. + pub(crate) tool_call: serde_json::Value, + pub(crate) parked_at: Instant, + /// When the read loop refuses the request if no verdict has arrived. + /// + /// Authoritative. Anything shown to the approver must come from this, not + /// from a configured timeout: the read loop may have clamped it to fit + /// the turn's hard deadline. + pub(crate) expires_at: Instant, + /// The one way back in. + pub(crate) verdict_tx: tokio::sync::oneshot::Sender, +} + +/// Tells the main loop how a parked request ended. +/// +/// Sent only after the answer is on the wire, so `Allowed` means the agent +/// really was told `allow_once`. A successful `send` of a verdict promises no +/// such thing: the write that follows it can still fail. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ApprovalSettled { + pub(crate) park_id: ParkId, + pub(crate) turn_id: String, + pub(crate) settlement: Settlement, + pub(crate) waited: Duration, +} + +/// Everything the read loop tells the main loop about parked requests. +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "the main loop's approval route reads these; until it lands only tests do" + ) +)] +#[derive(Debug)] +pub(crate) enum ApprovalEvent { + Ask(ApprovalAsk), + Settled(ApprovalSettled), +} + +/// One turn's capability to park requests, built by the dispatcher. +/// +/// Passed to the prompt call as a parameter rather than installed on the +/// connection. A link left on the connection would be picked up by whichever +/// read loop ran next, and a session's `initial_message` runs its own read +/// loop before the turn's, so the first guest turn of every new session could +/// never park. +#[derive(Debug, Clone)] +pub(crate) struct ApprovalLink { + /// Unbounded so handing off an ask never awaits. A bounded send could + /// stall the one loop that has to keep answering the agent. + pub(crate) events_tx: tokio::sync::mpsc::UnboundedSender, + pub(crate) turn_id: String, + pub(crate) limits: GateLimits, +} + +/// A turn's link plus the number of people it has asked so far. +/// +/// Built only inside the gated prompt read loop. The setup loop and the +/// cancel drain never hold one, which is what makes parking impossible there: +/// neither is a turn, so neither has anyone to ask. +#[derive(Debug)] +pub(crate) struct ParkGate { + link: ApprovalLink, + asks_charged: u8, +} + +impl ParkGate { + pub(crate) fn new(link: ApprovalLink) -> Self { + Self { + link, + asks_charged: 0, + } + } + + pub(crate) fn limits(&self) -> &GateLimits { + &self.link.limits + } + + pub(crate) fn turn_id(&self) -> &str { + &self.link.turn_id + } + + pub(crate) fn asks_charged(&self) -> u8 { + self.asks_charged + } + + pub(crate) fn charge(&mut self) { + self.asks_charged = self.asks_charged.saturating_add(1); + } + + pub(crate) fn refund(&mut self) { + self.asks_charged = self.asks_charged.saturating_sub(1); + } + + /// Hand an event to the main loop. + /// + /// `false` means the main loop is gone. The caller then refuses at once + /// instead of waiting out an ask nobody will ever see. + pub(crate) fn send(&self, event: ApprovalEvent) -> bool { + self.link.events_tx.send(event).is_ok() + } +} + +/// Whether a request that needs an approver may park. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Admission { + Park { expires_at: Instant }, + Refuse(RefusalReason), +} + +/// Decide whether a request that needs an approver may park. +/// +/// The session check comes first: a request for another session is never +/// parked, whatever the caps say. The time budget comes last, so a request +/// refused by a cap is reported as that cap. +pub(crate) fn admit( + parked_now: usize, + asks_charged: u8, + session_matches: bool, + ask_expiry: Option, + limits: &GateLimits, +) -> Admission { + if !session_matches { + return Admission::Refuse(RefusalReason::SessionMismatch); + } + if parked_now >= limits.max_parked { + return Admission::Refuse(RefusalReason::TooManyParked); + } + if asks_charged >= limits.max_asks_per_turn { + return Admission::Refuse(RefusalReason::AskBudgetExhausted); + } + match ask_expiry { + Some(expires_at) => Admission::Park { expires_at }, + None => Admission::Refuse(RefusalReason::NoTimeBudget), + } +} + +/// What became of a parked request's verdict channel. +#[derive(Debug)] +pub(crate) enum VerdictOutcome { + Delivered(ApprovalVerdict), + /// The sender was dropped without a verdict. + Closed, + /// No verdict had arrived by the expiry. + Expired, + /// The turn ended first. + TurnEnded, +} + +/// What to write to the agent for one outcome, and what to report about it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct Resolution { + /// Write `allow_once`. Otherwise write `reject_once`. + pub(crate) allow: bool, + pub(crate) settlement: Settlement, + pub(crate) refusal: Option, + /// Give the ask back to the turn's budget. + pub(crate) refund_ask: bool, +} + +/// Map an outcome to an answer. +/// +/// Total, and deny by default: exactly one shape allows, a delivered `Allow` +/// whose digest matches. The digest is checked before the decision, so a +/// verdict for another request is refused whatever it says. +pub(crate) fn resolve(parked_digest: &RequestDigest, outcome: VerdictOutcome) -> Resolution { + let deny = |settlement: Settlement, reason: RefusalReason| Resolution { + allow: false, + settlement, + refusal: Some(reason), + refund_ask: false, + }; + match outcome { + VerdictOutcome::Delivered(v) if v.digest != *parked_digest => { + deny(Settlement::DigestMismatch, RefusalReason::DigestMismatch) + } + VerdictOutcome::Delivered(ApprovalVerdict { + decision: ApprovalDecision::Allow { via }, + .. + }) => Resolution { + allow: true, + settlement: Settlement::Allowed { via }, + refusal: None, + refund_ask: via == AllowVia::StandingGrant, + }, + VerdictOutcome::Delivered(ApprovalVerdict { + decision: ApprovalDecision::Deny(r), + .. + }) => deny(Settlement::Denied(r), RefusalReason::Denied(r)), + VerdictOutcome::Closed => deny( + Settlement::VerdictChannelClosed, + RefusalReason::VerdictChannelClosed, + ), + VerdictOutcome::Expired => deny(Settlement::Expired, RefusalReason::Expired), + VerdictOutcome::TurnEnded => deny(Settlement::TurnEnded, RefusalReason::TurnEnded), + } +} + +/// Serialize `value` with object keys sorted at every depth. +/// +/// Explicit rather than trusting serde_json's map order: the digest must not +/// change if some crate in the build turns on serde_json's `preserve_order`. +pub(crate) fn canonical_json(value: &serde_json::Value) -> Vec { + fn canonicalize(v: &serde_json::Value) -> serde_json::Value { + match v { + serde_json::Value::Object(map) => { + let mut keys: Vec<&String> = map.keys().collect(); + keys.sort(); + let mut out = serde_json::Map::new(); + for k in keys { + out.insert(k.clone(), canonicalize(&map[k])); + } + serde_json::Value::Object(out) + } + serde_json::Value::Array(items) => { + serde_json::Value::Array(items.iter().map(canonicalize).collect()) + } + other => other.clone(), + } + } + serde_json::to_vec(&canonicalize(value)).expect("a serde_json::Value always serializes") +} + +/// Digest one permission request, as the read loop saw it. +/// +/// SHA-256 over a domain separator, then each field prefixed with its length, +/// so two different requests cannot hash the same bytes by moving a boundary +/// (`"ab"` + `"c"` against `"a"` + `"bc"`). A missing string field hashes as +/// `""`, and a missing `rawInput` as JSON `null`, which differs from `{}`. +pub(crate) fn request_digest(session_id: &str, tool_call: &serde_json::Value) -> RequestDigest { + let text = |key: &str| tool_call.get(key).and_then(|v| v.as_str()).unwrap_or(""); + let raw_input = canonical_json( + tool_call + .get("rawInput") + .unwrap_or(&serde_json::Value::Null), + ); + let mut hasher = Sha256::new(); + hasher.update(DIGEST_DOMAIN); + for field in [ + session_id.as_bytes(), + text("toolCallId").as_bytes(), + text("kind").as_bytes(), + text("title").as_bytes(), + raw_input.as_slice(), + ] { + hasher.update((field.len() as u64).to_be_bytes()); + hasher.update(field); + } + let out = hasher.finalize(); + let mut digest = [0u8; 32]; + digest.copy_from_slice(&out); + RequestDigest(digest) +} + +/// A short name for the tool, for people: its title, else its kind, else +/// `(untitled)`. +pub(crate) fn tool_label(tool_call: &serde_json::Value) -> String { + tool_call + .get("title") + .and_then(|v| v.as_str()) + .or_else(|| tool_call.get("kind").and_then(|v| v.as_str())) + .unwrap_or("(untitled)") + .to_string() +} + +/// Which of the read loop's deadlines comes due next. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Fire { + /// The agent has been silent for the idle timeout. + Idle, + /// The turn reached its hard deadline, credit included. + Hard, + /// A parked request reached its expiry. It is refused and the turn + /// carries on. + Approval, +} + +/// The read loop's deadlines, with waiting on an approver taken into account. +/// +/// Replaces three loose locals (the idle deadline, the hard deadline and the +/// time of the last activity) because parking changes how all three behave, +/// and the rules are easier to get right, and to test, in one place that is +/// handed the time instead of reading it. +/// +/// - **Idle is suspended while anything is parked.** The agent is silent +/// because it is waiting on us, not because it is stuck. +/// - **Idle restarts from the moment a wait ends,** not from the agent's last +/// line before it. Otherwise a turn that waited longer than the idle +/// timeout would die the instant the owner said yes. +/// - **Hard is credited for the time spent waiting,** continuously while the +/// wait is open, so an ask made just before the hard deadline still gets +/// its full window. The credit is capped per turn, so asking over and over +/// cannot buy unlimited wall clock. +#[derive(Debug, Clone)] +pub(crate) struct TurnClock { + idle_timeout: Duration, + idle_deadline: Instant, + last_activity_at: Instant, + hard_base: Instant, + credit_cap: Duration, + credit_banked: Duration, + wait_started: Option, +} + +impl TurnClock { + pub(crate) fn new( + now: Instant, + idle_timeout: Duration, + hard_deadline: Instant, + credit_cap: Duration, + ) -> Self { + Self { + idle_timeout, + idle_deadline: now + idle_timeout, + last_activity_at: now, + hard_base: hard_deadline, + credit_cap, + credit_banked: Duration::ZERO, + wait_started: None, + } + } + + /// The agent wrote something: restart the idle window. + pub(crate) fn on_activity(&mut self, now: Instant) { + self.idle_deadline = now + self.idle_timeout; + self.last_activity_at = now; + } + + /// Record whether anything is parked right now. + /// + /// Call once per loop iteration, before reading any deadline. The change + /// to nothing parked banks the wait's credit and restarts idle from `now`. + pub(crate) fn sync(&mut self, now: Instant, any_parked: bool) { + match (self.wait_started, any_parked) { + (None, true) => self.wait_started = Some(now), + (Some(started), false) => { + self.credit_banked = self + .credit_banked + .saturating_add(now.saturating_duration_since(started)) + .min(self.credit_cap); + self.wait_started = None; + self.idle_deadline = now + self.idle_timeout; + self.last_activity_at = now; + } + _ => {} + } + } + + fn banked(&self) -> Duration { + self.credit_banked.min(self.credit_cap) + } + + /// Credit earned so far, a wait that is still open included. + pub(crate) fn credit_at(&self, now: Instant) -> Duration { + let open = self + .wait_started + .map(|started| now.saturating_duration_since(started)) + .unwrap_or_default(); + self.credit_banked.saturating_add(open).min(self.credit_cap) + } + + /// The hard deadline with the credit earned so far. + pub(crate) fn effective_hard(&self, now: Instant) -> Instant { + self.hard_base + self.credit_at(now) + } + + /// The hard deadline to aim at while a wait that began at `started` is + /// open. + /// + /// Credit accrues as fast as time passes, so the deadline only comes due + /// once the cap is reached. Aiming straight at that point gives the loop + /// a fixed target instead of one that moves with every tick. The one + /// exception is a deadline that had already passed when the wait began: + /// no credit was owed, and it fires at once. + fn hard_if_waiting_since(&self, started: Instant) -> Instant { + let at_start = self.hard_base + self.banked(); + if at_start <= started { + at_start + } else { + self.hard_base + self.credit_cap + } + } + + /// The next deadline to sleep until, and which one it is. + /// + /// While anything is parked, idle is not a candidate, and an expiry that + /// ties with the hard deadline wins: refusing one request and carrying on + /// is gentler than ending the turn. With nothing parked this is the rule + /// the loop always had: idle when it is strictly earlier, otherwise hard. + pub(crate) fn next_deadline( + &self, + now: Instant, + earliest_expiry: Option, + ) -> (Instant, Fire) { + match self.wait_started { + Some(started) => { + let hard = self.hard_if_waiting_since(started); + match earliest_expiry { + Some(expiry) if expiry <= hard => (expiry, Fire::Approval), + _ => (hard, Fire::Hard), + } + } + None => { + let hard = self.effective_hard(now); + if self.idle_deadline < hard { + (self.idle_deadline, Fire::Idle) + } else { + (hard, Fire::Hard) + } + } + } + } + + /// Push the hard deadline out to at least `candidate`, keeping the credit + /// already earned on top. Never moves it earlier. Returns whether it + /// moved. + pub(crate) fn renew_hard(&mut self, now: Instant, candidate: Instant) -> bool { + let effective = self.effective_hard(now); + if candidate > effective { + self.hard_base += candidate - effective; + true + } else { + false + } + } + + /// How long the agent has been silent, for the hard-timeout error. + pub(crate) fn silence(&self, now: Instant) -> Duration { + now.saturating_duration_since(self.last_activity_at) + } + + /// When an ask made at `now` must expire. + /// + /// The full approval timeout, clamped to land `hard_margin` inside the + /// hard deadline the ask would be waiting against. `None` when that + /// leaves less than `min_ask_window`: asking a question nobody can answer + /// in time is worse than refusing straight away. + pub(crate) fn ask_expiry(&self, now: Instant, limits: &GateLimits) -> Option { + let hard = self.hard_if_waiting_since(self.wait_started.unwrap_or(now)); + let latest = hard.checked_sub(limits.hard_margin)?; + let expiry = (now + limits.approval_timeout).min(latest); + (expiry >= now + limits.min_ask_window).then_some(expiry) + } + + #[cfg(test)] + pub(crate) fn idle_deadline(&self) -> Instant { + self.idle_deadline + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + const S: fn(u64) -> Duration = Duration::from_secs; + + fn clock(now: Instant, idle: u64, hard_in: u64, cap: u64) -> TurnClock { + TurnClock::new(now, S(idle), now + S(hard_in), S(cap)) + } + + // ── Deadlines ──────────────────────────────────────────────────────────── + + #[test] + fn idle_is_not_a_candidate_while_parked() { + let t0 = Instant::now(); + let mut c = clock(t0, 1, 7200, 1800); + c.sync(t0, true); + let (at, fire) = c.next_deadline(t0 + S(5), Some(t0 + S(600))); + assert_eq!( + (at, fire), + (t0 + S(600), Fire::Approval), + "a 1s idle timeout must not fire while a person is being asked" + ); + } + + /// The regression where a turn dies the instant the owner says yes: the + /// idle window must restart when the wait ends, not run on from the + /// agent's last line before it. + #[test] + fn resume_recomputes_idle_from_now_not_from_the_last_line() { + let t0 = Instant::now(); + let mut c = clock(t0, 60, 7200, 1800); + c.on_activity(t0); + c.sync(t0 + S(1), true); + c.sync(t0 + S(600), false); + assert_eq!(c.idle_deadline(), t0 + S(660)); + assert_eq!( + c.next_deadline(t0 + S(600), None), + (t0 + S(660), Fire::Idle) + ); + } + + /// Credit runs while the wait is open, not only once it closes, so an ask + /// made just before the hard deadline is not killed partway through. + #[test] + fn hard_is_credited_for_the_wait_continuously() { + let t0 = Instant::now(); + let mut c = clock(t0, 60, 100, 1800); + c.sync(t0 + S(90), true); + assert_eq!(c.effective_hard(t0 + S(500)), t0 + S(510)); + c.sync(t0 + S(390), false); + assert_eq!(c.effective_hard(t0 + S(390)), t0 + S(400)); + } + + #[test] + fn credit_is_capped_per_turn_across_waits() { + let t0 = Instant::now(); + let mut c = clock(t0, 60, 7200, 1800); + for i in 0..3u64 { + let start = t0 + S(i * 2000); + c.sync(start, true); + c.sync(start + S(900), false); + } + assert_eq!( + c.credit_at(t0 + S(10_000)), + S(1800), + "three 15-minute waits bank only the 30-minute cap" + ); + assert_eq!(c.effective_hard(t0 + S(10_000)), t0 + S(7200 + 1800)); + } + + #[test] + fn hard_target_is_stable_while_parked_so_the_loop_cannot_spin() { + let t0 = Instant::now(); + let mut c = clock(t0, 60, 5, 1800); + c.sync(t0 + S(1), true); + let a = c.next_deadline(t0 + S(2), None); + let b = c.next_deadline(t0 + S(900), None); + assert_eq!(a, b, "sleep_until must not chase a moving deadline"); + assert_eq!(a, (t0 + S(5 + 1800), Fire::Hard)); + } + + /// A park that races past the hard deadline gets no credit it was never + /// owed. + #[test] + fn hard_that_was_already_due_when_the_wait_began_fires_at_once() { + let t0 = Instant::now(); + let mut c = clock(t0, 60, 5, 1800); + c.sync(t0 + S(6), true); + let (at, fire) = c.next_deadline(t0 + S(6), Some(t0 + S(600))); + assert_eq!((at, fire), (t0 + S(5), Fire::Hard)); + } + + #[test] + fn approval_wins_a_tie_with_hard() { + let t0 = Instant::now(); + let mut c = clock(t0, 60, 100, 0); + c.sync(t0, true); + assert_eq!( + c.next_deadline(t0, Some(t0 + S(100))), + (t0 + S(100), Fire::Approval), + "refusing one request and carrying on beats ending the turn" + ); + } + + #[test] + fn unparked_classification_is_todays_rule() { + let t0 = Instant::now(); + let c = clock(t0, 60, 100, 1800); + assert_eq!(c.next_deadline(t0, None), (t0 + S(60), Fire::Idle)); + let c = clock(t0, 100, 100, 1800); + assert_eq!( + c.next_deadline(t0, None), + (t0 + S(100), Fire::Hard), + "a tie with nothing parked goes to hard, as it always has" + ); + } + + #[test] + fn steer_renewal_is_monotonic_and_keeps_accrued_credit() { + let t0 = Instant::now(); + let mut c = clock(t0, 60, 100, 1800); + assert!( + !c.renew_hard(t0, t0 + S(50)), + "renewal never moves hard earlier" + ); + assert_eq!(c.effective_hard(t0), t0 + S(100)); + c.sync(t0 + S(10), true); + assert!(c.renew_hard(t0 + S(20), t0 + S(500))); + assert_eq!(c.effective_hard(t0 + S(20)), t0 + S(500)); + assert_eq!( + c.effective_hard(t0 + S(30)), + t0 + S(510), + "credit keeps accruing on top of a renewal" + ); + } + + // ── Ask windows and admission ──────────────────────────────────────────── + + #[test] + fn ask_window_gets_full_timeout_near_the_hard_deadline_when_credit_remains() { + let t0 = Instant::now(); + let c = clock(t0, 60, 100, 1800); + let limits = GateLimits::default(); + assert_eq!(c.ask_expiry(t0 + S(90), &limits), Some(t0 + S(690))); + } + + #[test] + fn ask_window_is_clamped_inside_the_hard_margin_and_refused_when_too_small() { + let t0 = Instant::now(); + let limits = GateLimits::default(); + let c = clock(t0, 60, 300, 0); + assert_eq!(c.ask_expiry(t0, &limits), Some(t0 + S(290))); + let c = clock(t0, 60, 35, 0); + assert_eq!( + c.ask_expiry(t0, &limits), + None, + "25s is too short to answer in, so no ask is made" + ); + } + + #[test] + fn admission_caps() { + let t0 = Instant::now(); + let l = GateLimits::default(); + let e = Some(t0 + S(600)); + assert_eq!( + admit(4, 0, true, e, &l), + Admission::Refuse(RefusalReason::TooManyParked) + ); + assert_eq!( + admit(0, 6, true, e, &l), + Admission::Refuse(RefusalReason::AskBudgetExhausted) + ); + assert_eq!( + admit(0, 0, false, e, &l), + Admission::Refuse(RefusalReason::SessionMismatch) + ); + assert_eq!( + admit(0, 0, true, None, &l), + Admission::Refuse(RefusalReason::NoTimeBudget) + ); + assert_eq!( + admit(3, 5, true, e, &l), + Admission::Park { + expires_at: t0 + S(600) + } + ); + } + + #[test] + fn production_limits_credit_three_full_waits() { + let l = GateLimits::default(); + assert_eq!(l.approval_timeout, DEFAULT_APPROVAL_TIMEOUT); + assert_eq!(l.credit_cap, S(1800)); + assert_eq!(GateLimits::production(S(60)).credit_cap, S(180)); + } + + // ── Binding a verdict to its request ───────────────────────────────────── + + #[test] + fn only_a_matching_allow_allows() { + let d = RequestDigest([1; 32]); + let other = RequestDigest([2; 32]); + let v = |digest, decision| VerdictOutcome::Delivered(ApprovalVerdict { digest, decision }); + let allow = ApprovalDecision::Allow { + via: AllowVia::Approver, + }; + let person = resolve(&d, v(d, allow)); + assert!(person.allow && !person.refund_ask); + let grant = resolve( + &d, + v( + d, + ApprovalDecision::Allow { + via: AllowVia::StandingGrant, + }, + ), + ); + assert!( + grant.allow && grant.refund_ask, + "a grant paged nobody, so it gives the ask back" + ); + for o in [ + v(other, allow), + v(d, ApprovalDecision::Deny(DenyReason::Approver)), + VerdictOutcome::Closed, + VerdictOutcome::Expired, + VerdictOutcome::TurnEnded, + ] { + let r = resolve(&d, o); + assert!(!r.allow && r.refusal.is_some() && !r.refund_ask, "{r:?}"); + } + } + + #[test] + fn digest_ignores_key_order_but_binds_every_field() { + let a = json!({"toolCallId":"t1","kind":"execute","title":"Bash","rawInput":{"b":1,"a":{"y":2,"x":3}}}); + let b = json!({"rawInput":{"a":{"x":3,"y":2},"b":1},"title":"Bash","kind":"execute","toolCallId":"t1"}); + assert_eq!(request_digest("s", &a), request_digest("s", &b)); + let base = request_digest("s", &a); + assert_ne!(base, request_digest("s2", &a), "the session must be bound"); + for (k, v) in [ + ("toolCallId", json!("t2")), + ("kind", json!("edit")), + ("title", json!("Bash2")), + ("rawInput", json!({"b":1})), + ] { + let mut m = a.clone(); + m[k] = v; + assert_ne!(base, request_digest("s", &m), "{k} must be bound"); + } + let split1 = json!({"toolCallId":"ab","kind":"c"}); + let split2 = json!({"toolCallId":"a","kind":"bc"}); + assert_ne!( + request_digest("s", &split1), + request_digest("s", &split2), + "fields are length-prefixed, so a moved boundary changes the digest" + ); + let null_in = json!({"toolCallId":"t"}); + let empty_in = json!({"toolCallId":"t","rawInput":{}}); + assert_ne!( + request_digest("s", &null_in), + request_digest("s", &empty_in) + ); + } + + /// The observer feed's `settlement` and `via` values are read by the + /// desktop, so they are part of the wire and must not drift. + #[test] + fn settlement_names_are_the_observer_contract() { + let names: Vec<&str> = [ + Settlement::Allowed { + via: AllowVia::Approver, + }, + Settlement::Denied(DenyReason::Approver), + Settlement::Expired, + Settlement::VerdictChannelClosed, + Settlement::DigestMismatch, + Settlement::TurnEnded, + ] + .iter() + .map(Settlement::as_str) + .collect(); + assert_eq!( + names, + [ + "allowed", + "denied", + "expired", + "verdict_channel_closed", + "digest_mismatch", + "turn_ended" + ] + ); + assert_eq!(AllowVia::Approver.as_str(), "approver"); + assert_eq!(AllowVia::StandingGrant.as_str(), "standing_grant"); + } +} diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index d870a3ca5d8..f52943b6075 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1,6 +1,7 @@ #![deny(unsafe_code)] mod acp; +mod approval_gate; mod config; mod engram_fetch; mod filter; diff --git a/crates/buzz-acp/src/pending_permissions.rs b/crates/buzz-acp/src/pending_permissions.rs index 4fcddf6ba26..cbacf65bbdc 100644 --- a/crates/buzz-acp/src/pending_permissions.rs +++ b/crates/buzz-acp/src/pending_permissions.rs @@ -24,15 +24,97 @@ //! cleanup inside the loop runs. This collection therefore lives on the //! connection, outliving the dropped future, so `cancel_with_cleanup` can still //! answer every outstanding request. +//! +//! # A parked request is still an owed request +//! +//! A guest turn may *park* a request instead of answering it: nothing is +//! written to the agent until an approver's verdict arrives or the wait +//! expires. Parking is recorded on the request's existing entry, not in a +//! second collection, because the exits above do not care why a request is +//! unanswered. A parked id comes back from `next_owed` like any other, so a +//! dropped prompt future still gets it answered by `cancel_with_cleanup`. A +//! second collection would give every exit two places to look, and the exit +//! that missed one would leave the agent blocked forever. +//! +//! The parked state holds the only receiver a verdict for that request can +//! arrive on. Dropping the entry drops the receiver, and that is how the main +//! loop learns an ask is dead: its `send` fails. + +use std::future::Future; +use std::pin::Pin; +use std::task::Poll; + +use tokio::sync::oneshot; +use tokio::time::Instant; + +use crate::approval_gate::{ApprovalVerdict, ParkId, RequestDigest}; + +/// A request waiting on an approver. +/// +/// Holds everything needed to answer it later without the original message. +/// Both option ids are the agent's own, looked up by kind when the request +/// arrived, because the ids are the agent's to choose. +#[derive(Debug)] +pub(crate) struct ParkedAsk { + pub(crate) park_id: ParkId, + /// Written only for a delivered allow whose digest matches. + pub(crate) allow_option_id: String, + /// Written for every other outcome. + pub(crate) reject_option_id: String, + /// Compared with every verdict before it is applied. + pub(crate) digest: RequestDigest, + /// For the refusal notice and the logs. + pub(crate) tool: String, + pub(crate) parked_at: Instant, + pub(crate) expires_at: Instant, + /// The only way a verdict for this request can arrive. + pub(crate) verdict_rx: oneshot::Receiver, +} + +/// A parked request taken out of waiting, with what is needed to refuse it. +/// +/// Its receiver is already gone by the time anyone holds one of these. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct RetiredAsk { + pub(crate) rpc_id: serde_json::Value, + pub(crate) park_id: ParkId, + pub(crate) digest: RequestDigest, + pub(crate) reject_option_id: String, + pub(crate) tool: String, + pub(crate) parked_at: Instant, +} + +/// Why a request could not be parked. +/// +/// Carries no payload. Handing the rejected ask back would make the `Result` +/// large enough to trip `clippy::result_large_err`, and the caller has no use +/// for it beyond refusing the request. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ParkRejected { + /// The id is not owed an answer, so there is nothing to park. + NotOwed, + /// The id is already parked. One request waits on one verdict. + AlreadyParked, +} + +/// One request owed an answer, and the approver it is waiting on, if any. +#[derive(Debug)] +struct OwedAnswer { + rpc_id: serde_json::Value, + parked: Option, +} /// Ids of `session/request_permission` requests awaiting a response. /// /// `serde_json::Value` because JSON-RPC 2.0 permits both numeric and string /// ids. Order is preserved: requests are answered oldest first, so a cancel /// resolves them in the order the agent asked. -#[derive(Debug, Default, Clone, PartialEq)] +/// +/// Neither `Clone` nor `PartialEq`: a parked entry owns a oneshot receiver, +/// which has neither, and a copy of a receiver would be a second way in. +#[derive(Debug, Default)] pub(crate) struct PendingPermissions { - ids: Vec, + entries: Vec, } impl PendingPermissions { @@ -41,47 +123,188 @@ impl PendingPermissions { /// Idempotent: an id already present is not duplicated, so a retransmitted /// request cannot cause two responses to be written for it. pub(crate) fn record(&mut self, id: serde_json::Value) { - if !self.ids.contains(&id) { - self.ids.push(id); + if !self.contains(&id) { + self.entries.push(OwedAnswer { + rpc_id: id, + parked: None, + }); } } + /// Whether `id` is owed an answer, parked or not. + /// + /// The permission handler checks this first, so a retransmit of a request + /// it still owes is ignored rather than answered twice or parked twice. + pub(crate) fn contains(&self, id: &serde_json::Value) -> bool { + self.entries.iter().any(|entry| entry.rpc_id == *id) + } + /// Forget one request, once its response is on the wire. /// /// Call this *after* a successful write, never before. Forgetting first /// means a failed write leaves this saying "answered" when nothing was /// sent, and the cancel path then skips the request — the deadlock this /// ordering exists to prevent. Siblings keep their own claim on an answer. + /// + /// Forgetting a parked request drops its verdict receiver, so the main + /// loop's `send` fails from then on. pub(crate) fn forget(&mut self, id: &serde_json::Value) { - self.ids.retain(|pending| pending != id); + self.entries.retain(|entry| entry.rpc_id != *id); } - /// The oldest request still owed an answer. + /// The oldest request still owed an answer, parked or not. pub(crate) fn next_owed(&self) -> Option<&serde_json::Value> { - self.ids.first() + self.entries.first().map(|entry| &entry.rpc_id) + } + + /// Park a request that is owed an answer. + /// + /// The id must already be recorded and not yet parked. It stays owed: + /// parking changes how it will be answered, never whether. + pub(crate) fn park( + &mut self, + id: &serde_json::Value, + ask: ParkedAsk, + ) -> Result<(), ParkRejected> { + match self.entries.iter_mut().find(|entry| entry.rpc_id == *id) { + Some(entry) if entry.parked.is_none() => { + entry.parked = Some(ask); + Ok(()) + } + Some(_) => Err(ParkRejected::AlreadyParked), + None => Err(ParkRejected::NotOwed), + } + } + + /// How many requests are waiting on an approver. + /// + /// The per-connection cap counts these rather than every owed id: an id + /// that is not parked is answered within the same loop iteration. + pub(crate) fn parked_count(&self) -> usize { + self.entries + .iter() + .filter(|entry| entry.parked.is_some()) + .count() + } + + /// The soonest any parked request expires, for the read loop's next + /// deadline. + pub(crate) fn earliest_expiry(&self) -> Option { + self.entries + .iter() + .filter_map(|entry| entry.parked.as_ref().map(|ask| ask.expires_at)) + .min() + } + + /// The parked request that expired first, if any has expired by `now`. + pub(crate) fn first_expired(&self, now: Instant) -> Option { + self.entries + .iter() + .filter_map(|entry| entry.parked.as_ref()) + .filter(|ask| ask.expires_at <= now) + .min_by_key(|ask| ask.expires_at) + .map(|ask| ask.park_id) + } + + /// Stop waiting on one parked request and hand back what is needed to + /// answer it. + /// + /// The id stays owed until its answer is written and [`forget`] is called: + /// the same write-then-forget order as every other answer, so a failed + /// write leaves it for `cancel_with_cleanup`. + /// + /// [`forget`]: Self::forget + pub(crate) fn take_parked( + &mut self, + park_id: ParkId, + ) -> Option<(serde_json::Value, ParkedAsk)> { + let entry = self.entries.iter_mut().find(|entry| { + entry + .parked + .as_ref() + .is_some_and(|ask| ask.park_id == park_id) + })?; + let ask = entry.parked.take()?; + Some((entry.rpc_id.clone(), ask)) + } + + /// Stop waiting on every parked request at once. + /// + /// Every receiver is dropped inside this call, before the caller writes + /// anything, so the main loop's `send` fails from here on even if the + /// writes that follow never happen. A receiver left alive across those + /// writes would let the main loop believe it had delivered a verdict to a + /// request that is about to be refused. The ids stay owed. + pub(crate) fn retire_all_parked(&mut self) -> Vec { + self.entries + .iter_mut() + .filter_map(|entry| { + let ask = entry.parked.take()?; + Some(RetiredAsk { + rpc_id: entry.rpc_id.clone(), + park_id: ask.park_id, + digest: ask.digest, + reject_option_id: ask.reject_option_id, + tool: ask.tool, + parked_at: ask.parked_at, + }) + }) + .collect() + } + + /// Resolves with the first parked request whose verdict channel is ready: + /// a verdict, or `Err` when the sender was dropped. + /// + /// Pending forever when nothing is parked. Lazy, because `tokio::select!` + /// builds the future for a disabled branch too, and building it must not + /// touch any receiver. + /// + /// The caller MUST `take_parked` the returned id before polling again: a + /// tokio oneshot receiver panics if it is polled after it has completed. + pub(crate) fn next_verdict( + &mut self, + ) -> impl Future)> + '_ + { + std::future::poll_fn(move |cx| { + for entry in self.entries.iter_mut() { + if let Some(ask) = entry.parked.as_mut() { + if let Poll::Ready(result) = Pin::new(&mut ask.verdict_rx).poll(cx) { + return Poll::Ready((ask.park_id, result)); + } + } + } + Poll::Pending + }) } /// How many requests are still owed an answer. /// - /// Test-only for now. The approval leg will read it for the concurrency - /// cap; gated rather than `allow(dead_code)` so the day it has a - /// production caller is the day the gate comes off. + /// Test-only. The per-connection cap counts parked requests + /// ([`parked_count`](Self::parked_count)), not every owed id; gated rather + /// than `allow(dead_code)` so the day it has a production caller is the + /// day the gate comes off. #[cfg(test)] pub(crate) fn len(&self) -> usize { - self.ids.len() + self.entries.len() } /// Whether every request received has been answered. #[cfg(test)] pub(crate) fn is_empty(&self) -> bool { - self.ids.is_empty() + self.entries.is_empty() } } #[cfg(test)] mod tests { - use super::PendingPermissions; + use super::{ParkedAsk, PendingPermissions}; + use crate::approval_gate::{ + AllowVia, ApprovalDecision, ApprovalVerdict, ParkId, RequestDigest, + }; use serde_json::json; + use std::time::Duration; + use tokio::sync::oneshot; + use tokio::time::Instant; #[test] fn a_recorded_request_is_owed_until_it_is_forgotten() { @@ -174,4 +397,164 @@ mod tests { } assert!(pending.is_empty()); } + + // ── Parked requests ────────────────────────────────────────────────────── + + fn ask(expires_in: u64) -> (ParkedAsk, oneshot::Sender) { + let (tx, rx) = oneshot::channel(); + let now = Instant::now(); + ( + ParkedAsk { + park_id: ParkId::mint(), + allow_option_id: "allow".into(), + reject_option_id: "reject".into(), + digest: RequestDigest([7; 32]), + tool: "Bash".into(), + parked_at: now, + expires_at: now + Duration::from_secs(expires_in), + verdict_rx: rx, + }, + tx, + ) + } + + /// The dropped-future exit: `cancel_with_cleanup` answers whatever + /// `next_owed` returns, so a parked request must be among them. + #[test] + fn a_parked_request_is_still_owed_so_cancel_answers_it() { + let mut pending = PendingPermissions::default(); + pending.record(json!(1)); + let (a, _tx) = ask(600); + pending.park(&json!(1), a).unwrap(); + assert_eq!(pending.next_owed(), Some(&json!(1))); + assert_eq!(pending.parked_count(), 1); + assert!(pending.contains(&json!(1))); + } + + #[test] + fn park_requires_a_recorded_unparked_id() { + let mut pending = PendingPermissions::default(); + let (a, _tx) = ask(600); + assert!( + pending.park(&json!(1), a).is_err(), + "an id nobody owes cannot park" + ); + pending.record(json!(1)); + let (a, _tx1) = ask(600); + pending.park(&json!(1), a).unwrap(); + let (b, _tx2) = ask(600); + assert!( + pending.park(&json!(1), b).is_err(), + "never park one id twice" + ); + } + + /// Parked requests keep the write-then-forget order: a failed write after + /// taking the parked state still leaves the id for cancel. + #[test] + fn taking_a_parked_state_leaves_the_id_owed_until_forgotten() { + let mut pending = PendingPermissions::default(); + pending.record(json!(1)); + let (a, _tx) = ask(600); + let id = a.park_id; + pending.park(&json!(1), a).unwrap(); + let (rpc, _) = pending.take_parked(id).unwrap(); + assert_eq!(rpc, json!(1)); + assert_eq!(pending.parked_count(), 0); + assert_eq!(pending.next_owed(), Some(&json!(1))); + pending.forget(&json!(1)); + assert!(pending.next_owed().is_none()); + } + + /// The exit drain retires before it writes, so the main loop sees every + /// ask die even when the writes to the agent then fail. + #[test] + fn retiring_drops_every_receiver_and_keeps_ids_owed() { + let mut pending = PendingPermissions::default(); + let mut txs = Vec::new(); + for id in 1..=2 { + pending.record(json!(id)); + let (a, tx) = ask(600); + pending.park(&json!(id), a).unwrap(); + txs.push(tx); + } + let retired = pending.retire_all_parked(); + assert_eq!(retired.len(), 2); + assert!( + txs.iter().all(|tx| tx.is_closed()), + "the main loop must see the asks die" + ); + assert_eq!(pending.next_owed(), Some(&json!(1))); + } + + #[test] + fn forgetting_a_parked_request_closes_its_verdict_channel() { + let mut pending = PendingPermissions::default(); + pending.record(json!(1)); + let (a, tx) = ask(600); + pending.park(&json!(1), a).unwrap(); + pending.forget(&json!(1)); + assert!( + tx.is_closed(), + "a forgotten request's verdict must have nowhere to land" + ); + } + + #[tokio::test(start_paused = true)] + async fn expiry_queries() { + let mut pending = PendingPermissions::default(); + pending.record(json!(1)); + pending.record(json!(2)); + let (a, _t1) = ask(600); + let (b, _t2) = ask(300); + let b_id = b.park_id; + pending.park(&json!(1), a).unwrap(); + pending.park(&json!(2), b).unwrap(); + let now = Instant::now(); + assert_eq!( + pending.earliest_expiry(), + Some(now + Duration::from_secs(300)) + ); + assert_eq!(pending.first_expired(now), None); + assert_eq!( + pending.first_expired(now + Duration::from_secs(300)), + Some(b_id) + ); + } + + #[tokio::test(start_paused = true)] + async fn next_verdict_returns_the_ready_one_and_reports_a_dropped_sender() { + let mut pending = PendingPermissions::default(); + pending.record(json!(1)); + pending.record(json!(2)); + let (a, tx_a) = ask(600); + let (b, tx_b) = ask(600); + let (a_id, b_id) = (a.park_id, b.park_id); + pending.park(&json!(1), a).unwrap(); + pending.park(&json!(2), b).unwrap(); + tx_b.send(ApprovalVerdict { + digest: RequestDigest([7; 32]), + decision: ApprovalDecision::Allow { + via: AllowVia::Approver, + }, + }) + .unwrap(); + let (id, res) = pending.next_verdict().await; + assert_eq!(id, b_id); + assert!(res.is_ok()); + pending.take_parked(id).unwrap(); + drop(tx_a); + let (id, res) = pending.next_verdict().await; + assert_eq!(id, a_id); + assert!( + res.is_err(), + "a dropped sender is a closed channel, never a verdict" + ); + pending.take_parked(id).unwrap(); + let idle = tokio::time::timeout(Duration::from_secs(1), pending.next_verdict()).await; + assert!( + idle.is_err(), + "nothing parked: pending, and no poll-after-complete panic" + ); + } } diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 005eee8ce8c..11a02b1939a 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -2514,6 +2514,7 @@ pub async fn run_prompt_task( // its own standing context, with no inbound request behind // it. Nobody is being answered here, so there is no guest. crate::acp::TurnAuthority::Owner, + None, ) .await; @@ -2843,6 +2844,7 @@ pub async fn run_prompt_task( ctx.idle_timeout, ctx.max_turn_duration, turn_authority, + None, ) => result, } } @@ -2855,6 +2857,7 @@ pub async fn run_prompt_task( ctx.idle_timeout, ctx.max_turn_duration, turn_authority, + None, ) => result, mode = rx => { let control_signal = mode.unwrap_or(ControlSignal::Cancel); From d083507e4dc23196f776e0317a5172a57de2e4a7 Mon Sep 17 00:00:00 2001 From: QuicksilverSlick Date: Fri, 11 Sep 2026 10:57:49 -0500 Subject: [PATCH 2/2] fix(buzz-acp): report a prompt error read just before a control signal With the slice A review fix, the prompt stops counting as in flight as soon as its response is read, and that includes a JSON-RPC error. If a steer, interrupt or stop then drops the prompt future during its exit drain, the pool finds no prompt in flight and takes its "completed before the control signal" branch. That branch reported every such turn as a success: the batch was recorded as delivered and never retried. The client now keeps the error it read until the prompt call returns, and that branch takes it. A failed turn is reported as the error it was, with no delivery record and an Error metric. Its batch follows the signal, as a cancel's would: a steer or interrupt re-prompts it, and a stop or rotate drops it. The copy is cleared when the prompt call returns and again when the next prompt starts, so it can never be reported against a later turn. Each test fails with its part of the fix undone: - a_prompt_error_read_before_a_drop_is_kept_for_the_pool - an_error_the_prompt_returned_is_not_kept_for_the_pool - test_classify_completed_before_control_signal_reports_a_read_error Co-Authored-By: Claude Opus 5 --- crates/buzz-acp/src/acp.rs | 99 +++++++++++++++++++++- crates/buzz-acp/src/pool.rs | 165 ++++++++++++++++++++++++++++++++---- 2 files changed, 246 insertions(+), 18 deletions(-) diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 75bd8bb3a32..84bd1d48b12 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -204,6 +204,17 @@ pub struct AcpClient { /// Used by [`cancel_with_cleanup`] to drain the correct response. /// Set in [`session_prompt_with_idle_timeout`]; consumed in [`cancel_with_cleanup`]. last_prompt_id: Option, + /// The JSON-RPC error that answered the prompt, kept only while the pool + /// may still need it. + /// + /// The prompt's result normally carries the error back to the pool. But + /// the pool can drop the prompt future in its exit drain, after the error + /// was read. It then finds no prompt in flight and takes its "completed + /// before the control signal" branch, which would otherwise report the + /// turn as a success: its batch recorded as delivered and never retried. + /// That branch takes this instead. Cleared when the prompt call returns, + /// and again when the next prompt starts. + prompt_error_before_drain: Option<(i64, String)>, /// Hard deadline for the current turn, set by `session_prompt_with_idle_timeout`. /// Inherited by `cancel_with_cleanup` so the drain loop shares the same budget /// rather than starting a fresh timer (prevents double-jeopardy). @@ -660,6 +671,7 @@ impl AcpClient { current_turn_authority: TurnAuthority::Guest, permission_refusals: Vec::new(), last_prompt_id: None, + prompt_error_before_drain: None, current_hard_deadline: None, observer: None, observer_agent_index: None, @@ -919,6 +931,9 @@ impl AcpClient { self.goose_usage.begin_turn(session_id); self.standard_usage.begin_turn(session_id); + // Before the first yield, like `last_prompt_id`: an error kept from an + // earlier turn must never be reported as this one's. + self.prompt_error_before_drain = None; self.last_prompt_id = Some(self.next_id); let id = self.next_id; self.next_id += 1; @@ -970,6 +985,9 @@ impl AcpClient { // be decided with its authority again. The next prompt sets its own, // and a cancel drain that follows a timeout decides as a guest. self.current_turn_authority = TurnAuthority::Guest; + // The result carries any error back itself; the copy was only for a + // pool that dropped this future before it could return. + self.prompt_error_before_drain = None; // On timeout errors, leave current_hard_deadline set so cancel_with_cleanup // can inherit the remaining budget. Clear it on all other outcomes. @@ -1010,6 +1028,16 @@ impl AcpClient { self.last_prompt_id.is_some() } + /// Take the JSON-RPC error that answered the prompt, if the pool dropped + /// the prompt future after it was read. For the pool's "completed before + /// the control signal" branch only; everywhere else the prompt's own + /// result carries the error. + pub(crate) fn take_prompt_error_before_drain(&mut self) -> Option { + self.prompt_error_before_drain + .take() + .map(|(code, message)| AcpError::AgentError { code, message }) + } + /// Most recently observed goose `_meta.goose.activeRunId` from a /// `session_info_update`, if any. /// @@ -1564,11 +1592,15 @@ impl AcpClient { // Cleared here, a drop mid-drain takes the pool's "turn completed // before the control signal" branch, which runs no cleanup. Whatever // the drain had not answered stays owed, and the next prompt answers - // it before it is sent. + // it before it is sent. An error is kept for that branch, so a turn + // that failed is not reported as one that completed. if matches!(result, Ok(_) | Err(AcpError::AgentError { .. })) && self.last_prompt_id == Some(expected_id) { self.last_prompt_id = None; + if let Err(AcpError::AgentError { code, message }) = &result { + self.prompt_error_before_drain = Some((*code, message.clone())); + } } self.answer_owed_on_exit(gate.as_ref()).await; result @@ -4004,6 +4036,71 @@ mod tests { stalled.shutdown().await; } + /// The pool drops the prompt future before it can return its error, so the + /// error has to wait on the client for the pool to take. Were it lost, the + /// pool would report a failed turn as completed: its batch recorded as + /// delivered and never retried. + #[tokio::test] + async fn a_prompt_error_read_before_a_drop_is_kept_for_the_pool() { + let error = serde_json::json!({ + "jsonrpc": "2.0", + "id": PROMPT_ID, + "error": {"code": -32000, "message": "the agent failed the turn"}, + }); + let (mut client, mut stalled) = drop_in_the_exit_drain(error).await; + match client.take_prompt_error_before_drain() { + Some(AcpError::AgentError { code, message }) => { + assert_eq!(code, -32000); + assert_eq!(message, "the agent failed the turn"); + } + other => panic!("the pool must get the error the turn ended with, got {other:?}"), + } + assert!( + client.take_prompt_error_before_drain().is_none(), + "the error is reported once" + ); + stalled.shutdown().await; + } + + /// When the prompt call returns, its result already carries the error. A + /// copy left behind could be reported against a later turn. + #[tokio::test] + async fn an_error_the_prompt_returned_is_not_kept_for_the_pool() { + // Two cats, as in the owed-ids test below: the queued error answers the + // prompt as soon as it is sent. A fresh client's first request is id 0. + let mut client = spawn_inert_client().await; + let mut wire = spawn_inert_client().await; + let error = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "error": {"code": -32000, "message": "the agent failed the turn"}, + }); + client + .write_ndjson(&error) + .await + .expect("cat takes the response"); + std::mem::swap(&mut client.stdin, &mut wire.stdin); + + let result = client + .session_prompt_with_idle_timeout( + PARK_SESSION, + "hello", + std::time::Duration::from_secs(5), + std::time::Duration::from_secs(10), + TurnAuthority::Guest, + None, + ) + .await; + assert!( + matches!(result, Err(AcpError::AgentError { .. })), + "got {result:?}" + ); + assert!( + client.take_prompt_error_before_drain().is_none(), + "the returned result carried the error, so nothing is kept" + ); + } + /// The other side of the two above: a prompt that timed out still has its /// response coming, so it stays in flight for the pool's cancel to drain. #[tokio::test] diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 11a02b1939a..b0801b53894 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -2948,20 +2948,29 @@ pub async fn run_prompt_task( } } } else { - // Race 1 resolution: turn completed naturally before cancel - // could fire. last_prompt_id is None — cleared by - // session_prompt_with_idle_timeout() on success. The prompt - // future was dropped by select! — its Ok result is gone. + // Race 1 resolution: the read loop consumed the prompt's + // response before cancel could fire. last_prompt_id is + // None, cleared once the result, or a JSON-RPC error, was + // read. The prompt future was dropped by select!, so its + // result is gone; an error waits on the client instead + // (take_prompt_error_before_drain), so a failed turn is + // not reported as a completed one. // // Note: this `else` branch (last_prompt_id is None) cannot // fire during the pre-prompt phase because `biased` select! // polls the prompt arm first. That arm sets last_prompt_id // synchronously before its first yield point, so by the time // the cancel arm can win, last_prompt_id is already Some. - // This branch only fires when the turn genuinely completed - // and last_prompt_id was cleared by the success path. + // This branch only fires once the prompt's response was + // read and last_prompt_id cleared. // // MUST send a PromptResult or the main loop deadlocks. + let completed = classify_completed_before_control_signal( + &ctx, + agent.acp.take_prompt_error_before_drain(), + control_signal.clone(), + batch, + ); if matches!( control_signal, ControlSignal::Rotate | ControlSignal::SwitchModel { .. } @@ -2976,15 +2985,24 @@ pub async fn run_prompt_task( "control signal arrived but turn already completed — treating as success" ); } - log_stop_reason(&source, &StopReason::EndTurn); - if let PromptSource::Channel(scope) = &source { - let standing_sent = !agent.has_system_prompt_support(); - record_scope_delivery_success( - &mut agent, - scope.clone(), - standing_sent, - &pending_delivered_event_ids, + if let PromptOutcome::Error(error) = &completed.outcome { + tracing::error!( + target: "pool::prompt", + "session_prompt error, read just before the control signal: {error}" ); + } else { + log_stop_reason(&source, &StopReason::EndTurn); + } + if completed.delivered { + if let PromptSource::Channel(scope) = &source { + let standing_sent = !agent.has_system_prompt_support(); + record_scope_delivery_success( + &mut agent, + scope.clone(), + standing_sent, + &pending_delivered_event_ids, + ); + } } apply_completed_before_control_signal( &mut agent.state, @@ -2998,7 +3016,7 @@ pub async fn run_prompt_task( observer_channel_id, &session_id, &turn_id, - Some(buzz_core::agent_turn_metric::StopReason::EndTurn), + Some(completed.metric), ) .await; send_prompt_result( @@ -3006,8 +3024,8 @@ pub async fn run_prompt_task( &turn_id, agent, source, - PromptOutcome::Ok(StopReason::EndTurn), - None, // turn succeeded — batch was processed, no requeue + completed.outcome, + completed.retry_batch, ); return; } @@ -4555,6 +4573,49 @@ fn classify_control_cancel_failure( } } +/// What the Race 1 branch of [`run_prompt_task`] reports: a control signal +/// won the `select!` after the read loop had consumed the prompt's response, +/// so the prompt future, and the result it held, were dropped. +struct CompletedBeforeControlSignal { + outcome: PromptOutcome, + retry_batch: Option, + metric: buzz_core::agent_turn_metric::StopReason, + /// Only a turn that completed has delivered its batch. + delivered: bool, +} + +/// Classify a turn the read loop had finished when a control signal dropped +/// its prompt future. `prompt_error` is the JSON-RPC error that answered the +/// prompt, if one did ([`AcpClient::take_prompt_error_before_drain`]). +/// +/// A turn that completed is reported as completed, whatever the signal: its +/// batch was processed. A turn that failed is reported as the error it was, +/// like the undropped `Err` arm, with no delivery record. Its batch's fate +/// follows the signal, as a cancel's would: a steer or interrupt re-prompts +/// it, a stop or rotate drops it. +fn classify_completed_before_control_signal( + ctx: &PromptContext, + prompt_error: Option, + signal: ControlSignal, + batch: Option, +) -> CompletedBeforeControlSignal { + match prompt_error { + None => CompletedBeforeControlSignal { + outcome: PromptOutcome::Ok(StopReason::EndTurn), + // The turn succeeded: the batch was processed, so no requeue. + retry_batch: None, + metric: buzz_core::agent_turn_metric::StopReason::EndTurn, + delivered: true, + }, + Some(error) => CompletedBeforeControlSignal { + outcome: PromptOutcome::Error(error), + retry_batch: requeue_cancelled_batch(ctx, signal, batch), + metric: buzz_core::agent_turn_metric::StopReason::Error, + delivered: false, + }, + } +} + /// How a turn's source is named in the `pool::prompt` log lines. /// /// Shared by the turn-start and turn-stop lines so a log can be read as pairs. @@ -7840,6 +7901,76 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" ); } + #[test] + fn test_classify_completed_before_control_signal_reports_a_read_error() { + let ctx = { + let mut ctx = make_prompt_context_no_owner(); + ctx.dedup_mode = DedupMode::Queue; + ctx + }; + + // Completed: a success, not requeued, delivered, whatever the signal. + for signal in [ + ControlSignal::Steer, + ControlSignal::Cancel, + ControlSignal::Rotate, + ] { + let completed = classify_completed_before_control_signal( + &ctx, + None, + signal.clone(), + Some(one_event_batch(Uuid::new_v4())), + ); + assert_outcome_matches(&completed.outcome, "Ok"); + assert!( + completed.retry_batch.is_none(), + "{signal:?}: a completed batch is not requeued" + ); + assert!( + completed.delivered, + "{signal:?}: a completed batch was delivered" + ); + assert_eq!( + completed.metric, + buzz_core::agent_turn_metric::StopReason::EndTurn + ); + } + + // Errored: reported as the error and never delivered. The batch + // follows the signal, as a cancel's would. + let cases = [ + (ControlSignal::Steer, Some(CancelReason::Steer)), + (ControlSignal::Interrupt, Some(CancelReason::Interrupt)), + (ControlSignal::Cancel, None), + (ControlSignal::Rotate, None), + ]; + for (signal, expected_reason) in cases { + let completed = classify_completed_before_control_signal( + &ctx, + Some(AcpError::AgentError { + code: -32000, + message: "the agent failed the turn".into(), + }), + signal.clone(), + Some(one_event_batch(Uuid::new_v4())), + ); + assert_outcome_matches(&completed.outcome, "Error"); + assert!( + !completed.delivered, + "{signal:?}: a failed turn delivered nothing" + ); + assert_eq!( + completed.metric, + buzz_core::agent_turn_metric::StopReason::Error + ); + assert_eq!( + completed.retry_batch.map(|batch| batch.cancel_reason), + expected_reason.map(Some), + "{signal:?}: batch fate" + ); + } + } + #[test] fn test_classify_control_cancel_failure_crosses_error_outcome_and_batch_fate() { let ctx = {