systemprompt_loader/subprocess/
control.rs

1//! Async process control: liveness, ownership, port holders, termination.
2//!
3//! Everything a supervisor needs to decide whether a recorded pid is still
4//! one of ours and to stop it. Blocking work — identity reads that retry,
5//! `lsof`/`netstat`, every Windows primitive — runs under `spawn_blocking`;
6//! the POSIX signal and reap syscalls are non-blocking and run inline.
7//!
8//! A child this process spawned and that has since exited stays a zombie until
9//! it is reaped, and `kill(pid, 0)` reports a zombie as alive. Liveness here
10//! therefore reaps an exited child of ours first and then treats a zombie as
11//! exited, so a stop never waits out its grace period on a corpse.
12//!
13//! Copyright (c) systemprompt.io — Business Source License 1.1.
14//! See <https://systemprompt.io> for licensing details.
15
16use std::time::Duration;
17
18use async_trait::async_trait;
19use systemprompt_identifiers::ServiceName;
20use systemprompt_traits::ProcessLiveness;
21
22use super::{ChildKind, SupervisionError};
23
24#[cfg(unix)]
25use super::posix as platform;
26#[cfg(windows)]
27use super::winnt as platform;
28
29const EXIT_POLL: Duration = Duration::from_millis(25);
30const KILL_SETTLE: Duration = Duration::from_secs(5);
31
32/// How a termination request ended.
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
34pub enum Termination {
35    AlreadyExited,
36    Exited,
37    Killed,
38}
39
40/// How a stop of a recorded, marker-verified child ended.
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub enum StopOutcome {
43    NotRunning,
44    NotOurs,
45    Stopped(Termination),
46}
47
48pub async fn is_running(pid: u32) -> bool {
49    blocking_on_windows(move || !platform::has_exited(pid))
50        .await
51        .unwrap_or(false)
52}
53
54pub async fn owns(pid: u32, kind: ChildKind, service: &ServiceName) -> bool {
55    let service = service.clone();
56    let verdict = tokio::task::spawn_blocking(move || {
57        !platform::has_exited(pid)
58            && super::live_environ(pid).is_some_and(|environ| kind.identifies(&environ, &service))
59    })
60    .await;
61    match verdict {
62        Ok(owned) => owned,
63        Err(e) => {
64            tracing::warn!(pid, error = %e, "Child identity check did not complete; treating as not ours");
65            false
66        },
67    }
68}
69
70/// [`ProcessLiveness`] over this module's spawn markers: a pid is ours when it
71/// is alive and identifies as a supervised agent or MCP child named `service`.
72#[derive(Debug, Clone, Copy, Default)]
73pub struct SubprocessLiveness;
74
75#[async_trait]
76impl ProcessLiveness for SubprocessLiveness {
77    async fn owns(&self, pid: u32, service: &ServiceName) -> bool {
78        owns(pid, ChildKind::Agent, service).await || owns(pid, ChildKind::Mcp, service).await
79    }
80}
81
82pub async fn pids_listening_on(port: u16) -> Result<Vec<u32>, SupervisionError> {
83    if port == 0 {
84        return Ok(Vec::new());
85    }
86    tokio::task::spawn_blocking(move || super::ports::listening_pids(port)).await?
87}
88
89pub async fn process_group(pid: u32) -> Option<u32> {
90    blocking_on_windows(move || platform::process_group(pid))
91        .await
92        .ok()
93        .flatten()
94}
95
96pub async fn terminate_gracefully(
97    pid: u32,
98    grace: Duration,
99) -> Result<Termination, SupervisionError> {
100    terminate(pid, grace, false).await
101}
102
103pub async fn terminate_group_gracefully(
104    pid: u32,
105    grace: Duration,
106) -> Result<Termination, SupervisionError> {
107    let leader = blocking_on_windows(move || platform::is_group_leader(pid)).await?;
108    terminate(pid, grace, leader).await
109}
110
111pub async fn stop_owned(
112    pid: u32,
113    kind: ChildKind,
114    service: &ServiceName,
115    grace: Duration,
116) -> Result<StopOutcome, SupervisionError> {
117    if !is_running(pid).await {
118        return Ok(StopOutcome::NotRunning);
119    }
120    if !owns(pid, kind, service).await {
121        tracing::warn!(
122            pid,
123            service = %service,
124            "Recorded pid is alive but carries no matching child marker; leaving it untouched"
125        );
126        return Ok(StopOutcome::NotOurs);
127    }
128    terminate_group_gracefully(pid, grace)
129        .await
130        .map(StopOutcome::Stopped)
131}
132
133async fn terminate(
134    pid: u32,
135    grace: Duration,
136    group: bool,
137) -> Result<Termination, SupervisionError> {
138    if !is_running(pid).await {
139        return Ok(Termination::AlreadyExited);
140    }
141    let delivery = blocking_on_windows(move || platform::send(pid, false, group)).await??;
142    if delivery == platform::Delivery::NoSuchProcess || wait_for_exit(pid, grace).await {
143        return Ok(Termination::Exited);
144    }
145    let delivery = blocking_on_windows(move || platform::send(pid, true, group)).await??;
146    if delivery == platform::Delivery::NoSuchProcess || wait_for_exit(pid, KILL_SETTLE).await {
147        return Ok(Termination::Killed);
148    }
149    Err(SupervisionError::Survived { pid })
150}
151
152async fn wait_for_exit(pid: u32, within: Duration) -> bool {
153    let deadline = tokio::time::Instant::now() + within;
154    loop {
155        if !is_running(pid).await {
156            return true;
157        }
158        if tokio::time::Instant::now() >= deadline {
159            return false;
160        }
161        tokio::time::sleep(EXIT_POLL).await;
162    }
163}
164
165#[cfg(unix)]
166fn blocking_on_windows<T, F>(work: F) -> impl Future<Output = Result<T, SupervisionError>>
167where
168    F: FnOnce() -> T + Send + 'static,
169    T: Send + 'static,
170{
171    std::future::ready(Ok(work()))
172}
173
174#[cfg(windows)]
175fn blocking_on_windows<T, F>(work: F) -> impl Future<Output = Result<T, SupervisionError>>
176where
177    F: FnOnce() -> T + Send + 'static,
178    T: Send + 'static,
179{
180    async move { Ok(tokio::task::spawn_blocking(work).await?) }
181}
systemprompt_loader/subprocess/control.rs

systemprompt_loader/subprocess/
control.rs
