bundle/v0/tests/test_optimizer_probe.py

from __future__ import annotations

import json
import hashlib
import subprocess
from pathlib import Path

import numpy as np
import pytest

from loss_aware_dro_repro import optimizer_probe as probe_module
from loss_aware_dro_repro.core import (
    CONFIG_ROOT,
    ContractError,
    canonical_bytes,
    sha256_value,
)
from loss_aware_dro_repro.optimizer_probe import (
    PROBE_EVIDENCE_SCALE,
    STOPPING_ESCALATION_EVIDENCE_SCALE,
    _mutate_launch_ledger,
    _literal_relative_improvement,
    _expected_probe_bindings,
    _optimizer_update,
    _relative_improvement,
    _run_trajectory,
    _stopping_diagnostics,
    load_optimizer_probe_manifest,
    run_optimizer_probe,
    validate_optimizer_probe,
)


def test_manifest_freezes_two_fidelity_paths_and_four_representative_tasks():
    manifest = load_optimizer_probe_manifest(CONFIG_ROOT / "optimizer_probe_v1.json")
    assert manifest["evidence_scale"] == PROBE_EVIDENCE_SCALE
    assert manifest["probe_cap_iterations"] == 250
    assert len(manifest["task_selectors"]) == 4
    assert [item["id"] for item in manifest["optimizers"]] == [
        "paper_plain_gradient_descent",
        "released_code_adam",
    ]
    assert manifest["optimizers"][0]["gradient_clip"] == [-1000.0, 1000.0]
    assert manifest["optimizers"][1]["gradient_clip"] == [-10000.0, 10000.0]
    assert manifest["acceptance"][
        "paired_initial_scientific_invariant_required"
    ] is True
    assert not any(manifest["authority"].values())


def test_stopping_escalation_manifest_freezes_exact_profile():
    manifest = load_optimizer_probe_manifest(
        CONFIG_ROOT / "optimizer_stopping_escalation_v1.json"
    )
    assert manifest["evidence_scale"] == STOPPING_ESCALATION_EVIDENCE_SCALE
    assert manifest["probe_cap_iterations"] == 5000
    assert manifest["checkpoint_iterations_zero_based"] == [
        0,
        1,
        2,
        4,
        9,
        24,
        49,
        99,
        249,
        499,
        999,
        1999,
        4999,
    ]
    assert manifest["task_selectors"] == [
        "portfolio_gaussian_main/d026/r04/n050",
        "portfolio_gmm/d005/r04/n050",
        "regression_absolute_main/d005/r04/n030",
        "regression_squared/d005/r04/n030",
    ]
    assert [item["id"] for item in manifest["optimizers"]] == [
        "paper_plain_gradient_descent",
        "released_code_adam",
    ]
    assert manifest["acceptance"]["claim_eligible"] is False
    assert manifest["acceptance"]["cost_freeze_eligible"] is False
    assert not any(manifest["authority"].values())


@pytest.mark.parametrize(
    ("field", "value", "message"),
    [
        ("probe_cap_iterations", 4999, "cap changed"),
        (
            "checkpoint_iterations_zero_based",
            [0, 4999],
            "checkpoint schedule changed",
        ),
        (
            "fixed_command",
            "python scripts/run_optimizer_probe.py --config configs/optimizer_probe_v1.json --output <immutable-output-dir>",
            "fixed command changed",
        ),
        ("probe_id", "loss-aware-optimizer-escalation-v2", "unsupported"),
    ],
)
def test_stopping_escalation_manifest_rejects_profile_drift(
    tmp_path, field, value, message
):
    manifest = load_optimizer_probe_manifest(
        CONFIG_ROOT / "optimizer_stopping_escalation_v1.json"
    )
    manifest[field] = value
    path = tmp_path / "optimizer_stopping_escalation_v1.json"
    path.write_text(json.dumps(manifest), encoding="utf-8")
    with pytest.raises(ContractError, match=message):
        load_optimizer_probe_manifest(path)


def test_stopping_escalation_manifest_rejects_renamed_config(tmp_path):
    manifest = load_optimizer_probe_manifest(
        CONFIG_ROOT / "optimizer_stopping_escalation_v1.json"
    )
    path = tmp_path / "renamed.json"
    path.write_text(json.dumps(manifest), encoding="utf-8")
    with pytest.raises(ContractError, match="config filename"):
        load_optimizer_probe_manifest(path)


