diff --git a/bt-daemon/Cargo.lock b/bt-daemon/Cargo.lock index 18d1bf23..46a11702 100644 --- a/bt-daemon/Cargo.lock +++ b/bt-daemon/Cargo.lock @@ -239,7 +239,7 @@ dependencies = [ [[package]] name = "braintrust-sdk-rust" version = "0.1.0-alpha.2" -source = "git+https://github.com/braintrustdata/braintrust-sdk-rust?rev=019e428ab80d0da4c00dab383d4b0d1fdf4486fd#019e428ab80d0da4c00dab383d4b0d1fdf4486fd" +source = "git+https://github.com/braintrustdata/braintrust-sdk-rust?rev=ac70efca9d2e72e8d5446d73601312e08f1c6148#ac70efca9d2e72e8d5446d73601312e08f1c6148" dependencies = [ "anyhow", "arc-swap", @@ -278,6 +278,7 @@ dependencies = [ "clap", "fs2", "libc", + "parking_lot", "regex", "reqwest", "rquickjs", @@ -1125,6 +1126,15 @@ version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "log" version = "0.4.33" @@ -1240,6 +1250,29 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + [[package]] name = "percent-encoding" version = "2.3.2" @@ -1422,6 +1455,15 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + [[package]] name = "regex" version = "1.13.1" @@ -1642,6 +1684,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "security-framework" version = "3.7.0" diff --git a/bt-daemon/Cargo.toml b/bt-daemon/Cargo.toml index b4ff59d1..d359c724 100644 --- a/bt-daemon/Cargo.toml +++ b/bt-daemon/Cargo.toml @@ -18,7 +18,7 @@ required-features = ["cli"] [dependencies] # Keep this pinned exactly, matching the dependency policy used by `bt`. -braintrust-sdk-rust = { git = "https://github.com/braintrustdata/braintrust-sdk-rust", rev = "019e428ab80d0da4c00dab383d4b0d1fdf4486fd" } +braintrust-sdk-rust = { git = "https://github.com/braintrustdata/braintrust-sdk-rust", rev = "ac70efca9d2e72e8d5446d73601312e08f1c6148" } base64 = "0.22" anyhow = "1" async-trait = "0.1" @@ -46,6 +46,7 @@ windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Securit [dev-dependencies] axum = "0.8" bytes = "1" +parking_lot = "0.12" reqwest = { version = "0.12", default-features = false, features = ["json", "stream"] } wiremock = "0.6" zstd = "0.13" diff --git a/bt-daemon/docs/protocol.md b/bt-daemon/docs/protocol.md index 9b29323d..bb0a010d 100644 --- a/bt-daemon/docs/protocol.md +++ b/bt-daemon/docs/protocol.md @@ -247,8 +247,30 @@ downgrade it. } ``` -The Braintrust sink records the capture package version in -`context.span_origin.version`. +The shared delivery path snapshots provenance independently on the coding-agent +session root and each turn: +`context.span_origin.name = "braintrust.plugin."`, +`context.span_origin.version` is the span-creating event's capture package +version, `context.span_origin..version` is its coding-agent +`source_version`, and `context.span_origin.bt.version` is the daemon version at +span creation. `` is the canonical name, such as `codex` or `claude-code`, +not a literal `agent` key. Unknown plugin and agent versions are explicit JSON +`null`; they never fall back to another component's version. The +destination-scoped delivery ledger persists each snapshot before export; later +updates, resume, and replay reuse it unchanged, even across upgrades. +New turns capture new snapshots instead of inheriting the root's versions. +Provisional roots retain their original snapshot while waiting for work. +An unmaterialized root or turn replayed for a new destination uses its +span-creating journaled event, not the event that triggered recovery. Other +descendants do not receive these snapshots. Transcript backfills capture +versions available at materialization, not unrecorded historical versions. See the +[span-origin contract](../../docs/coding-agent-tracing-plugin-spec.md#621-span-origin-versions). + +For acknowledged spans from legacy ledgers without saved snapshots, stateless +updates omit `context.span_origin` entirely, including SDK defaults and inherited +parent origin fields. This preserves existing backend provenance without guessing +historical versions; output, timing, and unrelated propagated context still merge +normally. New turns receive fresh snapshots. Field notes: diff --git a/bt-daemon/src/delivery_ledger.rs b/bt-daemon/src/delivery_ledger.rs index 934d0d18..fef15c13 100644 --- a/bt-daemon/src/delivery_ledger.rs +++ b/bt-daemon/src/delivery_ledger.rs @@ -8,10 +8,10 @@ //! destination still receives the full trace. use crate::sink::Sink; -use crate::translate::{SpanOp, SpanRow}; +use crate::translate::{OriginSnapshot, SpanOp, SpanRow, SpanType}; use crate::wire::SessionConfig; use sha2::{Digest, Sha256}; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; #[derive(serde::Serialize, serde::Deserialize, Default)] @@ -20,6 +20,8 @@ struct LedgerFile { completed_span_ids: HashSet, #[serde(default, alias = "late_merge_span_ids")] late_merge_ids: HashSet, + #[serde(default)] + span_origins: HashMap, } struct DeliveryLedger { @@ -36,7 +38,7 @@ impl DeliveryLedger { source: &str, session_id: &str, config: &SessionConfig, - ) -> anyhow::Result { + ) -> anyhow::Result<(Self, HashMap)> { let fingerprint = serde_json::json!({ "api_url": config.auth.api_url, "org_id": config.auth.org_id, @@ -55,13 +57,16 @@ impl DeliveryLedger { Err(error) if error.kind() == std::io::ErrorKind::NotFound => LedgerFile::default(), Err(error) => return Err(error.into()), }; - Ok(Self { - path, - known: persisted.completed_span_ids, - pending: HashSet::new(), - known_late_merges: persisted.late_merge_ids, - pending_late_merges: HashSet::new(), - }) + Ok(( + Self { + path, + known: persisted.completed_span_ids, + pending: HashSet::new(), + known_late_merges: persisted.late_merge_ids, + pending_late_merges: HashSet::new(), + }, + persisted.span_origins, + )) } fn filter(&self, ops: &[SpanOp]) -> Vec { @@ -95,13 +100,20 @@ impl DeliveryLedger { } } - async fn commit(&mut self) -> anyhow::Result<()> { + async fn commit( + &mut self, + span_origins: &HashMap, + ) -> anyhow::Result<()> { if self.pending.is_empty() && self.pending_late_merges.is_empty() { return Ok(()); } self.known.extend(self.pending.drain()); self.known_late_merges .extend(self.pending_late_merges.drain()); + self.save(span_origins).await + } + + async fn save(&self, span_origins: &HashMap) -> anyhow::Result<()> { let parent = self.path.parent().expect("ledger path has a parent"); tokio::fs::create_dir_all(parent).await?; let temp = self @@ -112,6 +124,7 @@ impl DeliveryLedger { serde_json::to_vec(&LedgerFile { completed_span_ids: self.known.clone(), late_merge_ids: self.known_late_merges.clone(), + span_origins: span_origins.clone(), })?, ) .await?; @@ -139,18 +152,25 @@ fn late_merge_id(row: &SpanRow, key: &str) -> String { pub(crate) struct LedgerSink { inner: Box, ledger: Option, + span_origins: HashMap, + origins_dirty: bool, + plugin_version: Option, + source_version: Option, + bt_version: String, + attached_parent: Option, } impl LedgerSink { pub(crate) async fn new( inner: Box, - data_dir: &Path, + data_dir: Option<&Path>, source: &str, session_id: &str, config: Option<&SessionConfig>, + bt_version: &str, ) -> Self { - let ledger = match config { - Some(config) if config.destination.is_some() => { + let loaded = match (data_dir, config) { + (Some(data_dir), Some(config)) if config.destination.is_some() => { DeliveryLedger::load(data_dir, source, session_id, config) .await .map_err(|error| { @@ -160,7 +180,20 @@ impl LedgerSink { } _ => None, }; - Self { inner, ledger } + let (ledger, span_origins) = match loaded { + Some((ledger, origins)) => (Some(ledger), origins), + None => (None, HashMap::new()), + }; + Self { + inner, + ledger, + span_origins, + origins_dirty: false, + plugin_version: None, + source_version: None, + bt_version: bt_version.to_owned(), + attached_parent: config.and_then(|config| config.attached_span_ids().0), + } } } @@ -168,10 +201,20 @@ impl LedgerSink { impl Sink for LedgerSink { fn configure(&mut self, config: &SessionConfig) { self.inner.configure(config); + self.attached_parent = config.attached_span_ids().0; + } + + fn set_capture_versions(&mut self, plugin_version: Option<&str>, source_version: Option<&str>) { + if self.plugin_version.as_deref() != plugin_version { + self.plugin_version = plugin_version.map(str::to_owned); + } + if self.source_version.as_deref() != source_version { + self.source_version = source_version.map(str::to_owned); + } } async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result { - let filtered = self + let mut filtered = self .ledger .as_ref() .map(|ledger| ledger.filter(ops)) @@ -179,6 +222,45 @@ impl Sink for LedgerSink { if filtered.is_empty() { return Ok(0); } + for op in &mut filtered { + let is_insert = matches!(op, SpanOp::Insert(_)); + let row = match op { + SpanOp::Insert(row) | SpanOp::Merge(row) => row, + }; + let is_root = row.span_type == SpanType::Task + && match self.attached_parent.as_deref() { + Some(parent) => { + row.parent_span_ids.len() == 1 + && row.parent_span_ids[0] == parent + && row.span_id != parent + } + None => row.parent_span_ids.is_empty(), + }; + if is_insert && (is_root || row.is_turn) { + if let std::collections::hash_map::Entry::Vacant(entry) = + self.span_origins.entry(row.span_id.clone()) + { + entry.insert(OriginSnapshot { + plugin_version: self.plugin_version.clone(), + bt_version: self.bt_version.clone(), + source_version: self.source_version.clone(), + }); + self.origins_dirty = true; + } + } + // Stamp only known session roots and turns, including stateless updates. + // This also discards provenance supplied by a span-processing plugin. + row.origin = self.span_origins.get(&row.span_id).cloned(); + } + if self.origins_dirty { + // Persist birth versions before the SDK can queue a span. A crash + // between delivery and its checkpoint must not change its provenance. + // Do not acknowledge pending span deliveries before the sink flushes. + if let Some(ledger) = &self.ledger { + ledger.save(&self.span_origins).await?; + } + self.origins_dirty = false; + } let emitted = self.inner.emit(&filtered).await?; // A deferred lifecycle root has not reached the backend yet. Do not // record it as delivered: a subsequent cold-worker replay needs to @@ -194,7 +276,7 @@ impl Sink for LedgerSink { async fn flush(&mut self) -> anyhow::Result<()> { self.inner.flush().await?; if let Some(ledger) = &mut self.ledger { - ledger.commit().await?; + ledger.commit(&self.span_origins).await?; } Ok(()) } @@ -213,7 +295,8 @@ mod tests { use super::*; use crate::translate::SpanRow; use crate::wire::{BackendAuth, FlushMode, TraceDestination}; - use std::sync::{Arc, Mutex}; + use parking_lot::Mutex; + use std::sync::Arc; #[derive(Default)] struct RecordingSink { @@ -223,7 +306,7 @@ mod tests { #[async_trait::async_trait] impl Sink for RecordingSink { async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result { - self.emitted.lock().unwrap().extend_from_slice(ops); + self.emitted.lock().extend_from_slice(ops); Ok(ops.len() as u64) } @@ -288,6 +371,84 @@ mod tests { }) } + #[tokio::test] + async fn open_turn_snapshot_survives_recovery_before_delivery_checkpoint() { + let temp = tempfile::tempdir().unwrap(); + let config = config("project-a"); + let mut first = LedgerSink::new( + Box::new(RecordingSink::default()), + Some(temp.path()), + "codex", + "session-1", + Some(&config), + "bt1", + ) + .await; + first.set_capture_versions(None, Some("native1")); + let turn = SpanRow { + span_id: "turn-1".into(), + root_span_id: "root".into(), + parent_span_ids: vec!["root".into()], + is_turn: true, + ..Default::default() + }; + first.emit(&[SpanOp::Insert(turn.clone())]).await.unwrap(); + // Lose in-memory state without flushing/checkpointing delivery. + drop(first); + + let output = Arc::new(Mutex::new(Vec::new())); + let mut recovered = LedgerSink::new( + Box::new(RecordingSink { + emitted: output.clone(), + }), + Some(temp.path()), + "codex", + "session-1", + Some(&config), + "bt2", + ) + .await; + recovered.set_capture_versions(Some("v2"), Some("native2")); + recovered + .emit(&[ + SpanOp::Insert(turn.clone()), + partial_merge("turn-1"), + SpanOp::Insert(SpanRow { + span_id: "turn-2".into(), + ..turn.clone() + }), + SpanOp::Insert(SpanRow { + span_id: "subagent-container".into(), + is_turn: false, + ..turn + }), + ]) + .await + .unwrap(); + let rows = output.lock(); + assert_eq!(rows.len(), 4); + for op in rows.iter() { + let row = match op { + SpanOp::Insert(row) | SpanOp::Merge(row) => row, + }; + let expected = match row.span_id.as_str() { + "turn-1" => Some(OriginSnapshot { + plugin_version: None, + source_version: Some("native1".into()), + bt_version: "bt1".into(), + }), + "turn-2" => Some(OriginSnapshot { + plugin_version: Some("v2".into()), + source_version: Some("native2".into()), + bt_version: "bt2".into(), + }), + "subagent-container" => None, + unexpected => panic!("unexpected span: {unexpected}"), + }; + assert_eq!(row.origin, expected, "{}", row.span_id); + } + } + #[tokio::test] async fn a_destination_receives_a_terminal_span_only_once_across_sink_instances() { let temp = tempfile::tempdir().unwrap(); @@ -297,15 +458,16 @@ mod tests { }; let mut first = LedgerSink::new( Box::new(first), - temp.path(), + Some(temp.path()), "codex", "session-1", Some(&config("project-a")), + "test", ) .await; first.emit(&[terminal("span-1")]).await.unwrap(); first.flush().await.unwrap(); - assert_eq!(first_output.lock().unwrap().len(), 1); + assert_eq!(first_output.lock().len(), 1); let mut same_destination = config("project-a"); same_destination.additional_metadata = Some(serde_json::json!({"run_id": "new"})); @@ -315,16 +477,17 @@ mod tests { }; let mut repeated = LedgerSink::new( Box::new(repeated), - temp.path(), + Some(temp.path()), "codex", "session-1", Some(&same_destination), + "test", ) .await; assert_eq!(repeated.emit(&[terminal("span-1")]).await.unwrap(), 0); assert_eq!(repeated.emit(&[partial_merge("span-1")]).await.unwrap(), 0); repeated.flush().await.unwrap(); - assert!(repeated_output.lock().unwrap().is_empty()); + assert!(repeated_output.lock().is_empty()); } #[tokio::test] @@ -333,10 +496,11 @@ mod tests { let first = RecordingSink::default(); let mut first = LedgerSink::new( Box::new(first), - temp.path(), + Some(temp.path()), "grok", "session-1", Some(&config("project-a")), + "test", ) .await; assert_eq!(first.emit(&[terminal("span-1")]).await.unwrap(), 1); @@ -345,10 +509,11 @@ mod tests { let second = RecordingSink::default(); let mut second = LedgerSink::new( Box::new(second), - temp.path(), + Some(temp.path()), "grok", "session-1", Some(&config("project-a")), + "test", ) .await; assert_eq!( @@ -374,10 +539,11 @@ mod tests { let third = RecordingSink::default(); let mut third = LedgerSink::new( Box::new(third), - temp.path(), + Some(temp.path()), "grok", "session-1", Some(&config("project-a")), + "test", ) .await; assert_eq!( @@ -396,10 +562,11 @@ mod tests { let first = RecordingSink::default(); let mut first = LedgerSink::new( Box::new(first), - temp.path(), + Some(temp.path()), "codex", "session-1", Some(&config("project-a")), + "test", ) .await; assert_eq!( @@ -414,10 +581,11 @@ mod tests { let resumed = RecordingSink::default(); let mut resumed = LedgerSink::new( Box::new(resumed), - temp.path(), + Some(temp.path()), "codex", "session-1", Some(&config("project-a")), + "test", ) .await; assert_eq!( @@ -446,10 +614,11 @@ mod tests { }; let mut initial = LedgerSink::new( Box::new(initial), - temp.path(), + Some(temp.path()), "claude-code", "session-1", Some(&config("project-a")), + "test", ) .await; initial.emit(&[terminal("span-1")]).await.unwrap(); @@ -461,15 +630,16 @@ mod tests { }; let mut replay = LedgerSink::new( Box::new(replay), - temp.path(), + Some(temp.path()), "claude-code", "session-1", Some(&config("project-b")), + "test", ) .await; assert_eq!(replay.emit(&[terminal("span-1")]).await.unwrap(), 1); replay.flush().await.unwrap(); - assert_eq!(replay_output.lock().unwrap().len(), 1); + assert_eq!(replay_output.lock().len(), 1); } #[tokio::test] async fn an_authoritative_terminal_merge_records_completion_and_late_delivery() { @@ -477,10 +647,11 @@ mod tests { let first = RecordingSink::default(); let mut first = LedgerSink::new( Box::new(first), - temp.path(), + Some(temp.path()), "grok", "session-1", Some(&config("project-a")), + "test", ) .await; let mut authoritative = terminal("span-1"); @@ -494,10 +665,11 @@ mod tests { let repeated = RecordingSink::default(); let mut repeated = LedgerSink::new( Box::new(repeated), - temp.path(), + Some(temp.path()), "grok", "session-1", Some(&config("project-a")), + "test", ) .await; assert_eq!(repeated.emit(&[terminal("span-1")]).await.unwrap(), 0); diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index defe8b7b..f259717a 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -56,7 +56,7 @@ pub(crate) struct SessionOptions { pub session_id: String, pub translator_session_id: String, pub source: String, - pub plugin_version: Option, + pub bt_version: String, pub replay: Option, pub config: crate::wire::SessionConfig, pub correlation_key: String, @@ -88,7 +88,7 @@ impl Session { session_id, translator_session_id, source, - plugin_version, + bt_version, replay, config, correlation_key, @@ -107,7 +107,7 @@ impl Session { session_id: session_id.clone(), translator_session_id, source: source.clone(), - plugin_version, + bt_version, translators, sink_factory, counters: counters.clone(), @@ -450,7 +450,7 @@ struct SessionActor { session_id: String, translator_session_id: String, source: String, - plugin_version: Option, + bt_version: String, translators: Arc, sink_factory: Arc, counters: Arc, @@ -508,11 +508,7 @@ impl SessionActor { return; } }; - let sink = match self.sink_factory.create( - &self.session_id, - &self.source, - self.plugin_version.as_deref(), - ) { + let sink = match self.sink_factory.create(&self.session_id, &self.source) { Ok(s) => s, Err(e) => { self.set_error(format!("sink init failed: {e}")); @@ -547,10 +543,11 @@ impl SessionActor { let mut sink: Box = Box::new( LedgerSink::new( sink, - &self.data_dir, + Some(&self.data_dir), &self.source, &self.session_id, Some(&self.config), + &self.bt_version, ) .await, ); @@ -582,6 +579,10 @@ impl SessionActor { ctx.config = Some(cfg.clone()); self.refresh_permalink(sink.as_ref()); } + sink.set_capture_versions( + env.plugin_version.as_deref(), + env.source_version.as_deref(), + ); let translated = translator.handle(&env, &ctx); let (correlation_changed, delivered) = self .emit_translator_batches( @@ -847,6 +848,7 @@ impl SessionActor { payload.insert("_bt_transcript_replay".to_string(), serde_json::json!(true)); } } + sink.set_capture_versions(env.plugin_version.as_deref(), env.source_version.as_deref()); let translated = translator.handle(&env, ctx); if entry_through <= plan.acknowledged_through { self.replay_without_delivery(translator, ctx, translated) diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 3c7b7f35..ae371a72 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -72,4 +72,6 @@ pub use trace_runtime::{ run_trace, HostRouteSelection, RouteRequirements, TraceHostContext, TraceHostServices, }; pub use transcript_import::{import_transcript, import_transcripts, run_import}; -pub use translate::{AgentTranslator, Registry, SessionCtx, SpanOp, SpanRow, SpanType}; +pub use translate::{ + AgentTranslator, OriginSnapshot, Registry, SessionCtx, SpanOp, SpanRow, SpanType, +}; diff --git a/bt-daemon/src/main.rs b/bt-daemon/src/main.rs index c1740938..9e9714fa 100644 --- a/bt-daemon/src/main.rs +++ b/bt-daemon/src/main.rs @@ -251,11 +251,7 @@ async fn main() { let mut opts = if debug_sink { debug_serve_options(VERSION, &data_dir) } else { - let cfg = BraintrustSinkConfig { - api_url, - app_url, - version: VERSION.to_string(), - }; + let cfg = BraintrustSinkConfig { api_url, app_url }; braintrust_serve_options(VERSION, cfg, Arc::new(Registry::default_agents())) }; opts.auth_provider = Some(Arc::new(EnvironmentAuthProvider { diff --git a/bt-daemon/src/server/sessions.rs b/bt-daemon/src/server/sessions.rs index d871f194..55985a53 100644 --- a/bt-daemon/src/server/sessions.rs +++ b/bt-daemon/src/server/sessions.rs @@ -90,7 +90,7 @@ impl Daemon { session_id: env.session_id.clone(), translator_session_id, source: env.source.clone(), - plugin_version: env.plugin_version.clone(), + bt_version: self.version.clone(), replay: Some(replay), config, correlation_key: key.correlation_key(), diff --git a/bt-daemon/src/sink/braintrust.rs b/bt-daemon/src/sink/braintrust.rs index 34555a65..1a53619a 100644 --- a/bt-daemon/src/sink/braintrust.rs +++ b/bt-daemon/src/sink/braintrust.rs @@ -14,9 +14,9 @@ use crate::translate::{SpanOp, SpanRow, SpanType}; use crate::wire::{SessionConfig, TraceDestination}; use braintrust_sdk_rust::{ BraintrustClient, ParentSpanInfo, SpanComponents, SpanHandle, SpanLog, SpanObjectType, - SpanOrigin, SpanType as SdkSpanType, DEFAULT_API_URL, DEFAULT_APP_URL, + SpanType as SdkSpanType, DEFAULT_API_URL, DEFAULT_APP_URL, }; -use serde_json::{Map, Value}; +use serde_json::{json, Map, Value}; use std::collections::{HashMap, VecDeque}; use std::io::Write; use std::sync::Arc; @@ -29,13 +29,11 @@ use tokio::sync::Mutex as AsyncMutex; pub struct BraintrustSinkConfig { pub api_url: Option, pub app_url: Option, - pub version: String, } /// Lazily-built, shared-by-URL client pool. struct ClientCache { state: AsyncMutex, - version: String, } const CLIENT_CACHE_CAPACITY: usize = 16; @@ -103,10 +101,9 @@ impl ClientCacheState { } impl ClientCache { - fn new(version: String) -> Self { + fn new() -> Self { Self { state: AsyncMutex::new(ClientCacheState::default()), - version, } } @@ -125,7 +122,6 @@ impl ClientCache { } let client = BraintrustClient::builder() .skip_login(true) - .span_origin(SpanOrigin::new().version(self.version.clone())) .api_url(api_url.to_string()) .app_url(app_url.to_string()) .build() @@ -147,32 +143,24 @@ pub struct BraintrustSinkFactory { cache: Arc, default_api_url: Option, default_app_url: Option, - version: String, } impl BraintrustSinkFactory { pub fn new(cfg: BraintrustSinkConfig) -> Self { Self { - cache: Arc::new(ClientCache::new(cfg.version.clone())), + cache: Arc::new(ClientCache::new()), default_api_url: cfg.api_url, default_app_url: cfg.app_url, - version: cfg.version, } } } impl SinkFactory for BraintrustSinkFactory { - fn create( - &self, - _session_id: &str, - source: &str, - plugin_version: Option<&str>, - ) -> anyhow::Result> { + fn create(&self, _session_id: &str, source: &str) -> anyhow::Result> { Ok(Box::new(BraintrustSink { cache: self.cache.clone(), default_api_url: self.default_api_url.clone(), default_app_url: self.default_app_url.clone(), - version: plugin_version.unwrap_or(&self.version).to_string(), source: source.to_string(), creds: None, urls: None, @@ -205,7 +193,6 @@ struct BraintrustSink { cache: Arc, default_api_url: Option, default_app_url: Option, - version: String, source: String, creds: Option, /// Resolved `(api_url, app_url)` for this session, from its config. @@ -268,13 +255,6 @@ impl BraintrustSink { Ok(client) } - fn span_origin(&self) -> SpanOrigin { - SpanOrigin::new() - .name(format!("braintrust.plugin.{}", self.source)) - .version(self.version.clone()) - .instrumentation("braintrust-plugin") - } - fn ensure_handle(&mut self, client: &BraintrustClient, row: &SpanRow) -> anyhow::Result<()> { if self.open.contains_key(&row.span_id) { return Ok(()); @@ -291,8 +271,7 @@ impl BraintrustSink { .span_type(map_span_type(row.span_type)) .span_id(row.span_id.clone()) .row_id(row.span_id.clone()) - .parent_info(parent) - .span_origin(self.span_origin()); + .parent_info(parent); if let Some(org_name) = &creds.org_name { builder = builder.org_name(org_name.clone()); } @@ -306,7 +285,7 @@ impl BraintrustSink { fn update_open(&mut self, client: &BraintrustClient, row: &SpanRow) -> anyhow::Result<()> { self.ensure_handle(client, row)?; let handle = self.open.get(&row.span_id).expect("just inserted"); - handle.log(build_log(row, self.span_origin())?); + handle.log(build_log(row, &self.source, false)?); if let Some(end) = row.end_ms { handle.end_with_time(ms_to_secs(end)); // SpanHandle retains the complete accumulated input/output. Once a @@ -323,13 +302,25 @@ impl BraintrustSink { .as_ref() .ok_or_else(|| anyhow::anyhow!("session has no credentials/config yet"))?; let project = self.project(creds); - let components = self.span_components(row, creds, &project).to_str(); + let mut components = self.span_components(row, creds, &project); + if row.origin.is_none() { + // An attached parent's provenance is not the existing span's origin. + // Legacy spans have no birth snapshot with which to override it. + if let Some(context) = components + .propagated_event + .as_mut() + .and_then(|event| event.get_mut("context")) + .and_then(Value::as_object_mut) + { + context.remove("span_origin"); + } + } client .update_span_with_credentials( creds.token.clone(), creds.org_id.clone(), - &components, - build_log(row, self.span_origin())?, + &components.to_str(), + build_log(row, &self.source, true)?, ) .map_err(|error| anyhow::anyhow!("braintrust span merge failed: {error}")) } @@ -627,9 +618,26 @@ fn ms_to_secs(ms: i64) -> f64 { ms as f64 / 1000.0 } -fn build_log(row: &SpanRow, origin: SpanOrigin) -> anyhow::Result { - // Repeat the plugin origin so stateless merges cannot use SDK defaults. - let mut builder = SpanLog::builder().span_origin(origin); +fn build_log(row: &SpanRow, source: &str, is_merge: bool) -> anyhow::Result { + let mut builder = SpanLog::builder(); + if let Some(origin) = &row.origin { + // The ledger supplies the immutable birth snapshot on every root/turn op. + // Explicit null prevents SDK version defaults on spans with unknown + // plugin versions, including stateless merges after close or recovery. + builder = builder.context(json!({ + "span_origin": { + "name": format!("braintrust.plugin.{source}"), + "version": origin.plugin_version, + "bt": { "version": origin.bt_version }, + (source): { "version": origin.source_version }, + "instrumentation": { "name": "braintrust-plugin" }, + }, + })); + } else if is_merge { + // An update without a saved snapshot must leave existing provenance + // untouched, not replace legacy plugin attribution with SDK defaults. + builder = builder.skip_span_origin(); + } // The span's display name is carried on the log event, not the builder. // An empty name means "unchanged" (many merge ops use `..Default::default()` @@ -703,3 +711,93 @@ fn serialized_len(op: &SpanOp) -> serde_json::Result { serde_json::to_writer(&mut counter, op)?; Ok(counter.0) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::translate::OriginSnapshot; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + #[tokio::test] + async fn root_context_retains_sdk_environment_on_insert_and_stateless_merge() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/version")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/logs3")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) + .mount(&server) + .await; + let client = BraintrustClient::builder() + .skip_login(true) + .api_url(server.uri()) + .app_url(server.uri()) + .environment("test", Some("sink-test")) + .build() + .await + .unwrap(); + let row = SpanRow { + span_id: "root".into(), + root_span_id: "root".into(), + origin: Some(OriginSnapshot { + plugin_version: None, + bt_version: "birth-bt".into(), + source_version: None, + }), + ..Default::default() + }; + let handle = client + .span_builder_with_credentials("sk-test", "org-test") + .span_id(row.span_id.clone()) + .row_id(row.span_id.clone()) + .parent_info(ParentSpanInfo::ProjectLogs { + object_id: "proj-test".into(), + }) + .build(); + handle.log(build_log(&row, "codex", false).unwrap()); + handle.end_with_time(2.0); + client.flush().await.unwrap(); + + let mut components = SpanComponents::new(SpanObjectType::ProjectLogs); + components.object_id = Some("proj-test".into()); + components.row_id = Some(row.span_id.clone()); + components.span_id = Some(row.span_id.clone()); + components.root_span_id = Some(row.root_span_id.clone()); + client + .update_span_with_credentials( + "sk-test", + "org-test", + &components.to_str(), + build_log(&row, "codex", true).unwrap(), + ) + .unwrap(); + client.flush().await.unwrap(); + + let requests = server.received_requests().await.unwrap(); + let rows: Vec = requests + .iter() + .filter(|request| request.url.path() == "/logs3") + .flat_map(|request| { + serde_json::from_slice::(&request.body).unwrap()["rows"] + .as_array() + .unwrap() + .clone() + }) + .collect(); + assert!(rows.len() >= 2); + for row in rows { + let origin = &row["context"]["span_origin"]; + assert_eq!(origin["name"], "braintrust.plugin.codex"); + assert_eq!(origin.get("version"), Some(&Value::Null)); + assert_eq!(origin["bt"]["version"], "birth-bt"); + assert_eq!( + origin["environment"], + json!({"type": "test", "name": "sink-test"}) + ); + } + } +} diff --git a/bt-daemon/src/sink/debug.rs b/bt-daemon/src/sink/debug.rs index 97cd33d3..95bc57d4 100644 --- a/bt-daemon/src/sink/debug.rs +++ b/bt-daemon/src/sink/debug.rs @@ -16,12 +16,7 @@ pub struct DebugSinkFactory { } impl SinkFactory for DebugSinkFactory { - fn create( - &self, - session_id: &str, - source: &str, - _plugin_version: Option<&str>, - ) -> anyhow::Result> { + fn create(&self, session_id: &str, source: &str) -> anyhow::Result> { std::fs::create_dir_all(&self.dir)?; let name = format!("{}.ndjson", sanitize(session_id)); // Fall back to a stable digest when the id cannot be one file name. diff --git a/bt-daemon/src/sink/mod.rs b/bt-daemon/src/sink/mod.rs index 043cb966..220e494c 100644 --- a/bt-daemon/src/sink/mod.rs +++ b/bt-daemon/src/sink/mod.rs @@ -27,6 +27,14 @@ pub trait Sink: Send { let _ = config; } + /// Versions of the event currently being translated, including replay. + fn set_capture_versions( + &mut self, + _plugin_version: Option<&str>, + _source_version: Option<&str>, + ) { + } + /// Emit span ops. Returns the number of rows written, for status counters. async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result; @@ -46,14 +54,8 @@ pub trait Sink: Send { } } -/// Builds a sink per session. `source` and `plugin_version` identify the -/// instrumentation that captured the events and are used to stamp -/// `context.span_origin` centrally. +/// Builds a sink per session. Capture provenance belongs to span creation, +/// not to the event that happens to create the session actor. pub trait SinkFactory: Send + Sync { - fn create( - &self, - session_id: &str, - source: &str, - plugin_version: Option<&str>, - ) -> anyhow::Result>; + fn create(&self, session_id: &str, source: &str) -> anyhow::Result>; } diff --git a/bt-daemon/src/span_processor.rs b/bt-daemon/src/span_processor.rs index ca6afbfb..f11fa9d9 100644 --- a/bt-daemon/src/span_processor.rs +++ b/bt-daemon/src/span_processor.rs @@ -238,7 +238,7 @@ pub fn process( engine.failed_plugins.remove(plugin); } let candidate = engine.call(plugin, &row, operation, source, session_id); - let candidate = match candidate { + let mut candidate = match candidate { Ok(candidate) if ( candidate.span_id.as_str(), @@ -277,6 +277,8 @@ pub fn process( break; } }; + // Turn identity belongs to the translator, not the JSON transform. + candidate.is_turn = row.is_turn; row = candidate; } Ok(()) diff --git a/bt-daemon/src/trace_runtime.rs b/bt-daemon/src/trace_runtime.rs index d69938a9..310ecfbb 100644 --- a/bt-daemon/src/trace_runtime.rs +++ b/bt-daemon/src/trace_runtime.rs @@ -182,7 +182,6 @@ fn serve_options(host: &TraceHostContext) -> ServeOptions { let cfg = BraintrustSinkConfig { api_url: None, app_url: None, - version: host.version.clone(), }; let mut options = braintrust_serve_options(&host.version, cfg, Arc::new(Registry::default_agents())); diff --git a/bt-daemon/src/transcript_import/pipeline.rs b/bt-daemon/src/transcript_import/pipeline.rs index fde2ebf3..5112a8a1 100644 --- a/bt-daemon/src/transcript_import/pipeline.rs +++ b/bt-daemon/src/transcript_import/pipeline.rs @@ -309,24 +309,18 @@ impl ImportProcessor { Some(live) => live, None => { let translator = self.opts.translators.create(&env.source, &sid); - let sink = self.opts.sink_factory.create( - &sid, - &env.source, - env.plugin_version.as_deref(), - )?; - let sink: Box = match &self.ledger_dir { - Some(ledger_dir) => Box::new( - delivery_ledger::LedgerSink::new( - sink, - ledger_dir, - &env.source, - &sid, - env.config.as_ref(), - ) - .await, - ), - None => sink, - }; + let sink = self.opts.sink_factory.create(&sid, &env.source)?; + let sink: Box = Box::new( + delivery_ledger::LedgerSink::new( + sink, + self.ledger_dir.as_deref(), + &env.source, + &sid, + env.config.as_ref(), + &self.opts.version, + ) + .await, + ); self.sessions.insert( sid.clone(), ImportLive { @@ -357,6 +351,8 @@ impl ImportProcessor { live.sink.configure(cfg); live.ctx.config = Some(cfg.clone()); } + live.sink + .set_capture_versions(env.plugin_version.as_deref(), env.source_version.as_deref()); let ops = live.translator.handle(&env, &live.ctx)?; Self::emit_translator_batches(live, ops).await?; } diff --git a/bt-daemon/src/translate/antigravity.rs b/bt-daemon/src/translate/antigravity.rs index 017eca3a..380a9217 100644 --- a/bt-daemon/src/translate/antigravity.rs +++ b/bt-daemon/src/translate/antigravity.rs @@ -221,6 +221,7 @@ impl AntigravityTranslator { parent_span_ids: vec![self.session_span_id.clone()], name: format!("Turn {}", self.turn_count), span_type: SpanType::Task, + is_turn: true, start_ms: Some(ts_ms), input: nonempty_value(input), metadata: Some(json!({"turn_number": self.turn_count})), diff --git a/bt-daemon/src/translate/claude.rs b/bt-daemon/src/translate/claude.rs index 9bcb6171..09ec9095 100644 --- a/bt-daemon/src/translate/claude.rs +++ b/bt-daemon/src/translate/claude.rs @@ -521,6 +521,7 @@ impl ClaudeTranslator { parent_span_ids: vec![parent_id.clone()], name, span_type: SpanType::Task, + is_turn: true, start_ms: Some(event.ts_ms), input: event.payload.get("prompt").cloned(), metadata: (!metadata.is_empty()).then_some(Value::Object(metadata)), diff --git a/bt-daemon/src/translate/codex.rs b/bt-daemon/src/translate/codex.rs index 39c5c528..1bdf9f9a 100644 --- a/bt-daemon/src/translate/codex.rs +++ b/bt-daemon/src/translate/codex.rs @@ -828,6 +828,9 @@ impl CodexTranslator { parent_span_ids: vec![scope.turn_parent_span_id.clone()], name: format!("turn: {turn_id}"), span_type: SpanType::Task, + // Native compaction hooks identify maintenance turns before catch-up. + is_turn: self.compaction_trigger_by_turn.get(&turn_id).is_none() + && !self.compaction_spans.contains(&turn_id), start_ms: Some(ts), metadata: Some(json!({ "turn_id": turn_id, "model": scope.model })), ..Default::default() diff --git a/bt-daemon/src/translate/cursor/mod.rs b/bt-daemon/src/translate/cursor/mod.rs index 969d9e88..89af6a88 100644 --- a/bt-daemon/src/translate/cursor/mod.rs +++ b/bt-daemon/src/translate/cursor/mod.rs @@ -356,6 +356,7 @@ impl CursorTranslator { &format!("Turn {}", self.turn_seq), SpanType::Task, ); + row.is_turn = true; row.start_ms = Some(ts); row.input = prompt.as_ref().map(|p| json!(p)); row.metadata = Some( diff --git a/bt-daemon/src/translate/debug.rs b/bt-daemon/src/translate/debug.rs index 484b439d..b954898a 100644 --- a/bt-daemon/src/translate/debug.rs +++ b/bt-daemon/src/translate/debug.rs @@ -63,6 +63,8 @@ impl AgentTranslator for DebugTranslator { error: None, late_merge_key: None, tags: None, + is_turn: false, + origin: None, })); } @@ -84,6 +86,8 @@ impl AgentTranslator for DebugTranslator { error: None, late_merge_key: None, tags: None, + is_turn: false, + origin: None, })); Ok(ops) diff --git a/bt-daemon/src/translate/grok.rs b/bt-daemon/src/translate/grok.rs index 9ddbdfa4..9712c59d 100644 --- a/bt-daemon/src/translate/grok.rs +++ b/bt-daemon/src/translate/grok.rs @@ -485,6 +485,7 @@ impl GrokTranslator { parent_span_ids: vec![self.session_span_id.clone()], name: format!("Turn {}", self.turn_seq), span_type: SpanType::Task, + is_turn: true, start_ms: Some(ts_ms), input, metadata: Some(json!({"transcript_turn_key": turn_key})), diff --git a/bt-daemon/src/translate/mod.rs b/bt-daemon/src/translate/mod.rs index 1e4ff483..eef0de36 100644 --- a/bt-daemon/src/translate/mod.rs +++ b/bt-daemon/src/translate/mod.rs @@ -48,6 +48,14 @@ pub enum SpanType { Tool, } +/// Immutable versions captured when a session root or turn is first created. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OriginSnapshot { + pub plugin_version: Option, + pub bt_version: String, + pub source_version: Option, +} + /// A resolved span row, ready for a sink to insert or merge. Field set is the /// subset every current plugin uses; extend as translators need more. #[derive(Debug, Clone, Default, Serialize, Deserialize)] @@ -81,6 +89,12 @@ pub struct SpanRow { /// Labels for filtering in Braintrust (e.g. `compaction`, `permission-request`). #[serde(default, skip_serializing_if = "Option::is_none")] pub tags: Option>, + /// Translator-owned marker for turn creation, not other task spans. + #[serde(skip)] + pub is_turn: bool, + /// Delivery-owned creation snapshot, present on session roots and turns. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub origin: Option, } /// A span operation. `Insert` creates (or replaces) a row; `Merge` updates an diff --git a/bt-daemon/src/translate/opencode.rs b/bt-daemon/src/translate/opencode.rs index 4214ec3a..878a1f6a 100644 --- a/bt-daemon/src/translate/opencode.rs +++ b/bt-daemon/src/translate/opencode.rs @@ -750,6 +750,7 @@ impl OpenCodeTranslator { parent_span_ids: vec![state.root_span_id.clone()], name: format!("Turn {}", state.turn_number), span_type: SpanType::Task, + is_turn: true, start_ms: Some(event.ts_ms), input: (!input.is_empty()).then_some(Value::String(input)), metadata: Some( diff --git a/bt-daemon/src/translate/pi.rs b/bt-daemon/src/translate/pi.rs index 22e1ffde..3054862d 100644 --- a/bt-daemon/src/translate/pi.rs +++ b/bt-daemon/src/translate/pi.rs @@ -613,6 +613,7 @@ impl PiTranslator { parent_span_ids: vec![self.root_span_id.clone()], name: format!("Turn {}", self.turn_seq), span_type: SpanType::Task, + is_turn: true, start_ms: Some(ts), input: Some(input), metadata: Some(metadata), diff --git a/bt-daemon/tests/braintrust_sink.rs b/bt-daemon/tests/braintrust_sink.rs index 7c684dcb..e694818e 100644 --- a/bt-daemon/tests/braintrust_sink.rs +++ b/bt-daemon/tests/braintrust_sink.rs @@ -5,7 +5,8 @@ use braintrust_sdk_rust::{SpanComponents, SpanObjectType}; use bt_daemon::wire::{BackendAuth, FlushMode, SessionConfig, TraceDestination}; use bt_daemon::{ - BraintrustSinkConfig, BraintrustSinkFactory, SinkFactory, SpanOp, SpanRow, SpanType, + BraintrustSinkConfig, BraintrustSinkFactory, OriginSnapshot, SinkFactory, SpanOp, SpanRow, + SpanType, }; use serde_json::{json, Value}; use wiremock::matchers::{method, path}; @@ -55,6 +56,12 @@ fn row( error: None, late_merge_key: None, tags: None, + origin: parents.is_empty().then(|| OriginSnapshot { + plugin_version: None, + bt_version: "test".into(), + source_version: None, + }), + is_turn: false, } } @@ -132,9 +139,8 @@ async fn explicit_project_id_does_not_register_a_project_name() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: Some(base.clone()), app_url: Some(base.clone()), - version: "test".into(), }); - let mut sink = factory.create("sess-id", "antigravity", None).unwrap(); + let mut sink = factory.create("sess-id", "antigravity").unwrap(); let mut config = session_config(&base); config.destination = Some(TraceDestination::ProjectLogs { project_id: Some("proj-existing".into()), @@ -181,10 +187,9 @@ async fn multi_profile_sessions_route_to_their_own_backend() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: None, app_url: None, - version: "test".into(), }); - let mut sink_a = factory.create("sess-a", "codex", None).unwrap(); + let mut sink_a = factory.create("sess-a", "codex").unwrap(); sink_a.configure(&session_config(&base_a)); sink_a .emit(&[SpanOp::Insert(row( @@ -200,7 +205,7 @@ async fn multi_profile_sessions_route_to_their_own_backend() { .unwrap(); sink_a.flush().await.unwrap(); - let mut sink_b = factory.create("sess-b", "codex", None).unwrap(); + let mut sink_b = factory.create("sess-b", "codex").unwrap(); sink_b.configure(&session_config(&base_b)); sink_b .emit(&[SpanOp::Insert(row( @@ -234,9 +239,8 @@ async fn merge_with_empty_name_does_not_clobber_the_original_name() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: Some(base.clone()), app_url: Some(base.clone()), - version: "test".into(), }); - let mut sink = factory.create("sess-1", "codex", None).unwrap(); + let mut sink = factory.create("sess-1", "codex").unwrap(); sink.configure(&session_config(&base)); let named = row("s1", "s1", &[], "codex: myapp", SpanType::Task, 1, None); @@ -262,9 +266,8 @@ async fn provisional_lifecycle_root_is_not_exported_without_work() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: Some(base.clone()), app_url: Some(base.clone()), - version: "test".into(), }); - let mut sink = factory.create("empty", "codex", None).unwrap(); + let mut sink = factory.create("empty", "codex").unwrap(); sink.configure(&session_config(&base)); let mut root = row("root", "root", &[], "codex", SpanType::Task, 1, None); @@ -292,9 +295,8 @@ async fn provisional_root_survives_a_nonterminal_flush_until_work_arrives() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: Some(base.clone()), app_url: Some(base.clone()), - version: "test".into(), }); - let mut sink = factory.create("flush", "codex", None).unwrap(); + let mut sink = factory.create("flush", "codex").unwrap(); sink.configure(&session_config(&base)); let mut root = row("root", "root", &[], "codex", SpanType::Task, 1, None); @@ -317,6 +319,12 @@ async fn provisional_root_survives_a_nonterminal_flush_until_work_arrives() { let bodies = logs3_bodies(&server).await; assert!(bodies.contains("codex")); assert!(bodies.contains("Turn 1")); + let rows = logs3_rows(&server).await; + let root = rows.iter().find(|row| row["span_id"] == "root").unwrap(); + let origin = &root["context"]["span_origin"]; + assert_eq!(origin["name"], "braintrust.plugin.codex"); + assert_eq!(origin.get("version"), Some(&Value::Null)); + assert_eq!(origin["bt"]["version"], "test"); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -326,9 +334,8 @@ async fn work_releases_only_its_direct_provisional_root() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: Some(base.clone()), app_url: Some(base.clone()), - version: "test".into(), }); - let mut sink = factory.create("multi-root", "opencode", None).unwrap(); + let mut sink = factory.create("multi-root", "opencode").unwrap(); sink.configure(&session_config(&base)); let mut root_a = row("root-a", "root-a", &[], "root A", SpanType::Task, 1, None); @@ -368,9 +375,8 @@ async fn provisional_root_is_exported_before_its_first_work_row() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: Some(base.clone()), app_url: Some(base.clone()), - version: "test".into(), }); - let mut sink = factory.create("work", "codex", None).unwrap(); + let mut sink = factory.create("work", "codex").unwrap(); sink.configure(&session_config(&base)); let mut root = row("root", "root", &[], "codex", SpanType::Task, 1, None); @@ -408,21 +414,20 @@ async fn late_merge_updates_a_completed_span_without_an_open_handle() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: Some(base.clone()), app_url: Some(base.clone()), - version: "test".into(), }); - // The same cached SDK client serves every source. Provenance must stay - // session-specific, including when the plugin version falls back to bt's. - for (source, plugin_version) in [ - ("codex", Some("1.2.3")), - ("claude", Some("2.3.4")), - ("grok", Some("3.4.5")), - ("opencode", Some("4.5.6")), - ("pi", Some("5.6.7")), - ("antigravity", None), + // Stateless child merges must retain linkage without adding root provenance. + for source in [ + "codex", + "claude-code", + "grok", + "opencode", + "pi", + "cursor", + "antigravity", ] { let session_id = format!("sess-late-{source}"); - let span_id = format!("finished-{source}"); - let mut sink = factory.create(&session_id, source, plugin_version).unwrap(); + let span_id = format!("finished-{session_id}"); + let mut sink = factory.create(&session_id, source).unwrap(); sink.configure(&session_config(&base)); sink.emit(&[SpanOp::Insert(row( @@ -448,8 +453,12 @@ async fn late_merge_updates_a_completed_span_without_an_open_handle() { sink.flush().await.unwrap(); // Recovery also starts without an open handle for an existing span. + let recovery_factory = BraintrustSinkFactory::new(BraintrustSinkConfig { + api_url: Some(base.clone()), + app_url: Some(base.clone()), + }); drop(sink); - let mut recovered = factory.create(&session_id, source, plugin_version).unwrap(); + let mut recovered = recovery_factory.create(&session_id, source).unwrap(); recovered.configure(&session_config(&base)); recovered .emit(&[SpanOp::Merge(SpanRow { @@ -477,13 +486,6 @@ async fn late_merge_updates_a_completed_span_without_an_open_handle() { assert_eq!(update["root_span_id"], "trace-root"); assert_eq!(update["span_parents"], json!(["turn-parent"])); } - for row in rows { - let origin = &row["context"]["span_origin"]; - assert_eq!(origin["name"], format!("braintrust.plugin.{source}")); - assert_eq!(origin["version"], plugin_version.unwrap_or("test")); - assert_eq!(origin["instrumentation"]["name"], "braintrust-plugin"); - assert!(row["metadata"].get("bt_daemon_version").is_none()); - } } } @@ -494,9 +496,8 @@ async fn a_resumed_session_can_extend_an_already_ended_root() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: Some(base.clone()), app_url: Some(base.clone()), - version: "test".into(), }); - let mut sink = factory.create("sess-resumed", "antigravity", None).unwrap(); + let mut sink = factory.create("sess-resumed", "antigravity").unwrap(); sink.configure(&session_config(&base)); let root = row( @@ -549,9 +550,8 @@ async fn attached_trace_children_keep_the_external_root() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: Some(base.clone()), app_url: Some(base.clone()), - version: "test".into(), }); - let mut sink = factory.create("sess-1", "codex", None).unwrap(); + let mut sink = factory.create("sess-1", "codex").unwrap(); let mut config = session_config(&base); let mut components = SpanComponents::new(SpanObjectType::ProjectLogs); components.object_id = Some("proj-parent".into()); @@ -586,9 +586,8 @@ async fn exported_parent_preserves_object_root_and_propagated_event() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: Some(base.clone()), app_url: Some(base.clone()), - version: "test".into(), }); - let mut sink = factory.create("sess-parent", "codex", None).unwrap(); + let mut sink = factory.create("sess-parent", "codex").unwrap(); let mut config = session_config(&base); let mut components = SpanComponents::new(SpanObjectType::Experiment); components.object_id = Some("exp-parent".into()); @@ -600,7 +599,7 @@ async fn exported_parent_preserves_object_root_and_propagated_event() { )])); config.destination = Some(TraceDestination::ParentSpan { components }); sink.configure(&config); - sink.emit(&[SpanOp::Insert(row( + let mut root = row( "session-root", "daemon-internal-root", &["external-parent"], @@ -608,9 +607,13 @@ async fn exported_parent_preserves_object_root_and_propagated_event() { SpanType::Task, 1, Some(2), - ))]) - .await - .unwrap(); + ); + root.origin = Some(OriginSnapshot { + plugin_version: Some("attached-plugin".into()), + bt_version: "attached-bt".into(), + source_version: Some("attached-codex".into()), + }); + sink.emit(&[SpanOp::Insert(root)]).await.unwrap(); sink.flush().await.unwrap(); let bodies = logs3_bodies(&server).await; @@ -623,6 +626,67 @@ async fn exported_parent_preserves_object_root_and_propagated_event() { ] { assert!(bodies.contains(expected), "{expected} absent: {bodies}"); } + let rows = logs3_rows(&server).await; + let root = rows + .iter() + .find(|row| row["span_id"] == "session-root") + .unwrap(); + let origin = &root["context"]["span_origin"]; + assert_eq!(origin["name"], "braintrust.plugin.codex"); + assert_eq!(origin["version"], "attached-plugin"); + assert_eq!(origin["bt"]["version"], "attached-bt"); + assert_eq!(origin["codex"]["version"], "attached-codex"); + assert!(origin.get("agent").is_none()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn legacy_merge_preserves_origin_without_losing_attached_context() { + let server = mock_backend().await; + let base = server.uri(); + let factory = BraintrustSinkFactory::new(BraintrustSinkConfig::default()); + let mut sink = factory.create("legacy-session", "codex").unwrap(); + let mut config = session_config(&base); + let mut components = SpanComponents::new(SpanObjectType::Experiment); + components.object_id = Some("exp-parent".into()); + components.span_id = Some("external-parent".into()); + components.root_span_id = Some("external-root".into()); + components.propagated_event = Some(serde_json::Map::from_iter([ + ("tenant".into(), json!("acme")), + ( + "context".into(), + json!({ + "request_id": "request-1", + "span_origin": {"name": "parent-instrumentation", "version": "parent-version"} + }), + ), + ])); + config.destination = Some(TraceDestination::ParentSpan { components }); + sink.configure(&config); + sink.emit(&[SpanOp::Merge(SpanRow { + span_id: "legacy-root".into(), + root_span_id: "internal-root".into(), + parent_span_ids: vec!["external-parent".into()], + output: Some(json!("late output")), + end_ms: Some(2_000), + ..Default::default() + })]) + .await + .unwrap(); + sink.flush().await.unwrap(); + + let rows = logs3_rows(&server).await; + let update = rows + .iter() + .find(|row| row["span_id"] == "legacy-root") + .unwrap(); + assert_eq!(update["_is_merge"], true); + assert_eq!(update["output"], "late output"); + assert_eq!(update["metrics"]["end"], 2.0); + assert_eq!(update["experiment_id"], "exp-parent"); + assert_eq!(update["root_span_id"], "external-root"); + assert_eq!(update["span_parents"], json!(["external-parent"])); + assert_eq!(update["tenant"], "acme"); + assert_eq!(update["context"], json!({"request_id": "request-1"})); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -650,14 +714,13 @@ async fn braintrust_sink_delivers_spans_to_collector() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: Some(base.clone()), app_url: Some(base.clone()), - version: "test".into(), }); - let mut sink = factory.create("sess-1", "codex", Some("0.9.0")).unwrap(); + let mut sink = factory.create("sess-1", "codex").unwrap(); sink.configure(&session_config(&base)); // A session root (task) and a child tool span under it. - let root = row( + let mut root = row( "rootspan1", "rootspan1", &[], @@ -666,6 +729,7 @@ async fn braintrust_sink_delivers_spans_to_collector() { 1000, None, ); + root.origin.as_mut().unwrap().plugin_version = Some("0.9.0".into()); let tool = row( "toolspan1", "rootspan1", @@ -709,18 +773,31 @@ async fn braintrust_sink_delivers_spans_to_collector() { ); assert!(bodies.contains("codex: sess-1"), "root span name missing"); assert!(bodies.contains("\"command\""), "tool input missing"); - assert!( - bodies.contains("braintrust.plugin.codex"), - "shared span origin name missing" - ); - assert!( - bodies.contains("0.9.0"), - "plugin version missing from shared span origin" - ); - assert!( - !bodies.contains("bt_daemon_version"), - "daemon version should not clutter span metadata: {bodies}" - ); + for span_id in ["rootspan1", "toolspan1"] { + let rows: Vec<_> = logs3_rows(&server) + .await + .into_iter() + .filter(|row| row["span_id"] == span_id) + .collect(); + assert!(!rows.is_empty(), "{span_id}: no exported rows"); + for row in rows { + let origin = &row["context"]["span_origin"]; + if span_id == "rootspan1" { + assert_eq!(origin["name"], "braintrust.plugin.codex"); + assert_eq!(origin.get("version"), Some(&json!("0.9.0"))); + assert_eq!(origin["bt"]["version"], "test"); + assert_eq!(origin["codex"].get("version"), Some(&Value::Null)); + assert!(origin.get("agent").is_none()); + assert_eq!(origin["instrumentation"]["name"], "braintrust-plugin"); + } else { + assert_eq!(origin["name"], "braintrust.sdk.rust"); + assert_ne!(origin.get("version"), Some(&json!("0.9.0"))); + assert!(origin.get("bt").is_none()); + assert_eq!(origin["instrumentation"]["name"], "braintrust-rust-sdk"); + } + assert!(row["metadata"].get("bt_daemon_version").is_none()); + } + } // Project registration happened (org_name path, no login). assert!( @@ -743,9 +820,8 @@ async fn experiment_sessions_use_experiment_object_type_and_id() { let factory = BraintrustSinkFactory::new(BraintrustSinkConfig { api_url: Some(base.clone()), app_url: Some(base.clone()), - version: "test".into(), }); - let mut sink = factory.create("sess-exp", "claude-code", None).unwrap(); + let mut sink = factory.create("sess-exp", "claude-code").unwrap(); let mut config = session_config(&base); config.destination = Some(TraceDestination::Experiment { experiment_id: "exp-42".into(), @@ -787,3 +863,91 @@ async fn experiment_sessions_use_experiment_object_type_and_id() { "experiment spans were routed as project logs: {bodies}" ); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn root_origin_survives_open_closed_and_recovered_merges() { + let server = mock_backend().await; + let base = server.uri(); + let factory = BraintrustSinkFactory::new(BraintrustSinkConfig::default()); + for (source, plugin_version) in [ + ("codex", Some("1.2.3")), + ("claude-code", Some("2.3.4")), + ("grok", Some("3.4.5")), + ("opencode", Some("4.5.6")), + ("pi", Some("5.6.7")), + ("cursor", Some("6.7.8")), + ("antigravity", None), + ("codex", None), + ] { + let session_id = format!("root-{source}-{}", plugin_version.unwrap_or("unknown")); + let source_version = plugin_version.map(|_| "native-version"); + let origin = OriginSnapshot { + plugin_version: plugin_version.map(str::to_string), + bt_version: "birth-bt".into(), + source_version: source_version.map(str::to_string), + }; + let mut root = row( + &session_id, + &session_id, + &[], + "session root", + SpanType::Task, + 1, + None, + ); + root.origin = Some(origin.clone()); + let mut sink = factory.create(&session_id, source).unwrap(); + sink.configure(&session_config(&base)); + sink.emit(&[SpanOp::Insert(root)]).await.unwrap(); + sink.flush().await.unwrap(); + for (status, end_ms) in [("open", None), ("closed", Some(2)), ("late", None)] { + sink.emit(&[SpanOp::Merge(SpanRow { + span_id: session_id.clone(), + root_span_id: session_id.clone(), + origin: Some(origin.clone()), + output: Some(json!({"status": status})), + end_ms, + ..Default::default() + })]) + .await + .unwrap(); + sink.flush().await.unwrap(); + } + drop(sink); + let recovery_factory = BraintrustSinkFactory::new(BraintrustSinkConfig::default()); + let mut recovered = recovery_factory.create(&session_id, source).unwrap(); + recovered.configure(&session_config(&base)); + recovered + .emit(&[SpanOp::Merge(SpanRow { + span_id: session_id.clone(), + root_span_id: session_id.clone(), + origin: Some(origin), + output: Some(json!({"status": "recovered"})), + ..Default::default() + })]) + .await + .unwrap(); + recovered.flush().await.unwrap(); + + let rows: Vec<_> = logs3_rows(&server) + .await + .into_iter() + .filter(|row| row["span_id"] == session_id) + .collect(); + for status in ["open", "closed", "late", "recovered"] { + assert!( + rows.iter().any(|row| row["output"]["status"] == status), + "{session_id}: missing {status} update: {rows:?}" + ); + } + for row in rows { + let origin = &row["context"]["span_origin"]; + assert_eq!(origin["name"], format!("braintrust.plugin.{source}")); + assert_eq!(origin.get("version"), Some(&json!(plugin_version))); + assert_eq!(origin["bt"]["version"], "birth-bt"); + assert_eq!(origin[source].get("version"), Some(&json!(source_version))); + assert!(origin.get("agent").is_none()); + assert_eq!(origin["instrumentation"]["name"], "braintrust-plugin"); + } + } +} diff --git a/bt-daemon/tests/cursor_pipeline.rs b/bt-daemon/tests/cursor_pipeline.rs index dacbf00e..aafdba4f 100644 --- a/bt-daemon/tests/cursor_pipeline.rs +++ b/bt-daemon/tests/cursor_pipeline.rs @@ -381,7 +381,7 @@ impl Sink for BufferedSink { } impl SinkFactory for BufferedSink { - fn create(&self, _: &str, _: &str, _: Option<&str>) -> anyhow::Result> { + fn create(&self, _: &str, _: &str) -> anyhow::Result> { Ok(Box::new(Self(self.0.clone()))) } } diff --git a/bt-daemon/tests/distributed_tracing.rs b/bt-daemon/tests/distributed_tracing.rs index 10592a8c..3a31ffe4 100644 --- a/bt-daemon/tests/distributed_tracing.rs +++ b/bt-daemon/tests/distributed_tracing.rs @@ -47,12 +47,7 @@ impl RecordingSinkFactory { } impl SinkFactory for RecordingSinkFactory { - fn create( - &self, - session_id: &str, - source: &str, - _plugin_version: Option<&str>, - ) -> anyhow::Result> { + fn create(&self, session_id: &str, source: &str) -> anyhow::Result> { let record = Arc::new(RecordedSession { source: source.into(), ..RecordedSession::default() diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index c9284074..e1bb3317 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -32,12 +32,7 @@ struct TrackingSinkFactory { } impl SinkFactory for TrackingSinkFactory { - fn create( - &self, - session_id: &str, - _source: &str, - _plugin_version: Option<&str>, - ) -> anyhow::Result> { + fn create(&self, session_id: &str, _source: &str) -> anyhow::Result> { Ok(Box::new(TrackingSink { session_id: session_id.to_string(), flushes: self.flushes.clone(), @@ -67,7 +62,7 @@ impl Sink for SlowSink { struct SlowSinkFactory; impl SinkFactory for SlowSinkFactory { - fn create(&self, _: &str, _: &str, _: Option<&str>) -> anyhow::Result> { + fn create(&self, _: &str, _: &str) -> anyhow::Result> { Ok(Box::new(SlowSink)) } } @@ -104,7 +99,7 @@ impl Sink for GateSink { } impl SinkFactory for GateSinkFactory { - fn create(&self, _: &str, _: &str, _: Option<&str>) -> anyhow::Result> { + fn create(&self, _: &str, _: &str) -> anyhow::Result> { Ok(Box::new(GateSink { blocked: self.blocked.clone(), gate: self.gate.clone(), @@ -149,12 +144,7 @@ struct RouteRecordingSinkFactory { } impl SinkFactory for RouteRecordingSinkFactory { - fn create( - &self, - _session_id: &str, - _source: &str, - _plugin_version: Option<&str>, - ) -> anyhow::Result> { + fn create(&self, _session_id: &str, _source: &str) -> anyhow::Result> { let record = Arc::new(RouteSinkRecord::default()); self.sinks.lock().unwrap().push(record.clone()); Ok(Box::new(RouteRecordingSink { record })) @@ -2563,31 +2553,17 @@ esac handle.await.unwrap(); } -#[cfg(all(feature = "cli", unix))] -type CreatedDebugSinks = Arc)>>>; - #[cfg(all(feature = "cli", unix))] struct FlushTrackingDebugSinkFactory { inner: DebugSinkFactory, flushes: Arc>>, - created: CreatedDebugSinks, } #[cfg(all(feature = "cli", unix))] impl SinkFactory for FlushTrackingDebugSinkFactory { - fn create( - &self, - session_id: &str, - source: &str, - plugin_version: Option<&str>, - ) -> anyhow::Result> { - self.created.lock().unwrap().push(( - session_id.to_string(), - source.to_string(), - plugin_version.map(str::to_string), - )); + fn create(&self, session_id: &str, source: &str) -> anyhow::Result> { Ok(Box::new(FlushTrackingDebugSink { - inner: self.inner.create(session_id, source, plugin_version)?, + inner: self.inner.create(session_id, source)?, session_id: session_id.to_string(), flushes: self.flushes.clone(), })) @@ -2629,7 +2605,6 @@ async fn start_grok_debug_daemon_at( data_dir: PathBuf, socket: PathBuf, flushes: Arc>>, - created: CreatedDebugSinks, ) -> tokio::task::JoinHandle<()> { let args = ServeArgs { socket: Some(socket.clone()), @@ -2645,7 +2620,6 @@ async fn start_grok_debug_daemon_at( dir: data_dir.join("spans"), }, flushes, - created, }), auth_provider: Some(Arc::new(TestAuthProvider { calls: Mutex::new(Vec::new()), @@ -2847,14 +2821,7 @@ async fn packaged_grok_hook_replays_bounded_transcripts_to_isolated_debug_routes }); let flushes = Arc::new(Mutex::new(HashMap::new())); - let created = Arc::new(Mutex::new(Vec::new())); - let first = start_grok_debug_daemon_at( - data_dir.clone(), - socket.clone(), - flushes.clone(), - created.clone(), - ) - .await; + let first = start_grok_debug_daemon_at(data_dir.clone(), socket.clone(), flushes.clone()).await; invoke_grok_hook( &socket, plugin_version, @@ -2958,13 +2925,8 @@ async fn packaged_grok_hook_replays_bounded_transcripts_to_isolated_debug_routes std::fs::write(&primary_updates, &updates).unwrap(); std::fs::write(&primary_events, &events).unwrap(); - let second = start_grok_debug_daemon_at( - data_dir.clone(), - socket.clone(), - flushes.clone(), - created.clone(), - ) - .await; + let second = + start_grok_debug_daemon_at(data_dir.clone(), socket.clone(), flushes.clone()).await; let primary_end = serde_json::json!({ "sessionId": "grok-primary", "hookEventName": "session_end", @@ -3229,12 +3191,6 @@ async fn packaged_grok_hook_replays_bounded_transcripts_to_isolated_debug_routes assert!(primary_text.contains("primary-route")); assert!(!primary_text.contains("decoy-route")); - let created = created.lock().unwrap().clone(); - assert_eq!(created.len(), 3); - assert!(created.iter().all(|(_, source, version)| { - source == "grok" && version.as_deref() == Some(plugin_version) - })); - shutdown(&socket).await; second.await.unwrap(); } diff --git a/bt-daemon/tests/root_provenance.rs b/bt-daemon/tests/root_provenance.rs new file mode 100644 index 00000000..e24c09b0 --- /dev/null +++ b/bt-daemon/tests/root_provenance.rs @@ -0,0 +1,764 @@ +//! Consumer-visible provenance through capture, journal replay, dispatch, and HTTP ingest. + +use async_trait::async_trait; +use braintrust_sdk_rust::{SpanComponents, SpanObjectType}; +use bt_daemon::wire::{AuthSelection, BackendAuth, Envelope, SessionRoute, TraceDestination}; +use bt_daemon::{ + flush_session, forward_envelope, run_serve, run_status, shutdown_daemon, source_journal_path, + AuthLease, AuthProvider, AuthResolveReason, BraintrustSinkConfig, BraintrustSinkFactory, + HostInfo, Registry, ServeArgs, ServeOptions, StatusArgs, +}; +use serde_json::{json, Value}; +use std::ffi::OsString; +use std::io::Write; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +struct BackendAuthProvider(String); + +#[async_trait] +impl AuthProvider for BackendAuthProvider { + async fn resolve( + &self, + selection: &AuthSelection, + _reason: AuthResolveReason, + ) -> anyhow::Result { + Ok(AuthLease { + selection: selection.clone().canonicalized()?, + auth: BackendAuth { + token: "sk-test".into(), + api_url: Some(self.0.clone()), + app_url: Some(self.0.clone()), + org_name: Some("acme".into()), + org_id: Some("org-1".into()), + }, + expires_at_ms: None, + }) + } +} + +async fn mock_backend() -> MockServer { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/version")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/api/project/register")) + .respond_with( + ResponseTemplate::new(200).set_body_json(json!({ "project": { "id": "proj-1" } })), + ) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/logs3")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) + .mount(&server) + .await; + server +} + +fn test_endpoint(tmp: &Path) -> PathBuf { + #[cfg(unix)] + { + tmp.join("d.sock") + } + #[cfg(windows)] + { + let _ = tmp; + PathBuf::from(format!(r"\\.\pipe\bt-root-origin-{}", uuid::Uuid::new_v4())) + } +} + +struct Daemon { + socket: PathBuf, + host: HostInfo, + task: tokio::task::JoinHandle>, +} + +impl Daemon { + async fn start(tmp: &Path, server: &MockServer, version: &str) -> Self { + let socket = test_endpoint(tmp); + let args = ServeArgs { + socket: Some(socket.clone()), + data_dir: Some(tmp.join("data")), + idle_timeout_secs: 0, + session_idle_timeout_secs: 0, + }; + let opts = ServeOptions { + version: version.into(), + translators: Arc::new(Registry::default_agents()), + sink_factory: Arc::new(BraintrustSinkFactory::new(BraintrustSinkConfig { + api_url: Some(server.uri()), + app_url: Some(server.uri()), + })), + auth_provider: Some(Arc::new(BackendAuthProvider(server.uri()))), + }; + let task = tokio::spawn(run_serve(args, opts)); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if let Ok(Some(_)) = run_status(StatusArgs { + socket: Some(socket.clone()), + session_id: None, + }) + .await + { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("daemon did not become ready"); + Self { + socket, + host: HostInfo { + serve_argv: vec![OsString::from("unused")], + version: version.into(), + }, + task, + } + } + + async fn send(&self, event: &Envelope) { + forward_envelope(event, &self.socket, &self.host, true) + .await + .unwrap(); + } + + async fn flush(&self, session: &str) { + let result = flush_session(session, &self.socket, 5_000).await.unwrap(); + assert!(result.flushed, "delivery did not drain: {result:?}"); + assert_eq!(result.pending, 0, "pending delivery: {result:?}"); + } + + async fn stop(self) { + shutdown_daemon(&self.socket).await.unwrap(); + self.task.await.unwrap().unwrap(); + } +} + +fn event( + session: &str, + source: &str, + name: &str, + version: Option<&str>, + source_version: Option<&str>, + ts_ms: i64, +) -> Envelope { + Envelope { + source: source.into(), + source_version: source_version.map(str::to_owned), + plugin_version: version.map(str::to_owned), + session_id: session.into(), + event: name.into(), + ts_ms, + managed_run_id: None, + payload: json!({ + "session_id": session, + "hook_event_name": name, + "prompt": format!("prompt at {ts_ms}"), + "prompt_id": format!("prompt-{ts_ms}"), + }), + route: Some(SessionRoute { + destination: Some(TraceDestination::ProjectLogs { + project_id: Some("proj-1".into()), + project_name: None, + }), + ..SessionRoute::default() + }), + config: None, + capture: None, + } +} + +async fn logs3_rows(server: &MockServer) -> Vec { + server + .received_requests() + .await + .unwrap() + .iter() + .filter(|request| request.url.path() == "/logs3") + .flat_map(|request| { + serde_json::from_slice::(&request.body).unwrap()["rows"] + .as_array() + .expect("logs3 request must contain rows") + .clone() + }) + .collect() +} + +fn root_id(rows: &[Value], session: &str) -> String { + rows.iter() + .find(|row| row["metadata"]["session_id"] == session) + .expect("session root was not ingested")["span_id"] + .as_str() + .unwrap() + .to_owned() +} + +fn assert_provenance( + rows: &[Value], + span: &str, + source: &str, + plugin_version: Option<&str>, + source_version: Option<&str>, + bt_version: &str, +) { + let spans: Vec<_> = rows.iter().filter(|row| row["span_id"] == span).collect(); + assert!( + !spans.is_empty(), + "expected ingested insert/update for {span}" + ); + for row in spans { + let origin = &row["context"]["span_origin"]; + assert_eq!(origin["name"], format!("braintrust.plugin.{source}")); + // get() distinguishes explicit null from a missing version field. + assert_eq!(origin.get("version"), Some(&json!(plugin_version))); + assert_eq!(origin[source].get("version"), Some(&json!(source_version))); + assert!(origin.get("agent").is_none(), "generic agent key: {row}"); + assert_eq!(origin["bt"]["version"], bt_version); + assert_eq!(origin["instrumentation"]["name"], "braintrust-plugin"); + assert!(row["metadata"].get("bt_daemon_version").is_none()); + } +} + +fn turn_id(rows: &[Value], prompt_ts: i64) -> String { + rows.iter() + .find(|row| row["input"] == format!("prompt at {prompt_ts}")) + .expect("turn was not ingested")["span_id"] + .as_str() + .unwrap() + .to_owned() +} + +fn assert_non_turn_provenance(rows: &[Value], root: &str, turns: &[&str]) { + let children: Vec<_> = rows + .iter() + .filter(|row| { + row["root_span_id"] == root + && row["span_id"] != root + && !turns.iter().any(|turn| row["span_id"] == *turn) + }) + .collect(); + assert!(!children.is_empty(), "expected ingested non-turn work"); + for row in children { + if let Some(origin) = row["context"].get("span_origin") { + assert_eq!(origin["name"], "braintrust.sdk.rust"); + assert!(origin.get("bt").is_none()); + assert!(origin.get("claude-code").is_none()); + assert!(origin.get("agent").is_none()); + assert_ne!(origin["instrumentation"]["name"], "braintrust-plugin"); + } else { + assert_eq!(row["_is_merge"], true, "insert omitted SDK origin: {row}"); + } + assert!(row["metadata"].get("bt_daemon_version").is_none()); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn root_and_turn_birth_snapshots_survive_version_changes_and_daemon_restart() { + let server = mock_backend().await; + let tmp = tempfile::tempdir().unwrap(); + let plugin = tmp.path().join("origin-plugin.mjs"); + std::fs::write( + &plugin, + "export default span => ({...span, metadata: {...span.metadata, origin_plugin: true}})", + ) + .unwrap(); + let capture = |session, name, plugin_version, source_version, ts_ms| { + let mut envelope = event( + session, + "claude-code", + name, + plugin_version, + source_version, + ts_ms, + ); + envelope.route.as_mut().unwrap().span_plugins = vec![plugin.clone()]; + envelope + }; + let first = Daemon::start(tmp.path(), &server, "bt1").await; + let mut sessions = Vec::new(); + for (session, birth_plugin, birth_source) in [ + ("known-birth", Some("v1"), Some("native1")), + ("unknown-birth", None, None), + ] { + first + .send(&capture( + session, + "UserPromptSubmit", + birth_plugin, + birth_source, + 1_000, + )) + .await; + first.flush(session).await; + let rows = logs3_rows(&server).await; + let root = root_id(&rows, session); + let session_rows: Vec<_> = rows + .into_iter() + .filter(|row| row["root_span_id"] == root) + .collect(); + let turn1 = turn_id(&session_rows, 1_000); + assert_provenance( + &session_rows, + &root, + "claude-code", + birth_plugin, + birth_source, + "bt1", + ); + assert_provenance( + &session_rows, + &turn1, + "claude-code", + birth_plugin, + birth_source, + "bt1", + ); + + // Opening a new turn also updates the previous turn. Neither unknown + // nor known birth versions may be overwritten by this newer capture. + first + .send(&capture( + session, + "UserPromptSubmit", + Some("v2"), + Some("native2"), + 2_000, + )) + .await; + let mut tool = capture(session, "PreToolUse", Some("v2"), Some("native2"), 2_100); + tool.payload["tool_use_id"] = json!("tool-1"); + tool.payload["tool_name"] = json!("Read"); + tool.payload["tool_input"] = json!({"file_path": "/tmp/example"}); + first.send(&tool).await; + let mut completed = tool; + completed.event = "PostToolUse".into(); + completed.payload["hook_event_name"] = json!("PostToolUse"); + completed.payload["tool_response"] = json!("file contents"); + completed.plugin_version = Some("v3".into()); + completed.source_version = Some("native3".into()); + completed.ts_ms = 2_200; + first.send(&completed).await; + first.flush(session).await; + let rows = logs3_rows(&server).await; + let session_rows: Vec<_> = rows + .into_iter() + .filter(|row| row["root_span_id"] == root) + .collect(); + let turn2 = turn_id(&session_rows, 2_000); + assert_ne!(turn1, turn2); + assert_provenance( + &session_rows, + &root, + "claude-code", + birth_plugin, + birth_source, + "bt1", + ); + assert_provenance( + &session_rows, + &turn1, + "claude-code", + birth_plugin, + birth_source, + "bt1", + ); + assert_provenance( + &session_rows, + &turn2, + "claude-code", + Some("v2"), + Some("native2"), + "bt1", + ); + assert_non_turn_provenance(&session_rows, &root, &[&turn1, &turn2]); + assert!(session_rows + .iter() + .any(|row| { row["span_id"] == turn2 && row["metadata"]["origin_plugin"] == true })); + sessions.push((session, birth_plugin, birth_source, root, turn1, turn2)); + } + first.stop().await; + + let second = Daemon::start(tmp.path(), &server, "bt2").await; + for (session, birth_plugin, birth_source, root, turn1, turn2) in sessions { + let before = logs3_rows(&server).await.len(); + second + .send(&capture( + session, + "UserPromptSubmit", + Some("v3"), + Some("native3"), + 3_000, + )) + .await; + second + .send(&capture( + session, + "SessionEnd", + Some("v4"), + Some("native4"), + 4_000, + )) + .await; + second.flush(session).await; + let rows = logs3_rows(&server).await; + let turn3 = turn_id(&rows[before..], 3_000); + assert_ne!(turn3, turn1); + assert_ne!(turn3, turn2); + // Graceful shutdown completed these spans; recovery must not redeliver them. + assert!(rows[before..].iter().all(|row| { + row["span_id"] != root && row["span_id"] != turn1 && row["span_id"] != turn2 + })); + assert_provenance( + &rows[before..], + &turn3, + "claude-code", + Some("v3"), + Some("native3"), + "bt2", + ); + // Check every exported update, including any replayed history, rather + // than just the final merged row visible to a consumer. + assert_provenance( + &rows, + &root, + "claude-code", + birth_plugin, + birth_source, + "bt1", + ); + assert_provenance( + &rows, + &turn1, + "claude-code", + birth_plugin, + birth_source, + "bt1", + ); + assert_provenance( + &rows, + &turn2, + "claude-code", + Some("v2"), + Some("native2"), + "bt1", + ); + assert_non_turn_provenance(&rows, &root, &[&turn1, &turn2, &turn3]); + assert!(rows[before..] + .iter() + .any(|row| { row["span_id"] == turn3 && row["metadata"]["origin_plugin"] == true })); + } + second.stop().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn legacy_acknowledged_spans_receive_late_updates_without_inventing_origins() { + let server = mock_backend().await; + let tmp = tempfile::tempdir().unwrap(); + let session = "legacy-origin"; + let capture = |name, plugin_version, source_version, ts_ms| { + event( + session, + "claude-code", + name, + Some(plugin_version), + Some(source_version), + ts_ms, + ) + }; + let first = Daemon::start(tmp.path(), &server, "bt1").await; + first + .send(&capture("UserPromptSubmit", "v1", "native1", 1_000)) + .await; + first.flush(session).await; + // Shutdown completes the spans, but the acknowledged journal still has an + // active turn. A later Stop must deliver new output, not replay old inserts. + first.stop().await; + let initial_rows = logs3_rows(&server).await; + let root = root_id(&initial_rows, session); + let old_turn = turn_id(&initial_rows, 1_000); + for span in [&root, &old_turn] { + assert_provenance( + &initial_rows, + span, + "claude-code", + Some("v1"), + Some("native1"), + "bt1", + ); + assert!(initial_rows + .iter() + .any(|row| row["span_id"] == *span && row["metrics"]["end"] == 1.0)); + } + + let data_dir = tmp.path().join("data"); + let journal = source_journal_path(&data_dir, "claude-code", session); + let acknowledged_history = std::fs::read(&journal).unwrap(); + assert!(std::str::from_utf8(&acknowledged_history) + .unwrap() + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .any(|record| { + record["_bt_record_type"] == "delivery_checkpoint" + && record["through"].as_u64().is_some_and(|offset| offset > 0) + })); + let ledger_paths: Vec<_> = std::fs::read_dir(data_dir.join("delivery-ledger")) + .unwrap() + .map(|entry| entry.unwrap().path()) + .collect(); + assert_eq!(ledger_paths.len(), 1); + let ledger_path = &ledger_paths[0]; + let mut ledger: Value = serde_json::from_slice(&std::fs::read(ledger_path).unwrap()).unwrap(); + let completed = ledger["completed_span_ids"].clone(); + for span in [&root, &old_turn] { + assert!(completed.as_array().unwrap().contains(&json!(span))); + assert!(ledger["span_origins"].get(span.as_str()).is_some()); + } + // Simulate the old schema, preserving all completion/late-merge IDs and + // journal checkpoints. Missing provenance is not a snapshot of null versions. + let ledger_object = ledger.as_object_mut().unwrap(); + ledger_object.remove("span_origins"); + ledger_object.remove("root_origins"); + assert_eq!(ledger["completed_span_ids"], completed); + assert!(ledger.get("span_origins").is_none()); + assert!(ledger.get("root_origins").is_none()); + std::fs::write(ledger_path, serde_json::to_vec(&ledger).unwrap()).unwrap(); + assert_eq!(std::fs::read(&journal).unwrap(), acknowledged_history); + + let second = Daemon::start(tmp.path(), &server, "bt2").await; + let mut late_stop = capture("Stop", "v2", "native2", 2_000); + late_stop.payload["last_assistant_message"] = json!("late legacy answer"); + second.send(&late_stop).await; + second.flush(session).await; + let rows = logs3_rows(&server).await; + let late_rows = &rows[initial_rows.len()..]; + for span in [&root, &old_turn] { + let updates: Vec<_> = late_rows + .iter() + .filter(|row| row["span_id"] == *span) + .collect(); + assert!(!updates.is_empty(), "missing late legacy update for {span}"); + assert!(updates.iter().any(|row| row["metrics"]["end"] == 2.0)); + for row in updates { + assert_eq!(row["_is_merge"], true, "historical insert replayed: {row}"); + assert!( + row["context"].get("span_origin").is_none(), + "legacy update must omit provenance entirely, not set null: {row}" + ); + } + } + assert!(late_rows + .iter() + .any(|row| { row["span_id"] == old_turn && row["output"] == "late legacy answer" })); + + second + .send(&capture("UserPromptSubmit", "v3", "native3", 3_000)) + .await; + let mut stop = capture("Stop", "v4", "native4", 4_000); + stop.payload["last_assistant_message"] = json!("fresh answer"); + second.send(&stop).await; + second.flush(session).await; + second.stop().await; + let rows = logs3_rows(&server).await; + let fresh_turn = turn_id(&rows[initial_rows.len()..], 3_000); + assert_ne!(fresh_turn, old_turn); + assert_provenance( + &rows, + &fresh_turn, + "claude-code", + Some("v3"), + Some("native3"), + "bt2", + ); + assert!(rows.iter().any(|row| { + row["span_id"] == fresh_turn + && row["output"] == "fresh answer" + && row["metrics"]["end"] == 4.0 + })); + + // Repeated recovery must not turn historical capture versions into a + // fabricated birth snapshot, or restore SDK provenance on the next merge. + let before_restart = rows.len(); + let third = Daemon::start(tmp.path(), &server, "bt3").await; + third + .send(&capture("SessionEnd", "v5", "native5", 5_000)) + .await; + third.flush(session).await; + third.stop().await; + let rows = logs3_rows(&server).await; + assert!(rows[before_restart..] + .iter() + .any(|row| row["span_id"] == root && row["metrics"]["end"] == 5.0)); + for row in rows[initial_rows.len()..] + .iter() + .filter(|row| row["span_id"] == root || row["span_id"] == old_turn) + { + assert_eq!(row["_is_merge"], true); + assert!( + row["context"].get("span_origin").is_none(), + "legacy origin was invented after recovery: {row}" + ); + } + let ledger: Value = serde_json::from_slice(&std::fs::read(ledger_path).unwrap()).unwrap(); + for key in ["span_origins", "root_origins"] { + for span in [&root, &old_turn] { + assert!( + ledger[key].get(span.as_str()).is_none(), + "legacy birth snapshot was invented in {key}: {ledger}" + ); + } + } + assert!(ledger["span_origins"].get(fresh_turn.as_str()).is_some()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn newer_capture_replays_unacknowledged_root_with_its_journaled_plugin_version() { + let server = mock_backend().await; + let tmp = tempfile::tempdir().unwrap(); + let daemon = Daemon::start(tmp.path(), &server, "bt1").await; + let session = "unacknowledged-history"; + let birth = event( + session, + "claude-code", + "UserPromptSubmit", + Some("v1"), + Some("native1"), + 1_000, + ); + // Seed the capture-before-dispatch crash window after startup recovery has + // finished. The newer live event, not startup, must trigger history replay. + let journal = source_journal_path(&tmp.path().join("data"), "claude-code", session); + std::fs::create_dir_all(journal.parent().unwrap()).unwrap(); + let mut raw = serde_json::to_vec(&birth.redacted()).unwrap(); + raw.push(b'\n'); + std::fs::write(journal, raw).unwrap(); + daemon + .send(&event( + session, + "claude-code", + "SessionEnd", + Some("v2"), + Some("native2"), + 2_000, + )) + .await; + daemon.flush(session).await; + let rows = logs3_rows(&server).await; + let root = root_id(&rows, session); + assert_provenance( + &rows, + &root, + "claude-code", + Some("v1"), + Some("native1"), + "bt1", + ); + let turn = turn_id(&rows, 1_000); + assert_provenance( + &rows, + &turn, + "claude-code", + Some("v1"), + Some("native1"), + "bt1", + ); + assert!(rows.iter().any(|row| row["input"] == "prompt at 1000")); + daemon.stop().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn attached_provisional_root_keeps_birth_snapshot_until_work_after_restart() { + let server = mock_backend().await; + let tmp = tempfile::tempdir().unwrap(); + let transcript = tmp.path().join("codex.jsonl"); + let session = "attached-provisional"; + let metadata = json!({ + "timestamp": "2026-01-01T00:00:01Z", + "type": "session_meta", + "payload": { "id": session, "cli_version": "native-codex-version" } + }); + std::fs::write(&transcript, format!("{metadata}\n")).unwrap(); + let mut components = SpanComponents::new(SpanObjectType::ProjectLogs); + components.object_id = Some("proj-1".into()); + components.span_id = Some("external-parent".into()); + components.root_span_id = Some("external-root".into()); + let mut birth = event( + session, + "codex", + "SessionStart", + Some("v1"), + Some("native1"), + 1_767_225_601_000, + ); + birth.payload = json!({ + "session_id": session, + "hook_event_name": "SessionStart", + "transcript_path": transcript, + "source": "startup" + }); + birth.route.as_mut().unwrap().destination = Some(TraceDestination::ParentSpan { components }); + + let first = Daemon::start(tmp.path(), &server, "bt1").await; + first.send(&birth).await; + first.flush(session).await; + assert!( + logs3_rows(&server).await.is_empty(), + "provisional root exported before work" + ); + first.stop().await; + assert!( + logs3_rows(&server).await.is_empty(), + "shutdown exported provisional root" + ); + + let second = Daemon::start(tmp.path(), &server, "bt2").await; + let work = json!({ + "timestamp": "2026-01-01T00:00:02Z", + "type": "event_msg", + "payload": { "type": "task_started", "turn_id": "turn-after-upgrade" } + }); + writeln!( + std::fs::OpenOptions::new() + .append(true) + .open(&transcript) + .unwrap(), + "{work}" + ) + .unwrap(); + let mut capture = birth; + capture.plugin_version = Some("v2".into()); + capture.source_version = Some("native2".into()); + capture.event = "Stop".into(); + capture.ts_ms += 1_000; + capture.payload["hook_event_name"] = json!("Stop"); + second.send(&capture).await; + second.flush(session).await; + let rows = logs3_rows(&server).await; + let root = root_id(&rows, session); + assert_ne!(root, "external-parent"); + assert_ne!(root, "external-root"); + assert_provenance(&rows, &root, "codex", Some("v1"), Some("native1"), "bt1"); + let turn = rows + .iter() + .find(|row| row["span_id"] != root) + .expect("turn after restart was not ingested")["span_id"] + .as_str() + .unwrap(); + assert_provenance(&rows, turn, "codex", Some("v2"), Some("native2"), "bt2"); + for row in &rows { + assert_eq!(row["root_span_id"], "external-root"); + if row["span_id"] == root { + assert_eq!(row["span_parents"], json!(["external-parent"])); + } else { + assert_eq!(row["span_parents"], json!([root])); + } + } + second.stop().await; +} diff --git a/docs/coding-agent-tracing-plugin-spec.md b/docs/coding-agent-tracing-plugin-spec.md index 8e385086..14501afa 100644 --- a/docs/coding-agent-tracing-plugin-spec.md +++ b/docs/coding-agent-tracing-plugin-spec.md @@ -255,6 +255,77 @@ A process shutdown/session-end event closes the current process lifetime. Since agents can resume sessions, the translator MUST remain replay-safe if later activity reuses the native session identifier. +#### 6.2.1 Span-origin versions + +The shared delivery path MUST snapshot `context.span_origin` independently when +a coding-agent session root or turn span is first created. This is span-creation +provenance, not a record of every version used within a turn. Other descendants +do not receive these snapshots; ordinary SDK provenance is independent of this +contract. Translators identify turns explicitly, not by task type or display name. + +| Field | Required value | +|-------|----------------| +| `name` | Stable plugin identity: `braintrust.plugin.`, using the canonical source. | +| `version` | Hooks/plugin package version from the event that creates the span, or explicit JSON `null` when unknown. | +| `bt.version` | Version string of the `bt` daemon that first creates the span snapshot. | +| `.version` | Coding-agent version from that event's `source_version`, or explicit JSON `null` when unknown. The key is the canonical source (`codex`, `claude-code`, `antigravity`, `cursor`, `grok`, `opencode`, or `pi`), not `agent`. | + +For example: + +```json +{ + "context": { + "span_origin": { + "name": "braintrust.plugin.codex", + "version": "2.0.2", + "bt": { "version": "0.20.0" }, + "codex": { "version": "0.123.0" }, + "instrumentation": { "name": "braintrust-plugin" } + } + } +} +``` + +The plugin version MUST come from the capture package, not the coding agent, +`bt`, or the underlying SDK. Adapters MUST forward it when discoverable. If it +is unavailable (for example, a versionless Antigravity manifest or a transcript +import without capture provenance), the sink MUST emit explicit `null`; it +MUST NOT substitute another component's version or allow SDK defaults to do so. +This explicit-null rule overrides general guidance to omit unknown provenance. + +Each snapshot MUST remain unchanged on later events, span updates, plugin or +agent upgrades, daemon restarts, and resume. Updates MAY repeat the original +snapshot to prevent SDK defaults from overwriting it, but MUST NOT substitute +current versions. Unknown plugin and coding-agent versions MUST remain `null` +even if a later event supplies a version. Each new turn MUST capture its own +snapshot rather than inherit the session root's versions. + +Actor creation is not span creation. When journal replay creates a previously +unmaterialized root or turn, its plugin and agent versions MUST come from the +journaled event creating that span, not the newer event that triggered recovery. +If a snapshot already exists for that span and destination, replay MUST reuse +it, including its original `bt.version`. A new destination creates its own +snapshots using the original span-creating events and the daemon doing delivery. +Transcript backfills snapshot versions available at materialization; they MUST +NOT claim to reconstruct historical per-turn versions that were not captured. + +Legacy delivery ledgers may contain acknowledged spans without origin snapshots. +Updates to those spans MUST preserve their existing backend provenance: omit +`context.span_origin` rather than inject SDK defaults, copy an attached parent's +origin, write `null`, or invent historical versions from the current daemon. +Missing snapshots are distinct from snapshots containing unknown versions. +Updates MUST still deliver their output, metrics, and other context normally; +new turns capture fresh snapshots under the rules above. + +A turn later reclassified as compaction retains its creation snapshot. Tasks +already known to represent compaction or subagent containers are not turns. + +A provisional root MUST retain the snapshot taken when it was created, even if +delivery waits for a later descendant event. When tracing attaches to an +external parent, this contract applies to the coding-agent session root below +that parent and its turns; it MUST NOT modify the external root. Existing +instrumentation and environment provenance MUST be preserved. + ### 6.3 Turns Each user request and the agent work attributable to it MUST form one child @@ -511,7 +582,7 @@ features that require native evidence remain conditional under section 3. | 1. Foundation | ROUTE-01 | Project routing | A session can resolve and pin a Braintrust project destination without placing credentials in agent configuration. | | 2. Minimum trace | TRACE-01 | Deterministic session root | A replay-safe task span represents the session, preserves external attachment, and can be extended after process shutdown or resume. | | 2. Minimum trace | TRACE-02 | Per-turn task spans | Each user request has a child task span containing native input, final output, timing, and completion status when observable. | -| 2. Minimum trace | META-01 | Session, version, and workspace metadata | The root includes available native session ID, source and plugin versions, cwd, workspace, execution mode, and transcript provenance. | +| 2. Minimum trace | META-01 | Session, version, and workspace metadata | The root includes available native session ID, agent version, cwd, workspace, execution mode, and transcript provenance; roots and turns independently snapshot plugin, `bt`, and coding-agent versions per section 6.2.1. | | 2. Minimum trace | REL-02 | Deterministic IDs and duplicate handling | Stable native identities and deterministic fallbacks make duplicate delivery and journal replay converge on the same logical spans. | | 2. Minimum trace | REL-03 | Bounded correlation state | Prompt, operation, transcript, and deduplication state have explicit bounds without dropping currently open work silently. | | 2. Minimum trace | TRACE-14 | Concurrent activity correlation | Overlapping turns and operations correlate by native identity rather than a single implicit current slot; ambiguous ancestry fails safe. |