diff --git a/src-tauri/src/acp/connection.rs b/src-tauri/src/acp/connection.rs index 5a52fda29..5b6c01deb 100644 --- a/src-tauri/src/acp/connection.rs +++ b/src-tauri/src/acp/connection.rs @@ -1,6 +1,8 @@ use std::collections::{BTreeMap, HashMap, HashSet}; +use std::future::Future; use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::time::Duration; use sacp::schema::{ BlobResourceContents, CancelNotification, ClientCapabilities, ContentBlock, ContentChunk, @@ -33,7 +35,9 @@ use crate::acp::error::AcpError; use crate::acp::file_system_runtime::{FileSystemRuntime, FileSystemRuntimeError}; use crate::acp::registry::{self, AgentDistribution}; use crate::acp::session_state::SessionState; -use crate::acp::terminal_runtime::{TerminalRuntime, TerminalRuntimeError}; +use crate::acp::terminal_runtime::{ + TerminalCleanupReport, TerminalRuntime, TerminalRuntimeError, +}; use crate::acp::types::{ AcpEvent, AvailableCommandInfo, ConnectionInfo, ConnectionStatus, GrokEffortSpec, PermissionOptionInfo, PlanEntryInfo, PromptCapabilitiesInfo, PromptInputBlock, @@ -46,6 +50,15 @@ use crate::network::proxy; use crate::web::event_bridge::{emit_with_state, EventEmitter}; const DEFAULT_COMMAND_COLOR_ENV: [(&str, &str); 1] = [("CLICOLOR_FORCE", "1")]; +const CANCEL_TERMINAL_CLEANUP_DEADLINE: Duration = Duration::from_secs(5); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum CancelCleanupOutcome { + Completed, + Failed(usize), + TimedOut, + TaskFailed, +} fn merge_agent_env( env: &[(&'static str, &'static str)], @@ -2296,6 +2309,88 @@ fn canonical_spec_to_mcp_server(name: &str, spec: &serde_json::Value) -> Result< } } +async fn finish_cancelled_turn_after_cleanup( + state: &Arc>, + emitter: &EventEmitter, + session_id: &str, + agent_type: AgentType, + cleanup: F, + deadline: Duration, +) -> CancelCleanupOutcome +where + F: Future + Send + 'static, +{ + let started = std::time::Instant::now(); + let cleanup_task = tokio::spawn(cleanup); + let (outcome, diagnostic) = match tokio::time::timeout(deadline, cleanup_task).await { + Ok(Ok(report)) if report.is_clean() => (CancelCleanupOutcome::Completed, None), + Ok(Ok(report)) => { + let count = report.failure_count(); + ( + CancelCleanupOutcome::Failed(count), + Some(format!( + "{count} terminal cleanup operation(s) failed; cancellation state was recovered" + )), + ) + } + Ok(Err(err)) => ( + CancelCleanupOutcome::TaskFailed, + Some(format!( + "terminal cleanup task failed; cancellation state was recovered: {err}" + )), + ), + Err(_) => ( + CancelCleanupOutcome::TimedOut, + Some(format!( + "terminal cleanup exceeded {} seconds; cancellation state was recovered and cleanup continues in background", + deadline.as_secs() + )), + ), + }; + + if let Some(message) = diagnostic { + tracing::warn!( + session_id, + agent_type = %agent_type, + elapsed_ms = started.elapsed().as_millis(), + ?outcome, + "[ACP] forcing cancelled turn state convergence after terminal cleanup issue" + ); + emit_with_state( + state, + emitter, + AcpEvent::Error { + message, + agent_type: agent_type.to_string(), + code: Some("terminal_cleanup_incomplete".to_string()), + terminal: false, + }, + ) + .await; + } else { + tracing::info!( + session_id, + agent_type = %agent_type, + elapsed_ms = started.elapsed().as_millis(), + ?outcome, + "[ACP] terminal cleanup completed for cancelled turn" + ); + } + + emit_with_state( + state, + emitter, + AcpEvent::TurnComplete { + session_id: session_id.to_string(), + stop_reason: "cancelled".to_string(), + agent_type: agent_type.to_string(), + }, + ) + .await; + + outcome +} + /// The main ACP connection loop. #[allow(clippy::too_many_arguments)] #[tracing::instrument( @@ -4893,12 +4988,6 @@ async fn run_conversation_loop<'a>( Agent, CancelNotification::new(sid.clone()), ); - // Also terminate any command runtimes created for this - // session so cancellation does not hang on long-running - // terminal tools. - terminal_runtime - .release_all_for_session(sid.0.as_ref()) - .await; tracked_terminal_tool_calls.clear(); // Also cancel any pending permission requests let mut locked = perms.lock().await; @@ -4908,18 +4997,26 @@ async fn run_conversation_loop<'a>( )); } drop(locked); - // Immediately emit TurnComplete so the frontend - // transitions out of "prompting" and the user can - // send new messages. Don't wait for the agent -- - // it may be slow to respond or not respond at all. - emit_with_state( + + // Give isolated terminal groups a bounded chance to + // finish their staged shutdown. The cleanup task owns + // the removed terminal handles and remains detached if + // this outer deadline expires. In every outcome the + // helper emits TurnComplete, reopening prompt admission. + let cleanup_runtime = Arc::clone(&terminal_runtime); + let cleanup_session_id = sid.0.to_string(); + let completion_session_id = cleanup_session_id.clone(); + let _ = finish_cancelled_turn_after_cleanup( state, emitter, - AcpEvent::TurnComplete { - session_id: sid.0.to_string(), - stop_reason: "cancelled".into(), - agent_type: agent_type.to_string(), + &completion_session_id, + agent_type, + async move { + cleanup_runtime + .release_all_for_session(&cleanup_session_id) + .await }, + CANCEL_TERMINAL_CLEANUP_DEADLINE, ) .await; // Cascade-cancel any in-flight delegations owned by @@ -6506,6 +6603,75 @@ mod tests { ToolCallContent::Diff(d) } + #[tokio::test] + async fn terminal_cleanup_timeout_still_completes_cancelled_turn() { + let mut initial = SessionState::new( + "cancel-timeout-conn".to_string(), + AgentType::Grok, + None, + "test-window".to_string(), + None, + ); + initial.status = ConnectionStatus::Prompting; + initial.turn_in_flight = true; + initial.apply_event(&AcpEvent::ToolCall { + tool_call_id: "blocked-terminal".to_string(), + title: "terminal".to_string(), + kind: "execute".to_string(), + status: "in_progress".to_string(), + content: None, + raw_input: None, + raw_output: None, + locations: None, + meta: None, + images: None, + }); + let state = Arc::new(RwLock::new(initial)); + let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>(); + let (finished_tx, finished_rx) = tokio::sync::oneshot::channel::<()>(); + + let outcome = finish_cancelled_turn_after_cleanup( + &state, + &EventEmitter::Noop, + "cancel-timeout-session", + AgentType::Grok, + async move { + let _ = release_rx.await; + let _ = finished_tx.send(()); + TerminalCleanupReport::default() + }, + Duration::from_millis(25), + ) + .await; + + { + let settled = state.read().await; + assert_eq!(outcome, CancelCleanupOutcome::TimedOut); + assert_eq!(settled.status, ConnectionStatus::Connected); + assert!( + !settled.turn_in_flight, + "the next prompt admission gate must reopen" + ); + assert!(settled.active_tool_calls.is_empty()); + assert!(settled.pending_permission.is_none()); + assert_eq!( + settled + .last_error + .as_ref() + .and_then(|error| error.code.as_deref()), + Some("terminal_cleanup_incomplete") + ); + } + + release_tx + .send(()) + .expect("detached cleanup still owns receiver"); + tokio::time::timeout(Duration::from_secs(1), finished_rx) + .await + .expect("detached cleanup finishes") + .expect("cleanup completion signal"); + } + #[test] fn classify_load_failure_resource_not_found_maps_to_code() { assert_eq!( diff --git a/src-tauri/src/acp/terminal_runtime.rs b/src-tauri/src/acp/terminal_runtime.rs index ed0f200ba..dc41a77f4 100644 --- a/src-tauri/src/acp/terminal_runtime.rs +++ b/src-tauri/src/acp/terminal_runtime.rs @@ -1,6 +1,8 @@ use std::collections::{BTreeMap, HashMap}; use std::path::PathBuf; use std::process::Stdio; +#[cfg(unix)] +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -21,6 +23,15 @@ const DEFAULT_OUTPUT_BYTE_LIMIT: u64 = 1_000_000; /// inherit the pipe handle and keep it open long after the direct child /// exits, turning `wait_for_exit` into a silent hang. const READER_DRAIN_GRACE: Duration = Duration::from_millis(200); +const CHILD_EXIT_POLL_INTERVAL: Duration = Duration::from_millis(25); +#[cfg(unix)] +const SIGINT_GRACE: Duration = Duration::from_millis(500); +#[cfg(unix)] +const SIGTERM_GRACE: Duration = Duration::from_millis(1_500); +#[cfg(unix)] +const SIGKILL_GRACE: Duration = Duration::from_millis(2_000); +#[cfg(unix)] +const DESCENDANT_LEASE_DURATION: Duration = Duration::from_secs(5); #[derive(Debug)] pub enum TerminalRuntimeError { @@ -37,6 +48,21 @@ impl TerminalRuntimeError { } } +#[derive(Debug, Default)] +pub(crate) struct TerminalCleanupReport { + failures: Vec<(String, String)>, +} + +impl TerminalCleanupReport { + pub(crate) fn is_clean(&self) -> bool { + self.failures.is_empty() + } + + pub(crate) fn failure_count(&self) -> usize { + self.failures.len() + } +} + #[derive(Debug, Default, Clone)] struct TerminalSnapshot { output: String, @@ -45,22 +71,422 @@ struct TerminalSnapshot { exit_status: Option, } +#[cfg(unix)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ProcessGroupKey { + pgid: libc::pid_t, + generation: u64, +} + +#[cfg(unix)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ProcessGroupPresence { + Present, + Missing, +} + +#[cfg(unix)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ProcessGroupSignalResult { + Delivered, + Missing, +} + +#[cfg(unix)] +trait UnixProcessGroupBackend: Send + Sync { + fn probe(&self, key: ProcessGroupKey) -> Result; + + fn signal( + &self, + key: ProcessGroupKey, + signal: libc::c_int, + ) -> Result; +} + +#[cfg(unix)] +struct LibcProcessGroupBackend; + +#[cfg(unix)] +impl UnixProcessGroupBackend for LibcProcessGroupBackend { + fn probe(&self, key: ProcessGroupKey) -> Result { + let result = unsafe { libc::kill(-key.pgid, 0) }; + if result == 0 { + return Ok(ProcessGroupPresence::Present); + } + + let err = std::io::Error::last_os_error(); + match err.raw_os_error() { + Some(libc::ESRCH) => Ok(ProcessGroupPresence::Missing), + Some(libc::EPERM) => Ok(ProcessGroupPresence::Present), + _ => Err(TerminalRuntimeError::Internal(format!( + "failed to query terminal process group pgid={} generation={}: {err}", + key.pgid, key.generation + ))), + } + } + + fn signal( + &self, + key: ProcessGroupKey, + signal: libc::c_int, + ) -> Result { + let result = unsafe { libc::kill(-key.pgid, signal) }; + if result == 0 { + return Ok(ProcessGroupSignalResult::Delivered); + } + + let err = std::io::Error::last_os_error(); + if err.raw_os_error() == Some(libc::ESRCH) { + return Ok(ProcessGroupSignalResult::Missing); + } + Err(TerminalRuntimeError::Internal(format!( + "failed to signal terminal process group pgid={} generation={} signal={signal}: {err}", + key.pgid, key.generation + ))) + } +} + +#[cfg(unix)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum UnixProcessGroupState { + OwnedLeaderAlive { + key: ProcessGroupKey, + }, + OwnedDescendants { + key: ProcessGroupKey, + deadline: tokio::time::Instant, + }, + Retired, +} + +#[cfg(unix)] +struct UnixProcessGroupLease { + terminal_id: String, + session_id: String, + pid: u32, + backend: Arc, + state: Mutex, + cleanup_gate: Mutex<()>, +} + +#[cfg(unix)] +impl UnixProcessGroupLease { + fn new( + terminal_id: String, + session_id: String, + pid: u32, + generation: u64, + backend: Arc, + ) -> Result { + let pgid = libc::pid_t::try_from(pid).map_err(|_| { + TerminalRuntimeError::Internal(format!("terminal pid {pid} does not fit pid_t")) + })?; + Ok(Self { + terminal_id, + session_id, + pid, + backend, + state: Mutex::new(UnixProcessGroupState::OwnedLeaderAlive { + key: ProcessGroupKey { pgid, generation }, + }), + cleanup_gate: Mutex::new(()), + }) + } + + async fn observe_leader_exit( + &self, + observed_at: tokio::time::Instant, + ) -> Result<(), TerminalRuntimeError> { + let mut state = self.state.lock().await; + let UnixProcessGroupState::OwnedLeaderAlive { key } = *state else { + return Ok(()); + }; + + match self.backend.probe(key) { + Ok(ProcessGroupPresence::Missing) => { + *state = UnixProcessGroupState::Retired; + tracing::info!( + terminal_id = %self.terminal_id, + session_id = %self.session_id, + pid = self.pid, + pgid = key.pgid, + generation = key.generation, + "[ACP] retired terminal process-group lease after leader exit" + ); + Ok(()) + } + Ok(ProcessGroupPresence::Present) => { + let deadline = observed_at + DESCENDANT_LEASE_DURATION; + *state = UnixProcessGroupState::OwnedDescendants { key, deadline }; + tracing::info!( + terminal_id = %self.terminal_id, + session_id = %self.session_id, + pid = self.pid, + pgid = key.pgid, + generation = key.generation, + lease_ms = DESCENDANT_LEASE_DURATION.as_millis(), + "[ACP] started bounded descendant process-group lease" + ); + Ok(()) + } + Err(err) => { + *state = UnixProcessGroupState::Retired; + Err(err) + } + } + } + + fn active_key( + state: &mut UnixProcessGroupState, + now: tokio::time::Instant, + ) -> Option<(ProcessGroupKey, Option)> { + match *state { + UnixProcessGroupState::OwnedLeaderAlive { key } => Some((key, None)), + UnixProcessGroupState::OwnedDescendants { key, deadline } if now < deadline => { + Some((key, Some(deadline))) + } + UnixProcessGroupState::OwnedDescendants { .. } => { + *state = UnixProcessGroupState::Retired; + None + } + UnixProcessGroupState::Retired => None, + } + } + + async fn signal_active( + &self, + signal: libc::c_int, + stage: &'static str, + ) -> Result)>, TerminalRuntimeError> { + let mut state = self.state.lock().await; + let Some((key, deadline)) = Self::active_key(&mut state, tokio::time::Instant::now()) + else { + return Ok(None); + }; + + // A live leader anchors its PID/PGID, but descendants only have a + // bounded numeric-PGID lease. Revalidate that lease immediately before + // every signal so an already-disappeared group is retired instead of + // receiving a signal after a delayed release. POSIX still permits the + // final probe-to-signal syscall race; both calls stay under the same + // terminal state lock to avoid a wider in-process race. + if deadline.is_some() { + match self.backend.probe(key) { + Ok(ProcessGroupPresence::Present) => {} + Ok(ProcessGroupPresence::Missing) => { + *state = UnixProcessGroupState::Retired; + return Ok(None); + } + Err(err) => { + *state = UnixProcessGroupState::Retired; + return Err(err); + } + } + + // A synchronous backend can itself consume the final lease time. + // Expiry is absolute and non-renewable, so do not let a successful + // probe grant time for a later signal. + if Self::active_key(&mut state, tokio::time::Instant::now()).is_none() { + return Ok(None); + } + } + + match self.backend.signal(key, signal) { + Ok(ProcessGroupSignalResult::Delivered) => { + tracing::info!( + terminal_id = %self.terminal_id, + session_id = %self.session_id, + pid = self.pid, + pgid = key.pgid, + generation = key.generation, + signal_stage = stage, + "[ACP] signalled terminal process group" + ); + Ok(Some((key, deadline))) + } + Ok(ProcessGroupSignalResult::Missing) => { + *state = UnixProcessGroupState::Retired; + Ok(None) + } + Err(err) => { + *state = UnixProcessGroupState::Retired; + Err(err) + } + } + } + + async fn probe_active(&self) -> Result { + let mut state = self.state.lock().await; + let Some((key, _)) = Self::active_key(&mut state, tokio::time::Instant::now()) else { + return Ok(false); + }; + + match self.backend.probe(key) { + Ok(ProcessGroupPresence::Present) => Ok(true), + Ok(ProcessGroupPresence::Missing) => { + *state = UnixProcessGroupState::Retired; + Ok(false) + } + Err(err) => { + *state = UnixProcessGroupState::Retired; + Err(err) + } + } + } + + async fn retire(&self) { + *self.state.lock().await = UnixProcessGroupState::Retired; + } + + async fn cleanup_with_stages( + &self, + stages: &[(libc::c_int, &'static str, Duration)], + terminal: Option<&TerminalInstance>, + ) -> Result<(), TerminalRuntimeError> { + let _cleanup_guard = self.cleanup_gate.lock().await; + let started = std::time::Instant::now(); + let mut last_key = None; + + for &(signal, stage, grace) in stages { + let Some((key, descendant_deadline)) = self.signal_active(signal, stage).await? else { + return Ok(()); + }; + last_key = Some(key); + + let now = tokio::time::Instant::now(); + let stage_deadline = descendant_deadline + .map(|lease_deadline| (now + grace).min(lease_deadline)) + .unwrap_or(now + grace); + + loop { + if let Some(terminal) = terminal { + terminal.refresh_exit_status().await?; + } + + let now = tokio::time::Instant::now(); + if !self.probe_active().await? { + tracing::info!( + terminal_id = %self.terminal_id, + session_id = %self.session_id, + pid = self.pid, + pgid = key.pgid, + generation = key.generation, + elapsed_ms = started.elapsed().as_millis(), + signal_stage = stage, + "[ACP] terminal process-group lease retired" + ); + return Ok(()); + } + + if now >= stage_deadline { + break; + } + tokio::time::sleep( + CHILD_EXIT_POLL_INTERVAL.min(stage_deadline.saturating_duration_since(now)), + ) + .await; + } + } + + self.retire().await; + let key = last_key.expect("non-empty terminal signal stages"); + Err(TerminalRuntimeError::Internal(format!( + "terminal process group pgid={} generation={} survived SIGKILL deadline", + key.pgid, key.generation + ))) + } + + async fn cleanup(&self, terminal: &TerminalInstance) -> Result<(), TerminalRuntimeError> { + self.cleanup_with_stages( + &[ + (libc::SIGINT, "sigint", SIGINT_GRACE), + (libc::SIGTERM, "sigterm", SIGTERM_GRACE), + (libc::SIGKILL, "sigkill", SIGKILL_GRACE), + ], + Some(terminal), + ) + .await + } +} + +#[cfg(all(test, unix))] +#[derive(Clone)] +struct ExitObservationBarrier { + reached: Arc, + resume: Arc, +} + struct TerminalInstance { + terminal_id: String, session_id: String, + #[cfg(unix)] + process_group: UnixProcessGroupLease, + /// Serializes observing a direct-child exit with the corresponding Unix + /// process-group lease transition. A cleanup must not see an absent child + /// while the lease still describes a live leader. + #[cfg(unix)] + leader_exit_observation_gate: Mutex<()>, output_limit: Option, child: Mutex>, snapshot: Mutex, reader_handles: Mutex>>, + #[cfg(all(test, unix))] + exit_observation_barrier: std::sync::Mutex>, } impl TerminalInstance { - fn new(session_id: String, output_limit: Option, child: tokio::process::Child) -> Self { - Self { + fn new( + terminal_id: String, + session_id: String, + output_limit: Option, + child: tokio::process::Child, + #[cfg_attr(not(unix), allow(unused_variables))] pid: u32, + #[cfg(unix)] generation: u64, + #[cfg(unix)] process_group_backend: Arc, + ) -> Result { + #[cfg(unix)] + let process_group = UnixProcessGroupLease::new( + terminal_id.clone(), + session_id.clone(), + pid, + generation, + process_group_backend, + )?; + + Ok(Self { + terminal_id, session_id, + #[cfg(unix)] + process_group, + #[cfg(unix)] + leader_exit_observation_gate: Mutex::new(()), output_limit: output_limit.and_then(|v| usize::try_from(v).ok()), child: Mutex::new(Some(child)), snapshot: Mutex::new(TerminalSnapshot::default()), reader_handles: Mutex::new(Vec::new()), + #[cfg(all(test, unix))] + exit_observation_barrier: std::sync::Mutex::new(None), + }) + } + + #[cfg(all(test, unix))] + fn install_exit_observation_barrier(&self, barrier: ExitObservationBarrier) { + *self + .exit_observation_barrier + .lock() + .expect("exit observation barrier lock") = Some(barrier); + } + + #[cfg(all(test, unix))] + async fn pause_after_reap_before_lease_observation(&self) { + let barrier = self + .exit_observation_barrier + .lock() + .expect("exit observation barrier lock") + .clone(); + if let Some(barrier) = barrier { + barrier.reached.wait().await; + barrier.resume.wait().await; } } @@ -102,6 +528,43 @@ impl TerminalInstance { } } + #[cfg(unix)] + let (maybe_status, process_group_result) = { + let _observation_guard = self.leader_exit_observation_gate.lock().await; + let mut child_guard = self.child.lock().await; + if let Some(child) = child_guard.as_mut() { + match child.try_wait() { + Ok(Some(status)) => { + // Do not make the child absent visible until the group + // lease has either become bounded descendants or has + // retired. The child mutex is released before the + // async lease observation. + drop(child_guard); + + #[cfg(all(test, unix))] + self.pause_after_reap_before_lease_observation().await; + + let process_group_result = self + .process_group + .observe_leader_exit(tokio::time::Instant::now()) + .await; + + *self.child.lock().await = None; + (Some(status), process_group_result) + } + Ok(None) => (None, Ok(())), + Err(err) => { + return Err(TerminalRuntimeError::Internal(format!( + "failed to query terminal exit status: {err}" + ))) + } + } + } else { + (None, Ok(())) + } + }; + + #[cfg(not(unix))] let maybe_status = { let mut child_guard = self.child.lock().await; if let Some(child) = child_guard.as_mut() { @@ -135,76 +598,51 @@ impl TerminalInstance { self.drain_readers().await; let mut snapshot = self.snapshot.lock().await; snapshot.exit_status = Some(map_exit_status(status)); + + #[cfg(unix)] + process_group_result?; } Ok(()) } async fn wait_for_exit(&self) -> Result { - self.refresh_exit_status().await?; - let cached_exit = self.snapshot.lock().await.exit_status.clone(); - if let Some(exit_status) = cached_exit { - self.drain_readers().await; - return Ok(exit_status); + loop { + self.refresh_exit_status().await?; + if let Some(exit_status) = self.snapshot.lock().await.exit_status.clone() { + return Ok(exit_status); + } + tokio::time::sleep(CHILD_EXIT_POLL_INTERVAL).await; } + } - let exit_status = { - let mut child_guard = self.child.lock().await; - let Some(child) = child_guard.as_mut() else { - return Err(TerminalRuntimeError::Internal( - "terminal process missing while waiting for exit".to_string(), - )); - }; - let status = child.wait().await.map_err(|err| { - TerminalRuntimeError::Internal(format!( - "failed waiting for terminal process to exit: {err}" - )) - })?; - *child_guard = None; - map_exit_status(status) - }; - - self.drain_readers().await; - - let mut snapshot = self.snapshot.lock().await; - snapshot.exit_status = Some(exit_status.clone()); - Ok(exit_status) + #[cfg(unix)] + async fn kill_command(&self) -> Result<(), TerminalRuntimeError> { + self.refresh_exit_status().await?; + let cleanup_result = self.process_group.cleanup(self).await; + let refresh_result = self.refresh_exit_status().await; + cleanup_result.and(refresh_result) } + #[cfg(not(unix))] async fn kill_command(&self) -> Result<(), TerminalRuntimeError> { self.refresh_exit_status().await?; - let already_exited = self.snapshot.lock().await.exit_status.is_some(); - if already_exited { - self.drain_readers().await; + if self.snapshot.lock().await.exit_status.is_some() { return Ok(()); } - let exit_status = { - let mut child_guard = self.child.lock().await; - let Some(child) = child_guard.as_mut() else { - return Ok(()); - }; - - if let Some(pid) = child.id() { - if let Err(err) = kill_tree::tokio::kill_tree(pid).await { - tracing::error!("[ACP] kill_tree failed for pid {pid}: {err}"); - } + let pid = self + .child + .lock() + .await + .as_ref() + .and_then(tokio::process::Child::id); + if let Some(pid) = pid { + if let Err(err) = kill_tree::tokio::kill_tree(pid).await { + tracing::error!("[ACP] kill_tree failed for pid {pid}: {err}"); } - - let status = child.wait().await.map_err(|err| { - TerminalRuntimeError::Internal(format!( - "failed to wait for killed terminal process: {err}" - )) - })?; - *child_guard = None; - map_exit_status(status) - }; - - self.drain_readers().await; - - let mut snapshot = self.snapshot.lock().await; - snapshot.exit_status = Some(exit_status); - Ok(()) + } + self.wait_for_exit().await.map(|_| ()) } async fn snapshot(&self) -> TerminalSnapshot { @@ -214,6 +652,10 @@ impl TerminalInstance { pub struct TerminalRuntime { terminals: Mutex, + #[cfg(unix)] + process_group_backend: Arc, + #[cfg(unix)] + next_process_group_generation: AtomicU64, /// Base environment merged into every spawned terminal command before /// the agent's per-request `env` is applied. This is where the codeg /// git credential helper (`GIT_CONFIG_*`) lives so an agent that runs @@ -248,11 +690,28 @@ impl TerminalRuntime { pub fn with_base_env(base_env: BTreeMap) -> Self { Self { terminals: Mutex::new(HashMap::new()), + #[cfg(unix)] + process_group_backend: Arc::new(LibcProcessGroupBackend), + #[cfg(unix)] + next_process_group_generation: AtomicU64::new(1), base_env, default_cwd: None, } } + #[cfg(unix)] + fn allocate_process_group_generation(&self) -> Result { + self.next_process_group_generation + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |generation| { + generation.checked_add(1) + }) + .map_err(|_| { + TerminalRuntimeError::Internal( + "terminal process-group generation space exhausted".to_string(), + ) + }) + } + /// Set the fallback working directory used when a `terminal/create` request /// does not specify its own `cwd`. Chainable after `with_base_env`. pub fn with_default_cwd(mut self, default_cwd: Option) -> Self { @@ -260,6 +719,12 @@ impl TerminalRuntime { self } + #[cfg(all(test, unix))] + fn with_process_group_backend(mut self, backend: Arc) -> Self { + self.process_group_backend = backend; + self + } + /// Apply stdio, working directory, and environment to a freshly built /// terminal command. Shared by the direct-exec and shell-fallback spawn /// paths in `create_terminal` so both honor the same cwd precedence and @@ -300,6 +765,11 @@ impl TerminalRuntime { for env_var in &request.env { command.env(&env_var.name, &env_var.value); } + + #[cfg(unix)] + { + command.process_group(0); + } } pub async fn create_terminal( @@ -329,6 +799,9 @@ impl TerminalRuntime { )); } + #[cfg(unix)] + let process_group_generation = self.allocate_process_group_generation()?; + // Spawn the command. Try a direct exec first so a real program — one // resolved on PATH, an absolute path, or a relative/space-containing // path reachable through the request's cwd and env — runs exactly as @@ -368,15 +841,34 @@ impl TerminalRuntime { } }; + let pid = child.id().ok_or_else(|| { + TerminalRuntimeError::Internal("spawned terminal has no process id".to_string()) + })?; let stdout = child.stdout.take(); let stderr = child.stderr.take(); let terminal_id = format!("term_{}", uuid::Uuid::new_v4().simple()); let terminal = Arc::new(TerminalInstance::new( + terminal_id.clone(), request.session_id.to_string(), Some(output_byte_limit), child, - )); + pid, + #[cfg(unix)] + process_group_generation, + #[cfg(unix)] + Arc::clone(&self.process_group_backend), + )?); + + #[cfg(unix)] + tracing::info!( + terminal_id = %terminal_id, + session_id = %request.session_id, + pid, + pgid = pid, + generation = process_group_generation, + "[ACP] spawned isolated terminal process group" + ); let mut handles: Vec> = Vec::new(); if let Some(reader) = stdout { @@ -508,7 +1000,7 @@ impl TerminalRuntime { Ok(ReleaseTerminalResponse::new()) } - pub async fn release_all_for_session(&self, session_id: &str) { + pub(crate) async fn release_all_for_session(&self, session_id: &str) -> TerminalCleanupReport { let removed = { let mut terminals = self.terminals.lock().await; let ids: Vec = terminals @@ -526,11 +1018,28 @@ impl TerminalRuntime { removed }; - for terminal in removed { - if let Err(err) = terminal.kill_command().await { - tracing::error!("[ACP] Failed to release terminal during cleanup: {err:?}"); - } - } + let results = futures::future::join_all(removed.into_iter().map(|terminal| async move { + let terminal_id = terminal.terminal_id.clone(); + (terminal_id, terminal.kill_command().await) + })) + .await; + + let failures = results + .into_iter() + .filter_map(|(terminal_id, result)| { + result.err().map(|err| { + tracing::error!( + terminal_id = %terminal_id, + session_id, + error = ?err, + "[ACP] terminal session cleanup failed" + ); + (terminal_id, format!("{err:?}")) + }) + }) + .collect(); + + TerminalCleanupReport { failures } } async fn find_terminal( @@ -673,8 +1182,710 @@ fn decode_available_utf8(pending: &mut Vec) -> String { #[cfg(all(test, unix))] mod tests { use super::*; + use std::collections::VecDeque; + use std::time::Instant; + + use kill_tree::Config; use sacp::schema::{EnvVariable, SessionId, WaitForTerminalExitRequest}; + fn init_test_tracing() { + let _ = tracing_subscriber::fmt().with_test_writer().try_init(); + } + + #[derive(Debug, Clone, PartialEq, Eq)] + enum BackendCall { + Probe(ProcessGroupKey), + Signal(ProcessGroupKey, libc::c_int), + } + + #[derive(Default)] + struct MockProcessGroupBackend { + calls: std::sync::Mutex>, + probes: std::sync::Mutex>, + signals: std::sync::Mutex>, + probe_delay: Option, + call_tx: Option>, + } + + impl MockProcessGroupBackend { + fn with_probes(probes: impl IntoIterator) -> Self { + Self { + probes: std::sync::Mutex::new(probes.into_iter().collect()), + ..Self::default() + } + } + + fn with_signals(signals: impl IntoIterator) -> Self { + Self { + signals: std::sync::Mutex::new(signals.into_iter().collect()), + ..Self::default() + } + } + + fn with_probe_delay( + probes: impl IntoIterator, + probe_delay: Duration, + ) -> Self { + Self { + probes: std::sync::Mutex::new(probes.into_iter().collect()), + probe_delay: Some(probe_delay), + ..Self::default() + } + } + + fn with_call_tx( + mut self, + call_tx: tokio::sync::mpsc::UnboundedSender, + ) -> Self { + self.call_tx = Some(call_tx); + self + } + + fn record_call(&self, call: BackendCall) { + self.calls + .lock() + .expect("mock calls lock") + .push(call.clone()); + if let Some(call_tx) = &self.call_tx { + let _ = call_tx.send(call); + } + } + + fn calls(&self) -> Vec { + self.calls.lock().expect("mock calls lock").clone() + } + + fn delivered_signals(&self) -> Vec { + self.calls() + .into_iter() + .filter_map(|call| match call { + BackendCall::Signal(_, signal) => Some(signal), + BackendCall::Probe(_) => None, + }) + .collect() + } + + fn clear_calls(&self) { + self.calls.lock().expect("mock calls lock").clear(); + } + + fn push_probe(&self, presence: ProcessGroupPresence) { + self.probes + .lock() + .expect("mock probes lock") + .push_back(presence); + } + } + + impl UnixProcessGroupBackend for MockProcessGroupBackend { + fn probe( + &self, + key: ProcessGroupKey, + ) -> Result { + self.record_call(BackendCall::Probe(key)); + let result = self + .probes + .lock() + .expect("mock probes lock") + .pop_front() + .unwrap_or(ProcessGroupPresence::Present); + if let Some(delay) = self.probe_delay { + std::thread::sleep(delay); + } + Ok(result) + } + + fn signal( + &self, + key: ProcessGroupKey, + signal: libc::c_int, + ) -> Result { + self.record_call(BackendCall::Signal(key, signal)); + Ok(self + .signals + .lock() + .expect("mock signals lock") + .pop_front() + .unwrap_or(ProcessGroupSignalResult::Delivered)) + } + } + + const TEST_PROCESS_GROUP_KEY: ProcessGroupKey = ProcessGroupKey { + pgid: 42_424, + generation: 7, + }; + + fn test_process_group_lease( + state: UnixProcessGroupState, + backend: Arc, + ) -> UnixProcessGroupLease { + UnixProcessGroupLease { + terminal_id: "test-terminal".to_string(), + session_id: "test-session".to_string(), + pid: TEST_PROCESS_GROUP_KEY.pgid as u32, + backend, + state: Mutex::new(state), + cleanup_gate: Mutex::new(()), + } + } + + fn zero_grace_stages() -> [(libc::c_int, &'static str, Duration); 3] { + [ + (libc::SIGINT, "sigint", Duration::ZERO), + (libc::SIGTERM, "sigterm", Duration::ZERO), + (libc::SIGKILL, "sigkill", Duration::ZERO), + ] + } + + #[tokio::test] + async fn leader_exit_with_descendants_starts_bounded_lease() { + let backend = Arc::new(MockProcessGroupBackend::default()); + let lease = test_process_group_lease( + UnixProcessGroupState::OwnedLeaderAlive { + key: TEST_PROCESS_GROUP_KEY, + }, + backend, + ); + let observed_at = tokio::time::Instant::now(); + + lease + .observe_leader_exit(observed_at) + .await + .expect("observe leader exit"); + + assert_eq!( + *lease.state.lock().await, + UnixProcessGroupState::OwnedDescendants { + key: TEST_PROCESS_GROUP_KEY, + deadline: observed_at + DESCENDANT_LEASE_DURATION, + } + ); + } + + #[tokio::test] + async fn leader_exit_without_descendants_retires_before_delayed_release() { + let backend = Arc::new(MockProcessGroupBackend::with_probes([ + ProcessGroupPresence::Missing, + ])); + let lease = test_process_group_lease( + UnixProcessGroupState::OwnedLeaderAlive { + key: TEST_PROCESS_GROUP_KEY, + }, + backend.clone(), + ); + lease + .observe_leader_exit(tokio::time::Instant::now()) + .await + .expect("observe leader exit"); + backend.clear_calls(); + + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("delayed cleanup"); + + assert!(backend.calls().is_empty()); + assert!(backend.delivered_signals().is_empty()); + assert_eq!(*lease.state.lock().await, UnixProcessGroupState::Retired); + } + + #[tokio::test] + async fn naturally_exited_terminal_delayed_release_sends_no_group_signal() { + let backend = Arc::new(MockProcessGroupBackend::with_probes([ + ProcessGroupPresence::Missing, + ])); + let runtime = TerminalRuntime::with_base_env(BTreeMap::new()) + .with_process_group_backend(backend.clone()); + let session_id = SessionId::new("natural-exit-delayed-release".to_string()); + let request = CreateTerminalRequest::new(session_id.clone(), "/bin/true".to_string()); + let response = runtime + .create_terminal(request) + .await + .expect("create short terminal"); + + runtime + .wait_for_terminal_exit(WaitForTerminalExitRequest::new( + session_id.clone(), + response.terminal_id, + )) + .await + .expect("observe natural exit"); + backend.clear_calls(); + tokio::time::sleep(Duration::from_millis(25)).await; + + let report = runtime.release_all_for_session(session_id.0.as_ref()).await; + + assert!(report.is_clean()); + assert!(backend.calls().is_empty()); + assert!(backend.delivered_signals().is_empty()); + } + + #[tokio::test] + async fn concurrent_cleanup_never_signals_after_reap_before_lease_observation() { + let (call_tx, mut call_rx) = tokio::sync::mpsc::unbounded_channel(); + let backend = Arc::new( + MockProcessGroupBackend::with_probes([ProcessGroupPresence::Missing]) + .with_call_tx(call_tx), + ); + let runtime = TerminalRuntime::with_base_env(BTreeMap::new()) + .with_process_group_backend(backend.clone()); + let session_id = SessionId::new("reap-before-lease-observation".to_string()); + let response = runtime + .create_terminal(CreateTerminalRequest::new( + session_id.clone(), + "/bin/true".to_string(), + )) + .await + .expect("create short terminal"); + let terminal = runtime + .find_terminal(&response.terminal_id.to_string(), session_id.0.as_ref()) + .await + .expect("find test terminal"); + let barrier = ExitObservationBarrier { + reached: Arc::new(tokio::sync::Barrier::new(2)), + resume: Arc::new(tokio::sync::Barrier::new(2)), + }; + terminal.install_exit_observation_barrier(barrier.clone()); + + let refresh_terminal = terminal.clone(); + let refresh = tokio::spawn(async move { + loop { + refresh_terminal.refresh_exit_status().await?; + if refresh_terminal.snapshot.lock().await.exit_status.is_some() { + return Ok::<(), TerminalRuntimeError>(()); + } + tokio::time::sleep(CHILD_EXIT_POLL_INTERVAL).await; + } + }); + tokio::time::timeout(Duration::from_secs(1), barrier.reached.wait()) + .await + .expect("direct child did not reach controlled reap observation"); + let leader_key = match *terminal.process_group.state.lock().await { + UnixProcessGroupState::OwnedLeaderAlive { key } => key, + other => panic!("lease changed before the controlled observation point: {other:?}"), + }; + assert!( + terminal.child.lock().await.is_some(), + "child must not become absent before the lease observation commits" + ); + + let cleanup_terminal = terminal.clone(); + let cleanup = tokio::spawn(async move { cleanup_terminal.kill_command().await }); + let early_call = tokio::time::timeout(Duration::from_millis(50), call_rx.recv()).await; + assert!( + early_call.is_err(), + "cleanup signalled or probed while leader exit observation was paused: {early_call:?}" + ); + + tokio::time::timeout(Duration::from_secs(1), barrier.resume.wait()) + .await + .expect("controlled reap observation did not resume"); + refresh.await.expect("join refresh").expect("refresh exit"); + cleanup.await.expect("join cleanup").expect("cleanup exit"); + + assert_eq!(backend.calls(), vec![BackendCall::Probe(leader_key)]); + assert!(backend.delivered_signals().is_empty()); + assert_eq!( + *terminal.process_group.state.lock().await, + UnixProcessGroupState::Retired + ); + + // A later user of the same numeric PGID remains outside this terminal's + // retired lease and receives no further backend operation. + backend.push_probe(ProcessGroupPresence::Present); + let calls_after_retirement = backend.calls(); + terminal.kill_command().await.expect("repeated cleanup"); + assert_eq!(backend.calls(), calls_after_retirement); + assert!(backend.delivered_signals().is_empty()); + } + + #[tokio::test] + async fn descendant_lease_release_before_deadline_can_signal_original_group() { + let backend = Arc::new(MockProcessGroupBackend::with_probes([ + ProcessGroupPresence::Present, + ProcessGroupPresence::Present, + ProcessGroupPresence::Present, + ProcessGroupPresence::Present, + ProcessGroupPresence::Present, + ProcessGroupPresence::Missing, + ])); + let lease = test_process_group_lease( + UnixProcessGroupState::OwnedDescendants { + key: TEST_PROCESS_GROUP_KEY, + deadline: tokio::time::Instant::now() + DESCENDANT_LEASE_DURATION, + }, + backend.clone(), + ); + + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("cleanup descendants"); + + assert_eq!( + backend.delivered_signals(), + vec![libc::SIGINT, libc::SIGTERM, libc::SIGKILL] + ); + assert!(backend.calls().iter().all(|call| match call { + BackendCall::Probe(key) | BackendCall::Signal(key, _) => { + *key == TEST_PROCESS_GROUP_KEY + } + })); + } + + #[tokio::test] + async fn missing_descendant_group_retires_before_signal_and_ignores_later_reuse() { + let backend = Arc::new(MockProcessGroupBackend::with_probes([ + ProcessGroupPresence::Missing, + ])); + let lease = test_process_group_lease( + UnixProcessGroupState::OwnedDescendants { + key: TEST_PROCESS_GROUP_KEY, + deadline: tokio::time::Instant::now() + DESCENDANT_LEASE_DURATION, + }, + backend.clone(), + ); + + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("missing descendant group cleanup"); + + assert_eq!( + backend.calls(), + vec![BackendCall::Probe(TEST_PROCESS_GROUP_KEY)] + ); + assert!(backend.delivered_signals().is_empty()); + assert_eq!(*lease.state.lock().await, UnixProcessGroupState::Retired); + + // The original group is gone. A later owner reusing its numeric PGID + // must not be queried or signalled by this terminal. + backend.push_probe(ProcessGroupPresence::Present); + let calls_after_retirement = backend.calls(); + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("repeated cleanup after simulated reuse"); + + assert_eq!(backend.calls(), calls_after_retirement); + assert!(backend.delivered_signals().is_empty()); + } + + #[tokio::test] + async fn descendant_lease_expiry_during_probe_retires_before_signal() { + let backend = Arc::new(MockProcessGroupBackend::with_probe_delay( + [ProcessGroupPresence::Present], + Duration::from_millis(10), + )); + let lease = test_process_group_lease( + UnixProcessGroupState::OwnedDescendants { + key: TEST_PROCESS_GROUP_KEY, + deadline: tokio::time::Instant::now() + Duration::from_millis(1), + }, + backend.clone(), + ); + + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("expired descendant group cleanup"); + + assert_eq!( + backend.calls(), + vec![BackendCall::Probe(TEST_PROCESS_GROUP_KEY)] + ); + assert!(backend.delivered_signals().is_empty()); + assert_eq!(*lease.state.lock().await, UnixProcessGroupState::Retired); + } + + #[tokio::test] + async fn descendant_probe_esrch_between_stages_retires_before_next_signal() { + let backend = Arc::new(MockProcessGroupBackend::with_probes([ + ProcessGroupPresence::Present, + ProcessGroupPresence::Present, + ProcessGroupPresence::Missing, + ])); + let lease = test_process_group_lease( + UnixProcessGroupState::OwnedDescendants { + key: TEST_PROCESS_GROUP_KEY, + deadline: tokio::time::Instant::now() + DESCENDANT_LEASE_DURATION, + }, + backend.clone(), + ); + + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("stage cleanup"); + + assert_eq!(backend.delivered_signals(), vec![libc::SIGINT]); + assert_eq!( + backend.calls(), + vec![ + BackendCall::Probe(TEST_PROCESS_GROUP_KEY), + BackendCall::Signal(TEST_PROCESS_GROUP_KEY, libc::SIGINT), + BackendCall::Probe(TEST_PROCESS_GROUP_KEY), + BackendCall::Probe(TEST_PROCESS_GROUP_KEY), + ] + ); + assert_eq!(*lease.state.lock().await, UnixProcessGroupState::Retired); + } + + #[tokio::test] + async fn descendant_signal_esrch_retires_and_repeat_cleanup_is_noop() { + let backend = Arc::new(MockProcessGroupBackend { + probes: std::sync::Mutex::new([ProcessGroupPresence::Present].into_iter().collect()), + signals: std::sync::Mutex::new( + [ProcessGroupSignalResult::Missing].into_iter().collect(), + ), + ..MockProcessGroupBackend::default() + }); + let lease = test_process_group_lease( + UnixProcessGroupState::OwnedDescendants { + key: TEST_PROCESS_GROUP_KEY, + deadline: tokio::time::Instant::now() + DESCENDANT_LEASE_DURATION, + }, + backend.clone(), + ); + + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("signal ESRCH cleanup"); + let calls_after_first_cleanup = backend.calls(); + + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("repeated cleanup after signal ESRCH"); + + assert_eq!(backend.calls(), calls_after_first_cleanup); + assert_eq!( + backend.delivered_signals(), + vec![libc::SIGINT], + "the backend observed the attempted signal, but ESRCH delivered it to no group" + ); + assert_eq!(*lease.state.lock().await, UnixProcessGroupState::Retired); + } + + #[tokio::test] + async fn expired_descendant_lease_retires_without_backend_call() { + let backend = Arc::new(MockProcessGroupBackend::default()); + let lease = test_process_group_lease( + UnixProcessGroupState::OwnedDescendants { + key: TEST_PROCESS_GROUP_KEY, + deadline: tokio::time::Instant::now(), + }, + backend.clone(), + ); + + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("expire descendants lease"); + + assert!(backend.calls().is_empty()); + assert_eq!(*lease.state.lock().await, UnixProcessGroupState::Retired); + } + + #[tokio::test] + async fn retired_lease_ignores_simulated_pgid_reuse() { + let backend = Arc::new(MockProcessGroupBackend::with_probes([ + ProcessGroupPresence::Missing, + ])); + let lease = test_process_group_lease( + UnixProcessGroupState::OwnedLeaderAlive { + key: TEST_PROCESS_GROUP_KEY, + }, + backend.clone(), + ); + lease + .observe_leader_exit(tokio::time::Instant::now()) + .await + .expect("retire original group"); + backend.clear_calls(); + backend.push_probe(ProcessGroupPresence::Present); + + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("cleanup reused numeric pgid"); + + assert!(backend.calls().is_empty()); + assert!(backend.delivered_signals().is_empty()); + } + + #[tokio::test] + async fn esrch_retires_permanently_and_repeat_cleanup_is_noop() { + let backend = Arc::new(MockProcessGroupBackend::with_signals([ + ProcessGroupSignalResult::Missing, + ])); + let lease = test_process_group_lease( + UnixProcessGroupState::OwnedLeaderAlive { + key: TEST_PROCESS_GROUP_KEY, + }, + backend.clone(), + ); + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("first cleanup"); + let calls_after_first = backend.calls(); + + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("repeat cleanup"); + + assert_eq!(backend.calls(), calls_after_first); + assert_eq!(*lease.state.lock().await, UnixProcessGroupState::Retired); + } + + #[tokio::test] + async fn concurrent_cleanup_runs_one_signal_sequence() { + let backend = Arc::new(MockProcessGroupBackend::with_probes([ + ProcessGroupPresence::Present, + ProcessGroupPresence::Present, + ProcessGroupPresence::Missing, + ])); + let lease = Arc::new(test_process_group_lease( + UnixProcessGroupState::OwnedLeaderAlive { + key: TEST_PROCESS_GROUP_KEY, + }, + backend.clone(), + )); + let stages = zero_grace_stages(); + + let (left, right) = tokio::join!( + lease.cleanup_with_stages(&stages, None), + lease.cleanup_with_stages(&stages, None), + ); + + left.expect("left cleanup"); + right.expect("right cleanup"); + assert_eq!( + backend.delivered_signals(), + vec![libc::SIGINT, libc::SIGTERM, libc::SIGKILL] + ); + } + + #[tokio::test] + async fn live_leader_keeps_staged_signal_sequence() { + let backend = Arc::new(MockProcessGroupBackend::with_probes([ + ProcessGroupPresence::Present, + ProcessGroupPresence::Present, + ProcessGroupPresence::Missing, + ])); + let lease = test_process_group_lease( + UnixProcessGroupState::OwnedLeaderAlive { + key: TEST_PROCESS_GROUP_KEY, + }, + backend.clone(), + ); + + lease + .cleanup_with_stages(&zero_grace_stages(), None) + .await + .expect("cleanup live leader"); + + assert_eq!( + backend.delivered_signals(), + vec![libc::SIGINT, libc::SIGTERM, libc::SIGKILL] + ); + } + + #[tokio::test] + async fn leader_exit_with_live_descendants_is_not_retired_immediately() { + let backend = Arc::new(MockProcessGroupBackend::default()); + let lease = test_process_group_lease( + UnixProcessGroupState::OwnedLeaderAlive { + key: TEST_PROCESS_GROUP_KEY, + }, + backend, + ); + + lease + .observe_leader_exit(tokio::time::Instant::now()) + .await + .expect("observe descendants"); + + assert!(matches!( + *lease.state.lock().await, + UnixProcessGroupState::OwnedDescendants { .. } + )); + } + + async fn terminal_pid( + runtime: &TerminalRuntime, + session_id: &SessionId, + terminal_id: &str, + ) -> u32 { + let terminal = runtime + .find_terminal(terminal_id, session_id.0.as_ref()) + .await + .expect("test terminal exists"); + let pid = terminal + .child + .lock() + .await + .as_ref() + .and_then(tokio::process::Child::id) + .expect("test terminal has a direct child pid"); + pid + } + + fn pid_exists(pid: u32) -> bool { + let result = unsafe { libc::kill(pid as libc::pid_t, 0) }; + if result == 0 { + return true; + } + std::io::Error::last_os_error().raw_os_error() != Some(libc::ESRCH) + } + + async fn emergency_kill_exact_tree(pid: u32, terminal: &Arc) { + let pid = pid as libc::pid_t; + let _ = unsafe { libc::kill(pid, libc::SIGSTOP) }; + let config = Config { + signal: "SIGKILL".to_string(), + include_target: true, + }; + let _ = kill_tree::tokio::kill_tree_with_config(pid as u32, &config).await; + let _ = unsafe { libc::kill(pid, libc::SIGKILL) }; + let _ = tokio::time::timeout(Duration::from_secs(1), terminal.wait_for_exit()).await; + for _ in 0..100 { + if !pid_exists(pid as u32) { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + + async fn spawn_shell( + runtime: &TerminalRuntime, + session_id: &SessionId, + script: &str, + ) -> (String, u32) { + let mut request = CreateTerminalRequest::new(session_id.clone(), "/bin/sh".to_string()); + request.args = vec!["-c".into(), script.into()]; + let response = runtime + .create_terminal(request) + .await + .expect("create test terminal"); + let terminal_id = response.terminal_id.to_string(); + let pid = terminal_pid(runtime, session_id, &terminal_id).await; + (terminal_id, pid) + } + + async fn wait_for_output(terminal: &Arc, needle: &str) -> bool { + for _ in 0..100 { + if terminal.snapshot().await.output.contains(needle) { + return true; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + false + } + /// Regression: when an ACP agent calls `terminal/create` (e.g. to run /// `git push`), the runtime's base env — populated by the connection /// layer with the codeg credential helper's `GIT_CONFIG_*` keys — @@ -830,8 +2041,7 @@ mod tests { // Genuine shell operators must evaluate, not be passed as literal args. let session_id = SessionId::new("shell-ops".to_string()); - let request = - CreateTerminalRequest::new(session_id.clone(), "true && echo OK".to_string()); + let request = CreateTerminalRequest::new(session_id.clone(), "true && echo OK".to_string()); let output = run_and_capture(&runtime, &session_id, request).await; assert!( output.contains("OK"), @@ -866,8 +2076,7 @@ mod tests { let runtime = TerminalRuntime::with_base_env(BTreeMap::new()); let session_id = SessionId::new("direct-exec".to_string()); - let mut request = - CreateTerminalRequest::new(session_id.clone(), "/bin/echo".to_string()); + let mut request = CreateTerminalRequest::new(session_id.clone(), "/bin/echo".to_string()); request.args = vec!["hello world".into()]; let output = run_and_capture(&runtime, &session_id, request).await; assert!( @@ -944,11 +2153,16 @@ mod tests { /// check, run in codeg's own cwd, would shell-wrap and split the path. #[tokio::test] async fn relative_executable_with_spaces_runs_in_effective_cwd() { + use std::io::Write; use std::os::unix::fs::PermissionsExt; let dir = tempfile::tempdir().expect("temp dir"); let exe = dir.path().join("my tool"); // space in the file name - std::fs::write(&exe, "#!/bin/sh\necho ran-relative\n").expect("write script"); + let mut file = std::fs::File::create(&exe).expect("create script"); + file.write_all(b"#!/bin/sh\necho ran-relative\n") + .expect("write script"); + file.sync_all().expect("sync script before exec"); + drop(file); let mut perms = std::fs::metadata(&exe).expect("metadata").permissions(); perms.set_mode(0o755); std::fs::set_permissions(&exe, perms).expect("chmod"); @@ -965,4 +2179,183 @@ mod tests { "relative space-containing exe was not run in the effective cwd; got:\n{output}" ); } + + #[tokio::test] + async fn unix_terminal_isolated_in_own_process_group() { + init_test_tracing(); + let runtime = TerminalRuntime::with_base_env(BTreeMap::new()); + let session_id = SessionId::new("pgid-isolation".to_string()); + let (_terminal_id, pid) = spawn_shell(&runtime, &session_id, "sleep 60").await; + + let child_pgid = unsafe { libc::getpgid(pid as libc::pid_t) }; + let runner_pgid = unsafe { libc::getpgrp() }; + runtime.release_all_for_session(session_id.0.as_ref()).await; + + assert_eq!( + child_pgid, pid as libc::pid_t, + "child must lead its own process group" + ); + assert_ne!( + child_pgid, runner_pgid, + "child must not share the test runner process group" + ); + } + + #[tokio::test] + async fn exited_leader_keeps_short_lease_for_live_same_group_descendant() { + init_test_tracing(); + let runtime = TerminalRuntime::with_base_env(BTreeMap::new()); + let session_id = SessionId::new("exited-leader-live-descendant".to_string()); + let (terminal_id, leader_pid) = spawn_shell( + &runtime, + &session_id, + "sleep 60 & child=$!; printf 'descendant=%s\\n' \"$child\"; exit 0", + ) + .await; + let terminal = runtime + .find_terminal(&terminal_id, session_id.0.as_ref()) + .await + .expect("terminal exists"); + assert!( + wait_for_output(&terminal, "descendant=").await, + "shell did not report its test-owned descendant" + ); + let output = terminal.snapshot().await.output; + let descendant_pid = output + .lines() + .find_map(|line| line.strip_prefix("descendant=")) + .and_then(|pid| pid.parse::().ok()) + .expect("parse descendant pid"); + + terminal + .wait_for_exit() + .await + .expect("observe direct child exit"); + let descendant_pgid = unsafe { libc::getpgid(descendant_pid as libc::pid_t) }; + let state_after_leader_exit = *terminal.process_group.state.lock().await; + + assert_eq!( + descendant_pgid, leader_pid as libc::pid_t, + "descendant must remain in the terminal's original process group" + ); + assert!( + pid_exists(descendant_pid), + "descendant must still be alive when the leader exit is observed" + ); + assert!(matches!( + state_after_leader_exit, + UnixProcessGroupState::OwnedDescendants { + key: ProcessGroupKey { pgid, .. }, + deadline, + } if pgid == leader_pid as libc::pid_t && deadline > tokio::time::Instant::now() + )); + + let report = runtime.release_all_for_session(session_id.0.as_ref()).await; + if pid_exists(descendant_pid) { + let _ = unsafe { libc::kill(descendant_pid as libc::pid_t, libc::SIGKILL) }; + } + + assert!(report.is_clean(), "descendant cleanup should succeed"); + assert!( + !pid_exists(descendant_pid), + "same-group descendant survived release inside its lease" + ); + } + + #[tokio::test] + async fn session_cleanup_kills_stubborn_group_without_touching_other_session() { + init_test_tracing(); + let runtime = TerminalRuntime::with_base_env(BTreeMap::new()); + let victim_session = SessionId::new("stubborn-victim".to_string()); + let other_session = SessionId::new("unrelated-session".to_string()); + let (victim_terminal_id, victim_pid) = spawn_shell( + &runtime, + &victim_session, + "trap '' INT TERM; printf 'ready\\n'; while :; do sleep 60 & wait $!; done", + ) + .await; + let victim_terminal = runtime + .find_terminal(&victim_terminal_id, victim_session.0.as_ref()) + .await + .expect("victim terminal exists"); + if !wait_for_output(&victim_terminal, "ready\n").await { + emergency_kill_exact_tree(victim_pid, &victim_terminal).await; + panic!("stubborn shell did not install signal traps in time"); + } + let (_other_terminal, other_pid) = + spawn_shell(&runtime, &other_session, "while :; do sleep 60; done").await; + + let started = Instant::now(); + let cleanup = tokio::time::timeout( + Duration::from_secs(5), + runtime.release_all_for_session(victim_session.0.as_ref()), + ) + .await; + + if cleanup.is_err() { + emergency_kill_exact_tree(victim_pid, &victim_terminal).await; + } + let other_still_alive = pid_exists(other_pid); + runtime + .release_all_for_session(other_session.0.as_ref()) + .await; + + assert!( + cleanup.is_ok(), + "stubborn terminal cleanup exceeded five seconds" + ); + assert!( + !pid_exists(victim_pid), + "victim direct child survived cleanup" + ); + assert!( + other_still_alive, + "cleanup signalled a terminal from another session" + ); + assert!(started.elapsed() < Duration::from_secs(5)); + } + + #[tokio::test] + async fn in_flight_wait_does_not_block_session_cleanup() { + init_test_tracing(); + let runtime = Arc::new(TerminalRuntime::with_base_env(BTreeMap::new())); + let session_id = SessionId::new("wait-does-not-lock-cancel".to_string()); + let (terminal_id, pid) = spawn_shell(&runtime, &session_id, "sleep 60").await; + let terminal = runtime + .find_terminal(&terminal_id, session_id.0.as_ref()) + .await + .expect("terminal exists"); + + let waiter_runtime = Arc::clone(&runtime); + let waiter_session = session_id.clone(); + let waiter_terminal = terminal_id.clone(); + let wait_task = tokio::spawn(async move { + waiter_runtime + .wait_for_terminal_exit(WaitForTerminalExitRequest::new( + waiter_session, + waiter_terminal, + )) + .await + }); + tokio::time::sleep(Duration::from_millis(100)).await; + + let cleanup = tokio::time::timeout( + Duration::from_secs(5), + runtime.release_all_for_session(session_id.0.as_ref()), + ) + .await; + if cleanup.is_err() { + emergency_kill_exact_tree(pid, &terminal).await; + } + let wait_result = tokio::time::timeout(Duration::from_secs(1), wait_task).await; + + assert!( + cleanup.is_ok(), + "terminal wait held the child mutex across cancellation" + ); + assert!( + wait_result.is_ok(), + "terminal wait did not observe cancellation exit" + ); + } }