diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 0799b173be8..84bd1d48b12 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. /// @@ -192,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). @@ -501,6 +524,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 { @@ -644,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, @@ -850,6 +878,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 +886,7 @@ impl AcpClient { idle_timeout, max_duration, authority, + approvals, ) .await } @@ -874,12 +904,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. @@ -894,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; @@ -905,22 +945,49 @@ 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; + // 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. @@ -961,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. /// @@ -1048,8 +1125,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 +1202,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 +1219,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 +1237,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 +1373,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 +1467,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 +1511,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 +1525,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 +1535,87 @@ 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. 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 + } + + /// 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 +1638,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 +1681,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 +1839,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 +1916,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 +1988,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 +2052,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 +2061,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 +2283,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 +2292,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 +2320,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 +2346,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 +2548,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 +2886,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 +2937,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 +3396,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,45 +3458,1199 @@ mod tests { } } - use super::*; + // ── 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_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)); + 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_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_tool_call() -> serde_json::Value { + probe_request_with_id(7)["params"]["toolCall"].clone() } - #[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) - ); + /// 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 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] + 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) @@ -5004,6 +6835,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..b0801b53894 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); @@ -2945,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 { .. } @@ -2973,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, @@ -2995,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( @@ -3003,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; } @@ -4552,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. @@ -7837,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 = {