Skip to main content

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}