From 86877bbd7b4d8ae38b78c44f58abcb8f6ea1b7d5 Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Wed, 7 Oct 2026 13:33:17 -0700 Subject: [PATCH 1/4] feat(daemon): stamp turn_span_id on every span under a turn Translators mark the span that opens a user turn (SpanRow::turn_root). A shared TurnLineage wrapper, applied to every translator in Registry, copies that span's id into metadata.turn_span_id on the turn and all of its descendants, including subagent activity. The outermost turn wins, so nested agent turns roll up to the user turn that spawned them. Rows whose ancestry resolves late are re-emitted as late merges. This lets per-turn cost be aggregated with a flat filter instead of a trace tree walk. Co-Authored-By: Claude Opus 5.5 (1M context) --- bt-daemon/src/lib.rs | 4 +- bt-daemon/src/translate/antigravity.rs | 1 + bt-daemon/src/translate/claude.rs | 1 + bt-daemon/src/translate/codex.rs | 1 + bt-daemon/src/translate/cursor/mod.rs | 1 + bt-daemon/src/translate/debug.rs | 2 + bt-daemon/src/translate/grok.rs | 1 + bt-daemon/src/translate/mod.rs | 11 +- bt-daemon/src/translate/opencode.rs | 1 + bt-daemon/src/translate/pi.rs | 1 + bt-daemon/src/translate/turn_lineage.rs | 289 +++++++++++++++++++++++ bt-daemon/tests/braintrust_sink.rs | 1 + bt-daemon/tests/claude_translator.rs | 22 +- bt-daemon/tests/codex_translator.rs | 4 +- bt-daemon/tests/cursor_translator.rs | 21 +- bt-daemon/tests/grok_translator.rs | 13 +- bt-daemon/tests/pi_translator.rs | 6 +- bt-daemon/tests/support/span_identity.rs | 58 ++++- 18 files changed, 427 insertions(+), 11 deletions(-) create mode 100644 bt-daemon/src/translate/turn_lineage.rs diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 3c7b7f35..97696c09 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, 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 017eca3a..1595e3ce 100644 --- a/bt-daemon/src/translate/antigravity.rs +++ b/bt-daemon/src/translate/antigravity.rs @@ -224,6 +224,7 @@ impl AntigravityTranslator { start_ms: Some(ts_ms), input: nonempty_value(input), metadata: Some(json!({"turn_number": self.turn_count})), + turn_root: true, ..Default::default() })); self.turn = Some(Turn { diff --git a/bt-daemon/src/translate/claude.rs b/bt-daemon/src/translate/claude.rs index 9bcb6171..d1e1a3f3 100644 --- a/bt-daemon/src/translate/claude.rs +++ b/bt-daemon/src/translate/claude.rs @@ -524,6 +524,7 @@ impl ClaudeTranslator { start_ms: Some(event.ts_ms), input: event.payload.get("prompt").cloned(), metadata: (!metadata.is_empty()).then_some(Value::Object(metadata)), + turn_root: true, ..Default::default() })); let turn_id = Arc::new(id.clone()); diff --git a/bt-daemon/src/translate/codex.rs b/bt-daemon/src/translate/codex.rs index 39c5c528..9337b72a 100644 --- a/bt-daemon/src/translate/codex.rs +++ b/bt-daemon/src/translate/codex.rs @@ -830,6 +830,7 @@ impl CodexTranslator { span_type: SpanType::Task, start_ms: Some(ts), metadata: Some(json!({ "turn_id": turn_id, "model": scope.model })), + turn_root: true, ..Default::default() })); scope.open_turns.push(OpenTurn { diff --git a/bt-daemon/src/translate/cursor/mod.rs b/bt-daemon/src/translate/cursor/mod.rs index 969d9e88..0d6b735e 100644 --- a/bt-daemon/src/translate/cursor/mod.rs +++ b/bt-daemon/src/translate/cursor/mod.rs @@ -361,6 +361,7 @@ impl CursorTranslator { row.metadata = Some( json!({"turn_boundary_source":origin, "start_time_estimated":origin != "beforeSubmitPrompt", "generation_id":generation}), ); + row.turn_root = true; 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 484b439d..4bac2474 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_root: false, tags: None, })); } @@ -83,6 +84,7 @@ impl AgentTranslator for DebugTranslator { metrics: None, error: None, late_merge_key: None, + turn_root: false, tags: None, })); diff --git a/bt-daemon/src/translate/grok.rs b/bt-daemon/src/translate/grok.rs index 9ddbdfa4..9b122073 100644 --- a/bt-daemon/src/translate/grok.rs +++ b/bt-daemon/src/translate/grok.rs @@ -488,6 +488,7 @@ impl GrokTranslator { start_ms: Some(ts_ms), input, metadata: Some(json!({"transcript_turn_key": turn_key})), + turn_root: true, ..Default::default() })); self.current_turn = Some(OpenTurn { diff --git a/bt-daemon/src/translate/mod.rs b/bt-daemon/src/translate/mod.rs index 1e4ff483..c468cf9b 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; @@ -78,6 +80,11 @@ pub struct SpanRow { #[serde(skip)] #[doc(hidden)] pub late_merge_key: Option, + /// Translator-only marker for the span that opens a user turn. + /// [`turn_lineage`] stamps its id onto the turn and every descendant. + #[serde(skip)] + #[doc(hidden)] + pub turn_root: bool, /// Labels for filtering in Braintrust (e.g. `compaction`, `permission-request`). #[serde(default, skip_serializing_if = "Option::is_none")] pub tags: Option>, @@ -227,7 +234,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 4214ec3a..95f6ea6f 100644 --- a/bt-daemon/src/translate/opencode.rs +++ b/bt-daemon/src/translate/opencode.rs @@ -755,6 +755,7 @@ impl OpenCodeTranslator { metadata: Some( json!({"turn_number":state.turn_number,"model":model,"loaded_skill_names":skills}), ), + turn_root: true, ..Default::default() })); ops diff --git a/bt-daemon/src/translate/pi.rs b/bt-daemon/src/translate/pi.rs index 22e1ffde..6c09bee9 100644 --- a/bt-daemon/src/translate/pi.rs +++ b/bt-daemon/src/translate/pi.rs @@ -616,6 +616,7 @@ impl PiTranslator { start_ms: Some(ts), input: Some(input), metadata: Some(metadata), + turn_root: true, ..Default::default() })); ops diff --git a/bt-daemon/src/translate/turn_lineage.rs b/bt-daemon/src/translate/turn_lineage.rs new file mode 100644 index 00000000..499b2105 --- /dev/null +++ b/bt-daemon/src/translate/turn_lineage.rs @@ -0,0 +1,289 @@ +//! Stamps every span with the user turn that owns it. +//! +//! Translators only mark the span that opens a turn ([`SpanRow::turn_root`]). +//! This wrapper copies that span's id into `metadata.turn_span_id` on the turn +//! and on every descendant, including nested subagent activity, so a turn's +//! cost can be aggregated with a flat filter instead of a tree walk. +//! +//! The outermost turn wins: a subagent's own turns roll up to the user turn +//! that spawned them. A row whose ancestry is not yet known is remembered and +//! re-emitted as a late merge once its parent resolves. +//! +//! Translator state is rebuilt by journal replay, which also replays through +//! this wrapper, so the lineage maps need no persistence of their own. + +use super::{AgentTranslator, SessionCtx, SpanOp, SpanRow}; +use serde_json::{Map, Value}; +use std::collections::HashMap; + +pub const TURN_SPAN_ID_KEY: &str = "turn_span_id"; +const LATE_MERGE_KEY: &str = "turn_lineage"; + +struct Unresolved { + root_span_id: String, + parent_span_ids: Vec, +} + +pub(crate) struct TurnLineage { + inner: Box, + turn_of: HashMap, + parent_of: HashMap, + unresolved_children: HashMap>, +} + +impl TurnLineage { + pub(crate) fn new(inner: Box) -> Self { + Self { + inner, + turn_of: HashMap::new(), + parent_of: HashMap::new(), + unresolved_children: HashMap::new(), + } + } + + fn stamp(&mut self, ops: Vec) -> Vec { + let mut out = Vec::with_capacity(ops.len()); + for mut op in ops { + let is_insert = matches!(op, SpanOp::Insert(_)); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = &mut op; + let span_id = row.span_id.clone(); + let already_stamped = self.turn_of.contains_key(&span_id); + let turn = self.resolve(row); + // Merges keep earlier metadata, so only rows that create or replace + // a span, or the first row to resolve it, need the key. + if let Some(turn) = turn.as_ref().filter(|_| is_insert || !already_stamped) { + set_turn(row, turn); + } + out.push(op); + if let Some(turn) = turn { + self.release_children(&span_id, &turn, &mut out); + } + } + out + } + + fn resolve(&mut self, row: &SpanRow) -> Option { + if let Some(parent) = row.parent_span_ids.first() { + self.parent_of + .entry(row.span_id.clone()) + .or_insert_with(|| parent.clone()); + } + if let Some(turn) = self.turn_of.get(&row.span_id) { + return Some(turn.clone()); + } + let parent = self.parent_of.get(&row.span_id).cloned(); + let inherited = parent + .as_ref() + .and_then(|parent| self.turn_of.get(parent)) + .cloned(); + let Some(turn) = inherited.or_else(|| row.turn_root.then(|| row.span_id.clone())) else { + if let Some(parent) = parent { + self.unresolved_children + .entry(parent) + .or_default() + .entry(row.span_id.clone()) + .or_insert_with(|| Unresolved { + root_span_id: row.root_span_id.clone(), + parent_span_ids: row.parent_span_ids.clone(), + }); + } + return None; + }; + self.turn_of.insert(row.span_id.clone(), turn.clone()); + Some(turn) + } + + fn release_children(&mut self, span_id: &str, turn: &str, out: &mut Vec) { + let mut resolved = vec![span_id.to_owned()]; + while let Some(parent) = resolved.pop() { + let Some(children) = self.unresolved_children.remove(&parent) else { + continue; + }; + for (child, unresolved) in children { + self.turn_of.insert(child.clone(), turn.to_owned()); + let mut row = SpanRow { + span_id: child.clone(), + root_span_id: unresolved.root_span_id, + parent_span_ids: unresolved.parent_span_ids, + late_merge_key: Some(LATE_MERGE_KEY.into()), + ..Default::default() + }; + set_turn(&mut row, turn); + out.push(SpanOp::Merge(row)); + resolved.push(child); + } + } + } +} + +fn set_turn(row: &mut SpanRow, turn: &str) { + 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.to_owned())); + } +} + +impl AgentTranslator for TurnLineage { + fn handle( + &mut self, + event: &crate::wire::Envelope, + ctx: &SessionCtx, + ) -> anyhow::Result> { + let ops = self.inner.handle(event, ctx)?; + Ok(self.stamp(ops)) + } + + fn drain_pending(&mut self, ctx: &SessionCtx) -> anyhow::Result>> { + let ops = self.inner.drain_pending(ctx)?; + Ok(ops.map(|ops| self.stamp(ops))) + } + + fn checkpoint(&mut self, ctx: &SessionCtx) -> anyhow::Result> { + let ops = self.inner.checkpoint(ctx)?; + Ok(self.stamp(ops)) + } + + fn finalize(&mut self, ctx: &SessionCtx) -> anyhow::Result> { + let ops = self.inner.finalize(ctx)?; + Ok(self.stamp(ops)) + } + + fn flush(&mut self, ctx: &SessionCtx) -> anyhow::Result> { + let ops = self.inner.flush(ctx)?; + Ok(self.stamp(ops)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::translate::SpanType; + use crate::wire::Envelope; + use std::collections::VecDeque; + + struct Scripted(VecDeque>); + + impl AgentTranslator for Scripted { + fn handle(&mut self, _event: &Envelope, _ctx: &SessionCtx) -> anyhow::Result> { + Ok(self.0.pop_front().unwrap_or_default()) + } + } + + fn row(span_id: &str, parent: Option<&str>, span_type: SpanType) -> SpanRow { + SpanRow { + span_id: span_id.into(), + root_span_id: "root".into(), + parent_span_ids: parent.map(|p| vec![p.to_owned()]).unwrap_or_default(), + name: span_id.into(), + span_type, + ..Default::default() + } + } + + fn turn(span_id: &str, parent: &str) -> SpanRow { + SpanRow { + turn_root: true, + ..row(span_id, Some(parent), SpanType::Task) + } + } + + fn run(batches: Vec>) -> Vec { + let count = batches.len(); + let mut lineage = TurnLineage::new(Box::new(Scripted(batches.into()))); + let ctx = SessionCtx { + session_id: "s".into(), + config: None, + }; + let event: Envelope = serde_json::from_value(serde_json::json!({ + "v": 1, "source": "debug", "session_id": "s", "event": "e", "ts_ms": 0, "payload": {} + })) + .unwrap(); + (0..count) + .flat_map(|_| lineage.handle(&event, &ctx).unwrap()) + .collect() + } + + fn turn_of(ops: &[SpanOp], span_id: &str) -> Vec> { + ops.iter() + .map(|op| match op { + SpanOp::Insert(row) | SpanOp::Merge(row) => row, + }) + .filter(|row| row.span_id == span_id) + .map(|row| { + row.metadata + .as_ref() + .and_then(|metadata| metadata.get(TURN_SPAN_ID_KEY)) + .and_then(Value::as_str) + .map(str::to_owned) + }) + .collect() + } + + #[test] + fn stamps_turn_and_all_descendants() { + let ops = run(vec![vec![ + SpanOp::Insert(row("root", None, SpanType::Task)), + SpanOp::Insert(turn("turn", "root")), + SpanOp::Insert(row("llm", Some("turn"), SpanType::Llm)), + SpanOp::Insert(row("subagent", Some("turn"), SpanType::Task)), + SpanOp::Insert(row("nested-llm", Some("subagent"), SpanType::Llm)), + SpanOp::Merge(SpanRow { + span_id: "llm".into(), + root_span_id: "root".into(), + ..Default::default() + }), + ]]); + assert_eq!(turn_of(&ops, "root"), [None]); + assert_eq!(turn_of(&ops, "turn"), [Some("turn".into())]); + assert_eq!(turn_of(&ops, "llm"), [Some("turn".into()), None]); + assert_eq!(turn_of(&ops, "nested-llm"), [Some("turn".into())]); + } + + #[test] + fn outermost_turn_wins_for_nested_agent_turns() { + let ops = run(vec![vec![ + SpanOp::Insert(turn("user-turn", "root")), + SpanOp::Insert(row("spawn", Some("user-turn"), SpanType::Tool)), + SpanOp::Insert(turn("subagent-turn", "spawn")), + SpanOp::Insert(row("subagent-llm", Some("subagent-turn"), SpanType::Llm)), + ]]); + assert_eq!(turn_of(&ops, "subagent-turn"), [Some("user-turn".into())]); + assert_eq!(turn_of(&ops, "subagent-llm"), [Some("user-turn".into())]); + } + + #[test] + fn late_parent_resolution_emits_late_merges_for_waiting_descendants() { + let ops = run(vec![ + vec![ + SpanOp::Insert(row("llm", Some("turn"), SpanType::Llm)), + SpanOp::Insert(row("child", Some("llm"), SpanType::Tool)), + ], + vec![SpanOp::Insert(turn("turn", "root"))], + ]); + assert_eq!(turn_of(&ops, "llm"), [None, Some("turn".into())]); + assert_eq!(turn_of(&ops, "child"), [None, Some("turn".into())]); + let late: Vec<_> = ops + .iter() + .filter_map(|op| match op { + SpanOp::Merge(row) => row.late_merge_key.as_deref(), + SpanOp::Insert(_) => None, + }) + .collect(); + assert_eq!(late, [LATE_MERGE_KEY, LATE_MERGE_KEY]); + } + + #[test] + fn spans_outside_any_turn_are_left_untouched() { + let ops = run(vec![vec![ + SpanOp::Insert(row("root", None, SpanType::Task)), + SpanOp::Insert(row("session-compaction", Some("root"), SpanType::Task)), + ]]); + assert_eq!(turn_of(&ops, "session-compaction"), [None]); + let SpanOp::Insert(row) = &ops[1] else { + panic!("expected insert"); + }; + assert!(row.metadata.is_none()); + } +} diff --git a/bt-daemon/tests/braintrust_sink.rs b/bt-daemon/tests/braintrust_sink.rs index 7c684dcb..9de95d5c 100644 --- a/bt-daemon/tests/braintrust_sink.rs +++ b/bt-daemon/tests/braintrust_sink.rs @@ -54,6 +54,7 @@ fn row( metrics: None, error: None, late_merge_key: None, + turn_root: false, tags: 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..cd62bc46 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; @@ -1536,6 +1536,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/pi_translator.rs b/bt-daemon/tests/pi_translator.rs index 9c9683bc..eb95335c 100644 --- a/bt-daemon/tests/pi_translator.rs +++ b/bt-daemon/tests/pi_translator.rs @@ -3,7 +3,7 @@ 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 std::collections::HashMap; @@ -226,7 +226,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 +950,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..dffc334b 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,56 @@ 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) carries the outermost +/// such turn's id, and spans outside any turn carry none. 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 { + let previous = stamped.insert(row.span_id.clone(), turn.to_owned()); + assert!( + previous.as_deref().is_none_or(|previous| previous == turn), + "span {} moved between turns", + row.span_id + ); + } + } + 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() +} From 3e580d17968e76255c5e4a78f068bcd866f43372 Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Wed, 7 Oct 2026 14:44:19 -0700 Subject: [PATCH 2/4] fix(daemon): correct turn_span_id when turns arrive before their parents A turn span emitted before the span that spawned it used to claim itself as a turn permanently, leaving its subtree attributed to the nested turn. Turns are now derived from current knowledge (parent's turn, else self if a turn root) and any change is propagated to affected descendants as late merges. Late merge keys include the turn id so successive corrections to a span are not deduplicated by the delivery ledger. Adds regression tests for the out-of-order subagent case and repeated corrections, and lineage checks for Antigravity, OpenCode (including child sessions) and Pi. Co-Authored-By: Claude Opus 5.5 (1M context) --- bt-daemon/src/translate/turn_lineage.rs | 200 +++++++++++++++------- bt-daemon/tests/antigravity_translator.rs | 3 +- bt-daemon/tests/opencode_translator.rs | 4 +- bt-daemon/tests/pi_translator.rs | 3 +- bt-daemon/tests/support/span_identity.rs | 15 +- 5 files changed, 153 insertions(+), 72 deletions(-) diff --git a/bt-daemon/src/translate/turn_lineage.rs b/bt-daemon/src/translate/turn_lineage.rs index 499b2105..0c5493bb 100644 --- a/bt-daemon/src/translate/turn_lineage.rs +++ b/bt-daemon/src/translate/turn_lineage.rs @@ -5,39 +5,45 @@ //! and on every descendant, including nested subagent activity, so a turn's //! cost can be aggregated with a flat filter instead of a tree walk. //! -//! The outermost turn wins: a subagent's own turns roll up to the user turn -//! that spawned them. A row whose ancestry is not yet known is remembered and -//! re-emitted as a late merge once its parent resolves. +//! A span's turn is always derived from what is known so far: its parent's +//! turn if the parent has one, otherwise itself if it is a turn root. The +//! outermost turn therefore wins, so a subagent's own turns roll up to the user +//! turn that spawned them. Rows can arrive in any order; whenever a span's turn +//! changes, its affected descendants are corrected with late merges. //! //! Translator state is rebuilt by journal replay, which also replays through //! this wrapper, so the lineage maps need no persistence of their own. use super::{AgentTranslator, SessionCtx, SpanOp, SpanRow}; use serde_json::{Map, Value}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; pub const TURN_SPAN_ID_KEY: &str = "turn_span_id"; -const LATE_MERGE_KEY: &str = "turn_lineage"; +const LATE_MERGE_KEY_PREFIX: &str = "turn_lineage"; -struct Unresolved { +struct Identity { root_span_id: String, parent_span_ids: Vec, } pub(crate) struct TurnLineage { inner: Box, - turn_of: HashMap, + identity: HashMap, parent_of: HashMap, - unresolved_children: HashMap>, + children: HashMap>, + turn_roots: HashSet, + turn_of: HashMap, } impl TurnLineage { pub(crate) fn new(inner: Box) -> Self { Self { inner, - turn_of: HashMap::new(), + identity: HashMap::new(), parent_of: HashMap::new(), - unresolved_children: HashMap::new(), + children: HashMap::new(), + turn_roots: HashSet::new(), + turn_of: HashMap::new(), } } @@ -47,70 +53,90 @@ impl TurnLineage { let is_insert = matches!(op, SpanOp::Insert(_)); let (SpanOp::Insert(row) | SpanOp::Merge(row)) = &mut op; let span_id = row.span_id.clone(); - let already_stamped = self.turn_of.contains_key(&span_id); - let turn = self.resolve(row); + self.observe(row); + let previous = self.turn_of.get(&span_id).cloned(); + let turn = self.derive(&span_id); // Merges keep earlier metadata, so only rows that create or replace - // a span, or the first row to resolve it, need the key. - if let Some(turn) = turn.as_ref().filter(|_| is_insert || !already_stamped) { + // a span, or that change its turn, need the key. + if let Some(turn) = turn + .as_ref() + .filter(|turn| is_insert || previous.as_ref() != Some(*turn)) + { set_turn(row, turn); + self.turn_of.insert(span_id.clone(), turn.clone()); } out.push(op); - if let Some(turn) = turn { - self.release_children(&span_id, &turn, &mut out); + if turn.is_some() && turn != previous { + self.correct_descendants(&span_id, &mut out); } } out } - fn resolve(&mut self, row: &SpanRow) -> Option { - if let Some(parent) = row.parent_span_ids.first() { - self.parent_of - .entry(row.span_id.clone()) - .or_insert_with(|| parent.clone()); + fn observe(&mut self, row: &SpanRow) { + if row.turn_root { + self.turn_roots.insert(row.span_id.clone()); } - if let Some(turn) = self.turn_of.get(&row.span_id) { - return Some(turn.clone()); + let identity = self + .identity + .entry(row.span_id.clone()) + .or_insert_with(|| Identity { + root_span_id: row.root_span_id.clone(), + parent_span_ids: Vec::new(), + }); + if identity.parent_span_ids.is_empty() { + identity.parent_span_ids = row.parent_span_ids.clone(); } - let parent = self.parent_of.get(&row.span_id).cloned(); - let inherited = parent - .as_ref() - .and_then(|parent| self.turn_of.get(parent)) - .cloned(); - let Some(turn) = inherited.or_else(|| row.turn_root.then(|| row.span_id.clone())) else { - if let Some(parent) = parent { - self.unresolved_children - .entry(parent) + if let Some(parent) = row.parent_span_ids.first() { + if !self.parent_of.contains_key(&row.span_id) { + self.parent_of.insert(row.span_id.clone(), parent.clone()); + self.children + .entry(parent.clone()) .or_default() - .entry(row.span_id.clone()) - .or_insert_with(|| Unresolved { - root_span_id: row.root_span_id.clone(), - parent_span_ids: row.parent_span_ids.clone(), - }); + .push(row.span_id.clone()); } - return None; - }; - self.turn_of.insert(row.span_id.clone(), turn.clone()); - Some(turn) + } + } + + fn derive(&self, span_id: &str) -> Option { + let inherited = self + .parent_of + .get(span_id) + .and_then(|parent| self.turn_of.get(parent)); + inherited.cloned().or_else(|| { + self.turn_roots + .contains(span_id) + .then(|| span_id.to_owned()) + }) } - fn release_children(&mut self, span_id: &str, turn: &str, out: &mut Vec) { - let mut resolved = vec![span_id.to_owned()]; - while let Some(parent) = resolved.pop() { - let Some(children) = self.unresolved_children.remove(&parent) else { + fn correct_descendants(&mut self, span_id: &str, out: &mut Vec) { + let mut changed = vec![span_id.to_owned()]; + while let Some(parent) = changed.pop() { + let Some(children) = self.children.get(&parent).cloned() else { continue; }; - for (child, unresolved) in children { - self.turn_of.insert(child.clone(), turn.to_owned()); + for child in children { + let Some(turn) = self.derive(&child) else { + continue; + }; + if self.turn_of.get(&child) == Some(&turn) { + continue; + } + self.turn_of.insert(child.clone(), turn.clone()); + let Some(identity) = self.identity.get(&child) else { + continue; + }; let mut row = SpanRow { span_id: child.clone(), - root_span_id: unresolved.root_span_id, - parent_span_ids: unresolved.parent_span_ids, - late_merge_key: Some(LATE_MERGE_KEY.into()), + root_span_id: identity.root_span_id.clone(), + parent_span_ids: identity.parent_span_ids.clone(), + late_merge_key: Some(format!("{LATE_MERGE_KEY_PREFIX}:{turn}")), ..Default::default() }; - set_turn(&mut row, turn); + set_turn(&mut row, &turn); out.push(SpanOp::Merge(row)); - resolved.push(child); + changed.push(child); } } } @@ -221,6 +247,15 @@ mod tests { .collect() } + fn late_merge_keys(ops: &[SpanOp]) -> Vec<&str> { + ops.iter() + .filter_map(|op| match op { + SpanOp::Merge(row) => row.late_merge_key.as_deref(), + SpanOp::Insert(_) => None, + }) + .collect() + } + #[test] fn stamps_turn_and_all_descendants() { let ops = run(vec![vec![ @@ -264,14 +299,59 @@ mod tests { ]); assert_eq!(turn_of(&ops, "llm"), [None, Some("turn".into())]); assert_eq!(turn_of(&ops, "child"), [None, Some("turn".into())]); - let late: Vec<_> = ops - .iter() - .filter_map(|op| match op { - SpanOp::Merge(row) => row.late_merge_key.as_deref(), - SpanOp::Insert(_) => None, - }) - .collect(); - assert_eq!(late, [LATE_MERGE_KEY, LATE_MERGE_KEY]); + assert_eq!( + late_merge_keys(&ops), + ["turn_lineage:turn", "turn_lineage:turn"] + ); + } + + #[test] + fn subagent_turn_emitted_before_its_spawn_span_is_corrected_to_the_user_turn() { + let ops = run(vec![ + vec![SpanOp::Insert(turn("user-turn", "root"))], + vec![ + SpanOp::Insert(turn("subagent-turn", "spawn")), + SpanOp::Insert(row("subagent-llm", Some("subagent-turn"), SpanType::Llm)), + ], + vec![SpanOp::Insert(row( + "spawn", + Some("user-turn"), + SpanType::Tool, + ))], + ]); + assert_eq!( + turn_of(&ops, "subagent-turn"), + [Some("subagent-turn".into()), Some("user-turn".into())] + ); + assert_eq!( + turn_of(&ops, "subagent-llm"), + [Some("subagent-turn".into()), Some("user-turn".into())] + ); + } + + #[test] + fn each_correction_of_a_span_gets_its_own_late_merge_key() { + let ops = run(vec![ + vec![SpanOp::Insert(turn("inner", "middle"))], + vec![SpanOp::Insert(turn("middle", "outer"))], + vec![SpanOp::Insert(turn("outer", "root"))], + ]); + assert_eq!( + turn_of(&ops, "inner"), + [ + Some("inner".into()), + Some("middle".into()), + Some("outer".into()) + ] + ); + assert_eq!( + late_merge_keys(&ops), + [ + "turn_lineage:middle", + "turn_lineage:outer", + "turn_lineage:outer" + ] + ); } #[test] 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/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 eb95335c..5bb5695a 100644 --- a/bt-daemon/tests/pi_translator.rs +++ b/bt-daemon/tests/pi_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, 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() diff --git a/bt-daemon/tests/support/span_identity.rs b/bt-daemon/tests/support/span_identity.rs index dffc334b..b24d2907 100644 --- a/bt-daemon/tests/support/span_identity.rs +++ b/bt-daemon/tests/support/span_identity.rs @@ -48,9 +48,11 @@ 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) carries the outermost -/// such turn's id, and spans outside any turn carry none. Returns how many -/// spans were stamped so callers can assert the fixture exercised lineage. +/// 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(); @@ -74,12 +76,7 @@ pub(crate) fn assert_turn_lineage(ops: &[SpanOp], is_turn: impl Fn(&SpanRow) -> .and_then(|metadata| metadata.get(TURN_SPAN_ID_KEY)) .and_then(Value::as_str); if let Some(turn) = turn { - let previous = stamped.insert(row.span_id.clone(), turn.to_owned()); - assert!( - previous.as_deref().is_none_or(|previous| previous == turn), - "span {} moved between turns", - row.span_id - ); + stamped.insert(row.span_id.clone(), turn.to_owned()); } } for span in &spans { From bb3c98702e7e9135ded0281ec9e2fa8f066b7d27 Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Wed, 7 Oct 2026 15:35:20 -0700 Subject: [PATCH 3/4] refactor(daemon): record turn ownership in translators instead of rebuilding it Translators already track which native turn each operation belongs to in order to parent it, so they now record the owning user turn on each row (SpanRow::turn_span_id). Nested activity records the user turn that spawned it: Claude via Turn/AgentOrigin user_turn_id, Codex via a per-scope owner set at subagent start, and OpenCode via an owner inherited by child sessions. TurnLineage becomes a stateless step that publishes the value as metadata.turn_span_id after each translator's metadata allowlist. This removes the per-session lineage maps, late-merge corrections and emission-order dependence. New translator state is bounded like the existing native-id caches. Co-Authored-By: Claude Opus 5.5 (1M context) --- bt-daemon/src/translate/antigravity.rs | 5 +- bt-daemon/src/translate/claude.rs | 33 ++- bt-daemon/src/translate/codex.rs | 27 +- bt-daemon/src/translate/cursor/mod.rs | 5 +- bt-daemon/src/translate/debug.rs | 4 +- bt-daemon/src/translate/grok.rs | 5 +- bt-daemon/src/translate/mod.rs | 7 +- bt-daemon/src/translate/opencode.rs | 29 +- bt-daemon/src/translate/pi.rs | 9 +- bt-daemon/src/translate/turn_lineage.rs | 347 +++--------------------- bt-daemon/tests/braintrust_sink.rs | 2 +- 11 files changed, 150 insertions(+), 323 deletions(-) diff --git a/bt-daemon/src/translate/antigravity.rs b/bt-daemon/src/translate/antigravity.rs index 1595e3ce..fb9da2a4 100644 --- a/bt-daemon/src/translate/antigravity.rs +++ b/bt-daemon/src/translate/antigravity.rs @@ -224,7 +224,7 @@ impl AntigravityTranslator { start_ms: Some(ts_ms), input: nonempty_value(input), metadata: Some(json!({"turn_number": self.turn_count})), - turn_root: true, + turn_span_id: Some(span_id.clone()), ..Default::default() })); self.turn = Some(Turn { @@ -275,6 +275,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 +335,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 +396,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 d1e1a3f3..daa67574 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(), @@ -524,9 +528,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_root: true, + 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 +648,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 +679,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 +694,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 +737,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 +790,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 +1058,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 +1093,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 9337b72a..e1c2d289 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,6 +841,8 @@ impl CodexTranslator { return; } let span_id = ids::span_id(&self.session_id, &format!("turn:{turn_id}")); + let owner = scope.turn_owner(&span_id); + 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(), @@ -830,7 +851,7 @@ impl CodexTranslator { span_type: SpanType::Task, start_ms: Some(ts), metadata: Some(json!({ "turn_id": turn_id, "model": scope.model })), - turn_root: true, + turn_span_id: Some(owner), ..Default::default() })); scope.open_turns.push(OpenTurn { @@ -927,6 +948,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: Some(scope.turn_owner(&turn_span)), parent_span_ids: vec![turn_span], name, span_type: SpanType::Llm, @@ -1090,6 +1112,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: Some(scope.turn_owner(&turn_span)), parent_span_ids: vec![turn_span], name, span_type: SpanType::Tool, @@ -1335,6 +1358,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: Some(scope.turn_owner(&turn_span)), parent_span_ids: vec![turn_span.clone()], name: name.clone(), span_type: SpanType::Llm, @@ -1520,6 +1544,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 0d6b735e..22446d47 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() @@ -361,7 +364,7 @@ impl CursorTranslator { row.metadata = Some( json!({"turn_boundary_source":origin, "start_time_estimated":origin != "beforeSubmitPrompt", "generation_id":generation}), ); - row.turn_root = true; + 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 4bac2474..cef6b585 100644 --- a/bt-daemon/src/translate/debug.rs +++ b/bt-daemon/src/translate/debug.rs @@ -62,7 +62,7 @@ impl AgentTranslator for DebugTranslator { metrics: None, error: None, late_merge_key: None, - turn_root: false, + turn_span_id: None, tags: None, })); } @@ -84,7 +84,7 @@ impl AgentTranslator for DebugTranslator { metrics: None, error: None, late_merge_key: None, - turn_root: false, + turn_span_id: None, tags: None, })); diff --git a/bt-daemon/src/translate/grok.rs b/bt-daemon/src/translate/grok.rs index 9b122073..8d8432dc 100644 --- a/bt-daemon/src/translate/grok.rs +++ b/bt-daemon/src/translate/grok.rs @@ -488,7 +488,7 @@ impl GrokTranslator { start_ms: Some(ts_ms), input, metadata: Some(json!({"transcript_turn_key": turn_key})), - turn_root: true, + turn_span_id: Some(span_id.clone()), ..Default::default() })); self.current_turn = Some(OpenTurn { @@ -583,6 +583,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 +665,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 +717,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 c468cf9b..02dda3e5 100644 --- a/bt-daemon/src/translate/mod.rs +++ b/bt-daemon/src/translate/mod.rs @@ -80,11 +80,12 @@ pub struct SpanRow { #[serde(skip)] #[doc(hidden)] pub late_merge_key: Option, - /// Translator-only marker for the span that opens a user turn. - /// [`turn_lineage`] stamps its id onto the turn and every descendant. + /// 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_root: bool, + 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>, diff --git a/bt-daemon/src/translate/opencode.rs b/bt-daemon/src/translate/opencode.rs index 95f6ea6f..5e8340fd 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), @@ -755,7 +771,7 @@ impl OpenCodeTranslator { metadata: Some( json!({"turn_number":state.turn_number,"model":model,"loaded_skill_names":skills}), ), - turn_root: true, + turn_span_id: Some(state.owner_turn_span_id.clone().unwrap_or(turn_id)), ..Default::default() })); ops @@ -954,6 +970,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 +1022,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 +1099,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 +1226,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 6c09bee9..0487eb23 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), @@ -616,7 +618,7 @@ impl PiTranslator { start_ms: Some(ts), input: Some(input), metadata: Some(metadata), - turn_root: true, + turn_span_id: Some(id), ..Default::default() })); ops @@ -738,6 +740,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 +798,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 +838,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 index 0c5493bb..41e35df0 100644 --- a/bt-daemon/src/translate/turn_lineage.rs +++ b/bt-daemon/src/translate/turn_lineage.rs @@ -1,153 +1,46 @@ -//! Stamps every span with the user turn that owns it. +//! Publishes the user turn that owns each span. //! -//! Translators only mark the span that opens a turn ([`SpanRow::turn_root`]). -//! This wrapper copies that span's id into `metadata.turn_span_id` on the turn -//! and on every descendant, including nested subagent activity, so a turn's -//! cost can be aggregated with a flat filter instead of a tree walk. +//! 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. //! -//! A span's turn is always derived from what is known so far: its parent's -//! turn if the parent has one, otherwise itself if it is a turn root. The -//! outermost turn therefore wins, so a subagent's own turns roll up to the user -//! turn that spawned them. Rows can arrive in any order; whenever a span's turn -//! changes, its affected descendants are corrected with late merges. -//! -//! Translator state is rebuilt by journal replay, which also replays through -//! this wrapper, so the lineage maps need no persistence of their own. +//! 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}; -use std::collections::{HashMap, HashSet}; pub const TURN_SPAN_ID_KEY: &str = "turn_span_id"; -const LATE_MERGE_KEY_PREFIX: &str = "turn_lineage"; - -struct Identity { - root_span_id: String, - parent_span_ids: Vec, -} pub(crate) struct TurnLineage { inner: Box, - identity: HashMap, - parent_of: HashMap, - children: HashMap>, - turn_roots: HashSet, - turn_of: HashMap, } impl TurnLineage { pub(crate) fn new(inner: Box) -> Self { - Self { - inner, - identity: HashMap::new(), - parent_of: HashMap::new(), - children: HashMap::new(), - turn_roots: HashSet::new(), - turn_of: HashMap::new(), - } - } - - fn stamp(&mut self, ops: Vec) -> Vec { - let mut out = Vec::with_capacity(ops.len()); - for mut op in ops { - let is_insert = matches!(op, SpanOp::Insert(_)); - let (SpanOp::Insert(row) | SpanOp::Merge(row)) = &mut op; - let span_id = row.span_id.clone(); - self.observe(row); - let previous = self.turn_of.get(&span_id).cloned(); - let turn = self.derive(&span_id); - // Merges keep earlier metadata, so only rows that create or replace - // a span, or that change its turn, need the key. - if let Some(turn) = turn - .as_ref() - .filter(|turn| is_insert || previous.as_ref() != Some(*turn)) - { - set_turn(row, turn); - self.turn_of.insert(span_id.clone(), turn.clone()); - } - out.push(op); - if turn.is_some() && turn != previous { - self.correct_descendants(&span_id, &mut out); - } - } - out - } - - fn observe(&mut self, row: &SpanRow) { - if row.turn_root { - self.turn_roots.insert(row.span_id.clone()); - } - let identity = self - .identity - .entry(row.span_id.clone()) - .or_insert_with(|| Identity { - root_span_id: row.root_span_id.clone(), - parent_span_ids: Vec::new(), - }); - if identity.parent_span_ids.is_empty() { - identity.parent_span_ids = row.parent_span_ids.clone(); - } - if let Some(parent) = row.parent_span_ids.first() { - if !self.parent_of.contains_key(&row.span_id) { - self.parent_of.insert(row.span_id.clone(), parent.clone()); - self.children - .entry(parent.clone()) - .or_default() - .push(row.span_id.clone()); - } - } - } - - fn derive(&self, span_id: &str) -> Option { - let inherited = self - .parent_of - .get(span_id) - .and_then(|parent| self.turn_of.get(parent)); - inherited.cloned().or_else(|| { - self.turn_roots - .contains(span_id) - .then(|| span_id.to_owned()) - }) + Self { inner } } +} - fn correct_descendants(&mut self, span_id: &str, out: &mut Vec) { - let mut changed = vec![span_id.to_owned()]; - while let Some(parent) = changed.pop() { - let Some(children) = self.children.get(&parent).cloned() else { - continue; - }; - for child in children { - let Some(turn) = self.derive(&child) else { - continue; - }; - if self.turn_of.get(&child) == Some(&turn) { - continue; - } - self.turn_of.insert(child.clone(), turn.clone()); - let Some(identity) = self.identity.get(&child) else { - continue; - }; - let mut row = SpanRow { - span_id: child.clone(), - root_span_id: identity.root_span_id.clone(), - parent_span_ids: identity.parent_span_ids.clone(), - late_merge_key: Some(format!("{LATE_MERGE_KEY_PREFIX}:{turn}")), - ..Default::default() - }; - set_turn(&mut row, &turn); - out.push(SpanOp::Merge(row)); - changed.push(child); - } - } +fn stamp(mut ops: Vec) -> Vec { + for op in &mut ops { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + publish_turn(row); } + ops } -fn set_turn(row: &mut SpanRow, turn: &str) { +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.to_owned())); + metadata.insert(TURN_SPAN_ID_KEY.into(), Value::String(turn)); } } @@ -157,213 +50,59 @@ impl AgentTranslator for TurnLineage { event: &crate::wire::Envelope, ctx: &SessionCtx, ) -> anyhow::Result> { - let ops = self.inner.handle(event, ctx)?; - Ok(self.stamp(ops)) + Ok(stamp(self.inner.handle(event, ctx)?)) } fn drain_pending(&mut self, ctx: &SessionCtx) -> anyhow::Result>> { - let ops = self.inner.drain_pending(ctx)?; - Ok(ops.map(|ops| self.stamp(ops))) + Ok(self.inner.drain_pending(ctx)?.map(stamp)) } fn checkpoint(&mut self, ctx: &SessionCtx) -> anyhow::Result> { - let ops = self.inner.checkpoint(ctx)?; - Ok(self.stamp(ops)) + Ok(stamp(self.inner.checkpoint(ctx)?)) } fn finalize(&mut self, ctx: &SessionCtx) -> anyhow::Result> { - let ops = self.inner.finalize(ctx)?; - Ok(self.stamp(ops)) + Ok(stamp(self.inner.finalize(ctx)?)) } fn flush(&mut self, ctx: &SessionCtx) -> anyhow::Result> { - let ops = self.inner.flush(ctx)?; - Ok(self.stamp(ops)) + Ok(stamp(self.inner.flush(ctx)?)) } } #[cfg(test)] mod tests { use super::*; - use crate::translate::SpanType; - use crate::wire::Envelope; - use std::collections::VecDeque; + use serde_json::json; - struct Scripted(VecDeque>); - - impl AgentTranslator for Scripted { - fn handle(&mut self, _event: &Envelope, _ctx: &SessionCtx) -> anyhow::Result> { - Ok(self.0.pop_front().unwrap_or_default()) - } - } - - fn row(span_id: &str, parent: Option<&str>, span_type: SpanType) -> SpanRow { - SpanRow { - span_id: span_id.into(), + fn row(turn: Option<&str>, metadata: Option) -> SpanOp { + SpanOp::Insert(SpanRow { + span_id: "span".into(), root_span_id: "root".into(), - parent_span_ids: parent.map(|p| vec![p.to_owned()]).unwrap_or_default(), - name: span_id.into(), - span_type, + turn_span_id: turn.map(str::to_owned), + metadata, ..Default::default() - } - } - - fn turn(span_id: &str, parent: &str) -> SpanRow { - SpanRow { - turn_root: true, - ..row(span_id, Some(parent), SpanType::Task) - } - } - - fn run(batches: Vec>) -> Vec { - let count = batches.len(); - let mut lineage = TurnLineage::new(Box::new(Scripted(batches.into()))); - let ctx = SessionCtx { - session_id: "s".into(), - config: None, - }; - let event: Envelope = serde_json::from_value(serde_json::json!({ - "v": 1, "source": "debug", "session_id": "s", "event": "e", "ts_ms": 0, "payload": {} - })) - .unwrap(); - (0..count) - .flat_map(|_| lineage.handle(&event, &ctx).unwrap()) - .collect() - } - - fn turn_of(ops: &[SpanOp], span_id: &str) -> Vec> { - ops.iter() - .map(|op| match op { - SpanOp::Insert(row) | SpanOp::Merge(row) => row, - }) - .filter(|row| row.span_id == span_id) - .map(|row| { - row.metadata - .as_ref() - .and_then(|metadata| metadata.get(TURN_SPAN_ID_KEY)) - .and_then(Value::as_str) - .map(str::to_owned) - }) - .collect() - } - - fn late_merge_keys(ops: &[SpanOp]) -> Vec<&str> { - ops.iter() - .filter_map(|op| match op { - SpanOp::Merge(row) => row.late_merge_key.as_deref(), - SpanOp::Insert(_) => None, - }) - .collect() - } - - #[test] - fn stamps_turn_and_all_descendants() { - let ops = run(vec![vec![ - SpanOp::Insert(row("root", None, SpanType::Task)), - SpanOp::Insert(turn("turn", "root")), - SpanOp::Insert(row("llm", Some("turn"), SpanType::Llm)), - SpanOp::Insert(row("subagent", Some("turn"), SpanType::Task)), - SpanOp::Insert(row("nested-llm", Some("subagent"), SpanType::Llm)), - SpanOp::Merge(SpanRow { - span_id: "llm".into(), - root_span_id: "root".into(), - ..Default::default() - }), - ]]); - assert_eq!(turn_of(&ops, "root"), [None]); - assert_eq!(turn_of(&ops, "turn"), [Some("turn".into())]); - assert_eq!(turn_of(&ops, "llm"), [Some("turn".into()), None]); - assert_eq!(turn_of(&ops, "nested-llm"), [Some("turn".into())]); - } - - #[test] - fn outermost_turn_wins_for_nested_agent_turns() { - let ops = run(vec![vec![ - SpanOp::Insert(turn("user-turn", "root")), - SpanOp::Insert(row("spawn", Some("user-turn"), SpanType::Tool)), - SpanOp::Insert(turn("subagent-turn", "spawn")), - SpanOp::Insert(row("subagent-llm", Some("subagent-turn"), SpanType::Llm)), - ]]); - assert_eq!(turn_of(&ops, "subagent-turn"), [Some("user-turn".into())]); - assert_eq!(turn_of(&ops, "subagent-llm"), [Some("user-turn".into())]); - } - - #[test] - fn late_parent_resolution_emits_late_merges_for_waiting_descendants() { - let ops = run(vec![ - vec![ - SpanOp::Insert(row("llm", Some("turn"), SpanType::Llm)), - SpanOp::Insert(row("child", Some("llm"), SpanType::Tool)), - ], - vec![SpanOp::Insert(turn("turn", "root"))], - ]); - assert_eq!(turn_of(&ops, "llm"), [None, Some("turn".into())]); - assert_eq!(turn_of(&ops, "child"), [None, Some("turn".into())]); - assert_eq!( - late_merge_keys(&ops), - ["turn_lineage:turn", "turn_lineage:turn"] - ); + }) } - #[test] - fn subagent_turn_emitted_before_its_spawn_span_is_corrected_to_the_user_turn() { - let ops = run(vec![ - vec![SpanOp::Insert(turn("user-turn", "root"))], - vec![ - SpanOp::Insert(turn("subagent-turn", "spawn")), - SpanOp::Insert(row("subagent-llm", Some("subagent-turn"), SpanType::Llm)), - ], - vec![SpanOp::Insert(row( - "spawn", - Some("user-turn"), - SpanType::Tool, - ))], - ]); - assert_eq!( - turn_of(&ops, "subagent-turn"), - [Some("subagent-turn".into()), Some("user-turn".into())] - ); - assert_eq!( - turn_of(&ops, "subagent-llm"), - [Some("subagent-turn".into()), Some("user-turn".into())] - ); + fn metadata(op: &SpanOp) -> Option<&Value> { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + row.metadata.as_ref() } #[test] - fn each_correction_of_a_span_gets_its_own_late_merge_key() { - let ops = run(vec![ - vec![SpanOp::Insert(turn("inner", "middle"))], - vec![SpanOp::Insert(turn("middle", "outer"))], - vec![SpanOp::Insert(turn("outer", "root"))], - ]); - assert_eq!( - turn_of(&ops, "inner"), - [ - Some("inner".into()), - Some("middle".into()), - Some("outer".into()) - ] - ); + fn publishes_the_translator_recorded_turn_alongside_existing_metadata() { + let ops = stamp(vec![row(Some("turn"), Some(json!({"model": "m"})))]); assert_eq!( - late_merge_keys(&ops), - [ - "turn_lineage:middle", - "turn_lineage:outer", - "turn_lineage:outer" - ] + metadata(&ops[0]), + Some(&json!({"model": "m", "turn_span_id": "turn"})) ); } #[test] - fn spans_outside_any_turn_are_left_untouched() { - let ops = run(vec![vec![ - SpanOp::Insert(row("root", None, SpanType::Task)), - SpanOp::Insert(row("session-compaction", Some("root"), SpanType::Task)), - ]]); - assert_eq!(turn_of(&ops, "session-compaction"), [None]); - let SpanOp::Insert(row) = &ops[1] else { - panic!("expected insert"); - }; - assert!(row.metadata.is_none()); + 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/braintrust_sink.rs b/bt-daemon/tests/braintrust_sink.rs index 9de95d5c..96824a88 100644 --- a/bt-daemon/tests/braintrust_sink.rs +++ b/bt-daemon/tests/braintrust_sink.rs @@ -54,7 +54,7 @@ fn row( metrics: None, error: None, late_merge_key: None, - turn_root: false, + turn_span_id: None, tags: None, } } From 16c573d0e8af379e698121f8cae24983b75b99fe Mon Sep 17 00:00:00 2001 From: max-braintrust Date: Thu, 8 Oct 2026 13:57:46 -0700 Subject: [PATCH 4/4] fix(daemon): exclude Codex maintenance compactions from turn ownership Turns that native compaction hooks identify as maintenance (is_turn = false) no longer own themselves; in the main scope they and their synthetic compaction llm carry no turn_span_id, so per-turn cost rollups do not charge a fake turn. Child rows now follow their turn's recorded owner, keeping a span's turn_span_id equal to its parent's. Compaction discovered only during catch-up still opens as a turn, matching is_turn. Co-Authored-By: Claude Opus 5.5 (1M context) --- bt-daemon/src/translate/codex.rs | 27 +++++++---- bt-daemon/tests/codex_translator.rs | 75 +++++++++++++++++++++++++++++ 2 files changed, 93 insertions(+), 9 deletions(-) diff --git a/bt-daemon/src/translate/codex.rs b/bt-daemon/src/translate/codex.rs index 0691c589..3585e099 100644 --- a/bt-daemon/src/translate/codex.rs +++ b/bt-daemon/src/translate/codex.rs @@ -841,20 +841,29 @@ impl CodexTranslator { return; } let span_id = ids::span_id(&self.session_id, &format!("turn:{turn_id}")); - let owner = scope.turn_owner(&span_id); - self.turn_owners.insert(span_id.clone(), owner.clone()); + // 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: Some(owner), + turn_span_id: owner, ..Default::default() })); scope.open_turns.push(OpenTurn { @@ -951,7 +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: Some(scope.turn_owner(&turn_span)), + turn_span_id: self.turn_owners.get(&turn_span).cloned(), parent_span_ids: vec![turn_span], name, span_type: SpanType::Llm, @@ -1115,7 +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: Some(scope.turn_owner(&turn_span)), + turn_span_id: self.turn_owners.get(&turn_span).cloned(), parent_span_ids: vec![turn_span], name, span_type: SpanType::Tool, @@ -1361,7 +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: Some(scope.turn_owner(&turn_span)), + turn_span_id: self.turn_owners.get(&turn_span).cloned(), parent_span_ids: vec![turn_span.clone()], name: name.clone(), span_type: SpanType::Llm, diff --git a/bt-daemon/tests/codex_translator.rs b/bt-daemon/tests/codex_translator.rs index cd62bc46..9844ce34 100644 --- a/bt-daemon/tests/codex_translator.rs +++ b/bt-daemon/tests/codex_translator.rs @@ -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();