def test_manifest_rejects_cap_as_convergence(tmp_path):
    manifest = load_optimizer_probe_manifest(CONFIG_ROOT / "optimizer_probe_v1.json")
    manifest["stopping_rules"]["cap_is_not_convergence"] = False
    path = tmp_path / "optimizer_probe_v1.json"
    path.write_text(json.dumps(manifest), encoding="utf-8")
    with pytest.raises(ContractError, match="stopping semantics"):
        load_optimizer_probe_manifest(path)


def _git(repo: Path, *arguments: str) -> str:
    completed = subprocess.run(
        ["git", "-C", str(repo), *arguments],
        capture_output=True,
        text=True,
        check=True,
    )
    return completed.stdout.strip()


def _write_commit_snapshot_fixture(repo: Path) -> tuple[Path, dict, str]:
    lane = repo / "reproductions" / "papers" / "loss-aware-dro-ot"
    package = lane / "src" / "loss_aware_dro_repro"
    (lane / "configs").mkdir(parents=True)
    (lane / "scripts").mkdir()
    (lane / "environment").mkdir()
    package.mkdir(parents=True)
    manifest = load_optimizer_probe_manifest(CONFIG_ROOT / "optimizer_probe_v1.json")
    (lane / "configs" / "optimizer_probe_v1.json").write_bytes(
        (CONFIG_ROOT / "optimizer_probe_v1.json").read_bytes()
    )
    (lane / "configs" / "paper_scale.json").write_bytes(
        (CONFIG_ROOT / "paper_scale.json").read_bytes()
    )
    (lane / "scripts" / "run_optimizer_probe.py").write_text(
        "print('snapshot runner')\n", encoding="utf-8"
    )
    (lane / "environment" / "scientific-freeze.txt").write_text(
        "snapshot-freeze\n", encoding="utf-8"
    )
    (package / "snapshot.py").write_text("VALUE = 1\n", encoding="utf-8")
    _git(repo, "init")
    _git(repo, "config", "user.email", "snapshot@example.invalid")
    _git(repo, "config", "user.name", "Snapshot Test")
    _git(repo, "add", ".")
    _git(repo, "commit", "-m", "frozen optimizer probe snapshot")
    return lane, manifest, _git(repo, "rev-parse", "HEAD")


def test_commit_snapshot_bindings_ignore_current_source_changes(
    monkeypatch, tmp_path
):
    lane, manifest, commit = _write_commit_snapshot_fixture(tmp_path)
    monkeypatch.setattr(probe_module, "LANE_ROOT", lane)
    config = lane / "configs" / "optimizer_probe_v1.json"
    before = _expected_probe_bindings(config, manifest, commit)
    (lane / "src" / "loss_aware_dro_repro" / "snapshot.py").write_text(
        "VALUE = 999\n", encoding="utf-8"
    )
    after = _expected_probe_bindings(config, manifest, commit)
    assert after == before
    assert _git(tmp_path, "status", "--short") == (
        "M reproductions/papers/loss-aware-dro-ot/"
        "src/loss_aware_dro_repro/snapshot.py"
    )


def test_commit_snapshot_rejects_forged_historical_commit(monkeypatch, tmp_path):
    lane, manifest, _commit = _write_commit_snapshot_fixture(tmp_path)
    monkeypatch.setattr(probe_module, "LANE_ROOT", lane)
    config = lane / "configs" / "optimizer_probe_v1.json"
    with pytest.raises(ContractError, match="bound commit validation"):
        _expected_probe_bindings(config, manifest, "f" * 40)


def test_relative_improvement_matches_released_signed_rule():
    assert _relative_improvement(None, 2.0) is None
    assert _relative_improvement(2.0, 1.0) == 0.5
    assert _relative_improvement(2.0, 3.0) == -0.5
    assert _relative_improvement(0.0, 0.25) == 0.25


def test_paper_literal_denominator_is_distinct_and_zero_is_undefined():
    assert _literal_relative_improvement(None, 1.0) is None
    assert _literal_relative_improvement(0.0, 1.0) is None
    assert _literal_relative_improvement(2.0, 1.0) == 0.5
    assert _literal_relative_improvement(-2.0, -3.0) == -0.5


