bundle/v0/tests/test_stopping_sample.py

from __future__ import annotations

import gzip
import hashlib
import json
from collections import Counter, defaultdict
from pathlib import Path

import pytest

from loss_aware_dro_repro import stopping_sample as sample_module
from loss_aware_dro_repro import batch_control as batch_module
from loss_aware_dro_repro.batch_control import (
    build_stopping_sample_batch_manifest,
    lane_file_snapshot,
)
from loss_aware_dro_repro.core import ContractError, canonical_bytes, load_plan, sha256_value
from loss_aware_dro_repro.empirical_task import (
    normalize_empirical_execution_config,
    run_empirical_task,
)
from loss_aware_dro_repro.matrix import expand_tasks
from loss_aware_dro_repro.paper_task import normalize_execution_config, run_gaussian_task
from loss_aware_dro_repro.stopping_sample import (
    analyze_stopping_sample,
    build_stopping_sample_receipt,
    load_stopping_sample_manifest,
    select_stopping_sample,
    validate_stopping_sample,
)
from loss_aware_dro_repro.trace_streams import IncrementalJsonlWriter


EXPECTED_HASH = "78ef2f219e75ec50fae0b2dffab75781e9b4dd36e9327a77b592404341c000be"


def _task_list_hash(tasks):
    payload = json.dumps(
        [task["task_id"] for task in tasks],
        ensure_ascii=False,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def _receipt(task, *, runtime=10.0, bytes_total=1600):
    byte_values = {
        "iteration_trace": bytes_total - 1300,
        "checkpoint_states": 500,
        "result": 300,
        "solver_receipt": 100,
        "stopping_receipt": 100,
        "lineage_receipt": 200,
        "oos_receipt": 100,
    }
    receipt = {
        "schema_version": 1,
        "receipt_type": "stopping_storage_task_diagnostic",
        "builder_contract_version": 1,
        "task_id": task["task_id"],
        "task_hash": task["task_hash"],
        "status": "success",
        "route": "paper_plain_gradient_descent",
        "diagnostic_cap_iterations": 5000,
        "claim_eligible": False,
        "cost_freeze_eligible": False,
        "full_matrix_freeze_eligible": False,
        "authority": {
            "paid_compute": False,
            "remote_compute": False,
            "external_inference": False,
            "push": False,
            "publish": False,
        },
        "run_identity": "sha256:" + "1" * 64,
        "implementation_commit": "3" * 40,
        "plan_hash": load_stopping_sample_manifest()["plan_hash"],
        "execution_config_hash": "sha256:" + "5" * 64,
        "source_tree_hash": "sha256:" + "2" * 64,
        "source_file_hashes": {"src/example.py": "4" * 64},
        "test_only_provenance": True,
        "runtime_seconds": runtime,
        "iterations_executed": 5000,
        "stop_label": "censored_at_diagnostic_cap",
        "stop_reason": "maximum_outer_iterations_reached",
        "artifact_bytes": {**byte_values, "total": bytes_total},
        "artifact_hashes": {name: "a" * 64 for name in byte_values},
        "solver": {"statuses": {"optimal": 5001}, "maximum_residual": 1e-10},
    }
    receipt["artifact_set_binding"] = sample_module._artifact_set_binding(receipt)
    return receipt


@pytest.fixture(scope="module")
def selected_tasks():
    return select_stopping_sample()


@pytest.fixture(scope="module")
def complete_receipts(selected_tasks):
    return [_receipt(task) for task in selected_tasks]


def _analyze(receipts, **kwargs):
    return analyze_stopping_sample(
        receipts,
        _test_provenance_override={"mode": "synthetic_unit_receipts_v1"},
        **kwargs,
    )


def _write_json(path: Path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(canonical_bytes(value) + b"\n")


def _sha(path: Path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def _production_output(
    root: Path, task, task_entry, *, iterations=1, capped=False
):
    root.mkdir(parents=True)
    run_identity = task_entry["scientific_run_identity"]
    source_tree_hash = task_entry["scientific_source_tree_hash"]
    implementation_commit = task_entry["implementation_commit"]
    execution = task_entry["normalized_execution_config"]
    execution_config_hash = sha256_value(execution)
    trace_path = root / "iteration_trace.jsonl.gz"
    checkpoint_path = root / "checkpoint_states.jsonl.gz"
    with IncrementalJsonlWriter(trace_path) as writer:
        for iteration in range(iterations):
            writer.write(
                {
                    "schema_version": 2,
                    "record_type": "scalar_optimization_iteration",
                    "run_identity": run_identity,
                    "task_id": task["task_id"],
                    "iteration": iteration,
                    "solver_status": "optimal",
                    "solver_residual_maximum": 1e-10,
                }
            )
    with IncrementalJsonlWriter(checkpoint_path) as writer:
        writer.write(
            {
                "schema_version": 2,
                "record_type": "full_optimization_checkpoint",
                "run_identity": run_identity,
                "task_id": task["task_id"],
                "iteration": iterations,
                "checkpoint_reasons": ["terminal"],
                "full_state": {
                    "solver_status": "optimal",
                    "solver_residual_maximum": 2e-10,
                },
            }
        )
    reason = (
        "maximum_outer_iterations_reached"
        if capped
        else "paper_total_penalized_objective_improvement_below_tolerance"
    )
    identity = {
        "run_identity": run_identity,
        "task_hash": task["task_hash"],
        "execution_config_hash": execution_config_hash,
        "source_tree_hash": source_tree_hash,
    }
    solver = {
        "schema_version": 1,
        **identity,
        "status_counts": {"optimal": iterations + 1},
        "all_pass": True,
    }
    stopping = {
        "schema_version": 2,
        **identity,
        "iterations": iterations,
        "stop_reason": reason,
    }
    oos = {"schema_version": 1, "run_identity": run_identity}
    lineage = {
        "schema_version": 1,
        **identity,
        "task_id": task["task_id"],
        "plan_hash": load_stopping_sample_manifest()["plan_hash"],
        "execution_config": execution,
        "source_files": task_entry["scientific_source_files"],
        "implementation_commit": implementation_commit,
        "iteration_trace_bytes": trace_path.stat().st_size,
        "checkpoint_state_bytes": checkpoint_path.stat().st_size,
        "claim_eligible": False,
    }
    for name, value in (
        ("solver_receipt.json", solver),
        ("stopping_receipt.json", stopping),
        ("lineage_receipt.json", lineage),
        ("oos_receipt.json", oos),
    ):
        _write_json(root / name, value)
    artifact_hashes = {
        "iteration_trace": _sha(trace_path),
        "checkpoint_states": _sha(checkpoint_path),
        "solver_receipt": _sha(root / "solver_receipt.json"),
        "stopping_receipt": _sha(root / "stopping_receipt.json"),
        "lineage_receipt": _sha(root / "lineage_receipt.json"),
        "oos_receipt": _sha(root / "oos_receipt.json"),
    }
    result = {
        "schema_version": 2,
        "task_id": task["task_id"],
        "task_hash": task["task_hash"],
        "run_identity": run_identity,
        "plan_hash": load_stopping_sample_manifest()["plan_hash"],
        "status": "success",
        "implementation": {"commit": implementation_commit},
        "runtime": {"duration_seconds": 1.25, "cost_usd": 0.0},
        "scientific_contract": {"stopping_rule": "paper_total_phi_literal"},
        "optimization": {
            "iterations": iterations,
            "stop_reason": reason,
            "iteration_trace_bytes": trace_path.stat().st_size,
            "checkpoint_state_bytes": checkpoint_path.stat().st_size,
        },
        "artifact_hashes": artifact_hashes,
    }
    _write_json(root / "result.json", result)
    return result


@pytest.fixture(scope="module")
def production_batch_manifest(selected_tasks):
    source = lane_file_snapshot()
    snapshot = {
        "commit": "3" * 40,
        "clean": True,
        "lane_source_hash": source["hash"],
        "files": source["files"],
    }
    manifest, _receipt = build_stopping_sample_batch_manifest(
        snapshot_provider=lambda: snapshot,
    )
    return manifest


def _materialize_batch_evidence(
    tmp_path: Path,
    manifest,
    task,
    *,
    iterations=1,
    capped=False,
):
    entry = next(row for row in manifest["tasks"] if row["task_id"] == task["task_id"])
    manifest_path = tmp_path / "batch-manifest.json"
    _write_json(manifest_path, manifest)
    output_root = tmp_path / "batch-output"
    output = output_root / entry["output_relpath"]
    result = _production_output(
        output, task, entry, iterations=iterations, capped=capped
    )
    reservation = {
        "schema_version": 1,
        "event": "reserved",
        "manifest_hash": manifest["manifest_hash"],
        "scientific_run_identity": entry["scientific_run_identity"],
        "task_id": task["task_id"],
        "reserved_at": "2026-07-19T00:00:00Z",
    }
    identity_slug = entry["scientific_run_identity"].removeprefix("sha256:")
    _write_json(
        output_root / "control" / "reservations" / f"{identity_slug}.json",
        reservation,
    )
    success = {
        "schema_version": 1,
        "event": "success",
        "manifest_hash": manifest["manifest_hash"],
        "scientific_run_identity": entry["scientific_run_identity"],
        "task_id": task["task_id"],
        "result_sha256": _sha(output / "result.json"),
        "recorded_at": "2026-07-19T00:00:01Z",
    }
    ledger = output_root / "control" / "launch-ledger.jsonl"
    ledger.parent.mkdir(parents=True, exist_ok=True)
    ledger.write_bytes(canonical_bytes(reservation) + b"\n" + canonical_bytes(success) + b"\n")
    return result, manifest_path, output_root, output


def _materialize_complete_analysis_evidence(
    tmp_path: Path,
    manifest,
    selected,
    receipts,
):
    manifest_path = tmp_path / "batch-manifest.json"
    _write_json(manifest_path, manifest)
    output_root = tmp_path / "batch-output"
    control_root = output_root / "control"
    builder = sample_module.stopping_sample_implementation_binding()
    receipt_by_id = {receipt["task_id"]: dict(receipt) for receipt in receipts}
    launch_events = []
    result_rows = []
    reservations = {}
    successes = {}
    for entry in manifest["tasks"]:
        task_id = entry["task_id"]
        receipt = receipt_by_id[task_id]
        receipt.pop("test_only_provenance")
        receipt.update(
            {
                "run_identity": entry["scientific_run_identity"],
                "implementation_commit": entry["implementation_commit"],
                "execution_config_hash": entry["execution_config_hash"],
                "source_tree_hash": entry["scientific_source_tree_hash"],
                "source_file_hashes": entry["scientific_source_files"],
                "builder_implementation": builder,
            }
        )
        result_row = {
            "run_identity": entry["scientific_run_identity"],
            "task_id": task_id,
        }
        result_bytes = canonical_bytes(result_row) + b"\n"
        receipt["artifact_hashes"] = {
            **receipt["artifact_hashes"],
            "result": hashlib.sha256(result_bytes).hexdigest(),
        }
        reservation = {
            "schema_version": 1,
            "event": "reserved",
            "manifest_hash": manifest["manifest_hash"],
            "scientific_run_identity": entry["scientific_run_identity"],
            "task_id": task_id,
            "reserved_at": "2026-07-19T00:00:00Z",
        }
        success = {
            "schema_version": 1,
            "event": "success",
            "manifest_hash": manifest["manifest_hash"],
            "scientific_run_identity": entry["scientific_run_identity"],
            "task_id": task_id,
            "result_sha256": receipt["artifact_hashes"]["result"],
            "recorded_at": "2026-07-19T00:00:01Z",
        }
        identity_slug = entry["scientific_run_identity"].removeprefix("sha256:")
        reservation_path = control_root / "reservations" / f"{identity_slug}.json"
        _write_json(reservation_path, reservation)
        reservations[task_id] = hashlib.sha256(reservation_path.read_bytes()).hexdigest()
        successes[task_id] = success
        launch_events.extend((reservation, success))
        result_rows.append(result_row)
    launch_path = control_root / "launch-ledger.jsonl"
    launch_path.parent.mkdir(parents=True, exist_ok=True)
    launch_path.write_bytes(
        b"".join(canonical_bytes(event) + b"\n" for event in launch_events)
    )
    launch_hash = hashlib.sha256(launch_path.read_bytes()).hexdigest()
    for task_id, receipt in receipt_by_id.items():
        receipt["batch_evidence"] = {
            "manifest_hash": manifest["manifest_hash"],
            "manifest_file_sha256": hashlib.sha256(manifest_path.read_bytes()).hexdigest(),
            "repository_lane_source_hash": manifest["repository"]["lane_source_hash"],
            "builder_source_tree_hash": builder["source_tree_hash"],
            "reservation_sha256": reservations[task_id],
            "launch_ledger_sha256": launch_hash,
            "failure_ledger_sha256": None,
            "success_event_sha256": sha256_value(successes[task_id]),
        }
        receipt["artifact_set_binding"] = sample_module._artifact_set_binding(receipt)
    ordered_receipts = [receipt_by_id[entry["task_id"]] for entry in manifest["tasks"]]
    result_bytes = b"".join(canonical_bytes(row) + b"\n" for row in result_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 result_rows
            ],
            "results_sha256": result_hash,
        }
    )
    aggregate_dir = output_root / "aggregates" / aggregate_identity.removeprefix("sha256:")
    results_path = aggregate_dir / "results.jsonl"
    results_path.parent.mkdir(parents=True, exist_ok=True)
    results_path.write_bytes(result_bytes)
    pass_path = batch_module._policy_receipt_path(control_root, manifest, "passed")
    batch_module._immutable_policy_receipt(
        pass_path,
        {
            "schema_version": 1,
            "kind": "issuance_passed",
            "manifest_hash": manifest["manifest_hash"],
            "policy": manifest["execution"]["issuance_policy"],
            "block_reason": None,
            "evaluated_after_completed_tasks": 100,
            "elapsed_wall_seconds": 1.0,
            "point_projected_wall_seconds": 4.0,
            "conservative_upper_wall_seconds": 5.0,
            "wall_target_seconds": 1500.0,
            "issued_task_count_at_gate": 108,
            "in_flight_at_gate": 8,
        },
        manifest,
    )
    summary_path = aggregate_dir / "summary.json"
    _write_json(
        summary_path,
        {
            "schema_version": 1,
            "aggregate_identity": aggregate_identity,
            "manifest_hash": manifest["manifest_hash"],
            "expected_task_count": 400,
            "validated_success_count": 400,
            "rejected": [],
            "complete": True,
            "artifact_hashes": {"results.jsonl": result_hash},
            "issuance_pass_receipt_sha256": hashlib.sha256(
                pass_path.read_bytes()
            ).hexdigest(),
            "authority": batch_module.DENIED_AUTHORITY,
            "scientific_verdicts": {"C1": "HOLD", "C2": "HOLD", "C3": "HOLD"},
        },
    )
    return {
        "receipts": ordered_receipts,
        "manifest_path": manifest_path,
        "output_root": output_root,
        "summary_path": summary_path,
        "results_path": results_path,
        "pass_path": pass_path,
    }


def _rebind_success_event(output_root: Path, output: Path):
    ledger = output_root / "control" / "launch-ledger.jsonl"
    rows = [json.loads(line) for line in ledger.read_text(encoding="utf-8").splitlines()]
    success = next(row for row in rows if row["event"] == "success")
    success["result_sha256"] = _sha(output / "result.json")
    ledger.write_bytes(b"".join(canonical_bytes(row) + b"\n" for row in rows))


def _ledger_rows(output_root: Path):
    ledger = output_root / "control" / "launch-ledger.jsonl"
    return ledger, [
        json.loads(line) for line in ledger.read_text(encoding="utf-8").splitlines()
    ]


def _write_ledger(path: Path, rows):
    path.write_bytes(b"".join(canonical_bytes(row) + b"\n" for row in rows))


def _rewrite_and_rebind(root: Path, filename: str, value):
    _write_json(root / filename, value)
    result = json.loads((root / "result.json").read_text(encoding="utf-8"))
    key = filename.removesuffix(".json")
    result["artifact_hashes"][key] = _sha(root / filename)
    _write_json(root / "result.json", result)


def test_exact_sample_hash_and_marginals(selected_tasks):
    assert load_stopping_sample_manifest()["batch_issuance_policy"] == {
        "type": "completion_projection_gate_v1",
        "evaluate_after_completed_tasks": 100,
        "wall_target_seconds": 1500.0,
        "upper_multiplier": 1.25,
        "max_in_flight": 8,
    }
    assert len(selected_tasks) == len({task["task_id"] for task in selected_tasks}) == 400
    assert _task_list_hash(selected_tasks) == EXPECTED_HASH
    validate_stopping_sample(selected_tasks)
    counts = Counter(task["suite"] for task in selected_tasks)
    assert counts == {
        "portfolio_gaussian_main": 100,
        "portfolio_gaussian_coverage_ablation": 100,
        "regression_absolute_main": 40,
        "portfolio_gaussian_highdim": 40,
        "portfolio_discrete": 40,
        "portfolio_gmm": 40,
        "regression_squared": 40,
    }
    marginals = defaultdict(lambda: defaultdict(Counter))
    for task in selected_tasks:
        for field in ("distribution_id", "sample_size", "replicate"):
            marginals[task["suite"]][field][task[field]] += 1
    for suite in ("portfolio_gaussian_main", "portfolio_gaussian_coverage_ablation"):
        assert set(marginals[suite]["distribution_id"].values()) == {2}
        assert set(marginals[suite]["sample_size"].values()) == {10}
        assert set(marginals[suite]["replicate"].values()) == {10}
    for suite in ("portfolio_gaussian_highdim", "portfolio_discrete", "portfolio_gmm"):
        assert set(marginals[suite]["distribution_id"].values()) == {4}
        assert set(marginals[suite]["sample_size"].values()) == {4}
        assert set(marginals[suite]["replicate"].values()) == {4}
    for suite in ("regression_absolute_main", "regression_squared"):
        assert set(marginals[suite]["distribution_id"].values()) == {4}
        assert set(marginals[suite]["sample_size"].values()) == {8}
        assert set(marginals[suite]["replicate"].values()) == {4}


def test_production_normalizers_satisfy_sample_receipt_execution_contract():
    plan = load_plan()
    contract = load_stopping_sample_manifest()["receipt_contract"]
    requested = {
        "schema_version": 2,
        "execution_scale": "stopping_storage_sample",
        "max_outer_iterations": 5000,
    }
    normalized = [normalize_execution_config(plan, requested)]
    normalized.extend(
        normalize_empirical_execution_config(plan, requested, suite=suite)
        for suite in (
            "portfolio_discrete",
            "portfolio_gmm",
            "regression_absolute_main",
            "regression_squared",
        )
    )
    for execution in normalized:
        assert execution["execution_scale"] == contract["required_execution_scale"]
        assert execution["optimizer"] == contract["required_optimizer"]
        assert (
            execution["max_outer_iterations"]
            == contract["required_max_outer_iterations"]
        )
        assert execution["store_every"] == contract["required_store_every"]
        assert execution["claim_eligible"] is False


def test_bootstrap_is_deterministic_and_scope_limited(complete_receipts):
    first = _analyze(complete_receipts)
    second = _analyze(complete_receipts)
    assert first == second
    assert first["stratum_count"] == 60
    assert first["population_task_count"] == 14000
    assert first["precision_pass"] is True
    assert first["analysis_implementation"] == (
        sample_module.stopping_sample_implementation_binding()
    )
    assert first["input_receipt_set_binding"].startswith("sha256:")
    assert first["weighted_estimates"]["cpu_core_seconds_total_at_cap"]["total"] == 140000.0
    assert first["weighted_estimates"]["storage_bytes_total_at_cap"]["total"] == 22400000.0
    assert first["weighted_estimates"]["restricted_mean_iterations_min_T_5000"]["mean"] == 5000.0
    assert first["weighted_estimates"]["trigger_probability_by_5000"]["mean"] == 0.0
    assert first["weighted_estimates"]["censoring_probability_at_5000"]["mean"] == 1.0
    assert first["cap_is_convergence"] is False
    assert first["eligibility"] == {
        "claim_eligible": False,
        "cost_freeze_eligible": False,
        "full_matrix_freeze_eligible": False,
        "diagnostic_resource_estimate_eligible": True,
    }


def test_analyzer_rejects_synthetic_receipts_without_private_test_hook(
    complete_receipts,
):
    with pytest.raises(ContractError, match="complete batch evidence"):
        analyze_stopping_sample(complete_receipts)


def test_private_test_hook_rejects_production_provenance(complete_receipts):
    receipts = [dict(receipt) for receipt in complete_receipts]
    receipts[0]["builder_implementation"] = (
        sample_module.stopping_sample_implementation_binding()
    )
    receipts[0]["artifact_set_binding"] = sample_module._artifact_set_binding(
        receipts[0]
    )
    with pytest.raises(ContractError, match="contains production provenance"):
        _analyze(receipts)


def test_production_analysis_binds_complete_aggregate_and_semantic_pass(
    tmp_path,
    selected_tasks,
    complete_receipts,
    production_batch_manifest,
):
    evidence = _materialize_complete_analysis_evidence(
        tmp_path,
        production_batch_manifest,
        selected_tasks,
        complete_receipts,
    )
    analysis = analyze_stopping_sample(
        evidence["receipts"],
        batch_manifest_path=evidence["manifest_path"],
        batch_output_root=evidence["output_root"],
        aggregate_summary_path=evidence["summary_path"],
        aggregate_results_path=evidence["results_path"],
        issuance_pass_receipt_path=evidence["pass_path"],
    )
    binding = analysis["analysis_evidence"]
    assert binding["mode"] == "production_batch_completion_v1"
    assert binding["manifest_hash"] == production_batch_manifest["manifest_hash"]
    assert binding["aggregate_identity"].startswith("sha256:")
    assert len(binding["issuance_pass_receipt_sha256"]) == 64
    assert binding["aggregate_results_sha256"] == hashlib.sha256(
        evidence["results_path"].read_bytes()
    ).hexdigest()


@pytest.mark.parametrize("mutation", ["missing_pass", "tampered_result_row", "blocker"])
def test_production_analysis_rejects_incomplete_or_rebound_batch_evidence(
    tmp_path,
    selected_tasks,
    complete_receipts,
    production_batch_manifest,
    mutation,
):
    evidence = _materialize_complete_analysis_evidence(
        tmp_path,
        production_batch_manifest,
        selected_tasks,
        complete_receipts,
    )
    if mutation == "missing_pass":
        evidence["pass_path"].unlink()
        message = "cannot load production artifact"
    elif mutation == "tampered_result_row":
        rows = evidence["results_path"].read_bytes().splitlines(keepends=True)
        row = json.loads(rows[0])
        row["run_identity"] = "sha256:" + "f" * 64
        rows[0] = canonical_bytes(row) + b"\n"
        evidence["results_path"].write_bytes(b"".join(rows))
        message = "aggregate row differs"
    else:
        blocker_path = batch_module._policy_receipt_path(
            evidence["output_root"] / "control",
            production_batch_manifest,
            "blocked",
        )
        _write_json(blocker_path, {"present": True})
        message = "issuance blocker"
    with pytest.raises(ContractError, match=message):
        analyze_stopping_sample(
            evidence["receipts"],
            batch_manifest_path=evidence["manifest_path"],
            batch_output_root=evidence["output_root"],
            aggregate_summary_path=evidence["summary_path"],
            aggregate_results_path=evidence["results_path"],
            issuance_pass_receipt_path=evidence["pass_path"],
        )


def test_missing_stratum_is_rejected(complete_receipts):
    receipts = [
        receipt
        for receipt in complete_receipts
        if not (
            receipt["task_id"].startswith("portfolio_gmm/")
            and receipt["task_id"].endswith("/n010")
        )
    ]
    with pytest.raises(ContractError, match="missing a required stratum"):
        _analyze(receipts)


def test_forged_or_unselected_task_is_rejected(selected_tasks, complete_receipts):
    selected_ids = {task["task_id"] for task in selected_tasks}
    target = selected_tasks[0]
    forged_task = next(
        task
        for task in expand_tasks(load_plan())
        if task["task_id"] not in selected_ids
        and task["suite"] == target["suite"]
        and task["sample_size"] == target["sample_size"]
    )
    receipts = [dict(receipt) for receipt in complete_receipts]
    receipts[0] = {**receipts[0], "task_id": forged_task["task_id"], "task_hash": forged_task["task_hash"]}
    with pytest.raises(ContractError, match="forged or unselected"):
        _analyze(receipts)


def test_failed_task_blocks_analysis(complete_receipts):
    receipts = [dict(receipt) for receipt in complete_receipts]
    receipts[10] = {**receipts[10], "status": "failed"}
    with pytest.raises(ContractError, match="task failed"):
        _analyze(receipts)


def test_analyzer_rejects_rebound_artifact_byte_total(complete_receipts):
    receipts = [dict(receipt) for receipt in complete_receipts]
    rebound = dict(receipts[0]["artifact_bytes"])
    rebound["iteration_trace"] += 1
    rebound["total"] += 1
    receipts[0] = {**receipts[0], "artifact_bytes": rebound}
    with pytest.raises(ContractError, match="artifact set binding is inconsistent"):
        _analyze(receipts)


@pytest.mark.parametrize(
    ("changes", "message"),
    [
        (
            {"stop_label": "converged_at_cap"},
            "may not be labeled convergence",
        ),
        (
            {
                "stop_label": "paper_rule_triggered",
                "stop_reason": "maximum_outer_iterations_reached",
            },
            "trigger label is inconsistent",
        ),
        (
            {
                "iterations_executed": 4999,
                "stop_label": "censored_at_diagnostic_cap",
            },
            "censoring label is inconsistent",
        ),
    ],
)
def test_censoring_and_trigger_labels_are_enforced(
    complete_receipts, changes, message
):
    receipts = [dict(receipt) for receipt in complete_receipts]
    receipts[0] = {**receipts[0], **changes}
    with pytest.raises(ContractError, match=message):
        _analyze(receipts)


def test_precision_failure_is_reported_without_promotion(
    selected_tasks, complete_receipts
):
    receipts = []
    for index, (task, receipt) in enumerate(zip(selected_tasks, complete_receipts)):
        value = 1_000_000.0 if index % 40 == 0 else 1.0
        receipts.append({**receipt, "runtime_seconds": value})
    analysis = _analyze(receipts)
    assert analysis["precision_pass"] is False
    assert any(value.endswith(":runtime_seconds") for value in analysis["precision_failures"])
    assert analysis["eligibility"]["diagnostic_resource_estimate_eligible"] is False
    assert analysis["eligibility"]["claim_eligible"] is False
    assert analysis["eligibility"]["cost_freeze_eligible"] is False
    assert analysis["eligibility"]["full_matrix_freeze_eligible"] is False


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("claim_eligible", True),
        ("cost_freeze_eligible", True),
        ("full_matrix_freeze_eligible", True),
        (
            "authority",
            {
                "paid_compute": True,
                "remote_compute": False,
                "external_inference": False,
                "push": False,
                "publish": False,
            },
        ),
    ],
)
def test_receipt_cannot_promote_scientific_or_cost_eligibility(
    complete_receipts, field, value
):
    receipts = [dict(receipt) for receipt in complete_receipts]
    receipts[0] = {**receipts[0], field: value}
    with pytest.raises(ContractError, match="eligibility or authority promotion"):
        _analyze(receipts)


