From 4f587fb526355948630cda427c1144c79542cd0f Mon Sep 17 00:00:00 2001 From: Jakob Stender Guldberg Date: Tue, 8 Sep 2026 13:40:16 +0200 Subject: [PATCH] fix(llm): stop CLI backends re-reading the repo and blowing the argv limit Two independent problems in the `claude` and `codex` CLI backends. The prompt was passed as a single argv entry. Linux caps one argument at MAX_ARG_STRLEN (128 KiB), and a group's diff plus file excerpts routinely exceeds that, so "analyze this flow" failed the spawn outright with `Argument list too long` (E2BIG) before the model was ever reached. Prompts now go on stdin, which has no such limit. Separately, the agent addendum told every spawn to "use your built-in read/search tools to inspect files, plans, and git state" even though the prompt already carries the diffs and file excerpts. Each pass therefore paid to re-derive context it had just been handed. Measured on an identical prompt, dropping that instruction and the unreachable MCP, plugin and slash-command prefix took a pass from 57,475 to 26,191 prefix tokens (-54%), with one fewer tool call and turn. Also bounds the previously unbounded metadata batch fan-out, kills child processes when their future is dropped, logs each spawn at debug level, and adds rustfmt and clippy to the devshell. --- crates/diffcore-core/src/llm/claude_cli.rs | 265 +++++++++++++-------- crates/diffcore-core/src/llm/codex_cli.rs | 35 ++- crates/diffcore-core/src/llm/metadata.rs | 55 ++++- flake.nix | 2 + 4 files changed, 249 insertions(+), 108 deletions(-) diff --git a/crates/diffcore-core/src/llm/claude_cli.rs b/crates/diffcore-core/src/llm/claude_cli.rs index 714a3d8..1c52263 100644 --- a/crates/diffcore-core/src/llm/claude_cli.rs +++ b/crates/diffcore-core/src/llm/claude_cli.rs @@ -1,4 +1,4 @@ -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::process::{Command as StdCommand, Stdio}; use async_trait::async_trait; @@ -15,9 +15,9 @@ use super::{ }; const AGENT_ADDENDUM: &str = - "You are running inside the target repository root. Use your built-in \ -read/search tools or safe inspection commands when needed to inspect files, plans, and git state. \ -Do not modify files. Return only the final JSON object that matches the provided schema."; + "The prompt already contains every diff and file excerpt you need. Do not read files, run \ +commands, or otherwise inspect the repository. Do not modify files. Return only the final JSON \ +object that matches the provided schema."; #[derive(Debug, Clone)] pub struct ClaudeCliProvider { @@ -65,97 +65,150 @@ impl ClaudeCliProvider { let schema_text = serde_json::to_string(&schema::flatten_json_schema(json_schema)) .map_err(|e| LlmError::ParseResponse(format!("Failed to serialize schema: {}", e)))?; - let mut command = Command::new(&claude_path); - command - .current_dir(self.workdir()) - .arg("--print") - .arg("--output-format") - .arg("stream-json") - .arg("--verbose") - .arg("--json-schema") - .arg(schema_text) - .arg("--system-prompt") - .arg(format!("{}\n\n{}", system_prompt, AGENT_ADDENDUM)) - .arg(user_prompt); - - if let Some(model) = self.selected_model() { - command.arg("--model").arg(model); - } + let workdir = self.workdir(); + let args = build_args(&schema_text, &system_prompt, self.selected_model()); + let stdout = run_claude(&claude_path, &workdir, &args, &user_prompt).await?; - command.stdout(Stdio::piped()).stderr(Stdio::piped()); - super::emit_activity(super::ActivityUpdate::info( - "claude", - "Launching Claude Code", - Some("claude.launch".to_string()), - )); + parse_structured_response(&stdout) + } +} - let mut child = command - .spawn() - .map_err(|e| LlmError::CommandFailed(format!("Failed to launch claude: {}", e)))?; - let stdout = child.stdout.take().ok_or_else(|| { - LlmError::CommandFailed("Failed to capture claude stdout".to_string()) - })?; - let stderr = child.stderr.take().ok_or_else(|| { - LlmError::CommandFailed("Failed to capture claude stderr".to_string()) - })?; - let activity_callback = super::current_activity_callback(); - - let stdout_activity_callback = activity_callback.clone(); - let stdout_task = tokio::spawn(async move { - collect_claude_stream(stdout, "stdout", true, stdout_activity_callback).await - }); - let stderr_task = tokio::spawn(async move { - collect_claude_stream(stderr, "stderr", false, activity_callback).await - }); +fn build_args(schema_text: &str, system_prompt: &str, model: Option<&str>) -> Vec { + let mut args = vec![ + "--print".to_string(), + "--output-format".to_string(), + "stream-json".to_string(), + "--verbose".to_string(), + "--json-schema".to_string(), + schema_text.to_string(), + "--system-prompt".to_string(), + format!("{}\n\n{}", system_prompt, AGENT_ADDENDUM), + // The user's MCP servers, plugins and slash commands are loaded into every + // spawn's prefix and none of them are reachable from a --print run. + "--strict-mcp-config".to_string(), + "--mcp-config".to_string(), + r#"{"mcpServers":{}}"#.to_string(), + "--disable-slash-commands".to_string(), + ]; + + if let Some(model) = model { + args.push("--model".to_string()); + args.push(model.to_string()); + } - let timeout_sleep = sleep(Duration::from_secs(cli_timeout_secs())); - tokio::pin!(timeout_sleep); + args +} - let status = tokio::select! { - result = child.wait() => result.map_err(|e| LlmError::CommandFailed(format!("Failed to wait for claude: {}", e)))?, - _ = &mut timeout_sleep => { - let _ = child.kill().await; - return Err(LlmError::Timeout(cli_timeout_secs())); - } - }; +fn parse_structured_response(stdout: &str) -> Result { + let structured = parse_claude_structured_output(stdout).ok_or_else(|| { + LlmError::ParseResponse(format!( + "Claude response did not include structured_output: {}", + redact_api_keys(&truncate_to_token_budget(stdout, 400)) + )) + })?; + + serde_json::from_value(structured).map_err(|e| { + LlmError::ParseResponse(format!( + "Failed to parse Claude structured output: {} — response: {}", + e, + redact_api_keys(&truncate_to_token_budget(stdout, 400)) + )) + }) +} - let stdout = stdout_task - .await - .map_err(|e| { - LlmError::CommandFailed(format!("Failed to join claude stdout task: {}", e)) - })? - .map_err(|e| LlmError::CommandFailed(format!("Failed to read claude stdout: {}", e)))?; - let stderr = stderr_task - .await - .map_err(|e| { - LlmError::CommandFailed(format!("Failed to join claude stderr task: {}", e)) - })? - .map_err(|e| LlmError::CommandFailed(format!("Failed to read claude stderr: {}", e)))?; - - if !status.success() { - let combined = format!("{}\n{}", stderr, stdout); - let message = redact_api_keys(&truncate_to_token_budget(&combined, 400)); - return Err(LlmError::CommandFailed(format!( - "claude --print exited with {}: {}", - status, message - ))); +async fn run_claude( + claude_path: &Path, + workdir: &Path, + args: &[String], + prompt: &str, +) -> Result { + let mut command = Command::new(claude_path); + command.current_dir(workdir).args(args); + command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + log::debug!( + target: "claude_cli", + "spawn {} with {} args (cwd {}, {} byte prompt on stdin)", + claude_path.display(), + args.len(), + workdir.display(), + prompt.len() + ); + super::emit_activity(super::ActivityUpdate::info( + "claude", + format!("Launching Claude Code ({} KiB prompt)", prompt.len() / 1024), + Some("claude.launch".to_string()), + )); + + let mut child = command + .spawn() + .map_err(|e| LlmError::CommandFailed(format!("Failed to launch claude: {}", e)))?; + + // The prompt goes on stdin, never argv: Linux caps a single argument at + // MAX_ARG_STRLEN (128 KiB) and a group's diff routinely exceeds that. + let mut stdin = child + .stdin + .take() + .ok_or_else(|| LlmError::CommandFailed("Failed to capture claude stdin".to_string()))?; + let prompt = prompt.to_string(); + let stdin_task = tokio::spawn(async move { + use tokio::io::AsyncWriteExt; + let _ = stdin.write_all(prompt.as_bytes()).await; + let _ = stdin.shutdown().await; + }); + + let stdout = child + .stdout + .take() + .ok_or_else(|| LlmError::CommandFailed("Failed to capture claude stdout".to_string()))?; + let stderr = child + .stderr + .take() + .ok_or_else(|| LlmError::CommandFailed("Failed to capture claude stderr".to_string()))?; + let activity_callback = super::current_activity_callback(); + + let stdout_activity_callback = activity_callback.clone(); + let stdout_task = tokio::spawn(async move { + collect_claude_stream(stdout, "stdout", true, stdout_activity_callback).await + }); + let stderr_task = tokio::spawn(async move { + collect_claude_stream(stderr, "stderr", false, activity_callback).await + }); + + let timeout_sleep = sleep(Duration::from_secs(cli_timeout_secs())); + tokio::pin!(timeout_sleep); + + let status = tokio::select! { + result = child.wait() => result.map_err(|e| LlmError::CommandFailed(format!("Failed to wait for claude: {}", e)))?, + _ = &mut timeout_sleep => { + let _ = child.kill().await; + return Err(LlmError::Timeout(cli_timeout_secs())); } + }; - let structured = parse_claude_structured_output(&stdout).ok_or_else(|| { - LlmError::ParseResponse(format!( - "Claude response did not include structured_output: {}", - redact_api_keys(&truncate_to_token_budget(&stdout, 400)) - )) - })?; - - serde_json::from_value(structured).map_err(|e| { - LlmError::ParseResponse(format!( - "Failed to parse Claude structured output: {} — response: {}", - e, - redact_api_keys(&truncate_to_token_budget(&stdout, 400)) - )) - }) + let _ = stdin_task.await; + let stdout = stdout_task + .await + .map_err(|e| LlmError::CommandFailed(format!("Failed to join claude stdout task: {}", e)))? + .map_err(|e| LlmError::CommandFailed(format!("Failed to read claude stdout: {}", e)))?; + let stderr = stderr_task + .await + .map_err(|e| LlmError::CommandFailed(format!("Failed to join claude stderr task: {}", e)))? + .map_err(|e| LlmError::CommandFailed(format!("Failed to read claude stderr: {}", e)))?; + + if !status.success() { + let combined = format!("{}\n{}", stderr, stdout); + let message = redact_api_keys(&truncate_to_token_budget(&combined, 400)); + return Err(LlmError::CommandFailed(format!( + "claude --print exited with {}: {}", + status, message + ))); } + + Ok(stdout) } pub fn cli_timeout_secs() -> u64 { @@ -509,16 +562,11 @@ fn extract_claude_structured_output(parsed: &serde_json::Value) -> Option = requests .into_iter() .map(|request| { let provider = Arc::clone(&provider); - tokio::spawn(async move { provider.describe_groups(&request).await }) + let permits = Arc::clone(&permits); + tokio::spawn(async move { + let _permit = permits.acquire_owned().await; + provider.describe_groups(&request).await + }) }) .collect(); @@ -354,6 +363,8 @@ pub fn metadata_user_prompt(request: &MetadataRequest) -> String { clippy::print_stderr )] mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + use super::*; use crate::git::{DiffHunk, FileStatus}; use crate::types::{ @@ -594,10 +605,14 @@ mod tests { } /// A provider that answers every batch, after a delay that inverts completion - /// order relative to dispatch order. + /// order relative to dispatch order, recording the peak number of calls it was + /// ever handling at once. + #[derive(Default)] struct BatchingProvider { batches_seen: Arc>>, delay_ms: u64, + in_flight: AtomicUsize, + max_in_flight: AtomicUsize, } #[async_trait::async_trait] @@ -644,10 +659,13 @@ mod tests { .trim_start_matches('g') .parse() .unwrap(); + let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1; + self.max_in_flight.fetch_max(now, Ordering::SeqCst); tokio::time::sleep(std::time::Duration::from_millis( self.delay_ms * (10 - first as u64), )) .await; + self.in_flight.fetch_sub(1, Ordering::SeqCst); if let Ok(mut seen) = self.batches_seen.lock() { seen.push(first); } @@ -674,6 +692,7 @@ mod tests { let provider = Arc::new(BatchingProvider { batches_seen: Arc::new(std::sync::Mutex::new(Vec::new())), delay_ms: 5, + ..Default::default() }); run_metadata_pass(provider.clone(), &mut groups, &[], 3) @@ -698,6 +717,32 @@ mod tests { assert_eq!(seen, vec![6, 3, 0], "batches completed out of dispatch order"); } + #[tokio::test] + async fn batch_fan_out_stays_within_the_concurrency_cap() { + let mut groups: Vec = (0..9) + .map(|i| make_group(&format!("g{}", i), i, &["src/a.ts"])) + .collect(); + let provider = Arc::new(BatchingProvider { + batches_seen: Arc::new(std::sync::Mutex::new(Vec::new())), + delay_ms: 2, + ..Default::default() + }); + + run_metadata_pass(provider.clone(), &mut groups, &[], 1) + .await + .unwrap(); + + let peak = provider.max_in_flight.load(Ordering::SeqCst); + assert!( + peak <= MAX_CONCURRENT_BATCHES, + "{} batches in flight at once, cap is {}", + peak, + MAX_CONCURRENT_BATCHES + ); + assert!(peak > 1, "batches must still overlap"); + assert!(groups.iter().all(|g| g.description.is_some())); + } + struct FailingProvider; #[async_trait::async_trait] diff --git a/flake.nix b/flake.nix index 3b4b399..5545f7c 100644 --- a/flake.nix +++ b/flake.nix @@ -49,6 +49,8 @@ packages = [ pkgs.sccache pkgs.mold + pkgs.rustfmt + pkgs.clippy pkgs.nodejs pkgs.playwright-driver.browsers ];