Merge remote-tracking branch 'origin/main'

lib/crates/fabro-cli/src/commands/server/stop.rs

@@ -16,17 +16,24 @@ pub(crate) async fn stop_server(storage_dir: &Path, timeout: Duration) -> Result
 
     fabro_proc::sigterm(record.pid);
 
+    // Use the zombie-aware predicate here: this loop is commonly driven
+    // against a child of the calling process (tests, install/uninstall
+    // in-process shutdowns, a foreground-launching shell). A zombie
+    // child would otherwise satisfy `process_running` until its parent
+    // waits, causing us to burn the whole `timeout` on an already-dead
+    // process. The `ps` cost (~2 ms per poll) is trivial compared to
+    // the 10 s timeout and is only paid while the process still exists.
     let poll_interval = Duration::from_millis(100);
     let mut elapsed = Duration::ZERO;
     while elapsed < timeout {
-        if !fabro_proc::process_running(record.pid) {
+        if !fabro_proc::process_running_strict(record.pid) {
             break;
         }
         time::sleep(poll_interval).await;
         elapsed += poll_interval;
     }
 
-    if fabro_proc::process_running(record.pid) {
+    if fabro_proc::process_running_strict(record.pid) {
         fabro_proc::sigkill(record.pid);
         time::sleep(Duration::from_millis(100)).await;
     }

lib/crates/fabro-cli/tests/it/cmd/pr_list.rs

@@ -1,4 +1,13 @@
+#![allow(
+    clippy::absolute_paths,
+    reason = "This test module prefers explicit type paths over extra imports."
+)]
+
 use fabro_test::{fabro_snapshot, test_context};
