test(mcp,api): close verification gaps in daemon resilience tests

hyperdb-api/tests/process_tests.rs

@@ -72,24 +72,71 @@ fn callback_connection_shutdowns_hyperd_after_parent_kill() {
 
 fn wait_for_reported_pid(pid_file: &std::path::Path, timeout: Duration) -> Result<u32, String> {
     let deadline = std::time::Instant::now() + timeout;
+    let mut last_unparseable: Option<String> = None;
     loop {
         match fs::read_to_string(pid_file) {
-            Ok(contents) => {
-                return contents
-                    .trim()
-                    .parse()
-                    .map_err(|error| format!("invalid PID report {contents:?}: {error}"));
-            }
+            Ok(contents) => match contents.trim().parse::<u32>() {
+                Ok(pid) => return Ok(pid),
+                Err(_) => {
+                    // The child reports its PID with `fs::write`, which is
+                    // `File::create` (truncates/creates) then `write_all` —
+                    // not atomic. A poll can land in the window where the
+                    // file exists but is still empty or partially written;
+                    // treat that the same as "not created yet" rather than a
+                    // fatal parse error, and keep polling to the deadline.
+                    last_unparseable = Some(contents);
+                }
+            },
             Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
             Err(error) => return Err(format!("could not read PID report: {error}")),
         }
         if std::time::Instant::now() >= deadline {
-            return Err(format!("no PID report appeared at {}", pid_file.display()));
+            return Err(match last_unparseable {
+                Some(contents) => format!(
+                    "PID report at {} never became parseable before the deadline; last read {contents:?}",
+                    pid_file.display()
+                ),
+                None => format!("no PID report appeared at {}", pid_file.display()),
+            });
         }
         thread::sleep(Duration::from_millis(20));
     }
 }
 
+/// The parent's `fs::write` of the PID report is `File::create` (truncates or
+/// creates, leaving an empty file momentarily) followed by `write_all` — not
+/// atomic. A poll landing in that window used to make `"".parse::<u32>()`
+/// panic the whole test; it must instead be treated as "not ready yet" and
+/// retried to the deadline.
+#[test]
+fn wait_for_reported_pid_retries_past_a_torn_write() {
+    let temp_dir = tempfile::tempdir().expect("create temp dir for torn-write simulation");
+    let pid_file = temp_dir.path().join("hyperd-pid");
+
+    let writer_pid_file = pid_file.clone();
+    let writer = thread::spawn(move || {
+        // Reproduces the exact non-atomic sequence: create (truncate) first,
+        // leaving the file present-but-empty for a deliberate window, then
+        // write the real content — same as the production `fs::write` call
+        // this helper polls for, just with the empty window stretched out
+        // long enough that a fast poller is guaranteed to observe it.
+        fs::File::create(&writer_pid_file).expect("create pid file (torn-write simulation)");
+        thread::sleep(Duration::from_millis(150));
+        fs::write(&writer_pid_file, "4242").expect("finish torn-write simulation");
+    });
+
+    let result = wait_for_reported_pid(&pid_file, Duration::from_secs(5));
+    writer
+        .join()
+        .expect("torn-write simulation thread must not panic");
+
+    assert_eq!(
+        result,
+        Ok(4242),
+        "a read landing in the create/write_all gap must be retried, not treated as a fatal parse error"
+    );
+}
+
 fn bounded_process_exit_poll(pid: u32, timeout: Duration) -> bool {
     let deadline = std::time::Instant::now() + timeout;
     loop {

hyperdb-mcp/CHANGELOG.md

@@ -232,6 +232,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
   `daemon status --port` now probes that exact health port, discovered-daemon
   error reports target the effective health port, and best-effort health I/O
   no longer retains the engine mutex.
+- **Health-listener connections could be torn down before the client sent its
+  first byte, on macOS and other BSD-derived kernels.** `HealthListener::bind`
+  puts the listening socket in non-blocking mode so its accept loop can poll
+  for shutdown; on BSD kernels (unlike Linux) `accept()` propagates that
+  `O_NONBLOCK` flag to the accepted socket, so the connection handler's first
+  `read_line` returned `WouldBlock` in microseconds and closed the connection
+  before a client had a chance to write a command. Liveness checks, restart
+  reporting, and heartbeats all depend on this connection surviving long
+  enough to receive one line, so accepted connections are now explicitly
+  forced back into blocking mode.
 - **Attachment contention is actionable for persistent *and* user attaches.**
   A lock conflict (SQLSTATE `55006`, or a legacy "already attached" / "file is
   locked" phrase from older hyperd) now returns `RESOURCE_BUSY` with the

hyperdb-mcp/src/daemon/health.rs

@@ -136,21 +136,20 @@ impl HealthListener {
                 break;
             }
 
-            match self.listener.accept() {
-                Ok((stream, _addr)) => {
-                    if let Err(error) = stream.set_nonblocking(false) {
-                        warn!(
-                            error = %error,
-                            "could not make accepted health connection blocking"
-                        );
-                        continue;
-                    }
+            match accept_and_force_blocking(&self.listener) {
+                Ok(AcceptedConnection::Ready(stream)) => {
                     let state = Arc::clone(&state);
                     let info = Arc::clone(&info);
                     std::thread::spawn(move || {
                         handle_client(stream, &state, &info);
                     });
                 }
+                Ok(AcceptedConnection::ForceBlockingFailed(error)) => {
+                    warn!(
+                        error = %error,
+                        "could not make accepted health connection blocking"
+                    );
+                }
                 Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                     // Poll tightly: the doctor network phase budgets only a
                     // few hundred ms for a STATUS round-trip, and on slow CI
@@ -169,6 +168,40 @@ impl HealthListener {
     }
 }
 
+/// Outcome of [`accept_and_force_blocking`] once a connection has actually
+/// been accepted (as opposed to `accept()` itself erroring, which is
+/// propagated through the outer `io::Result`).
+enum AcceptedConnection {
+    /// Accepted and confirmed blocking; ready to hand to [`handle_client`].
+    Ready(TcpStream),
+    /// Accepted, but the attempt to clear the non-blocking flag failed. The
+    /// connection is dropped; the caller logs and moves on to the next
+    /// `accept()`.
+    ForceBlockingFailed(std::io::Error),
+}
+
+/// Accept one connection and force it into blocking mode.
+///
+/// [`HealthListener::bind`] puts the *listening* socket into non-blocking
+/// mode so [`HealthListener::run`]'s loop can poll `should_shutdown` between
+/// accepts. On BSD-derived kernels — macOS and other BSDs, but **not**
+/// Linux, which keeps a newly accepted socket's blocking mode independent of
+/// the listener's — `accept()` propagates the listening socket's
+/// `O_NONBLOCK` flag to the accepted socket. Left non-blocking, the accepted
+/// stream would return `WouldBlock` from `read_line` in
+/// [`handle_client`] in microseconds — typically before the client has even
+/// written its first byte — tearing the connection down before it ever
+/// received a command. `set_nonblocking(false)` below undoes that
+/// propagation unconditionally, which is a no-op (not a bug) on platforms
+/// that never had the problem.
+fn accept_and_force_blocking(listener: &TcpListener) -> std::io::Result<AcceptedConnection> {
+    let (stream, _addr) = listener.accept()?;
+    match stream.set_nonblocking(false) {
+        Ok(()) => Ok(AcceptedConnection::Ready(stream)),
+        Err(error) => Ok(AcceptedConnection::ForceBlockingFailed(error)),
+    }
+}
+
 fn status_json(info: &Mutex<DaemonInfo>) -> String {
     let snapshot = info.lock().expect("DaemonInfo mutex poisoned").clone();
     match DaemonRecord::with_current_identity(&snapshot) {
@@ -703,4 +736,65 @@ mod tests {
             "newline-free health responses beyond the 64 KiB protocol limit must be rejected; {outcome}"
         );
     }
+
+    /// Asserts the accepted socket's actual blocking state via `fcntl`,
+    /// rather than inferring it from read-timing behavior. On Linux an
+    /// accepted socket is blocking regardless of the listener's mode, so a
+    /// behavioral test (send late, expect it to still be read) passes
+    /// trivially there even with `accept_and_force_blocking`'s
+    /// `set_nonblocking(false)` reverted — it would only catch a regression
+    /// on the BSD-derived kernels (macOS and other BSDs) that actually
+    /// propagate `O_NONBLOCK` to accepted sockets. Reading the `O_NONBLOCK`
+    /// flag directly makes the test verify the real contract everywhere,
+    /// rather than a platform-dependent behavioral proxy for it.
+    #[cfg(unix)]
+    #[test]
+    fn accept_and_force_blocking_clears_nonblocking_flag() {
+        use std::os::unix::io::AsRawFd;
+
+        let listener = HealthListener::bind(0).expect("bind test health listener");
+        let port = listener.port;
+
+        let client = std::thread::spawn(move || {
+            // Held for the duration of the accept below; dropped (and thus
+            // closed) only once this thread returns.
+            std::net::TcpStream::connect(("127.0.0.1", port)).expect("connect test client")
+        });
+
+        let accept_deadline = Instant::now() + Duration::from_secs(2);
+        let accepted = loop {
+            match accept_and_force_blocking(&listener.listener) {
+                Ok(AcceptedConnection::Ready(stream)) => break stream,
+                Ok(AcceptedConnection::ForceBlockingFailed(error)) => {
+                    panic!("force-blocking the accepted test connection failed: {error}")
+                }
+                Err(ref error) if error.kind() == std::io::ErrorKind::WouldBlock => {
+                    assert!(
+                        Instant::now() < accept_deadline,
+                        "timed out waiting to accept the test client connection"
+                    );
+                    std::thread::sleep(Duration::from_millis(2));
+                }
+                Err(error) => panic!("accept failed: {error}"),
+            }
+        };
+        client.join().expect("test client thread must not panic");
+
+        // SAFETY: `accepted` is a live, owned, valid socket for the duration
+        // of this call; `F_GETFL` only reads flags and mutates nothing.
+        let flags = unsafe { libc::fcntl(accepted.as_raw_fd(), libc::F_GETFL) };
+        assert!(
+            flags >= 0,
+            "fcntl(F_GETFL) on the accepted test connection failed: {}",
+            std::io::Error::last_os_error()
+        );
+        assert_eq!(
+            flags & libc::O_NONBLOCK,
+            0,
+            "accepted health connection must be blocking (O_NONBLOCK must be clear); \
+             reverting accept_and_force_blocking's set_nonblocking(false) call would \
+             leave O_NONBLOCK set on BSD-derived kernels that propagate it from the \
+             listening socket"
+        );
+    }
 }

hyperdb-mcp/tests/daemon_tests.rs

@@ -314,8 +314,9 @@ fn health_listener_waits_for_command_after_accept() {
                 }
             }
 
-            // Keep the already-accepted first socket idle for more than two
-            // additional 100ms listener polls before sending its first command.
+            // Keep the already-accepted first socket idle for well over the
+            // listener's 5ms accept-loop poll interval before sending its
+            // first command.
             std::thread::sleep(Duration::from_millis(350));
             idle_client.ping("delayed first-client")
         })();
@@ -1232,7 +1233,10 @@ fn takeover_decision_both_unparseable_reuses() {
 // ─── Integration tests: full daemon lifecycle with real hyperd ─────────────────
 
 #[test]
-#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"]
+#[cfg_attr(
+    target_os = "macos",
+    ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"
+)]
 fn daemon_mode_engine_connects_to_shared_hyperd() {
     let _lock = acquire_env_lock();
     let daemon = TestDaemon::start();
@@ -1251,7 +1255,10 @@ fn daemon_mode_engine_connects_to_shared_hyperd() {
 }
 
 #[test]
-#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"]
+#[cfg_attr(
+    target_os = "macos",
+    ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"
+)]
 fn daemon_mode_two_engines_share_same_hyperd() {
     let _lock = acquire_env_lock();
     let daemon = TestDaemon::start();
@@ -1294,7 +1301,10 @@ fn daemon_mode_two_engines_share_same_hyperd() {
 }
 
 #[test]
-#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"]
+#[cfg_attr(
+    target_os = "macos",
+    ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"
+)]
 fn daemon_mode_persistent_database_file_survives_engine_drop() {
     let _lock = acquire_env_lock();
     let _daemon = TestDaemon::start();
@@ -1319,7 +1329,10 @@ fn daemon_mode_persistent_database_file_survives_engine_drop() {
 }
 
 #[test]
-#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"]
+#[cfg_attr(
+    target_os = "macos",
+    ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"
+)]
 fn daemon_mode_persistent_engine_data_is_queryable() {
     let _lock = acquire_env_lock();
     let daemon = TestDaemon::start();
@@ -1348,7 +1361,10 @@ fn daemon_mode_persistent_engine_data_is_queryable() {
 
 #[cfg(unix)]
 #[test]
-#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"]
+#[cfg_attr(
+    target_os = "macos",
+    ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"
+)]
 fn hyperd_monitor_detects_killed_hyperd_and_restarts() {
     let _lock = acquire_env_lock();
     let daemon = TestDaemon::start();
@@ -1376,7 +1392,10 @@ fn hyperd_monitor_detects_killed_hyperd_and_restarts() {
 
 #[cfg(unix)]
 #[test]
-#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"]
+#[cfg_attr(
+    target_os = "macos",
+    ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"
+)]
 fn client_report_triggers_restart_after_kill() {
     let _lock = acquire_env_lock();
     let daemon = TestDaemon::start();
@@ -1404,7 +1423,10 @@ fn client_report_triggers_restart_after_kill() {
 
 #[cfg(unix)]
 #[test]
-#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"]
+#[cfg_attr(
+    target_os = "macos",
+    ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"
+)]
 fn engine_recovers_after_hyperd_killed() {
     // End-to-end test: the user-visible behavior of this whole feature.
     // 1. Start daemon + create an Engine (= an MCP client connection).
@@ -1456,7 +1478,10 @@ fn engine_recovers_after_hyperd_killed() {
 }
 
 #[test]
-#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"]
+#[cfg_attr(
+    target_os = "macos",
+    ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"
+)]
 fn daemon_mode_ephemeral_database_cleaned_up_on_drop() {
     let _lock = acquire_env_lock();
     let _daemon = TestDaemon::start();