from __future__ import annotations

import ast
from dataclasses import dataclass
import hashlib
import json
from pathlib import Path
from types import MappingProxyType, SimpleNamespace

import numpy as np
import pytest

from glyph_assets import (
    AssetSegment,
    GenerationAttemptRecord,
    GlyphAssetError,
    SemanticPlanCommitment,
    compute_ordered_attempt_ledger_sha256,
)
from glyph_hybrid_plan import GlyphHybridPlanError
from latency_timing import latency_stage
from production import count_speech_units, pcm16_audio_output
from quality_runtime import CandidateGenerationContext


ROOT = Path(__file__).resolve().parents[1]
APP_PATH = ROOT / "app.py"


class FakeGradioError(Exception):
    pass


def _isolated_nodes(names: tuple[str, ...], namespace: dict[str, object]):
    namespace.setdefault("latency_stage", latency_stage)
    tree = ast.parse(APP_PATH.read_text(encoding="utf-8"))
    selected = [
        node
        for node in tree.body
        if isinstance(node, (ast.FunctionDef, ast.ClassDef)) and node.name in names
    ]
    observed = {node.name for node in selected}
    assert observed == set(names)
    module = ast.Module(
        body=[
            ast.ImportFrom(
                module="__future__",
                names=[ast.alias(name="annotations")],
                level=0,
            ),
            *selected,
        ],
        type_ignores=[],
    )
    ast.fix_missing_locations(module)
    exec(compile(module, APP_PATH, "exec"), namespace)
    return SimpleNamespace(**{name: namespace[name] for name in names})


def _function_source(name: str) -> str:
    source = APP_PATH.read_text(encoding="utf-8")
    tree = ast.parse(source)
    node = next(
        item
        for item in tree.body
        if isinstance(item, ast.FunctionDef) and item.name == name
    )
    result = ast.get_source_segment(source, node)
    assert result is not None
    return result


def test_schema7_adapter_uses_frozen_per_return_gate_thresholds():
    source = APP_PATH.read_text(encoding="utf-8")
    start = source.index(
        "_GLYPH_HYBRID_EVIDENCE_ADAPTER = HardenedGlyphHybridAdapter("
    )
    end = source.index("\ndef _network_fragment_proof_rows", start)
    adapter_assignment = source[start:end]

    assert (
        "min_speaker_similarity=(\n"
        "                QUALITY_RELEASE_MIN_SPEAKER_SIMILARITY"
    ) in adapter_assignment
    assert (
        "max_boundary_speaker_drop=(\n"
        "                QUALITY_RELEASE_MAX_BOUNDARY_SPEAKER_DROP"
    ) in adapter_assignment
    assert (
        "min_squim_stoi=QUALITY_PREFERRED_MIN_SQUIM_STOI"
        in adapter_assignment
    )
    assert (
        "min_squim_pesq=QUALITY_PREFERRED_MIN_SQUIM_PESQ"
        in adapter_assignment
    )


def test_profile_selection_requires_exact_default_speaker_object_identity():
    default = object()
    namespace = {
        "DEFAULT_SPEAKER": "內建語者 B",
        "DEFAULT_CENTROID": default,
        "SPEAKERS": {
            "內建語者 A": object(),
            "內建語者 B": default,
        },
    }
    selected = _isolated_nodes(
        ("_glyph_profile_selected",),
        namespace,
    )._glyph_profile_selected

    assert selected("內建語者 B") is True
    assert selected("內建語者 A") is False
    assert selected("female_voice") is False
    namespace["SPEAKERS"]["內建語者 B"] = object()
    assert selected("內建語者 B") is False


