omne_cli/

claude_proc.rs

//! `claude -p --output-format stream-json` subprocess client.
//!
//! Unit 9 substrate: drive a `claude` subprocess, parse its stream-json
//! output, reconstruct assistant text into lines, and capture the
//! concatenated text to `.omne/var/runs/<run_id>/nodes/<node_id>.out`.
//! The sentinel scanner (Unit 10) and node executor (Unit 11) consume
//! the reconstructed-line iterator exposed here.
//!
//! The module is intentionally split into two layers:
//!
//! - [`StreamParser`] — a generic `BufRead` → `AssistantLine` iterator.
//!   Unit tests and downstream reuse (future replay/inspection tooling)
//!   drive this directly with canned fixtures; no subprocess required.
//! - [`ClaudeProcess`] — the subprocess wrapper that owns a `Child`,
//!   drains its stderr on a background thread (so the pipe buffer can
//!   never deadlock a long-running `claude -p` on a quiet run), and
//!   delegates iteration to a `StreamParser<BufReader<ChildStdout>>`.
//!
//! Reconstruction contract (plan R8, spike 2026-04-15):
//!
//! - Each stream-json line is one envelope. We only care about
//!   `{"type":"assistant", "message": { "id": ..., "content": [...] }}`
//!   entries. All other top-level types are skipped — including
//!   `system.hook_response`, which SessionStart hooks emit under `-p`
//!   (spike-confirmed noise) and which must never leak into the
//!   sentinel scanner.
//! - `claude -p` may split a single logical message across multiple
//!   envelopes that share `message.id`. Text blocks are therefore
//!   accumulated into a per-`id` buffer and flushed when a new id is
//!   observed or the stream ends. Flushing splits the buffer on `\n`,
//!   trims each line, drops empties, and yields them in order.
//! - All received text blocks are written verbatim to the capture file
//!   as they arrive. No per-line rewriting: the capture is the raw
//!   concatenation of every `text` block from every assistant message,
//!   which matches the plan's "concatenated assistant text" contract.

#![allow(dead_code)]

use std::collections::{HashMap, VecDeque};
use std::ffi::OsStr;
use std::fs::{File, OpenOptions};
use std::io::{BufRead, BufReader, Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, ChildStdout, Command, ExitStatus, Stdio};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::Duration;

use thiserror::Error;
use wait_timeout::ChildExt;

/// Default binary name probed on `PATH`.
pub const DEFAULT_BIN: &str = "claude";

/// Wall-clock budget for [`preflight`] (i.e. `claude --version`). Long
/// enough for a slow Windows process spawn; short enough to fail fast
/// when the host binary is wedged. Runtime invocations (`stream`) have
/// no timeout at this layer — Unit 11's executor owns node-level
/// deadlines.
pub const PREFLIGHT_TIMEOUT: Duration = Duration::from_secs(10);

/// Errors surfaced by the `claude -p` subprocess client.
#[derive(Debug, Error)]
pub enum Error {
    /// `claude` is not on `PATH`, or the explicit override path does
    /// not point at an executable. Distinct from [`Error::Spawn`] so
    /// `omne validate` and the executor can emit an install hint.
    #[error(
        "claude binary not found on PATH\n\
         hint: install Claude Code and ensure `claude` is on PATH"
    )]
    HostMissing,

    /// `std::process::Command::spawn` failed for a reason other than
    /// the binary being missing (permission denied, I/O error on the
    /// pipe allocation, etc).
    #[error("failed to launch claude: {source}")]
    Spawn {
        #[source]
        source: std::io::Error,
    },

    /// The preflight subprocess did not exit within
    /// [`PREFLIGHT_TIMEOUT`]. The child is killed before this error is
    /// returned so no zombie leaks.
    #[error("claude did not exit within {elapsed:?}")]
    Timeout { elapsed: Duration },

    /// The child exited with a non-zero status. The stderr tail is
    /// carried verbatim so callers can surface it in their own error
    /// reporting.
    #[error("claude exited with {status}\nstderr: {stderr}")]
    ExitedNonZero { status: ExitStatus, stderr: String },

    /// I/O error reading the child's stdout or awaiting its exit.
    #[error("I/O error on claude stream: {source}")]
    Io {
        #[source]
        source: std::io::Error,
    },

    /// I/O error writing the per-node capture file.
    #[error("capture I/O on {path}: {source}")]
    Capture {
        path: PathBuf,
        #[source]
        source: std::io::Error,
    },
}

