use crate::agent_result::{IdleTimeoutCommit, IdleTimeoutRecord};
use crate::git::hermetic_command;
use crate::state::State;
use std::io::{BufRead, BufReader, Write};
use std::os::unix::process::CommandExt;
use std::path::Path;
use std::process::Stdio;
use std::sync::mpsc;
use std::time::Duration;
use tracing::{debug, info, warn};
#[derive(Debug, thiserror::Error)]
pub enum MonitorError {
#[error("failed to spawn monitor: {0}")]
Io(#[from] std::io::Error),
#[error("project path is not valid UTF-8")]
NonUtf8Path,
#[error("could not determine devflow binary path")]
NoBinaryPath,
#[error("supervised child exposed no {0} pipe")]
NoChildPipe(&'static str),
}
pub const DEFAULT_IDLE_TIMEOUT_SECS: u64 = 120;
pub const IDLE_TIMEOUT_FLOOR_SECS: u64 = 120;
pub const IDLE_TIMEOUT_ENV: &str = "DEVFLOW_CLAUDE_IDLE_TIMEOUT_SECS";
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IdleTimeoutResolution {
Default,
Configured,
Clamped {
configured: u64,
},
Unparseable {
raw: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IdleTimeoutSetting {
pub timeout: Duration,
pub resolution: IdleTimeoutResolution,
}
impl IdleTimeoutSetting {
pub fn clamped(&self) -> bool {
matches!(self.resolution, IdleTimeoutResolution::Clamped { .. })
}
pub fn notice(&self) -> Option<String> {
match &self.resolution {
IdleTimeoutResolution::Default | IdleTimeoutResolution::Configured => None,
IdleTimeoutResolution::Clamped { configured } => Some(format!(
"{IDLE_TIMEOUT_ENV}={configured} is below the {IDLE_TIMEOUT_FLOOR_SECS}s floor \
and was CLAMPED; {}s is in force. A shorter window kills healthy runs: a 12s \
bound terminated a live, healthy run in 2 of 7 measured trials.",
self.timeout.as_secs()
)),
IdleTimeoutResolution::Unparseable { raw } => Some(format!(
"{IDLE_TIMEOUT_ENV}={raw:?} could not be parsed as a whole number of seconds; \
the {}s default is in force. If you meant to RAISE the timeout, this did not \
do it.",
self.timeout.as_secs()
)),
}
}
}
pub fn parse_idle_timeout_secs(raw: Option<String>) -> IdleTimeoutSetting {
let floor = Duration::from_secs(IDLE_TIMEOUT_FLOOR_SECS);
let Some(trimmed) = raw.as_deref().map(str::trim).filter(|s| !s.is_empty()) else {
return IdleTimeoutSetting {
timeout: floor,
resolution: IdleTimeoutResolution::Default,
};
};
let Ok(configured) = trimmed.parse::<u64>() else {
return IdleTimeoutSetting {
timeout: floor,
resolution: IdleTimeoutResolution::Unparseable {
raw: trimmed.to_string(),
},
};
};
if configured < IDLE_TIMEOUT_FLOOR_SECS {
IdleTimeoutSetting {
timeout: floor,
resolution: IdleTimeoutResolution::Clamped { configured },
}
} else {
IdleTimeoutSetting {
timeout: Duration::from_secs(configured),
resolution: IdleTimeoutResolution::Configured,
}
}
}
pub fn idle_timeout_setting() -> IdleTimeoutSetting {
parse_idle_timeout_secs(std::env::var("DEVFLOW_CLAUDE_IDLE_TIMEOUT_SECS").ok())
}
pub enum MonitorLaunch {
PipeOwning {
prompt: String,
},
Legacy,
}
pub fn spawn_monitor(
state: &State,
program: &str,
args: &[String],
envs: &[(String, String)],
launch: MonitorLaunch,
) -> Result<u32, MonitorError> {
spawn_monitor_inner(state, program, args, envs, launch, true)
}
fn spawn_monitor_inner(
state: &State,
program: &str,
args: &[String],
envs: &[(String, String)],
launch: MonitorLaunch,
run_advance: bool,
) -> Result<u32, MonitorError> {
let project_root = state
.project_root
.to_str()
.ok_or(MonitorError::NonUtf8Path)?;
let binary = std::env::current_exe()
.map_err(|_| MonitorError::NoBinaryPath)?
.to_str()
.ok_or(MonitorError::NonUtf8Path)?
.to_string();
info!(
"spawning monitor for phase {}: {program} {}",
state.phase,
args.join(" ")
);
let stdout_file = crate::agent_result::stdout_path(&state.project_root, state.phase);
let stderr_file = crate::agent_result::stderr_path(&state.project_root, state.phase);
let exit_file = crate::agent_result::exit_code_path(&state.project_root, state.phase);
let pid_file = crate::agent_result::agent_pid_path(&state.project_root, state.phase);
if let Some(parent) = stdout_file.parent() {
crate::workflow::ensure_devflow_dir(parent)?;
}
let stdout_file = stdout_file.to_str().ok_or(MonitorError::NonUtf8Path)?;
let stderr_file = stderr_file.to_str().ok_or(MonitorError::NonUtf8Path)?;
let exit_file = exit_file.to_str().ok_or(MonitorError::NonUtf8Path)?;
let pid_file = pid_file.to_str().ok_or(MonitorError::NonUtf8Path)?;
let workdir_path = state
.worktree_path
.as_deref()
.unwrap_or(&state.project_root);
let workdir = workdir_path.to_str().ok_or(MonitorError::NonUtf8Path)?;
if let MonitorLaunch::PipeOwning { prompt } = launch {
let _ = run_advance;
if !envs.is_empty() {
warn!(
"pipe-owning monitor: {} adapter env var(s) will not survive the \
inner hermetic_command scrub — thread them explicitly before \
routing an env-setting adapter through this arm",
envs.len()
);
}
let idle = idle_timeout_setting();
if let Some(notice) = idle.notice() {
warn!("{notice}");
println!("{notice}");
}
let prompt_file = crate::agent_result::prompt_path(&state.project_root, state.phase);
std::fs::write(&prompt_file, &prompt)?;
let prompt_file = prompt_file.to_str().ok_or(MonitorError::NonUtf8Path)?;
let child = hermetic_command(&binary, workdir_path)
.arg("__monitor")
.arg("--project")
.arg(project_root)
.arg("--phase")
.arg(state.phase.to_string())
.arg("--workdir")
.arg(workdir)
.arg("--prompt-file")
.arg(prompt_file)
.arg("--idle-timeout-secs")
.arg(idle.timeout.as_secs().to_string())
.arg("--")
.arg(program)
.args(args)
.envs(envs.iter().map(|(k, v)| (k.as_str(), v.as_str())))
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()?;
let pid = child.id();
info!("pipe-owning monitor spawned with pid {pid}");
return Ok(pid);
}
let advance_tail = if run_advance {
format!(
"; {binary} advance {project_root} --phase {phase}",
binary = shell_escape(&binary),
project_root = shell_escape(project_root),
phase = state.phase,
)
} else {
String::new()
};
let script = format!(
"apid=''; cleanup() {{ [ -n \"$apid\" ] && kill \"$apid\" 2>/dev/null; exit 0; }}; \
trap cleanup TERM INT; \
cd {workdir} || exit 1; \
\"$@\" > {stdout_file} 2>{stderr_file} & \
apid=$!; echo $apid > {pid_file}; \
wait $apid; echo $? > {exit_file}{advance_tail}",
workdir = shell_escape(workdir),
stdout_file = shell_escape(stdout_file),
stderr_file = shell_escape(stderr_file),
exit_file = shell_escape(exit_file),
pid_file = shell_escape(pid_file),
);
let child = hermetic_command("sh", workdir_path)
.arg("-c")
.arg(&script)
.arg("sh")
.arg(program)
.args(args)
.envs(envs.iter().map(|(k, v)| (k.as_str(), v.as_str())))
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()?;
let pid = child.id();
info!("monitor spawned with pid {pid}");
Ok(pid)
}
#[derive(Default, PartialEq, Eq, Debug, Clone, Copy)]
enum BackgroundTaskState {
#[default]
NeverAnnounced,
Pending(usize),
Unreadable,
}
#[derive(Default)]
pub struct CloseRule {
marker_seen: bool,
background_tasks: BackgroundTaskState,
}
impl CloseRule {
pub fn observe(&mut self, line: &str) {
let Ok(event) = serde_json::from_str::<serde_json::Value>(line) else {
return;
};
if crate::agent_result::event_is_top_level_result_marker(&event) {
self.marker_seen = true;
}
if event.get("type").and_then(serde_json::Value::as_str) == Some("system")
&& event.get("subtype").and_then(serde_json::Value::as_str)
== Some("background_tasks_changed")
{
self.background_tasks = match event.get("tasks").and_then(serde_json::Value::as_array) {
Some(tasks) => BackgroundTaskState::Pending(tasks.len()),
None => BackgroundTaskState::Unreadable,
};
}
}
pub fn should_close(&self) -> bool {
self.marker_seen
&& matches!(
self.background_tasks,
BackgroundTaskState::NeverAnnounced | BackgroundTaskState::Pending(0)
)
}
}
pub fn user_turn_line(prompt: &str) -> String {
serde_json::json!({
"type": "user",
"message": { "role": "user", "content": prompt },
})
.to_string()
}
#[allow(clippy::too_many_arguments)]
pub fn run_pipe_owning_monitor(
project_root: &Path,
phase: u32,
workdir: &Path,
prompt: &str,
idle_timeout: Duration,
program: &str,
args: &[String],
envs: &[(String, String)],
) -> Result<i32, MonitorError> {
let stdout_file = crate::agent_result::stdout_path(project_root, phase);
let stderr_file = crate::agent_result::stderr_path(project_root, phase);
let exit_file = crate::agent_result::exit_code_path(project_root, phase);
let pid_file = crate::agent_result::agent_pid_path(project_root, phase);
if let Some(parent) = stdout_file.parent() {
crate::workflow::ensure_devflow_dir(parent)?;
}
let stderr_handle = std::fs::File::create(&stderr_file)?;
let mut capture = std::fs::File::create(&stdout_file)?;
let mut child = hermetic_command(program, workdir)
.args(args)
.envs(envs.iter().map(|(k, v)| (k.as_str(), v.as_str())))
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::from(stderr_handle))
.process_group(0)
.spawn()?;
let child_pid = child.id();
std::fs::write(&pid_file, format!("{child_pid}\n"))?;
let mut child_stdin = child
.stdin
.take()
.ok_or(MonitorError::NoChildPipe("stdin"))?;
let child_stdout = child
.stdout
.take()
.ok_or(MonitorError::NoChildPipe("stdout"))?;
let (close_tx, close_rx) = mpsc::channel::<()>();
let turn = user_turn_line(prompt);
let writer = std::thread::spawn(move || {
let wrote = child_stdin
.write_all(turn.as_bytes())
.and_then(|()| child_stdin.write_all(b"\n"))
.and_then(|()| child_stdin.flush());
if let Err(err) = wrote {
warn!("could not write the initial user turn to the child's stdin: {err}");
return;
}
let _ = close_rx.recv();
drop(child_stdin);
});
let (line_tx, line_rx) = mpsc::channel::<String>();
let reader = std::thread::spawn(move || {
let mut reader_buf = BufReader::new(child_stdout);
let mut raw = Vec::new();
loop {
raw.clear();
match reader_buf.read_until(b'\n', &mut raw) {
Ok(0) => break, Ok(_) => {}
Err(err) => {
warn!("stdout read error, treating as EOF: {err}");
break;
}
}
while raw.last().is_some_and(|b| *b == b'\n' || *b == b'\r') {
raw.pop();
}
let line = String::from_utf8_lossy(&raw).into_owned();
if let Err(err) = writeln!(capture, "{line}") {
warn!("could not append to the capture file: {err}");
}
let _ = capture.flush();
if line_tx.send(line).is_err() {
break;
}
}
});
let mut rule = CloseRule::default();
let mut close_signalled = false;
loop {
match line_rx.recv_timeout(idle_timeout) {
Ok(line) => {
if close_signalled {
continue;
}
rule.observe(&line);
if rule.should_close() {
let _ = close_tx.send(());
close_signalled = true;
}
}
Err(mpsc::RecvTimeoutError::Disconnected) => break,
Err(mpsc::RecvTimeoutError::Timeout) => {
if close_signalled {
info!(
"no output for {idle_timeout:?} after the close rule released stdin; \
the stage already reported — proceeding to reap, NOT recording a timeout"
);
break;
}
fire_idle_timeout(project_root, phase, workdir, child_pid, idle_timeout);
break;
}
}
}
drop(close_tx);
let status = child.wait()?;
let code = status.code().unwrap_or_else(|| {
use std::os::unix::process::ExitStatusExt;
status.signal().map_or(-1, |signal| 128 + signal)
});
std::fs::write(&exit_file, format!("{code}\n"))?;
let _ = writer.join();
let _ = reader.join();
info!("supervised child {child_pid} exited with code {code}");
Ok(code)
}
fn fire_idle_timeout(
project_root: &Path,
phase: u32,
workdir: &Path,
child_pid: u32,
idle: Duration,
) {
let idle_secs = idle.as_secs();
warn!("idle timeout: no output from the supervised child for {idle_secs}s");
let (commits, enumeration_note) = enumerate_phase_commits(workdir, phase);
let write_error =
write_idle_timeout_record(project_root, phase, idle_secs, child_pid, &commits)
.err()
.map(|err| err.to_string());
if let Some(err) = &write_error {
warn!("idle timeout: could not persist the verdict: {err}");
}
let terminated = terminate_child_group(child_pid);
let named: Vec<String> = commits
.iter()
.map(|commit| {
let short: String = commit.sha.chars().take(7).collect();
format!("{short} {}", commit.subject)
})
.collect();
let mut entry = format!(
"[idle-timeout] no output for {idle_secs}s; terminated agent pid {child_pid} \
(verified dead: {terminated}). {} commit(s) on the phase branch, NONE rolled back{}{}",
named.len(),
if named.is_empty() {
String::new()
} else {
format!(": {}", named.join("; "))
},
enumeration_note
.map(|note| format!(" [commit enumeration degraded: {note}]"))
.unwrap_or_default(),
);
if let Some(err) = write_error {
entry.push_str(&format!(" [verdict file could not be written: {err}]"));
}
warn!("{entry}");
append_monitor_log(project_root, phase, &entry);
}
fn enumerate_phase_commits(workdir: &Path, phase: u32) -> (Vec<IdleTimeoutCommit>, Option<String>) {
let git_flow = crate::config::GitFlowConfig::default();
let branch = format!("{}phase-{:02}", git_flow.feature_prefix, phase);
let range = format!("{}..{branch}", git_flow.develop);
let output = match crate::git::git_command(workdir)
.args(["log", "--format=%H %s", &range])
.output()
{
Ok(output) => output,
Err(err) => return (Vec::new(), Some(format!("git log could not run: {err}"))),
};
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
return (
Vec::new(),
Some(format!("git log {range} failed: {stderr}")),
);
}
let commits = String::from_utf8_lossy(&output.stdout)
.lines()
.filter_map(|line| {
let line = line.trim();
if line.is_empty() {
return None;
}
let (sha, subject) = line.split_once(' ').unwrap_or((line, ""));
Some(IdleTimeoutCommit {
sha: sha.to_string(),
subject: subject.to_string(),
})
})
.collect();
(commits, None)
}
fn write_idle_timeout_record(
project_root: &Path,
phase: u32,
idle_secs: u64,
child_pid: u32,
commits: &[IdleTimeoutCommit],
) -> std::io::Result<()> {
let record = IdleTimeoutRecord {
status: crate::agent_result::AgentStatus::IdleTimeout
.as_wire_str()
.to_string(),
idle_secs,
agent_pid: child_pid,
written_at: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0),
commits: commits.to_vec(),
};
let json = serde_json::to_string(&record)
.map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
let path = crate::agent_result::idle_timeout_path(project_root, phase);
if let Some(parent) = path.parent() {
crate::workflow::ensure_devflow_dir(parent)?;
}
let mut file = std::fs::File::create(&path)?;
file.write_all(json.as_bytes())?;
file.flush()?;
file.sync_all()
}
fn terminate_child_group(child_pid: u32) -> bool {
let Ok(signed) = libc::pid_t::try_from(child_pid) else {
warn!("idle timeout: child pid {child_pid} does not fit pid_t; not signalling");
return false;
};
if signed <= 1 {
warn!("idle timeout: refusing to signal group for pid {signed}");
return false;
}
unsafe {
libc::kill(-signed, libc::SIGTERM);
}
let dead = crate::agent::terminate_and_verify(
child_pid,
crate::agent::TERMINATE_VERIFY_WAIT,
crate::agent::TERMINATE_VERIFY_POLL,
);
unsafe {
libc::kill(-signed, libc::SIGKILL);
}
dead
}
fn append_monitor_log(project_root: &Path, phase: u32, entry: &str) {
let path = crate::agent_result::monitor_log_path(project_root, phase);
if let Ok(mut file) = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&path)
{
let _ = writeln!(file, "{entry}");
}
}
pub fn wait_for_agent_pid(project_root: &Path, phase: u32) -> Option<u32> {
let path = crate::agent_result::agent_pid_path(project_root, phase);
debug!("polling for agent PID for phase {phase}");
for _ in 0..50 {
if let Ok(contents) = std::fs::read_to_string(&path)
&& let Ok(pid) = contents.trim().parse::<u32>()
{
return Some(pid);
}
std::thread::sleep(Duration::from_millis(20));
}
debug!("agent PID not found for phase {phase} after polling");
None
}
fn shell_escape(s: &str) -> String {
format!("'{}'", s.replace('\'', "'\\''"))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::mode::Mode;
use crate::stage::Stage;
use crate::state::{AgentKind, State};
fn state_in(root: &Path) -> State {
let mut state = State::new(4, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Code;
state
}
const INIT_LINE: &str = r#"{"type":"system","subtype":"init","cwd":"/tmp/work","session_id":"s-1","tools":["Task","Bash"],"uuid":"u-init"}"#;
fn bg_tasks_line(count: usize) -> String {
let tasks: Vec<String> = (0..count)
.map(|i| {
format!(
r#"{{"task_id":"t{i}","task_type":"local_agent","description":"child {i}"}}"#
)
})
.collect();
format!(
r#"{{"type":"system","subtype":"background_tasks_changed","tasks":[{}],"uuid":"u-bg{count}","session_id":"s-1"}}"#,
tasks.join(",")
)
}
fn result_line(marker: &str) -> String {
format!(
r#"{{"type":"result","subtype":"success","is_error":false,"num_turns":3,"stop_reason":"end_turn","session_id":"s-1","uuid":"u-res","result":"{marker}"}}"#
)
}
fn coalesced_result_line(marker: &str) -> String {
format!(
r#"{{"type":"result","subtype":"success","is_error":false,"num_turns":2,"stop_reason":"end_turn","origin":{{"kind":"task-notification"}},"session_id":"s-1","uuid":"u-res-coalesced","result":"{marker}"}}"#
)
}
fn subagent_result_line(marker: &str) -> String {
result_line(marker).replacen('{', r#"{"parent_tool_use_id":"toolu_child","#, 1)
}
const MARKER: &str = r#"All done.\nDEVFLOW_RESULT: {\"status\":\"success\",\"commits\":3}"#;
const NO_MARKER: &str = "Acknowledged; nothing to report.";
fn observe_all(lines: &[String]) -> CloseRule {
let mut rule = CloseRule::default();
for line in lines {
rule.observe(line);
}
rule
}
#[test]
fn close_rule_requires_both_marker_and_drained_background_tasks() {
let drained_but_unmarked = observe_all(&[
INIT_LINE.to_string(),
bg_tasks_line(1),
bg_tasks_line(0),
r#"{"type":"result","result":"DEVFLOW_RESULT: {\"status\":\"succ"#.to_string(),
"progress: still working".to_string(),
result_line(NO_MARKER),
]);
assert!(
!drained_but_unmarked.should_close(),
"the drain alone must never close stdin: 30c/30d measured the \
drain-to-final-result lag at 4.54-11.51s across 14 trials, and \
closing at the drain would have truncated the final orchestrator \
turn in all seven 30d trials"
);
let marked_but_pending =
observe_all(&[INIT_LINE.to_string(), bg_tasks_line(1), result_line(MARKER)]);
assert!(
!marked_but_pending.should_close(),
"a marker while a background task is still announced must not \
close stdin — the pending child's task-notification turn would \
have nowhere to be delivered"
);
}
#[test]
fn unreadable_first_announcement_does_not_satisfy_the_drain_arm() {
let unreadable_first = observe_all(&[
INIT_LINE.to_string(),
r#"{"type":"system","subtype":"background_tasks_changed","tasks":null}"#.to_string(),
result_line(MARKER),
]);
assert!(
!unreadable_first.should_close(),
"an unreadable FIRST announcement must not be indistinguishable \
from never-announced — closing here would release stdin while \
the CLI has said a task exists whose count could not be read"
);
let never_announced = observe_all(&[INIT_LINE.to_string(), result_line(MARKER)]);
assert!(
never_announced.should_close(),
"a stage that never announces background tasks at all must still \
close on its marker alone — conflating NeverAnnounced with \
Unreadable would hang every ordinary stage for the full idle \
timeout"
);
let unreadable_after_pending = observe_all(&[
INIT_LINE.to_string(),
bg_tasks_line(1),
r#"{"type":"system","subtype":"background_tasks_changed","tasks":"not-an-array"}"#
.to_string(),
result_line(MARKER),
]);
assert!(
!unreadable_after_pending.should_close(),
"an unreadable announcement following a real pending count must \
still block closing, not silently forget the pending task"
);
}
#[test]
fn close_rule_is_vacuously_drained_when_no_background_tasks_event_appears() {
let rule = observe_all(&[
INIT_LINE.to_string(),
"starting up".to_string(),
r#"{"type":"assist"#.to_string(),
result_line(MARKER),
]);
assert!(
rule.should_close(),
"a stage that never announced a background task is drained by \
definition; only the marker arm has anything to satisfy"
);
}
#[test]
fn coalesced_completions_do_not_undercount_children() {
let rule = observe_all(&[
INIT_LINE.to_string(),
bg_tasks_line(2),
bg_tasks_line(0),
coalesced_result_line(MARKER),
]);
assert!(
rule.should_close(),
"two announced children, one drain event and one coalesced result \
must still close — a rule that matched result events against \
child count would stall here forever"
);
let undrained = observe_all(&[
INIT_LINE.to_string(),
bg_tasks_line(2),
coalesced_result_line(MARKER),
]);
assert!(
!undrained.should_close(),
"control: it is the drained list that decides, not the arrival of \
a result event"
);
}
#[test]
fn marker_inside_a_non_top_level_result_does_not_satisfy_the_close_rule() {
let subagent = observe_all(&[INIT_LINE.to_string(), subagent_result_line(MARKER)]);
assert!(
!subagent.should_close(),
"a subagent-origin result carrying a marker must not close the \
stream — same provenance hole constraint 9 item 2 closed for the \
stage verdict"
);
let top_level = observe_all(&[INIT_LINE.to_string(), result_line(MARKER)]);
assert!(
top_level.should_close(),
"control: the same event without a parent id is authoritative"
);
}
#[test]
fn shell_escape_wraps_basic_strings() {
assert_eq!(shell_escape("hello"), "'hello'");
assert_eq!(shell_escape("hello world"), "'hello world'");
assert_eq!(shell_escape("/tmp/devflow"), "'/tmp/devflow'");
}
#[test]
fn pipe_owning_monitor_delivers_prompt_via_stdin_and_captures_stream() {
const SENTINEL: &str = "TRACER-PROMPT-SENTINEL";
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 4u32;
std::fs::create_dir_all(root.join(".devflow")).unwrap();
let eof_file = root.join("stdin-eof");
let early_file = root.join("stdin-closed-early");
let prompt = format!("first line with a \" quote\n{SENTINEL}");
let script = format!(
r#"
set -u
IFS= read -r turn || {{ echo "NO_INITIAL_TURN_ON_STDIN" >&2; exit 91; }}
case "$turn" in
*{SENTINEL}*) ;;
*) echo "INITIAL_TURN_MISSING_PROMPT: $turn" >&2; exit 92 ;;
esac
# Probe: block on stdin until EOF, then record it. stdout is redirected so
# this subshell does not hold the capture pipe open after the main shell exits.
#
# `exec 3<&0` then `cat <&3` is load-bearing, not a flourish: POSIX assigns
# /dev/null to a BACKGROUNDED list's stdin before any explicit redirection
# when job control is off. A bare `( cat > /dev/null ) &` therefore reads EOF
# instantly and reports an early close that never happened. The explicit
# `<&3` is applied after that default and overrides it.
exec 3<&0
( cat <&3 > /dev/null; printf 'EOF\n' > '{eof}' ) > /dev/null 2>&1 &
printf '%s\n' '{{"type":"system","subtype":"init","session_id":"tracer-1"}}'
printf '%s\n' '{{"type":"system","subtype":"background_tasks_changed","tasks":[{{"task_id":"t1","task_type":"local_agent"}}]}}'
printf '%s\n' '{{"type":"system","subtype":"background_tasks_changed","tasks":[]}}'
# The drain has landed but no marker has. A correct monitor is still holding
# stdin open here; sample it and record the violation if it is not.
sleep 0.5
if [ -f '{eof}' ]; then printf 'EARLY\n' > '{early}'; fi
printf '%s\n' '{{"type":"result","subtype":"success","is_error":false,"session_id":"tracer-1","result":"DEVFLOW_RESULT: {{\"status\":\"success\",\"commits\":2}}"}}'
# Bounded wait for EOF: a monitor that never closes stdin must fail the
# assertions below, not hang the suite.
i=0
while [ $i -lt 100 ] && [ ! -f '{eof}' ]; do
sleep 0.1
i=$((i+1))
done
exit 0
"#,
eof = eof_file.display(),
early = early_file.display(),
);
let code = run_pipe_owning_monitor(
root,
phase,
root,
&prompt,
Duration::from_secs(20),
"sh",
&["-c".to_string(), script],
&[],
)
.expect("pipe-owning monitor should supervise the stub to completion");
let stderr = std::fs::read_to_string(crate::agent_result::stderr_path(root, phase))
.unwrap_or_default();
assert_eq!(
code, 0,
"stub exited {code}; 91 = no initial turn arrived on stdin, \
92 = the turn arrived but did not carry the prompt (a JSON \
escaping regression tears it across lines). stderr: {stderr:?}"
);
assert!(
!early_file.exists(),
"the monitor closed the child's stdin BEFORE the close rule was \
satisfied — the drain had landed but no DEVFLOW_RESULT marker had. \
Constraint 4's AND cannot be honoured once stdin is gone: a \
task-notification turn would have nowhere to be delivered."
);
assert!(
eof_file.exists(),
"the monitor never closed the child's stdin at all; the close rule \
should have fired once the marker arrived with the task list drained"
);
let capture =
std::fs::read_to_string(crate::agent_result::stdout_path(root, phase)).unwrap();
for expected in [
r#""subtype":"init""#,
r#""task_id":"t1""#,
r#""tasks":[]"#,
r#""type":"result""#,
] {
assert!(
capture.contains(expected),
"capture is missing {expected}; got:\n{capture}"
);
}
assert!(
crate::agent_result::capture_is_claude_stream(&capture),
"the capture must classify as a Claude stream-json document — \
this is what makes 30b's stream parser reachable at all:\n{capture}"
);
let result = crate::agent_result::evaluate_layer1(root, phase)
.expect("Layer 1 must decide this capture");
assert_eq!(
result.status,
crate::agent_result::AgentStatus::Success,
"Layer 1 verdict from the stream capture: {result:?}"
);
let exit = std::fs::read_to_string(crate::agent_result::exit_code_path(root, phase))
.expect("the monitor must record the child's exit code");
assert_eq!(exit.trim(), "0", "exit file contents: {exit:?}");
}
#[test]
fn non_utf8_byte_does_not_truncate_the_capture() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 11u32;
std::fs::create_dir_all(root.join(".devflow")).unwrap();
let script = r#"
set -u
IFS= read -r _turn || exit 91
printf '%s\n' '{"type":"system","subtype":"init","session_id":"utf8-1"}'
printf 'raw-\377-bytes\n'
printf '%s\n' '{"type":"system","subtype":"background_tasks_changed","tasks":[]}'
printf '%s\n' '{"type":"result","subtype":"success","is_error":false,"session_id":"utf8-1","result":"DEVFLOW_RESULT: {\"status\":\"success\"}"}'
exit 0
"#;
let code = run_pipe_owning_monitor(
root,
phase,
root,
"prompt",
Duration::from_secs(20),
"sh",
&["-c".to_string(), script.to_string()],
&[],
)
.expect("the monitor must survive a non-UTF-8 byte on the child's stdout");
assert_eq!(code, 0, "stub should exit cleanly");
let capture =
std::fs::read_to_string(crate::agent_result::stdout_path(root, phase)).unwrap();
assert!(
capture.contains(r#""type":"result""#),
"the terminal result event was lost: a non-UTF-8 byte earlier in the \
stream truncated the capture. This is the regression:\n{capture}"
);
assert!(
capture.contains("raw-"),
"the undecodable line itself must still be teed (lossily), since the \
capture is the verbatim record:\n{capture}"
);
let result = crate::agent_result::evaluate_layer1(root, phase)
.expect("Layer 1 must still decide a capture that contained a bad byte");
assert_eq!(
result.status,
crate::agent_result::AgentStatus::Success,
"verdict after lossy decode: {result:?}"
);
}
#[test]
fn no_idle_timeout_is_recorded_when_the_child_is_merely_slow_to_exit() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 12u32;
std::fs::create_dir_all(root.join(".devflow")).unwrap();
let script = r#"
set -u
IFS= read -r _turn || exit 91
printf '%s\n' '{"type":"system","subtype":"init","session_id":"slow-1"}'
printf '%s\n' '{"type":"system","subtype":"background_tasks_changed","tasks":[]}'
printf '%s\n' '{"type":"result","subtype":"success","is_error":false,"session_id":"slow-1","result":"DEVFLOW_RESULT: {\"status\":\"success\"}"}'
# Everything has been said; the close rule fires here. Now wind down slowly,
# well past the injected idle window, emitting nothing.
sleep 3
exit 0
"#;
let code = run_pipe_owning_monitor(
root,
phase,
root,
"prompt",
Duration::from_millis(600),
"sh",
&["-c".to_string(), script.to_string()],
&[],
)
.expect("a slow-exiting child that already reported is not a failure");
assert!(
!crate::agent_result::idle_timeout_path(root, phase).exists(),
"an idle-timeout verdict was written for a stage that had ALREADY \
emitted its terminal marker and drained its tasks — silence after a \
deliberate close is expected, not a hang"
);
assert_eq!(code, 0, "the child exited cleanly, if slowly");
let result = crate::agent_result::evaluate_layer1(root, phase)
.expect("Layer 1 must decide this capture");
assert_eq!(
result.status,
crate::agent_result::AgentStatus::Success,
"a completed stage must not be reported as a timeout: {result:?}"
);
}
#[test]
fn a_signal_killed_child_records_128_plus_signal_not_minus_one() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 13u32;
std::fs::create_dir_all(root.join(".devflow")).unwrap();
let script = r#"
set -u
IFS= read -r _turn || exit 91
printf '%s\n' '{"type":"system","subtype":"init","session_id":"sig-1"}'
kill -9 $$
"#;
let code = run_pipe_owning_monitor(
root,
phase,
root,
"prompt",
Duration::from_secs(20),
"sh",
&["-c".to_string(), script.to_string()],
&[],
)
.expect("the monitor must reap a signal-killed child");
assert_eq!(
code, 137,
"SIGKILL(9) must be recorded as 128+9=137, the value \
`evaluate_layer2` and `reconcile_stream_success_against_exit_code` \
map to ResourceKilled/GateInfra. -1 means the signal was discarded."
);
let exit = std::fs::read_to_string(crate::agent_result::exit_code_path(root, phase))
.expect("the monitor must record the exit code");
assert_eq!(exit.trim(), "137", "exit file contents: {exit:?}");
}
#[test]
fn shell_escape_handles_single_quotes() {
assert_eq!(shell_escape("can't"), "'can'\\''t'");
assert_eq!(shell_escape("a'b'c"), "'a'\\''b'\\''c'");
}
#[test]
fn shell_escape_handles_empty_string() {
assert_eq!(shell_escape(""), "''");
}
#[test]
fn wait_for_agent_pid_returns_pid_when_file_exists() {
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir_all(dir.path().join(".devflow")).unwrap();
std::fs::write(
crate::agent_result::agent_pid_path(dir.path(), 4),
"12345\n",
)
.unwrap();
assert_eq!(wait_for_agent_pid(dir.path(), 4), Some(12345));
}
#[test]
fn wait_for_agent_pid_returns_none_when_file_missing() {
let dir = tempfile::tempdir().unwrap();
assert_eq!(wait_for_agent_pid(dir.path(), 4), None);
}
#[test]
fn wait_for_agent_pid_returns_none_for_garbage_content() {
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir_all(dir.path().join(".devflow")).unwrap();
std::fs::write(
crate::agent_result::agent_pid_path(dir.path(), 4),
"not-a-pid",
)
.unwrap();
assert_eq!(wait_for_agent_pid(dir.path(), 4), None);
}
#[test]
fn spawn_monitor_captures_agent_pid_and_output() {
let dir = tempfile::tempdir().unwrap();
let state = state_in(dir.path());
let args = vec!["-c".to_string(), "echo MONITOR_READY".to_string()];
let monitor_pid = spawn_monitor(&state, "sh", &args, &[], MonitorLaunch::Legacy).unwrap();
assert!(monitor_pid > 0);
let agent_pid = wait_for_agent_pid(dir.path(), state.phase)
.expect("monitor should record the agent pid");
assert!(agent_pid > 0);
let stdout_path = crate::agent_result::stdout_path(dir.path(), state.phase);
let mut captured = String::new();
for _ in 0..100 {
if let Ok(contents) = std::fs::read_to_string(&stdout_path)
&& contents.contains("MONITOR_READY")
{
captured = contents;
break;
}
std::thread::sleep(Duration::from_millis(20));
}
assert!(
captured.contains("MONITOR_READY"),
"expected MONITOR_READY in captured stdout, got {captured:?}"
);
}
fn proc_snapshot(pid: u32) -> String {
let Ok(status) = std::fs::read_to_string(format!("/proc/{pid}/status")) else {
return format!("GONE (no /proc/{pid})");
};
let field = |key: &str| {
status
.lines()
.find(|l| l.starts_with(key))
.map(|l| l.split_whitespace().skip(1).collect::<Vec<_>>().join(" "))
.unwrap_or_else(|| "?".into())
};
let cmdline = std::fs::read(format!("/proc/{pid}/cmdline"))
.map(|raw| {
let joined = raw
.split(|&b| b == 0)
.filter(|a| !a.is_empty())
.map(|a| String::from_utf8_lossy(a).into_owned())
.collect::<Vec<_>>()
.join(" ");
if joined.is_empty() {
"<empty>".to_string()
} else {
joined
}
})
.unwrap_or_else(|e| format!("<unreadable: {e}>"));
format!(
"ALIVE Name={} State={} PPid={} cmdline=[{cmdline}]",
field("Name:"),
field("State:"),
field("PPid:")
)
}
#[test]
fn sigterm_to_monitor_also_kills_the_agent() {
let dir = tempfile::tempdir().unwrap();
let state = state_in(dir.path());
let args = vec!["-c".to_string(), "sleep 30".to_string()];
let monitor_pid = spawn_monitor(&state, "sh", &args, &[], MonitorLaunch::Legacy).unwrap();
let agent_pid = wait_for_agent_pid(dir.path(), state.phase)
.expect("monitor should record the agent pid");
assert!(
crate::agent::agent_running(agent_pid),
"agent should be running before SIGTERM"
);
let monitor_before = proc_snapshot(monitor_pid);
let agent_before = proc_snapshot(agent_pid);
let kill_rc = unsafe { libc::kill(monitor_pid as libc::pid_t, libc::SIGTERM) };
let kill_err = if kill_rc == 0 {
"ok".to_string()
} else {
format!("errno {}", std::io::Error::last_os_error())
};
let mut still_running = true;
for _ in 0..250 {
if !crate::agent::agent_running(agent_pid) {
still_running = false;
break;
}
std::thread::sleep(Duration::from_millis(20));
}
let monitor_after = proc_snapshot(monitor_pid);
let agent_after = proc_snapshot(agent_pid);
let pidfile =
std::fs::read_to_string(crate::agent_result::agent_pid_path(dir.path(), state.phase))
.unwrap_or_else(|e| format!("<unreadable: {e}>"));
assert!(
!still_running,
"agent (pid {agent_pid}) was orphaned — still running after monitor SIGTERM\n\
\x20 monitor pid: {monitor_pid}\n\
\x20 kill(TERM) rc: {kill_rc} ({kill_err})\n\
\x20 monitor before: {monitor_before}\n\
\x20 monitor after: {monitor_after}\n\
\x20 agent pid: {agent_pid}\n\
\x20 agent before: {agent_before}\n\
\x20 agent after: {agent_after}\n\
\x20 pidfile contents: {}\n\
Read the monitor's `after` line first. GONE means the shell died \
without running its trap — most likely SIGTERM arrived before \
`trap` was installed, or it was killed rather than handling the \
signal, either way leaving the agent unreaped. STILL ALIVE means \
the trap never fired or `kill $apid` failed, so compare the agent \
pid against the pidfile and check the agent's PPid: if PPid is not \
the monitor, `$!` did not name the process we are polling. If the \
agent's Name is `sh` rather than `sleep`, the agent shell forked \
rather than exec'd, so killing it leaves its own child behind.",
pidfile.trim()
);
}
#[test]
fn spawn_monitor_runs_agent_in_worktree_but_captures_in_project_root() {
let dir = tempfile::tempdir().unwrap();
let worktree = dir.path().join(".worktrees/phase-04");
std::fs::create_dir_all(&worktree).unwrap();
let mut state = state_in(dir.path());
state.worktree_path = Some(worktree.clone());
let args = vec!["-c".to_string(), "pwd; echo WORKTREE_READY".to_string()];
let monitor_pid = spawn_monitor(&state, "sh", &args, &[], MonitorLaunch::Legacy).unwrap();
assert!(monitor_pid > 0);
let agent_pid = wait_for_agent_pid(dir.path(), state.phase)
.expect("monitor should record the agent pid in the main project");
assert!(agent_pid > 0);
let stdout_path = crate::agent_result::stdout_path(dir.path(), state.phase);
let mut captured = String::new();
for _ in 0..100 {
if let Ok(contents) = std::fs::read_to_string(&stdout_path)
&& contents.contains("WORKTREE_READY")
{
captured = contents;
break;
}
std::thread::sleep(Duration::from_millis(20));
}
assert!(
captured.contains(&worktree.display().to_string()),
"agent did not run in worktree cwd; captured stdout: {captured:?}"
);
assert!(
stdout_path.exists(),
"stdout capture missing in main .devflow"
);
assert!(
!crate::agent_result::stdout_path(&worktree, state.phase).exists(),
"stdout capture should not be written under the worktree"
);
}
fn git(root: &Path, args: &[&str]) {
let ok = crate::test_support::git_command(root)
.args(args)
.output()
.unwrap()
.status
.success();
assert!(ok, "git {args:?} failed");
}
fn init_repo(root: &Path) {
git(root, &["init", "-q"]);
git(root, &["config", "user.email", "test@example.com"]);
git(root, &["config", "user.name", "Test"]);
}
#[test]
fn spawn_monitor_agent_git_calls_resolve_workdir_not_a_hostile_git_dir() {
const INNER_ROOT: &str = "DEVFLOW_27_MONITOR_INNER_ROOT";
if let Ok(root) = std::env::var(INNER_ROOT) {
let root = std::path::PathBuf::from(root);
let state = state_in(&root);
let args = vec![
"-c".to_string(),
"git rev-parse --absolute-git-dir".to_string(),
];
spawn_monitor(&state, "sh", &args, &[], MonitorLaunch::Legacy).unwrap();
wait_for_agent_pid(&root, state.phase).expect("monitor should record the agent pid");
let stdout_path = crate::agent_result::stdout_path(&root, state.phase);
let mut captured = String::new();
for _ in 0..100 {
if let Ok(contents) = std::fs::read_to_string(&stdout_path)
&& !contents.trim().is_empty()
{
captured = contents;
break;
}
std::thread::sleep(Duration::from_millis(20));
}
let resolved = std::fs::canonicalize(captured.trim())
.expect("agent's reported git-dir must exist on disk");
let expected =
std::fs::canonicalize(root.join(".git")).expect("caller repo .git must exist");
assert_eq!(
resolved, expected,
"agent's git call resolved to a hostile GIT_DIR's \
repository instead of the caller's own workdir: \
got {resolved:?}, want {expected:?}"
);
return;
}
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("caller-repo");
std::fs::create_dir_all(&root).unwrap();
init_repo(&root);
let foreign = tempfile::tempdir().unwrap();
init_repo(foreign.path());
let exe = std::env::current_exe().expect("current_exe for child re-invocation");
let out = std::process::Command::new(&exe)
.arg("spawn_monitor_agent_git_calls_resolve_workdir_not_a_hostile_git_dir")
.arg("--test-threads=1")
.env(INNER_ROOT, root.to_str().unwrap())
.env("GIT_DIR", foreign.path().join(".git"))
.output()
.expect("spawn hostile child test process");
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(
stdout.contains("1 passed"),
"child test process must have run exactly the inner test; \
stdout:\n{stdout}"
);
assert!(
out.status.success(),
"monitor-spawned agent (hostile GIT_DIR pointed at an \
unrelated foreign repository) must still resolve its git \
calls against the caller's own workdir; child exit status \
{:?}\nstdout:\n{stdout}",
out.status
);
}
#[test]
fn spawn_monitor_treats_agent_args_as_literal_argv() {
let dir = tempfile::tempdir().unwrap();
let state = state_in(dir.path());
let payload = "value; touch INJECTED";
let args = vec![
"-c".to_string(),
"printf '%s\\n' \"$0\"; echo ARGV_SAFE".to_string(),
payload.to_string(),
];
spawn_monitor(&state, "sh", &args, &[], MonitorLaunch::Legacy).unwrap();
wait_for_agent_pid(dir.path(), state.phase).expect("monitor should record the agent pid");
let stdout_path = crate::agent_result::stdout_path(dir.path(), state.phase);
let mut captured = String::new();
for _ in 0..100 {
if let Ok(contents) = std::fs::read_to_string(&stdout_path)
&& contents.contains("ARGV_SAFE")
{
captured = contents;
break;
}
std::thread::sleep(Duration::from_millis(20));
}
assert!(
captured.contains(payload),
"literal argv missing: {captured:?}"
);
assert!(captured.contains("ARGV_SAFE"));
assert!(!dir.path().join("INJECTED").exists());
}
#[test]
fn idle_timeout_secs_clamps_below_floor_and_logs() {
let setting = parse_idle_timeout_secs(Some("5".to_string()));
assert_eq!(setting.timeout, Duration::from_secs(120));
assert!(setting.clamped(), "the clamp must be observable as a value");
assert_eq!(
setting.resolution,
IdleTimeoutResolution::Clamped { configured: 5 }
);
let notice = setting.notice().expect("a clamp owes a loud notice");
for fragment in ["5", "120", IDLE_TIMEOUT_ENV] {
assert!(
notice.contains(fragment),
"notice must name {fragment:?}; got: {notice}"
);
}
}
#[test]
fn idle_timeout_secs_accepts_values_above_floor() {
let setting = parse_idle_timeout_secs(Some("300".to_string()));
assert_eq!(setting.timeout, Duration::from_secs(300));
assert!(!setting.clamped());
assert_eq!(setting.resolution, IdleTimeoutResolution::Configured);
assert_eq!(
setting.notice(),
None,
"an honoured value is unremarkable and must not shout"
);
let exact = parse_idle_timeout_secs(Some("120".to_string()));
assert_eq!(exact.resolution, IdleTimeoutResolution::Configured);
assert!(!exact.clamped());
}
#[test]
fn idle_timeout_secs_defaults_to_the_floor() {
let floor = Duration::from_secs(IDLE_TIMEOUT_FLOOR_SECS);
for raw in [None, Some(String::new()), Some(" ".to_string())] {
let setting = parse_idle_timeout_secs(raw.clone());
assert_eq!(setting.timeout, floor, "raw {raw:?} must yield the floor");
assert_eq!(setting.resolution, IdleTimeoutResolution::Default);
assert_eq!(setting.notice(), None, "nothing chosen is not an error");
}
for raw in ["banana", "60O", "-5", "30.5"] {
let setting = parse_idle_timeout_secs(Some(raw.to_string()));
assert_eq!(setting.timeout, floor, "raw {raw:?} must yield the floor");
assert_eq!(
setting.resolution,
IdleTimeoutResolution::Unparseable {
raw: raw.to_string()
}
);
assert!(
setting.notice().is_some(),
"a typo that silently halves an intended timeout must be loud: {raw:?}"
);
}
}
#[test]
fn idle_timer_resets_on_every_stream_line() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 6u32;
std::fs::create_dir_all(root.join(".devflow")).unwrap();
let script = r#"
set -u
IFS= read -r turn || exit 91
i=0
while [ $i -lt 12 ]; do
printf '%s\n' '{"type":"system","subtype":"heartbeat","n":'"$i"'}'
sleep 0.1
i=$((i+1))
done
printf '%s\n' '{"type":"result","subtype":"success","is_error":false,"session_id":"idle-1","result":"DEVFLOW_RESULT: {\"status\":\"success\"}"}'
exit 0
"#;
let started = std::time::Instant::now();
let code = run_pipe_owning_monitor(
root,
phase,
root,
"prompt",
Duration::from_millis(400),
"sh",
&["-c".to_string(), script.to_string()],
&[],
)
.expect("a chatty child must be supervised to completion");
let elapsed = started.elapsed();
assert_eq!(code, 0, "the chatty child must exit cleanly, not be killed");
assert!(
!crate::agent_result::idle_timeout_path(root, phase).exists(),
"no timeout may fire while the child is still emitting lines"
);
assert!(
elapsed > Duration::from_millis(400),
"the run must outlast the idle window, else it proves nothing \
about resetting: {elapsed:?}"
);
let capture =
std::fs::read_to_string(crate::agent_result::stdout_path(root, phase)).unwrap();
assert_eq!(
capture.matches("heartbeat").count(),
12,
"all twelve resets must have been observed: {capture:?}"
);
}
#[test]
fn idle_timeout_writes_side_channel_before_terminating_child() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().to_path_buf();
let phase = 7u32;
std::fs::create_dir_all(root.join(".devflow")).unwrap();
let script = r#"
set -u
IFS= read -r turn || exit 91
trap '' TERM
printf '%s\n' '{"type":"system","subtype":"init","session_id":"idle-2"}'
sleep 120
"#;
let verdict = crate::agent_result::idle_timeout_path(&root, phase);
let pid_file = crate::agent_result::agent_pid_path(&root, phase);
let watcher = std::thread::spawn(move || {
let deadline = std::time::Instant::now() + Duration::from_secs(30);
let mut pid: Option<u32> = None;
while std::time::Instant::now() < deadline {
if pid.is_none() {
pid = std::fs::read_to_string(&pid_file)
.ok()
.and_then(|s| s.trim().parse::<u32>().ok());
}
if verdict.exists() {
return pid.map(crate::agent::agent_running);
}
std::thread::sleep(Duration::from_millis(5));
}
None
});
let code = run_pipe_owning_monitor(
&root,
phase,
&root,
"prompt",
Duration::from_millis(250),
"sh",
&["-c".to_string(), script.to_string()],
&[],
)
.expect("a silent child must still produce a supervised outcome");
let observed = watcher.join().expect("watcher thread panicked");
assert_eq!(
observed,
Some(true),
"the verdict must be on disk while the child is STILL ALIVE. \
Some(false) = written after termination (the D-05 violation); \
None = the verdict never appeared at all"
);
let raw = std::fs::read_to_string(crate::agent_result::idle_timeout_path(&root, phase))
.expect("verdict file must be readable");
let record: IdleTimeoutRecord = serde_json::from_str(&raw).expect("verdict must parse");
assert_eq!(record.status, "idle_timeout");
assert_eq!(record.idle_secs, 0, "250ms truncates to 0 whole seconds");
assert!(record.agent_pid > 1);
let result = crate::agent_result::evaluate_layer1(&root, phase)
.expect("Layer 1 must decide a timed-out run");
assert_eq!(
result.status,
crate::agent_result::AgentStatus::IdleTimeout,
"the monitor's verdict must survive all the way to the oracle"
);
assert!(
crate::agent_result::exit_code_path(&root, phase).exists(),
"the exit file must still be written so advance() is reachable"
);
let _ = code;
let log = std::fs::read_to_string(crate::agent_result::monitor_log_path(&root, phase))
.expect("the monitor must log its own timeout");
assert!(log.contains("idle-timeout"), "log entry missing: {log:?}");
}
fn init_repo_with_feature_commits(root: &Path, phase: u32, commits: usize) {
let git = |args: &[&str]| {
let output = crate::git::git_command(root).args(args).output().unwrap();
assert!(
output.status.success(),
"git {args:?} failed: {}",
String::from_utf8_lossy(&output.stderr)
);
};
git(&["init"]);
git(&["config", "user.email", "devflow@example.com"]);
git(&["config", "user.name", "DevFlow Tests"]);
git(&["config", "commit.gpgsign", "false"]);
git(&["config", "core.hooksPath", "/dev/null"]);
git(&["checkout", "-b", "develop"]);
std::fs::write(root.join("README.md"), "base\n").unwrap();
git(&["add", "README.md"]);
git(&["commit", "-m", "base"]);
let branch = format!("feature/phase-{phase:02}");
git(&["checkout", "-b", &branch]);
for i in 0..commits {
let name = format!("work-{i}.txt");
std::fs::write(root.join(&name), "work\n").unwrap();
git(&["add", &name]);
git(&["commit", "-m", &format!("feat: agent work {i}")]);
}
}
fn commit_count(root: &Path, phase: u32) -> u32 {
let range = format!("develop..feature/phase-{phase:02}");
let output = crate::git::git_command(root)
.args(["rev-list", "--count", &range])
.output()
.unwrap();
String::from_utf8_lossy(&output.stdout)
.trim()
.parse()
.unwrap()
}
#[test]
fn idle_timeout_does_not_roll_back_commits() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 8u32;
init_repo_with_feature_commits(root, phase, 2);
std::fs::create_dir_all(root.join(".devflow")).unwrap();
let before = commit_count(root, phase);
assert_eq!(before, 2, "fixture precondition");
let script = r#"
set -u
IFS= read -r turn || exit 91
printf '%s\n' '{"type":"system","subtype":"init","session_id":"idle-3"}'
sleep 120
"#;
run_pipe_owning_monitor(
root,
phase,
root,
"prompt",
Duration::from_millis(250),
"sh",
&["-c".to_string(), script.to_string()],
&[],
)
.expect("a silent child must still produce a supervised outcome");
assert_eq!(
commit_count(root, phase),
before,
"an idle timeout must never roll back, reset, or revert a commit"
);
let raw = std::fs::read_to_string(crate::agent_result::idle_timeout_path(root, phase))
.expect("verdict file must exist");
let record: IdleTimeoutRecord = serde_json::from_str(&raw).expect("verdict must parse");
assert_eq!(
record.commits.len(),
2,
"the verdict must NAME the commits, not merely leave them alone"
);
for commit in &record.commits {
assert_eq!(commit.sha.len(), 40, "full sha expected: {commit:?}");
assert!(
commit.subject.starts_with("feat: agent work"),
"subject must survive enumeration: {commit:?}"
);
}
let result = crate::agent_result::evaluate_layer1(root, phase).unwrap();
assert_eq!(result.commits, Some(2));
let reason = result.reason.unwrap();
assert!(
reason.contains("NONE of them were rolled back"),
"reason: {reason}"
);
}
}