diff --git a/CHANGELOG.md b/CHANGELOG.md index ca2bc62..0e3ad86 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,15 @@ +## [0.1.8] - 2026-05-21 + +### Fixed + +- Local CLI-agent providers now drain stdout and stderr while the child process runs, preventing verbose Codex or Claude Code executions from timing out because an output pipe filled before process exit. +- CLI-agent failure diagnostics now keep bounded stdout and stderr captures and report whether either stream was truncated. +- Truncated successful stdout from local CLI agents now returns a clear invalid-provider-response error instead of a misleading JSON parse failure. + +### Changed + +- Documented that Codex `temperature` and output-token settings are not emitted as unsupported CLI flags; runtime `args` remain the escape hatch for documented Codex config overrides. + ## [0.1.7] - 2026-05-18 ### Fixed diff --git a/Cargo.lock b/Cargo.lock index 2e6e662..77aa8da 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -80,7 +80,7 @@ checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af" [[package]] name = "bitloops-inference" -version = "0.1.7" +version = "0.1.8" dependencies = [ "assert_cmd", "bitloops-inference-protocol", @@ -98,7 +98,7 @@ dependencies = [ [[package]] name = "bitloops-inference-protocol" -version = "0.1.7" +version = "0.1.8" dependencies = [ "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index dd58423..b3dd3b0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,7 +6,7 @@ members = [ resolver = "2" [workspace.package] -version = "0.1.7" +version = "0.1.8" edition = "2024" license = "Apache-2.0" diff --git a/README.md b/README.md index 576e944..b808324 100644 --- a/README.md +++ b/README.md @@ -100,7 +100,9 @@ thinking_level = "extra_high" `thinking_level` is optional and only supported by local CLI-agent drivers. For `codex_exec`, supported values are `low`, `medium`, `high`, `extra_high`, and `xhigh`; both `extra_high` and `xhigh` run Codex with `model_reasoning_effort="xhigh"`. For `claude_code_print`, supported values are Claude Code's native effort names: `low`, `medium`, `high`, `xhigh`, and `max`. -`codex_exec` writes a temporary JSON Schema file, runs `codex exec --output-schema --output-last-message `, and returns the parsed result file as `parsed_json`. `claude_code_print` runs `claude -p --model --output-format json --input-format text --json-schema --allowedTools Read,Grep,Glob`, writes the combined prompt to stdin, and returns Claude Code's JSON output as `parsed_json`. When `thinking_level` is present, it is passed as provider-specific CLI configuration. The `--json-schema` argument is included when the inference request metadata contains `json_schema`. +`codex_exec` writes a temporary JSON Schema file, runs `codex exec --output-schema --output-last-message `, and returns the parsed result file as `parsed_json`. `claude_code_print` runs `claude -p --model --output-format json --input-format text --json-schema --allowedTools Read,Grep,Glob`, writes the combined prompt to stdin, and returns Claude Code's JSON output as `parsed_json`. When `thinking_level` is present, it is passed as provider-specific CLI configuration. Codex CLI config overrides that are not first-class Bitloops fields can be passed with runtime `args`; Bitloops does not emit unsupported Codex `temperature` or output-token flags. The `--json-schema` argument is included when the inference request metadata contains `json_schema`. + +Local CLI-agent stdout and stderr are drained while the child process is running, so verbose tools cannot block on full output pipes before exiting. Failure diagnostics include bounded stdout and stderr captures, with truncation flags when a stream exceeds the capture limit. ## How Bitloops calls it diff --git a/crates/bitloops-inference/src/provider/cli_agent.rs b/crates/bitloops-inference/src/provider/cli_agent.rs index f4529f8..5ca3fb4 100644 --- a/crates/bitloops-inference/src/provider/cli_agent.rs +++ b/crates/bitloops-inference/src/provider/cli_agent.rs @@ -1,7 +1,8 @@ use std::fs; -use std::io::Write; +use std::io::{Read, Write}; use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; +use std::sync::mpsc; use std::thread; use std::time::{Duration, Instant}; @@ -30,6 +31,9 @@ impl CodexExecProvider { let command = self.runtime_command()?; let mut args = self.profile.runtime_args.clone(); args.push("exec".to_string()); + // Codex CLI currently exposes reasoning effort as a config override. Other + // Bitloops profile fields such as temperature and max_output_tokens are not + // emitted here because Codex does not document stable config keys for them. if let Some(thinking_level) = self.profile.thinking_level { let Some(effort) = thinking_level.codex_reasoning_effort() else { return Err(ProviderError::invalid_config( @@ -103,6 +107,8 @@ impl InferenceProvider for CodexExecProvider { Some(json!({ "path": result_path.display().to_string() })), ) })? + } else if output.stdout_truncated { + return Err(truncated_stdout_error("codex_exec", output.stdout)); } else { output.stdout.clone() }; @@ -177,6 +183,9 @@ impl InferenceProvider for ClaudeCodePrintProvider { fn infer(&self, request: &InferenceRequest) -> Result { let command = self.build_command(request)?; let output = run_cli_command(command, self.profile.timeout_secs)?; + if output.stdout_truncated { + return Err(truncated_stdout_error("claude_code_print", output.stdout)); + } let parsed_json = parse_claude_json(&output.stdout)?; Ok(InferenceResponse { @@ -221,6 +230,16 @@ struct CliCommand { #[derive(Debug, Clone, PartialEq, Eq)] struct CliOutput { stdout: String, + stdout_truncated: bool, +} + +const MAX_CLI_CAPTURE_BYTES: usize = 4 * 1024 * 1024; +const OUTPUT_COLLECT_TIMEOUT: Duration = Duration::from_secs(1); + +#[derive(Debug, Clone, PartialEq, Eq)] +struct CapturedStream { + text: String, + truncated: bool, } fn prompt_for_cli_agent(request: &InferenceRequest) -> String { @@ -266,6 +285,68 @@ fn write_json_file(path: &Path, value: &Value) -> Result<(), ProviderError> { }) } +fn spawn_stream_reader( + command_name: &str, + stream_name: &'static str, + mut stream: R, +) -> mpsc::Receiver> +where + R: Read + Send + 'static, +{ + let (sender, receiver) = mpsc::channel(); + let command_name = command_name.to_string(); + + thread::spawn(move || { + let mut bytes = Vec::new(); + let mut truncated = false; + let mut buffer = [0_u8; 8192]; + + loop { + match stream.read(&mut buffer) { + Ok(0) => break, + Ok(count) => { + let remaining = MAX_CLI_CAPTURE_BYTES.saturating_sub(bytes.len()); + if remaining > 0 { + let keep = remaining.min(count); + bytes.extend_from_slice(&buffer[..keep]); + } + if count > remaining { + truncated = true; + } + } + Err(err) => { + let _ = sender.send(Err(format!( + "failed to read {stream_name} from CLI agent `{command_name}`: {err}" + ))); + return; + } + } + } + + let _ = sender.send(Ok(CapturedStream { + text: String::from_utf8_lossy(&bytes).to_string(), + truncated, + })); + }); + + receiver +} + +fn collect_stream_output( + receiver: mpsc::Receiver>, + command_name: &str, + stream_name: &str, +) -> Result { + receiver + .recv_timeout(OUTPUT_COLLECT_TIMEOUT) + .map_err(|err| { + ProviderError::provider_transport_error(format!( + "failed to collect {stream_name} from CLI agent `{command_name}`: {err}" + )) + })? + .map_err(ProviderError::provider_transport_error) +} + fn run_cli_command(command: CliCommand, timeout_secs: u64) -> Result { let mut process = Command::new(&command.command); process.args(&command.args); @@ -287,6 +368,22 @@ fn run_cli_command(command: CliCommand, timeout_secs: u64) -> Result Result Result> = None; let deadline = Instant::now() + Duration::from_secs(timeout_secs); - loop { + let status = loop { if stdin_result.is_none() && let Some(receiver) = &stdin_result_receiver { match receiver.try_recv() { Ok(result) => stdin_result = Some(result), - Err(std::sync::mpsc::TryRecvError::Empty) => {} - Err(std::sync::mpsc::TryRecvError::Disconnected) => { + Err(mpsc::TryRecvError::Empty) => {} + Err(mpsc::TryRecvError::Disconnected) => { stdin_result = Some(Err(format!( "failed to write stdin to CLI agent `{}`: stdin writer stopped", command.command @@ -326,7 +423,7 @@ fn run_cli_command(command: CliCommand, timeout_secs: u64) -> Result break, + Ok(Some(status)) => break status, Ok(None) if Instant::now() >= deadline => { let _ = child.kill(); let _ = child.wait(); @@ -344,7 +441,7 @@ fn run_cli_command(command: CliCommand, timeout_secs: u64) -> Result Result Result Result { @@ -389,6 +486,17 @@ fn parse_cli_json(text: &str) -> Result { }) } +fn truncated_stdout_error(provider_name: &str, stdout: String) -> ProviderError { + ProviderError::invalid_provider_response( + format!("{provider_name} stdout exceeded the CLI capture limit"), + Some(json!({ + "stdout": stdout, + "stdout_truncated": true, + "max_capture_bytes": MAX_CLI_CAPTURE_BYTES, + })), + ) +} + fn parse_claude_json(text: &str) -> Result { let outer = parse_cli_json(text)?; if let Some(structured_output) = outer.get("structured_output") { @@ -515,6 +623,49 @@ mod tests { ); } + #[test] + fn codex_exec_preserves_runtime_args_before_exec_subcommand() { + let provider = CodexExecProvider::new(profile( + ProviderKind::CodexExec, + "codex", + &["-c", "model_verbosity=\"low\""], + )); + let command = provider + .build_command( + &request(), + Path::new("/tmp/schema.json"), + Path::new("/tmp/result.json"), + ) + .expect("command"); + + assert_eq!( + &command.args[..4], + &[ + "-c".to_string(), + "model_verbosity=\"low\"".to_string(), + "exec".to_string(), + "--model".to_string(), + ] + ); + } + + #[test] + fn codex_exec_does_not_emit_unsupported_temperature_or_token_args() { + let provider = CodexExecProvider::new(profile(ProviderKind::CodexExec, "codex", &[])); + let command = provider + .build_command( + &request(), + Path::new("/tmp/schema.json"), + Path::new("/tmp/result.json"), + ) + .expect("command"); + + let joined = command.args.join(" "); + assert!(!joined.contains("temperature")); + assert!(!joined.contains("max_output_tokens")); + assert!(!joined.contains("model_max_output_tokens")); + } + #[test] fn claude_code_print_builds_stdin_command_with_model_schema_and_tools() { let provider = @@ -580,6 +731,18 @@ mod tests { assert_eq!(error.code, "invalid_provider_response"); } + #[test] + fn truncated_stdout_error_includes_capture_limit() { + let error = truncated_stdout_error("claude_code_print", "partial".to_string()); + + assert_eq!(error.code, "invalid_provider_response"); + assert!(error.message.contains("claude_code_print")); + let details = error.details.expect("details"); + assert_eq!(details["stdout"], "partial"); + assert_eq!(details["stdout_truncated"], true); + assert_eq!(details["max_capture_bytes"], MAX_CLI_CAPTURE_BYTES); + } + #[test] fn driver_failure_includes_status_and_stderr() { let error = run_cli_command( @@ -603,6 +766,34 @@ mod tests { assert_eq!(details["stderr"], "stderr text\n"); } + #[test] + fn driver_failure_with_large_stderr_returns_bounded_diagnostics() { + let error = run_cli_command( + CliCommand { + command: "/bin/sh".to_string(), + args: vec![ + "-c".to_string(), + "echo stdout text; dd if=/dev/zero bs=1024 count=5120 2>/dev/null | tr '\\000' e >&2; exit 7".to_string(), + ], + cwd: None, + stdin: None, + }, + 5, + ) + .expect_err("failing command should return provider error"); + + assert_eq!(error.code, "invalid_provider_response"); + let details = error.details.expect("details"); + assert!(details["status"].as_str().unwrap().contains('7')); + assert_eq!(details["stdout"], "stdout text\n"); + assert_eq!(details["stdout_truncated"], false); + assert_eq!(details["stderr_truncated"], true); + assert_eq!( + details["stderr"].as_str().expect("stderr").len(), + MAX_CLI_CAPTURE_BYTES + ); + } + #[test] fn run_cli_command_writes_configured_stdin() { let output = run_cli_command( @@ -619,6 +810,45 @@ mod tests { assert_eq!(output.stdout, "system\n\nuser"); } + #[test] + fn run_cli_command_drains_large_stderr_before_waiting_for_exit() { + let output = run_cli_command( + CliCommand { + command: "/bin/sh".to_string(), + args: vec![ + "-c".to_string(), + "printf '{\"summary\":\"ok\"}'; dd if=/dev/zero bs=1024 count=1024 2>/dev/null | tr '\\000' x >&2".to_string(), + ], + cwd: None, + stdin: None, + }, + 2, + ) + .expect("large stderr should not block child exit"); + + assert_eq!(output.stdout, "{\"summary\":\"ok\"}"); + } + + #[test] + fn run_cli_command_drains_large_stdout_before_waiting_for_exit() { + let output = run_cli_command( + CliCommand { + command: "/bin/sh".to_string(), + args: vec![ + "-c".to_string(), + "printf '{\"summary\":\"ok\"}\\n'; dd if=/dev/zero bs=1024 count=1024 2>/dev/null | tr '\\000' x".to_string(), + ], + cwd: None, + stdin: None, + }, + 2, + ) + .expect("large stdout should not block child exit"); + + assert!(output.stdout.starts_with("{\"summary\":\"ok\"}\n")); + assert!(output.stdout.len() >= 1024 * 1024); + } + #[test] fn run_cli_command_timeout_includes_blocked_stdin_write() { let (sender, receiver) = std::sync::mpsc::channel();