fix(daemon): reject dead structured sessions

Serialize replay reads with event appends, acquire the live receiver under that lock, and drain the event pump during shutdown. Recovery now removes its FIFO when opening the log fails.

Cargo.toml

@@ -10,6 +10,7 @@ repository = "https://git.naps.pt/naps62/maestro"
 publish = false
 
 [workspace.dependencies]
+agent-client-protocol = { version = "2.1", features = ["unstable_tool_call_name"] }
 axum = { version = "0.7", features = ["ws"] }
 chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] }
 clap = { version = "4.5", features = ["derive", "env"] }
@@ -17,6 +18,7 @@ crossterm = { version = "0.28", features = ["event-stream"] }
 dirs = "5"
 futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
 libc = "0.2"
+maestro-acp = { path = "crates/maestro-acp" }
 maestro-config = { path = "crates/maestro-config" }
 # `default-features = false` so each consumer opts into `haiku-autoname`
 # through its own feature of the same name (#131); a workspace entry that
@@ -38,6 +40,7 @@ tempfile = "3"
 thiserror = "1"
 tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time", "net", "io-util", "fs", "signal", "process"] }
 tokio-tungstenite = "0.21"
+tokio-util = { version = "0.7", features = ["compat"] }
 toml = "0.8"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }

crates/maestro-acp/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "maestro-acp"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+publish.workspace = true
+description = "ACP client role: one claude-code-acp child per structured session"
+
+[dependencies]
+agent-client-protocol.workspace = true
+libc.workspace = true
+serde_json.workspace = true
+thiserror.workspace = true
+tokio.workspace = true
+tokio-util.workspace = true
+tracing.workspace = true
+
+[dev-dependencies]
+tempfile.workspace = true

crates/maestro-acp/src/lib.rs

