test(daemon): cover projection lifecycle

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

@@ -158,6 +158,7 @@ enum Command {
 
 enum ChildCommand {
     Signal(libc::c_int),
+    Reap(oneshot::Sender<bool>),
 }
 
 /// Why the handshake did not produce a session id.
@@ -206,6 +207,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
@@ -392,6 +405,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

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)
@@ -36,6 +38,7 @@ import json
 import os
 import signal
 import sys
+import time
 
 LOG = os.environ.get("FAKE_ACP_LOG")
 
@@ -113,6 +116,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":

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

@@ -170,6 +170,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,
     }
 }

crates/maestro-daemon/src/projection.rs

@@ -0,0 +1,608 @@
+//! 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,
+    turn_active: bool,
+    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()
+    }
+
+    /// Apply a transport-owned prompt lifecycle transition.
+    pub fn turn_lifecycle(&mut self, lifecycle: TurnLifecycle) -> ProjectionUpdate {
+        match lifecycle {
+            TurnLifecycle::Begin => {
+                self.turn_active = true;
+                self.state.turn.responding = true;
+                self.state.turn.thinking = true;
+                self.turn_tool_calls.clear();
+                self.state.turn.active_tool_calls.clear();
+            }
+            TurnLifecycle::Finish | TurnLifecycle::Cancel | TurnLifecycle::Failure => {
+                self.turn_active = false;
+                self.state.turn.responding = false;
+                self.state.turn.thinking = false;
+                self.turn_tool_calls.clear();
+                self.state.turn.active_tool_calls.clear();
+            }
+        }
+        ProjectionUpdate {
+            transcript: None,
+            state: self.state(),
+        }
+    }
+
+    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());
+
+        projection.turn_lifecycle(TurnLifecycle::Begin);
+        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,
+        ] {
+            projection.turn_lifecycle(TurnLifecycle::Begin);
+            projection.turn_lifecycle(terminal);
+            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();
+        projection.turn_lifecycle(TurnLifecycle::Begin);
+        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);
+        assert!(projection.state().turn.active_tool_calls.is_empty());
+        assert_eq!(projection.snapshot(1).rows.len(), 1);
+
+        projection.turn_lifecycle(TurnLifecycle::Begin);
+        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;
 
@@ -21,6 +21,7 @@ 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};
@@ -106,6 +107,8 @@ pub struct SessionRecord {
 struct AcpEventStream {
     log: EventLog,
     live: broadcast::Sender<LoggedEvent>,
+    projection: Projection,
+    projected_live: broadcast::Sender<ProjectionUpdate>,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -127,6 +130,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.
@@ -144,6 +149,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 +201,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 +249,8 @@ impl SessionManager {
                 events,
                 viewports: Mutex::new(Viewports::new()),
                 next_viewport: AtomicU64::new(0),
+                shutting_down: AtomicBool::new(false),
+                structured_lifecycle: Mutex::new(()),
             }),
         }
     }
@@ -343,19 +357,84 @@ impl SessionManager {
             .map_err(SessionError::Sidecar)
     }
 
+    /// 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 Some(record) = sessions.get(&id) else {
+                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,
+    ) -> 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 projected = stream.projection.turn_lifecycle(lifecycle);
+        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 +442,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,8 +470,8 @@ impl SessionManager {
             .filter_map(|(_, _, done)| done.clone())
             .map(wait_for_event_pump);
         futures_util::future::join_all(drains).await;
-        for (id, _, _) in children {
-            mark_session_dead(&self.inner, id).await;
+        for (id, _, _) in live_children {
+            mark_session_dead_after_shutdown(&self.inner, id).await;
         }
     }
 
@@ -457,6 +554,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 +670,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 +1036,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 +1051,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 +1067,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 +1547,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 +1608,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 +1670,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 +1715,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 +1792,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 +1822,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;
             };
@@ -1656,7 +1841,7 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
     }
 }
 
-/// Mark a session whose tmux session has disappeared as [`SessionStatus::Dead`]
+/// Mark a session whose backing process has disappeared as [`SessionStatus::Dead`]
 /// and settle its worktree per the natural-exit rule (#82).
 ///
 /// The record, its [`PtyBridge`] and its sidecar all survive, so a client can
@@ -1668,9 +1853,29 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
 /// alone and stays attached to the dead record for a later forced dismissal
 /// (#81), which leaves it on disk too.
 ///
-/// Dead sessions do not survive a daemon restart: [`SessionManager::recover`]
-/// adopts only what tmux still holds.
+/// Terminal sessions that die naturally do not survive a daemon restart:
+/// [`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) {
+    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) {
@@ -1690,21 +1895,23 @@ async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
         }
     };
     let Some((mut session, worktree)) = flipped else {
-        return;
+        return false;
     };
     tracing::info!(%id, "session process gone — marking dead");
 
