diff --git a/src/openhuman/memory/chat.rs b/src/openhuman/memory/chat.rs index 023faf9cd..e52452377 100644 --- a/src/openhuman/memory/chat.rs +++ b/src/openhuman/memory/chat.rs @@ -13,7 +13,7 @@ use async_trait::async_trait; use crate::openhuman::config::{Config, DEFAULT_CLOUD_LLM_MODEL}; use crate::openhuman::inference::provider::{ - create_chat_provider, provider_for_role, ChatMessage, Provider, + create_chat_provider, provider_for_role, ChatMessage, ChatRequest, Provider, UsageInfo, }; /// One pair of prompt messages handed to the memory LLM backend. @@ -35,6 +35,24 @@ pub trait ChatProvider: Send + Sync { async fn chat_for_text(&self, prompt: &ChatPrompt) -> Result { self.chat_for_json(prompt).await } + + /// Like [`chat_for_text`], but also surfaces the provider-reported + /// [`UsageInfo`] (real token counts + `charged_amount_usd`) when the + /// backing provider returns it. + /// + /// The default implementation simply runs `chat_for_text` and reports + /// `None` usage, so implementors (e.g. test doubles, external impls) + /// that don't thread usage keep compiling unchanged. The production + /// `InferenceChatProvider` overrides this to route through the + /// inference `Provider::chat` API, which already parses usage out of + /// the backend response (see `compatible::extract_usage`). + async fn chat_for_text_with_usage( + &self, + prompt: &ChatPrompt, + ) -> Result<(String, Option)> { + let text = self.chat_for_text(prompt).await?; + Ok((text, None)) + } } struct InferenceChatProvider { @@ -54,6 +72,17 @@ impl InferenceChatProvider { } async fn run(&self, prompt: &ChatPrompt) -> Result { + let (text, _usage) = self.run_with_usage(prompt).await?; + Ok(text) + } + + /// Run the prompt through the inference `Provider::chat` API and return + /// both the text and the provider-reported usage. Memory historically + /// called `chat_with_history` (which returns only `String` and drops + /// the parsed usage); routing through `chat` instead lets us thread the + /// real token counts + `charged_amount_usd` into the sync audit log + /// (issue #3110) without re-deriving them from `body.len() / 4`. + async fn run_with_usage(&self, prompt: &ChatPrompt) -> Result<(String, Option)> { log::debug!( "[memory::chat] provider={} kind={} model={} sys_chars={} user_chars={}", self.display, @@ -68,19 +97,43 @@ impl InferenceChatProvider { ChatMessage::user(prompt.user.clone()), ]; - let text = self + let request = ChatRequest { + messages: &messages, + tools: None, + stream: None, + }; + + let response = self .inner - .chat_with_history(&messages, &self.model, prompt.temperature) + .chat(request, &self.model, prompt.temperature) .await?; + // Fail fast on a missing body rather than masking it as an empty + // string: an empty summary would still be ingested (and, post-#3110, + // counted against the run's real charge) as if it were valid output. + // The caller's fallback path (`fallback_summary`) is the correct + // recovery for a silent provider, and it only runs on `Err`. + let Some(text) = response.text else { + anyhow::bail!( + "inference provider '{}' returned no text for {} summarise request", + self.display, + prompt.kind + ); + }; + let usage = response.usage; + log::debug!( - "[memory::chat] provider={} kind={} response_chars={}", + "[memory::chat] provider={} kind={} response_chars={} usage_present={} input_tokens={} output_tokens={} charged_usd={}", self.display, prompt.kind, - text.len() + text.len(), + usage.is_some(), + usage.as_ref().map(|u| u.input_tokens).unwrap_or(0), + usage.as_ref().map(|u| u.output_tokens).unwrap_or(0), + usage.as_ref().map(|u| u.charged_amount_usd).unwrap_or(0.0), ); - Ok(text) + Ok((text, usage)) } } @@ -97,6 +150,13 @@ impl ChatProvider for InferenceChatProvider { async fn chat_for_text(&self, prompt: &ChatPrompt) -> Result { self.run(prompt).await } + + async fn chat_for_text_with_usage( + &self, + prompt: &ChatPrompt, + ) -> Result<(String, Option)> { + self.run_with_usage(prompt).await + } } fn routed_memory_config(config: &Config) -> Config { @@ -262,4 +322,22 @@ mod tests { assert_eq!(p.chat_for_json(&prompt).await.unwrap(), "hello"); assert_eq!(p.calls.load(std::sync::atomic::Ordering::SeqCst), 1); } + + #[tokio::test] + async fn chat_for_text_with_usage_default_impl_reports_no_usage() { + // A provider that doesn't override `chat_for_text_with_usage` + // (here the `chat_for_json`-only `StaticChatProvider`) must still + // return its text, with `None` usage — so summarise() falls back + // to the estimate rather than reporting a bogus zero charge. + let p = StaticChatProvider::new("summary text"); + let prompt = ChatPrompt { + system: "sys".into(), + user: "u".into(), + temperature: 0.0, + kind: "test", + }; + let (text, usage) = p.chat_for_text_with_usage(&prompt).await.unwrap(); + assert_eq!(text, "summary text"); + assert!(usage.is_none()); + } } diff --git a/src/openhuman/memory_sources/sync.rs b/src/openhuman/memory_sources/sync.rs index 08d8d80ab..f0c82316d 100644 --- a/src/openhuman/memory_sources/sync.rs +++ b/src/openhuman/memory_sources/sync.rs @@ -150,6 +150,7 @@ pub async fn sync_source(source: MemorySourceEntry, config: Config) -> Result<() input_tokens: 0, output_tokens: 0, estimated_cost_usd: 0.0, + actual_charged_usd: None, duration_ms, success: true, error: None, @@ -182,6 +183,7 @@ pub async fn sync_source(source: MemorySourceEntry, config: Config) -> Result<() input_tokens: 0, output_tokens: 0, estimated_cost_usd: 0.0, + actual_charged_usd: None, duration_ms, success: false, error: Some(error.clone()), @@ -379,7 +381,11 @@ async fn check_and_rebuild_tree(source: &MemorySourceEntry, config: &Config) { scope = %scope, files = outcome.files_read, batches = outcome.batches, - cost = %format!("${:.4}", outcome.estimated_cost_usd), + cost = %format!( + "${:.4}", + outcome.actual_charged_usd.unwrap_or(outcome.estimated_cost_usd) + ), + cost_is_actual = outcome.actual_charged_usd.is_some(), "[memory_sources:sync] rebuild complete" ); } diff --git a/src/openhuman/memory_sync/sources/audit.rs b/src/openhuman/memory_sync/sources/audit.rs index 3e84ddc49..f8d532107 100644 --- a/src/openhuman/memory_sync/sources/audit.rs +++ b/src/openhuman/memory_sync/sources/audit.rs @@ -21,12 +21,28 @@ pub struct SyncAuditEntry { pub items_fetched: u32, /// Number of summarise batches produced. pub batches: u32, - /// Estimated input tokens fed to the summariser (sum of item bodies / 4). + /// Input tokens fed to the summariser. Provider-reported when the + /// backend returned usage for every batch; otherwise estimated as the + /// sum of item bodies / 4. See [`SyncAuditEntry::actual_charged_usd`] + /// for the signal of which path produced the cost figure. pub input_tokens: u64, - /// Output tokens produced by the summariser. + /// Output tokens produced by the summariser. Provider-reported when + /// available, else estimated. pub output_tokens: u64, - /// Estimated cost in USD (input + output at model pricing). + /// Estimated cost in USD (input + output at hardcoded model pricing). + /// Always populated as a fallback so old audit entries — and runs + /// where the backend reported no charge — still render a cost. Prefer + /// [`SyncAuditEntry::actual_charged_usd`] when it is `Some`. pub estimated_cost_usd: f64, + /// Real amount billed by the backend in USD (sum of + /// `openhuman.billing.charged_amount_usd` across batches), when the + /// provider reported it for the run. `None` for runs that fell back to + /// the local estimate and for audit entries written before issue + /// #3110 (the `#[serde(default)]` keeps those deserialising). When + /// `Some`, this is the authoritative cost; renderers should show it in + /// preference to `estimated_cost_usd`. + #[serde(default)] + pub actual_charged_usd: Option, /// Duration of the sync in milliseconds. pub duration_ms: u64, /// Whether the sync completed successfully. @@ -99,6 +115,141 @@ pub fn estimate_cost_usd(input_tokens: u64, output_tokens: u64) -> f64 { input_cost + output_cost } +/// Accumulates per-batch token/charge figures over a sync run and decides +/// whether the run's totals may be reported as provider-"real" or must fall +/// back to the local `len/4` estimate. +/// +/// The rule (issue #3110): provider-reported tokens/charges are only +/// promoted to the run-level audit figure when **every** batch in the run +/// carried that signal. A run where some batches reported usage and others +/// fell back (provider silent / fallback summary) would otherwise produce a +/// partial "real" total that undercounts the run — worse than the estimate. +/// In that mixed case we keep the estimate, which covers all batches. +#[derive(Debug, Default, Clone)] +pub struct RealCostAccumulator { + total_batches: u32, + /// Number of batches that reported provider token usage (`input_tokens` + /// or `output_tokens` non-zero). + batches_with_usage: u32, + /// Number of batches that reported a provider charge. + batches_with_charge: u32, + est_input_tokens: u64, + est_output_tokens: u64, + real_input_tokens: u64, + real_output_tokens: u64, + real_charged_usd: f64, +} + +impl RealCostAccumulator { + pub fn new() -> Self { + Self::default() + } + + /// Fold one batch into the accumulator. + /// + /// `est_input` / `est_output` are the local-heuristic token counts for + /// the batch and are always summed. `real_input` / `real_output` are the + /// provider-reported counts (`0` = "no usage" sentinel). `charge` is the + /// provider charge for the batch when reported. + pub fn add_batch( + &mut self, + est_input: u64, + est_output: u64, + real_input: u64, + real_output: u64, + charge: Option, + ) { + self.total_batches += 1; + self.est_input_tokens += est_input; + self.est_output_tokens += est_output; + + // `input_tokens == 0 && output_tokens == 0` is the sentinel for "no + // usage" (set by the fallback path and by providers that don't report + // usage), so a batch only counts as carrying real usage when one of + // them is non-zero. + if real_input > 0 || real_output > 0 { + self.batches_with_usage += 1; + self.real_input_tokens += real_input; + self.real_output_tokens += real_output; + } + if let Some(charge) = charge { + self.batches_with_charge += 1; + self.real_charged_usd += charge; + } + } + + /// True when every batch reported provider token usage. Only then are the + /// real token totals complete enough to replace the estimate. + fn usage_is_complete(&self) -> bool { + self.total_batches > 0 && self.batches_with_usage == self.total_batches + } + + /// True when every batch reported a provider charge. Only then is the + /// summed charge a faithful total for the run. + fn charge_is_complete(&self) -> bool { + self.total_batches > 0 && self.batches_with_charge == self.total_batches + } + + /// Input tokens to record on the audit entry: real total when complete + /// across all batches, else the estimate. + pub fn audit_input_tokens(&self) -> u64 { + if self.usage_is_complete() { + self.real_input_tokens + } else { + self.est_input_tokens + } + } + + /// Output tokens to record on the audit entry. + pub fn audit_output_tokens(&self) -> u64 { + if self.usage_is_complete() { + self.real_output_tokens + } else { + self.est_output_tokens + } + } + + /// The hardcoded-pricing estimate over the run's estimated tokens — + /// always recorded as the fallback cost. + pub fn estimated_cost(&self) -> f64 { + estimate_cost_usd(self.est_input_tokens, self.est_output_tokens) + } + + /// The authoritative provider charge for the run when every batch + /// reported one, else `None` (falls back to the estimate downstream). + pub fn actual_charged_usd(&self) -> Option { + if self.charge_is_complete() { + Some(self.real_charged_usd) + } else { + None + } + } + + /// True when this run's token figures came from complete provider usage. + pub fn usage_is_real(&self) -> bool { + self.usage_is_complete() + } +} + +impl SyncAuditEntry { + /// The cost figure to display for this run: the real backend charge + /// when the provider reported one ([`Self::actual_charged_usd`]), + /// otherwise the hardcoded-pricing estimate ([`Self::estimated_cost_usd`]). + /// + /// Old audit entries (written before issue #3110) have no + /// `actual_charged_usd`, so they transparently fall back to the + /// estimate and keep rendering as before. + pub fn effective_cost_usd(&self) -> f64 { + self.actual_charged_usd.unwrap_or(self.estimated_cost_usd) + } + + /// True when the cost figure came from the backend's real billing + /// signal rather than the hardcoded-pricing estimate. + pub fn cost_is_actual(&self) -> bool { + self.actual_charged_usd.is_some() + } +} + #[cfg(test)] mod tests { use super::*; @@ -111,6 +262,58 @@ mod tests { assert!((cost - 0.0049).abs() < 0.0001); } + #[test] + fn accumulator_all_batches_real_promotes_usage_and_charge() { + let mut acc = RealCostAccumulator::new(); + acc.add_batch(1_000, 100, 900, 90, Some(0.005)); + acc.add_batch(1_000, 100, 800, 80, Some(0.004)); + + assert!(acc.usage_is_real()); + assert_eq!(acc.audit_input_tokens(), 1_700); + assert_eq!(acc.audit_output_tokens(), 170); + let charge = acc.actual_charged_usd().expect("charge complete"); + assert!((charge - 0.009).abs() < 1e-9); + } + + #[test] + fn accumulator_mixed_usage_falls_back_to_estimate() { + // One batch reports real usage, the second is silent (fallback). + // A partial real total (900/90) would *undercount* the run versus + // the estimate (2000/200), so we must keep the estimate. + let mut acc = RealCostAccumulator::new(); + acc.add_batch(1_000, 100, 900, 90, Some(0.005)); + acc.add_batch(1_000, 100, 0, 0, None); + + assert!(!acc.usage_is_real()); + assert_eq!(acc.audit_input_tokens(), 2_000); + assert_eq!(acc.audit_output_tokens(), 200); + // Charge incomplete (only one batch reported) → no actual charge. + assert_eq!(acc.actual_charged_usd(), None); + } + + #[test] + fn accumulator_usage_complete_but_charge_partial() { + // Every batch reports usage, but only one reports a charge. Tokens + // promote to real; charge stays None because the run-level sum would + // be missing the second batch's charge. + let mut acc = RealCostAccumulator::new(); + acc.add_batch(1_000, 100, 900, 90, Some(0.005)); + acc.add_batch(1_000, 100, 800, 80, None); + + assert!(acc.usage_is_real()); + assert_eq!(acc.audit_input_tokens(), 1_700); + assert_eq!(acc.actual_charged_usd(), None); + } + + #[test] + fn accumulator_no_batches_uses_estimate() { + let acc = RealCostAccumulator::new(); + assert!(!acc.usage_is_real()); + assert_eq!(acc.audit_input_tokens(), 0); + assert_eq!(acc.actual_charged_usd(), None); + assert_eq!(acc.estimated_cost(), 0.0); + } + #[test] fn append_creates_file_and_writes_jsonl() { let tmp = tempfile::TempDir::new().unwrap(); @@ -126,6 +329,7 @@ mod tests { input_tokens: 50_000, output_tokens: 5_000, estimated_cost_usd: 0.225, + actual_charged_usd: None, duration_ms: 12_000, success: true, error: None, @@ -139,4 +343,78 @@ mod tests { assert_eq!(lines.len(), 2); assert!(lines[0].contains("src_123")); } + + fn entry_with_costs(estimated: f64, actual: Option) -> SyncAuditEntry { + SyncAuditEntry { + timestamp: Utc::now(), + source_id: "src".to_string(), + source_kind: "github_repo".to_string(), + scope: "github:org/repo".to_string(), + items_fetched: 1, + batches: 1, + input_tokens: 100, + output_tokens: 10, + estimated_cost_usd: estimated, + actual_charged_usd: actual, + duration_ms: 1, + success: true, + error: None, + } + } + + #[test] + fn effective_cost_prefers_actual_charge_when_present() { + // An entry built from a provider `UsageInfo` carries the real + // backend charge — `effective_cost_usd` must return it, not the + // hardcoded-pricing estimate. + let entry = entry_with_costs(0.0049, Some(0.0123)); + assert!(entry.cost_is_actual()); + assert!((entry.effective_cost_usd() - 0.0123).abs() < f64::EPSILON); + } + + #[test] + fn effective_cost_falls_back_to_estimate_without_usage() { + // No provider usage → fall back to the estimate. + let entry = entry_with_costs(0.0049, None); + assert!(!entry.cost_is_actual()); + assert!((entry.effective_cost_usd() - 0.0049).abs() < f64::EPSILON); + } + + #[test] + fn old_entry_without_actual_field_deserializes_and_renders_estimate() { + // A pre-#3110 audit line has no `actual_charged_usd` key. The + // `#[serde(default)]` must let it deserialize, and the entry must + // render its estimate via `effective_cost_usd`. + let legacy = r#"{ + "timestamp": "2024-01-01T00:00:00Z", + "source_id": "src_old", + "source_kind": "github_repo", + "scope": "github:org/repo", + "items_fetched": 10, + "batches": 1, + "input_tokens": 50000, + "output_tokens": 5000, + "estimated_cost_usd": 0.0049, + "duration_ms": 12000, + "success": true + }"#; + let entry: SyncAuditEntry = serde_json::from_str(legacy).unwrap(); + assert_eq!(entry.actual_charged_usd, None); + assert!(!entry.cost_is_actual()); + assert!((entry.effective_cost_usd() - 0.0049).abs() < f64::EPSILON); + } + + #[test] + fn new_entry_roundtrips_actual_charge_through_jsonl() { + let tmp = tempfile::TempDir::new().unwrap(); + let path = tmp.path().join("roundtrip_audit.jsonl"); + let entry = entry_with_costs(0.0049, Some(0.0200)); + append_jsonl(&path, &entry).unwrap(); + + let content = std::fs::read_to_string(&path).unwrap(); + let line = content.lines().next().unwrap(); + let parsed: SyncAuditEntry = serde_json::from_str(line).unwrap(); + assert_eq!(parsed.actual_charged_usd, Some(0.0200)); + assert!((parsed.effective_cost_usd() - 0.0200).abs() < f64::EPSILON); + } } diff --git a/src/openhuman/memory_sync/sources/github.rs b/src/openhuman/memory_sync/sources/github.rs index 5e342fc5c..a56646602 100644 --- a/src/openhuman/memory_sync/sources/github.rs +++ b/src/openhuman/memory_sync/sources/github.rs @@ -16,7 +16,7 @@ use crate::openhuman::memory_store::content::raw::{self as raw_store, RawItem}; use crate::openhuman::memory_store::trees::types::TreeKind; use crate::openhuman::memory_store::trees::types::INPUT_TOKEN_BUDGET; use crate::openhuman::memory_sync::sources::audit::{ - append_audit_entry, estimate_cost_usd, SyncAuditEntry, + append_audit_entry, RealCostAccumulator, SyncAuditEntry, }; use crate::openhuman::memory_sync::traits::{SyncOutcome, SyncPipeline, SyncPipelineKind}; use crate::openhuman::memory_tree::ingest::{ingest_summary, SummaryIngestInput}; @@ -212,14 +212,15 @@ pub async fn run_github_sync( )), ); - let mut total_input_tokens: u64 = 0; - let mut total_output_tokens: u64 = 0; + // Token/charge accounting across the run. The estimate (`body.len() / 4` + // heuristic) is always summed; provider-reported figures only replace it + // when every batch reported them (issue #3110). See `RealCostAccumulator`. + let mut cost = RealCostAccumulator::new(); for (batch_idx, (batch_inputs, batch_labels, batch_basenames)) in batches.into_iter().enumerate() { let batch_input_tokens: u64 = batch_inputs.iter().map(|i| i.token_count as u64).sum(); - total_input_tokens += batch_input_tokens; let ctx = SummaryContext { tree_id: &tree.id, @@ -240,7 +241,13 @@ pub async fn run_github_sync( } }; - total_output_tokens += output.token_count as u64; + cost.add_batch( + batch_input_tokens, + output.token_count as u64, + output.input_tokens, + output.output_tokens, + output.charged_amount_usd, + ); let time_start = batch_inputs .iter() @@ -279,7 +286,29 @@ pub async fn run_github_sync( } let duration_ms = start.elapsed().as_millis() as u64; - let cost = estimate_cost_usd(total_input_tokens, total_output_tokens); + + // Provider token counts are recorded only when *every* batch reported + // usage; a mixed run keeps the `len/4` estimate (which covers all + // batches) rather than a partial real total that would undercount. + // `estimated_cost_usd` is always populated as the fallback. Issue #3110. + let any_real_usage = cost.usage_is_real(); + let audit_input_tokens = cost.audit_input_tokens(); + let audit_output_tokens = cost.audit_output_tokens(); + let estimated_cost = cost.estimated_cost(); + let actual_charged_usd = cost.actual_charged_usd(); + // Cost surfaced to the user/logs: real charge when present, else estimate. + let display_cost = actual_charged_usd.unwrap_or(estimated_cost); + + tracing::info!( + source_id = %source_id, + usage_is_real = any_real_usage, + actual_charge = actual_charged_usd.is_some(), + input_tokens = audit_input_tokens, + output_tokens = audit_output_tokens, + estimated_cost_usd = %format!("{estimated_cost:.4}"), + actual_charged_usd = ?actual_charged_usd, + "[memory_sync:github] sync cost accounting" + ); append_audit_entry( config, @@ -290,9 +319,10 @@ pub async fn run_github_sync( scope: repo_scope.clone(), items_fetched: input_count as u32, batches: batch_count as u32, - input_tokens: total_input_tokens, - output_tokens: total_output_tokens, - estimated_cost_usd: cost, + input_tokens: audit_input_tokens, + output_tokens: audit_output_tokens, + estimated_cost_usd: estimated_cost, + actual_charged_usd, duration_ms, success: true, error: None, @@ -305,7 +335,7 @@ pub async fn run_github_sync( Some(kind_str), Some(source_id), Some(format!( - "{input_count} items → {batch_count} summary(ies) ({total_input_tokens} in / {total_output_tokens} out tokens, ${cost:.4})" + "{input_count} items → {batch_count} summary(ies) ({audit_input_tokens} in / {audit_output_tokens} out tokens, ${display_cost:.4})" )), ); @@ -313,7 +343,7 @@ pub async fn run_github_sync( records_ingested: input_count as u32, more_pending: false, note: Some(format!( - "{input_count} items → {batch_count} summary(ies) (${cost:.4})" + "{input_count} items → {batch_count} summary(ies) (${display_cost:.4})" )), }) } diff --git a/src/openhuman/memory_sync/sources/rebuild.rs b/src/openhuman/memory_sync/sources/rebuild.rs index 0642add23..af70c3018 100644 --- a/src/openhuman/memory_sync/sources/rebuild.rs +++ b/src/openhuman/memory_sync/sources/rebuild.rs @@ -20,7 +20,7 @@ use crate::openhuman::memory_store::content::paths::slugify_source_id; use crate::openhuman::memory_store::content::raw::raw_source_dir; use crate::openhuman::memory_store::trees::types::{TreeKind, INPUT_TOKEN_BUDGET}; use crate::openhuman::memory_sync::sources::audit::{ - append_audit_entry, estimate_cost_usd, SyncAuditEntry, + append_audit_entry, RealCostAccumulator, SyncAuditEntry, }; use crate::openhuman::memory_tree::ingest::{ingest_summary, SummaryIngestInput}; use crate::openhuman::memory_tree::summarise::{ @@ -28,13 +28,17 @@ use crate::openhuman::memory_tree::summarise::{ }; /// Outcome of a rebuild operation. -#[derive(Clone, Debug)] +#[derive(Clone, Debug, Default)] pub struct RebuildOutcome { pub files_read: usize, pub batches: usize, pub input_tokens: u64, pub output_tokens: u64, pub estimated_cost_usd: f64, + /// Real amount billed by the backend in USD when the provider reported + /// usage for the run; `None` when it fell back to the estimate. Issue + /// #3110. Prefer this over `estimated_cost_usd` when `Some`. + pub actual_charged_usd: Option, } /// Check whether a source needs tree rebuilding: raw files exist on disk @@ -91,13 +95,7 @@ pub async fn rebuild_tree_from_raw(config: &Config, scope: &str) -> Result Result Result Result Result Result Result Result { } /// Output of a summarise call. -#[derive(Clone, Debug)] +#[derive(Clone, Debug, Default)] pub struct SummaryOutput { pub content: String, pub token_count: u32, @@ -71,6 +71,20 @@ pub struct SummaryOutput { /// design note). pub entities: Vec, pub topics: Vec, + /// Provider-reported prompt token count for this summarise call, when + /// the backend returned usage. `0` when usage was unavailable (e.g. + /// the [`fallback_summary`] path, or a provider that doesn't report + /// usage) — callers should fall back to their own estimate in that + /// case. Threaded into the sync audit log (issue #3110). + pub input_tokens: u64, + /// Provider-reported completion token count for this summarise call. + /// `0` when usage was unavailable (see [`Self::input_tokens`]). + pub output_tokens: u64, + /// Amount billed for this summarise call in USD, from the backend's + /// `openhuman.billing.charged_amount_usd`. `None` when the provider + /// did not report a charge — callers fall back to the hardcoded + /// pricing estimate (issue #3110). + pub charged_amount_usd: Option, } /// Fold `inputs` into a single summary by making one chat-provider call. @@ -95,12 +109,7 @@ pub async fn summarise( let body = build_user_prompt(inputs, per_input_cap); if body.trim().is_empty() { - return Ok(SummaryOutput { - content: String::new(), - token_count: 0, - entities: Vec::new(), - topics: Vec::new(), - }); + return Ok(SummaryOutput::default()); } let provider = @@ -122,18 +131,35 @@ pub async fn summarise( ctx.token_budget, ); - let raw = provider - .chat_for_text(&prompt) + let (raw, usage) = provider + .chat_for_text_with_usage(&prompt) .await .with_context(|| format!("memory_tree::summarise: provider={}", provider.name()))?; let (content, token_count) = clamp_to_budget(raw.trim(), effective_budget); + + // Prefer provider-reported usage (real token counts + charged amount) + // over our `body.len() / 4` estimate. `None`/zero means the backend + // didn't surface usage; downstream callers fall back to estimates. + let input_tokens = usage.as_ref().map(|u| u.input_tokens).unwrap_or(0); + let output_tokens = usage.as_ref().map(|u| u.output_tokens).unwrap_or(0); + let charged_amount_usd = usage.as_ref().and_then(|u| { + if u.charged_amount_usd > 0.0 { + Some(u.charged_amount_usd) + } else { + None + } + }); + log::debug!( - "[memory_tree::summarise] sealed tree_id={} level={} inputs={} tokens={}", + "[memory_tree::summarise] sealed tree_id={} level={} inputs={} tokens={} usage_input={} usage_output={} charged_usd={:?}", ctx.tree_id, ctx.target_level, inputs.len(), token_count, + input_tokens, + output_tokens, + charged_amount_usd, ); Ok(SummaryOutput { @@ -141,6 +167,9 @@ pub async fn summarise( token_count, entities: Vec::new(), topics: Vec::new(), + input_tokens, + output_tokens, + charged_amount_usd, }) } @@ -160,11 +189,17 @@ pub fn fallback_summary(inputs: &[SummaryInput], budget: u32) -> SummaryOutput { } let joined = parts.join("\n\n"); let (content, token_count) = clamp_to_budget(&joined, budget); + // No provider call happened on the fallback path, so there is no + // real usage to report — leave token counts at 0 and charge at None + // so callers fall back to their estimate. SummaryOutput { content, token_count, entities: Vec::new(), topics: Vec::new(), + input_tokens: 0, + output_tokens: 0, + charged_amount_usd: None, } } @@ -264,4 +299,139 @@ mod tests { assert!(out.content.contains("kept")); assert_eq!(out.content.matches("— ").count(), 1); } + + #[test] + fn fallback_reports_no_provider_usage() { + // The fallback path makes no provider call, so it must report + // zero/None usage — github/rebuild then fall back to the estimate. + let inputs = vec![sample_input("a", "hello")]; + let out = fallback_summary(&inputs, 10_000); + assert_eq!(out.input_tokens, 0); + assert_eq!(out.output_tokens, 0); + assert_eq!(out.charged_amount_usd, None); + } + + /// Test `ChatProvider` that reports provider usage (real token counts + /// + a backend charge) so we can prove `summarise` threads it into + /// `SummaryOutput`. + struct UsageReportingProvider { + response: String, + usage: Option, + } + + #[async_trait::async_trait] + impl crate::openhuman::memory::chat::ChatProvider for UsageReportingProvider { + fn name(&self) -> &str { + "test:usage-reporting" + } + + async fn chat_for_json(&self, _prompt: &ChatPrompt) -> Result { + Ok(self.response.clone()) + } + + async fn chat_for_text_with_usage( + &self, + _prompt: &ChatPrompt, + ) -> Result<( + String, + Option, + )> { + Ok((self.response.clone(), self.usage.clone())) + } + } + + fn summary_ctx<'a>(tree_id: &'a str) -> SummaryContext<'a> { + SummaryContext { + tree_id, + tree_kind: TreeKind::Source, + target_level: 1, + token_budget: 5_000, + } + } + + #[tokio::test] + async fn summarise_threads_provider_usage_into_output() { + use crate::openhuman::inference::provider::UsageInfo; + use crate::openhuman::memory::chat::test_override; + + let provider = std::sync::Arc::new(UsageReportingProvider { + response: "a folded summary".to_string(), + usage: Some(UsageInfo { + input_tokens: 1_234, + output_tokens: 56, + charged_amount_usd: 0.0078, + ..Default::default() + }), + }); + + let cfg = Config::default(); + let inputs = vec![sample_input("a", "raw content to fold")]; + let ctx = summary_ctx("tree:test"); + + let out = + test_override::with_provider(provider, async { summarise(&cfg, &inputs, &ctx).await }) + .await + .unwrap(); + + assert_eq!(out.input_tokens, 1_234); + assert_eq!(out.output_tokens, 56); + assert_eq!(out.charged_amount_usd, Some(0.0078)); + assert!(out.content.contains("folded summary")); + } + + #[tokio::test] + async fn summarise_leaves_usage_empty_when_provider_reports_none() { + use crate::openhuman::memory::chat::test_override; + + let provider = std::sync::Arc::new(UsageReportingProvider { + response: "a folded summary".to_string(), + usage: None, + }); + + let cfg = Config::default(); + let inputs = vec![sample_input("a", "raw content to fold")]; + let ctx = summary_ctx("tree:test"); + + let out = + test_override::with_provider(provider, async { summarise(&cfg, &inputs, &ctx).await }) + .await + .unwrap(); + + // No usage → callers must fall back to their estimate. + assert_eq!(out.input_tokens, 0); + assert_eq!(out.output_tokens, 0); + assert_eq!(out.charged_amount_usd, None); + } + + #[tokio::test] + async fn summarise_treats_zero_charge_as_absent() { + use crate::openhuman::inference::provider::UsageInfo; + use crate::openhuman::memory::chat::test_override; + + // A provider that reports token counts but a zero charge (backend + // didn't surface billing) — token counts flow through, but the + // charge must be `None` so callers fall back to the estimate. + let provider = std::sync::Arc::new(UsageReportingProvider { + response: "a folded summary".to_string(), + usage: Some(UsageInfo { + input_tokens: 100, + output_tokens: 10, + charged_amount_usd: 0.0, + ..Default::default() + }), + }); + + let cfg = Config::default(); + let inputs = vec![sample_input("a", "raw content to fold")]; + let ctx = summary_ctx("tree:test"); + + let out = + test_override::with_provider(provider, async { summarise(&cfg, &inputs, &ctx).await }) + .await + .unwrap(); + + assert_eq!(out.input_tokens, 100); + assert_eq!(out.output_tokens, 10); + assert_eq!(out.charged_amount_usd, None); + } }