From 591cdcd748150e986160c34640f02ac361a48917 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 4 Aug 2026 22:16:49 +0800 Subject: [PATCH 1/6] Add live transcript attach and managed agent runs --- bt-daemon/README.md | 11 +- bt-daemon/docs/protocol.md | 9 + bt-daemon/src/lib.rs | 416 +++++++++++++++++++++++++---- bt-daemon/src/main.rs | 21 +- bt-daemon/src/transcript_import.rs | 219 ++++++++++++++- 5 files changed, 620 insertions(+), 56 deletions(-) diff --git a/bt-daemon/README.md b/bt-daemon/README.md index b1b414d..1bf4235 100644 --- a/bt-daemon/README.md +++ b/bt-daemon/README.md @@ -6,7 +6,7 @@ event→trace state machine and sends spans to Braintrust out-of-band. See [`docs/protocol.md`](docs/protocol.md) for the wire contract. > **Placeholder name** — the real name is TBD. The subcommand framing -> (`serve` / `hook` / `status` / `import`) should survive a rename. +> (`serve` / `hook` / `status` / `import` / `run`) should survive a rename. ## Layout @@ -16,7 +16,7 @@ One self-contained Cargo crate, liftable to its own repo by copying - `src/wire` — the wire protocol module: envelope types + JSON-RPC framing. - `src/translate` and `src/sink` — agent state machines and Braintrust output. - `src/lib.rs` — the embeddable library: clap `Args` + async entry points - (`run_serve`, `run_hook`, `run_status`, `run_import`). This is what `bt` + (`run_serve`, `run_hook`, `run_status`, `run_import`, `run_traced`). This is what `bt` depends on. - `src/main.rs` — the standalone **`bt-daemon` binary**, compiled only with the `cli` feature for isolated testing/development. Env/flag static-token @@ -84,6 +84,13 @@ that transcript, and sends them through the normal translator and sink to create a trace for the past session. Hook-only facts absent from a native transcript are not invented. +Add `--attach` to keep following an active Codex or Claude transcript until +Ctrl-C. `run [ARGS...]` launches the selected agent with +inherited stdio and injects live Braintrust hooks for that invocation, so it +does not depend on the tracing plugin being installed or enabled. Managed runs +suppress inherited Braintrust plugin hooks to avoid logging the same session +twice; the injected hooks still use the normal daemon translator and sink. + ## Status Phases 0–5 are implemented: protocol, daemon lifecycle, Braintrust sink, diff --git a/bt-daemon/docs/protocol.md b/bt-daemon/docs/protocol.md index 1ba9392..3125b71 100644 --- a/bt-daemon/docs/protocol.md +++ b/bt-daemon/docs/protocol.md @@ -242,6 +242,15 @@ routing synthetic lifecycle events through the regular translator. `--parent ` attaches it below an exported span and is mutually exclusive with an object destination. +`--attach` keeps a single translator and sink alive, tails new native records, +and finalizes the active turn on Ctrl-C. `run [ARGS...]` +launches the selected agent with inherited stdio and injects live hook +configuration for that invocation, so it works without plugin setup. A private +inherited environment marker makes installed Braintrust plugin hooks no-op for +that managed child, while a private hook flag authorizes the injected hook process. +The resulting native hook events follow the regular journal, translator, and +sink path; transcript tailing remains specific to `import --attach`. + - **Journal (WAL).** Every accepted event is appended (auth-redacted) to `/journal/.ndjson` before/at enqueue. `data_dir` defaults to `$XDG_STATE_HOME/braintrust/bt-daemon` or diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 210fe58..8b1a111 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -33,6 +33,7 @@ pub use translate::{ use braintrust_sdk_rust::SpanComponents; use clap::{Args, ValueEnum}; +use std::ffi::OsString; use std::path::PathBuf; use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; @@ -87,6 +88,10 @@ pub struct HookArgs { /// JSON object merged into root-span metadata. #[arg(long)] pub additional_metadata: Option, + /// Marks the hook definition injected by `run`; inherited plugin hooks do + /// not carry this flag and are suppressed for the managed child. + #[arg(long, hide = true)] + pub managed_run_hook: bool, } /// Arguments for `status`. @@ -114,6 +119,10 @@ pub struct ImportArgs { /// Attach the imported session below an exported Braintrust span. #[arg(long, value_name = "SPAN_COMPONENTS", conflicts_with = "destination")] pub parent: Option, + /// Keep following the transcript until Ctrl-C, importing new turns as the + /// coding-agent session grows. + #[arg(long)] + pub attach: bool, } #[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] @@ -123,6 +132,35 @@ pub enum ImportSource { Claude, } +/// Arguments for launching a coding agent under transcript-based tracing. +#[derive(Debug, Clone, Args)] +#[command(trailing_var_arg = true)] +pub struct RunArgs { + /// Coding agent to launch. + #[arg(value_enum)] + pub source: RunSource, + /// Arguments forwarded verbatim to the coding agent. + #[arg(allow_hyphen_values = true)] + pub agent_args: Vec, +} + +/// Front-end command used by a managed agent run to forward one hook payload. +/// +/// The standalone binary uses `[bt-daemon, hook]`; the embedded `bt` front-end +/// uses its own equivalent prefix. `run_traced` appends `--source `. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RunHookCommand { + pub program: OsString, + pub args: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] +pub enum RunSource { + Codex, + #[value(name = "claude", alias = "claude-code")] + Claude, +} + /// Run the daemon until shutdown. pub async fn run_serve(args: ServeArgs, opts: ServeOptions) -> anyhow::Result<()> { server::run(args, opts).await @@ -139,6 +177,12 @@ pub async fn run_hook( mut route: SessionRoute, host: HostInfo, ) -> anyhow::Result<()> { + // A managed run injects its own hook definitions. Suppress an inherited + // Braintrust plugin hook for the same child, but allow the injected hook + // process, which carries the second marker. + if std::env::var_os("_BT_TRACE_MANAGED_RUN").is_some() && !args.managed_run_hook { + return Ok(()); + } let settings = settings::SharedSettings::load(); if !settings.tracing_enabled() { return Ok(()); @@ -322,7 +366,173 @@ pub async fn run_import( .or(args.destination); apply_import_destination(&mut config, destination)?; let file = transcript_import::resolve_transcript(&args.session_id, args.source)?; - import_transcript(&file, args.source, opts, config).await + if args.attach { + attach_transcript(&file, args.source, opts, config).await + } else { + import_transcript(&file, args.source, opts, config).await + } +} + +/// Launch a coding agent with inherited stdio and inject Braintrust hooks for +/// this invocation, without requiring the tracing plugin to be installed or +/// enabled globally. +pub async fn run_traced( + args: RunArgs, + hook_command: RunHookCommand, +) -> anyhow::Result { + let executable = match args.source { + RunSource::Codex => "codex", + RunSource::Claude => "claude", + }; + let injected_args = managed_run_args(args.source, &hook_command)?; + let mut child = tokio::process::Command::new(executable) + .args(injected_args) + .args(args.agent_args) + .env("_BT_TRACE_MANAGED_RUN", "1") + .spawn() + .map_err(|error| anyhow::anyhow!("failed to launch {executable}: {error}"))?; + let interrupt = tokio::signal::ctrl_c(); + tokio::pin!(interrupt); + + tokio::select! { + status = child.wait() => Ok(status?), + result = &mut interrupt => { + result?; + child.start_kill()?; + Ok(child.wait().await?) + } + } +} + +fn managed_run_args( + source: RunSource, + hook_command: &RunHookCommand, +) -> anyhow::Result> { + let source_name = match source { + RunSource::Codex => "codex", + RunSource::Claude => "claude", + }; + let unix_command = managed_hook_shell_command(hook_command, source_name, false)?; + let windows_command = managed_hook_shell_command(hook_command, source_name, true)?; + match source { + RunSource::Codex => Ok(codex_managed_run_args(&unix_command, &windows_command)), + RunSource::Claude => Ok(claude_managed_run_args(if cfg!(windows) { + &windows_command + } else { + &unix_command + })?), + } +} + +fn managed_hook_shell_command( + hook_command: &RunHookCommand, + source: &str, + windows: bool, +) -> anyhow::Result { + let mut argv = Vec::with_capacity(hook_command.args.len() + 4); + argv.push(hook_command.program.clone()); + argv.extend(hook_command.args.iter().cloned()); + argv.push(OsString::from("--source")); + argv.push(OsString::from(source)); + argv.push(OsString::from("--managed-run-hook")); + let mut rendered = Vec::with_capacity(argv.len()); + for arg in argv { + let arg = arg + .into_string() + .map_err(|_| anyhow::anyhow!("managed hook command contains non-Unicode argv"))?; + rendered.push(if windows { + quote_windows_command_arg(&arg) + } else { + quote_unix_shell_arg(&arg) + }); + } + Ok(rendered.join(" ")) +} + +fn quote_unix_shell_arg(arg: &str) -> String { + format!("'{}'", arg.replace('\'', "'\"'\"'")) +} + +fn quote_windows_command_arg(arg: &str) -> String { + format!("\"{}\"", arg.replace('\\', "/").replace('"', "\"\"")) +} + +const CODEX_RUN_HOOK_EVENTS: &[&str] = &[ + "SessionStart", + "UserPromptSubmit", + "PreToolUse", + "PermissionRequest", + "PostToolUse", + "PreCompact", + "PostCompact", + "SubagentStart", + "SubagentStop", + "Stop", + "SessionEnd", +]; + +const CLAUDE_RUN_HOOK_EVENTS: &[&str] = &[ + "SessionStart", + "Setup", + "UserPromptSubmit", + "UserPromptExpansion", + "PreToolUse", + "PermissionRequest", + "PermissionDenied", + "PostToolUse", + "PostToolUseFailure", + "PostToolBatch", + "PreCompact", + "PostCompact", + "Notification", + "MessageDisplay", + "SubagentStart", + "SubagentStop", + "TaskCreated", + "TaskCompleted", + "Stop", + "StopFailure", + "SessionEnd", +]; + +fn codex_managed_run_args(unix_command: &str, windows_command: &str) -> Vec { + let unix_command = serde_json::to_string(unix_command).expect("serialize hook command"); + let windows_command = + serde_json::to_string(windows_command).expect("serialize Windows hook command"); + let mut args = vec![ + OsString::from("--enable"), + OsString::from("hooks"), + OsString::from("--dangerously-bypass-hook-trust"), + ]; + for event in CODEX_RUN_HOOK_EVENTS { + args.push(OsString::from("-c")); + args.push(OsString::from(format!( + "hooks.{event}=[{{hooks=[{{type=\"command\",command={unix_command},commandWindows={windows_command}}}]}}]" + ))); + } + args +} + +fn claude_managed_run_args(command: &str) -> anyhow::Result> { + let hook = serde_json::json!({ + "hooks": [{ + "hooks": [{ + "type": "command", + "command": command, + "async": false + }] + }] + }); + let hooks = CLAUDE_RUN_HOOK_EVENTS + .iter() + .map(|event| ((*event).to_string(), hook.clone())) + .collect::>(); + Ok(vec![ + OsString::from("--settings"), + OsString::from(serde_json::to_string( + &serde_json::json!({ "hooks": hooks }), + )?), + ]) } fn apply_import_destination( @@ -350,65 +560,111 @@ pub async fn import_transcript( opts: ServeOptions, config: Option, ) -> anyhow::Result<()> { - use std::collections::HashMap; let entries = transcript_import::transcript_envelopes(file, source)?; + let mut processor = ImportProcessor::new(opts, config); + processor.process(entries).await?; + processor.finish().await +} - struct Live { - translator: Box, - sink: Box, - ctx: SessionCtx, - pending_ops: usize, +async fn attach_transcript( + file: &std::path::Path, + source: ImportSource, + opts: ServeOptions, + config: Option, +) -> anyhow::Result<()> { + let mut tail = transcript_import::TranscriptTail::new(file.to_path_buf(), source); + let mut processor = ImportProcessor::new(opts, config); + let shutdown = tokio::signal::ctrl_c(); + tokio::pin!(shutdown); + loop { + processor.process(tail.poll(false)?).await?; + tokio::select! { + result = &mut shutdown => { + result?; + break; + } + _ = tokio::time::sleep(std::time::Duration::from_millis(500)) => {} + } + } + processor.process(tail.poll(true)?).await?; + processor.finish().await +} + +struct ImportLive { + translator: Box, + sink: Box, + ctx: SessionCtx, + pending_ops: usize, +} + +struct ImportProcessor { + sessions: std::collections::HashMap, + opts: ServeOptions, + config: Option, +} + +impl ImportProcessor { + fn new(opts: ServeOptions, config: Option) -> Self { + Self { + sessions: std::collections::HashMap::new(), + opts, + config, + } } - let mut sessions: HashMap = HashMap::new(); - - for mut env in entries { - env.config = config.clone(); - let sid = env.session_id.clone(); - let live = match sessions.get_mut(&sid) { - Some(l) => l, - None => { - let translator = opts.translators.create(&env.source, &sid); - let sink = opts.sink_factory.create(&sid, &env.source)?; - sessions.insert( - sid.clone(), - Live { - translator, - sink, - ctx: SessionCtx { - session_id: sid.clone(), - config: None, + + async fn process(&mut self, entries: Vec) -> anyhow::Result<()> { + for mut env in entries { + env.config = self.config.clone(); + let sid = env.session_id.clone(); + let live = match self.sessions.get_mut(&sid) { + Some(live) => live, + None => { + let translator = self.opts.translators.create(&env.source, &sid); + let sink = self.opts.sink_factory.create(&sid, &env.source)?; + self.sessions.insert( + sid.clone(), + ImportLive { + translator, + sink, + ctx: SessionCtx { + session_id: sid.clone(), + config: None, + }, + pending_ops: 0, }, - pending_ops: 0, - }, - ); - sessions.get_mut(&sid).unwrap() + ); + self.sessions.get_mut(&sid).unwrap() + } + }; + if let Some(cfg) = &env.config { + live.sink.configure(cfg); + live.ctx.config = Some(cfg.clone()); } - }; - if let Some(cfg) = &env.config { - live.sink.configure(cfg); - live.ctx.config = Some(cfg.clone()); - } - let ops = live.translator.handle(&env, &live.ctx)?; - // Imports can contain tens of thousands of SDK log commands. Bound the - // number queued between drains without serializing one network flush - // for every native turn boundary. - const FLUSH_OPS: usize = 500; - for chunk in ops.chunks(FLUSH_OPS) { - live.sink.emit(chunk).await?; - live.pending_ops += chunk.len(); - if live.pending_ops >= FLUSH_OPS { - live.sink.flush().await?; - live.pending_ops = 0; + let ops = live.translator.handle(&env, &live.ctx)?; + // Imports can contain tens of thousands of SDK log commands. Bound the + // number queued between drains without serializing one network flush + // for every native turn boundary. + const FLUSH_OPS: usize = 500; + for chunk in ops.chunks(FLUSH_OPS) { + live.sink.emit(chunk).await?; + live.pending_ops += chunk.len(); + if live.pending_ops >= FLUSH_OPS { + live.sink.flush().await?; + live.pending_ops = 0; + } } } + Ok(()) } - for (_sid, mut live) in sessions { - let ops = live.translator.flush(&live.ctx)?; - live.sink.emit(&ops).await?; - live.sink.flush().await?; + async fn finish(self) -> anyhow::Result<()> { + for (_sid, mut live) in self.sessions { + let ops = live.translator.flush(&live.ctx)?; + live.sink.emit(&ops).await?; + live.sink.flush().await?; + } + Ok(()) } - Ok(()) } /// Build a Phase-1 debug [`ServeOptions`]: debug translator registry + a debug @@ -509,4 +765,64 @@ mod tests { .to_string() .contains("import destination requires a resolved Braintrust session configuration")); } + + fn test_run_hook_command() -> RunHookCommand { + RunHookCommand { + program: OsString::from("/opt/Braintrust CLI/bt"), + args: vec![OsString::from("agents"), OsString::from("hook")], + } + } + + #[test] + fn codex_managed_run_injects_live_hooks() { + let args = managed_run_args(RunSource::Codex, &test_run_hook_command()).unwrap(); + assert_eq!(args[0], "--enable"); + assert_eq!(args[1], "hooks"); + assert_eq!(args[2], "--dangerously-bypass-hook-trust"); + assert_eq!( + args.iter().filter(|arg| *arg == "-c").count(), + CODEX_RUN_HOOK_EVENTS.len() + ); + let config = args + .iter() + .find_map(|arg| { + let arg = arg.to_str()?; + arg.starts_with("hooks.SessionStart=").then_some(arg) + }) + .unwrap(); + assert!(config.contains("--managed-run-hook")); + assert!(config.contains("agents")); + assert!(config.contains("hook")); + assert!(config.contains("--source")); + assert!(config.contains("codex")); + assert!(!config.contains("transcript")); + } + + #[test] + fn claude_managed_run_injects_live_hooks() { + let args = managed_run_args(RunSource::Claude, &test_run_hook_command()).unwrap(); + assert_eq!(args[0], "--settings"); + let settings: serde_json::Value = serde_json::from_str(args[1].to_str().unwrap()).unwrap(); + let hooks = settings["hooks"].as_object().unwrap(); + assert_eq!(hooks.len(), CLAUDE_RUN_HOOK_EVENTS.len()); + let command = hooks["SessionStart"]["hooks"][0]["hooks"][0]["command"] + .as_str() + .unwrap(); + assert!(command.contains("--managed-run-hook")); + assert!(command.contains("agents")); + assert!(command.contains("hook")); + assert!(command.contains("--source")); + assert!(command.contains("claude")); + assert!(!command.contains("transcript")); + } + + #[test] + fn managed_hook_commands_quote_frontend_paths() { + let hook = test_run_hook_command(); + let unix = managed_hook_shell_command(&hook, "codex", false).unwrap(); + assert!(unix.contains("'/opt/Braintrust CLI/bt' 'agents' 'hook' '--source' 'codex'")); + let windows = managed_hook_shell_command(&hook, "claude", true).unwrap(); + assert!(windows + .contains("\"/opt/Braintrust CLI/bt\" \"agents\" \"hook\" \"--source\" \"claude\"")); + } } diff --git a/bt-daemon/src/main.rs b/bt-daemon/src/main.rs index 65ab227..c442dad 100644 --- a/bt-daemon/src/main.rs +++ b/bt-daemon/src/main.rs @@ -9,7 +9,7 @@ use bt_daemon::wire::{AuthSelection, BackendAuth, SessionRoute, TraceDestination use bt_daemon::{ braintrust_serve_options, paths, run_hook, run_import, run_serve, run_status, AuthLease, AuthProvider, AuthResolveReason, BraintrustSinkConfig, DebugSinkFactory, HookArgs, HostInfo, - ImportArgs, Registry, ServeArgs, ServeOptions, StatusArgs, + ImportArgs, Registry, RunArgs, RunHookCommand, ServeArgs, ServeOptions, StatusArgs, run_traced, }; use clap::{Args, Parser, Subcommand}; use std::ffi::OsString; @@ -106,6 +106,8 @@ enum Command { Status(StatusArgs), /// Import a past Codex or Claude Code session by its resume id. Import(ImportArgs), + /// Launch a coding agent with live tracing hooks for this invocation. + Run(RunArgs), } /// Non-secret session selection. Credentials are resolved by the daemon. @@ -214,5 +216,22 @@ async fn main() { std::process::exit(1); } } + Command::Run(args) => { + let exe = std::env::current_exe() + .map(OsString::from) + .unwrap_or_else(|_| OsString::from("bt-daemon")); + let hook_command = RunHookCommand { + program: exe, + args: vec![OsString::from("hook")], + }; + match run_traced(args, hook_command).await { + Ok(status) if status.success() => {} + Ok(status) => std::process::exit(status.code().unwrap_or(1)), + Err(error) => { + eprintln!("bt-daemon run: {error}"); + std::process::exit(1); + } + } + } } } diff --git a/bt-daemon/src/transcript_import.rs b/bt-daemon/src/transcript_import.rs index 0a18150..a8e2a24 100644 --- a/bt-daemon/src/transcript_import.rs +++ b/bt-daemon/src/transcript_import.rs @@ -9,11 +9,15 @@ pub(crate) fn resolve_transcript( source: ImportSource, ) -> anyhow::Result { validate_session_id(session_id)?; + resolve_transcript_in(session_id, source, &transcript_roots(source)) +} + +fn transcript_roots(source: ImportSource) -> Vec { let home = std::env::var_os("HOME") .or_else(|| std::env::var_os("USERPROFILE")) .map(PathBuf::from) .unwrap_or_else(|| PathBuf::from(".")); - let roots = match source { + match source { ImportSource::Codex => { let codex_home = std::env::var_os("CODEX_HOME") .map(PathBuf::from) @@ -29,8 +33,7 @@ pub(crate) fn resolve_transcript( .unwrap_or_else(|| home.join(".claude")); vec![claude_home.join("projects")] } - }; - resolve_transcript_in(session_id, source, &roots) + } } fn resolve_transcript_in( @@ -162,6 +165,128 @@ pub(crate) fn transcript_envelopes( } } +/// Incrementally converts a growing native transcript into synthetic hook +/// events for one persistent translator. The final poll closes the active +/// turn/session; ordinary polls keep the newest turn open. +pub(crate) struct TranscriptTail { + path: PathBuf, + source: ImportSource, + started: bool, + completed_turns: usize, + active_turn: Option, + codex_checkpoints: usize, + last_len: u64, +} + +impl TranscriptTail { + pub(crate) fn new(path: PathBuf, source: ImportSource) -> Self { + Self { + path, + source, + started: false, + completed_turns: 0, + active_turn: None, + codex_checkpoints: 0, + last_len: 0, + } + } + + pub(crate) fn poll(&mut self, finalize: bool) -> anyhow::Result> { + let events = match transcript_envelopes(&self.path, self.source) { + Ok(events) => events, + Err(_) if !finalize => return Ok(Vec::new()), + Err(error) => return Err(error), + }; + let len = std::fs::metadata(&self.path)?.len(); + match self.source { + ImportSource::Codex => self.poll_codex(events, len, finalize), + ImportSource::Claude => self.poll_claude(events, len, finalize), + } + } + + fn poll_codex( + &mut self, + events: Vec, + len: u64, + finalize: bool, + ) -> anyhow::Result> { + if events.len() < 2 { + bail!("Codex import did not produce session boundary events"); + } + let mut out = Vec::new(); + if !self.started { + out.push(events[0].clone()); + self.started = true; + } + let checkpoints = &events[1..events.len() - 1]; + out.extend(checkpoints.iter().skip(self.codex_checkpoints).cloned()); + self.codex_checkpoints = checkpoints.len(); + let mut tail = events.last().cloned().unwrap(); + if finalize { + out.push(tail); + } else if len != self.last_len { + tail.event = "ImportCheckpoint".into(); + if let Some(payload) = tail.payload.as_object_mut() { + payload.insert("hook_event_name".into(), json!("ImportCheckpoint")); + } + out.push(tail); + } + self.last_len = len; + Ok(out) + } + + fn poll_claude( + &mut self, + events: Vec, + len: u64, + finalize: bool, + ) -> anyhow::Result> { + if events.len() < 2 || !(events.len() - 2).is_multiple_of(2) { + bail!("Claude import did not produce turn boundary pairs"); + } + let mut out = Vec::new(); + if !self.started { + out.push(events[0].clone()); + self.started = true; + } + let turn_count = (events.len() - 2) / 2; + let completed_target = if finalize { + turn_count + } else { + turn_count.saturating_sub(1) + }; + while self.completed_turns < completed_target { + let turn = self.completed_turns; + if self.active_turn != Some(turn) { + out.push(events[1 + turn * 2].clone()); + } + out.push(events[2 + turn * 2].clone()); + self.completed_turns += 1; + self.active_turn = None; + } + if !finalize && turn_count > 0 { + let active = turn_count - 1; + if self.active_turn != Some(active) { + out.push(events[1 + active * 2].clone()); + self.active_turn = Some(active); + } + if len != self.last_len { + let mut checkpoint = events.last().cloned().unwrap(); + checkpoint.event = "ImportCheckpoint".into(); + if let Some(payload) = checkpoint.payload.as_object_mut() { + payload.insert("hook_event_name".into(), json!("ImportCheckpoint")); + } + out.push(checkpoint); + } + } + if finalize { + out.push(events.last().cloned().unwrap()); + } + self.last_len = len; + Ok(out) + } +} + fn codex_envelopes(path: &Path, records: &[Value]) -> anyhow::Result> { let meta = records .iter() @@ -576,4 +701,92 @@ mod tests { 3 ); } + + #[test] + fn codex_tail_keeps_session_open_until_final_poll() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("session-123.jsonl"); + let mut records = vec![ + json!({"timestamp":"2026-01-01T00:00:01Z","type":"session_meta","payload":{"id":"session-123"}}), + json!({"timestamp":"2026-01-01T00:00:02Z","type":"event_msg","payload":{"type":"task_started","turn_id":"turn-1"}}), + json!({"timestamp":"2026-01-01T00:00:03Z","type":"response_item","payload":{"type":"message","role":"assistant"}}), + ]; + let write = |records: &[Value]| { + std::fs::write( + &path, + records + .iter() + .map(Value::to_string) + .collect::>() + .join("\n"), + ) + .unwrap(); + }; + write(&records); + let mut tail = TranscriptTail::new(path.clone(), ImportSource::Codex); + let first = tail.poll(false).unwrap(); + assert_eq!(first.first().unwrap().event, "SessionStart"); + assert_eq!(first.last().unwrap().event, "ImportCheckpoint"); + assert!(first.iter().all(|event| event.event != "Stop")); + + records.push(json!({"timestamp":"2026-01-01T00:00:04Z","type":"event_msg","payload":{"type":"task_complete","turn_id":"turn-1"}})); + write(&records); + let second = tail.poll(false).unwrap(); + assert!(second.iter().all(|event| event.event != "SessionStart")); + assert_eq!(second.last().unwrap().event, "ImportCheckpoint"); + assert_eq!(tail.poll(true).unwrap().last().unwrap().event, "Stop"); + } + + #[test] + fn claude_tail_closes_only_completed_turns() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("session-123.jsonl"); + let mut records = vec![ + json!({"type":"user","sessionId":"session-123","timestamp":"2026-01-01T00:00:01Z","message":{"content":"one"}}), + json!({"type":"assistant","sessionId":"session-123","timestamp":"2026-01-01T00:00:02Z","message":{"content":"answer one"}}), + ]; + let write = |records: &[Value]| { + std::fs::write( + &path, + records + .iter() + .map(Value::to_string) + .collect::>() + .join("\n"), + ) + .unwrap(); + }; + write(&records); + let mut tail = TranscriptTail::new(path.clone(), ImportSource::Claude); + let first = tail.poll(false).unwrap(); + assert_eq!( + first + .iter() + .map(|event| event.event.as_str()) + .collect::>(), + vec!["SessionStart", "UserPromptSubmit", "ImportCheckpoint"] + ); + + records.extend([ + json!({"type":"user","sessionId":"session-123","timestamp":"2026-01-01T00:00:03Z","message":{"content":"two"}}), + json!({"type":"assistant","sessionId":"session-123","timestamp":"2026-01-01T00:00:04Z","message":{"content":"answer two"}}), + ]); + write(&records); + let second = tail.poll(false).unwrap(); + assert_eq!( + second + .iter() + .map(|event| event.event.as_str()) + .collect::>(), + vec!["Stop", "UserPromptSubmit", "ImportCheckpoint"] + ); + assert_eq!( + tail.poll(true) + .unwrap() + .iter() + .map(|event| event.event.as_str()) + .collect::>(), + vec!["Stop", "SessionEnd"] + ); + } } From c3334b555f43ebc5ed74ad503fc482296529357c Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 5 Aug 2026 01:42:32 +0800 Subject: [PATCH 2/6] Clarify live-hook run tracing --- bt-daemon/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 8b1a111..c45798f 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -132,7 +132,7 @@ pub enum ImportSource { Claude, } -/// Arguments for launching a coding agent under transcript-based tracing. +/// Arguments for launching a coding agent with invocation-local live hooks. #[derive(Debug, Clone, Args)] #[command(trailing_var_arg = true)] pub struct RunArgs { From 99e709125f8915f91daacfe5888bd623ae177bfa Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 5 Aug 2026 01:59:48 +0800 Subject: [PATCH 3/6] Use normal Codex hook trust --- bt-daemon/README.md | 3 +++ bt-daemon/docs/protocol.md | 2 ++ bt-daemon/src/lib.rs | 10 ++++------ 3 files changed, 9 insertions(+), 6 deletions(-) diff --git a/bt-daemon/README.md b/bt-daemon/README.md index 1bf4235..fa44f74 100644 --- a/bt-daemon/README.md +++ b/bt-daemon/README.md @@ -90,6 +90,9 @@ inherited stdio and injects live Braintrust hooks for that invocation, so it does not depend on the tracing plugin being installed or enabled. Managed runs suppress inherited Braintrust plugin hooks to avoid logging the same session twice; the injected hooks still use the normal daemon translator and sink. +Codex applies its normal hook-review flow, so the first run requires trusting +the injected Braintrust hook through `/hooks`; later runs reuse that trust while +the hook definition remains unchanged. ## Status diff --git a/bt-daemon/docs/protocol.md b/bt-daemon/docs/protocol.md index 3125b71..49f315f 100644 --- a/bt-daemon/docs/protocol.md +++ b/bt-daemon/docs/protocol.md @@ -248,6 +248,8 @@ launches the selected agent with inherited stdio and injects live hook configuration for that invocation, so it works without plugin setup. A private inherited environment marker makes installed Braintrust plugin hooks no-op for that managed child, while a private hook flag authorizes the injected hook process. +Codex does not bypass hook trust: the user reviews the injected hook once through +`/hooks`, and Codex reuses its hash-based trust while the definition is unchanged. The resulting native hook events follow the regular journal, translator, and sink path; transcript tailing remains specific to `import --attach`. diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index c45798f..4487d63 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -499,11 +499,7 @@ fn codex_managed_run_args(unix_command: &str, windows_command: &str) -> Vec Date: Wed, 5 Aug 2026 04:08:12 +0800 Subject: [PATCH 4/6] Unify transcript import modes --- bt-daemon/src/lib.rs | 28 ++++++++-------------------- bt-daemon/tests/replay.rs | 36 +++++++++++++++++++++++++++--------- 2 files changed, 35 insertions(+), 29 deletions(-) diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 4487d63..15f191f 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -366,11 +366,7 @@ pub async fn run_import( .or(args.destination); apply_import_destination(&mut config, destination)?; let file = transcript_import::resolve_transcript(&args.session_id, args.source)?; - if args.attach { - attach_transcript(&file, args.source, opts, config).await - } else { - import_transcript(&file, args.source, opts, config).await - } + import_transcript(&file, args.source, opts, config, args.attach).await } /// Launch a coding agent with inherited stdio and inject Braintrust hooks for @@ -555,34 +551,26 @@ pub async fn import_transcript( source: ImportSource, opts: ServeOptions, config: Option, -) -> anyhow::Result<()> { - let entries = transcript_import::transcript_envelopes(file, source)?; - let mut processor = ImportProcessor::new(opts, config); - processor.process(entries).await?; - processor.finish().await -} - -async fn attach_transcript( - file: &std::path::Path, - source: ImportSource, - opts: ServeOptions, - config: Option, + attach: bool, ) -> anyhow::Result<()> { let mut tail = transcript_import::TranscriptTail::new(file.to_path_buf(), source); let mut processor = ImportProcessor::new(opts, config); let shutdown = tokio::signal::ctrl_c(); tokio::pin!(shutdown); + let mut finalizing = !attach; loop { - processor.process(tail.poll(false)?).await?; + processor.process(tail.poll(finalizing)?).await?; + if finalizing { + break; + } tokio::select! { result = &mut shutdown => { result?; - break; + finalizing = true; } _ = tokio::time::sleep(std::time::Duration::from_millis(500)) => {} } } - processor.process(tail.poll(true)?).await?; processor.finish().await } diff --git a/bt-daemon/tests/replay.rs b/bt-daemon/tests/replay.rs index c4d0e16..5ac4c02 100644 --- a/bt-daemon/tests/replay.rs +++ b/bt-daemon/tests/replay.rs @@ -55,9 +55,15 @@ async fn imports_native_codex_rollout_through_codex_translator() { ); let output = tmp.path().join("spans"); - import_transcript(&transcript, ImportSource::Codex, options(&output), None) - .await - .unwrap(); + import_transcript( + &transcript, + ImportSource::Codex, + options(&output), + None, + false, + ) + .await + .unwrap(); let rows = rows(&output.join("codex-past.ndjson")); assert_eq!(inserted(&rows, "task"), 3, "session and two turns"); @@ -94,9 +100,15 @@ async fn imports_native_claude_transcript_with_multiple_turns_and_tools() { ); let output = tmp.path().join("spans"); - import_transcript(&transcript, ImportSource::Claude, options(&output), None) - .await - .unwrap(); + import_transcript( + &transcript, + ImportSource::Claude, + options(&output), + None, + false, + ) + .await + .unwrap(); let rows = rows(&output.join("claude-past.ndjson")); assert_eq!(inserted(&rows, "task"), 3, "session and two turns"); @@ -130,9 +142,15 @@ async fn imports_non_monotonic_claude_records_into_their_native_turns() { ); let output = tmp.path().join("spans"); - import_transcript(&transcript, ImportSource::Claude, options(&output), None) - .await - .unwrap(); + import_transcript( + &transcript, + ImportSource::Claude, + options(&output), + None, + false, + ) + .await + .unwrap(); let rows = rows(&output.join("claude-non-monotonic.ndjson")); assert_eq!(inserted(&rows, "task"), 4, "session and three turns"); From bf55192f8d8438301b020165da80982983928e8e Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Thu, 6 Aug 2026 02:09:02 +0800 Subject: [PATCH 5/6] Propagate selected routes through managed runs --- bt-daemon/src/lib.rs | 3 +++ bt-daemon/src/main.rs | 17 +++++++++++------ 2 files changed, 14 insertions(+), 6 deletions(-) diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 15f191f..dc0196f 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -375,16 +375,19 @@ pub async fn run_import( pub async fn run_traced( args: RunArgs, hook_command: RunHookCommand, + route: SessionRoute, ) -> anyhow::Result { let executable = match args.source { RunSource::Codex => "codex", RunSource::Claude => "claude", }; let injected_args = managed_run_args(args.source, &hook_command)?; + let route = serde_json::to_string(&route)?; let mut child = tokio::process::Command::new(executable) .args(injected_args) .args(args.agent_args) .env("_BT_TRACE_MANAGED_RUN", "1") + .env(settings::SESSION_ROUTE_ENV, route) .spawn() .map_err(|error| anyhow::anyhow!("failed to launch {executable}: {error}"))?; let interrupt = tokio::signal::ctrl_c(); diff --git a/bt-daemon/src/main.rs b/bt-daemon/src/main.rs index c442dad..f2b8032 100644 --- a/bt-daemon/src/main.rs +++ b/bt-daemon/src/main.rs @@ -7,9 +7,9 @@ use async_trait::async_trait; use bt_daemon::wire::{AuthSelection, BackendAuth, SessionRoute, TraceDestination}; use bt_daemon::{ - braintrust_serve_options, paths, run_hook, run_import, run_serve, run_status, AuthLease, - AuthProvider, AuthResolveReason, BraintrustSinkConfig, DebugSinkFactory, HookArgs, HostInfo, - ImportArgs, Registry, RunArgs, RunHookCommand, ServeArgs, ServeOptions, StatusArgs, run_traced, + braintrust_serve_options, paths, run_hook, run_import, run_serve, run_status, run_traced, + AuthLease, AuthProvider, AuthResolveReason, BraintrustSinkConfig, DebugSinkFactory, HookArgs, + HostInfo, ImportArgs, Registry, RunArgs, RunHookCommand, ServeArgs, ServeOptions, StatusArgs, }; use clap::{Args, Parser, Subcommand}; use std::ffi::OsString; @@ -107,7 +107,12 @@ enum Command { /// Import a past Codex or Claude Code session by its resume id. Import(ImportArgs), /// Launch a coding agent with live tracing hooks for this invocation. - Run(RunArgs), + Run { + #[command(flatten)] + route: RouteArgs, + #[command(flatten)] + args: RunArgs, + }, } /// Non-secret session selection. Credentials are resolved by the daemon. @@ -216,7 +221,7 @@ async fn main() { std::process::exit(1); } } - Command::Run(args) => { + Command::Run { route, args } => { let exe = std::env::current_exe() .map(OsString::from) .unwrap_or_else(|_| OsString::from("bt-daemon")); @@ -224,7 +229,7 @@ async fn main() { program: exe, args: vec![OsString::from("hook")], }; - match run_traced(args, hook_command).await { + match run_traced(args, hook_command, route.into_route()).await { Ok(status) if status.success() => {} Ok(status) => std::process::exit(status.code().unwrap_or(1)), Err(error) => { From f13bee16b6835e521c5b903d6802f613b302a0fc Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Thu, 6 Aug 2026 23:11:36 +0800 Subject: [PATCH 6/6] Isolate managed run settings from setup --- bt-daemon/README.md | 5 ++ bt-daemon/docs/protocol.md | 5 ++ bt-daemon/src/lib.rs | 25 +++++++- bt-daemon/src/main.rs | 28 ++++++--- bt-daemon/src/settings.rs | 120 ++++++++++++++++++++++++++++++++++--- 5 files changed, 165 insertions(+), 18 deletions(-) diff --git a/bt-daemon/README.md b/bt-daemon/README.md index fa44f74..cb7e561 100644 --- a/bt-daemon/README.md +++ b/bt-daemon/README.md @@ -94,6 +94,11 @@ Codex applies its normal hook-review flow, so the first run requires trusting the injected Braintrust hook through `/hooks`; later runs reuse that trust while the hook definition remains unchanged. +Managed-run settings are scoped to the launched agent process tree. They enable +tracing for that invocation and override the persistent setup route without +rewriting it, so ordinary agent sessions and concurrent managed runs may use +different profiles, organizations, projects, experiments, or parent spans. + ## Status Phases 0–5 are implemented: protocol, daemon lifecycle, Braintrust sink, diff --git a/bt-daemon/docs/protocol.md b/bt-daemon/docs/protocol.md index 49f315f..4ed3065 100644 --- a/bt-daemon/docs/protocol.md +++ b/bt-daemon/docs/protocol.md @@ -252,6 +252,11 @@ Codex does not bypass hook trust: the user reviews the injected hook once throug `/hooks`, and Codex reuses its hash-based trust while the definition is unchanged. The resulting native hook events follow the regular journal, translator, and sink path; transcript tailing remains specific to `import --attach`. +The managed child inherits a non-secret invocation-settings value containing +`traceToBraintrust: true` and its immutable `SessionRoute`. This overrides the +persistent setup route only within that process tree. Other agent processes +continue using setup settings, and concurrent managed runs can select distinct +profiles, organizations, and destinations while sharing one daemon. - **Journal (WAL).** Every accepted event is appended (auth-redacted) to `/journal/.ndjson` before/at enqueue. `data_dir` diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index dc0196f..75b755a 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -377,17 +377,22 @@ pub async fn run_traced( hook_command: RunHookCommand, route: SessionRoute, ) -> anyhow::Result { + if route.destination.is_none() { + anyhow::bail!( + "managed run requires a trace destination; select a project, object destination, or parent span" + ); + } let executable = match args.source { RunSource::Codex => "codex", RunSource::Claude => "claude", }; let injected_args = managed_run_args(args.source, &hook_command)?; - let route = serde_json::to_string(&route)?; + let invocation_settings = serde_json::to_string(&settings::InvocationSettings::enabled(route))?; let mut child = tokio::process::Command::new(executable) .args(injected_args) .args(args.agent_args) .env("_BT_TRACE_MANAGED_RUN", "1") - .env(settings::SESSION_ROUTE_ENV, route) + .env(settings::INVOCATION_SETTINGS_ENV, invocation_settings) .spawn() .map_err(|error| anyhow::anyhow!("failed to launch {executable}: {error}"))?; let interrupt = tokio::signal::ctrl_c(); @@ -760,6 +765,22 @@ mod tests { } } + #[tokio::test] + async fn managed_run_rejects_a_missing_destination_before_launch() { + let error = run_traced( + RunArgs { + source: RunSource::Codex, + agent_args: Vec::new(), + }, + test_run_hook_command(), + SessionRoute::default(), + ) + .await + .unwrap_err(); + + assert!(error.to_string().contains("requires a trace destination")); + } + #[test] fn codex_managed_run_injects_live_hooks() { let args = managed_run_args(RunSource::Codex, &test_run_hook_command()).unwrap(); diff --git a/bt-daemon/src/main.rs b/bt-daemon/src/main.rs index f2b8032..2fb7f6e 100644 --- a/bt-daemon/src/main.rs +++ b/bt-daemon/src/main.rs @@ -122,10 +122,16 @@ struct RouteArgs { profile: Option, #[arg(long = "org", env = "BRAINTRUST_ORG_NAME")] org_name: Option, - #[arg(long, env = "BRAINTRUST_PROJECT", conflicts_with = "destination")] + #[arg( + long, + env = "BRAINTRUST_PROJECT", + conflicts_with_all = ["destination", "parent"] + )] project: Option, - #[arg(long, env = "BRAINTRUST_DESTINATION")] + #[arg(long, env = "BRAINTRUST_DESTINATION", conflicts_with = "parent")] destination: Option, + #[arg(long, value_name = "SPAN_COMPONENTS")] + parent: Option, } impl RouteArgs { @@ -135,13 +141,17 @@ impl RouteArgs { profile: self.profile, org_name: self.org_name, }, - destination: self.destination.or_else(|| { - self.project - .map(|project_name| TraceDestination::ProjectLogs { - project_id: None, - project_name: Some(project_name), - }) - }), + destination: self + .parent + .map(|components| TraceDestination::ParentSpan { components }) + .or(self.destination) + .or_else(|| { + self.project + .map(|project_name| TraceDestination::ProjectLogs { + project_id: None, + project_name: Some(project_name), + }) + }), ..SessionRoute::default() } } diff --git a/bt-daemon/src/settings.rs b/bt-daemon/src/settings.rs index 807e042..b543d80 100644 --- a/bt-daemon/src/settings.rs +++ b/bt-daemon/src/settings.rs @@ -6,10 +6,27 @@ use crate::paths; use crate::wire::SessionRoute; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use std::path::Path; -pub(crate) const SESSION_ROUTE_ENV: &str = "BT_TRACE_SESSION_ROUTE"; +pub(crate) const INVOCATION_SETTINGS_ENV: &str = "BT_TRACE_INVOCATION_SETTINGS"; + +/// Non-secret settings scoped to one `bt trace run` process tree. +#[derive(Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct InvocationSettings { + pub trace_to_braintrust: bool, + pub route: SessionRoute, +} + +impl InvocationSettings { + pub(crate) fn enabled(route: SessionRoute) -> Self { + Self { + trace_to_braintrust: true, + route, + } + } +} #[derive(Debug, Default, Deserialize)] #[serde(rename_all = "camelCase")] @@ -20,11 +37,25 @@ pub(crate) struct SharedSettings { impl SharedSettings { pub(crate) fn load() -> Self { - let mut settings = Self::load_from(&paths::settings_path(None)); - if let Ok(raw) = std::env::var(SESSION_ROUTE_ENV) { - match serde_json::from_str(&raw) { - Ok(route) => settings.route = Some(route), - Err(error) => tracing::warn!("managed run route ignored: {error}"), + let invocation = std::env::var(INVOCATION_SETTINGS_ENV).ok(); + Self::load_from_sources(&paths::settings_path(None), invocation.as_deref()) + } + + fn load_from_sources(path: &Path, invocation: Option<&str>) -> Self { + let mut settings = Self::load_from(path); + if let Some(raw) = invocation { + match serde_json::from_str::(raw) { + Ok(invocation) => { + settings.trace_to_braintrust = Some(invocation.trace_to_braintrust); + settings.route = Some(invocation.route); + } + Err(error) => { + tracing::warn!("managed run settings ignored: {error}"); + // Never fall back to the persistent route for a managed + // child whose invocation selection cannot be decoded. + settings.trace_to_braintrust = Some(false); + settings.route = None; + } } } settings @@ -129,6 +160,81 @@ mod tests { assert!(!SharedSettings::default().tracing_enabled_with(None)); } + #[test] + fn invocation_settings_override_setup_without_mutating_it() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("config.json"); + std::fs::write( + &path, + r#"{ + "traceToBraintrust": false, + "route": { + "auth": {"profile": "global", "org_name": "global-org"}, + "destination": {"type": "project_logs", "project_name": "global-project"} + } + }"#, + ) + .unwrap(); + + let invocation = |profile: &str, project: &str| { + serde_json::to_string(&InvocationSettings::enabled(SessionRoute { + auth: crate::wire::AuthSelection { + profile: Some(profile.to_string()), + org_name: Some(format!("{profile}-org")), + }, + destination: Some(crate::wire::TraceDestination::ProjectLogs { + project_id: None, + project_name: Some(project.to_string()), + }), + ..SessionRoute::default() + })) + .unwrap() + }; + + let work = + SharedSettings::load_from_sources(&path, Some(&invocation("work", "work-project"))); + let personal = SharedSettings::load_from_sources( + &path, + Some(&invocation("personal", "personal-project")), + ); + let global = SharedSettings::load_from_sources(&path, None); + + assert!(work.tracing_enabled_with(None)); + assert!(personal.tracing_enabled_with(None)); + assert_eq!(work.route.unwrap().auth.profile.as_deref(), Some("work")); + assert_eq!( + personal.route.unwrap().auth.profile.as_deref(), + Some("personal") + ); + assert!(!global.tracing_enabled_with(None)); + let global_route = global.route.unwrap(); + assert_eq!(global_route.auth.profile.as_deref(), Some("global")); + assert_eq!( + global_route.destination.unwrap().project_name(), + Some("global-project") + ); + } + + #[test] + fn malformed_invocation_settings_do_not_fall_back_to_setup_route() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("config.json"); + std::fs::write( + &path, + r#"{ + "traceToBraintrust": true, + "route": { + "destination": {"type": "project_logs", "project_name": "global-project"} + } + }"#, + ) + .unwrap(); + + let settings = SharedSettings::load_from_sources(&path, Some("{")); + assert!(!settings.tracing_enabled_with(None)); + assert!(settings.route.is_none()); + } + #[test] fn boolean_environment_values_match_launcher_contract() { for value in ["1", "true", "TRUE", "yes", "on"] {