From ea90ed6bb8e6c5a0345ace3409d38f339ac360f6 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Thu, 15 Oct 2026 20:57:18 +0800 Subject: [PATCH 1/4] Isolate routing submission stage --- bt-daemon/src/pipeline/mod.rs | 1 + bt-daemon/src/pipeline/plugins.rs | 7 ++- bt-daemon/src/pipeline/routing.rs | 73 ++++++++++++++++---------- bt-daemon/src/pipeline/submission.rs | 76 ++++++++++++++++++++++++++++ 4 files changed, 129 insertions(+), 28 deletions(-) create mode 100644 bt-daemon/src/pipeline/submission.rs diff --git a/bt-daemon/src/pipeline/mod.rs b/bt-daemon/src/pipeline/mod.rs index 19ee0f93..371bd9bb 100644 --- a/bt-daemon/src/pipeline/mod.rs +++ b/bt-daemon/src/pipeline/mod.rs @@ -4,6 +4,7 @@ pub(crate) mod delivery; pub(crate) mod plugins; pub(crate) mod prepare; pub(crate) mod routing; +pub(crate) mod submission; pub(crate) use delivery::{Completion, Delivery}; diff --git a/bt-daemon/src/pipeline/plugins.rs b/bt-daemon/src/pipeline/plugins.rs index 9fa2e1cd..322aff18 100644 --- a/bt-daemon/src/pipeline/plugins.rs +++ b/bt-daemon/src/pipeline/plugins.rs @@ -11,7 +11,7 @@ pub(crate) enum PluginOutcome { } pub(crate) struct Plugins { - pub paths: Vec, + paths: Vec, pub source: String, pub session_id: String, blocked: Option<(Vec, PluginRejected)>, @@ -44,6 +44,11 @@ impl Plugins { blocked: None, } } + + pub fn configure(&mut self, paths: &[PathBuf]) { + self.paths.clear(); + self.paths.extend_from_slice(paths); + } } #[async_trait::async_trait] diff --git a/bt-daemon/src/pipeline/routing.rs b/bt-daemon/src/pipeline/routing.rs index 86efe4cb..d8b23608 100644 --- a/bt-daemon/src/pipeline/routing.rs +++ b/bt-daemon/src/pipeline/routing.rs @@ -1,6 +1,10 @@ //! Routing owns its sink behind a bounded typed port. Session dispatch is its facade. use super::delivery::{Delivery, InputPort, OutputPort}; use super::plugins::{PluginOutcome, Plugins}; +use super::prepare::Prepare; +use super::submission::{ + Input as SubmissionInput, Operation, Outcome as SubmissionOutcome, Submission, +}; use super::{Completion, Stage}; use crate::{sink::Sink, translate::SpanOp, wire::SessionConfig}; use parking_lot::Mutex; @@ -12,11 +16,6 @@ struct Request { operation: Operation, reply: oneshot::Sender, } - -enum Operation { - Emit(Vec), - Flush, -} struct Response { result: anyhow::Result, pending: bool, @@ -58,7 +57,7 @@ pub(crate) struct RoutingSink { impl RoutingSink { pub fn spawn( - mut sink: Box, + sink: Box, source: String, session_id: String, data_dir: PathBuf, @@ -66,21 +65,21 @@ impl RoutingSink { ) -> Self { let (tx, mut rx) = mpsc::channel::>(16); tokio::spawn(async move { - let mut stage = Plugins::new(&source, &session_id); - let mut prepare = super::prepare::Prepare::new(&source); + let mut plugins = Plugins::new(&source, &session_id); + let mut prepare = Prepare::new(&source); + let mut submission = Submission::new(sink); while let Ok(Some(delivery)) = InputPort::receive(&mut rx).await { let request = delivery.value(); - sink.configure(&request.config); - let result = match &request.operation { + let (result, state) = match &request.operation { Operation::Emit(ops) => { let mut ops = ops.clone(); prepare.apply(&mut ops, &request.config); - stage.paths = request.config.span_plugins.clone(); + plugins.configure(&request.config.span_plugins); let mut prepared = Vec::with_capacity(ops.len()); let mut failure = None; let mut report = false; for op in &ops { - match stage.process(op).await { + match plugins.process(op).await { PluginOutcome::Prepared(op) => prepared.push(*op), PluginOutcome::Filtered => {} PluginOutcome::Failed(error) => { @@ -103,23 +102,43 @@ impl RoutingSink { } } } - let emitted = if prepared.is_empty() { - Ok(0) - } else { - sink.emit(&prepared).await + let submit_input = SubmissionInput { + config: request.config.clone(), + operation: Operation::Emit(prepared), + }; + let SubmissionOutcome::Emitted { + result: emitted, + state, + } = submission.process(&submit_input).await + else { + unreachable!("emit request returned a flush outcome") }; match (emitted, failure) { - (Err(error), _) => Err(error), - (Ok(emitted), Some(cause)) => Err(EmitFailure { - emitted, - report, - cause, - } - .into()), - (Ok(emitted), None) => Ok(emitted), + (Err(error), _) => (Err(error), state), + (Ok(emitted), Some(cause)) => ( + Err(EmitFailure { + emitted, + report, + cause, + } + .into()), + state, + ), + (Ok(emitted), None) => (Ok(emitted), state), } } - Operation::Flush => sink.flush().await.map(|()| 0), + Operation::Flush => { + let submit_input = SubmissionInput { + config: request.config.clone(), + operation: Operation::Flush, + }; + let SubmissionOutcome::Flushed { result, state } = + submission.process(&submit_input).await + else { + unreachable!("flush request returned an emit outcome") + }; + (result.map(|()| 0), state) + } }; if let Err(error) = &result { *last_error.lock() = Some(error.to_string()); @@ -136,8 +155,8 @@ impl RoutingSink { }; let _ = request.reply.send(Response { result, - pending: sink.has_pending_delivery(), - permalink: sink.permalink(), + pending: state.pending, + permalink: state.permalink, }); } }); diff --git a/bt-daemon/src/pipeline/submission.rs b/bt-daemon/src/pipeline/submission.rs new file mode 100644 index 00000000..f9870e80 --- /dev/null +++ b/bt-daemon/src/pipeline/submission.rs @@ -0,0 +1,76 @@ +//! The route-local sink adapter is one stage in the routing container. +use super::Stage; +use crate::{sink::Sink, translate::SpanOp, wire::SessionConfig}; + +pub(crate) enum Operation { + Emit(Vec), + Flush, +} + +pub(crate) struct Input { + pub config: SessionConfig, + pub operation: Operation, +} + +pub(crate) struct State { + pub pending: bool, + pub permalink: Option, +} + +pub(crate) enum Outcome { + Emitted { + result: anyhow::Result, + state: State, + }, + Flushed { + result: anyhow::Result<()>, + state: State, + }, +} + +pub(crate) struct Submission { + sink: Box, +} + +impl Submission { + pub fn new(sink: Box) -> Self { + Self { sink } + } + + fn state(&self) -> State { + State { + pending: self.sink.has_pending_delivery(), + permalink: self.sink.permalink(), + } + } +} + +#[async_trait::async_trait] +impl Stage for Submission { + type Input = Input; + type Outcome = Outcome; + + async fn process(&mut self, input: &Self::Input) -> Self::Outcome { + self.sink.configure(&input.config); + match &input.operation { + Operation::Emit(ops) => { + let result = if ops.is_empty() { + Ok(0) + } else { + self.sink.emit(ops).await + }; + Outcome::Emitted { + result, + state: self.state(), + } + } + Operation::Flush => { + let result = self.sink.flush().await; + Outcome::Flushed { + result, + state: self.state(), + } + } + } + } +} From 6bcd8a40b818fc1609212c55527b5aff7d3a8590 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Thu, 15 Oct 2026 21:05:12 +0800 Subject: [PATCH 2/4] Classify route failures at stage boundaries --- bt-daemon/src/pipeline/consumer.rs | 2 +- bt-daemon/src/pipeline/recovery.rs | 117 +++++++++++++++++---------- bt-daemon/src/pipeline/routing.rs | 66 ++++++++------- bt-daemon/src/pipeline/submission.rs | 57 +++++++++++-- 4 files changed, 164 insertions(+), 78 deletions(-) diff --git a/bt-daemon/src/pipeline/consumer.rs b/bt-daemon/src/pipeline/consumer.rs index fec51b26..21cede50 100644 --- a/bt-daemon/src/pipeline/consumer.rs +++ b/bt-daemon/src/pipeline/consumer.rs @@ -123,7 +123,7 @@ impl ConsumerOptions { } fn routing(&self, error: &anyhow::Error, config: &SessionConfig, attempt: u32) -> Held { - if let Some(partial) = error.downcast_ref::() { + if let Some(partial) = error.downcast_ref::() { self.counters .spans_emitted .fetch_add(partial.emitted, Ordering::Relaxed); diff --git a/bt-daemon/src/pipeline/recovery.rs b/bt-daemon/src/pipeline/recovery.rs index 8030bf10..82147924 100644 --- a/bt-daemon/src/pipeline/recovery.rs +++ b/bt-daemon/src/pipeline/recovery.rs @@ -1,5 +1,5 @@ //! Failure outcomes wait for explicit evidence before re-entering processing. -use super::{plugins::PluginRejected, Stage}; +use super::Stage; use crate::wire::{RecoveryInfo, SessionConfig}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -55,6 +55,50 @@ pub(crate) enum Condition { }, } +/// Stage adapters classify failures before handing them to the shared recovery +/// state. The retry runner consumes this value without knowing SDK/plugin types. +#[derive(Debug, Clone)] +pub(crate) enum RetryCondition { + Plugin { + path: PathBuf, + digest: Option<[u8; 32]>, + processors: Vec>, + }, + Credentials, + Configuration { + recheck: bool, + }, + Transient, +} + +pub(crate) struct RouteFailure { + pub cause: anyhow::Error, + pub retry: RetryCondition, + pub emitted: u64, + pub report: bool, +} + +impl std::fmt::Display for RouteFailure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.cause.fmt(f) + } +} +impl std::fmt::Debug for RouteFailure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RouteFailure") + .field("cause", &self.cause.to_string()) + .field("retry", &self.retry) + .field("emitted", &self.emitted) + .field("report", &self.report) + .finish() + } +} +impl std::error::Error for RouteFailure { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.cause.as_ref()) + } +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub(crate) struct Failure { pub boundary: Boundary, @@ -130,49 +174,36 @@ impl Failure { } } pub fn routing(error: &anyhow::Error, config: &SessionConfig, attempt: u32) -> Self { - let error = error - .downcast_ref::() - .map_or(error, |failure| &failure.cause); - let mut failure = Self::delay(Boundary::Sink, 0, format!("{error:#}"), attempt); - if let Some(plugin) = error.downcast_ref::() { - failure.boundary = Boundary::Plugin; - failure.condition = Condition::Plugin { - path: plugin.failure.path.clone(), - digest: plugin.digest, - chain: chain(config), - processors: Some(plugin.processors.clone()), - }; - } else if let Some(error) = error.downcast_ref::() { - use braintrust_sdk_rust::BraintrustError; - match error { - BraintrustError::Api { status: 401, .. } => { - failure.boundary = Boundary::Credentials; - failure.condition = Condition::Credentials { - recheck_ms: now_ms().saturating_add(1_000), - }; - } - BraintrustError::Api { - status: 403 | 404, .. - } => { - failure.condition = Condition::Configuration { - signature: configuration(config), - recheck_ms: Some(now_ms().saturating_add(60_000)), - } - } - BraintrustError::Api { - status: 408 | 429 | 500..=599, - .. - } - | BraintrustError::Http(_) - | BraintrustError::Network(_) - | BraintrustError::ChannelClosed => {} - _ => { - failure.condition = Condition::Configuration { - signature: configuration(config), - recheck_ms: None, - } - } + let classified = error.downcast_ref::(); + let cause = classified.map_or(error, |failure| &failure.cause); + let mut failure = Self::delay(Boundary::Sink, 0, format!("{cause:#}"), attempt); + match classified.map(|failure| &failure.retry) { + Some(RetryCondition::Plugin { + path, + digest, + processors, + }) => { + failure.boundary = Boundary::Plugin; + failure.condition = Condition::Plugin { + path: path.clone(), + digest: *digest, + chain: chain(config), + processors: Some(processors.clone()), + }; + } + Some(RetryCondition::Credentials) => { + failure.boundary = Boundary::Credentials; + failure.condition = Condition::Credentials { + recheck_ms: now_ms().saturating_add(1_000), + }; + } + Some(RetryCondition::Configuration { recheck }) => { + failure.condition = Condition::Configuration { + signature: configuration(config), + recheck_ms: recheck.then(|| now_ms().saturating_add(60_000)), + }; } + Some(RetryCondition::Transient) | None => {} } if !config.auth.token.is_empty() { failure.reason = failure.reason.replace(&config.auth.token, ""); diff --git a/bt-daemon/src/pipeline/routing.rs b/bt-daemon/src/pipeline/routing.rs index d8b23608..610a7d96 100644 --- a/bt-daemon/src/pipeline/routing.rs +++ b/bt-daemon/src/pipeline/routing.rs @@ -2,6 +2,7 @@ use super::delivery::{Delivery, InputPort, OutputPort}; use super::plugins::{PluginOutcome, Plugins}; use super::prepare::Prepare; +use super::recovery::{RetryCondition, RouteFailure}; use super::submission::{ Input as SubmissionInput, Operation, Outcome as SubmissionOutcome, Submission, }; @@ -22,24 +23,6 @@ struct Response { permalink: Option, } -/// Rows accepted before a mandatory plugin failed still count as emitted. -#[derive(Debug)] -pub(crate) struct EmitFailure { - pub emitted: u64, - pub report: bool, - pub cause: anyhow::Error, -} -impl std::fmt::Display for EmitFailure { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - self.cause.fmt(f) - } -} -impl std::error::Error for EmitFailure { - fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { - Some(self.cause.as_ref()) - } -} - struct Forward; #[async_trait::async_trait] impl Completion for Forward { @@ -94,11 +77,18 @@ impl RoutingSink { tracing::warn!(%diagnostic, "plugin diagnostic could not be saved"); } } - failure = Some(anyhow::Error::new(error)); + failure = Some(( + anyhow::Error::new(error.clone()), + RetryCondition::Plugin { + path: error.failure.path.clone(), + digest: error.digest, + processors: error.processors.clone(), + }, + )); } PluginOutcome::Unavailable(error) => { report = true; - failure = Some(error); + failure = Some((error, RetryCondition::Transient)); } } } @@ -114,14 +104,22 @@ impl RoutingSink { unreachable!("emit request returned a flush outcome") }; match (emitted, failure) { - (Err(error), _) => (Err(error), state), - (Ok(emitted), Some(cause)) => ( - Err(EmitFailure { + (Err(error), _) => ( + Err(anyhow::Error::new(RouteFailure { + cause: error.cause, + retry: error.retry, + emitted: 0, + report: false, + })), + state, + ), + (Ok(emitted), Some((cause, retry))) => ( + Err(anyhow::Error::new(RouteFailure { + cause, + retry, emitted, report, - cause, - } - .into()), + })), state, ), (Ok(emitted), None) => (Ok(emitted), state), @@ -137,7 +135,17 @@ impl RoutingSink { else { unreachable!("flush request returned an emit outcome") }; - (result.map(|()| 0), state) + ( + result.map(|()| 0).map_err(|error| { + anyhow::Error::new(RouteFailure { + cause: error.cause, + retry: error.retry, + emitted: 0, + report: false, + }) + }), + state, + ) } }; if let Err(error) = &result { @@ -268,12 +276,12 @@ mod tests { }) }; let failure = sink.emit(&[op("accept"), op("reject")]).await.unwrap_err(); - let first = failure.downcast_ref::().unwrap(); + let first = failure.downcast_ref::().unwrap(); assert_eq!(first.emitted, 1); assert!(first.report); assert_eq!(emitted.load(Ordering::Relaxed), 1); let failure = sink.emit(&[op("accept")]).await.unwrap_err(); - let cached = failure.downcast_ref::().unwrap(); + let cached = failure.downcast_ref::().unwrap(); assert_eq!(cached.emitted, 0); assert!( !cached.report, diff --git a/bt-daemon/src/pipeline/submission.rs b/bt-daemon/src/pipeline/submission.rs index f9870e80..5252163f 100644 --- a/bt-daemon/src/pipeline/submission.rs +++ b/bt-daemon/src/pipeline/submission.rs @@ -1,5 +1,5 @@ //! The route-local sink adapter is one stage in the routing container. -use super::Stage; +use super::{recovery::RetryCondition, Stage}; use crate::{sink::Sink, translate::SpanOp, wire::SessionConfig}; pub(crate) enum Operation { @@ -19,15 +19,62 @@ pub(crate) struct State { pub(crate) enum Outcome { Emitted { - result: anyhow::Result, + result: Result, state: State, }, Flushed { - result: anyhow::Result<()>, + result: Result<(), Failure>, state: State, }, } +pub(crate) struct Failure { + pub cause: anyhow::Error, + pub retry: RetryCondition, +} + +impl std::fmt::Display for Failure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.cause.fmt(f) + } +} +impl std::fmt::Debug for Failure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SubmissionFailure") + .field("cause", &self.cause.to_string()) + .field("retry", &self.retry) + .finish() + } +} +impl std::error::Error for Failure { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.cause.as_ref()) + } +} + +impl Failure { + fn classify(cause: anyhow::Error) -> Self { + let retry = match cause.downcast_ref::() { + Some(braintrust_sdk_rust::BraintrustError::Api { status: 401, .. }) => { + RetryCondition::Credentials + } + Some(braintrust_sdk_rust::BraintrustError::Api { + status: 403 | 404, .. + }) => RetryCondition::Configuration { recheck: true }, + Some(braintrust_sdk_rust::BraintrustError::Api { + status: 408 | 429 | 500..=599, + .. + }) + | Some(braintrust_sdk_rust::BraintrustError::Http(_)) + | Some(braintrust_sdk_rust::BraintrustError::Network(_)) + | Some(braintrust_sdk_rust::BraintrustError::ChannelClosed) + | None => RetryCondition::Transient, + Some(_) => RetryCondition::Configuration { recheck: false }, + }; + Self { cause, retry } + } +} + pub(crate) struct Submission { sink: Box, } @@ -57,7 +104,7 @@ impl Stage for Submission { let result = if ops.is_empty() { Ok(0) } else { - self.sink.emit(ops).await + self.sink.emit(ops).await.map_err(Failure::classify) }; Outcome::Emitted { result, @@ -65,7 +112,7 @@ impl Stage for Submission { } } Operation::Flush => { - let result = self.sink.flush().await; + let result = self.sink.flush().await.map_err(Failure::classify); Outcome::Flushed { result, state: self.state(), From 85eb4c7fad0b40fd16d0096b7d73803a72d93616 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Sat, 17 Oct 2026 00:54:37 +0800 Subject: [PATCH 3/4] Preserve completion-only retry through stage error wrappers --- bt-daemon/src/pipeline/consumer.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bt-daemon/src/pipeline/consumer.rs b/bt-daemon/src/pipeline/consumer.rs index 21cede50..0dc416f0 100644 --- a/bt-daemon/src/pipeline/consumer.rs +++ b/bt-daemon/src/pipeline/consumer.rs @@ -161,7 +161,7 @@ impl ConsumerOptions { ) })?; sink.flush().await.map_err(|error| { - if error.downcast_ref::().is_some() { + if error.chain().any(|cause| cause.is::()) { self.storage(&error, attempt, Work::Flush) } else { self.routing(&error, config, attempt) From a4ad11f22eb6b188fbdbc9041f9bad43ddc94c16 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Sat, 17 Oct 2026 02:35:55 +0800 Subject: [PATCH 4/4] Document durable SpanOp recovery and route completion --- bt-daemon/README.md | 53 ++++++++++++++++++++++++++++----------------- 1 file changed, 33 insertions(+), 20 deletions(-) diff --git a/bt-daemon/README.md b/bt-daemon/README.md index bf3aa57b..941b2ac5 100644 --- a/bt-daemon/README.md +++ b/bt-daemon/README.md @@ -158,10 +158,10 @@ bt trace run --plugin ./local.mjs codex -- "summarize this change" bt trace import codex SESSION_ID --plugin ./sanitize-history.mjs ``` -The journal stores raw input events, not transformed spans. After daemon -recovery, replayed events therefore pass through the resumed session's current -route: ordinary sessions use the current globally configured plugins, while a -managed session continues using only that run's isolated plugins. +The daemon journals raw input events and canonical SpanOps produced by translation. +Route settings and plugins are applied after SpanOp persistence, so routing retries +reuse the saved operations without rerunning translation. Ordinary sessions use +their configured plugins; managed sessions retain their run's isolated plugin chain. `context.operation` is `"insert"` or `"merge"`; `context.source` and `context.session_id` identify the translated event stream; and `context.env` @@ -173,12 +173,12 @@ The environment map is captured from the daemon process when each worker-local span processor is constructed. Plugins execute in bounded, thread-local QuickJS runtimes with no filesystem or network host APIs. Modules must be self-contained and transforms must be stateless: module globals belong to a -worker thread, not a session. Every configured plugin is mandatory. If any -plugin fails, the daemon discards that span operation instead of delivering -untransformed data, and that worker continues discarding operations that would -use the failed plugin. Modifying the plugin file causes workers to retry it. The -raw event remains journaled, so restarting the daemon after fixing the plugin -replays the withheld data through the current chain. +worker thread, not a session. Every configured plugin is mandatory. A plugin +failure holds the affected route at its earliest unresolved SpanOp; other routes +continue independently. The daemon persists the failure and waits for the failed +file or another plugin in its configured chain to change, or for the chain itself +to change. It then retries from the SpanOp journal without requiring a restart or +another agent event. Restarting alone does not retry an unchanged failing chain. Plugin failures are deduplicated in the daemon's private local state, including the raw QuickJS exception and stack. Inspect them with: @@ -297,9 +297,17 @@ different profiles, organizations, projects, experiments, or parent spans. ## Recovery and storage Capture requests return after the event is flushed to its journal. Daemon -workers handle authentication, translation, and delivery. Restart recovery -replays journaled events with deterministic span IDs, so repeated delivery -updates the same spans. +workers handle authentication, translation, and delivery. Translation commits +canonical SpanOps and its consumed event position to a SQLite journal. Each route +has an independent durable head that advances only after checked sink acceptance +or intentional filtering. Plugin, authentication, and delivery recovery resumes +from unresolved SpanOps without replaying translation. Retries use deterministic +span IDs, so repeated delivery updates the same spans. + +Failures retain their reason and retry condition, including across restarts. +The daemon waits for the relevant plugin/configuration change, fresh credentials, +or retry delay before trying again. `bt trace doctor` reports the affected stage, +pending position, reason, and retry eligibility. Claude, Codex, and Cursor events reference daemon-owned transcript mirrors. Cursor preserves separate immutable generations when its native transcript is rewritten. @@ -308,13 +316,18 @@ separate updates and events mirrors. Replay does not depend on the original transcript staying at its old path. The daemon streams journals and transcripts, applies queue backpressure, and -retires idle sessions. Retired sessions are rebuilt from their journals when -another event arrives. Journals and mirrors preserve conversation content; -credentials are excluded from journaled routing data. - -Recovery files are removed after seven days without modification. Cleanup runs -at startup and hourly while the daemon is running. This retention period is -currently fixed in `src/server.rs`. +retires idle sessions. Retirement preserves unresolved work and route heads; +startup resumes pending subscriptions without waiting for another event. +Translators rebuild their in-memory state from committed input when more +translation is needed, suppressing already committed output. Journals and mirrors +preserve conversation content; credentials are excluded from journaled routing data. + +Completed, inactive session partitions become eligible for cleanup after seven +days without modification. Cleanup also requires translation to be caught up, +all captured routes to have registered heads, every head to be complete, and no +pending recovery. Retained event logs protect the transcript mirrors and +correlation state needed for reconstruction. Cleanup runs at startup and hourly; +the retention period is fixed in `src/server/maintenance.rs`. ## Platform and integration coverage