fix(daemon): reject dead structured sessions

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

@@ -42,9 +42,10 @@ use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use agent_client_protocol::schema::v1::{
-    CancelNotification, ClientCapabilities, FileSystemCapabilities, Implementation,
-    InitializeRequest, LoadSessionRequest, 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};
@@ -153,11 +154,16 @@ fn is_executable(path: &Path) -> bool {
 
 enum Command {
     Cancel,
+    Prompt {
+        text: String,
+        done: oneshot::Sender<Result<(), String>>,
+    },
     Close,
 }
 
 enum ChildCommand {
     Signal(libc::c_int),
+    Reap(oneshot::Sender<bool>),
 }
 
 /// Why the handshake did not produce a session id.
@@ -206,6 +212,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
@@ -223,6 +241,18 @@ impl AcpSession {
         let _ = self.commands.send(Command::Cancel);
     }
 
+    /// Send a user prompt and receive completion once the agent ends the turn.
+    pub fn prompt(&self, text: String) -> Result<oneshot::Receiver<Result<(), String>>, String> {
+        let (done_tx, done_rx) = oneshot::channel();
+        self.commands
+            .send(Command::Prompt {
+                text,
+                done: done_tx,
+            })
+            .map_err(|_| "ACP connection is closed".to_string())?;
+        Ok(done_rx)
+    }
+
     /// End the child: `session/cancel`, wait `cancel_grace`, `SIGTERM`, wait
     /// `term_grace`, `SIGKILL`. Returns once the child has been reaped.
     pub async fn terminate(&self, cancel_grace: Duration, term_grace: Duration) {
@@ -392,6 +422,21 @@ async fn supervise_child(
                     // 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
@@ -534,6 +579,27 @@ async fn run_connection<T>(
                                 SessionId::new(session_id.as_str()),
                             ));
                         }
+                        Some(Command::Prompt { text, done }) => {
+                            let request = PromptRequest::new(
+                                SessionId::new(session_id.as_str()),
+                                vec![ContentBlock::from(text)],
+                            );
+                            let done = Arc::new(Mutex::new(Some(done)));
+                            let callback_done = done.clone();
+                            if let Err(error) = cx.send_request(request).on_receiving_result(move |result| {
+                                let callback_done = callback_done.clone();
+                                async move {
+                                    if let Some(done) = callback_done.lock().expect("prompt sender").take() {
+                                        let _ = done.send(result.map(|_| ()).map_err(|e| e.to_string()));
+                                    }
+                                    Ok(())
+                                }
+                            }) {
+                                if let Some(done) = done.lock().expect("prompt sender").take() {
+                                    let _ = done.send(Err(error.to_string()));
+                                }
+                            }
+                        }
                         Some(Command::Close) | None => break,
                     },
                     _ = cx.incoming_closed() => break,

crates/maestro-acp/tests/fake_acp_agent.py

@@ -17,6 +17,8 @@ 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)
@@ -27,6 +29,10 @@ Environment knobs, all optional:
                             notifications ("replayed <i>") before answering,
                             the way a real agent streams history back
   FAKE_ACP_LOAD_FAIL=1    answer `session/load` with a JSON-RPC error
+  FAKE_ACP_DELAY_PROMPT_MS=n  wait before answering `session/prompt`
+  FAKE_ACP_PROMPT_FAIL=1  answer every `session/prompt` with a JSON-RPC error
+  FAKE_ACP_PROMPT_FAIL_ONCE=1  answer only the first `session/prompt` with a
+                             JSON-RPC error
 
 Signals: SIGUSR1 emits one `agent_message_chunk` notification ("tick <n>",
 counting from 1) and logs `emit tick <n>`, so a test can make the agent
@@ -36,6 +42,7 @@ import json
 import os
 import signal
 import sys
+import time
 
 LOG = os.environ.get("FAKE_ACP_LOG")
 
@@ -62,6 +69,7 @@ signal.signal(signal.SIGTERM, on_sigterm)
 
 session_id = f"fake-{os.getpid()}"
 ticks = 0
+prompt_failures = 0
 
 
 def chunk(text):
@@ -113,6 +121,7 @@ for raw in sys.stdin:
             "authMethods": [],
         }})
     elif method == "session/new":
+        time.sleep(int(os.environ.get("FAKE_ACP_DELAY_SESSION_NEW_MS", "0")) / 1000)
         send({"jsonrpc": "2.0", "id": rid, "result": {"sessionId": session_id}})
         send(chunk("hello from fake"))
         if os.environ.get("FAKE_ACP_EXIT_AFTER_NEW") == "1":
@@ -133,6 +142,16 @@ for raw in sys.stdin:
             send(chunk("final on cancel"))
         if os.environ.get("FAKE_ACP_EXIT_ON_CANCEL") == "1":
             sys.exit(0)
+    elif method == "session/prompt":
+        time.sleep(int(os.environ.get("FAKE_ACP_DELAY_PROMPT_MS", "0")) / 1000)
+        fail_once = os.environ.get("FAKE_ACP_PROMPT_FAIL_ONCE") == "1"
+        if os.environ.get("FAKE_ACP_PROMPT_FAIL") == "1" or (fail_once and prompt_failures == 0):
+            prompt_failures += 1
+            send({"jsonrpc": "2.0", "id": rid,
+                  "error": {"code": -32000, "message": "fake prompt failure"}})
+        else:
+            send({"jsonrpc": "2.0", "id": rid,
+                  "result": {"stopReason": "end_turn"}})
     elif rid is not None:
         send({"jsonrpc": "2.0", "id": rid,
               "error": {"code": -32601, "message": f"method not found: {method}"}})

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

crates/maestro-daemon/src/projection.rs

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

crates/maestro-daemon/src/session.rs

@@ -7,7 +7,7 @@
 
 use std::collections::{HashMap, HashSet};
 use std::path::PathBuf;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -15,12 +15,13 @@ use chrono::Utc;
 use maestro_acp::{AcpSession, SpawnConfig};
 use maestro_protocol::{
     ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session,
-    SessionStatus, SessionView, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
+    SessionStatus, SessionView, StructuredReplayEvent, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
 };
 use tokio::sync::{broadcast, watch, Mutex};
 use uuid::Uuid;
 
 use crate::panes::Panes;
+use crate::projection::{Projection, ProjectionUpdate, TurnLifecycle};
 use crate::pty::{pipe_pane_command, PtyBridge};
 use crate::sidecar::{EventCursor, EventLog, LoggedEvent, SidecarError, SidecarStore};
 use crate::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord};
@@ -102,10 +103,21 @@ pub struct SessionRecord {
     acp_event_pump: Option<watch::Receiver<bool>>,
 }
 
