fix(maestro): avoid nested session locks

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

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

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

@@ -307,11 +307,10 @@ async fn rename_session(
     }
 }
 
-/// `POST /api/sessions/:id/activity` (#105): an agent reports its own state.
+/// `POST /api/sessions/:id/activity` (#105): an agent reports a terminal
-/// The report holds above the daemon's inference for a TTL, then inference
+/// session's state. Structured sessions acknowledge and ignore these reports
-/// resumes — an agent that dies mid-run cannot pin its session at `Error`
+/// because their activity comes from ACP. Unknown session is 404; an unknown
-/// forever. Unknown session is 404; an unknown state value is rejected by the
+/// state value is rejected by the JSON decoder (422). Answers `204 No Content`.
-/// JSON decoder (422) rather than coerced. Answers `204 No Content`.
 async fn report_activity(
     State(s): State<LocalApiState>,
     Path(id): Path<Uuid>,

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;
 
@@ -36,6 +37,7 @@ pub struct Projection {
     pending_tool_updates: HashMap<String, Vec<maestro_acp::schema::ToolCallUpdateFields>>,
     state: ReducedState,
     active_turn: Option<u64>,
+    turn_failed: bool,
     next_turn: u64,
     turn_tool_calls: HashSet<String>,
 }
@@ -109,11 +111,21 @@ impl Projection {
         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();
@@ -141,6 +153,7 @@ impl Projection {
             return None;
         }
         self.active_turn = None;
+        self.turn_failed = lifecycle == TurnLifecycle::Failure;
         self.state.turn.responding = false;
         self.state.turn.thinking = false;
         self.turn_tool_calls.clear();
@@ -210,6 +223,21 @@ impl Projection {
         })
     }
 
+    /// Undo an `append`ed row that a caller decides never happened — used
+    /// when persisting the row's own event fails (#184), so a snapshot taken
+    /// after the failure cannot show a row durable replay would never
+    /// reproduce. Only ever called for a row that was just appended and
+    /// never patched, so removing it by id fully reverses `append`.
+    pub fn rollback_row(&mut self, id: &str) {
+        let Some(index) = self.row_index.remove(id) else {
+            return;
+        };
+        self.rows.remove(index);
+        for (shifted_index, row) in self.rows.iter().enumerate().skip(index) {
+            self.row_index.insert(row.id.clone(), shifted_index);
+        }
+    }
+
     fn content(
         &mut self,
         seq: u64,
@@ -712,6 +740,74 @@ mod tests {
         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();

crates/maestro-daemon/src/session.rs

@@ -25,7 +25,7 @@ use crate::panes::Panes;
 use crate::projection::{Projection, ProjectionUpdate, TurnLifecycle};
 use crate::pty::{pipe_pane_command, PtyBridge};
 use crate::sidecar::{
-    EventCursor, EventLog, LoggedEvent, PersistedEvent, SidecarError, SidecarStore,
+    AppendError, EventCursor, EventLog, LoggedEvent, PersistedEvent, SidecarError, SidecarStore,
 };
 use crate::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord};
 use crate::tmux::TmuxError;
@@ -535,6 +535,8 @@ impl SessionManager {
             return Ok(false);
         };
         let _ = stream.projected_live.send(projected);
+        drop(stream);
+        sync_structured_activity(&self.inner, id).await;
         Ok(true)
     }
 
@@ -558,6 +560,8 @@ impl SessionManager {
         }
         let (turn, update) = stream.projection.begin_turn();
         let _ = stream.projected_live.send(update);
+        drop(stream);
+        sync_structured_activity(&self.inner, id).await;
         Ok(BeginAcpTurn::Started(acp, turn))
     }
 
@@ -614,6 +618,8 @@ impl SessionManager {
                                 };
                                 manager.finish_queued_acp_turn(id, turn, lifecycle).await;
                             });
+                            drop(stream);
+                            sync_structured_activity(&self.inner, id).await;
                             return Ok(QueueAcpPrompt::Queued);
                         }
                         Err(message) => {
@@ -625,6 +631,8 @@ impl SessionManager {
                                     "newly claimed turn remains current while stream is locked",
                                 );
                             let _ = stream.projected_live.send(projected);
+                            drop(stream);
+                            sync_structured_activity(&self.inner, id).await;
                             return Ok(QueueAcpPrompt::Failed(message));
                         }
                     }
@@ -669,6 +677,8 @@ impl SessionManager {
         if let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) {
             let _ = stream.projected_live.send(projected);
         }