/// Session lifecycle for one `claude -p` invocation.
///
/// Locked by Unit 0 spike: `--session-id` on the first iteration of a
/// `fresh_context: false` loop, `--resume` on every subsequent
/// iteration. The id is a UUID (not a ULID) — `claude --session-id`
/// rejects non-UUID strings. Unit 11's loop controller allocates and
/// passes this value.
#[derive(Debug, Clone, Eq, PartialEq)]
pub enum Session {
    /// First invocation: set deterministic session id via
    /// `--session-id <uuid>`.
    New(String),
    /// Subsequent invocations: resume via `--resume <uuid>`.
    Resume(String),
}

/// Arguments used to build a `claude -p` invocation.
#[derive(Debug, Clone)]
pub struct SpawnOpts {
    /// Positional prompt argument passed last to `claude -p`.
    pub prompt: String,
    /// Working directory for the child process. Unit 11 sets this to
    /// the per-run worktree.
    pub cwd: PathBuf,
    /// Optional `--model <id>`. Omitted when `None`.
    pub model: Option<String>,
    /// Comma-joined into `--allowed-tools`. Empty when the node places
    /// no restriction.
    pub allowed_tools: Vec<String>,
    /// Loop session flag (Unit 11); `None` for one-shot nodes.
    pub session: Option<Session>,
    /// Escape hatch for per-distro flags (e.g. `--bare`,
    /// `--dangerously-skip-permissions`). Appended verbatim before the
    /// prompt.
    pub extra_args: Vec<String>,
    /// Override binary path. Defaults to [`DEFAULT_BIN`] on `PATH`.
    pub bin: Option<PathBuf>,
}

impl SpawnOpts {
    /// Minimal constructor for the common case: prompt + cwd, no
    /// model / tools / session.
    pub fn new(prompt: impl Into<String>, cwd: impl Into<PathBuf>) -> Self {
        Self {
            prompt: prompt.into(),
            cwd: cwd.into(),
            model: None,
            allowed_tools: Vec::new(),
            session: None,
            extra_args: Vec::new(),
            bin: None,
        }
    }
}

/// How [`StreamParser`] opens the capture file.
///
/// Non-loop and first-iteration loop nodes want `Truncate` so a replayed
/// run does not concatenate stale text from a prior attempt. Subsequent
/// loop iterations want `Append` so all iterations accumulate in the
/// same `nodes/<id>.out` (with iteration markers written separately by
/// the executor). Default: `Truncate`.
#[derive(Debug, Clone, Copy, Eq, PartialEq, Default)]
pub enum CaptureMode {
    #[default]
    Truncate,
    Append,
}

/// One reconstructed line of assistant text.
///
/// `message_id` is the `message.id` field carried by every `assistant`
/// stream-json envelope. Unit 10's sentinel scanner does not need it
/// today (matches are on `text` alone), but Unit 11's telemetry and
/// future replay tooling benefit from knowing which turn produced each
/// line, so it is surfaced rather than hidden.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct AssistantLine {
    pub text: String,
    pub message_id: String,
}

/// Run `claude --version` and return the reported version string.
///
/// `bin = None` uses the `PATH`-resolved `claude` binary. An explicit
/// path is useful for tests and for environments where multiple
/// installations coexist.
pub fn preflight(bin: Option<&Path>) -> Result<String, Error> {
    let bin_os: &OsStr = bin
        .map(|b| b.as_os_str())
        .unwrap_or_else(|| OsStr::new(DEFAULT_BIN));

    // Pre-probe for HostMissing so the error surfaces cleanly instead
    // of going through the ambiguous `Command::spawn` error path.
    if bin.is_none() && which::which(DEFAULT_BIN).is_err() {
        return Err(Error::HostMissing);
    }

    let mut child = Command::new(bin_os)
        .arg("--version")
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .map_err(map_spawn_error)?;

    let status = match child
        .wait_timeout(PREFLIGHT_TIMEOUT)
        .map_err(|source| Error::Io { source })?
    {
        Some(status) => status,
        None => {
            let _ = child.kill();
            let _ = child.wait();
            return Err(Error::Timeout {
                elapsed: PREFLIGHT_TIMEOUT,
            });
        }
    };

    let mut stdout = String::new();
    if let Some(mut s) = child.stdout.take() {
        s.read_to_string(&mut stdout)
            .map_err(|source| Error::Io { source })?;
    }
    let mut stderr = String::new();
    if let Some(mut s) = child.stderr.take() {
        s.read_to_string(&mut stderr)
            .map_err(|source| Error::Io { source })?;
    }

    if !status.success() {
        return Err(Error::ExitedNonZero { status, stderr });
    }
    Ok(stdout.trim().to_string())
}

