fix(maestro): replace agent_session_id when a report names a new one

A resume issues a new conversation id, so write-once left the stored id
pointing at the pre-resume conversation and every later resume reattached
it. Rewrite the sidecar when the reported id differs from the stored one;
a repeated report of the same id still writes nothing.

Closes #176

crates/maestro-daemon/src/session.rs

@@ -87,6 +87,11 @@ pub struct SessionRecord {
     /// (`POST /.../focus`), `None` while unfocused. Daemon-local; never rides
     /// the wire.
     pub focused_since: Option<std::time::Instant>,
+    /// Set for as long as a [`SessionManager::resume`] is between its two
+    /// takes of the map lock (#175). The spawn happens with the lock released,
+    /// so the `Running` check alone lets a second request through; this is the
+    /// state the second one is refused on. Daemon-local; never rides the wire.
+    pub resuming: bool,
     /// Sticky acknowledgement applied by [`crate::activity::apply_focus_suppression`]:
     /// once the pane has been focused long enough to clear an `AwaitingInput`
     /// glow, this holds through focus loss until the session does new work.
@@ -595,6 +600,7 @@ impl SessionManager {
                     naming: false,
                     focused_since: None,
                     awaiting_suppressed: false,
+                    resuming: false,
                 },
             );
         }
@@ -646,15 +652,23 @@ impl SessionManager {
         id: Uuid,
         presets: &[maestro_protocol::SpawnPreset],
     ) -> Result<Session, SessionError> {
+        // The spawn below runs with the map lock released, so the `Running`
+        // check alone is not enough to keep two resumes apart (#175): both
+        // would pass it, both would open a `PtyBridge` on the same FIFO, and
+        // the loser would report a tmux failure instead of the conflict it
+        // is. The record is marked in-flight under this first lock, and every
+        // path out of here clears it again.
         let mut session = {
-            let sessions = self.inner.sessions.lock().await;
-            let rec = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
-            if rec.session.status == SessionStatus::Running {
+            let mut sessions = self.inner.sessions.lock().await;
+            let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
+            if rec.session.status == SessionStatus::Running || rec.resuming {
                 return Err(SessionError::AlreadyRunning(id));
             }
+            rec.resuming = true;
             rec.session.clone()
         };
         if !session.cwd.exists() {
+            self.clear_resuming(id).await;
             return Err(SessionError::CwdMissing(session.cwd.clone()));
         }
         let command = crate::resume::resume_command(&session, presets);
@@ -686,6 +700,7 @@ impl SessionManager {
         if let Err(e) = spawned {
             // `bridge` drops here, closing the FIFO reader. Nothing else was
             // touched, so the session stays exactly as dormant as it was.
+            self.clear_resuming(id).await;
             return Err(SessionError::Tmux(e));
         }
 
@@ -705,6 +720,7 @@ impl SessionManager {
             // sidecar does not describe.
             let tmux = self.inner.tmux.clone();
             let _ = tmux_blocking(move || tmux.kill_session(id)).await;
+            self.clear_resuming(id).await;
             return Err(SessionError::Sidecar(e));
         }
 
@@ -712,9 +728,17 @@ impl SessionManager {
             let mut sessions = self.inner.sessions.lock().await;
             let Some(rec) = sessions.get_mut(&id) else {
                 // Killed while we were spawning. Undo the pane instead of
-                // leaving one behind with no record of it.
+                // leaving one behind with no record of it — and delete the
+                // sidecar the write above just put back (#175). The kill
+                // deleted it before that write landed, so leaving it would
+                // have the next recovery register a dormant ghost for a
+                // session that is gone and whose worktree is already removed.
+                drop(sessions);
                 let tmux = self.inner.tmux.clone();
                 let _ = tmux_blocking(move || tmux.kill_session(id)).await;
+                if let Err(e) = self.inner.sidecar.delete(id) {
+                    tracing::warn!(error = ?e, %id, "resume: failed to delete the sidecar of a session killed mid-resume");
+                }
                 return Err(SessionError::NotFound(id));
             };
             rec.session = session.clone();
@@ -724,6 +748,7 @@ impl SessionManager {
             rec.settled = false;
             rec.focused_since = None;
             rec.awaiting_suppressed = false;
+            rec.resuming = false;
         }
         let _ = self.inner.events.send(CatalogEvent::SessionUpdated {
             session: session.clone(),
@@ -731,6 +756,16 @@ impl SessionManager {
         Ok(session)
     }
 
+    /// Drop the in-flight marker [`Self::resume`] set under its first lock
+    /// (#175), so a refused or failed resume can be retried. A no-op when the
+    /// record is already gone: a kill that landed mid-resume took the marker
+    /// with it.
+    async fn clear_resuming(&self, id: Uuid) {
+        if let Some(rec) = self.inner.sessions.lock().await.get_mut(&id) {
+            rec.resuming = false;
+        }
+    }
+
     /// Best-effort removal of a worktree created earlier in `create` when a
     /// later step fails. No-op when the request had no worktree.
     fn rollback_worktree(&self, base: &Option<PathBuf>, path: &Option<PathBuf>) {
@@ -922,6 +957,7 @@ impl SessionManager {
                     naming: false,
                     focused_since: None,
                     awaiting_suppressed: false,
+                    resuming: false,
                 },
             );
             let _ = self
@@ -1048,6 +1084,7 @@ impl SessionManager {
                         naming: false,
                         focused_since: None,
                         awaiting_suppressed: false,
+                        resuming: false,
                     },
                 );
                 registered.push(session);
@@ -1268,10 +1305,12 @@ impl SessionManager {
     /// and stays silent on the event stream at steady state.
     ///
     /// `agent_session_id` (#161) is the agent's own conversation id, parsed
-    /// by the caller from the hook payload's `session_id` field. It is
-    /// written once — the first report that carries one wins — and the
-    /// sidecar is only rewritten on that transition, never on every report,
-    /// since an agent's session id never changes mid-session.
+    /// by the caller from the hook payload's `session_id` field. The
+    /// sidecar is rewritten only when the reported id differs from the
+    /// stored one — a repeated report of the same id is pure churn from the
+    /// hook's `PreToolUse` cadence and writes nothing, but a resume (#176)
+    /// issues a new conversation id and must replace the stale one, or
+    /// every later resume reattaches the pre-resume conversation.
     pub async fn report_activity(
         &self,
         id: Uuid,
@@ -1293,16 +1332,16 @@ impl SessionManager {
                 state,
                 self.inner.report_ttl,
             ));
-            let newly_identified =
-                rec.session.agent_session_id.is_none() && agent_session_id.is_some();
-            if newly_identified {
+            let id_changed =
+                agent_session_id.is_some() && rec.session.agent_session_id != agent_session_id;
+            if id_changed {
                 rec.session.agent_session_id = agent_session_id;
                 self.inner
                     .sidecar
                     .write(&rec.session)
                     .map_err(SessionError::Sidecar)?;
             }
-            (changed || newly_identified).then(|| rec.session.clone())
+            (changed || id_changed).then(|| rec.session.clone())
         };
         if let Some(session) = updated {
             let _ = self

crates/maestro-daemon/tests/activity_report.rs

@@ -187,10 +187,30 @@ async fn a_reported_session_id_is_recorded_on_the_session() {
     assert_eq!(s.agent_session_id.as_deref(), Some("conv-abc123"));
 }
 
-/// #161: a session id does not change mid-session, so once set it is never
-/// overwritten by a later report — even one that names a different id.
+/// #161: repeated reports of the same id are pure churn from the hook's
+/// `PreToolUse` cadence — the sidecar must not be rewritten for them.
 #[tokio::test]
-async fn a_reported_session_id_is_written_once_and_never_replaced() {
+async fn a_repeated_report_of_the_same_session_id_writes_nothing() {
+    let fx = fixture(Duration::from_secs(300)).await;
+    let id = spawn_sleeping(&fx.manager).await;
+
+    fx.manager
+        .report_activity(id, ActivityState::Active, None, Some("conv-first".into()))
+        .await
+        .unwrap();
+    fx.manager
+        .report_activity(id, ActivityState::Active, None, Some("conv-first".into()))
+        .await
+        .unwrap();
+
+    let s = observed(&fx.manager, id).await;
+    assert_eq!(s.agent_session_id.as_deref(), Some("conv-first"));
+}
+
+/// #176: a resume issues a new conversation id. The stored id must move to
+/// it, or every later resume reattaches the stale pre-resume conversation.
+#[tokio::test]
+async fn a_reported_session_id_is_replaced_when_it_differs() {
     let fx = fixture(Duration::from_secs(300)).await;
     let id = spawn_sleeping(&fx.manager).await;
 
@@ -204,7 +224,14 @@ async fn a_reported_session_id_is_written_once_and_never_replaced() {
         .unwrap();
 
     let s = observed(&fx.manager, id).await;
-    assert_eq!(s.agent_session_id.as_deref(), Some("conv-first"));
+    assert_eq!(s.agent_session_id.as_deref(), Some("conv-second"));
+
+    let stored = fx
+        .sidecar
+        .read(id)
+        .expect("sidecar read")
+        .expect("sidecar present");
+    assert_eq!(stored.agent_session_id.as_deref(), Some("conv-second"));
 }
 
 #[tokio::test]

crates/maestro-daemon/tests/resume.rs

@@ -357,3 +357,237 @@ async fn post_resume_on_an_unknown_session_is_404() {
         .unwrap();
     assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND);
 }
+
+// ---------------------------------------------------------------------------
+// Races (#175)
+// ---------------------------------------------------------------------------
+
+/// Real tmux, with the *first* `new_session_piped` held inside the call until
+/// the test releases it. Later calls pass straight through, so a second resume
+/// racing the first never blocks on a gate the first one owns.
+///
+/// This is what makes the two races below deterministic rather than timing
+/// dependent: the resume under test is parked in exactly the window the bug
+/// lives in, and the racing operation runs to completion before it moves.
+#[derive(Debug)]
+struct GatedSpawn {
+    inner: maestro_daemon::tmux::Tmux,
+    entered: tokio::sync::mpsc::UnboundedSender<()>,
+    release: std::sync::Mutex<std::sync::mpsc::Receiver<()>>,
+    gated: std::sync::atomic::AtomicBool,
+}
+
+/// The test's end of a [`GatedSpawn`].
+struct Gate {
+    entered: tokio::sync::mpsc::UnboundedReceiver<()>,
+    release: std::sync::mpsc::SyncSender<()>,
+}
+
+impl Gate {
+    /// Wait until the gated spawn is parked mid-resume.
+    async fn wait_entered(&mut self) {
+        tokio::time::timeout(Duration::from_secs(10), self.entered.recv())
+            .await
+            .expect("resume reached the tmux spawn")
+            .expect("gate alive");
+    }
+
+    fn release(&self) {
+        self.release.send(()).expect("gate receiver alive");
+    }
+}
+
+fn gated_panes(inner: maestro_daemon::tmux::Tmux) -> (Arc<GatedSpawn>, Gate) {
+    let (entered_tx, entered_rx) = tokio::sync::mpsc::unbounded_channel();
+    let (release_tx, release_rx) = std::sync::mpsc::sync_channel(1);
+    (
+        Arc::new(GatedSpawn {
+            inner,
+            entered: entered_tx,
+            release: std::sync::Mutex::new(release_rx),
+            gated: std::sync::atomic::AtomicBool::new(true),
+        }),
+        Gate {
+            entered: entered_rx,
+            release: release_tx,
+        },
+    )
+}
+
+impl maestro_daemon::panes::Panes for GatedSpawn {
+    fn new_session_piped(
+        &self,
+        id: Uuid,
+        cwd: &std::path::Path,
+        command: &str,
+        session_env: &std::collections::BTreeMap<String, String>,
+        sink_command: &str,
+    ) -> Result<(), maestro_daemon::tmux::TmuxError> {
+        if self.gated.swap(false, std::sync::atomic::Ordering::SeqCst) {
+            let _ = self.entered.send(());
+            let _ = self.release.lock().expect("gate mutex").recv();
+        }
+        self.inner
+            .new_session_piped(id, cwd, command, session_env, sink_command)
+    }
+
+    fn kill_session(&self, id: Uuid) -> Result<(), maestro_daemon::tmux::TmuxError> {
+        self.inner.kill_session(id)
+    }
+
+    fn list_sessions(&self) -> Result<Vec<Uuid>, maestro_daemon::tmux::TmuxError> {
+        self.inner.list_sessions()
+    }
+
+    fn pane_info(
+        &self,
+        id: Uuid,
+    ) -> Result<Option<maestro_daemon::tmux::PaneInfo>, maestro_daemon::tmux::TmuxError> {
+        self.inner.pane_info(id)
+    }
+
+    fn send_keys(&self, id: Uuid, bytes: &[u8]) -> Result<(), maestro_daemon::tmux::TmuxError> {
+        self.inner.send_keys(id, bytes)
+    }
+
+    fn pipe_pane(&self, id: Uuid, sink: &str) -> Result<(), maestro_daemon::tmux::TmuxError> {
+        self.inner.pipe_pane(id, sink)
+    }
+
+    fn resize_window(
+        &self,
+        id: Uuid,
+        cols: u16,
+        rows: u16,
+    ) -> Result<(), maestro_daemon::tmux::TmuxError> {
+        self.inner.resize_window(id, cols, rows)
+    }
+
+    fn capture_pane(&self, id: Uuid) -> Result<Option<Vec<u8>>, maestro_daemon::tmux::TmuxError> {
+        self.inner.capture_pane(id)
+    }
+}
+
+/// A dormant session whose manager's tmux spawn can be parked by the test.
+async fn dormant_gated(fx: &DaemonFixture, cwd: &TempDir) -> (SessionManager, Session, Gate) {
+    let session = fx
+        .manager
+        .create(CreateSessionRequest {
+            cwd: cwd.path().to_path_buf(),
+            command: "sleep 30".into(),
+            env: HashMap::new(),
+            metadata: HashMap::new(),
+            worktree: None,
+        })
+        .await
+        .expect("create");
+    fx.tmux.tmux.kill_server().expect("kill tmux server");
+
+    let (panes, gate) = gated_panes(fx.tmux.tmux.clone());
+    let manager = SessionManager::new(
+        panes,
+        maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir()),
+        maestro_daemon::worktree::WorktreeManager::new(None),
+        fx.fifo_dir(),
+    );
+    manager.recover().await.expect("recover");
+    assert_eq!(
+        manager.status(session.id).await,
+        Some(SessionStatus::Dormant)
+    );
+    (manager, session, gate)
+}
+
+/// A kill that lands while a resume is inside the tmux spawn must not leave
+/// the sidecar behind. The kill deletes it; the resume writes it back a moment
+/// later and then discovers its own record is gone. The sidecar it resurrected
+/// would otherwise make the next recovery register a ghost dormant session for
+/// a session that was killed.
+#[tokio::test]
+async fn a_kill_during_resume_leaves_no_sidecar_behind() {
+    let fx = DaemonFixture::new();
+    let cwd = TempDir::new().unwrap();
+    let (manager, session, mut gate) = dormant_gated(&fx, &cwd).await;
+    let id = session.id;
+
+    let resuming = tokio::spawn({
+        let manager = manager.clone();
+        async move { manager.resume(id, &[]).await }
+    });
+    gate.wait_entered().await;
+
+    manager.kill(id, false).await.expect("kill mid-resume");
+    gate.release();
+
+    let err = resuming.await.unwrap().expect_err("record is gone");
+    assert!(matches!(err, SessionError::NotFound(got) if got == id));
+
+    assert!(
+        maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir())
+            .read(id)
+            .expect("read sidecar")
+            .is_none(),
+        "the killed session's sidecar must not be resurrected by the losing resume"
+    );
+
+    // And so the next recovery registers no ghost.
+    let after = revive(&fx);
+    after.recover().await.expect("recover");
+    assert!(
+        after.snapshot().await.is_empty(),
+        "a killed session must not come back dormant"
+    );
+}
+
+/// Two resumes of the same session must not both proceed. The second is
+/// refused on the state — `already_running` — not on its tmux spawn colliding
+/// with the first one's.
+#[tokio::test]
+async fn a_second_resume_is_refused_while_the_first_is_in_flight() {
+    let fx = DaemonFixture::new();
+    let cwd = TempDir::new().unwrap();
+    let (manager, session, mut gate) = dormant_gated(&fx, &cwd).await;
+    let id = session.id;
+
+    let resuming = tokio::spawn({
+        let manager = manager.clone();
+        async move { manager.resume(id, &[]).await }
+    });
+    gate.wait_entered().await;
+
+    let err = manager
+        .resume(id, &[])
+        .await
+        .expect_err("a resume is already in flight");
+    assert!(matches!(err, SessionError::AlreadyRunning(got) if got == id));
+    assert_eq!(err.to_error_info().code, "already_running");
+
+    gate.release();
+    let first = resuming.await.unwrap().expect("the first resume wins");
+    assert_eq!(first.status, SessionStatus::Running);
+    assert_eq!(
+        fx.tmux.tmux.list_sessions().expect("list"),
+        vec![id],
+        "exactly one pane for the session"
+    );
+}
+
+/// A refused resume must not leave the session marked in-flight: once the
+/// reason for the refusal is gone, resuming works.
+#[tokio::test]
+async fn a_refused_resume_can_be_retried() {
+    let fx = DaemonFixture::new();
+    let cwd = TempDir::new().unwrap();
+    let (manager, session) = dormant_session(&fx, &cwd, "sleep 30", "retried").await;
+
+    std::fs::remove_dir_all(cwd.path()).unwrap();
+    let err = manager
+        .resume(session.id, &[])
+        .await
+        .expect_err("cwd is gone");
+    assert_eq!(err.to_error_info().code, "cwd_missing");
+
+    std::fs::create_dir_all(cwd.path()).unwrap();
+    let resumed = manager.resume(session.id, &[]).await.expect("resume");
+    assert_eq!(resumed.status, SessionStatus::Running);
+}

docs/RESTART.md

@@ -98,7 +98,13 @@ conversation id is respawned on the command it was created with. `Session::comma
 always keeps that original command, so a second resume resolves the same preset.
 Resume is refused with `409` for a session that is already running
 (`already_running`) and for one whose `cwd` has disappeared (`cwd_missing`) —
-neither is something a respawn could fix. The other thing to do with a dormant
+neither is something a respawn could fix. A session with a resume already in
+flight is refused as `already_running` too (#175): the respawn happens with
+the session map unlocked, so the record is marked in-flight before it starts
+and the second request is answered on that state rather than on its tmux spawn
+colliding with the first one's. A resume that loses a race with a kill takes
+its own sidecar back off disk, so the killed session does not return as a
+dormant ghost on the next recovery. The other thing to do with a dormant
 session is dismiss it, which removes the sidecar so the next recovery does not
 bring it back.