From ecbd3ee0283e595b8751cf8afd58d666d876fa7d Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 22:13:19 +0800 Subject: [PATCH 1/7] 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 | 80 ++++- 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, 954 insertions(+), 15 deletions(-) create mode 100644 bt-daemon/src/span_processor.rs diff --git a/bt-daemon/Cargo.lock b/bt-daemon/Cargo.lock index 225091e..14ed0c1 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", @@ -599,6 +606,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" @@ -757,6 +770,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" @@ -1417,6 +1435,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" @@ -1473,6 +1500,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 65784c7..ad9306c 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 57acb25..a71bb08 100644 --- a/bt-daemon/docs/protocol.md +++ b/bt-daemon/docs/protocol.md @@ -194,6 +194,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", @@ -205,7 +206,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"] } } ``` @@ -226,6 +228,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. @@ -269,8 +273,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 @@ -337,8 +342,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 fdfad18..7aa7318 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -12,6 +12,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}; @@ -295,6 +296,11 @@ impl BatchMode { } } +struct PluginState { + env: BTreeMap, + disabled: bool, +} + impl SessionActor { async fn run(self, mut rx: mpsc::Receiver) { let mut translator = match self @@ -345,18 +351,26 @@ 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, reply) => { let correlation_barrier = is_tool_lifecycle_event(&env.event); let mut reply = Some(reply); + 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()); @@ -371,6 +385,7 @@ impl SessionActor { &mut translator, &mut sink, &ctx, + &mut plugins, translated, BatchMode::Live, ) @@ -398,7 +413,7 @@ impl SessionActor { let _ = reply.send(()); } SessionMsg::Flush(reply) => { - self.checkpoint_and_flush(&mut translator, &mut sink, &ctx) + self.checkpoint_and_flush(&mut translator, &mut sink, &ctx, &mut plugins) .await; let _ = reply.send(self.counters.queued.load(Ordering::Relaxed)); } @@ -407,6 +422,7 @@ impl SessionActor { &mut translator, &mut sink, &ctx, + &mut plugins, BatchMode::TerminalFinalize, ) .await; @@ -417,6 +433,7 @@ impl SessionActor { &mut translator, &mut sink, &ctx, + &mut plugins, BatchMode::ShutdownFinalize, ) .await; @@ -435,6 +452,7 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, + plugins: &mut PluginState, first: anyhow::Result>, mode: BatchMode, ) -> bool { @@ -457,7 +475,38 @@ impl SessionActor { &ops, ); } - 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); } @@ -484,6 +533,7 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, + plugins: &mut PluginState, ) { let Some(plan) = &self.replay else { return; @@ -510,7 +560,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; } @@ -524,7 +574,14 @@ impl SessionActor { env.source = self.source.clone(); let translated = translator.handle(&env, ctx); let _ = self - .emit_translator_batches(translator, sink, ctx, translated, BatchMode::Replay) + .emit_translator_batches( + translator, + sink, + ctx, + plugins, + translated, + BatchMode::Replay, + ) .await; } } @@ -534,10 +591,18 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, + plugins: &mut PluginState, ) { let translated = translator.checkpoint(ctx); let correlation_changed = self - .emit_translator_batches(translator, sink, ctx, translated, BatchMode::Checkpoint) + .emit_translator_batches( + translator, + sink, + ctx, + plugins, + translated, + BatchMode::Checkpoint, + ) .await; self.persist_correlation_if_changed(correlation_changed) .await; @@ -549,11 +614,12 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, + plugins: &mut PluginState, mode: BatchMode, ) { let translated = translator.finalize(ctx); let correlation_changed = self - .emit_translator_batches(translator, sink, ctx, translated, mode) + .emit_translator_batches(translator, sink, ctx, plugins, translated, mode) .await; self.persist_correlation_if_changed(correlation_changed) .await; diff --git a/bt-daemon/src/journal.rs b/bt-daemon/src/journal.rs index 5869e65..aa36d13 100644 --- a/bt-daemon/src/journal.rs +++ b/bt-daemon/src/journal.rs @@ -352,6 +352,7 @@ pub fn envelope_from_redacted(r: RedactedEnvelope) -> Envelope { managed_run_id: r.managed_run_id, capture: r.capture, 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 64a50c3..d5c6b8d 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -23,6 +23,7 @@ mod server; mod settings; mod setup; mod sink; +mod span_processor; mod trace_command; mod trace_runtime; mod transcript_import; @@ -213,6 +214,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)] @@ -257,6 +262,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, @@ -341,6 +350,11 @@ pub async fn run_hook( .filter(|value| !value.is_empty()), capture: None, payload, + plugin_env: if route.span_plugins.is_empty() { + Default::default() + } else { + span_processor::environment() + }, route: Some(route), config: None, }; @@ -553,6 +567,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 = import_parent_components(&args)? .map(|components| wire::TraceDestination::ParentSpan { components }) .or_else(|| args.destination.clone()); @@ -657,8 +688,11 @@ fn import_parent_components(args: &ImportArgs) -> anyhow::Result 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" @@ -786,6 +820,19 @@ impl ManagedRunRuntime { } } +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, @@ -1026,6 +1073,7 @@ pub async fn import_transcripts( } struct ImportLive { + source: String, translator: Box, sink: Box, ctx: SessionCtx, @@ -1033,6 +1081,8 @@ struct ImportLive { span_ids: std::collections::HashSet, root_span_id: Option, destination: Option, + plugin_env: std::collections::BTreeMap, + plugins_disabled: bool, } struct ImportProcessor { @@ -1066,6 +1116,7 @@ impl ImportProcessor { self.sessions.insert( sid.clone(), ImportLive { + source: env.source.clone(), translator, sink, ctx: SessionCtx { @@ -1082,6 +1133,12 @@ impl ImportProcessor { .config .as_ref() .and_then(|config| config.destination.clone()), + 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() @@ -1091,6 +1148,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?; } @@ -1117,7 +1177,41 @@ impl ImportProcessor { live.root_span_id = Some(row.root_span_id.clone()); } } - 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?; @@ -1404,6 +1498,7 @@ mod tests { parent_project: None, attach: true, additional_metadata: None, + plugin: Vec::new(), }; assert!(validate_import_selection(&args) .unwrap_err() @@ -1482,6 +1577,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 4aa98f1..f129868 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 8258eaf..50d1466 100644 --- a/bt-daemon/src/setup.rs +++ b/bt-daemon/src/setup.rs @@ -475,6 +475,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)?); for key in [ @@ -1077,4 +1084,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 f0ae928..fca9500 100644 --- a/bt-daemon/src/trace_command.rs +++ b/bt-daemon/src/trace_command.rs @@ -98,6 +98,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. @@ -150,6 +154,7 @@ mod tests { TraceCommand::Setup(SetupArgs { agent: SetupAgent::Claude, additional_metadata: Some(ref value), + .. }) if value == r#"{"setup":true}"# )); @@ -260,4 +265,50 @@ mod tests { assert!(Cli::try_parse_from(["bt", "setup", "antigravity", "--disable"]).is_err()); } + + #[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 1438ec2..1e39ee6 100644 --- a/bt-daemon/src/trace_runtime.rs +++ b/bt-daemon/src/trace_runtime.rs @@ -178,6 +178,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(), }) } @@ -305,6 +306,12 @@ async fn doctor_output(host: &TraceHostContext, args: DoctorArgs) -> DoctorComma } } +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 { @@ -319,6 +326,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) => { @@ -373,6 +383,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?; let summaries = run_import(import_args, serve_options(&host), Some(config)).await?; print_output(TraceCommandOutput::import(summaries), host.output_format) @@ -388,6 +403,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() { @@ -549,6 +571,7 @@ mod tests { TraceCommand::Setup(SetupArgs { agent: SetupAgent::OpenCode, additional_metadata: None, + plugin: Vec::new(), }), true, ), @@ -556,6 +579,7 @@ mod tests { TraceCommand::Run(RunArgs { source: RunSource::Codex, additional_metadata: None, + plugin: Vec::new(), agent_args: Vec::new(), }), false, @@ -635,6 +659,7 @@ mod tests { parent_project: 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 b4f539e..5e6a2cf 100644 --- a/bt-daemon/src/transcript_import/mod.rs +++ b/bt-daemon/src/transcript_import/mod.rs @@ -306,6 +306,7 @@ fn envelope( managed_run_id: None, capture: 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 7969dc1..8d837dc 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 operating-system process observed while capturing an event. /// @@ -56,6 +58,10 @@ pub struct Envelope { pub capture: 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. @@ -138,6 +144,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 { @@ -147,12 +157,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. @@ -166,6 +187,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. @@ -326,6 +349,7 @@ mod tests { }], }), 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 { source: AuthSource::SavedProfile, @@ -349,6 +373,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 b1d054d..6bf3b85 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 eb2dfa3..0104a35 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, capture: None, @@ -270,6 +271,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, capture: None, @@ -370,6 +372,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, capture: None, @@ -460,6 +463,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, capture: None, @@ -580,6 +584,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, capture: None, @@ -717,6 +722,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, capture: None, diff --git a/bt-daemon/tests/codex_translator.rs b/bt-daemon/tests/codex_translator.rs index 896cd88..368e4de 100644 --- a/bt-daemon/tests/codex_translator.rs +++ b/bt-daemon/tests/codex_translator.rs @@ -69,6 +69,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, capture: None, @@ -507,6 +508,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 210e718..973475b 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, capture: None, diff --git a/bt-daemon/tests/pi_translator.rs b/bt-daemon/tests/pi_translator.rs index fa4d5a6..1d92ff8 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, capture: None, diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 7d358f8..0e820a4 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -137,6 +137,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, @@ -1151,6 +1152,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; @@ -1534,6 +1649,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 c286a58..0555b61 100644 --- a/bt-daemon/tests/replay.rs +++ b/bt-daemon/tests/replay.rs @@ -107,6 +107,7 @@ async fn attached_import_summary_reports_the_effective_parent_root() { }), flush_mode: FlushMode::FireAndForget, additional_metadata: None, + span_plugins: Vec::new(), }; let summaries = import_transcript( @@ -123,6 +124,56 @@ async fn attached_import_summary_reports_the_effective_parent_root() { assert_eq!(summaries[0].root_span_id.as_deref(), Some(parent_root)); } +#[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 8a180a8..d22bc26 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 { @@ -77,6 +78,7 @@ export interface DaemonEnvelope { ts_ms: number managed_run_id?: string payload: unknown + plugin_env?: Record route?: DaemonSessionRoute } @@ -167,6 +169,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 9036c67..e620db2 100644 --- a/src/runtime/js-daemon-client/tests/client.test.ts +++ b/src/runtime/js-daemon-client/tests/client.test.ts @@ -127,6 +127,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, @@ -143,6 +144,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 082fdb8762628485ba59512bc4845780659871aa Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 22:38:51 +0800 Subject: [PATCH 2/7] 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 a71bb08..f3323a4 100644 --- a/bt-daemon/docs/protocol.md +++ b/bt-daemon/docs/protocol.md @@ -194,7 +194,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", @@ -228,8 +227,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. @@ -273,9 +270,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 7aa7318..83f777a 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -12,7 +12,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}; @@ -297,7 +296,6 @@ impl BatchMode { } struct PluginState { - env: BTreeMap, disabled: bool, } @@ -351,10 +349,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. @@ -368,9 +363,6 @@ impl SessionActor { SessionMsg::Event(env, reply) => { let correlation_barrier = is_tool_lifecycle_event(&env.event); let mut reply = Some(reply); - 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()); @@ -491,7 +483,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 aa36d13..5869e65 100644 --- a/bt-daemon/src/journal.rs +++ b/bt-daemon/src/journal.rs @@ -352,7 +352,6 @@ pub fn envelope_from_redacted(r: RedactedEnvelope) -> Envelope { managed_run_id: r.managed_run_id, capture: r.capture, 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 d5c6b8d..8cd992a 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -350,11 +350,6 @@ pub async fn run_hook( .filter(|value| !value.is_empty()), capture: None, payload, - plugin_env: if route.span_plugins.is_empty() { - Default::default() - } else { - span_processor::environment() - }, route: Some(route), config: None, }; @@ -1081,7 +1076,6 @@ struct ImportLive { span_ids: std::collections::HashSet, root_span_id: Option, destination: Option, - plugin_env: std::collections::BTreeMap, plugins_disabled: bool, } @@ -1133,11 +1127,6 @@ impl ImportProcessor { .config .as_ref() .and_then(|config| config.destination.clone()), - plugin_env: if env.plugin_env.is_empty() { - crate::span_processor::environment() - } else { - env.plugin_env.clone() - }, plugins_disabled: false, }, ); @@ -1148,9 +1137,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?; } @@ -1194,7 +1180,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 5e6a2cf..b4f539e 100644 --- a/bt-daemon/src/transcript_import/mod.rs +++ b/bt-daemon/src/transcript_import/mod.rs @@ -306,7 +306,6 @@ fn envelope( managed_run_id: None, capture: 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 8d837dc..d737e39 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 operating-system process observed while capturing an event. @@ -58,10 +57,6 @@ pub struct Envelope { pub capture: 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. @@ -349,7 +344,6 @@ mod tests { }], }), 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 { source: AuthSource::SavedProfile, diff --git a/bt-daemon/tests/claude_translator.rs b/bt-daemon/tests/claude_translator.rs index 0104a35..eb2dfa3 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, capture: None, @@ -271,7 +270,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, capture: None, @@ -372,7 +370,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, capture: None, @@ -463,7 +460,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, capture: None, @@ -584,7 +580,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, capture: None, @@ -722,7 +717,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, capture: None, diff --git a/bt-daemon/tests/codex_translator.rs b/bt-daemon/tests/codex_translator.rs index 368e4de..a17b140 100644 --- a/bt-daemon/tests/codex_translator.rs +++ b/bt-daemon/tests/codex_translator.rs @@ -69,7 +69,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, capture: None, diff --git a/bt-daemon/tests/opencode_translator.rs b/bt-daemon/tests/opencode_translator.rs index 973475b..210e718 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, capture: None, diff --git a/bt-daemon/tests/pi_translator.rs b/bt-daemon/tests/pi_translator.rs index 1d92ff8..fa4d5a6 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, capture: None, diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 0e820a4..9fd5599 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -137,7 +137,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, @@ -1153,14 +1152,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( @@ -1176,10 +1175,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(); @@ -1187,10 +1182,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 @@ -1216,7 +1207,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 d22bc26..788e9e7 100644 --- a/src/runtime/js-daemon-client/src/index.ts +++ b/src/runtime/js-daemon-client/src/index.ts @@ -78,7 +78,6 @@ export interface DaemonEnvelope { ts_ms: number managed_run_id?: string payload: unknown - plugin_env?: Record route?: DaemonSessionRoute } @@ -169,17 +168,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 e620db2..e776cfa 100644 --- a/src/runtime/js-daemon-client/tests/client.test.ts +++ b/src/runtime/js-daemon-client/tests/client.test.ts @@ -144,10 +144,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 85e682c1692873da1587ea5b90f76a6673c01366 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 22:55:06 +0800 Subject: [PATCH 3/7] Isolate span plugin failures per worker --- bt-daemon/README.md | 7 +- bt-daemon/src/dispatch.rs | 74 ++++++--------- bt-daemon/src/lib.rs | 72 +++++++++------ bt-daemon/src/span_processor.rs | 153 +++++++++++++++++++++++++------- bt-daemon/tests/pipeline.rs | 25 ++++-- 5 files changed, 218 insertions(+), 113 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 83f777a..4eef38f 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -295,10 +295,6 @@ impl BatchMode { } } -struct PluginState { - disabled: bool, -} - impl SessionActor { async fn run(self, mut rx: mpsc::Receiver) { let mut translator = match self @@ -349,14 +345,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 { @@ -377,7 +371,6 @@ impl SessionActor { &mut translator, &mut sink, &ctx, - &mut plugins, translated, BatchMode::Live, ) @@ -405,7 +398,7 @@ impl SessionActor { let _ = reply.send(()); } SessionMsg::Flush(reply) => { - self.checkpoint_and_flush(&mut translator, &mut sink, &ctx, &mut plugins) + self.checkpoint_and_flush(&mut translator, &mut sink, &ctx) .await; let _ = reply.send(self.counters.queued.load(Ordering::Relaxed)); } @@ -414,7 +407,6 @@ impl SessionActor { &mut translator, &mut sink, &ctx, - &mut plugins, BatchMode::TerminalFinalize, ) .await; @@ -425,7 +417,6 @@ impl SessionActor { &mut translator, &mut sink, &ctx, - &mut plugins, BatchMode::ShutdownFinalize, ) .await; @@ -444,7 +435,6 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, - plugins: &mut PluginState, first: anyhow::Result>, mode: BatchMode, ) -> bool { @@ -467,36 +457,35 @@ impl SessionActor { &ops, ); } - 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); @@ -524,7 +513,6 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, - plugins: &mut PluginState, ) { let Some(plan) = &self.replay else { return; @@ -569,7 +557,6 @@ impl SessionActor { translator, sink, ctx, - plugins, translated, BatchMode::Replay, ) @@ -582,7 +569,6 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, - plugins: &mut PluginState, ) { let translated = translator.checkpoint(ctx); let correlation_changed = self @@ -590,7 +576,6 @@ impl SessionActor { translator, sink, ctx, - plugins, translated, BatchMode::Checkpoint, ) @@ -605,12 +590,11 @@ impl SessionActor { translator: &mut Box, sink: &mut Box, ctx: &SessionCtx, - plugins: &mut PluginState, mode: BatchMode, ) { let translated = translator.finalize(ctx); let correlation_changed = self - .emit_translator_batches(translator, sink, ctx, plugins, translated, mode) + .emit_translator_batches(translator, sink, ctx, translated, mode) .await; self.persist_correlation_if_changed(correlation_changed) .await; diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 8cd992a..2939749 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -1076,7 +1076,6 @@ struct ImportLive { span_ids: std::collections::HashSet, root_span_id: Option, destination: Option, - plugins_disabled: bool, } struct ImportProcessor { @@ -1127,7 +1126,6 @@ impl ImportProcessor { .config .as_ref() .and_then(|config| config.destination.clone()), - plugins_disabled: false, }, ); self.sessions.get_mut(&sid).unwrap() @@ -1163,39 +1161,41 @@ impl ImportProcessor { live.root_span_id = Some(row.root_span_id.clone()); } } - 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 { @@ -1362,6 +1362,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 9fd5599..91843d2 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -1222,13 +1222,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(); @@ -1243,12 +1253,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 2240d62419183f592e1090a7c705da8ae3ac5d7f Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 23:02:45 +0800 Subject: [PATCH 4/7] 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 14ed0c1..48d32b1 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", @@ -1520,6 +1521,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 ad9306c..47b4872 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 20e0099c7faa92203a487f16c317269c5cd20472 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 23:11:27 +0800 Subject: [PATCH 5/7] 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 b7c68e100a7171ed70e33757691f26b5ec55a9d1 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 21 Aug 2026 23:24:46 +0800 Subject: [PATCH 6/7] 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 3a20d759ee0f742b090c278ca7161716704904ce Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Sat, 22 Aug 2026 00:15:45 +0800 Subject: [PATCH 7/7] Isolate run and import span plugins --- bt-daemon/README.md | 21 +++---- bt-daemon/src/dispatch.rs | 16 +----- bt-daemon/src/lib.rs | 101 ++++++++++++++++++++++++++------- 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 ++ 7 files changed, 100 insertions(+), 71 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/dispatch.rs b/bt-daemon/src/dispatch.rs index 4eef38f..13d52dc 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -553,13 +553,7 @@ impl SessionActor { env.source = self.source.clone(); let translated = translator.handle(&env, ctx); let _ = self - .emit_translator_batches( - translator, - sink, - ctx, - translated, - BatchMode::Replay, - ) + .emit_translator_batches(translator, sink, ctx, translated, BatchMode::Replay) .await; } } @@ -572,13 +566,7 @@ impl SessionActor { ) { let translated = translator.checkpoint(ctx); let correlation_changed = self - .emit_translator_batches( - translator, - sink, - ctx, - translated, - BatchMode::Checkpoint, - ) + .emit_translator_batches(translator, sink, ctx, translated, BatchMode::Checkpoint) .await; self.persist_correlation_if_changed(correlation_changed) .await; diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 2939749..d7f400b 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -214,8 +214,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, } @@ -262,8 +262,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. @@ -562,9 +562,27 @@ 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 = import_parent_components(&args)? + .map(|components| wire::TraceDestination::ParentSpan { components }) + .or_else(|| args.destination.clone()); + 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, @@ -575,19 +593,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 = import_parent_components(&args)? - .map(|components| wire::TraceDestination::ParentSpan { components }) - .or_else(|| args.destination.clone()); - 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<()> { @@ -685,9 +694,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" @@ -815,6 +822,11 @@ impl ManagedRunRuntime { } } +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() @@ -1398,6 +1410,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([ @@ -1596,6 +1654,7 @@ mod tests { RunArgs { source: RunSource::Codex, additional_metadata: None, + plugin: Vec::new(), agent_args: vec![OsString::from("--dangerously-bypass-hook-trust")], }, test_run_hook_command(), diff --git a/bt-daemon/src/settings.rs b/bt-daemon/src/settings.rs index f129868..4aa98f1 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 1e39ee6..4eda398 100644 --- a/bt-daemon/src/trace_runtime.rs +++ b/bt-daemon/src/trace_runtime.rs @@ -306,12 +306,6 @@ async fn doctor_output(host: &TraceHostContext, args: DoctorArgs) -> DoctorComma } } -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 { @@ -383,11 +377,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?; let summaries = run_import(import_args, serve_options(&host), Some(config)).await?; print_output(TraceCommandOutput::import(summaries), host.output_format) @@ -403,13 +392,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 d737e39..8fb3cc1 100644 --- a/bt-daemon/src/wire/envelope.rs +++ b/bt-daemon/src/wire/envelope.rs @@ -140,7 +140,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 91843d2..367aba0 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -1213,6 +1213,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();