def test_manifest_is_local_scope_limited_and_frozen():
    manifest = load_stopping_sample_manifest()
    assert manifest["expected_task_count"] == 400
    assert manifest["expected_task_id_list_sha256"] == EXPECTED_HASH
    assert manifest["diagnostic_contract"]["diagnostic_cap_iterations"] == 5000
    assert manifest["diagnostic_contract"]["cap_is_convergence"] is False
    assert manifest["eligibility"]["claim_eligible"] is False
    assert manifest["eligibility"]["cost_freeze_eligible"] is False
    assert manifest["eligibility"]["full_matrix_freeze_eligible"] is False
    assert not any(manifest["authority"].values())


def test_receipt_builder_hashes_every_production_output(
    monkeypatch, tmp_path, selected_tasks, production_batch_manifest
):
    monkeypatch.setattr(sample_module, "validate_result", lambda *args, **kwargs: None)
    result, manifest_path, output_root, output = _materialize_batch_evidence(
        tmp_path, production_batch_manifest, selected_tasks[0]
    )
    receipt = build_stopping_sample_receipt(
        output,
        selected_tasks[0],
        result_context=result,
        batch_manifest_path=manifest_path,
        batch_output_root=output_root,
    )
    expected = {
        "iteration_trace",
        "checkpoint_states",
        "result",
        "solver_receipt",
        "stopping_receipt",
        "lineage_receipt",
        "oos_receipt",
    }
    assert set(receipt["artifact_hashes"]) == expected
    assert set(receipt["artifact_bytes"]) == expected | {"total"}
    assert receipt["artifact_bytes"]["total"] == sum(
        receipt["artifact_bytes"][name] for name in expected
    )
    assert receipt["solver"] == {
        "statuses": {"optimal": 2},
        "maximum_residual": 2e-10,
    }
    assert receipt["batch_evidence"]["manifest_hash"] == production_batch_manifest[
        "manifest_hash"
    ]
    assert receipt["batch_evidence"]["failure_ledger_sha256"] is None
    assert receipt["builder_implementation"] == (
        sample_module.stopping_sample_implementation_binding()
    )
    assert (
        receipt["batch_evidence"]["builder_source_tree_hash"]
        == receipt["builder_implementation"]["source_tree_hash"]
    )
    assert receipt["stop_label"] == "paper_rule_triggered"


