glyph_assets.py

"""Immutable glyph assets and deterministic hybrid PCM16 assembly.

This module is intentionally isolated from ``app`` and ``quality_runtime``.
It is the small, auditable runtime boundary between a frozen, externally
evaluated glyph bundle and a later inference integration.

The v1 contract has three important properties:

* a manifest is accepted only when its externally pinned byte hash, internal
  bundle hash, exact grammar, exact 32-entry ordering, WAV containers, decoded
  PCM and immutable asset cores all agree;
* ECAPA, SQUIM and ASR numbers in the manifest are provenance attestations,
  not runtime measurements.  Loading a bundle never turns those claims into a
  quality pass; the release evaluator must run the pinned models again;
* hybrid output is PCM16 from the first join onward.  Carrier segments are
  conditioned once before insertion, complete asset PCM remains byte-exact,
  and no global speed or gain transform exists after insertion.
"""

from __future__ import annotations

import hashlib
import io
import json
import math
import os
import stat
import wave
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from types import MappingProxyType
from typing import Any, Iterable, Mapping, Sequence

import numpy as np


SHA256_HEX_LENGTH = 64
SAMPLE_RATE = 48_000
PCM_SAMPLE_FORMAT = "pcm_s16le"
MANIFEST_SCHEMA = "bluemagpie.glyph-assets.manifest.v1"
JOIN_PROFILE_ID = "bluemagpie.glyph-hybrid-join.v1"
PLAN_SCHEMA = "bluemagpie.glyph-assets.plan.v1"
LEDGER_SCHEMA = "bluemagpie.glyph-assets.ledger.v1"
SEMANTIC_COMMITMENT_SCHEMA = "bluemagpie.semantic-plan-commitment.v1"
ATTEMPT_LEDGER_SCHEMA = "bluemagpie.generation-attempt-ledger.v1"

ASSET_COUNT = 32
MAX_MANIFEST_BYTES = 1_048_576
MAX_ASSET_CONTAINER_BYTES = 2_000_000
MAX_ASSET_SAMPLES = 480_000
MAX_CARRIER_SAMPLES = 5_760_000
MAX_OUTPUT_SAMPLES = 14_400_000

CARRIER_BOUNDARY_PAUSE_SAMPLES = 1_920  # 40 ms at 48 kHz.
CARRIER_EDGE_FADE_SAMPLES = 144  # 3 ms at 48 kHz.
TAIL_FADE_SAMPLES = 144  # 3 ms at 48 kHz.
FINAL_SILENCE_SAMPLES = 8_640  # 180 ms at 48 kHz.
ASSET_LEADING_FADE_SAMPLES = 144  # Frozen into each asset before hashing.

CARRIER_RMS_FLOOR = 0.07
CARRIER_PEAK_LIMIT = 0.95
CARRIER_MAX_GAIN = 3.0
MAX_BOUNDARY_ABSOLUTE_JUMP = 0.10
MAX_GENERATED_CHUNK_BUDGET = 32
MAX_GENERATED_TEXT_UNIT_BUDGET = 800
MAX_ATTEMPT_RECORDS = 4_096

MIN_ECAPA_SIMILARITY = 0.25
MIN_SQUIM_STOI = 0.72
MIN_SQUIM_PESQ = 1.20
MIN_SQUIM_SI_SDR = -20.0
MAX_ENDPOINT_TAIL_ENERGY_RATIO = 0.02

_HEX_CHARS = frozenset("0123456789abcdef")
_INTERNAL_SEGMENT_PREFIX = "__bluemagpie_internal__:"
_LEGACY_RESERVED_SEGMENT_IDS = frozenset(("final-silence",))


class GlyphAssetError(ValueError):
    """Raised when immutable asset or assembly evidence is invalid."""


def _ordinal_token(index: int) -> str:
    digits = (
        "零",
        "一",
        "二",
        "三",
        "四",
        "五",
        "六",
        "七",
        "八",
        "九",
    )
    if index < 10:
        return digits[index]
    if index == 10:
        return "十"
    if index < 20:
        return "十" + digits[index - 10]
    if index == 20:
        return "二十"
    return "二十" + digits[index - 20]


@dataclass(frozen=True)
class GlyphSpec:
    asset_id: str
    raw_byte: str
    canonical_token: str


V1_GLYPH_SPECS = tuple(
    GlyphSpec(
        asset_id=f"letter-{letter}",
        raw_byte=letter,
        canonical_token=f"第{_ordinal_token(index)}個英文字母",
    )
    for index, letter in enumerate("abcdefghijklmnopqrstuvwxyz", start=1)
) + (
    GlyphSpec("symbol-at", "@", "網路符號小老鼠"),
    GlyphSpec("symbol-dot", ".", "網路符號點"),
    GlyphSpec("symbol-hyphen", "-", "這個符號叫做橫線"),
    GlyphSpec("symbol-underscore", "_", "這個符號叫做底線"),
    GlyphSpec("symbol-colon", ":", "這個符號叫做冒號"),
    GlyphSpec("symbol-slash", "/", "網路符號斜線"),
)


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


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


V1_GRAMMAR_SHA256 = _sha256_bytes(
    _canonical_json_bytes(
        {
            "schema": "bluemagpie.ordinal-glyph-grammar.v1",
            "glyphs": [
                {
                    "id": spec.asset_id,
                    "raw_byte": spec.raw_byte,
                    "canonical_token": spec.canonical_token,
                }
                for spec in V1_GLYPH_SPECS
            ],
        }
    )
)


V1_JOIN_PROFILE = MappingProxyType(
    {
        "id": JOIN_PROFILE_ID,
        "sample_rate": SAMPLE_RATE,
        "speed": 1.0,
        "asset_join_mode": "asset_pcm_concat_v1",
        "asset_transform": "none",
        "asset_speed": 1.0,
        "asset_gain": 1.0,
        "carrier_rms_floor": CARRIER_RMS_FLOOR,
        "carrier_peak_limit": CARRIER_PEAK_LIMIT,
        "carrier_max_gain": CARRIER_MAX_GAIN,
        "carrier_boundary_pause_samples": CARRIER_BOUNDARY_PAUSE_SAMPLES,
        "carrier_edge_fade_samples": CARRIER_EDGE_FADE_SAMPLES,
        "asset_leading_fade_samples": ASSET_LEADING_FADE_SAMPLES,
        "max_boundary_absolute_jump": MAX_BOUNDARY_ABSOLUTE_JUMP,
        "tail_fade_samples": TAIL_FADE_SAMPLES,
        "final_silence_samples": FINAL_SILENCE_SAMPLES,
        "max_generated_chunk_budget": MAX_GENERATED_CHUNK_BUDGET,
        "max_generated_text_unit_budget": MAX_GENERATED_TEXT_UNIT_BUDGET,
        "max_asset_samples": MAX_ASSET_SAMPLES,
        "max_carrier_samples": MAX_CARRIER_SAMPLES,
        "max_output_samples": MAX_OUTPUT_SAMPLES,
    }
)


def _require_exact_keys(
    value: Any,
    expected: Iterable[str],
    *,
    context: str,
) -> Mapping[str, Any]:
    if not isinstance(value, dict):
        raise GlyphAssetError(f"{context} must be an object")
    expected_set = frozenset(expected)
    actual_set = frozenset(value)
    if actual_set != expected_set:
        missing = sorted(expected_set - actual_set)
        extra = sorted(actual_set - expected_set)
        raise GlyphAssetError(
            f"{context} keys do not match schema; missing={missing}, extra={extra}"
        )
    return value


def _require_string(value: Any, *, context: str) -> str:
    if not isinstance(value, str) or not value or "\x00" in value:
        raise GlyphAssetError(f"{context} must be a non-empty string")
    return value


def _require_sha256(value: Any, *, context: str) -> str:
    text = _require_string(value, context=context)
    if (
        len(text) != SHA256_HEX_LENGTH
        or text.lower() != text
        or any(char not in _HEX_CHARS for char in text)
    ):
        raise GlyphAssetError(f"{context} must be a lowercase SHA-256 hex digest")
    return text


def _require_int(
    value: Any,
    *,
    context: str,
    minimum: int | None = None,
    maximum: int | None = None,
) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        raise GlyphAssetError(f"{context} must be an integer")
    if minimum is not None and value < minimum:
        raise GlyphAssetError(f"{context} must be >= {minimum}")
    if maximum is not None and value > maximum:
        raise GlyphAssetError(f"{context} must be <= {maximum}")
    return value


def _require_finite(
    value: Any,
    *,
    context: str,
    minimum: float | None = None,
    maximum: float | None = None,
) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise GlyphAssetError(f"{context} must be numeric")
    result = float(value)
    if not math.isfinite(result):
        raise GlyphAssetError(f"{context} must be finite")
    if minimum is not None and result < minimum:
        raise GlyphAssetError(f"{context} must be >= {minimum}")
    if maximum is not None and result > maximum:
        raise GlyphAssetError(f"{context} must be <= {maximum}")
    return result


def _immutable_json(value: Any) -> Any:
    if isinstance(value, dict):
        return MappingProxyType(
            {str(key): _immutable_json(item) for key, item in value.items()}
        )
    if isinstance(value, list):
        return tuple(_immutable_json(item) for item in value)
    return value


def _reject_duplicate_json_keys(
    pairs: list[tuple[str, Any]],
) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise GlyphAssetError(f"asset manifest contains duplicate JSON key: {key!r}")
        result[key] = value
    return result


@dataclass(frozen=True)
class ArtifactPin:
    name: str
    uri: str
    revision: str
    sha256: str


@dataclass(frozen=True)
class EcapaAttestation:
    cosine_similarity: float
    minimum_cosine_similarity: float


@dataclass(frozen=True)
class SquimAttestation:
    stoi: float
    pesq: float
    si_sdr: float
    minimum_stoi: float
    minimum_pesq: float
    minimum_si_sdr: float


@dataclass(frozen=True)
class EndpointAttestation:
    stop_reason: str
    turbo_cer: float
    large_v3_cer: float
    prefix_cer: float
    suffix_cer: float
    extra_tail_units: int
    tail_energy_ratio: float
    maximum_tail_energy_ratio: float
    terminal_silence_samples: int


@dataclass(frozen=True)
class PcmMeasurements:
    rms: float
    peak: float
    active_duration_seconds: float
    terminal_silence_samples: int


@dataclass(frozen=True)
class GenerationAttemptRecord:
    """Content-free record for one ordered model-generator invocation."""

    ordinal: int
    attempt_id: str
    seed: int
    text_sha256: str
    endpoint_evidence_sha256: str
    generated_chunk_count: int
    generated_text_units: int


@dataclass(frozen=True)
class SemanticPlanCommitment:
    """Opaque external semantic proof and its frozen generation accounting.

    This module does not decide whether a text plan or inverse proof is
    semantically correct.  It does require the caller's externally computed
    hashes, complete ordered attempt ledger and honest 32/800 request
    accounting to survive unchanged in both the audio plan and ledger.
    """

    external_semantic_plan_sha256: str
    inverse_proof_sha256: str
    ordered_attempt_ledger_sha256: str
    attempt_record_count: int
    assembled_carrier_count: int
    request_model_generator_invocations: int
    request_generated_chunk_count: int
    request_generated_text_units: int
    request_generated_chunk_budget: int
    request_generated_text_unit_budget: int
    ordered_attempt_records: tuple[GenerationAttemptRecord, ...]


