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

refactor(agent): extract claude driver into hive-claude crate

Browse source
This commit is contained in:
müde 2026-07-05 18:50:12 +02:00
commit a3b66241d1
13 changed files with 907 additions and 442 deletions
Download patch file Download diff file Expand all files Collapse all files

630
hive-ag3nt/src/turn.rs
View file

@ -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 <title>` 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 {