car-server-core 0.37.0

//! Paired A/B comparator: two arms — each an ([`ArmSpec`]) **engine on a
//! backbone** — over the **same task and the same `OutcomeContract`**.
//!
//! The original claim was "CAR + gpt-5.5 codes as well as Codex + gpt-5.5, and
//! here is where its harness wins." That is only meaningful when both arms run
//! the *same* model — ALE's finding is that backbone choice is ~3× the spread of
//! harness choice, so native-on-Haiku vs Codex-on-gpt-5.5 measures the model,
//! not the runtime.
//!
//! Arms therefore carry their own backbone, which lets one comparator express
//! the whole family (see `docs/proposals/coder-ab-value-surface.md`):
//! the same-backbone harness delta, the **cross-tier diagonal**
//! (`native@gpt-5.4` vs `codex@gpt-5.5` — what the runtime is worth in model
//! tiers), and **ablation** (native vs native with a mechanism off, the only
//! shape that isolates a single mechanism).
//!
//! The fairness rule is now *checked* rather than asserted
//! ([`AbReport::same_backbone`]). It previously could not be: the report held one
//! operator-supplied `backbone` string while the pin reached only the native
//! loop, so the external arm ran on whatever its CLI was configured for.
//!
//! Design mirrors `bench/car_bench/ale`'s orchestrator, ported to Rust and to
//! the in-process coder path:
//!
//! - **Honest denominator.** An arm that never actually ran (worktree/setup/
//!   inference-transport failure) is `infra_failed` and excluded — never scored
//!   as a task-0. The paired stats are computed over the tasks *both* arms
//!   scored, exactly like `compare_pair`'s `treatment ∩ control` intersection.
//! - **Paired significance.** McNemar's test over the discordant pairs
//!   (treatment-pass/control-fail vs treatment-fail/control-pass) — the right test
//!   for two harnesses judged on the same items, and the regression gate Slice 4
//!   re-runs to decide whether an applied change was a real improvement or noise.
//! - **Cost axis.** Per-arm mean cost and cost-per-pass, so "as good" can be
//!   qualified by "and cheaper/dearer."
//!
//! The execution seam is injected ([`AbArmRunner`]) — the same
//! injected-closure philosophy as `car-builder`'s generate seam and
//! `car-verify::cwm`'s `EffectModel` — so the statistics core is unit-testable
//! with scripted arms and no live inference. The live runner (real
//! `run_native_loop` / `run_external_loop` in fresh worktrees) is wired by the
//! `car coder-ab` CLI.

use std::collections::HashMap;

use async_trait::async_trait;
use car_eventlog::harness_adapt::{diagnose_from_jsonl, HarnessIntervention, InterventionLayer};
use serde::{Deserialize, Serialize};

use super::contract::OutcomeContract;

/// One coding task both arms attempt, verified against the identical contract.
/// The corpus is a JSONL of these — curated from CAR's own merged PRs / closed
/// issues (revert the fix, keep the PR's tests as the contract, have both arms
/// re-derive it).
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct AbTask {
    /// Stable id (e.g. the PR number or issue slug).
    pub id: String,
    /// The natural-language task both arms are given.
    pub intent: String,
    /// The pass/fail ground truth both arms are verified against — identical
    /// across arms so the only variable is the harness.
    pub contract: OutcomeContract,
    /// Repo the task runs against. `None` = the CLI's `--repo`/cwd default; each
    /// arm still works in its own throwaway worktree off it.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub repo: Option<String>,
}

/// Which engine an arm drives.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ArmEngine {
    /// CAR's native loop.
    Native,
    /// An external CLI session (`codex`, `claude-code`, `gemini`).
    External(String),
}

impl ArmEngine {
    /// Stable engine key (`"native"`, `"codex"`).
    pub fn label(&self) -> String {
        match self {
            Self::Native => "native".to_string(),
            Self::External(id) => id.clone(),
        }
    }
}

/// A fully-specified arm: **which engine, on which backbone**.
///
/// The backbone lives here rather than once per report because the interesting
/// experiments need the two arms to differ on it:
///
/// - **same-backbone** (`native@gpt-5.5` vs `codex@gpt-5.5`) — the harness delta,
///   the original claim;
/// - **cross-tier diagonal** (`native@gpt-5.4` vs `codex@gpt-5.5`) — what the
///   runtime is worth in *model tiers*, which is the economically legible claim
///   (see `docs/proposals/coder-ab-value-surface.md`);
/// - **ablation** (`native@gpt-5.5` vs `native@gpt-5.5` with a mechanism off) —
///   isolates one mechanism, which a CAR-vs-Codex comparison can never do since
///   it confounds the harness with everything else about Codex.
///
/// The old shape (an `Arm` enum plus one report-level `backbone`) could express
/// only the first, and even then it did not *enforce* it: the pin reached the
/// native loop alone, so the external arm silently ran on its CLI's configured
/// default. `model` here is threaded to whichever engine runs, so the invariant
/// is now the runtime's job, not the operator's memory.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ArmSpec {
    pub engine: ArmEngine,
    /// The pinned backbone. `None` = the engine's own default (adaptive routing
    /// for native; the CLI's config for external) — unpinned, so a same-backbone
    /// claim is unverifiable.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub model: Option<String>,
}

impl ArmSpec {
    /// CAR's native loop on `model`.
    pub fn native(model: Option<String>) -> Self {
        Self {
            engine: ArmEngine::Native,
            model,
        }
    }

