From 1b6d4dbfd815b13b8d9081c2bb912a340bbe45ff Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 21 Oct 2026 01:22:04 +0900 Subject: [PATCH 1/6] fix: preserve workdir failure diagnostics across transport --- crates/server-api/src/lib.rs | 8 ++ crates/tools/src/bash.rs | 14 +-- crates/tools/src/file_target.rs | 4 +- crates/tools/tests/integration.rs | 7 +- crates/workdir/src/external.rs | 6 ++ crates/workdir/src/http.rs | 83 ++++++++++++++++-- crates/workdir/src/lib.rs | 56 +++++++++++- crates/workdir/src/local.rs | 18 ++-- crates/workdir/src/local/command_transport.rs | 1 + .../src/local/command_transport/tests.rs | 32 +++++++ crates/workdir/src/router.rs | 2 +- crates/workdir/tests/failure_diagnostics.rs | 85 +++++++++++++++++++ crates/worker-runtime/src/http_server.rs | 11 +++ .../tests/command_transport_tests.rs | 36 ++++++++ crates/worker/src/checkout_http_tests.rs | 1 + .../src/feature/builtin/manage_workdir.rs | 17 +++- crates/worker/tests/controller_test.rs | 4 +- crates/workspace-server/src/server.rs | 23 +++++ .../src/server/tests/api_logging_tests.rs | 59 +++++++++++++ crates/yoi/src/workdir_share.rs | 2 + 20 files changed, 440 insertions(+), 29 deletions(-) create mode 100644 crates/workdir/tests/failure_diagnostics.rs diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index 1de9be8e8..3b448d409 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -132,6 +132,8 @@ pub struct WorkdirOperationApiError { pub workdir_log: Option, #[serde(skip)] pub denial_reason: Option, + #[serde(skip)] + pub failure_reason: Option, } /// Internal operational correlation. Producers supply fixed-length ID fingerprints and @@ -157,6 +159,7 @@ impl WorkdirOperationApiError { status: error.status, workdir_log: None, denial_reason: None, + failure_reason: None, } } @@ -170,6 +173,7 @@ impl WorkdirOperationApiError { status, workdir_log: None, denial_reason: error.denial_reason, + failure_reason: error.failure_reason, } } } @@ -15047,10 +15051,12 @@ mod workdir_error_diagnostic_tests { #[test] fn workdir_internal_diagnostics_never_enter_public_error_body_or_schema() { let mut error = WorkdirOperationApiError::provider(workdir::http::WorkdirTransportError { + failure_reason: None, code: workdir::http::WorkdirTransportErrorCode::Denied, message: "Workdir operation was denied".to_string(), denial_reason: Some(workdir::WorkdirDenialReason::OsPermissionDenied), }); + error.failure_reason = Some(workdir::WorkdirFailureReason::HttpTimeout); error.workdir_log = Some(WorkdirOperationLogContext { operation: "write", workspace_id_hash: "internal-workspace-fingerprint".to_string(), @@ -15064,10 +15070,12 @@ mod workdir_error_diagnostic_tests { assert_eq!(api_macros::HttpError::status_code(&error), 403); let decoded: WorkdirOperationApiError = serde_json::from_value(wire).unwrap(); assert!(decoded.denial_reason.is_none()); + assert!(decoded.failure_reason.is_none()); assert!(decoded.workdir_log.is_none()); let schema = serde_json::to_value(schemars::schema_for!(WorkdirOperationApiError)).unwrap(); assert!(schema["properties"].get("workdir_log").is_none()); assert!(schema["properties"].get("denial_reason").is_none()); + assert!(schema["properties"].get("failure_reason").is_none()); } } diff --git a/crates/tools/src/bash.rs b/crates/tools/src/bash.rs index 74dcee471..efeaa3acf 100644 --- a/crates/tools/src/bash.rs +++ b/crates/tools/src/bash.rs @@ -158,11 +158,15 @@ async fn command_page( ) -> Result { for attempt in 0..3 { match session.command_output(request.clone()).await { - Err( - workdir::WorkdirError::Transport(_) - | workdir::WorkdirError::Unavailable(_) - | workdir::WorkdirError::OutcomeUnknown(_), - ) if attempt < 2 => { + Err(error) + if attempt < 2 + && matches!( + error.classification_source(), + workdir::WorkdirError::Transport(_) + | workdir::WorkdirError::Unavailable(_) + | workdir::WorkdirError::OutcomeUnknown(_) + ) => + { tokio::time::sleep(std::time::Duration::from_millis(100 << attempt)).await; } result => return result, diff --git a/crates/tools/src/file_target.rs b/crates/tools/src/file_target.rs index f090431ac..c3bfb535e 100644 --- a/crates/tools/src/file_target.rs +++ b/crates/tools/src/file_target.rs @@ -147,7 +147,9 @@ impl FileTarget { pub(crate) fn checked_error(error: workdir::WorkdirError) -> ToolError { use workdir::WorkdirError; let code = match error.classification_source() { - WorkdirError::DenialContext { .. } => unreachable!("diagnostic context unwrapped"), + WorkdirError::DenialContext { .. } | WorkdirError::FailureContext { .. } => { + unreachable!("diagnostic context unwrapped") + } WorkdirError::Conflict(_) | WorkdirError::NotFound(_) => "checkout_stale", WorkdirError::Denied { .. } | WorkdirError::OutOfScope { .. } diff --git a/crates/tools/tests/integration.rs b/crates/tools/tests/integration.rs index accfb59ef..f7b1dd2da 100644 --- a/crates/tools/tests/integration.rs +++ b/crates/tools/tests/integration.rs @@ -546,9 +546,10 @@ impl workdir::WorkdirSession for ExternalValidatingSession { .is_ok() { // The production provider has committed a page, but its response is lost. - return Err(workdir::WorkdirError::Transport( - "injected lost response".into(), - )); + return Err( + workdir::WorkdirError::Transport("injected lost response".into()) + .with_failure_reason(workdir::WorkdirFailureReason::HttpTimeout), + ); } Ok(output) } diff --git a/crates/workdir/src/external.rs b/crates/workdir/src/external.rs index 867d24498..c1c00bfd8 100644 --- a/crates/workdir/src/external.rs +++ b/crates/workdir/src/external.rs @@ -525,6 +525,8 @@ pub struct ExternalWorkdirOperationError { code: crate::http::WorkdirTransportErrorCode, #[serde(default, skip_serializing_if = "Option::is_none")] denial_reason: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + failure_reason: Option, } impl ExternalWorkdirOperationError { @@ -534,11 +536,13 @@ impl ExternalWorkdirOperationError { Self { code: error.code, denial_reason: error.denial_reason, + failure_reason: error.failure_reason, } } pub fn into_transport_error(self) -> WorkdirTransportError { WorkdirTransportError { + failure_reason: self.failure_reason, code: self.code, denial_reason: self.denial_reason, message: format!("External Workdir provider reported {}", self.code.as_str()), @@ -762,6 +766,7 @@ mod tests { ] { let external = ExternalWorkdirOperationError::from_transport_error(WorkdirTransportError { + failure_reason: None, code: Code::Denied, message: "/home/operator/private provider-secret".repeat(4096), denial_reason: reason, @@ -788,6 +793,7 @@ mod tests { #[test] fn provider_failures_are_closed_and_drop_provider_authored_diagnostics() { let error = ExternalWorkdirOperationError::from_transport_error(WorkdirTransportError { + failure_reason: None, code: crate::http::WorkdirTransportErrorCode::Internal, denial_reason: None, message: "/home/operator/private provider-secret\ncontrol".to_string(), diff --git a/crates/workdir/src/http.rs b/crates/workdir/src/http.rs index 7464943da..8a035395c 100644 --- a/crates/workdir/src/http.rs +++ b/crates/workdir/src/http.rs @@ -275,6 +275,8 @@ pub struct WorkdirTransportError { /// Internal provider diagnostics. Public response boundaries may omit this. #[serde(default, skip_serializing_if = "Option::is_none")] pub denial_reason: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub failure_reason: Option, } impl WorkdirTransportError { @@ -282,6 +284,11 @@ impl WorkdirTransportError { pub fn from_workdir_error(error: &WorkdirError) -> Self { use WorkdirTransportErrorCode as Code; let (code, message) = match error { + WorkdirError::FailureContext { reason, source } => { + let mut transport = Self::from_workdir_error(source); + transport.failure_reason = Some(*reason); + return transport; + } WorkdirError::DenialContext { reason, source } => { let mut transport = Self::from_workdir_error(source); transport.denial_reason = Some(*reason); @@ -301,6 +308,7 @@ impl WorkdirTransportError { code: Code::Unsupported, message: format!("Workdir capability {capability:?} is not available"), denial_reason: None, + failure_reason: None, }; } WorkdirError::UnsupportedOperation(_) => { @@ -359,6 +367,7 @@ impl WorkdirTransportError { code, message: message.to_string(), denial_reason: error.denial_reason(), + failure_reason: error.failure_reason(), } } @@ -408,6 +417,11 @@ impl WorkdirTransportError { Code::Transport => WorkdirError::Transport(self.message), Code::Internal => WorkdirError::OperationFailed, }; + let error = if let Some(reason) = self.failure_reason { + error.with_failure_reason(reason) + } else { + error + }; // Preserve any supplied reason independently of the public classification. // Without a reason, retain the existing classification's deliberate fallback. if let Some(reason) = self.denial_reason { @@ -557,9 +571,10 @@ mod client { operation: WorkdirSessionOperation, ) -> Result { if self.closed.load(Ordering::Acquire) { - return Err(WorkdirError::Unavailable( - "Workdir session is closed".to_string(), - )); + return Err( + WorkdirError::Unavailable("Workdir session is closed".to_string()) + .with_failure_reason(crate::WorkdirFailureReason::SessionClosed), + ); } let url = endpoint( &self.base_url, @@ -767,6 +782,10 @@ mod client { let outcome = WorkdirError::OutcomeUnknown( "checkout transport failed; effects may have occurred".into(), ); + let outcome = match error.failure_reason() { + Some(reason) => outcome.with_failure_reason(reason), + None => outcome, + }; Err(match error.denial_reason() { Some(reason) => WorkdirError::DenialContext { reason, @@ -966,8 +985,10 @@ mod client { ) -> Result { if response.status().is_success() { let bytes = bounded_response_bytes(response).await?; - serde_json::from_slice(&bytes) - .map_err(|_| WorkdirError::Transport("Runtime Workdir response was invalid".into())) + serde_json::from_slice(&bytes).map_err(|_| { + WorkdirError::Transport("Runtime Workdir response was invalid".into()) + .with_failure_reason(crate::WorkdirFailureReason::HttpResponseInvalid) + }) } else { Err(decode_error(response).await) } @@ -983,14 +1004,16 @@ mod client { { return Err(WorkdirError::Transport( "Runtime Workdir response exceeded transport bound".into(), - )); + ) + .with_failure_reason(crate::WorkdirFailureReason::HttpResponseTooLarge)); } let mut bytes = Vec::new(); while let Some(chunk) = response.chunk().await.map_err(http_unavailable)? { if chunk.len() > MAX_RESPONSE_BYTES.saturating_sub(bytes.len()) { return Err(WorkdirError::Transport( "Runtime Workdir response exceeded transport bound".into(), - )); + ) + .with_failure_reason(crate::WorkdirFailureReason::HttpResponseTooLarge)); } bytes.extend_from_slice(&chunk); } @@ -1003,13 +1026,22 @@ mod client { .map(WorkdirTransportError::into_workdir_error) .unwrap_or_else(|_| { WorkdirError::Transport("Runtime Workdir error response was invalid".into()) + .with_failure_reason(crate::WorkdirFailureReason::HttpResponseInvalid) }), Err(error) => error, } } - fn http_unavailable(_error: reqwest::Error) -> WorkdirError { + fn http_unavailable(error: reqwest::Error) -> WorkdirError { + let reason = if error.is_timeout() { + crate::WorkdirFailureReason::HttpTimeout + } else if error.is_connect() { + crate::WorkdirFailureReason::HttpConnect + } else { + crate::WorkdirFailureReason::HttpTransfer + }; WorkdirError::Transport("Runtime Workdir HTTP request failed".to_string()) + .with_failure_reason(reason) } #[cfg(test)] @@ -1017,6 +1049,40 @@ mod client { use super::*; use std::sync::Mutex; + #[tokio::test] + async fn http_failure_reasons_do_not_copy_urls_or_credentials() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let client = Client::builder().no_proxy().build().unwrap(); + // A listening socket that never replies makes a real HTTP timeout. + let error = client + .get(format!("http://{address}/private?token=secret")) + .timeout(std::time::Duration::from_millis(25)) + .send() + .await + .unwrap_err(); + assert!(error.is_timeout()); + let error = http_unavailable(error); + assert_eq!( + error.failure_reason(), + Some(crate::WorkdirFailureReason::HttpTimeout) + ); + assert!(!format!("{error:?}").contains("secret")); + assert!(!format!("{error:?}").contains("/private")); + drop(listener); + let error = client + .get(format!("http://{address}/private?token=secret")) + .timeout(std::time::Duration::from_secs(2)) + .send() + .await + .unwrap_err(); + assert!(error.is_connect()); + assert_eq!( + http_unavailable(error).failure_reason(), + Some(crate::WorkdirFailureReason::HttpConnect) + ); + } + #[derive(Debug, Default)] struct RejectAuthorization { requests: Arc)>>>, @@ -1193,6 +1259,7 @@ mod tests { (WorkdirTransportErrorCode::Internal, 500, "operation failed"), ] { let transport = WorkdirTransportError { + failure_reason: None, code, denial_reason: None, message: if code == WorkdirTransportErrorCode::Conflict { diff --git a/crates/workdir/src/lib.rs b/crates/workdir/src/lib.rs index 7f052d5be..1a515ac16 100644 --- a/crates/workdir/src/lib.rs +++ b/crates/workdir/src/lib.rs @@ -401,8 +401,44 @@ impl From<&str> for WorkdirDenial { } } +/// Bounded operational diagnostics, never provider text, paths or credentials. +/// Reasons describe an observation, not retry permission or proof of no effects. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum WorkdirFailureReason { + SessionClosed, + ProviderUnavailable, + CommandTerminalMissing, + HttpTimeout, + HttpConnect, + HttpTransfer, + HttpResponseInvalid, + HttpResponseTooLarge, +} + +impl WorkdirFailureReason { + pub const fn as_str(self) -> &'static str { + match self { + Self::SessionClosed => "session_closed", + Self::ProviderUnavailable => "provider_unavailable", + Self::CommandTerminalMissing => "command_terminal_missing", + Self::HttpTimeout => "http_timeout", + Self::HttpConnect => "http_connect", + Self::HttpTransfer => "http_transfer", + Self::HttpResponseInvalid => "http_response_invalid", + Self::HttpResponseTooLarge => "http_response_too_large", + } + } +} + #[derive(Debug, thiserror::Error)] pub enum WorkdirError { + #[error("{source}")] + FailureContext { + reason: WorkdirFailureReason, + #[source] + source: Box, + }, #[error("Workdir operation denied: {0}")] Denied(WorkdirDenial), @@ -521,12 +557,29 @@ impl WorkdirError { /// and `denial_reason`: enrichment must not alter behavior or retry policy. pub fn classification_source(&self) -> &Self { let mut error = self; - while let Self::DenialContext { source, .. } = error { + while let Self::DenialContext { source, .. } | Self::FailureContext { source, .. } = error { error = source; } error } + pub fn with_failure_reason(self, reason: WorkdirFailureReason) -> Self { + Self::FailureContext { + reason, + source: Box::new(self), + } + } + + pub fn failure_reason(&self) -> Option { + match self { + Self::FailureContext { reason, .. } => Some(*reason), + Self::DenialContext { source, .. } => source.failure_reason(), + Self::SessionClosed => Some(WorkdirFailureReason::SessionClosed), + Self::Unavailable(_) => Some(WorkdirFailureReason::ProviderUnavailable), + _ => None, + } + } + pub fn denied(reason: WorkdirDenialReason, message: impl Into) -> Self { Self::Denied(WorkdirDenial { reason: Some(reason), @@ -540,6 +593,7 @@ impl WorkdirError { match self { Self::Denied(denial) => denial.reason, Self::DenialContext { reason, .. } => Some(*reason), + Self::FailureContext { source, .. } => source.denial_reason(), Self::OutOfScope(_) => Some(Reason::PathOutOfScope), Self::SymlinkOutOfScope { .. } => Some(Reason::SymlinkTargetOutOfScope), Self::ReadOnly(_) => Some(Reason::PathReadOnly), diff --git a/crates/workdir/src/local.rs b/crates/workdir/src/local.rs index c84970b0f..344cab6a8 100644 --- a/crates/workdir/src/local.rs +++ b/crates/workdir/src/local.rs @@ -1210,7 +1210,8 @@ impl LocalWorkdirSession { Err(WorkdirError::Unavailable(format!( "Workdir session for {} is closed", self.inner.workdir.id() - ))) + )) + .with_failure_reason(crate::WorkdirFailureReason::SessionClosed)) } else { Ok(()) } @@ -2546,11 +2547,18 @@ mod tests { .expect("close should wake command output waiters") .unwrap() .unwrap_err(); - assert!(matches!(waiter_error, WorkdirError::Unavailable(_))); + assert!(matches!( + waiter_error.classification_source(), + WorkdirError::Unavailable(_) + )); + assert_eq!( + waiter_error.failure_reason(), + Some(crate::WorkdirFailureReason::SessionClosed) + ); assert!(matches!( WorkdirSession::command_status(&session, command).await, - Err(WorkdirError::Unavailable(_)) + Err(ref error) if matches!(error.classification_source(), WorkdirError::Unavailable(_)) )); assert!(matches!( WorkdirSession::read( @@ -2563,7 +2571,7 @@ mod tests { }, ) .await, - Err(WorkdirError::Unavailable(_)) + Err(ref error) if matches!(error.classification_source(), WorkdirError::Unavailable(_)) )); let restored = LocalWorkdirSession::materialized_bound( @@ -3531,7 +3539,7 @@ mod tests { .await .unwrap() .unwrap(), - Err(WorkdirError::Unavailable(message)) if message.contains("is closed") + Err(ref error) if matches!(error.classification_source(), WorkdirError::Unavailable(message) if message.contains("is closed")) )); assert!( session.command_snapshot().is_empty(), diff --git a/crates/workdir/src/local/command_transport.rs b/crates/workdir/src/local/command_transport.rs index 5b800c0ab..1af30b1fc 100644 --- a/crates/workdir/src/local/command_transport.rs +++ b/crates/workdir/src/local/command_transport.rs @@ -113,6 +113,7 @@ impl Transport { WorkdirError::Unavailable( "command transport ended before terminal fence".into(), ) + .with_failure_reason(crate::WorkdirFailureReason::CommandTerminalMissing) })?; self.terminal = Some(terminal.clone()); page.terminal = Some(terminal); diff --git a/crates/workdir/src/local/command_transport/tests.rs b/crates/workdir/src/local/command_transport/tests.rs index 796a1f93f..12f20b061 100644 --- a/crates/workdir/src/local/command_transport/tests.rs +++ b/crates/workdir/src/local/command_transport/tests.rs @@ -4,6 +4,38 @@ use super::*; use tempfile::TempDir; +#[tokio::test] +async fn missing_terminal_is_diagnosed_without_changing_error_classification() { + let (sender, receiver) = mpsc::channel(1); + let (_status, terminal_status) = watch::channel(None); + drop(sender); + let mut transport = Transport { + receiver, + pending: None, + cursor: 0, + replay: None, + terminal: None, + terminal_status, + }; + let error = transport + .read(&CommandOutputRequest { + handle: CommandHandle("diagnostic-test".into()), + cursor: 0, + limit: 16, + wait: true, + }) + .await + .unwrap_err(); + assert_eq!( + error.failure_reason(), + Some(crate::WorkdirFailureReason::CommandTerminalMissing) + ); + assert!(matches!( + error.classification_source(), + WorkdirError::Unavailable(_) + )); +} + struct Harness { session: LocalWorkdirSession, _root: TempDir, diff --git a/crates/workdir/src/router.rs b/crates/workdir/src/router.rs index 362504eab..8810b4435 100644 --- a/crates/workdir/src/router.rs +++ b/crates/workdir/src/router.rs @@ -842,7 +842,7 @@ mod tests { assert!(router.is_empty()); assert!(matches!( underlying.command_status(handle).await, - Err(WorkdirError::Unavailable(_)) + Err(ref error) if matches!(error.classification_source(), WorkdirError::Unavailable(_)) )); } diff --git a/crates/workdir/tests/failure_diagnostics.rs b/crates/workdir/tests/failure_diagnostics.rs new file mode 100644 index 000000000..a94f4fe9f --- /dev/null +++ b/crates/workdir/tests/failure_diagnostics.rs @@ -0,0 +1,85 @@ +use workdir::external::ExternalWorkdirOperationError; +use workdir::http::{WorkdirTransportError, WorkdirTransportErrorCode as Code}; +use workdir::{WorkdirError, WorkdirFailureReason as Reason}; + +#[test] +fn failure_reasons_survive_internal_transports_without_provider_text() { + for reason in [ + Reason::SessionClosed, + Reason::ProviderUnavailable, + Reason::CommandTerminalMissing, + Reason::HttpTimeout, + Reason::HttpConnect, + Reason::HttpTransfer, + Reason::HttpResponseInvalid, + Reason::HttpResponseTooLarge, + ] { + let source = WorkdirError::Unavailable("secret-token /private/path".into()) + .with_failure_reason(reason); + let transport = WorkdirTransportError::from_workdir_error(&source); + let wire = serde_json::to_string(&transport).unwrap(); + assert!(!wire.contains("secret-token")); + assert!(!wire.contains("/private/path")); + assert_eq!(transport.code, Code::Unavailable); + assert_eq!(transport.code.http_status(), 503); + assert_eq!(transport.message, "Workdir session is unavailable"); + assert_eq!(serde_json::to_value(reason).unwrap(), reason.as_str()); + let decoded: WorkdirTransportError = serde_json::from_str(&wire).unwrap(); + let external = ExternalWorkdirOperationError::from_transport_error(decoded); + let external_wire = serde_json::to_string(&external).unwrap(); + let external: ExternalWorkdirOperationError = serde_json::from_str(&external_wire).unwrap(); + let error = external.into_transport_error().into_workdir_error(); + assert_eq!(error.failure_reason(), Some(reason)); + assert!(matches!( + error.classification_source(), + WorkdirError::Unavailable(_) + )); + assert_eq!( + WorkdirTransportError::from_workdir_error(&error).failure_reason, + Some(reason) + ); + } +} + +#[test] +fn legacy_errors_and_nested_context_preserve_classification() { + let wire = r#"{"code":"unavailable","message":"Workdir session is unavailable"}"#; + let old: WorkdirTransportError = serde_json::from_str(wire).unwrap(); + assert!(old.failure_reason.is_none()); + assert!(matches!( + old.into_workdir_error(), + WorkdirError::Unavailable(_) + )); + let closed = WorkdirTransportError::from_workdir_error(&WorkdirError::SessionClosed); + assert_eq!(closed.failure_reason, Some(Reason::SessionClosed)); + // Reason-looking text never becomes a typed diagnosis. + let generic = WorkdirError::Unavailable("http_timeout".into()); + assert_eq!(generic.failure_reason(), Some(Reason::ProviderUnavailable)); + let error = WorkdirError::DenialContext { + reason: workdir::WorkdirDenialReason::OsPermissionDenied, + source: Box::new( + WorkdirError::OutcomeUnknown("do not retry".into()) + .with_failure_reason(Reason::HttpTransfer), + ), + }; + let error = WorkdirTransportError::from_workdir_error(&error).into_workdir_error(); + assert!(matches!( + error.classification_source(), + WorkdirError::OutcomeUnknown(_) + )); + assert_eq!( + error.denial_reason(), + Some(workdir::WorkdirDenialReason::OsPermissionDenied) + ); + assert_eq!(error.failure_reason(), Some(Reason::HttpTransfer)); +} + +#[test] +fn provider_cannot_inject_free_text_failure_reason() { + assert!( + serde_json::from_str::( + r#"{"code":"unavailable","message":"x","failure_reason":"secret /path"}"# + ) + .is_err() + ); +} diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 57b7d0eb6..a1faefc73 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -2146,6 +2146,7 @@ impl RuntimeHttpWorkdirError { Self { status, payload: WorkdirTransportError { + failure_reason: None, denial_reason: None, code, message: message.into(), @@ -2214,6 +2215,16 @@ impl From for RuntimeHttpWorkdirError { impl IntoResponse for RuntimeHttpWorkdirError { fn into_response(self) -> Response { + if let Some(reason) = self.payload.failure_reason { + tracing::warn!( + target: "worker_runtime::workdir", + event = "workdir_provider_failure", + status = self.status.as_u16(), + code = self.payload.code.as_str(), + failure_reason = reason.as_str(), + "Workdir provider operation failed" + ); + } (self.status, Json(self.payload)).into_response() } } 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 643e722ab..de1150159 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 @@ -156,6 +156,42 @@ impl Harness { } } +#[tokio::test] +async fn real_http_closed_session_retains_failure_reason() { + let harness = Harness::new().await; + let handle = harness + .remote + .start_command(CommandRequest { + transport_byte_budget: 1024, + command: "printf done".into(), + timeout_secs: 5, + cwd: WorkdirPath::root(), + tool_call_id: None, + }) + .await + .unwrap(); + // Keep the HTTP session identity registered, but close the underlying provider. + harness.session.close().await.unwrap(); + let error = harness + .remote + .command_output(CommandOutputRequest { + handle, + cursor: 0, + limit: 16, + wait: true, + }) + .await + .unwrap_err(); + assert_eq!( + error.failure_reason(), + Some(workdir::WorkdirFailureReason::SessionClosed) + ); + assert!(matches!( + error.classification_source(), + workdir::WorkdirError::Unavailable(_) + )); +} + #[tokio::test] async fn real_http_command_pages_preserve_binary_streams_replay_and_nonzero_exit() { tokio::time::timeout(Duration::from_secs(20), async { diff --git a/crates/worker/src/checkout_http_tests.rs b/crates/worker/src/checkout_http_tests.rs index 9c50f3d0f..d6489ae6c 100644 --- a/crates/worker/src/checkout_http_tests.rs +++ b/crates/worker/src/checkout_http_tests.rs @@ -57,6 +57,7 @@ impl ProviderState { return Err(( StatusCode::UNAUTHORIZED, AxumJson(WorkdirTransportError { + failure_reason: None, denial_reason: None, code: WorkdirTransportErrorCode::Denied, message: "provider authorization denied".into(), diff --git a/crates/worker/src/feature/builtin/manage_workdir.rs b/crates/worker/src/feature/builtin/manage_workdir.rs index ad35374e2..12cfdb65d 100644 --- a/crates/worker/src/feature/builtin/manage_workdir.rs +++ b/crates/worker/src/feature/builtin/manage_workdir.rs @@ -409,10 +409,20 @@ impl WorkdirSession for WorkspaceAttachedWorkdirSession { "mismatched checkout mutation response".into(), )), Ok(_) => Err(Self::mismatch("checkout_execute")), - Err(WorkdirError::Transport(_) | WorkdirError::Unavailable(_)) if mutation => { - Err(WorkdirError::OutcomeUnknown( + Err(error) + if mutation + && matches!( + error.classification_source(), + WorkdirError::Transport(_) | WorkdirError::Unavailable(_) + ) => + { + let outcome = WorkdirError::OutcomeUnknown( "checkout mutation response lost; inspect effects before retry".into(), - )) + ); + Err(match error.failure_reason() { + Some(reason) => outcome.with_failure_reason(reason), + None => outcome, + }) } Err(error) => Err(error), } @@ -1230,6 +1240,7 @@ mod tests { WorkspaceResponse { status, body: serde_json::to_string(&workdir::http::WorkdirTransportError { + failure_reason: None, denial_reason: None, code, message: message.to_string(), diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 077eadc99..52ebdc62c 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -480,7 +480,7 @@ async fn shutdown_closes_bound_workdir_session() { assert!(matches!( session.command_status(command).await, - Err(WorkdirError::Unavailable(_)) + Err(ref error) if matches!(error.classification_source(), WorkdirError::Unavailable(_)) )); } @@ -758,7 +758,7 @@ async fn controller_startup_failure_closes_bound_workdir_session() { tool_call_id: None, }) .await, - Err(WorkdirError::Unavailable(_)) + Err(ref error) if matches!(error.classification_source(), WorkdirError::Unavailable(_)) )); } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index a8ddde657..76d394da4 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1411,6 +1411,10 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession { let outcome = workdir::WorkdirError::OutcomeUnknown( "checkout provider response lost; inspect effects before retry".into(), ); + let outcome = match error.failure_reason() { + Some(reason) => outcome.with_failure_reason(reason), + None => outcome, + }; Err(match error.denial_reason() { Some(reason) => workdir::WorkdirError::DenialContext { reason, @@ -13501,6 +13505,7 @@ async fn log_failed_api_response(request: Request, next: Next) -> Response { attachment_alias_hash = context.map(|ctx| ctx.attachment_alias_hash.as_str()), request_token_id_hash = context.and(request_token_id_hash.as_deref()), denial_reason = error.and_then(|error| error.denial_reason).map(|reason| reason.as_str()), + failure_reason = error.and_then(|error| error.failure_reason).map(|reason| reason.as_str()), status = event.status, kind = event.kind.unwrap_or("unknown"), message = event.message.unwrap_or(""), @@ -20989,9 +20994,14 @@ impl InternalWorkdirOperationError { Error::WorkdirSessionOpenFailed { transport, .. } => transport.denial_reason, _ => None, }; + let failure_reason = match &error.error { + Error::WorkdirSessionOpenFailed { transport, .. } => transport.failure_reason, + _ => None, + }; let mut error = server_api::WorkdirOperationApiError::api(error.into_repository_api_error()); error.denial_reason = reason; + error.failure_reason = failure_reason; error } Self::Provider(error) => server_api::WorkdirOperationApiError::provider(error), @@ -27162,6 +27172,7 @@ async fn serve_external_workdir_provider( Some(ExternalProviderCommand::Operation { operation_id, operation, response }) => { if revoking { let _ = response.send(Err(WorkdirTransportError { + failure_reason: None, denial_reason: None, code: workdir::http::WorkdirTransportErrorCode::Unavailable, message: "External Workdir provider shutdown is in progress".to_string(), @@ -27175,6 +27186,7 @@ async fn serve_external_workdir_provider( }; if pending.len() >= pending_limit { let _ = response.send(Err(WorkdirTransportError { + failure_reason: None, denial_reason: None, code: workdir::http::WorkdirTransportErrorCode::Unavailable, message: "External Workdir provider is at its in-flight operation limit".to_string(), @@ -27186,6 +27198,7 @@ async fn serve_external_workdir_provider( Ok(operation) => operation, Err(message) => { let _ = response.send(Err(WorkdirTransportError { + failure_reason: None, denial_reason: None, code: workdir::http::WorkdirTransportErrorCode::InvalidRequest, message, @@ -27243,6 +27256,7 @@ async fn serve_external_workdir_provider( Err(error.into_transport_error()) } ExternalWorkdirOperationOutcome::Cancelled => Err(WorkdirTransportError { + failure_reason: None, denial_reason: None, code: workdir::http::WorkdirTransportErrorCode::Unavailable, message: "External Workdir operation was cancelled".to_string(), @@ -36896,6 +36910,7 @@ struct ApiErrorLog { diagnostics: Vec, workdir_log: Option, denial_reason: Option, + failure_reason: Option, } impl ApiErrorLog { @@ -36910,6 +36925,7 @@ impl ApiErrorLog { diagnostics: Vec::new(), workdir_log: error.workdir_log.clone(), denial_reason: error.denial_reason, + failure_reason: error.failure_reason, }; } let mut log = Self::from_repository_error(&server_api::RepositoryApiError::new( @@ -36924,6 +36940,7 @@ impl ApiErrorLog { )); log.workdir_log = error.workdir_log.clone(); log.denial_reason = error.denial_reason; + log.failure_reason = error.failure_reason; if log.workdir_log.is_some() { // Existing public API details are unchanged; do not copy request-derived // alias/handle text or session-open errors into operational diagnostics. @@ -36940,6 +36957,7 @@ impl ApiErrorLog { Self { workdir_log: None, denial_reason: None, + failure_reason: None, kind: error .diagnostics .first() @@ -37254,6 +37272,7 @@ impl IntoResponse for ApiError { let log = ApiErrorLog { workdir_log: None, denial_reason: None, + failure_reason: None, kind: self .diagnostics .first() @@ -37601,6 +37620,7 @@ mod tests { assert!(matches!(operation, WorkdirSessionOperation::Read(_))); response .send(Err(WorkdirTransportError { + failure_reason: None, denial_reason: None, code: workdir::http::WorkdirTransportErrorCode::NotFound, message: "logical file was not found".to_string(), @@ -41788,6 +41808,7 @@ mod tests { let error = ApiErrorLog { workdir_log: None, denial_reason: None, + failure_reason: None, kind: "ticket_backend_error".to_string(), message: "sqlite error: FOREIGN KEY constraint failed".to_string(), diagnostics: vec![RuntimeDiagnostic { @@ -41929,6 +41950,7 @@ mod tests { ), ] { let response = InternalWorkdirOperationError::Provider(WorkdirTransportError { + failure_reason: None, denial_reason: None, code, message: "safe provider message".to_string(), @@ -41953,6 +41975,7 @@ mod tests { workdir_operation_transport_error(workdir::WorkdirError::InvalidGlob("[".to_string())); let remote = workdir_operation_transport_error( WorkdirTransportError { + failure_reason: None, denial_reason: None, code: workdir::http::WorkdirTransportErrorCode::InvalidRequest, message: "Workdir operation request is invalid".to_string(), diff --git a/crates/workspace-server/src/server/tests/api_logging_tests.rs b/crates/workspace-server/src/server/tests/api_logging_tests.rs index 0b2320b17..fc883168b 100644 --- a/crates/workspace-server/src/server/tests/api_logging_tests.rs +++ b/crates/workspace-server/src/server/tests/api_logging_tests.rs @@ -563,6 +563,7 @@ async fn generated_workdir_provider_errors_preserve_response_and_log_details() { )); response .send(Err(WorkdirTransportError { + failure_reason: None, denial_reason: reason, code, message: "body-secret /private/provider/path".to_string(), @@ -600,6 +601,64 @@ async fn generated_workdir_provider_errors_preserve_response_and_log_details() { 64 ); } + for reason in [ + workdir::WorkdirFailureReason::SessionClosed, + workdir::WorkdirFailureReason::CommandTerminalMissing, + workdir::WorkdirFailureReason::HttpTimeout, + ] { + let request = runtime_source_request( + &identity, + Some(&fixture.worker.worker_id), + "POST", + &path, + serde_json::to_vec(&server_api::CurrentWorkerWorkdirOperationRequest { + target_workdir: "checkout".into(), + operation: WorkdirSessionOperation::CommandOutput(workdir::CommandOutputRequest { + handle: handle.clone(), + cursor: 0, + limit: 16, + wait: true, + }), + }) + .unwrap(), + ); + let provider = async { + let Some(ExternalProviderCommand::Operation { + operation, + response, + .. + }) = commands.recv().await + else { + panic!("expected output request") + }; + assert!(matches!( + operation, + WorkdirSessionOperation::CommandOutput(_) + )); + response + .send(Err(WorkdirTransportError { + code: Code::Unavailable, + message: "body-secret /private/provider/path".into(), + denial_reason: None, + failure_reason: Some(reason), + })) + .unwrap(); + }; + let ((status, body, events), ()) = + tokio::join!(request_with_logs(app.clone(), request), provider); + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!( + body, + json!({"code":"unavailable", "message":"Workdir session is unavailable"}) + ); + assert_eq!(events.len(), 1, "{events:?}"); + assert_eq!(events[0]["failure_reason"], reason.as_str()); + assert_eq!(events[0]["operation"], "command_output"); + assert_eq!(events[0]["workdir_stage"], "provider_dispatch"); + let log = serde_json::to_string(&events).unwrap(); + assert!(!log.contains("body-secret")); + assert!(!log.contains("/private/provider/path")); + } let provider = async { let Some(ExternalProviderCommand::Operation { response, .. }) = commands.recv().await else { diff --git a/crates/yoi/src/workdir_share.rs b/crates/yoi/src/workdir_share.rs index 6dc60bc1c..f760be2b0 100644 --- a/crates/yoi/src/workdir_share.rs +++ b/crates/yoi/src/workdir_share.rs @@ -836,6 +836,7 @@ where .unwrap_or_else(|_| ExternalWorkdirOperationOutcome::Failed { error: ExternalWorkdirOperationError::from_transport_error( WorkdirTransportError { + failure_reason: None, denial_reason: None, code: workdir::http::WorkdirTransportErrorCode::Internal, message: "External Workdir operation executor failed".to_string(), @@ -954,6 +955,7 @@ async fn execute_operation( Ok(result) => ExternalWorkdirOperationOutcome::Completed { result }, Err(message) => ExternalWorkdirOperationOutcome::Failed { error: ExternalWorkdirOperationError::from_transport_error(WorkdirTransportError { + failure_reason: None, denial_reason: None, code: workdir::http::WorkdirTransportErrorCode::InvalidRequest, message, From aa0fb2f787b29e6a3efc2a1268f5e09a08e1c841 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 21 Oct 2026 04:48:43 +0900 Subject: [PATCH 2/6] fix: continue workdir output reads after long-poll expiry --- crates/workdir/src/http.rs | 61 ++++++++------ .../tests/command_transport_tests.rs | 66 +++++++++++++++ crates/worker/src/checkout_http_tests.rs | 82 ++++++++++++++++--- 3 files changed, 173 insertions(+), 36 deletions(-) diff --git a/crates/workdir/src/http.rs b/crates/workdir/src/http.rs index 8a035395c..f99f804c5 100644 --- a/crates/workdir/src/http.rs +++ b/crates/workdir/src/http.rs @@ -898,34 +898,43 @@ mod client { "command byte limit is outside transport bounds".into(), )); } - let wait = request.wait; - match self - .operate(WorkdirSessionOperation::CommandOutput(request.clone())) - .await? - { - WorkdirSessionOperationResult::CommandOutput(result) => { - let size = result - .chunks - .iter() - .try_fold(0usize, |size, chunk| size.checked_add(chunk.bytes.len())); - if !result.transport_bounds_valid() - || size.is_none_or(|size| { - size > request.limit - || request.cursor.checked_add(size as u64) - != Some(result.next_cursor) - }) - { - return Err(Self::mismatch("bounded command byte page")); + loop { + let started = tokio::time::Instant::now(); + match self + .operate(WorkdirSessionOperation::CommandOutput(request.clone())) + .await? + { + WorkdirSessionOperationResult::CommandOutput(result) => { + let size = result + .chunks + .iter() + .try_fold(0usize, |size, chunk| size.checked_add(chunk.bytes.len())); + if !result.transport_bounds_valid() + || size.is_none_or(|size| { + size > request.limit + || request.cursor.checked_add(size as u64) + != Some(result.next_cursor) + }) + { + return Err(Self::mismatch("bounded command byte page")); + } + if request.wait && result.chunks.is_empty() && result.terminal.is_none() { + // A normal long-poll expiry is not a transport failure. + // Continue at the same cursor, without spending the caller's + // error retry budget or acknowledging any bytes. Pace peers + // that return empty pages immediately; a real long poll has + // already consumed this delay. The caller owns cancellation + // and the overall deadline. + tokio::time::sleep_until( + started + std::time::Duration::from_millis(100), + ) + .await; + continue; + } + return Ok(result); } - if wait && result.chunks.is_empty() && result.terminal.is_none() { - // A waiting request must block at the provider. Never spin HTTP - // requests against an immediate empty response. A caller may retry - // this exact cursor with its own bounded transport retry policy. - return Err(Self::mismatch("waiting command byte page")); - } - return Ok(result); + _ => return Err(Self::mismatch("command_output")), } - _ => return Err(Self::mismatch("command_output")), } } 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 de1150159..902694621 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 @@ -156,6 +156,72 @@ impl Harness { } } +#[tokio::test] +async fn real_http_waiting_command_survives_long_poll_expiry() { + tokio::time::timeout(Duration::from_secs(45), async { + let harness = Harness::new().await; + let handle = harness + .remote + .start_command(CommandRequest { + transport_byte_budget: 1024, + command: "printf ready; while [ ! -e release ]; do sleep 0.05; done; printf after; printf err >&2; exit 7".into(), + timeout_secs: 40, + cwd: WorkdirPath::root(), + tool_call_id: None, + }) + .await + .unwrap(); + let mut request = CommandOutputRequest { + handle, + cursor: 0, + limit: 1024, + wait: true, + }; + let ready = harness.remote.command_output(request.clone()).await.unwrap(); + assert_eq!(ready.chunks.iter().flat_map(|c| c.bytes.iter().copied()).collect::>(), b"ready"); + assert!(ready.terminal.is_none()); + request.cursor = ready.next_cursor; + + // Keep the command silent until the production HTTP handler has expired + // its 20-second long poll. The client must keep waiting, not fail or + // expose an empty page as completed output. + let waiting = harness.remote.command_output(request.clone()); + tokio::pin!(waiting); + tokio::select! { + result = &mut waiting => panic!("waiting read ended before release: {result:?}"), + _ = tokio::time::sleep(Duration::from_secs(21)) => {} + } + std::fs::write(harness._root.path().join("release"), b"").unwrap(); + let mut page = waiting.await.unwrap(); + let mut stdout = Vec::new(); + let mut stderr = Vec::new(); + let terminal = loop { + // The retry cursor still replays the exact nonempty/terminal page. + assert_eq!(harness.remote.command_output(request.clone()).await.unwrap(), page); + for chunk in &page.chunks { + assert_eq!(chunk.sequence, request.cursor); + request.cursor += chunk.bytes.len() as u64; + match chunk.stream { + CommandStream::Stdout => stdout.extend_from_slice(&chunk.bytes), + CommandStream::Stderr => stderr.extend_from_slice(&chunk.bytes), + } + } + assert_eq!(page.next_cursor, request.cursor); + if let Some(terminal) = page.terminal { + break terminal; + } + page = harness.remote.command_output(request.clone()).await.unwrap(); + }; + assert_eq!(stdout, b"after"); + assert_eq!(stderr, b"err"); + assert_eq!(terminal.status, workdir::CommandStatus::Failed); + assert_eq!(terminal.exit_code, Some(7)); + harness.remote.close().await.unwrap(); + }) + .await + .expect("bounded long-poll continuation over HTTP"); +} + #[tokio::test] async fn real_http_closed_session_retains_failure_reason() { let harness = Harness::new().await; diff --git a/crates/worker/src/checkout_http_tests.rs b/crates/worker/src/checkout_http_tests.rs index d6489ae6c..827a5b623 100644 --- a/crates/worker/src/checkout_http_tests.rs +++ b/crates/worker/src/checkout_http_tests.rs @@ -886,21 +886,32 @@ async fn remote_native_bash_retains_streams_quota_and_replays_lost_terminal_thro } #[tokio::test] -async fn waiting_remote_command_rejects_empty_response_without_http_busy_loop() { +async fn waiting_remote_command_paces_empty_responses_and_keeps_cursor_until_output() { watchdog(async { let f = HttpFixture::new(WorkdirSessionCapabilities::ALL).await; let remote = f.open(TOKEN).await.unwrap(); - *f.state.command_fault.lock().unwrap() = Some(CommandResponseFault::EmptyWaiting); - let error = remote - .command_output(workdir::CommandOutputRequest { - handle: workdir::CommandHandle("unused".into()), - cursor: 0, - limit: 1024, - wait: true, + let handle = remote + .start_command(workdir::CommandRequest { + command: "printf done; exit 7".into(), + cwd: WorkdirPath::root(), + timeout_secs: 5, + transport_byte_budget: 1024, + tool_call_id: None, }) .await - .unwrap_err(); - assert!(matches!(error, WorkdirError::Unavailable(_))); + .unwrap(); + *f.state.command_fault.lock().unwrap() = Some(CommandResponseFault::EmptyWaiting); + let mut request = workdir::CommandOutputRequest { + handle, + cursor: 0, + limit: 1024, + wait: false, + }; + // Nonwaiting polls still return their empty page after one HTTP request. + let page = remote.command_output(request.clone()).await.unwrap(); + assert!(page.chunks.is_empty()); + assert!(page.terminal.is_none()); + assert_eq!(page.next_cursor, 0); assert_eq!( f.state .operations @@ -911,6 +922,57 @@ async fn waiting_remote_command_rejects_empty_response_without_http_busy_loop() .count(), 1 ); + f.state.operations.lock().unwrap().clear(); + + request.wait = true; + // Repeated immediate empty responses remain pending, are paced rather + // than spun, and can still be stopped by the caller's deadline. + assert!( + tokio::time::timeout( + Duration::from_millis(350), + remote.command_output(request.clone()), + ) + .await + .is_err() + ); + let operations = f.state.operations.lock().unwrap().clone(); + assert!( + (1..=4).contains(&operations.len()), + "empty polls were not paced: {}", + operations.len() + ); + for operation in &operations { + let WorkdirSessionOperation::CommandOutput(output) = operation else { + panic!("unexpected operation: {operation:?}"); + }; + assert_eq!(output, &request); + } + tokio::time::sleep(Duration::from_millis(150)).await; + assert_eq!( + f.state.operations.lock().unwrap().len(), + operations.len(), + "dropped waiting read kept polling" + ); + + // After the empty replies stop, the exact same cursor still retrieves + // the bytes and terminal from the original command, without a restart. + *f.state.command_fault.lock().unwrap() = None; + let mut bytes = Vec::new(); + 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); + request.cursor += chunk.bytes.len() as u64; + bytes.extend(chunk.bytes); + } + assert_eq!(page.next_cursor, request.cursor); + if let Some(terminal) = page.terminal { + assert_eq!(terminal.exit_code, Some(7)); + break; + } + } + assert_eq!(bytes, b"done"); remote.close().await.unwrap(); }) .await; From 62b2d27b4a35fe6d3f7f96261bf0ca4e545ed94b Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 21 Oct 2026 05:40:39 +0900 Subject: [PATCH 3/6] 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, From a1748b552c52fd5f68de6ba852421fd778aec010 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 21 Oct 2026 05:52:48 +0900 Subject: [PATCH 4/6] fix: validate closed-session diagnostics and fault quotas --- crates/tools/src/bash_output.rs | 37 +++++++++ .../src/local/command_transport/tests.rs | 83 +++++++++++++++++++ crates/worker-runtime/src/http_server.rs | 6 +- 3 files changed, 125 insertions(+), 1 deletion(-) diff --git a/crates/tools/src/bash_output.rs b/crates/tools/src/bash_output.rs index fad3b02e9..6e9db7783 100644 --- a/crates/tools/src/bash_output.rs +++ b/crates/tools/src/bash_output.rs @@ -612,4 +612,41 @@ mod tests { Some(workdir::CommandOutputTruncation::ByteBudget) ); } + + #[test] + fn incomplete_collection_with_quota_retains_both_causes_without_claiming_complete_output() { + let root = BashOutputView::new(); + let mut writer = root.begin().unwrap(); + writer.append(CommandStream::Stdout, b"abc"); + let terminal: workdir::CommandTerminal = serde_json::from_value(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":9, "stderr_bytes":0, "stdout_delivered_bytes":3, "stderr_delivered_bytes":0, + "truncation_reason":"byte_budget" + })).unwrap(); + let metadata = writer.finish_transport(&terminal).unwrap(); + assert!(!metadata.terminal.transport_complete); + assert_eq!(metadata.terminal.exit_code, Some(0)); + assert_eq!( + metadata.terminal.output_collection, + Some(terminal.output_collection) + ); + assert_eq!( + metadata.transport_truncation, + Some(workdir::CommandOutputTruncation::ByteBudget) + ); + assert_eq!(metadata.stdout_seen, 9); + assert_eq!(metadata.stdout_saved, 3); + assert!(metadata.storage_truncated); + assert!(metadata.retention_limit.is_none()); + assert_eq!( + root.read(&metadata.id, CommandStream::Stdout, 0, 100) + .unwrap(), + b"abc" + ); + assert_eq!( + serde_json::to_value(root.metadata(&metadata.id).unwrap()).unwrap(), + serde_json::to_value(metadata).unwrap() + ); + } } diff --git a/crates/workdir/src/local/command_transport/tests.rs b/crates/workdir/src/local/command_transport/tests.rs index 324ce614e..647e1db16 100644 --- a/crates/workdir/src/local/command_transport/tests.rs +++ b/crates/workdir/src/local/command_transport/tests.rs @@ -783,3 +783,86 @@ async fn spawn_fault_retains_safe_reason_without_claiming_output_collection() { #[cfg(feature = "http-client")] mod http; + +#[tokio::test] +async fn drain_failure_and_byte_budget_keep_independent_loss_diagnostics() { + for stream in [CommandStream::Stdout, CommandStream::Stderr] { + let ready = Gate::default(); + let exited = Gate::default(); + let process = FaultProcess { + ready: ready.clone(), + exited: exited.clone(), + wait_error: None, + }; + let reader = FaultReader { + prefix: b"partial", + prefix_read: ready, + fail_ready: exited, + }; + 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, + 3, + "quota-fault", + &telemetry, + cancel_rx, + &sender, + ) + .await + } + CommandStream::Stderr => { + supervise( + process, + &b""[..], + reader, + 10, + 3, + "quota-fault", + &telemetry, + cancel_rx, + &sender, + ) + .await + } + } + }) + .await + .unwrap(); + let Message::Bytes(chunk) = receiver.recv().await.unwrap(); + assert_eq!(chunk.stream, stream); + assert_eq!(chunk.bytes, b"par"); + assert!(receiver.try_recv().is_err()); + assert_eq!(terminal.exit_code, Some(0)); + assert_eq!(terminal.stdout_bytes + terminal.stderr_bytes, 7); + assert_eq!( + terminal.stdout_delivered_bytes + terminal.stderr_delivered_bytes, + 3 + ); + assert_eq!( + terminal.truncation_reason, + Some(crate::CommandOutputTruncation::ByteBudget) + ); + assert_eq!( + terminal.output_collection, + CommandOutputCollection::Incomplete { + failure: CommandFailure { + stage: match stream { + CommandStream::Stdout => CommandFailureStage::StdoutDrain, + CommandStream::Stderr => CommandFailureStage::StderrDrain, + }, + reason: CommandIoErrorKind::PermissionDenied, + os_code: None, + }, + } + ); + assert!(terminal.metadata_valid()); + } +} diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index a1faefc73..aefc66973 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -3206,9 +3206,13 @@ mod tests { .await .expect_err("close must be terminal"); assert!(matches!( - closed_error, + closed_error.classification_source(), workdir::WorkdirError::Unavailable(_) )); + assert_eq!( + closed_error.failure_reason(), + Some(workdir::WorkdirFailureReason::SessionClosed) + ); assert!( state .workdir_sessions From 60eabea19adaf65888428b922f71179cb1eec4fe Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 21 Oct 2026 06:13:16 +0900 Subject: [PATCH 5/6] fix: assert retained remote session close diagnostics --- crates/workspace-server/src/hosts.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 4027b221a..18fe40e22 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -8866,7 +8866,14 @@ mod tests { }) .await .expect_err("closed session must reject local operation without another request"); - assert!(matches!(error, WorkdirError::Unavailable(_))); + assert!(matches!( + error.classification_source(), + WorkdirError::Unavailable(_) + )); + assert_eq!( + error.failure_reason(), + Some(workdir::WorkdirFailureReason::SessionClosed) + ); server.join().expect("mock remote server finished"); } From 6181426abd675f4313d6300c6a01ddeb2ccdd6df Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 21 Oct 2026 06:27:54 +0900 Subject: [PATCH 6/6] fix: assert retained checkout outcome diagnostics --- crates/workspace-server/src/server.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index a1b36997f..9bd590949 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -37791,10 +37791,15 @@ mod tests { ))) .unwrap(); } + let error = write.await.unwrap().unwrap_err(); assert!(matches!( - write.await.unwrap(), - Err(workdir::WorkdirError::OutcomeUnknown(_)) + error.classification_source(), + workdir::WorkdirError::OutcomeUnknown(_) )); + assert_eq!( + error.failure_reason(), + lose_response.then_some(workdir::WorkdirFailureReason::ProviderUnavailable) + ); } }