diff --git a/crates/flowproof-adapters/src/agent_runner.rs b/crates/flowproof-adapters/src/agent_runner.rs index 51b73fe..38e7c6b 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::{mpsc, Arc, Mutex}; 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,58 @@ pub fn run_against_contained( #[cfg(test)] mod tests { + /// 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..d5e2a8f 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,58 @@ 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