tests/test_eval_phase3_adversarial.py

"""CPU-only adversarial contract for the Phase 3 objective-only evaluator.

Expected public API (specified before implementation)::

    from music3lab import eval as phase3
    authority = phase3.prepare_corpus(sources=[Path(...), ...],
        output_root=Path(...), ffmpeg_path="ffmpeg", required_frames=44_032,
        guard_seconds=5)
    result = phase3.evaluate_objective_only(authority=authority,
        candidate_root=Path(...), protected_baseline=...,
        optional_evaluators=("clap", "ast", "whisper"))
    phase3.verify_evaluation_bundle(Path(...), authority=authority)

``prepare_corpus`` retains source and ffmpeg by descriptor; normalizes through
ffmpeg; makes SHA-named 0444 copies; uses raw hashes for a 1/3/4 source-level
split; creates one content-blind, SHA-derived 44,032-frame crop with 5s guards;
and seals heldout sources from tuning. ``evaluate_objective_only`` only accepts
deterministic numerical scores, returns unavailable optional evaluators as
``MISSING``, and never turns a missing render into zero/pass. Both APIs reject
rehashed manifest/path/mode/link/crop/aggregate/candidate/publication attacks.
Promotion is Pareto + hard floors + protected regression + bootstrap CI, never
a learned or subjective scalar. These CPU-only synthetic WAVs test integrity,
not listening preference.
"""
from __future__ import annotations

import hashlib
import json
import os
from pathlib import Path
import stat
import wave

import numpy as np
import pytest

from music3lab import eval as phase3


RATE, FRAMES = 44_100, 44_032


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


def _wav(path: Path, *, frames: int, kind: str, seed: int) -> Path:
    """Deterministic non-musical stereo WAV with distinct source bytes."""
    index = np.arange(frames, dtype=np.float64)
    if kind == "tone":
        left = np.sin(2 * np.pi * (193 + seed) * index / RATE)
        right = np.cos(2 * np.pi * (313 + seed) * index / RATE)
    elif kind == "noise":
        rng = np.random.default_rng(seed)
        left, right = rng.normal(0, 0.08, frames), rng.normal(0, 0.08, frames)
    elif kind == "silence":
        left = right = np.zeros(frames)
    else:
        raise ValueError(kind)
    pcm = np.column_stack((left, right)).clip(-1, 1)
    with wave.open(str(path), "wb") as handle:
        handle.setnchannels(2); handle.setsampwidth(2); handle.setframerate(RATE)
        handle.writeframes((pcm * 32767).astype("<i2").tobytes())
    return path


@pytest.fixture
def sources(tmp_path: Path) -> list[Path]:
    return [_wav(tmp_path / f"source-{n}.wav", frames=12 * RATE + n,
                 kind="tone" if n % 2 else "noise", seed=n) for n in range(8)]


@pytest.fixture
def authority(tmp_path: Path, sources: list[Path]):
    return phase3.prepare_corpus(sources=sources, output_root=tmp_path / "corpus",
        ffmpeg_path="ffmpeg", required_frames=FRAMES, guard_seconds=5)


def _manifest(authority) -> dict:
    return json.loads((Path(authority.root) / "manifest.json").read_text())


def _resign(path: Path) -> None:
    """Attacker recomputes ordinary self-described hash after a mutation."""
    value = json.loads(path.read_text())
    value["semantic_digest"] = phase3.semantic_digest({k: v for k, v in value.items() if k != "semantic_digest"})
    path.write_text(json.dumps(value, sort_keys=True, separators=(",", ":")))


def test_corpus_is_immutable_hash_named_and_split_by_raw_source(authority, sources: list[Path]) -> None:
    manifest = _manifest(authority)
    assert {name: len(manifest["splits"][name]) for name in ("calibration", "validation", "heldout")} == {"calibration": 1, "validation": 3, "heldout": 4}
    assert set().union(*map(set, manifest["splits"].values())) == {_sha(path) for path in sources}
    assert sum(map(len, manifest["splits"].values())) == 8
    for record in manifest["sources"].values():
        copy_path = Path(authority.root) / record["corpus_path"]
        assert copy_path.name == record["raw_sha256"] + ".wav"
        assert _sha(copy_path) == record["normalized_sha256"]
        assert stat.S_IMODE(copy_path.stat().st_mode) == 0o444


def test_split_has_no_raw_pcm_or_crop_leakage(authority) -> None:
    manifest, seen_raw, seen_pcm, seen_crop = _manifest(authority), set(), set(), set()
    for raw_hashes in manifest["splits"].values():
        for raw_hash in raw_hashes:
            record = manifest["sources"][raw_hash]
            assert raw_hash not in seen_raw
            assert record["normalized_sha256"] not in seen_pcm
            assert record["crop"]["sha256"] not in seen_crop
            seen_raw.add(raw_hash); seen_pcm.add(record["normalized_sha256"]); seen_crop.add(record["crop"]["sha256"])


