fix(maestro): key activity watermark by stream

crates/maestro-acp/tests/fake_acp_agent.py

@@ -19,6 +19,8 @@ Environment knobs, all optional:
   FAKE_ACP_INIT_FAIL=1    answer `initialize` with a JSON-RPC error
   FAKE_ACP_DELAY_SESSION_NEW_MS=n  wait n milliseconds before answering
                              `session/new`
+  FAKE_ACP_SESSION_NEW_GATE=path  wait for `path` to exist before answering
+                             `session/new`
   FAKE_ACP_EXIT_AFTER_NEW=1  exit 0 right after answering `session/new`
   FAKE_ACP_EXIT_ON_CANCEL=1  exit 0 on `session/cancel` (default: keep running
                              so the SIGTERM path is exercised)
@@ -133,6 +135,9 @@ for raw in sys.stdin:
             "authMethods": [],
         }})
     elif method == "session/new":
+        gate = os.environ.get("FAKE_ACP_SESSION_NEW_GATE")
+        while gate and not os.path.exists(gate):
+            time.sleep(0.01)
         time.sleep(int(os.environ.get("FAKE_ACP_DELAY_SESSION_NEW_MS", "0")) / 1000)
         send({"jsonrpc": "2.0", "id": rid, "result": {"sessionId": session_id}})
         send(chunk("hello from fake"))

crates/maestro-daemon/src/session.rs

@@ -6,9 +6,12 @@
 //! handler interact with tmux.
 
 use std::collections::{HashMap, HashSet};
+use std::future::Future;
 use std::path::PathBuf;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
-use std::sync::Arc;
+use std::pin::Pin;
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
+use std::sync::{Arc, Weak};
+use std::task::{Context, Poll};
 use std::time::Duration;
 
 use chrono::Utc;
@@ -18,7 +21,7 @@ use maestro_protocol::{
     SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, WorktreeSpec, NAME_KEY,
     NAME_SOURCE_KEY,
 };
-use tokio::sync::{broadcast, oneshot, watch, Mutex};
+use tokio::sync::{broadcast, oneshot, watch, Mutex, Notify};
 use uuid::Uuid;
 
 use crate::panes::Panes;
@@ -99,8 +102,12 @@ pub struct SessionRecord {
     /// outside the sessions lock. Daemon-local; never rides the wire.
     pub acp: Option<Arc<AcpSession>>,
     /// Open log plus its post-append live sender. Their shared lock is the
-    /// replay/live seam used by the structured WebSocket in #138.
+    /// replay/live seam used by the structured WebSocket in #138. Callers may
+    /// clone this handle under `sessions`, but must never hold either guard
+    /// while acquiring the other.
     acp_events: Option<Arc<Mutex<AcpEventStream>>>,
+    /// Latest activity snapshot applied, keyed to its originating stream.
+    structured_activity_watermark: Option<StructuredActivityWatermark>,
     /// Completion signal for the task draining ACP notifications into the
     /// event log. Shutdown waits for it before returning.
     acp_event_pump: Option<watch::Receiver<bool>>,
@@ -115,6 +122,12 @@ impl SessionRecord {
     }
 }
 
+#[derive(Debug)]
+struct StructuredActivityWatermark {
+    stream: Weak<Mutex<AcpEventStream>>,
+    sequence: u64,
+}
+
 #[derive(Debug)]
 struct AcpEventStream {
     log: EventLog,
@@ -126,6 +139,14 @@ struct AcpEventStream {
     /// claimed: it is queued onto ACP before another operation can proceed.
     operation: Arc<Mutex<()>>,
     dispatch: AcpDispatchState,
+    activity_sequence: u64,
+}
+
+impl AcpEventStream {
+    fn activity_snapshot(&mut self) -> (u64, ActivityState) {
+        self.activity_sequence += 1;
+        (self.activity_sequence, self.projection.activity())
+    }
 }
 
 #[derive(Debug, Default)]
@@ -263,9 +284,99 @@ struct Inner {
     /// Once daemon shutdown starts, it owns structured-session finalization.
     /// Background passes must not classify those deliberate exits as natural.
     shutting_down: AtomicBool,
-    /// Serializes structured creation with shutdown so every adapter child is
-    /// either rejected or included in shutdown's final drain.
-    structured_lifecycle: Mutex<()>,
+    /// Admission is serialized with shutdown, while admitted handshakes run
+    /// independently and are counted until registration or rollback finishes.
+    structured_lifecycle: StructuredLifecycle,
+}
+
+#[derive(Debug, Default)]
+struct StructuredLifecycle {
+    gate: std::sync::Mutex<()>,
+    in_flight: AtomicUsize,
+    settled: Notify,
+}
+
+impl StructuredLifecycle {
+    fn admit<'a>(
+        &'a self,
+        shutting_down: &AtomicBool,
+    ) -> Result<StructuredCreate<'a>, SessionError> {
+        let _gate = self.gate.lock().expect("structured lifecycle gate");
+        if shutting_down.load(Ordering::Acquire) {
+            return Err(SessionError::ShuttingDown);
+        }
+        self.in_flight.fetch_add(1, Ordering::AcqRel);
+        Ok(StructuredCreate { lifecycle: self })
+    }
+
+    async fn wait_until_settled(&self) {
+        loop {
+            let settled = self.settled.notified();
+            if self.in_flight.load(Ordering::Acquire) == 0 {
+                return;
+            }
+            settled.await;
+        }
+    }
+}
+
+struct StructuredCreate<'a> {
+    lifecycle: &'a StructuredLifecycle,
+}
+
+impl Drop for StructuredCreate<'_> {
+    fn drop(&mut self) {
+        if self.lifecycle.in_flight.fetch_sub(1, Ordering::AcqRel) == 1 {
+            self.lifecycle.settled.notify_one();
+        }
+    }
+}
+
+struct AdmittedStructuredCreate<'a> {
+    create: Pin<Box<dyn Future<Output = Result<Session, SessionError>> + Send + 'a>>,
+    /// Field order matters: dropping a cancelled create rolls back its inner
+    /// future before this permit releases shutdown's in-flight barrier.
+    _permit: StructuredCreate<'a>,
+}
+
+impl Future for AdmittedStructuredCreate<'_> {
+    type Output = Result<Session, SessionError>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.create.as_mut().poll(cx)
+    }
+}
+
+struct PendingCreate<'a> {
+    inner: &'a Inner,
+    id: Uuid,
+    fifo_path: PathBuf,
+    worktree_path: Option<PathBuf>,
+    worktree_base_repo: Option<PathBuf>,
+    committed: bool,
+}
+
+impl PendingCreate<'_> {
+    fn commit(&mut self) {
+        self.committed = true;
+    }
+}
+
+impl Drop for PendingCreate<'_> {
+    fn drop(&mut self) {
+        if self.committed {
+            return;
+        }
+        let _ = std::fs::remove_file(&self.fifo_path);
+        if let Err(error) = self.inner.sidecar.delete(self.id) {
+            tracing::warn!(%error, id = %self.id, "rollback: failed to delete create artifacts");
+        }
+        if let (Some(base), Some(path)) = (&self.worktree_base_repo, &self.worktree_path) {
+            if let Err(error) = self.inner.worktree_mgr.remove(base, path) {
+                tracing::warn!(%error, ?path, "rollback: failed to remove worktree after partial create");
+            }
+        }
+    }
 }
 
 impl SessionManager {
@@ -309,7 +420,7 @@ impl SessionManager {
                 viewports: Mutex::new(Viewports::new()),
                 next_viewport: AtomicU64::new(0),
                 shutting_down: AtomicBool::new(false),
-                structured_lifecycle: Mutex::new(()),
+                structured_lifecycle: StructuredLifecycle::default(),
             }),
         }
     }
