from __future__ import annotations from dataclasses import asdict, replace import hashlib import json import random import numpy as np import pytest import glyph_hybrid_evidence as evidence_module from glyph_hybrid_evidence import ( EVIDENCE_SCHEMA, EVIDENCE_VERSION, ArtifactPin, BudgetCaps, EndpointEvidence, EvidenceRebuildInputs, GatePolicy, GlyphHybridEvidenceError, RendererContract, TrustedAsset, TrustedAssetLoadCall, TrustedAsrCall, TrustedCarrierSelection, TrustedEcapaCall, TrustedGateBinding, TrustedGlyphBundle, TrustedModelGenerationCall, TrustedRenderSegment, TrustedRendererPlan, TrustedSquimCall, compute_trusted_asset_entry_sha256, compute_trusted_bundle_digests, format_hybrid_evidence, hybrid_evidence_sha256, rebuild_hybrid_evidence, validate_hybrid_evidence, ) def _sha(value: bytes | str) -> str: if isinstance(value, str): value = value.encode("utf-8") return hashlib.sha256(value).hexdigest() def _policy_sha(policy: GatePolicy) -> str: return _sha( json.dumps( asdict(policy), ensure_ascii=True, allow_nan=False, separators=(",", ":"), sort_keys=True, ).encode("ascii") ) def _pcm( sample_count: int, amplitude: int, phase: int = 0, terminal_silence_samples: int = 0, ) -> bytes: indexes = np.arange(sample_count, dtype=np.int64) values = np.where((indexes + phase) % 2, amplitude, -amplitude) if terminal_silence_samples: values[-terminal_silence_samples:] = 0 return values.astype(" tuple[bytes, EndpointEvidence]: assert pin.kind == "model" assert 0 <= seed < 2**32 assert policy_id == "quality-primary" assert policy_sha256 == GENERATION_POLICY_SHA256 assert sample_rate == 48_000 amplitude = {"甲": 4_000, "乙": 4_500}[text] return ( _pcm( 48_000, amplitude, terminal_silence_samples=480, ), EndpointEvidence( stop_reason="stop_threshold", tail_energy_ratio=0.0, terminal_silence_samples=480, generated_steps=4, hard_cap_steps=5, ), ) def _asr_evaluator( pin: ArtifactPin, pcm16_le: bytes, sample_rate: int, ) -> str: assert pin.kind == "asr" assert sample_rate == 48_000 samples = np.frombuffer(pcm16_le, dtype=" tuple[float, float, float]: assert pin.kind == "ecapa" assert pcm16_le assert sample_rate == 48_000 assert speaker_anchor_sha256 == SPEAKER_ANCHOR_SHA256 return 0.55, 0.52, 0.50 def _squim_evaluator( pin: ArtifactPin, pcm16_le: bytes, sample_rate: int, ) -> tuple[float, float, float]: assert pin.kind == "squim" assert pcm16_le assert sample_rate == 48_000 return 0.91, 1.55, 4.0 def _renderer_evaluator( contract: RendererContract, bundle: TrustedGlyphBundle, raw_input: str, ) -> TrustedRendererPlan: assert len(bundle.assets) == 1 asset = bundle.assets[0] asset_start = raw_input.index(asset.raw_text) asset_end = asset_start + len(asset.raw_text) left = raw_input[:asset_start] right = raw_input[asset_end:] canonical = left + asset.canonical_text + right segments = ( TrustedRenderSegment( segment_id="rebuilt-left", kind="carrier", raw_start=0, raw_end=asset_start, canonical_start=0, canonical_end=len(left), raw_text=left, canonical_text=left, output_start_sample=0, output_end_sample=0, ), TrustedRenderSegment( segment_id="rebuilt-asset", kind="asset", raw_start=asset_start, raw_end=asset_end, canonical_start=len(left), canonical_end=len(left) + len(asset.canonical_text), raw_text=asset.raw_text, canonical_text=asset.canonical_text, output_start_sample=0, output_end_sample=0, asset_id=asset.asset_id, ), TrustedRenderSegment( segment_id="rebuilt-right", kind="carrier", raw_start=asset_end, raw_end=len(raw_input), canonical_start=len(left) + len(asset.canonical_text), canonical_end=len(canonical), raw_text=right, canonical_text=right, output_start_sample=0, output_end_sample=0, ), ) return TrustedRendererPlan( grammar_id=contract.grammar_id, grammar_sha256=contract.grammar_sha256, profile_id=contract.profile_id, profile_sha256=contract.profile_sha256, join_profile_id=contract.join_profile_id, join_profile_sha256=contract.join_profile_sha256, raw_input=raw_input, canonical_target=canonical, segments=segments, ) def _pin(name: str) -> ArtifactPin: if name == "source": kind = "source" elif name == "tts-model": kind = "model" elif name.startswith("asr-"): kind = "asr" else: kind = name return ArtifactPin( name=name, kind=kind, uri=f"hf://frozen/{name}", revision=f"{name}-revision", sha256=_sha(f"{name}:artifact"), config_sha256=_sha(f"{name}:config"), ) def _breeze25_profile_pin( name: str, decoder_profile: str, ) -> ArtifactPin: return ArtifactPin( name=name, kind="asr", uri=f"hf://{BREEZE25_MODEL_ID}", revision=BREEZE25_REVISION, sha256=BREEZE25_WEIGHT_SHA256, config_sha256=_sha(f"breeze25:{decoder_profile}"), ) def _inputs() -> EvidenceRebuildInputs: raw = "甲a乙" canonical = "甲第一個英文字母乙" carrier_left = _pcm( 48_000, 4_000, terminal_silence_samples=480, ) asset_pcm = _pcm(24_000, 3_000, phase=1) carrier_right = _pcm( 48_000, 4_500, terminal_silence_samples=480, ) output = carrier_left + asset_pcm + carrier_right asset = TrustedAsset( asset_id="letter-a", raw_text="a", canonical_text="第一個英文字母", manifest_entry_sha256="0" * 64, pcm16_le=asset_pcm, ) asset = replace( asset, manifest_entry_sha256=compute_trusted_asset_entry_sha256(asset), ) manifest_sha256, bundle_sha256 = compute_trusted_bundle_digests( (asset,) ) bundle = TrustedGlyphBundle( manifest_sha256=manifest_sha256, bundle_sha256=bundle_sha256, assets=(asset,), ) grammar_sha = _sha("grammar") profile_sha = _sha("profile") join_sha = _sha("join") policy = GatePolicy() contract = RendererContract( grammar_id="ordinal-glyph-v1", grammar_sha256=grammar_sha, profile_id="hybrid-request-v1", profile_sha256=profile_sha, manifest_sha256=bundle.manifest_sha256, bundle_sha256=bundle.bundle_sha256, join_profile_id="pcm16-exact-v1", join_profile_sha256=join_sha, gate_policy_sha256=_policy_sha(policy), speaker_anchor_sha256=SPEAKER_ANCHOR_SHA256, generation_policy_ids=("quality-primary",), generation_policy_sha256s=(GENERATION_POLICY_SHA256,), source_pins=(_pin("source"),), model_pins=(_pin("tts-model"),), evaluator_pins=( _breeze25_profile_pin( BREEZE25_PRIMARY_PIN_NAME, "fixed-zh", ), _breeze25_profile_pin( BREEZE25_CONFIRMATION_PIN_NAME, "auto-language", ), _pin("ecapa"), _pin("squim"), ), ) segments = ( TrustedRenderSegment( segment_id="carrier-0", kind="carrier", raw_start=0, raw_end=1, canonical_start=0, canonical_end=1, raw_text="甲", canonical_text="甲", output_start_sample=0, output_end_sample=48_000, ), TrustedRenderSegment( segment_id="asset-0", kind="asset", raw_start=1, raw_end=2, canonical_start=1, canonical_end=8, raw_text="a", canonical_text="第一個英文字母", output_start_sample=48_000, output_end_sample=72_000, asset_id="letter-a", ), TrustedRenderSegment( segment_id="carrier-1", kind="carrier", raw_start=2, raw_end=3, canonical_start=8, canonical_end=9, raw_text="乙", canonical_text="乙", output_start_sample=72_000, output_end_sample=120_000, ), ) plan = TrustedRendererPlan( grammar_id=contract.grammar_id, grammar_sha256=grammar_sha, profile_id=contract.profile_id, profile_sha256=profile_sha, join_profile_id=contract.join_profile_id, join_profile_sha256=join_sha, raw_input=raw, canonical_target=canonical, segments=segments, ) endpoint = EndpointEvidence( stop_reason="stop_threshold", tail_energy_ratio=0.0, terminal_silence_samples=480, generated_steps=4, hard_cap_steps=5, ) calls = ( TrustedModelGenerationCall( call_id="model-0", segment_id="carrier-0", attempt_ordinal=0, text="甲", seed=45_678, policy_id="quality-primary", policy_sha256=GENERATION_POLICY_SHA256, model_pin_name="tts-model", output_pcm16_le=carrier_left, endpoint=endpoint, ), TrustedModelGenerationCall( call_id="model-1", segment_id="carrier-1", attempt_ordinal=0, text="乙", seed=45_678, policy_id="quality-primary", policy_sha256=GENERATION_POLICY_SHA256, model_pin_name="tts-model", output_pcm16_le=carrier_right, endpoint=endpoint, ), TrustedAssetLoadCall( call_id="asset-load-0", segment_id="asset-0", asset_id="letter-a", manifest_entry_sha256=asset.manifest_entry_sha256, loaded_pcm16_le=asset_pcm, ), TrustedAsrCall( call_id="asr-final", scope="final", evaluator_pin_name=BREEZE25_PRIMARY_PIN_NAME, input_pcm16_le=output, target_text=canonical, transcript_text=canonical, ), TrustedEcapaCall( call_id="ecapa-final", scope="final", evaluator_pin_name="ecapa", input_pcm16_le=output, speaker_anchor_sha256=SPEAKER_ANCHOR_SHA256, centroid_similarity=0.55, begin_similarity=0.52, end_similarity=0.50, ), TrustedSquimCall( call_id="squim-final", scope="final", evaluator_pin_name="squim", input_pcm16_le=output, stoi=0.91, pesq=1.55, si_sdr=4.0, ), TrustedAsrCall( call_id="asr-independent", scope="independent", evaluator_pin_name=BREEZE25_CONFIRMATION_PIN_NAME, input_pcm16_le=output, target_text=canonical, transcript_text=canonical, ), TrustedEcapaCall( call_id="ecapa-independent", scope="independent", evaluator_pin_name="ecapa", input_pcm16_le=output, speaker_anchor_sha256=SPEAKER_ANCHOR_SHA256, centroid_similarity=0.55, begin_similarity=0.52, end_similarity=0.50, ), TrustedSquimCall( call_id="squim-independent", scope="independent", evaluator_pin_name="squim", input_pcm16_le=output, stoi=0.91, pesq=1.55, si_sdr=4.0, ), ) active_speech_samples = evidence_module._active_speech_samples( output, policy, ) return EvidenceRebuildInputs( contract=contract, plan=plan, bundle=bundle, assembled_pcm16_le=output, public_pcm16_le=output, calls=calls, carrier_selections=( TrustedCarrierSelection( segment_id="carrier-0", path_rank=1, attempted_call_ids=("model-0",), selected_call_id="model-0", ), TrustedCarrierSelection( segment_id="carrier-1", path_rank=1, attempted_call_ids=("model-1",), selected_call_id="model-1", ), ), gate_bindings=( TrustedGateBinding( scope="final", asr_call_id="asr-final", ecapa_call_id="ecapa-final", squim_call_id="squim-final", active_speech_samples=active_speech_samples, ), TrustedGateBinding( scope="independent", asr_call_id="asr-independent", ecapa_call_id="ecapa-independent", squim_call_id="squim-independent", active_speech_samples=active_speech_samples, ), ), content_commitment_key=CONTENT_COMMITMENT_KEY, generation_evaluator=_generation_evaluator, renderer_evaluator=_renderer_evaluator, asr_evaluator=_asr_evaluator, ecapa_evaluator=_ecapa_evaluator, squim_evaluator=_squim_evaluator, gate_policy=policy, ) def _canonical_mutation(serialized: bytes, mutate) -> bytes: value = json.loads(serialized) mutate(value) return json.dumps( value, ensure_ascii=True, allow_nan=False, separators=(",", ":"), sort_keys=True, ).encode("ascii") def test_rebuild_formats_content_free_schema7_and_validates_independently(): inputs = _inputs() evidence = rebuild_hybrid_evidence(inputs) serialized = format_hybrid_evidence(inputs) validated = validate_hybrid_evidence(serialized, inputs) assert evidence == validated assert evidence.schema == EVIDENCE_SCHEMA assert evidence.version == EVIDENCE_VERSION assert evidence.request_proof.final_gate.passed is True assert evidence.request_proof.independent_gate.passed is True assert evidence.request_proof.additional_strict_gates == replace( evidence.request_proof.additional_strict_gates, strict_inverse_reconstruction=True, asset_reconstruction=True, public_pcm_equality=True, ) assert hybrid_evidence_sha256(serialized, inputs) == _sha(serialized) decoded = serialized.decode("ascii") for secret in ( "甲", "乙", "第一個英文字母", "\\u7532", "\\u4e59", ( "\\u7b2c\\u4e00\\u500b\\u82f1" "\\u6587\\u5b57\\u6bcd" ), "letter-a", "hf://frozen/", "tts-model-revision", ): assert secret not in decoded assert ( evidence.request_proof.content_commitment_scheme == "hmac-sha256-v1" ) assert evidence.request_proof.raw_input_sha256 != _sha( inputs.plan.raw_input ) assert evidence.request_proof.canonical_target_sha256 != _sha( inputs.plan.canonical_target ) public_payload = json.loads(serialized) assert evidence.renderer_contract_sha256 == _sha( json.dumps( public_payload["renderer_contract"], ensure_ascii=True, allow_nan=False, separators=(",", ":"), sort_keys=True, ).encode("ascii") ) call_by_id = { item.call_id: item for item in evidence.runtime_call_ledger } for gate in ( evidence.request_proof.final_gate, evidence.request_proof.independent_gate, ): assert ( call_by_id[gate.asr_call_id].target_sha256 == evidence.request_proof.canonical_target_sha256 ) segment_by_id = { item.segment_id: item for item in evidence.request_proof.segments } asset_segment = next( item for item in segment_by_id.values() if item.kind == "asset" ) asset_load = next( item for item in evidence.runtime_call_ledger if item.kind == "asset_load" ) assert ( asset_load.asset_id, asset_load.manifest_entry_sha256, asset_load.loaded_pcm_sha256, asset_load.loaded_sample_count, ) == ( asset_segment.asset_id, asset_segment.asset_manifest_entry_sha256, asset_segment.realized_pcm_sha256, asset_segment.output_sample_count, ) for selection in evidence.carrier_selections: carrier_segment = segment_by_id[selection.segment_id] selected_model = call_by_id[selection.selected_call_id] assert ( selected_model.text_sha256, selected_model.output_pcm_sha256, selected_model.output_sample_count, ) == ( carrier_segment.carrier_text_sha256, carrier_segment.realized_pcm_sha256, carrier_segment.output_sample_count, ) plain_contract_sha = _sha( json.dumps( asdict(inputs.contract), ensure_ascii=True, allow_nan=False, separators=(",", ":"), sort_keys=True, ).encode("ascii") ) assert evidence.renderer_contract_sha256 != plain_contract_sha def test_semantic_plan_is_recomputed_by_the_pinned_renderer(): inputs = _inputs() segments = list(inputs.plan.segments) segments[0] = replace( segments[0], canonical_text="錯", ) changed_plan = replace( inputs.plan, canonical_target="錯第一個英文字母乙", segments=tuple(segments), ) with pytest.raises( GlyphHybridEvidenceError, match="pinned renderer rebuild", ): rebuild_hybrid_evidence( replace(inputs, plan=changed_plan) ) def test_caller_controlled_segment_and_call_ids_are_replaced_by_opaque_ordinals(): inputs = _inputs() secret_segment_id = "raw-a-carrier-secret" secret_call_id = "raw-a-model-secret" segments = ( replace(inputs.plan.segments[0], segment_id=secret_segment_id), *inputs.plan.segments[1:], ) calls = ( replace( inputs.calls[0], call_id=secret_call_id, segment_id=secret_segment_id, ), *inputs.calls[1:], ) selections = ( replace( inputs.carrier_selections[0], segment_id=secret_segment_id, attempted_call_ids=(secret_call_id,), selected_call_id=secret_call_id, ), inputs.carrier_selections[1], ) varied = replace( inputs, plan=replace(inputs.plan, segments=segments), calls=calls, carrier_selections=selections, ) serialized = format_hybrid_evidence(varied) assert secret_segment_id.encode("ascii") not in serialized assert secret_call_id.encode("ascii") not in serialized validate_hybrid_evidence(serialized, varied) def test_all_required_budgets_are_recomputed_in_fixed_order(): evidence = rebuild_hybrid_evidence(_inputs()) observed = { budget.name: (budget.actual, budget.cap) for budget in evidence.request_proof.budgets } assert list(observed) == [ "model_invocations", "generated_chunks", "generated_speech_units", "asset_count", "asset_atoms", "asset_canonical_units", "asset_audio_samples", "asr_calls", "ecapa_calls", "squim_calls", ] assert observed["model_invocations"] == (2, 32) assert observed["generated_chunks"] == (2, 32) assert observed["generated_speech_units"] == (2, 800) assert observed["asset_count"] == (1, 32) assert observed["asset_atoms"] == (1, 128) assert observed["asset_canonical_units"] == (7, 512) assert observed["asset_audio_samples"] == (24_000, 61_440_000) assert observed["asr_calls"] == (2, 96) assert observed["ecapa_calls"] == (2, 96) assert observed["squim_calls"] == (2, 96) @pytest.mark.parametrize( "mutate", ( lambda value: value.__setitem__("schema", "other.schema.v7"), lambda value: value.__setitem__("version", 6), lambda value: value["renderer_contract"]["source_pins"][0].__setitem__( "sha256", "0" * 64 ), lambda value: value["request_proof"].__setitem__( "raw_input_sha256", "0" * 64 ), lambda value: value["request_proof"]["segments"].reverse(), lambda value: value["request_proof"]["segments"].pop(), lambda value: value["request_proof"]["segments"].append( value["request_proof"]["segments"][0].copy() ), lambda value: value["request_proof"]["segments"][0].__setitem__( "raw_end", 2 ), lambda value: value["request_proof"]["segments"][1].__setitem__( "realized_pcm_sha256", "0" * 64 ), lambda value: value["request_proof"]["budgets"][0].__setitem__( "actual", 1 ), lambda value: value["runtime_call_ledger"].pop(), lambda value: value["runtime_call_ledger"].reverse(), lambda value: value["carrier_selections"][0].__setitem__( "selected_call_id", "model-1" ), lambda value: value["request_proof"]["final_gate"].__setitem__( "passed", False ), lambda value: value["request_proof"][ "additional_strict_gates" ].__setitem__("asset_reconstruction", False), lambda value: value.__setitem__("extra", True), ), ) def test_canonical_tampering_never_survives_independent_rebuild(mutate): inputs = _inputs() serialized = format_hybrid_evidence(inputs) tampered = _canonical_mutation(serialized, mutate) with pytest.raises( GlyphHybridEvidenceError, match="independent trusted rebuild", ): validate_hybrid_evidence(tampered, inputs) def test_noncanonical_duplicate_key_and_wrong_root_fail_closed(): inputs = _inputs() serialized = format_hybrid_evidence(inputs) with pytest.raises(GlyphHybridEvidenceError, match="not canonical"): validate_hybrid_evidence(serialized + b" ", inputs) with pytest.raises(GlyphHybridEvidenceError, match="duplicate"): validate_hybrid_evidence(b'{"schema":1,"schema":2}', inputs) with pytest.raises(GlyphHybridEvidenceError, match="root"): validate_hybrid_evidence(b"[]", inputs) def test_plan_range_reorder_omit_duplicate_and_borrow_fail_before_formatting(): inputs = _inputs() segments = inputs.plan.segments invalid_plans = ( replace(inputs.plan, segments=(segments[1], segments[0], segments[2])), replace(inputs.plan, segments=segments[:-1]), replace(inputs.plan, segments=(segments[0], segments[0], *segments[1:])), replace( inputs.plan, segments=( replace(segments[0], raw_end=2, raw_text="甲a"), *segments[1:], ), ), ) for plan in invalid_plans: with pytest.raises(GlyphHybridEvidenceError): rebuild_hybrid_evidence(replace(inputs, plan=plan)) borrowed = replace( inputs.carrier_selections[0], attempted_call_ids=("model-1",), selected_call_id="model-1", ) with pytest.raises(GlyphHybridEvidenceError, match="borrows"): rebuild_hybrid_evidence( replace( inputs, carrier_selections=( borrowed, inputs.carrier_selections[1], ), ) ) def test_asset_hash_bytes_load_and_public_output_mismatches_fail_closed(): inputs = _inputs() asset_call = inputs.calls[2] assert isinstance(asset_call, TrustedAssetLoadCall) tampered_asset_call = replace( asset_call, loaded_pcm16_le=_pcm(24_000, 3_100), ) with pytest.raises(GlyphHybridEvidenceError, match="trusted bundle"): rebuild_hybrid_evidence( replace( inputs, calls=(*inputs.calls[:2], tampered_asset_call, *inputs.calls[3:]), ) ) asset_segment = inputs.plan.segments[1] bad_output = bytearray(inputs.assembled_pcm16_le) bad_output[asset_segment.output_start_sample * 2] ^= 1 with pytest.raises(GlyphHybridEvidenceError, match="public PCM"): rebuild_hybrid_evidence( replace(inputs, public_pcm16_le=bytes(bad_output)) ) with pytest.raises( GlyphHybridEvidenceError, match="reconstructed PCM|strict hybrid gates", ): rebuild_hybrid_evidence( replace( inputs, assembled_pcm16_le=bytes(bad_output), public_pcm16_le=bytes(bad_output), ) ) def test_hidden_duplicate_or_unattributed_runtime_calls_fail_closed(): inputs = _inputs() model = inputs.calls[0] assert isinstance(model, TrustedModelGenerationCall) hidden = replace(model, call_id="hidden-model", attempt_ordinal=1, seed=45_679) with pytest.raises(GlyphHybridEvidenceError, match="disclosed and attributed"): rebuild_hybrid_evidence(replace(inputs, calls=inputs.calls + (hidden,))) duplicate = replace(model, call_id=inputs.calls[1].call_id) with pytest.raises(GlyphHybridEvidenceError, match="unique"): rebuild_hybrid_evidence( replace(inputs, calls=(duplicate, *inputs.calls[1:])) ) with pytest.raises(GlyphHybridEvidenceError, match="asset segment"): rebuild_hybrid_evidence( replace(inputs, calls=inputs.calls[:2] + inputs.calls[3:]) ) def test_model_and_all_other_hard_budgets_are_enforced_and_caps_cannot_expand(): inputs = _inputs() for caps, reason in ( (replace(inputs.budget_caps, model_invocations=1), "model_invocations"), (replace(inputs.budget_caps, generated_chunks=1), "generated_chunks"), ( replace(inputs.budget_caps, generated_speech_units=1), "generated_speech_units", ), (replace(inputs.budget_caps, asset_count=0), "asset_count"), (replace(inputs.budget_caps, asset_atoms=0), "asset_atoms"), ( replace(inputs.budget_caps, asset_canonical_units=2), "asset_canonical_units", ), ( replace(inputs.budget_caps, asset_audio_samples=23_999), "asset_audio_samples", ), (replace(inputs.budget_caps, asr_calls=1), "asr_calls"), (replace(inputs.budget_caps, ecapa_calls=1), "ecapa_calls"), (replace(inputs.budget_caps, squim_calls=1), "squim_calls"), ): with pytest.raises(GlyphHybridEvidenceError, match=reason): rebuild_hybrid_evidence(replace(inputs, budget_caps=caps)) for caps, reason in ( (replace(inputs.budget_caps, model_invocations=33), "invocation cap"), (replace(inputs.budget_caps, generated_chunks=33), "chunk cap"), ( replace(inputs.budget_caps, generated_speech_units=801), "speech-unit cap", ), ): with pytest.raises(GlyphHybridEvidenceError, match=reason): rebuild_hybrid_evidence(replace(inputs, budget_caps=caps)) def test_logged_pass_boolean_cannot_replace_numeric_gate_evidence(): inputs = _inputs() squim = inputs.calls[5] assert isinstance(squim, TrustedSquimCall) failed = replace(squim, pesq=0.5) with pytest.raises( GlyphHybridEvidenceError, match="pinned evaluator rebuild|must both pass", ): rebuild_hybrid_evidence( replace( inputs, calls=(*inputs.calls[:5], failed, *inputs.calls[6:]), ) ) def test_evaluator_calls_must_use_exact_pcm_pins_scope_and_subject_attribution(): inputs = _inputs() asr = inputs.calls[3] assert isinstance(asr, TrustedAsrCall) bad_pcm = replace(asr, input_pcm16_le=_pcm(120_000, 4_000)) with pytest.raises(GlyphHybridEvidenceError, match="exact output PCM"): rebuild_hybrid_evidence( replace(inputs, calls=(*inputs.calls[:3], bad_pcm, *inputs.calls[4:])) ) bad_pin = replace(asr, evaluator_pin_name="untrusted-asr") with pytest.raises(GlyphHybridEvidenceError, match="unpinned"): rebuild_hybrid_evidence( replace(inputs, calls=(*inputs.calls[:3], bad_pin, *inputs.calls[4:])) ) borrowed_pin = replace(asr, evaluator_pin_name="squim") with pytest.raises(GlyphHybridEvidenceError, match="non-ASR"): rebuild_hybrid_evidence( replace( inputs, calls=(*inputs.calls[:3], borrowed_pin, *inputs.calls[4:]), ) ) bad_subject = replace(asr, subject_model_call_id="model-0") with pytest.raises( GlyphHybridEvidenceError, match="local evaluator|whole-output", ): rebuild_hybrid_evidence( replace( inputs, calls=(*inputs.calls[:3], bad_subject, *inputs.calls[4:]), ) ) model = inputs.calls[0] assert isinstance(model, TrustedModelGenerationCall) local_asr = TrustedAsrCall( call_id="asr-local-wrong-target", scope="local-primary", evaluator_pin_name=BREEZE25_PRIMARY_PIN_NAME, input_pcm16_le=model.output_pcm16_le, target_text="錯", transcript_text="甲", subject_model_call_id=model.call_id, ) with pytest.raises(GlyphHybridEvidenceError, match="local ASR target"): rebuild_hybrid_evidence( replace(inputs, calls=inputs.calls + (local_asr,)) ) def test_source_model_evaluator_and_renderer_pins_are_exactly_bound(): inputs = _inputs() serialized = format_hybrid_evidence(inputs) changed_source = replace( inputs.contract.source_pins[0], revision="different-revision", ) changed_inputs = replace( inputs, contract=replace(inputs.contract, source_pins=(changed_source,)), ) with pytest.raises(GlyphHybridEvidenceError, match="independent"): validate_hybrid_evidence(serialized, changed_inputs) changed_plan = replace(inputs.plan, profile_sha256=_sha("other-profile")) with pytest.raises(GlyphHybridEvidenceError, match="contract pins"): rebuild_hybrid_evidence(replace(inputs, plan=changed_plan)) changed_bundle = replace(inputs.bundle, bundle_sha256=_sha("other-bundle")) with pytest.raises( GlyphHybridEvidenceError, match="contract pins|actual asset bytes", ): rebuild_hybrid_evidence(replace(inputs, bundle=changed_bundle)) def test_bundle_entry_manifest_and_bundle_digests_are_recomputed_from_bytes(): inputs = _inputs() asset = inputs.bundle.assets[0] changed_asset = replace(asset, canonical_text="錯") changed_bundle = replace(inputs.bundle, assets=(changed_asset,)) with pytest.raises( GlyphHybridEvidenceError, match="entry digest", ): rebuild_hybrid_evidence( replace(inputs, bundle=changed_bundle) ) changed_pcm_asset = replace( asset, pcm16_le=_pcm(24_000, 3_100), ) changed_pcm_asset = replace( changed_pcm_asset, manifest_entry_sha256=compute_trusted_asset_entry_sha256( changed_pcm_asset ), ) with pytest.raises( GlyphHybridEvidenceError, match="actual asset bytes", ): rebuild_hybrid_evidence( replace( inputs, bundle=replace( inputs.bundle, assets=(changed_pcm_asset,), ), ) ) def test_control_segments_cannot_bypass_audio_or_speech_budgets(): inputs = _inputs() segments = list(inputs.plan.segments) segments[0] = replace(segments[0], kind="control") calls = tuple( call for call in inputs.calls if call.call_id != "model-0" ) changed_plan = replace( inputs.plan, segments=tuple(segments), ) with pytest.raises( GlyphHybridEvidenceError, match="speech-free zero-audio", ): rebuild_hybrid_evidence( replace( inputs, plan=changed_plan, calls=calls, carrier_selections=(inputs.carrier_selections[1],), renderer_evaluator=( lambda contract, bundle, raw_input: changed_plan ), ) ) def test_alignment_detects_balanced_onset_deletion_and_tail_insertion(): target = "".join(chr(0x4E00 + index) for index in range(100)) transcript = target[1:] + "尾" metrics = evidence_module._asr_metrics(target, transcript) assert metrics["cer"] == pytest.approx(0.02) assert metrics["prefix_deletions"] == 1 assert metrics["suffix_deletions"] == 0 assert metrics["extra_tail_units"] == 1 def test_asr_metrics_preserve_production_semantic_availability_gate(): target = "一般文字中放入 a&b@example.tw 之後繼續說明完整內容。" transcript = target.replace("a&b@example.tw", "b@example.tw") metrics = evidence_module._asr_metrics(target, transcript) assert metrics["cer"] < 0.08 assert metrics["network_protected_spans"] == 1 assert metrics["network_protected_spans_passed"] is False assert metrics["semantic_available"] is False def test_endpoint_and_active_speech_are_recomputed_from_exact_pcm(): inputs = _inputs() model = inputs.calls[0] assert isinstance(model, TrustedModelGenerationCall) bad_endpoint = replace( model.endpoint, tail_energy_ratio=0.01, ) with pytest.raises( GlyphHybridEvidenceError, match="tail energy", ): rebuild_hybrid_evidence( replace( inputs, calls=( replace(model, endpoint=bad_endpoint), *inputs.calls[1:], ), generation_evaluator=( lambda pin, text, seed, policy_id, policy_sha256, sample_rate: ( model.output_pcm16_le, bad_endpoint, ) if text == model.text else _generation_evaluator( pin, text, seed, policy_id, policy_sha256, sample_rate, ) ), ) ) forged_endpoint = replace( model.endpoint, generated_steps=model.endpoint.generated_steps - 1, ) with pytest.raises( GlyphHybridEvidenceError, match="does not match pinned replay", ): rebuild_hybrid_evidence( replace( inputs, calls=( replace(model, endpoint=forged_endpoint), *inputs.calls[1:], ), ) ) bindings = ( replace( inputs.gate_bindings[0], active_speech_samples=( inputs.gate_bindings[0].active_speech_samples - 1 ), ), inputs.gate_bindings[1], ) with pytest.raises( GlyphHybridEvidenceError, match="active speech samples", ): rebuild_hybrid_evidence( replace(inputs, gate_bindings=bindings) ) def test_independent_asr_profile_anchor_and_generation_policy_are_frozen(): inputs = _inputs() independent_asr = inputs.calls[6] assert isinstance(independent_asr, TrustedAsrCall) same_asr = replace( independent_asr, evaluator_pin_name=BREEZE25_PRIMARY_PIN_NAME, ) with pytest.raises( GlyphHybridEvidenceError, match="distinct pinned decoder profiles", ): rebuild_hybrid_evidence( replace( inputs, calls=( *inputs.calls[:6], same_asr, *inputs.calls[7:], ), ) ) different_weights_pin = replace( inputs.contract.evaluator_pins[1], sha256=_sha("different-breeze25-weights"), ) with pytest.raises( GlyphHybridEvidenceError, match="share the exact pinned model weights", ): rebuild_hybrid_evidence( replace( inputs, contract=replace( inputs.contract, evaluator_pins=( inputs.contract.evaluator_pins[0], different_weights_pin, *inputs.contract.evaluator_pins[2:], ), ), ) ) foreign_weights_pins = tuple( replace( pin, uri="hf://untrusted/other-asr", revision="foreign-revision", sha256=_sha("foreign-shared-weights"), ) if pin.kind == "asr" else pin for pin in inputs.contract.evaluator_pins ) with pytest.raises( GlyphHybridEvidenceError, match="official pinned Breeze ASR 25 weights", ): rebuild_hybrid_evidence( replace( inputs, contract=replace( inputs.contract, evaluator_pins=foreign_weights_pins, ), ) ) swapped_calls = list(inputs.calls) swapped_calls[3] = replace( inputs.calls[3], evaluator_pin_name=BREEZE25_CONFIRMATION_PIN_NAME, ) swapped_calls[6] = replace( inputs.calls[6], evaluator_pin_name=BREEZE25_PRIMARY_PIN_NAME, ) with pytest.raises( GlyphHybridEvidenceError, match="honest Breeze primary and confirmation logical pin names", ): rebuild_hybrid_evidence( replace(inputs, calls=tuple(swapped_calls)) ) ecapa = inputs.calls[4] assert isinstance(ecapa, TrustedEcapaCall) wrong_anchor = replace( ecapa, speaker_anchor_sha256=_sha("wrong-anchor"), ) with pytest.raises( GlyphHybridEvidenceError, match="unpinned speaker anchor", ): rebuild_hybrid_evidence( replace( inputs, calls=( *inputs.calls[:4], wrong_anchor, *inputs.calls[5:], ), ) ) model = inputs.calls[0] assert isinstance(model, TrustedModelGenerationCall) wrong_policy = replace( model, policy_sha256=_sha("unfrozen-policy"), ) with pytest.raises( GlyphHybridEvidenceError, match="unpinned generation policy", ): rebuild_hybrid_evidence( replace( inputs, calls=(wrong_policy, *inputs.calls[1:]), ) ) @pytest.mark.parametrize( "field", ( "asset_count", "asset_atoms", "asset_canonical_units", "asset_audio_samples", "asr_calls", "ecapa_calls", "squim_calls", ), ) def test_all_nonmodel_budget_caps_have_immutable_hard_maxima(field): inputs = _inputs() expanded = replace( inputs.budget_caps, **{field: getattr(inputs.budget_caps, field) + 1}, ) with pytest.raises( GlyphHybridEvidenceError, match="cap exceeds hard maximum", ): rebuild_hybrid_evidence( replace(inputs, budget_caps=expanded) ) def test_unused_bundle_payload_cannot_bypass_hard_resource_maxima( monkeypatch, ): inputs = _inputs() monkeypatch.setattr( evidence_module, "MAX_ASSET_CANONICAL_UNITS", 1, ) with pytest.raises( GlyphHybridEvidenceError, match="bundle canonical units exceed hard maximum", ): rebuild_hybrid_evidence(inputs) monkeypatch.setattr( evidence_module, "MAX_ASSET_CANONICAL_UNITS", 512, ) monkeypatch.setattr( evidence_module, "MAX_ASSET_AUDIO_SAMPLES", 100, ) with pytest.raises( GlyphHybridEvidenceError, match="bundle audio samples exceed hard maximum", ): rebuild_hybrid_evidence(inputs) def test_runtime_ledger_rejects_noncausal_whole_output_evaluators(): inputs = _inputs() reordered_calls = inputs.calls[3:] + inputs.calls[:3] with pytest.raises( GlyphHybridEvidenceError, match="precedes production calls", ): rebuild_hybrid_evidence( replace(inputs, calls=reordered_calls) ) def test_content_commitments_require_the_trusted_request_key(): inputs = _inputs() serialized = format_hybrid_evidence(inputs) changed_key_inputs = replace( inputs, content_commitment_key=hashlib.sha256( b"different-content-key" ).digest(), ) with pytest.raises( GlyphHybridEvidenceError, match="independent trusted rebuild", ): validate_hybrid_evidence(serialized, changed_key_inputs) def test_seeded_property_random_pcm_and_attempts_remain_deterministic(): rng = random.Random(0xB10E_5A17) baseline = _inputs() for case in range(30): calls = list(baseline.calls) selections = list(baseline.carrier_selections) extras: list[TrustedModelGenerationCall] = [] for segment_index, call_index in enumerate((0, 1)): selected = calls[call_index] assert isinstance(selected, TrustedModelGenerationCall) attempt_ids = [selected.call_id] for attempt_index in range(rng.randint(0, 2)): extra = replace( selected, call_id=( f"case-{case}-segment-{segment_index}" f"-attempt-{attempt_index}" ), attempt_ordinal=attempt_index + 1, seed=rng.randrange(2**32), output_pcm16_le=_pcm( len(selected.output_pcm16_le) // 2, rng.randint(1_000, 6_000), phase=attempt_index, terminal_silence_samples=480, ), ) extras.append(extra) attempt_ids.append(extra.call_id) selections[segment_index] = replace( selections[segment_index], attempted_call_ids=tuple(attempt_ids), ) model_calls = tuple(calls[:2] + extras) replay_by_key = { ( call.model_pin_name, call.text, call.seed, call.policy_id, call.policy_sha256, ): (call.output_pcm16_le, call.endpoint) for call in model_calls } def replay_generation( pin: ArtifactPin, text: str, seed: int, policy_id: str, policy_sha256: str, sample_rate: int, ) -> tuple[bytes, EndpointEvidence]: assert sample_rate == 48_000 return replay_by_key[ (pin.name, text, seed, policy_id, policy_sha256) ] varied = replace( baseline, calls=tuple(model_calls + tuple(calls[2:])), carrier_selections=tuple(selections), generation_evaluator=replay_generation, ) first = format_hybrid_evidence(varied) second = format_hybrid_evidence(varied) assert first == second validate_hybrid_evidence(first, varied)