def test_receipt_builder_accepts_actual_one_step_runner_artifacts(
    tmp_path, selected_tasks
):
    task = next(task for task in selected_tasks if task["suite"] == "portfolio_discrete")
    output = tmp_path / "actual-runner-output"
    result = run_empirical_task(
        task["task_id"],
        output,
        requested_execution_config={
            "schema_version": 2,
            "execution_scale": "test",
            "max_outer_iterations": 1,
            "oos_sample_count": 257,
            "oos_chunk_size": 31,
        },
    )
    test_batch_evidence = {
        "run_identity": result["run_identity"],
        "status": "success",
        "failed_run_identities": [],
    }
    receipt = build_stopping_sample_receipt(
        output,
        task,
        result_context=result,
        _test_execution_override={
            "execution_scale": "test",
            "max_outer_iterations": 1,
        },
        _test_batch_evidence_override=test_batch_evidence,
    )

    with gzip.open(output / "iteration_trace.jsonl.gz", "rt", encoding="utf-8") as stream:
        scalar_rows = [json.loads(line) for line in stream if line.strip()]
    with gzip.open(output / "checkpoint_states.jsonl.gz", "rt", encoding="utf-8") as stream:
        checkpoint_rows = [json.loads(line) for line in stream if line.strip()]
    terminal = checkpoint_rows[-1]
    expected_maximum_residual = max(
        [row["solver_residual_maximum"] for row in scalar_rows]
        + [terminal["full_state"]["solver_residual_maximum"]]
    )
    solver_receipt = json.loads(
        (output / "solver_receipt.json").read_text(encoding="utf-8")
    )

    assert receipt["iterations_executed"] == 1
    assert sum(receipt["solver"]["statuses"].values()) == 2
    assert receipt["solver"]["statuses"] == solver_receipt["status_counts"]
    assert receipt["solver"]["maximum_residual"] == expected_maximum_residual
    assert terminal["iteration"] == receipt["iterations_executed"]
    assert "terminal" in terminal["checkpoint_reasons"]
    assert receipt["stop_label"] == "censored_at_diagnostic_cap"


