fix: bound daemon memory so a long session cannot exhaust it

bt-daemon/README.md

@@ -114,10 +114,18 @@ Phases 0–5 are implemented: protocol, daemon lifecycle, Braintrust sink,
 Codex and Claude translators, `bt daemon` integration, and thin hook shims for
 both shipped plugins. Restart recovery replays the redacted journal with
 deterministic span ids, so resubmitted rows merge into the same spans instead
-of creating duplicates. Claude lifecycle entries embed transcript snapshots, so
-recovery does not depend on mutable external paths. Explicit turn/session-end
+of creating duplicates. Claude lifecycle entries reference a daemon-owned
+transcript mirror, so recovery does not depend on mutable external paths
+without re-recording the transcript on every turn. Explicit turn/session-end
 flushes are bounded, and sessions can target project logs or an experiment.
 
+Memory is bounded end to end, while on-disk records stay complete: the daemon
+never holds a transcript or a whole journal in memory, mirroring and replay
+both stream, session queues apply backpressure, and sessions that go quiet are
+retired and rebuilt from their journal on the next event. Nothing on disk —
+journal, mirror, or conversation content — is capped or truncated; only
+in-memory caches are bounded, and each is re-derivable from disk.
+
 Windows named-pipe transport, detached spawning, lifecycle handover, and
 cross-platform pipeline tests are implemented. The remaining host follow-ups
 are OpenCode and pi, which are not present in this monorepo.

bt-daemon/docs/protocol.md

