fix(maestro): serialize structured shutdown

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.

Cargo.lock

@@ -1558,6 +1558,7 @@ version = "0.1.0"
 dependencies = [
  "chrono",
  "futures-util",
+ "libc",
  "maestro-protocol",
  "maestro-testing",
  "maestro-tui",

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)]
@@ -138,6 +156,10 @@ enum Command {
     Close,
 }
 
+enum ChildCommand {
+    Signal(libc::c_int),
+}
+
 /// Why the handshake did not produce a session id.
 enum ReadyError {
     /// The adapter answered with an error.
@@ -162,6 +184,7 @@ pub struct AcpSession {
     session_id: String,
     pid: u32,
     commands: mpsc::UnboundedSender<Command>,
+    child_commands: mpsc::UnboundedSender<ChildCommand>,
     updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
     exit: watch::Receiver<Option<ExitStatus>>,
 }
@@ -203,9 +226,6 @@ impl AcpSession {
     /// End the child: `session/cancel`, wait `cancel_grace`, `SIGTERM`, wait
     /// `term_grace`, `SIGKILL`. Returns once the child has been reaped.
     pub async fn terminate(&self, cancel_grace: Duration, term_grace: Duration) {
-        if self.has_exited() {
-            return;
-        }
         self.cancel();
         if tokio::time::timeout(cancel_grace, self.wait())
             .await
@@ -227,16 +247,14 @@ impl AcpSession {
     }
 
     fn signal(&self, sig: libc::c_int) {
-        signal_pid(self.pid, sig);
+        let _ = self.child_commands.send(ChildCommand::Signal(sig));
     }
 }
 
 impl Drop for AcpSession {
     fn drop(&mut self) {
+        self.signal(libc::SIGTERM);
         let _ = self.commands.send(Command::Close);
-        if !self.has_exited() {
-            self.signal(libc::SIGTERM);
-        }
     }
 }
 
@@ -283,11 +301,8 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
     tokio::spawn(drain_stderr(stderr, pid, stderr_tail.clone()));
 
     let (exit_tx, exit_rx) = watch::channel(None);
-    tokio::spawn(async move {
-        let status = child.wait().await.ok();
-        tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
-        let _ = exit_tx.send(status);
-    });
+    let (child_commands_tx, child_commands_rx) = mpsc::unbounded_channel();
+    tokio::spawn(supervise_child(child, pid, child_commands_rx, exit_tx));
 
     let (updates_tx, updates_rx) = mpsc::unbounded_channel();
     let (commands_tx, commands_rx) = mpsc::unbounded_channel();
@@ -296,6 +311,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,
@@ -308,6 +324,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
                 session_id,
                 pid,
                 commands: commands_tx,
+                child_commands: child_commands_tx,
                 updates: Mutex::new(Some(updates_rx)),
                 exit: exit_rx,
             });
@@ -337,12 +354,12 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
     };
     // No half-started adapter survives a failed spawn.
     let _ = commands_tx.send(Command::Close);
-    signal_pid(pid, libc::SIGTERM);
+    let _ = child_commands_tx.send(ChildCommand::Signal(libc::SIGTERM));
     if tokio::time::timeout(TERM_GRACE, wait_exit(&exit_rx))
         .await
         .is_err()
     {
-        signal_pid(pid, libc::SIGKILL);
+        let _ = child_commands_tx.send(ChildCommand::Signal(libc::SIGKILL));
         wait_exit(&exit_rx).await;
     }
     Err(failure)
@@ -354,11 +371,39 @@ async fn wait_exit(exit: &watch::Receiver<Option<ExitStatus>>) -> Option<ExitSta
     *status
 }
 
