fix(maestro): make FaultyLogSink honor the Write contract on partial writes

write() reported Ok(partial) and Err from the same call, which Write
forbids. Defer the error to the following call so write_all observes
a genuine short write, then a failure — matching what a real fd does.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

CLAUDE.md

@@ -92,6 +92,12 @@ clients — how it opens, the sigils, the key model, the match-and-rank algorith
 the two matchers must agree on, and the fixture table they are tested against.
 Read it before changing either bar or either matcher.
 
+`docs/STRUCTURED-VIEW.md` is the contract for a structured session (an agent
+run as a daemon child over ACP instead of in a tmux pane) — the eight
+transcript row kinds, the two projections and their cursor/replay semantics,
+the handshake and declined capabilities, and the approval nonce rules. Read
+it before changing anything that renders a structured session.
+
 `nix run .#dev` (or `just dev`) brings up the whole local stack under
 process-compose: a debug daemon on 127.0.0.1:7071, then Vite on :5173 once the
 daemon answers `/api/info`. The SPA proxies `/api/*` back to the daemon, so only

Cargo.lock

@@ -1558,6 +1558,7 @@ version = "0.1.0"
 dependencies = [
  "chrono",
  "futures-util",
+ "libc",
  "maestro-protocol",
  "maestro-testing",
  "maestro-tui",

Cargo.toml

@@ -10,7 +10,7 @@ repository = "https://git.naps.pt/naps62/maestro"
 publish = false
 
 [workspace.dependencies]
-agent-client-protocol = "2.1"
+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"] }

crates/maestro-acp/src/lib.rs

@@ -19,6 +19,12 @@
 //! 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
@@ -31,13 +37,15 @@ 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, FileSystemCapabilities, Implementation,
-    InitializeRequest, NewSessionRequest, RequestPermissionOutcome, RequestPermissionRequest,
-    RequestPermissionResponse, SessionId, SessionNotification,
+    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};
@@ -82,6 +90,10 @@ pub struct SpawnConfig {
     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 {
@@ -90,8 +102,24 @@ impl SpawnConfig {
             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
+    }
+}
+
+/// A blocked ACP permission request delivered to the daemon.
+#[derive(Debug)]
+pub struct PermissionRequest {
+    /// The prompt turn that was active when ACP delivered this callback.
+    pub turn: u64,
+    pub request: RequestPermissionRequest,
+    pub response: oneshot::Sender<RequestPermissionOutcome>,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -135,9 +163,44 @@ fn is_executable(path: &Path) -> bool {
 
 enum Command {
     Cancel,
+    Prompt {
+        text: String,
+        turn: u64,
+        done: oneshot::Sender<Result<(), String>>,
+    },
     Close,
 }
 
+enum ChildCommand {
+    Signal(libc::c_int),
+    Reap(oneshot::Sender<bool>),
+}
+
+#[derive(Debug, Default)]
+struct PermissionTurnState {
+    active: Option<u64>,
+}
+
+impl PermissionTurnState {
+    fn begin(&mut self, turn: u64) {
+        self.active = Some(turn);
+    }
+
+    fn cancel_active(&mut self) {
+        self.active = None;
+    }
+
+    fn finish(&mut self, turn: u64) {
+        if self.active == Some(turn) {
+            self.active = None;
+        }
+    }
+
+    fn callback_turn(&self) -> Option<u64> {
+        self.active
+    }
+}
+
 /// Why the handshake did not produce a session id.
 enum ReadyError {
     /// The adapter answered with an error.
@@ -162,7 +225,9 @@ pub struct AcpSession {
     session_id: String,
     pid: u32,
     commands: mpsc::UnboundedSender<Command>,
+    child_commands: mpsc::UnboundedSender<ChildCommand>,
     updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
+    permission_requests: Mutex<Option<mpsc::UnboundedReceiver<PermissionRequest>>>,
     exit: watch::Receiver<Option<ExitStatus>>,
 }
 
@@ -183,6 +248,18 @@ impl AcpSession {
         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
@@ -194,18 +271,39 @@ impl AcpSession {
         self.updates.lock().expect("updates mutex").take()
     }
 
+    pub fn take_permission_requests(&self) -> Option<mpsc::UnboundedReceiver<PermissionRequest>> {
+        self.permission_requests
+            .lock()
+            .expect("permission requests")
+            .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,
+        turn: u64,
+    ) -> Result<oneshot::Receiver<Result<(), String>>, String> {
+        let (done_tx, done_rx) = oneshot::channel();
+        self.commands
+            .send(Command::Prompt {
+                text,
+                turn,
+                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) {
-        if self.has_exited() {
-            return;
-        }
         self.cancel();
         if tokio::time::timeout(cancel_grace, self.wait())
             .await
@@ -227,16 +325,14 @@ impl AcpSession {
     }
 
     fn signal(&self, sig: libc::c_int) {
-        signal_pid(self.pid, sig);
+        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);
-        if !self.has_exited() {
-            self.signal(libc::SIGTERM);
-        }
     }
 }
 
@@ -283,20 +379,20 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
     tokio::spawn(drain_stderr(stderr, pid, stderr_tail.clone()));
 
     let (exit_tx, exit_rx) = watch::channel(None);
-    tokio::spawn(async move {
-        let status = child.wait().await.ok();
-        tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
-        let _ = exit_tx.send(status);
-    });
+    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 (permission_tx, permission_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,
+        permission_tx,
         commands_rx,
         ready_tx,
     ));
@@ -308,7 +404,9 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
                 session_id,
                 pid,
                 commands: commands_tx,
+                child_commands: child_commands_tx,
                 updates: Mutex::new(Some(updates_rx)),
+                permission_requests: Mutex::new(Some(permission_rx)),
                 exit: exit_rx,
             });
         }
@@ -337,12 +435,12 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
     };
     // No half-started adapter survives a failed spawn.
     let _ = commands_tx.send(Command::Close);
-    signal_pid(pid, libc::SIGTERM);
+    let _ = child_commands_tx.send(ChildCommand::Signal(libc::SIGTERM));
     if tokio::time::timeout(TERM_GRACE, wait_exit(&exit_rx))
         .await
         .is_err()
     {
-        signal_pid(pid, libc::SIGKILL);
+        let _ = child_commands_tx.send(ChildCommand::Signal(libc::SIGKILL));
         wait_exit(&exit_rx).await;
     }
     Err(failure)
@@ -354,11 +452,54 @@ async fn wait_exit(exit: &watch::Receiver<Option<ExitStatus>>) -> Option<ExitSta
     *status
 }
 
-fn signal_pid(pid: u32, sig: libc::c_int) {
-    // ESRCH just means the child is already gone.
-    // SAFETY: plain syscall on a pid we spawned and have not reaped.
-    unsafe {
-        libc::kill(pid as libc::pid_t, sig);
+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;
+                }
+            },
+        }
     }
 }
 
@@ -375,17 +516,33 @@ fn client_capabilities() -> ClientCapabilities {
 async fn run_connection<T>(
     transport: T,
     cwd: PathBuf,
+    load_session: Option<String>,
     updates_tx: mpsc::UnboundedSender<SessionNotification>,
+    permission_requests: mpsc::UnboundedSender<PermissionRequest>,
     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();
+    // ACP callbacks have no turn id. The daemon never queues a later prompt
+    // until a cancelled prompt settles, so this active turn is unambiguous.
+    let permission_turn = Arc::new(Mutex::new(PermissionTurnState::default()));
+    let permission_turn_seen = permission_turn.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(())
@@ -394,15 +551,30 @@ async fn run_connection<T>(
         )
         .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,
-                ))
+                let turn = permission_turn_seen
+                    .lock()
+                    .expect("permission turn state")
+                    .callback_turn();
+                let Some(turn) = turn else {
+                    return responder.respond(RequestPermissionResponse::new(
+                        RequestPermissionOutcome::Cancelled,
+                    ));
+                };
+                let (response, received) = oneshot::channel();
+                if permission_requests
+                    .send(PermissionRequest {
+                        turn,
+                        request,
+                        response,
+                    })
+                    .is_err()
+                {
+                    return responder.respond(RequestPermissionResponse::new(
+                        RequestPermissionOutcome::Cancelled,
+                    ));
+                }
+                let outcome = received.await.unwrap_or(RequestPermissionOutcome::Cancelled);
+                responder.respond(RequestPermissionResponse::new(outcome))
             },
             agent_client_protocol::on_receive_request!(),
         )
@@ -410,32 +582,113 @@ async fn run_connection<T>(
             let init = InitializeRequest::new(ProtocolVersion::V1)
                 .client_capabilities(client_capabilities())
                 .client_info(Implementation::new("maestro", env!("CARGO_PKG_VERSION")));
-            if let Err(e) = cx.send_request(init).block_task().await {
-                let _ = ready_tx.send(Err(ready_error(e)));
-                return Ok(());
-            }
-            let created = match cx
-                .send_request(NewSessionRequest::new(cwd))
-                .block_task()
-                .await
-            {
+            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 = created.session_id.0.to_string();
+            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) => {
+                            permission_turn
+                                .lock()
+                                .expect("permission turn state")
+                                .cancel_active();
                             let _ = cx.send_notification(CancelNotification::new(
                                 SessionId::new(session_id.as_str()),
                             ));
                         }
+                        Some(Command::Prompt { text, turn, done }) => {
+                            permission_turn
+                                .lock()
+                                .expect("permission turn state")
+                                .begin(turn);
+                            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();
+                            let callback_turn = permission_turn.clone();
+                            if let Err(error) = cx.send_request(request).on_receiving_result(move |result| {
+                                let callback_done = callback_done.clone();
+                                let callback_turn = callback_turn.clone();
+                                async move {
+                                    callback_turn
+                                        .lock()
+                                        .expect("permission turn state")
+                                        .finish(turn);
+                                    if let Some(done) = callback_done.lock().expect("prompt sender").take() {
+                                        let _ = done.send(result.map(|_| ()).map_err(|e| e.to_string()));
+                                    }
+                                    Ok(())
+                                }
+                            }) {
+                                permission_turn
+                                    .lock()
+                                    .expect("permission turn state")
+                                    .finish(turn);
+                                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,

crates/maestro-acp/tests/fake_acp_agent.py

@@ -17,14 +17,35 @@ Environment knobs, all optional:
                             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_CANCEL_SETTLES=1  complete the outstanding prompt on 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
+  FAKE_ACP_PERMISSION_AFTER_CANCEL=1  defer the first prompt's permission
+                             callback until a later prompt
+
+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")
 
@@ -49,13 +70,36 @@ def on_sigterm(_signum, _frame):
 
 signal.signal(signal.SIGTERM, on_sigterm)
 
+session_id = f"fake-{os.getpid()}"
+ticks = 0
+prompt_failures = 0
+pending_prompt_id = None
+pending_permission_id = None
+delayed_permission = False
+
+
+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, '')}")
 
-session_id = f"fake-{os.getpid()}"
-
 for raw in sys.stdin:
     raw = raw.strip()
     if not raw:
@@ -64,6 +108,12 @@ for raw in sys.stdin:
     method = msg.get("method")
     if method is None:
         # A response to something we sent; nothing to do.
+        if msg.get("id") == pending_permission_id:
+            log("permission response " + json.dumps(msg.get("result"), sort_keys=True))
+            send({"jsonrpc": "2.0", "id": pending_prompt_id,
+                  "result": {"stopReason": "end_turn"}})
+            pending_prompt_id = None
+            pending_permission_id = None
         continue
     log(f"recv {method}")
     rid = msg.get("id")
@@ -77,22 +127,85 @@ for raw in sys.stdin:
             continue
         send({"jsonrpc": "2.0", "id": rid, "result": {
             "protocolVersion": params.get("protocolVersion", 1),
-            "agentCapabilities": {"loadSession": True},
+            "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({"jsonrpc": "2.0", "method": "session/update", "params": {
-            "sessionId": session_id,
-            "update": {"sessionUpdate": "agent_message_chunk",
-                       "content": {"type": "text", "text": "hello from fake"}},
-        }})
+        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_CANCEL_SETTLES") == "1"
+                and pending_prompt_id is not None):
+            send({"jsonrpc": "2.0", "id": pending_prompt_id,
+                  "result": {"stopReason": "cancelled"}})
+            pending_prompt_id = None
+            pending_permission_id = None
         if os.environ.get("FAKE_ACP_EXIT_ON_CANCEL") == "1":
             sys.exit(0)
