vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

consume hive-claude via git dependency instead of an in-tree copy

Browse source 
fixes hyperhive/hyperhive#2841. hive-claude was extracted to its own
repo a while back (hyperhive/hive-claude) but this workspace kept
vendoring a path dependency on an in-tree copy - two copies with
nothing keeping them in sync.

drops "hive-claude" from workspace members, switches the dependency
to a git source pinned at hive-claude's current main (2900cc2), and
deletes the in-tree hive-claude/ directory.

this is the workspace's first git-sourced cargo dependency. verified:
cargo build/test/clippy -p hive-agent (the only consumer) all clean,
nix fmt 0 files changed, nix flake check all green (crane vendors the
git dep from Cargo.lock via its own fixed-output derivation, no extra
plumbing needed). network reachability + credentials for the
canonical host checked separately on hive-claude#1 - public repo, no
auth needed, hive-ci confirmed reachable; hive-priv (host netns) is
the one path nobody can test from a container, atlas is running a
one-agent canary rebuild once this merges as the real confirmation.
This commit is contained in:
damocles 2026-07-29 21:38:52 +02:00 • committed by mara
commit 2e0096caf4
14 changed files with 2 additions and 1686 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Cargo.lock generated
View file

@ -1661,10 +1661,10 @@ dependencies = [
[[package]] [[package]]
name = "hive-claude" name = "hive-claude"
version = "0.1.0" version = "0.1.0"
source = "git+https://forge.darkest.space/hyperhive/hive-claude?branch=main#4ee1940e91b0f713f1767fc5ba84426a82d562f8"
dependencies = [ dependencies = [
"serde", "serde",
"serde_json", "serde_json",
"tempfile",
"thiserror 2.0.18", "thiserror 2.0.18",
"tokio", "tokio",
"tracing", "tracing",

3
Cargo.toml
View file

@ -7,7 +7,6 @@ members = [
"hive-core-agent-sock", "hive-core-agent-sock",
"hive-bash-mcp", "hive-bash-mcp",
"hive-c0re", "hive-c0re",
"hive-claude",
"hive-screen-mcp", "hive-screen-mcp",
"hive-forge", "hive-forge",
"hive-forge-notify", "hive-forge-notify",
@ -57,7 +56,7 @@ hive-sh4re = { path = "hive-sh4re" }
hive-agent-sock = { path = "hive-agent-sock" } hive-agent-sock = { path = "hive-agent-sock" }
hive-jobq = { path = "hive-jobq" } hive-jobq = { path = "hive-jobq" }
hive-core-agent-sock = { path = "hive-core-agent-sock" } hive-core-agent-sock = { path = "hive-core-agent-sock" }
hive-claude = { path = "hive-claude" } hive-claude = { git = "https://forge.darkest.space/hyperhive/hive-claude", branch = "main" }
hive-host-sock = { path = "hive-host-sock" } hive-host-sock = { path = "hive-host-sock" }
hive-priv-sock = { path = "hive-priv-sock" } hive-priv-sock = { path = "hive-priv-sock" }
hive-sock-client = { path = "hive-sock-client" } hive-sock-client = { path = "hive-sock-client" }

18
hive-claude/Cargo.toml
View file

@ -1,18 +0,0 @@
[package]
name = "hive-claude"
edition.workspace = true
version.workspace = true
readme = "README.md"
[lints]
workspace = true
[dependencies]
serde = { workspace = true }
serde_json.workspace = true
thiserror.workspace = true
tokio.workspace = true
tracing.workspace = true
[dev-dependencies]
tempfile = "3"

75
hive-claude/README.md
View file

@ -1,75 +0,0 @@
# 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
Two layers — reach for the high-level one:
- **`InfiniteSession { name, store, policy }`** — a durable session that keeps
itself alive across the context window. `.run(&config, prompt, &sink)` →
`Result<Progress, Error>` does resume-or-create, compacts **reactively** on
overflow (compact + retry once), and **proactively** after a clean turn when
the policy says so (optional checkpoint turn, then compact). `.compact(…)`
forces one. `Progress { created, compacted, telemetry }` reports what happened
— including the turn's parsed `Telemetry`, finalised at the `result` event.
- **`Claude::run(&config, &attach, prompt, &sink)`** → `Result<(), Error>` —
the low-level driver: one turn, one `Attach` target (`Resume` / `Create` /
`Continue` / `OneOff`). A clean turn is `Ok(())`; every other state is an
`Error` variant.
Supporting pieces:
- **`CompactionPolicy`** — decides *when* to compact. **`PercentPolicy`**
(`percent`, `default_window`, `checkpoint_prompt`) compacts at a percent of
the model window; **`NeverCompact`** never does.
- **`Config`** — the invocation (model, effort, cwd, prompt/MCP files, tools,
extra args).
- **`Sink`** — a trait with no-op defaults; implement what you care about to
observe stream events, non-JSON stdout, and stderr. `NoopSink` ignores all.
- **`SessionStore`** — locate and archive on-disk sessions by title.
- **`Telemetry`** (`context`, `cost`, `context_window`, `model`) — everything
the driver parses from a turn's stream, returned in `Progress`. **`Usage`** is
the minimal slice (`context_tokens`, `context_window`) the policy sees.
`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`.**

346
hive-claude/src/classify.rs
View file

@ -1,346 +0,0 @@
//! 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.
//!
//! Stdout events are parsed into a typed [`StreamEvent`] (a serde-tagged enum)
//! rather than walked as a `serde_json::Value` — the event-type dispatch and
//! the terminal result event's fields (`is_error`, `api_error_status`) are
//! typed, so a shape drift is a clean fallback instead of a silent
//! mis-detection. The marker *strings* still gate detection (they're the
//! API's error text); typing removes the fragile field-probing around them.
use std::sync::atomic::{AtomicBool, Ordering};
use serde::Deserialize;
/// 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 session is unauthenticated — either the API
/// rejected the request (401, an expired/revoked OAuth session) or the CLI
/// gave up refreshing its local OAuth token before ever making a request.
/// Sourced from claude-code's `api_retry` JSON events and its human-readable
/// give-up lines (two distinct give-up messages: a rejected request, and a
/// failed local token refresh).
const AUTH_FAIL_MARKERS: [&str; 4] = [
"\"error\":\"authentication_failed\"",
"\"error_status\":401",
"Failed to authenticate. API Error: 401",
"Failed to authenticate: OAuth session expired and could not be refreshed",
];
/// 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",
];
/// The claude-code stdout stream-json events we classify, internally tagged
/// on `type`. Typing the dispatch (and the `result` event's fields below)
/// replaces walking a `serde_json::Value` with `get("field").and_then(...)`,
/// so a field-shape drift is a compile error / a clean fallback rather than a
/// silent mis-detection. Unknown event types fall into [`StreamEvent::Other`]
/// (scanned raw like any other control event).
#[derive(Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum StreamEvent {
/// Model-authored turn output — never scanned (it can quote any marker).
Assistant,
/// Model-authored tool results — never scanned.
User,
/// The terminal result event (see [`ResultEvent`]).
Result(ResultEvent),
/// A control `error` event — the only event a rate-limit marker is
/// trusted on (a model *discussing* a rate limit can't forge this type).
Error,
/// `system` and any other/unknown control event — claude-authored, so
/// scanned raw for the failure markers.
#[serde(other)]
Other,
}
/// The terminal `result` event's fields we act on. `subtype` is deliberately
/// absent: it reads `"success"` even on a hard failure, so it's not a usable
/// error signal — `is_error` is the discriminator (the CLI sets it from the
/// real outcome; the model can't forge it). On `is_error` the `result` /
/// `error` fields carry claude-code's own failure text.
#[derive(Deserialize)]
struct ResultEvent {
#[serde(default)]
is_error: bool,
/// claude-code's failure text on `is_error` (a genuine "Prompt is too
/// long" / auth message); the model's final answer on success.
#[serde(default)]
result: Option<String>,
/// Structured error payload (a string, or an object like
/// `{"type":"rate_limit_error"}`) present on some failures.
#[serde(default)]
error: Option<serde_json::Value>,
/// API status on an auth failure (e.g. `401`) — a typed signal that
/// doesn't depend on the human-readable string landing in `result`.
#[serde(default)]
api_error_status: Option<u32>,
}
/// 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 CLI-authored line — stderr, or a non-JSON stdout line — for every
/// marker. These bytes are always claude-code's own output, never model
/// conversation, so all markers are trusted.
pub(crate) fn scan_cli_line(&self, line: &str) {
self.scan_failure_markers(line);
if RATE_LIMIT_MARKERS.iter().any(|m| line.contains(m)) {
self.rate_limited.store(true, Ordering::Relaxed);
}
}
/// Scan a parsed stdout JSON event. **Skips model-authored `assistant` /
/// `user` message events**, whose serialized content can quote any marker
/// verbatim (an agent discussing this very code, say) — a false positive
/// that would otherwise trip a needless compact/retry or a spurious
/// auth/session error. Every real signal here is emitted *instead of* a
/// model turn (the API rejected the prompt, the auth failed, or `--resume`
/// missed before any inference), so it can only appear on a control event
/// (`error` / `result` / `system`) or as raw non-JSON text — never inside
/// an assistant/user message. This holds whatever exact shape claude-code
/// uses for the message, so the gate can't suppress a genuine signal.
pub(crate) fn scan_stdout_json(&self, event: &serde_json::Value, raw: &str) {
// Type the event instead of walking the `Value`. A parse failure
// (missing / non-string `type`) is rare and falls through to a raw
// scan — the safe default, matching the old "unknown type" branch.
match StreamEvent::deserialize(event) {
// Model-authored output — its serialized content can quote any
// marker verbatim (an agent discussing this very code), so it is
// never scanned. A genuine signal is emitted *instead of* a model
// turn, so it never rides an assistant/user event.
Ok(StreamEvent::Assistant | StreamEvent::User) => {}
// The terminal result event. On success its `result` field is the
// model's final answer (same trust level as an assistant message
// — must NOT scan). On a FAILED result (`is_error`) the `result` /
// `error` fields carry claude-code's OWN failure text — the real
// event (verified against captured stream-json) is
// {"type":"result","is_error":true,"subtype":"success",
// "result":"Prompt is too long","terminal_reason":"blocking_limit"}
// — note `subtype` is "success" even on a hard failure, so
// `is_error` is the discriminator (the model can't forge it).
Ok(StreamEvent::Result(r)) => {
if r.is_error {
if let Some(text) = &r.result {
self.scan_failure_markers(text);
}
if let Some(err) = &r.error {
// `to_string()` re-serializes the JSON value (a string
// value round-trips to `"...quoted..."`), but the marker
// substring still appears inside the quoted form, so the
// scan finds it either way.
self.scan_failure_markers(&err.to_string());
}
// Typed 401: an auth failure that set the status code but
// may not have put the human-readable string in `result`.
if r.api_error_status == Some(401) {
self.auth_failed.store(true, Ordering::Relaxed);
}
}
}
// A control `error` event — claude-authored end to end, and the
// only place a rate-limit marker is trusted.
Ok(StreamEvent::Error) => {
self.scan_failure_markers(raw);
if RATE_LIMIT_MARKERS.iter().any(|m| raw.contains(m)) {
self.rate_limited.store(true, Ordering::Relaxed);
}
}
// `system` / unknown control events (claude-authored) → scan raw.
Ok(StreamEvent::Other) | Err(_) => self.scan_failure_markers(raw),
}
}
/// The prompt-too-long / auth-failed / session-not-found markers. Callers
/// gate *where* this runs (see `scan_cli_line` / `scan_stdout_json`).
fn scan_failure_markers(&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);
}
}
/// 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
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::Error;
fn json(raw: &str) -> serde_json::Value {
serde_json::from_str(raw).unwrap()
}
#[test]
fn assistant_content_quoting_marker_is_ignored() {
// An agent discussing this code emits the marker verbatim in an
// assistant message — must NOT trip a sentinel.
let s = Sentinels::default();
let raw = r#"{"type":"assistant","message":{"content":[{"type":"text","text":"the CLI prints Prompt is too long on overflow"}]}}"#;
s.scan_stdout_json(&json(raw), raw);
assert!(s.soft_error().is_none());
}
#[test]
fn genuine_failure_result_is_detected() {
// Both fixtures are the real claude-code failure shape, verified against
// captured stream-json: the marker lives in the `result` field with
// `is_error: true` and — counterintuitively — `subtype: "success"`. The
// earlier unconditional `result` scrub blinded this; `is_error` is the
// discriminator that restores it (and the model can't forge it).
let s = Sentinels::default();
let raw = r#"{"type":"result","is_error":true,"subtype":"success","result":"Prompt is too long","terminal_reason":"blocking_limit","stop_reason":"stop_sequence"}"#;
s.scan_stdout_json(&json(raw), raw);
assert!(matches!(s.soft_error(), Some(Error::PromptTooLong)));
let s = Sentinels::default();
let raw = r#"{"type":"result","is_error":true,"subtype":"success","result":"Failed to authenticate. API Error: 401 Invalid authentication credentials","api_error_status":401}"#;
s.scan_stdout_json(&json(raw), raw);
assert!(matches!(s.soft_error(), Some(Error::AuthFailed)));
}
#[test]
fn typed_api_error_status_401_trips_auth_without_string_marker() {
// The typed `api_error_status` field catches an auth failure even when
// the human-readable 401 string isn't in `result` — a signal the old
// substring scan would have missed.
let s = Sentinels::default();
let raw = r#"{"type":"result","is_error":true,"subtype":"success","result":"request failed","api_error_status":401}"#;
s.scan_stdout_json(&json(raw), raw);
assert!(matches!(s.soft_error(), Some(Error::AuthFailed)));
}
#[test]
fn error_payload_object_on_result_is_scanned() {
// A failure result can carry the marker in a structured `error`
// payload rather than `result`; the typed `error: Value` is stringified
// and scanned.
let s = Sentinels::default();
let raw = r#"{"type":"result","is_error":true,"error":"Prompt is too long","result":"ok"}"#;
s.scan_stdout_json(&json(raw), raw);
assert!(matches!(s.soft_error(), Some(Error::PromptTooLong)));
}
#[test]
fn result_field_model_text_quoting_marker_is_ignored() {
// The terminal `result` event's `result` field is the model's final
// answer. A marker quoted there (the model discussing this very code,
// or echoing an API error string) must NOT trip a sentinel — that was
// a turn-kill DoS. Covers prompt-too-long, auth, and session markers.
for marker in [
"Prompt is too long",
"Failed to authenticate. API Error: 401",
"does not match any session title",
] {
let s = Sentinels::default();
let raw = format!(
r#"{{"type":"result","subtype":"success","is_error":false,"result":"the harness scans stdout for {marker} — see classify.rs"}}"#
);
s.scan_stdout_json(&json(&raw), &raw);
assert!(
s.soft_error().is_none(),
"marker {marker:?} in the model-authored result field must be ignored"
);
}
}
#[test]
fn result_control_field_still_trips_even_with_clean_result_text() {
// Scrubbing `result` must not blind us to a genuine signal in a
// control field of the same event.
let s = Sentinels::default();
let raw = r#"{"type":"result","subtype":"error","is_error":true,"error":"Failed to authenticate. API Error: 401","result":"ok"}"#;
s.scan_stdout_json(&json(raw), raw);
assert!(matches!(s.soft_error(), Some(Error::AuthFailed)));
}
#[test]
fn raw_non_json_marker_is_detected() {
let s = Sentinels::default();
s.scan_cli_line("API Error: Prompt is too long");
assert!(matches!(s.soft_error(), Some(Error::PromptTooLong)));
}
#[test]
fn oauth_refresh_failure_trips_auth_failed() {
// A local OAuth token-refresh failure never reaches an API request —
// no `error_status`/401 anywhere — so it needs its own marker rather
// than riding the request-rejected 401 text. Observed in the wild as
// a turn that failed with a bare non-zero exit and no recognized
// sentinel: the harness never transitioned to needs-login because
// this exact give-up line wasn't a marker yet.
let s = Sentinels::default();
s.scan_cli_line("Failed to authenticate: OAuth session expired and could not be refreshed");
assert!(matches!(s.soft_error(), Some(Error::AuthFailed)));
}
#[test]
fn rate_limit_still_only_on_error_event() {
// A non-error control event mentioning the marker must not trip it.
let s = Sentinels::default();
let raw = r#"{"type":"result","summary":"we hit a rate_limit_error earlier"}"#;
s.scan_stdout_json(&json(raw), raw);
assert!(s.soft_error().is_none());
// A genuine error event does.
let raw2 = r#"{"type":"error","error":{"type":"rate_limit_error"}}"#;
s.scan_stdout_json(&json(raw2), raw2);
assert!(matches!(s.soft_error(), Some(Error::RateLimited)));
}
}