-fn signal_pid(pid: u32, sig: libc::c_int) {
-    // ESRCH just means the child is already gone.
-    // SAFETY: plain syscall on a pid we spawned and have not reaped.
-    unsafe {
-        libc::kill(pid as libc::pid_t, sig);
+async fn supervise_child(
+    mut child: tokio::process::Child,
+    pid: u32,
+    mut commands: mpsc::UnboundedReceiver<ChildCommand>,
+    exit_tx: watch::Sender<Option<ExitStatus>>,
+) {
+    loop {
+        tokio::select! {
+            status = child.wait() => {
+                let status = status.ok();
+                tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
+                let _ = exit_tx.send(status);
+                return;
+            }
+            command = commands.recv() => match command {
+                Some(ChildCommand::Signal(sig)) => {
+                    // This task alone owns `child`. It cannot reap the child
+                    // between receiving this command and sending the signal,
+                    // so its PID cannot be reused in that window.
+                    unsafe { libc::kill(pid as libc::pid_t, sig) };
+                }
+                None => {
+                    // The session vanished without a command receiver. This
+                    // task still owns the unreaped child, so this cannot hit
+                    // a reused PID.
+                    unsafe { libc::kill(pid as libc::pid_t, libc::SIGTERM) };
+                    let status = child.wait().await.ok();
+                    tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
+                    let _ = exit_tx.send(status);
+                    return;
+                }
+            },
+        }
     }
 }
 
@@ -375,17 +420,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 +466,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

@@ -17,14 +17,28 @@ Environment knobs, all optional:
                             signal SIGTERM
   FAKE_ACP_ECHO_ENV       name of one env var to log at start
   FAKE_ACP_INIT_FAIL=1    answer `initialize` with a JSON-RPC error
+  FAKE_ACP_DELAY_SESSION_NEW_MS=n  wait n milliseconds before answering
+                             `session/new`
   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_IGNORE_EOF=1      keep running after stdin closes, until signalled
+  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
 import signal
 import sys
+import time
 
 LOG = os.environ.get("FAKE_ACP_LOG")
 
@@ -49,13 +63,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,22 +110,37 @@ 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":
+        time.sleep(int(os.environ.get("FAKE_ACP_DELAY_SESSION_NEW_MS", "0")) / 1000)
         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:
         send({"jsonrpc": "2.0", "id": rid,
               "error": {"code": -32601, "message": f"method not found: {method}"}})
+
+if os.environ.get("FAKE_ACP_IGNORE_EOF") == "1":
+    while True:
+        signal.pause()

crates/maestro-acp/tests/spawn.rs

@@ -69,6 +69,24 @@ impl Fake {
         }
         panic!("log never contained {needle:?}:\n{}", self.log());
     }
+
+    fn pid(&self) -> u32 {
+        self.log()
+            .lines()
+            .find_map(|line| line.strip_prefix("start pid="))
+            .and_then(|rest| rest.split(' ').next())
+            .and_then(|pid| pid.parse().ok())
+            .expect("pid in log")
+    }
+
+    async fn wait_for_exit(&self) {
+        let pid = self.pid();
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while pid_alive(pid) && Instant::now() < deadline {
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+        assert!(!pid_alive(pid), "child {pid} still alive");
+    }
 }
 
 fn pid_alive(pid: u32) -> bool {
@@ -274,6 +292,90 @@ async fn child_exit_before_handshake_reports_exited_early() {
     }
 }
 
+#[tokio::test]
+async fn drop_terminates_and_reaps_the_child() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path());
+    config.env.insert("FAKE_ACP_IGNORE_EOF".into(), "1".into());
+    let session = maestro_acp::spawn(config).await.expect("spawn");
+
+    drop(session);
+
+    fake.wait_for_log("signal SIGTERM").await;
+    fake.wait_for_exit().await;
+}
+
+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/cli.rs

@@ -122,21 +122,23 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
     // shutdown, and both must reach the adapter children below (#134). The
     // unit's `KillMode=process` signals only the daemon, so nothing else
     // would end them.
-    let shutdown = async {
+    let shutdown_manager = manager.clone();
+    let shutdown = async move {
         let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
             .expect("install SIGTERM handler");
         tokio::select! {
             _ = tokio::signal::ctrl_c() => tracing::info!("ctrl-c received — shutting down"),
             _ = sigterm.recv() => tracing::info!("SIGTERM received — shutting down"),
         }
+        // Axum waits for upgraded WebSockets to drain. Structured children
+        // must be settled before that wait, while the signal path still owns
+        // the shutdown order (#179).
+        shutdown_manager.shutdown().await;
     };
 
     axum::serve(listener, router)
         .tcp_nodelay(true)
         .with_graceful_shutdown(shutdown)
         .await?;
-    // tmux sessions outlive the daemon by design; structured sessions'
-    // adapter children do not (#134).
-    manager.shutdown().await;
     Ok(())
 }