@dataclass(frozen=True)
class GlyphAsset:
    asset_id: str
    raw_byte: str
    canonical_token: str
    relative_path: str
    container_sha256: str
    pcm_sha256: str
    sample_rate: int
    channels: int
    sample_format: str
    sample_count: int
    rms: float
    peak: float
    active_duration_seconds: float
    core_start_sample: int
    core_end_sample: int
    core_pcm_sha256: str
    ecapa: EcapaAttestation
    squim: SquimAttestation
    endpoint: EndpointAttestation
    container_bytes: bytes
    pcm16_le: bytes

    def decoded_pcm16(self) -> np.ndarray:
        result = np.frombuffer(self.pcm16_le, dtype="<i2").copy()
        result.flags.writeable = False
        return result

    @property
    def raw_byte_value(self) -> int:
        return self.raw_byte.encode("ascii")[0]

    @property
    def runtime_quality_verified(self) -> bool:
        """Manifest attestations are never a runtime quality measurement."""

        return False


@dataclass(frozen=True)
class GlyphAssetBundle:
    manifest_path: Path
    manifest_bytes: bytes
    manifest_sha256: str
    manifest_schema: str
    bundle_version: str
    grammar_sha256: str
    bundle_sha256: str
    speaker_anchor_sha256: str
    join_profile: Mapping[str, Any]
    source_pins: tuple[ArtifactPin, ...]
    model_pins: tuple[ArtifactPin, ...]
    evaluator_pins: tuple[ArtifactPin, ...]
    assets: tuple[GlyphAsset, ...]

    @property
    def runtime_quality_verified(self) -> bool:
        """A successful integrity load is not an ECAPA/SQUIM/ASR release pass."""

        return False

    def asset_by_id(self, asset_id: str) -> GlyphAsset:
        matches = tuple(asset for asset in self.assets if asset.asset_id == asset_id)
        if len(matches) != 1:
            raise GlyphAssetError(f"asset id is not uniquely available: {asset_id!r}")
        return matches[0]

    def asset_by_raw_byte(self, raw_byte: str | bytes | int) -> GlyphAsset:
        if isinstance(raw_byte, bytes):
            if len(raw_byte) != 1:
                raise GlyphAssetError("raw byte lookup requires exactly one byte")
            value = raw_byte.decode("ascii")
        elif isinstance(raw_byte, int):
            if raw_byte < 0 or raw_byte > 127:
                raise GlyphAssetError("raw byte integer must be ASCII")
            value = bytes((raw_byte,)).decode("ascii")
        elif isinstance(raw_byte, str):
            if len(raw_byte.encode("ascii", errors="strict")) != 1:
                raise GlyphAssetError("raw byte lookup requires one ASCII byte")
            value = raw_byte
        else:
            raise GlyphAssetError("unsupported raw byte lookup type")
        matches = tuple(asset for asset in self.assets if asset.raw_byte == value)
        if len(matches) != 1:
            raise GlyphAssetError(f"raw byte is not uniquely available: {value!r}")
        return matches[0]


def _attempt_record_projection(
    value: GenerationAttemptRecord,
    *,
    expected_ordinal: int,
) -> Mapping[str, Any]:
    context = f"semantic_commitment.ordered_attempt_records[{expected_ordinal}]"
    if not isinstance(value, GenerationAttemptRecord):
        raise GlyphAssetError(f"{context} must be a GenerationAttemptRecord")
    ordinal = _require_int(
        value.ordinal,
        context=f"{context}.ordinal",
        minimum=0,
        maximum=MAX_ATTEMPT_RECORDS - 1,
    )
    if ordinal != expected_ordinal:
        raise GlyphAssetError("ordered attempt ledger ordinals are not contiguous")
    attempt_id = _require_string(value.attempt_id, context=f"{context}.attempt_id")
    seed = _require_int(
        value.seed,
        context=f"{context}.seed",
        minimum=0,
        maximum=(2**31) - 1,
    )
    text_sha256 = _require_sha256(
        value.text_sha256,
        context=f"{context}.text_sha256",
    )
    endpoint_evidence_sha256 = _require_sha256(
        value.endpoint_evidence_sha256,
        context=f"{context}.endpoint_evidence_sha256",
    )
    generated_chunk_count = _require_int(
        value.generated_chunk_count,
        context=f"{context}.generated_chunk_count",
        minimum=0,
        maximum=1,
    )
    generated_text_units = _require_int(
        value.generated_text_units,
        context=f"{context}.generated_text_units",
        minimum=0,
        maximum=MAX_GENERATED_TEXT_UNIT_BUDGET,
    )
    if generated_chunk_count == 0 and generated_text_units != 0:
        raise GlyphAssetError(
            "attempt without a generated chunk must account zero text units"
        )
    if generated_chunk_count == 1 and generated_text_units == 0:
        raise GlyphAssetError(
            "generated attempt must account at least one text unit"
        )
    return {
        "ordinal": ordinal,
        "attempt_id": attempt_id,
        "seed": seed,
        "text_sha256": text_sha256,
        "endpoint_evidence_sha256": endpoint_evidence_sha256,
        "generated_chunk_count": generated_chunk_count,
        "generated_text_units": generated_text_units,
    }


def _ordered_attempt_ledger_projection(
    records: Sequence[GenerationAttemptRecord],
) -> Mapping[str, Any]:
    if not isinstance(records, Sequence):
        raise GlyphAssetError("ordered attempt ledger must be a sequence")
    if not 0 <= len(records) <= MAX_ATTEMPT_RECORDS:
        raise GlyphAssetError(
            "ordered attempt ledger exceeds the bounded record limit"
        )
    projected = [
        _attempt_record_projection(record, expected_ordinal=index)
        for index, record in enumerate(records)
    ]
    attempt_ids = [record["attempt_id"] for record in projected]
    if len(set(attempt_ids)) != len(attempt_ids):
        raise GlyphAssetError("ordered attempt ledger contains duplicate attempt id")
    return {
        "schema": ATTEMPT_LEDGER_SCHEMA,
        "records": projected,
    }


def compute_ordered_attempt_ledger_sha256(
    records: Sequence[GenerationAttemptRecord],
) -> str:
    """Hash a structurally valid, content-free ordered request attempt ledger."""

    return _sha256_bytes(
        _canonical_json_bytes(_ordered_attempt_ledger_projection(records))
    )


def _semantic_commitment_projection(
    value: SemanticPlanCommitment,
    *,
    selected_carrier_provenance: Sequence[Mapping[str, Any]],
) -> Mapping[str, Any]:
    if not isinstance(value, SemanticPlanCommitment):
        raise GlyphAssetError(
            "semantic_commitment must be a SemanticPlanCommitment"
        )
    external_plan_sha = _require_sha256(
        value.external_semantic_plan_sha256,
        context="semantic_commitment.external_semantic_plan_sha256",
    )
    inverse_proof_sha = _require_sha256(
        value.inverse_proof_sha256,
        context="semantic_commitment.inverse_proof_sha256",
    )
    ordered_attempt_ledger_sha = _require_sha256(
        value.ordered_attempt_ledger_sha256,
        context="semantic_commitment.ordered_attempt_ledger_sha256",
    )
    attempt_record_count = _require_int(
        value.attempt_record_count,
        context="semantic_commitment.attempt_record_count",
        minimum=0,
        maximum=MAX_ATTEMPT_RECORDS,
    )
    assembled_carrier_count = _require_int(
        value.assembled_carrier_count,
        context="semantic_commitment.assembled_carrier_count",
        minimum=0,
        maximum=MAX_GENERATED_CHUNK_BUDGET,
    )
    request_model_generator_invocations = _require_int(
        value.request_model_generator_invocations,
        context="semantic_commitment.request_model_generator_invocations",
        minimum=0,
        maximum=MAX_ATTEMPT_RECORDS,
    )
    request_generated_chunk_count = _require_int(
        value.request_generated_chunk_count,
        context="semantic_commitment.request_generated_chunk_count",
        minimum=0,
        maximum=MAX_GENERATED_CHUNK_BUDGET,
    )
    request_generated_text_units = _require_int(
        value.request_generated_text_units,
        context="semantic_commitment.request_generated_text_units",
        minimum=0,
        maximum=MAX_GENERATED_TEXT_UNIT_BUDGET,
    )
    request_generated_chunk_budget = _require_int(
        value.request_generated_chunk_budget,
        context="semantic_commitment.request_generated_chunk_budget",
    )
    request_generated_text_unit_budget = _require_int(
        value.request_generated_text_unit_budget,
        context="semantic_commitment.request_generated_text_unit_budget",
    )
    if request_generated_chunk_budget != MAX_GENERATED_CHUNK_BUDGET:
        raise GlyphAssetError("semantic commitment chunk budget is not frozen")
    if request_generated_text_unit_budget != MAX_GENERATED_TEXT_UNIT_BUDGET:
        raise GlyphAssetError("semantic commitment text-unit budget is not frozen")
    records = value.ordered_attempt_records
    if not isinstance(records, tuple):
        raise GlyphAssetError(
            "semantic commitment ordered attempt records must be an immutable tuple"
        )
    ledger_projection = _ordered_attempt_ledger_projection(records)
    actual_ledger_sha = _sha256_bytes(_canonical_json_bytes(ledger_projection))
    if actual_ledger_sha != ordered_attempt_ledger_sha:
        raise GlyphAssetError("ordered attempt ledger hash mismatch")
    if attempt_record_count != len(records):
        raise GlyphAssetError("ordered attempt ledger record count mismatch")
    if request_model_generator_invocations != len(records):
        raise GlyphAssetError(
            "request generator invocation total does not match attempt ledger"
        )
    actual_generated_chunks = sum(
        record["generated_chunk_count"]
        for record in ledger_projection["records"]
    )
    actual_generated_text_units = sum(
        record["generated_text_units"]
        for record in ledger_projection["records"]
    )
    if request_generated_chunk_count != actual_generated_chunks:
        raise GlyphAssetError(
            "request generated chunk total does not match attempt ledger"
        )
    if request_generated_text_units != actual_generated_text_units:
        raise GlyphAssetError(
            "request generated text-unit total does not match attempt ledger"
        )
    if assembled_carrier_count != len(selected_carrier_provenance):
        raise GlyphAssetError(
            "assembled carrier total does not match final carrier segments"
        )
    if assembled_carrier_count == 0:
        if (
            records
            or attempt_record_count != 0
            or request_model_generator_invocations != 0
            or request_generated_chunk_count != 0
            or request_generated_text_units != 0
        ):
            raise GlyphAssetError(
                "asset-only assembly must have an exact zero-work attempt ledger"
            )
    elif not records:
        raise GlyphAssetError(
            "carrier assembly requires a non-empty attempt ledger"
        )
    if request_model_generator_invocations < len(selected_carrier_provenance):
        raise GlyphAssetError(
            "request generator invocations omit selected carrier provenance"
        )
    if request_generated_chunk_count < len(selected_carrier_provenance):
        raise GlyphAssetError(
            "request generated chunks omit selected carrier provenance"
        )
    records_by_id = {
        record["attempt_id"]: record for record in ledger_projection["records"]
    }
    selected_ids: set[str] = set()
    for index, selected in enumerate(selected_carrier_provenance):
        context = f"selected_carrier_provenance[{index}]"
        selected_attempt_id = _require_string(
            selected.get("selected_attempt_id"),
            context=f"{context}.selected_attempt_id",
        )
        if selected_attempt_id in selected_ids:
            raise GlyphAssetError(
                "multiple final carriers claim the same selected attempt"
            )
        selected_ids.add(selected_attempt_id)
        record = records_by_id.get(selected_attempt_id)
        if record is None or record["generated_chunk_count"] != 1:
            raise GlyphAssetError(
                "selected carrier is not attributable to a generated attempt"
            )
        selected_seed = _require_int(
            selected.get("selected_seed"),
            context=f"{context}.selected_seed",
            minimum=0,
            maximum=(2**31) - 1,
        )
        selected_text_sha = _require_sha256(
            selected.get("selected_text_sha256"),
            context=f"{context}.selected_text_sha256",
        )
        selected_endpoint_sha = _require_sha256(
            selected.get("selected_endpoint_evidence_sha256"),
            context=f"{context}.selected_endpoint_evidence_sha256",
        )
        if (
            selected_seed != record["seed"]
            or selected_text_sha != record["text_sha256"]
            or selected_endpoint_sha != record["endpoint_evidence_sha256"]
        ):
            raise GlyphAssetError(
                "selected carrier provenance does not match its attempt record"
            )
    return {
        "schema": SEMANTIC_COMMITMENT_SCHEMA,
        "external_semantic_plan_sha256": external_plan_sha,
        "inverse_proof_sha256": inverse_proof_sha,
        "attempt_ledger_schema": ATTEMPT_LEDGER_SCHEMA,
        "ordered_attempt_ledger_sha256": ordered_attempt_ledger_sha,
        "attempt_record_count": attempt_record_count,
        "assembled_carrier_count": assembled_carrier_count,
        "request_model_generator_invocations": (
            request_model_generator_invocations
        ),
        "request_generated_chunk_count": request_generated_chunk_count,
        "request_generated_text_units": request_generated_text_units,
        "request_generated_chunk_budget": request_generated_chunk_budget,
        "request_generated_text_unit_budget": (
            request_generated_text_unit_budget
        ),
    }


