"""Hardened production adapter for schema-7 glyph-hybrid evidence. The adapter is the only bridge from the hosted runtime's text-bearing objects to :mod:`glyph_hybrid_evidence`. It rebuilds the renderer plan, replays every model attempt, reruns two explicitly distinct decoder profiles over the same pinned Breeze ASR 25 weights plus ECAPA and SQUIM on exact public PCM, validates the immutable assembly, and only then returns content-free canonical JSON. """ from __future__ import annotations from dataclasses import asdict, dataclass, is_dataclass import hashlib import json import math from pathlib import Path import secrets from typing import Any, Callable, Mapping, Sequence import numpy as np import glyph_ascii_v1 as glyph_ascii_module import glyph_assets as glyph_assets_module import glyph_hybrid_evidence as evidence import glyph_hybrid_plan as glyph_plan_module import quality_runtime as quality_runtime_module from glyph_assets import ( AssetSegment, CarrierSegment, condition_carrier_v1, verify_hybrid_assembly, ) from glyph_hybrid_plan import build_glyph_hybrid_plan from quality_runtime import ( SQUIM_OBJECTIVE_MAX_WINDOWS, SQUIM_OBJECTIVE_SAMPLE_RATE, SQUIM_OBJECTIVE_WEIGHT_SHA256, SQUIM_OBJECTIVE_WINDOW_SECONDS, release_speaker_evidence_from_audio, squim_objective_evidence_from_audio, ) ADAPTER_SCHEMA = "bluemagpie.glyph-hybrid.adapter.v1" GENERATION_POLICY_SCHEMA = "bluemagpie.glyph-generation-policy.v1" ASR_TASK = "transcribe" BREEZE25_MODEL_ID = getattr( quality_runtime_module, "BREEZE25_MODEL_ID", "MediaTek-Research/Breeze-ASR-25", ) BREEZE25_REVISION = getattr( quality_runtime_module, "BREEZE25_REVISION", "cffe7ccb404d025296a00758d0a33468bec3a9d0", ) BREEZE25_PRIMARY_PIN_NAME = "breeze25-primary-zh" BREEZE25_CONFIRMATION_PIN_NAME = "breeze25-confirmation-auto" BREEZE25_PRIMARY_DECODER_PROFILE = "breeze25-greedy-zh-v1" BREEZE25_CONFIRMATION_DECODER_PROFILE = "breeze25-greedy-auto-v1" BREEZE25_PRIMARY_LANGUAGE = "zh" BREEZE25_CONFIRMATION_LANGUAGE = None BREEZE25_WEIGHT_SHA256 = getattr( quality_runtime_module, "BREEZE25_WEIGHT_SHA256", "c5d952b3bc03ea277209aff0ef5b5c4c055d74449ff794c02d8f4e315fdef6b6", ) BREEZE25_SAMPLE_RATE = getattr( quality_runtime_module, "BREEZE25_SAMPLE_RATE", 16_000, ) BREEZE25_ATTENTION_IMPLEMENTATION = getattr( quality_runtime_module, "BREEZE25_ATTENTION_IMPLEMENTATION", "eager", ) BREEZE25_RETURN_ATTENTION_MASK = getattr( quality_runtime_module, "BREEZE25_RETURN_ATTENTION_MASK", True, ) BREEZE25_MAX_MICROBATCH_SEGMENTS = getattr( quality_runtime_module, "BREEZE25_MAX_MICROBATCH_SEGMENTS", 6, ) def _unavailable_breeze25_transcriber(*args: Any, **kwargs: Any) -> str: del args, kwargs raise RuntimeError( "the pinned Breeze ASR 25 runtime is unavailable" ) transcribe_breeze25 = getattr( quality_runtime_module, "transcribe_breeze25", _unavailable_breeze25_transcriber, ) class GlyphHybridAdapterError(RuntimeError): """Raised when production data cannot satisfy hardened schema 7.""" @dataclass(frozen=True) class GlyphAdapterRuntime: """Exact hosted dependencies required for an independent rebuild.""" sample_rate: int inference_steps: int model_repo_id: str model_revision: str ecapa_repo_id: str ecapa_revision: str speaker_anchor_sha256: str speaker_centroid: Any generation_lock: Any generate_chunk: Callable[..., Any] generation_policy_resolver: Callable[[int], Any] generation_cfg_resolver: Callable[[int], float] speaker_encoder_factory: Callable[[], Any] speaker_anchor_factory: Callable[[], np.ndarray] gate_policy: evidence.GatePolicy asr_max_new_tokens: int asr_max_verification_segments: int renderer_builder: Callable[..., Any] = build_glyph_hybrid_plan assembly_verifier: Callable[..., None] = verify_hybrid_assembly carrier_conditioner: Callable[[CarrierSegment], bytes] = condition_carrier_v1 breeze_primary_transcriber: Callable[..., str] = transcribe_breeze25 breeze_confirmation_transcriber: Callable[..., str] = transcribe_breeze25 speaker_measure: Callable[..., Any] = release_speaker_evidence_from_audio squim_measure: Callable[..., Any] = squim_objective_evidence_from_audio def _canonical_json_bytes(value: Any) -> bytes: try: return json.dumps( value, ensure_ascii=True, allow_nan=False, separators=(",", ":"), sort_keys=True, ).encode("ascii") except (TypeError, ValueError) as exc: raise GlyphHybridAdapterError( f"adapter value is not canonical JSON: {exc}" ) from exc def _sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def _sha256_payload(value: Any) -> str: return _sha256_bytes(_canonical_json_bytes(value)) def _pcm_waveform(pcm16_le: bytes) -> np.ndarray: if type(pcm16_le) is not bytes or not pcm16_le or len(pcm16_le) % 2: raise GlyphHybridAdapterError("exact PCM must be non-empty PCM16 bytes") return ( np.frombuffer(pcm16_le, dtype=" str: result = str(value).strip().rstrip("/") for prefix in ("hf://", "https://huggingface.co/"): if result.casefold().startswith(prefix): result = result[len(prefix) :] break return result.casefold() def _select_hf_pin( pins: Sequence[Any], *, repo_id: str, revision: str, label: str, ) -> Any: expected_repo = _normalized_hf_uri(repo_id) matches = tuple( pin for pin in pins if _normalized_hf_uri(getattr(pin, "uri", "")) == expected_repo and getattr(pin, "revision", None) == revision ) if len(matches) != 1: raise GlyphHybridAdapterError( f"exact {label} artifact pin is unavailable" ) return matches[0] def _select_squim_pin(pins: Sequence[Any]) -> Any: matches = tuple( pin for pin in pins if getattr(pin, "sha256", None) == SQUIM_OBJECTIVE_WEIGHT_SHA256 and "squim" in str(getattr(pin, "name", "")).casefold() ) if len(matches) != 1: raise GlyphHybridAdapterError( "exact pinned SQUIM artifact is unavailable" ) return matches[0] def _artifact_pin( pin: Any, *, kind: str, config: Mapping[str, Any], logical_name: str | None = None, ) -> evidence.ArtifactPin: return evidence.ArtifactPin( name=str(logical_name if logical_name is not None else pin.name), kind=kind, uri=str(pin.uri), revision=str(pin.revision), sha256=str(pin.sha256), config_sha256=_sha256_payload(dict(config)), ) def _source_file_pin( name: str, path: str | Path, *, config: Mapping[str, Any], ) -> evidence.ArtifactPin: selected = Path(path).resolve(strict=True) payload = selected.read_bytes() digest = _sha256_bytes(payload) return evidence.ArtifactPin( name=name, kind="source", uri=f"repo://{selected.name}", revision=digest[:40], sha256=digest, config_sha256=_sha256_payload(dict(config)), ) def _runtime_pins( runtime: GlyphAdapterRuntime, bundle: Any, ) -> tuple[ tuple[evidence.ArtifactPin, ...], evidence.ArtifactPin, evidence.ArtifactPin, evidence.ArtifactPin, evidence.ArtifactPin, evidence.ArtifactPin, ]: model_pin = _select_hf_pin( tuple(bundle.model_pins), repo_id=runtime.model_repo_id, revision=runtime.model_revision, label="BlueMagpie model", ) breeze_weights_pin = _select_hf_pin( tuple(bundle.evaluator_pins), repo_id=BREEZE25_MODEL_ID, revision=BREEZE25_REVISION, label="Breeze ASR 25", ) if getattr(breeze_weights_pin, "sha256", None) != BREEZE25_WEIGHT_SHA256: raise GlyphHybridAdapterError( "exact Breeze ASR 25 weight pin is unavailable" ) ecapa_pin = _select_hf_pin( tuple(bundle.evaluator_pins), repo_id=runtime.ecapa_repo_id, revision=runtime.ecapa_revision, label="ECAPA", ) squim_pin = _select_squim_pin(tuple(bundle.evaluator_pins)) source_pins = tuple( _artifact_pin( pin, kind="source", config={ "schema": ADAPTER_SCHEMA, "role": "bundle-source", "artifact_sha256": str(pin.sha256), }, ) for pin in tuple(bundle.source_pins) ) + ( _source_file_pin( "glyph-ascii-runtime-source", glyph_ascii_module.__file__, config={ "schema": ADAPTER_SCHEMA, "role": "grammar-and-inverse-proof", }, ), _source_file_pin( "glyph-hybrid-plan-source", glyph_plan_module.__file__, config={ "schema": ADAPTER_SCHEMA, "role": "renderer", }, ), _source_file_pin( "glyph-assets-runtime-source", glyph_assets_module.__file__, config={ "schema": ADAPTER_SCHEMA, "role": "join-and-carrier-conditioner", }, ), _source_file_pin( "glyph-hybrid-evidence-source", evidence.__file__, config={ "schema": ADAPTER_SCHEMA, "role": "evidence-rebuild-and-validation", }, ), _source_file_pin( "glyph-hybrid-adapter-source", __file__, config={ "schema": ADAPTER_SCHEMA, "role": "production-evidence-adapter", }, ), _source_file_pin( "glyph-generation-runtime-source", runtime.generate_chunk.__code__.co_filename, config={ "schema": ADAPTER_SCHEMA, "role": "generation-replay", }, ), ) model = _artifact_pin( model_pin, kind="model", config={ "schema": ADAPTER_SCHEMA, "repo_id": runtime.model_repo_id, "revision": runtime.model_revision, "sample_rate": runtime.sample_rate, "inference_steps": runtime.inference_steps, }, ) shared_asr_config = { "schema": ADAPTER_SCHEMA, "task": ASR_TASK, "max_new_tokens": runtime.asr_max_new_tokens, "max_verification_segments": ( runtime.asr_max_verification_segments ), "sample_rate": BREEZE25_SAMPLE_RATE, "attention_implementation": BREEZE25_ATTENTION_IMPLEMENTATION, "return_attention_mask": BREEZE25_RETURN_ATTENTION_MASK, "max_microbatch_segments": BREEZE25_MAX_MICROBATCH_SEGMENTS, "do_sample": False, "num_beams": 1, "repo_id": BREEZE25_MODEL_ID, "revision": BREEZE25_REVISION, "shared_weights_sha256": str(breeze_weights_pin.sha256), } breeze_primary = _artifact_pin( breeze_weights_pin, kind="asr", logical_name=BREEZE25_PRIMARY_PIN_NAME, config={ **shared_asr_config, "decoder_profile": BREEZE25_PRIMARY_DECODER_PROFILE, "language": BREEZE25_PRIMARY_LANGUAGE, }, ) breeze_confirmation = _artifact_pin( breeze_weights_pin, kind="asr", logical_name=BREEZE25_CONFIRMATION_PIN_NAME, config={ **shared_asr_config, "decoder_profile": BREEZE25_CONFIRMATION_DECODER_PROFILE, "language": "auto", }, ) ecapa = _artifact_pin( ecapa_pin, kind="ecapa", config={ "schema": ADAPTER_SCHEMA, "repo_id": runtime.ecapa_repo_id, "revision": runtime.ecapa_revision, "device": "cpu", "measurement": "release-full-plus-thirds-v1", "active_top_db": runtime.gate_policy.active_voice_top_db, "speaker_anchor_sha256": runtime.speaker_anchor_sha256, }, ) squim = _artifact_pin( squim_pin, kind="squim", config={ "schema": ADAPTER_SCHEMA, "weight_sha256": SQUIM_OBJECTIVE_WEIGHT_SHA256, "sample_rate": SQUIM_OBJECTIVE_SAMPLE_RATE, "window_seconds": SQUIM_OBJECTIVE_WINDOW_SECONDS, "max_windows": SQUIM_OBJECTIVE_MAX_WINDOWS, "device": "cpu", }, ) return ( source_pins, model, breeze_primary, breeze_confirmation, ecapa, squim, ) def _trusted_bundle( bundle: Any, manifest_entries: Mapping[str, str], ) -> evidence.TrustedGlyphBundle: assets: list[evidence.TrustedAsset] = [] for asset in tuple(bundle.assets): provisional = evidence.TrustedAsset( asset_id=str(asset.asset_id), raw_text=str(asset.raw_byte), canonical_text=str(asset.canonical_token), manifest_entry_sha256="0" * 64, pcm16_le=bytes(asset.pcm16_le), ) entry_sha256 = evidence.compute_trusted_asset_entry_sha256( provisional ) if manifest_entries.get(provisional.asset_id) != entry_sha256: raise GlyphHybridAdapterError( "profile manifest entry does not match exact asset bytes" ) assets.append( evidence.TrustedAsset( **{ **provisional.__dict__, "manifest_entry_sha256": entry_sha256, } ) ) manifest_sha256, bundle_sha256 = evidence.compute_trusted_bundle_digests( tuple(assets) ) return evidence.TrustedGlyphBundle( manifest_sha256=manifest_sha256, bundle_sha256=bundle_sha256, assets=tuple(assets), ) def _profile_sha256(plan: Any, bundle: Any) -> str: return _sha256_payload( { "schema": ADAPTER_SCHEMA, "profile_id": str(plan.profile_id), "normalization_id": str(plan.normalization_id), "boundary_contract_id": str(plan.boundary_contract_id), "grammar_id": str(plan.grammar_id), "grammar_sha256": str(bundle.grammar_sha256), } ) def _join_sha256(bundle: Any) -> str: return _sha256_payload(dict(bundle.join_profile)) def _trusted_semantic_plan( raw_text: str, plan: Any, bundle: Any, ) -> evidence.TrustedRendererPlan: segments = tuple( evidence.TrustedRenderSegment( segment_id=f"renderer-segment-{index}", kind=( "asset" if segment.kind == "asset_atom" else "carrier" if segment.kind == "carrier" else "control" ), raw_start=int(segment.absolute_raw_start), raw_end=int(segment.absolute_raw_end), canonical_start=int(segment.canonical_start), canonical_end=int(segment.canonical_end), raw_text=str(segment.raw_text), canonical_text=str(segment.canonical_text), output_start_sample=0, output_end_sample=0, asset_id=( str(segment.asset_id) if segment.kind == "asset_atom" else None ), ) for index, segment in enumerate(tuple(plan.segments)) ) return evidence.TrustedRendererPlan( grammar_id=str(plan.grammar_id), grammar_sha256=str(bundle.grammar_sha256), profile_id=str(plan.profile_id), profile_sha256=_profile_sha256(plan, bundle), join_profile_id=str(bundle.join_profile["id"]), join_profile_sha256=_join_sha256(bundle), raw_input=raw_text, canonical_target=str(plan.canonical_target), segments=segments, ) def _trusted_realized_plan( raw_text: str, plan: Any, bundle: Any, assembly: Any, ) -> evidence.TrustedRendererPlan: ledger_spans = tuple(assembly.ledger.spans) span_index = 0 output_cursor = 0 result: list[evidence.TrustedRenderSegment] = [] def add_silence(span: Any, raw_cursor: int, canonical_cursor: int) -> None: nonlocal output_cursor if ( span.kind not in {"pause", "final_silence"} or int(span.start_sample) != output_cursor or int(span.end_sample) <= output_cursor ): raise GlyphHybridAdapterError( "assembly join-silence provenance is invalid" ) result.append( evidence.TrustedRenderSegment( segment_id=f"join-silence-{len(result)}", kind="silence", raw_start=raw_cursor, raw_end=raw_cursor, canonical_start=canonical_cursor, canonical_end=canonical_cursor, raw_text="", canonical_text="", output_start_sample=output_cursor, output_end_sample=int(span.end_sample), ) ) output_cursor = int(span.end_sample) for segment in tuple(plan.segments): raw_start = int(segment.absolute_raw_start) raw_end = int(segment.absolute_raw_end) canonical_start = int(segment.canonical_start) canonical_end = int(segment.canonical_end) if segment.kind in {"control", "separator"}: result.append( evidence.TrustedRenderSegment( segment_id=f"plan-control-{segment.ordinal}", kind="control", raw_start=raw_start, raw_end=raw_end, canonical_start=canonical_start, canonical_end=canonical_end, raw_text=str(segment.raw_text), canonical_text=str(segment.canonical_text), output_start_sample=output_cursor, output_end_sample=output_cursor, ) ) continue while ( span_index < len(ledger_spans) and ledger_spans[span_index].kind == "pause" ): add_silence( ledger_spans[span_index], raw_start, canonical_start, ) span_index += 1 if span_index >= len(ledger_spans): raise GlyphHybridAdapterError( "renderer audio segment is absent from assembly ledger" ) span = ledger_spans[span_index] expected_kind = ( "asset" if segment.kind == "asset_atom" else "carrier" ) expected_segment_id = f"glyph-plan-segment-{segment.ordinal}" if ( span.kind != expected_kind or span.segment_id != expected_segment_id or int(span.start_sample) != output_cursor or int(span.end_sample) <= output_cursor ): raise GlyphHybridAdapterError( "renderer and assembly segment provenance disagree" ) result.append( evidence.TrustedRenderSegment( segment_id=expected_segment_id, kind=expected_kind, raw_start=raw_start, raw_end=raw_end, canonical_start=canonical_start, canonical_end=canonical_end, raw_text=str(segment.raw_text), canonical_text=str(segment.canonical_text), output_start_sample=output_cursor, output_end_sample=int(span.end_sample), asset_id=( str(segment.asset_id) if expected_kind == "asset" else None ), ) ) output_cursor = int(span.end_sample) span_index += 1 while span_index < len(ledger_spans): add_silence( ledger_spans[span_index], len(raw_text), len(str(plan.canonical_target)), ) span_index += 1 if output_cursor != len(assembly.pcm16_le) // 2: raise GlyphHybridAdapterError( "renderer evidence does not cover exact assembled PCM" ) semantic = _trusted_semantic_plan(raw_text, plan, bundle) return evidence.TrustedRendererPlan( grammar_id=semantic.grammar_id, grammar_sha256=semantic.grammar_sha256, profile_id=semantic.profile_id, profile_sha256=semantic.profile_sha256, join_profile_id=semantic.join_profile_id, join_profile_sha256=semantic.join_profile_sha256, raw_input=semantic.raw_input, canonical_target=semantic.canonical_target, segments=tuple(result), ) def _policy_payload( runtime: GlyphAdapterRuntime, reservation: Any, model_pin: evidence.ArtifactPin, ) -> tuple[str, str, Any, float]: candidate_ordinal = int(reservation.candidate_ordinal) network_conditioned = bool(reservation.network_conditioned) policy = runtime.generation_policy_resolver(candidate_ordinal) cfg = float(runtime.generation_cfg_resolver(candidate_ordinal)) if not math.isfinite(cfg): raise GlyphHybridAdapterError("generation replay CFG is invalid") if is_dataclass(policy): policy_value = asdict(policy) elif isinstance(getattr(policy, "__dict__", None), dict): policy_value = dict(policy.__dict__) else: raise GlyphHybridAdapterError( "generation policy is not canonically serializable" ) payload = { "schema": GENERATION_POLICY_SCHEMA, "model_pin_name": model_pin.name, "model_sha256": model_pin.sha256, "candidate_ordinal": candidate_ordinal, "network_conditioned": network_conditioned, "scheduled_cfg": cfg, "inference_steps": runtime.inference_steps, "sample_rate": runtime.sample_rate, "policy": policy_value, } digest = _sha256_payload(payload) return f"generation-policy-{digest[:24]}", digest, policy, cfg def _conditioned_attempt_pcm( runtime: GlyphAdapterRuntime, completed: Any, outcome: Any, ) -> bytes: return runtime.carrier_conditioner( CarrierSegment( segment_id=str(completed.reservation.carrier_segment_id), audio=np.asarray(outcome.audio, dtype=np.float32).reshape(-1), selected_attempt_id=str(completed.record.attempt_id), selected_seed=int(completed.record.seed), selected_text_sha256=str(completed.record.text_sha256), selected_endpoint_evidence_sha256=str( completed.record.endpoint_evidence_sha256 ), sample_rate=runtime.sample_rate, speed=1.0, ) ) def _endpoint( pcm16_le: bytes, outcome: Any, policy: evidence.GatePolicy, ) -> evidence.EndpointEvidence: return evidence.EndpointEvidence( stop_reason=str(outcome.stop_reason), tail_energy_ratio=evidence.measure_endpoint_tail_energy_ratio( pcm16_le, policy, ), terminal_silence_samples=( evidence.measure_terminal_silence_samples( pcm16_le, policy, ) ), generated_steps=int(outcome.generated_steps), hard_cap_steps=int(outcome.hard_stop_steps), ) def _verification_result(value: Any, *, scope: str) -> Any: verification = getattr(value, "verification", value) if getattr(verification, "passed", None) is not True: raise GlyphHybridAdapterError( f"{scope} production verification did not pass" ) rows = tuple(getattr(verification, "candidate_results", ())) if len(rows) != 1: raise GlyphHybridAdapterError( f"{scope} production verification is not whole-output" ) return rows[0] class HardenedGlyphHybridAdapter: """Callable adapter installed into ``app._GLYPH_HYBRID_EVIDENCE_ADAPTER``.""" def __init__(self, runtime: GlyphAdapterRuntime) -> None: if type(runtime) is not GlyphAdapterRuntime: raise TypeError("runtime must be GlyphAdapterRuntime") self._runtime = runtime def __call__( self, *, raw_text: str, plan: Any, profile_state: Any, segments: tuple[AssetSegment | CarrierSegment, ...], assembly: Any, semantic_commitment: Any, request_ledger: Any, final_verification: Any, independent_final_verification: Any, ) -> bytes: runtime = self._runtime if ( type(raw_text) is not str or not raw_text or runtime.sample_rate != int(assembly.sample_rate) or runtime.gate_policy.sample_rate != runtime.sample_rate or runtime.asr_max_new_tokens <= 0 or runtime.asr_max_verification_segments <= 0 ): raise GlyphHybridAdapterError( "adapter request or runtime contract is invalid" ) if profile_state.bundle is None: raise GlyphHybridAdapterError("trusted glyph bundle is unavailable") bundle = profile_state.bundle manifest_entries = dict( profile_state.asset_entry_sha256_by_asset_id ) runtime.assembly_verifier( assembly, trusted_bundle=bundle, original_segments=segments, expected_semantic_commitment=semantic_commitment, ) final_row = _verification_result( final_verification, scope="final", ) _verification_result( independent_final_verification, scope="independent", ) trusted_bundle = _trusted_bundle(bundle, manifest_entries) trusted_plan = _trusted_realized_plan( raw_text, plan, bundle, assembly, ) ( source_pins, model_pin, breeze_primary_pin, breeze_confirmation_pin, ecapa_pin, squim_pin, ) = _runtime_pins(runtime, bundle) anchor = np.asarray( runtime.speaker_anchor_factory(), dtype=" tuple[bytes, evidence.EndpointEvidence]: key = ( pin.name, text, seed, policy_id, policy_sha256, sample_rate, ) cached = replay_cache.get(key) if cached is not None: return cached replay_items = replay_by_key.get(key) if replay_items is None or pin != model_pin: raise GlyphHybridAdapterError( "model replay request is outside the trusted ledger" ) replayed: list[tuple[bytes, evidence.EndpointEvidence]] = [] for item, policy, cfg in replay_items: reservation = item.reservation try: with runtime.generation_lock: outcome = runtime.generate_chunk( str(reservation.text), runtime.speaker_centroid, cfg=cfg, steps=runtime.inference_steps, request_seed=int(reservation.seed), policy=policy, network_conditioned=bool( reservation.network_conditioned ), ) except Exception as exc: raise GlyphHybridAdapterError( "deterministic model replay failed" ) from exc replayed_pcm = _conditioned_attempt_pcm( runtime, item, outcome, ) replayed.append( ( replayed_pcm, _endpoint( replayed_pcm, outcome, runtime.gate_policy, ), ) ) result = replayed[0] if any(item != result for item in replayed[1:]): raise GlyphHybridAdapterError( "identical generation inputs replayed inconsistently" ) replay_cache[key] = result return result asset_by_id = { asset.asset_id: asset for asset in trusted_bundle.assets } asset_calls: list[evidence.TrustedAssetLoadCall] = [] for index, segment in enumerate(segments): if not isinstance(segment, AssetSegment): continue asset = asset_by_id.get(segment.asset_id) if asset is None: raise GlyphHybridAdapterError( "assembled asset is absent from trusted bundle" ) asset_calls.append( evidence.TrustedAssetLoadCall( call_id=f"asset-load-{index}", segment_id=str(segment.segment_id), asset_id=str(segment.asset_id), manifest_entry_sha256=asset.manifest_entry_sha256, loaded_pcm16_le=asset.pcm16_le, ) ) selections: list[evidence.TrustedCarrierSelection] = [] carrier_rank = 0 for segment in segments: if not isinstance(segment, CarrierSegment): continue carrier_rank += 1 attempt_ids = tuple( attempts_by_segment.get(str(segment.segment_id), ()) ) if segment.selected_attempt_id not in attempt_ids: raise GlyphHybridAdapterError( "selected carrier has no replayable attempt" ) selections.append( evidence.TrustedCarrierSelection( segment_id=str(segment.segment_id), path_rank=carrier_rank, attempted_call_ids=attempt_ids, selected_call_id=str(segment.selected_attempt_id), ) ) if bool(completed) != bool(selections): raise GlyphHybridAdapterError( "pure-asset/model-work accounting is inconsistent" ) expected_pins = { breeze_primary_pin.name: breeze_primary_pin, breeze_confirmation_pin.name: breeze_confirmation_pin, ecapa_pin.name: ecapa_pin, squim_pin.name: squim_pin, } asr_cache: dict[tuple[str, str, int], str] = {} ecapa_cache: dict[ tuple[str, str, int, str], tuple[float, float, float], ] = {} squim_cache: dict[ tuple[str, str, int], tuple[float, float, float], ] = {} def asr_evaluator( pin: evidence.ArtifactPin, pcm16_le: bytes, sample_rate: int, ) -> str: if expected_pins.get(pin.name) != pin: raise GlyphHybridAdapterError("ASR pin changed during rebuild") key = pin.name, _sha256_bytes(pcm16_le), sample_rate if key not in asr_cache: waveform = _pcm_waveform(pcm16_le) if pin == breeze_primary_pin: transcriber = runtime.breeze_primary_transcriber language = BREEZE25_PRIMARY_LANGUAGE elif pin == breeze_confirmation_pin: transcriber = runtime.breeze_confirmation_transcriber language = BREEZE25_CONFIRMATION_LANGUAGE else: raise GlyphHybridAdapterError( "ASR evaluator kind is not pinned" ) transcript = transcriber( waveform, sample_rate, language=language, task=ASR_TASK, max_new_tokens=runtime.asr_max_new_tokens, max_verification_segments=( runtime.asr_max_verification_segments ), ) if type(transcript) is not str: raise GlyphHybridAdapterError( "ASR evaluator returned a non-string transcript" ) asr_cache[key] = transcript.strip() return asr_cache[key] def ecapa_evaluator( pin: evidence.ArtifactPin, pcm16_le: bytes, sample_rate: int, speaker_anchor_sha256: str, ) -> tuple[float, float, float]: if ( pin != ecapa_pin or speaker_anchor_sha256 != runtime.speaker_anchor_sha256 ): raise GlyphHybridAdapterError( "ECAPA pin or speaker anchor changed" ) key = ( pin.name, _sha256_bytes(pcm16_le), sample_rate, speaker_anchor_sha256, ) if key not in ecapa_cache: measured = runtime.speaker_measure( _pcm_waveform(pcm16_le), sample_rate, runtime.speaker_encoder_factory(), anchor, device="cpu", active_top_db=( runtime.gate_policy.active_voice_top_db ), ) values = ( float(measured.similarity), float(measured.begin_similarity), float(measured.end_similarity), ) if not all(math.isfinite(value) for value in values): raise GlyphHybridAdapterError( "ECAPA returned non-finite evidence" ) ecapa_cache[key] = values return ecapa_cache[key] def squim_evaluator( pin: evidence.ArtifactPin, pcm16_le: bytes, sample_rate: int, ) -> tuple[float, float, float]: if pin != squim_pin: raise GlyphHybridAdapterError("SQUIM pin changed") key = pin.name, _sha256_bytes(pcm16_le), sample_rate if key not in squim_cache: measured = runtime.squim_measure( _pcm_waveform(pcm16_le), sample_rate, ) values = ( float(measured.stoi), float(measured.pesq), float(measured.si_sdr), ) if not all(math.isfinite(value) for value in values): raise GlyphHybridAdapterError( "SQUIM returned non-finite evidence" ) squim_cache[key] = values return squim_cache[key] output_pcm = bytes(assembly.pcm16_le) breeze_primary_transcript = asr_evaluator( breeze_primary_pin, output_pcm, runtime.sample_rate, ) breeze_confirmation_transcript = asr_evaluator( breeze_confirmation_pin, output_pcm, runtime.sample_rate, ) speaker_values = ecapa_evaluator( ecapa_pin, output_pcm, runtime.sample_rate, runtime.speaker_anchor_sha256, ) squim_values = squim_evaluator( squim_pin, output_pcm, runtime.sample_rate, ) final_comparison = getattr(final_row, "comparison", None) if ( final_comparison is None or not breeze_primary_transcript or not breeze_confirmation_transcript ): raise GlyphHybridAdapterError( "semantic evaluator output is unavailable" ) calls: list[evidence.TrustedRuntimeCall] = [ *model_calls, *asset_calls, ] gate_bindings: list[evidence.TrustedGateBinding] = [] active_samples = evidence.measure_active_speech_samples( output_pcm, runtime.gate_policy, ) for scope, asr_pin, transcript in ( ( "final", breeze_primary_pin, breeze_primary_transcript, ), ( "independent", breeze_confirmation_pin, breeze_confirmation_transcript, ), ): asr_call_id = f"asr-{scope}" ecapa_call_id = f"ecapa-{scope}" squim_call_id = f"squim-{scope}" calls.extend( ( evidence.TrustedAsrCall( call_id=asr_call_id, scope=scope, evaluator_pin_name=asr_pin.name, input_pcm16_le=output_pcm, target_text=str(plan.canonical_target), transcript_text=transcript, ), evidence.TrustedEcapaCall( call_id=ecapa_call_id, scope=scope, evaluator_pin_name=ecapa_pin.name, input_pcm16_le=output_pcm, speaker_anchor_sha256=( runtime.speaker_anchor_sha256 ), centroid_similarity=speaker_values[0], begin_similarity=speaker_values[1], end_similarity=speaker_values[2], ), evidence.TrustedSquimCall( call_id=squim_call_id, scope=scope, evaluator_pin_name=squim_pin.name, input_pcm16_le=output_pcm, stoi=squim_values[0], pesq=squim_values[1], si_sdr=squim_values[2], ), ) ) gate_bindings.append( evidence.TrustedGateBinding( scope=scope, asr_call_id=asr_call_id, ecapa_call_id=ecapa_call_id, squim_call_id=squim_call_id, active_speech_samples=active_samples, ) ) def renderer_evaluator( contract: evidence.RendererContract, rebuilt_bundle: evidence.TrustedGlyphBundle, request_text: str, ) -> evidence.TrustedRendererPlan: if ( contract.manifest_sha256 != rebuilt_bundle.manifest_sha256 or contract.bundle_sha256 != rebuilt_bundle.bundle_sha256 or rebuilt_bundle != trusted_bundle ): raise GlyphHybridAdapterError( "renderer rebuild bundle changed" ) fresh = runtime.renderer_builder( request_text, asset_entry_sha256_by_asset_id={ asset.asset_id: asset.manifest_entry_sha256 for asset in rebuilt_bundle.assets }, ) if fresh is None: raise GlyphHybridAdapterError( "pinned renderer did not reproduce the request" ) return _trusted_semantic_plan( request_text, fresh, bundle, ) generation_policy_items = tuple(sorted(policies.items())) gate_policy_sha256 = _sha256_payload( asdict(runtime.gate_policy) ) contract = evidence.RendererContract( grammar_id=trusted_plan.grammar_id, grammar_sha256=trusted_plan.grammar_sha256, profile_id=trusted_plan.profile_id, profile_sha256=trusted_plan.profile_sha256, manifest_sha256=trusted_bundle.manifest_sha256, bundle_sha256=trusted_bundle.bundle_sha256, join_profile_id=trusted_plan.join_profile_id, join_profile_sha256=trusted_plan.join_profile_sha256, gate_policy_sha256=gate_policy_sha256, speaker_anchor_sha256=runtime.speaker_anchor_sha256, generation_policy_ids=tuple( item[0] for item in generation_policy_items ), generation_policy_sha256s=tuple( item[1] for item in generation_policy_items ), source_pins=source_pins, model_pins=(model_pin,), evaluator_pins=( breeze_primary_pin, breeze_confirmation_pin, ecapa_pin, squim_pin, ), ) rebuild_inputs = evidence.EvidenceRebuildInputs( contract=contract, plan=trusted_plan, bundle=trusted_bundle, assembled_pcm16_le=output_pcm, public_pcm16_le=bytes(assembly.public_pcm16_le), calls=tuple(calls), carrier_selections=tuple(selections), gate_bindings=tuple(gate_bindings), content_commitment_key=secrets.token_bytes(32), generation_evaluator=generation_evaluator, renderer_evaluator=renderer_evaluator, asr_evaluator=asr_evaluator, ecapa_evaluator=ecapa_evaluator, squim_evaluator=squim_evaluator, budget_caps=evidence.BudgetCaps( model_invocations=32, generated_chunks=32, generated_speech_units=800, ), gate_policy=runtime.gate_policy, ) serialized = evidence.format_hybrid_evidence(rebuild_inputs) evidence.validate_hybrid_evidence(serialized, rebuild_inputs) return serialized __all__ = ( "ADAPTER_SCHEMA", "GlyphHybridAdapterError", "GlyphAdapterRuntime", "HardenedGlyphHybridAdapter", )