remove redundant todo md

CLAUDE.md

@@ -33,9 +33,16 @@ hand-maintained per-file tree drifts out of sync with the code.
   forge / matrix provisioning, per-container stats, and the axum
   operator dashboard (`dashboard.rs`). Largest crate.
 - **`hive-ag3nt/`** — in-container harness; one `hive` binary for every
-  agent. Turn loop (`turn.rs`), embedded MCP server (`mcp.rs`), per-agent
-  web UI (`web_ui.rs`), event + turn-stats sqlite sinks, login flow,
-  system-prompt renderer, forge-notify subscriber.
+  agent. Turn-loop *policy* layer (`turn.rs`) over the `hive-claude`
+  driver, embedded MCP server (`mcp.rs`), per-agent web UI (`web_ui/`
+  module dir), event + turn-stats sqlite sinks, login flow, system-prompt
+  renderer, forge-notify subscriber.
+- **`hive-claude/`** — reusable, app-agnostic driver for headless
+  `claude --print`: spawns the CLI, streams + classifies stream-json,
+  parses per-turn `Telemetry`, and drives a durable self-compacting
+  `InfiniteSession` (name + `SessionStore` + `CompactionPolicy`). Uses
+  `thiserror` (it's a library); the `hive-*` binaries consume it with
+  `anyhow`. See `hive-claude/README.md`.
 - **`hive-priv/`** — minimal root privileged-helper, socket-activated at
   `/run/hive/priv.sock`; performs the few root operations (bind-mount
   edits, nsenter) the unprivileged `hive-c0re` delegates to it. See

Cargo.lock

@@ -1333,6 +1333,7 @@ dependencies = [
  "axum",
  "clap",
  "futures-util",
+ "hive-claude",
  "hive-sh4re",
  "reqwest",
  "rmcp",
@@ -1392,6 +1393,16 @@ dependencies = [
  "tracing-subscriber",
 ]
 
+[[package]]
+name = "hive-claude"
+version = "0.1.0"
+dependencies = [
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+ "tokio",
+]
+
 [[package]]
 name = "hive-forge"
 version = "0.1.0"

Cargo.toml

@@ -4,6 +4,7 @@ members = [
   "hive-ag3nt",
   "hive-bash-mcp",
   "hive-c0re",
+  "hive-claude",
   "hive-forge",
   "hive-matrix-mcp",
   "hive-priv",
@@ -35,6 +36,8 @@ chrono = { version = "0.4", default-features = false, features = [
 clap = { version = "4", features = ["derive"] }
 clap_complete = "4"
 hive-sh4re = { path = "hive-sh4re" }
+hive-claude = { path = "hive-claude" }
+thiserror = "2"
 tower-http = { version = "0.6", features = ["fs"] }
 rmcp = { version = "1.7", default-features = false, features = [
   "server",

TODO.md

@@ -1,6 +0,0 @@
-# Hyperhive TODOs
-
-The backlog moved to the forge issue tracker:
-<http://localhost:3000/hyperhive/hyperhive/issues>
-
-Operator/agent trust-boundary design rationale: [`docs/boundary.md`](docs/boundary.md).

docs/turn-loop.md

@@ -137,13 +137,25 @@ if a new inbox message arrives before this turn ends"). The
 
 ```
 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 \
   --mcp-config /run/hive/claude-mcp-config.json --strict-mcp-config \
   --tools <builtins> --allowedTools <builtins+mcp>
 # wake prompt piped over stdin
 ```
 
+**Crate split.** The generic subprocess mechanics — spawning
+`claude --print`, streaming + classifying stream-json, session
+lookup/archive, and the durable-session compaction loop — live in the
+reusable **`hive-claude`** crate (`hive_claude::{Claude, InfiniteSession,
+Attach, CompactionPolicy, PercentPolicy, Telemetry, Sink, SessionStore}`;
+see `hive-claude/README.md`). `hive_ag3nt::turn` is the hyperhive **policy
+layer** on top: it builds the per-turn config from the bus, bridges the
+output stream onto the event bus (`BusSink`), and owns the compaction /
+auto-reset / retry decisions in `drive_turn`. The lib returns everything it
+parsed from a turn (usage, cost, context window, resolved model) as
+`Telemetry`, which the policy layer applies to the bus.
+
 Hive-enforced settings ship at `/etc/claude-code/managed-settings.json`
 (claude-code's canonical managed-settings path — precedence #1,
 read-only, un-overridable), wired in `nix/templates/harness-base.nix`
@@ -180,45 +192,71 @@ The effective window drives watermarks and is exposed at runtime
 via `/api/state.context_window_tokens` so the UI can show a
 percentage-of-window ctx badge.
 
-`--continue` keeps a persistent session per agent (claude stores
-sessions in `~/.claude/projects/`, which is bind-mounted
-persistently). 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.
-A one-shot `--continue` suppression is available via
-`POST /api/new-session` (or `/new-session` slash command in the
-per-agent terminal) — `Bus::take_skip_continue()` flips an
-`AtomicBool` once per turn, the next claude invocation drops
-`--continue`, every subsequent turn resumes normal behaviour.
+**Session identity — a constant title.** Every turn keys on one fixed,
+harness-owned session title (`turn::session_title()`, default
+`hive-session`, override `HIVE_SESSION_TITLE`). The durable
+`hive_claude::InfiniteSession` (built once by the serve loop via
+`turn::make_session`, then reused) `--resume <title>`s it; the *first* use
+(bootstrap, post-archive, post-purge) misses and the session re-runs the
+same prompt once with `--name <title>` to mint it. That single self-heal
+rule is the whole identity system — there is **no** scraped session-id
+file. Because the
+title is constant, `/compact` and its post-compact retry provably target
+the same session (killing the old "compact ran on a different/empty
+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
 
 claude's own in-session auto-compact is off (via the managed settings
-at `/etc/claude-code/managed-settings.json`); hyperhive owns it
-explicitly in `turn::drive_turn`. There are two triggers:
+at `/etc/claude-code/managed-settings.json`); hyperhive owns it. The
+`hive_claude::InfiniteSession` keeps the session alive across the context
+window with two triggers baked into its `run`:
 
-- **Reactive** — claude-code prints `Prompt is too long` (the
-  `PROMPT_TOO_LONG_MARKER`). The session is *already* past the context
-  window, so no turn can run on it — `drive_turn` runs `/compact`
-  straight away and retries the same wake-up prompt once. No
+- **Reactive** — claude-code prints `Prompt is too long`. The session is
+  *already* past the window, so no turn can run on it — the session
+  `/compact`s straight away and retries the same wake-up prompt once. No
   notes-checkpoint turn is possible here: the detail is gone.
-- **Proactive** — a turn finishes cleanly but the last inference's
-  context size (`Bus::last_ctx_usage().context_tokens()`) is at or
-  above a watermark. While the session is still healthy, `drive_turn`
-  injects one synthetic *notes-checkpoint* turn (`CHECKPOINT_PROMPT`
-  — "context is filling up, flush durable state into `/state` now")
-  and *then* runs `/compact`. This gives the agent a chance to
-  persist in-flight task state, decisions, and file paths before the
-  conversation detail collapses into a summary.
+- **Proactive** — a turn finishes cleanly but the last inference's context
+  size crossed the policy watermark. While the session is still healthy it
+  runs one synthetic *notes-checkpoint* turn (`CHECKPOINT_PROMPT` —
+  "context is filling up, flush durable state into `/state` now") and
+  *then* `/compact`s, so the agent can persist in-flight state before the
+  detail collapses into a summary.
 
-The compact watermark defaults to **75% of `context_window_tokens(model)`**
-(dynamically derived — 150k for haiku, 750k for sonnet/opus). Override
-with `HIVE_COMPACT_WATERMARK_TOKENS` (absolute token count); set to `0`
-to disable proactive compaction entirely (the reactive path always
-applies). The proactive path is best-effort — a failed checkpoint turn
-or `/compact` is surfaced as a `Note` but never fails the turn that
-already succeeded. The operator can also force a compaction any time
-via `/api/compact`.
+The **when** is a `hive_claude::CompactionPolicy` injected by the harness:
+`turn::make_session` builds a `PercentPolicy` that compacts once the
+model-reported context fill reaches `HIVE_COMPACT_WATERMARK_PERCENT`
+(default **75%**), falling back to `events::context_window_tokens(model)`
+for the window on turns the model didn't report one. `0` disables proactive
+compaction (the reactive path always applies). The proactive path is
+best-effort — a failed checkpoint or `/compact` never fails the turn that
+already succeeded.
+
+The operator can force a compaction any time via `POST /api/compact`. It's
+**deferred**: the handler sets `Bus::request_compact()` and returns
+immediately; the harness runs the `/compact` at the next turn boundary
+(end of the in-flight turn in `drive_turn`, or — when the agent is idle —
+in `turn::run_pending_compact` on the serve loop's next empty poll). This
+lets `/api/compact` work mid-turn instead of only when idle, without racing
+a live claude process.
 
 To disable proactive compaction for a specific agent, use the nix option:
 
@@ -226,25 +264,28 @@ To disable proactive compaction for a specific agent, use the nix option:
 hyperhive.autoCompact = false;  # default true
 ```
 
-Setting `autoCompact = false` is equivalent to `HIVE_COMPACT_WATERMARK_TOKENS=0`.
-Useful for agents running large-context models (sonnet/opus) where the 75%
-heuristic fires before the session is actually full — the reactive path
-(compact-on-overflow when the session hits the hard limit) still applies.
+Setting `autoCompact = false` sets `HIVE_COMPACT_WATERMARK_TOKENS=0`, which
+the percent resolver still honours as a disable. Useful for large-context
+models (sonnet/opus) where the 75% heuristic fires before the session is
+actually full — the reactive path (compact-on-overflow at the hard limit)
+still applies.
 
-- **Auto session-reset** — a third path that fires when both
-  conditions hold: context is ≥ a watermark (`HIVE_AUTO_RESET_WATERMARK_TOKENS`,
-  default **50% of `context_window_tokens(model)`**) AND the time since
-  the last turn exceeds the assumed prompt-cache TTL
-  (`HIVE_CACHE_TTL_SECS`, default `3600`).
-  Claude's prompt cache lives ~5 minutes; if the cache is already
-  cold, resuming with `--continue` pays the full re-upload cost of
-  the current context with no benefit over starting fresh. So:
-  `drive_turn` injects one `AUTO_RESET_CHECKPOINT_PROMPT` notes turn
-  ("flush state to files, cache is cold") then arms
-  `Bus::take_skip_continue()` for the real turn — the next turn runs
-  without `--continue`, starting a fresh session. Unlike proactive
-  compaction the session is dropped entirely, not compacted. Set
-  `HIVE_AUTO_RESET_WATERMARK_TOKENS=0` to disable.
+- **Auto session-reset** — a third path (`turn::maybe_auto_reset`,
+  pre-turn) that fires when both conditions hold: context is ≥ a watermark
+  (`HIVE_AUTO_RESET_WATERMARK_TOKENS`, default **50% of
+  `context_window_tokens(model)`**) AND the time since the last turn
+  exceeds the assumed prompt-cache TTL (`HIVE_CACHE_TTL_SECS`, default
+  `3600`). Claude's prompt cache goes cold after a while; once it's cold,
+  `--resume`-ing a large session pays the full re-upload cost with no
+  benefit over starting fresh. So `drive_turn` **archives** the current
+  session (same mechanism as the operator reset — rename `<uuid>.jsonl` →
+  `.archived`) so the next turn's `--resume <title>` misses and starts
+  fresh. Unlike proactive compaction the session is dropped entirely, not
+  compacted — and *no* preceding checkpoint turn runs, because any turn
+  before the reset would just re-warm the cache and defeat the purpose.
+  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
 back to the parent's cwd in dev), so any relative path in a tool
@@ -264,7 +305,7 @@ the embedded MCP server.
 Whenever hive-c0re starts / restarts / rebuilds a container, it
 also drops a `system` message into the agent's inbox via
 `Coordinator::kick_agent` — a one-line "you were just (re)started,
-check /state/ for your notes, --continue session is intact". The
+check /state/ for your notes, your session is intact". The
 next turn picks it up like any other inbox message.
 
 ### On-boot files
@@ -317,10 +358,16 @@ socket at `/run/hive/` once at startup:
   empty-string env vars and missing env vars round-trip the
   same way.
 
-The shared per-turn plumbing lives in `hive_ag3nt::turn::{write_mcp_config,
-write_settings, write_system_prompt, run_turn, drive_turn,
-emit_turn_end, wait_for_login, compact_session}` — the single code path
-every agent role runs through.
+The per-turn plumbing lives in `hive_ag3nt::turn`: `write_mcp_config` /
+`write_system_prompt` (on-boot files), `make_session` (builds the durable
+`InfiniteSession`, once), `drive_turn` (the policy state machine —
+reset/auto-reset, the turn, 401-retry, deferred-compact-at-turn-end),
+`run_pending_compact` (idle operator compact), `BusSink` (stream → bus +
+`Telemetry` applied via `apply_telemetry`), `emit_turn_end`, `session_title`
+/ `session_store` / `archive_session` (identity + turn-boundary reset). The
+actual claude spawn, stream classification, and the reactive/proactive
+compaction loop are in the `hive-claude` crate. Login-wait
+(`wait_for_login`) lives in `hive_ag3nt::login`.
 
 ### Reference docs (`hyperhive.docs.enable`)
 
@@ -685,9 +732,10 @@ addition to the broadcast channel and the events history. Variants:
 
 The harness flips state at the relevant transitions
 (`set_state(Thinking)` before `drive_turn`, `set_state(Idle)`
-after; `set_state(Compacting)` around `compact_session`). Exposed
-via `/api/state.turn_state` + `turn_state_since` (unix seconds);
-the agent page renders this rather than deriving from SSE events.
+after; `set_state(Compacting)` around an idle operator compact in
+`turn::run_pending_compact`). Exposed via `/api/state.turn_state` +
+`turn_state_since` (unix seconds); the agent page renders this rather
+than deriving from SSE events.
 
 ### Tool envelope
 

docs/web-ui/agent.md

@@ -261,9 +261,11 @@ Slash commands today:
 - `/cancel` — `POST /api/cancel` → host shellouts `pkill -INT
   claude`, emits a Note. Also surfaces as a `■ cancel turn`
   button in the state row while state=thinking.
-- `/compact` — `POST /api/compact` → host spawns
-  `turn::compact_session` in the background; output streams into
-  the live panel.
+- `/compact` — `POST /api/compact` → sets the deferred
+  `Bus::request_compact()` flag and returns immediately. The harness
+  runs the `/compact` at the next turn boundary (end of the in-flight
+  turn, or `turn::run_pending_compact` when idle); output streams into
+  the live panel. Works mid-turn, not only when idle.
 - `/model <name>` — `POST /api/model` flipping `Bus::set_model`.
   Takes effect on the next turn; persisted to
   `/state/hyperhive-model` so the override survives harness
@@ -450,7 +452,7 @@ token-efficiency chips derived from the bucket sums: **cache
 hit-rate** (`cache_read` over all input-side tokens) and
 **tokens/turn**. When `reminder_stats` is present (fetched via
 `ReminderRollup` RPC and merged into the snapshot in
-`web_ui.rs::api_stats`) three more chips appear: **reminders
+`web_ui/stats.rs::api_stats`) three more chips appear: **reminders
 scheduled / delivered / pending** for the window. When the
 per-session capture has data, a **first-turn ctx** chip shows the
 input tokens of the most recent fresh claude session's first turn —

docs/web-ui/shape.md

@@ -185,7 +185,7 @@ listener: read `data-confirm`, swap the button to a spinner, POST
 `application/x-www-form-urlencoded`, re-enable the button on success
 (refreshState may keep the form mounted, so we don't rely on a
 re-render), call `refreshState()`. State shapes live in
-`dashboard.rs::StateSnapshot` and `web_ui.rs::StateSnapshot` — when
+`dashboard.rs::StateSnapshot` and `web_ui/state.rs::StateSnapshot` — when
 adding state fields, plumb through the snapshot struct and the
 relevant domain module (`assets/swarm.js`, `assets/call.js`, etc.) render function.
 

hive-ag3nt/Cargo.toml

@@ -12,6 +12,7 @@ axum.workspace = true
 reqwest.workspace = true
 futures-util = "0.3"
 clap.workspace = true
+hive-claude.workspace = true
 hive-sh4re.workspace = true
 rmcp.workspace = true
 rusqlite.workspace = true

hive-ag3nt/src/bin/hive.rs

@@ -8,7 +8,6 @@ use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
-use hive_ag3nt::web_ui::TurnLock;
 
 use anyhow::Result;
 use clap::{Parser, Subcommand};
@@ -429,7 +428,6 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
         }
     }
     let files = turn::TurnFiles::prepare(socket, &label).await?;
-    let turn_lock: TurnLock = Arc::new(tokio::sync::Mutex::new(()));
     // Plugin install failures come back as a Vec<String> — route each
     // through `<parent>` via the `send_to_parent` failure-notify path.
     // The broker resolves `<parent>` per `topology::parent_of`;
@@ -454,18 +452,15 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
         login_state.clone(),
         bus.clone(),
         socket.to_path_buf(),
-        files.clone(),
-        turn_lock.clone(),
     );
     tokio::spawn(async move {
-        let (label, port, login_state, bus, socket, files, turn_lock) = web_ui_args;
-        if let Err(e) = web_ui::serve(label, port, login_state, bus, socket, files, turn_lock).await
-        {
+        let (label, port, login_state, bus, socket) = web_ui_args;
+        if let Err(e) = web_ui::serve(label, port, login_state, bus, socket).await {
             tracing::error!(error = %e, "web_ui::serve exited with error");
         }
     });
     if matches!(initial, LoginState::NeedsLogin) {
-        turn::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await;
+        login::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await;
     } else {
         // Clear any stale `hyperhive-needs-login` sentinel left over
         // from a prior boot — `online` status writes the sentinel
@@ -480,7 +475,6 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
         bus,
         stats,
         &files,
-        turn_lock,
     )
     .await
 }
@@ -502,10 +496,12 @@ async fn serve_loop<S: Surface>(
     bus: Bus,
     stats: Option<TurnStats>,
     files: &turn::TurnFiles,
-    turn_lock: TurnLock,
 ) -> Result<()> {
     tracing::info!(socket = %socket.display(), "harness serve");
     S::requeue_inflight(socket).await;
+    // The durable claude session, built once and reused for every turn +
+    // idle compaction below (it's effectively stateless).
+    let session = turn::make_session(&bus);
     // Set when a turn calls `request_next_turn` and no real work is
     // pending — the next iteration drives this synthetic message
     // in-process instead of long-polling the broker. Never
@@ -517,7 +513,13 @@ async fn serve_loop<S: Surface>(
             None => match S::recv_next(socket).await {
                 RecvOutcome::Message(first) => first,
                 RecvOutcome::Empty => {
-                    tokio::time::sleep(interval).await;
+                    // Idle: no message this poll. Service a queued operator
+                    // `/compact` here so it runs even when no turn is driving
+                    // (the in-flight case is handled at the end of drive_turn).
+                    let compacted = turn::run_pending_compact(files, &bus, &session).await;
+                    if !compacted {
+                        tokio::time::sleep(interval).await;
+                    }
                     continue;
                 }
                 RecvOutcome::TransportError => {
@@ -538,7 +540,7 @@ async fn serve_loop<S: Surface>(
                         &bus,
                         stats.as_ref(),
                         files,
-                        &turn_lock,
+                        &session,
                         graceful_stop_message(),
                     )
                     .await;
@@ -547,10 +549,11 @@ async fn serve_loop<S: Surface>(
                 }
             },
         };
-        let ctrl = handle_turn::<S>(socket, &bus, stats.as_ref(), files, &turn_lock, next).await;
+        let ctrl =
+            handle_turn::<S>(socket, &bus, stats.as_ref(), files, &session, next).await;
         if ctrl.auth_failed {
             *login_state.lock().unwrap() = LoginState::NeedsLogin;
-            turn::wait_for_login(
+            login::wait_for_login(
                 &claude_dir,
                 login_state.clone(),
                 &bus,
@@ -575,7 +578,7 @@ async fn handle_turn<S: Surface>(
     bus: &Bus,
     stats: Option<&TurnStats>,
     files: &turn::TurnFiles,
-    turn_lock: &TurnLock,
+    session: &turn::AgentSession,
     first: hive_sh4re::DeliveredMessage,
 ) -> TurnControl {
     let from = first.from;
@@ -595,10 +598,7 @@ async fn handle_turn<S: Surface>(
     let started_instant = std::time::Instant::now();
     let model_at_start = bus.model();
     let prompt = serve_common::format_wake_prompt(msg_id, &from, &body, unread, redelivered);
-    let outcome = {
-        let _guard = turn_lock.lock().await;
-        turn::drive_turn(&prompt, files, bus).await
-    };
+    let outcome = turn::drive_turn(&prompt, files, bus, session).await;
     turn::emit_turn_end(bus, &outcome);
     bus.set_state(TurnState::Idle);
     if matches!(

hive-ag3nt/src/events.rs

@@ -12,6 +12,7 @@ use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 
+use hive_claude::TokenUsage;
 use rusqlite::{Connection, params};
 use serde::{Deserialize, Serialize};
 use tokio::sync::broadcast;
@@ -432,111 +433,6 @@ impl EventStore {
     }
 }
 
-/// Token usage emitted by claude in the final `result` stream-json event.
-/// All counts are in tokens. `None` fields mean the server didn't report them.
-#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
-pub struct TokenUsage {
-    pub input_tokens: u64,
-    pub output_tokens: u64,
-    pub cache_read_input_tokens: u64,
-    pub cache_creation_input_tokens: u64,
-}
-
-impl TokenUsage {
-    /// Total context consumed this turn (input + cache reads + cache writes).
-    /// This is the per-inference context footprint that counts against the
-    /// model's `contextWindow` limit. Tracked from the last `assistant` event
-    /// in the stream-json (per-inference usage, not the cumulative `result`
-    /// event which sums across all inferences in a tool-heavy turn and can
-    /// far exceed the per-inference window).
-    #[must_use]
-    pub fn context_tokens(&self) -> u64 {
-        self.input_tokens + self.cache_read_input_tokens + self.cache_creation_input_tokens
-    }
-
-    /// Parse usage from the terminal `result` stream-json event. This is the
-    /// **cumulative** sum across every inference in the turn — useful as a
-    /// cost signal, but NOT the current context size (a tool-heavy turn
-    /// sums per-call cached prompts and easily exceeds the model window).
-    #[must_use]
-    pub fn from_stream_event(v: &serde_json::Value) -> Option<Self> {
-        if v.get("type").and_then(|t| t.as_str()) != Some("result") {
-            return None;
-        }
-        Some(Self::from_usage_obj(v.get("usage")?))
-    }
-
-    /// Parse usage from a per-inference `assistant` event's
-    /// `.message.usage` block. Each turn fires one of these for every
-    /// model call; tracking the LAST one over the turn gives the actual
-    /// conversation context size — the number to watch for compaction.
-    #[must_use]
-    pub fn from_assistant_event(v: &serde_json::Value) -> Option<Self> {
-        if v.get("type").and_then(|t| t.as_str()) != Some("assistant") {
-            return None;
-        }
-        Some(Self::from_usage_obj(v.get("message")?.get("usage")?))
-    }
-
-    fn from_usage_obj(u: &serde_json::Value) -> Self {
-        let field = |k: &str| u.get(k).and_then(serde_json::Value::as_u64).unwrap_or(0);
-        Self {
-            input_tokens: field("input_tokens"),
-            output_tokens: field("output_tokens"),
-            cache_read_input_tokens: field("cache_read_input_tokens"),
-            cache_creation_input_tokens: field("cache_creation_input_tokens"),
-        }
-    }
-
-    /// Extract the per-inference context-window limit from a `result`
-    /// stream-json event's `modelUsage` map. The API reports this as
-    /// `modelUsage.<model-name>.contextWindow`; we take the first non-zero
-    /// value across all model keys.
-    ///
-    /// Returns `None` if the event is not a `result` type or has no
-    /// `contextWindow` field. The returned value is the authoritative
-    /// per-inference active window (e.g. 200 000 for `claude-sonnet-4-6`).
-    /// It may be smaller than the full prompt-cache capacity (which can
-    /// be several million tokens via cache reads).
-    #[must_use]
-    pub fn context_window_from_result_event(v: &serde_json::Value) -> Option<u64> {
-        if v.get("type").and_then(|t| t.as_str()) != Some("result") {
-            return None;
-        }
-        let model_usage = v.get("modelUsage")?;
-        let map = model_usage.as_object()?;
-        for (_model, stats) in map {
-            if let Some(w) = stats
-                .get("contextWindow")
-                .and_then(serde_json::Value::as_u64)
-                && w > 0
-            {
-                return Some(w);
-            }
-        }
-        None
-    }
-
-    /// Extract the *resolved* model id from an `assistant` stream-json
-    /// event (`message.model`). Unlike the requested `--model` name
-    /// (which may be a short alias like `opus` or a default), the API
-    /// echoes the concrete version it actually ran on (e.g.
-    /// `claude-opus-4-8`). Recording the resolved id (not the requested
-    /// name) is what lets the ST4TS model-mix + cost rollup label the
-    /// exact version that ran. Returns `None` for non-assistant events
-    /// or ones missing `message.model`.
-    #[must_use]
-    pub fn model_from_assistant_event(v: &serde_json::Value) -> Option<String> {
-        if v.get("type").and_then(|t| t.as_str()) != Some("assistant") {
-            return None;
-        }
-        v.get("message")
-            .and_then(|m| m.get("model"))
-            .and_then(serde_json::Value::as_str)
-            .filter(|s| !s.is_empty())
-            .map(ToOwned::to_owned)
-    }
-}
 
 /// Authoritative turn-loop state. The harness owns it; the web UI
 /// reads via `/api/state` and renders. Lives alongside the bus
@@ -680,21 +576,28 @@ pub struct Bus {
     /// `container_view` can surface the status on the dashboard without
     /// a live socket call.
     rate_limited: Arc<AtomicBool>,
-    /// One-shot: next `run_claude` call drops `--continue`, starting
-    /// a fresh claude session. Set by `POST /api/new-session` from
-    /// the per-agent web UI; consumed (cleared back to false) by the
-    /// next turn. Subsequent turns resume normal `--continue`
-    /// behavior. Atomic so the consumer can take-and-clear without a
-    /// lock.
-    skip_continue_once: Arc<AtomicBool>,
+    /// One-shot: operator asked to reset the claude session (`POST
+    /// /api/new-session`). Consumed at the *next* turn boundary by
+    /// `turn::drive_turn`, which archives the current session so the turn
+    /// starts fresh. Deferred (not applied in the handler) so the archive
+    /// never races a claude process mid-write — one claude per container,
+    /// serialized by the serve loop, so the turn boundary is the only point
+    /// where no session file is open.
+    session_reset_pending: Arc<AtomicBool>,
+    /// One-shot: run `/compact` after the next turn ends. Consumed at the end
+    /// of the current/next turn by `turn::drive_turn`. Deferring to the turn
+    /// boundary keeps compaction from racing a live claude process mid-turn.
+    compact_pending: Arc<AtomicBool>,
     /// 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
     /// every `turn_stats` row until the next fresh session. `None` before
     /// the first fresh turn or when the stats db is unavailable.
     session_id: Arc<Mutex<Option<i64>>>,
-    /// One-shot: `run_claude` flips this true when it suppresses
-    /// `--continue` (a fresh session). The bin loop takes-and-clears it
-    /// after the turn to decide whether to mint a new `sessions` row.
+    /// One-shot: `run_claude` flips this true when it creates a fresh claude
+    /// session (`--name <title>` on a title miss). The bin loop takes-and-
+    /// 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>,
     /// Per-turn tool-call counter. Reset by the bin loop between
     /// turns via `take_tool_calls`. Populated by `observe_stream` as
@@ -773,7 +676,8 @@ impl Bus {
             last_ctx_usage: Arc::new(Mutex::new(None)),
             last_cost_usage: Arc::new(Mutex::new(None)),
             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)),
+            compact_pending: Arc::new(AtomicBool::new(false)),
             session_id: Arc::new(Mutex::new(None)),
             fresh_session: Arc::new(AtomicBool::new(false)),
             tool_calls: Arc::new(Mutex::new(std::collections::HashMap::new())),
@@ -797,23 +701,37 @@ impl Bus {
         self.event_seq.fetch_add(1, Ordering::SeqCst) + 1
     }
 
-    /// Arm the one-shot: the next claude invocation will run without
-    /// `--continue`, dropping any prior session context. Idempotent
-    /// — calling twice in a row before the next turn still consumes
-    /// to a single fresh-start.
-    pub fn request_new_session(&self) {
-        self.skip_continue_once.store(true, Ordering::SeqCst);
+    /// Request a session reset (operator `POST /api/new-session`). Deferred:
+    /// the flag is consumed at the next turn boundary by `drive_turn`, which
+    /// archives the current session so no claude process is mid-write when the
+    /// file is renamed. Idempotent — two clicks before the next turn still
+    /// archive once.
+    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
-    /// run claude without `--continue` for this turn.
+    /// Take + clear the session-reset one-shot. Returns true iff `drive_turn`
+    /// should archive the current session before this turn.
     #[must_use]
-    pub fn take_skip_continue(&self) -> bool {
-        self.skip_continue_once.swap(false, Ordering::SeqCst)
+    pub fn take_session_reset(&self) -> bool {
+        self.session_reset_pending.swap(false, Ordering::SeqCst)
+    }
+
+    /// Request a compaction after the next turn ends (deferred to the turn
+    /// boundary). Idempotent.
+    pub fn request_compact(&self) {
+        self.compact_pending.store(true, Ordering::SeqCst);
+    }
+
+    /// Take + clear the compact one-shot. Returns true iff `drive_turn` should
+    /// compact at the end of this turn.
+    #[must_use]
+    pub fn take_compact(&self) -> bool {
+        self.compact_pending.swap(false, Ordering::SeqCst)
     }
 
     /// 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) {
         self.fresh_session.store(true, Ordering::SeqCst);
     }
@@ -1203,8 +1121,8 @@ impl Default for Bus {
 #[cfg(test)]
 mod tests {
     use super::{
-        BusEvent, DEFAULT_EFFORT, EFFORT_LEVELS, LiveEvent, StoredEvent, TokenUsage,
-        forge_cursor_from_json, is_valid_effort,
+        BusEvent, DEFAULT_EFFORT, EFFORT_LEVELS, LiveEvent, StoredEvent, forge_cursor_from_json,
+        is_valid_effort,
     };
     use serde_json::json;
 
@@ -1283,29 +1201,4 @@ mod tests {
         assert!(EFFORT_LEVELS.contains(&DEFAULT_EFFORT));
         assert_eq!(DEFAULT_EFFORT, "medium");
     }
-
-    #[test]
-    fn resolved_model_from_assistant_event() {
-        let v = json!({
-            "type": "assistant",
-            "message": { "model": "claude-opus-4-8", "role": "assistant" }
-        });
-        assert_eq!(
-            TokenUsage::model_from_assistant_event(&v),
-            Some("claude-opus-4-8".to_owned())
-        );
-    }
-
-    #[test]
-    fn resolved_model_ignores_non_assistant_and_missing() {
-        // Wrong event type.
-        let result = json!({ "type": "result", "message": { "model": "claude-opus-4-8" } });
-        assert_eq!(TokenUsage::model_from_assistant_event(&result), None);
-        // Assistant event missing message.model.
-        let no_model = json!({ "type": "assistant", "message": { "role": "assistant" } });
-        assert_eq!(TokenUsage::model_from_assistant_event(&no_model), None);
-        // Empty model string is treated as absent.
-        let empty = json!({ "type": "assistant", "message": { "model": "" } });
-        assert_eq!(TokenUsage::model_from_assistant_event(&empty), None);
-    }
 }

hive-ag3nt/src/login.rs

@@ -9,6 +9,10 @@
 //! exact layout is locked in.
 
 use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use crate::events::Bus;
 
 /// Returns the Claude credentials directory for this agent. Delegates
 /// to `paths::claude_dir`, which reads `$HOME/.claude`. The service
@@ -56,3 +60,216 @@ impl LoginState {
         }
     }
 }
+
+/// Block until the bound `~/.claude/` dir contains a session that
+/// post-dates this call, polling on a `poll_ms` interval (min 2s).
+/// Flips `state` to `Online` when login lands; caller resumes its
+/// serve loop. Snapshots the dir at entry and only resumes when the
+/// snapshot advances (mtime OR file-count change), avoiding the
+/// infinite-401 loop a bare-existence check would produce when stale
+/// credentials are already on disk. Mtime-snapshot resumption rationale
+/// and `DirSnapshot` two-axis design: see
+/// [`docs/turn-loop.md::The loop`](../../docs/turn-loop.md).
+///
+/// # Panics
+///
+/// Panics if the internal login-state lock is poisoned.
+pub async fn wait_for_login(
+    claude_dir: &Path,
+    state: Arc<Mutex<LoginState>>,
+    bus: &Bus,
+    poll_ms: u64,
+) {
+    tracing::warn!(
+        claude_dir = %claude_dir.display(),
+        "no claude session — staying in partial-run mode (web UI only)"
+    );
+    // Announce `needs_login_idle` to the bus so the sentinel file
+    // (`{state_dir}/hyperhive-needs-login`) gets written on every entry
+    // path — cold-boot, 401-mid-turn, and `/api/logout`. The host's
+    // `auth_failed_sentinel` reads that file to surface `needs_login`
+    // on the dashboard. Idempotent — `emit_status` is a `write` on a
+    // small empty file, so re-entering this function after a transient
+    // operator action is a no-op for the on-disk state.
+    bus.emit_status("needs_login_idle");
+    let snapshot = snapshot_dir(claude_dir);
+    let probe = Duration::from_millis(poll_ms.max(2000));
+    loop {
+        tokio::time::sleep(probe).await;
+        if session_refreshed(snapshot, snapshot_dir(claude_dir)) {
+            tracing::info!("claude session refreshed — entering turn loop");
+            *state.lock().unwrap() = LoginState::Online;
+            bus.emit_status("online");
+            return;
+        }
+    }
+}
+
+/// Snapshot of the credentials dir at a point in time: number of
+/// regular files + newest `mtime` across them. The two axes are both
+/// load-bearing for `wait_for_login`'s refresh check (`session_refreshed`):
+/// mtime catches the common case (re-login overwrites an existing
+/// credentials file in-place), `file_count` catches the pathological case
+/// where `meta.modified()` errors on every file (exotic fs, NFS quirks)
+/// so the mtime axis stays `None` forever but new files still trigger a
+/// resume. Defaults to `{0, None}` on `read_dir` failure (missing or
+/// unreadable dir) — `wait_for_login` then resumes when files first
+/// appear.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+struct DirSnapshot {
+    file_count: usize,
+    newest_mtime: Option<std::time::SystemTime>,
+}
+
+fn snapshot_dir(dir: &Path) -> DirSnapshot {
+    let Ok(entries) = std::fs::read_dir(dir) else {
+        return DirSnapshot::default();
+    };
+    let mut snap = DirSnapshot::default();
+    for entry in entries.flatten() {
+        if !entry.file_type().is_ok_and(|t| t.is_file()) {
+            continue;
+        }
+        snap.file_count += 1;
+        let Ok(meta) = entry.metadata() else { continue };
+        let Ok(mtime) = meta.modified() else { continue };
+        if snap.newest_mtime.is_none_or(|cur| mtime > cur) {
+            snap.newest_mtime = Some(mtime);
+        }
+    }
+    snap
+}
+
+/// Has the credentials dir been written since `prev`? Used as the
+/// exit condition for `wait_for_login`:
+///
+/// - `file_count` changed → something was added or removed, treat as
+///   refresh (covers the "all files have unreadable mtime" edge case).
+/// - `newest_mtime` advanced → existing file was rewritten in place
+///   (the common claude re-login path).
+/// - prev had no mtime (empty or all-unreadable) and now has one →
+///   first useful signal we've seen, treat as refresh.
+fn session_refreshed(prev: DirSnapshot, now: DirSnapshot) -> bool {
+    if now.file_count != prev.file_count {
+        return true;
+    }
+    match (prev.newest_mtime, now.newest_mtime) {
+        (None, Some(_)) => true,
+        (Some(p), Some(n)) => n > p,
+        _ => false,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::fs;
+    use std::time::{Duration, SystemTime};
+
+    use super::{DirSnapshot, session_refreshed, snapshot_dir};
+
+    #[test]
+    fn snapshot_dir_empty_dir_is_default() {
+        let dir = tempfile::tempdir().unwrap();
+        let snap = snapshot_dir(dir.path());
+        assert_eq!(snap.file_count, 0);
+        assert!(snap.newest_mtime.is_none());
+    }
+
+    #[test]
+    fn snapshot_dir_missing_dir_is_default() {
+        // Defensive: a nonexistent dir must NOT panic. Bind mounts that
+        // disappear mid-poll (host purge during operator intervention)
+        // would otherwise crash the harness.
+        let missing = tempfile::tempdir()
+            .unwrap()
+            .path()
+            .join("never-created-subdir");
+        let snap = snapshot_dir(&missing);
+        assert_eq!(snap, DirSnapshot::default());
+    }
+
+    #[test]
+    fn snapshot_dir_picks_latest_mtime_and_counts_files() {
+        let dir = tempfile::tempdir().unwrap();
+        fs::write(dir.path().join("old.json"), b"{}").unwrap();
+        // Sleep so the second file's mtime is strictly greater than
+        // the first on filesystems with low timestamp resolution.
+        std::thread::sleep(Duration::from_millis(20));
+        let newer_path = dir.path().join("newer.json");
+        fs::write(&newer_path, b"{}").unwrap();
+        let snap = snapshot_dir(dir.path());
+        assert_eq!(snap.file_count, 2);
+        let newer_meta = fs::metadata(&newer_path).unwrap().modified().unwrap();
+        assert_eq!(snap.newest_mtime, Some(newer_meta));
+    }
+
+    #[test]
+    fn session_refreshed_first_login_flips_on_any_file() {
+        // Empty-dir snapshot → any file appearing means a fresh
+        // login landed. First-time login semantics.
+        let dir = tempfile::tempdir().unwrap();
+        let snapshot = snapshot_dir(dir.path());
+        assert!(!session_refreshed(snapshot, snapshot_dir(dir.path())));
+        fs::write(dir.path().join("credentials.json"), b"{}").unwrap();
+        assert!(session_refreshed(snapshot, snapshot_dir(dir.path())));
+    }
+
+    #[test]
+    fn session_refreshed_stale_creds_dont_flip_immediately() {
+        // Stale credentials.json already exists at entry; wait_for_login
+        // must NOT immediately return — it would loop straight into
+        // another 401-failing turn.
+        let dir = tempfile::tempdir().unwrap();
+        fs::write(dir.path().join("credentials.json"), b"{}").unwrap();
+        let snapshot = snapshot_dir(dir.path());
+        assert_eq!(snapshot.file_count, 1);
+        // No change to the file → loop must NOT exit.
+        assert!(!session_refreshed(snapshot, snapshot_dir(dir.path())));
+    }
+
+    #[test]
+    fn session_refreshed_after_creds_rewrite_flips() {
+        // After the stale-creds snapshot, the operator's `/login/code`
+        // flow lands a refreshed credentials file — its mtime bumps
+        // strictly past the snapshot and wait_for_login resumes.
+        let dir = tempfile::tempdir().unwrap();
+        fs::write(dir.path().join("credentials.json"), b"{}").unwrap();
+        let snapshot = snapshot_dir(dir.path());
+        std::thread::sleep(Duration::from_millis(20));
+        fs::write(dir.path().join("credentials.json"), b"{\"v\":2}").unwrap();
+        assert!(session_refreshed(snapshot, snapshot_dir(dir.path())));
+    }
+
+    #[test]
+    fn session_refreshed_snapshot_with_future_mtime_doesnt_flip() {
+        // Defensive: a snapshot set to a future timestamp (e.g. clock
+        // skew between snapshot and probe) must keep waiting until a
+        // file's mtime actually exceeds it, not return on first poll.
+        let dir = tempfile::tempdir().unwrap();
+        fs::write(dir.path().join("credentials.json"), b"{}").unwrap();
+        let snapshot = DirSnapshot {
+            file_count: 1,
+            newest_mtime: Some(SystemTime::now() + Duration::from_hours(1)),
+        };
+        assert!(!session_refreshed(snapshot, snapshot_dir(dir.path())));
+    }
+
+    #[test]
+    fn session_refreshed_count_change_flips_when_mtime_unreadable() {
+        // Defensive: if all files have unreadable `meta.modified()`
+        // (exotic fs / NFS), newest_mtime stays `None` forever — but
+        // file_count axis still catches new files appearing. Simulated
+        // here by forging a snapshot with file_count=1 + no mtime, then
+        // writing a second file.
+        let dir = tempfile::tempdir().unwrap();
+        fs::write(dir.path().join("a"), b"{}").unwrap();
+        let forged = DirSnapshot {
+            file_count: 1,
+            newest_mtime: None,
+        };
+        fs::write(dir.path().join("b"), b"{}").unwrap();
+        // Real snapshot has file_count=2, so refresh fires even
+        // though the mtime axis would be inconclusive.
+        assert!(session_refreshed(forged, snapshot_dir(dir.path())));
+    }
+}

hive-ag3nt/src/mcp.rs

@@ -2330,8 +2330,10 @@ pub fn render_claude_config(agent_binary: &str, socket: &std::path::Path) -> Str
 }
 
 #[cfg(test)]
-mod recv_hint_tests {
+mod tests {
     use super::{IDLE_WAIT_HINT, SocketReply, format_recv};
+    use super::{SERVER_NAME, allowed_mcp_tools};
+    use hive_sh4re::ToolGroup;
 
     #[test]
     fn empty_recv_after_wait_appends_idle_hint() {
@@ -2345,12 +2347,6 @@ mod recv_hint_tests {
         let out = format_recv(Ok(SocketReply::Messages(vec![])), false);
         assert_eq!(out, "(empty)");
     }
-}
-
-#[cfg(test)]
-mod allowed_tools_tests {
-    use super::{SERVER_NAME, allowed_mcp_tools};
-    use hive_sh4re::ToolGroup;
 
     fn qualified(tool: &str) -> String {
         format!("mcp__{SERVER_NAME}__{tool}")

hive-ag3nt/src/turn_stats.rs

@@ -268,8 +268,8 @@ impl TurnStats {
     pub fn last_usage(
         &self,
     ) -> (
-        Option<crate::events::TokenUsage>,
-        Option<crate::events::TokenUsage>,
+        Option<hive_claude::TokenUsage>,
+        Option<hive_claude::TokenUsage>,
     ) {
         let conn = self.inner.lock().unwrap();
         conn.query_row(
@@ -285,19 +285,19 @@ impl TurnStats {
                 let g = |i: usize| -> rusqlite::Result<u64> {
                     Ok(u64::try_from(row.get::<_, i64>(i)?).unwrap_or(0))
                 };
-                let cost = crate::events::TokenUsage {
+                let cost = hive_claude::TokenUsage {
                     input_tokens: g(0)?,
                     output_tokens: g(1)?,
                     cache_read_input_tokens: g(2)?,
                     cache_creation_input_tokens: g(3)?,
                 };
-                let last = crate::events::TokenUsage {
+                let last = hive_claude::TokenUsage {
                     input_tokens: g(4)?,
                     output_tokens: g(5)?,
                     cache_read_input_tokens: g(6)?,
                     cache_creation_input_tokens: g(7)?,
                 };
-                let ctx = if last == crate::events::TokenUsage::default() {
+                let ctx = if last == hive_claude::TokenUsage::default() {
                     None
                 } else {
                     Some(last)

hive-ag3nt/src/web_ui/actions.rs

@@ -0,0 +1,172 @@
+//! Operator action POST handlers (send, cancel, compact, model, effort, reset).
+
+use axum::{
+    Form,
+    extract::State,
+    http::StatusCode,
+    response::{IntoResponse, Response},
+};
+use serde::Deserialize;
+
+use crate::client;
+
+use super::{AppState, SOCKET_FETCH_TIMEOUT, error_response};
+
+#[derive(Deserialize)]
+pub(super) struct SendForm {
+    body: String,
+}
+
+pub(super) async fn post_send(
+    State(state): State<AppState>,
+    Form(form): Form<SendForm>,
+) -> Response {
+    let body = form.body.trim().to_owned();
+    if body.is_empty() {
+        return error_response(StatusCode::BAD_REQUEST, "send: `body` required");
+    }
+    let result = match tokio::time::timeout(
+        SOCKET_FETCH_TIMEOUT,
+        client::request::<_, hive_sh4re::Response>(
+            &state.socket,
+            &hive_sh4re::Request::OperatorMsg { body },
+        ),
+    )
+    .await
+    {
+        Ok(Ok(hive_sh4re::Response::Ok)) => Ok(()),
+        Ok(Ok(hive_sh4re::Response::Err { message })) => Err(message),
+        Ok(Ok(other)) => Err(format!("unexpected response: {other:?}")),
+        Ok(Err(e)) => Err(format!("transport: {e:#}")),
+        Err(_) => Err("timed out — hive-c0re busy, retry".to_owned()),
+    };
+    match result {
+        // 200 instead of 303 → the client doesn't refetch /api/state.
+        // The operator message becomes a broker `Sent` (already shown
+        // server-side in the dashboard); on the agent side, the
+        // resulting `TurnStart` SSE event drives the terminal + the
+        // inbox row gets consumed by the time `TurnEnd` fires the
+        // existing turn-end refresh.
+        Ok(()) => (axum::http::StatusCode::OK, "ok").into_response(),
+        Err(e) => error_response(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            &format!("send failed: {e}"),
+        ),
+    }
+}
+
+pub(super) async fn post_cancel_turn(State(state): State<AppState>) -> Response {
+    let out = tokio::process::Command::new("pkill")
+        .args(["-INT", "claude"])
+        .output()
+        .await;
+    let note = match out {
+        Ok(o) if o.status.success() => "operator: /cancel — sent SIGINT to claude".to_owned(),
+        Ok(o) if o.status.code() == Some(1) => {
+            "operator: /cancel — no claude process to interrupt".to_owned()
+        }
+        Ok(o) => format!(
+            "operator: /cancel — pkill exited {} stderr={}",
+            o.status,
+            String::from_utf8_lossy(&o.stderr).trim()
+        ),
+        Err(e) => format!("operator: /cancel — pkill failed: {e}"),
+    };
+    state
+        .bus
+        .emit(crate::events::LiveEvent::Note { text: note });
+    (axum::http::StatusCode::OK, "ok").into_response()
+}
+
+/// Operator-initiated `/compact`. Deferred: sets the `compact_pending` flag
+/// that `turn::drive_turn` consumes at the end of the current/next turn, so it
+/// works while a turn is in flight (a mid-turn compaction would race the live
+/// claude process) rather than only when the agent is idle. Returns 200
+/// immediately; the compaction stream lands in the live panel when it runs.
+pub(super) async fn post_compact(State(state): State<AppState>) -> Response {
+    state.bus.request_compact();
+    state.bus.emit(crate::events::LiveEvent::Note {
+        text: "operator: /compact queued — runs at the end of the current turn".into(),
+    });
+    (axum::http::StatusCode::OK, "ok").into_response()
+}
+
+/// Request a session reset. The current session is archived (its backing
+/// `<uuid>.jsonl` renamed out of claude's resolution glob) at the next turn
+/// boundary, so the following turn's `--resume` misses and self-heals into a
+/// freshly-named session. History is preserved on disk, not deleted.
+///
+/// Deferred (a one-shot flag consumed by `drive_turn`) rather than applied
+/// here: renaming the session file while a claude turn is mid-write would
+/// race the live process. Between turns there is no open session file (one
+/// claude per container, serialized by the serve loop), so the archive is
+/// safe there. Useful when the session-resume context is poisoned (claude
+/// went off the rails, hit an unrecoverable refusal, etc.) and a full reset
+/// is cheaper than asking claude to forget mid-stream.
+pub(super) async fn post_new_session(State(state): State<AppState>) -> Response {
+    state.bus.request_session_reset();
+    state.bus.emit(crate::events::LiveEvent::Note {
+        text: "operator: session reset queued — takes effect at the next turn".into(),
+    });
+    (axum::http::StatusCode::OK, "ok").into_response()
+}
+
+#[derive(Deserialize)]
+pub(super) struct ModelForm {
+    model: String,
+}
+
+/// Switch the model for future turns. The current turn (if any)
+/// keeps its model; `/model <name>` applies starting with the next
+/// `recv` cycle. Empty / whitespace-only inputs are rejected. No
+/// claude-side validation — we just hand the string through to
+/// `claude --model <name>`; an unknown model surfaces as a turn
+/// failure in the live panel and the operator can revert.
+pub(super) async fn post_set_model(
+    State(state): State<AppState>,
+    Form(form): Form<ModelForm>,
+) -> Response {
+    let name = form.model.trim();
+    if name.is_empty() {
+        return error_response(StatusCode::BAD_REQUEST, "model: name required");
+    }
+    state.bus.set_model(name);
+    state.bus.emit(crate::events::LiveEvent::Note {
+        text: format!("operator: /model — claude model set to '{name}' for future turns"),
+    });
+    tracing::info!(%name, "operator set model");
+    (axum::http::StatusCode::OK, "ok").into_response()
+}
+
+#[derive(Deserialize)]
+pub(super) struct EffortForm {
+    effort: String,
+}
+
+/// Switch the claude effort level for future sessions. Operator-only
+/// (the dashboard picker POSTs here through the gateway). Validated
+/// server-side against [`crate::events::EFFORT_LEVELS`] — an out-of-set
+/// value is rejected rather than handed to `claude --effort`, since an
+/// unknown level would fail every subsequent launch. Applies on the next
+/// session start (no mid-session swap).
+pub(super) async fn post_set_effort(
+    State(state): State<AppState>,
+    Form(form): Form<EffortForm>,
+) -> Response {
+    let level = form.effort.trim();
+    if !crate::events::is_valid_effort(level) {
+        return error_response(
+            StatusCode::BAD_REQUEST,
+            &format!(
+                "effort: level must be one of {}",
+                crate::events::EFFORT_LEVELS.join(", ")
+            ),
+        );
+    }
+    state.bus.set_effort(level);
+    state.bus.emit(crate::events::LiveEvent::Note {
+        text: format!("operator: /effort — claude effort set to '{level}' for future sessions"),
+    });
+    tracing::info!(%level, "operator set effort");
+    (axum::http::StatusCode::OK, "ok").into_response()
+}

hive-ag3nt/src/web_ui/auth.rs

@@ -0,0 +1,141 @@
+//! Login / logout flow handlers (`/login/*`, `/api/logout`).
+
+use std::sync::Arc;
+
+use axum::{
+    Form,
+    extract::State,
+    http::StatusCode,
+    response::{IntoResponse, Response},
+};
+use serde::Deserialize;
+
+use crate::login::LoginState;
+use crate::login_session::{LoginSession, drop_if_finished};
+
+use super::{AppState, error_response};
+
+pub(super) async fn post_login_start(State(state): State<AppState>) -> Response {
+    drop_if_finished(&state.session);
+    {
+        let guard = state.session.lock().unwrap();
+        if guard.is_some() {
+            return (axum::http::StatusCode::OK, "ok").into_response();
+        }
+    }
+    match LoginSession::start() {
+        Ok(session) => {
+            *state.session.lock().unwrap() = Some(Arc::new(session));
+            // Flip status from needs_login_idle → needs_login_in_progress
+            // so the web UI's badge + polling kick in (polling is still
+            // the right tool for the streaming session output during
+            // the login flow itself; events drop the poll for
+            // *everything else*).
+            state.bus.emit_status("needs_login_in_progress");
+            (axum::http::StatusCode::OK, "ok").into_response()
+        }
+        Err(e) => error_response(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            &format!("login start failed: {e:#}"),
+        ),
+    }
+}
+
+#[derive(Deserialize)]
+pub(super) struct CodeForm {
+    code: String,
+}
+
+pub(super) async fn post_login_code(
+    State(state): State<AppState>,
+    Form(form): Form<CodeForm>,
+) -> Response {
+    let session = state.session.lock().unwrap().clone();
+    let Some(session) = session else {
+        return error_response(StatusCode::CONFLICT, "no login session running");
+    };
+    if let Err(e) = session.submit_code(&form.code).await {
+        return error_response(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            &format!("submit code failed: {e:#}"),
+        );
+    }
+    (axum::http::StatusCode::OK, "ok").into_response()
+}
+
+pub(super) async fn post_login_cancel(State(state): State<AppState>) -> Response {
+    let session = state.session.lock().unwrap().take();
+    if let Some(session) = session {
+        session.close_stdin().await;
+        session.kill();
+    }
+    // Back to needs_login_idle (LoginState unchanged, session gone).
+    state.bus.emit_status("needs_login_idle");
+    (axum::http::StatusCode::OK, "ok").into_response()
+}
+
+/// OAuth credential filenames inside `paths::claude_dir()`. Wiping
+/// only these (and not the rest of `~/.claude/`) preserves session
+/// history so `claude --continue` keeps working after a fresh login.
+/// Rationale + the previous wholesale-wipe shape we replaced live in
+/// [`docs/web-ui/agent.md::Per-agent endpoints`](../../../docs/web-ui/agent.md)
+/// (the `/api/logout` bullet).
+const CRED_FILE_NAMES: &[&str] = &[".credentials.json", "mcp-needs-auth-cache.json"];
+
+/// Operator-driven `/logout`: SIGINT claude, delete the credential
+/// files in `CRED_FILE_NAMES`, flip `LoginState::NeedsLogin`. The
+/// turn loop's next iteration parks into `wait_for_login` which
+/// resumes when a fresh credentials file appears via `/login/code`.
+/// Always returns 200 with a body describing what happened. See
+/// [`docs/web-ui/agent.md::Per-agent endpoints`](../../../docs/web-ui/agent.md)
+/// (the `/api/logout` bullet) for the three-step rationale +
+/// preservation invariants.
+pub(super) async fn post_logout(State(state): State<AppState>) -> Response {
+    // Step 1: SIGINT claude (best-effort, matches `post_cancel_turn`).
+    let _ = tokio::process::Command::new("pkill")
+        .args(["-INT", "claude"])
+        .output()
+        .await;
+    // Step 2: delete OAuth credential files only — preserve session
+    // history files alongside them.
+    let dir = crate::paths::claude_dir();
+    let mut warnings: Vec<String> = Vec::new();
+    let mut wiped: Vec<&str> = Vec::new();
+    for name in CRED_FILE_NAMES {
+        let path = dir.join(name);
+        match tokio::fs::remove_file(&path).await {
+            Ok(()) => wiped.push(name),
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
+                // Already gone — operator clicked /logout while
+                // already logged out, or the file simply didn't exist
+                // for this agent. Idempotent.
+            }
+            Err(e) => warnings.push(format!("{name}: {e}")),
+        }
+    }
+    let wipe_summary = if wiped.is_empty() {
+        "no credential files present (already logged out)".to_owned()
+    } else {
+        format!("wiped {}", wiped.join(", "))
+    };
+    let warn_suffix = if warnings.is_empty() {
+        String::new()
+    } else {
+        format!(" (warnings: {})", warnings.join("; "))
+    };
+    // Step 3: flip LoginState + emit Note. Turn loop sees the flip on
+    // its next iteration and parks into wait_for_login.
+    *state.login.lock().unwrap() = LoginState::NeedsLogin;
+    state.bus.emit(crate::events::LiveEvent::Note {
+        text: format!(
+            "operator: /logout — {wipe_summary} in {}{warn_suffix}",
+            dir.display()
+        ),
+    });
+    state.bus.emit_status("needs_login_idle");
+    (
+        axum::http::StatusCode::OK,
+        format!("ok: {wipe_summary} in {}{warn_suffix}", dir.display()),
+    )
+        .into_response()
+}

hive-ag3nt/src/web_ui/mod.rs

@@ -0,0 +1,269 @@
+//! Per-container HTTP UI. SPA shape: `GET /` returns a static shell;
+//! `GET /static/*` serves CSS + JS; `GET /api/state` returns the page
+//! state as JSON; the JS app renders. Live events stream on
+//! `/events/stream`. Action POSTs (`/send`, `/login/*`) return either a
+//! 303 Redirect (for browsers that submit the form normally) or just
+//! 200 OK — the JS app re-fetches `/api/state` afterwards.
+//!
+//! Handlers are split by concern into the submodules below; this file owns the
+//! shared [`AppState`], the listener + router wiring in [`serve`], and a couple
+//! of small shared helpers ([`error_response`], [`SOCKET_FETCH_TIMEOUT`]).
+
+mod actions;
+mod auth;
+mod screen;
+mod state;
+mod stats;
+mod stream;
+
+use std::net::SocketAddr;
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex};
+
+use anyhow::{Context, Result};
+use axum::{
+    Router,
+    http::StatusCode,
+    response::{IntoResponse, Response},
+    routing::{get, post},
+};
+use tower_http::services::ServeDir;
+
+use crate::events::Bus;
+use crate::login::LoginState;
+use crate::login_session::LoginSession;
+
+/// Deadline for broker-backed fetches on web-UI request paths. The
+/// page's critical fields (status, turn state, usage) are all
+/// in-memory; a busy or stalled hive-c0re must degrade the
+/// socket-backed extras (inbox rows, loose ends, reminder stats)
+/// instead of hanging the whole response — an unbounded await here is
+/// what let `/api/state` stall long enough to bork the terminal.
+const SOCKET_FETCH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
+
+/// Live login state for the web UI. The harness updates this in place as it
+/// transitions between `NeedsLogin` and `Online`; the UI reads on each
+/// render.
+pub type LoginStateCell = Arc<Mutex<LoginState>>;
+
+#[derive(Clone)]
+struct AppState {
+    label: String,
+    login: LoginStateCell,
+    session: Arc<Mutex<Option<Arc<LoginSession>>>>,
+    bus: Bus,
+    socket: PathBuf,
+    /// VNC port from the `HIVE_GUI_VNC_PORT` env var at startup.
+    /// `None` when unset (gui not enabled for this agent).
+    gui_vnc_port: Option<u16>,
+}
+
+/// Bind the per-container web listener and serve the SPA.
+///
+/// `HIVE_WEB_SOCKET` opt-in selects unix-socket vs TCP binding; the
+/// dual-mode transition + gateway-side consumer live in
+/// [`docs/web-ui/shape.md::Listener bind`](../../../docs/web-ui/shape.md) and
+/// [`docs/gateway.md::Per-agent unix-socket upstream`](../../../docs/gateway.md).
+///
+/// # Errors
+///
+/// Returns an error if neither the TCP listener (default) nor the
+/// unix-socket bind (`HIVE_WEB_SOCKET`, if set) can be acquired, or
+/// if `HIVE_STATIC_DIR` is missing.
+pub async fn serve(
+    label: String,
+    port: u16,
+    login: LoginStateCell,
+    bus: Bus,
+    socket: PathBuf,
+) -> Result<()> {
+    let gui_vnc_port = read_gui_vnc_port();
+    let static_dir: PathBuf = std::env::var_os("HIVE_STATIC_DIR")
+        .map(PathBuf::from)
+        .context(
+            "HIVE_STATIC_DIR env var not set — point it at the merged \
+             per-agent dist (see hyperhive.frontend.mergedDist in nix)",
+        )?;
+    if !static_dir.is_dir() {
+        anyhow::bail!(
+            "HIVE_STATIC_DIR ({}) is not a directory",
+            static_dir.display()
+        );
+    }
+    tracing::info!(static_dir = %static_dir.display(), "web UI static dir resolved");
+    let state = AppState {
+        label,
+        login,
+        session: Arc::new(Mutex::new(None)),
+        bus,
+        socket,
+        gui_vnc_port,
+    };
+    let app = Router::new()
+        .route("/api/state", get(state::api_state))
+        .route("/api/dashboard-state", get(state::api_dashboard_state))
+        .route("/events/stream", get(stream::events_stream))
+        .route("/events/history", get(stream::events_history))
+        .route("/send", post(actions::post_send))
+        .route("/login/start", post(auth::post_login_start))
+        .route("/login/code", post(auth::post_login_code))
+        .route("/login/cancel", post(auth::post_login_cancel))
+        .route("/api/cancel", post(actions::post_cancel_turn))
+        .route("/api/compact", post(actions::post_compact))
+        .route("/api/model", post(actions::post_set_model))
+        .route("/api/effort", post(actions::post_set_effort))
+        .route("/api/new-session", post(actions::post_new_session))
+        .route("/api/logout", post(auth::post_logout))
+        .route("/api/loose-ends", get(stats::api_loose_ends))
+        .route("/api/bash-tasks", get(stats::api_bash_tasks))
+        .route("/api/stats", get(stats::api_stats))
+        .route("/screen/ws", get(screen::screen_ws))
+        .route("/icon", get(screen::serve_icon))
+        // Anything else (`/`, `/stats`, `/screen`, `/static/*`)
+        // falls through to the merged dist. ServeDir auto-appends
+        // `.html` when the URL is a bare path that matches a file
+        // (so `/stats` → `dist/stats.html`, `/screen` → `dist/
+        // screen.html`). Per-agent `extraFiles` additions are
+        // already layered into this same directory (see
+        // hyperhive.frontend.mergedDist in nix).
+        .fallback_service(ServeDir::new(&static_dir))
+        .with_state(state);
+    // `HIVE_WEB_SOCKET` opt-in: when set + non-empty, bind a
+    // `UnixListener` at the given path. Empty string treated as
+    // unset so a stray `HIVE_WEB_SOCKET=` doesn't trap us into an
+    // un-bindable empty path. Falls through to the TCP path below
+    // otherwise. See docs/gateway.md::Per-agent unix-socket upstream
+    // for the gateway-side consumer.
+    if let Some(socket_path) = std::env::var_os("HIVE_WEB_SOCKET")
+        && !socket_path.is_empty()
+    {
+        let path = PathBuf::from(socket_path);
+        let listener = bind_unix(&path)?;
+        tracing::info!(socket = %path.display(), "web UI listening on unix socket");
+        axum::serve(listener, app).await?;
+        return Ok(());
+    }
+    let addr = SocketAddr::from(([0, 0, 0, 0], port));
+    let listener = bind_with_retry(addr, "web UI").await?;
+    tracing::info!(%port, "web UI listening on tcp");
+    axum::serve(listener, app).await?;
+    Ok(())
+}
+
+/// Bind a `UnixListener` at `path` and drop a `.bound` marker next
+/// to it so c0re's gateway-map writer knows the socket is live.
+/// Best-effort unlinks any stale socket left from a crashed previous
+/// harness (clean exit removes it, but `bind(2)` refuses to overwrite
+/// an existing file) and `mkdir -p`s the parent for first-boot. Mode
+/// `0o666` — world-accessible so the gateway container's nginx process
+/// can `connect(2)` without sharing a group with the agent user.
+/// The per-agent subdir (`/run/hive-agent/<name>/`) is only accessible
+/// to containers that have it bind-mounted, so world-accessible sockets
+/// are not a material risk.
+///
+/// Marker-gating + the gateway-side consumer: see
+/// [`docs/gateway.md::Per-agent unix-socket upstream`](../../../docs/gateway.md).
+fn bind_unix(path: &Path) -> Result<tokio::net::UnixListener> {
+    use std::os::unix::fs::PermissionsExt;
+    if let Some(parent) = path.parent() {
+        std::fs::create_dir_all(parent)
+            .with_context(|| format!("create socket parent dir {}", parent.display()))?;
+    }
+    // Best-effort: ENOENT is fine (no stale file); any other error
+    // surfaces via the bind below with a clearer "AddrInUse" / perms
+    // message than a partial cleanup would.
+    let _ = std::fs::remove_file(path);
+    let listener = tokio::net::UnixListener::bind(path)
+        .with_context(|| format!("bind unix socket at {}", path.display()))?;
+    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666))
+        .with_context(|| format!("set perms on {}", path.display()))?;
+    // Best-effort ready marker: failed write isn't fatal (the harness
+    // still binds + serves), it just means the gateway side keeps the
+    // TCP upstream for one more sync tick.
+    if let Some(parent) = path.parent() {
+        let marker = parent.join("hyperhive-socket-bound");
+        if let Err(e) = std::fs::write(&marker, b"") {
+            tracing::warn!(
+                marker = %marker.display(), error = %e,
+                "failed to write hyperhive-socket-bound marker — gateway may keep TCP upstream"
+            );
+        }
+    }
+    Ok(listener)
+}
+
+/// Maximum bind attempts before `bind_with_retry` gives up on `AddrInUse`.
+const MAX_BIND_ATTEMPTS: u32 = 12;
+
+/// Bind a TCP listener with `SO_REUSEADDR` set, retrying on `AddrInUse` with
+/// exponential backoff capped at 2s, up to [`MAX_BIND_ATTEMPTS`] attempts. If
+/// the port is still held after the final attempt, returns the `AddrInUse`
+/// error rather than looping forever (a genuine collision needs the operator,
+/// not an unbounded wait).
+///
+/// Retry rationale + dashboard-banner-on-real-collision:
+/// see [`docs/web-ui/shape.md::Listener bind`](../../../docs/web-ui/shape.md).
+async fn bind_with_retry(addr: SocketAddr, label: &str) -> Result<tokio::net::TcpListener> {
+    let mut delay_ms = 250u64;
+    let mut attempts = 0u32;
+    loop {
+        match try_bind(addr) {
+            Ok(l) => {
+                if attempts > 0 {
+                    tracing::info!(
+                        %addr, attempts,
+                        "{label}: bind succeeded after retry"
+                    );
+                }
+                return Ok(l);
+            }
+            Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => {
+                let attempt = attempts + 1;
+                if attempt >= MAX_BIND_ATTEMPTS {
+                    return Err(e).with_context(|| {
+                        format!("bind {label} on {addr}: still AddrInUse after {attempt} attempts")
+                    });
+                }
+                tracing::warn!(
+                    %addr, attempt,
+                    "{label}: AddrInUse, retrying in {delay_ms}ms"
+                );
+                tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
+                attempts += 1;
+                delay_ms = (delay_ms * 2).min(2000);
+            }
+            Err(e) => {
+                return Err(e).with_context(|| format!("bind {label} on {addr}"));
+            }
+        }
+    }
+}
+
+fn try_bind(addr: SocketAddr) -> std::io::Result<tokio::net::TcpListener> {
+    let sock = match addr {
+        SocketAddr::V4(_) => tokio::net::TcpSocket::new_v4()?,
+        SocketAddr::V6(_) => tokio::net::TcpSocket::new_v6()?,
+    };
+    sock.set_reuseaddr(true)?;
+    sock.bind(addr)?;
+    sock.listen(1024)
+}
+
+/// The fixed VNC port weston bound, from the `HIVE_GUI_VNC_PORT` env var
+/// the harness service sets when gui is enabled (see weston-vnc.nix).
+/// `None` when unset (gui not enabled for this agent) or unparseable.
+/// The port is a fixed, container-local value — no per-agent hashing, no
+/// marker file — because network isolation is unconditional (each agent
+/// has its own netns, so the port can't collide across containers).
+fn read_gui_vnc_port() -> Option<u16> {
+    std::env::var("HIVE_GUI_VNC_PORT").ok()?.parse().ok()
+}
+
+fn error_response(status: StatusCode, message: &str) -> Response {
+    // Plain text — JS app surfaces in `alert()`, HTML wrapping would just
+    // be noise. Status is per-caller: 400 for bad input, 409 for a
+    // retryable state conflict (turn in flight / hive-c0re busy), 500 only
+    // for a genuine server/transport failure — the frontend shows the code
+    // in its alert, so a benign "busy, retry" must not read as a 500.
+    (status, message.to_owned()).into_response()
+}

hive-ag3nt/src/web_ui/screen.rs

@@ -0,0 +1,96 @@
+//! VNC screen websocket relay + agent icon.
+
+use axum::{
+    extract::State,
+    http::StatusCode,
+    response::{IntoResponse, Response},
+};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+use super::AppState;
+
+/// This agent's icon. Serves the operator-configured SVG from
+/// `/etc/hyperhive/icon.svg` (set via the `hyperhive.icon` agent.nix
+/// option) when present, otherwise the bundled default hyperhive logo.
+/// Always returns an image, so consumers (dashboard, favicon) can hit
+/// `/icon` unconditionally without probing whether one is configured.
+pub(super) async fn serve_icon() -> impl IntoResponse {
+    // Per-agent icon overrides go through `/etc/hyperhive/icon.svg`
+    // (set via the `hyperhive.icon` agent.nix option); the bundled
+    // default is resolved at runtime from
+    // `$HIVE_ASSETS_DIR/branding/hyperhive.svg`. If neither file can
+    // be read we serve an empty body — keeps the response a valid SVG
+    // content-type without a panic on a misconfigured container.
+    let body = std::fs::read_to_string("/etc/hyperhive/icon.svg").unwrap_or_else(|_| {
+        std::fs::read_to_string(hive_sh4re::assets::branding_svg()).unwrap_or_default()
+    });
+    ([("content-type", "image/svg+xml")], body)
+}
+
+/// WebSocket handler: upgrade then pump bytes between the WS client and
+/// the VNC server on `127.0.0.1:<vnc_port>`. Returns 404 when gui is not
+/// enabled for this agent.
+pub(super) async fn screen_ws(
+    ws: axum::extract::ws::WebSocketUpgrade,
+    State(state): State<AppState>,
+) -> Response {
+    let Some(vnc_port) = state.gui_vnc_port else {
+        return (StatusCode::NOT_FOUND, "gui not enabled for this agent").into_response();
+    };
+    ws.on_upgrade(move |socket| relay_ws_vnc(socket, vnc_port))
+}
+
+/// Pure byte pump: forwards raw bytes between the WebSocket client and
+/// the VNC TCP stream. Transparent to any RFB variant (plain, `VeNCrypt`).
+async fn relay_ws_vnc(socket: axum::extract::ws::WebSocket, vnc_port: u16) {
+    // Import futures traits locally so they don't conflict with
+    // tokio_stream::StreamExt used at module scope.
+    use axum::extract::ws::Message;
+    use futures_util::{SinkExt, StreamExt as _};
+
+    let addr = format!("127.0.0.1:{vnc_port}");
+    let Ok(tcp) = tokio::net::TcpStream::connect(&addr).await else {
+        tracing::warn!(%addr, "screen/ws: could not connect to VNC server");
+        return;
+    };
+    let (mut tcp_rx, mut tcp_tx) = tcp.into_split();
+    let (mut ws_tx, mut ws_rx) = socket.split();
+
+    // WS → TCP
+    let ws_to_tcp = tokio::spawn(async move {
+        while let Some(Ok(msg)) = futures_util::StreamExt::next(&mut ws_rx).await {
+            match msg {
+                Message::Binary(data) if tcp_tx.write_all(&data).await.is_err() => {
+                    break;
+                }
+                Message::Close(_) => break,
+                _ => {} // ping/pong/text: ignore
+            }
+        }
+    });
+
+    // TCP → WS
+    let tcp_to_ws = tokio::spawn(async move {
+        let mut buf = vec![0u8; 8192];
+        loop {
+            match tcp_rx.read(&mut buf).await {
+                Ok(0) | Err(_) => break,
+                Ok(n) => {
+                    if ws_tx
+                        .send(Message::Binary(buf[..n].to_vec().into()))
+                        .await
+                        .is_err()
+                    {
+                        break;
+                    }
+                }
+            }
+        }
+    });
+
+    // Wait for either direction to close, then let both tasks drop.
+    tokio::select! {
+        _ = ws_to_tcp => {}
+        _ = tcp_to_ws => {}
+    }
+}

hive-ag3nt/src/web_ui/state.rs

@@ -0,0 +1,434 @@
+//! `/api/state` + `/api/dashboard-state` snapshot builders.
+
+use axum::extract::State;
+use serde::Serialize;
+
+use crate::client;
+use crate::login::LoginState;
+use crate::login_session::drop_if_finished;
+
+use super::{AppState, SOCKET_FETCH_TIMEOUT};
+
+pub(super) async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> {
+    // Capture seq *before* any reads so the dedupe contract is
+    // "events with seq > snapshot.seq are post-snapshot, never missed."
+    let seq = state.bus.current_seq();
+    drop_if_finished(&state.session);
+    let login = *state.login.lock().unwrap();
+    let session_snapshot = state.session.lock().unwrap().clone();
+    let (status, session_view) = match (login, session_snapshot) {
+        (LoginState::Online, _) if state.bus.is_rate_limited() => ("rate_limited", None),
+        (LoginState::Online, _) => ("online", None),
+        (LoginState::NeedsLogin, None) => ("needs_login_idle", None),
+        (LoginState::NeedsLogin, Some(s)) => (
+            "needs_login_in_progress",
+            Some(SessionView {
+                url: s.url(),
+                output: s.output(),
+                finished: s.finished(),
+                exit_note: s.exit_note(),
+            }),
+        ),
+    };
+    let dashboard_port = std::env::var("HIVE_DASHBOARD_PORT")
+        .ok()
+        .and_then(|s| s.parse::<u16>().ok())
+        .unwrap_or(7000);
+    let inbox = recent_inbox(&state.socket).await;
+    let (turn_state, turn_state_since) = state.bus.state_snapshot();
+    let model = state.bus.model();
+    let context_window_tokens = state
+        .bus
+        .api_context_window()
+        .unwrap_or_else(|| crate::events::context_window_tokens(&model));
+    let ctx_usage = state.bus.last_ctx_usage();
+    let cost_usage = state.bus.last_cost_usage();
+    let effort = state.bus.effort();
+    axum::Json(StateSnapshot {
+        seq,
+        label: state.label.clone(),
+        qualified_label: crate::identity::qualify(&state.label),
+        dashboard_port,
+        status,
+        session: session_view,
+        inbox,
+        turn_state,
+        turn_state_since,
+        model,
+        context_window_tokens,
+        ctx_usage,
+        cost_usage,
+        links: agent_links(&state.label, state.gui_vnc_port.is_some()),
+        forge_public_url: std::env::var("HIVE_FORGE_PUBLIC_URL")
+            .ok()
+            .filter(|s| !s.is_empty()),
+        hive_name: crate::identity::hive_name(),
+        swarm_name: crate::identity::swarm_name(),
+        available_models: available_models(),
+        effort,
+        available_efforts: crate::events::EFFORT_LEVELS
+            .iter()
+            .map(ToString::to_string)
+            .collect(),
+    })
+}
+
+/// Lean snapshot of the agent-owned fields that the dashboard card
+/// needs. Served at `GET /api/dashboard-state` (accessible through the
+/// gateway at `/agent/<name>/api/dashboard-state`). The dashboard
+/// fetches this once per running agent to get fresh, agent-authoritative
+/// values instead of relying on hive-c0re's periodic file-reads.
+///
+/// Structural fields (running, `needs_update`, `deployed_sha`, parent, …)
+/// continue to come from hive-c0re's `/api/state`; this endpoint covers
+/// only the fields the agent itself is the source of truth for.
+pub(super) async fn api_dashboard_state(
+    State(state): State<AppState>,
+) -> axum::Json<DashboardState> {
+    let (status_text, status_set_at) = read_own_status();
+    let rate_limited = state.bus.is_rate_limited();
+    let model = state.bus.model();
+    let context_window_tokens = state
+        .bus
+        .api_context_window()
+        .unwrap_or_else(|| crate::events::context_window_tokens(&model));
+    // Full context-window size = input + cache-read + cache-creation. Using
+    // raw `input_tokens` here reported only the *uncached* sliver, which is
+    // ~0 once prompt caching kicks in — so every card showed `ctx·0k`. Match
+    // the agent page (which ships the whole `ctx_usage` and sums it) and the
+    // cache-TTL logic in turn.rs, both of which use `context_tokens()`.
+    let ctx_tokens = state.bus.last_ctx_usage().map(|u| u.context_tokens());
+    axum::Json(DashboardState {
+        status_text,
+        status_set_at,
+        ctx_tokens,
+        context_window_tokens,
+        rate_limited,
+        links: agent_links(&state.label, state.gui_vnc_port.is_some()),
+    })
+}
+
+#[derive(Serialize)]
+pub(super) struct StateSnapshot {
+    /// Bus seq at the moment this snapshot was assembled. Clients dedupe
+    /// their buffered SSE traffic against this value: events with
+    /// `seq <= snapshot.seq` are already reflected (or pre-date the
+    /// snapshot); `seq > snapshot.seq` is post-snapshot. Reset to 0 on
+    /// harness restart — clients treat reconnect as a fresh world.
+    seq: u64,
+    label: String,
+    /// Hive-qualified long name (`${label}@${hyperhive.domain}`) when
+    /// the host has been configured for a multi-hive swarm; falls back
+    /// to the short label when the hive domain env var is unset.
+    /// The frontend uses this for the page title / agent self-introduction;
+    /// when it equals `label`, the page renders the short form unchanged.
+    qualified_label: String,
+    dashboard_port: u16,
+    /// `"online"` | `"rate_limited"` | `"needs_login_idle"` | `"needs_login_in_progress"`.
+    status: &'static str,
+    /// Present when `status == "needs_login_in_progress"`.
+    session: Option<SessionView>,
+    /// Last N messages addressed to this agent, newest-first. Pulled
+    /// from the broker via the per-agent socket on each render.
+    /// Empty on transport failure.
+    inbox: Vec<hive_sh4re::InboxRow>,
+    /// Authoritative turn-loop state from the harness and the unix
+    /// timestamp the state was entered. The JS computes the age
+    /// client-side off this rather than tracking it from SSE events.
+    turn_state: crate::events::TurnState,
+    turn_state_since: i64,
+    /// Currently-active claude model name. Reflected on the page so
+    /// the operator can see what they just switched to (and what's
+    /// in flight). Mutable at runtime via `POST /api/model`.
+    model: String,
+    /// Effective context-window token budget for the current model.
+    /// Primary source: API-reported `modelUsage.*.contextWindow` from
+    /// the last result event (authoritative per-inference active window).
+    /// Falls back to `HIVE_CONTEXT_WINDOW_TOKENS_*` env vars, then 200 000.
+    /// Consumers (e.g. dashboard badge) use this to render ctx-usage %.
+    context_window_tokens: u64,
+    /// Last-inference token usage from the most recent completed
+    /// turn — represents the current context-window size at turn-end.
+    /// `null` until the first turn finishes.
+    ctx_usage: Option<hive_claude::TokenUsage>,
+    /// Cumulative token usage across the most recent turn's inferences
+    /// (cost signal). `null` until the first turn finishes.
+    cost_usage: Option<hive_claude::TokenUsage>,
+    /// Navigation links for this agent page. Also served via
+    /// `DashboardState.links` (`GET /api/dashboard-state`) for the
+    /// dashboard card's icon strip. Both are produced by `agent_links()`
+    /// — single source of truth. See [`docs/web-ui/dashboard.md::Container row`]
+    /// for the frontend resolver + which links appear in which conditions.
+    links: Vec<AgentLink>,
+    /// Public URL of the forge served by hive-gateway (e.g.
+    /// `"https://forge.pr1ma.darkest.space"`). Sourced from
+    /// `HIVE_FORGE_PUBLIC_URL`; `None` when `forge.behindGateway=false`
+    /// or the env var is absent. The frontend uses this to build forge
+    /// nav-strip links instead of hardcoding `<hostname>:3000`.
+    forge_public_url: Option<String>,
+    /// Human name of this hive instance (e.g. `"pr1ma"`). Sourced
+    /// from `HYPERHIVE_HIVE_NAME`; `None` when unset. The frontend
+    /// uses this for the page `<title>` and header label so browser
+    /// tabs disambiguate when multiple hives are open in parallel.
+    hive_name: Option<String>,
+    /// Human name of the swarm (e.g. `"constellat1on"`). Sourced from
+    /// `HYPERHIVE_SWARM_NAME`; `None` when unset.
+    swarm_name: Option<String>,
+    /// Ordered list of model short-names the operator has declared as
+    /// available on this hive. Sourced from `HIVE_AVAILABLE_MODELS`
+    /// (comma-separated, set by `services.hyperhive.availableModels`).
+    /// Falls back to `["haiku", "sonnet", "opus"]` when the env var is
+    /// absent or empty. The frontend model quick-picker renders one button
+    /// per entry in this list, so operators can add new models or drop
+    /// ones they don't want without touching the frontend code.
+    available_models: Vec<String>,
+    /// Currently-active claude effort level. Reflected on the page so the
+    /// operator's effort picker shows the live selection. Mutable at
+    /// runtime via `POST /api/effort`; applies on the next session.
+    effort: String,
+    /// Selectable effort levels for the picker, ascending. Fixed set
+    /// (`low`, `medium`, `high`, `xhigh`, `max`) — sourced from
+    /// [`crate::events::EFFORT_LEVELS`], not operator-configurable like
+    /// `available_models`. The frontend renders one button per entry.
+    available_efforts: Vec<String>,
+}
+
+#[derive(Serialize)]
+struct SessionView {
+    /// First `https://…` claude emitted on stdout, if any.
+    url: Option<String>,
+    /// Accumulated stdout + stderr.
+    output: String,
+    finished: bool,
+    exit_note: Option<String>,
+}
+
+/// One navigation link in the agent page header row. The same JSON
+/// shape appears in both `StateSnapshot.links` (`GET /api/state`,
+/// per-agent page) and `DashboardState.links` (`GET /api/dashboard-state`,
+/// dashboard card icon strip). `agent_links()` is the single source
+/// of truth for what links an agent exposes.
+#[derive(Serialize)]
+struct AgentLink {
+    /// `kind = Container | Forge` → path; `kind = External` → full URL.
+    /// The frontend prepends the right base before rendering.
+    url: String,
+    icon: String,
+    label: String,
+    kind: AgentLinkKind,
+}
+
+/// Resolution hint for `AgentLink.url`. The agent backend can't know
+/// which hostname the browser sees (especially when the dashboard
+/// proxies the call from a different origin), so it labels each link
+/// and lets the frontend prepend the right base.
+#[derive(Serialize, Clone, Copy)]
+#[serde(rename_all = "snake_case")]
+enum AgentLinkKind {
+    /// `url` is a path on the agent's container web UI (`/stats`,
+    /// `/screen`). Agent page: same-origin path. Dashboard:
+    /// `http://<host>:<container.port><url>`.
+    Container,
+    /// `url` is a path on the local Forgejo (`/<label>`,
+    /// `/agent-configs/<label>`). Both surfaces:
+    /// `http://<host>:3000<url>`.
+    Forge,
+    /// `url` is already a fully-qualified absolute URL — use as-is.
+    /// Agent-declared `hyperhive.dashboardLinks` extras arrive here.
+    External,
+}
+
+#[derive(serde::Serialize)]
+pub(super) struct DashboardState {
+    /// Free-text status set by `set_status`, read directly from the
+    /// `hyperhive-status` file the harness writes. `None` when unset.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    status_text: Option<String>,
+    /// Unix timestamp (seconds) when the status file was last written.
+    /// `None` when no status is set.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    status_set_at: Option<i64>,
+    /// Full context-window size from the most recent completed turn
+    /// (`ctx_usage.context_tokens()` = input + cache-read + cache-creation).
+    /// `None` until the first turn finishes. Drives the `ctx·Nk` card badge.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    ctx_tokens: Option<u64>,
+    /// Effective context-window budget for the current model. Same
+    /// derivation as `StateSnapshot::context_window_tokens`.
+    context_window_tokens: u64,
+    /// True while the harness is parked after a rate-limit response.
+    rate_limited: bool,
+    /// Navigation links for the dashboard card's icon strip. This is
+    /// the authoritative source — includes the screen link (GUI agents)
+    /// which hive-c0re's disk-based fallback cannot determine.
+    links: Vec<AgentLink>,
+}
+
+/// Read the agent's own free-text status and the timestamp when it was
+/// set, directly from the `hyperhive-status` file in the state dir.
+/// Mirrors `hive_c0re::container_view::read_agent_status` but runs
+/// inside the agent container using its own state dir.
+fn read_own_status() -> (Option<String>, Option<i64>) {
+    let path = crate::paths::state_dir().join("hyperhive-status");
+    let meta = std::fs::metadata(&path).ok();
+    let text = std::fs::read_to_string(&path)
+        .ok()
+        .as_deref()
+        .map(str::trim)
+        .filter(|t| !t.is_empty())
+        .map(str::to_owned);
+    let mtime = meta.and_then(|m| {
+        m.modified().ok().and_then(|t| {
+            t.duration_since(std::time::UNIX_EPOCH)
+                .ok()
+                .and_then(|d| i64::try_from(d.as_secs()).ok())
+        })
+    });
+    if text.is_none() {
+        (None, None)
+    } else {
+        (text, mtime)
+    }
+}
+
+/// Build the navigation link list for the agent page header. URLs
+/// are paths (relative) for `Container`/`Forge` targets and absolute
+/// for `External`; the frontend resolves each against its `kind`
+/// against the right base so the backend never has to guess the
+/// operator's browser host. See
+/// [`docs/web-ui/dashboard.md::Container row`](../../../docs/web-ui/dashboard.md) for
+/// the resolver + how `deployed:<sha>` ships alongside.
+fn agent_links(label: &str, gui_enabled: bool) -> Vec<AgentLink> {
+    let mut links = Vec::new();
+
+    links.push(AgentLink {
+        url: "stats.html".to_owned(),
+        icon: "📊".to_owned(),
+        label: "stats".to_owned(),
+        kind: AgentLinkKind::Container,
+    });
+
+    if gui_enabled {
+        links.push(AgentLink {
+            url: "screen.html".to_owned(),
+            icon: "🖥".to_owned(),
+            label: "screen".to_owned(),
+            kind: AgentLinkKind::Container,
+        });
+    }
+
+    if crate::paths::state_dir().join("forge-token").is_file() {
+        links.push(AgentLink {
+            url: format!("/{label}"),
+            icon: "⬡".to_owned(),
+            label: "forge".to_owned(),
+            kind: AgentLinkKind::Forge,
+        });
+        links.push(AgentLink {
+            url: format!("/agent-configs/{label}"),
+            icon: "↳".to_owned(),
+            label: "config".to_owned(),
+            kind: AgentLinkKind::Forge,
+        });
+    }
+
+    // Agent-declared extras (`hyperhive.dashboardLinks` → the
+    // `hive-dashboard-links` NixOS oneshot writes them to
+    // `{state_dir}/hyperhive-dashboard-links.json`). Shape on disk
+    // is `{label, icon, url}` with absolute URLs — those become
+    // `kind = External` links, passed through verbatim.
+    let extras_path = crate::paths::state_dir().join("hyperhive-dashboard-links.json");
+    if let Ok(text) = std::fs::read_to_string(&extras_path)
+        && !text.trim().is_empty()
+        && let Ok(extras) = serde_json::from_str::<Vec<ExtraLink>>(&text)
+    {
+        for e in extras {
+            links.push(AgentLink {
+                url: e.url,
+                icon: e.icon,
+                label: e.label,
+                kind: AgentLinkKind::External,
+            });
+        }
+    }
+
+    links
+}
+
+/// On-disk shape of `hyperhive-dashboard-links.json` (the
+/// `hive-dashboard-links` NixOS oneshot's output). Mapped to
+/// `AgentLink { kind: External }` inside `agent_links`.
+#[derive(serde::Deserialize)]
+struct ExtraLink {
+    label: String,
+    #[serde(default)]
+    icon: String,
+    url: String,
+}
+
+/// Best-effort: pull the last 30 messages addressed to us via the
+/// per-agent / manager socket. Empty list on any transport / decode
+/// failure — the inbox section is decorative, not authoritative.
+async fn recent_inbox(socket: &std::path::Path) -> Vec<hive_sh4re::InboxRow> {
+    const LIMIT: u64 = 30;
+    // Deadline-bounded: `/api/state` must render even when hive-c0re is
+    // busy — an empty inbox section beats a hung snapshot.
+    match tokio::time::timeout(
+        SOCKET_FETCH_TIMEOUT,
+        client::request::<_, hive_sh4re::Response>(
+            socket,
+            &hive_sh4re::Request::Recent { limit: LIMIT },
+        ),
+    )
+    .await
+    {
+        Ok(Ok(hive_sh4re::Response::Recent { rows })) => rows,
+        _ => Vec::new(),
+    }
+}
+
+/// Fetch reminder activity stats from the broker via the per-agent /
+/// manager socket. Returns None on any transport / decode failure — the
+/// stats are decorative, not authoritative.
+pub(super) async fn fetch_reminder_stats(
+    socket: &std::path::Path,
+    window_secs: u64,
+) -> Option<hive_sh4re::ReminderStats> {
+    match tokio::time::timeout(
+        SOCKET_FETCH_TIMEOUT,
+        client::request::<_, hive_sh4re::Response>(
+            socket,
+            &hive_sh4re::Request::ReminderRollup {
+                since_secs: window_secs,
+                agent: None,
+            },
+        ),
+    )
+    .await
+    {
+        Ok(Ok(hive_sh4re::Response::ReminderRollup(stats))) => Some(stats),
+        _ => None,
+    }
+}
+
+/// Read `HIVE_AVAILABLE_MODELS` (comma-separated short names injected by
+/// `services.hyperhive.availableModels`) and return the parsed list.
+/// Falls back to `["haiku", "sonnet", "opus"]` when the env var is absent
+/// or resolves to an empty list after trimming.
+fn available_models() -> Vec<String> {
+    const DEFAULT: &[&str] = &["haiku", "sonnet", "opus"];
+    let raw = match std::env::var("HIVE_AVAILABLE_MODELS") {
+        Ok(v) if !v.trim().is_empty() => v,
+        _ => return DEFAULT.iter().map(ToString::to_string).collect(),
+    };
+    let models: Vec<String> = raw
+        .split(',')
+        .map(|s| s.trim().to_string())
+        .filter(|s| !s.is_empty())
+        .collect();
+    if models.is_empty() {
+        DEFAULT.iter().map(ToString::to_string).collect()
+    } else {
+        models
+    }
+}

hive-ag3nt/src/web_ui/stats.rs

@@ -0,0 +1,131 @@
+//! Stats + loose-ends + bash-tasks read endpoints.
+
+use axum::extract::State;
+use axum::http::StatusCode;
+use axum::response::{IntoResponse, Response};
+use serde::Deserialize;
+
+use crate::client;
+
+use super::state::fetch_reminder_stats;
+use super::{AppState, SOCKET_FETCH_TIMEOUT, error_response};
+
+#[derive(Deserialize)]
+pub(super) struct StatsQuery {
+    window: Option<String>,
+}
+
+pub(super) async fn api_stats(
+    State(state): State<AppState>,
+    axum::extract::Query(q): axum::extract::Query<StatsQuery>,
+) -> axum::Json<crate::stats::Snapshot> {
+    let window = crate::stats::Window::parse(q.window.as_deref().unwrap_or("24h"));
+    let mut snapshot = crate::stats::snapshot_default(window);
+    // Pass the window span to the reminder-stats RPC so the broker
+    // filters its counts to the same time range as the chart data.
+    let window_secs = window.span_secs();
+    let window_secs_u = u64::try_from(window_secs).unwrap_or(0);
+    snapshot.reminder_stats = fetch_reminder_stats(&state.socket, window_secs_u).await;
+    axum::Json(snapshot)
+}
+
+/// Proxy this agent's loose-ends list via the per-agent socket. The
+/// web UI surfaces the result as a collapsible section in the page
+/// so the operator can see at a glance what's pending against the
+/// agent (questions asked by it, peer questions targeting it,
+/// reminders it scheduled, approvals for the manager). Same data
+/// the `mcp__hyperhive__get_loose_ends` tool sees from inside the
+/// container.
+pub(super) async fn api_loose_ends(State(state): State<AppState>) -> Response {
+    let loose_ends: Vec<hive_sh4re::LooseEnd> = match tokio::time::timeout(
+        SOCKET_FETCH_TIMEOUT,
+        client::request::<_, hive_sh4re::Response>(
+            &state.socket,
+            &hive_sh4re::Request::GetLooseEnds { agent: None },
+        ),
+    )
+    .await
+    {
+        Ok(Ok(hive_sh4re::Response::LooseEnds { loose_ends })) => loose_ends,
+        Ok(Ok(hive_sh4re::Response::Err { message })) => {
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                &format!("get_loose_ends: {message}"),
+            );
+        }
+        Ok(Ok(other)) => {
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                &format!("unexpected response: {other:?}"),
+            );
+        }
+        Ok(Err(e)) => {
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                &format!("transport: {e:#}"),
+            );
+        }
+        Err(_) => {
+            return error_response(
+                StatusCode::CONFLICT,
+                "get_loose_ends: timed out — hive-c0re busy, retry",
+            );
+        }
+    };
+    axum::Json(serde_json::json!({ "loose_ends": loose_ends })).into_response()
+}
+
+/// `GET /api/bash-tasks` — snapshot of this agent's in-flight bash tasks.
+///
+/// The `hive-bash-mcp` daemon runs in this same container and writes one
+/// `<id>.json` ([`hive_sh4re::TaskFile`]) per task under the harness
+/// `bash-tasks/` dir. This reads that dir and returns the tasks still
+/// `Pending` or `Running`, so the agent page can show what's running without
+/// going through the broker. Snapshot only — the page polls/refreshes it like
+/// `/api/loose-ends`; there's no live SSE push for task state yet. Unreadable
+/// or malformed files (incl. the daemon's `.json.tmp` scratch writes, which
+/// don't match the `.json` extension) are skipped so one stray file can't
+/// fail the whole list.
+pub(super) async fn api_bash_tasks() -> Response {
+    let dir = crate::paths::harness_dir().join("bash-tasks");
+    // The dir scan + per-file reads are blocking fs I/O; run them off the
+    // async executor so a slow or large tasks dir can't stall other requests.
+    let tasks = tokio::task::spawn_blocking(move || {
+        let mut tasks: Vec<hive_sh4re::TaskFile> = Vec::new();
+        let Ok(rd) = std::fs::read_dir(&dir) else {
+            return tasks;
+        };
+        for entry in rd.flatten() {
+            let path = entry.path();
+            if path.extension().and_then(|e| e.to_str()) != Some("json") {
+                continue;
+            }
+            let Ok(text) = std::fs::read_to_string(&path) else {
+                continue;
+            };
+            let Ok(task) = serde_json::from_str::<hive_sh4re::TaskFile>(&text) else {
+                continue;
+            };
+            if matches!(
+                task.status,
+                hive_sh4re::TaskStatus::Pending | hive_sh4re::TaskStatus::Running
+            ) {
+                tasks.push(task);
+            }
+        }
+        // Running before Pending, then oldest-first so a long-runner sits on top.
+        tasks.sort_by(|a, b| {
+            let rank = |s: &hive_sh4re::TaskStatus| match s {
+                hive_sh4re::TaskStatus::Running => 0,
+                _ => 1,
+            };
+            rank(&a.status)
+                .cmp(&rank(&b.status))
+                .then(a.created_at.cmp(&b.created_at))
+        });
+        tasks
+    })
+    .await
+    .unwrap_or_default();
+    axum::Json(serde_json::json!({ "tasks": tasks })).into_response()
+}

hive-ag3nt/src/web_ui/stream.rs

@@ -0,0 +1,70 @@
+//! Live SSE event stream + history endpoints.
+
+use std::convert::Infallible;
+
+use axum::Json;
+use axum::extract::{Query, State};
+use axum::response::sse::{Event, KeepAlive, Sse};
+use serde::Deserialize;
+use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream};
+
+use super::AppState;
+
+/// Query params for the paginated history endpoint.
+#[derive(Debug, Deserialize)]
+pub(super) struct HistoryParams {
+    /// Cursor: only return events with sqlite row id < `before`.
+    /// Omit for the initial (most-recent) page.
+    before: Option<i64>,
+    /// Page size (default 100, capped at `HISTORY_CAPACITY`).
+    limit: Option<usize>,
+}
+
+pub(super) async fn events_history(
+    State(state): State<AppState>,
+    Query(params): Query<HistoryParams>,
+) -> Json<serde_json::Value> {
+    use crate::events::HISTORY_CAPACITY;
+    let limit = params.limit.unwrap_or(100).min(HISTORY_CAPACITY);
+    let before = params.before;
+    let is_initial = before.is_none();
+
+    // Capture seq *before* the read on initial loads so the SSE dedupe
+    // window is "drop buffered events you've already seen in history",
+    // never "lose an event that fired between the read and the seq."
+    // On paginated loads (`before` is set) seq is not needed.
+    let seq = if is_initial {
+        Some(state.bus.current_seq())
+    } else {
+        None
+    };
+
+    let (events, min_id, has_more) = state.bus.history_page(before, limit);
+    let mut resp = serde_json::json!({
+        "events": events,
+        "min_id": min_id,
+        "has_more": has_more,
+    });
+    if let Some(s) = seq {
+        resp["seq"] = serde_json::json!(s);
+    }
+    Json(resp)
+}
+
+pub(super) async fn events_stream(
+    State(state): State<AppState>,
+) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
+    tracing::info!("sse: client subscribed");
+    let rx = state.bus.subscribe();
+    // Drop a "hello" note into the bus so every new subscriber sees at
+    // least one event immediately and can clear the connecting placeholder.
+    state.bus.emit(crate::events::LiveEvent::Note {
+        text: "live stream attached".into(),
+    });
+    let stream = BroadcastStream::new(rx).filter_map(|res| {
+        let ev = res.ok()?;
+        let json = serde_json::to_string(&ev).ok()?;
+        Some(Ok(Event::default().data(json)))
+    });
+    Sse::new(stream).keep_alive(KeepAlive::default())
+}

hive-claude/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "hive-claude"
+edition.workspace = true
+version.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+serde = { workspace = true }
+serde_json.workspace = true
+thiserror.workspace = true
+tokio.workspace = true

hive-claude/README.md

@@ -0,0 +1,75 @@
+# hive-claude
+
+A small, reusable async driver for headless `claude --print` (Claude Code CLI)
+sessions. It spawns the CLI, streams and classifies the `stream-json` output,
+and reports the result. It knows only about the Claude Code CLI — **no
+hyperhive types, policy, watermarks, or logging.** Compaction, retry, and reset
+policy belong to the caller.
+
+## When to use it
+
+Reach for this crate whenever you need to run the `claude` CLI from Rust and
+react to how a turn ended. It is the shared substrate under
+`hive-ag3nt`'s turn loop; new callers (tools, tests, other agents) should build
+on it rather than shelling out to `claude` by hand.
+
+## Shape
+
+Two layers — reach for the high-level one:
+
+- **`InfiniteSession { name, store, policy }`** — a durable session that keeps
+  itself alive across the context window. `.run(&config, prompt, &sink)` →
+  `Result<Progress, Error>` does resume-or-create, compacts **reactively** on
+  overflow (compact + retry once), and **proactively** after a clean turn when
+  the policy says so (optional checkpoint turn, then compact). `.compact(…)`
+  forces one. `Progress { created, compacted, telemetry }` reports what happened
+  — including the turn's parsed `Telemetry`, finalised at the `result` event.
+- **`Claude::run(&config, &attach, prompt, &sink)`** → `Result<(), Error>` —
+  the low-level driver: one turn, one `Attach` target (`Resume` / `Create` /
+  `Continue` / `OneOff`). A clean turn is `Ok(())`; every other state is an
+  `Error` variant.
+
+Supporting pieces:
+
+- **`CompactionPolicy`** — decides *when* to compact. **`PercentPolicy`**
+  (`percent`, `default_window`, `checkpoint_prompt`) compacts at a percent of
+  the model window; **`NeverCompact`** never does.
+- **`Config`** — the invocation (model, effort, cwd, prompt/MCP files, tools,
+  extra args).
+- **`Sink`** — a trait with no-op defaults; implement what you care about to
+  observe stream events, non-JSON stdout, and stderr. `NoopSink` ignores all.
+- **`SessionStore`** — locate and archive on-disk sessions by title.
+- **`Telemetry`** (`context`, `cost`, `context_window`, `model`) — everything
+  the driver parses from a turn's stream, returned in `Progress`. **`Usage`** is
+  the minimal slice (`context_tokens`, `context_window`) the policy sees.
+
+`Error` unifies the two things that can stop a turn: recognized **sentinels**
+(`PromptTooLong`, `RateLimited`, `AuthFailed`, `SessionNotFound`) and **hard
+failures** (`Spawn`, `Stdin`, `Wait`, `Exit`, `Io`). Sentinels are expected
+control-flow, not crashes — the caller compacts, parks, re-auths, or creates a
+session in response.
+
+```rust
+use hive_claude::{Claude, Config, Error, NoopSink, Session};
+
+let config = Config { model: "haiku".into(), ..Default::default() };
+match Claude::run(&config, &Session::Resume("my-session".into()), "hello", &NoopSink).await {
+    Ok(()) => {}
+    Err(Error::PromptTooLong) => { /* compact + retry */ }
+    Err(Error::RateLimited)   => { /* park + retry */ }
+    Err(other)                => eprintln!("claude: {other}"),
+}
+```
+
+## `thiserror` here, `anyhow` in the apps
+
+This is a **library**, so it returns a concrete, matchable `Error` enum built
+with `thiserror`: callers can tell a rate-limit from a spawn failure and act
+accordingly. Libraries should never force their callers into `anyhow`'s
+type-erased error.
+
+The **applications** (the `hive-*` binaries) use `anyhow` instead — at the top
+level you usually only want to add context and bubble a failure up, not match
+on it. A `hive_claude::Error` converts into an `anyhow::Error` for free at the
+`?` boundary. Rule of thumb: **libraries return `thiserror` enums, binaries
+consume them with `anyhow`.**

hive-claude/src/classify.rs

@@ -0,0 +1,107 @@
+//! Sentinel detection: mapping claude-code CLI output onto [`crate::Error`]
+//! variants.
+//!
+//! These marker strings are claude-code CLI knowledge, not app knowledge. They
+//! are empirically stable across CLI versions; if one drifts the run degrades
+//! gracefully (a clean turn, or a hard [`crate::Error::Exit`] on a non-zero
+//! exit) rather than misbehaving.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+
+/// Emitted when the prompt/context exceeds the model's window.
+const PROMPT_TOO_LONG_MARKER: &str = "Prompt is too long";
+
+/// Substrings indicating the API refused for rate-limit / usage-cap / credit
+/// reasons. On stdout these are only trusted inside a JSON `error` event (see
+/// [`Sentinels::scan_stdout_json`] / [`Sentinels::scan_rate_limit_text`]) so a
+/// model *discussing* a rate limit in prose can't trigger a false positive.
+const RATE_LIMIT_MARKERS: [&str; 5] = [
+    "rate_limit_error",
+    "overloaded_error",
+    "Credit balance is too low",
+    "Usage limit reached",
+    "Request rate limit exceeded",
+];
+
+/// Substrings indicating the API rejected the request as unauthenticated (401)
+/// — an expired/revoked OAuth session. Sourced from claude-code's `api_retry`
+/// JSON events and its human-readable give-up line.
+const AUTH_FAIL_MARKERS: [&str; 3] = [
+    "\"error\":\"authentication_failed\"",
+    "\"error_status\":401",
+    "Failed to authenticate. API Error: 401",
+];
+
+/// Substrings indicating `--resume` could not resolve its target: no session
+/// with the given title, or no conversation with the given id.
+const SESSION_NOT_FOUND_MARKERS: [&str; 2] = [
+    "does not match any session title",
+    "No conversation found with session ID",
+];
+
+/// Shared, lock-free sentinel flags accumulated while both output streams are
+/// pumped concurrently. Read once after the child exits.
+#[derive(Default)]
+pub(crate) struct Sentinels {
+    prompt_too_long: AtomicBool,
+    rate_limited: AtomicBool,
+    auth_failed: AtomicBool,
+    session_not_found: AtomicBool,
+}
+
+impl Sentinels {
+    /// Scan a raw line (stdout or stderr) for the always-on markers:
+    /// prompt-too-long, auth-failed, session-not-found. Rate-limit is handled
+    /// separately because on stdout it must only fire on JSON `error` events.
+    pub(crate) fn scan_line(&self, line: &str) {
+        if line.contains(PROMPT_TOO_LONG_MARKER) {
+            self.prompt_too_long.store(true, Ordering::Relaxed);
+        }
+        if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) {
+            self.auth_failed.store(true, Ordering::Relaxed);
+        }
+        if SESSION_NOT_FOUND_MARKERS.iter().any(|m| line.contains(m)) {
+            self.session_not_found.store(true, Ordering::Relaxed);
+        }
+    }
+
+    /// Trust a rate-limit hit only on a JSON `error` event (so a model
+    /// *discussing* a rate limit in prose can't trigger it). The `type` gate
+    /// needs the parsed `event`; the marker match runs on `raw`, the original
+    /// line — the same bytes, so we don't re-serialize the value.
+    pub(crate) fn scan_stdout_json(&self, event: &serde_json::Value, raw: &str) {
+        if event.get("type").and_then(|t| t.as_str()) == Some("error")
+            && RATE_LIMIT_MARKERS.iter().any(|m| raw.contains(m))
+        {
+            self.rate_limited.store(true, Ordering::Relaxed);
+        }
+    }
+
+    /// Trust a rate-limit hit on raw text (non-JSON stdout, or any stderr) —
+    /// these are CLI messages, not conversation content.
+    pub(crate) fn scan_rate_limit_text(&self, line: &str) {
+        if RATE_LIMIT_MARKERS.iter().any(|m| line.contains(m)) {
+            self.rate_limited.store(true, Ordering::Relaxed);
+        }
+    }
+
+    /// The recognized-sentinel error, if any fired — `None` means no sentinel
+    /// (so the run either completed or failed hard on its exit code). The
+    /// sentinels keep a fixed priority (too-long > rate > auth); a
+    /// session-not-found can only arise on a resume that made no model call,
+    /// so it never coincides with the others.
+    pub(crate) fn soft_error(&self) -> Option<crate::Error> {
+        use crate::Error;
+        if self.prompt_too_long.load(Ordering::Relaxed) {
+            Some(Error::PromptTooLong)
+        } else if self.rate_limited.load(Ordering::Relaxed) {
+            Some(Error::RateLimited)
+        } else if self.auth_failed.load(Ordering::Relaxed) {
+            Some(Error::AuthFailed)
+        } else if self.session_not_found.load(Ordering::Relaxed) {
+            Some(Error::SessionNotFound)
+        } else {
+            None
+        }
+    }
+}

hive-claude/src/config.rs

@@ -0,0 +1,59 @@
+//! Invocation config: how to build one `claude --print` command line.
+
+use std::path::PathBuf;
+
+/// How a run attaches to a claude session — the low-level session flag. Kept
+/// separate from [`Config`] so one config can drive resume + create +
+/// `/compact` of the same logical session. For a self-managing durable
+/// session, prefer [`crate::InfiniteSession`] over hand-picking an `Attach`.
+#[derive(Debug, Clone)]
+pub enum Attach {
+    /// `--resume <id-or-title>` — resume an existing session by UUID or by the
+    /// display title set via [`Attach::Create`]. Yields
+    /// [`crate::Error::SessionNotFound`] if nothing matches.
+    Resume(String),
+    /// `--name <title>` — start a new session carrying the given display title
+    /// (persisted as a `custom-title` event, which `--resume <title>` later
+    /// resolves against).
+    Create(String),
+    /// `--continue` — resume the most recent session in the cwd. Ambiguous
+    /// when other claude processes share the cwd; prefer titled sessions.
+    Continue,
+    /// No session flag — a one-off, unnamed session.
+    OneOff,
+}
+
+/// Everything needed to build one headless `claude --print` invocation, minus
+/// the [`Attach`] target (passed separately to [`crate::Claude::run`]).
+///
+/// Fields map one-to-one to CLI flags; `None`/empty means "don't pass the
+/// flag". `--print --verbose --output-format stream-json` are always set by
+/// the driver and are not configurable here (the driver depends on the
+/// stream-json shape).
+#[derive(Debug, Clone, Default)]
+pub struct Config {
+    /// `--model`. Empty omits the flag (claude falls back to its own default).
+    pub model: String,
+    /// `--effort <level>`. `None` omits the flag.
+    pub effort: Option<String>,
+    /// Working directory for the child. Claude derives its per-project session
+    /// dir from this path. `None` inherits the parent process cwd.
+    pub cwd: Option<PathBuf>,
+    /// `--system-prompt-file <path>`.
+    pub system_prompt_file: Option<PathBuf>,
+    /// `--mcp-config <path>`.
+    pub mcp_config: Option<PathBuf>,
+    /// Pass `--strict-mcp-config` (only the configured MCP servers, no
+    /// discovery).
+    pub strict_mcp_config: bool,
+    /// `--tools <expr>` — the built-in tool allow-list expression.
+    pub tools: Option<String>,
+    /// `--allowedTools <expr>`.
+    pub allowed_tools: Option<String>,
+    /// `--add-dir <path>` (repeatable) — extra readable directories.
+    pub add_dirs: Vec<PathBuf>,
+    /// Any additional raw args appended verbatim after the ones above.
+    pub extra_args: Vec<String>,
+    /// Program to spawn. `None` defaults to `claude` (resolved on `PATH`).
+    pub program: Option<String>,
+}

hive-claude/src/driver.rs

@@ -0,0 +1,179 @@
+//! The subprocess driver: spawn claude, pump + classify its streams, and
+//! assemble the result.
+
+use std::collections::VecDeque;
+use std::process::Stdio;
+
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::process::{ChildStderr, ChildStdout, Command};
+
+use crate::classify::Sentinels;
+use crate::{Attach, Config, Error, Result, Sink};
+
+/// Default program name spawned when [`Config::program`] is unset.
+const DEFAULT_PROGRAM: &str = "claude";
+
+/// How many trailing stderr lines to keep for [`Error::Exit`].
+const STDERR_TAIL_LINES: usize = 20;
+
+/// The low-level driver entry point. A namespace for the run function — there
+/// is nothing to construct; call `Claude::run(…)` directly. For a durable,
+/// self-compacting session, use [`crate::InfiniteSession`] instead.
+pub struct Claude;
+
+impl Claude {
+    /// Spawn one headless `claude --print` turn, stream its output through
+    /// `sink`, and report the result.
+    ///
+    /// The wake prompt is written to claude's stdin. stdout (`stream-json`)
+    /// and stderr are pumped concurrently while the child runs. A clean turn
+    /// returns `Ok(())`; every non-completion state — recognized sentinel or
+    /// hard failure — is an [`Error`] variant, so callers branch with a single
+    /// `match`.
+    ///
+    /// # Errors
+    ///
+    /// - A recognized sentinel: [`Error::PromptTooLong`], [`Error::RateLimited`],
+    ///   [`Error::AuthFailed`], [`Error::SessionNotFound`].
+    /// - [`Error::Spawn`] if the binary can't be launched.
+    /// - [`Error::Stdin`] / [`Error::Wait`] on stdin-write / child-wait failure.
+    /// - [`Error::Exit`] on a non-zero exit that raised no sentinel.
+    pub async fn run(
+        config: &Config,
+        attach: &Attach,
+        prompt: &str,
+        sink: &impl Sink,
+    ) -> Result<()> {
+        let program = config.program.as_deref().unwrap_or(DEFAULT_PROGRAM);
+        let mut cmd = build_command(program, config, attach);
+
+        let mut child = cmd
+            .stdin(Stdio::piped())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .spawn()
+            .map_err(|source| Error::Spawn {
+                program: program.to_string(),
+                source,
+            })?;
+
+        if let Some(mut stdin) = child.stdin.take() {
+            stdin.write_all(prompt.as_bytes()).await.map_err(Error::Stdin)?;
+            // Best-effort flush/close; claude sees EOF and starts the turn.
+            stdin.shutdown().await.ok();
+        }
+        let stdout = child.stdout.take().expect("stdout piped");
+        let stderr = child.stderr.take().expect("stderr piped");
+
+        let sentinels = Sentinels::default();
+        // Pump both streams and wait for exit concurrently on this task — no
+        // `spawn`, so the sink needn't be `'static` and borrows stay simple.
+        let ((), stderr_tail, status) = tokio::join!(
+            pump_stdout(stdout, sink, &sentinels),
+            pump_stderr(stderr, sink, &sentinels),
+            child.wait(),
+        );
+        let status = status.map_err(Error::Wait)?;
+
+        // A recognized sentinel takes precedence over the exit code; otherwise
+        // a non-zero exit with no sentinel is a hard failure.
+        if let Some(sentinel) = sentinels.soft_error() {
+            return Err(sentinel);
+        }
+        if !status.success() {
+            return Err(Error::Exit {
+                status,
+                stderr_tail,
+            });
+        }
+        Ok(())
+    }
+}
+
+/// Assemble the argv. `--print --verbose --output-format stream-json` are
+/// mandatory (the driver parses that shape); everything else is gated on the
+/// [`Config`] / [`Attach`].
+fn build_command(program: &str, config: &Config, attach: &Attach) -> Command {
+    let mut cmd = Command::new(program);
+    if let Some(cwd) = &config.cwd {
+        cmd.current_dir(cwd);
+    }
+    cmd.arg("--print")
+        .arg("--verbose")
+        .arg("--output-format")
+        .arg("stream-json");
+    if !config.model.is_empty() {
+        cmd.arg("--model").arg(&config.model);
+    }
+    if let Some(effort) = &config.effort {
+        cmd.arg("--effort").arg(effort);
+    }
+    match attach {
+        Attach::Resume(id) => {
+            cmd.arg("--resume").arg(id);
+        }
+        Attach::Create(title) => {
+            cmd.arg("--name").arg(title);
+        }
+        Attach::Continue => {
+            cmd.arg("--continue");
+        }
+        Attach::OneOff => {}
+    }
+    if let Some(path) = &config.system_prompt_file {
+        cmd.arg("--system-prompt-file").arg(path);
+    }
+    if let Some(path) = &config.mcp_config {
+        cmd.arg("--mcp-config").arg(path);
+    }
+    if config.strict_mcp_config {
+        cmd.arg("--strict-mcp-config");
+    }
+    if let Some(tools) = &config.tools {
+        cmd.arg("--tools").arg(tools);
+    }
+    if let Some(allowed) = &config.allowed_tools {
+        cmd.arg("--allowedTools").arg(allowed);
+    }
+    for dir in &config.add_dirs {
+        cmd.arg("--add-dir").arg(dir);
+    }
+    for extra in &config.extra_args {
+        cmd.arg(extra);
+    }
+    cmd
+}
+
+/// Read stdout line by line: classify each line, parse JSON, hand events (or
+/// raw non-JSON lines) to the sink.
+async fn pump_stdout(stdout: ChildStdout, sink: &impl Sink, sentinels: &Sentinels) {
+    let mut lines = BufReader::new(stdout).lines();
+    while let Ok(Some(line)) = lines.next_line().await {
+        sentinels.scan_line(&line);
+        if let Ok(event) = serde_json::from_str::<serde_json::Value>(&line) {
+            sentinels.scan_stdout_json(&event, &line);
+            sink.on_event(&event);
+        } else {
+            sentinels.scan_rate_limit_text(&line);
+            sink.on_stdout_line(&line);
+        }
+    }
+}
+
+/// Read stderr line by line: classify, forward to the sink, and retain the
+/// last [`STDERR_TAIL_LINES`] for a possible [`Error::Exit`]. Returns the
+/// newline-joined tail.
+async fn pump_stderr(stderr: ChildStderr, sink: &impl Sink, sentinels: &Sentinels) -> String {
+    let mut lines = BufReader::new(stderr).lines();
+    let mut tail: VecDeque<String> = VecDeque::with_capacity(STDERR_TAIL_LINES);
+    while let Ok(Some(line)) = lines.next_line().await {
+        sentinels.scan_line(&line);
+        sentinels.scan_rate_limit_text(&line);
+        sink.on_stderr_line(&line);
+        if tail.len() >= STDERR_TAIL_LINES {
+            tail.pop_front();
+        }
+        tail.push_back(line);
+    }
+    tail.into_iter().collect::<Vec<_>>().join("\n")
+}

hive-claude/src/error.rs

@@ -0,0 +1,77 @@
+//! Typed errors for the driver. See the crate-level docs for why this is a
+//! `thiserror` enum rather than `anyhow`.
+
+use std::process::ExitStatus;
+
+use thiserror::Error;
+
+/// Why a claude run did not complete cleanly. A normal, finished turn is
+/// `Ok(())`; everything else is one of these variants.
+///
+/// Two families share the enum on purpose, so a caller can handle them with a
+/// single `match` on the `Result`:
+///
+/// - **Recognized sentinels** — [`Error::PromptTooLong`],
+///   [`Error::RateLimited`], [`Error::AuthFailed`], [`Error::SessionNotFound`].
+///   These are *expected* non-completion states parsed from claude's output,
+///   not crashes; callers typically compact, park-and-retry, re-auth, or
+///   create-a-session in response.
+/// - **Hard failures** — [`Error::Spawn`], [`Error::Stdin`], [`Error::Wait`],
+///   [`Error::Exit`], [`Error::Io`]. The process couldn't run, or died with no
+///   recognizable reason.
+#[derive(Debug, Error)]
+pub enum Error {
+    /// `Prompt is too long` — the session is past the model's context window.
+    #[error("prompt is too long for the model's context window")]
+    PromptTooLong,
+
+    /// The API refused for rate-limit / usage-cap / credit-balance reasons.
+    #[error("request was rate-limited or hit a usage/credit cap")]
+    RateLimited,
+
+    /// The API rejected the request with 401 (auth/session expired or revoked).
+    #[error("authentication failed (HTTP 401)")]
+    AuthFailed,
+
+    /// `--resume` matched no session for the given id or title (e.g. the title
+    /// was never created, or its backing file was moved away).
+    #[error("no session matched the requested id or title")]
+    SessionNotFound,
+
+    /// The `claude` binary could not be spawned (not on `PATH`, not
+    /// executable, …).
+    #[error("failed to spawn `{program}`: {source}")]
+    Spawn {
+        /// The program name we tried to run.
+        program: String,
+        /// The underlying spawn error.
+        #[source]
+        source: std::io::Error,
+    },
+
+    /// Writing the prompt to claude's stdin failed.
+    #[error("writing prompt to claude stdin failed: {0}")]
+    Stdin(#[source] std::io::Error),
+
+    /// Awaiting the child process failed.
+    #[error("waiting on claude failed: {0}")]
+    Wait(#[source] std::io::Error),
+
+    /// claude exited non-zero and raised none of the recognized sentinels.
+    /// `stderr_tail` is the last handful of stderr lines (empty if there were
+    /// none), included so the caller can surface a real diagnostic.
+    #[error("claude exited {status}\n{stderr_tail}")]
+    Exit {
+        /// The child's exit status.
+        status: ExitStatus,
+        /// Tail of stderr, newline-joined; empty when claude wrote nothing.
+        stderr_tail: String,
+    },
+
+    /// A filesystem operation (session lookup / archive) failed.
+    #[error("session store i/o failed: {0}")]
+    Io(#[from] std::io::Error),
+}
+
+/// Convenience alias for results from this crate.
+pub type Result<T> = std::result::Result<T, Error>;

hive-claude/src/lib.rs

@@ -0,0 +1,74 @@
+//! `hive-claude` — a small, reusable async driver for headless
+//! `claude --print` (Claude Code CLI) sessions.
+//!
+//! It spawns the CLI, streams and classifies its `stream-json` output, and
+//! reports the result as a `Result<(), Error>`: a clean turn is `Ok(())`, and
+//! every non-completion state — both recognized sentinels (rate-limit,
+//! prompt-too-long, …) and hard failures (spawn, non-zero exit) — is a variant
+//! of the single [`Error`] enum, so callers branch with one `match`. The crate
+//! also locates and archives on-disk sessions by title ([`SessionStore`]). It
+//! knows only about the Claude Code CLI — no application types, hard-coded
+//! watermarks, or logging. Callers wire streaming output through a [`Sink`].
+//!
+//! Two layers:
+//!
+//! - [`Claude::run`] — the low-level driver: one turn, one [`Attach`] target.
+//! - [`InfiniteSession`] — a durable session (name + [`SessionStore`] +
+//!   [`CompactionPolicy`]) that keeps itself alive across the context window by
+//!   compacting reactively (on overflow) and proactively (per policy — e.g.
+//!   [`PercentPolicy`]). This is the one you usually want.
+//!
+//! # `thiserror` here, `anyhow` in the apps
+//!
+//! This is a **library**, so it exposes a concrete, matchable error enum built
+//! with [`thiserror`](https://docs.rs/thiserror): a caller can distinguish
+//! `Error::Exit { status, .. }` from `Error::Spawn { .. }` and branch on it.
+//! A library should never force its callers to reach into `anyhow`'s
+//! type-erased error to find out what went wrong.
+//!
+//! The **applications** in this workspace (the `hive-*` binaries) use
+//! [`anyhow`](https://docs.rs/anyhow) instead. At the top level you usually
+//! only want to attach context and log or bubble a failure up — not match on
+//! it — and `anyhow::Result` + `?` + `.context()` is the ergonomic fit.
+//! `anyhow::Error` implements `From<E>` for any `std::error::Error`, so a
+//! `hive_claude::Error` converts into an `anyhow::Error` for free at the `?`
+//! boundary. Rule of thumb: **libraries return `thiserror` enums, binaries
+//! consume them with `anyhow`.**
+//!
+//! # Example
+//!
+//! ```no_run
+//! # async fn ex() {
+//! use hive_claude::{Attach, Claude, Config, Error, NoopSink};
+//!
+//! let config = Config {
+//!     model: "haiku".into(),
+//!     ..Default::default()
+//! };
+//! match Claude::run(&config, &Attach::Resume("my-session".into()), "hello", &NoopSink).await {
+//!     Ok(()) => {}
+//!     Err(Error::PromptTooLong) => { /* caller compacts + retries */ }
+//!     Err(Error::RateLimited) => { /* caller parks + retries */ }
+//!     Err(other) => eprintln!("claude: {other}"),
+//! }
+//! # }
+//! ```
+
+mod classify;
+mod config;
+mod driver;
+mod error;
+mod policy;
+mod session;
+mod sink;
+mod store;
+mod telemetry;
+
+pub use config::{Attach, Config};
+pub use driver::Claude;
+pub use error::{Error, Result};
+pub use policy::{CompactionPolicy, NeverCompact, PercentPolicy};
+pub use session::{InfiniteSession, Progress};
+pub use sink::{NoopSink, Sink};
+pub use store::SessionStore;
+pub use telemetry::{Telemetry, TokenUsage, Usage};

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),
+        }));
+    }
+}

hive-claude/src/session.rs

@@ -0,0 +1,180 @@
+//! A durable, self-compacting ("infinite") claude session.
+
+use std::sync::Mutex;
+
+use serde_json::Value;
+
+use crate::{Attach, Claude, CompactionPolicy, Config, Error, Result, SessionStore, Sink, Telemetry};
+
+/// A named claude session that outlives the model's context window by
+/// compacting itself. Bundles the three things a durable session needs:
+///
+/// - a **name** (the constant session title it resumes / creates under),
+/// - a **store** ([`SessionStore`], to find the backing file so a resume vs.
+///   create is decided without a wasted spawn), and
+/// - a **policy** ([`CompactionPolicy`], deciding *when* to compact).
+///
+/// [`InfiniteSession::run`] keeps the session alive across turns:
+///
+/// - **resume-or-create** — resumes the titled session, creating it on first
+///   use (or after its file was archived away);
+/// - **reactive** — if a turn overflows ([`Error::PromptTooLong`]), it compacts
+///   and retries the same prompt once;
+/// - **proactive** — after a clean turn it consults the policy and, if due,
+///   runs an optional checkpoint turn then compacts.
+///
+/// Resetting/archiving the session is intentionally *not* part of this type —
+/// that stays with the caller.
+pub struct InfiniteSession<P: CompactionPolicy> {
+    name: String,
+    store: SessionStore,
+    policy: P,
+}
+
+/// What [`InfiniteSession::run`] did, beyond streaming the turn to the sink.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct Progress {
+    /// The turn resumed no existing session — a fresh one was created.
+    pub created: bool,
+    /// A compaction ran (reactively on overflow, or proactively per policy).
+    pub compacted: bool,
+    /// Everything parsed from the answering turn's stream (usage, cost,
+    /// context window, resolved model). The authoritative copy — finalised at
+    /// the turn's `result` event.
+    pub telemetry: Telemetry,
+}
+
+impl<P: CompactionPolicy> InfiniteSession<P> {
+    /// Build a durable session for `name`, backed by `store`, governed by
+    /// `policy`.
+    pub fn new(name: impl Into<String>, store: SessionStore, policy: P) -> Self {
+        Self {
+            name: name.into(),
+            store,
+            policy,
+        }
+    }
+
+    /// Run one turn, keeping the session infinite (see the type docs).
+    ///
+    /// # Errors
+    ///
+    /// Propagates any non-`SessionNotFound` [`Error`] from the underlying
+    /// runs. A reactive retry that *still* overflows surfaces as
+    /// [`Error::PromptTooLong`]; rate-limit / auth / hard failures propagate
+    /// unchanged for the caller to handle.
+    pub async fn run(&self, config: &Config, prompt: &str, sink: &impl Sink) -> Result<Progress> {
+        let meter = TelemetrySink::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; the retry is the answering
+                // turn, so its telemetry is what we report.
+                self.compact(config, sink).await?;
+                let retry = TelemetrySink::new(sink);
+                let created = self.attempt(config, prompt, &retry).await?;
+                return Ok(Progress {
+                    created,
+                    compacted: true,
+                    telemetry: retry.snapshot(),
+                });
+            }
+            Err(other) => return Err(other),
+        };
+        let telemetry = meter.snapshot();
+
+        // Proactive: the turn completed on a healthy session. If the policy
+        // says it's due, checkpoint (best-effort) then compact.
+        if self.policy.should_compact(telemetry.usage()) {
+            if let Some(checkpoint) = self.policy.checkpoint_prompt() {
+                let _ = self.attempt(config, checkpoint, sink).await;
+            }
+            let _ = self.compact(config, sink).await;
+            return Ok(Progress {
+                created,
+                compacted: true,
+                telemetry,
+            });
+        }
+        Ok(Progress {
+            created,
+            compacted: false,
+            telemetry,
+        })
+    }
+
+    /// Force a `/compact` on the session (e.g. operator-driven). Resume-only:
+    /// if the session doesn't exist there is nothing to compact, so a
+    /// [`Error::SessionNotFound`] is swallowed as a no-op `Ok`.
+    ///
+    /// # Errors
+    ///
+    /// Propagates any error other than `SessionNotFound` from the compact run.
+    pub async fn compact(&self, config: &Config, sink: &impl Sink) -> Result<()> {
+        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 turn's
+/// [`Telemetry`] from the stream. Cheap; the driver calls it synchronously
+/// from one reader task, so the `Mutex` only satisfies `&self`.
+struct TelemetrySink<'a, S: Sink> {
+    inner: &'a S,
+    telemetry: Mutex<Telemetry>,
+}
+
+impl<'a, S: Sink> TelemetrySink<'a, S> {
+    fn new(inner: &'a S) -> Self {
+        Self {
+            inner,
+            telemetry: Mutex::new(Telemetry::default()),
+        }
+    }
+
+    fn snapshot(&self) -> Telemetry {
+        self.telemetry.lock().unwrap().clone()
+    }
+}
+
+impl<S: Sink> Sink for TelemetrySink<'_, S> {
+    fn on_event(&self, event: &Value) {
+        self.telemetry.lock().unwrap().observe(event);
+        self.inner.on_event(event);
+    }
+
+    fn on_stdout_line(&self, line: &str) {
+        self.inner.on_stdout_line(line);
+    }
+
+    fn on_stderr_line(&self, line: &str) {
+        self.inner.on_stderr_line(line);
+    }
+}

hive-claude/src/sink.rs

@@ -0,0 +1,30 @@
+//! Streaming-output consumer hook.
+
+use serde_json::Value;
+
+/// Consumer callbacks for a claude run's output streams. Every method has a
+/// no-op default, so an implementor overrides only what it needs.
+///
+/// Methods are called synchronously from the stdout/stderr readers as lines
+/// arrive, so keep them cheap — the idiomatic body forwards to a channel or an
+/// event bus rather than blocking. The driver handles sentinel classification
+/// (rate-limit, prompt-too-long, …) itself; a sink only *observes* the stream.
+pub trait Sink {
+    /// A parsed `stream-json` object from stdout (assistant/result/system/…
+    /// event). The driver has already classified it for sentinels.
+    fn on_event(&self, _event: &Value) {}
+
+    /// A stdout line that was not valid JSON — occasional Claude Code CLI
+    /// chatter rather than conversation content.
+    fn on_stdout_line(&self, _line: &str) {}
+
+    /// A stderr line, delivered verbatim. The last several are also retained
+    /// by the driver for [`crate::Error::Exit`].
+    fn on_stderr_line(&self, _line: &str) {}
+}
+
+/// A [`Sink`] that discards everything. Useful for fire-and-forget runs where
+/// only the [`crate::Outcome`] matters.
+pub struct NoopSink;
+
+impl Sink for NoopSink {}

hive-claude/src/store.rs

@@ -0,0 +1,95 @@
+//! Locating and archiving on-disk claude sessions by title.
+
+use std::io::BufRead as _;
+use std::path::PathBuf;
+
+use crate::{Error, Result};
+
+/// On-disk claude session store scoped to one working directory.
+///
+/// Claude Code keeps sessions under
+/// `<claude_home>/projects/<slug(cwd)>/<uuid>.jsonl`, one project dir per cwd.
+/// This type locates and archives those files by their display title. It is
+/// generic Claude Code layout knowledge — no application specifics.
+#[derive(Debug, Clone)]
+pub struct SessionStore {
+    claude_home: PathBuf,
+    cwd: PathBuf,
+}
+
+impl SessionStore {
+    /// Build a store for `cwd`, with claude's home dir (normally `~/.claude`)
+    /// at `claude_home`.
+    pub fn new(claude_home: impl Into<PathBuf>, cwd: impl Into<PathBuf>) -> Self {
+        Self {
+            claude_home: claude_home.into(),
+            cwd: cwd.into(),
+        }
+    }
+
+    /// `<claude_home>/projects/<slug>` for this cwd. Claude slugises the
+    /// absolute cwd by replacing every `/` and `.` with `-` (verified against
+    /// claude 2.1.197 — e.g. `/agents/iris/state` → `-agents-iris-state`).
+    #[must_use]
+    pub fn project_dir(&self) -> PathBuf {
+        let slug: String = self
+            .cwd
+            .to_string_lossy()
+            .chars()
+            .map(|c| if c == '/' || c == '.' { '-' } else { c })
+            .collect();
+        self.claude_home.join("projects").join(slug)
+    }
+
+    /// Find the `<uuid>.jsonl` in the project dir whose `customTitle` equals
+    /// `title` (the value `--name` sets, stored in a `custom-title` event).
+    ///
+    /// Reads each session file line by line and stops at the first match, so a
+    /// huge transcript isn't slurped into memory. Returns `None` if no session
+    /// carries the title or the project dir is absent. Non-`.jsonl` files
+    /// (including anything already archived to `*.jsonl.archived`) are skipped.
+    #[must_use]
+    pub fn find_by_title(&self, title: &str) -> Option<PathBuf> {
+        let marker = format!("\"customTitle\":\"{title}\"");
+        for entry in std::fs::read_dir(self.project_dir()).ok()?.flatten() {
+            let path = entry.path();
+            if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
+                continue;
+            }
+            let Ok(file) = std::fs::File::open(&path) else {
+                continue;
+            };
+            if std::io::BufReader::new(file)
+                .lines()
+                .map_while(std::result::Result::ok)
+                .any(|line| line.contains(&marker))
+            {
+                return Some(path);
+            }
+        }
+        None
+    }
+
+    /// Archive the session titled `title` by renaming its backing file
+    /// `<uuid>.jsonl` → `<uuid>.jsonl.archived`. That drops it out of claude's
+    /// `*.jsonl` resolution glob (so a later `--resume <title>` misses) while
+    /// preserving the full transcript on disk. Only the file with a matching
+    /// `customTitle` is touched.
+    ///
+    /// Returns the archived file's new path, or `None` if no session carried
+    /// the title.
+    ///
+    /// # Errors
+    ///
+    /// [`Error::Io`] if the rename fails.
+    pub fn archive_by_title(&self, title: &str) -> Result<Option<PathBuf>> {
+        let Some(path) = self.find_by_title(title) else {
+            return Ok(None);
+        };
+        let mut target = path.clone().into_os_string();
+        target.push(".archived");
+        let target = PathBuf::from(target);
+        std::fs::rename(&path, &target).map_err(Error::Io)?;
+        Ok(Some(target))
+    }
+}

hive-claude/src/telemetry.rs

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