sloop/runner/
local.rs

1use std::fs;
2use std::io::{self, Read, Write};
3use std::os::unix::fs::PermissionsExt;
4use std::os::unix::process::CommandExt;
5use std::path::{Path, PathBuf};
6use std::process::{Child, Command, Stdio};
7
8use base64::Engine;
9use base64::engine::general_purpose::URL_SAFE_NO_PAD as BASE64_TOKEN;
10use tokio::net::UnixListener;
11
12use super::{
13    AgentProcessCheckpoint, ExecProcessCheckpoint, ExecutionEvidence, ExecutionFailure,
14    ProcessIdentity, RunnerError, StageExecution, StageHooks, StageOrder, WorkerCredentials,
15};
16use crate::clock::Clock;
17use crate::run_log::{OutputSource, OutputStream, RunLogWriter};
18
19const WORKER_BOOTSTRAP_PROMPT: &str = include_str!("../worker-instructions.md").trim_ascii();
20
21/// A launched agent whose worker socket can be registered before supervision
22/// moves to a blocking task.
23pub struct LaunchedAgent {
24    child: Child,
25    readers: Vec<std::thread::JoinHandle<bool>>,
26    process: ProcessIdentity,
27    started_at_ms: i64,
28    worker_listener: Option<UnixListener>,
29    worker: WorkerCredentials,
30}
31
32pub struct AgentCompletion {
33    pub evidence: ExecutionEvidence,
34    pub wait_error: Option<String>,
35}
36
37impl LaunchedAgent {
38    pub fn process(&self) -> ProcessIdentity {
39        self.process
40    }
41
42    pub fn worker(&self) -> &WorkerCredentials {
43        &self.worker
44    }
45
46    pub fn take_worker_listener(&mut self) -> UnixListener {
47        self.worker_listener
48            .take()
49            .expect("worker listener is taken only once")
50    }
51
52    pub fn wait(mut self, clock: &dyn Clock) -> AgentCompletion {
53        let (exit_code, wait_error) = match self.child.wait() {
54            Ok(status) => (status.code(), None),
55            Err(error) => (None, Some(error.to_string())),
56        };
57        let stragglers_killed = kill_straggler_process_group(self.process.process_group_id);
58        let mut output_capture_complete = true;
59        for reader in self.readers {
60            output_capture_complete &= reader.join().unwrap_or(false);
61        }
62        AgentCompletion {
63            evidence: ExecutionEvidence {
64                exit_code,
65                started_at_ms: self.started_at_ms,
66                finished_at_ms: clock.now_ms(),
67                process: Some(self.process),
68                output_capture_complete,
69                stragglers_killed,
70            },
71            wait_error,
72        }
73    }
74}
75
76pub fn launch_agent<H: StageHooks>(
77    order: StageOrder,
78    hooks: &H,
79    clock: &dyn Clock,
80) -> Result<LaunchedAgent, RunnerError<H::Error>> {
81    let StageExecution::Agent(launch) = &order.execution else {
82        return Err(RunnerError::Execution(
83            "agent launch requires an agent stage order".into(),
84        ));
85    };
86    let Some(program) = launch.argv.first() else {
87        return Err(RunnerError::Execution("agent argv is empty".into()));
88    };
89
90    fs::create_dir_all(
91        order
92            .worktree
93            .parent()
94            .ok_or_else(|| RunnerError::Execution("worktree has no parent".into()))?,
95    )
96    .map_err(|error| RunnerError::Execution(error.to_string()))?;
97    let git = Command::new("git")
98        .args(["worktree", "add", "--quiet", "-b", &order.branch])
99        .arg(&order.worktree)
100        .current_dir(&launch.repository)
101        .output()
102        .map_err(|error| RunnerError::Execution(error.to_string()))?;
103    if !git.status.success() {
104        return Err(RunnerError::Execution(format!(
105            "git worktree add failed: {}",
106            String::from_utf8_lossy(&git.stderr).trim()
107        )));
108    }
109
110    let output_log = RunLogWriter::open(&order.output_path)
111        .map_err(|error| RunnerError::Execution(error.to_string()))?;
112    let worker_token = generate_worker_token().map_err(RunnerError::Execution)?;
113    let socket_path = &launch.worker_socket_path;
114    fs::create_dir_all(socket_path.parent().expect("worker sockets have a parent"))
115        .map_err(|error| RunnerError::Execution(error.to_string()))?;
116    let _ = fs::remove_file(socket_path);
117    let worker_listener = UnixListener::bind(socket_path)
118        .map_err(|error| RunnerError::Execution(error.to_string()))?;
119    fs::set_permissions(socket_path, fs::Permissions::from_mode(0o600))
120        .map_err(|error| RunnerError::Execution(error.to_string()))?;
121
122    let mut command = Command::new(program);
123    command
124        .args(&launch.argv[1..])
125        .current_dir(&order.worktree)
126        .envs(launch.environment.iter().cloned())
127        .env("SLOOP_SOCKET", socket_path)
128        .env("SLOOP_TOKEN", &worker_token)
129        .stdin(Stdio::null())
130        .stdout(Stdio::piped())
131        .stderr(Stdio::piped())
132        .process_group(0);
133    let mut child = command
134        .spawn()
135        .map_err(|error| RunnerError::Execution(error.to_string()))?;
136    let readers = vec![
137        spawn_output_reader(
138            child.stdout.take().expect("stdout was piped"),
139            output_log.clone(),
140            OutputSource::Agent,
141            None,
142            OutputStream::Stdout,
143        ),
144        spawn_output_reader(
145            child.stderr.take().expect("stderr was piped"),
146            output_log,
147            OutputSource::Agent,
148            None,
149            OutputStream::Stderr,
150        ),
151    ];
152
153    let pid = child.id();
154    let process = ProcessIdentity {
155        pid,
156        start_time: process_start_time(pid),
157        process_group_id: pid,
158    };
159    let started_at_ms = clock.now_ms();
160    let worker = WorkerCredentials {
161        socket: socket_path.clone(),
162        token: worker_token,
163    };
164    let checkpoint = AgentProcessCheckpoint {
165        run_id: order.run_id,
166        branch: order.branch,
167        worktree: order.worktree,
168        process,
169        worker: worker.clone(),
170        started_at_ms,
171    };
172    if let Err(error) = hooks.record_agent_process(&checkpoint) {
173        kill_process_group(process);
174        let _ = child.wait();
175        for reader in readers {
176            let _ = reader.join();
177        }
178        return Err(RunnerError::Hook(error));
179    }
180
181    Ok(LaunchedAgent {
182        child,
183        readers,
184        process,
185        started_at_ms,
186        worker_listener: Some(worker_listener),
187        worker,
188    })
189}
190
191pub fn run_exec_stage<H: StageHooks>(
192    order: &StageOrder,
193    hooks: &H,
194    clock: &dyn Clock,
195) -> Result<ExecutionEvidence, ExecutionFailure<H::Error>> {
196    let started_at_ms = clock.now_ms();
197    let failed = |process, error| ExecutionFailure {
198        evidence: ExecutionEvidence {
199            exit_code: None,
200            started_at_ms,
201            finished_at_ms: clock.now_ms(),
202            process,
203            output_capture_complete: false,
204            stragglers_killed: false,
205        },
206        error,
207    };
208    let StageExecution::Exec(launch) = &order.execution else {
209        return Err(failed(
210            None,
211            RunnerError::Execution("exec launch requires an exec stage order".into()),
212        ));
213    };
214    let Some(program) = launch.argv.first() else {
215        return Err(failed(
216            None,
217            RunnerError::Execution("exec argv is empty".into()),
218        ));
219    };
220    if hooks.cancellation_requested(&order.run_id) {
221        return Err(failed(
222            None,
223            RunnerError::Execution("stage cancelled before launch".into()),
224        ));
225    }
226    let output_log = RunLogWriter::open(&order.output_path).map_err(|error| {
227        failed(
228            None,
229            RunnerError::Execution(format!("cannot open run output: {error}")),
230        )
231    })?;
232
233    let mut command = if launch.worker.is_some() {
234        let mut command = Command::new("sh");
235        command
236            .args([
237                "-c",
238                "IFS= read -r _ || exit 125; exec \"$@\"",
239                "sloop-stage",
240            ])
241            .args(&launch.argv);
242        command
243    } else {
244        let mut command = Command::new(program);
245        command.args(&launch.argv[1..]);
246        command
247    };
248    command
249        .current_dir(&order.worktree)
250        .envs(launch.environment.iter().cloned())
251        .stdout(Stdio::piped())
252        .stderr(Stdio::piped())
253        .process_group(0);
254    if let Some(worker) = &launch.worker {
255        command
256            .env("SLOOP_SOCKET", &worker.socket)
257            .env("SLOOP_TOKEN", &worker.token)
258            .stdin(Stdio::piped());
259    } else {
260        command
261            .env_remove("SLOOP_SOCKET")
262            .env_remove("SLOOP_TOKEN")
263            .stdin(Stdio::null());
264    }
265    let mut child = command
266        .spawn()
267        .map_err(|error| failed(None, RunnerError::Execution(error.to_string())))?;
268    let mut gate = launch
269        .worker
270        .as_ref()
271        .map(|_| child.stdin.take().expect("reported stage stdin was piped"));
272    let pid = child.id();
273    let process = ProcessIdentity {
274        pid,
275        start_time: process_start_time(pid),
276        process_group_id: pid,
277    };
278    let Some(start_time) = process.start_time else {
279        kill_process_group(process);
280        let _ = child.wait();
281        return Err(failed(
282            Some(process),
283            RunnerError::Execution("cannot identify exec process".into()),
284        ));
285    };
286    let readers = vec![
287        spawn_output_reader(
288            child.stdout.take().expect("stdout was piped"),
289            output_log.clone(),
290            OutputSource::Aftercare,
291            Some(order.stage.clone()),
292            OutputStream::Stdout,
293        ),
294        spawn_output_reader(
295            child.stderr.take().expect("stderr was piped"),
296            output_log,
297            OutputSource::Aftercare,
298            Some(order.stage.clone()),
299            OutputStream::Stderr,
300        ),
301    ];
302    if order.stage == "test" {
303        wait_for_test_hook("before-test-process-checkpoint");
304    }
305    wait_for_test_hook(&format!(
306        "before-aftercare-process-checkpoint-{}",
307        order.stage
308    ));
309    let checkpoint = ExecProcessCheckpoint {
310        run_id: order.run_id.clone(),
311        stage: order.stage.clone(),
312        process: ProcessIdentity {
313            start_time: Some(start_time),
314            ..process
315        },
316        started_at_ms: clock.now_ms(),
317    };
318    if let Err(error) = hooks.record_exec_process(&checkpoint) {
319        kill_process_group_if_matches(checkpoint.process);
320        let _ = child.wait();
321        join_readers(readers);
322        return Err(failed(Some(process), RunnerError::Hook(error)));
323    }
324    wait_for_test_hook(&format!(
325        "after-aftercare-process-checkpoint-{}",
326        order.stage
327    ));
328    if hooks.cancellation_requested(&order.run_id) {
329        kill_process_group_if_matches(checkpoint.process);
330        let _ = child.wait();
331        join_readers(readers);
332        return Err(failed(
333            Some(process),
334            RunnerError::Execution("stage cancelled after launch".into()),
335        ));
336    }
337    if let Some(gate) = gate.as_mut()
338        && gate.write_all(b"run\n").is_err()
339    {
340        kill_process_group_if_matches(checkpoint.process);
341        let _ = child.wait();
342        join_readers(readers);
343        return Err(failed(
344            Some(process),
345            RunnerError::Execution("cannot release exec process".into()),
346        ));
347    }
348    drop(gate);
349
350    let status = child.wait();
351    let stragglers_killed = kill_straggler_process_group(pid);
352    let output_capture_complete = join_readers(readers);
353    if !output_capture_complete {
354        return Err(ExecutionFailure {
355            evidence: ExecutionEvidence {
356                exit_code: None,
357                started_at_ms,
358                finished_at_ms: clock.now_ms(),
359                process: Some(process),
360                output_capture_complete: false,
361                stragglers_killed,
362            },
363            error: RunnerError::Execution("exec output capture was incomplete".into()),
364        });
365    }
366    let status = match status {
367        Ok(status) => status,
368        Err(error) => {
369            return Err(ExecutionFailure {
370                evidence: ExecutionEvidence {
371                    exit_code: None,
372                    started_at_ms,
373                    finished_at_ms: clock.now_ms(),
374                    process: Some(process),
375                    output_capture_complete: true,
376                    stragglers_killed,
377                },
378                error: RunnerError::Execution(error.to_string()),
379            });
380        }
381    };
382    Ok(ExecutionEvidence {
383        exit_code: status.code(),
384        started_at_ms,
385        finished_at_ms: clock.now_ms(),
386        process: Some(process),
387        output_capture_complete: true,
388        stragglers_killed,
389    })
390}
391
392fn spawn_output_reader(
393    pipe: impl Read + Send + 'static,
394    log: RunLogWriter,
395    source: OutputSource,
396    stage: Option<String>,
397    stream: OutputStream,
398) -> std::thread::JoinHandle<bool> {
399    std::thread::spawn(move || {
400        let mut pipe = pipe;
401        let mut buffer = [0u8; 8192];
402        loop {
403            match pipe.read(&mut buffer) {
404                Ok(0) => return true,
405                Ok(read) => {
406                    if log
407                        .append(source, stage.as_deref(), stream, &buffer[..read])
408                        .is_err()
409                    {
410                        return false;
411                    }
412                }
413                Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
414                Err(_) => return false,
415            }
416        }
417    })
418}
419
420fn join_readers(readers: Vec<std::thread::JoinHandle<bool>>) -> bool {
421    readers.into_iter().fold(true, |complete, reader| {
422        complete & reader.join().unwrap_or(false)
423    })
424}
425
426pub fn compose_worker_prompt(root: &Path) -> Result<String, String> {
427    let path = root.join(".agents/sloop/instructions.md");
428    match fs::read_to_string(&path) {
429        Ok(instructions) => Ok(format!("{WORKER_BOOTSTRAP_PROMPT}\n\n{instructions}")),
430        Err(error) if error.kind() == io::ErrorKind::NotFound => {
431            Ok(WORKER_BOOTSTRAP_PROMPT.to_owned())
432        }
433        Err(error) => Err(format!("cannot read {}: {error}", path.display())),
434    }
435}
436
437fn generate_worker_token() -> Result<String, String> {
438    let mut bytes = [0u8; 32];
439    let mut urandom = fs::File::open("/dev/urandom").map_err(|error| error.to_string())?;
440    urandom
441        .read_exact(&mut bytes)
442        .map_err(|error| error.to_string())?;
443    Ok(BASE64_TOKEN.encode(bytes))
444}
445
446pub fn run_output_path(state_dir: &Path, run_id: &str) -> PathBuf {
447    state_dir.join("runs").join(run_id).join("output.ndjson")
448}
449
450pub fn worker_socket_path(runtime_dir: &Path, run_id: &str) -> PathBuf {
451    // macOS caps Unix socket paths at 104 bytes, and the runtime root under
452    // `$TMPDIR` is long there, so worker sockets sit directly in the runtime
453    // directory and use the short run id. The full 32-hex id pushed a macOS
454    // path past the cap and every agent spawn failed at bind.
455    runtime_dir.join(format!("w{}.sock", crate::run_ref::short(run_id)))
456}
457
458pub fn process_start_time(pid: u32) -> Option<i64> {
459    #[cfg(target_os = "linux")]
460    {
461        let stat = fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
462        let after_command = &stat[stat.rfind(')')? + 1..];
463        after_command
464            .split_whitespace()
465            .nth(19)
466            .and_then(|field| field.parse().ok())
467    }
468
469    #[cfg(not(target_os = "linux"))]
470    {
471        let output = Command::new("ps")
472            .args(["-o", "lstart=", "-p", &pid.to_string()])
473            .output()
474            .ok()?;
475        if !output.status.success() || output.stdout.iter().all(u8::is_ascii_whitespace) {
476            return None;
477        }
478        let mut hash = 0xcbf29ce484222325_u64;
479        for byte in output.stdout {
480            hash ^= u64::from(byte);
481            hash = hash.wrapping_mul(0x100000001b3);
482        }
483        Some(hash as i64)
484    }
485}
486
487pub fn process_identity_matches(pid: u32, expected_start_time: Option<i64>) -> bool {
488    matches!(
489        (expected_start_time, process_start_time(pid)),
490        (Some(expected), Some(actual)) if expected == actual
491    )
492}
493
494pub fn kill_straggler_process_group(group: u32) -> bool {
495    let group = -(group as libc::pid_t);
496    let stragglers_present = unsafe { libc::kill(group, 0) } == 0;
497    if stragglers_present {
498        unsafe {
499            libc::kill(group, libc::SIGKILL);
500        }
501    }
502    stragglers_present
503}
504
505fn kill_process_group(process: ProcessIdentity) {
506    unsafe {
507        libc::kill(-(process.process_group_id as libc::pid_t), libc::SIGKILL);
508    }
509}
510
511fn kill_process_group_if_matches(process: ProcessIdentity) {
512    if process_identity_matches(process.pid, process.start_time) {
513        kill_process_group(process);
514    }
515}
516
517#[cfg(debug_assertions)]
518pub fn wait_for_test_hook(name: &str) {
519    use std::time::Duration;
520
521    let Some(directory) = std::env::var_os("SLOOP_TEST_HOOK_DIR").map(PathBuf::from) else {
522        return;
523    };
524    let armed = directory.join(format!("{name}.armed"));
525    if !armed.is_file() {
526        return;
527    }
528    let reached = directory.join(format!("{name}.reached"));
529    let release = directory.join(format!("{name}.release"));
530    if fs::write(&reached, b"").is_err() {
531        return;
532    }
533    while !release.is_file() {
534        std::thread::sleep(Duration::from_millis(10));
535    }
536}
537
538#[cfg(not(debug_assertions))]
539pub fn wait_for_test_hook(_name: &str) {}
540
541#[cfg(test)]
542mod tests {
543    use std::convert::Infallible;
544    use std::fs;
545
546    use tempfile::tempdir;
547
548    use super::{WORKER_BOOTSTRAP_PROMPT, compose_worker_prompt, run_exec_stage};
549    use crate::clock::SystemClock;
550    use crate::config::{AgentTarget, expand_agent_cmd};
551    use crate::runner::{
552        AgentProcessCheckpoint, ExecLaunch, ExecProcessCheckpoint, StageExecution, StageHooks,
553        StageOrder,
554    };
555
556    struct NoopHooks;
557
558    impl StageHooks for NoopHooks {
559        type Error = Infallible;
560
561        fn cancellation_requested(&self, _run_id: &str) -> bool {
562            false
563        }
564
565        fn record_agent_process(
566            &self,
567            _checkpoint: &AgentProcessCheckpoint,
568        ) -> Result<(), Self::Error> {
569            Ok(())
570        }
571
572        fn record_exec_process(
573            &self,
574            _checkpoint: &ExecProcessCheckpoint,
575        ) -> Result<(), Self::Error> {
576            Ok(())
577        }
578    }
579
580    fn target(cmd: &[&str], model: Option<&str>, effort: Option<&str>) -> AgentTarget {
581        AgentTarget {
582            cmd: cmd.iter().map(|argument| (*argument).to_owned()).collect(),
583            model: model.map(str::to_owned),
584            effort: effort.map(str::to_owned),
585        }
586    }
587
588    #[test]
589    fn agent_command_expands_ticket_model_and_effort() {
590        let template = target(
591            &[
592                "agent",
593                "--model={model}",
594                "--effort",
595                "{effort}",
596                "prompt={prompt}",
597            ],
598            None,
599            None,
600        );
601
602        assert_eq!(
603            expand_agent_cmd(&template, Some("sonnet"), Some("medium"), "assignment").unwrap(),
604            [
605                "agent",
606                "--model=sonnet",
607                "--effort",
608                "medium",
609                "prompt=assignment"
610            ]
611        );
612    }
613
614    #[test]
615    fn agent_command_rejects_a_missing_ticket_field() {
616        let template = target(&["agent", "{model}"], None, Some("medium"));
617
618        assert_eq!(
619            expand_agent_cmd(&template, None, Some("medium"), "assignment"),
620            Err("does not specify `model`".to_owned())
621        );
622    }
623
624    #[test]
625    fn agent_command_falls_back_to_target_defaults() {
626        let template = target(
627            &["agent", "{model}", "{effort}", "{prompt}"],
628            Some("opus"),
629            Some("high"),
630        );
631
632        assert_eq!(
633            expand_agent_cmd(&template, None, None, "assignment").unwrap(),
634            ["agent", "opus", "high", "assignment"]
635        );
636    }
637
638    #[test]
639    fn agent_command_prefers_ticket_values_over_target_defaults() {
640        let template = target(
641            &["agent", "{model}", "{effort}", "{prompt}"],
642            Some("opus"),
643            Some("high"),
644        );
645
646        assert_eq!(
647            expand_agent_cmd(&template, Some("haiku"), Some("low"), "assignment").unwrap(),
648            ["agent", "haiku", "low", "assignment"]
649        );
650    }
651
652    #[test]
653    fn scripted_exec_returns_factual_evidence_without_a_daemon_or_store() {
654        let directory = tempdir().unwrap();
655        let order = StageOrder {
656            run_id: "R1".into(),
657            stage: "check".into(),
658            execution: StageExecution::Exec(ExecLaunch {
659                argv: vec!["sh".into(), "-c".into(), "printf runner-output".into()],
660                worker: None,
661                environment: Vec::new(),
662            }),
663            worktree: directory.path().into(),
664            branch: "sloop/T1-a1-R1".into(),
665            output_path: directory.path().join("output.ndjson"),
666        };
667
668        let evidence = run_exec_stage(&order, &NoopHooks, &SystemClock).unwrap();
669
670        assert_eq!(evidence.exit_code, Some(0));
671        assert!(evidence.output_capture_complete);
672        assert!(evidence.process.is_some());
673    }
674
675    #[test]
676    fn worker_prompt_uses_the_builtin_when_instructions_are_absent() {
677        let root = tempdir().unwrap();
678
679        assert_eq!(
680            compose_worker_prompt(root.path()).unwrap(),
681            WORKER_BOOTSTRAP_PROMPT
682        );
683    }
684
685    #[test]
686    fn worker_prompt_appends_repository_instructions() {
687        let root = tempdir().unwrap();
688        fs::create_dir_all(root.path().join(".agents/sloop")).unwrap();
689        fs::write(
690            root.path().join(".agents/sloop/instructions.md"),
691            "Use repository conventions.\n",
692        )
693        .unwrap();
694
695        assert_eq!(
696            compose_worker_prompt(root.path()).unwrap(),
697            format!("{WORKER_BOOTSTRAP_PROMPT}\n\nUse repository conventions.\n")
698        );
699    }
700
701    #[test]
702    fn worker_socket_paths_fit_the_macos_socket_length_cap() {
703        // A representative macOS runtime directory: `$TMPDIR` on macOS is
704        // `/var/folders/<2>/<30>/T/`, followed by the sloop runtime key. The
705        // tightest real layout adds a `tempdir` test prefix on top.
706        let runtime_dir = std::path::Path::new(
707            "/var/folders/24/8k48jl6d249_n_qfxwsl6xvm0000gn/T/.tmpAbC123/sloop/0123456789abcdef",
708        );
709        let socket = super::worker_socket_path(runtime_dir, &"ab".repeat(16));
710
711        assert!(
712            socket.as_os_str().len() <= 104,
713            "worker socket path is {} bytes, over the 104-byte macOS cap: {}",
714            socket.as_os_str().len(),
715            socket.display()
716        );
717    }
718}
sloop/runner/local.rs

sloop/runner/
local.rs
