glyph_hybrid_plan.py

"""Fail-closed request planner for the frozen ``glyph_ascii_v1`` profile.

This module is deliberately limited to text and provenance.  It does not load
audio, generate speech, join waveforms, or alter the legacy frontend.  A
request either receives one complete, reversible hybrid plan or remains on the
unchanged legacy path.

The profile scans *maximal printable-ASCII runs*.  A run is network-looking
when a case-insensitive preflight finds a network marker anywhere within it.
Every such complete run must independently satisfy the strict, native-
lowercase ``glyph_ascii_v1`` grammar.  There is no stripping, case folding,
Unicode repair, substring borrowing, or partial activation.

Canonical boundary contract
---------------------------

Every non-empty canonical payload is separated from the next payload by
exactly one ASCII space.  The same rule applies between glyph atoms.  Spaces
are represented by explicit, zero-raw/zero-audio separator segments.  Carrier
punctuation comes only from the existing zh-TW production normalizer; this
planner injects no lexical cue or punctuation.
"""

from __future__ import annotations

from collections.abc import Mapping
from dataclasses import asdict, dataclass
import hashlib
import json
import re
from types import MappingProxyType
import unicodedata

from glyph_ascii_v1 import (
    GLYPH_ASCII_V1_GRAMMAR_ID,
    GlyphAsciiV1Error,
    GlyphAsciiV1Proof,
    glyph_ascii_v1_asset_manifest_sha256,
    glyph_ascii_v1_identifier_kind,
    inverse_glyph_ascii_v1,
    render_glyph_ascii_v1,
)
from production import (
    contains_network_identifier,
    count_speech_units,
    normalize_spoken_forms,
)


GLYPH_HYBRID_PROFILE_ID = "glyph_hybrid_request_v1"
GLYPH_HYBRID_NORMALIZATION_ID = "production.normalize_spoken_forms:zh-TW"
GLYPH_HYBRID_BOUNDARY_ID = "single_ascii_space_zero_audio_v1"

# These reservations are independent from the ordinary cascade ledger.  The
# planner runs before generation, so a request cannot consume model or asset
# work and then discover that it exceeded a profile cap.
GLYPH_HYBRID_MAX_RAW_CHARS = 360
GLYPH_HYBRID_MAX_IDENTIFIERS = 8
GLYPH_HYBRID_MAX_ASSET_ATOMS = 128
GLYPH_HYBRID_MAX_ASSET_CANONICAL_UNITS = 512
GLYPH_HYBRID_ASSET_AUDIO_PLACEHOLDER_SAMPLES_PER_ATOM = 144_000
GLYPH_HYBRID_MAX_ASSET_AUDIO_PLACEHOLDER_SAMPLES = (
    GLYPH_HYBRID_MAX_ASSET_ATOMS
    * GLYPH_HYBRID_ASSET_AUDIO_PLACEHOLDER_SAMPLES_PER_ATOM
)
GLYPH_HYBRID_MAX_MODEL_GENERATED_CHUNKS = 32
GLYPH_HYBRID_MAX_MODEL_GENERATED_TEXT_UNITS = 800

_PRINTABLE_ASCII_RUN_RE = re.compile(r"[ -~]+", flags=re.ASCII)
_SHA256_RE = re.compile(r"[0-9a-f]{64}\Z", flags=re.ASCII)
_NETWORK_MARKERS = ("http://", "https://", "ftp://", "www.", "@", "://")
_ZH_CARRIER_COMPATIBILITY_PUNCTUATION = frozenset(
    "，。！？；：、（）【】「」『』《》〈〉“”‘’［］｛｝"
    "…⋯—–．｡､〔〕〖〗〘〙〚〛﹁﹂︰﹐﹑﹒﹔﹕﹖﹗"
)
_BIDI_UNSAFE = frozenset(
    {
        "BN",
        "LRE",
        "LRO",
        "RLE",
        "RLO",
        "PDF",
        "LRI",
        "RLI",
        "FSI",
        "PDI",
    }
)
_SEGMENT_KINDS = frozenset({"carrier", "asset_atom", "control", "separator"})
_AUDIO_SOURCES = frozenset({"model", "asset", "zero_audio"})


class GlyphHybridPlanError(ValueError):
    """Raised when a plan, proof, manifest, or expected request is invalid."""


