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::{ExitStatus, Stdio};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, 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;

/// Idle-watchdog probe cadence: re-check output silence this often while the
/// child runs. Fine-grained enough to fire within ~one probe of the deadline,
/// cheap enough to ignore.
const IDLE_PROBE: Duration = Duration::from_secs(5);

/// 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::IdleTimeout`] if `config.idle_timeout` is set and no stdout
    ///   line arrives within that window (the child is killed).
    /// - [`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();
        // Idle watchdog clock: `last_activity` holds seconds-since-`base` of the
        // last stdout line, bumped by the pump; the waiter reads it to detect a
        // fully silent stall. Monotonic (`Instant`) so a wall-clock jump can't
        // spuriously fire it.
        let base = Instant::now();
        let last_activity = AtomicU64::new(0);
        // 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, timed_out)) = tokio::join!(
            pump_stdout(stdout, sink, &sentinels, base, &last_activity),
            pump_stderr(stderr, sink, &sentinels),
            wait_with_idle(&mut child, base, &last_activity, config.idle_timeout),
        );
        let status = status.map_err(Error::Wait)?;

        // A recognized sentinel takes precedence over everything (most specific
        // reason). Then an idle-watchdog kill; then a plain non-zero exit.
        if let Some(sentinel) = sentinels.soft_error() {
            return Err(sentinel);
        }
        if timed_out {
            return Err(Error::IdleTimeout);
        }
        if !status.success() {
            return Err(Error::Exit {
                status,
                stderr_tail,
            });
        }
        Ok(())
    }
}

/// Wait for the child to exit, enforcing the optional idle watchdog. With no
/// `idle_timeout` this is a plain `child.wait()`. Otherwise it re-checks on a
/// fixed probe cadence: if no stdout line arrived within `idle_timeout` (per
/// `last_activity`, bumped by [`pump_stdout`]), it kills the child and reaps
/// it. Returns the exit status and whether the watchdog fired.
async fn wait_with_idle(
    child: &mut Child,
    base: Instant,
    last_activity: &AtomicU64,
    idle_timeout: Option<Duration>,
) -> (std::io::Result<ExitStatus>, bool) {
    let Some(window) = idle_timeout else {
        return (child.wait().await, false);
    };
    // `child.wait()` is cancel-safe, so dropping it on a probe timeout doesn't
    // lose the exit.
    loop {
        match tokio::time::timeout(IDLE_PROBE, child.wait()).await {
            Ok(status) => return (status, false),
            Err(_probe_expired) => {
                let last = Duration::from_secs(last_activity.load(Ordering::Relaxed));
                if base.elapsed().saturating_sub(last) >= window {
                    let _ = child.kill().await;
                    return (child.wait().await, true);
                }
            }
        }
    }
}

/// 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,
    base: Instant,
    last_activity: &AtomicU64,
) {
    let mut lines = BufReader::new(stdout).lines();
    while let Ok(Some(line)) = lines.next_line().await {
        // Poke the idle watchdog: any stdout line resets the silence timer.
        // Seconds granularity is plenty — the probe cadence is coarser still.
        last_activity.store(base.elapsed().as_secs(), Ordering::Relaxed);
        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")
}