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