diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c55f3e..1379321 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - `tole doctor` (#378): tiered provider diagnostics — `offline` (wiring) and `catalog` (live `/models`, the model must be served); each blocker carries its fix command, the exit code is non-zero when the requested tier is not reached, and keys / URL credentials are never printed. +- Per-user-turn token stats (#379): `tole status` lists input / output / cache-read / cache-write / reasoning tokens per turn and the status JSON (`GET /sessions/{id}/status`, `tole_session_status`) gains an additive `usage_report.turns` array. Derived from the durable usage ledger, so old sessions read the same way and nothing is stored twice; a value no step reported is `n/a` / `null`, never 0, and no elapsed time is invented. ### Changed - Memory loop injection hygiene (#384): the recalled batch is hard-capped at 24 items / 64 KiB and deduplicated; a malformed recall payload (non-array, or a hit without content) now fails the whole batch visibly instead of silently injecting fewer items; `tole serve`/MCP sessions report recall failures on stderr like the CLI does. @@ -30,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `read_file` rejects an empty `path` with a clear message instead of a confusing "is a directory" error. ### Added +- Per-user-turn token stats (#379): `tole status` lists input / output / cache-read / cache-write / reasoning tokens per turn and the status JSON (`GET /sessions/{id}/status`, `tole_session_status`) gains an additive `usage_report.turns` array. Derived from the durable usage ledger, so old sessions read the same way and nothing is stored twice; a value no step reported is `n/a` / `null`, never 0, and no elapsed time is invented. - `tole status` and the serve status JSON now report token usage, cache-hit rate and a per-step request-size split (system+tools vs history, stored per step under `tole_wire` in the usage record); `fact/mission` gains `cached_tokens`. Instrumentation only: the wire, budgets and durable entry shapes are unchanged (#211, part 1 of 3). ### Changed diff --git a/README.md b/README.md index 1fe2fdf..2b4d338 100644 --- a/README.md +++ b/README.md @@ -298,7 +298,10 @@ the usage derived from the ledger for any session: prompt / completion / reasoning / cached tokens, the cache-hit rate, and the request size in characters split into system+tools vs history at the first and last step (#211; recorded per step under `tole_wire`; sessions recorded before it read -`n/a`). Reported only — `--max-tokens` accounting is unchanged. +`n/a`). It also lists per-user-turn stats (#379: input / output / cache-read / +cache-write / reasoning tokens, `turns` in the status JSON), summed from the same +ledger so they survive restarts; a value no step reported is `n/a` / `null`, +never 0. Reported only — `--max-tokens` accounting is unchanged. ## Approval gates diff --git a/crates/tole-cli/src/main.rs b/crates/tole-cli/src/main.rs index a3b5df0..96ebc15 100644 --- a/crates/tole-cli/src/main.rs +++ b/crates/tole-cli/src/main.rs @@ -2145,6 +2145,10 @@ fn status_command(sessions_dir: &Path, id: &str) -> Result<()> { for line in tole_cli::usage_report::usage_report(usage).render_lines() { println!("{line}"); } + // Issue #379: per user-turn stats from the same ledger. + for line in tole_cli::usage_report::turn_lines(storage.entries(), usage) { + println!("{line}"); + } Ok(()) } diff --git a/crates/tole-cli/src/serve.rs b/crates/tole-cli/src/serve.rs index 2a1e6cf..cef4654 100644 --- a/crates/tole-cli/src/serve.rs +++ b/crates/tole-cli/src/serve.rs @@ -675,7 +675,7 @@ fn route(state: &State, method: &str, path: &str, body: &str) -> (u16, serde_jso Ok(guard) => ( Some(guard.entries().len()), // Issue #211: additive usage report. - tole_cli::usage_report::usage_report(guard.usages()).to_json(), + tole_cli::usage_report::usage_json(guard.entries(), guard.usages()), ), // turn in flight — skip the count and report Err(_) => (None, json!(null)), diff --git a/crates/tole-cli/src/session_tools.rs b/crates/tole-cli/src/session_tools.rs index fc75459..a3152e7 100644 --- a/crates/tole-cli/src/session_tools.rs +++ b/crates/tole-cli/src/session_tools.rs @@ -402,7 +402,7 @@ impl Tool for SessionStatusTool { json!(guard.entries().len()), // Issue #211: additive usage report (same object as the // serve status JSON). - crate::usage_report::usage_report(guard.usages()).to_json(), + crate::usage_report::usage_json(guard.entries(), guard.usages()), ), Err(_) => (json!(null), json!(null)), // turn in flight }; diff --git a/crates/tole-cli/src/usage_report.rs b/crates/tole-cli/src/usage_report.rs index 509c25a..635002e 100644 --- a/crates/tole-cli/src/usage_report.rs +++ b/crates/tole-cli/src/usage_report.rs @@ -11,6 +11,7 @@ //! measured zero. use serde_json::{json, Value}; +use tole_core::entry::Entry; use tole_core::storage::UsageRecord; /// One step's request-size split, in characters (see @@ -166,6 +167,167 @@ pub fn usage_report(usages: &[UsageRecord]) -> UsageReport { r } +/// Cache-write (creation) tokens of one step: Anthropic-style +/// `cache_creation_input_tokens`, or the gateway variant +/// `prompt_tokens_details.cache_write_tokens`. +fn cache_write_of(u: &Value) -> Option { + u64_at(u, &["cache_creation_input_tokens"]) + .or_else(|| u64_at(u, &["prompt_tokens_details", "cache_write_tokens"])) +} + +/// Provider-reported token stats of one user turn (#379): the sum over the +/// provider steps of the turn. Derived at read time from the durable usage +/// ledger, so there is no second copy to drift and sessions recorded +/// before this existed read the same way. +/// +/// Every field is `None` until at least one step reported it: unknown is +/// never coerced to 0. No elapsed time is carried — it is not measured, so +/// it is not shown. +#[derive(Debug, Clone, PartialEq)] +pub struct TurnStats { + /// 1-based position of the turn in the session. + pub turn: u64, + /// Id of the user message that opened the turn. + pub user_entry: String, + /// Provider steps of the turn that reported usage. + pub steps: u64, + pub input_tokens: Option, + pub output_tokens: Option, + pub cache_read_tokens: Option, + pub cache_write_tokens: Option, + pub reasoning_tokens: Option, + /// Sum of the ledger rows that carry a cost; `None` when none does. + pub cost_usd: Option, +} + +fn add(acc: &mut Option, x: Option) { + if let Some(x) = x { + *acc = Some(acc.unwrap_or(0).saturating_add(x)); + } +} + +/// A turn opens at a user message, except the stop-gate's own feedback +/// message (the turn continues after it). +fn opens_turn(e: &Entry) -> bool { + e.kind.as_str() == "message" + && e.payload["role"] == json!("user") + && e.payload["stop_gate_feedback"] != json!(true) +} + +/// Fold the usage ledger into per-user-turn stats. A usage row belongs to +/// the turn that was open at the entry it is attached to. Never panics; +/// rows attached to unknown entries (or before the first user message) are +/// ignored rather than guessed into a turn. +pub fn turn_stats(entries: &[Entry], usages: &[UsageRecord]) -> Vec { + let starts: Vec = entries + .iter() + .enumerate() + .filter(|(_, e)| opens_turn(e)) + .map(|(i, _)| i) + .collect(); + let mut out: Vec = starts + .iter() + .enumerate() + .map(|(n, &i)| TurnStats { + turn: n as u64 + 1, + user_entry: entries[i].id.clone(), + steps: 0, + input_tokens: None, + output_tokens: None, + cache_read_tokens: None, + cache_write_tokens: None, + reasoning_tokens: None, + cost_usd: None, + }) + .collect(); + let index: std::collections::HashMap<&str, usize> = entries + .iter() + .enumerate() + .map(|(i, e)| (e.id.as_str(), i)) + .collect(); + for rec in usages { + let Some(&at) = index.get(rec.entry_id.as_str()) else { + continue; + }; + let open = starts.partition_point(|&s| s <= at); + let Some(t) = open.checked_sub(1).and_then(|k| out.get_mut(k)) else { + continue; + }; + if let Some(c) = rec.cost_usd { + t.cost_usd = Some(t.cost_usd.unwrap_or(0.0) + c); + } + if !reports_provider_usage(rec) { + continue; + } + let u = &rec.usage; + t.steps += 1; + add(&mut t.input_tokens, u64_at(u, &["prompt_tokens"])); + add(&mut t.output_tokens, u64_at(u, &["completion_tokens"])); + add(&mut t.cache_read_tokens, cached_of(u)); + add(&mut t.cache_write_tokens, cache_write_of(u)); + add( + &mut t.reasoning_tokens, + u64_at(u, &["completion_tokens_details", "reasoning_tokens"]), + ); + } + out +} + +/// The `usage_report` status object plus the additive per-turn array. +pub fn usage_json(entries: &[Entry], usages: &[UsageRecord]) -> Value { + let mut j = usage_report(usages).to_json(); + j["turns"] = Value::Array( + turn_stats(entries, usages) + .iter() + .map(TurnStats::to_json) + .collect(), + ); + j +} + +impl TurnStats { + /// Unknown values are `null` (never a fabricated zero). + pub fn to_json(&self) -> Value { + json!({ + "turn": self.turn, + "user_entry": self.user_entry, + "steps": self.steps, + "input_tokens": self.input_tokens, + "output_tokens": self.output_tokens, + "cache_read_tokens": self.cache_read_tokens, + "cache_write_tokens": self.cache_write_tokens, + "reasoning_tokens": self.reasoning_tokens, + "cost_usd": self.cost_usd, + }) + } + + /// One human line; unknown reads `n/a`. + pub fn render_line(&self) -> String { + let n = |v: Option| v.map_or("n/a".to_string(), |x| x.to_string()); + format!( + "turn {}: {} steps, {} in / {} out / {} cache read / {} cache write / {} reasoning", + self.turn, + self.steps, + n(self.input_tokens), + n(self.output_tokens), + n(self.cache_read_tokens), + n(self.cache_write_tokens), + n(self.reasoning_tokens), + ) + } +} + +/// Per-turn lines for `tole status` (empty when the session has no turn). +pub fn turn_lines(entries: &[Entry], usages: &[UsageRecord]) -> Vec { + let turns = turn_stats(entries, usages); + if turns.is_empty() { + return Vec::new(); + } + let mut lines = vec!["per turn:".to_string()]; + lines.extend(turns.iter().map(|t| format!(" {}", t.render_line()))); + lines +} + fn wire_json(w: &WirePoint) -> Value { json!({ "prefix_chars": w.prefix_chars, @@ -429,4 +591,193 @@ mod tests { assert_eq!(w.messages, None); assert_eq!(wire_json(&w)["messages"], Value::Null); } + + // ---- per-turn stats (#379) ---- + + fn entry(seq: u64, kind: &str, payload: Value) -> Entry { + Entry { + seq, + id: format!("e_{seq}"), + parent_id: None, + kind: tole_core::entry::EntryType::from_string(kind), + timestamp: 0, + payload, + } + } + + fn user(seq: u64) -> Entry { + entry(seq, "message", json!({"role": "user", "text": "hi"})) + } + + fn assistant(seq: u64) -> Entry { + entry(seq, "message", json!({"role": "assistant", "text": "ok"})) + } + + fn rec_at(entry_id: &str, usage: Value) -> UsageRecord { + UsageRecord { + id: String::new(), + entry_id: entry_id.into(), + usage, + cost_usd: None, + } + } + + #[test] + fn usage_is_grouped_by_user_turn_and_summed() { + let entries = vec![ + user(1), + assistant(2), + user(3), + // The stop gate's own feedback does not open a turn. + entry( + 4, + "message", + json!({"role": "user", "text": "gate", "stop_gate_feedback": true}), + ), + assistant(5), + ]; + let usages = vec![ + rec_at("e_1", json!({"prompt_tokens": 10, "completion_tokens": 2})), + rec_at("e_3", json!({"prompt_tokens": 20, "completion_tokens": 4})), + rec_at("e_4", json!({"prompt_tokens": 30, "completion_tokens": 6})), + ]; + let t = turn_stats(&entries, &usages); + assert_eq!(t.len(), 2); + assert_eq!( + (t[0].turn, t[0].user_entry.as_str(), t[0].steps), + (1, "e_1", 1) + ); + assert_eq!((t[0].input_tokens, t[0].output_tokens), (Some(10), Some(2))); + assert_eq!( + (t[1].turn, t[1].user_entry.as_str(), t[1].steps), + (2, "e_3", 2) + ); + assert_eq!( + (t[1].input_tokens, t[1].output_tokens), + (Some(50), Some(10)) + ); + } + + #[test] + fn unknown_stays_unknown_never_zero() { + let entries = vec![user(1), assistant(2), user(3), assistant(4)]; + let usages = vec![ + // Turn 1: prompt reported, nothing else. + rec_at("e_1", json!({"prompt_tokens": 7})), + // Turn 2: wire-only row — no provider usage at all. + rec_at("e_3", json!({"tole_wire": wire(1, 2, 3)})), + ]; + let t = turn_stats(&entries, &usages); + assert_eq!(t[0].input_tokens, Some(7)); + assert_eq!(t[0].output_tokens, None); + assert_eq!(t[0].cache_read_tokens, None); + assert_eq!(t[0].cache_write_tokens, None); + assert_eq!(t[0].reasoning_tokens, None); + assert_eq!(t[0].cost_usd, None); + assert_eq!(t[1].steps, 0); + assert_eq!(t[1].input_tokens, None); + let j = t[1].to_json(); + assert_eq!(j["input_tokens"], Value::Null); + assert_eq!(j["cost_usd"], Value::Null); + assert!( + t[1].render_line().contains("n/a in / n/a out"), + "{}", + t[1].render_line() + ); + } + + #[test] + fn a_step_that_reports_nothing_does_not_hide_the_one_that_does() { + let entries = vec![user(1), assistant(2)]; + let usages = vec![ + rec_at("e_1", json!({"prompt_tokens": 5, "cached_tokens": 3})), + rec_at("e_2", json!({"prompt_tokens": 6})), + ]; + let t = turn_stats(&entries, &usages); + assert_eq!(t[0].steps, 2); + assert_eq!(t[0].input_tokens, Some(11)); + // Only one step reported the cache: its value, not a diluted sum. + assert_eq!(t[0].cache_read_tokens, Some(3)); + } + + #[test] + fn cache_write_and_cost_are_read_from_known_variants() { + let entries = vec![user(1)]; + let mut a = rec_at("e_1", json!({"cache_creation_input_tokens": 4})); + a.cost_usd = Some(0.25); + let mut b = rec_at( + "e_1", + json!({"prompt_tokens_details": {"cache_write_tokens": 6}}), + ); + b.cost_usd = Some(0.5); + let t = turn_stats(&entries, &[a, b]); + assert_eq!(t[0].cache_write_tokens, Some(10)); + assert_eq!(t[0].cost_usd, Some(0.75)); + } + + #[test] + fn old_and_odd_sessions_still_fold() { + // No usage at all (a session recorded before usage existed). + let t = turn_stats(&[user(1), assistant(2)], &[]); + assert_eq!(t.len(), 1); + assert_eq!((t[0].steps, t[0].input_tokens), (0, None)); + // No turn at all. + assert!(turn_stats(&[], &[rec_at("e_1", json!({"prompt_tokens": 1}))]).is_empty()); + assert!(turn_lines(&[], &[]).is_empty()); + // A row attached to an unknown entry is ignored, not guessed. + let t = turn_stats(&[user(1)], &[rec_at("e_99", json!({"prompt_tokens": 9}))]); + assert_eq!(t[0].input_tokens, None); + } + + #[test] + fn stats_survive_a_restart_and_match_the_provider_block_exactly() { + use tole_core::entry::{EntryType, NewEntry}; + use tole_core::storage::{Commit, JsonlStorage, Storage}; + let dir = std::env::temp_dir().join(format!("tole-turnstats-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let provider_usage = json!({ + "prompt_tokens": 123, + "completion_tokens": 45, + "prompt_tokens_details": {"cached_tokens": 100}, + "completion_tokens_details": {"reasoning_tokens": 8} + }); + let before = { + let mut s = JsonlStorage::create(&dir, "ts", None).unwrap(); + let u = s + .commit(Commit::new().entry(NewEntry::root( + EntryType::new(EntryType::MESSAGE), + json!({"role": "user", "text": "hi"}), + ))) + .unwrap(); + s.commit(Commit::new().usage(UsageRecord { + id: String::new(), + entry_id: u[0].id.clone(), + usage: provider_usage.clone(), + cost_usd: None, + })) + .unwrap(); + turn_stats(s.entries(), s.usages()) + }; + // "Restart": reopen from disk. + let s = JsonlStorage::open(dir.join("ts.jsonl")).unwrap(); + let after = turn_stats(s.entries(), s.usages()); + assert_eq!(before, after); + assert_eq!(after.len(), 1); + let t = &after[0]; + assert_eq!( + ( + t.input_tokens, + t.output_tokens, + t.cache_read_tokens, + t.reasoning_tokens + ), + (Some(123), Some(45), Some(100), Some(8)) + ); + assert_eq!(t.cache_write_tokens, None, "nothing invented"); + let j = usage_json(s.entries(), s.usages()); + assert_eq!(j["turns"][0]["input_tokens"], json!(123)); + assert_eq!(j["turns"][0]["cache_write_tokens"], Value::Null); + let _ = std::fs::remove_dir_all(&dir); + } } diff --git a/docs/mobile-control.md b/docs/mobile-control.md index 4dbc683..c8c0b9c 100644 --- a/docs/mobile-control.md +++ b/docs/mobile-control.md @@ -48,6 +48,7 @@ meaning). Its shape — every unknown value is `null`, never a fabricated 0: | `cached_tokens` | sum of cached prompt tokens (`prompt_tokens_details.cached_tokens`, falling back to `cached_read_tokens` and the other gateway spellings); `null` if no step reported it | | `cache_hit_rate` | cached / prompt over the steps that reported cached tokens; `null` if unknown | | `wire` | `null` for sessions recorded before #211, else `{first, last, history_growth_per_step}`; `first`/`last` = `{prefix_chars, system_chars, tools_chars, history_chars, messages}` of the first/last step that has it (characters, not bytes; `prefix_chars` = system + tools) | +| `turns` | per user turn (#379), derived from the same ledger: `[{turn, user_entry, steps, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, reasoning_tokens, cost_usd}]`. Each token field is the sum over the turn's steps that reported it and `null` when none did (unknown is never 0); no elapsed time is carried. `[]` for a session with no turn | The mission cost report (`fact/mission`) is rendered by `tole status`; it gained an additive `cached_tokens` (`null` when no step reported it). The