Isolate routing submission stage

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
 

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::<super::routing::EmitFailure>() {
+        if let Some(partial) = error.downcast_ref::<super::recovery::RouteFailure>() {
             self.counters
                 .spans_emitted
                 .fetch_add(partial.emitted, Ordering::Relaxed);
@@ -161,7 +161,7 @@ impl ConsumerOptions {
                     )
                 })?;
                 sink.flush().await.map_err(|error| {
-                    if error.downcast_ref::<CommitFailure>().is_some() {
+                    if error.chain().any(|cause| cause.is::<CommitFailure>()) {
                         self.storage(&error, attempt, Work::Flush)
                     } else {
                         self.routing(&error, config, attempt)

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

bt-daemon/src/pipeline/plugins.rs

@@ -11,7 +11,7 @@ pub(crate) enum PluginOutcome {
 }
 
 pub(crate) struct Plugins {
-    pub paths: Vec<PathBuf>,
+    paths: Vec<PathBuf>,
     pub source: String,
     pub session_id: String,
     blocked: Option<(Vec<PathBuf>, 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]

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<Option<[u8; 32]>>,
+    },
+    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::<super::routing::EmitFailure>()
-            .map_or(error, |failure| &failure.cause);
-        let mut failure = Self::delay(Boundary::Sink, 0, format!("{error:#}"), attempt);
-        if let Some(plugin) = error.downcast_ref::<PluginRejected>() {
-            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::<braintrust_sdk_rust::BraintrustError>() {
-            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::<RouteFailure>();
+        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, "<redacted>");