/// Build the `Command` that [`spawn`] would launch. Exposed so tests
/// and Unit 11's executor can inspect / adjust the argv without
/// actually spawning.
pub fn build_command(opts: &SpawnOpts) -> Command {
    let bin_os: &OsStr = opts
        .bin
        .as_deref()
        .map(|p| p.as_os_str())
        .unwrap_or_else(|| OsStr::new(DEFAULT_BIN));

    let mut cmd = Command::new(bin_os);
    cmd.current_dir(&opts.cwd);
    cmd.args(["-p", "--output-format", "stream-json", "--verbose"]);

    if let Some(model) = &opts.model {
        cmd.args(["--model", model]);
    }
    if !opts.allowed_tools.is_empty() {
        cmd.arg("--allowed-tools").arg(opts.allowed_tools.join(","));
    }
    match &opts.session {
        Some(Session::New(uuid)) => {
            cmd.args(["--session-id", uuid]);
        }
        Some(Session::Resume(uuid)) => {
            cmd.args(["--resume", uuid]);
        }
        None => {}
    }
    for extra in &opts.extra_args {
        cmd.arg(extra);
    }

    // Positional prompt goes last so it is unambiguous even if an
    // `extra_args` entry starts with `-`.
    cmd.arg(&opts.prompt);

    cmd.stdin(Stdio::null());
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());
    cmd
}

/// Spawn `claude -p` per [`SpawnOpts`] and return the `Child`.
///
/// Stdin is redirected from `/dev/null` so the subprocess never waits
/// for interactive input; stdout and stderr are piped so the caller
/// can feed them to [`ClaudeProcess`].
pub fn spawn(opts: &SpawnOpts) -> Result<Child, Error> {
    build_command(opts).spawn().map_err(map_spawn_error)
}

/// Wire a live `Child` into a [`ClaudeProcess`] that yields
/// [`AssistantLine`] values and writes the reconstructed text to
/// `capture_path` (truncating any prior content).
pub fn stream(child: Child, capture_path: &Path) -> Result<ClaudeProcess, Error> {
    ClaudeProcess::from_child(child, capture_path)
}

/// Translate a `Command::spawn` failure into [`Error::HostMissing`]
/// when the OS reports `NotFound`, otherwise [`Error::Spawn`]. Shared
/// between [`preflight`] and [`spawn`].
fn map_spawn_error(source: std::io::Error) -> Error {
    if source.kind() == std::io::ErrorKind::NotFound {
        Error::HostMissing
    } else {
        Error::Spawn { source }
    }
}

/// Parses stream-json envelopes from any `BufRead` and yields
/// [`AssistantLine`] values, while capturing all reconstructed text
/// to a backing file.
///
/// The parser owns the capture file so iteration and capture advance
/// in lockstep; callers who just want the lines can drop the iterator
/// and the capture file closes on its own. Flushing happens on every
/// message transition and on EOF.
pub struct StreamParser<R: BufRead> {
    reader: R,
    capture: File,
    capture_path: PathBuf,
    /// Per-message-id text buffer. Keyed by `message.id` because
    /// `claude -p` may emit multiple envelopes with the same id when
    /// partial-message streaming is active.
    buffers: HashMap<String, String>,
    /// Insertion order of ids observed, so flush-on-EOF yields lines
    /// in deterministic order across messages. We also push to this
    /// on new-id observation, and pop the id when it is flushed.
    order: VecDeque<String>,
    /// Id of the most recently seen `assistant` envelope. Used to
    /// detect id transitions (flush prior) without scanning `order`.
    current_id: Option<String>,
    /// Lines ready to be yielded. Populated at flush time.
    pending: VecDeque<AssistantLine>,
    finished: bool,
}

impl<R: BufRead> StreamParser<R> {
    /// Build a parser over `reader`, creating or truncating
    /// `capture_path` (and its parent directory) for the reconstructed
    /// text. Equivalent to [`StreamParser::with_mode`] with
    /// [`CaptureMode::Truncate`].
    pub fn new(reader: R, capture_path: &Path) -> Result<Self, Error> {
        Self::with_mode(reader, capture_path, CaptureMode::Truncate)
    }

