tests/test_glyph_assets.py

from __future__ import annotations

import hashlib
import json
import math
import os
import shutil
import wave
from dataclasses import dataclass, replace
from pathlib import Path

import numpy as np
import pytest

import glyph_assets as ga
from glyph_ascii_v1 import GLYPH_ASCII_V1_SYMBOLS


def _sha256(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def _write_pcm16_wav(
    path: Path,
    pcm: np.ndarray,
    *,
    sample_rate: int = ga.SAMPLE_RATE,
    channels: int = 1,
    sample_width: int = 2,
) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    array = np.asarray(pcm, dtype="<i2")
    if channels == 2:
        array = np.repeat(array[:, None], 2, axis=1).reshape(-1)
    with wave.open(str(path), "wb") as writer:
        writer.setnchannels(channels)
        writer.setsampwidth(sample_width)
        writer.setframerate(sample_rate)
        if sample_width == 2:
            writer.writeframes(array.astype("<i2").tobytes())
        else:
            writer.writeframes(np.full(array.size, 128, dtype=np.uint8).tobytes())


def _asset_pcm(index: int, sample_count: int = 12_000) -> np.ndarray:
    pcm = np.zeros(sample_count, dtype="<i2")
    start = 240
    end = sample_count - ga.FINAL_SILENCE_SAMPLES
    phase = np.arange(end - start, dtype=np.float64)
    frequency = 170.0 + index * 11.0
    amplitude = 5_000.0 + index * 101.0
    voiced = np.rint(
        amplitude
        * np.sin(2.0 * np.pi * frequency * phase / ga.SAMPLE_RATE)
    ).astype("<i2")
    trailing_fade = np.linspace(
        1.0,
        0.0,
        num=ga.TAIL_FADE_SAMPLES,
        endpoint=True,
        dtype=np.float64,
    )
    voiced[-ga.TAIL_FADE_SAMPLES :] = np.rint(
        voiced[-ga.TAIL_FADE_SAMPLES :].astype(np.float64) * trailing_fade
    ).astype("<i2")
    pcm[start:end] = voiced
    return pcm


def _pin(name: str, index: int) -> dict[str, object]:
    return {
        "name": name,
        "uri": f"https://example.invalid/{name}",
        "revision": f"revision-{index}",
        "sha256": hashlib.sha256(f"{name}-{index}".encode()).hexdigest(),
    }


def _entry_for_file(
    spec: ga.GlyphSpec,
    *,
    index: int,
    relative_path: str,
    path: Path,
    core_start: int = 360,
    core_end: int = 3_000,
) -> dict[str, object]:
    container = path.read_bytes()
    with wave.open(str(path), "rb") as reader:
        sample_count = reader.getnframes()
        pcm16_le = reader.readframes(sample_count)
    measurements = ga.measure_pcm16(pcm16_le)
    return {
        "id": spec.asset_id,
        "raw_byte": spec.raw_byte,
        "canonical_token": spec.canonical_token,
        "relative_path": relative_path,
        "container_sha256": _sha256(container),
        "pcm_sha256": _sha256(pcm16_le),
        "sample_rate": ga.SAMPLE_RATE,
        "channels": 1,
        "sample_format": ga.PCM_SAMPLE_FORMAT,
        "sample_count": sample_count,
        "rms": measurements.rms,
        "peak": measurements.peak,
        "active_duration_seconds": measurements.active_duration_seconds,
        "core_start_sample": core_start,
        "core_end_sample": core_end,
        "core_pcm_sha256": _sha256(pcm16_le[core_start * 2 : core_end * 2]),
        "ecapa": {
            "cosine_similarity": 0.31 + index * 0.001,
            "minimum_cosine_similarity": ga.MIN_ECAPA_SIMILARITY,
        },
        "squim": {
            "stoi": 0.80,
            "pesq": 1.50,
            "si_sdr": 1.0,
            "minimum_stoi": ga.MIN_SQUIM_STOI,
            "minimum_pesq": ga.MIN_SQUIM_PESQ,
            "minimum_si_sdr": ga.MIN_SQUIM_SI_SDR,
        },
        "endpoint": {
            "stop_reason": "stop_token",
            "turbo_cer": 0.0,
            "large_v3_cer": 0.0,
            "prefix_cer": 0.0,
            "suffix_cer": 0.0,
            "extra_tail_units": 0,
            "tail_energy_ratio": 0.001,
            "maximum_tail_energy_ratio": ga.MAX_ENDPOINT_TAIL_ENERGY_RATIO,
            "terminal_silence_samples": measurements.terminal_silence_samples,
        },
    }


@dataclass
class BundleFixture:
    root: Path
    manifest_path: Path
    manifest: dict[str, object]
    manifest_sha256: str

    def write(self, *, recompute_bundle: bool = True) -> str:
        if recompute_bundle:
            self.manifest["bundle_sha256"] = ga.compute_bundle_sha256(
                str(self.manifest["bundle_version"]),
                self.manifest["assets"],
            )
        data = json.dumps(
            self.manifest,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        ).encode()
        self.manifest_path.write_bytes(data)
        self.manifest_sha256 = _sha256(data)
        return self.manifest_sha256

    def load(self) -> ga.GlyphAssetBundle:
        return ga.load_glyph_asset_bundle(
            self.manifest_path,
            expected_manifest_sha256=self.manifest_sha256,
        )


@pytest.fixture
def bundle_fixture(tmp_path: Path) -> BundleFixture:
    entries: list[dict[str, object]] = []
    for index, spec in enumerate(ga.V1_GLYPH_SPECS):
        relative_path = f"assets/{index:02d}-{spec.asset_id}.wav"
        path = tmp_path / relative_path
        _write_pcm16_wav(path, _asset_pcm(index))
        entries.append(
            _entry_for_file(
                spec,
                index=index,
                relative_path=relative_path,
                path=path,
            )
        )
    manifest: dict[str, object] = {
        "manifest_schema": ga.MANIFEST_SCHEMA,
        "bundle_version": "speaker-b-ordinal-glyphs-v1",
        "grammar_sha256": ga.V1_GRAMMAR_SHA256,
        "bundle_sha256": "0" * 64,
        "speaker_anchor_sha256": hashlib.sha256(b"speaker-b-anchor").hexdigest(),
        "join_profile": dict(ga.V1_JOIN_PROFILE),
        "source_pins": [_pin("asset-builder", 1)],
        "model_pins": [_pin("bluemagpie-checkpoint", 2)],
        "evaluator_pins": [
            _pin("whisper-turbo", 3),
            _pin("whisper-large-v3", 4),
            _pin("ecapa", 5),
            _pin("squim", 6),
        ],
        "assets": entries,
    }
    result = BundleFixture(
        root=tmp_path,
        manifest_path=tmp_path / "manifest.json",
        manifest=manifest,
        manifest_sha256="",
    )
    result.write()
    return result


def test_loads_exact_immutable_32_entry_manifest(bundle_fixture: BundleFixture):
    bundle = bundle_fixture.load()

    assert len(bundle.assets) == ga.ASSET_COUNT == 32
    assert bundle.grammar_sha256 == ga.V1_GRAMMAR_SHA256
    assert bundle.join_profile == ga.V1_JOIN_PROFILE
    assert bundle.asset_by_raw_byte("a").asset_id == "letter-a"
    assert bundle.asset_by_raw_byte(b"@").asset_id == "symbol-at"
    assert bundle.asset_by_raw_byte(ord("/")).asset_id == "symbol-slash"
    assert (
        bundle.asset_by_id("letter-z").canonical_token
        == "第二十六個英文字母"
    )
    with pytest.raises(TypeError):
        bundle.join_profile["speed"] = 0.9
    decoded = bundle.assets[0].decoded_pcm16()
    assert not decoded.flags.writeable
    with pytest.raises(ValueError):
        decoded[0] = 1


def test_renderer_and_bundle_share_the_exact_expanded_32_entry_grammar():
    renderer_rows = tuple(
        (
            symbol.asset_id,
            chr(symbol.raw_byte),
            symbol.canonical_token,
        )
        for symbol in GLYPH_ASCII_V1_SYMBOLS
    )
    bundle_rows = tuple(
        (spec.asset_id, spec.raw_byte, spec.canonical_token)
        for spec in ga.V1_GLYPH_SPECS
    )

    assert len(renderer_rows) == len(bundle_rows) == ga.ASSET_COUNT == 32
    assert renderer_rows == bundle_rows


def test_external_manifest_hash_is_mandatory_and_detects_rewrite(
    bundle_fixture: BundleFixture,
):
    original_sha = bundle_fixture.manifest_sha256
    bundle_fixture.manifest["bundle_version"] = "attacker-rewrite"
    bundle_fixture.write()

    with pytest.raises(ga.GlyphAssetError, match="manifest hash mismatch"):
        ga.load_glyph_asset_bundle(
            bundle_fixture.manifest_path,
            expected_manifest_sha256=original_sha,
        )
    with pytest.raises(ga.GlyphAssetError, match="lowercase SHA-256"):
        ga.load_glyph_asset_bundle(
            bundle_fixture.manifest_path,
            expected_manifest_sha256="not-a-hash",
        )


def test_internal_bundle_hash_detects_asset_metadata_rewrite(
    bundle_fixture: BundleFixture,
):
    bundle_fixture.manifest["assets"][0]["rms"] = 0.123
    bundle_fixture.write(recompute_bundle=False)

    with pytest.raises(ga.GlyphAssetError, match="bundle hash mismatch"):
        bundle_fixture.load()


def test_container_tamper_detected_even_when_manifest_bytes_are_unchanged(
    bundle_fixture: BundleFixture,
):
    path = bundle_fixture.root / bundle_fixture.manifest["assets"][0]["relative_path"]
    data = bytearray(path.read_bytes())
    data[-7] ^= 0x01
    path.write_bytes(data)

    with pytest.raises(ga.GlyphAssetError, match="container hash mismatch"):
        bundle_fixture.load()


def test_wav_hash_and_decode_use_same_nofollow_open_under_toctou_swap(
    bundle_fixture: BundleFixture,
    monkeypatch: pytest.MonkeyPatch,
):
    entry = bundle_fixture.manifest["assets"][0]
    target = bundle_fixture.root / entry["relative_path"]
    original_container_sha = _sha256(target.read_bytes())
    replacement = bundle_fixture.root / "replacement.wav"
    _write_pcm16_wav(replacement, _asset_pcm(31))
    replacement_sha = _sha256(replacement.read_bytes())
    real_wave_open = ga.wave.open
    swapped = False

    def swap_path_then_decode(source, mode):
        nonlocal swapped
        # The decoder must receive the already-read immutable container, never
        # a path that can be exchanged after hashing.
        assert not isinstance(source, (str, os.PathLike))
        if not swapped:
            os.replace(replacement, target)
            swapped = True
        return real_wave_open(source, mode)

    monkeypatch.setattr(ga.wave, "open", swap_path_then_decode)

    bundle = bundle_fixture.load()

    assert swapped
    assert bundle.assets[0].container_sha256 == original_container_sha
    assert _sha256(bundle.assets[0].container_bytes) == original_container_sha
    assert _sha256(target.read_bytes()) == replacement_sha != original_container_sha


@pytest.mark.parametrize(
    "mutation,match",
    [
        (lambda assets: assets.reverse(), "frozen v1 grammar/order"),
        (lambda assets: assets.pop(), "exactly 32"),
        (
            lambda assets: assets[1].__setitem__("id", assets[0]["id"]),
            "frozen v1 grammar/order",
        ),
        (
            lambda assets: assets[1].__setitem__(
                "raw_byte", assets[0]["raw_byte"]
            ),
            "frozen v1 grammar/order",
        ),
        (
            lambda assets: assets[1].__setitem__(
                "canonical_token", assets[0]["canonical_token"]
            ),
            "frozen v1 grammar/order",
        ),
        (
            lambda assets: assets[0].__setitem__("raw_byte", "A"),
            "frozen v1 grammar/order",
        ),
    ],
    ids=("reorder", "omit", "duplicate-id", "duplicate-raw", "duplicate-token", "substitute"),
)
def test_frozen_grammar_rejects_reorder_omit_duplicate_and_substitute(
    bundle_fixture: BundleFixture,
    mutation,
    match: str,
):
    mutation(bundle_fixture.manifest["assets"])
    bundle_fixture.write()

    with pytest.raises(ga.GlyphAssetError, match=match):
        bundle_fixture.load()


def test_duplicate_audio_hash_is_rejected_after_each_file_verifies(
    bundle_fixture: BundleFixture,
):
    entries = bundle_fixture.manifest["assets"]
    first = entries[0]
    second = entries[1]
    first_path = bundle_fixture.root / first["relative_path"]
    second_path = bundle_fixture.root / second["relative_path"]
    shutil.copyfile(first_path, second_path)
    for field in (
        "container_sha256",
        "pcm_sha256",
        "sample_count",
        "rms",
        "peak",
        "active_duration_seconds",
        "core_start_sample",
        "core_end_sample",
        "core_pcm_sha256",
        "endpoint",
    ):
        second[field] = first[field]
    bundle_fixture.write()

    with pytest.raises(ga.GlyphAssetError, match="duplicate container hash"):
        bundle_fixture.load()


def test_missing_asset_is_rejected(bundle_fixture: BundleFixture):
    entry = bundle_fixture.manifest["assets"][5]
    (bundle_fixture.root / entry["relative_path"]).unlink()

    with pytest.raises(ga.GlyphAssetError, match="missing"):
        bundle_fixture.load()


def test_asset_path_traversal_is_rejected_before_open(
    bundle_fixture: BundleFixture,
):
    entry = bundle_fixture.manifest["assets"][0]
    entry["relative_path"] = "../outside.wav"
    bundle_fixture.write()

    with pytest.raises(ga.GlyphAssetError, match="unsafe components"):
        bundle_fixture.load()


def test_asset_symlink_is_rejected(bundle_fixture: BundleFixture):
    entry = bundle_fixture.manifest["assets"][0]
    original = bundle_fixture.root / entry["relative_path"]
    target = bundle_fixture.root / "real-copy.wav"
    shutil.copyfile(original, target)
    original.unlink()
    original.symlink_to(target)

    with pytest.raises(ga.GlyphAssetError, match="symlink"):
        bundle_fixture.load()


def test_manifest_symlink_is_rejected(bundle_fixture: BundleFixture):
    link = bundle_fixture.root / "manifest-link.json"
    link.symlink_to(bundle_fixture.manifest_path)

    with pytest.raises(ga.GlyphAssetError, match="manifest must not be a symlink"):
        ga.load_glyph_asset_bundle(
            link,
            expected_manifest_sha256=bundle_fixture.manifest_sha256,
        )


@pytest.mark.parametrize(
    "sample_rate,channels,sample_width,match",
    [
        (44_100, 1, 2, "48 kHz"),
        (48_000, 2, 2, "mono"),
        (48_000, 1, 1, "PCM16"),
    ],
    ids=("wrong-rate", "stereo", "pcm8"),
)
def test_decoded_wav_format_is_recomputed_not_trusted(
    bundle_fixture: BundleFixture,
    sample_rate: int,
    channels: int,
    sample_width: int,
    match: str,
):
    entry = bundle_fixture.manifest["assets"][0]
    path = bundle_fixture.root / entry["relative_path"]
    _write_pcm16_wav(
        path,
        _asset_pcm(0),
        sample_rate=sample_rate,
        channels=channels,
        sample_width=sample_width,
    )
    entry["container_sha256"] = _sha256(path.read_bytes())
    bundle_fixture.write()

    with pytest.raises(ga.GlyphAssetError, match=match):
        bundle_fixture.load()


def test_declared_oversized_asset_is_rejected(bundle_fixture: BundleFixture):
    bundle_fixture.manifest["assets"][0]["sample_count"] = (
        ga.MAX_ASSET_SAMPLES + 1
    )
    bundle_fixture.write()

    with pytest.raises(ga.GlyphAssetError, match="must be <="):
        bundle_fixture.load()


def test_asset_requires_180_ms_of_exact_zero_tail(
    bundle_fixture: BundleFixture,
):
    entries = bundle_fixture.manifest["assets"]
    original = entries[0]
    path = bundle_fixture.root / original["relative_path"]
    pcm = _asset_pcm(0)
    # This is below the active-duration threshold, so only an exact-zero check
    # can detect it.
    pcm[-1] = 1
    _write_pcm16_wav(path, pcm)
    entries[0] = _entry_for_file(
        ga.V1_GLYPH_SPECS[0],
        index=0,
        relative_path=original["relative_path"],
        path=path,
    )
    bundle_fixture.write()

    with pytest.raises(ga.GlyphAssetError, match="180 ms of exact zero PCM"):
        bundle_fixture.load()


@pytest.mark.parametrize(
    "sample_index,value,match",
    [
        (0, 10_000, "first PCM code must be exact zero"),
        (1, 10_000, "3 ms leading fade envelope"),
    ],
    ids=("accepted-click-boundary", "leading-envelope-bypass"),
)
def test_manifest_recomputes_leading_click_contract(
    bundle_fixture: BundleFixture,
    sample_index: int,
    value: int,
    match: str,
):
    original = bundle_fixture.manifest["assets"][0]
    path = bundle_fixture.root / original["relative_path"]
    pcm = _asset_pcm(0)
    pcm[sample_index] = value
    _write_pcm16_wav(path, pcm)
    bundle_fixture.manifest["assets"][0] = _entry_for_file(
        ga.V1_GLYPH_SPECS[0],
        index=0,
        relative_path=original["relative_path"],
        path=path,
    )
    bundle_fixture.write()

    with pytest.raises(ga.GlyphAssetError, match=match):
        bundle_fixture.load()


def test_fake_silent_pcm_with_passing_attestations_never_becomes_verified(
    bundle_fixture: BundleFixture,
):
    original = bundle_fixture.manifest["assets"][0]
    path = bundle_fixture.root / original["relative_path"]
    _write_pcm16_wav(path, np.zeros(12_000, dtype="<i2"))
    silent_entry = _entry_for_file(
        ga.V1_GLYPH_SPECS[0],
        index=0,
        relative_path=original["relative_path"],
        path=path,
    )
    # _entry_for_file deliberately writes apparently passing ECAPA/SQUIM/ASR
    # provenance; integrity checks still know the PCM contains no speech.
    bundle_fixture.manifest["assets"][0] = silent_entry
    bundle_fixture.write()

    with pytest.raises(ga.GlyphAssetError, match="must contain active speech"):
        bundle_fixture.load()


@pytest.mark.parametrize(
    "field_path,value,match",
    [
        (("rms",), float("nan"), "not canonical JSON"),
        (
            ("ecapa", "minimum_cosine_similarity"),
            0.24,
            "attested threshold",
        ),
        (("squim", "minimum_stoi"), 0.70, "attested thresholds"),
        (
            ("endpoint", "maximum_tail_energy_ratio"),
            0.03,
            "attested tail threshold",
        ),
        (("endpoint", "stop_reason"), "claimed-pass", "allowed completion"),
    ],
)
def test_attestation_schema_and_profile_are_validated_without_claiming_a_pass(
    bundle_fixture: BundleFixture,
    field_path: tuple[str, ...],
    value: object,
    match: str,
):
    target = bundle_fixture.manifest["assets"][0]
    for key in field_path[:-1]:
        target = target[key]
    target[field_path[-1]] = value
    if isinstance(value, float) and not math.isfinite(value):
        data = json.dumps(
            bundle_fixture.manifest,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
            allow_nan=True,
        ).encode()
        bundle_fixture.manifest_path.write_bytes(data)
        bundle_fixture.manifest_sha256 = _sha256(data)
    else:
        bundle_fixture.write()

    with pytest.raises(ga.GlyphAssetError, match=match):
        bundle_fixture.load()


def test_failing_attested_metrics_are_provenance_not_runtime_verification(
    bundle_fixture: BundleFixture,
):
    first = bundle_fixture.manifest["assets"][0]
    first["ecapa"]["cosine_similarity"] = -0.8
    first["squim"]["stoi"] = 0.01
    first["squim"]["pesq"] = -0.4
    first["squim"]["si_sdr"] = -200.0
    first["endpoint"]["turbo_cer"] = 1.0
    first["endpoint"]["large_v3_cer"] = 1.0
    first["endpoint"]["prefix_cer"] = 1.0
    first["endpoint"]["suffix_cer"] = 1.0
    first["endpoint"]["extra_tail_units"] = 10
    first["endpoint"]["tail_energy_ratio"] = 1.0
    bundle_fixture.write()

    bundle = bundle_fixture.load()

    assert bundle.runtime_quality_verified is False
    assert bundle.assets[0].runtime_quality_verified is False


def test_untrusted_pass_boolean_is_not_part_of_the_schema(
    bundle_fixture: BundleFixture,
):
    bundle_fixture.manifest["assets"][0]["ecapa"]["passed"] = True
    bundle_fixture.write()

    with pytest.raises(ga.GlyphAssetError, match="keys do not match schema"):
        bundle_fixture.load()


def _carrier(segment_id: str, *, frequency: float = 220.0) -> ga.CarrierSegment:
    samples = np.arange(3_000, dtype=np.float64)
    audio = (
        0.04 * np.sin(2.0 * np.pi * frequency * samples / ga.SAMPLE_RATE)
    ).astype(np.float32)
    seed = int.from_bytes(
        hashlib.sha256(f"{segment_id}:seed".encode()).digest()[:4],
        "big",
    ) % (2**31)
    return ga.CarrierSegment(
        segment_id=segment_id,
        audio=audio,
        selected_attempt_id=f"attempt:{segment_id}",
        selected_seed=seed,
        selected_text_sha256=_sha256(f"{segment_id}:text".encode()),
        selected_endpoint_evidence_sha256=_sha256(
            f"{segment_id}:endpoint".encode()
        ),
    )


def _example_segments() -> tuple[ga.HybridSegment, ...]:
    return (
        _carrier("carrier-before"),
        ga.AssetSegment("glyph-a", "letter-a"),
        ga.AssetSegment("glyph-at", "symbol-at"),
        _carrier("carrier-after", frequency=260.0),
        ga.AssetSegment("glyph-z", "letter-z"),
    )


def _commitment(
    segments: tuple[ga.HybridSegment, ...],
    *,
    salt: str = "test-plan",
    discarded_attempts: int = 0,
) -> ga.SemanticPlanCommitment:
    carriers = tuple(
        segment for segment in segments if isinstance(segment, ga.CarrierSegment)
    )
    records: list[ga.GenerationAttemptRecord] = []
    for carrier in carriers:
        records.append(
            ga.GenerationAttemptRecord(
                ordinal=len(records),
                attempt_id=carrier.selected_attempt_id,
                seed=carrier.selected_seed,
                text_sha256=carrier.selected_text_sha256,
                endpoint_evidence_sha256=(
                    carrier.selected_endpoint_evidence_sha256
                ),
                generated_chunk_count=1,
                generated_text_units=10,
            )
        )
    for index in range(discarded_attempts):
        attempt_id = f"discarded:{salt}:{index}"
        records.append(
            ga.GenerationAttemptRecord(
                ordinal=len(records),
                attempt_id=attempt_id,
                seed=10_000 + index,
                text_sha256=_sha256(f"{attempt_id}:text".encode()),
                endpoint_evidence_sha256=_sha256(
                    f"{attempt_id}:endpoint".encode()
                ),
                generated_chunk_count=1,
                generated_text_units=7,
            )
        )
    ordered_records = tuple(records)
    return ga.SemanticPlanCommitment(
        external_semantic_plan_sha256=_sha256(f"{salt}:semantic".encode()),
        inverse_proof_sha256=_sha256(f"{salt}:inverse".encode()),
        ordered_attempt_ledger_sha256=(
            ga.compute_ordered_attempt_ledger_sha256(ordered_records)
        ),
        attempt_record_count=len(ordered_records),
        assembled_carrier_count=len(carriers),
        request_model_generator_invocations=len(ordered_records),
        request_generated_chunk_count=sum(
            record.generated_chunk_count for record in ordered_records
        ),
        request_generated_text_units=sum(
            record.generated_text_units for record in ordered_records
        ),
        request_generated_chunk_budget=ga.MAX_GENERATED_CHUNK_BUDGET,
        request_generated_text_unit_budget=(
            ga.MAX_GENERATED_TEXT_UNIT_BUDGET
        ),
        ordered_attempt_records=ordered_records,
    )


def _commitment_with_records(
    commitment: ga.SemanticPlanCommitment,
    records: tuple[ga.GenerationAttemptRecord, ...],
) -> ga.SemanticPlanCommitment:
    return replace(
        commitment,
        ordered_attempt_ledger_sha256=(
            ga.compute_ordered_attempt_ledger_sha256(records)
        ),
        attempt_record_count=len(records),
        request_model_generator_invocations=len(records),
        request_generated_chunk_count=sum(
            record.generated_chunk_count for record in records
        ),
        request_generated_text_units=sum(
            record.generated_text_units for record in records
        ),
        ordered_attempt_records=records,
    )


def _asset_only_zero_work_commitment(
    *,
    salt: str = "asset-only-zero-work",
) -> ga.SemanticPlanCommitment:
    records: tuple[ga.GenerationAttemptRecord, ...] = ()
    return ga.SemanticPlanCommitment(
        external_semantic_plan_sha256=_sha256(f"{salt}:semantic".encode()),
        inverse_proof_sha256=_sha256(f"{salt}:inverse".encode()),
        ordered_attempt_ledger_sha256=(
            ga.compute_ordered_attempt_ledger_sha256(records)
        ),
        attempt_record_count=0,
        assembled_carrier_count=0,
        request_model_generator_invocations=0,
        request_generated_chunk_count=0,
        request_generated_text_units=0,
        request_generated_chunk_budget=ga.MAX_GENERATED_CHUNK_BUDGET,
        request_generated_text_unit_budget=ga.MAX_GENERATED_TEXT_UNIT_BUDGET,
        ordered_attempt_records=records,
    )


def test_asset_only_assembly_accepts_exact_zero_work_attempt_ledger(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = (
        ga.AssetSegment("glyph-a", "letter-a"),
        ga.AssetSegment("glyph-at", "symbol-at"),
    )
    commitment = _asset_only_zero_work_commitment()
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle,
        segments,
        semantic_commitment=commitment,
    )

    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )

    assert assembly.ledger.attempt_record_count == 0
    assert assembly.ledger.request_model_generator_invocations == 0
    assert assembly.ledger.request_generated_chunk_count == 0
    assert assembly.ledger.request_generated_text_units == 0
    ga.verify_hybrid_assembly(
        assembly,
        trusted_bundle=bundle,
        original_segments=segments,
        expected_semantic_commitment=commitment,
    )


