gate4agent_wake/
spawn.rs

1//! Last resort: start a NEW provider process that resumes the session for
2//! one turn. Only for a session marked headless (no client holds it open).
3//! A session that has a live client is never spawned into: two writers on
4//! one transcript is how sessions get corrupted.
5//!
6//! | Kind | Command (prompt never in argv when the CLI reads stdin) |
7//! |---|---|
8//! | grok/local | `grok --resume <id> -p <prompt>` |
9//! | codex/local | `codex exec resume <id> -` (prompt on stdin) |
10//! | kimi_code/local | `kimi -S <id> -p <prompt>` |
11//! | claude_code/local | `claude --resume <id> -p` (prompt on stdin) |
12//! | cursor/local | `agent --resume <id> -p <prompt>` |
13//! | codex/web | `codex cloud exec --env <env> <prompt>` (new cloud task) |
14//!
15//! The binary is `M4A_<PROVIDER>_BIN` or the vendor name on `PATH`.
16
17use std::io::Write;
18use std::path::PathBuf;
19use std::process::{Command, Stdio};
20
21use super::{
22    wake_prompt, ProviderKind, ProviderSession, SessionKind, Surface, WakeAdapter, WakeError,
23    WakeLetter, WakeOutcome,
24};
25
26/// Env var naming the cloud environment for `codex cloud exec --env`.
27pub const CODEX_CLOUD_ENV_ENV: &str = "M4A_CODEX_CLOUD_ENV";
28
29/// Env override for a provider binary (`M4A_CODEX_BIN`, ...).
30pub fn bin_env(provider: ProviderKind) -> &'static str {
31    match provider {
32        ProviderKind::Grok => "M4A_GROK_BIN",
33        ProviderKind::KimiCode => "M4A_KIMI_BIN",
34        ProviderKind::ClaudeCode => "M4A_CLAUDE_BIN",
35        ProviderKind::Codex => "M4A_CODEX_BIN",
36        ProviderKind::Cursor => "M4A_CURSOR_AGENT_BIN",
37    }
38}
39
40/// Default program name on `PATH`.
41pub fn default_program(provider: ProviderKind) -> &'static str {
42    match provider {
43        ProviderKind::Grok => "grok",
44        ProviderKind::KimiCode => "kimi",
45        ProviderKind::ClaudeCode => "claude",
46        ProviderKind::Codex => "codex",
47        ProviderKind::Cursor => "agent",
48    }
49}
50
51/// Argv (after the program) and the stdin payload for one resume turn.
52pub fn resume_command(
53    kind: SessionKind,
54    session_id: &str,
55    prompt: &str,
56    cloud_env: Option<&str>,
57) -> Result<(Vec<String>, Option<String>), WakeError> {
58    let id = session_id.to_string();
59    let p = prompt.to_string();
60    Ok(match (kind.surface, kind.provider) {
61        (Surface::Local, ProviderKind::Grok) => (vec!["--resume".into(), id, "-p".into(), p], None),
62        (Surface::Local, ProviderKind::Codex) => (
63            vec!["exec".into(), "resume".into(), id, "-".into()],
64            Some(p),
65        ),
66        (Surface::Local, ProviderKind::KimiCode) => (vec!["-S".into(), id, "-p".into(), p], None),
67        (Surface::Local, ProviderKind::ClaudeCode) => {
68            (vec!["--resume".into(), id, "-p".into()], Some(p))
69        }
70        (Surface::Local, ProviderKind::Cursor) => {
71            (vec!["--resume".into(), id, "-p".into(), p], None)
72        }
73        (Surface::Web, ProviderKind::Codex) => {
74            let Some(env) = cloud_env.filter(|env| !env.is_empty()) else {
75                return Err(WakeError::Unavailable(format!(
76                    "{CODEX_CLOUD_ENV_ENV} unset"
77                )));
78            };
79            (
80                vec![
81                    "cloud".into(),
82                    "exec".into(),
83                    "--env".into(),
84                    env.to_string(),
85                    p,
86                ],
87                None,
88            )
89        }
90        _ => return Err(WakeError::NoInboundTrigger(kind)),
91    })
92}
93
94/// Spawns one resume turn for a headless session. Does not wait for it.
95pub struct ResumeSpawnAdapter {
96    kind: SessionKind,
97    program: Option<PathBuf>,
98    cloud_env: Option<String>,
99}
100
101impl ResumeSpawnAdapter {
102    /// Adapter for `kind`. `program` overrides the binary.
103    pub fn new(kind: SessionKind, program: Option<PathBuf>, cloud_env: Option<String>) -> Self {
104        Self {
105            kind,
106            program,
107            cloud_env,
108        }
109    }
110
111    /// Program from `M4A_<PROVIDER>_BIN` or `PATH`.
112    pub fn from_env(kind: SessionKind) -> Self {
113        let program = std::env::var_os(bin_env(kind.provider))
114            .filter(|value| !value.is_empty())
115            .map(PathBuf::from);
116        let cloud_env = std::env::var(CODEX_CLOUD_ENV_ENV)
117            .ok()
118            .filter(|v| !v.is_empty());
119        Self::new(kind, program, cloud_env)
120    }
121
122    fn program(&self) -> PathBuf {
123        self.program
124            .clone()
125            .unwrap_or_else(|| PathBuf::from(default_program(self.kind.provider)))
126    }
127}
128
129impl WakeAdapter for ResumeSpawnAdapter {
130    fn kind(&self) -> SessionKind {
131        self.kind
132    }
133
134    fn probe(&self, session: &ProviderSession) -> Result<(), WakeError> {
135        if !session.headless {
136            return Err(WakeError::Unavailable(
137                "session has a live client; resume spawn is headless-only".into(),
138            ));
139        }
140        resume_command(
141            self.kind,
142            &session.session_id,
143            "",
144            self.cloud_env.as_deref(),
145        )
146        .map(|_| ())
147    }
148
149    fn wake(
150        &mut self,
151        session: &ProviderSession,
152        letter: &WakeLetter<'_>,
153    ) -> Result<WakeOutcome, WakeError> {
154        self.probe(session)?;
155        let prompt = wake_prompt(session, letter);
156        let (args, stdin) = resume_command(
157            self.kind,
158            &session.session_id,
159            &prompt,
160            self.cloud_env.as_deref(),
161        )?;
162        let mut command = Command::new(self.program());
163        command
164            .args(&args)
165            .stdin(if stdin.is_some() {
166                Stdio::piped()
167            } else {
168                Stdio::null()
169            })
170            .stdout(Stdio::null())
171            .stderr(Stdio::null());
172        #[cfg(windows)]
173        {
174            use std::os::windows::process::CommandExt;
175            command.creation_flags(0x0800_0000); // CREATE_NO_WINDOW
176        }
177        if let Some(cwd) = session.cwd.as_deref().filter(|cwd| cwd.is_dir()) {
178            command.current_dir(cwd);
179        }
180        let mut child = command
181            .spawn()
182            .map_err(|err| WakeError::Unavailable(format!("spawn: {}", err.kind())))?;
183        if let (Some(text), Some(mut pipe)) = (stdin, child.stdin.take()) {
184            pipe.write_all(text.as_bytes())
185                .map_err(|err| WakeError::Transport(format!("spawn stdin: {}", err.kind())))?;
186        }
187        // Reap in the background; the turn runs on its own.
188        std::thread::spawn(move || {
189            let _ = child.wait();
190        });
191        Ok(WakeOutcome::Delivered)
192    }
193}
194
195#[cfg(test)]
196mod tests {
197    use super::*;
198    use crate::tests::{letter, session};
199
200    #[test]
201    fn commands_keep_prompt_off_argv_where_the_cli_reads_stdin() {
202        let (args, stdin) =
203            resume_command(SessionKind::local(ProviderKind::Codex), "t1", "hi", None).unwrap();
204        assert_eq!(args, ["exec", "resume", "t1", "-"]);
205        assert_eq!(stdin.as_deref(), Some("hi"));
206        let (args, stdin) = resume_command(
207            SessionKind::local(ProviderKind::ClaudeCode),
208            "c1",
209            "hi",
210            None,
211        )
212        .unwrap();
213        assert!(!args.iter().any(|a| a == "hi") && stdin.is_some());
214        assert!(resume_command(SessionKind::web(ProviderKind::Codex), "x", "hi", None).is_err());
215        assert!(resume_command(SessionKind::web(ProviderKind::KimiCode), "x", "hi", None).is_err());
216    }
217
218    #[test]
219    fn never_spawns_into_a_session_with_a_live_client() {
220        let kind = SessionKind::local(ProviderKind::KimiCode);
221        let mut adapter = ResumeSpawnAdapter::new(kind, Some("/bin/false".into()), None);
222        let s = session(kind);
223        assert!(!s.headless);
224        let err = adapter.wake(&s, &letter("x")).unwrap_err().to_string();
225        assert!(err.contains("headless-only"), "{err}");
226    }
227
228    #[cfg(unix)]
229    #[test]
230    fn headless_session_gets_one_resume_process() {
231        let dir = std::env::temp_dir().join(format!("m4a-spawn-{}", std::process::id()));
232        let _ = std::fs::create_dir_all(&dir);
233        let out = dir.join("argv");
234        let script = dir.join("fake-codex");
235        std::fs::write(
236            &script,
237            format!(
238                "#!/bin/sh\necho \"$@\" > {0}.tmp\ncat >> {0}.tmp\nmv {0}.tmp {0}\n",
239                out.display()
240            ),
241        )
242        .unwrap();
243        use std::os::unix::fs::PermissionsExt;
244        std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
245        let kind = SessionKind::local(ProviderKind::Codex);
246        let mut adapter = ResumeSpawnAdapter::new(kind, Some(script), None);
247        let mut s = session(kind);
248        s.headless = true;
249        assert_eq!(
250            adapter.wake(&s, &letter("ping")).unwrap(),
251            WakeOutcome::Delivered
252        );
253        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
254        while !out.exists() && std::time::Instant::now() < deadline {
255            std::thread::sleep(std::time::Duration::from_millis(20));
256        }
257        let text = std::fs::read_to_string(&out).unwrap();
258        assert!(text.starts_with("exec resume s-1 -"));
259        assert!(text.trim_end().ends_with("ping"));
260        let _ = std::fs::remove_dir_all(&dir);
261    }
262}
gate4agent_wake/spawn.rs

gate4agent_wake/
spawn.rs
