diff --git a/crates/maestro-daemon/src/projection.rs b/crates/maestro-daemon/src/projection.rs index 6b984e1..30a0460 100644 --- a/crates/maestro-daemon/src/projection.rs +++ b/crates/maestro-daemon/src/projection.rs @@ -173,10 +173,17 @@ impl Projection { } } - pub fn restore_approval(&mut self, row: TranscriptRow) { - if !self.row_index.contains_key(&row.id) { + pub fn restore_approval(&mut self, row: TranscriptRow) -> ProjectionUpdate { + let transcript = if !self.row_index.contains_key(&row.id) { self.row_index.insert(row.id.clone(), self.rows.len()); - self.rows.push(row); + self.rows.push(row.clone()); + Some(TranscriptDelta::Append { row }) + } else { + None + }; + ProjectionUpdate { + transcript, + state: self.state(), } } @@ -699,4 +706,24 @@ mod tests { assert!(projection.state().pending_approval.is_none()); assert_eq!(projection.snapshot(0).rows, vec![row]); } + + #[test] + fn recovered_approval_is_historical_not_pending() { + let mut live = Projection::default(); + let TranscriptDelta::Append { row } = live + .request_approval("daemon-nonce".into(), serde_json::json!({"tool":"bash"})) + .transcript + .unwrap() + else { + panic!("approval row") + }; + let mut recovered = Projection::default(); + let update = recovered.restore_approval(row.clone()); + assert_eq!( + update.transcript, + Some(TranscriptDelta::Append { row: row.clone() }) + ); + assert!(update.state.pending_approval.is_none()); + assert_eq!(recovered.snapshot(0).rows, vec![row]); + } } diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 9dfdb0a..900707f 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -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, 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(¬ification) { + let event = PersistedEvent::Notification(Box::new(notification)); + match events.log.append(&event) { Ok(seq) => { - let projected = events.projection.apply(seq, ¬ification); - 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, events: Arc>) { let Some(mut requests) = acp.take_permission_requests() else { return; @@ -1811,9 +1813,14 @@ fn start_acp_permission_pump(acp: &Arc, events: Arc 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 { + 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 + ); + } +} diff --git a/crates/maestro-daemon/src/sidecar.rs b/crates/maestro-daemon/src/sidecar.rs index 05aef06..e9df838 100644 --- a/crates/maestro-daemon/src/sidecar.rs +++ b/crates/maestro-daemon/src/sidecar.rs @@ -7,8 +7,8 @@ //! advertised but with empty metadata (handled by the caller, not here). //! //! A structured session (#135) also owns `.events.jsonl` next to its -//! sidecar: every `session/update` notification its adapter sent, one JSON -//! object per line, appended before it is broadcast. The log is the source of +//! sidecar: every `session/update` notification and daemon approval row is one +//! JSON object, appended before it is broadcast. The log is the source of //! truth for the transcript; `seq` is the line's 1-based position, so a //! client that holds seq `n` asks for [`EventCursor::Since`]`(n)`. A line //! that does not parse — including the partial line a hard kill leaves at @@ -62,38 +62,6 @@ impl SidecarStore { self.dir.join(format!("{id}.recover")) } - fn approvals_path_for(&self, id: Uuid) -> PathBuf { - self.dir.join(format!("{id}.approvals.jsonl")) - } - - /// Approval rows are a daemon projection, not ACP notifications. Keep - /// only their display payload so a restart retains history but never - /// resurrects the daemon-local pending nonce. - pub fn append_approval(&self, id: Uuid, row: &TranscriptRow) -> Result<(), SidecarError> { - std::fs::create_dir_all(&self.dir)?; - let mut file = OpenOptions::new() - .create(true) - .append(true) - .open(self.approvals_path_for(id))?; - serde_json::to_writer(&mut file, row)?; - file.write_all(b"\n")?; - file.flush()?; - Ok(()) - } - - pub fn read_approvals(&self, id: Uuid) -> Result, SidecarError> { - let file = match File::open(self.approvals_path_for(id)) { - Ok(file) => file, - Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), - Err(error) => return Err(error.into()), - }; - Ok(BufReader::new(file) - .lines() - .map_while(Result::ok) - .filter_map(|line| serde_json::from_str(&line).ok()) - .collect()) - } - /// Mark `id` as eligible for one structured-session recovery attempt. pub fn mark_for_recovery(&self, id: Uuid) -> Result<(), SidecarError> { std::fs::create_dir_all(&self.dir)?; @@ -182,10 +150,10 @@ impl SidecarStore { } continue; } - let Ok(notification) = serde_json::from_slice::(&line) else { + let Ok(event) = PersistedEvent::decode(&line) else { continue; }; - out.push_back(LoggedEvent { seq, notification }); + out.push_back(LoggedEvent { seq, event }); match cursor { EventCursor::Since(_) if out.len() >= limit => break, EventCursor::Before(_) if out.len() > limit => { @@ -222,7 +190,7 @@ impl SidecarStore { /// whichever does not exist. pub fn delete(&self, id: Uuid) -> Result<(), SidecarError> { remove_if_present(&self.events_path_for(id))?; - remove_if_present(&self.approvals_path_for(id))?; + remove_if_present(&self.dir.join(format!("{id}.approvals.jsonl")))?; remove_if_present(&self.recovery_path_for(id))?; remove_if_present(&self.path_for(id)) } @@ -280,11 +248,52 @@ pub enum EventCursor { } /// One line of a session's event log. -#[derive(Debug, Clone, PartialEq, serde::Serialize)] +#[derive(Debug, Clone, PartialEq)] pub struct LoggedEvent { /// 1-based line position in the log; the cursor for [`EventCursor`]. pub seq: u64, - pub notification: SessionNotification, + pub event: PersistedEvent, +} + +/// A single entry in the structured event history. +/// +/// ACP notifications retain their original JSON representation, so existing +/// event logs remain readable. Daemon-generated rows use an envelope so they +/// can share the same cursor sequence without pretending to be ACP traffic. +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(tag = "maestroEvent", rename_all = "snake_case")] +pub enum PersistedEvent { + Approval { + row: TranscriptRow, + }, + #[serde(skip)] + Notification(Box), +} + +impl PersistedEvent { + fn decode(line: &[u8]) -> Result { + serde_json::from_slice::(line) + .map(|notification| Self::Notification(Box::new(notification))) + .or_else(|_| serde_json::from_slice(line)) + } +} + +impl From for PersistedEvent { + fn from(notification: SessionNotification) -> Self { + Self::Notification(Box::new(notification)) + } +} + +impl From<&SessionNotification> for PersistedEvent { + fn from(notification: &SessionNotification) -> Self { + Self::Notification(Box::new(notification.clone())) + } +} + +impl From<&PersistedEvent> for PersistedEvent { + fn from(event: &PersistedEvent) -> Self { + event.clone() + } } /// An open, append-only handle on one session's event log. Obtained from @@ -305,8 +314,12 @@ 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, notification: &SessionNotification) -> Result { - let mut line = serde_json::to_vec(notification)?; + pub fn append(&mut self, event: impl Into) -> Result { + let event = event.into(); + let mut line = match &event { + PersistedEvent::Notification(notification) => serde_json::to_vec(notification)?, + _ => serde_json::to_vec(&event)?, + }; line.push(b'\n'); self.file.write_all(&line)?; self.file.flush()?; diff --git a/crates/maestro-daemon/tests/acp.rs b/crates/maestro-daemon/tests/acp.rs index f707eb4..2f71888 100644 --- a/crates/maestro-daemon/tests/acp.rs +++ b/crates/maestro-daemon/tests/acp.rs @@ -83,7 +83,10 @@ impl Fake { } fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str { - match &event.notification.update { + let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else { + panic!("unexpected daemon event") + }; + match ¬ification.update { SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content { ContentBlock::Text(text) => &text.text, other => panic!("unexpected content {other:?}"), diff --git a/crates/maestro-daemon/tests/sidecar.rs b/crates/maestro-daemon/tests/sidecar.rs index 336d86a..e5f1612 100644 --- a/crates/maestro-daemon/tests/sidecar.rs +++ b/crates/maestro-daemon/tests/sidecar.rs @@ -6,8 +6,8 @@ use std::collections::HashMap; use std::path::PathBuf; use chrono::{DateTime, Utc}; -use maestro_daemon::sidecar::SidecarStore; -use maestro_protocol::{Session, SessionStatus, SessionView}; +use maestro_daemon::sidecar::{PersistedEvent, SidecarStore}; +use maestro_protocol::{Session, SessionStatus, SessionView, TranscriptRow, TranscriptRowKind}; use serde_json::Value; use tempfile::TempDir; use uuid::Uuid; @@ -186,6 +186,13 @@ fn text_of(n: &SessionNotification) -> &str { } } +fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str { + let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else { + panic!("expected ACP notification") + }; + text_of(notification) +} + #[test] fn append_assigns_line_position_as_seq_and_reads_back() { let dir = TempDir::new().unwrap(); @@ -194,9 +201,9 @@ fn append_assigns_line_position_as_seq_and_reads_back() { let mut log = store.open_events(id).expect("open"); assert_eq!(log.last_seq(), 0); - assert_eq!(log.append(¬e("a")).unwrap(), 1); - assert_eq!(log.append(¬e("b")).unwrap(), 2); - assert_eq!(log.append(¬e("c")).unwrap(), 3); + assert_eq!(log.append(note("a")).unwrap(), 1); + assert_eq!(log.append(note("b")).unwrap(), 2); + assert_eq!(log.append(note("c")).unwrap(), 3); assert_eq!(log.last_seq(), 3); let path = store.events_path_for(id); @@ -211,9 +218,7 @@ fn append_assigns_line_position_as_seq_and_reads_back() { let all = store.read_events(id, EventCursor::Since(0), 100).unwrap(); assert_eq!(all.iter().map(|e| e.seq).collect::>(), vec![1, 2, 3]); assert_eq!( - all.iter() - .map(|e| text_of(&e.notification)) - .collect::>(), + all.iter().map(event_text).collect::>(), vec!["a", "b", "c"] ); } @@ -225,7 +230,7 @@ fn since_reads_forward_and_before_reads_backward_with_limit() { let id = Uuid::new_v4(); let mut log = store.open_events(id).unwrap(); for i in 1..=6 { - log.append(¬e(&format!("n{i}"))).unwrap(); + log.append(note(&format!("n{i}"))).unwrap(); } // `since` is exclusive: what a client that already holds seq 2 needs. @@ -278,7 +283,7 @@ fn malformed_lines_are_skipped_but_keep_their_seq() { let store = SidecarStore::new(dir.path()); let id = Uuid::new_v4(); let mut log = store.open_events(id).unwrap(); - log.append(¬e("first")).unwrap(); + log.append(note("first")).unwrap(); drop(log); // Garbage in the middle and a line that is JSON but not a notification. @@ -289,11 +294,11 @@ fn malformed_lines_are_skipped_but_keep_their_seq() { std::fs::write(&path, raw).unwrap(); let mut log = store.open_events(id).unwrap(); assert_eq!(log.last_seq(), 3); - assert_eq!(log.append(¬e("fourth")).unwrap(), 4); + assert_eq!(log.append(note("fourth")).unwrap(), 4); let all = store.read_events(id, EventCursor::Since(0), 10).unwrap(); assert_eq!(all.iter().map(|e| e.seq).collect::>(), vec![1, 4]); - assert_eq!(text_of(&all[1].notification), "fourth"); + assert_eq!(event_text(&all[1]), "fourth"); } #[test] @@ -303,7 +308,7 @@ fn truncated_last_line_still_opens_and_the_next_append_starts_a_fresh_line() { let store = SidecarStore::new(dir.path()); let id = Uuid::new_v4(); let mut log = store.open_events(id).unwrap(); - log.append(¬e("whole")).unwrap(); + log.append(note("whole")).unwrap(); drop(log); let path = store.events_path_for(id); @@ -319,10 +324,10 @@ fn truncated_last_line_still_opens_and_the_next_append_starts_a_fresh_line() { // glue onto it; the partial keeps a seq of its own. let mut log = store.open_events(id).unwrap(); assert_eq!(log.last_seq(), 2); - assert_eq!(log.append(¬e("after")).unwrap(), 3); + assert_eq!(log.append(note("after")).unwrap(), 3); let all = store.read_events(id, EventCursor::Since(0), 10).unwrap(); assert_eq!(all.iter().map(|e| e.seq).collect::>(), vec![1, 3]); - assert_eq!(text_of(&all[1].notification), "after"); + assert_eq!(event_text(&all[1]), "after"); let raw = std::fs::read_to_string(&path).unwrap(); assert_eq!(raw.lines().count(), 3, "{raw}"); } @@ -334,8 +339,10 @@ fn delete_removes_the_event_log_and_load_all_ignores_it() { let id = Uuid::new_v4(); store.write(&sample(id)).unwrap(); let mut log = store.open_events(id).unwrap(); - log.append(¬e("x")).unwrap(); + log.append(note("x")).unwrap(); drop(log); + let legacy_approvals = dir.path().join(format!("{id}.approvals.jsonl")); + std::fs::write(&legacy_approvals, b"obsolete").unwrap(); assert!(store.events_path_for(id).exists()); let loaded = store.load_all().unwrap(); @@ -344,6 +351,47 @@ fn delete_removes_the_event_log_and_load_all_ignores_it() { store.delete(id).unwrap(); assert!(!store.path_for(id).exists()); assert!(!store.events_path_for(id).exists()); + assert!(!legacy_approvals.exists()); // Idempotent with the log already gone. store.delete(id).unwrap(); } + +#[test] +fn approval_rows_share_event_cursors_with_legacy_acp_lines() { + let dir = TempDir::new().unwrap(); + let store = SidecarStore::new(dir.path()); + let id = Uuid::new_v4(); + let legacy = serde_json::to_vec(¬e("before")).unwrap(); + std::fs::create_dir_all(dir.path()).unwrap(); + std::fs::write(store.events_path_for(id), [legacy, b"\n".to_vec()].concat()).unwrap(); + + let mut log = store.open_events(id).unwrap(); + assert_eq!( + log.append(PersistedEvent::Approval { + row: TranscriptRow { + id: "approval:secret".into(), + kind: TranscriptRowKind::Approval, + data: serde_json::json!({"tool": "bash"}), + }, + }) + .unwrap(), + 2 + ); + assert_eq!(log.append(note("after")).unwrap(), 3); + + let all = store.read_events(id, EventCursor::Since(0), 10).unwrap(); + assert_eq!( + all.iter().map(|event| event.seq).collect::>(), + [1, 2, 3] + ); + assert!(matches!(all[1].event, PersistedEvent::Approval { .. })); + assert_eq!(event_text(&all[0]), "before"); + assert_eq!(event_text(&all[2]), "after"); + + let since = store.read_events(id, EventCursor::Since(1), 1).unwrap(); + assert_eq!(since[0].seq, 2); + let before = store + .read_events(id, EventCursor::Before(Some(3)), 1) + .unwrap(); + assert_eq!(before[0].seq, 2); +}