    /// Build a parser over `reader` with explicit capture-open mode.
    ///
    /// [`CaptureMode::Truncate`] wipes prior content; [`CaptureMode::Append`]
    /// positions writes at end-of-file so loop iterations accumulate in
    /// one file (plan R11 "same `nodes/<id>.out` with iteration markers").
    pub fn with_mode(reader: R, capture_path: &Path, mode: CaptureMode) -> Result<Self, Error> {
        if let Some(parent) = capture_path.parent() {
            if !parent.as_os_str().is_empty() {
                std::fs::create_dir_all(parent).map_err(|source| Error::Capture {
                    path: parent.to_path_buf(),
                    source,
                })?;
            }
        }
        let mut open = OpenOptions::new();
        open.create(true).write(true);
        match mode {
            CaptureMode::Truncate => {
                open.truncate(true);
            }
            CaptureMode::Append => {
                open.append(true);
            }
        }
        let capture = open.open(capture_path).map_err(|source| Error::Capture {
            path: capture_path.to_path_buf(),
            source,
        })?;
        Ok(Self {
            reader,
            capture,
            capture_path: capture_path.to_path_buf(),
            buffers: HashMap::new(),
            order: VecDeque::new(),
            current_id: None,
            pending: VecDeque::new(),
            finished: false,
        })
    }

    /// Ingest one raw line. Parses as JSON; dispatches on top-level
    /// `type`. Returns `Ok(())` for skippable envelopes and for
    /// malformed lines (logged + dropped per plan R-deferred
    /// "Streamed JSON parsing resilience").
    fn ingest_line(&mut self, raw: &str) -> Result<(), Error> {
        let trimmed = raw.trim();
        if trimmed.is_empty() {
            return Ok(());
        }
        let env: serde_json::Value = match serde_json::from_str(trimmed) {
            Ok(v) => v,
            Err(err) => {
                eprintln!("claude_proc: skipping malformed stream-json line: {err}");
                return Ok(());
            }
        };
        let ty = env.get("type").and_then(|t| t.as_str()).unwrap_or("");
        if ty != "assistant" {
            // Silent skip for the envelopes the plan explicitly
            // excludes (system/user/result/rate_limit_event) and for
            // unknown future types. `system.hook_response` lands here
            // and is the load-bearing spike-confirmed noise source.
            return Ok(());
        }

        let Some(message) = env.get("message") else {
            return Ok(());
        };
        let id = message
            .get("id")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();

        // New id ≠ currently-open id → flush the prior message first
        // so the yield order is <prior lines>, then <this message's
        // lines> (on subsequent ingest or EOF).
        if let Some(prev) = self.current_id.as_ref() {
            if prev != &id {
                let prev = prev.clone();
                self.flush_message(&prev);
            }
        }
        if !self.buffers.contains_key(&id) {
            self.order.push_back(id.clone());
        }
        self.current_id = Some(id.clone());

        let empty = Vec::new();
        let content = message
            .get("content")
            .and_then(|c| c.as_array())
            .unwrap_or(&empty);
        let mut appended = String::new();
        for block in content {
            let block_type = block.get("type").and_then(|t| t.as_str()).unwrap_or("");
            if block_type != "text" {
                continue;
            }
            if let Some(text) = block.get("text").and_then(|v| v.as_str()) {
                appended.push_str(text);
            }
        }
        if !appended.is_empty() {
            self.buffers.entry(id).or_default().push_str(&appended);
            self.capture
                .write_all(appended.as_bytes())
                .map_err(|source| Error::Capture {
                    path: self.capture_path.clone(),
                    source,
                })?;
        }
        Ok(())
    }

    /// Split `buffers[id]` on `\n`, trim and drop empties, push the
    /// survivors onto `pending` in order. Also removes `id` from
    /// `order` so the iterator does not re-flush it.
    fn flush_message(&mut self, id: &str) {
        let Some(buf) = self.buffers.remove(id) else {
            return;
        };
        // Keep `order` in sync so EOF-flush does not re-yield.
        if let Some(pos) = self.order.iter().position(|x| x == id) {
            self.order.remove(pos);
        }
        for piece in buf.split('\n') {
            let trimmed = piece.trim();
            if trimmed.is_empty() {
                continue;
            }
            self.pending.push_back(AssistantLine {
                text: trimmed.to_string(),
                message_id: id.to_string(),
            });
        }
    }