    /// An external CLI (`codex`) on `model`.
    pub fn external(id: impl Into<String>, model: Option<String>) -> Self {
        Self {
            engine: ArmEngine::External(id.into()),
            model,
        }
    }

    /// Report label: `native@gpt-5.4`, `codex@gpt-5.5`, or bare `native` when
    /// unpinned. Two arms in one run can share an engine (ablation) or a model
    /// (the harness delta), so the label carries both to stay unambiguous.
    pub fn label(&self) -> String {
        match &self.model {
            Some(m) if !m.trim().is_empty() => format!("{}@{}", self.engine.label(), m),
            _ => self.engine.label(),
        }
    }
}

/// A single arm's result on a single task, normalized so both the native
/// [`LoopOutcome`](super::native_loop::LoopOutcome) and the external CLI's
/// result map onto one shape.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct ArmOutcome {
    /// Every contract check passed (the ground-truth win condition).
    pub passed: bool,
    /// Contract-evaluation rounds actually executed.
    pub iterations: u32,
    /// USD spent on inference for this arm's run (0.0 when unknown — the native
    /// loop does not always meter; the external CLI reports `total_cost_usd`).
    #[serde(default)]
    pub cost_usd: f64,
    /// Wall-clock for the arm's run.
    pub wall_ms: u64,
    /// Terminal error (cancellation, repeated inference failure). Distinct from
    /// `infra_failed`: an arm can error late (task-attributable) after really
    /// running.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    /// Where the arm's transcript/event stream landed, for Slice 3 attribution.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub transcript_path: Option<String>,
    /// The arm never actually attempted the task — worktree/setup/CLI-launch/
    /// transport failure before any real work. Excluded from the honest
    /// denominator (ALE's infra-vs-task split); NOT a scored task-0.
    #[serde(default)]
    pub infra_failed: bool,
}

impl ArmOutcome {
    /// An infra failure: the arm couldn't be attempted. Kept out of the scored
    /// denominator.
    pub fn infra(reason: impl Into<String>, wall_ms: u64) -> Self {
        Self {
            passed: false,
            iterations: 0,
            cost_usd: 0.0,
            wall_ms,
            error: Some(reason.into()),
            transcript_path: None,
            infra_failed: true,
        }
    }

    /// Whether this outcome counts toward the scored denominator.
    pub fn scorable(&self) -> bool {
        !self.infra_failed
    }
}

/// Both arms' outcomes on one task — the paired unit.
///
/// `treatment`/`control` (ALE's `compare_pair` vocabulary, which this module's
/// statistics already mirror) rather than `native`/`external`: the arms are no
/// longer engine-typed, and under ablation both are native. The asymmetry is
/// kept deliberately — `treatment` is the arm under test, `control` the
/// reference — because the delta's sign and `ab_loop`'s quality bar both need to
/// know which arm is "us".
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct AbCell {
    pub task_id: String,
    #[serde(alias = "native")]
    pub treatment: ArmOutcome,
    #[serde(alias = "external")]
    pub control: ArmOutcome,
    /// The exact arms that produced this cell.
    ///
    /// Per-cell, not per-report, because the resume checkpoint is keyed by
    /// corpus path alone: without provenance, running `flask@gpt-5.5` and then
    /// resuming the same corpus at `flask@gpt-5.4` silently reuses the 5.5 cells
    /// and folds two backbones into one report — a contaminated result that
    /// looks perfectly clean. [`AbCell::matches`] is the guard.
    ///
    /// `None` = a legacy checkpoint written before provenance existed; treated
    /// as *unknown*, hence never reusable.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub treatment_spec: Option<ArmSpec>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub control_spec: Option<ArmSpec>,
}

impl AbCell {
    /// The cell is *paired-scorable* only when BOTH arms really ran — the
    /// intersection ALE's `compare_pair` measures the delta over.
    pub fn paired_scorable(&self) -> bool {
        self.treatment.scorable() && self.control.scorable()
    }

    /// Whether this cell was produced by exactly these two arms, and is
    /// therefore safe to reuse on resume. Unknown provenance is never a match —
    /// re-running a task is cheap next to publishing a mixed-backbone report.
    pub fn matches(&self, treatment: &ArmSpec, control: &ArmSpec) -> bool {
        self.treatment_spec.as_ref() == Some(treatment)
            && self.control_spec.as_ref() == Some(control)
    }
}

/// McNemar's chi-square (df=1) critical value at α=0.05.
const CHI2_CRIT_05: f64 = 3.841;

/// Aggregate paired statistics over the scorable intersection.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct PairedStats {
    /// Tasks both arms scored (the honest denominator).
    pub paired_tasks: usize,
    pub treatment_passes: usize,
    pub control_passes: usize,
    pub treatment_pass_rate: f64,
    pub control_pass_rate: f64,
    /// `treatment_pass_rate - control_pass_rate` (>0 = the arm under test ahead).
    pub pass_rate_delta: f64,
    /// Concordant/discordant breakdown of the paired outcomes.
    pub both_pass: usize,
    /// Discordant, treatment's favor (McNemar's `b`).
    pub treatment_only: usize,
    /// Discordant, control's favor (McNemar's `c`).
    pub control_only: usize,
    pub both_fail: usize,
    /// McNemar's chi-square with continuity correction: `(|b-c|-1)^2/(b+c)`.
    /// `0.0` when there are no discordant pairs.
    pub mcnemar_chi2: f64,
    /// `mcnemar_chi2 > 3.841` — the delta is unlikely to be noise at α=0.05.
    pub mcnemar_significant_05: bool,
    /// Arms that couldn't be attempted (reported, excluded from the delta).
    pub treatment_infra_failures: usize,
    pub control_infra_failures: usize,
    /// Mean inference cost per scorable arm run.
    pub treatment_mean_cost_usd: f64,
    pub control_mean_cost_usd: f64,
    /// Total cost / passes — `None` when an arm never passed (undefined).
    pub treatment_cost_per_pass: Option<f64>,
    pub control_cost_per_pass: Option<f64>,
}

