From ab63b77411a73c65a942f7f84755540c929a90e0 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 21 Oct 2026 02:20:46 +0800 Subject: [PATCH] Recover transient span-plugin timeouts and report active holds --- bt-daemon/README.md | 15 ++- bt-daemon/src/command_output.rs | 45 ++++++- bt-daemon/src/dispatch.rs | 4 + bt-daemon/src/pipeline/consumer.rs | 39 +++++- bt-daemon/src/pipeline/plugins.rs | 9 ++ bt-daemon/src/pipeline/recovery.rs | 116 +++++++++++++++++- bt-daemon/src/pipeline/routing.rs | 1 + bt-daemon/src/server/rpc.rs | 3 + bt-daemon/src/span_processor.rs | 191 ++++++++++++++++++++++++++--- bt-daemon/src/trace_runtime.rs | 10 ++ bt-daemon/src/wire/methods.rs | 7 ++ 11 files changed, 411 insertions(+), 29 deletions(-) diff --git a/bt-daemon/README.md b/bt-daemon/README.md index 8f78c933..b01c2484 100644 --- a/bt-daemon/README.md +++ b/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 diff --git a/bt-daemon/src/command_output.rs b/bt-daemon/src/command_output.rs index 704e7f09..ccc0a5ec 100644 --- a/bt-daemon/src/command_output.rs +++ b/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, + /// Current holds, separate from persisted historical plugin errors. + #[serde(skip_serializing_if = "Vec::is_empty")] + pub recovering_sessions: Vec, #[serde(skip_serializing_if = "Option::is_none")] pub error: Option, } @@ -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!( diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 47baecb1..9522bc95 100644 --- a/bt-daemon/src/dispatch.rs +++ b/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(); diff --git a/bt-daemon/src/pipeline/consumer.rs b/bt-daemon/src/pipeline/consumer.rs index 0dc416f0..6876c240 100644 --- a/bt-daemon/src/pipeline/consumer.rs +++ b/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); #[async_trait::async_trait] impl crate::pipeline::Completion for Acknowledged { diff --git a/bt-daemon/src/pipeline/plugins.rs b/bt-daemon/src/pipeline/plugins.rs index 322aff18..2b75de8d 100644 --- a/bt-daemon/src/pipeline/plugins.rs +++ b/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); } diff --git a/bt-daemon/src/pipeline/recovery.rs b/bt-daemon/src/pipeline/recovery.rs index 82147924..45a72cea 100644 --- a/bt-daemon/src/pipeline/recovery.rs +++ b/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>>, + #[serde(default, skip_serializing_if = "Option::is_none")] + timeout: Option>, }, Configuration { signature: String, @@ -55,11 +57,18 @@ pub(crate) enum Condition { }, } +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub(crate) struct TimeoutRetry { + recheck_ms: Option, + 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>, @@ -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::(); 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 + )); + } + } +} diff --git a/bt-daemon/src/pipeline/routing.rs b/bt-daemon/src/pipeline/routing.rs index 610a7d96..6c8ea0d9 100644 --- a/bt-daemon/src/pipeline/routing.rs +++ b/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(), diff --git a/bt-daemon/src/server/rpc.rs b/bt-daemon/src/server/rpc.rs index 0f16c2d1..f8e8c938 100644 --- a/bt-daemon/src/server/rpc.rs +++ b/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, diff --git a/bt-daemon/src/span_processor.rs b/bt-daemon/src/span_processor.rs index f706496e..b67152c6 100644 --- a/bt-daemon/src/span_processor.rs +++ b/bt-daemon/src/span_processor.rs @@ -11,13 +11,25 @@ use sha2::{Digest, Sha256}; use std::cell::RefCell; use std::collections::{BTreeMap, HashMap}; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; const MEMORY_LIMIT_BYTES: usize = 64 * 1024 * 1024; const STACK_LIMIT_BYTES: usize = 512 * 1024; -const CALL_TIMEOUT: Duration = Duration::from_millis(50); +const CALL_TIMEOUT: Duration = Duration::from_secs(1); + +#[derive(Debug, thiserror::Error)] +#[error("span plugin exceeded its {budget_ms} ms budget during {phase} (total {elapsed_ms} ms; input conversion {input_ms} ms, execution {execution_ms} ms, output conversion {output_ms} ms): {exception}")] +struct PluginTimeout { + budget_ms: u128, + phase: &'static str, + elapsed_ms: u128, + input_ms: u128, + execution_ms: u128, + output_ms: u128, + exception: String, +} thread_local! { static ENGINE: RefCell> = const { RefCell::new(None) }; @@ -46,6 +58,7 @@ struct Engine { context: Context, started: Instant, deadline_ms: Arc, + interrupted: Arc, // Must drop after every Context and persistent JavaScript value. _runtime: Runtime, } @@ -59,6 +72,7 @@ struct CachedModule { struct FailedPlugin { fingerprint: Option, message: String, + timed_out: bool, } #[derive(Clone, Copy, PartialEq, Eq)] @@ -75,9 +89,16 @@ impl Engine { let deadline_ms = Arc::new(AtomicU64::new(0)); let interrupt_deadline = deadline_ms.clone(); let interrupt_started = started; + let interrupted = Arc::new(AtomicBool::new(false)); + let interrupt_fired = interrupted.clone(); runtime.set_interrupt_handler(Some(Box::new(move || { let deadline = interrupt_deadline.load(Ordering::Relaxed); - deadline != 0 && interrupt_started.elapsed().as_millis() as u64 >= deadline + let expired = + deadline != 0 && interrupt_started.elapsed().as_millis() as u64 >= deadline; + if expired { + interrupt_fired.store(true, Ordering::Relaxed); + } + expired }))); let context = Context::full(&runtime)?; Ok(Self { @@ -88,6 +109,7 @@ impl Engine { context, started, deadline_ms, + interrupted, _runtime: runtime, }) } @@ -107,7 +129,8 @@ impl Engine { } let digest = Sha256::digest(&source); let name = format!("bt-span-plugin:{digest:x}"); - self.arm_deadline(); + self.arm_deadline(CALL_TIMEOUT); + let started = Instant::now(); let function = self.context.with(|ctx| -> anyhow::Result<_> { let (module, promise) = Module::declare(ctx.clone(), name, source) .catch(&ctx) @@ -126,6 +149,21 @@ impl Engine { Ok(Persistent::save(&ctx, function)) }); self.deadline_ms.store(0, Ordering::Relaxed); + if self.interrupted.load(Ordering::Relaxed) || started.elapsed() >= CALL_TIMEOUT { + return Err(PluginTimeout { + budget_ms: CALL_TIMEOUT.as_millis(), + phase: "module loading", + elapsed_ms: started.elapsed().as_millis(), + input_ms: 0, + execution_ms: 0, + output_ms: 0, + exception: function + .err() + .map(|error| error.to_string()) + .unwrap_or_default(), + } + .into()); + } let function = function?; self.modules.insert( path.to_path_buf(), @@ -144,6 +182,18 @@ impl Engine { operation: Operation, source: &str, session_id: &str, + ) -> anyhow::Result { + self.call_with_budget(path, row, operation, source, session_id, CALL_TIMEOUT) + } + + fn call_with_budget( + &mut self, + path: &Path, + row: &SpanRow, + operation: Operation, + source: &str, + session_id: &str, + budget: Duration, ) -> anyhow::Result { let function = self.load(path)?; let context = PluginContext { @@ -152,30 +202,63 @@ impl Engine { session_id, env: &self.env, }; - self.arm_deadline(); + self.arm_deadline(budget); + let started = Instant::now(); + let mut phase = "input conversion"; + let mut input_ms = 0; + let mut execution_ms = 0; + let mut output_ms = 0; let result = self.context.with(|ctx| -> anyhow::Result { let function = function.restore(&ctx)?; let row = rquickjs_serde::to_value(ctx.clone(), row)?; let context = rquickjs_serde::to_value(ctx.clone(), &context)?; + input_ms = started.elapsed().as_millis(); + anyhow::ensure!( + started.elapsed() < budget, + "deadline expired before plugin execution" + ); + phase = "plugin execution"; + let executing = Instant::now(); let result = function .call::<_, rquickjs::Value<'_>>((row, context)) .catch(&ctx) - .map_err(|error| anyhow::anyhow!(error.to_string()))?; + .map_err(|error| anyhow::anyhow!(error.to_string())); + execution_ms = executing.elapsed().as_millis(); + let result = result?; if result.as_promise().is_some() { anyhow::bail!("plugin returned a Promise; span plugins must be synchronous"); } - Ok(rquickjs_serde::from_value_strict(result)?) + phase = "output conversion"; + let converting = Instant::now(); + let result = rquickjs_serde::from_value_strict(result); + output_ms = converting.elapsed().as_millis(); + Ok(result?) }); self.deadline_ms.store(0, Ordering::Relaxed); + if self.interrupted.load(Ordering::Relaxed) || started.elapsed() >= budget { + if phase == "input conversion" { + input_ms = started.elapsed().as_millis(); + } + return Err(PluginTimeout { + budget_ms: budget.as_millis(), + phase, + elapsed_ms: started.elapsed().as_millis(), + input_ms, + execution_ms, + output_ms, + exception: result + .err() + .map(|error| error.to_string()) + .unwrap_or_default(), + } + .into()); + } result } - fn arm_deadline(&self) { - let deadline = self - .started - .elapsed() - .saturating_add(CALL_TIMEOUT) - .as_millis() as u64; + fn arm_deadline(&self, budget: Duration) { + self.interrupted.store(false, Ordering::Relaxed); + let deadline = self.started.elapsed().saturating_add(budget).as_millis() as u64; self.deadline_ms.store(deadline.max(1), Ordering::Relaxed); } } @@ -184,6 +267,7 @@ impl Engine { pub struct PluginFailure { pub path: PathBuf, pub message: String, + pub timed_out: bool, /// Only the first failure on this worker needs to be logged and persisted. pub newly_seen: bool, } @@ -252,12 +336,13 @@ fn process_inner( .failed_plugins .get(plugin) .cloned() - .filter(|_| use_failed_cache) + .filter(|failed| use_failed_cache && !failed.timed_out) { if failed.fingerprint == plugin_fingerprint(plugin) { failure = Some(PluginFailure { path: plugin.clone(), message: failed.message, + timed_out: failed.timed_out, newly_seen: false, }); break; @@ -283,24 +368,27 @@ fn process_inner( let message = "changed immutable span identity fields".to_owned(); engine.failed_plugins.insert( plugin.clone(), - failed_plugin(engine.last_fingerprint, &message), + failed_plugin(engine.last_fingerprint, &message, false), ); failure = Some(PluginFailure { path: plugin.clone(), message, + timed_out: false, newly_seen: true, }); break; } Err(error) => { + let timed_out = error.downcast_ref::().is_some(); let message = error.to_string(); engine.failed_plugins.insert( plugin.clone(), - failed_plugin(engine.last_fingerprint, &message), + failed_plugin(engine.last_fingerprint, &message, timed_out), ); failure = Some(PluginFailure { path: plugin.clone(), message, + timed_out, newly_seen: true, }); break; @@ -324,10 +412,15 @@ fn process_inner( }) } -fn failed_plugin(fingerprint: Option, message: &str) -> FailedPlugin { +fn failed_plugin( + fingerprint: Option, + message: &str, + timed_out: bool, +) -> FailedPlugin { FailedPlugin { fingerprint, message: message.to_owned(), + timed_out, } } @@ -452,6 +545,66 @@ mod tests { assert_eq!(processed.metadata.unwrap()["second"], true); } + #[test] + fn conversion_deadline_is_reported_before_the_filter_runs() { + let dir = tempfile::tempdir().unwrap(); + let plugin = dir.path().join("guard.mjs"); + std::fs::write( + &plugin, + "export default () => { throw new Error('filter ran'); }", + ) + .unwrap(); + let error = Engine::new() + .unwrap() + .call_with_budget( + &plugin, + &row(), + Operation::Insert, + "pi", + "session", + Duration::ZERO, + ) + .unwrap_err(); + let timeout = error.downcast_ref::().unwrap(); + assert_eq!(timeout.phase, "input conversion"); + assert_eq!(timeout.execution_ms, 0); + assert!(!timeout.exception.contains("filter ran")); + } + + #[test] + fn redacts_large_input_and_preserves_property_error_details() { + let dir = tempfile::tempdir().unwrap(); + let plugin = dir.path().join("redact.mjs"); + std::fs::write( + &plugin, + "export default span => {delete span.metadata.username; return span}", + ) + .unwrap(); + let mut span = row(); + span.input = Some(serde_json::json!((0..10_000) + .map(|_| "x".repeat(100)) + .collect::>())); + span.metadata = Some(serde_json::json!({"username":"private","source":"pi"})); + let processed = Engine::new() + .unwrap() + .call(&plugin, &span, Operation::Insert, "pi", "session") + .unwrap(); + assert_eq!(processed.input, span.input); + assert_eq!(processed.metadata, Some(serde_json::json!({"source":"pi"}))); + assert_eq!(span.metadata.unwrap()["username"], "private"); + + std::fs::write( + &plugin, + "export default span => {span.missing.workspace; return span}", + ) + .unwrap(); + let result = process(&[plugin], &SpanOp::Insert(row()), "pi", "session").unwrap(); + let failure = result.failure.unwrap(); + assert!(!failure.timed_out); + assert!(failure.message.contains("workspace")); + assert!(result.op.is_none()); + } + #[test] fn rejects_identity_changes_and_drops_the_operation() { let dir = tempfile::tempdir().unwrap(); @@ -478,7 +631,9 @@ mod tests { std::fs::write(&runaway, "export default span => { while (true) {} }").unwrap(); let started = Instant::now(); let result = process(&[runaway], &SpanOp::Insert(row()), "codex", "session").unwrap(); - assert!(result.failure.is_some()); + let failure = result.failure.as_ref().unwrap(); + assert!(failure.timed_out); + assert!(failure.message.contains("plugin execution")); assert!(result.op.is_none()); assert!(started.elapsed() < Duration::from_secs(2)); diff --git a/bt-daemon/src/trace_runtime.rs b/bt-daemon/src/trace_runtime.rs index 684c97b6..bc90c766 100644 --- a/bt-daemon/src/trace_runtime.rs +++ b/bt-daemon/src/trace_runtime.rs @@ -352,6 +352,14 @@ async fn diagnose_daemon( Ok(Err(error)) => return unreachable(error.to_string()), Ok(Ok(status)) => { let mut session_errors = Vec::new(); + let recovering_sessions = status + .sessions + .iter() + .filter(|session| { + is_same_agent(source, &session.source) && session.recovery.is_some() + }) + .cloned() + .collect(); let errors = status .sessions .into_iter() @@ -370,6 +378,7 @@ async fn diagnose_daemon( DaemonDiagnostic { version: Some(status.daemon_version), session_errors, + recovering_sessions, ..DaemonDiagnostic::default() } } @@ -1644,6 +1653,7 @@ mod tests { version: Some(version.into()), auth: Some(auth), session_errors: vec![session_error.into()], + recovering_sessions: Vec::new(), error: None, }; let shell = diagnostic("ready", None); diff --git a/bt-daemon/src/wire/methods.rs b/bt-daemon/src/wire/methods.rs index 7461d03e..1d4146ec 100644 --- a/bt-daemon/src/wire/methods.rs +++ b/bt-daemon/src/wire/methods.rs @@ -93,6 +93,9 @@ pub struct StatusResult { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RecoveryInfo { + /// Retry attempts already made for this held delivery. + #[serde(default)] + pub attempt: u32, pub stage: String, pub position: u64, pub reason: String, @@ -109,6 +112,10 @@ pub struct SessionStatus { #[serde(default, skip_serializing_if = "Option::is_none")] pub recovery: Option, pub queued: u64, + /// Canonical span operations not yet acknowledged for this route. Distinct + /// from native events still awaiting translation and from unique spans. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub queued_span_ops: Option, pub spans_emitted: u64, #[serde(default, skip_serializing_if = "Option::is_none")] pub permalink: Option,