@dataclass(frozen=True)
class HybridSegmentSpec:
    """One exact raw/canonical segment in assembly order.

    ``raw_text`` is retained for carrier/control provenance and for the one
    ASCII byte represented by an asset atom.  Separator segments have empty
    raw text and a zero-width raw range.
    """

    ordinal: int
    kind: str
    audio_source: str
    absolute_raw_start: int
    absolute_raw_end: int
    canonical_start: int
    canonical_end: int
    raw_text: str
    canonical_text: str
    raw_sha256: str
    canonical_sha256: str
    identifier_ordinal: int | None
    atom_ordinal: int | None
    asset_id: str | None
    manifest_entry_sha256: str | None


@dataclass(frozen=True)
class HybridReservation:
    """Pre-generation resource reservation for one complete request."""

    raw_chars: int
    identifiers: int
    asset_atoms: int
    asset_canonical_units: int
    asset_audio_placeholder_samples: int
    model_generated_chunks: int
    model_generated_text_units: int


@dataclass(frozen=True)
class HybridRenderPlan:
    """Complete request-level proof and canonical assembly plan."""

    profile_id: str
    grammar_id: str
    normalization_id: str
    boundary_contract_id: str
    raw_input_sha256: str
    canonical_target: str
    canonical_target_sha256: str
    asset_manifest_logical_sha256: str
    identifier_proofs: tuple[GlyphAsciiV1Proof, ...]
    segments: tuple[HybridSegmentSpec, ...]
    reservation: HybridReservation
    plan_sha256: str


@dataclass(frozen=True)
class _RawPiece:
    kind: str
    start: int
    end: int
    text: str