+use fabro_types::run_event::PullRequestCreatedProps;
+use fabro_types::{EventBody, RunEvent, RunId};
+
+use super::support::{server_endpoint, setup_completed_fast_dry_run};
 
 #[test]
 fn help() {
@@ -26,17 +35,61 @@ fn help() {
     ");
 }
 
+// Seed a PR event against this test's own run so the store is guaranteed to
+// have at least one entry; `fabro pr list` then must load GitHub credentials
+// and fail, regardless of what peer tests have left in the shared store.
 #[test]
 fn pr_list_missing_github_credentials_errors() {
     let context = test_context!();
+    let run = setup_completed_fast_dry_run(&context);
+    let run_id: RunId = run.run_id.parse().unwrap();
+
+    let runtime = tokio::runtime::Runtime::new().unwrap();
+    runtime.block_on(async {
+        let (client, base_url) =
+            server_endpoint(&context.storage_dir).expect("server endpoint should exist");
+        let event = RunEvent {
+            id: ulid::Ulid::new().to_string(),
+            ts: chrono::Utc::now(),
+            run_id,
+            node_id: None,
+            node_label: None,
+            stage_id: None,
+            parallel_group_id: None,
+            parallel_branch_id: None,
+            session_id: None,
+            parent_session_id: None,
+            tool_call_id: None,
+            actor: None,
+            body: EventBody::PullRequestCreated(PullRequestCreatedProps {
+                pr_url:      "https://github.com/fabro-sh/fabro/pull/123".to_string(),
+                pr_number:   123,
+                owner:       "fabro-sh".to_string(),
+                repo:        "fabro".to_string(),
+                base_branch: "main".to_string(),
+                head_branch: "fabro/run/demo".to_string(),
+                title:       "Map the constellations".to_string(),
+                draft:       false,
+            }),
+        };
+        client
+            .post(format!("{base_url}/api/v1/runs/{run_id}/events"))
+            .json(&event)
+            .send()
+            .await
+            .unwrap()
+            .error_for_status()
+            .unwrap();
+    });
+
     let mut cmd = context.command();
     cmd.args(["pr", "list"]);
 
     fabro_snapshot!(context.filters(), cmd, @"
-    success: true
-    exit_code: 0
+    success: false
+    exit_code: 1
     ----- stdout -----
-    No pull requests found.
     ----- stderr -----
+    error: GitHub credentials required — run `fabro install` or set GITHUB_TOKEN
     ");
 }

lib/crates/fabro-proc/src/flock.rs

@@ -20,6 +20,25 @@ pub fn try_flock_exclusive(file: &File) -> io::Result<bool> {
     }
 }
 
+/// Try to acquire a shared (read) lock on `file` without blocking.
+///
+/// Returns `Ok(true)` if the lock was acquired, `Ok(false)` if another
+/// process/fd already holds an incompatible lock, and `Err` for unexpected
+/// errors.
+pub fn try_flock_shared(file: &File) -> io::Result<bool> {
+    // SAFETY: flock() on a valid fd is safe; LOCK_NB makes it non-blocking.
+    let ret = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_SH | libc::LOCK_NB) };
+    if ret == 0 {
+        Ok(true)
+    } else {
+        let err = io::Error::last_os_error();
+        match err.raw_os_error() {
+            Some(libc::EWOULDBLOCK) => Ok(false),
+            _ => Err(err),
+        }
+    }
+}
+
 /// Release any lock held on `file`.
 pub fn flock_unlock(file: &File) -> io::Result<()> {
     // SAFETY: flock() with LOCK_UN on a valid fd is safe.

lib/crates/fabro-proc/src/lib.rs

@@ -11,14 +11,14 @@ mod signal;
 mod title;
 
 #[cfg(unix)]
-pub use flock::{flock_unlock, try_flock_exclusive};
+pub use flock::{flock_unlock, try_flock_exclusive, try_flock_shared};
 #[cfg(target_os = "linux")]
 pub use pre_exec::pre_exec_pdeathsig;
 #[cfg(unix)]
 pub use pre_exec::pre_exec_setpgid;
 #[cfg(unix)]
 pub use pre_exec::pre_exec_setsid;
-pub use signal::{process_exists, process_group_alive, process_running};
+pub use signal::{process_exists, process_group_alive, process_running, process_running_strict};
 #[cfg(unix)]
 pub use signal::{
     sigkill, sigkill_process_group, sigterm, sigterm_process_group, sigusr1, sigusr2,

lib/crates/fabro-proc/src/signal.rs

@@ -20,17 +20,35 @@ pub fn process_exists(pid: u32) -> bool {
 
 /// Check whether a process with the given PID is still running.
 ///
-/// On Unix, this treats zombie / defunct processes as not running even though
-/// they still have a visible PID until their parent reaps them. If the
-/// follow-up `ps` probe fails, this falls back to `process_exists(pid)` to
-/// preserve the old conservative behavior.
+/// On Unix, delegates to `process_exists` (a `kill(pid, 0)` probe). Unreaped
+/// zombies count as running here because the callers are hot paths (test
+/// harness marker scans, daemon-liveness probes) where a zombie window is
+/// sub-millisecond and the ~2 ms cost of an authoritative zombie check via
+/// `ps` would dominate. For the narrow set of callers that genuinely need
+/// to treat zombies as stopped (notably the `fabro server stop` polling
+/// loop, which can wait out its full timeout on a zombie child), use
+/// `process_running_strict`.
 pub fn process_running(pid: u32) -> bool {
+    process_exists(pid)
+}
+
+/// Like `process_running`, but treats zombie / defunct processes as not
+/// running.
+///
+/// On Unix, follows a cheap `kill(pid, 0)` probe with a `ps` shell-out to
+/// read the process state character and excludes `Z`/`z` entries. Falls
+/// back to `process_exists(pid)` when the `ps` probe fails, preserving the
+/// old conservative behavior. On non-unix, identical to `process_exists`.
+///
+/// Prefer `process_running` unless you are polling for a child you cannot
+/// `wait()` on — the `ps` invocation costs ~2 ms per call on macOS and is
+/// wasted on hot paths that have no zombie exposure.
+pub fn process_running_strict(pid: u32) -> bool {
     #[cfg(unix)]
     {
         if !process_exists(pid) {
             return false;
         }
-
         unix_process_state(pid).is_none_or(|state| !matches!(state, 'Z' | 'z'))
     }
     #[cfg(not(unix))]
@@ -52,7 +70,6 @@ fn unix_process_state(pid: u32) -> Option<char> {
     if !output.status.success() {
         return None;
     }
-
     String::from_utf8_lossy(&output.stdout)
         .chars()
         .find(|ch| !ch.is_whitespace())
@@ -155,22 +172,23 @@ mod tests {
     use std::process::{Command, Stdio};
     use std::time::Duration;
 
-    use super::{process_exists, process_group_alive, process_running};
+    use super::{process_exists, process_group_alive, process_running, process_running_strict};
     use crate::pre_exec::pre_exec_setpgid;
 
     #[test]
     fn process_running_returns_true_for_current_process() {
         assert!(process_exists(std::process::id()));
         assert!(process_running(std::process::id()));
+        assert!(process_running_strict(std::process::id()));
     }
 
     #[cfg(unix)]
     #[test]
     #[expect(
         clippy::disallowed_methods,
-        reason = "process-state test needs to spawn a short-lived child and intentionally leave it unreaped"
+        reason = "zombie-detection test needs to spawn a short-lived child and intentionally leave it unreaped"
     )]
-    fn process_running_returns_false_for_unreaped_zombie_child() {
+    fn process_running_strict_returns_false_for_unreaped_zombie_child() {
         let mut child = Command::new("sh")
             .args(["-c", "exit 0"])
             .spawn()
@@ -184,8 +202,12 @@ mod tests {
             "unreaped zombie should still have a visible pid"
         );
         assert!(
-            !process_running(pid),
-            "unreaped zombie should not count as a running process"
+            process_running(pid),
+            "cheap process_running treats zombies as alive by design"
+        );
+        assert!(
+            !process_running_strict(pid),
+            "process_running_strict should treat zombies as stopped"
         );
 
         let _status = child.wait().expect("child should remain waitable");

lib/crates/fabro-server/tests/it/scenario/lifecycle.rs

@@ -249,7 +249,17 @@ async fn full_http_lifecycle_cancel() {
     )
     .await;
     assert_eq!(body["status"], "running");
-    assert_eq!(body["pending_control"], "cancel");
+    // `pending_control` is computed from the store projection after the cancel
+    // event is appended AND the worker is signaled. The worker is sitting at a
+    // human gate; once notified it can emit a clearing event before this
+    // handler re-reads the projection, so the response can legitimately
+    // observe either the still-pending "cancel" or a null where the worker
+    // already consumed it. Durable convergence is asserted below.
+    let pending_control = &body["pending_control"];
+    assert!(
+        pending_control == "cancel" || pending_control.is_null(),
+        "expected pending_control to be \"cancel\" or null, got {pending_control}"
+    );
 
     // Verify the durable store view converges to cancelled failure.
     let body = wait_for_run_state(&app, &run_id, "failed", "cancelled").await;

lib/crates/fabro-test/src/lib.rs

@@ -16,7 +16,6 @@ use assert_cmd::Command;
 use fabro_config::Storage;
 use fabro_types::RunId;
 use regex::Regex;
-use serde::Serialize;
 use serde_json::{Map, Value, json};
 use toml::Value as TomlValue;
 use toml::map::Map as TomlMap;
@@ -294,18 +293,35 @@ enum SessionMode {
     Process,
 }
 
-#[derive(Debug, Serialize)]
-struct ClientMarker {
-    pid:           u32,
-    touched_at_ms: u128,
-}
-
 static SESSION_REFS: OnceLock<Mutex<HashMap<PathBuf, usize>>> = OnceLock::new();
 
 fn session_refs() -> &'static Mutex<HashMap<PathBuf, usize>> {
     SESSION_REFS.get_or_init(|| Mutex::new(HashMap::new()))
 }
 
+// Gate the stale-session reap so it fires at most once per process. The
+// reap scans /tmp/fx/{n-*,p-*} to clean up after a prior nextest run
+// that crashed; it's correctness-for-safety work that does not need to
+// happen on every TestContext::new. One gate per SessionMode preserves
+// the existing call structure without introducing cross-mode coupling.
+static NEXTEST_REAPED: OnceLock<()> = OnceLock::new();
+static PROCESS_REAPED: OnceLock<()> = OnceLock::new();
+
+// Advisory-lock-based peer-presence marker. Each test process opens
+// `<session_root>/clients/<pid>` once, holds a shared (LOCK_SH) flock
+// for the lifetime of any live TestContext in the process, and
+// releases it explicitly in `cleanup_session_root` when the refcount
+// drops to zero. Peers detect liveness by attempting LOCK_EX: if it
+// succeeds, the owner is gone (normal exit, panic, SIGKILL, or zombie
+// — the kernel releases flocks at process exit in every case), and
+// the stale marker file is removed.
+//
+// The slot stores the marker path so subsequent rebounds in the same
+// process (e.g., after a drop-to-zero followed by a new TestContext)
+// can re-validate the invariant: a process only ever participates in
+// one session root.
+static MARKER_HANDLE: Mutex<Option<(PathBuf, File)>> = Mutex::new(None);
+
 #[expect(
     clippy::disallowed_methods,
     reason = "This synchronous test-support helper uses uuidgen when available to create stable unique case IDs."
@@ -329,13 +345,6 @@ fn test_case_id() -> String {
     ulid
 }
 
-fn current_timestamp_ms() -> u128 {
-    SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .expect("system time should be after unix epoch")
-        .as_millis()
-}
-
 fn current_pid() -> u32 {
     std::process::id()
 }
@@ -467,6 +476,18 @@ fn with_session_lock<T>(root: &Path, f: impl FnOnce() -> T) -> T {
     result
 }
 
+// Iterate `<root>/clients/` and for each marker file attempt a
+// non-blocking exclusive flock. Success means the previous owner has
+// released the lock (normal exit, panic, SIGKILL, or zombie — the
+// kernel releases flocks at process exit regardless), and the stale
+// marker file is removed. `EWOULDBLOCK` means the owner is still
+// alive and holding LOCK_SH.
+//
+// Same-process subtlety: if our own PID's marker file is in the
+// listing, we have LOCK_SH on it via `MARKER_HANDLE`. On Linux and
+// macOS, flock locks are per-open-file-description, so a fresh
+// `open()` here returns an FD that sees the shared lock and correctly
+// reports EWOULDBLOCK when asked for LOCK_EX.
 fn live_marker_count(root: &Path) -> usize {
     let clients_dir = session_clients_dir(root);
     let Ok(entries) = std::fs::read_dir(&clients_dir) else {
@@ -480,30 +501,73 @@ fn live_marker_count(root: &Path) -> usize {
                 .to_string_lossy()
                 .parse::<u32>()
                 .ok()
-                .map(|pid| (pid, entry.path()))
+                .map(|_pid| entry.path())
         })
-        .filter(|(pid, path)| {
-            if fabro_proc::process_running(*pid) {
-                true
-            } else {
+        .filter(|path| {
+            // Open read-write so LOCK_EX has the access mode it expects
+            // on the widest set of platforms. If the file is missing
+            // between read_dir and open, treat it as already gone.
+            let Ok(file) = std::fs::OpenOptions::new()
+                .read(true)
+                .write(true)
+                .open(path)
+            else {
+                return false;
+            };
+            // If the lock is acquired, the previous owner is gone; drop
+            // the file handle (releasing our just-acquired lock) and
+            // remove the marker. Anything else (`Ok(false)` meaning
+            // still held, `Err(_)` for unexpected IO errors) is treated
+            // conservatively as live.
+            if matches!(fabro_proc::try_flock_exclusive(&file), Ok(true)) {
+                drop(file);
                 let _ = std::fs::remove_file(path);
                 false
+            } else {
+                true
             }
         })
         .count()
 }
 
+// Open (or reopen, after a session-drop-to-zero) the per-process
+// marker file and hold a shared advisory lock on it in
+// `MARKER_HANDLE`. Called from inside the `with_session_lock` block
+// of `TestContext::new`.
 fn write_marker(root: &Path) {
-    let marker = ClientMarker {
-        pid:           current_pid(),
-        touched_at_ms: current_timestamp_ms(),
-    };
-    let marker_path = session_marker_path(root, marker.pid);
+    let marker_path = session_marker_path(root, current_pid());
     ensure_parent_dir(&marker_path);
-    let contents =
-        serde_json::to_vec(&marker).expect("client marker should serialize to JSON bytes");
-    std::fs::write(&marker_path, contents)
-        .unwrap_or_else(|err| panic!("failed to write {}: {err}", marker_path.display()));
+
+    let mut slot = MARKER_HANDLE.lock().expect("MARKER_HANDLE lock poisoned");
+
+    if let Some((existing_path, _)) = slot.as_ref() {
+        debug_assert_eq!(
+            existing_path, &marker_path,
+            "marker handle path drifted — session root changed mid-process?"
+        );
+        if marker_path.exists() {
+            return;
+        }
+        // Marker was removed (e.g., by a peer's reap) while we thought
+        // we still owned it. Re-establish by replacing the handle.
+        slot.take();
+    }
+
+    let file = std::fs::OpenOptions::new()
+        .create(true)
+        .truncate(true)
+        .read(true)
+        .write(true)
+        .open(&marker_path)
+        .unwrap_or_else(|err| panic!("failed to open {}: {err}", marker_path.display()));
+    let acquired = fabro_proc::try_flock_shared(&file)
+        .unwrap_or_else(|err| panic!("failed to flock {}: {err}", marker_path.display()));
+    assert!(
+        acquired,
+        "unexpected contention acquiring LOCK_SH on freshly created marker {}",
+        marker_path.display()
+    );
+    *slot = Some((marker_path, file));
 }
 
 fn managed_storage_settings(storage_dir: &Path, rest: &str) -> String {
@@ -898,6 +962,17 @@ fn reap_isolated_servers(root: &Path) {
 
 fn cleanup_session_root(root: &Path) {
     with_session_lock(root, || {
+        // Release our own advisory lock first so `live_marker_count`
+        // below can observe that no one is holding the marker file,
+        // then remove the file. Order matters: if we unlinked before
+        // dropping the handle, peers would still see our LOCK_SH on
+        // the (now-unlinked but still open) inode and count us as
+        // live, preventing the server teardown.
+        {
+            let mut slot = MARKER_HANDLE.lock().expect("MARKER_HANDLE lock poisoned");
+            // Dropping the File closes the FD and releases LOCK_SH.
+            slot.take();
+        }
         let marker_path = session_marker_path(root, current_pid());
         let _ = std::fs::remove_file(&marker_path);
         let live_count = live_marker_count(root);
@@ -964,8 +1039,8 @@ impl TestContext {
             .expect("failed to create temp dir");
         let root_path = context_root.path().to_path_buf();
         let (_, test_run_id, session_paths) = session_paths();
-        reap_stale_session_roots(SessionMode::Nextest);
-        reap_stale_session_roots(SessionMode::Process);
+        NEXTEST_REAPED.get_or_init(|| reap_stale_session_roots(SessionMode::Nextest));
+        PROCESS_REAPED.get_or_init(|| reap_stale_session_roots(SessionMode::Process));
         with_session_lock(&session_paths.root, || {
             std::fs::create_dir_all(session_clients_dir(&session_paths.root)).unwrap_or_else(
                 |err| {