fix: wait for Docker ACP stdio termination

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

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]

lib/crates/fabro-sandbox/src/docker.rs

@@ -884,22 +884,46 @@ struct DockerStdioProcessControl {
     container_id: String,
     exec_id:      String,
     stop_file:    String,
-    termination:  TokioMutex<Option<StdioProcessTermination>>,
+    state:        DockerStdioProcessState,
+}
+
+#[derive(Default)]
+struct DockerStdioProcessState {
+    stop_requested: TokioMutex<bool>,
+    termination:    TokioMutex<Option<StdioProcessTermination>>,
+}
+
+impl DockerStdioProcessState {
+    async fn cached_termination(&self) -> Option<StdioProcessTermination> {
+        *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<StdioProcessTermination> {
-        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");