From 487e62a9ca2a997340387326e9ef2a19b5f0753c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Sun, 5 Jul 2026 20:06:30 +0200 Subject: [PATCH] feat(hive-claude): parse turn telemetry in the lib, return it from run --- hive-ag3nt/src/events.rs | 112 +--------------------- hive-ag3nt/src/turn.rs | 86 ++++++++--------- hive-claude/README.md | 8 +- hive-claude/src/lib.rs | 4 +- hive-claude/src/session.rs | 49 +++++----- hive-claude/src/telemetry.rs | 181 +++++++++++++++++++++++++++++++++++ hive-claude/src/usage.rs | 47 --------- 7 files changed, 260 insertions(+), 227 deletions(-) create mode 100644 hive-claude/src/telemetry.rs delete mode 100644 hive-claude/src/usage.rs diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index a433ae91..a9cf80ca 100644 --- a/hive-ag3nt/src/events.rs +++ b/hive-ag3nt/src/events.rs @@ -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 { - 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 { - 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..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 { - 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 { - 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); - } } diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index be112cf8..4f2455ce 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -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, -} - -#[derive(Default)] -struct BusSinkState { - last_inference: Option, - last_model: Option, } 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 ` misses and self-heals into a fresh `--name <title>` /// session. Delegates the rename to [`hive_claude::SessionStore::archive_by_title`] diff --git a/hive-claude/README.md b/hive-claude/README.md index 1a8525d4..957b7776 100644 --- a/hive-claude/README.md +++ b/hive-claude/README.md @@ -22,7 +22,8 @@ Two layers — reach for the high-level one: `Result<Progress, Error>` does resume-or-create, compacts **reactively** on overflow (compact + retry once), and **proactively** after a clean turn when the policy says so (optional checkpoint turn, then compact). `.compact(…)` - forces one. `Progress { created, compacted }` reports what happened. + 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 @@ -38,8 +39,9 @@ Supporting pieces: - **`Sink`** — a trait with no-op defaults; implement what you care about to observe stream events, non-JSON stdout, and stderr. `NoopSink` ignores all. - **`SessionStore`** — locate and archive on-disk sessions by title. -- **`Usage`** — the minimal context signal (`context_tokens`, - `context_window`) the policy sees. +- **`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 diff --git a/hive-claude/src/lib.rs b/hive-claude/src/lib.rs index 068d8c5d..40cde76a 100644 --- a/hive-claude/src/lib.rs +++ b/hive-claude/src/lib.rs @@ -62,7 +62,7 @@ mod policy; mod session; mod sink; mod store; -mod usage; +mod telemetry; pub use config::{Attach, Config}; pub use driver::Claude; @@ -71,4 +71,4 @@ pub use policy::{CompactionPolicy, NeverCompact, PercentPolicy}; pub use session::{InfiniteSession, Progress}; pub use sink::{NoopSink, Sink}; pub use store::SessionStore; -pub use usage::Usage; +pub use telemetry::{Telemetry, TokenUsage, Usage}; diff --git a/hive-claude/src/session.rs b/hive-claude/src/session.rs index ae2d7059..4fded3b5 100644 --- a/hive-claude/src/session.rs +++ b/hive-claude/src/session.rs @@ -4,7 +4,7 @@ use std::sync::Mutex; use serde_json::Value; -use crate::{Attach, Claude, CompactionPolicy, Config, Error, Result, SessionStore, Sink, Usage, usage}; +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: @@ -32,12 +32,16 @@ pub struct InfiniteSession<P: CompactionPolicy> { } /// What [`InfiniteSession::run`] did, beyond streaming the turn to the sink. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +#[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> { @@ -60,26 +64,30 @@ impl<P: CompactionPolicy> InfiniteSession<P> { /// [`Error::PromptTooLong`]; rate-limit / auth / hard failures propagate /// unchanged for the caller to handle. pub async fn run(&self, config: &Config, prompt: &str, sink: &impl Sink) -> Result<Progress> { - let meter = UsageSink::new(sink); + let 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. + // 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 created = self.attempt(config, prompt, 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(meter.snapshot()) { + if self.policy.should_compact(telemetry.usage()) { if let Some(checkpoint) = self.policy.checkpoint_prompt() { let _ = self.attempt(config, checkpoint, sink).await; } @@ -87,11 +95,13 @@ impl<P: CompactionPolicy> InfiniteSession<P> { return Ok(Progress { created, compacted: true, + telemetry, }); } Ok(Progress { created, compacted: false, + telemetry, }) } @@ -133,35 +143,30 @@ impl<P: CompactionPolicy> InfiniteSession<P> { } } -/// A [`Sink`] that forwards to an inner sink while accumulating the minimal -/// [`Usage`] the policy needs from the stream. Cheap; the driver calls it -/// synchronously from one reader task, so the `Mutex` only satisfies `&self`. -struct UsageSink<'a, S: Sink> { +/// 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, - usage: Mutex<Usage>, + telemetry: Mutex<Telemetry>, } -impl<'a, S: Sink> UsageSink<'a, S> { +impl<'a, S: Sink> TelemetrySink<'a, S> { fn new(inner: &'a S) -> Self { Self { inner, - usage: Mutex::new(Usage::default()), + telemetry: Mutex::new(Telemetry::default()), } } - fn snapshot(&self) -> Usage { - *self.usage.lock().unwrap() + fn snapshot(&self) -> Telemetry { + self.telemetry.lock().unwrap().clone() } } -impl<S: Sink> Sink for UsageSink<'_, S> { +impl<S: Sink> Sink for TelemetrySink<'_, S> { fn on_event(&self, event: &Value) { - if let Some(tokens) = usage::context_tokens(event) { - self.usage.lock().unwrap().context_tokens = tokens; - } - if let Some(window) = usage::context_window(event) { - self.usage.lock().unwrap().context_window = Some(window); - } + self.telemetry.lock().unwrap().observe(event); self.inner.on_event(event); } diff --git a/hive-claude/src/telemetry.rs b/hive-claude/src/telemetry.rs new file mode 100644 index 00000000..3c1c490f --- /dev/null +++ b/hive-claude/src/telemetry.rs @@ -0,0 +1,181 @@ +//! 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_json::Value; + +/// Token counts from one `usage` block. All in tokens; missing fields read `0`. +#[derive(Debug, Clone, Copy, Default, 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 { + /// 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)] +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)] +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)); + } +} diff --git a/hive-claude/src/usage.rs b/hive-claude/src/usage.rs deleted file mode 100644 index 8bb83839..00000000 --- a/hive-claude/src/usage.rs +++ /dev/null @@ -1,47 +0,0 @@ -//! Minimal context-usage signal parsed from the stream, used to drive -//! [`crate::CompactionPolicy`]. This is deliberately small — just what a -//! compaction decision needs. Consumers that want full per-turn accounting -//! parse the raw events in their own [`crate::Sink`]. - -use serde_json::Value; - -/// The context footprint of the most recent inference in a turn. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -pub struct Usage { - /// Tokens in the last inference's context (input + cache reads + cache - /// writes) — what counts against the model's window. `0` until the first - /// `assistant` event is seen. - pub context_tokens: u64, - /// The model-reported active context window (`modelUsage.*.contextWindow` - /// on the terminal `result` event), if the turn reported one. - pub context_window: Option<u64>, -} - -/// Per-inference context size from an `assistant` event's `message.usage`. -/// Tracking the *last* one over a turn gives the live conversation size (the -/// cumulative `result` usage double-counts tool-call prompts and overshoots). -pub(crate) fn context_tokens(event: &Value) -> Option<u64> { - if event.get("type").and_then(Value::as_str) != Some("assistant") { - return None; - } - let usage = event.get("message")?.get("usage")?; - let field = |k: &str| usage.get(k).and_then(Value::as_u64).unwrap_or(0); - Some(field("input_tokens") + field("cache_read_input_tokens") + field("cache_creation_input_tokens")) -} - -/// The per-inference active window from a `result` event's `modelUsage` map -/// (first non-zero `contextWindow` across model keys). This is the limit the -/// model actually enforces, which can be far below the prompt-cache capacity. -pub(crate) fn context_window(event: &Value) -> Option<u64> { - if event.get("type").and_then(Value::as_str) != Some("result") { - return None; - } - for (_model, stats) in event.get("modelUsage")?.as_object()? { - if let Some(w) = stats.get("contextWindow").and_then(Value::as_u64) - && w > 0 - { - return Some(w); - } - } - None -}