From c9cd2295bd1cbd6f562b72735cece55a90845ce8 Mon Sep 17 00:00:00 2001 From: isactornberg Date: Fri, 2 Oct 2026 10:45:46 +0200 Subject: [PATCH 1/2] The daemon lets go of a client the moment it hangs up instead of at the next broadcast, and a client that subscribes no longer misses a change made while its Snapshot was taken - A subscribed client that disconnected (a TUI that quit, the idle check `nebula upgrade` makes) kept its socket open in the daemon until the next broadcast: the task forwarding broadcasts to it held a sender and only noticed on its next send, so the connection's writer never ended. An idle daemon collected one socket per such client. `handle_client` now keeps that task's handle and aborts it when the client's requests end, the way it aborts the attach forwarders. A second `Subscribe` on one connection replaces the first forwarder instead of adding one. - `Subscribe` now subscribes to the broadcasts before it takes the Snapshot, not after. A status change or an upsert made between the two used to reach nobody; now it arrives after the Snapshot. A client folds those in by id, so one the Snapshot already carries changes nothing. Tests: a_subscribed_client_that_hangs_up_is_let_go_by_an_idle_daemon (fails without the fix: the daemon never closes its half). --- crates/nebula-daemon/src/server.rs | 18 +++++++++++++++-- crates/nebula/tests/e2e_pty.rs | 31 ++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/crates/nebula-daemon/src/server.rs b/crates/nebula-daemon/src/server.rs index 27d5b918..e2a440e2 100644 --- a/crates/nebula-daemon/src/server.rs +++ b/crates/nebula-daemon/src/server.rs @@ -58,6 +58,8 @@ async fn handle_client(daemon: Arc, stream: UnixStream) -> Result<()> { // Per-connection attach state: forward-task handles keyed by session. let mut attached: HashMap> = HashMap::new(); + // The task forwarding broadcasts to a subscribed client. + let mut subscription: Option> = None; let mut handshaken = false; let result: Result<()> = async { @@ -91,6 +93,10 @@ async fn handle_client(daemon: Arc, stream: UnixStream) -> Result<()> { break; } ClientRequest::Subscribe => { + // Subscribed before the snapshot is taken, so a change + // made between the two is not lost: it arrives after the + // Snapshot, where a client folds it in by id. + let mut rx = daemon.events.subscribe(); let snapshot = daemon.snapshot().unwrap_or(ServerEvent::Snapshot { projects: vec![], worktrees: vec![], @@ -101,9 +107,8 @@ async fn handle_client(daemon: Arc, stream: UnixStream) -> Result<()> { ui_state: None, }); let _ = out_tx.send(snapshot).await; - let mut rx = daemon.events.subscribe(); let tx = out_tx.clone(); - tokio::spawn(async move { + let forward = tokio::spawn(async move { loop { match rx.recv().await { Ok(ev) => { @@ -118,6 +123,9 @@ async fn handle_client(daemon: Arc, stream: UnixStream) -> Result<()> { } } }); + if let Some(old) = subscription.replace(forward) { + old.abort(); + } } ClientRequest::Attach { session: sref, @@ -592,6 +600,12 @@ async fn handle_client(daemon: Arc, stream: UnixStream) -> Result<()> { h.abort(); daemon.note_detached(&sref); } + // The forwarder holds a sender and waits on the next broadcast: left + // running, it keeps the writer, and this client's socket, open until + // the daemon next has something to say. + if let Some(forward) = subscription { + forward.abort(); + } drop(out_tx); let _ = writer_task.await; result diff --git a/crates/nebula/tests/e2e_pty.rs b/crates/nebula/tests/e2e_pty.rs index 81033661..3386bbd7 100644 --- a/crates/nebula/tests/e2e_pty.rs +++ b/crates/nebula/tests/e2e_pty.rs @@ -3994,6 +3994,37 @@ async fn subscribe(c: &mut UnixStream) -> Vec { .await } +/// A subscribed client that hangs up is let go at once, with nothing +/// broadcast in between: the daemon closes its side of the connection as +/// soon as it reads the hang-up. It used to keep the socket open until the +/// next broadcast, so an idle daemon collected one per one-shot client. +#[tokio::test] +async fn a_subscribed_client_that_hangs_up_is_let_go_by_an_idle_daemon() { + use tokio::io::AsyncWriteExt; + let env = TestEnv::new(); + let mut daemon = env.spawn_daemon(); + let mut c = connect(&env.sock()).await; + handshake(&mut c).await; + subscribe(&mut c).await; + + // Hang up the writing half only: the daemon reads the end of the + // requests, and this half stays open to see the daemon close its own. + c.shutdown().await.unwrap(); + let closed = tokio::time::timeout(EVENT_TIMEOUT, async { + while let Ok(Some(_)) = read_frame::(&mut c).await {} + }) + .await; + assert!( + closed.is_ok(), + "the daemon kept a hung-up subscriber's connection open" + ); + + let mut c = connect(&env.sock()).await; + handshake(&mut c).await; + write_frame(&mut c, &ClientRequest::Shutdown).await.unwrap(); + wait_for_exit(&mut daemon); +} + /// `nebula spawn ""` from inside a session, end to end over real /// processes: the CLI (what the model runs) makes the daemon start a second /// agent in the caller's worktree — booted at once, on the default name so From 231d62c32a612f864257097b7ca1f2d649b6039e Mon Sep 17 00:00:00 2001 From: isactornberg Date: Fri, 2 Oct 2026 10:54:39 +0200 Subject: [PATCH 2/2] Two e2e tests stop depending on the machine they run on: the sibling-spawn test keeps Codex's hooks out of the developer's own ~/.codex, and the CRUD test no longer takes the echo of its typed line for the command's output - nebula_spawn_cli_starts_a_sibling_session_in_the_same_worktree spawns a Codex sibling, and a Codex spawn installs nebula's managed hooks into Codex's home. The test's daemon ran with no `CODEX_HOME`, so a run installed them into `~/.codex/hooks.json` on the machine it ran on. It now pins `CODEX_HOME` to its own temp dir, as codex_hooks_install_and_drive_status already does. - full_crud_attach_and_restart_persistence typed `echo ; pwd` and read on until the marker had shown twice, taking the second for the command's output. A line typed before the shell's prompt is up is echoed by the tty and again when the line editor redraws it, so the marker showed twice before the command had run, and the cwd assertion that follows failed: about 1 run in 20 on its own, more with the rest of the file running. It now types `pwd; echo ` and waits for the marker at the start of a line, which only the command prints, after `pwd`'s output. Test-only. --- crates/nebula/tests/e2e_pty.rs | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/crates/nebula/tests/e2e_pty.rs b/crates/nebula/tests/e2e_pty.rs index 3386bbd7..2d7964b5 100644 --- a/crates/nebula/tests/e2e_pty.rs +++ b/crates/nebula/tests/e2e_pty.rs @@ -358,14 +358,17 @@ async fn full_crud_attach_and_restart_persistence() { &mut c, &ClientRequest::Input { session: sref.clone(), - data: format!("echo {marker}; pwd\n").into_bytes(), + data: format!("pwd; echo {marker}\n").into_bytes(), }, ) .await .unwrap(); let events = read_events_until(&mut c, EVENT_TIMEOUT, |evs| { let text = String::from_utf8_lossy(&collected_output(evs)).into_owned(); - text.matches(marker).count() >= 2 + // The typed line is echoed, and echoed again when the shell's line + // editor redraws what was typed before its prompt. Only a marker at + // the start of a line is the command's own, printed after `pwd`'s. + text.contains(&format!("\n{marker}")) }) .await; // The shell runs in the worktree directory. @@ -4049,7 +4052,13 @@ async fn nebula_spawn_cli_starts_a_sibling_session_in_the_same_worktree() { ) .unwrap(); make_executable(&script); - let mut daemon = env.spawn_daemon_with_agent_cmd(script.to_str().unwrap()); + // A Codex sibling installs Codex's managed hooks into Codex's home: + // this test's own, never the developer's `~/.codex`. + let codex_home = env.tmp.path().join("codex-home"); + let mut daemon = env.spawn_daemon_with( + script.to_str().unwrap(), + &[(env::CODEX_HOME, codex_home.to_str().unwrap())], + ); let mut c = connect(&env.sock()).await; handshake(&mut c).await;