fix(maestro): close ACP replay seams

Serialize replay reads with event appends, acquire the live receiver under that lock, and drain the event pump during shutdown. Recovery now removes its FIFO when opening the log fails.

crates/maestro-acp/src/lib.rs

@@ -19,6 +19,12 @@
 //! is dropped while nobody reads: #135 takes the receiver and persists each
 //! notification to the per-session event log before broadcasting it.
 //!
+//! A respawn after a daemon restart sets [`SpawnConfig::load_session`] and
+//! the handshake runs `session/load` with the stored id instead of
+//! `session/new`. The history the agent streams back while that request is
+//! in flight is *not* delivered: the daemon's event log already holds it,
+//! and replaying it would double every line.
+//!
 //! The child is a child of the daemon, full stop. It shares the daemon's
 //! process group, is marked to receive `SIGTERM` when the daemon dies
 //! (`PR_SET_PDEATHSIG` on Linux), and [`AcpSession::terminate`] is what both
@@ -31,13 +37,14 @@ use std::collections::{HashMap, VecDeque};
 use std::ffi::OsString;
 use std::path::{Path, PathBuf};
 use std::process::{ExitStatus, Stdio};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use agent_client_protocol::schema::v1::{
     CancelNotification, ClientCapabilities, FileSystemCapabilities, Implementation,
-    InitializeRequest, NewSessionRequest, RequestPermissionOutcome, RequestPermissionRequest,
-    RequestPermissionResponse, SessionId, SessionNotification,
+    InitializeRequest, LoadSessionRequest, NewSessionRequest, RequestPermissionOutcome,
+    RequestPermissionRequest, RequestPermissionResponse, SessionId, SessionNotification,
 };
 use agent_client_protocol::schema::ProtocolVersion;
 use agent_client_protocol::{ByteStreams, Client};
@@ -82,6 +89,10 @@ pub struct SpawnConfig {
     pub env: HashMap<String, String>,
     /// Upper bound on the handshake; [`HANDSHAKE_TIMEOUT`] by default.
     pub handshake_timeout: Duration,
+    /// An existing ACP session id to resume with `session/load` (#135)
+    /// instead of opening a fresh one. The adapter must advertise
+    /// `loadSession`; the handshake fails otherwise.
+    pub load_session: Option<String>,
 }
 
 impl SpawnConfig {
@@ -90,8 +101,15 @@ impl SpawnConfig {
             cwd: cwd.into(),
             env,
             handshake_timeout: HANDSHAKE_TIMEOUT,
+            load_session: None,
         }
     }
