car_inference/

outcome.rs

//! Outcome tracking — learn from inference results to improve routing.
//!
//! Two observation channels:
//! 1. **Conversation signals** — implicit feedback from what happens after an inference
//!    call (user moved on = accepted, user corrected = rejected, re-asked = rejected).
//! 2. **Git-diff tracking** — for code generation, compare suggestions to actual commits
//!    (ground truth, no classification model needed).
//!
//! Every inference call produces an `InferenceOutcome`. Outcomes accumulate into
//! `ModelProfile`s with per-task statistics. The adaptive router uses profiles
//! to make data-driven model selection.

use std::collections::{HashMap, HashSet};
use std::time::{SystemTime, UNIX_EPOCH};

use serde::{Deserialize, Serialize};

/// Task type for outcome tracking. Maps to ModelCapability but at the call level.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum InferenceTask {
    Generate,
    Embed,
    Classify,
    Code,
    Reasoning,
}

impl std::fmt::Display for InferenceTask {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            InferenceTask::Generate => write!(f, "generate"),
            InferenceTask::Embed => write!(f, "embed"),
            InferenceTask::Classify => write!(f, "classify"),
            InferenceTask::Code => write!(f, "code"),
            InferenceTask::Reasoning => write!(f, "reasoning"),
        }
    }
}

/// A single inference invocation record.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InferenceOutcome {
    /// Unique trace ID for this invocation.
    pub trace_id: String,
    /// Model that was used.
    pub model_id: String,
    /// Task type.
    pub task: InferenceTask,
    /// How the model was selected.
    pub routing_reason: String,
    /// Wall-clock latency in milliseconds.
    pub latency_ms: u64,
    /// Input tokens (estimated). For a prompt-cached provider this is the
    /// *uncached* prefix only; the cached portion is split out below.
    pub input_tokens: usize,
    /// Output tokens (estimated).
    pub output_tokens: usize,
    /// Prompt-cache hit input tokens (Anthropic `cache_read_input_tokens`),
    /// billed at ~0.1× base input. `0` for uncached / non-caching providers.
    #[serde(default)]
    pub cache_read_input_tokens: usize,
    /// Prompt-cache write input tokens (Anthropic `cache_creation_input_tokens`),
    /// billed at ~1.25×/2× base input. `0` for uncached / non-caching providers.
    #[serde(default)]
    pub cache_creation_input_tokens: usize,
    /// Outcome from conversation signal inference.
    pub inferred_outcome: Option<InferredOutcome>,
    /// Outcome from git-diff tracking (code only).
    pub code_outcome: Option<CodeOutcome>,
    /// Error message if inference failed.
    pub error: Option<String>,
    /// Unix timestamp.
    pub timestamp: u64,
    /// Whether a *mechanical* success has already been credited to the
    /// model profile for this call (booked at completion when it produced
    /// output with no error — see [`OutcomeTracker::record_complete`]).
    /// Guards against the later pending-sweep or a downstream quality
    /// signal double-counting the same call. In-memory only (`pending` is
    /// never persisted), so it's skipped from (de)serialization.
    #[serde(skip)]
    pub success_credited: bool,
    /// Unix time the call ended (completed, or ran out without a completion —
    /// see [`OutcomeTracker::note_elapsed`]). `None` while it is in flight.
    /// The pending sweep counts its grace period from here, not from
    /// `timestamp` (the start), so a call that itself runs longer than the
    /// sweep TTL can still be graded after it finishes. In-memory only.
    #[serde(skip)]
    pub completed_at: Option<u64>,
}

/// Outcome inferred from conversation flow (implicit feedback).
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum InferredOutcome {
    /// User moved on, built on the response.
    Accepted { confidence: f64 },
    /// User used the result but modified it.
    AcceptedWithEdits { confidence: f64 },
    /// User corrected, re-asked, or explicitly rejected.
    Rejected { confidence: f64 },
    /// No follow-up signal (session ended, inconclusive).
    Inconclusive,
}

impl InferredOutcome {
    /// Convert to a quality score (0.0 = bad, 1.0 = good).
    pub fn quality_score(&self) -> Option<f64> {
        match self {
            InferredOutcome::Accepted { confidence } => Some(*confidence),
            InferredOutcome::AcceptedWithEdits { confidence } => Some(confidence * 0.7),
            InferredOutcome::Rejected { confidence } => Some((1.0 - confidence) * 0.3),
            InferredOutcome::Inconclusive => None,
        }
    }

    pub fn is_success(&self) -> Option<bool> {
        match self {
            InferredOutcome::Accepted { .. } => Some(true),
            InferredOutcome::AcceptedWithEdits { .. } => Some(true),
            InferredOutcome::Rejected { .. } => Some(false),
            InferredOutcome::Inconclusive => None,
        }
    }
}

/// Outcome from git-diff comparison (code generation ground truth).
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum CodeOutcome {
    /// Suggestion was applied as-is (exact or near-exact match in diff).
    Applied,
    /// User changed the same file but differently (partial adoption).
    Modified,
    /// File unchanged despite suggestion (rejected / not used).
    Ignored,
    /// AST structural diff: signature was changed (breaking change).
    SignatureChanged,
    /// AST structural diff: body was modified but signature preserved (non-breaking).
    BodyModified,
    /// AST structural diff: new symbol was added.
    SymbolAdded,
}

impl CodeOutcome {
    pub fn quality_score(&self) -> f64 {
        match self {
            CodeOutcome::Applied => 1.0,
            CodeOutcome::SignatureChanged => 0.8,
            CodeOutcome::BodyModified => 0.7,
            CodeOutcome::SymbolAdded => 0.7,
            CodeOutcome::Modified => 0.6,
            CodeOutcome::Ignored => 0.1,
        }
    }

    pub fn is_success(&self) -> bool {
        !matches!(self, CodeOutcome::Ignored)
    }
}

/// What `ema_quality` and `quality_observations` mean in a saved profile.
///
/// - 0: a generation failure was graded 0.0 and counted as quality evidence,
///   and a success was never graded, so the evidence could only ever say
///   "worse". One stray failure took a third of a model's catalog quality
///   away for good (`reliability`'s `e/(e+K)` with W=2, K=4).
/// - 1: only real grades (accept/edit/reject, external grades) move quality.
///   A failure is availability evidence: it feeds `fail_count`, the circuit
///   breaker and the router's completion rate, not answer quality.
pub const QUALITY_EVIDENCE_VERSION: u32 = 1;

/// Bring a profile saved under an older [`QUALITY_EVIDENCE_VERSION`] to the
/// current one. Version 0's quality evidence cannot be separated into real
/// grades and failures, and failures dominate it, so it is dropped: quality
/// returns to neutral with no evidence and the router sits at the catalog
/// estimate until real grades arrive. Availability counters and external
/// grades are kept.
fn migrate_quality_evidence(p: &mut ModelProfile) {
    if p.quality_evidence_version >= QUALITY_EVIDENCE_VERSION {
        return;
    }
    p.ema_quality = NEUTRAL_QUALITY;
    p.quality_observations = 0;
    p.prior_sample_size = 0;
    for ts in p.task_stats.values_mut() {
        ts.ema_quality = NEUTRAL_QUALITY;
        ts.quality_observations = 0;
        ts.prior_sample_size = 0;
    }
    p.quality_evidence_version = QUALITY_EVIDENCE_VERSION;
}

/// The neutral quality prior a fresh profile starts at.
const NEUTRAL_QUALITY: f64 = 0.5;

/// Per-task statistics within a model profile.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct TaskStats {
    pub calls: u64,
    pub successes: u64,
    pub failures: u64,
    /// Running average latency in ms.
    pub avg_latency_ms: f64,
    /// Exponential moving average of quality score.
    pub ema_quality: f64,
    /// Number of benchmark cases behind `ema_quality` when it is a *seeded
    /// cold-start prior* (i.e. `calls == 0`). Lets the router shrink the
    /// seeded score toward its schema estimate by sample size. `0` means no
    /// benchmark evidence — the router then sits at the schema estimate.
    #[serde(default)]
    pub prior_sample_size: usize,
    /// Number of *graded* quality observations that have moved `ema_quality`
    /// for this task — accept/edit/reject signals, NOT mechanical successes or
    /// failures (which say whether a call ran, not how good its answer was;
    /// see [`QUALITY_EVIDENCE_VERSION`]). The router trusts
    /// `ema_quality` in proportion to this count, so a task with many ungraded
    /// successes stays anchored to its schema estimate instead of a hollow
    /// neutral EMA.
    #[serde(default)]
    pub quality_observations: u64,
    /// Graded results reported from outside this daemon
    /// (`outcomes.record_external_grade`), one entry per source key (see
    /// [`external_grade_key`]). Kept apart from `ema_quality` on purpose:
    /// these grades are not tied to a call this daemon served, so they never
    /// touch the served counters, the ledger replay in [`plan_reconcile`], or
    /// the bootstrap floor. The router pools them into `reliability` through
    /// [`TaskStats::external_evidence`], whose caps bound what one source, one
    /// reporter, and all of them together can weigh.
    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
    pub external_grades: std::collections::BTreeMap<String, ExternalGradeStats>,
}

impl Default for TaskStats {
    /// A task's first grade moves quality from neutral, as the overall
    /// profile's does. Starting at 0.0 made a perfect first grade read as a
    /// poor one.
    fn default() -> Self {
        Self {
            calls: 0,
            successes: 0,
            failures: 0,
            avg_latency_ms: 0.0,
            ema_quality: NEUTRAL_QUALITY,
            prior_sample_size: 0,
            quality_observations: 0,
            external_grades: Default::default(),
        }
    }
}

impl TaskStats {
    pub fn success_rate(&self) -> f64 {
        let total = self.successes + self.failures;
        if total == 0 {
            return 0.5;
        } // prior: assume neutral
        self.successes as f64 / total as f64
    }

    /// Whether this task has any served or seeded evidence of its own, as
    /// opposed to holding only external grades. An entry that external grades
    /// created must not shadow the overall profile's served evidence in the
    /// router, and its `ema_quality` (the `Default` 0.0) is not a measurement.
    /// An entry with no external grades always counts as served, so the
    /// router's choice between task and overall stats is unchanged for it.
    pub fn has_served_evidence(&self) -> bool {
        self.calls > 0
            || self.successes > 0
            || self.failures > 0
            || self.prior_sample_size > 0
            || self.quality_observations > 0
            || self.external_grades.is_empty()
    }

    /// The pooled external evidence for this task now: `(evidence, mean)`.
    /// See [`Self::external_evidence_at`].
    pub fn external_evidence(&self) -> (f64, f64) {
        self.external_evidence_at(now_unix())
    }

    /// The pooled external evidence for this task at `now`: `(evidence,
    /// mean)`, `(0.0, 0.0)` when nothing counts. Three caps apply in turn:
    ///
    /// 1. each source counts its observations up to
    ///    [`EXTERNAL_GRADE_EVIDENCE_CAP`];
    /// 2. all of one reporter's sources together count at most
    ///    [`EXTERNAL_GRADE_REPORTER_EVIDENCE_CAP`], so a reporter cannot add
    ///    weight by rotating source names;
    /// 3. every reporter together counts at most
    ///    [`EXTERNAL_GRADE_TOTAL_EVIDENCE_CAP`].
    ///
    /// A cap scales a group's weight down without changing its mean. A source
    /// not updated for [`EXTERNAL_GRADE_MAX_AGE_SECS`] counts for nothing, so
    /// a stale grade stops steering routing on its own.
    pub fn external_evidence_at(&self, now: u64) -> (f64, f64) {
        // reporter -> (evidence, evidence-weighted quality sum)
        let mut by_reporter: std::collections::BTreeMap<Option<&str>, (f64, f64)> =
            std::collections::BTreeMap::new();
        for (key, grade) in &self.external_grades {
            if grade.is_stale_at(now) {
                continue;
            }
            let e = grade.evidence() as f64;
            let entry = by_reporter
                .entry(external_grade_reporter(key))
                .or_insert((0.0, 0.0));
            entry.0 += e;
            entry.1 += e * grade.quality;
        }
        let mut evidence = 0.0;
        let mut weighted = 0.0;
        for (e, w) in by_reporter.into_values() {
            if e <= 0.0 {
                continue;
            }
            let counted = e.min(EXTERNAL_GRADE_REPORTER_EVIDENCE_CAP as f64);
            evidence += counted;
            weighted += counted * (w / e);
        }
        if evidence <= 0.0 {
            return (0.0, 0.0);
        }
        let mean = weighted / evidence;
        (evidence.min(EXTERNAL_GRADE_TOTAL_EVIDENCE_CAP as f64), mean)
    }
}

/// Most evidence one external source contributes to a model's task, in
/// benchmark-case units. Reports beyond the cap keep updating that source's
/// quality but add no further weight.
pub const EXTERNAL_GRADE_EVIDENCE_CAP: u64 = 8;
/// Most evidence all of one reporter's sources together contribute to a
/// model's task. With the router's prior pseudocount of 4, one reporter moves
/// `reliability` at most two thirds of the way from the catalog prior to the
/// quality it reports, however many source names it uses.
pub const EXTERNAL_GRADE_REPORTER_EVIDENCE_CAP: u64 = 8;
/// Most evidence every external grade together contributes to a model's task:
/// with the prior pseudocount of 4, at most four fifths of the way from the
/// catalog prior to the pooled external quality. Served, graded calls are not
/// capped, so a model CAR serves is steered by its own record over time.
pub const EXTERNAL_GRADE_TOTAL_EVIDENCE_CAP: u64 = 16;
/// Most distinct sources one reporter may keep on one model's task. A source
/// that has gone stale (see [`EXTERNAL_GRADE_MAX_AGE_SECS`]) is evicted to
/// make room instead of refusing the new one.
pub const MAX_EXTERNAL_GRADE_SOURCES: usize = 8;
/// A source not updated for this long (90 days) no longer counts toward
/// routing, and its slot can be taken by a new source. Reporting under it
/// again starts it afresh rather than averaging with the stale quality.
pub const EXTERNAL_GRADE_MAX_AGE_SECS: u64 = 90 * 24 * 60 * 60;
/// Longest accepted reporter identity, in bytes.
pub const MAX_EXTERNAL_GRADE_REPORTER_LEN: usize = 160;
/// Largest `n` (graded cases) one report may carry.
pub const MAX_EXTERNAL_GRADE_CASES: u32 = 100;
/// Longest accepted source name, in bytes.
pub const MAX_EXTERNAL_GRADE_SOURCE_LEN: usize = 64;

/// One external source's graded evidence for a model's task.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ExternalGradeStats {
    /// The source's current quality for this model and task, in `[0, 1]`: a
    /// running mean over its reports whose window is capped at
    /// [`EXTERNAL_GRADE_EVIDENCE_CAP`] cases, so a newer verdict keeps moving
    /// it after the cap is reached.
    pub quality: f64,
    /// Graded cases this source has reported, uncapped.
    pub observations: u64,
    /// Unix time of the source's latest report.
    pub updated_at: u64,
}

impl ExternalGradeStats {
    /// The evidence this source counts for on its own: its observations,
    /// capped. The reporter and total caps in
    /// [`TaskStats::external_evidence_at`] may count it for less.
    pub fn evidence(&self) -> u64 {
        self.observations.min(EXTERNAL_GRADE_EVIDENCE_CAP)
    }

    /// Whether this source was last updated more than
    /// [`EXTERNAL_GRADE_MAX_AGE_SECS`] before `now`.
    pub fn is_stale_at(&self, now: u64) -> bool {
        now.saturating_sub(self.updated_at) > EXTERNAL_GRADE_MAX_AGE_SECS
    }
}

/// The key an external grade is stored under in
/// [`TaskStats::external_grades`]: the source name for a report from an
/// operator (`reporter` is `None`), `<reporter>/<source>` otherwise. A source
/// name cannot contain `/`, so one reporter can neither write under another's
/// key nor under an operator's.
pub fn external_grade_key(reporter: Option<&str>, source: &str) -> String {
    match reporter {
        Some(reporter) => format!("{reporter}/{source}"),
        None => source.to_string(),
    }
}

/// The reporter part of an external grade key: `None` for an operator's.
pub fn external_grade_reporter(key: &str) -> Option<&str> {
    key.split_once('/').map(|(reporter, _)| reporter)
}

/// Why an external grade was not recorded.
#[derive(Debug, Clone, PartialEq)]
pub enum ExternalGradeRefusal {
    /// `quality` was not a finite number in `[0, 1]`.
    InvalidQuality,
    /// `n` was zero or above [`MAX_EXTERNAL_GRADE_CASES`].
    InvalidCases,
    /// The source name was empty, too long, or used characters outside
    /// `[A-Za-z0-9._:-]`.
    InvalidSource,
    /// The model id was empty or longer than 256 bytes.
    InvalidModel,
    /// The reporter identity was empty, too long, or contained `/` or a
    /// character outside printable ASCII. Set by the daemon, never the caller.
    InvalidReporter,
    /// The reporter already keeps [`MAX_EXTERNAL_GRADE_SOURCES`] other current
    /// sources on this task.
    SourceLimitReached,
}

impl ExternalGradeRefusal {
    /// Stable machine-readable reason, carried on the wire as `data.reason`.
    pub fn reason(&self) -> &'static str {
        match self {
            Self::InvalidQuality => "invalid_quality",
            Self::InvalidCases => "invalid_n",
            Self::InvalidSource => "invalid_source",
            Self::InvalidModel => "invalid_model",
            Self::InvalidReporter => "invalid_reporter",
            Self::SourceLimitReached => "source_limit_reached",
        }
    }
}

