car_inference/outcome.rs
1//! Outcome tracking — learn from inference results to improve routing.
2//!
3//! Two observation channels:
4//! 1. **Conversation signals** — implicit feedback from what happens after an inference
5//! call (user moved on = accepted, user corrected = rejected, re-asked = rejected).
6//! 2. **Git-diff tracking** — for code generation, compare suggestions to actual commits
7//! (ground truth, no classification model needed).
8//!
9//! Every inference call produces an `InferenceOutcome`. Outcomes accumulate into
10//! `ModelProfile`s with per-task statistics. The adaptive router uses profiles
11//! to make data-driven model selection.
12
13use std::collections::{HashMap, HashSet};
14use std::time::{SystemTime, UNIX_EPOCH};
15
16use serde::{Deserialize, Serialize};
17
18/// Task type for outcome tracking. Maps to ModelCapability but at the call level.
19#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
20#[serde(rename_all = "snake_case")]
21pub enum InferenceTask {
22 Generate,
23 Embed,
24 Classify,
25 Code,
26 Reasoning,
27}
28
29impl std::fmt::Display for InferenceTask {
30 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
31 match self {
32 InferenceTask::Generate => write!(f, "generate"),
33 InferenceTask::Embed => write!(f, "embed"),
34 InferenceTask::Classify => write!(f, "classify"),
35 InferenceTask::Code => write!(f, "code"),
36 InferenceTask::Reasoning => write!(f, "reasoning"),
37 }
38 }
39}
40
41/// A single inference invocation record.
42#[derive(Debug, Clone, Serialize, Deserialize)]
43pub struct InferenceOutcome {
44 /// Unique trace ID for this invocation.
45 pub trace_id: String,
46 /// Model that was used.
47 pub model_id: String,
48 /// Task type.
49 pub task: InferenceTask,
50 /// How the model was selected.
51 pub routing_reason: String,
52 /// Wall-clock latency in milliseconds.
53 pub latency_ms: u64,
54 /// Input tokens (estimated). For a prompt-cached provider this is the
55 /// *uncached* prefix only; the cached portion is split out below.
56 pub input_tokens: usize,
57 /// Output tokens (estimated).
58 pub output_tokens: usize,
59 /// Prompt-cache hit input tokens (Anthropic `cache_read_input_tokens`),
60 /// billed at ~0.1× base input. `0` for uncached / non-caching providers.
61 #[serde(default)]
62 pub cache_read_input_tokens: usize,
63 /// Prompt-cache write input tokens (Anthropic `cache_creation_input_tokens`),
64 /// billed at ~1.25×/2× base input. `0` for uncached / non-caching providers.
65 #[serde(default)]
66 pub cache_creation_input_tokens: usize,
67 /// Outcome from conversation signal inference.
68 pub inferred_outcome: Option<InferredOutcome>,
69 /// Outcome from git-diff tracking (code only).
70 pub code_outcome: Option<CodeOutcome>,
71 /// Error message if inference failed.
72 pub error: Option<String>,
73 /// Unix timestamp.
74 pub timestamp: u64,
75 /// Whether a *mechanical* success has already been credited to the
76 /// model profile for this call (booked at completion when it produced
77 /// output with no error — see [`OutcomeTracker::record_complete`]).
78 /// Guards against the later pending-sweep or a downstream quality
79 /// signal double-counting the same call. In-memory only (`pending` is
80 /// never persisted), so it's skipped from (de)serialization.
81 #[serde(skip)]
82 pub success_credited: bool,
83}
84
85/// Outcome inferred from conversation flow (implicit feedback).
86#[derive(Debug, Clone, Serialize, Deserialize)]
87#[serde(tag = "type", rename_all = "snake_case")]
88pub enum InferredOutcome {
89 /// User moved on, built on the response.
90 Accepted { confidence: f64 },
91 /// User used the result but modified it.
92 AcceptedWithEdits { confidence: f64 },
93 /// User corrected, re-asked, or explicitly rejected.
94 Rejected { confidence: f64 },
95 /// No follow-up signal (session ended, inconclusive).
96 Inconclusive,
97}
98
99impl InferredOutcome {
100 /// Convert to a quality score (0.0 = bad, 1.0 = good).
101 pub fn quality_score(&self) -> Option<f64> {
102 match self {
103 InferredOutcome::Accepted { confidence } => Some(*confidence),
104 InferredOutcome::AcceptedWithEdits { confidence } => Some(confidence * 0.7),
105 InferredOutcome::Rejected { confidence } => Some((1.0 - confidence) * 0.3),
106 InferredOutcome::Inconclusive => None,
107 }
108 }
109
110 pub fn is_success(&self) -> Option<bool> {
111 match self {
112 InferredOutcome::Accepted { .. } => Some(true),
113 InferredOutcome::AcceptedWithEdits { .. } => Some(true),
114 InferredOutcome::Rejected { .. } => Some(false),
115 InferredOutcome::Inconclusive => None,
116 }
117 }
118}
119
120/// Outcome from git-diff comparison (code generation ground truth).
121#[derive(Debug, Clone, Serialize, Deserialize)]
122#[serde(tag = "type", rename_all = "snake_case")]
123pub enum CodeOutcome {
124 /// Suggestion was applied as-is (exact or near-exact match in diff).
125 Applied,
126 /// User changed the same file but differently (partial adoption).
127 Modified,
128 /// File unchanged despite suggestion (rejected / not used).
129 Ignored,
130 /// AST structural diff: signature was changed (breaking change).
131 SignatureChanged,
132 /// AST structural diff: body was modified but signature preserved (non-breaking).
133 BodyModified,
134 /// AST structural diff: new symbol was added.
135 SymbolAdded,
136}
137
138impl CodeOutcome {
139 pub fn quality_score(&self) -> f64 {
140 match self {
141 CodeOutcome::Applied => 1.0,
142 CodeOutcome::SignatureChanged => 0.8,
143 CodeOutcome::BodyModified => 0.7,
144 CodeOutcome::SymbolAdded => 0.7,
145 CodeOutcome::Modified => 0.6,
146 CodeOutcome::Ignored => 0.1,
147 }
148 }
149
150 pub fn is_success(&self) -> bool {
151 !matches!(self, CodeOutcome::Ignored)
152 }
153}
154
155/// Backfill `quality_observations` for a profile hydrated from a pre-upgrade
156/// snapshot (where the field deserializes to 0). Failures are always graded,
157/// so `fail_count` is a safe lower bound on how many observations moved the
158/// EMA — using it preserves learned distrust across the upgrade. Only fills
159/// when the count is still 0, so profiles written by this version are untouched.
160///
161/// A profile at neutral quality learned no distrust, so there is nothing to
162/// preserve: [`plan_reconcile`] returns a profile whose failures were all
163/// no-answer to exactly that state, and refilling its weight from
164/// `fail_count` would undo the rebuild on every load.
165fn backfill_quality_observations(p: &mut ModelProfile) {
166 if p.quality_observations == 0 && p.fail_count > 0 && p.ema_quality < NEUTRAL_QUALITY {
167 p.quality_observations = p.fail_count;
168 }
169 for ts in p.task_stats.values_mut() {
170 if ts.quality_observations == 0 && ts.failures > 0 && ts.ema_quality < NEUTRAL_QUALITY {
171 ts.quality_observations = ts.failures;
172 }
173 }
174}
175
176/// The neutral quality prior a fresh profile starts at.
177const NEUTRAL_QUALITY: f64 = 0.5;
178
179/// Per-task statistics within a model profile.
180#[derive(Debug, Clone, Default, Serialize, Deserialize)]
181pub struct TaskStats {
182 pub calls: u64,
183 pub successes: u64,
184 pub failures: u64,
185 /// Running average latency in ms.
186 pub avg_latency_ms: f64,
187 /// Exponential moving average of quality score.
188 pub ema_quality: f64,
189 /// Number of benchmark cases behind `ema_quality` when it is a *seeded
190 /// cold-start prior* (i.e. `calls == 0`). Lets the router shrink the
191 /// seeded score toward its schema estimate by sample size. `0` means no
192 /// benchmark evidence — the router then sits at the schema estimate.
193 #[serde(default)]
194 pub prior_sample_size: usize,
195 /// Number of *graded* quality observations that have moved `ema_quality`
196 /// for this task — accept/edit/reject signals and failures, NOT mechanical
197 /// successes (which leave answer quality unknown). The router trusts
198 /// `ema_quality` in proportion to this count, so a task with many ungraded
199 /// successes stays anchored to its schema estimate instead of a hollow
200 /// neutral EMA.
201 #[serde(default)]
202 pub quality_observations: u64,
203}
204
205impl TaskStats {
206 pub fn success_rate(&self) -> f64 {
207 let total = self.successes + self.failures;
208 if total == 0 {
209 return 0.5;
210 } // prior: assume neutral
211 self.successes as f64 / total as f64
212 }
213}
214
215/// Per-model performance profile, built from observed outcomes.
216#[derive(Debug, Clone, Serialize, Deserialize)]
217pub struct ModelProfile {
218 pub model_id: String,
219 pub total_calls: u64,
220 pub success_count: u64,
221 pub fail_count: u64,
222 pub total_latency_ms: u64,
223 /// Total estimated input tokens across all calls. For prompt-cached
224 /// providers this is the *uncached* prefix only; cached tokens are in
225 /// the two fields below so each bucket can be priced at its own rate.
226 #[serde(default)]
227 pub total_input_tokens: u64,
228 /// Total estimated output tokens across all calls.
229 #[serde(default)]
230 pub total_output_tokens: u64,
231 /// Total prompt-cache *read* (hit) input tokens — billed at ~0.1× input.
232 #[serde(default)]
233 pub total_cache_read_input_tokens: u64,
234 /// Total prompt-cache *write* input tokens — billed at ~1.25×/2× input.
235 #[serde(default)]
236 pub total_cache_creation_input_tokens: u64,
237 /// Per-task statistics.
238 pub task_stats: HashMap<String, TaskStats>,
239 /// Overall EMA quality score (0.0 - 1.0).
240 pub ema_quality: f64,
241 /// Number of benchmark cases behind `ema_quality` when it is a *seeded
242 /// cold-start prior* (`total_calls == 0`). See `TaskStats::prior_sample_size`.
243 #[serde(default)]
244 pub prior_sample_size: usize,
245 /// Graded quality observations that have moved `ema_quality` (overall).
246 /// See `TaskStats::quality_observations`.
247 #[serde(default)]
248 pub quality_observations: u64,
249 /// Derived metric: quality per 1K total tokens. Populated on export
250 /// (not on every update) so it always reflects the latest snapshot.
251 /// Inspired by Meta-Harness: context-token efficiency is a first-class
252 /// optimization target, so it needs to be visible in model_stats.
253 #[serde(default)]
254 pub quality_per_1k_tokens: f64,
255 /// Last updated (unix timestamp).
256 pub updated_at: u64,
257}
258
259impl ModelProfile {
260 pub fn new(model_id: String) -> Self {
261 Self {
262 model_id,
263 total_calls: 0,
264 success_count: 0,
265 fail_count: 0,
266 total_latency_ms: 0,
267 total_input_tokens: 0,
268 total_output_tokens: 0,
269 total_cache_read_input_tokens: 0,
270 total_cache_creation_input_tokens: 0,
271 task_stats: HashMap::new(),
272 ema_quality: NEUTRAL_QUALITY,
273 prior_sample_size: 0,
274 quality_observations: 0,
275 quality_per_1k_tokens: 0.0,
276 updated_at: now_unix(),
277 }
278 }
279
280 /// Success rate over *resolved* outcomes, or `None` when nothing has
281 /// resolved yet. Returns `None` rather than a fabricated 0.5 so callers
282 /// (all of which are display/reporting surfaces) can render "no resolved
283 /// signal" instead of a misleading confident "50%" for a never-measured
284 /// model. The router does NOT use this — its Thompson sampler reads the
285 /// raw `success_count`/`fail_count` and blends its own schema-derived
286 /// Beta prior (`adaptive_router`), so no 0.5 floor belongs here. Mirrors
287 /// `LaneUsage::success_rate`'s Option semantics.
288 pub fn success_rate_resolved(&self) -> Option<f64> {
289 let resolved = self.success_count + self.fail_count;
290 if resolved == 0 {
291 None
292 } else {
293 Some(self.success_count as f64 / resolved as f64)
294 }
295 }
296
297 pub fn avg_latency_ms(&self) -> f64 {
298 if self.total_calls == 0 {
299 return 0.0;
300 }
301 self.total_latency_ms as f64 / self.total_calls as f64
302 }
303
304 /// Same degradation pattern as SkillStats: fail_count > success_count + threshold.
305 pub fn should_degrade(&self, threshold: u64) -> bool {
306 self.fail_count > self.success_count + threshold
307 }
308
309 /// Get stats for a specific task type.
310 pub fn task_stats(&self, task: InferenceTask) -> Option<&TaskStats> {
311 self.task_stats.get(&task.to_string())
312 }
313
314 /// Total tokens observed across all calls (all input buckets + output).
315 /// Includes prompt-cache read and write tokens — they are real tokens the
316 /// provider processed, so price-free token efficiency must count them.
317 pub fn total_tokens(&self) -> u64 {
318 self.total_input_tokens
319 + self.total_cache_read_input_tokens
320 + self.total_cache_creation_input_tokens
321 + self.total_output_tokens
322 }
323
324 /// Quality per 1000 tokens: `ema_quality * 1000 / total_tokens`.
325 /// Returns 0.0 before any tokens have been observed.
326 pub fn compute_quality_per_1k_tokens(&self) -> f64 {
327 let total = self.total_tokens();
328 if total == 0 {
329 return 0.0;
330 }
331 self.ema_quality * 1000.0 / total as f64
332 }
333
334 /// Tokens spent per *successful* outcome: total tokens ÷ successes. `None`
335 /// before any success has resolved. This is PRICE-FREE token efficiency —
336 /// NOT dollar cost-per-outcome: a $0.10/Mtok and a $10/Mtok model with equal
337 /// token counts score identically here. For the dollar metric the
338 /// outcome-scoreboard vision wants, use [`Self::usd_per_success`] with the
339 /// model's catalog prices. Pairs with [`Self::success_rate_resolved`].
340 pub fn tokens_per_success(&self) -> Option<f64> {
341 if self.success_count == 0 {
342 None
343 } else {
344 Some(self.total_tokens() as f64 / self.success_count as f64)
345 }
346 }
347
348 /// Dollar cost per *successful* outcome — the honest "cost-per-outcome":
349 /// input+output token spend priced at the model's catalog rates, divided by
350 /// successes. `None` before any success. Prices are USD per 1M tokens
351 /// (from the catalog/registry `cost`), supplied by the caller so the tracker
352 /// stays catalog-free. This is the metric the scoreboard should lead with —
353 /// "cry once" is only legible in dollars-per-outcome, not tokens.
354 pub fn usd_per_success(
355 &self,
356 input_per_mtok: f64,
357 output_per_mtok: f64,
358 cache: CacheRates,
359 ) -> Option<f64> {
360 if self.success_count == 0 {
361 return None;
362 }
363 let usd = priced_input_usd(
364 self.total_input_tokens,
365 self.total_cache_read_input_tokens,
366 self.total_cache_creation_input_tokens,
367 input_per_mtok,
368 cache,
369 ) + self.total_output_tokens as f64 * output_per_mtok / 1_000_000.0;
370 Some(usd / self.success_count as f64)
371 }
372}
373
374/// Per-provider prompt-cache economics, expressed relative to the base input
375/// rate. The providers differ structurally: Anthropic uses explicit
376/// breakpoints with a deep read discount and a write premium; OpenAI caches
377/// automatically with a shallower read discount and no write charge. The
378/// durable ledger is catalog-free, so the caller derives these from the model
379/// schema's protocol (see [`crate::schema::ApiProtocol::cache_rates`]) and
380/// passes them in alongside the per-token prices.
381#[derive(Debug, Clone, Copy, PartialEq)]
382pub struct CacheRates {
383 /// Cached-read (hit) price ÷ base input. Anthropic ~0.1×, OpenAI ~0.5×.
384 pub read_mult: f64,
385 /// Cache-write price ÷ base input. Anthropic ~1.25× (5-minute TTL; 1-hour
386 /// is 2× but the ledger doesn't record which TTL, so writes bill at 1.25×
387 /// — exact for the default path, a conservative under-estimate for 1-hour).
388 /// OpenAI has no write charge (`0`) and reports no write tokens.
389 pub write_mult: f64,
390}
391
392impl CacheRates {
393 /// Anthropic / Bedrock-Claude: deep read discount + write premium.
394 pub const ANTHROPIC: Self = Self {
395 read_mult: 0.1,
396 write_mult: 1.25,
397 };
398 /// OpenAI (incl. Azure OpenAI): automatic caching, ~0.5× read, no write.
399 pub const OPENAI: Self = Self {
400 read_mult: 0.5,
401 write_mult: 0.0,
402 };
403 /// Providers without parsed prompt caching (Google/Vertex, local). Their
404 /// cache token counts are always zero, so the multipliers are never
405 /// exercised — this is just an honest, inert default.
406 pub const NONE: Self = Self {
407 read_mult: 0.0,
408 write_mult: 0.0,
409 };
410}
411
412impl Default for CacheRates {
413 /// Anthropic was the first (and write-bucket-bearing) provider; defaulting
414 /// here keeps a missing-rates path conservative rather than free.
415 fn default() -> Self {
416 Self::ANTHROPIC
417 }
418}
419
420/// USD cost of the three input buckets (uncached prefix, cache read, cache
421/// write), each priced off the same base `input_per_mtok` rate with the
422/// provider's [`CacheRates`] applied. Shared by [`ModelProfile::usd_per_success`]
423/// and the durable [`crate::scoreboard::Scoreboard`] so both price caching the
424/// same way.
425pub fn priced_input_usd(
426 uncached_input_tokens: u64,
427 cache_read_input_tokens: u64,
428 cache_creation_input_tokens: u64,
429 input_per_mtok: f64,
430 cache: CacheRates,
431) -> f64 {
432 (uncached_input_tokens as f64 * input_per_mtok
433 + cache_read_input_tokens as f64 * input_per_mtok * cache.read_mult
434 + cache_creation_input_tokens as f64 * input_per_mtok * cache.write_mult)
435 / 1_000_000.0
436}
437
438/// EMA smoothing factor. Higher = more weight on recent observations.
439const EMA_ALPHA: f64 = 0.2;
440
441/// One resolved inference outcome — the durable, attributable "receipt"
442/// the concierge quotes ("routed to X at T because Y; latency 1.2s;
443/// outcome success q=0.9"). Append-only; one JSON line per resolution.
444/// Deliberately flat (success/quality/error pulled out of the richer
445/// `InferenceOutcome`) so it's stable to read back and cheap to reason
446/// over. Carries no prompt/output text — only routing-adjacent metadata.
447#[derive(Debug, Clone, Serialize, Deserialize)]
448pub struct OutcomeLedgerEntry {
449 pub trace_id: String,
450 pub model_id: String,
451 pub task: InferenceTask,
452 pub routing_reason: String,
453 pub latency_ms: u64,
454 pub input_tokens: usize,
455 pub output_tokens: usize,
456 /// Prompt-cache read (hit) input tokens, billed at ~0.1× input. Defaults
457 /// to 0 so receipts written before caching telemetry existed read back
458 /// (and price) as fully-uncached.
459 #[serde(default, skip_serializing_if = "is_zero_usize")]
460 pub cache_read_input_tokens: usize,
461 /// Prompt-cache write input tokens, billed at ~1.25×/2× input.
462 #[serde(default, skip_serializing_if = "is_zero_usize")]
463 pub cache_creation_input_tokens: usize,
464 #[serde(default, skip_serializing_if = "Option::is_none")]
465 pub success: Option<bool>,
466 #[serde(default, skip_serializing_if = "Option::is_none")]
467 pub quality: Option<f64>,
468 #[serde(default, skip_serializing_if = "Option::is_none")]
469 pub error: Option<String>,
470 /// Project/workspace the call belonged to — the key B1 groups paired
471 /// comparisons on. `None` until the capture path threads it (B1); kept
472 /// in the schema now so today's receipts are forward-compatible.
473 #[serde(default, skip_serializing_if = "Option::is_none")]
474 pub project_id: Option<String>,
475 /// Intent/use-case lane (finer than `task`), if the router computed
476 /// one. `None` until threaded (B1).
477 #[serde(default, skip_serializing_if = "Option::is_none")]
478 pub intent: Option<String>,
479 /// Why an unattributed receipt (`success: None` with an `error`) was not
480 /// the model's fault. `None` on attributed receipts and on unattributed
481 /// ones written before reasons were recorded.
482 #[serde(default, skip_serializing_if = "Option::is_none")]
483 pub attribution: Option<UnattributedReason>,
484 /// The CAR version that recorded this receipt. A model's failures under an
485 /// older CAR are not evidence against it: a release can fix the loader
486 /// (gemma-4's nested `vocab_size` failed every load until it did), and
487 /// counting those failures would retire weights for CAR's own bug. `None`
488 /// on receipts written before versions were recorded.
489 #[serde(default, skip_serializing_if = "Option::is_none")]
490 pub car_version: Option<String>,
491 pub timestamp: u64,
492}
493
494impl OutcomeLedgerEntry {
495 /// Whether this receipt is evidence the model cannot run: a failure the
496 /// model was blamed for, recorded by this CAR. A failure an older build
497 /// recorded may have been that build's bug (gemma-4's nested `vocab_size`
498 /// failed every load until a release read it), so it never condemns
499 /// weights this build might run.
500 ///
501 /// The rule scopes the portfolio's `cannot_load`, the one verdict that
502 /// deletes. Routing health keeps older builds' failures:
503 /// - the circuit breaker is per session, in memory, so nothing an older
504 /// build recorded holds it open after a restart;
505 /// - the lifetime `fail_count` feeds a Thompson-sampling posterior with a
506 /// weak prior, so a model with failures and no successes rarely wins an
507 /// adaptive pick against a proven one, whichever build recorded them;
508 /// - a blamed generation failure is a graded zero, so failures can pull a
509 /// local model's quality below the quality-first bootstrap floor. That
510 /// gate then leaves it out of latency-sensitive, quality-critical tasks
511 /// until other workloads lift its quality, or an explicit request or lane
512 /// default runs it.
513 ///
514 /// Subtracting old-build failures at reconciliation would need a once-only
515 /// stamp on every failure receipt at every release, and would hand a
516 /// genuinely broken model a clean slate each upgrade — which is why
517 /// [`plan_reconcile`] keeps the counters the profile's. Capping them would
518 /// not help either: with this prior, three failures and no success still
519 /// almost always lose to a proven model.
520 pub fn counts_against_model(&self) -> bool {
521 self.success == Some(false)
522 && self.attribution.is_none()
523 && self.car_version.as_deref() == Some(env!("CARGO_PKG_VERSION"))
524 }
525}
526
527/// Why a failed attempt was kept out of the model's health. Recorded on the
528/// receipt so a reader — the concierge quoting "refused for memory 20 times",
529/// a lane deciding what counts as friction — branches on a type instead of
530/// re-reading error text.
531#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
532#[serde(rename_all = "snake_case")]
533pub enum UnattributedReason {
534 /// The request asked for something this model cannot do (a mode, a
535 /// response format, a catalog precondition).
536 Capability,
537 /// The account behind the model: a missing key, an expired sign-in, no
538 /// credits, no workspace, a gateway with no upstream.
539 Account,
540 /// This machine could not hold the model while keeping CAR's memory
541 /// policy and emergency reserve. The model never ran.
542 ResourceBlocked,
543 /// A filter in front of the model ruled on the request's content.
544 ContentRefused,
545 /// The caller stopped it: a dropped stream, a spend limit, a controlled
546 /// termination.
547 Cancelled,
548 /// A model id the registry does not know and that never ran — a test
549 /// pinning a made-up id against the live daemon.
550 Unregistered,
551}
552
553/// Serde skip helper: keep the cache-token fields out of receipts that never
554/// touched the cache, so the durable ledger stays byte-identical to its
555/// pre-caching shape for the common (uncached) case.
556fn is_zero_usize(n: &usize) -> bool {
557 *n == 0
558}
559
560/// Cap on the in-memory ledger buffer between flushes — backstop against
561/// unbounded growth if the flush timer never runs. Oldest entries drop.
562const MAX_LEDGER_BUFFER: usize = 5000;
563
564/// Max characters of an error string kept in a ledger receipt. Errors are
565/// useful signal ("429", "context overflow") but can echo provider text
566/// or paths — cap the length so the privacy-bounded ledger never grows a
567/// large/sensitive blob. The classification lives in the prefix.
568const MAX_LEDGER_ERROR_CHARS: usize = 256;
569
570/// Truncate an error to [`MAX_LEDGER_ERROR_CHARS`] on a char boundary.
571fn redact_error(error: &str) -> String {
572 if error.chars().count() <= MAX_LEDGER_ERROR_CHARS {
573 return error.to_string();
574 }
575 let truncated: String = error.chars().take(MAX_LEDGER_ERROR_CHARS).collect();
576 format!("{truncated}…")
577}
578
579/// Classify an inference error as a **no-answer failure** — the call never
580/// produced an answer to grade — rather than a *generation* failure (the model
581/// returned content that was judged bad). Two families qualify:
582/// - transport/infrastructure: read timeout, connection reset, 5xx, EOF,
583/// overload.
584/// - auth/client rejection: 401/403/400/404, bad API key, `invalid_client`.
585/// The request was refused before any tokens were generated.
586///
587/// Such a call carries NO information about the model's *answer* quality — like a
588/// *mechanical* success, it leaves quality unknown — so it must not move the
589/// quality EMA. Counting these toward quality silently buries a good-but-slow,
590/// transiently-unreachable, or merely *mis-authenticated* model: observed live
591/// as `parslee/*` collapsing to `ema_quality ≈ 0.003` (from read-timeouts AND
592/// ~47 `HTTP 401 Unauthorized` from an undeployed gateway), which then
593/// deprioritized it under every workload. (Rate-limit / 429 is handled
594/// separately as a harsher *availability* penalty, not a quality one.)
595///
596/// This is the principled cut: a failure is answer-quality evidence ONLY when
597/// the model actually produced an answer. Auth/HTTP/transport rejections didn't,
598/// so the default for them is quality-neutral. We match on unambiguous textual
599/// forms, NOT bare status numbers (a bare "400"/"500" substring matches
600/// "1500ms", "tokens: 400", etc.) — a status-only error with no recognizable
601/// phrase falls through to the generation-failure path, which is the safe
602/// direction (it can only over-penalize a model that *did* answer, never
603/// silently bury one that didn't).
604fn is_no_answer_failure(error: &str) -> bool {
605 let e = error.to_ascii_lowercase();
606 const NEEDLES: &[&str] = &[
607 // --- transport / infrastructure (no answer produced) ---
608 "timeout",
609 "timed out",
610 "deadline",
611 "connection",
612 "connect error",
613 "reset by peer",
614 "broken pipe",
615 "network",
616 "unreachable",
617 "eof",
618 "stream closed",
619 "socket",
620 "503",
621 "502",
622 "504",
623 // A 500 is a server-side fault that almost always produced no answer,
624 // but bare "500" is too broad a substring (it matches "1500ms",
625 // "tokens: 500", a 500-dim error, …), so we match only its unambiguous
626 // textual form. A status-only "HTTP 500" with no body stays classified
627 // as a generation failure — accept that edge rather than risk
628 // false-positive hits on coincidental "500" substrings.
629 "internal server error",
630 "service unavailable",
631 "temporarily unavailable",
632 "overloaded",
633 // --- auth / client rejection (request refused, no answer produced) ---
634 // Same bare-number caution as 5xx: match textual forms, not "401"/"400".
635 // Kept HIGH-PRECISION: only phrases that unambiguously mean "the request
636 // was refused before generation." Deliberately NOT "not found" (matches a
637 // generation/parse error like "field not found") or "invalid token"
638 // (collides with tokenizer errors) — a rare 404 mis-scored as a bad
639 // answer is the safe direction; silently burying an auth-blocked model is
640 // not.
641 "unauthorized",
642 "unauthenticated",
643 "forbidden",
644 "permission denied",
645 "authentication",
646 "invalid_client",
647 "invalid api key",
648 "bad request",
649 ];
650 NEEDLES.iter().any(|n| e.contains(n))
651}
652
653/// Rewrite the ledger keeping only the most recent `max_entries` receipts
654/// (retention bound). No-op when under the cap or the file is absent.
655/// Atomic (temp + rename) so a crash mid-prune can't corrupt the ledger.
656pub fn prune_ledger(path: &std::path::Path, max_entries: usize) -> std::io::Result<()> {
657 if max_entries == 0 || !path.exists() {
658 return Ok(());
659 }
660 let entries = read_ledger(path, 0);
661 if entries.len() <= max_entries {
662 return Ok(());
663 }
664 rewrite_ledger(path, &entries[entries.len() - max_entries..])
665}
666
667/// Replace the ledger with `entries`: written to a uniquely named temp file
668/// in the same directory, made owner-only, synced, and renamed over the
669/// ledger, then the directory is synced. A shared temp name let a prune and a
670/// reconcile truncate each other's half-written file, and `fs::write` left the
671/// rewritten ledger world-readable — undoing the 0600 an append sets on a file
672/// holding error text.
673fn rewrite_ledger(path: &std::path::Path, entries: &[OutcomeLedgerEntry]) -> std::io::Result<()> {
674 use std::io::Write;
675 let dir = path
676 .parent()
677 .filter(|d| !d.as_os_str().is_empty())
678 .unwrap_or_else(|| std::path::Path::new("."));
679 let mut tmp = tempfile::Builder::new()
680 .prefix(".outcome_ledger.")
681 .suffix(".tmp")
682 .tempfile_in(dir)?;
683 #[cfg(unix)]
684 {
685 use std::os::unix::fs::PermissionsExt;
686 tmp.as_file()
687 .set_permissions(std::fs::Permissions::from_mode(0o600))?;
688 }
689 for e in entries {
690 let line = serde_json::to_string(e).map_err(std::io::Error::other)?;
691 tmp.write_all(line.as_bytes())?;
692 tmp.write_all(b"\n")?;
693 }
694 tmp.as_file().sync_all()?;
695 car_secrets::harden_owner_only(tmp.path());
696 tmp.persist(path).map_err(|e| e.error)?;
697 #[cfg(unix)]
698 if let Ok(d) = std::fs::File::open(dir) {
699 let _ = d.sync_all();
700 }
701 Ok(())
702}
703
704/// What [`plan_reconcile`] changed. All zero when the ledger and profiles
705/// were already consistent.
706#[derive(Debug, Default, Clone, PartialEq, Eq)]
707pub struct LedgerReconciliation {
708 /// Failure receipts an older build wrote for errors current code books as
709 /// unattributed, now marked with the reason it would record.
710 pub reclassified: usize,
711 /// Failure receipts for model ids the registry does not know and that
712 /// never ran, now marked [`UnattributedReason::Unregistered`].
713 pub unregistered: usize,
714 /// Profiles whose counters and quality evidence were rebuilt.
715 pub profiles_rebuilt: usize,
716 /// Profiles of never-run unregistered ids, removed.
717 pub profiles_removed: usize,
718}
719
720/// The reason current code would record for a legacy receipt's error, when it
721/// would not attribute that error to the model at all.
722///
723/// Receipts written before those errors were typed carry only their rendered
724/// text, so this is the one place that classifies by text — on history that
725/// has no type to read. Live booking routes on the [`crate::InferenceError`]
726/// variant and records an [`UnattributedReason`]; a receipt that carries one is
727/// never re-read here. The prefixes are the renderings of the producers named
728/// beside them; the ones whose producer no longer exists are kept as they
729/// were written.
730fn legacy_unattributed(error: &str) -> Option<UnattributedReason> {
731 use UnattributedReason as R;
732 const PREFIXES: &[(&str, UnattributedReason)] = &[
733 // `From<LocalAdmissionError> for InferenceError`: each verdict's
734 // recovery text, which is the `LocalResourceBlocked` Display.
735 (
736 "Not enough memory is free to start this model",
737 R::ResourceBlocked,
738 ),
739 ("This model needs about ", R::ResourceBlocked),
740 (
741 "Local model loading is disabled by the 0 GB allocation",
742 R::ResourceBlocked,
743 ),
744 ("CAR could not measure live memory", R::ResourceBlocked),
745 (
746 "This local model is being removed or maintained",
747 R::ResourceBlocked,
748 ),
749 (
750 "CAR is still confirming that the previous local model process exited",
751 R::ResourceBlocked,
752 ),
753 // `InferenceError::ProviderKeyMissing` (typed 2026-09-15); its Display
754 // kept the `InferenceFailed` rendering.
755 (
756 "inference failed: no API keys available for endpoint ",
757 R::Account,
758 ),
759 // The Parslee org lookup's 401, before it was `CredentialUnavailable`.
760 (
761 "inference failed: Parslee org lookup failed: HTTP 401",
762 R::Account,
763 ),
764 // A FoundationModels guardrail or refusal, before it was typed
765 // `ContentRefused`.
766 (
767 "inference failed: FoundationModels declined the request (",
768 R::ContentRefused,
769 ),
770 ];
771 if let Some((_, reason)) = PREFIXES
772 .iter()
773 .find(|(prefix, _)| error.starts_with(prefix))
774 {
775 return Some(*reason);
776 }
777 // `UnsupportedMode` rendered through `InferenceFailed`: anchored on the
778 // variant's own Display, not the bare word "mode".
779 (error.starts_with("inference failed: mode ") && error.contains(" not implemented on backend "))
780 .then_some(R::Capability)
781}
782
783/// The quality a profile and its tasks started from before any outcome moved
784/// them: the benchmark prior when one seeded the profile, otherwise neutral.
785/// Read from today's priors files, which may have changed since the profile
786/// was seeded — the best available stand-in, not a record.
787#[derive(Debug, Clone, Default)]
788pub struct QualitySeed {
789 pub overall: Option<f64>,
790 pub tasks: HashMap<String, f64>,
791}
792
793/// A planned reconciliation. Nothing is applied yet: `ledger`, when present,
794/// is what the ledger file must become, and `profiles` the corrected profile
795/// set to commit once it has.
796#[derive(Debug, Default)]
797pub struct ReconcilePlan {
798 pub report: LedgerReconciliation,
799 pub ledger: Option<Vec<OutcomeLedgerEntry>>,
800 pub profiles: Option<HashMap<String, ModelProfile>>,
801}
802
803/// Make the ledger and the persisted profiles agree about which failures were
804/// the model's.
805///
806/// Older builds booked failures a model was not responsible for — admission
807/// refusals of this machine's memory, missing keys, expired sign-ins, content
808/// rulings, ids that were never registered. This marks those receipts with the
809/// [`UnattributedReason`] current code would record and corrects each affected
810/// profile:
811///
812/// - **Counters stay the profile's.** The ledger is not a complete record of
813/// successes — `record_complete` credits a mechanical success to the profile
814/// and the receipt waits for a sweep that a short-lived process exits before
815/// — while every failure is written at once. Rebuilding counts from the
816/// ledger would pair a full failure history with a partial success history
817/// and bury exactly the local models this repair is for. So the failures
818/// stamped **in this run** are subtracted from `fail_count`, `total_calls`
819/// and the task's `failures`, the exact inverse of what `record_failure`
820/// added, and nothing else is touched. A profile breaking
821/// `total_calls >= success_count + fail_count` is clamped, not rebuilt.
822/// - **Quality evidence is replayed from the ledger.** Graded receipts are
823/// written when the grade arrives, so for these the ledger is complete:
824/// when a profile carries more `quality_observations` than its graded
825/// receipts account for — the misattributed case — the count and the EMA are
826/// replayed from those receipts in order, starting from the benchmark seed
827/// where one seeded the profile (`seeds`). The replay is a function of the
828/// ledger alone, so repeating it is harmless.
829///
830/// Subtraction must happen once per receipt, so it rides on the stamp. The
831/// plan touches nothing: the caller writes the stamped ledger **first**, and
832/// only once that succeeds commits `profiles` into the tracker and saves them.
833/// A failed write leaves the tracker untouched; a crash after the write keeps
834/// those failures counted (the conservative direction). Neither can subtract
835/// twice; rows an earlier version already
836/// rewrote to `success: None` subtract nothing, because that version already
837/// took them out of the counters. A reconciled state root has nothing left to
838/// stamp, so steady-state starts change nothing.
839///
840/// An id is unregistered only if the registry does not know it, no receipt
841/// shows it succeeding, and no profile credits it a success — an id that ran
842/// keeps its history whole. Nothing is deleted from the ledger.
843pub fn plan_reconcile(
844 path: &std::path::Path,
845 tracker: &OutcomeTracker,
846 is_registered: &dyn Fn(&str) -> bool,
847 seeds: &HashMap<String, QualitySeed>,
848) -> ReconcilePlan {
849 let mut plan = ReconcilePlan::default();
850 // Corrections are made on a copy and committed by the caller only once
851 // the stamped ledger is on disk — see [`OutcomeTracker::commit_reconciled`].
852 let mut profiles = tracker.profiles.clone();
853 let mut profiles_changed = false;
854 if !path.exists() {
855 return plan;
856 }
857 let mut entries = read_ledger(path, 0);
858
859 let ran: std::collections::HashSet<&str> = entries
860 .iter()
861 .filter(|e| e.success == Some(true))
862 .map(|e| e.model_id.as_str())
863 .collect();
864 let never_ran: std::collections::HashSet<String> = entries
865 .iter()
866 .map(|e| e.model_id.as_str())
867 .filter(|id| {
868 !ran.contains(id)
869 && !is_registered(id)
870 && profiles.get(*id).is_none_or(|p| p.success_count == 0)
871 })
872 .map(str::to_string)
873 .collect();
874
875 let mut touched = std::collections::BTreeSet::new();
876 // Failures stamped in this run, per model and task: what to take back out
877 // of the profile counters.
878 let mut unbooked: HashMap<String, HashMap<String, u64>> = HashMap::new();
879 for entry in &mut entries {
880 if entry.attribution.is_some() {
881 continue;
882 }
883 let legacy = entry.error.as_deref().and_then(legacy_unattributed);
884 let reason = if never_ran.contains(&entry.model_id) && entry.success == Some(false) {
885 UnattributedReason::Unregistered
886 } else if let Some(reason) = legacy.filter(|_| entry.success != Some(true)) {
887 // `success: None` here is either a row the first version of this
888 // pass already reclassified or an unattributed row written before
889 // reasons existed; stamping it changes nothing it is counted as.
890 reason
891 } else {
892 continue;
893 };
894 match reason {
895 UnattributedReason::Unregistered => plan.report.unregistered += 1,
896 _ if entry.success == Some(false) => plan.report.reclassified += 1,
897 _ => {}
898 }
899 if entry.success == Some(false) {
900 *unbooked
901 .entry(entry.model_id.clone())
902 .or_default()
903 .entry(entry.task.to_string())
904 .or_default() += 1;
905 }
906 entry.success = None;
907 entry.quality = None;
908 entry.attribution = Some(reason);
909 touched.insert(entry.model_id.clone());
910 }
911 let ledger_changed = !touched.is_empty();
912
913 // Clamp profiles whose counters contradict themselves.
914 for profile in profiles.values_mut() {
915 let floor = profile.success_count + profile.fail_count;
916 if profile.total_calls < floor {
917 profile.total_calls = floor;
918 plan.report.profiles_rebuilt += 1;
919 profiles_changed = true;
920 }
921 }
922
923 for model_id in &touched {
924 if never_ran.contains(model_id) {
925 if profiles.remove(model_id).is_some() {
926 plan.report.profiles_removed += 1;
927 profiles_changed = true;
928 }
929 continue;
930 }
931 let Some(profile) = profiles.get_mut(model_id) else {
932 continue;
933 };
934 let rows: Vec<&OutcomeLedgerEntry> =
935 entries.iter().filter(|e| &e.model_id == model_id).collect();
936 let seed = seeds.get(model_id);
937 let unbook = unbooked.get(model_id);
938 if correct_profile(profile, &rows, unbook, seed) {
939 plan.report.profiles_rebuilt += 1;
940 profiles_changed = true;
941 }
942 }
943
944 if ledger_changed {
945 plan.ledger = Some(entries);
946 }
947 if profiles_changed {
948 plan.profiles = Some(profiles);
949 }
950 plan
951}
952
953/// Write the ledger a [`ReconcilePlan`] produced. Done before the corrected
954/// profiles are committed, so a failed write leaves them as they were.
955pub fn write_reconciled_ledger(
956 path: &std::path::Path,
957 entries: &[OutcomeLedgerEntry],
958) -> std::io::Result<()> {
959 rewrite_ledger(path, entries)
960}
961
962/// Whether a failure receipt moved the quality EMA. Mirrors
963/// `record_failure`: a graded failure is recorded with `quality: Some(0.0)`,
964/// and a rate limit moves the EMA (×0.1) while recording no answer quality.
965fn failure_was_rate_limited(entry: &OutcomeLedgerEntry) -> bool {
966 entry.success == Some(false)
967 && entry.quality.is_none()
968 && entry
969 .error
970 .as_deref()
971 .is_some_and(|e| e.contains("429") || e.contains("RESOURCE_EXHAUSTED"))
972}
973
974/// Correct one profile: take the failures stamped in this run back out of its
975/// counters (`unbook`, per task), then replay its quality evidence from the
976/// graded receipts. Returns whether anything changed.
977fn correct_profile(
978 profile: &mut ModelProfile,
979 rows: &[&OutcomeLedgerEntry],
980 unbook: Option<&HashMap<String, u64>>,
981 seed: Option<&QualitySeed>,
982) -> bool {
983 let before = serde_json::to_value(&*profile).ok();
984
985 // Exactly what `record_failure` added: a call, a failure, a task failure.
986 for (task, n) in unbook.into_iter().flatten() {
987 profile.total_calls = profile.total_calls.saturating_sub(*n);
988 profile.fail_count = profile.fail_count.saturating_sub(*n);
989 if let Some(ts) = profile.task_stats.get_mut(task) {
990 ts.failures = ts.failures.saturating_sub(*n);
991 }
992 }
993
994 // Attributed receipts only: an unattributed one never touched a profile.
995 let attributed: Vec<&OutcomeLedgerEntry> = rows
996 .iter()
997 .copied()
998 .filter(|e| e.attribution.is_none())
999 .collect();
1000 let in_task =
1001 |e: &&OutcomeLedgerEntry, task: Option<&str>| task.is_none_or(|t| e.task.to_string() == t);
1002
1003 // Quality evidence: the graded receipts, in order, from the seed.
1004 let graded: Vec<&OutcomeLedgerEntry> = attributed
1005 .iter()
1006 .copied()
1007 .filter(|e| e.quality.is_some() || failure_was_rate_limited(e))
1008 .collect();
1009 let replay = |start: f64, task: Option<&str>| {
1010 let mut ema = start;
1011 let mut n = 0u64;
1012 for e in graded.iter().filter(|e| in_task(e, task)) {
1013 ema = match e.quality {
1014 Some(q) => ema * (1.0 - EMA_ALPHA) + q * EMA_ALPHA,
1015 None => ema * 0.1,
1016 };
1017 n += 1;
1018 }
1019 (ema, n)
1020 };
1021 // A seeded profile replays from its seed; one whose seed is gone (the
1022 // priors file was removed) cannot be replayed honestly and is left.
1023 let profile_start = match (profile.prior_sample_size, seed.and_then(|s| s.overall)) {
1024 (0, _) => Some(NEUTRAL_QUALITY),
1025 (_, Some(seeded)) => Some(seeded.clamp(0.0, 1.0)),
1026 (_, None) => None,
1027 };
1028 if let Some(start) = profile_start {
1029 let (ema, n) = replay(start, None);
1030 if profile.quality_observations > n {
1031 profile.ema_quality = ema;
1032 profile.quality_observations = n;
1033 }
1034 }
1035 for (task, ts) in profile.task_stats.iter_mut() {
1036 let seeded = seed.and_then(|s| s.tasks.get(task)).copied();
1037 let start = match (ts.prior_sample_size, seeded) {
1038 (0, _) => Some(TaskStats::default().ema_quality),
1039 (_, Some(seeded)) => Some(seeded.clamp(0.0, 1.0)),
1040 (_, None) => None,
1041 };
1042 let Some(start) = start else { continue };
1043 let (ema, n) = replay(start, Some(task));
1044 if ts.quality_observations > n {
1045 // With no graded evidence left an unseeded task returns to
1046 // neutral, which also keeps `backfill_quality_observations` from
1047 // refilling its weight from `failures` on the next load.
1048 ts.ema_quality = if n == 0 && ts.prior_sample_size == 0 {
1049 NEUTRAL_QUALITY
1050 } else {
1051 ema
1052 };
1053 ts.quality_observations = n;
1054 }
1055 }
1056 let changed = serde_json::to_value(&*profile).ok() != before;
1057 if changed {
1058 profile.updated_at = now_unix();
1059 }
1060 changed
1061}
1062
1063/// Append resolved outcome receipts to the JSONL ledger (one line each).
1064/// Append-only: no full-file rewrite, so this never triggers the
1065/// write-storm the profiles file had. Creates the file/parent on first
1066/// write.
1067pub fn append_ledger_entries(
1068 path: &std::path::Path,
1069 entries: &[OutcomeLedgerEntry],
1070) -> std::io::Result<()> {
1071 if entries.is_empty() {
1072 return Ok(());
1073 }
1074 if let Some(parent) = path.parent() {
1075 std::fs::create_dir_all(parent)?;
1076 }
1077 use std::io::Write;
1078 let mut opts = std::fs::OpenOptions::new();
1079 opts.create(true).append(true);
1080 // Local, privacy-bounded file (error text, soon project paths): owner
1081 // read/write only.
1082 #[cfg(unix)]
1083 {
1084 use std::os::unix::fs::OpenOptionsExt;
1085 opts.mode(0o600);
1086 }
1087 // On Windows `mode(0o600)` is ignored; lock the file owner-only via ACL on
1088 // a fresh create only (idempotent). No-op off Windows.
1089 let existed = path.exists();
1090 let mut f = opts.open(path)?;
1091 if !existed {
1092 car_secrets::harden_owner_only(path);
1093 }
1094 for e in entries {
1095 let line = serde_json::to_string(e).map_err(std::io::Error::other)?;
1096 f.write_all(line.as_bytes())?;
1097 f.write_all(b"\n")?;
1098 }
1099 Ok(())
1100}
1101
1102/// Read the most recent `limit` ledger entries (0 = all). Tolerant of
1103/// partial/garbage lines (skips them) so a torn append never poisons a
1104/// read — the consumer (UsageProfile) gets whatever is well-formed.
1105pub fn read_ledger(path: &std::path::Path, limit: usize) -> Vec<OutcomeLedgerEntry> {
1106 let Ok(content) = std::fs::read_to_string(path) else {
1107 return Vec::new();
1108 };
1109 let mut out: Vec<OutcomeLedgerEntry> = content
1110 .lines()
1111 .filter(|l| !l.trim().is_empty())
1112 .filter_map(|l| serde_json::from_str(l).ok())
1113 .collect();
1114 if limit > 0 && out.len() > limit {
1115 out = out.split_off(out.len() - limit);
1116 }
1117 out
1118}
1119
1120/// Tracks inference outcomes and builds performance profiles.
1121pub struct OutcomeTracker {
1122 /// In-memory profiles, keyed by model_id.
1123 profiles: HashMap<String, ModelProfile>,
1124 /// Pending outcomes: completed inference calls awaiting outcome signal.
1125 /// Keyed by trace_id.
1126 pending: HashMap<String, InferenceOutcome>,
1127 /// Counter for generating trace IDs.
1128 trace_counter: u64,
1129 /// Models excluded for this session (429/rate-limited). Hard exclusion.
1130 excluded: HashSet<String>,
1131 /// Set whenever a persisted field (a `ModelProfile`) changes; cleared
1132 /// on save. Lets the engine debounce disk writes — persist only when
1133 /// there's something new, instead of serializing the whole file after
1134 /// every inference call. `excluded`/`pending` are session-only and do
1135 /// not flip this.
1136 dirty: bool,
1137 /// Resolved outcome receipts awaiting append to the JSONL ledger.
1138 /// Drained by the engine on flush. Bounded by [`MAX_LEDGER_BUFFER`].
1139 ledger_buffer: std::collections::VecDeque<OutcomeLedgerEntry>,
1140}
1141
1142impl OutcomeTracker {
1143 pub fn new() -> Self {
1144 Self {
1145 profiles: HashMap::new(),
1146 pending: HashMap::new(),
1147 trace_counter: 0,
1148 excluded: HashSet::new(),
1149 dirty: false,
1150 ledger_buffer: std::collections::VecDeque::new(),
1151 }
1152 }
1153
1154 /// Push a resolved receipt to the ledger buffer (capped; oldest drops).
1155 fn push_ledger(&mut self, entry: OutcomeLedgerEntry) {
1156 if self.ledger_buffer.len() >= MAX_LEDGER_BUFFER {
1157 self.ledger_buffer.pop_front();
1158 }
1159 self.ledger_buffer.push_back(entry);
1160 }
1161
1162 /// Drain buffered receipts for the engine to append to the ledger file.
1163 pub fn drain_ledger(&mut self) -> Vec<OutcomeLedgerEntry> {
1164 self.ledger_buffer.drain(..).collect()
1165 }
1166
1167 /// Check if a model is excluded (rate-limited) for this session.
1168 pub fn is_excluded(&self, model_id: &str) -> bool {
1169 self.excluded.contains(model_id)
1170 }
1171
1172 /// Record that an inference call started. Returns a trace_id.
1173 pub fn record_start(
1174 &mut self,
1175 model_id: &str,
1176 task: InferenceTask,
1177 routing_reason: &str,
1178 ) -> String {
1179 self.trace_counter += 1;
1180 let trace_id = format!("t-{}-{}", now_unix(), self.trace_counter);
1181
1182 let outcome = InferenceOutcome {
1183 trace_id: trace_id.clone(),
1184 model_id: model_id.to_string(),
1185 task,
1186 // Length-capped: `routing_reason` is the only free-text field
1187 // that reaches the persisted ledger. INVARIANT: it must NEVER
1188 // embed prompt or output text — only a routing rationale
1189 // ("Code task -> Qwen3-4B"). The cap bounds accidental growth.
1190 routing_reason: redact_error(routing_reason),
1191 latency_ms: 0,
1192 input_tokens: 0,
1193 output_tokens: 0,
1194 cache_read_input_tokens: 0,
1195 cache_creation_input_tokens: 0,
1196 inferred_outcome: None,
1197 code_outcome: None,
1198 error: None,
1199 timestamp: now_unix(),
1200 success_credited: false,
1201 };
1202
1203 self.pending.insert(trace_id.clone(), outcome);
1204 trace_id
1205 }
1206
1207 /// Record completion of an inference call (timing + token counts).
1208 pub fn record_complete(
1209 &mut self,
1210 trace_id: &str,
1211 latency_ms: u64,
1212 input_tokens: usize,
1213 output_tokens: usize,
1214 ) {
1215 self.record_complete_cached(trace_id, latency_ms, input_tokens, output_tokens, 0, 0);
1216 }
1217
1218 /// Like [`Self::record_complete`] but also records the prompt-cache token
1219 /// split (`cache_read` = hit, `cache_creation` = write). `input_tokens`
1220 /// here is the *uncached* prefix only — for Anthropic that's
1221 /// `usage.input_tokens`, with the cached portion passed separately so cost
1222 /// accounting prices each bucket at its own rate. The plain 4-arg
1223 /// [`Self::record_complete`] forwards here with both cache buckets zeroed,
1224 /// which is correct for non-caching providers and local inference.
1225 pub fn record_complete_cached(
1226 &mut self,
1227 trace_id: &str,
1228 latency_ms: u64,
1229 input_tokens: usize,
1230 output_tokens: usize,
1231 cache_read_input_tokens: usize,
1232 cache_creation_input_tokens: usize,
1233 ) {
1234 if let Some(outcome) = self.pending.get_mut(trace_id) {
1235 outcome.latency_ms = latency_ms;
1236 outcome.input_tokens = input_tokens;
1237 outcome.output_tokens = output_tokens;
1238 outcome.cache_read_input_tokens = cache_read_input_tokens;
1239 outcome.cache_creation_input_tokens = cache_creation_input_tokens;
1240
1241 // Credit a *mechanical* success immediately when the call ran
1242 // and produced output with no error. This is symmetric with
1243 // `record_failure`, which books a failure the moment it happens
1244 // — previously a success was credited only by the 300s pending
1245 // sweep, so short-lived processes (the `car infer` CLI, eval
1246 // harnesses) exited before the sweep and NEVER recorded a
1247 // success, pinning working models at the 0.5 EMA prior / "0
1248 // successes" in the health UI (#312, only partially fixed by
1249 // the deferred sweep). `ema_quality` is untouched — only a real
1250 // downstream accept/edit signal moves quality. `success_credited`
1251 // stops the later sweep and any quality-signal resolution from
1252 // double-counting this same call.
1253 let mechanical_success = output_tokens > 0 && outcome.error.is_none();
1254 if mechanical_success {
1255 outcome.success_credited = true;
1256 }
1257 let model_id = outcome.model_id.clone();
1258 let task_key = outcome.task.to_string();
1259
1260 // Update profile with timing data (the `outcome` borrow of
1261 // `self.pending` ends here; `self.profiles` is a disjoint field).
1262 let profile = self
1263 .profiles
1264 .entry(model_id.clone())
1265 .or_insert_with(|| ModelProfile::new(model_id));
1266
1267 profile.total_calls += 1;
1268 profile.total_latency_ms += latency_ms;
1269 profile.total_input_tokens += input_tokens as u64;
1270 profile.total_output_tokens += output_tokens as u64;
1271 profile.total_cache_read_input_tokens += cache_read_input_tokens as u64;
1272 profile.total_cache_creation_input_tokens += cache_creation_input_tokens as u64;
1273 if mechanical_success {
1274 profile.success_count += 1;
1275 }
1276
1277 let ts = profile.task_stats.entry(task_key).or_default();
1278 ts.calls += 1;
1279 if mechanical_success {
1280 ts.successes += 1;
1281 }
1282 ts.avg_latency_ms =
1283 ts.avg_latency_ms + (latency_ms as f64 - ts.avg_latency_ms) / ts.calls as f64;
1284
1285 profile.updated_at = now_unix();
1286 self.dirty = true;
1287 }
1288 }
1289
1290 /// Record a failure.
1291 pub fn record_failure(&mut self, trace_id: &str, error: &str) {
1292 let mut ledger_entry = None;
1293 if let Some(outcome) = self.pending.get_mut(trace_id) {
1294 outcome.error = Some(error.to_string());
1295
1296 let profile = self
1297 .profiles
1298 .entry(outcome.model_id.clone())
1299 .or_insert_with(|| ModelProfile::new(outcome.model_id.clone()));
1300
1301 // A failed call is still a call: count it in `total_calls` so the
1302 // denominator includes failures (record_complete counts successes
1303 // the same way). Without this, `total_calls` tallied only
1304 // completions and could read *smaller* than `fail_count` — the
1305 // live symptom on parslee/* models (6 calls, 14 fails).
1306 profile.total_calls += 1;
1307 profile.fail_count += 1;
1308
1309 // Rate-limit errors (429) get a harsher penalty — the model is
1310 // guaranteed to fail again, so drop quality aggressively.
1311 let is_rate_limited = error.contains("429") || error.contains("RESOURCE_EXHAUSTED");
1312 // No-answer failures (read timeout, connection reset, 5xx, EOF, AND
1313 // auth/client rejections like 401/403/400) never produced an answer,
1314 // so answer quality is UNKNOWN — they must stay quality-neutral. The
1315 // call + failure are already booked above (`total_calls`/`fail_count`)
1316 // so availability and the circuit breaker still react; we just leave
1317 // `ema_quality` / `quality_observations` untouched. See
1318 // `is_no_answer_failure`.
1319 let is_no_answer = !is_rate_limited && is_no_answer_failure(error);
1320 if is_rate_limited {
1321 // Hard-exclude for the rest of this session (#13)
1322 self.excluded.insert(outcome.model_id.clone());
1323 profile.ema_quality *= 0.1;
1324 // A 429 is a real (availability) quality signal — it counts.
1325 profile.quality_observations += 1;
1326 } else if !is_no_answer {
1327 profile.ema_quality = profile.ema_quality * (1.0 - EMA_ALPHA) + 0.0 * EMA_ALPHA;
1328 // A generation failure is a real graded quality signal (it moved
1329 // ema_quality toward 0), so it counts toward quality-evidence.
1330 profile.quality_observations += 1;
1331 }
1332
1333 let task_key = outcome.task.to_string();
1334 let ts = profile.task_stats.entry(task_key).or_default();
1335 // `failures` is an availability counter — every failure counts,
1336 // no-answer failures included (it feeds degradation / circuit-breaking).
1337 ts.failures += 1;
1338 if is_rate_limited {
1339 ts.ema_quality *= 0.1;
1340 ts.quality_observations += 1;
1341 } else if !is_no_answer {
1342 ts.ema_quality *= 1.0 - EMA_ALPHA;
1343 ts.quality_observations += 1;
1344 }
1345
1346 profile.updated_at = now_unix();
1347 self.dirty = true;
1348
1349 ledger_entry = Some(OutcomeLedgerEntry {
1350 trace_id: outcome.trace_id.clone(),
1351 model_id: outcome.model_id.clone(),
1352 task: outcome.task,
1353 routing_reason: outcome.routing_reason.clone(),
1354 latency_ms: outcome.latency_ms,
1355 input_tokens: outcome.input_tokens,
1356 output_tokens: outcome.output_tokens,
1357 cache_read_input_tokens: outcome.cache_read_input_tokens,
1358 cache_creation_input_tokens: outcome.cache_creation_input_tokens,
1359 success: Some(false),
1360 // #374 follow-up: the ledger's `quality` is ANSWER quality,
1361 // recorded only when an answer was actually produced. A
1362 // *generation* failure produced a bad answer, so it's a real
1363 // graded `0.0` — mirroring the in-memory EMA's move toward 0 — so
1364 // the scoreboard's `avg_quality` and the shadow-calibration band
1365 // (#369) that fold this ledger see the same answer-quality signal
1366 // the live EMA does. A *no-answer* failure (transport/timeout/5xx
1367 // OR an auth/client rejection like 401/403/400) or a *429* produced
1368 // NO answer, so answer quality is unknown -> `None`, keeping
1369 // availability noise out of the answer-quality fold — the same
1370 // separation `record_failure` applies to the in-memory EMA above.
1371 //
1372 // NOTE: a 429 maps to `None` here INTENTIONALLY, even though the
1373 // in-memory path above counts it as a quality observation
1374 // (`ema_quality *= 0.1`). The in-memory EMA is a *routing-
1375 // availability* knob (a 429 means "keep steering away this
1376 // session"); the ledger field is *answer* quality, and a 429
1377 // produced no answer to grade. Don't "align" this to `Some(0.0)`
1378 // — that would inject availability noise into avg_quality / the
1379 // shadow band (the very thing #374 set out to remove).
1380 quality: if is_rate_limited || is_no_answer {
1381 None
1382 } else {
1383 Some(0.0)
1384 },
1385 error: Some(redact_error(error)),
1386 project_id: None,
1387 intent: None,
1388 attribution: None,
1389 car_version: Some(env!("CARGO_PKG_VERSION").to_string()),
1390 timestamp: now_unix(),
1391 });
1392 }
1393
1394 if let Some(entry) = ledger_entry {
1395 self.push_ledger(entry);
1396 }
1397
1398 // Failed outcomes don't need further tracking
1399 self.pending.remove(trace_id);
1400 }
1401
1402 /// Resolve a request/provider capability mismatch without treating it as a
1403 /// model-health or answer-quality failure.
1404 ///
1405 /// The attempt is still emitted to the receipt ledger for operator
1406 /// visibility, but it must not change the profile that adaptive routing
1407 /// uses for unrelated requests to the same model.
1408 pub fn record_capability_rejection(&mut self, trace_id: &str, error: &str) {
1409 self.record_not_model_fault(trace_id, error, UnattributedReason::Capability);
1410 }
1411
1412 /// Resolve a provider **account** rejection — bad/absent key, or out of
1413 /// credits — without treating it as a model failure.
1414 ///
1415 /// Same unattributed receipt as [`record_capability_rejection`](crate::outcome::OutcomeTracker::record_capability_rejection), different
1416 /// reason: the condition is account-wide, so it is not evidence about the
1417 /// model that happened to be selected. Booking it as a failure degrades the
1418 /// 30-day health EMA of every model the fallback chain touches, and that
1419 /// penalty outlives the top-up that fixes the account (Parslee-ai/car#650).
1420 pub fn record_account_rejection(&mut self, trace_id: &str, error: &str) {
1421 self.record_not_model_fault(trace_id, error, UnattributedReason::Account);
1422 }
1423
1424 /// Resolve a local admission refusal — the machine could not hold the
1425 /// model while keeping CAR's memory policy and emergency reserve — without
1426 /// treating it as a model failure.
1427 ///
1428 /// The model never ran. Booking the refusal as a failure measured the
1429 /// machine's free memory at that moment and charged it to the model: a
1430 /// 27B checkpoint refused five times on a busy afternoon read as 58%
1431 /// healthy for good, which is the number a concierge would retire it on.
1432 pub fn record_resource_rejection(&mut self, trace_id: &str, error: &str) {
1433 self.record_not_model_fault(trace_id, error, UnattributedReason::ResourceBlocked);
1434 }
1435
1436 /// Replace the profiles with a [`ReconcilePlan`]'s corrected set. Called
1437 /// only after the plan's ledger is on disk — committing first would let the
1438 /// periodic save persist a correction whose stamps never landed, and the
1439 /// next start would make it again.
1440 pub fn commit_reconciled(&mut self, profiles: HashMap<String, ModelProfile>) {
1441 self.profiles = profiles;
1442 self.dirty = true;
1443 }
1444
1445 /// Note how long an attempt ran before it ended without a completion — a
1446 /// dropped stream, for one — so its unattributed receipt keeps the time
1447 /// instead of reading as instantaneous. A stall the caller gave up on
1448 /// stays visible in the latency even though it is not booked as a failure.
1449 pub fn note_elapsed(&mut self, trace_id: &str, latency_ms: u64) {
1450 if let Some(outcome) = self.pending.get_mut(trace_id) {
1451 outcome.latency_ms = latency_ms;
1452 }
1453 }
1454
1455 /// Emit a receipt that records what happened without attributing it to the
1456 /// model: `success: None`, `quality: None`, so the ledger keeps the
1457 /// evidence while the routing profile stays untouched.
1458 pub fn record_not_model_fault(
1459 &mut self,
1460 trace_id: &str,
1461 error: &str,
1462 reason: UnattributedReason,
1463 ) {
1464 if let Some(outcome) = self.pending.remove(trace_id) {
1465 self.push_ledger(OutcomeLedgerEntry {
1466 trace_id: outcome.trace_id,
1467 model_id: outcome.model_id,
1468 task: outcome.task,
1469 routing_reason: outcome.routing_reason,
1470 latency_ms: outcome.latency_ms,
1471 input_tokens: outcome.input_tokens,
1472 output_tokens: outcome.output_tokens,
1473 cache_read_input_tokens: outcome.cache_read_input_tokens,
1474 cache_creation_input_tokens: outcome.cache_creation_input_tokens,
1475 success: None,
1476 quality: None,
1477 error: Some(redact_error(error)),
1478 project_id: None,
1479 intent: None,
1480 attribution: Some(reason),
1481 car_version: Some(env!("CARGO_PKG_VERSION").to_string()),
1482 timestamp: now_unix(),
1483 });
1484 }
1485 }
1486
1487 /// Record an inferred outcome from conversation signals.
1488 pub fn record_inferred_outcome(&mut self, trace_id: &str, outcome: InferredOutcome) {
1489 if let Some(pending) = self.pending.remove(trace_id) {
1490 self.apply_outcome(&pending, outcome.quality_score(), outcome.is_success());
1491 }
1492 }
1493
1494 /// Record an outcome from git-diff comparison (code generation).
1495 pub fn record_code_outcome(&mut self, trace_id: &str, outcome: CodeOutcome) {
1496 if let Some(pending) = self.pending.remove(trace_id) {
1497 self.apply_outcome(
1498 &pending,
1499 Some(outcome.quality_score()),
1500 Some(outcome.is_success()),
1501 );
1502 }
1503 }
1504
1505 /// Resolve all pending outcomes for a completed conversation turn.
1506 /// Called with the inferred outcomes from conversation signal analysis.
1507 pub fn resolve_pending_from_signals(&mut self, outcomes: Vec<(String, InferredOutcome)>) {
1508 for (trace_id, inferred) in outcomes {
1509 self.record_inferred_outcome(&trace_id, inferred);
1510 }
1511 }
1512
1513 /// Infer outcomes from a sequence of action results.
1514 ///
1515 /// In a reasoning session, each action's output feeds the next. If action N
1516 /// produced output and action N+1 succeeded using it, N was implicitly accepted.
1517 /// If N produced empty output or N+1 failed, N was implicitly rejected.
1518 ///
1519 /// Returns (trace_id, inferred_outcome) pairs ready for `resolve_pending_from_signals`.
1520 pub fn infer_outcomes_from_action_sequence(
1521 &self,
1522 action_results: &[(String, bool, f64, String)], // (trace_id, success, confidence, output)
1523 ) -> Vec<(String, InferredOutcome)> {
1524 let mut outcomes = Vec::new();
1525
1526 for (i, (trace_id, success, confidence, output)) in action_results.iter().enumerate() {
1527 if trace_id.is_empty() {
1528 continue; // No trace (e.g., memgine-only action)
1529 }
1530
1531 if !success {
1532 outcomes.push((
1533 trace_id.clone(),
1534 InferredOutcome::Rejected {
1535 confidence: *confidence,
1536 },
1537 ));
1538 continue;
1539 }
1540
1541 // Check if the next action used this one's output (implicit acceptance)
1542 let next_succeeded = action_results
1543 .get(i + 1)
1544 .map(|(_, s, _, _)| *s)
1545 .unwrap_or(true); // Last action: assume accepted if successful
1546
1547 let has_output = !output.trim().is_empty();
1548
1549 if has_output && next_succeeded {
1550 outcomes.push((
1551 trace_id.clone(),
1552 InferredOutcome::Accepted {
1553 confidence: *confidence,
1554 },
1555 ));
1556 } else if has_output && !next_succeeded {
1557 // Output existed but downstream failed — may not be this action's fault
1558 outcomes.push((
1559 trace_id.clone(),
1560 InferredOutcome::AcceptedWithEdits {
1561 confidence: confidence * 0.7,
1562 },
1563 ));
1564 } else {
1565 outcomes.push((trace_id.clone(), InferredOutcome::Inconclusive));
1566 }
1567 }
1568
1569 outcomes
1570 }
1571
1572 /// Get the profile for a model.
1573 pub fn profile(&self, model_id: &str) -> Option<&ModelProfile> {
1574 self.profiles.get(model_id)
1575 }
1576
1577 /// Whether a trace is still awaiting an outcome (in `pending`). Lets a
1578 /// caller observe, before calling `record_inferred_outcome`, whether the
1579 /// resolution will actually land or silently no-op (trace already resolved
1580 /// or swept) — so a lost outcome signal is countable, not invisible.
1581 pub fn has_pending(&self, trace_id: &str) -> bool {
1582 self.pending.contains_key(trace_id)
1583 }
1584
1585 /// Get all profiles.
1586 pub fn all_profiles(&self) -> &HashMap<String, ModelProfile> {
1587 &self.profiles
1588 }
1589
1590 /// Get pending trace IDs (for conversation signal analysis).
1591 pub fn pending_trace_ids(&self) -> Vec<String> {
1592 self.pending.keys().cloned().collect()
1593 }
1594
1595 /// Get a pending outcome by trace_id.
1596 pub fn get_pending(&self, trace_id: &str) -> Option<&InferenceOutcome> {
1597 self.pending.get(trace_id)
1598 }
1599
1600 /// Export profiles for serialization / persistence. Derived metrics
1601 /// (quality_per_1k_tokens) are recomputed on the way out so callers
1602 /// always see a consistent snapshot.
1603 pub fn export_profiles(&self) -> Vec<ModelProfile> {
1604 self.profiles
1605 .values()
1606 .cloned()
1607 .map(|mut p| {
1608 p.quality_per_1k_tokens = p.compute_quality_per_1k_tokens();
1609 p
1610 })
1611 .collect()
1612 }
1613
1614 /// Import profiles as a genuine mutation (benchmark priors, CLI import
1615 /// from the memgine fact graph, router merges). Marks the tracker
1616 /// dirty so the new profiles actually reach disk on the next flush.
1617 /// NOTE: hydration from disk does NOT go through here — see
1618 /// [`load_from_file`](crate::outcome::OutcomeTracker::load_from_file), which inserts directly and stays clean (loading
1619 /// is not a change).
1620 pub fn import_profiles(&mut self, profiles: Vec<ModelProfile>) {
1621 for p in profiles {
1622 self.profiles.insert(p.model_id.clone(), p);
1623 }
1624 self.dirty = true;
1625 }
1626
1627 /// Save profiles to a JSON file for cross-session persistence (#13).
1628 ///
1629 /// Atomic: serialize to a sibling temp file, then rename over the
1630 /// target. This is the durable receipt store — a torn `write()`
1631 /// (crash mid-write) would corrupt it, and `load_from_file` treats a
1632 /// parse failure as a hard error, which on next boot loses *all*
1633 /// history. The temp+rename makes a partial write impossible.
1634 pub fn save_to_file(&self, path: &std::path::Path) -> Result<(), std::io::Error> {
1635 let profiles = self.export_profiles();
1636 let json = serde_json::to_string_pretty(&profiles).map_err(std::io::Error::other)?;
1637 if let Some(parent) = path.parent() {
1638 std::fs::create_dir_all(parent)?;
1639 }
1640 let tmp = path.with_extension("json.tmp");
1641 std::fs::write(&tmp, json)?;
1642 std::fs::rename(&tmp, path)
1643 }
1644
1645 /// True if a persisted profile has changed since the last save.
1646 pub fn is_dirty(&self) -> bool {
1647 self.dirty
1648 }
1649
1650 /// Save only if dirty, clearing the flag on success. Returns whether
1651 /// a write happened. Lets callers persist cheaply on a timer without
1652 /// rewriting the whole file when nothing changed.
1653 pub fn save_if_dirty(&mut self, path: &std::path::Path) -> Result<bool, std::io::Error> {
1654 if !self.dirty {
1655 return Ok(false);
1656 }
1657 self.save_to_file(path)?;
1658 self.dirty = false;
1659 Ok(true)
1660 }
1661
1662 /// Load profiles from a JSON file for cross-session persistence (#13).
1663 pub fn load_from_file(&mut self, path: &std::path::Path) -> Result<usize, std::io::Error> {
1664 if !path.exists() {
1665 return Ok(0);
1666 }
1667 let json = std::fs::read_to_string(path)?;
1668 let profiles: Vec<ModelProfile> = serde_json::from_str(&json)
1669 .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
1670 let count = profiles.len();
1671 // Insert directly (NOT via import_profiles): hydration from disk is
1672 // not a change, so it must not flip the dirty flag — otherwise the
1673 // first idle flush would needlessly rewrite the file we just read.
1674 for mut p in profiles {
1675 // Migration: profiles persisted before `quality_observations`
1676 // existed deserialize it as 0, which would make the router forget
1677 // learned distrust (a model the EMA dropped on past failures would
1678 // revert to its schema estimate). Failures are always graded, so
1679 // `fail_count` is a safe lower bound — backfill from it so learned
1680 // distrust survives the upgrade. No-op for profiles written by this
1681 // version (a failure always bumps `quality_observations`, so the
1682 // count is already ≥ `fail_count`).
1683 backfill_quality_observations(&mut p);
1684 self.profiles.insert(p.model_id.clone(), p);
1685 }
1686 Ok(count)
1687 }
1688
1689 /// Apply a quality signal to the model's profile.
1690 fn apply_outcome(
1691 &mut self,
1692 pending: &InferenceOutcome,
1693 quality: Option<f64>,
1694 success: Option<bool>,
1695 ) {
1696 let profile = self
1697 .profiles
1698 .entry(pending.model_id.clone())
1699 .or_insert_with(|| ModelProfile::new(pending.model_id.clone()));
1700
1701 if let Some(q) = quality {
1702 profile.ema_quality = profile.ema_quality * (1.0 - EMA_ALPHA) + q * EMA_ALPHA;
1703 profile.quality_observations += 1;
1704
1705 let task_key = pending.task.to_string();
1706 let ts = profile.task_stats.entry(task_key).or_default();
1707 ts.ema_quality = ts.ema_quality * (1.0 - EMA_ALPHA) + q * EMA_ALPHA;
1708 ts.quality_observations += 1;
1709 }
1710
1711 if let Some(ok) = success {
1712 // The call may have already been credited a mechanical success at
1713 // completion (record_complete). Don't double-count it here.
1714 let already_credited = pending.success_credited;
1715 let task_key = pending.task.to_string();
1716 if ok {
1717 if !already_credited {
1718 profile.success_count += 1;
1719 let ts = profile.task_stats.entry(task_key).or_default();
1720 ts.successes += 1;
1721 }
1722 // else: already counted; the quality EMA above is the new info.
1723 } else {
1724 // A real downstream signal says this call was bad. If we
1725 // mechanically credited it as a success at completion,
1726 // reclassify: undo the success and book a failure instead.
1727 if already_credited {
1728 profile.success_count = profile.success_count.saturating_sub(1);
1729 let ts = profile.task_stats.entry(task_key.clone()).or_default();
1730 ts.successes = ts.successes.saturating_sub(1);
1731 }
1732 profile.fail_count += 1;
1733 let ts = profile.task_stats.entry(task_key).or_default();
1734 ts.failures += 1;
1735 }
1736 }
1737
1738 profile.updated_at = now_unix();
1739 self.dirty = true;
1740
1741 self.push_ledger(OutcomeLedgerEntry {
1742 trace_id: pending.trace_id.clone(),
1743 model_id: pending.model_id.clone(),
1744 task: pending.task,
1745 routing_reason: pending.routing_reason.clone(),
1746 latency_ms: pending.latency_ms,
1747 input_tokens: pending.input_tokens,
1748 output_tokens: pending.output_tokens,
1749 cache_read_input_tokens: pending.cache_read_input_tokens,
1750 cache_creation_input_tokens: pending.cache_creation_input_tokens,
1751 success,
1752 quality,
1753 error: None,
1754 project_id: None,
1755 intent: None,
1756 attribution: None,
1757 car_version: Some(env!("CARGO_PKG_VERSION").to_string()),
1758 timestamp: now_unix(),
1759 });
1760 }
1761
1762 /// Evict `pending` entries older than `ttl_secs` so the map can't grow
1763 /// without bound in a long-running daemon (only a few resolution paths
1764 /// ever drain it). For each evicted entry that actually *completed*
1765 /// (latency recorded), emit a terminal `Inconclusive` receipt — this
1766 /// also de-biases the ledger: without it, the ledger would only ever
1767 /// contain calls that happened to get a follow-up signal (a
1768 /// systematic skew toward reasoning-session traffic), and downstream
1769 /// stats would be computed over a non-representative sample. Returns
1770 /// the number of entries swept.
1771 pub fn sweep_pending(&mut self, ttl_secs: u64) -> usize {
1772 self.sweep_pending_at(ttl_secs, now_unix())
1773 }
1774
1775 /// Clock-injectable core of [`sweep_pending`] for deterministic tests.
1776 fn sweep_pending_at(&mut self, ttl_secs: u64, now: u64) -> usize {
1777 let cutoff = now.saturating_sub(ttl_secs);
1778 let expired: Vec<String> = self
1779 .pending
1780 .iter()
1781 .filter(|(_, o)| o.timestamp < cutoff)
1782 .map(|(id, _)| id.clone())
1783 .collect();
1784 for id in &expired {
1785 if let Some(o) = self.pending.remove(id) {
1786 // Only completed-but-unresolved calls become receipts; a
1787 // never-completed trace (latency 0) is in-flight/crashed —
1788 // a zero-everything receipt would be noise.
1789 if o.latency_ms > 0 {
1790 // A swept call completed but never received a downstream
1791 // quality signal. Recording *every* such call Inconclusive
1792 // (`success: None`) left signal-less models — notably ALL
1793 // local inference, which has no accept/edit feedback loop —
1794 // pinned at the 0.5 EMA prior with zero successes, so the
1795 // health UI rendered working models as "50%"/"0%" (#312).
1796 // Instead, credit a *mechanical* success when the call
1797 // actually returned output with no error: it ran and
1798 // produced tokens; we simply never learned whether the
1799 // answer was *good*. `quality` stays None so `ema_quality`
1800 // is only ever moved by a real quality signal. A completion
1801 // that produced no output stays Inconclusive (ambiguous).
1802 // Credit a mechanical success here ONLY if it wasn't
1803 // already booked at completion (record_complete now does
1804 // this immediately for the common case). A call credited
1805 // at completion still gets a `Some(true)` receipt below,
1806 // but its count is not incremented twice.
1807 let credit_now =
1808 o.output_tokens > 0 && o.error.is_none() && !o.success_credited;
1809 let was_success = o.success_credited || credit_now;
1810 if credit_now {
1811 let profile = self
1812 .profiles
1813 .entry(o.model_id.clone())
1814 .or_insert_with(|| ModelProfile::new(o.model_id.clone()));
1815 profile.success_count += 1;
1816 let ts = profile.task_stats.entry(o.task.to_string()).or_default();
1817 ts.successes += 1;
1818 profile.updated_at = now_unix();
1819 self.dirty = true;
1820 }
1821 self.push_ledger(OutcomeLedgerEntry {
1822 trace_id: o.trace_id,
1823 model_id: o.model_id,
1824 task: o.task,
1825 routing_reason: o.routing_reason,
1826 latency_ms: o.latency_ms,
1827 input_tokens: o.input_tokens,
1828 output_tokens: o.output_tokens,
1829 cache_read_input_tokens: o.cache_read_input_tokens,
1830 cache_creation_input_tokens: o.cache_creation_input_tokens,
1831 success: if was_success { Some(true) } else { None },
1832 quality: None,
1833 error: None,
1834 project_id: None,
1835 intent: None,
1836 attribution: None,
1837 car_version: Some(env!("CARGO_PKG_VERSION").to_string()),
1838 timestamp: now_unix(),
1839 });
1840 }
1841 }
1842 }
1843 expired.len()
1844 }
1845
1846 /// Check git diff for pending code suggestions and resolve outcomes.
1847 ///
1848 /// Two strategies:
1849 /// 1. **AST structural diff** (when `ast` feature is enabled): parse the old
1850 /// and new versions of changed files and compare at the symbol level.
1851 /// This gives precise outcomes: SignatureChanged, BodyModified, SymbolAdded.
1852 /// 2. **Text diff fallback**: token matching against the combined git diff.
1853 pub fn check_git_outcomes(&mut self, repo_dir: &std::path::Path) {
1854 let diff = match std::process::Command::new("git")
1855 .args(["diff", "--no-color"])
1856 .current_dir(repo_dir)
1857 .output()
1858 {
1859 Ok(output) => String::from_utf8_lossy(&output.stdout).to_string(),
1860 Err(_) => return,
1861 };
1862
1863 let staged_diff = match std::process::Command::new("git")
1864 .args(["diff", "--cached", "--no-color"])
1865 .current_dir(repo_dir)
1866 .output()
1867 {
1868 Ok(output) => String::from_utf8_lossy(&output.stdout).to_string(),
1869 Err(_) => String::new(),
1870 };
1871
1872 let combined_diff = format!("{}\n{}", diff, staged_diff);
1873
1874 if combined_diff.trim().is_empty() {
1875 return; // No changes at all
1876 }
1877
1878 // Try AST structural diff on changed files
1879 #[cfg(feature = "ast")]
1880 let ast_outcome = Self::check_git_outcomes_ast(repo_dir);
1881
1882 let code_traces: Vec<(String, String)> = self
1883 .pending
1884 .iter()
1885 .filter(|(_, o)| matches!(o.task, InferenceTask::Code))
1886 .map(|(id, o)| (id.clone(), o.model_id.clone()))
1887 .collect();
1888
1889 for (trace_id, _model_id) in code_traces {
1890 if let Some(pending) = self.pending.get(&trace_id) {
1891 // Try AST-based outcome first
1892 #[cfg(feature = "ast")]
1893 if let Some(ref ast_out) = ast_outcome {
1894 let pending_clone = pending.clone();
1895 self.apply_outcome(
1896 &pending_clone,
1897 Some(ast_out.quality_score()),
1898 Some(ast_out.is_success()),
1899 );
1900 continue;
1901 }
1902
1903 // Fallback: text token matching
1904 let output_tokens: Vec<&str> = pending
1905 .routing_reason
1906 .split_whitespace()
1907 .filter(|t| t.len() > 5)
1908 .collect();
1909
1910 let outcome = if output_tokens.iter().any(|t| combined_diff.contains(t)) {
1911 CodeOutcome::Applied
1912 } else {
1913 CodeOutcome::Modified
1914 };
1915
1916 let pending_clone = pending.clone();
1917 self.apply_outcome(
1918 &pending_clone,
1919 Some(outcome.quality_score()),
1920 Some(outcome.is_success()),
1921 );
1922 }
1923 }
1924 }
1925
1926 /// AST-based git outcome: parse changed files before and after, diff symbols.
1927 #[cfg(feature = "ast")]
1928 fn check_git_outcomes_ast(repo_dir: &std::path::Path) -> Option<CodeOutcome> {
1929 // Get list of changed files
1930 let name_only = std::process::Command::new("git")
1931 .args(["diff", "--name-only"])
1932 .current_dir(repo_dir)
1933 .output()
1934 .ok()?;
1935 let changed_files: Vec<&str> = std::str::from_utf8(&name_only.stdout)
1936 .ok()?
1937 .lines()
1938 .filter(|f| !f.is_empty())
1939 .collect();
1940
1941 if changed_files.is_empty() {
1942 return None;
1943 }
1944
1945 let mut has_sig_change = false;
1946 let mut has_body_change = false;
1947 let mut has_addition = false;
1948
1949 for file in &changed_files {
1950 // Only parse files tree-sitter supports
1951 if car_ast::Language::from_filename(file).is_none() {
1952 continue;
1953 }
1954
1955 // Get the HEAD version
1956 let old_content = std::process::Command::new("git")
1957 .args(["show", &format!("HEAD:{}", file)])
1958 .current_dir(repo_dir)
1959 .output()
1960 .ok()
1961 .and_then(|o| {
1962 if o.status.success() {
1963 String::from_utf8(o.stdout).ok()
1964 } else {
1965 None
1966 }
1967 });
1968
1969 // Get the working tree version
1970 let new_path = repo_dir.join(file);
1971 let new_content = std::fs::read_to_string(&new_path).ok();
1972
1973 match (old_content, new_content) {
1974 (Some(old), Some(new)) => {
1975 let old_parsed = car_ast::parse_file(&old, file);
1976 let new_parsed = car_ast::parse_file(&new, file);
1977
1978 if let (Some(old_p), Some(new_p)) = (old_parsed, new_parsed) {
1979 let changes = car_ast::diff_symbols(&old_p, &new_p);
1980 for change in &changes {
1981 match change {
1982 car_ast::SymbolChange::Added(_) => has_addition = true,
1983 car_ast::SymbolChange::Modified {
1984 signature_changed, ..
1985 } => {
1986 if *signature_changed {
1987 has_sig_change = true;
1988 } else {
1989 has_body_change = true;
1990 }
1991 }
1992 car_ast::SymbolChange::Removed(_) => has_sig_change = true,
1993 }
1994 }
1995 }
1996 }
1997 (None, Some(_)) => has_addition = true, // New file
1998 _ => {}
1999 }
2000 }
2001
2002 // Return the most significant outcome
2003 if has_sig_change {
2004 Some(CodeOutcome::SignatureChanged)
2005 } else if has_body_change {
2006 Some(CodeOutcome::BodyModified)
2007 } else if has_addition {
2008 Some(CodeOutcome::SymbolAdded)
2009 } else {
2010 None // No structural changes detected (maybe non-code files changed)
2011 }
2012 }
2013}
2014
2015impl Default for OutcomeTracker {
2016 fn default() -> Self {
2017 Self::new()
2018 }
2019}
2020
2021fn now_unix() -> u64 {
2022 SystemTime::now()
2023 .duration_since(UNIX_EPOCH)
2024 .unwrap_or_default()
2025 .as_secs()
2026}
2027
2028#[cfg(test)]
2029mod tests {
2030 use super::*;
2031
2032 #[test]
2033 fn only_a_failure_this_car_recorded_counts_against_a_model() {
2034 let entry = |success, version: Option<&str>| OutcomeLedgerEntry {
2035 trace_id: "t".into(),
2036 model_id: "m".into(),
2037 task: InferenceTask::Generate,
2038 routing_reason: String::new(),
2039 latency_ms: 0,
2040 input_tokens: 0,
2041 output_tokens: 0,
2042 cache_read_input_tokens: 0,
2043 cache_creation_input_tokens: 0,
2044 success: Some(success),
2045 quality: None,
2046 error: Some("config.json has no vocab_size".into()),
2047 project_id: None,
2048 intent: None,
2049 attribution: None,
2050 car_version: version.map(str::to_string),
2051 timestamp: 1,
2052 };
2053 let here = env!("CARGO_PKG_VERSION");
2054 assert!(entry(false, Some(here)).counts_against_model());
2055 // An older build's failure may be that build's bug.
2056 assert!(!entry(false, Some("0.0.1")).counts_against_model());
2057 assert!(!entry(false, None).counts_against_model());
2058 assert!(!entry(true, Some(here)).counts_against_model());
2059 }
2060
2061 #[test]
2062 fn lifecycle() {
2063 let mut tracker = OutcomeTracker::new();
2064
2065 // Start an inference call
2066 let trace = tracker.record_start(
2067 "qwen/qwen3-4b:q4_k_m",
2068 InferenceTask::Code,
2069 "Code task -> Qwen3-4B",
2070 );
2071
2072 // Complete it
2073 tracker.record_complete(&trace, 1200, 100, 50);
2074
2075 // Profile should have 1 call
2076 let profile = tracker.profile("qwen/qwen3-4b:q4_k_m").unwrap();
2077 assert_eq!(profile.total_calls, 1);
2078 assert_eq!(profile.avg_latency_ms(), 1200.0);
2079
2080 // Record positive outcome
2081 tracker.record_inferred_outcome(&trace, InferredOutcome::Accepted { confidence: 0.9 });
2082
2083 let profile = tracker.profile("qwen/qwen3-4b:q4_k_m").unwrap();
2084 assert_eq!(profile.success_count, 1);
2085 assert!(profile.ema_quality > 0.5); // should have gone up from 0.5
2086 }
2087
2088 #[test]
2089 fn failure_degrades() {
2090 // A failed call goes through record_failure ONLY — it never also
2091 // hits record_complete (a call either completes or errors, not both).
2092 // record_complete now credits a mechanical success, so mixing the two
2093 // on one trace would model a flow that does not occur.
2094 let mut tracker = OutcomeTracker::new();
2095 for _ in 0..5 {
2096 let trace = tracker.record_start("bad-model", InferenceTask::Generate, "test");
2097 // A *generation* failure (not a transport/timeout error, which is
2098 // quality-neutral — see `transport_failures_are_quality_neutral`).
2099 tracker.record_failure(&trace, "model produced malformed output");
2100 }
2101
2102 let profile = tracker.profile("bad-model").unwrap();
2103 assert_eq!(profile.fail_count, 5);
2104 assert_eq!(profile.success_count, 0);
2105 assert!(profile.should_degrade(2)); // 5 > 0 + 2
2106 assert!(profile.ema_quality < 0.3); // decayed toward 0
2107 }
2108
2109 #[test]
2110 fn code_outcome_ground_truth() {
2111 let mut tracker = OutcomeTracker::new();
2112
2113 let trace = tracker.record_start("qwen/qwen3-4b:q4_k_m", InferenceTask::Code, "code");
2114 tracker.record_complete(&trace, 500, 200, 100);
2115 tracker.record_code_outcome(&trace, CodeOutcome::Applied);
2116
2117 let profile = tracker.profile("qwen/qwen3-4b:q4_k_m").unwrap();
2118 assert_eq!(profile.success_count, 1);
2119 // EMA should reflect Applied quality (1.0): 0.5 * 0.8 + 1.0 * 0.2 = 0.6
2120 assert!((profile.ema_quality - 0.6).abs() < 0.01);
2121 }
2122
2123 #[test]
2124 fn per_task_stats() {
2125 let mut tracker = OutcomeTracker::new();
2126
2127 // Two code calls, one generate call
2128 for _ in 0..2 {
2129 let trace = tracker.record_start("m1", InferenceTask::Code, "code");
2130 tracker.record_complete(&trace, 1000, 100, 50);
2131 tracker.record_inferred_outcome(&trace, InferredOutcome::Accepted { confidence: 0.8 });
2132 }
2133 let trace = tracker.record_start("m1", InferenceTask::Generate, "gen");
2134 tracker.record_complete(&trace, 500, 50, 25);
2135 tracker.record_inferred_outcome(&trace, InferredOutcome::Rejected { confidence: 0.9 });
2136
2137 let profile = tracker.profile("m1").unwrap();
2138 assert_eq!(profile.total_calls, 3);
2139
2140 let code_stats = profile.task_stats(InferenceTask::Code).unwrap();
2141 assert_eq!(code_stats.calls, 2);
2142 assert_eq!(code_stats.successes, 2);
2143
2144 let gen_stats = profile.task_stats(InferenceTask::Generate).unwrap();
2145 assert_eq!(gen_stats.calls, 1);
2146 assert_eq!(gen_stats.failures, 1);
2147 }
2148
2149 #[test]
2150 fn export_populates_quality_per_1k_tokens() {
2151 let mut tracker = OutcomeTracker::new();
2152 let trace = tracker.record_start("m1", InferenceTask::Generate, "test");
2153 tracker.record_complete(&trace, 100, 800, 200); // 1000 tokens total
2154 tracker.record_inferred_outcome(&trace, InferredOutcome::Accepted { confidence: 1.0 });
2155
2156 let exported = tracker.export_profiles();
2157 assert_eq!(exported.len(), 1);
2158 let p = &exported[0];
2159 // ema_quality after one Accepted{1.0}: 0.5 * 0.8 + 1.0 * 0.2 = 0.6
2160 // quality_per_1k = 0.6 * 1000 / 1000 = 0.6
2161 assert!(
2162 (p.quality_per_1k_tokens - 0.6).abs() < 1e-6,
2163 "got {}",
2164 p.quality_per_1k_tokens
2165 );
2166 }
2167
2168 #[test]
2169 fn quality_per_1k_tokens_zero_without_tokens() {
2170 let profile = ModelProfile::new("x".into());
2171 assert_eq!(profile.compute_quality_per_1k_tokens(), 0.0);
2172 }
2173
2174 #[test]
2175 fn tokens_per_success_is_outcome_denominated() {
2176 let mut p = ModelProfile::new("x".into());
2177 // No success yet → None (not a fabricated number).
2178 assert_eq!(p.tokens_per_success(), None);
2179 // 900 total tokens over 3 successful outcomes → 300 tokens/success.
2180 p.total_input_tokens = 600;
2181 p.total_output_tokens = 300;
2182 p.success_count = 3;
2183 assert_eq!(p.tokens_per_success(), Some(300.0));
2184 // Dollar cost-per-success: (600*$1 + 300*$2)/1M / 3 successes.
2185 let usd = p.usd_per_success(1.0, 2.0, CacheRates::ANTHROPIC).unwrap();
2186 assert!((usd - ((600.0 * 1.0 + 300.0 * 2.0) / 1_000_000.0 / 3.0)).abs() < 1e-12);
2187 assert_eq!(
2188 ModelProfile::new("y".into()).usd_per_success(1.0, 2.0, CacheRates::ANTHROPIC),
2189 None
2190 );
2191 }
2192
2193 #[test]
2194 fn usd_per_success_prices_cache_buckets_separately() {
2195 let mut p = ModelProfile::new("cached".into());
2196 p.success_count = 1;
2197 p.total_input_tokens = 100; // uncached prefix @ 1.0×
2198 p.total_cache_read_input_tokens = 1000; // hit @ 0.1×
2199 p.total_cache_creation_input_tokens = 200; // write @ 1.25×
2200 p.total_output_tokens = 50; // output @ 2.0×
2201 // Each input bucket priced at its own multiplier off the $1/Mtok base.
2202 let expected_input = (100.0 * 1.0 + 1000.0 * 1.0 * 0.1 + 200.0 * 1.0 * 1.25) / 1_000_000.0;
2203 let expected = expected_input + 50.0 * 2.0 / 1_000_000.0;
2204 let usd = p.usd_per_success(1.0, 2.0, CacheRates::ANTHROPIC).unwrap();
2205 assert!((usd - expected).abs() < 1e-12, "got {usd}, want {expected}");
2206 // Token efficiency counts all real tokens, cache included.
2207 assert_eq!(p.total_tokens(), 100 + 1000 + 200 + 50);
2208 // A cache hit must be ~10× cheaper than counting it as full input,
2209 // which is the whole point of fixing the accounting.
2210 let mut naive = ModelProfile::new("naive".into());
2211 naive.success_count = 1;
2212 naive.total_input_tokens = 1300; // same tokens, all priced at 1.0×
2213 naive.total_output_tokens = 50;
2214 assert!(
2215 naive
2216 .usd_per_success(1.0, 2.0, CacheRates::ANTHROPIC)
2217 .unwrap()
2218 > usd
2219 );
2220 }
2221
2222 #[test]
2223 fn openai_cache_rates_price_reads_at_half_no_write_premium() {
2224 // OpenAI economics: cached reads at 0.5×, no write bucket. Same token
2225 // counts as Anthropic must cost MORE under OpenAI (0.5× vs 0.1× reads).
2226 let mut p = ModelProfile::new("gpt".into());
2227 p.success_count = 1;
2228 p.total_input_tokens = 100; // uncached @ 1.0×
2229 p.total_cache_read_input_tokens = 1000; // OpenAI hit @ 0.5×
2230 p.total_cache_creation_input_tokens = 0; // OpenAI has no write bucket
2231 let usd = p.usd_per_success(1.0, 2.0, CacheRates::OPENAI).unwrap();
2232 let expected = (100.0 * 1.0 + 1000.0 * 1.0 * 0.5) / 1_000_000.0;
2233 assert!((usd - expected).abs() < 1e-12, "got {usd}, want {expected}");
2234 // Same buckets under Anthropic rates (0.1× read) are cheaper.
2235 let anthropic = p.usd_per_success(1.0, 2.0, CacheRates::ANTHROPIC).unwrap();
2236 assert!(
2237 usd > anthropic,
2238 "OpenAI 0.5× read must exceed Anthropic 0.1×"
2239 );
2240 // And both beat pricing the cache hit as full uncached input (1.0×).
2241 let mut naive = ModelProfile::new("naive".into());
2242 naive.success_count = 1;
2243 naive.total_input_tokens = 1100;
2244 assert!(naive.usd_per_success(1.0, 2.0, CacheRates::OPENAI).unwrap() > usd);
2245 }
2246
2247 #[test]
2248 fn record_complete_cached_accumulates_cache_totals() {
2249 let mut tracker = OutcomeTracker::new();
2250 let trace = tracker.record_start("m", InferenceTask::Generate, "gen");
2251 tracker.record_complete_cached(&trace, 100, 40, 20, 800, 120);
2252 let p = tracker.profile("m").unwrap();
2253 assert_eq!(p.total_input_tokens, 40, "uncached prefix only");
2254 assert_eq!(p.total_cache_read_input_tokens, 800);
2255 assert_eq!(p.total_cache_creation_input_tokens, 120);
2256 }
2257
2258 #[test]
2259 fn dirty_flag_and_save_if_dirty() {
2260 let dir = std::env::temp_dir().join("car-outcome-dirty-test");
2261 let _ = std::fs::remove_dir_all(&dir);
2262 let path = dir.join("outcome_profiles.json");
2263
2264 let mut tracker = OutcomeTracker::new();
2265 // Fresh tracker is clean → save_if_dirty is a no-op.
2266 assert!(!tracker.is_dirty());
2267 assert!(!tracker.save_if_dirty(&path).unwrap());
2268 assert!(!path.exists());
2269
2270 // A recorded outcome dirties it.
2271 let trace = tracker.record_start("m1", InferenceTask::Generate, "router");
2272 tracker.record_complete(&trace, 100, 10, 20);
2273 assert!(tracker.is_dirty());
2274
2275 // save_if_dirty writes once and clears the flag.
2276 assert!(tracker.save_if_dirty(&path).unwrap());
2277 assert!(path.exists());
2278 assert!(!tracker.is_dirty());
2279
2280 // Second call with no new changes does not rewrite.
2281 assert!(!tracker.save_if_dirty(&path).unwrap());
2282
2283 // Loading profiles must NOT mark the tracker dirty.
2284 let mut fresh = OutcomeTracker::new();
2285 fresh.load_from_file(&path).unwrap();
2286 assert!(!fresh.is_dirty());
2287
2288 // But a genuine import (benchmark priors / CLI) MUST dirty, so the
2289 // imported profiles actually persist on the next flush.
2290 fresh.import_profiles(vec![ModelProfile::new("seeded".into())]);
2291 assert!(fresh.is_dirty());
2292
2293 let _ = std::fs::remove_dir_all(&dir);
2294 }
2295
2296 #[test]
2297 fn ledger_captures_resolved_outcomes() {
2298 let dir = std::env::temp_dir().join("car-outcome-ledger-test");
2299 let _ = std::fs::remove_dir_all(&dir);
2300 let path = dir.join("outcome_ledger.jsonl");
2301
2302 let mut tracker = OutcomeTracker::new();
2303
2304 // A success with a quality signal → one resolved receipt.
2305 let t1 = tracker.record_start("good-model", InferenceTask::Generate, "router:test");
2306 tracker.record_complete(&t1, 1200, 50, 100);
2307 tracker.record_inferred_outcome(&t1, InferredOutcome::Accepted { confidence: 0.9 });
2308
2309 // A failure → one resolved receipt with the error.
2310 let t2 = tracker.record_start("bad-model", InferenceTask::Code, "router:test");
2311 tracker.record_failure(&t2, "boom: 500");
2312
2313 let drained = tracker.drain_ledger();
2314 assert_eq!(drained.len(), 2);
2315 assert!(tracker.drain_ledger().is_empty(), "drain clears the buffer");
2316
2317 append_ledger_entries(&path, &drained).unwrap();
2318 let read = read_ledger(&path, 0);
2319 assert_eq!(read.len(), 2);
2320
2321 let good = read.iter().find(|e| e.model_id == "good-model").unwrap();
2322 assert_eq!(good.success, Some(true));
2323 assert!(good.quality.is_some());
2324 assert_eq!(good.routing_reason, "router:test");
2325 assert_eq!(good.latency_ms, 1200);
2326
2327 let bad = read.iter().find(|e| e.model_id == "bad-model").unwrap();
2328 assert_eq!(bad.success, Some(false));
2329 assert_eq!(bad.error.as_deref(), Some("boom: 500"));
2330
2331 // Read back limited to most recent 1.
2332 assert_eq!(read_ledger(&path, 1).len(), 1);
2333
2334 let _ = std::fs::remove_dir_all(&dir);
2335 }
2336
2337 #[test]
2338 fn ledger_redacts_long_errors_and_prunes() {
2339 let dir = std::env::temp_dir().join("car-outcome-privacy-test");
2340 let _ = std::fs::remove_dir_all(&dir);
2341 let path = dir.join("outcome_ledger.jsonl");
2342
2343 // Long error is truncated on capture.
2344 let mut tracker = OutcomeTracker::new();
2345 let t = tracker.record_start("m", InferenceTask::Generate, "r");
2346 let huge = "x".repeat(5000);
2347 tracker.record_failure(&t, &huge);
2348 let drained = tracker.drain_ledger();
2349 let err = drained[0].error.as_ref().unwrap();
2350 assert!(
2351 err.chars().count() <= MAX_LEDGER_ERROR_CHARS + 1,
2352 "error truncated"
2353 );
2354
2355 // Pruning keeps only the most recent N.
2356 let entries: Vec<OutcomeLedgerEntry> = (0..10)
2357 .map(|i| OutcomeLedgerEntry {
2358 trace_id: format!("t{i}"),
2359 model_id: "m".into(),
2360 task: InferenceTask::Generate,
2361 routing_reason: "r".into(),
2362 latency_ms: 1,
2363 input_tokens: 1,
2364 output_tokens: 1,
2365 cache_read_input_tokens: 0,
2366 cache_creation_input_tokens: 0,
2367 success: Some(true),
2368 quality: Some(1.0),
2369 error: None,
2370 project_id: None,
2371 intent: None,
2372 attribution: None,
2373 car_version: None,
2374 timestamp: i,
2375 })
2376 .collect();
2377 append_ledger_entries(&path, &entries).unwrap();
2378 prune_ledger(&path, 3).unwrap();
2379 let kept = read_ledger(&path, 0);
2380 assert_eq!(kept.len(), 3);
2381 assert_eq!(kept[0].trace_id, "t7"); // most recent 3: t7,t8,t9
2382 assert_eq!(kept[2].trace_id, "t9");
2383
2384 let _ = std::fs::remove_dir_all(&dir);
2385 }
2386
2387 #[test]
2388 fn sweep_pending_credits_mechanical_success_or_inconclusive() {
2389 let mut tracker = OutcomeTracker::new();
2390
2391 // Completed with output, no signal -> mechanical success (#312).
2392 let t1 = tracker.record_start("m", InferenceTask::Generate, "r");
2393 tracker.record_complete(&t1, 500, 10, 20);
2394 // Completed but produced NO output -> stays Inconclusive.
2395 let t2 = tracker.record_start("m", InferenceTask::Generate, "r");
2396 tracker.record_complete(&t2, 300, 5, 0);
2397 // A never-completed call (in-flight / crashed) -> no receipt.
2398 let _t3 = tracker.record_start("m", InferenceTask::Generate, "r");
2399
2400 // Evaluate the sweep against a clock 10s in the future so the
2401 // same-second pending entries are unambiguously past the cutoff.
2402 let swept = tracker.sweep_pending_at(0, now_unix() + 10);
2403 assert_eq!(swept, 3, "all pending entries evicted");
2404
2405 // Only the two completed calls become receipts.
2406 let mut receipts = tracker.drain_ledger();
2407 receipts.sort_by_key(|r| r.latency_ms);
2408 assert_eq!(receipts.len(), 2);
2409 // 300ms, no output -> Inconclusive
2410 assert_eq!(receipts[0].latency_ms, 300);
2411 assert_eq!(receipts[0].success, None);
2412 // 500ms, produced output -> mechanical success, quality untouched
2413 assert_eq!(receipts[1].latency_ms, 500);
2414 assert_eq!(receipts[1].success, Some(true));
2415 assert_eq!(receipts[1].quality, None);
2416
2417 // success_count is credited; ema_quality stays the neutral prior
2418 // (no real quality signal ever arrived).
2419 let p = tracker.profile("m").expect("profile exists");
2420 assert_eq!(p.success_count, 1);
2421 assert_eq!(p.ema_quality, 0.5);
2422 }
2423
2424 #[test]
2425 fn record_complete_credits_success_immediately() {
2426 // The headline fix: a call that completes with output is credited a
2427 // success at completion, not deferred to the 300s sweep — so a
2428 // short-lived process (CLI) that exits before any sweep still records
2429 // the success. Previously success_count stayed 0 and health read 0.5.
2430 let mut tracker = OutcomeTracker::new();
2431 let t = tracker.record_start("m", InferenceTask::Generate, "r");
2432 tracker.record_complete(&t, 500, 12, 20);
2433
2434 let p = tracker.profile("m").expect("profile exists");
2435 assert_eq!(p.success_count, 1, "success credited at completion");
2436 assert_eq!(p.total_calls, 1);
2437 assert_eq!(p.total_input_tokens, 12, "input tokens recorded, not 0");
2438 assert_eq!(p.fail_count, 0);
2439
2440 // A later sweep of the same (still-pending) call must NOT double-count.
2441 tracker.sweep_pending_at(0, now_unix() + 10);
2442 let p = tracker.profile("m").unwrap();
2443 assert_eq!(p.success_count, 1, "sweep does not re-credit");
2444 }
2445
2446 #[test]
2447 fn success_rate_resolved_distinguishes_no_signal_from_measured() {
2448 // Display surfaces must not show a never-measured model as a confident
2449 // "50%": `success_rate_resolved()` returns None until something
2450 // resolves, then the real ratio. (The router never derives a rate
2451 // here — it reads raw success/fail counts directly.)
2452 let mut p = ModelProfile::new("m".to_string());
2453 assert_eq!(
2454 p.success_rate_resolved(),
2455 None,
2456 "no resolved signal, not a fabricated 0.5"
2457 );
2458
2459 p.success_count = 3;
2460 p.fail_count = 1;
2461 assert_eq!(
2462 p.success_rate_resolved(),
2463 Some(0.75),
2464 "real rate once resolved"
2465 );
2466 }
2467
2468 #[test]
2469 fn quality_observations_count_only_graded_signals() {
2470 // #3: a mechanical success must NOT count as a graded quality
2471 // observation (it leaves answer quality unknown), but an accept/reject
2472 // signal and a failure must. The router keys reliability on this count.
2473 let mut tracker = OutcomeTracker::new();
2474
2475 // Mechanical success — credited, but quality stays unknown.
2476 let t = tracker.record_start("m", InferenceTask::Generate, "r");
2477 tracker.record_complete(&t, 500, 12, 20);
2478 let p = tracker.profile("m").unwrap();
2479 assert_eq!(p.success_count, 1);
2480 assert_eq!(
2481 p.quality_observations, 0,
2482 "mechanical success is not a graded quality observation"
2483 );
2484 assert!(
2485 (p.ema_quality - 0.5).abs() < 1e-9,
2486 "EMA untouched by mechanical success"
2487 );
2488
2489 // A real accept signal — graded.
2490 tracker.record_inferred_outcome(&t, InferredOutcome::Accepted { confidence: 0.9 });
2491 let p = tracker.profile("m").unwrap();
2492 assert_eq!(p.quality_observations, 1, "graded accept signal counts");
2493 assert!(p.ema_quality > 0.5, "graded accept moved the EMA up");
2494 assert_eq!(
2495 p.task_stats(InferenceTask::Generate)
2496 .unwrap()
2497 .quality_observations,
2498 1,
2499 "per-task graded count tracked too"
2500 );
2501
2502 // A failure — also graded (moves EMA toward 0).
2503 let t2 = tracker.record_start("m", InferenceTask::Generate, "r");
2504 tracker.record_failure(&t2, "boom");
2505 let p = tracker.profile("m").unwrap();
2506 assert_eq!(p.quality_observations, 2, "failure is a graded observation");
2507 }
2508
2509 #[test]
2510 fn record_complete_no_output_is_not_a_success() {
2511 let mut tracker = OutcomeTracker::new();
2512 let t = tracker.record_start("m", InferenceTask::Generate, "r");
2513 tracker.record_complete(&t, 300, 5, 0); // no output
2514 let p = tracker.profile("m").unwrap();
2515 assert_eq!(p.success_count, 0, "no output -> no mechanical success");
2516 assert_eq!(p.total_calls, 1);
2517 }
2518
2519 #[test]
2520 fn record_failure_counts_total_calls() {
2521 // A failed call is still a call. total_calls must include it so it can
2522 // never read smaller than fail_count (the parslee/* 6-calls/14-fails
2523 // symptom).
2524 let mut tracker = OutcomeTracker::new();
2525 for _ in 0..3 {
2526 let t = tracker.record_start("m", InferenceTask::Generate, "r");
2527 tracker.record_failure(&t, "boom 500");
2528 }
2529 let p = tracker.profile("m").unwrap();
2530 assert_eq!(p.fail_count, 3);
2531 assert_eq!(p.total_calls, 3, "failures counted in total_calls");
2532 assert!(p.fail_count <= p.total_calls);
2533 }
2534
2535 #[test]
2536 fn capability_rejection_does_not_change_model_profile() {
2537 let mut tracker = OutcomeTracker::new();
2538 let trace = tracker.record_start("m", InferenceTask::Generate, "r");
2539
2540 tracker.record_capability_rejection(&trace, "unsupported mode: json schema");
2541
2542 assert!(tracker.profile("m").is_none());
2543 assert!(!tracker.has_pending(&trace));
2544 assert_eq!(tracker.ledger_buffer.len(), 1);
2545 }
2546
2547 #[test]
2548 fn transport_failures_are_quality_neutral() {
2549 // A read-timeout / connection error means the model never produced an
2550 // answer — its answer quality is UNKNOWN, so transport failures must not
2551 // move the quality EMA or count as graded quality evidence. (Observed
2552 // live: parslee/* collapsed to ema_quality ≈ 0.003 after ~23 daemon
2553 // read-timeouts, which buried it under every workload.) They DO still
2554 // count for availability (total_calls / fail_count / circuit breaker).
2555 let mut tracker = OutcomeTracker::new();
2556 // One genuine generation failure establishes the baseline EMA + a
2557 // graded quality observation.
2558 let t = tracker.record_start("m", InferenceTask::Code, "r");
2559 tracker.record_failure(&t, "model produced malformed output");
2560 let baseline_ema = tracker.profile("m").unwrap().ema_quality;
2561 let baseline_qobs = tracker.profile("m").unwrap().quality_observations;
2562
2563 // Hammer with transport timeouts AND auth/client rejections — neither
2564 // produced an answer, so quality must not budge. The 401 case is the
2565 // live parslee/* symptom: ~47 "HTTP 401 Unauthorized" from an undeployed
2566 // gateway must not bury the model on *answer* quality.
2567 for _ in 0..5 {
2568 let t = tracker.record_start("m", InferenceTask::Code, "r");
2569 tracker.record_failure(&t, "daemon read timeout on infer after 30s");
2570 }
2571 for _ in 0..5 {
2572 let t = tracker.record_start("m", InferenceTask::Code, "r");
2573 tracker.record_failure(
2574 &t,
2575 "inference failed: Parslee org lookup failed: HTTP 401 Unauthorized: Authentication",
2576 );
2577 }
2578
2579 let p = tracker.profile("m").unwrap();
2580 assert_eq!(
2581 p.fail_count, 11,
2582 "no-answer failures still counted for availability"
2583 );
2584 assert_eq!(p.total_calls, 11, "every call counted (failures included)");
2585 assert!(
2586 (p.ema_quality - baseline_ema).abs() < 1e-9,
2587 "timeouts AND auth rejections must not move the quality EMA ({baseline_ema} -> {})",
2588 p.ema_quality
2589 );
2590 assert_eq!(
2591 p.quality_observations, baseline_qobs,
2592 "no-answer failures (transport + auth) are not graded quality evidence"
2593 );
2594 // Contrast: a real generation failure still penalizes quality.
2595 let t = tracker.record_start("m", InferenceTask::Code, "r");
2596 tracker.record_failure(&t, "model produced malformed output");
2597 assert!(
2598 tracker.profile("m").unwrap().ema_quality < baseline_ema,
2599 "a non-transport failure must still move the quality EMA down"
2600 );
2601 }
2602
2603 #[test]
2604 fn is_no_answer_failure_precision_boundary() {
2605 // The classifier gates whether a failure counts as answer-quality, so
2606 // pin both sides of the precision boundary it promises. Matches are on
2607 // the FAILURE error string only (never model output).
2608 // No-answer: transport/infra + auth/client rejection → quality-neutral.
2609 for s in [
2610 "daemon read timeout on infer after 30s",
2611 "connection reset by peer",
2612 "upstream returned 503 service unavailable",
2613 "HTTP 500 Internal Server Error",
2614 "stream closed unexpectedly (eof)",
2615 "HTTP 401 Unauthorized",
2616 "unauthenticated request",
2617 "HTTP 403 Forbidden",
2618 "permission denied by gateway",
2619 "Parslee org lookup failed: Authentication required",
2620 "oauth: invalid_client",
2621 "invalid api key",
2622 "HTTP 400 Bad Request: max context length exceeded",
2623 ] {
2624 assert!(
2625 is_no_answer_failure(s),
2626 "expected no-answer (quality-neutral): {s:?}"
2627 );
2628 }
2629 // Generation failures: the model produced output that was bad/unparseable.
2630 // These MUST stay generation failures (graded), NOT be swallowed as
2631 // no-answer — including the deliberately-excluded ambiguous phrases.
2632 for s in [
2633 "model produced malformed output",
2634 "field 'user' not found in model response", // not "not found" → no-answer
2635 "tokenizer: invalid token id 99999", // not "invalid token" → no-answer
2636 "response failed JSON schema validation",
2637 "tool call arguments did not validate",
2638 ] {
2639 assert!(
2640 !is_no_answer_failure(s),
2641 "expected generation failure (graded): {s:?}"
2642 );
2643 }
2644 }
2645
2646 #[test]
2647 fn ledger_quality_distinguishes_generation_from_transport_failure() {
2648 // The durable ledger's `quality` is ANSWER quality, mirroring the
2649 // in-memory EMA (#374 follow-up). A generation failure produced a bad
2650 // answer → graded `Some(0.0)`; a transport failure or a 429 produced no
2651 // answer → `None`. This keeps the scoreboard's avg_quality and the
2652 // shadow-calibration band (#369) — both of which fold this ledger — free
2653 // of availability noise while still seeing the genuine bad-answer signal.
2654 let mut tracker = OutcomeTracker::new();
2655
2656 let t = tracker.record_start("m", InferenceTask::Code, "r");
2657 tracker.record_failure(&t, "model produced malformed output"); // generation
2658 let t = tracker.record_start("m", InferenceTask::Code, "r");
2659 tracker.record_failure(&t, "daemon read timeout on infer after 30s"); // transport
2660 let t = tracker.record_start("m", InferenceTask::Code, "r");
2661 tracker.record_failure(&t, "429 RESOURCE_EXHAUSTED"); // rate-limit
2662 let t = tracker.record_start("m", InferenceTask::Code, "r");
2663 tracker.record_failure(&t, "upstream: 503 service unavailable"); // 5xx infra
2664 let t = tracker.record_start("m", InferenceTask::Code, "r");
2665 tracker.record_failure(&t, "HTTP 500 Internal Server Error"); // server fault
2666 let t = tracker.record_start("m", InferenceTask::Code, "r");
2667 // The live parslee/* failure: auth rejection, no answer produced.
2668 tracker.record_failure(
2669 &t,
2670 "inference failed: Parslee org lookup failed: HTTP 401 Unauthorized: Authentication",
2671 );
2672
2673 let entries = tracker.drain_ledger();
2674 assert_eq!(entries.len(), 6);
2675 // All are failures for availability accounting.
2676 assert!(entries.iter().all(|e| e.success == Some(false)));
2677 // Only the generation failure carries a graded answer-quality of 0.0.
2678 assert_eq!(
2679 entries[0].quality,
2680 Some(0.0),
2681 "generation failure -> graded 0.0 answer quality"
2682 );
2683 assert_eq!(
2684 entries[1].quality, None,
2685 "transport failure produced no answer -> quality unknown (None)"
2686 );
2687 assert_eq!(
2688 entries[2].quality, None,
2689 "429 produced no answer -> quality unknown (None), an availability event"
2690 );
2691 assert_eq!(
2692 entries[3].quality, None,
2693 "5xx infra failure produced no answer -> None"
2694 );
2695 assert_eq!(
2696 entries[4].quality, None,
2697 "an 'internal server error' (500) is a server fault, not a bad answer -> None"
2698 );
2699 assert_eq!(
2700 entries[5].quality, None,
2701 "a 401 auth rejection produced no answer -> None (not a bad-answer 0.0)"
2702 );
2703 }
2704
2705 #[test]
2706 fn real_failure_signal_reclassifies_mechanical_success() {
2707 // Completed with output -> mechanical success. A later Rejected signal
2708 // overrides it: undo the success, book a failure. No phantom success.
2709 let mut tracker = OutcomeTracker::new();
2710 let t = tracker.record_start("m", InferenceTask::Generate, "r");
2711 tracker.record_complete(&t, 100, 8, 15);
2712 assert_eq!(tracker.profile("m").unwrap().success_count, 1);
2713
2714 tracker.record_inferred_outcome(&t, InferredOutcome::Rejected { confidence: 0.9 });
2715 let p = tracker.profile("m").unwrap();
2716 assert_eq!(p.success_count, 0, "mechanical success undone");
2717 assert_eq!(p.fail_count, 1, "failure booked");
2718 }
2719
2720 #[test]
2721 fn real_success_signal_does_not_double_count() {
2722 let mut tracker = OutcomeTracker::new();
2723 let t = tracker.record_start("m", InferenceTask::Generate, "r");
2724 tracker.record_complete(&t, 100, 8, 15);
2725 tracker.record_inferred_outcome(&t, InferredOutcome::Accepted { confidence: 0.9 });
2726 let p = tracker.profile("m").unwrap();
2727 assert_eq!(
2728 p.success_count, 1,
2729 "Accepted on an already-credited call is not +2"
2730 );
2731 }
2732
2733 #[test]
2734 fn export_import() {
2735 let mut tracker = OutcomeTracker::new();
2736 let trace = tracker.record_start("m1", InferenceTask::Generate, "test");
2737 tracker.record_complete(&trace, 100, 10, 5);
2738 tracker.record_inferred_outcome(&trace, InferredOutcome::Accepted { confidence: 0.9 });
2739
2740 let exported = tracker.export_profiles();
2741 assert_eq!(exported.len(), 1);
2742
2743 let mut tracker2 = OutcomeTracker::new();
2744 tracker2.import_profiles(exported);
2745 assert!(tracker2.profile("m1").is_some());
2746 }
2747}