def test_receipt_builder_accepts_actual_gaussian_one_step_runner_artifacts(
    tmp_path, selected_tasks
):
    task = next(
        task
        for task in selected_tasks
        if task["suite"] == "portfolio_gaussian_highdim"
    )
    output = tmp_path / "actual-gaussian-runner-output"
    result = run_gaussian_task(
        task["task_id"],
        output,
        requested_execution_config={
            "schema_version": 2,
            "execution_scale": "test",
            "max_outer_iterations": 1,
        },
    )
    receipt = build_stopping_sample_receipt(
        output,
        task,
        result_context=result,
        _test_execution_override={
            "execution_scale": "test",
            "max_outer_iterations": 1,
        },
        _test_batch_evidence_override={
            "run_identity": result["run_identity"],
            "status": "success",
            "failed_run_identities": [],
        },
    )
    oos = json.loads((output / "oos_receipt.json").read_text(encoding="utf-8"))
    assert receipt["iterations_executed"] == 1
    assert receipt["stop_label"] == "censored_at_diagnostic_cap"
    assert receipt["artifact_hashes"]["oos_receipt"] == hashlib.sha256(
        (output / "oos_receipt.json").read_bytes()
    ).hexdigest()
    assert oos["evaluation"] == "analytic_true_gaussian_cvar"
    assert oos["initial_cvar"] == result["metrics"]["oos_initial"]
    assert oos["final_cvar"] == result["metrics"]["oos_final"]


