fix(adapters): persist the MCP lane before the agent can see the response

CHANGELOG.md

@@ -6,6 +6,31 @@ together).
 
 ## Unreleased
 
+### Fixed
+
+- **A tool call the agent really made could be missing from the evidence.**
+  0.8.0 fixed the MCP stand-in losing its whole recording when an agent killed
+  it, by persisting the lane after every captured call. It was not enough. The
+  stand-in forwarded the server's response to the agent BEFORE persisting it,
+  and the moment the agent has a response it may kill the stand-in — so the
+  final call of a session was lost while `record` still exited 0.
+
+  Found by the falsifiability suite, with the real server's own request log as
+  the oracle: the server was asked for `tools/call get_weather`, the lane held
+  only `initialize` and `tools/list`, and nothing said so. A lane that silently
+  runs short understates what the agent did, and every assertion evaluated over
+  it inherits the omission — including a guard claim that a tool was never
+  called.
+
+  Capture and persistence now happen BEFORE the response is forwarded. That
+  ordering is the fix: by the time the agent can act on a response, the evidence
+  for it is already on disk.
+
+  And a failed flush is no longer swallowed. With the write at stdin EOF no
+  longer load-bearing, an incremental flush that fails means evidence was lost,
+  so it now fails the record by name instead of minting a partial lane. Evidence
+  lost silently is the one outcome this tool cannot ship.
+
 ### Changed
 
 - **A control that stopped being certifiable passed the gate.**

crates/flowproof-adapters/src/mcp_stdio.rs

@@ -227,36 +227,41 @@ fn run_replay(plan: &McpPlan, out_path: &Path) -> Result<(), String> {
 
 /// Persist the captured lane NOW, so an abrupt shutdown cannot lose it.
 ///
-/// The authoritative write is still the one at stdin EOF, but reaching it
-/// requires the agent to close the stand-in's stdin and let it reap the real
-/// server. An agent that terminates its MCP subprocess abruptly never gets
-/// there - goose does exactly this - and the whole recording was lost, which
-/// then read as "the agent never spawned flowproof's MCP stand-in": a false
-/// negative pointing the adopter at wiring that was already correct.
+/// This is THE durability guarantee, not a best-effort supplement to the write
+/// at stdin EOF. Reaching EOF requires the agent to close the stand-in's stdin
+/// and let it reap the real server; an agent that terminates its MCP
+/// subprocess abruptly never gets there - goose does exactly this.
+///
+/// It runs BEFORE the response is forwarded to the agent. That ordering is the
+/// whole point: once the agent has the response it may kill the stand-in at
+/// any moment, so anything not yet persisted is lost. Forwarding first left a
+/// window in which the final call of a session vanished from the lane while
+/// `record` still exited 0 (#273).
 ///
 /// `write_out_atomic` is a temp-file rename, so a partial flush is never
-/// observable and the last one wins. Errors are swallowed: a failed
-/// incremental flush must not break a run that the EOF write may still save.
+/// observable and the last one wins. Errors are RETURNED rather than swallowed:
+/// with the EOF write no longer load-bearing, a failed flush means evidence was
+/// lost, and evidence lost silently is the one outcome this tool cannot ship.
 fn flush_lane(
     out_path: &std::path::Path,
     calls: &Mutex<BTreeMap<usize, McpCall>>,
     events: &Mutex<Vec<McpServerEvent>>,
-) {
+) -> Result<(), String> {
     let captured: Vec<McpCall> = calls
         .lock()
         .unwrap_or_else(|e| e.into_inner())
         .values()
         .cloned()
         .collect();
     let events: Vec<McpServerEvent> = events.lock().unwrap_or_else(|e| e.into_inner()).clone();
-    let _ = write_out_atomic(
+    write_out_atomic(
         out_path,
         &McpOut {
             calls: captured,
             events,
             ..Default::default()
         },
-    );
+    )
 }
 
 /// RECORD: two threads bridge the agent and the REAL server.
@@ -359,16 +364,12 @@ fn run_record(plan: &McpPlan, out_path: &Path) -> Result<(), String> {
                 if line.trim().is_empty() {
                     continue;
                 }
-                // Forward the server's line to the agent verbatim, so record
-                // is transparent.
-                {
-                    let mut out = write_lock.lock().unwrap_or_else(|e| e.into_inner());
-                    let _ = out.write_all(line.as_bytes());
-                    let _ = out.flush();
-                }
-                // Classify the server line by the JSON-RPC shape. It was
-                // already forwarded verbatim above; here we decide what, if
-                // anything, to CAPTURE from it.
+                // Classify and CAPTURE first, then forward. The order is
+                // load-bearing: the moment the agent has the response it may
+                // kill the stand-in, so anything not yet persisted is lost.
+                // Forwarding first left a window in which the final call of a
+                // session vanished from the lane while `record` still exited 0
+                // (#273).
                 if let Ok(msg) = serde_json::from_str::<Value>(line.trim()) {
                     let has_method = msg.get("method").and_then(Value::as_str);
                     let has_id = msg.get("id");
@@ -413,12 +414,29 @@ fn run_record(plan: &McpPlan, out_path: &Path) -> Result<(), String> {
                                         result,
                                     },
                                 );
-                                flush_lane(&out_path_b, &calls, &events);
+                                if let Err(detail) = flush_lane(&out_path_b, &calls, &events) {
+                                    let mut slot =
+                                        record_error.lock().unwrap_or_else(|e| e.into_inner());
+                                    if slot.is_none() {
+                                        *slot = Some(format!(
+                                            "the MCP lane for this server could not be \
+                                             persisted, so the recording is incomplete: \
+                                             {detail}"
+                                        ));
+                                    }
+                                }
                             }
                         }
                         (None, None) => {}
                     }
                 }
