glyph_hybrid_adapter.py

"""Hardened production adapter for schema-7 glyph-hybrid evidence.

The adapter is the only bridge from the hosted runtime's text-bearing objects
to :mod:`glyph_hybrid_evidence`. It rebuilds the renderer plan, replays every
model attempt, reruns two explicitly distinct decoder profiles over the same
pinned Breeze ASR 25 weights plus ECAPA and SQUIM on exact public PCM,
validates the immutable assembly, and only then returns content-free canonical
JSON.
"""

from __future__ import annotations

from dataclasses import asdict, dataclass, is_dataclass
import hashlib
import json
import math
from pathlib import Path
import secrets
from typing import Any, Callable, Mapping, Sequence

import numpy as np

import glyph_ascii_v1 as glyph_ascii_module
import glyph_assets as glyph_assets_module
import glyph_hybrid_evidence as evidence
import glyph_hybrid_plan as glyph_plan_module
import quality_runtime as quality_runtime_module
from glyph_assets import (
    AssetSegment,
    CarrierSegment,
    condition_carrier_v1,
    verify_hybrid_assembly,
)
from glyph_hybrid_plan import build_glyph_hybrid_plan
from quality_runtime import (
    SQUIM_OBJECTIVE_MAX_WINDOWS,
    SQUIM_OBJECTIVE_SAMPLE_RATE,
    SQUIM_OBJECTIVE_WEIGHT_SHA256,
    SQUIM_OBJECTIVE_WINDOW_SECONDS,
    release_speaker_evidence_from_audio,
    squim_objective_evidence_from_audio,
)


ADAPTER_SCHEMA = "bluemagpie.glyph-hybrid.adapter.v1"
GENERATION_POLICY_SCHEMA = "bluemagpie.glyph-generation-policy.v1"
ASR_TASK = "transcribe"
BREEZE25_MODEL_ID = getattr(
    quality_runtime_module,
    "BREEZE25_MODEL_ID",
    "MediaTek-Research/Breeze-ASR-25",
)
BREEZE25_REVISION = getattr(
    quality_runtime_module,
    "BREEZE25_REVISION",
    "cffe7ccb404d025296a00758d0a33468bec3a9d0",
)
BREEZE25_PRIMARY_PIN_NAME = "breeze25-primary-zh"
BREEZE25_CONFIRMATION_PIN_NAME = "breeze25-confirmation-auto"
BREEZE25_PRIMARY_DECODER_PROFILE = "breeze25-greedy-zh-v1"
BREEZE25_CONFIRMATION_DECODER_PROFILE = "breeze25-greedy-auto-v1"
BREEZE25_PRIMARY_LANGUAGE = "zh"
BREEZE25_CONFIRMATION_LANGUAGE = None
BREEZE25_WEIGHT_SHA256 = getattr(
    quality_runtime_module,
    "BREEZE25_WEIGHT_SHA256",
    "c5d952b3bc03ea277209aff0ef5b5c4c055d74449ff794c02d8f4e315fdef6b6",
)
BREEZE25_SAMPLE_RATE = getattr(
    quality_runtime_module,
    "BREEZE25_SAMPLE_RATE",
    16_000,
)
BREEZE25_ATTENTION_IMPLEMENTATION = getattr(
    quality_runtime_module,
    "BREEZE25_ATTENTION_IMPLEMENTATION",
    "eager",
)
BREEZE25_RETURN_ATTENTION_MASK = getattr(
    quality_runtime_module,
    "BREEZE25_RETURN_ATTENTION_MASK",
    True,
)
BREEZE25_MAX_MICROBATCH_SEGMENTS = getattr(
    quality_runtime_module,
    "BREEZE25_MAX_MICROBATCH_SEGMENTS",
    6,
)


def _unavailable_breeze25_transcriber(*args: Any, **kwargs: Any) -> str:
    del args, kwargs
    raise RuntimeError(
        "the pinned Breeze ASR 25 runtime is unavailable"
    )


transcribe_breeze25 = getattr(
    quality_runtime_module,
    "transcribe_breeze25",
    _unavailable_breeze25_transcriber,
)


class GlyphHybridAdapterError(RuntimeError):
    """Raised when production data cannot satisfy hardened schema 7."""


@dataclass(frozen=True)
class GlyphAdapterRuntime:
    """Exact hosted dependencies required for an independent rebuild."""

    sample_rate: int
    inference_steps: int
    model_repo_id: str
    model_revision: str
    ecapa_repo_id: str
    ecapa_revision: str
    speaker_anchor_sha256: str
    speaker_centroid: Any
    generation_lock: Any
    generate_chunk: Callable[..., Any]
    generation_policy_resolver: Callable[[int], Any]
    generation_cfg_resolver: Callable[[int], float]
    speaker_encoder_factory: Callable[[], Any]
    speaker_anchor_factory: Callable[[], np.ndarray]
    gate_policy: evidence.GatePolicy
    asr_max_new_tokens: int
    asr_max_verification_segments: int
    renderer_builder: Callable[..., Any] = build_glyph_hybrid_plan
    assembly_verifier: Callable[..., None] = verify_hybrid_assembly
    carrier_conditioner: Callable[[CarrierSegment], bytes] = condition_carrier_v1
    breeze_primary_transcriber: Callable[..., str] = transcribe_breeze25
    breeze_confirmation_transcriber: Callable[..., str] = transcribe_breeze25
    speaker_measure: Callable[..., Any] = release_speaker_evidence_from_audio
    squim_measure: Callable[..., Any] = squim_objective_evidence_from_audio