+    elif method == "session/prompt":
+        if os.environ.get("FAKE_ACP_PERMISSION_AFTER_CANCEL") == "1":
+            if pending_prompt_id is None:
+                pending_prompt_id = rid
+                delayed_permission = True
+                continue
+            if delayed_permission:
+                delayed_permission = False
+                pending_permission_id = "permission-1"
+                send({"jsonrpc": "2.0", "id": pending_permission_id,
+                      "method": "session/request_permission", "params": {
+                          "sessionId": session_id,
+                          "toolCall": {"toolCallId": "tool-1", "fields": {}},
+                          "options": [
+                              {"optionId": "once", "name": "Allow once", "kind": "allow_once"},
+                              {"optionId": "deny", "name": "Deny", "kind": "reject_once"}
+                          ]
+                      }})
+                continue
+        if os.environ.get("FAKE_ACP_REQUEST_PERMISSION") == "1":
+            pending_prompt_id = rid
+            pending_permission_id = "permission-1"
+            send({"jsonrpc": "2.0", "id": pending_permission_id,
+                  "method": "session/request_permission", "params": {
+                      "sessionId": session_id,
+                      "toolCall": {"toolCallId": "tool-1", "fields": {}},
+                      "options": [
+                          {"optionId": "once", "name": "Allow once", "kind": "allow_once"},
+                          {"optionId": "session", "name": "Allow session", "kind": "allow_always"},
+                          {"optionId": "deny", "name": "Deny", "kind": "reject_once"}
+                      ]
+                  }})
+            continue
+        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

@@ -69,6 +69,24 @@ impl Fake {
         }
         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 {
@@ -274,6 +292,90 @@ async fn child_exit_before_handshake_reports_exited_early() {
     }
 }
 
+#[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]

crates/maestro-daemon/src/activity.rs

@@ -66,6 +66,23 @@ pub fn decide(idle_for: Duration, awaiting_input: bool) -> ActivityState {
     }
 }
 
+/// Derive a structured session's sidebar state from its ACP projection.
+pub fn decide_structured(
+    turn_active: bool,
+    pending_approval: bool,
+    turn_failed: bool,
+) -> ActivityState {
+    if turn_failed {
+        ActivityState::Error
+    } else if pending_approval {
+        ActivityState::AwaitingInput
+    } else if turn_active {
+        ActivityState::Active
+    } else {
+        ActivityState::Idle
+    }
+}
+
 /// How long a session's pane must be continuously focused (a client's pane,
 /// not a sidebar highlight) before an `AwaitingInput` glow clears: the
 /// operator has looked, so the rail stops asking for attention.
@@ -136,6 +153,17 @@ mod tests {
         );
     }
 
+    #[test]
+    fn structured_activity_follows_lifecycle_with_approval_precedence() {
+        assert_eq!(decide_structured(false, false, false), ActivityState::Idle);
+        assert_eq!(decide_structured(true, false, false), ActivityState::Active);
+        assert_eq!(
+            decide_structured(true, true, false),
+            ActivityState::AwaitingInput
+        );
+        assert_eq!(decide_structured(false, false, true), ActivityState::Error);
+    }
+
     #[test]
     fn report_is_in_force_until_exactly_its_expiry() {
         let ttl = Duration::from_secs(300);

crates/maestro-daemon/src/cli.rs

@@ -122,21 +122,23 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
     // shutdown, and both must reach the adapter children below (#134). The
     // unit's `KillMode=process` signals only the daemon, so nothing else
     // would end them.
-    let shutdown = async {
+    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)
         .tcp_nodelay(true)
         .with_graceful_shutdown(shutdown)
         .await?;
-    // tmux sessions outlive the daemon by design; structured sessions'
-    // adapter children do not (#134).
-    manager.shutdown().await;
     Ok(())
 }

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,13 @@ use maestro_config::SpawnPreset;
 use maestro_protocol::{
     routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonInfo, GitRoot, NameBatch,
     RenameSessionRequest, ReportActivityRequest, SendInputRequest, Session, SetFocusRequest,
-    PROTOCOL_VERSION,
+    StructuredApprovalDecision, StructuredEvent, StructuredPromptRequest, PROTOCOL_VERSION,
 };
 use serde::Deserialize;
 use uuid::Uuid;
 
-use crate::session::SessionManager;
+use crate::session::{QueueAcpPrompt, SessionManager};
+use crate::sidecar::EventCursor;
 
 #[derive(Debug, Clone)]
 pub struct LocalApiState {
@@ -64,7 +65,15 @@ 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))
+        .route(
+            routes::patterns::STRUCTURED_APPROVAL,
+            post(structured_approval),
+        );
     #[cfg(feature = "haiku-autoname")]
     {
         router = router.route(routes::patterns::SESSION_PROMPT, post(session_prompt));
@@ -170,6 +179,7 @@ fn session_error_status(code: &str) -> StatusCode {
         // The session outlived its pane: nothing to type into, and nothing
         // the daemon can do about it.
         "not_running" => StatusCode::CONFLICT,
+        "shutting_down" => StatusCode::SERVICE_UNAVAILABLE,
         _ => StatusCode::BAD_GATEWAY,
     }
 }
@@ -297,11 +307,10 @@ async fn rename_session(
     }
 }
 
-/// `POST /api/sessions/:id/activity` (#105): an agent reports its own state.
-/// The report holds above the daemon's inference for a TTL, then inference
-/// resumes — an agent that dies mid-run cannot pin its session at `Error`
-/// forever. Unknown session is 404; an unknown state value is rejected by the
-/// JSON decoder (422) rather than coerced. Answers `204 No Content`.
+/// `POST /api/sessions/:id/activity` (#105): an agent reports a terminal
+/// session's state. Structured sessions acknowledge and ignore these reports
+/// because their activity comes from ACP. Unknown session is 404; an unknown
+/// state value is rejected by the JSON decoder (422). Answers `204 No Content`.
 async fn report_activity(
     State(s): State<LocalApiState>,
     Path(id): Path<Uuid>,
@@ -368,6 +377,126 @@ 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 {
+    match s.manager.queue_acp_prompt(id, body.prompt).await {
+        Ok(QueueAcpPrompt::Queued) => StatusCode::NO_CONTENT.into_response(),
+        Ok(QueueAcpPrompt::InFlight) => error_response(
+            StatusCode::CONFLICT,
+            "turn_in_flight",
+            "the structured session already has an active turn",
+        ),
+        Ok(QueueAcpPrompt::Unavailable) => error_response(
+            StatusCode::CONFLICT,
+            "not_structured",
+            "session is not a live structured session",
+        ),
+        Ok(QueueAcpPrompt::Failed(message)) => {
+            error_response(StatusCode::BAD_GATEWAY, "acp_prompt_failed", &message)
+        }
+        Err(e) => {
+            let info = e.to_error_info();
+            error_response(session_error_status(&info.code), &info.code, &info.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)
+        }
+    }
+}
+
+async fn structured_approval(
+    State(s): State<LocalApiState>,
+    Path((id, nonce)): Path<(Uuid, String)>,
+    Json(decision): Json<StructuredApprovalDecision>,
+) -> Response {
+    let decision = match decision {
+        StructuredApprovalDecision::AllowOnce => crate::session::ApprovalDecision::AllowOnce,
+        StructuredApprovalDecision::AllowForSession => {
+            crate::session::ApprovalDecision::AllowForSession
+        }
+        StructuredApprovalDecision::Deny => crate::session::ApprovalDecision::Deny,
+    };
+    match s.manager.resolve_acp_approval(id, &nonce, decision).await {
+        Ok(Some(())) => StatusCode::NO_CONTENT.into_response(),
+        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)
+        }
+    }
+}
+
 /// 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 +547,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,850 @@