@@ -827,10 +938,18 @@ impl SessionManager {
     /// is on disk after the restart is truthful. tmux sessions are left
     /// alone — surviving a restart is what they are for.
     pub async fn shutdown(&self) {
-        if self.inner.shutting_down.swap(true, Ordering::AcqRel) {
-            return;
+        {
+            let _gate = self
+                .inner
+                .structured_lifecycle
+                .gate
+                .lock()
+                .expect("structured lifecycle gate");
+            if self.inner.shutting_down.swap(true, Ordering::AcqRel) {
+                return;
+            }
         }
-        let _lifecycle = self.inner.structured_lifecycle.lock().await;
+        self.inner.structured_lifecycle.wait_until_settled().await;
         let children: Vec<(Uuid, Arc<AcpSession>, Option<watch::Receiver<bool>>)> = {
             let sessions = self.inner.sessions.lock().await;
             sessions
@@ -980,11 +1099,15 @@ impl SessionManager {
     /// Spawn a new session. On success returns the freshly-created `Session`.
     pub async fn create(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
         if req.view == SessionView::Structured {
-            let _lifecycle = self.inner.structured_lifecycle.lock().await;
-            if self.inner.shutting_down.load(Ordering::Acquire) {
-                return Err(SessionError::ShuttingDown);
+            let permit = self
+                .inner
+                .structured_lifecycle
+                .admit(&self.inner.shutting_down)?;
+            return AdmittedStructuredCreate {
+                create: Box::pin(self.create_inner(req)),
+                _permit: permit,
             }
-            return self.create_inner(req).await;
+            .await;
         }
         self.create_inner(req).await
     }
@@ -1022,22 +1145,17 @@ impl SessionManager {
         // Each side effect below unwinds the ones before it on failure, so a
         // partial create never leaks a worktree or — worse — a ghost tmux
         // session with no in-memory record (NAP-260).
-        let bridge =
-            match PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await {
-                Ok(b) => b,
-                Err(e) => {
-                    self.rollback_worktree(&worktree_base_repo, &worktree_path);
-                    return Err(e.into());
-                }
-            };
-        let mut session_env = match sanitize_env(&req.env) {
-            Ok(env) => env,
-            Err(e) => {
-                let _ = std::fs::remove_file(&fifo_path);
-                self.rollback_worktree(&worktree_base_repo, &worktree_path);
-                return Err(e);
-            }
+        let mut pending = PendingCreate {
+            inner: &self.inner,
+            id,
+            fifo_path: fifo_path.clone(),
+            worktree_path: worktree_path.clone(),
+            worktree_base_repo: worktree_base_repo.clone(),
+            committed: false,
         };
+        let bridge =
+            PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await?;
+        let mut session_env = sanitize_env(&req.env)?;
         // Identity + reporting target for the in-session hook (#106). Ours
         // win over whatever the request carried, so a client cannot spoof
         // another session's identity (#103).
@@ -1084,10 +1202,6 @@ impl SessionManager {
                         .await
                 };
                 if let Err(e) = spawned {
-                    // `bridge` drops at the end of this block, closing the
-                    // FIFO reader.
-                    let _ = std::fs::remove_file(&fifo_path);
-                    self.rollback_worktree(&worktree_base_repo, &worktree_path);
                     return Err(SessionError::Tmux(e));
                 }
                 (None, None)
@@ -1105,8 +1219,6 @@ impl SessionManager {
                     Err(error) => {
                         acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
                             .await;
-                        let _ = std::fs::remove_file(&fifo_path);
-                        self.rollback_worktree(&worktree_base_repo, &worktree_path);
                         return Err(SessionError::Sidecar(error));
                     }
                 };
@@ -1126,13 +1238,12 @@ impl SessionManager {
                         pending_approvals: HashMap::new(),
                         operation: Arc::new(Mutex::new(())),
                         dispatch: AcpDispatchState::default(),
+                        activity_sequence: 0,
                     })))
                 }
                 Err(e) => {
                     acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
                         .await;
-                    let _ = std::fs::remove_file(&fifo_path);
-                    self.rollback_worktree(&worktree_base_repo, &worktree_path);
                     return Err(SessionError::Sidecar(e));
                 }
             }
@@ -1201,9 +1312,6 @@ impl SessionManager {
                     let _ = tmux_blocking(move || tmux.kill_session(id)).await;
                 }
             }
-            let _ = std::fs::remove_file(&fifo_path);
-            let _ = self.inner.sidecar.delete(id);
-            self.rollback_worktree(&worktree_base_repo, &worktree_path);
             return Err(SessionError::Sidecar(e));
         }
 
@@ -1213,6 +1321,11 @@ impl SessionManager {
             .map(|(acp, events)| (acp.clone(), events.clone()));
         {
             let mut sessions = self.inner.sessions.lock().await;
+            let event_pump = event_pump.map(|(acp, events)| {
+                let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
+                start_acp_permission_pump(&self.inner, id, &acp, events);
+                done
+            });
             sessions.insert(
                 id,
                 SessionRecord {
@@ -1229,17 +1342,11 @@ impl SessionManager {
                     awaiting_suppressed: false,
                     acp,
                     acp_events,
-                    acp_event_pump: None,
+                    structured_activity_watermark: None,
+                    acp_event_pump: event_pump,
                 },
             );
         }
-        if let Some((acp, events)) = event_pump {
-            let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
-            start_acp_permission_pump(&self.inner, id, &acp, events);
-            if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
-                record.acp_event_pump = Some(done);
-            }
-        }
         let _ = self.inner.events.send(CatalogEvent::SessionAdded {
             session: session.clone(),
         });
@@ -1264,19 +1371,10 @@ impl SessionManager {
             }
         }
 
+        pending.commit();
         Ok(session)
     }
 