def _canonical_json_bytes(value: Any) -> bytes:
    try:
        return json.dumps(
            value,
            ensure_ascii=True,
            allow_nan=False,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("ascii")
    except (TypeError, ValueError) as exc:
        raise GlyphHybridAdapterError(
            f"adapter value is not canonical JSON: {exc}"
        ) from exc


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


def _sha256_payload(value: Any) -> str:
    return _sha256_bytes(_canonical_json_bytes(value))


def _pcm_waveform(pcm16_le: bytes) -> np.ndarray:
    if type(pcm16_le) is not bytes or not pcm16_le or len(pcm16_le) % 2:
        raise GlyphHybridAdapterError("exact PCM must be non-empty PCM16 bytes")
    return (
        np.frombuffer(pcm16_le, dtype="<i2").astype(np.float32)
        / np.float32(32768.0)
    )


def _normalized_hf_uri(value: str) -> str:
    result = str(value).strip().rstrip("/")
    for prefix in ("hf://", "https://huggingface.co/"):
        if result.casefold().startswith(prefix):
            result = result[len(prefix) :]
            break
    return result.casefold()


def _select_hf_pin(
    pins: Sequence[Any],
    *,
    repo_id: str,
    revision: str,
    label: str,
) -> Any:
    expected_repo = _normalized_hf_uri(repo_id)
    matches = tuple(
        pin
        for pin in pins
        if _normalized_hf_uri(getattr(pin, "uri", "")) == expected_repo
        and getattr(pin, "revision", None) == revision
    )
    if len(matches) != 1:
        raise GlyphHybridAdapterError(
            f"exact {label} artifact pin is unavailable"
        )
    return matches[0]


def _select_squim_pin(pins: Sequence[Any]) -> Any:
    matches = tuple(
        pin
        for pin in pins
        if getattr(pin, "sha256", None) == SQUIM_OBJECTIVE_WEIGHT_SHA256
        and "squim" in str(getattr(pin, "name", "")).casefold()
    )
    if len(matches) != 1:
        raise GlyphHybridAdapterError(
            "exact pinned SQUIM artifact is unavailable"
        )
    return matches[0]


def _artifact_pin(
    pin: Any,
    *,
    kind: str,
    config: Mapping[str, Any],
    logical_name: str | None = None,
) -> evidence.ArtifactPin:
    return evidence.ArtifactPin(
        name=str(logical_name if logical_name is not None else pin.name),
        kind=kind,
        uri=str(pin.uri),
        revision=str(pin.revision),
        sha256=str(pin.sha256),
        config_sha256=_sha256_payload(dict(config)),
    )


def _source_file_pin(
    name: str,
    path: str | Path,
    *,
    config: Mapping[str, Any],
) -> evidence.ArtifactPin:
    selected = Path(path).resolve(strict=True)
    payload = selected.read_bytes()
    digest = _sha256_bytes(payload)
    return evidence.ArtifactPin(
        name=name,
        kind="source",
        uri=f"repo://{selected.name}",
        revision=digest[:40],
        sha256=digest,
        config_sha256=_sha256_payload(dict(config)),
    )


def _runtime_pins(
    runtime: GlyphAdapterRuntime,
    bundle: Any,
) -> tuple[
    tuple[evidence.ArtifactPin, ...],
    evidence.ArtifactPin,
    evidence.ArtifactPin,
    evidence.ArtifactPin,
    evidence.ArtifactPin,
    evidence.ArtifactPin,
]:
    model_pin = _select_hf_pin(
        tuple(bundle.model_pins),
        repo_id=runtime.model_repo_id,
        revision=runtime.model_revision,
        label="BlueMagpie model",
    )
    breeze_weights_pin = _select_hf_pin(
        tuple(bundle.evaluator_pins),
        repo_id=BREEZE25_MODEL_ID,
        revision=BREEZE25_REVISION,
        label="Breeze ASR 25",
    )
    if getattr(breeze_weights_pin, "sha256", None) != BREEZE25_WEIGHT_SHA256:
        raise GlyphHybridAdapterError(
            "exact Breeze ASR 25 weight pin is unavailable"
        )
    ecapa_pin = _select_hf_pin(
        tuple(bundle.evaluator_pins),
        repo_id=runtime.ecapa_repo_id,
        revision=runtime.ecapa_revision,
        label="ECAPA",
    )
    squim_pin = _select_squim_pin(tuple(bundle.evaluator_pins))

    source_pins = tuple(
        _artifact_pin(
            pin,
            kind="source",
            config={
                "schema": ADAPTER_SCHEMA,
                "role": "bundle-source",
                "artifact_sha256": str(pin.sha256),
            },
        )
        for pin in tuple(bundle.source_pins)
    ) + (
        _source_file_pin(
            "glyph-ascii-runtime-source",
            glyph_ascii_module.__file__,
            config={
                "schema": ADAPTER_SCHEMA,
                "role": "grammar-and-inverse-proof",
            },
        ),
        _source_file_pin(
            "glyph-hybrid-plan-source",
            glyph_plan_module.__file__,
            config={
                "schema": ADAPTER_SCHEMA,
                "role": "renderer",
            },
        ),
        _source_file_pin(
            "glyph-assets-runtime-source",
            glyph_assets_module.__file__,
            config={
                "schema": ADAPTER_SCHEMA,
                "role": "join-and-carrier-conditioner",
            },
        ),
        _source_file_pin(
            "glyph-hybrid-evidence-source",
            evidence.__file__,
            config={
                "schema": ADAPTER_SCHEMA,
                "role": "evidence-rebuild-and-validation",
            },
        ),
        _source_file_pin(
            "glyph-hybrid-adapter-source",
            __file__,
            config={
                "schema": ADAPTER_SCHEMA,
                "role": "production-evidence-adapter",
            },
        ),
        _source_file_pin(
            "glyph-generation-runtime-source",
            runtime.generate_chunk.__code__.co_filename,
            config={
                "schema": ADAPTER_SCHEMA,
                "role": "generation-replay",
            },
        ),
    )
    model = _artifact_pin(
        model_pin,
        kind="model",
        config={
            "schema": ADAPTER_SCHEMA,
            "repo_id": runtime.model_repo_id,
            "revision": runtime.model_revision,
            "sample_rate": runtime.sample_rate,
            "inference_steps": runtime.inference_steps,
        },
    )
    shared_asr_config = {
        "schema": ADAPTER_SCHEMA,
        "task": ASR_TASK,
        "max_new_tokens": runtime.asr_max_new_tokens,
        "max_verification_segments": (
            runtime.asr_max_verification_segments
        ),
        "sample_rate": BREEZE25_SAMPLE_RATE,
        "attention_implementation": BREEZE25_ATTENTION_IMPLEMENTATION,
        "return_attention_mask": BREEZE25_RETURN_ATTENTION_MASK,
        "max_microbatch_segments": BREEZE25_MAX_MICROBATCH_SEGMENTS,
        "do_sample": False,
        "num_beams": 1,
        "repo_id": BREEZE25_MODEL_ID,
        "revision": BREEZE25_REVISION,
        "shared_weights_sha256": str(breeze_weights_pin.sha256),
    }
    breeze_primary = _artifact_pin(
        breeze_weights_pin,
        kind="asr",
        logical_name=BREEZE25_PRIMARY_PIN_NAME,
        config={
            **shared_asr_config,
            "decoder_profile": BREEZE25_PRIMARY_DECODER_PROFILE,
            "language": BREEZE25_PRIMARY_LANGUAGE,
        },
    )
    breeze_confirmation = _artifact_pin(
        breeze_weights_pin,
        kind="asr",
        logical_name=BREEZE25_CONFIRMATION_PIN_NAME,
        config={
            **shared_asr_config,
            "decoder_profile": BREEZE25_CONFIRMATION_DECODER_PROFILE,
            "language": "auto",
        },
    )
    ecapa = _artifact_pin(
        ecapa_pin,
        kind="ecapa",
        config={
            "schema": ADAPTER_SCHEMA,
            "repo_id": runtime.ecapa_repo_id,
            "revision": runtime.ecapa_revision,
            "device": "cpu",
            "measurement": "release-full-plus-thirds-v1",
            "active_top_db": runtime.gate_policy.active_voice_top_db,
            "speaker_anchor_sha256": runtime.speaker_anchor_sha256,
        },
    )
    squim = _artifact_pin(
        squim_pin,
        kind="squim",
        config={
            "schema": ADAPTER_SCHEMA,
            "weight_sha256": SQUIM_OBJECTIVE_WEIGHT_SHA256,
            "sample_rate": SQUIM_OBJECTIVE_SAMPLE_RATE,
            "window_seconds": SQUIM_OBJECTIVE_WINDOW_SECONDS,
            "max_windows": SQUIM_OBJECTIVE_MAX_WINDOWS,
            "device": "cpu",
        },
    )
    return (
        source_pins,
        model,
        breeze_primary,
        breeze_confirmation,
        ecapa,
        squim,
    )


def _trusted_bundle(
    bundle: Any,
    manifest_entries: Mapping[str, str],
) -> evidence.TrustedGlyphBundle:
    assets: list[evidence.TrustedAsset] = []
    for asset in tuple(bundle.assets):
        provisional = evidence.TrustedAsset(
            asset_id=str(asset.asset_id),
            raw_text=str(asset.raw_byte),
            canonical_text=str(asset.canonical_token),
            manifest_entry_sha256="0" * 64,
            pcm16_le=bytes(asset.pcm16_le),
        )
        entry_sha256 = evidence.compute_trusted_asset_entry_sha256(
            provisional
        )
        if manifest_entries.get(provisional.asset_id) != entry_sha256:
            raise GlyphHybridAdapterError(
                "profile manifest entry does not match exact asset bytes"
            )
        assets.append(
            evidence.TrustedAsset(
                **{
                    **provisional.__dict__,
                    "manifest_entry_sha256": entry_sha256,
                }
            )
        )
    manifest_sha256, bundle_sha256 = evidence.compute_trusted_bundle_digests(
        tuple(assets)
    )
    return evidence.TrustedGlyphBundle(
        manifest_sha256=manifest_sha256,
        bundle_sha256=bundle_sha256,
        assets=tuple(assets),
    )


def _profile_sha256(plan: Any, bundle: Any) -> str:
    return _sha256_payload(
        {
            "schema": ADAPTER_SCHEMA,
            "profile_id": str(plan.profile_id),
            "normalization_id": str(plan.normalization_id),
            "boundary_contract_id": str(plan.boundary_contract_id),
            "grammar_id": str(plan.grammar_id),
            "grammar_sha256": str(bundle.grammar_sha256),
        }
    )


def _join_sha256(bundle: Any) -> str:
    return _sha256_payload(dict(bundle.join_profile))


def _trusted_semantic_plan(
    raw_text: str,
    plan: Any,
    bundle: Any,
) -> evidence.TrustedRendererPlan:
    segments = tuple(
        evidence.TrustedRenderSegment(
            segment_id=f"renderer-segment-{index}",
            kind=(
                "asset"
                if segment.kind == "asset_atom"
                else "carrier"
                if segment.kind == "carrier"
                else "control"
            ),
            raw_start=int(segment.absolute_raw_start),
            raw_end=int(segment.absolute_raw_end),
            canonical_start=int(segment.canonical_start),
            canonical_end=int(segment.canonical_end),
            raw_text=str(segment.raw_text),
            canonical_text=str(segment.canonical_text),
            output_start_sample=0,
            output_end_sample=0,
            asset_id=(
                str(segment.asset_id)
                if segment.kind == "asset_atom"
                else None
            ),
        )
        for index, segment in enumerate(tuple(plan.segments))
    )
    return evidence.TrustedRendererPlan(
        grammar_id=str(plan.grammar_id),
        grammar_sha256=str(bundle.grammar_sha256),
        profile_id=str(plan.profile_id),
        profile_sha256=_profile_sha256(plan, bundle),
        join_profile_id=str(bundle.join_profile["id"]),
        join_profile_sha256=_join_sha256(bundle),
        raw_input=raw_text,
        canonical_target=str(plan.canonical_target),
        segments=segments,
    )


def _trusted_realized_plan(
    raw_text: str,
    plan: Any,
    bundle: Any,
    assembly: Any,
) -> evidence.TrustedRendererPlan:
    ledger_spans = tuple(assembly.ledger.spans)
    span_index = 0
    output_cursor = 0
    result: list[evidence.TrustedRenderSegment] = []

    def add_silence(span: Any, raw_cursor: int, canonical_cursor: int) -> None:
        nonlocal output_cursor
        if (
            span.kind not in {"pause", "final_silence"}
            or int(span.start_sample) != output_cursor
            or int(span.end_sample) <= output_cursor
        ):
            raise GlyphHybridAdapterError(
                "assembly join-silence provenance is invalid"
            )
        result.append(
            evidence.TrustedRenderSegment(
                segment_id=f"join-silence-{len(result)}",
                kind="silence",
                raw_start=raw_cursor,
                raw_end=raw_cursor,
                canonical_start=canonical_cursor,
                canonical_end=canonical_cursor,
                raw_text="",
                canonical_text="",
                output_start_sample=output_cursor,
                output_end_sample=int(span.end_sample),
            )
        )
        output_cursor = int(span.end_sample)

    for segment in tuple(plan.segments):
        raw_start = int(segment.absolute_raw_start)
        raw_end = int(segment.absolute_raw_end)
        canonical_start = int(segment.canonical_start)
        canonical_end = int(segment.canonical_end)
        if segment.kind in {"control", "separator"}:
            result.append(
                evidence.TrustedRenderSegment(
                    segment_id=f"plan-control-{segment.ordinal}",
                    kind="control",
                    raw_start=raw_start,
                    raw_end=raw_end,
                    canonical_start=canonical_start,
                    canonical_end=canonical_end,
                    raw_text=str(segment.raw_text),
                    canonical_text=str(segment.canonical_text),
                    output_start_sample=output_cursor,
                    output_end_sample=output_cursor,
                )
            )
            continue
        while (
            span_index < len(ledger_spans)
            and ledger_spans[span_index].kind == "pause"
        ):
            add_silence(
                ledger_spans[span_index],
                raw_start,
                canonical_start,
            )
            span_index += 1
        if span_index >= len(ledger_spans):
            raise GlyphHybridAdapterError(
                "renderer audio segment is absent from assembly ledger"
            )
        span = ledger_spans[span_index]
        expected_kind = (
            "asset" if segment.kind == "asset_atom" else "carrier"
        )
        expected_segment_id = f"glyph-plan-segment-{segment.ordinal}"
        if (
            span.kind != expected_kind
            or span.segment_id != expected_segment_id
            or int(span.start_sample) != output_cursor
            or int(span.end_sample) <= output_cursor
        ):
            raise GlyphHybridAdapterError(
                "renderer and assembly segment provenance disagree"
            )
        result.append(
            evidence.TrustedRenderSegment(
                segment_id=expected_segment_id,
                kind=expected_kind,
                raw_start=raw_start,
                raw_end=raw_end,
                canonical_start=canonical_start,
                canonical_end=canonical_end,
                raw_text=str(segment.raw_text),
                canonical_text=str(segment.canonical_text),
                output_start_sample=output_cursor,
                output_end_sample=int(span.end_sample),
                asset_id=(
                    str(segment.asset_id)
                    if expected_kind == "asset"
                    else None
                ),
            )
        )
        output_cursor = int(span.end_sample)
        span_index += 1

    while span_index < len(ledger_spans):
        add_silence(
            ledger_spans[span_index],
            len(raw_text),
            len(str(plan.canonical_target)),
        )
        span_index += 1
    if output_cursor != len(assembly.pcm16_le) // 2:
        raise GlyphHybridAdapterError(
            "renderer evidence does not cover exact assembled PCM"
        )
    semantic = _trusted_semantic_plan(raw_text, plan, bundle)
    return evidence.TrustedRendererPlan(
        grammar_id=semantic.grammar_id,
        grammar_sha256=semantic.grammar_sha256,
        profile_id=semantic.profile_id,
        profile_sha256=semantic.profile_sha256,
        join_profile_id=semantic.join_profile_id,
        join_profile_sha256=semantic.join_profile_sha256,
        raw_input=semantic.raw_input,
        canonical_target=semantic.canonical_target,
        segments=tuple(result),
    )


def _policy_payload(
    runtime: GlyphAdapterRuntime,
    reservation: Any,
    model_pin: evidence.ArtifactPin,
) -> tuple[str, str, Any, float]:
    candidate_ordinal = int(reservation.candidate_ordinal)
    network_conditioned = bool(reservation.network_conditioned)
    policy = runtime.generation_policy_resolver(candidate_ordinal)
    cfg = float(runtime.generation_cfg_resolver(candidate_ordinal))
    if not math.isfinite(cfg):
        raise GlyphHybridAdapterError("generation replay CFG is invalid")
    if is_dataclass(policy):
        policy_value = asdict(policy)
    elif isinstance(getattr(policy, "__dict__", None), dict):
        policy_value = dict(policy.__dict__)
    else:
        raise GlyphHybridAdapterError(
            "generation policy is not canonically serializable"
        )
    payload = {
        "schema": GENERATION_POLICY_SCHEMA,
        "model_pin_name": model_pin.name,
        "model_sha256": model_pin.sha256,
        "candidate_ordinal": candidate_ordinal,
        "network_conditioned": network_conditioned,
        "scheduled_cfg": cfg,
        "inference_steps": runtime.inference_steps,
        "sample_rate": runtime.sample_rate,
        "policy": policy_value,
    }
    digest = _sha256_payload(payload)
    return f"generation-policy-{digest[:24]}", digest, policy, cfg


def _conditioned_attempt_pcm(
    runtime: GlyphAdapterRuntime,
    completed: Any,
    outcome: Any,
) -> bytes:
    return runtime.carrier_conditioner(
        CarrierSegment(
            segment_id=str(completed.reservation.carrier_segment_id),
            audio=np.asarray(outcome.audio, dtype=np.float32).reshape(-1),
            selected_attempt_id=str(completed.record.attempt_id),
            selected_seed=int(completed.record.seed),
            selected_text_sha256=str(completed.record.text_sha256),
            selected_endpoint_evidence_sha256=str(
                completed.record.endpoint_evidence_sha256
            ),
            sample_rate=runtime.sample_rate,
            speed=1.0,
        )
    )


def _endpoint(
    pcm16_le: bytes,
    outcome: Any,
    policy: evidence.GatePolicy,
) -> evidence.EndpointEvidence:
    return evidence.EndpointEvidence(
        stop_reason=str(outcome.stop_reason),
        tail_energy_ratio=evidence.measure_endpoint_tail_energy_ratio(
            pcm16_le,
            policy,
        ),
        terminal_silence_samples=(
            evidence.measure_terminal_silence_samples(
                pcm16_le,
                policy,
            )
        ),
        generated_steps=int(outcome.generated_steps),
        hard_cap_steps=int(outcome.hard_stop_steps),
    )


def _verification_result(value: Any, *, scope: str) -> Any:
    verification = getattr(value, "verification", value)
    if getattr(verification, "passed", None) is not True:
        raise GlyphHybridAdapterError(
            f"{scope} production verification did not pass"
        )
    rows = tuple(getattr(verification, "candidate_results", ()))
    if len(rows) != 1:
        raise GlyphHybridAdapterError(
            f"{scope} production verification is not whole-output"
        )
    return rows[0]


class HardenedGlyphHybridAdapter:
    """Callable adapter installed into ``app._GLYPH_HYBRID_EVIDENCE_ADAPTER``."""

    def __init__(self, runtime: GlyphAdapterRuntime) -> None:
        if type(runtime) is not GlyphAdapterRuntime:
            raise TypeError("runtime must be GlyphAdapterRuntime")
        self._runtime = runtime

    def __call__(
        self,
        *,
        raw_text: str,
        plan: Any,
        profile_state: Any,
        segments: tuple[AssetSegment | CarrierSegment, ...],
        assembly: Any,
        semantic_commitment: Any,
        request_ledger: Any,
        final_verification: Any,
        independent_final_verification: Any,
    ) -> bytes:
        runtime = self._runtime
        if (
            type(raw_text) is not str
            or not raw_text
            or runtime.sample_rate != int(assembly.sample_rate)
            or runtime.gate_policy.sample_rate != runtime.sample_rate
            or runtime.asr_max_new_tokens <= 0
            or runtime.asr_max_verification_segments <= 0
        ):
            raise GlyphHybridAdapterError(
                "adapter request or runtime contract is invalid"
            )
        if profile_state.bundle is None:
            raise GlyphHybridAdapterError("trusted glyph bundle is unavailable")
        bundle = profile_state.bundle
        manifest_entries = dict(
            profile_state.asset_entry_sha256_by_asset_id
        )
        runtime.assembly_verifier(
            assembly,
            trusted_bundle=bundle,
            original_segments=segments,
            expected_semantic_commitment=semantic_commitment,
        )
        final_row = _verification_result(
            final_verification,
            scope="final",
        )
        _verification_result(
            independent_final_verification,
            scope="independent",
        )

        trusted_bundle = _trusted_bundle(bundle, manifest_entries)
        trusted_plan = _trusted_realized_plan(
            raw_text,
            plan,
            bundle,
            assembly,
        )
        (
            source_pins,
            model_pin,
            breeze_primary_pin,
            breeze_confirmation_pin,
            ecapa_pin,
            squim_pin,
        ) = _runtime_pins(runtime, bundle)

        anchor = np.asarray(
            runtime.speaker_anchor_factory(),
            dtype="<f4",
        ).reshape(-1)
        if (
            anchor.size == 0
            or not np.isfinite(anchor).all()
            or _sha256_bytes(anchor.tobytes(order="C"))
            != runtime.speaker_anchor_sha256
            or runtime.speaker_anchor_sha256
            != str(bundle.speaker_anchor_sha256)
        ):
            raise GlyphHybridAdapterError(
                "speaker anchor does not match the pinned runtime"
            )

        completed = tuple(getattr(request_ledger, "completed_attempts", ()))
        records = tuple(getattr(request_ledger, "records", ()))
        if (
            len(completed) != len(records)
            or any(
                item.record != record
                for item, record in zip(completed, records, strict=True)
            )
        ):
            raise GlyphHybridAdapterError(
                "generation attempt ledger is incomplete"
            )

        policies: dict[str, str] = {}
        replay_by_key: dict[
            tuple[str, str, int, str, str, int],
            list[tuple[Any, Any, float]],
        ] = {}
        model_calls: list[evidence.TrustedModelGenerationCall] = []
        attempts_by_segment: dict[str, list[str]] = {}
        attempt_ordinal_by_segment: dict[str, int] = {}
        for item in completed:
            reservation = item.reservation
            segment_id = str(reservation.carrier_segment_id)
            attempt_ordinal = attempt_ordinal_by_segment.get(segment_id, 0)
            attempt_ordinal_by_segment[segment_id] = attempt_ordinal + 1
            policy_id, policy_sha256, policy, cfg = _policy_payload(
                runtime,
                reservation,
                model_pin,
            )
            previous_policy = policies.setdefault(policy_id, policy_sha256)
            if previous_policy != policy_sha256:
                raise GlyphHybridAdapterError(
                    "generation policy id is ambiguous"
                )
            logged_pcm = _conditioned_attempt_pcm(
                runtime,
                item,
                item.endpoint_outcome,
            )
            logged_endpoint = _endpoint(
                logged_pcm,
                item.endpoint_outcome,
                runtime.gate_policy,
            )
            call_id = str(item.record.attempt_id)
            model_calls.append(
                evidence.TrustedModelGenerationCall(
                    call_id=call_id,
                    segment_id=segment_id,
                    attempt_ordinal=attempt_ordinal,
                    text=str(reservation.text),
                    seed=int(reservation.seed),
                    policy_id=policy_id,
                    policy_sha256=policy_sha256,
                    model_pin_name=model_pin.name,
                    output_pcm16_le=logged_pcm,
                    endpoint=logged_endpoint,
                )
            )
            attempts_by_segment.setdefault(segment_id, []).append(call_id)
            key = (
                model_pin.name,
                str(reservation.text),
                int(reservation.seed),
                policy_id,
                policy_sha256,
                runtime.sample_rate,
            )
            replay_by_key.setdefault(key, []).append((item, policy, cfg))

        replay_cache: dict[
            tuple[str, str, int, str, str, int],
            tuple[bytes, evidence.EndpointEvidence],
        ] = {}

        def generation_evaluator(
            pin: evidence.ArtifactPin,
            text: str,
            seed: int,
            policy_id: str,
            policy_sha256: str,
            sample_rate: int,
        ) -> tuple[bytes, evidence.EndpointEvidence]:
            key = (
                pin.name,
                text,
                seed,
                policy_id,
                policy_sha256,
                sample_rate,
            )
            cached = replay_cache.get(key)
            if cached is not None:
                return cached
            replay_items = replay_by_key.get(key)
            if replay_items is None or pin != model_pin:
                raise GlyphHybridAdapterError(
                    "model replay request is outside the trusted ledger"
                )
            replayed: list[tuple[bytes, evidence.EndpointEvidence]] = []
            for item, policy, cfg in replay_items:
                reservation = item.reservation
                try:
                    with runtime.generation_lock:
                        outcome = runtime.generate_chunk(
                            str(reservation.text),
                            runtime.speaker_centroid,
                            cfg=cfg,
                            steps=runtime.inference_steps,
                            request_seed=int(reservation.seed),
                            policy=policy,
                            network_conditioned=bool(
                                reservation.network_conditioned
                            ),
                        )
                except Exception as exc:
                    raise GlyphHybridAdapterError(
                        "deterministic model replay failed"
                    ) from exc
                replayed_pcm = _conditioned_attempt_pcm(
                    runtime,
                    item,
                    outcome,
                )
                replayed.append(
                    (
                        replayed_pcm,
                        _endpoint(
                            replayed_pcm,
                            outcome,
                            runtime.gate_policy,
                        ),
                    )
                )
            result = replayed[0]
            if any(item != result for item in replayed[1:]):
                raise GlyphHybridAdapterError(
                    "identical generation inputs replayed inconsistently"
                )
            replay_cache[key] = result
            return result

        asset_by_id = {
            asset.asset_id: asset for asset in trusted_bundle.assets
        }
        asset_calls: list[evidence.TrustedAssetLoadCall] = []
        for index, segment in enumerate(segments):
            if not isinstance(segment, AssetSegment):
                continue
            asset = asset_by_id.get(segment.asset_id)
            if asset is None:
                raise GlyphHybridAdapterError(
                    "assembled asset is absent from trusted bundle"
                )
            asset_calls.append(
                evidence.TrustedAssetLoadCall(
                    call_id=f"asset-load-{index}",
                    segment_id=str(segment.segment_id),
                    asset_id=str(segment.asset_id),
                    manifest_entry_sha256=asset.manifest_entry_sha256,
                    loaded_pcm16_le=asset.pcm16_le,
                )
            )

        selections: list[evidence.TrustedCarrierSelection] = []
        carrier_rank = 0
        for segment in segments:
            if not isinstance(segment, CarrierSegment):
                continue
            carrier_rank += 1
            attempt_ids = tuple(
                attempts_by_segment.get(str(segment.segment_id), ())
            )
            if segment.selected_attempt_id not in attempt_ids:
                raise GlyphHybridAdapterError(
                    "selected carrier has no replayable attempt"
                )
            selections.append(
                evidence.TrustedCarrierSelection(
                    segment_id=str(segment.segment_id),
                    path_rank=carrier_rank,
                    attempted_call_ids=attempt_ids,
                    selected_call_id=str(segment.selected_attempt_id),
                )
            )
        if bool(completed) != bool(selections):
            raise GlyphHybridAdapterError(
                "pure-asset/model-work accounting is inconsistent"
            )

        expected_pins = {
            breeze_primary_pin.name: breeze_primary_pin,
            breeze_confirmation_pin.name: breeze_confirmation_pin,
            ecapa_pin.name: ecapa_pin,
            squim_pin.name: squim_pin,
        }
        asr_cache: dict[tuple[str, str, int], str] = {}
        ecapa_cache: dict[
            tuple[str, str, int, str],
            tuple[float, float, float],
        ] = {}
        squim_cache: dict[
            tuple[str, str, int],
            tuple[float, float, float],
        ] = {}

        def asr_evaluator(
            pin: evidence.ArtifactPin,
            pcm16_le: bytes,
            sample_rate: int,
        ) -> str:
            if expected_pins.get(pin.name) != pin:
                raise GlyphHybridAdapterError("ASR pin changed during rebuild")
            key = pin.name, _sha256_bytes(pcm16_le), sample_rate
            if key not in asr_cache:
                waveform = _pcm_waveform(pcm16_le)
                if pin == breeze_primary_pin:
                    transcriber = runtime.breeze_primary_transcriber
                    language = BREEZE25_PRIMARY_LANGUAGE
                elif pin == breeze_confirmation_pin:
                    transcriber = runtime.breeze_confirmation_transcriber
                    language = BREEZE25_CONFIRMATION_LANGUAGE
                else:
                    raise GlyphHybridAdapterError(
                        "ASR evaluator kind is not pinned"
                    )
                transcript = transcriber(
                    waveform,
                    sample_rate,
                    language=language,
                    task=ASR_TASK,
                    max_new_tokens=runtime.asr_max_new_tokens,
                    max_verification_segments=(
                        runtime.asr_max_verification_segments
                    ),
                )
                if type(transcript) is not str:
                    raise GlyphHybridAdapterError(
                        "ASR evaluator returned a non-string transcript"
                    )
                asr_cache[key] = transcript.strip()
            return asr_cache[key]

        def ecapa_evaluator(
            pin: evidence.ArtifactPin,
            pcm16_le: bytes,
            sample_rate: int,
            speaker_anchor_sha256: str,
        ) -> tuple[float, float, float]:
            if (
                pin != ecapa_pin
                or speaker_anchor_sha256
                != runtime.speaker_anchor_sha256
            ):
                raise GlyphHybridAdapterError(
                    "ECAPA pin or speaker anchor changed"
                )
            key = (
                pin.name,
                _sha256_bytes(pcm16_le),
                sample_rate,
                speaker_anchor_sha256,
            )
            if key not in ecapa_cache:
                measured = runtime.speaker_measure(
                    _pcm_waveform(pcm16_le),
                    sample_rate,
                    runtime.speaker_encoder_factory(),
                    anchor,
                    device="cpu",
                    active_top_db=(
                        runtime.gate_policy.active_voice_top_db
                    ),
                )
                values = (
                    float(measured.similarity),
                    float(measured.begin_similarity),
                    float(measured.end_similarity),
                )
                if not all(math.isfinite(value) for value in values):
                    raise GlyphHybridAdapterError(
                        "ECAPA returned non-finite evidence"
                    )
                ecapa_cache[key] = values
            return ecapa_cache[key]

        def squim_evaluator(
            pin: evidence.ArtifactPin,
            pcm16_le: bytes,
            sample_rate: int,
        ) -> tuple[float, float, float]:
            if pin != squim_pin:
                raise GlyphHybridAdapterError("SQUIM pin changed")
            key = pin.name, _sha256_bytes(pcm16_le), sample_rate
            if key not in squim_cache:
                measured = runtime.squim_measure(
                    _pcm_waveform(pcm16_le),
                    sample_rate,
                )
                values = (
                    float(measured.stoi),
                    float(measured.pesq),
                    float(measured.si_sdr),
                )
                if not all(math.isfinite(value) for value in values):
                    raise GlyphHybridAdapterError(
                        "SQUIM returned non-finite evidence"
                    )
                squim_cache[key] = values
            return squim_cache[key]

        output_pcm = bytes(assembly.pcm16_le)
        breeze_primary_transcript = asr_evaluator(
            breeze_primary_pin,
            output_pcm,
            runtime.sample_rate,
        )
        breeze_confirmation_transcript = asr_evaluator(
            breeze_confirmation_pin,
            output_pcm,
            runtime.sample_rate,
        )
        speaker_values = ecapa_evaluator(
            ecapa_pin,
            output_pcm,
            runtime.sample_rate,
            runtime.speaker_anchor_sha256,
        )
        squim_values = squim_evaluator(
            squim_pin,
            output_pcm,
            runtime.sample_rate,
        )
        final_comparison = getattr(final_row, "comparison", None)
        if (
            final_comparison is None
            or not breeze_primary_transcript
            or not breeze_confirmation_transcript
        ):
            raise GlyphHybridAdapterError(
                "semantic evaluator output is unavailable"
            )

        calls: list[evidence.TrustedRuntimeCall] = [
            *model_calls,
            *asset_calls,
        ]
        gate_bindings: list[evidence.TrustedGateBinding] = []
        active_samples = evidence.measure_active_speech_samples(
            output_pcm,
            runtime.gate_policy,
        )
        for scope, asr_pin, transcript in (
            (
                "final",
                breeze_primary_pin,
                breeze_primary_transcript,
            ),
            (
                "independent",
                breeze_confirmation_pin,
                breeze_confirmation_transcript,
            ),
        ):
            asr_call_id = f"asr-{scope}"
            ecapa_call_id = f"ecapa-{scope}"
            squim_call_id = f"squim-{scope}"
            calls.extend(
                (
                    evidence.TrustedAsrCall(
                        call_id=asr_call_id,
                        scope=scope,
                        evaluator_pin_name=asr_pin.name,
                        input_pcm16_le=output_pcm,
                        target_text=str(plan.canonical_target),
                        transcript_text=transcript,
                    ),
                    evidence.TrustedEcapaCall(
                        call_id=ecapa_call_id,
                        scope=scope,
                        evaluator_pin_name=ecapa_pin.name,
                        input_pcm16_le=output_pcm,
                        speaker_anchor_sha256=(
                            runtime.speaker_anchor_sha256
                        ),
                        centroid_similarity=speaker_values[0],
                        begin_similarity=speaker_values[1],
                        end_similarity=speaker_values[2],
                    ),
                    evidence.TrustedSquimCall(
                        call_id=squim_call_id,
                        scope=scope,
                        evaluator_pin_name=squim_pin.name,
                        input_pcm16_le=output_pcm,
                        stoi=squim_values[0],
                        pesq=squim_values[1],
                        si_sdr=squim_values[2],
                    ),
                )
            )
            gate_bindings.append(
                evidence.TrustedGateBinding(
                    scope=scope,
                    asr_call_id=asr_call_id,
                    ecapa_call_id=ecapa_call_id,
                    squim_call_id=squim_call_id,
                    active_speech_samples=active_samples,
                )
            )

        def renderer_evaluator(
            contract: evidence.RendererContract,
            rebuilt_bundle: evidence.TrustedGlyphBundle,
            request_text: str,
        ) -> evidence.TrustedRendererPlan:
            if (
                contract.manifest_sha256
                != rebuilt_bundle.manifest_sha256
                or contract.bundle_sha256 != rebuilt_bundle.bundle_sha256
                or rebuilt_bundle != trusted_bundle
            ):
                raise GlyphHybridAdapterError(
                    "renderer rebuild bundle changed"
                )
            fresh = runtime.renderer_builder(
                request_text,
                asset_entry_sha256_by_asset_id={
                    asset.asset_id: asset.manifest_entry_sha256
                    for asset in rebuilt_bundle.assets
                },
            )
            if fresh is None:
                raise GlyphHybridAdapterError(
                    "pinned renderer did not reproduce the request"
                )
            return _trusted_semantic_plan(
                request_text,
                fresh,
                bundle,
            )

        generation_policy_items = tuple(sorted(policies.items()))
        gate_policy_sha256 = _sha256_payload(
            asdict(runtime.gate_policy)
        )
        contract = evidence.RendererContract(
            grammar_id=trusted_plan.grammar_id,
            grammar_sha256=trusted_plan.grammar_sha256,
            profile_id=trusted_plan.profile_id,
            profile_sha256=trusted_plan.profile_sha256,
            manifest_sha256=trusted_bundle.manifest_sha256,
            bundle_sha256=trusted_bundle.bundle_sha256,
            join_profile_id=trusted_plan.join_profile_id,
            join_profile_sha256=trusted_plan.join_profile_sha256,
            gate_policy_sha256=gate_policy_sha256,
            speaker_anchor_sha256=runtime.speaker_anchor_sha256,
            generation_policy_ids=tuple(
                item[0] for item in generation_policy_items
            ),
            generation_policy_sha256s=tuple(
                item[1] for item in generation_policy_items
            ),
            source_pins=source_pins,
            model_pins=(model_pin,),
            evaluator_pins=(
                breeze_primary_pin,
                breeze_confirmation_pin,
                ecapa_pin,
                squim_pin,
            ),
        )
        rebuild_inputs = evidence.EvidenceRebuildInputs(
            contract=contract,
            plan=trusted_plan,
            bundle=trusted_bundle,
            assembled_pcm16_le=output_pcm,
            public_pcm16_le=bytes(assembly.public_pcm16_le),
            calls=tuple(calls),
            carrier_selections=tuple(selections),
            gate_bindings=tuple(gate_bindings),
            content_commitment_key=secrets.token_bytes(32),
            generation_evaluator=generation_evaluator,
            renderer_evaluator=renderer_evaluator,
            asr_evaluator=asr_evaluator,
            ecapa_evaluator=ecapa_evaluator,
            squim_evaluator=squim_evaluator,
            budget_caps=evidence.BudgetCaps(
                model_invocations=32,
                generated_chunks=32,
                generated_speech_units=800,
            ),
            gate_policy=runtime.gate_policy,
        )
        serialized = evidence.format_hybrid_evidence(rebuild_inputs)
        evidence.validate_hybrid_evidence(serialized, rebuild_inputs)
        return serialized


__all__ = (
    "ADAPTER_SCHEMA",
    "GlyphHybridAdapterError",
    "GlyphAdapterRuntime",
    "HardenedGlyphHybridAdapter",
)