def _parse_pins(value: Any, *, context: str) -> tuple[ArtifactPin, ...]:
    if not isinstance(value, list) or not value:
        raise GlyphAssetError(f"{context} must be a non-empty list")
    pins: list[ArtifactPin] = []
    names: set[str] = set()
    for index, item in enumerate(value):
        item_context = f"{context}[{index}]"
        fields = _require_exact_keys(
            item,
            ("name", "uri", "revision", "sha256"),
            context=item_context,
        )
        name = _require_string(fields["name"], context=f"{item_context}.name")
        if name in names:
            raise GlyphAssetError(f"duplicate pin name in {context}: {name!r}")
        names.add(name)
        pins.append(
            ArtifactPin(
                name=name,
                uri=_require_string(fields["uri"], context=f"{item_context}.uri"),
                revision=_require_string(
                    fields["revision"], context=f"{item_context}.revision"
                ),
                sha256=_require_sha256(
                    fields["sha256"], context=f"{item_context}.sha256"
                ),
            )
        )
    return tuple(pins)


def _safe_relative_asset_path(root: Path, relative_path: Any) -> Path:
    text = _require_string(relative_path, context="asset.relative_path")
    if "\\" in text:
        raise GlyphAssetError("asset.relative_path must use POSIX separators")
    posix = PurePosixPath(text)
    if posix.is_absolute() or not posix.parts:
        raise GlyphAssetError("asset.relative_path must be relative")
    if any(part in ("", ".", "..") for part in posix.parts):
        raise GlyphAssetError("asset.relative_path contains unsafe components")

    current = root
    for part in posix.parts:
        current = current / part
        try:
            info = os.lstat(current)
        except FileNotFoundError as exc:
            raise GlyphAssetError(f"asset file is missing: {text}") from exc
        if stat.S_ISLNK(info.st_mode):
            raise GlyphAssetError(f"asset path contains a symlink: {text}")
    if not stat.S_ISREG(os.lstat(current).st_mode):
        raise GlyphAssetError(f"asset path is not a regular file: {text}")
    resolved_root = root.resolve(strict=True)
    resolved_file = current.resolve(strict=True)
    try:
        resolved_file.relative_to(resolved_root)
    except ValueError as exc:
        raise GlyphAssetError(f"asset path escapes bundle root: {text}") from exc
    return current


def pcm16_array_to_bytes(pcm16: np.ndarray) -> bytes:
    array = np.asarray(pcm16)
    if array.ndim != 1 or array.dtype.kind != "i" or array.dtype.itemsize != 2:
        raise GlyphAssetError("PCM array must be one-dimensional signed int16")
    little = np.asarray(array, dtype="<i2")
    return little.tobytes(order="C")


def pcm16_bytes_to_array(pcm16_le: bytes) -> np.ndarray:
    if not isinstance(pcm16_le, bytes) or len(pcm16_le) % 2:
        raise GlyphAssetError("PCM16 bytes must be immutable and frame-aligned")
    result = np.frombuffer(pcm16_le, dtype="<i2").copy()
    result.flags.writeable = False
    return result


def _active_bounds(pcm: np.ndarray) -> tuple[int, int]:
    if pcm.size == 0:
        return 0, 0
    normalized = np.abs(pcm.astype(np.float64)) / 32768.0
    peak = float(np.max(normalized))
    if peak == 0.0:
        return 0, 0
    threshold = peak * (10.0 ** (-35.0 / 20.0))
    active = np.flatnonzero(normalized >= threshold)
    if active.size == 0:
        return 0, 0
    return int(active[0]), int(active[-1]) + 1


def measure_pcm16(pcm16_le: bytes) -> PcmMeasurements:
    pcm = np.frombuffer(pcm16_le, dtype="<i2")
    if pcm.size == 0:
        raise GlyphAssetError("PCM16 audio must not be empty")
    normalized = pcm.astype(np.float64) / 32768.0
    rms = float(np.sqrt(np.mean(np.square(normalized), dtype=np.float64)))
    peak = float(np.max(np.abs(normalized)))
    active_start, active_end = _active_bounds(pcm)
    active_duration = (
        float(active_end - active_start) / SAMPLE_RATE
        if active_end > active_start
        else 0.0
    )
    terminal_silence = int(pcm.size - active_end) if active_end else int(pcm.size)
    return PcmMeasurements(
        rms=rms,
        peak=peak,
        active_duration_seconds=active_duration,
        terminal_silence_samples=terminal_silence,
    )


def _validate_asset_leading_edge(pcm: np.ndarray, *, context: str) -> None:
    """Recompute the frozen, builder-side 3 ms anti-click contract."""

    samples = np.asarray(pcm, dtype="<i2")
    if samples.size < ASSET_LEADING_FADE_SAMPLES + 1:
        raise GlyphAssetError(f"{context} is too short for the 3 ms leading fade")
    if int(samples[0]) != 0:
        raise GlyphAssetError(f"{context} first PCM code must be exact zero")
    leading = samples[:ASSET_LEADING_FADE_SAMPLES].astype(np.int64)
    envelope = np.floor(
        32768.0
        * np.linspace(
            0.0,
            1.0,
            num=ASSET_LEADING_FADE_SAMPLES,
            endpoint=True,
            dtype=np.float64,
        )
    ).astype(np.int64)
    if np.any(np.abs(leading) > envelope):
        raise GlyphAssetError(
            f"{context} violates the frozen 3 ms leading fade envelope"
        )
    edge_window = samples[: ASSET_LEADING_FADE_SAMPLES + 1].astype(np.int64)
    maximum_code_jump = int(
        math.floor(MAX_BOUNDARY_ABSOLUTE_JUMP * 32768.0)
    )
    if np.any(np.abs(np.diff(edge_window)) > maximum_code_jump):
        raise GlyphAssetError(
            f"{context} leading edge exceeds the frozen click limit"
        )


def _terminal_exact_zero_samples(pcm: np.ndarray) -> int:
    samples = np.asarray(pcm)
    nonzero = np.flatnonzero(samples)
    if nonzero.size == 0:
        return int(samples.size)
    return int(samples.size - int(nonzero[-1]) - 1)


def _read_regular_file_nofollow(
    path: Path,
    *,
    maximum_bytes: int,
    context: str,
) -> bytes:
    """Read one regular file through one no-follow descriptor.

    The descriptor is also the source of all size/type checks.  Callers must
    decode the returned bytes rather than reopen ``path``; this closes the
    hash-then-decode WAV replacement window.
    """

    nofollow = getattr(os, "O_NOFOLLOW", None)
    if nofollow is None:
        raise GlyphAssetError(f"{context} requires O_NOFOLLOW support")
    flags = os.O_RDONLY | nofollow
    if hasattr(os, "O_CLOEXEC"):
        flags |= os.O_CLOEXEC
    try:
        descriptor = os.open(path, flags)
    except (FileNotFoundError, OSError) as exc:
        raise GlyphAssetError(f"{context} could not be opened without symlinks") from exc
    try:
        before = os.fstat(descriptor)
        if not stat.S_ISREG(before.st_mode):
            raise GlyphAssetError(f"{context} must be a regular file")
        if before.st_size <= 0 or before.st_size > maximum_bytes:
            raise GlyphAssetError(f"{context} is empty or oversized")
        chunks: list[bytes] = []
        remaining = int(before.st_size)
        while remaining:
            chunk = os.read(descriptor, min(remaining, 1_048_576))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        data = b"".join(chunks)
        after = os.fstat(descriptor)
        identity_before = (
            before.st_dev,
            before.st_ino,
            before.st_size,
            before.st_mtime_ns,
            before.st_ctime_ns,
        )
        identity_after = (
            after.st_dev,
            after.st_ino,
            after.st_size,
            after.st_mtime_ns,
            after.st_ctime_ns,
        )
        if (
            len(data) != before.st_size
            or remaining != 0
            or identity_before != identity_after
        ):
            raise GlyphAssetError(f"{context} changed while it was being read")
        return data
    finally:
        os.close(descriptor)


def _decode_pcm16_wav_container(
    container: bytes,
    *,
    display_name: str,
    expected_sample_count: int,
) -> bytes:
    try:
        with wave.open(io.BytesIO(container), "rb") as reader:
            if reader.getnchannels() != 1:
                raise GlyphAssetError(f"asset must be mono: {display_name}")
            if reader.getsampwidth() != 2:
                raise GlyphAssetError(f"asset must be signed PCM16: {display_name}")
            if reader.getframerate() != SAMPLE_RATE:
                raise GlyphAssetError(f"asset must be 48 kHz: {display_name}")
            if reader.getcomptype() != "NONE":
                raise GlyphAssetError(
                    f"asset must be uncompressed PCM: {display_name}"
                )
            frame_count = reader.getnframes()
            if frame_count != expected_sample_count:
                raise GlyphAssetError(
                    f"asset sample count mismatch: {display_name}"
                )
            if frame_count <= 0 or frame_count > MAX_ASSET_SAMPLES:
                raise GlyphAssetError(
                    f"asset sample count is invalid or oversized: {display_name}"
                )
            pcm16_le = reader.readframes(frame_count)
            if reader.readframes(1):
                raise GlyphAssetError(
                    f"asset contains uncounted frames: {display_name}"
                )
    except (wave.Error, EOFError) as exc:
        raise GlyphAssetError(
            f"asset is not a valid PCM WAV: {display_name}"
        ) from exc
    return pcm16_le


def _read_pcm16_wav(
    path: Path,
    *,
    expected_container_sha256: str,
    expected_pcm_sha256: str,
    expected_sample_count: int,
) -> tuple[bytes, bytes]:
    container = _read_regular_file_nofollow(
        path,
        maximum_bytes=MAX_ASSET_CONTAINER_BYTES,
        context=f"asset container {path.name}",
    )
    actual_container_sha256 = _sha256_bytes(container)
    if actual_container_sha256 != expected_container_sha256:
        raise GlyphAssetError(f"asset container hash mismatch: {path.name}")

    pcm16_le = _decode_pcm16_wav_container(
        container,
        display_name=path.name,
        expected_sample_count=expected_sample_count,
    )
    if len(pcm16_le) != expected_sample_count * 2:
        raise GlyphAssetError(f"decoded asset byte count mismatch: {path.name}")
    if _sha256_bytes(pcm16_le) != expected_pcm_sha256:
        raise GlyphAssetError(f"decoded asset PCM hash mismatch: {path.name}")
    return container, pcm16_le


def _metric_close(actual: float, expected: float) -> bool:
    return math.isclose(actual, expected, rel_tol=1e-9, abs_tol=1e-12)


