diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 08d01e05..d1123eef 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -720,12 +720,7 @@ impl SessionActor { while let Some(ops) = next { if !ops.is_empty() { if mode.observes_correlation() { - let changed = self.correlation.observe_ops( - &self.correlation_key, - &self.route, - config, - &ops, - ); + let changed = self.observe_ops(&ops, config); correlation_changed |= changed; if changed { self.persist_correlation_if_changed(true).await; @@ -855,9 +850,7 @@ impl SessionActor { }; while let Some(ops) = next { if !ops.is_empty() { - let _ = - self.correlation - .observe_ops(&self.correlation_key, &self.route, config, &ops); + let _ = self.observe_ops(&ops, config); } next = match translator.drain_pending(ctx) { Ok(next) => next, @@ -900,6 +893,17 @@ impl SessionActor { delivered } + fn observe_ops( + &self, + ops: &[crate::translate::SpanOp], + config: &crate::wire::SessionConfig, + ) -> bool { + let mut routed = ops.to_vec(); + crate::pipeline::prepare::Prepare::new(&self.source).apply(&mut routed, config); + self.correlation + .observe_ops(&self.correlation_key, &self.route, config, &routed) + } + async fn persist_correlation_if_changed(&self, changed: bool) { if !changed { return; diff --git a/bt-daemon/src/pipeline/mod.rs b/bt-daemon/src/pipeline/mod.rs index 3adefa05..76d8ab52 100644 --- a/bt-daemon/src/pipeline/mod.rs +++ b/bt-daemon/src/pipeline/mod.rs @@ -2,6 +2,7 @@ pub(crate) mod delivery; pub(crate) mod plugins; +pub(crate) mod prepare; pub(crate) mod routing; pub(crate) use delivery::{Completion, Delivery}; diff --git a/bt-daemon/src/pipeline/prepare.rs b/bt-daemon/src/pipeline/prepare.rs new file mode 100644 index 00000000..a29e4f7e --- /dev/null +++ b/bt-daemon/src/pipeline/prepare.rs @@ -0,0 +1,199 @@ +//! Route overlay is independent of native translation and never mutates shared ops. +use crate::{ + translate::{SpanOp, SpanType}, + wire::SessionConfig, +}; +use std::collections::HashSet; + +pub(crate) struct Prepare { + source: String, + roots: HashSet, +} + +impl Prepare { + pub fn new(source: &str) -> Self { + Self { + source: source.into(), + roots: HashSet::new(), + } + } + + pub fn apply(&mut self, ops: &mut [SpanOp], config: &SessionConfig) { + let (parent, root) = config.attached_span_ids(); + let root = root.or_else(|| parent.clone()); + for op in ops { + let insert = matches!(op, SpanOp::Insert(_)); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + let is_root = self.roots.contains(&row.span_id) + || (row.span_type == SpanType::Task + && row.parent_span_ids.is_empty() + && row.span_id == row.root_span_id); + if is_root { + self.roots.insert(row.span_id.clone()); + } + // Native subagent identity is independent of route attachment. + let is_subagent = row.span_type == SpanType::Task + && !row.is_turn + && row.metadata.as_ref().is_some_and(|metadata| { + metadata + .get("is_subagent") + .and_then(|value| value.as_bool()) + == Some(true) + || (self.source == "codex" && metadata.get("agent_id").is_some()) + }); + if let Some(root) = &root { + row.root_span_id = root.clone(); + } + if is_root { + if let Some(parent) = &parent { + row.parent_span_ids = vec![parent.clone()]; + } + } + if insert && (is_root || (self.source == "opencode" && is_subagent)) { + let metadata = row.metadata.get_or_insert_with(|| serde_json::json!({})); + if let Some(metadata) = metadata.as_object_mut() { + if let Some(extra) = config + .additional_metadata + .as_ref() + .and_then(|v| v.as_object()) + { + for (key, value) in extra { + if !key.starts_with("_bt_") { + metadata.entry(key.clone()).or_insert_with(|| value.clone()); + } + } + } + if self.source == "codex" { + if let Some(project) = config.project_name() { + metadata.insert("project".into(), project.into()); + } + } + } + } + if insert && (is_root || is_subagent) && !config.tags.is_empty() { + row.tags = Some(config.tags.clone()); + } + } + } +} + +/// Retain the public translator/import API: route settings are applied here, +/// outside the native translators, while the daemon uses canonical output directly. +pub(crate) struct ConfiguredTranslator { + inner: Box, + prepare: Prepare, +} + +impl ConfiguredTranslator { + pub fn new(inner: Box, source: &str) -> Self { + Self { + inner, + prepare: Prepare::new(source), + } + } + fn apply(&mut self, mut ops: Vec, ctx: &crate::translate::SessionCtx) -> Vec { + if let Some(config) = &ctx.config { + self.prepare.apply(&mut ops, config); + } + ops + } +} + +impl crate::translate::AgentTranslator for ConfiguredTranslator { + fn handle( + &mut self, + env: &crate::wire::Envelope, + ctx: &crate::translate::SessionCtx, + ) -> anyhow::Result> { + let ops = self.inner.handle(env, ctx)?; + Ok(self.apply(ops, ctx)) + } + fn drain_pending( + &mut self, + ctx: &crate::translate::SessionCtx, + ) -> anyhow::Result>> { + let ops = self.inner.drain_pending(ctx)?; + Ok(ops.map(|ops| self.apply(ops, ctx))) + } + fn checkpoint(&mut self, ctx: &crate::translate::SessionCtx) -> anyhow::Result> { + let ops = self.inner.checkpoint(ctx)?; + Ok(self.apply(ops, ctx)) + } + fn finalize(&mut self, ctx: &crate::translate::SessionCtx) -> anyhow::Result> { + let ops = self.inner.finalize(ctx)?; + Ok(self.apply(ops, ctx)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + translate::{Registry, SessionCtx}, + wire::{BackendAuth, Envelope, SessionRoute}, + }; + + #[test] + fn canonical_translation_does_not_depend_on_route_customization() { + let registry = Registry::default_agents(); + let env = Envelope { + source: "claude-code".into(), + source_version: Some("test".into()), + plugin_version: Some("test".into()), + session_id: "neutral".into(), + event: "UserPromptSubmit".into(), + ts_ms: 1, + managed_run_id: None, + capture: None, + route: None, + config: None, + payload: serde_json::json!({"cwd": "/workspace", "session_id": "neutral", "prompt": "hello"}), + }; + let plain = SessionCtx { + session_id: "neutral".into(), + config: None, + }; + let config = SessionRoute { + tags: vec!["route-tag".into()], + additional_metadata: Some( + serde_json::json!({"custom": true, "source": "wrong", "_bt_private": true}), + ), + ..SessionRoute::default() + } + .with_auth(BackendAuth { + token: "not-persisted".into(), + api_url: None, + app_url: None, + org_name: None, + org_id: None, + }); + let customized = SessionCtx { + session_id: "neutral".into(), + config: Some(config.clone()), + }; + let mut a = registry + .create_checked_with_session_namespace("claude-code", "neutral") + .unwrap(); + let mut b = registry + .create_checked_with_session_namespace("claude-code", "neutral") + .unwrap(); + let canonical = a.handle(&env, &plain).unwrap(); + assert!(!canonical.is_empty()); + let other = b.handle(&env, &customized).unwrap(); + assert_eq!( + serde_json::to_value(&canonical).unwrap(), + serde_json::to_value(other).unwrap() + ); + let mut routed = canonical.clone(); + Prepare::new("claude-code").apply(&mut routed, &config); + let (SpanOp::Insert(root) | SpanOp::Merge(root)) = &routed[0]; + assert_eq!(root.tags.as_ref().unwrap(), &["route-tag"]); + let metadata = root.metadata.as_ref().unwrap(); + assert_eq!(metadata["custom"], true); + assert_eq!(metadata["source"], "claude-code"); + assert!(metadata.get("_bt_private").is_none()); + let (SpanOp::Insert(root) | SpanOp::Merge(root)) = &canonical[0]; + assert!(root.tags.is_none()); + assert!(root.metadata.as_ref().unwrap().get("custom").is_none()); + } +} diff --git a/bt-daemon/src/pipeline/routing.rs b/bt-daemon/src/pipeline/routing.rs index ffb65498..cb0a992c 100644 --- a/bt-daemon/src/pipeline/routing.rs +++ b/bt-daemon/src/pipeline/routing.rs @@ -68,6 +68,7 @@ impl RoutingSink { ) -> Self { let (tx, mut rx) = mpsc::channel::>(16); tokio::spawn(async move { + let mut prepare = super::prepare::Prepare::new(&source); while let Some(delivery) = InputPort::receive(&mut rx).await { let request = delivery.value(); sink.configure(&request.config); @@ -77,6 +78,8 @@ impl RoutingSink { ); let result = match &request.operation { Operation::Emit(ops) => { + let mut ops = ops.clone(); + prepare.apply(&mut ops, &request.config); let mut stage = Plugins { paths: request.config.span_plugins.clone(), source: source.clone(), @@ -85,7 +88,7 @@ impl RoutingSink { let mut prepared = Vec::with_capacity(ops.len()); let mut failure = None; let mut report = false; - for op in ops { + for op in &ops { match stage.process(op).await { PluginOutcome::Prepared(op) => prepared.push(*op), PluginOutcome::Filtered => {} diff --git a/bt-daemon/src/translate/antigravity.rs b/bt-daemon/src/translate/antigravity.rs index d825480a..d40d5a24 100644 --- a/bt-daemon/src/translate/antigravity.rs +++ b/bt-daemon/src/translate/antigravity.rs @@ -8,8 +8,7 @@ use super::git::GitMetadataCache; use super::tool::{with_tool_approval, without_nulls, ToolApproval}; use super::{ - local_username, root_tags, AgentTranslator, SessionCtx, SpanOp, SpanRow, SpanType, - TranslatorFactory, + local_username, AgentTranslator, SessionCtx, SpanOp, SpanRow, SpanType, TranslatorFactory, }; use crate::ids; use crate::wire::Envelope; @@ -95,28 +94,13 @@ impl AntigravityTranslator { return; } self.root_open = true; - let (parent_span_id, external_root_span_id) = ctx - .config - .as_ref() - .map(|config| config.attached_span_ids()) - .unwrap_or_default(); - if let Some(external_root) = external_root_span_id { - self.root_span_id = external_root; - } - self.session_parent_span_ids = parent_span_id.into_iter().collect(); - let workspace = event .payload .get("workspacePaths") .and_then(Value::as_array) .and_then(|paths| paths.first()) .and_then(Value::as_str); - let mut metadata = ctx - .config - .as_ref() - .and_then(|config| config.additional_metadata.clone()) - .and_then(|value| value.as_object().cloned()) - .unwrap_or_default(); + let mut metadata = serde_json::Map::new(); metadata.retain(|key, _| !key.starts_with("_bt_")); metadata.insert("session_id".into(), json!(ctx.session_id)); metadata.insert("conversation_id".into(), json!(ctx.session_id)); @@ -149,7 +133,7 @@ impl AntigravityTranslator { span_type: SpanType::Task, start_ms: Some(event.ts_ms), metadata: Some(Value::Object(metadata)), - tags: root_tags(ctx), + tags: None, ..Default::default() })); } diff --git a/bt-daemon/src/translate/claude.rs b/bt-daemon/src/translate/claude.rs index c6697a90..f3edb0f3 100644 --- a/bt-daemon/src/translate/claude.rs +++ b/bt-daemon/src/translate/claude.rs @@ -10,8 +10,7 @@ use super::git::GitMetadataCache; use super::recent::{RecentMap, RecentSet, RECENT_ID_CAPACITY}; use super::tool::{add_tool_approval, explicit_skill_metadata, nonempty_error_text, ToolApproval}; use super::{ - local_username, root_tags, AgentTranslator, SessionCtx, SpanOp, SpanRow, SpanType, - TranslatorFactory, + local_username, AgentTranslator, SessionCtx, SpanOp, SpanRow, SpanType, TranslatorFactory, }; use crate::ids; use crate::wire::Envelope; @@ -355,23 +354,9 @@ impl ClaudeTranslator { return; } self.root_open = true; - let (parent_span_id, root_span_id) = ctx - .config - .as_ref() - .map(|config| config.attached_span_ids()) - .unwrap_or_default(); - if let Some(root) = root_span_id { - self.root_span_id = root.clone(); - } - self.session_parent_span_ids = parent_span_id.into_iter().collect(); let cwd = hook.cwd.clone().unwrap_or_default(); let workspace = basename(&cwd); - let mut metadata = ctx - .config - .as_ref() - .and_then(|c| c.additional_metadata.clone()) - .and_then(|v| v.as_object().cloned()) - .unwrap_or_default(); + let mut metadata = serde_json::Map::new(); // Internal routing settings must never appear as user metadata. metadata.retain(|key, _| !key.starts_with("_bt_")); metadata.insert("session_id".into(), json!(ctx.session_id)); @@ -412,7 +397,7 @@ impl ClaudeTranslator { start_ms: Some(event.ts_ms), input: Some(json!(format!("Session: {workspace}"))), metadata: Some(Value::Object(metadata)), - tags: root_tags(ctx), + tags: None, ..Default::default() })); } diff --git a/bt-daemon/src/translate/codex.rs b/bt-daemon/src/translate/codex.rs index daef123e..8b654770 100644 --- a/bt-daemon/src/translate/codex.rs +++ b/bt-daemon/src/translate/codex.rs @@ -120,9 +120,6 @@ impl TranslatorFactory for crate::agents::Codex { session_source: None, permission_mode: None, root_cwd: None, - project: None, - additional_metadata: Map::new(), - tags: Vec::new(), main_path: None, // The main scope is created lazily once we learn its transcript path. scopes: HashMap::new(), @@ -247,9 +244,6 @@ struct CodexTranslator { session_source: Option, permission_mode: Option, root_cwd: Option, - project: Option, - additional_metadata: Map, - tags: Vec, main_path: Option, scopes: HashMap, spawn_turn_by_call_id: RecentMap, @@ -262,7 +256,7 @@ struct CodexTranslator { } impl AgentTranslator for CodexTranslator { - fn handle(&mut self, event: &Envelope, ctx: &SessionCtx) -> anyhow::Result> { + fn handle(&mut self, event: &Envelope, _ctx: &SessionCtx) -> anyhow::Result> { anyhow::ensure!( self.pending.is_none(), "Codex translator has pending catch-up work; drain it before handling another event" @@ -270,23 +264,6 @@ impl AgentTranslator for CodexTranslator { let payload = &event.payload; let mut ops = Vec::new(); - if let Some(config) = &ctx.config { - let attached = config.attached_span_ids(); - self.external_parent_span_id = attached.0; - self.effective_root_span_id = attached - .1 - .or_else(|| self.external_parent_span_id.clone()) - .unwrap_or_else(|| self.root_span_id.clone()); - self.project = config.project_name().map(ToOwned::to_owned); - self.additional_metadata = config - .additional_metadata - .as_ref() - .and_then(Value::as_object) - .cloned() - .unwrap_or_default(); - self.tags = config.tags.clone(); - } - // --- hook-specific side effects (before catch-up) --- match event.event.as_str() { "SessionStart" => { @@ -753,7 +730,7 @@ impl CodexTranslator { }; let cwd = str_field(payload, "cwd"); self.root_cwd = cwd.clone(); - let mut md = self.additional_metadata.clone(); + let mut md = Map::new(); for k in ["id", "cwd", "cli_version"] { if let Some(v) = str_field(payload, k) { md.insert( @@ -782,9 +759,7 @@ impl CodexTranslator { if let Some(m) = &scope.model { md.insert("model".into(), json!(m)); } - if let Some(project) = &self.project { - md.insert("project".into(), json!(project)); - } + md.insert("username".into(), json!(local_username())); scope.root_created = true; ops.push(SpanOp::Insert(SpanRow { @@ -800,7 +775,7 @@ impl CodexTranslator { "source": self.session_source, })), metadata: Some(Value::Object(md)), - tags: (!self.tags.is_empty()).then(|| self.tags.clone()), + tags: None, ..Default::default() })); } @@ -827,7 +802,7 @@ impl CodexTranslator { "agent_type": scope.agent_type, "transcript_path": scope.path, })), - tags: (!self.tags.is_empty()).then(|| self.tags.clone()), + tags: None, ..Default::default() })); } diff --git a/bt-daemon/src/translate/cursor/mod.rs b/bt-daemon/src/translate/cursor/mod.rs index d6f7e861..dbdc53bf 100644 --- a/bt-daemon/src/translate/cursor/mod.rs +++ b/bt-daemon/src/translate/cursor/mod.rs @@ -8,8 +8,7 @@ mod hooks; use super::git::GitMetadataCache; use super::recent::{RecentMap, RecentSet}; use super::{ - local_username, root_tags, AgentTranslator, SessionCtx, SpanOp, SpanRow, SpanType, - TranslatorFactory, + local_username, AgentTranslator, SessionCtx, SpanOp, SpanRow, SpanType, TranslatorFactory, }; use crate::{ids, wire::Envelope}; use hooks::{ @@ -315,13 +314,8 @@ impl CursorTranslator { return; } self.started = true; - let (parent, root) = ctx - .config - .as_ref() - .map(|c| c.attached_span_ids()) - .unwrap_or_default(); - self.trace_root = root.unwrap_or_else(|| self.root.clone()); - self.root_parents = parent.into_iter().collect(); + self.trace_root = self.root.clone(); + self.root_parents.clear(); let mut metadata = json!({"source":"cursor", "session_id":ctx.session_id, "username":local_username(), "os":std::env::consts::OS, "workspace":self.cwd, "trace_cursor_version":e.source_version, "trace_plugin_version":e.plugin_version}); @@ -336,7 +330,7 @@ impl CursorTranslator { span_type: SpanType::Task, start_ms: Some(e.ts_ms), metadata: Some(metadata), - tags: root_tags(ctx), + tags: None, ..Default::default() })); } @@ -1522,10 +1516,9 @@ impl CursorTranslator { ..Default::default() }; } - fn enrich(&self, ops: &mut [SpanOp], ctx: &SessionCtx) { + fn enrich(&self, ops: &mut [SpanOp], _ctx: &SessionCtx) { self.git.enrich_rows(self.cwd.as_deref(), ops); for op in ops { - let root_insert = matches!(op, SpanOp::Insert(row) if row.span_id == self.root); let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; if let Some(metadata) = row.metadata.as_mut().and_then(Value::as_object_mut) { // Only the reviewed Cursor fields reach span metadata. Raw @@ -1541,22 +1534,6 @@ impl CursorTranslator { | "is_interrupt" ) && value == &json!(false)) }); - if root_insert { - // Destination metadata is an explicit user customization, - // rather than additional automatically captured hook data. - if let Some(additional) = ctx - .config - .as_ref() - .and_then(|c| c.additional_metadata.as_ref()) - .and_then(Value::as_object) - { - for (key, value) in additional { - if !key.starts_with("_bt_") { - metadata.entry(key.clone()).or_insert_with(|| value.clone()); - } - } - } - } } } } diff --git a/bt-daemon/src/translate/grok.rs b/bt-daemon/src/translate/grok.rs index de9e2edb..d474844c 100644 --- a/bt-daemon/src/translate/grok.rs +++ b/bt-daemon/src/translate/grok.rs @@ -276,7 +276,7 @@ impl GrokTranslator { &mut self, ts_ms: i64, event: &Envelope, - ctx: &SessionCtx, + _ctx: &SessionCtx, ops: &mut Vec, ) { if self.root_open { @@ -300,22 +300,11 @@ impl GrokTranslator { })); return; } - let attached = ctx - .config - .as_ref() - .map(|config| config.attached_span_ids()) - .unwrap_or_default(); - self.root_span_id = attached.1.unwrap_or_else(|| self.session_span_id.clone()); - self.session_parent_span_ids = attached.0.clone().into_iter().collect(); + self.root_span_id = self.session_span_id.clone(); + self.session_parent_span_ids.clear(); self.root_open = true; - let mut metadata = ctx - .config - .as_ref() - .and_then(|config| config.additional_metadata.as_ref()) - .and_then(Value::as_object) - .cloned() - .unwrap_or_default(); + let mut metadata = serde_json::Map::new(); metadata.retain(|key, _| !key.starts_with("_bt_")); metadata.insert("source".into(), json!("grok")); metadata.insert("session_id".into(), json!(self.session_id)); diff --git a/bt-daemon/src/translate/mod.rs b/bt-daemon/src/translate/mod.rs index 0af0af23..7f5bfd3e 100644 --- a/bt-daemon/src/translate/mod.rs +++ b/bt-daemon/src/translate/mod.rs @@ -124,14 +124,6 @@ pub struct SessionCtx { pub config: Option, } -/// Tags configured for every root span in the current agent session. -pub(crate) fn root_tags(ctx: &SessionCtx) -> Option> { - ctx.config - .as_ref() - .map(|config| config.tags.clone()) - .filter(|tags| !tags.is_empty()) -} - /// A per-session state machine. One instance per session; `&mut self` so it /// can hold open-span maps, transcript offsets, etc. pub trait AgentTranslator: Send { @@ -236,7 +228,10 @@ impl Registry { .canonical_source(source) .ok_or_else(|| anyhow::anyhow!("unsupported coding-agent source {source:?}"))?; let namespace = crate::ids::session_namespace(canonical, session_id); - self.create_checked_with_session_namespace(canonical, &namespace) + let translator = self.create_checked_with_session_namespace(canonical, &namespace)?; + Ok(Box::new( + crate::pipeline::prepare::ConfiguredTranslator::new(translator, canonical), + )) } pub(crate) fn create_checked_with_session_namespace( diff --git a/bt-daemon/src/translate/opencode.rs b/bt-daemon/src/translate/opencode.rs index a7577a31..87159df0 100644 --- a/bt-daemon/src/translate/opencode.rs +++ b/bt-daemon/src/translate/opencode.rs @@ -5,8 +5,7 @@ use super::git::GitMetadataCache; use super::recent::RecentSet; use super::tool::{error_text, with_tool_approval, ToolApproval}; use super::{ - local_username, root_tags, AgentTranslator, SessionCtx, SpanOp, SpanRow, SpanType, - TranslatorFactory, + local_username, AgentTranslator, SessionCtx, SpanOp, SpanRow, SpanType, TranslatorFactory, }; use crate::ids; use crate::wire::Envelope; @@ -585,7 +584,7 @@ impl OpenCodeTranslator { parent_id: Option<&str>, info: Option<&Value>, ts: i64, - ctx: &SessionCtx, + _ctx: &SessionCtx, ) -> Vec { if self.sessions.contains_key(native_id) { return Vec::new(); @@ -606,24 +605,9 @@ impl OpenCodeTranslator { (root_span_id.clone(), Vec::new(), "Subagent".to_string()) } } else { - let external = ctx - .config - .as_ref() - .map(|c| c.attached_span_ids()) - .unwrap_or_default(); - ( - external.1.unwrap_or_else(|| root_span_id.clone()), - external.0.into_iter().collect(), - "OpenCode".to_string(), - ) + (root_span_id.clone(), Vec::new(), "OpenCode".to_string()) }; - let mut metadata = ctx - .config - .as_ref() - .and_then(|c| c.additional_metadata.as_ref()) - .and_then(Value::as_object) - .cloned() - .unwrap_or_default(); + let mut metadata = serde_json::Map::new(); metadata.insert("session_id".into(), Value::String(native_id.to_string())); metadata.insert("source".into(), Value::String("opencode".into())); metadata.insert("_bt_defer_root".into(), Value::Bool(true)); @@ -655,7 +639,7 @@ impl OpenCodeTranslator { span_type: SpanType::Task, start_ms: Some(ts), metadata: Some(Value::Object(metadata)), - tags: root_tags(ctx), + tags: None, ..Default::default() })] } diff --git a/bt-daemon/src/translate/pi.rs b/bt-daemon/src/translate/pi.rs index 52cc89ee..aea46afa 100644 --- a/bt-daemon/src/translate/pi.rs +++ b/bt-daemon/src/translate/pi.rs @@ -6,8 +6,7 @@ use super::tool::{ error_text, explicit_skill_metadata, with_tool_approval, without_nulls, ToolApproval, }; use super::{ - local_username, root_tags, AgentTranslator, SessionCtx, SpanOp, SpanRow, SpanType, - TranslatorFactory, + local_username, AgentTranslator, SessionCtx, SpanOp, SpanRow, SpanType, TranslatorFactory, }; use crate::ids; use crate::wire::Envelope; @@ -537,23 +536,9 @@ impl PiTranslator { // descendants and terminal aggregate merges from this point on. return Vec::new(); } - let attached = ctx - .config - .as_ref() - .map(|c| c.attached_span_ids()) - .unwrap_or_default(); - self.external_parent = attached.0; - self.effective_root_span_id = attached - .1 - .or_else(|| self.external_parent.clone()) - .unwrap_or_else(|| self.root_span_id.clone()); - let mut metadata = ctx - .config - .as_ref() - .and_then(|c| c.additional_metadata.as_ref()) - .and_then(Value::as_object) - .cloned() - .unwrap_or_default(); + self.external_parent = None; + self.effective_root_span_id = self.root_span_id.clone(); + let mut metadata = serde_json::Map::new(); metadata.insert("session_id".into(), json!(ctx.session_id)); metadata.insert("source".into(), json!("pi")); metadata.insert("_bt_defer_root".into(), json!(true)); @@ -591,7 +576,7 @@ impl PiTranslator { span_type: SpanType::Task, start_ms: Some(envelope.ts_ms), metadata: Some(Value::Object(metadata)), - tags: root_tags(ctx), + tags: None, ..Default::default() })] } diff --git a/bt-daemon/tests/codex_translator.rs b/bt-daemon/tests/codex_translator.rs index 5a7974d1..18f0cf72 100644 --- a/bt-daemon/tests/codex_translator.rs +++ b/bt-daemon/tests/codex_translator.rs @@ -1701,10 +1701,7 @@ fn codex_subagent_with_malformed_optional_type_nests_under_spawning_turn() { let reg = Registry::default_agents(); let mut tr = reg.create("codex", "s"); - let ctx = SessionCtx { - session_id: "s".into(), - config: None, - }; + let ctx = configured_ctx("s", json!({"team":"pipeline"})); let mut ops = Vec::new(); // The spawn_agent transcript record is first observed on this same @@ -1791,6 +1788,9 @@ fn codex_subagent_with_malformed_optional_type_nests_under_spawning_turn() { let main_turn = find(&rows, SpanType::Task, "turn: t1"); let subagent = find(&rows, SpanType::Task, "subagent: a1"); let sub_turn = find(&rows, SpanType::Task, "turn: st1"); + assert_eq!(root.tags.as_ref().unwrap(), &["ci", "docs"]); + assert_eq!(subagent.tags.as_ref().unwrap(), &["ci", "docs"]); + assert!(sub_turn.tags.is_none(), "agent tags must not tag its turns"); // Whole thing is one trace under the main root. for r in [main_turn, subagent, sub_turn] { diff --git a/bt-daemon/tests/opencode_translator.rs b/bt-daemon/tests/opencode_translator.rs index e0aba73c..ed11dead 100644 --- a/bt-daemon/tests/opencode_translator.rs +++ b/bt-daemon/tests/opencode_translator.rs @@ -340,7 +340,22 @@ fn opencode_child_sessions_share_the_parent_trace_root() { let mut translator = registry.create("opencode", "root-session"); let ctx = SessionCtx { session_id: "root-session".into(), - config: None, + config: Some( + SessionRoute { + tags: vec!["nested".into()], + additional_metadata: Some( + json!({"team":"pipeline", "source":"override", "_bt_private":true}), + ), + ..SessionRoute::default() + } + .with_auth(BackendAuth { + token: "test".into(), + api_url: None, + app_url: None, + org_name: None, + org_id: None, + }), + ), }; let mut ops = translator .handle( @@ -367,6 +382,20 @@ fn opencode_child_sessions_share_the_parent_trace_root() { .unwrap(), ); assert!(assert_turn_lineage(&ops, |row| row.name.starts_with("Turn ")) >= 2); + let roots: Vec<_> = ops + .iter() + .filter_map(|op| match op { + SpanOp::Insert(row) if row.span_type == SpanType::Task && !row.is_turn => Some(row), + _ => None, + }) + .collect(); + assert_eq!(roots.len(), 2); + for root in roots { + let metadata = root.metadata.as_ref().unwrap(); + assert_eq!(metadata["team"], "pipeline"); + assert_eq!(metadata["source"], "opencode"); + assert!(metadata.get("_bt_private").is_none()); + } let rows = reduce(ops); let parent = rows.values().find(|r| r.name == "OpenCode").unwrap(); let child = rows @@ -375,6 +404,9 @@ fn opencode_child_sessions_share_the_parent_trace_root() { .unwrap(); assert_eq!(child.root_span_id, parent.root_span_id); assert_eq!(child.parent_span_ids.len(), 1); + for root in [parent, child] { + assert_eq!(root.tags.as_ref().unwrap(), &["nested"]); + } } #[test]