From a25a0b46d4a3c22c1e0a22c91edefdfd888afcb3 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 11 May 2026 14:14:43 -0400 Subject: [PATCH] fix: wait for Docker ACP stdio termination --- .config/nextest.toml | 4 ++ lib/crates/fabro-acp/tests/session.rs | 12 +++++- lib/crates/fabro-sandbox/src/docker.rs | 51 +++++++++++++++++++++++--- 3 files changed, 59 insertions(+), 8 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index fe7f4f778..2d5880982 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -15,6 +15,10 @@ leak-timeout = "500ms" filter = "package(fabro-server) & test(all_spec_routes_are_routable)" slow-timeout = { period = "15s", terminate-after = 4 } + [[profile.default.overrides]] + filter = "package(fabro-devcontainer) & test(resolve_features_integration)" + slow-timeout = { period = "10s", terminate-after = 3 } + [[profile.default.overrides]] filter = "package(fabro-workflow)" slow-timeout = { period = "2s", terminate-after = 3 } diff --git a/lib/crates/fabro-acp/tests/session.rs b/lib/crates/fabro-acp/tests/session.rs index 3f57046c1..b535a79d1 100644 --- a/lib/crates/fabro-acp/tests/session.rs +++ b/lib/crates/fabro-acp/tests/session.rs @@ -119,17 +119,19 @@ async fn runs_inside_sandbox_and_uses_requested_cwd() { async fn cancellation_sends_session_cancel_and_returns_cancelled() { let tempdir = tempfile::tempdir().expect("create tempdir"); let cancel_path = tempdir.path().join("cancel.txt"); + let tempdir_path = tempdir.path().to_path_buf(); + let cancel_path_for_task = cancel_path.clone(); let cancel_token = CancellationToken::new(); let cancel_for_task = cancel_token.clone(); let task = tokio::spawn(async move { run_fake_agent( - tempdir.path(), + &tempdir_path, HashMap::from([ ("ACP_MODE".to_string(), "cancel".to_string()), ( "ACP_CANCEL_RECORD".to_string(), - cancel_path.to_string_lossy().into_owned(), + cancel_path_for_task.to_string_lossy().into_owned(), ), ]), Some(5_000), @@ -146,6 +148,12 @@ async fn cancellation_sends_session_cancel_and_returns_cancelled() { .expect_err("cancelled turn should error"); assert!(matches!(err, AcpError::Cancelled)); + assert_eq!( + read_to_string(cancel_path) + .await + .expect("read cancel record"), + "session/cancel\n" + ); } #[tokio::test] diff --git a/lib/crates/fabro-sandbox/src/docker.rs b/lib/crates/fabro-sandbox/src/docker.rs index 44e954aca..dae54b658 100644 --- a/lib/crates/fabro-sandbox/src/docker.rs +++ b/lib/crates/fabro-sandbox/src/docker.rs @@ -884,22 +884,46 @@ struct DockerStdioProcessControl { container_id: String, exec_id: String, stop_file: String, - termination: TokioMutex>, + state: DockerStdioProcessState, +} + +#[derive(Default)] +struct DockerStdioProcessState { + stop_requested: TokioMutex, + termination: TokioMutex>, +} + +impl DockerStdioProcessState { + async fn cached_termination(&self) -> Option { + *self.termination.lock().await + } + + async fn should_request_stop(&self) -> bool { + self.cached_termination().await.is_none() && !*self.stop_requested.lock().await + } + + async fn mark_stop_requested(&self) { + *self.stop_requested.lock().await = true; + } + + async fn cache_termination(&self, termination: StdioProcessTermination) { + *self.termination.lock().await = Some(termination); + } } #[async_trait] impl StdioProcessControl for DockerStdioProcessControl { async fn terminate(&self) -> crate::Result<()> { - if self.termination.lock().await.is_some() { + if !self.state.should_request_stop().await { return Ok(()); } request_docker_exec_stop_with(&self.docker, &self.container_id, &self.stop_file).await?; - *self.termination.lock().await = Some(StdioProcessTermination::cancelled()); + self.state.mark_stop_requested().await; Ok(()) } async fn wait(&self) -> crate::Result { - if let Some(termination) = *self.termination.lock().await { + if let Some(termination) = self.state.cached_termination().await { return Ok(termination); } @@ -912,7 +936,7 @@ impl StdioProcessControl for DockerStdioProcessControl { if inspect.running != Some(true) { let exit_code = inspect.exit_code.and_then(|code| i32::try_from(code).ok()); let termination = StdioProcessTermination::exited(exit_code); - *self.termination.lock().await = Some(termination); + self.state.cache_termination(termination).await; return Ok(termination); } time::sleep(std::time::Duration::from_millis(50)).await; @@ -1516,7 +1540,7 @@ impl Sandbox for DockerSandbox { container_id, exec_id, stop_file, - termination: TokioMutex::new(None), + state: DockerStdioProcessState::default(), }); if let Some(token) = cancel_token { @@ -2060,6 +2084,21 @@ mod tests { assert_eq!(output.stdout, b"abc\n"); } + #[tokio::test] + async fn docker_stdio_process_state_does_not_cache_cancelled_on_stop_request() { + let state = DockerStdioProcessState::default(); + + assert!(state.should_request_stop().await); + state.mark_stop_requested().await; + assert_eq!(state.cached_termination().await, None); + assert!(!state.should_request_stop().await); + + let termination = StdioProcessTermination::exited(Some(143)); + state.cache_termination(termination).await; + assert_eq!(state.cached_termination().await, Some(termination)); + assert!(!state.should_request_stop().await); + } + #[tokio::test] async fn controlled_shell_command_skips_user_command_when_stop_already_requested() { let tempdir = tempfile::tempdir().expect("tempdir should be created");