diff --git a/crates/maestro-daemon/src/sidecar.rs b/crates/maestro-daemon/src/sidecar.rs index 709d877..acd2b74 100644 --- a/crates/maestro-daemon/src/sidecar.rs +++ b/crates/maestro-daemon/src/sidecar.rs @@ -5,9 +5,22 @@ //! original `cwd`, command, env, metadata, etc. Files are deleted on //! `KillSession`. Sessions whose sidecar is missing at recovery time are still //! 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 +//! 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 +//! the end — keeps its seq and is skipped on read. `delete` removes the log +//! with the sidecar; `load_all` never reads one as a sidecar. +use std::collections::VecDeque; +use std::fs::{File, OpenOptions}; +use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; +use maestro_acp::schema::SessionNotification; use maestro_protocol::Session; use uuid::Uuid; @@ -38,6 +51,98 @@ impl SidecarStore { self.dir.join(format!("{id}.json")) } + /// Where the per-session event log lives (#135). + pub fn events_path_for(&self, id: Uuid) -> PathBuf { + self.dir.join(format!("{id}.events.jsonl")) + } + + /// Open the event log for `id` for appending, creating it (and the + /// directory) when missing. One handle per live session: the caller + /// keeps it for the session's lifetime rather than reopening per + /// notification. + /// + /// A log whose last line has no newline — what a hard kill mid-write + /// leaves behind — is terminated first, so the next append starts a + /// fresh line instead of gluing onto the fragment. The fragment keeps + /// the seq of its line and reads as malformed. + pub fn open_events(&self, id: Uuid) -> Result { + std::fs::create_dir_all(&self.dir)?; + let mut file = OpenOptions::new() + .create(true) + .read(true) + .append(true) + .open(self.events_path_for(id))?; + let mut lines = 0u64; + let mut last = None; + let mut reader = BufReader::new(&mut file); + let mut buf = [0u8; 64 * 1024]; + loop { + let n = reader.read(&mut buf)?; + if n == 0 { + break; + } + lines += buf[..n].iter().filter(|&&b| b == b'\n').count() as u64; + last = Some(buf[n - 1]); + } + if last.is_some_and(|b| b != b'\n') { + file.write_all(b"\n")?; + lines += 1; + } + file.seek(SeekFrom::End(0))?; + Ok(EventLog { + file, + last_seq: lines, + }) + } + + /// Read events from the log for `id`, at most `limit`, in ascending seq + /// order. A missing log is an empty log. Lines that fail to parse are + /// skipped but still consume a seq, so cursors stay stable. + pub fn read_events( + &self, + id: Uuid, + cursor: EventCursor, + limit: usize, + ) -> Result, SidecarError> { + let file = match File::open(self.events_path_for(id)) { + Ok(f) => f, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(e) => return Err(e.into()), + }; + if limit == 0 { + return Ok(Vec::new()); + } + let mut out: VecDeque = VecDeque::new(); + let mut seq = 0u64; + for line in BufReader::new(file).split(b'\n') { + let line = line?; + seq += 1; + let keep = match cursor { + EventCursor::Since(since) => seq > since, + EventCursor::Before(Some(before)) => seq < before, + EventCursor::Before(None) => true, + }; + if !keep { + if matches!(cursor, EventCursor::Before(Some(_))) { + break; + } + continue; + } + let Ok(notification) = serde_json::from_slice::(&line) else { + continue; + }; + out.push_back(LoggedEvent { seq, notification }); + match cursor { + EventCursor::Since(_) if out.len() >= limit => break, + EventCursor::Before(_) if out.len() > limit => { + out.pop_front(); + } + _ => {} + } + } + Ok(out.into()) + } + /// Atomically write `session` to `/.json`. Creates the directory /// if it does not exist. pub fn write(&self, session: &Session) -> Result<(), SidecarError> { @@ -59,13 +164,11 @@ impl SidecarStore { } } - /// Delete the sidecar for `id`. No-op if it does not exist. + /// Delete the sidecar and the event log for `id`. No-op for whichever + /// does not exist. pub fn delete(&self, id: Uuid) -> Result<(), SidecarError> { - match std::fs::remove_file(self.path_for(id)) { - Ok(()) => Ok(()), - Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), - Err(e) => Err(e.into()), - } + remove_if_present(&self.events_path_for(id))?; + remove_if_present(&self.path_for(id)) } /// Read every sidecar JSON in the directory. Files that fail to parse are @@ -81,6 +184,8 @@ impl SidecarStore { for entry in entries { let entry = entry?; let path = entry.path(); + // `.events.jsonl` logs (#135) sit in the same directory and are + // never sidecars. if path.extension().and_then(|s| s.to_str()) != Some("json") { continue; } @@ -95,3 +200,61 @@ impl SidecarStore { Ok(out) } } + +fn remove_if_present(path: &Path) -> Result<(), SidecarError> { + match std::fs::remove_file(path) { + Ok(()) => Ok(()), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(e) => Err(e.into()), + } +} + +/// Which slice of a session's event log to read. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EventCursor { + /// Forward: events with `seq` strictly greater than the cursor, the + /// first `limit` of them. `Since(0)` is the whole log from the start. + /// What a client catching up after a reconnect asks for. + Since(u64), + /// Backward: events with `seq` strictly less than the cursor, the last + /// `limit` of them, still returned in ascending order. `None` means the + /// tail of the log, for a first paint. The next page passes the smallest + /// seq it received. + Before(Option), +} + +/// One line of a session's event log. +#[derive(Debug, Clone, PartialEq)] +pub struct LoggedEvent { + /// 1-based line position in the log; the cursor for [`EventCursor`]. + pub seq: u64, + pub notification: SessionNotification, +} + +/// 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, + last_seq: u64, +} + +impl EventLog { + /// The seq of the last line, `0` for an empty log. + pub fn last_seq(&self) -> u64 { + self.last_seq + } + + /// Append one notification as a line and return its seq. The line is + /// 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)?; + line.push(b'\n'); + self.file.write_all(&line)?; + self.file.flush()?; + self.last_seq += 1; + Ok(self.last_seq) + } +} diff --git a/crates/maestro-daemon/tests/sidecar.rs b/crates/maestro-daemon/tests/sidecar.rs index f942e2d..8130c6d 100644 --- a/crates/maestro-daemon/tests/sidecar.rs +++ b/crates/maestro-daemon/tests/sidecar.rs @@ -146,3 +146,189 @@ fn write_creates_parent_directory() { assert!(nested.exists()); assert!(store.path_for(s.id).exists()); } + +// ---- per-session event log (#135) ---- + +use maestro_acp::schema::{ContentBlock, SessionNotification, SessionUpdate, TextContent}; +use maestro_daemon::sidecar::EventCursor; + +fn note(text: &str) -> SessionNotification { + SessionNotification::new( + "acp-1", + SessionUpdate::AgentMessageChunk(maestro_acp::schema::ContentChunk::new( + ContentBlock::Text(TextContent::new(text)), + )), + ) +} + +fn text_of(n: &SessionNotification) -> &str { + match &n.update { + SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content { + ContentBlock::Text(t) => &t.text, + other => panic!("unexpected content {other:?}"), + }, + other => panic!("unexpected update {other:?}"), + } +} + +#[test] +fn append_assigns_line_position_as_seq_and_reads_back() { + let dir = TempDir::new().unwrap(); + let store = SidecarStore::new(dir.path()); + let id = Uuid::new_v4(); + + 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.last_seq(), 3); + + let path = store.events_path_for(id); + assert!(path.ends_with(format!("{id}.events.jsonl")), "{path:?}"); + let raw = std::fs::read_to_string(&path).unwrap(); + assert_eq!(raw.lines().count(), 3, "one object per line:\n{raw}"); + for line in raw.lines() { + let v: Value = serde_json::from_str(line).expect("each line is one JSON object"); + assert_eq!(v["sessionId"], "acp-1"); + } + + 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::>(), + vec!["a", "b", "c"] + ); +} + +#[test] +fn since_reads_forward_and_before_reads_backward_with_limit() { + let dir = TempDir::new().unwrap(); + let store = SidecarStore::new(dir.path()); + 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(); + } + + // `since` is exclusive: what a client that already holds seq 2 needs. + let fwd = store.read_events(id, EventCursor::Since(2), 3).unwrap(); + assert_eq!(fwd.iter().map(|e| e.seq).collect::>(), vec![3, 4, 5]); + // Past the end: nothing, not an error. + assert!(store + .read_events(id, EventCursor::Since(6), 3) + .unwrap() + .is_empty()); + + // `before` is exclusive and pages backward: the `limit` events closest + // to the cursor, still in ascending seq order. + let back = store + .read_events(id, EventCursor::Before(Some(5)), 2) + .unwrap(); + assert_eq!(back.iter().map(|e| e.seq).collect::>(), vec![3, 4]); + // No cursor: the tail of the log, for a first paint. + let tail = store.read_events(id, EventCursor::Before(None), 2).unwrap(); + assert_eq!(tail.iter().map(|e| e.seq).collect::>(), vec![5, 6]); + // Paging past the start runs out cleanly. + let start = store + .read_events(id, EventCursor::Before(Some(2)), 10) + .unwrap(); + assert_eq!(start.iter().map(|e| e.seq).collect::>(), vec![1]); + assert!(store + .read_events(id, EventCursor::Before(Some(1)), 10) + .unwrap() + .is_empty()); +} + +#[test] +fn missing_log_reads_as_empty() { + let dir = TempDir::new().unwrap(); + let store = SidecarStore::new(dir.path()); + let id = Uuid::new_v4(); + assert!(store + .read_events(id, EventCursor::Since(0), 10) + .unwrap() + .is_empty()); + assert!(store + .read_events(id, EventCursor::Before(None), 10) + .unwrap() + .is_empty()); +} + +#[test] +fn malformed_lines_are_skipped_but_keep_their_seq() { + let dir = TempDir::new().unwrap(); + let store = SidecarStore::new(dir.path()); + let id = Uuid::new_v4(); + let mut log = store.open_events(id).unwrap(); + log.append(¬e("first")).unwrap(); + drop(log); + + // Garbage in the middle and a line that is JSON but not a notification. + let path = store.events_path_for(id); + let mut raw = std::fs::read_to_string(&path).unwrap(); + raw.push_str("{not json\n"); + raw.push_str("{\"unrelated\": true}\n"); + 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); + + 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"); +} + +#[test] +fn truncated_last_line_still_opens_and_the_next_append_starts_a_fresh_line() { + // A hard kill mid-write leaves a partial line with no newline. + let dir = TempDir::new().unwrap(); + let store = SidecarStore::new(dir.path()); + let id = Uuid::new_v4(); + let mut log = store.open_events(id).unwrap(); + log.append(¬e("whole")).unwrap(); + drop(log); + + let path = store.events_path_for(id); + let mut raw = std::fs::read(&path).unwrap(); + raw.extend_from_slice(br#"{"sessionId":"acp-1","update":{"sessionUpd"#); + std::fs::write(&path, &raw).unwrap(); + + // Reads: the partial line is skipped, nothing else is lost. + let all = store.read_events(id, EventCursor::Since(0), 10).unwrap(); + assert_eq!(all.iter().map(|e| e.seq).collect::>(), vec![1]); + + // Reopening terminates the partial line so the next append does not + // 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); + 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"); + let raw = std::fs::read_to_string(&path).unwrap(); + assert_eq!(raw.lines().count(), 3, "{raw}"); +} + +#[test] +fn delete_removes_the_event_log_and_load_all_ignores_it() { + let dir = TempDir::new().unwrap(); + let store = SidecarStore::new(dir.path()); + let id = Uuid::new_v4(); + store.write(&sample(id)).unwrap(); + let mut log = store.open_events(id).unwrap(); + log.append(¬e("x")).unwrap(); + drop(log); + assert!(store.events_path_for(id).exists()); + + let loaded = store.load_all().unwrap(); + assert_eq!(loaded.len(), 1, "the .jsonl must not be read as a sidecar"); + + store.delete(id).unwrap(); + assert!(!store.path_for(id).exists()); + assert!(!store.events_path_for(id).exists()); + // Idempotent with the log already gone. + store.delete(id).unwrap(); +}