hyperhive/hive-claude/src/driver.rs

//! The subprocess driver: spawn claude, pump + classify its streams, and
//! assemble the result.

use std::collections::VecDeque;
use std::process::Stdio;

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{ChildStderr, ChildStdout, Command};

use crate::classify::Sentinels;
use crate::{Attach, Config, Error, Result, Sink};

/// Default program name spawned when [`Config::program`] is unset.
const DEFAULT_PROGRAM: &str = "claude";

/// How many trailing stderr lines to keep for [`Error::Exit`].
const STDERR_TAIL_LINES: usize = 20;

/// The low-level driver entry point. A namespace for the run function — there
/// is nothing to construct; call `Claude::run(…)` directly. For a durable,
/// self-compacting session, use [`crate::InfiniteSession`] instead.
pub struct Claude;

impl Claude {
    /// Spawn one headless `claude --print` turn, stream its output through
    /// `sink`, and report the result.
    ///
    /// The wake prompt is written to claude's stdin. stdout (`stream-json`)
    /// and stderr are pumped concurrently while the child runs. A clean turn
    /// returns `Ok(())`; every non-completion state — recognized sentinel or
    /// hard failure — is an [`Error`] variant, so callers branch with a single
    /// `match`.
    ///
    /// # Errors
    ///
    /// - A recognized sentinel: [`Error::PromptTooLong`], [`Error::RateLimited`],
    ///   [`Error::AuthFailed`], [`Error::SessionNotFound`].
    /// - [`Error::Spawn`] if the binary can't be launched.
    /// - [`Error::Stdin`] / [`Error::Wait`] on stdin-write / child-wait failure.
    /// - [`Error::Exit`] on a non-zero exit that raised no sentinel.
    pub async fn run(
        config: &Config,
        attach: &Attach,
        prompt: &str,
        sink: &impl Sink,
    ) -> Result<()> {
        let program = config.program.as_deref().unwrap_or(DEFAULT_PROGRAM);
        let mut cmd = build_command(program, config, attach);

        let mut child = cmd
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .map_err(|source| Error::Spawn {
                program: program.to_string(),
                source,
            })?;

        if let Some(mut stdin) = child.stdin.take() {
            stdin
                .write_all(prompt.as_bytes())
                .await
                .map_err(Error::Stdin)?;
            // Best-effort flush/close; claude sees EOF and starts the turn.
            stdin.shutdown().await.ok();
        }
        let stdout = child.stdout.take().expect("stdout piped");
        let stderr = child.stderr.take().expect("stderr piped");

        let sentinels = Sentinels::default();
        // Pump both streams and wait for exit concurrently on this task — no
        // `spawn`, so the sink needn't be `'static` and borrows stay simple.
        let ((), stderr_tail, status) = tokio::join!(
            pump_stdout(stdout, sink, &sentinels),
            pump_stderr(stderr, sink, &sentinels),
            child.wait(),
        );
        let status = status.map_err(Error::Wait)?;

        // A recognized sentinel takes precedence over the exit code; otherwise
        // a non-zero exit with no sentinel is a hard failure.
        if let Some(sentinel) = sentinels.soft_error() {
            return Err(sentinel);
        }
        if !status.success() {
            return Err(Error::Exit {
                status,
                stderr_tail,
            });
        }
        Ok(())
    }
}

/// Assemble the argv. `--print --verbose --output-format stream-json` are
/// mandatory (the driver parses that shape); everything else is gated on the
/// [`Config`] / [`Attach`].
fn build_command(program: &str, config: &Config, attach: &Attach) -> Command {
    let mut cmd = Command::new(program);
    if let Some(cwd) = &config.cwd {
        cmd.current_dir(cwd);
    }
    cmd.arg("--print")
        .arg("--verbose")
        .arg("--output-format")
        .arg("stream-json");
    if !config.model.is_empty() {
        cmd.arg("--model").arg(&config.model);
    }
    if let Some(effort) = &config.effort {
        cmd.arg("--effort").arg(effort);
    }
    match attach {
        Attach::Resume(id) => {
            cmd.arg("--resume").arg(id);
        }
        Attach::Create(title) => {
            cmd.arg("--name").arg(title);
        }
        Attach::Continue => {
            cmd.arg("--continue");
        }
        Attach::OneOff => {}
    }
    if let Some(path) = &config.system_prompt_file {
        cmd.arg("--system-prompt-file").arg(path);
    }
    if let Some(path) = &config.mcp_config {
        cmd.arg("--mcp-config").arg(path);
    }
    if config.strict_mcp_config {
        cmd.arg("--strict-mcp-config");
    }
    if let Some(tools) = &config.tools {
        cmd.arg("--tools").arg(tools);
    }
    if let Some(allowed) = &config.allowed_tools {
        cmd.arg("--allowedTools").arg(allowed);
    }
    for dir in &config.add_dirs {
        cmd.arg("--add-dir").arg(dir);
    }
    for extra in &config.extra_args {
        cmd.arg(extra);
    }
    cmd
}

/// Read stdout line by line: classify each line, parse JSON, hand events (or
/// raw non-JSON lines) to the sink.
async fn pump_stdout(stdout: ChildStdout, sink: &impl Sink, sentinels: &Sentinels) {
    let mut lines = BufReader::new(stdout).lines();
    while let Ok(Some(line)) = lines.next_line().await {
        if let Ok(event) = serde_json::from_str::<serde_json::Value>(&line) {
            // JSON stdout: classify with the model-content gate so an
            // `assistant`/`user` message quoting a marker can't trip it.
            sentinels.scan_stdout_json(&event, &line);
            sink.on_event(&event);
        } else {
            // Non-JSON stdout is CLI text, not conversation — trust all markers.
            sentinels.scan_cli_line(&line);
            sink.on_stdout_line(&line);
        }
    }
}

/// Read stderr line by line: classify, forward to the sink, and retain the
/// last [`STDERR_TAIL_LINES`] for a possible [`Error::Exit`]. Returns the
/// newline-joined tail.
async fn pump_stderr(stderr: ChildStderr, sink: &impl Sink, sentinels: &Sentinels) -> String {
    let mut lines = BufReader::new(stderr).lines();
    let mut tail: VecDeque<String> = VecDeque::with_capacity(STDERR_TAIL_LINES);
    while let Ok(Some(line)) = lines.next_line().await {
        // stderr is always CLI output — trust all markers.
        sentinels.scan_cli_line(&line);
        sink.on_stderr_line(&line);
        if tail.len() >= STDERR_TAIL_LINES {
            tail.pop_front();
        }
        tail.push_back(line);
    }
    tail.into_iter().collect::<Vec<_>>().join("\n")
}