docs(maestro): fix disabled-unstable-variant claim in STRUCTURED-VIEW.md

A SessionUpdate variant gated behind an unstable ACP feature this build
doesn't enable is compiled out of the enum, not decoded into a Notice.
It fails deserialization the same as a genuinely unknown future tag.

CLAUDE.md

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

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};
@@ -112,6 +113,15 @@ impl SpawnConfig {
     }
 }
 
+/// A blocked ACP permission request delivered to the daemon.
+#[derive(Debug)]
+pub struct PermissionRequest {
+    /// The prompt turn that was active when ACP delivered this callback.
+    pub turn: u64,
+    pub request: RequestPermissionRequest,
+    pub response: oneshot::Sender<RequestPermissionOutcome>,
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum AcpError {
     /// The adapter is not on `PATH`. The message names the install command
@@ -153,6 +163,11 @@ fn is_executable(path: &Path) -> bool {
 
 enum Command {
     Cancel,
+    Prompt {
+        text: String,
+        turn: u64,
+        done: oneshot::Sender<Result<(), String>>,
+    },
     Close,
 }
 
@@ -161,6 +176,31 @@ enum ChildCommand {
     Reap(oneshot::Sender<bool>),
 }
 
+#[derive(Debug, Default)]
+struct PermissionTurnState {
+    active: Option<u64>,
+}
+
+impl PermissionTurnState {
+    fn begin(&mut self, turn: u64) {
+        self.active = Some(turn);
+    }
+
+    fn cancel_active(&mut self) {
+        self.active = None;
+    }
+
+    fn finish(&mut self, turn: u64) {
+        if self.active == Some(turn) {
+            self.active = None;
+        }
+    }
+
+    fn callback_turn(&self) -> Option<u64> {
+        self.active
+    }
+}
+
 /// Why the handshake did not produce a session id.
 enum ReadyError {
     /// The adapter answered with an error.
@@ -187,6 +227,7 @@ pub struct AcpSession {
     commands: mpsc::UnboundedSender<Command>,
     child_commands: mpsc::UnboundedSender<ChildCommand>,
     updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
+    permission_requests: Mutex<Option<mpsc::UnboundedReceiver<PermissionRequest>>>,
     exit: watch::Receiver<Option<ExitStatus>>,
 }
 
@@ -230,12 +271,36 @@ impl AcpSession {
         self.updates.lock().expect("updates mutex").take()
     }
 
+    pub fn take_permission_requests(&self) -> Option<mpsc::UnboundedReceiver<PermissionRequest>> {
+        self.permission_requests
+            .lock()
+            .expect("permission requests")
+            .take()
+    }
+
     /// Send `session/cancel`. Fire-and-forget: a closed connection means the
     /// child is already gone, which is the outcome cancel was after.
     pub fn cancel(&self) {
         let _ = self.commands.send(Command::Cancel);
     }
 
+    /// Send a user prompt and receive completion once the agent ends the turn.
+    pub fn prompt(
+        &self,
+        text: String,
+        turn: u64,
+    ) -> Result<oneshot::Receiver<Result<(), String>>, String> {
+        let (done_tx, done_rx) = oneshot::channel();
+        self.commands
+            .send(Command::Prompt {
+                text,
+                turn,
+                done: done_tx,
+            })
+            .map_err(|_| "ACP connection is closed".to_string())?;
+        Ok(done_rx)
+    }
+
     /// End the child: `session/cancel`, wait `cancel_grace`, `SIGTERM`, wait
     /// `term_grace`, `SIGKILL`. Returns once the child has been reaped.
     pub async fn terminate(&self, cancel_grace: Duration, term_grace: Duration) {
@@ -318,6 +383,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
     tokio::spawn(supervise_child(child, pid, child_commands_rx, exit_tx));
 
     let (updates_tx, updates_rx) = mpsc::unbounded_channel();
+    let (permission_tx, permission_rx) = mpsc::unbounded_channel();
     let (commands_tx, commands_rx) = mpsc::unbounded_channel();
     let (ready_tx, ready_rx) = oneshot::channel();
     let transport = ByteStreams::new(stdin.compat_write(), stdout.compat());
@@ -326,6 +392,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
         config.cwd.clone(),
         config.load_session.clone(),
         updates_tx,
+        permission_tx,
         commands_rx,
         ready_tx,
     ));
@@ -339,6 +406,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
                 commands: commands_tx,
                 child_commands: child_commands_tx,
                 updates: Mutex::new(Some(updates_rx)),
+                permission_requests: Mutex::new(Some(permission_rx)),
                 exit: exit_rx,
             });
         }
@@ -450,6 +518,7 @@ async fn run_connection<T>(
     cwd: PathBuf,
     load_session: Option<String>,
     updates_tx: mpsc::UnboundedSender<SessionNotification>,
+    permission_requests: mpsc::UnboundedSender<PermissionRequest>,
     mut commands_rx: mpsc::UnboundedReceiver<Command>,
     ready_tx: oneshot::Sender<Result<String, ReadyError>>,
 ) where
