diff --git a/crates/workdir/src/http.rs b/crates/workdir/src/http.rs index 8a035395c..f99f804c5 100644 --- a/crates/workdir/src/http.rs +++ b/crates/workdir/src/http.rs @@ -898,34 +898,43 @@ mod client { "command byte limit is outside transport bounds".into(), )); } - let wait = request.wait; - match self - .operate(WorkdirSessionOperation::CommandOutput(request.clone())) - .await? - { - WorkdirSessionOperationResult::CommandOutput(result) => { - let size = result - .chunks - .iter() - .try_fold(0usize, |size, chunk| size.checked_add(chunk.bytes.len())); - if !result.transport_bounds_valid() - || size.is_none_or(|size| { - size > request.limit - || request.cursor.checked_add(size as u64) - != Some(result.next_cursor) - }) - { - return Err(Self::mismatch("bounded command byte page")); + loop { + let started = tokio::time::Instant::now(); + match self + .operate(WorkdirSessionOperation::CommandOutput(request.clone())) + .await? + { + WorkdirSessionOperationResult::CommandOutput(result) => { + let size = result + .chunks + .iter() + .try_fold(0usize, |size, chunk| size.checked_add(chunk.bytes.len())); + if !result.transport_bounds_valid() + || size.is_none_or(|size| { + size > request.limit + || request.cursor.checked_add(size as u64) + != Some(result.next_cursor) + }) + { + return Err(Self::mismatch("bounded command byte page")); + } + if request.wait && result.chunks.is_empty() && result.terminal.is_none() { + // A normal long-poll expiry is not a transport failure. + // Continue at the same cursor, without spending the caller's + // error retry budget or acknowledging any bytes. Pace peers + // that return empty pages immediately; a real long poll has + // already consumed this delay. The caller owns cancellation + // and the overall deadline. + tokio::time::sleep_until( + started + std::time::Duration::from_millis(100), + ) + .await; + continue; + } + return Ok(result); } - if wait && result.chunks.is_empty() && result.terminal.is_none() { - // A waiting request must block at the provider. Never spin HTTP - // requests against an immediate empty response. A caller may retry - // this exact cursor with its own bounded transport retry policy. - return Err(Self::mismatch("waiting command byte page")); - } - return Ok(result); + _ => return Err(Self::mismatch("command_output")), } - _ => return Err(Self::mismatch("command_output")), } } diff --git a/crates/worker-runtime/src/http_server/tests/command_transport_tests.rs b/crates/worker-runtime/src/http_server/tests/command_transport_tests.rs index de1150159..902694621 100644 --- a/crates/worker-runtime/src/http_server/tests/command_transport_tests.rs +++ b/crates/worker-runtime/src/http_server/tests/command_transport_tests.rs @@ -156,6 +156,72 @@ impl Harness { } } +#[tokio::test] +async fn real_http_waiting_command_survives_long_poll_expiry() { + tokio::time::timeout(Duration::from_secs(45), async { + let harness = Harness::new().await; + let handle = harness + .remote + .start_command(CommandRequest { + transport_byte_budget: 1024, + command: "printf ready; while [ ! -e release ]; do sleep 0.05; done; printf after; printf err >&2; exit 7".into(), + timeout_secs: 40, + cwd: WorkdirPath::root(), + tool_call_id: None, + }) + .await + .unwrap(); + let mut request = CommandOutputRequest { + handle, + cursor: 0, + limit: 1024, + wait: true, + }; + let ready = harness.remote.command_output(request.clone()).await.unwrap(); + assert_eq!(ready.chunks.iter().flat_map(|c| c.bytes.iter().copied()).collect::>(), b"ready"); + assert!(ready.terminal.is_none()); + request.cursor = ready.next_cursor; + + // Keep the command silent until the production HTTP handler has expired + // its 20-second long poll. The client must keep waiting, not fail or + // expose an empty page as completed output. + let waiting = harness.remote.command_output(request.clone()); + tokio::pin!(waiting); + tokio::select! { + result = &mut waiting => panic!("waiting read ended before release: {result:?}"), + _ = tokio::time::sleep(Duration::from_secs(21)) => {} + } + std::fs::write(harness._root.path().join("release"), b"").unwrap(); + let mut page = waiting.await.unwrap(); + let mut stdout = Vec::new(); + let mut stderr = Vec::new(); + let terminal = loop { + // The retry cursor still replays the exact nonempty/terminal page. + assert_eq!(harness.remote.command_output(request.clone()).await.unwrap(), page); + for chunk in &page.chunks { + assert_eq!(chunk.sequence, request.cursor); + request.cursor += chunk.bytes.len() as u64; + match chunk.stream { + CommandStream::Stdout => stdout.extend_from_slice(&chunk.bytes), + CommandStream::Stderr => stderr.extend_from_slice(&chunk.bytes), + } + } + assert_eq!(page.next_cursor, request.cursor); + if let Some(terminal) = page.terminal { + break terminal; + } + page = harness.remote.command_output(request.clone()).await.unwrap(); + }; + assert_eq!(stdout, b"after"); + assert_eq!(stderr, b"err"); + assert_eq!(terminal.status, workdir::CommandStatus::Failed); + assert_eq!(terminal.exit_code, Some(7)); + harness.remote.close().await.unwrap(); + }) + .await + .expect("bounded long-poll continuation over HTTP"); +} + #[tokio::test] async fn real_http_closed_session_retains_failure_reason() { let harness = Harness::new().await; diff --git a/crates/worker/src/checkout_http_tests.rs b/crates/worker/src/checkout_http_tests.rs index d6489ae6c..827a5b623 100644 --- a/crates/worker/src/checkout_http_tests.rs +++ b/crates/worker/src/checkout_http_tests.rs @@ -886,21 +886,32 @@ async fn remote_native_bash_retains_streams_quota_and_replays_lost_terminal_thro } #[tokio::test] -async fn waiting_remote_command_rejects_empty_response_without_http_busy_loop() { +async fn waiting_remote_command_paces_empty_responses_and_keeps_cursor_until_output() { watchdog(async { let f = HttpFixture::new(WorkdirSessionCapabilities::ALL).await; let remote = f.open(TOKEN).await.unwrap(); - *f.state.command_fault.lock().unwrap() = Some(CommandResponseFault::EmptyWaiting); - let error = remote - .command_output(workdir::CommandOutputRequest { - handle: workdir::CommandHandle("unused".into()), - cursor: 0, - limit: 1024, - wait: true, + let handle = remote + .start_command(workdir::CommandRequest { + command: "printf done; exit 7".into(), + cwd: WorkdirPath::root(), + timeout_secs: 5, + transport_byte_budget: 1024, + tool_call_id: None, }) .await - .unwrap_err(); - assert!(matches!(error, WorkdirError::Unavailable(_))); + .unwrap(); + *f.state.command_fault.lock().unwrap() = Some(CommandResponseFault::EmptyWaiting); + let mut request = workdir::CommandOutputRequest { + handle, + cursor: 0, + limit: 1024, + wait: false, + }; + // Nonwaiting polls still return their empty page after one HTTP request. + let page = remote.command_output(request.clone()).await.unwrap(); + assert!(page.chunks.is_empty()); + assert!(page.terminal.is_none()); + assert_eq!(page.next_cursor, 0); assert_eq!( f.state .operations @@ -911,6 +922,57 @@ async fn waiting_remote_command_rejects_empty_response_without_http_busy_loop() .count(), 1 ); + f.state.operations.lock().unwrap().clear(); + + request.wait = true; + // Repeated immediate empty responses remain pending, are paced rather + // than spun, and can still be stopped by the caller's deadline. + assert!( + tokio::time::timeout( + Duration::from_millis(350), + remote.command_output(request.clone()), + ) + .await + .is_err() + ); + let operations = f.state.operations.lock().unwrap().clone(); + assert!( + (1..=4).contains(&operations.len()), + "empty polls were not paced: {}", + operations.len() + ); + for operation in &operations { + let WorkdirSessionOperation::CommandOutput(output) = operation else { + panic!("unexpected operation: {operation:?}"); + }; + assert_eq!(output, &request); + } + tokio::time::sleep(Duration::from_millis(150)).await; + assert_eq!( + f.state.operations.lock().unwrap().len(), + operations.len(), + "dropped waiting read kept polling" + ); + + // After the empty replies stop, the exact same cursor still retrieves + // the bytes and terminal from the original command, without a restart. + *f.state.command_fault.lock().unwrap() = None; + let mut bytes = Vec::new(); + loop { + let page = remote.command_output(request.clone()).await.unwrap(); + assert_eq!(remote.command_output(request.clone()).await.unwrap(), page); + for chunk in page.chunks { + assert_eq!(chunk.sequence, request.cursor); + request.cursor += chunk.bytes.len() as u64; + bytes.extend(chunk.bytes); + } + assert_eq!(page.next_cursor, request.cursor); + if let Some(terminal) = page.terminal { + assert_eq!(terminal.exit_code, Some(7)); + break; + } + } + assert_eq!(bytes, b"done"); remote.close().await.unwrap(); }) .await;