Add turn_span_id

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,
 };

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)),

bt-daemon/src/translate/claude.rs

@@ -264,6 +264,8 @@ struct ClaudeTranslator {
     continuation_count: u32,
     last_user_turn_id: Option<String>,
     agent_origins: RecentMap<String, AgentOrigin>,
+    /// User turn owning each turn or subagent span that can parent other spans.
+    span_owners: RecentMap<String, String>,
     user_turn_ends: RecentMap<String, i64>,
     prompt_turns: RecentMap<String, Arc<String>>,
     pending_prompts: VecDeque<PendingPrompt>,
@@ -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<String> {
+        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));
             }
         }
     }

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<String>,
     agent_type: Option<String>,
     spawning_turn_span_id: Option<String>,
+    /// User turn that owns this subagent's work; `None` for the main scope.
+    owner_turn_span_id: Option<String>,
     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<String, Scope>,
     spawn_turn_by_call_id: RecentMap<String, String>,
     spawn_turn_by_agent_id: RecentMap<String, String>,
+    turn_owners: RecentMap<String, String>,
     compaction_trigger_by_turn: RecentMap<String, String>,
     compaction_spans: RecentSet<String>,
     pending: Option<PendingWork>,
@@ -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,
         }
     }

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,

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,

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)

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<String>,
+    /// 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<String>,
     /// Labels for filtering in Braintrust (e.g. `compaction`, `permission-request`).
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub tags: Option<Vec<String>>,
@@ -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