fix(maestro): anchor structured TUI viewport to row identity

follow_selection=false stored scroll as a raw rendered-line offset into
lines rebuilt fresh every frame. A ProjectionUpdate::Patch growing a
streaming row above the viewport shifted every line below it, so the
raw offset kept pointing at a numerically-same but visually different
line and the viewport drifted even though the operator had scrolled
away deliberately.

Cache each frame's row-id-per-line layout and, before applying scroll
next frame, translate the old raw offset into (row id, intra-row line)
against that cache, then relocate it in the freshly rebuilt lines. Rows
above or intersecting the viewport growing, shrinking, or being
patched now shift the offset instead of the visible content. Falls
back to the nearest surviving neighbor row when the anchor row itself
is removed, and to the clamped raw offset when nothing nearby
survived. follow_selection and bottom-follow are unaffected — they
already recompute scroll from the live selection/tail each frame.

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-cli/src/attach.rs

@@ -19,6 +19,12 @@ use tokio_tungstenite::tungstenite::Message;
 pub enum AttachError {
     #[error("ws: {0}")]
     WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
+    #[error("daemon returned {status} ({code}): {message}")]
+    Rejected {
+        status: String,
+        code: String,
+        message: String,
+    },
     #[error("io: {0}")]
     Io(#[from] io::Error),
     #[error("termios: {0}")]
@@ -32,6 +38,21 @@ pub enum AttachError {
 }
 
 impl AttachError {
+    fn from_websocket(error: tokio_tungstenite::tungstenite::Error) -> Self {
+        if let tokio_tungstenite::tungstenite::Error::Http(response) = &error {
+            if let Some(body) = response.body().as_deref() {
+                if let Ok(error_body) = serde_json::from_slice::<maestro_protocol::ApiError>(body) {
+                    return Self::Rejected {
+                        status: response.status().to_string(),
+                        code: error_body.error.code,
+                        message: error_body.error.message,
+                    };
+                }
+            }
+        }
+        Self::WebSocket(error)
+    }
+
     /// True when the failure was an inability to reach the WS endpoint at the
     /// transport level (connection refused), as opposed to an HTTP-level
     /// rejection from a daemon that did answer.
@@ -111,7 +132,9 @@ fn current_winsize() -> Option<(u16, u16)> {
 /// byte that exits cleanly without killing the session.
 #[allow(clippy::result_large_err)]
 pub async fn run_attach_url(url: &str, detach_byte: u8) -> Result<(), AttachError> {
-    let (ws, _resp) = tokio_tungstenite::connect_async(url).await?;
+    let (ws, _resp) = tokio_tungstenite::connect_async(url)
+        .await
+        .map_err(AttachError::from_websocket)?;
     let (mut sink, mut stream) = ws.split();
 
     let raw = RawTty::enable()?;
@@ -233,4 +256,25 @@ mod tests {
             Err(AttachError::InvalidDetachKey(_))
         ));
     }
+
+    #[test]
+    fn attach_surfaces_structured_handshake_errors() {
+        let body = serde_json::to_vec(&maestro_protocol::ApiError::new(
+            "not_terminal",
+            "session is not a terminal session",
+        ))
+        .unwrap();
+        let response = tokio_tungstenite::tungstenite::http::Response::builder()
+            .status(409)
+            .body(Some(body))
+            .unwrap();
+
+        let error =
+            AttachError::from_websocket(tokio_tungstenite::tungstenite::Error::Http(response));
+
+        assert_eq!(
+            error.to_string(),
+            "daemon returned 409 Conflict (not_terminal): session is not a terminal session"
+        );
+    }
 }

crates/maestro-cli/src/sessions.rs

@@ -6,7 +6,7 @@ use std::io::Write;
 use maestro_protocol::display::abbreviate_dir;
 use maestro_protocol::{
     ActivityState, ClientError, CreateSessionRequest, DaemonClient, DaemonInfo, SendInputRequest,
-    Session, SessionStatus, WorktreeSpec,
+    Session, SessionStatus, SessionView, WorktreeSpec,
 };
 use serde_json::Value;
 
@@ -179,8 +179,8 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], info: &Daemon
     }
     writeln!(
         out,
-        "  {:<36}  {:<8}  {:<9}  {:<28}  {:<40}  {}",
+        "  {:<36}  {:<8}  {:<10}  {:<9}  {:<28}  {:<40}  {}",
-        "ID", "STATUS", "ACTIVITY", "DIR", "COMMAND", "BLOCKED"
+        "ID", "STATUS", "VIEW", "ACTIVITY", "DIR", "COMMAND", "BLOCKED"
     )
     .ok();
     for s in sorted {
@@ -190,6 +190,10 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], info: &Daemon
             SessionStatus::Running => "running",
             SessionStatus::Dead => "dead",
         };
+        let view = match s.view {
+            SessionView::Terminal => "terminal",
+            SessionView::Structured => "structured",
+        };
         let activity = match s.activity {
             ActivityState::Active => "active",
             ActivityState::Idle => "idle",
@@ -207,8 +211,8 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], info: &Daemon
         let blocked = blocked_label(s);
         writeln!(
             out,
-            "  {:<36}  {:<8}  {:<9}  {:<28}  {:<40}  {blocked}",
+            "  {:<36}  {:<8}  {:<10}  {:<9}  {:<28}  {:<40}  {blocked}",
-            s.id, status, activity, dir, command
+            s.id, status, view, activity, dir, command
         )
         .ok();
     }
