fix: integrate command output failure diagnostics

Cargo.lock

@@ -6497,6 +6497,7 @@ name = "workdir"
 version = "0.1.0"
 dependencies = [
  "async-trait",
+ "axum",
  "fs-operation",
  "libc",
  "manifest",
@@ -6598,6 +6599,7 @@ dependencies = [
  "tokio",
  "tokio-tungstenite 0.29.0",
  "toml",
+ "tools",
  "tower",
  "tracing",
  "tracing-subscriber",

crates/server-api/src/lib.rs

@@ -132,6 +132,8 @@ pub struct WorkdirOperationApiError {
     pub workdir_log: Option<WorkdirOperationLogContext>,
     #[serde(skip)]
     pub denial_reason: Option<workdir::WorkdirDenialReason>,
+    #[serde(skip)]
+    pub failure_reason: Option<workdir::WorkdirFailureReason>,
 }
 
 /// 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());
     }
 }
 

crates/tools/src/bash.rs

@@ -158,11 +158,15 @@ async fn command_page(
 ) -> Result<workdir::CommandOutput, workdir::WorkdirError> {
     for attempt in 0..3 {
         match session.command_output(request.clone()).await {
-            Err(
+            Err(error)
+                if attempt < 2
+                    && matches!(
+                        error.classification_source(),
                         workdir::WorkdirError::Transport(_)
                             | workdir::WorkdirError::Unavailable(_)
-                | workdir::WorkdirError::OutcomeUnknown(_),
-            ) if attempt < 2 => {
+                            | workdir::WorkdirError::OutcomeUnknown(_)
+                    ) =>
+            {
                 tokio::time::sleep(std::time::Duration::from_millis(100 << attempt)).await;
             }
             result => return result,
@@ -363,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 {
@@ -375,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(),

crates/tools/src/bash_output.rs

@@ -68,8 +68,10 @@ pub struct OutputRange {
 pub struct OutputTerminal {
     pub status: Option<CommandStatus>,
     pub exit_code: Option<i32>,
-    /// 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<workdir::CommandFailure>,
+    pub output_collection: Option<workdir::CommandOutputCollection>,
+    /// 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<BashOutputMetadata, String> {
         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())
@@ -602,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()
+        );
+    }
 }

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 { .. }

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<workdir::CommandOutputCollection>,
     mutate_after_first_binary_response: Option<std::path::PathBuf>,
 }
 
@@ -534,7 +535,11 @@ impl workdir::WorkdirSession for ExternalValidatingSession {
         &self,
         request: workdir::CommandOutputRequest,
     ) -> Result<workdir::CommandOutput, workdir::WorkdirError> {
-        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
@@ -546,9 +551,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)
     }
@@ -574,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");
@@ -599,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)();
@@ -617,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)();
@@ -1103,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)
@@ -1280,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());
@@ -1314,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();
+    }
+}

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

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<crate::WorkdirDenialReason>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    failure_reason: Option<crate::WorkdirFailureReason>,
 }
 
 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(),
@@ -918,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,
@@ -1076,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,

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<crate::WorkdirDenialReason>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub failure_reason: Option<crate::WorkdirFailureReason>,
 }
 
 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<WorkdirSessionOperationResult, WorkdirError> {
             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,
@@ -879,7 +898,8 @@ mod client {
                     "command byte limit is outside transport bounds".into(),
                 ));
             }
-            let wait = request.wait;
+            loop {
+                let started = tokio::time::Instant::now();
                 match self
                     .operate(WorkdirSessionOperation::CommandOutput(request.clone()))
                     .await?
@@ -898,17 +918,25 @@ mod client {
                         {
                             return Err(Self::mismatch("bounded command byte page"));
                         }
-                    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"));
+                        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);
                     }
                     _ => return Err(Self::mismatch("command_output")),
                 }
             }
