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

feat(hive-claude): add InfiniteSession (name+store+compaction policy)

Browse source
This commit is contained in:
müde 2026-07-05 19:38:16 +02:00
commit 80d819e444
8 changed files with 501 additions and 283 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -2,14 +2,15 @@
use std::path::PathBuf;
/// Which claude session a run should attach to. Kept separate from [`Config`]
/// so a single config can drive resume + create + `/compact` of the same
/// logical session across turns.
/// 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 Session {
pub enum Attach {
/// `--resume <id-or-title>` — resume an existing session by UUID or by the
/// display title set via [`Session::Create`]. Yields
/// [`crate::Outcome::SessionNotFound`] if nothing matches.
/// 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
@ -23,7 +24,7 @@ pub enum Session {
}
/// Everything needed to build one headless `claude --print` invocation, minus
/// the [`Session`] attachment (passed separately to [`crate::run`]).
/// 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

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

@ -1,5 +1,5 @@
//! The subprocess driver: spawn claude, pump + classify its streams, and
//! assemble an [`Outcome`].
//! assemble the result.
use std::collections::VecDeque;
use std::process::Stdio;
@ -8,7 +8,7 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{ChildStderr, ChildStdout, Command};
use crate::classify::Sentinels;
use crate::{Config, Error, Result, Session, Sink};
use crate::{Attach, Config, Error, Result, Sink};
/// Default program name spawned when [`Config::program`] is unset.
const DEFAULT_PROGRAM: &str = "claude";
@ -16,9 +16,9 @@ const DEFAULT_PROGRAM: &str = "claude";
/// How many trailing stderr lines to keep for [`Error::Exit`].
const STDERR_TAIL_LINES: usize = 20;
/// The driver entry point. A namespace for the run functions — there is
/// nothing to construct; call the associated functions directly
/// (`Claude::run(…)`, `Claude::run_resume_or_create(…)`).
/// 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 {
@ -40,12 +40,12 @@ impl Claude {
/// - [`Error::Exit`] on a non-zero exit that raised no sentinel.
pub async fn run(
config: &Config,
session: &Session,
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, session);
let mut cmd = build_command(program, config, attach);
let mut child = cmd
.stdin(Stdio::piped())
@ -88,42 +88,12 @@ impl Claude {
}
Ok(())
}
/// Resume a titled session, creating it on first use. Runs
/// `--resume <title>`; on [`Error::SessionNotFound`] it re-runs the *same
/// prompt* once with `--name <title>` to mint the session.
///
/// Returns `Ok(true)` when a fresh session was created (the resume missed),
/// `Ok(false)` when an existing session was resumed. This is the common
/// "one durable session per agent, self-healing on first boot / after the
/// file is archived away" pattern, kept generic here.
///
/// # Errors
///
/// Propagates any [`Error`] from the underlying [`Claude::run`] calls
/// (other than the `SessionNotFound` on the first attempt, which is handled
/// by creating).
pub async fn run_resume_or_create(
config: &Config,
title: &str,
prompt: &str,
sink: &impl Sink,
) -> Result<bool> {
match Self::run(config, &Session::Resume(title.to_string()), prompt, sink).await {
Err(Error::SessionNotFound) => {
Self::run(config, &Session::Create(title.to_string()), prompt, sink).await?;
Ok(true)
}
Ok(()) => Ok(false),
Err(other) => Err(other),
}
}
}
/// Assemble the argv. `--print --verbose --output-format stream-json` are
/// mandatory (the driver parses that shape); everything else is gated on the
/// [`Config`] / [`Session`].
fn build_command(program: &str, config: &Config, session: &Session) -> Command {
/// [`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);
@ -138,17 +108,17 @@ fn build_command(program: &str, config: &Config, session: &Session) -> Command {
if let Some(effort) = &config.effort {
cmd.arg("--effort").arg(effort);
}
match session {
Session::Resume(id) => {
match attach {
Attach::Resume(id) => {
cmd.arg("--resume").arg(id);
}
Session::Create(title) => {
Attach::Create(title) => {
cmd.arg("--name").arg(title);
}
Session::Continue => {
Attach::Continue => {
cmd.arg("--continue");
}
Session::OneOff => {}
Attach::OneOff => {}
}
if let Some(path) = &config.system_prompt_file {
cmd.arg("--system-prompt-file").arg(path);

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

@ -7,9 +7,16 @@
//! prompt-too-long, …) and hard failures (spawn, non-zero exit) — is a variant
//! of the single [`Error`] enum, so callers branch with one `match`. The crate
//! also locates and archives on-disk sessions by title ([`SessionStore`]). It
//! knows only about the Claude Code CLI — no application types, policy,
//! watermarks, or logging. Callers wire streaming output through a [`Sink`] and
//! layer their own compaction / retry / reset policy on top.
//! 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
//!
@ -32,13 +39,13 @@
//!
//! ```no_run
//! # async fn ex() {
//! use hive_claude::{Claude, Config, Error, NoopSink, Session};
//! use hive_claude::{Attach, Claude, Config, Error, NoopSink};
//!
//! let config = Config {
//! model: "haiku".into(),
//! ..Default::default()
//! };
//! match Claude::run(&config, &Session::Resume("my-session".into()), "hello", &NoopSink).await {
//! 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 */ }
@ -51,11 +58,17 @@ mod classify;
mod config;
mod driver;
mod error;
mod policy;
mod session;
mod sink;
mod store;
mod usage;
pub use config::{Config, Session};
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 usage::Usage;

130
hive-claude/src/policy.rs Normal file
View file

@ -0,0 +1,130 @@
//! 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),
}));
}
}

175
hive-claude/src/session.rs Normal file
View file

@ -0,0 +1,175 @@
//! 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, Usage, usage};
/// 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, Copy, 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,
}
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> {
let meter = UsageSink::new(sink);
let created = match self.attempt(config, prompt, &meter).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.
self.compact(config, sink).await?;
let created = self.attempt(config, prompt, sink).await?;
return Ok(Progress {
created,
compacted: true,
});
}
Err(other) => return Err(other),
};
// 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(meter.snapshot()) {
if let Some(checkpoint) = self.policy.checkpoint_prompt() {
let _ = self.attempt(config, checkpoint, sink).await;
}
let _ = self.compact(config, sink).await;
return Ok(Progress {
created,
compacted: true,
});
}
Ok(Progress {
created,
compacted: false,
})
}
/// 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<()> {
match Claude::run(config, &Attach::Resume(self.name.clone()), "/compact", sink).await {
Err(Error::SessionNotFound) => Ok(()),
other => other,
}
}
/// One resume-or-create turn. Uses the store to pick resume vs. create up
/// front (avoiding a wasted resume-miss spawn), and 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) -> Result<bool> {
let exists = self.store.find_by_title(&self.name).is_some();
let attach = if exists {
Attach::Resume(self.name.clone())
} else {
Attach::Create(self.name.clone())
};
match Claude::run(config, &attach, prompt, sink).await {
Ok(()) => Ok(!exists),
// We thought it existed but the resume missed (raced an archive) —
// self-heal by creating.
Err(Error::SessionNotFound) if exists => {
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 minimal
/// [`Usage`] the policy needs from the stream. Cheap; the driver calls it
/// synchronously from one reader task, so the `Mutex` only satisfies `&self`.
struct UsageSink<'a, S: Sink> {
inner: &'a S,
usage: Mutex<Usage>,
}
impl<'a, S: Sink> UsageSink<'a, S> {
fn new(inner: &'a S) -> Self {
Self {
inner,
usage: Mutex::new(Usage::default()),
}
}
fn snapshot(&self) -> Usage {
*self.usage.lock().unwrap()
}
}
impl<S: Sink> Sink for UsageSink<'_, S> {
fn on_event(&self, event: &Value) {
if let Some(tokens) = usage::context_tokens(event) {
self.usage.lock().unwrap().context_tokens = tokens;
}
if let Some(window) = usage::context_window(event) {
self.usage.lock().unwrap().context_window = Some(window);
}
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);
}
}