+//! 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::{
+    ActivityState, 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>,
+    turn_failed: bool,
+    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()
+    }
+
+    /// Sidebar activity derived from the live turn projection.
+    pub fn activity(&self) -> ActivityState {
+        crate::activity::decide_structured(
+            self.active_turn.is_some(),
+            self.state.pending_approval.is_some(),
+            self.turn_failed,
+        )
+    }
+
+    /// 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.turn_failed = false;
+        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.turn_failed = lifecycle == TurnLifecycle::Failure;
+        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
+    }
+
+    /// Add a daemon-originated permission row. The nonce belongs only in the
+    /// reduced state, which is the client control surface, never in ACP data.
+    pub fn request_approval(&mut self, nonce: String, request: Value) -> ProjectionUpdate {
+        let row = self.append(
+            format!("approval:{}", uuid::Uuid::new_v4()),
+            TranscriptRowKind::Approval,
+            request.clone(),
+        );
+        self.state.pending_approval = Some(serde_json::json!({
+            "nonce": nonce,
+            "request": request,
+        }));
+        ProjectionUpdate {
+            transcript: Some(row),
+            state: self.state(),
+        }
+    }
+
+    pub fn restore_approval(&mut self, row: TranscriptRow) -> ProjectionUpdate {
+        let transcript = if !self.row_index.contains_key(&row.id) {
+            self.row_index.insert(row.id.clone(), self.rows.len());
+            self.rows.push(row.clone());
+            Some(TranscriptDelta::Append { row })
+        } else {
+            None
+        };
+        ProjectionUpdate {
+            transcript,
+            state: self.state(),
+        }
+    }
+
+    /// Clear an approval only when it is still the one the client answered.
+    pub fn resolve_approval(&mut self, nonce: &str) -> Option<ProjectionUpdate> {
+        let pending = self.state.pending_approval.as_ref()?;
+        if pending.get("nonce").and_then(Value::as_str) != Some(nonce) {
+            return None;
+        }
+        self.state.pending_approval = None;
+        Some(ProjectionUpdate {
+            transcript: None,
+            state: self.state(),
+        })
+    }
+
+    /// Terminal turn paths must not leave a recovered snapshot permanently
+    /// blocked on an ACP request whose transport no longer exists.
+    pub fn clear_approval(&mut self) -> Option<ProjectionUpdate> {
+        self.state.pending_approval.take()?;
+        Some(ProjectionUpdate {
+            transcript: None,
+            state: self.state(),
+        })
+    }
+
+    /// Undo an `append`ed row that a caller decides never happened — used
+    /// when persisting the row's own event fails (#184), so a snapshot taken
+    /// after the failure cannot show a row durable replay would never
+    /// reproduce. Only ever called for a row that was just appended and
+    /// never patched, so removing it by id fully reverses `append`.
+    pub fn rollback_row(&mut self, id: &str) {
+        let Some(index) = self.row_index.remove(id) else {
+            return;
+        };
+        self.rows.remove(index);
+        for (shifted_index, row) in self.rows.iter().enumerate().skip(index) {
+            self.row_index.insert(row.id.clone(), shifted_index);
+        }
+    }
+
+    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");
+    }
+
+    #[test]
+    fn approval_history_survives_pending_cleanup() {
+        let mut projection = Projection::default();
+        let update =
+            projection.request_approval("secret".into(), serde_json::json!({"tool":"bash"}));
+        let TranscriptDelta::Append { row } = update.transcript.unwrap() else {
+            panic!("approval row")
+        };
+        assert_eq!(row.kind, TranscriptRowKind::Approval);
+        assert!(row.data.get("nonce").is_none());
+        assert!(!row.id.contains("secret"));
+        let persisted =
+            serde_json::to_string(&crate::sidecar::PersistedEvent::Approval { row: row.clone() })
+                .unwrap();
+        assert!(!persisted.contains("secret"));
+        assert_eq!(
+            projection.state().pending_approval.unwrap()["nonce"],
+            "secret"
+        );
+        projection.clear_approval();
+        assert!(projection.state().pending_approval.is_none());
+        assert_eq!(projection.snapshot(0).rows, vec![row]);
+    }
+
+    #[test]
+    fn rollback_row_removes_only_that_row_and_keeps_others_addressable() {
+        let mut projection = Projection::default();
+        projection.apply(
+            1,
+            &event(serde_json::json!({
+                "sessionUpdate": "user_message_chunk",
+                "messageId": "m1",
+                "content": {"type": "text", "text": "hi"},
+            })),
+        );
+        let TranscriptDelta::Append { row } = projection
+            .request_approval("nonce".into(), serde_json::json!({"tool": "bash"}))
+            .transcript
+            .unwrap()
+        else {
+            panic!("approval row")
+        };
+        // A row after the one being rolled back, so removing the approval
+        // (index 1) shifts this one from index 2 down to index 1 — proving
+        // `row_index` gets reindexed, not just `rows`.
+        projection.apply(
+            2,
+            &event(serde_json::json!({
+                "sessionUpdate": "user_message_chunk",
+                "messageId": "m2",
+                "content": {"type": "text", "text": "hey"},
+            })),
+        );
+        assert_eq!(projection.snapshot(0).rows.len(), 3);
+
+        projection.rollback_row(&row.id);
+
+        assert_eq!(projection.snapshot(0).rows.len(), 2);
+        assert_eq!(projection.snapshot(0).rows[0].kind, TranscriptRowKind::User);
+        assert_eq!(projection.snapshot(0).rows[1].kind, TranscriptRowKind::User);
+
+        // The shifted row must still be independently patchable at its new
+        // index.
+        let update = projection.apply(
+            3,
+            &event(serde_json::json!({
+                "sessionUpdate": "user_message_chunk",
+                "messageId": "m2",
+                "content": {"type": "text", "text": " there"},
+            })),
+        );
+        let TranscriptDelta::Patch { row } = update.transcript.unwrap() else {
+            panic!("patched row")
+        };
+        assert_eq!(row.data.pointer("/content/text").unwrap(), "hey there");
+        assert_eq!(projection.snapshot(0).rows.len(), 2);
+    }
+
+    #[test]
+    fn rollback_row_is_a_no_op_for_an_unknown_id() {
+        let mut projection = Projection::default();
+        projection.apply(
+            1,
+            &event(serde_json::json!({
+                "sessionUpdate": "user_message_chunk",
+                "content": {"type": "text", "text": "hi"},
+            })),
+        );
+        projection.rollback_row("approval:does-not-exist");
+        assert_eq!(projection.snapshot(0).rows.len(), 1);
+    }
+
+    #[test]
+    fn recovered_approval_is_historical_not_pending() {
+        let mut live = Projection::default();
+        let TranscriptDelta::Append { row } = live
+            .request_approval("daemon-nonce".into(), serde_json::json!({"tool":"bash"}))
+            .transcript
+            .unwrap()
+        else {
+            panic!("approval row")
+        };
+        let mut recovered = Projection::default();
+        let update = recovered.restore_approval(row.clone());
+        assert_eq!(
+            update.transcript,
+            Some(TranscriptDelta::Append { row: row.clone() })
+        );
+        assert!(update.state.pending_approval.is_none());
+        assert_eq!(recovered.snapshot(0).rows, vec![row]);
+        let replay = serde_json::to_string(&recovered.snapshot(1).rows).unwrap();
+        assert!(!replay.contains("daemon-nonce"));
+    }
+
+    #[test]
+    fn stale_terminal_turn_cannot_clear_newer_approval() {
+        let mut projection = Projection::default();
+        let (first, _) = projection.begin_turn();
+        projection.turn_lifecycle(TurnLifecycle::Cancel, first);
+        let (second, _) = projection.begin_turn();
+        projection.request_approval("second-nonce".into(), serde_json::json!({"tool":"bash"}));
+
+        assert!(projection
+            .turn_lifecycle(TurnLifecycle::Finish, first)
+            .is_none());
+        assert_eq!(projection.active_turn(), Some(second));
+        assert_eq!(
+            projection.state().pending_approval.unwrap()["nonce"],
+            "second-nonce"
+        );
+    }
+}

crates/maestro-daemon/src/sidecar.rs

@@ -5,10 +5,23 @@
 //! 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 and daemon approval row is one
+//! JSON object, 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_protocol::Session;
+use maestro_acp::schema::SessionNotification;
+use maestro_protocol::{Session, TranscriptRow};
 use uuid::Uuid;
 
 #[derive(Debug, thiserror::Error)]
@@ -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: Box::new(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(event) = PersistedEvent::decode(&line) else {
+                continue;
+            };
+            out.push_back(LoggedEvent { seq, event });
+            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,13 @@ 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)) {
-            Ok(()) => Ok(()),
-            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
-            Err(e) => Err(e.into()),
-        }
+        remove_if_present(&self.events_path_for(id))?;
+        remove_if_present(&self.dir.join(format!("{id}.approvals.jsonl")))?;
+        remove_if_present(&self.recovery_path_for(id))?;
+        remove_if_present(&self.path_for(id))
     }
 
     /// Read every sidecar JSON in the directory. Files that fail to parse are