def test_each_route_uses_its_own_primary_stop_but_records_all_diagnostics():
    paper = _stopping_diagnostics(
        optimizer_id="paper_plain_gradient_descent",
        previous_total=-2.0,
        current_total=-3.0,
        previous_lower=2.0,
        current_lower=1.0,
        tolerance=1e-6,
    )
    assert paper["primary_rule"] == (
        "paper_algorithm2_total_penalized_phi_literal_previous"
    )
    assert paper["primary_stop_trigger"] is True
    assert paper["paper_total_phi_literal"]["denominator_sign_inversion_risk"] is True
    assert paper["released_lower_objective_abs_denominator"][
        "stop_trigger_observed"
    ] is False

    released = _stopping_diagnostics(
        optimizer_id="released_code_adam",
        previous_total=2.0,
        current_total=1.0,
        previous_lower=2.0,
        current_lower=3.0,
        tolerance=1e-6,
    )
    assert released["primary_rule"] == (
        "released_nonpenalized_lower_objective_abs_previous"
    )
    assert released["primary_stop_trigger"] is True
    assert released["primary_trigger_caused_by_objective_worsening"] is True


def test_adam_bias_correction_and_post_transform_clip_are_exact():
    optimizer = {
        "id": "released_code_adam",
        "beta1": 0.9,
        "beta2": 0.999,
        "epsilon": 1e-8,
        "gradient_clip": [-0.5, 0.5],
    }
    raw = np.array([[2.0, 99.0], [-4.0, 8.0]])
    transformed, applied, moments = _optimizer_update(
        raw, optimizer, {"step": 0}
    )
    np.testing.assert_allclose(
        transformed, np.array([[1.0, 0.0], [-1.0, 1.0]]), atol=1e-7
    )
    np.testing.assert_allclose(applied, np.array([[0.5, 0.0], [-0.5, 0.5]]))
    assert moments["step"] == 1
    np.testing.assert_allclose(moments["first"], np.array([[0.2, 0.0], [-0.4, 0.8]]))


def test_plain_gd_clips_raw_gradient_without_adam_transform():
    optimizer = {
        "id": "paper_plain_gradient_descent",
        "gradient_clip": [-1.0, 1.0],
    }
    raw = np.array([[2.0, 99.0], [-4.0, 0.5]])
    transformed, applied, moments = _optimizer_update(raw, optimizer, {"step": 0})
    np.testing.assert_array_equal(transformed, np.array([[2.0, 0.0], [-4.0, 0.5]]))
    np.testing.assert_array_equal(applied, np.array([[1.0, 0.0], [-1.0, 0.5]]))
    assert moments == {"step": 1}


def _synthetic_manifest(cap: int = 3) -> dict:
    return {
        "evidence_scale": PROBE_EVIDENCE_SCALE,
        "probe_cap_iterations": cap,
        "checkpoint_iterations_zero_based": [0, cap - 1],
        "stopping_rules": {"tolerance": 1e-6},
        "acceptance": {
            "accepted_solver_statuses": ["optimal"],
            "solver_residual_max": 1e-7,
        },
    }


def _synthetic_state(objective: float) -> dict:
    return {
        "total_gradient": np.zeros((1, 1)),
        "total_objective": objective,
        "penalty": 0.0,
        "lower": {"objective": objective, "status": "optimal"},
    }


def _synthetic_payload(_state: dict) -> dict:
    return {
        "decision": [0.25],
        "bootstrap_distances": [0.1, 0.2, 0.3],
        "solver_residuals": {
            "contract_version": 2,
            "primal": 0.0,
            "dual": 0.0,
            "equality": 0.0,
            "cone": 0.0,
            "dual_cone": 0.0,
            "complementarity": 0.0,
            "duality_gap": 0.0,
            "solver_native_primal": 0.0,
            "solver_native_dual": 0.0,
            "primal_relative": 0.0,
            "dual_relative": 0.0,
            "equality_relative": 0.0,
            "cone_relative": 0.0,
            "dual_cone_relative": 0.0,
            "duality_gap_relative": 0.0,
            "complementarity_relative": 0.0,
        }
    }


