diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index c48e4875..ae28251a 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -9,7 +9,7 @@ use std::path::{Path, PathBuf}; use std::sync::Mutex; use anyhow::Result; -use hive_claude::{Claude, Config, Session, Sink}; +use hive_claude::{Config, InfiniteSession, PercentPolicy, Sink}; use serde_json::Value; use crate::events::{Bus, LiveEvent, TokenUsage}; @@ -59,6 +59,10 @@ const DEFAULT_RATE_LIMIT_SLEEP_SECS: u64 = 300; /// `0` to disable (always resume). const DEFAULT_CACHE_TTL_SECS: u64 = 3600; +/// Default proactive-compaction watermark, as a percent of the effective +/// context window. Overridable via `HIVE_COMPACT_WATERMARK_PERCENT`. +const DEFAULT_COMPACT_PERCENT: u8 = 75; + /// Synthetic wake prompt for the proactive notes-checkpoint turn. Not an /// inbox message — the harness injects it directly so the agent gets one /// turn to persist durable state before `/compact` collapses the @@ -213,42 +217,51 @@ fn cache_ttl_secs() -> u64 { env_u64_positive("HIVE_CACHE_TTL_SECS", DEFAULT_CACHE_TTL_SECS) } -/// Resolve the proactive-compaction watermark. Priority order: -/// 1. `HIVE_COMPACT_WATERMARK_TOKENS` env var (explicit override). -/// 2. 75% of `effective_context_window(bus)`. -/// -/// `0` disables proactive compaction (reactive path still applies). -fn compact_watermark_tokens(bus: &Bus) -> u64 { - env_u64("HIVE_COMPACT_WATERMARK_TOKENS").unwrap_or_else(|| effective_context_window(bus) * 3 / 4) +/// Proactive-compaction watermark as a percent of the effective context +/// window (default [`DEFAULT_COMPACT_PERCENT`]). `0` disables proactive +/// compaction — the reactive on-overflow path still applies. Reads +/// `HIVE_COMPACT_WATERMARK_PERCENT`; the legacy `autoCompact = false` switch +/// (which sets `HIVE_COMPACT_WATERMARK_TOKENS=0`) is still honoured as disable. +fn compact_percent() -> u8 { + if env_u64("HIVE_COMPACT_WATERMARK_TOKENS") == Some(0) { + return 0; + } + let pct = env_u64("HIVE_COMPACT_WATERMARK_PERCENT").unwrap_or(u64::from(DEFAULT_COMPACT_PERCENT)); + u8::try_from(pct.min(100)).unwrap_or(DEFAULT_COMPACT_PERCENT) } -/// Drive one turn end-to-end. Three paths layer on top of the raw `run_turn`: +/// Build the agent's durable session: constant title + on-disk store + a +/// percent-of-window compaction policy that checkpoints (`CHECKPOINT_PROMPT`) +/// before compacting. `default_window` feeds the policy the effective window +/// for turns where the model didn't itself report one. +fn infinite_session(bus: &Bus) -> InfiniteSession { + InfiniteSession::new( + session_title(), + session_store(), + PercentPolicy { + percent: compact_percent(), + default_window: Some(effective_context_window(bus)), + checkpoint_prompt: Some(CHECKPOINT_PROMPT.to_string()), + }, + ) +} + +/// Drive one turn end-to-end. The durable [`InfiniteSession`] owns the +/// resume-or-create + compaction loop (reactive on overflow, and proactive per +/// the percent policy — including the pre-compaction checkpoint turn). This +/// layer wraps it with the two hyperhive-specific concerns: /// -/// - **Auto-reset (pre-turn)** — context is large AND the prompt cache has -/// gone cold (idle gap ≥ cache TTL). Resuming would re-upload the full -/// transcript uncached at the same cost as a fresh start. The harness -/// archives the current session (so the next `--resume ` misses and -/// 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 -/// 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 -/// notes-checkpoint turn is possible here — the detail is gone. -/// - **Proactive (post-turn)** — the turn finished cleanly but its context -/// size has crept past the watermark: while the session is still healthy we -/// give the agent one dedicated turn to checkpoint its `/state` notes, then -/// compact. This keeps a later turn from hitting the reactive path (where -/// there is no chance to save anything first). The graceful-stop path takes -/// the same proactive route — a checkpoint turn before `/compact` is cheap -/// insurance and keeps a later cold-start resume small. +/// - **Session reset (pre-turn)** — an operator reset (`/api/new-session`) or +/// the auto-reset heuristic (context large AND prompt cache gone cold) +/// archives the current session at this turn boundary so the run starts +/// fresh. The two are mutually exclusive. This is deliberately *not* part of +/// the infinite-session abstraction — it's the hive escape hatch. +/// - **401 retry** — a transient token-refresh race can 401 once and clear, so +/// the whole turn is retried a single time before bubbling `AuthFailed` to +/// the serve loop (which parks for re-login). /// -/// Called once per turn by the `hive` serve loop (every agent role). +/// Called once per turn by the `hive` serve loop. pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome { - // 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 { @@ -259,120 +272,33 @@ pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutco // Heuristic: context large AND prompt cache gone cold. maybe_auto_reset(bus); } - let outcome = match run_turn(prompt, files, bus).await { - TurnOutcome::PromptTooLong => { - // Compact has its own three-flag surface (it's the same claude - // binary). Treat any non-Ok outcome the same as if `run_turn` - // had returned it — the serve loop already knows what to do - // with each variant, no point re-wrapping. PromptTooLong from - // /compact itself would be absurd recursion; bubble it up as - // a normal failure path. - match compact_session(files, bus).await { - TurnOutcome::Ok | TurnOutcome::Compacted => run_turn(prompt, files, bus).await, - other => return other, + let config = claude_config(bus, files); + let sink = BusSink::new(bus); + let session = infinite_session(bus); + let mut result = session.run(&config, prompt, &sink).await; + if matches!(result, Err(hive_claude::Error::AuthFailed)) { + bus.emit(LiveEvent::Note { + text: "got 401 — retrying once before parking for re-login".into(), + }); + result = session.run(&config, prompt, &sink).await; + } + match result { + Ok(progress) => { + if progress.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()), + }); + } + if progress.compacted { + TurnOutcome::Compacted + } else { + TurnOutcome::Ok } } - // Rate-limited: no point retrying immediately — bubble up so the - // serve loop can park + emit status before the next attempt. - TurnOutcome::RateLimited => return TurnOutcome::RateLimited, - // Auth failed: may be a transient token-refresh race or brief API - // hiccup. Retry once before bubbling up so the serve loop parks for - // re-login. The retry outcome is passed through unchanged — if it - // fails again the serve loop handles it as usual. - TurnOutcome::AuthFailed => { - bus.emit(LiveEvent::Note { - text: "got 401 — retrying once before parking for re-login".into(), - }); - run_turn(prompt, files, bus).await - } - other => other, - }; - // Proactive: a turn just completed on a still-healthy session. If its - // context crossed the watermark, run a separate notes-checkpoint turn so - // the agent can flush durable state, then compact before a later turn - // overflows into the reactive path. Best-effort — never changes the - // outcome of the turn that already succeeded, but records it as - // `Compacted` so turn stats can distinguish it from a plain `Ok`. - if matches!(outcome, TurnOutcome::Ok) && maybe_checkpoint_and_compact(files, bus).await { - return TurnOutcome::Compacted; - } - outcome -} - -/// Proactive post-turn compaction. If the last inference's context size -/// has crossed the watermark, run one notes-checkpoint turn so the agent -/// can persist durable state, then `/compact`. Best-effort: a failed -/// checkpoint or compaction is logged + surfaced as a Note but never -/// fails the turn that already succeeded. Returns `true` if compaction -/// was attempted (watermark crossed), `false` if skipped. -async fn maybe_checkpoint_and_compact(files: &TurnFiles, bus: &Bus) -> bool { - let Some(used) = watermark_crossed(bus) else { - return false; - }; - let watermark = compact_watermark_tokens(bus); - bus.emit(LiveEvent::Note { - text: format!( - "context at {used} tokens (watermark {watermark}) — running a \ - notes-checkpoint turn before /compact" - ), - }); - // Give the agent one turn to flush durable state into /state. If the - // session is somehow already too far gone to run even this, fall - // through to compaction anyway — the checkpoint is best-effort. - match run_turn(CHECKPOINT_PROMPT, files, bus).await { - TurnOutcome::Ok | TurnOutcome::Compacted => {} - TurnOutcome::PromptTooLong => bus.emit(LiveEvent::Note { - text: "checkpoint turn overflowed the window — compacting without it".into(), - }), - TurnOutcome::RateLimited => bus.emit(LiveEvent::Note { - text: "checkpoint turn was rate-limited — compacting anyway".into(), - }), - TurnOutcome::AuthFailed => bus.emit(LiveEvent::Note { - text: "checkpoint turn hit 401 — skipping compaction, parking for re-login".into(), - }), - TurnOutcome::Failed(e) => bus.emit(LiveEvent::Note { - text: format!("checkpoint turn failed ({e:#}) — compacting anyway"), - }), - } - do_compact(files, bus).await; - true -} - -/// Returns `Some(used_tokens)` when proactive compaction is enabled AND the -/// last inference's context size has reached the watermark; `None` when -/// compaction is disabled (watermark 0), there's no usage reading yet, or -/// the context is still below the watermark. -fn watermark_crossed(bus: &Bus) -> Option<u64> { - let watermark = compact_watermark_tokens(bus); - if watermark == 0 { - return None; // proactive compaction disabled - } - let used = bus.last_ctx_usage().map(|u| u.context_tokens())?; - (used >= watermark).then_some(used) -} - -/// Run `/compact`, surfacing each failure mode as a best-effort Note. -/// Never changes the outcome of the turn that already succeeded — the -/// next real turn surfaces any underlying issue (rate-limit / 401 / etc.) -/// through the normal path anyway. -async fn do_compact(files: &TurnFiles, bus: &Bus) { - match compact_session(files, bus).await { - TurnOutcome::Ok | TurnOutcome::Compacted => {} - TurnOutcome::PromptTooLong => bus.emit(LiveEvent::Note { - text: "/compact unexpectedly returned PromptTooLong — next turn will retry".into(), - }), - TurnOutcome::RateLimited => bus.emit(LiveEvent::Note { - text: "/compact was rate-limited — next turn will park + retry".into(), - }), - TurnOutcome::AuthFailed => bus.emit(LiveEvent::Note { - text: "/compact hit 401 — next turn will trigger the re-login flow".into(), - }), - TurnOutcome::Failed(e) => { - tracing::warn!(error = %format!("{e:#}"), "post-checkpoint compact failed"); - bus.emit(LiveEvent::Note { - text: format!("/compact after checkpoint failed: {e:#}"), - }); - } + Err(e) => error_to_turn(e), } } @@ -451,86 +377,30 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) { } } -/// Run one turn against the constant-title session, resuming it or creating -/// 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 { - 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), - } -} - -/// Run claude's built-in `/compact` slash command on the persistent -/// session. Takes the *same* params as `run_turn` because compact -/// re-initialises claude with the full session shape — same MCP -/// surface, same system prompt, same allowed-tools — so the post- -/// compact state matches a normal turn's. Only the prompt over stdin -/// differs (`/compact` vs the wake-up payload). -/// -/// Returns the same `TurnOutcome` shape as `run_turn` so callers can -/// react identically to all three failure flags (`prompt_too_long`, -/// `rate_limited`, `auth_failed`). The reactive caller bubbles any -/// non-Ok outcome up so the serve loop's normal handling (park + -/// retry on rate-limit, flip to `needs_login` on 401, etc.) kicks in; -/// the proactive post-checkpoint caller stays best-effort and only -/// emits a Note for each failure mode. +/// Operator-initiated `/compact` on the durable session (from the web UI). +/// Resume-only via [`InfiniteSession::compact`]: a missing session is a +/// harmless no-op (never mints an empty session just to compact it). Surfaces +/// the result as a Note and the usual `TurnOutcome`. pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> TurnOutcome { bus.emit(LiveEvent::Note { - text: "context overflow — running /compact on the persistent session".into(), + text: "running /compact on the session".into(), }); - // 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 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) => { + match infinite_session(bus).compact(&config, &sink).await { + Ok(()) => { bus.emit(LiveEvent::Note { - text: "no titled session to compact — skipping".into(), + text: "/compact done".into(), }); - TurnOutcome::Ok + TurnOutcome::Compacted + } + Err(e) => { + bus.emit(LiveEvent::Note { + text: format!("/compact failed: {e}"), + }); + error_to_turn(e) } - Err(e) => error_to_turn(e), - }; - match &outcome { - TurnOutcome::Ok | TurnOutcome::Compacted => bus.emit(LiveEvent::Note { - text: "/compact done".into(), - }), - TurnOutcome::PromptTooLong => bus.emit(LiveEvent::Note { - text: "/compact reported PromptTooLong — bubbling up".into(), - }), - TurnOutcome::RateLimited => bus.emit(LiveEvent::Note { - text: "/compact was rate-limited — bubbling up".into(), - }), - TurnOutcome::AuthFailed => bus.emit(LiveEvent::Note { - text: "/compact hit 401 — bubbling up for re-login flow".into(), - }), - TurnOutcome::Failed(e) => bus.emit(LiveEvent::Note { - text: format!("/compact failed: {e:#}"), - }), } - outcome } /// The constant session title for this agent. `HIVE_SESSION_TITLE` overrides diff --git a/hive-claude/README.md b/hive-claude/README.md index 5f8a3a1e..1a8525d4 100644 --- a/hive-claude/README.md +++ b/hive-claude/README.md @@ -15,19 +15,31 @@ on it rather than shelling out to `claude` by hand. ## Shape -- **`Claude::run(&config, &session, prompt, &sink)`** → `Result<(), Error>`. A - clean turn is `Ok(())`; every non-completion state is an [`Error`] variant, so - you branch with a single `match`. -- **`Claude::run_resume_or_create(&config, title, prompt, &sink)`** → - `Result<bool>`. Resumes a titled session, creating it on first use. - `Ok(true)` means a fresh session was minted. +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 }` reports what happened. +- **`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). **`Session`** — which session to attach to (`Resume` / `Create` - / `Continue` / `OneOff`). -- **`Sink`** — a trait with no-op defaults; implement the methods you care - about to observe stream events, non-JSON stdout, and stderr. Use `NoopSink` - when you only want the result. + 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. +- **`Usage`** — the minimal context signal (`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 diff --git a/hive-claude/src/config.rs b/hive-claude/src/config.rs index 09129e53..ae0cff93 100644 --- a/hive-claude/src/config.rs +++ b/hive-claude/src/config.rs @@ -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 diff --git a/hive-claude/src/driver.rs b/hive-claude/src/driver.rs index ad5f99a1..c40ea456 100644 --- a/hive-claude/src/driver.rs +++ b/hive-claude/src/driver.rs @@ -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); diff --git a/hive-claude/src/lib.rs b/hive-claude/src/lib.rs index 03934a50..068d8c5d 100644 --- a/hive-claude/src/lib.rs +++ b/hive-claude/src/lib.rs @@ -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; diff --git a/hive-claude/src/policy.rs b/hive-claude/src/policy.rs new file mode 100644 index 00000000..998b0968 --- /dev/null +++ b/hive-claude/src/policy.rs @@ -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), + })); + } +} diff --git a/hive-claude/src/session.rs b/hive-claude/src/session.rs new file mode 100644 index 00000000..ae2d7059 --- /dev/null +++ b/hive-claude/src/session.rs @@ -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); + } +} diff --git a/hive-claude/src/usage.rs b/hive-claude/src/usage.rs new file mode 100644 index 00000000..8bb83839 --- /dev/null +++ b/hive-claude/src/usage.rs @@ -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 +}