67
hive-claude/src/config.rs
View file

@ -1,67 +0,0 @@
//! Invocation config: how to build one `claude --print` command line.
use std::path::PathBuf;
use std::time::Duration;
/// How a run attaches to a claude session — the low-level session flag. Kept
/// separate from [`Config`] so one config can drive resume + create +
/// `/compact` of the same logical session. For a self-managing durable
/// session, prefer [`crate::InfiniteSession`] over hand-picking an `Attach`.
#[derive(Debug, Clone)]
pub enum Attach {
/// `--resume <id-or-title>` — resume an existing session by UUID or by the
/// display title set via [`Attach::Create`]. Yields
/// [`crate::Error::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 [`Attach`] target (passed separately to [`crate::Claude::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>,
/// Idle watchdog: kill the child and return [`crate::Error::IdleTimeout`]
/// if no stdout line arrives for this long. The timer resets on every
/// stdout line, so a large/slow but still-streaming turn is never cut;
/// only complete output silence trips it. `None` waits indefinitely.
/// The driver stays policy-free — the caller decides the window (and
/// whether to read it from the environment).
pub idle_timeout: Option<Duration>,
}

241
hive-claude/src/driver.rs
View file

@ -1,241 +0,0 @@
//! The subprocess driver: spawn claude, pump + classify its streams, and
//! assemble the result.
use std::collections::VecDeque;
use std::process::{ExitStatus, Stdio};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, ChildStderr, ChildStdout, Command};
use crate::classify::Sentinels;
use crate::{Attach, Config, Error, Result, Sink};
/// Default program name spawned when [`Config::program`] is unset.
const DEFAULT_PROGRAM: &str = "claude";
/// How many trailing stderr lines to keep for [`Error::Exit`].
const STDERR_TAIL_LINES: usize = 20;
/// Idle-watchdog probe cadence: re-check output silence this often while the
/// child runs. Fine-grained enough to fire within ~one probe of the deadline,
/// cheap enough to ignore.
const IDLE_PROBE: Duration = Duration::from_secs(5);
/// The low-level driver entry point. A namespace for the run function — there
/// is nothing to construct; call `Claude::run(…)` directly. For a durable,
/// self-compacting session, use [`crate::InfiniteSession`] instead.
pub struct Claude;
impl Claude {
/// Spawn one headless `claude --print` turn, stream its output through
/// `sink`, and report the result.
///
/// The wake prompt is written to claude's stdin. stdout (`stream-json`)
/// and stderr are pumped concurrently while the child runs. A clean turn
/// returns `Ok(())`; every non-completion state — recognized sentinel or
/// hard failure — is an [`Error`] variant, so callers branch with a single
/// `match`.
///
/// # Errors
///
/// - A recognized sentinel: [`Error::PromptTooLong`], [`Error::RateLimited`],
/// [`Error::AuthFailed`], [`Error::SessionNotFound`].
/// - [`Error::Spawn`] if the binary can't be launched.
/// - [`Error::Stdin`] / [`Error::Wait`] on stdin-write / child-wait failure.
/// - [`Error::IdleTimeout`] if `config.idle_timeout` is set and no stdout
/// line arrives within that window (the child is killed).
/// - [`Error::Exit`] on a non-zero exit that raised no sentinel.
pub async fn run(
config: &Config,
attach: &Attach,
prompt: &str,
sink: &impl Sink,
) -> Result<()> {
let program = config.program.as_deref().unwrap_or(DEFAULT_PROGRAM);
let mut cmd = build_command(program, config, attach);
let mut child = cmd
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.map_err(|source| Error::Spawn {
program: program.to_string(),
source,
})?;
if let Some(mut stdin) = child.stdin.take() {
stdin
.write_all(prompt.as_bytes())
.await
.map_err(Error::Stdin)?;
// Best-effort flush/close; claude sees EOF and starts the turn.
stdin.shutdown().await.ok();
}
let stdout = child.stdout.take().expect("stdout piped");
let stderr = child.stderr.take().expect("stderr piped");
let sentinels = Sentinels::default();
// Idle watchdog clock: `last_activity` holds seconds-since-`base` of the
// last stdout line, bumped by the pump; the waiter reads it to detect a
// fully silent stall. Monotonic (`Instant`) so a wall-clock jump can't
// spuriously fire it.
let base = Instant::now();
let last_activity = AtomicU64::new(0);
// Pump both streams and wait for exit concurrently on this task — no
// `spawn`, so the sink needn't be `'static` and borrows stay simple.
let ((), stderr_tail, (status, timed_out)) = tokio::join!(
pump_stdout(stdout, sink, &sentinels, base, &last_activity),
pump_stderr(stderr, sink, &sentinels),
wait_with_idle(&mut child, base, &last_activity, config.idle_timeout),
);
let status = status.map_err(Error::Wait)?;
// A recognized sentinel takes precedence over everything (most specific
// reason). Then an idle-watchdog kill; then a plain non-zero exit.
if let Some(sentinel) = sentinels.soft_error() {
return Err(sentinel);
}
if timed_out {
return Err(Error::IdleTimeout);
}
if !status.success() {
return Err(Error::Exit {
status,
stderr_tail,
});
}
Ok(())
}
}
/// Wait for the child to exit, enforcing the optional idle watchdog. With no
/// `idle_timeout` this is a plain `child.wait()`. Otherwise it re-checks on a
/// fixed probe cadence: if no stdout line arrived within `idle_timeout` (per
/// `last_activity`, bumped by [`pump_stdout`]), it kills the child and reaps
/// it. Returns the exit status and whether the watchdog fired.
async fn wait_with_idle(
child: &mut Child,
base: Instant,
last_activity: &AtomicU64,
idle_timeout: Option<Duration>,
) -> (std::io::Result<ExitStatus>, bool) {
let Some(window) = idle_timeout else {
return (child.wait().await, false);
};
// `child.wait()` is cancel-safe, so dropping it on a probe timeout doesn't
// lose the exit.
loop {
match tokio::time::timeout(IDLE_PROBE, child.wait()).await {
Ok(status) => return (status, false),
Err(_probe_expired) => {
let last = Duration::from_secs(last_activity.load(Ordering::Relaxed));
if base.elapsed().saturating_sub(last) >= window {
let _ = child.kill().await;
return (child.wait().await, true);
}
}
}
}
}
/// Assemble the argv. `--print --verbose --output-format stream-json` are
/// mandatory (the driver parses that shape); everything else is gated on the
/// [`Config`] / [`Attach`].
fn build_command(program: &str, config: &Config, attach: &Attach) -> Command {
let mut cmd = Command::new(program);
if let Some(cwd) = &config.cwd {
cmd.current_dir(cwd);
}
cmd.arg("--print")
.arg("--verbose")
.arg("--output-format")
.arg("stream-json");
if !config.model.is_empty() {
cmd.arg("--model").arg(&config.model);
}
if let Some(effort) = &config.effort {
cmd.arg("--effort").arg(effort);
}
match attach {
Attach::Resume(id) => {
cmd.arg("--resume").arg(id);
}
Attach::Create(title) => {
cmd.arg("--name").arg(title);
}
Attach::Continue => {
cmd.arg("--continue");
}
Attach::OneOff => {}
}
if let Some(path) = &config.system_prompt_file {
cmd.arg("--system-prompt-file").arg(path);
}
if let Some(path) = &config.mcp_config {
cmd.arg("--mcp-config").arg(path);
}
if config.strict_mcp_config {
cmd.arg("--strict-mcp-config");
}
if let Some(tools) = &config.tools {
cmd.arg("--tools").arg(tools);
}
if let Some(allowed) = &config.allowed_tools {
cmd.arg("--allowedTools").arg(allowed);
}
for dir in &config.add_dirs {
cmd.arg("--add-dir").arg(dir);
}
for extra in &config.extra_args {
cmd.arg(extra);
}
cmd
}
/// Read stdout line by line: classify each line, parse JSON, hand events (or
/// raw non-JSON lines) to the sink.
async fn pump_stdout(
stdout: ChildStdout,
sink: &impl Sink,
sentinels: &Sentinels,
base: Instant,
last_activity: &AtomicU64,
) {
let mut lines = BufReader::new(stdout).lines();
while let Ok(Some(line)) = lines.next_line().await {
// Poke the idle watchdog: any stdout line resets the silence timer.
// Seconds granularity is plenty — the probe cadence is coarser still.
last_activity.store(base.elapsed().as_secs(), Ordering::Relaxed);
if let Ok(event) = serde_json::from_str::<serde_json::Value>(&line) {
// JSON stdout: classify with the model-content gate so an
// `assistant`/`user` message quoting a marker can't trip it.
sentinels.scan_stdout_json(&event, &line);
sink.on_event(&event);
} else {
// Non-JSON stdout is CLI text, not conversation — trust all markers.
sentinels.scan_cli_line(&line);
sink.on_stdout_line(&line);
}
}
}
/// Read stderr line by line: classify, forward to the sink, and retain the
/// last [`STDERR_TAIL_LINES`] for a possible [`Error::Exit`]. Returns the
/// newline-joined tail.
async fn pump_stderr(stderr: ChildStderr, sink: &impl Sink, sentinels: &Sentinels) -> String {
let mut lines = BufReader::new(stderr).lines();
let mut tail: VecDeque<String> = VecDeque::with_capacity(STDERR_TAIL_LINES);
while let Ok(Some(line)) = lines.next_line().await {
// stderr is always CLI output — trust all markers.
sentinels.scan_cli_line(&line);
sink.on_stderr_line(&line);
if tail.len() >= STDERR_TAIL_LINES {
tail.pop_front();
}
tail.push_back(line);
}
tail.into_iter().collect::<Vec<_>>().join("\n")
}