@@ -81,6 +208,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 +224,419 @@ 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)]
+pub struct LoggedEvent {
+    /// 1-based line position in the log; the cursor for [`EventCursor`].
+    pub seq: u64,
+    pub event: PersistedEvent,
+}
+
+/// A single entry in the structured event history.
+///
+/// ACP notifications retain their original JSON representation, so existing
+/// event logs remain readable. Daemon-generated rows use an envelope so they
+/// can share the same cursor sequence without pretending to be ACP traffic.
+#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
+#[serde(tag = "maestroEvent", rename_all = "snake_case")]
+pub enum PersistedEvent {
+    Approval {
+        row: TranscriptRow,
+    },
+    #[serde(skip)]
+    Notification(Box<SessionNotification>),
+}
+
+impl PersistedEvent {
+    fn decode(line: &[u8]) -> Result<Self, serde_json::Error> {
+        serde_json::from_slice::<SessionNotification>(line)
+            .map(|notification| Self::Notification(Box::new(notification)))
+            .or_else(|_| serde_json::from_slice(line))
+    }
+}
+
+impl From<SessionNotification> for PersistedEvent {
+    fn from(notification: SessionNotification) -> Self {
+        Self::Notification(Box::new(notification))
+    }
+}
+
+impl From<&SessionNotification> for PersistedEvent {
+    fn from(notification: &SessionNotification) -> Self {
+        Self::Notification(Box::new(notification.clone()))
+    }
+}
+
+impl From<&PersistedEvent> for PersistedEvent {
+    fn from(event: &PersistedEvent) -> Self {
+        event.clone()
+    }
+}
+
+/// What an [`EventLog`] appends onto: something seekable by length and
+/// truncatable back to one, so a failed append can be rolled off the end
+/// again. A real [`File`] in production; an in-memory fault double in tests
+/// (#184) — the same [`EventLog::append`] rollback logic runs over both, so
+/// the behavior under test is the production behavior, not a stand-in for
+/// it.
+pub(crate) trait LogSink: Write + Send {
+    fn len(&mut self) -> std::io::Result<u64>;
+    fn truncate(&mut self, len: u64) -> std::io::Result<()>;
+}
+
+impl LogSink for File {
+    fn len(&mut self) -> std::io::Result<u64> {
+        Ok(self.metadata()?.len())
+    }
+
+    fn truncate(&mut self, len: u64) -> std::io::Result<()> {
+        self.set_len(len)
+    }
+}
+
+/// Outcome of a failed [`EventLog::append`].
+#[derive(Debug, thiserror::Error)]
+pub enum AppendError {
+    /// The write or flush failed, but the log was rolled back to exactly the
+    /// byte it was at before the call: `last_seq` and the file agree, and
+    /// this same [`EventLog`] is safe to keep appending to. The event itself
+    /// was never persisted or broadcast.
+    #[error("event append failed, rolled back cleanly: {0}")]
+    Failed(#[source] SidecarError),
+    /// The append failed *and* the rollback did not restore the log to its
+    /// prior length, so the file's line count and `last_seq` may now
+    /// disagree. The caller must not append to this [`EventLog`] again —
+    /// treat the session as dead rather than promise consistency it cannot
+    /// keep.
+    #[error("event append failed ({cause}) and rollback did not restore consistency ({rollback})")]
+    Fatal {
+        cause: SidecarError,
+        rollback: SidecarError,
+    },
+}
+
+impl From<SidecarError> for AppendError {
+    fn from(error: SidecarError) -> Self {
+        AppendError::Failed(error)
+    }
+}
+
+/// An open, append-only handle on one session's event log. Obtained from
+/// [`SidecarStore::open_events`]; held for the session's lifetime.
+pub struct EventLog {
+    file: Box<dyn LogSink>,
+    last_seq: u64,
+}
+
+impl std::fmt::Debug for EventLog {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("EventLog")
+            .field("last_seq", &self.last_seq)
+            .finish_non_exhaustive()
+    }
+}
+
+impl EventLog {
+    /// Wrap an arbitrary sink as an [`EventLog`], for tests that need to
+    /// inject a deterministic append failure (#184) — a partial write, a
+    /// flush that fails after the bytes are already down, or a rollback that
+    /// cannot itself complete.
+    #[cfg(test)]
+    pub(crate) fn for_test(file: impl LogSink + 'static, last_seq: u64) -> Self {
+        Self {
+            file: Box::new(file),
+            last_seq,
+        }
+    }
+
+    /// 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.
+    ///
+    /// A `write_all` that errors partway can leave an unterminated prefix on
+    /// disk; a `flush` that errors after a *complete* `write_all` leaves
+    /// ambiguity about whether that full, well-formed line should count —
+    /// counting it would silently shift every later seq by one, since
+    /// `last_seq` was never bumped for it. Both cases are rolled back to the
+    /// exact pre-append length before returning, so `last_seq` and the file
+    /// never disagree and the same [`EventLog`] can be appended to again
+    /// immediately. If the rollback itself fails, the log's consistency can
+    /// no longer be promised and [`AppendError::Fatal`] is returned instead.
+    pub fn append(&mut self, event: impl Into<PersistedEvent>) -> Result<u64, AppendError> {
+        let event = event.into();
+        let mut line = match &event {
+            PersistedEvent::Notification(notification) => serde_json::to_vec(notification)
+                .map_err(|e| AppendError::from(SidecarError::from(e)))?,
+            _ => {
+                serde_json::to_vec(&event).map_err(|e| AppendError::from(SidecarError::from(e)))?
+            }
+        };
+        line.push(b'\n');
+
+        let pre_len = self
+            .file
+            .len()
+            .map_err(|e| AppendError::from(SidecarError::from(e)))?;
+
+        if let Err(cause) = self.file.write_all(&line) {
+            return Err(self.rollback_after(pre_len, cause));
+        }
+        if let Err(cause) = self.file.flush() {
+            return Err(self.rollback_after(pre_len, cause));
+        }
+
+        self.last_seq += 1;
+        Ok(self.last_seq)
+    }
+
+    fn rollback_after(&mut self, pre_len: u64, cause: std::io::Error) -> AppendError {
+        match self.file.truncate(pre_len) {
+            Ok(()) => AppendError::Failed(SidecarError::Io(cause)),
+            Err(rollback) => AppendError::Fatal {
+                cause: SidecarError::Io(cause),
+                rollback: SidecarError::Io(rollback),
+            },
+        }
+    }
+}
+
+/// Bytes shared between a [`FaultyLogSink`] moved into an [`EventLog`] and
+/// the test that planted it, so the test can inspect what actually ended up
+/// on the "disk" after a fault without reaching into `EventLog`'s private
+/// `file` field.
+#[cfg(test)]
+#[derive(Clone, Default)]
+pub(crate) struct SharedBuf(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
+
+#[cfg(test)]
+impl SharedBuf {
+    pub(crate) fn snapshot(&self) -> Vec<u8> {
+        self.0.lock().unwrap().clone()
+    }
+}
+
+/// An in-memory [`LogSink`] double that injects one deterministic fault — a
+/// short write, a flush that fails after a full write, or a rollback that
+/// cannot itself complete — then behaves like a healthy sink from then on.
+/// Lets a test prove the *same* [`EventLog`] recovers, rather than one with a
+/// handle swapped in from a healthy disk, which production never does
+/// (#184).
+#[cfg(test)]
+#[derive(Default)]
+pub(crate) struct FaultyLogSink {
+    pub(crate) buf: SharedBuf,
+    /// Accept this many bytes on the next `write` call, then fail the
+    /// *following* call — a short write followed by an error, exactly what
+    /// `write_all` sees from a real fd that dies mid-line. `Write::write`
+    /// must not report an error on a call that made partial progress, so
+    /// the fault is delivered one call later, via `pending_write_error`.
+    pub(crate) fail_write_after: Option<usize>,
+    pub(crate) pending_write_error: bool,
+    pub(crate) fail_flush_once: bool,
+    pub(crate) fail_truncate: bool,
+}
+
+#[cfg(test)]
+impl Write for FaultyLogSink {
+    fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
+        if self.pending_write_error {
+            self.pending_write_error = false;
+            return Err(std::io::Error::other("fault: write failed mid-line"));
+        }
+        if let Some(allowed) = self.fail_write_after.take() {
+            let allowed = allowed.min(data.len());
+            if allowed == 0 {
+                // No progress was made, so it's fine to report the error
+                // from this same call.
+                return Err(std::io::Error::other("fault: write failed mid-line"));
+            }
+            self.buf
+                .0
+                .lock()
+                .unwrap()
+                .extend_from_slice(&data[..allowed]);
+            self.pending_write_error = true;
+            return Ok(allowed);
+        }
+        self.buf.0.lock().unwrap().extend_from_slice(data);
+        Ok(data.len())
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        if self.fail_flush_once {
+            self.fail_flush_once = false;
+            return Err(std::io::Error::other(
+                "fault: flush failed after a full write",
+            ));
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+impl LogSink for FaultyLogSink {
+    fn len(&mut self) -> std::io::Result<u64> {
+        Ok(self.buf.0.lock().unwrap().len() as u64)
+    }
+
+    fn truncate(&mut self, len: u64) -> std::io::Result<()> {
+        if self.fail_truncate {
+            return Err(std::io::Error::other("fault: rollback truncate failed"));
+        }
+        self.buf.0.lock().unwrap().truncate(len as usize);
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn sample_event(id: &str) -> PersistedEvent {
+        PersistedEvent::Approval {
+            row: TranscriptRow {
+                id: id.to_owned(),
+                kind: maestro_protocol::TranscriptRowKind::Approval,
+                data: serde_json::json!({"tool": "bash"}),
+            },
+        }
+    }
+
+    /// Replays `bytes` as a real on-disk log via the production read path,
+    /// so "live and durable agree" is checked with the same code a
+    /// reconnecting client would hit, not a hand-rolled line splitter.
+    fn replay(bytes: &[u8]) -> Vec<LoggedEvent> {
+        let dir = tempfile::tempdir().unwrap();
+        let store = SidecarStore::new(dir.path());
+        let id = Uuid::new_v4();
+        std::fs::create_dir_all(store.dir()).unwrap();
+        std::fs::write(store.events_path_for(id), bytes).unwrap();
+        store.read_events(id, EventCursor::Since(0), 100).unwrap()
+    }
+
+    #[test]
+    fn append_rolls_back_a_partial_write_and_the_same_instance_recovers() {
+        let buf = SharedBuf::default();
+        let writer = FaultyLogSink {
+            buf: buf.clone(),
+            fail_write_after: Some(5),
+            ..Default::default()
+        };
+        let mut log = EventLog::for_test(writer, 0);
+
+        let err = log.append(sample_event("r1")).unwrap_err();
+        assert!(
+            matches!(err, AppendError::Failed(_)),
+            "a repairable rollback must not be reported fatal"
+        );
+        assert_eq!(log.last_seq(), 0, "cursor must stay put on a failed append");
+        assert!(
+            buf.snapshot().is_empty(),
+            "the partial prefix a failed write_all leaves behind must be truncated away"
+        );
+
+        let seq = log
+            .append(sample_event("r1"))
+            .expect("the same instance must be able to append again after a clean rollback");
+        assert_eq!(seq, 1);
+        assert_eq!(log.last_seq(), 1);
+
+        let bytes = buf.snapshot();
+        assert_eq!(
+            bytes.iter().filter(|&&b| b == b'\n').count(),
+            1,
+            "exactly one well-formed line, not a partial prefix glued to the recovered append"
+        );
+
+        let replayed = replay(&bytes);
+        assert_eq!(replayed.len(), 1);
+        assert_eq!(replayed[0].seq, log.last_seq());
+        let PersistedEvent::Approval { row } = &replayed[0].event else {
+            panic!("approval event");
+        };
+        assert_eq!(row.id, "r1");
+    }
+
+    #[test]
+    fn append_rolls_back_when_flush_fails_after_a_full_write() {
+        let buf = SharedBuf::default();
+        let writer = FaultyLogSink {
+            buf: buf.clone(),
+            fail_flush_once: true,
+            ..Default::default()
+        };
+        let mut log = EventLog::for_test(writer, 0);
+
+        let err = log.append(sample_event("r1")).unwrap_err();
+        assert!(matches!(err, AppendError::Failed(_)));
+        assert_eq!(log.last_seq(), 0);
+        assert!(
+            buf.snapshot().is_empty(),
+            "a full-but-unflushed line must not survive uncounted by last_seq — it would \
+             silently shift every later seq by one"
+        );
+
+        let seq = log
+            .append(sample_event("r1"))
+            .expect("the same instance must recover once flush stops failing");
+        assert_eq!(seq, 1);
+
+        let bytes = buf.snapshot();
+        let replayed = replay(&bytes);
+        assert_eq!(replayed.len(), 1);
+        assert_eq!(replayed[0].seq, log.last_seq());
+    }
+
+    #[test]
+    fn append_reports_fatal_when_rollback_cannot_restore_consistency() {
+        let buf = SharedBuf::default();
+        let writer = FaultyLogSink {
+            buf,
+            fail_write_after: Some(3),
+            fail_truncate: true,
+            ..Default::default()
+        };
+        let mut log = EventLog::for_test(writer, 0);
+
+        let err = log.append(sample_event("r1")).unwrap_err();
+        assert!(
+            matches!(err, AppendError::Fatal { .. }),
+            "an unrepairable rollback must be surfaced as fatal, never silently treated as \
+             a clean recoverable failure"
+        );
+        assert_eq!(
+            log.last_seq(),
+            0,
+            "a fatal append must not advance the cursor either"
+        );
+    }
+}

crates/maestro-daemon/tests/acp.rs

@@ -6,11 +6,20 @@ mod common;
 
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use common::{wait_for_async, DaemonFixture};
+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_protocol::{ActivityState, CreateSessionRequest, SessionStatus, SessionView};
+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 {
@@ -61,13 +70,31 @@ impl Fake {
     fn pid(&self) -> u32 {
         self.log()
             .lines()
-            .find_map(|l| l.strip_prefix("start pid="))
-            .and_then(|rest| rest.split(' ').next())
-            .and_then(|p| p.parse().ok())
+            .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 {
+    let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else {
+        panic!("unexpected daemon event")
+    };
+    match &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,
@@ -109,6 +136,7 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
 
     assert_eq!(session.view, SessionView::Structured);
     assert_eq!(session.status, SessionStatus::Running);
+    assert_eq!(session.activity, ActivityState::Idle);
     let acp_id = session.acp_session_id.clone().expect("acp session id");
     assert!(acp_id.starts_with("fake-"), "{acp_id}");
 
@@ -140,20 +168,540 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
     .unwrap();
     assert_eq!(sidecar.acp_session_id.as_deref(), Some(acp_id.as_str()));
 
-    // The raw notification stream is reachable for #135, handed out once.
-    let mut updates = fx
+    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
-        .take_acp_updates(session.id)
+        .subscribe_acp_events_with_replay(session.id, EventCursor::Since(0), 10)
         .await
-        .expect("updates receiver");
-    assert!(fx.manager.take_acp_updates(session.id).await.is_none());
-    let first = tokio::time::timeout(Duration::from_secs(5), updates.recv())
+        .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("channel open");
-    assert_eq!(first.session_id.0.as_ref(), acp_id);
+        .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 structured_activity_tracks_turn_completion_and_failure() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_DELAY_PROMPT_MS".into(), "100".into());
+    env.insert("FAKE_ACP_PROMPT_FAIL_ONCE".into(), "1".into());
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .unwrap();
+
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "fail".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+    assert_eq!(
+        fx.manager
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|candidate| candidate.id == session.id)
+            .unwrap()
+            .activity,
+        ActivityState::Active
+    );
+    wait_for_async(Duration::from_secs(2), || async {
+        fx.manager
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|candidate| candidate.id == session.id)
+            .is_some_and(|candidate| candidate.activity == ActivityState::Error)
+    })
+    .await;
+
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "succeed".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+    assert_eq!(
+        fx.manager
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|candidate| candidate.id == session.id)
+            .unwrap()
+            .activity,
+        ActivityState::Active
+    );
+    wait_for_async(Duration::from_secs(2), || async {
+        fx.manager
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|candidate| candidate.id == session.id)
+            .is_some_and(|candidate| candidate.activity == ActivityState::Idle)
+    })
+    .await;
+
+    fx.manager.kill(session.id, false).await.unwrap();
+}
+
+#[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 cancelled_prompt_that_never_settles_fails_the_session() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_PERMISSION_AFTER_CANCEL".into(), "1".into());
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .unwrap();
+
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "first".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+    assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
+    let second = fx
+        .manager
+        .queue_acp_prompt(session.id, "second".into())
+        .await
+        .unwrap();
+    assert!(
+        matches!(second, maestro_daemon::session::QueueAcpPrompt::Unavailable),
+        "{second:?}"
+    );
+    assert_eq!(
+        fx.manager.status(session.id).await,
+        Some(SessionStatus::Dead)
+    );
+    assert!(!fake.log().contains("permission response"));
+}
+
+#[tokio::test]
+async fn cancellation_settlement_allows_the_next_prompt_permission() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_REQUEST_PERMISSION".into(), "1".into());
+    env.insert("FAKE_ACP_CANCEL_SETTLES".into(), "1".into());
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .unwrap();
+
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "first".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+    assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "second".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+
+    let nonce = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let (_, state, _) = fx
+                .manager
+                .subscribe_acp_projection(session.id, EventCursor::Since(0))
+                .await
+                .unwrap()
+                .unwrap();
+            if let Some(pending) = state.pending_approval {
+                break pending["nonce"].as_str().unwrap().to_owned();
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    assert!(fx
+        .manager
+        .resolve_acp_approval(
+            session.id,
+            &nonce,
+            maestro_daemon::session::ApprovalDecision::AllowOnce,
+        )
+        .await
+        .unwrap()
+        .is_some());
+    wait_for_async(Duration::from_secs(2), || {
+        let log = fake.log();
+        async move { log.contains("permission response") }
+    })
+    .await;
+    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)
+    );
+    assert_eq!(
+        revived
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|candidate| candidate.id == session.id)
+            .unwrap()
+            .activity,
+        ActivityState::Idle
+    );
+    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]
@@ -264,6 +812,176 @@ async fn shutdown_reaps_every_adapter_child() {
     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();
@@ -289,8 +1007,38 @@ async fn adapter_exit_marks_the_session_dead_and_keeps_it_listed() {
         fx.manager.status(session.id).await == Some(SessionStatus::Dead)
     })
     .await;