def _parse_ecapa(value: Any, *, context: str) -> EcapaAttestation:
    fields = _require_exact_keys(
        value,
        ("cosine_similarity", "minimum_cosine_similarity"),
        context=context,
    )
    similarity = _require_finite(
        fields["cosine_similarity"],
        context=f"{context}.cosine_similarity",
        minimum=-1.0,
        maximum=1.0,
    )
    threshold = _require_finite(
        fields["minimum_cosine_similarity"],
        context=f"{context}.minimum_cosine_similarity",
        minimum=-1.0,
        maximum=1.0,
    )
    if threshold != MIN_ECAPA_SIMILARITY:
        raise GlyphAssetError(
            f"{context} attested threshold is not the frozen v1 threshold"
        )
    return EcapaAttestation(similarity, threshold)


def _parse_squim(value: Any, *, context: str) -> SquimAttestation:
    fields = _require_exact_keys(
        value,
        (
            "stoi",
            "pesq",
            "si_sdr",
            "minimum_stoi",
            "minimum_pesq",
            "minimum_si_sdr",
        ),
        context=context,
    )
    stoi = _require_finite(
        fields["stoi"], context=f"{context}.stoi", minimum=0.0, maximum=1.0
    )
    pesq = _require_finite(
        fields["pesq"], context=f"{context}.pesq", minimum=-0.5, maximum=4.5
    )
    si_sdr = _require_finite(fields["si_sdr"], context=f"{context}.si_sdr")
    minimum_stoi = _require_finite(
        fields["minimum_stoi"],
        context=f"{context}.minimum_stoi",
        minimum=0.0,
        maximum=1.0,
    )
    minimum_pesq = _require_finite(
        fields["minimum_pesq"],
        context=f"{context}.minimum_pesq",
        minimum=-0.5,
        maximum=4.5,
    )
    minimum_si_sdr = _require_finite(
        fields["minimum_si_sdr"], context=f"{context}.minimum_si_sdr"
    )
    if (
        minimum_stoi != MIN_SQUIM_STOI
        or minimum_pesq != MIN_SQUIM_PESQ
        or minimum_si_sdr != MIN_SQUIM_SI_SDR
    ):
        raise GlyphAssetError(
            f"{context} attested thresholds are not the frozen v1 thresholds"
        )
    return SquimAttestation(
        stoi,
        pesq,
        si_sdr,
        minimum_stoi,
        minimum_pesq,
        minimum_si_sdr,
    )


def _parse_endpoint(value: Any, *, context: str) -> EndpointAttestation:
    fields = _require_exact_keys(
        value,
        (
            "stop_reason",
            "turbo_cer",
            "large_v3_cer",
            "prefix_cer",
            "suffix_cer",
            "extra_tail_units",
            "tail_energy_ratio",
            "maximum_tail_energy_ratio",
            "terminal_silence_samples",
        ),
        context=context,
    )
    stop_reason = _require_string(
        fields["stop_reason"], context=f"{context}.stop_reason"
    )
    if stop_reason not in ("stop_token", "verified_hard_cap"):
        raise GlyphAssetError(f"{context}.stop_reason is not an allowed completion")
    turbo_cer = _require_finite(
        fields["turbo_cer"], context=f"{context}.turbo_cer", minimum=0.0
    )
    large_v3_cer = _require_finite(
        fields["large_v3_cer"], context=f"{context}.large_v3_cer", minimum=0.0
    )
    prefix_cer = _require_finite(
        fields["prefix_cer"], context=f"{context}.prefix_cer", minimum=0.0
    )
    suffix_cer = _require_finite(
        fields["suffix_cer"], context=f"{context}.suffix_cer", minimum=0.0
    )
    extra_tail_units = _require_int(
        fields["extra_tail_units"],
        context=f"{context}.extra_tail_units",
        minimum=0,
    )
    tail_energy_ratio = _require_finite(
        fields["tail_energy_ratio"],
        context=f"{context}.tail_energy_ratio",
        minimum=0.0,
    )
    maximum_tail_energy_ratio = _require_finite(
        fields["maximum_tail_energy_ratio"],
        context=f"{context}.maximum_tail_energy_ratio",
        minimum=0.0,
    )
    terminal_silence_samples = _require_int(
        fields["terminal_silence_samples"],
        context=f"{context}.terminal_silence_samples",
        minimum=0,
    )
    if maximum_tail_energy_ratio != MAX_ENDPOINT_TAIL_ENERGY_RATIO:
        raise GlyphAssetError(
            f"{context} attested tail threshold is not frozen v1"
        )
    return EndpointAttestation(
        stop_reason,
        turbo_cer,
        large_v3_cer,
        prefix_cer,
        suffix_cer,
        extra_tail_units,
        tail_energy_ratio,
        maximum_tail_energy_ratio,
        terminal_silence_samples,
    )


def _parse_asset(
    value: Any,
    *,
    index: int,
    expected_spec: GlyphSpec,
    root: Path,
) -> GlyphAsset:
    context = f"assets[{index}]"
    fields = _require_exact_keys(
        value,
        (
            "id",
            "raw_byte",
            "canonical_token",
            "relative_path",
            "container_sha256",
            "pcm_sha256",
            "sample_rate",
            "channels",
            "sample_format",
            "sample_count",
            "rms",
            "peak",
            "active_duration_seconds",
            "core_start_sample",
            "core_end_sample",
            "core_pcm_sha256",
            "ecapa",
            "squim",
            "endpoint",
        ),
        context=context,
    )
    asset_id = _require_string(fields["id"], context=f"{context}.id")
    raw_byte = _require_string(fields["raw_byte"], context=f"{context}.raw_byte")
    canonical_token = _require_string(
        fields["canonical_token"], context=f"{context}.canonical_token"
    )
    if (
        asset_id != expected_spec.asset_id
        or raw_byte != expected_spec.raw_byte
        or canonical_token != expected_spec.canonical_token
    ):
        raise GlyphAssetError(
            f"{context} does not match the frozen v1 grammar/order"
        )
    try:
        encoded_raw = raw_byte.encode("ascii", errors="strict")
    except UnicodeEncodeError as exc:
        raise GlyphAssetError(f"{context}.raw_byte must be ASCII") from exc
    if len(encoded_raw) != 1:
        raise GlyphAssetError(f"{context}.raw_byte must be exactly one byte")

    relative_path = _require_string(
        fields["relative_path"], context=f"{context}.relative_path"
    )
    container_sha256 = _require_sha256(
        fields["container_sha256"], context=f"{context}.container_sha256"
    )
    pcm_sha256 = _require_sha256(
        fields["pcm_sha256"], context=f"{context}.pcm_sha256"
    )
    sample_rate = _require_int(
        fields["sample_rate"], context=f"{context}.sample_rate"
    )
    channels = _require_int(fields["channels"], context=f"{context}.channels")
    sample_format = _require_string(
        fields["sample_format"], context=f"{context}.sample_format"
    )
    if (
        sample_rate != SAMPLE_RATE
        or channels != 1
        or sample_format != PCM_SAMPLE_FORMAT
    ):
        raise GlyphAssetError(f"{context} format declaration is not PCM16/mono/48k")
    sample_count = _require_int(
        fields["sample_count"],
        context=f"{context}.sample_count",
        minimum=1,
        maximum=MAX_ASSET_SAMPLES,
    )
    rms = _require_finite(
        fields["rms"], context=f"{context}.rms", minimum=0.0, maximum=1.0
    )
    peak = _require_finite(
        fields["peak"], context=f"{context}.peak", minimum=0.0, maximum=1.0
    )
    active_duration_seconds = _require_finite(
        fields["active_duration_seconds"],
        context=f"{context}.active_duration_seconds",
        minimum=0.0,
    )
    core_start = _require_int(
        fields["core_start_sample"],
        context=f"{context}.core_start_sample",
        minimum=CARRIER_EDGE_FADE_SAMPLES,
    )
    core_end = _require_int(
        fields["core_end_sample"],
        context=f"{context}.core_end_sample",
        minimum=core_start + 1,
        maximum=sample_count - CARRIER_EDGE_FADE_SAMPLES,
    )
    core_pcm_sha256 = _require_sha256(
        fields["core_pcm_sha256"], context=f"{context}.core_pcm_sha256"
    )
    ecapa = _parse_ecapa(fields["ecapa"], context=f"{context}.ecapa")
    squim = _parse_squim(fields["squim"], context=f"{context}.squim")
    endpoint = _parse_endpoint(
        fields["endpoint"], context=f"{context}.endpoint"
    )

    asset_path = _safe_relative_asset_path(root, relative_path)
    container_bytes, pcm16_le = _read_pcm16_wav(
        asset_path,
        expected_container_sha256=container_sha256,
        expected_pcm_sha256=pcm_sha256,
        expected_sample_count=sample_count,
    )
    measurements = measure_pcm16(pcm16_le)
    if measurements.rms <= 0.0 or measurements.active_duration_seconds <= 0.0:
        raise GlyphAssetError(f"{context} decoded PCM must contain active speech")
    if not _metric_close(measurements.rms, rms):
        raise GlyphAssetError(f"{context}.rms does not match decoded PCM")
    if not _metric_close(measurements.peak, peak):
        raise GlyphAssetError(f"{context}.peak does not match decoded PCM")
    if not _metric_close(
        measurements.active_duration_seconds, active_duration_seconds
    ):
        raise GlyphAssetError(
            f"{context}.active_duration_seconds does not match decoded PCM"
        )
    if measurements.terminal_silence_samples != endpoint.terminal_silence_samples:
        raise GlyphAssetError(
            f"{context}.endpoint terminal silence does not match decoded PCM"
        )
    decoded_pcm = np.frombuffer(pcm16_le, dtype="<i2")
    _validate_asset_leading_edge(decoded_pcm, context=context)
    if (
        sample_count < FINAL_SILENCE_SAMPLES
        or np.any(decoded_pcm[-FINAL_SILENCE_SAMPLES:])
    ):
        raise GlyphAssetError(
            f"{context} must end in 180 ms of exact zero PCM"
        )
    core_bytes = pcm16_le[core_start * 2 : core_end * 2]
    if _sha256_bytes(core_bytes) != core_pcm_sha256:
        raise GlyphAssetError(f"{context}.core_pcm_sha256 mismatch")

    return GlyphAsset(
        asset_id=asset_id,
        raw_byte=raw_byte,
        canonical_token=canonical_token,
        relative_path=relative_path,
        container_sha256=container_sha256,
        pcm_sha256=pcm_sha256,
        sample_rate=sample_rate,
        channels=channels,
        sample_format=sample_format,
        sample_count=sample_count,
        rms=rms,
        peak=peak,
        active_duration_seconds=active_duration_seconds,
        core_start_sample=core_start,
        core_end_sample=core_end,
        core_pcm_sha256=core_pcm_sha256,
        ecapa=ecapa,
        squim=squim,
        endpoint=endpoint,
        container_bytes=container_bytes,
        pcm16_le=pcm16_le,
    )


def compute_bundle_sha256(bundle_version: str, raw_assets: Sequence[Any]) -> str:
    """Return the v1 bundle digest over the exact ordered asset evidence."""

    return _sha256_bytes(
        _canonical_json_bytes(
            {
                "bundle_version": bundle_version,
                "assets": list(raw_assets),
            }
        )
    )


def _pin_projection(pin: ArtifactPin) -> Mapping[str, Any]:
    return {
        "name": pin.name,
        "uri": pin.uri,
        "revision": pin.revision,
        "sha256": pin.sha256,
    }