85
hive-claude/src/error.rs
View file

@ -1,85 +0,0 @@
//! 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),
/// The child produced no stdout for longer than the configured idle
/// window (`Config::idle_timeout`) and was killed. Indicative of an
/// Anthropic API stall (e.g. a multi-retry connection storm that goes
/// silent for minutes). Callers typically park briefly and retry, like
/// the rate-limit path.
#[error("claude idle timeout: no output for the configured window")]
IdleTimeout,
/// 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>;

77
hive-claude/src/lib.rs
View file

@ -1,77 +0,0 @@
//! `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, hard-coded
//! watermarks, or logging. Callers wire streaming output through a [`Sink`].
//!
//! Two layers:
//!
//! - [`Claude::run`] — the low-level driver: one turn, one [`Attach`] target.
//! - [`InfiniteSession`] — a durable session (name + [`SessionStore`] +
//! [`CompactionPolicy`]) that keeps itself alive across the context window by
//! compacting reactively (on overflow) and proactively (per policy — e.g.
//! [`PercentPolicy`]). This is the one you usually want.
//!
//! # `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::{Attach, Claude, Config, Error, NoopSink};
//!
//! let config = Config {
//! model: "haiku".into(),
//! ..Default::default()
//! };
//! match Claude::run(&config, &Attach::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}"),
//! }
//! # }
//! ```
// The crate-level `//!` overview above is the rustdoc entry point — it belongs
// in source, not docs/, so the comment-block length lint should not flag it.
// lint:allow-long-comment
mod classify;
mod config;
mod driver;
mod error;
mod policy;
mod session;
mod sink;
mod store;
mod telemetry;
pub use config::{Attach, Config};
pub use driver::Claude;
pub use error::{Error, Result};
pub use policy::{CompactionPolicy, NeverCompact, PercentPolicy};
pub use session::{InfiniteSession, Progress};
pub use sink::{NoopSink, Sink};
pub use store::SessionStore;
pub use telemetry::{Telemetry, TokenUsage, Usage};