+    assert_eq!(
+        fx.manager
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|candidate| candidate.id == session.id)
+            .unwrap()
+            .activity,
+        ActivityState::Error
+    );
     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"
+    );
 
-    fx.manager.kill(session.id, false).await.expect("dismiss");
-    assert!(!fx.manager.has(session.id).await);
+    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/local_api.rs

@@ -4,12 +4,17 @@ 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::{
+    ActivityState, 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 +91,589 @@ 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")
+    }
+
+    fn log(&self) -> String {
+        std::fs::read_to_string(self.dir.path().join("agent.log")).unwrap_or_default()
+    }
+}
+
+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
+}
+
+async fn wait_for_session_activity(h: &Harness, id: Uuid, expected: ActivityState) {
+    tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let sessions: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
+                .await
+                .unwrap()
+                .json()
+                .await
+                .unwrap();
+            if sessions
+                .iter()
+                .find(|session| session.id == id)
+                .is_some_and(|session| session.activity == expected)
+            {
+                return;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("session {id} did not reach {expected:?}"));
+}
+
+#[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::time::timeout(Duration::from_secs(2), async {
+        loop {
+            if fake.log().matches("recv session/prompt").count() == 2 {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("both prompt commands reach ACP");
+    let log = fake.log();
+    let first_prompt = log.find("recv session/prompt").unwrap();
+    let cancel = log.find("recv session/cancel").unwrap();
+    let second_prompt = log.rfind("recv session/prompt").unwrap();
+    assert!(first_prompt < cancel && cancel < second_prompt, "{log}");
+}
+
+#[tokio::test]
+async fn structured_approval_is_secret_single_use_and_race_safe() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_REQUEST_PERMISSION".into(), "1".into());
+    let id = structured_session_with_env(&h, env).await;
+    let client = reqwest::Client::new();
+    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();
+    assert_eq!(
+        client
+            .post(url(
+                h.addr,
+                &format!("/api/sessions/{id}/structured/prompt")
+            ))
+            .json(&serde_json::json!({"prompt":"needs permission"}))
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        reqwest::StatusCode::NO_CONTENT
+    );
+
+    let nonce = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let event: serde_json::Value =
+                serde_json::from_str(&ws.next().await.unwrap().unwrap().into_text().unwrap())
+                    .unwrap();
+            if let Some(nonce) = event["state"]["pending_approval"]["nonce"].as_str() {
+                assert_eq!(event["transcript"]["row"]["kind"], "approval");
+                assert!(event["transcript"]["row"]["data"].get("nonce").is_none());
+                break nonce.to_owned();
+            }
+        }
+    })
+    .await
+    .unwrap();
+    wait_for_session_activity(&h, id, ActivityState::AwaitingInput).await;
+    assert!(
+        !fake.log().contains(&nonce),
+        "daemon nonce leaked to ACP agent"
+    );
+
+    let path = format!("/api/sessions/{id}/structured/approvals/{nonce}");
+    let (a, b) = tokio::join!(
+        client
+            .post(url(h.addr, &path))
+            .json(&serde_json::json!("allow_once"))
+            .send(),
+        client
+            .post(url(h.addr, &path))
+            .json(&serde_json::json!("deny"))
+            .send(),
+    );
+    assert_eq!(a.unwrap().status(), reqwest::StatusCode::NO_CONTENT);
+    assert_eq!(b.unwrap().status(), reqwest::StatusCode::NO_CONTENT);
+    assert_eq!(
+        client
+            .post(url(h.addr, &path))
+            .json(&serde_json::json!("deny"))
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        reqwest::StatusCode::NO_CONTENT
+    );
+    tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            if fake.log().contains("permission response") {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    let log = fake.log();
+    assert!(
+        log.contains("\"optionId\": \"once\"") || log.contains("\"optionId\": \"deny\""),
+        "{log}"
+    );
+    tokio::time::timeout(Duration::from_secs(1), async {
+        loop {
+            let event: serde_json::Value =
+                serde_json::from_str(&ws.next().await.unwrap().unwrap().into_text().unwrap())
+                    .unwrap();
+            if event["state"]["pending_approval"].is_null() {
+                break;
+            }
+        }
+    })
+    .await
+    .expect("resolved approval clears the reduced state");
+    wait_for_session_activity(&h, id, ActivityState::Idle).await;
+    assert_eq!(
+        client
+            .post(url(
+                h.addr,
+                &format!(
+                    "/api/sessions/{}/structured/approvals/{nonce}",
+                    Uuid::new_v4()
+                )
+            ))
+            .json(&serde_json::json!("deny"))
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        reqwest::StatusCode::NOT_FOUND
+    );
+}
+
+#[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");
+    wait_for_session_activity(&h, id, ActivityState::Error).await;
+    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;
@@ -789,6 +1377,73 @@ async fn events_ws_sends_a_snapshot_then_streams_deltas() {
     }
 }
 
+#[tokio::test]
+async fn events_ws_streams_structured_activity_without_hook_overrides() {
+    let h = spawn("konishi").await;
+    let fake = FakeAcp::new();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_DELAY_PROMPT_MS".into(), "500".into());
+    let (mut ws, _) = connect_async(ws_url(h.addr, "/api/events"))
+        .await
+        .expect("events ws");
+    assert!(matches!(
+        next_event(&mut ws).await,
+        maestro_protocol::CatalogEvent::Snapshot { .. }
+    ));
+
+    let id = structured_session_with_env(&h, env).await;
+    let maestro_protocol::CatalogEvent::SessionAdded { session } = next_event(&mut ws).await else {
+        panic!("expected structured SessionAdded")
+    };
+    assert_eq!(session.activity, ActivityState::Idle);
+
+    let client = reqwest::Client::new();
+    let prompt = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "hello"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(prompt.status(), reqwest::StatusCode::NO_CONTENT);
+    let report = client
+        .post(url(h.addr, &format!("/api/sessions/{id}/activity")))
+        .json(&serde_json::json!({"state": "Error"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(report.status(), reqwest::StatusCode::NO_CONTENT);
+    wait_for_session_activity(&h, id, ActivityState::Active).await;
+
+    let mut activities = Vec::new();
+    tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            if let maestro_protocol::CatalogEvent::SessionUpdated { session } =
+                next_event(&mut ws).await
+            {
+                activities.push(session.activity);
+                if session.activity == ActivityState::Idle {
+                    return;
+                }
+            }
+        }
+    })
+    .await
+    .expect("structured turn returns to idle");
+    assert!(
+        activities.contains(&ActivityState::Active),
+        "{activities:?}"
+    );
+    assert!(
+        !activities.contains(&ActivityState::Error),
+        "{activities:?}"
+    );
+
+    h.manager.kill(id, false).await.unwrap();
+}
+
 /// Each subscriber has its own lag domain now that the publisher task is gone:
 /// two clients connected at different times both get their own Snapshot.
 #[tokio::test]

crates/maestro-daemon/tests/sidecar.rs

@@ -6,8 +6,8 @@ use std::collections::HashMap;
 use std::path::PathBuf;
 
 use chrono::{DateTime, Utc};
-use maestro_daemon::sidecar::SidecarStore;
-use maestro_protocol::{Session, SessionStatus, SessionView};
+use maestro_daemon::sidecar::{PersistedEvent, SidecarStore};
+use maestro_protocol::{Session, SessionStatus, SessionView, TranscriptRow, TranscriptRowKind};
 use serde_json::Value;
 use tempfile::TempDir;
 use uuid::Uuid;
