Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 13 additions & 9 deletions bt-daemon/src/dispatch.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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;
Expand Down
1 change: 1 addition & 0 deletions bt-daemon/src/pipeline/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down
199 changes: 199 additions & 0 deletions bt-daemon/src/pipeline/prepare.rs
Original file line number Diff line number Diff line change
@@ -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<String>,
}

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);
Comment thread
Qard marked this conversation as resolved.
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<dyn crate::translate::AgentTranslator>,
prepare: Prepare,
}

impl ConfiguredTranslator {
pub fn new(inner: Box<dyn crate::translate::AgentTranslator>, source: &str) -> Self {
Self {
inner,
prepare: Prepare::new(source),
}
}
fn apply(&mut self, mut ops: Vec<SpanOp>, ctx: &crate::translate::SessionCtx) -> Vec<SpanOp> {
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<Vec<SpanOp>> {
let ops = self.inner.handle(env, ctx)?;
Ok(self.apply(ops, ctx))
}
fn drain_pending(
&mut self,
ctx: &crate::translate::SessionCtx,
) -> anyhow::Result<Option<Vec<SpanOp>>> {
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<Vec<SpanOp>> {
let ops = self.inner.checkpoint(ctx)?;
Ok(self.apply(ops, ctx))
}
fn finalize(&mut self, ctx: &crate::translate::SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
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());
}
}
5 changes: 4 additions & 1 deletion bt-daemon/src/pipeline/routing.rs
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ impl RoutingSink {
) -> Self {
let (tx, mut rx) = mpsc::channel::<Delivery<Request>>(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);
Expand All @@ -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(),
Expand All @@ -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 => {}
Expand Down
22 changes: 3 additions & 19 deletions bt-daemon/src/translate/antigravity.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -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()
}));
}
Expand Down
21 changes: 3 additions & 18 deletions bt-daemon/src/translate/claude.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -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()
}));
}
Expand Down
Loading
Loading