+                // Forwarded verbatim, and only now: record stays transparent
+                // to the agent, but never at the cost of the evidence.
+                {
+                    let mut out = write_lock.lock().unwrap_or_else(|e| e.into_inner());
+                    let _ = out.write_all(line.as_bytes());
+                    let _ = out.flush();
+                }
             }
         })
     };
@@ -475,7 +493,10 @@ fn run_record(plan: &McpPlan, out_path: &Path) -> Result<(), String> {
                     result: result.clone(),
                 },
             );
-            flush_lane(out_path, &calls, &events);
+            // Already in the right order: the mocked answer is persisted
+            // before the agent is told, so this path never raced. It only
+            // needed the failure to stop being invisible.
+            flush_lane(out_path, &calls, &events)?;
             let mut out = write_lock.lock().unwrap_or_else(|e| e.into_inner());
             write_result(&mut *out, &id, &result)?;
         } else {

crates/flowproof-cli/tests/falsifiability_mcp_kill.rs

@@ -0,0 +1,203 @@
+//! Falsifiability proof for the MCP stand-in's recording durability (#257).
+//!
+//! Before 0.8.0 the MCP lane was written only at stdin EOF. An agent that
+//! terminated its stand-in abruptly never got there, and the ENTIRE recording
+//! was lost. The fix persists the lane after every captured call, atomically.
+//!
+//! Nothing exercised it. Every MCP test in the suite closes stdin politely and
+//! waits -- exactly the path the defect did not affect -- so all of them would
+//! stay green if the fix were reverted tomorrow. That is the shape of an
+//! unguarded fix: the bug is gone, and nothing would notice its return.
+//!
+//! The violating input is one hostile agent
+//! (`tests/falsifiability/fixtures/mcp-killing-agent.py`), identical to the
+//! suite's polite one but for its last act: `proc.kill()` instead of
+//! `stdin.close(); wait()`.
+//!
+//! Two layers, and the second is the one that matters. A trace merely EXISTING
+//! proves nothing -- the model cassette would still be written even with the
+//! MCP lane empty, which is precisely what the defect produced. So the test
+//! reads the lane and requires the captured call to be in it.
+
+use std::io::{Read, Write};
+use std::net::TcpListener;
+use std::path::{Path, PathBuf};
+
+const FLOWPROOF_BIN: &str = env!("CARGO_BIN_EXE_flowproof");
+
+/// The deterministic stdio MCP server, logging every request it receives.
+const FAKE_MCP_SERVER: &str = r#"
+import json, sys
+log = open(sys.argv[1], "a")
+for line in sys.stdin:
+    line = line.strip()
+    if not line:
+        continue
+    msg = json.loads(line)
+    log.write(line + "\n"); log.flush()
+    mid = msg.get("id")
+    if mid is None:
+        continue
+    method = msg.get("method")
+    if method == "initialize":
+        result = {"protocolVersion": "2024-11-05",
+                  "serverInfo": {"name": "weather", "version": "1"},
+                  "capabilities": {"tools": {}}}
+    elif method == "tools/list":
+        result = {"tools": [{"name": "get_weather"}]}
+    elif method == "tools/call":
+        result = {"content": [{"type": "text", "text": "REAL:" + msg["params"]["name"]}],
+                  "isError": False}
+    else:
+        result = {}
+    sys.stdout.write(json.dumps({"jsonrpc": "2.0", "id": mid, "result": result}) + "\n")
+    sys.stdout.flush()
+"#;
+
+fn fixture() -> PathBuf {
+    Path::new(env!("CARGO_MANIFEST_DIR"))
+        .join("../../tests/falsifiability/fixtures/mcp-killing-agent.py")
+}
+
+fn work_dir(name: &str) -> PathBuf {
+    let dir = std::env::temp_dir().join(format!("flowproof-fals-mcpkill-{name}"));
+    std::fs::remove_dir_all(&dir).ok();
+    std::fs::create_dir_all(&dir).expect("work dir");
+    dir
+}
+
+fn read_http_request(stream: &mut std::net::TcpStream) -> String {
+    let mut buf = Vec::new();
+    let mut chunk = [0u8; 1024];
+    while let Ok(n) = stream.read(&mut chunk) {
+        if n == 0 {
+            break;
+        }
+        buf.extend_from_slice(&chunk[..n]);
+        let text = String::from_utf8_lossy(&buf).to_string();
+        let Some(head_end) = text.find("\r\n\r\n") else {
+            continue;
+        };
+        let len = text
+            .lines()
+            .find_map(|l| {
+                let (k, v) = l.split_once(':')?;
+                k.eq_ignore_ascii_case("content-length")
+                    .then(|| v.trim().parse::<usize>().ok())?
+            })
+            .unwrap_or(0);
+        if buf.len() >= head_end + 4 + len {
+            break;
+        }
+    }
+    String::from_utf8_lossy(&buf).to_string()
+}
+
+/// Replies `done` to everything: enough to satisfy the model-boundary guard.
+/// The MCP boundary is what is under test here, not the model one.
+fn fake_model() -> String {
+    let listener = TcpListener::bind("127.0.0.1:0").expect("bind");
+    let port = listener.local_addr().expect("addr").port();
+    std::thread::spawn(move || {
+        for stream in listener.incoming().take(16) {
+            let Ok(mut stream) = stream else { continue };
+            let _ = read_http_request(&mut stream);
+            let reply = serde_json::json!({
+                "choices": [{"index": 0, "finish_reason": "stop",
+                    "message": {"role": "assistant", "content": "done"}}]
+            })
+            .to_string();
+            let _ = stream.write_all(
+                format!(
+                    "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\n\
+                     content-length: {}\r\nconnection: close\r\n\r\n{reply}",
+                    reply.len()
+                )
+                .as_bytes(),
+            );
+            let _ = stream.flush();
+            let _ = stream.shutdown(std::net::Shutdown::Write);
+        }
+    });
+    format!("http://127.0.0.1:{port}/v1")
+}
+
+/// An agent that kills its stand-in must not take the recording with it.
+#[test]
+fn an_agent_that_kills_the_stand_in_still_leaves_a_recorded_lane() {
+    let dir = work_dir("kill");
+    let agent_py = dir.join("agent.py");
+    let server_py = dir.join("server.py");
+    let log = dir.join("server.log");
+    std::fs::copy(fixture(), &agent_py).expect("stage the hostile agent");
+    std::fs::write(&server_py, FAKE_MCP_SERVER).expect("server");
+
+    let spec = dir.join("weather.flow.yaml");
+    std::fs::write(
+        &spec,
+        format!(
+            "name: MCP weather, agent kills the stand-in\n\
+             app: agent\n\
+             agent:\n  command: python3 {agent}\n\
+             mcp:\n  - name: weather\n    command: python3 {server} {log}\n\
+             steps:\n\
+             \x20 - prompt: What is the weather?\n\
+             \x20 - assert: reply contains done\n",
+            agent = agent_py.display(),
+            server = server_py.display(),
+            log = log.display(),
+        ),
+    )
+    .expect("spec");
+
+    let out = std::process::Command::new(FLOWPROOF_BIN)
+        .arg("record")
+        .arg(&spec)
+        .current_dir(&dir)
+        .env("FLOWPROOF_AGENT_UPSTREAM", fake_model())
+        .env("FLOWPROOF_AGENT_KEY", "not-a-real-key")
+        .env("FLOWPROOF_MCP_EXE", FLOWPROOF_BIN)
+        .output()
+        .expect("record with the hostile agent");
+
+    let stdout = String::from_utf8_lossy(&out.stdout).to_string();
+    let stderr = String::from_utf8_lossy(&out.stderr).to_string();
+    assert!(
+        out.status.success(),
+        "an agent that kills its stand-in is rude, not wrong: the record must \
+         still succeed. stdout={stdout} stderr={stderr}"
+    );
+
+    // The real server WAS asked -- otherwise the exchange never happened and
+    // the lane would be legitimately empty, proving nothing.
+    let logged = std::fs::read_to_string(&log).expect("server log readable");
+    assert!(
+        logged.contains("get_weather"),
+        "the exchange must actually have happened, or an empty lane is \
+         vacuously 'preserved': {logged}"
+    );
+
+    // Layer 2: the lane survived the kill. A trace merely existing proves
+    // nothing -- the model cassette is written regardless, and an empty MCP
+    // lane beside it is exactly what the pre-0.8.0 defect produced.
+    let trace = dir.join("weather.trace.jsonl");
+    let contents = std::fs::read_to_string(&trace).expect("trace readable");
+    for needle in [
+        "\"mcp\"",
+        "\"weather\"",
+        "initialize",
+        "tools/call",
+        "get_weather",
+    ] {
+        assert!(
+            contents.contains(needle),
+            "the MCP lane must survive an abrupt kill, missing {needle}: {contents}"
+        );
+    }
+    assert!(
+        contents.contains("REAL:get_weather"),
+        "and the captured RESULT survived too, not just the request: {contents}"
+    );
+
+    std::fs::remove_dir_all(&dir).ok();
+}