impl std::fmt::Display for ExternalGradeRefusal {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::InvalidQuality => write!(f, "quality must be a number in [0, 1]"),
            Self::InvalidCases => {
                write!(f, "n must be between 1 and {MAX_EXTERNAL_GRADE_CASES}")
            }
            Self::InvalidSource => write!(
                f,
                "source must be 1-{MAX_EXTERNAL_GRADE_SOURCE_LEN} characters of [A-Za-z0-9._:-]"
            ),
            Self::InvalidModel => write!(f, "model_id must be 1-256 bytes"),
            Self::InvalidReporter => write!(
                f,
                "reporter must be 1-{MAX_EXTERNAL_GRADE_REPORTER_LEN} printable ASCII characters \
                 without `/`"
            ),
            Self::SourceLimitReached => write!(
                f,
                "this caller already keeps {MAX_EXTERNAL_GRADE_SOURCES} current external grade \
                 sources on this model's task; report under one of them, or wait for one to go \
                 stale after {} days",
                EXTERNAL_GRADE_MAX_AGE_SECS / 86_400
            ),
        }
    }
}

/// What resolving a batch of outcome signals did.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct ResolveReport {
    /// Signals that matched a pending trace and were recorded.
    pub resolved: usize,
    /// Trace ids that matched no pending trace — already resolved, never
    /// started by this tracker, or swept — so the signal was not recorded.
    pub dropped: Vec<String>,
}

/// Per-model performance profile, built from observed outcomes.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ModelProfile {
    pub model_id: String,
    pub total_calls: u64,
    pub success_count: u64,
    pub fail_count: u64,
    pub total_latency_ms: u64,
    /// Total estimated input tokens across all calls. For prompt-cached
    /// providers this is the *uncached* prefix only; cached tokens are in
    /// the two fields below so each bucket can be priced at its own rate.
    #[serde(default)]
    pub total_input_tokens: u64,
    /// Total estimated output tokens across all calls.
    #[serde(default)]
    pub total_output_tokens: u64,
    /// Total prompt-cache *read* (hit) input tokens — billed at ~0.1× input.
    #[serde(default)]
    pub total_cache_read_input_tokens: u64,
    /// Total prompt-cache *write* input tokens — billed at ~1.25×/2× input.
    #[serde(default)]
    pub total_cache_creation_input_tokens: u64,
    /// Per-task statistics.
    pub task_stats: HashMap<String, TaskStats>,
    /// Overall EMA quality score (0.0 - 1.0).
    pub ema_quality: f64,
    /// Number of benchmark cases behind `ema_quality` when it is a *seeded
    /// cold-start prior* (`total_calls == 0`). See `TaskStats::prior_sample_size`.
    #[serde(default)]
    pub prior_sample_size: usize,
    /// Graded quality observations that have moved `ema_quality` (overall).
    /// See `TaskStats::quality_observations`.
    #[serde(default)]
    pub quality_observations: u64,
    /// Derived metric: quality per 1K total tokens. Populated on export
    /// (not on every update) so it always reflects the latest snapshot.
    /// Inspired by Meta-Harness: context-token efficiency is a first-class
    /// optimization target, so it needs to be visible in model_stats.
    #[serde(default)]
    pub quality_per_1k_tokens: f64,
    /// The [`QUALITY_EVIDENCE_VERSION`] this profile's quality was recorded
    /// under. Absent (0) on profiles saved before the version existed.
    #[serde(default)]
    pub quality_evidence_version: u32,
    /// Last updated (unix timestamp).
    pub updated_at: u64,
}

impl ModelProfile {
    pub fn new(model_id: String) -> Self {
        Self {
            model_id,
            total_calls: 0,
            success_count: 0,
            fail_count: 0,
            total_latency_ms: 0,
            total_input_tokens: 0,
            total_output_tokens: 0,
            total_cache_read_input_tokens: 0,
            total_cache_creation_input_tokens: 0,
            task_stats: HashMap::new(),
            ema_quality: NEUTRAL_QUALITY,
            prior_sample_size: 0,
            quality_observations: 0,
            quality_per_1k_tokens: 0.0,
            quality_evidence_version: QUALITY_EVIDENCE_VERSION,
            updated_at: now_unix(),
        }
    }

    /// Success rate over *resolved* outcomes, or `None` when nothing has
    /// resolved yet. Returns `None` rather than a fabricated 0.5 so callers
    /// (all of which are display/reporting surfaces) can render "no resolved
    /// signal" instead of a misleading confident "50%" for a never-measured
    /// model. The router does NOT use this — its reliability estimate reads
    /// the raw `success_count`/`fail_count` with its own add-one prior
    /// (`adaptive_router`), so no 0.5 floor belongs here. Mirrors
    /// `LaneUsage::success_rate`'s Option semantics.
    pub fn success_rate_resolved(&self) -> Option<f64> {
        let resolved = self.success_count + self.fail_count;
        if resolved == 0 {
            None
        } else {
            Some(self.success_count as f64 / resolved as f64)
        }
    }

    pub fn avg_latency_ms(&self) -> f64 {
        if self.total_calls == 0 {
            return 0.0;
        }
        self.total_latency_ms as f64 / self.total_calls as f64
    }

    /// Same degradation pattern as SkillStats: fail_count > success_count + threshold.
    pub fn should_degrade(&self, threshold: u64) -> bool {
        self.fail_count > self.success_count + threshold
    }

    /// Get stats for a specific task type.
    pub fn task_stats(&self, task: InferenceTask) -> Option<&TaskStats> {
        self.task_stats.get(&task.to_string())
    }

    /// Total tokens observed across all calls (all input buckets + output).
    /// Includes prompt-cache read and write tokens — they are real tokens the
    /// provider processed, so price-free token efficiency must count them.
    pub fn total_tokens(&self) -> u64 {
        self.total_input_tokens
            + self.total_cache_read_input_tokens
            + self.total_cache_creation_input_tokens
            + self.total_output_tokens
    }

    /// Quality per 1000 tokens: `ema_quality * 1000 / total_tokens`.
    /// Returns 0.0 before any tokens have been observed.
    pub fn compute_quality_per_1k_tokens(&self) -> f64 {
        let total = self.total_tokens();
        if total == 0 {
            return 0.0;
        }
        self.ema_quality * 1000.0 / total as f64
    }

    /// Tokens spent per *successful* outcome: total tokens ÷ successes. `None`
    /// before any success has resolved. This is PRICE-FREE token efficiency —
    /// NOT dollar cost-per-outcome: a $0.10/Mtok and a $10/Mtok model with equal
    /// token counts score identically here. For the dollar metric the
    /// outcome-scoreboard vision wants, use [`Self::usd_per_success`] with the
    /// model's catalog prices. Pairs with [`Self::success_rate_resolved`].
    pub fn tokens_per_success(&self) -> Option<f64> {
        if self.success_count == 0 {
            None
        } else {
            Some(self.total_tokens() as f64 / self.success_count as f64)
        }
    }

    /// Dollar cost per *successful* outcome — the honest "cost-per-outcome":
    /// input+output token spend priced at the model's catalog rates, divided by
    /// successes. `None` before any success. Prices are USD per 1M tokens
    /// (from the catalog/registry `cost`), supplied by the caller so the tracker
    /// stays catalog-free. This is the metric the scoreboard should lead with —
    /// "cry once" is only legible in dollars-per-outcome, not tokens.
    pub fn usd_per_success(
        &self,
        input_per_mtok: f64,
        output_per_mtok: f64,
        cache: CacheRates,
    ) -> Option<f64> {
        if self.success_count == 0 {
            return None;
        }
        let usd = priced_input_usd(
            self.total_input_tokens,
            self.total_cache_read_input_tokens,
            self.total_cache_creation_input_tokens,
            input_per_mtok,
            cache,
        ) + self.total_output_tokens as f64 * output_per_mtok / 1_000_000.0;
        Some(usd / self.success_count as f64)
    }
}

/// Per-provider prompt-cache economics, expressed relative to the base input
/// rate. The providers differ structurally: Anthropic uses explicit
/// breakpoints with a deep read discount and a write premium; OpenAI caches
/// automatically with a shallower read discount and no write charge. The
/// durable ledger is catalog-free, so the caller derives these from the model
/// schema's protocol (see [`crate::schema::ApiProtocol::cache_rates`]) and
/// passes them in alongside the per-token prices.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct CacheRates {
    /// Cached-read (hit) price ÷ base input. Anthropic ~0.1×, OpenAI ~0.5×.
    pub read_mult: f64,
    /// Cache-write price ÷ base input. Anthropic ~1.25× (5-minute TTL; 1-hour
    /// is 2× but the ledger doesn't record which TTL, so writes bill at 1.25×
    /// — exact for the default path, a conservative under-estimate for 1-hour).
    /// OpenAI has no write charge (`0`) and reports no write tokens.
    pub write_mult: f64,
}

impl CacheRates {
    /// Anthropic / Bedrock-Claude: deep read discount + write premium.
    pub const ANTHROPIC: Self = Self {
        read_mult: 0.1,
        write_mult: 1.25,
    };
    /// OpenAI (incl. Azure OpenAI): automatic caching, ~0.5× read, no write.
    pub const OPENAI: Self = Self {
        read_mult: 0.5,
        write_mult: 0.0,
    };
    /// Providers without parsed prompt caching (Google/Vertex, local). Their
    /// cache token counts are always zero, so the multipliers are never
    /// exercised — this is just an honest, inert default.
    pub const NONE: Self = Self {
        read_mult: 0.0,
        write_mult: 0.0,
    };
}

impl Default for CacheRates {
    /// Anthropic was the first (and write-bucket-bearing) provider; defaulting
    /// here keeps a missing-rates path conservative rather than free.
    fn default() -> Self {
        Self::ANTHROPIC
    }
}

/// USD cost of the three input buckets (uncached prefix, cache read, cache
/// write), each priced off the same base `input_per_mtok` rate with the
/// provider's [`CacheRates`] applied. Shared by [`ModelProfile::usd_per_success`]
/// and the durable [`crate::scoreboard::Scoreboard`] so both price caching the
/// same way.
pub fn priced_input_usd(
    uncached_input_tokens: u64,
    cache_read_input_tokens: u64,
    cache_creation_input_tokens: u64,
    input_per_mtok: f64,
    cache: CacheRates,
) -> f64 {
    (uncached_input_tokens as f64 * input_per_mtok
        + cache_read_input_tokens as f64 * input_per_mtok * cache.read_mult
        + cache_creation_input_tokens as f64 * input_per_mtok * cache.write_mult)
        / 1_000_000.0
}

/// EMA smoothing factor. Higher = more weight on recent observations.
const EMA_ALPHA: f64 = 0.2;

/// One resolved inference outcome — the durable, attributable "receipt"
/// the concierge quotes ("routed to X at T because Y; latency 1.2s;
/// outcome success q=0.9"). Append-only; one JSON line per resolution.
/// Deliberately flat (success/quality/error pulled out of the richer
/// `InferenceOutcome`) so it's stable to read back and cheap to reason
/// over. Carries no prompt/output text — only routing-adjacent metadata.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OutcomeLedgerEntry {
    pub trace_id: String,
    pub model_id: String,
    pub task: InferenceTask,
    pub routing_reason: String,
    pub latency_ms: u64,
    pub input_tokens: usize,
    pub output_tokens: usize,
    /// Prompt-cache read (hit) input tokens, billed at ~0.1× input. Defaults
    /// to 0 so receipts written before caching telemetry existed read back
    /// (and price) as fully-uncached.
    #[serde(default, skip_serializing_if = "is_zero_usize")]
    pub cache_read_input_tokens: usize,
    /// Prompt-cache write input tokens, billed at ~1.25×/2× input.
    #[serde(default, skip_serializing_if = "is_zero_usize")]
    pub cache_creation_input_tokens: usize,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub success: Option<bool>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub quality: Option<f64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    /// Project/workspace the call belonged to — the key B1 groups paired
    /// comparisons on. `None` until the capture path threads it (B1); kept
    /// in the schema now so today's receipts are forward-compatible.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub project_id: Option<String>,
    /// Intent/use-case lane (finer than `task`), if the router computed
    /// one. `None` until threaded (B1).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub intent: Option<String>,
    /// Why an unattributed receipt (`success: None` with an `error`) was not
    /// the model's fault. `None` on attributed receipts and on unattributed
    /// ones written before reasons were recorded.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub attribution: Option<UnattributedReason>,
    /// The CAR version that recorded this receipt. A model's failures under an
    /// older CAR are not evidence against it: a release can fix the loader
    /// (gemma-4's nested `vocab_size` failed every load until it did), and
    /// counting those failures would retire weights for CAR's own bug. `None`
    /// on receipts written before versions were recorded.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub car_version: Option<String>,
    pub timestamp: u64,
}

impl OutcomeLedgerEntry {
    /// Whether this receipt is evidence the model cannot run: a failure the
    /// model was blamed for, recorded by this CAR. A failure an older build
    /// recorded may have been that build's bug (gemma-4's nested `vocab_size`
    /// failed every load until a release read it), so it never condemns
    /// weights this build might run.
    ///
    /// The rule scopes the portfolio's `cannot_load`, the one verdict that
    /// deletes. Routing health keeps older builds' failures:
    /// - the circuit breaker is per session, in memory, so nothing an older
    ///   build recorded holds it open after a restart;
    /// - the lifetime `fail_count` sets the router's completion rate,
    ///   `(completed + 1) / (resolved + 1)`, which scales the model's
    ///   reliability whichever build recorded the failures; successes earn it
    ///   back.
    ///
    /// Subtracting old-build failures at reconciliation would need a once-only
    /// stamp on every failure receipt at every release, and would hand a
    /// genuinely broken model a clean slate each upgrade — which is why
    /// [`plan_reconcile`] keeps the counters the profile's.
    pub fn counts_against_model(&self) -> bool {
        self.success == Some(false)
            && self.attribution.is_none()
            && self.car_version.as_deref() == Some(env!("CARGO_PKG_VERSION"))
    }
}

/// Why a failed attempt was kept out of the model's health. Recorded on the
/// receipt so a reader — the concierge quoting "refused for memory 20 times",
/// a lane deciding what counts as friction — branches on a type instead of
/// re-reading error text.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum UnattributedReason {
    /// The request asked for something this model cannot do (a mode, a
    /// response format, a catalog precondition).
    Capability,
    /// The account behind the model: a missing key, an expired sign-in, no
    /// credits, no workspace, a gateway with no upstream.
    Account,
    /// This machine could not hold the model while keeping CAR's memory
    /// policy and emergency reserve. The model never ran.
    ResourceBlocked,
    /// A filter in front of the model ruled on the request's content.
    ContentRefused,
    /// The caller stopped it: a dropped stream, a spend limit, a controlled
    /// termination.
    Cancelled,
    /// A model id the registry does not know and that never ran — a test
    /// pinning a made-up id against the live daemon.
    Unregistered,
}

/// Serde skip helper: keep the cache-token fields out of receipts that never
/// touched the cache, so the durable ledger stays byte-identical to its
/// pre-caching shape for the common (uncached) case.
fn is_zero_usize(n: &usize) -> bool {
    *n == 0
}

/// Cap on the in-memory ledger buffer between flushes — backstop against
/// unbounded growth if the flush timer never runs. Oldest entries drop.
const MAX_LEDGER_BUFFER: usize = 5000;

/// Max characters of an error string kept in a ledger receipt. Errors are
/// useful signal ("429", "context overflow") but can echo provider text
/// or paths — cap the length so the privacy-bounded ledger never grows a
/// large/sensitive blob. The classification lives in the prefix.
const MAX_LEDGER_ERROR_CHARS: usize = 256;

/// Truncate an error to [`MAX_LEDGER_ERROR_CHARS`] on a char boundary.
fn redact_error(error: &str) -> String {
    if error.chars().count() <= MAX_LEDGER_ERROR_CHARS {
        return error.to_string();
    }
    let truncated: String = error.chars().take(MAX_LEDGER_ERROR_CHARS).collect();
    format!("{truncated}…")
}

/// Classify an inference error as a **no-answer failure** — the call never
/// produced an answer to grade — rather than a *generation* failure (the model
/// returned content that was judged bad). Two families qualify:
///   - transport/infrastructure: read timeout, connection reset, 5xx, EOF,
///     overload.
///   - auth/client rejection: 401/403/400/404, bad API key, `invalid_client`.
///     The request was refused before any tokens were generated.
///
/// Such a call carries NO information about the model's *answer* quality — like a
/// *mechanical* success, it leaves quality unknown — so it must not move the
/// quality EMA. Counting these toward quality silently buries a good-but-slow,
/// transiently-unreachable, or merely *mis-authenticated* model: observed live
/// as `parslee/*` collapsing to `ema_quality ≈ 0.003` (from read-timeouts AND
/// ~47 `HTTP 401 Unauthorized` from an undeployed gateway), which then
/// deprioritized it under every workload. (Rate-limit / 429 is handled
/// separately as a harsher *availability* penalty, not a quality one.)
///
/// This is the principled cut: a failure is answer-quality evidence ONLY when
/// the model actually produced an answer. Auth/HTTP/transport rejections didn't,
/// so the default for them is quality-neutral. We match on unambiguous textual
/// forms, NOT bare status numbers (a bare "400"/"500" substring matches
/// "1500ms", "tokens: 400", etc.) — a status-only error with no recognizable
/// phrase falls through to the generation-failure path, which is the safe
/// direction (it can only over-penalize a model that *did* answer, never
/// silently bury one that didn't).
fn is_no_answer_failure(error: &str) -> bool {
    let e = error.to_ascii_lowercase();
    const NEEDLES: &[&str] = &[
        // --- transport / infrastructure (no answer produced) ---
        "timeout",
        "timed out",
        "deadline",
        "connection",
        "connect error",
        "reset by peer",
        "broken pipe",
        "network",
        "unreachable",
        "eof",
        "stream closed",
        "socket",
        "503",
        "502",
        "504",
        // A 500 is a server-side fault that almost always produced no answer,
        // but bare "500" is too broad a substring (it matches "1500ms",
        // "tokens: 500", a 500-dim error, …), so we match only its unambiguous
        // textual form. A status-only "HTTP 500" with no body stays classified
        // as a generation failure — accept that edge rather than risk
        // false-positive hits on coincidental "500" substrings.
        "internal server error",
        "service unavailable",
        "temporarily unavailable",
        "overloaded",
        // --- auth / client rejection (request refused, no answer produced) ---
        // Same bare-number caution as 5xx: match textual forms, not "401"/"400".
        // Kept HIGH-PRECISION: only phrases that unambiguously mean "the request
        // was refused before generation." Deliberately NOT "not found" (matches a
        // generation/parse error like "field not found") or "invalid token"
        // (collides with tokenizer errors) — a rare 404 mis-scored as a bad
        // answer is the safe direction; silently burying an auth-blocked model is
        // not.
        "unauthorized",
        "unauthenticated",
        "forbidden",
        "permission denied",
        "authentication",
        "invalid_client",
        "invalid api key",
        "bad request",
    ];
    NEEDLES.iter().any(|n| e.contains(n))
}