@@ -68,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();
@@ -146,3 +161,237 @@ 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:?}"),
+    }
+}
+
+fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
+    let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else {
+        panic!("expected ACP notification")
+    };
+    text_of(notification)
+}
+
+#[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(event_text).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!(event_text(&all[1]), "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!(event_text(&all[1]), "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);
+    let legacy_approvals = dir.path().join(format!("{id}.approvals.jsonl"));
+    std::fs::write(&legacy_approvals, b"obsolete").unwrap();
+    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());
+    assert!(!legacy_approvals.exists());
+    // Idempotent with the log already gone.
+    store.delete(id).unwrap();
+}
+
+#[test]
+fn approval_rows_share_event_cursors_with_legacy_acp_lines() {
+    let dir = TempDir::new().unwrap();
+    let store = SidecarStore::new(dir.path());
+    let id = Uuid::new_v4();
+    let legacy = serde_json::to_vec(&note("before")).unwrap();
+    std::fs::create_dir_all(dir.path()).unwrap();
+    std::fs::write(store.events_path_for(id), [legacy, b"\n".to_vec()].concat()).unwrap();
+
+    let mut log = store.open_events(id).unwrap();
+    assert_eq!(
+        log.append(PersistedEvent::Approval {
+            row: TranscriptRow {
+                id: "approval:secret".into(),
+                kind: TranscriptRowKind::Approval,
+                data: serde_json::json!({"tool": "bash"}),
+            },
+        })
+        .unwrap(),
+        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(|event| event.seq).collect::<Vec<_>>(),
+        [1, 2, 3]
+    );
+    assert!(matches!(all[1].event, PersistedEvent::Approval { .. }));
+    assert_eq!(event_text(&all[0]), "before");
+    assert_eq!(event_text(&all[2]), "after");
+
+    let since = store.read_events(id, EventCursor::Since(1), 1).unwrap();
+    assert_eq!(since[0].seq, 2);
+    let before = store
+        .read_events(id, EventCursor::Before(Some(3)), 1)
+        .unwrap();
+    assert_eq!(before[0].seq, 2);
+}

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;
 
@@ -36,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.
@@ -995,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/lib.rs

@@ -156,9 +156,9 @@ pub enum SessionStatus {
     Dead,
 }
 
-/// Inferred session activity, orthogonal to [`SessionStatus`] (NAP-340). The
-/// daemon derives this from tmux pane state, PTY quiescence, and `/proc`
-/// inspection — never by parsing the terminal byte stream.
+/// Session activity, orthogonal to [`SessionStatus`] (NAP-340). Terminal
+/// sessions use pane/process inference and agent reports; structured sessions
+/// use their ACP turn lifecycle (docs/SIDEBAR.md).
 ///
 /// Deserializes leniently (#110, SIDEBAR.md Precedence): an unknown variant
 /// from a newer daemon becomes [`ActivityState::Idle`] instead of failing the
@@ -166,15 +166,14 @@ pub enum SessionStatus {
 /// forgiven.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Default)]
 pub enum ActivityState {
-    /// Producing output, or the foreground process is busy.
+    /// Producing output, the foreground process is busy, or an ACP turn is active.
     Active,
     /// Foreground process alive but quiescent. The forward-compatible default.
     #[default]
     Idle,
-    /// Blocked on a `read()` from the pty — waiting on the user.
+    /// Blocked on terminal input or waiting for an ACP approval.
     AwaitingInput,
-    /// The agent running in the session reported a failure (#102). Never
-    /// inferred by the daemon; only set when an agent reports it (#105).
+    /// An agent reported a failure, or a structured session/turn failed.
     Error,
 }
 
@@ -755,6 +754,132 @@ 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,
+}
+
+/// Operator decision for a pending ACP permission request.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum StructuredApprovalDecision {
+    AllowOnce,
+    AllowForSession,
+    Deny,
+}
+
 // ---------------------------------------------------------------------------
 // Client ↔ Daemon control messages (text frames on the attach WS)
 // ---------------------------------------------------------------------------
@@ -1144,6 +1269,31 @@ mod tests {
         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

crates/maestro-protocol/src/routes.rs

@@ -71,6 +71,31 @@ 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")
+}
+
+/// `POST /api/sessions/:id/structured/approvals/:nonce` — answer ACP permission.
+pub fn structured_approval(id: Uuid, nonce: &str) -> String {
+    format!("{SESSIONS}/{id}/structured/approvals/{nonce}")
+}
+
 /// Axum-style patterns for the two parameterised routes, so the server
 /// registers the same paths [`session`] and [`attach`] build.
 pub mod patterns {
@@ -80,6 +105,11 @@ 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";
+    pub const STRUCTURED_APPROVAL: &str = "/api/sessions/:id/structured/approvals/:nonce";
 }
 
 /// A daemon base URL that could not be turned into a WebSocket base.
@@ -206,6 +236,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 +286,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 +320,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));
     }
 }

docs/INSTALL.md

@@ -18,10 +18,14 @@ wizard, and cleans up — only the unified `maestro` binary ends up persisted in
 
 **Updating is the same command.** Re-running `./install.sh` detects an existing
 systemd unit, refreshes it, and runs `systemctl --user restart maestro` instead