def test_network_route_is_pre_generation_fail_closed_and_never_partial_fallback():
    calls: list[tuple[str, object]] = []
    available = SimpleNamespace(
        available=True,
        asset_entry_sha256_by_asset_id=MappingProxyType({"asset": "0" * 64}),
    )
    current_state = {"value": available}
    namespace = {
        "gr": SimpleNamespace(Error=FakeGradioError),
        "GlyphHybridPlanError": GlyphHybridPlanError,
        "GlyphAssetError": GlyphAssetError,
        "_glyph_profile_selected": lambda speaker: speaker == "default",
        "_glyph_request_network_looking": (
            lambda text: any(marker in str(text).lower() for marker in ("@", "http://"))
        ),
        "_validate_glyph_runtime_controls": (
            lambda **kwargs: calls.append(("controls", kwargs))
        ),
        "_GLYPH_PROFILE_STATE": available,
        "build_glyph_hybrid_plan": (
            lambda raw, **kwargs: calls.append(("plan", raw)) or None
        ),
        "_synthesize_glyph_hybrid": (
            lambda *args, **kwargs: calls.append(("synthesize", args[0]))
            or (48_000, np.zeros(8, dtype=np.float32))
        ),
    }
    route = _isolated_nodes(
        ("_maybe_synthesize_glyph_speaker_request",),
        namespace,
    )._maybe_synthesize_glyph_speaker_request

    assert (
        route("普通文字", "default", cfg=3.0, steps=10, speed=1.0)
        is None
    )
    assert calls == []
    assert (
        route("bad@network", "other", cfg=3.0, steps=10, speed=1.0)
        is None
    )
    assert calls == []

    with pytest.raises(FakeGradioError, match="嚴格小寫 ASCII"):
        route("bad@network", "default", cfg=3.0, steps=10, speed=1.0)
    assert [name for name, _ in calls] == ["controls", "plan"]
    assert not any(name == "synthesize" for name, _ in calls)

    calls.clear()
    namespace["_GLYPH_PROFILE_STATE"] = SimpleNamespace(
        available=False,
        unavailable_reason="hash_mismatch",
    )
    with pytest.raises(FakeGradioError, match="資產目前不可用"):
        route("bad@network", "default", cfg=3.0, steps=10, speed=1.0)
    assert [name for name, _ in calls] == ["controls"]


def test_valid_strict_plan_is_the_only_network_path_that_reaches_synthesis():
    plan = object()
    calls: list[str] = []
    state = SimpleNamespace(
        available=True,
        asset_entry_sha256_by_asset_id=MappingProxyType({"asset": "0" * 64}),
    )
    namespace = {
        "gr": SimpleNamespace(Error=FakeGradioError),
        "GlyphHybridPlanError": GlyphHybridPlanError,
        "GlyphAssetError": GlyphAssetError,
        "_glyph_profile_selected": lambda speaker: speaker == "default",
        "_glyph_request_network_looking": lambda text: "@" in str(text),
        "_validate_glyph_runtime_controls": lambda **kwargs: calls.append("controls"),
        "_GLYPH_PROFILE_STATE": state,
        "build_glyph_hybrid_plan": (
            lambda *args, **kwargs: calls.append("plan") or plan
        ),
        "_synthesize_glyph_hybrid": (
            lambda raw, exact_plan, exact_state, **kwargs: (
                calls.append("synthesize")
                or (
                    (48_000, np.zeros(8, dtype=np.float32))
                    if exact_plan is plan and exact_state is state
                    else None
                )
            )
        ),
    }
    route = _isolated_nodes(
        ("_maybe_synthesize_glyph_speaker_request",),
        namespace,
    )._maybe_synthesize_glyph_speaker_request

    result = route(
        "mead@forest.org",
        "default",
        cfg=3.0,
        steps=10,
        speed=1.0,
    )

    assert result[0] == 48_000
    assert calls == ["controls", "plan", "synthesize"]