impl PairedStats {
    fn from_cells(cells: &[AbCell]) -> Self {
        let treatment_infra_failures = cells.iter().filter(|c| c.treatment.infra_failed).count();
        let control_infra_failures = cells.iter().filter(|c| c.control.infra_failed).count();

        // The honest denominator: only cells where BOTH arms really ran.
        let paired: Vec<&AbCell> = cells.iter().filter(|c| c.paired_scorable()).collect();
        let paired_tasks = paired.len();

        let treatment_passes = paired.iter().filter(|c| c.treatment.passed).count();
        let control_passes = paired.iter().filter(|c| c.control.passed).count();

        let rate = |n: usize| {
            if paired_tasks == 0 {
                0.0
            } else {
                n as f64 / paired_tasks as f64
            }
        };
        let treatment_pass_rate = rate(treatment_passes);
        let control_pass_rate = rate(control_passes);

        let both_pass = paired
            .iter()
            .filter(|c| c.treatment.passed && c.control.passed)
            .count();
        let treatment_only = paired
            .iter()
            .filter(|c| c.treatment.passed && !c.control.passed)
            .count();
        let control_only = paired
            .iter()
            .filter(|c| !c.treatment.passed && c.control.passed)
            .count();
        let both_fail = paired
            .iter()
            .filter(|c| !c.treatment.passed && !c.control.passed)
            .count();

        // McNemar with Yates continuity correction over the discordant pairs.
        let b = treatment_only as f64;
        let c = control_only as f64;
        let discordant = b + c;
        let mcnemar_chi2 = if discordant > 0.0 {
            let num = (b - c).abs() - 1.0;
            // Clamp the corrected numerator at 0 (|b-c| can be < 1).
            let num = num.max(0.0);
            num * num / discordant
        } else {
            0.0
        };
        let mcnemar_significant_05 = discordant > 0.0 && mcnemar_chi2 > CHI2_CRIT_05;

        // Cost axis over scorable arms only.
        let mean = |sel: &dyn Fn(&AbCell) -> Option<f64>| {
            let xs: Vec<f64> = cells.iter().filter_map(sel).collect();
            if xs.is_empty() {
                0.0
            } else {
                xs.iter().sum::<f64>() / xs.len() as f64
            }
        };
        let treatment_mean_cost_usd =
            mean(&|c| c.treatment.scorable().then_some(c.treatment.cost_usd));
        let control_mean_cost_usd = mean(&|c| c.control.scorable().then_some(c.control.cost_usd));

        let cost_per_pass = |sel: &dyn Fn(&AbCell) -> &ArmOutcome| {
            let scorable: Vec<&ArmOutcome> =
                cells.iter().map(sel).filter(|o| o.scorable()).collect();
            let passes = scorable.iter().filter(|o| o.passed).count();
            if passes == 0 {
                None
            } else {
                let total: f64 = scorable.iter().map(|o| o.cost_usd).sum();
                Some(total / passes as f64)
            }
        };
        let treatment_cost_per_pass = cost_per_pass(&|c| &c.treatment);
        let control_cost_per_pass = cost_per_pass(&|c| &c.control);

        Self {
            paired_tasks,
            treatment_passes,
            control_passes,
            treatment_pass_rate,
            control_pass_rate,
            pass_rate_delta: treatment_pass_rate - control_pass_rate,
            both_pass,
            treatment_only,
            control_only,
            both_fail,
            mcnemar_chi2,
            mcnemar_significant_05,
            treatment_infra_failures,
            control_infra_failures,
            treatment_mean_cost_usd,
            control_mean_cost_usd,
            treatment_cost_per_pass,
            control_cost_per_pass,
        }
    }
}

/// The full report: every paired cell plus the aggregate stats. Serializes to
/// the timestamped JSON the `car coder-ab` CLI writes to `bench/results`.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct AbReport {
    /// The arm under test.
    pub treatment: ArmSpec,
    /// The reference arm.
    pub control: ArmSpec,
    pub cells: Vec<AbCell>,
    pub stats: PairedStats,
}

impl AbReport {
    /// Fold cells into a report against the two arms that produced them.
    pub fn from_cells(cells: Vec<AbCell>, treatment: ArmSpec, control: ArmSpec) -> Self {
        let stats = PairedStats::from_cells(&cells);
        Self {
            treatment,
            control,
            cells,
            stats,
        }
    }