def _asset_manifest_projection(asset: GlyphAsset) -> Mapping[str, Any]:
    return {
        "id": asset.asset_id,
        "raw_byte": asset.raw_byte,
        "canonical_token": asset.canonical_token,
        "relative_path": asset.relative_path,
        "container_sha256": asset.container_sha256,
        "pcm_sha256": asset.pcm_sha256,
        "sample_rate": asset.sample_rate,
        "channels": asset.channels,
        "sample_format": asset.sample_format,
        "sample_count": asset.sample_count,
        "rms": asset.rms,
        "peak": asset.peak,
        "active_duration_seconds": asset.active_duration_seconds,
        "core_start_sample": asset.core_start_sample,
        "core_end_sample": asset.core_end_sample,
        "core_pcm_sha256": asset.core_pcm_sha256,
        "ecapa": {
            "cosine_similarity": asset.ecapa.cosine_similarity,
            "minimum_cosine_similarity": asset.ecapa.minimum_cosine_similarity,
        },
        "squim": {
            "stoi": asset.squim.stoi,
            "pesq": asset.squim.pesq,
            "si_sdr": asset.squim.si_sdr,
            "minimum_stoi": asset.squim.minimum_stoi,
            "minimum_pesq": asset.squim.minimum_pesq,
            "minimum_si_sdr": asset.squim.minimum_si_sdr,
        },
        "endpoint": {
            "stop_reason": asset.endpoint.stop_reason,
            "turbo_cer": asset.endpoint.turbo_cer,
            "large_v3_cer": asset.endpoint.large_v3_cer,
            "prefix_cer": asset.endpoint.prefix_cer,
            "suffix_cer": asset.endpoint.suffix_cer,
            "extra_tail_units": asset.endpoint.extra_tail_units,
            "tail_energy_ratio": asset.endpoint.tail_energy_ratio,
            "maximum_tail_energy_ratio": asset.endpoint.maximum_tail_energy_ratio,
            "terminal_silence_samples": asset.endpoint.terminal_silence_samples,
        },
    }


def _bundle_manifest_projection(bundle: GlyphAssetBundle) -> Mapping[str, Any]:
    assets = [_asset_manifest_projection(asset) for asset in bundle.assets]
    return {
        "manifest_schema": bundle.manifest_schema,
        "bundle_version": bundle.bundle_version,
        "grammar_sha256": bundle.grammar_sha256,
        "bundle_sha256": bundle.bundle_sha256,
        "speaker_anchor_sha256": bundle.speaker_anchor_sha256,
        "join_profile": dict(bundle.join_profile),
        "source_pins": [_pin_projection(pin) for pin in bundle.source_pins],
        "model_pins": [_pin_projection(pin) for pin in bundle.model_pins],
        "evaluator_pins": [_pin_projection(pin) for pin in bundle.evaluator_pins],
        "assets": assets,
    }


def _validate_loaded_asset_integrity(
    asset: GlyphAsset,
    *,
    expected_spec: GlyphSpec,
    index: int,
) -> None:
    context = f"trusted_bundle.assets[{index}]"
    if (
        asset.asset_id != expected_spec.asset_id
        or asset.raw_byte != expected_spec.raw_byte
        or asset.canonical_token != expected_spec.canonical_token
    ):
        raise GlyphAssetError(
            f"{context} labels do not match the frozen grammar/order"
        )
    if _sha256_bytes(asset.container_bytes) != asset.container_sha256:
        raise GlyphAssetError(f"{context} retained container hash mismatch")
    decoded = _decode_pcm16_wav_container(
        asset.container_bytes,
        display_name=asset.relative_path,
        expected_sample_count=asset.sample_count,
    )
    if decoded != asset.pcm16_le:
        raise GlyphAssetError(f"{context} retained container/PCM mismatch")
    if _sha256_bytes(asset.pcm16_le) != asset.pcm_sha256:
        raise GlyphAssetError(f"{context} retained PCM hash mismatch")
    if len(asset.pcm16_le) != asset.sample_count * 2:
        raise GlyphAssetError(f"{context} retained PCM length mismatch")
    measurements = measure_pcm16(asset.pcm16_le)
    if measurements.rms <= 0.0 or measurements.active_duration_seconds <= 0.0:
        raise GlyphAssetError(f"{context} retained PCM must contain active speech")
    if (
        not _metric_close(measurements.rms, asset.rms)
        or not _metric_close(measurements.peak, asset.peak)
        or not _metric_close(
            measurements.active_duration_seconds,
            asset.active_duration_seconds,
        )
        or measurements.terminal_silence_samples
        != asset.endpoint.terminal_silence_samples
    ):
        raise GlyphAssetError(f"{context} retained PCM measurements mismatch")
    pcm = np.frombuffer(asset.pcm16_le, dtype="<i2")
    _validate_asset_leading_edge(pcm, context=context)
    if (
        pcm.size < FINAL_SILENCE_SAMPLES
        or np.any(pcm[-FINAL_SILENCE_SAMPLES:])
    ):
        raise GlyphAssetError(f"{context} retained PCM final silence mismatch")
    if not (
        CARRIER_EDGE_FADE_SAMPLES
        <= asset.core_start_sample
        < asset.core_end_sample
        <= asset.sample_count - CARRIER_EDGE_FADE_SAMPLES
    ):
        raise GlyphAssetError(f"{context} retained core range mismatch")
    core = asset.pcm16_le[
        asset.core_start_sample * 2 : asset.core_end_sample * 2
    ]
    if _sha256_bytes(core) != asset.core_pcm_sha256:
        raise GlyphAssetError(f"{context} retained core hash mismatch")
    # These calls validate attestation schema/ranges and frozen profile IDs.
    # They deliberately do not execute or impersonate the pinned evaluators.
    projected = _asset_manifest_projection(asset)
    _parse_ecapa(projected["ecapa"], context=f"{context}.ecapa")
    _parse_squim(projected["squim"], context=f"{context}.squim")
    _parse_endpoint(projected["endpoint"], context=f"{context}.endpoint")