def test_trajectory_applies_update_then_records_exact_released_stop():
    objectives = iter([10.0, 11.0, 9.0])
    summary, trace, checkpoints = _run_trajectory(
        task={
            "task_id": "synthetic/task",
            "task_hash": "sha256:task",
            "hyperparameters": {"coverage_beta": 0.1},
        },
        initial_L=np.eye(1),
        evaluate=lambda _L: _synthetic_state(next(objectives)),
        state_payload=_synthetic_payload,
        optimizer={
            "id": "paper_plain_gradient_descent",
            "learning_rate": 1e-4,
            "gradient_clip": [-1000.0, 1000.0],
            "metric_eigenvalue_clip": [1e-6, 1e6],
        },
        manifest=_synthetic_manifest(),
        bootstrap_quantile_method="higher",
    )
    rows = [json.loads(line) for line in trace.splitlines()]
    checkpoint_rows = [json.loads(line) for line in checkpoints.splitlines()]
    assert len(rows) == 2
    assert rows[-1]["stop_trigger"] is True
    assert "primary_signed_relative_improvement" in rows[-1]
    assert rows[-1]["penalty"] == 0.0
    assert rows[-1]["iteration_seconds"] >= 0.0
    assert "signed_relative_nonpenalized_objective_improvement" not in rows[-1]
    assert summary["stop_trigger_observed"] is True
    assert summary["probe_cap_reached"] is False
    assert summary["stop_reason"] == (
        "optimizer_fidelity_primary_stopping_rule_triggered"
    )
    assert summary["initial_scientific_invariant_hash"].startswith("sha256:")
    assert summary["initial_scientific_invariant"]["epsilon"] == 0.3
    assert checkpoint_rows[-1]["iteration"] == 1
    assert "L_next" in checkpoint_rows[-1]


def test_cap_is_explicitly_not_convergence_and_trace_stays_compact():
    objectives = iter([10.0, 9.0, 8.0])
    summary, trace, checkpoints = _run_trajectory(
        task={
            "task_id": "synthetic/task",
            "task_hash": "sha256:task",
            "hyperparameters": {"coverage_beta": 0.1},
        },
        initial_L=np.eye(1),
        evaluate=lambda _L: _synthetic_state(next(objectives)),
        state_payload=_synthetic_payload,
        optimizer={
            "id": "paper_plain_gradient_descent",
            "learning_rate": 1e-4,
            "gradient_clip": [-1000.0, 1000.0],
            "metric_eigenvalue_clip": [1e-6, 1e6],
        },
        manifest=_synthetic_manifest(),
        bootstrap_quantile_method="higher",
    )
    rows = [json.loads(line) for line in trace.splitlines()]
    assert len(rows) == 3
    assert all("state" not in row and "L" not in row for row in rows)
    assert summary["probe_cap_reached"] is True
    assert summary["probe_cap_is_not_convergence"] is True
    assert summary["converged_label_allowed"] is False
    assert summary["stop_reason"] == "probe_cap_reached_without_stop_trigger"
    assert len(checkpoints.splitlines()) == 2


def test_launch_ledger_never_reuses_failed_identity(tmp_path):
    ledger = tmp_path / "ledger.json"
    output = tmp_path / "out"
    _mutate_launch_ledger(ledger, "sha256:fixed", output_dir=output, status="reserved")
    _mutate_launch_ledger(ledger, "sha256:fixed", output_dir=output, status="failed")
    with pytest.raises(ContractError, match="already reserved"):
        _mutate_launch_ledger(
            ledger, "sha256:fixed", output_dir=output, status="reserved"
        )


@pytest.mark.parametrize(
    "config_name",
    ["optimizer_probe_v1.json", "optimizer_stopping_escalation_v1.json"],
)
def test_runner_refuses_dirty_lane_before_creating_output(
    monkeypatch, tmp_path, config_name
):
    monkeypatch.setattr(
        "loss_aware_dro_repro.optimizer_probe._repository_state",
        lambda: ("a" * 40, False),
    )
    output = tmp_path / "must-not-exist"
    with pytest.raises(ContractError, match="paper lane is not clean"):
        run_optimizer_probe(CONFIG_ROOT / config_name, output)
    assert not output.exists()


def test_validator_rejects_incomplete_artifact_directory(tmp_path):
    (tmp_path / "optimizer-probe.json").write_text(
        json.dumps(
            {
                "evidence_scale": PROBE_EVIDENCE_SCALE,
                "claim_eligible": False,
                "cost_freeze_eligible": False,
                "pairs": [],
            }
        ),
        encoding="utf-8",
    )
    with pytest.raises(ContractError, match="file set mismatch"):
        validate_optimizer_probe(CONFIG_ROOT / "optimizer_probe_v1.json", tmp_path)