@@ -230,7 +234,7 @@ mod tests {
     use std::path::PathBuf;
 
     use chrono::{TimeZone, Utc};
-    use maestro_protocol::{ActivityState, Session, SessionStatus};
+    use maestro_protocol::{ActivityState, Session, SessionStatus, SessionView};
     use uuid::Uuid;
 
     use super::DaemonInfo;
@@ -286,6 +290,22 @@ mod tests {
         assert!(s.contains("idle"), "{s}");
     }
 
+    #[test]
+    fn table_exposes_terminal_and_structured_views() {
+        let terminal = session("/tmp", "bash", 1);
+        let mut structured = session("/tmp", "claude", 2);
+        structured.view = SessionView::Structured;
+
+        let mut buf = Vec::new();
+        super::print_session_table(&mut buf, &[terminal, structured], &info(""));
+        let out = String::from_utf8(buf).unwrap();
+        let lines: Vec<&str> = out.lines().collect();
+        let header = lines.iter().position(|line| line.contains("VIEW")).unwrap();
+
+        assert!(lines[header + 1].contains("terminal"), "{out}");
+        assert!(lines[header + 2].contains("structured"), "{out}");
+    }
+
     #[test]
     fn live_cwd_overrides_spawn_dir_in_table() {
         let mut s = session("/home/naps/maestro", "bash", 1);

crates/maestro-daemon/src/local_api.rs

@@ -26,7 +26,7 @@ use maestro_protocol::{
 use serde::Deserialize;
 use uuid::Uuid;
 
-use crate::session::{QueueAcpPrompt, SessionManager};
+use crate::session::{QueueAcpPrompt, SessionError, SessionManager};
 use crate::sidecar::EventCursor;
 
 #[derive(Debug, Clone)]
@@ -179,6 +179,7 @@ fn session_error_status(code: &str) -> StatusCode {
         // The session outlived its pane: nothing to type into, and nothing
         // the daemon can do about it.
         "not_running" => StatusCode::CONFLICT,
+        "not_terminal" => StatusCode::CONFLICT,
         "shutting_down" => StatusCode::SERVICE_UNAVAILABLE,
         _ => StatusCode::BAD_GATEWAY,
     }
@@ -534,12 +535,23 @@ async fn attach_ws(
     State(s): State<LocalApiState>,
     ws: axum::extract::WebSocketUpgrade,
 ) -> Result<Response, Response> {
-    let Some((replay, rx)) = s.manager.subscribe_pty_with_replay(id).await else {
+    let (replay, rx) = match s.manager.subscribe_pty_with_replay(id).await {
-        return Err(error_response(
+        Ok(subscription) => subscription,
-            StatusCode::NOT_FOUND,
+        Err(SessionError::NotFound(_)) => {
-            "unknown_session",
+            return Err(error_response(
-            "no such session on this daemon",
+                StatusCode::NOT_FOUND,
-        ));
+                "unknown_session",
+                "no such session on this daemon",
+            ));
+        }
+        Err(error) => {
+            let info = error.to_error_info();
+            return Err(error_response(
+                session_error_status(&info.code),
+                &info.code,
+                &info.message,
+            ));
+        }
     };
     let manager = s.manager.clone();
     Ok(ws.on_upgrade(move |socket| async move {

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::pin::Pin;
-use std::sync::Arc;
+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)]
@@ -169,6 +190,8 @@ pub enum SessionError {
     /// a stray keystroke on an attach socket, which is dropped.
     #[error("session is no longer running: {0}")]
     NotRunning(Uuid),
+    #[error("session is not a terminal session")]
+    NotTerminal(Uuid),
     #[error("daemon is shutting down")]
     ShuttingDown,
     /// The kill was refused (#79): the session's worktree holds unsaved work.
@@ -205,6 +228,7 @@ impl SessionError {
             SessionError::NotFound(_) => "not_found",
             SessionError::InvalidEnv(_) => "invalid_env",
             SessionError::NotRunning(_) => "not_running",
+            SessionError::NotTerminal(_) => "not_terminal",
             SessionError::ShuttingDown => "shutting_down",
             SessionError::KillGuarded(_) => "kill_guarded",
         };
@@ -260,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
+    /// Admission is serialized with shutdown, while admitted handshakes run
-    /// either rejected or included in shutdown's final drain.
+    /// independently and are counted until registration or rollback finishes.
-    structured_lifecycle: Mutex<()>,
+    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 {
@@ -306,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(),
             }),
         }
     }
@@ -824,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
@@ -879,11 +1001,17 @@ impl SessionManager {
         }
     }
 
-    /// Subscribe to a session's PTY broadcast. `None` if the session is
+    /// Subscribe to a terminal session's PTY broadcast.
-    /// unknown to this daemon.
+    pub async fn subscribe_pty(
-    pub async fn subscribe_pty(&self, id: Uuid) -> Option<broadcast::Receiver<Vec<u8>>> {
+        &self,
+        id: Uuid,
+    ) -> Result<broadcast::Receiver<Vec<u8>>, SessionError> {
         let sessions = self.inner.sessions.lock().await;
-        sessions.get(&id).map(|r| r.bridge.subscribe())
+        let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
+        if record.session.view != SessionView::Terminal {
+            return Err(SessionError::NotTerminal(id));
+        }
+        Ok(record.bridge.subscribe())
     }
 
     /// Subscribe to a session's PTY broadcast and return the replay buffer
@@ -897,11 +1025,14 @@ impl SessionManager {
     pub async fn subscribe_pty_with_replay(
         &self,
         id: Uuid,
-    ) -> Option<(Vec<u8>, broadcast::Receiver<Vec<u8>>)> {
+    ) -> Result<(Vec<u8>, broadcast::Receiver<Vec<u8>>), SessionError> {
         let tmux = self.inner.tmux.clone();
         let (result, updated) = {
             let mut sessions = self.inner.sessions.lock().await;
-            let rec = sessions.get_mut(&id)?;
+            let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
+            if rec.session.view != SessionView::Terminal {
+                return Err(SessionError::NotTerminal(id));
+            }
             rec.session.last_focused_at = Some(Utc::now());
             let updated = rec.session.clone();
             // For a live session, tmux's own screen beats the byte tail: the
@@ -932,7 +1063,7 @@ impl SessionManager {
             .inner
             .events
             .send(CatalogEvent::SessionUpdated { session: updated });
-        Some(result)
+        Ok(result)
     }
 
     pub async fn has(&self, id: Uuid) -> bool {
@@ -945,6 +1076,15 @@ impl SessionManager {
         sessions.get(&id).map(|r| r.session.status)
     }
 
+    async fn terminal_status(&self, id: Uuid) -> Result<SessionStatus, SessionError> {
+        let sessions = self.inner.sessions.lock().await;
+        let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
+        if record.session.view != SessionView::Terminal {
+            return Err(SessionError::NotTerminal(id));
+        }
+        Ok(record.session.status)
+    }
+
     /// Whether the natural-exit settle step (#82) has finished for this
     /// session: `Some(true)` once applied or refused, `Some(false)` while
     /// still pending, `None` if the session is unknown. `Dead` is observable
@@ -959,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;
+            let permit = self
-            if self.inner.shutting_down.load(Ordering::Acquire) {
+                .inner
-                return Err(SessionError::ShuttingDown);
+                .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
     }
@@ -1001,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 =
+        let mut pending = PendingCreate {
-            match PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await {
+            inner: &self.inner,
-                Ok(b) => b,
+            id,
-                Err(e) => {
+            fifo_path: fifo_path.clone(),
-                    self.rollback_worktree(&worktree_base_repo, &worktree_path);
+            worktree_path: worktree_path.clone(),
-                    return Err(e.into());
+            worktree_base_repo: worktree_base_repo.clone(),
-                }
+            committed: false,
-            };
-        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 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).
@@ -1063,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)
@@ -1084,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));
                     }
                 };
@@ -1105,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));
                 }
             }
@@ -1180,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));
         }
 
@@ -1192,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 {
@@ -1208,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(),
         });
@@ -1243,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 —
@@ -1435,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,
                 },
             );
@@ -1556,6 +1676,7 @@ impl SessionManager {
                             pending_approvals: HashMap::new(),
                             operation: Arc::new(Mutex::new(())),
                             dispatch: AcpDispatchState::default(),
+                            activity_sequence: 0,
                         })))
                     }
                     Err(error) => {
@@ -1603,6 +1724,7 @@ impl SessionManager {
                     awaiting_suppressed: false,
                     acp,
                     acp_events,
+                    structured_activity_watermark: None,
                     acp_event_pump: None,
                 },
             );
