diff --git a/crates/cm-api/src/microvm_turn_executor.rs b/crates/cm-api/src/microvm_turn_executor.rs index d6880e4..11ecf02 100644 --- a/crates/cm-api/src/microvm_turn_executor.rs +++ b/crates/cm-api/src/microvm_turn_executor.rs @@ -295,6 +295,7 @@ impl TurnExecutor for MicroVmTurnExecutor { // the honest value for "not measured on this path". tokens: 0, gated: Vec::new(), + spend: Default::default(), }) } } diff --git a/crates/cm-api/src/recursive_exec.rs b/crates/cm-api/src/recursive_exec.rs index 0d2ccd9..5c71b3a 100644 --- a/crates/cm-api/src/recursive_exec.rs +++ b/crates/cm-api/src/recursive_exec.rs @@ -173,6 +173,7 @@ impl TurnExecutor for SubTopologyExecutor { output: record.final_output, tokens: record.totals.tokens, gated, + spend: Default::default(), }) } } diff --git a/crates/cm-api/src/swarm.rs b/crates/cm-api/src/swarm.rs index 4249bd1..dfa7af7 100644 --- a/crates/cm-api/src/swarm.rs +++ b/crates/cm-api/src/swarm.rs @@ -86,6 +86,7 @@ fn step( output: output.into(), gated: Vec::new(), tokens: 0, + spend: Default::default(), } } diff --git a/crates/cm-api/src/topology_exec.rs b/crates/cm-api/src/topology_exec.rs index d6142de..45b23e7 100644 --- a/crates/cm-api/src/topology_exec.rs +++ b/crates/cm-api/src/topology_exec.rs @@ -683,6 +683,7 @@ impl ZeroClawDriveExecutor { { let mut output = String::new(); let mut tokens: u64 = 0; + let mut spend = cm_orchestrator::Spend::default(); let mut gated: Vec = Vec::new(); let mut trace = ToolTrace::default(); @@ -720,6 +721,23 @@ impl ZeroClawDriveExecutor { let input = v.get("input_tokens").and_then(|n| n.as_u64()).unwrap_or(0); let out = v.get("output_tokens").and_then(|n| n.as_u64()).unwrap_or(0); tokens = input + out; + // The frame has always carried these; only + // `tokens` was read, so every agent turn was + // charged with no record of who was paid. + spend = cm_orchestrator::Spend { + input_tokens: input, + output_tokens: out, + provider: v + .get("provider") + .and_then(|p| p.as_str()) + .filter(|p| !p.is_empty()) + .map(str::to_string), + model: v + .get("model") + .and_then(|m| m.as_str()) + .filter(|m| !m.is_empty()) + .map(str::to_string), + }; break; } "approval_request" => { @@ -807,6 +825,7 @@ impl ZeroClawDriveExecutor { output: output.trim().to_string(), tokens, gated, + spend, }, trace, )) @@ -968,7 +987,10 @@ mod tests { json!({"type": "session_start", "session_id": "s1", "resumed": false}), json!({"type": "chunk", "content": "hel"}), json!({"type": "chunk", "content": "lo"}), - json!({"type": "done", "input_tokens": 5, "output_tokens": 7}), + // The real frame carries model and provider; the executor read + // only the two token counts until 2026-09-14. + json!({"type": "done", "input_tokens": 5, "output_tokens": 7, + "model": "claude-sonnet-5", "provider": "anthropic"}), ] } @@ -1068,6 +1090,16 @@ mod tests { let out = exec.run_turn(req()).await.unwrap(); assert_eq!(out.output, "hello"); assert_eq!(out.tokens, 12); + assert_eq!( + out.spend, + cm_orchestrator::Spend { + input_tokens: 5, + output_tokens: 7, + provider: Some("anthropic".into()), + model: Some("claude-sonnet-5".into()), + }, + "the split and the provider must survive the done frame, not just the sum" + ); assert!(out.gated.is_empty()); } diff --git a/crates/cm-api/src/topology_worker.rs b/crates/cm-api/src/topology_worker.rs index 1e5db07..267f009 100644 --- a/crates/cm-api/src/topology_worker.rs +++ b/crates/cm-api/src/topology_worker.rs @@ -556,16 +556,35 @@ async fn drive( } if let Some(agent_id) = agent_of.get(&last.node_id).copied() { if last.tokens > 0 { - // The executor reports ONE total, not an in/out split. - // Credits price the sum, so cost is right; the columns - // record it as output rather than inventing a split. + // The split and the provider come from the runtime's + // `done` frame via `StepRecord.spend`. An executor + // that reports only a total leaves the split at 0/0 + // and the total goes on the output side, as before. + let (tin, tout) = if last.spend.input_tokens + last.spend.output_tokens > 0 + { + (last.spend.input_tokens, last.spend.output_tokens) + } else { + (0, last.tokens as u64) + }; + let mission_id: Option = sqlx::query_scalar::<_, Option>( + "SELECT mission_id FROM topology_runs WHERE id = $1", + ) + .bind(id) + .fetch_optional(&pool) + .await + .ok() + .flatten() + .flatten(); if let Err(e) = cm_billing::charge( &pool, cm_domain::WorkspaceId::from(workspace_id), cm_domain::AgentId::from(agent_id), None, - 0, - last.tokens as u64, + tin, + tout, + last.spend.provider.as_deref(), + last.spend.model.as_deref(), + mission_id, ) .await { diff --git a/crates/cm-billing/src/lib.rs b/crates/cm-billing/src/lib.rs index aaa0601..1dd01c0 100644 --- a/crates/cm-billing/src/lib.rs +++ b/crates/cm-billing/src/lib.rs @@ -36,21 +36,36 @@ pub async fn charge( run_id: Option, input_tokens: u64, output_tokens: u64, + // Who was paid, and for which mission. `None` where the executor did not + // say. Until 2026-09-14 no agent-side row carried these, so the larger + // half of the spend could not be asked per provider — the judge's half + // could, and that is how a plan emptying twice went unexplained. + provider: Option<&str>, + model: Option<&str>, + mission_id: Option, ) -> Result { let owed = credits_for_tokens(input_tokens + output_tokens); let mut tx = pool.begin().await?; - sqlx::query!( + // `sqlx::query`, not `query!`: the macro pins this statement to offline + // metadata that a schema change then has to regenerate against a live + // database, and the columns added by migration 0085 are nullable text + // and uuid — nothing here that a compile-time check would catch. + sqlx::query( "INSERT INTO usage_events - (workspace_id, agent_id, run_id, kind, tokens_in, tokens_out, credits) - VALUES ($1, $2, $3, 'llm_tokens', $4, $5, $6)", - workspace_id.as_uuid(), - agent_id.as_uuid(), - run_id, - input_tokens as i64, - output_tokens as i64, - sqlx::types::BigDecimal::from(owed), + (workspace_id, agent_id, run_id, kind, tokens_in, tokens_out, credits, + provider, model, mission_id) + VALUES ($1, $2, $3, 'llm_tokens', $4, $5, $6, $7, $8, $9)", ) + .bind(workspace_id.as_uuid()) + .bind(agent_id.as_uuid()) + .bind(run_id) + .bind(input_tokens as i64) + .bind(output_tokens as i64) + .bind(sqlx::types::BigDecimal::from(owed)) + .bind(provider) + .bind(model) + .bind(mission_id) .execute(&mut *tx) .await?; diff --git a/crates/cm-billing/tests/billing.rs b/crates/cm-billing/tests/billing.rs index d6c5aea..167bc6f 100644 --- a/crates/cm-billing/tests/billing.rs +++ b/crates/cm-billing/tests/billing.rs @@ -62,7 +62,7 @@ async fn charges_span_lots_oldest_first_and_record_usage() { .unwrap(); // 2500 tokens → 3 credits: drains the first lot (2) then one more. - let deducted = charge(&pool, ws.id, agent.id, Some(run_id), 1500, 1000) + let deducted = charge(&pool, ws.id, agent.id, Some(run_id), 1500, 1000, None, None, None) .await .unwrap(); assert_eq!(deducted, 3); @@ -96,7 +96,7 @@ async fn an_empty_workspace_records_usage_but_clamps_at_zero() { .unwrap(); // Owes 5, only 1 available: deducts 1, balance hits zero, never negative. - let deducted = charge(&pool, ws.id, agent.id, Some(run_id), 4000, 500) + let deducted = charge(&pool, ws.id, agent.id, Some(run_id), 4000, 500, None, None, None) .await .unwrap(); assert_eq!(deducted, 1); diff --git a/crates/cm-orchestrator/src/evolve.rs b/crates/cm-orchestrator/src/evolve.rs index 6ea1082..78379b3 100644 --- a/crates/cm-orchestrator/src/evolve.rs +++ b/crates/cm-orchestrator/src/evolve.rs @@ -145,6 +145,7 @@ mod tests { output: format!("{}<{}>", req.role, req.context.join("|")), tokens: 10, gated: vec![], + spend: Default::default(), }) } } diff --git a/crates/cm-orchestrator/src/harness.rs b/crates/cm-orchestrator/src/harness.rs index 2ccc252..07b6cda 100644 --- a/crates/cm-orchestrator/src/harness.rs +++ b/crates/cm-orchestrator/src/harness.rs @@ -147,6 +147,7 @@ mod tests { output: format!("{}:{}", req.role, req.context.join(" ")), tokens: 10, gated, + spend: Default::default(), }) } } diff --git a/crates/cm-orchestrator/src/lib.rs b/crates/cm-orchestrator/src/lib.rs index 6c754ad..9f060d8 100644 --- a/crates/cm-orchestrator/src/lib.rs +++ b/crates/cm-orchestrator/src/lib.rs @@ -95,15 +95,35 @@ pub struct TurnRequest { pub context: Vec, } +/// What a turn cost and who was paid — the part of the runtime's `done` frame +/// that `tokens` alone threw away. +/// +/// `tokens` stayed as the one total every reader already keys on. This is +/// the split beside it, plus the provider and model that answered, so the +/// spend can be asked per provider BEFORE a plan limit asks it for you. Judge +/// spend gained this on 2026-09-14 and agent spend did not, which left the +/// larger of the two invisible. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct Spend { + pub input_tokens: u64, + pub output_tokens: u64, + /// Provider family that answered (`anthropic`, `glm`, …), when the + /// runtime said. `None` on executors that do not report one. + pub provider: Option, + pub model: Option, +} + /// The result of a single agent turn. #[derive(Debug, Clone)] pub struct TurnOutcome { /// The turn's textual output. pub output: String, - /// Model tokens spent (cost proxy). + /// Model tokens spent (cost proxy). Input + output. pub tokens: u64, /// Any sandbox-leaving actions attempted during the turn. pub gated: Vec, + /// The split and the provider behind `tokens`. + pub spend: Spend, } /// Runs one safe agent turn. The real impl wraps `cm-runtime::Runtime` @@ -143,6 +163,10 @@ pub struct StepRecord { pub gated: Vec, /// Tokens it spent. pub tokens: u64, + /// The split and provider behind `tokens`. `default` so checkpoints + /// journaled before this field existed still load. + #[serde(default)] + pub spend: Spend, } /// The full record of a topology run (journal + final output + totals). @@ -264,6 +288,7 @@ where output: outcome.output, gated: outcome.gated, tokens: outcome.tokens, + spend: outcome.spend, }); // Hand the caller a durable snapshot to persist before the next turn. @@ -309,6 +334,7 @@ mod tests { output: format!("{}({})<{}>", req.role, req.node_id, req.context.join("|")), tokens: 10, gated, + spend: Default::default(), }) } } @@ -383,6 +409,7 @@ mod tests { output: format!("{}({})<{}>", req.role, req.node_id, req.context.join("|")), tokens: 10, gated: vec![], + spend: Default::default(), }) } } diff --git a/crates/cm-orchestrator/src/provider_executor.rs b/crates/cm-orchestrator/src/provider_executor.rs index 9744f6e..96dca42 100644 --- a/crates/cm-orchestrator/src/provider_executor.rs +++ b/crates/cm-orchestrator/src/provider_executor.rs @@ -89,6 +89,7 @@ impl TurnExecutor for ProviderExecutor { tokens, // Tool-free reasoning turns leave the sandbox nowhere. gated: vec![], + spend: Default::default(), }) } } diff --git a/crates/cm-orchestrator/src/workflow.rs b/crates/cm-orchestrator/src/workflow.rs index 2ef635c..ea07425 100644 --- a/crates/cm-orchestrator/src/workflow.rs +++ b/crates/cm-orchestrator/src/workflow.rs @@ -74,6 +74,7 @@ mod tests { output: format!("{}<{}>", req.node_id, req.context.join("|")), tokens: 5, gated: vec![], + spend: Default::default(), }) } } diff --git a/crates/cm-runtime/src/runtime.rs b/crates/cm-runtime/src/runtime.rs index 37f394e..045f050 100644 --- a/crates/cm-runtime/src/runtime.rs +++ b/crates/cm-runtime/src/runtime.rs @@ -875,6 +875,13 @@ impl Runtime { ); // Meter the run (§8.4). Billing failures never fail the run — the // usage ledger is the recovery path. + // This loop drives ONE provider with no fallback chain, so the model + // the request named is the model that answered. The family is taken + // only from an explicit `provider:` prefix — a bare model name is + // recorded as-is with no family rather than guessed at, and a chat + // run is not a mission. + let model = state.request.model.as_str(); + let provider = model.split_once(':').map(|(p, _)| p); if let Err(error) = cm_billing::charge( &self.inner.pool, state.workspace_id, @@ -882,6 +889,9 @@ impl Runtime { Some(run_id), state.input_tokens, state.output_tokens, + provider, + Some(model), + None, ) .await {