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]) } }