def _validate_loaded_bundle_integrity(bundle: GlyphAssetBundle) -> None:
    if not isinstance(bundle, GlyphAssetBundle):
        raise GlyphAssetError("trusted_bundle must be a loaded GlyphAssetBundle")
    if bundle.manifest_schema != MANIFEST_SCHEMA:
        raise GlyphAssetError("trusted bundle manifest schema mismatch")
    if bundle.grammar_sha256 != V1_GRAMMAR_SHA256:
        raise GlyphAssetError("trusted bundle grammar mismatch")
    if dict(bundle.join_profile) != dict(V1_JOIN_PROFILE):
        raise GlyphAssetError("trusted bundle join profile mismatch")
    if len(bundle.assets) != ASSET_COUNT:
        raise GlyphAssetError("trusted bundle asset count mismatch")
    if _sha256_bytes(bundle.manifest_bytes) != bundle.manifest_sha256:
        raise GlyphAssetError("trusted bundle retained manifest hash mismatch")
    try:
        retained_manifest = json.loads(
            bundle.manifest_bytes,
            object_pairs_hook=_reject_duplicate_json_keys,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise GlyphAssetError("trusted bundle retained manifest is invalid") from exc
    if retained_manifest != _bundle_manifest_projection(bundle):
        raise GlyphAssetError("trusted bundle retained manifest projection mismatch")
    raw_assets = [_asset_manifest_projection(asset) for asset in bundle.assets]
    recomputed_bundle_sha = compute_bundle_sha256(
        bundle.bundle_version,
        raw_assets,
    )
    if recomputed_bundle_sha != bundle.bundle_sha256:
        raise GlyphAssetError("trusted bundle recomputed bundle hash mismatch")
    for index, (asset, expected_spec) in enumerate(
        zip(bundle.assets, V1_GLYPH_SPECS)
    ):
        _validate_loaded_asset_integrity(
            asset,
            expected_spec=expected_spec,
            index=index,
        )
    uniqueness_fields = (
        tuple(asset.asset_id for asset in bundle.assets),
        tuple(asset.raw_byte for asset in bundle.assets),
        tuple(asset.canonical_token for asset in bundle.assets),
        tuple(asset.relative_path for asset in bundle.assets),
        tuple(asset.container_sha256 for asset in bundle.assets),
        tuple(asset.pcm_sha256 for asset in bundle.assets),
        tuple(asset.core_pcm_sha256 for asset in bundle.assets),
    )
    if any(len(set(values)) != len(values) for values in uniqueness_fields):
        raise GlyphAssetError("trusted bundle contains duplicate immutable evidence")


def load_glyph_asset_bundle(
    manifest_path: str | os.PathLike[str],
    *,
    expected_manifest_sha256: str,
) -> GlyphAssetBundle:
    """Integrity-load a frozen 32-entry v1 asset bundle.

    ``expected_manifest_sha256`` is mandatory and must originate outside the
    bundle.  Therefore an attacker cannot modify a WAV and simply rewrite both
    the per-file and bundle hashes inside the same manifest.  Recorded ECAPA,
    SQUIM and ASR values remain attestations only: this function never loads
    those models and never returns a runtime quality pass.
    """

    expected_manifest_sha256 = _require_sha256(
        expected_manifest_sha256, context="expected_manifest_sha256"
    )
    path = Path(manifest_path)
    try:
        info = os.lstat(path)
    except FileNotFoundError as exc:
        raise GlyphAssetError("asset manifest is missing") from exc
    if stat.S_ISLNK(info.st_mode):
        raise GlyphAssetError("asset manifest must not be a symlink")
    if not stat.S_ISREG(info.st_mode):
        raise GlyphAssetError("asset manifest must be a regular file")
    manifest_bytes = _read_regular_file_nofollow(
        path,
        maximum_bytes=MAX_MANIFEST_BYTES,
        context="asset manifest",
    )
    manifest_sha256 = _sha256_bytes(manifest_bytes)
    if manifest_sha256 != expected_manifest_sha256:
        raise GlyphAssetError("asset manifest hash mismatch")
    try:
        raw = json.loads(
            manifest_bytes,
            object_pairs_hook=_reject_duplicate_json_keys,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise GlyphAssetError("asset manifest is not valid UTF-8 JSON") from exc
    fields = _require_exact_keys(
        raw,
        (
            "manifest_schema",
            "bundle_version",
            "grammar_sha256",
            "bundle_sha256",
            "speaker_anchor_sha256",
            "join_profile",
            "source_pins",
            "model_pins",
            "evaluator_pins",
            "assets",
        ),
        context="manifest",
    )
    manifest_schema = _require_string(
        fields["manifest_schema"], context="manifest.manifest_schema"
    )
    if manifest_schema != MANIFEST_SCHEMA:
        raise GlyphAssetError("unsupported asset manifest schema")
    bundle_version = _require_string(
        fields["bundle_version"], context="manifest.bundle_version"
    )
    grammar_sha256 = _require_sha256(
        fields["grammar_sha256"], context="manifest.grammar_sha256"
    )
    if grammar_sha256 != V1_GRAMMAR_SHA256:
        raise GlyphAssetError("manifest grammar is not the frozen v1 grammar")
    bundle_sha256 = _require_sha256(
        fields["bundle_sha256"], context="manifest.bundle_sha256"
    )
    speaker_anchor_sha256 = _require_sha256(
        fields["speaker_anchor_sha256"],
        context="manifest.speaker_anchor_sha256",
    )
    join_profile = _require_exact_keys(
        fields["join_profile"], V1_JOIN_PROFILE.keys(), context="manifest.join_profile"
    )
    if join_profile != dict(V1_JOIN_PROFILE):
        raise GlyphAssetError("manifest join profile is not the frozen v1 profile")
    source_pins = _parse_pins(fields["source_pins"], context="manifest.source_pins")
    model_pins = _parse_pins(fields["model_pins"], context="manifest.model_pins")
    evaluator_pins = _parse_pins(
        fields["evaluator_pins"], context="manifest.evaluator_pins"
    )
    raw_assets = fields["assets"]
    if not isinstance(raw_assets, list) or len(raw_assets) != ASSET_COUNT:
        raise GlyphAssetError(f"manifest must contain exactly {ASSET_COUNT} assets")
    computed_bundle_sha256 = compute_bundle_sha256(bundle_version, raw_assets)
    if computed_bundle_sha256 != bundle_sha256:
        raise GlyphAssetError("manifest bundle hash mismatch")

    root = path.parent
    assets = tuple(
        _parse_asset(
            raw_asset,
            index=index,
            expected_spec=V1_GLYPH_SPECS[index],
            root=root,
        )
        for index, raw_asset in enumerate(raw_assets)
    )

    uniqueness_fields = {
        "id": [asset.asset_id for asset in assets],
        "raw byte": [asset.raw_byte for asset in assets],
        "canonical token": [asset.canonical_token for asset in assets],
        "relative path": [asset.relative_path for asset in assets],
        "container hash": [asset.container_sha256 for asset in assets],
        "PCM hash": [asset.pcm_sha256 for asset in assets],
        "core hash": [asset.core_pcm_sha256 for asset in assets],
    }
    for label, values in uniqueness_fields.items():
        if len(set(values)) != len(values):
            raise GlyphAssetError(f"manifest contains a duplicate {label}")

    return GlyphAssetBundle(
        manifest_path=path.resolve(strict=True),
        manifest_bytes=manifest_bytes,
        manifest_sha256=manifest_sha256,
        manifest_schema=manifest_schema,
        bundle_version=bundle_version,
        grammar_sha256=grammar_sha256,
        bundle_sha256=bundle_sha256,
        speaker_anchor_sha256=speaker_anchor_sha256,
        join_profile=_immutable_json(join_profile),
        source_pins=source_pins,
        model_pins=model_pins,
        evaluator_pins=evaluator_pins,
        assets=assets,
    )


@dataclass(frozen=True)
class CarrierSegment:
    segment_id: str
    audio: np.ndarray
    selected_attempt_id: str
    selected_seed: int
    selected_text_sha256: str
    selected_endpoint_evidence_sha256: str
    sample_rate: int = SAMPLE_RATE
    speed: float = 1.0


@dataclass(frozen=True)
class AssetSegment:
    segment_id: str
    asset_id: str
    speed: float = 1.0


HybridSegment = CarrierSegment | AssetSegment


@dataclass(frozen=True)
class _PreparedSegment:
    segment_id: str
    kind: str
    pcm16_le: bytes
    source_pcm_sha256: str
    asset: GlyphAsset | None
    selected_attempt_id: str | None
    selected_seed: int | None
    selected_text_sha256: str | None
    selected_endpoint_evidence_sha256: str | None


@dataclass(frozen=True)
class LedgerSpan:
    ordinal: int
    kind: str
    segment_id: str
    start_sample: int
    end_sample: int
    pcm_sha256: str
    source_pcm_sha256: str | None
    asset_id: str | None
    raw_byte: str | None
    canonical_token: str | None
    selected_attempt_id: str | None
    selected_seed: int | None
    selected_text_sha256: str | None
    selected_endpoint_evidence_sha256: str | None
    output_core_start_sample: int | None
    output_core_end_sample: int | None
    output_core_pcm_sha256: str | None


@dataclass(frozen=True)
class BoundaryJump:
    sample_index: int
    left_kind: str
    right_kind: str
    absolute_jump: float


@dataclass(frozen=True)
class AssemblyLedger:
    schema: str
    manifest_sha256: str
    bundle_sha256: str
    plan_sha256: str
    semantic_commitment_schema: str
    external_semantic_plan_sha256: str
    inverse_proof_sha256: str
    attempt_ledger_schema: str
    ordered_attempt_ledger_sha256: str
    attempt_record_count: int
    assembled_carrier_count: int
    request_model_generator_invocations: int
    request_generated_chunk_count: int
    request_generated_text_units: int
    request_generated_chunk_budget: int
    request_generated_text_unit_budget: int
    sample_rate: int
    spans: tuple[LedgerSpan, ...]
    boundary_jumps: tuple[BoundaryJump, ...]
    tail_fade_start_sample: int
    tail_fade_end_sample: int
    final_silence_start_sample: int
    final_silence_end_sample: int
    total_samples: int
    output_pcm_sha256: str


@dataclass(frozen=True)
class HybridAssembly:
    sample_rate: int
    pcm16_le: bytes
    ledger: AssemblyLedger

    @property
    def public_pcm16_le(self) -> bytes:
        return self.pcm16_le

    @property
    def verification_pcm16_le(self) -> bytes:
        return self.pcm16_le

    def public_pcm16(self) -> np.ndarray:
        return pcm16_bytes_to_array(self.pcm16_le)

    def verification_pcm16(self) -> np.ndarray:
        return pcm16_bytes_to_array(self.pcm16_le)


def _validate_speed(speed: Any, *, context: str) -> float:
    result = _require_finite(speed, context=context)
    if result != 1.0:
        raise GlyphAssetError(f"{context} must be exactly 1.0 in v1")
    return result


def _validate_user_segment_id(value: Any, *, context: str) -> str:
    segment_id = _require_string(value, context=context)
    if (
        segment_id.startswith(_INTERNAL_SEGMENT_PREFIX)
        or segment_id in _LEGACY_RESERVED_SEGMENT_IDS
        or (
            segment_id.startswith("pause-")
            and segment_id.removeprefix("pause-").isdigit()
        )
    ):
        raise GlyphAssetError(
            f"{context} uses a reserved internal segment id/namespace"
        )
    return segment_id


def _internal_pause_id(index: int) -> str:
    return f"{_INTERNAL_SEGMENT_PREFIX}pause:{index}"


def _internal_final_silence_id() -> str:
    return f"{_INTERNAL_SEGMENT_PREFIX}final-silence"


def _carrier_selected_provenance(
    segment: CarrierSegment,
    *,
    context: str,
) -> Mapping[str, Any]:
    return {
        "selected_attempt_id": _require_string(
            segment.selected_attempt_id,
            context=f"{context}.selected_attempt_id",
        ),
        "selected_seed": _require_int(
            segment.selected_seed,
            context=f"{context}.selected_seed",
            minimum=0,
            maximum=(2**31) - 1,
        ),
        "selected_text_sha256": _require_sha256(
            segment.selected_text_sha256,
            context=f"{context}.selected_text_sha256",
        ),
        "selected_endpoint_evidence_sha256": _require_sha256(
            segment.selected_endpoint_evidence_sha256,
            context=f"{context}.selected_endpoint_evidence_sha256",
        ),
    }


def _edge_fade_float(audio: np.ndarray, samples: int) -> np.ndarray:
    result = np.asarray(audio, dtype=np.float64).copy()
    count = min(samples, result.size)
    if count <= 0:
        return result
    ramp = np.linspace(0.0, 1.0, num=count, endpoint=True, dtype=np.float64)
    result[:count] *= ramp
    result[-count:] *= ramp[::-1]
    return result


def _float_to_pcm16(audio: np.ndarray) -> bytes:
    clipped = np.clip(audio, -1.0, 32767.0 / 32768.0)
    quantized = np.rint(clipped * 32768.0).astype("<i2")
    return quantized.tobytes(order="C")


def condition_carrier_v1(segment: CarrierSegment) -> bytes:
    """Apply the sole allowed carrier conditioning before asset insertion."""

    _validate_user_segment_id(segment.segment_id, context="carrier.segment_id")
    if segment.sample_rate != SAMPLE_RATE:
        raise GlyphAssetError("carrier sample rate must be 48 kHz")
    _validate_speed(segment.speed, context="carrier.speed")
    audio = np.asarray(segment.audio)
    if audio.ndim != 1:
        raise GlyphAssetError("carrier audio must be one-dimensional mono")
    if audio.dtype.kind != "f":
        raise GlyphAssetError("carrier audio must use a floating dtype")
    if audio.size <= 0 or audio.size > MAX_CARRIER_SAMPLES:
        raise GlyphAssetError("carrier audio is empty or oversized")
    working = audio.astype(np.float64, copy=True)
    if not np.all(np.isfinite(working)):
        raise GlyphAssetError("carrier audio contains non-finite samples")
    absolute_peak = float(np.max(np.abs(working)))
    if absolute_peak == 0.0:
        raise GlyphAssetError("carrier audio must not be silent")
    # Peak-scaled RMS avoids an intermediate square overflow for finite but
    # adversarially large float inputs.
    rms = float(
        absolute_peak
        * np.sqrt(
            np.mean(np.square(working / absolute_peak), dtype=np.float64)
        )
    )
    if rms <= 0.0:
        raise GlyphAssetError("carrier audio must not be silent")
    if rms < CARRIER_RMS_FLOOR:
        gain = min(CARRIER_MAX_GAIN, CARRIER_RMS_FLOOR / rms)
        working *= gain
    peak = float(np.max(np.abs(working)))
    if peak > CARRIER_PEAK_LIMIT:
        working *= CARRIER_PEAK_LIMIT / peak
    working = _edge_fade_float(working, CARRIER_EDGE_FADE_SAMPLES)
    return _float_to_pcm16(working)


def _prepare_segments(
    bundle: GlyphAssetBundle,
    segments: Sequence[HybridSegment],
) -> tuple[_PreparedSegment, ...]:
    if not isinstance(segments, Sequence) or not segments:
        raise GlyphAssetError("hybrid assembly requires at least one segment")
    prepared: list[_PreparedSegment] = []
    segment_ids: set[str] = set()
    for index, segment in enumerate(segments):
        context = f"segments[{index}]"
        if isinstance(segment, CarrierSegment):
            segment_id = _validate_user_segment_id(
                segment.segment_id, context=f"{context}.segment_id"
            )
            selected = _carrier_selected_provenance(segment, context=context)
            pcm16_le = condition_carrier_v1(segment)
            asset = None
            kind = "carrier"
            source_sha = _sha256_bytes(pcm16_le)
        elif isinstance(segment, AssetSegment):
            segment_id = _validate_user_segment_id(
                segment.segment_id, context=f"{context}.segment_id"
            )
            _validate_speed(segment.speed, context=f"{context}.speed")
            asset = bundle.asset_by_id(
                _require_string(segment.asset_id, context=f"{context}.asset_id")
            )
            # Assets are already endpoint-conditioned, PCM16 release objects.
            # Their complete decoded byte range, not only the speech core,
            # remains immutable through every v1 join.
            pcm16_le = asset.pcm16_le
            kind = "asset"
            source_sha = asset.pcm_sha256
            selected = {
                "selected_attempt_id": None,
                "selected_seed": None,
                "selected_text_sha256": None,
                "selected_endpoint_evidence_sha256": None,
            }
        else:
            raise GlyphAssetError(f"{context} has an unsupported segment type")
        if segment_id in segment_ids:
            raise GlyphAssetError(f"duplicate segment id: {segment_id!r}")
        segment_ids.add(segment_id)
        prepared.append(
            _PreparedSegment(
                segment_id=segment_id,
                kind=kind,
                pcm16_le=pcm16_le,
                source_pcm_sha256=source_sha,
                asset=asset,
                selected_attempt_id=selected["selected_attempt_id"],
                selected_seed=selected["selected_seed"],
                selected_text_sha256=selected["selected_text_sha256"],
                selected_endpoint_evidence_sha256=selected[
                    "selected_endpoint_evidence_sha256"
                ],
            )
        )
    return tuple(prepared)


def _prepared_selected_carrier_provenance(
    prepared: Sequence[_PreparedSegment],
) -> tuple[Mapping[str, Any], ...]:
    return tuple(
        {
            "selected_attempt_id": segment.selected_attempt_id,
            "selected_seed": segment.selected_seed,
            "selected_text_sha256": segment.selected_text_sha256,
            "selected_endpoint_evidence_sha256": (
                segment.selected_endpoint_evidence_sha256
            ),
        }
        for segment in prepared
        if segment.kind == "carrier"
    )


def _plan_projection(
    bundle: GlyphAssetBundle,
    prepared: Sequence[_PreparedSegment],
    semantic_commitment: SemanticPlanCommitment,
) -> Mapping[str, Any]:
    commitment = _semantic_commitment_projection(
        semantic_commitment,
        selected_carrier_provenance=(
            _prepared_selected_carrier_provenance(prepared)
        ),
    )
    return {
        "schema": PLAN_SCHEMA,
        "manifest_sha256": bundle.manifest_sha256,
        "bundle_sha256": bundle.bundle_sha256,
        "semantic_commitment": commitment,
        "join_profile": dict(V1_JOIN_PROFILE),
        "segments": [
            {
                "ordinal": index,
                "segment_id": segment.segment_id,
                "kind": segment.kind,
                "source_pcm_sha256": segment.source_pcm_sha256,
                "asset_id": segment.asset.asset_id if segment.asset else None,
                "raw_byte": segment.asset.raw_byte if segment.asset else None,
                "canonical_token": (
                    segment.asset.canonical_token if segment.asset else None
                ),
                "selected_attempt_id": segment.selected_attempt_id,
                "selected_seed": segment.selected_seed,
                "selected_text_sha256": segment.selected_text_sha256,
                "selected_endpoint_evidence_sha256": (
                    segment.selected_endpoint_evidence_sha256
                ),
            }
            for index, segment in enumerate(prepared)
        ],
    }


def compute_hybrid_plan_sha256(
    bundle: GlyphAssetBundle,
    segments: Sequence[HybridSegment],
    *,
    semantic_commitment: SemanticPlanCommitment,
) -> str:
    prepared = _prepare_segments(bundle, segments)
    return _sha256_bytes(
        _canonical_json_bytes(
            _plan_projection(bundle, prepared, semantic_commitment)
        )
    )


def _span_hash(output: np.ndarray, start: int, end: int) -> str:
    return _sha256_bytes(
        np.asarray(output[start:end], dtype="<i2").tobytes(order="C")
    )


def _validate_ledger(ledger: AssemblyLedger, pcm16_le: bytes) -> None:
    if ledger.schema != LEDGER_SCHEMA:
        raise GlyphAssetError("ledger schema mismatch")
    if ledger.sample_rate != SAMPLE_RATE:
        raise GlyphAssetError("ledger sample rate mismatch")
    if ledger.semantic_commitment_schema != SEMANTIC_COMMITMENT_SCHEMA:
        raise GlyphAssetError("ledger semantic commitment schema mismatch")
    if ledger.attempt_ledger_schema != ATTEMPT_LEDGER_SCHEMA:
        raise GlyphAssetError("ledger attempt commitment schema mismatch")
    _require_sha256(
        ledger.external_semantic_plan_sha256,
        context="ledger.external_semantic_plan_sha256",
    )
    _require_sha256(
        ledger.inverse_proof_sha256,
        context="ledger.inverse_proof_sha256",
    )
    _require_sha256(
        ledger.ordered_attempt_ledger_sha256,
        context="ledger.ordered_attempt_ledger_sha256",
    )
    attempt_record_count = _require_int(
        ledger.attempt_record_count,
        context="ledger.attempt_record_count",
        minimum=0,
        maximum=MAX_ATTEMPT_RECORDS,
    )
    assembled_carrier_count = _require_int(
        ledger.assembled_carrier_count,
        context="ledger.assembled_carrier_count",
        minimum=0,
        maximum=MAX_GENERATED_CHUNK_BUDGET,
    )
    generator_invocations = _require_int(
        ledger.request_model_generator_invocations,
        context="ledger.request_model_generator_invocations",
        minimum=0,
        maximum=MAX_ATTEMPT_RECORDS,
    )
    generated_chunks = _require_int(
        ledger.request_generated_chunk_count,
        context="ledger.request_generated_chunk_count",
        minimum=0,
        maximum=MAX_GENERATED_CHUNK_BUDGET,
    )
    _require_int(
        ledger.request_generated_text_units,
        context="ledger.request_generated_text_units",
        minimum=0,
        maximum=MAX_GENERATED_TEXT_UNIT_BUDGET,
    )
    if (
        ledger.request_generated_chunk_budget != MAX_GENERATED_CHUNK_BUDGET
        or ledger.request_generated_text_unit_budget
        != MAX_GENERATED_TEXT_UNIT_BUDGET
    ):
        raise GlyphAssetError("ledger request budget commitment is not frozen")
    if attempt_record_count != generator_invocations:
        raise GlyphAssetError("ledger attempt/invocation totals disagree")
    if assembled_carrier_count != sum(
        span.kind == "carrier" for span in ledger.spans
    ):
        raise GlyphAssetError("ledger assembled carrier total mismatch")
    if assembled_carrier_count == 0 and (
        attempt_record_count != 0
        or generator_invocations != 0
        or generated_chunks != 0
        or ledger.request_generated_text_units != 0
    ):
        raise GlyphAssetError("asset-only ledger contains model work")
    if (
        generator_invocations < assembled_carrier_count
        or generated_chunks < assembled_carrier_count
    ):
        raise GlyphAssetError("ledger omits selected carrier generation accounting")
    if ledger.total_samples * 2 != len(pcm16_le):
        raise GlyphAssetError("ledger total does not cover output")
    cursor = 0
    segment_ids: set[str] = set()
    selected_attempt_ids: set[str] = set()
    pcm = np.frombuffer(pcm16_le, dtype="<i2")
    for ordinal, span in enumerate(ledger.spans):
        if span.ordinal != ordinal:
            raise GlyphAssetError("ledger ordinals are not monotonic")
        if span.start_sample != cursor or span.end_sample <= span.start_sample:
            raise GlyphAssetError("ledger ranges do not provide exact monotonic coverage")
        if span.end_sample > ledger.total_samples:
            raise GlyphAssetError("ledger range exceeds output")
        segment_id = _require_string(
            span.segment_id,
            context="ledger span segment_id",
        )
        if segment_id in segment_ids:
            raise GlyphAssetError("ledger segment ids are not globally unique")
        segment_ids.add(segment_id)
        if span.kind in ("pause", "final_silence"):
            if not segment_id.startswith(_INTERNAL_SEGMENT_PREFIX):
                raise GlyphAssetError("ledger internal span escaped its namespace")
            if np.any(pcm[span.start_sample : span.end_sample]):
                raise GlyphAssetError("ledger internal silence span is not exact zero")
        elif span.kind in ("asset", "carrier"):
            _validate_user_segment_id(
                segment_id,
                context="ledger user segment id",
            )
        else:
            raise GlyphAssetError("ledger span has unsupported kind")
        if _span_hash(pcm, span.start_sample, span.end_sample) != span.pcm_sha256:
            raise GlyphAssetError("ledger span PCM hash mismatch")
        if span.kind == "carrier":
            selected_attempt_id = _require_string(
                span.selected_attempt_id,
                context="ledger carrier selected_attempt_id",
            )
            if selected_attempt_id in selected_attempt_ids:
                raise GlyphAssetError(
                    "ledger carriers contain duplicate selected attempt id"
                )
            selected_attempt_ids.add(selected_attempt_id)
            _require_int(
                span.selected_seed,
                context="ledger carrier selected_seed",
                minimum=0,
                maximum=(2**31) - 1,
            )
            _require_sha256(
                span.selected_text_sha256,
                context="ledger carrier selected_text_sha256",
            )
            _require_sha256(
                span.selected_endpoint_evidence_sha256,
                context="ledger carrier selected_endpoint_evidence_sha256",
            )
        elif (
            span.selected_attempt_id is not None
            or span.selected_seed is not None
            or span.selected_text_sha256 is not None
            or span.selected_endpoint_evidence_sha256 is not None
        ):
            raise GlyphAssetError(
                "non-carrier ledger span contains selected attempt provenance"
            )
        if span.kind == "asset":
            if (
                span.output_core_start_sample is None
                or span.output_core_end_sample is None
                or span.output_core_pcm_sha256 is None
            ):
                raise GlyphAssetError("asset ledger span is missing immutable core proof")
            if not (
                span.start_sample
                <= span.output_core_start_sample
                < span.output_core_end_sample
                <= span.end_sample
            ):
                raise GlyphAssetError("asset core range is outside its ledger span")
            if (
                _span_hash(
                    pcm,
                    span.output_core_start_sample,
                    span.output_core_end_sample,
                )
                != span.output_core_pcm_sha256
            ):
                raise GlyphAssetError("asset core hash mismatch in output")
            if span.source_pcm_sha256 != span.pcm_sha256:
                raise GlyphAssetError("asset full PCM changed during assembly")
        cursor = span.end_sample
    if cursor != ledger.total_samples:
        raise GlyphAssetError("ledger ranges do not cover the complete output")
    if len(ledger.boundary_jumps) != max(0, len(ledger.spans) - 1):
        raise GlyphAssetError("ledger boundary jump count mismatch")
    for jump, left, right in zip(
        ledger.boundary_jumps,
        ledger.spans,
        ledger.spans[1:],
    ):
        expected_index = left.end_sample
        expected_jump = (
            abs(int(pcm[expected_index]) - int(pcm[expected_index - 1]))
            / 32768.0
        )
        if (
            jump.sample_index != expected_index
            or jump.left_kind != left.kind
            or jump.right_kind != right.kind
            or not math.isfinite(jump.absolute_jump)
            or jump.absolute_jump != expected_jump
        ):
            raise GlyphAssetError("ledger boundary jump evidence mismatch")
        if expected_jump > MAX_BOUNDARY_ABSOLUTE_JUMP:
            raise GlyphAssetError("ledger boundary exceeds the frozen click limit")
    if (
        ledger.final_silence_end_sample != ledger.total_samples
        or ledger.final_silence_start_sample
        != ledger.total_samples - FINAL_SILENCE_SAMPLES
        or np.any(pcm[ledger.final_silence_start_sample :])
    ):
        raise GlyphAssetError("ledger final silence is not exact zero PCM")
    if (
        ledger.tail_fade_end_sample != ledger.final_silence_start_sample
        or ledger.tail_fade_start_sample < 0
        or ledger.tail_fade_start_sample
        != ledger.tail_fade_end_sample - TAIL_FADE_SAMPLES
    ):
        raise GlyphAssetError("ledger tail fade range mismatch")
    if _sha256_bytes(pcm16_le) != ledger.output_pcm_sha256:
        raise GlyphAssetError("ledger output PCM hash mismatch")


def assemble_hybrid_pcm16(
    bundle: GlyphAssetBundle,
    segments: Sequence[HybridSegment],
    *,
    semantic_commitment: SemanticPlanCommitment,
    expected_plan_sha256: str,
    speed: float = 1.0,
    post_gain: float = 1.0,
) -> HybridAssembly:
    """Assemble a plan-bound, deterministic public/verification waveform."""

    expected_plan_sha256 = _require_sha256(
        expected_plan_sha256, context="expected_plan_sha256"
    )
    _validate_speed(speed, context="assembly.speed")
    post_gain_value = _require_finite(post_gain, context="assembly.post_gain")
    if post_gain_value != 1.0:
        raise GlyphAssetError("assembly.post_gain must be exactly 1.0 in v1")
    prepared = _prepare_segments(bundle, segments)
    commitment_projection = _semantic_commitment_projection(
        semantic_commitment,
        selected_carrier_provenance=(
            _prepared_selected_carrier_provenance(prepared)
        ),
    )
    actual_plan_sha256 = _sha256_bytes(
        _canonical_json_bytes(
            _plan_projection(bundle, prepared, semantic_commitment)
        )
    )
    if actual_plan_sha256 != expected_plan_sha256:
        raise GlyphAssetError("hybrid segment plan hash mismatch")

    parts: list[np.ndarray] = []
    provisional: list[tuple[str, str, int, int, _PreparedSegment | None]] = []
    cursor = 0
    previous: _PreparedSegment | None = None
    for index, segment in enumerate(prepared):
        # Frozen ``asset_pcm_concat_v1`` has no hidden separator between two
        # decoder-exact assets.  A pause exists only to isolate a conditioned
        # carrier edge from an immutable asset (or another carrier).
        if previous is not None and (
            previous.kind == "carrier" or segment.kind == "carrier"
        ):
            pause = np.zeros(CARRIER_BOUNDARY_PAUSE_SAMPLES, dtype="<i2")
            parts.append(pause)
            provisional.append(
                (
                    "pause",
                    _internal_pause_id(index),
                    cursor,
                    cursor + pause.size,
                    None,
                )
            )
            cursor += int(pause.size)
        samples = np.frombuffer(segment.pcm16_le, dtype="<i2").copy()
        parts.append(samples)
        provisional.append(
            (segment.kind, segment.segment_id, cursor, cursor + samples.size, segment)
        )
        cursor += int(samples.size)
        previous = segment
    body = np.concatenate(parts).astype("<i2", copy=False)
    existing_final_silence = min(
        FINAL_SILENCE_SAMPLES,
        _terminal_exact_zero_samples(body),
    )
    missing_final_silence = FINAL_SILENCE_SAMPLES - existing_final_silence
    if cursor + missing_final_silence > MAX_OUTPUT_SAMPLES:
        raise GlyphAssetError("hybrid output exceeds the frozen v1 sample budget")
    if missing_final_silence:
        top_up_start = int(body.size)
        output = np.concatenate(
            (body, np.zeros(missing_final_silence, dtype="<i2"))
        ).astype("<i2", copy=False)
        provisional.append(
            (
                "final_silence",
                _internal_final_silence_id(),
                top_up_start,
                int(output.size),
                None,
            )
        )
    else:
        output = body
    final_silence_end = int(output.size)
    final_silence_start = final_silence_end - FINAL_SILENCE_SAMPLES
    tail_fade_end = final_silence_start
    tail_fade_start = tail_fade_end - TAIL_FADE_SAMPLES
    if tail_fade_start < 0:
        raise GlyphAssetError("assembled audio is shorter than the final tail contract")
    spans: list[LedgerSpan] = []
    for ordinal, (kind, segment_id, start, end, segment) in enumerate(provisional):
        asset = segment.asset if segment is not None else None
        core_start = start + asset.core_start_sample if asset else None
        core_end = start + asset.core_end_sample if asset else None
        spans.append(
            LedgerSpan(
                ordinal=ordinal,
                kind=kind,
                segment_id=segment_id,
                start_sample=start,
                end_sample=end,
                pcm_sha256=_span_hash(output, start, end),
                source_pcm_sha256=(
                    segment.source_pcm_sha256 if segment is not None else None
                ),
                asset_id=asset.asset_id if asset else None,
                raw_byte=asset.raw_byte if asset else None,
                canonical_token=asset.canonical_token if asset else None,
                selected_attempt_id=(
                    segment.selected_attempt_id if segment is not None else None
                ),
                selected_seed=(
                    segment.selected_seed if segment is not None else None
                ),
                selected_text_sha256=(
                    segment.selected_text_sha256 if segment is not None else None
                ),
                selected_endpoint_evidence_sha256=(
                    segment.selected_endpoint_evidence_sha256
                    if segment is not None
                    else None
                ),
                output_core_start_sample=core_start,
                output_core_end_sample=core_end,
                output_core_pcm_sha256=asset.core_pcm_sha256 if asset else None,
            )
        )

    jumps: list[BoundaryJump] = []
    for left, right in zip(spans, spans[1:]):
        boundary = left.end_sample
        absolute_jump = (
            abs(int(output[boundary]) - int(output[boundary - 1])) / 32768.0
        )
        if absolute_jump > MAX_BOUNDARY_ABSOLUTE_JUMP:
            raise GlyphAssetError("hybrid boundary exceeds the frozen click limit")
        jumps.append(
            BoundaryJump(
                sample_index=boundary,
                left_kind=left.kind,
                right_kind=right.kind,
                absolute_jump=float(absolute_jump),
            )
        )
    output_bytes = output.tobytes(order="C")
    ledger = AssemblyLedger(
        schema=LEDGER_SCHEMA,
        manifest_sha256=bundle.manifest_sha256,
        bundle_sha256=bundle.bundle_sha256,
        plan_sha256=actual_plan_sha256,
        semantic_commitment_schema=SEMANTIC_COMMITMENT_SCHEMA,
        external_semantic_plan_sha256=commitment_projection[
            "external_semantic_plan_sha256"
        ],
        inverse_proof_sha256=commitment_projection["inverse_proof_sha256"],
        attempt_ledger_schema=commitment_projection["attempt_ledger_schema"],
        ordered_attempt_ledger_sha256=commitment_projection[
            "ordered_attempt_ledger_sha256"
        ],
        attempt_record_count=commitment_projection["attempt_record_count"],
        assembled_carrier_count=commitment_projection[
            "assembled_carrier_count"
        ],
        request_model_generator_invocations=commitment_projection[
            "request_model_generator_invocations"
        ],
        request_generated_chunk_count=commitment_projection[
            "request_generated_chunk_count"
        ],
        request_generated_text_units=commitment_projection[
            "request_generated_text_units"
        ],
        request_generated_chunk_budget=commitment_projection[
            "request_generated_chunk_budget"
        ],
        request_generated_text_unit_budget=commitment_projection[
            "request_generated_text_unit_budget"
        ],
        sample_rate=SAMPLE_RATE,
        spans=tuple(spans),
        boundary_jumps=tuple(jumps),
        tail_fade_start_sample=tail_fade_start,
        tail_fade_end_sample=tail_fade_end,
        final_silence_start_sample=final_silence_start,
        final_silence_end_sample=final_silence_end,
        total_samples=int(output.size),
        output_pcm_sha256=_sha256_bytes(output_bytes),
    )
    _validate_ledger(ledger, output_bytes)
    return HybridAssembly(
        sample_rate=SAMPLE_RATE,
        pcm16_le=output_bytes,
        ledger=ledger,
    )


def verify_hybrid_assembly(
    assembly: HybridAssembly,
    *,
    trusted_bundle: GlyphAssetBundle,
    original_segments: Sequence[HybridSegment],
    expected_semantic_commitment: SemanticPlanCommitment,
) -> None:
    """Rebuild and verify output from trusted inputs, never ledger assertions.

    The trusted bundle is the immutable object returned by
    :func:`load_glyph_asset_bundle`; ``original_segments`` are the caller's
    ordered carrier inputs and asset selections.  The semantic commitment is
    externally computed and opaque here, but its hashes and honest 32/800
    accounting are hard-bound into the recomputed plan and ledger.
    """

    if not isinstance(assembly, HybridAssembly):
        raise GlyphAssetError("assembly must be a HybridAssembly")
    if assembly.sample_rate != SAMPLE_RATE:
        raise GlyphAssetError("assembly sample rate mismatch")
    if assembly.public_pcm16_le is not assembly.verification_pcm16_le:
        raise GlyphAssetError("public and verification PCM are not one immutable object")
    if assembly.public_pcm16_le != assembly.verification_pcm16_le:
        raise GlyphAssetError("public and verification PCM differ")
    _validate_loaded_bundle_integrity(trusted_bundle)
    prepared = _prepare_segments(trusted_bundle, original_segments)
    commitment_projection = _semantic_commitment_projection(
        expected_semantic_commitment,
        selected_carrier_provenance=(
            _prepared_selected_carrier_provenance(prepared)
        ),
    )
    expected_plan_sha256 = _sha256_bytes(
        _canonical_json_bytes(
            _plan_projection(
                trusted_bundle,
                prepared,
                expected_semantic_commitment,
            )
        )
    )
    if assembly.ledger.manifest_sha256 != trusted_bundle.manifest_sha256:
        raise GlyphAssetError("assembly manifest evidence mismatch")
    if assembly.ledger.bundle_sha256 != trusted_bundle.bundle_sha256:
        raise GlyphAssetError("assembly bundle evidence mismatch")
    if assembly.ledger.plan_sha256 != expected_plan_sha256:
        raise GlyphAssetError("assembly plan evidence mismatch")
    ledger_commitment = {
        "schema": assembly.ledger.semantic_commitment_schema,
        "external_semantic_plan_sha256": (
            assembly.ledger.external_semantic_plan_sha256
        ),
        "inverse_proof_sha256": assembly.ledger.inverse_proof_sha256,
        "attempt_ledger_schema": assembly.ledger.attempt_ledger_schema,
        "ordered_attempt_ledger_sha256": (
            assembly.ledger.ordered_attempt_ledger_sha256
        ),
        "attempt_record_count": assembly.ledger.attempt_record_count,
        "assembled_carrier_count": assembly.ledger.assembled_carrier_count,
        "request_model_generator_invocations": (
            assembly.ledger.request_model_generator_invocations
        ),
        "request_generated_chunk_count": (
            assembly.ledger.request_generated_chunk_count
        ),
        "request_generated_text_units": (
            assembly.ledger.request_generated_text_units
        ),
        "request_generated_chunk_budget": (
            assembly.ledger.request_generated_chunk_budget
        ),
        "request_generated_text_unit_budget": (
            assembly.ledger.request_generated_text_unit_budget
        ),
    }
    if ledger_commitment != commitment_projection:
        raise GlyphAssetError("assembly semantic commitment evidence mismatch")
    _validate_ledger(assembly.ledger, assembly.pcm16_le)

    expected = assemble_hybrid_pcm16(
        trusted_bundle,
        original_segments,
        semantic_commitment=expected_semantic_commitment,
        expected_plan_sha256=expected_plan_sha256,
    )
    if assembly.pcm16_le != expected.pcm16_le:
        raise GlyphAssetError("assembly output differs from trusted reconstruction")
    if assembly.ledger != expected.ledger:
        raise GlyphAssetError("assembly ledger differs from trusted reconstruction")

    # Keep the direct bit-equality obligations explicit rather than relying on
    # digest strings or dataclass equality alone.
    output = np.frombuffer(assembly.pcm16_le, dtype="<i2")
    user_spans = tuple(
        span
        for span in assembly.ledger.spans
        if span.kind in ("carrier", "asset")
    )
    if len(user_spans) != len(prepared):
        raise GlyphAssetError("assembly user-span count mismatch")
    for expected_segment, span in zip(prepared, user_spans):
        if (
            span.segment_id != expected_segment.segment_id
            or span.kind != expected_segment.kind
        ):
            raise GlyphAssetError("assembly ordered segment identity mismatch")
        source_pcm = np.frombuffer(expected_segment.pcm16_le, dtype="<i2")
        output_slice = output[span.start_sample : span.end_sample]
        if not np.array_equal(output_slice, source_pcm):
            raise GlyphAssetError("assembly segment PCM differs from trusted source")
        if expected_segment.asset is not None:
            asset = expected_segment.asset
            if (
                span.asset_id != asset.asset_id
                or span.raw_byte != asset.raw_byte
                or span.canonical_token != asset.canonical_token
                or span.source_pcm_sha256 != asset.pcm_sha256
            ):
                raise GlyphAssetError("assembly asset label/source evidence mismatch")
            output_core = output[
                span.output_core_start_sample : span.output_core_end_sample
            ]
            trusted_core = source_pcm[
                asset.core_start_sample : asset.core_end_sample
            ]
            if (
                not np.array_equal(output_core, trusted_core)
                or _sha256_bytes(
                    np.asarray(output_core, dtype="<i2").tobytes(order="C")
                )
                != asset.core_pcm_sha256
            ):
                raise GlyphAssetError("assembly asset core differs from trusted source")
        elif (
            span.asset_id is not None
            or span.raw_byte is not None
            or span.canonical_token is not None
            or span.source_pcm_sha256
            != _sha256_bytes(expected_segment.pcm16_le)
            or span.selected_attempt_id != expected_segment.selected_attempt_id
            or span.selected_seed != expected_segment.selected_seed
            or span.selected_text_sha256
            != expected_segment.selected_text_sha256
            or span.selected_endpoint_evidence_sha256
            != expected_segment.selected_endpoint_evidence_sha256
        ):
            raise GlyphAssetError("assembly carrier evidence mismatch")
    if _sha256_bytes(assembly.pcm16_le) != expected.ledger.output_pcm_sha256:
        raise GlyphAssetError("assembly recomputed output hash mismatch")