@@ -1626,10 +1748,9 @@ impl SessionManager {
     /// its pane is gone, so this is a no-op rather than a tmux failure the
     /// client would have to interpret.
     pub async fn send_keys(&self, id: Uuid, bytes: &[u8]) -> Result<(), SessionError> {
-        match self.status(id).await {
+        match self.terminal_status(id).await? {
-            None => return Err(SessionError::NotFound(id)),
+            SessionStatus::Dead => return Ok(()),
-            Some(SessionStatus::Dead) => return Ok(()),
+            SessionStatus::Running => {}
-            Some(SessionStatus::Running) => {}
         }
         let tmux = self.inner.tmux.clone();
         let bytes = bytes.to_vec();
@@ -1645,10 +1766,9 @@ impl SessionManager {
     /// asked for text to land somewhere, and a silent success would let a
     /// script believe an agent had been told something it never heard.
     pub async fn send_input(&self, id: Uuid, text: &str, submit: bool) -> Result<(), SessionError> {
-        match self.status(id).await {
+        match self.terminal_status(id).await? {
-            None => return Err(SessionError::NotFound(id)),
+            SessionStatus::Dead => return Err(SessionError::NotRunning(id)),
-            Some(SessionStatus::Dead) => return Err(SessionError::NotRunning(id)),
+            SessionStatus::Running => {}
-            Some(SessionStatus::Running) => {}
         }
         let tmux = self.inner.tmux.clone();
         let bytes = text.as_bytes().to_vec();
@@ -1668,10 +1788,9 @@ impl SessionManager {
     /// Resize a session's window. No-op for a dead session, as with
     /// [`Self::send_keys`].
     pub async fn resize(&self, id: Uuid, cols: u16, rows: u16) -> Result<(), SessionError> {
-        match self.status(id).await {
+        match self.terminal_status(id).await? {
-            None => return Err(SessionError::NotFound(id)),
+            SessionStatus::Dead => return Ok(()),
-            Some(SessionStatus::Dead) => return Ok(()),
+            SessionStatus::Running => {}
-            Some(SessionStatus::Running) => {}
         }
         let tmux = self.inner.tmux.clone();
         tmux_blocking(move || tmux.resize_window(id, cols, rows))
@@ -1703,6 +1822,7 @@ impl SessionManager {
         cols: u16,
         rows: u16,
     ) -> Result<(), SessionError> {
+        self.terminal_status(id).await?;
         let target = self
             .inner
             .viewports
@@ -2025,57 +2145,59 @@ async fn sync_structured_activity_from_stream(
     id: Uuid,
     events: Arc<Mutex<AcpEventStream>>,
 ) {
-    let event_stream = events.lock().await;
+    let (sequence, raw) = events.lock().await.activity_snapshot();
-    let raw = event_stream.projection.activity();
+    apply_structured_activity(inner, id, Some((&events, sequence)), raw).await;
-    // 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;
 }
 
 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 mut sessions = inner.sessions.lock().await;
+    let Some(record) = sessions.get_mut(&id) else {
-        let Some(record) = sessions.get_mut(&id) else {
+        return;
-            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())
-        }
     };
-    if let Some(session) = updated {
+    if record.session.view != SessionView::Structured {
-        let _ = inner.events.send(CatalogEvent::SessionUpdated { session });
+        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(),
+        });
     }
 }
 
@@ -2752,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,
+        apply_structured_activity, cancel_pending_approvals, permission_outcome,
-        run_structured_activity_pass, sync_structured_activity_from_stream, AcpDispatchState,
+        record_permission_request, run_structured_activity_pass,
-        AcpEventStream, ApprovalDecision, RecordOutcome, SessionManager,
+        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,
     };
@@ -2790,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,
         }
     }
 
@@ -2807,13 +2931,15 @@ mod tests {
     async fn manager_with_structured_stream() -> (
         tempfile::TempDir,
         SessionManager,
+        Arc<FakePanes>,
         uuid::Uuid,
         Arc<tokio::sync::Mutex<AcpEventStream>>,
         uuid::Uuid,
     ) {
         let state = tempfile::tempdir().unwrap();
+        let panes = Arc::new(FakePanes::new());
         let manager = SessionManager::new(
-            Arc::new(FakePanes::new()),
+            panes.clone(),
             SidecarStore::new(state.path().join("sessions")),
             WorktreeManager::new(None),
             state.path().join("fifos"),
@@ -2848,12 +2974,12 @@ mod tests {
             record.focused_since = Some(std::time::Instant::now() - FOCUS_CLEAR_HOLD);
             record.acp_events = Some(events.clone());
         }
-        (state, manager, structured.id, events, terminal.id)
+        (state, manager, panes, structured.id, events, terminal.id)
     }
 
     #[tokio::test]
     async fn blocked_structured_activity_read_does_not_block_the_catalog() {
-        let (state, manager, structured_id, events, terminal_id) =
+        let (state, manager, _panes, structured_id, events, terminal_id) =
             manager_with_structured_stream().await;
         let held_stream = events.lock().await;
         let activity_pass = run_structured_activity_pass(&manager.inner);
@@ -2898,9 +3024,136 @@ mod tests {
         assert_eq!(structured.activity, ActivityState::Idle);
     }
 
+    #[tokio::test]
+    async fn structured_sessions_reject_every_pty_operation_before_panes() {
+        let (_state, manager, panes, structured_id, _events, terminal_id) =
+            manager_with_structured_stream().await;
+        let before = panes.calls();
+
+        assert!(matches!(
+            manager.subscribe_pty(structured_id).await,
+            Err(super::SessionError::NotTerminal(id)) if id == structured_id
+        ));
+        assert!(matches!(
+            manager.subscribe_pty_with_replay(structured_id).await,
+            Err(super::SessionError::NotTerminal(id)) if id == structured_id
+        ));
+        assert!(matches!(
+            manager.send_keys(structured_id, b"ignored").await,
+            Err(super::SessionError::NotTerminal(id)) if id == structured_id
+        ));
+        assert!(matches!(
+            manager.send_input(structured_id, "ignored", false).await,
+            Err(super::SessionError::NotTerminal(id)) if id == structured_id
+        ));
+        assert!(matches!(
+            manager.resize(structured_id, 80, 24).await,
+            Err(super::SessionError::NotTerminal(id)) if id == structured_id
+        ));
+        assert!(matches!(
+            manager
+                .set_viewport(structured_id, manager.attach_viewport(), 80, 24)
+                .await,
+            Err(super::SessionError::NotTerminal(id)) if id == structured_id
+        ));
+
+        let after = panes.calls();
+        assert_eq!(after, before, "structured PTY operations touched panes");
+
+        manager
+            .send_input(terminal_id, "works", false)
+            .await
+            .expect("terminal input remains available");
+        manager
+            .resize(terminal_id, 100, 30)
+            .await
+            .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, structured_id, old_events, _terminal_id) =
+        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
@@ -2931,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();

crates/maestro-daemon/tests/local_api.rs

@@ -1311,6 +1311,39 @@ async fn attach_ws_streams_pty_bytes() {
     assert!(saw_hello, "PTY did not echo through the attach WS");
 }
 
+#[tokio::test]
+async fn pty_routes_reject_structured_sessions_as_not_terminal() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+
+    let response = reqwest::Client::new()
+        .post(url(h.addr, &format!("/api/sessions/{id}/input")))
+        .json(&maestro_protocol::SendInputRequest {
+            text: "ignored".into(),
+            submit: false,
+        })
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = response.json().await.unwrap();
+    assert_eq!(error.error.code, "not_terminal");
+    assert_eq!(error.error.message, "session is not a terminal session");
+
+    let error = connect_async(ws_url(h.addr, &format!("/api/attach/{id}")))
+        .await
+        .expect_err("structured session must not upgrade to a PTY socket");
+    let tokio_tungstenite::tungstenite::Error::Http(response) = error else {
+        panic!("expected an HTTP handshake rejection, got {error}");
+    };
+    assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
+    let body = response.body().as_deref().expect("ApiError response body");
+    let error: ApiError = serde_json::from_slice(body).unwrap();
+    assert_eq!(error.error.code, "not_terminal");
+    assert_eq!(error.error.message, "session is not a terminal session");
+}
+
 #[tokio::test]
 async fn attach_ws_for_unknown_session_returns_404() {
     let h = spawn("naps-laptop").await;

crates/maestro-tui/src/structured_pane.rs

@@ -90,6 +90,7 @@ impl StructuredPane {
 
         let viewport = usize::from(area.height);
         let max_scroll = lines.len().saturating_sub(viewport);
+        state.scroll = reconcile_scroll(&state.scroll, &state.line_rows, &lines);
         if state.follow_selection {
             if let Some(selected) = state.selected.as_deref() {
                 let first = lines
@@ -138,6 +139,7 @@ impl StructuredPane {
                 Rect::new(area.x, area.y + offset as u16, area.width, 1),
             );
         }
+        state.line_rows = lines.iter().map(|line| line.row_id.clone()).collect();
     }
 
     fn render_projection_status(&self, f: &mut Frame, area: Rect, state: &StructuredState) {
@@ -214,6 +216,96 @@ fn error_line(text: &str) -> VisualLine {
     }
 }
 
+/// Translate `scroll` — a raw line index into `old_rows` (the previous
+/// render's row-per-line layout) — into the equivalent index into `lines`
+/// (this render's freshly rebuilt layout), anchored on row identity rather
+/// than raw position. A row's wrapped lines above or intersecting the
+/// viewport growing, shrinking, or moving index therefore shifts `scroll`
+/// along with it instead of leaving the viewport looking at different
+/// content. Falls back to the nearest surviving neighbour row when the
+/// anchor row itself is gone, and to the raw index (clamped) when nothing
+/// nearby survived either.
+fn reconcile_scroll(scroll: &usize, old_rows: &[Option<String>], lines: &[VisualLine]) -> usize {
+    if old_rows.is_empty() {
+        return *scroll;
+    }
+    let old_index = (*scroll).min(old_rows.len() - 1);
+    let anchor = old_rows[old_index].clone();
+    match anchor {
+        Some(id) => match find_row_block(lines, &id) {
+            Some((start, len)) => {
+                start + (old_index - block_start(old_rows, old_index)).min(len - 1)
+            }
+            None => fallback_after_removal(old_rows, lines, old_index),
+        },
+        None => {
+            // Untagged tail lines (connecting/dead/error) carry no row identity;
+            // approximate by preserving distance from the end of the transcript.
+            let distance_from_end = old_rows.len() - old_index;
+            lines.len().saturating_sub(distance_from_end)
+        }
+    }
+}
+
+/// Index of the first line belonging to `id`, and how many lines it spans.
+/// A row's wrapped lines are always contiguous in `lines` — `build_lines`
+/// emits them as one run per row — so a single forward scan suffices.
+fn find_row_block(lines: &[VisualLine], id: &str) -> Option<(usize, usize)> {
+    let start = lines
+        .iter()
+        .position(|line| line.row_id.as_deref() == Some(id))?;
+    let mut end = start;
+    while end + 1 < lines.len() && lines[end + 1].row_id.as_deref() == Some(id) {
+        end += 1;
+    }
+    Some((start, end - start + 1))
+}
+
+/// First index of the contiguous run that `index` sits in within `rows`.
+fn block_start(rows: &[Option<String>], index: usize) -> usize {
+    let mut start = index;
+    while start > 0 && rows[start - 1] == rows[index] {
+        start -= 1;
+    }
+    start
+}
+
+/// The anchor row is gone (removed, or replaced wholesale by a fresh
+/// snapshot). Search outward from its old position, nearest first, for a
+/// still-live neighbour row and land on its first line; if none of the old
+/// layout's rows survived, fall back to the raw old index clamped into the
+/// new layout.
+fn fallback_after_removal(
+    old_rows: &[Option<String>],
+    lines: &[VisualLine],
+    old_index: usize,
+) -> usize {
+    let len = old_rows.len();
+    let mut distance = 1usize;
+    loop {
+        if distance > old_index && old_index + distance >= len {
+            return old_index.min(lines.len().saturating_sub(1));
+        }
+        if distance <= old_index {
+            if let Some((start, _)) = old_rows[old_index - distance]
+                .as_ref()
+                .and_then(|id| find_row_block(lines, id))
+            {
+                return start;
+            }
+        }
+        if old_index + distance < len {
+            if let Some((start, _)) = old_rows[old_index + distance]
+                .as_ref()
+                .and_then(|id| find_row_block(lines, id))
+            {
+                return start;
+            }
+        }
+        distance += 1;
+    }
+}
+
 fn build_lines(state: &StructuredState, width: usize) -> Vec<VisualLine> {
     let mut out = Vec::new();
     for row in &state.rows {
@@ -1272,4 +1364,213 @@ mod tests {
         assert!(text.contains("adapter exited"), "{text}");
         assert!(text.contains("request failed"), "{text}");
     }
+
+    fn test_pane() -> StructuredPane {
+        StructuredPane {
+            handle: StructuredHandle::test_handle(StructuredState::default()),
+            focused: true,
+            mode: Mode::Normal,
+            session_status: SessionStatus::Running,
+            activity_detail: None,
+        }
+    }
+
+    fn numbered_rows(state: &mut StructuredState, count: usize) {
+        for n in 0..count {
+            state.rows.push(row(
+                &format!("r{n}"),
+                TranscriptRowKind::AgentText,
+                json!({"content":{"type":"text","text":format!("line {n}")}}),
+            ));
+        }
+    }
+
+    fn grow_row(state: &mut StructuredState, id: &str) {
+        let index = state.rows.iter().position(|r| r.id == id).unwrap();
+        state.rows[index] = row(
+            id,
+            TranscriptRowKind::AgentText,
+            json!({"content":{"type":"text","text":"a very long streaming reply that keeps growing and will wrap across several lines now"}}),
+        );
+    }
+
+    fn render_transcript_lines(
+        pane: &StructuredPane,
+        state: &mut StructuredState,
+        width: u16,
+        height: u16,
+    ) -> Vec<String> {
+        let mut terminal = Terminal::new(TestBackend::new(width, height)).unwrap();
+        terminal
+            .draw(|frame| pane.render_transcript(frame, frame.area(), state))
+            .unwrap();
+        let buffer = terminal.backend().buffer().clone();
+        (0..height)
+            .map(|y| {
+                (0..width)
+                    .map(|x| buffer[(x, y)].symbol())
+                    .collect::<String>()
+            })
+            .collect()
+    }
+
+    #[test]
+    fn row_growing_above_the_viewport_keeps_the_visual_anchor() {
+        let pane = test_pane();
+        let mut state = StructuredState::default();
+        state.follow_selection = false;
+        numbered_rows(&mut state, 20);
+        state.scroll = 10;
+
+        let before = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert!(before[0].contains("line 10"), "{before:?}");
+
+        grow_row(&mut state, "r3");
+        let after = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert!(
+            after[0].contains("line 10"),
+            "a row growing above the viewport must not shift what's on screen: {after:?}"
+        );
+    }
+
+    #[test]
+    fn appending_a_row_below_the_viewport_keeps_the_visual_anchor() {
+        let pane = test_pane();
+        let mut state = StructuredState::default();
+        state.follow_selection = false;
+        numbered_rows(&mut state, 20);
+        state.scroll = 10;
+
+        let before = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert!(before[0].contains("line 10"), "{before:?}");
+
+        state.rows.push(row(
+            "r20",
+            TranscriptRowKind::AgentText,
+            json!({"content":{"type":"text","text":"line 20"}}),
+        ));
+        let after = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert_eq!(before, after, "growth below the viewport must not move it");
+    }
+
+    #[test]
+    fn removing_a_row_above_the_viewport_keeps_the_visual_anchor() {
+        let pane = test_pane();
+        let mut state = StructuredState::default();
+        state.follow_selection = false;
+        numbered_rows(&mut state, 20);
+        state.scroll = 10;
+
+        let before = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert!(before[0].contains("line 10"), "{before:?}");
+
+        state.rows.retain(|r| r.id != "r3");
+        let after = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert!(
+            after[0].contains("line 10"),
+            "a row removed above the viewport must not shift what's on screen: {after:?}"
+        );
+    }
+
+    #[test]
+    fn removing_the_anchor_row_falls_back_to_the_nearest_surviving_neighbor() {
+        let pane = test_pane();
+        let mut state = StructuredState::default();
+        state.follow_selection = false;
+        numbered_rows(&mut state, 20);
+        state.scroll = 10;
+
+        let before = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert!(before[0].contains("line 10"), "{before:?}");
+
+        state.rows.retain(|r| r.id != "r10");
+        let after = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert!(
+            after[0].contains("line 9"),
+            "removing the anchor row should fall back to its nearest surviving neighbor: {after:?}"
+        );
+    }
+
+    #[test]
+    fn patching_a_row_above_the_viewport_keeps_the_visual_anchor() {
+        let pane = test_pane();
+        let mut state = StructuredState::default();
+        state.follow_selection = false;
+        numbered_rows(&mut state, 20);
+        state.scroll = 10;
+
+        render_transcript_lines(&pane, &mut state, 30, 5);
+        // Simulate a `TranscriptDelta::Patch` on a collapsed tool row above the
+        // viewport expanding into a multi-line diff.
+        let index = state.rows.iter().position(|r| r.id == "r5").unwrap();
+        state.rows[index] = row(
+            "r5",
+            TranscriptRowKind::ToolDiff,
+            json!({"title":"Edit(x)","content":[{"type":"diff","path":"x","oldText":"a","newText":"a\nb\nc\nd"}]}),
+        );
+        let after = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert!(
+            after[0].contains("line 10"),
+            "a patched row above the viewport must not shift what's on screen: {after:?}"
+        );
+    }
+
+    #[test]
+    fn narrow_width_reconciles_without_panicking() {
+        let pane = test_pane();
+        let mut state = StructuredState::default();
+        state.follow_selection = false;
+        numbered_rows(&mut state, 20);
+        state.scroll = 10;
+
+        render_transcript_lines(&pane, &mut state, 1, 3);
+        grow_row(&mut state, "r3");
+        render_transcript_lines(&pane, &mut state, 1, 3);
+        assert!(state.scroll <= state.line_rows.len());
+    }
+
+    #[test]
+    fn follow_selection_without_a_selection_stays_pinned_to_the_live_edge() {
+        let pane = test_pane();
+        let mut state = StructuredState::default();
+        state.follow_selection = true;
+        state.selected = None;
+        numbered_rows(&mut state, 20);
+
+        let before = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert!(before.last().unwrap().contains("line 19"), "{before:?}");
+
+        state.rows.push(row(
+            "r20",
+            TranscriptRowKind::AgentText,
+            json!({"content":{"type":"text","text":"line 20"}}),
+        ));
+        let after = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert!(
+            after.last().unwrap().contains("line 20"),
+            "bottom-follow must track the live edge as rows are appended: {after:?}"
+        );
+    }
+
+    #[test]
+    fn follow_selection_keeps_the_selected_row_visible_when_a_row_above_it_grows() {
+        let pane = test_pane();
+        let mut state = StructuredState::default();
+        state.follow_selection = true;
+        numbered_rows(&mut state, 20);
+        state.selected = Some("r15".into());
+
+        let before = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert!(
+            before.iter().any(|line| line.contains("line 15")),
+            "{before:?}"
+        );
+
+        grow_row(&mut state, "r3");
+        let after = render_transcript_lines(&pane, &mut state, 30, 5);
+        assert!(
+            after.iter().any(|line| line.contains("line 15")),
+            "the selected row must stay visible while follow_selection is on: {after:?}"
+        );
+    }
 }

crates/maestro-tui/src/structured_task.rs

@@ -25,6 +25,12 @@ pub struct StructuredState {
     pub selected: Option<String>,
     pub expanded: HashSet<String>,
     pub scroll: usize,
+    /// Row id per rendered line from the last render pass, parallel to that
+    /// pass's `lines`. Lets the next render translate `scroll` — a raw index
+    /// into that now-stale layout — back into a row identity before mapping
+    /// it onto the freshly rebuilt layout, so growth/shrinkage above the
+    /// viewport shifts indices without shifting what's on screen.
+    pub(crate) line_rows: Vec<Option<String>>,
     pub follow_selection: bool,
     pub composer: TextInput,
     pub error: Option<String>,
@@ -42,6 +48,7 @@ impl Default for StructuredState {
             selected: None,
             expanded: HashSet::new(),
             scroll: 0,
+            line_rows: Vec::new(),
             follow_selection: true,
             composer: TextInput::default(),
             error: None,

docs/STRUCTURED-VIEW.md

@@ -162,8 +162,10 @@ choices, not daemon ones.
 **A client decides:**
 
 - How each row kind is drawn — markdown for `AgentText`/`AgentThought`, a
-  diff view for `ToolDiff`, a tool card for `ToolCall`, a checklist for
+  diff view for `ToolDiff`, a tool card for `ToolCall`, and a floor
-  `Plan`, and a floor presentation for `Notice`.
+  presentation for `Notice`. `Plan` is the exception: both clients omit it
+  from the transcript entirely and render the current plan in a separate
+  plan rail (web) / status line (TUI) instead.
 - How `responding` vs. `thinking` are distinguished visually, if at all.
 - How the operator is asked to resolve an `Approval` row — the daemon only
   exposes the three-way decision (`AllowOnce` / `AllowForSession` / `Deny`),

web/src/structured.test.ts

@@ -44,6 +44,22 @@ function row(id: string, kind: TranscriptRow["kind"], data: unknown): Transcript
   return { id, kind, data };
 }
 
+function toolRow(id: string, text: string): TranscriptRow {
+  return row(id, "tool_call", {
+    title: `Read(${id})`,
+    status: "completed",
+    content: [{ type: "content", content: { type: "text", text } }],
+  });
+}
+
+function diffRow(id: string, newText: string): TranscriptRow {
+  return row(id, "tool_diff", {
+    title: `Edit(${id})`,
+    status: "completed",
+    content: [{ type: "diff", path: "file.ts", oldText: "before", newText }],
+  });
+}
+
 const idleState: ReducedState = { turn: { responding: false, thinking: false, active_tool_calls: [] } };
 
 const allApprovalOptions = [
@@ -93,6 +109,121 @@ describe("structured connection: data flow", () => {
     conn.dispose();
   });
 
+  it("keeps snapshot plan rows out of the transcript while rendering the plan rail", () => {
+    const ws = new FakeSocket();
+    const conn = mount({ sockets: [ws] });
+    ws.open();
+    ws.emit({
+      type: "transcript_snapshot",
+      snapshot: {
+        cursor: 3,
+        rows: [
+          row("user:1", "user", { content: { type: "text", text: "before plan" } }),
+          row("plan", "plan", { entries: [{ content: "ship it", status: "pending" }] }),
+          row("agent:1", "agent_text", { content: { type: "text", text: "after plan" } }),
+        ],
+      },
+    });
+    ws.emit({
+      type: "reduced_state",
+      state: {
+        ...idleState,
+        plan: { entries: [{ content: "ship it", priority: "high", status: "pending" }] },
+      },
+    });
+
+    const transcript = conn.el.querySelector(".structured-rows")!;
+    expect(transcript.querySelectorAll(".st-row")).toHaveLength(2);
+    expect(transcript.querySelector('[data-row-id="plan"]')).toBeNull();
+    expect(transcript.textContent).toContain("before plan");
+    expect(transcript.textContent).toContain("after plan");
+    expect(transcript.textContent).not.toContain("[plan]");
+    expect(conn.el.querySelector(".structured-rail")!.textContent).toContain("ship it");
+    conn.dispose();
+  });
+
+  it("applies plan append, patch, and remove without disturbing surrounding rows", () => {
+    const ws = new FakeSocket();
+    const conn = mount({ sockets: [ws] });
+    ws.open();
+    ws.emit({
+      type: "transcript_snapshot",
+      snapshot: {
+        cursor: 1,
+        rows: [row("user:1", "user", { content: { type: "text", text: "keep me" } })],
+      },
+    });
+    ws.emit({ type: "reduced_state", state: idleState });
+
+    ws.emit({
+      type: "projection_update",
+      transcript: {
+        type: "append",
+        row: row("plan", "plan", { entries: [{ content: "first", status: "pending" }] }),
+      },
+      state: {
+        ...idleState,
+        plan: { entries: [{ content: "first", priority: "high", status: "pending" }] },
+      },
+    });
+    ws.emit({
+      type: "projection_update",
+      transcript: {
+        type: "append",
+        row: row("agent:1", "agent_text", { content: { type: "text", text: "still here" } }),
+      },
+      state: {
+        ...idleState,
+        plan: { entries: [{ content: "first", priority: "high", status: "pending" }] },
+      },
+    });
+    ws.emit({
+      type: "projection_update",
+      transcript: {
+        type: "patch",
+        row: row("plan", "plan", { entries: [{ content: "updated", status: "completed" }] }),
+      },
+      state: {
+        ...idleState,
+        plan: { entries: [{ content: "updated", priority: "high", status: "completed" }] },
+      },
+    });
+
+    const transcript = conn.el.querySelector(".structured-rows")!;
+    expect([...transcript.querySelectorAll<HTMLElement>(".st-row")].map((el) => el.dataset.rowId)).toEqual([
+      "user:1",
+      "agent:1",
+    ]);
+    expect(conn.el.querySelector(".structured-rail")!.textContent).toContain("updated");
+    expect(conn.el.querySelector(".structured-rail")!.textContent).not.toContain("first");
+
+    ws.emit({
+      type: "projection_update",
+      transcript: { type: "remove", id: "plan" },
+      state: idleState,
+    });
+    ws.emit({
+      type: "projection_update",
+      transcript: {
+        type: "patch",
+        row: row("agent:1", "agent_text", { content: { type: "text", text: "patched neighbor" } }),
+      },
+      state: idleState,
+    });
+    ws.emit({
+      type: "projection_update",
+      transcript: { type: "remove", id: "user:1" },
+      state: idleState,
+    });
+
+    expect(conn.el.querySelector(".structured-rail .st-plan")).toBeNull();
+    expect([...transcript.querySelectorAll<HTMLElement>(".st-row")].map((el) => el.dataset.rowId)).toEqual([
+      "agent:1",
+    ]);
+    expect(transcript.textContent).toContain("patched neighbor");
+    conn.dispose();
+  });
+
   it("appends and patches rows via projection_update without duplicating them", () => {
     const ws = new FakeSocket();
     const conn = mount({ sockets: [ws] });
@@ -535,6 +666,110 @@ describe("structured connection: reconnection/error/loading", () => {
     conn.dispose();
   });
 
+  it("does not mount plan rows from a reconnect snapshot", () => {
+    const first = new FakeSocket();
+    const second = new FakeSocket();
+    const conn = mount({ sockets: [first, second], shouldReconnect: () => true });
+    first.open();
+    first.emit({
+      type: "transcript_snapshot",
+      snapshot: {
+        cursor: 1,
+        rows: [row("agent:old", "agent_text", { content: { type: "text", text: "old" } })],
+      },
+    });
+
+    first.drop();
+    vi.advanceTimersByTime(1000);
+    second.open();
+    second.emit({
+      type: "transcript_snapshot",
+      snapshot: {
+        cursor: 4,
+        rows: [
+          row("user:new", "user", { content: { type: "text", text: "new" } }),
+          row("plan", "plan", { entries: [{ content: "reconnected", status: "in_progress" }] }),
+        ],
+      },
+    });
+    second.emit({
+      type: "reduced_state",
+      state: {
+        ...idleState,
+        plan: { entries: [{ content: "reconnected", priority: "high", status: "in_progress" }] },
+      },
+    });
+
+    const transcript = conn.el.querySelector(".structured-rows")!;
+    expect(transcript.textContent).toBe("Younew");
+    expect(transcript.querySelector('[data-row-id="plan"]')).toBeNull();
+    expect(conn.el.querySelector(".structured-rail")!.textContent).toContain("reconnected");
+    conn.dispose();
+  });
+
+  it("keeps surviving expanded rows open across a reconnect snapshot and leaves fresh rows collapsed", () => {
+    const first = new FakeSocket();
+    const second = new FakeSocket();
+    const conn = mount({ sockets: [first, second], shouldReconnect: () => true });
+    first.open();
+    first.emit({
+      type: "transcript_snapshot",
+      snapshot: { cursor: 2, rows: [toolRow("tool:1", "before"), diffRow("diff:1", "first")] },
+    });
+    for (const head of conn.el.querySelectorAll<HTMLButtonElement>(".st-tool-head")) head.click();
+
+    first.drop();
+    vi.advanceTimersByTime(1000);
+    second.open();
+    second.emit({
+      type: "transcript_snapshot",
+      snapshot: {
+        cursor: 3,
+        rows: [toolRow("tool:1", "after"), diffRow("diff:1", "second"), toolRow("tool:2", "new")],
+      },
+    });
+
+    expect(
+      conn.el.querySelector('[data-row-id="tool:1"] .st-tool-head')?.getAttribute("aria-expanded"),
+    ).toBe("true");
+    expect(
+      conn.el.querySelector('[data-row-id="diff:1"] .st-tool-head')?.getAttribute("aria-expanded"),
+    ).toBe("true");
+    expect(
+      conn.el.querySelector('[data-row-id="tool:2"] .st-tool-head')?.getAttribute("aria-expanded"),
+    ).toBe("false");
+    expect(conn.el.querySelector('[data-row-id="tool:1"] .st-tool-out')?.textContent).toContain("after");
+    conn.dispose();
+  });
+
+  it("prunes expanded IDs removed or made non-expandable by a snapshot", () => {
+    const ws = new FakeSocket();
+    const conn = mount({ sockets: [ws] });
+    ws.open();
+    ws.emit({
+      type: "transcript_snapshot",
+      snapshot: { cursor: 2, rows: [toolRow("tool:gone", "gone"), toolRow("tool:plain", "plain")] },
+    });
+    for (const head of conn.el.querySelectorAll<HTMLButtonElement>(".st-tool-head")) head.click();
+
+    ws.emit({
+      type: "transcript_snapshot",
+      snapshot: {
+        cursor: 3,
+        rows: [row("tool:plain", "agent_text", { content: { type: "text", text: "not a tool" } })],
+      },
+    });
+    ws.emit({
+      type: "transcript_snapshot",
+      snapshot: { cursor: 5, rows: [toolRow("tool:gone", "returned"), toolRow("tool:plain", "returned")] },
+    });
+
+    for (const head of conn.el.querySelectorAll<HTMLButtonElement>(".st-tool-head")) {
+      expect(head.getAttribute("aria-expanded")).toBe("false");
+    }
+    conn.dispose();
+  });
+
   it("keeps a local submission gated across a reconnect until begin-turn arrives", () => {
     vi.mocked(sendStructuredPrompt).mockImplementationOnce(() => new Promise(() => {}));
     const first = new FakeSocket();
@@ -697,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;
@@ -106,6 +107,7 @@ export function createStructuredConnection(
   let rows: TranscriptRow[] = [];
   const rowIndex = new Map<string, number>();
   const rowEls = new Map<string, HTMLElement>();
+  const expandedRowIds = new Set<string>();
   let state: ReducedState = { turn: idleTurn() };
   let pendingApprovalRowId: string | null = null;
   let latestSession: Session | undefined = store?.snapshot().sessions.get(sessionId);
@@ -116,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;
@@ -151,59 +212,106 @@ export function createStructuredConnection(
     }
   }
 
-  function mountRow(row: TranscriptRow): void {
+  function buildRowElement(row: TranscriptRow): HTMLElement | null {
-    const el = buildTranscriptRow(row, approvalStateFor(row));
+    const el = buildTranscriptRow(
+      row,
+      approvalStateFor(row),
+      expandedRowIds.has(row.id),
+      (expanded) => {
+        if (expanded) expandedRowIds.add(row.id);
+        else expandedRowIds.delete(row.id);
+        finishTranscriptLayout();
+      },
+    );
+    if (!el) return null;
+    if (el.querySelector(".st-tool-head")?.getAttribute("aria-expanded") !== "true") {
+      expandedRowIds.delete(row.id);
+    }
+    return el;
+  }
+
+  function insertRowElement(row: TranscriptRow, el: HTMLElement): void {
+    const index = rowIndex.get(row.id) ?? rows.length - 1;
+    let next: HTMLElement | null = null;
+    for (let i = index + 1; i < rows.length; i++) {
+      next = rowEls.get(rows[i].id) ?? null;
+      if (next) break;
+    }
     rowEls.set(row.id, el);
-    rowsEl.appendChild(el);
+    rowsEl.insertBefore(el, next);
+  }
+
+  function mountRow(row: TranscriptRow): void {
+    const el = buildRowElement(row);
+    if (el) insertRowElement(row, el);
   }
 
   function remountRow(row: TranscriptRow): void {
     const old = rowEls.get(row.id);
     const oldToolHead = old?.querySelector<HTMLButtonElement>(".st-tool-head") ?? null;
-    const wasExpanded = oldToolHead?.getAttribute("aria-expanded") === "true";
     const restoreFocus = document.activeElement === oldToolHead;
-    const el = buildTranscriptRow(row, approvalStateFor(row), wasExpanded);
+    const el = buildRowElement(row);
-    rowEls.set(row.id, el);
+    if (!el) {
-    if (old) old.replaceWith(el);
+      old?.remove();
-    else rowsEl.appendChild(el);
+      rowEls.delete(row.id);
+      expandedRowIds.delete(row.id);
+      return;
+    }
+    if (old) {
+      rowEls.set(row.id, el);
+      old.replaceWith(el);
+    } else {
+      insertRowElement(row, el);
+    }
     const newToolHead = el.querySelector<HTMLButtonElement>(".st-tool-head");
     if (restoreFocus) newToolHead?.focus();
   }
 
   function applySnapshot(snapshot: { rows: TranscriptRow[] }): void {
-    rows = [...snapshot.rows];
+    const forceBottom = !receivedSnapshot;
-    rowIndex.clear();
+    updateTranscriptLayout(() => {
-    rows.forEach((row, i) => rowIndex.set(row.id, i));
+      rows = [...snapshot.rows];
-    rowEls.clear();
+      const expandableIds = new Set(
-    rowsEl.replaceChildren();
+        rows.filter((row) => row.kind === "tool_call" || row.kind === "tool_diff").map((row) => row.id),
-    for (const row of rows) mountRow(row);
+      );
+      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") {
+    updateTranscriptLayout(() => {
-      rowIndex.set(delta.row.id, rows.length);
+      if (delta.type === "append") {
-      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 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;
+        const i = rowIndex.get(delta.id);
-        remountRow(delta.row);
+        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);
-      rowIndex.clear();
-      rows.forEach((row, idx) => rowIndex.set(row.id, idx));
-      rowEls.get(delta.id)?.remove();
-      rowEls.delete(delta.id);
-    }
   }
 
   function toolRowCount(): number {
@@ -252,10 +360,12 @@ export function createStructuredConnection(
     reconcileAwaitingBegin();
     const newPendingId = state.pending_approval ? lastRowIdOfKind("approval") : null;
     if (newPendingId !== pendingApprovalRowId) {
-      const prev = pendingApprovalRowId;
+      updateTranscriptLayout(() => {
-      pendingApprovalRowId = newPendingId;
+        const prev = pendingApprovalRowId;
-      if (prev !== null && rowIndex.has(prev)) remountRow(rows[rowIndex.get(prev)!]);
+        pendingApprovalRowId = newPendingId;
-      if (newPendingId !== null) remountRow(rows[rowIndex.get(newPendingId)!]);
+        if (prev !== null && rowIndex.has(prev)) remountRow(rows[rowIndex.get(prev)!]);
+        if (newPendingId !== null) remountRow(rows[rowIndex.get(newPendingId)!]);
+      });
     }
     renderPills();
     renderRailNow();
@@ -279,8 +389,10 @@ export function createStructuredConnection(
 
   // ---- composer ----
   function autoResize(): void {
-    input.style.height = "auto";
+    updateTranscriptLayout(() => {
-    input.style.height = `${Math.min(input.scrollHeight, 160)}px`;
+      input.style.height = "auto";
+      input.style.height = `${Math.min(input.scrollHeight, 160)}px`;
+    });
   }
   input.addEventListener("input", autoResize);
   input.addEventListener("keydown", (e) => {
@@ -336,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);
 
@@ -452,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,
@@ -459,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();
@@ -478,10 +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();
     },
   };

web/src/structuredRender.test.ts

@@ -9,7 +9,11 @@ import {
 } from "./structuredRender.js";
 import type { Session, TranscriptRow } from "./types.js";
 
-function row(kind: TranscriptRow["kind"], data: unknown, id = "r1"): TranscriptRow {
+function row<K extends TranscriptRow["kind"]>(
+  kind: K,
+  data: unknown,
+  id = "r1",
+): TranscriptRow & { kind: K } {
   return { id, kind, data };
 }
 
@@ -275,9 +279,16 @@ describe("buildTranscriptRow: approval", () => {
 });
 
 describe("buildTranscriptRow: notice / unknown kinds", () => {
-  it("renders a plan or notice row without crashing", () => {
+  it("renders notice and unknown row kinds safely", () => {
-    expect(() => buildTranscriptRow(row("notice", { sessionUpdate: "future" }))).not.toThrow();
+    expect(buildTranscriptRow(row("notice", { sessionUpdate: "future" }))?.textContent).toBe(
-    expect(() => buildTranscriptRow(row("plan", { entries: [] }))).not.toThrow();
+      "[notice]",
+    );
+    const unknown = { id: "r1", kind: "future_kind", data: {} } as unknown as TranscriptRow;
+    expect(buildTranscriptRow(unknown)?.textContent).toBe("[future_kind]");
+  });
+
+  it("returns no transcript element for a plan row", () => {
+    expect(buildTranscriptRow(row("plan", { entries: [] }))).toBeNull();
   });
 });
 

web/src/structuredRender.ts

@@ -187,7 +187,11 @@ function statusMeta(status: string | undefined, hasDiff: boolean): { text: strin
   return null;
 }
 
-function buildToolRow(row: TranscriptRow, expanded: boolean): HTMLElement {
+function buildToolRow(
+  row: TranscriptRow,
+  expanded: boolean,
+  onExpandedChange?: (expanded: boolean) => void,
+): HTMLElement {
   const tool = (row.data ?? {}) as AcpToolCallLike;
   const content = tool.content ?? [];
   const hasDiff = content.some((c) => c.type === "diff");
@@ -259,6 +263,7 @@ function buildToolRow(row: TranscriptRow, expanded: boolean): HTMLElement {
   };
   head.addEventListener("click", () => {
     setExpanded(head.getAttribute("aria-expanded") !== "true");
+    onExpandedChange?.(head.getAttribute("aria-expanded") === "true");
   });
   setExpanded(expanded);
 
@@ -362,13 +367,33 @@ function buildApprovalRow(row: TranscriptRow, approval: ApprovalRowState): HTMLE
   return wrap;
 }
 
-/** Build the DOM for one transcript row. `approval` is consulted only for an
+/** Build the DOM for one transcript row, or `null` for a plan shown only in
- *  `approval` row; pass it whenever the row might be that kind. */
+ *  the rail. `approval` is consulted only for an `approval` row; tool
+ *  expansion is supplied and reported explicitly. */
+export function buildTranscriptRow(
+  row: TranscriptRow & { kind: "plan" },
+  approval?: ApprovalRowState,
+  toolExpanded?: boolean,
+  onToolExpandedChange?: (expanded: boolean) => void,
+): null;
+export function buildTranscriptRow(
+  row: TranscriptRow & { kind: Exclude<TranscriptRow["kind"], "plan"> },
+  approval?: ApprovalRowState,
+  toolExpanded?: boolean,
+  onToolExpandedChange?: (expanded: boolean) => void,
+): HTMLElement;
+export function buildTranscriptRow(
+  row: TranscriptRow,
+  approval?: ApprovalRowState,
+  toolExpanded?: boolean,
+  onToolExpandedChange?: (expanded: boolean) => void,
+): HTMLElement | null;
 export function buildTranscriptRow(
   row: TranscriptRow,
   approval?: ApprovalRowState,
   toolExpanded = false,
-): HTMLElement {
+  onToolExpandedChange?: (expanded: boolean) => void,
+): HTMLElement | null {
   let built: HTMLElement;
   switch (row.kind) {
     case "user":
@@ -382,11 +407,13 @@ export function buildTranscriptRow(
       break;
     case "tool_call":
     case "tool_diff":
-      built = buildToolRow(row, toolExpanded);
+      built = buildToolRow(row, toolExpanded, onToolExpandedChange);
       break;
     case "approval":
       built = buildApprovalRow(row, approval ?? { pending: false, onDecide: () => Promise.resolve() });
       break;
+    case "plan":
+      return null;
     default:
       built = buildPlainRow(row);
       break;