bundle/v0/tests/test_full_matrix_analysis.py

from __future__ import annotations

import copy
import hashlib
import json
from pathlib import Path

import pytest

from loss_aware_dro_repro.batch_control import DENIED_AUTHORITY
from loss_aware_dro_repro.core import (
    ContractError,
    canonical_bytes,
    load_json,
    load_plan,
    plan_hash,
    sha256_value,
)
from loss_aware_dro_repro.full_matrix_analysis import (
    ANALYSIS_CONFIG_PATH,
    _validate_external_gradient_receipt,
    _validate_theorem_receipt,
    analyze_full_matrix_run,
    analyze_validated_full_matrix,
)
from loss_aware_dro_repro.matrix import expand_tasks


def _run_identity(task_id: str) -> str:
    return "sha256:" + hashlib.sha256(task_id.encode("utf-8")).hexdigest()


@pytest.fixture(scope="module")
def full_matrix_fixture() -> tuple[dict, dict, dict, list[dict], dict]:
    plan = load_plan()
    tasks = sorted(expand_tasks(plan), key=lambda task: task["task_id"])
    rows = []
    manifest_tasks = []
    for task in tasks:
        suite = task["suite"]
        coverage = suite != "portfolio_gaussian_coverage_ablation"
        run_identity = _run_identity(task["task_id"])
        rows.append(
            {
                "task_id": task["task_id"],
                "task_hash": task["task_hash"],
                "run_identity": run_identity,
                "plan_hash": plan_hash(plan),
                "dataset": {
                    "suite": suite,
                    "distribution_id": task["distribution_id"],
                    "replicate": task["replicate"],
                    "sample_size": task["sample_size"],
                },
                "solver": {"status": "optimal"},
                "optimization": {
                    "iterations": 25,
                    "initial_stationarity": 2.0,
                    "final_stationarity": 1.0,
                    "stopping": {
                        "reason": "paper_total_penalized_objective_improvement_below_tolerance"
                    },
                },
                "metrics": {
                    "relative_worst_case_improvement": 0.10,
                    "relative_oos_improvement": 0.08,
                    "coverage_final": coverage,
                },
            }
        )
        manifest_tasks.append(
            {
                "task_id": task["task_id"],
                "task_hash": task["task_hash"],
                "scientific_run_identity": run_identity,
            }
        )
    manifest = {
        "plan_hash": plan_hash(plan),
        "manifest_hash": "sha256:" + "1" * 64,
        "repository": {
            "commit": "a" * 40,
            "lane_source_hash": "sha256:" + "b" * 64,
        },
        "selection": {"mode": "all", "shard_count": 1, "shard_index": 0},
        "execution": {
            "requested_execution_config": {
                "schema_version": 2,
                "execution_scale": "stopping_storage_sample",
                "max_outer_iterations": 5000,
            }
        },
        "tasks": manifest_tasks,
    }
    result_bytes = b"".join(canonical_bytes(row) + b"\n" for row in rows)
    result_hash = hashlib.sha256(result_bytes).hexdigest()
    aggregate_identity = sha256_value(
        {
            "manifest_hash": manifest["manifest_hash"],
            "included": [(row["task_id"], row["run_identity"]) for row in rows],
            "results_sha256": result_hash,
        }
    )
    aggregate = {
        "schema_version": 1,
        "aggregate_identity": aggregate_identity,
        "manifest_hash": manifest["manifest_hash"],
        "expected_task_count": len(tasks),
        "validated_success_count": len(rows),
        "complete": True,
        "rejected": [],
        "artifact_hashes": {"results.jsonl": result_hash},
        "issuance_pass_receipt_sha256": None,
        "authority": DENIED_AUTHORITY,
        "scientific_verdicts": {"C1": "HOLD", "C2": "HOLD", "C3": "HOLD"},
    }
    contract = copy.deepcopy(load_json(ANALYSIS_CONFIG_PATH))
    contract["uncertainty"]["bootstrap_replicates"] = 100
    return plan, manifest, aggregate, rows, contract