def test_receipt_builder_rejects_omitted_output_file(
    monkeypatch, tmp_path, selected_tasks, production_batch_manifest
):
    monkeypatch.setattr(sample_module, "validate_result", lambda *args, **kwargs: None)
    _result, manifest_path, output_root, output = _materialize_batch_evidence(
        tmp_path, production_batch_manifest, selected_tasks[0]
    )
    (output / "oos_receipt.json").unlink()
    with pytest.raises(ContractError, match="artifact set mismatch"):
        build_stopping_sample_receipt(
            output,
            selected_tasks[0],
            batch_manifest_path=manifest_path,
            batch_output_root=output_root,
        )


def test_receipt_builder_rejects_rebound_byte_totals(
    monkeypatch, tmp_path, selected_tasks, production_batch_manifest
):
    monkeypatch.setattr(sample_module, "validate_result", lambda *args, **kwargs: None)
    _result, manifest_path, output_root, output = _materialize_batch_evidence(
        tmp_path, production_batch_manifest, selected_tasks[0]
    )
    lineage_path = output / "lineage_receipt.json"
    lineage = json.loads(lineage_path.read_text(encoding="utf-8"))
    lineage["iteration_trace_bytes"] += 1
    _rewrite_and_rebind(output, "lineage_receipt.json", lineage)
    _rebind_success_event(output_root, output)
    with pytest.raises(ContractError, match="byte counts are rebound"):
        build_stopping_sample_receipt(
            output,
            selected_tasks[0],
            batch_manifest_path=manifest_path,
            batch_output_root=output_root,
        )