def _write_synthetic_probe_directory(
    directory: Path,
    *,
    second_invariant_value: float = 1.0,
    second_trace_optimizer_id: str = "released_code_adam",
) -> tuple[dict, dict]:
    manifest = {
        "probe_id": "loss-aware-optimizer-fidelity-v1",
        "evidence_scale": PROBE_EVIDENCE_SCALE,
        "fixed_command": "python synthetic.py",
        "probe_cap_iterations": 1,
        "task_selectors": ["synthetic/task"],
        "checkpoint_iterations_zero_based": [0],
        "stopping_rules": {
            "tolerance": 1e-6,
            "first_iteration_eligible": 1,
        },
        "acceptance": {
            "accepted_solver_statuses": ["optimal"],
            "solver_residual_max": 1e-7,
        },
        "optimizers": [
            {"id": "paper_plain_gradient_descent"},
            {"id": "released_code_adam"},
        ],
    }
    pairs = []
    for optimizer_id in (
        "paper_plain_gradient_descent",
        "released_code_adam",
    ):
        relative = Path("task-00") / optimizer_id
        target = directory / relative
        target.mkdir(parents=True)
        trace_optimizer_id = (
            second_trace_optimizer_id
            if optimizer_id == "released_code_adam"
            else optimizer_id
        )
        trace = canonical_bytes(
            {
                "record_type": "optimizer_probe_iteration",
                "evidence_scale": PROBE_EVIDENCE_SCALE,
                "claim_eligible": False,
                "cost_freeze_eligible": False,
                "task_id": "synthetic/task",
                "optimizer_id": trace_optimizer_id,
                "iteration": 0,
                "nonpenalized_objective": 2.0,
                "penalty": 0.5,
                "total_objective": 2.5,
                "primary_signed_relative_improvement": None,
                "raw_gradient_fro": 0.1,
                "transformed_gradient_fro": 0.1,
                "applied_gradient_fro": 0.1,
                "solver_status": "optimal",
                "residual_maximum": 0.0,
                "iteration_seconds": 0.01,
                "stop_trigger": False,
                "cap_iteration": True,
                "stopping_diagnostics": _stopping_diagnostics(
                    optimizer_id=optimizer_id,
                    previous_total=None,
                    current_total=2.5,
                    previous_lower=None,
                    current_lower=2.0,
                    tolerance=1e-6,
                ),
            }
        ) + b"\n"
        checkpoints = canonical_bytes(
            {
                "record_type": "optimizer_probe_checkpoint",
                "task_id": "synthetic/task",
                "optimizer_id": optimizer_id,
                "iteration": 0,
                "L": [[1.0]],
                "metric": [[1.0]],
                "L_next": [[1.0]],
                "metric_next": [[1.0]],
                "raw_gradient": [[0.1]],
                "transformed_gradient": [[0.1]],
                "applied_gradient": [[0.1]],
                "state": {
                    "decision": [0.25],
                    "objective": 2.0,
                    "penalty": 0.5,
                    "total_objective": 2.5,
                    "solver_status": "optimal",
                    "solver_residuals": _synthetic_payload({})[
                        "solver_residuals"
                    ],
                },
            }
        ) + b"\n"
        invariant = {
            "epsilon": (
                second_invariant_value
                if optimizer_id == "released_code_adam"
                else 1.0
            ),
            "initial_L": [[1.0]],
            "initial_metric": [[1.0]],
            "initial_nonpenalized_lower_objective": 2.0,
            "initial_total_penalized_objective": 2.5,
            "initial_decision": [0.25],
            "initial_raw_gradient": [[0.1]],
            "dataset_task_lineage": {
                "task_id": "synthetic/task",
                "task_hash": "sha256:task",
                "suite": "synthetic",
                "distribution_id": 0,
                "replicate": 0,
                "sample_size": 3,
                "seeds": {"dataset": 1},
            },
        }
        summary = {
            "schema_version": 1,
            "evidence_scale": PROBE_EVIDENCE_SCALE,
            "claim_eligible": False,
            "cost_freeze_eligible": False,
            "converged_label_allowed": False,
            "task_id": "synthetic/task",
            "task_hash": "sha256:task",
            "optimizer_id": optimizer_id,
            "iterations_executed": 1,
            "stop_reason": "probe_cap_reached_without_stop_trigger",
            "probe_cap_reached": True,
            "probe_cap_is_not_convergence": True,
            "stop_trigger_observed": False,
            "stop_trigger": None,
            "maximum_solver_residual": 0.0,
            "terminal_metric_factor": [[1.0]],
            "terminal_metric": [[1.0]],
            "initial_scientific_invariant": invariant,
            "initial_scientific_invariant_hash": sha256_value(invariant),
            "trace_sha256": hashlib.sha256(trace).hexdigest(),
            "trace_bytes": len(trace),
            "checkpoints_sha256": hashlib.sha256(checkpoints).hexdigest(),
            "checkpoints_bytes": len(checkpoints),
        }
        trace_path = target / "trace.jsonl"
        checkpoints_path = target / "checkpoints.jsonl"
        summary_path = target / "summary.json"
        trace_path.write_bytes(trace)
        checkpoints_path.write_bytes(checkpoints)
        summary_path.write_bytes(canonical_bytes(summary) + b"\n")
        pairs.append(
            {
                "task_id": "synthetic/task",
                "task_hash": "sha256:task",
                "optimizer_id": optimizer_id,
                "summary_path": (relative / "summary.json").as_posix(),
                "summary_sha256": hashlib.sha256(summary_path.read_bytes()).hexdigest(),
                "trace_path": (relative / "trace.jsonl").as_posix(),
                "trace_sha256": summary["trace_sha256"],
                "checkpoints_path": (relative / "checkpoints.jsonl").as_posix(),
                "checkpoints_sha256": summary["checkpoints_sha256"],
                "initial_scientific_invariant_hash": summary[
                    "initial_scientific_invariant_hash"
                ],
            }
        )
    bindings = {
        "commit": "a" * 40,
        "source_tree_hash": "sha256:source",
        "source_files": {"src/example.py": "source-file-hash"},
        "config_file_sha256": "config-file-hash",
        "config_value_hash": "sha256:config-value",
        "plan_hash": "sha256:plan",
        "task_hashes": {"synthetic/task": "sha256:task"},
        "reference_commit": "b" * 40,
        "reference_source_sha256": "reference-source-hash",
        "fixed_command": "python synthetic.py",
    }
    run_key = sha256_value(
        {
            "probe_id": manifest["probe_id"],
            "commit": bindings["commit"],
            "fixed_command": bindings["fixed_command"],
            "source_tree_hash": bindings["source_tree_hash"],
            "config_hash": "sha256:" + bindings["config_file_sha256"],
            "config_value_hash": bindings["config_value_hash"],
            "plan_hash": bindings["plan_hash"],
            "task_hashes": bindings["task_hashes"],
        }
    )
    receipt = {
        "schema_version": 1,
        "probe_id": "loss-aware-optimizer-fidelity-v1",
        "evidence_scale": PROBE_EVIDENCE_SCALE,
        "claim_eligible": False,
        "cost_freeze_eligible": False,
        "converged_label_allowed": False,
        "run_key": run_key,
        "bindings": bindings,
        "pair_count": 2,
        "pairs": pairs,
        "authority": {
            "paid_compute": False,
            "remote_compute": False,
            "external_inference": False,
            "push": False,
            "publish": False,
        },
        "scientific_verdicts": {"C1": "HOLD", "C2": "HOLD", "C3": "HOLD"},
    }
    (directory / "optimizer-probe.json").write_bytes(canonical_bytes(receipt) + b"\n")
    return manifest, receipt