    /// Whether both arms ran the **same pinned backbone** — the fairness rule
    /// that makes `pass_rate_delta` a *harness* measurement rather than a model
    /// measurement (backbone choice is ~3× the harness spread).
    ///
    /// `None` when either arm is unpinned: the honest answer is then *unknown*,
    /// not "yes". This used to be unknowable — the report recorded a single
    /// `backbone` string the operator asserted, while the pin reached only the
    /// native loop. `false` is a legitimate, deliberate configuration (the
    /// cross-tier diagonal); what must never happen is *believing* it is `true`
    /// without having checked.
    /// `Some(false)` means the two arms carry **different pin strings** — which
    /// is not quite the same as "different models", because the id namespaces
    /// differ per engine (`parslee/reasoning` and `gpt-5.5` name one backbone by
    /// two routes). We cannot resolve that equivalence here, so the honest
    /// reading is "different pins, check what you intended", not "different
    /// tiers". Identical pins are unambiguous.
    pub fn same_backbone(&self) -> Option<bool> {
        let t = self.treatment.model.as_deref()?;
        let c = self.control.model.as_deref()?;
        Some(t == c)
    }

    /// A one-line human summary for the CLI table footer.
    pub fn summary_line(&self) -> String {
        let s = &self.stats;
        let sig = if s.mcnemar_significant_05 {
            "significant (p<.05)"
        } else {
            "not significant"
        };
        // Name the comparison being made, so a cross-tier diagonal can never be
        // read as a same-backbone harness delta.
        let kind = match self.same_backbone() {
            Some(true) => "same-backbone harness delta",
            Some(false) => "DIFFERENT PINS (not a same-backbone harness delta)",
            None => "UNPINNED (backbone unverified)",
        };
        format!(
            "{} {}/{} ({:.0}%) vs {} {}/{} ({:.0}%) over {} paired tasks — delta {:+.1} pp, McNemar χ²={:.2} {} [{}]",
            self.treatment.label(),
            s.treatment_passes,
            s.paired_tasks,
            s.treatment_pass_rate * 100.0,
            self.control.label(),
            s.control_passes,
            s.paired_tasks,
            s.control_pass_rate * 100.0,
            s.paired_tasks,
            s.pass_rate_delta * 100.0,
            s.mcnemar_chi2,
            sig,
            kind,
        )
    }
}

/// The injected execution seam: run one arm on one task, in its own fresh
/// worktree, verifying against `task.contract`. Never panics — an arm that
/// can't be attempted returns [`ArmOutcome::infra`]. Scripted in tests; the
/// live impl (real `run_native_loop` / `run_external_loop`) is wired by the CLI.
#[async_trait]
pub trait AbArmRunner: Send + Sync {
    async fn run_arm(&self, task: &AbTask, arm: &ArmSpec) -> ArmOutcome;
}

/// Run the full paired suite: each task through both arms, against the same
/// contract, folded into an [`AbReport`].
pub async fn run_ab_suite(
    tasks: &[AbTask],
    treatment: &ArmSpec,
    control: &ArmSpec,
    runner: &dyn AbArmRunner,
) -> AbReport {
    run_ab_suite_resumable(tasks, treatment, control, runner, Vec::new(), None, |_| {}).await
}

/// Resumable, incrementally-observable [`run_ab_suite`]. A long real-corpus run
/// is expensive, so this makes it interruptible and iterative:
///
/// - `done` are cells already scored (loaded from a durable checkpoint); their
///   tasks are skipped and the cells kept, so a killed run loses nothing — a
///   re-run picks up where it stopped. **Only cells produced by these exact two
///   arms are reused** ([`AbCell::matches`]); the rest are dropped and re-run,
///   because the checkpoint is keyed by corpus path alone and would otherwise
///   fold a previous run's *different backbone* into this report.
/// - `limit` caps how many NEW tasks to run this invocation (`None` = all
///   remaining): run one, inspect the result, fix whatever it surfaced, resume.
/// - `on_cell` fires the instant each new cell is scored, BEFORE the next task
///   starts — the caller persists it there (append to the checkpoint), which is
///   what makes a mid-run stop lossless.
///
/// Returns the report over ALL cells (previously `done` + newly run).
pub async fn run_ab_suite_resumable(
    tasks: &[AbTask],
    treatment: &ArmSpec,
    control: &ArmSpec,
    runner: &dyn AbArmRunner,
    done: Vec<AbCell>,
    limit: Option<usize>,
    mut on_cell: impl FnMut(&AbCell),
) -> AbReport {
    // Reuse only what THIS pair of arms produced. A cell from a different
    // backbone (or unknown provenance) is silent contamination, not a saving.
    let mut cells: Vec<AbCell> = done
        .into_iter()
        .filter(|c| c.matches(treatment, control))
        .collect();
    let done_ids: std::collections::HashSet<String> =
        cells.iter().map(|c| c.task_id.clone()).collect();
    let mut ran = 0usize;
    for task in tasks {
        if done_ids.contains(&task.id) {
            continue; // already scored by these arms — resume past it
        }
        if limit.is_some_and(|lim| ran >= lim) {
            break; // stop after this batch; the rest resume on the next run
        }
        let treatment_outcome = runner.run_arm(task, treatment).await;
        let control_outcome = runner.run_arm(task, control).await;
        let cell = AbCell {
            task_id: task.id.clone(),
            treatment: treatment_outcome,
            control: control_outcome,
            treatment_spec: Some(treatment.clone()),
            control_spec: Some(control.clone()),
        };
        on_cell(&cell); // durably persist BEFORE moving on
        cells.push(cell);
        ran += 1;
    }
    AbReport::from_cells(cells, treatment.clone(), control.clone())
}