def test_asset_only_assembly_rejects_hidden_or_fabricated_model_work(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = (ga.AssetSegment("glyph-a", "letter-a"),)
    fake = _commitment(
        segments,
        salt="fabricated-asset-only",
        discarded_attempts=1,
    )

    with pytest.raises(
        ga.GlyphAssetError,
        match="asset-only assembly must have an exact zero-work attempt ledger",
    ):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=fake,
        )


def test_carrier_assembly_still_rejects_empty_attempt_ledger(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = (_carrier("carrier-required"),)
    empty = replace(
        _asset_only_zero_work_commitment(salt="carrier-empty"),
        assembled_carrier_count=1,
    )

    with pytest.raises(
        ga.GlyphAssetError,
        match="carrier assembly requires a non-empty attempt ledger",
    ):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=empty,
        )


def test_hybrid_assembly_is_deterministic_public_equals_verified_and_idempotent(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle, segments, semantic_commitment=commitment
    )

    first = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    second = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )

    assert first == second
    assert first.public_pcm16_le is first.verification_pcm16_le
    assert first.public_pcm16_le == first.verification_pcm16_le
    assert np.array_equal(first.public_pcm16(), first.verification_pcm16())
    assert (
        ga.pcm16_array_to_bytes(ga.pcm16_bytes_to_array(first.pcm16_le))
        == first.pcm16_le
    )
    ga.verify_hybrid_assembly(
        first,
        trusted_bundle=bundle,
        original_segments=segments,
        expected_semantic_commitment=commitment,
    )


