From 7575f5d82904af22338c220ead629108c5dc8571 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Mon, 17 Aug 2026 12:54:55 +0800 Subject: [PATCH] fix: bound daemon memory so a long session cannot exhaust it A user reported the daemon reaching 20 GB before crashing. Nothing in the long-lived daemon bounded its in-memory footprint: not payloads, not replay, not the conversation history, and not the per-session maps. The dominant cost was transcript snapshotting. Every Claude lifecycle event (UserPromptSubmit/Stop/StopFailure/SubagentStop/SessionEnd) read the entire transcript into memory and journaled it verbatim, so a session re-recorded its whole growing transcript on every turn. On my own machine an 18 MB transcript had produced a 1.6 GB journal: 142 snapshot lines accounted for 1.54 GB while all 5,852 other lines totalled 19 MB. Replay then loaded that file with read_to_string and parsed every line into a Vec held at once, and accumulated every resulting span op -- each LLM row carrying a full history clone -- into a single Vec before emitting. That is the crash. The waste was total rather than incidental: the translator only ever consumes bytes past its cursor offset, so the entire prefix of every snapshot was dead weight in all cases. Only in-memory caches are bounded here. Nothing on disk -- journal, mirror, or conversation content -- is capped or truncated, because each is a durability record whose loss is silent; they are kept small by storing every byte once rather than by dropping data, and every in-memory structure is re-derivable from them. Changes: - Mirror transcripts instead of snapshotting them. New transcript_mirror module streams only new bytes into a daemon-owned file; the journal carries a reference plus the high-water offset. Each byte is stored once, and the offset bounds replay to exactly what the live run saw. Keeps the durability property that motivated snapshots -- the mirror is immune to the agent rewriting the path -- and makes journal growth linear in transcript size. - Stream journal replay. JournalReader reads entry by entry, bounded to the journal length recorded when the session was created, and each entry's spans are emitted as produced. Peak memory is one entry, not a whole session. - Retire idle sessions. Nothing ever removed a session from the six daemon maps, so translator state, sink handles, credential leases, and journal file handles accumulated until the process exited -- which for a continuously busy user never happened, since the idle watchdog needs every session quiet. Sessions now retire after --session-idle-timeout and rebuild from the journal on a later event. - Bound the session queue. Unbounded mpsc replaced with a bounded channel, so a stalled sink applies backpressure rather than accumulating events. The event is journaled before enqueue, so this costs latency, never data. - Drop two redundant full clones of the conversation history per turn. - Run journal/mirror GC periodically, not only at daemon startup. Backward compatible: journals written before mirroring inline the transcript as _bt_transcript_snapshot, and translators still accept that form. A test asserts both forms translate identically across all three Claude fixtures. Tests: journal growth stays flat across 40 turns on a growing transcript (would have caught the original bug), idle sessions retire and resume from their journal with stable span ids, and the mirror reference is self-contained. Co-Authored-By: Claude Opus 5 (1M context) --- bt-daemon/README.md | 12 +- bt-daemon/docs/protocol.md | 32 ++++- bt-daemon/src/dispatch.rs | 150 +++++++++++++++----- bt-daemon/src/journal.rs | 71 ++++++++-- bt-daemon/src/lib.rs | 6 + bt-daemon/src/server.rs | 158 +++++++++++++++------ bt-daemon/src/transcript_mirror.rs | 163 ++++++++++++++++++++++ bt-daemon/src/translate/claude.rs | 175 ++++++++++++----------- bt-daemon/tests/claude_translator.rs | 60 +++++++- bt-daemon/tests/pipeline.rs | 200 ++++++++++++++++++++++++++- 10 files changed, 847 insertions(+), 180 deletions(-) create mode 100644 bt-daemon/src/transcript_mirror.rs diff --git a/bt-daemon/README.md b/bt-daemon/README.md index 642c96a..2e3c346 100644 --- a/bt-daemon/README.md +++ b/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. diff --git a/bt-daemon/docs/protocol.md b/bt-daemon/docs/protocol.md index 6faf8e2..6bc832f 100644 --- a/bt-daemon/docs/protocol.md +++ b/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 + `/transcripts/.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 `/managed-runs/.ndjson`. `managed_run.flush` reads diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 7f8bf53..49a30bf 100644 --- a/bt-daemon/src/dispatch.rs +++ b/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, + tx: mpsc::Sender, pub counters: Arc, pub last_error: Arc>>, pub permalink: Arc>>, + last_activity: Mutex, } impl Session { @@ -43,12 +62,12 @@ impl Session { session_id: String, source: String, plugin_version: Option, - replay: Vec, + replay: Option, config: crate::wire::SessionConfig, translators: Arc, sink_factory: Arc, ) -> Arc { - 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, last_error: Arc>>, permalink: Arc>>, - replay: Vec, + replay: Option, config: crate::wire::SessionConfig, } impl SessionActor { - async fn run(self, mut rx: mpsc::UnboundedReceiver) { + async fn run(self, mut rx: mpsc::Receiver) { 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, sink: &mut Box, 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}")), } } diff --git a/bt-daemon/src/journal.rs b/bt-daemon/src/journal.rs index 69da587..17c71e7 100644 --- a/bt-daemon/src/journal.rs +++ b/bt-daemon/src/journal.rs @@ -13,7 +13,7 @@ use crate::wire::{BackendAuth, Envelope, RedactedEnvelope, SessionRoute}; use serde::{Deserialize, Serialize}; use std::collections::HashSet; use std::path::{Path, PathBuf}; -use tokio::io::AsyncWriteExt; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt}; pub fn journal_dir(data_dir: &Path) -> PathBuf { data_dir.join("journal") @@ -140,6 +140,12 @@ impl JournalWriter { /// Append one event in redacted form and flush to the OS. Not fsync'd per /// event (that would dominate hook latency); an OS crash can lose the last /// few lines, which replay tolerates. + /// + /// The journal is a durability record and is never capped or truncated: + /// dropping entries would silently cost recovery fidelity. Its size is + /// bounded instead by writing each transcript byte once (see + /// [`crate::transcript_mirror`]) and by age-based GC, and replay reads it + /// as a stream so a large journal never becomes a large allocation. pub async fn append(&mut self, env: &Envelope) -> anyhow::Result<()> { let mut line = serde_json::to_vec(&env.redacted())?; line.push(b'\n'); @@ -149,19 +155,60 @@ impl JournalWriter { } } -/// Read a journal file back into redacted envelopes (for replay/rebuild). -pub async fn read_journal(path: &Path) -> anyhow::Result> { - let data = tokio::fs::read_to_string(path).await?; - let mut out = Vec::new(); - for (i, line) in data.lines().enumerate() { - if line.trim().is_empty() { - continue; +/// Streaming reader over one session's journal. +/// +/// Replay must never materialize a whole journal: the file is read line by +/// line so peak memory is one entry, not the entire recorded session. +pub struct JournalReader { + lines: tokio::io::Lines>>, + path: PathBuf, + line_no: usize, +} + +impl JournalReader { + /// Open a journal for streaming, reading at most `through` bytes. + /// + /// That bound is what keeps replay from consuming the very event that + /// triggered the session's creation: the caller records the journal's + /// length before appending, so the actor replays strictly what was + /// already recovered state, never the live event still on its way to the + /// queue. `Ok(None)` means the session has no journal yet. + pub async fn open(path: &Path, through: u64) -> anyhow::Result> { + let file = match tokio::fs::File::open(path).await { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error.into()), + }; + Ok(Some(Self { + lines: tokio::io::BufReader::new(file.take(through)).lines(), + path: path.to_path_buf(), + line_no: 0, + })) + } + + /// The journal's current length, which is the bound a session created now + /// should replay through. A missing journal replays nothing. + pub async fn recorded_len(path: &Path) -> u64 { + tokio::fs::metadata(path) + .await + .map(|meta| meta.len()) + .unwrap_or(0) + } + + /// The next entry, or `None` at end of file. + pub async fn next_entry(&mut self) -> anyhow::Result> { + while let Some(line) = self.lines.next_line().await? { + self.line_no += 1; + if line.trim().is_empty() { + continue; + } + let env: RedactedEnvelope = serde_json::from_str(&line).map_err(|error| { + anyhow::anyhow!("journal {}:{}: {error}", self.path.display(), self.line_no) + })?; + return Ok(Some(env)); } - let env: RedactedEnvelope = serde_json::from_str(line) - .map_err(|e| anyhow::anyhow!("journal {}:{}: {e}", path.display(), i + 1))?; - out.push(env); + Ok(None) } - Ok(out) } /// Best-effort age-based journal collection. A failed stat/remove is logged diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 8e3fa52..1ec5532 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -24,6 +24,7 @@ mod sink; mod trace_command; mod trace_runtime; mod transcript_import; +mod transcript_mirror; mod translate; mod transport; @@ -69,6 +70,11 @@ pub struct ServeArgs { /// disables the watchdog. #[arg(long, default_value_t = 300)] pub idle_timeout_secs: u64, + /// Retire a session's in-memory state after this many seconds without + /// traffic. A later event rebuilds it from the journal. 0 disables + /// retirement. + #[arg(long, default_value_t = 300)] + pub session_idle_timeout_secs: u64, } /// Arguments for `hook`. diff --git a/bt-daemon/src/server.rs b/bt-daemon/src/server.rs index fc697c4..b9e58b7 100644 --- a/bt-daemon/src/server.rs +++ b/bt-daemon/src/server.rs @@ -2,7 +2,7 @@ //! serves JSON-RPC connections, and shuts down gracefully (idle timeout, //! `daemon.shutdown`, or SIGINT/SIGTERM). -use crate::dispatch::{hydrate_transcript_snapshot, Session}; +use crate::dispatch::{hydrate_transcript_reference, ReplayPlan, Session}; use crate::journal::{self, JournalWriter}; use crate::sink::SinkFactory; use crate::translate::Registry; @@ -239,33 +239,6 @@ impl Daemon { *self.last_activity.lock().unwrap() = Instant::now(); } - async fn replay_for( - &self, - session_id: &str, - route: &SessionRoute, - ) -> anyhow::Result> { - match journal::read_journal(&journal::journal_path(&self.data_dir, session_id)).await { - Ok(entries) => Ok(entries - .into_iter() - .filter(|entry| { - entry - .route - .as_ref() - .is_some_and(|candidate| candidate.same_route(route)) - }) - .map(journal::envelope_from_redacted) - .collect()), - Err(error) - if error - .downcast_ref::() - .is_some_and(|io| io.kind() == std::io::ErrorKind::NotFound) => - { - Ok(Vec::new()) - } - Err(error) => Err(error), - } - } - async fn session_for(&self, env: &Envelope, key: &DeliveryKey) -> anyhow::Result> { { let map = self.sessions.lock().unwrap(); @@ -273,22 +246,23 @@ impl Daemon { return Ok(session.clone()); } } - // Open the journal outside the lock (async I/O), then insert under it, - // resolving a race where two connections create the same session. let route = env .route .as_ref() .ok_or_else(|| anyhow::anyhow!("event is missing its session route"))?; - let replay = match self.replay_for(&env.session_id, route).await { - Ok(replay) => replay, - Err(error) => { - tracing::warn!(session_id = %env.session_id, "journal replay skipped: {error}"); - Vec::new() - } - }; let config = env.config.clone().ok_or_else(|| { anyhow::anyhow!("resolved route is missing its session configuration") })?; + // The actor streams the journal itself, so creating a session stays + // cheap and allocation-free here no matter how long the recorded + // session is. Bound it to what is recorded now, before this event is + // appended, so replay covers recovery only. + let journal_path = journal::journal_path(&self.data_dir, &env.session_id); + let replay = ReplayPlan { + through: journal::JournalReader::recorded_len(&journal_path).await, + journal_path, + route: route.clone(), + }; let mut map = self.sessions.lock().unwrap(); if let Some(s) = map.get(key) { return Ok(s.clone()); @@ -297,7 +271,7 @@ impl Daemon { env.session_id.clone(), env.source.clone(), env.plugin_version.clone(), - replay, + Some(replay), config, self.translators.clone(), self.sink_factory.clone(), @@ -306,8 +280,60 @@ impl Daemon { Ok(session) } + /// Drop every trace of one delivery pipeline. A session that goes quiet + /// must not pin its translator state, sink handles, credential lease, or + /// journal file for the rest of the daemon's life; deterministic span ids + /// mean a late event simply rebuilds it from the journal. + async fn retire_session(&self, key: &DeliveryKey) { + let lock = self.session_lock(&key.session_id); + let _guard = lock.lock().await; + + let session = { self.sessions.lock().unwrap().remove(key) }; + let Some(session) = session else { + return; + }; + session.shutdown().await; + + self.session_auth.lock().await.remove(key); + self.auth_errors.lock().unwrap().remove(key); + self.managed_run_sessions.lock().unwrap().retain(|_, keys| { + keys.remove(key); + !keys.is_empty() + }); + + // The journal writer and lock are keyed by session id, which several + // delivery pipelines can share; only release them once the last one + // for that id is gone. + let last = !self + .sessions + .lock() + .unwrap() + .keys() + .any(|other| other.session_id == key.session_id); + if last { + self.journals.lock().unwrap().remove(&key.session_id); + self.session_locks.lock().unwrap().remove(&key.session_id); + } + tracing::info!(session_id = %key.session_id, "session retired"); + } + + /// Delivery pipelines with no traffic for `idle_timeout` and nothing left + /// queued. + fn idle_sessions(&self, idle_timeout: Duration) -> Vec { + self.sessions + .lock() + .unwrap() + .iter() + .filter(|(_, session)| { + session.idle_for() >= idle_timeout + && session.counters.queued.load(Ordering::Relaxed) == 0 + }) + .map(|(key, _)| key.clone()) + .collect() + } + async fn append_to_journal(&self, env: &mut Envelope) -> anyhow::Result<()> { - hydrate_transcript_snapshot(env).await; + hydrate_transcript_reference(&self.data_dir, env).await; let existing = { self.journals.lock().unwrap().get(&env.session_id).cloned() }; let writer = match existing { Some(writer) => writer, @@ -468,10 +494,14 @@ pub async fn run(args: ServeArgs, opts: ServeOptions) -> anyhow::Result<()> { tracing::info!(socket = %socket.display(), "bt-daemon listening"); let daemon = Daemon::new(opts, data_dir); - journal::gc_old_journals(&daemon.data_dir, Duration::from_secs(7 * 24 * 60 * 60)).await; - journal::gc_old_managed_runs(&daemon.data_dir, Duration::from_secs(7 * 24 * 60 * 60)).await; + collect_garbage(&daemon.data_dir).await; let idle_timeout = Duration::from_secs(args.idle_timeout_secs); spawn_idle_watchdog(daemon.clone(), idle_timeout); + spawn_session_reaper( + daemon.clone(), + Duration::from_secs(args.session_idle_timeout_secs), + ); + spawn_gc(daemon.clone()); let accept_result = accept_loop(daemon.clone(), listener).await; @@ -619,6 +649,7 @@ async fn accept_event(daemon: &Arc, mut env: Envelope) -> Result<(), Str .map_err(|error| format!("journal failed: {error}"))?; session .enqueue(env) + .await .map_err(|error| format!("enqueue failed: {error}"))?; Ok(delivery_key) } @@ -806,6 +837,51 @@ impl Daemon { } } +/// How long recovery state is kept on disk. +const RETENTION: Duration = Duration::from_secs(7 * 24 * 60 * 60); +/// How often that retention is enforced while the daemon keeps running. +const GC_INTERVAL: Duration = Duration::from_secs(60 * 60); + +async fn collect_garbage(data_dir: &std::path::Path) { + journal::gc_old_journals(data_dir, RETENTION).await; + journal::gc_old_managed_runs(data_dir, RETENTION).await; + crate::transcript_mirror::gc_old_mirrors(data_dir, RETENTION).await; +} + +/// Retire delivery pipelines that have gone quiet. Without this, every session +/// the daemon ever saw keeps its translator state, sink handles, and journal +/// file handle alive until the process exits — which for a continuously busy +/// user is never, since the idle watchdog needs *all* sessions quiet. +fn spawn_session_reaper(daemon: Arc, idle_timeout: Duration) { + if idle_timeout.is_zero() { + return; // 0 disables retirement (useful in tests) + } + tokio::spawn(async move { + let tick = (idle_timeout / 4).max(Duration::from_secs(1)); + loop { + tokio::select! { + _ = daemon.shutdown.notified() => return, + _ = tokio::time::sleep(tick) => {} + } + for key in daemon.idle_sessions(idle_timeout) { + daemon.retire_session(&key).await; + } + } + }); +} + +fn spawn_gc(daemon: Arc) { + tokio::spawn(async move { + loop { + tokio::select! { + _ = daemon.shutdown.notified() => return, + _ = tokio::time::sleep(GC_INTERVAL) => {} + } + collect_garbage(&daemon.data_dir).await; + } + }); +} + fn spawn_idle_watchdog(daemon: Arc, idle_timeout: Duration) { if idle_timeout.is_zero() { return; // 0 disables the watchdog (useful in tests) diff --git a/bt-daemon/src/transcript_mirror.rs b/bt-daemon/src/transcript_mirror.rs new file mode 100644 index 0000000..18502d2 --- /dev/null +++ b/bt-daemon/src/transcript_mirror.rs @@ -0,0 +1,163 @@ +//! Daemon-owned append-only mirrors of agent transcript files. +//! +//! Claude transcript files are external mutable state, so a journaled +//! lifecycle event must stay replayable even after the agent rewrites or +//! deletes the path it came from. Embedding the whole transcript in every +//! lifecycle event bought that durability at quadratic cost: a session +//! re-journaled its entire (growing) transcript on every turn, so an 18 MB +//! transcript produced a 1.6 GB journal that replay then had to hold in +//! memory all at once. +//! +//! Mirroring stores each transcript byte exactly once. The journal carries +//! only a reference — the mirror path plus the high-water offset that existed +//! when the event was accepted — so replay reads the same bytes the live run +//! saw, straight off disk, without the daemon ever holding a transcript in +//! memory. + +use std::path::{Path, PathBuf}; +use tokio::io::{AsyncSeekExt, AsyncWriteExt}; +use uuid::Uuid; + +/// Namespace for mirror file names (distinct from the span-id namespace). +const NAMESPACE: Uuid = Uuid::from_u128(0x3d51_9a02_7c64_4b8f_9e17_a2c5_0d63_88f1); + +pub fn mirror_dir(data_dir: &Path) -> PathBuf { + data_dir.join("transcripts") +} + +/// A stable per-(session, transcript path) mirror file name. Keyed by both so +/// one session's main and subagent transcripts never collide, and so two +/// sessions reading the same path keep independent mirrors. +pub fn mirror_path(data_dir: &Path, session_id: &str, source: &str) -> PathBuf { + let name = format!("{session_id}\u{1f}{source}"); + let digest = Uuid::new_v5(&NAMESPACE, name.as_bytes()) + .simple() + .to_string(); + mirror_dir(data_dir).join(format!("{digest}.jsonl")) +} + +/// Append everything written to `source` since the last capture, returning the +/// mirror path and the mirror's resulting length. That length is the exact +/// high-water offset the caller should journal: replay bounded by it sees the +/// transcript as of this moment and no further. +/// +/// If `source` is shorter than the mirror (the agent rewrote or truncated it), +/// the mirror is rebuilt from scratch so it never interleaves two generations. +pub async fn capture( + data_dir: &Path, + session_id: &str, + source: &str, +) -> anyhow::Result<(PathBuf, u64)> { + let path = mirror_path(data_dir, session_id, source); + tokio::fs::create_dir_all(mirror_dir(data_dir)).await?; + + let mirrored = tokio::fs::metadata(&path) + .await + .map(|meta| meta.len()) + .unwrap_or(0); + let mut input = tokio::fs::File::open(source).await?; + let source_len = input.metadata().await?.len(); + + // A shorter source means the file was replaced; start the mirror over. + let restart = source_len < mirrored; + let from = if restart { 0 } else { mirrored }; + if from >= source_len { + return Ok((path, mirrored)); + } + + input.seek(std::io::SeekFrom::Start(from)).await?; + let mut mirror = tokio::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(restart) + .append(!restart) + .open(&path) + .await?; + // Streamed, never buffered whole: the delta is copied through a small + // fixed buffer, so mirroring an arbitrarily large transcript costs + // arbitrarily little memory. The copy is unbounded in bytes on purpose — + // the mirror is the durable record and must stay complete. + let copied = tokio::io::copy(&mut input, &mut mirror).await?; + mirror.flush().await?; + Ok((path, from + copied)) +} + +/// Best-effort age-based collection, mirroring journal GC. Mirrors are only +/// useful for as long as their journal survives. +pub async fn gc_old_mirrors(data_dir: &Path, max_age: std::time::Duration) { + let dir = mirror_dir(data_dir); + let Ok(mut entries) = tokio::fs::read_dir(&dir).await else { + return; + }; + let now = std::time::SystemTime::now(); + while let Ok(Some(entry)) = entries.next_entry().await { + let path = entry.path(); + if path.extension().and_then(|value| value.to_str()) != Some("jsonl") { + continue; + } + let old = entry + .metadata() + .await + .ok() + .and_then(|meta| meta.modified().ok()) + .and_then(|modified| now.duration_since(modified).ok()) + .is_some_and(|age| age > max_age); + if old { + let _ = tokio::fs::remove_file(&path).await; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn capture_is_incremental_and_reports_the_high_water_offset() { + let tmp = tempfile::tempdir().unwrap(); + let source = tmp.path().join("t.jsonl"); + tokio::fs::write(&source, b"one\n").await.unwrap(); + + let (mirror, first) = capture(tmp.path(), "s1", source.to_str().unwrap()) + .await + .unwrap(); + assert_eq!(first, 4); + + tokio::fs::write(&source, b"one\ntwo\n").await.unwrap(); + let (_, second) = capture(tmp.path(), "s1", source.to_str().unwrap()) + .await + .unwrap(); + assert_eq!(second, 8); + assert_eq!(tokio::fs::read(&mirror).await.unwrap(), b"one\ntwo\n"); + } + + #[tokio::test] + async fn a_truncated_source_restarts_the_mirror() { + let tmp = tempfile::tempdir().unwrap(); + let source = tmp.path().join("t.jsonl"); + tokio::fs::write(&source, b"aaaa\nbbbb\n").await.unwrap(); + capture(tmp.path(), "s1", source.to_str().unwrap()) + .await + .unwrap(); + + tokio::fs::write(&source, b"cc\n").await.unwrap(); + let (mirror, len) = capture(tmp.path(), "s1", source.to_str().unwrap()) + .await + .unwrap(); + assert_eq!(len, 3); + assert_eq!(tokio::fs::read(&mirror).await.unwrap(), b"cc\n"); + } + + #[tokio::test] + async fn separate_sessions_and_paths_get_separate_mirrors() { + let tmp = tempfile::tempdir().unwrap(); + assert_ne!( + mirror_path(tmp.path(), "s1", "/a.jsonl"), + mirror_path(tmp.path(), "s2", "/a.jsonl") + ); + assert_ne!( + mirror_path(tmp.path(), "s1", "/a.jsonl"), + mirror_path(tmp.path(), "s1", "/b.jsonl") + ); + } +} diff --git a/bt-daemon/src/translate/claude.rs b/bt-daemon/src/translate/claude.rs index c0144d7..979552e 100644 --- a/bt-daemon/src/translate/claude.rs +++ b/bt-daemon/src/translate/claude.rs @@ -441,8 +441,13 @@ impl ClaudeTranslator { .map(|cursor| std::mem::take(&mut cursor.buffered)) .unwrap_or_default(); let parsed = parse_transcript(&records, std::mem::take(&mut self.main_history)); - self.main_history = parsed.history.clone(); - self.emit_parsed(parsed, "main", parent, ops); + // Moved, not cloned: this used to deep-copy the whole conversation on + // every turn. The history is never trimmed — it is the model input + // these spans report, so dropping any of it would silently corrupt the + // trace. Its footprint is the active session's own conversation, and + // retiring an idle session releases it. + self.main_history = parsed.history; + self.emit_parsed(parsed.calls, parsed.tools, "main", parent, ops); } fn emit_transcript( @@ -453,17 +458,18 @@ impl ClaudeTranslator { ops: &mut Vec, ) { let parsed = parse_transcript(records, Vec::new()); - self.emit_parsed(parsed, scope, parent, ops); + self.emit_parsed(parsed.calls, parsed.tools, scope, parent, ops); } fn emit_parsed( &mut self, - parsed: ParsedTranscript, + calls: Vec, + tools: Vec, scope: &str, parent: &str, ops: &mut Vec, ) { - for call in parsed.calls { + for call in calls { let request_key = format!("{scope}:{}", call.request_id); if self.emitted_requests.insert(request_key.clone()) { let span_key = format!("{scope}:llm:{}", call.request_id); @@ -474,7 +480,7 @@ impl ClaudeTranslator { ))); } } - for tool in parsed.tools { + for tool in tools { if self.emitted_tools.insert(tool.call_id.clone()) { let span_key = format!("tool:{}", tool.call_id); ops.push(SpanOp::Insert(tool.into_row( @@ -503,8 +509,8 @@ impl ClaudeTranslator { let current_turn_rows = cursor.buffered.split_off(split); let previous_rows = std::mem::replace(&mut cursor.buffered, current_turn_rows); let parsed = parse_transcript(&previous_rows, std::mem::take(&mut self.main_history)); - self.main_history = parsed.history.clone(); - self.emit_parsed(parsed, "main", &parent, ops); + self.main_history = parsed.history; + self.emit_parsed(parsed.calls, parsed.tools, "main", &parent, ops); self.git .enrich_rows(self.last_turn_cwd.as_deref(), &mut ops[op_start..]); } @@ -1027,29 +1033,26 @@ impl TranscriptTool { } } -fn read_records_until(path: &str, offset: &mut u64, cutoff_ms: i64) -> Vec { - let Ok(mut file) = std::fs::File::open(path) else { - return Vec::new(); - }; - let len = file.metadata().map(|m| m.len()).unwrap_or(0); - if *offset > len { - *offset = 0; - } - if file.seek(SeekFrom::Start(*offset)).is_err() { - return Vec::new(); - } - let mut reader = std::io::BufReader::new(file); - read_buffered_until(&mut reader, offset, cutoff_ms) -} - -fn read_buffered_until( +/// Read newline-delimited records forward from `offset`, advancing it past +/// everything consumed. +/// +/// `cutoff_ms` stops before the first record stamped after the hook that is +/// being handled, so replaying against a transcript that already contains the +/// finished session does not pull in future turns. `through` stops at a byte +/// offset — the transcript mirror's high-water mark, which bounds a replayed +/// event to exactly the bytes the live run saw. +fn read_buffered_bounded( reader: &mut std::io::BufReader, offset: &mut u64, - cutoff_ms: i64, + cutoff_ms: Option, + through: Option, ) -> Vec { let mut records = Vec::new(); let mut line = String::new(); loop { + if through.is_some_and(|through| *offset >= through) { + break; + } line.clear(); let start = *offset; let Ok(read) = reader.read_line(&mut line) else { @@ -1058,11 +1061,16 @@ fn read_buffered_until( if read == 0 { break; } + if through.is_some_and(|through| start + read as u64 > through) { + break; + } let Ok(value) = serde_json::from_str::(line.trim()) else { *offset += read as u64; continue; }; - if parse_timestamp_ms(&value).is_some_and(|timestamp| timestamp > cutoff_ms) { + if cutoff_ms.is_some_and(|cutoff| { + parse_timestamp_ms(&value).is_some_and(|timestamp| timestamp > cutoff) + }) { *offset = start; break; } @@ -1072,26 +1080,12 @@ fn read_buffered_until( records } -fn read_event_records(event: &Envelope, path: &str, offset: &mut u64) -> Vec { - let import_through_offset = event - .payload - .get("_bt_import_through_offset") - .and_then(Value::as_u64); - let snapshot = event - .payload - .get("_bt_transcript_snapshot") - .filter(|snapshot| snapshot.get("path").and_then(Value::as_str) == Some(path)) - .and_then(|snapshot| snapshot.get("contents")) - .and_then(Value::as_str); - match (snapshot, import_through_offset) { - (Some(contents), Some(through)) => read_snapshot_through_offset(contents, offset, through), - (None, Some(through)) => read_records_through_offset(path, offset, through), - (Some(contents), None) => read_snapshot_until(contents, offset, event.ts_ms), - (None, None) => read_records_until(path, offset, event.ts_ms), - } -} - -fn read_records_through_offset(path: &str, offset: &mut u64, through: u64) -> Vec { +fn read_file_bounded( + path: &str, + offset: &mut u64, + cutoff_ms: Option, + through: Option, +) -> Vec { let Ok(mut file) = std::fs::File::open(path) else { return Vec::new(); }; @@ -1102,53 +1096,76 @@ fn read_records_through_offset(path: &str, offset: &mut u64, through: u64) -> Ve if file.seek(SeekFrom::Start(*offset)).is_err() { return Vec::new(); } - read_buffered_through_offset(&mut std::io::BufReader::new(file), offset, through.min(len)) + let through = through.map(|through| through.min(len)); + read_buffered_bounded( + &mut std::io::BufReader::new(file), + offset, + cutoff_ms, + through, + ) } -fn read_snapshot_through_offset(contents: &str, offset: &mut u64, through: u64) -> Vec { - if *offset > contents.len() as u64 { +fn read_snapshot_bounded( + contents: &str, + offset: &mut u64, + cutoff_ms: Option, + through: Option, +) -> Vec { + let len = contents.len() as u64; + if *offset > len { *offset = 0; } let mut reader = std::io::BufReader::new(std::io::Cursor::new(contents.as_bytes())); if reader.seek(SeekFrom::Start(*offset)).is_err() { return Vec::new(); } - read_buffered_through_offset(&mut reader, offset, through.min(contents.len() as u64)) + let through = through.map(|through| through.min(len)); + read_buffered_bounded(&mut reader, offset, cutoff_ms, through) } -fn read_buffered_through_offset( - reader: &mut std::io::BufReader, - offset: &mut u64, - through: u64, -) -> Vec { - let mut records = Vec::new(); - let mut line = String::new(); - while *offset < through { - line.clear(); - let start = *offset; - let Ok(read) = reader.read_line(&mut line) else { - break; - }; - if read == 0 || start + read as u64 > through { - break; - } - *offset += read as u64; - if let Ok(value) = serde_json::from_str::(line.trim()) { - records.push(value); +/// Where this event's transcript records should be read from, in preference +/// order: the daemon's own mirror, then an inline snapshot (older journals), +/// then the live path. +fn read_event_records(event: &Envelope, path: &str, offset: &mut u64) -> Vec { + let import_through_offset = event + .payload + .get("_bt_import_through_offset") + .and_then(Value::as_u64); + + let mirror = event + .payload + .get("_bt_transcript_mirror") + .filter(|mirror| mirror.get("path").and_then(Value::as_str) == Some(path)); + if let Some(mirror) = mirror { + let mirror_path = mirror.get("mirror").and_then(Value::as_str); + let through = mirror.get("through").and_then(Value::as_u64); + if let (Some(mirror_path), Some(through)) = (mirror_path, through) { + // The mirror is byte-identical to the source prefix, so offsets + // are interchangeable between them. An import offset is the + // tighter bound when present. + let through = import_through_offset.unwrap_or(through); + let cutoff = import_through_offset.is_none().then_some(event.ts_ms); + if std::fs::metadata(mirror_path).is_ok() { + return read_file_bounded(mirror_path, offset, cutoff, Some(through)); + } } } - records -} -fn read_snapshot_until(contents: &str, offset: &mut u64, cutoff_ms: i64) -> Vec { - if *offset > contents.len() as u64 { - *offset = 0; - } - let mut reader = std::io::BufReader::new(std::io::Cursor::new(contents.as_bytes())); - if reader.seek(SeekFrom::Start(*offset)).is_err() { - return Vec::new(); + // Journals written before transcript mirroring carry the whole file. + let snapshot = event + .payload + .get("_bt_transcript_snapshot") + .filter(|snapshot| snapshot.get("path").and_then(Value::as_str) == Some(path)) + .and_then(|snapshot| snapshot.get("contents")) + .and_then(Value::as_str); + match (snapshot, import_through_offset) { + (Some(contents), through @ Some(_)) => { + read_snapshot_bounded(contents, offset, None, through) + } + (Some(contents), None) => read_snapshot_bounded(contents, offset, Some(event.ts_ms), None), + (None, through @ Some(_)) => read_file_bounded(path, offset, None, through), + (None, None) => read_file_bounded(path, offset, Some(event.ts_ms), None), } - read_buffered_until(&mut reader, offset, cutoff_ms) } fn tool_metadata( diff --git a/bt-daemon/tests/claude_translator.rs b/bt-daemon/tests/claude_translator.rs index c94cac1..b8e4710 100644 --- a/bt-daemon/tests/claude_translator.rs +++ b/bt-daemon/tests/claude_translator.rs @@ -10,7 +10,20 @@ fn fixture(name: &str) -> PathBuf { .join(name) } +/// How a replayed event points at its transcript bytes. +#[derive(Clone, Copy, PartialEq)] +enum Source { + /// The daemon's append-only mirror plus a high-water offset (current). + Mirror, + /// The whole transcript inlined into the event (pre-mirror journals). + Snapshot, +} + fn replay(name: &str) -> Vec { + replay_from(name, Source::Mirror) +} + +fn replay_from(name: &str, source: Source) -> Vec { let dir = fixture(name); let contents = std::fs::read_to_string(dir.join("events.ndjson")).unwrap(); let first: Value = serde_json::from_str(contents.lines().next().unwrap()).unwrap(); @@ -34,10 +47,23 @@ fn replay(name: &str) -> Vec { let local = dir.join("transcripts").join(basename); if local.exists() { payload[field] = json!(local.to_str().unwrap()); - payload["_bt_transcript_snapshot"] = json!({ - "path": local.to_str().unwrap(), - "contents": std::fs::read_to_string(&local).unwrap() - }); + let contents = std::fs::read_to_string(&local).unwrap(); + payload[match source { + Source::Mirror => "_bt_transcript_mirror", + Source::Snapshot => "_bt_transcript_snapshot", + }] = match source { + // The mirror is byte-identical to the source prefix, so + // the fixture file stands in for it directly. + Source::Mirror => json!({ + "path": local.to_str().unwrap(), + "mirror": local.to_str().unwrap(), + "through": contents.len() as u64, + }), + Source::Snapshot => json!({ + "path": local.to_str().unwrap(), + "contents": contents, + }), + }; } } let ts_ms = chrono::DateTime::parse_from_rfc3339(record["ts"].as_str().unwrap()) @@ -88,6 +114,32 @@ fn reduce(ops: Vec) -> HashMap { rows } +/// Journals recorded before transcript mirroring inline the whole transcript. +/// Both forms must translate identically, so upgrading the daemon neither +/// changes live output nor breaks recovery from an existing journal. +#[test] +fn mirror_and_inline_snapshot_transcripts_translate_identically() { + for name in ["test-fixture", "example-simple", "subagent-compact"] { + let mirrored = reduce(replay_from(name, Source::Mirror)); + let inlined = reduce(replay_from(name, Source::Snapshot)); + assert_eq!( + mirrored.len(), + inlined.len(), + "{name}: span count differs between mirror and inline snapshot" + ); + for (span_id, row) in &mirrored { + let other = inlined + .get(span_id) + .unwrap_or_else(|| panic!("{name}: {span_id} missing from the inline replay")); + assert_eq!( + serde_json::to_value(row).unwrap(), + serde_json::to_value(other).unwrap(), + "{name}: {span_id} differs between mirror and inline snapshot" + ); + } + } +} + #[test] fn claude_real_fixture_matches_session_turn_tool_and_token_contract() { let rows = reduce(replay("test-fixture")); diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index e4a4cd2..aedc19f 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -218,6 +218,7 @@ async fn start_routed_daemon( socket: Some(socket.clone()), data_dir: Some(data_dir.clone()), idle_timeout_secs: 0, + session_idle_timeout_secs: 0, }; let mut opts = debug_serve_options("test", &data_dir); opts.auth_provider = Some(provider); @@ -267,6 +268,7 @@ async fn start_daemon_at_with( socket: Some(socket.clone()), data_dir: Some(data_dir), idle_timeout_secs: 0, + session_idle_timeout_secs: 0, }; let opts = ServeOptions { version: "test".to_string(), @@ -333,6 +335,7 @@ async fn start_daemon() -> ( socket: Some(socket.clone()), data_dir: Some(data_dir.clone()), idle_timeout_secs: 0, // disable the watchdog for the test + session_idle_timeout_secs: 0, }; let mut opts = debug_serve_options("test", &data_dir); opts.auth_provider = Some(Arc::new(TestAuthProvider { @@ -375,6 +378,7 @@ async fn start_tracking_daemon( socket: Some(socket.clone()), data_dir: Some(data_dir), idle_timeout_secs: 0, + session_idle_timeout_secs: 0, }; let handle = tokio::spawn(async move { let _ = run_serve(args, opts).await; @@ -388,6 +392,7 @@ async fn start_daemon_at(data_dir: PathBuf, socket: PathBuf) -> tokio::task::Joi socket: Some(socket.clone()), data_dir: Some(data_dir.clone()), idle_timeout_secs: 0, + session_idle_timeout_secs: 0, }; let mut opts = debug_serve_options("test", &data_dir); opts.auth_provider = Some(Arc::new(TestAuthProvider { @@ -833,6 +838,7 @@ async fn a_second_server_detects_the_existing_daemon() { socket: Some(socket.clone()), data_dir: Some(data_dir.clone()), idle_timeout_secs: 0, + session_idle_timeout_secs: 0, }; let result = tokio::time::timeout( Duration::from_secs(2), @@ -975,7 +981,7 @@ async fn restart_replays_journal_with_stable_span_ids_before_new_events() { } #[tokio::test] -async fn claude_boundary_journal_contains_a_self_contained_transcript_snapshot() { +async fn claude_boundary_journal_references_a_self_contained_transcript_mirror() { let (data_dir, socket, handle, tmp) = start_daemon().await; let transcript = tmp.path().join("claude.jsonl"); std::fs::write( @@ -998,9 +1004,197 @@ async fn claude_boundary_journal_contains_a_self_contained_transcript_snapshot() .unwrap(); let journal = std::fs::read_to_string(data_dir.join("journal/claude-journal.ndjson")).unwrap(); - assert!(journal.contains("_bt_transcript_snapshot")); - assert!(journal.contains("durable")); assert!(!journal.contains("sk-TOP-SECRET-abc123")); + + // The journal references the mirror rather than inlining the transcript, + // so re-journaling a growing transcript stays linear in its size. + let entry: serde_json::Value = serde_json::from_str(journal.lines().next().unwrap()).unwrap(); + let reference = &entry["payload"]["_bt_transcript_mirror"]; + assert_eq!(reference["path"], transcript.to_str().unwrap()); + assert!( + !journal.contains("durable"), + "the journal must not inline transcript contents: {journal}" + ); + + // The mirror is daemon-owned and survives the original being rewritten. + let mirror = std::path::PathBuf::from(reference["mirror"].as_str().unwrap()); + assert!(mirror.starts_with(data_dir.join("transcripts"))); + let mirrored = std::fs::read_to_string(&mirror).unwrap(); + assert!(mirrored.contains("durable")); + assert_eq!( + reference["through"].as_u64().unwrap(), + mirrored.len() as u64, + "the journaled offset must bound replay to the bytes captured here" + ); + handle.abort(); +} + +/// Start a daemon that retires sessions after `ttl_secs` of quiet. +async fn start_daemon_with_session_ttl( + ttl_secs: u64, +) -> ( + PathBuf, + PathBuf, + tokio::task::JoinHandle<()>, + tempfile::TempDir, +) { + let tmp = tempfile::tempdir().unwrap(); + let data_dir = tmp.path().join("data"); + let socket = test_endpoint(tmp.path()); + std::fs::create_dir_all(&data_dir).unwrap(); + let args = ServeArgs { + socket: Some(socket.clone()), + data_dir: Some(data_dir.clone()), + idle_timeout_secs: 0, + session_idle_timeout_secs: ttl_secs, + }; + let mut opts = debug_serve_options("test", &data_dir); + opts.auth_provider = Some(Arc::new(TestAuthProvider { + calls: Mutex::new(Vec::new()), + fail: false, + first_lease_expired: false, + })); + let handle = tokio::spawn(async move { + let _ = run_serve(args, opts).await; + }); + wait_for(&socket).await; + (data_dir, socket, handle, tmp) +} + +fn claude_stop(session_id: &str, transcript: &Path, ts_ms: i64) -> Envelope { + let mut env = envelope(session_id, "Stop", ts_ms); + env.source = "claude-code".into(); + env.payload = serde_json::json!({ + "session_id": session_id, + "hook_event_name": "Stop", + "transcript_path": transcript, + }); + env +} + +/// The 20 GB crash: every lifecycle event used to journal the whole transcript, +/// so a session's journal grew with the square of its transcript and replay +/// had to hold all of it in memory at once. Journal growth must stay tied to +/// the *number* of events, not to the transcript size times that number. +#[tokio::test] +async fn claude_journal_does_not_grow_with_the_transcript_on_every_turn() { + let (data_dir, socket, handle, tmp) = start_daemon().await; + let host = dummy_host(); + let transcript = tmp.path().join("claude.jsonl"); + + const TURNS: i64 = 40; + let mut contents = String::new(); + for turn in 0..TURNS { + // Each turn appends a chunky assistant record, as a real session does. + contents.push_str(&format!( + r#"{{"type":"assistant","timestamp":"2026-07-29T00:00:00Z","message":{{"id":"m{turn}","model":"claude","content":[{{"type":"text","text":"{}"}}]}}}}"#, + "x".repeat(20_000) + )); + contents.push('\n'); + std::fs::write(&transcript, &contents).unwrap(); + forward_envelope( + &claude_stop("grow", &transcript, 1_775_000_000_000 + turn), + &socket, + &host, + false, + ) + .await + .unwrap(); + } + flush_session("grow", &socket, 5000).await.unwrap(); + + let transcript_len = std::fs::metadata(&transcript).unwrap().len(); + let journal_len = std::fs::metadata(data_dir.join("journal/grow.ndjson")) + .unwrap() + .len(); + + // Re-journaling the transcript every turn would put this near + // TURNS * transcript_len / 2 (tens of megabytes). References are ~a few + // hundred bytes each. + assert!( + journal_len < 64 * 1024, + "journal grew with the transcript: {journal_len} bytes for {TURNS} events \ + over a {transcript_len}-byte transcript" + ); + + // The transcript is still captured durably -- once, in the mirror. + let mirrors: Vec<_> = std::fs::read_dir(data_dir.join("transcripts")) + .unwrap() + .map(|entry| entry.unwrap().path()) + .collect(); + assert_eq!(mirrors.len(), 1, "one mirror per (session, transcript)"); + assert_eq!( + std::fs::metadata(&mirrors[0]).unwrap().len(), + transcript_len, + "the mirror must hold the whole transcript exactly once" + ); + handle.abort(); +} + +/// A session that goes quiet must release its translator state, sink, journal +/// handle, and credential lease instead of pinning them until the process +/// exits -- which, for a continuously busy user, never happened. +#[tokio::test] +async fn idle_sessions_are_retired_and_can_resume_from_their_journal() { + let (data_dir, socket, handle, _tmp) = start_daemon_with_session_ttl(1).await; + let host = dummy_host(); + forward_envelope(&envelope("nap", "SessionStart", 1), &socket, &host, false) + .await + .unwrap(); + flush_session("nap", &socket, 5000).await.unwrap(); + + let live = |socket: PathBuf| async move { + run_status(StatusArgs { + socket: Some(socket), + session_id: Some("nap".into()), + }) + .await + .unwrap() + .unwrap() + .sessions + .len() + }; + assert_eq!(live(socket.clone()).await, 1, "session should be live"); + + let mut retired = false; + for _ in 0..100 { + tokio::time::sleep(Duration::from_millis(100)).await; + if live(socket.clone()).await == 0 { + retired = true; + break; + } + } + assert!(retired, "an idle session was never retired"); + + // Retirement is not data loss: a later event rebuilds the session from its + // journal, and deterministic span ids merge the re-emitted rows. + forward_envelope(&envelope("nap", "Stop", 2), &socket, &host, false) + .await + .unwrap(); + flush_session("nap", &socket, 5000).await.unwrap(); + let spans = std::fs::read_to_string(data_dir.join("spans/nap.ndjson")).unwrap(); + let ids: Vec = spans + .lines() + .map(|line| { + let row: serde_json::Value = serde_json::from_str(line).unwrap(); + row.get("Insert") + .or_else(|| row.get("Merge")) + .and_then(|body| body.get("span_id")) + .and_then(serde_json::Value::as_str) + .unwrap() + .to_owned() + }) + .collect(); + assert_eq!( + ids.len(), + 5, + "first delivery (2) + replay after retirement (2) + resumed event (1)" + ); + assert_eq!( + ids.iter().collect::>().len(), + 3, + "replay after retirement must reuse the original span ids, not mint new ones" + ); handle.abort(); }