Recover transient span-plugin timeouts and expose active holds

bt-daemon/README.md

@@ -175,7 +175,13 @@ 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. 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
+continue independently. Calls have a one-second total budget, including input
+conversion, synchronous execution, and output conversion. Timeout diagnostics
+identify the phase and include timings alongside the original exception.
+Timeouts receive at most two automatic retries with backoff from the same
+durable SpanOp position; unredacted operations are never delivered. Exhausted
+timeouts can recover after a runtime upgrade or a relevant plugin change.
+For other plugin errors, 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.
@@ -187,8 +193,11 @@ the raw QuickJS exception and stack. Inspect them with:
 bt trace doctor codex
 ```
 
-The doctor output reports the plugin path, exception, occurrence count, and
-timestamps. Managed-run diagnostics are copied out of their temporary daemon
+The doctor output separates historical plugin errors (path, exception, recorded
+occurrence count, and timestamps) from active delivery holds (session, stage,
+queued span operations, retry attempts, and recovery condition). Recorded
+occurrences do not count affected spans; queued span operations can include
+multiple updates to the same span. Managed-run diagnostics are copied out of their temporary daemon
 directory before it is removed.
 Plugins are trusted local code: although they have no host APIs, they can copy
 environment values into spans that are delivered to Braintrust. Read only the

bt-daemon/src/command_output.rs

@@ -123,6 +123,9 @@ pub struct DaemonDiagnostic {
     /// Distinct latest errors the daemon recorded for this agent's sessions.
     #[serde(skip_serializing_if = "Vec::is_empty")]
     pub session_errors: Vec<String>,
+    /// Current holds, separate from persisted historical plugin errors.
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub recovering_sessions: Vec<crate::wire::SessionStatus>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub error: Option<String>,
 }
@@ -271,12 +274,21 @@ impl TraceCommandOutput {
                 for error in &daemon.session_errors {
                     rendered.push_str(&format!("\nDaemon session error: {error}"));
                 }
+                for session in &daemon.recovering_sessions {
+                    if let Some(recovery) = &session.recovery {
+                        rendered.push_str(&format!(
+                            "\nActive delivery hold: {} / {}\nStage: {}\nQueued span operations: {}\nRetry attempts: {}\nWaiting for: {}",
+                            session.source, session.session_id, recovery.stage,
+                            session.queued_span_ops.map_or_else(|| "unknown".into(), |count| count.to_string()), recovery.attempt, recovery.retry
+                        ));
+                    }
+                }
                 for warning in &doctor.warnings {
                     rendered.push_str(&format!("\nWarning: {warning}"));
                 }
                 for diagnostic in &doctor.plugin_diagnostics {
                     rendered.push_str(&format!(
-                        "\nPlugin error: {} ({} occurrence{})\nFirst seen: {}\nLast seen: {}\n{}",
+                        "\nHistorical plugin error: {} ({} recorded occurrence{}; not an affected-span count)\nFirst seen: {}\nLast seen: {}\n{}",
                         diagnostic.plugin_path.display(),
                         diagnostic.occurrences,
                         if diagnostic.occurrences == 1 { "" } else { "s" },
@@ -532,6 +544,24 @@ mod tests {
                     error: Some("saved profile ID 'p' no longer exists".into()),
                 }),
                 session_errors: vec!["could not resolve Braintrust auth for codex".into()],
+                recovering_sessions: vec![crate::wire::SessionStatus {
+                    session_id: "held-session".into(),
+                    source: "codex".into(),
+                    route: None,
+                    recovery: Some(crate::wire::RecoveryInfo {
+                        stage: "plugin".into(),
+                        position: 1,
+                        reason: "deadline".into(),
+                        retry: "bounded retry".into(),
+                        eligible: false,
+                        attempt: 1,
+                    }),
+                    queued: 8,
+                    queued_span_ops: Some(7),
+                    spans_emitted: 0,
+                    permalink: None,
+                    last_error: None,
+                }],
                 error: None,
             },
             warnings: Vec::new(),
@@ -555,6 +585,19 @@ mod tests {
             value["daemon"]["session_errors"][0],
             "could not resolve Braintrust auth for codex"
         );
+        assert_eq!(
+            value["daemon"]["recovering_sessions"][0]["queued_span_ops"],
+            7
+        );
+        assert_eq!(
+            value["daemon"]["recovering_sessions"][0]["recovery"]["attempt"],
+            1
+        );
+        let human = output.render(OutputFormat::Human).unwrap();
+        assert!(human.contains("Active delivery hold: codex / held-session"));
+        assert!(human.contains("Queued span operations: 7"));
+        assert!(human.contains("Historical plugin error:"));
+        assert!(human.contains("not an affected-span count"));
         assert!(!rendered.contains("token"));
         assert!(!rendered.contains("api_key"));
         assert_eq!(

bt-daemon/src/dispatch.rs

@@ -208,6 +208,10 @@ impl Session {
             .saturating_add(self.source_progress.pending())
     }
 
+    pub(crate) fn pending_span_ops(&self) -> u64 {
+        self.route_progress.pending()
+    }
+
     /// Queue a journaled event without waiting for translation or delivery.
     pub async fn enqueue(&self, env: Envelope, journal_through: u64) -> anyhow::Result<()> {
         self.touch();

bt-daemon/src/pipeline/consumer.rs

@@ -131,7 +131,10 @@ impl ConsumerOptions {
                 tracing::warn!(session_id=%self.session_id, %error, "route processing failed");
             }
         }
-        Box::new((Failure::routing(error, config, attempt), Work::Resubmit))
+        Box::new((
+            Failure::routing(error, config, attempt, &self.version),
+            Work::Resubmit,
+        ))
     }
 
     // Keep the original acknowledgement obligations until checked delivery has
@@ -545,6 +548,40 @@ mod tests {
         let _ = forward.await;
         journal.close().await.unwrap();
     }
+
+    #[tokio::test]
+    async fn transient_plugin_timeout_replays_the_held_operation_without_an_edit() {
+        let tmp = tempfile::tempdir().unwrap();
+        let plugin = tmp.path().join("redact.mjs");
+        std::fs::write(&plugin, "let calls=0; export default span => {if(calls++===0) {while(true){}} return {...span,name:'redacted'};}").unwrap();
+        let journal = SpanJournal::open(&tmp.path().join("spans.sqlite"))
+            .await
+            .unwrap();
+        stored(&journal).await;
+        let route = SessionRoute {
+            span_plugins: vec![plugin],
+            ..Default::default()
+        };
+        let record = Arc::new(Recording::default());
+        let opts = options(tmp.path(), &route, record.clone());
+        let recovery = opts.recovery.clone();
+        let (consumer, progress, error) = spawn(journal.subscribe(&route, 0).await.unwrap(), opts);
+        wait_for(|| recovery.lock().is_some()).await;
+        assert_eq!(progress.completed(), 0);
+        assert_eq!(record.accepted.load(Ordering::SeqCst), 0);
+        assert_eq!(recovery.lock().as_ref().unwrap().stage, "plugin");
+        wait_for(|| progress.completed() == 1).await;
+        assert_eq!(record.accepted.load(Ordering::SeqCst), 1);
+        assert!(error.lock().is_none());
+        assert!(recovery.lock().is_none());
+        assert!(journal
+            .failure(&serde_json::to_string(&route).unwrap())
+            .await
+            .unwrap()
+            .is_none());
+        consumer.shutdown().await;
+        journal.close().await.unwrap();
+    }
     struct Acknowledged(Arc<AtomicU64>);
     #[async_trait::async_trait]
     impl crate::pipeline::Completion for Acknowledged {

bt-daemon/src/pipeline/plugins.rs

@@ -46,6 +46,15 @@ impl Plugins {
     }
 
     pub fn configure(&mut self, paths: &[PathBuf]) {
+        // One timeout attempt per submission; the durable consumer owns retry
+        // timing and attempt limits. Do not latch a transient deadline forever.
+        if self
+            .blocked
+            .as_ref()
+            .is_some_and(|(_, failure)| failure.failure.timed_out)
+        {
+            self.blocked = None;
+        }
         self.paths.clear();
         self.paths.extend_from_slice(paths);
     }

bt-daemon/src/pipeline/recovery.rs

@@ -39,6 +39,8 @@ pub(crate) enum Condition {
         chain: String,
         #[serde(default, skip_serializing_if = "Option::is_none")]
         processors: Option<Vec<Option<[u8; 32]>>>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        timeout: Option<Box<TimeoutRetry>>,
     },
     Configuration {
         signature: String,
@@ -55,11 +57,18 @@ pub(crate) enum Condition {
     },
 }
 
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub(crate) struct TimeoutRetry {
+    recheck_ms: Option<i64>,
+    version: String,
+}
+
 /// 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 {
+        timed_out: bool,
         path: PathBuf,
         digest: Option<[u8; 32]>,
         processors: Vec<Option<[u8; 32]>>,
@@ -173,12 +182,18 @@ impl Failure {
             },
         }
     }
-    pub fn routing(error: &anyhow::Error, config: &SessionConfig, attempt: u32) -> Self {
+    pub fn routing(
+        error: &anyhow::Error,
+        config: &SessionConfig,
+        attempt: u32,
+        version: &str,
+    ) -> Self {
         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 {
+                timed_out,
                 path,
                 digest,
                 processors,
@@ -189,6 +204,13 @@ impl Failure {
                     digest: *digest,
                     chain: chain(config),
                     processors: Some(processors.clone()),
+                    timeout: timed_out.then(|| {
+                        Box::new(TimeoutRetry {
+                            recheck_ms: (attempt < 2)
+                                .then(|| now_ms().saturating_add(500 * (1i64 << attempt.min(8)))),
+                            version: version.to_owned(),
+                        })
+                    }),
                 };
             }
             Some(RetryCondition::Credentials) => {
@@ -239,10 +261,22 @@ impl Failure {
     }
     pub fn info(&self, eligible: bool) -> RecoveryInfo {
         let retry = match &self.condition {
-            Condition::Plugin { path, .. } => format!(
-                "bytes/existence of {} or configured plugin chain changes",
-                path.display()
-            ),
+            Condition::Plugin { path, timeout, .. } => {
+                let mut retry = format!(
+                    "bytes/existence of {} or configured plugin chain changes",
+                    path.display()
+                );
+                if let Some(timeout) = timeout {
+                    if let Some(when) = timeout.recheck_ms {
+                        retry.push_str(&format!(" or timeout retry at {when}"));
+                    }
+                    retry.push_str(&format!(
+                        " or runtime revision changes from {}",
+                        timeout.version
+                    ));
+                }
+                retry
+            }
             Condition::Configuration { recheck_ms, .. } => match recheck_ms {
                 Some(when) => format!("relevant settings/processors change or recheck at {when}"),
                 None => "relevant settings or processors change".into(),
@@ -259,6 +293,7 @@ impl Failure {
             reason: self.reason.clone(),
             retry,
             eligible,
+            attempt: self.attempt,
         }
     }
     pub fn message(&self) -> String {
@@ -296,8 +331,12 @@ impl Stage for Recovery<'_> {
                 digest,
                 chain: previous,
                 processors,
+                timeout,
             } => {
-                self.config.is_some_and(|config| chain(config) != *previous)
+                timeout.as_ref().is_some_and(|timeout| {
+                    timeout.recheck_ms.is_some_and(|when| now_ms() >= when)
+                        || timeout.version != self.version
+                }) || self.config.is_some_and(|config| chain(config) != *previous)
                     || crate::span_processor::file_digest(path) != *digest
                     || processors.as_ref().is_some_and(|previous| {
                         self.config.is_some_and(|config| {
@@ -330,3 +369,68 @@ impl Stage for Recovery<'_> {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn timeout_retries_are_bounded_and_runtime_upgrades_unblock_them() {
+        let tmp = tempfile::tempdir().unwrap();
+        let path = tmp.path().join("plugin.mjs");
+        std::fs::write(&path, "export default span => span").unwrap();
+        let digest = crate::span_processor::file_digest(&path);
+        let config = crate::wire::SessionRoute {
+            span_plugins: vec![path.clone()],
+            ..Default::default()
+        }
+        .with_auth(crate::wire::BackendAuth {
+            token: String::new(),
+            api_url: None,
+            app_url: None,
+            org_id: None,
+            org_name: None,
+        });
+        let error = anyhow::Error::new(RouteFailure {
+            cause: anyhow::anyhow!("deadline"),
+            retry: RetryCondition::Plugin {
+                path,
+                digest,
+                processors: vec![digest],
+                timed_out: true,
+            },
+            emitted: 0,
+            report: true,
+        });
+        for attempt in 0..=2 {
+            let mut failure = Failure::routing(&error, &config, attempt, "old");
+            let Condition::Plugin { timeout, .. } = &mut failure.condition else {
+                panic!("plugin condition")
+            };
+            let timeout = timeout.as_mut().unwrap();
+            assert_eq!(timeout.recheck_ms.is_some(), attempt < 2);
+            if timeout.recheck_ms.is_some() {
+                timeout.recheck_ms = Some(0);
+            }
+            let json = serde_json::to_string(&failure).unwrap();
+            let failure: Failure = serde_json::from_str(&json).unwrap();
+            assert_eq!(failure.info(false).attempt, attempt);
+            let result = (Recovery {
+                config: Some(&config),
+                version: "old",
+            })
+            .process(&failure)
+            .await;
+            assert_eq!(matches!(result, RecoveryOutcome::Retry), attempt < 2);
+            assert!(matches!(
+                (Recovery {
+                    config: Some(&config),
+                    version: "new"
+                })
+                .process(&failure)
+                .await,
+                RecoveryOutcome::Retry
+            ));
+        }
+    }
+}

bt-daemon/src/pipeline/routing.rs

@@ -80,6 +80,7 @@ impl RoutingSink {
                                     failure = Some((
                                         anyhow::Error::new(error.clone()),
                                         RetryCondition::Plugin {
+                                            timed_out: error.failure.timed_out,
                                             path: error.failure.path.clone(),
                                             digest: error.digest,
                                             processors: error.processors.clone(),

bt-daemon/src/server/rpc.rs

@@ -335,6 +335,7 @@ impl Daemon {
                 source: s.source.clone(),
                 route: serde_json::from_str(&key.route).ok(),
                 queued: s.pending(),
+                queued_span_ops: Some(s.pending_span_ops()),
                 recovery: s
                     .recovery
                     .lock()
@@ -364,6 +365,7 @@ impl Daemon {
                 source: source.clone(),
                 route: serde_json::from_str(&key.route).ok(),
                 queued: 0,
+                queued_span_ops: None,
                 recovery: None,
                 spans_emitted: 0,
                 permalink: None,
@@ -414,6 +416,7 @@ impl Daemon {
                     source: first.env.source.clone(),
                     route: route.cloned(),
                     queued: state.events.len() as u64,
+                    queued_span_ops: None,
                     recovery: state.recovery.as_ref().map(|failure| failure.info(false)),
                     spans_emitted: 0,
                     permalink: None,