/// Rewrite the ledger keeping only the most recent `max_entries` receipts
/// (retention bound). No-op when under the cap or the file is absent.
/// Atomic (temp + rename) so a crash mid-prune can't corrupt the ledger.
pub fn prune_ledger(path: &std::path::Path, max_entries: usize) -> std::io::Result<()> {
    if max_entries == 0 || !path.exists() {
        return Ok(());
    }
    let entries = read_ledger(path, 0);
    if entries.len() <= max_entries {
        return Ok(());
    }
    rewrite_ledger(path, &entries[entries.len() - max_entries..])
}

/// Replace the ledger with `entries`: written to a uniquely named temp file
/// in the same directory, made owner-only, synced, and renamed over the
/// ledger, then the directory is synced. A shared temp name let a prune and a
/// reconcile truncate each other's half-written file, and `fs::write` left the
/// rewritten ledger world-readable — undoing the 0600 an append sets on a file
/// holding error text.
fn rewrite_ledger(path: &std::path::Path, entries: &[OutcomeLedgerEntry]) -> std::io::Result<()> {
    use std::io::Write;
    let dir = path
        .parent()
        .filter(|d| !d.as_os_str().is_empty())
        .unwrap_or_else(|| std::path::Path::new("."));
    let mut tmp = tempfile::Builder::new()
        .prefix(".outcome_ledger.")
        .suffix(".tmp")
        .tempfile_in(dir)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        tmp.as_file()
            .set_permissions(std::fs::Permissions::from_mode(0o600))?;
    }
    for e in entries {
        let line = serde_json::to_string(e).map_err(std::io::Error::other)?;
        tmp.write_all(line.as_bytes())?;
        tmp.write_all(b"\n")?;
    }
    tmp.as_file().sync_all()?;
    car_secrets::harden_owner_only(tmp.path());
    tmp.persist(path).map_err(|e| e.error)?;
    #[cfg(unix)]
    if let Ok(d) = std::fs::File::open(dir) {
        let _ = d.sync_all();
    }
    Ok(())
}

/// What [`plan_reconcile`] changed. All zero when the ledger and profiles
/// were already consistent.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct LedgerReconciliation {
    /// Failure receipts an older build wrote for errors current code books as
    /// unattributed, now marked with the reason it would record.
    pub reclassified: usize,
    /// Failure receipts for model ids the registry does not know and that
    /// never ran, now marked [`UnattributedReason::Unregistered`].
    pub unregistered: usize,
    /// Profiles whose counters and quality evidence were rebuilt.
    pub profiles_rebuilt: usize,
    /// Profiles of never-run unregistered ids, removed.
    pub profiles_removed: usize,
}

/// The reason current code would record for a legacy receipt's error, when it
/// would not attribute that error to the model at all.
///
/// Receipts written before those errors were typed carry only their rendered
/// text, so this is the one place that classifies by text — on history that
/// has no type to read. Live booking routes on the [`crate::InferenceError`]
/// variant and records an [`UnattributedReason`]; a receipt that carries one is
/// never re-read here. The prefixes are the renderings of the producers named
/// beside them; the ones whose producer no longer exists are kept as they
/// were written.
fn legacy_unattributed(error: &str) -> Option<UnattributedReason> {
    use UnattributedReason as R;
    const PREFIXES: &[(&str, UnattributedReason)] = &[
        // `From<LocalAdmissionError> for InferenceError`: each verdict's
        // recovery text, which is the `LocalResourceBlocked` Display.
        (
            "Not enough memory is free to start this model",
            R::ResourceBlocked,
        ),
        ("This model needs about ", R::ResourceBlocked),
        (
            "Local model loading is disabled by the 0 GB allocation",
            R::ResourceBlocked,
        ),
        ("CAR could not measure live memory", R::ResourceBlocked),
        (
            "This local model is being removed or maintained",
            R::ResourceBlocked,
        ),
        (
            "CAR is still confirming that the previous local model process exited",
            R::ResourceBlocked,
        ),
        // `InferenceError::ProviderKeyMissing` (typed 2026-09-15); its Display
        // kept the `InferenceFailed` rendering.
        (
            "inference failed: no API keys available for endpoint ",
            R::Account,
        ),
        // The Parslee org lookup's 401, before it was `CredentialUnavailable`.
        (
            "inference failed: Parslee org lookup failed: HTTP 401",
            R::Account,
        ),
        // A FoundationModels guardrail or refusal, before it was typed
        // `ContentRefused`.
        (
            "inference failed: FoundationModels declined the request (",
            R::ContentRefused,
        ),
    ];
    if let Some((_, reason)) = PREFIXES
        .iter()
        .find(|(prefix, _)| error.starts_with(prefix))
    {
        return Some(*reason);
    }
    // `UnsupportedMode` rendered through `InferenceFailed`: anchored on the
    // variant's own Display, not the bare word "mode".
    (error.starts_with("inference failed: mode ") && error.contains(" not implemented on backend "))
        .then_some(R::Capability)
}

/// The quality a profile and its tasks started from before any outcome moved
/// them: the benchmark prior when one seeded the profile, otherwise neutral.
/// Read from today's priors files, which may have changed since the profile
/// was seeded — the best available stand-in, not a record.
#[derive(Debug, Clone, Default)]
pub struct QualitySeed {
    pub overall: Option<f64>,
    pub tasks: HashMap<String, f64>,
}

/// A planned reconciliation. Nothing is applied yet: `ledger`, when present,
/// is what the ledger file must become, and `profiles` the corrected profile
/// set to commit once it has.
#[derive(Debug, Default)]
pub struct ReconcilePlan {
    pub report: LedgerReconciliation,
    pub ledger: Option<Vec<OutcomeLedgerEntry>>,
    pub profiles: Option<HashMap<String, ModelProfile>>,
}

/// Make the ledger and the persisted profiles agree about which failures were
/// the model's.
///
/// Older builds booked failures a model was not responsible for — admission
/// refusals of this machine's memory, missing keys, expired sign-ins, content
/// rulings, ids that were never registered. This marks those receipts with the
/// [`UnattributedReason`] current code would record and corrects each affected
/// profile:
///
/// - **Counters stay the profile's.** The ledger is not a complete record of
///   successes — `record_complete` credits a mechanical success to the profile
///   and the receipt waits for a sweep that a short-lived process exits before
///   — while every failure is written at once. Rebuilding counts from the
///   ledger would pair a full failure history with a partial success history
///   and bury exactly the local models this repair is for. So the failures
///   stamped **in this run** are subtracted from `fail_count`, `total_calls`
///   and the task's `failures`, the exact inverse of what `record_failure`
///   added, and nothing else is touched. A profile breaking
///   `total_calls >= success_count + fail_count` is clamped, not rebuilt.
/// - **Quality evidence is replayed from the ledger.** Graded receipts are
///   written when the grade arrives, so for these the ledger is complete:
///   when a profile carries more `quality_observations` than its graded
///   receipts account for — the misattributed case — the count and the EMA are
///   replayed from those receipts in order, starting from the benchmark seed
///   where one seeded the profile (`seeds`). The replay is a function of the
///   ledger alone, so repeating it is harmless.
///
/// Subtraction must happen once per receipt, so it rides on the stamp. The
/// plan touches nothing: the caller writes the stamped ledger **first**, and
/// only once that succeeds commits `profiles` into the tracker and saves them.
/// A failed write leaves the tracker untouched; a crash after the write keeps
/// those failures counted (the conservative direction). Neither can subtract
/// twice; rows an earlier version already
/// rewrote to `success: None` subtract nothing, because that version already
/// took them out of the counters. A reconciled state root has nothing left to
/// stamp, so steady-state starts change nothing.
///
/// An id is unregistered only if the registry does not know it, no receipt
/// shows it succeeding, and no profile credits it a success — an id that ran
/// keeps its history whole. Nothing is deleted from the ledger.
pub fn plan_reconcile(
    path: &std::path::Path,
    tracker: &OutcomeTracker,
    is_registered: &dyn Fn(&str) -> bool,
    seeds: &HashMap<String, QualitySeed>,
) -> ReconcilePlan {
    let mut plan = ReconcilePlan::default();
    // Corrections are made on a copy and committed by the caller only once
    // the stamped ledger is on disk — see [`OutcomeTracker::commit_reconciled`].
    let mut profiles = tracker.profiles.clone();
    let mut profiles_changed = false;
    if !path.exists() {
        return plan;
    }
    let mut entries = read_ledger(path, 0);

    let ran: std::collections::HashSet<&str> = entries
        .iter()
        .filter(|e| e.success == Some(true))
        .map(|e| e.model_id.as_str())
        .collect();
    let never_ran: std::collections::HashSet<String> = entries
        .iter()
        .map(|e| e.model_id.as_str())
        .filter(|id| {
            !ran.contains(id)
                && !is_registered(id)
                && profiles.get(*id).is_none_or(|p| p.success_count == 0)
        })
        .map(str::to_string)
        .collect();

    let mut touched = std::collections::BTreeSet::new();
    // Failures stamped in this run, per model and task: what to take back out
    // of the profile counters.
    let mut unbooked: HashMap<String, HashMap<String, u64>> = HashMap::new();
    for entry in &mut entries {
        if entry.attribution.is_some() {
            continue;
        }
        let legacy = entry.error.as_deref().and_then(legacy_unattributed);
        let reason = if never_ran.contains(&entry.model_id) && entry.success == Some(false) {
            UnattributedReason::Unregistered
        } else if let Some(reason) = legacy.filter(|_| entry.success != Some(true)) {
            // `success: None` here is either a row the first version of this
            // pass already reclassified or an unattributed row written before
            // reasons existed; stamping it changes nothing it is counted as.
            reason
        } else {
            continue;
        };
        match reason {
            UnattributedReason::Unregistered => plan.report.unregistered += 1,
            _ if entry.success == Some(false) => plan.report.reclassified += 1,
            _ => {}
        }
        if entry.success == Some(false) {
            *unbooked
                .entry(entry.model_id.clone())
                .or_default()
                .entry(entry.task.to_string())
                .or_default() += 1;
        }
        entry.success = None;
        entry.quality = None;
        entry.attribution = Some(reason);
        touched.insert(entry.model_id.clone());
    }
    let ledger_changed = !touched.is_empty();

    // Clamp profiles whose counters contradict themselves.
    for profile in profiles.values_mut() {
        let floor = profile.success_count + profile.fail_count;
        if profile.total_calls < floor {
            profile.total_calls = floor;
            plan.report.profiles_rebuilt += 1;
            profiles_changed = true;
        }
    }

    for model_id in &touched {
        if never_ran.contains(model_id) {
            if profiles.remove(model_id).is_some() {
                plan.report.profiles_removed += 1;
                profiles_changed = true;
            }
            continue;
        }
        let Some(profile) = profiles.get_mut(model_id) else {
            continue;
        };
        let rows: Vec<&OutcomeLedgerEntry> =
            entries.iter().filter(|e| &e.model_id == model_id).collect();
        let seed = seeds.get(model_id);
        let unbook = unbooked.get(model_id);
        if correct_profile(profile, &rows, unbook, seed) {
            plan.report.profiles_rebuilt += 1;
            profiles_changed = true;
        }
    }

    if ledger_changed {
        plan.ledger = Some(entries);
    }
    if profiles_changed {
        plan.profiles = Some(profiles);
    }
    plan
}

/// Write the ledger a [`ReconcilePlan`] produced. Done before the corrected
/// profiles are committed, so a failed write leaves them as they were.
pub fn write_reconciled_ledger(
    path: &std::path::Path,
    entries: &[OutcomeLedgerEntry],
) -> std::io::Result<()> {
    rewrite_ledger(path, entries)
}

/// Correct one profile: take the failures stamped in this run back out of its
/// counters (`unbook`, per task), then replay its quality evidence from the
/// graded receipts. Returns whether anything changed.
fn correct_profile(
    profile: &mut ModelProfile,
    rows: &[&OutcomeLedgerEntry],
    unbook: Option<&HashMap<String, u64>>,
    seed: Option<&QualitySeed>,
) -> bool {
    let before = serde_json::to_value(&*profile).ok();

    // Exactly what `record_failure` added: a call, a failure, a task failure.
    for (task, n) in unbook.into_iter().flatten() {
        profile.total_calls = profile.total_calls.saturating_sub(*n);
        profile.fail_count = profile.fail_count.saturating_sub(*n);
        if let Some(ts) = profile.task_stats.get_mut(task) {
            ts.failures = ts.failures.saturating_sub(*n);
        }
    }

    // Attributed receipts only: an unattributed one never touched a profile.
    let attributed: Vec<&OutcomeLedgerEntry> = rows
        .iter()
        .copied()
        .filter(|e| e.attribution.is_none())
        .collect();
    let in_task =
        |e: &&OutcomeLedgerEntry, task: Option<&str>| task.is_none_or(|t| e.task.to_string() == t);

    // Quality evidence: the graded receipts, in order, from the seed.
    // A failure receipt's `0.0` is answer quality for the scoreboard, not a
    // routing grade (QUALITY_EVIDENCE_VERSION).
    let graded: Vec<(&OutcomeLedgerEntry, f64)> = attributed
        .iter()
        .copied()
        .filter(|e| e.success != Some(false))
        .filter_map(|e| e.quality.map(|q| (e, q)))
        .collect();
    let replay = |start: f64, task: Option<&str>| {
        let mut ema = start;
        let mut n = 0u64;
        for (_, q) in graded.iter().filter(|(e, _)| in_task(e, task)) {
            ema = ema * (1.0 - EMA_ALPHA) + q * EMA_ALPHA;
            n += 1;
        }
        (ema, n)
    };
    // A seeded profile replays from its seed; one whose seed is gone (the
    // priors file was removed) cannot be replayed honestly and is left.
    let profile_start = match (profile.prior_sample_size, seed.and_then(|s| s.overall)) {
        (0, _) => Some(NEUTRAL_QUALITY),
        (_, Some(seeded)) => Some(seeded.clamp(0.0, 1.0)),
        (_, None) => None,
    };
    if let Some(start) = profile_start {
        let (ema, n) = replay(start, None);
        if profile.quality_observations > n {
            profile.ema_quality = ema;
            profile.quality_observations = n;
        }
    }
    for (task, ts) in profile.task_stats.iter_mut() {
        let seeded = seed.and_then(|s| s.tasks.get(task)).copied();
        let start = match (ts.prior_sample_size, seeded) {
            (0, _) => Some(TaskStats::default().ema_quality),
            (_, Some(seeded)) => Some(seeded.clamp(0.0, 1.0)),
            (_, None) => None,
        };
        let Some(start) = start else { continue };
        let (ema, n) = replay(start, Some(task));
        if ts.quality_observations > n {
            // With no graded evidence left an unseeded task returns to
            // neutral, which also keeps `backfill_quality_observations` from
            // refilling its weight from `failures` on the next load.
            ts.ema_quality = if n == 0 && ts.prior_sample_size == 0 {
                NEUTRAL_QUALITY
            } else {
                ema
            };
            ts.quality_observations = n;
        }
    }
    let changed = serde_json::to_value(&*profile).ok() != before;
    if changed {
        profile.updated_at = now_unix();
    }
    changed
}

/// Append resolved outcome receipts to the JSONL ledger (one line each).
/// Append-only: no full-file rewrite, so this never triggers the
/// write-storm the profiles file had. Creates the file/parent on first
/// write.
pub fn append_ledger_entries(
    path: &std::path::Path,
    entries: &[OutcomeLedgerEntry],
) -> std::io::Result<()> {
    if entries.is_empty() {
        return Ok(());
    }
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    use std::io::Write;
    let mut opts = std::fs::OpenOptions::new();
    opts.create(true).append(true);
    // Local, privacy-bounded file (error text, soon project paths): owner
    // read/write only.
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        opts.mode(0o600);
    }
    // On Windows `mode(0o600)` is ignored; lock the file owner-only via ACL on
    // a fresh create only (idempotent). No-op off Windows.
    let existed = path.exists();
    let mut f = opts.open(path)?;
    if !existed {
        car_secrets::harden_owner_only(path);
    }
    for e in entries {
        let line = serde_json::to_string(e).map_err(std::io::Error::other)?;
        f.write_all(line.as_bytes())?;
        f.write_all(b"\n")?;
    }
    Ok(())
}