// ---------------------------------------------------------------------------
// Slice 3 — mechanism attribution.
//
// The A/B tells us *that* CAR's coder lost a task; attribution tells us *why*,
// and — crucially — whether it's a loss CAR can fix. We fold each native loss's
// event stream through `car_eventlog::harness_adapt::diagnose`, which already
// recognizes the loss mechanisms (premature certification / ungrounded
// completion, truncation, turn-exhaustion, retry thrash, malformed calls) and
// maps them to a Life-Harness intervention layer. The operational split (the
// ALE finding "verify-mode can't fix clean-but-wrong"):
//
//   - a losing native cell whose events yield ≥1 intervention → HARNESS-ADDRESSABLE
//     (Slice 4 hands these to the Evolution Agent),
//   - a losing native cell that ran clean and simply produced a wrong answer,
//     yielding no diagnosed pattern → BACKBONE-BOUND (parked — no harness lever;
//     wait for a stronger model).
// ---------------------------------------------------------------------------

/// Attribution over one A/B round's treatment-arm losses.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct RoundAttribution {
    /// Treatment-losing paired tasks whose event stream yielded ≥1 intervention.
    pub harness_addressable_losses: Vec<String>,
    /// Treatment-losing paired tasks that ran clean but produced a wrong answer
    /// (no diagnosed pattern) — a backbone/domain signal, not a harness lever.
    pub backbone_bound_losses: Vec<String>,
    /// Interventions merged across all addressable losses, highest evidence
    /// first — the telemetry Slice 4 feeds to `evolution.run`.
    pub interventions: Vec<HarnessIntervention>,
}

impl RoundAttribution {
    /// Whether there is any harness-addressable lever to act on this round.
    pub fn has_lever(&self) -> bool {
        !self.interventions.is_empty()
    }
}

/// Whether a diagnosed layer is one the Evolution Agent can act on autonomously.
/// `ProceduralSkill` defers to CAR's separate skill-distillation path, so it is
/// not fed to the harness-evolution fixer here.
pub fn is_evolution_actionable(layer: InterventionLayer) -> bool {
    matches!(
        layer,
        InterventionLayer::EnvironmentContract
            | InterventionLayer::ActionRealization
            | InterventionLayer::TrajectoryRegulation
    )
}

