Fix recovery diagnostics and resume held session delivery

bt-daemon/README.md

@@ -314,6 +314,12 @@ 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.
 
+A later event with corrected delivery settings also resumes configuration-held
+subscriptions for the same source session, authentication selection, and
+destination. Each subscription keeps its original durable head and pinned backend;
+the correction is saved for restart recovery. Healthy routes and other destinations
+retain their own settings.
+
 Claude, Codex, and Cursor events reference daemon-owned transcript mirrors. Cursor
 preserves separate immutable generations when its native transcript is rewritten.
 Grok uses

bt-daemon/src/pipeline/journal.rs

@@ -83,7 +83,7 @@ impl SpanJournal {
             let db = Connection::open(path)?;
             db.busy_timeout(std::time::Duration::from_secs(5))?;
             let version: u32 = db.query_row("PRAGMA user_version", [], |row| row.get(0))?;
-            anyhow::ensure!(version <= 4, "unsupported SpanOp journal schema {version}");
+            anyhow::ensure!(version <= 5, "unsupported SpanOp journal schema {version}");
             db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;")?;
             if version < 3 {
                 db.execute_batch(
@@ -112,6 +112,9 @@ impl SpanJournal {
             if version < 4 {
                 db.execute_batch("BEGIN IMMEDIATE; CREATE TABLE root_candidates (span_id TEXT PRIMARY KEY, value BLOB NOT NULL); PRAGMA user_version=4; COMMIT")?;
             }
+            if version < 5 {
+                db.execute_batch("BEGIN IMMEDIATE; ALTER TABLE heads ADD COLUMN settings TEXT; ALTER TABLE heads ADD COLUMN settings_through INTEGER NOT NULL DEFAULT 0; PRAGMA user_version=5; COMMIT")?;
+            }
             let tail = db.query_row("SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) WHERE committed=1", [], |row| row.get(0))?;
             Ok((db, tail))
         })
@@ -424,6 +427,90 @@ impl SpanJournal {
         })
         .await
     }