def _install_synthetic_validator(monkeypatch, manifest: dict, receipt: dict) -> None:
    expected_bindings = json.loads(json.dumps(receipt["bindings"]))
    monkeypatch.setattr(
        probe_module, "load_optimizer_probe_manifest", lambda _path: manifest
    )
    monkeypatch.setattr(
        probe_module,
        "_expected_probe_bindings",
        lambda _config_path, _manifest, _commit: expected_bindings,
    )


def _mutate_first_trace_and_rebind(directory: Path, receipt: dict, mutate) -> None:
    pair = receipt["pairs"][0]
    trace_path = directory / pair["trace_path"]
    summary_path = directory / pair["summary_path"]
    row = json.loads(trace_path.read_text(encoding="utf-8"))
    mutate(row)
    trace_path.write_bytes(canonical_bytes(row) + b"\n")
    summary = json.loads(summary_path.read_text(encoding="utf-8"))
    summary["trace_sha256"] = hashlib.sha256(trace_path.read_bytes()).hexdigest()
    summary["trace_bytes"] = trace_path.stat().st_size
    summary_path.write_bytes(canonical_bytes(summary) + b"\n")
    pair["trace_sha256"] = summary["trace_sha256"]
    pair["summary_sha256"] = hashlib.sha256(summary_path.read_bytes()).hexdigest()
    (directory / "optimizer-probe.json").write_bytes(
        canonical_bytes(receipt) + b"\n"
    )


