From 037994af540c4d9e74ecb690af1afe8643250ad8 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 4 Aug 2026 22:16:49 +0800 Subject: [PATCH 1/4] Add live transcript attach and managed agent runs --- bt-daemon/README.md | 11 +- bt-daemon/docs/protocol.md | 9 + bt-daemon/src/lib.rs | 418 +++++++++++++++++++++++++---- bt-daemon/src/main.rs | 25 +- bt-daemon/src/transcript_import.rs | 219 ++++++++++++++- 5 files changed, 623 insertions(+), 59 deletions(-) diff --git a/bt-daemon/README.md b/bt-daemon/README.md index d633658..a0e07dc 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 @@ -81,6 +81,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 5247251..14ab544 100644 --- a/bt-daemon/docs/protocol.md +++ b/bt-daemon/docs/protocol.md @@ -245,6 +245,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 5dd2ec8..732783c 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}; @@ -97,6 +98,10 @@ pub struct HookArgs { /// logs. The Claude shim supplies this from CC_EXPERIMENT_ID. #[arg(long)] pub experiment_id: 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`. @@ -124,6 +129,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)] @@ -133,6 +142,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 @@ -149,6 +187,12 @@ pub async fn run_hook( mut config: SessionConfig, 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(()); @@ -359,7 +403,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( @@ -387,65 +597,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, - } - 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, - }, - pending_ops: 0, - }, - ); - sessions.get_mut(&sid).unwrap() +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; } - }; - if let Some(cfg) = &env.config { - live.sink.configure(cfg); - live.ctx.config = Some(cfg.clone()); + _ = 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 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; + } + + 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, + }, + ); + self.sessions.get_mut(&sid).unwrap() + } + }; + 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; + } } } + 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 @@ -544,4 +800,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 f30ff5b..736f155 100644 --- a/bt-daemon/src/main.rs +++ b/bt-daemon/src/main.rs @@ -6,9 +6,9 @@ use bt_daemon::wire::{BackendAuth, FlushMode, SessionConfig}; use bt_daemon::{ - braintrust_serve_options, paths, run_hook, run_import, run_serve, run_status, - BraintrustSinkConfig, DebugSinkFactory, HookArgs, HostInfo, ImportArgs, Registry, ServeArgs, - ServeOptions, StatusArgs, + braintrust_serve_options, paths, run_hook, run_import, run_serve, run_status, run_traced, + BraintrustSinkConfig, DebugSinkFactory, HookArgs, HostInfo, ImportArgs, Registry, RunArgs, + RunHookCommand, ServeArgs, ServeOptions, StatusArgs, }; use clap::{Args, Parser, Subcommand}; use std::ffi::OsString; @@ -69,6 +69,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), } /// Static-token backend auth from env/flags (no profile resolution). @@ -180,5 +182,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 5d94d00..43d18a9 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() @@ -575,4 +700,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 83a79f35bf2c43bd7ecd86466f677eccbc863b1f Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 5 Aug 2026 01:42:32 +0800 Subject: [PATCH 2/4] 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 732783c..9fbe37f 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -142,7 +142,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 3726fe8b2dde23e44e0309270849322d6ba5a699 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 5 Aug 2026 01:59:48 +0800 Subject: [PATCH 3/4] 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 a0e07dc..f4b613d 100644 --- a/bt-daemon/README.md +++ b/bt-daemon/README.md @@ -87,6 +87,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 14ab544..752af32 100644 --- a/bt-daemon/docs/protocol.md +++ b/bt-daemon/docs/protocol.md @@ -251,6 +251,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 9fbe37f..cfa6933 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -536,11 +536,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/4] 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 cfa6933..bb8830c 100644 --- a/bt-daemon/src/lib.rs +++ b/bt-daemon/src/lib.rs @@ -403,11 +403,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 @@ -592,34 +588,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 26e7972..897589d 100644 --- a/bt-daemon/tests/replay.rs +++ b/bt-daemon/tests/replay.rs @@ -54,9 +54,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"); @@ -93,9 +99,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"); @@ -129,9 +141,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");