/// Read the most recent `limit` ledger entries (0 = all). Tolerant of
/// partial/garbage lines (skips them) so a torn append never poisons a
/// read — the consumer (UsageProfile) gets whatever is well-formed.
pub fn read_ledger(path: &std::path::Path, limit: usize) -> Vec<OutcomeLedgerEntry> {
    let Ok(content) = std::fs::read_to_string(path) else {
        return Vec::new();
    };
    let mut out: Vec<OutcomeLedgerEntry> = content
        .lines()
        .filter(|l| !l.trim().is_empty())
        .filter_map(|l| serde_json::from_str(l).ok())
        .collect();
    if limit > 0 && out.len() > limit {
        out = out.split_off(out.len() - limit);
    }
    out
}

tokio::task_local! {
    /// Whether calls made inside the scope may teach the router (see
    /// [`scope_learning`]).
    static LEARNING: bool;
    /// Set once a call in the scope is dispatched to a model (see
    /// [`scope_dispatch_probe`]).
    static DISPATCHED: std::sync::Arc<std::sync::atomic::AtomicBool>;
}

/// Run `future`, setting `probe` once one of its calls is dispatched to a
/// model: when the engine starts an attempt on a chosen model, past routing
/// and every refusal raised before dispatch. A caller that meters a call uses
/// it to count only calls that reached a model.
pub async fn scope_dispatch_probe<F: std::future::Future>(
    probe: std::sync::Arc<std::sync::atomic::AtomicBool>,
    future: F,
) -> F::Output {
    DISPATCHED.scope(probe, future).await
}

/// Mark the call in this scope as dispatched to a model (see
/// [`scope_dispatch_probe`]): an attempt was recorded on a chosen model, or a
/// streamed request's bytes were handed to the HTTP client.
pub(crate) fn note_dispatch() {
    let _ = DISPATCHED.try_with(|probe| probe.store(true, std::sync::atomic::Ordering::SeqCst));
}

/// Run `future` with the outcome tracker learning from its calls or not. A
/// worker's calls run with `learn = false`, so the adaptive router's scores
/// and outcome history neither steer nor learn from them
/// (docs/proposals/orchestrator-workers.md, I5). Outside any scope every call
/// is learned from, as before.
pub async fn scope_learning<F: std::future::Future>(learn: bool, future: F) -> F::Output {
    LEARNING.scope(learn, future).await
}

pub(crate) fn learning_suppressed() -> bool {
    LEARNING.try_with(|learn| !*learn).unwrap_or(false)
}

/// How long a call that ended stays pending — gradeable through
/// `outcomes.resolve_pending` — before the daemon's sweep records it
/// ungraded: fifteen minutes from the call's END. Long enough for a judge
/// that is itself a multi-minute generation; mechanical success is already
/// credited at completion, so the wait costs only memory and receipt latency.
pub const PENDING_GRADE_TTL_SECS: u64 = 15 * 60;

/// How long a call still in flight stays pending before the sweep treats it
/// as crashed: two hours from its start. See [`OutcomeTracker::sweep_pending`].
pub const PENDING_IN_FLIGHT_TTL_SECS: u64 = 2 * 60 * 60;

/// Tracks inference outcomes and builds performance profiles.
pub struct OutcomeTracker {
    /// In-memory profiles, keyed by model_id.
    profiles: HashMap<String, ModelProfile>,
    /// Pending outcomes: completed inference calls awaiting outcome signal.
    /// Keyed by trace_id.
    pending: HashMap<String, InferenceOutcome>,
    /// Counter for generating trace IDs.
    trace_counter: u64,
    /// Models excluded for this session (429/rate-limited). Hard exclusion.
    excluded: HashSet<String>,
    /// Set whenever a persisted field (a `ModelProfile`) changes; cleared
    /// on save. Lets the engine debounce disk writes — persist only when
    /// there's something new, instead of serializing the whole file after
    /// every inference call. `excluded`/`pending` are session-only and do
    /// not flip this.
    dirty: bool,
    /// Resolved outcome receipts awaiting append to the JSONL ledger.
    /// Drained by the engine on flush. Bounded by [`MAX_LEDGER_BUFFER`].
    ledger_buffer: std::collections::VecDeque<OutcomeLedgerEntry>,
}

impl OutcomeTracker {
    pub fn new() -> Self {
        Self {
            profiles: HashMap::new(),
            pending: HashMap::new(),
            trace_counter: 0,
            excluded: HashSet::new(),
            dirty: false,
            ledger_buffer: std::collections::VecDeque::new(),
        }
    }

    /// Push a resolved receipt to the ledger buffer (capped; oldest drops).
    fn push_ledger(&mut self, entry: OutcomeLedgerEntry) {
        if self.ledger_buffer.len() >= MAX_LEDGER_BUFFER {
            self.ledger_buffer.pop_front();
        }
        self.ledger_buffer.push_back(entry);
    }

    /// Drain buffered receipts for the engine to append to the ledger file.
    pub fn drain_ledger(&mut self) -> Vec<OutcomeLedgerEntry> {
        self.ledger_buffer.drain(..).collect()
    }

    /// Check if a model is excluded (rate-limited) for this session.
    pub fn is_excluded(&self, model_id: &str) -> bool {
        self.excluded.contains(model_id)
    }

    /// Record that an inference call started. Returns a trace_id.
    ///
    /// Inside [`scope_learning`]`(false, ..)` the call is not recorded at all:
    /// the trace id is returned but nothing is pending under it, so its
    /// completion, failure and any later quality signal change no profile.
    pub fn record_start(
        &mut self,
        model_id: &str,
        task: InferenceTask,
        routing_reason: &str,
    ) -> String {
        self.trace_counter += 1;
        let trace_id = format!("t-{}-{}", now_unix(), self.trace_counter);
        note_dispatch();
        if learning_suppressed() {
            return trace_id;
        }

        let outcome = InferenceOutcome {
            trace_id: trace_id.clone(),
            model_id: model_id.to_string(),
            task,
            // Length-capped: `routing_reason` is the only free-text field
            // that reaches the persisted ledger. INVARIANT: it must NEVER
            // embed prompt or output text — only a routing rationale
            // ("Code task -> Qwen3-4B"). The cap bounds accidental growth.
            routing_reason: redact_error(routing_reason),
            latency_ms: 0,
            input_tokens: 0,
            output_tokens: 0,
            cache_read_input_tokens: 0,
            cache_creation_input_tokens: 0,
            inferred_outcome: None,
            code_outcome: None,
            error: None,
            timestamp: now_unix(),
            success_credited: false,
            completed_at: None,
        };

        self.pending.insert(trace_id.clone(), outcome);
        trace_id
    }

    /// Record completion of an inference call (timing + token counts).
    pub fn record_complete(
        &mut self,
        trace_id: &str,
        latency_ms: u64,
        input_tokens: usize,
        output_tokens: usize,
    ) {
        self.record_complete_cached(trace_id, latency_ms, input_tokens, output_tokens, 0, 0);
    }

    /// Like [`Self::record_complete`] but also records the prompt-cache token
    /// split (`cache_read` = hit, `cache_creation` = write). `input_tokens`
    /// here is the *uncached* prefix only — for Anthropic that's
    /// `usage.input_tokens`, with the cached portion passed separately so cost
    /// accounting prices each bucket at its own rate. The plain 4-arg
    /// [`Self::record_complete`] forwards here with both cache buckets zeroed,
    /// which is correct for non-caching providers and local inference.
    pub fn record_complete_cached(
        &mut self,
        trace_id: &str,
        latency_ms: u64,
        input_tokens: usize,
        output_tokens: usize,
        cache_read_input_tokens: usize,
        cache_creation_input_tokens: usize,
    ) {
        if let Some(outcome) = self.pending.get_mut(trace_id) {
            outcome.completed_at = Some(now_unix());
            outcome.latency_ms = latency_ms;
            outcome.input_tokens = input_tokens;
            outcome.output_tokens = output_tokens;
            outcome.cache_read_input_tokens = cache_read_input_tokens;
            outcome.cache_creation_input_tokens = cache_creation_input_tokens;

            // Credit a *mechanical* success immediately when the call ran
            // and produced output with no error. This is symmetric with
            // `record_failure`, which books a failure the moment it happens
            // — previously a success was credited only by the 300s pending
            // sweep, so short-lived processes (the `car infer` CLI, eval
            // harnesses) exited before the sweep and NEVER recorded a
            // success, pinning working models at the 0.5 EMA prior / "0
            // successes" in the health UI (#312, only partially fixed by
            // the deferred sweep). `ema_quality` is untouched — only a real
            // downstream accept/edit signal moves quality. `success_credited`
            // stops the later sweep and any quality-signal resolution from
            // double-counting this same call.
            let mechanical_success = output_tokens > 0 && outcome.error.is_none();
            if mechanical_success {
                outcome.success_credited = true;
            }
            let model_id = outcome.model_id.clone();
            let task_key = outcome.task.to_string();

            // Update profile with timing data (the `outcome` borrow of
            // `self.pending` ends here; `self.profiles` is a disjoint field).
            let profile = self
                .profiles
                .entry(model_id.clone())
                .or_insert_with(|| ModelProfile::new(model_id));

            profile.total_calls += 1;
            profile.total_latency_ms += latency_ms;
            profile.total_input_tokens += input_tokens as u64;
            profile.total_output_tokens += output_tokens as u64;
            profile.total_cache_read_input_tokens += cache_read_input_tokens as u64;
            profile.total_cache_creation_input_tokens += cache_creation_input_tokens as u64;
            if mechanical_success {
                profile.success_count += 1;
            }

            let ts = profile.task_stats.entry(task_key).or_default();
            ts.calls += 1;
            if mechanical_success {
                ts.successes += 1;
            }
            ts.avg_latency_ms =
                ts.avg_latency_ms + (latency_ms as f64 - ts.avg_latency_ms) / ts.calls as f64;

            profile.updated_at = now_unix();
            self.dirty = true;
        }
    }

    /// Record a failure.
    pub fn record_failure(&mut self, trace_id: &str, error: &str) {
        let mut ledger_entry = None;
        if let Some(outcome) = self.pending.get_mut(trace_id) {
            outcome.error = Some(error.to_string());

            let profile = self
                .profiles
                .entry(outcome.model_id.clone())
                .or_insert_with(|| ModelProfile::new(outcome.model_id.clone()));

            // A failed call is still a call: count it in `total_calls` so the
            // denominator includes failures (record_complete counts successes
            // the same way). Without this, `total_calls` tallied only
            // completions and could read *smaller* than `fail_count` — the
            // live symptom on parslee/* models (6 calls, 14 fails).
            profile.total_calls += 1;
            profile.fail_count += 1;

            // Rate-limit errors (429) get a harsher penalty — the model is
            // guaranteed to fail again, so drop quality aggressively.
            let is_rate_limited = error.contains("429") || error.contains("RESOURCE_EXHAUSTED");
            // No-answer failures (read timeout, connection reset, 5xx, EOF, AND
            // auth/client rejections like 401/403/400) never produced an answer,
            // so answer quality is UNKNOWN — they must stay quality-neutral. The
            // call + failure are already booked above (`total_calls`/`fail_count`)
            // so availability and the circuit breaker still react; we just leave
            // `ema_quality` / `quality_observations` untouched. See
            // `is_no_answer_failure`.
            let is_no_answer = !is_rate_limited && is_no_answer_failure(error);
            if is_rate_limited {
                // Hard-exclude for the rest of this session (#13)
                self.excluded.insert(outcome.model_id.clone());
            }
            // A failure is availability evidence (the counters above, the
            // circuit breaker, the sampler's failure count), never a grade of
            // answer quality: see QUALITY_EVIDENCE_VERSION.

            let task_key = outcome.task.to_string();
            let ts = profile.task_stats.entry(task_key).or_default();
            // `failures` is an availability counter — every failure counts,
            // no-answer failures included (it feeds degradation / circuit-breaking).
            ts.failures += 1;

            profile.updated_at = now_unix();
            self.dirty = true;

            ledger_entry = Some(OutcomeLedgerEntry {
                trace_id: outcome.trace_id.clone(),
                model_id: outcome.model_id.clone(),
                task: outcome.task,
                routing_reason: outcome.routing_reason.clone(),
                latency_ms: outcome.latency_ms,
                input_tokens: outcome.input_tokens,
                output_tokens: outcome.output_tokens,
                cache_read_input_tokens: outcome.cache_read_input_tokens,
                cache_creation_input_tokens: outcome.cache_creation_input_tokens,
                success: Some(false),
                // #374 follow-up: the ledger's `quality` is ANSWER quality for the
                // scoreboard's `avg_quality` and the shadow-calibration band
                // (#369). A *generation* failure is recorded there as `0.0`. A
                // *no-answer* failure (transport/timeout/5xx, 401/403/400) or a
                // *429* produced no answer, so it is `None`. Routing quality does
                // not read failure receipts at all (QUALITY_EVIDENCE_VERSION): the
                // reconcile replay skips `success: Some(false)` rows, as the live
                // tracker no longer grades failures.
                quality: if is_rate_limited || is_no_answer {
                    None
                } else {
                    Some(0.0)
                },
                error: Some(redact_error(error)),
                project_id: None,
                intent: None,
                attribution: None,
                car_version: Some(env!("CARGO_PKG_VERSION").to_string()),
                timestamp: now_unix(),
            });
        }

        if let Some(entry) = ledger_entry {
            self.push_ledger(entry);
        }

        // Failed outcomes don't need further tracking
        self.pending.remove(trace_id);
    }

    /// Resolve a request/provider capability mismatch without treating it as a
    /// model-health or answer-quality failure.
    ///
    /// The attempt is still emitted to the receipt ledger for operator
    /// visibility, but it must not change the profile that adaptive routing
    /// uses for unrelated requests to the same model.
    pub fn record_capability_rejection(&mut self, trace_id: &str, error: &str) {
        self.record_not_model_fault(trace_id, error, UnattributedReason::Capability);
    }

    /// Resolve a provider **account** rejection — bad/absent key, or out of
    /// credits — without treating it as a model failure.
    ///
    /// Same unattributed receipt as [`record_capability_rejection`](crate::outcome::OutcomeTracker::record_capability_rejection), different
    /// reason: the condition is account-wide, so it is not evidence about the
    /// model that happened to be selected. Booking it as a failure degrades the
    /// 30-day health EMA of every model the fallback chain touches, and that
    /// penalty outlives the top-up that fixes the account (Parslee-ai/car#650).
    pub fn record_account_rejection(&mut self, trace_id: &str, error: &str) {
        self.record_not_model_fault(trace_id, error, UnattributedReason::Account);
    }

    /// Resolve a local admission refusal — the machine could not hold the
    /// model while keeping CAR's memory policy and emergency reserve — without
    /// treating it as a model failure.
    ///
    /// The model never ran. Booking the refusal as a failure measured the
    /// machine's free memory at that moment and charged it to the model: a
    /// 27B checkpoint refused five times on a busy afternoon read as 58%
    /// healthy for good, which is the number a concierge would retire it on.
    pub fn record_resource_rejection(&mut self, trace_id: &str, error: &str) {
        self.record_not_model_fault(trace_id, error, UnattributedReason::ResourceBlocked);
    }

    /// Replace the profiles with a [`ReconcilePlan`]'s corrected set. Called
    /// only after the plan's ledger is on disk — committing first would let the
    /// periodic save persist a correction whose stamps never landed, and the
    /// next start would make it again.
    pub fn commit_reconciled(&mut self, profiles: HashMap<String, ModelProfile>) {
        self.profiles = profiles;
        self.dirty = true;
    }

    /// Note how long an attempt ran before it ended without a completion — a
    /// dropped stream, for one — so its unattributed receipt keeps the time
    /// instead of reading as instantaneous. A stall the caller gave up on
    /// stays visible in the latency even though it is not booked as a failure.
    pub fn note_elapsed(&mut self, trace_id: &str, latency_ms: u64) {
        if let Some(outcome) = self.pending.get_mut(trace_id) {
            outcome.latency_ms = latency_ms;
            outcome.completed_at.get_or_insert_with(now_unix);
        }
    }

    /// Emit a receipt that records what happened without attributing it to the
    /// model: `success: None`, `quality: None`, so the ledger keeps the
    /// evidence while the routing profile stays untouched.
    pub fn record_not_model_fault(
        &mut self,
        trace_id: &str,
        error: &str,
        reason: UnattributedReason,
    ) {
        if let Some(outcome) = self.pending.remove(trace_id) {
            self.push_ledger(OutcomeLedgerEntry {
                trace_id: outcome.trace_id,
                model_id: outcome.model_id,
                task: outcome.task,
                routing_reason: outcome.routing_reason,
                latency_ms: outcome.latency_ms,
                input_tokens: outcome.input_tokens,
                output_tokens: outcome.output_tokens,
                cache_read_input_tokens: outcome.cache_read_input_tokens,
                cache_creation_input_tokens: outcome.cache_creation_input_tokens,
                success: None,
                quality: None,
                error: Some(redact_error(error)),
                project_id: None,
                intent: None,
                attribution: Some(reason),
                car_version: Some(env!("CARGO_PKG_VERSION").to_string()),
                timestamp: now_unix(),
            });
        }
    }

    /// Record an inferred outcome from conversation signals.
    pub fn record_inferred_outcome(&mut self, trace_id: &str, outcome: InferredOutcome) {
        if let Some(pending) = self.pending.remove(trace_id) {
            self.apply_outcome(&pending, outcome.quality_score(), outcome.is_success());
        }
    }

    /// Record an outcome from git-diff comparison (code generation).
    pub fn record_code_outcome(&mut self, trace_id: &str, outcome: CodeOutcome) {
        if let Some(pending) = self.pending.remove(trace_id) {
            self.apply_outcome(
                &pending,
                Some(outcome.quality_score()),
                Some(outcome.is_success()),
            );
        }
    }