def test_receipt_builder_rejects_forged_lineage(
    monkeypatch, tmp_path, selected_tasks, production_batch_manifest
):
    monkeypatch.setattr(sample_module, "validate_result", lambda *args, **kwargs: None)
    _result, manifest_path, output_root, output = _materialize_batch_evidence(
        tmp_path, production_batch_manifest, selected_tasks[0]
    )
    lineage_path = output / "lineage_receipt.json"
    lineage = json.loads(lineage_path.read_text(encoding="utf-8"))
    lineage["task_hash"] = "sha256:" + "f" * 64
    _rewrite_and_rebind(output, "lineage_receipt.json", lineage)
    _rebind_success_event(output_root, output)
    with pytest.raises(ContractError, match="forged production lineage"):
        build_stopping_sample_receipt(
            output,
            selected_tasks[0],
            batch_manifest_path=manifest_path,
            batch_output_root=output_root,
        )


def test_receipt_builder_rejects_wrong_execution_scale(
    monkeypatch, tmp_path, selected_tasks, production_batch_manifest
):
    monkeypatch.setattr(sample_module, "validate_result", lambda *args, **kwargs: None)
    _result, manifest_path, output_root, output = _materialize_batch_evidence(
        tmp_path, production_batch_manifest, selected_tasks[0]
    )
    lineage_path = output / "lineage_receipt.json"
    lineage = json.loads(lineage_path.read_text(encoding="utf-8"))
    lineage["execution_config"]["execution_scale"] = "paper_scale"
    _rewrite_and_rebind(output, "lineage_receipt.json", lineage)
    _rebind_success_event(output_root, output)
    with pytest.raises(ContractError, match="execution scale is not frozen"):
        build_stopping_sample_receipt(
            output,
            selected_tasks[0],
            batch_manifest_path=manifest_path,
            batch_output_root=output_root,
        )


