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
21pub 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 runtime_dir.join("workers").join(format!("{run_id}.sock"))
452}
453
454pub fn process_start_time(pid: u32) -> Option<i64> {
455 #[cfg(target_os = "linux")]
456 {
457 let stat = fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
458 let after_command = &stat[stat.rfind(')')? + 1..];
459 after_command
460 .split_whitespace()
461 .nth(19)
462 .and_then(|field| field.parse().ok())
463 }
464
465 #[cfg(not(target_os = "linux"))]
466 {
467 let output = Command::new("ps")
468 .args(["-o", "lstart=", "-p", &pid.to_string()])
469 .output()
470 .ok()?;
471 if !output.status.success() || output.stdout.iter().all(u8::is_ascii_whitespace) {
472 return None;
473 }
474 let mut hash = 0xcbf29ce484222325_u64;
475 for byte in output.stdout {
476 hash ^= u64::from(byte);
477 hash = hash.wrapping_mul(0x100000001b3);
478 }
479 Some(hash as i64)
480 }
481}
482
483pub fn process_identity_matches(pid: u32, expected_start_time: Option<i64>) -> bool {
484 matches!(
485 (expected_start_time, process_start_time(pid)),
486 (Some(expected), Some(actual)) if expected == actual
487 )
488}
489
490pub fn kill_straggler_process_group(group: u32) -> bool {
491 let group = -(group as libc::pid_t);
492 let stragglers_present = unsafe { libc::kill(group, 0) } == 0;
493 if stragglers_present {
494 unsafe {
495 libc::kill(group, libc::SIGKILL);
496 }
497 }
498 stragglers_present
499}
500
501fn kill_process_group(process: ProcessIdentity) {
502 unsafe {
503 libc::kill(-(process.process_group_id as libc::pid_t), libc::SIGKILL);
504 }
505}
506
507fn kill_process_group_if_matches(process: ProcessIdentity) {
508 if process_identity_matches(process.pid, process.start_time) {
509 kill_process_group(process);
510 }
511}
512
513#[cfg(debug_assertions)]
514pub fn wait_for_test_hook(name: &str) {
515 use std::time::Duration;
516
517 let Some(directory) = std::env::var_os("SLOOP_TEST_HOOK_DIR").map(PathBuf::from) else {
518 return;
519 };
520 let armed = directory.join(format!("{name}.armed"));
521 if !armed.is_file() {
522 return;
523 }
524 let reached = directory.join(format!("{name}.reached"));
525 let release = directory.join(format!("{name}.release"));
526 if fs::write(&reached, b"").is_err() {
527 return;
528 }
529 while !release.is_file() {
530 std::thread::sleep(Duration::from_millis(10));
531 }
532}
533
534#[cfg(not(debug_assertions))]
535pub fn wait_for_test_hook(_name: &str) {}
536
537#[cfg(test)]
538mod tests {
539 use std::convert::Infallible;
540 use std::fs;
541
542 use tempfile::tempdir;
543
544 use super::{WORKER_BOOTSTRAP_PROMPT, compose_worker_prompt, run_exec_stage};
545 use crate::clock::SystemClock;
546 use crate::config::{AgentTarget, expand_agent_cmd};
547 use crate::runner::{
548 AgentProcessCheckpoint, ExecLaunch, ExecProcessCheckpoint, StageExecution, StageHooks,
549 StageOrder,
550 };
551
552 struct NoopHooks;
553
554 impl StageHooks for NoopHooks {
555 type Error = Infallible;
556
557 fn cancellation_requested(&self, _run_id: &str) -> bool {
558 false
559 }
560
561 fn record_agent_process(
562 &self,
563 _checkpoint: &AgentProcessCheckpoint,
564 ) -> Result<(), Self::Error> {
565 Ok(())
566 }
567
568 fn record_exec_process(
569 &self,
570 _checkpoint: &ExecProcessCheckpoint,
571 ) -> Result<(), Self::Error> {
572 Ok(())
573 }
574 }
575
576 fn target(cmd: &[&str], model: Option<&str>, effort: Option<&str>) -> AgentTarget {
577 AgentTarget {
578 cmd: cmd.iter().map(|argument| (*argument).to_owned()).collect(),
579 model: model.map(str::to_owned),
580 effort: effort.map(str::to_owned),
581 }
582 }
583
584 #[test]
585 fn agent_command_expands_ticket_model_and_effort() {
586 let template = target(
587 &[
588 "agent",
589 "--model={model}",
590 "--effort",
591 "{effort}",
592 "prompt={prompt}",
593 ],
594 None,
595 None,
596 );
597
598 assert_eq!(
599 expand_agent_cmd(&template, Some("sonnet"), Some("medium"), "assignment").unwrap(),
600 [
601 "agent",
602 "--model=sonnet",
603 "--effort",
604 "medium",
605 "prompt=assignment"
606 ]
607 );
608 }
609
610 #[test]
611 fn agent_command_rejects_a_missing_ticket_field() {
612 let template = target(&["agent", "{model}"], None, Some("medium"));
613
614 assert_eq!(
615 expand_agent_cmd(&template, None, Some("medium"), "assignment"),
616 Err("does not specify `model`".to_owned())
617 );
618 }
619
620 #[test]
621 fn agent_command_falls_back_to_target_defaults() {
622 let template = target(
623 &["agent", "{model}", "{effort}", "{prompt}"],
624 Some("opus"),
625 Some("high"),
626 );
627
628 assert_eq!(
629 expand_agent_cmd(&template, None, None, "assignment").unwrap(),
630 ["agent", "opus", "high", "assignment"]
631 );
632 }
633
634 #[test]
635 fn agent_command_prefers_ticket_values_over_target_defaults() {
636 let template = target(
637 &["agent", "{model}", "{effort}", "{prompt}"],
638 Some("opus"),
639 Some("high"),
640 );
641
642 assert_eq!(
643 expand_agent_cmd(&template, Some("haiku"), Some("low"), "assignment").unwrap(),
644 ["agent", "haiku", "low", "assignment"]
645 );
646 }
647
648 #[test]
649 fn scripted_exec_returns_factual_evidence_without_a_daemon_or_store() {
650 let directory = tempdir().unwrap();
651 let order = StageOrder {
652 run_id: "R1".into(),
653 stage: "check".into(),
654 execution: StageExecution::Exec(ExecLaunch {
655 argv: vec!["sh".into(), "-c".into(), "printf runner-output".into()],
656 worker: None,
657 environment: Vec::new(),
658 }),
659 worktree: directory.path().into(),
660 branch: "sloop/T1-a1-R1".into(),
661 output_path: directory.path().join("output.ndjson"),
662 };
663
664 let evidence = run_exec_stage(&order, &NoopHooks, &SystemClock).unwrap();
665
666 assert_eq!(evidence.exit_code, Some(0));
667 assert!(evidence.output_capture_complete);
668 assert!(evidence.process.is_some());
669 }
670
671 #[test]
672 fn worker_prompt_uses_the_builtin_when_instructions_are_absent() {
673 let root = tempdir().unwrap();
674
675 assert_eq!(
676 compose_worker_prompt(root.path()).unwrap(),
677 WORKER_BOOTSTRAP_PROMPT
678 );
679 }
680
681 #[test]
682 fn worker_prompt_appends_repository_instructions() {
683 let root = tempdir().unwrap();
684 fs::create_dir_all(root.path().join(".agents/sloop")).unwrap();
685 fs::write(
686 root.path().join(".agents/sloop/instructions.md"),
687 "Use repository conventions.\n",
688 )
689 .unwrap();
690
691 assert_eq!(
692 compose_worker_prompt(root.path()).unwrap(),
693 format!("{WORKER_BOOTSTRAP_PROMPT}\n\nUse repository conventions.\n")
694 );
695 }
696}