+        }
 
         async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError> {
             match self
@@ -966,8 +994,10 @@ mod client {
     ) -> Result<T, WorkdirError> {
         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 +1013,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 +1035,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 +1058,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<Mutex<Vec<(String, String, Vec<u8>)>>>,
@@ -1193,6 +1268,7 @@ mod tests {
             (WorkdirTransportErrorCode::Internal, 500, "operation failed"),
         ] {
             let transport = WorkdirTransportError {
+                failure_reason: None,
                 code,
                 denial_reason: None,
                 message: if code == WorkdirTransportErrorCode::Conflict {

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<WorkdirError>,
+    },
     #[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<WorkdirFailureReason> {
+        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<String>) -> 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),

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(),

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;
 
@@ -113,6 +114,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);
@@ -145,6 +147,33 @@ pub(super) fn start(
     telemetry: CommandTelemetry,
     environment: BTreeMap<String, String>,
     resources: Vec<Arc<dyn WorkdirSessionResource>>,
+) -> 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<F: std::future::Future<Output = crate::CommandTerminal> + Send + 'static>(
+    command_id: String,
+    telemetry: CommandTelemetry,
+    resources: Vec<Arc<dyn WorkdirSessionResource>>,
+    execution: impl FnOnce(watch::Receiver<bool>, mpsc::Sender<Message>) -> F + Send + 'static,
 ) -> LocalCommand {
     let (sender, receiver) = mpsc::channel(QUEUE_CHUNKS);
     let (cancel, cancel_rx) = watch::channel(false);
@@ -159,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.
@@ -191,10 +211,18 @@ struct ByteAccounting {
 }
 
 impl ByteAccounting {
-    fn terminal(&self, status: CommandStatus, exit_code: Option<i32>) -> crate::CommandTerminal {
+    fn terminal(
+        &self,
+        status: CommandStatus,
+        exit_code: Option<i32>,
+        process_failure: Option<CommandFailure>,
+        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,
@@ -214,7 +242,7 @@ async fn pump<R: AsyncRead + Unpin>(
     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();
@@ -227,7 +255,11 @@ async fn pump<R: AsyncRead + Unpin>(
         // 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 {
@@ -241,7 +273,15 @@ async fn pump<R: AsyncRead + Unpin>(
             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,
@@ -287,7 +327,7 @@ async fn execute(
     command_id: &str,
     telemetry: &CommandTelemetry,
     environment: BTreeMap<String, String>,
-    mut cancel: watch::Receiver<bool>,
+    cancel: watch::Receiver<bool>,
     sender: &mpsc::Sender<Message>,
 ) -> crate::CommandTerminal {
     telemetry.started(command_id, request.tool_call_id);
@@ -303,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<u32>;
+    fn start_kill(&mut self);
+    fn wait(
+        &mut self,
+    ) -> impl std::future::Future<Output = std::io::Result<std::process::ExitStatus>> + Send;
+}
+
+impl CommandProcess for tokio::process::Child {
+    fn id(&self) -> Option<u32> {
+        self.id()
+    }
+    fn start_kill(&mut self) {
+        let _ = self.start_kill();
+    }
+    async fn wait(&mut self) -> std::io::Result<std::process::ExitStatus> {
+        self.wait().await
+    }
+}
+
+fn process_result(
+    result: std::io::Result<std::process::ExitStatus>,
+) -> (CommandStatus, Option<i32>, Option<CommandFailure>) {
+    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<bool>,
+    sender: &mpsc::Sender<Message>,
+) -> 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,
                 };
-    accounting.terminal(status, exit_code)
+                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, process_failure, output_collection)
 }
 
 #[cfg(test)]

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,
@@ -488,3 +520,349 @@ 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<std::sync::Mutex<(bool, Option<std::task::Waker>)>>);
+
+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<std::io::ErrorKind>,
+}
+
+impl CommandProcess for FaultProcess {
+    fn id(&self) -> Option<u32> {
+        None
+    }
+    fn start_kill(&mut self) {
+        self.ready.open();
+    }
+    async fn wait(&mut self) -> std::io::Result<std::process::ExitStatus> {
+        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<std::io::Result<()>> {
+        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;
+
+#[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());
+    }
+}

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<Arc<LocalWorkdirSession>>,
+    Json(request): Json<WorkdirSessionOperationRequest>,
+) -> Json<WorkdirSessionOperationResult> {
+    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");
+}

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<std::io::ErrorKind> 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<i32>,
+}
+
+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<i32>,
+    pub process_failure: Option<CommandFailure>,
+    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<CommandOutputTruncation>,
 }
 
+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::<CommandFailure>(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::<CommandTerminal>(invalid)
+                    .unwrap()
+                    .metadata_valid()
+            );
+        }
+        let mut unclassified = wire.clone();
+        unclassified
+            .as_object_mut()
+            .unwrap()
+            .remove("output_collection");
+        assert!(serde_json::from_value::<CommandTerminal>(unclassified).is_err());
+        let mut raw = wire;
+        raw["message"] = serde_json::json!("/private TOKEN=secret");
+        assert!(serde_json::from_value::<CommandTerminal>(raw).is_err());
+    }
 }

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(_))
         ));
     }
 

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::<WorkdirTransportError>(
+            r#"{"code":"unavailable","message":"x","failure_reason":"secret /path"}"#
+        )
+        .is_err()
+    );
+}

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

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<workdir::WorkdirError> 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()
     }
 }
@@ -3195,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

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,13 +71,22 @@ impl Harness {
     }
 
     async fn new() -> Self {
-        let root = tempfile::tempdir().unwrap();
-        let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound(
+        Self::with_environment(tempfile::tempdir().unwrap(), BTreeMap::new()).await
+    }
+
+    async fn with_environment(
+        root: tempfile::TempDir,
+        command_environment: BTreeMap<String, String>,
+    ) -> 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);
@@ -156,6 +166,336 @@ 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::<Vec<_>>(), 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_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;
+    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 {
@@ -189,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());
@@ -333,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

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(),
@@ -885,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
@@ -910,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;

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(),

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(_))
     ));
 }
 

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");
     }
 

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<RuntimeDiagnostic>,
     workdir_log: Option<server_api::WorkdirOperationLogContext>,
     denial_reason: Option<workdir::WorkdirDenialReason>,
+    failure_reason: Option<workdir::WorkdirFailureReason>,
 }
 
 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(),
@@ -37771,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)
+            );
         }
     }
 
@@ -38025,6 +38050,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,
 
@@ -38127,6 +38154,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,
 
@@ -39219,6 +39248,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,
 
@@ -41788,6 +41819,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 +41961,7 @@ mod tests {
             ),
         ] {
             let response = InternalWorkdirOperationError::Provider(WorkdirTransportError {
+                failure_reason: None,
                 denial_reason: None,
                 code,
                 message: "safe provider message".to_string(),
@@ -41953,6 +41986,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(),

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 {

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,

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,