wip: fail conformance recording on incomplete runs

Merged integrated 5ef2e3a50 baseline before refactoring (afe46fca0). Events require TurnComplete and joined shutdown; MCP boot/catalog/call/shutdown and SSE replay carry unconditional deadline failures, never recordable timeout goldens. Reject empty families/cases, remove stale C02-05 exemption, and compare golden/error/observer output bytes exactly. Added real Engine, loopback MCP, and loopback SSE stalled-run regression controls plus deliberate byte drift checks. Source batch complete; governed focused build and golden qualification are pending. Rustfmt and git diff --check passed. No hosted/platform/provider/release proof claimed.

Signed-off-by: CodeWhale Bot <[email protected]>

crates/tui/src/conformance.rs

@@ -25,8 +25,8 @@
 //! must reproduce.
 //!
 //! Known limits, stated so nobody assumes them: the harness pins what the
-//! Rust side does today, including the one behavior an audit slice is
-//! changing (see `events/provider_error_after_tool_call.case.json`); it runs
+//! Rust side does today. A provider error after a collected tool call must
+//! settle it without executing it (C02-05); the invariant is mandatory. It runs
 //! no real provider, MCP server binary, or Node host. The `events`, `prompt`
 //! and `hooks` goldens were recorded on Unix and are compiled only there:
 //! hook fixtures are POSIX shell, and a Windows turn differs in shell and

crates/tui/src/conformance/events.rs