@@ -0,0 +1,632 @@
+//! ACP client role for the daemon (#134): one `claude-code-acp` child per
+//! structured session.
+//!
+//! The daemon is the client; the adapter is the agent. [`spawn`] starts the
+//! adapter as a child process with stdio pipes, in the session's `cwd` with
+//! the session's env, runs the `initialize` + `session/new` handshake, and
+//! hands back an [`AcpSession`] that owns the connection for the life of the
+//! session.
+//!
+//! **The handshake declares almost nothing.** `initialize` reports
+//! `fs.readTextFile`, `fs.writeTextFile` and `terminal` as `false`, so the
+//! adapter uses Claude Code's own Read / Write / Bash instead of asking the
+//! daemon to act on its behalf — exactly what already happens in the
+//! terminal view. `session/request_permission` is the one client method that
+//! gets a real implementation, in #139; here it is a stub that declines.
+//!
+//! Every `session/update` notification the adapter sends lands on an
+//! unbounded channel the session owns ([`AcpSession::take_updates`]). Nothing
+//! is dropped while nobody reads: #135 takes the receiver and persists each
+//! notification to the per-session event log before broadcasting it.
+//!
+//! A respawn after a daemon restart sets [`SpawnConfig::load_session`] and
+//! the handshake runs `session/load` with the stored id instead of
+//! `session/new`. The history the agent streams back while that request is
+//! in flight is *not* delivered: the daemon's event log already holds it,
+//! and replaying it would double every line.
+//!
+//! The child is a child of the daemon, full stop. It shares the daemon's
+//! process group, is marked to receive `SIGTERM` when the daemon dies
+//! (`PR_SET_PDEATHSIG` on Linux), and [`AcpSession::terminate`] is what both
+//! `DELETE /api/sessions/:id` and daemon shutdown call: `session/cancel`, a
+//! short wait, then `SIGTERM`, then `SIGKILL` as the last resort.
+
+#![deny(missing_debug_implementations)]
+
+use std::collections::{HashMap, VecDeque};
+use std::ffi::OsString;
+use std::path::{Path, PathBuf};
+use std::process::{ExitStatus, Stdio};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use agent_client_protocol::schema::v1::{
+    CancelNotification, ClientCapabilities, ContentBlock, FileSystemCapabilities, Implementation,
+    InitializeRequest, LoadSessionRequest, NewSessionRequest, PromptRequest,
+    RequestPermissionOutcome, RequestPermissionRequest, RequestPermissionResponse, SessionId,
+    SessionNotification,
+};
+use agent_client_protocol::schema::ProtocolVersion;
+use agent_client_protocol::{ByteStreams, Client};
+use tokio::io::AsyncBufReadExt;
+use tokio::sync::{mpsc, oneshot, watch};
+use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
+
+pub use agent_client_protocol::schema::v1 as schema;
+
+/// The adapter binary, resolved on `PATH`.
+pub const ADAPTER_BIN: &str = "claude-code-acp";
+
+/// How to install the adapter when it is missing. Named in the error so an
+/// operator staring at a failed session knows what to run.
+pub const INSTALL_COMMAND: &str =
+    "npm install -g --prefix ~/.local @zed-industries/claude-code-acp";
+
+/// How long `initialize` + `session/new` may take before the spawn is
+/// declared failed. The real adapter boots a node process and the Claude
+/// Code SDK behind it; a slow machine takes seconds, never a minute.
+pub const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(60);
+
+/// How long [`AcpSession::terminate`] waits after `session/cancel` before
+/// sending `SIGTERM`. Cancel is a courtesy that lets the adapter finish the
+/// turn it is on; the signal is what actually ends it.
+pub const CANCEL_GRACE: Duration = Duration::from_millis(500);
+
+/// How long [`AcpSession::terminate`] waits after `SIGTERM` before `SIGKILL`.
+pub const TERM_GRACE: Duration = Duration::from_secs(2);
+
+/// Bytes of the child's stderr kept for the failure message.
+const STDERR_TAIL_BYTES: usize = 4096;
+
+/// What [`spawn`] needs to start one adapter.
+#[derive(Debug, Clone)]
+pub struct SpawnConfig {
+    /// Working directory of the child and `cwd` of `session/new`.
+    pub cwd: PathBuf,
+    /// Environment laid over the daemon's own for the child. A `PATH` in
+    /// here is also what the adapter is resolved against, since that is the
+    /// `PATH` the child would see.
+    pub env: HashMap<String, String>,
+    /// Upper bound on the handshake; [`HANDSHAKE_TIMEOUT`] by default.
+    pub handshake_timeout: Duration,
+    /// An existing ACP session id to resume with `session/load` (#135)
+    /// instead of opening a fresh one. The adapter must advertise
+    /// `loadSession`; the handshake fails otherwise.
+    pub load_session: Option<String>,
+}
+
+impl SpawnConfig {
+    pub fn new(cwd: impl Into<PathBuf>, env: HashMap<String, String>) -> Self {
+        Self {
+            cwd: cwd.into(),
+            env,
+            handshake_timeout: HANDSHAKE_TIMEOUT,
+            load_session: None,
+        }
+    }
+
+    /// Resume `session_id` with `session/load` instead of `session/new`.
+    pub fn load_session(mut self, session_id: impl Into<String>) -> Self {
+        self.load_session = Some(session_id.into());
+        self
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum AcpError {
+    /// The adapter is not on `PATH`. The message names the install command
+    /// so the failed session is actionable rather than silently dead.
+    #[error("{ADAPTER_BIN} not found on PATH; install it with: {INSTALL_COMMAND}")]
+    AdapterMissing,
+    #[error("failed to start {ADAPTER_BIN}: {0}")]
+    Spawn(#[source] std::io::Error),
+    /// The adapter answered the handshake with an error.
+    #[error("ACP handshake failed: {0}")]
+    Handshake(String),
+    /// The adapter went away before the handshake finished. Carries the
+    /// tail of its stderr, which is where a missing Claude login shows up.
+    #[error("{ADAPTER_BIN} exited during the ACP handshake ({status}){stderr}")]
+    ExitedEarly { status: String, stderr: String },
+    #[error("ACP handshake timed out after {0:?}")]
+    HandshakeTimeout(Duration),
+}
+
+/// Resolve [`ADAPTER_BIN`] against the `PATH` the child would run with: the
+/// session env's `PATH` when it sets one, the daemon's own otherwise.
+pub fn resolve_adapter(env: &HashMap<String, String>) -> Option<PathBuf> {
+    let path: OsString = match env.get("PATH") {
+        Some(p) => OsString::from(p),
+        None => std::env::var_os("PATH")?,
+    };
+    std::env::split_paths(&path)
+        .filter(|dir| !dir.as_os_str().is_empty())
+        .map(|dir| dir.join(ADAPTER_BIN))
+        .find(|candidate| is_executable(candidate))
+}
+
+fn is_executable(path: &Path) -> bool {
+    use std::os::unix::fs::PermissionsExt;
+    std::fs::metadata(path)
+        .map(|m| m.is_file() && m.permissions().mode() & 0o111 != 0)
+        .unwrap_or(false)
+}
+
+enum Command {
+    Cancel,
+    Prompt {
+        text: String,
+        done: oneshot::Sender<Result<(), String>>,
+    },
+    Close,
+}
+
+enum ChildCommand {
+    Signal(libc::c_int),
+    Reap(oneshot::Sender<bool>),
+}
+
+/// Why the handshake did not produce a session id.
+enum ReadyError {
+    /// The adapter answered with an error.
+    Handshake(String),
+    /// The transport closed under the request: the child is gone.
+    Closed,
+}
+
+fn ready_error(e: agent_client_protocol::Error) -> ReadyError {
+    if agent_client_protocol::is_incoming_transport_closed(&e) {
+        ReadyError::Closed
+    } else {
+        ReadyError::Handshake(e.to_string())
+    }
+}
+
+/// A live ACP connection to one adapter child. Dropping it sends `SIGTERM`
+/// if the child is still running; [`AcpSession::terminate`] is the polite
+/// version.
+#[derive(Debug)]
+pub struct AcpSession {
+    session_id: String,
+    pid: u32,
+    commands: mpsc::UnboundedSender<Command>,
+    child_commands: mpsc::UnboundedSender<ChildCommand>,
+    updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
+    exit: watch::Receiver<Option<ExitStatus>>,
+}
+
+impl AcpSession {
+    /// The id `session/new` returned. What a respawn passes to
+    /// `session/load` (#135).
+    pub fn session_id(&self) -> &str {
+        &self.session_id
+    }
+
+    /// The adapter's pid.
+    pub fn pid(&self) -> u32 {
+        self.pid
+    }
+
+    /// Whether the child has been reaped.
+    pub fn has_exited(&self) -> bool {
+        self.exit.borrow().is_some()
+    }
+
+    /// Reap the child if it has already exited.
+    pub async fn reap_if_exited(&self) -> bool {
+        if self.has_exited() {
+            return true;
+        }
+        let (sent, received) = oneshot::channel();
+        if self.child_commands.send(ChildCommand::Reap(sent)).is_err() {
+            return self.has_exited();
+        }
+        received.await.unwrap_or_else(|_| self.has_exited())
+    }
+
+    /// Wait for the child to exit.
+    pub async fn wait(&self) -> Option<ExitStatus> {
+        wait_exit(&self.exit).await
+    }
+
+    /// The stream of raw `session/update` notifications, handed out once.
+    /// Notifications that arrived before the take are still queued on it.
+    pub fn take_updates(&self) -> Option<mpsc::UnboundedReceiver<SessionNotification>> {
+        self.updates.lock().expect("updates mutex").take()
+    }
+
+    /// Send `session/cancel`. Fire-and-forget: a closed connection means the
+    /// child is already gone, which is the outcome cancel was after.
+    pub fn cancel(&self) {
+        let _ = self.commands.send(Command::Cancel);
+    }
+
+    /// Send a user prompt and receive completion once the agent ends the turn.
+    pub fn prompt(&self, text: String) -> Result<oneshot::Receiver<Result<(), String>>, String> {
+        let (done_tx, done_rx) = oneshot::channel();
+        self.commands
+            .send(Command::Prompt {
+                text,
+                done: done_tx,
+            })
+            .map_err(|_| "ACP connection is closed".to_string())?;
+        Ok(done_rx)
+    }
+
+    /// End the child: `session/cancel`, wait `cancel_grace`, `SIGTERM`, wait
+    /// `term_grace`, `SIGKILL`. Returns once the child has been reaped.
+    pub async fn terminate(&self, cancel_grace: Duration, term_grace: Duration) {
+        self.cancel();
+        if tokio::time::timeout(cancel_grace, self.wait())
+            .await
+            .is_ok()
+        {
+            return;
+        }
+        self.signal(libc::SIGTERM);
+        if tokio::time::timeout(term_grace, self.wait()).await.is_ok() {
+            return;
+        }
+        tracing::warn!(
+            pid = self.pid,
+            "{ADAPTER_BIN} ignored SIGTERM; sending SIGKILL"
+        );
+        self.signal(libc::SIGKILL);
+        let _ = self.commands.send(Command::Close);
+        self.wait().await;
+    }
+
+    fn signal(&self, sig: libc::c_int) {
+        let _ = self.child_commands.send(ChildCommand::Signal(sig));
+    }
+}
+
+impl Drop for AcpSession {
+    fn drop(&mut self) {
+        self.signal(libc::SIGTERM);
+        let _ = self.commands.send(Command::Close);
+    }
+}
+
+/// Start the adapter for one session and run the handshake.
+///
+/// Fails without a running child: on any error the process is already
+/// signalled, so a failed session never leaves a stray adapter behind.
+pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
+    let adapter = resolve_adapter(&config.env).ok_or(AcpError::AdapterMissing)?;
+
+    let mut cmd = tokio::process::Command::new(&adapter);
+    cmd.current_dir(&config.cwd)
+        .envs(&config.env)
+        // Claude Code refuses to start when it thinks it is nested inside
+        // another Claude Code session; a daemon launched from one would
+        // otherwise never get a structured session up.
+        .env_remove("CLAUDECODE")
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        // If the daemon is dropped without a graceful shutdown the runtime
+        // still tears the task holding this `Child` down, and that kills it.
+        .kill_on_drop(true);
+    #[cfg(target_os = "linux")]
+    {
+        // Same process group as the daemon (no setpgid), and SIGTERM when the
+        // daemon dies however it dies: an agent that outlives its daemon is
+        // exactly the orphan #134 rules out.
+        // SAFETY: prctl is async-signal-safe and touches no shared state.
+        unsafe {
+            cmd.pre_exec(|| {
+                libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM);
+                Ok(())
+            });
+        }
+    }
+    let mut child = cmd.spawn().map_err(AcpError::Spawn)?;
+    let pid = child.id().expect("child pid is known before wait");
+    let stdin = child.stdin.take().expect("piped stdin");
+    let stdout = child.stdout.take().expect("piped stdout");
+    let stderr = child.stderr.take().expect("piped stderr");
+
+    let stderr_tail = Arc::new(Mutex::new(VecDeque::<u8>::new()));
+    tokio::spawn(drain_stderr(stderr, pid, stderr_tail.clone()));
+
+    let (exit_tx, exit_rx) = watch::channel(None);
+    let (child_commands_tx, child_commands_rx) = mpsc::unbounded_channel();
+    tokio::spawn(supervise_child(child, pid, child_commands_rx, exit_tx));
+
+    let (updates_tx, updates_rx) = mpsc::unbounded_channel();
+    let (commands_tx, commands_rx) = mpsc::unbounded_channel();
+    let (ready_tx, ready_rx) = oneshot::channel();
+    let transport = ByteStreams::new(stdin.compat_write(), stdout.compat());
+    tokio::spawn(run_connection(
+        transport,
+        config.cwd.clone(),
+        config.load_session.clone(),
+        updates_tx,
+        commands_rx,
+        ready_tx,
+    ));
+
+    let outcome = tokio::time::timeout(config.handshake_timeout, ready_rx).await;
+    let failure = match outcome {
+        Ok(Ok(Ok(session_id))) => {
+            return Ok(AcpSession {
+                session_id,
+                pid,
+                commands: commands_tx,
+                child_commands: child_commands_tx,
+                updates: Mutex::new(Some(updates_rx)),
+                exit: exit_rx,
+            });
+        }
+        Ok(Ok(Err(ReadyError::Handshake(message)))) => AcpError::Handshake(message),
+        Ok(Ok(Err(ReadyError::Closed))) | Ok(Err(_)) => {
+            // The connection ended before the handshake did: the child went
+            // away. Reap it so the status is real, then report.
+            let status = wait_exit(&exit_rx).await;
+            let stderr = {
+                let mut tail = stderr_tail.lock().expect("stderr tail");
+                tail.make_contiguous();
+                String::from_utf8_lossy(tail.as_slices().0)
+                    .trim()
+                    .to_string()
+            };
+            AcpError::ExitedEarly {
+                status: status.map_or_else(|| "unknown status".into(), |s| s.to_string()),
+                stderr: if stderr.is_empty() {
+                    String::new()
+                } else {
+                    format!(": {stderr}")
+                },
+            }
+        }
+        Err(_) => AcpError::HandshakeTimeout(config.handshake_timeout),
+    };
+    // No half-started adapter survives a failed spawn.
+    let _ = commands_tx.send(Command::Close);
+    let _ = child_commands_tx.send(ChildCommand::Signal(libc::SIGTERM));
+    if tokio::time::timeout(TERM_GRACE, wait_exit(&exit_rx))
+        .await
+        .is_err()
+    {
+        let _ = child_commands_tx.send(ChildCommand::Signal(libc::SIGKILL));
+        wait_exit(&exit_rx).await;
+    }
+    Err(failure)
+}
+
+async fn wait_exit(exit: &watch::Receiver<Option<ExitStatus>>) -> Option<ExitStatus> {
+    let mut exit = exit.clone();
+    let status = exit.wait_for(|s| s.is_some()).await.ok()?;
+    *status
+}
+
+async fn supervise_child(
+    mut child: tokio::process::Child,
+    pid: u32,
+    mut commands: mpsc::UnboundedReceiver<ChildCommand>,
+    exit_tx: watch::Sender<Option<ExitStatus>>,
+) {
+    loop {
+        tokio::select! {
+            status = child.wait() => {
+                let status = status.ok();
+                tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
+                let _ = exit_tx.send(status);
+                return;
+            }
+            command = commands.recv() => match command {
+                Some(ChildCommand::Signal(sig)) => {
+                    // This task alone owns `child`. It cannot reap the child
+                    // between receiving this command and sending the signal,
+                    // so its PID cannot be reused in that window.
+                    unsafe { libc::kill(pid as libc::pid_t, sig) };
+                }
+                Some(ChildCommand::Reap(done)) => match child.try_wait() {
+                    Ok(Some(status)) => {
+                        let _ = done.send(true);
+                        tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
+                        let _ = exit_tx.send(Some(status));
+                        return;
+                    }
+                    Ok(None) => {
+                        let _ = done.send(false);
+                    }
+                    Err(error) => {
+                        tracing::warn!(pid, %error, "could not reap {ADAPTER_BIN}");
+                        let _ = done.send(false);
+                    }
+                },
+                None => {
+                    // The session vanished without a command receiver. This
+                    // task still owns the unreaped child, so this cannot hit
+                    // a reused PID.
+                    unsafe { libc::kill(pid as libc::pid_t, libc::SIGTERM) };
+                    let status = child.wait().await.ok();
+                    tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
+                    let _ = exit_tx.send(status);
+                    return;
+                }
+            },
+        }
+    }
+}
+
+/// The capabilities the daemon declares: none of the filesystem or terminal
+/// surface, so the adapter falls back to Claude Code's own tools.
+fn client_capabilities() -> ClientCapabilities {
+    ClientCapabilities::new()
+        .fs(FileSystemCapabilities::new()
+            .read_text_file(false)
+            .write_text_file(false))
+        .terminal(false)
+}
+
+async fn run_connection<T>(
+    transport: T,
+    cwd: PathBuf,
+    load_session: Option<String>,
+    updates_tx: mpsc::UnboundedSender<SessionNotification>,
+    mut commands_rx: mpsc::UnboundedReceiver<Command>,
+    ready_tx: oneshot::Sender<Result<String, ReadyError>>,
+) where
+    T: agent_client_protocol::ConnectTo<Client> + 'static,
+{
+    // Set while `session/load` is in flight. Notification handlers run in
+    // the dispatch loop, in arrival order, so everything the agent streams
+    // before its load response sees the flag up and is dropped: that is the
+    // history the event log already holds (#135).
+    let replaying = Arc::new(AtomicBool::new(false));
+    let replaying_seen = replaying.clone();
+    let result = Client
+        .builder()
+        .name("maestro")
+        .on_receive_notification(
+            async move |notification: SessionNotification, _cx| {
+                if replaying_seen.load(Ordering::SeqCst) {
+                    tracing::trace!("dropping replayed notification during session/load");
+                    return Ok(());
+                }
+                // A dropped receiver only means nobody wants updates any more.
+                let _ = updates_tx.send(notification);
+                Ok(())
+            },
+            agent_client_protocol::on_receive_notification!(),
+        )
+        .on_receive_request(
+            async move |request: RequestPermissionRequest, responder, _cx| {
+                // Stub until #139: decline rather than grant something the
+                // operator never saw.
+                tracing::warn!(
+                    session = %request.session_id,
+                    "session/request_permission declined: not implemented yet (#139)"
+                );
+                responder.respond(RequestPermissionResponse::new(
+                    RequestPermissionOutcome::Cancelled,
+                ))
+            },
+            agent_client_protocol::on_receive_request!(),
+        )
+        .connect_with(transport, async move |cx| {
+            let init = InitializeRequest::new(ProtocolVersion::V1)
+                .client_capabilities(client_capabilities())
+                .client_info(Implementation::new("maestro", env!("CARGO_PKG_VERSION")));
+            let initialized = match cx.send_request(init).block_task().await {
+                Ok(r) => r,
+                Err(e) => {
+                    let _ = ready_tx.send(Err(ready_error(e)));
+                    return Ok(());
+                }
+            };
+            let session_id = match load_session {
+                Some(existing) => {
+                    if !initialized.agent_capabilities.load_session {
+                        let _ = ready_tx.send(Err(ReadyError::Handshake(format!(
+                            "{ADAPTER_BIN} does not advertise loadSession; cannot resume session {existing}"
+                        ))));
+                        return Ok(());
+                    }
+                    replaying.store(true, Ordering::SeqCst);
+                    let (loaded_tx, loaded_rx) = oneshot::channel();
+                    let replaying_done = replaying.clone();
+                    if let Err(e) = cx
+                        .send_request(LoadSessionRequest::new(existing.clone(), cwd))
+                        .on_receiving_result(move |result| {
+                            replaying_done.store(false, Ordering::SeqCst);
+                            async move {
+                                let _ = loaded_tx.send(result);
+                                Ok(())
+                            }
+                        })
+                    {
+                        let _ = ready_tx.send(Err(ready_error(e)));
+                        return Ok(());
+                    }
+                    match loaded_rx.await {
+                        Ok(Ok(_)) => {}
+                        Ok(Err(e)) => {
+                            let _ = ready_tx.send(Err(ready_error(e)));
+                            return Ok(());
+                        }
+                        Err(_) => {
+                            let _ = ready_tx.send(Err(ReadyError::Handshake(
+                                "session/load response was not delivered".into(),
+                            )));
+                            return Ok(());
+                        }
+                    }
+                    existing
+                }
+                None => match cx
+                    .send_request(NewSessionRequest::new(cwd))
+                    .block_task()
+                    .await
+                {
+                    Ok(created) => created.session_id.0.to_string(),
+                    Err(e) => {
+                        let _ = ready_tx.send(Err(ready_error(e)));
+                        return Ok(());
+                    }
+                },
+            };
+            let _ = ready_tx.send(Ok(session_id.clone()));
+
+            loop {
+                tokio::select! {
+                    cmd = commands_rx.recv() => match cmd {
+                        Some(Command::Cancel) => {
+                            let _ = cx.send_notification(CancelNotification::new(
+                                SessionId::new(session_id.as_str()),
+                            ));
+                        }
+                        Some(Command::Prompt { text, done }) => {
+                            let request = PromptRequest::new(
+                                SessionId::new(session_id.as_str()),
+                                vec![ContentBlock::from(text)],
+                            );
+                            let done = Arc::new(Mutex::new(Some(done)));
+                            let callback_done = done.clone();
+                            if let Err(error) = cx.send_request(request).on_receiving_result(move |result| {
+                                let callback_done = callback_done.clone();
+                                async move {
+                                    if let Some(done) = callback_done.lock().expect("prompt sender").take() {
+                                        let _ = done.send(result.map(|_| ()).map_err(|e| e.to_string()));
+                                    }
+                                    Ok(())
+                                }
+                            }) {
+                                if let Some(done) = done.lock().expect("prompt sender").take() {
+                                    let _ = done.send(Err(error.to_string()));
+                                }
+                            }
+                        }
+                        Some(Command::Close) | None => break,
+                    },
+                    _ = cx.incoming_closed() => break,
+                }
+            }
+            Ok(())
+        })
+        .await;
+    if let Err(e) = result {
+        tracing::debug!(error = %e, "ACP connection ended with an error");
+    }
+}
+
+async fn drain_stderr(
+    stderr: tokio::process::ChildStderr,
+    pid: u32,
+    tail: Arc<Mutex<VecDeque<u8>>>,
+) {
+    let mut lines = tokio::io::BufReader::new(stderr).lines();
+    while let Ok(Some(line)) = lines.next_line().await {
+        tracing::debug!(pid, "{ADAPTER_BIN}: {line}");
+        let mut tail = tail.lock().expect("stderr tail");
+        tail.extend(line.as_bytes());
+        tail.push_back(b'\n');
+        while tail.len() > STDERR_TAIL_BYTES {
+            tail.pop_front();
+        }
+        tail.make_contiguous();
+    }
+}

crates/maestro-acp/tests/fake_acp_agent.py

@@ -0,0 +1,161 @@
+#!/usr/bin/env python3
+"""A deterministic stand-in for `claude-code-acp` (#134).
+
+Speaks just enough JSON-RPC over stdio for the daemon's handshake:
+answers `initialize` and `session/new`, logs every method it receives, and
+exits on SIGTERM. Tests point `PATH` at a directory holding this file under
+the adapter's name, so the real adapter (which needs Claude auth and is
+slow) never runs in the deterministic suite.
+
+Environment knobs, all optional:
+
+  FAKE_ACP_LOG            append one line per event to this file:
+                            start pid=<pid> cwd=<cwd>
+                            env <NAME>=<value>       (for FAKE_ACP_ECHO_ENV)
+                            caps <json of clientCapabilities>
+                            recv <method>
+                            signal SIGTERM
+  FAKE_ACP_ECHO_ENV       name of one env var to log at start
+  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_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)
+  FAKE_ACP_IGNORE_EOF=1      keep running after stdin closes, until signalled
+  FAKE_ACP_EMIT_ON_CANCEL=1  emit a final notification before handling cancel
+  FAKE_ACP_NO_LOAD=1      advertise `loadSession: false`
+  FAKE_ACP_LOAD_REPLAY=n  on `session/load`, replay n `agent_message_chunk`
+                            notifications ("replayed <i>") before answering,
+                            the way a real agent streams history back
+  FAKE_ACP_LOAD_FAIL=1    answer `session/load` with a JSON-RPC error
+  FAKE_ACP_DELAY_PROMPT_MS=n  wait before answering `session/prompt`
+  FAKE_ACP_PROMPT_FAIL=1  answer every `session/prompt` with a JSON-RPC error
+  FAKE_ACP_PROMPT_FAIL_ONCE=1  answer only the first `session/prompt` with a
+                             JSON-RPC error
+
+Signals: SIGUSR1 emits one `agent_message_chunk` notification ("tick <n>",
+counting from 1) and logs `emit tick <n>`, so a test can make the agent
+speak on demand (#135).
+"""
+import json
+import os
+import signal
+import sys
+import time
+
+LOG = os.environ.get("FAKE_ACP_LOG")
+
+
+def log(line):
+    if not LOG:
+        return
+    with open(LOG, "a") as f:
+        f.write(line + "\n")
+        f.flush()
+
+
+def send(obj):
+    sys.stdout.write(json.dumps(obj) + "\n")
+    sys.stdout.flush()
+
+
+def on_sigterm(_signum, _frame):
+    log("signal SIGTERM")
+    sys.exit(0)
+
+
+signal.signal(signal.SIGTERM, on_sigterm)
+
+session_id = f"fake-{os.getpid()}"
+ticks = 0
+prompt_failures = 0
+
+
+def chunk(text):
+    return {"jsonrpc": "2.0", "method": "session/update", "params": {
+        "sessionId": session_id,
+        "update": {"sessionUpdate": "agent_message_chunk",
+                   "content": {"type": "text", "text": text}},
+    }}
+
+
+def on_sigusr1(_signum, _frame):
+    global ticks
+    ticks += 1
+    send(chunk(f"tick {ticks}"))
+    log(f"emit tick {ticks}")
+
+
+signal.signal(signal.SIGUSR1, on_sigusr1)
+
+log(f"start pid={os.getpid()} cwd={os.getcwd()}")
+echo = os.environ.get("FAKE_ACP_ECHO_ENV")
+if echo:
+    log(f"env {echo}={os.environ.get(echo, '')}")
+
+for raw in sys.stdin:
+    raw = raw.strip()
+    if not raw:
+        continue
+    msg = json.loads(raw)
+    method = msg.get("method")
+    if method is None:
+        # A response to something we sent; nothing to do.
+        continue
+    log(f"recv {method}")
+    rid = msg.get("id")
+    params = msg.get("params") or {}
+
+    if method == "initialize":
+        log("caps " + json.dumps(params.get("clientCapabilities", {}), sort_keys=True))
+        if os.environ.get("FAKE_ACP_INIT_FAIL") == "1":
+            send({"jsonrpc": "2.0", "id": rid,
+                  "error": {"code": -32000, "message": "fake init failure"}})
+            continue
+        send({"jsonrpc": "2.0", "id": rid, "result": {
+            "protocolVersion": params.get("protocolVersion", 1),
+            "agentCapabilities": {
+                "loadSession": os.environ.get("FAKE_ACP_NO_LOAD") != "1"},
+            "agentInfo": {"name": "fake-acp-agent", "version": "0.0.0"},
+            "authMethods": [],
+        }})
+    elif method == "session/new":
+        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"))
+        if os.environ.get("FAKE_ACP_EXIT_AFTER_NEW") == "1":
+            sys.exit(0)
+    elif method == "session/load":
+        session_id = params.get("sessionId")
+        log(f"recv session/load sessionId={session_id}")
+        if os.environ.get("FAKE_ACP_LOAD_FAIL") == "1":
+            send({"jsonrpc": "2.0", "id": rid,
+                  "error": {"code": -32000, "message": "fake load failure"}})
+            continue
+        for i in range(int(os.environ.get("FAKE_ACP_LOAD_REPLAY", "0"))):
+            send(chunk(f"replayed {i + 1}"))
+        send({"jsonrpc": "2.0", "id": rid, "result": {}})
+        send(chunk("hello after load"))
+    elif method == "session/cancel":
+        if os.environ.get("FAKE_ACP_EMIT_ON_CANCEL") == "1":
+            send(chunk("final on cancel"))
+        if os.environ.get("FAKE_ACP_EXIT_ON_CANCEL") == "1":
+            sys.exit(0)
+    elif method == "session/prompt":
+        time.sleep(int(os.environ.get("FAKE_ACP_DELAY_PROMPT_MS", "0")) / 1000)
+        fail_once = os.environ.get("FAKE_ACP_PROMPT_FAIL_ONCE") == "1"
+        if os.environ.get("FAKE_ACP_PROMPT_FAIL") == "1" or (fail_once and prompt_failures == 0):
+            prompt_failures += 1
+            send({"jsonrpc": "2.0", "id": rid,
+                  "error": {"code": -32000, "message": "fake prompt failure"}})
+        else:
+            send({"jsonrpc": "2.0", "id": rid,
+                  "result": {"stopReason": "end_turn"}})
+    elif rid is not None:
+        send({"jsonrpc": "2.0", "id": rid,
+              "error": {"code": -32601, "message": f"method not found: {method}"}})
+
+if os.environ.get("FAKE_ACP_IGNORE_EOF") == "1":
+    while True:
+        signal.pause()

crates/maestro-acp/tests/spawn.rs

@@ -0,0 +1,394 @@
+//! `maestro-acp` against the fake adapter in `tests/fake_acp_agent.py`, plus
+//! one ignored test against the real `claude-code-acp`.
+
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::time::{Duration, Instant};
+
+use maestro_acp::schema::{ContentBlock, SessionUpdate};
+use maestro_acp::{AcpError, SpawnConfig, ADAPTER_BIN, INSTALL_COMMAND};
+use tempfile::TempDir;
+
+fn fake_agent() -> PathBuf {
+    Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fake_acp_agent.py")
+}
+
+/// A temp `bin/` holding the fake agent under the adapter's name, plus a
+/// log file the agent appends to. `PATH` for the spawn is just that dir.
+struct Fake {
+    dir: TempDir,
+}
+
+impl Fake {
+    fn new() -> Self {
+        let dir = TempDir::new().unwrap();
+        let bin = dir.path().join("bin");
+        std::fs::create_dir_all(&bin).unwrap();
+        std::fs::copy(fake_agent(), bin.join(ADAPTER_BIN)).unwrap();
+        Self { dir }
+    }
+
+    fn log_path(&self) -> PathBuf {
+        self.dir.path().join("agent.log")
+    }
+
+    fn log(&self) -> String {
+        std::fs::read_to_string(self.log_path()).unwrap_or_default()
+    }
+
+    /// The fake's `bin/` first, so it shadows any real adapter, then the
+    /// process `PATH` so the script's `python3` still resolves.
+    fn env(&self) -> HashMap<String, String> {
+        let path = format!(
+            "{}:{}",
+            self.dir.path().join("bin").display(),
+            std::env::var("PATH").unwrap_or_default()
+        );
+        HashMap::from([
+            ("PATH".to_string(), path),
+            (
+                "FAKE_ACP_LOG".to_string(),
+                self.log_path().to_string_lossy().into_owned(),
+            ),
+        ])
+    }
+
+    fn config(&self, cwd: &Path) -> SpawnConfig {
+        let mut config = SpawnConfig::new(cwd, self.env());
+        config.handshake_timeout = Duration::from_secs(10);
+        config
+    }
+
+    async fn wait_for_log(&self, needle: &str) {
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while Instant::now() < deadline {
+            if self.log().contains(needle) {
+                return;
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+        panic!("log never contained {needle:?}:\n{}", self.log());
+    }
+
+    fn pid(&self) -> u32 {
+        self.log()
+            .lines()
+            .find_map(|line| line.strip_prefix("start pid="))
+            .and_then(|rest| rest.split(' ').next())
+            .and_then(|pid| pid.parse().ok())
+            .expect("pid in log")
+    }
+
+    async fn wait_for_exit(&self) {
+        let pid = self.pid();
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while pid_alive(pid) && Instant::now() < deadline {
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+        assert!(!pid_alive(pid), "child {pid} still alive");
+    }
+}
+
+fn pid_alive(pid: u32) -> bool {
+    Path::new(&format!("/proc/{pid}")).exists()
+        && std::fs::read_to_string(format!("/proc/{pid}/stat"))
+            .map(|s| !s.contains(") Z "))
+            .unwrap_or(false)
+}
+
+#[tokio::test]
+async fn spawn_runs_adapter_in_cwd_with_env_and_restrictive_caps() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path());
+    config
+        .env
+        .insert("FAKE_ACP_ECHO_ENV".into(), "MAESTRO_SESSION_ID".into());
+    config
+        .env
+        .insert("MAESTRO_SESSION_ID".into(), "sess-123".into());
+
+    let session = maestro_acp::spawn(config).await.expect("spawn");
+    assert!(session.session_id().starts_with("fake-"));
+    assert!(pid_alive(session.pid()));
+
+    let log = fake.log();
+    let cwd_canon = cwd.path().canonicalize().unwrap();
+    assert!(
+        log.contains(&format!("cwd={}", cwd_canon.display())),
+        "child cwd missing from log:\n{log}"
+    );
+    assert!(log.contains("env MAESTRO_SESSION_ID=sess-123"), "{log}");
+    assert!(log.contains("recv initialize"), "{log}");
+    assert!(log.contains("recv session/new"), "{log}");
+    let caps = log
+        .lines()
+        .find_map(|l| l.strip_prefix("caps "))
+        .expect("caps line");
+    let caps: serde_json::Value = serde_json::from_str(caps).unwrap();
+    assert_eq!(caps["fs"]["readTextFile"], false);
+    assert_eq!(caps["fs"]["writeTextFile"], false);
+    assert_eq!(caps["terminal"], false);
+
+    session
+        .terminate(Duration::from_millis(100), Duration::from_secs(2))
+        .await;
+}
+
+#[tokio::test]
+async fn missing_adapter_names_the_install_command() {
+    let empty = TempDir::new().unwrap();
+    let cwd = TempDir::new().unwrap();
+    let env = HashMap::from([(
+        "PATH".to_string(),
+        empty.path().to_string_lossy().into_owned(),
+    )]);
+    let err = maestro_acp::spawn(SpawnConfig::new(cwd.path(), env))
+        .await
+        .expect_err("no adapter on PATH");
+    assert!(matches!(err, AcpError::AdapterMissing));
+    let msg = err.to_string();
+    assert!(msg.contains(ADAPTER_BIN), "{msg}");
+    assert!(msg.contains(INSTALL_COMMAND), "{msg}");
+}
+
+#[tokio::test]
+async fn session_updates_reach_the_taken_receiver() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let session = maestro_acp::spawn(fake.config(cwd.path()))
+        .await
+        .expect("spawn");
+
+    let mut updates = session.take_updates().expect("first take");
+    assert!(session.take_updates().is_none(), "handed out once");
+    let first = tokio::time::timeout(Duration::from_secs(5), updates.recv())
+        .await
+        .expect("update within 5s")
+        .expect("channel open");
+    assert_eq!(first.session_id.0.as_ref(), session.session_id());
+    match first.update {
+        SessionUpdate::AgentMessageChunk(chunk) => match chunk.content {
+            ContentBlock::Text(t) => assert_eq!(t.text, "hello from fake"),
+            other => panic!("unexpected content {other:?}"),
+        },
+        other => panic!("unexpected update {other:?}"),
+    }
+
+    session
+        .terminate(Duration::from_millis(100), Duration::from_secs(2))
+        .await;
+}
+
+#[tokio::test]
+async fn terminate_sends_cancel_then_sigterm() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let session = maestro_acp::spawn(fake.config(cwd.path()))
+        .await
+        .expect("spawn");
+    let pid = session.pid();
+
+    session
+        .terminate(Duration::from_millis(200), Duration::from_secs(2))
+        .await;
+    assert!(session.has_exited());
+    fake.wait_for_log("signal SIGTERM").await;
+
+    let log = fake.log();
+    let cancel = log.find("recv session/cancel").expect("cancel logged");
+    let term = log.find("signal SIGTERM").expect("SIGTERM logged");
+    assert!(cancel < term, "cancel must precede SIGTERM:\n{log}");
+    assert!(!pid_alive(pid), "child {pid} still alive");
+}
+
+#[tokio::test]
+async fn terminate_is_satisfied_by_an_exit_on_cancel() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path());
+    config
+        .env
+        .insert("FAKE_ACP_EXIT_ON_CANCEL".into(), "1".into());
+    let session = maestro_acp::spawn(config).await.expect("spawn");
+
+    session
+        .terminate(Duration::from_secs(2), Duration::from_secs(2))
+        .await;
+    assert!(session.has_exited());
+    let log = fake.log();
+    assert!(log.contains("recv session/cancel"), "{log}");
+    assert!(!log.contains("signal SIGTERM"), "cancel was enough:\n{log}");
+}
+
+#[tokio::test]
+async fn handshake_error_fails_the_spawn_and_reaps_the_child() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path());
+    config.env.insert("FAKE_ACP_INIT_FAIL".into(), "1".into());
+
+    let err = maestro_acp::spawn(config).await.expect_err("init fails");
+    assert!(matches!(err, AcpError::Handshake(_)), "{err}");
+    assert!(err.to_string().contains("fake init failure"), "{err}");
+    // Closing the connection gives the fake EOF on stdin; the SIGTERM that
+    // follows may find it already gone. Either way it must not survive.
+    let pid: u32 = fake
+        .log()
+        .lines()
+        .find_map(|l| l.strip_prefix("start pid="))
+        .and_then(|rest| rest.split(' ').next())
+        .and_then(|p| p.parse().ok())
+        .expect("pid in log");
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while pid_alive(pid) && Instant::now() < deadline {
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    assert!(!pid_alive(pid), "child {pid} outlived the failed spawn");
+}
+
+#[tokio::test]
+async fn early_exit_is_observed_after_a_successful_handshake() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path());
+    config
+        .env
+        .insert("FAKE_ACP_EXIT_AFTER_NEW".into(), "1".into());
+    let session = maestro_acp::spawn(config).await.expect("spawn");
+
+    let status = tokio::time::timeout(Duration::from_secs(5), session.wait())
+        .await
+        .expect("exit within 5s")
+        .expect("status");
+    assert!(status.success(), "{status}");
+    assert!(session.has_exited());
+}
+
+#[tokio::test]
+async fn child_exit_before_handshake_reports_exited_early() {
+    // A "adapter" that dies before answering anything.
+    let dir = TempDir::new().unwrap();
+    let bin = dir.path().join("bin");
+    std::fs::create_dir_all(&bin).unwrap();
+    let script = bin.join(ADAPTER_BIN);
+    std::fs::write(&script, "#!/bin/sh\necho 'not logged in' >&2\nexit 3\n").unwrap();
+    {
+        use std::os::unix::fs::PermissionsExt;
+        std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
+    }
+    let cwd = TempDir::new().unwrap();
+    let env = HashMap::from([("PATH".to_string(), bin.to_string_lossy().into_owned())]);
+    let mut config = SpawnConfig::new(cwd.path(), env);
+    config.handshake_timeout = Duration::from_secs(10);
+
+    let err = maestro_acp::spawn(config).await.expect_err("child died");
+    match &err {
+        AcpError::ExitedEarly { status, stderr } => {
+            assert!(status.contains('3'), "{status}");
+            assert!(stderr.contains("not logged in"), "{stderr}");
+        }
+        other => panic!("unexpected {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn drop_terminates_and_reaps_the_child() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path());
+    config.env.insert("FAKE_ACP_IGNORE_EOF".into(), "1".into());
+    let session = maestro_acp::spawn(config).await.expect("spawn");
+
+    drop(session);
+
+    fake.wait_for_log("signal SIGTERM").await;
+    fake.wait_for_exit().await;
+}
+
+fn chunk_text(n: &maestro_acp::schema::SessionNotification) -> String {
+    match &n.update {
+        SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
+            ContentBlock::Text(t) => t.text.clone(),
+            other => panic!("unexpected content {other:?}"),
+        },
+        other => panic!("unexpected update {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn load_session_resumes_the_stored_id_and_drops_the_replay() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path()).load_session("fake-old");
+    config.env.insert("FAKE_ACP_LOAD_REPLAY".into(), "3".into());
+
+    let session = maestro_acp::spawn(config).await.expect("spawn");
+    assert_eq!(session.session_id(), "fake-old");
+    let log = fake.log();
+    assert!(
+        log.contains("recv session/load sessionId=fake-old"),
+        "{log}"
+    );
+    assert!(!log.contains("recv session/new"), "{log}");
+
+    // The three replayed chunks never reach the receiver: the event log
+    // already holds them. The first notification after the response does.
+    let mut updates = session.take_updates().expect("updates");
+    let first = tokio::time::timeout(Duration::from_secs(5), updates.recv())
+        .await
+        .expect("update within 5s")
+        .expect("channel open");
+    assert_eq!(chunk_text(&first), "hello after load");
+    // Nothing queued behind it either.
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    assert!(updates.try_recv().is_err(), "replay leaked through");
+
+    session
+        .terminate(Duration::from_millis(100), Duration::from_secs(2))
+        .await;
+}
+
+#[tokio::test]
+async fn load_session_requires_the_load_capability() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path()).load_session("fake-old");
+    config.env.insert("FAKE_ACP_NO_LOAD".into(), "1".into());
+
+    let err = maestro_acp::spawn(config)
+        .await
+        .expect_err("no loadSession");
+    assert!(matches!(err, AcpError::Handshake(_)), "{err}");
+    assert!(err.to_string().contains("loadSession"), "{err}");
+    assert!(!fake.log().contains("recv session/load"), "{}", fake.log());
+}
+
+#[tokio::test]
+async fn load_session_error_fails_the_spawn() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path()).load_session("fake-old");
+    config.env.insert("FAKE_ACP_LOAD_FAIL".into(), "1".into());
+
+    let err = maestro_acp::spawn(config).await.expect_err("load fails");
+    assert!(matches!(err, AcpError::Handshake(_)), "{err}");
+    assert!(err.to_string().contains("fake load failure"), "{err}");
+}
+
+/// End to end against the installed `claude-code-acp`. Needs Claude auth,
+/// takes seconds, so it is opt-in: `cargo test -p maestro-acp -- --ignored`.
+#[tokio::test]
+#[ignore]
+async fn real_adapter_handshake() {
+    let cwd = TempDir::new().unwrap();
+    let session = maestro_acp::spawn(SpawnConfig::new(cwd.path(), HashMap::new()))
+        .await
+        .expect("real claude-code-acp handshake");
+    assert!(!session.session_id().is_empty());
+    assert!(pid_alive(session.pid()));
+    session
+        .terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
+        .await;
+    assert!(session.has_exited());
+}

