use super::store::RoutineStore;
use super::{Routine, RoutineError, RunCause, RunRecord, RunStatus, MAX_RUNS};
use serde_json::Value;
use std::fs::{self, File};
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
use std::os::fd::AsRawFd;
use std::os::unix::fs::PermissionsExt;
use std::os::unix::process::CommandExt;
use std::process::{Command, Stdio};
use std::thread;
use std::time::{SystemTime, UNIX_EPOCH};
const FINAL_OUTPUT_BYTES: usize = 16 * 1024;
const STRUCTURED_LINE_BYTES: usize = 64 * 1024;
const MAX_RUN_LOG_BYTES: usize = 8 * 1024 * 1024;
const RUN_LOG_TAIL_BYTES: usize = 128 * 1024;
const TRUNCATION_MARKER: &[u8] = b"\n[asched: middle output truncated]\n";
pub fn expanded_argv(routine: &Routine) -> (Vec<String>, Option<&[u8]>) {
if routine.command.iter().any(|arg| arg == "{prompt}") {
(
routine
.command
.iter()
.map(|arg| {
if arg == "{prompt}" {
routine.prompt.clone()
} else {
arg.clone()
}
})
.collect(),
None,
)
} else {
(routine.command.clone(), Some(routine.prompt.as_bytes()))
}
}
pub fn execute(
store: &RoutineStore,
routine: &Routine,
scheduled: Option<i64>,
) -> Result<RunRecord, RoutineError> {
execute_supervised(store, routine, scheduled, |_| {})
}
pub fn execute_supervised(
store: &RoutineStore,
routine: &Routine,
scheduled: Option<i64>,
on_started: impl FnOnce(&RunRecord),
) -> Result<RunRecord, RoutineError> {
if routine.command.is_empty() {
return Err(RoutineError::Validation(
"command must contain at least one argv item".into(),
));
}
let cause = scheduled.map_or(RunCause::Manual, |scheduled_epoch_minute| RunCause::Cron {
scheduled_epoch_minute,
});
let record = prepare_run_record(store, routine, cause);
append_record(store, record.clone())?;
execute_prepared_supervised(store, routine, record, None, on_started)
}
pub(crate) fn prepare_run_record(
store: &RoutineStore,
routine: &Routine,
cause: RunCause,
) -> RunRecord {
let started = now_epoch();
let id = format!("{}-{}", started, now_nanos());
let run_dir = store.logs_dir().join(hex_name(&routine.name)).join(&id);
let scheduled_epoch_minute = match cause {
RunCause::Cron {
scheduled_epoch_minute,
} => Some(scheduled_epoch_minute),
_ => None,
};
RunRecord {
id,
routine: routine.name.clone(),
started_epoch: started,
finished_epoch: None,
cause,
scheduled_epoch_minute,
status: RunStatus::Running,
exit_code: None,
pid: None,
process_start: None,
final_output: String::new(),
stdout_path: run_dir.join("stdout.log"),
stderr_path: run_dir.join("stderr.log"),
}
}
pub(crate) fn execute_prepared_supervised(
store: &RoutineStore,
routine: &Routine,
mut record: RunRecord,
event_payload: Option<String>,
on_started: impl FnOnce(&RunRecord),
) -> Result<RunRecord, RoutineError> {
if routine.command.is_empty() {
return Err(RoutineError::Validation(
"command argv must not be empty".into(),
));
}
let lock_dir = store.logs_dir().join("locks");
fs::create_dir_all(&lock_dir)?;
fs::set_permissions(&lock_dir, fs::Permissions::from_mode(0o700))?;
let run_lock = std::fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(lock_dir.join(hex_name(&routine.name)))?;
run_lock.set_permissions(fs::Permissions::from_mode(0o600))?;
if unsafe { libc::flock(run_lock.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } != 0 {
return Err(RoutineError::AlreadyRunning(routine.name.clone()));
}
let run_dir = record
.stdout_path
.parent()
.ok_or_else(|| RoutineError::Corrupt("run output path has no parent".into()))?;
fs::create_dir_all(run_dir)?;
fs::set_permissions(run_dir, fs::Permissions::from_mode(0o700))?;
let stdout_path = record.stdout_path.clone();
let stderr_path = record.stderr_path.clone();
let (argv, stdin_bytes) = expanded_argv(routine);
let mut command = Command::new(&argv[0]);
command.args(&argv[1..]).current_dir(store.project());
if let Some(payload) = event_payload {
command.env("ASCHED_EVENT_PAYLOAD", payload);
}
command
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
command.process_group(0);
let mut child = match command.spawn() {
Ok(child) => child,
Err(error) => {
let stdout_file = File::create(&stdout_path)?;
stdout_file.set_permissions(fs::Permissions::from_mode(0o600))?;
let mut stderr_file = File::create(&stderr_path)?;
stderr_file.set_permissions(fs::Permissions::from_mode(0o600))?;
stderr_file.write_all(error.to_string().as_bytes())?;
stderr_file.sync_all()?;
record.status = RunStatus::SpawnFailed;
record.finished_epoch = Some(now_epoch());
record.final_output = error.to_string();
replace_record(store, record.clone())?;
return Ok(record);
}
};
record.pid = Some(child.id() as i32);
record.process_start = process_start(child.id() as i32);
if let Err(error) = replace_record(store, record.clone()) {
return fail_started_run(
store,
child,
record,
&format!("failed to persist spawned process: {error}"),
None,
None,
);
}
on_started(&record);
let stdout = match child.stdout.take() {
Some(stdout) => stdout,
None => return fail_started_run(store, child, record, "stdout pipe missing", None, None),
};
let stderr = match child.stderr.take() {
Some(stderr) => stderr,
None => return fail_started_run(store, child, record, "stderr pipe missing", None, None),
};
let out_thread = drain(stdout, stdout_path.clone());
let err_thread = drain(stderr, stderr_path.clone());
if let Some(mut stdin) = child.stdin.take() {
if let Some(bytes) = stdin_bytes {
if let Err(error) = stdin.write_all(bytes) {
return fail_started_run(
store,
child,
record,
&format!("prompt delivery failed: {error}"),
Some(out_thread),
Some(err_thread),
);
}
}
}
let pid = child.id() as i32;
let status = child.wait()?;
terminate_remaining_process_group(pid);
out_thread
.join()
.map_err(|_| RoutineError::Io("stdout reader panicked".into()))??;
err_thread
.join()
.map_err(|_| RoutineError::Io("stderr reader panicked".into()))??;
record.finished_epoch = Some(now_epoch());
record.exit_code = status.code();
let cancelled = store
.load_runtime()
.ok()
.and_then(|state| {
state
.runs
.get(&routine.name)
.and_then(|runs| runs.iter().find(|run| run.id == record.id))
.cloned()
})
.is_some_and(|run| run.status == RunStatus::Cancelled);
record.status = if cancelled {
RunStatus::Cancelled
} else if status.success() {
RunStatus::Succeeded
} else {
RunStatus::Failed
};
record.pid = None;
record.process_start = None;
record.final_output = extract_final_output_files(&stdout_path, &stderr_path);
replace_record(store, record.clone())?;
Ok(record)
}
fn fail_started_run(
store: &RoutineStore,
mut child: std::process::Child,
mut record: RunRecord,
message: &str,
out_thread: Option<thread::JoinHandle<Result<(), RoutineError>>>,
err_thread: Option<thread::JoinHandle<Result<(), RoutineError>>>,
) -> Result<RunRecord, RoutineError> {
let pid = child.id() as i32;
unsafe {
libc::kill(-pid, libc::SIGTERM);
}
let mut status = None;
for _ in 0..20 {
if status.is_none() {
match child.try_wait() {
Ok(Some(value)) => status = Some(value),
Ok(None) => {}
Err(_) => break,
}
}
if !process_group_exists(pid) {
break;
}
thread::sleep(std::time::Duration::from_millis(100));
}
if status.is_none() || process_group_exists(pid) {
unsafe {
libc::kill(-pid, libc::SIGKILL);
}
if status.is_none() {
status = child.wait().ok();
}
}
if let Some(handle) = out_thread {
let _ = handle.join();
}
if let Some(handle) = err_thread {
let _ = handle.join();
}
record.finished_epoch = Some(now_epoch());
record.status = RunStatus::Failed;
record.exit_code = status.and_then(|status| status.code());
record.pid = None;
record.process_start = None;
record.final_output = message.to_string();
replace_record(store, record)?;
Err(RoutineError::Io(message.to_string()))
}
fn process_group_exists(pgid: i32) -> bool {
if unsafe { libc::kill(-pgid, 0) } == 0 {
return true;
}
std::io::Error::last_os_error().raw_os_error() != Some(libc::ESRCH)
}
fn terminate_remaining_process_group(pgid: i32) {
if !process_group_exists(pgid) {
return;
}
unsafe {
libc::kill(-pgid, libc::SIGTERM);
}
for _ in 0..20 {
if !process_group_exists(pgid) {
return;
}
thread::sleep(std::time::Duration::from_millis(100));
}
for _ in 0..20 {
unsafe {
libc::kill(-pgid, libc::SIGKILL);
}
if !process_group_exists(pgid) {
return;
}
thread::sleep(std::time::Duration::from_millis(100));
}
}
pub(crate) fn process_start(pid: i32) -> Option<String> {
#[cfg(target_os = "macos")]
{
let mut info = std::mem::MaybeUninit::<libc::proc_bsdinfo>::zeroed();
let size = std::mem::size_of::<libc::proc_bsdinfo>() as i32;
let read = unsafe {
libc::proc_pidinfo(
pid,
libc::PROC_PIDTBSDINFO,
0,
info.as_mut_ptr().cast(),
size,
)
};
if read != size {
return None;
}
let info = unsafe { info.assume_init() };
Some(format!(
"{}:{}",
info.pbi_start_tvsec, info.pbi_start_tvusec
))
}
#[cfg(not(target_os = "macos"))]
{
let stat = fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
let after_name = stat.rsplit_once(") ")?.1;
after_name.split_whitespace().nth(19).map(str::to_string)
}
}
fn drain(
mut source: impl Read + Send + 'static,
path: std::path::PathBuf,
) -> thread::JoinHandle<Result<(), RoutineError>> {
thread::spawn(move || {
let mut file = File::create(path)?;
file.set_permissions(fs::Permissions::from_mode(0o600))?;
copy_bounded_log(&mut source, &mut file)?;
file.sync_all()?;
Ok(())
})
}
fn copy_bounded_log(mut source: impl Read, mut target: impl Write) -> Result<(), RoutineError> {
let prefix_limit = MAX_RUN_LOG_BYTES - RUN_LOG_TAIL_BYTES - TRUNCATION_MARKER.len();
let mut prefix_written = 0;
let mut tail = Vec::with_capacity(RUN_LOG_TAIL_BYTES);
let mut truncated = false;
let mut buffer = [0_u8; 16 * 1024];
loop {
let read = source.read(&mut buffer)?;
if read == 0 {
break;
}
let mut chunk = &buffer[..read];
if prefix_written < prefix_limit {
let keep = chunk.len().min(prefix_limit - prefix_written);
target.write_all(&chunk[..keep])?;
prefix_written += keep;
chunk = &chunk[keep..];
}
if chunk.is_empty() {
continue;
}
if tail.len() + chunk.len() > RUN_LOG_TAIL_BYTES {
truncated = true;
let discard = (tail.len() + chunk.len()) - RUN_LOG_TAIL_BYTES;
if discard >= tail.len() {
tail.clear();
let start = chunk.len().saturating_sub(RUN_LOG_TAIL_BYTES);
tail.extend_from_slice(&chunk[start..]);
} else {
tail.drain(..discard);
tail.extend_from_slice(chunk);
}
} else {
tail.extend_from_slice(chunk);
}
}
if truncated {
target.write_all(TRUNCATION_MARKER)?;
}
target.write_all(&tail)?;
Ok(())
}
fn append_record(store: &RoutineStore, record: RunRecord) -> Result<(), RoutineError> {
store.modify_runtime(|state| {
state
.runs
.entry(record.routine.clone())
.or_default()
.push(record);
Ok(())
})
}
fn replace_record(store: &RoutineStore, record: RunRecord) -> Result<(), RoutineError> {
let routine = record.routine.clone();
let removed = store.modify_runtime(|state| {
let runs = state.runs.entry(record.routine.clone()).or_default();
if let Some(existing) = runs.iter_mut().find(|run| run.id == record.id) {
*existing = record;
}
Ok(if runs.len() > MAX_RUNS {
runs.drain(0..runs.len() - MAX_RUNS).collect::<Vec<_>>()
} else {
vec![]
})
})?;
for old in removed {
prune_run_logs(store, &routine, &old);
}
Ok(())
}
pub(crate) fn prune_run_logs(store: &RoutineStore, routine: &str, run: &RunRecord) {
if run.routine != routine || !is_safe_run_id(&run.id) {
return;
}
let dir = store.logs_dir().join(hex_name(routine)).join(&run.id);
let _ = fs::remove_dir_all(dir);
}
fn is_safe_run_id(id: &str) -> bool {
let Some((seconds, nanos)) = id.split_once('-') else {
return false;
};
!seconds.is_empty()
&& !nanos.is_empty()
&& seconds.bytes().all(|byte| byte.is_ascii_digit())
&& nanos.bytes().all(|byte| byte.is_ascii_digit())
}
pub fn extract_final_output(stdout: &str, stderr: &str) -> String {
extract_final_output_bytes(stdout.as_bytes(), stderr.as_bytes())
}
fn extract_final_output_bytes(stdout: &[u8], stderr: &[u8]) -> String {
let stdout = String::from_utf8_lossy(stdout);
let stderr = String::from_utf8_lossy(stderr);
let mut codex = None;
let mut claude = None;
for line in stdout
.lines()
.filter(|line| line.len() <= STRUCTURED_LINE_BYTES)
{
let Ok(value) = serde_json::from_str::<Value>(line) else {
continue;
};
if value.get("type").and_then(Value::as_str) == Some("item.completed") {
let item = &value["item"];
if item.get("type").and_then(Value::as_str) == Some("agent_message") {
codex = item
.get("text")
.and_then(Value::as_str)
.map(|text| text_tail(text, FINAL_OUTPUT_BYTES));
}
}
if value.get("type").and_then(Value::as_str) == Some("result") {
claude = value
.get("result")
.and_then(Value::as_str)
.map(|text| text_tail(text, FINAL_OUTPUT_BYTES));
} else if value.get("type").and_then(Value::as_str) == Some("assistant") {
if let Some(text) = claude_assistant_text(&value) {
claude = Some(text_tail(&text, FINAL_OUTPUT_BYTES));
}
}
}
codex
.or(claude)
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(|| {
if !stdout.trim().is_empty() {
text_tail(stdout.trim(), FINAL_OUTPUT_BYTES)
} else {
text_tail(stderr.trim(), FINAL_OUTPUT_BYTES)
}
})
}
fn extract_final_output_files(
stdout_path: &std::path::Path,
stderr_path: &std::path::Path,
) -> String {
let (codex, claude) = File::open(stdout_path)
.ok()
.map(scan_structured_output)
.unwrap_or_default();
codex
.or(claude)
.filter(|output| !output.trim().is_empty())
.unwrap_or_else(|| {
let stdout = bounded_file_tail(stdout_path);
if stdout.is_empty() {
bounded_file_tail(stderr_path)
} else {
stdout
}
})
}
fn scan_structured_output(file: File) -> (Option<String>, Option<String>) {
let mut codex = None;
let mut claude = None;
let mut reader = BufReader::new(file);
loop {
let Some(line) = read_bounded_line(&mut reader, STRUCTURED_LINE_BYTES) else {
break;
};
let Some(line) = line else {
continue;
};
let Ok(value) = serde_json::from_slice::<Value>(&line) else {
continue;
};
if value.get("type").and_then(Value::as_str) == Some("item.completed") {
let item = &value["item"];
if item.get("type").and_then(Value::as_str) == Some("agent_message") {
codex = item
.get("text")
.and_then(Value::as_str)
.map(|text| text_tail(text, FINAL_OUTPUT_BYTES));
}
}
if value.get("type").and_then(Value::as_str) == Some("result") {
claude = value
.get("result")
.and_then(Value::as_str)
.map(|text| text_tail(text, FINAL_OUTPUT_BYTES));
} else if value.get("type").and_then(Value::as_str) == Some("assistant") {
if let Some(text) = claude_assistant_text(&value) {
claude = Some(text_tail(&text, FINAL_OUTPUT_BYTES));
}
}
}
(codex, claude)
}
fn read_bounded_line(reader: &mut impl BufRead, max_bytes: usize) -> Option<Option<Vec<u8>>> {
let mut line = Vec::new();
let mut oversized = false;
loop {
let buffer = reader.fill_buf().ok()?;
if buffer.is_empty() {
return (!line.is_empty() || oversized).then_some((!oversized).then_some(line));
}
let newline = buffer.iter().position(|byte| *byte == b'\n');
let consumed = newline.map_or(buffer.len(), |index| index + 1);
let content_len = newline.unwrap_or(buffer.len());
if !oversized && line.len().saturating_add(content_len) <= max_bytes {
line.extend_from_slice(&buffer[..content_len]);
} else {
oversized = true;
line.clear();
}
reader.consume(consumed);
if newline.is_some() {
return Some((!oversized).then_some(line));
}
}
}
fn bounded_file_tail(path: &std::path::Path) -> String {
const READ_CHUNK_BYTES: usize = 4 * 1024;
let Ok(mut file) = File::open(path) else {
return String::new();
};
let Ok(mut position) = file.seek(SeekFrom::End(0)) else {
return String::new();
};
let mut tail = Vec::with_capacity(FINAL_OUTPUT_BYTES);
let mut found_content = false;
while position > 0 && tail.len() < FINAL_OUTPUT_BYTES {
let chunk_len = usize::try_from(position.min(READ_CHUNK_BYTES as u64)).unwrap_or(0);
position -= chunk_len as u64;
if file.seek(SeekFrom::Start(position)).is_err() {
break;
}
let mut chunk = vec![0; chunk_len];
if file.read_exact(&mut chunk).is_err() {
break;
}
if !found_content {
while chunk.last().is_some_and(u8::is_ascii_whitespace) {
chunk.pop();
}
found_content = !chunk.is_empty();
}
if found_content {
let remaining = FINAL_OUTPUT_BYTES - tail.len();
let start = chunk.len().saturating_sub(remaining);
chunk.drain(..start);
chunk.extend_from_slice(&tail);
tail = chunk;
}
}
String::from_utf8_lossy(&tail).trim().to_string()
}
fn claude_assistant_text(value: &Value) -> Option<String> {
let content = value.get("message")?.get("content")?.as_array()?;
let text = content
.iter()
.filter(|block| block.get("type").and_then(Value::as_str) == Some("text"))
.filter_map(|block| block.get("text").and_then(Value::as_str))
.collect::<Vec<_>>()
.join("\n");
(!text.trim().is_empty()).then_some(text)
}
fn text_tail(text: &str, max_bytes: usize) -> String {
if text.len() <= max_bytes {
return text.to_string();
}
let mut start = text.len() - max_bytes;
while !text.is_char_boundary(start) {
start += 1;
}
text[start..].to_string()
}
fn hex_name(name: &str) -> String {
name.as_bytes().iter().map(|b| format!("{b:02x}")).collect()
}
fn now_epoch() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs() as i64
}
fn now_nanos() -> u128 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::routine::Trigger;
use std::cell::Cell;
use std::io::Cursor;
use std::path::PathBuf;
use std::rc::Rc;
use std::sync::atomic::{AtomicI32, AtomicU64, Ordering};
static NEXT: AtomicU64 = AtomicU64::new(0);
struct CountingReader {
source: Cursor<Vec<u8>>,
consumed: Rc<Cell<usize>>,
}
impl Read for CountingReader {
fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
let read = self.source.read(buffer)?;
self.consumed.set(self.consumed.get() + read);
Ok(read)
}
}
#[test]
fn given_log_within_cap_when_copied_then_every_byte_is_preserved_without_marker() {
let input = b"begin\ncomplete tail".to_vec();
let consumed = Rc::new(Cell::new(0));
let reader = CountingReader {
source: Cursor::new(input.clone()),
consumed: Rc::clone(&consumed),
};
let mut output = Vec::new();
copy_bounded_log(reader, &mut output).unwrap();
assert_eq!(
(
output == input,
output
.windows(TRUNCATION_MARKER.len())
.any(|window| window == TRUNCATION_MARKER),
consumed.get(),
),
(true, false, input.len())
);
}
#[test]
fn given_log_over_cap_when_copied_then_prefix_and_tail_are_bounded_and_input_is_consumed() {
let mut input = b"BEGIN".to_vec();
input.resize(MAX_RUN_LOG_BYTES + 4_096, b'x');
input.extend_from_slice(b"FINAL-TAIL");
let consumed = Rc::new(Cell::new(0));
let reader = CountingReader {
source: Cursor::new(input.clone()),
consumed: Rc::clone(&consumed),
};
let mut output = Vec::new();
copy_bounded_log(reader, &mut output).unwrap();
let markers = output
.windows(TRUNCATION_MARKER.len())
.filter(|window| *window == TRUNCATION_MARKER)
.count();
assert_eq!(
(
output.len() <= MAX_RUN_LOG_BYTES,
output.starts_with(b"BEGIN"),
output.ends_with(b"FINAL-TAIL"),
markers,
consumed.get(),
),
(true, true, true, 1, input.len())
);
}
#[test]
fn exact_prompt_is_replaced_otherwise_stdin_is_used() {
let replaced = Routine {
name: "x".into(),
trigger: Trigger::Cron("* * * * *".into()),
command: vec!["echo".into(), "{prompt}".into()],
prompt: "hi".into(),
enabled: true,
};
assert_eq!(expanded_argv(&replaced).0, vec!["echo", "hi"]);
assert!(expanded_argv(&replaced).1.is_none());
let stdin = Routine {
command: vec!["cat".into()],
..replaced
};
assert_eq!(expanded_argv(&stdin).1, Some("hi".as_bytes()));
}
#[test]
fn extracts_codex_claude_and_fallback_output() {
assert_eq!(extract_final_output("{\"type\":\"item.completed\",\"item\":{\"type\":\"agent_message\",\"text\":\"done\"}}\n", ""), "done");
assert_eq!(
extract_final_output("{\"type\":\"result\",\"result\":\"answer\"}\n", ""),
"answer"
);
assert_eq!(extract_final_output("plain\n", "err"), "plain");
assert_eq!(extract_final_output("", "err\n"), "err");
}
#[test]
fn plain_fallback_is_bounded_to_the_log_tail() {
let output = format!("HEAD\n{}TAIL", "progress\n".repeat(4_000));
let extracted = extract_final_output(&output, "");
assert!(extracted.len() <= 16 * 1024);
assert!(extracted.ends_with("TAIL"));
assert!(!extracted.contains("HEAD"));
}
#[test]
fn claude_extraction_ignores_nested_tool_payload_text() {
let output = concat!(
"{\"type\":\"assistant\",\"message\":{\"content\":[",
"{\"type\":\"text\",\"text\":\"final answer\"},",
"{\"type\":\"tool_use\",\"input\":{\"text\":\"tool payload\"}}]}}\n"
);
assert_eq!(extract_final_output(output, ""), "final answer");
}
#[test]
fn tool_only_structured_output_uses_bounded_plain_fallback() {
let output = concat!(
"{\"type\":\"assistant\",\"message\":{\"content\":[",
"{\"type\":\"tool_use\",\"input\":{\"text\":\"not an answer\"}}]}}\n"
);
assert_eq!(extract_final_output(output, ""), output.trim());
}
#[test]
fn invalid_utf8_is_preserved_in_plain_fallback() {
assert_eq!(
extract_final_output_bytes(b"before\xffafter\n", b""),
"before\u{fffd}after"
);
}
#[test]
fn file_extraction_streams_structured_output_and_bounds_plain_tail() {
let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../../target/routine-execution-tests")
.join(format!(
"{}-{}",
std::process::id(),
NEXT.fetch_add(1, Ordering::Relaxed)
));
fs::create_dir_all(&root).unwrap();
let stdout_path = root.join("stdout.log");
let stderr_path = root.join("stderr.log");
let mut stdout = File::create(&stdout_path).unwrap();
for _ in 0..10_000 {
writeln!(stdout, "progress output that must not be retained").unwrap();
}
writeln!(
stdout,
"{{\"type\":\"result\",\"result\":\"final answer\"}}"
)
.unwrap();
fs::write(&stderr_path, "error").unwrap();
assert_eq!(
extract_final_output_files(&stdout_path, &stderr_path),
"final answer"
);
fs::write(
&stdout_path,
format!("HEAD\n{}TAIL\n", "progress\n".repeat(4_000)),
)
.unwrap();
let fallback = extract_final_output_files(&stdout_path, &stderr_path);
assert!(fallback.len() <= 16 * 1024);
assert!(fallback.ends_with("TAIL"));
assert!(!fallback.contains("HEAD"));
let _ = fs::remove_dir_all(root);
}
#[test]
fn oversized_structured_line_is_discarded_without_hiding_later_result() {
let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../../target/routine-execution-tests")
.join(format!(
"{}-{}",
std::process::id(),
NEXT.fetch_add(1, Ordering::Relaxed)
));
fs::create_dir_all(&root).unwrap();
let stdout_path = root.join("stdout.log");
let stderr_path = root.join("stderr.log");
let mut stdout = File::create(&stdout_path).unwrap();
writeln!(
stdout,
"{{\"type\":\"result\",\"result\":\"{}\"}}",
"x".repeat(STRUCTURED_LINE_BYTES)
)
.unwrap();
writeln!(stdout, "{{\"type\":\"result\",\"result\":\"bounded\"}}").unwrap();
fs::write(&stderr_path, "").unwrap();
assert_eq!(
extract_final_output_files(&stdout_path, &stderr_path),
"bounded"
);
let _ = fs::remove_dir_all(root);
}
#[test]
fn empty_public_command_returns_validation_instead_of_panicking() {
let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../../target/routine-execution-tests")
.join(format!(
"{}-{}",
std::process::id(),
NEXT.fetch_add(1, Ordering::Relaxed)
));
let project =
fs::canonicalize(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../..")).unwrap();
let store = RoutineStore::new(root.clone(), &project).unwrap();
let routine = Routine {
name: "empty".into(),
trigger: Trigger::Cron("* * * * *".into()),
command: Vec::new(),
prompt: String::new(),
enabled: true,
};
assert!(matches!(
execute(&store, &routine, None),
Err(RoutineError::Validation(_))
));
assert!(!store.load_runtime().unwrap().runs.contains_key("empty"));
let _ = fs::remove_dir_all(root);
}
#[test]
fn successful_leader_exit_does_not_leave_pipe_holding_descendants() {
let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../../target/routine-execution-tests")
.join(format!(
"{}-{}",
std::process::id(),
NEXT.fetch_add(1, Ordering::Relaxed)
));
let project =
fs::canonicalize(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../..")).unwrap();
let store = RoutineStore::new(root.clone(), &project).unwrap();
let routine = Routine {
name: "descendant".into(),
trigger: Trigger::Cron("* * * * *".into()),
command: vec![
"/bin/sh".into(),
"-c".into(),
"(trap '' TERM; sleep 30) & exit 0".into(),
],
prompt: String::new(),
enabled: true,
};
let pid = AtomicI32::new(0);
let record = execute_supervised(&store, &routine, None, |record| {
pid.store(record.pid.unwrap(), Ordering::SeqCst);
})
.unwrap();
assert_eq!(record.status, RunStatus::Succeeded);
assert_eq!(unsafe { libc::kill(-pid.load(Ordering::SeqCst), 0) }, -1);
let _ = fs::remove_dir_all(root);
}
#[test]
fn retention_never_deletes_a_directory_from_persisted_output_paths() {
let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../../target/routine-execution-tests")
.join(format!(
"{}-{}",
std::process::id(),
NEXT.fetch_add(1, Ordering::Relaxed)
));
let project =
fs::canonicalize(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../..")).unwrap();
let store = RoutineStore::new(root.clone(), &project).unwrap();
let victim = root.join("must-survive");
fs::create_dir_all(&victim).unwrap();
fs::write(victim.join("marker"), "present").unwrap();
let record = RunRecord {
id: "../must-survive".into(),
routine: "safe".into(),
started_epoch: 1,
finished_epoch: Some(2),
scheduled_epoch_minute: None,
cause: RunCause::Manual,
status: RunStatus::Succeeded,
exit_code: Some(0),
pid: None,
process_start: None,
final_output: String::new(),
stdout_path: victim.join("stdout.log"),
stderr_path: victim.join("stderr.log"),
};
prune_run_logs(&store, "safe", &record);
assert!(victim.join("marker").exists());
let _ = fs::remove_dir_all(root);
}
#[test]
fn prompt_write_failure_terminates_and_reaps_process_group() {
let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../../target/routine-execution-tests")
.join(format!(
"{}-{}",
std::process::id(),
NEXT.fetch_add(1, Ordering::Relaxed)
));
let project =
fs::canonicalize(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../..")).unwrap();
let store = RoutineStore::new(root.clone(), &project).unwrap();
let routine = Routine {
name: "broken-stdin".into(),
trigger: Trigger::Cron("* * * * *".into()),
command: vec![
"/bin/sh".into(),
"-c".into(),
"exec 0<&-; trap '' TERM; sleep 30".into(),
],
prompt: "x".repeat(1024 * 1024),
enabled: true,
};
let pid = AtomicI32::new(0);
let result = execute_supervised(&store, &routine, None, |record| {
pid.store(record.pid.unwrap(), Ordering::SeqCst);
});
assert!(matches!(result, Err(RoutineError::Io(_))));
let pid = pid.load(Ordering::SeqCst);
assert!(pid > 0);
assert_eq!(unsafe { libc::kill(-pid, 0) }, -1);
let record = store.load_runtime().unwrap().runs["broken-stdin"]
.last()
.unwrap()
.clone();
assert_eq!(record.status, RunStatus::Failed);
assert!(record.pid.is_none());
let _ = fs::remove_dir_all(root);
}
}