/// Attribute the treatment-arm losses in `report`. `read_events(path)` returns the
/// event-log JSONL for a transcript path (live: read the file; tests: an
/// in-memory map). Only *paired-scorable* native losses are attributed — an
/// infra failure is not a harness lesson. Interventions from a layer that
/// defers elsewhere ([`is_evolution_actionable`] false) still mark a loss as
/// addressed but are dropped from the fixer feed.
pub fn attribute_round(
    report: &AbReport,
    read_events: impl Fn(&str) -> Option<String>,
    min_occurrences: usize,
) -> RoundAttribution {
    let mut addressable = Vec::new();
    let mut backbone = Vec::new();
    // Merge interventions across losing cells by (layer, target), summing
    // evidence so a pattern recurring across tasks ranks above a one-off.
    let mut merged: HashMap<(String, String), HarnessIntervention> = HashMap::new();

    for cell in &report.cells {
        if !cell.paired_scorable() || cell.treatment.passed {
            continue;
        }
        let jsonl = cell
            .treatment
            .transcript_path
            .as_deref()
            .and_then(&read_events)
            .unwrap_or_default();
        let diag = diagnose_from_jsonl(&jsonl, min_occurrences);
        let actionable: Vec<HarnessIntervention> = diag
            .interventions
            .into_iter()
            .filter(|iv| is_evolution_actionable(iv.layer))
            .collect();
        if actionable.is_empty() {
            backbone.push(cell.task_id.clone());
        } else {
            addressable.push(cell.task_id.clone());
            for iv in actionable {
                let key = (format!("{:?}", iv.layer), iv.target.clone());
                merged
                    .entry(key)
                    .and_modify(|e| e.evidence_count += iv.evidence_count)
                    .or_insert(iv);
            }
        }
    }

    let mut interventions: Vec<HarnessIntervention> = merged.into_values().collect();
    interventions.sort_by(|a, b| b.evidence_count.cmp(&a.evidence_count));
    RoundAttribution {
        harness_addressable_losses: addressable,
        backbone_bound_losses: backbone,
        interventions,
    }
}

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

    fn contract() -> OutcomeContract {
        OutcomeContract {
            description: "tests pass".into(),
            checks: vec![],
        }
    }

    fn task(id: &str) -> AbTask {
        AbTask {
            id: id.into(),
            intent: format!("do {id}"),
            contract: contract(),
            repo: None,
        }
    }

    fn done(passed: bool, cost: f64) -> ArmOutcome {
        ArmOutcome {
            passed,
            iterations: 1,
            cost_usd: cost,
            wall_ms: 10,
            error: None,
            transcript_path: None,
            infra_failed: false,
        }
    }

    /// The standard pair: CAR native vs codex, both pinned to one backbone.
    fn tspec() -> ArmSpec {
        ArmSpec::native(Some("parslee/reasoning".into()))
    }
    fn cspec() -> ArmSpec {
        ArmSpec::external("codex", Some("parslee/reasoning".into()))
    }

    /// A cell carrying the standard pair's provenance (so it is resume-reusable).
    fn cell_of(id: &str, t: ArmOutcome, c: ArmOutcome) -> AbCell {
        AbCell {
            task_id: id.into(),
            treatment: t,
            control: c,
            treatment_spec: Some(tspec()),
            control_spec: Some(cspec()),
        }
    }

    /// A scripted runner: outcomes keyed by (task_id, engine_label).
    struct Scripted(HashMap<(String, String), ArmOutcome>);

    #[async_trait]
    impl AbArmRunner for Scripted {
        async fn run_arm(&self, task: &AbTask, arm: &ArmSpec) -> ArmOutcome {
            self.0
                .get(&(task.id.clone(), arm.engine.label()))
                .cloned()
                .unwrap_or_else(|| ArmOutcome::infra("no script", 0))
        }
    }

    fn script(entries: Vec<(&str, &str, ArmOutcome)>) -> Scripted {
        Scripted(
            entries
                .into_iter()
                .map(|(t, a, o)| ((t.to_string(), a.to_string()), o))
                .collect(),
        )
    }

    #[tokio::test]
    async fn paired_delta_and_mcnemar_over_discordant_pairs() {
        // 5 tasks. native beats external on 3 discordant, loses 0, both-pass 1,
        // both-fail 1. b=3, c=0 → chi2 = (|3-0|-1)^2/3 = 4/3 = 1.333 (not sig).
        let tasks: Vec<AbTask> = (0..5).map(|i| task(&format!("t{i}"))).collect();
        let runner = script(vec![
            ("t0", "native", done(true, 0.10)),
            ("t0", "codex", done(false, 0.20)),
            ("t1", "native", done(true, 0.10)),
            ("t1", "codex", done(false, 0.20)),
            ("t2", "native", done(true, 0.10)),
            ("t2", "codex", done(false, 0.20)),
            ("t3", "native", done(true, 0.10)),
            ("t3", "codex", done(true, 0.20)),
            ("t4", "native", done(false, 0.10)),
            ("t4", "codex", done(false, 0.20)),
        ]);
        let report = run_ab_suite(&tasks, &tspec(), &cspec(), &runner).await;
        let s = &report.stats;
        assert_eq!(s.paired_tasks, 5);
        assert_eq!(s.treatment_passes, 4);
        assert_eq!(s.control_passes, 1);
        assert_eq!(s.treatment_only, 3);
        assert_eq!(s.control_only, 0);
        assert_eq!(s.both_pass, 1);
        assert_eq!(s.both_fail, 1);
        assert!((s.pass_rate_delta - 0.6).abs() < 1e-9);
        assert!((s.mcnemar_chi2 - 4.0 / 3.0).abs() < 1e-9);
        assert!(!s.mcnemar_significant_05);
        // Cost-per-pass: native 5 scorable runs × 0.10 = 0.50 / 4 passes.
        assert!((s.treatment_cost_per_pass.unwrap() - 0.50 / 4.0).abs() < 1e-9);
        // External never... passed once (t3): 5 × 0.20 = 1.0 / 1.
        assert!((s.control_cost_per_pass.unwrap() - 1.0).abs() < 1e-9);
    }

    #[tokio::test]
    async fn resumable_skips_done_caps_by_limit_and_observes_each_cell() {
        let tasks: Vec<AbTask> = (0..4).map(|i| task(&format!("t{i}"))).collect();
        let runner = script(vec![
            ("t0", "native", done(true, 0.0)),
            ("t0", "codex", done(false, 0.0)),
            ("t1", "native", done(true, 0.0)),
            ("t1", "codex", done(false, 0.0)),
            ("t2", "native", done(true, 0.0)),
            ("t2", "codex", done(false, 0.0)),
            ("t3", "native", done(true, 0.0)),
            ("t3", "codex", done(false, 0.0)),
        ]);
        let cell = |id: &str| cell_of(id, done(true, 0.0), done(false, 0.0));

        // t0 already scored (from a checkpoint). limit=1 → run exactly ONE new
        // task (t1), skip t0, leave t2/t3 for a later resume.
        let mut observed: Vec<String> = Vec::new();
        let report = run_ab_suite_resumable(
            &tasks,
            &tspec(),
            &cspec(),
            &runner,
            vec![cell("t0")],
            Some(1),
            |c| observed.push(c.task_id.clone()),
        )
        .await;
        assert_eq!(observed, vec!["t1"], "only the ONE new task fires on_cell");
        assert_eq!(report.stats.paired_tasks, 2, "t0 (resumed) + t1 (new)");

        // Resume: feed both done cells back, no limit → runs the remaining t2, t3.
        let mut observed2: Vec<String> = Vec::new();
        let report2 = run_ab_suite_resumable(
            &tasks,
            &tspec(),
            &cspec(),
            &runner,
            vec![cell("t0"), cell("t1")],
            None,
            |c| observed2.push(c.task_id.clone()),
        )
        .await;
        assert_eq!(observed2, vec!["t2", "t3"]);
        assert_eq!(report2.stats.paired_tasks, 4, "all scored once resumed");
    }

    /// The checkpoint is keyed by corpus path alone, so a re-run at a DIFFERENT
    /// backbone would otherwise resume the previous backbone's cells and fold
    /// two models into one report — contamination that looks perfectly clean.
    /// A cell is reusable only when both arms match exactly.
    #[tokio::test]
    async fn resume_refuses_cells_from_a_different_backbone() {
        let tasks: Vec<AbTask> = (0..2).map(|i| task(&format!("t{i}"))).collect();
        let runner = script(vec![
            ("t0", "native", done(false, 0.0)),
            ("t0", "codex", done(false, 0.0)),
            ("t1", "native", done(false, 0.0)),
            ("t1", "codex", done(false, 0.0)),
        ]);
        // A checkpoint written by an earlier gpt-5.5 run…
        let stale = AbCell {
            task_id: "t0".into(),
            treatment: done(true, 0.0),
            control: done(false, 0.0),
            treatment_spec: Some(ArmSpec::native(Some("gpt-5.5".into()))),
            control_spec: Some(ArmSpec::external("codex", Some("gpt-5.5".into()))),
        };
        // …must NOT be reused by a gpt-5.4 run of the same corpus.
        let t54 = ArmSpec::native(Some("gpt-5.4".into()));
        let c54 = ArmSpec::external("codex", Some("gpt-5.4".into()));
        let mut observed = Vec::new();
        let report = run_ab_suite_resumable(&tasks, &t54, &c54, &runner, vec![stale], None, |c| {
            observed.push(c.task_id.clone())
        })
        .await;
        assert_eq!(observed, vec!["t0", "t1"], "the stale cell is re-run");
        assert_eq!(report.stats.paired_tasks, 2);
        assert_eq!(
            report.stats.treatment_passes, 0,
            "the gpt-5.5 cell's pass must not leak into the gpt-5.4 report"
        );
        assert!(report.cells.iter().all(|c| c.matches(&t54, &c54)));

        // Legacy cells (no provenance) are unknown, hence never reusable.
        let legacy = AbCell {
            task_id: "t0".into(),
            treatment: done(true, 0.0),
            control: done(false, 0.0),
            treatment_spec: None,
            control_spec: None,
        };
        let report2 =
            run_ab_suite_resumable(&tasks, &t54, &c54, &runner, vec![legacy], None, |_| {}).await;
        assert_eq!(report2.stats.treatment_passes, 0);
    }

    /// A same-backbone pair is a harness delta; a cross-tier pair is not, and the
    /// report must say so rather than let the diagonal read as a harness win.
    #[test]
    fn same_backbone_is_checked_not_asserted() {
        let paired = |t: ArmSpec, c: ArmSpec| AbReport::from_cells(vec![], t, c);
        assert_eq!(
            paired(
                ArmSpec::native(Some("gpt-5.5".into())),
                ArmSpec::external("codex", Some("gpt-5.5".into()))
            )
            .same_backbone(),
            Some(true)
        );
        // Different pins: deliberate (the diagonal), and never a harness delta.
        let diagonal = paired(
            ArmSpec::native(Some("gpt-5.4".into())),
            ArmSpec::external("codex", Some("gpt-5.5".into())),
        );
        assert_eq!(diagonal.same_backbone(), Some(false));
        assert!(diagonal.summary_line().contains("DIFFERENT PINS"));
        // An unpinned arm makes the invariant unknowable — never silently "true".
        let unpinned = paired(
            ArmSpec::native(None),
            ArmSpec::external("codex", Some("gpt-5.5".into())),
        );
        assert_eq!(unpinned.same_backbone(), None);
        assert!(unpinned.summary_line().contains("UNPINNED"));
    }

    #[tokio::test]
    async fn strong_discordance_is_significant() {
        // b=10, c=0 → chi2 = (10-1)^2/10 = 8.1 > 3.841 → significant.
        let tasks: Vec<AbTask> = (0..10).map(|i| task(&format!("t{i}"))).collect();
        let mut entries = Vec::new();
        for i in 0..10 {
            let id = format!("t{i}");
            entries.push((id.clone(), "native".to_string(), done(true, 0.0)));
            entries.push((id.clone(), "codex".to_string(), done(false, 0.0)));
        }
        let runner = Scripted(entries.into_iter().map(|(t, a, o)| ((t, a), o)).collect());
        let report = run_ab_suite(&tasks, &tspec(), &cspec(), &runner).await;
        assert!((report.stats.mcnemar_chi2 - 8.1).abs() < 1e-9);
        assert!(report.stats.mcnemar_significant_05);
    }

    #[tokio::test]
    async fn infra_failures_are_excluded_from_the_denominator() {
        // 3 tasks; on t2 the external arm never ran (infra). That whole cell
        // drops from the paired denominator — not scored as an external loss.
        let tasks: Vec<AbTask> = (0..3).map(|i| task(&format!("t{i}"))).collect();
        let runner = script(vec![
            ("t0", "native", done(true, 0.0)),
            ("t0", "codex", done(true, 0.0)),
            ("t1", "native", done(true, 0.0)),
            ("t1", "codex", done(false, 0.0)),
            ("t2", "native", done(true, 0.0)),
            ("t2", "codex", ArmOutcome::infra("codex launch failed", 5)),
        ]);
        let report = run_ab_suite(&tasks, &tspec(), &cspec(), &runner).await;
        let s = &report.stats;
        assert_eq!(s.paired_tasks, 2, "t2 excluded — external infra failure");
        assert_eq!(s.control_infra_failures, 1);
        assert_eq!(s.treatment_passes, 2);
        assert_eq!(s.control_passes, 1);
    }

    #[tokio::test]
    async fn empty_suite_is_well_defined() {
        let report = run_ab_suite(&[], &tspec(), &cspec(), &NoRunner).await;
        assert_eq!(report.stats.paired_tasks, 0);
        assert_eq!(report.stats.treatment_pass_rate, 0.0);
        assert!(!report.stats.mcnemar_significant_05);
        assert_eq!(report.stats.treatment_cost_per_pass, None);
    }

    struct NoRunner;
    #[async_trait]
    impl AbArmRunner for NoRunner {
        async fn run_arm(&self, _task: &AbTask, _arm: &ArmSpec) -> ArmOutcome {
            ArmOutcome::infra("unused", 0)
        }
    }

    #[test]
    fn summary_line_is_readable() {
        let cells = vec![cell_of("t0", done(true, 0.1), done(false, 0.2))];
        let report = AbReport::from_cells(cells, tspec(), cspec());
        let line = report.summary_line();
        assert!(line.contains("native@parslee/reasoning 1/1"));
        assert!(line.contains("codex@parslee/reasoning 0/1"));
        // Both arms pinned to one model → an honest harness delta.
        assert!(line.contains("same-backbone harness delta"));
    }

    // --- Slice 3: attribution ---

    /// A native ArmOutcome that lost, tagged with a transcript key.
    fn lost_with_transcript(key: &str) -> ArmOutcome {
        ArmOutcome {
            passed: false,
            iterations: 3,
            cost_usd: 0.0,
            wall_ms: 10,
            error: None,
            transcript_path: Some(key.into()),
            infra_failed: false,
        }
    }

    /// A JSONL event stream with `n` runtime failures of `action` — a
    /// trajectory-regulation signature `diagnose` recognizes at `min≤n`.
    fn failing_transcript(action: &str, err: &str, n: usize) -> String {
        (0..n)
            .map(|_| {
                format!(
                    r#"{{"kind":"action_failed","action_id":"{action}","data":{{"error":"{err}"}}}}"#
                )
            })
            .collect::<Vec<_>>()
            .join("\n")
    }

    fn report_with(cells: Vec<AbCell>) -> AbReport {
        AbReport::from_cells(cells, tspec(), cspec())
    }

    #[test]
    fn treatment_loss_with_recurring_failure_is_harness_addressable() {
        // control won → a CAR-specific harness gap
        let cells = vec![cell_of(
            "t0",
            lost_with_transcript("t0.jsonl"),
            done(true, 0.0),
        )];
        let report = report_with(cells);
        let events: HashMap<String, String> = [(
            "t0.jsonl".to_string(),
            failing_transcript("run_command", "exited 1 at runtime", 3),
        )]
        .into_iter()
        .collect();
        let attr = attribute_round(&report, |p| events.get(p).cloned(), 2);
        assert_eq!(attr.harness_addressable_losses, vec!["t0".to_string()]);
        assert!(attr.backbone_bound_losses.is_empty());
        assert!(attr.has_lever());
        assert_eq!(attr.interventions.len(), 1);
        assert_eq!(
            attr.interventions[0].layer,
            InterventionLayer::TrajectoryRegulation
        );
        assert_eq!(attr.interventions[0].evidence_count, 3);
    }

    #[test]
    fn clean_treatment_loss_is_backbone_bound() {
        // No failure events — the arm ran clean and simply produced a wrong
        // answer. No harness lever; parked as a backbone signal.
        let cells = vec![cell_of(
            "t1",
            lost_with_transcript("t1.jsonl"),
            done(false, 0.0),
        )];
        let report = report_with(cells);
        let events: HashMap<String, String> = [("t1.jsonl".to_string(), String::new())]
            .into_iter()
            .collect();
        let attr = attribute_round(&report, |p| events.get(p).cloned(), 2);
        assert!(attr.harness_addressable_losses.is_empty());
        assert_eq!(attr.backbone_bound_losses, vec!["t1".to_string()]);
        assert!(!attr.has_lever());
    }

    #[test]
    fn interventions_merge_and_rank_across_losing_cells() {
        // Two losing cells fail the SAME action → evidence sums, one merged
        // intervention. A third cell fails a different action → ranked below.
        let cells = vec![
            cell_of("a", lost_with_transcript("a.jsonl"), done(true, 0.0)),
            cell_of("b", lost_with_transcript("b.jsonl"), done(true, 0.0)),
            cell_of("c", lost_with_transcript("c.jsonl"), done(true, 0.0)),
            // A passed treatment cell and an infra cell must be ignored.
            cell_of("d", done(true, 0.0), done(true, 0.0)),
        ];
        let report = report_with(cells);
        let events: HashMap<String, String> = [
            (
                "a.jsonl".to_string(),
                failing_transcript("run_command", "runtime boom", 2),
            ),
            (
                "b.jsonl".to_string(),
                failing_transcript("run_command", "runtime boom", 3),
            ),
            (
                "c.jsonl".to_string(),
                failing_transcript("edit_file", "runtime splat", 2),
            ),
        ]
        .into_iter()
        .collect();
        let attr = attribute_round(&report, |p| events.get(p).cloned(), 2);
        assert_eq!(attr.harness_addressable_losses.len(), 3);
        assert_eq!(
            attr.interventions.len(),
            2,
            "run_command merged, edit_file distinct"
        );
        // run_command evidence = 2 + 3 = 5, ranked first.
        assert_eq!(attr.interventions[0].target, "run_command");
        assert_eq!(attr.interventions[0].evidence_count, 5);
        assert_eq!(attr.interventions[1].target, "edit_file");
    }
}