diff --git a/Cargo.lock b/Cargo.lock index 744a340f..29eb0c04 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1333,6 +1333,7 @@ dependencies = [ "axum", "clap", "futures-util", + "hive-claude", "hive-sh4re", "reqwest", "rmcp", @@ -1392,6 +1393,15 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "hive-claude" +version = "0.1.0" +dependencies = [ + "serde_json", + "thiserror 2.0.18", + "tokio", +] + [[package]] name = "hive-forge" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 247e3637..128ba120 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "hive-ag3nt", "hive-bash-mcp", "hive-c0re", + "hive-claude", "hive-forge", "hive-matrix-mcp", "hive-priv", @@ -35,6 +36,8 @@ chrono = { version = "0.4", default-features = false, features = [ clap = { version = "4", features = ["derive"] } clap_complete = "4" hive-sh4re = { path = "hive-sh4re" } +hive-claude = { path = "hive-claude" } +thiserror = "2" tower-http = { version = "0.6", features = ["fs"] } rmcp = { version = "1.7", default-features = false, features = [ "server", diff --git a/hive-ag3nt/Cargo.toml b/hive-ag3nt/Cargo.toml index 655b5ab3..53f566e5 100644 --- a/hive-ag3nt/Cargo.toml +++ b/hive-ag3nt/Cargo.toml @@ -12,6 +12,7 @@ axum.workspace = true reqwest.workspace = true futures-util = "0.3" clap.workspace = true +hive-claude.workspace = true hive-sh4re.workspace = true rmcp.workspace = true rusqlite.workspace = true diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index d053a57b..3e428d38 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -1,20 +1,19 @@ -//! Per-turn claude invocation. The spawn shape, arg-vector, stdin plumbing, -//! and stream-json pumping are shared across all roles (there is only one -//! role: agent). +//! Per-turn claude policy layer. The generic subprocess mechanics — spawning +//! `claude --print`, streaming + classifying stream-json, session +//! lookup/archive — live in the `hive-claude` crate. This module owns the +//! hyperhive-specific policy on top: building the per-turn config from the +//! bus, bridging the output stream onto the event bus (`BusSink`), and the +//! compaction / auto-reset / retry state machine (`drive_turn`). -use std::collections::VecDeque; -use std::io::BufRead as _; use std::path::{Path, PathBuf}; -use std::process::Stdio; -use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; -use anyhow::{Result, bail}; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::process::Command; +use anyhow::Result; +use hive_claude::{Claude, Config, Session, Sink}; +use serde_json::Value; -use crate::events::{Bus, LiveEvent}; +use crate::events::{Bus, LiveEvent, TokenUsage}; use crate::login::LoginState; use crate::mcp; @@ -27,55 +26,12 @@ use crate::mcp; // notes persistence under `/state`). Unknown keys are silently ignored by // claude-code; if a key gets renamed we'll spot it because the // corresponding behavior will start firing mid-turn again. - -/// Regex-ish marker claude-code emits when context overflows. Same string -/// bitburner-agent watches for. Empirically reliable across claude-code -/// versions; if it ever changes, compaction won't fire and we'll see a -/// claude exit with a useful error in the live view. -const PROMPT_TOO_LONG_MARKER: &str = "Prompt is too long"; - -/// Substrings that indicate the Anthropic API is refusing the request due -/// to a rate limit, per-account usage cap, or exhausted credit balance. -/// Matched against both stdout and stderr; any hit returns -/// `TurnOutcome::RateLimited` so the serve loop can park + retry instead -/// of propagating a hard failure that looks identical to a crash. -const RATE_LIMIT_MARKERS: &[&str] = &[ - "rate_limit_error", - "overloaded_error", - "Credit balance is too low", - "Usage limit reached", - "Request rate limit exceeded", -]; - -/// Substrings that indicate the Anthropic API rejected the request as -/// unauthenticated — the OAuth session in `$HOME/.claude/` has expired -/// or been revoked. Surfaced as `TurnOutcome::AuthFailed`, which the -/// harness uses to flip the container into `needs_login_idle` so the -/// dashboard's re-auth flow takes over. Matched against both stdout -/// JSON `error` events and stderr; the markers come from claude-code's -/// `api_retry` events (`{"error":"authentication_failed", -/// "error_status":401,...}`) and the human-readable -/// "Failed to authenticate. API Error: 401" line claude prints on giveup. -/// See [`docs/turn-loop.md::The loop`](../../docs/turn-loop.md) for the -/// re-auth resumption path. -const AUTH_FAIL_MARKERS: &[&str] = &[ - "\"error\":\"authentication_failed\"", - "\"error_status\":401", - "Failed to authenticate. API Error: 401", -]; - -/// Substrings claude-code emits when a `--resume ` target can't be -/// resolved: either no session carries our constant title yet (first turn, -/// post-archive, post-purge) or a stale id was handed in. On a hit the -/// harness re-runs the SAME prompt once with `--name <title>` to mint a -/// fresh session titled `<title>`, so the agent self-heals instead of -/// failing `--resume` forever. Empirically matched against claude 2.1.197: -/// resume-by-title miss → "…does not match any session title"; bare stale -/// id → "No conversation found with session ID". -const TITLE_NOT_FOUND_MARKERS: &[&str] = &[ - "does not match any session title", - "No conversation found with session ID", -]; +// +// The subprocess mechanics — spawning `claude --print`, streaming + +// classifying stream-json, session lookup/archive — live in the generic +// `hive-claude` crate. This module is the hyperhive *policy* layer on top: +// it builds the per-turn [`Config`] from the bus, forwards the stream to the +// event bus via [`BusSink`], and owns compaction / auto-reset / retry. /// Fixed, harness-owned claude session title. Every turn / compact / /// checkpoint resumes THIS title (`--resume <title>`); the create path @@ -603,22 +559,29 @@ fn session_refreshed(prev: DirSnapshot, now: DirSnapshot) -> bool { } /// Run one turn against the constant-title session, resuming it or creating -/// it on first use. Delegates to [`run_claude_resume_or_create`]; the session -/// is pinned by a fixed `--name`/`--resume <title>` (NOT bare `--continue`, -/// which resumes the *latest* session in this cwd and lets a `choom` session -/// hijack the live harness context — a constant title is immune since choom -/// won't carry it). claude's in-session auto-compact is disabled via the -/// managed settings at `/etc/claude-code/managed-settings.json` so it doesn't -/// stall mid-turn — hyperhive owns compaction. +/// it on first use (`hive_claude::run_resume_or_create`). The session is +/// pinned by a fixed `--resume`/`--name <title>` (NOT bare `--continue`, which +/// resumes the *latest* session in this cwd and lets a `choom` session hijack +/// the live harness context — a constant title is immune since choom won't +/// carry it). claude's in-session auto-compact is disabled via the managed +/// settings at `/etc/claude-code/managed-settings.json` so it doesn't stall +/// mid-turn — hyperhive owns compaction. pub async fn run_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome { - match run_claude_resume_or_create(prompt, files, bus).await { - Ok(ClaudeResult::PromptTooLong) => TurnOutcome::PromptTooLong, - Ok(ClaudeResult::RateLimited) => TurnOutcome::RateLimited, - Ok(ClaudeResult::AuthFailed) => TurnOutcome::AuthFailed, - // `Ok`, and a residual `TitleNotFound` (create path also missed — - // shouldn't happen) both settle as a normal completed turn. - Ok(ClaudeResult::Ok | ClaudeResult::TitleNotFound) => TurnOutcome::Ok, - Err(e) => TurnOutcome::Failed(e), + let config = claude_config(bus, files); + let sink = BusSink::new(bus); + match Claude::run_resume_or_create(&config, &session_title(), prompt, &sink).await { + Ok(created) => { + if created { + // Fresh session minted this turn → flag it so the bin loop + // mints a `sessions` row + stamps its id onto this turn's stats. + bus.mark_fresh_session(); + bus.emit(LiveEvent::Note { + text: format!("created fresh session titled \"{}\"", session_title()), + }); + } + TurnOutcome::Ok + } + Err(e) => error_to_turn(e), } } @@ -643,19 +606,19 @@ pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> TurnOutcome { // Resume-only: never create a session just to compact it. If the titled // session doesn't exist there's genuinely nothing to compact — compacting // a freshly-minted empty session would just print "not enough messages" - // (the historical version-B failure), so treat a title miss as a no-op Ok. - let outcome = match run_claude("/compact", files, bus, false).await { - Ok(ClaudeResult::TitleNotFound) => { + // (the historical version-B failure), so treat a session miss as a no-op Ok. + let config = claude_config(bus, files); + let sink = BusSink::new(bus); + let session = Session::Resume(session_title()); + let outcome = match Claude::run(&config, &session, "/compact", &sink).await { + Ok(()) => TurnOutcome::Ok, + Err(hive_claude::Error::SessionNotFound) => { bus.emit(LiveEvent::Note { text: "no titled session to compact — skipping".into(), }); TurnOutcome::Ok } - Ok(ClaudeResult::PromptTooLong) => TurnOutcome::PromptTooLong, - Ok(ClaudeResult::RateLimited) => TurnOutcome::RateLimited, - Ok(ClaudeResult::AuthFailed) => TurnOutcome::AuthFailed, - Ok(ClaudeResult::Ok) => TurnOutcome::Ok, - Err(e) => TurnOutcome::Failed(e), + Err(e) => error_to_turn(e), }; match &outcome { TurnOutcome::Ok | TurnOutcome::Compacted => bus.emit(LiveEvent::Note { @@ -677,26 +640,6 @@ pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> TurnOutcome { outcome } -/// The recognized outcome of one `run_claude` invocation. Genuine failures -/// (spawn error, non-zero exit with no recognized sentinel) come back as -/// `Err` from `run_claude`; this enum captures every non-error outcome the -/// stdout/stderr classifier can distinguish. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum ClaudeResult { - /// Turn completed with no sentinel raised. - Ok, - /// `Prompt is too long` — the session is past the context window. - PromptTooLong, - /// API refused for rate-limit / usage-cap / credit reasons. - RateLimited, - /// API rejected with 401 (OAuth session expired/revoked). - AuthFailed, - /// `--resume <title>` matched no session (bootstrap / post-archive / - /// post-purge). Drives the one-shot `--name <title>` create retry in - /// [`run_claude_resume_or_create`]. - TitleNotFound, -} - /// The constant session title for this agent. `HIVE_SESSION_TITLE` overrides /// the compiled-in [`DEFAULT_SESSION_TITLE`]; each agent runs in its own /// container (own `~/.claude` + own `/state` cwd), so even the shared default @@ -710,9 +653,10 @@ pub fn session_title() -> String { .unwrap_or_else(|| DEFAULT_SESSION_TITLE.to_string()) } -/// The cwd claude is spawned in (mirrors `run_claude`): the agent's durable -/// `/state` dir when it exists, else the harness process cwd. Claude derives -/// its per-project session dir from this path. +/// The cwd claude is spawned in: the agent's durable `/state` dir when it +/// exists, else the harness process cwd. Claude derives its per-project +/// session dir from this path, so the same value feeds both the [`Config`] and +/// the [`hive_claude::SessionStore`]. fn session_cwd() -> PathBuf { let state_dir = crate::paths::state_dir(); if state_dir.is_dir() { @@ -722,87 +666,159 @@ fn session_cwd() -> PathBuf { } } -/// `~/.claude/projects/<slug>` for the current cwd. Claude slugises the -/// absolute cwd by replacing every `/` and `.` with `-` (empirically verified -/// against claude 2.1.197 — e.g. `/agents/iris/state` → `-agents-iris-state`). -/// Sessions (including any `choom` sessions sharing the cwd) live here as -/// `<uuid>.jsonl`. -fn claude_project_dir() -> PathBuf { - let cwd = session_cwd(); - let slug: String = cwd - .to_string_lossy() - .chars() - .map(|c| if c == '/' || c == '.' { '-' } else { c }) - .collect(); - crate::paths::claude_dir().join("projects").join(slug) +/// The on-disk session store for this agent (claude home + spawn cwd), used to +/// locate + archive the harness session by title. +fn session_store() -> hive_claude::SessionStore { + hive_claude::SessionStore::new(crate::paths::claude_dir(), session_cwd()) } -/// Find the `<uuid>.jsonl` in the current project dir whose `customTitle` -/// equals `title` (the value `--name` sets). Reads each session file -/// line-by-line and stops at the first marker hit, so a huge transcript isn't -/// slurped into memory. Returns `None` if no session carries the title -/// (bootstrap / post-archive) or the project dir is absent. Skips already -/// archived (`*.jsonl.archived`) files and any `choom` sessions that don't -/// carry our title. -fn find_session_file(title: &str) -> Option<PathBuf> { - let marker = format!("\"customTitle\":\"{title}\""); - let dir = claude_project_dir(); - for entry in std::fs::read_dir(&dir).ok()?.flatten() { - let path = entry.path(); - if path.extension().and_then(|e| e.to_str()) != Some("jsonl") { - continue; - } - let Ok(file) = std::fs::File::open(&path) else { - continue; - }; - if std::io::BufReader::new(file) - .lines() - .map_while(Result::ok) - .any(|line| line.contains(&marker)) - { - return Some(path); +/// Build the per-turn `hive_claude::Config` from the bus (model / effort) and +/// the materialised `TurnFiles` (system prompt + MCP config), plus the fixed +/// tool allow-lists and the optional docs `--add-dir`. +fn claude_config(bus: &Bus, files: &TurnFiles) -> Config { + let mut add_dirs = Vec::new(); + // hyperhive.docs.enable wires HIVE_DOCS_DIR to the in-container reference + // docs; expose it as an additional readable directory when set. + if let Some(docs_dir) = std::env::var_os("HIVE_DOCS_DIR") + && !docs_dir.is_empty() + { + add_dirs.push(PathBuf::from(docs_dir)); + } + let cwd = { + let state_dir = crate::paths::state_dir(); + state_dir.is_dir().then_some(state_dir) + }; + Config { + model: bus.model(), + effort: Some(bus.effort()), + cwd, + system_prompt_file: Some(files.system_prompt.clone()), + mcp_config: Some(files.mcp_config.clone()), + strict_mcp_config: true, + tools: Some(mcp::builtin_tools_arg()), + allowed_tools: Some(mcp::allowed_tools_arg()), + add_dirs, + ..Config::default() + } +} + +/// Map a `hive_claude::Error` onto the harness's `TurnOutcome`. The recognized +/// sentinels become their matching outcomes; a residual `SessionNotFound` +/// (create path itself missed — shouldn't happen) settles as `Ok`; genuine +/// failures become `Failed` (converting the typed lib error into `anyhow`). +fn error_to_turn(err: hive_claude::Error) -> TurnOutcome { + use hive_claude::Error; + match err { + Error::PromptTooLong => TurnOutcome::PromptTooLong, + Error::RateLimited => TurnOutcome::RateLimited, + Error::AuthFailed => TurnOutcome::AuthFailed, + Error::SessionNotFound => TurnOutcome::Ok, + other => TurnOutcome::Failed(other.into()), + } +} + +/// Bridges a claude run's output stream onto the hyperhive event bus: parses +/// per-turn token usage / resolved model / context-window from stream-json, +/// mirrors every event to the SSE bus, and surfaces non-JSON stdout + stderr +/// as Notes. Interior mutability (a `Mutex`) tracks the last inference across +/// events; the driver calls the `Sink` methods synchronously from one reader +/// task, so contention is nil — the lock only satisfies the `&self` trait +/// signature (and keeps `BusSink: Sync` for the driver's `Send` future). +struct BusSink<'a> { + bus: &'a Bus, + state: Mutex<BusSinkState>, +} + +#[derive(Default)] +struct BusSinkState { + last_inference: Option<TokenUsage>, + last_model: Option<String>, +} + +impl<'a> BusSink<'a> { + fn new(bus: &'a Bus) -> Self { + Self { + bus, + state: Mutex::new(BusSinkState::default()), } } - None +} + +impl Sink for BusSink<'_> { + fn on_event(&self, event: &Value) { + { + let mut st = self.state.lock().unwrap(); + // `last_inference` overwrites on every assistant event so at + // result-time it holds the most recent model call's usage — the + // actual context size. The `result` event carries the cumulative + // cost usage; both update the badges together. + if let Some(u) = TokenUsage::from_assistant_event(event) { + st.last_inference = Some(u); + } + if let Some(m) = TokenUsage::model_from_assistant_event(event) { + st.last_model = Some(m); + } + if let Some(cost) = TokenUsage::from_stream_event(event) { + let ctx = st.last_inference.unwrap_or(cost); + self.bus.record_turn_usage(ctx, cost); + self.bus.set_resolved_model(st.last_model.clone()); + } + } + // Seed the API-reported context-window from the result event's + // `modelUsage.*.contextWindow` — the authoritative active window for + // compaction watermarks. + if let Some(w) = TokenUsage::context_window_from_result_event(event) { + self.bus.set_api_context_window(w); + } + self.bus.observe_stream(event); + self.bus.emit(LiveEvent::Stream(event.clone())); + } + + fn on_stdout_line(&self, line: &str) { + self.bus.emit(LiveEvent::Note { + text: format!("(non-json) {line}"), + }); + } + + fn on_stderr_line(&self, line: &str) { + // Mirror to journald so post-mortems work without the web UI / events + // sqlite; the bus Note is what the dashboard renders. + tracing::warn!(line = %line, "claude stderr"); + self.bus.emit(LiveEvent::Note { + text: format!("stderr: {line}"), + }); + } } /// Archive (do NOT delete) the harness's own session so the next turn's /// `--resume <title>` misses and self-heals into a fresh `--name <title>` -/// session. Renames the backing `<uuid>.jsonl` → `<uuid>.jsonl.archived`, -/// which drops it out of claude's `*.jsonl` resolution glob while preserving -/// the full transcript on disk for forensics. Only the file carrying OUR -/// `customTitle` is touched — any `choom` sessions sharing the cwd are left -/// alone. Best-effort: emits a Note on success, on nothing-to-archive, and on -/// error; never fails a turn. Only ever called at a turn boundary (top of +/// session. Delegates the rename to [`hive_claude::SessionStore::archive_by_title`] +/// (which touches only the file carrying OUR `customTitle`, leaving any `choom` +/// session sharing the cwd alone) and surfaces the result as a Note. Best- +/// effort: never fails a turn. Only ever called at a turn boundary (top of /// `drive_turn` for an operator reset, or `maybe_auto_reset` pre-turn) so no /// claude process holds the session file open when it's renamed. fn archive_session(bus: &Bus) { let title = session_title(); - let Some(path) = find_session_file(&title) else { - bus.emit(LiveEvent::Note { - text: format!( - "no existing session titled \"{title}\" to archive — next turn starts fresh" - ), - }); - return; - }; - let mut target = path.clone().into_os_string(); - target.push(".archived"); - let target = PathBuf::from(target); - let name = path - .file_name() - .and_then(|n| n.to_str()) - .unwrap_or("?") - .to_string(); - match std::fs::rename(&path, &target) { - Ok(()) => { - tracing::info!(from = %path.display(), to = %target.display(), "archived claude session"); + match session_store().archive_by_title(&title) { + Ok(Some(path)) => { + let name = path + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("?") + .to_string(); + tracing::info!(path = %path.display(), "archived claude session"); bus.emit(LiveEvent::Note { text: format!("archived session \"{title}\" ({name}) — next turn starts fresh"), }); } + Ok(None) => bus.emit(LiveEvent::Note { + text: format!( + "no existing session titled \"{title}\" to archive — next turn starts fresh" + ), + }), Err(e) => { - tracing::warn!(error = %e, path = %path.display(), "failed to archive claude session"); + tracing::warn!(error = %e, "failed to archive claude session"); bus.emit(LiveEvent::Note { text: format!("failed to archive session \"{title}\": {e}"), }); @@ -810,280 +826,6 @@ fn archive_session(bus: &Bus) { } } -/// Resume the constant-title session, creating it on first use. Runs -/// `--resume <title>`; if claude reports the title doesn't resolve yet -/// (bootstrap / post-archive / post-purge), re-runs the SAME prompt once with -/// `--name <title>` to mint it. This is the single self-heal rule that -/// replaces the old scrape-persist-UUID machinery. -async fn run_claude_resume_or_create( - prompt: &str, - files: &TurnFiles, - bus: &Bus, -) -> Result<ClaudeResult> { - match run_claude(prompt, files, bus, false).await? { - ClaudeResult::TitleNotFound => run_claude(prompt, files, bus, true).await, - other => Ok(other), - } -} - -#[allow( - clippy::too_many_lines, - reason = "one linear subprocess driver: spawn claude, stream + classify \ - stdout/stderr, then assemble the outcome; splitting it would \ - fragment the streaming state across helpers" -)] -async fn run_claude( - prompt: &str, - files: &TurnFiles, - bus: &Bus, - create: bool, -) -> Result<ClaudeResult> { - // Keep the last STDERR_TAIL_LINES of stderr so a non-zero exit can - // include real context in the bail message (and downstream in the - // failure notification to the manager) instead of just "exit 1". - const STDERR_TAIL_LINES: usize = 20; - let model = bus.model(); - let effort = bus.effort(); - // Constant session identity. Every call keys on the same fixed title: - // `--resume <title>` normally, `--name <title>` on the create path (first - // use / post-archive / post-purge, driven by `run_claude_resume_or_create` - // on a title miss). We NEVER pass bare `--continue`: that resumes the - // latest session in this cwd, which a `choom` invocation (same cwd) can - // hijack — a constant title is immune since choom won't carry it. No - // scraped UUID, no persist file, so compaction + the post-compact retry - // provably target the same session. - let title = session_title(); - if create { - // Fresh session: flag it so the bin loop mints a new `sessions` row + - // stamps its id onto this turn's stats. - bus.mark_fresh_session(); - bus.emit(LiveEvent::Note { - text: format!("creating fresh session titled \"{title}\""), - }); - } - let mut cmd = Command::new("claude"); - // Spawn inside the agent's state dir so relative paths in tool calls - // (Read foo.md, Bash ls, Write notes.md) land in the durable dir - // instead of wherever the harness systemd unit started. Falls back - // silently if the dir is missing (dev / test without the bind mount). - let state_dir = crate::paths::state_dir(); - if state_dir.is_dir() { - cmd.current_dir(&state_dir); - } - cmd.arg("--print") - .arg("--verbose") - .arg("--output-format") - .arg("stream-json") - .arg("--model") - .arg(&model) - .arg("--effort") - .arg(&effort); - if create { - cmd.arg("--name").arg(&title); - } else { - cmd.arg("--resume").arg(&title); - } - cmd.arg("--system-prompt-file").arg(&files.system_prompt); - cmd.arg("--mcp-config") - .arg(&files.mcp_config) - .arg("--strict-mcp-config") - .arg("--tools") - .arg(mcp::builtin_tools_arg()) - .arg("--allowedTools") - .arg(mcp::allowed_tools_arg()); - // hyperhive.docs.enable wires HIVE_DOCS_DIR to the in-container - // reference-docs tree (with a generic CLAUDE.md pointer at its root). - // Expose it to claude as an additional directory so the docs are - // readable; harness-base.nix also sets - // CLAUDE_CODE_ADDITIONAL_DIRECTORIES_CLAUDE_MD=1 so claude loads that - // pointer additively (never replacing the agent's own CLAUDE.md). - // Unset (docs disabled) → the flag is not passed. - if let Some(docs_dir) = std::env::var_os("HIVE_DOCS_DIR") - && !docs_dir.is_empty() - { - cmd.arg("--add-dir").arg(&docs_dir); - } - let mut child = cmd - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn()?; - - if let Some(mut stdin) = child.stdin.take() { - stdin.write_all(prompt.as_bytes()).await?; - stdin.shutdown().await.ok(); - drop(stdin); - } - let stdout = child.stdout.take().expect("piped stdout"); - let stderr = child.stderr.take().expect("piped stderr"); - - let prompt_too_long = Arc::new(AtomicBool::new(false)); - let rate_limited = Arc::new(AtomicBool::new(false)); - let auth_failed = Arc::new(AtomicBool::new(false)); - // `--resume <title>` found no session carrying the title: the caller - // (`run_claude_resume_or_create`) re-runs once with `--name <title>`. - let title_not_found = Arc::new(AtomicBool::new(false)); - let flag_out = prompt_too_long.clone(); - let flag_err = prompt_too_long.clone(); - let rate_out = rate_limited.clone(); - let rate_err = rate_limited.clone(); - let auth_out = auth_failed.clone(); - let auth_err = auth_failed.clone(); - let notfound_out = title_not_found.clone(); - let notfound_err = title_not_found.clone(); - let bus_out = bus.clone(); - let bus_err = bus.clone(); - let pump_stdout = tokio::spawn(async move { - let mut reader = BufReader::new(stdout).lines(); - // Track usage as the turn unfolds. `last_inference` overwrites on - // every assistant event so at result-time it holds the most recent - // model call's usage — the actual context size. The `result` event - // carries the cumulative-across-the-turn usage (cost signal). Both - // get handed to `record_turn_usage` together so a single SSE - // event updates both badges. - let mut last_inference: Option<crate::events::TokenUsage> = None; - // Resolved model id (API-echoed `message.model`) from this turn's - // assistant events; recorded onto the bus at result-time so the - // per-turn stats label the concrete version that ran, not the - // requested `--model` alias. - let mut last_model: Option<String> = None; - while let Ok(Some(line)) = reader.next_line().await { - if line.contains(PROMPT_TOO_LONG_MARKER) { - flag_out.store(true, Ordering::Relaxed); - } - // Auth-fail check happens on the raw line first so we - // catch both the `api_retry` JSON events (which can land - // before they're fully parseable) and any stderr-shaped - // text that snuck onto stdout. - if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) { - auth_out.store(true, Ordering::Relaxed); - } - if TITLE_NOT_FOUND_MARKERS.iter().any(|m| line.contains(m)) { - notfound_out.store(true, Ordering::Relaxed); - } - if let Ok(v) = serde_json::from_str::<serde_json::Value>(&line) { - // Rate-limit detection: only fire on JSON `error` events, - // not on arbitrary text content. An agent discussing a past - // rate limit in its response would otherwise trigger a false - // positive (the full conversation flows through stdout as - // stream-json, so any text the model outputs is visible here). - if v.get("type").and_then(|t| t.as_str()) == Some("error") { - let raw = v.to_string(); - if RATE_LIMIT_MARKERS.iter().any(|m| raw.contains(m)) { - rate_out.store(true, Ordering::Relaxed); - } - } - if let Some(u) = crate::events::TokenUsage::from_assistant_event(&v) { - last_inference = Some(u); - } - if let Some(m) = crate::events::TokenUsage::model_from_assistant_event(&v) { - last_model = Some(m); - } - if let Some(cost) = crate::events::TokenUsage::from_stream_event(&v) { - // Fallback to `cost` if the turn somehow produced - // a result without any assistant event — keeps the - // ctx badge from going stale on a degenerate turn. - let ctx = last_inference.unwrap_or(cost); - bus_out.record_turn_usage(ctx, cost); - // Pin the resolved model for this turn's stats row - // (cleared to None if no assistant event reported one - // → stats sink falls back to the requested name). - bus_out.set_resolved_model(last_model.clone()); - } - // Seed the API-reported context-window from the result - // event's `modelUsage.*.contextWindow` field. This is - // the authoritative per-inference active window used for - // compaction watermarks — it reflects what the model - // actually enforces, which may differ from the Nix - // config (e.g. 200k active window on a 1M cache model). - if let Some(w) = crate::events::TokenUsage::context_window_from_result_event(&v) { - bus_out.set_api_context_window(w); - } - bus_out.observe_stream(&v); - bus_out.emit(LiveEvent::Stream(v)); - } else { - // Non-JSON stdout: raw text check is fine here since these - // are claude CLI messages, not conversation content. - if RATE_LIMIT_MARKERS.iter().any(|m| line.contains(m)) { - rate_out.store(true, Ordering::Relaxed); - } - bus_out.emit(LiveEvent::Note { - text: format!("(non-json) {line}"), - }); - } - } - }); - let stderr_tail: Arc<Mutex<VecDeque<String>>> = - Arc::new(Mutex::new(VecDeque::with_capacity(STDERR_TAIL_LINES))); - let tail_clone = stderr_tail.clone(); - let pump_stderr = tokio::spawn(async move { - let mut reader = BufReader::new(stderr).lines(); - while let Ok(Some(line)) = reader.next_line().await { - if line.contains(PROMPT_TOO_LONG_MARKER) { - flag_err.store(true, Ordering::Relaxed); - } - if RATE_LIMIT_MARKERS.iter().any(|m| line.contains(m)) { - rate_err.store(true, Ordering::Relaxed); - } - if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) { - auth_err.store(true, Ordering::Relaxed); - } - if TITLE_NOT_FOUND_MARKERS.iter().any(|m| line.contains(m)) { - notfound_err.store(true, Ordering::Relaxed); - } - // Mirror to journald so post-mortems work without the web UI - // or the events sqlite. The bus event is what the dashboard - // renders; the tracing line is what `journalctl -M <c> -b` - // surfaces when claude exits non-zero. - tracing::warn!(line = %line, "claude stderr"); - bus_err.emit(LiveEvent::Note { - text: format!("stderr: {line}"), - }); - let mut t = tail_clone.lock().unwrap(); - if t.len() >= STDERR_TAIL_LINES { - t.pop_front(); - } - t.push_back(line); - } - }); - - let status = child.wait().await?; - let _ = pump_stdout.await; - let _ = pump_stderr.await; - let too_long = prompt_too_long.load(Ordering::Relaxed); - let is_rate_limited = rate_limited.load(Ordering::Relaxed); - let is_auth_failed = auth_failed.load(Ordering::Relaxed); - let is_title_not_found = title_not_found.load(Ordering::Relaxed); - // A title miss is a clean exit-1 (no session to resume yet), so it must - // not be treated as a hard failure — the caller re-runs with `--name`. - if !status.success() - && !too_long - && !is_rate_limited - && !is_auth_failed - && !is_title_not_found - { - let tail = stderr_tail.lock().unwrap(); - if tail.is_empty() { - bail!("claude exited {status} (no stderr)"); - } - let tail_str = tail.iter().cloned().collect::<Vec<_>>().join("\n"); - bail!("claude exited {status}\nstderr tail:\n{tail_str}"); - } - // Assemble the single recognized outcome. The failure sentinels keep - // their historical priority (too-long > rate > auth); a title miss only - // ever arises on a resume that made no model call, so it can't coincide. - Ok(if too_long { - ClaudeResult::PromptTooLong - } else if is_rate_limited { - ClaudeResult::RateLimited - } else if is_auth_failed { - ClaudeResult::AuthFailed - } else if is_title_not_found { - ClaudeResult::TitleNotFound - } else { - ClaudeResult::Ok - }) -} #[cfg(test)] mod tests { diff --git a/hive-claude/Cargo.toml b/hive-claude/Cargo.toml new file mode 100644 index 00000000..6218d201 --- /dev/null +++ b/hive-claude/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "hive-claude" +edition.workspace = true +version.workspace = true + +[lints] +workspace = true + +[dependencies] +serde_json.workspace = true +thiserror.workspace = true +tokio.workspace = true diff --git a/hive-claude/README.md b/hive-claude/README.md new file mode 100644 index 00000000..5f8a3a1e --- /dev/null +++ b/hive-claude/README.md @@ -0,0 +1,61 @@ +# hive-claude + +A small, reusable async driver for headless `claude --print` (Claude Code CLI) +sessions. It spawns the CLI, streams and classifies the `stream-json` output, +and reports the result. It knows only about the Claude Code CLI — **no +hyperhive types, policy, watermarks, or logging.** Compaction, retry, and reset +policy belong to the caller. + +## When to use it + +Reach for this crate whenever you need to run the `claude` CLI from Rust and +react to how a turn ended. It is the shared substrate under +`hive-ag3nt`'s turn loop; new callers (tools, tests, other agents) should build +on it rather than shelling out to `claude` by hand. + +## Shape + +- **`Claude::run(&config, &session, prompt, &sink)`** → `Result<(), Error>`. A + clean turn is `Ok(())`; every non-completion state is an [`Error`] variant, so + you branch with a single `match`. +- **`Claude::run_resume_or_create(&config, title, prompt, &sink)`** → + `Result<bool>`. Resumes a titled session, creating it on first use. + `Ok(true)` means a fresh session was minted. +- **`Config`** — the invocation (model, effort, cwd, prompt/MCP files, tools, + extra args). **`Session`** — which session to attach to (`Resume` / `Create` + / `Continue` / `OneOff`). +- **`Sink`** — a trait with no-op defaults; implement the methods you care + about to observe stream events, non-JSON stdout, and stderr. Use `NoopSink` + when you only want the result. +- **`SessionStore`** — locate and archive on-disk sessions by title. + +`Error` unifies the two things that can stop a turn: recognized **sentinels** +(`PromptTooLong`, `RateLimited`, `AuthFailed`, `SessionNotFound`) and **hard +failures** (`Spawn`, `Stdin`, `Wait`, `Exit`, `Io`). Sentinels are expected +control-flow, not crashes — the caller compacts, parks, re-auths, or creates a +session in response. + +```rust +use hive_claude::{Claude, Config, Error, NoopSink, Session}; + +let config = Config { model: "haiku".into(), ..Default::default() }; +match Claude::run(&config, &Session::Resume("my-session".into()), "hello", &NoopSink).await { + Ok(()) => {} + Err(Error::PromptTooLong) => { /* compact + retry */ } + Err(Error::RateLimited) => { /* park + retry */ } + Err(other) => eprintln!("claude: {other}"), +} +``` + +## `thiserror` here, `anyhow` in the apps + +This is a **library**, so it returns a concrete, matchable `Error` enum built +with `thiserror`: callers can tell a rate-limit from a spawn failure and act +accordingly. Libraries should never force their callers into `anyhow`'s +type-erased error. + +The **applications** (the `hive-*` binaries) use `anyhow` instead — at the top +level you usually only want to add context and bubble a failure up, not match +on it. A `hive_claude::Error` converts into an `anyhow::Error` for free at the +`?` boundary. Rule of thumb: **libraries return `thiserror` enums, binaries +consume them with `anyhow`.** diff --git a/hive-claude/src/classify.rs b/hive-claude/src/classify.rs new file mode 100644 index 00000000..dad9fbec --- /dev/null +++ b/hive-claude/src/classify.rs @@ -0,0 +1,106 @@ +//! Sentinel detection: mapping claude-code CLI output onto [`crate::Error`] +//! variants. +//! +//! These marker strings are claude-code CLI knowledge, not app knowledge. They +//! are empirically stable across CLI versions; if one drifts the run degrades +//! gracefully (a clean turn, or a hard [`crate::Error::Exit`] on a non-zero +//! exit) rather than misbehaving. + +use std::sync::atomic::{AtomicBool, Ordering}; + +/// Emitted when the prompt/context exceeds the model's window. +const PROMPT_TOO_LONG_MARKER: &str = "Prompt is too long"; + +/// Substrings indicating the API refused for rate-limit / usage-cap / credit +/// reasons. On stdout these are only trusted inside a JSON `error` event (see +/// [`Sentinels::scan_stdout_json`] / [`Sentinels::scan_rate_limit_text`]) so a +/// model *discussing* a rate limit in prose can't trigger a false positive. +const RATE_LIMIT_MARKERS: [&str; 5] = [ + "rate_limit_error", + "overloaded_error", + "Credit balance is too low", + "Usage limit reached", + "Request rate limit exceeded", +]; + +/// Substrings indicating the API rejected the request as unauthenticated (401) +/// — an expired/revoked OAuth session. Sourced from claude-code's `api_retry` +/// JSON events and its human-readable give-up line. +const AUTH_FAIL_MARKERS: [&str; 3] = [ + "\"error\":\"authentication_failed\"", + "\"error_status\":401", + "Failed to authenticate. API Error: 401", +]; + +/// Substrings indicating `--resume` could not resolve its target: no session +/// with the given title, or no conversation with the given id. +const SESSION_NOT_FOUND_MARKERS: [&str; 2] = [ + "does not match any session title", + "No conversation found with session ID", +]; + +/// Shared, lock-free sentinel flags accumulated while both output streams are +/// pumped concurrently. Read once after the child exits. +#[derive(Default)] +pub(crate) struct Sentinels { + prompt_too_long: AtomicBool, + rate_limited: AtomicBool, + auth_failed: AtomicBool, + session_not_found: AtomicBool, +} + +impl Sentinels { + /// Scan a raw line (stdout or stderr) for the always-on markers: + /// prompt-too-long, auth-failed, session-not-found. Rate-limit is handled + /// separately because on stdout it must only fire on JSON `error` events. + pub(crate) fn scan_line(&self, line: &str) { + if line.contains(PROMPT_TOO_LONG_MARKER) { + self.prompt_too_long.store(true, Ordering::Relaxed); + } + if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) { + self.auth_failed.store(true, Ordering::Relaxed); + } + if SESSION_NOT_FOUND_MARKERS.iter().any(|m| line.contains(m)) { + self.session_not_found.store(true, Ordering::Relaxed); + } + } + + /// Trust a rate-limit hit on a JSON `error` event's serialized form. + pub(crate) fn scan_stdout_json(&self, event: &serde_json::Value) { + if event.get("type").and_then(|t| t.as_str()) == Some("error") + && RATE_LIMIT_MARKERS + .iter() + .any(|m| event.to_string().contains(m)) + { + self.rate_limited.store(true, Ordering::Relaxed); + } + } + + /// Trust a rate-limit hit on raw text (non-JSON stdout, or any stderr) — + /// these are CLI messages, not conversation content. + pub(crate) fn scan_rate_limit_text(&self, line: &str) { + if RATE_LIMIT_MARKERS.iter().any(|m| line.contains(m)) { + self.rate_limited.store(true, Ordering::Relaxed); + } + } + + /// The recognized-sentinel error, if any fired — `None` means no sentinel + /// (so the run either completed or failed hard on its exit code). The + /// sentinels keep a fixed priority (too-long > rate > auth); a + /// session-not-found can only arise on a resume that made no model call, + /// so it never coincides with the others. + pub(crate) fn soft_error(&self) -> Option<crate::Error> { + use crate::Error; + if self.prompt_too_long.load(Ordering::Relaxed) { + Some(Error::PromptTooLong) + } else if self.rate_limited.load(Ordering::Relaxed) { + Some(Error::RateLimited) + } else if self.auth_failed.load(Ordering::Relaxed) { + Some(Error::AuthFailed) + } else if self.session_not_found.load(Ordering::Relaxed) { + Some(Error::SessionNotFound) + } else { + None + } + } +} diff --git a/hive-claude/src/config.rs b/hive-claude/src/config.rs new file mode 100644 index 00000000..09129e53 --- /dev/null +++ b/hive-claude/src/config.rs @@ -0,0 +1,58 @@ +//! Invocation config: how to build one `claude --print` command line. + +use std::path::PathBuf; + +/// Which claude session a run should attach to. Kept separate from [`Config`] +/// so a single config can drive resume + create + `/compact` of the same +/// logical session across turns. +#[derive(Debug, Clone)] +pub enum Session { + /// `--resume <id-or-title>` — resume an existing session by UUID or by the + /// display title set via [`Session::Create`]. Yields + /// [`crate::Outcome::SessionNotFound`] if nothing matches. + Resume(String), + /// `--name <title>` — start a new session carrying the given display title + /// (persisted as a `custom-title` event, which `--resume <title>` later + /// resolves against). + Create(String), + /// `--continue` — resume the most recent session in the cwd. Ambiguous + /// when other claude processes share the cwd; prefer titled sessions. + Continue, + /// No session flag — a one-off, unnamed session. + OneOff, +} + +/// Everything needed to build one headless `claude --print` invocation, minus +/// the [`Session`] attachment (passed separately to [`crate::run`]). +/// +/// Fields map one-to-one to CLI flags; `None`/empty means "don't pass the +/// flag". `--print --verbose --output-format stream-json` are always set by +/// the driver and are not configurable here (the driver depends on the +/// stream-json shape). +#[derive(Debug, Clone, Default)] +pub struct Config { + /// `--model`. Empty omits the flag (claude falls back to its own default). + pub model: String, + /// `--effort <level>`. `None` omits the flag. + pub effort: Option<String>, + /// Working directory for the child. Claude derives its per-project session + /// dir from this path. `None` inherits the parent process cwd. + pub cwd: Option<PathBuf>, + /// `--system-prompt-file <path>`. + pub system_prompt_file: Option<PathBuf>, + /// `--mcp-config <path>`. + pub mcp_config: Option<PathBuf>, + /// Pass `--strict-mcp-config` (only the configured MCP servers, no + /// discovery). + pub strict_mcp_config: bool, + /// `--tools <expr>` — the built-in tool allow-list expression. + pub tools: Option<String>, + /// `--allowedTools <expr>`. + pub allowed_tools: Option<String>, + /// `--add-dir <path>` (repeatable) — extra readable directories. + pub add_dirs: Vec<PathBuf>, + /// Any additional raw args appended verbatim after the ones above. + pub extra_args: Vec<String>, + /// Program to spawn. `None` defaults to `claude` (resolved on `PATH`). + pub program: Option<String>, +} diff --git a/hive-claude/src/driver.rs b/hive-claude/src/driver.rs new file mode 100644 index 00000000..ad5f99a1 --- /dev/null +++ b/hive-claude/src/driver.rs @@ -0,0 +1,209 @@ +//! The subprocess driver: spawn claude, pump + classify its streams, and +//! assemble an [`Outcome`]. + +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::{Config, Error, Result, Session, 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 driver entry point. A namespace for the run functions — there is +/// nothing to construct; call the associated functions directly +/// (`Claude::run(…)`, `Claude::run_resume_or_create(…)`). +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, + session: &Session, + prompt: &str, + sink: &impl Sink, + ) -> Result<()> { + let program = config.program.as_deref().unwrap_or(DEFAULT_PROGRAM); + let mut cmd = build_command(program, config, session); + + 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(()) + } + + /// Resume a titled session, creating it on first use. Runs + /// `--resume <title>`; on [`Error::SessionNotFound`] it re-runs the *same + /// prompt* once with `--name <title>` to mint the session. + /// + /// Returns `Ok(true)` when a fresh session was created (the resume missed), + /// `Ok(false)` when an existing session was resumed. This is the common + /// "one durable session per agent, self-healing on first boot / after the + /// file is archived away" pattern, kept generic here. + /// + /// # Errors + /// + /// Propagates any [`Error`] from the underlying [`Claude::run`] calls + /// (other than the `SessionNotFound` on the first attempt, which is handled + /// by creating). + pub async fn run_resume_or_create( + config: &Config, + title: &str, + prompt: &str, + sink: &impl Sink, + ) -> Result<bool> { + match Self::run(config, &Session::Resume(title.to_string()), prompt, sink).await { + Err(Error::SessionNotFound) => { + Self::run(config, &Session::Create(title.to_string()), prompt, sink).await?; + Ok(true) + } + Ok(()) => Ok(false), + Err(other) => Err(other), + } + } +} + +/// Assemble the argv. `--print --verbose --output-format stream-json` are +/// mandatory (the driver parses that shape); everything else is gated on the +/// [`Config`] / [`Session`]. +fn build_command(program: &str, config: &Config, session: &Session) -> 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 session { + Session::Resume(id) => { + cmd.arg("--resume").arg(id); + } + Session::Create(title) => { + cmd.arg("--name").arg(title); + } + Session::Continue => { + cmd.arg("--continue"); + } + Session::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 { + sentinels.scan_line(&line); + if let Ok(event) = serde_json::from_str::<serde_json::Value>(&line) { + sentinels.scan_stdout_json(&event); + sink.on_event(&event); + } else { + sentinels.scan_rate_limit_text(&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 { + sentinels.scan_line(&line); + sentinels.scan_rate_limit_text(&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") +} diff --git a/hive-claude/src/error.rs b/hive-claude/src/error.rs new file mode 100644 index 00000000..e29deecd --- /dev/null +++ b/hive-claude/src/error.rs @@ -0,0 +1,77 @@ +//! Typed errors for the driver. See the crate-level docs for why this is a +//! `thiserror` enum rather than `anyhow`. + +use std::process::ExitStatus; + +use thiserror::Error; + +/// Why a claude run did not complete cleanly. A normal, finished turn is +/// `Ok(())`; everything else is one of these variants. +/// +/// Two families share the enum on purpose, so a caller can handle them with a +/// single `match` on the `Result`: +/// +/// - **Recognized sentinels** — [`Error::PromptTooLong`], +/// [`Error::RateLimited`], [`Error::AuthFailed`], [`Error::SessionNotFound`]. +/// These are *expected* non-completion states parsed from claude's output, +/// not crashes; callers typically compact, park-and-retry, re-auth, or +/// create-a-session in response. +/// - **Hard failures** — [`Error::Spawn`], [`Error::Stdin`], [`Error::Wait`], +/// [`Error::Exit`], [`Error::Io`]. The process couldn't run, or died with no +/// recognizable reason. +#[derive(Debug, Error)] +pub enum Error { + /// `Prompt is too long` — the session is past the model's context window. + #[error("prompt is too long for the model's context window")] + PromptTooLong, + + /// The API refused for rate-limit / usage-cap / credit-balance reasons. + #[error("request was rate-limited or hit a usage/credit cap")] + RateLimited, + + /// The API rejected the request with 401 (auth/session expired or revoked). + #[error("authentication failed (HTTP 401)")] + AuthFailed, + + /// `--resume` matched no session for the given id or title (e.g. the title + /// was never created, or its backing file was moved away). + #[error("no session matched the requested id or title")] + SessionNotFound, + + /// The `claude` binary could not be spawned (not on `PATH`, not + /// executable, …). + #[error("failed to spawn `{program}`: {source}")] + Spawn { + /// The program name we tried to run. + program: String, + /// The underlying spawn error. + #[source] + source: std::io::Error, + }, + + /// Writing the prompt to claude's stdin failed. + #[error("writing prompt to claude stdin failed: {0}")] + Stdin(#[source] std::io::Error), + + /// Awaiting the child process failed. + #[error("waiting on claude failed: {0}")] + Wait(#[source] std::io::Error), + + /// claude exited non-zero and raised none of the recognized sentinels. + /// `stderr_tail` is the last handful of stderr lines (empty if there were + /// none), included so the caller can surface a real diagnostic. + #[error("claude exited {status}\n{stderr_tail}")] + Exit { + /// The child's exit status. + status: ExitStatus, + /// Tail of stderr, newline-joined; empty when claude wrote nothing. + stderr_tail: String, + }, + + /// A filesystem operation (session lookup / archive) failed. + #[error("session store i/o failed: {0}")] + Io(#[from] std::io::Error), +} + +/// Convenience alias for results from this crate. +pub type Result<T> = std::result::Result<T, Error>; diff --git a/hive-claude/src/lib.rs b/hive-claude/src/lib.rs new file mode 100644 index 00000000..03934a50 --- /dev/null +++ b/hive-claude/src/lib.rs @@ -0,0 +1,61 @@ +//! `hive-claude` — a small, reusable async driver for headless +//! `claude --print` (Claude Code CLI) sessions. +//! +//! It spawns the CLI, streams and classifies its `stream-json` output, and +//! reports the result as a `Result<(), Error>`: a clean turn is `Ok(())`, and +//! every non-completion state — both recognized sentinels (rate-limit, +//! prompt-too-long, …) and hard failures (spawn, non-zero exit) — is a variant +//! of the single [`Error`] enum, so callers branch with one `match`. The crate +//! also locates and archives on-disk sessions by title ([`SessionStore`]). It +//! knows only about the Claude Code CLI — no application types, policy, +//! watermarks, or logging. Callers wire streaming output through a [`Sink`] and +//! layer their own compaction / retry / reset policy on top. +//! +//! # `thiserror` here, `anyhow` in the apps +//! +//! This is a **library**, so it exposes a concrete, matchable error enum built +//! with [`thiserror`](https://docs.rs/thiserror): a caller can distinguish +//! `Error::Exit { status, .. }` from `Error::Spawn { .. }` and branch on it. +//! A library should never force its callers to reach into `anyhow`'s +//! type-erased error to find out what went wrong. +//! +//! The **applications** in this workspace (the `hive-*` binaries) use +//! [`anyhow`](https://docs.rs/anyhow) instead. At the top level you usually +//! only want to attach context and log or bubble a failure up — not match on +//! it — and `anyhow::Result` + `?` + `.context()` is the ergonomic fit. +//! `anyhow::Error` implements `From<E>` for any `std::error::Error`, so a +//! `hive_claude::Error` converts into an `anyhow::Error` for free at the `?` +//! boundary. Rule of thumb: **libraries return `thiserror` enums, binaries +//! consume them with `anyhow`.** +//! +//! # Example +//! +//! ```no_run +//! # async fn ex() { +//! use hive_claude::{Claude, Config, Error, NoopSink, Session}; +//! +//! let config = Config { +//! model: "haiku".into(), +//! ..Default::default() +//! }; +//! match Claude::run(&config, &Session::Resume("my-session".into()), "hello", &NoopSink).await { +//! Ok(()) => {} +//! Err(Error::PromptTooLong) => { /* caller compacts + retries */ } +//! Err(Error::RateLimited) => { /* caller parks + retries */ } +//! Err(other) => eprintln!("claude: {other}"), +//! } +//! # } +//! ``` + +mod classify; +mod config; +mod driver; +mod error; +mod sink; +mod store; + +pub use config::{Config, Session}; +pub use driver::Claude; +pub use error::{Error, Result}; +pub use sink::{NoopSink, Sink}; +pub use store::SessionStore; diff --git a/hive-claude/src/sink.rs b/hive-claude/src/sink.rs new file mode 100644 index 00000000..0c82f6c2 --- /dev/null +++ b/hive-claude/src/sink.rs @@ -0,0 +1,30 @@ +//! Streaming-output consumer hook. + +use serde_json::Value; + +/// Consumer callbacks for a claude run's output streams. Every method has a +/// no-op default, so an implementor overrides only what it needs. +/// +/// Methods are called synchronously from the stdout/stderr readers as lines +/// arrive, so keep them cheap — the idiomatic body forwards to a channel or an +/// event bus rather than blocking. The driver handles sentinel classification +/// (rate-limit, prompt-too-long, …) itself; a sink only *observes* the stream. +pub trait Sink { + /// A parsed `stream-json` object from stdout (assistant/result/system/… + /// event). The driver has already classified it for sentinels. + fn on_event(&self, _event: &Value) {} + + /// A stdout line that was not valid JSON — occasional Claude Code CLI + /// chatter rather than conversation content. + fn on_stdout_line(&self, _line: &str) {} + + /// A stderr line, delivered verbatim. The last several are also retained + /// by the driver for [`crate::Error::Exit`]. + fn on_stderr_line(&self, _line: &str) {} +} + +/// A [`Sink`] that discards everything. Useful for fire-and-forget runs where +/// only the [`crate::Outcome`] matters. +pub struct NoopSink; + +impl Sink for NoopSink {} diff --git a/hive-claude/src/store.rs b/hive-claude/src/store.rs new file mode 100644 index 00000000..064aa924 --- /dev/null +++ b/hive-claude/src/store.rs @@ -0,0 +1,95 @@ +//! Locating and archiving on-disk claude sessions by title. + +use std::io::BufRead as _; +use std::path::PathBuf; + +use crate::{Error, Result}; + +/// On-disk claude session store scoped to one working directory. +/// +/// Claude Code keeps sessions under +/// `<claude_home>/projects/<slug(cwd)>/<uuid>.jsonl`, one project dir per cwd. +/// This type locates and archives those files by their display title. It is +/// generic Claude Code layout knowledge — no application specifics. +#[derive(Debug, Clone)] +pub struct SessionStore { + claude_home: PathBuf, + cwd: PathBuf, +} + +impl SessionStore { + /// Build a store for `cwd`, with claude's home dir (normally `~/.claude`) + /// at `claude_home`. + pub fn new(claude_home: impl Into<PathBuf>, cwd: impl Into<PathBuf>) -> Self { + Self { + claude_home: claude_home.into(), + cwd: cwd.into(), + } + } + + /// `<claude_home>/projects/<slug>` for this cwd. Claude slugises the + /// absolute cwd by replacing every `/` and `.` with `-` (verified against + /// claude 2.1.197 — e.g. `/agents/iris/state` → `-agents-iris-state`). + #[must_use] + pub fn project_dir(&self) -> PathBuf { + let slug: String = self + .cwd + .to_string_lossy() + .chars() + .map(|c| if c == '/' || c == '.' { '-' } else { c }) + .collect(); + self.claude_home.join("projects").join(slug) + } + + /// Find the `<uuid>.jsonl` in the project dir whose `customTitle` equals + /// `title` (the value `--name` sets, stored in a `custom-title` event). + /// + /// Reads each session file line by line and stops at the first match, so a + /// huge transcript isn't slurped into memory. Returns `None` if no session + /// carries the title or the project dir is absent. Non-`.jsonl` files + /// (including anything already archived to `*.jsonl.archived`) are skipped. + #[must_use] + pub fn find_by_title(&self, title: &str) -> Option<PathBuf> { + let marker = format!("\"customTitle\":\"{title}\""); + for entry in std::fs::read_dir(self.project_dir()).ok()?.flatten() { + let path = entry.path(); + if path.extension().and_then(|e| e.to_str()) != Some("jsonl") { + continue; + } + let Ok(file) = std::fs::File::open(&path) else { + continue; + }; + if std::io::BufReader::new(file) + .lines() + .map_while(std::result::Result::ok) + .any(|line| line.contains(&marker)) + { + return Some(path); + } + } + None + } + + /// Archive the session titled `title` by renaming its backing file + /// `<uuid>.jsonl` → `<uuid>.jsonl.archived`. That drops it out of claude's + /// `*.jsonl` resolution glob (so a later `--resume <title>` misses) while + /// preserving the full transcript on disk. Only the file with a matching + /// `customTitle` is touched. + /// + /// Returns the archived file's new path, or `None` if no session carried + /// the title. + /// + /// # Errors + /// + /// [`Error::Io`] if the rename fails. + pub fn archive_by_title(&self, title: &str) -> Result<Option<PathBuf>> { + let Some(path) = self.find_by_title(title) else { + return Ok(None); + }; + let mut target = path.clone().into_os_string(); + target.push(".archived"); + let target = PathBuf::from(target); + std::fs::rename(&path, &target).map_err(Error::Io)?; + Ok(Some(target)) + } +}