    /// Resolve all pending outcomes for a completed conversation turn.
    /// Called with the inferred outcomes from conversation signal analysis.
    pub fn resolve_pending_from_signals(&mut self, outcomes: Vec<(String, InferredOutcome)>) {
        self.resolve_pending_reporting(outcomes);
    }

    /// [`Self::resolve_pending_from_signals`], reporting which signals landed
    /// and which matched no pending trace. A dropped signal changes nothing,
    /// and the report is the only place that is visible.
    pub fn resolve_pending_reporting(
        &mut self,
        outcomes: Vec<(String, InferredOutcome)>,
    ) -> ResolveReport {
        let mut report = ResolveReport::default();
        for (trace_id, inferred) in outcomes {
            if self.has_pending(&trace_id) {
                self.record_inferred_outcome(&trace_id, inferred);
                report.resolved += 1;
            } else {
                report.dropped.push(trace_id);
            }
        }
        report
    }

    /// [`Self::record_external_grade_as`] for an operator (no reporter).
    pub fn record_external_grade(
        &mut self,
        model_id: &str,
        task: InferenceTask,
        quality: f64,
        n: u32,
        source: &str,
    ) -> Result<ExternalGradeStats, ExternalGradeRefusal> {
        self.record_external_grade_as(None, model_id, task, quality, n, source)
    }

    /// Record a graded result for `model_id` on `task` that is not tied to a
    /// call this tracker served — a head-to-head or eval run outside CAR.
    ///
    /// `quality` is the mean grade over `n` cases. `reporter` is who is
    /// reporting, as the daemon knows it (`None` for an operator); the grade
    /// is stored under [`external_grade_key`]`(reporter, source)` on the
    /// task's [`TaskStats::external_grades`], so a reporter can only update
    /// its own sources. The source's quality is a running mean whose window
    /// is capped at [`EXTERNAL_GRADE_EVIDENCE_CAP`] cases; how much routing
    /// weight it carries is decided when pooling, see
    /// [`TaskStats::external_evidence_at`]. A reporter's new source past
    /// [`MAX_EXTERNAL_GRADE_SOURCES`] evicts that reporter's stalest source if
    /// it has gone stale, and is refused otherwise. Served counters, the
    /// served quality EMA and the ledger are untouched. Returns the source's
    /// state after the report.
    pub fn record_external_grade_as(
        &mut self,
        reporter: Option<&str>,
        model_id: &str,
        task: InferenceTask,
        quality: f64,
        n: u32,
        source: &str,
    ) -> Result<ExternalGradeStats, ExternalGradeRefusal> {
        self.record_external_grade_at(reporter, model_id, task, quality, n, source, now_unix())
    }

    /// Clock-injectable core of [`Self::record_external_grade_as`].
    #[allow(clippy::too_many_arguments)]
    fn record_external_grade_at(
        &mut self,
        reporter: Option<&str>,
        model_id: &str,
        task: InferenceTask,
        quality: f64,
        n: u32,
        source: &str,
        now: u64,
    ) -> Result<ExternalGradeStats, ExternalGradeRefusal> {
        if model_id.trim().is_empty() || model_id.len() > 256 {
            return Err(ExternalGradeRefusal::InvalidModel);
        }
        if !quality.is_finite() || !(0.0..=1.0).contains(&quality) {
            return Err(ExternalGradeRefusal::InvalidQuality);
        }
        if n == 0 || n > MAX_EXTERNAL_GRADE_CASES {
            return Err(ExternalGradeRefusal::InvalidCases);
        }
        if source.is_empty()
            || source.len() > MAX_EXTERNAL_GRADE_SOURCE_LEN
            || !source
                .bytes()
                .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b':' | b'-'))
        {
            return Err(ExternalGradeRefusal::InvalidSource);
        }
        if let Some(reporter) = reporter {
            if reporter.is_empty()
                || reporter.len() > MAX_EXTERNAL_GRADE_REPORTER_LEN
                || !reporter.bytes().all(|b| b.is_ascii_graphic() && b != b'/')
            {
                return Err(ExternalGradeRefusal::InvalidReporter);
            }
        }
        let key = external_grade_key(reporter, source);

        // The reporter's own sources other than this one: at the limit, make
        // room by evicting the stalest if it has gone stale, else refuse.
        let mut evict = None;
        if let Some(sources) = self
            .profiles
            .get(model_id)
            .and_then(|p| p.task_stats.get(&task.to_string()))
            .map(|ts| &ts.external_grades)
        {
            if !sources.contains_key(&key) {
                let own: Vec<(&String, &ExternalGradeStats)> = sources
                    .iter()
                    .filter(|(k, _)| external_grade_reporter(k) == reporter)
                    .collect();
                if own.len() >= MAX_EXTERNAL_GRADE_SOURCES {
                    match own.into_iter().min_by_key(|(_, g)| g.updated_at) {
                        Some((stalest, g)) if g.is_stale_at(now) => {
                            evict = Some(stalest.clone());
                        }
                        _ => return Err(ExternalGradeRefusal::SourceLimitReached),
                    }
                }
            }
        }

        let profile = self
            .profiles
            .entry(model_id.to_string())
            .or_insert_with(|| ModelProfile::new(model_id.to_string()));
        let ts = profile.task_stats.entry(task.to_string()).or_default();
        if let Some(stalest) = evict {
            ts.external_grades.remove(&stalest);
        }
        let fresh = ExternalGradeStats {
            quality,
            observations: 0,
            updated_at: now,
        };
        let grade = ts.external_grades.entry(key).or_insert(fresh.clone());
        if grade.is_stale_at(now) {
            // A stale source starts over: its old verdict no longer counts.
            *grade = fresh;
        }
        let n = u64::from(n);
        let observations = grade.observations.saturating_add(n);
        let window = observations.min(EXTERNAL_GRADE_EVIDENCE_CAP);
        let weight = n.min(window) as f64 / window as f64;
        grade.quality = (grade.quality * (1.0 - weight) + quality * weight).clamp(0.0, 1.0);
        grade.observations = observations;
        grade.updated_at = now;
        let stats = grade.clone();
        profile.updated_at = now;
        self.dirty = true;
        Ok(stats)
    }

    /// Infer outcomes from a sequence of action results.
    ///
    /// In a reasoning session, each action's output feeds the next. If action N
    /// produced output and action N+1 succeeded using it, N was implicitly accepted.
    /// If N produced empty output or N+1 failed, N was implicitly rejected.
    ///
    /// Returns (trace_id, inferred_outcome) pairs ready for `resolve_pending_from_signals`.
    pub fn infer_outcomes_from_action_sequence(
        &self,
        action_results: &[(String, bool, f64, String)], // (trace_id, success, confidence, output)
    ) -> Vec<(String, InferredOutcome)> {
        let mut outcomes = Vec::new();

        for (i, (trace_id, success, confidence, output)) in action_results.iter().enumerate() {
            if trace_id.is_empty() {
                continue; // No trace (e.g., memgine-only action)
            }

            if !success {
                outcomes.push((
                    trace_id.clone(),
                    InferredOutcome::Rejected {
                        confidence: *confidence,
                    },
                ));
                continue;
            }

            // Check if the next action used this one's output (implicit acceptance)
            let next_succeeded = action_results
                .get(i + 1)
                .map(|(_, s, _, _)| *s)
                .unwrap_or(true); // Last action: assume accepted if successful

            let has_output = !output.trim().is_empty();

            if has_output && next_succeeded {
                outcomes.push((
                    trace_id.clone(),
                    InferredOutcome::Accepted {
                        confidence: *confidence,
                    },
                ));
            } else if has_output && !next_succeeded {
                // Output existed but downstream failed — may not be this action's fault
                outcomes.push((
                    trace_id.clone(),
                    InferredOutcome::AcceptedWithEdits {
                        confidence: confidence * 0.7,
                    },
                ));
            } else {
                outcomes.push((trace_id.clone(), InferredOutcome::Inconclusive));
            }
        }

        outcomes
    }

    /// Get the profile for a model.
    pub fn profile(&self, model_id: &str) -> Option<&ModelProfile> {
        self.profiles.get(model_id)
    }

    /// Whether a trace is still awaiting an outcome (in `pending`). Lets a
    /// caller observe, before calling `record_inferred_outcome`, whether the
    /// resolution will actually land or silently no-op (trace already resolved
    /// or swept) — so a lost outcome signal is countable, not invisible.
    pub fn has_pending(&self, trace_id: &str) -> bool {
        self.pending.contains_key(trace_id)
    }

    /// Get all profiles.
    pub fn all_profiles(&self) -> &HashMap<String, ModelProfile> {
        &self.profiles
    }

    /// Get pending trace IDs (for conversation signal analysis).
    pub fn pending_trace_ids(&self) -> Vec<String> {
        self.pending.keys().cloned().collect()
    }

    /// Get a pending outcome by trace_id.
    pub fn get_pending(&self, trace_id: &str) -> Option<&InferenceOutcome> {
        self.pending.get(trace_id)
    }

    /// Export profiles for serialization / persistence. Derived metrics
    /// (quality_per_1k_tokens) are recomputed on the way out so callers
    /// always see a consistent snapshot.
    pub fn export_profiles(&self) -> Vec<ModelProfile> {
        self.profiles
            .values()
            .cloned()
            .map(|mut p| {
                p.quality_per_1k_tokens = p.compute_quality_per_1k_tokens();
                p
            })
            .collect()
    }

    /// Import profiles as a genuine mutation (benchmark priors, CLI import
    /// from the memgine fact graph, router merges). Marks the tracker
    /// dirty so the new profiles actually reach disk on the next flush.
    /// NOTE: hydration from disk does NOT go through here — see
    /// [`load_from_file`](crate::outcome::OutcomeTracker::load_from_file), which inserts directly and stays clean (loading
    /// is not a change).
    pub fn import_profiles(&mut self, profiles: Vec<ModelProfile>) {
        for p in profiles {
            self.profiles.insert(p.model_id.clone(), p);
        }
        self.dirty = true;
    }

    /// Save profiles to a JSON file for cross-session persistence (#13).
    ///
    /// Atomic: serialize to a sibling temp file, then rename over the
    /// target. This is the durable receipt store — a torn `write()`
    /// (crash mid-write) would corrupt it, and `load_from_file` treats a
    /// parse failure as a hard error, which on next boot loses *all*
    /// history. The temp+rename makes a partial write impossible.
    pub fn save_to_file(&self, path: &std::path::Path) -> Result<(), std::io::Error> {
        let profiles = self.export_profiles();
        let json = serde_json::to_string_pretty(&profiles).map_err(std::io::Error::other)?;
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let tmp = path.with_extension("json.tmp");
        std::fs::write(&tmp, json)?;
        std::fs::rename(&tmp, path)
    }

    /// True if a persisted profile has changed since the last save.
    pub fn is_dirty(&self) -> bool {
        self.dirty
    }

    /// Save only if dirty, clearing the flag on success. Returns whether
    /// a write happened. Lets callers persist cheaply on a timer without
    /// rewriting the whole file when nothing changed.
    pub fn save_if_dirty(&mut self, path: &std::path::Path) -> Result<bool, std::io::Error> {
        if !self.dirty {
            return Ok(false);
        }
        self.save_to_file(path)?;
        self.dirty = false;
        Ok(true)
    }

    /// Load profiles from a JSON file for cross-session persistence (#13).
    pub fn load_from_file(&mut self, path: &std::path::Path) -> Result<usize, std::io::Error> {
        if !path.exists() {
            return Ok(0);
        }
        let json = std::fs::read_to_string(path)?;
        let profiles: Vec<ModelProfile> = serde_json::from_str(&json)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
        let count = profiles.len();
        // Insert directly (NOT via import_profiles): hydration from disk is
        // not a change, so it must not flip the dirty flag — otherwise the
        // first idle flush would needlessly rewrite the file we just read.
        for mut p in profiles {
            migrate_quality_evidence(&mut p);
            self.profiles.insert(p.model_id.clone(), p);
        }
        Ok(count)
    }

    /// Apply a quality signal to the model's profile.
    fn apply_outcome(
        &mut self,
        pending: &InferenceOutcome,
        quality: Option<f64>,
        success: Option<bool>,
    ) {
        let profile = self
            .profiles
            .entry(pending.model_id.clone())
            .or_insert_with(|| ModelProfile::new(pending.model_id.clone()));

        if let Some(q) = quality {
            profile.ema_quality = profile.ema_quality * (1.0 - EMA_ALPHA) + q * EMA_ALPHA;
            profile.quality_observations += 1;

            let task_key = pending.task.to_string();
            let ts = profile.task_stats.entry(task_key).or_default();
            ts.ema_quality = ts.ema_quality * (1.0 - EMA_ALPHA) + q * EMA_ALPHA;
            ts.quality_observations += 1;
        }

        if let Some(ok) = success {
            // The call may have already been credited a mechanical success at
            // completion (record_complete). Don't double-count it here.
            let already_credited = pending.success_credited;
            let task_key = pending.task.to_string();
            if ok {
                if !already_credited {
                    profile.success_count += 1;
                    let ts = profile.task_stats.entry(task_key).or_default();
                    ts.successes += 1;
                }
                // else: already counted; the quality EMA above is the new info.
            } else {
                // A real downstream signal says this call was bad. If we
                // mechanically credited it as a success at completion,
                // reclassify: undo the success and book a failure instead.
                if already_credited {
                    profile.success_count = profile.success_count.saturating_sub(1);
                    let ts = profile.task_stats.entry(task_key.clone()).or_default();
                    ts.successes = ts.successes.saturating_sub(1);
                }
                profile.fail_count += 1;
                let ts = profile.task_stats.entry(task_key).or_default();
                ts.failures += 1;
            }
        }

        profile.updated_at = now_unix();
        self.dirty = true;

        self.push_ledger(OutcomeLedgerEntry {
            trace_id: pending.trace_id.clone(),
            model_id: pending.model_id.clone(),
            task: pending.task,
            routing_reason: pending.routing_reason.clone(),
            latency_ms: pending.latency_ms,
            input_tokens: pending.input_tokens,
            output_tokens: pending.output_tokens,
            cache_read_input_tokens: pending.cache_read_input_tokens,
            cache_creation_input_tokens: pending.cache_creation_input_tokens,
            success,
            quality,
            error: None,
            project_id: None,
            intent: None,
            attribution: None,
            car_version: Some(env!("CARGO_PKG_VERSION").to_string()),
            timestamp: now_unix(),
        });
    }

    /// Evict `pending` entries older than `ttl_secs` so the map can't grow
    /// without bound in a long-running daemon (only a few resolution paths
    /// ever drain it). For each evicted entry that actually *completed*
    /// (latency recorded), emit a terminal `Inconclusive` receipt — this
    /// also de-biases the ledger: without it, the ledger would only ever
    /// contain calls that happened to get a follow-up signal (a
    /// systematic skew toward reasoning-session traffic), and downstream
    /// stats would be computed over a non-representative sample. Returns
    /// the number of entries swept.
    pub fn sweep_pending(&mut self, ttl_secs: u64) -> usize {
        self.sweep_pending_at(ttl_secs, now_unix())
    }

    /// Clock-injectable core of [`sweep_pending`] for deterministic tests.
    ///
    /// A call that ended is swept `ttl_secs` after it **ended**, so the grace
    /// period a grader has is the same however long the call ran; before
    /// car-releases#110 it counted from the start, and a call longer than the
    /// TTL could never be graded. A call still in flight is swept only after
    /// [`PENDING_IN_FLIGHT_TTL_SECS`] from its start (or `ttl_secs`, if
    /// longer): the bound that keeps a crashed call from living in `pending`
    /// forever, set well past any real generation so a long call is not
    /// evicted before its completion can land.
    fn sweep_pending_at(&mut self, ttl_secs: u64, now: u64) -> usize {
        let ended_cutoff = now.saturating_sub(ttl_secs);
        let in_flight_cutoff = now.saturating_sub(ttl_secs.max(PENDING_IN_FLIGHT_TTL_SECS));
        let expired: Vec<String> = self
            .pending
            .iter()
            .filter(|(_, o)| match o.completed_at {
                Some(ended) => ended < ended_cutoff,
                None => o.timestamp < in_flight_cutoff,
            })
            .map(|(id, _)| id.clone())
            .collect();
        for id in &expired {
            if let Some(o) = self.pending.remove(id) {
                // Only completed-but-unresolved calls become receipts; a
                // never-completed trace (latency 0) is in-flight/crashed —
                // a zero-everything receipt would be noise.
                if o.latency_ms > 0 {
                    // A swept call completed but never received a downstream
                    // quality signal. Recording *every* such call Inconclusive
                    // (`success: None`) left signal-less models — notably ALL
                    // local inference, which has no accept/edit feedback loop —
                    // pinned at the 0.5 EMA prior with zero successes, so the
                    // health UI rendered working models as "50%"/"0%" (#312).
                    // Instead, credit a *mechanical* success when the call
                    // actually returned output with no error: it ran and
                    // produced tokens; we simply never learned whether the
                    // answer was *good*. `quality` stays None so `ema_quality`
                    // is only ever moved by a real quality signal. A completion
                    // that produced no output stays Inconclusive (ambiguous).
                    // Credit a mechanical success here ONLY if it wasn't
                    // already booked at completion (record_complete now does
                    // this immediately for the common case). A call credited
                    // at completion still gets a `Some(true)` receipt below,
                    // but its count is not incremented twice.
                    let credit_now =
                        o.output_tokens > 0 && o.error.is_none() && !o.success_credited;
                    let was_success = o.success_credited || credit_now;
                    if credit_now {
                        let profile = self
                            .profiles
                            .entry(o.model_id.clone())
                            .or_insert_with(|| ModelProfile::new(o.model_id.clone()));
                        profile.success_count += 1;
                        let ts = profile.task_stats.entry(o.task.to_string()).or_default();
                        ts.successes += 1;
                        profile.updated_at = now_unix();
                        self.dirty = true;
                    }
                    self.push_ledger(OutcomeLedgerEntry {
                        trace_id: o.trace_id,
                        model_id: o.model_id,
                        task: o.task,
                        routing_reason: o.routing_reason,
                        latency_ms: o.latency_ms,
                        input_tokens: o.input_tokens,
                        output_tokens: o.output_tokens,
                        cache_read_input_tokens: o.cache_read_input_tokens,
                        cache_creation_input_tokens: o.cache_creation_input_tokens,
                        success: if was_success { Some(true) } else { None },
                        quality: None,
                        error: None,
                        project_id: None,
                        intent: None,
                        attribution: None,
                        car_version: Some(env!("CARGO_PKG_VERSION").to_string()),
                        timestamp: now_unix(),
                    });
                }
            }
        }
        expired.len()
    }

    /// Check git diff for pending code suggestions and resolve outcomes.
    ///
    /// Two strategies:
    /// 1. **AST structural diff** (when `ast` feature is enabled): parse the old
    ///    and new versions of changed files and compare at the symbol level.
    ///    This gives precise outcomes: SignatureChanged, BodyModified, SymbolAdded.
    /// 2. **Text diff fallback**: token matching against the combined git diff.
    pub fn check_git_outcomes(&mut self, repo_dir: &std::path::Path) {
        let diff = match std::process::Command::new("git")
            .args(["diff", "--no-color"])
            .current_dir(repo_dir)
            .output()
        {
            Ok(output) => String::from_utf8_lossy(&output.stdout).to_string(),
            Err(_) => return,
        };

        let staged_diff = match std::process::Command::new("git")
            .args(["diff", "--cached", "--no-color"])
            .current_dir(repo_dir)
            .output()
        {
            Ok(output) => String::from_utf8_lossy(&output.stdout).to_string(),
            Err(_) => String::new(),
        };

        let combined_diff = format!("{}\n{}", diff, staged_diff);

        if combined_diff.trim().is_empty() {
            return; // No changes at all
        }

        // Try AST structural diff on changed files
        #[cfg(feature = "ast")]
        let ast_outcome = Self::check_git_outcomes_ast(repo_dir);

        let code_traces: Vec<(String, String)> = self
            .pending
            .iter()
            .filter(|(_, o)| matches!(o.task, InferenceTask::Code))
            .map(|(id, o)| (id.clone(), o.model_id.clone()))
            .collect();

        for (trace_id, _model_id) in code_traces {
            if let Some(pending) = self.pending.get(&trace_id) {
                // Try AST-based outcome first
                #[cfg(feature = "ast")]
                if let Some(ref ast_out) = ast_outcome {
                    let pending_clone = pending.clone();
                    self.apply_outcome(
                        &pending_clone,
                        Some(ast_out.quality_score()),
                        Some(ast_out.is_success()),
                    );
                    continue;
                }

                // Fallback: text token matching
                let output_tokens: Vec<&str> = pending
                    .routing_reason
                    .split_whitespace()
                    .filter(|t| t.len() > 5)
                    .collect();

                let outcome = if output_tokens.iter().any(|t| combined_diff.contains(t)) {
                    CodeOutcome::Applied
                } else {
                    CodeOutcome::Modified
                };

                let pending_clone = pending.clone();
                self.apply_outcome(
                    &pending_clone,
                    Some(outcome.quality_score()),
                    Some(outcome.is_success()),
                );
            }
        }
    }

    /// AST-based git outcome: parse changed files before and after, diff symbols.
    #[cfg(feature = "ast")]
    fn check_git_outcomes_ast(repo_dir: &std::path::Path) -> Option<CodeOutcome> {
        // Get list of changed files
        let name_only = std::process::Command::new("git")
            .args(["diff", "--name-only"])
            .current_dir(repo_dir)
            .output()
            .ok()?;
        let changed_files: Vec<&str> = std::str::from_utf8(&name_only.stdout)
            .ok()?
            .lines()
            .filter(|f| !f.is_empty())
            .collect();

        if changed_files.is_empty() {
            return None;
        }

        let mut has_sig_change = false;
        let mut has_body_change = false;
        let mut has_addition = false;

        for file in &changed_files {
            // Only parse files tree-sitter supports
            if car_ast::Language::from_filename(file).is_none() {
                continue;
            }

            // Get the HEAD version
            let old_content = std::process::Command::new("git")
                .args(["show", &format!("HEAD:{}", file)])
                .current_dir(repo_dir)
                .output()
                .ok()
                .and_then(|o| {
                    if o.status.success() {
                        String::from_utf8(o.stdout).ok()
                    } else {
                        None
                    }
                });

            // Get the working tree version
            let new_path = repo_dir.join(file);
            let new_content = std::fs::read_to_string(&new_path).ok();

            match (old_content, new_content) {
                (Some(old), Some(new)) => {
                    let old_parsed = car_ast::parse_file(&old, file);
                    let new_parsed = car_ast::parse_file(&new, file);

                    if let (Some(old_p), Some(new_p)) = (old_parsed, new_parsed) {
                        let changes = car_ast::diff_symbols(&old_p, &new_p);
                        for change in &changes {
                            match change {
                                car_ast::SymbolChange::Added(_) => has_addition = true,
                                car_ast::SymbolChange::Modified {
                                    signature_changed, ..
                                } => {
                                    if *signature_changed {
                                        has_sig_change = true;
                                    } else {
                                        has_body_change = true;
                                    }
                                }
                                car_ast::SymbolChange::Removed(_) => has_sig_change = true,
                            }
                        }
                    }
                }
                (None, Some(_)) => has_addition = true, // New file
                _ => {}
            }
        }

        // Return the most significant outcome
        if has_sig_change {
            Some(CodeOutcome::SignatureChanged)
        } else if has_body_change {
            Some(CodeOutcome::BodyModified)
        } else if has_addition {
            Some(CodeOutcome::SymbolAdded)
        } else {
            None // No structural changes detected (maybe non-code files changed)
        }
    }
}