def test_asset_core_samples_and_hash_survive_join_and_final_tail(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = (
        _carrier("carrier"),
        ga.AssetSegment("last-glyph", "letter-z"),
    )
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle, segments, semantic_commitment=commitment
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    output = assembly.verification_pcm16()
    asset = bundle.asset_by_id("letter-z")
    asset_source = asset.decoded_pcm16()
    span = next(span for span in assembly.ledger.spans if span.kind == "asset")

    output_core = output[
        span.output_core_start_sample : span.output_core_end_sample
    ]
    source_core = asset_source[
        asset.core_start_sample : asset.core_end_sample
    ]
    assert np.array_equal(output_core, source_core)
    assert ga.pcm16_array_to_bytes(output_core) == ga.pcm16_array_to_bytes(
        source_core
    )
    assert _sha256(ga.pcm16_array_to_bytes(output_core)) == asset.core_pcm_sha256
    assert span.output_core_pcm_sha256 == asset.core_pcm_sha256
    assert span.output_core_end_sample <= assembly.ledger.tail_fade_start_sample
    assert span.pcm_sha256 == asset.pcm_sha256
    assert np.array_equal(
        output[span.start_sample : span.end_sample],
        asset_source,
    )


def test_existing_asset_180ms_zero_tail_is_not_appended_twice(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = (ga.AssetSegment("only-glyph", "letter-a"),)
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle,
        segments,
        semantic_commitment=commitment,
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    asset = bundle.asset_by_id("letter-a")

    assert len(assembly.pcm16_le) == len(asset.pcm16_le)
    assert assembly.pcm16_le == asset.pcm16_le
    assert all(span.kind != "final_silence" for span in assembly.ledger.spans)
    assert (
        assembly.ledger.final_silence_end_sample
        - assembly.ledger.final_silence_start_sample
        == ga.FINAL_SILENCE_SAMPLES
    )


def test_final_tail_only_tops_up_missing_exact_zero_samples(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = (_carrier("only-carrier"),)
    commitment = _commitment(segments)
    conditioned = ga.pcm16_bytes_to_array(ga.condition_carrier_v1(segments[0]))
    existing_zero = min(
        ga.FINAL_SILENCE_SAMPLES,
        ga._terminal_exact_zero_samples(conditioned),
    )
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle,
        segments,
        semantic_commitment=commitment,
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )

    assert (
        assembly.ledger.total_samples
        == conditioned.size + ga.FINAL_SILENCE_SAMPLES - existing_zero
    )
    final_span = next(
        span for span in assembly.ledger.spans if span.kind == "final_silence"
    )
    assert (
        final_span.end_sample - final_span.start_sample
        == ga.FINAL_SILENCE_SAMPLES - existing_zero
    )


def test_ledger_ranges_are_monotonic_exact_complete_and_zero_terminated(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle, segments, semantic_commitment=commitment
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    ledger = assembly.ledger
    output = assembly.verification_pcm16()

    assert ledger.spans[0].start_sample == 0
    assert ledger.spans[-1].end_sample == ledger.total_samples == output.size
    assert all(
        left.end_sample == right.start_sample
        for left, right in zip(ledger.spans, ledger.spans[1:])
    )
    # The two adjacent assets are plain decoder-exact PCM concatenation.
    assert [span.kind for span in ledger.spans].count("pause") == 3
    first_asset = next(span for span in ledger.spans if span.segment_id == "glyph-a")
    second_asset = next(
        span for span in ledger.spans if span.segment_id == "glyph-at"
    )
    assert first_asset.end_sample == second_asset.start_sample
    assert all(
        span.pcm_sha256 == span.source_pcm_sha256
        for span in ledger.spans
        if span.kind == "asset"
    )
    assert ledger.tail_fade_end_sample == ledger.final_silence_start_sample
    assert (
        ledger.tail_fade_end_sample - ledger.tail_fade_start_sample
        == ga.TAIL_FADE_SAMPLES
    )
    assert (
        ledger.final_silence_end_sample - ledger.final_silence_start_sample
        == ga.FINAL_SILENCE_SAMPLES
    )
    assert np.count_nonzero(output[ledger.final_silence_start_sample :]) == 0
    assert output[ledger.tail_fade_end_sample - 1] == 0


def test_boundary_jump_metrics_cover_every_span_boundary_and_are_silent(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle, segments, semantic_commitment=commitment
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )

    assert len(assembly.ledger.boundary_jumps) == len(assembly.ledger.spans) - 1
    assert all(
        jump.sample_index == left.end_sample == right.start_sample
        for jump, left, right in zip(
            assembly.ledger.boundary_jumps,
            assembly.ledger.spans,
            assembly.ledger.spans[1:],
        )
    )
    assert max(
        jump.absolute_jump for jump in assembly.ledger.boundary_jumps
    ) <= 1.0 / 32768.0


def test_assembler_rejects_accepted_click_boundary_even_with_forged_asset_object(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    victim = bundle.asset_by_id("letter-b")
    malicious_pcm = bytearray(victim.pcm16_le)
    malicious_pcm[:2] = np.asarray([10_000], dtype="<i2").tobytes()
    malicious_bytes = bytes(malicious_pcm)
    measurements = ga.measure_pcm16(malicious_bytes)
    forged_asset = replace(
        victim,
        pcm16_le=malicious_bytes,
        pcm_sha256=_sha256(malicious_bytes),
        rms=measurements.rms,
        peak=measurements.peak,
        active_duration_seconds=measurements.active_duration_seconds,
    )
    forged_bundle = replace(
        bundle,
        assets=tuple(
            forged_asset if asset.asset_id == "letter-b" else asset
            for asset in bundle.assets
        ),
    )
    segments = (
        ga.AssetSegment("left", "letter-a"),
        ga.AssetSegment("right", "letter-b"),
    )
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        forged_bundle,
        segments,
        semantic_commitment=commitment,
    )

    with pytest.raises(ga.GlyphAssetError, match="boundary.*click limit"):
        ga.assemble_hybrid_pcm16(
            forged_bundle,
            segments,
            semantic_commitment=commitment,
            expected_plan_sha256=plan_sha,
        )


@pytest.mark.parametrize(
    "mutate",
    [
        lambda segments: tuple(reversed(segments)),
        lambda segments: segments[:-1],
        lambda segments: segments[:1]
        + (ga.AssetSegment("glyph-a", "letter-b"),)
        + segments[2:],
    ],
    ids=("reorder", "omit", "substitute"),
)
def test_original_plan_hash_rejects_reorder_omit_and_substitute(
    bundle_fixture: BundleFixture,
    mutate,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle, segments, semantic_commitment=commitment
    )

    with pytest.raises(ga.GlyphAssetError, match="plan hash mismatch"):
        ga.assemble_hybrid_pcm16(
            bundle,
            mutate(segments),
            semantic_commitment=commitment,
            expected_plan_sha256=plan_sha,
        )


def test_duplicate_segment_id_is_rejected(bundle_fixture: BundleFixture):
    bundle = bundle_fixture.load()
    segments = (
        _carrier("same"),
        ga.AssetSegment("same", "letter-a"),
    )

    with pytest.raises(ga.GlyphAssetError, match="duplicate segment id"):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=_commitment(segments),
        )


@pytest.mark.parametrize(
    "segment_id",
    (
        "__bluemagpie_internal__:pause:1",
        "__bluemagpie_internal__:final-silence",
        "pause-1",
        "final-silence",
    ),
)
def test_user_segment_ids_cannot_enter_internal_namespace(
    bundle_fixture: BundleFixture,
    segment_id: str,
):
    bundle = bundle_fixture.load()
    segments = (ga.AssetSegment(segment_id, "letter-a"),)

    with pytest.raises(ga.GlyphAssetError, match="reserved internal"):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=_commitment(segments),
        )


def test_internal_and_user_segment_ids_are_globally_unique(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle,
        segments,
        semantic_commitment=commitment,
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    segment_ids = [span.segment_id for span in assembly.ledger.spans]

    assert len(segment_ids) == len(set(segment_ids))
    assert all(
        span.segment_id.startswith("__bluemagpie_internal__:")
        for span in assembly.ledger.spans
        if span.kind in ("pause", "final_silence")
    )


@pytest.mark.parametrize(
    "mutation",
    (
        lambda value: replace(
            value,
            external_semantic_plan_sha256=_sha256(b"changed-plan"),
        ),
        lambda value: replace(
            value,
            inverse_proof_sha256=_sha256(b"changed-inverse"),
        ),
    ),
    ids=("semantic-plan", "inverse-proof"),
)
def test_audio_plan_hard_binds_external_semantic_and_budget_commitments(
    bundle_fixture: BundleFixture,
    mutation,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle,
        segments,
        semantic_commitment=commitment,
    )

    with pytest.raises(ga.GlyphAssetError, match="plan hash mismatch"):
        ga.assemble_hybrid_pcm16(
            bundle,
            segments,
            semantic_commitment=mutation(commitment),
            expected_plan_sha256=plan_sha,
        )


def test_semantic_commitment_must_honestly_count_assembled_carriers(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)

    with pytest.raises(ga.GlyphAssetError, match="assembled carrier total"):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=replace(commitment, assembled_carrier_count=1),
        )
    with pytest.raises(ga.GlyphAssetError, match="chunk budget is not frozen"):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=replace(
                commitment,
                request_generated_chunk_budget=31,
            ),
        )
    with pytest.raises(ga.GlyphAssetError, match="text-unit budget is not frozen"):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=replace(
                commitment,
                request_generated_text_unit_budget=799,
            ),
        )


