From d874a612f37e022135ba59d2af35ebff236b3b02 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 22:13:19 +0800 Subject: [PATCH 1/8] Add JavaScript span transform plugins --- bt-daemon/Cargo.lock | 56 +++ bt-daemon/Cargo.toml | 1 + bt-daemon/README.md | 34 ++ bt-daemon/config.json.example | 5 +- bt-daemon/docs/protocol.md | 17 +- bt-daemon/src/dispatch.rs | 76 +++- bt-daemon/src/journal.rs | 1 + bt-daemon/src/lib.rs | 100 +++++- bt-daemon/src/settings.rs | 7 + bt-daemon/src/setup.rs | 35 ++ bt-daemon/src/span_processor.rs | 335 ++++++++++++++++++ bt-daemon/src/trace_command.rs | 51 +++ bt-daemon/src/trace_runtime.rs | 25 ++ bt-daemon/src/transcript_import/mod.rs | 1 + bt-daemon/src/wire/envelope.rs | 25 ++ bt-daemon/tests/braintrust_sink.rs | 1 + bt-daemon/tests/claude_translator.rs | 6 + bt-daemon/tests/codex_translator.rs | 2 + bt-daemon/tests/opencode_translator.rs | 1 + bt-daemon/tests/pi_translator.rs | 1 + bt-daemon/tests/pipeline.rs | 116 ++++++ bt-daemon/tests/replay.rs | 51 +++ src/runtime/js-daemon-client/src/index.ts | 13 + .../js-daemon-client/tests/client.test.ts | 5 + 24 files changed, 944 insertions(+), 21 deletions(-) create mode 100644 bt-daemon/src/span_processor.rs diff --git a/bt-daemon/Cargo.lock b/bt-daemon/Cargo.lock index dfc514e..76ec42f 100644 --- a/bt-daemon/Cargo.lock +++ b/bt-daemon/Cargo.lock @@ -11,6 +11,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "android_system_properties" version = "0.1.5" @@ -271,6 +277,7 @@ dependencies = [ "clap", "regex", "reqwest", + "rquickjs", "serde", "serde_json", "sha2", @@ -598,6 +605,12 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "foldhash" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -756,6 +769,11 @@ name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash", +] [[package]] name = "heck" @@ -1407,6 +1425,15 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "relative-path" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bca40a312222d8ba74837cb474edef44b37f561da5f773981007a10bbaa992b0" +dependencies = [ + "serde", +] + [[package]] name = "reqwest" version = "0.12.28" @@ -1463,6 +1490,35 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rquickjs" +version = "0.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e04e4eedfb060b503b5f0a2644abb890b0b3620d3fb674f9455f230014964e4" +dependencies = [ + "rquickjs-core", +] + +[[package]] +name = "rquickjs-core" +version = "0.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16e4f499ac5b943d97ee6dbc44f23c2c10426f420f7d2f1793d6318911b6608c" +dependencies = [ + "hashbrown", + "relative-path", + "rquickjs-sys", +] + +[[package]] +name = "rquickjs-sys" +version = "0.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a13ac243b86a74120814ef7e9e30ad5a2c1199b7b9963b1cf7c84e4cdc1cad99" +dependencies = [ + "cc", +] + [[package]] name = "rustc-hash" version = "2.1.3" diff --git a/bt-daemon/Cargo.toml b/bt-daemon/Cargo.toml index d404440..c7e7698 100644 --- a/bt-daemon/Cargo.toml +++ b/bt-daemon/Cargo.toml @@ -24,6 +24,7 @@ async-trait = "0.1" chrono = "0.4" clap = { version = "4", features = ["derive", "env"] } regex = "1" +rquickjs = "0.12.2" serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.10" diff --git a/bt-daemon/README.md b/bt-daemon/README.md index e5feabb..e0187c7 100644 --- a/bt-daemon/README.md +++ b/bt-daemon/README.md @@ -59,6 +59,40 @@ the default `bt` profile. Credentials and backend URLs are never stored here; production resolves and refreshes them through `bt`. `bt trace run` supplies a process-local settings overlay and never changes any of these files. +### JavaScript span plugins + +`--plugin PATH` registers a synchronous ES module that transforms each +sink-neutral span row after translation and immediately before delivery. Repeat +the flag to compose plugins from left to right. Setup persists its ordered list; +run and import append invocation plugins after the persisted list. + +```bash +bt trace setup codex --plugin ./redact.mjs --plugin ./tag-ci.mjs +bt trace run --plugin ./local.mjs codex -- "summarize this change" +bt trace import codex SESSION_ID --plugin ./sanitize-history.mjs +``` + +Each module must default-export a synchronous function. It receives a span and +`{ operation, source, session_id, env }`, and must return a JSON-compatible span +object. Span, root, and parent identities cannot be changed: + +```js +export default function mapSpan(span, context) { + return { + ...span, + metadata: { ...span.metadata, deployment: context.env.DEPLOYMENT }, + }; +} +``` + +The environment map comes from the event-producing hook or adapter and is sent +only over the private daemon transport; it is never written to the recovery +journal. 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. A +runtime failure disables the chain for that session and sends the original +translator output instead. + ### Additional root metadata `additional_metadata` is a JSON object merged into each traced session's root diff --git a/bt-daemon/config.json.example b/bt-daemon/config.json.example index 78355d2..ed1363b 100644 --- a/bt-daemon/config.json.example +++ b/bt-daemon/config.json.example @@ -14,6 +14,9 @@ "additional_metadata": { "team": "platform", "environment": "development" - } + }, + "span_plugins": [ + "/absolute/path/to/redact.mjs" + ] } } diff --git a/bt-daemon/docs/protocol.md b/bt-daemon/docs/protocol.md index 6bc832f..1d19bc2 100644 --- a/bt-daemon/docs/protocol.md +++ b/bt-daemon/docs/protocol.md @@ -188,6 +188,7 @@ Used for version handover and by tests. "ts_ms": 1753639552123, "managed_run_id": "invocation-uuid", "payload": { "…raw agent-native hook payload…": true }, + "plugin_env": { "CI": "true" }, "route": { "auth": { "profile": "work", @@ -199,7 +200,8 @@ Used for version handover and by tests. "project_name": "codex" }, "flush_mode": "fire_and_forget", - "additional_metadata": { "…": "…" } + "additional_metadata": { "…": "…" }, + "span_plugins": ["/absolute/path/redact.mjs"] } } ``` @@ -220,6 +222,8 @@ Field notes: daemon. - **`payload`** is opaque to transport and to everything except the translator for `source`. +- **`plugin_env`** is the event producer's string environment map, exposed to + span plugins as `context.env`. It is transported live but never journaled. - **`managed_run_id`** is present only for events inherited from a `bt trace run` process tree. It groups native sessions for the final invocation flush and is not trace metadata. @@ -247,8 +251,9 @@ Field notes: ### Redaction Live credentials returned by the host provider are **never** written to the -journal, logs, status, or RPC response. Envelopes journal only their non-secret -`route`, allowing restart recovery to resolve a fresh lease. +journal, logs, status, or RPC response. The live plugin environment is likewise +omitted because it commonly contains secrets. Envelopes journal only their +non-secret `route`, allowing restart recovery to resolve a fresh lease. ## Daemon lifecycle @@ -315,8 +320,10 @@ profiles, organizations, and destinations while sharing one daemon. `$HOME/.braintrust/state/bt-daemon` on Unix, and `%LOCALAPPDATA%\Braintrust\bt-daemon` on Windows. On restart the daemon rebuilds each route's unfinished correlation state independently, replaying - only the journal entries whose `route` matches that pipeline into a fresh - translator. The resulting rows may be resubmitted to repair delivery + only the journal entries whose delivery route matches that pipeline into a + fresh translator. Span plugin paths are ignored for this comparison so raw + events can be replayed through the current plugin chain. The resulting rows + may be resubmitted to repair delivery interrupted by a crash, but their deterministic ids target the same backend rows and must never create duplicate spans, and a route never receives another route's rows. Replay streams the journal and is bounded to the diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index d065898..7864278 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -11,6 +11,7 @@ use crate::sink::SinkFactory; use crate::translate::{Registry, SessionCtx}; use crate::wire::{Envelope, SessionRoute}; +use std::collections::BTreeMap; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; @@ -214,6 +215,11 @@ struct SessionActor { config: crate::wire::SessionConfig, } +struct PluginState { + env: BTreeMap, + disabled: bool, +} + impl SessionActor { async fn run(self, mut rx: mpsc::Receiver) { let mut translator = self.translators.create(&self.source, &self.session_id); @@ -251,16 +257,24 @@ impl SessionActor { session_id: self.session_id.clone(), config: Some(self.config.clone()), }; + let mut plugins = PluginState { + env: crate::span_processor::environment(), + disabled: false, + }; sink.configure(&self.config); self.refresh_permalink(sink.as_ref()); // Rebuild translator state before accepting the first new event. // Stable span ids make this both crash recovery and a complete copy // when an existing source session is sent to another destination. - self.replay_into(&mut translator, &mut sink, &ctx).await; + self.replay_into(&mut translator, &mut sink, &ctx, &mut plugins) + .await; while let Some(msg) = rx.recv().await { match msg { SessionMsg::Event(env) => { + if !env.plugin_env.is_empty() { + plugins.env = env.plugin_env.clone(); + } if let Some(cfg) = &env.config { sink.configure(cfg); ctx.config = Some(cfg.clone()); @@ -271,9 +285,9 @@ impl SessionActor { &mut translator, &mut sink, &ctx, + &mut plugins, translated, - "translate failed", - "sink emit failed", + ("translate failed", "sink emit failed"), ) .await; self.counters.queued.fetch_sub(1, Ordering::Relaxed); @@ -285,11 +299,13 @@ impl SessionActor { let _ = reply.send(()); } SessionMsg::Flush(reply) => { - self.drain_flush(&mut translator, &mut sink, &ctx).await; + self.drain_flush(&mut translator, &mut sink, &ctx, &mut plugins) + .await; let _ = reply.send(self.counters.queued.load(Ordering::Relaxed)); } SessionMsg::Shutdown(reply) => { - self.drain_flush(&mut translator, &mut sink, &ctx).await; + self.drain_flush(&mut translator, &mut sink, &ctx, &mut plugins) + .await; let _ = reply.send(()); break; } @@ -305,10 +321,11 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, + plugins: &mut PluginState, first: anyhow::Result>, - translate_error: &str, - emit_error: &str, + errors: (&str, &str), ) { + let (translate_error, emit_error) = errors; let mut next = match first { Ok(ops) => Some(ops), Err(e) => { @@ -318,7 +335,38 @@ impl SessionActor { }; while let Some(ops) = next { if !ops.is_empty() { - match sink.emit(&ops).await { + let processed = if plugins.disabled { + ops.clone() + } else { + let plugin_paths = ctx + .config + .as_ref() + .map(|config| config.span_plugins.as_slice()) + .unwrap_or_default(); + let transformed: anyhow::Result> = ops + .iter() + .map(|op| { + crate::span_processor::process( + plugin_paths, + op, + &self.source, + &self.session_id, + &plugins.env, + ) + }) + .collect(); + match transformed { + Ok(processed) => processed, + Err(error) => { + plugins.disabled = true; + self.set_error(format!( + "span plugin failed; disabled for session: {error}" + )); + ops.clone() + } + } + }; + match sink.emit(&processed).await { Ok(n) => { self.counters.spans_emitted.fetch_add(n, Ordering::Relaxed); } @@ -344,6 +392,7 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, + plugins: &mut PluginState, ) { let Some(plan) = &self.replay else { return; @@ -370,7 +419,7 @@ impl SessionActor { if !entry .route .as_ref() - .is_some_and(|candidate| candidate.same_route(&plan.route)) + .is_some_and(|candidate| candidate.same_replay_route(&plan.route)) { continue; } @@ -380,9 +429,9 @@ impl SessionActor { translator, sink, ctx, + plugins, translated, - "journal replay failed", - "sink replay emit failed", + ("journal replay failed", "sink replay emit failed"), ) .await; } @@ -393,15 +442,16 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, + plugins: &mut PluginState, ) { let translated = translator.flush(ctx); self.emit_translator_batches( translator, sink, ctx, + plugins, translated, - "translate flush failed", - "sink emit (flush) failed", + ("translate flush failed", "sink emit (flush) failed"), ) .await; if let Err(e) = sink.flush().await { diff --git a/bt-daemon/src/journal.rs b/bt-daemon/src/journal.rs index 17c71e7..e83759e 100644 --- a/bt-daemon/src/journal.rs +++ b/bt-daemon/src/journal.rs @@ -263,6 +263,7 @@ pub fn envelope_from_redacted(r: RedactedEnvelope) -> Envelope { ts_ms: r.ts_ms, managed_run_id: r.managed_run_id, payload: r.payload, + plugin_env: std::collections::BTreeMap::new(), route, config, } diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 762a8ad..5edb045 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -21,6 +21,7 @@ mod server; mod settings; mod setup; mod sink; +mod span_processor; mod trace_command; mod trace_runtime; mod transcript_import; @@ -163,6 +164,10 @@ pub struct ImportArgs { /// JSON object merged into every imported root span's metadata. #[arg(long, env = "BRAINTRUST_ADDITIONAL_METADATA")] pub additional_metadata: Option, + /// JavaScript span transform for this import. Repeat to compose transforms + /// after plugins from the agent's persistent setup. + #[arg(long, value_name = "PATH")] + pub plugin: Vec, } #[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] @@ -182,6 +187,10 @@ pub struct RunArgs { /// JSON object merged into root-span metadata for this invocation. #[arg(long, env = "BRAINTRUST_ADDITIONAL_METADATA")] pub additional_metadata: Option, + /// JavaScript span transform for this invocation. Repeat to compose + /// transforms after plugins from persistent setup. + #[arg(long, value_name = "PATH")] + pub plugin: Vec, /// Arguments forwarded verbatim to the coding agent. #[arg(allow_hyphen_values = true)] pub agent_args: Vec, @@ -261,6 +270,11 @@ pub async fn run_hook( .ok() .filter(|value| !value.is_empty()), payload, + plugin_env: if route.span_plugins.is_empty() { + Default::default() + } else { + span_processor::environment() + }, route: Some(route), config: None, }; @@ -465,6 +479,23 @@ pub async fn run_import( mut config: Option, ) -> anyhow::Result<()> { validate_import_selection(&args)?; + let command_plugins = resolve_span_plugin_paths(&args.plugin)?; + if !command_plugins.is_empty() { + let config = config.get_or_insert_with(|| SessionConfig { + auth: wire::BackendAuth { + token: String::new(), + api_url: None, + app_url: None, + org_name: None, + org_id: None, + }, + destination: None, + flush_mode: wire::FlushMode::FireAndForget, + additional_metadata: None, + span_plugins: Vec::new(), + }); + config.span_plugins.extend(command_plugins); + } let destination = args .parent .map(|components| wire::TraceDestination::ParentSpan { components }) @@ -493,8 +524,11 @@ fn validate_import_selection(args: &ImportArgs) -> anyhow::Result<()> { pub async fn run_traced( args: RunArgs, hook_command: RunHookCommand, - route: SessionRoute, + mut route: SessionRoute, ) -> anyhow::Result { + route + .span_plugins + .extend(resolve_span_plugin_paths(&args.plugin)?); if route.destination.is_none() { anyhow::bail!( "managed run requires a trace destination; select a project, object destination, or parent span" @@ -562,6 +596,19 @@ pub async fn run_traced( status } +pub(crate) fn resolve_span_plugin_paths(paths: &[PathBuf]) -> anyhow::Result> { + let paths: Vec<_> = paths + .iter() + .map(|path| { + path.canonicalize().map_err(|error| { + anyhow::anyhow!("could not resolve span plugin {}: {error}", path.display()) + }) + }) + .collect::>()?; + crate::span_processor::validate(&paths)?; + Ok(paths) +} + fn managed_run_args( source: RunSource, hook_command: &RunHookCommand, @@ -800,10 +847,13 @@ pub async fn import_transcripts( } struct ImportLive { + source: String, translator: Box, sink: Box, ctx: SessionCtx, pending_ops: usize, + plugin_env: std::collections::BTreeMap, + plugins_disabled: bool, } struct ImportProcessor { @@ -837,6 +887,7 @@ impl ImportProcessor { self.sessions.insert( sid.clone(), ImportLive { + source: env.source.clone(), translator, sink, ctx: SessionCtx { @@ -844,6 +895,12 @@ impl ImportProcessor { config: None, }, pending_ops: 0, + plugin_env: if env.plugin_env.is_empty() { + crate::span_processor::environment() + } else { + env.plugin_env.clone() + }, + plugins_disabled: false, }, ); self.sessions.get_mut(&sid).unwrap() @@ -853,6 +910,9 @@ impl ImportProcessor { live.sink.configure(cfg); live.ctx.config = Some(cfg.clone()); } + if !env.plugin_env.is_empty() { + live.plugin_env = env.plugin_env.clone(); + } let ops = live.translator.handle(&env, &live.ctx)?; Self::emit_translator_batches(live, ops).await?; } @@ -870,7 +930,41 @@ impl ImportProcessor { // for every native turn boundary. const FLUSH_OPS: usize = 500; for chunk in ops.chunks(FLUSH_OPS) { - live.sink.emit(chunk).await?; + let transformed = if live.plugins_disabled { + chunk.to_vec() + } else { + let plugins = live + .ctx + .config + .as_ref() + .map(|config| config.span_plugins.as_slice()) + .unwrap_or_default(); + let transformed: anyhow::Result> = chunk + .iter() + .map(|op| { + crate::span_processor::process( + plugins, + op, + &live.source, + &live.ctx.session_id, + &live.plugin_env, + ) + }) + .collect(); + match transformed { + Ok(transformed) => transformed, + Err(error) => { + live.plugins_disabled = true; + tracing::warn!( + session_id = %live.ctx.session_id, + %error, + "span plugin failed during import; disabled for session" + ); + chunk.to_vec() + } + } + }; + live.sink.emit(&transformed).await?; live.pending_ops += chunk.len(); if live.pending_ops >= FLUSH_OPS { live.sink.flush().await?; @@ -1037,6 +1131,7 @@ mod tests { parent: None, attach: true, additional_metadata: None, + plugin: Vec::new(), }; assert!(validate_import_selection(&args) .unwrap_err() @@ -1095,6 +1190,7 @@ mod tests { RunArgs { source: RunSource::Codex, additional_metadata: None, + plugin: Vec::new(), agent_args: Vec::new(), }, test_run_hook_command(), diff --git a/bt-daemon/src/settings.rs b/bt-daemon/src/settings.rs index dda7a9e..23367e1 100644 --- a/bt-daemon/src/settings.rs +++ b/bt-daemon/src/settings.rs @@ -84,6 +84,13 @@ impl AgentSettings { pub(crate) fn tracing_enabled(&self) -> bool { self.trace_to_braintrust.unwrap_or(false) } + + pub(crate) fn configured_span_plugins(source: &str) -> Vec { + Self::load_from(&paths::agent_settings_path(source, None)) + .route + .map(|route| route.span_plugins) + .unwrap_or_default() + } } #[cfg(test)] diff --git a/bt-daemon/src/setup.rs b/bt-daemon/src/setup.rs index 4792f78..4971767 100644 --- a/bt-daemon/src/setup.rs +++ b/bt-daemon/src/setup.rs @@ -341,6 +341,13 @@ fn enable_tracing_at(path: &Path, mut route: SessionRoute) -> anyhow::Result<()> .filter(|metadata| metadata.is_object()) .cloned(); } + if route.span_plugins.is_empty() { + route.span_plugins = settings + .get("route") + .and_then(|route| route.get("span_plugins")) + .and_then(|plugins| serde_json::from_value(plugins.clone()).ok()) + .unwrap_or_default(); + } settings.insert("trace_to_braintrust".into(), Value::Bool(true)); settings.insert("route".into(), serde_json::to_value(route)?); settings.remove("traceToBraintrust"); @@ -739,4 +746,32 @@ mod tests { assert_eq!(config["plugin"], serde_json::json!(["other"])); assert_eq!(config["model"], "test/model"); } + + #[test] + fn tracing_settings_preserve_plugins_until_setup_explicitly_replaces_them() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("braintrust.json"); + std::fs::write(&path, r#"{"route":{"span_plugins":["old.mjs"]}}"#).unwrap(); + + enable_tracing_at(&path, SessionRoute::default()).unwrap(); + let settings: Value = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap(); + assert_eq!( + settings["route"]["span_plugins"], + serde_json::json!(["old.mjs"]) + ); + + enable_tracing_at( + &path, + SessionRoute { + span_plugins: vec![PathBuf::from("first.mjs"), PathBuf::from("second.mjs")], + ..SessionRoute::default() + }, + ) + .unwrap(); + let settings: Value = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap(); + assert_eq!( + settings["route"]["span_plugins"], + serde_json::json!(["first.mjs", "second.mjs"]) + ); + } } diff --git a/bt-daemon/src/span_processor.rs b/bt-daemon/src/span_processor.rs new file mode 100644 index 0000000..dc33e9a --- /dev/null +++ b/bt-daemon/src/span_processor.rs @@ -0,0 +1,335 @@ +//! Synchronous JavaScript span transforms. +//! +//! Session actors already execute on Tokio's worker pool. Each worker thread +//! lazily owns one QuickJS runtime and module cache, so JavaScript values never +//! cross threads and unrelated workers can transform spans concurrently. + +use crate::translate::{SpanOp, SpanRow}; +use rquickjs::{Context, Function, Module, Persistent, Runtime}; +use serde::Serialize; +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::Arc; +use std::time::{Duration, Instant, SystemTime}; + +const MEMORY_LIMIT_BYTES: usize = 64 * 1024 * 1024; +const STACK_LIMIT_BYTES: usize = 512 * 1024; +const CALL_TIMEOUT: Duration = Duration::from_millis(50); +const MAX_RESULT_BYTES: usize = 8 * 1024 * 1024; + +thread_local! { + static ENGINE: RefCell> = const { RefCell::new(None) }; +} + +#[derive(Serialize)] +#[serde(rename_all = "snake_case")] +enum Operation { + Insert, + Merge, +} + +#[derive(Serialize)] +struct PluginContext<'a> { + operation: Operation, + source: &'a str, + session_id: &'a str, + env: &'a BTreeMap, +} + +struct Engine { + modules: HashMap, + context: Context, + started: Instant, + deadline_ms: Arc, + // Must drop after every Context and persistent JavaScript value. + _runtime: Runtime, +} + +struct CachedModule { + modified: Option, + len: u64, + function: Persistent>, +} + +impl Engine { + fn new() -> anyhow::Result { + let runtime = Runtime::new()?; + runtime.set_memory_limit(MEMORY_LIMIT_BYTES); + runtime.set_max_stack_size(STACK_LIMIT_BYTES); + let started = Instant::now(); + let deadline_ms = Arc::new(AtomicU64::new(0)); + let interrupt_deadline = deadline_ms.clone(); + let interrupt_started = started; + 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 context = Context::full(&runtime)?; + Ok(Self { + modules: HashMap::new(), + context, + started, + deadline_ms, + _runtime: runtime, + }) + } + + fn load(&mut self, path: &Path) -> anyhow::Result>> { + let metadata = std::fs::metadata(path) + .map_err(|error| anyhow::anyhow!("failed to inspect {}: {error}", path.display()))?; + let modified = metadata.modified().ok(); + if let Some(module) = self.modules.get(path) { + if module.modified == modified && module.len == metadata.len() { + return Ok(module.function.clone()); + } + } + let source = std::fs::read(path) + .map_err(|error| anyhow::anyhow!("failed to read {}: {error}", path.display()))?; + let digest = Sha256::digest(&source); + let name = format!("bt-span-plugin:{digest:x}"); + self.arm_deadline(); + let function = self.context.with(|ctx| -> anyhow::Result<_> { + let (module, promise) = Module::declare(ctx.clone(), name, source)?.eval()?; + promise.finish::<()>()?; + let function: Function<'_> = module + .get("default") + .map_err(|error| anyhow::anyhow!("default export is not a function: {error}"))?; + Ok(Persistent::save(&ctx, function)) + }); + self.deadline_ms.store(0, Ordering::Relaxed); + let function = function?; + self.modules.insert( + path.to_path_buf(), + CachedModule { + modified, + len: metadata.len(), + function: function.clone(), + }, + ); + Ok(function) + } + + fn call( + &mut self, + path: &Path, + row: &SpanRow, + context: &PluginContext<'_>, + ) -> anyhow::Result { + let function = self.load(path)?; + let row_json = serde_json::to_vec(row)?; + let context_json = serde_json::to_vec(context)?; + self.arm_deadline(); + let result = self.context.with(|ctx| -> anyhow::Result> { + let function = function.restore(&ctx)?; + let row = ctx.json_parse(row_json)?; + let context = ctx.json_parse(context_json)?; + let result = function.call::<_, rquickjs::Value<'_>>((row, context))?; + if result.as_promise().is_some() { + anyhow::bail!("plugin returned a Promise; span plugins must be synchronous"); + } + let json = ctx + .json_stringify(result)? + .ok_or_else(|| anyhow::anyhow!("plugin returned a non-JSON value"))?; + Ok(json.to_string()?.into_bytes()) + }); + self.deadline_ms.store(0, Ordering::Relaxed); + let json = result?; + if json.len() > MAX_RESULT_BYTES { + anyhow::bail!( + "plugin returned {} bytes, exceeding the {} byte limit", + json.len(), + MAX_RESULT_BYTES + ); + } + Ok(serde_json::from_slice(&json)?) + } + + fn arm_deadline(&self) { + let deadline = self + .started + .elapsed() + .saturating_add(CALL_TIMEOUT) + .as_millis() as u64; + self.deadline_ms.store(deadline.max(1), Ordering::Relaxed); + } +} + +/// Apply an ordered plugin chain on the worker thread currently executing the +/// session actor. A plugin failure is reported to the caller, which can fail +/// open with the unmodified translator output. +pub fn process( + plugins: &[PathBuf], + op: &SpanOp, + source: &str, + session_id: &str, + env: &BTreeMap, +) -> anyhow::Result { + if plugins.is_empty() { + return Ok(op.clone()); + } + let (operation, mut row) = match op { + SpanOp::Insert(row) => (Operation::Insert, row.clone()), + SpanOp::Merge(row) => (Operation::Merge, row.clone()), + }; + let original_ids = ( + row.span_id.clone(), + row.root_span_id.clone(), + row.parent_span_ids.clone(), + ); + let context = PluginContext { + operation, + source, + session_id, + env, + }; + ENGINE.with_borrow_mut(|slot| -> anyhow::Result<()> { + if slot.is_none() { + *slot = Some(Engine::new()?); + } + let engine = slot.as_mut().expect("engine initialized"); + for plugin in plugins { + row = engine.call(plugin, &row, &context)?; + if ( + row.span_id.as_str(), + row.root_span_id.as_str(), + &row.parent_span_ids, + ) != ( + original_ids.0.as_str(), + original_ids.1.as_str(), + &original_ids.2, + ) { + anyhow::bail!( + "plugin {} changed immutable span identity fields", + plugin.display() + ); + } + } + Ok(()) + })?; + Ok(match op { + SpanOp::Insert(_) => SpanOp::Insert(row), + SpanOp::Merge(_) => SpanOp::Merge(row), + }) +} + +/// Compile each module and verify that it default-exports a function. Explicit +/// CLI commands call this before persisting or launching with a plugin chain. +pub fn validate(plugins: &[PathBuf]) -> anyhow::Result<()> { + ENGINE.with_borrow_mut(|slot| -> anyhow::Result<()> { + if slot.is_none() { + *slot = Some(Engine::new()?); + } + let engine = slot.as_mut().expect("engine initialized"); + for plugin in plugins { + engine.load(plugin)?; + } + Ok(()) + }) +} + +pub fn environment() -> BTreeMap { + std::env::vars_os() + .filter_map(|(key, value)| Some((key.into_string().ok()?, value.into_string().ok()?))) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn row() -> SpanRow { + SpanRow { + span_id: "span".into(), + root_span_id: "root".into(), + name: "original".into(), + ..SpanRow::default() + } + } + + #[test] + fn composes_plugins_and_exposes_context_env() { + let dir = tempfile::tempdir().unwrap(); + let first = dir.path().join("first.mjs"); + let second = dir.path().join("second.mjs"); + std::fs::write( + &first, + "export default (span, context) => ({...span, name: `${context.source}:${context.env.TEAM}:${span.name}`})", + ) + .unwrap(); + std::fs::write( + &second, + "export default span => ({...span, metadata: {second: true}})", + ) + .unwrap(); + let env = BTreeMap::from([("TEAM".into(), "platform".into())]); + let processed = process( + &[first, second], + &SpanOp::Insert(row()), + "codex", + "session", + &env, + ) + .unwrap(); + let SpanOp::Insert(processed) = processed else { + panic!("expected insert") + }; + assert_eq!(processed.name, "codex:platform:original"); + assert_eq!(processed.metadata.unwrap()["second"], true); + } + + #[test] + fn rejects_identity_changes() { + let dir = tempfile::tempdir().unwrap(); + let plugin = dir.path().join("identity.mjs"); + std::fs::write( + &plugin, + "export default span => ({...span, span_id: 'different'})", + ) + .unwrap(); + let error = process( + &[plugin], + &SpanOp::Insert(row()), + "codex", + "session", + &BTreeMap::new(), + ) + .unwrap_err(); + assert!(error.to_string().contains("immutable span identity")); + } + + #[test] + fn interrupts_runaway_plugins_and_rejects_promises() { + let dir = tempfile::tempdir().unwrap(); + let runaway = dir.path().join("runaway.mjs"); + std::fs::write(&runaway, "export default span => { while (true) {} }").unwrap(); + let started = Instant::now(); + assert!(process( + &[runaway], + &SpanOp::Insert(row()), + "codex", + "session", + &BTreeMap::new(), + ) + .is_err()); + assert!(started.elapsed() < Duration::from_secs(2)); + + let asynchronous = dir.path().join("async.mjs"); + std::fs::write( + &asynchronous, + "export default async span => ({...span, name: 'later'})", + ) + .unwrap(); + let error = process( + &[asynchronous], + &SpanOp::Insert(row()), + "codex", + "session", + &BTreeMap::new(), + ) + .unwrap_err(); + assert!(error.to_string().contains("must be synchronous")); + } +} diff --git a/bt-daemon/src/trace_command.rs b/bt-daemon/src/trace_command.rs index 235ed44..5bed608 100644 --- a/bt-daemon/src/trace_command.rs +++ b/bt-daemon/src/trace_command.rs @@ -56,6 +56,10 @@ pub struct EnableArgs { /// JSON object persisted in this agent's tracing route and merged into root-span metadata. #[arg(long, global = true, env = "BRAINTRUST_ADDITIONAL_METADATA")] pub additional_metadata: Option, + /// JavaScript span transform to persist for this agent. Repeat to compose + /// transforms in order. + #[arg(long, global = true, value_name = "PATH")] + pub plugin: Vec, } /// Backwards-compatible API name for hosts that mounted the former setup command. @@ -106,6 +110,7 @@ mod tests { TraceCommand::Setup(SetupArgs { agent: SetupAgent::Claude, additional_metadata: Some(ref value), + .. }) if value == r#"{"setup":true}"# )); @@ -168,4 +173,50 @@ mod tests { }) if value == r#"{"import":true}"# )); } + + #[test] + fn public_commands_preserve_repeated_plugin_order() { + for args in [ + vec![ + "bt", + "setup", + "codex", + "--plugin", + "first.mjs", + "--plugin", + "second.mjs", + ], + vec![ + "bt", + "run", + "--plugin", + "first.mjs", + "--plugin", + "second.mjs", + "codex", + ], + vec![ + "bt", + "import", + "codex", + "session", + "--plugin", + "first.mjs", + "--plugin", + "second.mjs", + ], + ] { + let parsed = Cli::try_parse_from(args).unwrap(); + let plugins = match parsed.trace.command { + TraceCommand::Setup(args) => args.plugin, + TraceCommand::Run(args) => args.plugin, + TraceCommand::Import(args) => args.plugin, + _ => unreachable!(), + }; + assert_eq!( + plugins, + [PathBuf::from("first.mjs"), PathBuf::from("second.mjs")] + ); + } + } } diff --git a/bt-daemon/src/trace_runtime.rs b/bt-daemon/src/trace_runtime.rs index 39a8731..dd495c7 100644 --- a/bt-daemon/src/trace_runtime.rs +++ b/bt-daemon/src/trace_runtime.rs @@ -136,6 +136,7 @@ async fn session_config( destination: route.destination.clone(), flush_mode: route.flush_mode, additional_metadata: route.additional_metadata.clone(), + span_plugins: route.span_plugins.clone(), }) } @@ -185,6 +186,12 @@ fn print_output(output: TraceCommandOutput, format: OutputFormat) -> anyhow::Res Ok(()) } +fn configured_plugins(source: &str) -> anyhow::Result> { + crate::resolve_span_plugin_paths(&crate::settings::AgentSettings::configured_span_plugins( + source, + )) +} + /// Execute the complete mounted trace command. pub async fn run_trace(args: TraceArgs, host: TraceHostContext) -> anyhow::Result<()> { match args.command { @@ -198,6 +205,9 @@ pub async fn run_trace(args: TraceArgs, host: TraceHostContext) -> anyhow::Resul ) .await?; apply_additional_metadata(&mut route, enable_args.additional_metadata.as_deref())?; + if !enable_args.plugin.is_empty() { + route.span_plugins = crate::resolve_span_plugin_paths(&enable_args.plugin)?; + } print_output(run_enable(enable_args, route)?, host.output_format) } TraceCommand::Disable(disable_args) => { @@ -248,6 +258,11 @@ pub async fn run_trace(args: TraceArgs, host: TraceHostContext) -> anyhow::Resul }) .await?; apply_additional_metadata(&mut route, import_args.additional_metadata.as_deref())?; + let source = match import_args.source { + crate::ImportSource::Codex => "codex", + crate::ImportSource::Claude => "claude", + }; + route.span_plugins = configured_plugins(source)?; let config = session_config(&host, &route).await?; run_import(import_args, serve_options(&host), Some(config)).await } @@ -261,6 +276,13 @@ pub async fn run_trace(args: TraceArgs, host: TraceHostContext) -> anyhow::Resul ) .await?; apply_additional_metadata(&mut route, run_args.additional_metadata.as_deref())?; + let source = match run_args.source { + crate::RunSource::Codex => "codex", + crate::RunSource::Claude => "claude", + crate::RunSource::OpenCode => "opencode", + crate::RunSource::Pi => "pi", + }; + route.span_plugins = configured_plugins(source)?; let hook_command = child_command(&host.command, "hook"); let status = run_traced(run_args, hook_command, route).await?; if status.success() { @@ -416,10 +438,12 @@ mod tests { TraceCommand::Setup(SetupArgs { agent: SetupAgent::OpenCode, additional_metadata: None, + plugin: Vec::new(), }), TraceCommand::Run(RunArgs { source: RunSource::Codex, additional_metadata: None, + plugin: Vec::new(), agent_args: Vec::new(), }), ] { @@ -489,6 +513,7 @@ mod tests { parent: None, attach: false, additional_metadata: None, + plugin: Vec::new(), }; let error = run_trace( TraceArgs { diff --git a/bt-daemon/src/transcript_import/mod.rs b/bt-daemon/src/transcript_import/mod.rs index d546fee..42d8687 100644 --- a/bt-daemon/src/transcript_import/mod.rs +++ b/bt-daemon/src/transcript_import/mod.rs @@ -305,6 +305,7 @@ fn envelope( ts_ms, managed_run_id: None, payload, + plugin_env: Default::default(), route: None, config: None, } diff --git a/bt-daemon/src/wire/envelope.rs b/bt-daemon/src/wire/envelope.rs index f61cc4f..71c73d6 100644 --- a/bt-daemon/src/wire/envelope.rs +++ b/bt-daemon/src/wire/envelope.rs @@ -3,6 +3,8 @@ use braintrust_sdk_rust::SpanComponents; use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; +use std::path::PathBuf; /// One captured hook event, forwarded from a shim to the daemon. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -29,6 +31,10 @@ pub struct Envelope { pub managed_run_id: Option, /// The raw agent-native hook payload; opaque except to the translator. pub payload: serde_json::Value, + /// Environment visible to span plugins for this live event. It is sent over + /// the private daemon transport but deliberately omitted from the journal. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub plugin_env: BTreeMap, /// Non-secret, immutable routing intent for this session. New clients use /// this instead of resolving credentials themselves. The daemon host maps /// the selected profile and organization to live credentials. @@ -63,6 +69,10 @@ pub struct SessionRoute { pub flush_mode: FlushMode, #[serde(default, skip_serializing_if = "Option::is_none")] pub additional_metadata: Option, + /// Ordered JavaScript span transforms. Paths are resolved by explicit + /// setup, run, and import commands before entering persistent settings. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub span_plugins: Vec, } impl SessionRoute { @@ -72,12 +82,23 @@ impl SessionRoute { destination: self.destination.clone(), flush_mode: self.flush_mode, additional_metadata: self.additional_metadata.clone(), + span_plugins: self.span_plugins.clone(), } } pub fn same_route(&self, other: &Self) -> bool { serde_json::to_value(self).ok() == serde_json::to_value(other).ok() } + + /// Raw journal entries can be replayed through a newer plugin chain as + /// long as their Braintrust delivery route is otherwise unchanged. + pub fn same_replay_route(&self, other: &Self) -> bool { + let mut left = self.clone(); + let mut right = other.clone(); + left.span_plugins.clear(); + right.span_plugins.clear(); + left.same_route(&right) + } } /// Trace settings and backend credentials resolved by the shim. @@ -91,6 +112,8 @@ pub struct SessionConfig { pub flush_mode: FlushMode, #[serde(default, skip_serializing_if = "Option::is_none")] pub additional_metadata: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub span_plugins: Vec, } /// Where a session's root span should be logged. @@ -242,6 +265,7 @@ mod tests { ts_ms: 1_753_639_552_123, managed_run_id: Some("run-1".into()), payload: serde_json::json!({ "session_id": "sess-1", "tool_name": "shell" }), + plugin_env: BTreeMap::from([("SECRET".into(), "not-journaled".into())]), route: Some(SessionRoute { auth: AuthSelection { profile: Some("work".into()), @@ -264,6 +288,7 @@ mod tests { destination: None, flush_mode: FlushMode::FireAndForget, additional_metadata: None, + span_plugins: Vec::new(), }), } } diff --git a/bt-daemon/tests/braintrust_sink.rs b/bt-daemon/tests/braintrust_sink.rs index 4cc622a..79aa4d7 100644 --- a/bt-daemon/tests/braintrust_sink.rs +++ b/bt-daemon/tests/braintrust_sink.rs @@ -26,6 +26,7 @@ fn session_config(base: &str) -> SessionConfig { }), flush_mode: FlushMode::FireAndForget, additional_metadata: None, + span_plugins: Vec::new(), } } diff --git a/bt-daemon/tests/claude_translator.rs b/bt-daemon/tests/claude_translator.rs index e8d7e2f..5853ed0 100644 --- a/bt-daemon/tests/claude_translator.rs +++ b/bt-daemon/tests/claude_translator.rs @@ -78,6 +78,7 @@ fn replay_from(name: &str, source: Source) -> Vec { ts_ms, managed_run_id: None, payload, + plugin_env: Default::default(), route: None, config: None, }; @@ -269,6 +270,7 @@ fn claude_additional_metadata_reaches_roots_without_overriding_session_fields() ts_ms: 1, managed_run_id: None, payload: json!({"session_id":"session","cwd":"/workspace","prompt":"go"}), + plugin_env: Default::default(), route: None, config: None, }, @@ -368,6 +370,7 @@ fn claude_permission_denied_and_failed_tools_are_first_class_spans() { ts_ms: 1, managed_run_id: None, payload, + plugin_env: Default::default(), route: None, config: None, }; @@ -457,6 +460,7 @@ fn claude_pairs_tool_lifecycle_and_marks_explicit_skills_and_stop_failures() { ts_ms, managed_run_id: None, payload, + plugin_env: Default::default(), route: None, config: None, }; @@ -576,6 +580,7 @@ fn claude_groups_streamed_rows_and_reads_late_final_output_at_session_end() { ts_ms, managed_run_id: None, payload, + plugin_env: Default::default(), route: None, config: None, }; @@ -712,6 +717,7 @@ fn claude_large_catch_up_emits_one_historical_snapshot_per_batch() { ts_ms: 2_000_000_000_000, managed_run_id: None, payload, + plugin_env: Default::default(), route: None, config: None, }; diff --git a/bt-daemon/tests/codex_translator.rs b/bt-daemon/tests/codex_translator.rs index 955d574..72c3514 100644 --- a/bt-daemon/tests/codex_translator.rs +++ b/bt-daemon/tests/codex_translator.rs @@ -63,6 +63,7 @@ fn envelope(session: &str, event: &str, transcript_path: &str, extra: Value) -> ts_ms: 0, managed_run_id: None, payload, + plugin_env: Default::default(), route: None, config: None, } @@ -493,6 +494,7 @@ fn configured_ctx(session_id: &str, additional_metadata: Value) -> SessionCtx { }), flush_mode: FlushMode::FireAndForget, additional_metadata: Some(additional_metadata), + span_plugins: Vec::new(), }), } } diff --git a/bt-daemon/tests/opencode_translator.rs b/bt-daemon/tests/opencode_translator.rs index 0d4bc69..c041e71 100644 --- a/bt-daemon/tests/opencode_translator.rs +++ b/bt-daemon/tests/opencode_translator.rs @@ -13,6 +13,7 @@ fn event(name: &str, ts_ms: i64, payload: serde_json::Value) -> Envelope { ts_ms, managed_run_id: None, payload, + plugin_env: Default::default(), route: None, config: None, } diff --git a/bt-daemon/tests/pi_translator.rs b/bt-daemon/tests/pi_translator.rs index 1d089f8..1659c26 100644 --- a/bt-daemon/tests/pi_translator.rs +++ b/bt-daemon/tests/pi_translator.rs @@ -13,6 +13,7 @@ fn event(name: &str, ts_ms: i64, native: serde_json::Value) -> Envelope { ts_ms, managed_run_id: None, payload: json!({"event":native,"extension_version":"1.0.0","cwd":"."}), + plugin_env: Default::default(), route: None, config: None, } diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 44bbeae..480ef9d 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -133,6 +133,7 @@ fn envelope(session_id: &str, event: &str, ts_ms: i64) -> Envelope { ts_ms, managed_run_id: None, payload: serde_json::json!({ "session_id": session_id, "hook_event_name": event, "n": ts_ms }), + plugin_env: Default::default(), route: Some(SessionRoute { destination: Some(bt_daemon::wire::TraceDestination::ProjectLogs { project_id: None, @@ -980,6 +981,120 @@ async fn restart_replays_journal_with_stable_span_ids_before_new_events() { second.await.unwrap(); } +#[tokio::test] +async fn span_plugins_transform_live_and_replayed_rows_with_event_environment() { + let (data_dir, socket, first, tmp) = start_daemon().await; + let host = dummy_host(); + let first_plugin = tmp.path().join("first.mjs"); + let second_plugin = tmp.path().join("second.mjs"); + std::fs::write( + &first_plugin, + "export default (span, context) => ({...span, name: `${context.env.PREFIX}:${span.name}`})", + ) + .unwrap(); + std::fs::write( + &second_plugin, + "export default (span, context) => ({...span, name: `${context.operation}:current:${span.name}`})", + ) + .unwrap(); + + let mut start = envelope("plugin-replay", "SessionStart", 1); + start + .route + .as_mut() + .unwrap() + .span_plugins + .push(first_plugin); + start.plugin_env = std::collections::BTreeMap::from([ + ("PREFIX".into(), "live".into()), + ("PLUGIN_SECRET".into(), "must-not-be-journaled".into()), + ]); + forward_envelope(&start, &socket, &host, false) + .await + .unwrap(); + flush_session("plugin-replay", &socket, 5000).await.unwrap(); + shutdown(&socket).await; + first.await.unwrap(); + + let journal = std::fs::read_to_string(data_dir.join("journal/plugin-replay.ndjson")).unwrap(); + assert!(!journal.contains("PLUGIN_SECRET")); + assert!(!journal.contains("must-not-be-journaled")); + + let second = start_daemon_at(data_dir.clone(), socket.clone()).await; + let mut stop = envelope("plugin-replay", "Stop", 2); + stop.route + .as_mut() + .unwrap() + .span_plugins + .push(second_plugin); + forward_envelope(&stop, &socket, &host, false) + .await + .unwrap(); + flush_session("plugin-replay", &socket, 5000).await.unwrap(); + + let spans = std::fs::read_to_string(data_dir.join("spans/plugin-replay.ndjson")).unwrap(); + let names: Vec<_> = spans + .lines() + .filter_map(|line| { + let value: serde_json::Value = serde_json::from_str(line).unwrap(); + value + .get("Insert") + .or_else(|| value.get("Merge")) + .and_then(|row| row.get("name")) + .and_then(serde_json::Value::as_str) + .map(str::to_owned) + }) + .collect(); + assert!(names.iter().any(|name| name.starts_with("live:"))); + assert!( + names.iter().any(|name| name.contains(":current:")), + "the new plugin chain should process replayed and live rows: {names:?}" + ); + + shutdown(&socket).await; + second.await.unwrap(); +} + +#[tokio::test] +async fn a_failing_span_plugin_is_reported_and_fails_open() { + let (data_dir, socket, handle, tmp) = start_daemon().await; + let plugin = tmp.path().join("bad.mjs"); + std::fs::write( + &plugin, + "export default span => ({...span, span_id: 'corrupt'})", + ) + .unwrap(); + let mut env = envelope("plugin-failure", "SessionStart", 1); + env.route.as_mut().unwrap().span_plugins.push(plugin); + forward_envelope(&env, &socket, &dummy_host(), false) + .await + .unwrap(); + flush_session("plugin-failure", &socket, 5000) + .await + .unwrap(); + + let status = run_status(StatusArgs { + socket: Some(socket.clone()), + session_id: Some("plugin-failure".into()), + }) + .await + .unwrap() + .unwrap(); + assert!(status.sessions[0] + .last_error + .as_deref() + .is_some_and(|error| error.contains("span plugin failed"))); + let spans = std::fs::read_to_string(data_dir.join("spans/plugin-failure.ndjson")).unwrap(); + assert!(!spans.contains("corrupt")); + assert!( + !spans.is_empty(), + "the original rows should still be delivered" + ); + + shutdown(&socket).await; + handle.await.unwrap(); +} + #[tokio::test] async fn claude_boundary_journal_references_a_self_contained_transcript_mirror() { let (data_dir, socket, handle, tmp) = start_daemon().await; @@ -1310,6 +1425,7 @@ esac RunArgs { source: RunSource::Codex, additional_metadata: None, + plugin: Vec::new(), agent_args: vec![session_id.into(), mode.into()], }, RunHookCommand { diff --git a/bt-daemon/tests/replay.rs b/bt-daemon/tests/replay.rs index 4ac4120..d7d70f3 100644 --- a/bt-daemon/tests/replay.rs +++ b/bt-daemon/tests/replay.rs @@ -1,3 +1,4 @@ +use bt_daemon::wire::{BackendAuth, FlushMode, SessionConfig}; use bt_daemon::{ import_transcript, import_transcripts, DebugSinkFactory, ImportSource, Registry, ServeOptions, }; @@ -69,6 +70,56 @@ fn inserted(rows: &[Value], span_type: &str) -> usize { .count() } +#[tokio::test] +async fn import_uses_the_same_span_plugin_stage() { + let tmp = tempfile::tempdir().unwrap(); + let transcript = tmp.path().join("plugin-import.jsonl"); + write_jsonl( + &transcript, + &[ + json!({"timestamp":"2026-01-01T00:00:01Z","type":"session_meta","payload":{"id":"plugin-import","cwd":"/tmp/demo"}}), + json!({"timestamp":"2026-01-01T00:00:02Z","type":"event_msg","payload":{"type":"task_started","turn_id":"turn"}}), + json!({"timestamp":"2026-01-01T00:00:03Z","type":"event_msg","payload":{"type":"task_complete","last_agent_message":"done"}}), + ], + ); + let plugin = tmp.path().join("import.mjs"); + std::fs::write( + &plugin, + "export default span => ({...span, name: `imported:${span.name}`})", + ) + .unwrap(); + let output = tmp.path().join("spans"); + import_transcript( + &transcript, + ImportSource::Codex, + options(&output), + Some(SessionConfig { + auth: BackendAuth { + token: String::new(), + api_url: None, + app_url: None, + org_name: None, + org_id: None, + }, + destination: None, + flush_mode: FlushMode::FireAndForget, + additional_metadata: None, + span_plugins: vec![plugin], + }), + false, + ) + .await + .unwrap(); + + let output = rows(&output.join("plugin-import.ndjson")); + assert!(output + .iter() + .filter_map(|op| op.get("Insert")) + .all(|row| row["name"] + .as_str() + .is_some_and(|name| name.starts_with("imported:")))); +} + #[tokio::test] async fn imports_multiple_transcripts_in_one_invocation() { let tmp = tempfile::tempdir().unwrap(); diff --git a/src/runtime/js-daemon-client/src/index.ts b/src/runtime/js-daemon-client/src/index.ts index 1e6c6b9..91f7f2a 100644 --- a/src/runtime/js-daemon-client/src/index.ts +++ b/src/runtime/js-daemon-client/src/index.ts @@ -14,6 +14,7 @@ export interface DaemonSessionRoute { destination: unknown flush_mode?: "fire_and_forget" | "flush_on_turn_end" additional_metadata?: Record + span_plugins?: string[] } export interface DaemonTraceSettings { @@ -55,6 +56,7 @@ export interface DaemonEnvelope { ts_ms: number managed_run_id?: string payload: unknown + plugin_env?: Record route?: DaemonSessionRoute } @@ -145,6 +147,17 @@ export class DaemonClient { return this.serial(async () => { const event = { ...envelope, + ...(envelope.plugin_env + ? { plugin_env: envelope.plugin_env } + : envelope.route?.span_plugins?.length + ? { + plugin_env: Object.fromEntries( + Object.entries(process.env).filter( + (entry): entry is [string, string] => entry[1] !== undefined, + ), + ), + } + : {}), ...(this.options.pluginVersion ? { plugin_version: this.options.pluginVersion } : {}), ...(this.options.managedRunId ? { managed_run_id: this.options.managedRunId } : {}), } diff --git a/src/runtime/js-daemon-client/tests/client.test.ts b/src/runtime/js-daemon-client/tests/client.test.ts index 53eff79..2eef81a 100644 --- a/src/runtime/js-daemon-client/tests/client.test.ts +++ b/src/runtime/js-daemon-client/tests/client.test.ts @@ -110,6 +110,7 @@ test("serializes initialize, events, flush, and status over one connection", asy event: name, ts_ms: Date.now(), payload: {}, + route: { destination: {}, span_plugins: ["plugin.mjs"] }, }) assert.deepEqual(await Promise.all([client.log(envelope("one")), client.log(envelope("two"))]), [ true, @@ -126,6 +127,10 @@ test("serializes initialize, events, flush, and status over one connection", asy ]) assert.deepEqual(eventParams.map((params) => params.plugin_version), ["1.0.0", "1.0.0"]) assert.deepEqual(eventParams.map((params) => params.managed_run_id), ["run-123", "run-123"]) + assert.equal( + (eventParams[0]?.plugin_env as Record | undefined)?.PATH, + process.env.PATH, + ) await client.close() await new Promise((resolve) => server.close(() => resolve())) rmSync(temp, { recursive: true, force: true }) From 9e30d64e5fed80a8d951c38aa14b6e19a7a4a0dc Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 22:38:51 +0800 Subject: [PATCH 2/8] Read plugin environment from daemon workers --- bt-daemon/README.md | 13 ++-- bt-daemon/docs/protocol.md | 10 +-- bt-daemon/src/dispatch.rs | 11 +-- bt-daemon/src/journal.rs | 1 - bt-daemon/src/lib.rs | 15 ---- bt-daemon/src/span_processor.rs | 73 +++++++------------ bt-daemon/src/transcript_import/mod.rs | 1 - bt-daemon/src/wire/envelope.rs | 6 -- bt-daemon/tests/claude_translator.rs | 6 -- bt-daemon/tests/codex_translator.rs | 1 - bt-daemon/tests/opencode_translator.rs | 1 - bt-daemon/tests/pi_translator.rs | 1 - bt-daemon/tests/pipeline.rs | 16 +--- src/runtime/js-daemon-client/src/index.ts | 12 --- .../js-daemon-client/tests/client.test.ts | 4 - 15 files changed, 41 insertions(+), 130 deletions(-) diff --git a/bt-daemon/README.md b/bt-daemon/README.md index e0187c7..7d76988 100644 --- a/bt-daemon/README.md +++ b/bt-daemon/README.md @@ -85,13 +85,12 @@ export default function mapSpan(span, context) { } ``` -The environment map comes from the event-producing hook or adapter and is sent -only over the private daemon transport; it is never written to the recovery -journal. 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. A -runtime failure disables the chain for that session and sends the original -translator output instead. +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. A runtime failure disables the chain for that +session and sends the original translator output instead. ### Additional root metadata diff --git a/bt-daemon/docs/protocol.md b/bt-daemon/docs/protocol.md index 1d19bc2..b7e8c94 100644 --- a/bt-daemon/docs/protocol.md +++ b/bt-daemon/docs/protocol.md @@ -188,7 +188,6 @@ Used for version handover and by tests. "ts_ms": 1753639552123, "managed_run_id": "invocation-uuid", "payload": { "…raw agent-native hook payload…": true }, - "plugin_env": { "CI": "true" }, "route": { "auth": { "profile": "work", @@ -222,8 +221,6 @@ Field notes: daemon. - **`payload`** is opaque to transport and to everything except the translator for `source`. -- **`plugin_env`** is the event producer's string environment map, exposed to - span plugins as `context.env`. It is transported live but never journaled. - **`managed_run_id`** is present only for events inherited from a `bt trace run` process tree. It groups native sessions for the final invocation flush and is not trace metadata. @@ -251,9 +248,10 @@ Field notes: ### Redaction Live credentials returned by the host provider are **never** written to the -journal, logs, status, or RPC response. The live plugin environment is likewise -omitted because it commonly contains secrets. Envelopes journal only their -non-secret `route`, allowing restart recovery to resolve a fresh lease. +journal, logs, status, or RPC response. Envelopes journal only their non-secret +`route`, allowing restart recovery to resolve a fresh lease. Span plugins read +an environment snapshot captured inside their daemon worker process; it is not +part of the envelope or journal schema. ## Daemon lifecycle diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 7864278..c7f0972 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -11,7 +11,6 @@ use crate::sink::SinkFactory; use crate::translate::{Registry, SessionCtx}; use crate::wire::{Envelope, SessionRoute}; -use std::collections::BTreeMap; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; @@ -216,7 +215,6 @@ struct SessionActor { } struct PluginState { - env: BTreeMap, disabled: bool, } @@ -257,10 +255,7 @@ impl SessionActor { session_id: self.session_id.clone(), config: Some(self.config.clone()), }; - let mut plugins = PluginState { - env: crate::span_processor::environment(), - disabled: false, - }; + let mut plugins = PluginState { disabled: false }; sink.configure(&self.config); self.refresh_permalink(sink.as_ref()); // Rebuild translator state before accepting the first new event. @@ -272,9 +267,6 @@ impl SessionActor { while let Some(msg) = rx.recv().await { match msg { SessionMsg::Event(env) => { - if !env.plugin_env.is_empty() { - plugins.env = env.plugin_env.clone(); - } if let Some(cfg) = &env.config { sink.configure(cfg); ctx.config = Some(cfg.clone()); @@ -351,7 +343,6 @@ impl SessionActor { op, &self.source, &self.session_id, - &plugins.env, ) }) .collect(); diff --git a/bt-daemon/src/journal.rs b/bt-daemon/src/journal.rs index e83759e..17c71e7 100644 --- a/bt-daemon/src/journal.rs +++ b/bt-daemon/src/journal.rs @@ -263,7 +263,6 @@ pub fn envelope_from_redacted(r: RedactedEnvelope) -> Envelope { ts_ms: r.ts_ms, managed_run_id: r.managed_run_id, payload: r.payload, - plugin_env: std::collections::BTreeMap::new(), route, config, } diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 5edb045..fc1599b 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -270,11 +270,6 @@ pub async fn run_hook( .ok() .filter(|value| !value.is_empty()), payload, - plugin_env: if route.span_plugins.is_empty() { - Default::default() - } else { - span_processor::environment() - }, route: Some(route), config: None, }; @@ -852,7 +847,6 @@ struct ImportLive { sink: Box, ctx: SessionCtx, pending_ops: usize, - plugin_env: std::collections::BTreeMap, plugins_disabled: bool, } @@ -895,11 +889,6 @@ impl ImportProcessor { config: None, }, pending_ops: 0, - plugin_env: if env.plugin_env.is_empty() { - crate::span_processor::environment() - } else { - env.plugin_env.clone() - }, plugins_disabled: false, }, ); @@ -910,9 +899,6 @@ impl ImportProcessor { live.sink.configure(cfg); live.ctx.config = Some(cfg.clone()); } - if !env.plugin_env.is_empty() { - live.plugin_env = env.plugin_env.clone(); - } let ops = live.translator.handle(&env, &live.ctx)?; Self::emit_translator_batches(live, ops).await?; } @@ -947,7 +933,6 @@ impl ImportProcessor { op, &live.source, &live.ctx.session_id, - &live.plugin_env, ) }) .collect(); diff --git a/bt-daemon/src/span_processor.rs b/bt-daemon/src/span_processor.rs index dc33e9a..844ca52 100644 --- a/bt-daemon/src/span_processor.rs +++ b/bt-daemon/src/span_processor.rs @@ -24,7 +24,7 @@ thread_local! { static ENGINE: RefCell> = const { RefCell::new(None) }; } -#[derive(Serialize)] +#[derive(Clone, Copy, Serialize)] #[serde(rename_all = "snake_case")] enum Operation { Insert, @@ -41,6 +41,7 @@ struct PluginContext<'a> { struct Engine { modules: HashMap, + env: BTreeMap, context: Context, started: Instant, deadline_ms: Arc, @@ -70,6 +71,7 @@ impl Engine { let context = Context::full(&runtime)?; Ok(Self { modules: HashMap::new(), + env: environment(), context, started, deadline_ms, @@ -116,11 +118,19 @@ impl Engine { &mut self, path: &Path, row: &SpanRow, - context: &PluginContext<'_>, + operation: Operation, + source: &str, + session_id: &str, ) -> anyhow::Result { let function = self.load(path)?; + let context = PluginContext { + operation, + source, + session_id, + env: &self.env, + }; let row_json = serde_json::to_vec(row)?; - let context_json = serde_json::to_vec(context)?; + let context_json = serde_json::to_vec(&context)?; self.arm_deadline(); let result = self.context.with(|ctx| -> anyhow::Result> { let function = function.restore(&ctx)?; @@ -165,7 +175,6 @@ pub fn process( op: &SpanOp, source: &str, session_id: &str, - env: &BTreeMap, ) -> anyhow::Result { if plugins.is_empty() { return Ok(op.clone()); @@ -179,19 +188,13 @@ pub fn process( row.root_span_id.clone(), row.parent_span_ids.clone(), ); - let context = PluginContext { - operation, - source, - session_id, - env, - }; ENGINE.with_borrow_mut(|slot| -> anyhow::Result<()> { if slot.is_none() { *slot = Some(Engine::new()?); } let engine = slot.as_mut().expect("engine initialized"); for plugin in plugins { - row = engine.call(plugin, &row, &context)?; + row = engine.call(plugin, &row, operation, source, session_id)?; if ( row.span_id.as_str(), row.root_span_id.as_str(), @@ -230,7 +233,7 @@ pub fn validate(plugins: &[PathBuf]) -> anyhow::Result<()> { }) } -pub fn environment() -> BTreeMap { +fn environment() -> BTreeMap { std::env::vars_os() .filter_map(|(key, value)| Some((key.into_string().ok()?, value.into_string().ok()?))) .collect() @@ -256,7 +259,7 @@ mod tests { let second = dir.path().join("second.mjs"); std::fs::write( &first, - "export default (span, context) => ({...span, name: `${context.source}:${context.env.TEAM}:${span.name}`})", + "export default (span, context) => ({...span, name: `${context.source}:${context.env.PATH}:${span.name}`})", ) .unwrap(); std::fs::write( @@ -264,19 +267,15 @@ mod tests { "export default span => ({...span, metadata: {second: true}})", ) .unwrap(); - let env = BTreeMap::from([("TEAM".into(), "platform".into())]); - let processed = process( - &[first, second], - &SpanOp::Insert(row()), - "codex", - "session", - &env, - ) - .unwrap(); + let processed = + process(&[first, second], &SpanOp::Insert(row()), "codex", "session").unwrap(); let SpanOp::Insert(processed) = processed else { panic!("expected insert") }; - assert_eq!(processed.name, "codex:platform:original"); + assert_eq!( + processed.name, + format!("codex:{}:original", std::env::var("PATH").unwrap()) + ); assert_eq!(processed.metadata.unwrap()["second"], true); } @@ -289,14 +288,7 @@ mod tests { "export default span => ({...span, span_id: 'different'})", ) .unwrap(); - let error = process( - &[plugin], - &SpanOp::Insert(row()), - "codex", - "session", - &BTreeMap::new(), - ) - .unwrap_err(); + let error = process(&[plugin], &SpanOp::Insert(row()), "codex", "session").unwrap_err(); assert!(error.to_string().contains("immutable span identity")); } @@ -306,14 +298,7 @@ mod tests { let runaway = dir.path().join("runaway.mjs"); std::fs::write(&runaway, "export default span => { while (true) {} }").unwrap(); let started = Instant::now(); - assert!(process( - &[runaway], - &SpanOp::Insert(row()), - "codex", - "session", - &BTreeMap::new(), - ) - .is_err()); + assert!(process(&[runaway], &SpanOp::Insert(row()), "codex", "session").is_err()); assert!(started.elapsed() < Duration::from_secs(2)); let asynchronous = dir.path().join("async.mjs"); @@ -322,14 +307,8 @@ mod tests { "export default async span => ({...span, name: 'later'})", ) .unwrap(); - let error = process( - &[asynchronous], - &SpanOp::Insert(row()), - "codex", - "session", - &BTreeMap::new(), - ) - .unwrap_err(); + let error = + process(&[asynchronous], &SpanOp::Insert(row()), "codex", "session").unwrap_err(); assert!(error.to_string().contains("must be synchronous")); } } diff --git a/bt-daemon/src/transcript_import/mod.rs b/bt-daemon/src/transcript_import/mod.rs index 42d8687..d546fee 100644 --- a/bt-daemon/src/transcript_import/mod.rs +++ b/bt-daemon/src/transcript_import/mod.rs @@ -305,7 +305,6 @@ fn envelope( ts_ms, managed_run_id: None, payload, - plugin_env: Default::default(), route: None, config: None, } diff --git a/bt-daemon/src/wire/envelope.rs b/bt-daemon/src/wire/envelope.rs index 71c73d6..a96947a 100644 --- a/bt-daemon/src/wire/envelope.rs +++ b/bt-daemon/src/wire/envelope.rs @@ -3,7 +3,6 @@ use braintrust_sdk_rust::SpanComponents; use serde::{Deserialize, Serialize}; -use std::collections::BTreeMap; use std::path::PathBuf; /// One captured hook event, forwarded from a shim to the daemon. @@ -31,10 +30,6 @@ pub struct Envelope { pub managed_run_id: Option, /// The raw agent-native hook payload; opaque except to the translator. pub payload: serde_json::Value, - /// Environment visible to span plugins for this live event. It is sent over - /// the private daemon transport but deliberately omitted from the journal. - #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] - pub plugin_env: BTreeMap, /// Non-secret, immutable routing intent for this session. New clients use /// this instead of resolving credentials themselves. The daemon host maps /// the selected profile and organization to live credentials. @@ -265,7 +260,6 @@ mod tests { ts_ms: 1_753_639_552_123, managed_run_id: Some("run-1".into()), payload: serde_json::json!({ "session_id": "sess-1", "tool_name": "shell" }), - plugin_env: BTreeMap::from([("SECRET".into(), "not-journaled".into())]), route: Some(SessionRoute { auth: AuthSelection { profile: Some("work".into()), diff --git a/bt-daemon/tests/claude_translator.rs b/bt-daemon/tests/claude_translator.rs index 5853ed0..e8d7e2f 100644 --- a/bt-daemon/tests/claude_translator.rs +++ b/bt-daemon/tests/claude_translator.rs @@ -78,7 +78,6 @@ fn replay_from(name: &str, source: Source) -> Vec { ts_ms, managed_run_id: None, payload, - plugin_env: Default::default(), route: None, config: None, }; @@ -270,7 +269,6 @@ fn claude_additional_metadata_reaches_roots_without_overriding_session_fields() ts_ms: 1, managed_run_id: None, payload: json!({"session_id":"session","cwd":"/workspace","prompt":"go"}), - plugin_env: Default::default(), route: None, config: None, }, @@ -370,7 +368,6 @@ fn claude_permission_denied_and_failed_tools_are_first_class_spans() { ts_ms: 1, managed_run_id: None, payload, - plugin_env: Default::default(), route: None, config: None, }; @@ -460,7 +457,6 @@ fn claude_pairs_tool_lifecycle_and_marks_explicit_skills_and_stop_failures() { ts_ms, managed_run_id: None, payload, - plugin_env: Default::default(), route: None, config: None, }; @@ -580,7 +576,6 @@ fn claude_groups_streamed_rows_and_reads_late_final_output_at_session_end() { ts_ms, managed_run_id: None, payload, - plugin_env: Default::default(), route: None, config: None, }; @@ -717,7 +712,6 @@ fn claude_large_catch_up_emits_one_historical_snapshot_per_batch() { ts_ms: 2_000_000_000_000, managed_run_id: None, payload, - plugin_env: Default::default(), route: None, config: None, }; diff --git a/bt-daemon/tests/codex_translator.rs b/bt-daemon/tests/codex_translator.rs index 72c3514..5f5a12d 100644 --- a/bt-daemon/tests/codex_translator.rs +++ b/bt-daemon/tests/codex_translator.rs @@ -63,7 +63,6 @@ fn envelope(session: &str, event: &str, transcript_path: &str, extra: Value) -> ts_ms: 0, managed_run_id: None, payload, - plugin_env: Default::default(), route: None, config: None, } diff --git a/bt-daemon/tests/opencode_translator.rs b/bt-daemon/tests/opencode_translator.rs index c041e71..0d4bc69 100644 --- a/bt-daemon/tests/opencode_translator.rs +++ b/bt-daemon/tests/opencode_translator.rs @@ -13,7 +13,6 @@ fn event(name: &str, ts_ms: i64, payload: serde_json::Value) -> Envelope { ts_ms, managed_run_id: None, payload, - plugin_env: Default::default(), route: None, config: None, } diff --git a/bt-daemon/tests/pi_translator.rs b/bt-daemon/tests/pi_translator.rs index 1659c26..1d089f8 100644 --- a/bt-daemon/tests/pi_translator.rs +++ b/bt-daemon/tests/pi_translator.rs @@ -13,7 +13,6 @@ fn event(name: &str, ts_ms: i64, native: serde_json::Value) -> Envelope { ts_ms, managed_run_id: None, payload: json!({"event":native,"extension_version":"1.0.0","cwd":"."}), - plugin_env: Default::default(), route: None, config: None, } diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 480ef9d..57b2d82 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -133,7 +133,6 @@ fn envelope(session_id: &str, event: &str, ts_ms: i64) -> Envelope { ts_ms, managed_run_id: None, payload: serde_json::json!({ "session_id": session_id, "hook_event_name": event, "n": ts_ms }), - plugin_env: Default::default(), route: Some(SessionRoute { destination: Some(bt_daemon::wire::TraceDestination::ProjectLogs { project_id: None, @@ -982,14 +981,14 @@ async fn restart_replays_journal_with_stable_span_ids_before_new_events() { } #[tokio::test] -async fn span_plugins_transform_live_and_replayed_rows_with_event_environment() { +async fn span_plugins_transform_live_and_replayed_rows_with_daemon_environment() { let (data_dir, socket, first, tmp) = start_daemon().await; let host = dummy_host(); let first_plugin = tmp.path().join("first.mjs"); let second_plugin = tmp.path().join("second.mjs"); std::fs::write( &first_plugin, - "export default (span, context) => ({...span, name: `${context.env.PREFIX}:${span.name}`})", + "export default (span, context) => ({...span, name: `${context.env.PATH}:${span.name}`})", ) .unwrap(); std::fs::write( @@ -1005,10 +1004,6 @@ async fn span_plugins_transform_live_and_replayed_rows_with_event_environment() .unwrap() .span_plugins .push(first_plugin); - start.plugin_env = std::collections::BTreeMap::from([ - ("PREFIX".into(), "live".into()), - ("PLUGIN_SECRET".into(), "must-not-be-journaled".into()), - ]); forward_envelope(&start, &socket, &host, false) .await .unwrap(); @@ -1016,10 +1011,6 @@ async fn span_plugins_transform_live_and_replayed_rows_with_event_environment() shutdown(&socket).await; first.await.unwrap(); - let journal = std::fs::read_to_string(data_dir.join("journal/plugin-replay.ndjson")).unwrap(); - assert!(!journal.contains("PLUGIN_SECRET")); - assert!(!journal.contains("must-not-be-journaled")); - let second = start_daemon_at(data_dir.clone(), socket.clone()).await; let mut stop = envelope("plugin-replay", "Stop", 2); stop.route @@ -1045,7 +1036,8 @@ async fn span_plugins_transform_live_and_replayed_rows_with_event_environment() .map(str::to_owned) }) .collect(); - assert!(names.iter().any(|name| name.starts_with("live:"))); + let path_prefix = format!("{}:", std::env::var("PATH").unwrap()); + assert!(names.iter().any(|name| name.starts_with(&path_prefix))); assert!( names.iter().any(|name| name.contains(":current:")), "the new plugin chain should process replayed and live rows: {names:?}" diff --git a/src/runtime/js-daemon-client/src/index.ts b/src/runtime/js-daemon-client/src/index.ts index 91f7f2a..e99f214 100644 --- a/src/runtime/js-daemon-client/src/index.ts +++ b/src/runtime/js-daemon-client/src/index.ts @@ -56,7 +56,6 @@ export interface DaemonEnvelope { ts_ms: number managed_run_id?: string payload: unknown - plugin_env?: Record route?: DaemonSessionRoute } @@ -147,17 +146,6 @@ export class DaemonClient { return this.serial(async () => { const event = { ...envelope, - ...(envelope.plugin_env - ? { plugin_env: envelope.plugin_env } - : envelope.route?.span_plugins?.length - ? { - plugin_env: Object.fromEntries( - Object.entries(process.env).filter( - (entry): entry is [string, string] => entry[1] !== undefined, - ), - ), - } - : {}), ...(this.options.pluginVersion ? { plugin_version: this.options.pluginVersion } : {}), ...(this.options.managedRunId ? { managed_run_id: this.options.managedRunId } : {}), } diff --git a/src/runtime/js-daemon-client/tests/client.test.ts b/src/runtime/js-daemon-client/tests/client.test.ts index 2eef81a..1e9ec75 100644 --- a/src/runtime/js-daemon-client/tests/client.test.ts +++ b/src/runtime/js-daemon-client/tests/client.test.ts @@ -127,10 +127,6 @@ test("serializes initialize, events, flush, and status over one connection", asy ]) assert.deepEqual(eventParams.map((params) => params.plugin_version), ["1.0.0", "1.0.0"]) assert.deepEqual(eventParams.map((params) => params.managed_run_id), ["run-123", "run-123"]) - assert.equal( - (eventParams[0]?.plugin_env as Record | undefined)?.PATH, - process.env.PATH, - ) await client.close() await new Promise((resolve) => server.close(() => resolve())) rmSync(temp, { recursive: true, force: true }) From aee3a1b4d4e3f53e79361592fe1344f26b781ed9 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 22:55:06 +0800 Subject: [PATCH 3/8] Isolate span plugin failures per worker --- bt-daemon/README.md | 7 +- bt-daemon/src/dispatch.rs | 73 ++++++--------- bt-daemon/src/lib.rs | 72 +++++++++------ bt-daemon/src/span_processor.rs | 153 +++++++++++++++++++++++++------- bt-daemon/tests/pipeline.rs | 25 ++++-- 5 files changed, 218 insertions(+), 112 deletions(-) diff --git a/bt-daemon/README.md b/bt-daemon/README.md index 7d76988..20bdb19 100644 --- a/bt-daemon/README.md +++ b/bt-daemon/README.md @@ -64,7 +64,8 @@ process-local settings overlay and never changes any of these files. `--plugin PATH` registers a synchronous ES module that transforms each sink-neutral span row after translation and immediately before delivery. Repeat the flag to compose plugins from left to right. Setup persists its ordered list; -run and import append invocation plugins after the persisted list. +run and import append invocation plugins after the persisted list. Each path is +canonicalized to an absolute path before it is validated or stored. ```bash bt trace setup codex --plugin ./redact.mjs --plugin ./tag-ci.mjs @@ -89,8 +90,8 @@ 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. A runtime failure disables the chain for that -session and sends the original translator output instead. +worker thread, not a session. If a plugin fails, that worker reports and skips +only that plugin on subsequent spans; the remaining plugins continue to run. ### Additional root metadata diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index c7f0972..3f5d5db 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -214,10 +214,6 @@ struct SessionActor { config: crate::wire::SessionConfig, } -struct PluginState { - disabled: bool, -} - impl SessionActor { async fn run(self, mut rx: mpsc::Receiver) { let mut translator = self.translators.create(&self.source, &self.session_id); @@ -255,14 +251,12 @@ impl SessionActor { session_id: self.session_id.clone(), config: Some(self.config.clone()), }; - let mut plugins = PluginState { disabled: false }; sink.configure(&self.config); self.refresh_permalink(sink.as_ref()); // Rebuild translator state before accepting the first new event. // Stable span ids make this both crash recovery and a complete copy // when an existing source session is sent to another destination. - self.replay_into(&mut translator, &mut sink, &ctx, &mut plugins) - .await; + self.replay_into(&mut translator, &mut sink, &ctx).await; while let Some(msg) = rx.recv().await { match msg { @@ -277,7 +271,6 @@ impl SessionActor { &mut translator, &mut sink, &ctx, - &mut plugins, translated, ("translate failed", "sink emit failed"), ) @@ -291,13 +284,11 @@ impl SessionActor { let _ = reply.send(()); } SessionMsg::Flush(reply) => { - self.drain_flush(&mut translator, &mut sink, &ctx, &mut plugins) - .await; + self.drain_flush(&mut translator, &mut sink, &ctx).await; let _ = reply.send(self.counters.queued.load(Ordering::Relaxed)); } SessionMsg::Shutdown(reply) => { - self.drain_flush(&mut translator, &mut sink, &ctx, &mut plugins) - .await; + self.drain_flush(&mut translator, &mut sink, &ctx).await; let _ = reply.send(()); break; } @@ -313,7 +304,6 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, - plugins: &mut PluginState, first: anyhow::Result>, errors: (&str, &str), ) { @@ -327,36 +317,35 @@ impl SessionActor { }; while let Some(ops) = next { if !ops.is_empty() { - let processed = if plugins.disabled { - ops.clone() - } else { - let plugin_paths = ctx - .config - .as_ref() - .map(|config| config.span_plugins.as_slice()) - .unwrap_or_default(); - let transformed: anyhow::Result> = ops - .iter() - .map(|op| { - crate::span_processor::process( - plugin_paths, - op, - &self.source, - &self.session_id, - ) - }) - .collect(); - match transformed { - Ok(processed) => processed, + let plugin_paths = ctx + .config + .as_ref() + .map(|config| config.span_plugins.as_slice()) + .unwrap_or_default(); + let mut processed = Vec::with_capacity(ops.len()); + for op in &ops { + match crate::span_processor::process( + plugin_paths, + op, + &self.source, + &self.session_id, + ) { + Ok(result) => { + for failure in result.failures { + self.set_error(format!( + "span plugin {} failed; disabled on this worker: {}", + failure.path.display(), + failure.message + )); + } + processed.push(result.op); + } Err(error) => { - plugins.disabled = true; - self.set_error(format!( - "span plugin failed; disabled for session: {error}" - )); - ops.clone() + self.set_error(format!("span plugin processor failed: {error}")); + processed.push(op.clone()); } } - }; + } match sink.emit(&processed).await { Ok(n) => { self.counters.spans_emitted.fetch_add(n, Ordering::Relaxed); @@ -383,7 +372,6 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, - plugins: &mut PluginState, ) { let Some(plan) = &self.replay else { return; @@ -420,7 +408,6 @@ impl SessionActor { translator, sink, ctx, - plugins, translated, ("journal replay failed", "sink replay emit failed"), ) @@ -433,14 +420,12 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, - plugins: &mut PluginState, ) { let translated = translator.flush(ctx); self.emit_translator_batches( translator, sink, ctx, - plugins, translated, ("translate flush failed", "sink emit (flush) failed"), ) diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index fc1599b..89f4b7d 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -847,7 +847,6 @@ struct ImportLive { sink: Box, ctx: SessionCtx, pending_ops: usize, - plugins_disabled: bool, } struct ImportProcessor { @@ -889,7 +888,6 @@ impl ImportProcessor { config: None, }, pending_ops: 0, - plugins_disabled: false, }, ); self.sessions.get_mut(&sid).unwrap() @@ -916,39 +914,41 @@ impl ImportProcessor { // for every native turn boundary. const FLUSH_OPS: usize = 500; for chunk in ops.chunks(FLUSH_OPS) { - let transformed = if live.plugins_disabled { - chunk.to_vec() - } else { - let plugins = live - .ctx - .config - .as_ref() - .map(|config| config.span_plugins.as_slice()) - .unwrap_or_default(); - let transformed: anyhow::Result> = chunk - .iter() - .map(|op| { - crate::span_processor::process( - plugins, - op, - &live.source, - &live.ctx.session_id, - ) - }) - .collect(); - match transformed { - Ok(transformed) => transformed, + let plugins = live + .ctx + .config + .as_ref() + .map(|config| config.span_plugins.as_slice()) + .unwrap_or_default(); + let mut transformed = Vec::with_capacity(chunk.len()); + for op in chunk { + match crate::span_processor::process( + plugins, + op, + &live.source, + &live.ctx.session_id, + ) { + Ok(result) => { + for failure in result.failures { + tracing::warn!( + session_id = %live.ctx.session_id, + plugin = %failure.path.display(), + error = %failure.message, + "span plugin failed during import; disabled on this worker" + ); + } + transformed.push(result.op); + } Err(error) => { - live.plugins_disabled = true; tracing::warn!( session_id = %live.ctx.session_id, %error, - "span plugin failed during import; disabled for session" + "span plugin processor failed during import" ); - chunk.to_vec() + transformed.push(op.clone()); } } - }; + } live.sink.emit(&transformed).await?; live.pending_ops += chunk.len(); if live.pending_ops >= FLUSH_OPS { @@ -1060,6 +1060,22 @@ mod tests { assert_eq!(args.session_idle_timeout_secs, 30); } + #[test] + fn span_plugin_paths_are_canonicalized_before_use() { + let dir = tempfile::Builder::new() + .prefix("span-plugin-path-") + .tempdir_in(".") + .unwrap(); + let plugin = dir.path().join("plugin.mjs"); + std::fs::write(&plugin, "export default span => span").unwrap(); + let relative = PathBuf::from(dir.path().file_name().unwrap()).join("plugin.mjs"); + + let resolved = resolve_span_plugin_paths(&[relative]).unwrap(); + + assert_eq!(resolved, [plugin.canonicalize().unwrap()]); + assert!(resolved[0].is_absolute()); + } + #[test] fn additional_metadata_overrides_a_route_only_with_a_json_object() { let mut route = SessionRoute { diff --git a/bt-daemon/src/span_processor.rs b/bt-daemon/src/span_processor.rs index 844ca52..dc7e684 100644 --- a/bt-daemon/src/span_processor.rs +++ b/bt-daemon/src/span_processor.rs @@ -9,7 +9,7 @@ use rquickjs::{Context, Function, Module, Persistent, Runtime}; use serde::Serialize; use sha2::{Digest, Sha256}; use std::cell::RefCell; -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; @@ -41,6 +41,7 @@ struct PluginContext<'a> { struct Engine { modules: HashMap, + failed_plugins: HashSet, env: BTreeMap, context: Context, started: Instant, @@ -71,6 +72,7 @@ impl Engine { let context = Context::full(&runtime)?; Ok(Self { modules: HashMap::new(), + failed_plugins: HashSet::new(), env: environment(), context, started, @@ -167,17 +169,31 @@ impl Engine { } } +#[derive(Debug)] +pub struct PluginFailure { + pub path: PathBuf, + pub message: String, +} + +pub struct ProcessResult { + pub op: SpanOp, + pub failures: Vec, +} + /// Apply an ordered plugin chain on the worker thread currently executing the -/// session actor. A plugin failure is reported to the caller, which can fail -/// open with the unmodified translator output. +/// session actor. A failing plugin is skipped on subsequent calls handled by +/// this worker, while the rest of the ordered chain continues to run. pub fn process( plugins: &[PathBuf], op: &SpanOp, source: &str, session_id: &str, -) -> anyhow::Result { +) -> anyhow::Result { if plugins.is_empty() { - return Ok(op.clone()); + return Ok(ProcessResult { + op: op.clone(), + failures: Vec::new(), + }); } let (operation, mut row) = match op { SpanOp::Insert(row) => (Operation::Insert, row.clone()), @@ -188,33 +204,59 @@ pub fn process( row.root_span_id.clone(), row.parent_span_ids.clone(), ); + let mut failures = Vec::new(); ENGINE.with_borrow_mut(|slot| -> anyhow::Result<()> { if slot.is_none() { *slot = Some(Engine::new()?); } let engine = slot.as_mut().expect("engine initialized"); for plugin in plugins { - row = engine.call(plugin, &row, operation, source, session_id)?; - if ( - row.span_id.as_str(), - row.root_span_id.as_str(), - &row.parent_span_ids, - ) != ( - original_ids.0.as_str(), - original_ids.1.as_str(), - &original_ids.2, - ) { - anyhow::bail!( - "plugin {} changed immutable span identity fields", - plugin.display() - ); + if engine.failed_plugins.contains(plugin) { + continue; } + let candidate = engine.call(plugin, &row, operation, source, session_id); + let candidate = match candidate { + Ok(candidate) + if ( + candidate.span_id.as_str(), + candidate.root_span_id.as_str(), + &candidate.parent_span_ids, + ) == ( + original_ids.0.as_str(), + original_ids.1.as_str(), + &original_ids.2, + ) => + { + candidate + } + Ok(_) => { + let message = "changed immutable span identity fields".to_owned(); + engine.failed_plugins.insert(plugin.clone()); + failures.push(PluginFailure { + path: plugin.clone(), + message, + }); + continue; + } + Err(error) => { + engine.failed_plugins.insert(plugin.clone()); + failures.push(PluginFailure { + path: plugin.clone(), + message: error.to_string(), + }); + continue; + } + }; + row = candidate; } Ok(()) })?; - Ok(match op { - SpanOp::Insert(_) => SpanOp::Insert(row), - SpanOp::Merge(_) => SpanOp::Merge(row), + Ok(ProcessResult { + op: match op { + SpanOp::Insert(_) => SpanOp::Insert(row), + SpanOp::Merge(_) => SpanOp::Merge(row), + }, + failures, }) } @@ -267,9 +309,9 @@ mod tests { "export default span => ({...span, metadata: {second: true}})", ) .unwrap(); - let processed = - process(&[first, second], &SpanOp::Insert(row()), "codex", "session").unwrap(); - let SpanOp::Insert(processed) = processed else { + let result = process(&[first, second], &SpanOp::Insert(row()), "codex", "session").unwrap(); + assert!(result.failures.is_empty()); + let SpanOp::Insert(processed) = result.op else { panic!("expected insert") }; assert_eq!( @@ -280,7 +322,7 @@ mod tests { } #[test] - fn rejects_identity_changes() { + fn rejects_identity_changes_without_dropping_the_span() { let dir = tempfile::tempdir().unwrap(); let plugin = dir.path().join("identity.mjs"); std::fs::write( @@ -288,8 +330,15 @@ mod tests { "export default span => ({...span, span_id: 'different'})", ) .unwrap(); - let error = process(&[plugin], &SpanOp::Insert(row()), "codex", "session").unwrap_err(); - assert!(error.to_string().contains("immutable span identity")); + let result = process(&[plugin], &SpanOp::Insert(row()), "codex", "session").unwrap(); + assert_eq!(result.failures.len(), 1); + assert!(result.failures[0] + .message + .contains("immutable span identity")); + let SpanOp::Insert(processed) = result.op else { + panic!("expected insert") + }; + assert_eq!(processed.span_id, "span"); } #[test] @@ -298,7 +347,8 @@ mod tests { let runaway = dir.path().join("runaway.mjs"); std::fs::write(&runaway, "export default span => { while (true) {} }").unwrap(); let started = Instant::now(); - assert!(process(&[runaway], &SpanOp::Insert(row()), "codex", "session").is_err()); + let result = process(&[runaway], &SpanOp::Insert(row()), "codex", "session").unwrap(); + assert_eq!(result.failures.len(), 1); assert!(started.elapsed() < Duration::from_secs(2)); let asynchronous = dir.path().join("async.mjs"); @@ -307,8 +357,47 @@ mod tests { "export default async span => ({...span, name: 'later'})", ) .unwrap(); - let error = - process(&[asynchronous], &SpanOp::Insert(row()), "codex", "session").unwrap_err(); - assert!(error.to_string().contains("must be synchronous")); + let result = process(&[asynchronous], &SpanOp::Insert(row()), "codex", "session").unwrap(); + assert_eq!(result.failures.len(), 1); + assert!(result.failures[0].message.contains("must be synchronous")); + } + + #[test] + fn skips_only_the_failed_plugin_and_continues_the_chain() { + let dir = tempfile::tempdir().unwrap(); + let first = dir.path().join("first.mjs"); + let broken = dir.path().join("broken.mjs"); + let last = dir.path().join("last.mjs"); + std::fs::write( + &first, + "export default span => ({...span, name: `first:${span.name}`})", + ) + .unwrap(); + std::fs::write( + &broken, + "export default () => { throw new Error('broken') }", + ) + .unwrap(); + std::fs::write( + &last, + "export default span => ({...span, name: `last:${span.name}`})", + ) + .unwrap(); + let plugins = [first, broken.clone(), last]; + + let first_result = process(&plugins, &SpanOp::Insert(row()), "codex", "session").unwrap(); + assert_eq!(first_result.failures.len(), 1); + assert_eq!(first_result.failures[0].path, broken); + let SpanOp::Insert(first_row) = first_result.op else { + panic!("expected insert") + }; + assert_eq!(first_row.name, "last:first:original"); + + let second_result = process(&plugins, &SpanOp::Insert(row()), "codex", "session").unwrap(); + assert!(second_result.failures.is_empty()); + let SpanOp::Insert(second_row) = second_result.op else { + panic!("expected insert") + }; + assert_eq!(second_row.name, "last:first:original"); } } diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 57b2d82..29e5e05 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -1051,13 +1051,23 @@ async fn span_plugins_transform_live_and_replayed_rows_with_daemon_environment() async fn a_failing_span_plugin_is_reported_and_fails_open() { let (data_dir, socket, handle, tmp) = start_daemon().await; let plugin = tmp.path().join("bad.mjs"); + let later_plugin = tmp.path().join("later.mjs"); std::fs::write( &plugin, "export default span => ({...span, span_id: 'corrupt'})", ) .unwrap(); + std::fs::write( + &later_plugin, + "export default span => ({...span, name: `after-failure:${span.name}`})", + ) + .unwrap(); let mut env = envelope("plugin-failure", "SessionStart", 1); - env.route.as_mut().unwrap().span_plugins.push(plugin); + env.route + .as_mut() + .unwrap() + .span_plugins + .extend([plugin, later_plugin]); forward_envelope(&env, &socket, &dummy_host(), false) .await .unwrap(); @@ -1072,12 +1082,17 @@ async fn a_failing_span_plugin_is_reported_and_fails_open() { .await .unwrap() .unwrap(); - assert!(status.sessions[0] - .last_error - .as_deref() - .is_some_and(|error| error.contains("span plugin failed"))); + assert!( + status.sessions[0] + .last_error + .as_deref() + .is_some_and(|error| error.contains("failed; disabled on this worker")), + "unexpected plugin status: {:?}", + status.sessions[0].last_error + ); let spans = std::fs::read_to_string(data_dir.join("spans/plugin-failure.ndjson")).unwrap(); assert!(!spans.contains("corrupt")); + assert!(spans.contains("after-failure:")); assert!( !spans.is_empty(), "the original rows should still be delivered" From 8db55096b467fec56fc258a8565ac487c87a38b7 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 23:02:45 +0800 Subject: [PATCH 4/8] Use direct serde bridge for span plugins --- bt-daemon/Cargo.lock | 11 +++++++++++ bt-daemon/Cargo.toml | 1 + bt-daemon/src/span_processor.rs | 29 ++++++++++------------------- 3 files changed, 22 insertions(+), 19 deletions(-) diff --git a/bt-daemon/Cargo.lock b/bt-daemon/Cargo.lock index 76ec42f..7b482aa 100644 --- a/bt-daemon/Cargo.lock +++ b/bt-daemon/Cargo.lock @@ -278,6 +278,7 @@ dependencies = [ "regex", "reqwest", "rquickjs", + "rquickjs-serde", "serde", "serde_json", "sha2", @@ -1510,6 +1511,16 @@ dependencies = [ "rquickjs-sys", ] +[[package]] +name = "rquickjs-serde" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04cf0aa631f8d0c5051db35f9f59899c34074d7b6be280c03fd4ce6165d0ed35" +dependencies = [ + "rquickjs", + "serde", +] + [[package]] name = "rquickjs-sys" version = "0.12.2" diff --git a/bt-daemon/Cargo.toml b/bt-daemon/Cargo.toml index c7e7698..0718e0f 100644 --- a/bt-daemon/Cargo.toml +++ b/bt-daemon/Cargo.toml @@ -25,6 +25,7 @@ chrono = "0.4" clap = { version = "4", features = ["derive", "env"] } regex = "1" rquickjs = "0.12.2" +rquickjs-serde = "0.6.1" serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.10" diff --git a/bt-daemon/src/span_processor.rs b/bt-daemon/src/span_processor.rs index dc7e684..d7afce2 100644 --- a/bt-daemon/src/span_processor.rs +++ b/bt-daemon/src/span_processor.rs @@ -18,7 +18,6 @@ use std::time::{Duration, Instant, SystemTime}; const MEMORY_LIMIT_BYTES: usize = 64 * 1024 * 1024; const STACK_LIMIT_BYTES: usize = 512 * 1024; const CALL_TIMEOUT: Duration = Duration::from_millis(50); -const MAX_RESULT_BYTES: usize = 8 * 1024 * 1024; thread_local! { static ENGINE: RefCell> = const { RefCell::new(None) }; @@ -131,32 +130,19 @@ impl Engine { session_id, env: &self.env, }; - let row_json = serde_json::to_vec(row)?; - let context_json = serde_json::to_vec(&context)?; self.arm_deadline(); - let result = self.context.with(|ctx| -> anyhow::Result> { + let result = self.context.with(|ctx| -> anyhow::Result { let function = function.restore(&ctx)?; - let row = ctx.json_parse(row_json)?; - let context = ctx.json_parse(context_json)?; + let row = rquickjs_serde::to_value(ctx.clone(), row)?; + let context = rquickjs_serde::to_value(ctx.clone(), &context)?; let result = function.call::<_, rquickjs::Value<'_>>((row, context))?; if result.as_promise().is_some() { anyhow::bail!("plugin returned a Promise; span plugins must be synchronous"); } - let json = ctx - .json_stringify(result)? - .ok_or_else(|| anyhow::anyhow!("plugin returned a non-JSON value"))?; - Ok(json.to_string()?.into_bytes()) + Ok(rquickjs_serde::from_value_strict(result)?) }); self.deadline_ms.store(0, Ordering::Relaxed); - let json = result?; - if json.len() > MAX_RESULT_BYTES { - anyhow::bail!( - "plugin returned {} bytes, exceeding the {} byte limit", - json.len(), - MAX_RESULT_BYTES - ); - } - Ok(serde_json::from_slice(&json)?) + result } fn arm_deadline(&self) { @@ -360,6 +346,11 @@ mod tests { let result = process(&[asynchronous], &SpanOp::Insert(row()), "codex", "session").unwrap(); assert_eq!(result.failures.len(), 1); assert!(result.failures[0].message.contains("must be synchronous")); + + let non_json = dir.path().join("non-json.mjs"); + std::fs::write(&non_json, "export default () => Symbol('not-json')").unwrap(); + let result = process(&[non_json], &SpanOp::Insert(row()), "codex", "session").unwrap(); + assert_eq!(result.failures.len(), 1); } #[test] From 24a1c59dad4d66fdeb1ee0e7f8460d3879d02f25 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 23:09:38 +0800 Subject: [PATCH 5/8] Restore named dispatch error arguments --- bt-daemon/src/dispatch.rs | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 3f5d5db..83aae51 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -272,7 +272,8 @@ impl SessionActor { &mut sink, &ctx, translated, - ("translate failed", "sink emit failed"), + "translate failed", + "sink emit failed", ) .await; self.counters.queued.fetch_sub(1, Ordering::Relaxed); @@ -305,9 +306,9 @@ impl SessionActor { sink: &mut Box, ctx: &SessionCtx, first: anyhow::Result>, - errors: (&str, &str), + translate_error: &str, + emit_error: &str, ) { - let (translate_error, emit_error) = errors; let mut next = match first { Ok(ops) => Some(ops), Err(e) => { @@ -409,7 +410,8 @@ impl SessionActor { sink, ctx, translated, - ("journal replay failed", "sink replay emit failed"), + "journal replay failed", + "sink replay emit failed", ) .await; } @@ -427,7 +429,8 @@ impl SessionActor { sink, ctx, translated, - ("translate flush failed", "sink emit (flush) failed"), + "translate flush failed", + "sink emit (flush) failed", ) .await; if let Err(e) = sink.flush().await { From 311d27f389b39c3d5f67a201abaf55cf7a86feb9 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 23:11:27 +0800 Subject: [PATCH 6/8] Normalize Windows plugin environment keys --- bt-daemon/src/span_processor.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/bt-daemon/src/span_processor.rs b/bt-daemon/src/span_processor.rs index d7afce2..2fc8aa9 100644 --- a/bt-daemon/src/span_processor.rs +++ b/bt-daemon/src/span_processor.rs @@ -263,7 +263,16 @@ pub fn validate(plugins: &[PathBuf]) -> anyhow::Result<()> { fn environment() -> BTreeMap { std::env::vars_os() - .filter_map(|(key, value)| Some((key.into_string().ok()?, value.into_string().ok()?))) + .filter_map(|(key, value)| { + let key = key.into_string().ok()?; + let value = value.into_string().ok()?; + // Windows environment variable names are case-insensitive, while + // JavaScript object properties are not. Use a stable casing there + // so portable plugins can read conventional names such as PATH. + #[cfg(windows)] + let key = key.to_ascii_uppercase(); + Some((key, value)) + }) .collect() } From 6429a64c25607be886546bd762da3cae24056dcb Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 23:24:46 +0800 Subject: [PATCH 7/8] Document JavaScript span plugin examples --- bt-daemon/README.md | 76 ++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 68 insertions(+), 8 deletions(-) diff --git a/bt-daemon/README.md b/bt-daemon/README.md index 20bdb19..0158992 100644 --- a/bt-daemon/README.md +++ b/bt-daemon/README.md @@ -67,31 +67,91 @@ the flag to compose plugins from left to right. Setup persists its ordered list; run and import append invocation plugins after the persisted list. Each path is canonicalized to an absolute path before it is validated or stored. -```bash -bt trace setup codex --plugin ./redact.mjs --plugin ./tag-ci.mjs -bt trace run --plugin ./local.mjs codex -- "summarize this change" -bt trace import codex SESSION_ID --plugin ./sanitize-history.mjs -``` - Each module must default-export a synchronous function. It receives a span and `{ operation, source, session_id, env }`, and must return a JSON-compatible span object. Span, root, and parent identities cannot be changed: ```js -export default function mapSpan(span, context) { +// redact.mjs +function redact(value) { + if (typeof value === "string") { + return value.replace(/sk-[A-Za-z0-9_-]+/g, "[REDACTED]"); + } + if (Array.isArray(value)) return value.map(redact); + if (value && typeof value === "object") { + return Object.fromEntries( + Object.entries(value).map(([key, child]) => [key, redact(child)]), + ); + } + return value; +} + +export default function redactSpan(span) { + const next = { ...span }; + for (const field of ["input", "output", "error"]) { + if (field in next) next[field] = redact(next[field]); + } + return next; +} +``` + +The context can drive a second transform without changing the first one: + +```js +// tag-ci.mjs +export default function tagCi(span, context) { + if (!context.env.CI) return span; + return { ...span, - metadata: { ...span.metadata, deployment: context.env.DEPLOYMENT }, + tags: [...new Set([...(span.tags ?? []), "ci"])], + metadata: { + ...(span.metadata ?? {}), + deployment: context.env.DEPLOYMENT_ENV ?? "unknown", + trace_source: context.source, + }, }; } ``` +Register both transforms persistently for ordinary Codex sessions. The +redactor runs first and its returned span becomes the tagger's input: + +```bash +bt trace setup codex --plugin ./redact.mjs --plugin ./tag-ci.mjs +``` + +`run` and `import` plugins apply only to that command and run after any plugins +saved by setup: + +```bash +# The invocation order is redact.mjs, tag-ci.mjs, then local.mjs. +bt trace run --plugin ./local.mjs codex -- "summarize this change" + +# Imported transcript spans pass through the setup plugins first, followed by +# sanitize-history.mjs. +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 plugin chain supplied by +the resumed session's current route. + +`context.operation` is `"insert"` or `"merge"`; `context.source` and +`context.session_id` identify the translated event stream; and `context.env` +contains the daemon process environment. Environment variable names are +uppercased on Windows so common lookups such as `context.env.PATH` remain +portable. + 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. If a plugin fails, that worker reports and skips only that plugin on subsequent spans; the remaining plugins continue to run. +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 +specific variables needed by the transform; never attach `context.env` itself. ### Additional root metadata From 41bd374abf793ddbb9a8106a7a57d258494198c6 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Sat, 22 Aug 2026 00:15:45 +0800 Subject: [PATCH 8/8] Isolate run and import span plugins --- bt-daemon/README.md | 21 +++---- bt-daemon/src/lib.rs | 102 ++++++++++++++++++++++++++------- bt-daemon/src/settings.rs | 7 --- bt-daemon/src/trace_runtime.rs | 18 ------ bt-daemon/src/wire/envelope.rs | 2 +- bt-daemon/tests/pipeline.rs | 6 ++ 6 files changed, 98 insertions(+), 58 deletions(-) diff --git a/bt-daemon/README.md b/bt-daemon/README.md index 0158992..d408288 100644 --- a/bt-daemon/README.md +++ b/bt-daemon/README.md @@ -63,8 +63,9 @@ process-local settings overlay and never changes any of these files. `--plugin PATH` registers a synchronous ES module that transforms each sink-neutral span row after translation and immediately before delivery. Repeat -the flag to compose plugins from left to right. Setup persists its ordered list; -run and import append invocation plugins after the persisted list. Each path is +the flag to compose plugins from left to right. `enable` persists its ordered list +for ordinary agent sessions. Managed runs and imports are isolated from that +list and use only the `--plugin` flags passed to their command. Each path is canonicalized to an absolute path before it is validated or stored. Each module must default-export a synchronous function. It receives a span and @@ -118,24 +119,24 @@ Register both transforms persistently for ordinary Codex sessions. The redactor runs first and its returned span becomes the tagger's input: ```bash -bt trace setup codex --plugin ./redact.mjs --plugin ./tag-ci.mjs +bt trace enable codex --plugin ./redact.mjs --plugin ./tag-ci.mjs ``` -`run` and `import` plugins apply only to that command and run after any plugins -saved by setup: +`run` and `import` plugins apply only to that command. They replace, rather than +merge with, plugins saved by `enable`: ```bash -# The invocation order is redact.mjs, tag-ci.mjs, then local.mjs. +# Only local.mjs runs; redact.mjs and tag-ci.mjs remain global enable behavior. bt trace run --plugin ./local.mjs codex -- "summarize this change" -# Imported transcript spans pass through the setup plugins first, followed by -# sanitize-history.mjs. +# Only sanitize-history.mjs transforms spans produced by this import. 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 plugin chain supplied by -the resumed session's current route. +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. `context.operation` is `"insert"` or `"merge"`; `context.source` and `context.session_id` identify the translated event stream; and `context.env` diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 89f4b7d..c36a31c 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -164,8 +164,8 @@ pub struct ImportArgs { /// JSON object merged into every imported root span's metadata. #[arg(long, env = "BRAINTRUST_ADDITIONAL_METADATA")] pub additional_metadata: Option, - /// JavaScript span transform for this import. Repeat to compose transforms - /// after plugins from the agent's persistent setup. + /// JavaScript span transform for this import. Repeat to compose an isolated + /// transform chain; persistent setup plugins are not included. #[arg(long, value_name = "PATH")] pub plugin: Vec, } @@ -187,8 +187,8 @@ pub struct RunArgs { /// JSON object merged into root-span metadata for this invocation. #[arg(long, env = "BRAINTRUST_ADDITIONAL_METADATA")] pub additional_metadata: Option, - /// JavaScript span transform for this invocation. Repeat to compose - /// transforms after plugins from persistent setup. + /// JavaScript span transform for this invocation. Repeat to compose an + /// isolated transform chain; persistent setup plugins are not included. #[arg(long, value_name = "PATH")] pub plugin: Vec, /// Arguments forwarded verbatim to the coding agent. @@ -474,9 +474,28 @@ pub async fn run_import( mut config: Option, ) -> anyhow::Result<()> { validate_import_selection(&args)?; - let command_plugins = resolve_span_plugin_paths(&args.plugin)?; - if !command_plugins.is_empty() { - let config = config.get_or_insert_with(|| SessionConfig { + apply_import_span_plugins(&mut config, &args.plugin)?; + let destination = args + .parent + .map(|components| wire::TraceDestination::ParentSpan { components }) + .or(args.destination); + apply_import_destination(&mut config, destination)?; + let files = transcript_import::resolve_transcripts(&args.session_ids, args.all, args.source)?; + if args.attach { + return import_transcript(&files[0], args.source, opts, config, true).await; + } + import_transcripts(&files, args.source, opts, config).await +} + +fn apply_import_span_plugins( + config: &mut Option, + plugins: &[PathBuf], +) -> anyhow::Result<()> { + let plugins = resolve_span_plugin_paths(plugins)?; + if let Some(config) = config.as_mut() { + config.span_plugins = plugins; + } else if !plugins.is_empty() { + *config = Some(SessionConfig { auth: wire::BackendAuth { token: String::new(), api_url: None, @@ -487,20 +506,10 @@ pub async fn run_import( destination: None, flush_mode: wire::FlushMode::FireAndForget, additional_metadata: None, - span_plugins: Vec::new(), + span_plugins: plugins, }); - config.span_plugins.extend(command_plugins); } - let destination = args - .parent - .map(|components| wire::TraceDestination::ParentSpan { components }) - .or(args.destination); - apply_import_destination(&mut config, destination)?; - let files = transcript_import::resolve_transcripts(&args.session_ids, args.all, args.source)?; - if args.attach { - return import_transcript(&files[0], args.source, opts, config, true).await; - } - import_transcripts(&files, args.source, opts, config).await + Ok(()) } fn validate_import_selection(args: &ImportArgs) -> anyhow::Result<()> { @@ -521,9 +530,7 @@ pub async fn run_traced( hook_command: RunHookCommand, mut route: SessionRoute, ) -> anyhow::Result { - route - .span_plugins - .extend(resolve_span_plugin_paths(&args.plugin)?); + apply_run_span_plugins(&mut route, &args.plugin)?; if route.destination.is_none() { anyhow::bail!( "managed run requires a trace destination; select a project, object destination, or parent span" @@ -591,6 +598,11 @@ pub async fn run_traced( status } +fn apply_run_span_plugins(route: &mut SessionRoute, plugins: &[PathBuf]) -> anyhow::Result<()> { + route.span_plugins = resolve_span_plugin_paths(plugins)?; + Ok(()) +} + pub(crate) fn resolve_span_plugin_paths(paths: &[PathBuf]) -> anyhow::Result> { let paths: Vec<_> = paths .iter() @@ -1096,6 +1108,52 @@ mod tests { .contains("invalid --additional-metadata JSON")); } + #[test] + fn managed_run_plugins_replace_inherited_plugins() { + let temp = tempfile::tempdir().unwrap(); + let plugin = temp.path().join("run.mjs"); + std::fs::write(&plugin, "export default span => span").unwrap(); + let mut route = SessionRoute { + span_plugins: vec![PathBuf::from("persisted.mjs")], + ..SessionRoute::default() + }; + + apply_run_span_plugins(&mut route, &[]).unwrap(); + assert!(route.span_plugins.is_empty()); + + apply_run_span_plugins(&mut route, std::slice::from_ref(&plugin)).unwrap(); + assert_eq!(route.span_plugins, [plugin.canonicalize().unwrap()]); + } + + #[test] + fn import_plugins_replace_inherited_plugins() { + let temp = tempfile::tempdir().unwrap(); + let plugin = temp.path().join("import.mjs"); + std::fs::write(&plugin, "export default span => span").unwrap(); + let mut config = Some(SessionConfig { + auth: wire::BackendAuth { + token: String::new(), + api_url: None, + app_url: None, + org_name: None, + org_id: None, + }, + destination: None, + flush_mode: wire::FlushMode::FireAndForget, + additional_metadata: None, + span_plugins: vec![PathBuf::from("persisted.mjs")], + }); + + apply_import_span_plugins(&mut config, &[]).unwrap(); + assert!(config.as_ref().unwrap().span_plugins.is_empty()); + + apply_import_span_plugins(&mut config, std::slice::from_ref(&plugin)).unwrap(); + assert_eq!( + config.unwrap().span_plugins, + [plugin.canonicalize().unwrap()] + ); + } + #[test] fn import_args_accept_multiple_sessions_or_all() { let explicit = ImportCli::try_parse_from([ diff --git a/bt-daemon/src/settings.rs b/bt-daemon/src/settings.rs index 23367e1..dda7a9e 100644 --- a/bt-daemon/src/settings.rs +++ b/bt-daemon/src/settings.rs @@ -84,13 +84,6 @@ impl AgentSettings { pub(crate) fn tracing_enabled(&self) -> bool { self.trace_to_braintrust.unwrap_or(false) } - - pub(crate) fn configured_span_plugins(source: &str) -> Vec { - Self::load_from(&paths::agent_settings_path(source, None)) - .route - .map(|route| route.span_plugins) - .unwrap_or_default() - } } #[cfg(test)] diff --git a/bt-daemon/src/trace_runtime.rs b/bt-daemon/src/trace_runtime.rs index dd495c7..499dcfc 100644 --- a/bt-daemon/src/trace_runtime.rs +++ b/bt-daemon/src/trace_runtime.rs @@ -186,12 +186,6 @@ fn print_output(output: TraceCommandOutput, format: OutputFormat) -> anyhow::Res Ok(()) } -fn configured_plugins(source: &str) -> anyhow::Result> { - crate::resolve_span_plugin_paths(&crate::settings::AgentSettings::configured_span_plugins( - source, - )) -} - /// Execute the complete mounted trace command. pub async fn run_trace(args: TraceArgs, host: TraceHostContext) -> anyhow::Result<()> { match args.command { @@ -258,11 +252,6 @@ pub async fn run_trace(args: TraceArgs, host: TraceHostContext) -> anyhow::Resul }) .await?; apply_additional_metadata(&mut route, import_args.additional_metadata.as_deref())?; - let source = match import_args.source { - crate::ImportSource::Codex => "codex", - crate::ImportSource::Claude => "claude", - }; - route.span_plugins = configured_plugins(source)?; let config = session_config(&host, &route).await?; run_import(import_args, serve_options(&host), Some(config)).await } @@ -276,13 +265,6 @@ pub async fn run_trace(args: TraceArgs, host: TraceHostContext) -> anyhow::Resul ) .await?; apply_additional_metadata(&mut route, run_args.additional_metadata.as_deref())?; - let source = match run_args.source { - crate::RunSource::Codex => "codex", - crate::RunSource::Claude => "claude", - crate::RunSource::OpenCode => "opencode", - crate::RunSource::Pi => "pi", - }; - route.span_plugins = configured_plugins(source)?; let hook_command = child_command(&host.command, "hook"); let status = run_traced(run_args, hook_command, route).await?; if status.success() { diff --git a/bt-daemon/src/wire/envelope.rs b/bt-daemon/src/wire/envelope.rs index a96947a..384ad3a 100644 --- a/bt-daemon/src/wire/envelope.rs +++ b/bt-daemon/src/wire/envelope.rs @@ -65,7 +65,7 @@ pub struct SessionRoute { #[serde(default, skip_serializing_if = "Option::is_none")] pub additional_metadata: Option, /// Ordered JavaScript span transforms. Paths are resolved by explicit - /// setup, run, and import commands before entering persistent settings. + /// setup, run, and import commands before entering a session route. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub span_plugins: Vec, } diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 29e5e05..ea5c604 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -1042,6 +1042,12 @@ async fn span_plugins_transform_live_and_replayed_rows_with_daemon_environment() names.iter().any(|name| name.contains(":current:")), "the new plugin chain should process replayed and live rows: {names:?}" ); + assert!( + !names + .iter() + .any(|name| name.contains(&format!(":current:{path_prefix}"))), + "recovery should replace the journal's old plugin chain with the resumed route: {names:?}" + ); shutdown(&socket).await; second.await.unwrap();