impl Default for OutcomeTracker {
    fn default() -> Self {
        Self::new()
    }
}

pub(crate) fn now_unix() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

#[cfg(test)]
mod quality_evidence_tests {
    use super::*;

    /// A generation failure is availability evidence only: it is counted, and
    /// answer quality stays where it was.
    #[test]
    fn a_failure_counts_against_availability_not_quality() {
        let mut tracker = OutcomeTracker::new();
        let trace = tracker.record_start("m", InferenceTask::Code, "write a parser");
        tracker.record_failure(&trace, "model returned malformed tool call JSON");
        let p = tracker.profile("m").unwrap();
        assert_eq!(p.fail_count, 1);
        assert_eq!(
            (p.ema_quality, p.quality_observations),
            (NEUTRAL_QUALITY, 0)
        );
        let ts = p.task_stats(InferenceTask::Code).unwrap();
        assert_eq!(ts.failures, 1);
        assert_eq!(
            (ts.ema_quality, ts.quality_observations),
            (NEUTRAL_QUALITY, 0)
        );
    }

    /// A task's first grade moves quality from neutral, as the profile's does.
    #[test]
    fn a_new_task_starts_at_neutral_quality() {
        assert_eq!(TaskStats::default().ema_quality, NEUTRAL_QUALITY);
    }

    /// A profile as today's code writes it (no `quality_evidence_version`)
    /// after a run of graded failures: its quality evidence is dropped on load,
    /// its availability counters are kept.
    #[test]
    fn a_version_zero_profile_loads_without_its_failure_derived_quality() {
        let mut tracker = OutcomeTracker::new();
        for _ in 0..5 {
            let trace = tracker.record_start("m", InferenceTask::Code, "x");
            tracker.record_failure(&trace, "generation failed");
        }
        let mut saved = serde_json::to_value(tracker.profile("m").unwrap()).unwrap();
        // What a version-0 writer stored after those five failures.
        saved["ema_quality"] = serde_json::json!(0.12);
        saved["quality_observations"] = serde_json::json!(5);
        saved["task_stats"]["code"]["ema_quality"] = serde_json::json!(0.07);
        saved["task_stats"]["code"]["quality_observations"] = serde_json::json!(5);
        saved
            .as_object_mut()
            .unwrap()
            .remove("quality_evidence_version");

        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("outcome_profiles.json");
        std::fs::write(&path, serde_json::to_string(&vec![saved]).unwrap()).unwrap();
        let mut loaded = OutcomeTracker::new();
        loaded.load_from_file(&path).unwrap();

        let p = loaded.profile("m").unwrap();
        assert_eq!(p.fail_count, 5, "availability survives");
        assert_eq!(
            (p.ema_quality, p.quality_observations),
            (NEUTRAL_QUALITY, 0)
        );
        let ts = p.task_stats(InferenceTask::Code).unwrap();
        assert_eq!(ts.failures, 5);
        assert_eq!(
            (ts.ema_quality, ts.quality_observations),
            (NEUTRAL_QUALITY, 0)
        );
        assert_eq!(p.quality_evidence_version, QUALITY_EVIDENCE_VERSION);
    }
}

#[cfg(test)]
mod tests {
    /// A worker's call teaches the router nothing: inside
    /// `scope_learning(false, ..)` its start, completion and failure change no
    /// model profile. The same calls outside the scope do.
    #[tokio::test]
    async fn an_unlearned_call_changes_no_profile() {
        let mut tracker = OutcomeTracker::new();
        super::scope_learning(false, async {
            let trace = tracker.record_start("m", InferenceTask::Generate, "pinned");
            tracker.record_complete(&trace, 10, 5, 5);
            let failed = tracker.record_start("m", InferenceTask::Generate, "pinned");
            tracker.record_failure(&failed, "boom");
        })
        .await;
        assert!(tracker.profile("m").is_none());
        let trace = tracker.record_start("m", InferenceTask::Generate, "pinned");
        tracker.record_complete(&trace, 10, 5, 5);
        assert_eq!(
            tracker.profile("m").map(|profile| profile.total_calls),
            Some(1)
        );
    }

    use super::*;

    #[test]
    fn only_a_failure_this_car_recorded_counts_against_a_model() {
        let entry = |success, version: Option<&str>| OutcomeLedgerEntry {
            trace_id: "t".into(),
            model_id: "m".into(),
            task: InferenceTask::Generate,
            routing_reason: String::new(),
            latency_ms: 0,
            input_tokens: 0,
            output_tokens: 0,
            cache_read_input_tokens: 0,
            cache_creation_input_tokens: 0,
            success: Some(success),
            quality: None,
            error: Some("config.json has no vocab_size".into()),
            project_id: None,
            intent: None,
            attribution: None,
            car_version: version.map(str::to_string),
            timestamp: 1,
        };
        let here = env!("CARGO_PKG_VERSION");
        assert!(entry(false, Some(here)).counts_against_model());
        // An older build's failure may be that build's bug.
        assert!(!entry(false, Some("0.0.1")).counts_against_model());
        assert!(!entry(false, None).counts_against_model());
        assert!(!entry(true, Some(here)).counts_against_model());
    }

    #[test]
    fn lifecycle() {
        let mut tracker = OutcomeTracker::new();

        // Start an inference call
        let trace = tracker.record_start(
            "qwen/qwen3-4b:q4_k_m",
            InferenceTask::Code,
            "Code task -> Qwen3-4B",
        );

        // Complete it
        tracker.record_complete(&trace, 1200, 100, 50);

        // Profile should have 1 call
        let profile = tracker.profile("qwen/qwen3-4b:q4_k_m").unwrap();
        assert_eq!(profile.total_calls, 1);
        assert_eq!(profile.avg_latency_ms(), 1200.0);

        // Record positive outcome
        tracker.record_inferred_outcome(&trace, InferredOutcome::Accepted { confidence: 0.9 });

        let profile = tracker.profile("qwen/qwen3-4b:q4_k_m").unwrap();
        assert_eq!(profile.success_count, 1);
        assert!(profile.ema_quality > 0.5); // should have gone up from 0.5
    }

    #[test]
    fn failure_degrades() {
        // A failed call goes through record_failure ONLY — it never also
        // hits record_complete (a call either completes or errors, not both).
        // record_complete now credits a mechanical success, so mixing the two
        // on one trace would model a flow that does not occur.
        let mut tracker = OutcomeTracker::new();
        for _ in 0..5 {
            let trace = tracker.record_start("bad-model", InferenceTask::Generate, "test");
            // A *generation* failure (not a transport/timeout error, which is
            // quality-neutral — see `transport_failures_are_quality_neutral`).
            tracker.record_failure(&trace, "model produced malformed output");
        }

        let profile = tracker.profile("bad-model").unwrap();
        assert_eq!(profile.fail_count, 5);
        assert_eq!(profile.success_count, 0);
        assert!(profile.should_degrade(2)); // 5 > 0 + 2
                                            // Availability degrades; answer quality is not graded by a failure
                                            // (QUALITY_EVIDENCE_VERSION).
        assert_eq!(profile.ema_quality, NEUTRAL_QUALITY);
    }

    #[test]
    fn code_outcome_ground_truth() {
        let mut tracker = OutcomeTracker::new();

        let trace = tracker.record_start("qwen/qwen3-4b:q4_k_m", InferenceTask::Code, "code");
        tracker.record_complete(&trace, 500, 200, 100);
        tracker.record_code_outcome(&trace, CodeOutcome::Applied);

        let profile = tracker.profile("qwen/qwen3-4b:q4_k_m").unwrap();
        assert_eq!(profile.success_count, 1);
        // EMA should reflect Applied quality (1.0): 0.5 * 0.8 + 1.0 * 0.2 = 0.6
        assert!((profile.ema_quality - 0.6).abs() < 0.01);
    }

    #[test]
    fn per_task_stats() {
        let mut tracker = OutcomeTracker::new();

        // Two code calls, one generate call
        for _ in 0..2 {
            let trace = tracker.record_start("m1", InferenceTask::Code, "code");
            tracker.record_complete(&trace, 1000, 100, 50);
            tracker.record_inferred_outcome(&trace, InferredOutcome::Accepted { confidence: 0.8 });
        }
        let trace = tracker.record_start("m1", InferenceTask::Generate, "gen");
        tracker.record_complete(&trace, 500, 50, 25);
        tracker.record_inferred_outcome(&trace, InferredOutcome::Rejected { confidence: 0.9 });

        let profile = tracker.profile("m1").unwrap();
        assert_eq!(profile.total_calls, 3);

        let code_stats = profile.task_stats(InferenceTask::Code).unwrap();
        assert_eq!(code_stats.calls, 2);
        assert_eq!(code_stats.successes, 2);

        let gen_stats = profile.task_stats(InferenceTask::Generate).unwrap();
        assert_eq!(gen_stats.calls, 1);
        assert_eq!(gen_stats.failures, 1);
    }

    #[test]
    fn export_populates_quality_per_1k_tokens() {
        let mut tracker = OutcomeTracker::new();
        let trace = tracker.record_start("m1", InferenceTask::Generate, "test");
        tracker.record_complete(&trace, 100, 800, 200); // 1000 tokens total
        tracker.record_inferred_outcome(&trace, InferredOutcome::Accepted { confidence: 1.0 });

        let exported = tracker.export_profiles();
        assert_eq!(exported.len(), 1);
        let p = &exported[0];
        // ema_quality after one Accepted{1.0}: 0.5 * 0.8 + 1.0 * 0.2 = 0.6
        // quality_per_1k = 0.6 * 1000 / 1000 = 0.6
        assert!(
            (p.quality_per_1k_tokens - 0.6).abs() < 1e-6,
            "got {}",
            p.quality_per_1k_tokens
        );
    }

    #[test]
    fn quality_per_1k_tokens_zero_without_tokens() {
        let profile = ModelProfile::new("x".into());
        assert_eq!(profile.compute_quality_per_1k_tokens(), 0.0);
    }

    #[test]
    fn tokens_per_success_is_outcome_denominated() {
        let mut p = ModelProfile::new("x".into());
        // No success yet → None (not a fabricated number).
        assert_eq!(p.tokens_per_success(), None);
        // 900 total tokens over 3 successful outcomes → 300 tokens/success.
        p.total_input_tokens = 600;
        p.total_output_tokens = 300;
        p.success_count = 3;
        assert_eq!(p.tokens_per_success(), Some(300.0));
        // Dollar cost-per-success: (600*$1 + 300*$2)/1M / 3 successes.
        let usd = p.usd_per_success(1.0, 2.0, CacheRates::ANTHROPIC).unwrap();
        assert!((usd - ((600.0 * 1.0 + 300.0 * 2.0) / 1_000_000.0 / 3.0)).abs() < 1e-12);
        assert_eq!(
            ModelProfile::new("y".into()).usd_per_success(1.0, 2.0, CacheRates::ANTHROPIC),
            None
        );
    }

    #[test]
    fn usd_per_success_prices_cache_buckets_separately() {
        let mut p = ModelProfile::new("cached".into());
        p.success_count = 1;
        p.total_input_tokens = 100; // uncached prefix @ 1.0×
        p.total_cache_read_input_tokens = 1000; // hit @ 0.1×
        p.total_cache_creation_input_tokens = 200; // write @ 1.25×
        p.total_output_tokens = 50; // output @ 2.0×
                                    // Each input bucket priced at its own multiplier off the $1/Mtok base.
        let expected_input = (100.0 * 1.0 + 1000.0 * 1.0 * 0.1 + 200.0 * 1.0 * 1.25) / 1_000_000.0;
        let expected = expected_input + 50.0 * 2.0 / 1_000_000.0;
        let usd = p.usd_per_success(1.0, 2.0, CacheRates::ANTHROPIC).unwrap();
        assert!((usd - expected).abs() < 1e-12, "got {usd}, want {expected}");
        // Token efficiency counts all real tokens, cache included.
        assert_eq!(p.total_tokens(), 100 + 1000 + 200 + 50);
        // A cache hit must be ~10× cheaper than counting it as full input,
        // which is the whole point of fixing the accounting.
        let mut naive = ModelProfile::new("naive".into());
        naive.success_count = 1;
        naive.total_input_tokens = 1300; // same tokens, all priced at 1.0×
        naive.total_output_tokens = 50;
        assert!(
            naive
                .usd_per_success(1.0, 2.0, CacheRates::ANTHROPIC)
                .unwrap()
                > usd
        );
    }

