fix(sessions): compose verified Codex rollout continuations

crates/freshell-codex/src/sidecar_test_support.rs

@@ -131,7 +131,7 @@ pub(crate) fn store_in(dir: &tempfile::TempDir) -> Arc<CodexSidecarStore> {
 
 /// The committed fake app-server fixture (repo-owned test harness).
 pub(crate) fn fake_app_server_fixture() -> std::path::PathBuf {
-    std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+    std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
         .join("../../test/fixtures/coding-cli/codex-app-server/fake-app-server.mjs")
 }
 

crates/freshell-codex/tests/launch_lifecycle.rs

@@ -753,10 +753,11 @@ async fn manager_exit_for_unknown_terminal_is_a_noop() {
 // ── D-C-R sidecar planning budget (S5.e precondition) ─────────────────────────────
 
 /// A [`FakeRuntime`]-shaped runtime whose `ensure_ready` blocks on a shared
-/// [`tokio::sync::Notify`] so plans stay in flight until the test releases
-/// them — the knob that keeps budget permits occupied.
+/// zero-permit semaphore so plans stay in flight until the test releases
+/// them. Release permits survive the handoff to a plan that has not entered
+/// `ensure_ready` yet.
 struct BlockingRuntime {
-    release: Arc<tokio::sync::Notify>,
+    release: Arc<tokio::sync::Semaphore>,
 }
 
 impl CodexLaunchRuntime for BlockingRuntime {
@@ -765,7 +766,11 @@ impl CodexLaunchRuntime for BlockingRuntime {
         cwd: Option<String>,
     ) -> BoxFuture<'_, Result<CodexRuntimeReady, String>> {
         Box::pin(async move {
-            self.release.notified().await;
+            self.release
+                .acquire()
+                .await
+                .expect("test release gate remains open")
+                .forget();
             // Released: stand up the file's real loopback echo upstream so
             // the plan completes against a real socket.
             let inner = FakeRuntime::start().await;
@@ -788,9 +793,9 @@ impl CodexLaunchRuntime for BlockingRuntime {
 
 fn blocking_test_runtime_factory() -> (
     freshell_codex::launch_lifecycle::CodexRuntimeFactory,
-    Arc<tokio::sync::Notify>,
+    Arc<tokio::sync::Semaphore>,
 ) {
-    let release = Arc::new(tokio::sync::Notify::new());
+    let release = Arc::new(tokio::sync::Semaphore::new(0));
     let factory_release = release.clone();
     let factory: freshell_codex::launch_lifecycle::CodexRuntimeFactory = Box::new(move |_plan| {
         let rt = Arc::new(BlockingRuntime {
@@ -840,9 +845,11 @@ async fn third_concurrent_plan_fails_fast_on_the_sidecar_budget() {
         err.to_string().contains("planning budget exhausted"),
         "{err}"
     );
-    release.notify_waiters();
-    let _ = a.await;
-    let _ = b.await;
+    release.add_permits(2);
+    let launch_a = a.await.expect("join").expect("first plan completes");
+    let launch_b = b.await.expect("join").expect("second plan completes");
+    manager.discard(launch_a).await;
+    manager.discard(launch_b).await;
 }
 
 // ── graceful restore/resume S1 (P2): restore-class plans queue, never die ─────────
@@ -1060,20 +1067,20 @@ async fn restore_class_queue_overflow_fails_loud_as_queue_full() {
         ),
         "{err}"
     );
-    // Drain: release the parked plans (BlockingRuntime parks on a Notify;
-    // the queued waiter parks again after the holder finishes, so notify twice).
-    release.notify_waiters();
+    // Release both plans before either join. The queued waiter enters the
+    // runtime only after the holder gives up the planning permit.
+    release.add_permits(2);
     let launch = holder.await.expect("join").expect("holder plan completes");
     manager.discard(launch).await;
-    release.notify_waiters();
     let launch2 = queued.await.expect("join").expect("queued plan completes");
     manager.discard(launch2).await;
+    assert_eq!(manager.plan_queue_depth(), 0, "plan queue drained");
 }
 
 // ── the spawn integration leg: real child + real proxy + fake TUI ─────────────────
 
 fn fake_app_server_command() -> String {
-    let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+    let fixture = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
         .join("../../test/fixtures/coding-cli/codex-app-server/fake-app-server.mjs");
     format!("node {}", fixture.display())
 }
@@ -1441,7 +1448,7 @@ async fn plan_retry_spawns_fresh_after_claimed_reattach_ensure_ready_fails() {
     // `sidecar_reconcile_tests::spawn_own_fake_app_server`; test binaries
     // cannot share code — the repo's copy-with-attribution convention).
     let survivor_ownership = "codex-sidecar-a7000003-cccc-4ccc-8ccc-cccccccccccc";
-    let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+    let fixture = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
         .join("../../test/fixtures/coding-cli/codex-app-server/fake-app-server.mjs");
     let bind_unused_ws_url = || {
         let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");

crates/freshell-server/src/extensions.rs

@@ -608,74 +608,11 @@ mod tests {
         // actually fire (absence from the registry alone was also true of the
         // old lenient port).
         //
-        // Capture strategy (matching freshell-freshagent's documented
-        // investigation): a set_global_default subscriber installed EXACTLY
-        // ONCE per test binary via OnceLock. Thread-local set_default proved
-        // nondeterministic under parallel `cargo test` (callsite interest
-        // caching); the global layer observes every event, and this test
-        // filters by its unique temp manifest_path.
-        use std::collections::BTreeMap;
-        use std::sync::{Arc, Mutex, OnceLock};
-        use tracing::field::{Field, Visit};
-        use tracing::{Event, Level, Subscriber};
-        use tracing_subscriber::layer::{Context, Layer, SubscriberExt};
-
-        struct Captured {
-            message: String,
-            fields: BTreeMap<String, String>,
-        }
-        #[derive(Default)]
-        struct V {
-            message: String,
-            fields: BTreeMap<String, String>,
-        }
-        impl Visit for V {
-            fn record_debug(&mut self, f: &Field, v: &dyn std::fmt::Debug) {
-                let r = format!("{v:?}");
-                if f.name() == "message" {
-                    self.message = r;
-                } else {
-                    self.fields.insert(f.name().into(), r);
-                }
-            }
-            fn record_str(&mut self, f: &Field, v: &str) {
-                if f.name() == "message" {
-                    self.message = v.into();
-                } else {
-                    self.fields.insert(f.name().into(), v.into());
-                }
-            }
-        }
-        struct CaptureLayer {
-            events: Arc<Mutex<Vec<Captured>>>,
-        }
-        impl<S: Subscriber> Layer<S> for CaptureLayer {
-            fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
-                if *event.metadata().level() != Level::WARN {
-                    return;
-                }
-                let mut v = V::default();
-                event.record(&mut v);
-                self.events.lock().expect("capture lock").push(Captured {
-                    message: v.message,
-                    fields: v.fields,
-                });
-            }
-        }
-
-        static GLOBAL_EVENTS: OnceLock<Arc<Mutex<Vec<Captured>>>> = OnceLock::new();
-        let events = GLOBAL_EVENTS.get_or_init(|| {
-            let events = Arc::new(Mutex::new(Vec::new()));
-            let layer = CaptureLayer {
-                events: Arc::clone(&events),
-            };
-            // Ignore the error case: some OTHER test installed a global
-            // subscriber first — then this assertion would fail noisily
-            // below, but no freshell-server test does that today.
-            let _ =
-                tracing::subscriber::set_global_default(tracing_subscriber::registry().with(layer));
-            events
-        });
+        // A shared process-global collector avoids competing test modules
+        // installing subscribers with disconnected event buffers. Thread-local
+        // set_default proved nondeterministic under parallel cargo test because
+        // of callsite interest caching; this test filters by its unique path.
+        let events = crate::test_trace_capture::captured_trace_events();
 
         let root = tmp();
         let bad_json = root.join("bad-json");
@@ -693,15 +630,14 @@ mod tests {
 
         let root_marker = root.display().to_string();
         let mine: Vec<String> = events
-            .lock()
-            .expect("capture lock")
-            .iter()
+            .snapshot()
+            .into_iter()
             .filter(|e| {
                 e.fields
                     .get("manifest_path")
                     .is_some_and(|p| p.contains(&root_marker))
             })
-            .map(|e| e.message.clone())
+            .map(|e| e.message)
             .collect();
         assert_eq!(
             mine.len(),

crates/freshell-server/src/main.rs

@@ -80,6 +80,8 @@ pub(crate) mod test_clock_gate;
 mod test_clock_router;
 #[cfg(test)]
 pub(crate) mod test_env_lock;
+#[cfg(test)]
+mod test_trace_capture;
 mod updater;
 
 use std::net::IpAddr;
@@ -2745,6 +2747,7 @@ async fn main() -> ExitCode {
         metadata: session_metadata_store.clone(),
         // STATUS-STRIP: sessions.cloned pages are client-ordered per instance.
         server_instance: Arc::clone(&server_instance_id),
+        collision_signatures: Arc::default(),
         // Unified agent names (Task 7 review M1): captured AFTER the boot
         // consolidation above ran — once the receipt committed, a scoped
         // coding-agent row's displayed title never consults the migrated