def _aggregate_for_rows(manifest: dict, rows: list[dict]) -> dict:
    result_bytes = b"".join(canonical_bytes(row) + b"\n" for row in rows)
    result_hash = hashlib.sha256(result_bytes).hexdigest()
    return {
        "schema_version": 1,
        "aggregate_identity": sha256_value(
            {
                "manifest_hash": manifest["manifest_hash"],
                "included": [
                    (row.get("task_id"), row.get("run_identity")) for row in rows
                ],
                "results_sha256": result_hash,
            }
        ),
        "manifest_hash": manifest["manifest_hash"],
        "expected_task_count": 14000,
        "validated_success_count": len(rows),
        "complete": len(rows) == 14000,
        "rejected": (
            []
            if len(rows) == 14000
            else [{"task_id": "missing", "reason": "synthetic missing fixture"}]
        ),
        "artifact_hashes": {"results.jsonl": result_hash},
        "issuance_pass_receipt_sha256": None,
        "authority": DENIED_AUTHORITY,
        "scientific_verdicts": {"C1": "HOLD", "C2": "HOLD", "C3": "HOLD"},
    }


def _analyze(
    fixture: tuple[dict, dict, dict, list[dict], dict],
    *,
    rows: list[dict] | None = None,
    aggregate: dict | None = None,
    contract: dict | None = None,
) -> dict:
    plan, manifest, fixture_aggregate, fixture_rows, fixture_contract = fixture
    selected_rows = fixture_rows if rows is None else rows
    selected_aggregate = (
        fixture_aggregate
        if aggregate is None and rows is None
        else aggregate or _aggregate_for_rows(manifest, selected_rows)
    )
    return analyze_validated_full_matrix(
        manifest,
        selected_aggregate,
        selected_rows,
        plan=plan,
        analysis_contract=contract or fixture_contract,
    )


def _write_gradient_receipt(
    tmp_path: Path, plan: dict, manifest: dict
) -> tuple[dict, Path]:
    selectors = next(
        suite["task_selectors"]
        for suite in plan["diagnostic_suites"]
        if suite["name"] == "hypergradient_finite_difference"
    )
    checks = []
    for index, task_id in enumerate(selectors):
        artifact_payload = {
            "task_id": task_id,
            "step": 0.0001,
            "analytic": [1.0, 2.0],
            "finite_difference": [1.0, 2.0],
            "absolute_error": 0.0,
            "relative_error": 0.0,
        }
        artifact_path = tmp_path / f"gradient-{index}.json"
        artifact_path.write_text(json.dumps(artifact_payload), encoding="utf-8")
        checks.append(
            {
                **artifact_payload,
                "passed": True,
                "artifact_path": artifact_path.name,
                "artifact_sha256": hashlib.sha256(artifact_path.read_bytes()).hexdigest(),
            }
        )
    receipt = {
        "schema_version": 1,
        "kind": "independent_finite_difference_validation",
        "plan_hash": plan_hash(plan),
        "implementation": {
            "commit": manifest["repository"]["commit"],
            "source_tree_hash": manifest["repository"]["lane_source_hash"],
            "working_tree_clean": True,
        },
        "command": ["python", "scripts/run_hypergradient_validation.py"],
        "error_definition": {
            "absolute": "max_absolute_component_difference",
            "relative": "l2_difference_over_max_l2_analytic_l2_finite_difference_1e-12",
            "pass_rule": "absolute_error_le_threshold_or_relative_error_le_threshold",
        },
        "thresholds": {"absolute_error_max": 0.000001, "relative_error_max": 0.001},
        "task_selectors": selectors,
        "checks": checks,
        "all_pass": True,
    }
    receipt_path = tmp_path / "gradient-receipt.json"
    receipt_path.write_text(json.dumps(receipt), encoding="utf-8")
    return receipt, receipt_path