    #[test]
    fn openai_cache_rates_price_reads_at_half_no_write_premium() {
        // OpenAI economics: cached reads at 0.5×, no write bucket. Same token
        // counts as Anthropic must cost MORE under OpenAI (0.5× vs 0.1× reads).
        let mut p = ModelProfile::new("gpt".into());
        p.success_count = 1;
        p.total_input_tokens = 100; // uncached @ 1.0×
        p.total_cache_read_input_tokens = 1000; // OpenAI hit @ 0.5×
        p.total_cache_creation_input_tokens = 0; // OpenAI has no write bucket
        let usd = p.usd_per_success(1.0, 2.0, CacheRates::OPENAI).unwrap();
        let expected = (100.0 * 1.0 + 1000.0 * 1.0 * 0.5) / 1_000_000.0;
        assert!((usd - expected).abs() < 1e-12, "got {usd}, want {expected}");
        // Same buckets under Anthropic rates (0.1× read) are cheaper.
        let anthropic = p.usd_per_success(1.0, 2.0, CacheRates::ANTHROPIC).unwrap();
        assert!(
            usd > anthropic,
            "OpenAI 0.5× read must exceed Anthropic 0.1×"
        );
        // And both beat pricing the cache hit as full uncached input (1.0×).
        let mut naive = ModelProfile::new("naive".into());
        naive.success_count = 1;
        naive.total_input_tokens = 1100;
        assert!(naive.usd_per_success(1.0, 2.0, CacheRates::OPENAI).unwrap() > usd);
    }

    #[test]
    fn record_complete_cached_accumulates_cache_totals() {
        let mut tracker = OutcomeTracker::new();
        let trace = tracker.record_start("m", InferenceTask::Generate, "gen");
        tracker.record_complete_cached(&trace, 100, 40, 20, 800, 120);
        let p = tracker.profile("m").unwrap();
        assert_eq!(p.total_input_tokens, 40, "uncached prefix only");
        assert_eq!(p.total_cache_read_input_tokens, 800);
        assert_eq!(p.total_cache_creation_input_tokens, 120);
    }

