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/projection.rs

@@ -223,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,
@@ -725,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;
@@ -2003,6 +2003,41 @@ fn apply_persisted_event(projection: &mut Projection, event: &LoggedEvent) -> Pr
 }
 
 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 {
@@ -2011,12 +2046,19 @@ async fn sync_structured_activity(inner: &Arc<Inner>, id: Uuid) {
         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 if let Some(events) = &record.acp_events {
-            events.lock().await.projection.activity()
         } else {
-            ActivityState::Error
+            raw
         };
         let activity = crate::activity::apply_focus_suppression(
             raw,
@@ -2047,6 +2089,7 @@ fn start_acp_permission_pump(
         return;
     };
     let weak_inner = Arc::downgrade(inner);
+    let weak_acp = Arc::downgrade(acp);
     tokio::spawn(async move {
         while let Some(PermissionRequest {
             turn,
@@ -2059,41 +2102,23 @@ fn start_acp_permission_pump(
                 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;
-                }
-            };
-            // ACP requests serialize through its connection. Keep the newest
-            // one authoritative if an adapter nonetheless overlaps them.
-            cancel_pending_approvals(&mut events);
-            events.pending_approvals.insert(
-                nonce.clone(),
-                PendingApproval {
-                    response,
-                    options: request.options,
-                },
-            );
-            let update = events.projection.request_approval(nonce, data);
-            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);
-                    drop(events);
-                    if let Some(inner) = weak_inner.upgrade() {
-                        sync_structured_activity(&inner, id).await;
-                    }
-                    continue;
-                }
-                let seq = events.log.last_seq();
-                let _ = events.live.send(LoggedEvent { seq, event });
             }
-            let _ = events.projected_live.send(update);
             drop(events);
+            if fatal {
+                tracing::error!(%id, "approval history append rollback did not restore consistency");
+                if let Some(acp) = weak_acp.upgrade() {
+                    acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
+                        .await;
+                }
+                if let Some(inner) = weak_inner.upgrade() {
+                    mark_session_dead(&inner, id).await;
+                }
+                break;
+            }
             if let Some(inner) = weak_inner.upgrade() {
                 sync_structured_activity(&inner, id).await;
             }
@@ -2101,6 +2126,71 @@ fn start_acp_permission_pump(
     });
 }
 
+/// 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
@@ -2653,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"
+        );
+    }
 }

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