crates/maestro-daemon/src/local_api.rs

@@ -170,6 +170,7 @@ fn session_error_status(code: &str) -> StatusCode {
         // The session outlived its pane: nothing to type into, and nothing
         // the daemon can do about it.
         "not_running" => StatusCode::CONFLICT,
+        "shutting_down" => StatusCode::SERVICE_UNAVAILABLE,
         _ => StatusCode::BAD_GATEWAY,
     }
 }

crates/maestro-daemon/src/session.rs

@@ -7,7 +7,7 @@
 
 use std::collections::{HashMap, HashSet};
 use std::path::PathBuf;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -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)]
@@ -115,6 +127,8 @@ pub enum SessionError {
     /// a stray keystroke on an attach socket, which is dropped.
     #[error("session is no longer running: {0}")]
     NotRunning(Uuid),
+    #[error("daemon is shutting down")]
+    ShuttingDown,
     /// The kill was refused (#79): the session's worktree holds unsaved work.
     /// Nothing happened — the session keeps running and the tree is untouched,
     /// so it stays the operator's handle on their own unfinished work.
@@ -132,6 +146,7 @@ impl SessionError {
             SessionError::NotFound(_) => "not_found",
             SessionError::InvalidEnv(_) => "invalid_env",
             SessionError::NotRunning(_) => "not_running",
+            SessionError::ShuttingDown => "shutting_down",
             SessionError::KillGuarded(_) => "kill_guarded",
         };
         let unsaved_work = match self {
@@ -183,6 +198,12 @@ struct Inner {
     /// client that reconnects never collides with the registration its
     /// previous socket has not finished releasing.
     next_viewport: AtomicU64,
+    /// Once daemon shutdown starts, it owns structured-session finalization.
+    /// Background passes must not classify those deliberate exits as natural.
+    shutting_down: AtomicBool,
+    /// Serializes structured creation with shutdown so every adapter child is
+    /// either rejected or included in shutdown's final drain.
+    structured_lifecycle: Mutex<()>,
 }
 
 impl SessionManager {
@@ -225,6 +246,8 @@ impl SessionManager {
                 events,
                 viewports: Mutex::new(Viewports::new()),
                 next_viewport: AtomicU64::new(0),
+                shutting_down: AtomicBool::new(false),
+                structured_lifecycle: Mutex::new(()),
             }),
         }
     }
@@ -294,18 +317,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 +360,20 @@ 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>)> = {
+        if self.inner.shutting_down.swap(true, Ordering::AcqRel) {
+            return;
+        }
+        let _lifecycle = self.inner.structured_lifecycle.lock().await;
+        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.session.status == SessionStatus::Running)
+                        .then_some(())
+                        .and_then(|()| r.acp.clone())
+                        .map(|acp| (*id, acp, r.acp_event_pump.clone()))
+                })
                 .collect()
         };
         if children.is_empty() {
@@ -328,10 +382,15 @@ 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 {
-            mark_session_dead(&self.inner, id).await;
+        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_after_shutdown(&self.inner, id).await;
         }
     }
 
