"""Content-free, independently rebuildable evidence for hybrid glyph audio. This module defines the future schema-7 evidence boundary without changing ``app`` or the legacy schema-6 runtime. Evidence JSON is intentionally not a self-authenticating report: it can only be formatted or accepted by rebuilding it from trusted renderer text, an integrity-checked glyph bundle, exact PCM, and the complete ordered runtime-call ledger. The public evidence contains hashes, scalar measurements, ranges and counts, but no request, carrier, canonical-target, ASR transcript, stable glyph id, or per-glyph dictionary hash. Small-domain commitments use a trusted, request-scoped HMAC key which is never serialized. Evaluator measurements are rerun through independently supplied, artifact-pinned callbacks. """ from __future__ import annotations from dataclasses import asdict, dataclass, is_dataclass import hashlib import hmac import json import math import re import unicodedata from typing import Any, Callable, Iterable, Mapping, Sequence, TypeAlias import numpy as np EVIDENCE_SCHEMA = "bluemagpie.glyph-hybrid.evidence.v7" EVIDENCE_VERSION = 7 PCM_SAMPLE_FORMAT = "pcm_s16le" MAX_MODEL_INVOCATIONS = 32 MAX_GENERATED_CHUNKS = 32 MAX_GENERATED_SPEECH_UNITS = 800 MAX_ASSET_COUNT = 32 MAX_ASSET_ATOMS = 128 MAX_ASSET_CANONICAL_UNITS = 512 MAX_ASSET_AUDIO_SAMPLES = 61_440_000 MAX_ASR_CALLS = 96 MAX_ECAPA_CALLS = 96 MAX_SQUIM_CALLS = 96 CONTENT_COMMITMENT_KEY_BYTES = 32 BREEZE25_MODEL_ID = "MediaTek-Research/Breeze-ASR-25" BREEZE25_REVISION = "cffe7ccb404d025296a00758d0a33468bec3a9d0" BREEZE25_WEIGHT_SHA256 = ( "c5d952b3bc03ea277209aff0ef5b5c4c055d74449ff794c02d8f4e315fdef6b6" ) BREEZE25_PRIMARY_PIN_NAME = "breeze25-primary-zh" BREEZE25_CONFIRMATION_PIN_NAME = "breeze25-confirmation-auto" _SHA256_RE = re.compile(r"[0-9a-f]{64}\Z", flags=re.ASCII) _ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/+-]{0,127}\Z", flags=re.ASCII) _SEGMENT_KINDS = frozenset(("carrier", "asset", "control", "silence")) _CALL_KINDS = frozenset( ("model_generation", "asset_load", "asr", "ecapa", "squim") ) _FINAL_SCOPES = frozenset(("final", "independent")) _GENERATION_STOP_REASONS = frozenset( ("hard_stop", "native_stop", "stop_threshold") ) _BUDGET_NAMES = ( "model_invocations", "generated_chunks", "generated_speech_units", "asset_count", "asset_atoms", "asset_canonical_units", "asset_audio_samples", "asr_calls", "ecapa_calls", "squim_calls", ) class GlyphHybridEvidenceError(ValueError): """Raised when trusted inputs or serialized evidence fail closed.""" def _canonical_json_bytes(value: Any) -> bytes: try: return json.dumps( value, ensure_ascii=True, allow_nan=False, separators=(",", ":"), sort_keys=True, ).encode("ascii") except (TypeError, ValueError) as exc: raise GlyphHybridEvidenceError( f"value cannot be encoded as canonical JSON: {exc}" ) from exc def _sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def _sha256_text(value: str) -> str: return _sha256_bytes(value.encode("utf-8")) def _content_commitment(key: bytes, domain: str, value: bytes) -> str: return hmac.new( key, domain.encode("ascii") + b"\x00" + value, hashlib.sha256, ).hexdigest() def _pcm_commitment(key: bytes, pcm16_le: bytes) -> str: """Return one request-local commitment shared by every exact PCM binding.""" return _content_commitment(key, "runtime-pcm", pcm16_le) def _require_content_commitment_key(value: Any) -> bytes: if type(value) is not bytes or len(value) != CONTENT_COMMITMENT_KEY_BYTES: raise GlyphHybridEvidenceError( "content_commitment_key must be exactly 32 trusted bytes" ) return value def _public_asset_ref(key: bytes, asset_id: str) -> str: return ( "asset-" + _content_commitment( key, "asset-id", asset_id.encode("utf-8"), )[:24] ) def _require_sha256(value: Any, *, field: str) -> str: if type(value) is not str or _SHA256_RE.fullmatch(value) is None: raise GlyphHybridEvidenceError( f"{field} must be a lowercase SHA-256 digest" ) return value def _require_id(value: Any, *, field: str) -> str: if type(value) is not str or _ID_RE.fullmatch(value) is None: raise GlyphHybridEvidenceError(f"{field} is not a bounded identifier") return value def _require_text(value: Any, *, field: str, allow_empty: bool = False) -> str: if type(value) is not str or "\x00" in value: raise GlyphHybridEvidenceError(f"{field} must be plain text") if not allow_empty and not value: raise GlyphHybridEvidenceError(f"{field} must not be empty") return value def _normalized_hf_uri(value: str) -> str: normalized = value.strip().rstrip("/") for prefix in ("hf://", "https://huggingface.co/"): if normalized.casefold().startswith(prefix): normalized = normalized[len(prefix) :] break return normalized.casefold() def _require_int( value: Any, *, field: str, minimum: int = 0, maximum: int = 2**63 - 1, ) -> int: if type(value) is not int or value < minimum or value > maximum: raise GlyphHybridEvidenceError( f"{field} must be an integer in [{minimum}, {maximum}]" ) return value def _require_float( value: Any, *, field: str, minimum: float | None = None, maximum: float | None = None, ) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise GlyphHybridEvidenceError(f"{field} must be numeric") result = float(value) if not math.isfinite(result): raise GlyphHybridEvidenceError(f"{field} must be finite") if minimum is not None and result < minimum: raise GlyphHybridEvidenceError(f"{field} must be >= {minimum}") if maximum is not None and result > maximum: raise GlyphHybridEvidenceError(f"{field} must be <= {maximum}") return result def _require_pcm16(value: Any, *, field: str) -> bytes: if type(value) is not bytes or not value or len(value) % 2: raise GlyphHybridEvidenceError( f"{field} must be non-empty little-endian PCM16 bytes" ) return value def _payload(value: Any) -> Any: if is_dataclass(value): return { field: _payload(item) for field, item in asdict(value).items() } if isinstance(value, tuple): return [_payload(item) for item in value] if isinstance(value, list): return [_payload(item) for item in value] if isinstance(value, dict): return {str(key): _payload(item) for key, item in value.items()} return value def _is_cjk(character: str) -> bool: value = ord(character) return ( 0x3400 <= value <= 0x4DBF or 0x4E00 <= value <= 0x9FFF or 0xF900 <= value <= 0xFAFF or 0x20000 <= value <= 0x3134F ) def _count_speech_units(text: str) -> int: """Count pre-normalized hybrid canonical units without runtime imports.""" units = 0 ascii_buffer: list[str] = [] def flush_ascii() -> None: nonlocal units if not ascii_buffer: return token = "".join(ascii_buffer) divisor = 2 if token.isdigit() else 4 units += max(1, math.ceil(len(token) / divisor)) ascii_buffer.clear() for character in text.casefold(): if character.isascii() and character.isalnum(): ascii_buffer.append(character) elif _is_cjk(character): flush_ascii() units += 1 else: flush_ascii() flush_ascii() return units @dataclass(frozen=True) class ArtifactPin: """Exact immutable source, model, or evaluator artifact pin.""" name: str kind: str uri: str revision: str sha256: str config_sha256: str @dataclass(frozen=True) class RendererContract: """Frozen renderer, asset, join, source, model and evaluator contract.""" grammar_id: str grammar_sha256: str profile_id: str profile_sha256: str manifest_sha256: str bundle_sha256: str join_profile_id: str join_profile_sha256: str gate_policy_sha256: str speaker_anchor_sha256: str generation_policy_ids: tuple[str, ...] generation_policy_sha256s: tuple[str, ...] source_pins: tuple[ArtifactPin, ...] model_pins: tuple[ArtifactPin, ...] evaluator_pins: tuple[ArtifactPin, ...] @dataclass(frozen=True) class TrustedAsset: """One integrity-checked asset supplied to the independent rebuilder.""" asset_id: str raw_text: str canonical_text: str manifest_entry_sha256: str pcm16_le: bytes @dataclass(frozen=True) class TrustedGlyphBundle: """Minimal trusted bundle view required by schema 7.""" manifest_sha256: str bundle_sha256: str assets: tuple[TrustedAsset, ...] def compute_trusted_asset_entry_sha256(asset: TrustedAsset) -> str: """Recompute one manifest entry from its exact metadata and PCM bytes.""" if type(asset) is not TrustedAsset: raise GlyphHybridEvidenceError("asset must be TrustedAsset") asset_id = _require_id(asset.asset_id, field="asset.asset_id") raw_text = _require_text(asset.raw_text, field="asset.raw_text") canonical_text = _require_text( asset.canonical_text, field="asset.canonical_text", ) pcm = _require_pcm16(asset.pcm16_le, field="asset.pcm16_le") return _sha256_bytes( _canonical_json_bytes( { "asset_id": asset_id, "raw_text": raw_text, "canonical_text": canonical_text, "pcm_sha256": _sha256_bytes(pcm), "pcm_sample_count": len(pcm) // 2, } ) ) def compute_trusted_bundle_digests( assets: Sequence[TrustedAsset], ) -> tuple[str, str]: """Recompute logical manifest and bundle digests from exact asset bytes.""" if type(assets) not in {tuple, list} or not assets: raise GlyphHybridEvidenceError("assets must be a non-empty sequence") entries: list[dict[str, Any]] = [] seen_ids: set[str] = set() for asset in assets: if type(asset) is not TrustedAsset: raise GlyphHybridEvidenceError("assets must contain TrustedAsset values") if asset.asset_id in seen_ids: raise GlyphHybridEvidenceError("asset ids must be unique") seen_ids.add(asset.asset_id) entry_sha256 = compute_trusted_asset_entry_sha256(asset) if asset.manifest_entry_sha256 != entry_sha256: raise GlyphHybridEvidenceError( "asset manifest entry digest does not match trusted bytes" ) entries.append( { "asset_id": asset.asset_id, "entry_sha256": entry_sha256, } ) entries.sort(key=lambda item: item["asset_id"]) manifest_sha256 = _sha256_bytes( _canonical_json_bytes( { "schema": "bluemagpie.glyph-hybrid.manifest.v1", "entries": entries, } ) ) bundle_sha256 = _sha256_bytes( _canonical_json_bytes( { "schema": "bluemagpie.glyph-hybrid.bundle.v1", "manifest_sha256": manifest_sha256, "assets": [ { "asset_id": asset.asset_id, "pcm_sha256": _sha256_bytes(asset.pcm16_le), "pcm_sample_count": len(asset.pcm16_le) // 2, } for asset in sorted(assets, key=lambda item: item.asset_id) ], } ) ) return manifest_sha256, bundle_sha256 @dataclass(frozen=True) class TrustedRenderSegment: """Text-bearing renderer segment; text is never emitted in evidence.""" segment_id: str kind: str raw_start: int raw_end: int canonical_start: int canonical_end: int raw_text: str canonical_text: str output_start_sample: int output_end_sample: int asset_id: str | None = None @dataclass(frozen=True) class TrustedRendererPlan: """Complete renderer plan supplied outside the evidence JSON.""" grammar_id: str grammar_sha256: str profile_id: str profile_sha256: str join_profile_id: str join_profile_sha256: str raw_input: str canonical_target: str segments: tuple[TrustedRenderSegment, ...] @dataclass(frozen=True) class EndpointEvidence: """Measured endpoint evidence for a model generation call.""" stop_reason: str tail_energy_ratio: float terminal_silence_samples: int generated_steps: int hard_cap_steps: int @dataclass(frozen=True) class TrustedModelGenerationCall: call_id: str segment_id: str attempt_ordinal: int text: str seed: int policy_id: str policy_sha256: str model_pin_name: str output_pcm16_le: bytes endpoint: EndpointEvidence @dataclass(frozen=True) class TrustedAssetLoadCall: call_id: str segment_id: str asset_id: str manifest_entry_sha256: str loaded_pcm16_le: bytes @dataclass(frozen=True) class TrustedAsrCall: call_id: str scope: str evaluator_pin_name: str input_pcm16_le: bytes target_text: str transcript_text: str subject_model_call_id: str | None = None @dataclass(frozen=True) class TrustedEcapaCall: call_id: str scope: str evaluator_pin_name: str input_pcm16_le: bytes speaker_anchor_sha256: str centroid_similarity: float begin_similarity: float end_similarity: float subject_model_call_id: str | None = None @dataclass(frozen=True) class TrustedSquimCall: call_id: str scope: str evaluator_pin_name: str input_pcm16_le: bytes stoi: float pesq: float si_sdr: float subject_model_call_id: str | None = None TrustedRuntimeCall: TypeAlias = ( TrustedModelGenerationCall | TrustedAssetLoadCall | TrustedAsrCall | TrustedEcapaCall | TrustedSquimCall ) AsrEvaluator: TypeAlias = Callable[[ArtifactPin, bytes, int], str] EcapaEvaluator: TypeAlias = Callable[ [ArtifactPin, bytes, int, str], tuple[float, float, float], ] SquimEvaluator: TypeAlias = Callable[ [ArtifactPin, bytes, int], tuple[float, float, float], ] GenerationEvaluator: TypeAlias = Callable[ [ArtifactPin, str, int, str, str, int], tuple[bytes, EndpointEvidence], ] RendererEvaluator: TypeAlias = Callable[ [RendererContract, TrustedGlyphBundle, str], TrustedRendererPlan, ] @dataclass(frozen=True) class TrustedCarrierSelection: """Every attempted carrier call and the one exact selected call.""" segment_id: str path_rank: int attempted_call_ids: tuple[str, ...] selected_call_id: str @dataclass(frozen=True) class TrustedGateBinding: """Bind a whole-output gate to three exact evaluator calls.""" scope: str asr_call_id: str ecapa_call_id: str squim_call_id: str active_speech_samples: int @dataclass(frozen=True) class BudgetCaps: model_invocations: int = MAX_MODEL_INVOCATIONS generated_chunks: int = MAX_GENERATED_CHUNKS generated_speech_units: int = MAX_GENERATED_SPEECH_UNITS asset_count: int = MAX_ASSET_COUNT asset_atoms: int = MAX_ASSET_ATOMS asset_canonical_units: int = MAX_ASSET_CANONICAL_UNITS asset_audio_samples: int = MAX_ASSET_AUDIO_SAMPLES asr_calls: int = MAX_ASR_CALLS ecapa_calls: int = MAX_ECAPA_CALLS squim_calls: int = MAX_SQUIM_CALLS @dataclass(frozen=True) class GatePolicy: """Numeric current-gate policy; no caller-supplied pass boolean exists.""" sample_rate: int = 48_000 max_cer: float = 0.08 max_prefix_cer: float = 0.50 max_suffix_cer: float = 0.50 max_prefix_deletions: int = 0 max_suffix_deletions: int = 0 max_extra_tail_units: int = 0 max_pace_cps: float = 4.30 min_speaker_similarity: float = 0.105 max_boundary_speaker_drop: float = 0.095 min_squim_stoi: float = 0.72 min_squim_pesq: float = 1.20 min_squim_si_sdr: float = -20.0 min_pcm_rms: float = 0.005 max_pcm_peak: float = 1.0 max_endpoint_tail_energy_ratio: float = 0.02 active_voice_top_db: float = 35.0 active_voice_frame_ms: float = 25.0 active_voice_hop_ms: float = 10.0 active_voice_min_rms: float = 1.0e-4 endpoint_tail_window_ms: float = 5.0 endpoint_silence_amplitude: float = 1.0e-4 @dataclass(frozen=True) class EvidenceRebuildInputs: """All trusted material required to independently rebuild schema 7.""" contract: RendererContract plan: TrustedRendererPlan bundle: TrustedGlyphBundle assembled_pcm16_le: bytes public_pcm16_le: bytes calls: tuple[TrustedRuntimeCall, ...] carrier_selections: tuple[TrustedCarrierSelection, ...] gate_bindings: tuple[TrustedGateBinding, ...] content_commitment_key: bytes generation_evaluator: GenerationEvaluator renderer_evaluator: RendererEvaluator asr_evaluator: AsrEvaluator ecapa_evaluator: EcapaEvaluator squim_evaluator: SquimEvaluator budget_caps: BudgetCaps = BudgetCaps() gate_policy: GatePolicy = GatePolicy() @dataclass(frozen=True) class RenderSegmentEvidence: ordinal: int segment_id: str kind: str raw_start: int raw_end: int canonical_start: int canonical_end: int raw_fragment_sha256: str canonical_fragment_sha256: str carrier_text_sha256: str | None asset_id: str | None asset_manifest_entry_sha256: str | None output_start_sample: int output_end_sample: int output_sample_count: int realized_pcm_sha256: str @dataclass(frozen=True) class HardBudgetEvidence: name: str actual: int cap: int @dataclass(frozen=True) class ModelGenerationCallEvidence: ordinal: int call_id: str kind: str segment_id: str attempt_ordinal: int text_sha256: str generated_speech_units: int generated_chunks: int seed: int policy_id: str policy_sha256: str model_pin_name: str output_pcm_sha256: str output_sample_count: int stop_reason: str tail_energy_ratio: float terminal_silence_samples: int generated_steps: int hard_cap_steps: int @dataclass(frozen=True) class AssetLoadCallEvidence: ordinal: int call_id: str kind: str segment_id: str asset_id: str manifest_entry_sha256: str loaded_pcm_sha256: str loaded_sample_count: int @dataclass(frozen=True) class AsrCallEvidence: ordinal: int call_id: str kind: str scope: str evaluator_pin_name: str input_pcm_sha256: str input_sample_count: int target_sha256: str transcript_sha256: str target_units: int hypothesis_units: int edit_distance: int cer: float prefix_cer: float suffix_cer: float prefix_deletions: int suffix_deletions: int extra_tail_units: int semantic_available: bool network_protected_spans: int network_protected_spans_passed: bool subject_model_call_id: str | None @dataclass(frozen=True) class EcapaCallEvidence: ordinal: int call_id: str kind: str scope: str evaluator_pin_name: str input_pcm_sha256: str input_sample_count: int speaker_anchor_sha256: str centroid_similarity: float begin_similarity: float end_similarity: float measurement_sha256: str subject_model_call_id: str | None @dataclass(frozen=True) class SquimCallEvidence: ordinal: int call_id: str kind: str scope: str evaluator_pin_name: str input_pcm_sha256: str input_sample_count: int stoi: float pesq: float si_sdr: float measurement_sha256: str subject_model_call_id: str | None RuntimeCallEvidence: TypeAlias = ( ModelGenerationCallEvidence | AssetLoadCallEvidence | AsrCallEvidence | EcapaCallEvidence | SquimCallEvidence ) @dataclass(frozen=True) class CarrierSelectionEvidence: segment_id: str path_rank: int attempted_call_ids: tuple[str, ...] selected_call_id: str selection_sha256: str @dataclass(frozen=True) class GateProjection: scope: str policy_sha256: str pcm_sha256: str asr_call_id: str ecapa_call_id: str squim_call_id: str target_units: int hypothesis_units: int edit_distance: int cer: float prefix_cer: float suffix_cer: float prefix_deletions: int suffix_deletions: int extra_tail_units: int semantic_available: bool network_protected_spans: int network_protected_spans_passed: bool pace_cps: float speaker_similarity: float boundary_speaker_drop: float squim_stoi: float squim_pesq: float squim_si_sdr: float pcm_rms: float pcm_peak: float endpoint_tail_energy_ratio: float endpoint_evidence_complete: bool passed: bool rejection_reasons: tuple[str, ...] @dataclass(frozen=True) class AdditionalStrictGates: strict_inverse_reconstruction: bool asset_reconstruction: bool public_pcm_equality: bool @dataclass(frozen=True) class RequestProof: content_commitment_scheme: str raw_input_sha256: str raw_input_codepoints: int raw_input_utf8_bytes: int canonical_target_sha256: str canonical_target_codepoints: int canonical_target_speech_units: int semantic_plan_sha256: str inverse_proof_sha256: str realized_audio_plan_sha256: str assembled_pcm_sha256: str public_pcm_sha256: str output_sample_rate: int output_channels: int output_sample_format: str output_sample_count: int output_byte_count: int segments: tuple[RenderSegmentEvidence, ...] budgets: tuple[HardBudgetEvidence, ...] ledger_sha256: str selected_carriers_sha256: str gate_policy_sha256: str final_gate: GateProjection independent_gate: GateProjection additional_strict_gates: AdditionalStrictGates @dataclass(frozen=True) class HybridEvidence: schema: str version: int renderer_contract: RendererContract renderer_contract_sha256: str request_proof: RequestProof runtime_call_ledger: tuple[RuntimeCallEvidence, ...] carrier_selections: tuple[CarrierSelectionEvidence, ...] evidence_sha256: str def _validate_pin(pin: ArtifactPin, *, field: str) -> None: if type(pin) is not ArtifactPin: raise GlyphHybridEvidenceError(f"{field} must contain ArtifactPin values") _require_id(pin.name, field=f"{field}.name") _require_id(pin.kind, field=f"{field}.kind") _require_text(pin.uri, field=f"{field}.uri") _require_text(pin.revision, field=f"{field}.revision") _require_sha256(pin.sha256, field=f"{field}.sha256") _require_sha256(pin.config_sha256, field=f"{field}.config_sha256") def _validate_contract( contract: RendererContract, plan: TrustedRendererPlan, bundle: TrustedGlyphBundle, gate_policy_sha256: str, ) -> None: if type(contract) is not RendererContract: raise GlyphHybridEvidenceError("contract must be RendererContract") for field, value in ( ("grammar_id", contract.grammar_id), ("profile_id", contract.profile_id), ("join_profile_id", contract.join_profile_id), ): _require_id(value, field=f"contract.{field}") for field, value in ( ("grammar_sha256", contract.grammar_sha256), ("profile_sha256", contract.profile_sha256), ("manifest_sha256", contract.manifest_sha256), ("bundle_sha256", contract.bundle_sha256), ("join_profile_sha256", contract.join_profile_sha256), ("gate_policy_sha256", contract.gate_policy_sha256), ("speaker_anchor_sha256", contract.speaker_anchor_sha256), ): _require_sha256(value, field=f"contract.{field}") if ( contract.grammar_id != plan.grammar_id or contract.grammar_sha256 != plan.grammar_sha256 or contract.profile_id != plan.profile_id or contract.profile_sha256 != plan.profile_sha256 or contract.join_profile_id != plan.join_profile_id or contract.join_profile_sha256 != plan.join_profile_sha256 ): raise GlyphHybridEvidenceError("renderer plan does not match contract pins") if ( contract.manifest_sha256 != bundle.manifest_sha256 or contract.bundle_sha256 != bundle.bundle_sha256 ): raise GlyphHybridEvidenceError("trusted bundle does not match contract pins") if contract.gate_policy_sha256 != gate_policy_sha256: raise GlyphHybridEvidenceError("gate policy does not match contract pin") if ( type(contract.generation_policy_ids) is not tuple or type(contract.generation_policy_sha256s) is not tuple or len(contract.generation_policy_ids) != len(contract.generation_policy_sha256s) ): raise GlyphHybridEvidenceError( "contract generation policy pins are malformed" ) generation_policy_ids = tuple( _require_id( policy_id, field=f"contract.generation_policy_ids[{index}]", ) for index, policy_id in enumerate( contract.generation_policy_ids ) ) generation_policy_digests = tuple( _require_sha256( digest, field=f"contract.generation_policy_sha256s[{index}]", ) for index, digest in enumerate( contract.generation_policy_sha256s ) ) if ( len(set(generation_policy_ids)) != len(generation_policy_ids) or len(set(generation_policy_digests)) != len(generation_policy_digests) ): raise GlyphHybridEvidenceError( "generation policy pins must be unique" ) pin_names: set[str] = set() for group_name, pins in ( ("source_pins", contract.source_pins), ("model_pins", contract.model_pins), ("evaluator_pins", contract.evaluator_pins), ): if not pins: raise GlyphHybridEvidenceError(f"contract.{group_name} must not be empty") for index, pin in enumerate(pins): _validate_pin(pin, field=f"contract.{group_name}[{index}]") if group_name == "source_pins" and pin.kind != "source": raise GlyphHybridEvidenceError("source pin kind must be source") if group_name == "model_pins" and pin.kind != "model": raise GlyphHybridEvidenceError("model pin kind must be model") if ( group_name == "evaluator_pins" and pin.kind not in {"asr", "ecapa", "squim"} ): raise GlyphHybridEvidenceError( "evaluator pin kind must be asr, ecapa or squim" ) if pin.name in pin_names: raise GlyphHybridEvidenceError("artifact pin names must be unique") pin_names.add(pin.name) return None def _content_free_contract( contract: RendererContract, content_commitment_key: bytes, ) -> RendererContract: def redact(pin: ArtifactPin) -> ArtifactPin: return ArtifactPin( name=pin.name, kind=pin.kind, uri=( "hmac-sha256:" + _content_commitment( content_commitment_key, "artifact-uri", pin.uri.encode("utf-8"), ) ), revision=( "hmac-sha256:" + _content_commitment( content_commitment_key, "artifact-revision", pin.revision.encode("utf-8"), ) ), sha256=pin.sha256, config_sha256=pin.config_sha256, ) return RendererContract( grammar_id=contract.grammar_id, grammar_sha256=contract.grammar_sha256, profile_id=contract.profile_id, profile_sha256=contract.profile_sha256, manifest_sha256=contract.manifest_sha256, bundle_sha256=contract.bundle_sha256, join_profile_id=contract.join_profile_id, join_profile_sha256=contract.join_profile_sha256, gate_policy_sha256=contract.gate_policy_sha256, speaker_anchor_sha256=contract.speaker_anchor_sha256, generation_policy_ids=contract.generation_policy_ids, generation_policy_sha256s=contract.generation_policy_sha256s, source_pins=tuple(redact(pin) for pin in contract.source_pins), model_pins=tuple(redact(pin) for pin in contract.model_pins), evaluator_pins=tuple( redact(pin) for pin in contract.evaluator_pins ), ) def _validate_bundle(bundle: TrustedGlyphBundle) -> dict[str, TrustedAsset]: if type(bundle) is not TrustedGlyphBundle: raise GlyphHybridEvidenceError("bundle must be TrustedGlyphBundle") _require_sha256(bundle.manifest_sha256, field="bundle.manifest_sha256") _require_sha256(bundle.bundle_sha256, field="bundle.bundle_sha256") if not bundle.assets: raise GlyphHybridEvidenceError("bundle must contain assets") if len(bundle.assets) > MAX_ASSET_COUNT: raise GlyphHybridEvidenceError("bundle asset count exceeds hard maximum") result: dict[str, TrustedAsset] = {} raw_values: set[str] = set() canonical_units = 0 audio_samples = 0 for index, asset in enumerate(bundle.assets): field = f"bundle.assets[{index}]" if type(asset) is not TrustedAsset: raise GlyphHybridEvidenceError(f"{field} must be TrustedAsset") _require_id(asset.asset_id, field=f"{field}.asset_id") _require_text(asset.raw_text, field=f"{field}.raw_text") _require_text(asset.canonical_text, field=f"{field}.canonical_text") _require_sha256( asset.manifest_entry_sha256, field=f"{field}.manifest_entry_sha256", ) _require_pcm16(asset.pcm16_le, field=f"{field}.pcm16_le") if asset.manifest_entry_sha256 != compute_trusted_asset_entry_sha256( asset ): raise GlyphHybridEvidenceError( "asset manifest entry digest does not match trusted bytes" ) if asset.asset_id in result: raise GlyphHybridEvidenceError("asset ids must be unique") if asset.raw_text in raw_values: raise GlyphHybridEvidenceError("asset raw values must be unique") result[asset.asset_id] = asset raw_values.add(asset.raw_text) canonical_units += _count_speech_units(asset.canonical_text) audio_samples += len(asset.pcm16_le) // 2 if canonical_units > MAX_ASSET_CANONICAL_UNITS: raise GlyphHybridEvidenceError( "bundle canonical units exceed hard maximum" ) if audio_samples > MAX_ASSET_AUDIO_SAMPLES: raise GlyphHybridEvidenceError( "bundle audio samples exceed hard maximum" ) manifest_sha256, bundle_sha256 = compute_trusted_bundle_digests( tuple(result.values()) ) if ( bundle.manifest_sha256 != manifest_sha256 or bundle.bundle_sha256 != bundle_sha256 ): raise GlyphHybridEvidenceError( "trusted bundle digests do not match actual asset bytes" ) return result def _segment_semantic_projection( ordinal: int, segment: TrustedRenderSegment, ) -> dict[str, Any]: return { "ordinal": ordinal, "segment_id": segment.segment_id, "kind": segment.kind, "raw_start": segment.raw_start, "raw_end": segment.raw_end, "canonical_start": segment.canonical_start, "canonical_end": segment.canonical_end, "raw_text": segment.raw_text, "canonical_text": segment.canonical_text, "asset_id": segment.asset_id, } def _renderer_semantic_projection(plan: TrustedRendererPlan) -> dict[str, Any]: return { "grammar_id": plan.grammar_id, "grammar_sha256": plan.grammar_sha256, "profile_id": plan.profile_id, "profile_sha256": plan.profile_sha256, "join_profile_id": plan.join_profile_id, "join_profile_sha256": plan.join_profile_sha256, "raw_input": plan.raw_input, "canonical_target": plan.canonical_target, "segments": [ { "ordinal": ordinal, "kind": segment.kind, "raw_start": segment.raw_start, "raw_end": segment.raw_end, "canonical_start": segment.canonical_start, "canonical_end": segment.canonical_end, "raw_text": segment.raw_text, "canonical_text": segment.canonical_text, "asset_id": segment.asset_id, } for ordinal, segment in enumerate( segment for segment in plan.segments if segment.kind != "silence" ) ], } def _validate_renderer_rebuild(inputs: EvidenceRebuildInputs) -> None: """Recompute semantic segmentation through the pinned renderer boundary. Segment identifiers and realized PCM ranges are deliberately excluded: identifiers are runtime-local labels, while exact ranges are separately bound to selected model outputs and asset bytes. """ try: rebuilt = inputs.renderer_evaluator( inputs.contract, inputs.bundle, inputs.plan.raw_input, ) except Exception as exc: raise GlyphHybridEvidenceError( "pinned renderer failed during rebuild" ) from exc if type(rebuilt) is not TrustedRendererPlan: raise GlyphHybridEvidenceError( "pinned renderer returned an invalid semantic plan" ) if _renderer_semantic_projection(rebuilt) != _renderer_semantic_projection( inputs.plan ): raise GlyphHybridEvidenceError( "renderer plan does not match pinned renderer rebuild" ) def _build_segments( plan: TrustedRendererPlan, asset_by_id: Mapping[str, TrustedAsset], output_pcm: bytes, content_commitment_key: bytes, ) -> tuple[ tuple[RenderSegmentEvidence, ...], str, str, str, bool, bool, ]: if type(plan) is not TrustedRendererPlan: raise GlyphHybridEvidenceError("plan must be TrustedRendererPlan") raw = _require_text(plan.raw_input, field="plan.raw_input") canonical = _require_text(plan.canonical_target, field="plan.canonical_target") if not plan.segments: raise GlyphHybridEvidenceError("plan must contain segments") raw_cursor = 0 canonical_cursor = 0 output_cursor = 0 reconstructed_raw: list[str] = [] reconstructed_canonical: list[str] = [] evidence: list[RenderSegmentEvidence] = [] semantic_projection: list[dict[str, Any]] = [] realized_projection: list[dict[str, Any]] = [] asset_reconstruction = True seen_ids: set[str] = set() for ordinal, segment in enumerate(plan.segments): field = f"plan.segments[{ordinal}]" if type(segment) is not TrustedRenderSegment: raise GlyphHybridEvidenceError(f"{field} has an invalid type") _require_id(segment.segment_id, field=f"{field}.segment_id") if segment.segment_id in seen_ids: raise GlyphHybridEvidenceError("segment ids must be unique") seen_ids.add(segment.segment_id) if segment.kind not in _SEGMENT_KINDS: raise GlyphHybridEvidenceError(f"{field}.kind is invalid") raw_text = _require_text( segment.raw_text, field=f"{field}.raw_text", allow_empty=True, ) canonical_text = _require_text( segment.canonical_text, field=f"{field}.canonical_text", allow_empty=True, ) raw_start = _require_int(segment.raw_start, field=f"{field}.raw_start") raw_end = _require_int(segment.raw_end, field=f"{field}.raw_end") canonical_start = _require_int( segment.canonical_start, field=f"{field}.canonical_start", ) canonical_end = _require_int( segment.canonical_end, field=f"{field}.canonical_end", ) output_start = _require_int( segment.output_start_sample, field=f"{field}.output_start_sample", ) output_end = _require_int( segment.output_end_sample, field=f"{field}.output_end_sample", ) if raw_start != raw_cursor or raw_end < raw_start: raise GlyphHybridEvidenceError("raw ranges must be ordered and contiguous") if canonical_start != canonical_cursor or canonical_end < canonical_start: raise GlyphHybridEvidenceError( "canonical ranges must be ordered and contiguous" ) if output_start != output_cursor or output_end < output_start: raise GlyphHybridEvidenceError( "realized PCM ranges must be ordered and contiguous" ) if raw_end - raw_start != len(raw_text): raise GlyphHybridEvidenceError("raw range length does not match text") if canonical_end - canonical_start != len(canonical_text): raise GlyphHybridEvidenceError( "canonical range length does not match text" ) if raw[raw_start:raw_end] != raw_text: raise GlyphHybridEvidenceError("segment does not match trusted raw input") if canonical[canonical_start:canonical_end] != canonical_text: raise GlyphHybridEvidenceError( "segment does not match trusted canonical target" ) pcm_start = output_start * 2 pcm_end = output_end * 2 if pcm_end > len(output_pcm): raise GlyphHybridEvidenceError("segment PCM range exceeds output") span_pcm = output_pcm[pcm_start:pcm_end] asset_id: str | None = None asset_entry_sha256: str | None = None carrier_text_sha256: str | None = None if segment.kind == "asset": if segment.asset_id is None: raise GlyphHybridEvidenceError("asset segment is missing asset id") trusted_asset_id = _require_id( segment.asset_id, field=f"{field}.asset_id", ) asset = asset_by_id.get(trusted_asset_id) if asset is None: raise GlyphHybridEvidenceError("asset segment references unknown asset") if raw_text != asset.raw_text or canonical_text != asset.canonical_text: raise GlyphHybridEvidenceError( "asset text does not exactly reconstruct the trusted asset" ) if span_pcm != asset.pcm16_le: asset_reconstruction = False asset_id = _public_asset_ref( content_commitment_key, trusted_asset_id, ) asset_entry_sha256 = _content_commitment( content_commitment_key, "asset-manifest-entry", asset.manifest_entry_sha256.encode("ascii"), ) else: if segment.asset_id is not None: raise GlyphHybridEvidenceError( "non-asset segment must not contain asset id" ) if segment.kind == "carrier": if not raw_text or not canonical_text or not span_pcm: raise GlyphHybridEvidenceError( "carrier must have raw, canonical and realized PCM content" ) carrier_text_sha256 = "present" elif segment.kind == "silence": if ( raw_text or canonical_text or output_start == output_end or not span_pcm or np.any(np.frombuffer(span_pcm, dtype=" tuple[str, ...]: units: list[str] = [] for character in unicodedata.normalize("NFKC", text).casefold(): category = unicodedata.category(character) if category[0] in {"L", "N"}: units.append(character) return tuple(units) def _asr_metrics( target: str, transcript: str, ) -> dict[str, int | float | bool]: # Import lazily to keep schema construction lightweight while using the # exact production normalization, Mandarin endpoint units and deterministic # alignment contract for semantic completion evidence. from production import compare_asr_text comparison = compare_asr_text( target, transcript, locale="zh-TW", prefix_units=6, suffix_units=6, max_cer=1.0e9, max_prefix_cer=1.0e9, max_suffix_cer=1.0e9, max_extra_tail_units=2**31 - 1, ) return { "target_units": len(comparison.target_text), "hypothesis_units": len(comparison.transcript_text), "edit_distance": comparison.edit_distance, "cer": comparison.cer, "prefix_cer": comparison.prefix_cer, "suffix_cer": comparison.suffix_cer, "prefix_deletions": comparison.prefix_deletions, "suffix_deletions": comparison.suffix_deletions, "extra_tail_units": comparison.extra_tail_units, "semantic_available": comparison.passed, "network_protected_spans": comparison.network_protected_spans, "network_protected_spans_passed": ( comparison.network_protected_spans_passed ), } def _measurement_sha256(payload: Mapping[str, Any]) -> str: return _sha256_bytes(_canonical_json_bytes(dict(payload))) def _build_calls( inputs: EvidenceRebuildInputs, segment_by_id: Mapping[str, TrustedRenderSegment], asset_by_id: Mapping[str, TrustedAsset], public_segment_id_by_trusted_id: Mapping[str, str], content_commitment_key: bytes, ) -> tuple[ tuple[RuntimeCallEvidence, ...], dict[str, RuntimeCallEvidence], dict[str, TrustedRuntimeCall], ]: model_pin_by_name = { pin.name: pin for pin in inputs.contract.model_pins } generation_policies = set( zip( inputs.contract.generation_policy_ids, inputs.contract.generation_policy_sha256s, strict=True, ) ) evaluator_pin_by_name = { pin.name: pin for pin in inputs.contract.evaluator_pins } trusted_by_id: dict[str, TrustedRuntimeCall] = {} evidence_by_id: dict[str, RuntimeCallEvidence] = {} evidence: list[RuntimeCallEvidence] = [] public_call_id_by_trusted_id: dict[str, str] = {} for ordinal, call in enumerate(inputs.calls): call_id = _require_id( getattr(call, "call_id", None), field=f"calls[{ordinal}].call_id", ) if call_id in public_call_id_by_trusted_id: raise GlyphHybridEvidenceError("runtime call ids must be unique") public_call_id_by_trusted_id[call_id] = f"call-{ordinal:04d}" for ordinal, call in enumerate(inputs.calls): call_id = _require_id(getattr(call, "call_id", None), field="call.call_id") trusted_by_id[call_id] = call public_call_id = public_call_id_by_trusted_id[call_id] item: RuntimeCallEvidence if type(call) is TrustedModelGenerationCall: segment = segment_by_id.get(call.segment_id) if segment is None or segment.kind != "carrier": raise GlyphHybridEvidenceError( "model call must reference one carrier segment" ) if call.text != segment.canonical_text: raise GlyphHybridEvidenceError( "model call text must exactly match carrier canonical text" ) _require_int( call.attempt_ordinal, field=f"calls[{ordinal}].attempt_ordinal", ) _require_int( call.seed, field=f"calls[{ordinal}].seed", maximum=2**32 - 1, ) _require_id(call.policy_id, field=f"calls[{ordinal}].policy_id") _require_sha256( call.policy_sha256, field=f"calls[{ordinal}].policy_sha256", ) if ( call.policy_id, call.policy_sha256, ) not in generation_policies: raise GlyphHybridEvidenceError( "model call uses an unpinned generation policy" ) model_pin = model_pin_by_name.get(call.model_pin_name) if model_pin is None: raise GlyphHybridEvidenceError("model call uses an unpinned model") output = _require_pcm16( call.output_pcm16_le, field=f"calls[{ordinal}].output_pcm16_le", ) endpoint = call.endpoint if type(endpoint) is not EndpointEvidence: raise GlyphHybridEvidenceError("model endpoint evidence is missing") try: replayed = inputs.generation_evaluator( model_pin, call.text, call.seed, call.policy_id, call.policy_sha256, inputs.gate_policy.sample_rate, ) except Exception as exc: raise GlyphHybridEvidenceError( "pinned model generation failed during rebuild" ) from exc if type(replayed) is not tuple or len(replayed) != 2: raise GlyphHybridEvidenceError( "pinned model generation returned invalid replay evidence" ) replayed_output = _require_pcm16( replayed[0], field=f"calls[{ordinal}].replayed_output_pcm16_le", ) replayed_endpoint = replayed[1] if type(replayed_endpoint) is not EndpointEvidence: raise GlyphHybridEvidenceError( "pinned model generation returned an invalid endpoint" ) if ( replayed_output != output or replayed_endpoint != endpoint ): raise GlyphHybridEvidenceError( "model output or endpoint does not match pinned replay" ) stop_reason = _require_id( endpoint.stop_reason, field=f"calls[{ordinal}].endpoint.stop_reason", ) if stop_reason not in _GENERATION_STOP_REASONS: raise GlyphHybridEvidenceError( "model stop reason is outside the frozen policy" ) tail = _require_float( endpoint.tail_energy_ratio, field=f"calls[{ordinal}].endpoint.tail_energy_ratio", minimum=0.0, ) silence = _require_int( endpoint.terminal_silence_samples, field=f"calls[{ordinal}].endpoint.terminal_silence_samples", ) generated_steps = _require_int( endpoint.generated_steps, field=f"calls[{ordinal}].endpoint.generated_steps", minimum=1, ) hard_cap_steps = _require_int( endpoint.hard_cap_steps, field=f"calls[{ordinal}].endpoint.hard_cap_steps", minimum=1, ) if ( generated_steps > hard_cap_steps or hard_cap_steps > 2_000 or ( stop_reason == "hard_stop" and generated_steps != hard_cap_steps ) ): raise GlyphHybridEvidenceError("generated steps exceed hard cap") measured_tail = _endpoint_tail_energy_ratio( output, inputs.gate_policy, ) measured_silence = _terminal_silence_samples( output, inputs.gate_policy, ) if not math.isclose( tail, measured_tail, rel_tol=0.0, abs_tol=1.0e-12, ): raise GlyphHybridEvidenceError( "endpoint tail energy does not match exact model PCM" ) if silence != measured_silence: raise GlyphHybridEvidenceError( "terminal silence does not match exact model PCM" ) generated_units = _count_speech_units(call.text) if generated_units <= 0: raise GlyphHybridEvidenceError( "model generation text has no speech units" ) item = ModelGenerationCallEvidence( ordinal=ordinal, call_id=public_call_id, kind="model_generation", segment_id=public_segment_id_by_trusted_id[call.segment_id], attempt_ordinal=call.attempt_ordinal, text_sha256=_content_commitment( content_commitment_key, "semantic-text", call.text.encode("utf-8"), ), generated_speech_units=generated_units, generated_chunks=1, seed=call.seed, policy_id=call.policy_id, policy_sha256=call.policy_sha256, model_pin_name=call.model_pin_name, output_pcm_sha256=_pcm_commitment( content_commitment_key, output, ), output_sample_count=len(output) // 2, stop_reason=stop_reason, tail_energy_ratio=tail, terminal_silence_samples=silence, generated_steps=generated_steps, hard_cap_steps=hard_cap_steps, ) elif type(call) is TrustedAssetLoadCall: segment = segment_by_id.get(call.segment_id) if segment is None or segment.kind != "asset": raise GlyphHybridEvidenceError( "asset load must reference one asset segment" ) asset = asset_by_id.get(call.asset_id) if asset is None or segment.asset_id != call.asset_id: raise GlyphHybridEvidenceError("asset load attribution is invalid") entry_sha = _require_sha256( call.manifest_entry_sha256, field=f"calls[{ordinal}].manifest_entry_sha256", ) loaded = _require_pcm16( call.loaded_pcm16_le, field=f"calls[{ordinal}].loaded_pcm16_le", ) if ( entry_sha != asset.manifest_entry_sha256 or loaded != asset.pcm16_le ): raise GlyphHybridEvidenceError( "asset load does not match trusted bundle bytes" ) item = AssetLoadCallEvidence( ordinal=ordinal, call_id=public_call_id, kind="asset_load", segment_id=public_segment_id_by_trusted_id[call.segment_id], asset_id=_public_asset_ref( content_commitment_key, call.asset_id, ), manifest_entry_sha256=_content_commitment( content_commitment_key, "asset-manifest-entry", entry_sha.encode("ascii"), ), loaded_pcm_sha256=_pcm_commitment( content_commitment_key, loaded, ), loaded_sample_count=len(loaded) // 2, ) elif type(call) is TrustedAsrCall: evaluator_pin = evaluator_pin_by_name.get(call.evaluator_pin_name) if evaluator_pin is None: raise GlyphHybridEvidenceError("ASR call uses an unpinned evaluator") if evaluator_pin.kind != "asr": raise GlyphHybridEvidenceError("ASR call uses a non-ASR pin") scope = _require_id(call.scope, field=f"calls[{ordinal}].scope") pcm = _require_pcm16( call.input_pcm16_le, field=f"calls[{ordinal}].input_pcm16_le", ) target = _require_text( call.target_text, field=f"calls[{ordinal}].target_text", ) transcript = _require_text( call.transcript_text, field=f"calls[{ordinal}].transcript_text", allow_empty=True, ) try: recomputed_transcript = inputs.asr_evaluator( evaluator_pin, pcm, inputs.gate_policy.sample_rate, ) except Exception as exc: raise GlyphHybridEvidenceError( "pinned ASR evaluator failed during rebuild" ) from exc recomputed_transcript = _require_text( recomputed_transcript, field=f"calls[{ordinal}].recomputed_transcript", allow_empty=True, ) if transcript != recomputed_transcript: raise GlyphHybridEvidenceError( "ASR transcript does not match pinned evaluator rebuild" ) transcript = recomputed_transcript metrics = _asr_metrics(target, transcript) subject_ref: str | None = None if call.subject_model_call_id is not None: subject_ref = public_call_id_by_trusted_id.get( call.subject_model_call_id ) if subject_ref is None: raise GlyphHybridEvidenceError( "ASR call references unknown model call" ) item = AsrCallEvidence( ordinal=ordinal, call_id=public_call_id, kind="asr", scope=scope, evaluator_pin_name=call.evaluator_pin_name, input_pcm_sha256=_pcm_commitment( content_commitment_key, pcm, ), input_sample_count=len(pcm) // 2, target_sha256=_content_commitment( content_commitment_key, "semantic-text", target.encode("utf-8"), ), transcript_sha256=_content_commitment( content_commitment_key, "asr-transcript", transcript.encode("utf-8"), ), target_units=int(metrics["target_units"]), hypothesis_units=int(metrics["hypothesis_units"]), edit_distance=int(metrics["edit_distance"]), cer=float(metrics["cer"]), prefix_cer=float(metrics["prefix_cer"]), suffix_cer=float(metrics["suffix_cer"]), prefix_deletions=int(metrics["prefix_deletions"]), suffix_deletions=int(metrics["suffix_deletions"]), extra_tail_units=int(metrics["extra_tail_units"]), semantic_available=bool(metrics["semantic_available"]), network_protected_spans=int( metrics["network_protected_spans"] ), network_protected_spans_passed=bool( metrics["network_protected_spans_passed"] ), subject_model_call_id=subject_ref, ) elif type(call) is TrustedEcapaCall: evaluator_pin = evaluator_pin_by_name.get(call.evaluator_pin_name) if evaluator_pin is None: raise GlyphHybridEvidenceError( "ECAPA call uses an unpinned evaluator" ) if evaluator_pin.kind != "ecapa": raise GlyphHybridEvidenceError("ECAPA call uses a non-ECAPA pin") scope = _require_id(call.scope, field=f"calls[{ordinal}].scope") pcm = _require_pcm16( call.input_pcm16_le, field=f"calls[{ordinal}].input_pcm16_le", ) anchor_sha = _require_sha256( call.speaker_anchor_sha256, field=f"calls[{ordinal}].speaker_anchor_sha256", ) if anchor_sha != inputs.contract.speaker_anchor_sha256: raise GlyphHybridEvidenceError( "ECAPA call uses an unpinned speaker anchor" ) centroid = _require_float( call.centroid_similarity, field=f"calls[{ordinal}].centroid_similarity", minimum=-1.0, maximum=1.0, ) begin = _require_float( call.begin_similarity, field=f"calls[{ordinal}].begin_similarity", minimum=-1.0, maximum=1.0, ) end = _require_float( call.end_similarity, field=f"calls[{ordinal}].end_similarity", minimum=-1.0, maximum=1.0, ) try: recomputed_ecapa = inputs.ecapa_evaluator( evaluator_pin, pcm, inputs.gate_policy.sample_rate, anchor_sha, ) except Exception as exc: raise GlyphHybridEvidenceError( "pinned ECAPA evaluator failed during rebuild" ) from exc if ( type(recomputed_ecapa) is not tuple or len(recomputed_ecapa) != 3 ): raise GlyphHybridEvidenceError( "pinned ECAPA evaluator returned invalid evidence" ) recomputed_ecapa_values = tuple( _require_float( value, field=f"calls[{ordinal}].recomputed_ecapa[{index}]", minimum=-1.0, maximum=1.0, ) for index, value in enumerate(recomputed_ecapa) ) if any( not math.isclose( observed, expected, rel_tol=0.0, abs_tol=1.0e-12, ) for observed, expected in zip( (centroid, begin, end), recomputed_ecapa_values, ) ): raise GlyphHybridEvidenceError( "ECAPA measurements do not match pinned evaluator rebuild" ) centroid, begin, end = recomputed_ecapa_values measurement = { "anchor_sha256": anchor_sha, "centroid_similarity": centroid, "begin_similarity": begin, "end_similarity": end, } subject_ref = None if call.subject_model_call_id is not None: subject_ref = public_call_id_by_trusted_id.get( call.subject_model_call_id ) if subject_ref is None: raise GlyphHybridEvidenceError( "ECAPA call references unknown model call" ) item = EcapaCallEvidence( ordinal=ordinal, call_id=public_call_id, kind="ecapa", scope=scope, evaluator_pin_name=call.evaluator_pin_name, input_pcm_sha256=_pcm_commitment( content_commitment_key, pcm, ), input_sample_count=len(pcm) // 2, speaker_anchor_sha256=anchor_sha, centroid_similarity=centroid, begin_similarity=begin, end_similarity=end, measurement_sha256=_measurement_sha256(measurement), subject_model_call_id=subject_ref, ) elif type(call) is TrustedSquimCall: evaluator_pin = evaluator_pin_by_name.get(call.evaluator_pin_name) if evaluator_pin is None: raise GlyphHybridEvidenceError( "SQUIM call uses an unpinned evaluator" ) if evaluator_pin.kind != "squim": raise GlyphHybridEvidenceError("SQUIM call uses a non-SQUIM pin") scope = _require_id(call.scope, field=f"calls[{ordinal}].scope") pcm = _require_pcm16( call.input_pcm16_le, field=f"calls[{ordinal}].input_pcm16_le", ) stoi = _require_float( call.stoi, field=f"calls[{ordinal}].stoi", minimum=0.0, maximum=1.0, ) pesq = _require_float( call.pesq, field=f"calls[{ordinal}].pesq", minimum=-0.5, maximum=5.0, ) si_sdr = _require_float( call.si_sdr, field=f"calls[{ordinal}].si_sdr", minimum=-100.0, maximum=100.0, ) try: recomputed_squim = inputs.squim_evaluator( evaluator_pin, pcm, inputs.gate_policy.sample_rate, ) except Exception as exc: raise GlyphHybridEvidenceError( "pinned SQUIM evaluator failed during rebuild" ) from exc if ( type(recomputed_squim) is not tuple or len(recomputed_squim) != 3 ): raise GlyphHybridEvidenceError( "pinned SQUIM evaluator returned invalid evidence" ) recomputed_squim_values = tuple( _require_float( value, field=f"calls[{ordinal}].recomputed_squim[{index}]", ) for index, value in enumerate(recomputed_squim) ) if any( not math.isclose( observed, expected, rel_tol=0.0, abs_tol=1.0e-12, ) for observed, expected in zip( (stoi, pesq, si_sdr), recomputed_squim_values, ) ): raise GlyphHybridEvidenceError( "SQUIM measurements do not match pinned evaluator rebuild" ) stoi, pesq, si_sdr = recomputed_squim_values measurement = {"stoi": stoi, "pesq": pesq, "si_sdr": si_sdr} subject_ref = None if call.subject_model_call_id is not None: subject_ref = public_call_id_by_trusted_id.get( call.subject_model_call_id ) if subject_ref is None: raise GlyphHybridEvidenceError( "SQUIM call references unknown model call" ) item = SquimCallEvidence( ordinal=ordinal, call_id=public_call_id, kind="squim", scope=scope, evaluator_pin_name=call.evaluator_pin_name, input_pcm_sha256=_pcm_commitment( content_commitment_key, pcm, ), input_sample_count=len(pcm) // 2, stoi=stoi, pesq=pesq, si_sdr=si_sdr, measurement_sha256=_measurement_sha256(measurement), subject_model_call_id=subject_ref, ) else: raise GlyphHybridEvidenceError( f"calls[{ordinal}] has unsupported runtime call type" ) evidence.append(item) evidence_by_id[call_id] = item if not evidence: raise GlyphHybridEvidenceError("runtime call ledger must not be empty") return tuple(evidence), evidence_by_id, trusted_by_id def _pcm_measurements(pcm16_le: bytes) -> tuple[float, float]: pcm = np.frombuffer(pcm16_le, dtype=" int: signal = ( np.frombuffer(pcm16_le, dtype="= threshold]: start = int(raw_start) end = min(signal.size, start + frame) if intervals and start <= intervals[-1][1]: intervals[-1][1] = max(intervals[-1][1], end) else: intervals.append([start, end]) return sum(end - start for start, end in intervals) def _endpoint_tail_energy_ratio( pcm16_le: bytes, policy: GatePolicy, ) -> float: signal = ( np.frombuffer(pcm16_le, dtype=" int: signal = np.abs( np.frombuffer(pcm16_le, dtype=" threshold: break count += 1 return count def measure_active_speech_samples( pcm16_le: bytes, policy: GatePolicy, ) -> int: """Measure the frozen active-interval union from exact PCM16 bytes.""" pcm = _require_pcm16(pcm16_le, field="pcm16_le") _gate_policy_sha256(policy) return _active_speech_samples(pcm, policy) def measure_endpoint_tail_energy_ratio( pcm16_le: bytes, policy: GatePolicy, ) -> float: """Measure the frozen endpoint tail ratio from exact PCM16 bytes.""" pcm = _require_pcm16(pcm16_le, field="pcm16_le") _gate_policy_sha256(policy) return _endpoint_tail_energy_ratio(pcm, policy) def measure_terminal_silence_samples( pcm16_le: bytes, policy: GatePolicy, ) -> int: """Measure the exact terminal-silence run from PCM16 bytes.""" pcm = _require_pcm16(pcm16_le, field="pcm16_le") _gate_policy_sha256(policy) return _terminal_silence_samples(pcm, policy) def _gate_policy_sha256(policy: GatePolicy) -> str: if type(policy) is not GatePolicy: raise GlyphHybridEvidenceError("gate_policy must be GatePolicy") _require_int(policy.sample_rate, field="gate_policy.sample_rate", minimum=1) _require_float(policy.max_cer, field="gate_policy.max_cer", minimum=0.0) _require_float( policy.max_prefix_cer, field="gate_policy.max_prefix_cer", minimum=0.0, ) _require_float( policy.max_suffix_cer, field="gate_policy.max_suffix_cer", minimum=0.0, ) _require_int( policy.max_prefix_deletions, field="gate_policy.max_prefix_deletions", ) _require_int( policy.max_suffix_deletions, field="gate_policy.max_suffix_deletions", ) _require_int( policy.max_extra_tail_units, field="gate_policy.max_extra_tail_units", ) for field in ( "max_pace_cps", "min_speaker_similarity", "max_boundary_speaker_drop", "min_squim_stoi", "min_squim_pesq", "min_squim_si_sdr", "min_pcm_rms", "max_pcm_peak", "max_endpoint_tail_energy_ratio", ): _require_float(getattr(policy, field), field=f"gate_policy.{field}") for field in ( "active_voice_top_db", "active_voice_frame_ms", "active_voice_hop_ms", "active_voice_min_rms", "endpoint_tail_window_ms", "endpoint_silence_amplitude", ): _require_float( getattr(policy, field), field=f"gate_policy.{field}", minimum=0.0, ) if ( policy.active_voice_frame_ms <= 0.0 or policy.active_voice_hop_ms <= 0.0 or policy.active_voice_min_rms <= 0.0 or policy.endpoint_tail_window_ms <= 0.0 ): raise GlyphHybridEvidenceError( "gate policy detector durations and RMS floor must be positive" ) return _sha256_bytes(_canonical_json_bytes(_payload(policy))) def _build_selections( inputs: EvidenceRebuildInputs, segment_by_id: Mapping[str, TrustedRenderSegment], public_segment_id_by_trusted_id: Mapping[str, str], evidence_by_call_id: Mapping[str, RuntimeCallEvidence], trusted_by_call_id: Mapping[str, TrustedRuntimeCall], output_pcm: bytes, ) -> tuple[tuple[CarrierSelectionEvidence, ...], str, tuple[str, ...]]: carrier_ids = { segment.segment_id for segment in segment_by_id.values() if segment.kind == "carrier" } selections: list[CarrierSelectionEvidence] = [] selected_ids: list[str] = [] attributed_attempts: set[str] = set() seen_segments: set[str] = set() for ordinal, selection in enumerate(inputs.carrier_selections): if type(selection) is not TrustedCarrierSelection: raise GlyphHybridEvidenceError("carrier selection type is invalid") if selection.segment_id not in carrier_ids: raise GlyphHybridEvidenceError( "carrier selection references non-carrier segment" ) if selection.segment_id in seen_segments: raise GlyphHybridEvidenceError( "carrier segment has duplicate selection evidence" ) seen_segments.add(selection.segment_id) path_rank = _require_int( selection.path_rank, field=f"carrier_selections[{ordinal}].path_rank", minimum=1, ) if not selection.attempted_call_ids: raise GlyphHybridEvidenceError("carrier selection has no attempts") if len(set(selection.attempted_call_ids)) != len( selection.attempted_call_ids ): raise GlyphHybridEvidenceError("carrier attempts contain duplicates") if selection.selected_call_id not in selection.attempted_call_ids: raise GlyphHybridEvidenceError("selected call is not an attempted call") attempts: list[ModelGenerationCallEvidence] = [] for call_id in selection.attempted_call_ids: _require_id(call_id, field="carrier attempted call id") item = evidence_by_call_id.get(call_id) trusted = trusted_by_call_id.get(call_id) if ( type(item) is not ModelGenerationCallEvidence or type(trusted) is not TrustedModelGenerationCall or trusted.segment_id != selection.segment_id ): raise GlyphHybridEvidenceError( "carrier attempt borrows an unrelated runtime call" ) if call_id in attributed_attempts: raise GlyphHybridEvidenceError( "model attempt is attributed more than once" ) attributed_attempts.add(call_id) attempts.append(item) if [item.ordinal for item in attempts] != sorted( item.ordinal for item in attempts ): raise GlyphHybridEvidenceError( "carrier attempts must preserve runtime ledger order" ) if [item.attempt_ordinal for item in attempts] != list( range(len(attempts)) ): raise GlyphHybridEvidenceError( "carrier attempt ordinals must be contiguous from zero" ) selected = evidence_by_call_id[selection.selected_call_id] assert isinstance(selected, ModelGenerationCallEvidence) segment = segment_by_id[selection.segment_id] span = output_pcm[ segment.output_start_sample * 2 : segment.output_end_sample * 2 ] if selected.output_pcm_sha256 != _pcm_commitment( inputs.content_commitment_key, span, ): raise GlyphHybridEvidenceError( "selected model output is not exact realized carrier PCM" ) selection_payload = { "segment_id": public_segment_id_by_trusted_id[selection.segment_id], "path_rank": path_rank, "attempted_call_ids": [item.call_id for item in attempts], "selected_call_id": selected.call_id, } selections.append( CarrierSelectionEvidence( segment_id=public_segment_id_by_trusted_id[ selection.segment_id ], path_rank=path_rank, attempted_call_ids=tuple(item.call_id for item in attempts), selected_call_id=selected.call_id, selection_sha256=_sha256_bytes( _canonical_json_bytes(selection_payload) ), ) ) selected_ids.append(selection.selected_call_id) if seen_segments != carrier_ids: raise GlyphHybridEvidenceError( "every carrier segment needs exactly one selected path" ) all_model_ids = { trusted_call_id for trusted_call_id, item in evidence_by_call_id.items() if isinstance(item, ModelGenerationCallEvidence) } if attributed_attempts != all_model_ids: raise GlyphHybridEvidenceError( "every model call must be disclosed and attributed exactly once" ) selections_tuple = tuple(selections) return ( selections_tuple, _sha256_bytes(_canonical_json_bytes(_payload(selections_tuple))), tuple(selected_ids), ) def _validate_call_attribution( inputs: EvidenceRebuildInputs, segment_by_id: Mapping[str, TrustedRenderSegment], asset_by_id: Mapping[str, TrustedAsset], evidence_by_call_id: Mapping[str, RuntimeCallEvidence], trusted_by_call_id: Mapping[str, TrustedRuntimeCall], output_pcm: bytes, gate_call_ids: set[str], content_commitment_key: bytes, ) -> None: asset_segment_ids = { segment.segment_id for segment in segment_by_id.values() if segment.kind == "asset" } loaded_segment_ids: set[str] = set() model_ids = { trusted_call_id for trusted_call_id, item in evidence_by_call_id.items() if isinstance(item, ModelGenerationCallEvidence) } production_ordinals = [ item.ordinal for item in evidence_by_call_id.values() if isinstance( item, (ModelGenerationCallEvidence, AssetLoadCallEvidence), ) ] last_production_ordinal = max(production_ordinals, default=-1) for trusted_call_id, item in evidence_by_call_id.items(): trusted = trusted_by_call_id[trusted_call_id] if isinstance(item, AssetLoadCallEvidence): assert isinstance(trusted, TrustedAssetLoadCall) if trusted.segment_id in loaded_segment_ids: raise GlyphHybridEvidenceError( "asset segment has duplicate load calls" ) loaded_segment_ids.add(trusted.segment_id) segment = segment_by_id[trusted.segment_id] span = output_pcm[ segment.output_start_sample * 2 : segment.output_end_sample * 2 ] asset = asset_by_id[trusted.asset_id] expected_loaded_commitment = _pcm_commitment( content_commitment_key, span, ) if ( span != asset.pcm16_le or item.loaded_pcm_sha256 != expected_loaded_commitment ): raise GlyphHybridEvidenceError( "asset load is not exact public reconstructed PCM" ) elif isinstance(item, (AsrCallEvidence, EcapaCallEvidence, SquimCallEvidence)): subject = getattr(trusted, "subject_model_call_id", None) if subject is None: if trusted_call_id not in gate_call_ids: raise GlyphHybridEvidenceError( "whole-output evaluator call is not gate-attributed" ) if item.input_pcm_sha256 != _pcm_commitment( content_commitment_key, output_pcm, ): raise GlyphHybridEvidenceError( "whole-output evaluator did not receive exact output PCM" ) if item.ordinal <= last_production_ordinal: raise GlyphHybridEvidenceError( "whole-output evaluator precedes production calls" ) else: if subject not in model_ids: raise GlyphHybridEvidenceError( "local evaluator call references unknown model call" ) model_item = evidence_by_call_id[subject] assert isinstance(model_item, ModelGenerationCallEvidence) if item.ordinal <= model_item.ordinal: raise GlyphHybridEvidenceError( "local evaluator precedes its model call" ) if item.input_pcm_sha256 != model_item.output_pcm_sha256: raise GlyphHybridEvidenceError( "local evaluator input does not match model-call PCM" ) if ( isinstance(item, AsrCallEvidence) and item.target_sha256 != model_item.text_sha256 ): raise GlyphHybridEvidenceError( "local ASR target does not match model-call text" ) if loaded_segment_ids != asset_segment_ids: raise GlyphHybridEvidenceError( "every asset segment needs exactly one disclosed asset load" ) def _build_gate_projection( binding: TrustedGateBinding, *, policy: GatePolicy, policy_sha256: str, canonical_target: str, output_pcm: bytes, evidence_by_call_id: Mapping[str, RuntimeCallEvidence], selected_model_call_ids: Sequence[str], content_commitment_key: bytes, ) -> GateProjection: if type(binding) is not TrustedGateBinding: raise GlyphHybridEvidenceError("gate binding type is invalid") if binding.scope not in _FINAL_SCOPES: raise GlyphHybridEvidenceError("gate binding scope is invalid") asr = evidence_by_call_id.get(binding.asr_call_id) ecapa = evidence_by_call_id.get(binding.ecapa_call_id) squim = evidence_by_call_id.get(binding.squim_call_id) if ( type(asr) is not AsrCallEvidence or type(ecapa) is not EcapaCallEvidence or type(squim) is not SquimCallEvidence ): raise GlyphHybridEvidenceError( "gate binding must reference ASR, ECAPA and SQUIM calls" ) if not (asr.scope == ecapa.scope == squim.scope == binding.scope): raise GlyphHybridEvidenceError("gate call scopes do not match binding") if any( item.subject_model_call_id is not None for item in (asr, ecapa, squim) ): raise GlyphHybridEvidenceError( "final gates must use whole-output evaluator calls" ) output_sha = _pcm_commitment(content_commitment_key, output_pcm) if any( item.input_pcm_sha256 != output_sha for item in (asr, ecapa, squim) ): raise GlyphHybridEvidenceError( "gate evaluators did not receive exact output PCM" ) if asr.target_sha256 != _content_commitment( content_commitment_key, "semantic-text", canonical_target.encode("utf-8"), ): raise GlyphHybridEvidenceError("ASR target does not match canonical target") claimed_active_samples = _require_int( binding.active_speech_samples, field=f"{binding.scope}.active_speech_samples", minimum=1, maximum=len(output_pcm) // 2, ) active_samples = _active_speech_samples(output_pcm, policy) if active_samples <= 0: raise GlyphHybridEvidenceError("output PCM contains no active speech") if claimed_active_samples != active_samples: raise GlyphHybridEvidenceError( "active speech samples do not match exact output PCM" ) pace = asr.target_units / (active_samples / policy.sample_rate) boundary_drop = max(0.0, ecapa.begin_similarity - ecapa.end_similarity) pcm_rms, pcm_peak = _pcm_measurements(output_pcm) endpoint_items = [ evidence_by_call_id[call_id] for call_id in selected_model_call_ids ] endpoint_complete = all( isinstance(item, ModelGenerationCallEvidence) and item.generated_steps <= item.hard_cap_steps and bool(item.stop_reason) for item in endpoint_items ) endpoint_tail = max( ( item.tail_energy_ratio for item in endpoint_items if isinstance(item, ModelGenerationCallEvidence) ), default=0.0, ) reasons: list[str] = [] checks = ( ("cer", asr.cer <= policy.max_cer), ("prefix_cer", asr.prefix_cer <= policy.max_prefix_cer), ("suffix_cer", asr.suffix_cer <= policy.max_suffix_cer), ( "prefix_deletions", asr.prefix_deletions <= policy.max_prefix_deletions, ), ( "suffix_deletions", asr.suffix_deletions <= policy.max_suffix_deletions, ), ( "extra_tail_units", asr.extra_tail_units <= policy.max_extra_tail_units, ), ("semantic_available", asr.semantic_available), ( "network_protected_spans", asr.network_protected_spans_passed, ), ("pace", pace <= policy.max_pace_cps), ( "speaker_similarity", ecapa.centroid_similarity >= policy.min_speaker_similarity, ), ( "boundary_speaker_drop", boundary_drop <= policy.max_boundary_speaker_drop, ), ("squim_stoi", squim.stoi >= policy.min_squim_stoi), ("squim_pesq", squim.pesq >= policy.min_squim_pesq), ("squim_si_sdr", squim.si_sdr >= policy.min_squim_si_sdr), ("pcm_rms", pcm_rms >= policy.min_pcm_rms), ("pcm_peak", pcm_peak <= policy.max_pcm_peak), ("endpoint_evidence", endpoint_complete), ( "endpoint_tail_energy", endpoint_tail <= policy.max_endpoint_tail_energy_ratio, ), ) reasons.extend(name for name, passed in checks if not passed) return GateProjection( scope=binding.scope, policy_sha256=policy_sha256, pcm_sha256=output_sha, asr_call_id=asr.call_id, ecapa_call_id=ecapa.call_id, squim_call_id=squim.call_id, target_units=asr.target_units, hypothesis_units=asr.hypothesis_units, edit_distance=asr.edit_distance, cer=asr.cer, prefix_cer=asr.prefix_cer, suffix_cer=asr.suffix_cer, prefix_deletions=asr.prefix_deletions, suffix_deletions=asr.suffix_deletions, extra_tail_units=asr.extra_tail_units, semantic_available=asr.semantic_available, network_protected_spans=asr.network_protected_spans, network_protected_spans_passed=( asr.network_protected_spans_passed ), pace_cps=pace, speaker_similarity=ecapa.centroid_similarity, boundary_speaker_drop=boundary_drop, squim_stoi=squim.stoi, squim_pesq=squim.pesq, squim_si_sdr=squim.si_sdr, pcm_rms=pcm_rms, pcm_peak=pcm_peak, endpoint_tail_energy_ratio=endpoint_tail, endpoint_evidence_complete=endpoint_complete, passed=not reasons, rejection_reasons=tuple(reasons), ) def _validate_caps(caps: BudgetCaps) -> dict[str, int]: if type(caps) is not BudgetCaps: raise GlyphHybridEvidenceError("budget_caps must be BudgetCaps") values = { name: _require_int( getattr(caps, name), field=f"budget_caps.{name}", minimum=0, ) for name in _BUDGET_NAMES } if values["model_invocations"] > MAX_MODEL_INVOCATIONS: raise GlyphHybridEvidenceError("model invocation cap exceeds hard maximum") if values["generated_chunks"] > MAX_GENERATED_CHUNKS: raise GlyphHybridEvidenceError("generated chunk cap exceeds hard maximum") if values["generated_speech_units"] > MAX_GENERATED_SPEECH_UNITS: raise GlyphHybridEvidenceError( "generated speech-unit cap exceeds hard maximum" ) hard_maxima = { "asset_count": MAX_ASSET_COUNT, "asset_atoms": MAX_ASSET_ATOMS, "asset_canonical_units": MAX_ASSET_CANONICAL_UNITS, "asset_audio_samples": MAX_ASSET_AUDIO_SAMPLES, "asr_calls": MAX_ASR_CALLS, "ecapa_calls": MAX_ECAPA_CALLS, "squim_calls": MAX_SQUIM_CALLS, } for name, maximum in hard_maxima.items(): if values[name] > maximum: raise GlyphHybridEvidenceError( f"{name} cap exceeds hard maximum" ) return values def _build_budgets( inputs: EvidenceRebuildInputs, segments: Sequence[RenderSegmentEvidence], calls: Sequence[RuntimeCallEvidence], ) -> tuple[HardBudgetEvidence, ...]: caps = _validate_caps(inputs.budget_caps) asset_segments = [item for item in segments if item.kind == "asset"] unique_assets = {item.asset_id for item in asset_segments} model_calls = [ item for item in calls if isinstance(item, ModelGenerationCallEvidence) ] actual = { "model_invocations": len(model_calls), "generated_chunks": sum(item.generated_chunks for item in model_calls), "generated_speech_units": sum( item.generated_speech_units for item in model_calls ), "asset_count": len(unique_assets), "asset_atoms": len(asset_segments), "asset_canonical_units": sum( item.canonical_end - item.canonical_start for item in asset_segments ), "asset_audio_samples": sum( item.output_sample_count for item in asset_segments ), "asr_calls": sum(isinstance(item, AsrCallEvidence) for item in calls), "ecapa_calls": sum(isinstance(item, EcapaCallEvidence) for item in calls), "squim_calls": sum(isinstance(item, SquimCallEvidence) for item in calls), } evidence: list[HardBudgetEvidence] = [] for name in _BUDGET_NAMES: if actual[name] > caps[name]: raise GlyphHybridEvidenceError( f"{name} actual {actual[name]} exceeds hard cap {caps[name]}" ) evidence.append( HardBudgetEvidence(name=name, actual=actual[name], cap=caps[name]) ) return tuple(evidence) def _evidence_without_digest(evidence: HybridEvidence) -> dict[str, Any]: payload = _payload(evidence) payload.pop("evidence_sha256") return payload def rebuild_hybrid_evidence(inputs: EvidenceRebuildInputs) -> HybridEvidence: """Independently rebuild schema 7 from trusted, text-bearing runtime data. The function rejects a failed final/independent gate. There is no API parameter for a caller-provided ``passed`` value. """ if type(inputs) is not EvidenceRebuildInputs: raise GlyphHybridEvidenceError("inputs must be EvidenceRebuildInputs") content_commitment_key = _require_content_commitment_key( inputs.content_commitment_key ) for name in ( "generation_evaluator", "renderer_evaluator", "asr_evaluator", "ecapa_evaluator", "squim_evaluator", ): if not callable(getattr(inputs, name)): raise GlyphHybridEvidenceError( f"{name} must be a trusted callable" ) output_pcm = _require_pcm16( inputs.assembled_pcm16_le, field="assembled_pcm16_le", ) public_pcm = _require_pcm16(inputs.public_pcm16_le, field="public_pcm16_le") public_pcm_equality = output_pcm == public_pcm if not public_pcm_equality: raise GlyphHybridEvidenceError("public PCM differs from assembled PCM") policy_sha = _gate_policy_sha256(inputs.gate_policy) asset_by_id = _validate_bundle(inputs.bundle) _validate_contract( inputs.contract, inputs.plan, inputs.bundle, policy_sha, ) _validate_renderer_rebuild(inputs) public_contract = _content_free_contract( inputs.contract, content_commitment_key, ) contract_sha = _sha256_bytes( _canonical_json_bytes(_payload(public_contract)) ) ( segments, semantic_plan_sha, inverse_proof_sha, realized_plan_sha, strict_inverse, asset_reconstruction, ) = _build_segments( inputs.plan, asset_by_id, output_pcm, content_commitment_key, ) segment_by_id = { segment.segment_id: segment for segment in inputs.plan.segments } public_segment_id_by_trusted_id = { segment.segment_id: f"segment-{ordinal:04d}" for ordinal, segment in enumerate(inputs.plan.segments) } call_evidence, call_by_id, trusted_call_by_id = _build_calls( inputs, segment_by_id, asset_by_id, public_segment_id_by_trusted_id, content_commitment_key, ) selections, selected_sha, selected_model_call_ids = _build_selections( inputs, segment_by_id, public_segment_id_by_trusted_id, call_by_id, trusted_call_by_id, output_pcm, ) if len(inputs.gate_bindings) != 2: raise GlyphHybridEvidenceError( "exactly final and independent gate bindings are required" ) binding_by_scope: dict[str, TrustedGateBinding] = {} for binding in inputs.gate_bindings: if type(binding) is not TrustedGateBinding: raise GlyphHybridEvidenceError("gate binding type is invalid") if binding.scope in binding_by_scope: raise GlyphHybridEvidenceError("gate binding scopes must be unique") binding_by_scope[binding.scope] = binding if set(binding_by_scope) != _FINAL_SCOPES: raise GlyphHybridEvidenceError( "both final and independent gate bindings are required" ) final_asr = call_by_id.get( binding_by_scope["final"].asr_call_id ) independent_asr = call_by_id.get( binding_by_scope["independent"].asr_call_id ) if ( type(final_asr) is not AsrCallEvidence or type(independent_asr) is not AsrCallEvidence ): raise GlyphHybridEvidenceError( "final and independent gates require ASR calls" ) evaluator_pin_by_name = { pin.name: pin for pin in inputs.contract.evaluator_pins } final_asr_pin = evaluator_pin_by_name[final_asr.evaluator_pin_name] independent_asr_pin = evaluator_pin_by_name[ independent_asr.evaluator_pin_name ] if ( final_asr.evaluator_pin_name == independent_asr.evaluator_pin_name or final_asr_pin.config_sha256 == independent_asr_pin.config_sha256 ): raise GlyphHybridEvidenceError( "ASR gates must use distinct pinned decoder profiles" ) if ( final_asr_pin.uri, final_asr_pin.revision, final_asr_pin.sha256, ) != ( independent_asr_pin.uri, independent_asr_pin.revision, independent_asr_pin.sha256, ): raise GlyphHybridEvidenceError( "ASR decoder profiles must share the exact pinned model weights" ) if ( _normalized_hf_uri(final_asr_pin.uri) != BREEZE25_MODEL_ID.casefold() or final_asr_pin.revision != BREEZE25_REVISION or final_asr_pin.sha256 != BREEZE25_WEIGHT_SHA256 ): raise GlyphHybridEvidenceError( "ASR decoder profiles must use the official pinned Breeze ASR 25 " "weights" ) if ( final_asr.evaluator_pin_name != BREEZE25_PRIMARY_PIN_NAME or independent_asr.evaluator_pin_name != BREEZE25_CONFIRMATION_PIN_NAME ): raise GlyphHybridEvidenceError( "ASR gates must use the honest Breeze primary and confirmation " "logical pin names" ) gate_call_ids = { call_id for binding in binding_by_scope.values() for call_id in ( binding.asr_call_id, binding.ecapa_call_id, binding.squim_call_id, ) } _validate_call_attribution( inputs, segment_by_id, asset_by_id, call_by_id, trusted_call_by_id, output_pcm, gate_call_ids, content_commitment_key, ) final_gate = _build_gate_projection( binding_by_scope["final"], policy=inputs.gate_policy, policy_sha256=policy_sha, canonical_target=inputs.plan.canonical_target, output_pcm=output_pcm, evidence_by_call_id=call_by_id, selected_model_call_ids=selected_model_call_ids, content_commitment_key=content_commitment_key, ) independent_gate = _build_gate_projection( binding_by_scope["independent"], policy=inputs.gate_policy, policy_sha256=policy_sha, canonical_target=inputs.plan.canonical_target, output_pcm=output_pcm, evidence_by_call_id=call_by_id, selected_model_call_ids=selected_model_call_ids, content_commitment_key=content_commitment_key, ) if not final_gate.passed or not independent_gate.passed: raise GlyphHybridEvidenceError( "final and independent current gates must both pass" ) budgets = _build_budgets(inputs, segments, call_evidence) ledger_sha = _sha256_bytes(_canonical_json_bytes(_payload(call_evidence))) strict_gates = AdditionalStrictGates( strict_inverse_reconstruction=strict_inverse, asset_reconstruction=asset_reconstruction, public_pcm_equality=public_pcm_equality, ) if not all(asdict(strict_gates).values()): raise GlyphHybridEvidenceError("additional strict hybrid gates failed") request_proof = RequestProof( content_commitment_scheme="hmac-sha256-v1", raw_input_sha256=_content_commitment( content_commitment_key, "raw-input", inputs.plan.raw_input.encode("utf-8"), ), raw_input_codepoints=len(inputs.plan.raw_input), raw_input_utf8_bytes=len(inputs.plan.raw_input.encode("utf-8")), canonical_target_sha256=_content_commitment( content_commitment_key, "semantic-text", inputs.plan.canonical_target.encode("utf-8"), ), canonical_target_codepoints=len(inputs.plan.canonical_target), canonical_target_speech_units=_count_speech_units( inputs.plan.canonical_target ), semantic_plan_sha256=semantic_plan_sha, inverse_proof_sha256=inverse_proof_sha, realized_audio_plan_sha256=realized_plan_sha, assembled_pcm_sha256=_sha256_bytes(output_pcm), public_pcm_sha256=_sha256_bytes(public_pcm), output_sample_rate=inputs.gate_policy.sample_rate, output_channels=1, output_sample_format=PCM_SAMPLE_FORMAT, output_sample_count=len(output_pcm) // 2, output_byte_count=len(output_pcm), segments=segments, budgets=budgets, ledger_sha256=ledger_sha, selected_carriers_sha256=selected_sha, gate_policy_sha256=policy_sha, final_gate=final_gate, independent_gate=independent_gate, additional_strict_gates=strict_gates, ) provisional = HybridEvidence( schema=EVIDENCE_SCHEMA, version=EVIDENCE_VERSION, renderer_contract=public_contract, renderer_contract_sha256=contract_sha, request_proof=request_proof, runtime_call_ledger=call_evidence, carrier_selections=selections, evidence_sha256="0" * 64, ) digest = _sha256_bytes( _canonical_json_bytes(_evidence_without_digest(provisional)) ) return HybridEvidence( schema=provisional.schema, version=provisional.version, renderer_contract=provisional.renderer_contract, renderer_contract_sha256=provisional.renderer_contract_sha256, request_proof=provisional.request_proof, runtime_call_ledger=provisional.runtime_call_ledger, carrier_selections=provisional.carrier_selections, evidence_sha256=digest, ) def format_hybrid_evidence(inputs: EvidenceRebuildInputs) -> bytes: """Return canonical content-free schema-7 JSON rebuilt from trusted input.""" evidence = rebuild_hybrid_evidence(inputs) return _canonical_json_bytes(_payload(evidence)) def _reject_duplicate_keys( pairs: list[tuple[str, Any]], ) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise GlyphHybridEvidenceError( f"duplicate JSON object key is forbidden: {key!r}" ) result[key] = value return result def validate_hybrid_evidence( serialized: bytes, trusted_inputs: EvidenceRebuildInputs, ) -> HybridEvidence: """Accept evidence only if an independent trusted rebuild is byte-identical.""" if type(serialized) is not bytes or not serialized: raise GlyphHybridEvidenceError("serialized evidence must be non-empty bytes") try: decoded = serialized.decode("ascii") observed = json.loads(decoded, object_pairs_hook=_reject_duplicate_keys) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise GlyphHybridEvidenceError(f"invalid evidence JSON: {exc}") from exc if type(observed) is not dict: raise GlyphHybridEvidenceError("evidence root must be an object") if _canonical_json_bytes(observed) != serialized: raise GlyphHybridEvidenceError("evidence JSON is not canonical") expected = rebuild_hybrid_evidence(trusted_inputs) expected_payload = _payload(expected) if observed != expected_payload: raise GlyphHybridEvidenceError( "evidence does not match independent trusted rebuild" ) return expected def hybrid_evidence_sha256( serialized: bytes, trusted_inputs: EvidenceRebuildInputs, ) -> str: """Validate and return the canonical serialized evidence byte hash.""" validate_hybrid_evidence(serialized, trusted_inputs) return _sha256_bytes(serialized) __all__ = ( "EVIDENCE_SCHEMA", "EVIDENCE_VERSION", "MAX_MODEL_INVOCATIONS", "MAX_GENERATED_CHUNKS", "MAX_GENERATED_SPEECH_UNITS", "MAX_ASSET_COUNT", "MAX_ASSET_ATOMS", "MAX_ASSET_CANONICAL_UNITS", "MAX_ASSET_AUDIO_SAMPLES", "MAX_ASR_CALLS", "MAX_ECAPA_CALLS", "MAX_SQUIM_CALLS", "GlyphHybridEvidenceError", "ArtifactPin", "RendererContract", "TrustedAsset", "TrustedGlyphBundle", "compute_trusted_asset_entry_sha256", "compute_trusted_bundle_digests", "measure_active_speech_samples", "measure_endpoint_tail_energy_ratio", "measure_terminal_silence_samples", "TrustedRenderSegment", "TrustedRendererPlan", "EndpointEvidence", "TrustedModelGenerationCall", "TrustedAssetLoadCall", "TrustedAsrCall", "TrustedEcapaCall", "TrustedSquimCall", "GenerationEvaluator", "RendererEvaluator", "TrustedCarrierSelection", "TrustedGateBinding", "BudgetCaps", "GatePolicy", "EvidenceRebuildInputs", "RenderSegmentEvidence", "HardBudgetEvidence", "ModelGenerationCallEvidence", "AssetLoadCallEvidence", "AsrCallEvidence", "EcapaCallEvidence", "SquimCallEvidence", "CarrierSelectionEvidence", "GateProjection", "AdditionalStrictGates", "RequestProof", "HybridEvidence", "rebuild_hybrid_evidence", "format_hybrid_evidence", "validate_hybrid_evidence", "hybrid_evidence_sha256", )