def test_crop_is_content_blind_sha_derived_exact_length_and_guarded(authority) -> None:
    manifest = _manifest(authority)
    for raw_hash, record in manifest["sources"].items():
        crop = record["crop"]
        assert crop["frames"] == FRAMES
        assert crop["start_frame"] == phase3.content_blind_crop_start(raw_sha256=raw_hash, total_frames=record["normalized_frames"], required_frames=FRAMES, guard_frames=5 * RATE)
        assert 5 * RATE <= crop["start_frame"]
        assert crop["start_frame"] + FRAMES <= record["normalized_frames"] - 5 * RATE
        assert _sha(Path(authority.root) / crop["path"]) == crop["sha256"]


@pytest.mark.parametrize("mutator", [
    lambda value: value["sources"][next(iter(value["sources"]))]["crop"].__setitem__("start_frame", 0),
    lambda value: value["splits"].__setitem__("heldout", value["splits"]["validation"]),
    lambda value: value.__setitem__("aggregate_sha256", "0" * 64),
])
def test_rehashed_manifest_crop_split_and_aggregate_attacks_reject(authority, mutator) -> None:
    root = Path(authority.root); path = root / "manifest.json"; value = _manifest(authority)
    mutator(value); path.write_text(json.dumps(value)); _resign(path)
    with pytest.raises((ValueError, RuntimeError), match="crop|split|aggregate|authority|digest"):
        phase3.verify_corpus(authority)


def test_source_and_ffmpeg_authority_are_retained_and_checked(authority) -> None:
    manifest = _manifest(authority)
    assert manifest["source_authority"]["retained"] is True
    assert manifest["ffmpeg_authority"]["retained"] is True
    assert len(manifest["ffmpeg_authority"]["sha256"]) == 64
    assert Path(manifest["source_authority"]["descriptor_path"]).exists()
    with pytest.raises((ValueError, RuntimeError), match="ffmpeg|source|authority"):
        phase3.verify_corpus(authority, ffmpeg_path="/bin/false")


@pytest.mark.parametrize("attack", ("symlink", "hardlink", "writable-mode", "swapped-path"))
def test_corpus_path_mode_and_link_attacks_reject(authority, tmp_path: Path, attack: str) -> None:
    manifest = _manifest(authority); root = Path(authority.root)
    record = next(iter(manifest["sources"].values())); target = root / record["corpus_path"]
    if attack == "symlink":
        victim = tmp_path / "victim.wav"; victim.write_bytes(target.read_bytes()); target.unlink(); target.symlink_to(victim)
    elif attack == "hardlink":
        alias = root / "alias.wav"; os.link(target, alias); record["corpus_path"] = alias.name
        (root / "manifest.json").write_text(json.dumps(manifest)); _resign(root / "manifest.json")
    elif attack == "writable-mode":
        target.chmod(0o644)
    else:
        record["corpus_path"] = "../outside.wav"; (root / "manifest.json").write_text(json.dumps(manifest)); _resign(root / "manifest.json")
    with pytest.raises((ValueError, RuntimeError), match="path|link|mode|authority|immutable"):
        phase3.verify_corpus(authority)


@pytest.mark.parametrize("corruption, expected", [
    ("identity", "unchanged"), ("clipping", "worse"), ("noise", "worse"),
    ("silence", "worse"), ("phase_inversion", "worse"), ("lowpass", "worse"),
    ("duplicate", "reject"), ("shuffle", "reject"),
])
def test_objective_canaries_have_fixed_non_subjective_directions(authority, tmp_path: Path, corruption: str, expected: str) -> None:
    candidate = phase3.make_synthetic_candidate(authority, output_root=tmp_path / corruption, corruption=corruption)
    result = phase3.evaluate_objective_only(authority=authority, candidate_root=candidate, protected_baseline=None)
    if expected == "reject":
        assert result.status == "REJECTED"
    elif expected == "unchanged":
        assert result.objective["waveform_nmse"] == pytest.approx(0.0)
    else:
        assert result.status == "EVALUATED" and result.objective["waveform_nmse"] > 0


def test_rejects_nan_short_audio_and_missing_edit_is_not_executed(authority, tmp_path: Path) -> None:
    for corruption in ("nan", "short"):
        candidate = phase3.make_synthetic_candidate(authority, output_root=tmp_path / corruption, corruption=corruption)
        assert phase3.evaluate_objective_only(authority=authority, candidate_root=candidate, protected_baseline=None).status == "REJECTED"
    missing = phase3.make_synthetic_candidate(authority, output_root=tmp_path / "missing", corruption="missing")
    result = phase3.evaluate_objective_only(authority=authority, candidate_root=missing, protected_baseline=None)
    assert result.status == result.promotion == "NOT_EXECUTED" and not result.passed


