naps62/maestro
Archived
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(maestro): roll back phantom approval row on log append failure

Browse Source 
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.
This commit is contained in:
Miguel Palhas
2026-09-05 04:34:47 +01:00
parent 51f96cfb98
commit c067f48a24
4 changed files with 270 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/maestro-daemon/src/session.rs
+185 -35
View File
@@ -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<maestro_acp::schema::RequestPermissionOutcome>,
) -> Option<ProjectionUpdate> {
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<String> {
match outcome {
RequestPermissionOutcome::Selected(selected) => Some(selected.option_id.to_string()),