@@ -414,6 +473,17 @@ impl SessionManager {
 
     /// Spawn a new session. On success returns the freshly-created `Session`.
     pub async fn create(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
+        if req.view == SessionView::Structured {
+            let _lifecycle = self.inner.structured_lifecycle.lock().await;
+            if self.inner.shutting_down.load(Ordering::Acquire) {
+                return Err(SessionError::ShuttingDown);
+            }
+            return self.create_inner(req).await;
+        }
+        self.create_inner(req).await
+    }
+
+    async fn create_inner(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
         let (effective_cwd, worktree_path, worktree_base_repo) = match &req.worktree {
             Some(spec) => {
                 let WorktreeSpec {
@@ -518,6 +588,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 +668,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 +694,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 +919,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 +929,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 +1386,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,
@@ -1181,6 +1465,9 @@ async fn probe_worktree(
 /// One pass of the liveness watcher: diff the in-memory session set against
 /// the tmux-reported one, and reap anything we still track but tmux has lost.
 async fn run_liveness_pass(inner: &Arc<Inner>) {
+    if inner.shutting_down.load(Ordering::Acquire) {
+        return;
+    }
     // Snapshot the in-memory ids first; if we own no sessions there is
     // nothing to reap and no reason to spawn a tmux subprocess (this also
     // avoids racing with the very first `new-session` that would otherwise
@@ -1240,6 +1527,9 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
 /// `SessionUpdated` only when `activity` or `foreground` actually changes — so a
 /// steady state produces no diffs. Runs on the existing liveness tick.
 async fn run_activity_pass(inner: &Arc<Inner>) {
+    if inner.shutting_down.load(Ordering::Acquire) {
+        return;
+    }
     let ids: Vec<Uuid> = {
         let sessions = inner.sessions.lock().await;
         if sessions.is_empty() {
@@ -1282,6 +1572,9 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
 
         let updated = {
             let mut sessions = inner.sessions.lock().await;
+            if inner.shutting_down.load(Ordering::Acquire) {
+                return;
+            }
             let Some(rec) = sessions.get_mut(&id) else {
                 continue;
             };
@@ -1356,6 +1649,9 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
 /// error must not read as "nothing blocks". Like the activity pass, this only
 /// emits `SessionUpdated` when the value actually changed.
 async fn run_blocked_work_pass(inner: &Arc<Inner>) {
+    if inner.shutting_down.load(Ordering::Acquire) {
+        return;
+    }
     let targets: Vec<(Uuid, (PathBuf, PathBuf))> = {
         let sessions = inner.sessions.lock().await;
         sessions
@@ -1383,6 +1679,9 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
         };
         let updated = {
             let mut sessions = inner.sessions.lock().await;
+            if inner.shutting_down.load(Ordering::Acquire) {
+                return;
+            }
             let Some(rec) = sessions.get_mut(&id) else {
                 continue;
             };
@@ -1399,7 +1698,7 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
     }
 }
 
-/// Mark a session whose tmux session has disappeared as [`SessionStatus::Dead`]
+/// Mark a session whose backing process has disappeared as [`SessionStatus::Dead`]
 /// and settle its worktree per the natural-exit rule (#82).
 ///
 /// The record, its [`PtyBridge`] and its sidecar all survive, so a client can
@@ -1411,9 +1710,27 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
 /// alone and stays attached to the dead record for a later forced dismissal
 /// (#81), which leaves it on disk too.
 ///
-/// Dead sessions do not survive a daemon restart: [`SessionManager::recover`]
-/// adopts only what tmux still holds.
+/// Terminal sessions that die naturally do not survive a daemon restart:
+/// [`SessionManager::recover`] adopts only what tmux still holds. Structured
+/// sessions stopped for daemon shutdown retain their sidecars for ACP recovery.
 async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
+    if inner.shutting_down.load(Ordering::Acquire) {
+        return;
+    }
+    mark_session_dead_with_worktree_settlement(inner, id, true).await;
+}
+
+/// A daemon restart terminates ACP children, but it is not a natural adapter
+/// exit: the sidecar is the recovery inventory, including its worktree.
+async fn mark_session_dead_after_shutdown(inner: &Arc<Inner>, id: Uuid) {
+    mark_session_dead_with_worktree_settlement(inner, id, false).await;
+}
+
+async fn mark_session_dead_with_worktree_settlement(
+    inner: &Arc<Inner>,
+    id: Uuid,
+    settle_worktree: bool,
+) {
     let flipped = {
         let mut sessions = inner.sessions.lock().await;
         match sessions.get_mut(&id) {
@@ -1437,17 +1754,19 @@ async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
     };
     tracing::info!(%id, "session process gone — marking dead");
 
-    if let Some(wt) = &worktree {
-        if settle_worktree_on_exit(inner, id, wt).await {
-            // The tree is settled (removed); a stale "in the way" would
-            // describe something the session no longer holds.
-            session.blocked_work = None;
-            let mut sessions = inner.sessions.lock().await;
-            if let Some(record) = sessions.get_mut(&id) {
-                record.session.worktree = None;
-                record.session.blocked_work = None;
-                record.worktree_path = None;
-                record.worktree_base_repo = None;
+    if settle_worktree {
+        if let Some(wt) = &worktree {
+            if settle_worktree_on_exit(inner, id, wt).await {
+                // The tree is settled (removed); a stale "in the way" would
+                // describe something the session no longer holds.
+                session.blocked_work = None;
+                let mut sessions = inner.sessions.lock().await;
+                if let Some(record) = sessions.get_mut(&id) {
+                    record.session.worktree = None;
+                    record.session.blocked_work = None;
+                    record.worktree_path = None;
+                    record.worktree_base_repo = None;
+                }
             }
         }
     }
@@ -1542,6 +1861,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,11 +6,18 @@ 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 common::{init_bare_repo, init_repo, push_all_to_origin, wait_for_async, DaemonFixture};
+use maestro_acp::schema::{ContentBlock, SessionUpdate};
 use maestro_acp::{ADAPTER_BIN, INSTALL_COMMAND};
-use maestro_protocol::{ActivityState, CreateSessionRequest, SessionStatus, SessionView};
+use maestro_daemon::sidecar::{EventCursor, SidecarStore};
+use maestro_daemon::worktree::WorktreeManager;
+use maestro_daemon::SessionManager;
+use maestro_protocol::{
+    ActivityState, CreateSessionRequest, SessionStatus, SessionView, WorktreeSpec,
+};
 use tempfile::TempDir;
 
 fn fake_agent() -> PathBuf {
@@ -61,13 +68,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 +162,232 @@ 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 shutdown_preserves_a_pristine_structured_worktree_for_recovery() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let repo = TempDir::new().unwrap();
+    init_repo(repo.path());
+    let origin = TempDir::new().unwrap();
+    init_bare_repo(origin.path());
+
+    let branch = "feature/restart";
+    let worktree = repo
+        .path()
+        .join("worktrees")
+        .join("feature")
+        .join("restart");
+    let session = fx
+        .manager
+        .create(CreateSessionRequest {
+            view: SessionView::Structured,
+            cwd: repo.path().to_path_buf(),
+            command: "claude".into(),
+            env: fake.env(),
+            metadata: HashMap::new(),
+            worktree: Some(WorktreeSpec {
+                base_repo: repo.path().to_path_buf(),
+                branch: branch.into(),
+                create_branch: true,
+            }),
+        })
+        .await
+        .expect("create");
+    assert_eq!(session.cwd, worktree);
+    push_all_to_origin(repo.path(), origin.path());
+
+    fx.manager.shutdown().await;
+
+    assert!(
+        worktree.is_dir(),
+        "shutdown must not settle a pristine worktree"
+    );
+    let sidecar: maestro_protocol::Session = serde_json::from_slice(
+        &std::fs::read(fx.sidecar_dir().join(format!("{}.json", session.id))).unwrap(),
+    )
+    .unwrap();
+    assert_eq!(
+        sidecar.worktree.as_ref().map(|wt| &wt.path),
+        Some(&worktree)
+    );
+
+    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)
+    );
+    assert!(fake.log().contains("recv session/load"), "{}", fake.log());
+    assert!(worktree.is_dir(), "recovery must retain the worktree");
+
+    revived.kill(session.id, false).await.expect("dismiss");
+    assert!(
+        !worktree.exists(),
+        "operator dismissal still cleans a pristine tree"
+    );
+}
+
+#[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 +498,82 @@ async fn shutdown_reaps_every_adapter_child() {
     assert_eq!(sidecar.status, SessionStatus::Dead);
 }
 
+#[tokio::test]
+async fn shutdown_waits_for_structured_creation_and_rejects_new_creates() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_DELAY_SESSION_NEW_MS".into(), "500".into());
+
+    let create_cwd = cwd.path().to_path_buf();
+    let manager = fx.manager.clone();
+    let create = tokio::spawn(async move { manager.create(structured(&create_cwd, env)).await });
+    wait_for_async(Duration::from_secs(5), || async {
+        fake.log().contains("recv session/new")
+    })
+    .await;
+
+    let manager = fx.manager.clone();
+    let shutdown = tokio::spawn(async move { manager.shutdown().await });
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    assert!(
+        !shutdown.is_finished(),
+        "shutdown returned before create finished"
+    );
+
+    let session = create.await.expect("create task").expect("create");
+    let pid = fake.pid();
+    shutdown.await.expect("shutdown task");
+
+    wait_for_exit(pid).await;
+    assert!(fake.log().contains("signal SIGTERM"), "{}", fake.log());
+    assert_eq!(
+        fx.manager.status(session.id).await,
+        Some(SessionStatus::Dead)
+    );
+    assert!(matches!(
+        fx.manager.create(structured(cwd.path(), fake.env())).await,
+        Err(maestro_daemon::SessionError::ShuttingDown)
+    ));
+
+    // A repeated signal must not attempt to terminate the child again.
+    fx.manager.shutdown().await;
+    assert_eq!(fake.log().matches("signal SIGTERM").count(), 1);
+}
+
+#[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();
+}

crates/maestro-e2e/Cargo.toml

@@ -13,6 +13,7 @@ path = "src/lib.rs"
 [dependencies]
 chrono.workspace = true
 futures-util.workspace = true
+libc.workspace = true
 maestro-protocol = { workspace = true, features = ["client"] }
 maestro-testing.workspace = true
 serde.workspace = true

crates/maestro-e2e/src/daemon.rs

@@ -87,6 +87,14 @@ impl Daemon {
         }
     }
 
+    /// Send the daemon the same SIGTERM systemd uses, without waiting for its
+    /// graceful HTTP shutdown to finish.
+    pub fn send_sigterm(&self) {
+        let child = self.child.as_ref().expect("daemon is running");
+        let result = unsafe { libc::kill(child.id() as libc::pid_t, libc::SIGTERM) };
+        assert_eq!(result, 0, "send SIGTERM to daemon");
+    }
+
     /// Re-spawn the daemon on the same port + state. The previous tmux
     /// server (under `TMUX_TMPDIR`) is left running so `recover()` can adopt
     /// its sessions.

crates/maestro-e2e/tests/e2e.rs

@@ -15,7 +15,7 @@ use std::time::{Duration, Instant};
 use futures_util::{SinkExt, StreamExt};
 use maestro_e2e::util::wait_for_async;
 use maestro_e2e::{Daemon, TestClient};
-use maestro_protocol::{CreateSessionRequest, WorktreeSpec};
+use maestro_protocol::{CreateSessionRequest, SessionStatus, SessionView, WorktreeSpec};
 use tempfile::TempDir;
 use tokio_tungstenite::tungstenite::Message as WsMessage;
 
@@ -36,6 +36,61 @@ fn create_req(cwd: &std::path::Path, command: &str) -> CreateSessionRequest {
     }
 }
 
+struct FakeAcp {
+    _dir: TempDir,
+    env: HashMap<String, String>,
+}
+
+fn fake_acp(log: &std::path::Path) -> FakeAcp {
+    let dir = tempfile::tempdir().expect("fake adapter dir");
+    let adapter = dir.path().join("claude-code-acp");
+    std::fs::copy(
+        concat!(
+            env!("CARGO_MANIFEST_DIR"),
+            "/../maestro-acp/tests/fake_acp_agent.py"
+        ),
+        &adapter,
+    )
+    .expect("copy fake adapter");
+    let mut permissions = std::fs::metadata(&adapter)
+        .expect("adapter metadata")
+        .permissions();
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+        permissions.set_mode(0o755);
+    }
+    std::fs::set_permissions(&adapter, permissions).expect("make fake adapter executable");
+    let env = HashMap::from([
+        (
+            "PATH".into(),
+            format!(
+                "{}:{}",
+                dir.path().display(),
+                std::env::var("PATH").unwrap_or_default()
+            ),
+        ),
+        ("FAKE_ACP_LOG".into(), log.to_string_lossy().into_owned()),
+    ]);
+    FakeAcp { _dir: dir, env }
+}
+
+fn fake_acp_pid(log: &std::path::Path) -> Option<u32> {
+    std::fs::read_to_string(log).ok()?.lines().find_map(|line| {
+        line.strip_prefix("start pid=")?
+            .split(' ')
+            .next()?
+            .parse()
+            .ok()
+    })
+}
+
+fn pid_alive(pid: u32) -> bool {
+    std::fs::read_to_string(format!("/proc/{pid}/stat"))
+        .map(|stat| !stat.contains(") Z "))
+        .unwrap_or(false)
+}
+
 /// A worktree spec built through its wire shape. The daemon and clients ship
 /// together (#87 dropped `cleanup_on_kill` from the struct), so the JSON
 /// object carries only the fields the lifecycle actually honours.
@@ -995,6 +1050,66 @@ async fn daemon_restart_recovers_session() {
     client.kill_session(s.id).await.expect("kill");
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn sigterm_stops_structured_child_before_events_ws_drains() {
+    let (mut daemon, client) = bring_up().await;
+    let cwd = TempDir::new().unwrap();
+    let log = cwd.path().join("fake-acp.log");
+    let fake = fake_acp(&log);
+    let session = client
+        .spawn_session(&CreateSessionRequest {
+            view: SessionView::Structured,
+            cwd: cwd.path().to_path_buf(),
+            command: "claude".into(),
+            env: fake.env.clone(),
+            metadata: HashMap::new(),
+            worktree: None,
+        })
+        .await
+        .expect("spawn structured session");
+    let ws_url = daemon.client().events_url().expect("events URL");
+    let (_events_ws, _) = tokio_tungstenite::connect_async(ws_url)
+        .await
+        .expect("events WS handshake");
+    wait_for_async("fake ACP starts", Duration::from_secs(5), || {
+        let log = log.clone();
+        async move { fake_acp_pid(&log).is_some() }
+    })
+    .await;
+    let pid = fake_acp_pid(&log).expect("fake ACP pid");
+
+    daemon.send_sigterm();
+
+    wait_for_async(
+        "fake ACP exits after SIGTERM",
+        Duration::from_secs(5),
+        || async { !pid_alive(pid) },
+    )
+    .await;
+    let sidecar_path = daemon
+        .state_dir()
+        .join("sessions")
+        .join(format!("{}.json", session.id));
+    wait_for_async(
+        "dead structured session sidecar",
+        Duration::from_secs(5),
+        || {
+            let sidecar_path = sidecar_path.clone();
+            async move {
+                std::fs::read(&sidecar_path)
+                    .ok()
+                    .and_then(|json| {
+                        serde_json::from_slice::<maestro_protocol::Session>(&json).ok()
+                    })
+                    .is_some_and(|saved| saved.status == SessionStatus::Dead)
+            }
+        },
+    )
+    .await;
+
+    daemon.stop();
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn concurrent_spawns_each_get_unique_ids() {
     let (_daemon, client) = bring_up().await;

docs/INSTALL.md

@@ -208,8 +208,10 @@ When the adapter is missing the spawn still creates the session, marked dead
 with the install command above as its activity detail, so the sidebar says
 what to run. The adapter is a child of the daemon: it ends with a `DELETE
 /api/sessions/:id` (`session/cancel`, then `SIGTERM`) and with the daemon
-itself. Terminal sessions keep surviving restarts through tmux; structured
-ones do not yet.
+itself. A daemon restart ends an in-flight structured turn, but preserves its
+worktree, session record, and event-log transcript. The replacement daemon
+starts a new adapter, resumes it with `session/load`, and replays the
+transcript. Terminal sessions keep surviving restarts through tmux.
 
 ## The browser UI