def test_objective_only_rejects_learned_or_subjective_scores(authority, tmp_path: Path) -> None:
    candidate = phase3.make_synthetic_candidate(authority, output_root=tmp_path / "identity", corruption="identity")
    for score in ("clap", "mos", "human_preference", "learned_quality", "llm_judge"):
        with pytest.raises((ValueError, RuntimeError), match="objective|learned|subjective|forbidden"):
            phase3.evaluate_objective_only(authority=authority, candidate_root=candidate, protected_baseline=None, requested_scores=(score,))


def test_missing_optional_evaluators_are_explicit_not_zero_or_pass(authority, tmp_path: Path) -> None:
    candidate = phase3.make_synthetic_candidate(authority, output_root=tmp_path / "identity", corruption="identity")
    result = phase3.evaluate_objective_only(authority=authority, candidate_root=candidate, protected_baseline=None, optional_evaluators=("clap", "ast", "whisper"))
    assert all(result.optional[name]["status"] == "MISSING" and result.optional[name]["value"] is None for name in ("clap", "ast", "whisper"))


def test_board_enforces_hard_floors_protected_regressions_pareto_and_bootstrap_ci() -> None:
    board = phase3.PromotionBoard(hard_floors={"validation.waveform_nmse": 0.02}, protected_regressions={"heldout.waveform_nmse": 0.0}, bootstrap_samples=200, seed=41)
    baseline = {"validation.waveform_nmse": 0.04, "heldout.waveform_nmse": 0.04}
    strata = {"validation": [0.04, 0.02, 0.01], "heldout": [0.04, 0.03, 0.02, 0.03]}
    accepted = board.decide(baseline=baseline, candidate={"validation.waveform_nmse": 0.01, "heldout.waveform_nmse": 0.03}, strata=strata)
    assert accepted.status == "PROMOTE" and accepted.bootstrap_ci["validation.waveform_nmse"][1] < 0
    for candidate in ({"validation.waveform_nmse": 0.03, "heldout.waveform_nmse": 0.01}, {"validation.waveform_nmse": 0.01, "heldout.waveform_nmse": 0.05}, {"validation.waveform_nmse": 0.01, "heldout.waveform_nmse": 0.04}):
        assert board.decide(baseline=baseline, candidate=candidate, strata=strata).status == "REJECT"


def test_sealed_heldout_cannot_be_opened_during_candidate_tuning(authority, tmp_path: Path) -> None:
    with pytest.raises((PermissionError, ValueError, RuntimeError), match="heldout|sealed|tuning|access"):
        phase3.open_tuning_view(authority=authority, candidate_root=tmp_path / "candidate", requested_split="heldout")
    assert phase3.open_tuning_view(authority=authority, candidate_root=tmp_path / "candidate", requested_split="validation").split == "validation"


def test_candidate_digest_and_manifest_last_publication_detect_rehashed_attacks(authority, tmp_path: Path) -> None:
    candidate = phase3.make_synthetic_candidate(authority, output_root=tmp_path / "candidate", corruption="identity")
    result = phase3.evaluate_objective_only(authority=authority, candidate_root=candidate, protected_baseline=None)
    output, staged = tmp_path / "published", tmp_path / "staged"
    phase3.publish_evaluation(result=result, output_root=output, staging_root=staged)
    assert (output / "manifest.json").exists() and phase3.publication_order(output)[-1] == "manifest.json"
    for relative in ("candidate.json", "aggregate.json"):
        path = output / relative; value = json.loads(path.read_text()); value["forged"] = True; path.write_text(json.dumps(value)); _resign(path)
        with pytest.raises((ValueError, RuntimeError), match="candidate|aggregate|manifest|digest|authority"):
            phase3.verify_evaluation_bundle(output, authority=authority)
        path.unlink(); phase3.publish_evaluation(result=result, output_root=output, staging_root=staged)


def test_manifest_last_failure_rolls_back_without_partial_publication(authority, tmp_path: Path) -> None:
    candidate = phase3.make_synthetic_candidate(authority, output_root=tmp_path / "candidate", corruption="identity")
    result = phase3.evaluate_objective_only(authority=authority, candidate_root=candidate, protected_baseline=None)
    published, staged = tmp_path / "published", tmp_path / "staged"
    with pytest.raises(OSError):
        phase3.publish_evaluation(result=result, output_root=published, staging_root=staged, inject_failure_before_manifest=True)
    assert not published.exists() or not (published / "manifest.json").exists()