systemprompt_loader/subprocess/
spawn.rs

1//! The one sanctioned way to start an agent or MCP child, and the in-place
2//! re-exec that stamps the API server with its identity marker.
3//!
4//! Every spawn runs on one dedicated thread and, where the platform offers
5//! it, the kernel is asked to `SIGTERM` the child if this process dies. The
6//! parent-death signal follows the thread that forked the child, which is why
7//! the spawner is a single long-lived thread rather than whichever runtime
8//! worker happened to call: a worker that exits would take its children with
9//! it. [`Spawner`] owns that thread: it is started once, at the composition
10//! root, and joined when the last handle to the spawner is dropped.
11//!
12//! Copyright (c) systemprompt.io — Business Source License 1.1.
13//! See <https://systemprompt.io> for licensing details.
14
15use std::process::{Child, Command};
16use std::sync::mpsc::{Sender, channel};
17use std::thread::JoinHandle;
18
19use systemprompt_identifiers::ServiceName;
20
21use super::ChildKind;
22
23type SpawnReply = Sender<std::io::Result<Child>>;
24type SpawnRequest = (Command, SpawnReply);
25
26#[derive(Debug)]
27pub struct Spawner {
28    requests: Option<Sender<SpawnRequest>>,
29    thread: Option<JoinHandle<()>>,
30}
31
32/// How [`stamp_api_server`] returned instead of re-executing.
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
34pub enum ApiServerStamp {
35    Stamped,
36    Unverifiable,
37}
38
39pub fn mark_child(cmd: &mut Command, kind: ChildKind, service: &ServiceName) {
40    match kind {
41        ChildKind::Agent | ChildKind::Mcp => {
42            cmd.env(systemprompt_models::subprocess::SUBPROCESS_MARKER_ENV, "1");
43        },
44        ChildKind::Api => {},
45    }
46    cmd.env(kind.marker_env(), service.as_str());
47}
48
49// Why: the API server is launched by an operator, a service manager or a
50// container runtime, never spawned by us, so the marker that later proves its
51// identity cannot be stamped at spawn. `execve` keeps the pid, so replacing
52// the image once with the marker added is invisible to whatever supervises
53// the process. An environment that cannot be read back (no platform probe, a
54// hardened binary) is left unstamped rather than re-executed in a loop.
55#[cfg(unix)]
56pub fn stamp_api_server() -> std::io::Result<ApiServerStamp> {
57    use std::os::unix::process::CommandExt;
58
59    let service = super::api_server_service();
60    let Some(environ) = super::live_environ(std::process::id())
61        .filter(|environ| systemprompt_models::subprocess::environ_has_entries(environ))
62    else {
63        return Ok(ApiServerStamp::Unverifiable);
64    };
65    if ChildKind::Api.identifies(&environ, &service) {
66        return Ok(ApiServerStamp::Stamped);
67    }
68    let mut cmd = Command::new(std::env::current_exe()?);
69    cmd.args(std::env::args_os().skip(1));
70    mark_child(&mut cmd, ChildKind::Api, &service);
71    Err(cmd.exec())
72}
73
74#[cfg(not(unix))]
75pub const fn stamp_api_server() -> std::io::Result<ApiServerStamp> {
76    Ok(ApiServerStamp::Unverifiable)
77}
78
79impl Spawner {
80    pub fn new() -> std::io::Result<Self> {
81        let (tx, rx) = channel::<SpawnRequest>();
82        let thread = std::thread::Builder::new()
83            .name("subprocess-spawner".to_owned())
84            .spawn(move || {
85                while let Ok((mut cmd, reply)) = rx.recv() {
86                    let outcome = spawn_on_this_thread(&mut cmd);
87                    if let Err(undelivered) = reply.send(outcome)
88                        && let Ok(mut child) = undelivered.0
89                    {
90                        if let Err(error) = child.kill() {
91                            tracing::warn!(error = %error, "Failed to stop unclaimed subprocess");
92                        }
93                        if let Err(error) = child.wait() {
94                            tracing::warn!(error = %error, "Failed to reap unclaimed subprocess");
95                        }
96                    }
97                }
98            })?;
99        Ok(Self {
100            requests: Some(tx),
101            thread: Some(thread),
102        })
103    }
104
105    pub fn spawn_supervised(&self, cmd: Command) -> std::io::Result<u32> {
106        let child = self.spawn_owned_supervised(cmd)?;
107        let pid = child.id();
108        drop(child);
109        Ok(pid)
110    }
111
112    pub fn spawn_owned_supervised(&self, cmd: Command) -> std::io::Result<Child> {
113        let sender = self
114            .requests
115            .as_ref()
116            .ok_or_else(|| std::io::Error::other("subprocess spawner is shut down"))?;
117        let (reply_tx, reply_rx) = channel();
118        sender
119            .send((cmd, reply_tx))
120            .map_err(std::io::Error::other)?;
121        reply_rx.recv().map_err(std::io::Error::other)?
122    }
123}
124
125impl Drop for Spawner {
126    fn drop(&mut self) {
127        self.requests.take();
128        if let Some(thread) = self.thread.take()
129            && thread.join().is_err()
130        {
131            tracing::warn!("Subprocess spawner thread panicked");
132        }
133    }
134}
135
136fn spawn_on_this_thread(cmd: &mut Command) -> std::io::Result<Child> {
137    #[cfg(target_os = "linux")]
138    super::linux::arm_parent_death_signal(cmd);
139    cmd.spawn()
140}
141
142// Why: On Unix, process group 0 assigns the child's PID as its process group
143// ID.
144#[cfg(unix)]
145pub fn place_in_own_process_group(command: &mut Command) {
146    use std::os::unix::process::CommandExt;
147    command.process_group(0);
148}
149
150#[cfg(windows)]
151pub fn place_in_own_process_group(command: &mut Command) {
152    use std::os::windows::process::CommandExt;
153    const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200;
154    command.creation_flags(CREATE_NEW_PROCESS_GROUP);
155}
systemprompt_loader/subprocess/spawn.rs

systemprompt_loader/subprocess/
spawn.rs