def test_request_attempt_ledger_accounts_rejected_candidates_and_selection():
    namespace = {
        "dataclass": dataclass,
        "np": np,
        "hashlib": hashlib,
        "GenerationAttemptRecord": GenerationAttemptRecord,
        "CandidateGenerationContext": CandidateGenerationContext,
        "count_speech_units": count_speech_units,
        "GLYPH_MAX_GENERATED_CHUNK_BUDGET": 32,
        "GLYPH_MAX_GENERATED_TEXT_UNIT_BUDGET": 800,
        "_canonical_json_sha256": (
            lambda value: hashlib.sha256(
                json.dumps(
                    value,
                    ensure_ascii=True,
                    separators=(",", ":"),
                    sort_keys=True,
                ).encode("ascii")
            ).hexdigest()
        ),
    }
    api = _isolated_nodes(
        (
            "_GlyphAttemptReservation",
            "_GlyphCompletedAttempt",
            "_GlyphRequestAttemptLedger",
            "_GlyphCarrierGenerationAudit",
        ),
        namespace,
    )
    ledger = api._GlyphRequestAttemptLedger()
    audit = api._GlyphCarrierGenerationAudit(ledger, "carrier-0")

    for candidate_index, text in enumerate(("第一個候選", "第二個候選")):
        context = CandidateGenerationContext(
            candidate_index=candidate_index,
            seed=100 + candidate_index,
            chunk_indices=(0,),
            chunk_candidate_ordinals=(candidate_index,),
        )
        reservation = audit.begin(
            (text,),
            100 + candidate_index,
            context,
            candidate_index,
            False,
        )
        audit.complete(
            reservation,
            (
                SimpleNamespace(
                    stop_reason="threshold",
                    endpoint_energy_ratio=0.001,
                    generated_steps=5,
                    hard_stop_steps=7,
                    audio=np.ones(32, dtype=np.float32),
                ),
            ),
        )

    selected_waveform = np.linspace(-0.1, 0.1, 64, dtype=np.float32)
    audit.select(
        SimpleNamespace(chunk_candidate_indices=(1,)),
        selected_waveform,
    )

    assert len(ledger.records) == 2
    assert len(ledger.completed_attempts) == 2
    assert [record.ordinal for record in ledger.records] == [0, 1]
    assert audit.selected() == ledger.records[1]
    assert np.array_equal(audit.selected_waveform(), selected_waveform)
    assert compute_ordered_attempt_ledger_sha256(ledger.records)