-    /// 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>) {
-        if let (Some(base), Some(path)) = (base, path) {
-            if let Err(e) = self.inner.worktree_mgr.remove(base, path) {
-                tracing::warn!(error = %e, ?path, "rollback: failed to remove worktree after partial create");
-            }
-        }
-    }
-
     /// Kill a session and remove its on-disk artifacts.
     ///
     /// Refused (#79) when the session's worktree holds unsaved work —
@@ -1456,6 +1554,7 @@ impl SessionManager {
                     // daemon; only tmux-backed sessions can be adopted.
                     acp: None,
                     acp_events: None,
+                    structured_activity_watermark: None,
                     acp_event_pump: None,
                 },
             );
@@ -1577,6 +1676,7 @@ impl SessionManager {
                             pending_approvals: HashMap::new(),
                             operation: Arc::new(Mutex::new(())),
                             dispatch: AcpDispatchState::default(),
+                            activity_sequence: 0,
                         })))
                     }
                     Err(error) => {
@@ -1624,6 +1724,7 @@ impl SessionManager {
                     awaiting_suppressed: false,
                     acp,
                     acp_events,
+                    structured_activity_watermark: None,
                     acp_event_pump: None,
                 },
             );
@@ -2044,57 +2145,59 @@ async fn sync_structured_activity_from_stream(
     id: Uuid,
     events: Arc<Mutex<AcpEventStream>>,
 ) {
-    let event_stream = events.lock().await;
-    let raw = event_stream.projection.activity();
-    // Stream state must stay fixed until its catalog transition is applied;
-    // every other path releases `sessions` before awaiting this stream lock.
-    apply_structured_activity(inner, id, Some(&events), raw).await;
+    let (sequence, raw) = events.lock().await.activity_snapshot();
+    apply_structured_activity(inner, id, Some((&events, sequence)), raw).await;
 }
 
 async fn apply_structured_activity(
     inner: &Arc<Inner>,
     id: Uuid,
-    expected_events: Option<&Arc<Mutex<AcpEventStream>>>,
+    expected_events: Option<(&Arc<Mutex<AcpEventStream>>, u64)>,
     raw: ActivityState,
 ) {
-    let updated = {
-        let mut sessions = inner.sessions.lock().await;
-        let Some(record) = sessions.get_mut(&id) else {
-            return;
-        };
-        if record.session.view != SessionView::Structured {
-            return;
-        }
-        let current_events = record.acp_events.as_ref();
-        let stream_matches = match (expected_events, current_events) {
-            (Some(expected), Some(current)) => Arc::ptr_eq(expected, current),
-            (None, None) => true,
-            _ => false,
-        };
-        if !stream_matches {
-            return;
-        }
-        let raw = if record.session.status == SessionStatus::Dead {
-            ActivityState::Error
-        } else {
-            raw
-        };
-        let activity = crate::activity::apply_focus_suppression(
-            raw,
-            record.focused_since,
-            std::time::Instant::now(),
-            &mut record.awaiting_suppressed,
-        );
-        if record.session.activity == activity {
-            None
-        } else {
-            record.session.activity = activity;
-            record.session.activity_since = Utc::now();
-            Some(record.session.clone())
-        }
+    let mut sessions = inner.sessions.lock().await;
+    let Some(record) = sessions.get_mut(&id) else {
+        return;
     };
-    if let Some(session) = updated {
-        let _ = inner.events.send(CatalogEvent::SessionUpdated { session });
+    if record.session.view != SessionView::Structured {
+        return;
+    }
+    let current_events = record.acp_events.as_ref();
+    match (expected_events, current_events) {
+        (Some((expected, sequence)), Some(current)) if Arc::ptr_eq(expected, current) => {
+            let stream = Arc::downgrade(current);
+            if record
+                .structured_activity_watermark
+                .as_ref()
+                .is_some_and(|watermark| {
+                    watermark.stream.ptr_eq(&stream) && sequence < watermark.sequence
+                })
+            {
+                return;
+            }
+            record.structured_activity_watermark =
+                Some(StructuredActivityWatermark { stream, sequence });
+        }
+        (None, None) => {}
+        _ => return,
+    }
+    let raw = if record.session.status == SessionStatus::Dead {
+        ActivityState::Error
+    } else {
+        raw
+    };
+    let activity = crate::activity::apply_focus_suppression(
+        raw,
+        record.focused_since,
+        std::time::Instant::now(),
+        &mut record.awaiting_suppressed,
+    );
+    if record.session.activity != activity {
+        record.session.activity = activity;
+        record.session.activity_since = Utc::now();
+        let _ = inner.events.send(CatalogEvent::SessionUpdated {
+            session: record.session.clone(),
+        });
     }
 }
 
@@ -2771,16 +2874,17 @@ mod tests {
         PermissionOption, PermissionOptionKind, RequestPermissionOutcome, RequestPermissionRequest,
         ToolCallUpdate, ToolCallUpdateFields,
     };
-    use maestro_protocol::{ActivityState, CreateSessionRequest, SessionView};
+    use maestro_protocol::{ActivityState, CatalogEvent, CreateSessionRequest, SessionView};
 
     use super::{
-        cancel_pending_approvals, permission_outcome, record_permission_request,
-        run_structured_activity_pass, sync_structured_activity_from_stream, AcpDispatchState,
-        AcpEventStream, ApprovalDecision, RecordOutcome, SessionManager,
+        apply_structured_activity, cancel_pending_approvals, permission_outcome,
+        record_permission_request, run_structured_activity_pass,
+        sync_structured_activity_from_stream, AcpDispatchState, AcpEventStream, ApprovalDecision,
+        RecordOutcome, SessionManager,
     };
     use crate::activity::FOCUS_CLEAR_HOLD;
     use crate::panes::FakePanes;
-    use crate::projection::Projection;
+    use crate::projection::{Projection, TurnLifecycle};
     use crate::sidecar::{
         EventCursor, EventLog, FaultyLogSink, PersistedEvent, SharedBuf, SidecarStore,
     };
@@ -2809,6 +2913,7 @@ mod tests {
             pending_approvals: std::collections::HashMap::new(),
             operation: std::sync::Arc::new(tokio::sync::Mutex::new(())),
             dispatch: AcpDispatchState::default(),
+            activity_sequence: 0,
         }
     }
 
@@ -2965,6 +3070,87 @@ mod tests {
             .expect("terminal resize remains available");
     }
 
+    #[tokio::test]
+    async fn structured_activity_releases_stream_before_waiting_for_catalog() {
+        let (_state, manager, _panes, structured_id, events, _terminal_id) =
+            manager_with_structured_stream().await;
+        let sessions = manager.inner.sessions.lock().await;
+        let sync =
+            sync_structured_activity_from_stream(&manager.inner, structured_id, events.clone());
+        tokio::pin!(sync);
+        assert!(matches!(futures_util::poll!(sync.as_mut()), Poll::Pending));
+
+        assert!(
+            events.try_lock().is_ok(),
+            "activity sync held the event stream while waiting for the catalog"
+        );
+
+        drop(sessions);
+        tokio::time::timeout(Duration::from_secs(1), sync)
+            .await
+            .expect("activity sync did not finish");
+    }
+
+    #[tokio::test]
+    async fn newer_structured_activity_snapshot_cannot_be_overtaken() {
+        let (_state, manager, _panes, structured_id, events, _terminal_id) =
+            manager_with_structured_stream().await;
+        let mut catalog = manager.events();
+        manager
+            .inner
+            .sessions
+            .lock()
+            .await
+            .get_mut(&structured_id)
+            .unwrap()
+            .focused_since = None;
+
+        let (older_sequence, older_activity) = events.lock().await.activity_snapshot();
+        let (newer_sequence, newer_activity) = {
+            let mut event_stream = events.lock().await;
+            let turn = event_stream.projection.active_turn().unwrap();
+            event_stream.projection.clear_approval();
+            event_stream
+                .projection
+                .turn_lifecycle(TurnLifecycle::Finish, turn);
+            event_stream.activity_snapshot()
+        };
+
+        apply_structured_activity(
+            &manager.inner,
+            structured_id,
+            Some((&events, newer_sequence)),
+            newer_activity,
+        )
+        .await;
+        apply_structured_activity(
+            &manager.inner,
+            structured_id,
+            Some((&events, older_sequence)),
+            older_activity,
+        )
+        .await;
+
+        let CatalogEvent::SessionUpdated { session } = catalog.try_recv().unwrap() else {
+            panic!("unexpected catalog event");
+        };
+        assert_eq!(session.activity, ActivityState::Idle);
+        assert!(matches!(
+            catalog.try_recv(),
+            Err(tokio::sync::broadcast::error::TryRecvError::Empty)
+        ));
+        assert_eq!(
+            manager
+                .snapshot()
+                .await
+                .into_iter()
+                .find(|session| session.id == structured_id)
+                .unwrap()
+                .activity,
+            ActivityState::Idle
+        );
+    }
+
     #[tokio::test]
     async fn delayed_activity_from_a_replaced_stream_is_discarded() {
         let (_state, manager, _panes, structured_id, old_events, _terminal_id) =
@@ -2998,6 +3184,73 @@ mod tests {
         ));
     }
 
