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