+
+    /// Resume `session_id` with `session/load` instead of `session/new`.
+    pub fn load_session(mut self, session_id: impl Into<String>) -> Self {
+        self.load_session = Some(session_id.into());
+        self
+    }
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -296,6 +314,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
     tokio::spawn(run_connection(
         transport,
         config.cwd.clone(),
+        config.load_session.clone(),
         updates_tx,
         commands_rx,
         ready_tx,
@@ -375,17 +394,28 @@ fn client_capabilities() -> ClientCapabilities {
 async fn run_connection<T>(
     transport: T,
     cwd: PathBuf,
+    load_session: Option<String>,
     updates_tx: mpsc::UnboundedSender<SessionNotification>,
     mut commands_rx: mpsc::UnboundedReceiver<Command>,
     ready_tx: oneshot::Sender<Result<String, ReadyError>>,
 ) where
     T: agent_client_protocol::ConnectTo<Client> + 'static,
 {
+    // Set while `session/load` is in flight. Notification handlers run in
+    // the dispatch loop, in arrival order, so everything the agent streams
+    // before its load response sees the flag up and is dropped: that is the
+    // history the event log already holds (#135).
+    let replaying = Arc::new(AtomicBool::new(false));
+    let replaying_seen = replaying.clone();
     let result = Client
         .builder()
         .name("maestro")
         .on_receive_notification(
             async move |notification: SessionNotification, _cx| {
+                if replaying_seen.load(Ordering::SeqCst) {
+                    tracing::trace!("dropping replayed notification during session/load");
+                    return Ok(());
+                }
                 // A dropped receiver only means nobody wants updates any more.
                 let _ = updates_tx.send(notification);
                 Ok(())
@@ -410,22 +440,64 @@ async fn run_connection<T>(
             let init = InitializeRequest::new(ProtocolVersion::V1)
                 .client_capabilities(client_capabilities())
                 .client_info(Implementation::new("maestro", env!("CARGO_PKG_VERSION")));
-            if let Err(e) = cx.send_request(init).block_task().await {
-                let _ = ready_tx.send(Err(ready_error(e)));
-                return Ok(());
-            }
-            let created = match cx
-                .send_request(NewSessionRequest::new(cwd))
-                .block_task()
-                .await
-            {
+            let initialized = match cx.send_request(init).block_task().await {
                 Ok(r) => r,
                 Err(e) => {
                     let _ = ready_tx.send(Err(ready_error(e)));
                     return Ok(());
                 }
             };
-            let session_id = created.session_id.0.to_string();
+            let session_id = match load_session {
+                Some(existing) => {
+                    if !initialized.agent_capabilities.load_session {
+                        let _ = ready_tx.send(Err(ReadyError::Handshake(format!(
+                            "{ADAPTER_BIN} does not advertise loadSession; cannot resume session {existing}"
+                        ))));
+                        return Ok(());
+                    }
+                    replaying.store(true, Ordering::SeqCst);
+                    let (loaded_tx, loaded_rx) = oneshot::channel();
+                    let replaying_done = replaying.clone();
+                    if let Err(e) = cx
+                        .send_request(LoadSessionRequest::new(existing.clone(), cwd))
+                        .on_receiving_result(move |result| {
+                            replaying_done.store(false, Ordering::SeqCst);
+                            async move {
+                                let _ = loaded_tx.send(result);
+                                Ok(())
+                            }
+                        })
+                    {
+                        let _ = ready_tx.send(Err(ready_error(e)));
+                        return Ok(());
+                    }
+                    match loaded_rx.await {
+                        Ok(Ok(_)) => {}
+                        Ok(Err(e)) => {
+                            let _ = ready_tx.send(Err(ready_error(e)));
+                            return Ok(());
+                        }
+                        Err(_) => {
+                            let _ = ready_tx.send(Err(ReadyError::Handshake(
+                                "session/load response was not delivered".into(),
+                            )));
+                            return Ok(());
+                        }
+                    }
+                    existing
+                }
+                None => match cx
+                    .send_request(NewSessionRequest::new(cwd))
+                    .block_task()
+                    .await
+                {
+                    Ok(created) => created.session_id.0.to_string(),
+                    Err(e) => {
+                        let _ = ready_tx.send(Err(ready_error(e)));
+                        return Ok(());
+                    }
+                },
+            };
             let _ = ready_tx.send(Ok(session_id.clone()));
 
             loop {

crates/maestro-acp/tests/fake_acp_agent.py

@@ -20,6 +20,16 @@ Environment knobs, all optional:
   FAKE_ACP_EXIT_AFTER_NEW=1  exit 0 right after answering `session/new`
   FAKE_ACP_EXIT_ON_CANCEL=1  exit 0 on `session/cancel` (default: keep running
                              so the SIGTERM path is exercised)
+  FAKE_ACP_EMIT_ON_CANCEL=1  emit a final notification before handling cancel
+  FAKE_ACP_NO_LOAD=1      advertise `loadSession: false`
+  FAKE_ACP_LOAD_REPLAY=n  on `session/load`, replay n `agent_message_chunk`
+                            notifications ("replayed <i>") before answering,
+                            the way a real agent streams history back
+  FAKE_ACP_LOAD_FAIL=1    answer `session/load` with a JSON-RPC error
+
+Signals: SIGUSR1 emits one `agent_message_chunk` notification ("tick <n>",
+counting from 1) and logs `emit tick <n>`, so a test can make the agent
+speak on demand (#135).
 """
 import json
 import os
@@ -49,13 +59,32 @@ def on_sigterm(_signum, _frame):
 
 signal.signal(signal.SIGTERM, on_sigterm)
 
+session_id = f"fake-{os.getpid()}"
+ticks = 0
+
+
+def chunk(text):
+    return {"jsonrpc": "2.0", "method": "session/update", "params": {
+        "sessionId": session_id,
+        "update": {"sessionUpdate": "agent_message_chunk",
+                   "content": {"type": "text", "text": text}},
+    }}
+
+
+def on_sigusr1(_signum, _frame):
+    global ticks
+    ticks += 1
+    send(chunk(f"tick {ticks}"))
+    log(f"emit tick {ticks}")
+
+
+signal.signal(signal.SIGUSR1, on_sigusr1)
+
 log(f"start pid={os.getpid()} cwd={os.getcwd()}")
 echo = os.environ.get("FAKE_ACP_ECHO_ENV")
 if echo:
     log(f"env {echo}={os.environ.get(echo, '')}")
 
-session_id = f"fake-{os.getpid()}"
-
 for raw in sys.stdin:
     raw = raw.strip()
     if not raw:
@@ -77,20 +106,30 @@ for raw in sys.stdin:
             continue
         send({"jsonrpc": "2.0", "id": rid, "result": {
             "protocolVersion": params.get("protocolVersion", 1),
-            "agentCapabilities": {"loadSession": True},
+            "agentCapabilities": {
+                "loadSession": os.environ.get("FAKE_ACP_NO_LOAD") != "1"},
             "agentInfo": {"name": "fake-acp-agent", "version": "0.0.0"},
             "authMethods": [],
         }})
     elif method == "session/new":
         send({"jsonrpc": "2.0", "id": rid, "result": {"sessionId": session_id}})
-        send({"jsonrpc": "2.0", "method": "session/update", "params": {
-            "sessionId": session_id,
-            "update": {"sessionUpdate": "agent_message_chunk",
-                       "content": {"type": "text", "text": "hello from fake"}},
-        }})
+        send(chunk("hello from fake"))
         if os.environ.get("FAKE_ACP_EXIT_AFTER_NEW") == "1":
             sys.exit(0)
+    elif method == "session/load":
+        session_id = params.get("sessionId")
+        log(f"recv session/load sessionId={session_id}")
+        if os.environ.get("FAKE_ACP_LOAD_FAIL") == "1":
+            send({"jsonrpc": "2.0", "id": rid,
+                  "error": {"code": -32000, "message": "fake load failure"}})
+            continue
+        for i in range(int(os.environ.get("FAKE_ACP_LOAD_REPLAY", "0"))):
+            send(chunk(f"replayed {i + 1}"))
+        send({"jsonrpc": "2.0", "id": rid, "result": {}})
+        send(chunk("hello after load"))
     elif method == "session/cancel":
+        if os.environ.get("FAKE_ACP_EMIT_ON_CANCEL") == "1":
+            send(chunk("final on cancel"))
         if os.environ.get("FAKE_ACP_EXIT_ON_CANCEL") == "1":
             sys.exit(0)
     elif rid is not None:

crates/maestro-acp/tests/spawn.rs

@@ -274,6 +274,76 @@ async fn child_exit_before_handshake_reports_exited_early() {
     }
 }
 
+fn chunk_text(n: &maestro_acp::schema::SessionNotification) -> String {
+    match &n.update {
+        SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
+            ContentBlock::Text(t) => t.text.clone(),
+            other => panic!("unexpected content {other:?}"),
+        },
+        other => panic!("unexpected update {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn load_session_resumes_the_stored_id_and_drops_the_replay() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path()).load_session("fake-old");
+    config.env.insert("FAKE_ACP_LOAD_REPLAY".into(), "3".into());
+
+    let session = maestro_acp::spawn(config).await.expect("spawn");
+    assert_eq!(session.session_id(), "fake-old");
+    let log = fake.log();
+    assert!(
+        log.contains("recv session/load sessionId=fake-old"),
+        "{log}"
+    );
+    assert!(!log.contains("recv session/new"), "{log}");
+
+    // The three replayed chunks never reach the receiver: the event log
+    // already holds them. The first notification after the response does.
+    let mut updates = session.take_updates().expect("updates");
+    let first = tokio::time::timeout(Duration::from_secs(5), updates.recv())
+        .await
+        .expect("update within 5s")
+        .expect("channel open");
+    assert_eq!(chunk_text(&first), "hello after load");
+    // Nothing queued behind it either.
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    assert!(updates.try_recv().is_err(), "replay leaked through");
+
+    session
+        .terminate(Duration::from_millis(100), Duration::from_secs(2))
+        .await;
+}
+
+#[tokio::test]
+async fn load_session_requires_the_load_capability() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path()).load_session("fake-old");
+    config.env.insert("FAKE_ACP_NO_LOAD".into(), "1".into());
+
+    let err = maestro_acp::spawn(config)
+        .await
+        .expect_err("no loadSession");
+    assert!(matches!(err, AcpError::Handshake(_)), "{err}");
+    assert!(err.to_string().contains("loadSession"), "{err}");
+    assert!(!fake.log().contains("recv session/load"), "{}", fake.log());
+}
+
+#[tokio::test]
+async fn load_session_error_fails_the_spawn() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path()).load_session("fake-old");
+    config.env.insert("FAKE_ACP_LOAD_FAIL".into(), "1".into());
+
+    let err = maestro_acp::spawn(config).await.expect_err("load fails");
+    assert!(matches!(err, AcpError::Handshake(_)), "{err}");
+    assert!(err.to_string().contains("fake load failure"), "{err}");
+}
+
 /// End to end against the installed `claude-code-acp`. Needs Claude auth,
 /// takes seconds, so it is opt-in: `cargo test -p maestro-acp -- --ignored`.
 #[tokio::test]

crates/maestro-daemon/src/session.rs

@@ -17,12 +17,12 @@ use maestro_protocol::{
     ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session,
     SessionStatus, SessionView, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
 };
-use tokio::sync::{broadcast, Mutex};
+use tokio::sync::{broadcast, watch, Mutex};
 use uuid::Uuid;
 
 use crate::panes::Panes;
 use crate::pty::{pipe_pane_command, PtyBridge};
-use crate::sidecar::{SidecarError, SidecarStore};
+use crate::sidecar::{EventCursor, EventLog, LoggedEvent, SidecarError, SidecarStore};
 use crate::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord};
 use crate::tmux::TmuxError;
 use crate::viewport::{ViewportId, Viewports};
@@ -94,6 +94,18 @@ pub struct SessionRecord {
     /// adapter never came up. Shared so a kill or shutdown can terminate it
     /// outside the sessions lock. Daemon-local; never rides the wire.
     pub acp: Option<Arc<AcpSession>>,
+    /// Open log plus its post-append live sender. Their shared lock is the
+    /// replay/live seam used by the structured WebSocket in #138.
+    acp_events: Option<Arc<Mutex<AcpEventStream>>>,
+    /// Completion signal for the task draining ACP notifications into the
+    /// event log. Shutdown waits for it before returning.
+    acp_event_pump: Option<watch::Receiver<bool>>,
+}
+
+#[derive(Debug)]
+struct AcpEventStream {
+    log: EventLog,
+    live: broadcast::Sender<LoggedEvent>,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -294,18 +306,41 @@ impl SessionManager {
         sessions.values().map(|r| r.session.clone()).collect()
     }
 
-    /// The raw `session/update` stream of a structured session (#134),
-    /// handed out once; `None` for a terminal session, a session whose
-    /// adapter never came up, or a stream already taken. Notifications that
-    /// arrived before the take are still queued. #135 takes it to feed the
-    /// per-session event log.
-    pub async fn take_acp_updates(
+    /// Read replay and acquire its live continuation under the same lock.
+    /// Events are therefore present exactly once across the returned seam.
+    pub async fn subscribe_acp_events_with_replay(
         &self,
         id: Uuid,
-    ) -> Option<tokio::sync::mpsc::UnboundedReceiver<maestro_acp::schema::SessionNotification>>
-    {
-        let sessions = self.inner.sessions.lock().await;
-        sessions.get(&id)?.acp.as_ref()?.take_updates()
+        cursor: EventCursor,
+        limit: usize,
+    ) -> Result<Option<(Vec<LoggedEvent>, broadcast::Receiver<LoggedEvent>)>, SessionError> {
+        let stream = {
+            let sessions = self.inner.sessions.lock().await;
+            let Some(record) = sessions.get(&id) else {
+                return Ok(None);
+            };
+            let Some(stream) = &record.acp_events else {
+                return Ok(None);
+            };
+            stream.clone()
+        };
+        let stream = stream.lock().await;
+        let live = stream.live.subscribe();
+        let replay = self.inner.sidecar.read_events(id, cursor, limit)?;
+        Ok(Some((replay, live)))
+    }
+
+    /// Read a cursor-bounded slice of the durable structured-session log.
+    pub fn read_acp_events(
+        &self,
+        id: Uuid,
+        cursor: EventCursor,
+        limit: usize,
+    ) -> Result<Vec<LoggedEvent>, SessionError> {
+        self.inner
+            .sidecar
+            .read_events(id, cursor, limit)
+            .map_err(SessionError::Sidecar)
     }
 
     /// End every adapter child before the daemon exits (#134): each gets
@@ -314,12 +349,15 @@ impl SessionManager {
     /// is on disk after the restart is truthful. tmux sessions are left
     /// alone — surviving a restart is what they are for.
     pub async fn shutdown(&self) {
-        let children: Vec<(Uuid, Arc<AcpSession>)> = {
+        let children: Vec<(Uuid, Arc<AcpSession>, Option<watch::Receiver<bool>>)> = {
             let sessions = self.inner.sessions.lock().await;
             sessions
                 .iter()
-                .filter(|(_, r)| r.session.status == SessionStatus::Running)
-                .filter_map(|(id, r)| r.acp.clone().map(|acp| (*id, acp)))
+                .filter_map(|(id, r)| {
+                    r.acp
+                        .clone()
+                        .map(|acp| (*id, acp, r.acp_event_pump.clone()))
+                })
                 .collect()
         };
         if children.is_empty() {
@@ -328,9 +366,14 @@ impl SessionManager {
         tracing::info!(count = children.len(), "shutdown: ending adapter children");
         let terminations = children
             .iter()
-            .map(|(_, acp)| acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE));
+            .map(|(_, acp, _)| acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE));
         futures_util::future::join_all(terminations).await;
-        for (id, _) in children {
+        let drains = children
+            .iter()
+            .filter_map(|(_, _, done)| done.clone())
+            .map(wait_for_event_pump);
+        futures_util::future::join_all(drains).await;
+        for (id, _, _) in children {
             mark_session_dead(&self.inner, id).await;
         }
     }
@@ -518,6 +561,24 @@ impl SessionManager {
             }
         };
 
+        let acp_events = if let Some(acp) = &acp {
+            match self.inner.sidecar.open_events(id) {
+                Ok(log) => {
+                    let (tx, _) = broadcast::channel(EVENTS_CAPACITY);
+                    Some(Arc::new(Mutex::new(AcpEventStream { log, live: tx })))
+                }
+                Err(e) => {
+                    acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
+                        .await;
+                    let _ = std::fs::remove_file(&fifo_path);
+                    self.rollback_worktree(&worktree_base_repo, &worktree_path);
+                    return Err(SessionError::Sidecar(e));
+                }
+            }
+        } else {
+            None
+        };
+
         // Capture spawn-history inputs before `req` is partially moved into
         // `session`. `history_cwd` is the *chosen* working dir (the request
         // `cwd`), not a generated worktree checkout path — that's what the
@@ -580,10 +641,15 @@ impl SessionManager {
                 }
             }
             let _ = std::fs::remove_file(&fifo_path);
+            let _ = self.inner.sidecar.delete(id);
             self.rollback_worktree(&worktree_base_repo, &worktree_path);
             return Err(SessionError::Sidecar(e));
         }
 
+        let event_pump = acp
+            .as_ref()
+            .zip(acp_events.as_ref())
+            .map(|(acp, events)| (acp.clone(), events.clone()));
         {
             let mut sessions = self.inner.sessions.lock().await;
             sessions.insert(
@@ -601,9 +667,17 @@ impl SessionManager {
                     focused_since: None,
                     awaiting_suppressed: false,
                     acp,
+                    acp_events,
+                    acp_event_pump: None,
                 },
             );
         }
+        if let Some((acp, events)) = event_pump {
+            let done = start_acp_event_pump(&self.inner, id, &acp, events);
+            if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
+                record.acp_event_pump = Some(done);
+            }
+        }
         let _ = self.inner.events.send(CatalogEvent::SessionAdded {
             session: session.clone(),
         });
@@ -818,6 +892,8 @@ impl SessionManager {
                     // A structured session's adapter died with the previous
                     // daemon; only tmux-backed sessions can be adopted.
                     acp: None,
+                    acp_events: None,
+                    acp_event_pump: None,
                 },
             );
             let _ = self
@@ -826,6 +902,138 @@ impl SessionManager {
                 .send(CatalogEvent::SessionAdded { session });
             adopted += 1;
         }
+
+        // Structured sessions have no tmux process to discover. Their
+        // sidecars are the recovery inventory: when an ACP id exists, start a
+        // fresh adapter process and load that conversation into it.
+        for mut session in self
+            .inner
+            .sidecar
+            .load_all()?
+            .into_iter()
+            .filter(|session| session.view == SessionView::Structured)
+        {
+            if self.inner.sessions.lock().await.contains_key(&session.id) {
+                continue;
+            }
+
+            let id = session.id;
+            let mut session_env = sanitize_env(&session.env)?;
+            let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo"));
+            let bridge =
+                PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await?;
+            session_env.insert("MAESTRO_SESSION_ID".into(), id.to_string());
+            if let Some(url) = self
+                .inner
+                .daemon_url
+                .lock()
+                .expect("daemon_url mutex")
+                .clone()
+            {
+                session_env.insert("MAESTRO_URL".into(), url);
+            }
+
+            let (acp, failure) = match session.acp_session_id.clone() {
+                Some(acp_session_id) => {
+                    let config = SpawnConfig::new(
+                        &session.cwd,
+                        session_env
+                            .iter()
+                            .map(|(key, value)| (key.clone(), value.clone()))
+                            .collect(),
+                    )
+                    .load_session(acp_session_id);
+                    match maestro_acp::spawn(config).await {
+                        Ok(acp) => (Some(Arc::new(acp)), None),
+                        Err(error) => {
+                            tracing::warn!(%id, %error, "structured session: adapter failed to resume");
+                            (None, Some(error.to_string()))
+                        }
+                    }
+                }
+                None => (
+                    None,
+                    Some("structured session has no stored ACP session id".into()),
+                ),
+            };
+            session.status = if failure.is_some() {
+                SessionStatus::Dead
+            } else {
+                SessionStatus::Running
+            };
+            session.activity = if failure.is_some() {
+                ActivityState::Error
+            } else {
+                ActivityState::Active
+            };
+            session.activity_detail = failure;
+
+            let acp_events = if let Some(acp) = &acp {
+                match self.inner.sidecar.open_events(id) {
+                    Ok(log) => {
+                        let (tx, _) = broadcast::channel(EVENTS_CAPACITY);
+                        Some(Arc::new(Mutex::new(AcpEventStream { log, live: tx })))
+                    }
+                    Err(error) => {
+                        acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
+                            .await;
+                        let _ = std::fs::remove_file(&fifo_path);
+                        return Err(SessionError::Sidecar(error));
+                    }
+                }
+            } else {
+                None
+            };
+            if let Err(error) = self.inner.sidecar.write(&session) {
+                if let Some(acp) = &acp {
+                    acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
+                        .await;
+                }
+                let _ = std::fs::remove_file(&fifo_path);
+                return Err(SessionError::Sidecar(error));
+            }
+            let event_pump = acp
+                .as_ref()
+                .zip(acp_events.as_ref())
+                .map(|(acp, events)| (acp.clone(), events.clone()));
+
+            self.inner.sessions.lock().await.insert(
+                id,
+                SessionRecord {
+                    session: session.clone(),
+                    fifo_path,
+                    worktree_path: session
+                        .worktree
+                        .as_ref()
+                        .map(|worktree| worktree.path.clone()),
+                    worktree_base_repo: session
+                        .worktree
+                        .as_ref()
+                        .map(|worktree| worktree.base_repo.clone()),
+                    bridge,
+                    reported_activity: None,
+                    settled: false,
+                    #[cfg(feature = "haiku-autoname")]
+                    naming: false,
+                    focused_since: None,
+                    awaiting_suppressed: false,
+                    acp,
+                    acp_events,
+                    acp_event_pump: None,
+                },
+            );
+            if let Some((acp, events)) = event_pump {
+                let done = start_acp_event_pump(&self.inner, id, &acp, events);
+                if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
+                    record.acp_event_pump = Some(done);
+                }
+            }
+            let _ = self
+                .inner
+                .events
+                .send(CatalogEvent::SessionAdded { session });
+            adopted += 1;
+        }
         Ok(adopted)
     }
 
@@ -1151,6 +1359,55 @@ fn sanitize_env(
     Ok(out)
 }
 
+fn start_acp_event_pump(
+    inner: &Arc<Inner>,
+    id: Uuid,
+    acp: &Arc<AcpSession>,
+    events: Arc<Mutex<AcpEventStream>>,
+) -> watch::Receiver<bool> {
+    let (done_tx, done_rx) = watch::channel(false);
+    let Some(mut updates) = acp.take_updates() else {
+        tracing::error!(%id, "structured session notification stream was already taken");
+        let _ = done_tx.send(true);
+        return done_rx;
+    };
+    let weak_inner = Arc::downgrade(inner);
+    let weak_acp = Arc::downgrade(acp);
+    tokio::spawn(async move {
+        while let Some(notification) = updates.recv().await {
+            let append_error = {
+                let mut events = events.lock().await;
+                match events.log.append(&notification) {
+                    Ok(seq) => {
+                        let _ = events.live.send(LoggedEvent { seq, notification });
+                        None
+                    }
+                    Err(error) => Some(error),
+                }
+            };
+            if let Some(error) = append_error {
+                tracing::error!(%id, %error, "structured event log append failed");
+                if let Some(acp) = weak_acp.upgrade() {
+                    acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
+                        .await;
+                }
+                if let Some(inner) = weak_inner.upgrade() {
+                    mark_session_dead(&inner, id).await;
+                }
+                break;
+            }
+        }
+        let _ = done_tx.send(true);
+    });
+    done_rx
+}
+
+async fn wait_for_event_pump(mut done: watch::Receiver<bool>) {
+    if !*done.borrow() {
+        let _ = done.wait_for(|finished| *finished).await;
+    }
+}
+
 async fn tmux_blocking<T, F>(f: F) -> T
 where
     F: FnOnce() -> T + Send + 'static,
@@ -1542,6 +1799,9 @@ async fn cleanup_record(
         acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
             .await;
     }
+    if let Some(done) = record.acp_event_pump.clone() {
+        wait_for_event_pump(done).await;
+    }
     let _ = std::fs::remove_file(&record.fifo_path);
     if let Err(e) = inner.sidecar.delete(id) {
         tracing::warn!(error = ?e, %id, "cleanup: failed to delete sidecar");

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/acp.rs

@@ -6,10 +6,15 @@ mod common;
 
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use common::{wait_for_async, DaemonFixture};
+use maestro_acp::schema::{ContentBlock, SessionUpdate};
 use maestro_acp::{ADAPTER_BIN, INSTALL_COMMAND};
+use maestro_daemon::sidecar::{EventCursor, SidecarStore};
+use maestro_daemon::worktree::WorktreeManager;
+use maestro_daemon::SessionManager;
 use maestro_protocol::{ActivityState, CreateSessionRequest, SessionStatus, SessionView};
 use tempfile::TempDir;
 
@@ -61,13 +66,28 @@ impl Fake {
     fn pid(&self) -> u32 {
         self.log()
             .lines()
-            .find_map(|l| l.strip_prefix("start pid="))
-            .and_then(|rest| rest.split(' ').next())
-            .and_then(|p| p.parse().ok())
+            .filter_map(|line| {
+                line.strip_prefix("start pid=")?
+                    .split(' ')
+                    .next()?
+                    .parse()
+                    .ok()
+            })
+            .next_back()
             .expect("pid in log")
     }
 }
 
+fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
+    match &event.notification.update {
+        SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
+            ContentBlock::Text(text) => &text.text,
+            other => panic!("unexpected content {other:?}"),
+        },
+        other => panic!("unexpected update {other:?}"),
+    }
+}
+
 fn structured(cwd: &Path, env: HashMap<String, String>) -> CreateSessionRequest {
     CreateSessionRequest {
         view: SessionView::Structured,
@@ -140,20 +160,162 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
     .unwrap();
     assert_eq!(sidecar.acp_session_id.as_deref(), Some(acp_id.as_str()));
 
-    // The raw notification stream is reachable for #135, handed out once.
-    let mut updates = fx
+    wait_for_async(Duration::from_secs(5), || async {
+        fx.manager
+            .read_acp_events(session.id, EventCursor::Since(0), 10)
+            .is_ok_and(|events| !events.is_empty())
+    })
+    .await;
+
+    // Live delivery happens only after the notification is durable.
+    let (replay, mut updates) = fx
         .manager
-        .take_acp_updates(session.id)
+        .subscribe_acp_events_with_replay(session.id, EventCursor::Since(0), 10)
         .await
-        .expect("updates receiver");
-    assert!(fx.manager.take_acp_updates(session.id).await.is_none());
-    let first = tokio::time::timeout(Duration::from_secs(5), updates.recv())
+        .expect("replay read")
+        .expect("event subscription");
+    assert_eq!(replay.len(), 1);
+    assert!(updates.try_recv().is_err(), "replay duplicated onto live");
+    unsafe {
+        libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1);
+    }
+    let live = tokio::time::timeout(Duration::from_secs(5), updates.recv())
         .await
         .expect("an update within 5s")
-        .expect("channel open");
-    assert_eq!(first.session_id.0.as_ref(), acp_id);
+        .expect("broadcast open");
+    assert_eq!(event_text(&live), "tick 1");
+    let stored = fx
+        .manager
+        .read_acp_events(session.id, EventCursor::Since(live.seq - 1), 1)
+        .expect("read event logged before broadcast");
+    assert_eq!(stored, vec![live]);
 
     fx.manager.kill(session.id, false).await.expect("kill");
+    assert!(!fx
+        .sidecar_dir()
+        .join(format!("{}.events.jsonl", session.id))
+        .exists());
+}
+
+#[tokio::test]
+async fn recovery_loads_the_acp_session_without_duplicating_history() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_LOAD_REPLAY".into(), "2".into());
+
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .expect("create");
+    let acp_id = session.acp_session_id.clone().expect("acp id");
+    wait_for_async(Duration::from_secs(5), || async {
+        fx.manager
+            .read_acp_events(session.id, EventCursor::Since(0), 10)
+            .is_ok_and(|events| events.len() == 1)
+    })
+    .await;
+    unsafe {
+        libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1);
+    }
+    wait_for_async(Duration::from_secs(5), || async {
+        fx.manager
+            .read_acp_events(session.id, EventCursor::Since(0), 10)
+            .is_ok_and(|events| events.len() == 2)
+    })
+    .await;
+
+    let old_pid = fake.pid();
+    fx.manager.shutdown().await;
+    wait_for_exit(old_pid).await;
+
+    let revived = SessionManager::new(
+        Arc::new(fx.tmux.tmux.clone()),
+        SidecarStore::new(fx.sidecar_dir()),
+        WorktreeManager::new(None),
+        fx.fifo_dir(),
+    );
+    assert_eq!(revived.recover().await.expect("recover"), 1);
+    assert_eq!(
+        revived.status(session.id).await,
+        Some(SessionStatus::Running)
+    );
+    let log = fake.log();
+    assert!(
+        log.contains(&format!("recv session/load sessionId={acp_id}")),
+        "{log}"
+    );
+    wait_for_async(Duration::from_secs(5), || async {
+        revived
+            .read_acp_events(session.id, EventCursor::Since(0), 10)
+            .is_ok_and(|events| events.len() == 3)
+    })
+    .await;
+
+    let (replay, mut live) = revived
+        .subscribe_acp_events_with_replay(session.id, EventCursor::Since(0), 10)
+        .await
+        .expect("replay read")
+        .expect("recovered event subscription");
+    assert_eq!(replay.len(), 3);
+    assert!(live.try_recv().is_err(), "replay duplicated onto live");
+    let new_pid = fake.pid();
+    assert_ne!(new_pid, old_pid);
+    unsafe {
+        libc::kill(new_pid as libc::pid_t, libc::SIGUSR1);
+    }
+    let event = tokio::time::timeout(Duration::from_secs(5), live.recv())
+        .await
+        .expect("live event after recovery")
+        .expect("broadcast open");
+    assert_eq!(event.seq, 4);
+    assert_eq!(event_text(&event), "tick 1");
+
+    let stored = revived
+        .read_acp_events(session.id, EventCursor::Since(0), 10)
+        .expect("recovered log");
+    assert_eq!(stored.len(), 4, "session/load replay must not be appended");
+    assert_eq!(
+        stored.iter().map(event_text).collect::<Vec<_>>(),
+        ["hello from fake", "tick 1", "hello after load", "tick 1"]
+    );
+
+    revived.kill(session.id, false).await.expect("kill");
+}
+
+#[tokio::test]
+async fn failed_recovery_cleans_its_fifo() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), fake.env()))
+        .await
+        .expect("create");
+    let old_pid = fake.pid();
+    fx.manager.shutdown().await;
+    wait_for_exit(old_pid).await;
+
+    let events_path = fx
+        .sidecar_dir()
+        .join(format!("{}.events.jsonl", session.id));
+    std::fs::remove_file(&events_path).unwrap();
+    std::fs::create_dir(&events_path).unwrap();
+    let revived = SessionManager::new(
+        Arc::new(fx.tmux.tmux.clone()),
+        SidecarStore::new(fx.sidecar_dir()),
+        WorktreeManager::new(None),
+        fx.fifo_dir(),
+    );
+
+    assert!(revived.recover().await.is_err());
+    assert!(!fx.fifo_dir().join(format!("{}.fifo", session.id)).exists());
+    let failed_pid = fake.pid();
+    assert_ne!(failed_pid, old_pid);
+    wait_for_exit(failed_pid).await;
 }
 
 #[tokio::test]
@@ -264,6 +426,38 @@ async fn shutdown_reaps_every_adapter_child() {
     assert_eq!(sidecar.status, SessionStatus::Dead);
 }
 
+#[tokio::test]
+async fn shutdown_drains_final_notifications_to_the_log() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_EMIT_ON_CANCEL".into(), "1".into());
+
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .expect("create");
+    wait_for_async(Duration::from_secs(5), || async {
+        fx.manager
+            .read_acp_events(session.id, EventCursor::Since(0), 10)
+            .is_ok_and(|events| events.len() == 1)
+    })
+    .await;
+
+    fx.manager.shutdown().await;
+
+    let stored = fx
+        .manager
+        .read_acp_events(session.id, EventCursor::Since(0), 10)
+        .expect("event log after shutdown");
+    assert_eq!(
+        stored.iter().map(event_text).collect::<Vec<_>>(),
+        ["hello from fake", "final on cancel"]
+    );
+}
+
 #[tokio::test]
 async fn adapter_exit_marks_the_session_dead_and_keeps_it_listed() {
     let fx = DaemonFixture::new();

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();
+}