    /// Flush every outstanding buffer. Called when the stream reaches
    /// EOF so trailing partial-line text from the last message still
    /// reaches the consumer.
    fn flush_all(&mut self) {
        self.current_id = None;
        while let Some(id) = self.order.pop_front() {
            // flush_message removes from `order`, but since we've
            // already drained it here, re-inserting the id as the
            // argument and letting flush_message work on `buffers`
            // only is fine.
            let buf = match self.buffers.remove(&id) {
                Some(b) => b,
                None => continue,
            };
            for piece in buf.split('\n') {
                let trimmed = piece.trim();
                if trimmed.is_empty() {
                    continue;
                }
                self.pending.push_back(AssistantLine {
                    text: trimmed.to_string(),
                    message_id: id.clone(),
                });
            }
        }
    }

    /// Flush the capture file to disk. Called by
    /// [`ClaudeProcess::finish`] before reporting exit status so a
    /// crashing caller cannot leave buffered capture data unflushed.
    fn flush_capture(&mut self) -> Result<(), Error> {
        self.capture.flush().map_err(|source| Error::Capture {
            path: self.capture_path.clone(),
            source,
        })
    }
}

impl<R: BufRead> Iterator for StreamParser<R> {
    type Item = Result<AssistantLine, Error>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            if let Some(line) = self.pending.pop_front() {
                return Some(Ok(line));
            }
            if self.finished {
                return None;
            }
            let mut raw = String::new();
            match self.reader.read_line(&mut raw) {
                Ok(0) => {
                    self.finished = true;
                    self.flush_all();
                    if let Err(e) = self.flush_capture() {
                        return Some(Err(e));
                    }
                }
                Ok(_) => {
                    if let Err(e) = self.ingest_line(&raw) {
                        return Some(Err(e));
                    }
                }
                Err(source) => {
                    self.finished = true;
                    return Some(Err(Error::Io { source }));
                }
            }
        }
    }
}

/// Subprocess wrapper around a live `claude -p` [`Child`] that yields
/// [`AssistantLine`] values and, on [`ClaudeProcess::finish`], returns
/// the child's exit status and captured stderr.
///
/// Stderr is drained on a dedicated thread as soon as the process is
/// wrapped. Without this, a chatty stderr (warnings, deprecation
/// notices) would fill its pipe buffer and block the child until
/// somebody reads it — which no one does during a sentinel-driven
/// loop iteration that only cares about stdout.
pub struct ClaudeProcess {
    parser: StreamParser<BufReader<ChildStdout>>,
    child: Arc<Mutex<Option<Child>>>,
    stderr_thread: Option<JoinHandle<Vec<u8>>>,
}

/// Cloneable handle that can kill a live [`ClaudeProcess`].
///
/// Returned by [`ClaudeProcess::killer`]. The executor's timeout
/// watchdog thread holds one while the main thread drains the parser;
/// if the deadline elapses before the stream ends, the watchdog calls
/// [`ChildKiller::kill`] and the main thread's next `read_line` returns
/// `Ok(0)` once the child's stdout pipe closes.
#[derive(Clone)]
pub struct ChildKiller {
    child: Arc<Mutex<Option<Child>>>,
}

impl ChildKiller {
    /// Send SIGKILL (or its platform equivalent) to the child if it is
    /// still alive. A no-op once [`ClaudeProcess::finish`] has taken
    /// the child out.
    pub fn kill(&self) -> std::io::Result<()> {
        let mut guard = match self.child.lock() {
            Ok(g) => g,
            Err(poisoned) => poisoned.into_inner(),
        };
        if let Some(c) = guard.as_mut() {
            return c.kill();
        }
        Ok(())
    }
}

impl ClaudeProcess {
    /// Wrap `child` and start draining its stderr.
    ///
    /// The child's `stdout` and `stderr` handles must both be piped
    /// (i.e. the child was spawned via [`spawn`] or via
    /// [`build_command`]). A missing stdout is fatal because the
    /// parser has nothing to read. Equivalent to
    /// [`ClaudeProcess::from_child_with_mode`] with
    /// [`CaptureMode::Truncate`].
    pub fn from_child(child: Child, capture_path: &Path) -> Result<Self, Error> {
        Self::from_child_with_mode(child, capture_path, CaptureMode::Truncate)
    }