def test_discarded_and_retry_attempts_are_legal_and_fully_accounted(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments, discarded_attempts=3)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle,
        segments,
        semantic_commitment=commitment,
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )

    assert commitment.assembled_carrier_count == 2
    assert commitment.request_model_generator_invocations == 5
    assert commitment.request_generated_chunk_count == 5
    assert commitment.request_generated_text_units == 41
    assert assembly.ledger.attempt_record_count == 5
    ga.verify_hybrid_assembly(
        assembly,
        trusted_bundle=bundle,
        original_segments=segments,
        expected_semantic_commitment=commitment,
    )


def test_hidden_attempt_record_rejected_by_external_ledger_commitment(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    hidden = ga.GenerationAttemptRecord(
        ordinal=len(commitment.ordered_attempt_records),
        attempt_id="hidden:retry",
        seed=77,
        text_sha256=_sha256(b"hidden:text"),
        endpoint_evidence_sha256=_sha256(b"hidden:endpoint"),
        generated_chunk_count=1,
        generated_text_units=7,
    )
    forged = replace(
        commitment,
        ordered_attempt_records=commitment.ordered_attempt_records + (hidden,),
    )

    with pytest.raises(ga.GlyphAssetError, match="ledger hash mismatch"):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=forged,
        )


def test_omitted_discarded_attempt_rejected_by_external_ledger_commitment(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments, discarded_attempts=1)
    forged = replace(
        commitment,
        ordered_attempt_records=commitment.ordered_attempt_records[:-1],
    )

    with pytest.raises(ga.GlyphAssetError, match="ledger hash mismatch"):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=forged,
        )


