From e9a89e91ce8dad5ed2c32a046dea8bf19343a942 Mon Sep 17 00:00:00 2001 From: Amin Chirazi <32016576+AminChirazi@users.noreply.github.com> Date: Tue, 28 Jul 2026 10:15:26 +0400 Subject: [PATCH 1/2] fix(agent): replay agent cassettes in directory mode, and bound the pipe drain Two bugs that together made `app: agent` flows unusable in CI. Both were hit recording DataMaker's agent suite; neither is visible from the code alone. 1. `run_suite` had no `app: agent` branch. Directory mode fell through to the step-replay loader, which parses a trace one JSON object per line. An agent cassette is a single `{app, mocks, cassette}` document, so it failed on line 1 and every agent flow errored with "invalid trace line: EOF while parsing an object at line 1 column 1" - traces `flowproof record` had just written, and that `flowproof run ` replayed green one at a time. Directory mode is what a suite and a CI job invoke, so the whole suite was dead there while single-spec runs looked fine. The branch mirrors the one `run_one` already had, including the containment line and the before/after hooks. `RunReport::agent` carries the verdict: its trace_id is neither "skipped" nor "errored", because the suite tallies those by that field and an agent flow that ran belongs in the ran/passed count. 2. `read_pipes` waited for EOF, unbounded, after the wait. EOF arrives when the last write end closes, not when the child dies. An agent that spawns its own server hands that grandchild the same stdout, so killing the child at the 300s deadline closed nothing and `read_to_string` blocked forever. A DataMaker replay sat 26 minutes on a 300-second timeout, produced no output and had to be killed by hand - the exact failure the deadline's own comment says it prevents. Draining now starts at spawn on its own thread and is bounded by a grace period, keeping whatever partial output arrived, since for a run that already went wrong that text is the only diagnostic there is. Starting at spawn also fixes a second latent bug in the same lines: draining only after the wait means a child that outwrites the ~64KB pipe buffer blocks in `write`, never exits, and is killed at the timeout - reported as a hung agent when nothing was wrong with it. Each fix has a regression test, and each was confirmed to fail without it: mutating out the suite branch reproduces the exact "invalid trace line" error, and mutating the drain back to an unbounded wait hangs the test past 120s. Reported-in: #183 --- crates/flowproof-adapters/src/agent_runner.rs | 157 ++++++++++++++++-- crates/flowproof-cli/src/lib.rs | 82 +++++++++ crates/flowproof-cli/tests/agent_flow_e2e.rs | 43 +++++ crates/flowproof-replay/src/report.rs | 34 ++++ 4 files changed, 302 insertions(+), 14 deletions(-) diff --git a/crates/flowproof-adapters/src/agent_runner.rs b/crates/flowproof-adapters/src/agent_runner.rs index 51b73fe..28cd073 100644 --- a/crates/flowproof-adapters/src/agent_runner.rs +++ b/crates/flowproof-adapters/src/agent_runner.rs @@ -17,6 +17,7 @@ use std::collections::BTreeMap; use std::io::Read; use std::process::{Command, Stdio}; +use std::sync::{Arc, Mutex, mpsc}; use std::time::{Duration, Instant}; use flowproof_trace::cassette::{Cassette, Divergence}; @@ -353,18 +354,82 @@ fn wait_to_deadline( (status, timed_out) } -/// Drain a child's stdout and stderr pipes to strings. -fn read_pipes(child: &mut std::process::Child) -> (String, String) { - let read = |pipe: Option<&mut dyn Read>| { - let mut buffer = String::new(); - if let Some(pipe) = pipe { - let _ = pipe.read_to_string(&mut buffer); - } - buffer - }; - let stdout = read(child.stdout.as_mut().map(|p| p as &mut dyn Read)); - let stderr = read(child.stderr.as_mut().map(|p| p as &mut dyn Read)); - (stdout, stderr) +/// How long to keep draining a pipe after the child is gone. +/// +/// Only reached when the write end outlived the child (see [`PipeDrain`]); +/// a normal exit closes the pipe and the drain finishes at once. +const PIPE_DRAIN_GRACE: Duration = Duration::from_secs(5); + +/// A child pipe being drained on its own thread. +/// +/// TWO BUGS LIVE WHERE THIS USED TO BE A BLOCKING `read_to_string` AFTER +/// THE WAIT, and neither is guessable from the old four-line body: +/// +/// 1. **The wait deadlocked against the pipe buffer.** Draining only after +/// the child exits means a child that writes more than the OS pipe +/// buffer (~64 KB) blocks in `write` forever, never exits, and is then +/// killed at the timeout - reported as a hung agent when it was really +/// a full pipe. Draining starts at spawn now, so the child always has +/// somewhere to write. +/// +/// 2. **`read_to_string` waited for EOF, which a GRANDCHILD can withhold.** +/// EOF arrives when the last write end closes, not when the child dies. +/// An agent that spawns its own server (`opencode serve` under the +/// OpenCode SDK) hands that grandchild the same stdout, so killing the +/// child at the timeout closed nothing and the read blocked forever. A +/// flowproof run against DataMaker's agent suite sat for 26 minutes on a +/// 300-second timeout, produced no output, and had to be killed by hand. +/// That is the failure the deadline above says it prevents: "an agent +/// that hangs ... would otherwise take the whole suite down with it". +/// +/// So the drain is bounded and the partial output is kept: whatever arrived +/// before the grace expired is returned rather than discarded, because for +/// a run that already went wrong that text is the only diagnostic there is. +struct PipeDrain { + buffer: Arc>, + finished: mpsc::Receiver<()>, +} + +impl PipeDrain { + /// Start draining `pipe` immediately, before the caller waits on the child. + fn start(pipe: Option) -> Self { + let buffer = Arc::new(Mutex::new(String::new())); + let (done, finished) = mpsc::channel(); + let sink = Arc::clone(&buffer); + std::thread::spawn(move || { + if let Some(mut pipe) = pipe { + // Chunked rather than `read_to_string` so a partial read is + // still visible in `buffer` when the grace expires. + let mut chunk = [0u8; 8192]; + loop { + match pipe.read(&mut chunk) { + Ok(0) | Err(_) => break, + Ok(n) => { + if let Ok(mut sink) = sink.lock() { + sink.push_str(&String::from_utf8_lossy(&chunk[..n])); + } + } + } + } + } + let _ = done.send(()); + }); + Self { buffer, finished } + } + + /// Take what was drained, waiting at most `grace` for the pipe to close. + /// + /// The reader thread is deliberately left running when the grace expires: + /// it is blocked on a pipe held open by a process flowproof does not own, + /// it holds nothing but its own buffer, and the process exits shortly + /// after. Killing it is not possible in safe Rust and not worth it. + fn collect(self, grace: Duration) -> String { + let _ = self.finished.recv_timeout(grace); + self.buffer + .lock() + .map(|buffer| buffer.clone()) + .unwrap_or_default() + } } /// Spawn the agent against an ALREADY-STARTED proxy and wait for it to @@ -387,8 +452,13 @@ pub fn run_against( source, })?; + // Drain BEFORE waiting, not after: a child that fills the pipe buffer + // blocks in `write` and never reaches the exit the wait is waiting for. + let out_drain = PipeDrain::start(child.stdout.take()); + let err_drain = PipeDrain::start(child.stderr.take()); let (status, timed_out) = wait_to_deadline(&mut child, timeout); - let (stdout, stderr) = read_pipes(&mut child); + let stdout = out_drain.collect(PIPE_DRAIN_GRACE); + let stderr = err_drain.collect(PIPE_DRAIN_GRACE); let log = proxy.log(); let run = AgentRun { @@ -442,8 +512,13 @@ pub fn run_against_contained( source, })?; + // Same ordering as the uncontained path: drain from spawn, so the pipe + // buffer can never be what stops the child from exiting. + let out_drain = PipeDrain::start(child.stdout.take()); + let err_drain = PipeDrain::start(child.stderr.take()); let (status, timed_out) = wait_to_deadline(&mut child, timeout); - let (stdout, stderr) = read_pipes(&mut child); + let stdout = out_drain.collect(PIPE_DRAIN_GRACE); + let stderr = err_drain.collect(PIPE_DRAIN_GRACE); let egress = supervisor.stop_and_collect(); let log = proxy.log(); @@ -476,6 +551,60 @@ pub fn run_against_contained( #[cfg(test)] mod tests { + use super::*; + + /// The 26-minute hang, reduced to its cause. + /// + /// A backgrounded `sleep` inherits the shell's stdout and holds the write + /// end open long after the shell itself exits - structurally the same + /// thing `opencode serve` does when an agent SDK spawns it. EOF therefore + /// never arrives, and the `read_to_string` this replaced waited for EOF, + /// so the drain outlived the process by however long the grandchild ran. + #[cfg(unix)] + #[test] + fn a_grandchild_holding_the_pipe_cannot_outlast_the_grace() { + let mut child = Command::new("sh") + .arg("-c") + .arg("sleep 30 & echo hi") + .stdout(Stdio::piped()) + .spawn() + .expect("spawn"); + let drain = PipeDrain::start(child.stdout.take()); + let _ = child.wait(); + + let started = Instant::now(); + let out = drain.collect(Duration::from_millis(500)); + + assert!( + started.elapsed() < Duration::from_secs(5), + "the drain must be bounded by the grace, not by the grandchild; took {:?}", + started.elapsed() + ); + // Bounded must not mean lossy: what did arrive is the only diagnostic + // a hung run leaves behind. + assert!(out.contains("hi"), "partial output must survive: {out:?}"); + } + + /// The second bug in the same place: draining only AFTER the wait means a + /// child that outwrites the OS pipe buffer (~64 KB) blocks in `write`, + /// never exits, and is killed at the timeout - reported as a hung agent + /// when nothing was wrong with it. Draining from spawn keeps it moving. + #[cfg(unix)] + #[test] + fn a_child_that_outwrites_the_pipe_buffer_still_exits() { + let mut child = Command::new("sh") + .arg("-c") + .arg("yes flowproof | head -c 200000") + .stdout(Stdio::piped()) + .spawn() + .expect("spawn"); + let drain = PipeDrain::start(child.stdout.take()); + let (_status, timed_out) = wait_to_deadline(&mut child, Duration::from_secs(20)); + let out = drain.collect(PIPE_DRAIN_GRACE); + + assert!(!timed_out, "a chatty child must not look like a hung one"); + assert_eq!(out.len(), 200_000, "every byte the child wrote is captured"); + } /// The gap a real adopter hit: their client reads AI_GATEWAY_URL, and /// the proxy's port is not known when the spec is written, so a static diff --git a/crates/flowproof-cli/src/lib.rs b/crates/flowproof-cli/src/lib.rs index 99a2159..ea75ea0 100644 --- a/crates/flowproof-cli/src/lib.rs +++ b/crates/flowproof-cli/src/lib.rs @@ -5,6 +5,7 @@ mod agent_flow; mod capture; use std::path::{Path, PathBuf}; +use std::time::Instant; use clap::{Parser, Subcommand, ValueEnum}; use flowproof_agent::FlowSpec; @@ -687,6 +688,34 @@ fn errored_flow( reports.push(report); } +/// Replay one agent flow inside a suite run, with the suite's hooks. +/// +/// `Err` is a HARNESS fault (a failing seed or cleanup hook); the agent's own +/// verdict comes back as the inner `Result`, so a failing flow is a failing +/// flow rather than a broken suite. Cleanup runs whichever way replay went, +/// matching the ordering the step-replay path uses below. +fn run_agent_flow_in_suite( + spec_path: &Path, + spec: &FlowSpec, + trace_path: &Path, + manifest: &flowproof_agent::SuiteManifest, + json: bool, +) -> Result, String> { + if let Some(cmd) = &manifest.before_each { + run_hook(cmd, spec_path, "before_each")?; + } + // The containment tier prints on every agent run, pass or fail - the + // single-spec path does the same, and a suite must not hide it. + if !json { + println!("{}", agent_flow::containment(spec).report_line()); + } + let outcome = agent_flow::replay(spec, trace_path); + if let Some(cmd) = &manifest.after_each { + run_hook(cmd, spec_path, "after_each")?; + } + Ok(outcome) +} + pub fn run_suite(dir: &Path, json: bool, retries: u8, missing: MissingTrace) -> Result { let mut specs = Vec::new(); discover_specs(dir, &mut specs)?; @@ -807,6 +836,59 @@ pub fn run_suite(dir: &Path, json: bool, retries: u8, missing: MissingTrace) -> } } } + // Agent flows replay their CASSETTE, not the step trace, exactly as + // the single-spec path at `run_one` does. + // + // Without this branch the suite fell through to `load_trace` below, + // which parses a UI trace one JSON object per line. An agent cassette + // is a single `{app, mocks, cassette}` document, so every agent flow + // in a directory run errored with "invalid trace line" - traces that + // `flowproof record` had just written, and that `flowproof run ` + // replayed fine one at a time. Directory mode is what a suite and CI + // invoke, so agent flows were effectively unrunnable there. + if gated_spec.app.id() == "agent" { + let started = Instant::now(); + match run_agent_flow_in_suite(spec_path, &gated_spec, &trace_path, &manifest, json) { + Ok(outcome) => { + let report = flowproof_replay::RunReport::agent( + &gated_spec.name, + outcome.as_ref().err().map(String::as_str), + started.elapsed().as_millis() as u64, + ); + if !json { + match &outcome { + Ok(()) => println!( + "[PASS] {} ({} ms)", + report.name, report.duration_ms + ), + Err(why) => println!("[FAIL] {} — {why}", report.name), + } + } + // Agent flows produce no run bundle, so there is no + // per-flow result path - the suite record below still + // carries the verdict. + flows.push(serde_json::json!({ + "spec": spec_path, + "report": report, + "report_path": null, + })); + reports.push(report); + } + // A hook fault is a harness fault, not a verdict about the + // agent: same treatment every other flow's hook failure gets. + Err(e) => { + errored_flow( + spec_path, + &gated_spec.name, + e, + json, + &mut flows, + &mut reports, + ); + } + } + continue; + } // Seed before the flow; a failing hook fails the flow, not the run. if let Some(cmd) = &manifest.before_each { if let Err(e) = run_hook(cmd, spec_path, "before_each") { diff --git a/crates/flowproof-cli/tests/agent_flow_e2e.rs b/crates/flowproof-cli/tests/agent_flow_e2e.rs index 2066a55..d97a425 100644 --- a/crates/flowproof-cli/tests/agent_flow_e2e.rs +++ b/crates/flowproof-cli/tests/agent_flow_e2e.rs @@ -506,3 +506,46 @@ fn audit_renders_the_control_map_in_yaml_and_json() { std::fs::remove_dir_all(&dir).ok(); } + +/// A recorded agent flow must replay when the suite is run by DIRECTORY, +/// not only when the spec is named directly. +/// +/// This is the gap that made agent flows unusable in CI. `run_suite` had no +/// `app: agent` branch, so it fell through to the step-replay loader, which +/// reads a trace one JSON object per line. An agent cassette is a single +/// `{app, mocks, cassette}` document, so the loader failed on line 1 and +/// every agent flow in the directory errored with "invalid trace line" - +/// traces `flowproof record` had just written, and that `flowproof run +/// ` replayed green one at a time. +/// +/// Directory mode is what a suite, a `pnpm test` script and a CI job all +/// invoke, so this asserts the two modes agree. +#[test] +fn a_recorded_agent_flow_replays_in_directory_mode() { + let _env = lock_env(); + let dir = work_dir("suite-dispatch"); + let agent_py = dir.join("agent.py"); + std::fs::write(&agent_py, FAKE_AGENT).expect("agent"); + let spec = write_spec(&dir, &agent_py); + + std::env::set_var("FLOWPROOF_AGENT_UPSTREAM", fake_model()); + let code = flowproof_cli::run_cli(["record", spec.to_str().expect("utf8")]); + assert_eq!(code, 0, "recording an agent flow should succeed"); + + // No model at all for the replay: a stray real call fails loudly. + std::env::remove_var("FLOWPROOF_AGENT_UPSTREAM"); + std::env::remove_var("OPENAI_BASE_URL"); + + // The single-spec path, which already worked. + let single = flowproof_cli::run_cli(["run", spec.to_str().expect("utf8")]); + assert_eq!(single, 0, "replaying the spec directly must pass"); + + // The DIRECTORY path, which errored before this branch existed. + let suite = flowproof_cli::run_cli(["run", dir.to_str().expect("utf8")]); + assert_eq!( + suite, 0, + "the same flow must replay when the suite is run by directory" + ); + + std::fs::remove_dir_all(&dir).ok(); +} diff --git a/crates/flowproof-replay/src/report.rs b/crates/flowproof-replay/src/report.rs index 89bc676..b672641 100644 --- a/crates/flowproof-replay/src/report.rs +++ b/crates/flowproof-replay/src/report.rs @@ -138,6 +138,40 @@ impl RunReport { } } + /// The report for an `app: agent` flow, whose verdict comes from the + /// cassette replay rather than from steps. + /// + /// `trace_id` is neither "skipped" nor "errored" because the suite counts + /// those two by that field: an agent flow RAN, so it belongs in the + /// ran/passed tally like any other flow. A failure here is a verdict about + /// the agent, not a broken harness - harness faults keep going through + /// [`RunReport::errored`]. + pub fn agent(name: &str, failure: Option<&str>, duration_ms: u64) -> Self { + let passed = failure.is_none(); + Self { + name: name.to_string(), + trace_id: "agent".into(), + passed, + degraded: false, + steps: vec![StepResult { + id: "s0001".into(), + intent: "agent cassette replayed".into(), + status: if passed { + StepStatus::Passed + } else { + StepStatus::Failed + }, + detail: failure.map(str::to_string), + started_ms: 0, + duration_ms, + selector_tier: None, + degraded: false, + }], + duration_ms, + recording: None, + } + } + /// A synthetic report for a flow that never ran (no trace recorded, /// skip condition). `passed: true` — a skip is not a failure, matching /// JUnit semantics — with one skipped step carrying the reason, so the From 37308b75f4f7ae66c13433c1793fc7afd5d5097c Mon Sep 17 00:00:00 2001 From: Amin Chirazi <32016576+AminChirazi@users.noreply.github.com> Date: Tue, 28 Jul 2026 10:26:21 +0400 Subject: [PATCH 2/2] style: rustfmt, and drop a duplicate test import CI runs `cargo fmt --all --check` and `cargo clippy --workspace --all-targets --all-features -- -D warnings`; I had run neither in that exact form. The test module already had its own `use super::*` further down, so the one I added at the top made it redundant. --- crates/flowproof-adapters/src/agent_runner.rs | 4 +--- crates/flowproof-cli/src/lib.rs | 7 +++---- 2 files changed, 4 insertions(+), 7 deletions(-) diff --git a/crates/flowproof-adapters/src/agent_runner.rs b/crates/flowproof-adapters/src/agent_runner.rs index 28cd073..38e7c6b 100644 --- a/crates/flowproof-adapters/src/agent_runner.rs +++ b/crates/flowproof-adapters/src/agent_runner.rs @@ -17,7 +17,7 @@ use std::collections::BTreeMap; use std::io::Read; use std::process::{Command, Stdio}; -use std::sync::{Arc, Mutex, mpsc}; +use std::sync::{mpsc, Arc, Mutex}; use std::time::{Duration, Instant}; use flowproof_trace::cassette::{Cassette, Divergence}; @@ -551,8 +551,6 @@ pub fn run_against_contained( #[cfg(test)] mod tests { - use super::*; - /// The 26-minute hang, reduced to its cause. /// /// A backgrounded `sleep` inherits the shell's stdout and holds the write diff --git a/crates/flowproof-cli/src/lib.rs b/crates/flowproof-cli/src/lib.rs index ea75ea0..d5e2a8f 100644 --- a/crates/flowproof-cli/src/lib.rs +++ b/crates/flowproof-cli/src/lib.rs @@ -857,10 +857,9 @@ pub fn run_suite(dir: &Path, json: bool, retries: u8, missing: MissingTrace) -> ); if !json { match &outcome { - Ok(()) => println!( - "[PASS] {} ({} ms)", - report.name, report.duration_ms - ), + Ok(()) => { + println!("[PASS] {} ({} ms)", report.name, report.duration_ms) + } Err(why) => println!("[FAIL] {} — {why}", report.name), } }