devflow_core/monitor.rs
1//! Background monitor daemon.
2//!
3//! Spawns a detached child process that *owns* the coding agent: it launches
4//! the agent, captures its stdout and exit code into `.devflow/`, and — when
5//! the agent exits — runs `devflow advance` to advance the stage machine.
6//!
7//! Owning the agent is the key fix over a CLI-scoped capture thread: because
8//! the monitor outlives `devflow start`, the agent's stdout keeps flowing into
9//! the capture file and its exit code is still reaped after the CLI exits.
10//!
11//! This is the core automation primitive — no cron, no scheduler,
12//! no agent cooperation needed.
13
14use crate::agent_result::{IdleTimeoutCommit, IdleTimeoutRecord};
15use crate::git::hermetic_command;
16use crate::state::State;
17use std::io::{BufRead, BufReader, Write};
18use std::os::unix::process::CommandExt;
19use std::path::Path;
20use std::process::Stdio;
21use std::sync::mpsc;
22use std::time::Duration;
23use tracing::{debug, info, warn};
24
25/// Errors produced by monitor operations.
26#[derive(Debug, thiserror::Error)]
27pub enum MonitorError {
28 /// Spawning the monitor process failed.
29 #[error("failed to spawn monitor: {0}")]
30 Io(#[from] std::io::Error),
31 /// Project path is not valid UTF-8.
32 #[error("project path is not valid UTF-8")]
33 NonUtf8Path,
34 /// Could not determine the current executable path.
35 #[error("could not determine devflow binary path")]
36 NoBinaryPath,
37 /// A child spawned with piped stdio did not expose one of its pipes.
38 #[error("supervised child exposed no {0} pipe")]
39 NoChildPipe(&'static str),
40}
41
42/// Idle-timeout default in seconds (D-02): the measured constraint-8 floor.
43///
44/// Plan 31-02 supplies the configurable-and-clamped reader that can only raise
45/// this. Until then `spawn_monitor` passes this literal to the monitor process.
46///
47/// Raised 30s -> 120s on 2026-08-03 by direct measurement; see
48/// [`IDLE_TIMEOUT_FLOOR_SECS`] for the trials and the reasoning. The previous
49/// value's "~4.2x margin" was computed against a workload that never entered a
50/// long foreground tool call, and did not transfer to one.
51pub const DEFAULT_IDLE_TIMEOUT_SECS: u64 = 120;
52
53/// The floor an idle timeout can never be configured below (D-02/D-04, 31-02).
54///
55/// **Raised 30s -> 120s on 2026-08-03, and the reasoning that set 30s was
56/// wrong — read this before touching it again.**
57///
58/// The original ≥30s floor cited "~4.2x margin" against an every-line signal
59/// whose observed max was 7.09s. Both numbers were real; the inference was not.
60/// Phase 30d measured *backgrounded* 10s/22s sleeps, where the agent is never
61/// sitting inside a long foreground tool call. Under one, the CLI emits
62/// `tool_progress` keepalives on a **fixed 30.00s interval**, so a healthy,
63/// hard-working child produces a 30.00s gap between stream lines — dead level
64/// with a 30s timeout, and on the wrong side of it, since the timer starts when
65/// the previous line is *processed* while the keepalive arrives 30s after it
66/// was *sent*, plus pipe latency.
67///
68/// Measured 2026-08-03, CLI 2.1.220, five workload-controlled trials across two
69/// unrelated workload types (each verified to have actually run — elapsed >=
70/// the workload duration, no `tool_use_error`), plus a negative control:
71///
72/// | workload | gaps > 5s |
73/// |---------------------------|------------------------|
74/// | 90s busy loop x3 | ~26.4, **30.00**, ~30.0 |
75/// | `cargo test --workspace` x2 | ~26.4, **30.00**, ~16 |
76/// | control (no long call) | max 2.2 |
77///
78/// Variance across all five: ±0.02s. `cargo test --workspace` is not a contrived
79/// case — it sits inside DevFlow's own post-merge gate, so the old floor would
80/// have killed healthy Code stages on the common path.
81///
82/// 120s is 4x the measured cadence: it survives **three** consecutive missed
83/// keepalives. That headroom is the point — the hazard is not a slightly larger
84/// gap but a *dropped* keepalive, which doubles the interval outright. 90s
85/// (two missed) is the lowest defensible value; do not go below it.
86///
87/// Do NOT lower it, and note that no configuration can. Phase 30d measured a
88/// 12-second bound killing a LIVE, HEALTHY run in 2 of 7 trials.
89///
90/// **What the five trials do not establish:** one machine, idle, one CLI
91/// version, two workload types. They show the 30.00s cadence is real and
92/// reproducible; they do not prove the interval is fixed across load, hardware,
93/// or CLI versions. That is precisely why this floor sits well above the
94/// observed maximum rather than near it.
95///
96/// Because the default IS the floor, the value can only ever be raised.
97pub const IDLE_TIMEOUT_FLOOR_SECS: u64 = 120;
98
99/// The environment variable that raises the idle timeout above its floor.
100pub const IDLE_TIMEOUT_ENV: &str = "DEVFLOW_CLAUDE_IDLE_TIMEOUT_SECS";
101
102/// How [`parse_idle_timeout_secs`] arrived at the timeout now in force.
103///
104/// A distinct enum rather than the plain `clamped: bool` the plan sketched:
105/// there are FOUR distinguishable resolutions, not two, and the loud operator
106/// notice needs to name the value that was configured — which a bool cannot
107/// carry. `ValidateOutcome` in `pipeline_outcomes.rs` makes the same argument
108/// for the same reason.
109#[derive(Debug, Clone, PartialEq, Eq)]
110pub enum IdleTimeoutResolution {
111 /// Nothing was configured; the default — which is the floor — is in force.
112 Default,
113 /// A configured value at or above the floor is in force verbatim.
114 Configured,
115 /// A configured value BELOW the floor was raised to it (D-04).
116 Clamped {
117 /// What the operator asked for, for the notice to name.
118 configured: u64,
119 },
120 /// A value was set but could not be parsed; the default is in force.
121 ///
122 /// Loud for the same reason the clamp is. An operator who meant `600` and
123 /// typed `60O` silently gets the 120s default, and a legitimately slow stage then dies
124 /// on a timeout nobody chose. `parse_gate_max_unattended_age` substitutes
125 /// silently in this case and is the anti-pattern here, not the precedent.
126 Unparseable {
127 /// The raw value, echoed back so the typo is visible.
128 raw: String,
129 },
130}
131
132/// A resolved idle timeout together with how it was arrived at.
133#[derive(Debug, Clone, PartialEq, Eq)]
134pub struct IdleTimeoutSetting {
135 /// The window that must elapse with NO line on the child's stdout.
136 pub timeout: Duration,
137 /// How that value was reached — observable to the caller as a VALUE, not
138 /// only as a log line, so a test can assert on it directly.
139 pub resolution: IdleTimeoutResolution,
140}
141
142impl IdleTimeoutSetting {
143 /// Whether the floor clamp engaged.
144 pub fn clamped(&self) -> bool {
145 matches!(self.resolution, IdleTimeoutResolution::Clamped { .. })
146 }
147
148 /// The loud, operator-facing notice this resolution owes, if any.
149 ///
150 /// `None` for the two unremarkable cases. `Some` exactly when a value the
151 /// operator supplied is NOT the value in force — the case that must never
152 /// pass silently.
153 pub fn notice(&self) -> Option<String> {
154 match &self.resolution {
155 IdleTimeoutResolution::Default | IdleTimeoutResolution::Configured => None,
156 IdleTimeoutResolution::Clamped { configured } => Some(format!(
157 "{IDLE_TIMEOUT_ENV}={configured} is below the {IDLE_TIMEOUT_FLOOR_SECS}s floor \
158 and was CLAMPED; {}s is in force. A shorter window kills healthy runs: a 12s \
159 bound terminated a live, healthy run in 2 of 7 measured trials.",
160 self.timeout.as_secs()
161 )),
162 IdleTimeoutResolution::Unparseable { raw } => Some(format!(
163 "{IDLE_TIMEOUT_ENV}={raw:?} could not be parsed as a whole number of seconds; \
164 the {}s default is in force. If you meant to RAISE the timeout, this did not \
165 do it.",
166 self.timeout.as_secs()
167 )),
168 }
169 }
170}
171
172/// Resolve a raw idle-timeout override into the value actually in force.
173///
174/// Pure — no environment access — so it is unit-testable directly rather than
175/// by mutating process-global env. That shape is copied from
176/// `devflow-cli`'s four `parse_*` timeout readers; their BEHAVIOUR is
177/// deliberately not copied, because none of them clamps against a floor and
178/// none logs when a fallback engages. There is no clamp-and-log precedent
179/// anywhere in this workspace; this is the first (D-04).
180pub fn parse_idle_timeout_secs(raw: Option<String>) -> IdleTimeoutSetting {
181 let floor = Duration::from_secs(IDLE_TIMEOUT_FLOOR_SECS);
182
183 // An unset variable and an EMPTY one are the same intent: nothing chosen.
184 // Only a non-empty value that fails to parse is a typo worth shouting at.
185 let Some(trimmed) = raw.as_deref().map(str::trim).filter(|s| !s.is_empty()) else {
186 return IdleTimeoutSetting {
187 timeout: floor,
188 resolution: IdleTimeoutResolution::Default,
189 };
190 };
191
192 let Ok(configured) = trimmed.parse::<u64>() else {
193 return IdleTimeoutSetting {
194 timeout: floor,
195 resolution: IdleTimeoutResolution::Unparseable {
196 raw: trimmed.to_string(),
197 },
198 };
199 };
200
201 if configured < IDLE_TIMEOUT_FLOOR_SECS {
202 IdleTimeoutSetting {
203 timeout: floor,
204 resolution: IdleTimeoutResolution::Clamped { configured },
205 }
206 } else {
207 IdleTimeoutSetting {
208 timeout: Duration::from_secs(configured),
209 resolution: IdleTimeoutResolution::Configured,
210 }
211 }
212}
213
214/// The thin environment wrapper over [`parse_idle_timeout_secs`].
215///
216/// The variable name is spelled out as a STRING LITERAL here rather than
217/// passed as [`IDLE_TIMEOUT_ENV`], and that is deliberate.
218/// `doc_check::source_read_env_vars` recognises a variable only when it is read
219/// through a literal inside `std::env::var("...")`; reading it through the
220/// const compiles and works identically but makes the variable INVISIBLE to
221/// the operator-doc parity gate, which would then pass green while the
222/// variable went undocumented. Verified by removing this variable's row from
223/// `OPERATIONS.md` and confirming `doc_check` reddens.
224pub fn idle_timeout_setting() -> IdleTimeoutSetting {
225 parse_idle_timeout_secs(std::env::var("DEVFLOW_CLAUDE_IDLE_TIMEOUT_SECS").ok())
226}
227
228/// Which supervision shape [`spawn_monitor`] should launch.
229///
230/// This is a MODE selection on one supervisor, not two monitors: both arms
231/// write the same capture, exit-code and agent-pid files under `.devflow/`,
232/// and both end by advancing the same stage machine. Nothing downstream needs
233/// to know which arm ran.
234pub enum MonitorLaunch {
235 /// Phase 31: a Rust supervisor that owns BOTH of the child's pipes,
236 /// delivers `prompt` as a JSON user turn on the child's stdin, and holds
237 /// that stdin open past the child's first turn so a task-notification turn
238 /// can still be delivered (constraint 4).
239 PipeOwning {
240 /// The stage prompt, delivered on the child's stdin rather than argv.
241 prompt: String,
242 },
243 /// The pre-31 detached `sh` script: stdin is `/dev/null`, stdout is
244 /// redirected to the capture file by the shell, and the script waits on
245 /// the agent then runs `devflow advance`. Every non-Claude adapter, every
246 /// stage not yet widened by D-09/D-10's rollout, and the checkpoint-resume
247 /// relaunch all run through here, unchanged.
248 Legacy,
249}
250
251/// Spawn a background monitor that owns the agent for the given workflow state.
252///
253/// The monitor is a detached process that:
254/// 1. Launches the agent (`program` + `args`) with stdout captured to the
255/// phase stdout file, recording the agent PID to the agent-pid file
256/// 2. Waits for the agent to exit and records its exit code to the exit file
257/// 3. Runs `devflow advance --phase N` to advance the workflow through its
258/// remaining stages
259///
260/// `launch` selects the supervision shape — see [`MonitorLaunch`].
261///
262/// Returns the PID of the spawned monitor.
263pub fn spawn_monitor(
264 state: &State,
265 program: &str,
266 args: &[String],
267 envs: &[(String, String)],
268 launch: MonitorLaunch,
269) -> Result<u32, MonitorError> {
270 spawn_monitor_inner(state, program, args, envs, launch, true)
271}
272
273fn spawn_monitor_inner(
274 state: &State,
275 program: &str,
276 args: &[String],
277 envs: &[(String, String)],
278 launch: MonitorLaunch,
279 run_advance: bool,
280) -> Result<u32, MonitorError> {
281 let project_root = state
282 .project_root
283 .to_str()
284 .ok_or(MonitorError::NonUtf8Path)?;
285
286 let binary = std::env::current_exe()
287 .map_err(|_| MonitorError::NoBinaryPath)?
288 .to_str()
289 .ok_or(MonitorError::NonUtf8Path)?
290 .to_string();
291
292 info!(
293 "spawning monitor for phase {}: {program} {}",
294 state.phase,
295 args.join(" ")
296 );
297
298 let stdout_file = crate::agent_result::stdout_path(&state.project_root, state.phase);
299 let stderr_file = crate::agent_result::stderr_path(&state.project_root, state.phase);
300 let exit_file = crate::agent_result::exit_code_path(&state.project_root, state.phase);
301 let pid_file = crate::agent_result::agent_pid_path(&state.project_root, state.phase);
302
303 // Ensure the capture directory exists before the detached process runs.
304 if let Some(parent) = stdout_file.parent() {
305 crate::workflow::ensure_devflow_dir(parent)?;
306 }
307
308 let stdout_file = stdout_file.to_str().ok_or(MonitorError::NonUtf8Path)?;
309 let stderr_file = stderr_file.to_str().ok_or(MonitorError::NonUtf8Path)?;
310 let exit_file = exit_file.to_str().ok_or(MonitorError::NonUtf8Path)?;
311 let pid_file = pid_file.to_str().ok_or(MonitorError::NonUtf8Path)?;
312
313 // The agent runs in its worktree when worktree mode is active; otherwise it
314 // runs in the project root. Capture/state files and the `devflow check`
315 // calls below always use the main project root, regardless of cwd.
316 let workdir_path = state
317 .worktree_path
318 .as_deref()
319 .unwrap_or(&state.project_root);
320 let workdir = workdir_path.to_str().ok_or(MonitorError::NonUtf8Path)?;
321
322 if let MonitorLaunch::PipeOwning { prompt } = launch {
323 // `run_advance` is not consulted on this arm: the `__monitor`
324 // subcommand always advances after reaping, and `spawn_monitor` is the
325 // only caller of this function — it hardcodes `true`. Adding a
326 // `--no-advance` flag for a case nothing exercises would be an
327 // untested branch; add it when a caller actually needs it.
328 let _ = run_advance;
329
330 // The adapter's extra env rides down by INHERITANCE here (set via
331 // `.envs(...)` on the `__monitor` process below), and that is only
332 // sufficient because the sole adapter routed through this arm —
333 // Claude — declares no extra env at all
334 // (`codex_disables_signing_via_env_others_do_not` asserts this).
335 // Widening this arm to an adapter that DOES set env requires
336 // threading it explicitly to `run_pipe_owning_monitor`: the inner
337 // `hermetic_command` scrubs `GIT_CONFIG_COUNT`, which neutralises any
338 // inherited `GIT_CONFIG_KEY_n` pair (Codex's unsigned-commit
339 // override is exactly that shape). Loud rather than silent, and in
340 // the CLI process where an operator can actually see it.
341 if !envs.is_empty() {
342 warn!(
343 "pipe-owning monitor: {} adapter env var(s) will not survive the \
344 inner hermetic_command scrub — thread them explicitly before \
345 routing an env-setting adapter through this arm",
346 envs.len()
347 );
348 }
349
350 // D-04: resolve and clamp the idle timeout HERE, in the parent, and
351 // hand the monitor the already-resolved integer.
352 //
353 // The placement is the whole point. `spawn_monitor` runs inside
354 // `devflow start`, attached to the operator's terminal; the monitor is
355 // a detached process whose stdio is all `Stdio::null()`, so a warning
356 // logged there scrolls into nothing. A silent clamp is the exact
357 // failure class this project keeps paying for, so the notice goes to
358 // BOTH `tracing::warn!` and stdout — the log for the record, stdout
359 // for the human who is watching right now.
360 let idle = idle_timeout_setting();
361 if let Some(notice) = idle.notice() {
362 warn!("{notice}");
363 println!("{notice}");
364 }
365
366 // The prompt travels as a FILE, not argv: argv has a hard length
367 // ceiling and DevFlow stage prompts routinely exceed what is safe to
368 // pass positionally.
369 let prompt_file = crate::agent_result::prompt_path(&state.project_root, state.phase);
370 std::fs::write(&prompt_file, &prompt)?;
371 let prompt_file = prompt_file.to_str().ok_or(MonitorError::NonUtf8Path)?;
372
373 // Re-exec THIS binary as its hidden `__monitor` subcommand. The
374 // monitor must outlive `devflow start`/`advance`, so it has to be a
375 // distinct OS process; re-exec needs no daemonization primitive beyond
376 // `spawn()`-without-`wait()`, which is exactly what the `sh` monitor
377 // below already relies on.
378 //
379 // Ordering is load-bearing for the same reason the Legacy arm's
380 // comment gives: `hermetic_command` does its `env_remove`s at
381 // construction and `.envs(...)` runs after, so deliberate
382 // configuration survives while inherited pollution does not.
383 let child = hermetic_command(&binary, workdir_path)
384 .arg("__monitor")
385 .arg("--project")
386 .arg(project_root)
387 .arg("--phase")
388 .arg(state.phase.to_string())
389 .arg("--workdir")
390 .arg(workdir)
391 .arg("--prompt-file")
392 .arg(prompt_file)
393 .arg("--idle-timeout-secs")
394 .arg(idle.timeout.as_secs().to_string())
395 .arg("--")
396 .arg(program)
397 .args(args)
398 .envs(envs.iter().map(|(k, v)| (k.as_str(), v.as_str())))
399 .stdin(Stdio::null())
400 .stdout(Stdio::null())
401 .stderr(Stdio::null())
402 .spawn()?;
403
404 let pid = child.id();
405 info!("pipe-owning monitor spawned with pid {pid}");
406 return Ok(pid);
407 }
408
409 // Shell script that launches the agent in the background, captures its
410 // stdout and exit code, then advances the workflow. Because this process
411 // is the agent's parent, capture survives the CLI exiting.
412 //
413 // stderr is captured to a separate file so it cannot corrupt the (possibly
414 // JSON) stdout capture that DevFlow parses for DEVFLOW_RESULT. Inspect
415 // .devflow/phase-NN-stderr.log for agent error output on failures.
416 //
417 // `devflow advance --phase N` evaluates the agent result, moves the stage
418 // machine forward, and (for an agent stage) spawns the next monitor
419 // itself. The phase is recorded here at spawn time so advance's identity
420 // never depends on a shared state singleton (13-DEFERRED-CR-03): under
421 // `devflow parallel`, each phase's monitor advances exactly its own
422 // stage machine.
423 //
424 // Traps SIGTERM and SIGINT for clean shutdown. WR-08 (13-REVIEW.md):
425 // the trap must also kill the backgrounded agent ($apid) — previously
426 // it only exited the monitor shell itself, orphaning the agent so it
427 // kept running/committing unsupervised with nothing left to call
428 // `devflow advance` once it finished. `apid` is initialized to empty
429 // before the trap is installed so a signal arriving before the agent is
430 // even backgrounded doesn't reference an unset variable.
431 let advance_tail = if run_advance {
432 format!(
433 "; {binary} advance {project_root} --phase {phase}",
434 binary = shell_escape(&binary),
435 project_root = shell_escape(project_root),
436 phase = state.phase,
437 )
438 } else {
439 String::new()
440 };
441 let script = format!(
442 "apid=''; cleanup() {{ [ -n \"$apid\" ] && kill \"$apid\" 2>/dev/null; exit 0; }}; \
443 trap cleanup TERM INT; \
444 cd {workdir} || exit 1; \
445 \"$@\" > {stdout_file} 2>{stderr_file} & \
446 apid=$!; echo $apid > {pid_file}; \
447 wait $apid; echo $? > {exit_file}{advance_tail}",
448 workdir = shell_escape(workdir),
449 stdout_file = shell_escape(stdout_file),
450 stderr_file = shell_escape(stderr_file),
451 exit_file = shell_escape(exit_file),
452 pid_file = shell_escape(pid_file),
453 );
454
455 // 27-REVIEW WR-03: built through `hermetic_command`, not a bare
456 // `Command::new("sh")`. This is the spawn that launches the coding agent
457 // itself, and the comment below is precisely the hazard: whatever
458 // environment this `sh` carries rides down into the agent and into every
459 // git command the agent runs. An inherited `GIT_DIR` here would silently
460 // retarget the phase's real commits at a repository the operator never
461 // named — the worst case this phase exists to prevent, on its
462 // highest-consequence call site.
463 //
464 // Ordering is load-bearing: `hermetic_command` does its `env_remove`s at
465 // construction, and `.envs(...)` below runs after, so an adapter that
466 // deliberately sets one of these variables still wins. Deliberate
467 // configuration survives; inherited pollution does not. That is what
468 // keeps Codex's unsigned-commit override (`GIT_CONFIG_*`) working.
469 let child = hermetic_command("sh", workdir_path)
470 .arg("-c")
471 .arg(&script)
472 .arg("sh")
473 .arg(program)
474 .args(args)
475 // Adapter-scoped env (e.g. Codex's unsigned-commit override) rides
476 // the whole monitor chain: sh → agent → its git children (13-06).
477 .envs(envs.iter().map(|(k, v)| (k.as_str(), v.as_str())))
478 .stdin(Stdio::null())
479 .stdout(Stdio::null())
480 .stderr(Stdio::null())
481 .spawn()?;
482
483 let pid = child.id();
484 info!("monitor spawned with pid {pid}");
485 Ok(pid)
486}
487
488/// Constraint 4's close rule as a pure, line-fed state machine: stdin may be
489/// released only once a `DEVFLOW_RESULT` marker has appeared inside a
490/// TOP-LEVEL `result` event **and** the background-task list has drained.
491///
492/// An `AND` of two arms, neither sufficient alone:
493///
494/// - **Marker arm.** Satisfied only by
495/// [`crate::agent_result::event_is_top_level_result_marker`] — a composition
496/// of the existing `is_top_level` predicate and the existing marker parser,
497/// never a looser text search. The CLI echoes the operator's prompt back
498/// into the same stdout, and DevFlow's own stage prompts discuss
499/// `DEVFLOW_RESULT` markers at length, so marker text alone is not evidence
500/// (T-31-01; the same echo produced the checkpoint false positive 30-05
501/// fixed).
502/// - **Drain arm.** Satisfied when no `background_tasks_changed` event has
503/// ever announced anything (vacuous — the common single-plan case) or when
504/// the most recent one carried an empty list.
505///
506/// **The drain alone is never a stop signal.** 30c/30d measured the
507/// drain-to-final-`result` lag at 4.54–11.51s across 14 trials; closing at the
508/// drain would have truncated the final orchestrator turn in all seven 30d
509/// trials.
510///
511/// **Never count `result` events.** Constraint 7: the CLI coalesces
512/// completions, so a wave whose children finish together produces one `result`
513/// for several of them — a shape superficially indistinguishable from "one
514/// child delivered, one lost". The drained list is the only thing separating
515/// those two. Per 30-04 the drain arm is *defensive rather than load-bearing*
516/// (n=2 Mode B trials delivered everything without it); that is the recorded
517/// reason to keep it cheaply, not a reason to drop it.
518///
519/// **A line that does not parse as JSON is ignored by this rule** — it can
520/// neither satisfy nor block either arm — but it is still teed verbatim to the
521/// capture file by the reader thread. A torn line therefore cannot silently
522/// decide anything, and cannot be silently lost either.
523/// The three states a `background_tasks_changed` announcement can leave the
524/// close rule in. Kept as a named enum rather than `Option<usize>` (999.75 /
525/// DEN-96, fixed 2026-08-04): a plain `Option` cannot distinguish "no
526/// announcement has ever arrived" from "an announcement arrived but its
527/// `tasks` field was not a readable array", and both used to collapse onto
528/// `None`. `should_close()` treats `None` as permission to close — so an
529/// unparseable *first* announcement closed stdin exactly when a background
530/// task was actually pending, which is the 999.64 orphan shape reachable
531/// through the guard built to prevent it.
532#[derive(Default, PartialEq, Eq, Debug, Clone, Copy)]
533enum BackgroundTaskState {
534 /// No `background_tasks_changed` event has been observed at all. Vacuously
535 /// drained: a stage that never backgrounds anything must still be able to
536 /// close on its marker, or every non-backgrounding stage would hang for
537 /// the full idle timeout.
538 #[default]
539 NeverAnnounced,
540 /// The last announcement carried a readable `tasks` array of this length.
541 /// `Pending(0)` is a real drain; `Pending(n>0)` blocks closing.
542 Pending(usize),
543 /// An announcement arrived — `type: "system"`,
544 /// `subtype: "background_tasks_changed"` — but its `tasks` field was not a
545 /// readable JSON array. Distinct from `NeverAnnounced` specifically so it
546 /// does NOT satisfy `should_close()`: the CLI said tasks might exist and
547 /// this rule could not read the count, so the safe assumption is that
548 /// something is still pending, not that nothing ever was.
549 Unreadable,
550}
551
552#[derive(Default)]
553pub struct CloseRule {
554 marker_seen: bool,
555 background_tasks: BackgroundTaskState,
556}
557
558impl CloseRule {
559 /// Fold one raw stdout line into the rule.
560 pub fn observe(&mut self, line: &str) {
561 let Ok(event) = serde_json::from_str::<serde_json::Value>(line) else {
562 return;
563 };
564 if crate::agent_result::event_is_top_level_result_marker(&event) {
565 self.marker_seen = true;
566 }
567 if event.get("type").and_then(serde_json::Value::as_str) == Some("system")
568 && event.get("subtype").and_then(serde_json::Value::as_str)
569 == Some("background_tasks_changed")
570 {
571 self.background_tasks = match event.get("tasks").and_then(serde_json::Value::as_array) {
572 Some(tasks) => BackgroundTaskState::Pending(tasks.len()),
573 // The announcement exists but its `tasks` field could not be
574 // read as an array. Distinct from "never announced" — see the
575 // enum doc comment. This is the fix: previously this arm did
576 // nothing, leaving `pending_background_tasks` at its prior
577 // value, which on the FIRST announcement was `None` —
578 // indistinguishable from vacuous drain, and so treated as
579 // permission to close exactly when it should not have been.
580 None => BackgroundTaskState::Unreadable,
581 };
582 }
583 }
584
585 /// Whether both arms hold and the child's stdin may be released.
586 pub fn should_close(&self) -> bool {
587 self.marker_seen
588 && matches!(
589 self.background_tasks,
590 BackgroundTaskState::NeverAnnounced | BackgroundTaskState::Pending(0)
591 )
592 }
593}
594
595/// The single place the stdin wire shape is constructed: one line of JSON
596/// carrying the initial user turn for a `--input-format stream-json` child.
597///
598/// Shape (`{"type":"user","message":{"role":"user","content":<prompt>}}`) is
599/// reproduced from the three archived Phase 30 harnesses, which all wrote
600/// exactly this and got a working turn back.
601///
602/// Built with `serde_json` rather than `format!` so the prompt is ESCAPED, not
603/// interpolated. A stage prompt is arbitrary text containing quotes, newlines
604/// and backslashes; interpolating it would produce a torn JSON line the CLI
605/// rejects, and a prompt could then alter the surrounding document's structure.
606pub fn user_turn_line(prompt: &str) -> String {
607 serde_json::json!({
608 "type": "user",
609 "message": { "role": "user", "content": prompt },
610 })
611 .to_string()
612}
613
614/// Supervise a `stream-json` child, owning both of its pipes, until the close
615/// rule is satisfied and the child exits. Returns the child's exit code, which
616/// is also written to the phase exit file.
617///
618/// This runs INSIDE the detached `__monitor` process, not in the CLI.
619///
620/// Threading model (constraint 4 / T-31-04). Three participants:
621/// - a **writer thread** owning the child's stdin: it writes the initial user
622/// turn, then BLOCKS on a channel rather than returning. It drops stdin only
623/// when told to, because constraint 4's `AND` can never be honoured if stdin
624/// is already gone — a task-notification turn arriving after the child's
625/// first turn would have nowhere to be delivered.
626/// - a **reader thread** owning the child's stdout: it tees each line verbatim
627/// to the capture file and forwards it to the supervisor. Dropping its
628/// sender at EOF is what surfaces `Disconnected` below.
629/// - the **supervisor** (this function's own thread), which applies the close
630/// rule and reaps.
631///
632/// The write and the read MUST be on independent threads. Writing the prompt
633/// synchronously before reading stdout is the textbook two-pipe deadlock: it
634/// passes every short-prompt smoke test and hangs on exactly the context-heavy
635/// production stages that matter (the Linux pipe buffer is commonly 64KiB and
636/// a DevFlow stage prompt can exceed that in one write).
637#[allow(clippy::too_many_arguments)]
638pub fn run_pipe_owning_monitor(
639 project_root: &Path,
640 phase: u32,
641 workdir: &Path,
642 prompt: &str,
643 idle_timeout: Duration,
644 program: &str,
645 args: &[String],
646 envs: &[(String, String)],
647) -> Result<i32, MonitorError> {
648 let stdout_file = crate::agent_result::stdout_path(project_root, phase);
649 let stderr_file = crate::agent_result::stderr_path(project_root, phase);
650 let exit_file = crate::agent_result::exit_code_path(project_root, phase);
651 let pid_file = crate::agent_result::agent_pid_path(project_root, phase);
652 if let Some(parent) = stdout_file.parent() {
653 crate::workflow::ensure_devflow_dir(parent)?;
654 }
655
656 // stderr goes to its own file so it cannot corrupt the JSONL stdout
657 // capture DevFlow parses — the same separation the Legacy script's
658 // `2>{stderr_file}` provides.
659 let stderr_handle = std::fs::File::create(&stderr_file)?;
660 // One handle, opened once, truncating at open and appending line by line.
661 // Truncate-at-open reproduces the Legacy arm's `>` redirection exactly, so
662 // a capture from a previous attempt can never be mixed into this one's
663 // (the launch path archives the prior capture first, but relying on that
664 // to make an append-mode open safe would be an unstated coupling).
665 let mut capture = std::fs::File::create(&stdout_file)?;
666
667 let mut child = hermetic_command(program, workdir)
668 .args(args)
669 .envs(envs.iter().map(|(k, v)| (k.as_str(), v.as_str())))
670 .stdin(Stdio::piped())
671 .stdout(Stdio::piped())
672 .stderr(Stdio::from(stderr_handle))
673 // T-31-05: make the child its own process-group leader so a later
674 // group signal cannot reach this monitor's own ancestors. Verified
675 // source shows the pre-31 `spawn_monitor` had NO session or group
676 // configuration at all — detachment came only from the parent not
677 // waiting — so this closes a gap rather than preserving one.
678 // Full `setsid()` session detachment is deliberately NOT done: no
679 // forensics record cites a SIGHUP-related monitor loss, so there is
680 // no evidence it buys anything. `pre_exec` calling `libc::setsid()`
681 // is the one-line follow-on if such a loss ever surfaces.
682 .process_group(0)
683 .spawn()?;
684
685 // Recorded immediately, before any pipe work: `wait_for_agent_pid` polls
686 // for this and the rest of DevFlow's liveness reporting depends on it.
687 let child_pid = child.id();
688 std::fs::write(&pid_file, format!("{child_pid}\n"))?;
689
690 let mut child_stdin = child
691 .stdin
692 .take()
693 .ok_or(MonitorError::NoChildPipe("stdin"))?;
694 let child_stdout = child
695 .stdout
696 .take()
697 .ok_or(MonitorError::NoChildPipe("stdout"))?;
698
699 let (close_tx, close_rx) = mpsc::channel::<()>();
700 let turn = user_turn_line(prompt);
701 let writer = std::thread::spawn(move || {
702 let wrote = child_stdin
703 .write_all(turn.as_bytes())
704 .and_then(|()| child_stdin.write_all(b"\n"))
705 .and_then(|()| child_stdin.flush());
706 if let Err(err) = wrote {
707 warn!("could not write the initial user turn to the child's stdin: {err}");
708 return;
709 }
710 // Deliberately NOT dropping stdin here — see this function's doc.
711 // Either signal (an explicit close, or the supervisor dropping its
712 // sender) means the same thing: stop holding the pipe open.
713 let _ = close_rx.recv();
714 drop(child_stdin);
715 });
716
717 let (line_tx, line_rx) = mpsc::channel::<String>();
718 let reader = std::thread::spawn(move || {
719 // `read_until` + `from_utf8_lossy`, NOT `BufRead::lines()` (peer review
720 // 2026-08-03, CRITICAL). `lines()` yields `Err(InvalidData)` on a single
721 // non-UTF-8 byte, and the previous code treated any read error as EOF —
722 // so one bad byte silently truncated the capture and dropped every later
723 // line INCLUDING the terminal `DEVFLOW_RESULT` marker. That is precisely
724 // the boundary-truncation class constraint 9 exists for, manufactured by
725 // the supervisor itself rather than by a dying writer.
726 //
727 // Decoding is now lossy and NON-fatal: undecodable bytes become U+FFFD
728 // and the line still reaches the capture and the close rule. A genuine
729 // I/O error still ends the loop, because that one really is EOF.
730 let mut reader_buf = BufReader::new(child_stdout);
731 let mut raw = Vec::new();
732 loop {
733 raw.clear();
734 match reader_buf.read_until(b'\n', &mut raw) {
735 Ok(0) => break, // real EOF
736 Ok(_) => {}
737 Err(err) => {
738 warn!("stdout read error, treating as EOF: {err}");
739 break;
740 }
741 }
742 while raw.last().is_some_and(|b| *b == b'\n' || *b == b'\r') {
743 raw.pop();
744 }
745 let line = String::from_utf8_lossy(&raw).into_owned();
746 // Tee VERBATIM before any interpretation: the whole Layer 1
747 // cascade reads this file, and a line the close rule ignores
748 // (unparseable noise, interleaved prose) must still reach it.
749 if let Err(err) = writeln!(capture, "{line}") {
750 warn!("could not append to the capture file: {err}");
751 }
752 let _ = capture.flush();
753 if line_tx.send(line).is_err() {
754 break;
755 }
756 }
757 // Dropping `line_tx` here is what surfaces `Disconnected` below.
758 });
759
760 // Constraint 4's close rule lives in `CloseRule` so it can be unit-tested
761 // by feeding it lines, with no child process per case.
762 let mut rule = CloseRule::default();
763 let mut close_signalled = false;
764
765 loop {
766 match line_rx.recv_timeout(idle_timeout) {
767 Ok(line) => {
768 if close_signalled {
769 continue;
770 }
771 rule.observe(&line);
772 if rule.should_close() {
773 let _ = close_tx.send(());
774 close_signalled = true;
775 }
776 }
777 Err(mpsc::RecvTimeoutError::Disconnected) => break,
778 Err(mpsc::RecvTimeoutError::Timeout) => {
779 // AFTER a deliberate close, silence is EXPECTED, not a hang
780 // (peer review 2026-08-03, CRITICAL). The close rule fires only
781 // once the agent has emitted its terminal marker AND background
782 // tasks have drained — at which point it has said everything it
783 // intends to say and is merely winding down. Firing the idle
784 // timeout here wrote an authoritative `IdleTimeout` verdict OVER
785 // a completed, successful stage; and because `evaluate_layer1`
786 // reads that side channel FIRST, by design, so that nothing can
787 // shadow a real timeout, the bogus verdict outranked the real
788 // success and could not be recovered from. The mechanism that
789 // protects a true timeout is what made a false one fatal.
790 //
791 // Break instead: the reap path below already bounds a child that
792 // will not exit, via `terminate_and_verify`.
793 if close_signalled {
794 info!(
795 "no output for {idle_timeout:?} after the close rule released stdin; \
796 the stage already reported — proceeding to reap, NOT recording a timeout"
797 );
798 break;
799 }
800 // No outer wall-clock bound exists anywhere in this loop, and
801 // none may be added (D-03). `recv_timeout` measures the gap
802 // since the LAST LINE, so a healthy 47-minute stage that keeps
803 // emitting is never touched — every line the reader thread
804 // forwards resets the window naturally, which is D-01's
805 // every-line signal rather than a milestone-only one. There is
806 // no single wall-clock value that is safe for both a hang and
807 // a legitimately long stage, which is why constraint 5
808 // rejected one.
809 fire_idle_timeout(project_root, phase, workdir, child_pid, idle_timeout);
810 break;
811 }
812 }
813 }
814
815 // Guarantee stdin is released before waiting. A child still holding an
816 // open stdin may never exit, and `child.wait()` would then block forever.
817 drop(close_tx);
818
819 let status = child.wait()?;
820 // A signal-killed child has NO exit code — `status.code()` is `None`, and
821 // the previous `unwrap_or(-1)` threw the signal away (peer review
822 // 2026-08-03, found independently by both reviewers and by the 31-04 plan
823 // review as W1). That silently defeated the classification 31-04 took care
824 // to preserve: `evaluate_layer2` and
825 // `reconcile_stream_success_against_exit_code` map **137** to
826 // `ResourceKilled` (routed to `GateInfra` — an infrastructure fault) and
827 // **127** to `AgentUnavailable`. Recording `-1` matched neither, so a real
828 // OOM kill arrived as a generic `Failed` and routed to `GateReview`, asking
829 // an operator to code-review a stage that was killed by the kernel.
830 //
831 // `128 + signal` is the shell convention those constants already encode:
832 // SIGKILL(9) -> 137, SIGTERM(15) -> 143. `-1` is now reachable only when a
833 // status is neither exited nor signalled, which POSIX does not define.
834 let code = status.code().unwrap_or_else(|| {
835 use std::os::unix::process::ExitStatusExt;
836 status.signal().map_or(-1, |signal| 128 + signal)
837 });
838 std::fs::write(&exit_file, format!("{code}\n"))?;
839
840 let _ = writer.join();
841 let _ = reader.join();
842
843 info!("supervised child {child_pid} exited with code {code}");
844 Ok(code)
845}
846
847/// The idle-timeout firing sequence, in the ONE order it may run (D-05).
848///
849/// 1. Enumerate the commits the agent made.
850/// 2. Write the authoritative verdict to its side-channel file, and fsync it.
851/// 3. **Only then** terminate the child.
852/// 4. Append a loud entry to the monitor's own log.
853///
854/// Step 3 must not precede step 2, and reversing them is not a stylistic
855/// choice. Between "the child is dead" and "an authoritative result exists"
856/// there is a window in which the verdict cascade sees a dead process, no
857/// Layer-1 answer, and some commits on the branch — and Layer 2 scores exactly
858/// that as `Success`. That is 999.64 reborn inside its own fix. A bare kill
859/// with no record is the other half of the same failure: exit code 137 reads
860/// as `ResourceKilled`, blaming an OOM that never happened.
861///
862/// **Nothing here rolls back, resets, or reverts a commit** (D-07, T-31-09).
863/// The commit log is READ and never written. A timeout can be a false
864/// positive, and destroying real work on a false positive is unrecoverable —
865/// this repo treats irreversible operations as needing review, not tests.
866///
867/// Scoped to the `PipeOwning` arm alone: `Legacy` keeps today's behaviour, and
868/// Codex/OpenCode keep theirs. The 120-second floor was measured against
869/// Claude's stream cadence (a fixed 30.00s `tool_progress` keepalive), and
870/// applying it to an agent whose output cadence has never been measured would
871/// be a behaviour prediction — the thing constraint 1 forbids.
872///
873/// Every step is best-effort and none can abort the sequence. A failure to
874/// enumerate, write, or log must still leave the child terminated and the
875/// stage machine advancing to a never-silent gate; the operator loses detail,
876/// never the verdict.
877fn fire_idle_timeout(
878 project_root: &Path,
879 phase: u32,
880 workdir: &Path,
881 child_pid: u32,
882 idle: Duration,
883) {
884 let idle_secs = idle.as_secs();
885 warn!("idle timeout: no output from the supervised child for {idle_secs}s");
886
887 // 1. Enumerate. A failure degrades to an empty list plus a note; it never
888 // aborts, because a missing commit list must not cost the verdict.
889 let (commits, enumeration_note) = enumerate_phase_commits(workdir, phase);
890
891 // 2. Write, flush, fsync. This completing is the ONLY thing that stops
892 // Layer 2 from later scoring partial commits as Success.
893 let write_error =
894 write_idle_timeout_record(project_root, phase, idle_secs, child_pid, &commits)
895 .err()
896 .map(|err| err.to_string());
897 if let Some(err) = &write_error {
898 warn!("idle timeout: could not persist the verdict: {err}");
899 }
900
901 // 3. Only now is it safe to kill.
902 let terminated = terminate_child_group(child_pid);
903
904 // 4. Loud, durable, and readable after the fact.
905 let named: Vec<String> = commits
906 .iter()
907 .map(|commit| {
908 let short: String = commit.sha.chars().take(7).collect();
909 format!("{short} {}", commit.subject)
910 })
911 .collect();
912 let mut entry = format!(
913 "[idle-timeout] no output for {idle_secs}s; terminated agent pid {child_pid} \
914 (verified dead: {terminated}). {} commit(s) on the phase branch, NONE rolled back{}{}",
915 named.len(),
916 if named.is_empty() {
917 String::new()
918 } else {
919 format!(": {}", named.join("; "))
920 },
921 enumeration_note
922 .map(|note| format!(" [commit enumeration degraded: {note}]"))
923 .unwrap_or_default(),
924 );
925 if let Some(err) = write_error {
926 entry.push_str(&format!(" [verdict file could not be written: {err}]"));
927 }
928 warn!("{entry}");
929 append_monitor_log(project_root, phase, &entry);
930}
931
932/// Enumerate the commits on this phase's feature branch, as
933/// `(commits, degradation note)`.
934///
935/// Same range construction `evaluate_layer2`'s commit COUNT uses
936/// (`{develop}..{feature_prefix}phase-NN`) — the same question asked with
937/// `git log` instead of `rev-list --count`, so the two can never disagree
938/// about which commits are the agent's.
939///
940/// Never returns an error. Every failure path yields an empty list and a note
941/// naming what went wrong: the operator losing the commit NAMES is bad, the
942/// operator losing the VERDICT is the failure this whole plan exists to
943/// prevent.
944fn enumerate_phase_commits(workdir: &Path, phase: u32) -> (Vec<IdleTimeoutCommit>, Option<String>) {
945 let git_flow = crate::config::GitFlowConfig::default();
946 let branch = format!("{}phase-{:02}", git_flow.feature_prefix, phase);
947 let range = format!("{}..{branch}", git_flow.develop);
948
949 let output = match crate::git::git_command(workdir)
950 .args(["log", "--format=%H %s", &range])
951 .output()
952 {
953 Ok(output) => output,
954 Err(err) => return (Vec::new(), Some(format!("git log could not run: {err}"))),
955 };
956
957 if !output.status.success() {
958 let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
959 return (
960 Vec::new(),
961 Some(format!("git log {range} failed: {stderr}")),
962 );
963 }
964
965 let commits = String::from_utf8_lossy(&output.stdout)
966 .lines()
967 .filter_map(|line| {
968 let line = line.trim();
969 if line.is_empty() {
970 return None;
971 }
972 // `%H %s` — a sha, one space, then the subject, which may itself
973 // contain spaces. `split_once` is therefore correct and `split`
974 // is not. A subject-less commit still yields an empty subject
975 // rather than being dropped.
976 let (sha, subject) = line.split_once(' ').unwrap_or((line, ""));
977 Some(IdleTimeoutCommit {
978 sha: sha.to_string(),
979 subject: subject.to_string(),
980 })
981 })
982 .collect();
983
984 (commits, None)
985}
986
987/// Write the idle-timeout verdict and get it onto the platter before returning.
988///
989/// `sync_all` is not decoration: D-05's guarantee is that the result exists
990/// before anything can race it, and a buffered write that is still in the page
991/// cache when the process is signalled has not achieved that.
992fn write_idle_timeout_record(
993 project_root: &Path,
994 phase: u32,
995 idle_secs: u64,
996 child_pid: u32,
997 commits: &[IdleTimeoutCommit],
998) -> std::io::Result<()> {
999 let record = IdleTimeoutRecord {
1000 status: crate::agent_result::AgentStatus::IdleTimeout
1001 .as_wire_str()
1002 .to_string(),
1003 idle_secs,
1004 agent_pid: child_pid,
1005 written_at: std::time::SystemTime::now()
1006 .duration_since(std::time::UNIX_EPOCH)
1007 .map(|d| d.as_secs())
1008 .unwrap_or(0),
1009 commits: commits.to_vec(),
1010 };
1011 let json = serde_json::to_string(&record)
1012 .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
1013
1014 let path = crate::agent_result::idle_timeout_path(project_root, phase);
1015 if let Some(parent) = path.parent() {
1016 crate::workflow::ensure_devflow_dir(parent)?;
1017 }
1018 let mut file = std::fs::File::create(&path)?;
1019 file.write_all(json.as_bytes())?;
1020 file.flush()?;
1021 file.sync_all()
1022}
1023
1024/// Terminate the supervised child's whole process group, returning the
1025/// VERIFIED fact of whether the leader is dead.
1026///
1027/// Acts on `child_pid`, which came from the in-memory `Child` handle — never
1028/// on the on-disk pid file (T-31-07). That distinction is what makes the
1029/// negative-pid signal below safe at all: while this monitor still holds the
1030/// unwaited `Child`, the kernel cannot recycle that pid, so it cannot come to
1031/// mean some unrelated process between spawn and now. A pid re-read from disk
1032/// carries no such guarantee.
1033///
1034/// Three steps, and the middle one is borrowed whole rather than reimplemented:
1035///
1036/// 1. `SIGTERM` to the GROUP. `.process_group(0)` at spawn made the child its
1037/// own group leader, so its pid IS its pgid and `-pid` reaches its whole
1038/// subtree — the tool subprocesses a coding agent leaves behind, which a
1039/// leader-only signal would orphan. It cannot reach this monitor: the
1040/// monitor stayed in its own inherited group, which is precisely what
1041/// `.process_group(0)` bought (T-31-05).
1042/// 2. [`crate::agent::terminate_and_verify`] for the leader — reused, not
1043/// rewritten. It owns the `SIGTERM` → poll → `SIGKILL` → re-poll
1044/// escalation and returns a verified liveness fact instead of an
1045/// assumption. 999.44 measured 15 of 15 orphaned wrappers surviving
1046/// `SIGTERM`, so the escalation is not optional.
1047/// 3. `SIGKILL` to the group, sweeping any survivor the leader's own
1048/// escalation did not cover. Unconditional by design: at this point the run
1049/// is over, everything in the group is the agent's subtree, and a `kill` to
1050/// an empty group is a no-op `ESRCH`.
1051///
1052/// The `signed > 1` guard is load-bearing twice over. `kill(-1, sig)` signals
1053/// every process the caller may signal, and `kill(0, sig)` signals the
1054/// caller's own group — the two catastrophic cases `agent::terminate` already
1055/// documents, reachable here through the negation rather than through a
1056/// hostile pid file.
1057fn terminate_child_group(child_pid: u32) -> bool {
1058 let Ok(signed) = libc::pid_t::try_from(child_pid) else {
1059 warn!("idle timeout: child pid {child_pid} does not fit pid_t; not signalling");
1060 return false;
1061 };
1062 if signed <= 1 {
1063 warn!("idle timeout: refusing to signal group for pid {signed}");
1064 return false;
1065 }
1066
1067 // SAFETY: `signed > 1`, so `-signed < -1` and the two catastrophic
1068 // targets (`0` = our own group, `-1` = everything) are both excluded.
1069 unsafe {
1070 libc::kill(-signed, libc::SIGTERM);
1071 }
1072
1073 let dead = crate::agent::terminate_and_verify(
1074 child_pid,
1075 crate::agent::TERMINATE_VERIFY_WAIT,
1076 crate::agent::TERMINATE_VERIFY_POLL,
1077 );
1078
1079 // SAFETY: same guard as above.
1080 unsafe {
1081 libc::kill(-signed, libc::SIGKILL);
1082 }
1083
1084 dead
1085}
1086
1087/// Append one line to the monitor's own log, creating it if needed.
1088///
1089/// Best-effort: the monitor's stdio is null, so this file is the only place a
1090/// "log loudly" obligation can actually land, but failing to write it must
1091/// never abort a termination sequence already in progress.
1092fn append_monitor_log(project_root: &Path, phase: u32, entry: &str) {
1093 let path = crate::agent_result::monitor_log_path(project_root, phase);
1094 if let Ok(mut file) = std::fs::OpenOptions::new()
1095 .create(true)
1096 .append(true)
1097 .open(&path)
1098 {
1099 let _ = writeln!(file, "{entry}");
1100 }
1101}
1102
1103/// Poll for the agent PID that the monitor records, for up to ~1 second.
1104///
1105/// Returns the PID once the monitor has launched the agent, or `None` if it
1106/// does not appear in time (the monitor still runs; only the display PID is lost).
1107pub fn wait_for_agent_pid(project_root: &Path, phase: u32) -> Option<u32> {
1108 let path = crate::agent_result::agent_pid_path(project_root, phase);
1109 debug!("polling for agent PID for phase {phase}");
1110 for _ in 0..50 {
1111 if let Ok(contents) = std::fs::read_to_string(&path)
1112 && let Ok(pid) = contents.trim().parse::<u32>()
1113 {
1114 return Some(pid);
1115 }
1116 std::thread::sleep(Duration::from_millis(20));
1117 }
1118 debug!("agent PID not found for phase {phase} after polling");
1119 None
1120}
1121
1122/// Escape a string for safe use in a single-quoted shell context.
1123fn shell_escape(s: &str) -> String {
1124 format!("'{}'", s.replace('\'', "'\\''"))
1125}
1126
1127#[cfg(test)]
1128mod tests {
1129 use super::*;
1130 use crate::mode::Mode;
1131 use crate::stage::Stage;
1132 use crate::state::{AgentKind, State};
1133
1134 fn state_in(root: &Path) -> State {
1135 let mut state = State::new(4, AgentKind::Claude, Mode::Auto, root.to_path_buf());
1136 state.stage = Stage::Code;
1137 state
1138 }
1139
1140 // ---- close-rule fixtures ------------------------------------------
1141 //
1142 // Key names, nesting and event types are taken from the real archived
1143 // capture at
1144 // `.planning/phases/30-keep-the-session-alive-past-turn-end/30a-evidence/raw_output_v3.jsonl`
1145 // (lines 5, 8, 19, 44 and 54), not invented: `tasks` is an array of
1146 // objects with `task_id`/`task_type`/`description`, the drained event is
1147 // the same event with `tasks":[]`, and a coalesced completion carries
1148 // `origin.kind == "task-notification"` on an ordinary `result`. Volumes
1149 // and identifiers are generalized; shapes are not.
1150
1151 const INIT_LINE: &str = r#"{"type":"system","subtype":"init","cwd":"/tmp/work","session_id":"s-1","tools":["Task","Bash"],"uuid":"u-init"}"#;
1152
1153 /// A `system`/`background_tasks_changed` event announcing `count` tasks.
1154 /// `count == 0` is the DRAINED shape (v3 line 44).
1155 fn bg_tasks_line(count: usize) -> String {
1156 let tasks: Vec<String> = (0..count)
1157 .map(|i| {
1158 format!(
1159 r#"{{"task_id":"t{i}","task_type":"local_agent","description":"child {i}"}}"#
1160 )
1161 })
1162 .collect();
1163 format!(
1164 r#"{{"type":"system","subtype":"background_tasks_changed","tasks":[{}],"uuid":"u-bg{count}","session_id":"s-1"}}"#,
1165 tasks.join(",")
1166 )
1167 }
1168
1169 /// A top-level `result` event. `marker` is the `result` field's text —
1170 /// the agent's own final message, where a `DEVFLOW_RESULT:` line lives.
1171 fn result_line(marker: &str) -> String {
1172 format!(
1173 r#"{{"type":"result","subtype":"success","is_error":false,"num_turns":3,"stop_reason":"end_turn","session_id":"s-1","uuid":"u-res","result":"{marker}"}}"#
1174 )
1175 }
1176
1177 /// The v3 line-54 shape: ONE `result` closing out work that several
1178 /// children contributed to, tagged with the task-notification origin.
1179 fn coalesced_result_line(marker: &str) -> String {
1180 format!(
1181 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}"}}"#
1182 )
1183 }
1184
1185 /// Same envelope, forwarded from a subagent rather than authored by the
1186 /// orchestrator session.
1187 fn subagent_result_line(marker: &str) -> String {
1188 result_line(marker).replacen('{', r#"{"parent_tool_use_id":"toolu_child","#, 1)
1189 }
1190
1191 /// A success marker as it appears INSIDE a `result` string field — the
1192 /// quotes are escaped because the field is itself JSON.
1193 const MARKER: &str = r#"All done.\nDEVFLOW_RESULT: {\"status\":\"success\",\"commits\":3}"#;
1194 const NO_MARKER: &str = "Acknowledged; nothing to report.";
1195
1196 fn observe_all(lines: &[String]) -> CloseRule {
1197 let mut rule = CloseRule::default();
1198 for line in lines {
1199 rule.observe(line);
1200 }
1201 rule
1202 }
1203
1204 /// Constraint 4 is an `AND`, and neither arm is sufficient alone. Both
1205 /// halves are asserted here because a rule that accidentally became an
1206 /// `OR` still passes any test that only ever feeds it both.
1207 #[test]
1208 fn close_rule_requires_both_marker_and_drained_background_tasks() {
1209 // Arm A: the drain lands, but no marker ever appears in a top-level
1210 // result. Closing here truncates the run before its verdict exists.
1211 // The torn line carrying marker TEXT is the negative control: a line
1212 // that does not parse as JSON must not be able to satisfy the marker
1213 // arm through the back door.
1214 let drained_but_unmarked = observe_all(&[
1215 INIT_LINE.to_string(),
1216 bg_tasks_line(1),
1217 bg_tasks_line(0),
1218 r#"{"type":"result","result":"DEVFLOW_RESULT: {\"status\":\"succ"#.to_string(),
1219 "progress: still working".to_string(),
1220 result_line(NO_MARKER),
1221 ]);
1222 assert!(
1223 !drained_but_unmarked.should_close(),
1224 "the drain alone must never close stdin: 30c/30d measured the \
1225 drain-to-final-result lag at 4.54-11.51s across 14 trials, and \
1226 closing at the drain would have truncated the final orchestrator \
1227 turn in all seven 30d trials"
1228 );
1229
1230 // Arm B: the marker lands while a child is still pending.
1231 let marked_but_pending =
1232 observe_all(&[INIT_LINE.to_string(), bg_tasks_line(1), result_line(MARKER)]);
1233 assert!(
1234 !marked_but_pending.should_close(),
1235 "a marker while a background task is still announced must not \
1236 close stdin — the pending child's task-notification turn would \
1237 have nowhere to be delivered"
1238 );
1239 }
1240
1241 /// 999.75 / DEN-96, fixed 2026-08-04. The FIRST `background_tasks_changed`
1242 /// announcement carries an unparseable `tasks` field (`null`, not an
1243 /// array). Before the fix, an unreadable announcement left the field at
1244 /// its prior value — which on the first announcement was the same `None`
1245 /// used for "nothing was ever announced", so `should_close()` treated it
1246 /// as a vacuous drain and closed stdin with a task genuinely pending. This
1247 /// is the 999.64 orphan shape, reachable through the guard built to
1248 /// prevent it.
1249 #[test]
1250 fn unreadable_first_announcement_does_not_satisfy_the_drain_arm() {
1251 let unreadable_first = observe_all(&[
1252 INIT_LINE.to_string(),
1253 r#"{"type":"system","subtype":"background_tasks_changed","tasks":null}"#.to_string(),
1254 result_line(MARKER),
1255 ]);
1256 assert!(
1257 !unreadable_first.should_close(),
1258 "an unreadable FIRST announcement must not be indistinguishable \
1259 from never-announced — closing here would release stdin while \
1260 the CLI has said a task exists whose count could not be read"
1261 );
1262
1263 // Negative control: the identical sequence, but with NO announcement
1264 // at all, must still close — this is the ordinary non-backgrounding
1265 // stage, and the fix must not regress it into hanging for the idle
1266 // timeout on every run that never backgrounds anything.
1267 let never_announced = observe_all(&[INIT_LINE.to_string(), result_line(MARKER)]);
1268 assert!(
1269 never_announced.should_close(),
1270 "a stage that never announces background tasks at all must still \
1271 close on its marker alone — conflating NeverAnnounced with \
1272 Unreadable would hang every ordinary stage for the full idle \
1273 timeout"
1274 );
1275
1276 // A LATER unreadable announcement, after a real pending count was
1277 // already known, must also block — the fix must not accidentally
1278 // treat Unreadable as forgiving once real state exists.
1279 let unreadable_after_pending = observe_all(&[
1280 INIT_LINE.to_string(),
1281 bg_tasks_line(1),
1282 r#"{"type":"system","subtype":"background_tasks_changed","tasks":"not-an-array"}"#
1283 .to_string(),
1284 result_line(MARKER),
1285 ]);
1286 assert!(
1287 !unreadable_after_pending.should_close(),
1288 "an unreadable announcement following a real pending count must \
1289 still block closing, not silently forget the pending task"
1290 );
1291 }
1292
1293 /// The common case: a single-plan stage that never dispatches anything.
1294 /// The drain arm is satisfied VACUOUSLY, because nothing was ever
1295 /// announced — an implementation that waited for a literal empty-list
1296 /// event would hang every such stage until the idle timeout.
1297 ///
1298 /// The interleaved noise lines also pin the other half of the rule's
1299 /// tolerance: a torn JSON line and a prose line are ignored for the rule
1300 /// (they can neither satisfy nor block it) while still being teed to the
1301 /// capture by the reader thread.
1302 #[test]
1303 fn close_rule_is_vacuously_drained_when_no_background_tasks_event_appears() {
1304 let rule = observe_all(&[
1305 INIT_LINE.to_string(),
1306 "starting up".to_string(),
1307 r#"{"type":"assist"#.to_string(),
1308 result_line(MARKER),
1309 ]);
1310 assert!(
1311 rule.should_close(),
1312 "a stage that never announced a background task is drained by \
1313 definition; only the marker arm has anything to satisfy"
1314 );
1315 }
1316
1317 /// Constraint 7. The CLI COALESCES completions: two children can finish
1318 /// into one `result` event, and two announced tasks can drain to an empty
1319 /// list in a single `background_tasks_changed`. Counting `result` events
1320 /// therefore silently undercounts any wave whose completions cluster —
1321 /// and that shape is superficially indistinguishable from "one child
1322 /// delivered, one lost". The drained list is the only thing separating
1323 /// them, so the rule asserts on the list state and never on a count.
1324 ///
1325 /// Per 30-04 the drain arm is DEFENSIVE rather than load-bearing: n=2
1326 /// Mode B trials delivered everything without it. That is the documented
1327 /// reason to keep it cheaply — "defensive" is not "removable".
1328 #[test]
1329 fn coalesced_completions_do_not_undercount_children() {
1330 let rule = observe_all(&[
1331 INIT_LINE.to_string(),
1332 bg_tasks_line(2),
1333 // BOTH children drain in ONE event...
1334 bg_tasks_line(0),
1335 // ...and complete into ONE result.
1336 coalesced_result_line(MARKER),
1337 ]);
1338 assert!(
1339 rule.should_close(),
1340 "two announced children, one drain event and one coalesced result \
1341 must still close — a rule that matched result events against \
1342 child count would stall here forever"
1343 );
1344
1345 // Negative control: the SAME single coalesced result with the drain
1346 // withheld must NOT close. Without this, the assertion above is also
1347 // satisfied by a rule that simply closes on any result event, and the
1348 // test would be measuring nothing.
1349 let undrained = observe_all(&[
1350 INIT_LINE.to_string(),
1351 bg_tasks_line(2),
1352 coalesced_result_line(MARKER),
1353 ]);
1354 assert!(
1355 !undrained.should_close(),
1356 "control: it is the drained list that decides, not the arrival of \
1357 a result event"
1358 );
1359 }
1360
1361 /// T-31-01. The CLI echoes the operator's prompt back into the same
1362 /// stdout, and DevFlow's own stage prompts discuss `DEVFLOW_RESULT`
1363 /// markers at length — so marker TEXT is not evidence of a verdict. Only
1364 /// a marker inside an event that is both `type: "result"` and top-level
1365 /// counts, reusing the one provenance predicate rather than inventing a
1366 /// second notion of trustworthiness.
1367 #[test]
1368 fn marker_inside_a_non_top_level_result_does_not_satisfy_the_close_rule() {
1369 let subagent = observe_all(&[INIT_LINE.to_string(), subagent_result_line(MARKER)]);
1370 assert!(
1371 !subagent.should_close(),
1372 "a subagent-origin result carrying a marker must not close the \
1373 stream — same provenance hole constraint 9 item 2 closed for the \
1374 stage verdict"
1375 );
1376
1377 // Control: the identical envelope WITHOUT the planted parent id is
1378 // top-level and legitimately closes. Without this the assertion above
1379 // would also pass against a rule that never closes at all.
1380 let top_level = observe_all(&[INIT_LINE.to_string(), result_line(MARKER)]);
1381 assert!(
1382 top_level.should_close(),
1383 "control: the same event without a parent id is authoritative"
1384 );
1385 }
1386
1387 #[test]
1388 fn shell_escape_wraps_basic_strings() {
1389 assert_eq!(shell_escape("hello"), "'hello'");
1390 assert_eq!(shell_escape("hello world"), "'hello world'");
1391 assert_eq!(shell_escape("/tmp/devflow"), "'/tmp/devflow'");
1392 }
1393
1394 /// The Phase 31 tracer: ONE Claude-shaped stage driven end to end through
1395 /// the pipe-owning supervisor.
1396 ///
1397 /// The stub behaves like the real CLI on the two axes under test and no
1398 /// others: it takes its initial turn from stdin, and it keeps stdin open
1399 /// as a channel it can still be spoken to on. It is a `sh` script because
1400 /// the wire behaviour is the subject, not the binary.
1401 ///
1402 /// **The early-close negative control is the point of the probe files.**
1403 /// A stub that merely blocks on stdin EOF before exiting cannot fail:
1404 /// whether the monitor closes stdin immediately after the write or only
1405 /// after the close rule is satisfied, the stub still eventually sees EOF
1406 /// and still exits 0. So the stub instead SAMPLES stdin liveness at a
1407 /// moment when a correct monitor provably has not closed it — after the
1408 /// drain, before any marker — and records `EARLY` if it is already gone.
1409 /// Two files that must disagree: `eof` must exist at the end, `early`
1410 /// must never exist.
1411 ///
1412 /// **The prompt sentinel is a negative control on JSON escaping.** The
1413 /// sentinel sits on the SECOND line of a multi-line prompt containing a
1414 /// double quote. `user_turn_line` escapes it, so the whole prompt arrives
1415 /// as one physical line and the stub's single `read` sees the sentinel. A
1416 /// `format!`-interpolated implementation would emit a torn two-line
1417 /// document, the stub's `read` would return only the first line, and the
1418 /// sentinel check would fail — which is exactly what should happen.
1419 #[test]
1420 fn pipe_owning_monitor_delivers_prompt_via_stdin_and_captures_stream() {
1421 const SENTINEL: &str = "TRACER-PROMPT-SENTINEL";
1422
1423 let dir = tempfile::tempdir().unwrap();
1424 let root = dir.path();
1425 let phase = 4u32;
1426 std::fs::create_dir_all(root.join(".devflow")).unwrap();
1427
1428 let eof_file = root.join("stdin-eof");
1429 let early_file = root.join("stdin-closed-early");
1430
1431 // A quote on line one, the sentinel on line two — see the doc above.
1432 let prompt = format!("first line with a \" quote\n{SENTINEL}");
1433
1434 let script = format!(
1435 r#"
1436set -u
1437IFS= read -r turn || {{ echo "NO_INITIAL_TURN_ON_STDIN" >&2; exit 91; }}
1438case "$turn" in
1439 *{SENTINEL}*) ;;
1440 *) echo "INITIAL_TURN_MISSING_PROMPT: $turn" >&2; exit 92 ;;
1441esac
1442
1443# Probe: block on stdin until EOF, then record it. stdout is redirected so
1444# this subshell does not hold the capture pipe open after the main shell exits.
1445#
1446# `exec 3<&0` then `cat <&3` is load-bearing, not a flourish: POSIX assigns
1447# /dev/null to a BACKGROUNDED list's stdin before any explicit redirection
1448# when job control is off. A bare `( cat > /dev/null ) &` therefore reads EOF
1449# instantly and reports an early close that never happened. The explicit
1450# `<&3` is applied after that default and overrides it.
1451exec 3<&0
1452( cat <&3 > /dev/null; printf 'EOF\n' > '{eof}' ) > /dev/null 2>&1 &
1453
1454printf '%s\n' '{{"type":"system","subtype":"init","session_id":"tracer-1"}}'
1455printf '%s\n' '{{"type":"system","subtype":"background_tasks_changed","tasks":[{{"task_id":"t1","task_type":"local_agent"}}]}}'
1456printf '%s\n' '{{"type":"system","subtype":"background_tasks_changed","tasks":[]}}'
1457
1458# The drain has landed but no marker has. A correct monitor is still holding
1459# stdin open here; sample it and record the violation if it is not.
1460sleep 0.5
1461if [ -f '{eof}' ]; then printf 'EARLY\n' > '{early}'; fi
1462
1463printf '%s\n' '{{"type":"result","subtype":"success","is_error":false,"session_id":"tracer-1","result":"DEVFLOW_RESULT: {{\"status\":\"success\",\"commits\":2}}"}}'
1464
1465# Bounded wait for EOF: a monitor that never closes stdin must fail the
1466# assertions below, not hang the suite.
1467i=0
1468while [ $i -lt 100 ] && [ ! -f '{eof}' ]; do
1469 sleep 0.1
1470 i=$((i+1))
1471done
1472exit 0
1473"#,
1474 eof = eof_file.display(),
1475 early = early_file.display(),
1476 );
1477
1478 let code = run_pipe_owning_monitor(
1479 root,
1480 phase,
1481 root,
1482 &prompt,
1483 Duration::from_secs(20),
1484 "sh",
1485 &["-c".to_string(), script],
1486 &[],
1487 )
1488 .expect("pipe-owning monitor should supervise the stub to completion");
1489
1490 let stderr = std::fs::read_to_string(crate::agent_result::stderr_path(root, phase))
1491 .unwrap_or_default();
1492 assert_eq!(
1493 code, 0,
1494 "stub exited {code}; 91 = no initial turn arrived on stdin, \
1495 92 = the turn arrived but did not carry the prompt (a JSON \
1496 escaping regression tears it across lines). stderr: {stderr:?}"
1497 );
1498
1499 assert!(
1500 !early_file.exists(),
1501 "the monitor closed the child's stdin BEFORE the close rule was \
1502 satisfied — the drain had landed but no DEVFLOW_RESULT marker had. \
1503 Constraint 4's AND cannot be honoured once stdin is gone: a \
1504 task-notification turn would have nowhere to be delivered."
1505 );
1506 assert!(
1507 eof_file.exists(),
1508 "the monitor never closed the child's stdin at all; the close rule \
1509 should have fired once the marker arrived with the task list drained"
1510 );
1511
1512 let capture =
1513 std::fs::read_to_string(crate::agent_result::stdout_path(root, phase)).unwrap();
1514 for expected in [
1515 r#""subtype":"init""#,
1516 r#""task_id":"t1""#,
1517 r#""tasks":[]"#,
1518 r#""type":"result""#,
1519 ] {
1520 assert!(
1521 capture.contains(expected),
1522 "capture is missing {expected}; got:\n{capture}"
1523 );
1524 }
1525 assert!(
1526 crate::agent_result::capture_is_claude_stream(&capture),
1527 "the capture must classify as a Claude stream-json document — \
1528 this is what makes 30b's stream parser reachable at all:\n{capture}"
1529 );
1530
1531 let result = crate::agent_result::evaluate_layer1(root, phase)
1532 .expect("Layer 1 must decide this capture");
1533 assert_eq!(
1534 result.status,
1535 crate::agent_result::AgentStatus::Success,
1536 "Layer 1 verdict from the stream capture: {result:?}"
1537 );
1538
1539 let exit = std::fs::read_to_string(crate::agent_result::exit_code_path(root, phase))
1540 .expect("the monitor must record the child's exit code");
1541 assert_eq!(exit.trim(), "0", "exit file contents: {exit:?}");
1542 }
1543
1544 /// Peer review 2026-08-03, CRITICAL: `BufRead::lines()` yields
1545 /// `Err(InvalidData)` on one non-UTF-8 byte, and the reader treated any read
1546 /// error as EOF — silently truncating the capture and dropping every later
1547 /// line, INCLUDING the terminal marker. The supervisor manufactured exactly
1548 /// the boundary-truncation failure constraint 9 exists to defend against.
1549 ///
1550 /// **What this does NOT establish:** that the real `claude` CLI ever emits
1551 /// non-UTF-8 on this stream. It emits JSON, which should be valid UTF-8. This
1552 /// pins the supervisor's robustness, not a demonstrated CLI behaviour.
1553 #[test]
1554 fn non_utf8_byte_does_not_truncate_the_capture() {
1555 let dir = tempfile::tempdir().unwrap();
1556 let root = dir.path();
1557 let phase = 11u32;
1558 std::fs::create_dir_all(root.join(".devflow")).unwrap();
1559
1560 // A raw 0xFF is invalid UTF-8 in any position. It sits BETWEEN two good
1561 // lines, so a reader that dies on it loses the marker that follows.
1562 let script = r#"
1563set -u
1564IFS= read -r _turn || exit 91
1565printf '%s\n' '{"type":"system","subtype":"init","session_id":"utf8-1"}'
1566printf 'raw-\377-bytes\n'
1567printf '%s\n' '{"type":"system","subtype":"background_tasks_changed","tasks":[]}'
1568printf '%s\n' '{"type":"result","subtype":"success","is_error":false,"session_id":"utf8-1","result":"DEVFLOW_RESULT: {\"status\":\"success\"}"}'
1569exit 0
1570"#;
1571
1572 let code = run_pipe_owning_monitor(
1573 root,
1574 phase,
1575 root,
1576 "prompt",
1577 Duration::from_secs(20),
1578 "sh",
1579 &["-c".to_string(), script.to_string()],
1580 &[],
1581 )
1582 .expect("the monitor must survive a non-UTF-8 byte on the child's stdout");
1583 assert_eq!(code, 0, "stub should exit cleanly");
1584
1585 let capture =
1586 std::fs::read_to_string(crate::agent_result::stdout_path(root, phase)).unwrap();
1587 assert!(
1588 capture.contains(r#""type":"result""#),
1589 "the terminal result event was lost: a non-UTF-8 byte earlier in the \
1590 stream truncated the capture. This is the regression:\n{capture}"
1591 );
1592 assert!(
1593 capture.contains("raw-"),
1594 "the undecodable line itself must still be teed (lossily), since the \
1595 capture is the verbatim record:\n{capture}"
1596 );
1597 let result = crate::agent_result::evaluate_layer1(root, phase)
1598 .expect("Layer 1 must still decide a capture that contained a bad byte");
1599 assert_eq!(
1600 result.status,
1601 crate::agent_result::AgentStatus::Success,
1602 "verdict after lossy decode: {result:?}"
1603 );
1604 }
1605
1606 /// Peer review 2026-08-03, CRITICAL: after the close rule released stdin the
1607 /// supervisor kept timing out on silence and fired `fire_idle_timeout`,
1608 /// writing an authoritative `IdleTimeout` verdict OVER a stage that had
1609 /// already reported success. `evaluate_layer1` reads that side channel first
1610 /// — by design, so nothing can shadow a real timeout — so the bogus verdict
1611 /// won and was unrecoverable.
1612 ///
1613 /// The timeout here (600ms) is injected short deliberately; the child sleeps
1614 /// well past it AFTER the marker. **What this does NOT establish:** that the
1615 /// 120s production floor is right — that rests on the keepalive measurement
1616 /// in `31-IDLE-GAP-MEASUREMENTS.md`, not on this test.
1617 #[test]
1618 fn no_idle_timeout_is_recorded_when_the_child_is_merely_slow_to_exit() {
1619 let dir = tempfile::tempdir().unwrap();
1620 let root = dir.path();
1621 let phase = 12u32;
1622 std::fs::create_dir_all(root.join(".devflow")).unwrap();
1623
1624 let script = r#"
1625set -u
1626IFS= read -r _turn || exit 91
1627printf '%s\n' '{"type":"system","subtype":"init","session_id":"slow-1"}'
1628printf '%s\n' '{"type":"system","subtype":"background_tasks_changed","tasks":[]}'
1629printf '%s\n' '{"type":"result","subtype":"success","is_error":false,"session_id":"slow-1","result":"DEVFLOW_RESULT: {\"status\":\"success\"}"}'
1630# Everything has been said; the close rule fires here. Now wind down slowly,
1631# well past the injected idle window, emitting nothing.
1632sleep 3
1633exit 0
1634"#;
1635
1636 let code = run_pipe_owning_monitor(
1637 root,
1638 phase,
1639 root,
1640 "prompt",
1641 Duration::from_millis(600),
1642 "sh",
1643 &["-c".to_string(), script.to_string()],
1644 &[],
1645 )
1646 .expect("a slow-exiting child that already reported is not a failure");
1647
1648 assert!(
1649 !crate::agent_result::idle_timeout_path(root, phase).exists(),
1650 "an idle-timeout verdict was written for a stage that had ALREADY \
1651 emitted its terminal marker and drained its tasks — silence after a \
1652 deliberate close is expected, not a hang"
1653 );
1654 assert_eq!(code, 0, "the child exited cleanly, if slowly");
1655
1656 let result = crate::agent_result::evaluate_layer1(root, phase)
1657 .expect("Layer 1 must decide this capture");
1658 assert_eq!(
1659 result.status,
1660 crate::agent_result::AgentStatus::Success,
1661 "a completed stage must not be reported as a timeout: {result:?}"
1662 );
1663 }
1664
1665 /// Peer review 2026-08-03 (found independently by BOTH reviewers and by the
1666 /// 31-04 plan review as W1): `status.code()` is `None` for a signal-killed
1667 /// child, and `unwrap_or(-1)` discarded the signal. `-1` matches neither the
1668 /// 137 nor the 127 arm, so a kernel OOM kill arrived as a generic `Failed`
1669 /// and routed to `GateReview` — asking a human to code-review a stage the
1670 /// kernel killed — instead of `GateInfra`.
1671 ///
1672 /// This asserts on what the monitor ACTUALLY writes for a real SIGKILL. The
1673 /// pre-existing arbitration test hardcoded `"137\n"` into its fixture, so it
1674 /// passed green against this defect the entire time — which is why this test
1675 /// spawns a child and kills it rather than writing the file itself.
1676 #[test]
1677 fn a_signal_killed_child_records_128_plus_signal_not_minus_one() {
1678 let dir = tempfile::tempdir().unwrap();
1679 let root = dir.path();
1680 let phase = 13u32;
1681 std::fs::create_dir_all(root.join(".devflow")).unwrap();
1682
1683 // SIGKILL itself: no exit code exists, only a termination signal.
1684 let script = r#"
1685set -u
1686IFS= read -r _turn || exit 91
1687printf '%s\n' '{"type":"system","subtype":"init","session_id":"sig-1"}'
1688kill -9 $$
1689"#;
1690
1691 let code = run_pipe_owning_monitor(
1692 root,
1693 phase,
1694 root,
1695 "prompt",
1696 Duration::from_secs(20),
1697 "sh",
1698 &["-c".to_string(), script.to_string()],
1699 &[],
1700 )
1701 .expect("the monitor must reap a signal-killed child");
1702
1703 assert_eq!(
1704 code, 137,
1705 "SIGKILL(9) must be recorded as 128+9=137, the value \
1706 `evaluate_layer2` and `reconcile_stream_success_against_exit_code` \
1707 map to ResourceKilled/GateInfra. -1 means the signal was discarded."
1708 );
1709 let exit = std::fs::read_to_string(crate::agent_result::exit_code_path(root, phase))
1710 .expect("the monitor must record the exit code");
1711 assert_eq!(exit.trim(), "137", "exit file contents: {exit:?}");
1712 }
1713
1714 #[test]
1715 fn shell_escape_handles_single_quotes() {
1716 assert_eq!(shell_escape("can't"), "'can'\\''t'");
1717 assert_eq!(shell_escape("a'b'c"), "'a'\\''b'\\''c'");
1718 }
1719
1720 #[test]
1721 fn shell_escape_handles_empty_string() {
1722 assert_eq!(shell_escape(""), "''");
1723 }
1724
1725 #[test]
1726 fn wait_for_agent_pid_returns_pid_when_file_exists() {
1727 let dir = tempfile::tempdir().unwrap();
1728 std::fs::create_dir_all(dir.path().join(".devflow")).unwrap();
1729 std::fs::write(
1730 crate::agent_result::agent_pid_path(dir.path(), 4),
1731 "12345\n",
1732 )
1733 .unwrap();
1734
1735 assert_eq!(wait_for_agent_pid(dir.path(), 4), Some(12345));
1736 }
1737
1738 #[test]
1739 fn wait_for_agent_pid_returns_none_when_file_missing() {
1740 let dir = tempfile::tempdir().unwrap();
1741
1742 assert_eq!(wait_for_agent_pid(dir.path(), 4), None);
1743 }
1744
1745 #[test]
1746 fn wait_for_agent_pid_returns_none_for_garbage_content() {
1747 let dir = tempfile::tempdir().unwrap();
1748 std::fs::create_dir_all(dir.path().join(".devflow")).unwrap();
1749 std::fs::write(
1750 crate::agent_result::agent_pid_path(dir.path(), 4),
1751 "not-a-pid",
1752 )
1753 .unwrap();
1754
1755 assert_eq!(wait_for_agent_pid(dir.path(), 4), None);
1756 }
1757
1758 #[test]
1759 fn spawn_monitor_captures_agent_pid_and_output() {
1760 let dir = tempfile::tempdir().unwrap();
1761 let state = state_in(dir.path());
1762 // Stub agent: write a known marker to stdout, then exit cleanly.
1763 let args = vec!["-c".to_string(), "echo MONITOR_READY".to_string()];
1764
1765 let monitor_pid = spawn_monitor(&state, "sh", &args, &[], MonitorLaunch::Legacy).unwrap();
1766 assert!(monitor_pid > 0);
1767
1768 // Observable side effect #1: the monitor records the agent PID to its
1769 // pid file with valid numeric content.
1770 let agent_pid = wait_for_agent_pid(dir.path(), state.phase)
1771 .expect("monitor should record the agent pid");
1772 assert!(agent_pid > 0);
1773
1774 // Observable side effect #2: the agent's stdout is captured to the
1775 // phase stdout file (proving the monitor actually ran the agent).
1776 let stdout_path = crate::agent_result::stdout_path(dir.path(), state.phase);
1777 let mut captured = String::new();
1778 for _ in 0..100 {
1779 if let Ok(contents) = std::fs::read_to_string(&stdout_path)
1780 && contents.contains("MONITOR_READY")
1781 {
1782 captured = contents;
1783 break;
1784 }
1785 std::thread::sleep(Duration::from_millis(20));
1786 }
1787 assert!(
1788 captured.contains("MONITOR_READY"),
1789 "expected MONITOR_READY in captured stdout, got {captured:?}"
1790 );
1791 }
1792
1793 /// WR-08 (13-REVIEW.md): sending SIGTERM/SIGINT to the monitor must also
1794 /// terminate the agent it owns. Before the fix, `cleanup()` only exited
1795 /// the monitor shell, leaving the agent orphaned and running/committing
1796 /// unsupervised with nothing left to call `devflow advance` for it.
1797 /// A one-line identity/state summary of a pid, for failure diagnostics.
1798 /// `Name`/`State`/`PPid` come from `/proc/<pid>/status`; the cmdline
1799 /// distinguishes a shell that exec'd its command from one that forked it.
1800 /// Test-only; never used in a decision.
1801 fn proc_snapshot(pid: u32) -> String {
1802 let Ok(status) = std::fs::read_to_string(format!("/proc/{pid}/status")) else {
1803 return format!("GONE (no /proc/{pid})");
1804 };
1805 let field = |key: &str| {
1806 status
1807 .lines()
1808 .find(|l| l.starts_with(key))
1809 .map(|l| l.split_whitespace().skip(1).collect::<Vec<_>>().join(" "))
1810 .unwrap_or_else(|| "?".into())
1811 };
1812 let cmdline = std::fs::read(format!("/proc/{pid}/cmdline"))
1813 .map(|raw| {
1814 let joined = raw
1815 .split(|&b| b == 0)
1816 .filter(|a| !a.is_empty())
1817 .map(|a| String::from_utf8_lossy(a).into_owned())
1818 .collect::<Vec<_>>()
1819 .join(" ");
1820 if joined.is_empty() {
1821 "<empty>".to_string()
1822 } else {
1823 joined
1824 }
1825 })
1826 .unwrap_or_else(|e| format!("<unreadable: {e}>"));
1827 format!(
1828 "ALIVE Name={} State={} PPid={} cmdline=[{cmdline}]",
1829 field("Name:"),
1830 field("State:"),
1831 field("PPid:")
1832 )
1833 }
1834
1835 #[test]
1836 fn sigterm_to_monitor_also_kills_the_agent() {
1837 let dir = tempfile::tempdir().unwrap();
1838 let state = state_in(dir.path());
1839 // Stub agent that runs long enough to observe: sleeps well past the
1840 // window this test needs to send SIGTERM and check liveness.
1841 let args = vec!["-c".to_string(), "sleep 30".to_string()];
1842
1843 let monitor_pid = spawn_monitor(&state, "sh", &args, &[], MonitorLaunch::Legacy).unwrap();
1844 let agent_pid = wait_for_agent_pid(dir.path(), state.phase)
1845 .expect("monitor should record the agent pid");
1846 assert!(
1847 crate::agent::agent_running(agent_pid),
1848 "agent should be running before SIGTERM"
1849 );
1850
1851 // Snapshot both processes before signalling. This assertion fails in
1852 // containerised CI and cannot be reproduced locally, and a bare
1853 // "still running" message discards everything that could explain it
1854 // — the same antipattern that made 999.47 expensive to diagnose.
1855 let monitor_before = proc_snapshot(monitor_pid);
1856 let agent_before = proc_snapshot(agent_pid);
1857
1858 // SIGTERM the monitor, as an operator (or lock.rs's stale-holder
1859 // reclaim path) would to abort a run.
1860 let kill_rc = unsafe { libc::kill(monitor_pid as libc::pid_t, libc::SIGTERM) };
1861 let kill_err = if kill_rc == 0 {
1862 "ok".to_string()
1863 } else {
1864 format!("errno {}", std::io::Error::last_os_error())
1865 };
1866
1867 // The agent should be killed promptly by the monitor's trap —
1868 // poll rather than sleep a fixed amount to keep this fast and
1869 // avoid flaking under load. (Window widened to 5s: at 2s this
1870 // still flaked under a fully parallel workspace test run.)
1871 //
1872 // 2026-07-26: this was widened 5s -> 15s for the containerised CI
1873 // job and STILL failed, then reverted to 5s. That widening was a
1874 // mistake: 15s is far beyond any plausible trap-and-kill latency,
1875 // so the agent is not being reaped SLOWLY, it is not being reaped.
1876 // Buying silence with a bigger number would have hidden a real
1877 // defect behind a green check — the exact false negative this
1878 // repository keeps getting bitten by.
1879 //
1880 // The trap mechanism itself is verified working: DevFlow's real
1881 // monitor script shape was run under both `bash` and `dash` (the
1882 // container's /bin/sh is dash, the Fedora host's is bash) and both
1883 // killed the backgrounded agent correctly. So the defect is in how
1884 // the agent is spawned or identified under container timing, not in
1885 // the shell trap — see 999.47, whose confirmed transient fork/exec
1886 // window is the prime suspect for the same class of failure here.
1887 //
1888 // Leave this red until that is fixed. Do NOT widen it again.
1889 let mut still_running = true;
1890 for _ in 0..250 {
1891 if !crate::agent::agent_running(agent_pid) {
1892 still_running = false;
1893 break;
1894 }
1895 std::thread::sleep(Duration::from_millis(20));
1896 }
1897 let monitor_after = proc_snapshot(monitor_pid);
1898 let agent_after = proc_snapshot(agent_pid);
1899 let pidfile =
1900 std::fs::read_to_string(crate::agent_result::agent_pid_path(dir.path(), state.phase))
1901 .unwrap_or_else(|e| format!("<unreadable: {e}>"));
1902
1903 assert!(
1904 !still_running,
1905 "agent (pid {agent_pid}) was orphaned — still running after monitor SIGTERM\n\
1906 \x20 monitor pid: {monitor_pid}\n\
1907 \x20 kill(TERM) rc: {kill_rc} ({kill_err})\n\
1908 \x20 monitor before: {monitor_before}\n\
1909 \x20 monitor after: {monitor_after}\n\
1910 \x20 agent pid: {agent_pid}\n\
1911 \x20 agent before: {agent_before}\n\
1912 \x20 agent after: {agent_after}\n\
1913 \x20 pidfile contents: {}\n\
1914 Read the monitor's `after` line first. GONE means the shell died \
1915 without running its trap — most likely SIGTERM arrived before \
1916 `trap` was installed, or it was killed rather than handling the \
1917 signal, either way leaving the agent unreaped. STILL ALIVE means \
1918 the trap never fired or `kill $apid` failed, so compare the agent \
1919 pid against the pidfile and check the agent's PPid: if PPid is not \
1920 the monitor, `$!` did not name the process we are polling. If the \
1921 agent's Name is `sh` rather than `sleep`, the agent shell forked \
1922 rather than exec'd, so killing it leaves its own child behind.",
1923 pidfile.trim()
1924 );
1925 }
1926
1927 #[test]
1928 fn spawn_monitor_runs_agent_in_worktree_but_captures_in_project_root() {
1929 let dir = tempfile::tempdir().unwrap();
1930 let worktree = dir.path().join(".worktrees/phase-04");
1931 std::fs::create_dir_all(&worktree).unwrap();
1932 let mut state = state_in(dir.path());
1933 state.worktree_path = Some(worktree.clone());
1934
1935 // Stub agent: print its cwd so the test proves the monitor changed
1936 // directories before launching the agent.
1937 let args = vec!["-c".to_string(), "pwd; echo WORKTREE_READY".to_string()];
1938
1939 let monitor_pid = spawn_monitor(&state, "sh", &args, &[], MonitorLaunch::Legacy).unwrap();
1940 assert!(monitor_pid > 0);
1941
1942 let agent_pid = wait_for_agent_pid(dir.path(), state.phase)
1943 .expect("monitor should record the agent pid in the main project");
1944 assert!(agent_pid > 0);
1945
1946 let stdout_path = crate::agent_result::stdout_path(dir.path(), state.phase);
1947 let mut captured = String::new();
1948 for _ in 0..100 {
1949 if let Ok(contents) = std::fs::read_to_string(&stdout_path)
1950 && contents.contains("WORKTREE_READY")
1951 {
1952 captured = contents;
1953 break;
1954 }
1955 std::thread::sleep(Duration::from_millis(20));
1956 }
1957
1958 assert!(
1959 captured.contains(&worktree.display().to_string()),
1960 "agent did not run in worktree cwd; captured stdout: {captured:?}"
1961 );
1962 assert!(
1963 stdout_path.exists(),
1964 "stdout capture missing in main .devflow"
1965 );
1966 assert!(
1967 !crate::agent_result::stdout_path(&worktree, state.phase).exists(),
1968 "stdout capture should not be written under the worktree"
1969 );
1970 }
1971
1972 /// Build the fixture repositories through the scrubbing constructor, as
1973 /// every other test module in this phase does (`version.rs:1102`).
1974 ///
1975 /// A bare `Command::new("git")` here would itself inherit an ambient
1976 /// hostile `GIT_DIR` — so under this phase's own acceptance command
1977 /// (`GIT_DIR=<throwaway>/.git cargo test -p devflow-core ...`) the
1978 /// fixture setup would target the throwaway repository instead of
1979 /// `root`, and the test below would fail for a reason that has nothing
1980 /// to do with the behavior it is guarding.
1981 fn git(root: &Path, args: &[&str]) {
1982 let ok = crate::test_support::git_command(root)
1983 .args(args)
1984 .output()
1985 .unwrap()
1986 .status
1987 .success();
1988 assert!(ok, "git {args:?} failed");
1989 }
1990
1991 fn init_repo(root: &Path) {
1992 git(root, &["init", "-q"]);
1993 git(root, &["config", "user.email", "test@example.com"]);
1994 git(root, &["config", "user.name", "Test"]);
1995 }
1996
1997 /// 27-REVIEW WR-03: the `sh` this function spawns owns the coding
1998 /// agent, and whatever environment rides down with it reaches every git
1999 /// command the agent runs (`sh` -> agent -> agent's git children). This
2000 /// proves the scrub with a real spawned agent process, not by
2001 /// inspecting the `Command` object: the agent shells out to
2002 /// `git rev-parse --absolute-git-dir`, and the resolved path must be
2003 /// the caller's own workdir, never a hostile `GIT_DIR` pointed at an
2004 /// unrelated foreign repository.
2005 ///
2006 /// Mirrors `tag_reads_resolve_caller_root_under_a_hostile_git_dir`
2007 /// (version.rs, 27-03/WR-01): `GIT_DIR` is never set on this test
2008 /// process itself (Rust 2024 `unsafe`, unsound under threaded tests —
2009 /// Phase 25 D-14), only on one freshly spawned child re-invoking this
2010 /// binary filtered to this test.
2011 #[test]
2012 fn spawn_monitor_agent_git_calls_resolve_workdir_not_a_hostile_git_dir() {
2013 const INNER_ROOT: &str = "DEVFLOW_27_MONITOR_INNER_ROOT";
2014
2015 if let Ok(root) = std::env::var(INNER_ROOT) {
2016 // Inner mode: GIT_DIR points at a foreign repository unrelated
2017 // to `root`, scoped to this child process only.
2018 let root = std::path::PathBuf::from(root);
2019 let state = state_in(&root);
2020 let args = vec![
2021 "-c".to_string(),
2022 "git rev-parse --absolute-git-dir".to_string(),
2023 ];
2024
2025 spawn_monitor(&state, "sh", &args, &[], MonitorLaunch::Legacy).unwrap();
2026 wait_for_agent_pid(&root, state.phase).expect("monitor should record the agent pid");
2027
2028 let stdout_path = crate::agent_result::stdout_path(&root, state.phase);
2029 let mut captured = String::new();
2030 for _ in 0..100 {
2031 if let Ok(contents) = std::fs::read_to_string(&stdout_path)
2032 && !contents.trim().is_empty()
2033 {
2034 captured = contents;
2035 break;
2036 }
2037 std::thread::sleep(Duration::from_millis(20));
2038 }
2039
2040 let resolved = std::fs::canonicalize(captured.trim())
2041 .expect("agent's reported git-dir must exist on disk");
2042 let expected =
2043 std::fs::canonicalize(root.join(".git")).expect("caller repo .git must exist");
2044 assert_eq!(
2045 resolved, expected,
2046 "agent's git call resolved to a hostile GIT_DIR's \
2047 repository instead of the caller's own workdir: \
2048 got {resolved:?}, want {expected:?}"
2049 );
2050 return;
2051 }
2052
2053 // Outer mode: a real repository at `root`, and an unrelated
2054 // foreign repository whose .git must never leak into the agent's
2055 // environment.
2056 let dir = tempfile::tempdir().unwrap();
2057 let root = dir.path().join("caller-repo");
2058 std::fs::create_dir_all(&root).unwrap();
2059 init_repo(&root);
2060
2061 let foreign = tempfile::tempdir().unwrap();
2062 init_repo(foreign.path());
2063
2064 let exe = std::env::current_exe().expect("current_exe for child re-invocation");
2065 let out = std::process::Command::new(&exe)
2066 // Substring filter, NOT `--exact`: the binary's real test name
2067 // is module-qualified (`monitor::tests::spawn_monitor_...`), so
2068 // `--exact` against the bare name matches nothing, runs zero
2069 // tests, and still exits 0 — a false green.
2070 .arg("spawn_monitor_agent_git_calls_resolve_workdir_not_a_hostile_git_dir")
2071 .arg("--test-threads=1")
2072 .env(INNER_ROOT, root.to_str().unwrap())
2073 .env("GIT_DIR", foreign.path().join(".git"))
2074 .output()
2075 .expect("spawn hostile child test process");
2076
2077 let stdout = String::from_utf8_lossy(&out.stdout);
2078 // Assert the child actually RAN the test, not merely that it
2079 // exited 0. A filter that matches nothing exits 0 with "0 passed".
2080 assert!(
2081 stdout.contains("1 passed"),
2082 "child test process must have run exactly the inner test; \
2083 stdout:\n{stdout}"
2084 );
2085 assert!(
2086 out.status.success(),
2087 "monitor-spawned agent (hostile GIT_DIR pointed at an \
2088 unrelated foreign repository) must still resolve its git \
2089 calls against the caller's own workdir; child exit status \
2090 {:?}\nstdout:\n{stdout}",
2091 out.status
2092 );
2093 }
2094
2095 #[test]
2096 fn spawn_monitor_treats_agent_args_as_literal_argv() {
2097 let dir = tempfile::tempdir().unwrap();
2098 let state = state_in(dir.path());
2099 let payload = "value; touch INJECTED";
2100 let args = vec![
2101 "-c".to_string(),
2102 "printf '%s\\n' \"$0\"; echo ARGV_SAFE".to_string(),
2103 payload.to_string(),
2104 ];
2105
2106 spawn_monitor(&state, "sh", &args, &[], MonitorLaunch::Legacy).unwrap();
2107 wait_for_agent_pid(dir.path(), state.phase).expect("monitor should record the agent pid");
2108
2109 let stdout_path = crate::agent_result::stdout_path(dir.path(), state.phase);
2110 let mut captured = String::new();
2111 for _ in 0..100 {
2112 if let Ok(contents) = std::fs::read_to_string(&stdout_path)
2113 && contents.contains("ARGV_SAFE")
2114 {
2115 captured = contents;
2116 break;
2117 }
2118 std::thread::sleep(Duration::from_millis(20));
2119 }
2120
2121 assert!(
2122 captured.contains(payload),
2123 "literal argv missing: {captured:?}"
2124 );
2125 assert!(captured.contains("ARGV_SAFE"));
2126 assert!(!dir.path().join("INJECTED").exists());
2127 }
2128
2129 // ---- idle timeout (31-02, D-01..D-08) --------------------------------
2130
2131 /// D-04: a value below the floor is raised to it, and the fact is
2132 /// observable to the CALLER as a value — not only as a log line a test
2133 /// would have to capture stdout to see.
2134 #[test]
2135 fn idle_timeout_secs_clamps_below_floor_and_logs() {
2136 let setting = parse_idle_timeout_secs(Some("5".to_string()));
2137
2138 assert_eq!(setting.timeout, Duration::from_secs(120));
2139 assert!(setting.clamped(), "the clamp must be observable as a value");
2140 assert_eq!(
2141 setting.resolution,
2142 IdleTimeoutResolution::Clamped { configured: 5 }
2143 );
2144
2145 // The notice must NAME the configured value, the floor, and the value
2146 // actually in force — a clamp that says only "clamped" leaves the
2147 // operator guessing which of the three numbers won.
2148 let notice = setting.notice().expect("a clamp owes a loud notice");
2149 for fragment in ["5", "120", IDLE_TIMEOUT_ENV] {
2150 assert!(
2151 notice.contains(fragment),
2152 "notice must name {fragment:?}; got: {notice}"
2153 );
2154 }
2155 }
2156
2157 /// The floor raises, it never lowers: a value above it survives verbatim
2158 /// and reports no clamp.
2159 #[test]
2160 fn idle_timeout_secs_accepts_values_above_floor() {
2161 let setting = parse_idle_timeout_secs(Some("300".to_string()));
2162
2163 assert_eq!(setting.timeout, Duration::from_secs(300));
2164 assert!(!setting.clamped());
2165 assert_eq!(setting.resolution, IdleTimeoutResolution::Configured);
2166 assert_eq!(
2167 setting.notice(),
2168 None,
2169 "an honoured value is unremarkable and must not shout"
2170 );
2171
2172 // Boundary: exactly the floor is CONFIGURED, not CLAMPED. An
2173 // off-by-one here would report a clamp that never happened and train
2174 // operators to ignore the notice.
2175 let exact = parse_idle_timeout_secs(Some("120".to_string()));
2176 assert_eq!(exact.resolution, IdleTimeoutResolution::Configured);
2177 assert!(!exact.clamped());
2178 }
2179
2180 /// Absent, empty, and unparseable all resolve to the floor. The three are
2181 /// NOT equivalent in loudness: nothing configured is silent, a typo is not.
2182 #[test]
2183 fn idle_timeout_secs_defaults_to_the_floor() {
2184 let floor = Duration::from_secs(IDLE_TIMEOUT_FLOOR_SECS);
2185
2186 for raw in [None, Some(String::new()), Some(" ".to_string())] {
2187 let setting = parse_idle_timeout_secs(raw.clone());
2188 assert_eq!(setting.timeout, floor, "raw {raw:?} must yield the floor");
2189 assert_eq!(setting.resolution, IdleTimeoutResolution::Default);
2190 assert_eq!(setting.notice(), None, "nothing chosen is not an error");
2191 }
2192
2193 for raw in ["banana", "60O", "-5", "30.5"] {
2194 let setting = parse_idle_timeout_secs(Some(raw.to_string()));
2195 assert_eq!(setting.timeout, floor, "raw {raw:?} must yield the floor");
2196 assert_eq!(
2197 setting.resolution,
2198 IdleTimeoutResolution::Unparseable {
2199 raw: raw.to_string()
2200 }
2201 );
2202 assert!(
2203 setting.notice().is_some(),
2204 "a typo that silently halves an intended timeout must be loud: {raw:?}"
2205 );
2206 }
2207 }
2208
2209 /// D-01/D-03: every line resets the window, and there is no outer
2210 /// wall-clock bound. A child that keeps talking for FOUR times the idle
2211 /// timeout is never terminated.
2212 ///
2213 /// The timeout is injected short (400ms) rather than using the 120s
2214 /// production default — this measures the RESET MECHANISM, and does so at
2215 /// a scale the suite can afford. **What it does not establish:** that 120s
2216 /// is the right production value. That rests on the 2026-08-03 keepalive
2217 /// measurement recorded on [`IDLE_TIMEOUT_FLOOR_SECS`], not on this test.
2218 #[test]
2219 fn idle_timer_resets_on_every_stream_line() {
2220 let dir = tempfile::tempdir().unwrap();
2221 let root = dir.path();
2222 let phase = 6u32;
2223 std::fs::create_dir_all(root.join(".devflow")).unwrap();
2224
2225 // 12 lines x 100ms = 1.2s of talking against a 400ms window. Any
2226 // implementation that resets on milestones only, or that imposes an
2227 // outer bound, kills this child before it finishes.
2228 let script = r#"
2229set -u
2230IFS= read -r turn || exit 91
2231i=0
2232while [ $i -lt 12 ]; do
2233 printf '%s\n' '{"type":"system","subtype":"heartbeat","n":'"$i"'}'
2234 sleep 0.1
2235 i=$((i+1))
2236done
2237printf '%s\n' '{"type":"result","subtype":"success","is_error":false,"session_id":"idle-1","result":"DEVFLOW_RESULT: {\"status\":\"success\"}"}'
2238exit 0
2239"#;
2240
2241 let started = std::time::Instant::now();
2242 let code = run_pipe_owning_monitor(
2243 root,
2244 phase,
2245 root,
2246 "prompt",
2247 Duration::from_millis(400),
2248 "sh",
2249 &["-c".to_string(), script.to_string()],
2250 &[],
2251 )
2252 .expect("a chatty child must be supervised to completion");
2253 let elapsed = started.elapsed();
2254
2255 assert_eq!(code, 0, "the chatty child must exit cleanly, not be killed");
2256 assert!(
2257 !crate::agent_result::idle_timeout_path(root, phase).exists(),
2258 "no timeout may fire while the child is still emitting lines"
2259 );
2260 assert!(
2261 elapsed > Duration::from_millis(400),
2262 "the run must outlast the idle window, else it proves nothing \
2263 about resetting: {elapsed:?}"
2264 );
2265
2266 let capture =
2267 std::fs::read_to_string(crate::agent_result::stdout_path(root, phase)).unwrap();
2268 assert_eq!(
2269 capture.matches("heartbeat").count(),
2270 12,
2271 "all twelve resets must have been observed: {capture:?}"
2272 );
2273 }
2274
2275 /// D-05, and the assertion the whole ordering exists for.
2276 ///
2277 /// The observation is made LIVE, by a watcher thread sampling the child's
2278 /// liveness at the first instant the verdict file exists — not by
2279 /// inspecting order after the fact, which cannot distinguish
2280 /// write-then-kill from kill-then-write.
2281 ///
2282 /// Its own negative control is structural: if the implementation wrote the
2283 /// verdict AFTER terminating, the watcher would sample a dead child and
2284 /// this test fails with `Some(false)`. The stub ignores `SIGTERM` so the
2285 /// window in which "file exists AND child alive" is observable is the full
2286 /// `TERMINATE_VERIFY_WAIT`, rather than a microsecond race.
2287 ///
2288 /// **What the duration of this test measures:** almost entirely
2289 /// `agent::TERMINATE_VERIFY_WAIT` (3s), because the stub refuses `SIGTERM`
2290 /// and must be escalated to `SIGKILL`. The 250ms idle window is a rounding
2291 /// error against it.
2292 #[test]
2293 fn idle_timeout_writes_side_channel_before_terminating_child() {
2294 let dir = tempfile::tempdir().unwrap();
2295 let root = dir.path().to_path_buf();
2296 let phase = 7u32;
2297 std::fs::create_dir_all(root.join(".devflow")).unwrap();
2298
2299 // One line, then silence. `trap '' TERM` widens the observation
2300 // window to the full SIGTERM->SIGKILL escalation.
2301 let script = r#"
2302set -u
2303IFS= read -r turn || exit 91
2304trap '' TERM
2305printf '%s\n' '{"type":"system","subtype":"init","session_id":"idle-2"}'
2306sleep 120
2307"#;
2308
2309 let verdict = crate::agent_result::idle_timeout_path(&root, phase);
2310 let pid_file = crate::agent_result::agent_pid_path(&root, phase);
2311 let watcher = std::thread::spawn(move || {
2312 let deadline = std::time::Instant::now() + Duration::from_secs(30);
2313 let mut pid: Option<u32> = None;
2314 while std::time::Instant::now() < deadline {
2315 if pid.is_none() {
2316 pid = std::fs::read_to_string(&pid_file)
2317 .ok()
2318 .and_then(|s| s.trim().parse::<u32>().ok());
2319 }
2320 if verdict.exists() {
2321 // Sample liveness at the FIRST moment the verdict exists.
2322 return pid.map(crate::agent::agent_running);
2323 }
2324 std::thread::sleep(Duration::from_millis(5));
2325 }
2326 None
2327 });
2328
2329 let code = run_pipe_owning_monitor(
2330 &root,
2331 phase,
2332 &root,
2333 "prompt",
2334 Duration::from_millis(250),
2335 "sh",
2336 &["-c".to_string(), script.to_string()],
2337 &[],
2338 )
2339 .expect("a silent child must still produce a supervised outcome");
2340
2341 let observed = watcher.join().expect("watcher thread panicked");
2342 assert_eq!(
2343 observed,
2344 Some(true),
2345 "the verdict must be on disk while the child is STILL ALIVE. \
2346 Some(false) = written after termination (the D-05 violation); \
2347 None = the verdict never appeared at all"
2348 );
2349
2350 // The verdict must also be readable and correct, not merely present.
2351 let raw = std::fs::read_to_string(crate::agent_result::idle_timeout_path(&root, phase))
2352 .expect("verdict file must be readable");
2353 let record: IdleTimeoutRecord = serde_json::from_str(&raw).expect("verdict must parse");
2354 assert_eq!(record.status, "idle_timeout");
2355 assert_eq!(record.idle_secs, 0, "250ms truncates to 0 whole seconds");
2356 assert!(record.agent_pid > 1);
2357
2358 // And the whole cascade must agree: Layer 1 reports the timeout.
2359 let result = crate::agent_result::evaluate_layer1(&root, phase)
2360 .expect("Layer 1 must decide a timed-out run");
2361 assert_eq!(
2362 result.status,
2363 crate::agent_result::AgentStatus::IdleTimeout,
2364 "the monitor's verdict must survive all the way to the oracle"
2365 );
2366
2367 // The child was killed, so it has no ordinary exit code — the point is
2368 // that the stage machine still reaches a gate rather than hanging.
2369 assert!(
2370 crate::agent_result::exit_code_path(&root, phase).exists(),
2371 "the exit file must still be written so advance() is reachable"
2372 );
2373 let _ = code;
2374
2375 // The loud monitor-log entry (D-04/D-07's readable-after-the-fact
2376 // obligation) must exist too — the monitor's stdio is null, so this
2377 // file is the only place it can land.
2378 let log = std::fs::read_to_string(crate::agent_result::monitor_log_path(&root, phase))
2379 .expect("the monitor must log its own timeout");
2380 assert!(log.contains("idle-timeout"), "log entry missing: {log:?}");
2381 }
2382
2383 /// Minimal git repo: `develop` plus a `feature/phase-NN` branch carrying
2384 /// `commits` extra commits.
2385 fn init_repo_with_feature_commits(root: &Path, phase: u32, commits: usize) {
2386 let git = |args: &[&str]| {
2387 let output = crate::git::git_command(root).args(args).output().unwrap();
2388 assert!(
2389 output.status.success(),
2390 "git {args:?} failed: {}",
2391 String::from_utf8_lossy(&output.stderr)
2392 );
2393 };
2394 git(&["init"]);
2395 git(&["config", "user.email", "devflow@example.com"]);
2396 git(&["config", "user.name", "DevFlow Tests"]);
2397 git(&["config", "commit.gpgsign", "false"]);
2398 git(&["config", "core.hooksPath", "/dev/null"]);
2399 git(&["checkout", "-b", "develop"]);
2400 std::fs::write(root.join("README.md"), "base\n").unwrap();
2401 git(&["add", "README.md"]);
2402 git(&["commit", "-m", "base"]);
2403
2404 let branch = format!("feature/phase-{phase:02}");
2405 git(&["checkout", "-b", &branch]);
2406 for i in 0..commits {
2407 let name = format!("work-{i}.txt");
2408 std::fs::write(root.join(&name), "work\n").unwrap();
2409 git(&["add", &name]);
2410 git(&["commit", "-m", &format!("feat: agent work {i}")]);
2411 }
2412 }
2413
2414 fn commit_count(root: &Path, phase: u32) -> u32 {
2415 let range = format!("develop..feature/phase-{phase:02}");
2416 let output = crate::git::git_command(root)
2417 .args(["rev-list", "--count", &range])
2418 .output()
2419 .unwrap();
2420 String::from_utf8_lossy(&output.stdout)
2421 .trim()
2422 .parse()
2423 .unwrap()
2424 }
2425
2426 /// D-07/T-31-09: a timeout READS the commit log and never writes to it. A
2427 /// timeout can be a false positive, and destroying real work on a false
2428 /// positive is unrecoverable.
2429 ///
2430 /// The "commits were enumerated" half is this test's negative control, and
2431 /// it is not optional: if enumeration silently returned nothing, "no
2432 /// commits were rolled back" would be trivially, vacuously true.
2433 #[test]
2434 fn idle_timeout_does_not_roll_back_commits() {
2435 let dir = tempfile::tempdir().unwrap();
2436 let root = dir.path();
2437 let phase = 8u32;
2438 init_repo_with_feature_commits(root, phase, 2);
2439 std::fs::create_dir_all(root.join(".devflow")).unwrap();
2440
2441 let before = commit_count(root, phase);
2442 assert_eq!(before, 2, "fixture precondition");
2443
2444 // No TERM trap here: the child dies promptly, keeping this test fast.
2445 let script = r#"
2446set -u
2447IFS= read -r turn || exit 91
2448printf '%s\n' '{"type":"system","subtype":"init","session_id":"idle-3"}'
2449sleep 120
2450"#;
2451
2452 run_pipe_owning_monitor(
2453 root,
2454 phase,
2455 root,
2456 "prompt",
2457 Duration::from_millis(250),
2458 "sh",
2459 &["-c".to_string(), script.to_string()],
2460 &[],
2461 )
2462 .expect("a silent child must still produce a supervised outcome");
2463
2464 assert_eq!(
2465 commit_count(root, phase),
2466 before,
2467 "an idle timeout must never roll back, reset, or revert a commit"
2468 );
2469
2470 // NEGATIVE CONTROL: enumeration must actually have found them, else
2471 // the assertion above is vacuous.
2472 let raw = std::fs::read_to_string(crate::agent_result::idle_timeout_path(root, phase))
2473 .expect("verdict file must exist");
2474 let record: IdleTimeoutRecord = serde_json::from_str(&raw).expect("verdict must parse");
2475 assert_eq!(
2476 record.commits.len(),
2477 2,
2478 "the verdict must NAME the commits, not merely leave them alone"
2479 );
2480 for commit in &record.commits {
2481 assert_eq!(commit.sha.len(), 40, "full sha expected: {commit:?}");
2482 assert!(
2483 commit.subject.starts_with("feat: agent work"),
2484 "subject must survive enumeration: {commit:?}"
2485 );
2486 }
2487
2488 // And the operator-facing reason names them.
2489 let result = crate::agent_result::evaluate_layer1(root, phase).unwrap();
2490 assert_eq!(result.commits, Some(2));
2491 let reason = result.reason.unwrap();
2492 assert!(
2493 reason.contains("NONE of them were rolled back"),
2494 "reason: {reason}"
2495 );
2496 }
2497}