+    #[tokio::test]
+    async fn replacement_accepts_new_activity_after_a_high_old_watermark() {
+        let (_state, manager, _panes, structured_id, old_events, _terminal_id) =
+            manager_with_structured_stream().await;
+        let replacement = Arc::new(tokio::sync::Mutex::new(stream(
+            manager
+                .inner
+                .sidecar
+                .open_events(uuid::Uuid::new_v4())
+                .expect("open replacement event log"),
+        )));
+        manager
+            .inner
+            .sessions
+            .lock()
+            .await
+            .get_mut(&structured_id)
+            .unwrap()
+            .focused_since = None;
+        let _ = old_events.lock().await.activity_snapshot();
+        let (old_high_sequence, old_high_activity) = old_events.lock().await.activity_snapshot();
+        apply_structured_activity(
+            &manager.inner,
+            structured_id,
+            Some((&old_events, old_high_sequence)),
+            old_high_activity,
+        )
+        .await;
+        let (delayed_old_sequence, delayed_old_activity) =
+            old_events.lock().await.activity_snapshot();
+
+        {
+            let mut sessions = manager.inner.sessions.lock().await;
+            let record = sessions.get_mut(&structured_id).unwrap();
+            record.acp_events = Some(replacement.clone());
+            record.session.activity = ActivityState::Error;
+        }
+
+        let (new_sequence, new_activity) = replacement.lock().await.activity_snapshot();
+        assert!(new_sequence < old_high_sequence);
+        apply_structured_activity(
+            &manager.inner,
+            structured_id,
+            Some((&replacement, new_sequence)),
+            new_activity,
+        )
+        .await;
+        apply_structured_activity(
+            &manager.inner,
+            structured_id,
+            Some((&old_events, delayed_old_sequence)),
+            delayed_old_activity,
+        )
+        .await;
+
+        let sessions = manager.inner.sessions.lock().await;
+        let record = sessions.get(&structured_id).unwrap();
+        assert_eq!(record.session.activity, ActivityState::Idle);
+        let watermark = record.structured_activity_watermark.as_ref().unwrap();
+        assert_eq!(watermark.sequence, new_sequence);
+        assert!(watermark.stream.ptr_eq(&Arc::downgrade(&replacement)));
+        assert!(Arc::ptr_eq(
+            record.acp_events.as_ref().unwrap(),
+            &replacement
+        ));
+    }
+
     /// A handle that can never append *or* roll back: opened read-only, so
     /// both `write_all` and the rollback `truncate` fail with the OS's own
     /// error (`ftruncate` requires a writable fd too) instead of a

crates/maestro-daemon/tests/acp.rs

@@ -950,6 +950,191 @@ async fn shutdown_waits_for_structured_creation_and_rejects_new_creates() {
     assert_eq!(fake.log().matches("signal SIGTERM").count(), 1);
 }
 
