From ce296820ac242e4ba91aa5574b1e38b8c3f4d2a4 Mon Sep 17 00:00:00 2001 From: Victor Garcia Date: Sat, 8 Aug 2026 01:10:49 -0700 Subject: [PATCH] `bullpen run --json`: consume a run as a stream instead of scraping a terminal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A program driving `bullpen run` had nothing to read. The final answer arrived as raw text deltas on stdout with no framing, tool activity was human prose gated behind -v on stderr, and the Event channel the loop already publishes had three of its four variants discarded by the CLI printer. Completion could only be inferred from stdout closing, which is indistinguishable from a crash. With --json, stdout carries one JSON object per line, flushed as each event happens so a consumer can act mid-flight, and the last line is always a `result` object carrying the session id, the final text, cumulative usage and an error flag. Without the flag nothing about the output changes. Notes: - The event kind strings live in crates/cli/src/json.rs as `const KIND_*`, hand-written rather than derived from the `Event` variant names. Once anything reads this stream the strings are a compatibility surface, and the crate that owns the wire should be the crate that owns the names — renaming an internal variant must not move a wire value. `event_json` is exhaustive on `Event` so a fifth variant fails to compile rather than silently vanishing from the stream. - The text-delta sink stays attached under --json and its text is dropped. Detaching it would flip the agent from complete_streaming to complete, a behavior change hidden behind an output flag. Deltas are not events; the per-turn AssistantText is the only text event. - Tool payloads are capped against bullpen-agent's MAX_TOOL_RESULT_BYTES, which this makes pub rather than duplicating the literal. The cap is genuinely needed here: run_tool_with_events emits ToolEnd with the *uncapped* output, and cap_result only applies as a result enters the transcript. The `result` text is deliberately not capped — that is the deliverable, not a payload. - `--bg --json` prints one `dispatched` object rather than a terminal object. A detached run has no stream and no completion to report. - --json is not forwarded across the dispatch boundary into the child, for the same reason -v is not: the child's stdout and stderr share one log file, so NDJSON there would interleave with prose. Refs #9 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01PfAfAujueuZ3rDTiL9apx3 --- ARCHITECTURE.md | 9 +- README.md | 1 + crates/agent/src/lib.rs | 5 +- crates/cli/src/json.rs | 288 ++++++++++++++++++++++++++++++++++++++++ crates/cli/src/main.rs | 50 ++++++- 5 files changed, 345 insertions(+), 8 deletions(-) create mode 100644 crates/cli/src/json.rs diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 3303db6..a4ccd84 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -88,13 +88,18 @@ Enforced in `bullpen-agent`, tested in its unit suite: provider call, even after the max-turns fuse trips. - A `max_tokens` stop returns the partial text as a distinct error; it is never silently continued. -- Tool results are capped (256 KiB) before entering the transcript. +- Tool results are capped (256 KiB) before entering the transcript. The same + cap bounds tool payloads on the CLI's JSON event stream. ## Event model The loop emits `Event` values (assistant text, tool start/end, turn done) on an optional channel. Rendering lives entirely outside the loop — the headless -CLI prints them; the future TUI consumes the same stream. A gone subscriber +CLI either renders them for a human or serializes them (`run --json`, one +object per line on stdout, terminated by a `result` object so completion is +never inferred from EOF); the future TUI consumes the same stream. The wire +names for the event kinds are owned by the CLI, not derived from the `Event` +variants, so renaming a variant cannot break a consumer. A gone subscriber never stops the loop. ## Persistence and durable execution diff --git a/README.md b/README.md index f1c9cb8..fba7287 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,7 @@ cd your-project bullpen run "find the failing test and explain why it fails" bullpen run --sandbox "refactor the retry logic" # confine writes bullpen run -v "..." # tool activity on stderr +bullpen run --json "..." # NDJSON event stream on stdout ``` Sessions are resumable by id prefix, with the provider they were created diff --git a/crates/agent/src/lib.rs b/crates/agent/src/lib.rs index baed126..eb71042 100644 --- a/crates/agent/src/lib.rs +++ b/crates/agent/src/lib.rs @@ -25,8 +25,9 @@ use bullpen_llm::{ use bullpen_tools::{Registry, ToolCtx}; use tokio::sync::mpsc::UnboundedSender; -/// Cap on any single tool result entering the transcript. -const MAX_TOOL_RESULT_BYTES: usize = 262_144; // 256 KiB +/// Cap on any single tool result the loop hands on — to the transcript, and +/// to anything downstream that re-emits event payloads. +pub const MAX_TOOL_RESULT_BYTES: usize = 262_144; // 256 KiB #[derive(Debug, Clone)] pub struct AgentConfig { diff --git a/crates/cli/src/json.rs b/crates/cli/src/json.rs new file mode 100644 index 0000000..608d8a2 --- /dev/null +++ b/crates/cli/src/json.rs @@ -0,0 +1,288 @@ +//! `bullpen run --json` — the run as newline-delimited JSON. +//! +//! The event kinds below are hand-written string constants, not names derived +//! from the `Event` variants they came from. `bullpen_agent::Event` is +//! internal state: renaming a variant is a refactor, and it must not silently +//! change the wire. Same argument `sessions_json` makes for field names — the +//! moment anything reads this stream, the strings are a compatibility surface +//! and the CLI is the crate that owns it. +//! +//! Every builder here is pure. The one impure function, [`emit`], is the only +//! place the stream touches stdout, so ordering is whatever order the caller +//! calls it in. + +use std::io::Write; + +use bullpen_agent::{Event, MAX_TOOL_RESULT_BYTES}; +use bullpen_llm::Usage; +use serde_json::{Value, json}; + +pub const KIND_ASSISTANT_TEXT: &str = "assistant_text"; +pub const KIND_TOOL_START: &str = "tool_start"; +pub const KIND_TOOL_END: &str = "tool_end"; +pub const KIND_TURN_DONE: &str = "turn_done"; +pub const KIND_RESULT: &str = "result"; +pub const KIND_DISPATCHED: &str = "dispatched"; + +/// One event as one line of the stream. Exhaustive on purpose: a fifth +/// `Event` variant must fail to compile here rather than vanish from the wire. +/// Pure so it can be tested without a terminal. +pub fn event_json(event: &Event) -> Value { + match event { + Event::AssistantText { text } => json!({ + "kind": KIND_ASSISTANT_TEXT, + "text": text, + }), + Event::ToolStart { id, name, input } => { + let (input, truncated) = cap_input(input); + json!({ + "kind": KIND_TOOL_START, + "id": id, + "name": name, + "input": input, + "input_truncated": truncated, + }) + } + Event::ToolEnd { + id, + name, + output, + is_error, + } => { + let (output, truncated) = cap_text(output); + json!({ + "kind": KIND_TOOL_END, + "id": id, + "name": name, + "output": output, + "output_truncated": truncated, + "is_error": is_error, + }) + } + Event::TurnDone { usage } => json!({ + "kind": KIND_TURN_DONE, + "usage": usage_json(*usage), + }), + } +} + +/// The terminal object. Carries the outcome outright so a consumer never has +/// to infer completion from the stream closing. +/// Pure so it can be tested without a terminal. +pub fn result_json(session_id: &str, text: &str, usage: Usage, error: Option<&str>) -> Value { + json!({ + "kind": KIND_RESULT, + "session_id": session_id, + "text": text, + "usage": usage_json(usage), + "error": error.is_some(), + "message": error, + }) +} + +/// The whole stream for a `--bg` dispatch: the run itself happens in another +/// process, so there is nothing to stream but the handle to it. +/// Pure so it can be tested without a terminal. +pub fn dispatched_json(session_id: &str, pid: u32) -> Value { + json!({ + "kind": KIND_DISPATCHED, + "session_id": session_id, + "pid": pid, + }) +} + +fn usage_json(usage: Usage) -> Value { + json!({ + "input_tokens": usage.input_tokens, + "output_tokens": usage.output_tokens, + }) +} + +/// Truncate to the shared 256 KiB cap on a char boundary, reporting whether it +/// had to. The flag replaces `cap_result`'s inline marker: a consumer parsing +/// JSON should not have to scan the payload for a sentinel. +fn cap_text(s: &str) -> (String, bool) { + if s.len() <= MAX_TOOL_RESULT_BYTES { + return (s.to_string(), false); + } + let mut end = MAX_TOOL_RESULT_BYTES; + while !s.is_char_boundary(end) { + end -= 1; + } + (s[..end].to_string(), true) +} + +/// Tool input stays a JSON value while it fits. Over the cap it becomes the +/// truncated serialization as a string — a clipped object would not parse. +fn cap_input(input: &Value) -> (Value, bool) { + let text = input.to_string(); + if text.len() <= MAX_TOOL_RESULT_BYTES { + return (input.clone(), false); + } + let (capped, _) = cap_text(&text); + (Value::String(capped), true) +} + +/// Write one line and flush it, so a consumer reads the run mid-flight. +/// Failures are dropped for the same reason the delta streamer drops them: a +/// closed stdout must not take the run down with it. +pub fn emit(value: &Value) { + let mut out = std::io::stdout(); + let _ = out.write_all(value.to_string().as_bytes()); + let _ = out.write_all(b"\n"); + let _ = out.flush(); +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn usage(input: u64, output: u64) -> Usage { + Usage { + input_tokens: input, + output_tokens: output, + } + } + + #[test] + fn event_kinds_are_stable_wire_strings() { + assert_eq!(KIND_ASSISTANT_TEXT, "assistant_text"); + assert_eq!(KIND_TOOL_START, "tool_start"); + assert_eq!(KIND_TOOL_END, "tool_end"); + assert_eq!(KIND_TURN_DONE, "turn_done"); + assert_eq!(KIND_RESULT, "result"); + assert_eq!(KIND_DISPATCHED, "dispatched"); + } + + #[test] + fn assistant_text_carries_the_turn_text() { + assert_eq!( + event_json(&Event::AssistantText { + text: "hello".into() + }), + json!({ "kind": "assistant_text", "text": "hello" }) + ); + } + + #[test] + fn tool_start_emits_its_input_as_a_json_value_when_small() { + assert_eq!( + event_json(&Event::ToolStart { + id: "t1".into(), + name: "bash".into(), + input: json!({ "command": "ls" }), + }), + json!({ + "kind": "tool_start", + "id": "t1", + "name": "bash", + "input": { "command": "ls" }, + "input_truncated": false, + }) + ); + } + + #[test] + fn tool_start_input_over_the_cap_becomes_a_truncated_string() { + let event = Event::ToolStart { + id: "t1".into(), + name: "bash".into(), + input: json!({ "command": "x".repeat(MAX_TOOL_RESULT_BYTES) }), + }; + let value = event_json(&event); + assert_eq!(value["input_truncated"], json!(true)); + let input = value["input"].as_str().unwrap(); + assert!(input.len() <= MAX_TOOL_RESULT_BYTES); + assert!(input.starts_with(r#"{"command":"xxx"#)); + } + + #[test] + fn tool_end_output_is_truncated_on_a_char_boundary() { + let event = Event::ToolEnd { + id: "t1".into(), + name: "bash".into(), + // Three-byte chars: the cap lands mid-character. + output: "☃".repeat(MAX_TOOL_RESULT_BYTES), + is_error: false, + }; + let value = event_json(&event); + assert_eq!(value["output_truncated"], json!(true)); + let output = value["output"].as_str().unwrap(); + assert!(output.len() <= MAX_TOOL_RESULT_BYTES); + assert!(output.chars().all(|c| c == '☃')); + } + + #[test] + fn tool_end_flags_provider_errors() { + assert_eq!( + event_json(&Event::ToolEnd { + id: "t1".into(), + name: "bash".into(), + output: "no such file".into(), + is_error: true, + }), + json!({ + "kind": "tool_end", + "id": "t1", + "name": "bash", + "output": "no such file", + "output_truncated": false, + "is_error": true, + }) + ); + } + + #[test] + fn turn_done_carries_cumulative_usage() { + assert_eq!( + event_json(&Event::TurnDone { + usage: usage(120, 34) + }), + json!({ + "kind": "turn_done", + "usage": { "input_tokens": 120, "output_tokens": 34 }, + }) + ); + } + + #[test] + fn result_carries_the_full_session_id_not_the_display_prefix() { + let id = "0123456789abcdef0123456789abcdef"; + assert_eq!( + result_json(id, "the answer", usage(9, 4), None), + json!({ + "kind": "result", + "session_id": id, + "text": "the answer", + "usage": { "input_tokens": 9, "output_tokens": 4 }, + "error": false, + "message": null, + }) + ); + } + + #[test] + fn result_on_error_keeps_the_partial_text_and_sets_the_error_flag() { + assert_eq!( + result_json("abc", "half an ans", usage(9, 4), Some("provider stopped")), + json!({ + "kind": "result", + "session_id": "abc", + "text": "half an ans", + "usage": { "input_tokens": 9, "output_tokens": 4 }, + "error": true, + "message": "provider stopped", + }) + ); + } + + #[test] + fn dispatched_carries_the_full_id_and_pid() { + let id = "0123456789abcdef0123456789abcdef"; + assert_eq!( + dispatched_json(id, 4242), + json!({ "kind": "dispatched", "session_id": id, "pid": 4242 }) + ); + } +} diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index 262a20a..3b1e97f 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -6,11 +6,12 @@ mod agents; mod bg; +mod json; use std::sync::Arc; use anyhow::{Context, bail}; -use bullpen_agent::{Agent, AgentConfig, Event}; +use bullpen_agent::{Agent, AgentConfig, AgentError, Event}; use bullpen_auth::codex::{CodexAuth, CodexCliBorrow, StoredCodex}; use bullpen_auth::{AuthFile, Credential, openrouter, pkce::Pkce}; use bullpen_llm::Provider; @@ -49,6 +50,9 @@ enum Command { /// Show tool activity on stderr while running. #[arg(short, long)] verbose: bool, + /// Stream the run as newline-delimited JSON on stdout. + #[arg(long)] + json: bool, /// Confine writes to the workspace (and system temp). On macOS this /// also runs shell commands under Seatbelt; elsewhere only the file /// tools are confined (see --sandbox notes). @@ -175,6 +179,7 @@ async fn main() -> anyhow::Result<()> { model, resume, verbose, + json, sandbox, sandbox_strict, bg, @@ -185,6 +190,7 @@ async fn main() -> anyhow::Result<()> { model, resume, verbose, + json, sandbox, sandbox_strict, bg, @@ -329,6 +335,7 @@ async fn run( model: Option, resume: Option, verbose: bool, + json: bool, sandbox: bool, sandbox_strict: bool, bg: bool, @@ -354,7 +361,11 @@ async fn run( } let pid = bg::spawn_detached(&session.id, &prompt, &extra)?; store.set_run_status(&session.id, "running", Some(pid as i64))?; - println!("dispatched {} (pid {pid})", &session.id[..8]); + if json { + json::emit(&json::dispatched_json(&session.id, pid)); + } else { + println!("dispatched {} (pid {pid})", &session.id[..8]); + } eprintln!( " bullpen agents watch it\n bullpen logs {} tail its output\n bullpen run -r {} \"...\" continue it", &session.id[..8], @@ -422,6 +433,11 @@ async fn run( let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); let printer = tokio::spawn(async move { while let Some(event) = rx.recv().await { + // One task owns stdout under --json, so event order is the order + // the loop produced them in. + if json { + json::emit(&json::event_json(&event)); + } if !verbose { continue; } @@ -442,6 +458,11 @@ async fn run( use std::io::Write; let mut out = std::io::stdout(); while let Some(text) = delta_rx.recv().await { + // The sink stays attached under --json so the agent still takes + // its streaming path; only the raw text is dropped. + if json { + continue; + } let _ = out.write_all(text.as_bytes()); let _ = out.flush(); } @@ -498,10 +519,29 @@ async fn run( let _ = store.set_run_status(&session.id, final_status, None); } + // Both consumer tasks have joined, so this is provably the last line of + // the stream. It carries the outcome outright: a consumer must never have + // to infer completion from stdout closing. + if json { + let (text, error) = match &result { + Ok(text) => (text.as_str(), None), + Err(e @ AgentError::Truncated { partial }) => (partial.as_str(), Some(e.to_string())), + Err(e) => ("", Some(e.to_string())), + }; + json::emit(&json::result_json( + &session.id, + text, + usage, + error.as_deref(), + )); + } + match result { Ok(_) => { // The answer already streamed to stdout; just close the line. - println!(); + if !json { + println!(); + } eprintln!( "[session {} · {} · {} in / {} out tokens]", &session.id[..8], @@ -512,7 +552,9 @@ async fn run( Ok(()) } Err(e) => { - println!(); + if !json { + println!(); + } bail!("agent error (session {} saved): {e}", &session.id[..8]) } }