134
hive-claude/src/policy.rs
View file

@ -1,134 +0,0 @@
//! When a durable session should compact.
use crate::Usage;
/// Decides, after a completed turn, whether an [`crate::InfiniteSession`]
/// should proactively compact — and what to say in the optional checkpoint
/// turn that runs first. Injected by the caller so the driver stays free of
/// any app-specific policy.
pub trait CompactionPolicy {
/// Given the last turn's context [`Usage`], compact now (before the window
/// fills)?
fn should_compact(&self, usage: Usage) -> bool;
/// Prompt for a pre-compaction checkpoint turn (a chance for the agent to
/// flush durable state before detail collapses into a summary), or `None`
/// to compact without one. Default: `None`.
fn checkpoint_prompt(&self) -> Option<&str> {
None
}
}
/// Compact once the context reaches `percent` of the model window.
///
/// The window is the one the model reported this turn ([`Usage::context_window`]),
/// falling back to [`PercentPolicy::default_window`] when the turn reported
/// none (e.g. a degenerate turn with no `result` usage). `percent == 0`
/// disables proactive compaction entirely.
#[derive(Debug, Clone, Default)]
pub struct PercentPolicy {
/// Watermark as a percent of the context window (e.g. `75`). `0` disables.
pub percent: u8,
/// Window to assume when the turn didn't report one. `None` → never
/// compact until a window is observed.
pub default_window: Option<u64>,
/// Prompt for the pre-compaction checkpoint turn; `None` skips it.
pub checkpoint_prompt: Option<String>,
}
impl CompactionPolicy for PercentPolicy {
fn should_compact(&self, usage: Usage) -> bool {
if self.percent == 0 {
return false;
}
let Some(window) = usage
.context_window
.or(self.default_window)
.filter(|&w| w > 0)
else {
return false;
};
usage.context_tokens.saturating_mul(100) >= u64::from(self.percent) * window
}
fn checkpoint_prompt(&self) -> Option<&str> {
self.checkpoint_prompt.as_deref()
}
}
/// A policy that never compacts. Turns run until the session overflows and the
/// reactive path in [`crate::InfiniteSession::run`] takes over.
#[derive(Debug, Clone, Copy, Default)]
pub struct NeverCompact;
impl CompactionPolicy for NeverCompact {
fn should_compact(&self, _usage: Usage) -> bool {
false
}
}
#[cfg(test)]
mod tests {
use super::{CompactionPolicy, NeverCompact, PercentPolicy};
use crate::Usage;
fn policy(percent: u8, default_window: Option<u64>) -> PercentPolicy {
PercentPolicy {
percent,
default_window,
checkpoint_prompt: None,
}
}
#[test]
fn fires_at_or_above_watermark() {
let p = policy(75, None);
// 75% of a 200k window = 150k.
assert!(!p.should_compact(Usage {
context_tokens: 149_999,
context_window: Some(200_000),
}));
assert!(p.should_compact(Usage {
context_tokens: 150_000,
context_window: Some(200_000),
}));
}
#[test]
fn zero_percent_disables() {
assert!(!policy(0, Some(200_000)).should_compact(Usage {
context_tokens: 199_999,
context_window: Some(200_000),
}));
}
#[test]
fn falls_back_to_default_window_when_unreported() {
let p = policy(50, Some(100_000));
assert!(p.should_compact(Usage {
context_tokens: 50_000,
context_window: None,
}));
// Reported window takes precedence over the default.
assert!(!p.should_compact(Usage {
context_tokens: 50_000,
context_window: Some(200_000),
}));
}
#[test]
fn no_window_anywhere_never_fires() {
assert!(!policy(75, None).should_compact(Usage {
context_tokens: u64::MAX,
context_window: None,
}));
}
#[test]
fn never_compact_is_never() {
assert!(!NeverCompact.should_compact(Usage {
context_tokens: u64::MAX,
context_window: Some(1),
}));
}
}