def _sha256_text(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def _require_sha256(value: object, *, field: str) -> str:
    if type(value) is not str or _SHA256_RE.fullmatch(value) is None:
        raise GlyphHybridPlanError(f"{field} must be a lowercase SHA-256 digest")
    return value


def _plan_payload(plan: HybridRenderPlan) -> dict[str, object]:
    payload = asdict(plan)
    payload.pop("plan_sha256")
    return payload


def _logical_plan_sha256(plan: HybridRenderPlan) -> str:
    payload = json.dumps(
        _plan_payload(plan),
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    ).encode("ascii")
    return hashlib.sha256(payload).hexdigest()


def glyph_hybrid_plan_sha256(plan: HybridRenderPlan) -> str:
    """Validate the plan type and return its recomputed logical digest."""

    if type(plan) is not HybridRenderPlan:
        raise GlyphHybridPlanError("hybrid plan has an invalid type")
    return _logical_plan_sha256(plan)


def _network_looking(run: str) -> bool:
    folded = run.lower()
    return any(marker in folded for marker in _NETWORK_MARKERS)


def _request_has_unsafe_unicode(raw_request: str) -> bool:
    """Reject characters that can split or visually rewrite an identifier.

    Han text and ordinary zh-TW punctuation remain eligible.  Non-ASCII Latin,
    Greek, Cyrillic, combining marks, compatibility forms that fold to
    printable ASCII, controls, and bidi formatting are intentionally outside
    this narrow profile.
    """

    for character in raw_request:
        if character.isascii():
            if unicodedata.category(character)[0] == "C":
                return True
            continue
        category = unicodedata.category(character)
        bidi = unicodedata.bidirectional(character)
        if category[0] in {"C", "M"} or bidi in _BIDI_UNSAFE:
            return True
        if character in _ZH_CARRIER_COMPATIBILITY_PUNCTUATION:
            continue
        folded = unicodedata.normalize("NFKC", character)
        if folded != character and any(
            folded_character.isascii()
            and 0x20 <= ord(folded_character) <= 0x7E
            for folded_character in folded
        ):
            return True
        unicode_name = unicodedata.name(character, "")
        if any(
            script in unicode_name
            for script in ("LATIN", "GREEK", "CYRILLIC")
        ):
            return True
    return False


def _raw_pieces(raw_request: str) -> tuple[_RawPiece, ...] | None:
    network_runs: list[re.Match[str]] = []
    for match in _PRINTABLE_ASCII_RUN_RE.finditer(raw_request):
        run = match.group(0)
        if not _network_looking(run):
            continue
        try:
            # A marker anywhere in a maximal printable-ASCII run makes that
            # complete run security-relevant.  Only the exact, lowercase v1
            # grammar may activate; never borrow a valid-looking substring.
            glyph_ascii_v1_identifier_kind(run)
        except GlyphAsciiV1Error:
            return None
        network_runs.append(match)
    if not network_runs:
        return None
    if len(network_runs) > GLYPH_HYBRID_MAX_IDENTIFIERS:
        return None

    pieces: list[_RawPiece] = []
    cursor = 0
    for match in network_runs:
        if cursor < match.start():
            pieces.append(
                _RawPiece(
                    kind="carrier",
                    start=cursor,
                    end=match.start(),
                    text=raw_request[cursor : match.start()],
                )
            )
        pieces.append(
            _RawPiece(
                kind="identifier",
                start=match.start(),
                end=match.end(),
                text=match.group(0),
            )
        )
        cursor = match.end()
    if cursor < len(raw_request):
        pieces.append(
            _RawPiece(
                kind="carrier",
                start=cursor,
                end=len(raw_request),
                text=raw_request[cursor:],
            )
        )
    return tuple(pieces)


def _carrier_canonical(raw_fragment: str) -> str | None:
    # A network form outside the strict maximal run would make activation
    # partial.  Reject both before and after legacy normalization.
    if contains_network_identifier(raw_fragment):
        return None
    try:
        canonical = normalize_spoken_forms(raw_fragment, locale="zh-TW")
    except (TypeError, ValueError):
        return None
    if (
        contains_network_identifier(canonical)
        or canonical != " ".join(canonical.split())
    ):
        return None
    return canonical


def _manifest_snapshot(
    asset_entry_sha256_by_asset_id: Mapping[str, str],
) -> tuple[Mapping[str, str], str]:
    if not isinstance(asset_entry_sha256_by_asset_id, Mapping):
        raise GlyphHybridPlanError("glyph asset manifest entries must be a mapping")
    try:
        # A caller-supplied Mapping may compute or cycle values on each
        # access. Materialize it exactly once into an owned plain dict, then
        # expose only an immutable view for every subsequent proof operation.
        owned = dict(asset_entry_sha256_by_asset_id.items())
    except Exception as error:
        raise GlyphHybridPlanError(
            "glyph asset manifest cannot be materialized"
        ) from error
    snapshot: Mapping[str, str] = MappingProxyType(owned)
    try:
        digest = glyph_ascii_v1_asset_manifest_sha256(snapshot)
    except GlyphAsciiV1Error as error:
        raise GlyphHybridPlanError("glyph asset manifest is invalid") from error
    return snapshot, digest


def _segment(
    *,
    ordinal: int,
    kind: str,
    audio_source: str,
    absolute_raw_start: int,
    absolute_raw_end: int,
    canonical_start: int,
    raw_text: str,
    canonical_text: str,
    identifier_ordinal: int | None = None,
    atom_ordinal: int | None = None,
    asset_id: str | None = None,
    manifest_entry_sha256: str | None = None,
) -> HybridSegmentSpec:
    return HybridSegmentSpec(
        ordinal=ordinal,
        kind=kind,
        audio_source=audio_source,
        absolute_raw_start=absolute_raw_start,
        absolute_raw_end=absolute_raw_end,
        canonical_start=canonical_start,
        canonical_end=canonical_start + len(canonical_text),
        raw_text=raw_text,
        canonical_text=canonical_text,
        raw_sha256=_sha256_text(raw_text),
        canonical_sha256=_sha256_text(canonical_text),
        identifier_ordinal=identifier_ordinal,
        atom_ordinal=atom_ordinal,
        asset_id=asset_id,
        manifest_entry_sha256=manifest_entry_sha256,
    )


def build_glyph_hybrid_plan(
    raw_request: str,
    *,
    asset_entry_sha256_by_asset_id: Mapping[str, str],
) -> HybridRenderPlan | None:
    """Build one complete strict plan, or return ``None`` without side effects.

    ``None`` means that the whole request must remain on the unchanged legacy
    path.  Invalid manifest configuration raises because silently falling back
    would conceal a deployment-integrity error.
    """

    if type(raw_request) is not str:
        raise GlyphHybridPlanError("raw request must be an exact string")
    manifest_snapshot, manifest_sha256 = _manifest_snapshot(
        asset_entry_sha256_by_asset_id
    )
    return _build_glyph_hybrid_plan_from_snapshot(
        raw_request,
        manifest_snapshot=manifest_snapshot,
        manifest_sha256=manifest_sha256,
    )


def _build_glyph_hybrid_plan_from_snapshot(
    raw_request: str,
    *,
    manifest_snapshot: Mapping[str, str],
    manifest_sha256: str,
) -> HybridRenderPlan | None:
    """Build using one already validated, immutable manifest snapshot."""

    if (
        not raw_request
        or len(raw_request) > GLYPH_HYBRID_MAX_RAW_CHARS
        or _request_has_unsafe_unicode(raw_request)
    ):
        return None
    pieces = _raw_pieces(raw_request)
    if pieces is None:
        return None

    segments: list[HybridSegmentSpec] = []
    identifier_proofs: list[GlyphAsciiV1Proof] = []
    canonical_parts: list[str] = []
    canonical_cursor = 0
    last_payload_raw_boundary: int | None = None
    asset_canonical_units = 0
    model_generated_chunks = 0
    model_generated_text_units = 0

    def append_separator(raw_boundary: int) -> None:
        nonlocal canonical_cursor
        separator = _segment(
            ordinal=len(segments),
            kind="separator",
            audio_source="zero_audio",
            absolute_raw_start=raw_boundary,
            absolute_raw_end=raw_boundary,
            canonical_start=canonical_cursor,
            raw_text="",
            canonical_text=" ",
        )
        segments.append(separator)
        canonical_parts.append(separator.canonical_text)
        canonical_cursor = separator.canonical_end

    def prepare_payload(raw_boundary: int) -> None:
        if canonical_parts and canonical_parts[-1] != " ":
            append_separator(raw_boundary)

    for piece in pieces:
        if piece.kind == "carrier":
            canonical = _carrier_canonical(piece.text)
            if canonical is None:
                return None
            speech_units = count_speech_units(canonical)
            if canonical:
                prepare_payload(piece.start)
            kind = "carrier" if speech_units > 0 else "control"
            audio_source = "model" if kind == "carrier" else "zero_audio"
            carrier = _segment(
                ordinal=len(segments),
                kind=kind,
                audio_source=audio_source,
                absolute_raw_start=piece.start,
                absolute_raw_end=piece.end,
                canonical_start=canonical_cursor,
                raw_text=piece.text,
                canonical_text=canonical,
            )
            segments.append(carrier)
            if canonical:
                canonical_parts.append(canonical)
                canonical_cursor = carrier.canonical_end
                last_payload_raw_boundary = piece.end
            if kind == "carrier":
                model_generated_chunks += 1
                model_generated_text_units += speech_units
            continue

        if piece.kind != "identifier":
            raise GlyphHybridPlanError("internal raw piece kind is invalid")
        try:
            # Strict eligibility is checked before adding any segment.  There
            # is no chance to render a valid prefix of an invalid run.
            if canonical_parts and canonical_parts[-1] != " ":
                append_separator(piece.start)
            canonical, proof = render_glyph_ascii_v1(
                piece.text,
                asset_entry_sha256_by_asset_id=manifest_snapshot,
                absolute_raw_start=piece.start,
                canonical_start=canonical_cursor,
            )
        except GlyphAsciiV1Error:
            return None
        if proof.asset_manifest_sha256 != manifest_sha256:
            raise GlyphHybridPlanError(
                "identifier proof manifest does not match plan header"
            )

        identifier_ordinal = len(identifier_proofs)
        identifier_proofs.append(proof)
        for atom_index, atom in enumerate(proof.atoms):
            if atom_index:
                append_separator(atom.absolute_raw_start)
            raw_character = chr(atom.raw_byte)
            asset = _segment(
                ordinal=len(segments),
                kind="asset_atom",
                audio_source="asset",
                absolute_raw_start=atom.absolute_raw_start,
                absolute_raw_end=atom.absolute_raw_end,
                canonical_start=atom.canonical_start,
                raw_text=raw_character,
                canonical_text=atom.canonical_token,
                identifier_ordinal=identifier_ordinal,
                atom_ordinal=atom.ordinal,
                asset_id=atom.asset_id,
                manifest_entry_sha256=atom.manifest_entry_sha256,
            )
            segments.append(asset)
            canonical_parts.append(asset.canonical_text)
            canonical_cursor = asset.canonical_end
            asset_canonical_units += count_speech_units(atom.canonical_token)
        if canonical != "".join(
            segment.canonical_text
            for segment in segments
            if (
                segment.identifier_ordinal == identifier_ordinal
                or (
                    segment.kind == "separator"
                    and proof.canonical_start
                    <= segment.canonical_start
                    < proof.canonical_end
                )
            )
        ):
            raise GlyphHybridPlanError("core identifier rendering was not preserved")
        last_payload_raw_boundary = piece.end

    # ``last_payload_raw_boundary`` is an internal construction assertion.  A
    # valid network request necessarily emitted at least one canonical atom.
    if last_payload_raw_boundary is None or not identifier_proofs:
        return None
    canonical_target = "".join(canonical_parts)
    asset_atoms = sum(len(proof.atoms) for proof in identifier_proofs)
    reservation = HybridReservation(
        raw_chars=len(raw_request),
        identifiers=len(identifier_proofs),
        asset_atoms=asset_atoms,
        asset_canonical_units=asset_canonical_units,
        asset_audio_placeholder_samples=(
            asset_atoms
            * GLYPH_HYBRID_ASSET_AUDIO_PLACEHOLDER_SAMPLES_PER_ATOM
        ),
        model_generated_chunks=model_generated_chunks,
        model_generated_text_units=model_generated_text_units,
    )
    if (
        reservation.identifiers > GLYPH_HYBRID_MAX_IDENTIFIERS
        or reservation.asset_atoms > GLYPH_HYBRID_MAX_ASSET_ATOMS
        or reservation.asset_canonical_units
        > GLYPH_HYBRID_MAX_ASSET_CANONICAL_UNITS
        or reservation.asset_audio_placeholder_samples
        > GLYPH_HYBRID_MAX_ASSET_AUDIO_PLACEHOLDER_SAMPLES
        or reservation.model_generated_chunks
        > GLYPH_HYBRID_MAX_MODEL_GENERATED_CHUNKS
        or reservation.model_generated_text_units
        > GLYPH_HYBRID_MAX_MODEL_GENERATED_TEXT_UNITS
    ):
        return None

    unfinished = HybridRenderPlan(
        profile_id=GLYPH_HYBRID_PROFILE_ID,
        grammar_id=GLYPH_ASCII_V1_GRAMMAR_ID,
        normalization_id=GLYPH_HYBRID_NORMALIZATION_ID,
        boundary_contract_id=GLYPH_HYBRID_BOUNDARY_ID,
        raw_input_sha256=_sha256_text(raw_request),
        canonical_target=canonical_target,
        canonical_target_sha256=_sha256_text(canonical_target),
        asset_manifest_logical_sha256=manifest_sha256,
        identifier_proofs=tuple(identifier_proofs),
        segments=tuple(segments),
        reservation=reservation,
        plan_sha256="",
    )
    return HybridRenderPlan(
        **{
            **unfinished.__dict__,
            "plan_sha256": _logical_plan_sha256(unfinished),
        }
    )


def inverse_glyph_hybrid_plan(
    plan: HybridRenderPlan,
    *,
    asset_entry_sha256_by_asset_id: Mapping[str, str],
    expected_raw_request: str,
) -> str:
    """Validate from first principles and reconstruct the exact raw request.

    The original request is mandatory external provenance.  This prevents an
    attacker from replacing a whole internally-consistent plan and merely
    recomputing its unkeyed digests.
    """

    if type(plan) is not HybridRenderPlan:
        raise GlyphHybridPlanError("hybrid plan has an invalid type")
    if type(expected_raw_request) is not str:
        raise GlyphHybridPlanError("expected raw request must be an exact string")
    manifest_snapshot, manifest_sha256 = _manifest_snapshot(
        asset_entry_sha256_by_asset_id
    )
    if (
        type(plan.identifier_proofs) is not tuple
        or type(plan.segments) is not tuple
        or type(plan.reservation) is not HybridReservation
        or plan.profile_id != GLYPH_HYBRID_PROFILE_ID
        or plan.grammar_id != GLYPH_ASCII_V1_GRAMMAR_ID
        or plan.normalization_id != GLYPH_HYBRID_NORMALIZATION_ID
        or plan.boundary_contract_id != GLYPH_HYBRID_BOUNDARY_ID
        or _require_sha256(
            plan.raw_input_sha256,
            field="plan.raw_input_sha256",
        )
        != plan.raw_input_sha256
        or _require_sha256(
            plan.canonical_target_sha256,
            field="plan.canonical_target_sha256",
        )
        != plan.canonical_target_sha256
        or _require_sha256(
            plan.asset_manifest_logical_sha256,
            field="plan.asset_manifest_logical_sha256",
        )
        != plan.asset_manifest_logical_sha256
        or _require_sha256(plan.plan_sha256, field="plan.plan_sha256")
        != plan.plan_sha256
        or plan.asset_manifest_logical_sha256 != manifest_sha256
        or plan.raw_input_sha256 != _sha256_text(expected_raw_request)
        or plan.canonical_target_sha256 != _sha256_text(plan.canonical_target)
        or plan.plan_sha256 != _logical_plan_sha256(plan)
    ):
        raise GlyphHybridPlanError("hybrid plan header is inconsistent")

    reconstructed_parts: list[str] = []
    canonical_parts: list[str] = []
    raw_cursor = 0
    canonical_cursor = 0
    asset_segments: dict[tuple[int, int], HybridSegmentSpec] = {}
    previous_nonempty_payload = False
    for ordinal, segment in enumerate(plan.segments):
        if type(segment) is not HybridSegmentSpec:
            raise GlyphHybridPlanError("hybrid segment has an invalid type")
        integer_fields = (
            segment.ordinal,
            segment.absolute_raw_start,
            segment.absolute_raw_end,
            segment.canonical_start,
            segment.canonical_end,
        )
        if any(type(value) is not int for value in integer_fields):
            raise GlyphHybridPlanError("hybrid segment integer field is invalid")
        if (
            segment.ordinal != ordinal
            or segment.kind not in _SEGMENT_KINDS
            or segment.audio_source not in _AUDIO_SOURCES
            or type(segment.raw_text) is not str
            or type(segment.canonical_text) is not str
            or segment.absolute_raw_start != raw_cursor
            or segment.absolute_raw_end < segment.absolute_raw_start
            or segment.canonical_start != canonical_cursor
            or segment.canonical_end
            != segment.canonical_start + len(segment.canonical_text)
            or _require_sha256(
                segment.raw_sha256,
                field="segment.raw_sha256",
            )
            != _sha256_text(segment.raw_text)
            or _require_sha256(
                segment.canonical_sha256,
                field="segment.canonical_sha256",
            )
            != _sha256_text(segment.canonical_text)
        ):
            raise GlyphHybridPlanError("hybrid segment is inconsistent")

        if segment.kind == "separator":
            if (
                segment.audio_source != "zero_audio"
                or segment.absolute_raw_end != segment.absolute_raw_start
                or segment.raw_text
                or segment.canonical_text != " "
                or any(
                    value is not None
                    for value in (
                        segment.identifier_ordinal,
                        segment.atom_ordinal,
                        segment.asset_id,
                        segment.manifest_entry_sha256,
                    )
                )
                or not previous_nonempty_payload
            ):
                raise GlyphHybridPlanError("separator contract is inconsistent")
            previous_nonempty_payload = False
        elif segment.kind == "asset_atom":
            if (
                segment.audio_source != "asset"
                or segment.absolute_raw_end != segment.absolute_raw_start + 1
                or len(segment.raw_text) != 1
                or not segment.raw_text.isascii()
                or not segment.canonical_text
                or type(segment.identifier_ordinal) is not int
                or type(segment.atom_ordinal) is not int
                or type(segment.asset_id) is not str
                or _require_sha256(
                    segment.manifest_entry_sha256,
                    field="segment.manifest_entry_sha256",
                )
                != segment.manifest_entry_sha256
                or previous_nonempty_payload
            ):
                raise GlyphHybridPlanError("asset segment contract is inconsistent")
            key = (segment.identifier_ordinal, segment.atom_ordinal)
            if key in asset_segments:
                raise GlyphHybridPlanError("asset segment provenance is duplicated")
            asset_segments[key] = segment
            reconstructed_parts.append(segment.raw_text)
            raw_cursor = segment.absolute_raw_end
            previous_nonempty_payload = True
        elif segment.kind in {"carrier", "control"}:
            if (
                segment.absolute_raw_end <= segment.absolute_raw_start
                or len(segment.raw_text)
                != segment.absolute_raw_end - segment.absolute_raw_start
                or any(
                    value is not None
                    for value in (
                        segment.identifier_ordinal,
                        segment.atom_ordinal,
                        segment.asset_id,
                        segment.manifest_entry_sha256,
                    )
                )
            ):
                raise GlyphHybridPlanError("carrier/control provenance is inconsistent")
            canonical = _carrier_canonical(segment.raw_text)
            speech_units = (
                count_speech_units(segment.canonical_text)
                if canonical is not None
                else -1
            )
            if (
                canonical is None
                or canonical != segment.canonical_text
                or (
                    segment.kind == "carrier"
                    and (
                        segment.audio_source != "model"
                        or speech_units <= 0
                    )
                )
                or (
                    segment.kind == "control"
                    and (
                        segment.audio_source != "zero_audio"
                        or speech_units != 0
                    )
                )
                or (segment.canonical_text and previous_nonempty_payload)
            ):
                raise GlyphHybridPlanError("carrier/control contract is inconsistent")
            reconstructed_parts.append(segment.raw_text)
            raw_cursor = segment.absolute_raw_end
            if segment.canonical_text:
                previous_nonempty_payload = True
        else:
            raise GlyphHybridPlanError("unknown hybrid segment kind")

        canonical_parts.append(segment.canonical_text)
        canonical_cursor = segment.canonical_end

    reconstructed = "".join(reconstructed_parts)
    canonical = "".join(canonical_parts)
    if (
        raw_cursor != len(reconstructed)
        or reconstructed != expected_raw_request
        or canonical != plan.canonical_target
        or canonical_cursor != len(canonical)
        or canonical.startswith(" ")
        or canonical.endswith(" ")
        or "  " in canonical
    ):
        raise GlyphHybridPlanError("hybrid segments do not provide exact coverage")

    expected_asset_keys: set[tuple[int, int]] = set()
    previous_raw_end = -1
    previous_canonical_end = -1
    for identifier_ordinal, proof in enumerate(plan.identifier_proofs):
        if type(proof) is not GlyphAsciiV1Proof:
            raise GlyphHybridPlanError("identifier proof has an invalid type")
        if (
            _require_sha256(
                proof.asset_manifest_sha256,
                field="proof.asset_manifest_sha256",
            )
            != plan.asset_manifest_logical_sha256
        ):
            raise GlyphHybridPlanError(
                "identifier proof manifest does not match plan header"
            )
        try:
            raw_identifier = inverse_glyph_ascii_v1(
                proof,
                asset_entry_sha256_by_asset_id=manifest_snapshot,
            )
        except GlyphAsciiV1Error as error:
            raise GlyphHybridPlanError("identifier proof is invalid") from error
        if (
            proof.absolute_raw_start <= previous_raw_end
            or proof.canonical_start <= previous_canonical_end
            or expected_raw_request[
                proof.absolute_raw_start : proof.absolute_raw_end
            ]
            != raw_identifier
            or plan.canonical_target[
                proof.canonical_start : proof.canonical_end
            ]
            != " ".join(atom.canonical_token for atom in proof.atoms)
        ):
            raise GlyphHybridPlanError("identifier proof range is inconsistent")
        previous_raw_end = proof.absolute_raw_end
        previous_canonical_end = proof.canonical_end
        for atom in proof.atoms:
            key = (identifier_ordinal, atom.ordinal)
            segment = asset_segments.get(key)
            if (
                segment is None
                or segment.absolute_raw_start != atom.absolute_raw_start
                or segment.absolute_raw_end != atom.absolute_raw_end
                or segment.canonical_start != atom.canonical_start
                or segment.canonical_end != atom.canonical_end
                or ord(segment.raw_text) != atom.raw_byte
                or segment.canonical_text != atom.canonical_token
                or segment.asset_id != atom.asset_id
                or segment.manifest_entry_sha256
                != atom.manifest_entry_sha256
            ):
                raise GlyphHybridPlanError("asset segment does not match its proof")
            expected_asset_keys.add(key)
    if set(asset_segments) != expected_asset_keys:
        raise GlyphHybridPlanError("asset segment set does not match identifier proofs")

    expected_plan = _build_glyph_hybrid_plan_from_snapshot(
        reconstructed,
        manifest_snapshot=manifest_snapshot,
        manifest_sha256=manifest_sha256,
    )
    if (
        expected_plan is None
        or type(expected_plan) is not type(plan)
        or expected_plan != plan
    ):
        raise GlyphHybridPlanError("hybrid plan is not the unique expected plan")
    return reconstructed