From 62b71776f551a18ca729ec93a837b7158547b533 Mon Sep 17 00:00:00 2001 From: Blomios Date: Tue, 14 Jul 2026 23:42:40 +0200 Subject: [PATCH] =?UTF-8?q?fix(background-task):=20d=C3=A9coupler=20la=20d?= =?UTF-8?q?=C3=A9tection=20de=20fin=20de=20t=C3=A2che=20du=20drain=20outpu?= =?UTF-8?q?t=20PTY=20(#2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Le runner de tâches de fond détectait la fin d'un process via l'EOF du drain de sortie PTY, couplant le cycle de vie du process au flux d'output. Un output encore ouvert (ou drainé ailleurs) pouvait masquer ou retarder la détection de fin. Ajoute wait/try_wait au port figé PtyPort : - wait : bloquant, rend un ExitStatus idempotent ; - try_wait : non bloquant, rend Option ; - exit status mémorisé pour garder wait/try_wait/kill cohérents. Implémentation dans PortablePtyAdapter (état d'exit mémorisé). Le CommandBackgroundRunner détecte désormais la fin via pty.wait (select sur cancel/deadline) au lieu de l'EOF du drain. Tous les fakes PtyPort du workspace sont complétés en conséquence. Le tee live UI reste hors périmètre (traité en #58). Aucun breaking IPC/front. Tests : infrastructure/tests/background_task_runner.rs (nouveau) + pty_adapter.rs verts, non-régression application/app-tauri OK (hors échecs réseau du sandbox). Co-Authored-By: Claude Opus 4.8 --- crates/app-tauri/src/state.rs | 12 + crates/application/tests/agent_lifecycle.rs | 6 + .../application/tests/change_agent_profile.rs | 6 + .../application/tests/orchestrator_service.rs | 6 + .../application/tests/structured_launch_d3.rs | 6 + crates/application/tests/terminal_usecases.rs | 8 + crates/application/tests/workstate_actions.rs | 6 + crates/domain/src/ports.rs | 24 ++ .../src/background_task/runner.rs | 46 ++- crates/infrastructure/src/input/mod.rs | 6 + crates/infrastructure/src/pty/mod.rs | 128 +++++++- .../tests/background_task_runner.rs | 291 ++++++++++++++++++ crates/infrastructure/tests/mcp_server.rs | 6 + .../tests/orchestrator_watcher.rs | 6 + crates/infrastructure/tests/pty_adapter.rs | 30 ++ 15 files changed, 550 insertions(+), 37 deletions(-) create mode 100644 crates/infrastructure/tests/background_task_runner.rs diff --git a/crates/app-tauri/src/state.rs b/crates/app-tauri/src/state.rs index d16be76..298c2c9 100644 --- a/crates/app-tauri/src/state.rs +++ b/crates/app-tauri/src/state.rs @@ -4191,6 +4191,12 @@ mod mcp_serve_peer_tests { fn scrollback(&self, _h: &PtyHandle) -> Result, PtyError> { Ok(Vec::new()) } + async fn wait(&self, _h: &PtyHandle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + fn try_wait(&self, _h: &PtyHandle) -> Result, PtyError> { + Ok(Some(ExitStatus { code: Some(0) })) + } async fn kill(&self, _h: &PtyHandle) -> Result { Ok(ExitStatus { code: Some(0) }) } @@ -5574,6 +5580,12 @@ mod mcp_e2e_loopback_tests { fn scrollback(&self, _h: &PtyHandle) -> Result, PtyError> { Ok(Vec::new()) } + async fn wait(&self, _h: &PtyHandle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + fn try_wait(&self, _h: &PtyHandle) -> Result, PtyError> { + Ok(Some(ExitStatus { code: Some(0) })) + } async fn kill(&self, _h: &PtyHandle) -> Result { Ok(ExitStatus { code: Some(0) }) } diff --git a/crates/application/tests/agent_lifecycle.rs b/crates/application/tests/agent_lifecycle.rs index a5b1234..55e848c 100644 --- a/crates/application/tests/agent_lifecycle.rs +++ b/crates/application/tests/agent_lifecycle.rs @@ -554,6 +554,12 @@ impl PtyPort for FakePty { fn scrollback(&self, _handle: &PtyHandle) -> Result, PtyError> { Ok(Vec::new()) } + async fn wait(&self, _handle: &PtyHandle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + fn try_wait(&self, _handle: &PtyHandle) -> Result, PtyError> { + Ok(Some(ExitStatus { code: Some(0) })) + } async fn kill(&self, _handle: &PtyHandle) -> Result { Ok(ExitStatus { code: Some(0) }) } diff --git a/crates/application/tests/change_agent_profile.rs b/crates/application/tests/change_agent_profile.rs index 4f69b3b..2a6caa7 100644 --- a/crates/application/tests/change_agent_profile.rs +++ b/crates/application/tests/change_agent_profile.rs @@ -427,6 +427,12 @@ impl PtyPort for FakePty { fn scrollback(&self, _handle: &PtyHandle) -> Result, PtyError> { Ok(Vec::new()) } + async fn wait(&self, _handle: &PtyHandle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + fn try_wait(&self, _handle: &PtyHandle) -> Result, PtyError> { + Ok(Some(ExitStatus { code: Some(0) })) + } async fn kill(&self, handle: &PtyHandle) -> Result { self.kills.lock().unwrap().push(handle.session_id); Ok(ExitStatus { code: Some(0) }) diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index a4bd2d6..753ac5f 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -370,6 +370,12 @@ impl PtyPort for FakePty { fn scrollback(&self, _handle: &PtyHandle) -> Result, PtyError> { Ok(Vec::new()) } + async fn wait(&self, _handle: &PtyHandle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + fn try_wait(&self, _handle: &PtyHandle) -> Result, PtyError> { + Ok(Some(ExitStatus { code: Some(0) })) + } async fn kill(&self, handle: &PtyHandle) -> Result { self.kills.lock().unwrap().push(handle.session_id); Ok(ExitStatus { code: Some(0) }) diff --git a/crates/application/tests/structured_launch_d3.rs b/crates/application/tests/structured_launch_d3.rs index a20f2ba..c6b74de 100644 --- a/crates/application/tests/structured_launch_d3.rs +++ b/crates/application/tests/structured_launch_d3.rs @@ -357,6 +357,12 @@ impl PtyPort for FakePty { fn scrollback(&self, _handle: &PtyHandle) -> Result, PtyError> { Ok(Vec::new()) } + async fn wait(&self, _handle: &PtyHandle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + fn try_wait(&self, _handle: &PtyHandle) -> Result, PtyError> { + Ok(Some(ExitStatus { code: Some(0) })) + } async fn kill(&self, handle: &PtyHandle) -> Result { self.kills.lock().unwrap().push(handle.session_id); Ok(ExitStatus { code: Some(0) }) diff --git a/crates/application/tests/terminal_usecases.rs b/crates/application/tests/terminal_usecases.rs index f53d7d1..c00aef5 100644 --- a/crates/application/tests/terminal_usecases.rs +++ b/crates/application/tests/terminal_usecases.rs @@ -106,6 +106,14 @@ impl PtyPort for FakePty { Ok(Vec::new()) } + async fn wait(&self, _handle: &PtyHandle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + + fn try_wait(&self, _handle: &PtyHandle) -> Result, PtyError> { + Ok(Some(ExitStatus { code: Some(0) })) + } + async fn kill(&self, handle: &PtyHandle) -> Result { let mut inner = self.0.lock().unwrap(); inner.calls.push(Call::Kill { diff --git a/crates/application/tests/workstate_actions.rs b/crates/application/tests/workstate_actions.rs index 565526b..b9ab999 100644 --- a/crates/application/tests/workstate_actions.rs +++ b/crates/application/tests/workstate_actions.rs @@ -85,6 +85,12 @@ impl PtyPort for FakePty { fn scrollback(&self, _handle: &PtyHandle) -> Result, PtyError> { Ok(Vec::new()) } + async fn wait(&self, _handle: &PtyHandle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + fn try_wait(&self, _handle: &PtyHandle) -> Result, PtyError> { + Ok(Some(ExitStatus { code: Some(0) })) + } async fn kill(&self, handle: &PtyHandle) -> Result { self.kills.lock().unwrap().push(handle.session_id); Ok(ExitStatus { code: Some(0) }) diff --git a/crates/domain/src/ports.rs b/crates/domain/src/ports.rs index 5dc68f2..a24f754 100644 --- a/crates/domain/src/ports.rs +++ b/crates/domain/src/ports.rs @@ -988,8 +988,32 @@ pub trait PtyPort: Send + Sync { /// [`PtyError`] if the handle is unknown. fn scrollback(&self, handle: &PtyHandle) -> Result, PtyError>; + /// Waits for the PTY process to finish naturally and returns its exit status. + /// + /// Idempotent while the session remains registered: the first waiter records + /// the process status, and subsequent calls return that memorized status. + /// + /// # Errors + /// [`PtyError`] if the handle is unknown. + async fn wait(&self, handle: &PtyHandle) -> Result; + + /// Non-blocking process status check. + /// + /// Returns `Ok(None)` while the process is still alive and `Ok(Some(status))` + /// once it has exited. Like [`wait`](Self::wait), the observed exit status is + /// memorized until the session is purged. + /// + /// # Errors + /// [`PtyError`] if the handle is unknown. + fn try_wait(&self, handle: &PtyHandle) -> Result, PtyError>; + /// Kills the PTY's process, returning its exit status. /// + /// Idempotent while the session remains registered: if the process has + /// already exited, returns the memorized status. Implementations may purge + /// the session as part of kill/cleanup; after purge, calls return + /// [`PtyError::NotFound`]. + /// /// # Errors /// [`PtyError`] on failure. async fn kill(&self, handle: &PtyHandle) -> Result; diff --git a/crates/infrastructure/src/background_task/runner.rs b/crates/infrastructure/src/background_task/runner.rs index d5b22a3..d577c4a 100644 --- a/crates/infrastructure/src/background_task/runner.rs +++ b/crates/infrastructure/src/background_task/runner.rs @@ -17,21 +17,19 @@ //! //! # Lifecycle of one task //! -//! `spawn` opens the PTY and starts a detached worker. The worker watches the -//! PTY's output stream for EOF (natural process exit), a cancel notification, or -//! the deadline, whichever comes first. It then snapshots the bounded output tail -//! from the PTY scrollback, reaps the real exit status via +//! `spawn` opens the PTY and starts a detached worker. The worker waits on the +//! PTY process status via [`PtyPort::wait`](domain::ports::PtyPort::wait), a +//! cancel notification, or the deadline, whichever comes first. It then snapshots +//! the bounded output tail from the PTY scrollback, cleans up via //! [`PtyPort::kill`](domain::ports::PtyPort::kill), maps that to a terminal //! [`BackgroundTaskResult`], and sends the completion. //! //! # Output tail & the live tee //! //! A PTY **merges** stdout and stderr onto one stream, so the completion carries -//! the merged output in `stdout_tail` and leaves `stderr_tail` empty. Exit -//! detection reads the PTY's single-consumer output subscription; a *live tee* -//! that re-subscribes the same PTY (to render it in an xterm cell) would supersede -//! the runner's subscription and break EOF detection — teeing a running command -//! therefore requires a multi-consumer PTY broadcast, which is out of scope here. +//! the merged output in `stdout_tail` and leaves `stderr_tail` empty. Completion +//! detection is deliberately decoupled from output subscriptions; the tail is read +//! from PTY scrollback, while process lifecycle comes from `wait`. use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; @@ -56,7 +54,7 @@ const DEFAULT_SIZE: PtySize = PtySize { rows: 24, cols: 80 }; /// Why the worker's terminal wait resolved. enum WaitOutcome { - /// The process exited on its own (stream hit EOF). + /// The process exited on its own. Exited, /// A [`cancel`](BackgroundTaskRunner::cancel) was requested. Cancelled, @@ -184,8 +182,8 @@ impl CommandBackgroundRunner { deadline_ms: Option, clock: &Arc, ) -> WaitOutcome { - let eof = Self::wait_for_eof(Arc::clone(pty), control.pty_handle.clone()); - tokio::pin!(eof); + let process_exit = pty.wait(&control.pty_handle); + tokio::pin!(process_exit); let sleep_ms = deadline_ms.map(|deadline| { let now = u64::try_from(clock.now_millis().max(0)).unwrap_or(0); @@ -194,30 +192,22 @@ impl CommandBackgroundRunner { match sleep_ms { Some(ms) => tokio::select! { - () = &mut eof => WaitOutcome::Exited, + result = &mut process_exit => { + let _ = result; + WaitOutcome::Exited + }, () = control.cancel.notified() => WaitOutcome::Cancelled, () = tokio::time::sleep(std::time::Duration::from_millis(ms)) => WaitOutcome::Expired, }, None => tokio::select! { - () = &mut eof => WaitOutcome::Exited, + result = &mut process_exit => { + let _ = result; + WaitOutcome::Exited + }, () = control.cancel.notified() => WaitOutcome::Cancelled, }, } } - - /// Resolves when the PTY output stream ends (the process has exited). - /// - /// The domain `OutputStream` is a blocking iterator, so it is drained on a - /// blocking task and its bytes discarded — the persisted tail is read from the - /// PTY scrollback instead, which the reader thread fills regardless of - /// subscribers. If the handle is already unknown, resolves immediately so the - /// worker proceeds to reap. - async fn wait_for_eof(pty: Arc, handle: PtyHandle) { - let Ok(stream) = pty.subscribe_output(&handle) else { - return; - }; - let _ = tokio::task::spawn_blocking(move || for _chunk in stream {}).await; - } } #[async_trait] diff --git a/crates/infrastructure/src/input/mod.rs b/crates/infrastructure/src/input/mod.rs index eac3343..68cf453 100644 --- a/crates/infrastructure/src/input/mod.rs +++ b/crates/infrastructure/src/input/mod.rs @@ -1674,6 +1674,12 @@ mod tests { fn scrollback(&self, _handle: &Handle) -> Result, PtyError> { Ok(Vec::new()) } + async fn wait(&self, _handle: &Handle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + fn try_wait(&self, _handle: &Handle) -> Result, PtyError> { + Ok(Some(ExitStatus { code: Some(0) })) + } async fn kill(&self, _handle: &Handle) -> Result { Ok(ExitStatus { code: Some(0) }) } diff --git a/crates/infrastructure/src/pty/mod.rs b/crates/infrastructure/src/pty/mod.rs index 87c4146..923693f 100644 --- a/crates/infrastructure/src/pty/mod.rs +++ b/crates/infrastructure/src/pty/mod.rs @@ -40,6 +40,7 @@ use std::io::{Read, Write}; use std::sync::mpsc::{self, Receiver, Sender}; use std::sync::{Arc, Mutex}; use std::thread::JoinHandle; +use std::time::Duration; use async_trait::async_trait; use portable_pty::{Child, CommandBuilder, MasterPty, NativePtySystem, PtySystem, SlavePty}; @@ -57,6 +58,9 @@ const READ_BUF: usize = 8 * 1024; /// oldest bytes are dropped so a re-attaching view repaints recent history. const SCROLLBACK_CAP: usize = 100 * 1024; +type SharedChild = Arc>>>; +type SharedExitStatus = Arc>>; + /// The shared output hub of one PTY: a bounded scrollback ring buffer plus the /// **single** currently-subscribed receiver. The reader thread feeds both; a /// view subscribes on (re-)attach, and each new subscription supersedes the @@ -129,8 +133,12 @@ struct LivePty { master: Box, /// Writer into the PTY (child stdin). writer: Box, - /// The spawned child process. - child: Box, + /// The spawned child process, retained until wait/try_wait/kill observes its + /// exit status. + child: SharedChild, + /// Memorized exit status. Kept while the PTY session is registered so + /// wait/try_wait/kill are idempotent relative to each other. + exit_status: SharedExitStatus, /// Shared scrollback + subscriber hub, fed by the reader thread. output: Arc>, /// Handle of the reader thread, joined on kill. @@ -241,6 +249,88 @@ fn to_command(spec: &SpawnSpec) -> CommandBuilder { cmd } +fn portable_status(status: &portable_pty::ExitStatus) -> ExitStatus { + ExitStatus { + code: exit_code(status), + } +} + +fn try_wait_shared( + child: &SharedChild, + exit_status: &SharedExitStatus, +) -> Result, PtyError> { + if let Some(status) = *exit_status + .lock() + .map_err(|_| PtyError::Io("pty exit status poisoned".to_owned()))? + { + return Ok(Some(status)); + } + + let observed = { + let mut child_guard = child + .lock() + .map_err(|_| PtyError::Io("pty child lock poisoned".to_owned()))?; + let Some(child) = child_guard.as_mut() else { + return Ok(*exit_status + .lock() + .map_err(|_| PtyError::Io("pty exit status poisoned".to_owned()))?); + }; + + match child.try_wait().map_err(|e| PtyError::Io(e.to_string()))? { + Some(status) => { + let status = portable_status(&status); + *child_guard = None; + Some(status) + } + None => None, + } + }; + + if let Some(status) = observed { + *exit_status + .lock() + .map_err(|_| PtyError::Io("pty exit status poisoned".to_owned()))? = Some(status); + Ok(Some(status)) + } else { + Ok(None) + } +} + +fn kill_shared( + child: &SharedChild, + exit_status: &SharedExitStatus, +) -> Result { + if let Some(status) = *exit_status + .lock() + .map_err(|_| PtyError::Io("pty exit status poisoned".to_owned()))? + { + return Ok(status); + } + + let status = { + let mut child_guard = child + .lock() + .map_err(|_| PtyError::Io("pty child lock poisoned".to_owned()))?; + let Some(child) = child_guard.as_mut() else { + return exit_status + .lock() + .map_err(|_| PtyError::Io("pty exit status poisoned".to_owned()))? + .ok_or_else(|| PtyError::Io("pty child missing exit status".to_owned())); + }; + + let _ = child.kill(); + let raw = child.wait().map_err(|e| PtyError::Io(e.to_string()))?; + let status = portable_status(&raw); + *child_guard = None; + status + }; + + *exit_status + .lock() + .map_err(|_| PtyError::Io("pty exit status poisoned".to_owned()))? = Some(status); + Ok(status) +} + #[async_trait] impl PtyPort for PortablePtyAdapter { async fn spawn(&self, spec: SpawnSpec, size: PtySize) -> Result { @@ -311,7 +401,8 @@ impl PtyPort for PortablePtyAdapter { let live = LivePty { master: pair.master, writer, - child, + child: Arc::new(Mutex::new(Some(child))), + exit_status: Arc::new(Mutex::new(None)), output, reader: Some(reader_handle), }; @@ -373,6 +464,27 @@ impl PtyPort for PortablePtyAdapter { Ok(snapshot) } + async fn wait(&self, handle: &PtyHandle) -> Result { + loop { + if let Some(status) = self.try_wait(handle)? { + return Ok(status); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + } + + fn try_wait(&self, handle: &PtyHandle) -> Result, PtyError> { + let (child, exit_status) = { + let map = self + .sessions + .lock() + .map_err(|_| PtyError::Io("pty registry poisoned".to_owned()))?; + let live = map.get(&handle.session_id).ok_or(PtyError::NotFound)?; + (Arc::clone(&live.child), Arc::clone(&live.exit_status)) + }; + try_wait_shared(&child, &exit_status) + } + async fn kill(&self, handle: &PtyHandle) -> Result { // Remove from the registry so the writer/master drop and the child is // fully owned here while we tear it down. @@ -384,9 +496,9 @@ impl PtyPort for PortablePtyAdapter { map.remove(&handle.session_id).ok_or(PtyError::NotFound)? }; - // Ask the child to terminate, then wait for its real status. - let _ = live.child.kill(); - let status = live.child.wait().map_err(|e| PtyError::Io(e.to_string()))?; + // Ask the child to terminate, unless wait/try_wait already memorized its + // natural exit status. + let status = kill_shared(&live.child, &live.exit_status)?; // Dropping master/writer closes the PTY; the reader thread then sees EOF. // Dropping the broadcast hub drops every subscriber's sender, so any @@ -395,9 +507,7 @@ impl PtyPort for PortablePtyAdapter { let _ = reader.join(); } - Ok(ExitStatus { - code: exit_code(&status), - }) + Ok(status) } } diff --git a/crates/infrastructure/tests/background_task_runner.rs b/crates/infrastructure/tests/background_task_runner.rs new file mode 100644 index 0000000..e60e1aa --- /dev/null +++ b/crates/infrastructure/tests/background_task_runner.rs @@ -0,0 +1,291 @@ +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use async_trait::async_trait; +use domain::ports::{ + BackgroundTaskRunner, Clock, ExitStatus, OutputStream, PtyError, PtyHandle, PtyPort, SpawnSpec, +}; +use domain::{ + AgentId, BackgroundTaskKind, BackgroundTaskResult, BackgroundTaskWakePolicy, ProjectId, + ProjectPath, PtySize, SessionId, TaskId, +}; +use infrastructure::CommandBackgroundRunner; +use uuid::Uuid; + +#[derive(Default)] +struct FakeClock(Mutex); + +impl FakeClock { + fn set(&self, now: i64) { + *self.0.lock().unwrap() = now; + } +} + +impl Clock for FakeClock { + fn now_millis(&self) -> i64 { + *self.0.lock().unwrap() + } +} + +#[derive(Default)] +struct FakePtyState { + handle: Option, + status: Option, + kills: usize, + scrollback: Vec, +} + +#[derive(Default)] +struct FakePty { + state: Mutex, + wait_changed: tokio::sync::Notify, + subscribe_count: AtomicUsize, +} + +impl FakePty { + fn complete(&self, code: Option) { + self.state.lock().unwrap().status = Some(ExitStatus { code }); + self.wait_changed.notify_waiters(); + } + + fn set_scrollback(&self, bytes: &[u8]) { + self.state.lock().unwrap().scrollback = bytes.to_vec(); + } + + fn last_handle(&self) -> PtyHandle { + self.state + .lock() + .unwrap() + .handle + .clone() + .expect("spawned handle") + } + + fn kill_count(&self) -> usize { + self.state.lock().unwrap().kills + } + + fn subscribe_count(&self) -> usize { + self.subscribe_count.load(Ordering::SeqCst) + } +} + +#[async_trait] +impl PtyPort for FakePty { + async fn spawn(&self, _spec: SpawnSpec, _size: PtySize) -> Result { + let handle = PtyHandle { + session_id: SessionId::from_uuid(Uuid::from_u128(0x777)), + }; + self.state.lock().unwrap().handle = Some(handle.clone()); + Ok(handle) + } + + fn write(&self, _handle: &PtyHandle, _data: &[u8]) -> Result<(), PtyError> { + Ok(()) + } + + fn resize(&self, _handle: &PtyHandle, _size: PtySize) -> Result<(), PtyError> { + Ok(()) + } + + fn subscribe_output(&self, _handle: &PtyHandle) -> Result { + self.subscribe_count.fetch_add(1, Ordering::SeqCst); + Ok(Box::new(std::iter::empty())) + } + + fn scrollback(&self, _handle: &PtyHandle) -> Result, PtyError> { + Ok(self.state.lock().unwrap().scrollback.clone()) + } + + async fn wait(&self, _handle: &PtyHandle) -> Result { + loop { + if let Some(status) = self.state.lock().unwrap().status { + return Ok(status); + } + self.wait_changed.notified().await; + } + } + + fn try_wait(&self, _handle: &PtyHandle) -> Result, PtyError> { + Ok(self.state.lock().unwrap().status) + } + + async fn kill(&self, _handle: &PtyHandle) -> Result { + let mut state = self.state.lock().unwrap(); + state.kills += 1; + let status = state.status.get_or_insert(ExitStatus { code: Some(130) }); + Ok(*status) + } +} + +fn id(n: u128) -> Uuid { + Uuid::from_u128(n) +} + +fn spawn_spec(task_id: TaskId, deadline_ms: Option) -> domain::ports::BackgroundTaskSpec { + domain::ports::BackgroundTaskSpec { + task_id, + project_id: ProjectId::from_uuid(id(1)), + owner_agent_id: AgentId::from_uuid(id(2)), + kind: BackgroundTaskKind::Command { + label: "test command".to_owned(), + }, + wake_policy: BackgroundTaskWakePolicy::RecordOnly, + command: Some(SpawnSpec { + command: "/bin/sh".to_owned(), + args: vec!["-c".to_owned(), "echo test".to_owned()], + cwd: ProjectPath::new("/tmp").unwrap(), + env: Vec::new(), + context_plan: None, + sandbox: None, + }), + deadline_ms, + } +} + +fn runner_with( + pty: Arc, + clock: Arc, +) -> ( + CommandBackgroundRunner, + domain::ports::BackgroundCompletionStream, +) { + let runner = CommandBackgroundRunner::new(pty, clock); + let completions = runner.subscribe_completions(); + (runner, completions) +} + +#[tokio::test] +async fn completes_from_wait_without_draining_output() { + let pty = Arc::new(FakePty::default()); + pty.set_scrollback(b"hello from scrollback"); + let clock = Arc::new(FakeClock::default()); + let (runner, mut completions) = runner_with(Arc::clone(&pty), clock); + let task_id = TaskId::from_uuid(id(10)); + + runner + .spawn(spawn_spec(task_id, None)) + .await + .expect("spawn succeeds"); + pty.complete(Some(0)); + + let completion = tokio::time::timeout( + Duration::from_secs(1), + tokio::task::spawn_blocking(move || completions.next().expect("completion")), + ) + .await + .expect("completion arrives") + .expect("completion thread joins"); + + assert_eq!(completion.task_id, task_id); + match completion.result { + BackgroundTaskResult::Success { + exit_code, + stdout_tail, + .. + } => { + assert_eq!(exit_code, Some(0)); + assert_eq!(stdout_tail.as_deref(), Some("hello from scrollback")); + } + other => panic!("expected success, got {other:?}"), + } + assert_eq!(pty.subscribe_count(), 0, "runner must not drain output"); + assert_eq!(pty.kill_count(), 1, "natural exit is cleaned up once"); +} + +#[tokio::test] +async fn ui_subscriber_does_not_interfere_with_runner_completion() { + let pty = Arc::new(FakePty::default()); + let clock = Arc::new(FakeClock::default()); + let (runner, mut completions) = runner_with(Arc::clone(&pty), clock); + let task_id = TaskId::from_uuid(id(11)); + + runner + .spawn(spawn_spec(task_id, None)) + .await + .expect("spawn succeeds"); + let handle = pty.last_handle(); + let _ui_stream = pty.subscribe_output(&handle).expect("ui subscribes"); + pty.complete(Some(0)); + + let completion = tokio::time::timeout( + Duration::from_secs(1), + tokio::task::spawn_blocking(move || completions.next().expect("completion")), + ) + .await + .expect("completion arrives") + .expect("completion thread joins"); + + assert_eq!(completion.task_id, task_id); + assert!(matches!( + completion.result, + BackgroundTaskResult::Success { .. } + )); + assert_eq!( + pty.subscribe_count(), + 1, + "only the explicit UI subscription should exist" + ); +} + +#[tokio::test] +async fn deadline_expires_and_kills_when_wait_never_resolves() { + let pty = Arc::new(FakePty::default()); + let clock = Arc::new(FakeClock::default()); + clock.set(100); + let (runner, mut completions) = runner_with(Arc::clone(&pty), Arc::clone(&clock)); + let task_id = TaskId::from_uuid(id(12)); + + runner + .spawn(spawn_spec(task_id, Some(101))) + .await + .expect("spawn succeeds"); + + let completion = tokio::time::timeout( + Duration::from_secs(1), + tokio::task::spawn_blocking(move || completions.next().expect("completion")), + ) + .await + .expect("completion arrives") + .expect("completion thread joins"); + + assert_eq!(completion.task_id, task_id); + assert!(matches!( + completion.result, + BackgroundTaskResult::Expired { .. } + )); + assert_eq!(pty.kill_count(), 1); + assert_eq!(pty.subscribe_count(), 0); +} + +#[tokio::test] +async fn cancel_wins_even_if_output_is_still_available() { + let pty = Arc::new(FakePty::default()); + pty.set_scrollback(b"still streaming"); + let clock = Arc::new(FakeClock::default()); + let (runner, mut completions) = runner_with(Arc::clone(&pty), clock); + let task_id = TaskId::from_uuid(id(13)); + + runner + .spawn(spawn_spec(task_id, None)) + .await + .expect("spawn succeeds"); + runner.cancel(task_id).await.expect("cancel succeeds"); + + let completion = tokio::time::timeout( + Duration::from_secs(1), + tokio::task::spawn_blocking(move || completions.next().expect("completion")), + ) + .await + .expect("completion arrives") + .expect("completion thread joins"); + + assert_eq!(completion.task_id, task_id); + assert!(matches!( + completion.result, + BackgroundTaskResult::Cancelled { .. } + )); + assert_eq!(pty.kill_count(), 1); + assert_eq!(pty.subscribe_count(), 0); +} diff --git a/crates/infrastructure/tests/mcp_server.rs b/crates/infrastructure/tests/mcp_server.rs index fd8b6b3..7a8095b 100644 --- a/crates/infrastructure/tests/mcp_server.rs +++ b/crates/infrastructure/tests/mcp_server.rs @@ -295,6 +295,12 @@ impl PtyPort for FakePty { fn scrollback(&self, _h: &PtyHandle) -> Result, PtyError> { Ok(Vec::new()) } + async fn wait(&self, _h: &PtyHandle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + fn try_wait(&self, _h: &PtyHandle) -> Result, PtyError> { + Ok(Some(ExitStatus { code: Some(0) })) + } async fn kill(&self, _h: &PtyHandle) -> Result { Ok(ExitStatus { code: Some(0) }) } diff --git a/crates/infrastructure/tests/orchestrator_watcher.rs b/crates/infrastructure/tests/orchestrator_watcher.rs index f5eeccf..93784dc 100644 --- a/crates/infrastructure/tests/orchestrator_watcher.rs +++ b/crates/infrastructure/tests/orchestrator_watcher.rs @@ -294,6 +294,12 @@ impl PtyPort for FakePty { fn scrollback(&self, _h: &PtyHandle) -> Result, PtyError> { Ok(Vec::new()) } + async fn wait(&self, _h: &PtyHandle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + fn try_wait(&self, _h: &PtyHandle) -> Result, PtyError> { + Ok(Some(ExitStatus { code: Some(0) })) + } async fn kill(&self, _h: &PtyHandle) -> Result { Ok(ExitStatus { code: Some(0) }) } diff --git a/crates/infrastructure/tests/pty_adapter.rs b/crates/infrastructure/tests/pty_adapter.rs index f2ecb92..f3747c1 100644 --- a/crates/infrastructure/tests/pty_adapter.rs +++ b/crates/infrastructure/tests/pty_adapter.rs @@ -180,6 +180,34 @@ async fn scrollback_retains_recent_output_for_repaint() { let _ = pty.kill(&handle).await; } +#[tokio::test] +async fn wait_reports_short_command_exit_without_output_drain() { + let pty = PortablePtyAdapter::new(); + let handle = pty + .spawn(sh_spec("printf wait-decoupled"), size()) + .await + .expect("spawn"); + + let status = pty.wait(&handle).await.expect("wait succeeds"); + assert_eq!(status.code, Some(0)); + + let sb = pty + .scrollback(&handle) + .expect("scrollback readable after wait"); + let text = String::from_utf8_lossy(&sb); + assert!( + text.contains("wait-decoupled"), + "reader thread still collects output independently, got {text:?}" + ); + + let cleanup = pty.kill(&handle).await.expect("cleanup kill succeeds"); + assert_eq!(cleanup, status, "cleanup returns memorized exit status"); + assert!( + pty.scrollback(&handle).is_err(), + "cleanup purges the session" + ); +} + #[tokio::test] async fn scrollback_is_bounded_to_cap_and_keeps_most_recent_bytes() { // Emit clearly more than 100 KB of deterministic output, then assert the @@ -228,6 +256,8 @@ async fn write_resize_kill_on_unknown_handle_are_not_found() { assert_eq!(pty.write(&ghost, b"x"), Err(PtyError::NotFound)); assert_eq!(pty.resize(&ghost, size()), Err(PtyError::NotFound)); assert!(pty.subscribe_output(&ghost).is_err()); + assert_eq!(pty.try_wait(&ghost), Err(PtyError::NotFound)); + assert_eq!(pty.wait(&ghost).await, Err(PtyError::NotFound)); assert_eq!(pty.kill(&ghost).await, Err(PtyError::NotFound)); }