def test_pure_asset_integration_makes_zero_model_calls_and_still_emits_schema7():
    pcm = np.array([0, 100, -100, 0], dtype="<i2")
    pcm_bytes = pcm.tobytes()

    class FakeAssembly:
        sample_rate = 48_000
        pcm16_le = pcm_bytes
        public_pcm16_le = pcm_bytes
        verification_pcm16_le = pcm_bytes

        def public_pcm16(self):
            return pcm.copy()

    class EmptyLedger:
        records: tuple[GenerationAttemptRecord, ...] = ()

    plan = SimpleNamespace(
        plan_sha256="1" * 64,
        canonical_target="第一個英文字母",
        canonical_target_sha256=hashlib.sha256(
            "第一個英文字母".encode("utf-8")
        ).hexdigest(),
        segments=(
            SimpleNamespace(
                ordinal=0,
                kind="asset_atom",
                asset_id="letter-a",
            ),
        ),
    )
    bundle = object()
    state = SimpleNamespace(
        available=True,
        bundle=bundle,
        asset_entry_sha256_by_asset_id=MappingProxyType({"letter-a": "2" * 64}),
    )
    events: list[tuple[str, object]] = []
    final = SimpleNamespace()
    namespace = {
        "np": np,
        "hashlib": hashlib,
        "GlyphHybridPlanError": GlyphHybridPlanError,
        "AssetSegment": AssetSegment,
        "CarrierSegment": object,
        "SemanticPlanCommitment": SemanticPlanCommitment,
        "_GlyphRequestAttemptLedger": EmptyLedger,
        "glyph_hybrid_plan_sha256": lambda value: value.plan_sha256,
        "inverse_glyph_hybrid_plan": lambda *args, **kwargs: "a@b.co",
        "_glyph_inverse_proof_sha256": lambda *args: "3" * 64,
        "resolve_request_seed": lambda *args: 123,
        "secrets": SimpleNamespace(randbelow=lambda upper: 123),
        "_validate_glyph_carrier_text": lambda text: None,
        "_GlyphCarrierGenerationAudit": object,
        "_synthesize": lambda *args, **kwargs: (_ for _ in ()).throw(
            AssertionError("pure asset request called the model")
        ),
        "DEFAULT_CENTROID": object(),
        "SR": 48_000,
        "compute_ordered_attempt_ledger_sha256": (
            compute_ordered_attempt_ledger_sha256
        ),
        "GLYPH_MAX_GENERATED_CHUNK_BUDGET": 32,
        "GLYPH_MAX_GENERATED_TEXT_UNIT_BUDGET": 800,
        "compute_hybrid_plan_sha256": (
            lambda *args, **kwargs: events.append(("plan", kwargs)) or "4" * 64
        ),
        "assemble_hybrid_pcm16": (
            lambda *args, **kwargs: events.append(("assemble", kwargs))
            or FakeAssembly()
        ),
        "verify_hybrid_assembly": (
            lambda *args, **kwargs: events.append(("verify", kwargs))
        ),
        "pcm16_audio_output": pcm16_audio_output,
        "_speaker_anchor_array": lambda centroid: np.ones(2, dtype=np.float32),
        "WholeWaveformVerificationCache": object,
        "_verify_trajectory_audio": lambda *args, **kwargs: final,
        "QUALITY_FINAL_ASR_MAX_NEW_TOKENS": 440,
        "transcribe_breeze25": object(),
        "_transcribe_glyph_hybrid_breeze_primary": object(),
        "_transcribe_glyph_hybrid_breeze_confirmation": object(),
        "_require_glyph_final_verification": lambda value: None,
        "_verify_independent_whole_audio": lambda *args, **kwargs: final,
        "FinalOutputRejectedError": RuntimeError,
        "gr": SimpleNamespace(Error=FakeGradioError),
        "_emit_glyph_hybrid_evidence": (
            lambda **kwargs: events.append(("schema7", kwargs))
        ),
    }
    synthesize = _isolated_nodes(
        ("_synthesize_glyph_hybrid",),
        namespace,
    )._synthesize_glyph_hybrid

    sample_rate, waveform = synthesize(
        "a@b.co",
        plan,
        state,
        cfg=3.0,
        steps=10,
        speed=1.0,
    )

    assert sample_rate == 48_000
    assert np.array_equal(
        pcm16_audio_output(sample_rate, waveform)[1],
        pcm,
    )
    assert [name for name, _ in events] == [
        "plan",
        "assemble",
        "verify",
        "schema7",
    ]
    assembly_kwargs = events[1][1]
    assert assembly_kwargs["speed"] == 1.0
    assert assembly_kwargs["post_gain"] == 1.0
    evidence_kwargs = events[-1][1]
    assert evidence_kwargs["request_ledger"].records == ()