@@ -81,6 +81,10 @@ pub(super) struct ScriptedProvider {
 
 impl ScriptedProvider {
     pub(super) fn from_script(model: &str, script: &Value) -> Self {
+        assert!(
+            script.as_array().is_some_and(|steps| !steps.is_empty()),
+            "provider_script must contain at least one response"
+        );
         let steps = script
             .as_array()
             .expect("provider_script is an array")
@@ -251,7 +255,7 @@ fn workspace_listing(workspace: &Path) -> BTreeMap<String, String> {
 
 pub(super) struct TurnRecord {
     pub(super) events: Vec<Event>,
-    pub(super) timed_out: bool,
+    pub(super) failure: Option<String>,
 }
 
 /// Drive one scripted turn inside `sandbox` on a fresh current-thread
@@ -261,6 +265,15 @@ pub(super) fn run_scripted_turn(
     sandbox: &Sandbox,
     case: &Value,
     script: &Value,
+) -> (TurnRecord, std::sync::Arc<ScriptedProvider>) {
+    run_scripted_turn_with_deadline(sandbox, case, script, event_deadline())
+}
+
+fn run_scripted_turn_with_deadline(
+    sandbox: &Sandbox,
+    case: &Value,
+    script: &Value,
+    deadline: Duration,
 ) -> (TurnRecord, std::sync::Arc<ScriptedProvider>) {
     let model = case["model"].as_str().expect("case.model");
     let provider = std::sync::Arc::new(ScriptedProvider::from_script(model, script));
@@ -281,7 +294,7 @@ pub(super) fn run_scripted_turn(
         let client: crate::core::model_client::SharedModelClient = provider.clone();
         let (engine, handle) = Engine::new_with_model_client(engine_config, &config, client);
         let op = send_message_op(case, &config);
-        drive_turn(engine, handle, op, &driver).await
+        drive_turn(engine, handle, op, &driver, deadline).await
     });
     drop(runtime);
     (record, provider)
@@ -294,22 +307,35 @@ async fn drive_turn(
     handle: crate::core::engine::EngineHandle,
     op: Op,
     driver: &[Value],
+    deadline: Duration,
 ) -> TurnRecord {
-    let run = tokio::spawn(engine.run());
+    let mut run = tokio::spawn(engine.run());
     handle.send(op).await.expect("send conformance turn");
     let rx = handle.rx_event.clone();
     let ids = protocol_ids();
     let mut events = Vec::new();
     let mut fired = vec![false; driver.len()];
-    let mut timed_out = false;
+    let mut failure = None;
+    let mut completed = false;
+    let expires = tokio::time::Instant::now() + deadline;
     loop {
-        let next =
-            tokio::time::timeout(event_deadline(), async { rx.write().await.recv().await }).await;
+        if tokio::time::Instant::now() >= expires {
+            failure = Some(format!(
+                "harness timeout: engine turn did not complete within {deadline:?}"
+            ));
+            break;
+        }
+        let next = golden::complete_within(
+            "engine turn",
+            expires.saturating_duration_since(tokio::time::Instant::now()),
+            async { rx.write().await.recv().await },
+        )
+        .await;
         let event = match next {
             Ok(Some(event)) => event,
             Ok(None) => break,
-            Err(_) => {
-                timed_out = true;
+            Err(error) => {
+                failure = Some(error);
                 break;
             }
         };
@@ -338,16 +364,45 @@ async fn drive_turn(
         let terminal = matches!(event, Event::TurnComplete { .. });
         events.push(event);
         if terminal {
+            completed = true;
             break;
         }
     }
-    let _ = handle.send(Op::Shutdown).await;
-    let _ = tokio::time::timeout(event_deadline(), run).await;
+    if !completed && failure.is_none() {
+        failure = Some("engine closed before TurnComplete".to_string());
+    }
+    handle.cancel();
+    match golden::complete_within(
+        "engine shutdown request",
+        deadline,
+        handle.send(Op::Shutdown),
+    )
+    .await
+    {
+        Ok(Ok(())) => {}
+        Ok(Err(error)) => {
+            failure.get_or_insert_with(|| format!("engine shutdown request failed: {error}"));
+        }
+        Err(error) => {
+            failure.get_or_insert(error);
+        }
+    }
+    match golden::complete_within("engine shutdown", deadline, &mut run).await {
+        Ok(Ok(_)) => {}
+        Ok(Err(error)) => {
+            failure.get_or_insert_with(|| format!("engine task failed: {error}"));
+        }
+        Err(error) => {
+            failure.get_or_insert(error);
+            run.abort();
+            let _ = run.await;
+        }
+    }
     let mut rx = rx.write().await;
     while let Ok(event) = rx.try_recv() {
         events.push(event);
     }
-    TurnRecord { events, timed_out }
+    TurnRecord { events, failure }
 }
 
 fn rule_matches(when: &Value, projected: &Value) -> bool {
@@ -447,44 +502,31 @@ fn check_invariants(case: &Value, workspace: &Path, provider: &ScriptedProvider)
 }
 
 fn run_case(name: &str, case: &Value, failures: &mut Failures) {
+    run_case_with_deadline(name, case, failures, event_deadline());
+}
+
+fn run_case_with_deadline(name: &str, case: &Value, failures: &mut Failures, deadline: Duration) {
     let sandbox = Sandbox::new(case);
     let workspace = sandbox.workspace.clone();
-    let (record, provider) = run_scripted_turn(&sandbox, case, &case["provider_script"]);
+    let (record, provider) =
+        run_scripted_turn_with_deadline(&sandbox, case, &case["provider_script"], deadline);
+    if let Some(error) = record.failure {
+        failures.push(name, error);
+        return;
+    }
     let mut masker = sandbox.masker(VOLATILE_KEYS);
     let mut lines = normalize_events(&record.events, &mut masker);
-    let mut summary = json!({
+    let summary = json!({
         "model_requests": provider.stream_requests(),
         "non_streaming_requests": provider.other_requests(),
         "workspace_after": workspace_listing(&workspace),
     });
-    if record.timed_out {
-        summary["harness_timeout"] =
-            json!(format!("no engine event within {:?}", event_deadline()));
-    }
     lines.push(json!({ "harness_summary": summary }));
 
     let violations = check_invariants(case, &workspace, &provider);
-    let expected_change = case.get("expected_to_change");
-    match (violations.is_empty(), expected_change) {
-        (false, None) => {
-            failures.push(name, format!("invariant broken: {}", violations.join("; ")))
-        }
-        (false, Some(change)) => eprintln!(
-            "conformance: `{name}` pins current behavior that {} is changing ({}): {}",
-            change["owner"].as_str().unwrap_or("an audit slice"),
-            change["finding"].as_str().unwrap_or("unnamed finding"),
-            violations.join("; ")
-        ),
-        (true, Some(change)) if !golden::update_mode() => failures.push(
-            name,
-            format!(
-                "the invariant now holds, so {} has landed: re-record this golden with {}=1 \
-                 and delete `expected_to_change` from the case",
-                change["finding"].as_str().unwrap_or("the expected change"),
-                golden::UPDATE_ENV
-            ),
-        ),
-        _ => {}
+    if !violations.is_empty() {
+        failures.push(name, format!("invariant broken: {}", violations.join("; ")));
+        return;
     }
 
     failures.record(
@@ -496,6 +538,25 @@ fn run_case(name: &str, case: &Value, failures: &mut Failures) {
     );
 }
 
+#[test]
+fn harness_timeout_rejects_a_real_stalled_turn() {
+    let mut case = golden::read_case(FAMILY, "plain_answer");
+    case["provider_script"][0]["then"] = json!("hang");
+    case["provider_script"][0]["events"] = json!([]);
+    let mut failures = Failures::default();
+    // This is the same recording path as normal cases. In update mode it must
+    // still fail and must not create this deliberately absent golden.
+    let name = "harness_timeout_control";
+    let path = golden::family_dir(FAMILY).join(format!("{name}.golden.jsonl"));
+    assert!(!path.exists());
+    run_case_with_deadline(name, &case, &mut failures, Duration::from_secs(1));
+    assert!(failures.contains("harness timeout"));
+    assert!(
+        !path.exists(),
+        "an incomplete turn was recorded as a golden"
+    );
+}
+
 #[test]
 fn golden_turn_events_match() {
     let names = golden::case_names(FAMILY);

crates/tui/src/conformance/golden.rs

@@ -64,13 +64,12 @@ pub(super) fn update_mode() -> bool {
 /// instead. Returns a failure description rather than panicking so a family
 /// can report every drifted case in one run.
 pub(super) fn check_golden(path: &Path, actual: &str) -> Result<(), String> {
-    let expected = std::fs::read_to_string(path)
-        .ok()
-        .map(|text| text.replace("\r\n", "\n"));
+    let update = update_mode();
+    let expected = std::fs::read_to_string(path).ok();
     if expected.as_deref() == Some(actual) {
         return Ok(());
     }
-    if update_mode() {
+    if update {
         if let Some(parent) = path.parent() {
             std::fs::create_dir_all(parent).expect("create golden dir");
         }
@@ -91,6 +90,18 @@ pub(super) fn check_golden(path: &Path, actual: &str) -> Result<(), String> {
     ))
 }
 
+/// A harness deadline is missing evidence, never a provider outcome or a
+/// recordable golden. Keep this boundary shared by every asynchronous family.
+pub(super) async fn complete_within<T>(
+    operation: &str,
+    deadline: std::time::Duration,
+    future: impl std::future::Future<Output = T>,
+) -> Result<T, String> {
+    tokio::time::timeout(deadline, future)
+        .await
+        .map_err(|_| format!("harness timeout: {operation} did not complete within {deadline:?}"))
+}
+
 /// Line-oriented first difference with a little context — enough to see what
 /// moved without dumping a whole transcript into the failure.
 fn first_difference(expected: &str, actual: &str) -> String {
@@ -266,8 +277,8 @@ impl Masker {
     }
 }
 
-/// Stable snake_case name of a `ToolError` variant; the variant, not the
-/// message wording, is the cross-implementation contract.
+/// Stable snake_case name of a `ToolError` variant. The golden also pins its
+/// full detail bytes; migration does not weaken either part of the contract.
 pub(super) fn tool_error_kind(error: &ToolError) -> &'static str {
     match error {
         ToolError::InvalidInput { .. } => "invalid_input",
@@ -357,6 +368,9 @@ pub(super) fn write_workspace(workspace: &Path, case: &Value) {
 pub(super) struct Failures(Vec<String>);
 
 impl Failures {
+    pub(super) fn contains(&self, message: &str) -> bool {
+        self.0.iter().any(|failure| failure.contains(message))
+    }
     pub(super) fn record(&mut self, case: &str, result: Result<(), String>) {
         if let Err(message) = result {
             self.0.push(format!("[{case}] {message}"));
@@ -368,6 +382,7 @@ impl Failures {
     }
 
     pub(super) fn finish(self, family: &str, cases: usize) {
+        assert!(cases > 0, "conformance family `{family}` ran no cases");
         assert!(
             self.0.is_empty(),
             "conformance family `{family}`: {} of {cases} case(s) failed\n\n{}",
@@ -377,3 +392,29 @@ impl Failures {
         eprintln!("conformance family `{family}`: {cases} case(s) matched");
     }
 }
+
+#[test]
+fn golden_comparison_rejects_changed_bytes_and_missing_output() {
+    let _lock = lock_test_env();
+    let _update = EnvVarGuard::set(UPDATE_ENV, "0");
+    let dir = tempfile::tempdir().expect("temporary golden");
+    let path = dir.path().join("case.golden.txt");
+    std::fs::write(&path, "expected\n").expect("write golden");
+    assert!(check_golden(&path, "expected\n").is_ok());
+    assert!(
+        check_golden(&path, "different\n")
+            .unwrap_err()
+            .contains("golden drift")
+    );
+    assert!(check_golden(&path, "expected\r\n").is_err());
+    assert!(check_golden(&dir.path().join("missing"), "").is_err());
+    // A CRLF golden must not be silently normalized either.
+    std::fs::write(&path, "expected\r\n").expect("write CRLF golden");
+    assert!(check_golden(&path, "expected\n").is_err());
+}
+
+#[test]
+#[should_panic(expected = "ran no cases")]
+fn empty_family_cannot_pass() {
+    Failures::default().finish("empty_control", 0);
+}

crates/tui/src/conformance/hooks.rs

@@ -124,6 +124,12 @@ fn tool_outcome(case: &Value) -> Result<ToolResult, ToolError> {
 }
 
 fn run_case(name: &str, case: &Value, failures: &mut Failures) {
+    assert!(
+        case["hooks"]
+            .as_array()
+            .is_some_and(|hooks| !hooks.is_empty()),
+        "hook case must contain a hook"
+    );
     let sandbox = Sandbox::new(case);
     let capture = sandbox.workspace.join(".conformance-capture");
     std::fs::create_dir_all(&capture).expect("capture dir");
@@ -178,7 +184,9 @@ fn run_case(name: &str, case: &Value, failures: &mut Failures) {
                 "success": result.success,
                 "exit_code": result.exit_code,
                 "background": result.background,
-                "stdout": result.stdout.trim(),
+                "stdout": result.stdout,
+                "stderr": result.stderr,
+                "error": result.error,
             })).collect::<Vec<_>>() })
         }
         other => panic!("unknown hook event `{other}`"),
@@ -196,6 +204,10 @@ fn run_case(name: &str, case: &Value, failures: &mut Failures) {
         })
         .collect();
     let mut golden_value = json!({ "outcome": outcome, "hooks": hooks });
+    if !hooks.iter().any(|hook| hook["ran"] == true) {
+        failures.push(name, "no configured hook produced a capture");
+        return;
+    }
     let mut masker = sandbox
         .masker(&[])
         .literal(executor.session_id(), "<HOOK_SESSION>");

crates/tui/src/conformance/mcp.rs

@@ -16,7 +16,7 @@
 //!
 //! Normalization: the server URL/port is masked; results are
 //! `{"ok": {success, content, metadata, content_blocks}}` with JSON content
-//! parsed, or `{"err": {kind, detail}}` where `detail` is informative text.
+//! parsed, or `{"err": {kind, detail}}` with exact detail bytes.
 //! `server_received` lists the side-effecting requests the server saw
 //! (`tools/call`, `resources/read`, `prompts/get`) — a call that must not be
 //! sent, or must not be replayed, shows up there.
@@ -62,6 +62,28 @@ trait McpDispatchUnderTest: Send + Sync {
     async fn shutdown(&self);
 }
 
+#[tokio::test]
+async fn harness_timeout_rejects_a_real_unanswered_mcp_call() {
+    let mut case = golden::read_case(FAMILY, "tools_resources_prompts");
+    let held = case["steps"]
+        .as_array()
+        .expect("steps")
+        .iter()
+        .find(|step| step["cancel"] == "after_server_holds")
+        .expect("held-call fixture")
+        .clone();
+    let mut held = held;
+    held.as_object_mut().expect("step").remove("cancel");
+    case["steps"] = json!([held]);
+    let _sandbox = Sandbox::new(&Value::Null);
+    let outcome =
+        run_transcript_with_deadline(&case, McpPoolDispatch::boxed, Duration::from_secs(1)).await;
+    assert!(
+        matches!(outcome, Err(error) if error.contains("harness timeout: MCP call")),
+        "an unanswered MCP call became recordable output"
+    );
+}
+
 type DispatchFactory = fn(config: McpConfig) -> Box<dyn McpDispatchUnderTest>;
 
 /// Every dispatch runs every transcript against the same golden.
@@ -159,13 +181,18 @@ impl TranscriptServer {
         let spec = Arc::new(spec);
         let (log, notify) = (Arc::clone(&received), Arc::clone(&held));
         let task = tokio::spawn(async move {
+            let mut connections = tokio::task::JoinSet::new();
             loop {
-                let Ok((socket, _)) = listener.accept().await else {
+                let accepted = tokio::select! {
+                    accepted = listener.accept() => accepted,
+                    _ = connections.join_next(), if !connections.is_empty() => continue,
+                };
+                let Ok((socket, _)) = accepted else {
                     break;
                 };
                 let (spec, log, notify) =
                     (Arc::clone(&spec), Arc::clone(&log), Arc::clone(&notify));
-                tokio::spawn(async move {
+                connections.spawn(async move {
                     answer(socket, &spec, &log, &notify).await;
                 });
             }
@@ -342,7 +369,22 @@ fn normalize_call(result: &Result<RichToolResult, ToolError>) -> Value {
     }
 }
 
-async fn run_transcript(case: &Value, factory: DispatchFactory) -> (Value, TranscriptServer) {
+async fn run_transcript(
+    case: &Value,
+    factory: DispatchFactory,
+) -> Result<(Value, TranscriptServer), String> {
+    run_transcript_with_deadline(case, factory, CALL_DEADLINE).await
+}
+
+async fn run_transcript_with_deadline(
+    case: &Value,
+    factory: DispatchFactory,
+    deadline: Duration,
+) -> Result<(Value, TranscriptServer), String> {
+    let scripted_steps = case["steps"].as_array().expect("case.steps");
+    if scripted_steps.is_empty() {
+        return Err("MCP transcript has no host steps".to_string());
+    }
     let server = TranscriptServer::start(case["server"].clone()).await;
     let server_name = case["server_name"].as_str().unwrap_or("conformance");
     let config: McpConfig = serde_json::from_value(json!({
@@ -355,15 +397,18 @@ async fn run_transcript(case: &Value, factory: DispatchFactory) -> (Value, Trans
     .expect("transcript MCP config");
     let dispatch = factory(config);
     let mut steps = Vec::new();
-    let boot = match dispatch.boot().await {
+    let boot = match golden::complete_within("MCP boot", deadline, dispatch.boot()).await? {
         Ok(()) => json!("ok"),
         Err(detail) => json!({ "err": { "detail": detail } }),
     };
     steps.push(json!({ "op": "boot", "outcome": boot }));
-    for step in case["steps"].as_array().expect("case.steps") {
+    for step in scripted_steps {
         match step["op"].as_str().expect("step.op") {
             "catalog" => {
-                let catalog = serde_json::to_value(dispatch.catalog().await).expect("catalog");
+                let catalog = serde_json::to_value(
+                    golden::complete_within("MCP catalog", deadline, dispatch.catalog()).await?,
+                )
+                .expect("catalog");
                 steps.push(json!({ "op": "catalog", "tools": catalog }));
             }
             "call" => {
@@ -377,31 +422,27 @@ async fn run_transcript(case: &Value, factory: DispatchFactory) -> (Value, Trans
                             cancel.cancel();
                         })
                     });
-                let outcome = tokio::time::timeout(
-                    CALL_DEADLINE,
+                let outcome = golden::complete_within(
+                    "MCP call",
+                    deadline,
                     dispatch.call(tool, step["input"].clone(), cancel),
                 )
                 .await;
                 if let Some(canceller) = canceller {
                     canceller.abort();
                 }
-                let outcome = match outcome {
-                    Ok(result) => normalize_call(&result),
-                    Err(_) => {
-                        json!({ "harness_timeout": format!("no answer within {CALL_DEADLINE:?}") })
-                    }
-                };
+                let outcome = normalize_call(&outcome?);
                 steps.push(json!({ "op": "call", "tool": tool, "outcome": outcome }));
             }
             other => panic!("unknown transcript op `{other}`"),
         }
     }
-    dispatch.shutdown().await;
+    golden::complete_within("MCP shutdown", deadline, dispatch.shutdown()).await?;
     let received = server.received.lock().expect("log").clone();
-    (
+    Ok((
         json!({ "steps": steps, "server_received": received }),
         server,
-    )
+    ))
 }
 
 #[test]
@@ -418,7 +459,13 @@ fn mcp_transcripts_match_goldens_for_every_dispatch() {
                 .enable_all()
                 .build()
                 .expect("runtime");
-            let (mut outcome, server) = runtime.block_on(run_transcript(&case, *factory));
+            let (mut outcome, server) = match runtime.block_on(run_transcript(&case, *factory)) {
+                Ok(completed) => completed,
+                Err(error) => {
+                    failures.push(&format!("{name} via {dispatch}"), error);
+                    continue;
+                }
+            };
             let mut masker = sandbox
                 .masker(&[])
                 .literal(&server.url, "<SERVER_URL>")

crates/tui/src/conformance/prompt.rs

@@ -85,8 +85,8 @@ fn capture_prefix(name: &str, case: &Value) -> Result<Prefix, String> {
         .cloned()
         .unwrap_or_else(default_script);
     let (record, provider) = run_scripted_turn(&sandbox, case, &script);
-    if record.timed_out {
-        return Err("the turn did not complete".to_string());
+    if let Some(error) = record.failure {
+        return Err(error);
     }
     let requests = provider.captured();
     let first = requests

crates/tui/src/conformance/sse.rs

@@ -6,7 +6,7 @@
 //! wire adapter yields. The first golden line is the request the adapter
 //! sent (method and path only — the body is the prompt family's business);
 //! every later line is one normalized [`super::stream_json`] event, or one
-//! `stream_failure` / `open_failure` record whose `detail` is informative.
+//! `stream_failure` / `open_failure` record with exact `detail` bytes.
 
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
@@ -33,6 +33,18 @@ struct ReceivedRequest {
     path: String,
 }
 
+#[tokio::test]
+async fn harness_timeout_rejects_a_real_stalled_sse_connection() {
+    let mut case = golden::read_case(FAMILY, "anthropic_thinking_text_usage");
+    case["hold_open"] = json!(true);
+    case["framing"] = json!("chunked");
+    let outcome = replay_with_deadline(&case, Vec::new(), Duration::from_secs(1)).await;
+    assert!(
+        matches!(outcome, Err(error) if error.contains("harness timeout: SSE replay")),
+        "a stalled SSE connection became recordable output"
+    );
+}
+
 /// One-shot-per-connection loopback server that answers every request with
 /// the recording. `close` framing ends the body by closing the socket (a
 /// provider that stops sending); `chunked` framing uses HTTP chunks, and
@@ -55,6 +67,7 @@ struct Framing {
     chunked: bool,
     transport_cut: bool,
     chunk_bytes: usize,
+    hold_open: bool,
 }
 
 impl LoopbackRecording {
@@ -66,14 +79,19 @@ impl LoopbackRecording {
         let requests = Arc::new(Mutex::new(Vec::new()));
         let captured = Arc::clone(&requests);
         let task = tokio::spawn(async move {
+            let mut connections = tokio::task::JoinSet::new();
             loop {
-                let Ok((socket, _)) = listener.accept().await else {
+                let accepted = tokio::select! {
+                    accepted = listener.accept() => accepted,
+                    _ = connections.join_next(), if !connections.is_empty() => continue,
+                };
+                let Ok((socket, _)) = accepted else {
                     break;
                 };
                 let body = body.clone();
                 let framing = framing.clone();
                 let captured = Arc::clone(&captured);
-                tokio::spawn(async move {
+                connections.spawn(async move {
                     serve(socket, body, framing, captured).await;
                 });
             }
@@ -153,6 +171,11 @@ async fn serve(
         // Give the client a chance to observe a split line before the rest.
         tokio::task::yield_now().await;
     }
+    if framing.hold_open {
+        // A stalled provider for the harness timeout regression. The listener
+        // owns this task, so dropping the replay cancels the open socket too.
+        std::future::pending::<()>().await;
+    }
     if framing.chunked && !framing.transport_cut {
         let _ = socket.write_all(b"0\r\n\r\n").await;
     }
@@ -250,11 +273,20 @@ fn request_for_case(case: &Value) -> MessageRequest {
     }
 }
 
-async fn replay(case: &Value, recording: Vec<u8>) -> Vec<Value> {
+async fn replay(case: &Value, recording: Vec<u8>) -> Result<Vec<Value>, String> {
+    replay_with_deadline(case, recording, STREAM_DEADLINE).await
+}
+
+async fn replay_with_deadline(
+    case: &Value,
+    recording: Vec<u8>,
+    deadline: Duration,
+) -> Result<Vec<Value>, String> {
     let framing = Framing {
         chunked: case["framing"].as_str() == Some("chunked"),
         transport_cut: case["transport_cut"].as_bool().unwrap_or(false),
         chunk_bytes: case["chunk_bytes"].as_u64().unwrap_or(0) as usize,
+        hold_open: case["hold_open"].as_bool().unwrap_or(false),
     };
     let server = LoopbackRecording::start(recording, framing).await;
     let route = case["route"].as_str().expect("case.route");
@@ -268,7 +300,7 @@ async fn replay(case: &Value, recording: Vec<u8>) -> Vec<Value> {
     };
 
     let mut lines = Vec::new();
-    let outcome = tokio::time::timeout(STREAM_DEADLINE, async {
+    let events = golden::complete_within("SSE replay", deadline, async {
         let mut events = Vec::new();
         match client.create_message_stream(request_for_case(case)).await {
             Err(error) => events.push(json!({
@@ -289,13 +321,10 @@ async fn replay(case: &Value, recording: Vec<u8>) -> Vec<Value> {
         }
         events
     })
-    .await;
-    let events = outcome.unwrap_or_else(|_| {
-        vec![json!({
-            "type": "harness_timeout",
-            "detail": format!("stream did not end within {STREAM_DEADLINE:?}"),
-        })]
-    });
+    .await?;
+    if events.is_empty() {
+        return Err("SSE replay produced no event or provider error".to_string());
+    }
 
     let requests = server.requests.lock().expect("request log").clone();
     lines.push(json!({
@@ -305,7 +334,7 @@ async fn replay(case: &Value, recording: Vec<u8>) -> Vec<Value> {
             .collect::<Vec<_>>(),
     }));
     lines.extend(events);
-    lines
+    Ok(lines)
 }
 
 #[tokio::test]
@@ -320,7 +349,13 @@ async fn recorded_sse_streams_match_goldens() {
             .map_or_else(|| format!("{name}.sse"), str::to_string);
         let recording = std::fs::read(dir.join(&recording_name))
             .unwrap_or_else(|error| panic!("read {recording_name}: {error}"));
-        let mut lines = replay(&case, recording).await;
+        let mut lines = match replay(&case, recording).await {
+            Ok(lines) => lines,
+            Err(error) => {
+                failures.push(name, error);
+                continue;
+            }
+        };
         let mut masker = golden::Masker::new(&[]);
         for line in &mut lines {
             masker.value(line);
@@ -329,7 +364,7 @@ async fn recorded_sse_streams_match_goldens() {
         // the events family can feed these goldens straight back in.
         for line in lines.iter().skip(1) {
             let kind = line["type"].as_str().unwrap_or_default();
-            if matches!(kind, "stream_failure" | "open_failure" | "harness_timeout") {
+            if matches!(kind, "stream_failure" | "open_failure") {
                 continue;
             }
             if let Err(message) = stream_json::round_trips(line) {

crates/tui/tests/fixtures/conformance/README.md

@@ -20,7 +20,10 @@ CODEWHALE_CONFORMANCE_UPDATE=1 cargo test -p codewhale-tui --lib -- conformance:
 
 Default is compare-only. Update mode rewrites drifted goldens and refuses to
 run when `CI` is set. A golden change is a behavior change and is reviewed as
-one.
+one. Harness deadlines, incomplete turns, broken invariants, and empty cases
+are failures in both modes and cannot be recorded as goldens. Provider timeout
+errors are legitimate observable outcomes; a harness timing out is missing
+evidence.
 
 Platforms: `sse` and `mcp` run everywhere. `events`, `prompt` and `hooks`
 goldens were recorded on Unix and compile only there — hook fixtures are POSIX
@@ -83,10 +86,9 @@ left in the workspace (path → sha256 prefix). `invariants` in a case are
 checked independently of the golden (`workspace_file_absent`,
 `workspace_file_present`, `max_model_requests`).
 
-A case with `expected_to_change` pins current behavior that a named change is
-fixing. Its invariant is expected to fail today; when the invariant starts to
-hold, the runner fails until the golden is re-recorded and the block deleted.
-`events/provider_error_after_tool_call` is that case for C02-05 (#6561).
+`events/provider_error_after_tool_call` requires the C02-05 (#6561) authority:
+a failed response executes no collected tool call and issues no retry. Its
+invariants are checked before recording, with no exemption for known defects.
 
 ### MCP transcripts (`mcp` goldens)
 
@@ -127,7 +129,8 @@ in a command runs a helper that saves the hook's stdin and every documented
 environment variable (`CODEWHALE_*` / `DEEPSEEK_*`, see `HOOK_ENV_CONTRACT`).
 The golden records the outcome — `{"admit": {requires_approval, updated_input,
 additional_context}}` or `{"refuse": {kind, detail}}` for `tool_call_before`,
-the observer results for `tool_call_after` — and, per hook, whether it ran,
+the observer results (including exact stdout, stderr and error) for
+`tool_call_after` — and, per hook, whether it ran,
 its stdin (the schema-1 document, parsed) and its contract environment. The
 hook session id and temp paths are masked. POSIX shell: Unix only.
 
@@ -153,6 +156,7 @@ case, and by passing the Rust runner once it is wired in:
   regression net around it while its edges move. A host-side consumer of
   `EventMsg` can read these goldens as its contract.
 
-Fields named `detail` carry human-readable error text; the Rust runner
-compares them exactly, a second implementation may compare them loosely (the
-`kind` / `type` beside them is the contract).
+Fields named `detail` carry human-readable error text and are compared exactly,
+including by a second implementation. Only the explicitly documented masks
+apply. No whitespace, line-ending, error-text, or event-order normalization may
+be added to make a drift pass. `.gitattributes` pins fixture checkout bytes to LF.

crates/tui/tests/fixtures/conformance/events/provider_error_after_tool_call.case.json

@@ -120,11 +120,5 @@
     {
       "max_model_requests": 1
     }
-  ],
-  "expected_to_change": {
-    "finding": "C02-05",
-    "issue": "#6561",
-    "owner": "opus/0101-audit-core",
-    "reason": "Current source still executes tool calls collected before a terminal stream error; the core audit slice makes the turn settle them as not-started. When that lands, re-record this golden and delete this block."
-  }
+  ]
 }