def test_direct_synthetic_transform_is_diagnostic_only_and_distribution_first(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    first = _analyze(full_matrix_fixture)
    second = _analyze(full_matrix_fixture)
    assert first == second
    assert first["claim_eligibility"]["eligible"] is False
    assert first["completeness"]["complete"] is False
    assert first["completeness"]["structurally_complete"] is True
    assert first["claim_verdicts"] == {
        claim_id: "inconclusive" for claim_id in ("A1", "A2", "A3", "A4", "A5", "A6")
    }
    assert list(first["diagnostic_candidate_verdicts"]) == [
        "A1", "A2", "A3", "A4", "A5", "A6"
    ]
    assert first["diagnostic_candidate_verdicts"]["A6"] == "verified"
    assert first["diagnostic_candidate_verdicts"]["A2"] == "inconclusive"
    point = first["A6"]["improvement_reporting_points"][
        "regression_absolute_main/n010"
    ]
    assert point["aggregation_unit"] == "distribution_id"
    assert point["distribution_count"] == 10
    assert point["observation_count"] == 100
    assert point["ci"]["replicates"] == 100
    assert point["mean"] == pytest.approx(0.08)


def test_all_forged_row_ids_fail_exact_membership(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    rows = copy.deepcopy(full_matrix_fixture[3])
    for index, row in enumerate(rows):
        row["task_id"] = f"forged/{index:05d}"
    analysis = _analyze(full_matrix_fixture, rows=rows)
    assert analysis["completeness"]["structurally_complete"] is False
    assert any(
        "aggregate task identity/hash pairs" in issue
        for issue in analysis["completeness"]["binding_issues"]
    )
    assert analysis["claim_verdicts"]["A1"] == "inconclusive"
    assert analysis["claim_verdicts"]["A6"] == "inconclusive"


def test_forged_aggregate_hash_and_identity_fail_binding(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    aggregate = copy.deepcopy(full_matrix_fixture[2])
    aggregate["aggregate_identity"] = "sha256:" + "f" * 64
    aggregate["artifact_hashes"]["results.jsonl"] = "0" * 64
    analysis = _analyze(full_matrix_fixture, aggregate=aggregate)
    issues = analysis["completeness"]["binding_issues"]
    assert any("results hash" in issue for issue in issues)
    assert any("aggregate identity" in issue for issue in issues)
    assert analysis["completeness"]["structurally_complete"] is False


def test_boolean_schema_version_and_unknown_config_key_fail_closed(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    boolean = copy.deepcopy(full_matrix_fixture[4])
    boolean["schema_version"] = True
    with pytest.raises(ContractError, match="schema violation"):
        _analyze(full_matrix_fixture, contract=boolean)
    extra = copy.deepcopy(full_matrix_fixture[4])
    extra["trust_me"] = True
    with pytest.raises(ContractError, match="schema violation"):
        _analyze(full_matrix_fixture, contract=extra)


def test_cap_hit_is_censoring_and_candidate_verdict_is_partial_with_improvement(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    rows = copy.deepcopy(full_matrix_fixture[3])
    target = next(row for row in rows if row["dataset"]["suite"] == "portfolio_gaussian_main")
    target["optimization"]["iterations"] = 5000
    target["optimization"]["stopping"]["reason"] = "maximum_outer_iterations_reached"
    analysis = _analyze(full_matrix_fixture, rows=rows)
    assert analysis["diagnostic_candidate_verdicts"]["A1"] == "partially_verified"
    assert analysis["diagnostic_candidate_verdicts"]["A5"] == "partially_verified"
    assert analysis["stopping"]["overall"]["counts"]["censored_at_5000"] == 1
    assert analysis["claim_verdicts"]["A5"] == "inconclusive"


def test_cap_plus_systematic_negative_improvement_is_not_coverage_only_partial(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    rows = copy.deepcopy(full_matrix_fixture[3])
    for row in rows:
        if row["dataset"]["suite"] != "portfolio_gaussian_coverage_ablation":
            row["metrics"]["relative_worst_case_improvement"] = -0.2
            row["metrics"]["relative_oos_improvement"] = -0.1
    target = next(row for row in rows if row["dataset"]["suite"] == "portfolio_gaussian_main")
    target["optimization"]["iterations"] = 5000
    target["optimization"]["stopping"]["reason"] = "maximum_outer_iterations_reached"
    analysis = _analyze(full_matrix_fixture, rows=rows)
    assert analysis["A5"]["criteria"]["all_lower_bounds_at_target"] is True
    assert analysis["diagnostic_candidate_verdicts"]["A4"] == "inconclusive"


def test_worsening_stop_is_not_labeled_convergence(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    rows = copy.deepcopy(full_matrix_fixture[3])
    target = next(row for row in rows if row["dataset"]["suite"] == "regression_absolute_main")
    target["optimization"]["stopping"]["reason"] = "paper_total_penalized_objective_worsened"
    analysis = _analyze(full_matrix_fixture, rows=rows)
    assert analysis["diagnostic_candidate_verdicts"]["A6"] == "partially_verified"
    assert analysis["stopping"]["overall"]["counts"]["stopped_due_to_objective_worsening"] == 1


def test_closed_numeric_gradient_receipt_passes_and_minimal_assertion_fails(
    tmp_path: Path,
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict],
) -> None:
    plan, manifest = full_matrix_fixture[0], full_matrix_fixture[1]
    receipt, receipt_path = _write_gradient_receipt(tmp_path, plan, manifest)
    status = _validate_external_gradient_receipt(
        receipt, plan, receipt_path=receipt_path, manifest=manifest
    )
    assert status["status"] == "pass"
    assert status["all_pass"] is True
    diagnostic = analyze_validated_full_matrix(
        manifest,
        full_matrix_fixture[2],
        full_matrix_fixture[3],
        plan=plan,
        analysis_contract=full_matrix_fixture[4],
        gradient_receipt=receipt,
        gradient_receipt_path=receipt_path,
    )
    assert diagnostic["A3"]["external_finite_difference_receipt"]["status"] == "pass"
    assert diagnostic["diagnostic_candidate_verdicts"]["A3"] == "partially_verified"
    assert diagnostic["A3"]["evidence_scope"]["source_formula_audit"] == "missing"
    assert diagnostic["A3"]["evidence_scope"]["nonsmooth_boundary_evidence"] == "missing"
    assert diagnostic["claim_eligibility"]["eligible"] is False
    assert diagnostic["claim_verdicts"]["A3"] == "inconclusive"
    minimal = {
        "plan_hash": plan_hash(plan),
        "all_pass": True,
        "task_selectors": receipt["task_selectors"],
    }
    status = _validate_external_gradient_receipt(
        minimal, plan, receipt_path=receipt_path, manifest=manifest
    )
    assert status["status"] == "fail"
    assert status["all_pass"] is False


def test_a2_stays_inconclusive_without_strict_theorem_receipt(
    tmp_path: Path,
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict],
) -> None:
    plan, manifest = full_matrix_fixture[0], full_matrix_fixture[1]
    analysis = _analyze(full_matrix_fixture)
    assert analysis["diagnostic_candidate_verdicts"]["A2"] == "inconclusive"
    assert analysis["A2"]["theorem_receipt"]["status"] == "missing"

    forged = {"kind": "mathematical_theorem_verification", "verdict": "verified"}
    forged_path = tmp_path / "forged-theorem.json"
    forged_path.write_text(json.dumps(forged), encoding="utf-8")
    status = _validate_theorem_receipt(
        forged, receipt_path=forged_path, manifest=manifest
    )
    assert status["status"] == "fail"
    assert status["verified"] is False

    valid = {
        "schema_version": 1,
        "kind": "mathematical_theorem_verification",
        "claim_id": "A2",
        "openreview_id": "K1EPPO9t2c",
        "paper_sha256": plan["paper"]["paper_sha256"],
        "theorem_text": "Theorem 5.1 establishes that the proposed hypergradient descent procedure converges to a critical point of the bilevel problem under mild conditions, including square-summable step sizes (Section 5.1, Theorem 5.1).",
        "implementation": {
            "commit": manifest["repository"]["commit"],
            "source_tree_hash": manifest["repository"]["lane_source_hash"],
            "working_tree_clean": True,
        },
        "assumptions": [
            {"name": "all theorem assumptions", "status": "proved", "evidence": "independent line-by-line proof audit"}
        ],
        "proof_audit": {
            "complete": True,
            "counterexample_search": "No counterexample satisfying every assumption was found.",
            "reviewer": "independent mathematical reviewer",
            "reviewed_at": "2026-07-20T08:00:00Z",
        },
        "verdict": "verified",
    }
    valid_path = tmp_path / "valid-theorem.json"
    valid_path.write_text(json.dumps(valid), encoding="utf-8")
    status = _validate_theorem_receipt(
        valid, receipt_path=valid_path, manifest=manifest
    )
    assert status["status"] == "verification_disabled"
    assert status["structurally_valid"] is True
    assert status["verified"] is False
    analysis = analyze_validated_full_matrix(
        manifest,
        full_matrix_fixture[2],
        full_matrix_fixture[3],
        plan=plan,
        analysis_contract=full_matrix_fixture[4],
        theorem_receipt=valid,
        theorem_receipt_path=valid_path,
    )
    assert analysis["diagnostic_candidate_verdicts"]["A2"] == "inconclusive"
    assert analysis["A2"]["theorem_receipt"]["verified"] is False


def test_a4_negative_slopes_cannot_promote_without_exact_setup_and_estimand_binding(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    rows = copy.deepcopy(full_matrix_fixture[3])
    for row in rows:
        if row["dataset"]["suite"] == "portfolio_gaussian_main":
            sample_size = row["dataset"]["sample_size"]
            row["metrics"]["relative_worst_case_improvement"] = (110 - sample_size) / 100
            row["metrics"]["relative_oos_improvement"] = (110 - sample_size) / 200
    analysis = _analyze(full_matrix_fixture, rows=rows)
    assert analysis["diagnostic_candidate_verdicts"]["A4"] == "inconclusive"
    setup = analysis["A4"]["anchored_setup_audit"]
    assert setup["exact_binding_pass"] is False
    assert setup["k"]["status"] == "unbound"
    assert setup["J"]["status"] == "unbound"
    assert setup["n_b"] == {"status": "match", "observed": 20}
    assert setup["relative_improvement_estimand"]["status"] == "unbound"
    sensitivity = analysis["extensions"]["a4_parameter_mismatch_sensitivity"]
    assert sensitivity["role"] == "extension_only"
    assert sensitivity["all_metric_slope_upper_bounds_negative"] is True
    assert all(
        summary["ci"]["upper"] < 0.0
        for summary in sensitivity["sample_size_slopes"].values()
    )


def test_a6_consistency_gate_rejects_one_negative_trial_even_when_ci_is_positive(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    rows = copy.deepcopy(full_matrix_fixture[3])
    for row in rows:
        if (
            row["dataset"]["suite"] == "regression_absolute_main"
            and row["dataset"]["distribution_id"] == 1
        ):
            row["metrics"]["relative_oos_improvement"] = -0.01
    analysis = _analyze(full_matrix_fixture, rows=rows)
    assert analysis["A6"]["criteria"]["all_lower_bounds_positive"] is True
    assert analysis["A6"]["criteria"]["every_one_of_ten_trial_means_positive"] is False
    assert analysis["diagnostic_candidate_verdicts"]["A6"] == "partially_verified"


def test_gradient_artifact_tamper_fails_recomputation(
    tmp_path: Path,
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict],
) -> None:
    plan, manifest = full_matrix_fixture[0], full_matrix_fixture[1]
    receipt, receipt_path = _write_gradient_receipt(tmp_path, plan, manifest)
    (tmp_path / receipt["checks"][0]["artifact_path"]).write_text("{}", encoding="utf-8")
    status = _validate_external_gradient_receipt(
        receipt, plan, receipt_path=receipt_path, manifest=manifest
    )
    assert status["status"] == "fail"
    assert status["check_statuses"][0]["artifact_hash_matches"] is False


def test_uncensored_systematic_negative_effect_is_falsified_and_ids_retained(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    rows = copy.deepcopy(full_matrix_fixture[3])
    first_negative_id = None
    for row in rows:
        if row["dataset"]["suite"] != "portfolio_gaussian_coverage_ablation":
            row["metrics"]["relative_worst_case_improvement"] = -0.10
            row["metrics"]["relative_oos_improvement"] = -0.05
            if row["dataset"]["suite"] == "regression_absolute_main":
                first_negative_id = first_negative_id or row["task_id"]
    analysis = _analyze(full_matrix_fixture, rows=rows)
    assert analysis["diagnostic_candidate_verdicts"]["A6"] == "falsified"
    all_retained_ids = {
        task_id
        for point in analysis["A6"]["improvement_reporting_points"].values()
        for task_id in point["nonpositive_task_ids"]
    }
    assert first_negative_id in all_retained_ids


def test_supplementary_c3_failures_and_caps_block_c3_verification(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    rows = copy.deepcopy(full_matrix_fixture[3])
    supplementary = {
        "portfolio_gaussian_highdim",
        "portfolio_discrete",
        "portfolio_gmm",
        "regression_squared",
    }
    count = 0
    for row in rows:
        if row["dataset"]["suite"] in supplementary:
            count += 1
            row["metrics"]["relative_worst_case_improvement"] = -1.0
            row["metrics"]["relative_oos_improvement"] = -1.0
            row["optimization"]["iterations"] = 5000
            row["optimization"]["stopping"]["reason"] = "maximum_outer_iterations_reached"
    analysis = _analyze(full_matrix_fixture, rows=rows)
    assert count == 3500
    assert analysis["diagnostic_candidate_verdicts"]["A5"] == "verified"
    assert analysis["diagnostic_candidate_verdicts"]["A6"] == "verified"
    assert analysis["extensions"]["scored"] is False
    reported_suites = {
        key.split("/")[0]
        for key in analysis["extensions"]["supplementary_improvement_reporting_points"]
    }
    assert supplementary <= reported_suites


def test_missing_output_and_unknown_stop_fail_closed(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    missing = copy.deepcopy(full_matrix_fixture[3][1:])
    analysis = _analyze(full_matrix_fixture, rows=missing)
    assert analysis["completeness"]["structurally_complete"] is False
    assert analysis["diagnostic_candidate_verdicts"]["A1"] == "inconclusive"
    assert analysis["diagnostic_candidate_verdicts"]["A6"] == "inconclusive"
    unknown = copy.deepcopy(full_matrix_fixture[3])
    unknown[0]["optimization"]["stopping"]["reason"] = "looks_converged"
    with pytest.raises(ContractError, match="unknown stopping reason"):
        _analyze(full_matrix_fixture, rows=unknown)


def _runtime_aggregate_fixture(tmp_path: Path) -> tuple[Path, Path, dict, dict, Path]:
    manifest_hash = "sha256:" + "1" * 64
    manifest_path = tmp_path / "manifest.json"
    manifest_path.write_text(json.dumps({"manifest_hash": manifest_hash}), encoding="utf-8")
    output_root = tmp_path / "output"
    row = {"task_id": "synthetic", "run_identity": "sha256:" + "2" * 64}
    result_bytes = canonical_bytes(row) + b"\n"
    result_hash = hashlib.sha256(result_bytes).hexdigest()
    aggregate_identity = sha256_value(
        {
            "manifest_hash": manifest_hash,
            "included": [(row["task_id"], row["run_identity"])],
            "results_sha256": result_hash,
        }
    )
    aggregate_root = output_root / "aggregates" / aggregate_identity.removeprefix("sha256:")
    aggregate_root.mkdir(parents=True)
    results_path = aggregate_root / "results.jsonl"
    results_path.write_bytes(result_bytes)
    summary = {
        "schema_version": 1,
        "aggregate_identity": aggregate_identity,
        "manifest_hash": manifest_hash,
        "expected_task_count": 1,
        "validated_success_count": 1,
        "rejected": [],
        "complete": True,
        "artifact_hashes": {"results.jsonl": result_hash},
        "issuance_pass_receipt_sha256": None,
        "authority": DENIED_AUTHORITY,
        "scientific_verdicts": {"C1": "HOLD", "C2": "HOLD", "C3": "HOLD"},
    }
    summary_path = aggregate_root / "summary.json"
    summary_path.write_text(json.dumps(summary), encoding="utf-8")
    returned = {
        **summary,
        "aggregate_path": aggregate_root.as_posix(),
        "summary_file_sha256": hashlib.sha256(summary_path.read_bytes()).hexdigest(),
    }
    return manifest_path, output_root, summary, returned, results_path


def test_runtime_path_constructs_private_ordinary_validation_context(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    import loss_aware_dro_repro.full_matrix_analysis as module

    manifest_path, output_root, _, returned, _ = _runtime_aggregate_fixture(tmp_path)
    monkeypatch.setattr(module, "validate_batch_manifest", lambda manifest: None)
    monkeypatch.setattr(module, "aggregate_validated_successes", lambda manifest, output: returned)
    monkeypatch.setattr(
        module,
        "_analysis_source_provenance",
        lambda: {
            "commit": "a" * 40,
            "clean": True,
            "source_tree_hash": "sha256:" + "b" * 64,
            "files": {"analysis.py": "c" * 64},
        },
    )

    def capture(manifest: dict, summary: dict, rows: list[dict], **kwargs: object) -> dict:
        context = kwargs.get("_ordinary_validation_context")
        assert context is not None
        assert getattr(context, "sentinel") is module._ORDINARY_VALIDATION_SENTINEL
        return {"context_received": True, "rows": rows}

    monkeypatch.setattr(module, "analyze_validated_full_matrix", capture)
    result = analyze_full_matrix_run(manifest_path, output_root)
    assert result == {
        "context_received": True,
        "rows": [{"run_identity": "sha256:" + "2" * 64, "task_id": "synthetic"}],
    }


def test_runtime_path_rejects_tampered_and_noncanonical_aggregate(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    import loss_aware_dro_repro.full_matrix_analysis as module

    manifest_path, output_root, _, returned, results_path = _runtime_aggregate_fixture(tmp_path)
    monkeypatch.setattr(module, "validate_batch_manifest", lambda manifest: None)
    monkeypatch.setattr(module, "aggregate_validated_successes", lambda manifest, output: returned)
    results_path.write_text("{}\n", encoding="utf-8")
    with pytest.raises(ContractError, match="results hash"):
        analyze_full_matrix_run(manifest_path, output_root)


def test_custom_contract_provenance_does_not_impersonate_committed_file(
    full_matrix_fixture: tuple[dict, dict, dict, list[dict], dict]
) -> None:
    analysis = _analyze(full_matrix_fixture)
    assert analysis["analysis_contract"]["source"] == "caller_supplied_diagnostic_override"
    assert analysis["analysis_contract"]["path"] is None
    assert analysis["analysis_contract"]["file_sha256"] is None
    assert analysis["analysis_contract"]["matches_committed_file"] is False