def test_receipt_builder_labels_cap_as_censoring_not_convergence(
    monkeypatch, tmp_path, selected_tasks, production_batch_manifest
):
    monkeypatch.setattr(sample_module, "validate_result", lambda *args, **kwargs: None)
    result, manifest_path, output_root, output = _materialize_batch_evidence(
        tmp_path,
        production_batch_manifest,
        selected_tasks[0],
        iterations=5000,
        capped=True,
    )
    receipt = build_stopping_sample_receipt(
        output,
        selected_tasks[0],
        result_context=result,
        batch_manifest_path=manifest_path,
        batch_output_root=output_root,
    )
    assert receipt["iterations_executed"] == 5000
    assert receipt["stop_label"] == "censored_at_diagnostic_cap"
    assert "converg" not in receipt["stop_label"]


def test_receipt_builder_rejects_failed_launch_identity(
    monkeypatch, tmp_path, selected_tasks, production_batch_manifest
):
    monkeypatch.setattr(sample_module, "validate_result", lambda *args, **kwargs: None)
    _result, manifest_path, output_root, output = _materialize_batch_evidence(
        tmp_path, production_batch_manifest, selected_tasks[0]
    )
    entry = next(
        row
        for row in production_batch_manifest["tasks"]
        if row["task_id"] == selected_tasks[0]["task_id"]
    )
    failure = {
        "schema_version": 1,
        "event": "failed",
        "manifest_hash": production_batch_manifest["manifest_hash"],
        "scientific_run_identity": entry["scientific_run_identity"],
        "task_id": selected_tasks[0]["task_id"],
    }
    failure_path = output_root / "control" / "failure-ledger.jsonl"
    failure_path.write_bytes(canonical_bytes(failure) + b"\n")
    with pytest.raises(ContractError, match="failure event"):
        build_stopping_sample_receipt(
            output,
            selected_tasks[0],
            batch_manifest_path=manifest_path,
            batch_output_root=output_root,
        )