def test_schema7_adapter_is_mandatory_canonical_content_free_and_pcm_bound(capsys):
    raw = "a@b.co"
    canonical = "第一個英文字母"
    pcm = np.array([0, 1, -1, 0], dtype="<i2").tobytes()
    plan = SimpleNamespace(
        plan_sha256="1" * 64,
        canonical_target_sha256=hashlib.sha256(
            canonical.encode("utf-8")
        ).hexdigest(),
    )
    assembly = SimpleNamespace(pcm16_le=pcm)
    ledger = SimpleNamespace(records=())
    speaker_anchor_sha256 = "7" * 64

    def payload():
        renderer_contract = {
            "speaker_anchor_sha256": speaker_anchor_sha256,
        }
        canonical_commitment = "c" * 64
        asset_id = "asset-opaque"
        asset_entry = "d" * 64
        asset_pcm = "e" * 64
        result = {
            "carrier_selections": [],
            "evidence_sha256": "",
            "renderer_contract": renderer_contract,
            "renderer_contract_sha256": hashlib.sha256(
                json.dumps(
                    renderer_contract,
                    ensure_ascii=True,
                    separators=(",", ":"),
                    sort_keys=True,
                ).encode("ascii")
            ).hexdigest(),
            "request_proof": {
                "additional_strict_gates": {
                    "asset_reconstruction": True,
                    "public_pcm_equality": True,
                    "strict_inverse_reconstruction": True,
                },
                "assembled_pcm_sha256": hashlib.sha256(pcm).hexdigest(),
                "budgets": [
                    {"actual": 0, "cap": 32, "name": "model_invocations"},
                    {"actual": 0, "cap": 32, "name": "generated_chunks"},
                    {
                        "actual": 0,
                        "cap": 800,
                        "name": "generated_speech_units",
                    },
                    {"actual": 1, "cap": 32, "name": "asset_count"},
                    {"actual": 1, "cap": 128, "name": "asset_atoms"},
                    {
                        "actual": 7,
                        "cap": 512,
                        "name": "asset_canonical_units",
                    },
                    {
                        "actual": 4,
                        "cap": 61_440_000,
                        "name": "asset_audio_samples",
                    },
                    {"actual": 2, "cap": 96, "name": "asr_calls"},
                    {"actual": 2, "cap": 96, "name": "ecapa_calls"},
                    {"actual": 2, "cap": 96, "name": "squim_calls"},
                ],
                "canonical_target_sha256": canonical_commitment,
                "content_commitment_scheme": "hmac-sha256-v1",
                "final_gate": {
                    "asr_call_id": "call-asr-final",
                    "passed": True,
                    "scope": "final",
                },
                "independent_gate": {
                    "asr_call_id": "call-asr-independent",
                    "passed": True,
                    "scope": "independent",
                },
                "public_pcm_sha256": hashlib.sha256(pcm).hexdigest(),
                "raw_input_sha256": "f" * 64,
                "segments": [
                    {
                        "asset_id": asset_id,
                        "asset_manifest_entry_sha256": asset_entry,
                        "kind": "asset",
                        "output_sample_count": len(pcm) // 2,
                        "realized_pcm_sha256": asset_pcm,
                        "segment_id": "segment-0000",
                    }
                ],
            },
            "runtime_call_ledger": [
                {
                    "asset_id": asset_id,
                    "call_id": "call-asset",
                    "kind": "asset_load",
                    "loaded_pcm_sha256": asset_pcm,
                    "loaded_sample_count": len(pcm) // 2,
                    "manifest_entry_sha256": asset_entry,
                    "segment_id": "segment-0000",
                },
                {
                    "call_id": "call-asr-final",
                    "kind": "asr",
                    "scope": "final",
                    "subject_model_call_id": None,
                    "target_sha256": canonical_commitment,
                },
                {
                    "call_id": "call-asr-independent",
                    "kind": "asr",
                    "scope": "independent",
                    "subject_model_call_id": None,
                    "target_sha256": canonical_commitment,
                },
            ],
            "schema": "bluemagpie.glyph-hybrid.evidence.v7",
            "version": 7,
        }
        digest_payload = dict(result)
        digest_payload.pop("evidence_sha256")
        result["evidence_sha256"] = hashlib.sha256(
            json.dumps(
                digest_payload,
                ensure_ascii=True,
                separators=(",", ":"),
                sort_keys=True,
            ).encode("ascii")
        ).hexdigest()
        return result

    namespace = {
        "json": json,
        "hashlib": hashlib,
        "_GLYPH_HYBRID_EVIDENCE_ADAPTER": None,
        "GLYPH_MAX_GENERATED_CHUNK_BUDGET": 32,
        "GLYPH_MAX_GENERATED_TEXT_UNIT_BUDGET": 800,
        "GLYPH_SPEAKER_ANCHOR_SHA256": speaker_anchor_sha256,
        "GLYPH_EVIDENCE_LOG_PREFIX": "[BlueMagpie][glyph-hybrid-evidence] ",
        "_canonical_json_sha256": (
            lambda value: hashlib.sha256(
                json.dumps(
                    value,
                    ensure_ascii=True,
                    separators=(",", ":"),
                    sort_keys=True,
                ).encode("ascii")
            ).hexdigest()
        ),
    }
    emit = _isolated_nodes(
        ("_emit_glyph_hybrid_evidence",),
        namespace,
    )._emit_glyph_hybrid_evidence
    kwargs = {
        "raw_text": raw,
        "plan": plan,
        "profile_state": object(),
        "segments": (),
        "assembly": assembly,
        "semantic_commitment": object(),
        "request_ledger": ledger,
        "final_verification": object(),
        "independent_final_verification": object(),
    }

    with pytest.raises(RuntimeError, match="adapter is unavailable"):
        emit(**kwargs)

    serialized = json.dumps(
        payload(),
        ensure_ascii=True,
        allow_nan=False,
        separators=(",", ":"),
        sort_keys=True,
    ).encode("ascii")
    namespace["_GLYPH_HYBRID_EVIDENCE_ADAPTER"] = lambda **values: serialized
    emit(**kwargs)
    output = capsys.readouterr().out
    assert output == (
        "[BlueMagpie][glyph-hybrid-evidence] "
        + serialized.decode("ascii")
        + "\n"
    )

    for mutate in (
        lambda value: value["request_proof"].__setitem__(
            "content_commitment_scheme",
            "plain-sha256",
        ),
        lambda value: value.__setitem__(
            "renderer_contract_sha256",
            "0" * 64,
        ),
        lambda value: value["runtime_call_ledger"][0].__setitem__(
            "loaded_pcm_sha256",
            "0" * 64,
        ),
        lambda value: value["runtime_call_ledger"][1].__setitem__(
            "target_sha256",
            "0" * 64,
        ),
    ):
        unbound = payload()
        mutate(unbound)
        unbound_digest_payload = dict(unbound)
        unbound_digest_payload.pop("evidence_sha256")
        unbound["evidence_sha256"] = hashlib.sha256(
            json.dumps(
                unbound_digest_payload,
                ensure_ascii=True,
                separators=(",", ":"),
                sort_keys=True,
            ).encode("ascii")
        ).hexdigest()
        unbound_bytes = json.dumps(
            unbound,
            ensure_ascii=True,
            allow_nan=False,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("ascii")
        namespace["_GLYPH_HYBRID_EVIDENCE_ADAPTER"] = (
            lambda **values: unbound_bytes
        )
        with pytest.raises(RuntimeError, match="not bound"):
            emit(**kwargs)

    carrier_bound = payload()
    carrier_bound["request_proof"]["segments"].append(
        {
            "carrier_text_sha256": "1" * 64,
            "kind": "carrier",
            "output_sample_count": len(pcm) // 2,
            "realized_pcm_sha256": "2" * 64,
            "segment_id": "segment-0001",
        }
    )
    carrier_bound["runtime_call_ledger"].append(
        {
            "call_id": "call-model",
            "kind": "model_generation",
            "output_pcm_sha256": "2" * 64,
            "output_sample_count": len(pcm) // 2,
            "segment_id": "segment-0001",
            "text_sha256": "1" * 64,
        }
    )
    carrier_bound["carrier_selections"] = [
        {
            "segment_id": "segment-0001",
            "selected_call_id": "call-model",
        }
    ]
    carrier_actuals = {
        "model_invocations": 1,
        "generated_chunks": 1,
        "generated_speech_units": 1,
    }
    for row in carrier_bound["request_proof"]["budgets"]:
        if row["name"] in carrier_actuals:
            row["actual"] = carrier_actuals[row["name"]]
    carrier_digest_payload = dict(carrier_bound)
    carrier_digest_payload.pop("evidence_sha256")
    carrier_bound["evidence_sha256"] = hashlib.sha256(
        json.dumps(
            carrier_digest_payload,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("ascii")
    ).hexdigest()
    carrier_bytes = json.dumps(
        carrier_bound,
        ensure_ascii=True,
        allow_nan=False,
        separators=(",", ":"),
        sort_keys=True,
    ).encode("ascii")
    namespace["_GLYPH_HYBRID_EVIDENCE_ADAPTER"] = (
        lambda **values: carrier_bytes
    )
    carrier_kwargs = {
        **kwargs,
        "request_ledger": SimpleNamespace(
            records=(
                SimpleNamespace(
                    generated_chunk_count=1,
                    generated_text_units=1,
                ),
            )
        ),
    }
    emit(**carrier_kwargs)

    carrier_bound["runtime_call_ledger"][-1][
        "output_pcm_sha256"
    ] = "3" * 64
    carrier_digest_payload = dict(carrier_bound)
    carrier_digest_payload.pop("evidence_sha256")
    carrier_bound["evidence_sha256"] = hashlib.sha256(
        json.dumps(
            carrier_digest_payload,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("ascii")
    ).hexdigest()
    carrier_bytes = json.dumps(
        carrier_bound,
        ensure_ascii=True,
        allow_nan=False,
        separators=(",", ":"),
        sort_keys=True,
    ).encode("ascii")
    namespace["_GLYPH_HYBRID_EVIDENCE_ADAPTER"] = (
        lambda **values: carrier_bytes
    )
    with pytest.raises(RuntimeError, match="not bound"):
        emit(**carrier_kwargs)

    leaking = payload()
    leaking["request_proof"]["raw_text"] = raw
    leaking_digest_payload = dict(leaking)
    leaking_digest_payload.pop("evidence_sha256")
    leaking["evidence_sha256"] = hashlib.sha256(
        json.dumps(
            leaking_digest_payload,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("ascii")
    ).hexdigest()
    leaking_bytes = json.dumps(
        leaking,
        ensure_ascii=True,
        allow_nan=False,
        separators=(",", ":"),
        sort_keys=True,
    ).encode("ascii")
    namespace["_GLYPH_HYBRID_EVIDENCE_ADAPTER"] = lambda **values: leaking_bytes
    with pytest.raises(RuntimeError, match="not bound"):
        emit(**kwargs)


def test_carrier_schema6_logs_are_suppressed_but_legacy_schema6_is_unchanged():
    source = _function_source("_synthesize")

    assert "if generation_audit is None:\n            print(line)" in source
    assert source.count("emit_public_cascade_evidence(") == 4
    assert source.count("format_cascade_evidence_log(") == 3
    assert "generation_audit.select(cascade, waveform)" in source


def test_wrappers_select_profile_before_generation_and_reference_stays_legacy():
    speaker = _function_source("tts_speaker")
    longform = _function_source("tts_longform")
    reference = _function_source("tts_reference")

    for source in (speaker, longform):
        assert source.index("_maybe_synthesize_glyph_speaker_request(") < (
            source.index("*_synthesize(")
        )
        assert "if glyph_result is not None:" in source
    assert "_maybe_synthesize_glyph_speaker_request(" not in reference
    assert "*_synthesize(" in reference


def test_hybrid_assembly_has_no_post_asset_global_audio_transform():
    source = _function_source("_synthesize_glyph_hybrid")

    for forbidden in (
        "_apply_speed(",
        "apply_loudness_floor(",
        "finish_audio(",
        "join_audio_chunks_variable(",
        "match_chunk_rms(",
        "crossfade",
    ):
        assert forbidden not in source
    assert "speed=1.0" in source
    assert "post_gain=1.0" in source
    assert "pcm16_audio_output(assembly.sample_rate, waveform)" in source


def test_hybrid_only_breeze25_wrappers_pin_the_160_segment_override():
    primary = _function_source("_transcribe_glyph_hybrid_breeze_primary")
    confirmation = _function_source(
        "_transcribe_glyph_hybrid_breeze_confirmation"
    )
    hybrid = _function_source("_synthesize_glyph_hybrid")

    assert (
        "max_verification_segments: int = "
        "GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS"
    ) in primary
    assert (
        "max_verification_segments: int = "
        "GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS"
    ) in confirmation
    for source in (primary, confirmation):
        assert "transcribe_breeze25(" in source
        assert "language=language" in source
        assert "task=task" in source
        assert "max_new_tokens=max_new_tokens" in source
        assert "max_verification_segments=max_verification_segments" in source
    assert "transcriber=_transcribe_glyph_hybrid_breeze_primary" in hybrid
    assert "transcriber=_transcribe_glyph_hybrid_breeze_confirmation" in hybrid