Feat/cs/stability

src/app.rs

@@ -341,11 +341,12 @@ impl App {
     /// the socket.
     pub fn connect(rows: u16, cols: u16) -> io::Result<Self> {
         let (stream, origin) = crate::daemon::connect_ready()?;
-        // Split the stream here (the fallible part) so the transport factory in
-        // `assemble` (which owns the wake sender) stays infallible.
+        // Create the reader and control handles before `assemble`: its
+        // transport factory cannot return an `io::Result`.
         let read = stream.try_clone()?;
+        let ctrl = stream.try_clone()?;
         let mut app = Self::assemble(rows, cols, move |_, _, wait_tx| {
-            Box::new(SocketTransport::from_halves(stream, read, wait_tx))
+            Box::new(SocketTransport::from_halves(stream, read, ctrl, wait_tx))
         });
         app.daemon_backed = true;
         // Report when a running daemon could not apply this invocation's
@@ -361,7 +362,8 @@ impl App {
             // Reconnection must not block the active UI indefinitely.
             let stream = crate::daemon::connect_ready_bounded()?;
             let read = stream.try_clone()?;
-            Ok(SocketTransport::from_halves(stream, read, wait_tx))
+            let ctrl = stream.try_clone()?;
+            Ok(SocketTransport::from_halves(stream, read, ctrl, wait_tx))
         };
         match build() {
             Ok(t) => {

src/app_tests.rs

@@ -104,21 +104,39 @@ fn shift(code: KeyCode) -> KeyEvent {
     KeyEvent::new(code, KeyModifiers::SHIFT)
 }
 
+/// Construct a stable live task without spawning a child. `Active` and
+/// unparked keep an untagged view in Running; a child can exit during the test
+/// and move its row to Completed.
+fn view(id: u64, cwd: PathBuf, tagged: bool, group: Option<&str>) -> TaskView {
+    TaskView {
+        id,
+        command: "true".to_string(),
+        cwd,
+        tagged,
+        group: group.map(str::to_string),
+        name: None,
+        lifecycle: Lifecycle::Active,
+        parked: false,
+        preview: Preview::floor(String::new()),
+        started_ago: Duration::ZERO,
+        quiet_ago: Some(Duration::ZERO),
+        finished_ago: None,
+    }
+}
+
 /// Selection is bound to a task id, so a reorder (here: tagging a task into
 /// the "In use" bucket) must not move the highlight to a different task.
 #[test]
 fn selection_follows_task_across_reorder() {
     let mut app = App::new_local(30, 100);
     let dir = app.invocation_dir.clone();
-    app.spawn_in("sleep 5", dir.clone()); // id 1
-    app.spawn_in("sleep 5", dir); // id 2
-    app.pump();
+    app.views = vec![view(1, dir.clone(), false, None), view(2, dir, false, None)];
     app.resolve_selection();
     assert_eq!(app.selected_id, Some(1));
 
-    // Tag id 2 -> it sorts into the "In use" bucket, ahead of id 1.
-    app.transport.send(Command::Tag { id: 2, on: true });
-    app.pump();
+    // Tagging id 2 moves it into "In use," ahead of id 1. Mutate the injected
+    // snapshot directly: a pump would replace it with the empty core snapshot.
+    app.views[1].tagged = true;
 
     let order = app.display_order();
     assert_eq!(app.views[order[0]].id, 2, "tagged task should sort first");
@@ -134,9 +152,10 @@ fn selection_follows_task_across_reorder() {
 fn dir_mode_groups_by_cwd() {
     let mut app = App::new_local(30, 100);
     let inv = app.invocation_dir.clone();
-    app.spawn_in("sleep 5", inv); // id 1, invocation dir
-    app.spawn_in("sleep 5", PathBuf::from("/tmp")); // id 2, /tmp
-    app.pump();
+    app.views = vec![
+        view(1, inv, false, None),                   // invocation dir
+        view(2, PathBuf::from("/tmp"), false, None), // /tmp
+    ];
 
     app.group_mode = GroupMode::State;
     let s = app.sections();
@@ -2169,11 +2188,11 @@ fn find_palette_opens_on_slash_only_with_tasks() {
 fn find_candidates_follow_display_order() {
     let mut app = App::new_local(30, 100);
     let inv = app.invocation_dir.clone();
-    app.spawn_in("sleep 5", inv.clone()); // id 1
-    app.spawn_in("sleep 5", inv.clone()); // id 2
-    app.spawn_in("sleep 5", inv); // id 3
-    app.transport.send(Command::Tag { id: 3, on: true });
-    app.pump();
+    app.views = vec![
+        view(1, inv.clone(), false, None),
+        view(2, inv.clone(), false, None),
+        view(3, inv, true, None),
+    ];
     let order: Vec<u64> = app
         .display_order()
         .into_iter()
@@ -2191,9 +2210,10 @@ fn find_candidates_follow_display_order() {
 fn find_empty_input_lists_every_task() {
     let mut app = App::new_local(30, 100);
     let inv = app.invocation_dir.clone();
-    app.spawn_in("sleep 5", inv.clone()); // id 1
-    app.spawn_grouped("sleep 5", inv, "alpha"); // id 2
-    app.pump();
+    app.views = vec![
+        view(1, inv.clone(), false, None),
+        view(2, inv, false, Some("alpha")),
+    ];
     app.on_key_dashboard(key(KeyCode::Char('/')));
     assert_eq!(find_ids(&app), vec![1, 2]);
 

src/daemon.rs

@@ -17,7 +17,7 @@
 
 use std::{
     fs,
-    io::{self, ErrorKind, Read, Write},
+    io::{self, ErrorKind, Read, Seek, Write},
     net::Shutdown,
     os::unix::{
         fs::{DirBuilderExt, MetadataExt, PermissionsExt},
@@ -33,7 +33,7 @@ use std::{
         mpsc::channel,
     },
     thread,
-    time::{Duration, Instant},
+    time::Duration,
 };
 
 use nix::{
@@ -48,7 +48,7 @@ use crate::{
     path::FLEETCOM_RUNTIME_DIR,
     protocol::{
         Command, Event, LaunchContext, PROTOCOL_VERSION, decode_command, decode_event,
-        decode_hello, encode_command, encode_event, encode_hello, hello_version,
+        decode_hello, encode_event, encode_hello, hello_version,
     },
     supervisor::{self, Supervisor},
 };
@@ -367,12 +367,18 @@ fn no_daemon() -> io::Result<()> {
 /// `--kill` must work while someone else is attached. The pid comes from the
 /// lock file (trustworthy while the flock is held: the holder wrote it), and
 /// daemon exit releases the flock, so acquiring it is the completion signal.
-/// A no-op (with a message) if no daemon is running.
+/// A no-op (with a message) if no daemon is running. A held flock without a
+/// usable pid is an error: it may be the interval between lock acquisition and
+/// pid publication, so it cannot be treated as the no-daemon case.
 pub fn run_kill() -> io::Result<()> {
-    let dir = runtime_dir();
+    run_kill_in(&runtime_dir())
+}
+
+/// Run the `--kill` operation against an explicit runtime directory.
+fn run_kill_in(dir: &Path) -> io::Result<()> {
     // The lock PID is a signal target, and the socket receives the client's
     // environment, so validate the directory before reading either file.
-    ensure_runtime_dir(&dir)?;
+    ensure_runtime_dir(dir)?;
     let lock_path = dir.join("daemon.lock");
     let Ok(file) = fs::OpenOptions::new()
         .read(true)
@@ -387,13 +393,25 @@ pub fn run_kill() -> io::Result<()> {
         Err((file, _)) => file,
     };
 
-    let mut pid_str = String::new();
-    file.read_to_string(&mut pid_str)?;
-    let Some(pid) = crate::task::positive_pid(pid_str.trim()) else {
-        // Without a usable pid, fall back to a Shutdown frame over the socket.
-        // Bound the fallback because an attached client can keep the daemon
-        // from accepting this connection.
-        return kill_via_socket_at(&socket_path(), KILL_SOCKET_TIMEOUT);
+    // `run_daemon` acquires the flock before replacing the pid. Retry briefly
+    // to cover that publication interval.
+    let mut pid = None;
+    for _ in 0..20 {
+        let mut pid_str = String::new();
+        file.seek(io::SeekFrom::Start(0))?;
+        file.read_to_string(&mut pid_str)?;
+        pid = crate::task::positive_pid(pid_str.trim());
+        if pid.is_some() {
+            break;
+        }
+        thread::sleep(Duration::from_millis(10));
+    }
+    let Some(pid) = pid else {
+        return Err(io::Error::new(
+            ErrorKind::TimedOut,
+            "the daemon holds the lock but has not written its pid (it may \
+             still be starting); retry",
+        ));
     };
 
     // ESRCH means the daemon exited between the lock probe and here; the flock
@@ -419,93 +437,6 @@ pub fn run_kill() -> io::Result<()> {
     ))
 }
 
-/// Timeout applied to blocking socket-fallback kill operations.
-const KILL_SOCKET_TIMEOUT: Duration = Duration::from_secs(10);
-
-/// Timeout reported when the socket-fallback kill exchange does not finish.
-fn kill_handshake_timeout() -> io::Error {
-    io::Error::new(
-        ErrorKind::TimedOut,
-        "the daemon is running but did not complete the kill handshake in \
-         time (another client may be attached); retry after it detaches, or \
-         send SIGTERM to the daemon process directly",
-    )
-}
-
-/// Set the read timeout to the remaining deadline budget.
-fn arm_read_deadline(s: &UnixStream, deadline: Instant) -> io::Result<()> {
-    let left = deadline.saturating_duration_since(Instant::now());
-    if left.is_zero() {
-        return Err(kill_handshake_timeout());
-    }
-    s.set_read_timeout(Some(left))
-}
-
-/// Convert either platform representation of a socket timeout into the
-/// kill-handshake timeout.
-fn deadline_mapped(e: io::Error) -> io::Error {
-    if is_timeout(&e) {
-        kill_handshake_timeout()
-    } else {
-        e
-    }
-}
-
-/// When the lock lacks a valid PID, send `Shutdown` over the socket and bound
-/// handshake and completion I/O by `budget`.
-fn kill_via_socket_at(path: &Path, budget: Duration) -> io::Result<()> {
-    match UnixStream::connect(path) {
-        Ok(mut s) => {
-            let (kind, payload) = encode_hello(&LaunchContext::here());
-            kill_exchange(&mut s, budget, kind, &payload)
-        }
-        Err(_) => no_daemon(),
-    }
-}
-
-/// Drive the bounded Shutdown exchange with a pre-encoded hello frame.
-fn kill_exchange(
-    s: &mut UnixStream,
-    budget: Duration,
-    hello_kind: u8,
-    hello_payload: &[u8],
-) -> io::Result<()> {
-    let deadline = Instant::now() + budget;
-    s.set_write_timeout(Some(budget))?;
-
-    write_frame(s, hello_kind, hello_payload).map_err(deadline_mapped)?;
-    arm_read_deadline(s, deadline)?;
-    let (kind, payload) = read_frame(s).map_err(deadline_mapped)?;
-    check_hello_ack(kind, &payload)?;
-
-    let (kind, payload) = encode_command(&Command::Shutdown);
-    write_frame(s, kind, &payload).map_err(deadline_mapped)?;
-    // Socket closure signals completion. Re-arm each read with the remaining
-    // budget so the loop cannot outlive the deadline.
-    let mut buf = [0u8; 256];
-    loop {
-        arm_read_deadline(s, deadline)?;
-        match s.read(&mut buf) {
-            Ok(0) => return Ok(()),
-            Ok(_) => {}
-            Err(e) if is_timeout(&e) => return Err(kill_handshake_timeout()),
-            // Retry interrupted reads; the deadline still bounds the loop.
-            Err(e) if e.kind() == ErrorKind::Interrupted => {}
-            // The daemon closing mid-drain is completion, same as `Ok(0)`.
-            Err(e)
-                if matches!(
-                    e.kind(),
-                    ErrorKind::ConnectionReset | ErrorKind::BrokenPipe | ErrorKind::UnexpectedEof
-                ) =>
-            {
-                return Ok(());
-            }
-            // Propagate other errors because they do not confirm daemon exit.
-            Err(e) => return Err(e),
-        }
-    }
-}
-
 /// The daemon entry point (`fleetcom --daemon`). Binds the socket and serves clients
 /// until an explicit shutdown. The supervisor is created once and persists across
 /// reconnects: tasks outlive any single client.
@@ -923,66 +854,25 @@ mod tests {
         assert!(notice.contains("--kill"), "{notice}");
     }
 
-    /// A missing hello response times out the kill exchange.
+    /// A held flock without a usable pid is an error: an existing lock file
+    /// reaches the no-daemon path only when its flock is acquirable.
     #[test]
-    fn kill_via_socket_bounds_the_handshake_wait() {
-        let base = temp("kill_socket_mute");
-        fs::create_dir_all(&*base).unwrap();
-        let sock = base.join("mute.sock");
-        // Leave the connection queued in the listener backlog.
-        let _listener = UnixListener::bind(&sock).unwrap();
-        let start = Instant::now();
-        let err = kill_via_socket_at(&sock, Duration::from_millis(200)).unwrap_err();
+    fn kill_with_a_held_lock_and_no_pid_is_an_error() {
+        let base = temp("kill_lock_no_pid");
+        let dir = base.join("runtime");
+        ensure_runtime_dir(&dir).unwrap();
+        let holder = fs::OpenOptions::new()
+            .create(true)
+            .write(true)
+            .truncate(false)
+            .open(dir.join("daemon.lock"))
+            .unwrap();
+        // The second open creates a distinct open-file description, so it
+        // contends with this lock even within one process.
+        let _held = Flock::lock(holder, FlockArg::LockExclusiveNonblock).unwrap();
+        let err = run_kill_in(&dir).unwrap_err();
         assert_eq!(err.kind(), ErrorKind::TimedOut);
-        assert!(err.to_string().contains("kill handshake"), "{err}");
-        assert!(
-            start.elapsed() < Duration::from_secs(5),
-            "the deadline must fire, not the test's timeout"
-        );
-    }
-
-    /// A blocked hello write reports the kill-handshake timeout.
-    #[test]
-    fn kill_exchange_maps_a_write_timeout() {
-        let base = temp("kill_socket_bigenv");
-        fs::create_dir_all(&*base).unwrap();
-        let sock = base.join("mute.sock");
-        let _listener = UnixListener::bind(&sock).unwrap();
-        let mut s = UnixStream::connect(&sock).unwrap();
-        // The listener never accepts, so this payload fills the send buffer.
-        let oversized = vec![0u8; 8 * 1024 * 1024];
-        let err = kill_exchange(&mut s, Duration::from_millis(200), 0, &oversized).unwrap_err();
-        assert_eq!(err.kind(), ErrorKind::TimedOut);
-        assert!(err.to_string().contains("kill handshake"), "{err}");
-    }
-
-    /// A daemon that keeps the socket open after Shutdown times out the drain.
-    #[test]
-    fn kill_via_socket_bounds_the_drain_wait() {
-        let base = temp("kill_socket_drain");
-        fs::create_dir_all(&*base).unwrap();
-        let sock = base.join("stuck.sock");
-        let listener = UnixListener::bind(&sock).unwrap();
-        let server = thread::spawn(move || {
-            let (mut s, _) = listener.accept().unwrap();
-            let _ = read_frame(&mut s); // hello
-            let (kind, payload) = encode_event(&Event::HelloOk);
-            let _ = write_frame(&mut s, kind, &payload);
-            let _ = read_frame(&mut s); // Shutdown, swallowed
-            // Hold the socket open until the client drops its end.
-            let _ = s.read(&mut [0u8; 16]);
-        });
-        let err = kill_via_socket_at(&sock, Duration::from_millis(300)).unwrap_err();
-        assert_eq!(err.kind(), ErrorKind::TimedOut);
-        server.join().unwrap();
-    }
-
-    /// A missing socket makes the fallback a no-op.
-    #[test]
-    fn kill_via_socket_without_a_socket_is_a_noop() {
-        let base = temp("kill_socket_absent");
-        fs::create_dir_all(&*base).unwrap();
-        assert!(kill_via_socket_at(&base.join("absent.sock"), Duration::from_millis(100)).is_ok());
+        assert!(err.to_string().contains("pid"), "{err}");
     }
 
     /// Remove group and other read/execute permissions from a valid directory.

src/harness/omp.rs

@@ -15,8 +15,8 @@
 //! Sessions live at `<sessions root>/<encoded cwd>/<iso ts>_<uuidv7>.jsonl`.
 //! The harness home *is* the sessions root: `PI_CODING_AGENT_SESSION_DIR`
 //! names a sessions directory outright, so no agent-dir value can express it.
-//! That override also flattens the store — it is passed straight through as the
-//! session file's parent and the bucket level is never computed — so
+//! That override also flattens the store: it is passed straight through as the
+//! session file's parent and the bucket level is never computed, so
 //! correlation scans the root and one level below it.
 //!
 //! Correlation does not derive bucket names. It enumerates the root and its