def _mutate_first_checkpoint_and_rebind(
    directory: Path, receipt: dict, mutate
) -> None:
    pair = receipt["pairs"][0]
    checkpoint_path = directory / pair["checkpoints_path"]
    summary_path = directory / pair["summary_path"]
    row = json.loads(checkpoint_path.read_text(encoding="utf-8"))
    mutate(row)
    checkpoint_path.write_bytes(canonical_bytes(row) + b"\n")
    summary = json.loads(summary_path.read_text(encoding="utf-8"))
    summary["checkpoints_sha256"] = hashlib.sha256(
        checkpoint_path.read_bytes()
    ).hexdigest()
    summary["checkpoints_bytes"] = checkpoint_path.stat().st_size
    summary_path.write_bytes(canonical_bytes(summary) + b"\n")
    pair["checkpoints_sha256"] = summary["checkpoints_sha256"]
    pair["summary_sha256"] = hashlib.sha256(summary_path.read_bytes()).hexdigest()
    (directory / "optimizer-probe.json").write_bytes(
        canonical_bytes(receipt) + b"\n"
    )


def test_validator_accepts_exact_paired_initial_invariant(monkeypatch, tmp_path):
    manifest, _ = _write_synthetic_probe_directory(tmp_path)
    receipt = json.loads((tmp_path / "optimizer-probe.json").read_text())
    _install_synthetic_validator(monkeypatch, manifest, receipt)
    validated = validate_optimizer_probe(Path("unused.json"), tmp_path)
    assert validated["pair_count"] == 2


def test_validator_rejects_paired_initial_state_drift(monkeypatch, tmp_path):
    manifest, receipt = _write_synthetic_probe_directory(
        tmp_path, second_invariant_value=1.01
    )
    _install_synthetic_validator(monkeypatch, manifest, receipt)
    with pytest.raises(ContractError, match="paired initial scientific state differs"):
        validate_optimizer_probe(Path("unused.json"), tmp_path)


def test_validator_rejects_trace_identity_even_when_hashes_are_self_consistent(
    monkeypatch, tmp_path
):
    manifest, receipt = _write_synthetic_probe_directory(
        tmp_path, second_trace_optimizer_id="paper_plain_gradient_descent"
    )
    _install_synthetic_validator(monkeypatch, manifest, receipt)
    with pytest.raises(ContractError, match="row identity"):
        validate_optimizer_probe(Path("unused.json"), tmp_path)


def test_validator_rejects_noncontiguous_trace_with_rebound_hashes(
    monkeypatch, tmp_path
):
    manifest, receipt = _write_synthetic_probe_directory(tmp_path)
    _install_synthetic_validator(monkeypatch, manifest, receipt)
    pair = receipt["pairs"][0]
    trace_path = tmp_path / pair["trace_path"]
    summary_path = tmp_path / pair["summary_path"]
    row = json.loads(trace_path.read_text(encoding="utf-8"))
    row["iteration"] = 1
    trace_path.write_bytes(canonical_bytes(row) + b"\n")
    summary = json.loads(summary_path.read_text(encoding="utf-8"))
    summary["trace_sha256"] = hashlib.sha256(trace_path.read_bytes()).hexdigest()
    summary["trace_bytes"] = trace_path.stat().st_size
    summary_path.write_bytes(canonical_bytes(summary) + b"\n")
    pair["trace_sha256"] = summary["trace_sha256"]
    pair["summary_sha256"] = hashlib.sha256(summary_path.read_bytes()).hexdigest()
    (tmp_path / "optimizer-probe.json").write_bytes(
        canonical_bytes(receipt) + b"\n"
    )
    with pytest.raises(ContractError, match="trace iteration sequence"):
        validate_optimizer_probe(Path("unused.json"), tmp_path)


