diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index cd049830688..baea6f48c66 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -9,7 +9,6 @@ //! 5. [`AcpClient::session_cancel`] / [`AcpClient::cancel_with_cleanup`] — cancel in-flight turn use futures_util::StreamExt; -use tokio::io::AsyncWriteExt; use tokio::process::{Child, ChildStdin, ChildStdout}; use tokio_util::codec::{FramedRead, LinesCodec, LinesCodecError}; @@ -18,6 +17,9 @@ use crate::usage::{ PromptResponseUsage, StandardAdapterKind, StandardUsageTracker, TurnUsage, UsageTracker, }; +#[path = "acp_frame_writer.rs"] +mod frame_writer; + /// Maximum allowed size of a single NDJSON line from the agent's stdout. /// Lines exceeding this limit are rejected to prevent OOM from rogue agents. const MAX_LINE_SIZE: usize = 10_000_000; // 10 MB @@ -144,8 +146,8 @@ fn build_initialize_params() -> serde_json::Value { pub struct AcpClient { /// The agent child process (kept alive to prevent zombie). child: Child, - /// Write end of the agent's stdin pipe. - stdin: ChildStdin, + /// Sole stdin writer; None after an interrupted/failed frame closes it. + stdin: Option, /// Framed reader over the agent's stdout pipe (line-oriented, bounded). /// Uses `LinesCodec::new_with_max_length` to enforce MAX_LINE_SIZE at the /// read level — prevents OOM from rogue agents writing infinite non-newline bytes. @@ -559,7 +561,7 @@ impl AcpClient { Ok(Self { child, - stdin, + stdin: Some(stdin), reader: FramedRead::new(stdout, LinesCodec::new_with_max_length(MAX_LINE_SIZE)), next_id: 0, pending_permission_id: None, @@ -1085,12 +1087,10 @@ impl AcpClient { async fn write_ndjson(&mut self, value: &serde_json::Value) -> Result<(), AcpError> { const WRITE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); let line = serde_json::to_string(value)?; - tokio::time::timeout(WRITE_TIMEOUT, async { - self.stdin.write_all(line.as_bytes()).await?; - self.stdin.write_all(b"\n").await?; - self.stdin.flush().await?; - Ok::<(), std::io::Error>(()) - }) + tokio::time::timeout( + WRITE_TIMEOUT, + frame_writer::write_frame(&mut self.stdin, line.as_bytes()), + ) .await .map_err(|_| AcpError::WriteTimeout(WRITE_TIMEOUT))? .map_err(AcpError::Io)?; diff --git a/crates/buzz-acp/src/acp_frame_writer.rs b/crates/buzz-acp/src/acp_frame_writer.rs new file mode 100644 index 00000000000..27f15977589 --- /dev/null +++ b/crates/buzz-acp/src/acp_frame_writer.rs @@ -0,0 +1,233 @@ +//! Cancellation-safe ownership of the sole ACP stdin writer. + +use tokio::io::{AsyncWrite, AsyncWriteExt}; + +/// Lend the writer back only after a complete frame has been written. Dropping +/// this future (control select, outer deadline, or task abort) or returning an +/// I/O error drops the owned writer instead. For ChildStdin this closes the pipe; +/// the empty slot also rejects every subsequent request/cleanup write as a +/// transport error, using the existing pool retirement and retry policy. +pub(super) async fn write_frame( + slot: &mut Option, + body: &[u8], +) -> std::io::Result<()> { + let mut writer = slot.take().ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "ACP stdin closed by an incomplete or failed frame write", + ) + })?; + writer.write_all(body).await?; + writer.write_all(b"\n").await?; + writer.flush().await?; + *slot = Some(writer); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + use tokio::io::AsyncReadExt; + + // A bounded duplex is a deterministic AsyncWrite seam: capacity pins the + // exact body/LF suspension point; the production ChildStdin test below + // exercises the real transport and prompt/cleanup ownership as well. + async fn aborted_frame(capacity: usize) { + let (writer, mut reader) = tokio::io::duplex(capacity); + let mut slot = Some(writer); + assert!( + tokio::time::timeout(Duration::from_millis(20), write_frame(&mut slot, b"{}")) + .await + .is_err() + ); + assert!(slot.is_none()); + assert_eq!( + write_frame(&mut slot, b"cancel").await.unwrap_err().kind(), + std::io::ErrorKind::BrokenPipe + ); + let mut bytes = Vec::new(); + reader.read_to_end(&mut bytes).await.unwrap(); + assert_eq!(bytes, b"{}"[..capacity]); + } + + #[tokio::test] + async fn aborted_body_closes_before_reuse() { + aborted_frame(1).await; + } + + #[tokio::test] + async fn aborted_lf_closes_before_reuse() { + aborted_frame(2).await; + } + + #[tokio::test] + async fn io_failure_closes_before_reuse() { + let (writer, reader) = tokio::io::duplex(4); + let mut slot = Some(writer); + drop(reader); + assert!(write_frame(&mut slot, b"{}").await.is_err()); + assert!(slot.is_none()); + assert!(write_frame(&mut slot, b"cancel").await.is_err()); + } + + #[tokio::test] + async fn io_failure_after_prefix_closes_before_reuse() { + let (writer, mut reader) = tokio::io::duplex(1); + let mut slot = Some(writer); + let (result, ()) = tokio::join!(write_frame(&mut slot, b"{}"), async move { + assert_eq!(reader.read_u8().await.unwrap(), b'{'); + drop(reader); + }); + assert!(result.is_err()); + assert!(slot.is_none()); + assert!(write_frame(&mut slot, b"cancel").await.is_err()); + } + + #[tokio::test] + async fn completed_frames_keep_writer() { + let (writer, mut reader) = tokio::io::duplex(32); + let mut slot = Some(writer); + write_frame(&mut slot, b"{}").await.unwrap(); + write_frame(&mut slot, b"{\"cancel\":true}").await.unwrap(); + drop(slot); + let mut bytes = Vec::new(); + reader.read_to_end(&mut bytes).await.unwrap(); + assert_eq!(bytes, b"{}\n{\"cancel\":true}\n"); + } + #[cfg(unix)] + async fn wait_for(path: &std::path::Path) { + tokio::time::timeout(Duration::from_secs(5), async { + while !path.exists() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("child fixture did not reach checkpoint"); + } + + #[cfg(unix)] + #[tokio::test] + async fn real_pipe_aborted_prompt_rejects_cleanup_and_later_requests() { + use super::super::{AcpClient, AcpError}; + let dir = std::env::temp_dir().join(format!("buzz-frame-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).unwrap(); + let release = dir.join("release"); + let capture = dir.join("capture"); + let done = dir.join("done"); + let script = format!( + "while [ ! -e '{}' ]; do sleep 0.01; done; cat > '{}'; touch '{}'; sleep 10", + release.display(), + capture.display(), + done.display() + ); + let mut client = AcpClient::spawn("bash", &["-c".into(), script], &[], false) + .await + .unwrap(); + let prompt = "x".repeat(300_000); + // Recreates the pool's control-select ownership, not its relay loop. + // The child does not read until AFTER cleanup attempts have returned. + tokio::select! { + biased; + result = client.session_prompt_with_idle_timeout( + "sess-test", &prompt, Duration::from_secs(60), Duration::from_secs(60) + ) => panic!("large write should remain blocked: {result:?}"), + _ = tokio::time::sleep(Duration::from_millis(100)) => {} + } + assert!(client.has_in_flight_prompt()); + assert!( + matches!(client.cancel_with_cleanup_grace("sess-test", Duration::from_secs(5)).await, + Err(AcpError::Io(ref e)) if e.kind() == std::io::ErrorKind::BrokenPipe) + ); + assert!(matches!( + client.session_cancel("sess-test").await, + Err(AcpError::Io(_)) + )); + assert!(matches!(client.initialize().await, Err(AcpError::Io(_)))); + std::fs::write(&release, []).unwrap(); + wait_for(&done).await; + let bytes = std::fs::read(&capture).unwrap(); + assert!(!bytes.is_empty(), "real pipe must contain a written prefix"); + assert!(bytes.len() < prompt.len()); + assert!(!bytes.contains(&b'\n'), "no cleanup frame may be appended"); + client.shutdown().await; + std::fs::remove_dir_all(dir).unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn real_pipe_completed_prompt_still_allows_normal_cancel() { + use super::super::AcpClient; + let dir = std::env::temp_dir().join(format!("buzz-frame-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).unwrap(); + let capture = dir.join("capture"); + let ready = dir.join("ready"); + let script = format!( + "read -r line; printf '%s\\n' \"$line\" > '{}'; touch '{}'; \ + read -r line; printf '%s\\n' \"$line\" >> '{}'; \ + printf '%s\\n' '{{\"jsonrpc\":\"2.0\",\"id\":0,\"result\":{{\"stopReason\":\"cancelled\"}}}}'; sleep 10", + capture.display(), + ready.display(), + capture.display() + ); + let mut client = AcpClient::spawn("bash", &["-c".into(), script], &[], false) + .await + .unwrap(); + tokio::select! { + biased; + result = client.session_prompt_with_idle_timeout( + "sess-test", "hello", Duration::from_secs(60), Duration::from_secs(60) + ) => panic!("response should wait for cancel: {result:?}"), + _ = wait_for(&ready) => {} + } + assert!(client + .cancel_with_cleanup_grace("sess-test", Duration::from_secs(5)) + .await + .is_ok()); + let text = std::fs::read_to_string(&capture).unwrap(); + let frames: Vec = text + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + assert_eq!(frames.len(), 2); + assert_eq!(frames[0]["method"], "session/prompt"); + assert_eq!(frames[1]["method"], "session/cancel"); + assert!(client.stdin.is_some()); + client.shutdown().await; + std::fs::remove_dir_all(dir).unwrap(); + } + #[cfg(unix)] + #[tokio::test] + async fn application_error_does_not_close_transport() { + use super::super::{AcpClient, AcpError}; + let script = r#"read -r line +printf '%s\n' '{"jsonrpc":"2.0","id":0,"error":{"code":-32602,"message":"bad input"}}' +read -r line +printf '%s\n' '{"jsonrpc":"2.0","id":1,"result":{"stopReason":"end_turn"}}' +sleep 10"#; + let mut client = AcpClient::spawn("bash", &["-c".into(), script.into()], &[], false) + .await + .unwrap(); + assert!(matches!( + client + .session_prompt_with_idle_timeout( + "sess-test", + "first", + Duration::from_secs(5), + Duration::from_secs(5) + ) + .await, + Err(AcpError::AgentError { .. }) + )); + assert!(client + .session_prompt_with_idle_timeout( + "sess-test", + "second", + Duration::from_secs(5), + Duration::from_secs(5) + ) + .await + .is_ok()); + client.shutdown().await; + } +} diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 6c272187f3c..d778e470b29 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -4367,9 +4367,20 @@ fn try_native_steer( true } Err(e) => { + // Structured fixed-reason label from the stock admission owner + // (`pool::send_steer`): task_absent / sender_absent / + // mailbox_full / mailbox_closed. Only admission refusals reach + // this arm — the ack watcher is spawned solely on `Ok(())`, so + // ack-native write failures are logged by the main loop's + // SteerAck arm instead and are never conflated with admission. + // The label never carries request content. + let reason = e + .admission_reason() + .map(|reason| reason.as_str()) + .unwrap_or("unclassified"); tracing::info!( channel = %channel_id, - error = ?e, + reason, "non-cancelling steer not accepted — falling back to cancel+merge" ); false @@ -4377,6 +4388,284 @@ fn try_native_steer( } } +// ── try_native_steer fallback-log tests ─────────────────────────────────────── +// +// Regression for the production tracing event `try_native_steer`'s Err arm +// emits before the caller falls back to the universal cancel+merge path. The +// `send_steer` admission-reason tests in pool.rs pin the refusal *labels* +// through `SteerError::admission_reason` alone — mutating this log's `reason` +// field leaves that suite green. These tests instead drive the REAL +// `try_native_steer` (real pool, real queue, real signed event, real steer +// body construction) through each of the four admission refusal branches and +// pin the log itself: the exact reason label, the exact production message, +// exactly the `channel`+`reason` fields (never request or error content), and +// the unchanged `false` return that keeps the caller on the fallback. +#[cfg(test)] +mod try_native_steer_fallback_log_tests { + use super::*; + use crate::pool::{SteerRequest, TaskMeta}; + use nostr::{EventBuilder, Keys, Kind}; + use tracing_subscriber::layer::SubscriberExt; + + /// Sentinel carried in the real event content — it flows into the steer + /// request body `try_native_steer` builds, and the fallback log must + /// never surface it. + const SECRET_REQUEST_CONTENT: &str = "SECRET-STEER-REQUEST-CONTENT"; + + /// One captured fallback log event: the exact values recorded for the + /// message/channel/reason fields, plus the names of any field beyond + /// that fixed vocabulary. + #[derive(Debug, Default)] + struct FallbackLog { + message: Option, + channel: Option, + reason: Option, + unexpected_fields: Vec, + } + + /// Records event field values (str values via `record_str`, Display and + /// format_args values via `record_debug`, exactly as tracing routes them) + /// and collects any field outside the fixed message/channel/reason set. + #[derive(Debug, Default)] + struct Recorder { + log: FallbackLog, + } + + impl Recorder { + fn store(&mut self, name: &str, value: String) { + match name { + "message" => self.log.message = Some(value), + "channel" => self.log.channel = Some(value), + "reason" => self.log.reason = Some(value), + _ => self.log.unexpected_fields.push(name.to_string()), + } + } + } + + impl tracing::field::Visit for Recorder { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.store(field.name(), value.to_string()); + } + + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.store(field.name(), format!("{value:?}")); + } + } + + /// Captures INFO events on `buzz_acp` carrying a `reason` field — the + /// fallback log's signature — mirroring the workspace's Layer+Visit + /// capture fixture (buzz-agent `count_silent_turn_warnings`). + struct Capture { + logs: Arc>>, + } + + impl tracing_subscriber::Layer for Capture { + fn on_event( + &self, + event: &tracing::Event<'_>, + _: tracing_subscriber::layer::Context<'_, S>, + ) { + if *event.metadata().level() != tracing::Level::INFO { + return; + } + if event.metadata().target() != "buzz_acp" { + return; + } + let mut recorder = Recorder::default(); + event.record(&mut recorder); + if recorder.log.reason.is_some() { + self.logs + .lock() + .unwrap() + .push(std::mem::take(&mut recorder.log)); + } + } + } + + /// Conversation scope for the steered channel — the same shape the + /// pool.rs admission fixtures use. + fn steer_scope() -> scope::SessionScope { + scope::SessionScope::Conversation { + channel_id: Uuid::nil(), + } + } + + /// A real signed kind:20001 stream message whose content becomes part of + /// the real steer request body `try_native_steer` builds. + fn stream_event() -> nostr::Event { + EventBuilder::new( + Kind::Custom(KIND_STREAM_MESSAGE as u16), + SECRET_REQUEST_CONTENT, + ) + .sign_with_keys(&Keys::generate()) + .expect("sign test stream message") + } + + /// Insert an in-flight task_map entry for `scope` carrying `steer_tx`, + /// mirroring `mark_agent_busy` — the existing seam for simulating an + /// in-flight prompt task without spawning a real agent turn (same shape + /// as the pool.rs `mark_agent_busy_with_steer_tx` fixture). + fn mark_in_flight_with_steer_tx( + pool: &mut AgentPool, + busy_scope: scope::SessionScope, + steer_tx: Option>, + ) { + let abort = pool.join_set.spawn(async {}); + pool.task_map_mut().insert( + abort.id(), + TaskMeta { + agent_index: 0, + channel_id: Some(busy_scope.channel_id()), + scope: Some(busy_scope), + turn_id: "t".into(), + recoverable_batch: None, + control_tx: None, + steer_tx, + successful_steer_deliveries: HashSet::new(), + }, + ); + } + + /// Drive the REAL `try_native_steer` exactly as the production caller + /// (`QueuedNormalListenerEvent::steer_or_interrupt`) does — event already + /// pushed into the queue, steer-eligible event, live steer-ack channel — + /// under a capturing subscriber. Returns the function's return value and + /// the fallback logs it emitted on this thread. + fn try_native_steer_capturing(pool: &mut AgentPool) -> (bool, Vec) { + let busy_scope = steer_scope(); + let event = stream_event(); + let mut queue = EventQueue::new(DedupMode::Queue); + // Caller invariant: the event is already queued before the steer + // attempt (see `try_native_steer`'s doc comment). + assert!( + queue.push(QueuedEvent { + channel_id: Uuid::nil(), + scope: busy_scope.clone(), + event: event.clone(), + received_at: std::time::Instant::now(), + prompt_tag: "mention".into(), + }), + "queued event must be accepted before the steer attempt" + ); + let (steer_ack_tx, _steer_ack_rx) = mpsc::unbounded_channel::(); + let logs = Arc::new(std::sync::Mutex::new(Vec::new())); + let subscriber = tracing_subscriber::registry().with(Capture { logs: logs.clone() }); + let returned = tracing::subscriber::with_default(subscriber, || { + try_native_steer( + pool, + &mut queue, + busy_scope.clone(), + event, + "mention".into(), + &steer_ack_tx, + ) + }); + let logs_out = std::mem::take(&mut *logs.lock().unwrap()); + (returned, logs_out) + } + + /// Assert the captured logs are exactly one production fallback event for + /// `expected_reason`: exact message, exact reason, exact channel, no + /// additional fields, and no request or error content anywhere. + fn assert_single_fallback_log(logs: &[FallbackLog], expected_reason: &str) { + assert_eq!( + logs.len(), + 1, + "exactly one fallback log must be emitted per refusal, got {logs:?}" + ); + let log = &logs[0]; + assert_eq!( + log.message.as_deref(), + Some("non-cancelling steer not accepted — falling back to cancel+merge"), + "fallback log message must stay the exact production bytes" + ); + assert_eq!( + log.reason.as_deref(), + Some(expected_reason), + "fallback log reason must be the exact admission label" + ); + let expected_channel = Uuid::nil().to_string(); + assert_eq!( + log.channel.as_deref(), + Some(expected_channel.as_str()), + "fallback log channel must be the scope's channel id" + ); + assert!( + log.unexpected_fields.is_empty(), + "fallback log must carry only channel+reason — no request/error \ + content fields: {:?}", + log.unexpected_fields + ); + let rendered = format!("{log:?}"); + assert!( + !rendered.contains(SECRET_REQUEST_CONTENT), + "fallback log must not leak request content: {rendered}" + ); + } + + /// No in-flight task owns the scope: `send_steer` refuses with + /// `PromptCompleted`, classified as `task_absent`. + #[test] + fn try_native_steer_logs_task_absent_reason_and_keeps_fallback() { + let mut pool = AgentPool::from_slots(vec![]); + let (returned, logs) = try_native_steer_capturing(&mut pool); + assert!( + !returned, + "refused native steer must keep the cancel+merge fallback" + ); + assert_single_fallback_log(&logs, "task_absent"); + } + + /// The in-flight task has no steer sender installed: `sender_absent`. + #[tokio::test] + async fn try_native_steer_logs_sender_absent_reason_and_keeps_fallback() { + let mut pool = AgentPool::from_slots(vec![]); + mark_in_flight_with_steer_tx(&mut pool, steer_scope(), None); + let (returned, logs) = try_native_steer_capturing(&mut pool); + assert!( + !returned, + "refused native steer must keep the cancel+merge fallback" + ); + assert_single_fallback_log(&logs, "sender_absent"); + } + + /// The capacity-1 steer mailbox already holds one in-flight steer: + /// `mailbox_full`. + #[tokio::test] + async fn try_native_steer_logs_mailbox_full_reason_and_keeps_fallback() { + let mut pool = AgentPool::from_slots(vec![]); + let (tx, _rx) = tokio::sync::mpsc::channel::(1); + mark_in_flight_with_steer_tx(&mut pool, steer_scope(), Some(tx.clone())); + let (ack_tx, _ack_rx) = tokio::sync::oneshot::channel::(); + tx.try_send(SteerRequest { + prompt_blocks: vec!["first in-flight steer".into()], + ack_tx, + }) + .expect("capacity-1 mailbox accepts the first in-flight steer"); + let (returned, logs) = try_native_steer_capturing(&mut pool); + assert!( + !returned, + "refused native steer must keep the cancel+merge fallback" + ); + assert_single_fallback_log(&logs, "mailbox_full"); + } + + /// The read loop's steer receiver is torn down: `mailbox_closed`. + #[tokio::test] + async fn try_native_steer_logs_mailbox_closed_reason_and_keeps_fallback() { + let mut pool = AgentPool::from_slots(vec![]); + let (tx, rx) = tokio::sync::mpsc::channel::(1); + mark_in_flight_with_steer_tx(&mut pool, steer_scope(), Some(tx)); + drop(rx); // read loop torn down before this steer arrived + let (returned, logs) = try_native_steer_capturing(&mut pool); + assert!( + !returned, + "refused native steer must keep the cancel+merge fallback" + ); + assert_single_fallback_log(&logs, "mailbox_closed"); + } +} + // ── dispatch_pending ────────────────────────────────────────────────────────── /// Flush queued work to available agents. diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 250221badca..66331455b0d 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -540,7 +540,11 @@ pub enum SteerError { /// running or just ended. AgentError { code: i64, message: String }, /// Transport-level failure: write error, read EOF, JSON-RPC framing - /// violation, etc. The string carries the underlying `AcpError`'s display. + /// violation, etc. For read-loop (post-admission) failures the string + /// carries the underlying `AcpError`'s display. For admission refusals + /// returned by [`AgentPool::send_steer`] it instead carries the fixed + /// [`SteerAdmissionReason`] label so callers can log a structured + /// reason without dumping request content. Transport(String), /// At steer-write time neither steer transport was available: no /// `expectedRunId` (`AcpClient::active_run_id` was `None`, so the @@ -576,6 +580,80 @@ pub enum SteerError { PromptCompleted, } +/// Fixed-vocabulary reason the stock admission owner +/// ([`AgentPool::send_steer`]) refused a native-steer request before any +/// wire write was attempted. +/// +/// These are structured diagnostic labels only — never request content, +/// agent output, or tokens. They classify *admission* (getting the request +/// into the in-flight read loop's capacity-1 steer mailbox) and are +/// deliberately distinct from post-admission failures: a steer that was +/// admitted and then failed at the wire keeps [`SteerError::Transport`] +/// carrying the underlying `AcpError` display, and `AgentError` / +/// `ExpectedRunIdMissing` / `OutcomeRejected` keep their own variants (see +/// [`SteerError::admission_reason`]). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SteerAdmissionReason { + /// No in-flight task owns the scope. Returned as + /// [`SteerError::PromptCompleted`], which keeps its distinct + /// release-and-normal-dispatch semantics — a separate branch, not a + /// transport refusal. + TaskAbsent, + /// The in-flight task has no `steer_tx` sender installed. + SenderAbsent, + /// The capacity-1 steer mailbox already holds one in-flight steer. + MailboxFull, + /// The read loop's steer receiver has been torn down. + MailboxClosed, +} + +impl SteerAdmissionReason { + /// Fixed, greppable label. Carried in the admission refusal's + /// [`SteerError::Transport`] string and in the main loop's fallback + /// log `reason` field; never contains request content. + pub fn as_str(self) -> &'static str { + match self { + Self::TaskAbsent => "task_absent", + Self::SenderAbsent => "sender_absent", + Self::MailboxFull => "mailbox_full", + Self::MailboxClosed => "mailbox_closed", + } + } + + /// Wrap the reason as the transport-refusal error `send_steer` returns, + /// preserving the historical `Err(SteerError::Transport(_))` return + /// contract for these branches. + fn transport_refusal(self) -> SteerError { + SteerError::Transport(self.as_str().to_owned()) + } +} + +impl SteerError { + /// Structured admission-refusal classification for errors produced by + /// the stock admission owner, [`AgentPool::send_steer`]. Returns `None` + /// for every post-admission error — ack-native steer write failures + /// (`Transport` carrying an `AcpError` display), `AgentError`, + /// `ExpectedRunIdMissing`, and `OutcomeRejected` — so admission + /// refusals are never conflated with wire failures. The fixed labels + /// mirror [`SteerAdmissionReason::as_str`] and are pinned together by + /// unit tests. + pub fn admission_reason(&self) -> Option { + match self { + Self::PromptCompleted => Some(SteerAdmissionReason::TaskAbsent), + Self::Transport(msg) => match msg.as_str() { + "sender_absent" => Some(SteerAdmissionReason::SenderAbsent), + "mailbox_full" => Some(SteerAdmissionReason::MailboxFull), + "mailbox_closed" => Some(SteerAdmissionReason::MailboxClosed), + // Any other Transport string is a post-admission wire + // failure (an AcpError display from the read loop), not an + // admission refusal. + _ => None, + }, + _ => None, + } + } +} + /// Outcome of a mid-turn steer, sent from the read loop back to the /// main loop's ack watcher. #[derive(Debug)] @@ -1106,9 +1184,11 @@ impl AgentPool { /// Returns `Ok(())` if the request was accepted by the read loop's /// receiver (capacity-1 mpsc; one slot is the single in-flight steer /// write). Returns `Err(SteerError::Transport(_))` on `Full`/`Closed` - /// (already-in-flight write, or read loop torn down). Callers must - /// fall back to the universal `ControlSignal::Steer` cancel+merge path - /// on `Err`. + /// (already-in-flight write, or read loop torn down); the `Transport` + /// string is the fixed [`SteerAdmissionReason`] label + /// (`sender_absent`/`mailbox_full`/`mailbox_closed`), never request + /// content. Callers must fall back to the universal + /// `ControlSignal::Steer` cancel+merge path on `Err`. /// /// This does **not** spawn the ack watcher — the caller owns the /// oneshot `ack_tx` inside `SteerRequest` and is responsible for @@ -1135,9 +1215,20 @@ impl AgentPool { let tx = meta .steer_tx .as_ref() - .ok_or_else(|| SteerError::Transport("steer_tx not installed".into()))?; - tx.try_send(request) - .map_err(|e| SteerError::Transport(e.to_string())) + .ok_or_else(|| SteerAdmissionReason::SenderAbsent.transport_refusal())?; + match tx.try_send(request) { + Ok(()) => Ok(()), + // Capacity-1 mailbox already holds one in-flight steer write. + // The refused request (including its oneshot ack) comes back + // inside the error and is dropped here, exactly as before. + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { + Err(SteerAdmissionReason::MailboxFull.transport_refusal()) + } + // Read loop receiver torn down before this request arrived. + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { + Err(SteerAdmissionReason::MailboxClosed.transport_refusal()) + } + } } /// Durably associate a successful steer with the exact ACP session that @@ -8762,6 +8853,191 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" // Reaching here without a panic is the test. } + // ── send_steer admission-reason tests ───────────────────────────────── + // + // These pin the fixed structured admission-refusal labels the stock + // admission owner (`send_steer`) reports, and that the fallback log's + // `reason` classification (via `SteerError::admission_reason`) matches + // them branch for branch without touching request content. + + /// Insert an in-flight task_map entry for `scope` carrying `steer_tx`, + /// mirroring `mark_agent_busy` — the existing seam for simulating an + /// in-flight prompt task without spawning a real agent turn. + fn mark_agent_busy_with_steer_tx( + pool: &mut AgentPool, + scope: SessionScope, + steer_tx: Option>, + ) { + let abort = pool.join_set.spawn(async {}); + pool.task_map_mut().insert( + abort.id(), + TaskMeta { + agent_index: 0, + channel_id: Some(scope.channel_id()), + scope: Some(scope), + turn_id: "t".into(), + recoverable_batch: None, + control_tx: None, + steer_tx, + successful_steer_deliveries: HashSet::new(), + }, + ); + } + + /// A minimal steer request; the ack oneshot is dropped with the request + /// when admission refuses it, exactly as in production. + fn steer_request() -> SteerRequest { + let (ack_tx, _ack_rx) = tokio::sync::oneshot::channel::(); + SteerRequest { + prompt_blocks: vec!["steer block".into()], + ack_tx, + } + } + + #[test] + fn send_steer_reports_task_absent_when_no_task_in_flight() { + let mut pool = AgentPool::from_slots(vec![]); + let scope = conv(Uuid::nil()); + let err = pool.send_steer(&scope, steer_request()).unwrap_err(); + // Unchanged return shape: the separate task-absent branch keeps + // PromptCompleted's release-and-normal-dispatch semantics. + assert!(matches!(err, SteerError::PromptCompleted)); + assert_eq!( + err.admission_reason(), + Some(SteerAdmissionReason::TaskAbsent) + ); + } + + #[tokio::test] + async fn send_steer_reports_sender_absent_when_no_steer_sender_installed() { + let mut pool = AgentPool::from_slots(vec![]); + let scope = conv(Uuid::nil()); + // The existing seam inserts the in-flight task with steer_tx: None. + mark_agent_busy(&mut pool, 0, scope.clone()); + let err = pool.send_steer(&scope, steer_request()).unwrap_err(); + assert!(matches!(&err, SteerError::Transport(msg) if msg == "sender_absent")); + assert_eq!( + err.admission_reason(), + Some(SteerAdmissionReason::SenderAbsent) + ); + } + + #[tokio::test] + async fn send_steer_reports_mailbox_full_when_one_steer_already_in_flight() { + let mut pool = AgentPool::from_slots(vec![]); + let scope = conv(Uuid::nil()); + let (tx, mut rx) = tokio::sync::mpsc::channel::(1); + mark_agent_busy_with_steer_tx(&mut pool, scope.clone(), Some(tx.clone())); + // Fill the capacity-1 mailbox: one steer write already in flight. + tx.try_send(steer_request()) + .expect("capacity-1 mailbox accepts the first in-flight steer"); + let err = pool.send_steer(&scope, steer_request()).unwrap_err(); + assert!(matches!(&err, SteerError::Transport(msg) if msg == "mailbox_full")); + assert_eq!( + err.admission_reason(), + Some(SteerAdmissionReason::MailboxFull) + ); + // The already-admitted request is untouched by the refused second + // attempt: send results are preserved. + let admitted = rx.try_recv().expect("first in-flight steer still queued"); + assert_eq!(admitted.prompt_blocks, vec!["steer block".to_string()]); + } + + #[tokio::test] + async fn send_steer_reports_mailbox_closed_after_receiver_dropped() { + let mut pool = AgentPool::from_slots(vec![]); + let scope = conv(Uuid::nil()); + let (tx, rx) = tokio::sync::mpsc::channel::(1); + mark_agent_busy_with_steer_tx(&mut pool, scope.clone(), Some(tx)); + drop(rx); // read loop torn down before this request arrived + let err = pool.send_steer(&scope, steer_request()).unwrap_err(); + assert!(matches!(&err, SteerError::Transport(msg) if msg == "mailbox_closed")); + assert_eq!( + err.admission_reason(), + Some(SteerAdmissionReason::MailboxClosed) + ); + } + + #[tokio::test] + async fn send_steer_ok_hands_request_to_read_loop_receiver() { + let mut pool = AgentPool::from_slots(vec![]); + let scope = conv(Uuid::nil()); + let (tx, mut rx) = tokio::sync::mpsc::channel::(1); + mark_agent_busy_with_steer_tx(&mut pool, scope.clone(), Some(tx)); + let (ack_tx, mut ack_rx) = tokio::sync::oneshot::channel::(); + pool.send_steer( + &scope, + SteerRequest { + prompt_blocks: vec!["steer block".into()], + ack_tx, + }, + ) + .expect("admission succeeds with a live receiver"); + let admitted = rx + .recv() + .await + .expect("request reaches the read loop mailbox"); + assert_eq!(admitted.prompt_blocks, vec!["steer block".to_string()]); + admitted + .ack_tx + .send(SteerAck::PromptCompletedNeutral) + .expect("ack oneshot survives admission"); + assert!(matches!( + ack_rx.try_recv(), + Ok(SteerAck::PromptCompletedNeutral) + )); + } + + #[test] + fn steer_admission_reason_labels_round_trip_with_send_steer_refusals() { + // send_steer writes exactly these labels via transport_refusal; + // admission_reason must classify them back without drift. + for reason in [ + SteerAdmissionReason::SenderAbsent, + SteerAdmissionReason::MailboxFull, + SteerAdmissionReason::MailboxClosed, + ] { + assert_eq!(reason.transport_refusal().admission_reason(), Some(reason)); + } + assert_eq!( + SteerError::PromptCompleted.admission_reason(), + Some(SteerAdmissionReason::TaskAbsent) + ); + // Fixed vocabulary the fallback log reports. + assert_eq!(SteerAdmissionReason::TaskAbsent.as_str(), "task_absent"); + assert_eq!(SteerAdmissionReason::SenderAbsent.as_str(), "sender_absent"); + assert_eq!(SteerAdmissionReason::MailboxFull.as_str(), "mailbox_full"); + assert_eq!( + SteerAdmissionReason::MailboxClosed.as_str(), + "mailbox_closed" + ); + } + + #[test] + fn steer_admission_reason_excludes_post_admission_errors() { + // Ack-native write failures are built in the read loop AFTER + // admission (acp.rs carries the AcpError display); they and the + // other ack outcomes must never classify as admission refusals. + let post_admission = [ + SteerError::Transport("I/O error: broken pipe mid-write".into()), + SteerError::AgentError { + code: -32601, + message: "method not found".into(), + }, + SteerError::ExpectedRunIdMissing, + SteerError::OutcomeRejected { + outcome: "failed".into(), + }, + ]; + for err in post_admission { + assert_eq!( + err.admission_reason(), + None, + "{err:?} is post-admission and must not classify as admission" + ); + } + } + // ── NIP-AM emit-hook unit tests ──────────────────────────────────────── /// `acp_stop_to_core` maps all ACP stop reasons to the correct NIP-AM