+#[tokio::test]
+async fn structured_handshakes_run_independently() {
+    let fx = DaemonFixture::new();
+    let blocked = Fake::new();
+    let ready = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let gate = cwd.path().join("release-first");
+    let mut blocked_env = blocked.env();
+    blocked_env.insert(
+        "FAKE_ACP_SESSION_NEW_GATE".into(),
+        gate.to_string_lossy().into_owned(),
+    );
+
+    let first_cwd = cwd.path().to_path_buf();
+    let manager = fx.manager.clone();
+    let first =
+        tokio::spawn(async move { manager.create(structured(&first_cwd, blocked_env)).await });
+    wait_for_async(Duration::from_secs(5), || async {
+        blocked.log().contains("recv session/new")
+    })
+    .await;
+
+    let second = tokio::time::timeout(
+        Duration::from_secs(5),
+        fx.manager.create(structured(cwd.path(), ready.env())),
+    )
+    .await
+    .expect("second handshake was blocked by the first")
+    .expect("second create");
+    assert_eq!(second.status, SessionStatus::Running);
+    assert!(!first.is_finished(), "the gate must still block the first");
+
+    std::fs::write(&gate, b"release").unwrap();
+    let first = first.await.expect("first task").expect("first create");
+    fx.manager.kill(first.id, false).await.expect("kill first");
+    fx.manager
+        .kill(second.id, false)
+        .await
+        .expect("kill second");
+}
+
+#[tokio::test]
+async fn shutdown_waits_for_every_admitted_structured_create() {
+    let fx = DaemonFixture::new();
+    let a = Fake::new();
+    let b = Fake::new();
+    let late = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let gate_a = cwd.path().join("release-a");
+    let gate_b = cwd.path().join("release-b");
+    let mut env_a = a.env();
+    let mut env_b = b.env();
+    env_a.insert(
+        "FAKE_ACP_SESSION_NEW_GATE".into(),
+        gate_a.to_string_lossy().into_owned(),
+    );
+    env_b.insert(
+        "FAKE_ACP_SESSION_NEW_GATE".into(),
+        gate_b.to_string_lossy().into_owned(),
+    );
+
+    let create_a = {
+        let manager = fx.manager.clone();
+        let cwd = cwd.path().to_path_buf();
+        tokio::spawn(async move { manager.create(structured(&cwd, env_a)).await })
+    };
+    let create_b = {
+        let manager = fx.manager.clone();
+        let cwd = cwd.path().to_path_buf();
+        tokio::spawn(async move { manager.create(structured(&cwd, env_b)).await })
+    };
+    wait_for_async(Duration::from_secs(5), || async {
+        a.log().contains("recv session/new") && b.log().contains("recv session/new")
+    })
+    .await;
+
+    let shutdown = {
+        let manager = fx.manager.clone();
+        tokio::spawn(async move { manager.shutdown().await })
+    };
+    wait_for_async(Duration::from_secs(5), || async {
+        matches!(
+            fx.manager.create(structured(cwd.path(), late.env())).await,
+            Err(maestro_daemon::SessionError::ShuttingDown)
+        )
+    })
+    .await;
+    assert!(!shutdown.is_finished());
+
+    std::fs::write(&gate_a, b"release").unwrap();
+    let session_a = create_a.await.expect("create a task").expect("create a");
+    assert!(
+        !shutdown.is_finished(),
+        "shutdown ignored the second create"
+    );
+
+    std::fs::write(&gate_b, b"release").unwrap();
+    let session_b = create_b.await.expect("create b task").expect("create b");
+    shutdown.await.expect("shutdown task");
+
+    wait_for_exit(a.pid()).await;
+    wait_for_exit(b.pid()).await;
+    assert_eq!(
+        fx.manager.status(session_a.id).await,
+        Some(SessionStatus::Dead)
+    );
+    assert_eq!(
+        fx.manager.status(session_b.id).await,
+        Some(SessionStatus::Dead)
+    );
+}
+
+#[tokio::test]
+async fn cancelling_a_structured_handshake_rolls_back_and_settles_admission() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let repo = TempDir::new().unwrap();
+    init_repo(repo.path());
+    let gate = repo.path().join("release-create");
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_ECHO_ENV".into(), "MAESTRO_SESSION_ID".into());
+    env.insert(
+        "FAKE_ACP_SESSION_NEW_GATE".into(),
+        gate.to_string_lossy().into_owned(),
+    );
+    let request = CreateSessionRequest {
+        view: SessionView::Structured,
+        cwd: repo.path().to_path_buf(),
+        command: "claude".into(),
+        env,
+        metadata: HashMap::new(),
+        worktree: Some(WorktreeSpec {
+            base_repo: repo.path().to_path_buf(),
+            branch: "cancelled-create".into(),
+            create_branch: true,
+        }),
+    };
+
+    let create = {
+        let manager = fx.manager.clone();
+        tokio::spawn(async move { manager.create(request).await })
+    };
+    wait_for_async(Duration::from_secs(5), || async {
+        fake.log().contains("recv session/new")
+    })
+    .await;
+    let id = fake
+        .log()
+        .lines()
+        .find_map(|line| line.strip_prefix("env MAESTRO_SESSION_ID="))
+        .and_then(|id| uuid::Uuid::parse_str(id).ok())
+        .expect("session id in adapter log");
+    let pid = fake.pid();
+
+    create.abort();
+    let shutdown = {
+        let manager = fx.manager.clone();
+        tokio::spawn(async move { manager.shutdown().await })
+    };
+    assert!(create.await.unwrap_err().is_cancelled());
+    tokio::time::timeout(Duration::from_secs(5), shutdown)
+        .await
+        .expect("cancelled admission kept shutdown waiting")
+        .expect("shutdown task");
+    wait_for_exit(pid).await;
+    wait_for_async(Duration::from_secs(5), || async {
+        !fx.fifo_dir().join(format!("{id}.fifo")).exists()
+            && !repo.path().join("cancelled-create").exists()
+    })
+    .await;
+    assert!(fx.manager.snapshot().await.is_empty());
+    let id = id.to_string();
+    assert!(
+        std::fs::read_dir(fx.sidecar_dir())
+            .into_iter()
+            .flatten()
+            .all(|entry| !entry
+                .unwrap()
+                .file_name()
+                .to_string_lossy()
+                .starts_with(&id)),
+        "cancelled create left a sidecar artifact"
+    );
+}
+
 #[tokio::test]
 async fn shutdown_drains_final_notifications_to_the_log() {
     let fx = DaemonFixture::new();

web/src/structured.test.ts

@@ -932,6 +932,298 @@ describe("structured connection: reconnection/error/loading", () => {
   });
 });
 
