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

fix(agent): pin claude session by constant title, archive on reset

Browse source
This commit is contained in:
müde 2026-07-05 18:13:54 +02:00
commit e35acca814
4 changed files with 369 additions and 164 deletions
Download patch file Download diff file Expand all files Collapse all files

78
docs/turn-loop.md
View file

@ -137,7 +137,7 @@ if a new inbox message arrives before this turn ends"). The
``` ```
claude --print --verbose --output-format stream-json --model <name> \ claude --print --verbose --output-format stream-json --model <name> \
--effort <level> --continue \ --effort <level> --resume <title> # or --name <title> on first use \
--system-prompt-file /run/hive/claude-system-prompt.md \ --system-prompt-file /run/hive/claude-system-prompt.md \
--mcp-config /run/hive/claude-mcp-config.json --strict-mcp-config \ --mcp-config /run/hive/claude-mcp-config.json --strict-mcp-config \
--tools <builtins> --allowedTools <builtins+mcp> --tools <builtins> --allowedTools <builtins+mcp>
@ -180,16 +180,34 @@ The effective window drives watermarks and is exposed at runtime
via `/api/state.context_window_tokens` so the UI can show a via `/api/state.context_window_tokens` so the UI can show a
percentage-of-window ctx badge. percentage-of-window ctx badge.
`--continue` keeps a persistent session per agent (claude stores **Session identity — a constant title.** Every turn keys on one fixed,
sessions in `~/.claude/projects/`, which is bind-mounted harness-owned session title (`turn::session_title()`, default
persistently). Auto-compact and auto-memory are disabled via the `hive-session`, override `HIVE_SESSION_TITLE`). A turn `--resume <title>`s
managed settings at `/etc/claude-code/managed-settings.json` because it; the *first* use (bootstrap, post-archive, post-purge) misses and
hyperhive owns compaction — see [Compaction](#compaction) below. `turn::run_claude_resume_or_create` re-runs the same prompt once with
A one-shot `--continue` suppression is available via `--name <title>` to mint it. That single self-heal rule is the whole
`POST /api/new-session` (or `/new-session` slash command in the identity system — there is **no** scraped session-id file. Because the
per-agent terminal) — `Bus::take_skip_continue()` flips an title is constant, `/compact` and its post-compact retry provably target
`AtomicBool` once per turn, the next claude invocation drops the same session (killing the old "compact ran on a different/empty
`--continue`, every subsequent turn resumes normal behaviour. session" bug), and a `choom` invocation in the same cwd can't hijack the
context (it won't carry our title). claude stores sessions in
`~/.claude/projects/<cwd-slug>/<uuid>.jsonl` (bind-mounted persistently);
`--name` writes the title into the file as a `custom-title` event, which
is what `--resume <title>` resolves against. We never pass bare
`--continue` (it resumes the *latest* session in the cwd — the hijack
vector). Auto-compact and auto-memory are disabled via the managed
settings at `/etc/claude-code/managed-settings.json` because hyperhive
owns compaction — see [Compaction](#compaction) below.
**Session reset** is available via `POST /api/new-session` (or
`/new-session` slash command). It does *not* touch the session inline —
that would race a mid-write claude process. Instead `Bus::request_session_reset()`
sets a one-shot flag consumed at the next turn boundary by `drive_turn`,
which **archives** the current session: the backing `<uuid>.jsonl` is
renamed to `<uuid>.jsonl.archived` (dropped out of claude's `*.jsonl`
resolution glob, history preserved on disk, only the file carrying *our*
title — any `choom` session sharing the cwd is left alone). The next
turn's `--resume <title>` then misses and self-heals into a fresh session.
### Compaction ### Compaction
@ -231,20 +249,22 @@ Useful for agents running large-context models (sonnet/opus) where the 75%
heuristic fires before the session is actually full — the reactive path heuristic fires before the session is actually full — the reactive path
(compact-on-overflow when the session hits the hard limit) still applies. (compact-on-overflow when the session hits the hard limit) still applies.
- **Auto session-reset** — a third path that fires when both - **Auto session-reset** — a third path (`turn::maybe_auto_reset`,
conditions hold: context is ≥ a watermark (`HIVE_AUTO_RESET_WATERMARK_TOKENS`, pre-turn) that fires when both conditions hold: context is ≥ a watermark
default **50% of `context_window_tokens(model)`**) AND the time since (`HIVE_AUTO_RESET_WATERMARK_TOKENS`, default **50% of
the last turn exceeds the assumed prompt-cache TTL `context_window_tokens(model)`**) AND the time since the last turn
(`HIVE_CACHE_TTL_SECS`, default `3600`). exceeds the assumed prompt-cache TTL (`HIVE_CACHE_TTL_SECS`, default
Claude's prompt cache lives ~5 minutes; if the cache is already `3600`). Claude's prompt cache goes cold after a while; once it's cold,
cold, resuming with `--continue` pays the full re-upload cost of `--resume`-ing a large session pays the full re-upload cost with no
the current context with no benefit over starting fresh. So: benefit over starting fresh. So `drive_turn` **archives** the current
`drive_turn` injects one `AUTO_RESET_CHECKPOINT_PROMPT` notes turn session (same mechanism as the operator reset — rename `<uuid>.jsonl` →
("flush state to files, cache is cold") then arms `.archived`) so the next turn's `--resume <title>` misses and starts
`Bus::take_skip_continue()` for the real turn — the next turn runs fresh. Unlike proactive compaction the session is dropped entirely, not
without `--continue`, starting a fresh session. Unlike proactive compacted — and *no* preceding checkpoint turn runs, because any turn
compaction the session is dropped entirely, not compacted. Set before the reset would just re-warm the cache and defeat the purpose.
`HIVE_AUTO_RESET_WATERMARK_TOKENS=0` to disable. Set `HIVE_AUTO_RESET_WATERMARK_TOKENS=0` to disable. Auto-reset and the
operator reset are mutually exclusive per turn (both archive → fresh
turn), so an explicit operator reset short-circuits the heuristic.
The child runs with `cwd = /state` (when the bind exists; falls The child runs with `cwd = /state` (when the bind exists; falls
back to the parent's cwd in dev), so any relative path in a tool back to the parent's cwd in dev), so any relative path in a tool
@ -318,9 +338,11 @@ socket at `/run/hive/` once at startup:
same way. same way.
The shared per-turn plumbing lives in `hive_ag3nt::turn::{write_mcp_config, The shared per-turn plumbing lives in `hive_ag3nt::turn::{write_mcp_config,
write_settings, write_system_prompt, run_turn, drive_turn, write_system_prompt, run_turn, drive_turn, emit_turn_end, wait_for_login,
emit_turn_end, wait_for_login, compact_session}` — the single code path compact_session, session_title}` — the single code path every agent role
every agent role runs through. runs through. Session identity is internal to the module:
`run_claude_resume_or_create` (resume-or-create self-heal) and
`archive_session` (turn-boundary reset) sit under `run_turn` / `drive_turn`.
### Reference docs (`hyperhive.docs.enable`) ### Reference docs (`hyperhive.docs.enable`)

48
hive-ag3nt/src/events.rs
View file

@ -680,21 +680,24 @@ pub struct Bus {
/// `container_view` can surface the status on the dashboard without /// `container_view` can surface the status on the dashboard without
/// a live socket call. /// a live socket call.
rate_limited: Arc<AtomicBool>, rate_limited: Arc<AtomicBool>,
/// One-shot: next `run_claude` call drops `--continue`, starting /// One-shot: operator asked to reset the claude session (`POST
/// a fresh claude session. Set by `POST /api/new-session` from /// /api/new-session`). Consumed at the *next* turn boundary by
/// the per-agent web UI; consumed (cleared back to false) by the /// `turn::drive_turn`, which archives the current session so the turn
/// next turn. Subsequent turns resume normal `--continue` /// starts fresh. Deferred (not applied in the handler) so the archive
/// behavior. Atomic so the consumer can take-and-clear without a /// never races a claude process mid-write — one claude per container,
/// lock. /// serialized by the serve loop, so the turn boundary is the only point
skip_continue_once: Arc<AtomicBool>, /// where no session file is open.
session_reset_pending: Arc<AtomicBool>,
/// Current fresh-claude-session id (FK to `sessions.id`). Set by the /// Current fresh-claude-session id (FK to `sessions.id`). Set by the
/// bin loop after minting a session row on a fresh start; stamped onto /// bin loop after minting a session row on a fresh start; stamped onto
/// every `turn_stats` row until the next fresh session. `None` before /// every `turn_stats` row until the next fresh session. `None` before
/// the first fresh turn or when the stats db is unavailable. /// the first fresh turn or when the stats db is unavailable.
session_id: Arc<Mutex<Option<i64>>>, session_id: Arc<Mutex<Option<i64>>>,
/// One-shot: `run_claude` flips this true when it suppresses /// One-shot: `run_claude` flips this true when it creates a fresh claude
/// `--continue` (a fresh session). The bin loop takes-and-clears it /// session (`--name <title>` on a title miss). The bin loop takes-and-
/// after the turn to decide whether to mint a new `sessions` row. /// clears it after the turn to decide whether to mint a new `sessions`
/// row. Session identity itself is handled entirely in `turn.rs` via the
/// constant title + archive — this flag is purely a stats signal.
fresh_session: Arc<AtomicBool>, fresh_session: Arc<AtomicBool>,
/// Per-turn tool-call counter. Reset by the bin loop between /// Per-turn tool-call counter. Reset by the bin loop between
/// turns via `take_tool_calls`. Populated by `observe_stream` as /// turns via `take_tool_calls`. Populated by `observe_stream` as
@ -773,7 +776,7 @@ impl Bus {
last_ctx_usage: Arc::new(Mutex::new(None)), last_ctx_usage: Arc::new(Mutex::new(None)),
last_cost_usage: Arc::new(Mutex::new(None)), last_cost_usage: Arc::new(Mutex::new(None)),
rate_limited: Arc::new(AtomicBool::new(was_rate_limited)), rate_limited: Arc::new(AtomicBool::new(was_rate_limited)),
skip_continue_once: Arc::new(AtomicBool::new(false)), session_reset_pending: Arc::new(AtomicBool::new(false)),
session_id: Arc::new(Mutex::new(None)), session_id: Arc::new(Mutex::new(None)),
fresh_session: Arc::new(AtomicBool::new(false)), fresh_session: Arc::new(AtomicBool::new(false)),
tool_calls: Arc::new(Mutex::new(std::collections::HashMap::new())), tool_calls: Arc::new(Mutex::new(std::collections::HashMap::new())),
@ -797,23 +800,24 @@ impl Bus {
self.event_seq.fetch_add(1, Ordering::SeqCst) + 1 self.event_seq.fetch_add(1, Ordering::SeqCst) + 1
} }
/// Arm the one-shot: the next claude invocation will run without /// Request a session reset (operator `POST /api/new-session`). Deferred:
/// `--continue`, dropping any prior session context. Idempotent /// the flag is consumed at the next turn boundary by `drive_turn`, which
/// — calling twice in a row before the next turn still consumes /// archives the current session so no claude process is mid-write when the
/// to a single fresh-start. /// file is renamed. Idempotent — two clicks before the next turn still
pub fn request_new_session(&self) { /// archive once.
self.skip_continue_once.store(true, Ordering::SeqCst); pub fn request_session_reset(&self) {
self.session_reset_pending.store(true, Ordering::SeqCst);
} }
/// Take + clear the one-shot. Returns true iff the caller should /// Take + clear the session-reset one-shot. Returns true iff `drive_turn`
/// run claude without `--continue` for this turn. /// should archive the current session before this turn.
#[must_use] #[must_use]
pub fn take_skip_continue(&self) -> bool { pub fn take_session_reset(&self) -> bool {
self.skip_continue_once.swap(false, Ordering::SeqCst) self.session_reset_pending.swap(false, Ordering::SeqCst)
} }
/// Mark that the current turn started a fresh claude session. /// Mark that the current turn started a fresh claude session.
/// `run_claude` calls this when it suppresses `--continue`. /// `run_claude` calls this when it creates a new titled session.
pub fn mark_fresh_session(&self) { pub fn mark_fresh_session(&self) {
self.fresh_session.store(true, Ordering::SeqCst); self.fresh_session.store(true, Ordering::SeqCst);
} }

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

@ -3,6 +3,7 @@
//! role: agent). //! role: agent).
use std::collections::VecDeque; use std::collections::VecDeque;
use std::io::BufRead as _;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::process::Stdio; use std::process::Stdio;
use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
@ -63,18 +64,28 @@ const AUTH_FAIL_MARKERS: &[&str] = &[
"Failed to authenticate. API Error: 401", "Failed to authenticate. API Error: 401",
]; ];
/// Substring claude-code emits when `--resume <id>` is handed a session id /// Substrings claude-code emits when a `--resume <title>` target can't be
/// that doesn't exist in this cwd's project (stale persisted id, or a crash /// resolved: either no session carries our constant title yet (first turn,
/// before the first turn ever completed a `.jsonl`). On a hit we clear the /// post-archive, post-purge) or a stale id was handed in. On a hit the
/// persisted id so the NEXT turn starts a fresh session and re-captures — /// harness re-runs the SAME prompt once with `--name <title>` to mint a
/// the agent self-heals instead of failing `--resume` forever. /// fresh session titled `<title>`, so the agent self-heals instead of
const SESSION_NOT_FOUND_MARKER: &str = "No conversation found with session ID"; /// 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",
];
/// Name of the harness-owned file under `paths::harness_dir()` that holds /// Fixed, harness-owned claude session title. Every turn / compact /
/// the claude session id to resume. Written after every turn with the id /// checkpoint resumes THIS title (`--resume <title>`); the create path
/// claude reported on its stream (the id can change across resume turns in /// names it (`--name <title>`). One constant identity per agent means
/// some claude-code versions, so we always rewrite with the last-seen value). /// compaction and the post-compact retry provably target the same session —
const CLAUDE_SESSION_ID_FILE: &str = "claude-session-id"; /// there is no scraped UUID to go stale, empty, or diverge. Each agent runs
/// in its own container (own `~/.claude` + own `/state` cwd), so even the
/// shared default never collides across agents. Override via
/// `HIVE_SESSION_TITLE`.
const DEFAULT_SESSION_TITLE: &str = "hive-session";
/// How long to sleep after detecting a rate-limit before re-entering the /// How long to sleep after detecting a rate-limit before re-entering the
/// serve loop. Overridable via `HIVE_RATE_LIMIT_SLEEP_SECS`. Default is /// serve loop. Overridable via `HIVE_RATE_LIMIT_SLEEP_SECS`. Default is
@ -267,9 +278,10 @@ fn compact_watermark_tokens(bus: &Bus) -> u64 {
/// ///
/// - **Auto-reset (pre-turn)** — context is large AND the prompt cache has /// - **Auto-reset (pre-turn)** — context is large AND the prompt cache has
/// gone cold (idle gap ≥ cache TTL). Resuming would re-upload the full /// gone cold (idle gap ≥ cache TTL). Resuming would re-upload the full
/// transcript uncached at the same cost as a fresh start. The harness runs /// transcript uncached at the same cost as a fresh start. The harness
/// one checkpoint turn (agent flushes state), then arms a one-shot /// archives the current session (so the next `--resume <title>` misses and
/// `request_new_session` so the actual turn starts fresh. /// self-heals into a fresh `--name <title>`); no checkpoint turn runs here
/// because any turn before the archive would just re-warm the cache.
/// - **Reactive (on overflow)** — `run_turn` returns `PromptTooLong`: the /// - **Reactive (on overflow)** — `run_turn` returns `PromptTooLong`: the
/// session is already past the context window and *no* turn can run on it, /// session is already past the context window and *no* turn can run on it,
/// so we compact immediately and retry the same wake-up prompt once. No /// so we compact immediately and retry the same wake-up prompt once. No
@ -284,7 +296,21 @@ fn compact_watermark_tokens(bus: &Bus) -> u64 {
/// ///
/// Called once per turn by the `hive` serve loop (every agent role). /// Called once per turn by the `hive` serve loop (every agent role).
pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome { pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome {
maybe_auto_reset(bus); // Start this turn on a fresh session when either trigger fires. Both
// archive the current session at this turn boundary (no claude is mid-
// write — one claude per container, serialized by the serve loop) and
// produce the same end state, so they're mutually exclusive: an explicit
// operator reset makes the auto-reset heuristic moot for this turn.
if bus.take_session_reset() {
// Operator-requested (deferred from `POST /api/new-session`).
bus.emit(LiveEvent::Note {
text: "operator: resetting session — archiving before this turn".into(),
});
archive_session(bus);
} else {
// Heuristic: context large AND prompt cache gone cold.
maybe_auto_reset(bus);
}
let outcome = match run_turn(prompt, files, bus).await { let outcome = match run_turn(prompt, files, bus).await {
TurnOutcome::PromptTooLong => { TurnOutcome::PromptTooLong => {
// Compact has its own three-flag surface (it's the same claude // Compact has its own three-flag surface (it's the same claude
@ -403,10 +429,11 @@ async fn do_compact(files: &TurnFiles, bus: &Bus) {
} }
/// Pre-turn auto-reset check. If context is large AND the prompt cache has /// Pre-turn auto-reset check. If context is large AND the prompt cache has
/// gone cold (idle time >= cache TTL), arm `request_new_session` so the /// gone cold (idle time >= cache TTL), archive the current session so the
/// next wake-up turn starts fresh. No preceding checkpoint turn — running /// next wake-up turn's `--resume <title>` misses and self-heals into a fresh
/// any turn before the reset would re-upload and re-warm the cache, which /// `--name <title>` session. No preceding checkpoint turn — running any turn
/// defeats the cost-optimisation purpose entirely. /// before the reset would re-upload and re-warm the cache, which defeats the
/// cost-optimisation purpose entirely.
fn maybe_auto_reset(bus: &Bus) { fn maybe_auto_reset(bus: &Bus) {
let watermark = auto_reset_watermark_tokens(bus); let watermark = auto_reset_watermark_tokens(bus);
if watermark == 0 { if watermark == 0 {
@ -437,7 +464,7 @@ fn maybe_auto_reset(bus: &Bus) {
— dropping session (cache cold, fresh start is equally cheap)" — dropping session (cache cold, fresh start is equally cheap)"
), ),
}); });
bus.request_new_session(); archive_session(bus);
} }
/// Emit the per-turn `TurnEnd` event + log line. Single owner so outcome /// Emit the per-turn `TurnEnd` event + log line. Single owner so outcome
@ -575,22 +602,22 @@ fn session_refreshed(prev: DirSnapshot, now: DirSnapshot) -> bool {
} }
} }
/// Spawn `claude` for one turn and pump `stream-json` stdout into the /// Run one turn against the constant-title session, resuming it or creating
/// live event bus. Prompt goes over stdin (variadic /// it on first use. Delegates to [`run_claude_resume_or_create`]; the session
/// `--allowedTools`/`--tools` would otherwise eat a trailing positional /// is pinned by a fixed `--name`/`--resume <title>` (NOT bare `--continue`,
/// prompt). The session is persistent across turns via `--resume <id>` /// which resumes the *latest* session in this cwd and lets a `choom` session
/// against the harness's own captured session id (NOT bare `--continue`, /// hijack the live harness context — a constant title is immune since choom
/// which resumes the *latest* session in this cwd and so lets a `choom` /// won't carry it). claude's in-session auto-compact is disabled via the
/// session hijack the live harness context). claude's in-session /// managed settings at `/etc/claude-code/managed-settings.json` so it doesn't
/// 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. /// stall mid-turn — hyperhive owns compaction.
pub async fn run_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome { pub async fn run_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome {
match run_claude(prompt, files, bus).await { match run_claude_resume_or_create(prompt, files, bus).await {
Ok((true, _, _)) => TurnOutcome::PromptTooLong, Ok(ClaudeResult::PromptTooLong) => TurnOutcome::PromptTooLong,
Ok((_, true, _)) => TurnOutcome::RateLimited, Ok(ClaudeResult::RateLimited) => TurnOutcome::RateLimited,
Ok((_, _, true)) => TurnOutcome::AuthFailed, Ok(ClaudeResult::AuthFailed) => TurnOutcome::AuthFailed,
Ok(_) => TurnOutcome::Ok, // `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), Err(e) => TurnOutcome::Failed(e),
} }
} }
@ -613,11 +640,21 @@ pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> TurnOutcome {
bus.emit(LiveEvent::Note { bus.emit(LiveEvent::Note {
text: "context overflow — running /compact on the persistent session".into(), text: "context overflow — running /compact on the persistent session".into(),
}); });
let outcome = match run_claude("/compact", files, bus).await { // Resume-only: never create a session just to compact it. If the titled
Ok((true, _, _)) => TurnOutcome::PromptTooLong, // session doesn't exist there's genuinely nothing to compact — compacting
Ok((_, true, _)) => TurnOutcome::RateLimited, // a freshly-minted empty session would just print "not enough messages"
Ok((_, _, true)) => TurnOutcome::AuthFailed, // (the historical version-B failure), so treat a title miss as a no-op Ok.
Ok(_) => TurnOutcome::Ok, let outcome = match run_claude("/compact", files, bus, false).await {
Ok(ClaudeResult::TitleNotFound) => {
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) => TurnOutcome::Failed(e),
}; };
match &outcome { match &outcome {
@ -640,45 +677,190 @@ pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> TurnOutcome {
outcome 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
/// never collides across agents.
#[must_use]
pub fn session_title() -> String {
std::env::var("HIVE_SESSION_TITLE")
.ok()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.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.
fn session_cwd() -> PathBuf {
let state_dir = crate::paths::state_dir();
if state_dir.is_dir() {
state_dir
} else {
std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
}
}
/// `~/.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)
}
/// 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);
}
}
None
}
/// 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
/// `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");
bus.emit(LiveEvent::Note {
text: format!("archived session \"{title}\" ({name}) — next turn starts fresh"),
});
}
Err(e) => {
tracing::warn!(error = %e, path = %path.display(), "failed to archive claude session");
bus.emit(LiveEvent::Note {
text: format!("failed to archive session \"{title}\": {e}"),
});
}
}
}
/// 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( #[allow(
clippy::too_many_lines, clippy::too_many_lines,
reason = "one linear subprocess driver: spawn claude, stream + classify \ reason = "one linear subprocess driver: spawn claude, stream + classify \
stdout/stderr, then assemble the outcome; splitting it would \ stdout/stderr, then assemble the outcome; splitting it would \
fragment the streaming state across helpers" fragment the streaming state across helpers"
)] )]
async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool, bool, bool)> { 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 // 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 // include real context in the bail message (and downstream in the
// failure notification to the manager) instead of just "exit 1". // failure notification to the manager) instead of just "exit 1".
const STDERR_TAIL_LINES: usize = 20; const STDERR_TAIL_LINES: usize = 20;
let model = bus.model(); let model = bus.model();
let effort = bus.effort(); let effort = bus.effort();
// Resolve which claude session to resume. We NEVER pass bare // Constant session identity. Every call keys on the same fixed title:
// `--continue`: that resumes the latest session in this cwd, which a // `--resume <title>` normally, `--name <title>` on the create path (first
// `choom` invocation (same cwd) can hijack, wiping the harness context. // use / post-archive / post-purge, driven by `run_claude_resume_or_create`
// Instead we `--resume <id>` against the id claude reported on a prior // on a title miss). We NEVER pass bare `--continue`: that resumes the
// turn, persisted under `harness_dir()/claude-session-id`. // latest session in this cwd, which a `choom` invocation (same cwd) can
let persist_path = crate::paths::harness_dir().join(CLAUDE_SESSION_ID_FILE); // hijack — a constant title is immune since choom won't carry it. No
let resume_id: Option<String> = if bus.take_skip_continue() { // scraped UUID, no persist file, so compaction + the post-compact retry
// Fresh session requested: mint a new one (no --resume). Flag it so // provably target the same session.
// the bin loop mints a new `sessions` row + stamps its id onto this let title = session_title();
// turn's stats. Drop any stale persisted id — the new id claude if create {
// reports this turn is captured + written below. // 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.mark_fresh_session();
let _ = std::fs::remove_file(&persist_path);
bus.emit(LiveEvent::Note { bus.emit(LiveEvent::Note {
text: "fresh session (continue suppressed for this turn)".into(), text: format!("creating fresh session titled \"{title}\""),
}); });
None }
} else {
// Continue: resume OUR captured id. Absent (first turn / just
// self-healed from a stale id) → fall through to a fresh session
// and capture the new id below.
match std::fs::read_to_string(&persist_path) {
Ok(s) if !s.trim().is_empty() => Some(s.trim().to_string()),
_ => None,
}
};
let mut cmd = Command::new("claude"); let mut cmd = Command::new("claude");
// Spawn inside the agent's state dir so relative paths in tool calls // 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 // (Read foo.md, Bash ls, Write notes.md) land in the durable dir
@ -696,8 +878,10 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
.arg(&model) .arg(&model)
.arg("--effort") .arg("--effort")
.arg(&effort); .arg(&effort);
if let Some(id) = &resume_id { if create {
cmd.arg("--resume").arg(id); cmd.arg("--name").arg(&title);
} else {
cmd.arg("--resume").arg(&title);
} }
cmd.arg("--system-prompt-file").arg(&files.system_prompt); cmd.arg("--system-prompt-file").arg(&files.system_prompt);
cmd.arg("--mcp-config") cmd.arg("--mcp-config")
@ -736,21 +920,17 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
let prompt_too_long = Arc::new(AtomicBool::new(false)); let prompt_too_long = Arc::new(AtomicBool::new(false));
let rate_limited = Arc::new(AtomicBool::new(false)); let rate_limited = Arc::new(AtomicBool::new(false));
let auth_failed = Arc::new(AtomicBool::new(false)); let auth_failed = Arc::new(AtomicBool::new(false));
// `--resume` against a stale/missing id: clear the persist file so the // `--resume <title>` found no session carrying the title: the caller
// next turn self-heals into a fresh session. // (`run_claude_resume_or_create`) re-runs once with `--name <title>`.
let session_not_found = Arc::new(AtomicBool::new(false)); let title_not_found = Arc::new(AtomicBool::new(false));
// Last `session_id` claude reported on its stream this turn; persisted
// after the child exits so the next turn `--resume`s it.
let session_id_seen = Arc::new(Mutex::new(None::<String>));
let flag_out = prompt_too_long.clone(); let flag_out = prompt_too_long.clone();
let flag_err = prompt_too_long.clone(); let flag_err = prompt_too_long.clone();
let rate_out = rate_limited.clone(); let rate_out = rate_limited.clone();
let rate_err = rate_limited.clone(); let rate_err = rate_limited.clone();
let auth_out = auth_failed.clone(); let auth_out = auth_failed.clone();
let auth_err = auth_failed.clone(); let auth_err = auth_failed.clone();
let notfound_out = session_not_found.clone(); let notfound_out = title_not_found.clone();
let notfound_err = session_not_found.clone(); let notfound_err = title_not_found.clone();
let session_id_out = session_id_seen.clone();
let bus_out = bus.clone(); let bus_out = bus.clone();
let bus_err = bus.clone(); let bus_err = bus.clone();
let pump_stdout = tokio::spawn(async move { let pump_stdout = tokio::spawn(async move {
@ -778,19 +958,10 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) { if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) {
auth_out.store(true, Ordering::Relaxed); auth_out.store(true, Ordering::Relaxed);
} }
if line.contains(SESSION_NOT_FOUND_MARKER) { if TITLE_NOT_FOUND_MARKERS.iter().any(|m| line.contains(m)) {
notfound_out.store(true, Ordering::Relaxed); notfound_out.store(true, Ordering::Relaxed);
} }
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&line) { if let Ok(v) = serde_json::from_str::<serde_json::Value>(&line) {
// Track the session id claude reports (init + result events
// both carry it). Persisted after exit so the next turn
// `--resume`s it; re-captured each turn since the id can
// change across resumes in some claude-code versions.
if let Some(sid) = v.get("session_id").and_then(|s| s.as_str())
&& !sid.is_empty()
{
*session_id_out.lock().unwrap() = Some(sid.to_string());
}
// Rate-limit detection: only fire on JSON `error` events, // Rate-limit detection: only fire on JSON `error` events,
// not on arbitrary text content. An agent discussing a past // not on arbitrary text content. An agent discussing a past
// rate limit in its response would otherwise trigger a false // rate limit in its response would otherwise trigger a false
@ -857,7 +1028,7 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) { if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) {
auth_err.store(true, Ordering::Relaxed); auth_err.store(true, Ordering::Relaxed);
} }
if line.contains(SESSION_NOT_FOUND_MARKER) { if TITLE_NOT_FOUND_MARKERS.iter().any(|m| line.contains(m)) {
notfound_err.store(true, Ordering::Relaxed); notfound_err.store(true, Ordering::Relaxed);
} }
// Mirror to journald so post-mortems work without the web UI // Mirror to journald so post-mortems work without the web UI
@ -882,19 +1053,15 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
let too_long = prompt_too_long.load(Ordering::Relaxed); let too_long = prompt_too_long.load(Ordering::Relaxed);
let is_rate_limited = rate_limited.load(Ordering::Relaxed); let is_rate_limited = rate_limited.load(Ordering::Relaxed);
let is_auth_failed = auth_failed.load(Ordering::Relaxed); let is_auth_failed = auth_failed.load(Ordering::Relaxed);
// Session-id bookkeeping. On a stale/missing `--resume` id, drop the let is_title_not_found = title_not_found.load(Ordering::Relaxed);
// persist file so the next turn starts fresh and self-heals. Otherwise // A title miss is a clean exit-1 (no session to resume yet), so it must
// rewrite it with the id claude reported this turn (handles the id // not be treated as a hard failure — the caller re-runs with `--name`.
// changing across resumes in some claude-code versions). if !status.success()
if session_not_found.load(Ordering::Relaxed) { && !too_long
let _ = std::fs::remove_file(&persist_path); && !is_rate_limited
} else if let Some(sid) = session_id_seen.lock().unwrap().clone() { && !is_auth_failed
if let Some(parent) = persist_path.parent() { && !is_title_not_found
let _ = std::fs::create_dir_all(parent); {
}
let _ = std::fs::write(&persist_path, sid);
}
if !status.success() && !too_long && !is_rate_limited && !is_auth_failed {
let tail = stderr_tail.lock().unwrap(); let tail = stderr_tail.lock().unwrap();
if tail.is_empty() { if tail.is_empty() {
bail!("claude exited {status} (no stderr)"); bail!("claude exited {status} (no stderr)");
@ -902,7 +1069,20 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
let tail_str = tail.iter().cloned().collect::<Vec<_>>().join("\n"); let tail_str = tail.iter().cloned().collect::<Vec<_>>().join("\n");
bail!("claude exited {status}\nstderr tail:\n{tail_str}"); bail!("claude exited {status}\nstderr tail:\n{tail_str}");
} }
Ok((too_long, is_rate_limited, is_auth_failed)) // 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)] #[cfg(test)]

29
hive-ag3nt/src/web_ui.rs
View file

@ -1134,23 +1134,22 @@ async fn post_compact(State(state): State<AppState>) -> Response {
(axum::http::StatusCode::OK, "ok").into_response() (axum::http::StatusCode::OK, "ok").into_response()
} }
/// Cancel the in-flight claude turn. Coarse-grained: shells out /// Request a session reset. The current session is archived (its backing
/// `pkill -INT claude` since there's at most one claude per container. /// `<uuid>.jsonl` renamed out of claude's resolution glob) at the next turn
/// SIGINT (not SIGTERM) so claude flushes anything in-flight and emits a /// boundary, so the following turn's `--resume` misses and self-heals into a
/// final result row. Emits a Note so the operator sees the cancel /// freshly-named session. History is preserved on disk, not deleted.
/// landed; the actual state transition back to `idle` happens when ///
/// `run_claude` wakes up and the harness emits `TurnEnd`. /// Deferred (a one-shot flag consumed by `drive_turn`) rather than applied
/// Arm a one-shot: the next claude turn drops `--continue`, starting a /// here: renaming the session file while a claude turn is mid-write would
/// fresh session. Subsequent turns resume normal `--continue` /// race the live process. Between turns there is no open session file (one
/// behavior. Idempotent before the next turn fires — calling twice /// claude per container, serialized by the serve loop), so the archive is
/// still results in a single fresh start. Useful when the /// safe there. Useful when the session-resume context is poisoned (claude
/// session-resume context is poisoned (claude went off the rails, /// went off the rails, hit an unrecoverable refusal, etc.) and a full reset
/// hit an unrecoverable refusal, etc.) and a full reset is cheaper /// is cheaper than asking claude to forget mid-stream.
/// than asking claude to forget mid-stream.
async fn post_new_session(State(state): State<AppState>) -> Response { async fn post_new_session(State(state): State<AppState>) -> Response {
state.bus.request_new_session(); state.bus.request_session_reset();
state.bus.emit(crate::events::LiveEvent::Note { state.bus.emit(crate::events::LiveEvent::Note {
text: "operator: new session armed — next turn runs without --continue".into(), text: "operator: session reset queued — takes effect at the next turn".into(),
}); });
(axum::http::StatusCode::OK, "ok").into_response() (axum::http::StatusCode::OK, "ok").into_response()
} }