+        drop(stream);
+        sync_structured_activity(&self.inner, id).await;
     }
 
     async fn fail_cancelled_acp_turn(
@@ -741,6 +751,7 @@ impl SessionManager {
             .expect("active turn remains current while stream is locked");
         let _ = stream.projected_live.send(projected);
         drop(stream);
+        sync_structured_activity(&self.inner, id).await;
         if let Some(turn) = cancelled_queued_turn {
             self.schedule_cancelled_acp_turn_failure(id, event_stream, turn);
         }
@@ -789,6 +800,8 @@ impl SessionManager {
         if let Some(update) = stream.projection.resolve_approval(nonce) {
             let _ = stream.projected_live.send(update);
         }
+        drop(stream);
+        sync_structured_activity(&self.inner, id).await;
         Ok(Some(()))
     }
 
@@ -1133,13 +1146,13 @@ impl SessionManager {
                 SessionStatus::Running
             },
             created_at: Utc::now(),
-            // A freshly-spawned session is producing its initial output; the
+            // A terminal spawn begins active until its watcher sees otherwise;
-            // activity watcher refines `foreground`/`activity` on its next tick.
+            // a structured spawn has no turn yet and begins idle.
             foreground: None,
-            activity: if acp_failure.is_some() {
+            activity: match (req.view, acp_failure.is_some()) {
-                ActivityState::Error
+                (_, true) => ActivityState::Error,
-            } else {
+                (SessionView::Structured, false) => ActivityState::Idle,
-                ActivityState::Active
+                (SessionView::Terminal, false) => ActivityState::Active,
             },
             activity_detail: acp_failure,
             last_activity_at: Utc::now(),
@@ -1201,7 +1214,7 @@ impl SessionManager {
         }
         if let Some((acp, events)) = event_pump {
             let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
-            start_acp_permission_pump(&acp, events);
+            start_acp_permission_pump(&self.inner, id, &acp, events);
             if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
                 record.acp_event_pump = Some(done);
             }
@@ -1331,6 +1344,7 @@ impl SessionManager {
                     return;
                 };
                 run_liveness_pass(&inner).await;
+                run_structured_activity_pass(&inner).await;
                 // Same tick, no new timer (NAP-340): refresh activity for the
                 // survivors and emit SessionUpdated on any transition.
                 run_activity_pass(&inner).await;
@@ -1499,11 +1513,15 @@ impl SessionManager {
                 SessionStatus::Dead
             };
             if recoverable {
-                session.activity = if failure.is_some() {
+                let activity = if failure.is_some() {
                     ActivityState::Error
                 } else {
-                    ActivityState::Active
+                    ActivityState::Idle
                 };
+                if session.activity != activity {
+                    session.activity_since = Utc::now();
+                }
+                session.activity = activity;
                 session.activity_detail = failure;
             }
 
@@ -1590,7 +1608,7 @@ impl SessionManager {
             );
             if let Some((acp, events)) = event_pump {
                 let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
-                start_acp_permission_pump(&acp, events);
+                start_acp_permission_pump(&self.inner, id, &acp, events);
                 if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
                     record.acp_event_pump = Some(done);
                 }
@@ -1750,12 +1768,10 @@ impl SessionManager {
         Ok(session)
     }
 
-    /// Accept an agent-reported activity state (#105), the only path that can
+    /// Accept an agent-reported activity state (#105) for a terminal session.
-    /// produce `ActivityState::Error`. The report outranks the activity pass's
+    /// Structured sessions derive their state from ACP instead, so their
-    /// own inference until its TTL lapses — so an agent that dies mid-run
+    /// inherited hooks are acknowledged but ignored. An unknown session is
-    /// cannot pin its session at `Error` forever. An unknown session is
+    /// [`SessionError::NotFound`]; the reported `detail` is capped server-side.
-    /// [`SessionError::NotFound`]; the reported `detail` is capped
-    /// server-side.
     ///
     /// A report that changes the effective state restarts `activity_since`
     /// (#104's clock); re-reporting an unchanged state only extends the TTL
@@ -1769,6 +1785,9 @@ impl SessionManager {
         let updated = {
             let mut sessions = self.inner.sessions.lock().await;
             let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
+            if rec.session.view == SessionView::Structured {
+                return Ok(());
+            }
             let detail = crate::activity::truncate_detail(detail);
             let changed = rec.session.activity != state || rec.session.activity_detail != detail;
             if rec.session.activity != state {
@@ -1983,10 +2002,94 @@ fn apply_persisted_event(projection: &mut Projection, event: &LoggedEvent) -> Pr
     }
 }
 
-fn start_acp_permission_pump(acp: &Arc<AcpSession>, events: Arc<Mutex<AcpEventStream>>) {
+async fn sync_structured_activity(inner: &Arc<Inner>, id: Uuid) {
+    let events = {
+        let sessions = inner.sessions.lock().await;
+        let Some(record) = sessions.get(&id) else {
+            return;
+        };
+        if record.session.view != SessionView::Structured {
+            return;
+        }
+        record.acp_events.clone()
+    };
+
+    match events {
+        Some(events) => sync_structured_activity_from_stream(inner, id, events).await,
+        None => apply_structured_activity(inner, id, None, ActivityState::Error).await,
+    }
+}
+
+async fn sync_structured_activity_from_stream(
+    inner: &Arc<Inner>,
+    id: Uuid,
+    events: Arc<Mutex<AcpEventStream>>,
+) {
+    let event_stream = events.lock().await;
+    let raw = event_stream.projection.activity();
+    // Stream state must stay fixed until its catalog transition is applied;
+    // every other path releases `sessions` before awaiting this stream lock.
+    apply_structured_activity(inner, id, Some(&events), raw).await;
+}
+
+async fn apply_structured_activity(
+    inner: &Arc<Inner>,
+    id: Uuid,
+    expected_events: Option<&Arc<Mutex<AcpEventStream>>>,
+    raw: ActivityState,
+) {
+    let updated = {
+        let mut sessions = inner.sessions.lock().await;
+        let Some(record) = sessions.get_mut(&id) else {
+            return;
+        };
+        if record.session.view != SessionView::Structured {
+            return;
+        }
+        let current_events = record.acp_events.as_ref();
+        let stream_matches = match (expected_events, current_events) {
+            (Some(expected), Some(current)) => Arc::ptr_eq(expected, current),
+            (None, None) => true,
+            _ => false,
+        };
+        if !stream_matches {
+            return;
+        }
+        let raw = if record.session.status == SessionStatus::Dead {
+            ActivityState::Error
+        } else {
+            raw
+        };
+        let activity = crate::activity::apply_focus_suppression(
+            raw,
+            record.focused_since,
+            std::time::Instant::now(),
+            &mut record.awaiting_suppressed,
+        );
+        if record.session.activity == activity {
+            None
+        } else {
+            record.session.activity = activity;
+            record.session.activity_since = Utc::now();
+            Some(record.session.clone())
+        }
+    };
+    if let Some(session) = updated {
+        let _ = inner.events.send(CatalogEvent::SessionUpdated { session });
+    }
+}
+
+fn start_acp_permission_pump(
+    inner: &Arc<Inner>,
+    id: Uuid,
+    acp: &Arc<AcpSession>,
+    events: Arc<Mutex<AcpEventStream>>,
+) {
     let Some(mut requests) = acp.take_permission_requests() else {
         return;
     };
+    let weak_inner = Arc::downgrade(inner);
+    let weak_acp = Arc::downgrade(acp);
     tokio::spawn(async move {
         while let Some(PermissionRequest {
             turn,
@@ -1999,40 +2102,95 @@ fn start_acp_permission_pump(acp: &Arc<AcpSession>, events: Arc<Mutex<AcpEventSt
                 let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
                 continue;
             }
-            let nonce = Uuid::new_v4().to_string();
+            let outcome = record_permission_request(&mut events, request, response);
-            let data = match serde_json::to_value(&request) {
+            let fatal = matches!(outcome, RecordOutcome::Fatal);
-                Ok(value) => value,
+            if let RecordOutcome::Live(update) = outcome {
-                Err(_) => {
+                let _ = events.projected_live.send(*update);
-                    let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
+            }
-                    continue;
+            drop(events);
-                }
+            if fatal {
-            };
+                tracing::error!(%id, "approval history append rollback did not restore consistency");
-            // ACP requests serialize through its connection. Keep the newest
+                if let Some(acp) = weak_acp.upgrade() {
-            // one authoritative if an adapter nonetheless overlaps them.
+                    acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
-            cancel_pending_approvals(&mut events);
+                        .await;
-            events.pending_approvals.insert(
+                }
-                nonce.clone(),
+                if let Some(inner) = weak_inner.upgrade() {
-                PendingApproval {
+                    mark_session_dead(&inner, id).await;
-                    response,
+                }
-                    options: request.options,
+                break;
-                },
+            }
-            );
+            if let Some(inner) = weak_inner.upgrade() {
-            let update = events.projection.request_approval(nonce, data);
+                sync_structured_activity(&inner, id).await;
-            if let Some(TranscriptDelta::Append { row }) = &update.transcript {
-                let event = PersistedEvent::Approval { row: row.clone() };
-                if let Err(error) = events.log.append(&event) {
-                    tracing::error!(%error, "approval history append failed");
-                    cancel_pending_approvals(&mut events);
-                    continue;
-                }
-                let seq = events.log.last_seq();
-                let _ = events.live.send(LoggedEvent { seq, event });
             }
-            let _ = events.projected_live.send(update);
         }
     });
 }
 
+/// What became of one [`record_permission_request`] call.
+enum RecordOutcome {
+    /// Persisted; broadcast this live.
+    Live(Box<ProjectionUpdate>),
+    /// Not persisted, but the log rolled back cleanly: the approval was
+    /// cancelled and the session lives on.
+    Dropped,
+    /// Not persisted, and the rollback did not restore a consistent log.
+    /// The caller must treat the session as dead rather than keep appending
+    /// to an `EventLog` whose `last_seq` may no longer match its file.
+    Fatal,
+}
+
+/// Install `request` as the pending approval — cancelling whatever was
+/// already pending, so the newest request wins (#139) — and persist its
+/// transcript row before it is exposed live. On append failure the row is
+/// rolled back from the projection so a later `TranscriptSnapshot` cannot
+/// show a row durable replay could never reproduce (#184); the ACP callback
+/// still resolves `Cancelled`, exactly as any other cancelled approval does.
+fn record_permission_request(
+    events: &mut AcpEventStream,
+    request: maestro_acp::schema::RequestPermissionRequest,
+    response: oneshot::Sender<maestro_acp::schema::RequestPermissionOutcome>,
+) -> RecordOutcome {
+    let nonce = Uuid::new_v4().to_string();
+    let data = match serde_json::to_value(&request) {
+        Ok(value) => value,
+        Err(_) => {
+            let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
+            return RecordOutcome::Dropped;
+        }
+    };
+    cancel_pending_approvals(events);
+    events.pending_approvals.insert(
+        nonce.clone(),
+        PendingApproval {
+            response,
+            options: request.options,
+        },
+    );
+    let update = events.projection.request_approval(nonce, data);
+    let Some(TranscriptDelta::Append { row }) = &update.transcript else {
+        return RecordOutcome::Live(Box::new(update));
+    };
+    let event = PersistedEvent::Approval { row: row.clone() };
+    match events.log.append(&event) {
+        Ok(seq) => {
+            let _ = events.live.send(LoggedEvent { seq, event });
+            RecordOutcome::Live(Box::new(update))
+        }
+        Err(AppendError::Failed(error)) => {
+            tracing::error!(%error, "approval history append failed");
+            events.projection.rollback_row(&row.id);
+            cancel_pending_approvals(events);
+            RecordOutcome::Dropped
+        }
+        Err(fatal @ AppendError::Fatal { .. }) => {
+            tracing::error!(error = %fatal, "approval history append failed unrecoverably");
+            events.projection.rollback_row(&row.id);
+            cancel_pending_approvals(events);
+            RecordOutcome::Fatal
+        }
+    }
+}
+
 fn cancel_pending_approvals(events: &mut AcpEventStream) {
     for (_, pending) in events.pending_approvals.drain() {
         let _ = pending
@@ -2176,7 +2334,7 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
     }
 }
 
-/// Refresh each live session's inferred activity (NAP-340) and emit
+/// Refresh each live terminal session's inferred activity (NAP-340) and emit
 /// `SessionUpdated` only when `activity` or `foreground` actually changes — so a
 /// steady state produces no diffs. Runs on the existing liveness tick.
 async fn run_activity_pass(inner: &Arc<Inner>) {
@@ -2192,7 +2350,10 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
         // kept as-is so the UI can still show what it was doing when it died.
         sessions
             .iter()
-            .filter(|(_, r)| r.session.status == SessionStatus::Running)
+            .filter(|(_, r)| {
+                r.session.status == SessionStatus::Running
+                    && r.session.view == SessionView::Terminal
+            })
             .map(|(id, _)| *id)
             .collect()
     };
@@ -2290,6 +2451,21 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
     }
 }
 
+/// Reapply focus suppression to structured activity on the liveness cadence.
+async fn run_structured_activity_pass(inner: &Arc<Inner>) {
+    let ids: Vec<Uuid> = inner
+        .sessions
+        .lock()
+        .await
+        .iter()
+        .filter(|(_, record)| record.session.view == SessionView::Structured)
+        .map(|(id, _)| *id)
+        .collect();
+    for id in ids {
+        sync_structured_activity(inner, id).await;
+    }
+}
+
 /// Refresh each running session's blocked-work indicator (#83): what its
 /// worktree holds right now that would block an unforced kill. Rides the
 /// liveness tick on its own (slower) cadence — see
@@ -2391,6 +2567,12 @@ async fn mark_session_dead_with_worktree_settlement(
         match sessions.get_mut(&id) {
             Some(record) if record.session.status == SessionStatus::Running => {
                 record.session.status = SessionStatus::Dead;
+                if record.session.view == SessionView::Structured
+                    && record.session.activity != ActivityState::Error
+                {
+                    record.session.activity = ActivityState::Error;
+                    record.session.activity_since = Utc::now();
+                }
                 // Release anyone attached; the scrollback survives for a later
                 // attach.
                 record.bridge.seal();
@@ -2561,13 +2743,322 @@ async fn cleanup_record(
 
 #[cfg(test)]
 mod tests {
-    use super::{permission_outcome, ApprovalDecision};
+    use std::fs::File;
-    use maestro_acp::schema::{PermissionOption, PermissionOptionKind, RequestPermissionOutcome};
+    use std::sync::Arc;
+    use std::task::Poll;
+    use std::time::Duration;
+
+    use maestro_acp::schema::{
+        PermissionOption, PermissionOptionKind, RequestPermissionOutcome, RequestPermissionRequest,
+        ToolCallUpdate, ToolCallUpdateFields,
+    };
+    use maestro_protocol::{ActivityState, CreateSessionRequest, SessionView};
+
+    use super::{
+        cancel_pending_approvals, permission_outcome, record_permission_request,
+        run_structured_activity_pass, sync_structured_activity_from_stream, AcpDispatchState,
+        AcpEventStream, ApprovalDecision, RecordOutcome, SessionManager,
+    };
+    use crate::activity::FOCUS_CLEAR_HOLD;
+    use crate::panes::FakePanes;
+    use crate::projection::Projection;
+    use crate::sidecar::{
+        EventCursor, EventLog, FaultyLogSink, PersistedEvent, SharedBuf, SidecarStore,
+    };
+    use crate::worktree::WorktreeManager;
 
     fn option(id: &str, kind: PermissionOptionKind) -> PermissionOption {
         PermissionOption::new(id.to_owned(), id, kind)
     }
 
+    fn permission_request() -> RequestPermissionRequest {
+        RequestPermissionRequest::new(
+            "acp-session",
+            ToolCallUpdate::new("tool-1", ToolCallUpdateFields::new()),
+            vec![option("allow-once", PermissionOptionKind::AllowOnce)],
+        )
+    }
+
+    fn stream(log: EventLog) -> AcpEventStream {
+        let (live, _live_rx) = tokio::sync::broadcast::channel(16);
+        let (projected_live, _projected_rx) = tokio::sync::broadcast::channel(16);
+        AcpEventStream {
+            log,
+            live,
+            projection: Projection::default(),
+            projected_live,
+            pending_approvals: std::collections::HashMap::new(),
+            operation: std::sync::Arc::new(tokio::sync::Mutex::new(())),
+            dispatch: AcpDispatchState::default(),
+        }
+    }
+
+    fn terminal_request(cwd: &std::path::Path) -> CreateSessionRequest {
+        CreateSessionRequest {
+            view: SessionView::Terminal,
+            cwd: cwd.to_path_buf(),
+            command: "sh".into(),
+            env: std::collections::HashMap::new(),
+            metadata: std::collections::HashMap::new(),
+            worktree: None,
+        }
+    }
+
+    async fn manager_with_structured_stream() -> (
+        tempfile::TempDir,
+        SessionManager,
+        uuid::Uuid,
+        Arc<tokio::sync::Mutex<AcpEventStream>>,
+        uuid::Uuid,
+    ) {
+        let state = tempfile::tempdir().unwrap();
+        let manager = SessionManager::new(
+            Arc::new(FakePanes::new()),
+            SidecarStore::new(state.path().join("sessions")),
+            WorktreeManager::new(None),
+            state.path().join("fifos"),
+        );
+        let structured = manager
+            .create(terminal_request(state.path()))
+            .await
+            .expect("create structured fixture");
+        let terminal = manager
+            .create(terminal_request(state.path()))
+            .await
+            .expect("create unrelated terminal");
+        let events = Arc::new(tokio::sync::Mutex::new(stream(
+            manager
+                .inner
+                .sidecar
+                .open_events(structured.id)
+                .expect("open event log"),
+        )));
+        {
+            let mut event_stream = events.lock().await;
+            event_stream.projection.begin_turn();
+            event_stream
+                .projection
+                .request_approval("nonce".into(), serde_json::json!({}));
+        }
+        {
+            let mut sessions = manager.inner.sessions.lock().await;
+            let record = sessions.get_mut(&structured.id).unwrap();
+            record.session.view = SessionView::Structured;
+            record.session.activity = ActivityState::AwaitingInput;
+            record.focused_since = Some(std::time::Instant::now() - FOCUS_CLEAR_HOLD);
+            record.acp_events = Some(events.clone());
+        }
+        (state, manager, structured.id, events, terminal.id)
+    }
+
+    #[tokio::test]
+    async fn blocked_structured_activity_read_does_not_block_the_catalog() {
+        let (state, manager, structured_id, events, terminal_id) =
+            manager_with_structured_stream().await;
+        let held_stream = events.lock().await;
+        let activity_pass = run_structured_activity_pass(&manager.inner);
+        tokio::pin!(activity_pass);
+        assert!(matches!(
+            futures_util::poll!(activity_pass.as_mut()),
+            Poll::Pending
+        ));
+
+        tokio::time::timeout(Duration::from_secs(1), manager.snapshot())
+            .await
+            .expect("catalog snapshot blocked behind another session's event stream");
+        tokio::time::timeout(
+            Duration::from_secs(1),
+            manager.send_keys(terminal_id, b"responsive"),
+        )
+        .await
+        .expect("terminal input blocked behind another session's event stream")
+        .expect("terminal input");
+        let created = tokio::time::timeout(
+            Duration::from_secs(1),
+            manager.create(terminal_request(state.path())),
+        )
+        .await
+        .expect("session create blocked behind another session's event stream")
+        .expect("create unrelated session");
+        tokio::time::timeout(Duration::from_secs(1), manager.kill(created.id, false))
+            .await
+            .expect("session kill blocked behind another session's event stream")
+            .expect("kill unrelated session");
+
+        drop(held_stream);
+        tokio::time::timeout(Duration::from_secs(1), activity_pass)
+            .await
+            .expect("activity pass did not resume");
+        let structured = manager
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|session| session.id == structured_id)
+            .unwrap();
+        assert_eq!(structured.activity, ActivityState::Idle);
+    }
+
+    #[tokio::test]
+    async fn delayed_activity_from_a_replaced_stream_is_discarded() {
+        let (_state, manager, structured_id, old_events, _terminal_id) =
+            manager_with_structured_stream().await;
+        let replacement = Arc::new(tokio::sync::Mutex::new(stream(
+            manager
+                .inner
+                .sidecar
+                .open_events(uuid::Uuid::new_v4())
+                .expect("open replacement event log"),
+        )));
+        let mut sessions = manager.inner.sessions.lock().await;
+        let sync =
+            sync_structured_activity_from_stream(&manager.inner, structured_id, old_events.clone());
+        tokio::pin!(sync);
+        assert!(matches!(futures_util::poll!(sync.as_mut()), Poll::Pending));
+        let record = sessions.get_mut(&structured_id).unwrap();
+        record.acp_events = Some(replacement.clone());
+        record.session.activity = ActivityState::Error;
+        drop(sessions);
+
+        tokio::time::timeout(Duration::from_secs(1), sync)
+            .await
+            .expect("stale activity sync did not finish");
+        let sessions = manager.inner.sessions.lock().await;
+        let record = sessions.get(&structured_id).unwrap();
+        assert_eq!(record.session.activity, ActivityState::Error);
+        assert!(Arc::ptr_eq(
+            record.acp_events.as_ref().unwrap(),
+            &replacement
+        ));
+    }
+
+    /// A handle that can never append *or* roll back: opened read-only, so
+    /// both `write_all` and the rollback `truncate` fail with the OS's own
+    /// error (`ftruncate` requires a writable fd too) instead of a
+    /// hand-rolled one. A genuinely dead fd like this is exactly the case
+    /// where consistency cannot be promised, so it must come back fatal.
+    fn unwritable_log(path: &std::path::Path, last_seq: u64) -> EventLog {
+        EventLog::for_test(File::open(path).expect("open read-only handle"), last_seq)
+    }
+
+    #[tokio::test]
+    async fn failed_approval_append_with_no_usable_rollback_is_fatal_and_leaves_no_phantom_row() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = SidecarStore::new(dir.path());
+        let id = uuid::Uuid::new_v4();
+        // Create the log file for real so a read-only handle onto it exists,
+        // then swap in the broken handle: the write is what must fail, not
+        // the open.
+        drop(store.open_events(id).unwrap());
+        let mut events = stream(unwritable_log(&store.events_path_for(id), 0));
+
+        let (response_tx, response_rx) = tokio::sync::oneshot::channel();
+        let outcome = record_permission_request(&mut events, permission_request(), response_tx);
+
+        assert!(
+            matches!(outcome, RecordOutcome::Fatal),
+            "a write failure whose rollback also fails must be reported fatal, not dropped"
+        );
+        assert!(
+            events.pending_approvals.is_empty(),
+            "the failed nonce must not linger as pending"
+        );
+        assert_eq!(
+            response_rx.await.unwrap(),
+            RequestPermissionOutcome::Cancelled,
+            "ACP must see the withdrawn-request outcome, not a hang"
+        );
+
+        let state = events.projection.state();
+        assert!(
+            state.pending_approval.is_none(),
+            "reduced state must not advertise a pending approval that never persisted"
+        );
+        assert!(
+            events.projection.snapshot(0).rows.is_empty(),
+            "the live projection must not carry a row durable replay cannot reproduce"
+        );
+
+        // Durable replay of the untouched real log must agree with the live
+        // projection: nothing happened.
+        let replayed = store
+            .read_events(id, EventCursor::Since(0), 100)
+            .expect("replay");
+        assert!(
+            replayed.is_empty(),
+            "durable replay must not diverge from the rolled-back live projection"
+        );
+        assert_eq!(
+            events.log.last_seq(),
+            0,
+            "cursor must stay put on a failed append"
+        );
+    }
+
+    #[tokio::test]
+    async fn approval_after_a_failed_one_persists_and_replays_cleanly() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = SidecarStore::new(dir.path());
+        let id = uuid::Uuid::new_v4();
+
+        // A fault that fails the write once and then behaves like a healthy
+        // sink — the same `EventLog` instance must recover, unlike a handle
+        // swapped in from a healthy disk, which production never does
+        // (#184).
+        let buf = SharedBuf::default();
+        let sink = FaultyLogSink {
+            buf: buf.clone(),
+            fail_write_after: Some(0),
+            ..Default::default()
+        };
+        let mut events = stream(EventLog::for_test(sink, 0));
+
+        let (first_tx, first_rx) = tokio::sync::oneshot::channel();
+        assert!(matches!(
+            record_permission_request(&mut events, permission_request(), first_tx),
+            RecordOutcome::Dropped
+        ));
+        assert_eq!(first_rx.await.unwrap(), RequestPermissionOutcome::Cancelled);
+        assert_eq!(
+            events.log.last_seq(),
+            0,
+            "the failed attempt must not have advanced the cursor"
+        );
+
+        // Same instance, no swap: the sink is healthy again, so the next
+        // request over the same connection must not be poisoned by the
+        // earlier failure (concurrent/ordering audit for #184).
+        let (second_tx, _second_rx) = tokio::sync::oneshot::channel();
+        let outcome = record_permission_request(&mut events, permission_request(), second_tx);
+
+        assert!(
+            matches!(outcome, RecordOutcome::Live(_)),
+            "a healthy append must be exposed live"
+        );
+        assert_eq!(events.pending_approvals.len(), 1);
+        let rows = events.projection.snapshot(0).rows;
+        assert_eq!(rows.len(), 1, "no phantom row survives from the failure");
+
+        // Durable replay via the real read path, over the bytes the same
+        // instance actually wrote.
+        std::fs::create_dir_all(store.dir()).unwrap();
+        std::fs::write(store.events_path_for(id), buf.snapshot()).unwrap();
+        let replayed = store
+            .read_events(id, EventCursor::Since(0), 100)
+            .expect("replay");
+        assert_eq!(replayed.len(), 1, "the failed attempt left no line behind");
+        let PersistedEvent::Approval { row } = &replayed[0].event else {
+            panic!("approval event")
+        };
+        assert_eq!(
+            row.id, rows[0].id,
+            "live and durable must agree on identity"
+        );
+        assert_eq!(replayed[0].seq, events.log.last_seq());
+
+        cancel_pending_approvals(&mut events);
+        assert!(events.pending_approvals.is_empty());
+    }
+
     fn selected_id(outcome: RequestPermissionOutcome) -> Option<String> {
         match outcome {
             RequestPermissionOutcome::Selected(selected) => Some(selected.option_id.to_string()),

crates/maestro-daemon/src/sidecar.rs

@@ -112,7 +112,7 @@ impl SidecarStore {
         }
         file.seek(SeekFrom::End(0))?;
         Ok(EventLog {
-            file,
+            file: Box::new(file),
             last_seq: lines,
         })
     }
@@ -296,15 +296,82 @@ impl From<&PersistedEvent> for PersistedEvent {
     }
 }
 
+/// 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
@@ -314,16 +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, event: impl Into<PersistedEvent>) -> Result<u64, SidecarError> {
+    ///
+    /// 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)?,
+            PersistedEvent::Notification(notification) => serde_json::to_vec(notification)
-            _ => serde_json::to_vec(&event)?,
+                .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

@@ -136,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}");
 
@@ -204,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();
@@ -476,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}")),
@@ -925,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

@@ -11,7 +11,8 @@ use common::DaemonFixture;
 use futures_util::{SinkExt, StreamExt};
 use maestro_daemon::local_api;
 use maestro_protocol::{
-    ApiError, CreateSessionRequest, Session, SessionStatus, SessionView, WorktreeSpec,
+    ActivityState, ApiError, CreateSessionRequest, Session, SessionStatus, SessionView,
+    WorktreeSpec,
 };
 use tempfile::TempDir;
 use tokio::net::TcpListener;
@@ -167,6 +168,29 @@ async fn structured_session_with_env(h: &Harness, env: HashMap<String, String>)
         .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;
@@ -318,6 +342,7 @@ async fn structured_approval_is_secret_single_use_and_race_safe() {
     })
     .await
     .unwrap();
+    wait_for_session_activity(&h, id, ActivityState::AwaitingInput).await;
     assert!(
         !fake.log().contains(&nonce),
         "daemon nonce leaked to ACP agent"
@@ -373,6 +398,7 @@ async fn structured_approval_is_secret_single_use_and_race_safe() {
     })
     .await
     .expect("resolved approval clears the reduced state");
+    wait_for_session_activity(&h, id, ActivityState::Idle).await;
     assert_eq!(
         client
             .post(url(
@@ -429,6 +455,7 @@ async fn structured_prompt_failure_clears_the_turn() {
     })
     .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"}))
@@ -1350,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-protocol/src/lib.rs

@@ -156,9 +156,9 @@ pub enum SessionStatus {
     Dead,
 }
 
-/// Inferred session activity, orthogonal to [`SessionStatus`] (NAP-340). The
+/// Session activity, orthogonal to [`SessionStatus`] (NAP-340). Terminal
-/// daemon derives this from tmux pane state, PTY quiescence, and `/proc`
+/// sessions use pane/process inference and agent reports; structured sessions
-/// inspection — never by parsing the terminal byte stream.
+/// 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
+    /// An agent reported a failure, or a structured session/turn failed.
-    /// inferred by the daemon; only set when an agent reports it (#105).
     Error,
 }
 

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
+This applies uniformly to `AwaitingInput` from terminal inference, an agent
-own inference or an agent's report (#105) — since the rail draws them
+report (#105), or a structured session's pending approval, since the rail draws
-identically. It does not touch `Error`; that state's own TTL (below) is
+them identically. It does not touch `Error`; terminal reports keep their TTL,
-unrelated and unaffected by focus.
+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

@@ -302,7 +302,14 @@ directory: the sidecar itself (`<id>.json`), plus two next to it:
   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.
+  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
@@ -314,9 +321,15 @@ 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: it cancels the pending approval and lets the session continue,
+   narrower: as long as the log rolled back cleanly, it only cancels the
-   without marking it dead.) `mark_session_dead` applies the same
+   pending approval and rolls the unpersisted `Approval` row back out of the
-   natural-exit worktree rule
+   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
@@ -343,7 +356,7 @@ 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`/`Active`; failure (no stored `acp_session_id`, or the handshake
+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.