@@ -107,7 +107,10 @@ The hot path. Params are the **Envelope** (see below). Request result:
 ```
 `accepted: true` means enqueued to the session's ordered queue and journaled.
 The daemon never fails the caller's turn for a downstream (Braintrust) error;
-those are handled asynchronously and surfaced via `status.get`.
+those are handled asynchronously and surfaced via `status.get`. The queue is
+bounded, so a session whose sink has stalled applies backpressure here instead
+of accumulating events without limit; the event is already journaled by then,
+so this costs latency, never data.
 
 ### `session.flush` (request)
 
@@ -262,6 +265,13 @@ journal, logs, status, or RPC response. Envelopes journal only their non-secret
   handle, so it retries the exclusive claim without filesystem cleanup.
 - **Idle exit.** The daemon exits after `--idle-timeout` (default 300 s) with
   zero active sessions and empty queues.
+- **Session retirement.** A delivery pipeline with no traffic for
+  `--session-idle-timeout` (default 300 s) and an empty queue is flushed and
+  dropped: its translator state, sink handles, credential lease, and journal
+  handle are released rather than held for the daemon's lifetime. A later
+  event rebuilds it from the journal, and deterministic span ids merge the
+  re-emitted rows. This matters because the idle exit above requires *every*
+  session to be quiet, which for a continuously active user never happens.
 - **Version handover.** `initialize` compares versions. A newer client sends
   `daemon.shutdown`, waits until the endpoint no longer accepts connections,
   and spawns its own daemon. In-flight session state is rebuilt from the
@@ -309,8 +319,24 @@ profiles, organizations, and destinations while sharing one daemon.
   translator. The resulting rows may be resubmitted to repair delivery
   interrupted by a crash, but their deterministic ids target the same backend
   rows and must never create duplicate spans, and a route never receives
-  another route's rows.
-  Journals are GC'd after 7 days.
+  another route's rows. Replay streams the journal and is bounded to the
+  bytes recorded before the replaying session was created, so recovering a
+  long session costs no more memory than running it. The journal itself is
+  never capped or truncated — dropping entries would silently cost recovery
+  fidelity — so its size is governed by writing each transcript byte once
+  (below) and by GC after 7 days.
+- **Transcript mirrors.** Claude transcript files are external mutable state,
+  so a lifecycle event must stay replayable after the agent rewrites the path
+  it came from. The daemon appends new transcript bytes to
+  `<data_dir>/transcripts/<uuid>.jsonl` — one mirror per
+  (session_id, transcript path) — and the journal entry carries only
+  `_bt_transcript_mirror: {path, mirror, through}`. `through` is the mirror's
+  high-water offset at acceptance, which bounds replay to exactly the bytes
+  the live run saw. Storing each byte once keeps a journal proportional to a
+  session's transcript rather than to its transcript times its turn count.
+  Entries written before mirroring instead carry the whole transcript inline
+  as `_bt_transcript_snapshot`, which translators still accept. Mirrors are
+  GC'd after 7 days like the journal.
 - **Managed-run acceptance records.** Alongside the journal, each accepted
   event that carries a `managed_run_id` also appends `{session_id, route}` to
   `<data_dir>/managed-runs/<managed_run_id>.ndjson`. `managed_run.flush` reads

bt-daemon/src/dispatch.rs

@@ -10,11 +10,19 @@
 
 use crate::sink::SinkFactory;
 use crate::translate::{Registry, SessionCtx};
-use crate::wire::Envelope;
+use crate::wire::{Envelope, SessionRoute};
+use std::path::PathBuf;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
+use std::time::Instant;
 use tokio::sync::{mpsc, oneshot};
 
+/// Bound on a session's in-flight queue. Enqueue awaits a slot rather than
+/// letting a stalled sink accumulate events without limit; the daemon has
+/// already journaled anything waiting here, so backpressure costs latency,
+/// never data.
+const QUEUE_CAPACITY: usize = 1024;
+
 #[derive(Default)]
 pub struct Counters {
     pub queued: AtomicU64,
@@ -28,13 +36,24 @@ enum SessionMsg {
     Shutdown(oneshot::Sender<()>),
 }
 
+/// Where to rebuild a session's translator state from, streamed at startup.
+pub struct ReplayPlan {
+    pub journal_path: PathBuf,
+    pub route: SessionRoute,
+    /// Replay stops here — the journal's length when this session was
+    /// created, so the event creating it is not replayed and then delivered
+    /// a second time from the queue.
+    pub through: u64,
+}
+
 /// Handle to one live session: its queue plus observable counters/state.
 pub struct Session {
     pub source: String,
-    tx: mpsc::UnboundedSender<SessionMsg>,
+    tx: mpsc::Sender<SessionMsg>,
     pub counters: Arc<Counters>,
     pub last_error: Arc<Mutex<Option<String>>>,
     pub permalink: Arc<Mutex<Option<String>>>,
+    last_activity: Mutex<Instant>,
 }
 
 impl Session {
@@ -43,12 +62,12 @@ impl Session {
         session_id: String,
         source: String,
         plugin_version: Option<String>,
-        replay: Vec<Envelope>,
+        replay: Option<ReplayPlan>,
         config: crate::wire::SessionConfig,
         translators: Arc<Registry>,
         sink_factory: Arc<dyn SinkFactory>,
     ) -> Arc<Session> {
-        let (tx, rx) = mpsc::unbounded_channel();
+        let (tx, rx) = mpsc::channel(QUEUE_CAPACITY);
         let counters = Arc::new(Counters::default());
         let last_error = Arc::new(Mutex::new(None));
         let permalink = Arc::new(Mutex::new(None));
@@ -73,23 +92,36 @@ impl Session {
             counters,
             last_error,
             permalink,
+            last_activity: Mutex::new(Instant::now()),
         })
     }
 
     /// Enqueue an event after the daemon has journaled it.
-    pub fn enqueue(&self, env: Envelope) -> anyhow::Result<()> {
+    pub async fn enqueue(&self, env: Envelope) -> anyhow::Result<()> {
+        self.touch();
         self.counters.queued.fetch_add(1, Ordering::Relaxed);
         self.tx
             .send(SessionMsg::Event(Box::new(env)))
+            .await
             .map_err(|_| anyhow::anyhow!("session actor is gone"))?;
         Ok(())
     }
 
+    fn touch(&self) {
+        *self.last_activity.lock().unwrap() = Instant::now();
+    }
+
+    /// How long since this session last saw traffic. Drives idle retirement.
+    pub fn idle_for(&self) -> std::time::Duration {
+        self.last_activity.lock().unwrap().elapsed()
+    }
+
     /// Ask the actor to drain and flush its sink, bounded by `timeout`.
     /// Returns `(flushed, pending)`.
     pub async fn flush(&self, timeout: std::time::Duration) -> (bool, u64) {
+        self.touch();
         let (reply_tx, reply_rx) = oneshot::channel();
-        if self.tx.send(SessionMsg::Flush(reply_tx)).is_err() {
+        if self.tx.send(SessionMsg::Flush(reply_tx)).await.is_err() {
             return (false, self.counters.queued.load(Ordering::Relaxed));
         }
         match tokio::time::timeout(timeout, reply_rx).await {
@@ -104,26 +136,30 @@ impl Session {
         let (reply_tx, reply_rx) = oneshot::channel();
         self.tx
             .send(SessionMsg::Configure(Box::new(config), reply_tx))
+            .await
             .map_err(|_| anyhow::anyhow!("session actor is gone"))?;
         reply_rx
             .await
             .map_err(|_| anyhow::anyhow!("session actor dropped configuration reply"))
     }
 
-    /// Drain, flush, and stop the actor (used on daemon shutdown).
+    /// Drain, flush, and stop the actor (used on daemon shutdown and when an
+    /// idle session is retired).
     pub async fn shutdown(&self) {
         let (reply_tx, reply_rx) = oneshot::channel();
-        if self.tx.send(SessionMsg::Shutdown(reply_tx)).is_ok() {
+        if self.tx.send(SessionMsg::Shutdown(reply_tx)).await.is_ok() {
             let _ = reply_rx.await;
         }
     }
 }
 
-/// Claude transcript files are external mutable state. Capture them in the
-/// journal at lifecycle boundaries so recovery/replay does not depend on a
-/// path that Claude may later rewrite or delete. Fail open: a missing file is
-/// handled by the translator exactly as before.
-pub(crate) async fn hydrate_transcript_snapshot(env: &mut Envelope) {
+/// Claude transcript files are external mutable state. Mirror them into
+/// daemon-owned storage at lifecycle boundaries and journal only a reference,
+/// so recovery/replay does not depend on a path that Claude may later rewrite
+/// or delete — and so the transcript is stored once rather than re-copied into
+/// every event. Fail open: without a reference the translator reads the live
+/// path exactly as before.
+pub(crate) async fn hydrate_transcript_reference(data_dir: &std::path::Path, env: &mut Envelope) {
     if env.source != "claude-code"
         || !matches!(
             env.event.as_str(),
@@ -145,13 +181,22 @@ pub(crate) async fn hydrate_transcript_snapshot(env: &mut Envelope) {
     else {
         return;
     };
-    let Ok(contents) = tokio::fs::read_to_string(&path).await else {
-        return;
-    };
+    let (mirror, through) =
+        match crate::transcript_mirror::capture(data_dir, &env.session_id, &path).await {
+            Ok(captured) => captured,
+            Err(error) => {
+                tracing::debug!(session_id = %env.session_id, %error, "transcript mirror skipped");
+                return;
+            }
+        };
     if let Some(payload) = env.payload.as_object_mut() {
         payload.insert(
-            "_bt_transcript_snapshot".to_string(),
-            serde_json::json!({ "path": path, "contents": contents }),
+            "_bt_transcript_mirror".to_string(),
+            serde_json::json!({
+                "path": path,
+                "mirror": mirror.to_string_lossy(),
+                "through": through,
+            }),
         );
     }
 }
@@ -165,12 +210,12 @@ struct SessionActor {
     counters: Arc<Counters>,
     last_error: Arc<Mutex<Option<String>>>,
     permalink: Arc<Mutex<Option<String>>>,
-    replay: Vec<Envelope>,
+    replay: Option<ReplayPlan>,
     config: crate::wire::SessionConfig,
 }
 
 impl SessionActor {
-    async fn run(self, mut rx: mpsc::UnboundedReceiver<SessionMsg>) {
+    async fn run(self, mut rx: mpsc::Receiver<SessionMsg>) {
         let mut translator = self.translators.create(&self.source, &self.session_id);
         let mut sink = match self.sink_factory.create(
             &self.session_id,
@@ -211,8 +256,7 @@ impl SessionActor {
         // Rebuild translator state before accepting the first new event.
         // Stable span ids make this both crash recovery and a complete copy
         // when an existing source session is sent to another destination.
-        self.replay_into(&mut translator, &mut sink, &ctx, &self.replay)
-            .await;
+        self.replay_into(&mut translator, &mut sink, &ctx).await;
 
         while let Some(msg) = rx.recv().await {
             match msg {
@@ -252,28 +296,62 @@ impl SessionActor {
         }
     }
 
+    /// Stream the journal through the translator, emitting each entry's spans
+    /// as they are produced. Nothing is accumulated across entries: peak
+    /// memory is one journal entry and the ops it yields, so recovering a
+    /// long session costs the same as running it.
     async fn replay_into(
         &self,
         translator: &mut Box<dyn crate::translate::AgentTranslator>,
         sink: &mut Box<dyn crate::sink::Sink>,
         ctx: &SessionCtx,
-        replay: &[Envelope],
     ) {
-        let mut replay_ops = Vec::new();
-        for env in replay {
-            match translator.handle(env, ctx) {
-                Ok(mut ops) => replay_ops.append(&mut ops),
-                Err(e) => self.set_error(format!("journal replay failed: {e}")),
-            }
-        }
-        if replay_ops.is_empty() {
+        let Some(plan) = &self.replay else {
             return;
-        }
-        match sink.emit(&replay_ops).await {
-            Ok(n) => {
-                self.counters.spans_emitted.fetch_add(n, Ordering::Relaxed);
+        };
+        let mut reader = match crate::journal::JournalReader::open(&plan.journal_path, plan.through)
+            .await
+        {
+            Ok(Some(reader)) => reader,
+            Ok(None) => return,
+            Err(error) => {
+                tracing::warn!(session_id = %self.session_id, "journal replay skipped: {error}");
+                return;
+            }
+        };
+        loop {
+            let entry = match reader.next_entry().await {
+                Ok(Some(entry)) => entry,
+                Ok(None) => break,
+                Err(error) => {
+                    tracing::warn!(session_id = %self.session_id, "journal replay stopped: {error}");
+                    break;
+                }
+            };
+            if !entry
+                .route
+                .as_ref()
+                .is_some_and(|candidate| candidate.same_route(&plan.route))
+            {
+                continue;
+            }
+            let env = crate::journal::envelope_from_redacted(entry);
+            let ops = match translator.handle(&env, ctx) {
+                Ok(ops) => ops,
+                Err(e) => {
+                    self.set_error(format!("journal replay failed: {e}"));
+                    continue;
+                }
+            };
+            if ops.is_empty() {
+                continue;
+            }
+            match sink.emit(&ops).await {
+                Ok(n) => {
+                    self.counters.spans_emitted.fetch_add(n, Ordering::Relaxed);
+                }
+                Err(e) => self.set_error(format!("sink replay emit failed: {e}")),
             }
-            Err(e) => self.set_error(format!("sink replay emit failed: {e}")),
         }
     }