def test_omitted_selected_attempt_rejected_even_with_recomputed_ledger_hash(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    omitted = _commitment_with_records(
        commitment,
        tuple(
            replace(record, ordinal=index)
            for index, record in enumerate(
                commitment.ordered_attempt_records[1:]
            )
        ),
    )

    with pytest.raises(
        ga.GlyphAssetError,
        match="omit selected carrier|not attributable",
    ):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=omitted,
        )


def test_duplicate_attempt_id_is_rejected_before_hash_acceptance(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    first, second = commitment.ordered_attempt_records
    duplicate = (
        first,
        replace(second, attempt_id=first.attempt_id),
    )
    forged = replace(commitment, ordered_attempt_records=duplicate)

    with pytest.raises(ga.GlyphAssetError, match="duplicate attempt id"):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=forged,
        )


@pytest.mark.parametrize(
    "field,value_delta,match",
    (
        ("attempt_record_count", 1, "record count mismatch"),
        (
            "request_model_generator_invocations",
            1,
            "invocation total",
        ),
        ("request_generated_chunk_count", 1, "chunk total"),
        ("request_generated_text_units", 1, "text-unit total"),
    ),
)
def test_wrong_request_attempt_totals_are_rejected(
    bundle_fixture: BundleFixture,
    field: str,
    value_delta: int,
    match: str,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    forged = replace(
        commitment,
        **{field: getattr(commitment, field) + value_delta},
    )

    with pytest.raises(ga.GlyphAssetError, match=match):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=forged,
        )