@@ -461,6 +530,10 @@ async fn run_connection<T>(
     // history the event log already holds (#135).
     let replaying = Arc::new(AtomicBool::new(false));
     let replaying_seen = replaying.clone();
+    // ACP callbacks have no turn id. The daemon never queues a later prompt
+    // until a cancelled prompt settles, so this active turn is unambiguous.
+    let permission_turn = Arc::new(Mutex::new(PermissionTurnState::default()));
+    let permission_turn_seen = permission_turn.clone();
     let result = Client
         .builder()
         .name("maestro")
@@ -478,15 +551,30 @@ async fn run_connection<T>(
         )
         .on_receive_request(
             async move |request: RequestPermissionRequest, responder, _cx| {
-                // Stub until #139: decline rather than grant something the
-                // operator never saw.
-                tracing::warn!(
-                    session = %request.session_id,
-                    "session/request_permission declined: not implemented yet (#139)"
-                );
-                responder.respond(RequestPermissionResponse::new(
-                    RequestPermissionOutcome::Cancelled,
-                ))
+                let turn = permission_turn_seen
+                    .lock()
+                    .expect("permission turn state")
+                    .callback_turn();
+                let Some(turn) = turn else {
+                    return responder.respond(RequestPermissionResponse::new(
+                        RequestPermissionOutcome::Cancelled,
+                    ));
+                };
+                let (response, received) = oneshot::channel();
+                if permission_requests
+                    .send(PermissionRequest {
+                        turn,
+                        request,
+                        response,
+                    })
+                    .is_err()
+                {
+                    return responder.respond(RequestPermissionResponse::new(
+                        RequestPermissionOutcome::Cancelled,
+                    ));
+                }
+                let outcome = received.await.unwrap_or(RequestPermissionOutcome::Cancelled);
+                responder.respond(RequestPermissionResponse::new(outcome))
             },
             agent_client_protocol::on_receive_request!(),
         )
@@ -558,10 +646,49 @@ async fn run_connection<T>(
                 tokio::select! {
                     cmd = commands_rx.recv() => match cmd {
                         Some(Command::Cancel) => {
+                            permission_turn
+                                .lock()
+                                .expect("permission turn state")
+                                .cancel_active();
                             let _ = cx.send_notification(CancelNotification::new(
                                 SessionId::new(session_id.as_str()),
                             ));
                         }
+                        Some(Command::Prompt { text, turn, done }) => {
+                            permission_turn
+                                .lock()
+                                .expect("permission turn state")
+                                .begin(turn);
+                            let request = PromptRequest::new(
+                                SessionId::new(session_id.as_str()),
+                                vec![ContentBlock::from(text)],
+                            );
+                            let done = Arc::new(Mutex::new(Some(done)));
+                            let callback_done = done.clone();
+                            let callback_turn = permission_turn.clone();
+                            if let Err(error) = cx.send_request(request).on_receiving_result(move |result| {
+                                let callback_done = callback_done.clone();
+                                let callback_turn = callback_turn.clone();
+                                async move {
+                                    callback_turn
+                                        .lock()
+                                        .expect("permission turn state")
+                                        .finish(turn);
+                                    if let Some(done) = callback_done.lock().expect("prompt sender").take() {
+                                        let _ = done.send(result.map(|_| ()).map_err(|e| e.to_string()));
+                                    }
+                                    Ok(())
+                                }
+                            }) {
+                                permission_turn
+                                    .lock()
+                                    .expect("permission turn state")
+                                    .finish(turn);
+                                if let Some(done) = done.lock().expect("prompt sender").take() {
+                                    let _ = done.send(Err(error.to_string()));
+                                }
+                            }
+                        }
                         Some(Command::Close) | None => break,
                     },
                     _ = cx.incoming_closed() => break,

crates/maestro-acp/tests/fake_acp_agent.py

@@ -24,11 +24,18 @@ Environment knobs, all optional:
                              so the SIGTERM path is exercised)
   FAKE_ACP_IGNORE_EOF=1      keep running after stdin closes, until signalled
   FAKE_ACP_EMIT_ON_CANCEL=1  emit a final notification before handling cancel
+  FAKE_ACP_CANCEL_SETTLES=1  complete the outstanding prompt on cancel
   FAKE_ACP_NO_LOAD=1      advertise `loadSession: false`
   FAKE_ACP_LOAD_REPLAY=n  on `session/load`, replay n `agent_message_chunk`
                             notifications ("replayed <i>") before answering,
                             the way a real agent streams history back
   FAKE_ACP_LOAD_FAIL=1    answer `session/load` with a JSON-RPC error
+  FAKE_ACP_DELAY_PROMPT_MS=n  wait before answering `session/prompt`
+  FAKE_ACP_PROMPT_FAIL=1  answer every `session/prompt` with a JSON-RPC error
+  FAKE_ACP_PROMPT_FAIL_ONCE=1  answer only the first `session/prompt` with a
+                             JSON-RPC error
+  FAKE_ACP_PERMISSION_AFTER_CANCEL=1  defer the first prompt's permission
+                             callback until a later prompt
 
 Signals: SIGUSR1 emits one `agent_message_chunk` notification ("tick <n>",
 counting from 1) and logs `emit tick <n>`, so a test can make the agent
@@ -65,6 +72,10 @@ signal.signal(signal.SIGTERM, on_sigterm)
 
 session_id = f"fake-{os.getpid()}"
 ticks = 0
+prompt_failures = 0
+pending_prompt_id = None
+pending_permission_id = None
+delayed_permission = False
 
 
 def chunk(text):
@@ -97,6 +108,12 @@ for raw in sys.stdin:
     method = msg.get("method")
     if method is None:
         # A response to something we sent; nothing to do.
+        if msg.get("id") == pending_permission_id:
+            log("permission response " + json.dumps(msg.get("result"), sort_keys=True))
+            send({"jsonrpc": "2.0", "id": pending_prompt_id,
+                  "result": {"stopReason": "end_turn"}})
+            pending_prompt_id = None
+            pending_permission_id = None
         continue
     log(f"recv {method}")
     rid = msg.get("id")
@@ -135,8 +152,56 @@ for raw in sys.stdin:
     elif method == "session/cancel":
         if os.environ.get("FAKE_ACP_EMIT_ON_CANCEL") == "1":
             send(chunk("final on cancel"))
+        if (os.environ.get("FAKE_ACP_CANCEL_SETTLES") == "1"
+                and pending_prompt_id is not None):
+            send({"jsonrpc": "2.0", "id": pending_prompt_id,
+                  "result": {"stopReason": "cancelled"}})
+            pending_prompt_id = None
+            pending_permission_id = None
         if os.environ.get("FAKE_ACP_EXIT_ON_CANCEL") == "1":
             sys.exit(0)
+    elif method == "session/prompt":
+        if os.environ.get("FAKE_ACP_PERMISSION_AFTER_CANCEL") == "1":
+            if pending_prompt_id is None:
+                pending_prompt_id = rid
+                delayed_permission = True
+                continue
+            if delayed_permission:
+                delayed_permission = False
+                pending_permission_id = "permission-1"
+                send({"jsonrpc": "2.0", "id": pending_permission_id,
+                      "method": "session/request_permission", "params": {
+                          "sessionId": session_id,
+                          "toolCall": {"toolCallId": "tool-1", "fields": {}},
+                          "options": [
+                              {"optionId": "once", "name": "Allow once", "kind": "allow_once"},
+                              {"optionId": "deny", "name": "Deny", "kind": "reject_once"}
+                          ]
+                      }})
+                continue
+        if os.environ.get("FAKE_ACP_REQUEST_PERMISSION") == "1":
+            pending_prompt_id = rid
+            pending_permission_id = "permission-1"
+            send({"jsonrpc": "2.0", "id": pending_permission_id,
+                  "method": "session/request_permission", "params": {
+                      "sessionId": session_id,
+                      "toolCall": {"toolCallId": "tool-1", "fields": {}},
+                      "options": [
+                          {"optionId": "once", "name": "Allow once", "kind": "allow_once"},
+                          {"optionId": "session", "name": "Allow session", "kind": "allow_always"},
+                          {"optionId": "deny", "name": "Deny", "kind": "reject_once"}
+                      ]
+                  }})
+            continue
+        time.sleep(int(os.environ.get("FAKE_ACP_DELAY_PROMPT_MS", "0")) / 1000)
+        fail_once = os.environ.get("FAKE_ACP_PROMPT_FAIL_ONCE") == "1"
+        if os.environ.get("FAKE_ACP_PROMPT_FAIL") == "1" or (fail_once and prompt_failures == 0):
+            prompt_failures += 1
+            send({"jsonrpc": "2.0", "id": rid,
+                  "error": {"code": -32000, "message": "fake prompt failure"}})
+        else:
+            send({"jsonrpc": "2.0", "id": rid,
+                  "result": {"stopReason": "end_turn"}})
     elif rid is not None:
         send({"jsonrpc": "2.0", "id": rid,
               "error": {"code": -32601, "message": f"method not found: {method}"}})

crates/maestro-daemon/src/activity.rs

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

crates/maestro-daemon/src/local_api.rs

@@ -21,12 +21,13 @@ use maestro_config::SpawnPreset;
 use maestro_protocol::{
     routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonInfo, GitRoot, NameBatch,
     RenameSessionRequest, ReportActivityRequest, SendInputRequest, Session, SetFocusRequest,
-    PROTOCOL_VERSION,
+    StructuredApprovalDecision, StructuredEvent, StructuredPromptRequest, PROTOCOL_VERSION,
 };
 use serde::Deserialize;
 use uuid::Uuid;
 
-use crate::session::SessionManager;
+use crate::session::{QueueAcpPrompt, SessionManager};
+use crate::sidecar::EventCursor;
 
 #[derive(Debug, Clone)]
 pub struct LocalApiState {
@@ -64,7 +65,15 @@ pub fn router(state: LocalApiState) -> Router {
         .route(routes::NAMES, get(names))
         .route(routes::INFO, get(daemon_info))
         .route(routes::EVENTS, any(events_ws))
-        .route(routes::patterns::ATTACH, any(attach_ws));
+        .route(routes::patterns::ATTACH, any(attach_ws))
+        .route(routes::patterns::STRUCTURED, any(structured_ws))
+        .route(routes::patterns::STRUCTURED_REPLAY, get(structured_replay))
+        .route(routes::patterns::STRUCTURED_PROMPT, post(structured_prompt))
+        .route(routes::patterns::STRUCTURED_CANCEL, post(structured_cancel))
+        .route(
+            routes::patterns::STRUCTURED_APPROVAL,
+            post(structured_approval),
+        );
     #[cfg(feature = "haiku-autoname")]
     {
         router = router.route(routes::patterns::SESSION_PROMPT, post(session_prompt));
@@ -298,11 +307,10 @@ async fn rename_session(
     }
 }
 
-/// `POST /api/sessions/:id/activity` (#105): an agent reports its own state.
-/// The report holds above the daemon's inference for a TTL, then inference
-/// resumes — an agent that dies mid-run cannot pin its session at `Error`
-/// forever. Unknown session is 404; an unknown state value is rejected by the
-/// JSON decoder (422) rather than coerced. Answers `204 No Content`.
+/// `POST /api/sessions/:id/activity` (#105): an agent reports a terminal
+/// session's state. Structured sessions acknowledge and ignore these reports
+/// because their activity comes from ACP. Unknown session is 404; an unknown
+/// state value is rejected by the JSON decoder (422). Answers `204 No Content`.
 async fn report_activity(
     State(s): State<LocalApiState>,
     Path(id): Path<Uuid>,
@@ -369,6 +377,126 @@ fn error_response(status: StatusCode, code: &str, message: &str) -> Response {
     (status, Json(ApiError::new(code, message))).into_response()
 }
 
+#[derive(Debug, Deserialize)]
+struct ReplayQuery {
+    since: Option<u64>,
+    before: Option<u64>,
+    limit: Option<usize>,
+}
+
+const MAX_STRUCTURED_REPLAY_LIMIT: usize = 500;
+
+async fn structured_replay(
+    State(s): State<LocalApiState>,
+    Path(id): Path<Uuid>,
+    Query(query): Query<ReplayQuery>,
+) -> Response {
+    if query.since.is_some() && query.before.is_some() {
+        return error_response(
+            StatusCode::BAD_REQUEST,
+            "invalid_cursor",
+            "use either since or before, not both",
+        );
+    }
+    let cursor = match query.before {
+        Some(before) => EventCursor::Before(Some(before)),
+        None => EventCursor::Since(query.since.unwrap_or(0)),
+    };
+    match s
+        .manager
+        .read_acp_projection_replay(
+            id,
+            cursor,
+            query
+                .limit
+                .unwrap_or(100)
+                .clamp(1, MAX_STRUCTURED_REPLAY_LIMIT),
+        )
+        .await
+    {
+        Ok(Some(events)) => Json(events).into_response(),
+        // A terminal session, or a structured session whose adapter is no
+        // longer live, cannot provide a coherent replay continuation.
+        Ok(None) => error_response(
+            StatusCode::CONFLICT,
+            "not_structured",
+            "session is not a live structured session",
+        ),
+        Err(e) => {
+            let info = e.to_error_info();
+            error_response(session_error_status(&info.code), &info.code, &info.message)
+        }
+    }
+}
+
+async fn structured_prompt(
+    State(s): State<LocalApiState>,
+    Path(id): Path<Uuid>,
+    Json(body): Json<StructuredPromptRequest>,
+) -> Response {
+    match s.manager.queue_acp_prompt(id, body.prompt).await {
+        Ok(QueueAcpPrompt::Queued) => StatusCode::NO_CONTENT.into_response(),
+        Ok(QueueAcpPrompt::InFlight) => error_response(
+            StatusCode::CONFLICT,
+            "turn_in_flight",
+            "the structured session already has an active turn",
+        ),
+        Ok(QueueAcpPrompt::Unavailable) => error_response(
+            StatusCode::CONFLICT,
+            "not_structured",
+            "session is not a live structured session",
+        ),
+        Ok(QueueAcpPrompt::Failed(message)) => {
+            error_response(StatusCode::BAD_GATEWAY, "acp_prompt_failed", &message)
+        }
+        Err(e) => {
+            let info = e.to_error_info();
+            error_response(session_error_status(&info.code), &info.code, &info.message)
+        }
+    }
+}
+
+async fn structured_cancel(State(s): State<LocalApiState>, Path(id): Path<Uuid>) -> Response {
+    match s.manager.cancel_acp_turn(id).await {
+        Ok(true) => StatusCode::NO_CONTENT.into_response(),
+        Ok(false) => error_response(
+            StatusCode::CONFLICT,
+            "not_structured",
+            "session is not a live structured session",
+        ),
+        Err(e) => {
+            let info = e.to_error_info();
+            error_response(session_error_status(&info.code), &info.code, &info.message)
+        }
+    }
+}
+
+async fn structured_approval(
+    State(s): State<LocalApiState>,
+    Path((id, nonce)): Path<(Uuid, String)>,
+    Json(decision): Json<StructuredApprovalDecision>,
+) -> Response {
+    let decision = match decision {
+        StructuredApprovalDecision::AllowOnce => crate::session::ApprovalDecision::AllowOnce,
+        StructuredApprovalDecision::AllowForSession => {
+            crate::session::ApprovalDecision::AllowForSession
+        }
+        StructuredApprovalDecision::Deny => crate::session::ApprovalDecision::Deny,
+    };
+    match s.manager.resolve_acp_approval(id, &nonce, decision).await {
+        Ok(Some(())) => StatusCode::NO_CONTENT.into_response(),
+        Ok(None) => error_response(
+            StatusCode::CONFLICT,
+            "not_structured",
+            "session is not a live structured session",
+        ),
+        Err(e) => {
+            let info = e.to_error_info();
+            error_response(session_error_status(&info.code), &info.code, &info.message)
+        }
+    }
+}
+
 /// Convenience constructor — pulls the bits `LocalApiState` needs from
 /// `DaemonConfig` so callers don't have to assemble it by hand. The project
 /// scanner starts with its cache cold; wire
@@ -419,6 +547,75 @@ async fn attach_ws(
     }))
 }
 
+#[allow(clippy::result_large_err)]
+async fn structured_ws(
+    Path(id): Path<Uuid>,
+    State(s): State<LocalApiState>,
+    ws: axum::extract::WebSocketUpgrade,
+) -> Result<Response, Response> {
+    let Some((snapshot, state, rx)) = s
+        .manager
+        .subscribe_acp_projection(id, EventCursor::Since(0))
+        .await
+        .map_err(|e| {
+            let info = e.to_error_info();
+            error_response(session_error_status(&info.code), &info.code, &info.message)
+        })?
+    else {
+        return Err(error_response(
+            StatusCode::CONFLICT,
+            "not_structured",
+            "session is not a live structured session",
+        ));
+    };
+    Ok(ws.on_upgrade(move |socket| async move {
+        let _ = pump_structured(socket, snapshot, state, rx).await;
+    }))
+}
+
+async fn pump_structured(
+    socket: axum::extract::ws::WebSocket,
+    snapshot: maestro_protocol::TranscriptSnapshot,
+    state: maestro_protocol::ReducedState,
+    mut rx: tokio::sync::broadcast::Receiver<crate::projection::ProjectionUpdate>,
+) -> Result<(), axum::Error> {
+    use axum::extract::ws::Message;
+    use futures_util::{SinkExt, StreamExt};
+
+    let (mut sink, mut stream) = socket.split();
+    for event in [
+        StructuredEvent::TranscriptSnapshot { snapshot },
+        StructuredEvent::ReducedState { state },
+    ] {
+        if sink
+            .send(Message::Text(
+                serde_json::to_string(&event).expect("structured event encode"),
+            ))
+            .await
+            .is_err()
+        {
+            return Ok(());
+        }
+    }
+    loop {
+        tokio::select! {
+            update = rx.recv() => match update {
+                Ok(update) => {
+                    let event = StructuredEvent::ProjectionUpdate { transcript: update.transcript, state: update.state };
+                    if sink.send(Message::Text(serde_json::to_string(&event).expect("structured event encode"))).await.is_err() { break; }
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => break,
+                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+            },
+            frame = stream.next() => match frame {
+                None | Some(Err(_)) | Some(Ok(Message::Close(_))) => break,
+                Some(Ok(_)) => {},
+            },
+        }
+    }
+    Ok(())
+}
+
 // ---------------------------------------------------------------------------
 // WS /api/events — live catalog deltas for browser/TUI subscribers (NAP-248)
 // ---------------------------------------------------------------------------

crates/maestro-daemon/src/projection.rs

@@ -4,7 +4,8 @@ use std::collections::{HashMap, HashSet};
 
 use maestro_acp::schema::{ContentChunk, SessionNotification, SessionUpdate, ToolCall};
 use maestro_protocol::{
-    ReducedState, TranscriptDelta, TranscriptRow, TranscriptRowKind, TranscriptSnapshot,
+    ActivityState, ReducedState, TranscriptDelta, TranscriptRow, TranscriptRowKind,
+    TranscriptSnapshot,
 };
 use serde_json::Value;
 
@@ -35,7 +36,9 @@ pub struct Projection {
     tools: HashMap<String, ToolCall>,
     pending_tool_updates: HashMap<String, Vec<maestro_acp::schema::ToolCallUpdateFields>>,
     state: ReducedState,
-    turn_active: bool,
+    active_turn: Option<u64>,
+    turn_failed: bool,
+    next_turn: u64,
     turn_tool_calls: HashSet<String>,
 }
 
@@ -104,27 +107,134 @@ impl Projection {
         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();
-            }
+    pub fn turn_active(&self) -> bool {
+        self.active_turn.is_some()
+    }
+
+    /// Sidebar activity derived from the live turn projection.
+    pub fn activity(&self) -> ActivityState {
+        crate::activity::decide_structured(
+            self.active_turn.is_some(),
+            self.state.pending_approval.is_some(),
+            self.turn_failed,
+        )
+    }
+
+    /// Start a prompt turn and return its monotonically increasing token.
+    pub fn begin_turn(&mut self) -> (u64, ProjectionUpdate) {
+        self.next_turn += 1;
+        let turn = self.next_turn;
+        self.active_turn = Some(turn);
+        self.turn_failed = false;
+        self.state.turn.responding = true;
+        self.state.turn.thinking = true;
+        self.turn_tool_calls.clear();
+        self.state.turn.active_tool_calls.clear();
+        (
+            turn,
+            ProjectionUpdate {
+                transcript: None,
+                state: self.state(),
+            },
+        )
+    }
+
+    /// End `turn` only when it remains the active prompt turn.
+    pub fn turn_lifecycle(
+        &mut self,
+        lifecycle: TurnLifecycle,
+        turn: u64,
+    ) -> Option<ProjectionUpdate> {
+        if !matches!(
+            lifecycle,
+            TurnLifecycle::Finish | TurnLifecycle::Cancel | TurnLifecycle::Failure
+        ) || self.active_turn != Some(turn)
+        {
+            return None;
         }
-        ProjectionUpdate {
+        self.active_turn = None;
+        self.turn_failed = lifecycle == TurnLifecycle::Failure;
+        self.state.turn.responding = false;
+        self.state.turn.thinking = false;
+        self.turn_tool_calls.clear();
+        self.state.turn.active_tool_calls.clear();
+        Some(ProjectionUpdate {
             transcript: None,
             state: self.state(),
+        })
+    }
+
+    pub fn active_turn(&self) -> Option<u64> {
+        self.active_turn
+    }
+
+    /// Add a daemon-originated permission row. The nonce belongs only in the
+    /// reduced state, which is the client control surface, never in ACP data.
+    pub fn request_approval(&mut self, nonce: String, request: Value) -> ProjectionUpdate {
+        let row = self.append(
+            format!("approval:{}", uuid::Uuid::new_v4()),
+            TranscriptRowKind::Approval,
+            request.clone(),
+        );
+        self.state.pending_approval = Some(serde_json::json!({
+            "nonce": nonce,
+            "request": request,
+        }));
+        ProjectionUpdate {
+            transcript: Some(row),
+            state: self.state(),
+        }
+    }
+
+    pub fn restore_approval(&mut self, row: TranscriptRow) -> ProjectionUpdate {
+        let transcript = if !self.row_index.contains_key(&row.id) {
+            self.row_index.insert(row.id.clone(), self.rows.len());
+            self.rows.push(row.clone());
+            Some(TranscriptDelta::Append { row })
+        } else {
+            None
+        };
+        ProjectionUpdate {
+            transcript,
+            state: self.state(),
+        }
+    }
+
+    /// Clear an approval only when it is still the one the client answered.
+    pub fn resolve_approval(&mut self, nonce: &str) -> Option<ProjectionUpdate> {
+        let pending = self.state.pending_approval.as_ref()?;
+        if pending.get("nonce").and_then(Value::as_str) != Some(nonce) {
+            return None;
+        }
+        self.state.pending_approval = None;
+        Some(ProjectionUpdate {
+            transcript: None,
+            state: self.state(),
+        })
+    }
+
+    /// Terminal turn paths must not leave a recovered snapshot permanently
+    /// blocked on an ACP request whose transport no longer exists.
+    pub fn clear_approval(&mut self) -> Option<ProjectionUpdate> {
+        self.state.pending_approval.take()?;
+        Some(ProjectionUpdate {
+            transcript: None,
+            state: self.state(),
+        })
+    }
+
+    /// Undo an `append`ed row that a caller decides never happened — used
+    /// when persisting the row's own event fails (#184), so a snapshot taken
+    /// after the failure cannot show a row durable replay would never
+    /// reproduce. Only ever called for a row that was just appended and
+    /// never patched, so removing it by id fully reverses `append`.
+    pub fn rollback_row(&mut self, id: &str) {
+        let Some(index) = self.row_index.remove(id) else {
+            return;
+        };
+        self.rows.remove(index);
+        for (shifted_index, row) in self.rows.iter().enumerate().skip(index) {
+            self.row_index.insert(row.id.clone(), shifted_index);
         }
     }
 
@@ -230,28 +340,28 @@ impl Projection {
     }
 
     fn agent_text(&mut self) {
-        if self.turn_active {
+        if self.turn_active() {
             self.state.turn.responding = true;
             self.state.turn.thinking = false;
         }
     }
 
     fn agent_thought(&mut self) {
-        if self.turn_active {
+        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 {
+        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) {
+        if !self.turn_active() || !self.turn_tool_calls.contains(tool_id) {
             return;
         }
         let active = self.tools.get(tool_id).is_some_and(|tool| {
@@ -328,6 +438,16 @@ mod tests {
         .unwrap()
     }
 
+    #[test]
+    fn a_session_update_tag_the_acp_dependency_does_not_know_fails_to_deserialize() {
+        // `SessionUpdate` has no unknown-variant fallback, so this never reaches `Projection::apply`.
+        let result: Result<SessionNotification, _> = serde_json::from_value(serde_json::json!({
+            "sessionId": "acp-1",
+            "update": {"sessionUpdate": "some_future_update_variant"},
+        }));
+        assert!(result.is_err());
+    }
+
     #[test]
     fn folds_content_tool_diff_plan_and_notice_rows() {
         let mut projection = Projection::default();
@@ -538,7 +658,7 @@ mod tests {
         );
         assert_eq!(projection.state().turn, Default::default());
 
-        projection.turn_lifecycle(TurnLifecycle::Begin);
+        let (_, _) = projection.begin_turn();
         assert!(projection.state().turn.responding);
         assert!(projection.state().turn.thinking);
         projection.apply(
@@ -558,8 +678,8 @@ mod tests {
             TurnLifecycle::Cancel,
             TurnLifecycle::Failure,
         ] {
-            projection.turn_lifecycle(TurnLifecycle::Begin);
-            projection.turn_lifecycle(terminal);
+            let (turn, _) = projection.begin_turn();
+            projection.turn_lifecycle(terminal, turn);
             assert!(!projection.state().turn.responding);
             assert!(!projection.state().turn.thinking);
         }
@@ -571,7 +691,7 @@ mod tests {
         next_id: &str,
     ) {
         let mut projection = Projection::default();
-        projection.turn_lifecycle(TurnLifecycle::Begin);
+        let (turn, _) = projection.begin_turn();
         projection.apply(
             1,
             &event(serde_json::json!({
@@ -581,11 +701,11 @@ mod tests {
         );
         assert_eq!(projection.state().turn.active_tool_calls, [unfinished_id]);
 
-        projection.turn_lifecycle(terminal);
+        projection.turn_lifecycle(terminal, turn);
         assert!(projection.state().turn.active_tool_calls.is_empty());
         assert_eq!(projection.snapshot(1).rows.len(), 1);
 
-        projection.turn_lifecycle(TurnLifecycle::Begin);
+        projection.begin_turn();
         projection.apply(
             2,
             &event(serde_json::json!({
@@ -605,4 +725,136 @@ mod tests {
     fn failed_turn_does_not_reactivate_unfinished_tools() {
         terminal_turn_drops_unfinished_tools(TurnLifecycle::Failure, "failed-1", "next-1");
     }
+
+    #[test]
+    fn approval_history_survives_pending_cleanup() {
+        let mut projection = Projection::default();
+        let update =
+            projection.request_approval("secret".into(), serde_json::json!({"tool":"bash"}));
+        let TranscriptDelta::Append { row } = update.transcript.unwrap() else {
+            panic!("approval row")
+        };
+        assert_eq!(row.kind, TranscriptRowKind::Approval);
+        assert!(row.data.get("nonce").is_none());
+        assert!(!row.id.contains("secret"));
+        let persisted =
+            serde_json::to_string(&crate::sidecar::PersistedEvent::Approval { row: row.clone() })
+                .unwrap();
+        assert!(!persisted.contains("secret"));
+        assert_eq!(
+            projection.state().pending_approval.unwrap()["nonce"],
+            "secret"
+        );
+        projection.clear_approval();
+        assert!(projection.state().pending_approval.is_none());
+        assert_eq!(projection.snapshot(0).rows, vec![row]);
+    }
+
+    #[test]
+    fn rollback_row_removes_only_that_row_and_keeps_others_addressable() {
+        let mut projection = Projection::default();
+        projection.apply(
+            1,
+            &event(serde_json::json!({
+                "sessionUpdate": "user_message_chunk",
+                "messageId": "m1",
+                "content": {"type": "text", "text": "hi"},
+            })),
+        );
+        let TranscriptDelta::Append { row } = projection
+            .request_approval("nonce".into(), serde_json::json!({"tool": "bash"}))
+            .transcript
+            .unwrap()
+        else {
+            panic!("approval row")
+        };
+        // A row after the one being rolled back, so removing the approval
+        // (index 1) shifts this one from index 2 down to index 1 — proving
+        // `row_index` gets reindexed, not just `rows`.
+        projection.apply(
+            2,
+            &event(serde_json::json!({
+                "sessionUpdate": "user_message_chunk",
+                "messageId": "m2",
+                "content": {"type": "text", "text": "hey"},
+            })),
+        );
+        assert_eq!(projection.snapshot(0).rows.len(), 3);
+
+        projection.rollback_row(&row.id);
+
+        assert_eq!(projection.snapshot(0).rows.len(), 2);
+        assert_eq!(projection.snapshot(0).rows[0].kind, TranscriptRowKind::User);
+        assert_eq!(projection.snapshot(0).rows[1].kind, TranscriptRowKind::User);
+
+        // The shifted row must still be independently patchable at its new
+        // index.
+        let update = projection.apply(
+            3,
+            &event(serde_json::json!({
+                "sessionUpdate": "user_message_chunk",
+                "messageId": "m2",
+                "content": {"type": "text", "text": " there"},
+            })),
+        );
+        let TranscriptDelta::Patch { row } = update.transcript.unwrap() else {
+            panic!("patched row")
+        };
+        assert_eq!(row.data.pointer("/content/text").unwrap(), "hey there");
+        assert_eq!(projection.snapshot(0).rows.len(), 2);
+    }
+
+    #[test]
+    fn rollback_row_is_a_no_op_for_an_unknown_id() {
+        let mut projection = Projection::default();
+        projection.apply(
+            1,
+            &event(serde_json::json!({
+                "sessionUpdate": "user_message_chunk",
+                "content": {"type": "text", "text": "hi"},
+            })),
+        );
+        projection.rollback_row("approval:does-not-exist");
+        assert_eq!(projection.snapshot(0).rows.len(), 1);
+    }
+
+    #[test]
+    fn recovered_approval_is_historical_not_pending() {
+        let mut live = Projection::default();
+        let TranscriptDelta::Append { row } = live
+            .request_approval("daemon-nonce".into(), serde_json::json!({"tool":"bash"}))
+            .transcript
+            .unwrap()
+        else {
+            panic!("approval row")
+        };
+        let mut recovered = Projection::default();
+        let update = recovered.restore_approval(row.clone());
+        assert_eq!(
+            update.transcript,
+            Some(TranscriptDelta::Append { row: row.clone() })
+        );
+        assert!(update.state.pending_approval.is_none());
+        assert_eq!(recovered.snapshot(0).rows, vec![row]);
+        let replay = serde_json::to_string(&recovered.snapshot(1).rows).unwrap();
+        assert!(!replay.contains("daemon-nonce"));
+    }
+
+    #[test]
+    fn stale_terminal_turn_cannot_clear_newer_approval() {
+        let mut projection = Projection::default();
+        let (first, _) = projection.begin_turn();
+        projection.turn_lifecycle(TurnLifecycle::Cancel, first);
+        let (second, _) = projection.begin_turn();
+        projection.request_approval("second-nonce".into(), serde_json::json!({"tool":"bash"}));
+
+        assert!(projection
+            .turn_lifecycle(TurnLifecycle::Finish, first)
+            .is_none());
+        assert_eq!(projection.active_turn(), Some(second));
+        assert_eq!(
+            projection.state().pending_approval.unwrap()["nonce"],
+            "second-nonce"
+        );
+    }
 }

crates/maestro-daemon/src/sidecar.rs

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

crates/maestro-daemon/tests/acp.rs

@@ -83,7 +83,10 @@ impl Fake {
 }
 
 fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
-    match &event.notification.update {
+    let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else {
+        panic!("unexpected daemon event")
+    };
+    match &notification.update {
         SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
             ContentBlock::Text(text) => &text.text,
             other => panic!("unexpected content {other:?}"),
@@ -133,6 +136,7 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
 
     assert_eq!(session.view, SessionView::Structured);
     assert_eq!(session.status, SessionStatus::Running);
+    assert_eq!(session.activity, ActivityState::Idle);
     let acp_id = session.acp_session_id.clone().expect("acp session id");
     assert!(acp_id.starts_with("fake-"), "{acp_id}");
 
@@ -201,6 +205,77 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
         .exists());
 }
 
+#[tokio::test]
+async fn structured_activity_tracks_turn_completion_and_failure() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_DELAY_PROMPT_MS".into(), "100".into());
+    env.insert("FAKE_ACP_PROMPT_FAIL_ONCE".into(), "1".into());
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .unwrap();
+
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "fail".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+    assert_eq!(
+        fx.manager
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|candidate| candidate.id == session.id)
+            .unwrap()
+            .activity,
+        ActivityState::Active
+    );
+    wait_for_async(Duration::from_secs(2), || async {
+        fx.manager
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|candidate| candidate.id == session.id)
+            .is_some_and(|candidate| candidate.activity == ActivityState::Error)
+    })
+    .await;
+
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "succeed".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+    assert_eq!(
+        fx.manager
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|candidate| candidate.id == session.id)
+            .unwrap()
+            .activity,
+        ActivityState::Active
+    );
+    wait_for_async(Duration::from_secs(2), || async {
+        fx.manager
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|candidate| candidate.id == session.id)
+            .is_some_and(|candidate| candidate.activity == ActivityState::Idle)
+    })
+    .await;
+
+    fx.manager.kill(session.id, false).await.unwrap();
+}
+
 #[tokio::test]
 async fn projection_snapshot_and_live_stream_share_one_fold() {
     let fx = DaemonFixture::new();
@@ -232,11 +307,14 @@ async fn projection_snapshot_and_live_stream_share_one_fold() {
     assert_eq!(snapshot.rows[0].kind, TranscriptRowKind::AgentText);
     assert_eq!(state.turn, Default::default());
 
-    assert!(fx
+    let maestro_daemon::session::BeginAcpTurn::Started(_, turn) = fx
         .manager
-        .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Begin)
+        .begin_acp_turn(session.id)
         .await
-        .expect("begin lifecycle"));
+        .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")
@@ -264,7 +342,7 @@ async fn projection_snapshot_and_live_stream_share_one_fold() {
 
     assert!(fx
         .manager
-        .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish)
+        .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish, turn)
         .await
         .expect("finish lifecycle"));
     let finish = tokio::time::timeout(Duration::from_secs(5), live.recv())
@@ -277,6 +355,148 @@ async fn projection_snapshot_and_live_stream_share_one_fold() {
     fx.manager.kill(session.id, false).await.expect("kill");
 }
 
+#[tokio::test]
+async fn stale_turn_completion_cannot_end_the_next_turn() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), fake.env()))
+        .await
+        .unwrap();
+
+    let maestro_daemon::session::BeginAcpTurn::Started(_, first) =
+        fx.manager.begin_acp_turn(session.id).await.unwrap()
+    else {
+        panic!("first turn should start");
+    };
+    assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
+    let maestro_daemon::session::BeginAcpTurn::Started(_, second) =
+        fx.manager.begin_acp_turn(session.id).await.unwrap()
+    else {
+        panic!("second turn should start");
+    };
+    assert!(second > first);
+    assert!(!fx
+        .manager
+        .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish, first)
+        .await
+        .unwrap());
+    assert!(matches!(
+        fx.manager.begin_acp_turn(session.id).await.unwrap(),
+        maestro_daemon::session::BeginAcpTurn::InFlight
+    ));
+    assert!(fx
+        .manager
+        .apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish, second)
+        .await
+        .unwrap());
+
+    fx.manager.kill(session.id, false).await.unwrap();
+}
+
+#[tokio::test]
+async fn cancelled_prompt_that_never_settles_fails_the_session() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_PERMISSION_AFTER_CANCEL".into(), "1".into());
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .unwrap();
+
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "first".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+    assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
+    let second = fx
+        .manager
+        .queue_acp_prompt(session.id, "second".into())
+        .await
+        .unwrap();
+    assert!(
+        matches!(second, maestro_daemon::session::QueueAcpPrompt::Unavailable),
+        "{second:?}"
+    );
+    assert_eq!(
+        fx.manager.status(session.id).await,
+        Some(SessionStatus::Dead)
+    );
+    assert!(!fake.log().contains("permission response"));
+}
+
+#[tokio::test]
+async fn cancellation_settlement_allows_the_next_prompt_permission() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_REQUEST_PERMISSION".into(), "1".into());
+    env.insert("FAKE_ACP_CANCEL_SETTLES".into(), "1".into());
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .unwrap();
+
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "first".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+    assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "second".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+
+    let nonce = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let (_, state, _) = fx
+                .manager
+                .subscribe_acp_projection(session.id, EventCursor::Since(0))
+                .await
+                .unwrap()
+                .unwrap();
+            if let Some(pending) = state.pending_approval {
+                break pending["nonce"].as_str().unwrap().to_owned();
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    assert!(fx
+        .manager
+        .resolve_acp_approval(
+            session.id,
+            &nonce,
+            maestro_daemon::session::ApprovalDecision::AllowOnce,
+        )
+        .await
+        .unwrap()
+        .is_some());
+    wait_for_async(Duration::from_secs(2), || {
+        let log = fake.log();
+        async move { log.contains("permission response") }
+    })
+    .await;
+    fx.manager.kill(session.id, false).await.unwrap();
+}
+
 #[tokio::test]
 async fn recovery_loads_the_acp_session_without_duplicating_history() {
     let fx = DaemonFixture::new();
@@ -328,6 +548,16 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
         revived.status(session.id).await,
         Some(SessionStatus::Running)
     );
+    assert_eq!(
+        revived
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|candidate| candidate.id == session.id)
+            .unwrap()
+            .activity,
+        ActivityState::Idle
+    );
     let log = fake.log();
     assert!(
         log.contains(&format!("recv session/load sessionId={acp_id}")),
@@ -777,6 +1007,16 @@ async fn adapter_exit_marks_the_session_dead_and_keeps_it_listed() {
         fx.manager.status(session.id).await == Some(SessionStatus::Dead)
     })
     .await;
+    assert_eq!(
+        fx.manager
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|candidate| candidate.id == session.id)
+            .unwrap()
+            .activity,
+        ActivityState::Error
+    );
     assert!(fx.manager.has(session.id).await);
     assert!(
         !fx.sidecar_dir()

crates/maestro-daemon/tests/local_api.rs

@@ -4,12 +4,17 @@ mod common;
 
 use std::collections::HashMap;
 use std::net::{Ipv4Addr, SocketAddr};
+use std::path::Path;
 use std::time::Duration;
 
 use common::DaemonFixture;
 use futures_util::{SinkExt, StreamExt};
 use maestro_daemon::local_api;
-use maestro_protocol::{ApiError, CreateSessionRequest, Session, WorktreeSpec};
+use maestro_protocol::{
+    ActivityState, ApiError, CreateSessionRequest, Session, SessionStatus, SessionView,
+    WorktreeSpec,
+};
+use tempfile::TempDir;
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
 use tokio_tungstenite::tungstenite::Message as WsMessage;
@@ -86,6 +91,589 @@ fn ws_url(addr: SocketAddr, path: &str) -> String {
     format!("ws://{addr}{path}")
 }
 
+struct FakeAcp {
+    dir: TempDir,
+}
+
+impl FakeAcp {
+    fn new() -> Self {
+        use std::os::unix::fs::PermissionsExt;
+
+        let dir = TempDir::new().unwrap();
+        let bin = dir.path().join("bin");
+        std::fs::create_dir_all(&bin).unwrap();
+        let adapter = bin.join("claude-code-acp");
+        std::fs::copy(
+            Path::new(env!("CARGO_MANIFEST_DIR")).join("../maestro-acp/tests/fake_acp_agent.py"),
+            &adapter,
+        )
+        .unwrap();
+        std::fs::set_permissions(&adapter, std::fs::Permissions::from_mode(0o755)).unwrap();
+        Self { dir }
+    }
+
+    fn env(&self) -> HashMap<String, String> {
+        HashMap::from([
+            (
+                "PATH".into(),
+                format!(
+                    "{}:{}",
+                    self.dir.path().join("bin").display(),
+                    std::env::var("PATH").unwrap()
+                ),
+            ),
+            (
+                "FAKE_ACP_LOG".into(),
+                self.dir.path().join("agent.log").display().to_string(),
+            ),
+        ])
+    }
+
+    fn pid(&self) -> u32 {
+        std::fs::read_to_string(self.dir.path().join("agent.log"))
+            .unwrap_or_default()
+            .lines()
+            .filter_map(|line| {
+                line.strip_prefix("start pid=")?
+                    .split(' ')
+                    .next()?
+                    .parse()
+                    .ok()
+            })
+            .next_back()
+            .expect("fake ACP started")
+    }
+
+    fn log(&self) -> String {
+        std::fs::read_to_string(self.dir.path().join("agent.log")).unwrap_or_default()
+    }
+}
+
+async fn structured_session(h: &Harness, fake: &FakeAcp) -> Uuid {
+    structured_session_with_env(h, fake.env()).await
+}
+
+async fn structured_session_with_env(h: &Harness, env: HashMap<String, String>) -> Uuid {
+    h.manager
+        .create(CreateSessionRequest {
+            view: SessionView::Structured,
+            cwd: "/tmp".into(),
+            command: "claude".into(),
+            env,
+            metadata: HashMap::new(),
+            worktree: None,
+        })
+        .await
+        .unwrap()
+        .id
+}
+
+async fn wait_for_session_activity(h: &Harness, id: Uuid, expected: ActivityState) {
+    tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let sessions: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
+                .await
+                .unwrap()
+                .json()
+                .await
+                .unwrap();
+            if sessions
+                .iter()
+                .find(|session| session.id == id)
+                .is_some_and(|session| session.activity == expected)
+            {
+                return;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("session {id} did not reach {expected:?}"));
+}
+
+#[tokio::test]
+async fn structured_ws_sends_snapshot_then_live_projection() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+    let (mut ws, _) = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
+        .await
+        .unwrap();
+
+    let first = ws.next().await.unwrap().unwrap().into_text().unwrap();
+    let second = ws.next().await.unwrap().unwrap().into_text().unwrap();
+    assert_eq!(
+        serde_json::from_str::<serde_json::Value>(&first).unwrap()["type"],
+        "transcript_snapshot"
+    );
+    assert_eq!(
+        serde_json::from_str::<serde_json::Value>(&second).unwrap()["type"],
+        "reduced_state"
+    );
+
+    let response = reqwest::Client::new()
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "hello"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);
+    let live = tokio::time::timeout(Duration::from_secs(1), ws.next())
+        .await
+        .expect("projection update arrives on the subscribed stream")
+        .unwrap()
+        .unwrap()
+        .into_text()
+        .unwrap();
+    assert_eq!(
+        serde_json::from_str::<serde_json::Value>(&live).unwrap()["type"],
+        "projection_update"
+    );
+}
+
+#[tokio::test]
+async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_DELAY_PROMPT_MS".into(), "200".into());
+    let id = structured_session_with_env(&h, env).await;
+    let client = reqwest::Client::new();
+    let first = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "first"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(first.status(), reqwest::StatusCode::NO_CONTENT);
+    let second = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "second"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(second.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = second.json().await.unwrap();
+    assert_eq!(error.error.code, "turn_in_flight");
+
+    let cancel = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/cancel"),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(cancel.status(), reqwest::StatusCode::NO_CONTENT);
+    let next = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "next"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(next.status(), reqwest::StatusCode::NO_CONTENT);
+    tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            if fake.log().matches("recv session/prompt").count() == 2 {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("both prompt commands reach ACP");
+    let log = fake.log();
+    let first_prompt = log.find("recv session/prompt").unwrap();
+    let cancel = log.find("recv session/cancel").unwrap();
+    let second_prompt = log.rfind("recv session/prompt").unwrap();
+    assert!(first_prompt < cancel && cancel < second_prompt, "{log}");
+}
+
+#[tokio::test]
+async fn structured_approval_is_secret_single_use_and_race_safe() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_REQUEST_PERMISSION".into(), "1".into());
+    let id = structured_session_with_env(&h, env).await;
+    let client = reqwest::Client::new();
+    let (mut ws, _) = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
+        .await
+        .unwrap();
+    ws.next().await.unwrap().unwrap();
+    ws.next().await.unwrap().unwrap();
+    assert_eq!(
+        client
+            .post(url(
+                h.addr,
+                &format!("/api/sessions/{id}/structured/prompt")
+            ))
+            .json(&serde_json::json!({"prompt":"needs permission"}))
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        reqwest::StatusCode::NO_CONTENT
+    );
+
+    let nonce = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let event: serde_json::Value =
+                serde_json::from_str(&ws.next().await.unwrap().unwrap().into_text().unwrap())
+                    .unwrap();
+            if let Some(nonce) = event["state"]["pending_approval"]["nonce"].as_str() {
+                assert_eq!(event["transcript"]["row"]["kind"], "approval");
+                assert!(event["transcript"]["row"]["data"].get("nonce").is_none());
+                break nonce.to_owned();
+            }
+        }
+    })
+    .await
+    .unwrap();
+    wait_for_session_activity(&h, id, ActivityState::AwaitingInput).await;
+    assert!(
+        !fake.log().contains(&nonce),
+        "daemon nonce leaked to ACP agent"
+    );
+
+    let path = format!("/api/sessions/{id}/structured/approvals/{nonce}");
+    let (a, b) = tokio::join!(
+        client
+            .post(url(h.addr, &path))
+            .json(&serde_json::json!("allow_once"))
+            .send(),
+        client
+            .post(url(h.addr, &path))
+            .json(&serde_json::json!("deny"))
+            .send(),
+    );
+    assert_eq!(a.unwrap().status(), reqwest::StatusCode::NO_CONTENT);
+    assert_eq!(b.unwrap().status(), reqwest::StatusCode::NO_CONTENT);
+    assert_eq!(
+        client
+            .post(url(h.addr, &path))
+            .json(&serde_json::json!("deny"))
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        reqwest::StatusCode::NO_CONTENT
+    );
+    tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            if fake.log().contains("permission response") {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    let log = fake.log();
+    assert!(
+        log.contains("\"optionId\": \"once\"") || log.contains("\"optionId\": \"deny\""),
+        "{log}"
+    );
+    tokio::time::timeout(Duration::from_secs(1), async {
+        loop {
+            let event: serde_json::Value =
+                serde_json::from_str(&ws.next().await.unwrap().unwrap().into_text().unwrap())
+                    .unwrap();
+            if event["state"]["pending_approval"].is_null() {
+                break;
+            }
+        }
+    })
+    .await
+    .expect("resolved approval clears the reduced state");
+    wait_for_session_activity(&h, id, ActivityState::Idle).await;
+    assert_eq!(
+        client
+            .post(url(
+                h.addr,
+                &format!(
+                    "/api/sessions/{}/structured/approvals/{nonce}",
+                    Uuid::new_v4()
+                )
+            ))
+            .json(&serde_json::json!("deny"))
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        reqwest::StatusCode::NOT_FOUND
+    );
+}
+
+#[tokio::test]
+async fn structured_prompt_failure_clears_the_turn() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_PROMPT_FAIL_ONCE".into(), "1".into());
+    let id = structured_session_with_env(&h, env).await;
+    let client = reqwest::Client::new();
+    let path = format!("/api/sessions/{id}/structured/prompt");
+    let (mut ws, _) = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
+        .await
+        .unwrap();
+    ws.next().await.unwrap().unwrap();
+    ws.next().await.unwrap().unwrap();
+    let first = client
+        .post(url(h.addr, &path))
+        .json(&serde_json::json!({"prompt":"fails"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(first.status(), reqwest::StatusCode::NO_CONTENT);
+    let mut saw_active_turn = false;
+    tokio::time::timeout(Duration::from_secs(1), async {
+        while let Some(Ok(message)) = ws.next().await {
+            let event: serde_json::Value =
+                serde_json::from_str(&message.into_text().unwrap()).unwrap();
+            let turn = &event["state"]["turn"];
+            if turn["responding"] == true {
+                saw_active_turn = true;
+            }
+            if saw_active_turn && turn["responding"] == false && turn["thinking"] == false {
+                return;
+            }
+        }
+        panic!("structured websocket closed before the failed turn cleared");
+    })
+    .await
+    .expect("failed turn clears promptly");
+    wait_for_session_activity(&h, id, ActivityState::Error).await;
+    let second = client
+        .post(url(h.addr, &path))
+        .json(&serde_json::json!({"prompt":"available"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(second.status(), reqwest::StatusCode::NO_CONTENT);
+    assert!(second.bytes().await.unwrap().is_empty());
+}
+
+#[tokio::test]
+async fn structured_prompt_rejects_unknown_and_unavailable_sessions() {
+    let h = spawn("naps-laptop").await;
+    let client = reqwest::Client::new();
+    let prompt = |id| {
+        client
+            .post(url(
+                h.addr,
+                &format!("/api/sessions/{id}/structured/prompt"),
+            ))
+            .json(&serde_json::json!({"prompt": "hello"}))
+    };
+
+    let unknown = prompt(Uuid::new_v4()).send().await.unwrap();
+    assert_eq!(unknown.status(), reqwest::StatusCode::NOT_FOUND);
+    let error: ApiError = unknown.json().await.unwrap();
+    assert_eq!(error.error.code, "not_found");
+
+    let terminal = h
+        .manager
+        .create(CreateSessionRequest {
+            view: SessionView::Terminal,
+            cwd: "/tmp".into(),
+            command: "sh".into(),
+            env: HashMap::new(),
+            metadata: HashMap::new(),
+            worktree: None,
+        })
+        .await
+        .unwrap();
+    let non_structured = prompt(terminal.id).send().await.unwrap();
+    assert_eq!(non_structured.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = non_structured.json().await.unwrap();
+    assert_eq!(error.error.code, "not_structured");
+
+    let fake = FakeAcp::new();
+    let dead = structured_session(&h, &fake).await;
+    h.manager.shutdown().await;
+    assert_eq!(h.manager.status(dead).await, Some(SessionStatus::Dead));
+    let unavailable = prompt(dead).send().await.unwrap();
+    assert_eq!(unavailable.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = unavailable.json().await.unwrap();
+    assert_eq!(error.error.code, "not_structured");
+}
+
+#[tokio::test]
+async fn structured_replay_honors_cursors_and_limit() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+    let client = reqwest::Client::new();
+
+    let events: Vec<serde_json::Value> = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let events: Vec<serde_json::Value> = client
+                .get(url(
+                    h.addr,
+                    &format!("/api/sessions/{id}/structured/replay?since=0&limit=1"),
+                ))
+                .send()
+                .await
+                .unwrap()
+                .json()
+                .await
+                .unwrap();
+            if !events.is_empty() {
+                return events;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(events.len(), 1);
+    assert_eq!(events[0]["cursor"], 1);
+    assert!(events[0].get("notification").is_none());
+
+    let invalid = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/replay?since=0&before=2"),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(invalid.status(), reqwest::StatusCode::BAD_REQUEST);
+}
+
+#[tokio::test]
+async fn structured_replay_pages_projected_events_in_cursor_order() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+    let client = reqwest::Client::new();
+    unsafe { libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1) };
+    let events: Vec<serde_json::Value> = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let events: Vec<serde_json::Value> = client
+                .get(url(
+                    h.addr,
+                    &format!("/api/sessions/{id}/structured/replay?before=99&limit=2"),
+                ))
+                .send()
+                .await
+                .unwrap()
+                .json()
+                .await
+                .unwrap();
+            if events.len() == 2 {
+                return events;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(
+        events
+            .iter()
+            .map(|event| event["cursor"].as_u64().unwrap())
+            .collect::<Vec<_>>(),
+        vec![1, 2]
+    );
+    let boundary: Vec<serde_json::Value> = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/replay?before=2&limit=10"),
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert_eq!(boundary.len(), 1);
+    assert_eq!(boundary[0]["cursor"], 1);
+}
+
+#[tokio::test]
+async fn structured_replay_rejects_unknown_terminal_and_dead_sessions() {
+    let h = spawn("naps-laptop").await;
+    let client = reqwest::Client::new();
+    let unknown = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{}/structured/replay", Uuid::new_v4()),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(unknown.status(), reqwest::StatusCode::NOT_FOUND);
+    let terminal = h
+        .manager
+        .create(CreateSessionRequest {
+            view: SessionView::Terminal,
+            cwd: "/tmp".into(),
+            command: "sh".into(),
+            env: HashMap::new(),
+            metadata: HashMap::new(),
+            worktree: None,
+        })
+        .await
+        .unwrap();
+    let response = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{}/structured/replay", terminal.id),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
+
+    let fake = FakeAcp::new();
+    let dead = structured_session(&h, &fake).await;
+    h.manager.shutdown().await;
+    assert_eq!(h.manager.status(dead).await, Some(SessionStatus::Dead));
+    let response = client
+        .get(url(
+            h.addr,
+            &format!("/api/sessions/{dead}/structured/replay"),
+        ))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), reqwest::StatusCode::CONFLICT);
+    let error: ApiError = response.json().await.unwrap();
+    assert_eq!(error.error.code, "not_structured");
+}
+
+#[tokio::test]
+async fn structured_ws_rejects_a_dead_session() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let id = structured_session(&h, &fake).await;
+    h.manager.shutdown().await;
+    assert_eq!(h.manager.status(id).await, Some(SessionStatus::Dead));
+
+    let err = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
+        .await
+        .expect_err("dead structured session must not upgrade");
+    let message = err.to_string();
+    assert!(
+        message.contains("409") || message.to_lowercase().contains("conflict"),
+        "{message}"
+    );
+}
+
 #[tokio::test]
 async fn info_returns_daemon_identity() {
     let h = spawn("naps-laptop").await;
@@ -789,6 +1377,73 @@ async fn events_ws_sends_a_snapshot_then_streams_deltas() {
     }
 }
 
+#[tokio::test]
+async fn events_ws_streams_structured_activity_without_hook_overrides() {
+    let h = spawn("konishi").await;
+    let fake = FakeAcp::new();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_DELAY_PROMPT_MS".into(), "500".into());
+    let (mut ws, _) = connect_async(ws_url(h.addr, "/api/events"))
+        .await
+        .expect("events ws");
+    assert!(matches!(
+        next_event(&mut ws).await,
+        maestro_protocol::CatalogEvent::Snapshot { .. }
+    ));
+
+    let id = structured_session_with_env(&h, env).await;
+    let maestro_protocol::CatalogEvent::SessionAdded { session } = next_event(&mut ws).await else {
+        panic!("expected structured SessionAdded")
+    };
+    assert_eq!(session.activity, ActivityState::Idle);
+
+    let client = reqwest::Client::new();
+    let prompt = client
+        .post(url(
+            h.addr,
+            &format!("/api/sessions/{id}/structured/prompt"),
+        ))
+        .json(&serde_json::json!({"prompt": "hello"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(prompt.status(), reqwest::StatusCode::NO_CONTENT);
+    let report = client
+        .post(url(h.addr, &format!("/api/sessions/{id}/activity")))
+        .json(&serde_json::json!({"state": "Error"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(report.status(), reqwest::StatusCode::NO_CONTENT);
+    wait_for_session_activity(&h, id, ActivityState::Active).await;
+
+    let mut activities = Vec::new();
+    tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            if let maestro_protocol::CatalogEvent::SessionUpdated { session } =
+                next_event(&mut ws).await
+            {
+                activities.push(session.activity);
+                if session.activity == ActivityState::Idle {
+                    return;
+                }
+            }
+        }
+    })
+    .await
+    .expect("structured turn returns to idle");
+    assert!(
+        activities.contains(&ActivityState::Active),
+        "{activities:?}"
+    );
+    assert!(
+        !activities.contains(&ActivityState::Error),
+        "{activities:?}"
+    );
+
+    h.manager.kill(id, false).await.unwrap();
+}
+
 /// Each subscriber has its own lag domain now that the publisher task is gone:
 /// two clients connected at different times both get their own Snapshot.
 #[tokio::test]

crates/maestro-daemon/tests/sidecar.rs

@@ -6,8 +6,8 @@ use std::collections::HashMap;
 use std::path::PathBuf;
 
 use chrono::{DateTime, Utc};
-use maestro_daemon::sidecar::SidecarStore;
-use maestro_protocol::{Session, SessionStatus, SessionView};
+use maestro_daemon::sidecar::{PersistedEvent, SidecarStore};
+use maestro_protocol::{Session, SessionStatus, SessionView, TranscriptRow, TranscriptRowKind};
 use serde_json::Value;
 use tempfile::TempDir;
 use uuid::Uuid;
@@ -186,6 +186,13 @@ fn text_of(n: &SessionNotification) -> &str {
     }
 }
 
+fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
+    let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else {
+        panic!("expected ACP notification")
+    };
+    text_of(notification)
+}
+
 #[test]
 fn append_assigns_line_position_as_seq_and_reads_back() {
     let dir = TempDir::new().unwrap();
@@ -194,9 +201,9 @@ fn append_assigns_line_position_as_seq_and_reads_back() {
 
     let mut log = store.open_events(id).expect("open");
     assert_eq!(log.last_seq(), 0);
-    assert_eq!(log.append(&note("a")).unwrap(), 1);
-    assert_eq!(log.append(&note("b")).unwrap(), 2);
-    assert_eq!(log.append(&note("c")).unwrap(), 3);
+    assert_eq!(log.append(note("a")).unwrap(), 1);
+    assert_eq!(log.append(note("b")).unwrap(), 2);
+    assert_eq!(log.append(note("c")).unwrap(), 3);
     assert_eq!(log.last_seq(), 3);
 
     let path = store.events_path_for(id);
@@ -211,9 +218,7 @@ fn append_assigns_line_position_as_seq_and_reads_back() {
     let all = store.read_events(id, EventCursor::Since(0), 100).unwrap();
     assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 2, 3]);
     assert_eq!(
-        all.iter()
-            .map(|e| text_of(&e.notification))
-            .collect::<Vec<_>>(),
+        all.iter().map(event_text).collect::<Vec<_>>(),
         vec!["a", "b", "c"]
     );
 }
@@ -225,7 +230,7 @@ fn since_reads_forward_and_before_reads_backward_with_limit() {
     let id = Uuid::new_v4();
     let mut log = store.open_events(id).unwrap();
     for i in 1..=6 {
-        log.append(&note(&format!("n{i}"))).unwrap();
+        log.append(note(&format!("n{i}"))).unwrap();
     }
 
     // `since` is exclusive: what a client that already holds seq 2 needs.
@@ -278,7 +283,7 @@ fn malformed_lines_are_skipped_but_keep_their_seq() {
     let store = SidecarStore::new(dir.path());
     let id = Uuid::new_v4();
     let mut log = store.open_events(id).unwrap();
-    log.append(&note("first")).unwrap();
+    log.append(note("first")).unwrap();
     drop(log);
 
     // Garbage in the middle and a line that is JSON but not a notification.
@@ -289,11 +294,11 @@ fn malformed_lines_are_skipped_but_keep_their_seq() {
     std::fs::write(&path, raw).unwrap();
     let mut log = store.open_events(id).unwrap();
     assert_eq!(log.last_seq(), 3);
-    assert_eq!(log.append(&note("fourth")).unwrap(), 4);
+    assert_eq!(log.append(note("fourth")).unwrap(), 4);
 
     let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
     assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 4]);
-    assert_eq!(text_of(&all[1].notification), "fourth");
+    assert_eq!(event_text(&all[1]), "fourth");
 }
 
 #[test]
@@ -303,7 +308,7 @@ fn truncated_last_line_still_opens_and_the_next_append_starts_a_fresh_line() {
     let store = SidecarStore::new(dir.path());
     let id = Uuid::new_v4();
     let mut log = store.open_events(id).unwrap();
-    log.append(&note("whole")).unwrap();
+    log.append(note("whole")).unwrap();
     drop(log);
 
     let path = store.events_path_for(id);
@@ -319,10 +324,10 @@ fn truncated_last_line_still_opens_and_the_next_append_starts_a_fresh_line() {
     // glue onto it; the partial keeps a seq of its own.
     let mut log = store.open_events(id).unwrap();
     assert_eq!(log.last_seq(), 2);
-    assert_eq!(log.append(&note("after")).unwrap(), 3);
+    assert_eq!(log.append(note("after")).unwrap(), 3);
     let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
     assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 3]);
-    assert_eq!(text_of(&all[1].notification), "after");
+    assert_eq!(event_text(&all[1]), "after");
     let raw = std::fs::read_to_string(&path).unwrap();
     assert_eq!(raw.lines().count(), 3, "{raw}");
 }
@@ -334,8 +339,10 @@ fn delete_removes_the_event_log_and_load_all_ignores_it() {
     let id = Uuid::new_v4();
     store.write(&sample(id)).unwrap();
     let mut log = store.open_events(id).unwrap();
-    log.append(&note("x")).unwrap();
+    log.append(note("x")).unwrap();
     drop(log);
+    let legacy_approvals = dir.path().join(format!("{id}.approvals.jsonl"));
+    std::fs::write(&legacy_approvals, b"obsolete").unwrap();
     assert!(store.events_path_for(id).exists());
 
     let loaded = store.load_all().unwrap();
@@ -344,6 +351,47 @@ fn delete_removes_the_event_log_and_load_all_ignores_it() {
     store.delete(id).unwrap();
     assert!(!store.path_for(id).exists());
     assert!(!store.events_path_for(id).exists());
+    assert!(!legacy_approvals.exists());
     // Idempotent with the log already gone.
     store.delete(id).unwrap();
 }
+
+#[test]
+fn approval_rows_share_event_cursors_with_legacy_acp_lines() {
+    let dir = TempDir::new().unwrap();
+    let store = SidecarStore::new(dir.path());
+    let id = Uuid::new_v4();
+    let legacy = serde_json::to_vec(&note("before")).unwrap();
+    std::fs::create_dir_all(dir.path()).unwrap();
+    std::fs::write(store.events_path_for(id), [legacy, b"\n".to_vec()].concat()).unwrap();
+
+    let mut log = store.open_events(id).unwrap();
+    assert_eq!(
+        log.append(PersistedEvent::Approval {
+            row: TranscriptRow {
+                id: "approval:secret".into(),
+                kind: TranscriptRowKind::Approval,
+                data: serde_json::json!({"tool": "bash"}),
+            },
+        })
+        .unwrap(),
+        2
+    );
+    assert_eq!(log.append(note("after")).unwrap(), 3);
+
+    let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
+    assert_eq!(
+        all.iter().map(|event| event.seq).collect::<Vec<_>>(),
+        [1, 2, 3]
+    );
+    assert!(matches!(all[1].event, PersistedEvent::Approval { .. }));
+    assert_eq!(event_text(&all[0]), "before");
+    assert_eq!(event_text(&all[2]), "after");
+
+    let since = store.read_events(id, EventCursor::Since(1), 1).unwrap();
+    assert_eq!(since[0].seq, 2);
+    let before = store
+        .read_events(id, EventCursor::Before(Some(3)), 1)
+        .unwrap();
+    assert_eq!(before[0].seq, 2);
+}

crates/maestro-protocol/src/lib.rs

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

crates/maestro-protocol/src/routes.rs

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

docs/INSTALL.md

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

docs/RESTART.md

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

docs/SIDEBAR.md

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

docs/STRUCTURED-VIEW.md

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