+impl SessionRecord {
+    fn is_live_structured(&self) -> bool {
+        self.session.view == SessionView::Structured
+            && self.session.status == SessionStatus::Running
+            && self.acp.is_some()
+            && self.acp_events.is_some()
+    }
+}
+
 #[derive(Debug)]
 struct AcpEventStream {
     log: EventLog,
     live: broadcast::Sender<LoggedEvent>,
+    projection: Projection,
+    projected_live: broadcast::Sender<ProjectionUpdate>,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -127,6 +139,8 @@ pub enum SessionError {
     /// a stray keystroke on an attach socket, which is dropped.
     #[error("session is no longer running: {0}")]
     NotRunning(Uuid),
+    #[error("daemon is shutting down")]
+    ShuttingDown,
     /// The kill was refused (#79): the session's worktree holds unsaved work.
     /// Nothing happened — the session keeps running and the tree is untouched,
     /// so it stays the operator's handle on their own unfinished work.
@@ -134,6 +148,14 @@ pub enum SessionError {
     KillGuarded(maestro_protocol::InTheWay),
 }
 
+/// The result of claiming the next prompt turn for a structured ACP session.
+#[derive(Debug)]
+pub enum BeginAcpTurn {
+    Started(Arc<AcpSession>, u64),
+    InFlight,
+    Unavailable,
+}
+
 impl SessionError {
     pub fn to_error_info(&self) -> ErrorInfo {
         let code = match self {
@@ -144,6 +166,7 @@ impl SessionError {
             SessionError::NotFound(_) => "not_found",
             SessionError::InvalidEnv(_) => "invalid_env",
             SessionError::NotRunning(_) => "not_running",
+            SessionError::ShuttingDown => "shutting_down",
             SessionError::KillGuarded(_) => "kill_guarded",
         };
         let unsaved_work = match self {
@@ -195,6 +218,12 @@ struct Inner {
     /// client that reconnects never collides with the registration its
     /// previous socket has not finished releasing.
     next_viewport: AtomicU64,
+    /// Once daemon shutdown starts, it owns structured-session finalization.
+    /// Background passes must not classify those deliberate exits as natural.
+    shutting_down: AtomicBool,
+    /// Serializes structured creation with shutdown so every adapter child is
+    /// either rejected or included in shutdown's final drain.
+    structured_lifecycle: Mutex<()>,
 }
 
 impl SessionManager {
@@ -237,6 +266,8 @@ impl SessionManager {
                 events,
                 viewports: Mutex::new(Viewports::new()),
                 next_viewport: AtomicU64::new(0),
+                shutting_down: AtomicBool::new(false),
+                structured_lifecycle: Mutex::new(()),
             }),
         }
     }
@@ -319,6 +350,9 @@ impl SessionManager {
             let Some(record) = sessions.get(&id) else {
                 return Ok(None);
             };
+            if !record.is_live_structured() {
+                return Ok(None);
+            }
             let Some(stream) = &record.acp_events else {
                 return Ok(None);
             };
@@ -343,19 +377,191 @@ impl SessionManager {
             .map_err(SessionError::Sidecar)
     }
 
+    /// Replay durable structured-session data as client-neutral projections.
+    ///
+    /// The complete log is folded before selecting the requested cursor range,
+    /// so each returned state is exactly the state at its cursor.
+    pub async fn read_acp_projection_replay(
+        &self,
+        id: Uuid,
+        cursor: EventCursor,
+        limit: usize,
+    ) -> Result<Option<Vec<StructuredReplayEvent>>, SessionError> {
+        let stream = {
+            let sessions = self.inner.sessions.lock().await;
+            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
+            if !record.is_live_structured() {
+                return Ok(None);
+            }
+            record.acp_events.clone()
+        };
+        let Some(stream) = stream else {
+            return Ok(None);
+        };
+        let stream = stream.lock().await;
+        let events = self
+            .inner
+            .sidecar
+            .read_events(id, EventCursor::Since(0), usize::MAX)?;
+        let mut projection = Projection::default();
+        let mut replay = Vec::new();
+        for event in events {
+            let update = projection.apply(event.seq, &event.notification);
+            let include = match cursor {
+                EventCursor::Since(since) => event.seq > since,
+                EventCursor::Before(Some(before)) => event.seq < before,
+                EventCursor::Before(None) => true,
+            };
+            if include {
+                replay.push(StructuredReplayEvent {
+                    cursor: event.seq,
+                    transcript: update.transcript,
+                    state: update.state,
+                });
+            }
+        }
+        drop(stream);
+        match cursor {
+            EventCursor::Since(_) => replay.truncate(limit),
+            EventCursor::Before(_) if replay.len() > limit => {
+                replay.drain(..replay.len() - limit);
+            }
+            _ => {}
+        }
+        Ok(Some(replay))
+    }
+
+    /// Subscribe to the two projections of a structured session's ACP log.
+    ///
+    /// The transcript snapshot and whole reduced state are folded once by the
+    /// event pump. Holding the stream lock while subscribing makes the live
+    /// continuation start strictly after that snapshot, with no replay/live
+    /// overlap. A cursor is accepted for the future transport layer; a
+    /// snapshot intentionally remains complete so a tool patch always has its
+    /// original row available, while reduced state is always full-history.
+    pub async fn subscribe_acp_projection(
+        &self,
+        id: Uuid,
+        _cursor: EventCursor,
+    ) -> Result<
+        Option<(
+            maestro_protocol::TranscriptSnapshot,
+            maestro_protocol::ReducedState,
+            broadcast::Receiver<ProjectionUpdate>,
+        )>,
+        SessionError,
+    > {
+        let stream = {
+            let sessions = self.inner.sessions.lock().await;
+            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
+            if !record.is_live_structured() {
+                return Ok(None);
+            }
+            let Some(stream) = &record.acp_events else {
+                return Ok(None);
+            };
+            stream.clone()
+        };
+        let stream = stream.lock().await;
+        let live = stream.projected_live.subscribe();
+        Ok(Some((
+            stream.projection.snapshot(stream.log.last_seq()),
+            stream.projection.state(),
+            live,
+        )))
+    }
+
+    /// Publish a prompt lifecycle transition without adding a synthetic ACP
+    /// event. The prompt route will call this seam in #138.
+    pub async fn apply_acp_turn_lifecycle(
+        &self,
+        id: Uuid,
+        lifecycle: TurnLifecycle,
+        turn: u64,
+    ) -> Result<bool, SessionError> {
+        let stream = {
+            let sessions = self.inner.sessions.lock().await;
+            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
+            record.acp_events.clone()
+        };
+        let Some(stream) = stream else {
+            return Ok(false);
+        };
+        let mut stream = stream.lock().await;
+        let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) else {
+            return Ok(false);
+        };
+        let _ = stream.projected_live.send(projected);
+        Ok(true)
+    }
+
+    /// Begin an ACP turn only when the previous one has ended. The projection
+    /// lock makes the rejection and lifecycle update one indivisible action.
+    pub async fn begin_acp_turn(&self, id: Uuid) -> Result<BeginAcpTurn, SessionError> {
+        let (acp, stream) = {
+            let sessions = self.inner.sessions.lock().await;
+            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
+            if !record.is_live_structured() {
+                return Ok(BeginAcpTurn::Unavailable);
+            }
+            (record.acp.clone(), record.acp_events.clone())
+        };
+        let (Some(acp), Some(stream)) = (acp, stream) else {
+            return Ok(BeginAcpTurn::Unavailable);
+        };
+        let mut stream = stream.lock().await;
+        if stream.projection.turn_active() {
+            return Ok(BeginAcpTurn::InFlight);
+        }
+        let (turn, update) = stream.projection.begin_turn();
+        let _ = stream.projected_live.send(update);
+        Ok(BeginAcpTurn::Started(acp, turn))
+    }
+
+    /// Cancel an ACP turn and immediately clear its client-facing lifecycle.
+    pub async fn cancel_acp_turn(&self, id: Uuid) -> Result<bool, SessionError> {
+        let (acp, stream) = {
+            let sessions = self.inner.sessions.lock().await;
+            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
+            if !record.is_live_structured() {
+                return Ok(false);
+            }
+            (record.acp.clone(), record.acp_events.clone())
+        };
+        let (Some(acp), Some(stream)) = (acp, stream) else {
+            return Ok(false);
+        };
+        let mut stream = stream.lock().await;
+        let Some(turn) = stream.projection.active_turn() else {
+            return Ok(true);
+        };
+        acp.cancel();
+        let projected = stream
+            .projection
+            .turn_lifecycle(TurnLifecycle::Cancel, turn)
+            .expect("active turn remains current while stream is locked");
+        let _ = stream.projected_live.send(projected);
+        Ok(true)
+    }
+
     /// End every adapter child before the daemon exits (#134): each gets
     /// `session/cancel`, then SIGTERM, concurrently. The sessions stay in
     /// the map, marked Dead and written through to their sidecars, so what
     /// is on disk after the restart is truthful. tmux sessions are left
     /// alone — surviving a restart is what they are for.
     pub async fn shutdown(&self) {
+        if self.inner.shutting_down.swap(true, Ordering::AcqRel) {
+            return;
+        }
+        let _lifecycle = self.inner.structured_lifecycle.lock().await;
         let children: Vec<(Uuid, Arc<AcpSession>, Option<watch::Receiver<bool>>)> = {
             let sessions = self.inner.sessions.lock().await;
             sessions
                 .iter()
                 .filter_map(|(id, r)| {
-                    r.acp
-                        .clone()
+                    (r.session.status == SessionStatus::Running)
+                        .then_some(())
+                        .and_then(|()| r.acp.clone())
                         .map(|acp| (*id, acp, r.acp_event_pump.clone()))
                 })
                 .collect()
@@ -363,8 +569,26 @@ impl SessionManager {
         if children.is_empty() {
             return;
         }
-        tracing::info!(count = children.len(), "shutdown: ending adapter children");
-        let terminations = children
+        let reaped =
+            futures_util::future::join_all(children.iter().map(|(_, acp, _)| acp.reap_if_exited()))
+                .await;
+        let mut natural_exits = Vec::new();
+        let mut live_children = Vec::new();
+        for (child, exited) in children.iter().cloned().zip(reaped) {
+            if exited {
+                natural_exits.push(child.0);
+            } else {
+                live_children.push(child);
+            }
+        }
+        for id in natural_exits {
+            mark_session_dead_with_worktree_settlement(&self.inner, id, true).await;
+        }
+        tracing::info!(
+            count = live_children.len(),
+            "shutdown: ending adapter children"
+        );
+        let terminations = live_children
             .iter()
             .map(|(_, acp, _)| acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE));
         futures_util::future::join_all(terminations).await;
@@ -373,7 +597,7 @@ impl SessionManager {
             .filter_map(|(_, _, done)| done.clone())
             .map(wait_for_event_pump);
         futures_util::future::join_all(drains).await;
-        for (id, _, _) in children {
+        for (id, _, _) in live_children {
             mark_session_dead_after_shutdown(&self.inner, id).await;
         }
     }
@@ -457,6 +681,17 @@ impl SessionManager {
 
     /// Spawn a new session. On success returns the freshly-created `Session`.
     pub async fn create(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
+        if req.view == SessionView::Structured {
+            let _lifecycle = self.inner.structured_lifecycle.lock().await;
+            if self.inner.shutting_down.load(Ordering::Acquire) {
+                return Err(SessionError::ShuttingDown);
+            }
+            return self.create_inner(req).await;
+        }
+        self.create_inner(req).await
+    }
+
+    async fn create_inner(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
         let (effective_cwd, worktree_path, worktree_base_repo) = match &req.worktree {
             Some(spec) => {
                 let WorktreeSpec {
@@ -562,10 +797,35 @@ impl SessionManager {
         };
 
         let acp_events = if let Some(acp) = &acp {
+            let persisted =
+                match self
+                    .inner
+                    .sidecar
+                    .read_events(id, EventCursor::Since(0), usize::MAX)
+                {
+                    Ok(events) => events,
+                    Err(error) => {
+                        acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
+                            .await;
+                        let _ = std::fs::remove_file(&fifo_path);
+                        self.rollback_worktree(&worktree_base_repo, &worktree_path);
+                        return Err(SessionError::Sidecar(error));
+                    }
+                };
             match self.inner.sidecar.open_events(id) {
                 Ok(log) => {
                     let (tx, _) = broadcast::channel(EVENTS_CAPACITY);
-                    Some(Arc::new(Mutex::new(AcpEventStream { log, live: tx })))
+                    let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
+                    let mut projection = Projection::default();
+                    for event in &persisted {
+                        projection.apply(event.seq, &event.notification);
+                    }
+                    Some(Arc::new(Mutex::new(AcpEventStream {
+                        log,
+                        live: tx,
+                        projection,
+                        projected_live: projected_tx,
+                    })))
                 }
                 Err(e) => {
                     acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
@@ -903,9 +1163,9 @@ impl SessionManager {
             adopted += 1;
         }
 
-        // Structured sessions have no tmux process to discover. Their
-        // sidecars are the recovery inventory: when an ACP id exists, start a
-        // fresh adapter process and load that conversation into it.
+        // Structured sessions have no tmux process to discover. A durable
+        // marker written during graceful shutdown is the only evidence that a
+        // sidecar may be resumed; older and naturally-dead sidecars stay dead.
         for mut session in self
             .inner
             .sidecar
@@ -918,6 +1178,7 @@ impl SessionManager {
             }
 
             let id = session.id;
+            let recoverable = self.inner.sidecar.take_recovery_marker(id)?;
             let mut session_env = sanitize_env(&session.env)?;
             let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo"));
             let bridge =
@@ -933,46 +1194,80 @@ impl SessionManager {
                 session_env.insert("MAESTRO_URL".into(), url);
             }
 
-            let (acp, failure) = match session.acp_session_id.clone() {
-                Some(acp_session_id) => {
-                    let config = SpawnConfig::new(
-                        &session.cwd,
-                        session_env
-                            .iter()
-                            .map(|(key, value)| (key.clone(), value.clone()))
-                            .collect(),
-                    )
-                    .load_session(acp_session_id);
-                    match maestro_acp::spawn(config).await {
-                        Ok(acp) => (Some(Arc::new(acp)), None),
-                        Err(error) => {
-                            tracing::warn!(%id, %error, "structured session: adapter failed to resume");
-                            (None, Some(error.to_string()))
+            let (acp, failure) = if recoverable {
+                match session.acp_session_id.clone() {
+                    Some(acp_session_id) => {
+                        let config = SpawnConfig::new(
+                            &session.cwd,
+                            session_env
+                                .iter()
+                                .map(|(key, value)| (key.clone(), value.clone()))
+                                .collect(),
+                        )
+                        .load_session(acp_session_id);
+                        match maestro_acp::spawn(config).await {
+                            Ok(acp) => (Some(Arc::new(acp)), None),
+                            Err(error) => {
+                                tracing::warn!(%id, %error, "structured session: adapter failed to resume");
+                                (None, Some(error.to_string()))
+                            }
                         }
                     }
+                    None => (
+                        None,
+                        Some("structured session has no stored ACP session id".into()),
+                    ),
                 }
-                None => (
-                    None,
-                    Some("structured session has no stored ACP session id".into()),
-                ),
+            } else {
+                (None, None)
             };
-            session.status = if failure.is_some() {
+            session.status = if recoverable {
+                if failure.is_some() {
+                    SessionStatus::Dead
+                } else {
+                    SessionStatus::Running
+                }
+            } else {
                 SessionStatus::Dead
-            } else {
-                SessionStatus::Running
             };
-            session.activity = if failure.is_some() {
-                ActivityState::Error
-            } else {
-                ActivityState::Active
-            };
-            session.activity_detail = failure;
+            if recoverable {
+                session.activity = if failure.is_some() {
+                    ActivityState::Error
+                } else {
+                    ActivityState::Active
+                };
+                session.activity_detail = failure;
+            }
 
             let acp_events = if let Some(acp) = &acp {
+                let persisted =
+                    match self
+                        .inner
+                        .sidecar
+                        .read_events(id, EventCursor::Since(0), usize::MAX)
+                    {
+                        Ok(events) => events,
+                        Err(error) => {
+                            acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
+                                .await;
+                            let _ = std::fs::remove_file(&fifo_path);
+                            return Err(SessionError::Sidecar(error));
+                        }
+                    };
                 match self.inner.sidecar.open_events(id) {
                     Ok(log) => {
                         let (tx, _) = broadcast::channel(EVENTS_CAPACITY);
-                        Some(Arc::new(Mutex::new(AcpEventStream { log, live: tx })))
+                        let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
+                        let mut projection = Projection::default();
+                        for event in &persisted {
+                            projection.apply(event.seq, &event.notification);
+                        }
+                        Some(Arc::new(Mutex::new(AcpEventStream {
+                            log,
+                            live: tx,
+                            projection,
+                            projected_live: projected_tx,
+                        })))
                     }
                     Err(error) => {
                         acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
@@ -1379,7 +1674,9 @@ fn start_acp_event_pump(
                 let mut events = events.lock().await;
                 match events.log.append(&notification) {
                     Ok(seq) => {
+                        let projected = events.projection.apply(seq, &notification);
                         let _ = events.live.send(LoggedEvent { seq, notification });
+                        let _ = events.projected_live.send(projected);
                         None
                     }
                     Err(error) => Some(error),
@@ -1438,6 +1735,9 @@ async fn probe_worktree(
 /// One pass of the liveness watcher: diff the in-memory session set against
 /// the tmux-reported one, and reap anything we still track but tmux has lost.
 async fn run_liveness_pass(inner: &Arc<Inner>) {
+    if inner.shutting_down.load(Ordering::Acquire) {
+        return;
+    }
     // Snapshot the in-memory ids first; if we own no sessions there is
     // nothing to reap and no reason to spawn a tmux subprocess (this also
     // avoids racing with the very first `new-session` that would otherwise
@@ -1497,6 +1797,9 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
 /// `SessionUpdated` only when `activity` or `foreground` actually changes — so a
 /// steady state produces no diffs. Runs on the existing liveness tick.
 async fn run_activity_pass(inner: &Arc<Inner>) {
+    if inner.shutting_down.load(Ordering::Acquire) {
+        return;
+    }
     let ids: Vec<Uuid> = {
         let sessions = inner.sessions.lock().await;
         if sessions.is_empty() {
@@ -1539,6 +1842,9 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
 
         let updated = {
             let mut sessions = inner.sessions.lock().await;
+            if inner.shutting_down.load(Ordering::Acquire) {
+                return;
+            }
             let Some(rec) = sessions.get_mut(&id) else {
                 continue;
             };
@@ -1613,6 +1919,9 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
 /// error must not read as "nothing blocks". Like the activity pass, this only
 /// emits `SessionUpdated` when the value actually changed.
 async fn run_blocked_work_pass(inner: &Arc<Inner>) {
+    if inner.shutting_down.load(Ordering::Acquire) {
+        return;
+    }
     let targets: Vec<(Uuid, (PathBuf, PathBuf))> = {
         let sessions = inner.sessions.lock().await;
         sessions
@@ -1640,6 +1949,9 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
         };
         let updated = {
             let mut sessions = inner.sessions.lock().await;
+            if inner.shutting_down.load(Ordering::Acquire) {
+                return;
+            }
             let Some(rec) = sessions.get_mut(&id) else {
                 continue;
             };
@@ -1672,20 +1984,25 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
 /// [`SessionManager::recover`] adopts only what tmux still holds. Structured
 /// sessions stopped for daemon shutdown retain their sidecars for ACP recovery.
 async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
+    if inner.shutting_down.load(Ordering::Acquire) {
+        return;
+    }
     mark_session_dead_with_worktree_settlement(inner, id, true).await;
 }
 
 /// A daemon restart terminates ACP children, but it is not a natural adapter
 /// exit: the sidecar is the recovery inventory, including its worktree.
 async fn mark_session_dead_after_shutdown(inner: &Arc<Inner>, id: Uuid) {
-    mark_session_dead_with_worktree_settlement(inner, id, false).await;
+    if mark_session_dead_with_worktree_settlement(inner, id, false).await {
+        let _ = inner.sidecar.mark_for_recovery(id);
+    }
 }
 
 async fn mark_session_dead_with_worktree_settlement(
     inner: &Arc<Inner>,
     id: Uuid,
     settle_worktree: bool,
-) {
+) -> bool {
     let flipped = {
         let mut sessions = inner.sessions.lock().await;
         match sessions.get_mut(&id) {
@@ -1705,7 +2022,7 @@ async fn mark_session_dead_with_worktree_settlement(
         }
     };
     let Some((mut session, worktree)) = flipped else {
-        return;
+        return false;
     };
     tracing::info!(%id, "session process gone — marking dead");
 
@@ -1736,6 +2053,7 @@ async fn mark_session_dead_with_worktree_settlement(
 
     let _ = inner.sidecar.write(&session);
     let _ = inner.events.send(CatalogEvent::SessionUpdated { session });
+    true
 }
 
 /// Apply the exit rule (#82) to a vanished session's worktree: probe what

crates/maestro-daemon/src/sidecar.rs

@@ -56,6 +56,28 @@ impl SidecarStore {
         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
@@ -164,10 +186,11 @@ impl SidecarStore {
         }
     }
 
-    /// Delete the sidecar and the event log for `id`. No-op for whichever
-    /// 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> {
         remove_if_present(&self.events_path_for(id))?;
+        remove_if_present(&self.recovery_path_for(id))?;
         remove_if_present(&self.path_for(id))
     }
 
@@ -224,7 +247,7 @@ pub enum EventCursor {
 }
 
 /// One line of a session's event log.
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, serde::Serialize)]
 pub struct LoggedEvent {
     /// 1-based line position in the log; the cursor for [`EventCursor`].
     pub seq: u64,

crates/maestro-daemon/tests/acp.rs

@@ -12,11 +12,13 @@ use std::time::{Duration, Instant};
 use common::{init_bare_repo, init_repo, push_all_to_origin, wait_for_async, DaemonFixture};
 use maestro_acp::schema::{ContentBlock, SessionUpdate};
 use maestro_acp::{ADAPTER_BIN, INSTALL_COMMAND};
+use maestro_daemon::projection::TurnLifecycle;
 use maestro_daemon::sidecar::{EventCursor, SidecarStore};
 use maestro_daemon::worktree::WorktreeManager;
 use maestro_daemon::SessionManager;
 use maestro_protocol::{
-    ActivityState, CreateSessionRequest, SessionStatus, SessionView, WorktreeSpec,
+    ActivityState, CreateSessionRequest, SessionStatus, SessionView, TranscriptDelta,
+    TranscriptRowKind, WorktreeSpec,
 };
 use tempfile::TempDir;
 
@@ -199,6 +201,126 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
         .exists());
 }
 
+#[tokio::test]
+async fn projection_snapshot_and_live_stream_share_one_fold() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), fake.env()))
+        .await
+        .expect("create");
+
+    wait_for_async(Duration::from_secs(5), || async {
+        fx.manager
+            .read_acp_events(session.id, EventCursor::Since(0), 10)
+            .is_ok_and(|events| events.len() == 1)
+    })
+    .await;
+
+    // The cursor does not truncate either projection: a reconnect needs the
+    // source row for future patches and whole reduced state from the session.
+    let (snapshot, state, mut live) = fx
+        .manager
+        .subscribe_acp_projection(session.id, EventCursor::Since(1))
+        .await
+        .expect("projection subscription")
+        .expect("structured session");
+    assert_eq!(snapshot.cursor, 1);
+    assert_eq!(snapshot.rows.len(), 1);
+    assert_eq!(snapshot.rows[0].kind, TranscriptRowKind::AgentText);
+    assert_eq!(state.turn, Default::default());
+
+    let maestro_daemon::session::BeginAcpTurn::Started(_, turn) = fx
+        .manager
+        .begin_acp_turn(session.id)
+        .await
+        .expect("begin lifecycle")
+    else {
+        panic!("structured turn should start");
+    };
+    let begin = tokio::time::timeout(Duration::from_secs(5), live.recv())
+        .await
+        .expect("begin lifecycle broadcast")
+        .expect("projection broadcast");
+    assert_eq!(begin.transcript, None);
+    assert!(begin.state.turn.responding);
+    assert!(begin.state.turn.thinking);
+
+    unsafe {
+        libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1);
+    }
+    let agent_update = tokio::time::timeout(Duration::from_secs(5), live.recv())
+        .await
+        .expect("live projection")
+        .expect("projection broadcast");
+    match agent_update.transcript {
+        Some(TranscriptDelta::Append { row }) => {
+            assert_eq!(row.kind, TranscriptRowKind::AgentText);
+            assert_eq!(row.data["content"]["text"], "tick 1");
+        }
+        other => panic!("expected one appended transcript row, got {other:?}"),
+    }
+    assert!(agent_update.state.turn.responding);
+    assert!(!agent_update.state.turn.thinking);
+
+    assert!(fx
+        .manager
+        .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish, turn)
+        .await
+        .expect("finish lifecycle"));
+    let finish = tokio::time::timeout(Duration::from_secs(5), live.recv())
+        .await
+        .expect("finish lifecycle broadcast")
+        .expect("projection broadcast");
+    assert_eq!(finish.transcript, None);
+    assert_eq!(finish.state.turn, Default::default());
+
+    fx.manager.kill(session.id, false).await.expect("kill");
+}
+
+#[tokio::test]
+async fn stale_turn_completion_cannot_end_the_next_turn() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), fake.env()))
+        .await
+        .unwrap();
+
+    let maestro_daemon::session::BeginAcpTurn::Started(_, first) =
+        fx.manager.begin_acp_turn(session.id).await.unwrap()
+    else {
+        panic!("first turn should start");
+    };
+    assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
+    let maestro_daemon::session::BeginAcpTurn::Started(_, second) =
+        fx.manager.begin_acp_turn(session.id).await.unwrap()
+    else {
+        panic!("second turn should start");
+    };
+    assert!(second > first);
+    assert!(!fx
+        .manager
+        .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish, first)
+        .await
+        .unwrap());
+    assert!(matches!(
+        fx.manager.begin_acp_turn(session.id).await.unwrap(),
+        maestro_daemon::session::BeginAcpTurn::InFlight
+    ));
+    assert!(fx
+        .manager
+        .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish, second)
+        .await
+        .unwrap());
+
+    fx.manager.kill(session.id, false).await.unwrap();
+}
+
 #[tokio::test]
 async fn recovery_loads_the_acp_session_without_duplicating_history() {
     let fx = DaemonFixture::new();
@@ -232,6 +354,12 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
     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()),
@@ -498,6 +626,144 @@ 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();
@@ -556,7 +822,27 @@ async fn adapter_exit_marks_the_session_dead_and_keeps_it_listed() {
     })
     .await;
     assert!(fx.manager.has(session.id).await);
+    assert!(
+        !fx.sidecar_dir()
+            .join(format!("{}.recover", session.id))
+            .exists(),
+        "a naturally exited adapter is never marked for recovery"
+    );
 
-    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,16 @@ mod common;
 
 use std::collections::HashMap;
 use std::net::{Ipv4Addr, SocketAddr};
+use std::path::Path;
 use std::time::Duration;
 
 use common::DaemonFixture;
 use futures_util::{SinkExt, StreamExt};
 use maestro_daemon::local_api;
-use maestro_protocol::{ApiError, CreateSessionRequest, Session, WorktreeSpec};
+use maestro_protocol::{
+    ApiError, CreateSessionRequest, Session, SessionStatus, SessionView, WorktreeSpec,
+};
+use tempfile::TempDir;
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
 use tokio_tungstenite::tungstenite::Message as WsMessage;
@@ -86,6 +90,430 @@ fn ws_url(addr: SocketAddr, path: &str) -> String {
     format!("ws://{addr}{path}")
 }
 
+struct FakeAcp {
+    dir: TempDir,
+}
+
+impl FakeAcp {
+    fn new() -> Self {
+        use std::os::unix::fs::PermissionsExt;
+
+        let dir = TempDir::new().unwrap();
+        let bin = dir.path().join("bin");
+        std::fs::create_dir_all(&bin).unwrap();
+        let adapter = bin.join("claude-code-acp");
+        std::fs::copy(
+            Path::new(env!("CARGO_MANIFEST_DIR")).join("../maestro-acp/tests/fake_acp_agent.py"),
+            &adapter,
+        )
+        .unwrap();
+        std::fs::set_permissions(&adapter, std::fs::Permissions::from_mode(0o755)).unwrap();
+        Self { dir }
+    }
+
+    fn env(&self) -> HashMap<String, String> {
+        HashMap::from([
+            (
+                "PATH".into(),
+                format!(
+                    "{}:{}",
+                    self.dir.path().join("bin").display(),
+                    std::env::var("PATH").unwrap()
+                ),
+            ),
+            (
+                "FAKE_ACP_LOG".into(),
+                self.dir.path().join("agent.log").display().to_string(),
+            ),
+        ])
+    }
+
+    fn pid(&self) -> u32 {
+        std::fs::read_to_string(self.dir.path().join("agent.log"))
+            .unwrap_or_default()
+            .lines()
+            .filter_map(|line| {
+                line.strip_prefix("start pid=")?
+                    .split(' ')
+                    .next()?
+                    .parse()
+                    .ok()
+            })
+            .next_back()
+            .expect("fake ACP started")
+    }
+}
+
+async fn structured_session(h: &Harness, fake: &FakeAcp) -> Uuid {
+    structured_session_with_env(h, fake.env()).await
+}
+
+async fn structured_session_with_env(h: &Harness, env: HashMap<String, String>) -> Uuid {
+    h.manager
+        .create(CreateSessionRequest {
+            view: SessionView::Structured,
+            cwd: "/tmp".into(),
+            command: "claude".into(),
+            env,
+            metadata: HashMap::new(),
+            worktree: None,
+        })
+        .await
+        .unwrap()
+        .id
+}
+
+#[tokio::test]
+async fn structured_ws_sends_snapshot_then_live_projection() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+    let (mut ws, _) = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
+        .await
+        .unwrap();
+
+    let first = ws.next().await.unwrap().unwrap().into_text().unwrap();
+    let second = ws.next().await.unwrap().unwrap().into_text().unwrap();
+    assert_eq!(
+        serde_json::from_str::<serde_json::Value>(&first).unwrap()["type"],
+        "transcript_snapshot"
+    );
+    assert_eq!(
+        serde_json::from_str::<serde_json::Value>(&second).unwrap()["type"],
+        "reduced_state"
+    );
+
+    let response = reqwest::Client::new()
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "hello"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);
+    let live = tokio::time::timeout(Duration::from_secs(1), ws.next())
+        .await
+        .expect("projection update arrives on the subscribed stream")
+        .unwrap()
+        .unwrap()
+        .into_text()
+        .unwrap();
+    assert_eq!(
+        serde_json::from_str::<serde_json::Value>(&live).unwrap()["type"],
+        "projection_update"
+    );
+}
+
+#[tokio::test]
+async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_DELAY_PROMPT_MS".into(), "200".into());
+    let id = structured_session_with_env(&h, env).await;
+    let client = reqwest::Client::new();
+    let first = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "first"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(first.status(), reqwest::StatusCode::NO_CONTENT);
+    let second = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "second"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(second.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = second.json().await.unwrap();
+    assert_eq!(error.error.code, "turn_in_flight");
+
+    let cancel = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/cancel"),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(cancel.status(), reqwest::StatusCode::NO_CONTENT);
+    let next = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "next"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(next.status(), reqwest::StatusCode::NO_CONTENT);
+}
+
+#[tokio::test]
+async fn structured_prompt_failure_clears_the_turn() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_PROMPT_FAIL_ONCE".into(), "1".into());
+    let id = structured_session_with_env(&h, env).await;
+    let client = reqwest::Client::new();
+    let path = format!("/api/sessions/{id}/structured/prompt");
+    let (mut ws, _) = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
+        .await
+        .unwrap();
+    ws.next().await.unwrap().unwrap();
+    ws.next().await.unwrap().unwrap();
+    let first = client
+        .post(url(h.addr, &path))
+        .json(&serde_json::json!({"prompt":"fails"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(first.status(), reqwest::StatusCode::NO_CONTENT);
+    let mut saw_active_turn = false;
+    tokio::time::timeout(Duration::from_secs(1), async {
+        while let Some(Ok(message)) = ws.next().await {
+            let event: serde_json::Value =
+                serde_json::from_str(&message.into_text().unwrap()).unwrap();
+            let turn = &event["state"]["turn"];
+            if turn["responding"] == true {
+                saw_active_turn = true;
+            }
+            if saw_active_turn && turn["responding"] == false && turn["thinking"] == false {
+                return;
+            }
+        }
+        panic!("structured websocket closed before the failed turn cleared");
+    })
+    .await
+    .expect("failed turn clears promptly");
+    let second = client
+        .post(url(h.addr, &path))
+        .json(&serde_json::json!({"prompt":"available"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(second.status(), reqwest::StatusCode::NO_CONTENT);
+    assert!(second.bytes().await.unwrap().is_empty());
+}
+
+#[tokio::test]
+async fn structured_prompt_rejects_unknown_and_unavailable_sessions() {
+    let h = spawn("naps-laptop").await;
+    let client = reqwest::Client::new();
+    let prompt = |id| {
+        client
+            .post(url(
+                h.addr,
+                &format!("/api/sessions/{id}/structured/prompt"),
+            ))
+            .json(&serde_json::json!({"prompt": "hello"}))
+    };
+
+    let unknown = prompt(Uuid::new_v4()).send().await.unwrap();
+    assert_eq!(unknown.status(), reqwest::StatusCode::NOT_FOUND);
+    let error: ApiError = unknown.json().await.unwrap();
+    assert_eq!(error.error.code, "not_found");
+
+    let terminal = h
+        .manager
+        .create(CreateSessionRequest {
+            view: SessionView::Terminal,
+            cwd: "/tmp".into(),
+            command: "sh".into(),
+            env: HashMap::new(),
+            metadata: HashMap::new(),
+            worktree: None,
+        })
+        .await
+        .unwrap();
+    let non_structured = prompt(terminal.id).send().await.unwrap();
+    assert_eq!(non_structured.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = non_structured.json().await.unwrap();
+    assert_eq!(error.error.code, "not_structured");
+
+    let fake = FakeAcp::new();
+    let dead = structured_session(&h, &fake).await;
+    h.manager.shutdown().await;
+    assert_eq!(h.manager.status(dead).await, Some(SessionStatus::Dead));
+    let unavailable = prompt(dead).send().await.unwrap();
+    assert_eq!(unavailable.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = unavailable.json().await.unwrap();
+    assert_eq!(error.error.code, "not_structured");
+}
+
+#[tokio::test]
+async fn structured_replay_honors_cursors_and_limit() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+    let client = reqwest::Client::new();
+
+    let events: Vec<serde_json::Value> = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let events: Vec<serde_json::Value> = client
+                .get(url(
+                    h.addr,
+                    &format!("/api/sessions/{id}/structured/replay?since=0&limit=1"),
+                ))
+                .send()
+                .await
+                .unwrap()
+                .json()
+                .await
+                .unwrap();
+            if !events.is_empty() {
+                return events;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(events.len(), 1);
+    assert_eq!(events[0]["cursor"], 1);
+    assert!(events[0].get("notification").is_none());
+
+    let invalid = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/replay?since=0&before=2"),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(invalid.status(), reqwest::StatusCode::BAD_REQUEST);
+}
+
+#[tokio::test]
+async fn structured_replay_pages_projected_events_in_cursor_order() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+    let client = reqwest::Client::new();
+    unsafe { libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1) };
+    let events: Vec<serde_json::Value> = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let events: Vec<serde_json::Value> = client
+                .get(url(
+                    h.addr,
+                    &format!("/api/sessions/{id}/structured/replay?before=99&limit=2"),
+                ))
+                .send()
+                .await
+                .unwrap()
+                .json()
+                .await
+                .unwrap();
+            if events.len() == 2 {
+                return events;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(
+        events
+            .iter()
+            .map(|event| event["cursor"].as_u64().unwrap())
+            .collect::<Vec<_>>(),
+        vec![1, 2]
+    );
+    let boundary: Vec<serde_json::Value> = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/replay?before=2&limit=10"),
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert_eq!(boundary.len(), 1);
+    assert_eq!(boundary[0]["cursor"], 1);
+}
+
+#[tokio::test]
+async fn structured_replay_rejects_unknown_terminal_and_dead_sessions() {
+    let h = spawn("naps-laptop").await;
+    let client = reqwest::Client::new();
+    let unknown = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{}/structured/replay", Uuid::new_v4()),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(unknown.status(), reqwest::StatusCode::NOT_FOUND);
+    let terminal = h
+        .manager
+        .create(CreateSessionRequest {
+            view: SessionView::Terminal,
+            cwd: "/tmp".into(),
+            command: "sh".into(),
+            env: HashMap::new(),
+            metadata: HashMap::new(),
+            worktree: None,
+        })
+        .await
+        .unwrap();
+    let response = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{}/structured/replay", terminal.id),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
+
+    let fake = FakeAcp::new();
+    let dead = structured_session(&h, &fake).await;
+    h.manager.shutdown().await;
+    assert_eq!(h.manager.status(dead).await, Some(SessionStatus::Dead));
+    let response = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{dead}/structured/replay"),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = response.json().await.unwrap();
+    assert_eq!(error.error.code, "not_structured");
+}
+
+#[tokio::test]
+async fn structured_ws_rejects_a_dead_session() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+    h.manager.shutdown().await;
+    assert_eq!(h.manager.status(id).await, Some(SessionStatus::Dead));
+
+    let err = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
+        .await
+        .expect_err("dead structured session must not upgrade");
+    let message = err.to_string();
+    assert!(
+        message.contains("409") || message.to_lowercase().contains("conflict"),
+        "{message}"
+    );
+}
+
 #[tokio::test]
 async fn info_returns_daemon_identity() {
     let h = spawn("naps-laptop").await;

crates/maestro-daemon/tests/sidecar.rs

@@ -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();

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

@@ -755,6 +755,123 @@ pub enum CatalogEvent {
     },
 }
 
+// ---------------------------------------------------------------------------
+// Structured-session projections
+// ---------------------------------------------------------------------------
+
+/// One client-neutral item in a structured session transcript.
+///
+/// `data` preserves the ACP payload without committing clients to an ACP
+/// dependency. Rendering choices, including markdown and tool-card treatment,
+/// stay with the client.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct TranscriptRow {
+    /// Stable within one session, so later deltas can address this row.
+    pub id: String,
+    pub kind: TranscriptRowKind,
+    pub data: Value,
+}
+
+/// The semantic shape of a [`TranscriptRow`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum TranscriptRowKind {
+    User,
+    AgentText,
+    AgentThought,
+    ToolCall,
+    ToolDiff,
+    Approval,
+    Plan,
+    Notice,
+}
+
+/// A complete transcript at a durable event-log position.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct TranscriptSnapshot {
+    pub cursor: u64,
+    #[serde(default)]
+    pub rows: Vec<TranscriptRow>,
+}
+
+/// One durable, client-neutral replay step from a structured session.
+///
+/// Results are ordered by increasing `cursor`. `since` returns the first
+/// matching steps; `before` returns the final matching steps, still increasing.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct StructuredReplayEvent {
+    /// Durable ACP-log position represented by this projected step.
+    pub cursor: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub transcript: Option<TranscriptDelta>,
+    pub state: ReducedState,
+}
+
+/// A change to a transcript row.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum TranscriptDelta {
+    Append { row: TranscriptRow },
+    Patch { row: TranscriptRow },
+    Remove { id: String },
+}
+
+/// Per-turn indicators derived from the ACP stream.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct TurnState {
+    #[serde(default)]
+    pub responding: bool,
+    #[serde(default)]
+    pub thinking: bool,
+    #[serde(default)]
+    pub active_tool_calls: Vec<String>,
+}
+
+/// A whole-session view derived from ACP notifications.
+///
+/// Values whose shape ACP owns are carried as JSON so this wire protocol stays
+/// forward-compatible when ACP adds fields or variants.
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+pub struct ReducedState {
+    #[serde(default)]
+    pub turn: TurnState,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub pending_approval: Option<Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub plan: Option<Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub usage: Option<Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub context: Option<Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub model: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub mode: Option<String>,
+}
+
+/// Server messages sent over a structured-session WebSocket.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum StructuredEvent {
+    TranscriptSnapshot {
+        snapshot: TranscriptSnapshot,
+    },
+    ReducedState {
+        state: ReducedState,
+    },
+    ProjectionUpdate {
+        #[serde(skip_serializing_if = "Option::is_none")]
+        transcript: Option<TranscriptDelta>,
+        state: ReducedState,
+    },
+}
+
+/// HTTP body for `POST /api/sessions/:id/structured/prompt`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct StructuredPromptRequest {
+    pub prompt: String,
+}
+
 // ---------------------------------------------------------------------------
 // Client ↔ Daemon control messages (text frames on the attach WS)
 // ---------------------------------------------------------------------------
@@ -1144,6 +1261,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,26 @@ pub fn attach(id: Uuid) -> String {
     format!("{API_PREFIX}/attach/{id}")
 }
 
+/// `WS /api/structured/:id` — transcript/state snapshots and live deltas.
+pub fn structured(id: Uuid) -> String {
+    format!("{API_PREFIX}/structured/{id}")
+}
+
+/// `GET /api/sessions/:id/structured/replay` — cursor-bounded ACP history.
+pub fn structured_replay(id: Uuid) -> String {
+    format!("{SESSIONS}/{id}/structured/replay")
+}
+
+/// `POST /api/sessions/:id/structured/prompt` — send an ACP prompt.
+pub fn structured_prompt(id: Uuid) -> String {
+    format!("{SESSIONS}/{id}/structured/prompt")
+}
+
+/// `POST /api/sessions/:id/structured/cancel` — cancel an ACP turn.
+pub fn structured_cancel(id: Uuid) -> String {
+    format!("{SESSIONS}/{id}/structured/cancel")
+}
+
 /// Axum-style patterns for the two parameterised routes, so the server
 /// registers the same paths [`session`] and [`attach`] build.
 pub mod patterns {
@@ -80,6 +100,10 @@ pub mod patterns {
     pub const SESSION_PROMPT: &str = "/api/sessions/:id/prompt";
     pub const SESSION_FOCUS: &str = "/api/sessions/:id/focus";
     pub const ATTACH: &str = "/api/attach/:id";
+    pub const STRUCTURED: &str = "/api/structured/:id";
+    pub const STRUCTURED_REPLAY: &str = "/api/sessions/:id/structured/replay";
+    pub const STRUCTURED_PROMPT: &str = "/api/sessions/:id/structured/prompt";
+    pub const STRUCTURED_CANCEL: &str = "/api/sessions/:id/structured/cancel";
 }
 
 /// A daemon base URL that could not be turned into a WebSocket base.
@@ -206,6 +230,22 @@ mod tests {
             attach(id()),
             "/api/attach/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50"
         );
+        assert_eq!(
+            structured(id()),
+            "/api/structured/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50"
+        );
+        assert_eq!(
+            structured_replay(id()),
+            "/api/sessions/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50/structured/replay"
+        );
+        assert_eq!(
+            structured_prompt(id()),
+            "/api/sessions/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50/structured/prompt"
+        );
+        assert_eq!(
+            structured_cancel(id()),
+            "/api/sessions/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50/structured/cancel"
+        );
         assert_eq!(
             attach_url("https://laptop.lan:7070", id()).unwrap(),
             "wss://laptop.lan:7070/api/attach/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50"
@@ -240,6 +280,22 @@ mod tests {
             patterns::ATTACH.trim_end_matches(":id"),
             attach(id()).trim_end_matches(&id().to_string())
         );
+        assert_eq!(
+            patterns::STRUCTURED.trim_end_matches(":id"),
+            structured(id()).trim_end_matches(&id().to_string())
+        );
+        assert_eq!(
+            patterns::STRUCTURED_REPLAY.trim_end_matches(":id/structured/replay"),
+            session(id()).trim_end_matches(&id().to_string())
+        );
+        assert_eq!(
+            patterns::STRUCTURED_PROMPT.trim_end_matches(":id/structured/prompt"),
+            session(id()).trim_end_matches(&id().to_string())
+        );
+        assert_eq!(
+            patterns::STRUCTURED_CANCEL.trim_end_matches(":id/structured/cancel"),
+            session(id()).trim_end_matches(&id().to_string())
+        );
     }
 
     #[test]
@@ -258,5 +314,9 @@ mod tests {
             assert!(path.starts_with(API_PREFIX), "{path} escapes {API_PREFIX}");
         }
         assert!(attach(id()).starts_with(API_PREFIX));
+        assert!(structured(id()).starts_with(API_PREFIX));
+        assert!(structured_replay(id()).starts_with(API_PREFIX));
+        assert!(structured_prompt(id()).starts_with(API_PREFIX));
+        assert!(structured_cancel(id()).starts_with(API_PREFIX));
     }
 }