@pytest.mark.parametrize(
    "field,value,match",
    (
        ("selected_attempt_id", "missing:selected", "not attributable"),
        ("selected_seed", 123_456, "does not match its attempt"),
        (
            "selected_text_sha256",
            _sha256(b"wrong:selected:text"),
            "does not match its attempt",
        ),
        (
            "selected_endpoint_evidence_sha256",
            _sha256(b"wrong:selected:endpoint"),
            "does not match its attempt",
        ),
    ),
)
def test_each_final_carrier_requires_exact_selected_attempt_attribution(
    bundle_fixture: BundleFixture,
    field: str,
    value: object,
    match: str,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    forged_first = replace(segments[0], **{field: value})
    forged_segments = (forged_first,) + segments[1:]

    with pytest.raises(ga.GlyphAssetError, match=match):
        ga.compute_hybrid_plan_sha256(
            bundle,
            forged_segments,
            semantic_commitment=commitment,
        )


def test_request_chunk_and_text_unit_cap_exceed_are_rejected(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    over_chunks = _commitment(segments, discarded_attempts=31)

    with pytest.raises(ga.GlyphAssetError, match="must be <= 32"):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=over_chunks,
        )

    commitment = _commitment(segments)
    records = list(commitment.ordered_attempt_records)
    records[0] = replace(records[0], generated_text_units=800)
    over_units = _commitment_with_records(commitment, tuple(records))
    with pytest.raises(ga.GlyphAssetError, match="must be <= 800"):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=over_units,
        )


