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

fix(daemon): unify approval replay log

Browse Source
This commit is contained in:
Miguel Palhas
2026-09-05 02:44:59 +01:00
parent d9815d86c6
commit 8773671afd
5 changed files with 251 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/maestro-daemon/src/session.rs
+99 -21
View File
@@ -24,7 +24,9 @@ use uuid::Uuid;
use crate::panes::Panes;
use crate::projection::{Projection, ProjectionUpdate, TurnLifecycle};
use crate::pty::{pipe_pane_command, PtyBridge};
use crate::sidecar::{EventCursor, EventLog, LoggedEvent, SidecarError, SidecarStore};
use crate::sidecar::{
EventCursor, EventLog, LoggedEvent, PersistedEvent, SidecarError, SidecarStore,
};
use crate::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord};
use crate::tmux::TmuxError;
use crate::viewport::{ViewportId, Viewports};
@@ -115,8 +117,6 @@ impl SessionRecord {
#[derive(Debug)]
struct AcpEventStream {
id: Uuid,
sidecar: SidecarStore,
log: EventLog,
live: broadcast::Sender<LoggedEvent>,
projection: Projection,
@@ -424,7 +424,7 @@ impl SessionManager {
let mut projection = Projection::default();
let mut replay = Vec::new();
for event in events {
let update = projection.apply(event.seq, &event.notification);
let update = apply_persisted_event(&mut projection, &event);
let include = match cursor {
EventCursor::Since(since) => event.seq > since,
EventCursor::Before(Some(before)) => event.seq < before,
@@ -889,14 +889,9 @@ impl SessionManager {
let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
let mut projection = Projection::default();
for event in &persisted {
projection.apply(event.seq, &event.notification);
}
for row in self.inner.sidecar.read_approvals(id)? {
projection.restore_approval(row);
apply_persisted_event(&mut projection, event);
}
Some(Arc::new(Mutex::new(AcpEventStream {
id,
sidecar: self.inner.sidecar.clone(),
log,
live: tx,
projection,
@@ -1338,14 +1333,9 @@ impl SessionManager {
let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
let mut projection = Projection::default();
for event in &persisted {
projection.apply(event.seq, &event.notification);
}
for row in self.inner.sidecar.read_approvals(id)? {
projection.restore_approval(row);
apply_persisted_event(&mut projection, event);
}
Some(Arc::new(Mutex::new(AcpEventStream {
id,
sidecar: self.inner.sidecar.clone(),
log,
live: tx,
projection,
@@ -1757,10 +1747,15 @@ fn start_acp_event_pump(
while let Some(notification) = updates.recv().await {
let append_error = {
let mut events = events.lock().await;
match events.log.append(&notification) {
let event = PersistedEvent::Notification(Box::new(notification));
match events.log.append(&event) {
Ok(seq) => {
let projected = events.projection.apply(seq, &notification);
let _ = events.live.send(LoggedEvent { seq, notification });
let logged = LoggedEvent {
seq,
event: event.clone(),
};
let projected = apply_persisted_event(&mut events.projection, &logged);
let _ = events.live.send(logged);
let _ = events.projected_live.send(projected);
None
}
@@ -1784,6 +1779,13 @@ fn start_acp_event_pump(
done_rx
}
fn apply_persisted_event(projection: &mut Projection, event: &LoggedEvent) -> ProjectionUpdate {
match &event.event {
PersistedEvent::Notification(notification) => projection.apply(event.seq, notification),
PersistedEvent::Approval { row } => projection.restore_approval(row.clone()),
}
}
fn start_acp_permission_pump(acp: &Arc<AcpSession>, events: Arc<Mutex<AcpEventStream>>) {
let Some(mut requests) = acp.take_permission_requests() else {
return;
@@ -1811,9 +1813,14 @@ fn start_acp_permission_pump(acp: &Arc<AcpSession>, events: Arc<Mutex<AcpEventSt
);
let update = events.projection.request_approval(nonce, data);
if let Some(TranscriptDelta::Append { row }) = &update.transcript {
if let Err(error) = events.sidecar.append_approval(events.id, row) {
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);
}
@@ -1841,7 +1848,23 @@ fn permission_outcome(
let kind = match decision {
ApprovalDecision::AllowOnce => PermissionOptionKind::AllowOnce,
ApprovalDecision::AllowForSession => PermissionOptionKind::AllowAlways,
ApprovalDecision::Deny => PermissionOptionKind::RejectOnce,
ApprovalDecision::Deny => {
// Prefer a one-shot refusal; use a permanent refusal only when
// that is the sole rejection option the ACP request offered.
return options
.iter()
.find(|option| option.kind == PermissionOptionKind::RejectOnce)
.or_else(|| {
options
.iter()
.find(|option| option.kind == PermissionOptionKind::RejectAlways)
})
.map_or(RequestPermissionOutcome::Cancelled, |option| {
RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new(
option.option_id.clone(),
))
});
}
};
options.iter().find(|option| option.kind == kind).map_or(
RequestPermissionOutcome::Cancelled,
@@ -2329,3 +2352,58 @@ async fn cleanup_record(
}
let _ = inner.events.send(CatalogEvent::SessionRemoved { id });
}
#[cfg(test)]
mod tests {
use super::{permission_outcome, ApprovalDecision};
use maestro_acp::schema::{PermissionOption, PermissionOptionKind, RequestPermissionOutcome};
fn option(id: &str, kind: PermissionOptionKind) -> PermissionOption {
PermissionOption::new(id.to_owned(), id, kind)
}
fn selected_id(outcome: RequestPermissionOutcome) -> Option<String> {
match outcome {
RequestPermissionOutcome::Selected(selected) => Some(selected.option_id.to_string()),
RequestPermissionOutcome::Cancelled => None,
_ => None,
}
}
#[test]
fn permission_decisions_select_only_offered_options() {
let all = [
option("allow-once", PermissionOptionKind::AllowOnce),
option("allow-always", PermissionOptionKind::AllowAlways),
option("reject-once", PermissionOptionKind::RejectOnce),
option("reject-always", PermissionOptionKind::RejectAlways),
];
assert_eq!(
selected_id(permission_outcome(ApprovalDecision::AllowOnce, &all)).as_deref(),
Some("allow-once")
);
assert_eq!(
selected_id(permission_outcome(ApprovalDecision::AllowForSession, &all)).as_deref(),
Some("allow-always")
);
assert_eq!(
selected_id(permission_outcome(ApprovalDecision::Deny, &all)).as_deref(),
Some("reject-once")
);
assert_eq!(
selected_id(permission_outcome(
ApprovalDecision::Deny,
&[option("reject-always", PermissionOptionKind::RejectAlways)],
))
.as_deref(),
Some("reject-always")
);
assert_eq!(
selected_id(permission_outcome(
ApprovalDecision::Deny,
&[option("allow-once", PermissionOptionKind::AllowOnce)],
)),
None
);
}
}