diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index ae371a72..9c586f5a 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -74,4 +74,5 @@ pub use trace_runtime::{ pub use transcript_import::{import_transcript, import_transcripts, run_import}; pub use translate::{ AgentTranslator, OriginSnapshot, Registry, SessionCtx, SpanOp, SpanRow, SpanType, + TURN_SPAN_ID_KEY, }; diff --git a/bt-daemon/src/translate/antigravity.rs b/bt-daemon/src/translate/antigravity.rs index 380a9217..d825480a 100644 --- a/bt-daemon/src/translate/antigravity.rs +++ b/bt-daemon/src/translate/antigravity.rs @@ -225,6 +225,7 @@ impl AntigravityTranslator { start_ms: Some(ts_ms), input: nonempty_value(input), metadata: Some(json!({"turn_number": self.turn_count})), + turn_span_id: Some(span_id.clone()), ..Default::default() })); self.turn = Some(Turn { @@ -275,6 +276,7 @@ impl AntigravityTranslator { span_id: span_id.clone(), root_span_id: self.root_span_id.clone(), parent_span_ids: vec![parent.clone()], + turn_span_id: Some(parent.clone()), name: model, span_type: SpanType::Llm, start_ms: Some(event.ts_ms), @@ -334,6 +336,7 @@ impl AntigravityTranslator { span_id: span_id.clone(), root_span_id: self.root_span_id.clone(), parent_span_ids: vec![parent.clone()], + turn_span_id: Some(parent.clone()), name: name.clone(), span_type: SpanType::Tool, start_ms: Some(event.ts_ms), @@ -394,6 +397,7 @@ impl AntigravityTranslator { span_id: span_id.clone(), root_span_id: self.root_span_id.clone(), parent_span_ids: vec![parent.clone()], + turn_span_id: Some(parent.clone()), name: name.clone(), span_type: SpanType::Tool, start_ms: Some(recovered_start_ms.unwrap_or(event.ts_ms)), diff --git a/bt-daemon/src/translate/claude.rs b/bt-daemon/src/translate/claude.rs index 09ec9095..9cdfe65a 100644 --- a/bt-daemon/src/translate/claude.rs +++ b/bt-daemon/src/translate/claude.rs @@ -264,6 +264,8 @@ struct ClaudeTranslator { continuation_count: u32, last_user_turn_id: Option, agent_origins: RecentMap, + /// User turn owning each turn or subagent span that can parent other spans. + span_owners: RecentMap, user_turn_ends: RecentMap, prompt_turns: RecentMap>, pending_prompts: VecDeque, @@ -309,6 +311,7 @@ impl ClaudeTranslator { continuation_count: 0, last_user_turn_id: None, agent_origins: RecentMap::default(), + span_owners: RecentMap::default(), user_turn_ends: RecentMap::default(), tool_seq: 0, prompt_turns: RecentMap::default(), @@ -515,6 +518,7 @@ impl ClaudeTranslator { metadata.insert("turn_trigger".into(), json!(trigger)); metadata.insert("task_id".into(), json!(task_id)); } + let owner = user_turn_id.clone().unwrap_or_else(|| id.clone()); ops.push(SpanOp::Insert(SpanRow { span_id: id.clone(), root_span_id: self.root_span_id.clone(), @@ -525,8 +529,10 @@ impl ClaudeTranslator { start_ms: Some(event.ts_ms), input: event.payload.get("prompt").cloned(), metadata: (!metadata.is_empty()).then_some(Value::Object(metadata)), + turn_span_id: Some(owner.clone()), ..Default::default() })); + self.span_owners.insert(id.clone(), owner); let turn_id = Arc::new(id.clone()); if let Some(prompt_id) = event.payload.get("prompt_id").and_then(Value::as_str) { self.prompt_turns @@ -643,6 +649,10 @@ impl ClaudeTranslator { self.turn.as_ref().map(|turn| turn.id.clone()) } + fn owner_of(&self, parent: &str) -> Option { + self.span_owners.get(parent).cloned() + } + fn ensure_subagent( &mut self, hook: &SubagentHook, @@ -670,6 +680,10 @@ impl ClaudeTranslator { .unwrap_or_else(|| self.last_user_turn_id.clone()), }); let parent_id = origin.parent_id.clone(); + let owner = origin + .user_turn_id + .clone() + .or_else(|| self.owner_of(&parent_id)); self.agent_origins.insert(hook.agent_id.clone(), origin); let agent_type = hook.agent_type.clone().unwrap_or_else(|| "agent".into()); let span_id = ids::span_id(&self.session_id, &format!("subagent:{}", hook.agent_id)); @@ -681,8 +695,12 @@ impl ClaudeTranslator { span_type: SpanType::Task, start_ms: Some(event.ts_ms), metadata: Some(json!({ "agent_id": hook.agent_id, "agent_type": agent_type })), + turn_span_id: owner.clone(), ..Default::default() })); + if let Some(owner) = owner { + self.span_owners.insert(span_id.clone(), owner); + } self.subagents.insert( hook.agent_id.clone(), Subagent { @@ -720,6 +738,7 @@ impl ClaudeTranslator { start_ms: Some(event.ts_ms), input: Some(input.clone()), metadata: Some(metadata.clone()), + turn_span_id: self.owner_of(&parent_id), ..Default::default() })); self.pending_tools.insert( @@ -772,6 +791,7 @@ impl ClaudeTranslator { ops.push(SpanOp::Insert(SpanRow { span_id: ids::span_id(&self.session_id, &format!("tool:{call_id}")), root_span_id: self.root_span_id.clone(), + turn_span_id: self.owner_of(&parent_id), parent_span_ids: vec![parent_id], name: tool_span_name(&tool_name, &input), span_type: SpanType::Tool, @@ -1039,11 +1059,13 @@ impl ClaudeTranslator { } else { parent }; - ops.push(SpanOp::Insert(call.into_row( + let mut row = call.into_row( ids::span_id(&self.session_id, &span_key), self.root_span_id.clone(), parent.to_string(), - ))); + ); + row.turn_span_id = self.owner_of(parent); + ops.push(SpanOp::Insert(row)); } } for tool in tools { @@ -1072,11 +1094,13 @@ impl ClaudeTranslator { } else { parent }; - ops.push(SpanOp::Insert(tool.into_row( + let mut row = tool.into_row( ids::span_id(&self.session_id, &span_key), self.root_span_id.clone(), parent.to_string(), - ))); + ); + row.turn_span_id = self.owner_of(parent); + ops.push(SpanOp::Insert(row)); } } } diff --git a/bt-daemon/src/translate/codex.rs b/bt-daemon/src/translate/codex.rs index 1bdf9f9a..3585e099 100644 --- a/bt-daemon/src/translate/codex.rs +++ b/bt-daemon/src/translate/codex.rs @@ -127,6 +127,7 @@ impl TranslatorFactory for crate::agents::Codex { // The main scope is created lazily once we learn its transcript path. scopes: HashMap::new(), spawn_turn_by_call_id: RecentMap::default(), + turn_owners: RecentMap::default(), spawn_turn_by_agent_id: RecentMap::default(), compaction_trigger_by_turn: RecentMap::default(), compaction_spans: RecentSet::default(), @@ -187,9 +188,19 @@ struct Scope { agent_id: Option, agent_type: Option, spawning_turn_span_id: Option, + /// User turn that owns this subagent's work; `None` for the main scope. + owner_turn_span_id: Option, subagent_ended: bool, } +impl Scope { + fn turn_owner(&self, turn_span_id: &str) -> String { + self.owner_turn_span_id + .clone() + .unwrap_or_else(|| turn_span_id.to_owned()) + } +} + enum DeferredHook { None, PostToolUse(Value), @@ -241,6 +252,7 @@ struct CodexTranslator { scopes: HashMap, spawn_turn_by_call_id: RecentMap, spawn_turn_by_agent_id: RecentMap, + turn_owners: RecentMap, compaction_trigger_by_turn: RecentMap, compaction_spans: RecentSet, pending: Option, @@ -541,6 +553,12 @@ impl CodexTranslator { .as_ref() .and_then(Value::as_str) .map(str::to_owned); + scope.owner_turn_span_id = (parent != self.root_span_id).then(|| { + self.turn_owners + .get(&parent) + .cloned() + .unwrap_or_else(|| parent.clone()) + }); scope.spawning_turn_span_id = Some(parent); self.scopes.insert(path, scope); } @@ -798,6 +816,7 @@ impl CodexTranslator { span_id: scope.turn_parent_span_id.clone(), root_span_id: self.effective_root_span_id.clone(), parent_span_ids: vec![parent], + turn_span_id: scope.owner_turn_span_id.clone(), name: format!("subagent: {agent_id}"), span_type: SpanType::Task, start_ms: Some(ts), @@ -822,17 +841,29 @@ impl CodexTranslator { return; } let span_id = ids::span_id(&self.session_id, &format!("turn:{turn_id}")); + // Native compaction hooks identify maintenance turns before catch-up. + let is_turn = self.compaction_trigger_by_turn.get(&turn_id).is_none() + && !self.compaction_spans.contains(&turn_id); + // A maintenance turn is not a user turn; inside a subagent it still + // belongs to the user turn that spawned the subagent. + let owner = if is_turn { + Some(scope.turn_owner(&span_id)) + } else { + scope.owner_turn_span_id.clone() + }; + if let Some(owner) = &owner { + self.turn_owners.insert(span_id.clone(), owner.clone()); + } ops.push(SpanOp::Insert(SpanRow { span_id: span_id.clone(), root_span_id: self.effective_root_span_id.clone(), 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), + is_turn, start_ms: Some(ts), metadata: Some(json!({ "turn_id": turn_id, "model": scope.model })), + turn_span_id: owner, ..Default::default() })); scope.open_turns.push(OpenTurn { @@ -929,6 +960,7 @@ impl CodexTranslator { ops.push(SpanOp::Insert(SpanRow { span_id: span_id.clone(), root_span_id: self.effective_root_span_id.clone(), + turn_span_id: self.turn_owners.get(&turn_span).cloned(), parent_span_ids: vec![turn_span], name, span_type: SpanType::Llm, @@ -1092,6 +1124,7 @@ impl CodexTranslator { ops.push(SpanOp::Insert(SpanRow { span_id: span_id.clone(), root_span_id: self.effective_root_span_id.clone(), + turn_span_id: self.turn_owners.get(&turn_span).cloned(), parent_span_ids: vec![turn_span], name, span_type: SpanType::Tool, @@ -1337,6 +1370,7 @@ impl CodexTranslator { ops.push(SpanOp::Insert(SpanRow { span_id: span_id.clone(), root_span_id: self.effective_root_span_id.clone(), + turn_span_id: self.turn_owners.get(&turn_span).cloned(), parent_span_ids: vec![turn_span.clone()], name: name.clone(), span_type: SpanType::Llm, @@ -1522,6 +1556,7 @@ impl Scope { agent_id: None, agent_type: None, spawning_turn_span_id: None, + owner_turn_span_id: None, subagent_ended: false, } } diff --git a/bt-daemon/src/translate/cursor/mod.rs b/bt-daemon/src/translate/cursor/mod.rs index 89af6a88..f2433702 100644 --- a/bt-daemon/src/translate/cursor/mod.rs +++ b/bt-daemon/src/translate/cursor/mod.rs @@ -267,10 +267,13 @@ impl CursorTranslator { } } fn row(&self, id: String, parent: String, name: &str, kind: SpanType) -> SpanRow { + // Every non-root Cursor span is parented directly to its turn. + let turn_span_id = (parent != self.root).then(|| parent.clone()); SpanRow { span_id: id, root_span_id: self.trace_root.clone(), parent_span_ids: vec![parent], + turn_span_id, name: name.into(), span_type: kind, ..Default::default() @@ -362,6 +365,7 @@ impl CursorTranslator { row.metadata = Some( json!({"turn_boundary_source":origin, "start_time_estimated":origin != "beforeSubmitPrompt", "generation_id":generation}), ); + row.turn_span_id = Some(id.clone()); ops.push(SpanOp::Insert(row)); let turn = Turn { ordinal: self.turn_seq, diff --git a/bt-daemon/src/translate/debug.rs b/bt-daemon/src/translate/debug.rs index b954898a..8d387afb 100644 --- a/bt-daemon/src/translate/debug.rs +++ b/bt-daemon/src/translate/debug.rs @@ -62,6 +62,7 @@ impl AgentTranslator for DebugTranslator { metrics: None, error: None, late_merge_key: None, + turn_span_id: None, tags: None, is_turn: false, origin: None, @@ -85,6 +86,7 @@ impl AgentTranslator for DebugTranslator { metrics: None, error: None, late_merge_key: None, + turn_span_id: None, tags: None, is_turn: false, origin: None, diff --git a/bt-daemon/src/translate/grok.rs b/bt-daemon/src/translate/grok.rs index 9712c59d..5dc29735 100644 --- a/bt-daemon/src/translate/grok.rs +++ b/bt-daemon/src/translate/grok.rs @@ -489,6 +489,7 @@ impl GrokTranslator { start_ms: Some(ts_ms), input, metadata: Some(json!({"transcript_turn_key": turn_key})), + turn_span_id: Some(span_id.clone()), ..Default::default() })); self.current_turn = Some(OpenTurn { @@ -583,6 +584,7 @@ impl GrokTranslator { span_id: span_id.clone(), root_span_id: self.root_span_id.clone(), parent_span_ids: vec![llm_parent_span_id.clone()], + turn_span_id: Some(llm_parent_span_id.clone()), name: format!("{model} call {sequence}"), span_type: SpanType::Llm, start_ms: Some(boundary_ms), @@ -664,6 +666,7 @@ impl GrokTranslator { span_id: span_id.clone(), root_span_id: self.root_span_id.clone(), parent_span_ids: vec![tool_parent_span_id.clone()], + turn_span_id: Some(tool_parent_span_id.clone()), name, span_type: SpanType::Tool, start_ms: Some(ts_ms), @@ -715,6 +718,7 @@ impl GrokTranslator { span_id: span_id.clone(), root_span_id: self.root_span_id.clone(), parent_span_ids: vec![parent_span_id.clone()], + turn_span_id: Some(parent_span_id.clone()), name: update .get("title") .and_then(Value::as_str) diff --git a/bt-daemon/src/translate/mod.rs b/bt-daemon/src/translate/mod.rs index eef0de36..2bd1b713 100644 --- a/bt-daemon/src/translate/mod.rs +++ b/bt-daemon/src/translate/mod.rs @@ -20,6 +20,8 @@ mod pi; pub(crate) use pi::request_config as pi_request_config; mod recent; mod tool; +mod turn_lineage; +pub use turn_lineage::TURN_SPAN_ID_KEY; use debug::DebugTranslatorFactory; @@ -86,6 +88,12 @@ pub struct SpanRow { #[serde(skip)] #[doc(hidden)] pub late_merge_key: Option, + /// Span id of the user turn that owns this row, recorded by the translator + /// from its native turn tracking. [`turn_lineage`] publishes it as + /// `metadata.turn_span_id`. + #[serde(skip)] + #[doc(hidden)] + pub turn_span_id: Option, /// Labels for filtering in Braintrust (e.g. `compaction`, `permission-request`). #[serde(default, skip_serializing_if = "Option::is_none")] pub tags: Option>, @@ -241,7 +249,9 @@ impl Registry { .ok_or_else(|| anyhow::anyhow!("unsupported coding-agent source {source:?}"))?; let factory = self.factories.get(canonical); let factory = factory.expect("canonical source must have a factory"); - Ok(factory.create(session_namespace)) + Ok(Box::new(turn_lineage::TurnLineage::new( + factory.create(session_namespace), + ))) } /// Create a known translator. Production ingress uses diff --git a/bt-daemon/src/translate/opencode.rs b/bt-daemon/src/translate/opencode.rs index 878a1f6a..a7577a31 100644 --- a/bt-daemon/src/translate/opencode.rs +++ b/bt-daemon/src/translate/opencode.rs @@ -335,6 +335,8 @@ struct NativeSession { effective_root_span_id: String, parent_span_ids: Vec, parent_session_id: Option, + /// For child sessions, the user turn in the root session that spawned them. + owner_turn_span_id: Option, root_idle: bool, current_turn_span_id: Option, turn_number: u32, @@ -360,6 +362,15 @@ struct NativeSession { history_tool_results: HashMap>, } +impl NativeSession { + /// The user turn that owns work in this session right now. + fn turn_owner(&self) -> Option { + self.owner_turn_span_id + .clone() + .or_else(|| self.current_turn_span_id.clone()) + } +} + #[derive(Default)] struct MessageHistory { entries: Vec, @@ -621,6 +632,9 @@ impl OpenCodeTranslator { metadata.insert("parent_session_id".into(), Value::String(parent.into())); metadata.insert("is_subagent".into(), Value::Bool(true)); } + let owner_turn_span_id = parent_id + .and_then(|parent| self.sessions.get(parent)) + .and_then(NativeSession::turn_owner); self.sessions.insert( native_id.to_string(), NativeSession { @@ -628,6 +642,7 @@ impl OpenCodeTranslator { effective_root_span_id: effective_root_span_id.clone(), parent_span_ids: parent_span_ids.clone(), parent_session_id: parent_id.map(str::to_owned), + owner_turn_span_id: owner_turn_span_id.clone(), ..Default::default() }, ); @@ -635,6 +650,7 @@ impl OpenCodeTranslator { span_id: root_span_id, root_span_id: effective_root_span_id, parent_span_ids, + turn_span_id: owner_turn_span_id, name, span_type: SpanType::Task, start_ms: Some(ts), @@ -745,7 +761,7 @@ impl OpenCodeTranslator { .and_then(Value::as_str); let skills = explicit_skills(&input); ops.push(SpanOp::Insert(SpanRow { - span_id: turn_id, + span_id: turn_id.clone(), root_span_id: state.effective_root_span_id.clone(), parent_span_ids: vec![state.root_span_id.clone()], name: format!("Turn {}", state.turn_number), @@ -756,6 +772,7 @@ impl OpenCodeTranslator { metadata: Some( json!({"turn_number":state.turn_number,"model":model,"loaded_skill_names":skills}), ), + turn_span_id: Some(state.owner_turn_span_id.clone().unwrap_or(turn_id)), ..Default::default() })); ops @@ -954,6 +971,12 @@ impl OpenCodeTranslator { vec![SpanOp::Insert(SpanRow { span_id: ids::span_id(&self.daemon_session_id, &format!("llm:{sid}:{mid}")), root_span_id: state.effective_root_span_id.clone(), + turn_span_id: Some( + state + .owner_turn_span_id + .clone() + .unwrap_or_else(|| turn.clone()), + ), parent_span_ids: vec![turn], name: format!("{provider}/{model}"), span_type: SpanType::Llm, @@ -1000,6 +1023,7 @@ impl OpenCodeTranslator { return vec![SpanOp::Insert(SpanRow { span_id: ids::span_id(&self.daemon_session_id, &format!("tool:{sid}:{call}")), root_span_id: s.effective_root_span_id.clone(), + turn_span_id: s.turn_owner(), parent_span_ids: vec![turn], name: tool.clone(), span_type: SpanType::Tool, @@ -1076,6 +1100,7 @@ impl OpenCodeTranslator { span_id: ids::span_id(&self.daemon_session_id, &format!("tool:{sid}:{call}")), root_span_id: s.effective_root_span_id.clone(), parent_span_ids: vec![parent_span_id], + turn_span_id: (!had_start).then(|| s.turn_owner()).flatten(), name, span_type: SpanType::Tool, start_ms: (!had_start).then(|| s.tool_starts.remove(&call).unwrap_or(ts_ms)), @@ -1202,6 +1227,7 @@ impl OpenCodeTranslator { span_id: ids::span_id(&self.daemon_session_id, &format!("tool:{sid}:{call}")), root_span_id: s.effective_root_span_id.clone(), parent_span_ids: vec![parent_span_id], + turn_span_id: (!had_start).then(|| s.turn_owner()).flatten(), name: title.unwrap_or(tool), span_type: SpanType::Tool, start_ms: (!had_start).then(|| s.tool_starts.remove(&call).unwrap_or(ts_ms)), diff --git a/bt-daemon/src/translate/pi.rs b/bt-daemon/src/translate/pi.rs index 3054862d..afc6dbd7 100644 --- a/bt-daemon/src/translate/pi.rs +++ b/bt-daemon/src/translate/pi.rs @@ -476,6 +476,7 @@ impl PiTranslator { ), root_span_id: self.effective_root_span_id.clone(), parent_span_ids: vec![turn.clone()], + turn_span_id: Some(turn.clone()), name: "llm".into(), span_type: SpanType::Llm, start_ms: Some(pending.start_ms), @@ -493,6 +494,7 @@ impl PiTranslator { span_id: self.tool_span_id(&call), root_span_id: self.effective_root_span_id.clone(), parent_span_ids: vec![self.tool_parent(turn, tool.parent_call.as_deref())], + turn_span_id: Some(turn.clone()), name: tool.name.clone(), span_type: SpanType::Tool, start_ms: Some(tool.start_ms), @@ -608,7 +610,7 @@ impl PiTranslator { } self.turn = Some((id.clone(), skills)); ops.push(SpanOp::Insert(SpanRow { - span_id: id, + span_id: id.clone(), root_span_id: self.effective_root_span_id.clone(), parent_span_ids: vec![self.root_span_id.clone()], name: format!("Turn {}", self.turn_seq), @@ -617,6 +619,7 @@ impl PiTranslator { start_ms: Some(ts), input: Some(input), metadata: Some(metadata), + turn_span_id: Some(id), ..Default::default() })); ops @@ -738,6 +741,7 @@ impl PiTranslator { ), root_span_id: self.effective_root_span_id.clone(), parent_span_ids: vec![turn.clone()], + turn_span_id: Some(turn.clone()), name: model.clone().unwrap_or_else(|| "llm".into()), span_type: SpanType::Llm, start_ms: Some(pending.start_ms), @@ -795,6 +799,7 @@ impl PiTranslator { span_id: self.tool_span_id(&id), root_span_id: self.effective_root_span_id.clone(), parent_span_ids: vec![parent], + turn_span_id: Some(turn.clone()), name: name.clone(), span_type: SpanType::Tool, start_ms: Some(ts), @@ -834,6 +839,7 @@ impl PiTranslator { span_id: self.tool_span_id(&call), root_span_id: self.effective_root_span_id.clone(), parent_span_ids: vec![self.tool_parent(turn, tracked.parent_call.as_deref())], + turn_span_id: pending.is_none().then(|| turn.clone()), name, span_type: SpanType::Tool, start_ms: pending.is_none().then_some(tracked.start_ms), diff --git a/bt-daemon/src/translate/turn_lineage.rs b/bt-daemon/src/translate/turn_lineage.rs new file mode 100644 index 00000000..41e35df0 --- /dev/null +++ b/bt-daemon/src/translate/turn_lineage.rs @@ -0,0 +1,108 @@ +//! Publishes the user turn that owns each span. +//! +//! Translators already track which native turn every operation belongs to in +//! order to parent it, so they record the owning user turn's span id on each +//! row ([`SpanRow::turn_span_id`]). Nested agent activity (subagents, child +//! sessions, continuation turns) records the user turn that spawned it, so a +//! turn's cost can be aggregated with a flat filter instead of a tree walk. +//! +//! This wrapper copies that value into `metadata.turn_span_id` after the +//! translator has applied its own metadata allowlist. It keeps no state. + +use super::{AgentTranslator, SessionCtx, SpanOp, SpanRow}; +use serde_json::{Map, Value}; + +pub const TURN_SPAN_ID_KEY: &str = "turn_span_id"; + +pub(crate) struct TurnLineage { + inner: Box, +} + +impl TurnLineage { + pub(crate) fn new(inner: Box) -> Self { + Self { inner } + } +} + +fn stamp(mut ops: Vec) -> Vec { + for op in &mut ops { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + publish_turn(row); + } + ops +} + +fn publish_turn(row: &mut SpanRow) { + let Some(turn) = row.turn_span_id.clone() else { + return; + }; + let metadata = row + .metadata + .get_or_insert_with(|| Value::Object(Map::new())); + if let Some(metadata) = metadata.as_object_mut() { + metadata.insert(TURN_SPAN_ID_KEY.into(), Value::String(turn)); + } +} + +impl AgentTranslator for TurnLineage { + fn handle( + &mut self, + event: &crate::wire::Envelope, + ctx: &SessionCtx, + ) -> anyhow::Result> { + Ok(stamp(self.inner.handle(event, ctx)?)) + } + + fn drain_pending(&mut self, ctx: &SessionCtx) -> anyhow::Result>> { + Ok(self.inner.drain_pending(ctx)?.map(stamp)) + } + + fn checkpoint(&mut self, ctx: &SessionCtx) -> anyhow::Result> { + Ok(stamp(self.inner.checkpoint(ctx)?)) + } + + fn finalize(&mut self, ctx: &SessionCtx) -> anyhow::Result> { + Ok(stamp(self.inner.finalize(ctx)?)) + } + + fn flush(&mut self, ctx: &SessionCtx) -> anyhow::Result> { + Ok(stamp(self.inner.flush(ctx)?)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn row(turn: Option<&str>, metadata: Option) -> SpanOp { + SpanOp::Insert(SpanRow { + span_id: "span".into(), + root_span_id: "root".into(), + turn_span_id: turn.map(str::to_owned), + metadata, + ..Default::default() + }) + } + + fn metadata(op: &SpanOp) -> Option<&Value> { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + row.metadata.as_ref() + } + + #[test] + fn publishes_the_translator_recorded_turn_alongside_existing_metadata() { + let ops = stamp(vec![row(Some("turn"), Some(json!({"model": "m"})))]); + assert_eq!( + metadata(&ops[0]), + Some(&json!({"model": "m", "turn_span_id": "turn"})) + ); + } + + #[test] + fn rows_without_a_turn_are_left_untouched() { + let ops = stamp(vec![row(None, None), row(None, Some(json!({"a": 1})))]); + assert_eq!(metadata(&ops[0]), None); + assert_eq!(metadata(&ops[1]), Some(&json!({"a": 1}))); + } +} diff --git a/bt-daemon/tests/antigravity_translator.rs b/bt-daemon/tests/antigravity_translator.rs index 743ef559..2cd8b9ba 100644 --- a/bt-daemon/tests/antigravity_translator.rs +++ b/bt-daemon/tests/antigravity_translator.rs @@ -5,7 +5,7 @@ use braintrust_sdk_rust::{SpanComponents, SpanObjectType}; use bt_daemon::wire::{BackendAuth, Envelope, SessionRoute, TraceDestination}; use bt_daemon::{Registry, SessionCtx, SpanOp, SpanRow, SpanType}; use serde_json::{json, Value}; -use span_identity::assert_merges_preserve_insert_identity; +use span_identity::{assert_merges_preserve_insert_identity, assert_turn_lineage}; use std::collections::HashMap; fn jsonl(records: &[Value]) -> (String, Vec) { @@ -282,6 +282,7 @@ fn hooks_and_full_transcript_build_model_and_tool_spans() { .unwrap(), ); + assert!(assert_turn_lineage(&ops, |row| row.name.starts_with("Turn ")) >= 3); let rows = reduce(ops); let root = rows .values() diff --git a/bt-daemon/tests/braintrust_sink.rs b/bt-daemon/tests/braintrust_sink.rs index e694818e..fcd8ed4e 100644 --- a/bt-daemon/tests/braintrust_sink.rs +++ b/bt-daemon/tests/braintrust_sink.rs @@ -55,6 +55,7 @@ fn row( metrics: None, error: None, late_merge_key: None, + turn_span_id: None, tags: None, origin: parents.is_empty().then(|| OriginSnapshot { plugin_version: None, diff --git a/bt-daemon/tests/claude_translator.rs b/bt-daemon/tests/claude_translator.rs index 26dbf517..e2c5ec9a 100644 --- a/bt-daemon/tests/claude_translator.rs +++ b/bt-daemon/tests/claude_translator.rs @@ -10,7 +10,7 @@ use braintrust_sdk_rust::{SpanComponents, SpanObjectType}; use bt_daemon::wire::{BackendAuth, Envelope, SessionRoute, TraceDestination}; use bt_daemon::{Registry, SessionCtx, SpanOp, SpanRow, SpanType}; use serde_json::{json, Value}; -use span_identity::assert_merges_preserve_insert_identity; +use span_identity::{assert_merges_preserve_insert_identity, assert_turn_lineage}; use std::collections::HashMap; use std::path::{Path, PathBuf}; @@ -148,6 +148,26 @@ fn reduce(ops: Vec) -> HashMap { rows } +/// Recorded sessions, including subagents and compaction, attribute every span +/// under a user turn to that turn so its cost can be aggregated flatly. +#[test] +fn every_span_under_a_turn_carries_its_turn_span_id() { + for name in ["test-fixture", "example-simple", "subagent-compact"] { + let ops = replay_from(name, Source::Mirror); + let llm_spans = ops + .iter() + .filter(|op| matches!(op, SpanOp::Insert(row) if row.span_type == SpanType::Llm)) + .count(); + let stamped = assert_turn_lineage(&ops, |row| { + row.name.starts_with("Turn ") || row.name.starts_with("Continuation") + }); + assert!( + stamped > llm_spans, + "{name}: expected turns, LLM calls, and tools" + ); + } +} + /// Journals recorded before transcript mirroring inline the whole transcript. /// Both forms must translate identically, so upgrading the daemon neither /// changes live output nor breaks recovery from an existing journal. diff --git a/bt-daemon/tests/codex_translator.rs b/bt-daemon/tests/codex_translator.rs index 34050ed4..9844ce34 100644 --- a/bt-daemon/tests/codex_translator.rs +++ b/bt-daemon/tests/codex_translator.rs @@ -14,7 +14,7 @@ use braintrust_sdk_rust::{SpanComponents, SpanObjectType}; use bt_daemon::wire::{BackendAuth, Envelope, FlushMode, SessionConfig, TraceDestination}; use bt_daemon::{Registry, SessionCtx, SpanOp, SpanRow, SpanType}; use serde_json::{json, Value}; -use span_identity::assert_merges_preserve_insert_identity; +use span_identity::{assert_merges_preserve_insert_identity, assert_turn_lineage}; use std::collections::HashMap; use std::io::Write; @@ -1270,6 +1270,9 @@ fn codex_compaction_relabels_turn_and_adds_compaction_llm() { .unwrap(), ); assert_merges_preserve_insert_identity(&ops); + // Compaction discovered only during catch-up opened as a turn, so its + // synthetic llm span follows that turn's ownership. + assert!(assert_turn_lineage(&ops, |row| row.is_turn) >= 2); let rows = reduce(ops); let compaction = find(&rows, SpanType::Task, "compaction"); @@ -1310,6 +1313,78 @@ fn codex_compaction_relabels_turn_and_adds_compaction_llm() { assert!(llm.end_ms.is_some()); } +#[test] +fn codex_hook_identified_compaction_is_not_a_user_turn() { + let tmp = tempfile::tempdir().unwrap(); + let t = tmp.path().join("rollout.jsonl"); + let tpath = t.to_str().unwrap(); + for v in [ + json!({ "timestamp": "2026-01-01T00:00:01Z", "type": "session_meta", "payload": { "id": "s", "cwd": "/x/app" } }), + json!({ "timestamp": "2026-01-01T00:00:02Z", "type": "turn_context", "payload": { "model": "gpt-5.5" } }), + ] { + append(&t, v); + } + + let reg = Registry::default_agents(); + let mut tr = reg.create("codex", "s"); + let ctx = SessionCtx { + session_id: "s".into(), + config: None, + }; + + let mut ops = Vec::new(); + ops.extend( + tr.handle(&envelope("s", "SessionStart", tpath, json!({})), &ctx) + .unwrap(), + ); + // PreCompact arrives before the compaction turn reaches the transcript. + ops.extend( + tr.handle( + &envelope( + "s", + "PreCompact", + tpath, + json!({ "turn_id": "c1", "trigger": "auto" }), + ), + &ctx, + ) + .unwrap(), + ); + for v in [ + json!({ "timestamp": "2026-01-01T00:00:03Z", "type": "event_msg", "payload": { "type": "task_started", "turn_id": "c1" } }), + json!({ "timestamp": "2026-01-01T00:00:04Z", "type": "compacted", "payload": { + "window_id": "w1", + "replacement_history": [{ "role": "user", "content": "kept" }] + } }), + json!({ "timestamp": "2026-01-01T00:00:05Z", "type": "event_msg", "payload": { "type": "token_count", "info": { "last_token_usage": { "input_tokens": 5000, "output_tokens": 50 } } } }), + ] { + append(&t, v); + } + ops.extend( + tr.handle( + &envelope( + "s", + "PostCompact", + tpath, + json!({ "turn_id": "c1", "trigger": "auto" }), + ), + &ctx, + ) + .unwrap(), + ); + + assert_merges_preserve_insert_identity(&ops); + // Maintenance compaction must not create a chargeable user turn. + assert_eq!(assert_turn_lineage(&ops, |row| row.is_turn), 0); + let llm_inserted = ops + .iter() + .any(|op| matches!(op, SpanOp::Insert(row) if row.span_type == SpanType::Llm)); + assert!( + llm_inserted, + "fixture should emit the synthetic compaction llm" + ); +} + #[test] fn codex_compaction_replaces_history_for_following_llms() { let tmp = tempfile::tempdir().unwrap(); @@ -1536,6 +1611,8 @@ fn codex_subagent_with_malformed_optional_type_nests_under_spawning_turn() { ); assert_merges_preserve_insert_identity(&ops); + // The subagent's own turn and model call roll up to the spawning user turn. + assert!(assert_turn_lineage(&ops, |row| row.name.starts_with("turn: ")) >= 5); let rows = reduce(ops); let root = find(&rows, SpanType::Task, "codex: app"); diff --git a/bt-daemon/tests/cursor_translator.rs b/bt-daemon/tests/cursor_translator.rs index 4d6c6c5e..fc8afaba 100644 --- a/bt-daemon/tests/cursor_translator.rs +++ b/bt-daemon/tests/cursor_translator.rs @@ -7,7 +7,7 @@ use braintrust_sdk_rust::{SpanComponents, SpanObjectType}; use bt_daemon::wire::{BackendAuth, Envelope, SessionRoute, TraceDestination}; use bt_daemon::{AgentTranslator, Registry, SessionCtx, SpanOp, SpanRow, SpanType}; use serde_json::{json, Value}; -use span_identity::IdentityLedger; +use span_identity::{assert_turn_lineage, IdentityLedger}; use std::collections::HashMap; use std::path::{Path, PathBuf}; @@ -158,6 +158,25 @@ fn real_capture(scenario: &str) -> Harness { harness.finish(); harness } +#[test] +fn real_cursor_spans_under_a_turn_carry_its_turn_span_id() { + for scenario in [ + "daemon-two-turn-success", + "headless-tools-error", + "headless-tools-success", + "interactive-multi-turn-error", + "interactive-multi-turn-success", + "plugin-headless-resume-success", + ] { + let h = real_capture(scenario); + let stamped = assert_turn_lineage(&h.ops, |row| row.name.starts_with("Turn ")); + assert!( + stamped > h.turns().len(), + "{scenario}: expected turn children" + ); + } +} + #[test] fn real_interactive_error_capture_has_two_turns_one_read_and_turn_scoped_usage() { let h = real_capture("interactive-multi-turn-error"); diff --git a/bt-daemon/tests/grok_translator.rs b/bt-daemon/tests/grok_translator.rs index 0980e7cb..1307c702 100644 --- a/bt-daemon/tests/grok_translator.rs +++ b/bt-daemon/tests/grok_translator.rs @@ -10,7 +10,7 @@ use braintrust_sdk_rust::{SpanComponents, SpanObjectType}; use bt_daemon::wire::{BackendAuth, Envelope, SessionRoute, TraceDestination}; use bt_daemon::{AgentTranslator, Registry, SessionCtx, SpanOp, SpanType}; use serde_json::json; -use span_identity::{assert_merges_preserve_insert_identity, IdentityLedger}; +use span_identity::{assert_merges_preserve_insert_identity, assert_turn_lineage, IdentityLedger}; use std::path::{Path, PathBuf}; fn fixture(name: &str) -> PathBuf { @@ -103,6 +103,17 @@ fn drain(translator: &mut dyn AgentTranslator) -> Vec { ops } +#[test] +fn grok_spans_under_a_turn_carry_its_turn_span_id() { + let updates = std::fs::metadata(fixture("updates.jsonl")).unwrap().len(); + let events = std::fs::metadata(fixture("events.jsonl")).unwrap().len(); + let mut translator = Registry::default_agents().create("grok", "grok-session"); + let ops = translator + .handle(&envelope(updates, events), &ctx()) + .unwrap(); + assert!(assert_turn_lineage(&ops, |row| row.name.starts_with("Turn ")) >= 3); +} + #[test] fn grok_transcript_builds_turn_llm_and_tool_spans_with_aggregate_usage() { let updates = std::fs::metadata(fixture("updates.jsonl")).unwrap().len(); diff --git a/bt-daemon/tests/opencode_translator.rs b/bt-daemon/tests/opencode_translator.rs index 218f8b8b..a6202f86 100644 --- a/bt-daemon/tests/opencode_translator.rs +++ b/bt-daemon/tests/opencode_translator.rs @@ -5,7 +5,7 @@ use braintrust_sdk_rust::{SpanComponents, SpanObjectType}; use bt_daemon::wire::{BackendAuth, Envelope, SessionRoute, TraceDestination}; use bt_daemon::{Registry, SessionCtx, SpanOp, SpanRow, SpanType}; use serde_json::json; -use span_identity::assert_merges_preserve_insert_identity; +use span_identity::{assert_merges_preserve_insert_identity, assert_turn_lineage}; use std::collections::HashMap; fn event(name: &str, ts_ms: i64, payload: serde_json::Value) -> Envelope { @@ -102,6 +102,7 @@ fn opencode_builds_turn_llm_tool_and_closes_the_session() { for event in events { ops.extend(translator.handle(&event, &ctx).unwrap()); } + assert!(assert_turn_lineage(&ops, |row| row.name.starts_with("Turn ")) >= 3); let rows = reduce(ops); assert_eq!( rows.values() @@ -359,6 +360,7 @@ fn opencode_child_sessions_share_the_parent_trace_root() { ) .unwrap(), ); + assert!(assert_turn_lineage(&ops, |row| row.name.starts_with("Turn ")) >= 2); let rows = reduce(ops); let parent = rows.values().find(|r| r.name == "OpenCode").unwrap(); let child = rows diff --git a/bt-daemon/tests/pi_translator.rs b/bt-daemon/tests/pi_translator.rs index 9c9683bc..5bb5695a 100644 --- a/bt-daemon/tests/pi_translator.rs +++ b/bt-daemon/tests/pi_translator.rs @@ -3,9 +3,9 @@ mod span_identity; use braintrust_sdk_rust::{SpanComponents, SpanObjectType}; use bt_daemon::wire::{BackendAuth, Envelope, SessionRoute, TraceDestination}; -use bt_daemon::{Registry, SessionCtx, SpanOp, SpanRow, SpanType}; +use bt_daemon::{Registry, SessionCtx, SpanOp, SpanRow, SpanType, TURN_SPAN_ID_KEY}; use serde_json::json; -use span_identity::assert_merges_preserve_insert_identity; +use span_identity::{assert_merges_preserve_insert_identity, assert_turn_lineage}; use std::collections::HashMap; fn event(name: &str, ts_ms: i64, native: serde_json::Value) -> Envelope { @@ -172,6 +172,7 @@ fn pi_builds_turn_llm_tool_compaction_and_shutdown_spans() { for event in events { ops.extend(translator.handle(&event, &ctx).unwrap()); } + assert!(assert_turn_lineage(&ops, |row| row.name.starts_with("Turn ")) >= 6); let rows = reduce(ops); assert_eq!( rows.values() @@ -226,7 +227,7 @@ fn pi_builds_turn_llm_tool_compaction_and_shutdown_spans() { // Only allowlisted request configuration and the available tools reach // metadata, never the provider payload's instructions or messages. assert_eq!( - metadata_without(first_llm, &[]), + metadata_without(first_llm, &[TURN_SPAN_ID_KEY]), json!({ "model": "gpt-5", "provider": "openai", @@ -950,7 +951,7 @@ fn pi_llm_metadata_selects_request_configuration_across_provider_payloads() { .values() .find(|row| row.span_type == SpanType::Llm && row.start_ms == Some(start)) .unwrap(); - metadata_without(row, &["model", "provider"]) + metadata_without(row, &["model", "provider", TURN_SPAN_ID_KEY]) }; let read = |strict: Option| { let mut function = json!({"name":"read","description":"Read","parameters":schema}); diff --git a/bt-daemon/tests/support/span_identity.rs b/bt-daemon/tests/support/span_identity.rs index 763f5479..b24d2907 100644 --- a/bt-daemon/tests/support/span_identity.rs +++ b/bt-daemon/tests/support/span_identity.rs @@ -1,5 +1,6 @@ -use bt_daemon::SpanOp; -use std::collections::HashMap; +use bt_daemon::{SpanOp, SpanRow, TURN_SPAN_ID_KEY}; +use serde_json::Value; +use std::collections::{HashMap, HashSet}; /// Remembers the identity each span was inserted with, so merges emitted in a /// later batch than their insert can still be checked. @@ -46,3 +47,53 @@ impl IdentityLedger { pub(crate) fn assert_merges_preserve_insert_identity(ops: &[SpanOp]) { IdentityLedger::default().check(ops); } + +/// Every span with a turn ancestor (or that is a turn) ends up carrying the +/// outermost such turn's id, and spans outside any turn carry none. A later +/// row may correct an earlier stamp, so the last value written wins. Returns +/// how many spans were stamped so callers can assert the fixture exercised +/// lineage. +#[allow(dead_code)] +pub(crate) fn assert_turn_lineage(ops: &[SpanOp], is_turn: impl Fn(&SpanRow) -> bool) -> usize { + let mut parent_of = HashMap::::new(); + let mut turns = HashSet::::new(); + let mut stamped = HashMap::::new(); + let mut spans = HashSet::::new(); + for op in ops { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + spans.insert(row.span_id.clone()); + if let Some(parent) = row.parent_span_ids.first() { + parent_of + .entry(row.span_id.clone()) + .or_insert_with(|| parent.clone()); + } + if matches!(op, SpanOp::Insert(_)) && is_turn(row) { + turns.insert(row.span_id.clone()); + } + let turn = row + .metadata + .as_ref() + .and_then(|metadata| metadata.get(TURN_SPAN_ID_KEY)) + .and_then(Value::as_str); + if let Some(turn) = turn { + stamped.insert(row.span_id.clone(), turn.to_owned()); + } + } + for span in &spans { + let mut expected = None; + let mut current = Some(span); + for _ in 0..=spans.len() { + let Some(id) = current else { break }; + if turns.contains(id) { + expected = Some(id); + } + current = parent_of.get(id); + } + assert_eq!( + stamped.get(span), + expected, + "span {span} has the wrong {TURN_SPAN_ID_KEY}" + ); + } + stamped.len() +}