def test_carrier_request_generator_invocations_must_match_nonempty_ledger(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = replace(
        _commitment(segments),
        request_model_generator_invocations=0,
    )

    with pytest.raises(
        ga.GlyphAssetError,
        match="request generator invocation total does not match attempt ledger",
    ):
        ga.compute_hybrid_plan_sha256(
            bundle,
            segments,
            semantic_commitment=commitment,
        )


def test_no_speed_or_gain_transform_is_permitted_after_asset_insertion(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    valid = _example_segments()
    commitment = _commitment(valid)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle, valid, semantic_commitment=commitment
    )

    with pytest.raises(ga.GlyphAssetError, match="must be exactly 1.0"):
        ga.assemble_hybrid_pcm16(
            bundle,
            valid,
            semantic_commitment=commitment,
            expected_plan_sha256=plan_sha,
            speed=0.99,
        )
    with pytest.raises(ga.GlyphAssetError, match="post_gain must be exactly 1.0"):
        ga.assemble_hybrid_pcm16(
            bundle,
            valid,
            semantic_commitment=commitment,
            expected_plan_sha256=plan_sha,
            post_gain=0.99,
        )
    with pytest.raises(ga.GlyphAssetError, match="must be exactly 1.0"):
        ga.compute_hybrid_plan_sha256(
            bundle,
            (ga.AssetSegment("glyph", "letter-a", speed=1.01),),
            semantic_commitment=_commitment(
                (ga.AssetSegment("glyph-ok", "letter-a"),)
            ),
        )
    with pytest.raises(ga.GlyphAssetError, match="must be exactly 1.0"):
        invalid_carrier = replace(
            _carrier("carrier"),
            audio=_carrier("unused").audio,
            speed=0.95,
        )
        ga.compute_hybrid_plan_sha256(
            bundle,
            (invalid_carrier,),
            semantic_commitment=_commitment((invalid_carrier,)),
        )


def test_each_carrier_is_conditioned_before_join_without_cross_segment_gain(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    first = _carrier("first", frequency=180.0)
    second = _carrier("second", frequency=320.0)
    segments = (
        first,
        ga.AssetSegment("asset", "letter-a"),
        second,
        ga.AssetSegment("last", "letter-b"),
    )
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle, segments, semantic_commitment=commitment
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    output = assembly.verification_pcm16()

    for carrier in (first, second):
        expected = ga.pcm16_bytes_to_array(ga.condition_carrier_v1(carrier))
        span = next(
            span
            for span in assembly.ledger.spans
            if span.segment_id == carrier.segment_id
        )
        actual = output[span.start_sample : span.end_sample]
        assert np.array_equal(actual, expected)


def test_carrier_nonfinite_audio_is_rejected(bundle_fixture: BundleFixture):
    bundle = bundle_fixture.load()
    audio = np.ones(1_000, dtype=np.float32)
    audio[500] = math.nan
    invalid_carrier = replace(_carrier("bad"), audio=audio)

    with pytest.raises(ga.GlyphAssetError, match="non-finite"):
        ga.compute_hybrid_plan_sha256(
            bundle,
            (invalid_carrier,),
            semantic_commitment=_commitment((invalid_carrier,)),
        )


def test_public_pcm_tamper_fails_independent_ledger_verification(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle, segments, semantic_commitment=commitment
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    tampered = bytearray(assembly.pcm16_le)
    tampered[1_001] ^= 0x01
    replaced = replace(assembly, pcm16_le=bytes(tampered))

    with pytest.raises(ga.GlyphAssetError, match="span PCM hash mismatch"):
        ga.verify_hybrid_assembly(
            replaced,
            trusted_bundle=bundle,
            original_segments=segments,
            expected_semantic_commitment=commitment,
        )


def test_boundary_jump_metric_tamper_is_recomputed(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle, segments, semantic_commitment=commitment
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    first_jump = replace(
        assembly.ledger.boundary_jumps[0],
        absolute_jump=0.99,
    )
    forged_ledger = replace(
        assembly.ledger,
        boundary_jumps=(first_jump,) + assembly.ledger.boundary_jumps[1:],
    )
    forged = replace(assembly, ledger=forged_ledger)

    with pytest.raises(ga.GlyphAssetError, match="boundary jump evidence"):
        ga.verify_hybrid_assembly(
            forged,
            trusted_bundle=bundle,
            original_segments=segments,
            expected_semantic_commitment=commitment,
        )


def test_forged_silence_and_self_consistent_digest_ledger_is_rejected(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle,
        segments,
        semantic_commitment=commitment,
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    silent_pcm = np.zeros(assembly.ledger.total_samples, dtype="<i2")
    silent_bytes = silent_pcm.tobytes()
    forged_spans = []
    for span in assembly.ledger.spans:
        span_bytes = silent_pcm[span.start_sample : span.end_sample].tobytes()
        span_sha = _sha256(span_bytes)
        core_sha = (
            _sha256(
                silent_pcm[
                    span.output_core_start_sample : span.output_core_end_sample
                ].tobytes()
            )
            if span.kind == "asset"
            else None
        )
        forged_spans.append(
            replace(
                span,
                pcm_sha256=span_sha,
                source_pcm_sha256=(
                    span_sha if span.kind in ("asset", "carrier") else None
                ),
                output_core_pcm_sha256=core_sha,
            )
        )
    forged_jumps = tuple(
        replace(jump, absolute_jump=0.0)
        for jump in assembly.ledger.boundary_jumps
    )
    forged_ledger = replace(
        assembly.ledger,
        spans=tuple(forged_spans),
        boundary_jumps=forged_jumps,
        output_pcm_sha256=_sha256(silent_bytes),
    )
    forged = replace(
        assembly,
        pcm16_le=silent_bytes,
        ledger=forged_ledger,
    )

    # The attacker updated every ledger digest consistently.  Verification
    # must still rebuild PCM from trusted carriers/assets and reject silence.
    with pytest.raises(ga.GlyphAssetError, match="trusted reconstruction"):
        ga.verify_hybrid_assembly(
            forged,
            trusted_bundle=bundle,
            original_segments=segments,
            expected_semantic_commitment=commitment,
        )


@pytest.mark.parametrize(
    "field,value",
    (
        ("asset_id", "letter-b"),
        ("raw_byte", "b"),
        ("canonical_token", "第二個英文字母"),
        ("segment_id", "attacker-relabelled"),
    ),
)
def test_verifier_rejects_mutated_asset_labels_and_segment_identity(
    bundle_fixture: BundleFixture,
    field: str,
    value: str,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle,
        segments,
        semantic_commitment=commitment,
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    index = next(
        index
        for index, span in enumerate(assembly.ledger.spans)
        if span.segment_id == "glyph-a"
    )
    forged_span = replace(assembly.ledger.spans[index], **{field: value})
    forged_spans = list(assembly.ledger.spans)
    forged_spans[index] = forged_span
    forged = replace(
        assembly,
        ledger=replace(assembly.ledger, spans=tuple(forged_spans)),
    )

    with pytest.raises(ga.GlyphAssetError, match="trusted reconstruction"):
        ga.verify_hybrid_assembly(
            forged,
            trusted_bundle=bundle,
            original_segments=segments,
            expected_semantic_commitment=commitment,
        )


@pytest.mark.parametrize(
    "field",
    ("bundle_sha256", "plan_sha256"),
)
def test_verifier_rejects_forged_bundle_and_plan_digests(
    bundle_fixture: BundleFixture,
    field: str,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle,
        segments,
        semantic_commitment=commitment,
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    forged = replace(
        assembly,
        ledger=replace(
            assembly.ledger,
            **{field: _sha256(f"forged-{field}".encode())},
        ),
    )

    with pytest.raises(ga.GlyphAssetError, match="evidence mismatch"):
        ga.verify_hybrid_assembly(
            forged,
            trusted_bundle=bundle,
            original_segments=segments,
            expected_semantic_commitment=commitment,
        )


def test_verifier_rejects_forged_semantic_commitment_ledger_fields(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle,
        segments,
        semantic_commitment=commitment,
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    forged = replace(
        assembly,
        ledger=replace(
            assembly.ledger,
            inverse_proof_sha256=_sha256(b"forged-inverse"),
        ),
    )

    with pytest.raises(ga.GlyphAssetError, match="semantic commitment"):
        ga.verify_hybrid_assembly(
            forged,
            trusted_bundle=bundle,
            original_segments=segments,
            expected_semantic_commitment=commitment,
        )


def test_verifier_reconditions_original_carrier_instead_of_trusting_ledger(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle,
        segments,
        semantic_commitment=commitment,
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    changed_segments = (
        _carrier("carrier-before", frequency=440.0),
    ) + segments[1:]

    with pytest.raises(ga.GlyphAssetError, match="plan evidence mismatch"):
        ga.verify_hybrid_assembly(
            assembly,
            trusted_bundle=bundle,
            original_segments=changed_segments,
            expected_semantic_commitment=commitment,
        )


def test_verifier_recomputes_trusted_bundle_not_just_ledger_bundle_string(
    bundle_fixture: BundleFixture,
):
    bundle = bundle_fixture.load()
    segments = _example_segments()
    commitment = _commitment(segments)
    plan_sha = ga.compute_hybrid_plan_sha256(
        bundle,
        segments,
        semantic_commitment=commitment,
    )
    assembly = ga.assemble_hybrid_pcm16(
        bundle,
        segments,
        semantic_commitment=commitment,
        expected_plan_sha256=plan_sha,
    )
    forged_bundle = replace(bundle, bundle_sha256=_sha256(b"forged-bundle"))

    with pytest.raises(ga.GlyphAssetError, match="retained manifest projection"):
        ga.verify_hybrid_assembly(
            assembly,
            trusted_bundle=forged_bundle,
            original_segments=segments,
            expected_semantic_commitment=commitment,
        )