    /// Same as [`ClaudeProcess::from_child`] but with explicit capture
    /// open mode. Loop iterations ≥ 2 pass [`CaptureMode::Append`] so
    /// prior iterations' text is preserved.
    pub fn from_child_with_mode(
        mut child: Child,
        capture_path: &Path,
        mode: CaptureMode,
    ) -> Result<Self, Error> {
        let stdout = child.stdout.take().ok_or_else(|| Error::Io {
            source: std::io::Error::new(
                std::io::ErrorKind::BrokenPipe,
                "claude child spawned without piped stdout",
            ),
        })?;
        let stderr_thread = child.stderr.take().map(|mut s| {
            thread::spawn(move || {
                let mut buf = Vec::new();
                let _ = s.read_to_end(&mut buf);
                buf
            })
        });
        let parser = StreamParser::with_mode(BufReader::new(stdout), capture_path, mode)?;
        Ok(Self {
            parser,
            child: Arc::new(Mutex::new(Some(child))),
            stderr_thread,
        })
    }

    /// Clone a kill-handle for a timeout watchdog. See [`ChildKiller`].
    pub fn killer(&self) -> ChildKiller {
        ChildKiller {
            child: Arc::clone(&self.child),
        }
    }

    /// Drain any remaining lines from the stream, wait for the child
    /// to exit, and return `(status, stderr)`.
    ///
    /// Does not call [`Error::ExitedNonZero`] itself — the caller
    /// decides whether a non-zero exit is actionable. Unit 11's
    /// executor maps this to [`Error::ExitedNonZero`] when the node
    /// contract considers a non-zero status a failure.
    pub fn finish(mut self) -> Result<(ExitStatus, String), Error> {
        // Drain pending iterator output so capture contains everything
        // the child produced before we close its stdout.
        for item in self.parser.by_ref() {
            item?;
        }
        self.parser.flush_capture()?;

        let child_opt = {
            let mut guard = match self.child.lock() {
                Ok(g) => g,
                Err(poisoned) => poisoned.into_inner(),
            };
            guard.take()
        };
        let mut child = child_opt.ok_or_else(|| Error::Io {
            source: std::io::Error::new(
                std::io::ErrorKind::BrokenPipe,
                "ClaudeProcess::finish called with no live child — \
                 finish already called or kill-drop left the Option None",
            ),
        })?;
        let status = child.wait().map_err(|source| Error::Io { source })?;
        let stderr_bytes = self
            .stderr_thread
            .take()
            .map(|h| h.join().unwrap_or_default())
            .unwrap_or_default();
        let stderr = String::from_utf8_lossy(&stderr_bytes).into_owned();
        Ok((status, stderr))
    }
}

impl Iterator for ClaudeProcess {
    type Item = Result<AssistantLine, Error>;

