sqlite_graphrag/commands/
claude_runner.rs

1//! Shared module for spawning Claude Code (`claude -p`) subprocesses.
2//!
3//! Eliminates duplication between `enrich.rs` and `ingest_claude.rs` (G02).
4//! Detects `terminal_reason: "max_turns"` in the JSON output (G03).
5
6use crate::errors::AppError;
7use std::path::Path;
8use std::process::{Command, Stdio};
9
10/// Minimum Claude Code version required for structured JSON output.
11const MIN_CLAUDE_VERSION: &str = "2.1.0";
12
13/// Environment variables whitelisted for the subprocess.
14const ENV_WHITELIST: &[&str] = &[
15    "PATH",
16    "HOME",
17    "USER",
18    "SHELL",
19    "TERM",
20    "LANG",
21    "XDG_CONFIG_HOME",
22    "XDG_DATA_HOME",
23    "XDG_RUNTIME_DIR",
24    "ANTHROPIC_API_KEY",
25    "CLAUDE_CONFIG_DIR",
26    "TMPDIR",
27    "TMP",
28    "TEMP",
29    "DYLD_FALLBACK_LIBRARY_PATH",
30];
31
32/// Windows-only environment variables.
33#[cfg(windows)]
34const ENV_WHITELIST_WINDOWS: &[&str] = &[
35    "LOCALAPPDATA",
36    "APPDATA",
37    "USERPROFILE",
38    "SystemRoot",
39    "COMSPEC",
40    "PATHEXT",
41    "HOMEPATH",
42    "HOMEDRIVE",
43];
44
45/// Default virtual memory limit for LLM subprocesses (4 GiB).
46const DEFAULT_SUBPROCESS_MEMORY_LIMIT_MB: u64 = 4096;
47
48/// Spawns a command with a virtual memory limit via `setrlimit(RLIMIT_AS)`.
49///
50/// On Linux, applies the limit in a `pre_exec` hook before the child process
51/// starts.  On non-Linux platforms, falls back to an unlimited spawn.
52/// The limit is read from `SQLITE_GRAPHRAG_SUBPROCESS_MEMORY_LIMIT_MB`
53/// (default: 4096 MiB).
54#[cfg(target_os = "linux")]
55pub fn spawn_with_memory_limit(cmd: &mut Command) -> std::io::Result<std::process::Child> {
56    use std::os::unix::process::CommandExt;
57    let max_mb: u64 = std::env::var("SQLITE_GRAPHRAG_SUBPROCESS_MEMORY_LIMIT_MB")
58        .ok()
59        .and_then(|v| v.parse().ok())
60        .unwrap_or(DEFAULT_SUBPROCESS_MEMORY_LIMIT_MB);
61    let max_bytes = max_mb * 1024 * 1024;
62    // SAFETY: pre_exec closure runs between fork() and exec() in the
63    // single-threaded child process — no other threads exist.
64    // libc::setsid and libc::setrlimit are async-signal-safe per POSIX.1-2008 §2.4.3.
65    // RLIMIT_AS limits virtual address space, not physical RSS.
66    // setsid failure with EPERM is tolerated (process already a session leader).
67    // On setrlimit failure, Err(last_os_error()) prevents exec.
68    unsafe {
69        cmd.pre_exec(move || {
70            let sid = libc::setsid();
71            if sid == -1 {
72                let err = std::io::Error::last_os_error();
73                if err.raw_os_error() != Some(libc::EPERM) {
74                    return Err(err);
75                }
76            }
77            let limit = libc::rlimit {
78                rlim_cur: max_bytes,
79                rlim_max: max_bytes,
80            };
81            if libc::setrlimit(libc::RLIMIT_AS, &limit) != 0 {
82                return Err(std::io::Error::last_os_error());
83            }
84            Ok(())
85        });
86    }
87    tracing::debug!(
88        target: "process",
89        program = ?cmd.get_program(),
90        args = ?cmd.get_args().collect::<Vec<_>>(),
91        "spawning external process"
92    );
93    cmd.spawn()
94}
95
96/// Spawns a command without memory limits (non-Linux fallback).
97/// On Unix (macOS, FreeBSD), applies setsid for process group isolation.
98#[cfg(not(target_os = "linux"))]
99pub fn spawn_with_memory_limit(cmd: &mut Command) -> std::io::Result<std::process::Child> {
100    #[cfg(unix)]
101    {
102        use std::os::unix::process::CommandExt;
103        // SAFETY: setsid() is async-signal-safe per POSIX.1-2008 §2.4.3.
104        // Creates independent session for cascade termination.
105        unsafe {
106            cmd.pre_exec(|| {
107                let sid = libc::setsid();
108                if sid == -1 {
109                    let err = std::io::Error::last_os_error();
110                    if err.raw_os_error() != Some(libc::EPERM) {
111                        return Err(err);
112                    }
113                }
114                Ok(())
115            });
116        }
117    }
118    tracing::debug!(
119        target: "process",
120        program = ?cmd.get_program(),
121        args = ?cmd.get_args().collect::<Vec<_>>(),
122        "spawning external process"
123    );
124    cmd.spawn()
125}
126
127/// Parsed output element from `claude -p --output-format json`.
128#[derive(Debug, serde::Deserialize)]
129pub struct ClaudeOutputElement {
130    pub r#type: Option<String>,
131    pub subtype: Option<String>,
132    #[serde(default)]
133    pub is_error: bool,
134    pub structured_output: Option<serde_json::Value>,
135    pub result: Option<String>,
136    pub total_cost_usd: Option<f64>,
137    pub error: Option<String>,
138    pub terminal_reason: Option<String>,
139    #[serde(rename = "apiKeySource")]
140    pub api_key_source: Option<String>,
141}
142
143/// Result of a successful Claude invocation.
144#[derive(Debug)]
145pub struct ClaudeResult {
146    pub value: serde_json::Value,
147    pub cost_usd: f64,
148    pub is_oauth: bool,
149}
150
151/// Validates that the Claude binary meets the minimum version requirement.
152pub fn validate_claude_version(binary: &Path) -> Result<String, AppError> {
153    let resolved = which::which(binary).map_err(|_| {
154        AppError::Validation(format!(
155            "executable '{}' not found in PATH; ensure it is installed and accessible",
156            binary.display()
157        ))
158    })?;
159    let output = Command::new(&resolved)
160        .arg("--version")
161        .stdin(Stdio::null())
162        .stdout(Stdio::piped())
163        .stderr(Stdio::piped())
164        .output()
165        .map_err(AppError::Io)?;
166
167    if !output.status.success() {
168        return Err(AppError::Validation(
169            "failed to run 'claude --version'".to_string(),
170        ));
171    }
172
173    let version_str = String::from_utf8(output.stdout)
174        .map_err(|_| AppError::Validation("claude --version output is not UTF-8".to_string()))?;
175    let version = version_str.trim().to_string();
176    let numeric = version.split([' ', '(']).next().unwrap_or("").trim();
177
178    fn parse_semver(s: &str) -> Option<(u64, u64, u64)> {
179        let parts: Vec<&str> = s.splitn(3, '.').collect();
180        if parts.len() < 2 {
181            return None;
182        }
183        let major = parts[0].parse::<u64>().ok()?;
184        let minor = parts[1].parse::<u64>().ok()?;
185        let patch = parts
186            .get(2)
187            .and_then(|p| p.parse::<u64>().ok())
188            .unwrap_or(0);
189        Some((major, minor, patch))
190    }
191
192    if let (Some(actual), Some(min)) = (parse_semver(numeric), parse_semver(MIN_CLAUDE_VERSION)) {
193        if actual < min {
194            return Err(AppError::Validation(format!(
195                "Claude Code version {numeric} is below minimum required {MIN_CLAUDE_VERSION}"
196            )));
197        }
198    }
199
200    Ok(version)
201}
202
203/// Builds a `Command` for `claude -p` with least-privilege environment.
204pub fn build_claude_command(
205    binary: &Path,
206    prompt: &str,
207    json_schema: &str,
208    model: Option<&str>,
209    max_turns: u32,
210) -> Command {
211    let mut cmd = Command::new(binary);
212
213    cmd.env_clear();
214    for var in ENV_WHITELIST {
215        if let Ok(val) = std::env::var(var) {
216            cmd.env(var, val);
217        }
218    }
219
220    #[cfg(windows)]
221    for var in ENV_WHITELIST_WINDOWS {
222        if let Ok(val) = std::env::var(var) {
223            cmd.env(var, val);
224        }
225    }
226
227    cmd.arg("-p")
228        .arg(prompt)
229        .arg("--output-format")
230        .arg("json")
231        .arg("--json-schema")
232        .arg(json_schema)
233        .arg("--max-turns")
234        .arg(max_turns.to_string())
235        .arg("--no-session-persistence");
236
237    if std::env::var("ANTHROPIC_API_KEY").is_ok() {
238        cmd.arg("--bare");
239    } else {
240        cmd.arg("--dangerously-skip-permissions")
241            .arg("--settings")
242            .arg(r#"{"hooks":{}}"#);
243    }
244
245    if let Some(m) = model {
246        cmd.arg("--model").arg(m);
247    }
248
249    cmd.stdin(Stdio::null())
250        .stdout(Stdio::piped())
251        .stderr(Stdio::piped());
252
253    cmd
254}
255
256/// Parses `claude -p --output-format json` output array.
257///
258/// G03: detects `terminal_reason: "max_turns"` and returns a specific error
259/// instead of a generic failure message.
260pub fn parse_claude_output(stdout: &str) -> Result<ClaudeResult, AppError> {
261    let elements: Vec<ClaudeOutputElement> = serde_json::from_str(stdout).map_err(|e| {
262        AppError::Validation(format!("failed to parse claude output as JSON array: {e}"))
263    })?;
264
265    let is_oauth = elements
266        .iter()
267        .find(|e| e.r#type.as_deref() == Some("system") && e.subtype.as_deref() == Some("init"))
268        .and_then(|e| e.api_key_source.as_deref())
269        .map(|s| s == "none")
270        .unwrap_or(false);
271
272    let result_elem = elements
273        .iter()
274        .find(|e| e.r#type.as_deref() == Some("result"))
275        .ok_or_else(|| {
276            AppError::Validation("claude output missing 'result' element".to_string())
277        })?;
278
279    // G03: detect max_turns exhaustion before checking is_error
280    if result_elem.terminal_reason.as_deref() == Some("max_turns") {
281        tracing::warn!(
282            target: "claude_runner",
283            "claude -p hit max_turns limit — hooks may have consumed turns"
284        );
285        return Err(AppError::Validation(
286            "claude -p hit max_turns: hooks may be consuming turns; increase --max-turns or disable hooks".to_string(),
287        ));
288    }
289
290    if result_elem.is_error {
291        let err_msg = result_elem
292            .error
293            .as_deref()
294            .or(result_elem.result.as_deref())
295            .unwrap_or("unknown error");
296        if err_msg.contains("rate_limit") || err_msg.contains("overloaded") {
297            return Err(AppError::RateLimited {
298                detail: err_msg.to_string(),
299            });
300        }
301        if err_msg.contains("Not logged in") || err_msg.contains("authentication") {
302            tracing::warn!(
303                target: "claude_runner",
304                "Claude Code authentication failed. Re-authenticate interactively with: claude"
305            );
306        }
307        return Err(AppError::Validation(format!(
308            "claude extraction failed: {err_msg}"
309        )));
310    }
311
312    let value = if let Some(v) = result_elem.structured_output.clone() {
313        v
314    } else if let Some(text) = &result_elem.result {
315        serde_json::from_str(text).map_err(|e| {
316            AppError::Validation(format!("failed to parse claude result field as JSON: {e}"))
317        })?
318    } else {
319        return Err(AppError::Validation(
320            "claude result missing structured_output and result field".into(),
321        ));
322    };
323
324    let cost = result_elem.total_cost_usd.unwrap_or(0.0);
325    Ok(ClaudeResult {
326        value,
327        cost_usd: cost,
328        is_oauth,
329    })
330}
331
332/// Calls `claude -p` with prompt and schema, waits with timeout, and parses output.
333///
334/// G03: parses stdout even on non-zero exit to detect `terminal_reason: "max_turns"`.
335pub fn run_claude(
336    binary: &Path,
337    prompt: &str,
338    json_schema: &str,
339    input_text: &str,
340    model: Option<&str>,
341    timeout_secs: u64,
342    max_turns: u32,
343) -> Result<ClaudeResult, AppError> {
344    use wait_timeout::ChildExt;
345
346    let full_prompt = format!("{prompt}\n\n{input_text}");
347    let mut cmd = build_claude_command(binary, &full_prompt, json_schema, model, max_turns);
348
349    let mut child = spawn_with_memory_limit(&mut cmd).map_err(|e| {
350        AppError::Io(std::io::Error::new(
351            e.kind(),
352            format!("failed to spawn claude: {e}"),
353        ))
354    })?;
355
356    let start = std::time::Instant::now();
357    let timeout = std::time::Duration::from_secs(timeout_secs);
358    let status = child.wait_timeout(timeout).map_err(AppError::Io)?;
359
360    match status {
361        Some(exit_status) => {
362            tracing::debug!(
363                target: "process",
364                exit_code = ?exit_status.code(),
365                elapsed_ms = start.elapsed().as_millis() as u64,
366                "external process completed"
367            );
368
369            let mut stdout_buf = Vec::new();
370            let mut stderr_buf = Vec::new();
371            if let Some(mut out) = child.stdout.take() {
372                std::io::Read::read_to_end(&mut out, &mut stdout_buf).map_err(AppError::Io)?;
373            }
374            if let Some(mut err) = child.stderr.take() {
375                std::io::Read::read_to_end(&mut err, &mut stderr_buf).map_err(AppError::Io)?;
376            }
377
378            let stdout_str = String::from_utf8(stdout_buf)
379                .map_err(|_| AppError::Validation("claude -p stdout is not valid UTF-8".into()))?;
380
381            // G03: parse stdout even on failure to detect terminal_reason
382            if !exit_status.success() {
383                if let Ok(result) = parse_claude_output(&stdout_str) {
384                    return Ok(result);
385                }
386                let stderr_str = String::from_utf8_lossy(&stderr_buf);
387                if stderr_str.contains("auth") || stderr_str.contains("login") {
388                    tracing::warn!(
389                        target: "claude_runner",
390                        "Claude Code authentication may have failed. Re-authenticate with: claude"
391                    );
392                }
393                return Err(AppError::Validation(format!(
394                    "claude -p exited with code {:?}: {}",
395                    exit_status.code(),
396                    stderr_str.trim()
397                )));
398            }
399
400            parse_claude_output(&stdout_str)
401        }
402        None => {
403            tracing::warn!(target: "claude_runner", timeout_secs, "claude -p timed out, terminating");
404            terminate_gracefully(&mut child, 3);
405            Err(AppError::Validation(format!(
406                "claude -p timed out after {timeout_secs} seconds"
407            )))
408        }
409    }
410}
411
412/// Terminates a child process gracefully: SIGTERM first, SIGKILL after grace period.
413#[cfg(unix)]
414pub fn terminate_gracefully(child: &mut std::process::Child, grace_secs: u64) {
415    use wait_timeout::ChildExt;
416    unsafe {
417        libc::kill(child.id() as i32, libc::SIGTERM);
418    }
419    match child.wait_timeout(std::time::Duration::from_secs(grace_secs)) {
420        Ok(Some(_)) => {}
421        _ => {
422            tracing::warn!(target: "process", pid = child.id(), "child ignored SIGTERM, sending SIGKILL");
423            let _ = child.kill();
424            let _ = child.wait();
425        }
426    }
427}
428
429/// Non-Unix fallback: kill immediately (Windows TerminateProcess).
430#[cfg(not(unix))]
431pub fn terminate_gracefully(child: &mut std::process::Child, _grace_secs: u64) {
432    let _ = child.kill();
433    let _ = child.wait();
434}
435
436#[cfg(test)]
437mod tests {
438    use super::*;
439
440    #[test]
441    fn parse_output_detects_max_turns() {
442        let stdout = r#"[{"type":"system","subtype":"init","apiKeySource":"none"},{"type":"result","is_error":false,"terminal_reason":"max_turns","structured_output":{"name":"t"}}]"#;
443        let err = parse_claude_output(stdout).unwrap_err();
444        assert!(
445            format!("{err}").contains("max_turns"),
446            "must detect max_turns in output"
447        );
448    }
449
450    #[test]
451    fn parse_output_extracts_structured_value() {
452        let stdout = r#"[{"type":"system","subtype":"init","apiKeySource":"none"},{"type":"result","is_error":false,"structured_output":{"key":"val"},"total_cost_usd":0.01}]"#;
453        let result = parse_claude_output(stdout).unwrap();
454        assert_eq!(result.value["key"], "val");
455        assert!((result.cost_usd - 0.01).abs() < f64::EPSILON);
456        assert!(result.is_oauth);
457    }
458
459    #[test]
460    fn parse_output_detects_rate_limit() {
461        let stdout = r#"[{"type":"result","is_error":true,"error":"rate_limit exceeded"}]"#;
462        let err = parse_claude_output(stdout).unwrap_err();
463        assert!(
464            matches!(err, AppError::RateLimited { .. }),
465            "expected AppError::RateLimited, got: {err}"
466        );
467    }
468}
sqlite_graphrag/commands/claude_runner.rs

sqlite_graphrag/commands/
claude_runner.rs
