From c067f48a245384291abbfb97012d80a0ece2779f Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 5 Sep 2026 04:34:47 +0100 Subject: [PATCH] fix(maestro): roll back phantom approval row on log append failure request_approval() appended the Approval row to the live projection before the daemon tried to persist it. On append failure the pending nonce was cancelled but the row stayed in the in-memory transcript, so a WebSocket snapshot taken afterward showed a row durable replay could never reproduce. Add Projection::rollback_row to undo the append and call it alongside the existing cancel_pending_approvals on failure. Extract the pump's per-request handling into record_permission_request so the failure path is directly testable without a live ACP connection. Fault-injection regression tests use a read-only file handle (EventLog::for_test, cfg(test)) to force append() to fail deterministically, then assert the live snapshot and durable replay agree (both empty) and a subsequent request still persists and replays correctly. Fixes #184. --- crates/maestro-daemon/src/projection.rs | 71 ++++++++ crates/maestro-daemon/src/session.rs | 220 ++++++++++++++++++++---- crates/maestro-daemon/src/sidecar.rs | 9 + docs/STRUCTURED-VIEW.md | 8 +- 4 files changed, 270 insertions(+), 38 deletions(-) diff --git a/crates/maestro-daemon/src/projection.rs b/crates/maestro-daemon/src/projection.rs index 043e6e6..68dd098 100644 --- a/crates/maestro-daemon/src/projection.rs +++ b/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,62 @@ 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") + }; + assert_eq!(projection.snapshot(0).rows.len(), 2); + + projection.rollback_row(&row.id); + + assert_eq!(projection.snapshot(0).rows.len(), 1); + assert_eq!(projection.snapshot(0).rows[0].kind, TranscriptRowKind::User); + + // The row after the rolled-back one must still be independently + // patchable — proves `row_index` was reindexed, not just `rows`. + let update = projection.apply( + 2, + &event(serde_json::json!({ + "sessionUpdate": "user_message_chunk", + "messageId": "m1", + "content": {"type": "text", "text": " there"}, + })), + ); + let TranscriptDelta::Patch { row } = update.transcript.unwrap() else { + panic!("patched row") + }; + assert_eq!(row.data.pointer("/content/text").unwrap(), "hi there"); + assert_eq!(projection.snapshot(0).rows.len(), 1); + } + + #[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(); diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index bdcba65..0690958 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -2059,40 +2059,10 @@ fn start_acp_permission_pump( let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled); continue; } - 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); - 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 update = record_permission_request(&mut events, request, response); + if let Some(update) = update { + let _ = events.projected_live.send(update); } - let _ = events.projected_live.send(update); drop(events); if let Some(inner) = weak_inner.upgrade() { sync_structured_activity(&inner, id).await; @@ -2101,6 +2071,51 @@ fn start_acp_permission_pump( }); } +/// 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. +/// Returns `None` when nothing should be broadcast live: an unserializable +/// request, or a persistence failure. +fn record_permission_request( + events: &mut AcpEventStream, + request: maestro_acp::schema::RequestPermissionRequest, + response: oneshot::Sender, +) -> Option { + 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 None; + } + }; + 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 Some(update); + }; + let event = PersistedEvent::Approval { row: row.clone() }; + if let Err(error) = events.log.append(&event) { + tracing::error!(%error, "approval history append failed"); + events.projection.rollback_row(&row.id); + cancel_pending_approvals(events); + return None; + } + let seq = events.log.last_seq(); + let _ = events.live.send(LoggedEvent { seq, event }); + Some(update) +} + fn cancel_pending_approvals(events: &mut AcpEventStream) { for (_, pending) in events.pending_approvals.drain() { let _ = pending @@ -2653,13 +2668,148 @@ async fn cleanup_record( #[cfg(test)] mod tests { - use super::{permission_outcome, ApprovalDecision}; - use maestro_acp::schema::{PermissionOption, PermissionOptionKind, RequestPermissionOutcome}; + use std::fs::File; + + use maestro_acp::schema::{ + PermissionOption, PermissionOptionKind, RequestPermissionOutcome, RequestPermissionRequest, + ToolCallUpdate, ToolCallUpdateFields, + }; + + use super::{ + cancel_pending_approvals, permission_outcome, record_permission_request, AcpDispatchState, + AcpEventStream, ApprovalDecision, + }; + use crate::projection::Projection; + use crate::sidecar::{EventCursor, EventLog, PersistedEvent, SidecarStore}; 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(), + } + } + + /// A handle that can never append: opened read-only, so `write_all` + /// fails with the OS's own error instead of a hand-rolled one — the + /// deterministic fault injection #184 asks for. + 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_leaves_no_phantom_row_and_replay_agrees() { + 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 update = record_permission_request(&mut events, permission_request(), response_tx); + + assert!( + update.is_none(), + "a failed persist must not be broadcast live" + ); + 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(); + drop(store.open_events(id).unwrap()); + let mut events = stream(unwritable_log(&store.events_path_for(id), 0)); + + let (first_tx, first_rx) = tokio::sync::oneshot::channel(); + assert!(record_permission_request(&mut events, permission_request(), first_tx).is_none()); + assert_eq!(first_rx.await.unwrap(), RequestPermissionOutcome::Cancelled); + + // A real disk recovers; the next request over the same connection + // must not be poisoned by the earlier failure (concurrent/ordering + // audit for #184). + events.log = store.open_events(id).unwrap(); + let (second_tx, _second_rx) = tokio::sync::oneshot::channel(); + let update = record_permission_request(&mut events, permission_request(), second_tx); + + assert!(update.is_some(), "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"); + + let replayed = store + .read_events(id, EventCursor::Since(0), 100) + .expect("replay"); + assert_eq!(replayed.len(), 1); + 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 { match outcome { RequestPermissionOutcome::Selected(selected) => Some(selected.option_id.to_string()), diff --git a/crates/maestro-daemon/src/sidecar.rs b/crates/maestro-daemon/src/sidecar.rs index e9df838..40edfba 100644 --- a/crates/maestro-daemon/src/sidecar.rs +++ b/crates/maestro-daemon/src/sidecar.rs @@ -305,6 +305,15 @@ pub struct EventLog { } impl EventLog { + /// Wrap an arbitrary file handle as an [`EventLog`], for tests that need + /// to inject a deterministic append failure (#184) — e.g. a handle + /// opened read-only, so `write_all` fails with the OS's own "bad file + /// descriptor" error instead of relying on a flaky filesystem fault. + #[cfg(test)] + pub(crate) fn for_test(file: File, last_seq: u64) -> Self { + Self { file, last_seq } + } + /// The seq of the last line, `0` for an empty log. pub fn last_seq(&self) -> u64 { self.last_seq diff --git a/docs/STRUCTURED-VIEW.md b/docs/STRUCTURED-VIEW.md index e207079..81bf52d 100644 --- a/docs/STRUCTURED-VIEW.md +++ b/docs/STRUCTURED-VIEW.md @@ -314,9 +314,11 @@ 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, - without marking it dead.) `mark_session_dead` applies the same - natural-exit worktree rule + narrower: it cancels the pending approval, rolls the unpersisted + `Approval` row back out of the live transcript, and lets the session + continue, without marking it dead — a snapshot taken after the failure + must never show a row durable replay could not reproduce (#184).) + `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