feat(maestro): add the per-session event log to SidecarStore (#135)

<sidecar_dir>/<id>.events.jsonl holds one session/update notification
per line; seq is the line's 1-based position. open_events returns an
append handle held for the session's lifetime, and terminates a
partial last line first so a hard kill mid-write never glues the next
event onto the fragment. read_events takes a Since (forward, exclusive)
or Before (backward, exclusive, None for the tail) cursor with a limit;
unparseable lines keep their seq and are skipped. delete removes the
log with the sidecar; load_all ignores .jsonl.

Each append is a single write_all to the kernel, no fsync: a killed
daemon loses nothing already written, and an fsync per streamed chunk
is the wrong trade for a log that is not meant to survive power loss.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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 `<id>.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<EventLog, SidecarError> {
+        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<Vec<LoggedEvent>, 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<LoggedEvent> = 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::<SessionNotification>(&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 `<dir>/<id>.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<u64>),
+}
+
+/// 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<u64, SidecarError> {
+        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)
+    }
+}

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(&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);
+    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<_>>(), vec![1, 2, 3]);
+    assert_eq!(
+        all.iter()
+            .map(|e| text_of(&e.notification))
+            .collect::<Vec<_>>(),
+        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(&note(&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<_>>(), 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<_>>(), 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<_>>(), 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<_>>(), 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(&note("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(&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<_>>(), 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(&note("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<_>>(), 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(&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<_>>(), 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(&note("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();
+}