Merge remote-tracking branch 'origin/work/T-772-command-output-integrity' into work/T-774-workdir-failure-diagnostics

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