225
hive-claude/src/session.rs
View file

@ -1,225 +0,0 @@
//! A durable, self-compacting ("infinite") claude session.
use std::sync::Mutex;
use serde_json::Value;
use crate::{
Attach, Claude, CompactionPolicy, Config, Error, Result, SessionStore, Sink, Telemetry,
};
/// A named claude session that outlives the model's context window by
/// compacting itself. Bundles the three things a durable session needs:
///
/// - a **name** (the constant session title it resumes / creates under),
/// - a **store** ([`SessionStore`], to find the backing file so a resume vs.
/// create is decided without a wasted spawn), and
/// - a **policy** ([`CompactionPolicy`], deciding *when* to compact).
///
/// [`InfiniteSession::run`] keeps the session alive across turns:
///
/// - **resume-or-create** — resumes the titled session, creating it on first
/// use (or after its file was archived away);
/// - **reactive** — if a turn overflows ([`Error::PromptTooLong`]), it compacts
/// and retries the same prompt once;
/// - **proactive** — after a clean turn it consults the policy and, if due,
/// runs an optional checkpoint turn then compacts.
///
/// Resetting/archiving the session is intentionally *not* part of this type —
/// that stays with the caller.
pub struct InfiniteSession<P: CompactionPolicy> {
name: String,
store: SessionStore,
policy: P,
}
/// What [`InfiniteSession::run`] did, beyond streaming the turn to the sink.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Progress {
/// The turn resumed no existing session — a fresh one was created.
pub created: bool,
/// A compaction ran (reactively on overflow, or proactively per policy).
pub compacted: bool,
/// Everything parsed from the answering turn's stream (usage, cost,
/// context window, resolved model). The authoritative copy — finalised at
/// the turn's `result` event.
pub telemetry: Telemetry,
}
impl<P: CompactionPolicy> InfiniteSession<P> {
/// Build a durable session for `name`, backed by `store`, governed by
/// `policy`.
pub fn new(name: impl Into<String>, store: SessionStore, policy: P) -> Self {
Self {
name: name.into(),
store,
policy,
}
}
/// Run one turn, keeping the session infinite (see the type docs).
///
/// # Errors
///
/// Propagates any non-`SessionNotFound` [`Error`] from the underlying
/// runs. A reactive retry that *still* overflows surfaces as
/// [`Error::PromptTooLong`]; rate-limit / auth / hard failures propagate
/// unchanged for the caller to handle.
pub async fn run(&self, config: &Config, prompt: &str, sink: &impl Sink) -> Result<Progress> {
// Resolve resume-vs-create once, up front, so `created` reflects the
// session state at the START of the run: the reactive-retry path can't
// read it from the retry (the session exists by then).
let resolved = self.store.find_by_title(&self.name);
let existed = resolved.is_some();
tracing::info!(
title = %self.name,
existed,
path = resolved.as_deref().map(|p| p.display().to_string()).unwrap_or_default(),
"InfiniteSession::run: resolved session before attach"
);
let meter = TelemetrySink::new(sink);
let created = match self.attempt(config, prompt, &meter, existed).await {
Ok(created) => created,
Err(Error::PromptTooLong) => {
// The session is already past the window — no turn can run on
// it and the detail is gone (no checkpoint possible). Compact,
// then retry the same prompt once; the retry is the answering
// turn, so its telemetry is what we report. The failed attempt
// created/resumed the session, so the retry resumes it, and
// `created` still reflects the pre-run state.
self.compact(config, sink).await?;
let retry = TelemetrySink::new(sink);
self.attempt(config, prompt, &retry, true).await?;
return Ok(Progress {
created: !existed,
compacted: true,
telemetry: retry.snapshot(),
});
}
Err(other) => return Err(other),
};
let telemetry = meter.snapshot();
// Proactive: the turn completed on a healthy session. If the policy
// says it's due, checkpoint (best-effort) then compact.
if self.policy.should_compact(telemetry.usage()) {
if let Some(checkpoint) = self.policy.checkpoint_prompt() {
let _ = self.attempt(config, checkpoint, sink, true).await;
}
// Only claim a compaction if it actually succeeded — the flag feeds
// stats + the auto-reset watermark, so a failed best-effort
// `/compact` must not report that the context shrank.
let compacted = self.compact(config, sink).await.is_ok();
return Ok(Progress {
created,
compacted,
telemetry,
});
}
Ok(Progress {
created,
compacted: false,
telemetry,
})
}
/// Force a `/compact` on the session (e.g. operator-driven). Resume-only:
/// if the session doesn't exist there is nothing to compact, so a
/// [`Error::SessionNotFound`] is swallowed as a no-op `Ok`.
///
/// # Errors
///
/// Propagates any error other than `SessionNotFound` from the compact run.
pub async fn compact(&self, config: &Config, sink: &impl Sink) -> Result<()> {
let resolved = self.store.find_by_title(&self.name);
tracing::info!(
title = %self.name,
path = resolved.as_deref().map(|p| p.display().to_string()).unwrap_or_default(),
"InfiniteSession::compact: attaching /compact to this session"
);
let result =
Claude::run(config, &Attach::Resume(self.name.clone()), "/compact", sink).await;
match result {
Err(Error::SessionNotFound) => {
tracing::info!(title = %self.name, "InfiniteSession::compact: session not found, no-op");
Ok(())
}
other => other,
}
}
/// One resume-or-create turn. `existed` is the caller's up-front
/// resume-vs-create decision (whether the titled session was on disk before
/// the run) — passing it in rather than re-checking keeps `created`
/// reporting consistent across the reactive-retry path. Still self-heals if
/// the backing file vanished between the check and the run. Returns whether
/// a fresh session was created.
async fn attempt(
&self,
config: &Config,
prompt: &str,
sink: &impl Sink,
existed: bool,
) -> Result<bool> {
let attach = if existed {
Attach::Resume(self.name.clone())
} else {
Attach::Create(self.name.clone())
};
tracing::info!(
title = %self.name,
mode = if existed { "resume" } else { "create" },
"InfiniteSession::attempt: attaching turn"
);
match Claude::run(config, &attach, prompt, sink).await {
Ok(()) => Ok(!existed),
// We thought it existed but the resume missed (raced an archive) —
// self-heal by creating.
Err(Error::SessionNotFound) if existed => {
tracing::warn!(
title = %self.name,
"InfiniteSession::attempt: resume missed (raced an archive?) — self-healing via create"
);
Claude::run(config, &Attach::Create(self.name.clone()), prompt, sink).await?;
Ok(true)
}
Err(other) => Err(other),
}
}
}
/// A [`Sink`] that forwards to an inner sink while accumulating the turn's
/// [`Telemetry`] from the stream. Cheap; the driver calls it synchronously
/// from one reader task, so the `Mutex` only satisfies `&self`.
struct TelemetrySink<'a, S: Sink> {
inner: &'a S,
telemetry: Mutex<Telemetry>,
}
impl<'a, S: Sink> TelemetrySink<'a, S> {
fn new(inner: &'a S) -> Self {
Self {
inner,
telemetry: Mutex::new(Telemetry::default()),
}
}
fn snapshot(&self) -> Telemetry {
self.telemetry.lock().unwrap().clone()
}
}
impl<S: Sink> Sink for TelemetrySink<'_, S> {
fn on_event(&self, event: &Value) {
self.telemetry.lock().unwrap().observe(event);
self.inner.on_event(event);
}
fn on_stdout_line(&self, line: &str) {
self.inner.on_stdout_line(line);
}
fn on_stderr_line(&self, line: &str) {
self.inner.on_stderr_line(line);
}
}