    fn next(&mut self) -> Option<Self::Item> {
        self.parser.next()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Cursor;
    use tempfile::TempDir;

    // ---- argv construction ---------------------------------------------

    fn collect_args(cmd: &Command) -> Vec<String> {
        cmd.get_args()
            .map(|a| a.to_string_lossy().into_owned())
            .collect()
    }

    #[test]
    fn build_command_sets_required_streaming_flags() {
        let opts = SpawnOpts::new("go", "/tmp");
        let cmd = build_command(&opts);
        let args = collect_args(&cmd);
        assert!(args
            .windows(4)
            .any(|w| w == ["-p", "--output-format", "stream-json", "--verbose"]));
        assert_eq!(args.last().unwrap(), "go");
    }

    #[test]
    fn build_command_omits_unset_model_and_tools() {
        let args = collect_args(&build_command(&SpawnOpts::new("x", "/tmp")));
        assert!(!args.iter().any(|a| a == "--model"));
        assert!(!args.iter().any(|a| a == "--allowed-tools"));
        assert!(!args.iter().any(|a| a == "--session-id"));
        assert!(!args.iter().any(|a| a == "--resume"));
    }

    #[test]
    fn build_command_joins_allowed_tools_with_commas() {
        let mut opts = SpawnOpts::new("x", "/tmp");
        opts.allowed_tools = vec!["Read".into(), "Bash".into()];
        let args = collect_args(&build_command(&opts));
        let tools_idx = args.iter().position(|a| a == "--allowed-tools").unwrap();
        assert_eq!(args[tools_idx + 1], "Read,Bash");
    }

    #[test]
    fn build_command_uses_session_id_for_new_session() {
        let mut opts = SpawnOpts::new("x", "/tmp");
        opts.session = Some(Session::New("abc-123".into()));
        let args = collect_args(&build_command(&opts));
        let idx = args.iter().position(|a| a == "--session-id").unwrap();
        assert_eq!(args[idx + 1], "abc-123");
        assert!(!args.iter().any(|a| a == "--resume"));
    }

    #[test]
    fn build_command_uses_resume_for_continued_session() {
        let mut opts = SpawnOpts::new("x", "/tmp");
        opts.session = Some(Session::Resume("abc-123".into()));
        let args = collect_args(&build_command(&opts));
        let idx = args.iter().position(|a| a == "--resume").unwrap();
        assert_eq!(args[idx + 1], "abc-123");
        assert!(!args.iter().any(|a| a == "--session-id"));
    }

    #[test]
    fn build_command_puts_prompt_last_even_with_extra_args() {
        let mut opts = SpawnOpts::new("go do it", "/tmp");
        opts.extra_args = vec!["--dangerously-skip-permissions".into()];
        let args = collect_args(&build_command(&opts));
        assert_eq!(args.last().unwrap(), "go do it");
        let danger_idx = args
            .iter()
            .position(|a| a == "--dangerously-skip-permissions")
            .unwrap();
        let prompt_idx = args.iter().position(|a| a == "go do it").unwrap();
        assert!(danger_idx < prompt_idx, "extra args precede the prompt");
    }

    // ---- parser fixtures -----------------------------------------------

    fn asst(id: &str, text: &str) -> String {
        serde_json::json!({
            "type": "assistant",
            "message": {
                "id": id,
                "content": [ { "type": "text", "text": text } ]
            }
        })
        .to_string()
    }

    fn asst_multi(id: &str, texts: &[&str]) -> String {
        let content: Vec<_> = texts
            .iter()
            .map(|t| serde_json::json!({ "type": "text", "text": t }))
            .collect();
        serde_json::json!({
            "type": "assistant",
            "message": { "id": id, "content": content }
        })
        .to_string()
    }

    fn run_parser(lines: &[String], capture: &Path) -> Vec<AssistantLine> {
        let body = lines.join("\n") + "\n";
        let parser = StreamParser::new(Cursor::new(body.into_bytes()), capture).expect("open");
        parser.map(|r| r.expect("parse ok")).collect()
    }

    #[test]
    fn parses_three_messages_totaling_five_lines() {
        let tmp = TempDir::new().unwrap();
        let cap = tmp.path().join("node.out");
        let lines = vec![
            asst("m1", "first\nsecond"),
            asst("m2", "third"),
            asst("m3", "fourth\nfifth"),
        ];
        let yielded = run_parser(&lines, &cap);
        assert_eq!(
            yielded.iter().map(|l| &l.text).collect::<Vec<_>>(),
            vec!["first", "second", "third", "fourth", "fifth"]
        );
        assert_eq!(yielded[0].message_id, "m1");
        assert_eq!(yielded[4].message_id, "m3");
    }

    #[test]
    fn joins_multiple_text_blocks_within_one_message() {
        let tmp = TempDir::new().unwrap();
        let cap = tmp.path().join("node.out");
        // Four "deltas" as four text blocks inside the same message;
        // the final text crosses an internal line boundary after join.
        let line = asst_multi(
            "m1",
            &["first part ", "of line\nsecond ", "line split ", "too"],
        );
        let yielded = run_parser(&[line], &cap);
        assert_eq!(
            yielded.iter().map(|l| &l.text).collect::<Vec<_>>(),
            vec!["first part of line", "second line split too"]
        );
    }

    #[test]
    fn joins_partial_message_across_envelopes_sharing_id() {
        // Same `message.id` across two envelopes — parser must
        // accumulate, not yield twice.
        let tmp = TempDir::new().unwrap();
        let cap = tmp.path().join("node.out");
        let lines = vec![asst("msame", "hello "), asst("msame", "world\ngoodbye")];
        let yielded = run_parser(&lines, &cap);
        assert_eq!(
            yielded.iter().map(|l| &l.text).collect::<Vec<_>>(),
            vec!["hello world", "goodbye"]
        );
    }

    #[test]
    fn skips_malformed_stream_json_line_and_continues() {
        let tmp = TempDir::new().unwrap();
        let cap = tmp.path().join("node.out");
        let lines = vec![
            asst("m1", "ok"),
            "{not valid json".to_string(),
            asst("m2", "still ok"),
        ];
        let yielded = run_parser(&lines, &cap);
        assert_eq!(
            yielded.iter().map(|l| &l.text).collect::<Vec<_>>(),
            vec!["ok", "still ok"]
        );
    }

    #[test]
    fn skips_unknown_top_level_type() {
        let tmp = TempDir::new().unwrap();
        let cap = tmp.path().join("node.out");
        let lines = vec![
            serde_json::json!({ "type": "thinking", "content": "pondering" }).to_string(),
            asst("m1", "visible"),
        ];
        let yielded = run_parser(&lines, &cap);
        assert_eq!(
            yielded.iter().map(|l| &l.text).collect::<Vec<_>>(),
            vec!["visible"]
        );
    }

    #[test]
    fn skips_system_hook_response_envelope() {
        // Spike-confirmed noise: SessionStart hooks fire under `-p`.
        let tmp = TempDir::new().unwrap();
        let cap = tmp.path().join("node.out");
        let lines = vec![
            serde_json::json!({
                "type": "system",
                "subtype": "hook_response",
                "hook_name": "SessionStart",
                "output": "CAVEMAN MODE ACTIVE"
            })
            .to_string(),
            asst("m1", "real content"),
        ];
        let yielded = run_parser(&lines, &cap);
        assert_eq!(
            yielded.iter().map(|l| &l.text).collect::<Vec<_>>(),
            vec!["real content"]
        );
    }

    #[test]
    fn skips_non_text_blocks_inside_assistant_message() {
        let tmp = TempDir::new().unwrap();
        let cap = tmp.path().join("node.out");
        // tool_use and thinking blocks must not leak into scanner
        // lines — only text blocks contribute.
        let line = serde_json::json!({
            "type": "assistant",
            "message": {
                "id": "m1",
                "content": [
                    { "type": "thinking", "thinking": "...internal..." },
                    { "type": "text", "text": "visible line" },
                    { "type": "tool_use", "name": "Read", "input": {} }
                ]
            }
        })
        .to_string();
        let yielded = run_parser(&[line], &cap);
        assert_eq!(
            yielded.iter().map(|l| &l.text).collect::<Vec<_>>(),
            vec!["visible line"]
        );
    }

    #[test]
    fn trims_whitespace_around_reconstructed_lines() {
        let tmp = TempDir::new().unwrap();
        let cap = tmp.path().join("node.out");
        let line = asst("m1", "   BLOCKED   \n  trailing\n");
        let yielded = run_parser(&[line], &cap);
        assert_eq!(
            yielded.iter().map(|l| &l.text).collect::<Vec<_>>(),
            vec!["BLOCKED", "trailing"]
        );
    }

    #[test]
    fn capture_file_mirrors_concatenated_assistant_text() {
        let tmp = TempDir::new().unwrap();
        let cap = tmp.path().join("deep").join("node.out");
        let lines = vec![
            asst("m1", "alpha\n"),
            asst_multi("m2", ["beta ", "gamma"].as_ref()),
        ];
        let _ = run_parser(&lines, &cap);
        let mut written = String::new();
        File::open(&cap)
            .unwrap()
            .read_to_string(&mut written)
            .unwrap();
        assert_eq!(written, "alpha\nbeta gamma");
    }

    #[test]
    fn empty_stream_yields_no_lines_and_creates_empty_capture() {
        let tmp = TempDir::new().unwrap();
        let cap = tmp.path().join("node.out");
        let parser = StreamParser::new(Cursor::new(Vec::<u8>::new()), &cap).unwrap();
        assert_eq!(parser.count(), 0);
        assert!(cap.exists());
        assert_eq!(std::fs::metadata(&cap).unwrap().len(), 0);
    }

    #[test]
    fn preflight_missing_binary_is_host_missing() {
        // Non-existent explicit path: NotFound → HostMissing.
        let err = preflight(Some(Path::new("/definitely/not/a/binary/omne-xyz"))).unwrap_err();
        assert!(
            matches!(err, Error::HostMissing),
            "expected HostMissing, got {err:?}"
        );
    }
}