test(kernel): dump daemon-side state when a dispatch never arrives (#174)

kernel/relayflowd/tests/crash_resume/agent.rs

@@ -17,8 +17,8 @@ use super::{
         start_run,
     },
     support::{
-        completed_step_count, journal_entries, kill_group, kill_process_group, read_pid,
-        resume_cli, wait_until,
+        completed_step_count, describe_stalled_resume, journal_entries, kill_group,
+        kill_process_group, read_pid, resume_cli, wait_until,
     },
 };
 
@@ -127,8 +127,17 @@ fn rung_c_sigkill_boundaries_resume_only_unfinished_steps_via_real_cli() {
         let _server = fixture.server();
         let mut worker = attached_worker(&fixture, "boundary-agent-stub");
         let run_id = super::support::only_run_id(&fixture.data_dir);
-        let resume = spawn_resume(&fixture, &run_id);
-        let dispatch = worker.event("step.dispatch").unwrap();
+        let mut resume = spawn_resume(&fixture, &run_id);
+        // Do not `.unwrap()` this. A missing dispatch is #174, and the whole
+        // difficulty there has been that the failure carries no daemon-side
+        // state -- so capture it here rather than losing it to the unwind.
+        let dispatch = match worker.event("step.dispatch") {
+            Ok(dispatch) => dispatch,
+            Err(error) => panic!(
+                "{label}: no step.dispatch after resume: {error}{}",
+                describe_stalled_resume(&fixture.data_dir, &mut resume)
+            ),
+        };
         assert!(!record_effect(&fixture, &mut worker, &dispatch).unwrap());
         complete(&mut worker, &dispatch).unwrap();
         let output = resume.wait_with_output().unwrap();

kernel/relayflowd/tests/crash_resume/concurrency.rs

@@ -28,12 +28,12 @@ fn run_start_dispatches_every_independent_lane_before_any_completion() {
     let mut worker = attached_worker(&fixture, "parallel-stub");
     let run_id = start_run(&fixture);
 
-    worker.set_read_timeout(Some(Duration::from_secs(1)));
+    worker.override_read_timeout(Some(Duration::from_secs(1)));
     let first = worker.event("step.dispatch").unwrap();
     let second = worker
         .event("step.dispatch")
         .expect("both independent lanes must dispatch before either completes");
-    worker.set_read_timeout(None);
+    worker.override_read_timeout(None);
     assert_eq!(first["step_id"], "lane-b");
     assert_eq!(second["step_id"], "lane-a");
 

kernel/relayflowd/tests/crash_resume/llm.rs

@@ -11,7 +11,10 @@ use super::{
     llm_support::{
         LlmFixture, ProtocolClient, ServerGuard, attached_worker, complete, spawn_resume, start_run,
     },
-    support::{journal_entries, kill_group, only_run_id, read_pid, resume_cli, wait_until},
+    support::{
+        describe_stalled_resume, journal_entries, kill_group, only_run_id, read_pid,
+        resume_cli, wait_until,
+    },
 };
 
 #[test]
@@ -106,8 +109,17 @@ fn sigkill_sweep_covers_before_and_between_the_rung_b_steps() {
         let _server = ServerGuard::start(&fixture);
         let mut worker = attached_worker(&fixture, "boundary-stub");
         let run_id = only_run_id(&fixture.data_dir);
-        let resume = spawn_resume(&fixture, &run_id);
-        let dispatch = worker.event("step.dispatch").unwrap();
+        let mut resume = spawn_resume(&fixture, &run_id);
+        // Do not `.unwrap()` this. A missing dispatch is #174, and the whole
+        // difficulty there has been that the failure carries no daemon-side
+        // state -- so capture it here rather than losing it to the unwind.
+        let dispatch = match worker.event("step.dispatch") {
+            Ok(dispatch) => dispatch,
+            Err(error) => panic!(
+                "{label}: no step.dispatch after resume: {error}{}",
+                describe_stalled_resume(&fixture.data_dir, &mut resume)
+            ),
+        };
         complete(&mut worker, &dispatch, json!({"answer": 4})).unwrap();
         let output = resume.wait_with_output().unwrap();
         assert!(

kernel/relayflowd/tests/crash_resume/llm_support.rs

@@ -235,17 +235,39 @@ pub struct ProtocolClient {
     reader: BufReader<UnixStream>,
     next_id: u64,
     events: Vec<Value>,
+    /// The ceiling currently in force, so a timeout can report the bound that
+    /// actually fired rather than the default constant.
+    read_timeout: Duration,
 }
 
+/// Ceiling on any single protocol read in a test.
+///
+/// The whole `crash_resume` target runs in about 38 seconds, so this is far
+/// longer than any legitimate wait; it exists only to convert "never" into a
+/// failure. See `read_frame` for why that matters.
+const READ_TIMEOUT: Duration = Duration::from_secs(60);
+
 impl ProtocolClient {
     pub fn connect(socket: &Path) -> Self {
         let stream = UnixStream::connect(socket).unwrap();
-        let reader = BufReader::new(stream.try_clone().unwrap());
+        let read_half = stream.try_clone().unwrap();
+        // Without this a frame that never arrives blocks forever. These tests
+        // SIGKILL a daemon and resume it, so "the dispatch never comes" is a
+        // reachable state, not a hypothetical -- and an unbounded read turns it
+        // into a silent hang that produces NO output at all. On GitHub runners
+        // that consumed the entire 30-minute step three times (#174), and the
+        // only evidence left behind was the harness's own
+        // "has been running for over 60 seconds" line.
+        read_half
+            .set_read_timeout(Some(READ_TIMEOUT))
+            .expect("set protocol read timeout");
+        let reader = BufReader::new(read_half);
         Self {
             stream,
             reader,
             next_id: 1,
             events: Vec::new(),
+            read_timeout: READ_TIMEOUT,
         }
     }
 
@@ -306,14 +328,62 @@ impl ProtocolClient {
         }
     }
 
-    pub fn set_read_timeout(&self, timeout: Option<Duration>) {
-        self.stream.set_read_timeout(timeout).unwrap();
+    /// Override the read ceiling. `None` restores the default -- it does NOT
+    /// make reads unbounded.
+    ///
+    /// Deliberately NOT named `set_read_timeout`: that name belongs to
+    /// `UnixStream`, where `None` means "block forever". Shadowing a std API
+    /// while inverting its meaning is a trap no docstring reliably defuses.
+    ///
+    /// That distinction is the point. Callers tighten the bound for a specific
+    /// assertion and then pass `None` to mean "back to normal". Two shapes use
+    /// it, and they are not the same:
+    ///
+    /// - `parallel_lifecycle.rs:138,208` probe for SILENCE -- 200ms, then
+    ///   assert the read errors.
+    /// - `concurrency.rs:31` tightens to 1s and expects the read to SUCCEED,
+    ///   so a missing dispatch fails fast instead of stalling the test.
+    ///
+    /// If `None` meant "block forever", every read after any of those would be
+    /// unbounded and the ceiling this type advertises would be a claim it does
+    /// not keep -- which is what two review lenses caught in the first revision
+    /// of this change.
+    pub fn override_read_timeout(&mut self, timeout: Option<Duration>) {
+        let timeout = timeout.unwrap_or(READ_TIMEOUT);
+        // Set it on the fd `read_frame` actually reads through -- the reader's,
+        // not `self.stream`. `try_clone` produces a separate descriptor, and
+        // while Linux and Darwin keep SO_RCVTIMEO on the shared socket (so
+        // writing to either fd happens to work today), that is a platform
+        // detail, not a guarantee. Going through the reader means the ceiling
+        // is set where it is read, with no hidden assumption to remember.
+        self.reader
+            .get_ref()
+            .set_read_timeout(Some(timeout))
+            .unwrap();
+        self.read_timeout = timeout;
     }
 
     fn read_frame(&mut self) -> Result<Value> {
         let mut line = String::new();
-        if self.reader.read_line(&mut line)? == 0 {
-            bail!("protocol connection closed")
+        match self.reader.read_line(&mut line) {
+            Ok(0) => bail!("protocol connection closed"),
+            Ok(_) => {}
+            // Name the timeout rather than letting it surface as a bare I/O
+            // error. A test that stops here is waiting for a frame the daemon
+            // never sent, and that sentence is the entire diagnosis.
+            Err(error)
+                if matches!(
+                    error.kind(),
+                    std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
+                ) =>
+            {
+                let waited = self.read_timeout;
+                bail!(
+                    "timed out after {waited:?} waiting for a protocol frame; \
+                     the daemon sent nothing (see #174)"
+                )
+            }
+            Err(error) => return Err(error.into()),
         }
         serde_json::from_str(&line).context("decode protocol frame")
     }

kernel/relayflowd/tests/crash_resume/parallel_lifecycle.rs

@@ -135,9 +135,9 @@ fn overlapping_agent_lanes_serialize_while_disjoint_lanes_merge_in_either_order(
     start_run(&fixture);
     let lane_b = worker.event("step.dispatch").unwrap();
     assert_eq!(lane_b["step_id"], "lane-b");
-    worker.set_read_timeout(Some(Duration::from_millis(200)));
+    worker.override_read_timeout(Some(Duration::from_millis(200)));
     assert!(worker.event("step.dispatch").is_err());
-    worker.set_read_timeout(None);
+    worker.override_read_timeout(None);
     complete_agent(&mut worker, &lane_b, &[("repo-b", "rB")]).unwrap();
     let lane_a = worker.event("step.dispatch").unwrap();
     assert_eq!(lane_a["pins"]["workspace"][0]["revision_id"], "rB");
@@ -205,9 +205,9 @@ fn overlapping_agent_conflict_survives_server_crash_and_resume() {
     let retried = replacement.event("step.dispatch").unwrap();
     assert_eq!(retried["step_id"], "lane-b");
     assert_eq!(retried["attempt"], 2);
-    replacement.set_read_timeout(Some(Duration::from_millis(200)));
+    replacement.override_read_timeout(Some(Duration::from_millis(200)));
     assert!(replacement.event("step.dispatch").is_err());
-    replacement.set_read_timeout(None);
+    replacement.override_read_timeout(None);
     complete_agent(&mut replacement, &retried, &[("repo-b", "rB")]).unwrap();
     let lane_a = replacement.event("step.dispatch").unwrap();
     complete_agent(&mut replacement, &lane_a, &[("repo-b", "rA")]).unwrap();

kernel/relayflowd/tests/crash_resume/support.rs

@@ -156,6 +156,55 @@ pub fn journal_entries(data_dir: &Path) -> Option<Vec<JournalEntry>> {
     journal.scan_segment(segment).ok()
 }
 
+/// Everything known about a stalled resume, as one panic message.
+///
+/// When a dispatch never arrives (#174) the useful state is all on the daemon
+/// side and none of it is captured today: the resumed child is still running,
+/// so `wait_with_output` is never reached and its output is discarded when the
+/// test unwinds; the journal is never read. Four occurrences produced four test
+/// names and nothing else.
+///
+/// This kills the child first -- it is wedged by definition, and without that
+/// the read below would block as long as the one that already timed out -- then
+/// reports its output and the run's journal together.
+pub fn describe_stalled_resume(data_dir: &Path, resume: &mut Child) -> String {
+    let mut report = String::new();
+
+    // Kill before read. The child is not going to finish on its own.
+    let _ = resume.kill();
+    let _ = resume.wait();
+
+    let mut stdout = String::new();
+    let mut stderr = String::new();
+    if let Some(mut handle) = resume.stdout.take() {
+        let _ = std::io::Read::read_to_string(&mut handle, &mut stdout);
+    }
+    if let Some(mut handle) = resume.stderr.take() {
+        let _ = std::io::Read::read_to_string(&mut handle, &mut stderr);
+    }
+    report.push_str(&format!(
+        "\n--- resume child ---\nstdout ({} bytes):\n{stdout}\nstderr ({} bytes):\n{stderr}\n",
+        stdout.len(),
+        stderr.len()
+    ));
+
+    // The journal says how far the run actually got, which is the question a
+    // missing dispatch raises: did the daemon resume and stall, or never resume?
+    match journal_entries(data_dir) {
+        None => report.push_str("--- journal --- absent (no run directory)\n"),
+        Some(entries) => {
+            report.push_str(&format!("--- journal ({} entries) ---\n", entries.len()));
+            for entry in &entries {
+                report.push_str(&format!(
+                    "  seq={} type={:?} step={:?}\n",
+                    entry.seq, entry.entry_type, entry.step_id
+                ));
+            }
+        }
+    }
+    report
+}
+
 pub fn completed_step_count(entries: &[JournalEntry]) -> usize {
     entries
         .iter()