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

feat(hive-claude): parse turn telemetry in the lib, return it from run

Browse source
This commit is contained in:
müde 2026-07-05 20:06:30 +02:00
commit 487e62a9ca
7 changed files with 260 additions and 227 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -453,89 +453,6 @@ impl TokenUsage {
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
@ -1225,8 +1142,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;
@ -1305,29 +1222,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);
}
}

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

@ -6,7 +6,6 @@
//! compaction / auto-reset / retry state machine (`drive_turn`).
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use anyhow::Result;
use hive_claude::{Config, InfiniteSession, PercentPolicy, Sink};
@ -285,6 +284,9 @@ pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutco
}
let outcome = match result {
Ok(progress) => {
// Apply the turn's parsed usage / model / context-window to the bus
// (badges, stats, auto-reset watermark input).
apply_telemetry(bus, &progress.telemetry);
if progress.created {
// Fresh session minted this turn → flag it so the bin loop
// mints a `sessions` row + stamps its id onto this turn's stats.
@ -495,59 +497,27 @@ fn error_to_turn(err: hive_claude::Error) -> TurnOutcome {
}
}
/// Bridges a claude run's output stream onto the hyperhive event bus: parses
/// per-turn token usage / resolved model / context-window from stream-json,
/// mirrors every event to the SSE bus, and surfaces non-JSON stdout + stderr
/// as Notes. Interior mutability (a `Mutex`) tracks the last inference across
/// events; the driver calls the `Sink` methods synchronously from one reader
/// task, so contention is nil — the lock only satisfies the `&self` trait
/// signature (and keeps `BusSink: Sync` for the driver's `Send` future).
/// Bridges a claude run's raw output stream onto the hyperhive event bus:
/// per-turn tool-call counting (`observe_stream`), the live SSE stream, and
/// non-JSON stdout + stderr as Notes. Stateless — usage/model/context-window
/// parsing lives in `hive-claude` and is applied from the run's returned
/// `Telemetry` (see `apply_telemetry`).
struct BusSink<'a> {
bus: &'a Bus,
state: Mutex<BusSinkState>,
}
#[derive(Default)]
struct BusSinkState {
last_inference: Option<TokenUsage>,
last_model: Option<String>,
}
impl<'a> BusSink<'a> {
fn new(bus: &'a Bus) -> Self {
Self {
bus,
state: Mutex::new(BusSinkState::default()),
}
Self { bus }
}
}
impl Sink for BusSink<'_> {
fn on_event(&self, event: &Value) {
{
let mut st = self.state.lock().unwrap();
// `last_inference` overwrites on every assistant event so at
// result-time it holds the most recent model call's usage — the
// actual context size. The `result` event carries the cumulative
// cost usage; both update the badges together.
if let Some(u) = TokenUsage::from_assistant_event(event) {
st.last_inference = Some(u);
}
if let Some(m) = TokenUsage::model_from_assistant_event(event) {
st.last_model = Some(m);
}
if let Some(cost) = TokenUsage::from_stream_event(event) {
let ctx = st.last_inference.unwrap_or(cost);
self.bus.record_turn_usage(ctx, cost);
self.bus.set_resolved_model(st.last_model.clone());
}
}
// Seed the API-reported context-window from the result event's
// `modelUsage.*.contextWindow` — the authoritative active window for
// compaction watermarks.
if let Some(w) = TokenUsage::context_window_from_result_event(event) {
self.bus.set_api_context_window(w);
}
// Raw-event concerns only: per-turn tool-call counting + the live SSE
// stream. Usage / model / context-window parsing lives in the lib now
// and is applied from the run's returned `Telemetry` (see `drive_turn`
// → `apply_telemetry`).
self.bus.observe_stream(event);
self.bus.emit(LiveEvent::Stream(event.clone()));
}
@ -568,6 +538,36 @@ impl Sink for BusSink<'_> {
}
}
/// Apply a completed turn's parsed [`hive_claude::Telemetry`] to the bus:
/// per-inference context usage + cumulative cost, the resolved model id, and
/// the API-reported context window (the authoritative window for the auto-reset
/// watermark). Skips a degenerate turn that parsed nothing so it doesn't reset
/// the badges to zero.
fn apply_telemetry(bus: &Bus, telemetry: &hive_claude::Telemetry) {
if telemetry.context.context_tokens() == 0 && telemetry.cost.context_tokens() == 0 {
return;
}
bus.record_turn_usage(
to_bus_usage(telemetry.context),
to_bus_usage(telemetry.cost),
);
bus.set_resolved_model(telemetry.model.clone());
if let Some(window) = telemetry.context_window {
bus.set_api_context_window(window);
}
}
/// Convert the lib's `TokenUsage` into the bus/stats `TokenUsage` (identical
/// fields; the two crates keep their own types to avoid coupling).
fn to_bus_usage(u: hive_claude::TokenUsage) -> TokenUsage {
TokenUsage {
input_tokens: u.input_tokens,
output_tokens: u.output_tokens,
cache_read_input_tokens: u.cache_read_input_tokens,
cache_creation_input_tokens: u.cache_creation_input_tokens,
}
}
/// Archive (do NOT delete) the harness's own session so the next turn's
/// `--resume <title>` misses and self-heals into a fresh `--name <title>`
/// session. Delegates the rename to [`hive_claude::SessionStore::archive_by_title`]