+describe("structured connection: transcript scrolling", () => {
+  let frames: Map<number, FrameRequestCallback>;
+  let nextFrame: number;
+  let resizeCallback: ResizeObserverCallback;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    vi.useFakeTimers();
+    frames = new Map();
+    nextFrame = 1;
+    vi.stubGlobal("requestAnimationFrame", (cb: FrameRequestCallback) => {
+      const id = nextFrame++;
+      frames.set(id, cb);
+      return id;
+    });
+    vi.stubGlobal("cancelAnimationFrame", (id: number) => {
+      frames.delete(id);
+    });
+    vi.stubGlobal("ResizeObserver", class {
+      constructor(callback: ResizeObserverCallback) {
+        resizeCallback = callback;
+      }
+      observe(): void {}
+      unobserve(): void {}
+      disconnect(): void {}
+    });
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.unstubAllGlobals();
+  });
+
+  function metrics(conn: ReturnType<typeof mount>, height: number, clientHeight: number, top = 0) {
+    const scroll = conn.el.querySelector<HTMLElement>(".structured-scroll")!;
+    const values = { height, clientHeight, top };
+    Object.defineProperties(scroll, {
+      scrollHeight: { configurable: true, get: () => values.height },
+      clientHeight: { configurable: true, get: () => values.clientHeight },
+      scrollTop: {
+        configurable: true,
+        get: () => values.top,
+        set: (value: number) => { values.top = value; },
+      },
+    });
+    return { scroll, values };
+  }
+
+  function runFrames(): void {
+    const pending = [...frames.entries()];
+    frames.clear();
+    for (const [, cb] of pending) cb(0);
+  }
+
+  function resizeBody(width: number): void {
+    resizeCallback(
+      [{ contentRect: { width } } as ResizeObserverEntry],
+      {} as ResizeObserver,
+    );
+  }
+
+  it("shows the bottom after the initial transcript snapshot is laid out", () => {
+    const ws = new FakeSocket();
+    const conn = mount({ sockets: [ws] });
+    const { values } = metrics(conn, 900, 300, 40);
+    ws.open();
+    ws.emit({
+      type: "transcript_snapshot",
+      snapshot: {
+        cursor: 1,
+        rows: [
+          row("agent:1", "agent_text", { content: { type: "text", text: "latest" } }),
+        ],
+      },
+    });
+
+    expect(values.top).toBe(40);
+    runFrames();
+    expect(values.top).toBe(600);
+    conn.dispose();
+  });
+
+  it("keeps pinned appends and patches at the bottom after layout", () => {
+    const ws = new FakeSocket();
+    const conn = mount({ sockets: [ws] });
+    const { values } = metrics(conn, 600, 200);
+    ws.open();
+    ws.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
+    runFrames();
+
+    values.height = 720;
+    ws.emit({
+      type: "projection_update",
+      transcript: {
+        type: "append",
+        row: row("agent:1", "agent_text", { content: { type: "text", text: "partial" } }),
+      },
+      state: idleState,
+    });
+    runFrames();
+    expect(values.top).toBe(520);
+
+    values.height = 780;
+    ws.emit({
+      type: "projection_update",
+      transcript: {
+        type: "patch",
+        row: row("agent:1", "agent_text", {
+          content: { type: "text", text: "a longer completion" },
+        }),
+      },
+      state: idleState,
+    });
+    runFrames();
+    expect(values.top).toBe(580);
+    conn.dispose();
+  });
+
+  it.each([
+    { position: "pinned", top: 800, expected: 880, manuallyScrolled: false },
+    { position: "manually scrolled", top: 180, expected: 180, manuallyScrolled: true },
+  ])("preserves the $position position when the composer grows", ({ top, expected, manuallyScrolled }) => {
+    const ws = new FakeSocket();
+    const conn = mount({ sockets: [ws] });
+    const { scroll, values } = metrics(conn, 1000, 200);
+    ws.open();
+    ws.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
+    runFrames();
+
+    values.top = top;
+    if (manuallyScrolled) scroll.dispatchEvent(new Event("scroll"));
+    const input = conn.el.querySelector<HTMLTextAreaElement>(".structured-input")!;
+    Object.defineProperty(input, "scrollHeight", { configurable: true, value: 120 });
+    input.value = "one\ntwo\nthree";
+    input.dispatchEvent(new Event("input"));
+    values.height = 1080;
+    runFrames();
+
+    expect(input.style.height).toBe("120px");
+    expect(values.top).toBe(expected);
+    conn.dispose();
+  });
+
+  it.each([
+    { position: "pinned", top: 800, expected: 850, manuallyScrolled: false },
+    { position: "manually scrolled", top: 180, expected: 180, manuallyScrolled: true },
+  ])("preserves the $position position across the narrow breakpoint", ({ top, expected, manuallyScrolled }) => {
+    const ws = new FakeSocket();
+    const conn = mount({ sockets: [ws] });
+    const { scroll, values } = metrics(conn, 1000, 200);
+    ws.open();
+    ws.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
+    runFrames();
+
+    values.top = top;
+    if (manuallyScrolled) scroll.dispatchEvent(new Event("scroll"));
+    resizeBody(500);
+    values.clientHeight = 150;
+    runFrames();
+
+    expect(conn.el.querySelector(".structured-body")!.classList).toContain("st-narrow");
+    expect(values.top).toBe(expected);
+    resizeBody(500);
+    expect(frames.size).toBe(0);
+    conn.dispose();
+  });
+
+  it("preserves a user-scrolled position across updates, approvals, and tool expansion", () => {
+    const ws = new FakeSocket();
+    const conn = mount({ sockets: [ws] });
+    const { scroll, values } = metrics(conn, 1000, 200);
+    ws.open();
+    ws.emit({ type: "transcript_snapshot", snapshot: { cursor: 1, rows: [toolRow("tool:1", "output")] } });
+    runFrames();
+
+    values.top = 180;
+    scroll.dispatchEvent(new Event("scroll"));
+    values.height = 1120;
+    ws.emit({
+      type: "projection_update",
+      transcript: { type: "append", row: row("agent:2", "agent_text", { content: { type: "text", text: "new" } }) },
+      state: idleState,
+    });
+    runFrames();
+    expect(values.top).toBe(180);
+
+    values.height = 1180;
+    ws.emit({
+      type: "projection_update",
+      transcript: {
+        type: "append",
+        row: row("approval:1", "approval", { toolCall: {}, options: allApprovalOptions }),
+      },
+      state: {
+        ...idleState,
+        pending_approval: {
+          nonce: "n1",
+          request: { sessionId: "s1", toolCall: {}, options: allApprovalOptions },
+        },
+      },
+    });
+    runFrames();
+    expect(values.top).toBe(180);
+
+    conn.el.querySelector<HTMLButtonElement>(".st-tool-head")!.click();
+    values.height = 1260;
+    runFrames();
+    expect(values.top).toBe(180);
+    conn.dispose();
+  });
+
+  it("resumes following when the user scrolls back near the bottom", () => {
+    const ws = new FakeSocket();
+    const conn = mount({ sockets: [ws] });
+    const { scroll, values } = metrics(conn, 1000, 200);
+    ws.open();
+    ws.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
+    runFrames();
+
+    values.top = 750;
+    scroll.dispatchEvent(new Event("scroll"));
+    values.height = 1100;
+    ws.emit({
+      type: "projection_update",
+      transcript: {
+        type: "append",
+        row: row("agent:1", "agent_text", { content: { type: "text", text: "follow" } }),
+      },
+      state: idleState,
+    });
+    runFrames();
+    expect(values.top).toBe(900);
+    conn.dispose();
+  });
+
+  it("preserves a user-scrolled position across a reconnect snapshot", () => {
+    const first = new FakeSocket();
+    const second = new FakeSocket();
+    const conn = mount({ sockets: [first, second], shouldReconnect: () => true });
+    const { scroll, values } = metrics(conn, 1000, 200);
+    first.open();
+    first.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
+    runFrames();
+
+    values.top = 160;
+    scroll.dispatchEvent(new Event("scroll"));
+    first.drop();
+    vi.advanceTimersByTime(1000);
+    second.open();
+    values.height = 1200;
+    second.emit({
+      type: "transcript_snapshot",
+      snapshot: {
+        cursor: 1,
+        rows: [
+          row("agent:1", "agent_text", {
+            content: { type: "text", text: "reconnected" },
+          }),
+        ],
+      },
+    });
+    runFrames();
+    expect(values.top).toBe(160);
+    conn.dispose();
+  });
+
+  it("cancels a pending transcript frame when disposed or parked", () => {
+    const ws = new FakeSocket();
+    const conn = mount({ sockets: [ws] });
+    const { values } = metrics(conn, 900, 300, 25);
+    ws.open();
+    ws.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
+    expect(frames.size).toBe(1);
+
+    conn.onUnmount();
+    expect(frames.size).toBe(0);
+    const input = conn.el.querySelector<HTMLTextAreaElement>(".structured-input")!;
+    Object.defineProperty(input, "scrollHeight", { configurable: true, value: 120 });
+    input.dispatchEvent(new Event("input"));
+    resizeBody(500);
+    expect(frames.size).toBe(0);
+    conn.onMount({ focus: false });
+    expect(frames.size).toBe(2);
+    conn.dispose();
+    expect(frames.size).toBe(0);
+    resizeBody(700);
+    expect(conn.el.querySelector(".structured-body")!.classList).toContain("st-narrow");
+    runFrames();
+    expect(values.top).toBe(25);
+  });
+});
+
 describe("structured connection: pane lifecycle", () => {
   beforeEach(() => vi.clearAllMocks());
 

web/src/structured.ts

@@ -80,6 +80,7 @@ export function createStructuredConnection(
   const stateEl = el.querySelector(".attach-state") as HTMLSpanElement;
   const pillsEl = el.querySelector(".st-pills") as HTMLSpanElement;
   const disconnectBtn = el.querySelector(".disconnect") as HTMLButtonElement;
+  const scrollEl = el.querySelector(".structured-scroll") as HTMLDivElement;
   const rowsEl = el.querySelector(".structured-rows") as HTMLDivElement;
   const railEl = el.querySelector(".structured-rail") as HTMLElement;
   const bodyEl = el.querySelector(".structured-body") as HTMLElement;
@@ -117,6 +118,65 @@ export function createStructuredConnection(
   let stateSocketGeneration = 0;
   let disposed = false;
 
+  // Within 64px, wheel and touch imprecision should still count as following.
+  const FOLLOW_BOTTOM_THRESHOLD_PX = 64;
+  let followingBottom = true;
+  let receivedSnapshot = false;
+  let scrollWorkActive = true;
+  let parkedScrollTop: number | null = null;
+  let scrollFrame: number | null = null;
+  let pendingScroll: { top: number; follow: boolean } | null = null;
+
+  function cancelScrollFrame(): void {
+    if (scrollFrame !== null) window.cancelAnimationFrame(scrollFrame);
+    scrollFrame = null;
+    pendingScroll = null;
+  }
+
+  function beginTranscriptLayout(forceBottom = false): void {
+    if (!scrollWorkActive || disposed) return;
+    if (pendingScroll) {
+      if (forceBottom) pendingScroll.follow = true;
+      return;
+    }
+    pendingScroll = {
+      top: scrollEl.scrollTop,
+      follow: forceBottom || followingBottom,
+    };
+  }
+
+  function finishTranscriptLayout(): void {
+    if (!scrollWorkActive || disposed || !pendingScroll || scrollFrame !== null) return;
+    scrollFrame = window.requestAnimationFrame(() => {
+      scrollFrame = null;
+      const pending = pendingScroll;
+      pendingScroll = null;
+      if (!pending || disposed || !scrollWorkActive) return;
+      scrollEl.scrollTop = pending.follow
+        ? Math.max(0, scrollEl.scrollHeight - scrollEl.clientHeight)
+        : pending.top;
+    });
+  }
+
+  function updateTranscriptLayout(update: () => void, forceBottom = false): void {
+    beginTranscriptLayout(forceBottom);
+    update();
+    finishTranscriptLayout();
+  }
+
+  function onTranscriptScroll(): void {
+    const distance = scrollEl.scrollHeight - scrollEl.clientHeight - scrollEl.scrollTop;
+    followingBottom = distance <= FOLLOW_BOTTOM_THRESHOLD_PX;
+    cancelScrollFrame();
+  }
+
+  function onTranscriptClick(event: Event): void {
+    if ((event.target as Element | null)?.closest(".st-tool-head")) beginTranscriptLayout();
+  }
+
+  scrollEl.addEventListener("scroll", onTranscriptScroll);
+  scrollEl.addEventListener("click", onTranscriptClick, true);
+
   function lastRowIdOfKind(kind: TranscriptRowKind): string | null {
     for (let i = rows.length - 1; i >= 0; i--) if (rows[i].kind === kind) return rows[i].id;
     return null;
@@ -160,6 +220,7 @@ export function createStructuredConnection(
       (expanded) => {
         if (expanded) expandedRowIds.add(row.id);
         else expandedRowIds.delete(row.id);
+        finishTranscriptLayout();
       },
     );
     if (!el) return null;
@@ -207,44 +268,50 @@ export function createStructuredConnection(
   }
 
   function applySnapshot(snapshot: { rows: TranscriptRow[] }): void {
-    rows = [...snapshot.rows];
-    const expandableIds = new Set(
-      rows.filter((row) => row.kind === "tool_call" || row.kind === "tool_diff").map((row) => row.id),
-    );
-    for (const id of expandedRowIds) if (!expandableIds.has(id)) expandedRowIds.delete(id);
-    rowIndex.clear();
-    rows.forEach((row, i) => rowIndex.set(row.id, i));
-    rowEls.clear();
-    rowsEl.replaceChildren();
-    for (const row of rows) mountRow(row);
+    const forceBottom = !receivedSnapshot;
+    updateTranscriptLayout(() => {
+      rows = [...snapshot.rows];
+      const expandableIds = new Set(
+        rows.filter((row) => row.kind === "tool_call" || row.kind === "tool_diff").map((row) => row.id),
+      );
+      for (const id of expandedRowIds) if (!expandableIds.has(id)) expandedRowIds.delete(id);
+      rowIndex.clear();
+      rows.forEach((row, i) => rowIndex.set(row.id, i));
+      rowEls.clear();
+      rowsEl.replaceChildren();
+      for (const row of rows) mountRow(row);
+    }, forceBottom);
+    receivedSnapshot = true;
   }
 
   function applyDelta(delta: TranscriptDelta | null | undefined): void {
     if (!delta) return;
-    if (delta.type === "append") {
-      rowIndex.set(delta.row.id, rows.length);
-      rows.push(delta.row);
-      mountRow(delta.row);
-    } else if (delta.type === "patch") {
-      const i = rowIndex.get(delta.row.id);
-      if (i === undefined) {
+    updateTranscriptLayout(() => {
+      if (delta.type === "append") {
         rowIndex.set(delta.row.id, rows.length);
         rows.push(delta.row);
         mountRow(delta.row);
+      } else if (delta.type === "patch") {
+        const i = rowIndex.get(delta.row.id);
+        if (i === undefined) {
+          rowIndex.set(delta.row.id, rows.length);
+          rows.push(delta.row);
+          mountRow(delta.row);
+        } else {
+          rows[i] = delta.row;
+          remountRow(delta.row);
+        }
       } else {
-        rows[i] = delta.row;
-        remountRow(delta.row);
+        const i = rowIndex.get(delta.id);
+        if (i === undefined) return;
+        rows.splice(i, 1);
+        expandedRowIds.delete(delta.id);
+        rowIndex.clear();
+        rows.forEach((row, idx) => rowIndex.set(row.id, idx));
+        rowEls.get(delta.id)?.remove();
+        rowEls.delete(delta.id);
       }
-    } else {
-      const i = rowIndex.get(delta.id);
-      if (i === undefined) return;
-      rows.splice(i, 1);
-      expandedRowIds.delete(delta.id);
-      rowIndex.clear();
-      rows.forEach((row, idx) => rowIndex.set(row.id, idx));
-      rowEls.get(delta.id)?.remove();
-      rowEls.delete(delta.id);
-    }
+    });
   }
 
   function toolRowCount(): number {
@@ -293,10 +360,12 @@ export function createStructuredConnection(
     reconcileAwaitingBegin();
     const newPendingId = state.pending_approval ? lastRowIdOfKind("approval") : null;
     if (newPendingId !== pendingApprovalRowId) {
-      const prev = pendingApprovalRowId;
-      pendingApprovalRowId = newPendingId;
-      if (prev !== null && rowIndex.has(prev)) remountRow(rows[rowIndex.get(prev)!]);
-      if (newPendingId !== null) remountRow(rows[rowIndex.get(newPendingId)!]);
+      updateTranscriptLayout(() => {
+        const prev = pendingApprovalRowId;
+        pendingApprovalRowId = newPendingId;
+        if (prev !== null && rowIndex.has(prev)) remountRow(rows[rowIndex.get(prev)!]);
+        if (newPendingId !== null) remountRow(rows[rowIndex.get(newPendingId)!]);
+      });
     }
     renderPills();
     renderRailNow();
@@ -320,8 +389,10 @@ export function createStructuredConnection(
 
   // ---- composer ----
   function autoResize(): void {
-    input.style.height = "auto";
-    input.style.height = `${Math.min(input.scrollHeight, 160)}px`;
+    updateTranscriptLayout(() => {
+      input.style.height = "auto";
+      input.style.height = `${Math.min(input.scrollHeight, 160)}px`;
+    });
   }
   input.addEventListener("input", autoResize);
   input.addEventListener("keydown", (e) => {
@@ -377,8 +448,11 @@ export function createStructuredConnection(
   // ---- narrow-pane rail collapse (a split pane, not just the viewport) ----
   const NARROW_BREAKPOINT = 560;
   const resizeObserver = new ResizeObserver((entries) => {
+    if (disposed) return;
     const width = entries[0]?.contentRect.width ?? bodyEl.clientWidth;
-    bodyEl.classList.toggle("st-narrow", width < NARROW_BREAKPOINT);
+    const narrow = width < NARROW_BREAKPOINT;
+    if (bodyEl.classList.contains("st-narrow") === narrow) return;
+    updateTranscriptLayout(() => bodyEl.classList.toggle("st-narrow", narrow));
   });
   resizeObserver.observe(bodyEl);
 
@@ -493,6 +567,12 @@ export function createStructuredConnection(
   // `mountToken` is bumped on every mount/unmount so a stale frame can tell
   // it is no longer the most recent one.
   let mountToken = 0;
+  let focusFrame: number | null = null;
+
+  function cancelFocusFrame(): void {
+    if (focusFrame !== null) window.cancelAnimationFrame(focusFrame);
+    focusFrame = null;
+  }
 
   return {
     sessionId,
@@ -500,13 +580,26 @@ export function createStructuredConnection(
     onMount(mountOpts?: { focus?: boolean }) {
       const focusOnMount = mountOpts?.focus !== false;
       const token = ++mountToken;
-      requestAnimationFrame(() => {
+      scrollWorkActive = true;
+      pendingScroll = {
+        top: parkedScrollTop ?? scrollEl.scrollTop,
+        follow: followingBottom,
+      };
+      parkedScrollTop = null;
+      finishTranscriptLayout();
+      cancelFocusFrame();
+      focusFrame = window.requestAnimationFrame(() => {
+        focusFrame = null;
         if (disposed || token !== mountToken) return; // parked/remounted since this was scheduled
         if (focusOnMount && !input.disabled) input.focus();
       });
     },
     onUnmount() {
       mountToken++;
+      parkedScrollTop = pendingScroll?.top ?? scrollEl.scrollTop;
+      scrollWorkActive = false;
+      cancelScrollFrame();
+      cancelFocusFrame();
     },
     focusTerminal() {
       if (!input.disabled) input.focus();
@@ -519,11 +612,16 @@ export function createStructuredConnection(
       if (disposed) return;
       disposed = true;
       mountToken++;
+      scrollWorkActive = false;
+      cancelScrollFrame();
+      cancelFocusFrame();
       cancelReconnect();
       resizeObserver.disconnect();
       unsubscribeStore?.();
       closeCurrentSocket();
       expandedRowIds.clear();
+      scrollEl.removeEventListener("scroll", onTranscriptScroll);
+      scrollEl.removeEventListener("click", onTranscriptClick, true);
       el.remove();
     },
   };