30
hive-claude/src/sink.rs
View file

@ -1,30 +0,0 @@
//! 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 {}

203
hive-claude/src/store.rs
View file

@ -1,203 +0,0 @@
//! 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, 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.
///
/// Unlike the old first-match-wins scan, this keeps going past the first
/// hit so a *second* same-titled file — which would otherwise resolve to
/// whichever one `read_dir` happens to yield first, i.e. filesystem-order
/// luck — logs a `tracing::warn!` instead of silently picking one. The
/// first match found is still what's returned (unchanged resolution
/// behaviour); this is a diagnostic for the "wrong session resumed"
/// report, not a fix.
#[must_use]
pub fn find_by_title(&self, title: &str) -> Option<PathBuf> {
let mut found: Option<PathBuf> = None;
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;
};
let matches = std::io::BufReader::new(file)
.lines()
.map_while(std::result::Result::ok)
.any(|line| line_sets_title(&line, title));
if !matches {
continue;
}
match &found {
None => found = Some(path),
Some(first) => {
tracing::warn!(
title,
first = %first.display(),
also = %path.display(),
"find_by_title: multiple session files share this title — \
resuming the first one found (readdir order, not deterministic)"
);
}
}
}
found
}
/// 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 {
tracing::info!(
title,
"archive_by_title: no session file found for this title"
);
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)?;
tracing::info!(
title,
from = %path.display(),
to = %target.display(),
"archive_by_title: renamed session file"
);
Ok(Some(target))
}
}
/// True if `line` is the session's `custom-title` event whose **top-level**
/// `customTitle` field equals `title`.
///
/// Parsing the line (rather than substring-matching the whole transcript) is
/// what makes this robust: a message that merely *quotes* the marker in its
/// content has no top-level `customTitle` key, so it can't cause a false match
/// on the wrong session file; the compact-vs-spaced JSON form is irrelevant;
/// and titles containing `"` / `\` are handled by the parser. Verified shape
/// (claude 2.1.x): `{"type":"custom-title","customTitle":"<title>",…}`.
///
/// The `contains` pre-check keeps the common case cheap — only the rare line
/// mentioning `customTitle` is parsed as JSON, not every transcript line.
fn line_sets_title(line: &str, title: &str) -> bool {
if !line.contains("customTitle") {
return false;
}
let Ok(value) = serde_json::from_str::<serde_json::Value>(line) else {
return false;
};
value.get("customTitle").and_then(|t| t.as_str()) == Some(title)
}
#[cfg(test)]
mod tests {
use super::*;
fn write(dir: &std::path::Path, name: &str, lines: &[&str]) {
std::fs::write(dir.join(name), lines.join("\n")).unwrap();
}
#[test]
fn find_by_title_matches_only_the_custom_title_event() {
let home = tempfile::tempdir().unwrap();
let cwd = std::path::Path::new("/agents/iris/state");
let store = SessionStore::new(home.path(), cwd);
let proj = store.project_dir();
std::fs::create_dir_all(&proj).unwrap();
// The real titled session.
write(
&proj,
"real.jsonl",
&[
r#"{"type":"summary","summary":"x"}"#,
r#"{"type":"custom-title","customTitle":"iris","sessionId":"real"}"#,
],
);
// A DIFFERENT session whose transcript merely quotes the marker string
// in message content — must NOT match.
write(
&proj,
"other.jsonl",
&[
r#"{"type":"custom-title","customTitle":"someone-else","sessionId":"other"}"#,
r#"{"type":"assistant","message":{"content":[{"type":"text","text":"the file had \"customTitle\":\"iris\" in it"}]}}"#,
],
);
assert_eq!(store.find_by_title("iris"), Some(proj.join("real.jsonl")));
assert_eq!(store.find_by_title("nobody"), None);
}
#[test]
fn find_by_title_tolerates_spaced_json_and_escapes() {
let home = tempfile::tempdir().unwrap();
let store = SessionStore::new(home.path(), std::path::Path::new("/x"));
let proj = store.project_dir();
std::fs::create_dir_all(&proj).unwrap();
// Spaced JSON form + a title needing JSON escaping.
write(
&proj,
"s.jsonl",
&[r#"{ "type": "custom-title", "customTitle": "a\"b" }"#],
);
assert_eq!(store.find_by_title("a\"b"), Some(proj.join("s.jsonl")));
}
}

182
hive-claude/src/telemetry.rs
View file

@ -1,182 +0,0 @@
//! What the driver parses out of a turn's stream-json output.
//!
//! The lib is the single source of truth for reading claude's usage/model
//! reporting. A turn's [`Telemetry`] is accumulated by the driver as events
//! stream and handed back from [`crate::InfiniteSession::run`]; consumers that
//! also want the raw events (for their own SSE / tool-call accounting) still
//! get them through their [`crate::Sink`].
use serde::{Deserialize, Serialize};
use serde_json::Value;
/// Token counts from one `usage` block. All in tokens; missing fields read `0`.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct TokenUsage {
pub input_tokens: u64,
pub output_tokens: u64,
pub cache_read_input_tokens: u64,
pub cache_creation_input_tokens: u64,
}
impl TokenUsage {
/// Context footprint counting against the model window: input + both cache
/// classes (not output).
#[must_use]
pub fn context_tokens(&self) -> u64 {
self.input_tokens + self.cache_read_input_tokens + self.cache_creation_input_tokens
}
fn from_obj(u: &Value) -> Self {
let field = |k: &str| u.get(k).and_then(Value::as_u64).unwrap_or(0);
Self {
input_tokens: field("input_tokens"),
output_tokens: field("output_tokens"),
cache_read_input_tokens: field("cache_read_input_tokens"),
cache_creation_input_tokens: field("cache_creation_input_tokens"),
}
}
}
/// Everything the driver tracks from one turn's stream-json output.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Telemetry {
/// Most recent per-inference usage (from `assistant` events) — the live
/// context footprint (the number to watch for compaction).
pub context: TokenUsage,
/// Cumulative usage across the turn (from the terminal `result` event) —
/// the cost signal. Sums per-call prompts and can exceed the window.
pub cost: TokenUsage,
/// Model-reported active context window (`modelUsage.*.contextWindow` on
/// the `result` event), if reported.
pub context_window: Option<u64>,
/// Resolved model id echoed by the API (`assistant.message.model`, e.g.
/// `claude-opus-4-8`) — the concrete version, not the requested alias.
pub model: Option<String>,
}
impl Telemetry {
/// Fold one stream-json event into the running telemetry.
pub(crate) fn observe(&mut self, event: &Value) {
match event.get("type").and_then(Value::as_str) {
Some("assistant") => {
let Some(message) = event.get("message") else {
return;
};
if let Some(usage) = message.get("usage") {
self.context = TokenUsage::from_obj(usage);
}
if let Some(model) = message
.get("model")
.and_then(Value::as_str)
.filter(|s| !s.is_empty())
{
self.model = Some(model.to_string());
}
}
Some("result") => {
if let Some(usage) = event.get("usage") {
self.cost = TokenUsage::from_obj(usage);
}
if let Some(window) = context_window_from_result(event) {
self.context_window = Some(window);
}
}
_ => {}
}
}
/// The minimal signal a [`crate::CompactionPolicy`] needs.
#[must_use]
pub fn usage(&self) -> Usage {
Usage {
context_tokens: self.context.context_tokens(),
context_window: self.context_window,
}
}
}
/// First non-zero `contextWindow` across the `result` event's `modelUsage` map.
fn context_window_from_result(event: &Value) -> Option<u64> {
for (_model, stats) in event.get("modelUsage")?.as_object()? {
if let Some(w) = stats.get("contextWindow").and_then(Value::as_u64)
&& w > 0
{
return Some(w);
}
}
None
}
/// The compaction signal: the live context size and the window it's measured
/// against. Derived from [`Telemetry`] via [`Telemetry::usage`].
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Usage {
/// Tokens in the last inference's context. `0` until the first `assistant`
/// event.
pub context_tokens: u64,
/// The model-reported active window, if the turn reported one.
pub context_window: Option<u64>,
}
#[cfg(test)]
mod tests {
use super::Telemetry;
use serde_json::json;
#[test]
fn context_tokens_sum_excludes_output() {
let mut t = Telemetry::default();
t.observe(&json!({
"type": "assistant",
"message": { "model": "claude-opus-4-8", "usage": {
"input_tokens": 100, "output_tokens": 999,
"cache_read_input_tokens": 20, "cache_creation_input_tokens": 5,
}}
}));
assert_eq!(t.context.context_tokens(), 125);
assert_eq!(t.model.as_deref(), Some("claude-opus-4-8"));
}
#[test]
fn last_assistant_wins() {
let mut t = Telemetry::default();
for n in [10, 20, 30] {
t.observe(&json!({
"type": "assistant",
"message": { "usage": { "input_tokens": n } }
}));
}
assert_eq!(t.context.input_tokens, 30);
}
#[test]
fn result_event_sets_cost_and_window() {
let mut t = Telemetry::default();
t.observe(&json!({
"type": "result",
"usage": { "input_tokens": 5_000, "output_tokens": 1_000 },
"modelUsage": { "claude-opus-4-8": { "contextWindow": 200_000 } }
}));
assert_eq!(t.cost.input_tokens, 5_000);
assert_eq!(t.context_window, Some(200_000));
}
#[test]
fn empty_model_ignored_and_non_events_no_op() {
let mut t = Telemetry::default();
t.observe(&json!({ "type": "assistant", "message": { "model": "" } }));
assert_eq!(t.model, None);
t.observe(&json!({ "type": "system", "subtype": "init" }));
assert_eq!(t, Telemetry::default());
}
#[test]
fn usage_view_derives_from_context_and_window() {
let mut t = Telemetry::default();
t.observe(&json!({ "type": "assistant", "message": { "usage": { "input_tokens": 42 } } }));
t.observe(&json!({ "type": "result", "modelUsage": { "m": { "contextWindow": 100 } } }));
let u = t.usage();
assert_eq!(u.context_tokens, 42);
assert_eq!(u.context_window, Some(100));
}
}