+    /// Keep the original subscription and backend binding while correcting settings
+    /// that can unblock an explicitly configuration-held delivery.
+    pub async fn reconfigure_held(
+        &self,
+        route: &SessionRoute,
+        through: u64,
+    ) -> anyhow::Result<Vec<(SessionRoute, SessionRoute)>> {
+        let route = route.clone();
+        self.with_db(move |db| {
+            let tx = db.transaction()?;
+            let updates = {
+                let mut query = tx.prepare(
+                    "SELECT route,failure,settings FROM heads WHERE failure IS NOT NULL AND settings_through<?",
+                )?;
+                let rows = query.query_map([through], |row| {
+                    Ok((
+                        row.get::<_, String>(0)?,
+                        row.get::<_, String>(1)?,
+                        row.get::<_, Option<String>>(2)?,
+                    ))
+                })?;
+                let mut updates = Vec::new();
+                for row in rows {
+                    let (key, raw, settings) = row?;
+                    let stored: SessionRoute = serde_json::from_str(&key)?;
+                    let Ok(failure) = serde_json::from_str::<Failure>(&raw) else {
+                        continue;
+                    };
+                    if !matches!(
+                        failure.condition,
+                        super::recovery::Condition::Configuration { .. } | super::recovery::Condition::Plugin { .. }
+                    ) || !stored.same_destination(&route)
+                    {
+                        continue;
+                    }
+                    let previous: SessionRoute = settings
+                        .as_deref()
+                        .map(serde_json::from_str)
+                        .transpose()?
+                        .unwrap_or_else(|| stored.clone());
+                    if previous.same_route(&route) {
+                        continue;
+                    }
+                    let mut corrected = route.clone();
+                    // Preserve the subscription's pinned credential selection.
+                    corrected.auth = stored.auth.clone();
+                    updates.push((key, stored, corrected));
+                }
+                updates
+            };
+            for (key, _, corrected) in &updates {
+                tx.execute(
+                    "UPDATE heads SET settings=?, settings_through=? WHERE route=?",
+                    params![serde_json::to_string(corrected)?, through, key],
+                )?;
+            }
+            tx.commit()?;
+            Ok(updates
+                .into_iter()
+                .map(|(_, stored, corrected)| (stored, corrected))
+                .collect())
+        })
+        .await
+    }
+
+    pub async fn route_settings(&self, route: &SessionRoute) -> anyhow::Result<SessionRoute> {
+        let fallback = route.clone();
+        let key = serde_json::to_string(route)?;
+        self.with_db(move |db| {
+            let settings: Option<String> = db
+                .query_row("SELECT settings FROM heads WHERE route=?", [key], |row| {
+                    row.get(0)
+                })
+                .optional()?
+                .flatten();
+            Ok(settings
+                .as_deref()
+                .map(serde_json::from_str)
+                .transpose()?
+                .unwrap_or(fallback))
+        })
+        .await
+    }
+
     pub async fn routes(&self) -> anyhow::Result<Vec<SessionRoute>> {
         self.with_db(|db| {
             let mut query = db.prepare("SELECT route FROM heads ORDER BY route")?;
@@ -1142,6 +1229,96 @@ mod tests {
             .await
             .unwrap();
     }
+    #[tokio::test]
+    async fn corrected_settings_survive_restart_without_advancing_or_retargeting_heads() {
+        let tmp = tempfile::tempdir().unwrap();
+        let path = tmp.path().join("spans.sqlite");
+        let journal = SpanJournal::open(&path).await.unwrap();
+        let original = SessionRoute {
+            destination: Some(crate::wire::TraceDestination::ProjectLogs {
+                project_id: Some("test-project".into()),
+                project_name: None,
+            }),
+            ..route("a")
+        };
+        let cursor = journal.subscribe(&original, 0).await.unwrap();
+        let failure = Failure {
+            boundary: Boundary::Sink,
+            position: 1,
+            through: None,
+            reason: "rejected row".into(),
+            attempt: 0,
+            condition: super::super::recovery::Condition::Configuration {
+                signature: "original".into(),
+                recheck_ms: None,
+            },
+        };
+        let raw = serde_json::to_string(&failure).unwrap();
+        journal
+            .set_failure(&cursor.route, Some(raw.clone()))
+            .await
+            .unwrap();
+        let mut corrected = original.clone();
+        corrected.tags.push("corrected".into());
+        for unrelated in [
+            SessionRoute {
+                destination: Some(crate::wire::TraceDestination::ProjectLogs {
+                    project_id: Some("other-project".into()),
+                    project_name: None,
+                }),
+                ..corrected.clone()
+            },
+            SessionRoute {
+                auth: crate::wire::AuthSelection {
+                    org_name: Some("other-org".into()),
+                    ..Default::default()
+                },
+                ..corrected.clone()
+            },
+        ] {
+            assert!(journal
+                .reconfigure_held(&unrelated, 20)
+                .await
+                .unwrap()
+                .is_empty());
+        }
+        assert_eq!(
+            journal
+                .reconfigure_held(&corrected, 20)
+                .await
+                .unwrap()
+                .len(),
+            1
+        );
+        assert_eq!(cursor.completed, 0);
+        assert_eq!(journal.failure(&cursor.route).await.unwrap(), Some(raw));
+        // A captured prefix must not revert a newer correction during startup replay.
+        assert!(journal
+            .reconfigure_held(&original, 10)
+            .await
+            .unwrap()
+            .is_empty());
+        journal.close().await.unwrap();
+        let restored = SpanJournal::open(&path).await.unwrap();
+        assert_eq!(
+            restored.route_settings(&original).await.unwrap().tags,
+            corrected.tags
+        );
+        assert_eq!(restored.routes().await.unwrap().len(), 1);
+        assert_eq!(restored.subscribe(&original, 0).await.unwrap().completed, 0);
+        restored.set_failure(&cursor.route, None).await.unwrap();
+        let mut newer = corrected.clone();
+        newer.tags.push("independent".into());
+        assert!(
+            restored
+                .reconfigure_held(&newer, 30)
+                .await
+                .unwrap()
+                .is_empty(),
+            "healthy independent subscriptions retain their settings"
+        );
+    }
+
     #[tokio::test]
     async fn close_releases_storage_even_when_a_blocking_read_was_cancelled() {
         let tmp = tempfile::tempdir().unwrap();

bt-daemon/src/pipeline/translation.rs

@@ -172,6 +172,18 @@ impl Translation {
             .map_err(|_| "translation control lost".to_owned())?
     }
 
+    pub async fn reconfigure_held(
+        &self,
+        route: &SessionRoute,
+        through: u64,
+    ) -> anyhow::Result<Vec<(SessionRoute, SessionRoute)>> {
+        self.journal.reconfigure_held(route, through).await
+    }
+
+    pub async fn route_settings(&self, route: &SessionRoute) -> anyhow::Result<SessionRoute> {
+        self.journal.route_settings(route).await
+    }
+
     /// Connect a route to the committed SpanOp stream without exposing its
     /// storage adapter to the route actor.
     pub async fn subscribe(

bt-daemon/src/server/correlation_store.rs

@@ -89,6 +89,9 @@ pub(super) struct PendingEvent {
     pub(super) env: Envelope,
     pub(super) replay_through: u64,
     pub(super) journal_through: u64,
+    /// Original capture position when startup extends the replay boundary.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(super) settings_through: Option<u64>,
 }
 
 impl<'de> Deserialize<'de> for PendingEvent {
@@ -103,6 +106,8 @@ impl<'de> Deserialize<'de> for PendingEvent {
                 env: Envelope,
                 replay_through: u64,
                 journal_through: u64,
+                #[serde(default)]
+                settings_through: Option<u64>,
             },
             Legacy(Envelope),
         }
@@ -111,15 +116,18 @@ impl<'de> Deserialize<'de> for PendingEvent {
                 env,
                 replay_through,
                 journal_through,
+                settings_through,
             } => Self {
                 env,
                 replay_through,
                 journal_through,
+                settings_through,
             },
             StoredPendingEvent::Legacy(env) => Self {
                 env,
                 replay_through: 0,
                 journal_through: 0,
+                settings_through: None,
             },
         })
     }
@@ -132,13 +140,29 @@ pub(super) async fn accept_event(daemon: &Arc<Daemon>, event: PendingEvent) -> R
     let mut state = daemon.pending_sessions.lock().get(&key).cloned();
     if state.is_none() {
         if daemon.standalone_links.lock().contains(&key) {
-            return accept_resolved_event(daemon, event).await;
+            return accept_established_event(
+                daemon,
+                &key,
+                event,
+                PendingSession {
+                    standalone: true,
+                    ..Default::default()
+                },
+            )
+            .await;
         }
         let route = daemon.automatic_links.lock().get(&key).cloned();
         if let Some(route) = route {
-            let mut event = event;
-            event.env.route = Some(route);
-            return accept_resolved_event(daemon, event).await;
+            return accept_established_event(
+                daemon,
+                &key,
+                event,
+                PendingSession {
+                    linked_route: Some(route),
+                    ..Default::default()
+                },
+            )
+            .await;
         }
         state = read_correlation_state(&daemon.data_dir, &key).await;
     }
@@ -295,6 +319,32 @@ pub(super) fn automatic_parent_route(
     Some(route)
 }
 
+/// Keep the normal established-session handoff fast, but retain any failure
+/// in the same durable recovery queue used during initial route discovery.
+async fn accept_established_event(
+    daemon: &Arc<Daemon>,
+    key: &str,
+    event: PendingEvent,
+    mut state: PendingSession,
+) -> Result<(), String> {
+    let mut resolved = event.clone();
+    if let Some(route) = &state.linked_route {
+        resolved.env.route = Some(route.clone());
+    }
+    if let Err(failure) = accept_resolved_event(daemon, resolved, 0).await {
+        let message = failure.message();
+        state.recovery = Some(*failure);
+        state.events.push(event);
+        // store_correlation registers the pending state before attempting I/O,
+        // so failed evidence persistence cannot hide or abandon the handoff.
+        if let Err(error) = store_correlation(daemon, key, &state).await {
+            tracing::warn!(%error, "established-session recovery evidence could not be persisted");
+        }
+        return Err(message);
+    }
+    Ok(())
+}
+
 pub(super) async fn store_correlation(
     daemon: &Daemon,
     key: &str,
@@ -333,18 +383,29 @@ pub(super) async fn finish_correlation(
         .recovery
         .take()
         .map_or(0, |failure| failure.attempt.saturating_add(1));
-    let result: Result<(), String> = async {
+    let storage_failure = |error: String| {
+        Box::new(crate::pipeline::recovery::Failure::storage(
+            &anyhow::anyhow!(error),
+            &daemon.version,
+            attempt,
+        ))
+    };
+    let result: Result<(), Box<crate::pipeline::recovery::Failure>> = async {
         // Persist the final route before delivering any events. A crash may
         // replay deterministic rows, but must never select a different parent.
-        store_correlation(daemon, key, &state).await?;
+        store_correlation(daemon, key, &state)
+            .await
+            .map_err(storage_failure)?;
         while let Some(event) = state.events.first() {
             let mut event = event.clone();
             if let Some(route) = &state.linked_route {
                 event.env.route = Some(route.clone());
             }
-            accept_resolved_event(daemon, event).await?;
+            accept_resolved_event(daemon, event, attempt).await?;
             state.events.remove(0);
-            store_correlation(daemon, key, &state).await?;
+            store_correlation(daemon, key, &state)
+                .await
+                .map_err(storage_failure)?;
         }
         Ok(())
     }
@@ -354,9 +415,7 @@ pub(super) async fn finish_correlation(
             .events
             .first()
             .map_or(0, |event| event.journal_through);
-        state.recovery = Some(
-            crate::pipeline::recovery::Failure::credentials(error.clone(), attempt).at(position),
-        );
+        state.recovery = Some((**error).clone().at(position));
         if let Err(error) = store_correlation(daemon, key, &state).await {
             tracing::warn!(%error,"pending recovery evidence could not be persisted");
         }
@@ -365,7 +424,7 @@ pub(super) async fn finish_correlation(
             .lock()
             .insert(key.to_string(), state);
     }
-    result
+    result.map_err(|failure| failure.message())
 }
 
 pub(super) async fn retry_pending_sessions(daemon: &Arc<Daemon>) -> Result<(), String> {