def test_receipt_builder_rejects_caller_only_production_context(
    monkeypatch, tmp_path, selected_tasks, production_batch_manifest
):
    monkeypatch.setattr(sample_module, "validate_result", lambda *args, **kwargs: None)
    result, _manifest_path, _output_root, output = _materialize_batch_evidence(
        tmp_path, production_batch_manifest, selected_tasks[0]
    )
    with pytest.raises(ContractError, match="production batch evidence is missing"):
        build_stopping_sample_receipt(
            output,
            selected_tasks[0],
            result_context=result,
        )


def test_receipt_builder_rejects_test_override_at_production_scale(
    monkeypatch, tmp_path, selected_tasks, production_batch_manifest
):
    monkeypatch.setattr(sample_module, "validate_result", lambda *args, **kwargs: None)
    result, _manifest_path, _output_root, output = _materialize_batch_evidence(
        tmp_path, production_batch_manifest, selected_tasks[0]
    )
    with pytest.raises(ContractError, match="forbidden at production scale"):
        build_stopping_sample_receipt(
            output,
            selected_tasks[0],
            result_context=result,
            _test_batch_evidence_override={
                "status": "success",
                "run_identity": result["run_identity"],
                "failed_run_identities": [],
            },
        )


def test_receipt_builder_rejects_missing_reservation_receipt(
    monkeypatch, tmp_path, selected_tasks, production_batch_manifest
):
    monkeypatch.setattr(sample_module, "validate_result", lambda *args, **kwargs: None)
    _result, manifest_path, output_root, output = _materialize_batch_evidence(
        tmp_path, production_batch_manifest, selected_tasks[0]
    )
    reservation = next((output_root / "control" / "reservations").iterdir())
    reservation.unlink()
    with pytest.raises(ContractError, match="cannot load production artifact"):
        build_stopping_sample_receipt(
            output,
            selected_tasks[0],
            batch_manifest_path=manifest_path,
            batch_output_root=output_root,
        )


@pytest.mark.parametrize(
    ("mutation", "message"),
    [
        ("missing_reservation", "exactly one reservation event"),
        ("missing_success", "exactly one success event"),
        ("duplicate_success", "exactly one success event"),
        ("duplicate_reservation", "exactly one reservation event"),
        ("mismatched_reservation", "differs from its receipt"),
        ("mismatched_task", "batch success event: task_id"),
        ("mismatched_identity", "batch success event: scientific_run_identity"),
        ("mismatched_result_hash", "batch success event: result_sha256"),
        ("failed_in_launch_ledger", "failure event"),
    ],
)
def test_receipt_builder_rejects_missing_duplicate_or_mismatched_batch_events(
    monkeypatch,
    tmp_path,
    selected_tasks,
    production_batch_manifest,
    mutation,
    message,
):
    monkeypatch.setattr(sample_module, "validate_result", lambda *args, **kwargs: None)
    _result, manifest_path, output_root, output = _materialize_batch_evidence(
        tmp_path, production_batch_manifest, selected_tasks[0]
    )
    ledger, rows = _ledger_rows(output_root)
    reservation = rows[0]
    success = rows[1]
    if mutation == "missing_reservation":
        rows = [success]
    elif mutation == "missing_success":
        rows = [reservation]
    elif mutation == "duplicate_success":
        rows.append(dict(success))
    elif mutation == "duplicate_reservation":
        rows.append(dict(reservation))
    elif mutation == "mismatched_reservation":
        reservation["reserved_at"] = "2026-07-19T00:00:02Z"
    elif mutation == "mismatched_task":
        success["task_id"] = selected_tasks[1]["task_id"]
    elif mutation == "mismatched_identity":
        success["scientific_run_identity"] = "sha256:" + "f" * 64
    elif mutation == "mismatched_result_hash":
        success["result_sha256"] = "f" * 64
    elif mutation == "failed_in_launch_ledger":
        rows.append(
            {
                "schema_version": 1,
                "event": "failed",
                "manifest_hash": production_batch_manifest["manifest_hash"],
                "scientific_run_identity": success["scientific_run_identity"],
                "task_id": selected_tasks[0]["task_id"],
            }
        )
    _write_ledger(ledger, rows)
    with pytest.raises(ContractError, match=message):
        build_stopping_sample_receipt(
            output,
            selected_tasks[0],
            batch_manifest_path=manifest_path,
            batch_output_root=output_root,
        )


def test_receipt_builder_rejects_tampered_batch_manifest(
    monkeypatch, tmp_path, selected_tasks, production_batch_manifest
):
    monkeypatch.setattr(sample_module, "validate_result", lambda *args, **kwargs: None)
    _result, manifest_path, output_root, output = _materialize_batch_evidence(
        tmp_path, production_batch_manifest, selected_tasks[0]
    )
    tampered = json.loads(manifest_path.read_text(encoding="utf-8"))
    tampered["tasks"][0]["task_hash"] = "sha256:" + "f" * 64
    _write_json(manifest_path, tampered)
    with pytest.raises(ContractError, match="batch manifest hash mismatch"):
        build_stopping_sample_receipt(
            output,
            selected_tasks[0],
            batch_manifest_path=manifest_path,
            batch_output_root=output_root,
        )