diff --git a/bt-daemon/README.md b/bt-daemon/README.md index d633658..f4b613d 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,16 @@ 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. +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 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..752af32 100644 --- a/bt-daemon/docs/protocol.md +++ b/bt-daemon/docs/protocol.md @@ -245,6 +245,17 @@ 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. +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`. + - **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..bb8830c 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 with invocation-local live hooks. +#[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,165 @@ 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 + import_transcript(&file, args.source, opts, config, args.attach).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")]; + 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( @@ -386,66 +588,104 @@ pub async fn import_transcript( source: ImportSource, opts: ServeOptions, config: Option, + attach: bool, ) -> anyhow::Result<()> { - use std::collections::HashMap; - let entries = transcript_import::transcript_envelopes(file, source)?; - - struct Live { - translator: Box, - sink: Box, - ctx: SessionCtx, - pending_ops: usize, + 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(finalizing)?).await?; + if finalizing { + break; + } + tokio::select! { + result = &mut shutdown => { + result?; + finalizing = true; + } + _ = tokio::time::sleep(std::time::Duration::from_millis(500)) => {} + } + } + 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 @@ -544,4 +784,66 @@ 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!(!args + .iter() + .any(|arg| arg == "--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"] + ); + } } 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");