47
hive-claude/src/usage.rs Normal file
View file

@ -0,0 +1,47 @@
//! Minimal context-usage signal parsed from the stream, used to drive
//! [`crate::CompactionPolicy`]. This is deliberately small — just what a
//! compaction decision needs. Consumers that want full per-turn accounting
//! parse the raw events in their own [`crate::Sink`].
use serde_json::Value;
/// The context footprint of the most recent inference in a turn.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Usage {
/// Tokens in the last inference's context (input + cache reads + cache
/// writes) — what counts against the model's window. `0` until the first
/// `assistant` event is seen.
pub context_tokens: u64,
/// The model-reported active context window (`modelUsage.*.contextWindow`
/// on the terminal `result` event), if the turn reported one.
pub context_window: Option<u64>,
}
/// Per-inference context size from an `assistant` event's `message.usage`.
/// Tracking the *last* one over a turn gives the live conversation size (the
/// cumulative `result` usage double-counts tool-call prompts and overshoots).
pub(crate) fn context_tokens(event: &Value) -> Option<u64> {
if event.get("type").and_then(Value::as_str) != Some("assistant") {
return None;
}
let usage = event.get("message")?.get("usage")?;
let field = |k: &str| usage.get(k).and_then(Value::as_u64).unwrap_or(0);
Some(field("input_tokens") + field("cache_read_input_tokens") + field("cache_creation_input_tokens"))
}
/// The per-inference active window from a `result` event's `modelUsage` map
/// (first non-zero `contextWindow` across model keys). This is the limit the
/// model actually enforces, which can be far below the prompt-cache capacity.
pub(crate) fn context_window(event: &Value) -> Option<u64> {
if event.get("type").and_then(Value::as_str) != Some("result") {
return None;
}
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
}