-    if let Some(wt) = &worktree {
-        if settle_worktree_on_exit(inner, id, wt).await {
-            // The tree is settled (removed); a stale "in the way" would
-            // describe something the session no longer holds.
-            session.blocked_work = None;
-            let mut sessions = inner.sessions.lock().await;
-            if let Some(record) = sessions.get_mut(&id) {
-                record.session.worktree = None;
-                record.session.blocked_work = None;
-                record.worktree_path = None;
-                record.worktree_base_repo = None;
+    if settle_worktree {
+        if let Some(wt) = &worktree {
+            if settle_worktree_on_exit(inner, id, wt).await {
+                // The tree is settled (removed); a stale "in the way" would
+                // describe something the session no longer holds.
+                session.blocked_work = None;
+                let mut sessions = inner.sessions.lock().await;
+                if let Some(record) = sessions.get_mut(&id) {
+                    record.session.worktree = None;
+                    record.session.blocked_work = None;
+                    record.worktree_path = None;
+                    record.worktree_base_repo = None;
+                }
             }
         }
     }
@@ -1719,6 +1926,7 @@ async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
 
     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))
     }
 

crates/maestro-daemon/tests/acp.rs

@@ -9,13 +9,17 @@ use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use common::{wait_for_async, DaemonFixture};
+use common::{init_bare_repo, init_repo, push_all_to_origin, wait_for_async, DaemonFixture};
 use maestro_acp::schema::{ContentBlock, SessionUpdate};
 use maestro_acp::{ADAPTER_BIN, INSTALL_COMMAND};
+use maestro_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};
+use maestro_protocol::{
+    ActivityState, CreateSessionRequest, SessionStatus, SessionView, TranscriptDelta,
+    TranscriptRowKind, WorktreeSpec,
+};
 use tempfile::TempDir;
 
 fn fake_agent() -> PathBuf {
@@ -197,6 +201,82 @@ 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());
+
+    assert!(fx
+        .manager
+        .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Begin)
+        .await
+        .expect("begin lifecycle"));
+    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)
+        .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 recovery_loads_the_acp_session_without_duplicating_history() {
     let fx = DaemonFixture::new();
@@ -230,6 +310,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()),
@@ -285,6 +371,76 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
     revived.kill(session.id, false).await.expect("kill");
 }
 
+#[tokio::test]
+async fn shutdown_preserves_a_pristine_structured_worktree_for_recovery() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let repo = TempDir::new().unwrap();
+    init_repo(repo.path());
+    let origin = TempDir::new().unwrap();
+    init_bare_repo(origin.path());
+
+    let branch = "feature/restart";
+    let worktree = repo
+        .path()
+        .join("worktrees")
+        .join("feature")
+        .join("restart");
+    let session = fx
+        .manager
+        .create(CreateSessionRequest {
+            view: SessionView::Structured,
+            cwd: repo.path().to_path_buf(),
+            command: "claude".into(),
+            env: fake.env(),
+            metadata: HashMap::new(),
+            worktree: Some(WorktreeSpec {
+                base_repo: repo.path().to_path_buf(),
+                branch: branch.into(),
+                create_branch: true,
+            }),
+        })
+        .await
+        .expect("create");
+    assert_eq!(session.cwd, worktree);
+    push_all_to_origin(repo.path(), origin.path());
+
+    fx.manager.shutdown().await;
+
+    assert!(
+        worktree.is_dir(),
+        "shutdown must not settle a pristine worktree"
+    );
+    let sidecar: maestro_protocol::Session = serde_json::from_slice(
+        &std::fs::read(fx.sidecar_dir().join(format!("{}.json", session.id))).unwrap(),
+    )
+    .unwrap();
+    assert_eq!(
+        sidecar.worktree.as_ref().map(|wt| &wt.path),
+        Some(&worktree)
+    );
+
+    let revived = SessionManager::new(
+        Arc::new(fx.tmux.tmux.clone()),
+        SidecarStore::new(fx.sidecar_dir()),
+        WorktreeManager::new(None),
+        fx.fifo_dir(),
+    );
+    assert_eq!(revived.recover().await.expect("recover"), 1);
+    assert_eq!(
+        revived.status(session.id).await,
+        Some(SessionStatus::Running)
+    );
+    assert!(fake.log().contains("recv session/load"), "{}", fake.log());
+    assert!(worktree.is_dir(), "recovery must retain the worktree");
+
+    revived.kill(session.id, false).await.expect("dismiss");
+    assert!(
+        !worktree.exists(),
+        "operator dismissal still cleans a pristine tree"
+    );
+}
+
 #[tokio::test]
 async fn failed_recovery_cleans_its_fifo() {
     let fx = DaemonFixture::new();
@@ -426,6 +582,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();
@@ -484,7 +778,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/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,87 @@ 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>,
+}
+
+/// 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>,
+}
+
 // ---------------------------------------------------------------------------
 // Client ↔ Daemon control messages (text frames on the attach WS)
 // ---------------------------------------------------------------------------
@@ -1144,6 +1225,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

docs/INSTALL.md

@@ -208,8 +208,10 @@ When the adapter is missing the spawn still creates the session, marked dead
 with the install command above as its activity detail, so the sidebar says
 what to run. The adapter is a child of the daemon: it ends with a `DELETE
 /api/sessions/:id` (`session/cancel`, then `SIGTERM`) and with the daemon
-itself. Terminal sessions keep surviving restarts through tmux; structured
-ones do not yet.
+itself. A daemon restart ends an in-flight structured turn, but preserves its
+worktree, session record, and event-log transcript. The replacement daemon
+starts a new adapter, resumes it with `session/load`, and replays the
+transcript. Terminal sessions keep surviving restarts through tmux.
 
 ## The browser UI