From 62b2d27b4a35fe6d3f7f96261bf0ca4e545ed94b Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 21 Oct 2026 05:40:39 +0900 Subject: [PATCH] fix: retain command failure stages and output collection state --- Cargo.lock | 2 + crates/tools/src/bash.rs | 8 +- crates/tools/src/bash_output.rs | 28 +- crates/tools/tests/integration.rs | 103 ++++++- crates/workdir/Cargo.toml | 1 + crates/workdir/src/external.rs | 4 + crates/workdir/src/local/command_transport.rs | 220 +++++++++++---- .../src/local/command_transport/tests.rs | 263 +++++++++++++++++ .../src/local/command_transport/tests/http.rs | 238 ++++++++++++++++ crates/workdir/src/operation.rs | 267 +++++++++++++++++- crates/worker-runtime/Cargo.toml | 1 + .../tests/command_transport_tests.rs | 264 ++++++++++++++++- crates/workspace-server/src/server.rs | 6 + .../server/tests/command_transport_tests.rs | 2 + 14 files changed, 1329 insertions(+), 78 deletions(-) create mode 100644 crates/workdir/src/local/command_transport/tests/http.rs diff --git a/Cargo.lock b/Cargo.lock index f2901cb1f..fd2babc6f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6497,6 +6497,7 @@ name = "workdir" version = "0.1.0" dependencies = [ "async-trait", + "axum", "fs-operation", "libc", "manifest", @@ -6597,6 +6598,7 @@ dependencies = [ "tokio", "tokio-tungstenite 0.29.0", "toml", + "tools", "tower", "tracing", "tracing-subscriber", diff --git a/crates/tools/src/bash.rs b/crates/tools/src/bash.rs index efeaa3acf..c3c3a6311 100644 --- a/crates/tools/src/bash.rs +++ b/crates/tools/src/bash.rs @@ -367,7 +367,7 @@ impl BashTool { guard.handle = None; let timed_out = terminal.status == workdir::CommandStatus::TimedOut; - let summary = + let mut summary = if cancellation_requested || terminal.status == workdir::CommandStatus::Cancelled { format!("$ {cmd_summary} (cancelled)") } else if timed_out { @@ -379,6 +379,12 @@ impl BashTool { None => format!("$ {cmd_summary} (terminated)"), } }; + if matches!( + terminal.output_collection, + workdir::CommandOutputCollection::Incomplete { .. } + ) { + summary.push_str(" (output incomplete)"); + } Ok(BashResult { summary, stdout_preview: String::from_utf8_lossy(&stdout_preview).into_owned(), diff --git a/crates/tools/src/bash_output.rs b/crates/tools/src/bash_output.rs index 538b50433..fad3b02e9 100644 --- a/crates/tools/src/bash_output.rs +++ b/crates/tools/src/bash_output.rs @@ -68,8 +68,10 @@ pub struct OutputRange { pub struct OutputTerminal { pub status: Option, pub exit_code: Option, - /// False on a forcibly dropped collector or failed transport: received bytes are retained, - /// but they are not a claim about the unobserved process tail. + pub process_failure: Option, + pub output_collection: Option, + /// True only for confirmed EOF collection (quota is reported separately). A terminal + /// fence or exit zero alone does not confirm the unobserved process tail. pub transport_complete: bool, } @@ -405,12 +407,9 @@ impl BashOutputWriter { terminal: &workdir::CommandTerminal, ) -> Result { let mut r = self.record.lock().unwrap(); - if terminal.stdout_delivered_bytes != r.stdout_seen + if !terminal.metadata_valid() + || terminal.stdout_delivered_bytes != r.stdout_seen || terminal.stderr_delivered_bytes != r.stderr_seen - || terminal.stdout_bytes < r.stdout_seen - || terminal.stderr_bytes < r.stderr_seen - || ((terminal.stdout_bytes != r.stdout_seen || terminal.stderr_bytes != r.stderr_seen) - != terminal.truncation_reason.is_some()) { return Err("command terminal counts do not fence delivered bytes".into()); } @@ -419,7 +418,12 @@ impl BashOutputWriter { r.terminal = Some(OutputTerminal { status: Some(terminal.status), exit_code: terminal.exit_code, - transport_complete: true, + transport_complete: matches!( + terminal.output_collection, + workdir::CommandOutputCollection::Complete + ), + process_failure: terminal.process_failure, + output_collection: Some(terminal.output_collection), }); Ok(r.metadata()) } @@ -439,6 +443,8 @@ impl Drop for BashOutputWriter { status: None, exit_code: None, transport_complete: false, + process_failure: None, + output_collection: None, }); } } @@ -456,6 +462,8 @@ mod tests { status: Some(CommandStatus::Completed), exit_code: Some(0), transport_complete: true, + process_failure: None, + output_collection: Some(workdir::CommandOutputCollection::Complete), }) .id } @@ -514,6 +522,8 @@ mod tests { status: Some(CommandStatus::Completed), exit_code: Some(0), transport_complete: true, + process_failure: None, + output_collection: Some(workdir::CommandOutputCollection::Complete), }); assert!(root.metadata(&metadata.id).is_some()); let _replacement = sibling.begin().unwrap(); @@ -582,7 +592,7 @@ mod tests { let root = BashOutputView::new(); let mut writer = root.begin().unwrap(); writer.append(CommandStream::Stdout, b"abc"); - let mut terminal = serde_json::json!({"status":"completed","exit_code":0,"stdout_bytes":9,"stderr_bytes":0,"stdout_delivered_bytes":4,"stderr_delivered_bytes":0,"truncation_reason":"byte_budget"}); + let mut terminal = serde_json::json!({"status":"completed","exit_code":0,"process_failure":null,"output_collection":{"state":"complete"},"stdout_bytes":9,"stderr_bytes":0,"stdout_delivered_bytes":4,"stderr_delivered_bytes":0,"truncation_reason":"byte_budget"}); assert!( writer .finish_transport(&serde_json::from_value(terminal.clone()).unwrap()) diff --git a/crates/tools/tests/integration.rs b/crates/tools/tests/integration.rs index f7b1dd2da..50ddfd32d 100644 --- a/crates/tools/tests/integration.rs +++ b/crates/tools/tests/integration.rs @@ -416,6 +416,7 @@ struct ExternalValidatingSession { truncate_first_binary_response: bool, command_response_losses: std::sync::atomic::AtomicUsize, lose_terminal_only: bool, + collection_override: Option, mutate_after_first_binary_response: Option, } @@ -534,7 +535,11 @@ impl workdir::WorkdirSession for ExternalValidatingSession { &self, request: workdir::CommandOutputRequest, ) -> Result { - let output = self.inner.command_output(request).await?; + let mut output = self.inner.command_output(request).await?; + if let (Some(collection), Some(terminal)) = (self.collection_override, &mut output.terminal) + { + terminal.output_collection = collection; + } if (!self.lose_terminal_only || output.terminal.is_some()) && self .command_response_losses @@ -575,6 +580,7 @@ async fn external_view_image_uses_a_protocol_bounded_request() { truncate_first_binary_response: false, command_response_losses: std::sync::atomic::AtomicUsize::new(0), lose_terminal_only: false, + collection_override: None, mutate_after_first_binary_response: None, }); let png = include_bytes!("fixtures/view-image/valid.png"); @@ -600,6 +606,7 @@ async fn external_view_image_rejects_truncated_and_cross_version_chunks() { truncate_first_binary_response: true, command_response_losses: std::sync::atomic::AtomicUsize::new(0), lose_terminal_only: false, + collection_override: None, mutate_after_first_binary_response: None, }); let (_meta, tool) = view_image_tool(truncated)(); @@ -618,6 +625,7 @@ async fn external_view_image_rejects_truncated_and_cross_version_chunks() { truncate_first_binary_response: false, command_response_losses: std::sync::atomic::AtomicUsize::new(0), lose_terminal_only: false, + collection_override: None, mutate_after_first_binary_response: Some(image_path), }); let (_meta, tool) = view_image_tool(changing)(); @@ -1104,7 +1112,20 @@ async fn bash_cancellation_returns_bounded_progress_as_terminal_output() { let execution = tokio::spawn(async move { executing.execute(r#"{"command":"printf 'before'; printf 'err-before' >&2; while :; do :; done","cwd":"."}"#, execution_context).await }); - await_command_output(&mut events).await; + // Cancellation must follow both writes, not merely the first stdout chunk. + tokio::time::timeout(std::time::Duration::from_secs(5), async { + let mut ready = [false; 2]; + while !ready.iter().all(|seen| *seen) { + if let workdir::CommandEvent::Output { stream, .. } = events.recv().await.unwrap() { + ready[match stream { + workdir::CommandStream::Stdout => 0, + workdir::CommandStream::Stderr => 1, + }] = true; + } + } + }) + .await + .expect("both streams written before cancellation"); bash.cancel_execution(&context).await.unwrap(); let ToolError::Cancelled(output) = tokio::time::timeout(std::time::Duration::from_secs(5), execution) @@ -1281,6 +1302,7 @@ async fn lost_page_and_terminal_responses_replay_without_restarting_or_double_ap mutate_after_first_binary_response: None, command_response_losses: std::sync::atomic::AtomicUsize::new(1), lose_terminal_only, + collection_override: None, }); let view = tools::bash_output::BashOutputView::new(); let definitions = core_builtin_tools(session.clone(), Tracker::new(), view.clone()); @@ -1315,3 +1337,80 @@ async fn lost_page_and_terminal_responses_replay_without_restarting_or_double_ap session.close().await.unwrap(); } } + +// Provider metadata is the fixture here; actual read/drain injection and HTTP +// forwarding are tested by workdir's production-supervisor suite. +#[tokio::test] +async fn exit_zero_with_incomplete_output_keeps_partial_bytes_and_safe_failure_in_retained_evidence() + { + use workdir::{ + CommandFailure, CommandFailureStage as Stage, CommandIoErrorKind, CommandOutputCollection, + }; + for stage in [ + Stage::StdoutRead, + Stage::StderrRead, + Stage::StdoutDrain, + Stage::StderrDrain, + Stage::OutputDelivery, + ] { + let root = TempDir::new().unwrap(); + let failure = CommandFailure { + stage, + reason: CommandIoErrorKind::PermissionDenied, + os_code: None, + }; + let collection = CommandOutputCollection::Incomplete { failure }; + let session: WorkdirSessionHandle = Arc::new(ExternalValidatingSession { + inner: LocalWorkdirSession::new( + Scope::writable(root.path()).unwrap(), + root.path().into(), + ), + truncate_first_binary_response: false, + mutate_after_first_binary_response: None, + command_response_losses: std::sync::atomic::AtomicUsize::new(0), + lose_terminal_only: false, + collection_override: Some(collection), + }); + let view = tools::bash_output::BashOutputView::new(); + let (_, bash) = tools::bash_tool(session.clone(), view.clone())(); + let out = call( + &bash, + json!({"cwd":".","command":"printf partial; printf err >&2"}), + ) + .await; + assert!(out.summary.ends_with("(output incomplete)")); + let body: serde_json::Value = serde_json::from_str(&out.content.unwrap()).unwrap(); + let metadata = &body["output"]; + assert_eq!(body["stdout_preview"], "partial"); + assert_eq!(body["stderr_preview"], "err"); + assert_eq!(body["preview_omitted"], false); + assert_eq!(metadata["terminal"]["exit_code"], 0); + assert_eq!(metadata["terminal"]["status"], "completed"); + assert_eq!(metadata["terminal"]["transport_complete"], false); + assert_eq!( + metadata["terminal"]["output_collection"], + serde_json::to_value(collection).unwrap() + ); + assert_eq!(metadata["storage_truncated"], false); + assert_eq!(metadata["transport_truncation"], serde_json::Value::Null); + assert_eq!(metadata["stdout_seen"], 7); + assert_eq!(metadata["stdout_delivered"], 7); + let id = metadata["id"].as_str().unwrap(); + assert_eq!( + serde_json::to_value(view.metadata(id).unwrap()).unwrap(), + *metadata + ); + assert_eq!( + view.read(id, workdir::CommandStream::Stdout, 0, 100) + .unwrap(), + b"partial" + ); + assert_eq!( + view.read(id, workdir::CommandStream::Stderr, 0, 100) + .unwrap(), + b"err" + ); + assert!(!metadata.to_string().contains(root.path().to_str().unwrap())); + session.close().await.unwrap(); + } +} diff --git a/crates/workdir/Cargo.toml b/crates/workdir/Cargo.toml index b8541897a..c972355ec 100644 --- a/crates/workdir/Cargo.toml +++ b/crates/workdir/Cargo.toml @@ -23,5 +23,6 @@ thiserror.workspace = true tokio = { workspace = true, features = ["io-util", "process", "rt", "sync", "time"] } [dev-dependencies] +axum.workspace = true serde_json.workspace = true tempfile.workspace = true diff --git a/crates/workdir/src/external.rs b/crates/workdir/src/external.rs index c1c00bfd8..e1d37d21c 100644 --- a/crates/workdir/src/external.rs +++ b/crates/workdir/src/external.rs @@ -924,6 +924,8 @@ mod tests { chunks: vec![], next_cursor: 0, terminal: Some(crate::CommandTerminal { + process_failure: None, + output_collection: crate::CommandOutputCollection::Complete, status: crate::CommandStatus::Completed, exit_code: Some(0), stdout_bytes: 0, @@ -1082,6 +1084,8 @@ mod tests { }], next_cursor: 3, terminal: Some(crate::CommandTerminal { + process_failure: None, + output_collection: crate::CommandOutputCollection::Complete, status: crate::CommandStatus::Completed, exit_code: Some(0), stdout_bytes: 100 * 1024 * 1024, diff --git a/crates/workdir/src/local/command_transport.rs b/crates/workdir/src/local/command_transport.rs index 1af30b1fc..0d5a261e4 100644 --- a/crates/workdir/src/local/command_transport.rs +++ b/crates/workdir/src/local/command_transport.rs @@ -3,6 +3,7 @@ //! A replay page survives a lost HTTP response until its next cursor is sent. use super::*; +use crate::{CommandFailure, CommandFailureStage, CommandIoErrorKind, CommandOutputCollection}; use tokio::io::{AsyncRead, AsyncReadExt}; use tokio::sync::mpsc; @@ -146,6 +147,33 @@ pub(super) fn start( telemetry: CommandTelemetry, environment: BTreeMap, resources: Vec>, +) -> LocalCommand { + let execution_id = command_id.clone(); + let execution_telemetry = telemetry.clone(); + start_with( + command_id, + telemetry, + resources, + move |cancel, sender| async move { + execute( + cwd, + request, + &execution_id, + &execution_telemetry, + environment, + cancel, + &sender, + ) + .await + }, + ) +} + +fn start_with + Send + 'static>( + command_id: String, + telemetry: CommandTelemetry, + resources: Vec>, + execution: impl FnOnce(watch::Receiver, mpsc::Sender) -> F + Send + 'static, ) -> LocalCommand { let (sender, receiver) = mpsc::channel(QUEUE_CHUNKS); let (cancel, cancel_rx) = watch::channel(false); @@ -160,16 +188,7 @@ pub(super) fn start( })); let task = tokio::spawn(async move { let _resources = resources; - let terminal = execute( - cwd, - request, - &command_id, - &telemetry, - environment, - cancel_rx, - &sender, - ) - .await; + let terminal = execution(cancel_rx, sender.clone()).await; telemetry.terminal(&command_id, terminal.status, terminal.exit_code); // Closing the byte queue fences this already-published terminal, without // requiring spare queue capacity or a receiver to release task resources. @@ -192,10 +211,18 @@ struct ByteAccounting { } impl ByteAccounting { - fn terminal(&self, status: CommandStatus, exit_code: Option) -> crate::CommandTerminal { + fn terminal( + &self, + status: CommandStatus, + exit_code: Option, + process_failure: Option, + output_collection: CommandOutputCollection, + ) -> crate::CommandTerminal { crate::CommandTerminal { status, exit_code, + process_failure, + output_collection, stdout_bytes: self.stdout, stderr_bytes: self.stderr, stdout_delivered_bytes: self.stdout_delivered, @@ -215,7 +242,7 @@ async fn pump( command_id: &str, mut remaining: u64, accounting: &mut ByteAccounting, -) -> std::io::Result<()> { +) -> Result<(), CommandFailure> { let mut stdout_done = false; let mut stderr_done = false; let mut stdout_decoder = CommandOutputDecoder::default(); @@ -228,7 +255,11 @@ async fn pump( // Before quota, reserve first to retain bounded pipe backpressure. let permit = if remaining > 0 { let Ok(permit) = sender.reserve().await else { - return Ok(()); + return Err(CommandFailure { + stage: CommandFailureStage::OutputDelivery, + reason: CommandIoErrorKind::Unknown, + os_code: None, + }); }; Some(permit) } else { @@ -242,7 +273,15 @@ async fn pump( CommandStream::Stdout => &stdout_buffer, CommandStream::Stderr => &stderr_buffer, }; - let count = result?; + let count = result.map_err(|error| { + CommandFailure::io( + match stream { + CommandStream::Stdout => CommandFailureStage::StdoutRead, + CommandStream::Stderr => CommandFailureStage::StderrRead, + }, + &error, + ) + })?; let decoder = match stream { CommandStream::Stdout => &mut stdout_decoder, CommandStream::Stderr => &mut stderr_decoder, @@ -288,7 +327,7 @@ async fn execute( command_id: &str, telemetry: &CommandTelemetry, environment: BTreeMap, - mut cancel: watch::Receiver, + cancel: watch::Receiver, sender: &mpsc::Sender, ) -> crate::CommandTerminal { telemetry.started(command_id, request.tool_call_id); @@ -304,74 +343,159 @@ async fn execute( .kill_on_drop(true); #[cfg(unix)] command.as_std_mut().process_group(0); - let mut accounting = ByteAccounting::default(); - let Ok(mut child) = command.spawn() else { - return accounting.terminal(CommandStatus::Failed, None); + let mut child = match command.spawn() { + Ok(child) => child, + Err(error) => { + return ByteAccounting::default().terminal( + CommandStatus::Failed, + None, + Some(CommandFailure::io(CommandFailureStage::Spawn, &error)), + CommandOutputCollection::NotStarted, + ); + } }; - let mut process_group = CommandProcessGroupGuard::new(child.id()); let stdout = child.stdout.take().expect("piped stdout"); let stderr = child.stderr.take().expect("piped stderr"); - let (status, exit_code) = { + supervise( + child, + stdout, + stderr, + request.timeout_secs, + request.transport_byte_budget, + command_id, + telemetry, + cancel, + sender, + ) + .await +} + +// Substitute only syscalls and readers: injected faults run the production supervisor. +trait CommandProcess: Send { + fn id(&self) -> Option; + fn start_kill(&mut self); + fn wait( + &mut self, + ) -> impl std::future::Future> + Send; +} + +impl CommandProcess for tokio::process::Child { + fn id(&self) -> Option { + self.id() + } + fn start_kill(&mut self) { + let _ = self.start_kill(); + } + async fn wait(&mut self) -> std::io::Result { + self.wait().await + } +} + +fn process_result( + result: std::io::Result, +) -> (CommandStatus, Option, Option) { + match result { + Ok(exit) => ( + if exit.success() { + CommandStatus::Completed + } else { + CommandStatus::Failed + }, + exit.code(), + None, + ), + Err(error) => ( + CommandStatus::Failed, + None, + Some(CommandFailure::io(CommandFailureStage::Wait, &error)), + ), + } +} + +#[allow(clippy::too_many_arguments)] +async fn supervise( + mut child: impl CommandProcess, + stdout: impl AsyncRead + Unpin, + stderr: impl AsyncRead + Unpin, + timeout_secs: u64, + byte_budget: u64, + command_id: &str, + telemetry: &CommandTelemetry, + mut cancel: watch::Receiver, + sender: &mpsc::Sender, +) -> crate::CommandTerminal { + let mut process_group = CommandProcessGroupGuard::new(child.id()); + let mut accounting = ByteAccounting::default(); + let (status, exit_code, process_failure, output_collection) = { let output = pump( stdout, stderr, sender, telemetry, command_id, - request.transport_byte_budget, + byte_budget, &mut accounting, ); tokio::pin!(output); - let mut drained = false; - let timeout = tokio::time::sleep(Duration::from_secs(request.timeout_secs.max(1))); + let mut collected = None; + let timeout = tokio::time::sleep(Duration::from_secs(timeout_secs.max(1))); tokio::pin!(timeout); let terminal = loop { tokio::select! { - result = &mut output, if !drained => { - drained = true; - if result.is_err() || sender.is_closed() { + result = &mut output, if collected.is_none() => { + let failed = result.is_err(); + collected = Some(result); + if failed { process_group.terminate(); - let _ = child.start_kill(); - let _ = child.wait().await; - break (CommandStatus::Failed, None); + child.start_kill(); + break process_result(child.wait().await); } } exit = child.wait() => { process_group.terminate(); - break match exit { - Ok(exit) => (if exit.success() { CommandStatus::Completed } else { CommandStatus::Failed }, exit.code()), - Err(_) => (CommandStatus::Failed, None), - }; + break process_result(exit); } _ = &mut timeout => { process_group.terminate(); - let _ = child.start_kill(); - let code = child.wait().await.ok().and_then(|exit| exit.code()); - break (CommandStatus::TimedOut, code); + child.start_kill(); + let (_, code, failure) = process_result(child.wait().await); + break (CommandStatus::TimedOut, code, failure); } _ = sender.closed() => { process_group.terminate(); - let _ = child.start_kill(); - let _ = child.wait().await; - break (CommandStatus::Cancelled, None); + child.start_kill(); + let (_, code, failure) = process_result(child.wait().await); + break (CommandStatus::Cancelled, code, failure); } changed = cancel.changed() => { if changed.is_err() || *cancel.borrow() { process_group.terminate(); - let _ = child.start_kill(); - let code = child.wait().await.ok().and_then(|exit| exit.code()); - break (CommandStatus::Cancelled, code); + child.start_kill(); + let (_, code, failure) = process_result(child.wait().await); + break (CommandStatus::Cancelled, code, failure); } } } }; - // Even after kill, preserve the bounded pipe tail before publishing terminal. - if !drained { - let _ = output.await; - } - terminal + // Exit is not pipe EOF. Retain read failures first observed during drain. + let result = match collected { + Some(result) => result, + None => output.await.map_err(|mut failure| { + failure.stage = match failure.stage { + CommandFailureStage::StdoutRead => CommandFailureStage::StdoutDrain, + CommandFailureStage::StderrRead => CommandFailureStage::StderrDrain, + stage => stage, + }; + failure + }), + }; + let collection = match result { + Ok(()) => CommandOutputCollection::Complete, + Err(failure) => CommandOutputCollection::Incomplete { failure }, + }; + (terminal.0, terminal.1, terminal.2, collection) }; - accounting.terminal(status, exit_code) + accounting.terminal(status, exit_code, process_failure, output_collection) } #[cfg(test)] diff --git a/crates/workdir/src/local/command_transport/tests.rs b/crates/workdir/src/local/command_transport/tests.rs index 12f20b061..324ce614e 100644 --- a/crates/workdir/src/local/command_transport/tests.rs +++ b/crates/workdir/src/local/command_transport/tests.rs @@ -520,3 +520,266 @@ async fn slow_receiver_backpressures_before_quota_but_full_queue_does_not_block_ assert_eq!(queued_bytes, QUEUE_CHUNKS * COMMAND_EVENT_CHUNK_BYTES); } } + +#[derive(Clone, Default)] +struct Gate(Arc)>>); + +impl Gate { + fn open(&self) { + let mut state = self.0.lock().unwrap(); + state.0 = true; + if let Some(waker) = state.1.take() { + waker.wake(); + } + } + + fn poll(&self, cx: &mut std::task::Context<'_>) -> std::task::Poll<()> { + let mut state = self.0.lock().unwrap(); + if state.0 { + std::task::Poll::Ready(()) + } else { + state.1 = Some(cx.waker().clone()); + std::task::Poll::Pending + } + } +} + +struct FaultProcess { + ready: Gate, + exited: Gate, + wait_error: Option, +} + +impl CommandProcess for FaultProcess { + fn id(&self) -> Option { + None + } + fn start_kill(&mut self) { + self.ready.open(); + } + async fn wait(&mut self) -> std::io::Result { + std::future::poll_fn(|cx| self.ready.poll(cx)).await; + self.exited.open(); + match self.wait_error { + Some(kind) => Err(std::io::Error::new( + kind, + "/private/path TOKEN=secret command-body", + )), + None => Ok(std::process::ExitStatus::default()), + } + } +} + +struct FaultReader { + prefix: &'static [u8], + prefix_read: Gate, + fail_ready: Gate, +} + +impl AsyncRead for FaultReader { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buffer: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + if !self.prefix.is_empty() { + let count = self.prefix.len().min(buffer.remaining()); + buffer.put_slice(&self.prefix[..count]); + self.prefix = &self.prefix[count..]; + self.prefix_read.open(); + return std::task::Poll::Ready(Ok(())); + } + std::task::ready!(self.fail_ready.poll(cx)); + std::task::Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::PermissionDenied, + "/private/path TOKEN=secret command-body", + ))) + } +} + +// Faults replace only OS wait/read boundaries, not collection or finalization. +#[tokio::test] +async fn read_and_post_exit_drain_faults_keep_prefix_exit_and_safe_original_reason() { + for stream in [CommandStream::Stdout, CommandStream::Stderr] { + for after_exit in [false, true] { + let ready = Gate::default(); + let exited = Gate::default(); + let fail_ready = if after_exit { + exited.clone() + } else { + let gate = Gate::default(); + gate.open(); + gate + }; + let reader = FaultReader { + prefix: b"partial", + prefix_read: if after_exit { + ready.clone() + } else { + Gate::default() + }, + fail_ready, + }; + let process = FaultProcess { + ready, + exited, + wait_error: None, + }; + let (sender, mut receiver) = mpsc::channel(QUEUE_CHUNKS); + let (_cancel, cancel_rx) = watch::channel(false); + let telemetry = CommandTelemetry::new(); + let terminal = tokio::time::timeout(Duration::from_secs(5), async { + match stream { + CommandStream::Stdout => { + supervise( + process, + reader, + &b""[..], + 10, + 100, + "fault", + &telemetry, + cancel_rx, + &sender, + ) + .await + } + CommandStream::Stderr => { + supervise( + process, + &b""[..], + reader, + 10, + 100, + "fault", + &telemetry, + cancel_rx, + &sender, + ) + .await + } + } + }) + .await + .unwrap(); + let Message::Bytes(chunk) = receiver.recv().await.unwrap(); + assert_eq!(chunk.bytes, b"partial"); + assert_eq!(chunk.stream, stream); + assert_eq!(chunk.sequence, 0); + assert_eq!(chunk.offset, 0); + assert!(receiver.try_recv().is_err()); + assert_eq!(terminal.status, CommandStatus::Completed); + assert_eq!(terminal.exit_code, Some(0)); + assert_eq!(terminal.process_failure, None); + assert_eq!(terminal.stdout_bytes + terminal.stderr_bytes, 7); + assert_eq!( + terminal.stdout_delivered_bytes + terminal.stderr_delivered_bytes, + 7 + ); + assert_eq!(terminal.truncation_reason, None); + assert_eq!( + terminal.output_collection, + CommandOutputCollection::Incomplete { + failure: CommandFailure { + stage: match (stream, after_exit) { + (CommandStream::Stdout, false) => CommandFailureStage::StdoutRead, + (CommandStream::Stdout, true) => CommandFailureStage::StdoutDrain, + (CommandStream::Stderr, false) => CommandFailureStage::StderrRead, + (CommandStream::Stderr, true) => CommandFailureStage::StderrDrain, + }, + reason: CommandIoErrorKind::PermissionDenied, + os_code: None + }, + } + ); + let wire = serde_json::to_string(&terminal).unwrap(); + for secret in ["/private", "TOKEN", "secret", "command-body"] { + assert!(!wire.contains(secret)); + } + } + } +} + +#[tokio::test] +async fn wait_fault_keeps_collected_bytes_and_does_not_invent_exit_information() { + let ready = Gate::default(); + ready.open(); + let process = FaultProcess { + ready, + exited: Gate::default(), + wait_error: Some(std::io::ErrorKind::InvalidData), + }; + let (sender, mut receiver) = mpsc::channel(QUEUE_CHUNKS); + let (_cancel, cancel_rx) = watch::channel(false); + let terminal = supervise( + process, + &b"out"[..], + &b"err"[..], + 10, + 100, + "wait-fault", + &CommandTelemetry::new(), + cancel_rx, + &sender, + ) + .await; + assert_eq!(terminal.status, CommandStatus::Failed); + assert_eq!(terminal.exit_code, None); + assert_eq!( + terminal.process_failure, + Some(CommandFailure { + stage: CommandFailureStage::Wait, + reason: CommandIoErrorKind::InvalidData, + os_code: None + }) + ); + assert_eq!( + terminal.output_collection, + CommandOutputCollection::Complete + ); + assert_eq!(terminal.stdout_bytes, 3); + assert_eq!(terminal.stderr_bytes, 3); + let mut chunks = Vec::new(); + while let Ok(Message::Bytes(chunk)) = receiver.try_recv() { + chunks.push(chunk); + } + assert_eq!(stream_bytes(&chunks, CommandStream::Stdout), b"out"); + assert_eq!(stream_bytes(&chunks, CommandStream::Stderr), b"err"); + assert!(!serde_json::to_string(&terminal).unwrap().contains("secret")); +} + +#[tokio::test] +async fn spawn_fault_retains_safe_reason_without_claiming_output_collection() { + let root = TempDir::new().unwrap(); + let (sender, _receiver) = mpsc::channel(QUEUE_CHUNKS); + let (_cancel, cancel_rx) = watch::channel(false); + let terminal = execute( + root.path().join("missing-secret-path"), + Harness::request("secret-command", 10, 100), + "spawn-fault", + &CommandTelemetry::new(), + BTreeMap::new(), + cancel_rx, + &sender, + ) + .await; + assert_eq!( + terminal.process_failure, + Some(CommandFailure { + stage: CommandFailureStage::Spawn, + reason: CommandIoErrorKind::NotFound, + os_code: Some(libc::ENOENT) + }) + ); + assert_eq!( + terminal.output_collection, + CommandOutputCollection::NotStarted + ); + assert_eq!(terminal.exit_code, None); + assert_eq!(terminal.status, CommandStatus::Failed); + assert_eq!(terminal.stdout_bytes + terminal.stderr_bytes, 0); + assert!(!serde_json::to_string(&terminal).unwrap().contains("secret")); +} + +#[cfg(feature = "http-client")] +mod http; diff --git a/crates/workdir/src/local/command_transport/tests/http.rs b/crates/workdir/src/local/command_transport/tests/http.rs new file mode 100644 index 000000000..0bed3d4dc --- /dev/null +++ b/crates/workdir/src/local/command_transport/tests/http.rs @@ -0,0 +1,238 @@ +//! Loopback integration of the real supervisor, byte queue/replay, local session, +//! HTTP dispatcher/client. OS wait/read are +//! deterministic faults; session open and start bind that injected process fixture. +//! Authorization is not asserted here (Runtime's signed HTTP suite owns it). + +use super::*; +use crate::http::*; +use axum::{Json, Router, extract::State, routing::post}; + +struct Server(tokio::task::JoinHandle<()>); +impl Drop for Server { + fn drop(&mut self) { + self.0.abort(); + } +} + +async fn operation( + State(session): State>, + Json(request): Json, +) -> Json { + if matches!(request.operation, WorkdirSessionOperation::CommandStart(_)) { + return Json(WorkdirSessionOperationResult::CommandStart(CommandHandle( + "injected".into(), + ))); + } + Json( + dispatch_workdir_session_operation(session.as_ref(), request.operation) + .await + .unwrap(), + ) +} + +#[tokio::test] +async fn pipe_and_wait_faults_cross_real_http_with_prefix_and_terminal_replay() { + tokio::time::timeout(Duration::from_secs(20), async { + for (stream, after_exit, wait_fault) in [ + (CommandStream::Stdout, false, false), + (CommandStream::Stderr, false, false), + (CommandStream::Stdout, true, false), + (CommandStream::Stderr, true, false), + (CommandStream::Stdout, false, true), + ] { + let root = TempDir::new().unwrap(); + let session = Arc::new(LocalWorkdirSession::new( + Scope::writable(root.path()).unwrap(), + root.path().into(), + )); + let ready = Gate::default(); + let exited = Gate::default(); + if wait_fault { + ready.open(); + } + let process = FaultProcess { + ready: ready.clone(), + exited: exited.clone(), + wait_error: wait_fault.then_some(std::io::ErrorKind::InvalidData), + }; + let fail_ready = if after_exit { + exited + } else { + let gate = Gate::default(); + gate.open(); + gate + }; + let reader = FaultReader { + prefix: b"partial", + prefix_read: if after_exit { ready } else { Gate::default() }, + fail_ready, + }; + let command = start_with( + "injected".into(), + CommandTelemetry::new(), + vec![], + move |cancel, sender| async move { + let telemetry = CommandTelemetry::new(); + if wait_fault { + supervise( + process, + &b"partial"[..], + &b""[..], + 10, + 100, + "injected", + &telemetry, + cancel, + &sender, + ) + .await + } else { + match stream { + CommandStream::Stdout => { + supervise( + process, + reader, + &b""[..], + 10, + 100, + "injected", + &telemetry, + cancel, + &sender, + ) + .await + } + CommandStream::Stderr => { + supervise( + process, + &b""[..], + reader, + 10, + 100, + "injected", + &telemetry, + cancel, + &sender, + ) + .await + } + } + } + }, + ); + session + .inner + .commands + .lock() + .await + .insert("injected".into(), command); + let app = Router::new() + .route( + "/v1/working-directories/{id}/sessions", + post(|| async { + Json(OpenWorkdirSessionResponse { + session_id: WorkdirSessionId::new("fault-session").unwrap(), + workdir_id: crate::Workdir::new("fault-workdir").id().clone(), + capabilities: crate::WorkdirSessionCapabilities::ALL, + }) + }), + ) + .route("/v1/workdir-sessions/{id}/operations", post(operation)) + .with_state(session.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let _server = Server(tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + })); + let remote: crate::WorkdirSessionHandle = Arc::new( + RemoteWorkdirSession::open( + reqwest::Client::builder().no_proxy().build().unwrap(), + base.parse().unwrap(), + "fixture-token", + crate::Workdir::new("fault-workdir").id().clone(), + OpenWorkdirSessionRequest::default(), + ) + .await + .unwrap(), + ); + let handle = CommandHandle("injected".into()); + let mut request = CommandOutputRequest { + handle, + cursor: 0, + limit: 1, + wait: true, + }; + let mut bytes = Vec::new(); + let terminal = loop { + let page = remote.command_output(request.clone()).await.unwrap(); + assert_eq!(remote.command_output(request.clone()).await.unwrap(), page); + for chunk in &page.chunks { + assert_eq!(chunk.sequence, request.cursor); + assert_eq!(chunk.offset, request.cursor); + assert_eq!(chunk.stream, stream); + bytes.extend_from_slice(&chunk.bytes); + request.cursor += chunk.bytes.len() as u64; + } + assert_eq!(page.next_cursor, request.cursor); + if let Some(terminal) = page.terminal { + break terminal; + } + }; + assert_eq!(bytes, b"partial"); + assert_eq!(terminal.truncation_reason, None); + assert_eq!(terminal.stdout_bytes + terminal.stderr_bytes, 7); + assert_eq!( + terminal.stdout_delivered_bytes + terminal.stderr_delivered_bytes, + 7 + ); + if wait_fault { + assert_eq!(terminal.status, CommandStatus::Failed); + assert_eq!(terminal.exit_code, None); + assert_eq!( + terminal.process_failure, + Some(CommandFailure { + stage: CommandFailureStage::Wait, + reason: CommandIoErrorKind::InvalidData, + os_code: None + }) + ); + assert_eq!( + terminal.output_collection, + CommandOutputCollection::Complete + ); + } else { + assert_eq!(terminal.status, CommandStatus::Completed); + assert_eq!(terminal.exit_code, Some(0)); + let stage = match (stream, after_exit) { + (CommandStream::Stdout, false) => CommandFailureStage::StdoutRead, + (CommandStream::Stderr, false) => CommandFailureStage::StderrRead, + (CommandStream::Stdout, true) => CommandFailureStage::StdoutDrain, + (CommandStream::Stderr, true) => CommandFailureStage::StderrDrain, + }; + assert_eq!( + terminal.output_collection, + CommandOutputCollection::Incomplete { + failure: CommandFailure { + stage, + reason: CommandIoErrorKind::PermissionDenied, + os_code: None + } + } + ); + } + let wire = serde_json::to_string(&terminal).unwrap(); + for secret in [ + "/private", + "TOKEN", + "secret", + "command-body", + "fixture-token", + ] { + assert!(!wire.contains(secret)); + } + session.close().await.unwrap(); + } + }) + .await + .expect("bounded loopback fault projection"); +} diff --git a/crates/workdir/src/operation.rs b/crates/workdir/src/operation.rs index 7e180e7aa..8cd671b53 100644 --- a/crates/workdir/src/operation.rs +++ b/crates/workdir/src/operation.rs @@ -132,12 +132,151 @@ pub enum CommandOutputTruncation { ByteBudget, } -/// Raw totals include discarded pipe bytes. Delivered totals count only prefixes -/// admitted to the transport queue, not page reads or retries by a consumer. +/// Closed diagnostic vocabulary: never serialize OS messages, paths, or errno text. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum CommandIoErrorKind { + NotFound, + PermissionDenied, + ConnectionRefused, + ConnectionReset, + ConnectionAborted, + NotConnected, + BrokenPipe, + AlreadyExists, + WouldBlock, + InvalidInput, + InvalidData, + TimedOut, + WriteZero, + Interrupted, + UnexpectedEof, + OutOfMemory, + HostUnreachable, + NetworkUnreachable, + AddrInUse, + AddrNotAvailable, + NetworkDown, + NotADirectory, + IsADirectory, + DirectoryNotEmpty, + ReadOnlyFilesystem, + StaleNetworkFileHandle, + StorageFull, + NotSeekable, + QuotaExceeded, + FileTooLarge, + ResourceBusy, + ExecutableFileBusy, + Deadlock, + CrossesDevices, + TooManyLinks, + InvalidFilename, + ArgumentListTooLong, + Unsupported, + Other, + Unknown, +} + +impl From for CommandIoErrorKind { + fn from(kind: std::io::ErrorKind) -> Self { + use std::io::ErrorKind as Io; + match kind { + Io::NotFound => Self::NotFound, + Io::PermissionDenied => Self::PermissionDenied, + Io::ConnectionRefused => Self::ConnectionRefused, + Io::ConnectionReset => Self::ConnectionReset, + Io::ConnectionAborted => Self::ConnectionAborted, + Io::NotConnected => Self::NotConnected, + Io::BrokenPipe => Self::BrokenPipe, + Io::AlreadyExists => Self::AlreadyExists, + Io::WouldBlock => Self::WouldBlock, + Io::InvalidInput => Self::InvalidInput, + Io::InvalidData => Self::InvalidData, + Io::TimedOut => Self::TimedOut, + Io::WriteZero => Self::WriteZero, + Io::Interrupted => Self::Interrupted, + Io::UnexpectedEof => Self::UnexpectedEof, + Io::OutOfMemory => Self::OutOfMemory, + Io::HostUnreachable => Self::HostUnreachable, + Io::NetworkUnreachable => Self::NetworkUnreachable, + Io::AddrInUse => Self::AddrInUse, + Io::AddrNotAvailable => Self::AddrNotAvailable, + Io::NetworkDown => Self::NetworkDown, + Io::NotADirectory => Self::NotADirectory, + Io::IsADirectory => Self::IsADirectory, + Io::DirectoryNotEmpty => Self::DirectoryNotEmpty, + Io::ReadOnlyFilesystem => Self::ReadOnlyFilesystem, + Io::StaleNetworkFileHandle => Self::StaleNetworkFileHandle, + Io::StorageFull => Self::StorageFull, + Io::NotSeekable => Self::NotSeekable, + Io::QuotaExceeded => Self::QuotaExceeded, + Io::FileTooLarge => Self::FileTooLarge, + Io::ResourceBusy => Self::ResourceBusy, + Io::ExecutableFileBusy => Self::ExecutableFileBusy, + Io::Deadlock => Self::Deadlock, + Io::CrossesDevices => Self::CrossesDevices, + Io::TooManyLinks => Self::TooManyLinks, + Io::InvalidFilename => Self::InvalidFilename, + Io::ArgumentListTooLong => Self::ArgumentListTooLong, + Io::Unsupported => Self::Unsupported, + Io::Other => Self::Other, + _ => Self::Unknown, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum CommandFailureStage { + Spawn, + Wait, + StdoutRead, + StderrRead, + StdoutDrain, + StderrDrain, + OutputDelivery, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct CommandFailure { + pub stage: CommandFailureStage, + pub reason: CommandIoErrorKind, + /// Observed numeric OS code, if supplied by the source. Never inferred from text. + pub os_code: Option, +} + +impl CommandFailure { + pub(crate) fn io(stage: CommandFailureStage, error: &std::io::Error) -> Self { + Self { + stage, + reason: error.kind().into(), + os_code: error.raw_os_error(), + } + } +} + +/// EOF collection is independent of exit status and the explicit byte budget. +/// Incomplete counts are lower bounds on bytes actually produced by the process. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)] +pub enum CommandOutputCollection { + NotStarted, + Complete, + Incomplete { failure: CommandFailure }, +} + +/// Raw totals include discarded *observed* pipe bytes, not an unknown failed tail. +/// Delivered totals count only prefixes admitted to the transport queue, not page +/// reads or retries. Neither terminal arrival nor exit zero implies EOF collection. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub struct CommandTerminal { pub status: CommandStatus, pub exit_code: Option, + pub process_failure: Option, + pub output_collection: CommandOutputCollection, pub stdout_bytes: u64, pub stderr_bytes: u64, pub stdout_delivered_bytes: u64, @@ -145,6 +284,54 @@ pub struct CommandTerminal { pub truncation_reason: Option, } +impl CommandTerminal { + /// Validate provider-authored metadata before publishing it as collected evidence. + pub fn metadata_valid(&self) -> bool { + let truncated = self.stdout_bytes > self.stdout_delivered_bytes + || self.stderr_bytes > self.stderr_delivered_bytes; + if self.status == CommandStatus::Running + || self.stdout_delivered_bytes > self.stdout_bytes + || self.stderr_delivered_bytes > self.stderr_bytes + || self.truncation_reason != truncated.then_some(CommandOutputTruncation::ByteBudget) + { + return false; + } + match self.process_failure { + Some(CommandFailure { + stage: CommandFailureStage::Spawn, + .. + }) => { + self.status == CommandStatus::Failed + && self.exit_code.is_none() + && self.output_collection == CommandOutputCollection::NotStarted + && self.stdout_bytes == 0 + && self.stderr_bytes == 0 + } + Some(CommandFailure { + stage: CommandFailureStage::Wait, + .. + }) => { + self.status != CommandStatus::Completed + && self.exit_code.is_none() + && self.collection_started() + } + Some(_) => false, + None => self.collection_started(), + } + } + + fn collection_started(&self) -> bool { + match self.output_collection { + CommandOutputCollection::NotStarted => false, + CommandOutputCollection::Complete => true, + CommandOutputCollection::Incomplete { failure } => !matches!( + failure.stage, + CommandFailureStage::Spawn | CommandFailureStage::Wait + ), + } + } +} + /// Sending next_cursor acknowledges this page. Only the current page may be /// retried; acknowledged bytes expire. Terminal fences all captured bytes, /// including on timeout/cancel. Consumers must pull while the process runs. @@ -186,19 +373,13 @@ impl CommandOutput { self.next_cursor <= MAX_COMMAND_TRANSPORT_BYTES && end.is_none_or(|end| end == self.next_cursor) && self.terminal.as_ref().is_none_or(|terminal| { - let truncated = terminal.stdout_bytes > terminal.stdout_delivered_bytes - || terminal.stderr_bytes > terminal.stderr_delivered_bytes; - stream_ends[0].is_none_or(|end| end == terminal.stdout_delivered_bytes) + terminal.metadata_valid() + && stream_ends[0].is_none_or(|end| end == terminal.stdout_delivered_bytes) && stream_ends[1].is_none_or(|end| end == terminal.stderr_delivered_bytes) - && terminal.status != CommandStatus::Running - && terminal.stdout_delivered_bytes <= terminal.stdout_bytes - && terminal.stderr_delivered_bytes <= terminal.stderr_bytes && terminal .stdout_delivered_bytes .checked_add(terminal.stderr_delivered_bytes) == Some(self.next_cursor) - && terminal.truncation_reason - == truncated.then_some(CommandOutputTruncation::ByteBudget) }) } } @@ -262,4 +443,70 @@ mod tests { .contains(&serde_json::json!("transport_byte_budget")) ); } + + #[test] + fn failure_wire_is_closed_and_does_not_infer_os_codes_from_custom_messages() { + let error = std::io::Error::new( + std::io::ErrorKind::Other, + "/host/private TOKEN=secret EACCES command", + ); + let failure = CommandFailure::io(CommandFailureStage::Wait, &error); + assert_eq!( + serde_json::to_value(failure).unwrap(), + serde_json::json!({"stage":"wait","reason":"other","os_code":null}) + ); + let os_error = std::io::Error::from_raw_os_error(libc::EIO); + let actual = CommandFailure::io(CommandFailureStage::StdoutDrain, &os_error); + assert_eq!(actual.os_code, Some(libc::EIO)); + for invalid in [ + serde_json::json!({"stage":"wait","reason":"/host/private TOKEN=secret"}), + serde_json::json!({"stage":"/host/private","reason":"other"}), + serde_json::json!({"stage":"wait","reason":"other","message":"secret"}), + ] { + assert!(serde_json::from_value::(invalid).is_err()); + } + } + + #[test] + fn provider_terminal_rejects_inconsistent_failure_domains_and_unstarted_byte_claims() { + let wire = serde_json::json!({"status":"completed","exit_code":0,"process_failure":null,"output_collection":{"state":"incomplete","failure":{"stage":"stdout_drain","reason":"other","os_code":null}},"stdout_bytes":3,"stderr_bytes":0,"stdout_delivered_bytes":3,"stderr_delivered_bytes":0,"truncation_reason":null}); + let terminal: CommandTerminal = serde_json::from_value(wire.clone()).unwrap(); + assert!(terminal.metadata_valid()); + for (field, value) in [ + ( + "output_collection", + serde_json::json!({"state":"not_started"}), + ), + ( + "output_collection", + serde_json::json!({"state":"incomplete","failure":{"stage":"wait","reason":"other"}}), + ), + ( + "process_failure", + serde_json::json!({"stage":"stdout_read","reason":"other"}), + ), + ( + "process_failure", + serde_json::json!({"stage":"spawn","reason":"not_found"}), + ), + ("truncation_reason", serde_json::json!("byte_budget")), + ] { + let mut invalid = wire.clone(); + invalid[field] = value; + assert!( + !serde_json::from_value::(invalid) + .unwrap() + .metadata_valid() + ); + } + let mut unclassified = wire.clone(); + unclassified + .as_object_mut() + .unwrap() + .remove("output_collection"); + assert!(serde_json::from_value::(unclassified).is_err()); + let mut raw = wire; + raw["message"] = serde_json::json!("/private TOKEN=secret"); + assert!(serde_json::from_value::(raw).is_err()); + } } diff --git a/crates/worker-runtime/Cargo.toml b/crates/worker-runtime/Cargo.toml index 0eba6a589..9b1c4ab49 100644 --- a/crates/worker-runtime/Cargo.toml +++ b/crates/worker-runtime/Cargo.toml @@ -56,6 +56,7 @@ server-api = { path = "../server-api" } workdir.workspace = true [dev-dependencies] +tools.workspace = true workdir = { workspace = true, features = ["http-client"] } futures.workspace = true agen.workspace = true diff --git a/crates/worker-runtime/src/http_server/tests/command_transport_tests.rs b/crates/worker-runtime/src/http_server/tests/command_transport_tests.rs index 902694621..fb4c3ffd7 100644 --- a/crates/worker-runtime/src/http_server/tests/command_transport_tests.rs +++ b/crates/worker-runtime/src/http_server/tests/command_transport_tests.rs @@ -3,6 +3,7 @@ //! pre-seeded session is a fixture; no Runtime Worker/materializer is claimed here. use super::*; +use std::collections::BTreeMap; use std::time::Duration; use workdir::http::{ RemoteWorkdirSession, WorkdirHttpAuthorization, WorkdirHttpAuthorizationFuture, @@ -70,14 +71,23 @@ impl Harness { } async fn new() -> Self { - let root = tempfile::tempdir().unwrap(); - let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound( - Workdir::new("wd-transport"), - root.path().to_path_buf(), - root.path().to_path_buf(), - SharedScope::new(Scope::writable(root.path()).unwrap()), - WorkdirSessionCapabilities::ALL, - )); + Self::with_environment(tempfile::tempdir().unwrap(), BTreeMap::new()).await + } + + async fn with_environment( + root: tempfile::TempDir, + command_environment: BTreeMap, + ) -> Self { + let session: WorkdirSessionHandle = + Arc::new(LocalWorkdirSession::materialized_bound_with_environment( + Workdir::new("wd-transport"), + root.path().to_path_buf(), + root.path().to_path_buf(), + SharedScope::new(Scope::writable(root.path()).unwrap()), + WorkdirSessionCapabilities::ALL, + command_environment, + Vec::new(), + )); let identity = RuntimeIdentityMaterial::generate("transport-key").unwrap(); let trust = workspace_trust(&identity); let state = RuntimeHttpState { @@ -222,6 +232,234 @@ async fn real_http_waiting_command_survives_long_poll_expiry() { .expect("bounded long-poll continuation over HTTP"); } +#[tokio::test] +async fn real_http_spawn_failure_preserves_safe_cause_counts_and_terminal_replay() { + tokio::time::timeout(Duration::from_secs(10), async { + let root = tempfile::tempdir().unwrap(); + let missing_path = root.path().join("private-spawn-path-must-not-exist"); + assert!(!missing_path.exists()); + let private_path = missing_path.to_str().unwrap().to_owned(); + let environment_secret = "spawn-environment-secret-must-not-leak"; + let command_secret = "spawn-command-secret-must-not-leak"; + let tool_call_secret = "spawn-tool-call-secret-must-not-leak"; + // This is an OS spawn failure in the production local provider, not a + // fabricated terminal: PATH contains no bash. The test process environment + // is untouched, so parallel command tests retain their normal PATH. + let harness = Harness::with_environment( + root, + BTreeMap::from([ + ("PATH".into(), private_path.clone()), + ("PRIVATE_AUTH_TOKEN".into(), environment_secret.into()), + ]), + ) + .await; + let command = format!("printf '{command_secret}'; printf ran > must-not-run"); + let handle = harness + .remote + .start_command(CommandRequest { + transport_byte_budget: 1024, + command: command.clone(), + timeout_secs: 5, + cwd: WorkdirPath::root(), + tool_call_id: Some(tool_call_secret.into()), + }) + .await + .unwrap(); + let request = CommandOutputRequest { + handle, + cursor: 0, + limit: 16, + wait: true, + }; + let page = harness.remote.command_output(request.clone()).await.unwrap(); + assert!(page.chunks.is_empty()); + assert_eq!(page.next_cursor, 0); + let terminal = page.terminal.as_ref().expect("spawn failure is terminal"); + assert_eq!(terminal.status, workdir::CommandStatus::Failed); + assert_eq!(terminal.exit_code, None); + assert_eq!( + terminal.process_failure, + Some(workdir::CommandFailure { + stage: workdir::CommandFailureStage::Spawn, + reason: workdir::CommandIoErrorKind::NotFound, + os_code: Some(libc::ENOENT), + }) + ); + assert_eq!( + terminal.output_collection, + workdir::CommandOutputCollection::NotStarted + ); + assert_eq!(terminal.stdout_bytes, 0); + assert_eq!(terminal.stderr_bytes, 0); + assert_eq!(terminal.stdout_delivered_bytes, 0); + assert_eq!(terminal.stderr_delivered_bytes, 0); + assert_eq!(terminal.truncation_reason, None); + + // A zero-byte terminal page remains replayable at its unchanged cursor; + // neither a retry nor a smaller nonwaiting read changes its accounting. + assert_eq!( + harness.remote.command_output(request.clone()).await.unwrap(), + page + ); + let replay_request = CommandOutputRequest { + wait: false, + limit: 1, + ..request + }; + assert_eq!( + harness + .remote + .command_output(replay_request.clone()) + .await + .unwrap(), + page + ); + let response = harness + .raw_operation( + serde_json::to_value(WorkdirSessionOperation::CommandOutput( + replay_request.clone(), + )) + .unwrap(), + ) + .await; + assert_eq!(response.status(), reqwest::StatusCode::OK); + let body = response.text().await.unwrap(); + for secret in [ + private_path.as_str(), + harness._root.path().to_str().unwrap(), + "private-spawn-path-must-not-exist", + "PRIVATE_AUTH_TOKEN", + environment_secret, + command_secret, + command.as_str(), + tool_call_secret, + ] { + assert!( + !body.contains(secret), + "private input leaked into HTTP terminal" + ); + } + let wire: serde_json::Value = serde_json::from_str(&body).unwrap(); + assert_eq!( + wire, + serde_json::json!({ + "operation": "command_output", + "result": { + "chunks": [], + "next_cursor": 0, + "terminal": { + "status": "failed", + "exit_code": null, + "process_failure": {"stage": "spawn", "reason": "not_found", "os_code": libc::ENOENT}, + "output_collection": {"state": "not_started"}, + "stdout_bytes": 0, + "stderr_bytes": 0, + "stdout_delivered_bytes": 0, + "stderr_delivered_bytes": 0, + "truncation_reason": null + } + } + }) + ); + assert_eq!( + harness.remote.command_output(replay_request).await.unwrap(), + page + ); + // Exercise the same real spawn failure through Bash, not a terminal + // fixture. Opening another remote client reuses the pre-seeded provider + // session and still passes through the signed HTTP middleware. + let bash_remote: WorkdirSessionHandle = Arc::new( + RemoteWorkdirSession::open_with_authorization( + harness.client.clone(), + harness.operation_url.join("/").unwrap(), + harness.authorization.clone(), + Workdir::new("wd-transport").id().clone(), + OpenWorkdirSessionRequest::default(), + ) + .await + .unwrap(), + ); + let view = tools::bash_output::BashOutputView::new(); + let (_, bash) = tools::bash_tool(bash_remote, view.clone())(); + let input = serde_json::json!({ + "command": command, + "cwd": ".", + "timeout": 5 + }) + .to_string(); + let output = bash + .execute( + &input, + agen::tool::ToolExecutionContext::new(tool_call_secret, "http-spawn-attempt", 0), + ) + .await + .expect("Bash returns a structured failed-command result"); + let content = output.content.as_deref().expect("Bash ToolOutput body"); + let tool_body: serde_json::Value = serde_json::from_str(content).unwrap(); + assert_eq!(tool_body["stdout_preview"], ""); + assert_eq!(tool_body["stderr_preview"], ""); + assert_eq!(tool_body["preview_omitted"], false); + assert_eq!( + tool_body["output"]["terminal"], + serde_json::json!({ + "status": "failed", + "exit_code": null, + "process_failure": {"stage": "spawn", "reason": "not_found", "os_code": libc::ENOENT}, + "output_collection": {"state": "not_started"}, + "transport_complete": false + }) + ); + let output_id = tool_body["output"]["id"].as_str().unwrap(); + let retained = view.metadata(output_id).expect("retained spawn failure"); + assert_eq!(retained.terminal.status, Some(workdir::CommandStatus::Failed)); + assert_eq!(retained.terminal.exit_code, None); + assert_eq!(retained.terminal.process_failure, terminal.process_failure); + assert_eq!( + retained.terminal.output_collection, + Some(workdir::CommandOutputCollection::NotStarted) + ); + assert!(!retained.terminal.transport_complete); + assert_eq!(retained.stdout_saved, 0); + assert_eq!(retained.stderr_saved, 0); + assert_eq!(retained.stdout_seen, 0); + assert_eq!(retained.stderr_seen, 0); + assert_eq!(retained.stdout_delivered, 0); + assert_eq!(retained.stderr_delivered, 0); + assert!(!retained.storage_truncated); + assert!(retained.retention_limit.is_none()); + assert_eq!(retained.transport_truncation, None); + let retained_wire = serde_json::to_value(&retained).unwrap(); + assert_eq!(retained_wire, tool_body["output"]); + for stream in [CommandStream::Stdout, CommandStream::Stderr] { + assert_eq!(view.read(output_id, stream, 0, 1), Some(Vec::new())); + } + for secret in [ + private_path.as_str(), + harness._root.path().to_str().unwrap(), + "private-spawn-path-must-not-exist", + "PRIVATE_AUTH_TOKEN", + environment_secret, + command_secret, + command.as_str(), + tool_call_secret, + ] { + assert!( + !content.contains(secret), + "private input leaked into Bash ToolOutput body" + ); + } + assert!(!harness._root.path().join("must-not-run").exists()); + harness.remote.close().await.unwrap(); + assert!(harness.session.command_snapshot().is_empty()); + assert_eq!( + serde_json::to_value(view.metadata(output_id).unwrap()).unwrap(), + retained_wire + ); + }) + .await + .expect("bounded real spawn failure over signed HTTP"); +} + #[tokio::test] async fn real_http_closed_session_retains_failure_reason() { let harness = Harness::new().await; @@ -291,6 +529,11 @@ async fn real_http_command_pages_preserve_binary_streams_replay_and_nonzero_exit assert_eq!(stderr, b"stderr\xe6\x97\xa5"); assert_eq!(terminal.status, workdir::CommandStatus::Failed); assert_eq!(terminal.exit_code, Some(9)); + assert_eq!(terminal.process_failure, None); + assert_eq!( + terminal.output_collection, + workdir::CommandOutputCollection::Complete + ); assert!(harness.remote.command_output(CommandOutputRequest { handle, cursor: 0, limit: 1, wait: false }).await.is_err()); harness.remote.close().await.unwrap(); assert!(harness.session.command_snapshot().is_empty()); @@ -435,6 +678,11 @@ async fn real_http_command_budget_bounds_unique_payload_and_reports_larger_seen_ ); assert_eq!(terminal.status, workdir::CommandStatus::Failed); assert_eq!(terminal.exit_code, Some(9)); + assert_eq!(terminal.process_failure, None); + assert_eq!( + terminal.output_collection, + workdir::CommandOutputCollection::Complete + ); assert!( harness .remote diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 76d394da4..a1b36997f 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -38045,6 +38045,8 @@ mod tests { chunks: Vec::new(), next_cursor: 0, terminal: Some(workdir::CommandTerminal { + process_failure: None, + output_collection: workdir::CommandOutputCollection::Complete, status: workdir::CommandStatus::Cancelled, exit_code: None, @@ -38147,6 +38149,8 @@ mod tests { chunks: Vec::new(), next_cursor: 0, terminal: Some(workdir::CommandTerminal { + process_failure: None, + output_collection: workdir::CommandOutputCollection::Complete, status: workdir::CommandStatus::Cancelled, exit_code: None, @@ -39239,6 +39243,8 @@ mod tests { chunks: Vec::new(), next_cursor: 0, terminal: Some(workdir::CommandTerminal { + process_failure: None, + output_collection: workdir::CommandOutputCollection::Complete, status: workdir::CommandStatus::Cancelled, exit_code: None, diff --git a/crates/workspace-server/src/server/tests/command_transport_tests.rs b/crates/workspace-server/src/server/tests/command_transport_tests.rs index af9d6416b..e27e764f0 100644 --- a/crates/workspace-server/src/server/tests/command_transport_tests.rs +++ b/crates/workspace-server/src/server/tests/command_transport_tests.rs @@ -39,6 +39,8 @@ fn page(cursor: u64, bytes: &[u8], terminal: bool) -> workdir::CommandOutput { }, next_cursor: cursor + bytes.len() as u64, terminal: terminal.then_some(workdir::CommandTerminal { + process_failure: None, + output_collection: workdir::CommandOutputCollection::Complete, status: workdir::CommandStatus::Cancelled, exit_code: None, stdout_bytes: cursor + bytes.len() as u64,