-of `start`. No `--update` flag needed. **The restart keeps your sessions:** the unit sets
+of `start`. No `--update` flag needed. **The restart keeps your terminal sessions:** the unit sets
 `KillMode=process`, so systemd signals only the daemon — the tmux server (and
-every session) survives, and the new daemon re-adopts them on start. A unit
-written before this default lacks the line; re-run `./install.sh` to refresh it.
+every tmux-backed session) survives, and the new daemon re-adopts them on
+start. Structured sessions (an agent run as a daemon child over ACP) are not
+tmux panes, so shutdown ends their active turn; the transcript survives and
+the next daemon attempts one resume (`docs/RESTART.md#structured-sessions`). A
+unit written before this default lacks the line; re-run `./install.sh` to
+refresh it.
 
 Inside the script, `cargo install --git https://git.naps.pt/naps62/maestro …`
 uses your already-configured git credentials for this host (HTTPS helper or
@@ -94,7 +98,10 @@ running a local daemon, for a host that only talks to another machine's.
 
 The daemon unit carries the same `KillMode=process` as the wizard-written one,
 so `systemctl --user restart maestro-daemon` leaves the tmux server and every
-session alive.
+terminal session alive. Structured sessions end their active turn on
+shutdown and resume once on the next daemon: ACP state comes back via
+`session/load`, while the event log only replays Maestro's own transcript
+projection (`docs/RESTART.md#structured-sessions`).
 
 ## Thin builds for single-role hosts
 
@@ -208,8 +215,13 @@ 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. Terminal sessions keep surviving restarts through tmux; structured
-ones do not yet.
+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. See
+`docs/STRUCTURED-VIEW.md` for the full contract — row kinds, projections, the
+handshake, and the approval and recovery rules — and `docs/RESTART.md` for
+what a client observes across the restart.
 
 ## The browser UI
 

docs/RESTART.md

@@ -2,10 +2,16 @@
 
 Maestro is upgraded in place: the operator reinstalls the binary and systemd
 restarts the daemon under every browser tab and TUI that is attached to it.
-tmux keeps the sessions alive across that (see `docs/INSTALL.md`), but clients
-lose their sockets. This document is the contract for what a client does about
-it — what the daemon publishes, how a client tells a restart from a blip, and
-what the user sees while it happens.
+tmux keeps **terminal** sessions alive across that (see `docs/INSTALL.md`),
+but clients lose their sockets either way. This document is the contract for
+what a client does about it — what the daemon publishes, how a client tells a
+restart from a blip, and what the user sees while it happens.
+
+A **structured** session (`docs/STRUCTURED-VIEW.md`) has no tmux pane behind
+it to survive on: its agent is a child of the daemon process itself, so a
+restart takes it down with the daemon. See
+[Structured sessions](#structured-sessions) below for exactly what that
+costs and what it doesn't.
 
 ## What the daemon publishes
 
@@ -68,6 +74,35 @@ resolves, so a blocked client is never a trapped one. It clears itself the
 moment a probe succeeds — there is no "retry" button, because the client is
 already retrying.
 
+## Structured sessions
+
+Everything above describes a **client's** view of a restart, and it holds for
+a structured session exactly as written — the same three verdicts, the same
+status indicator, the same overlay. What differs is what survives on the
+**daemon** side, because a structured session's agent is a daemon child, not
+a tmux pane:
+
+- A restart **ends the active turn**. There is no adapter process left to
+  finish it, and no tmux pane keeping it alive underneath.
+- The **transcript survives**, in the per-session event log
+  (`docs/STRUCTURED-VIEW.md#persistence-recovery-and-dead-sessions`) — a
+  graceful shutdown marks the session for one resume attempt. Recovery is
+  two separate operations, not one: the next daemon calls `session/load` so
+  the adapter restores its own ACP session, but whatever history the
+  adapter emits during that call is discarded, not replayed into anything —
+  the daemon already has that history on disk. Separately, the daemon
+  rebuilds its own projection by folding the persisted event log back
+  through it. An ungraceful death (a crash, `kill -9`) leaves no resume
+  marker, and the session stays dead permanently.
+- **Pending approvals and their nonces do not survive.** Both live only in
+  the old daemon's memory; any ACP permission request still mid-flight when
+  the daemon exits is answered `Cancelled` on the way down, and a resumed
+  session starts with no pending approval at all.
+
+`docs/STRUCTURED-VIEW.md` is the full contract for all of this — the event
+log's shape, the recovery marker, and exactly which kind of session death is
+resumable.
+
 ## Timing
 
 - **Grace window** — 1200 ms between the socket dropping and the overlay

docs/SIDEBAR.md

@@ -231,7 +231,24 @@ cutover points land within one tick.
 
 ## Precedence
 
-An agent-reported `ActivityState` (#105) overrides the daemon's inference for
+### Activity source by session kind
+
+The daemon supplies one `activity` field to both clients, but its source depends
+on the session kind:
+
+| Session kind | Source |
+|---|---|
+| Terminal | The existing agent-report override and five-minute TTL, falling back to PTY quiescence, foreground process and `/proc` input-block inference. |
+| Structured | ACP state directly: a pending approval is `AwaitingInput`; otherwise an in-flight turn is `Active`; no turn is `Idle`. A failed turn or dead structured session is `Error`, and the next turn clears a failed-turn error. |
+
+Structured sessions never enter the terminal inference path and ignore activity
+reports inherited through agent hooks. They have no PTY whose output or blocked
+read could describe their state. This is a second source, not a replacement for
+the terminal source.
+
+### Terminal report override
+
+For terminal sessions, an agent-reported `ActivityState` (#105) overrides the daemon's inference for
 **five minutes** (the TTL is fixed by the daemon; re-reporting extends it).
 When the TTL expires, inference resumes and the session's reported detail —
 if any — clears with the report. Clients never blend the two: whichever source
@@ -281,10 +298,10 @@ session is genuinely `Active` again — new work, not merely a refocus. Only
 after that fresh `Active` cycle does the *next* `AwaitingInput` glow again.
 Unfocusing and refocusing alone never re-arms it.
 
-This applies uniformly to `AwaitingInput` from either source — the daemon's
-own inference or an agent's report (#105) — since the rail draws them
-identically. It does not touch `Error`; that state's own TTL (below) is
-unrelated and unaffected by focus.
+This applies uniformly to `AwaitingInput` from terminal inference, an agent
+report (#105), or a structured session's pending approval, since the rail draws
+them identically. It does not touch `Error`; terminal reports keep their TTL,
+while a structured failed-turn error clears when the next turn starts.
 
 Each client reports its own pane's focus with `POST /api/sessions/:id/focus`
 (`{"focused": bool}`) whenever the pane it is showing changes — on attach, on

docs/STRUCTURED-VIEW.md

@@ -0,0 +1,372 @@
+# STRUCTURED-VIEW.md — the structured session contract
+
+The contract for a structured session (`view: Structured`, #133): an agent run
+as a daemon child over the Agent Client Protocol (ACP) instead of in a tmux
+pane. Alongside `docs/WORKTREES.md`, `docs/SIDEBAR.md` and `docs/RESTART.md`,
+read this before changing anything that renders a structured session — both
+the web and TUI renderers are written against it.
+
+It covers the eight transcript row kinds, what the daemon decides versus what
+a client decides, the two projections and their delta/cursor semantics, the
+ACP handshake and the capabilities it declines, the adapter's install
+command, and the approval nonce rules. **Non-goal: the detached runner.** It
+does not exist yet, and nothing here describes it.
+
+## What a structured session is
+
+A terminal session's agent runs inside a tmux pane; a structured session's
+agent runs as a subprocess of the daemon itself, speaking ACP over stdio. The
+daemon is the ACP **client**; the adapter process, `claude-code-acp`, is the
+ACP **agent** — it wraps Claude Code's own SDK and translates the model's
+tool calls into ACP `session/update` notifications instead of drawing them to
+a pane. `crates/maestro-acp` owns the daemon's side of that connection;
+`crates/maestro-daemon/src/projection.rs` folds what it receives into the two
+projections below.
+
+The child is a real child of the daemon process: same process group, and on
+Linux marked with `PR_SET_PDEATHSIG` so it dies with the daemon even if the
+daemon exits without cleaning up; other platforms have no such signal and
+depend on an orderly shutdown/drop to terminate it. The ordinary path on
+every platform is `AcpSession::terminate` (`session/cancel`, a short grace,
+`SIGTERM`, then `SIGKILL`), called from both `DELETE /api/sessions/:id` and
+daemon shutdown.
+`docs/WORKTREES.md`'s kill-guard rule — an unforced kill refused while the
+session's worktree holds unsaved work — applies to a structured session
+exactly as it does to a terminal one; only what gets terminated differs (an
+ACP handle instead of a tmux pane).
+
+## The adapter
+
+`ADAPTER_BIN` (`claude-code-acp`) is resolved by walking `PATH` — the
+session's own `PATH` env var if the spawn request set one, the daemon's
+otherwise — and taking the first executable file named `claude-code-acp` on
+it. Install it once per host:
+
+```sh
+npm install -g --prefix ~/.local @zed-industries/claude-code-acp
+```
+
+That is `maestro_acp::INSTALL_COMMAND`, and it is not just documentation: it
+is baked into `AcpError::AdapterMissing`'s message. When the adapter cannot
+be found, spawn still creates the session — **Dead**, with that install
+command as its `activity_detail` — so the sidebar names the fix instead of
+the session silently never appearing (see `docs/INSTALL.md` for the full
+install writeup, including the systemd `PATH` caveat). The adapter also needs
+a logged-in Claude Code (`claude /login`); a missing login surfaces the same
+way, via the adapter's own stderr tail in the failure message.
+
+The handshake is bounded by `HANDSHAKE_TIMEOUT` (60s) — long enough for a
+slow machine to boot a node process and the Claude Code SDK behind it, never
+long enough to hang a spawn indefinitely. Any handshake failure — missing
+adapter, early exit, a protocol error, or the timeout — leaves no orphaned
+process: `spawn` closes and reaps the child itself before returning the
+error.
+
+## The ACP handshake
+
+`initialize` is where the daemon states what kind of client it is. It
+declares, deliberately, almost nothing:
+
+| Capability | Declared | Why |
+| --- | --- | --- |
+| `fs.readTextFile` | `false` | The adapter uses Claude Code's own Read tool instead of asking the daemon to read a file on its behalf — the same tool a terminal-view Claude Code session uses. |
+| `fs.writeTextFile` | `false` | Same reasoning, for Write. |
+| `terminal` | `false` | The adapter runs Claude Code's own Bash tool rather than asking the daemon to open a terminal for it. |
+
+Declining all three keeps the daemon out of the file/terminal surface
+entirely — the adapter already has direct access to the session's `cwd`, and
+routing every read, write, and shell command through the daemon would only
+add a hop with no new capability. Every other capability namespace ACP
+defines (session extensions, auth, elicitation) is simply left at its
+default; the daemon does not advertise support because `claude-code-acp` has
+never asked for it, not because it was weighed and refused.
+
+The one client method that *is* implemented for real is
+`session/request_permission` (#139) — see [Approvals](#approvals) below.
+Before #139 it was a stub that declined every request; that stub is gone now
+that a live implementation exists.
+
+`session/new` opens a fresh ACP session on every ordinary spawn. A respawn
+after a daemon restart instead sets `SpawnConfig::load_session(id)` and runs
+`session/load` with the ACP session id the original handshake returned
+(`Session::acp_session_id` on the wire) — the adapter must advertise
+`agent_capabilities.load_session`, or the handshake fails outright naming
+that. Whatever history the adapter streams back while `session/load` is in
+flight is dropped on arrival, deliberately: the daemon's own event log
+already holds that history, and forwarding it too would duplicate every row
+in the projections below.
+
+## The eight transcript row kinds
+
+`TranscriptRowKind` (`maestro-protocol`) is the alphabet a structured
+transcript is written in. Every row has a stable `id` (so later deltas can
+address it), a `kind`, and a `data: Value` that preserves the underlying ACP
+payload — clients depend on ACP's *shape* without depending on ACP as a
+crate.
+
+| Kind | Carries | Row identity |
+| --- | --- | --- |
+| `User` | An echoed `user_message_chunk` — the prompt text as ACP saw it. | `user:<message_id>`, or `user:<seq>` when the chunk carries no id. |
+| `AgentText` | An `agent_message_chunk` — the agent's visible reply, streamed. | `agent:<message_id>` / `agent:<seq>`. Later chunks for the same id are appended into `data.content.text` in place, not appended as new rows. |
+| `AgentThought` | An `agent_thought_chunk` — the model's reasoning stream. | `thought:<message_id>` / `thought:<seq>`, coalesced the same way as `AgentText`. |
+| `ToolCall` | A `tool_call` / `tool_call_update` pair whose content carries no diff. | `tool:<tool_call_id>` — one row for the call's whole life; updates patch it in place, they never create a second row. |
+| `ToolDiff` | The same tool-call machinery, when the tool's content includes a `Diff` block. | Same `tool:<tool_call_id>` identity as `ToolCall`; the *kind* flips between the two as the tool's own content changes, not the row. |
+| `Approval` | A daemon-originated permission request (`RequestPermissionRequest`), serialized without its nonce. | `approval:<uuid>` — minted fresh per request, unrelated to the ACP request's own id. |
+| `Plan` | A `plan` update — the agent's current step list. | The fixed id `"plan"`. Each new plan **replaces** the row wholesale; entries are not diffed against the previous plan. |
+| `Notice` | The raw JSON of any `session/update` variant the projection does not special-case. | `notice:<seq>`. |
+
+`Notice` is the projection's forward-compatibility seam: ACP adds
+notification variants over time, and an unrecognised one becomes a generic
+row instead of being silently dropped or making the daemon reject the whole
+stream. A client that does not know what to do with a `Notice` can render it
+as an opaque, collapsed line — that is a reasonable floor, not a client bug.
+
+Rendering is entirely a client concern. The projection's job stops at
+folding ACP into these eight shapes; markdown formatting, diff rendering,
+tool-card treatment, and how a `Notice` is presented are all renderer
+choices, not daemon ones.
+
+## Daemon decisions vs. client decisions
+
+**The daemon decides:**
+
+- Which ACP `session/update` becomes which row kind, and under what id — the
+  fold in `projection.rs` is the one place this logic lives, so the web and
+  TUI renderers can never disagree about it.
+- Turn lifecycle: when a turn begins, and whether an ACP response, a cancel,
+  or a failure ends it (`TurnState.responding` / `.thinking` /
+  `.active_tool_calls` are derived, not renderer-computed).
+- Approval nonce minting, secrecy, and the single-active-approval rule (see
+  below) — a client never sees an ACP permission request before the daemon
+  has already turned it into an `Approval` row and a nonce.
+- What happens to an ACP update variant nobody has taught the daemon about
+  yet (falls back to `Notice` rather than erroring).
+- Persistence and replay: what is written to the event log, in what order,
+  and how a restart resumes it.
+
+**A client decides:**
+
+- How each row kind is drawn — markdown for `AgentText`/`AgentThought`, a
+  diff view for `ToolDiff`, a tool card for `ToolCall`, a checklist for
+  `Plan`, and a floor presentation for `Notice`.
+- How `responding` vs. `thinking` are distinguished visually, if at all.
+- How the operator is asked to resolve an `Approval` row — the daemon only
+  exposes the three-way decision (`AllowOnce` / `AllowForSession` / `Deny`),
+  never the UI that solicits it.
+- Scrolling, collapsing, and every other presentation-only behaviour.
+
+## The two projections
+
+`Projection` (daemon-only) is folded once from the durable ACP stream and
+exposes two independent views, both defined in `maestro-protocol`:
+
+- **`TranscriptSnapshot { cursor, rows }`** — the complete ordered list of
+  transcript rows built so far, plus the log position (`cursor`) it reflects.
+  It is intentionally always the *whole* transcript, not a windowed one: a
+  tool row patched much later must still find its original entry.
+- **`ReducedState`** — one whole-session snapshot folded over every event
+  seen: `turn` (the per-turn indicators above), `pending_approval`, `plan`,
+  `usage`/`context`, `model`, and `mode`. Like `TranscriptSnapshot`, it is
+  always full-history, never a delta — a client that only wants "is a turn
+  active right now" never has to replay anything to answer that.
+
+Changes to the transcript travel as `TranscriptDelta`, tagged `type`:
+`Append { row }` (a new row), `Patch { row }` (an existing row, addressed by
+`id`, replaced wholesale), and `Remove { id }`. `Remove` is defined on the
+wire for forward compatibility; nothing in the current projection emits it —
+every row, once appended, is only ever patched or left alone.
+
+### Cursor and replay semantics
+
+The event log's own line position is the cursor: 1-based, shared by ACP
+notifications and daemon-generated rows alike (`sidecar.rs`'s
+`PersistedEvent` envelope is what lets an `Approval` row share a cursor
+sequence with raw ACP JSON without pretending to be ACP traffic). Two ways to
+read it:
+
+- **`WS /api/structured/:id`** — on connect, sends a `TranscriptSnapshot` and
+  a `ReducedState` covering the whole log as it stands, then streams
+  `ProjectionUpdate { transcript, state }` for everything that happens next.
+  There is no client-supplied resume cursor on this socket: every connection
+  gets the full snapshot, not a gap-filled continuation, so a client never
+  needs the replay endpoint just to obtain the current transcript. Use
+  replay instead for cursor-bounded pagination through the durable log —
+  history from before this connection, fetched a page at a time.
+- **`GET /api/sessions/:id/structured/replay`** — cursor-bounded pagination
+  over the durable log, returning one `StructuredReplayEvent` (`cursor`,
+  `transcript`, `state`) per matching line. `?since=<n>` returns the first
+  page after `n` (`since=0` selects from the beginning, i.e. the first
+  page); a client pages forward from there using the cursor of the last
+  event returned. `?before=<n>` returns the page immediately before `n`,
+  still in ascending order — passing both is a `400`. `?limit` defaults to
+  100 and is clamped to 500. Each call re-folds
+  the entire log from scratch to compute the state at every returned cursor,
+  so a page reflects exactly the *durable* state at its cursor, not an
+  approximation of it — at the cost of redoing the whole fold per request.
+  That durable state is not always identical to what a live client saw at
+  that moment, though: turn lifecycle transitions are broadcast live without
+  a durable event backing them (see [Prompt and cancel](#prompt-and-cancel-bounded-liveness)),
+  and a restored `Approval` row carries no pending nonce, because the nonce
+  never reached disk (see [Approvals](#approvals)) — replay is exact and
+  reconstructible, not a perfect record of everything the live view showed.
+  A session that is not a live structured session (terminal, or a dead
+  adapter) answers `409 not_structured` from either route rather than a
+  coherent-looking but stale reply.
+
+## Approvals
+
+An ACP `session/request_permission` callback becomes an `Approval` row plus a
+pending decision, entirely on the daemon side:
+
+- **Generation.** The daemon mints the nonce itself — a fresh
+  `Uuid::new_v4()` string per request — rather than using anything ACP's own
+  request carries. ACP permission callbacks have no request id of their own
+  that would double as one.
+- **Live-only placement.** The nonce lives in exactly one place: the
+  in-memory `pending_approvals` map keyed by nonce, and its mirror in
+  `ReducedState.pending_approval.nonce` on the live wire. It is never part of
+  the `Approval` row's `data` — the row is `RequestPermissionRequest`
+  serialized *without* the nonce.
+- **Secrecy from the agent and from durable history.** The adapter itself
+  never learns the nonce — it is a daemon-to-client secret, never sent back
+  over the ACP connection. It is likewise never written to the event log:
+  restoring an `Approval` row from disk (`Projection::restore_approval`)
+  produces a historical row with no `pending_approval` in state, because the
+  nonce that would make it "pending" never made it to disk in the first
+  place. A restart therefore cannot resume a pending approval — see
+  [Persistence](#persistence-recovery-and-dead-sessions).
+- **One live approval at a time.** A new permission request mints its own
+  nonce, then cancels whatever was already pending (`cancel_pending_approvals`)
+  before installing itself as the new pending approval — ACP serializes
+  requests through one connection, so this only ever fires if an adapter
+  nonetheless overlaps two. Either way, the prior pending request is
+  cancelled/cleared before the new one is installed, so the newest request
+  wins.
+- **Single-use and duplicate resolution.**
+  `POST /api/sessions/:id/structured/approvals/:nonce` **always answers
+  `204`** for a known, live structured session — whether the nonce is the
+  one currently pending, one already consumed by an earlier answer, or one
+  that never existed. Only an unknown or non-structured *session* answers
+  otherwise (`404` / `409`). This is deliberate: two clients racing to answer the same
+  approval must not turn a harmless duplicate click into a client-visible
+  error. Whichever `POST` reaches the daemon first removes the nonce from
+  `pending_approvals` and is the one whose decision actually reaches the
+  adapter; every later `POST` for that nonce is a no-op `204`.
+- **Cancellation vs. denial.** A turn cancel or a session shutdown resolves
+  any pending approval with ACP's `Cancelled` outcome, not a `Deny` decision
+  — the adapter should read those as "the operation was withdrawn," not "the
+  operator refused it."
+- **Decision mapping.** `AllowOnce` and `AllowForSession` select the ACP
+  option of kind `AllowOnce` / `AllowAlways` from the request's own option
+  list. `Deny` prefers a one-shot refusal (`RejectOnce`) and falls back to a
+  permanent one (`RejectAlways`) only when that is the sole rejection option
+  offered; if neither is present, it answers `Cancelled`.
+
+## Prompt and cancel: bounded liveness
+
+`POST /api/sessions/:id/structured/prompt` and `.../cancel` are serialized
+against each other through one per-session `operation` lock
+(`AcpEventStream::operation`), because ACP permission callbacks carry no
+per-request id — the daemon's own notion of "the active turn" is the only
+thing a callback can be attributed to, so at most one turn may ever be
+in flight, and a cancelled turn must fully settle before the next prompt is
+allowed to claim the daemon's idea of "the active turn."
+
+- A prompt claims the turn and dispatches it to ACP as one operation. While a
+  turn is active, a second prompt gets `409 turn_in_flight` rather than
+  queuing — one active turn per session, never a hidden queue of prompts.
+- Cancel sends `session/cancel`, clears any pending approval as `Cancelled`,
+  and ends the turn's client-visible lifecycle immediately. If the cancelled
+  turn was still *dispatched* to ACP (waiting on its response), the daemon
+  cannot simply forget it: ACP has not confirmed the cancel yet, and a new
+  prompt claiming "the active turn" while the old one is still outstanding
+  would make the next permission callback ambiguous between the two.
+- So a cancelled-but-still-dispatched turn must **settle within
+  `CANCEL_GRACE` (500ms)** — either ACP acknowledges the prompt (however it
+  resolves) or that grace period runs out. If it settles in time, the next
+  prompt proceeds normally. If it does not, the daemon does not guess: it
+  marks the whole structured session dead (`fail_cancelled_acp_turn`) rather
+  than risk a callback resolving against the wrong turn. This is the bounded
+  half of "bounded liveness" — a structured session's prompt/cancel dance
+  either resolves promptly or the session is declared unusable, never left
+  in an indefinite ambiguous state.
+
+## Persistence, recovery, and dead sessions
+
+A structured session's durable state is three files in the sidecar
+directory: the sidecar itself (`<id>.json`), plus two next to it:
+
+- **`<id>.events.jsonl`** — the unified event log. Each line is either a raw
+  ACP `SessionNotification` (kept in its original JSON so old logs stay
+  readable) or a `{"maestroEvent": "approval", "row": …}` envelope for a
+  daemon-generated row. Both share the same 1-based line-position cursor. A
+  line a hard kill left half-written is terminated before the next append, so
+  it keeps its position and is skipped as malformed rather than corrupting
+  what follows. An append that fails *while the daemon is running* — a
+  `write_all` that errors after only a prefix reached disk, or a `flush` that
+  errors after a complete write — is rolled back to the exact byte length the
+  file was at before the call, so `last_seq` and the file's line count never
+  disagree and the same log handle can be appended to again immediately (#184).
+  If that rollback itself cannot restore the prior length, the append is
+  fatal: the log's consistency can no longer be promised, so the caller kills
+  the session instead of appending to it again.
+- **`<id>.recover`** — an empty marker file. Its only meaning is "the daemon
+  that wrote this shut down gracefully while this session was running, and
+  the next daemon may attempt to resume it." It is consumed (deleted) the
+  moment a recovery attempt reads it — one recovery attempt per graceful
+  shutdown, never a standing flag.
+
+Three ways a structured session ends up `Dead`, each with different
+persistence consequences:
+
+1. **The adapter exits, or a notification append to the event log fails,
+   while the daemon keeps running.** (An approval-row append failure is
+   narrower: as long as the log rolled back cleanly, it only cancels the
+   pending approval and rolls the unpersisted `Approval` row back out of the
+   live transcript, without marking the session dead — a snapshot taken after
+   the failure must never show a row durable replay could not reproduce
+   (#184). If the rollback itself did not restore the log's prior length,
+   this narrower path does not apply either: the session is marked dead the
+   same as any other append failure, since the log can no longer promise
+   `last_seq` matches its file.)
+   `mark_session_dead` applies the same natural-exit worktree rule
+   as a terminal session's own pane exiting (`docs/WORKTREES.md`): a pristine
+   worktree is removed, one holding unsaved work is left on disk. No recovery
+   marker is written. This death is permanent — a later daemon restart's
+   `recover()` finds the sidecar, sees no marker, and leaves the session
+   `Dead` without attempting to resume it.
+2. **A cancelled turn fails to settle within `CANCEL_GRACE`.** Same
+   consequence as (1): the session is marked dead through the same path, no
+   marker, no future resumption.
+3. **Graceful daemon shutdown**, while the session was still `Running`.
+   `mark_session_dead_after_shutdown` marks the session dead **without**
+   applying the worktree exit rule — the tree is kept exactly as it is,
+   pristine or not, because there is nothing "natural" about this exit — and
+   writes the `.recover` marker. The sidecar on disk therefore says `Dead`,
+   but is eligible for resumption.
+
+Recovery, on the next daemon's `recover()`: for every `Structured` sidecar
+not already adopted, the marker is consumed. When one was present and the
+sidecar carries a stored `acp_session_id`, the daemon spawns a fresh adapter
+with `SpawnConfig::load_session(acp_session_id)`, replays the
+entire durable event log into a new `Projection` (rebuilding the transcript
+and reduced state exactly as they were, with any restored `Approval` row
+landing as history — never as a re-armed pending approval, since its nonce
+never reached disk), and only then reopens the log for further appends. What
+the adapter streams back while its own `session/load` is in flight is
+discarded rather than appended, so the replayed history is not duplicated
+(see [The ACP handshake](#the-acp-handshake)). Success flips the session back
+to `Running`/`Idle`; failure (no stored `acp_session_id`, or the handshake
+itself failing) leaves it `Dead` with the failure reason as
+`activity_detail` — the same "tell the operator what to fix" pattern as a
+missing adapter binary.
+
+Pending approvals and their nonces never survive any of this. They live only
+in the in-memory `pending_approvals` map and the live `ReducedState`, both
+gone the instant the daemon that held them exits. Whatever ACP permission
+request was mid-flight when the old daemon died was already answered
+`Cancelled` on that daemon's own shutdown path, so a resumed session always
+starts with no pending approval.
+
+See `docs/RESTART.md` for what a *client* observes across a daemon restart,
+now that structured sessions exist alongside terminal ones.