    #[test]
    fn dirty_flag_and_save_if_dirty() {
        let dir = std::env::temp_dir().join("car-outcome-dirty-test");
        let _ = std::fs::remove_dir_all(&dir);
        let path = dir.join("outcome_profiles.json");

        let mut tracker = OutcomeTracker::new();
        // Fresh tracker is clean → save_if_dirty is a no-op.
        assert!(!tracker.is_dirty());
        assert!(!tracker.save_if_dirty(&path).unwrap());
        assert!(!path.exists());

        // A recorded outcome dirties it.
        let trace = tracker.record_start("m1", InferenceTask::Generate, "router");
        tracker.record_complete(&trace, 100, 10, 20);
        assert!(tracker.is_dirty());

        // save_if_dirty writes once and clears the flag.
        assert!(tracker.save_if_dirty(&path).unwrap());
        assert!(path.exists());
        assert!(!tracker.is_dirty());

        // Second call with no new changes does not rewrite.
        assert!(!tracker.save_if_dirty(&path).unwrap());

        // Loading profiles must NOT mark the tracker dirty.
        let mut fresh = OutcomeTracker::new();
        fresh.load_from_file(&path).unwrap();
        assert!(!fresh.is_dirty());

        // But a genuine import (benchmark priors / CLI) MUST dirty, so the
        // imported profiles actually persist on the next flush.
        fresh.import_profiles(vec![ModelProfile::new("seeded".into())]);
        assert!(fresh.is_dirty());

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn ledger_captures_resolved_outcomes() {
        let dir = std::env::temp_dir().join("car-outcome-ledger-test");
        let _ = std::fs::remove_dir_all(&dir);
        let path = dir.join("outcome_ledger.jsonl");

        let mut tracker = OutcomeTracker::new();

        // A success with a quality signal → one resolved receipt.
        let t1 = tracker.record_start("good-model", InferenceTask::Generate, "router:test");
        tracker.record_complete(&t1, 1200, 50, 100);
        tracker.record_inferred_outcome(&t1, InferredOutcome::Accepted { confidence: 0.9 });

        // A failure → one resolved receipt with the error.
        let t2 = tracker.record_start("bad-model", InferenceTask::Code, "router:test");
        tracker.record_failure(&t2, "boom: 500");

        let drained = tracker.drain_ledger();
        assert_eq!(drained.len(), 2);
        assert!(tracker.drain_ledger().is_empty(), "drain clears the buffer");

        append_ledger_entries(&path, &drained).unwrap();
        let read = read_ledger(&path, 0);
        assert_eq!(read.len(), 2);

        let good = read.iter().find(|e| e.model_id == "good-model").unwrap();
        assert_eq!(good.success, Some(true));
        assert!(good.quality.is_some());
        assert_eq!(good.routing_reason, "router:test");
        assert_eq!(good.latency_ms, 1200);

        let bad = read.iter().find(|e| e.model_id == "bad-model").unwrap();
        assert_eq!(bad.success, Some(false));
        assert_eq!(bad.error.as_deref(), Some("boom: 500"));

        // Read back limited to most recent 1.
        assert_eq!(read_ledger(&path, 1).len(), 1);

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn ledger_redacts_long_errors_and_prunes() {
        let dir = std::env::temp_dir().join("car-outcome-privacy-test");
        let _ = std::fs::remove_dir_all(&dir);
        let path = dir.join("outcome_ledger.jsonl");

        // Long error is truncated on capture.
        let mut tracker = OutcomeTracker::new();
        let t = tracker.record_start("m", InferenceTask::Generate, "r");
        let huge = "x".repeat(5000);
        tracker.record_failure(&t, &huge);
        let drained = tracker.drain_ledger();
        let err = drained[0].error.as_ref().unwrap();
        assert!(
            err.chars().count() <= MAX_LEDGER_ERROR_CHARS + 1,
            "error truncated"
        );

        // Pruning keeps only the most recent N.
        let entries: Vec<OutcomeLedgerEntry> = (0..10)
            .map(|i| OutcomeLedgerEntry {
                trace_id: format!("t{i}"),
                model_id: "m".into(),
                task: InferenceTask::Generate,
                routing_reason: "r".into(),
                latency_ms: 1,
                input_tokens: 1,
                output_tokens: 1,
                cache_read_input_tokens: 0,
                cache_creation_input_tokens: 0,
                success: Some(true),
                quality: Some(1.0),
                error: None,
                project_id: None,
                intent: None,
                attribution: None,
                car_version: None,
                timestamp: i,
            })
            .collect();
        append_ledger_entries(&path, &entries).unwrap();
        prune_ledger(&path, 3).unwrap();
        let kept = read_ledger(&path, 0);
        assert_eq!(kept.len(), 3);
        assert_eq!(kept[0].trace_id, "t7"); // most recent 3: t7,t8,t9
        assert_eq!(kept[2].trace_id, "t9");

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn sweep_pending_credits_mechanical_success_or_inconclusive() {
        let mut tracker = OutcomeTracker::new();

        // Completed with output, no signal -> mechanical success (#312).
        let t1 = tracker.record_start("m", InferenceTask::Generate, "r");
        tracker.record_complete(&t1, 500, 10, 20);
        // Completed but produced NO output -> stays Inconclusive.
        let t2 = tracker.record_start("m", InferenceTask::Generate, "r");
        tracker.record_complete(&t2, 300, 5, 0);
        // A never-completed call (in-flight / crashed) -> no receipt.
        let _t3 = tracker.record_start("m", InferenceTask::Generate, "r");

        // Evaluate the sweep against a clock 10s in the future so the
        // same-second pending entries are unambiguously past the cutoff.
        let swept = tracker.sweep_pending_at(0, now_unix() + 10);
        assert_eq!(
            swept, 2,
            "both ended calls evicted; the in-flight one waits"
        );
        // The in-flight call is evicted only past its own, longer bound.
        let swept = tracker.sweep_pending_at(0, now_unix() + PENDING_IN_FLIGHT_TTL_SECS + 10);
        assert_eq!(swept, 1, "a call that never ended is evicted eventually");

        // Only the two completed calls become receipts.
        let mut receipts = tracker.drain_ledger();
        receipts.sort_by_key(|r| r.latency_ms);
        assert_eq!(receipts.len(), 2);
        // 300ms, no output -> Inconclusive
        assert_eq!(receipts[0].latency_ms, 300);
        assert_eq!(receipts[0].success, None);
        // 500ms, produced output -> mechanical success, quality untouched
        assert_eq!(receipts[1].latency_ms, 500);
        assert_eq!(receipts[1].success, Some(true));
        assert_eq!(receipts[1].quality, None);

        // success_count is credited; ema_quality stays the neutral prior
        // (no real quality signal ever arrived).
        let p = tracker.profile("m").expect("profile exists");
        assert_eq!(p.success_count, 1);
        assert_eq!(p.ema_quality, 0.5);
    }

    #[test]
    fn record_complete_credits_success_immediately() {
        // The headline fix: a call that completes with output is credited a
        // success at completion, not deferred to the 300s sweep — so a
        // short-lived process (CLI) that exits before any sweep still records
        // the success. Previously success_count stayed 0 and health read 0.5.
        let mut tracker = OutcomeTracker::new();
        let t = tracker.record_start("m", InferenceTask::Generate, "r");
        tracker.record_complete(&t, 500, 12, 20);

        let p = tracker.profile("m").expect("profile exists");
        assert_eq!(p.success_count, 1, "success credited at completion");
        assert_eq!(p.total_calls, 1);
        assert_eq!(p.total_input_tokens, 12, "input tokens recorded, not 0");
        assert_eq!(p.fail_count, 0);

        // A later sweep of the same (still-pending) call must NOT double-count.
        tracker.sweep_pending_at(0, now_unix() + 10);
        let p = tracker.profile("m").unwrap();
        assert_eq!(p.success_count, 1, "sweep does not re-credit");
    }

    #[test]
    fn success_rate_resolved_distinguishes_no_signal_from_measured() {
        // Display surfaces must not show a never-measured model as a confident
        // "50%": `success_rate_resolved()` returns None until something
        // resolves, then the real ratio. (The router never derives a rate
        // here — it reads raw success/fail counts directly.)
        let mut p = ModelProfile::new("m".to_string());
        assert_eq!(
            p.success_rate_resolved(),
            None,
            "no resolved signal, not a fabricated 0.5"
        );

        p.success_count = 3;
        p.fail_count = 1;
        assert_eq!(
            p.success_rate_resolved(),
            Some(0.75),
            "real rate once resolved"
        );
    }

    #[test]
    fn quality_observations_count_only_graded_signals() {
        // #3: a mechanical success must NOT count as a graded quality
        // observation (it leaves answer quality unknown), but an accept/reject
        // signal and a failure must. The router keys reliability on this count.
        let mut tracker = OutcomeTracker::new();

        // Mechanical success — credited, but quality stays unknown.
        let t = tracker.record_start("m", InferenceTask::Generate, "r");
        tracker.record_complete(&t, 500, 12, 20);
        let p = tracker.profile("m").unwrap();
        assert_eq!(p.success_count, 1);
        assert_eq!(
            p.quality_observations, 0,
            "mechanical success is not a graded quality observation"
        );
        assert!(
            (p.ema_quality - 0.5).abs() < 1e-9,
            "EMA untouched by mechanical success"
        );

        // A real accept signal — graded.
        tracker.record_inferred_outcome(&t, InferredOutcome::Accepted { confidence: 0.9 });
        let p = tracker.profile("m").unwrap();
        assert_eq!(p.quality_observations, 1, "graded accept signal counts");
        assert!(p.ema_quality > 0.5, "graded accept moved the EMA up");
        assert_eq!(
            p.task_stats(InferenceTask::Generate)
                .unwrap()
                .quality_observations,
            1,
            "per-task graded count tracked too"
        );

        // A failure is availability evidence, not a grade
        // (QUALITY_EVIDENCE_VERSION).
        let t2 = tracker.record_start("m", InferenceTask::Generate, "r");
        tracker.record_failure(&t2, "boom");
        let p = tracker.profile("m").unwrap();
        assert_eq!(
            p.quality_observations, 1,
            "a failure is not a graded observation"
        );
    }

    #[test]
    fn record_complete_no_output_is_not_a_success() {
        let mut tracker = OutcomeTracker::new();
        let t = tracker.record_start("m", InferenceTask::Generate, "r");
        tracker.record_complete(&t, 300, 5, 0); // no output
        let p = tracker.profile("m").unwrap();
        assert_eq!(p.success_count, 0, "no output -> no mechanical success");
        assert_eq!(p.total_calls, 1);
    }

    #[test]
    fn record_failure_counts_total_calls() {
        // A failed call is still a call. total_calls must include it so it can
        // never read smaller than fail_count (the parslee/* 6-calls/14-fails
        // symptom).
        let mut tracker = OutcomeTracker::new();
        for _ in 0..3 {
            let t = tracker.record_start("m", InferenceTask::Generate, "r");
            tracker.record_failure(&t, "boom 500");
        }
        let p = tracker.profile("m").unwrap();
        assert_eq!(p.fail_count, 3);
        assert_eq!(p.total_calls, 3, "failures counted in total_calls");
        assert!(p.fail_count <= p.total_calls);
    }

    #[test]
    fn capability_rejection_does_not_change_model_profile() {
        let mut tracker = OutcomeTracker::new();
        let trace = tracker.record_start("m", InferenceTask::Generate, "r");

        tracker.record_capability_rejection(&trace, "unsupported mode: json schema");

        assert!(tracker.profile("m").is_none());
        assert!(!tracker.has_pending(&trace));
        assert_eq!(tracker.ledger_buffer.len(), 1);
    }

    #[test]
    fn transport_failures_are_quality_neutral() {
        // A read-timeout / connection error means the model never produced an
        // answer — its answer quality is UNKNOWN, so transport failures must not
        // move the quality EMA or count as graded quality evidence. (Observed
        // live: parslee/* collapsed to ema_quality ≈ 0.003 after ~23 daemon
        // read-timeouts, which buried it under every workload.) They DO still
        // count for availability (total_calls / fail_count / circuit breaker).
        let mut tracker = OutcomeTracker::new();
        // One genuine generation failure establishes the baseline EMA + a
        // graded quality observation.
        let t = tracker.record_start("m", InferenceTask::Code, "r");
        tracker.record_failure(&t, "model produced malformed output");
        let baseline_ema = tracker.profile("m").unwrap().ema_quality;
        let baseline_qobs = tracker.profile("m").unwrap().quality_observations;

        // Hammer with transport timeouts AND auth/client rejections — neither
        // produced an answer, so quality must not budge. The 401 case is the
        // live parslee/* symptom: ~47 "HTTP 401 Unauthorized" from an undeployed
        // gateway must not bury the model on *answer* quality.
        for _ in 0..5 {
            let t = tracker.record_start("m", InferenceTask::Code, "r");
            tracker.record_failure(&t, "daemon read timeout on infer after 30s");
        }
        for _ in 0..5 {
            let t = tracker.record_start("m", InferenceTask::Code, "r");
            tracker.record_failure(
                &t,
                "inference failed: Parslee org lookup failed: HTTP 401 Unauthorized: Authentication",
            );
        }

        let p = tracker.profile("m").unwrap();
        assert_eq!(
            p.fail_count, 11,
            "no-answer failures still counted for availability"
        );
        assert_eq!(p.total_calls, 11, "every call counted (failures included)");
        assert!(
            (p.ema_quality - baseline_ema).abs() < 1e-9,
            "timeouts AND auth rejections must not move the quality EMA ({baseline_ema} -> {})",
            p.ema_quality
        );
        assert_eq!(
            p.quality_observations, baseline_qobs,
            "no-answer failures (transport + auth) are not graded quality evidence"
        );
        // A generation failure is availability evidence too: it is counted
        // and leaves quality alone (QUALITY_EVIDENCE_VERSION).
        let t = tracker.record_start("m", InferenceTask::Code, "r");
        tracker.record_failure(&t, "model produced malformed output");
        let p = tracker.profile("m").unwrap();
        assert_eq!(p.fail_count, 12);
        assert_eq!(p.ema_quality, baseline_ema);
    }

    #[test]
    fn is_no_answer_failure_precision_boundary() {
        // The classifier gates whether a failure counts as answer-quality, so
        // pin both sides of the precision boundary it promises. Matches are on
        // the FAILURE error string only (never model output).
        // No-answer: transport/infra + auth/client rejection → quality-neutral.
        for s in [
            "daemon read timeout on infer after 30s",
            "connection reset by peer",
            "upstream returned 503 service unavailable",
            "HTTP 500 Internal Server Error",
            "stream closed unexpectedly (eof)",
            "HTTP 401 Unauthorized",
            "unauthenticated request",
            "HTTP 403 Forbidden",
            "permission denied by gateway",
            "Parslee org lookup failed: Authentication required",
            "oauth: invalid_client",
            "invalid api key",
            "HTTP 400 Bad Request: max context length exceeded",
        ] {
            assert!(
                is_no_answer_failure(s),
                "expected no-answer (quality-neutral): {s:?}"
            );
        }
        // Generation failures: the model produced output that was bad/unparseable.
        // These MUST stay generation failures (graded), NOT be swallowed as
        // no-answer — including the deliberately-excluded ambiguous phrases.
        for s in [
            "model produced malformed output",
            "field 'user' not found in model response", // not "not found" → no-answer
            "tokenizer: invalid token id 99999",        // not "invalid token" → no-answer
            "response failed JSON schema validation",
            "tool call arguments did not validate",
        ] {
            assert!(
                !is_no_answer_failure(s),
                "expected generation failure (graded): {s:?}"
            );
        }
    }

    #[test]
    fn ledger_quality_distinguishes_generation_from_transport_failure() {
        // The durable ledger's `quality` is ANSWER quality, mirroring the
        // in-memory EMA (#374 follow-up). A generation failure produced a bad
        // answer → graded `Some(0.0)`; a transport failure or a 429 produced no
        // answer → `None`. This keeps the scoreboard's avg_quality and the
        // shadow-calibration band (#369) — both of which fold this ledger — free
        // of availability noise while still seeing the genuine bad-answer signal.
        let mut tracker = OutcomeTracker::new();

        let t = tracker.record_start("m", InferenceTask::Code, "r");
        tracker.record_failure(&t, "model produced malformed output"); // generation
        let t = tracker.record_start("m", InferenceTask::Code, "r");
        tracker.record_failure(&t, "daemon read timeout on infer after 30s"); // transport
        let t = tracker.record_start("m", InferenceTask::Code, "r");
        tracker.record_failure(&t, "429 RESOURCE_EXHAUSTED"); // rate-limit
        let t = tracker.record_start("m", InferenceTask::Code, "r");
        tracker.record_failure(&t, "upstream: 503 service unavailable"); // 5xx infra
        let t = tracker.record_start("m", InferenceTask::Code, "r");
        tracker.record_failure(&t, "HTTP 500 Internal Server Error"); // server fault
        let t = tracker.record_start("m", InferenceTask::Code, "r");
        // The live parslee/* failure: auth rejection, no answer produced.
        tracker.record_failure(
            &t,
            "inference failed: Parslee org lookup failed: HTTP 401 Unauthorized: Authentication",
        );

        let entries = tracker.drain_ledger();
        assert_eq!(entries.len(), 6);
        // All are failures for availability accounting.
        assert!(entries.iter().all(|e| e.success == Some(false)));
        // Only the generation failure carries a graded answer-quality of 0.0.
        assert_eq!(
            entries[0].quality,
            Some(0.0),
            "generation failure -> graded 0.0 answer quality"
        );
        assert_eq!(
            entries[1].quality, None,
            "transport failure produced no answer -> quality unknown (None)"
        );
        assert_eq!(
            entries[2].quality, None,
            "429 produced no answer -> quality unknown (None), an availability event"
        );
        assert_eq!(
            entries[3].quality, None,
            "5xx infra failure produced no answer -> None"
        );
        assert_eq!(
            entries[4].quality, None,
            "an 'internal server error' (500) is a server fault, not a bad answer -> None"
        );
        assert_eq!(
            entries[5].quality, None,
            "a 401 auth rejection produced no answer -> None (not a bad-answer 0.0)"
        );
    }

    #[test]
    fn real_failure_signal_reclassifies_mechanical_success() {
        // Completed with output -> mechanical success. A later Rejected signal
        // overrides it: undo the success, book a failure. No phantom success.
        let mut tracker = OutcomeTracker::new();
        let t = tracker.record_start("m", InferenceTask::Generate, "r");
        tracker.record_complete(&t, 100, 8, 15);
        assert_eq!(tracker.profile("m").unwrap().success_count, 1);

        tracker.record_inferred_outcome(&t, InferredOutcome::Rejected { confidence: 0.9 });
        let p = tracker.profile("m").unwrap();
        assert_eq!(p.success_count, 0, "mechanical success undone");
        assert_eq!(p.fail_count, 1, "failure booked");
    }

    #[test]
    fn real_success_signal_does_not_double_count() {
        let mut tracker = OutcomeTracker::new();
        let t = tracker.record_start("m", InferenceTask::Generate, "r");
        tracker.record_complete(&t, 100, 8, 15);
        tracker.record_inferred_outcome(&t, InferredOutcome::Accepted { confidence: 0.9 });
        let p = tracker.profile("m").unwrap();
        assert_eq!(
            p.success_count, 1,
            "Accepted on an already-credited call is not +2"
        );
    }

    #[test]
    fn export_import() {
        let mut tracker = OutcomeTracker::new();
        let trace = tracker.record_start("m1", InferenceTask::Generate, "test");
        tracker.record_complete(&trace, 100, 10, 5);
        tracker.record_inferred_outcome(&trace, InferredOutcome::Accepted { confidence: 0.9 });

        let exported = tracker.export_profiles();
        assert_eq!(exported.len(), 1);

        let mut tracker2 = OutcomeTracker::new();
        tracker2.import_profiles(exported);
        assert!(tracker2.profile("m1").is_some());
    }

    #[test]
    fn a_call_longer_than_the_ttl_can_still_be_graded_after_it_ends() {
        // car-releases#110: a 372 s generation with a 300 s TTL counted from
        // the START was swept while in flight, so neither its completion nor
        // its grade ever landed.
        let ttl = 900;
        let mut tracker = OutcomeTracker::new();
        let t = tracker.record_start("m", InferenceTask::Generate, "r");
        let started = now_unix().saturating_sub(ttl + 600);
        tracker.pending.get_mut(&t).unwrap().timestamp = started;

        // Still running past the TTL: not evicted.
        assert_eq!(tracker.sweep_pending_at(ttl, now_unix()), 0);
        tracker.record_complete(&t, (ttl + 600) * 1000, 10, 20);
        // Ended just now: the grace period starts here.
        assert_eq!(tracker.sweep_pending_at(ttl, now_unix() + ttl / 2), 0);

        let report = tracker
            .resolve_pending_reporting(vec![(t, InferredOutcome::Accepted { confidence: 0.9 })]);
        assert_eq!(report.resolved, 1);
        assert!(report.dropped.is_empty());
        let p = tracker.profile("m").unwrap();
        assert_eq!(p.quality_observations, 1, "the grade moved the EMA");
        assert_eq!(p.total_calls, 1, "the completion was booked");
    }

    #[test]
    fn an_ended_call_is_swept_ttl_after_it_ends() {
        let mut tracker = OutcomeTracker::new();
        let t = tracker.record_start("m", InferenceTask::Generate, "r");
        tracker.record_complete(&t, 100, 1, 1);
        assert_eq!(tracker.sweep_pending_at(900, now_unix() + 899), 0);
        assert_eq!(tracker.sweep_pending_at(900, now_unix() + 902), 1);
        assert!(!tracker.has_pending(&t));
    }

    #[test]
    fn resolve_reports_signals_that_matched_no_pending_trace() {
        let mut tracker = OutcomeTracker::new();
        let live = tracker.record_start("m", InferenceTask::Generate, "r");
        tracker.record_complete(&live, 100, 1, 1);
        let report = tracker.resolve_pending_reporting(vec![
            (live.clone(), InferredOutcome::Accepted { confidence: 0.8 }),
            // The same trace twice: the second signal has nothing to resolve.
            (live.clone(), InferredOutcome::Accepted { confidence: 0.8 }),
            (
                "t-0-swept".into(),
                InferredOutcome::Rejected { confidence: 0.9 },
            ),
        ]);
        assert_eq!(report.resolved, 1);
        assert_eq!(report.dropped, vec![live, "t-0-swept".to_string()]);
        assert_eq!(tracker.profile("m").unwrap().quality_observations, 1);
    }

    #[test]
    fn external_grade_is_kept_per_source_and_leaves_served_counters_alone() {
        let mut tracker = OutcomeTracker::new();
        let stats = tracker
            .record_external_grade("gpt-x", InferenceTask::Code, 0.25, 6, "neo")
            .unwrap();
        assert_eq!(stats.quality, 0.25);
        assert_eq!(stats.observations, 6);
        assert_eq!(stats.evidence(), 6);

        let p = tracker.profile("gpt-x").unwrap();
        assert_eq!(p.total_calls, 0);
        assert_eq!(p.success_count + p.fail_count, 0);
        assert_eq!(p.quality_observations, 0, "served evidence untouched");
        assert_eq!(p.ema_quality, NEUTRAL_QUALITY, "served EMA untouched");
        let ts = p.task_stats(InferenceTask::Code).unwrap();
        assert_eq!(ts.quality_observations, 0);
        assert!(!ts.has_served_evidence());
        assert_eq!(ts.external_evidence(), (6.0, 0.25));
        assert!(tracker.is_dirty(), "an external grade must reach disk");
        assert!(tracker.drain_ledger().is_empty(), "no served receipt");
    }

    #[test]
    fn one_source_cannot_add_weight_past_its_cap() {
        let mut tracker = OutcomeTracker::new();
        let big = tracker
            .record_external_grade("m", InferenceTask::Generate, 0.0, 100, "bench")
            .unwrap();
        assert_eq!(big.observations, 100);
        assert_eq!(big.evidence(), EXTERNAL_GRADE_EVIDENCE_CAP);
        // Many small reports from the same source add no further weight, but
        // the newest verdict still moves the source's quality.
        for _ in 0..20 {
            tracker
                .record_external_grade("m", InferenceTask::Generate, 1.0, 1, "bench")
                .unwrap();
        }
        let ts = tracker
            .profile("m")
            .unwrap()
            .task_stats(InferenceTask::Generate)
            .unwrap();
        let (evidence, mean) = ts.external_evidence();
        assert_eq!(evidence, EXTERNAL_GRADE_EVIDENCE_CAP as f64);
        assert!(mean > 0.9, "newer reports keep moving quality: {mean}");
        assert_eq!(ts.external_grades["bench"].observations, 120);
    }

    #[test]
    fn external_grade_validation_and_source_limit() {
        let mut tracker = OutcomeTracker::new();
        let task = InferenceTask::Generate;
        let refused = |r: Result<ExternalGradeStats, ExternalGradeRefusal>| r.unwrap_err();
        assert_eq!(
            refused(tracker.record_external_grade("m", task, 1.5, 1, "s")),
            ExternalGradeRefusal::InvalidQuality
        );
        assert_eq!(
            refused(tracker.record_external_grade("m", task, f64::NAN, 1, "s")),
            ExternalGradeRefusal::InvalidQuality
        );
        assert_eq!(
            refused(tracker.record_external_grade("m", task, 0.5, 0, "s")),
            ExternalGradeRefusal::InvalidCases
        );
        assert_eq!(
            refused(tracker.record_external_grade(
                "m",
                task,
                0.5,
                MAX_EXTERNAL_GRADE_CASES + 1,
                "s"
            )),
            ExternalGradeRefusal::InvalidCases
        );
        assert_eq!(
            refused(tracker.record_external_grade("m", task, 0.5, 1, "has space")),
            ExternalGradeRefusal::InvalidSource
        );
        assert_eq!(
            refused(tracker.record_external_grade("m", task, 0.5, 1, &"s".repeat(65))),
            ExternalGradeRefusal::InvalidSource
        );
        assert_eq!(
            refused(tracker.record_external_grade(" ", task, 0.5, 1, "s")),
            ExternalGradeRefusal::InvalidModel
        );
        assert!(tracker.profile("m").is_none(), "a refusal records nothing");

        for i in 0..MAX_EXTERNAL_GRADE_SOURCES {
            tracker
                .record_external_grade("m", task, 0.5, 1, &format!("src-{i}"))
                .unwrap();
        }
        assert_eq!(
            refused(tracker.record_external_grade("m", task, 0.5, 1, "one-more")),
            ExternalGradeRefusal::SourceLimitReached
        );
        // A source already counted keeps reporting.
        tracker
            .record_external_grade("m", task, 0.9, 1, "src-0")
            .unwrap();
        // The limit is per task.
        tracker
            .record_external_grade("m", InferenceTask::Code, 0.5, 1, "one-more")
            .unwrap();
    }

    #[test]
    fn one_reporter_cannot_add_weight_by_rotating_source_names() {
        let mut tracker = OutcomeTracker::new();
        let task = InferenceTask::Code;
        for i in 0..MAX_EXTERNAL_GRADE_SOURCES {
            tracker
                .record_external_grade("m", task, 0.0, 8, &format!("alias-{i}"))
                .unwrap();
        }
        let ts = tracker.profile("m").unwrap().task_stats(task).unwrap();
        assert_eq!(
            ts.external_evidence(),
            (EXTERNAL_GRADE_REPORTER_EVIDENCE_CAP as f64, 0.0),
            "eight names from one reporter count as one reporter's cap"
        );
        // A second reporter is its own budget, and the pool is its mean.
        for i in 0..2 {
            tracker
                .record_external_grade_as(Some("agent:b"), "m", task, 1.0, 8, &format!("s{i}"))
                .unwrap();
        }
        let ts = tracker.profile("m").unwrap().task_stats(task).unwrap();
        assert_eq!(ts.external_evidence(), (16.0, 0.5));
        // A third reporter adds its mean but no weight past the total cap.
        tracker
            .record_external_grade_as(Some("agent:c"), "m", task, 1.0, 8, "s")
            .unwrap();
        let ts = tracker.profile("m").unwrap().task_stats(task).unwrap();
        let (evidence, mean) = ts.external_evidence();
        assert_eq!(evidence, EXTERNAL_GRADE_TOTAL_EVIDENCE_CAP as f64);
        assert!((mean - 2.0 / 3.0).abs() < 1e-12, "{mean}");
    }

    #[test]
    fn a_reporter_cannot_write_another_reporters_source() {
        let mut tracker = OutcomeTracker::new();
        let task = InferenceTask::Code;
        tracker
            .record_external_grade("m", task, 0.9, 8, "neo")
            .unwrap();
        tracker
            .record_external_grade_as(Some("agent:mallory"), "m", task, 0.0, 100, "neo")
            .unwrap();
        let ts = tracker.profile("m").unwrap().task_stats(task).unwrap();
        assert_eq!(
            ts.external_grades["neo"].quality, 0.9,
            "operator's untouched"
        );
        assert_eq!(ts.external_grades["agent:mallory/neo"].quality, 0.0);
        assert_eq!(
            tracker
                .record_external_grade_as(Some("agent:a/b"), "m", task, 0.5, 1, "x")
                .unwrap_err(),
            ExternalGradeRefusal::InvalidReporter,
            "a reporter cannot forge a key under another prefix"
        );
        // The source limit is per reporter: an agent at its limit does not
        // lock the operator, or another agent, out of the task.
        for i in 0..MAX_EXTERNAL_GRADE_SOURCES {
            let _ = tracker.record_external_grade_as(
                Some("agent:mallory"),
                "m",
                task,
                0.0,
                1,
                &format!("junk-{i}"),
            );
        }
        tracker
            .record_external_grade("m", task, 0.5, 1, "nightly")
            .unwrap();
        tracker
            .record_external_grade_as(Some("agent:other"), "m", task, 0.5, 1, "nightly")
            .unwrap();
    }

    #[test]
    fn a_stale_source_stops_counting_and_gives_up_its_slot() {
        let mut tracker = OutcomeTracker::new();
        let task = InferenceTask::Code;
        let then = 1_000_000_000;
        let later = then + EXTERNAL_GRADE_MAX_AGE_SECS + 1;
        for i in 0..MAX_EXTERNAL_GRADE_SOURCES {
            tracker
                .record_external_grade_at(None, "m", task, 0.0, 8, &format!("old-{i}"), then)
                .unwrap();
        }
        let ts = tracker.profile("m").unwrap().task_stats(task).unwrap();
        assert_eq!(ts.external_evidence_at(then).0, 8.0);
        assert_eq!(
            ts.external_evidence_at(later),
            (0.0, 0.0),
            "a stale grade no longer steers routing"
        );
        // While fresh, the limit refuses a new source; once stale, the
        // stalest gives up its slot.
        assert_eq!(
            tracker
                .record_external_grade_at(None, "m", task, 1.0, 1, "new", then + 1)
                .unwrap_err(),
            ExternalGradeRefusal::SourceLimitReached
        );
        tracker
            .record_external_grade_at(None, "m", task, 1.0, 1, "new", later)
            .unwrap();
        let ts = tracker.profile("m").unwrap().task_stats(task).unwrap();
        assert_eq!(ts.external_grades.len(), MAX_EXTERNAL_GRADE_SOURCES);
        assert!(ts.external_grades.contains_key("new"));
        assert_eq!(ts.external_evidence_at(later), (1.0, 1.0));
        // Reporting again under a stale source starts it afresh.
        let restarted = tracker
            .record_external_grade_at(None, "m", task, 1.0, 1, "old-7", later)
            .unwrap();
        assert_eq!((restarted.quality, restarted.observations), (1.0, 1));
    }

    #[test]
    fn external_grades_survive_a_save_and_load() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("profiles.json");
        let mut tracker = OutcomeTracker::new();
        tracker
            .record_external_grade("m", InferenceTask::Code, 0.75, 3, "neo")
            .unwrap();
        tracker.save_to_file(&path).unwrap();

        let mut loaded = OutcomeTracker::new();
        loaded.load_from_file(&path).unwrap();
        let ts = loaded
            .profile("m")
            .unwrap()
            .task_stats(InferenceTask::Code)
            .unwrap();
        assert_eq!(ts.external_grades["neo"].quality, 0.75);
        assert_eq!(ts.external_grades["neo"].observations, 3);

        // A profile written before external grades existed still loads.
        let old: TaskStats = serde_json::from_str(
            r#"{"calls":1,"successes":1,"failures":0,"avg_latency_ms":1.0,"ema_quality":0.5}"#,
        )
        .unwrap();
        assert!(old.external_grades.is_empty());
        assert!(!serde_json::to_string(&old)
            .unwrap()
            .contains("external_grades"));
    }
}