crates/maestro-cli/src/sessions.rs

@@ -51,6 +51,7 @@ pub async fn run_spawn(client: &DaemonClient, args: SpawnArgs) -> Result<Session
     };
 
     let body = CreateSessionRequest {
+        view: Default::default(),
         cwd: args.cwd.clone(),
         command: args.command.clone(),
         env,
@@ -236,6 +237,7 @@ mod tests {
 
     fn session(cwd: &str, command: &str, id: u128) -> Session {
         Session {
+            view: Default::default(),
             id: Uuid::from_u128(id),
             cwd: PathBuf::from(cwd),
             command: command.into(),
@@ -252,6 +254,7 @@ mod tests {
             current_cwd: None,
             blocked_work: None,
             last_focused_at: None,
+            acp_session_id: None,
         }
     }
 

crates/maestro-cli/tests/against_mock_daemon.rs

@@ -91,6 +91,7 @@ async fn spawn_round_trips_through_the_daemon() {
     let created = mock
         .client()
         .create_session(&maestro_protocol::CreateSessionRequest {
+            view: Default::default(),
             cwd: "/home/mock".into(),
             command: "bash".into(),
             env: Default::default(),

crates/maestro-daemon/Cargo.toml

@@ -28,6 +28,7 @@ dirs.workspace = true
 futures-util.workspace = true
 include_dir = { workspace = true, optional = true }
 libc.workspace = true
+maestro-acp.workspace = true
 mime_guess = { workspace = true, optional = true }
 maestro-config.workspace = true
 maestro-hooks = { workspace = true, default-features = false }
@@ -42,6 +43,7 @@ tracing-subscriber.workspace = true
 uuid.workspace = true
 
 [dev-dependencies]
+libc.workspace = true
 reqwest.workspace = true
 tokio-tungstenite.workspace = true
 urlencoding = "2"

crates/maestro-daemon/src/cli.rs

@@ -106,7 +106,7 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
     projects.start_background_refresh();
 
     let router = local_api::router(local_api::state_for(
-        manager,
+        manager.clone(),
         config.hostname.clone(),
         config.home_dir.clone(),
         config.spawn_presets.clone(),
@@ -118,9 +118,22 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
     let bound = listener.local_addr()?;
     tracing::info!(addr = %bound, "API listening");
 
-    let shutdown = async {
+    // Ctrl-C at a terminal or SIGTERM from systemd: either one is a
-        let _ = tokio::signal::ctrl_c().await;
+    // shutdown, and both must reach the adapter children below (#134). The
-        tracing::info!("ctrl-c received — shutting down");
+    // unit's `KillMode=process` signals only the daemon, so nothing else
+    // would end them.
+    let shutdown_manager = manager.clone();
+    let shutdown = async move {
+        let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
+            .expect("install SIGTERM handler");
+        tokio::select! {
+            _ = tokio::signal::ctrl_c() => tracing::info!("ctrl-c received — shutting down"),
+            _ = sigterm.recv() => tracing::info!("SIGTERM received — shutting down"),
+        }
+        // Axum waits for upgraded WebSockets to drain. Structured children
+        // must be settled before that wait, while the signal path still owns
+        // the shutdown order (#179).
+        shutdown_manager.shutdown().await;
     };
 
     axum::serve(listener, router)

crates/maestro-daemon/src/lib.rs

@@ -21,6 +21,7 @@ pub mod hooks;
 pub mod local_api;
 pub mod panes;
 pub mod proc;
+pub mod projection;
 pub mod projects;
 pub mod pty;
 pub mod release;

crates/maestro-daemon/src/local_api.rs

@@ -21,12 +21,14 @@ use maestro_config::SpawnPreset;
 use maestro_protocol::{
     routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonInfo, GitRoot, NameBatch,
     RenameSessionRequest, ReportActivityRequest, SendInputRequest, Session, SetFocusRequest,
-    PROTOCOL_VERSION,
+    StructuredEvent, StructuredPromptRequest, PROTOCOL_VERSION,
 };
 use serde::Deserialize;
 use uuid::Uuid;
 
+use crate::projection::TurnLifecycle;
 use crate::session::SessionManager;
+use crate::sidecar::EventCursor;
 
 #[derive(Debug, Clone)]
 pub struct LocalApiState {
@@ -64,7 +66,11 @@ pub fn router(state: LocalApiState) -> Router {
         .route(routes::NAMES, get(names))
         .route(routes::INFO, get(daemon_info))
         .route(routes::EVENTS, any(events_ws))
-        .route(routes::patterns::ATTACH, any(attach_ws));
+        .route(routes::patterns::ATTACH, any(attach_ws))
+        .route(routes::patterns::STRUCTURED, any(structured_ws))
+        .route(routes::patterns::STRUCTURED_REPLAY, get(structured_replay))
+        .route(routes::patterns::STRUCTURED_PROMPT, post(structured_prompt))
+        .route(routes::patterns::STRUCTURED_CANCEL, post(structured_cancel));
     #[cfg(feature = "haiku-autoname")]
     {
         router = router.route(routes::patterns::SESSION_PROMPT, post(session_prompt));
@@ -170,6 +176,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,
+        "shutting_down" => StatusCode::SERVICE_UNAVAILABLE,
         _ => StatusCode::BAD_GATEWAY,
     }
 }
@@ -368,6 +375,121 @@ fn error_response(status: StatusCode, code: &str, message: &str) -> Response {
     (status, Json(ApiError::new(code, message))).into_response()
 }
 
+#[derive(Debug, Deserialize)]
+struct ReplayQuery {
+    since: Option<u64>,
+    before: Option<u64>,
+    limit: Option<usize>,
+}
+
+const MAX_STRUCTURED_REPLAY_LIMIT: usize = 500;
+
+async fn structured_replay(
+    State(s): State<LocalApiState>,
+    Path(id): Path<Uuid>,
+    Query(query): Query<ReplayQuery>,
+) -> Response {
+    if query.since.is_some() && query.before.is_some() {
+        return error_response(
+            StatusCode::BAD_REQUEST,
+            "invalid_cursor",
+            "use either since or before, not both",
+        );
+    }
+    let cursor = match query.before {
+        Some(before) => EventCursor::Before(Some(before)),
+        None => EventCursor::Since(query.since.unwrap_or(0)),
+    };
+    match s
+        .manager
+        .read_acp_projection_replay(
+            id,
+            cursor,
+            query
+                .limit
+                .unwrap_or(100)
+                .clamp(1, MAX_STRUCTURED_REPLAY_LIMIT),
+        )
+        .await
+    {
+        Ok(Some(events)) => Json(events).into_response(),
+        // A terminal session, or a structured session whose adapter is no
+        // longer live, cannot provide a coherent replay continuation.
+        Ok(None) => error_response(
+            StatusCode::CONFLICT,
+            "not_structured",
+            "session is not a live structured session",
+        ),
+        Err(e) => {
+            let info = e.to_error_info();
+            error_response(session_error_status(&info.code), &info.code, &info.message)
+        }
+    }
+}
+
+async fn structured_prompt(
+    State(s): State<LocalApiState>,
+    Path(id): Path<Uuid>,
+    Json(body): Json<StructuredPromptRequest>,
+) -> Response {
+    let (acp, turn) = match s.manager.begin_acp_turn(id).await {
+        Ok(crate::session::BeginAcpTurn::Started(acp, turn)) => (acp, turn),
+        Ok(crate::session::BeginAcpTurn::InFlight) => {
+            return error_response(
+                StatusCode::CONFLICT,
+                "turn_in_flight",
+                "the structured session already has an active turn",
+            )
+        }
+        Ok(crate::session::BeginAcpTurn::Unavailable) => {
+            return error_response(
+                StatusCode::CONFLICT,
+                "not_structured",
+                "session is not a live structured session",
+            )
+        }
+        Err(e) => {
+            let info = e.to_error_info();
+            return error_response(session_error_status(&info.code), &info.code, &info.message);
+        }
+    };
+    match acp.prompt(body.prompt) {
+        Ok(done) => {
+            let manager = s.manager.clone();
+            tokio::spawn(async move {
+                let lifecycle = match done.await {
+                    Ok(Ok(())) => TurnLifecycle::Finish,
+                    Ok(Err(_)) | Err(_) => TurnLifecycle::Failure,
+                };
+                let _ = manager.apply_acp_turn_lifecycle(id, lifecycle, turn).await;
+            });
+            StatusCode::NO_CONTENT.into_response()
+        }
+        Err(message) => {
+            let _ = s
+                .manager
+                .apply_acp_turn_lifecycle(id, TurnLifecycle::Failure, turn)
+                .await;
+            error_response(StatusCode::BAD_GATEWAY, "acp_prompt_failed", &message)
+        }
+    }
+}
+
+async fn structured_cancel(State(s): State<LocalApiState>, Path(id): Path<Uuid>) -> Response {
+    match s.manager.cancel_acp_turn(id).await {
+        Ok(true) => StatusCode::NO_CONTENT.into_response(),
+        Ok(false) => error_response(
+            StatusCode::CONFLICT,
+            "not_structured",
+            "session is not a live structured session",
+        ),
+        Err(e) => {
+            let info = e.to_error_info();
+            error_response(session_error_status(&info.code), &info.code, &info.message)
+        }
+    }
+}
+
 /// Convenience constructor — pulls the bits `LocalApiState` needs from
 /// `DaemonConfig` so callers don't have to assemble it by hand. The project
 /// scanner starts with its cache cold; wire
@@ -418,6 +540,75 @@ async fn attach_ws(
     }))
 }
 
+#[allow(clippy::result_large_err)]
+async fn structured_ws(
+    Path(id): Path<Uuid>,
+    State(s): State<LocalApiState>,
+    ws: axum::extract::WebSocketUpgrade,
+) -> Result<Response, Response> {
+    let Some((snapshot, state, rx)) = s
+        .manager
+        .subscribe_acp_projection(id, EventCursor::Since(0))
+        .await
+        .map_err(|e| {
+            let info = e.to_error_info();
+            error_response(session_error_status(&info.code), &info.code, &info.message)
+        })?
+    else {
+        return Err(error_response(
+            StatusCode::CONFLICT,
+            "not_structured",
+            "session is not a live structured session",
+        ));
+    };
+    Ok(ws.on_upgrade(move |socket| async move {
+        let _ = pump_structured(socket, snapshot, state, rx).await;
+    }))
+}
+
+async fn pump_structured(
+    socket: axum::extract::ws::WebSocket,
+    snapshot: maestro_protocol::TranscriptSnapshot,
+    state: maestro_protocol::ReducedState,
+    mut rx: tokio::sync::broadcast::Receiver<crate::projection::ProjectionUpdate>,
+) -> Result<(), axum::Error> {
+    use axum::extract::ws::Message;
+    use futures_util::{SinkExt, StreamExt};
+
+    let (mut sink, mut stream) = socket.split();
+    for event in [
+        StructuredEvent::TranscriptSnapshot { snapshot },
+        StructuredEvent::ReducedState { state },
+    ] {
+        if sink
+            .send(Message::Text(
+                serde_json::to_string(&event).expect("structured event encode"),
+            ))
+            .await
+            .is_err()
+        {
+            return Ok(());
+        }
+    }
+    loop {
+        tokio::select! {
+            update = rx.recv() => match update {
+                Ok(update) => {
+                    let event = StructuredEvent::ProjectionUpdate { transcript: update.transcript, state: update.state };
+                    if sink.send(Message::Text(serde_json::to_string(&event).expect("structured event encode"))).await.is_err() { break; }
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => break,
+                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+            },
+            frame = stream.next() => match frame {
+                None | Some(Err(_)) | Some(Ok(Message::Close(_))) => break,
+                Some(Ok(_)) => {},
+            },
+        }
+    }
+    Ok(())
+}
+
 // ---------------------------------------------------------------------------
 // WS /api/events — live catalog deltas for browser/TUI subscribers (NAP-248)
 // ---------------------------------------------------------------------------

crates/maestro-daemon/src/projection.rs

@@ -0,0 +1,635 @@
+//! Fold persisted ACP notifications into client-neutral transcript and state.
+
+use std::collections::{HashMap, HashSet};
+
+use maestro_acp::schema::{ContentChunk, SessionNotification, SessionUpdate, ToolCall};
+use maestro_protocol::{
+    ReducedState, TranscriptDelta, TranscriptRow, TranscriptRowKind, TranscriptSnapshot,
+};
+use serde_json::Value;
+
+/// A transcript delta paired with the state after the same ACP notification.
+#[derive(Debug, Clone, PartialEq)]
+pub struct ProjectionUpdate {
+    pub transcript: Option<TranscriptDelta>,
+    pub state: ReducedState,
+}
+
+/// A prompt lifecycle transition supplied by the prompt transport (#138).
+///
+/// ACP notifications do not delimit turns, so replaying them must not infer
+/// an in-flight prompt from an agent chunk.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TurnLifecycle {
+    Begin,
+    Finish,
+    Cancel,
+    Failure,
+}
+
+/// The one-pass ACP fold used for both persisted replay and live events.
+#[derive(Debug, Default)]
+pub struct Projection {
+    rows: Vec<TranscriptRow>,
+    row_index: HashMap<String, usize>,
+    tools: HashMap<String, ToolCall>,
+    pending_tool_updates: HashMap<String, Vec<maestro_acp::schema::ToolCallUpdateFields>>,
+    state: ReducedState,
+    active_turn: Option<u64>,
+    next_turn: u64,
+    turn_tool_calls: HashSet<String>,
+}
+
+impl Projection {
+    /// Fold one durable event and return its client-visible changes.
+    pub fn apply(&mut self, seq: u64, notification: &SessionNotification) -> ProjectionUpdate {
+        let transcript = match &notification.update {
+            SessionUpdate::UserMessageChunk(chunk) => {
+                self.content(seq, chunk, TranscriptRowKind::User, "user")
+            }
+            SessionUpdate::AgentMessageChunk(chunk) => {
+                self.agent_text();
+                self.content(seq, chunk, TranscriptRowKind::AgentText, "agent")
+            }
+            SessionUpdate::AgentThoughtChunk(chunk) => {
+                self.agent_thought();
+                self.content(seq, chunk, TranscriptRowKind::AgentThought, "thought")
+            }
+            SessionUpdate::ToolCall(tool) => self.tool_call(tool),
+            SessionUpdate::ToolCallUpdate(update) => self.tool_update(update),
+            SessionUpdate::Plan(plan) => {
+                let data = to_value(plan);
+                self.state.plan = Some(data.clone());
+                Some(self.replace("plan".into(), TranscriptRowKind::Plan, data))
+            }
+            SessionUpdate::UsageUpdate(usage) => {
+                let value = to_value(usage);
+                self.state.usage = Some(value.clone());
+                self.state.context = Some(value);
+                None
+            }
+            SessionUpdate::CurrentModeUpdate(mode) => {
+                self.state.mode = Some(mode.current_mode_id.to_string());
+                None
+            }
+            SessionUpdate::ConfigOptionUpdate(options) => {
+                let value = to_value(options);
+                self.state.model = model_from_options(&value).or_else(|| self.state.model.clone());
+                None
+            }
+            // ACP deliberately adds notification variants over time. Known but
+            // non-transcript updates become a generic notice instead of being
+            // dropped or forcing clients to understand ACP's enum.
+            _ => Some(self.append(
+                format!("notice:{seq}"),
+                TranscriptRowKind::Notice,
+                to_value(&notification.update),
+            )),
+        };
+        ProjectionUpdate {
+            transcript,
+            state: self.state.clone(),
+        }
+    }
+
+    /// The current transcript at `cursor`.
+    pub fn snapshot(&self, cursor: u64) -> TranscriptSnapshot {
+        TranscriptSnapshot {
+            cursor,
+            rows: self.rows.clone(),
+        }
+    }
+
+    /// The state folded over every event seen so far.
+    pub fn state(&self) -> ReducedState {
+        self.state.clone()
+    }
+
+    pub fn turn_active(&self) -> bool {
+        self.active_turn.is_some()
+    }
+
+    /// Start a prompt turn and return its monotonically increasing token.
+    pub fn begin_turn(&mut self) -> (u64, ProjectionUpdate) {
+        self.next_turn += 1;
+        let turn = self.next_turn;
+        self.active_turn = Some(turn);
+        self.state.turn.responding = true;
+        self.state.turn.thinking = true;
+        self.turn_tool_calls.clear();
+        self.state.turn.active_tool_calls.clear();
+        (
+            turn,
+            ProjectionUpdate {
+                transcript: None,
+                state: self.state(),
+            },
+        )
+    }
+
+    /// End `turn` only when it remains the active prompt turn.
+    pub fn turn_lifecycle(
+        &mut self,
+        lifecycle: TurnLifecycle,
+        turn: u64,
+    ) -> Option<ProjectionUpdate> {
+        if !matches!(
+            lifecycle,
+            TurnLifecycle::Finish | TurnLifecycle::Cancel | TurnLifecycle::Failure
+        ) || self.active_turn != Some(turn)
+        {
+            return None;
+        }
+        self.active_turn = None;
+        self.state.turn.responding = false;
+        self.state.turn.thinking = false;
+        self.turn_tool_calls.clear();
+        self.state.turn.active_tool_calls.clear();
+        Some(ProjectionUpdate {
+            transcript: None,
+            state: self.state(),
+        })
+    }
+
+    pub fn active_turn(&self) -> Option<u64> {
+        self.active_turn
+    }
+
+    fn content(
+        &mut self,
+        seq: u64,
+        chunk: &ContentChunk,
+        kind: TranscriptRowKind,
+        prefix: &str,
+    ) -> Option<TranscriptDelta> {
+        let id = chunk
+            .message_id
+            .as_ref()
+            .map(|id| format!("{prefix}:{id}"))
+            .unwrap_or_else(|| format!("{prefix}:{seq}"));
+        let data = to_value(chunk);
+        if let Some(index) = self.row_index.get(&id).copied() {
+            let row = &mut self.rows[index];
+            append_text(&mut row.data, &data);
+            return Some(TranscriptDelta::Patch { row: row.clone() });
+        }
+        Some(self.append(id, kind, data))
+    }
+
+    fn tool_call(&mut self, tool: &ToolCall) -> Option<TranscriptDelta> {
+        let id = format!("tool:{}", tool.tool_call_id);
+        let tool_id = tool.tool_call_id.to_string();
+        let mut canonical = tool.clone();
+        if let Some(updates) = self.pending_tool_updates.get(&tool_id) {
+            for fields in updates {
+                canonical.update(fields.clone());
+            }
+        }
+        let kind = if tool_has_diff(tool) {
+            TranscriptRowKind::ToolDiff
+        } else {
+            TranscriptRowKind::ToolCall
+        };
+        self.tools.insert(tool_id, canonical.clone());
+        self.track_tool_call(&canonical);
+        Some(self.replace(
+            id,
+            if tool_has_diff(&canonical) {
+                TranscriptRowKind::ToolDiff
+            } else {
+                kind
+            },
+            to_value(&canonical),
+        ))
+    }
+
+    fn tool_update(
+        &mut self,
+        update: &maestro_acp::schema::ToolCallUpdate,
+    ) -> Option<TranscriptDelta> {
+        let tool_id = update.tool_call_id.to_string();
+        let Some(tool) = self.tools.get_mut(&tool_id) else {
+            self.pending_tool_updates
+                .entry(tool_id)
+                .or_default()
+                .push(update.fields.clone());
+            return None;
+        };
+        tool.update(update.fields.clone());
+        self.pending_tool_updates
+            .entry(tool_id.clone())
+            .or_default()
+            .push(update.fields.clone());
+        let row_id = format!("tool:{tool_id}");
+        let index = *self
+            .row_index
+            .get(&row_id)
+            .expect("a tracked tool call always has a transcript row");
+        let row = {
+            let row = &mut self.rows[index];
+            row.kind = if tool_has_diff(tool) {
+                TranscriptRowKind::ToolDiff
+            } else {
+                TranscriptRowKind::ToolCall
+            };
+            row.data = to_value(tool);
+            row.clone()
+        };
+        self.sync_active_tool(&tool_id);
+        Some(TranscriptDelta::Patch { row })
+    }
+
+    fn append(&mut self, id: String, kind: TranscriptRowKind, data: Value) -> TranscriptDelta {
+        let row = TranscriptRow { id, kind, data };
+        self.row_index.insert(row.id.clone(), self.rows.len());
+        self.rows.push(row.clone());
+        TranscriptDelta::Append { row }
+    }
+
+    fn replace(&mut self, id: String, kind: TranscriptRowKind, data: Value) -> TranscriptDelta {
+        let Some(index) = self.row_index.get(&id).copied() else {
+            return self.append(id, kind, data);
+        };
+        let row = &mut self.rows[index];
+        row.kind = kind;
+        row.data = data;
+        TranscriptDelta::Patch { row: row.clone() }
+    }
+
+    fn agent_text(&mut self) {
+        if self.turn_active() {
+            self.state.turn.responding = true;
+            self.state.turn.thinking = false;
+        }
+    }
+
+    fn agent_thought(&mut self) {
+        if self.turn_active() {
+            self.state.turn.responding = true;
+            self.state.turn.thinking = true;
+        }
+    }
+
+    fn track_tool_call(&mut self, tool: &ToolCall) {
+        if self.turn_active() {
+            self.turn_tool_calls.insert(tool.tool_call_id.to_string());
+            self.sync_active_tool(&tool.tool_call_id.to_string());
+        }
+    }
+
+    fn sync_active_tool(&mut self, tool_id: &str) {
+        if !self.turn_active() || !self.turn_tool_calls.contains(tool_id) {
+            return;
+        }
+        let active = self.tools.get(tool_id).is_some_and(|tool| {
+            matches!(
+                tool.status,
+                maestro_acp::schema::ToolCallStatus::Pending
+                    | maestro_acp::schema::ToolCallStatus::InProgress
+            )
+        });
+        let active_tool_calls = &mut self.state.turn.active_tool_calls;
+        if active {
+            if !active_tool_calls.iter().any(|id| id == tool_id) {
+                active_tool_calls.push(tool_id.to_owned());
+                active_tool_calls.sort();
+            }
+        } else {
+            active_tool_calls.retain(|id| id != tool_id);
+        }
+    }
+}
+
+fn to_value(value: &impl serde::Serialize) -> Value {
+    serde_json::to_value(value).expect("ACP schema values serialize")
+}
+
+fn tool_has_diff(tool: &ToolCall) -> bool {
+    tool.content
+        .iter()
+        .any(|content| matches!(content, maestro_acp::schema::ToolCallContent::Diff(_)))
+}
+
+fn append_text(existing: &mut Value, next: &Value) {
+    let Some(existing_text) = existing.pointer_mut("/content/text") else {
+        *existing = next.clone();
+        return;
+    };
+    let Some(next_text) = next.pointer("/content/text").and_then(Value::as_str) else {
+        *existing = next.clone();
+        return;
+    };
+    if let Some(text) = existing_text.as_str() {
+        *existing_text = Value::String(format!("{text}{next_text}"));
+    } else {
+        *existing = next.clone();
+    }
+}
+
+fn model_from_options(value: &Value) -> Option<String> {
+    value
+        .get("configOptions")?
+        .as_array()?
+        .iter()
+        .find(|option| {
+            option.get("id").and_then(Value::as_str) == Some("model")
+                || option.get("category").and_then(Value::as_str) == Some("model")
+        })?
+        .get("currentValue")?
+        .as_str()
+        .map(str::to_owned)
+}
+
+#[cfg(test)]
+mod tests {
+    use maestro_acp::schema::SessionNotification;
+    use maestro_protocol::{TranscriptDelta, TranscriptRowKind};
+
+    use super::{Projection, TurnLifecycle};
+
+    fn event(value: serde_json::Value) -> SessionNotification {
+        serde_json::from_value(serde_json::json!({
+            "sessionId": "acp-1",
+            "update": value,
+        }))
+        .unwrap()
+    }
+
+    #[test]
+    fn folds_content_tool_diff_plan_and_notice_rows() {
+        let mut projection = Projection::default();
+        let events = [
+            event(
+                serde_json::json!({"sessionUpdate":"user_message_chunk","content":{"type":"text","text":"hi"}}),
+            ),
+            event(
+                serde_json::json!({"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"hello"}}),
+            ),
+            event(
+                serde_json::json!({"sessionUpdate":"agent_thought_chunk","content":{"type":"text","text":"think"}}),
+            ),
+            event(
+                serde_json::json!({"sessionUpdate":"tool_call","toolCallId":"read-1","title":"Read","kind":"read"}),
+            ),
+            event(
+                serde_json::json!({"sessionUpdate":"tool_call","toolCallId":"edit-1","title":"Edit","content":[{"type":"diff","path":"x.rs","oldText":"a","newText":"b"}]}),
+            ),
+            event(serde_json::json!({"sessionUpdate":"plan","entries":[]})),
+            event(
+                serde_json::json!({"sessionUpdate":"available_commands_update","availableCommands":[]}),
+            ),
+        ];
+
+        let kinds: Vec<_> = events
+            .iter()
+            .enumerate()
+            .filter_map(|(i, event)| projection.apply((i + 1) as u64, event).transcript)
+            .map(|delta| match delta {
+                TranscriptDelta::Append { row } => row.kind,
+                other => panic!("expected append, got {other:?}"),
+            })
+            .collect();
+        assert_eq!(
+            kinds,
+            [
+                TranscriptRowKind::User,
+                TranscriptRowKind::AgentText,
+                TranscriptRowKind::AgentThought,
+                TranscriptRowKind::ToolCall,
+                TranscriptRowKind::ToolDiff,
+                TranscriptRowKind::Plan,
+                TranscriptRowKind::Notice,
+            ]
+        );
+    }
+
+    #[test]
+    fn tool_update_patches_the_original_stable_row() {
+        let mut projection = Projection::default();
+        let first = projection.apply(
+            4,
+            &event(serde_json::json!({
+                "sessionUpdate":"tool_call", "toolCallId":"bash-7", "title":"Run"
+            })),
+        );
+        let updated = projection.apply(
+            5,
+            &event(serde_json::json!({
+                "sessionUpdate":"tool_call_update", "toolCallId":"bash-7", "status":"completed"
+            })),
+        );
+        let id = match first.transcript.unwrap() {
+            TranscriptDelta::Append { row } => row.id,
+            other => panic!("expected append, got {other:?}"),
+        };
+        match updated.transcript.unwrap() {
+            TranscriptDelta::Patch { row } => {
+                assert_eq!(row.id, id);
+                assert_eq!(row.data["status"], "completed");
+            }
+            other => panic!("expected patch, got {other:?}"),
+        }
+        assert_eq!(projection.snapshot(5).rows.len(), 1);
+    }
+
+    #[test]
+    fn duplicate_tool_calls_patch_one_stable_row() {
+        let mut projection = Projection::default();
+        projection.apply(
+            1,
+            &event(serde_json::json!({
+                "sessionUpdate":"tool_call", "toolCallId":"bash-7", "title":"Run"
+            })),
+        );
+        let duplicate = projection.apply(
+            2,
+            &event(serde_json::json!({
+                "sessionUpdate":"tool_call", "toolCallId":"bash-7", "title":"Run again"
+            })),
+        );
+        match duplicate.transcript.unwrap() {
+            TranscriptDelta::Patch { row } => {
+                assert_eq!(row.id, "tool:bash-7");
+                assert_eq!(row.data["title"], "Run again");
+            }
+            other => panic!("expected patch, got {other:?}"),
+        }
+        assert_eq!(projection.snapshot(2).rows.len(), 1);
+    }
+
+    #[test]
+    fn early_tool_update_is_reconciled_when_the_call_arrives() {
+        let mut projection = Projection::default();
+        let early = projection.apply(
+            1,
+            &event(serde_json::json!({
+                "sessionUpdate":"tool_call_update", "toolCallId":"bash-7", "status":"completed"
+            })),
+        );
+        assert_eq!(early.transcript, None);
+        let call = projection.apply(
+            2,
+            &event(serde_json::json!({
+                "sessionUpdate":"tool_call", "toolCallId":"bash-7", "title":"Run", "status":"pending"
+            })),
+        );
+        match call.transcript.unwrap() {
+            TranscriptDelta::Append { row } => {
+                assert_eq!(row.id, "tool:bash-7");
+                assert_eq!(row.data["status"], "completed");
+            }
+            other => panic!("expected append, got {other:?}"),
+        }
+        assert!(projection.state().turn.active_tool_calls.is_empty());
+    }
+
+    #[test]
+    fn unknown_tool_stays_generic_and_keeps_its_raw_name() {
+        let mut projection = Projection::default();
+        let update = projection.apply(
+            1,
+            &event(serde_json::json!({
+                "sessionUpdate":"tool_call", "toolCallId":"custom-1", "title":"Custom",
+                "name":"vendor.custom", "kind":"future_kind"
+            })),
+        );
+        match update.transcript.unwrap() {
+            TranscriptDelta::Append { row } => {
+                assert_eq!(row.kind, TranscriptRowKind::ToolCall);
+                assert_eq!(row.data["name"], "vendor.custom");
+            }
+            other => panic!("expected generic tool row, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn folds_usage_mode_model_and_plan_into_whole_state() {
+        let mut projection = Projection::default();
+        for (seq, update) in [
+            serde_json::json!({"sessionUpdate":"usage_update","used":12,"size":200}),
+            serde_json::json!({"sessionUpdate":"current_mode_update","currentModeId":"plan"}),
+            serde_json::json!({"sessionUpdate":"config_option_update","configOptions":[{"id":"model","name":"Model","category":"model","type":"select","currentValue":"opus","options":[]}]}),
+            serde_json::json!({"sessionUpdate":"plan","entries":[]}),
+        ]
+        .into_iter()
+        .enumerate()
+        {
+            projection.apply((seq + 1) as u64, &event(update));
+        }
+        let state = projection.state();
+        assert_eq!(state.usage.as_ref().unwrap()["used"], 12);
+        assert_eq!(state.context.as_ref().unwrap()["size"], 200);
+        assert_eq!(state.mode.as_deref(), Some("plan"));
+        assert_eq!(state.model.as_deref(), Some("opus"));
+        assert_eq!(
+            state.plan.as_ref().unwrap()["entries"],
+            serde_json::json!([])
+        );
+    }
+
+    #[test]
+    fn plans_replace_one_stable_row_and_state() {
+        let mut projection = Projection::default();
+        let first = projection.apply(
+            1,
+            &event(serde_json::json!({"sessionUpdate":"plan","entries":[{"content":"first","priority":"high","status":"pending"}]})),
+        );
+        assert!(matches!(
+            first.transcript,
+            Some(TranscriptDelta::Append { .. })
+        ));
+        let replacement = projection.apply(
+            2,
+            &event(serde_json::json!({"sessionUpdate":"plan","entries":[{"content":"second","priority":"high","status":"completed"}]})),
+        );
+        match replacement.transcript.unwrap() {
+            TranscriptDelta::Patch { row } => {
+                assert_eq!(row.id, "plan");
+                assert_eq!(row.data["entries"][0]["content"], "second");
+            }
+            other => panic!("expected patch, got {other:?}"),
+        }
+        assert_eq!(projection.snapshot(2).rows.len(), 1);
+        assert_eq!(
+            projection.state().plan.unwrap()["entries"][0]["content"],
+            "second"
+        );
+    }
+
+    #[test]
+    fn turn_lifecycle_gates_chunks_and_normalizes_terminal_states() {
+        let mut projection = Projection::default();
+        projection.apply(
+            1,
+            &event(serde_json::json!({"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"replay"}})),
+        );
+        assert_eq!(projection.state().turn, Default::default());
+
+        let (_, _) = projection.begin_turn();
+        assert!(projection.state().turn.responding);
+        assert!(projection.state().turn.thinking);
+        projection.apply(
+            2,
+            &event(serde_json::json!({"sessionUpdate":"agent_thought_chunk","content":{"type":"text","text":"think"}})),
+        );
+        assert!(projection.state().turn.thinking);
+        projection.apply(
+            3,
+            &event(serde_json::json!({"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"answer"}})),
+        );
+        assert!(projection.state().turn.responding);
+        assert!(!projection.state().turn.thinking);
+
+        for terminal in [
+            TurnLifecycle::Finish,
+            TurnLifecycle::Cancel,
+            TurnLifecycle::Failure,
+        ] {
+            let (turn, _) = projection.begin_turn();
+            projection.turn_lifecycle(terminal, turn);
+            assert!(!projection.state().turn.responding);
+            assert!(!projection.state().turn.thinking);
+        }
+    }
+
+    fn terminal_turn_drops_unfinished_tools(
+        terminal: TurnLifecycle,
+        unfinished_id: &str,
+        next_id: &str,
+    ) {
+        let mut projection = Projection::default();
+        let (turn, _) = projection.begin_turn();
+        projection.apply(
+            1,
+            &event(serde_json::json!({
+                "sessionUpdate":"tool_call", "toolCallId":unfinished_id,
+                "title":"Unfinished", "status":"in_progress"
+            })),
+        );
+        assert_eq!(projection.state().turn.active_tool_calls, [unfinished_id]);
+
+        projection.turn_lifecycle(terminal, turn);
+        assert!(projection.state().turn.active_tool_calls.is_empty());
+        assert_eq!(projection.snapshot(1).rows.len(), 1);
+
+        projection.begin_turn();
+        projection.apply(
+            2,
+            &event(serde_json::json!({
+                "sessionUpdate":"tool_call", "toolCallId":next_id,
+                "title":"Next", "status":"pending"
+            })),
+        );
+        assert_eq!(projection.state().turn.active_tool_calls, [next_id]);
+    }
+
+    #[test]
+    fn cancelled_turn_does_not_reactivate_unfinished_tools() {
+        terminal_turn_drops_unfinished_tools(TurnLifecycle::Cancel, "cancelled-1", "next-1");
+    }
+
+    #[test]
+    fn failed_turn_does_not_reactivate_unfinished_tools() {
+        terminal_turn_drops_unfinished_tools(TurnLifecycle::Failure, "failed-1", "next-1");
+    }
+}

crates/maestro-daemon/src/sidecar.rs

@@ -5,9 +5,22 @@
 //! original `cwd`, command, env, metadata, etc. Files are deleted on
 //! `KillSession`. Sessions whose sidecar is missing at recovery time are still
 //! advertised but with empty metadata (handled by the caller, not here).
+//!
+//! A structured session (#135) also owns `<id>.events.jsonl` next to its
+//! sidecar: every `session/update` notification its adapter sent, one JSON
+//! object per line, appended before it is broadcast. The log is the source of
+//! truth for the transcript; `seq` is the line's 1-based position, so a
+//! client that holds seq `n` asks for [`EventCursor::Since`]`(n)`. A line
+//! that does not parse — including the partial line a hard kill leaves at
+//! the end — keeps its seq and is skipped on read. `delete` removes the log
+//! with the sidecar; `load_all` never reads one as a sidecar.
 
+use std::collections::VecDeque;
+use std::fs::{File, OpenOptions};
+use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
 
+use maestro_acp::schema::SessionNotification;
 use maestro_protocol::Session;
 use uuid::Uuid;
 
@@ -38,6 +51,120 @@ impl SidecarStore {
         self.dir.join(format!("{id}.json"))
     }
 
+    /// Where the per-session event log lives (#135).
+    pub fn events_path_for(&self, id: Uuid) -> PathBuf {
+        self.dir.join(format!("{id}.events.jsonl"))
+    }
+
+    /// Durable evidence that graceful daemon shutdown ended this structured
+    /// session and the next daemon may resume it.
+    pub fn recovery_path_for(&self, id: Uuid) -> PathBuf {
+        self.dir.join(format!("{id}.recover"))
+    }
+
+    /// Mark `id` as eligible for one structured-session recovery attempt.
+    pub fn mark_for_recovery(&self, id: Uuid) -> Result<(), SidecarError> {
+        std::fs::create_dir_all(&self.dir)?;
+        std::fs::write(self.recovery_path_for(id), [])?;
+        Ok(())
+    }
+
+    /// Consume the graceful-shutdown recovery marker for `id`.
+    pub fn take_recovery_marker(&self, id: Uuid) -> Result<bool, SidecarError> {
+        match std::fs::remove_file(self.recovery_path_for(id)) {
+            Ok(()) => Ok(true),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
+            Err(error) => Err(error.into()),
+        }
+    }
+
+    /// Open the event log for `id` for appending, creating it (and the
+    /// directory) when missing. One handle per live session: the caller
+    /// keeps it for the session's lifetime rather than reopening per
+    /// notification.
+    ///
+    /// A log whose last line has no newline — what a hard kill mid-write
+    /// leaves behind — is terminated first, so the next append starts a
+    /// fresh line instead of gluing onto the fragment. The fragment keeps
+    /// the seq of its line and reads as malformed.
+    pub fn open_events(&self, id: Uuid) -> Result<EventLog, SidecarError> {
+        std::fs::create_dir_all(&self.dir)?;
+        let mut file = OpenOptions::new()
+            .create(true)
+            .read(true)
+            .append(true)
+            .open(self.events_path_for(id))?;
+        let mut lines = 0u64;
+        let mut last = None;
+        let mut reader = BufReader::new(&mut file);
+        let mut buf = [0u8; 64 * 1024];
+        loop {
+            let n = reader.read(&mut buf)?;
+            if n == 0 {
+                break;
+            }
+            lines += buf[..n].iter().filter(|&&b| b == b'\n').count() as u64;
+            last = Some(buf[n - 1]);
+        }
+        if last.is_some_and(|b| b != b'\n') {
+            file.write_all(b"\n")?;
+            lines += 1;
+        }
+        file.seek(SeekFrom::End(0))?;
+        Ok(EventLog {
+            file,
+            last_seq: lines,
+        })
+    }
+
+    /// Read events from the log for `id`, at most `limit`, in ascending seq
+    /// order. A missing log is an empty log. Lines that fail to parse are
+    /// skipped but still consume a seq, so cursors stay stable.
+    pub fn read_events(
+        &self,
+        id: Uuid,
+        cursor: EventCursor,
+        limit: usize,
+    ) -> Result<Vec<LoggedEvent>, SidecarError> {
+        let file = match File::open(self.events_path_for(id)) {
+            Ok(f) => f,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+            Err(e) => return Err(e.into()),
+        };
+        if limit == 0 {
+            return Ok(Vec::new());
+        }
+        let mut out: VecDeque<LoggedEvent> = VecDeque::new();
+        let mut seq = 0u64;
+        for line in BufReader::new(file).split(b'\n') {
+            let line = line?;
+            seq += 1;
+            let keep = match cursor {
+                EventCursor::Since(since) => seq > since,
+                EventCursor::Before(Some(before)) => seq < before,
+                EventCursor::Before(None) => true,
+            };
+            if !keep {
+                if matches!(cursor, EventCursor::Before(Some(_))) {
+                    break;
+                }
+                continue;
+            }
+            let Ok(notification) = serde_json::from_slice::<SessionNotification>(&line) else {
+                continue;
+            };
+            out.push_back(LoggedEvent { seq, notification });
+            match cursor {
+                EventCursor::Since(_) if out.len() >= limit => break,
+                EventCursor::Before(_) if out.len() > limit => {
+                    out.pop_front();
+                }
+                _ => {}
+            }
+        }
+        Ok(out.into())
+    }
+
     /// Atomically write `session` to `<dir>/<id>.json`. Creates the directory
     /// if it does not exist.
     pub fn write(&self, session: &Session) -> Result<(), SidecarError> {
@@ -59,13 +186,12 @@ impl SidecarStore {
         }
     }
 
-    /// Delete the sidecar for `id`. No-op if it does not exist.
+    /// Delete the sidecar, event log, and recovery marker for `id`. No-op for
+    /// whichever does not exist.
     pub fn delete(&self, id: Uuid) -> Result<(), SidecarError> {
-        match std::fs::remove_file(self.path_for(id)) {
+        remove_if_present(&self.events_path_for(id))?;
-            Ok(()) => Ok(()),
+        remove_if_present(&self.recovery_path_for(id))?;
-            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
+        remove_if_present(&self.path_for(id))
-            Err(e) => Err(e.into()),
-        }
     }
 
     /// Read every sidecar JSON in the directory. Files that fail to parse are
@@ -81,6 +207,8 @@ impl SidecarStore {
         for entry in entries {
             let entry = entry?;
             let path = entry.path();
+            // `.events.jsonl` logs (#135) sit in the same directory and are
+            // never sidecars.
             if path.extension().and_then(|s| s.to_str()) != Some("json") {
                 continue;
             }
@@ -95,3 +223,61 @@ impl SidecarStore {
         Ok(out)
     }
 }
+
+fn remove_if_present(path: &Path) -> Result<(), SidecarError> {
+    match std::fs::remove_file(path) {
+        Ok(()) => Ok(()),
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
+        Err(e) => Err(e.into()),
+    }
+}
+
+/// Which slice of a session's event log to read.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum EventCursor {
+    /// Forward: events with `seq` strictly greater than the cursor, the
+    /// first `limit` of them. `Since(0)` is the whole log from the start.
+    /// What a client catching up after a reconnect asks for.
+    Since(u64),
+    /// Backward: events with `seq` strictly less than the cursor, the last
+    /// `limit` of them, still returned in ascending order. `None` means the
+    /// tail of the log, for a first paint. The next page passes the smallest
+    /// seq it received.
+    Before(Option<u64>),
+}
+
+/// One line of a session's event log.
+#[derive(Debug, Clone, PartialEq, serde::Serialize)]
+pub struct LoggedEvent {
+    /// 1-based line position in the log; the cursor for [`EventCursor`].
+    pub seq: u64,
+    pub notification: SessionNotification,
+}
+
+/// An open, append-only handle on one session's event log. Obtained from
+/// [`SidecarStore::open_events`]; held for the session's lifetime.
+#[derive(Debug)]
+pub struct EventLog {
+    file: File,
+    last_seq: u64,
+}
+
+impl EventLog {
+    /// The seq of the last line, `0` for an empty log.
+    pub fn last_seq(&self) -> u64 {
+        self.last_seq
+    }
+
+    /// Append one notification as a line and return its seq. The line is
+    /// handed to the kernel in a single write before this returns, so a
+    /// caller that broadcasts afterwards never announces an event the log
+    /// would not replay — short of the machine losing power.
+    pub fn append(&mut self, notification: &SessionNotification) -> Result<u64, SidecarError> {
+        let mut line = serde_json::to_vec(notification)?;
+        line.push(b'\n');
+        self.file.write_all(&line)?;
+        self.file.flush()?;
+        self.last_seq += 1;
+        Ok(self.last_seq)
+    }
+}

crates/maestro-daemon/tests/acp.rs

@@ -0,0 +1,848 @@
+//! Structured sessions (#134): a `claude-code-acp` child per session instead
+//! of a tmux pane. Runs against the fake adapter in
+//! `crates/maestro-acp/tests/fake_acp_agent.py`; nothing here needs Claude.
+
+mod common;
+
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use common::{init_bare_repo, init_repo, push_all_to_origin, wait_for_async, DaemonFixture};
+use maestro_acp::schema::{ContentBlock, SessionUpdate};
+use maestro_acp::{ADAPTER_BIN, INSTALL_COMMAND};
+use maestro_daemon::projection::TurnLifecycle;
+use maestro_daemon::sidecar::{EventCursor, SidecarStore};
+use maestro_daemon::worktree::WorktreeManager;
+use maestro_daemon::SessionManager;
+use maestro_protocol::{
+    ActivityState, CreateSessionRequest, SessionStatus, SessionView, TranscriptDelta,
+    TranscriptRowKind, WorktreeSpec,
+};
+use tempfile::TempDir;
+
+fn fake_agent() -> PathBuf {
+    Path::new(env!("CARGO_MANIFEST_DIR")).join("../maestro-acp/tests/fake_acp_agent.py")
+}
+
+/// The fake adapter installed as `claude-code-acp` in a temp `bin/`, with
+/// the log file it appends to. `env()` is what a spawn request carries.
+struct Fake {
+    dir: TempDir,
+}
+
+impl Fake {
+    fn new() -> Self {
+        let dir = TempDir::new().unwrap();
+        let bin = dir.path().join("bin");
+        std::fs::create_dir_all(&bin).unwrap();
+        std::fs::copy(fake_agent(), bin.join(ADAPTER_BIN)).unwrap();
+        Self { dir }
+    }
+
+    fn log_path(&self) -> PathBuf {
+        self.dir.path().join("agent.log")
+    }
+
+    fn log(&self) -> String {
+        std::fs::read_to_string(self.log_path()).unwrap_or_default()
+    }
+
+    /// Fake `bin/` first so it shadows any real adapter, then the process
+    /// `PATH` so the script's `python3` still resolves.
+    fn env(&self) -> HashMap<String, String> {
+        let path = format!(
+            "{}:{}",
+            self.dir.path().join("bin").display(),
+            std::env::var("PATH").unwrap_or_default()
+        );
+        HashMap::from([
+            ("PATH".to_string(), path),
+            (
+                "FAKE_ACP_LOG".to_string(),
+                self.log_path().to_string_lossy().into_owned(),
+            ),
+        ])
+    }
+
+    /// The pid the fake logged at start, once it has.
+    fn pid(&self) -> u32 {
+        self.log()
+            .lines()
+            .filter_map(|line| {
+                line.strip_prefix("start pid=")?
+                    .split(' ')
+                    .next()?
+                    .parse()
+                    .ok()
+            })
+            .next_back()
+            .expect("pid in log")
+    }
+}
+
+fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
+    match &event.notification.update {
+        SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
+            ContentBlock::Text(text) => &text.text,
+            other => panic!("unexpected content {other:?}"),
+        },
+        other => panic!("unexpected update {other:?}"),
+    }
+}
+
+fn structured(cwd: &Path, env: HashMap<String, String>) -> CreateSessionRequest {
+    CreateSessionRequest {
+        view: SessionView::Structured,
+        cwd: cwd.to_path_buf(),
+        command: "claude".into(),
+        env,
+        metadata: HashMap::new(),
+        worktree: None,
+    }
+}
+
+fn pid_alive(pid: u32) -> bool {
+    std::fs::read_to_string(format!("/proc/{pid}/stat"))
+        .map(|s| !s.contains(") Z "))
+        .unwrap_or(false)
+}
+
+async fn wait_for_exit(pid: u32) {
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while pid_alive(pid) && Instant::now() < deadline {
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    assert!(!pid_alive(pid), "pid {pid} still alive");
+}
+
+#[tokio::test]
+async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_ECHO_ENV".into(), "MAESTRO_SESSION_ID".into());
+
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .expect("create");
+
+    assert_eq!(session.view, SessionView::Structured);
+    assert_eq!(session.status, SessionStatus::Running);
+    let acp_id = session.acp_session_id.clone().expect("acp session id");
+    assert!(acp_id.starts_with("fake-"), "{acp_id}");
+
+    // The child ran in the session cwd with the session's env — including
+    // the identity the daemon injects, so the agent knows which session it
+    // is (#103).
+    let log = fake.log();
+    let cwd_canon = cwd.path().canonicalize().unwrap();
+    assert!(
+        log.contains(&format!("cwd={}", cwd_canon.display())),
+        "{log}"
+    );
+    assert!(
+        log.contains(&format!("env MAESTRO_SESSION_ID={}", session.id)),
+        "{log}"
+    );
+
+    // No tmux pane was created for it: the agent is the daemon's child.
+    let panes = fx.tmux.tmux.list_sessions().expect("list");
+    assert!(
+        !panes.contains(&session.id),
+        "structured session got a tmux pane"
+    );
+
+    // The sidecar carries the id a respawn will pass to session/load (#135).
+    let sidecar: maestro_protocol::Session = serde_json::from_slice(
+        &std::fs::read(fx.sidecar_dir().join(format!("{}.json", session.id))).unwrap(),
+    )
+    .unwrap();
+    assert_eq!(sidecar.acp_session_id.as_deref(), Some(acp_id.as_str()));
+
+    wait_for_async(Duration::from_secs(5), || async {
+        fx.manager
+            .read_acp_events(session.id, EventCursor::Since(0), 10)
+            .is_ok_and(|events| !events.is_empty())
+    })
+    .await;
+
+    // Live delivery happens only after the notification is durable.
+    let (replay, mut updates) = fx
+        .manager
+        .subscribe_acp_events_with_replay(session.id, EventCursor::Since(0), 10)
+        .await
+        .expect("replay read")
+        .expect("event subscription");
+    assert_eq!(replay.len(), 1);
+    assert!(updates.try_recv().is_err(), "replay duplicated onto live");
+    unsafe {
+        libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1);
+    }
+    let live = tokio::time::timeout(Duration::from_secs(5), updates.recv())
+        .await
+        .expect("an update within 5s")
+        .expect("broadcast open");
+    assert_eq!(event_text(&live), "tick 1");
+    let stored = fx
+        .manager
+        .read_acp_events(session.id, EventCursor::Since(live.seq - 1), 1)
+        .expect("read event logged before broadcast");
+    assert_eq!(stored, vec![live]);
+
+    fx.manager.kill(session.id, false).await.expect("kill");
+    assert!(!fx
+        .sidecar_dir()
+        .join(format!("{}.events.jsonl", session.id))
+        .exists());
+}
+
+#[tokio::test]
+async fn projection_snapshot_and_live_stream_share_one_fold() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), fake.env()))
+        .await
+        .expect("create");
+
+    wait_for_async(Duration::from_secs(5), || async {
+        fx.manager
+            .read_acp_events(session.id, EventCursor::Since(0), 10)
+            .is_ok_and(|events| events.len() == 1)
+    })
+    .await;
+
+    // The cursor does not truncate either projection: a reconnect needs the
+    // source row for future patches and whole reduced state from the session.
+    let (snapshot, state, mut live) = fx
+        .manager
+        .subscribe_acp_projection(session.id, EventCursor::Since(1))
+        .await
+        .expect("projection subscription")
+        .expect("structured session");
+    assert_eq!(snapshot.cursor, 1);
+    assert_eq!(snapshot.rows.len(), 1);
+    assert_eq!(snapshot.rows[0].kind, TranscriptRowKind::AgentText);
+    assert_eq!(state.turn, Default::default());
+
+    let maestro_daemon::session::BeginAcpTurn::Started(_, turn) = fx
+        .manager
+        .begin_acp_turn(session.id)
+        .await
+        .expect("begin lifecycle")
+    else {
+        panic!("structured turn should start");
+    };
+    let begin = tokio::time::timeout(Duration::from_secs(5), live.recv())
+        .await
+        .expect("begin lifecycle broadcast")
+        .expect("projection broadcast");
+    assert_eq!(begin.transcript, None);
+    assert!(begin.state.turn.responding);
+    assert!(begin.state.turn.thinking);
+
+    unsafe {
+        libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1);
+    }
+    let agent_update = tokio::time::timeout(Duration::from_secs(5), live.recv())
+        .await
+        .expect("live projection")
+        .expect("projection broadcast");
+    match agent_update.transcript {
+        Some(TranscriptDelta::Append { row }) => {
+            assert_eq!(row.kind, TranscriptRowKind::AgentText);
+            assert_eq!(row.data["content"]["text"], "tick 1");
+        }
+        other => panic!("expected one appended transcript row, got {other:?}"),
+    }
+    assert!(agent_update.state.turn.responding);
+    assert!(!agent_update.state.turn.thinking);
+
+    assert!(fx
+        .manager
+        .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish, turn)
+        .await
+        .expect("finish lifecycle"));
+    let finish = tokio::time::timeout(Duration::from_secs(5), live.recv())
+        .await
+        .expect("finish lifecycle broadcast")
+        .expect("projection broadcast");
+    assert_eq!(finish.transcript, None);
+    assert_eq!(finish.state.turn, Default::default());
+
+    fx.manager.kill(session.id, false).await.expect("kill");
+}
+
+#[tokio::test]
+async fn stale_turn_completion_cannot_end_the_next_turn() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), fake.env()))
+        .await
+        .unwrap();
+
+    let maestro_daemon::session::BeginAcpTurn::Started(_, first) =
+        fx.manager.begin_acp_turn(session.id).await.unwrap()
+    else {
+        panic!("first turn should start");
+    };
+    assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
+    let maestro_daemon::session::BeginAcpTurn::Started(_, second) =
+        fx.manager.begin_acp_turn(session.id).await.unwrap()
+    else {
+        panic!("second turn should start");
+    };
+    assert!(second > first);
+    assert!(!fx
+        .manager
+        .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish, first)
+        .await
+        .unwrap());
+    assert!(matches!(
+        fx.manager.begin_acp_turn(session.id).await.unwrap(),
+        maestro_daemon::session::BeginAcpTurn::InFlight
+    ));
+    assert!(fx
+        .manager
+        .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish, second)
+        .await
+        .unwrap());
+
+    fx.manager.kill(session.id, false).await.unwrap();
+}
+
+#[tokio::test]
+async fn recovery_loads_the_acp_session_without_duplicating_history() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_LOAD_REPLAY".into(), "2".into());
+
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .expect("create");
+    let acp_id = session.acp_session_id.clone().expect("acp id");
+    wait_for_async(Duration::from_secs(5), || async {
+        fx.manager
+            .read_acp_events(session.id, EventCursor::Since(0), 10)
+            .is_ok_and(|events| events.len() == 1)
+    })
+    .await;
+    unsafe {
+        libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1);
+    }
+    wait_for_async(Duration::from_secs(5), || async {
+        fx.manager
+            .read_acp_events(session.id, EventCursor::Since(0), 10)
+            .is_ok_and(|events| events.len() == 2)
+    })
+    .await;
+
+    let old_pid = fake.pid();
+    fx.manager.shutdown().await;
+    wait_for_exit(old_pid).await;
+    assert!(
+        fx.sidecar_dir()
+            .join(format!("{}.recover", session.id))
+            .exists(),
+        "graceful shutdown must record recovery provenance"
+    );
+
+    let revived = SessionManager::new(
+        Arc::new(fx.tmux.tmux.clone()),
+        SidecarStore::new(fx.sidecar_dir()),
+        WorktreeManager::new(None),
+        fx.fifo_dir(),
+    );
+    assert_eq!(revived.recover().await.expect("recover"), 1);
+    assert_eq!(
+        revived.status(session.id).await,
+        Some(SessionStatus::Running)
+    );
+    let log = fake.log();
+    assert!(
+        log.contains(&format!("recv session/load sessionId={acp_id}")),
+        "{log}"
+    );
+    wait_for_async(Duration::from_secs(5), || async {
+        revived
+            .read_acp_events(session.id, EventCursor::Since(0), 10)
+            .is_ok_and(|events| events.len() == 3)
+    })
+    .await;
+
+    let (replay, mut live) = revived
+        .subscribe_acp_events_with_replay(session.id, EventCursor::Since(0), 10)
+        .await
+        .expect("replay read")
+        .expect("recovered event subscription");
+    assert_eq!(replay.len(), 3);
+    assert!(live.try_recv().is_err(), "replay duplicated onto live");
+    let new_pid = fake.pid();
+    assert_ne!(new_pid, old_pid);
+    unsafe {
+        libc::kill(new_pid as libc::pid_t, libc::SIGUSR1);
+    }
+    let event = tokio::time::timeout(Duration::from_secs(5), live.recv())
+        .await
+        .expect("live event after recovery")
+        .expect("broadcast open");
+    assert_eq!(event.seq, 4);
+    assert_eq!(event_text(&event), "tick 1");
+
+    let stored = revived
+        .read_acp_events(session.id, EventCursor::Since(0), 10)
+        .expect("recovered log");
+    assert_eq!(stored.len(), 4, "session/load replay must not be appended");
+    assert_eq!(
+        stored.iter().map(event_text).collect::<Vec<_>>(),
+        ["hello from fake", "tick 1", "hello after load", "tick 1"]
+    );
+
+    revived.kill(session.id, false).await.expect("kill");
+}
+
+#[tokio::test]
+async fn shutdown_preserves_a_pristine_structured_worktree_for_recovery() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let repo = TempDir::new().unwrap();
+    init_repo(repo.path());
+    let origin = TempDir::new().unwrap();
+    init_bare_repo(origin.path());
+
+    let branch = "feature/restart";
+    let worktree = repo
+        .path()
+        .join("worktrees")
+        .join("feature")
+        .join("restart");
+    let session = fx
+        .manager
+        .create(CreateSessionRequest {
+            view: SessionView::Structured,
+            cwd: repo.path().to_path_buf(),
+            command: "claude".into(),
+            env: fake.env(),
+            metadata: HashMap::new(),
+            worktree: Some(WorktreeSpec {
+                base_repo: repo.path().to_path_buf(),
+                branch: branch.into(),
+                create_branch: true,
+            }),
+        })
+        .await
+        .expect("create");
+    assert_eq!(session.cwd, worktree);
+    push_all_to_origin(repo.path(), origin.path());
+
+    fx.manager.shutdown().await;
+
+    assert!(
+        worktree.is_dir(),
+        "shutdown must not settle a pristine worktree"
+    );
+    let sidecar: maestro_protocol::Session = serde_json::from_slice(
+        &std::fs::read(fx.sidecar_dir().join(format!("{}.json", session.id))).unwrap(),
+    )
+    .unwrap();
+    assert_eq!(
+        sidecar.worktree.as_ref().map(|wt| &wt.path),
+        Some(&worktree)
+    );
+
+    let revived = SessionManager::new(
+        Arc::new(fx.tmux.tmux.clone()),
+        SidecarStore::new(fx.sidecar_dir()),
+        WorktreeManager::new(None),
+        fx.fifo_dir(),
+    );
+    assert_eq!(revived.recover().await.expect("recover"), 1);
+    assert_eq!(
+        revived.status(session.id).await,
+        Some(SessionStatus::Running)
+    );
+    assert!(fake.log().contains("recv session/load"), "{}", fake.log());
+    assert!(worktree.is_dir(), "recovery must retain the worktree");
+
+    revived.kill(session.id, false).await.expect("dismiss");
+    assert!(
+        !worktree.exists(),
+        "operator dismissal still cleans a pristine tree"
+    );
+}
+
+#[tokio::test]
+async fn failed_recovery_cleans_its_fifo() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), fake.env()))
+        .await
+        .expect("create");
+    let old_pid = fake.pid();
+    fx.manager.shutdown().await;
+    wait_for_exit(old_pid).await;
+
+    let events_path = fx
+        .sidecar_dir()
+        .join(format!("{}.events.jsonl", session.id));
+    std::fs::remove_file(&events_path).unwrap();
+    std::fs::create_dir(&events_path).unwrap();
+    let revived = SessionManager::new(
+        Arc::new(fx.tmux.tmux.clone()),
+        SidecarStore::new(fx.sidecar_dir()),
+        WorktreeManager::new(None),
+        fx.fifo_dir(),
+    );
+
+    assert!(revived.recover().await.is_err());
+    assert!(!fx.fifo_dir().join(format!("{}.fifo", session.id)).exists());
+    let failed_pid = fake.pid();
+    assert_ne!(failed_pid, old_pid);
+    wait_for_exit(failed_pid).await;
+}
+
+#[tokio::test]
+async fn missing_adapter_yields_a_failed_session_naming_the_install_command() {
+    let fx = DaemonFixture::new();
+    let empty = TempDir::new().unwrap();
+    let cwd = TempDir::new().unwrap();
+    let env = HashMap::from([(
+        "PATH".to_string(),
+        empty.path().to_string_lossy().into_owned(),
+    )]);
+
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .expect("a failed session, not an error");
+
+    assert_eq!(session.status, SessionStatus::Dead);
+    assert_eq!(session.activity, ActivityState::Error);
+    let detail = session.activity_detail.clone().expect("actionable detail");
+    assert!(detail.contains(ADAPTER_BIN), "{detail}");
+    assert!(detail.contains(INSTALL_COMMAND), "{detail}");
+    assert_eq!(session.acp_session_id, None);
+
+    // It is a real session: listed, on disk, and dismissable like any other
+    // dead one.
+    assert!(fx.manager.has(session.id).await);
+    assert!(fx
+        .sidecar_dir()
+        .join(format!("{}.json", session.id))
+        .exists());
+    fx.manager.kill(session.id, false).await.expect("dismiss");
+    assert!(!fx.manager.has(session.id).await);
+}
+
+#[tokio::test]
+async fn kill_sends_cancel_then_sigterm_and_removes_the_sidecar() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut events = fx.manager.events();
+
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), fake.env()))
+        .await
+        .expect("create");
+    let pid = fake.pid();
+    assert!(pid_alive(pid));
+
+    fx.manager.kill(session.id, false).await.expect("kill");
+
+    wait_for_exit(pid).await;
+    let log = fake.log();
+    let cancel = log.find("recv session/cancel").expect("cancel logged");
+    let term = log.find("signal SIGTERM").expect("SIGTERM logged");
+    assert!(cancel < term, "cancel must precede SIGTERM:\n{log}");
+    assert!(!fx.manager.has(session.id).await);
+    assert!(!fx
+        .sidecar_dir()
+        .join(format!("{}.json", session.id))
+        .exists());
+
+    let mut removed = false;
+    while let Ok(ev) = events.try_recv() {
+        if matches!(ev, maestro_protocol::CatalogEvent::SessionRemoved { id } if id == session.id) {
+            removed = true;
+        }
+    }
+    assert!(removed, "SessionRemoved never broadcast");
+}
+
+#[tokio::test]
+async fn shutdown_reaps_every_adapter_child() {
+    let fx = DaemonFixture::new();
+    let cwd = TempDir::new().unwrap();
+    let a = Fake::new();
+    let b = Fake::new();
+
+    let sa = fx
+        .manager
+        .create(structured(cwd.path(), a.env()))
+        .await
+        .expect("create a");
+    let sb = fx
+        .manager
+        .create(structured(cwd.path(), b.env()))
+        .await
+        .expect("create b");
+    let (pa, pb) = (a.pid(), b.pid());
+    assert!(pid_alive(pa) && pid_alive(pb));
+
+    fx.manager.shutdown().await;
+
+    wait_for_exit(pa).await;
+    wait_for_exit(pb).await;
+    assert!(a.log().contains("signal SIGTERM"), "{}", a.log());
+    assert!(b.log().contains("signal SIGTERM"), "{}", b.log());
+    // Still listed, now Dead: the sidecar is truthful for whoever reads it
+    // after the restart (#135).
+    assert_eq!(fx.manager.status(sa.id).await, Some(SessionStatus::Dead));
+    assert_eq!(fx.manager.status(sb.id).await, Some(SessionStatus::Dead));
+    let sidecar: maestro_protocol::Session = serde_json::from_slice(
+        &std::fs::read(fx.sidecar_dir().join(format!("{}.json", sa.id))).unwrap(),
+    )
+    .unwrap();
+    assert_eq!(sidecar.status, SessionStatus::Dead);
+}
+
+#[tokio::test]
+async fn dismissal_during_shutdown_does_not_restore_recovery_artifacts() {
+    let fx = DaemonFixture::new();
+    let dismissed = Fake::new();
+    let blocker = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = dismissed.env();
+    env.insert("FAKE_ACP_EXIT_ON_CANCEL".into(), "1".into());
+
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .expect("create");
+    fx.manager
+        .create(structured(cwd.path(), blocker.env()))
+        .await
+        .expect("create blocker");
+
+    let manager = fx.manager.clone();
+    let shutdown = tokio::spawn(async move { manager.shutdown().await });
+    wait_for_async(Duration::from_secs(5), || async {
+        dismissed.log().contains("recv session/cancel")
+    })
+    .await;
+
+    fx.manager.kill(session.id, false).await.expect("dismiss");
+    assert!(
+        !fx.sidecar_dir()
+            .join(format!("{}.json", session.id))
+            .exists(),
+        "dismissal must finish before shutdown settles its stale snapshot"
+    );
+    shutdown.await.expect("shutdown task");
+
+    assert!(!fx.manager.has(session.id).await);
+    assert!(
+        !fx.sidecar_dir()
+            .join(format!("{}.recover", session.id))
+            .exists(),
+        "dismissal must leave no recovery marker"
+    );
+    assert!(
+        !fx.sidecar_dir()
+            .join(format!("{}.json", session.id))
+            .exists(),
+        "dismissal must leave no sidecar"
+    );
+    assert!(
+        !fx.sidecar_dir()
+            .join(format!("{}.events.jsonl", session.id))
+            .exists(),
+        "dismissal must leave no event log"
+    );
+}
+
+#[tokio::test]
+async fn shutdown_does_not_recover_an_adapter_that_already_exited() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_EXIT_AFTER_NEW".into(), "1".into());
+
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .expect("create");
+    wait_for_exit(fake.pid()).await;
+
+    fx.manager.shutdown().await;
+    assert!(
+        !fx.sidecar_dir()
+            .join(format!("{}.recover", session.id))
+            .exists(),
+        "a naturally exited adapter is not recoverable"
+    );
+
+    let revived = SessionManager::new(
+        Arc::new(fx.tmux.tmux.clone()),
+        SidecarStore::new(fx.sidecar_dir()),
+        WorktreeManager::new(None),
+        fx.fifo_dir(),
+    );
+    assert_eq!(revived.recover().await.expect("recover"), 1);
+    assert_eq!(revived.status(session.id).await, Some(SessionStatus::Dead));
+    assert!(
+        !fake.log().contains("recv session/load"),
+        "a naturally dead adapter must not be resumed: {}",
+        fake.log()
+    );
+}
+
+#[tokio::test]
+async fn shutdown_waits_for_structured_creation_and_rejects_new_creates() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_DELAY_SESSION_NEW_MS".into(), "500".into());
+
+    let create_cwd = cwd.path().to_path_buf();
+    let manager = fx.manager.clone();
+    let create = tokio::spawn(async move { manager.create(structured(&create_cwd, env)).await });
+    wait_for_async(Duration::from_secs(5), || async {
+        fake.log().contains("recv session/new")
+    })
+    .await;
+
+    let manager = fx.manager.clone();
+    let shutdown = tokio::spawn(async move { manager.shutdown().await });
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    assert!(
+        !shutdown.is_finished(),
+        "shutdown returned before create finished"
+    );
+
+    let session = create.await.expect("create task").expect("create");
+    let pid = fake.pid();
+    shutdown.await.expect("shutdown task");
+
+    wait_for_exit(pid).await;
+    assert!(fake.log().contains("signal SIGTERM"), "{}", fake.log());
+    assert_eq!(
+        fx.manager.status(session.id).await,
+        Some(SessionStatus::Dead)
+    );
+    assert!(matches!(
+        fx.manager.create(structured(cwd.path(), fake.env())).await,
+        Err(maestro_daemon::SessionError::ShuttingDown)
+    ));
+
+    // A repeated signal must not attempt to terminate the child again.
+    fx.manager.shutdown().await;
+    assert_eq!(fake.log().matches("signal SIGTERM").count(), 1);
+}
+
+#[tokio::test]
+async fn shutdown_drains_final_notifications_to_the_log() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_EMIT_ON_CANCEL".into(), "1".into());
+
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .expect("create");
+    wait_for_async(Duration::from_secs(5), || async {
+        fx.manager
+            .read_acp_events(session.id, EventCursor::Since(0), 10)
+            .is_ok_and(|events| events.len() == 1)
+    })
+    .await;
+
+    fx.manager.shutdown().await;
+
+    let stored = fx
+        .manager
+        .read_acp_events(session.id, EventCursor::Since(0), 10)
+        .expect("event log after shutdown");
+    assert_eq!(
+        stored.iter().map(event_text).collect::<Vec<_>>(),
+        ["hello from fake", "final on cancel"]
+    );
+}
+
+#[tokio::test]
+async fn adapter_exit_marks_the_session_dead_and_keeps_it_listed() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), fake.env()))
+        .await
+        .expect("create");
+    fx.manager.start_liveness_watcher(Duration::from_millis(50));
+
+    // The agent dies out from under the daemon: the liveness pass, not tmux,
+    // is what notices for a structured session.
+    let pid = fake.pid();
+    unsafe {
+        libc::kill(pid as libc::pid_t, libc::SIGTERM);
+    }
+    wait_for_exit(pid).await;
+
+    wait_for_async(Duration::from_secs(3), || async {
+        fx.manager.status(session.id).await == Some(SessionStatus::Dead)
+    })
+    .await;
+    assert!(fx.manager.has(session.id).await);
+    assert!(
+        !fx.sidecar_dir()
+            .join(format!("{}.recover", session.id))
+            .exists(),
+        "a naturally exited adapter is never marked for recovery"
+    );
+
+    let revived = SessionManager::new(
+        Arc::new(fx.tmux.tmux.clone()),
+        SidecarStore::new(fx.sidecar_dir()),
+        WorktreeManager::new(None),
+        fx.fifo_dir(),
+    );
+    assert_eq!(revived.recover().await.expect("recover"), 1);
+    assert_eq!(revived.status(session.id).await, Some(SessionStatus::Dead));
+    assert!(
+        !fake.log().contains("recv session/load"),
+        "a naturally dead adapter must not be resumed: {}",
+        fake.log()
+    );
+
+    revived.kill(session.id, false).await.expect("dismiss");
+    assert!(!revived.has(session.id).await);
+}