def test_validator_rejects_full_state_in_trace_with_rebound_hashes(
    monkeypatch, tmp_path
):
    manifest, receipt = _write_synthetic_probe_directory(tmp_path)
    _install_synthetic_validator(monkeypatch, manifest, receipt)
    pair = receipt["pairs"][0]
    trace_path = tmp_path / pair["trace_path"]
    summary_path = tmp_path / pair["summary_path"]
    row = json.loads(trace_path.read_text(encoding="utf-8"))
    row["L"] = [[1.0]]
    trace_path.write_bytes(canonical_bytes(row) + b"\n")
    summary = json.loads(summary_path.read_text(encoding="utf-8"))
    summary["trace_sha256"] = hashlib.sha256(trace_path.read_bytes()).hexdigest()
    summary["trace_bytes"] = trace_path.stat().st_size
    summary_path.write_bytes(canonical_bytes(summary) + b"\n")
    pair["trace_sha256"] = summary["trace_sha256"]
    pair["summary_sha256"] = hashlib.sha256(summary_path.read_bytes()).hexdigest()
    (tmp_path / "optimizer-probe.json").write_bytes(
        canonical_bytes(receipt) + b"\n"
    )
    with pytest.raises(ContractError, match="compact scalar evidence"):
        validate_optimizer_probe(Path("unused.json"), tmp_path)


@pytest.mark.parametrize(
    ("mutation", "message"),
    [
        (
            lambda row: row.__setitem__("total_objective", 2.6),
            "lower objective plus penalty",
        ),
        (
            lambda row: row.__setitem__("solver_status", "failed"),
            "unaccepted solver status",
        ),
        (
            lambda row: row["stopping_diagnostics"]["paper_total_phi_literal"].__setitem__(
                "objective_worsened", True
            ),
            "stopping diagnostics",
        ),
        (
            lambda row: row.__setitem__("iteration_seconds", float("nan")),
            "non-finite numeric value",
        ),
    ],
)
def test_validator_recomputes_trace_semantics_after_hash_rebinding(
    monkeypatch, tmp_path, mutation, message
):
    manifest, receipt = _write_synthetic_probe_directory(tmp_path)
    _install_synthetic_validator(monkeypatch, manifest, receipt)
    _mutate_first_trace_and_rebind(tmp_path, receipt, mutation)
    with pytest.raises(ContractError, match=message):
        validate_optimizer_probe(Path("unused.json"), tmp_path)


def test_validator_recomputes_bindings_and_run_key(monkeypatch, tmp_path):
    manifest, receipt = _write_synthetic_probe_directory(tmp_path)
    _install_synthetic_validator(monkeypatch, manifest, receipt)
    receipt["bindings"]["config_file_sha256"] = "attacker-rebound-config"
    receipt["run_key"] = sha256_value({"attacker": "self-consistent-story"})
    (tmp_path / "optimizer-probe.json").write_bytes(
        canonical_bytes(receipt) + b"\n"
    )
    with pytest.raises(ContractError, match="binding mismatch: config_file_sha256"):
        validate_optimizer_probe(Path("unused.json"), tmp_path)


def test_validator_rejects_hash_rebound_checkpoint_metric_tamper(
    monkeypatch, tmp_path
):
    manifest, receipt = _write_synthetic_probe_directory(tmp_path)
    _install_synthetic_validator(monkeypatch, manifest, receipt)
    _mutate_first_checkpoint_and_rebind(
        tmp_path, receipt, lambda row: row.__setitem__("metric", [[2.0]])
    )
    with pytest.raises(ContractError, match="metric disagrees with its factor"):
        validate_optimizer_probe(Path("unused.json"), tmp_path)


def test_validator_rejects_pair_reordering_and_noncanonical_receipt(
    monkeypatch, tmp_path
):
    manifest, receipt = _write_synthetic_probe_directory(tmp_path)
    _install_synthetic_validator(monkeypatch, manifest, receipt)
    receipt["pairs"].reverse()
    receipt_path = tmp_path / "optimizer-probe.json"
    receipt_path.write_bytes(canonical_bytes(receipt) + b"\n")
    with pytest.raises(ContractError, match="manifest order"):
        validate_optimizer_probe(Path("unused.json"), tmp_path)

    receipt["pairs"].reverse()
    receipt_path.write_text(json.dumps(receipt, indent=2), encoding="utf-8")
    with pytest.raises(ContractError, match="canonically serialized"):
        validate_optimizer_probe(Path("unused.json"), tmp_path)