crates/maestro-daemon/tests/activity_report.rs

@@ -42,6 +42,7 @@ async fn fixture(ttl: Duration) -> Fixture {
 async fn spawn_sleeping(manager: &SessionManager) -> Uuid {
     manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: "/tmp".into(),
             command: "sleep 60".into(),
             env: Default::default(),

crates/maestro-daemon/tests/attach.rs

@@ -75,6 +75,7 @@ async fn create_bash_session(h: &Harness) -> Uuid {
     let s = h
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd_path,
             command: "bash --noprofile --norc".into(),
             env: HashMap::new(),

crates/maestro-daemon/tests/autoname.rs

@@ -50,6 +50,7 @@ async fn spawn(manager: &SessionManager, name: Option<&str>) -> Uuid {
     };
     manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: "/tmp".into(),
             command: "sleep 60".into(),
             env: Default::default(),

crates/maestro-daemon/tests/blocked_work_cost.rs

@@ -19,6 +19,7 @@ use tempfile::TempDir;
 
 fn request(cwd: &std::path::Path, repo: &std::path::Path, branch: &str) -> CreateSessionRequest {
     CreateSessionRequest {
+        view: Default::default(),
         cwd: cwd.to_path_buf(),
         command: "sleep 30".into(),
         env: HashMap::new(),

crates/maestro-daemon/tests/local_api.rs

@@ -4,12 +4,16 @@ mod common;
 
 use std::collections::HashMap;
 use std::net::{Ipv4Addr, SocketAddr};
+use std::path::Path;
 use std::time::Duration;
 
 use common::DaemonFixture;
 use futures_util::{SinkExt, StreamExt};
 use maestro_daemon::local_api;
-use maestro_protocol::{ApiError, CreateSessionRequest, Session, WorktreeSpec};
+use maestro_protocol::{
+    ApiError, CreateSessionRequest, Session, SessionStatus, SessionView, WorktreeSpec,
+};
+use tempfile::TempDir;
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
 use tokio_tungstenite::tungstenite::Message as WsMessage;
@@ -86,6 +90,430 @@ fn ws_url(addr: SocketAddr, path: &str) -> String {
     format!("ws://{addr}{path}")
 }
 
+struct FakeAcp {
+    dir: TempDir,
+}
+
+impl FakeAcp {
+    fn new() -> Self {
+        use std::os::unix::fs::PermissionsExt;
+
+        let dir = TempDir::new().unwrap();
+        let bin = dir.path().join("bin");
+        std::fs::create_dir_all(&bin).unwrap();
+        let adapter = bin.join("claude-code-acp");
+        std::fs::copy(
+            Path::new(env!("CARGO_MANIFEST_DIR")).join("../maestro-acp/tests/fake_acp_agent.py"),
+            &adapter,
+        )
+        .unwrap();
+        std::fs::set_permissions(&adapter, std::fs::Permissions::from_mode(0o755)).unwrap();
+        Self { dir }
+    }
+
+    fn env(&self) -> HashMap<String, String> {
+        HashMap::from([
+            (
+                "PATH".into(),
+                format!(
+                    "{}:{}",
+                    self.dir.path().join("bin").display(),
+                    std::env::var("PATH").unwrap()
+                ),
+            ),
+            (
+                "FAKE_ACP_LOG".into(),
+                self.dir.path().join("agent.log").display().to_string(),
+            ),
+        ])
+    }
+
+    fn pid(&self) -> u32 {
+        std::fs::read_to_string(self.dir.path().join("agent.log"))
+            .unwrap_or_default()
+            .lines()
+            .filter_map(|line| {
+                line.strip_prefix("start pid=")?
+                    .split(' ')
+                    .next()?
+                    .parse()
+                    .ok()
+            })
+            .next_back()
+            .expect("fake ACP started")
+    }
+}
+
+async fn structured_session(h: &Harness, fake: &FakeAcp) -> Uuid {
+    structured_session_with_env(h, fake.env()).await
+}
+
+async fn structured_session_with_env(h: &Harness, env: HashMap<String, String>) -> Uuid {
+    h.manager
+        .create(CreateSessionRequest {
+            view: SessionView::Structured,
+            cwd: "/tmp".into(),
+            command: "claude".into(),
+            env,
+            metadata: HashMap::new(),
+            worktree: None,
+        })
+        .await
+        .unwrap()
+        .id
+}
+
+#[tokio::test]
+async fn structured_ws_sends_snapshot_then_live_projection() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+    let (mut ws, _) = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
+        .await
+        .unwrap();
+
+    let first = ws.next().await.unwrap().unwrap().into_text().unwrap();
+    let second = ws.next().await.unwrap().unwrap().into_text().unwrap();
+    assert_eq!(
+        serde_json::from_str::<serde_json::Value>(&first).unwrap()["type"],
+        "transcript_snapshot"
+    );
+    assert_eq!(
+        serde_json::from_str::<serde_json::Value>(&second).unwrap()["type"],
+        "reduced_state"
+    );
+
+    let response = reqwest::Client::new()
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "hello"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);
+    let live = tokio::time::timeout(Duration::from_secs(1), ws.next())
+        .await
+        .expect("projection update arrives on the subscribed stream")
+        .unwrap()
+        .unwrap()
+        .into_text()
+        .unwrap();
+    assert_eq!(
+        serde_json::from_str::<serde_json::Value>(&live).unwrap()["type"],
+        "projection_update"
+    );
+}
+
+#[tokio::test]
+async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_DELAY_PROMPT_MS".into(), "200".into());
+    let id = structured_session_with_env(&h, env).await;
+    let client = reqwest::Client::new();
+    let first = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "first"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(first.status(), reqwest::StatusCode::NO_CONTENT);
+    let second = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "second"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(second.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = second.json().await.unwrap();
+    assert_eq!(error.error.code, "turn_in_flight");
+
+    let cancel = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/cancel"),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(cancel.status(), reqwest::StatusCode::NO_CONTENT);
+    let next = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "next"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(next.status(), reqwest::StatusCode::NO_CONTENT);
+}
+
+#[tokio::test]
+async fn structured_prompt_failure_clears_the_turn() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_PROMPT_FAIL_ONCE".into(), "1".into());
+    let id = structured_session_with_env(&h, env).await;
+    let client = reqwest::Client::new();
+    let path = format!("/api/sessions/{id}/structured/prompt");
+    let (mut ws, _) = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
+        .await
+        .unwrap();
+    ws.next().await.unwrap().unwrap();
+    ws.next().await.unwrap().unwrap();
+    let first = client
+        .post(url(h.addr, &path))
+        .json(&serde_json::json!({"prompt":"fails"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(first.status(), reqwest::StatusCode::NO_CONTENT);
+    let mut saw_active_turn = false;
+    tokio::time::timeout(Duration::from_secs(1), async {
+        while let Some(Ok(message)) = ws.next().await {
+            let event: serde_json::Value =
+                serde_json::from_str(&message.into_text().unwrap()).unwrap();
+            let turn = &event["state"]["turn"];
+            if turn["responding"] == true {
+                saw_active_turn = true;
+            }
+            if saw_active_turn && turn["responding"] == false && turn["thinking"] == false {
+                return;
+            }
+        }
+        panic!("structured websocket closed before the failed turn cleared");
+    })
+    .await
+    .expect("failed turn clears promptly");
+    let second = client
+        .post(url(h.addr, &path))
+        .json(&serde_json::json!({"prompt":"available"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(second.status(), reqwest::StatusCode::NO_CONTENT);
+    assert!(second.bytes().await.unwrap().is_empty());
+}
+
+#[tokio::test]
+async fn structured_prompt_rejects_unknown_and_unavailable_sessions() {
+    let h = spawn("naps-laptop").await;
+    let client = reqwest::Client::new();
+    let prompt = |id| {
+        client
+            .post(url(
+                h.addr,
+                &format!("/api/sessions/{id}/structured/prompt"),
+            ))
+            .json(&serde_json::json!({"prompt": "hello"}))
+    };
+
+    let unknown = prompt(Uuid::new_v4()).send().await.unwrap();
+    assert_eq!(unknown.status(), reqwest::StatusCode::NOT_FOUND);
+    let error: ApiError = unknown.json().await.unwrap();
+    assert_eq!(error.error.code, "not_found");
+
+    let terminal = h
+        .manager
+        .create(CreateSessionRequest {
+            view: SessionView::Terminal,
+            cwd: "/tmp".into(),
+            command: "sh".into(),
+            env: HashMap::new(),
+            metadata: HashMap::new(),
+            worktree: None,
+        })
+        .await
+        .unwrap();
+    let non_structured = prompt(terminal.id).send().await.unwrap();
+    assert_eq!(non_structured.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = non_structured.json().await.unwrap();
+    assert_eq!(error.error.code, "not_structured");
+
+    let fake = FakeAcp::new();
+    let dead = structured_session(&h, &fake).await;
+    h.manager.shutdown().await;
+    assert_eq!(h.manager.status(dead).await, Some(SessionStatus::Dead));
+    let unavailable = prompt(dead).send().await.unwrap();
+    assert_eq!(unavailable.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = unavailable.json().await.unwrap();
+    assert_eq!(error.error.code, "not_structured");
+}
+
+#[tokio::test]
+async fn structured_replay_honors_cursors_and_limit() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+    let client = reqwest::Client::new();
+
+    let events: Vec<serde_json::Value> = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let events: Vec<serde_json::Value> = client
+                .get(url(
+                    h.addr,
+                    &format!("/api/sessions/{id}/structured/replay?since=0&limit=1"),
+                ))
+                .send()
+                .await
+                .unwrap()
+                .json()
+                .await
+                .unwrap();
+            if !events.is_empty() {
+                return events;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(events.len(), 1);
+    assert_eq!(events[0]["cursor"], 1);
+    assert!(events[0].get("notification").is_none());
+
+    let invalid = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/replay?since=0&before=2"),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(invalid.status(), reqwest::StatusCode::BAD_REQUEST);
+}
+
+#[tokio::test]
+async fn structured_replay_pages_projected_events_in_cursor_order() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+    let client = reqwest::Client::new();
+    unsafe { libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1) };
+    let events: Vec<serde_json::Value> = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let events: Vec<serde_json::Value> = client
+                .get(url(
+                    h.addr,
+                    &format!("/api/sessions/{id}/structured/replay?before=99&limit=2"),
+                ))
+                .send()
+                .await
+                .unwrap()
+                .json()
+                .await
+                .unwrap();
+            if events.len() == 2 {
+                return events;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(
+        events
+            .iter()
+            .map(|event| event["cursor"].as_u64().unwrap())
+            .collect::<Vec<_>>(),
+        vec![1, 2]
+    );
+    let boundary: Vec<serde_json::Value> = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/replay?before=2&limit=10"),
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert_eq!(boundary.len(), 1);
+    assert_eq!(boundary[0]["cursor"], 1);
+}
+
+#[tokio::test]
+async fn structured_replay_rejects_unknown_terminal_and_dead_sessions() {
+    let h = spawn("naps-laptop").await;
+    let client = reqwest::Client::new();
+    let unknown = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{}/structured/replay", Uuid::new_v4()),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(unknown.status(), reqwest::StatusCode::NOT_FOUND);
+    let terminal = h
+        .manager
+        .create(CreateSessionRequest {
+            view: SessionView::Terminal,
+            cwd: "/tmp".into(),
+            command: "sh".into(),
+            env: HashMap::new(),
+            metadata: HashMap::new(),
+            worktree: None,
+        })
+        .await
+        .unwrap();
+    let response = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{}/structured/replay", terminal.id),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
+
+    let fake = FakeAcp::new();
+    let dead = structured_session(&h, &fake).await;
+    h.manager.shutdown().await;
+    assert_eq!(h.manager.status(dead).await, Some(SessionStatus::Dead));
+    let response = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{dead}/structured/replay"),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = response.json().await.unwrap();
+    assert_eq!(error.error.code, "not_structured");
+}
+
+#[tokio::test]
+async fn structured_ws_rejects_a_dead_session() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+    h.manager.shutdown().await;
+    assert_eq!(h.manager.status(id).await, Some(SessionStatus::Dead));
+
+    let err = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
+        .await
+        .expect_err("dead structured session must not upgrade");
+    let message = err.to_string();
+    assert!(
+        message.contains("409") || message.to_lowercase().contains("conflict"),
+        "{message}"
+    );
+}
+
 #[tokio::test]
 async fn info_returns_daemon_identity() {
     let h = spawn("naps-laptop").await;
@@ -174,6 +602,7 @@ async fn list_sessions_starts_empty_and_reflects_creates() {
     assert!(empty.is_empty());
 
     let req = CreateSessionRequest {
+        view: Default::default(),
         cwd: "/tmp".into(),
         command: "sleep 60".into(),
         env: HashMap::new(),
@@ -390,6 +819,7 @@ async fn patch_session_renames_it() {
     let session = h
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: "/tmp".into(),
             command: "sleep 60".into(),
             env: HashMap::new(),
@@ -432,6 +862,7 @@ async fn post_focus_stamps_last_focused_at() {
     let session = h
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: "/tmp".into(),
             command: "sleep 60".into(),
             env: HashMap::new(),
@@ -465,6 +896,7 @@ async fn delete_owned_session_kills_it() {
     let session = h
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: "/tmp".into(),
             command: "sleep 60".into(),
             env: HashMap::new(),
@@ -486,6 +918,7 @@ async fn delete_owned_session_kills_it() {
 async fn spawn_via_manager(h: &Harness) -> Uuid {
     h.manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: "/tmp".into(),
             command: "sleep 60".into(),
             env: HashMap::new(),
@@ -595,6 +1028,7 @@ async fn delete_a_session_whose_worktree_holds_unsaved_work_returns_409_with_the
     let session = h
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: repo.path().to_path_buf(),
             command: "sleep 60".into(),
             env: HashMap::new(),
@@ -647,6 +1081,7 @@ async fn delete_with_force_ends_the_session_and_keeps_the_worktree() {
     let session = h
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: repo.path().to_path_buf(),
             command: "sleep 60".into(),
             env: HashMap::new(),
@@ -680,6 +1115,7 @@ async fn attach_ws_streams_pty_bytes() {
     let session = h
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: "/tmp".into(),
             command: "bash --noprofile --norc".into(),
             env: HashMap::new(),
@@ -748,6 +1184,7 @@ async fn events_ws_sends_a_snapshot_then_streams_deltas() {
     let created = h
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -792,6 +1229,7 @@ async fn every_events_subscriber_gets_its_own_snapshot() {
 
     h.manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),

crates/maestro-daemon/tests/rename.rs

@@ -51,6 +51,7 @@ async fn spawn_named(manager: &SessionManager, name: Option<&str>) -> Uuid {
     };
     manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: "/tmp".into(),
             command: "sleep 60".into(),
             env: Default::default(),

crates/maestro-daemon/tests/session_decisions.rs

@@ -64,6 +64,7 @@ impl Fx {
 
     fn request(&self, command: &str) -> CreateSessionRequest {
         CreateSessionRequest {
+            view: Default::default(),
             cwd: self.tmp.path().to_path_buf(),
             command: command.into(),
             env: HashMap::new(),

crates/maestro-daemon/tests/session_manager.rs

@@ -22,6 +22,7 @@ async fn create_then_kill_round_trip() {
     let session = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -63,6 +64,7 @@ async fn create_appends_to_spawn_history() {
     let session = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -92,6 +94,7 @@ async fn create_without_history_logs_nothing() {
     let cwd = TempDir::new().unwrap();
     fx.manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -114,6 +117,7 @@ async fn recover_adopts_existing_tmux_sessions() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -153,6 +157,7 @@ async fn create_captures_initial_output_in_replay_buffer() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             // The shell prints the marker before doing anything else and
             // then sleeps so the session stays alive. Before NAP-253 this
@@ -200,6 +205,7 @@ async fn attach_replay_is_the_current_screen_not_the_byte_tail() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+        view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             // Print a line, then switch to the alternate screen the way a
             // full-screen program does on startup. Only the second marker is
@@ -258,6 +264,7 @@ async fn attach_replay_carries_the_panes_mouse_modes() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+        view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             // Alternate screen, then button-event tracking (?1002) with SGR
             // coordinates (?1006) — what tmux and agent TUIs actually request.
@@ -311,6 +318,7 @@ async fn recover_re_pipes_pane_so_output_reaches_subscribers() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "bash --noprofile --norc".into(),
             env: HashMap::new(),
@@ -373,6 +381,7 @@ async fn pty_broadcast_delivers_bytes_to_subscribers() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "bash --noprofile --norc".into(),
             env: HashMap::new(),
@@ -423,6 +432,7 @@ async fn liveness_watcher_marks_dead_sessions_and_keeps_them_until_dismissed() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -488,6 +498,7 @@ async fn watcher_refreshes_the_blocked_work_indicator() {
 
     fx.manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: repo.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -580,6 +591,7 @@ async fn kill_removes_worktree_even_when_spec_says_keep() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: repo.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -617,6 +629,7 @@ async fn killing_refuses_a_worktree_with_unsaved_work_and_keeps_the_session() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: repo.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -680,6 +693,7 @@ async fn killing_refuses_until_the_branch_is_pushed() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: repo.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -729,6 +743,7 @@ async fn force_killing_a_guarded_worktree_leaves_it_on_disk() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: repo.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -774,6 +789,7 @@ async fn force_on_a_clear_worktree_removes_it_as_usual() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: repo.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -814,6 +830,7 @@ async fn killing_an_ignored_only_worktree_succeeds() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: repo.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -855,6 +872,7 @@ async fn natural_exit_of_a_guarded_worktree_leaves_it_on_disk() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: repo.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -916,6 +934,7 @@ async fn natural_exit_of_a_pristine_worktree_removes_it() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: repo.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -976,6 +995,7 @@ async fn liveness_watcher_leaves_running_sessions_alone() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -1010,6 +1030,7 @@ async fn worktree_base_repo_records_the_main_checkout_not_the_request() {
     let first = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: repo.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -1027,6 +1048,7 @@ async fn worktree_base_repo_records_the_main_checkout_not_the_request() {
     let second = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: inner_worktree.clone(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -1065,6 +1087,7 @@ async fn spawned_sessions_export_maestro_session_id_and_url() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -1103,6 +1126,7 @@ async fn a_freshly_created_session_has_never_been_focused() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -1128,6 +1152,7 @@ async fn set_focused_true_stamps_last_focused_at() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),
@@ -1176,6 +1201,7 @@ async fn attaching_stamps_last_focused_at() {
     let s = fx
         .manager
         .create(CreateSessionRequest {
+            view: Default::default(),
             cwd: cwd.path().to_path_buf(),
             command: "sleep 30".into(),
             env: HashMap::new(),

crates/maestro-daemon/tests/sidecar.rs

@@ -7,7 +7,7 @@ use std::path::PathBuf;
 
 use chrono::{DateTime, Utc};
 use maestro_daemon::sidecar::SidecarStore;
-use maestro_protocol::{Session, SessionStatus};
+use maestro_protocol::{Session, SessionStatus, SessionView};
 use serde_json::Value;
 use tempfile::TempDir;
 use uuid::Uuid;
@@ -32,6 +32,8 @@ fn sample(id: Uuid) -> Session {
         current_cwd: Some(PathBuf::from("/tmp/work/crates")),
         blocked_work: None,
         last_focused_at: None,
+        acp_session_id: None,
+        view: SessionView::Terminal,
     }
 }
 
@@ -66,15 +68,30 @@ fn delete_removes_file_and_is_idempotent() {
     let s = sample(id);
 
     store.write(&s).unwrap();
+    store.mark_for_recovery(id).unwrap();
     assert!(store.path_for(id).exists());
+    assert!(store.recovery_path_for(id).exists());
 
     store.delete(id).unwrap();
     assert!(!store.path_for(id).exists());
+    assert!(!store.recovery_path_for(id).exists());
 
     // second delete is a no-op
     store.delete(id).unwrap();
 }
 
+#[test]
+fn recovery_marker_is_consumed_once() {
+    let dir = TempDir::new().unwrap();
+    let store = SidecarStore::new(dir.path());
+    let id = Uuid::new_v4();
+
+    assert!(!store.take_recovery_marker(id).unwrap());
+    store.mark_for_recovery(id).unwrap();
+    assert!(store.take_recovery_marker(id).unwrap());
+    assert!(!store.take_recovery_marker(id).unwrap());
+}
+
 #[test]
 fn load_all_recovers_every_sidecar_and_skips_garbage() {
     let dir = TempDir::new().unwrap();
@@ -103,6 +120,37 @@ fn load_all_returns_empty_when_directory_missing() {
     assert!(store.load_all().unwrap().is_empty());
 }
 
+#[test]
+fn sidecar_missing_view_key_reads_as_terminal() {
+    // A pre-#133 sidecar on disk has no `view` field at all.
+    let dir = TempDir::new().unwrap();
+    let store = SidecarStore::new(dir.path());
+    let id = Uuid::new_v4();
+
+    let mut value = serde_json::to_value(sample(id)).unwrap();
+    value.as_object_mut().unwrap().remove("view");
+    std::fs::write(store.path_for(id), serde_json::to_vec(&value).unwrap()).unwrap();
+
+    let loaded = store.read(id).expect("read").expect("present");
+    assert_eq!(loaded.view, SessionView::Terminal);
+}
+
+#[test]
+fn sidecar_structured_view_round_trips() {
+    let dir = TempDir::new().unwrap();
+    let store = SidecarStore::new(dir.path());
+    let id = Uuid::new_v4();
+    let mut s = sample(id);
+    s.view = SessionView::Structured;
+
+    store.write(&s).expect("write");
+    let loaded = store.read(id).expect("read");
+    assert_eq!(
+        loaded.as_ref().map(|s| s.view),
+        Some(SessionView::Structured)
+    );
+}
+
 #[test]
 fn write_creates_parent_directory() {
     let dir = TempDir::new().unwrap();
@@ -113,3 +161,189 @@ fn write_creates_parent_directory() {
     assert!(nested.exists());
     assert!(store.path_for(s.id).exists());
 }
+
+// ---- per-session event log (#135) ----
+
+use maestro_acp::schema::{ContentBlock, SessionNotification, SessionUpdate, TextContent};
+use maestro_daemon::sidecar::EventCursor;
+
+fn note(text: &str) -> SessionNotification {
+    SessionNotification::new(
+        "acp-1",
+        SessionUpdate::AgentMessageChunk(maestro_acp::schema::ContentChunk::new(
+            ContentBlock::Text(TextContent::new(text)),
+        )),
+    )
+}
+
+fn text_of(n: &SessionNotification) -> &str {
+    match &n.update {
+        SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
+            ContentBlock::Text(t) => &t.text,
+            other => panic!("unexpected content {other:?}"),
+        },
+        other => panic!("unexpected update {other:?}"),
+    }
+}
+
+#[test]
+fn append_assigns_line_position_as_seq_and_reads_back() {
+    let dir = TempDir::new().unwrap();
+    let store = SidecarStore::new(dir.path());
+    let id = Uuid::new_v4();
+
+    let mut log = store.open_events(id).expect("open");
+    assert_eq!(log.last_seq(), 0);
+    assert_eq!(log.append(&note("a")).unwrap(), 1);
+    assert_eq!(log.append(&note("b")).unwrap(), 2);
+    assert_eq!(log.append(&note("c")).unwrap(), 3);
+    assert_eq!(log.last_seq(), 3);
+
+    let path = store.events_path_for(id);
+    assert!(path.ends_with(format!("{id}.events.jsonl")), "{path:?}");
+    let raw = std::fs::read_to_string(&path).unwrap();
+    assert_eq!(raw.lines().count(), 3, "one object per line:\n{raw}");
+    for line in raw.lines() {
+        let v: Value = serde_json::from_str(line).expect("each line is one JSON object");
+        assert_eq!(v["sessionId"], "acp-1");
+    }
+
+    let all = store.read_events(id, EventCursor::Since(0), 100).unwrap();
+    assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 2, 3]);
+    assert_eq!(
+        all.iter()
+            .map(|e| text_of(&e.notification))
+            .collect::<Vec<_>>(),
+        vec!["a", "b", "c"]
+    );
+}
+
+#[test]
+fn since_reads_forward_and_before_reads_backward_with_limit() {
+    let dir = TempDir::new().unwrap();
+    let store = SidecarStore::new(dir.path());
+    let id = Uuid::new_v4();
+    let mut log = store.open_events(id).unwrap();
+    for i in 1..=6 {
+        log.append(&note(&format!("n{i}"))).unwrap();
+    }
+
+    // `since` is exclusive: what a client that already holds seq 2 needs.
+    let fwd = store.read_events(id, EventCursor::Since(2), 3).unwrap();
+    assert_eq!(fwd.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![3, 4, 5]);
+    // Past the end: nothing, not an error.
+    assert!(store
+        .read_events(id, EventCursor::Since(6), 3)
+        .unwrap()
+        .is_empty());
+
+    // `before` is exclusive and pages backward: the `limit` events closest
+    // to the cursor, still in ascending seq order.
+    let back = store
+        .read_events(id, EventCursor::Before(Some(5)), 2)
+        .unwrap();
+    assert_eq!(back.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![3, 4]);
+    // No cursor: the tail of the log, for a first paint.
+    let tail = store.read_events(id, EventCursor::Before(None), 2).unwrap();
+    assert_eq!(tail.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![5, 6]);
+    // Paging past the start runs out cleanly.
+    let start = store
+        .read_events(id, EventCursor::Before(Some(2)), 10)
+        .unwrap();
+    assert_eq!(start.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1]);
+    assert!(store
+        .read_events(id, EventCursor::Before(Some(1)), 10)
+        .unwrap()
+        .is_empty());
+}
+
+#[test]
+fn missing_log_reads_as_empty() {
+    let dir = TempDir::new().unwrap();
+    let store = SidecarStore::new(dir.path());
+    let id = Uuid::new_v4();
+    assert!(store
+        .read_events(id, EventCursor::Since(0), 10)
+        .unwrap()
+        .is_empty());
+    assert!(store
+        .read_events(id, EventCursor::Before(None), 10)
+        .unwrap()
+        .is_empty());
+}
+
+#[test]
+fn malformed_lines_are_skipped_but_keep_their_seq() {
+    let dir = TempDir::new().unwrap();
+    let store = SidecarStore::new(dir.path());
+    let id = Uuid::new_v4();
+    let mut log = store.open_events(id).unwrap();
+    log.append(&note("first")).unwrap();
+    drop(log);
+
+    // Garbage in the middle and a line that is JSON but not a notification.
+    let path = store.events_path_for(id);
+    let mut raw = std::fs::read_to_string(&path).unwrap();
+    raw.push_str("{not json\n");
+    raw.push_str("{\"unrelated\": true}\n");
+    std::fs::write(&path, raw).unwrap();
+    let mut log = store.open_events(id).unwrap();
+    assert_eq!(log.last_seq(), 3);
+    assert_eq!(log.append(&note("fourth")).unwrap(), 4);
+
+    let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
+    assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 4]);
+    assert_eq!(text_of(&all[1].notification), "fourth");
+}
+
+#[test]
+fn truncated_last_line_still_opens_and_the_next_append_starts_a_fresh_line() {
+    // A hard kill mid-write leaves a partial line with no newline.
+    let dir = TempDir::new().unwrap();
+    let store = SidecarStore::new(dir.path());
+    let id = Uuid::new_v4();
+    let mut log = store.open_events(id).unwrap();
+    log.append(&note("whole")).unwrap();
+    drop(log);
+
+    let path = store.events_path_for(id);
+    let mut raw = std::fs::read(&path).unwrap();
+    raw.extend_from_slice(br#"{"sessionId":"acp-1","update":{"sessionUpd"#);
+    std::fs::write(&path, &raw).unwrap();
+
+    // Reads: the partial line is skipped, nothing else is lost.
+    let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
+    assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1]);
+
+    // Reopening terminates the partial line so the next append does not
+    // glue onto it; the partial keeps a seq of its own.
+    let mut log = store.open_events(id).unwrap();
+    assert_eq!(log.last_seq(), 2);
+    assert_eq!(log.append(&note("after")).unwrap(), 3);
+    let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
+    assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 3]);
+    assert_eq!(text_of(&all[1].notification), "after");
+    let raw = std::fs::read_to_string(&path).unwrap();
+    assert_eq!(raw.lines().count(), 3, "{raw}");
+}
+
+#[test]
+fn delete_removes_the_event_log_and_load_all_ignores_it() {
+    let dir = TempDir::new().unwrap();
+    let store = SidecarStore::new(dir.path());
+    let id = Uuid::new_v4();
+    store.write(&sample(id)).unwrap();
+    let mut log = store.open_events(id).unwrap();
+    log.append(&note("x")).unwrap();
+    drop(log);
+    assert!(store.events_path_for(id).exists());
+
+    let loaded = store.load_all().unwrap();
+    assert_eq!(loaded.len(), 1, "the .jsonl must not be read as a sidecar");
+
+    store.delete(id).unwrap();
+    assert!(!store.path_for(id).exists());
+    assert!(!store.events_path_for(id).exists());
+    // Idempotent with the log already gone.
+    store.delete(id).unwrap();
+}

crates/maestro-daemon/tests/worktree.rs

@@ -405,8 +405,10 @@ fn run_git(dir: &std::path::Path, args: &[&str]) {
     );
 }
 
+/// 49ab7e9: a new-branch add onto a name that already exists suffixes the
+/// branch (and so the directory) instead of failing, per docs/WORKTREES.md.
 #[test]
-fn add_fails_when_branch_already_exists() {
+fn add_new_suffixes_when_branch_already_exists() {
     let repo_dir = TempDir::new().unwrap();
     let repo = repo_dir.path().to_path_buf();
     init_repo(&repo);
@@ -423,9 +425,9 @@ fn add_fails_when_branch_already_exists() {
     let wt_root = TempDir::new().unwrap();
     let mgr = WorktreeManager::new(Some(wt_root.path().to_path_buf()));
 
-    let err = mgr.add_new(&repo, "feat/dup").unwrap_err();
+    let path = mgr.add_new(&repo, "feat/dup").unwrap();
-    let msg = format!("{err}");
+    assert!(path.ends_with("feat/dup-2"), "{}", path.display());
-    assert!(msg.contains("git exited"), "{msg}");
+    assert!(path.is_dir());
 }
 
 /// The project picker lists linked worktrees as pickable rows, so `base_repo`

crates/maestro-e2e/Cargo.toml

@@ -13,6 +13,7 @@ path = "src/lib.rs"
 [dependencies]
 chrono.workspace = true
 futures-util.workspace = true
+libc.workspace = true
 maestro-protocol = { workspace = true, features = ["client"] }
 maestro-testing.workspace = true
 serde.workspace = true

crates/maestro-e2e/src/daemon.rs

@@ -87,6 +87,14 @@ impl Daemon {
         }
     }
 
+    /// Send the daemon the same SIGTERM systemd uses, without waiting for its
+    /// graceful HTTP shutdown to finish.
+    pub fn send_sigterm(&self) {
+        let child = self.child.as_ref().expect("daemon is running");
+        let result = unsafe { libc::kill(child.id() as libc::pid_t, libc::SIGTERM) };
+        assert_eq!(result, 0, "send SIGTERM to daemon");
+    }
+
     /// Re-spawn the daemon on the same port + state. The previous tmux
     /// server (under `TMUX_TMPDIR`) is left running so `recover()` can adopt
     /// its sessions.

crates/maestro-e2e/tests/e2e.rs

@@ -15,7 +15,7 @@ use std::time::{Duration, Instant};
 use futures_util::{SinkExt, StreamExt};
 use maestro_e2e::util::wait_for_async;
 use maestro_e2e::{Daemon, TestClient};
-use maestro_protocol::{CreateSessionRequest, WorktreeSpec};
+use maestro_protocol::{CreateSessionRequest, SessionStatus, SessionView, WorktreeSpec};
 use tempfile::TempDir;
 use tokio_tungstenite::tungstenite::Message as WsMessage;
 
@@ -27,6 +27,7 @@ async fn bring_up() -> (Daemon, TestClient) {
 
 fn create_req(cwd: &std::path::Path, command: &str) -> CreateSessionRequest {
     CreateSessionRequest {
+        view: Default::default(),
         cwd: cwd.to_path_buf(),
         command: command.into(),
         env: HashMap::new(),
@@ -35,6 +36,61 @@ fn create_req(cwd: &std::path::Path, command: &str) -> CreateSessionRequest {
     }
 }
 
+struct FakeAcp {
+    _dir: TempDir,
+    env: HashMap<String, String>,
+}
+
+fn fake_acp(log: &std::path::Path) -> FakeAcp {
+    let dir = tempfile::tempdir().expect("fake adapter dir");
+    let adapter = dir.path().join("claude-code-acp");
+    std::fs::copy(
+        concat!(
+            env!("CARGO_MANIFEST_DIR"),
+            "/../maestro-acp/tests/fake_acp_agent.py"
+        ),
+        &adapter,
+    )
+    .expect("copy fake adapter");
+    let mut permissions = std::fs::metadata(&adapter)
+        .expect("adapter metadata")
+        .permissions();
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+        permissions.set_mode(0o755);
+    }
+    std::fs::set_permissions(&adapter, permissions).expect("make fake adapter executable");
+    let env = HashMap::from([
+        (
+            "PATH".into(),
+            format!(
+                "{}:{}",
+                dir.path().display(),
+                std::env::var("PATH").unwrap_or_default()
+            ),
+        ),
+        ("FAKE_ACP_LOG".into(), log.to_string_lossy().into_owned()),
+    ]);
+    FakeAcp { _dir: dir, env }
+}
+
+fn fake_acp_pid(log: &std::path::Path) -> Option<u32> {
+    std::fs::read_to_string(log).ok()?.lines().find_map(|line| {
+        line.strip_prefix("start pid=")?
+            .split(' ')
+            .next()?
+            .parse()
+            .ok()
+    })
+}
+
+fn pid_alive(pid: u32) -> bool {
+    std::fs::read_to_string(format!("/proc/{pid}/stat"))
+        .map(|stat| !stat.contains(") Z "))
+        .unwrap_or(false)
+}
+
 /// A worktree spec built through its wire shape. The daemon and clients ship
 /// together (#87 dropped `cleanup_on_kill` from the struct), so the JSON
 /// object carries only the fields the lifecycle actually honours.
@@ -49,6 +105,7 @@ fn worktree_spec(base_repo: &std::path::Path, branch: &str, create_branch: bool)
 /// Spawn request carrying a worktree spec against `base`.
 fn worktree_req(base: &std::path::Path, branch: &str, create_branch: bool) -> CreateSessionRequest {
     CreateSessionRequest {
+        view: Default::default(),
         cwd: base.to_path_buf(),
         command: "sleep 30".into(),
         env: HashMap::new(),
@@ -143,6 +200,7 @@ async fn kill_removes_clean_worktree() {
     init_repo(base.path());
 
     let req = CreateSessionRequest {
+        view: Default::default(),
         cwd: base.path().to_path_buf(),
         command: "sleep 30".into(),
         env: HashMap::new(),
@@ -196,6 +254,7 @@ async fn kill_refuses_worktree_with_uncommitted_changes() {
         .join("unsaved");
 
     let req = CreateSessionRequest {
+        view: Default::default(),
         cwd: base.path().to_path_buf(),
         command: "sleep 30".into(),
         env: HashMap::new(),
@@ -254,6 +313,7 @@ async fn spawn_on_existing_branch_checks_it_out() {
     run_git(&["branch", "feature/existing"], base.path());
 
     let req = CreateSessionRequest {
+        view: Default::default(),
         cwd: base.path().to_path_buf(),
         command: "sleep 30".into(),
         env: HashMap::new(),
@@ -323,8 +383,11 @@ async fn spawn_new_branch_creates_worktree_and_runs_in_it() {
     client.kill_session(s.id).await.expect("kill");
 }
 
-/// #66: a second spawn onto a branch that already has a worktree is rejected
+/// #66: a second spawn onto an *existing* branch that already has a worktree
-/// with a structured worktree error; the first worktree stays intact.
+/// is rejected with a structured worktree error; the first worktree stays
+/// intact. (A second `create_branch: true` spawn is a different case: 49ab7e9
+/// suffixes the new branch name instead, see
+/// `colliding_branch_name_gets_suffixed_too`.)
 #[tokio::test(flavor = "multi_thread")]
 async fn second_spawn_on_same_branch_is_rejected() {
     let (_daemon, client) = bring_up().await;
@@ -337,7 +400,7 @@ async fn second_spawn_on_same_branch_is_rejected() {
         .await
         .expect("first spawn");
     let dup = client
-        .spawn_session(&worktree_req(base.path(), "feature/dup", true))
+        .spawn_session(&worktree_req(base.path(), "feature/dup", false))
         .await;
     match dup {
         Err(maestro_e2e::ClientError::Api(maestro_protocol::ClientError::Api { info, .. })) => {
@@ -754,6 +817,7 @@ async fn natural_exit_settles_dirty_and_pristine_worktrees_differently() {
 
     let _dirty = client
         .spawn_session(&CreateSessionRequest {
+            view: Default::default(),
             cwd: dirty_repo.path().to_path_buf(),
             command: "sh -c 'echo unsaved > scratch.txt'".into(),
             env: HashMap::new(),
@@ -802,6 +866,7 @@ async fn natural_exit_settles_dirty_and_pristine_worktrees_differently() {
 
 fn create_req_cmd(cwd: &std::path::Path, command: &str, branch: &str) -> CreateSessionRequest {
     CreateSessionRequest {
+        view: Default::default(),
         cwd: cwd.to_path_buf(),
         command: command.into(),
         env: HashMap::new(),
@@ -985,6 +1050,66 @@ async fn daemon_restart_recovers_session() {
     client.kill_session(s.id).await.expect("kill");
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn sigterm_stops_structured_child_before_events_ws_drains() {
+    let (mut daemon, client) = bring_up().await;
+    let cwd = TempDir::new().unwrap();
+    let log = cwd.path().join("fake-acp.log");
+    let fake = fake_acp(&log);
+    let session = client
+        .spawn_session(&CreateSessionRequest {
+            view: SessionView::Structured,
+            cwd: cwd.path().to_path_buf(),
+            command: "claude".into(),
+            env: fake.env.clone(),
+            metadata: HashMap::new(),
+            worktree: None,
+        })
+        .await
+        .expect("spawn structured session");
+    let ws_url = daemon.client().events_url().expect("events URL");
+    let (_events_ws, _) = tokio_tungstenite::connect_async(ws_url)
+        .await
+        .expect("events WS handshake");
+    wait_for_async("fake ACP starts", Duration::from_secs(5), || {
+        let log = log.clone();
+        async move { fake_acp_pid(&log).is_some() }
+    })
+    .await;
+    let pid = fake_acp_pid(&log).expect("fake ACP pid");
+
+    daemon.send_sigterm();
+
+    wait_for_async(
+        "fake ACP exits after SIGTERM",
+        Duration::from_secs(5),
+        || async { !pid_alive(pid) },
+    )
+    .await;
+    let sidecar_path = daemon
+        .state_dir()
+        .join("sessions")
+        .join(format!("{}.json", session.id));
+    wait_for_async(
+        "dead structured session sidecar",
+        Duration::from_secs(5),
+        || {
+            let sidecar_path = sidecar_path.clone();
+            async move {
+                std::fs::read(&sidecar_path)
+                    .ok()
+                    .and_then(|json| {
+                        serde_json::from_slice::<maestro_protocol::Session>(&json).ok()
+                    })
+                    .is_some_and(|saved| saved.status == SessionStatus::Dead)
+            }
+        },
+    )
+    .await;
+
+    daemon.stop();
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn concurrent_spawns_each_get_unique_ids() {
     let (_daemon, client) = bring_up().await;

crates/maestro-protocol/src/display.rs

@@ -268,6 +268,7 @@ mod tests {
 
     fn session(cwd: &str) -> Session {
         Session {
+            view: Default::default(),
             id: Uuid::from_u128(1),
             cwd: PathBuf::from(cwd),
             command: "bash".into(),
@@ -284,6 +285,7 @@ mod tests {
             current_cwd: None,
             blocked_work: None,
             last_focused_at: None,
+            acp_session_id: None,
         }
     }
 

crates/maestro-protocol/src/lib.rs

@@ -116,6 +116,33 @@ pub struct Session {
     /// sorts a never-focused session after every session that has one.
     #[serde(default)]
     pub last_focused_at: Option<DateTime<Utc>>,
+    /// Whether this session's process is a plain terminal (PTY bytes through
+    /// tmux) or a structured agent driven over ACP rows (#133). Fixed at
+    /// spawn time and never changes for the life of the session. Defaults to
+    /// `Terminal` on the wire so every existing client and payload from a
+    /// daemon that predates the field keeps working untouched.
+    #[serde(default)]
+    pub view: SessionView,
+    /// The ACP session id the `claude-code-acp` adapter handed back from
+    /// `session/new` (#134). Only a [`SessionView::Structured`] session has
+    /// one, and only once its handshake succeeded; a later respawn passes it
+    /// to `session/load` so the conversation continues (#135). Distinct from
+    /// the hook-reported Claude Code conversation id (#161): that one is what
+    /// `claude --resume` takes, this one is what the ACP adapter takes, and
+    /// the two are minted by different processes with different lifetimes.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub acp_session_id: Option<String>,
+}
+
+/// A session's process shape (#133): a plain terminal, or a structured agent
+/// driven over ACP rows. Chosen at spawn and immutable for the session's
+/// life. `Terminal` is the default — every sidecar and `CreateSessionRequest`
+/// written before this field existed meant a terminal session.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
+pub enum SessionView {
+    #[default]
+    Terminal,
+    Structured,
 }
 
 /// Default for [`Session::last_activity_at`] when an older payload omits it.
@@ -352,6 +379,11 @@ pub struct CreateSessionRequest {
     pub metadata: HashMap<String, Value>,
     #[serde(default)]
     pub worktree: Option<WorktreeSpec>,
+    /// The process shape to spawn (#133). Defaulted so every existing client
+    /// and `curl` against the API keeps requesting a `Terminal` session
+    /// untouched.
+    #[serde(default)]
+    pub view: SessionView,
 }
 
 /// Result of `GET /api/dirs`: the immediate subdirectories of one directory.
@@ -723,6 +755,123 @@ pub enum CatalogEvent {
     },
 }
 
+// ---------------------------------------------------------------------------
+// Structured-session projections
+// ---------------------------------------------------------------------------
+
+/// One client-neutral item in a structured session transcript.
+///
+/// `data` preserves the ACP payload without committing clients to an ACP
+/// dependency. Rendering choices, including markdown and tool-card treatment,
+/// stay with the client.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct TranscriptRow {
+    /// Stable within one session, so later deltas can address this row.
+    pub id: String,
+    pub kind: TranscriptRowKind,
+    pub data: Value,
+}
+
+/// The semantic shape of a [`TranscriptRow`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum TranscriptRowKind {
+    User,
+    AgentText,
+    AgentThought,
+    ToolCall,
+    ToolDiff,
+    Approval,
+    Plan,
+    Notice,
+}
+
+/// A complete transcript at a durable event-log position.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct TranscriptSnapshot {
+    pub cursor: u64,
+    #[serde(default)]
+    pub rows: Vec<TranscriptRow>,
+}
+
+/// One durable, client-neutral replay step from a structured session.
+///
+/// Results are ordered by increasing `cursor`. `since` returns the first
+/// matching steps; `before` returns the final matching steps, still increasing.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct StructuredReplayEvent {
+    /// Durable ACP-log position represented by this projected step.
+    pub cursor: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub transcript: Option<TranscriptDelta>,
+    pub state: ReducedState,
+}
+
+/// A change to a transcript row.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum TranscriptDelta {
+    Append { row: TranscriptRow },
+    Patch { row: TranscriptRow },
+    Remove { id: String },
+}
+
+/// Per-turn indicators derived from the ACP stream.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct TurnState {
+    #[serde(default)]
+    pub responding: bool,
+    #[serde(default)]
+    pub thinking: bool,
+    #[serde(default)]
+    pub active_tool_calls: Vec<String>,
+}
+
+/// A whole-session view derived from ACP notifications.
+///
+/// Values whose shape ACP owns are carried as JSON so this wire protocol stays
+/// forward-compatible when ACP adds fields or variants.
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+pub struct ReducedState {
+    #[serde(default)]
+    pub turn: TurnState,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub pending_approval: Option<Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub plan: Option<Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub usage: Option<Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub context: Option<Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub model: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub mode: Option<String>,
+}
+
+/// Server messages sent over a structured-session WebSocket.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum StructuredEvent {
+    TranscriptSnapshot {
+        snapshot: TranscriptSnapshot,
+    },
+    ReducedState {
+        state: ReducedState,
+    },
+    ProjectionUpdate {
+        #[serde(skip_serializing_if = "Option::is_none")]
+        transcript: Option<TranscriptDelta>,
+        state: ReducedState,
+    },
+}
+
+/// HTTP body for `POST /api/sessions/:id/structured/prompt`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct StructuredPromptRequest {
+    pub prompt: String,
+}
+
 // ---------------------------------------------------------------------------
 // Client ↔ Daemon control messages (text frames on the attach WS)
 // ---------------------------------------------------------------------------
@@ -862,6 +1011,8 @@ mod tests {
             current_cwd: Some(PathBuf::from("/home/naps62/projects/maestro/crates")),
             blocked_work: None,
             last_focused_at: None,
+            view: SessionView::Terminal,
+            acp_session_id: None,
         }
     }
 
@@ -1073,6 +1224,7 @@ mod tests {
                 branch: "feat/foo".into(),
                 create_branch: true,
             }),
+            view: SessionView::Terminal,
         };
         assert_eq!(req, round_trip(&req));
     }
@@ -1087,6 +1239,74 @@ mod tests {
         assert!(req.worktree.is_none());
         assert!(req.env.is_empty());
         assert!(req.metadata.is_empty());
+        assert_eq!(req.view, SessionView::Terminal);
+    }
+
+    #[test]
+    fn session_view_defaults_to_terminal_on_legacy_payload() {
+        // A pre-#133 daemon or client omits `view`; it must deserialize to
+        // `Terminal` rather than failing, and `Structured` round-trips.
+        let json = serde_json::json!({
+            "id": Uuid::from_u128(1),
+            "cwd": "/tmp",
+            "command": "bash",
+            "status": "Running",
+            "created_at": "2026-05-24T12:00:00Z"
+        });
+        let s: Session = serde_json::from_value(json).unwrap();
+        assert_eq!(s.view, SessionView::Terminal);
+
+        let mut structured = sample_session();
+        structured.view = SessionView::Structured;
+        assert_eq!(structured, round_trip(&structured));
+    }
+
+    #[test]
+    fn structured_projection_types_round_trip_and_default() {
+        let row = TranscriptRow {
+            id: "tool:read-1".into(),
+            kind: TranscriptRowKind::ToolCall,
+            data: serde_json::json!({"name": "vendor.read", "title": "Read"}),
+        };
+        let snapshot = TranscriptSnapshot {
+            cursor: 7,
+            rows: vec![row.clone()],
+        };
+        assert_eq!(snapshot, round_trip(&snapshot));
+        assert_eq!(
+            TranscriptDelta::Patch { row: row.clone() },
+            round_trip(&TranscriptDelta::Patch { row })
+        );
+
+        let state: ReducedState = serde_json::from_value(serde_json::json!({})).unwrap();
+        assert_eq!(state, ReducedState::default());
+        assert_eq!(
+            serde_json::to_value(TranscriptRowKind::AgentThought).unwrap(),
+            serde_json::json!("agent_thought")
+        );
+    }
+
+    #[test]
+    fn acp_session_id_defaults_to_none_and_round_trips() {
+        // A payload from before #134 carries no `acp_session_id`; it must
+        // read as `None`, stay off the wire while unset, and round-trip
+        // once the adapter has handed one back.
+        let json = serde_json::json!({
+            "id": Uuid::from_u128(1),
+            "cwd": "/tmp",
+            "command": "bash",
+            "status": "Running",
+            "created_at": "2026-05-24T12:00:00Z"
+        });
+        let s: Session = serde_json::from_value(json).unwrap();
+        assert_eq!(s.acp_session_id, None);
+        let serialized = serde_json::to_value(&s).unwrap();
+        assert!(serialized.get("acp_session_id").is_none());
+
+        let mut structured = sample_session();
+        structured.view = SessionView::Structured;
+        structured.acp_session_id = Some("sess_01HZX".into());
+        assert_eq!(structured, round_trip(&structured));
     }
 
     /// A client that predates the new/existing discriminator sends no

crates/maestro-protocol/src/routes.rs

@@ -71,6 +71,26 @@ pub fn attach(id: Uuid) -> String {
     format!("{API_PREFIX}/attach/{id}")
 }
 
+/// `WS /api/structured/:id` — transcript/state snapshots and live deltas.
+pub fn structured(id: Uuid) -> String {
+    format!("{API_PREFIX}/structured/{id}")
+}
+
+/// `GET /api/sessions/:id/structured/replay` — cursor-bounded ACP history.
+pub fn structured_replay(id: Uuid) -> String {
+    format!("{SESSIONS}/{id}/structured/replay")
+}
+
+/// `POST /api/sessions/:id/structured/prompt` — send an ACP prompt.
+pub fn structured_prompt(id: Uuid) -> String {
+    format!("{SESSIONS}/{id}/structured/prompt")
+}
+
+/// `POST /api/sessions/:id/structured/cancel` — cancel an ACP turn.
+pub fn structured_cancel(id: Uuid) -> String {
+    format!("{SESSIONS}/{id}/structured/cancel")
+}
+
 /// Axum-style patterns for the two parameterised routes, so the server
 /// registers the same paths [`session`] and [`attach`] build.
 pub mod patterns {
@@ -80,6 +100,10 @@ pub mod patterns {
     pub const SESSION_PROMPT: &str = "/api/sessions/:id/prompt";
     pub const SESSION_FOCUS: &str = "/api/sessions/:id/focus";
     pub const ATTACH: &str = "/api/attach/:id";
+    pub const STRUCTURED: &str = "/api/structured/:id";
+    pub const STRUCTURED_REPLAY: &str = "/api/sessions/:id/structured/replay";
+    pub const STRUCTURED_PROMPT: &str = "/api/sessions/:id/structured/prompt";
+    pub const STRUCTURED_CANCEL: &str = "/api/sessions/:id/structured/cancel";
 }
 
 /// A daemon base URL that could not be turned into a WebSocket base.
@@ -206,6 +230,22 @@ mod tests {
             attach(id()),
             "/api/attach/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50"
         );
+        assert_eq!(
+            structured(id()),
+            "/api/structured/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50"
+        );
+        assert_eq!(
+            structured_replay(id()),
+            "/api/sessions/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50/structured/replay"
+        );
+        assert_eq!(
+            structured_prompt(id()),
+            "/api/sessions/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50/structured/prompt"
+        );
+        assert_eq!(
+            structured_cancel(id()),
+            "/api/sessions/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50/structured/cancel"
+        );
         assert_eq!(
             attach_url("https://laptop.lan:7070", id()).unwrap(),
             "wss://laptop.lan:7070/api/attach/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50"
@@ -240,6 +280,22 @@ mod tests {
             patterns::ATTACH.trim_end_matches(":id"),
             attach(id()).trim_end_matches(&id().to_string())
         );
+        assert_eq!(
+            patterns::STRUCTURED.trim_end_matches(":id"),
+            structured(id()).trim_end_matches(&id().to_string())
+        );
+        assert_eq!(
+            patterns::STRUCTURED_REPLAY.trim_end_matches(":id/structured/replay"),
+            session(id()).trim_end_matches(&id().to_string())
+        );
+        assert_eq!(
+            patterns::STRUCTURED_PROMPT.trim_end_matches(":id/structured/prompt"),
+            session(id()).trim_end_matches(&id().to_string())
+        );
+        assert_eq!(
+            patterns::STRUCTURED_CANCEL.trim_end_matches(":id/structured/cancel"),
+            session(id()).trim_end_matches(&id().to_string())
+        );
     }
 
     #[test]
@@ -258,5 +314,9 @@ mod tests {
             assert!(path.starts_with(API_PREFIX), "{path} escapes {API_PREFIX}");
         }
         assert!(attach(id()).starts_with(API_PREFIX));
+        assert!(structured(id()).starts_with(API_PREFIX));
+        assert!(structured_replay(id()).starts_with(API_PREFIX));
+        assert!(structured_prompt(id()).starts_with(API_PREFIX));
+        assert!(structured_cancel(id()).starts_with(API_PREFIX));
     }
 }

crates/maestro-testing/src/lib.rs

@@ -31,6 +31,7 @@ use uuid::Uuid;
 /// Build a plausible `Session` for a test to hand to [`MockDaemon::add_session`].
 pub fn session(id: Uuid, cwd: &str, command: &str) -> Session {
     Session {
+        view: Default::default(),
         id,
         cwd: cwd.into(),
         command: command.into(),
@@ -47,6 +48,7 @@ pub fn session(id: Uuid, cwd: &str, command: &str) -> Session {
         current_cwd: None,
         blocked_work: None,
         last_focused_at: None,
+        acp_session_id: None,
     }
 }
 

crates/maestro-tui/src/app.rs

@@ -1882,6 +1882,7 @@ mod tests {
             sessions: ids
                 .iter()
                 .map(|id| Session {
+                    view: Default::default(),
                     id: *id,
                     command: "bash".into(),
                     cwd: std::path::PathBuf::from("/"),
@@ -1898,6 +1899,7 @@ mod tests {
                     current_cwd: None,
                     blocked_work: None,
                     last_focused_at: None,
+                    acp_session_id: None,
                 })
                 .collect(),
         });
@@ -2063,6 +2065,7 @@ mod tests {
         let mut app = make_app();
         app.catalog.apply_event(CatalogEvent::Snapshot {
             sessions: vec![Session {
+                view: Default::default(),
                 id: sid,
                 command: "bash".into(),
                 cwd: std::path::PathBuf::from("/"),
@@ -2079,6 +2082,7 @@ mod tests {
                 current_cwd: None,
                 blocked_work: None,
                 last_focused_at: None,
+                acp_session_id: None,
             }],
         });
         app.dispatch(Action::SidebarNext);
@@ -2090,6 +2094,7 @@ mod tests {
         use chrono::Utc;
         use maestro_protocol::{Session, SessionStatus};
         Session {
+            view: Default::default(),
             id: sid,
             command: "bash".into(),
             cwd: cwd.into(),
@@ -2106,6 +2111,7 @@ mod tests {
             current_cwd: None,
             blocked_work: None,
             last_focused_at: None,
+            acp_session_id: None,
         }
     }
 
@@ -2777,6 +2783,7 @@ mod tests {
         );
         app.dispatch(Action::OpenSpawnModal);
         let req = CreateSessionRequest {
+            view: Default::default(),
             cwd: std::path::PathBuf::from("/srv/work"),
             command: "bash".into(),
             env: Default::default(),
@@ -3195,6 +3202,7 @@ mod tests {
         use chrono::Utc;
         use maestro_protocol::{CatalogEvent, Session, SessionStatus};
         let make = |n: u128, name: &str, focused: Option<chrono::DateTime<Utc>>| Session {
+            view: Default::default(),
             id: Uuid::from_u128(n),
             command: "bash".into(),
             cwd: std::path::PathBuf::from("/"),
@@ -3214,6 +3222,7 @@ mod tests {
             current_cwd: None,
             blocked_work: None,
             last_focused_at: focused,
+            acp_session_id: None,
         };
         let older = Utc::now() - chrono::Duration::seconds(60);
         app.catalog.apply_event(CatalogEvent::Snapshot {
@@ -3475,6 +3484,7 @@ mod tests {
         let sid = Uuid::from_u128(7);
         app.catalog.apply_event(CatalogEvent::Snapshot {
             sessions: vec![Session {
+                view: Default::default(),
                 id: sid,
                 command: "bash".into(),
                 cwd: std::path::PathBuf::from("/"),
@@ -3492,6 +3502,7 @@ mod tests {
                 current_cwd: None,
                 blocked_work: None,
                 last_focused_at: None,
+                acp_session_id: None,
             }],
         });
         app.dispatch(Action::KillHighlighted);
@@ -3511,6 +3522,7 @@ mod tests {
         };
         app.catalog.apply_event(CatalogEvent::Snapshot {
             sessions: vec![Session {
+                view: Default::default(),
                 id: sid,
                 command: "bash".into(),
                 cwd: std::path::PathBuf::from("/"),
@@ -3527,6 +3539,7 @@ mod tests {
                 current_cwd: None,
                 blocked_work: None,
                 last_focused_at: None,
+                acp_session_id: None,
             }],
         });
     }

crates/maestro-tui/src/catalog.rs

@@ -207,6 +207,7 @@ mod tests {
 
     fn session(id: u128, command: &str) -> Session {
         Session {
+            view: Default::default(),
             id: Uuid::from_u128(id),
             command: command.into(),
             cwd: std::path::PathBuf::from("/tmp"),
@@ -223,6 +224,7 @@ mod tests {
             current_cwd: None,
             blocked_work: None,
             last_focused_at: None,
+            acp_session_id: None,
         }
     }
 

crates/maestro-tui/src/command_bar.rs

@@ -793,6 +793,7 @@ mod tests {
 
     fn session(n: u128, command: &str) -> Session {
         Session {
+            view: Default::default(),
             id: Uuid::from_u128(n),
             cwd: PathBuf::from("/home/me/dev/api"),
             command: command.to_string(),
@@ -809,6 +810,7 @@ mod tests {
             current_cwd: None,
             blocked_work: None,
             last_focused_at: None,
+            acp_session_id: None,
         }
     }
 

crates/maestro-tui/src/draft.rs

@@ -694,6 +694,7 @@ impl DraftPane {
         }
         let worktree = self.build_worktree()?;
         Ok(CreateSessionRequest {
+            view: Default::default(),
             cwd,
             command: command.to_string(),
             env: Default::default(),

crates/maestro-tui/src/fast_forward.rs

@@ -108,6 +108,7 @@ impl FastForwardDefaults {
             return None;
         }
         Some(CreateSessionRequest {
+            view: Default::default(),
             cwd: project.path.clone(),
             command: command.to_string(),
             env: Default::default(),

crates/maestro-tui/src/sidebar.rs

@@ -484,6 +484,7 @@ mod tests {
         let mut metadata = HashMap::new();
         metadata.insert("name".into(), serde_json::Value::String(name.into()));
         Session {
+            view: Default::default(),
             id: Uuid::from_u128(id),
             command: "bash".into(),
             cwd: std::path::PathBuf::from("/tmp"),
@@ -500,6 +501,7 @@ mod tests {
             current_cwd: None,
             blocked_work: None,
             last_focused_at: None,
+            acp_session_id: None,
         }
     }
 
@@ -1250,6 +1252,7 @@ mod tests {
     #[test]
     fn falls_back_to_command_when_name_metadata_is_absent() {
         let s = Session {
+            view: Default::default(),
             id: Uuid::from_u128(1),
             command: "vim".into(),
             cwd: std::path::PathBuf::from("/tmp"),
@@ -1266,6 +1269,7 @@ mod tests {
             current_cwd: None,
             blocked_work: None,
             last_focused_at: None,
+            acp_session_id: None,
         };
         let sb = Sidebar {
             sessions: vec![s],

docs/INSTALL.md

@@ -188,6 +188,31 @@ An unknown name is rejected as an `[tui].theme` configuration error.
 The full action vocabulary is the `Action` enum in
 `crates/maestro-tui/src/action.rs`.
 
+## Structured sessions: `claude-code-acp`
+
+A session spawned with `view: Structured` (#134) runs Claude Code as a child
+of the daemon over the Agent Client Protocol instead of in a tmux pane. The
+daemon resolves the adapter binary, `claude-code-acp`, on its `PATH` at spawn
+time. Install it once per host:
+
+```sh
+npm install -g --prefix ~/.local @zed-industries/claude-code-acp
+```
+
+That puts `claude-code-acp` in `~/.local/bin`, which must be on the daemon's
+`PATH` (a systemd user unit does not read your shell profile; set
+`Environment=PATH=…` in the unit or install somewhere already on it). The
+adapter needs a logged-in Claude Code (`claude /login`).
+
+When the adapter is missing the spawn still creates the session, marked dead
+with the install command above as its activity detail, so the sidebar says
+what to run. The adapter is a child of the daemon: it ends with a `DELETE
+/api/sessions/:id` (`session/cancel`, then `SIGTERM`) and with the daemon
+itself. A daemon restart ends an in-flight structured turn, but preserves its
+worktree, session record, and event-log transcript. The replacement daemon
+starts a new adapter, resumes it with `session/load`, and replays the
+transcript. Terminal sessions keep surviving restarts through tmux.
+
 ## The browser UI
 
 The daemon serves the SPA off its own listener. Open

web/src/attach.ts

@@ -359,9 +359,6 @@ export function createAttachConnection(
   let ws: WebSocket;
   let reconnectMs = INITIAL_RECONNECT_MS;
   let reconnectTimer: number | null = null;
-  // Snapshot replay starts at row zero, which can leave Codex at its oldest line.
-  // Follow it once per socket; later scrolling remains under the operator's control.
-  let awaitingReplay = true;
 
   const sendControl = (msg: ClientControl) => {
     if (ws.readyState !== WebSocket.OPEN) return;
@@ -381,7 +378,6 @@ export function createAttachConnection(
   };
 
   function connect(): void {
-    awaitingReplay = true;
     ws = new WebSocket(attachWsUrl(sessionId));
     ws.binaryType = "arraybuffer";
 
@@ -402,20 +398,10 @@ export function createAttachConnection(
         return;
       }
       if (ev.data instanceof ArrayBuffer) {
-        const replay = awaitingReplay;
+        term.write(new Uint8Array(ev.data));
-        awaitingReplay = false;
-        term.write(new Uint8Array(ev.data), () => {
-          if (replay) term.scrollToBottom();
-        });
       } else if (ev.data instanceof Blob) {
         // Some browsers may default to Blob despite binaryType="arraybuffer".
-        void ev.data.arrayBuffer().then((b) => {
+        ev.data.arrayBuffer().then((b) => term.write(new Uint8Array(b)));
-          const replay = awaitingReplay;
-          awaitingReplay = false;
-          term.write(new Uint8Array(b), () => {
-            if (replay) term.scrollToBottom();
-          });
-        });
       }
     };
 

web/src/types.ts

@@ -56,6 +56,10 @@ export interface Session {
    * how the command bar's empty query sorts a never-focused session after every
    * session that has one (`docs/COMMAND-BAR.md`). */
   last_focused_at?: IsoTimestamp | null;
+  /** The ACP session id the `claude-code-acp` adapter returned from
+   * `session/new` (#134). Only a `Structured` session whose handshake
+   * succeeded has one; a respawn passes it to `session/load` (#135). */
+  acp_session_id?: string | null;
 }
 
 /** A session's daemon-created worktree (`SessionWorktree` on the wire). */