Add dual-ASR whole-output verification

README.md

@@ -39,7 +39,9 @@ speaker-reference 模式使用的 ECAPA encoder 固定在
 Barbet 另固定在 `6fcd7ce4aa37f2250a3242995bef0fbc3b026ba8`，
 避免 Space 重啟後在沒有程式版本變更的情況下改變模型或 speaker embedding 空間。
 候選語意驗證固定使用 Whisper large-v3-turbo revision
-`41f01f3fe87f28c78e2fbf8b568835947dd65ed9`。
+`41f01f3fe87f28c78e2fbf8b568835947dd65ed9`；exact-assembled whole-output
+語意驗證另固定使用 Whisper large-v3 revision
+`06f233fe06e710322aca913c1bc4249a0d71fce1`，兩者必須同時通過。
 
 | 設定 | 值 |
 |---|---:|
@@ -54,7 +56,7 @@ Barbet 另固定在 `6fcd7ce4aa37f2250a3242995bef0fbc3b026ba8`，
 | Hard stop | native-pace target steps, independent of playback pace |
 | Pace correction | whole completed chunk, pitch-preserving; active-region stretcher disabled |
 | Pace / speaker eligibility duration | union of active 25 ms RMS frames at 10 ms hop; internal pauses excluded |
-| Semantic verification | whole / first 6 / last 6 units, no lexical tail; fail closed |
+| Semantic verification | turbo local screen + turbo/full-large-v3 hard intersection on exact whole output; first/last 6 exact, no lexical tail; fail closed |
 | Local speaker verification | frozen request centroid + bounded window begin/end directional drift |
 | Early accept speaker preference | similarity ≥ 0.25 and directional drop ≤ 0.05 for gated chunks |
 | Local hard speaker gate | similarity ≥ 0.10 and directional drop ≤ 0.10 |
@@ -85,12 +87,17 @@ margin 固定 +1 latent step、`min_len` 固定為 2，不會為了播放目標
 以及下述唯一 boundary-only、drop 不超過 0.15 的 DP-only chunks。切段保留標點，並依逗號、分號或句末標點
 插入不同長度的停頓。輸出最後會套用保守的 RMS floor 與 peak limit。
 完成 RMS matching、edge fade、pause、crossfade 與使用者 speed 後，服務會再對最終整段 waveform
-執行一次 normalized target 的 prefix/whole/suffix/tail、pace 與 speaker anchor/boundary gate；這個
-post-join gate 失敗時不會回傳先前已通過的 chunk 音訊。
+執行 normalized target 的 prefix/whole/suffix/tail、pace 與 speaker anchor/boundary gate，並要求
+turbo 與 full large-v3 的語意 hard intersection；這個 post-join gate 失敗時不會回傳先前已通過的
+chunk 音訊。Full large-v3 不驗 local chunk，只驗 production assembler 產生的 exact whole candidate
+或 exact sequence path；其 rejection 會讓 cascade 繼續下一個 seed。
 此外，same-seed trajectory 只有在每個 local chunk 都通過後，才會先用同一個
 `_assemble_trajectory_audio` 組成實際播放版本並跑整段 semantic/pace/speaker gate；joined gate
 失敗會取消該 whole trajectory 的資格，但保留已通過的 local evidence 供 sequence DP 使用。
-這個額外 whole-ASR 只對 local 全數通過的候選執行，DP 選定後仍會保留上述最終 post-join gate。
+這個額外 dual-ASR whole gate 只對 local turbo 全數通過的候選執行，DP 選定後仍會保留上述最終
+post-join gate。每個 request 只在 waveform float32 bytes、sample rate、normalized target 與 pinned
+verifier profile 完全相同時重用 full large-v3 evidence；任何 sample 改變都會重新驗證。ASR loader
+異常、OOM、空 transcript 或無效 evidence 都 fail closed，不會降級為 turbo-only。
 若整個 candidate budget 都沒有 joined-safe whole trajectory，多 chunk request 才會進入 bounded
 k-best DP。已通過 local hard gate 的 chunk 可直接進入 DP；唯一 rejection reason 是
 `boundary_speaker_drop`、且 drop 不超過 0.15 的 chunk 也可作為 DP-only 候選。這個局部容忍

app.py

@@ -40,6 +40,8 @@ from production import (
 from quality_runtime import (
     BASE_GENERATION_POLICY,
     SAFE_DURATION_GENERATION_POLICY,
+    VERIFICATION_WHISPER_MODEL_ID,
+    VERIFICATION_WHISPER_REVISION,
     WHISPER_MODEL_ID,
     WHISPER_REVISION,
     CandidateVerification,
@@ -50,6 +52,7 @@ from quality_runtime import (
     NoQualifiedCandidateError,
     RELEASE_SPEAKER_TRIGGER_SECONDS,
     SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP,
+    WholeWaveformVerificationCache,
     active_voiced_duration_seconds,
     active_audio_rms_db,
     candidate_limit_for_chunk_budget,
@@ -63,6 +66,7 @@ from quality_runtime import (
     resolve_request_seed,
     run_adaptive_cascade,
     speaker_evidence_from_audio,
+    transcribe_verification_whisper,
     transcribe_whisper,
     trajectory_gate_evidence,
     verify_trajectory,
@@ -120,6 +124,13 @@ QUALITY_PREFERRED_MIN_SPEAKER_SIMILARITY = 0.25
 QUALITY_PREFERRED_MAX_BOUNDARY_SPEAKER_DROP = 0.05
 QUALITY_MAX_SEQUENCE_PATHS = 3
 SHORT_AUDIO_SPEAKER_GATE_SECONDS = 1.50
+VERIFICATION_ASR_PROFILE = (
+    f"{VERIFICATION_WHISPER_MODEL_ID}@{VERIFICATION_WHISPER_REVISION}:"
+    f"greedy:zh:short{QUALITY_PREFIX_SUFFIX_UNITS}:cer{QUALITY_MAX_CER}:"
+    f"prefix{QUALITY_PREFIX_SUFFIX_UNITS}=0:"
+    f"suffix{QUALITY_PREFIX_SUFFIX_UNITS}=0:tail=0:"
+    f"max_tokens{QUALITY_FINAL_ASR_MAX_NEW_TOKENS}"
+)
 
 
 print(f"[BlueMagpie] downloading model from {REPO_ID}@{MODEL_REVISION} ...")
@@ -128,6 +139,14 @@ print(f"[BlueMagpie] caching speaker encoder from {ECAPA_REPO_ID}@{ECAPA_REVISIO
 ECAPA_DIR = snapshot_download(ECAPA_REPO_ID, revision=ECAPA_REVISION)
 print(f"[BlueMagpie] caching quality ASR from {WHISPER_MODEL_ID}@{WHISPER_REVISION} ...")
 ASR_DIR = snapshot_download(WHISPER_MODEL_ID, revision=WHISPER_REVISION)
+print(
+    "[BlueMagpie] caching independent final ASR from "
+    f"{VERIFICATION_WHISPER_MODEL_ID}@{VERIFICATION_WHISPER_REVISION} ..."
+)
+VERIFICATION_ASR_DIR = snapshot_download(
+    VERIFICATION_WHISPER_MODEL_ID,
+    revision=VERIFICATION_WHISPER_REVISION,
+)
 tokenizer = PreTrainedTokenizerFast(tokenizer_file=os.path.join(MODEL_DIR, "tokenizer.json"))
 print(f"[BlueMagpie] loading model on device={DEVICE} ...")
 model = BlueMagpieModel.from_local(MODEL_DIR, tokenizer=tokenizer, training=False, device=DEVICE)
@@ -347,6 +366,8 @@ def _verify_trajectory_audio(
     asr_max_new_tokens: int = 128,
     *,
     release_speaker_gate: bool = False,
+    transcriber=transcribe_whisper,
+    semantic_only: bool = False,
 ):
     if len(trajectory) != len(chunks):
         return verify_trajectory(())
@@ -357,7 +378,7 @@ def _verify_trajectory_audio(
         prepared = prepare_candidate_audio(
             audio,
             SR,
-            transcriber=lambda waveform, sample_rate: transcribe_whisper(
+            transcriber=lambda waveform, sample_rate: transcriber(
                 waveform,
                 sample_rate,
                 max_new_tokens=asr_max_new_tokens,
@@ -386,9 +407,12 @@ def _verify_trajectory_audio(
         speaker_embedding = None
         rms_db = None
         measure_speaker = (
-            release_speaker_measurement_required(duration)
-            if release_speaker_gate
-            else duration >= SHORT_AUDIO_SPEAKER_GATE_SECONDS
+            not semantic_only
+            and (
+                release_speaker_measurement_required(duration)
+                if release_speaker_gate
+                else duration >= SHORT_AUDIO_SPEAKER_GATE_SECONDS
+            )
         )
         if measure_speaker:
             try:
@@ -453,6 +477,7 @@ def _verify_trajectory_audio(
         max_prefix_cer=0.0,
         max_suffix_cer=0.0,
         max_extra_tail_units=0,
+        speaker_gate_enabled=not semantic_only,
         short_audio_seconds=(
             RELEASE_SPEAKER_TRIGGER_SECONDS
             if release_speaker_gate
@@ -468,7 +493,37 @@ def _verify_trajectory_audio(
             if release_speaker_gate
             else QUALITY_MAX_BOUNDARY_SPEAKER_DROP
         ),
-        max_pace_cps=QUALITY_MAX_PACE_CPS,
+        max_pace_cps=None if semantic_only else QUALITY_MAX_PACE_CPS,
+    )
+
+
+def _verify_independent_whole_audio(
+    waveform: np.ndarray,
+    target_text: str,
+    anchor: np.ndarray,
+    cache: WholeWaveformVerificationCache,
+):
+    """Verify exact assembled audio with the independent full large-v3 ASR."""
+
+    def verifier(exact_waveform, sample_rate, exact_target):
+        if sample_rate != SR:
+            raise ValueError("independent verifier sample rate mismatch")
+        return _verify_trajectory_audio(
+            (exact_waveform,),
+            (exact_target,),
+            anchor,
+            1.0,
+            QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
+            transcriber=transcribe_verification_whisper,
+            semantic_only=True,
+        )
+
+    return cache.verify(
+        waveform,
+        SR,
+        target_text,
+        VERIFICATION_ASR_PROFILE,
+        verifier,
     )
 
 
@@ -532,6 +587,7 @@ def _qualify_candidate_trajectory_audio(
     whole_target_text: str,
     anchor: np.ndarray,
     playback_speed: float,
+    independent_cache: WholeWaveformVerificationCache,
     *,
     candidate_seed: int,
 ):
@@ -565,9 +621,30 @@ def _qualify_candidate_trajectory_audio(
             f"seed={candidate_seed} "
             f"{_verification_metric_log_fields(joined_verification)}"
         )
-    return CandidateVerification(
+        return CandidateVerification(
+            qualified,
+            joined_output=trajectory_gate_evidence(joined_verification),
+        )
+
+    independent_verification = _verify_independent_whole_audio(
+        waveform,
+        whole_target_text,
+        anchor,
+        independent_cache,
+    )
+    dual_qualified = qualify_trajectory_with_joined_output(
         qualified,
-        joined_output=trajectory_gate_evidence(joined_verification),
+        independent_verification,
+    )
+    if not dual_qualified.passed:
+        print(
+            "[BlueMagpie] candidate independent whole output rejected "
+            f"seed={candidate_seed} "
+            f"{_verification_metric_log_fields(independent_verification)}"
+        )
+    return CandidateVerification(
+        dual_qualified,
+        joined_output=trajectory_gate_evidence(independent_verification),
     )
 
 
@@ -577,6 +654,7 @@ def _verify_sequence_trajectory_audio(
     whole_target_text: str,
     anchor: np.ndarray,
     playback_speed: float,
+    independent_cache: WholeWaveformVerificationCache,
 ):
     """Verify one ranked DP path after exact production assembly."""
 
@@ -585,7 +663,7 @@ def _verify_sequence_trajectory_audio(
         chunks,
         playback_speed,
     )
-    verification = _verify_trajectory_audio(
+    turbo_verification = _verify_trajectory_audio(
         (waveform,),
         (whole_target_text,),
         anchor,
@@ -593,14 +671,33 @@ def _verify_sequence_trajectory_audio(
         QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
         release_speaker_gate=True,
     )
-    status = "verified" if verification.passed else "rejected"
+    status = "verified" if turbo_verification.passed else "rejected"
     print(
-        f"[BlueMagpie] sequence path {status} "
+        f"[BlueMagpie] sequence path turbo {status} "
         f"rank={sequence_result.sequence_path_rank} "
         f"chunk_candidates={sequence_result.chunk_candidate_indices} "
-        f"{_verification_metric_log_fields(verification)}"
+        f"{_verification_metric_log_fields(turbo_verification)}"
+    )
+    if not turbo_verification.passed:
+        return turbo_verification
+
+    independent_verification = _verify_independent_whole_audio(
+        waveform,
+        whole_target_text,
+        anchor,
+        independent_cache,
+    )
+    status = "verified" if independent_verification.passed else "rejected"
+    print(
+        f"[BlueMagpie] sequence path independent {status} "
+        f"rank={sequence_result.sequence_path_rank} "
+        f"chunk_candidates={sequence_result.chunk_candidate_indices} "
+        f"{_verification_metric_log_fields(independent_verification)}"
+    )
+    return qualify_trajectory_with_joined_output(
+        turbo_verification,
+        independent_verification,
     )
-    return verification
 
 
 def _synthesize(
@@ -648,6 +745,7 @@ def _synthesize(
         max_generated_text_units=QUALITY_MAX_GENERATED_TEXT_UNITS,
     )
     anchor = _speaker_anchor_array(centroid)
+    independent_cache = WholeWaveformVerificationCache()
     try:
         with _GENERATION_LOCK:
             cascade = run_adaptive_cascade(
@@ -667,6 +765,7 @@ def _synthesize(
                     text,
                     anchor,
                     speed,
+                    independent_cache,
                     candidate_seed=seed,
                 ),
                 max_candidates=max_candidates,
@@ -683,6 +782,7 @@ def _synthesize(
                         text,
                         anchor,
                         speed,
+                        independent_cache,
                     )
                 ),
                 max_sequence_paths=QUALITY_MAX_SEQUENCE_PATHS,
@@ -730,9 +830,23 @@ def _synthesize(
         release_speaker_gate=True,
     )
     final_evidence = trajectory_gate_evidence(final_verification)
+    independent_final_verification = None
     try:
         require_verified_final_output(final_verification)
+        independent_final_verification = _verify_independent_whole_audio(
+            waveform,
+            text,
+            anchor,
+            independent_cache,
+        )
+        final_evidence = trajectory_gate_evidence(independent_final_verification)
+        require_verified_final_output(independent_final_verification)
     except FinalOutputRejectedError as error:
+        independent_fields = (
+            "not_run"
+            if independent_final_verification is None
+            else _verification_metric_log_fields(independent_final_verification)
+        )
         print(
             format_cascade_evidence_log(
                 cascade.diagnostics,
@@ -744,9 +858,12 @@ def _synthesize(
         )
         print(
             "[BlueMagpie] final output rejected "
-            f"{_verification_metric_log_fields(final_verification)}"
+            f"turbo={_verification_metric_log_fields(final_verification)} "
+            f"independent={independent_fields}"
         )
         raise gr.Error("最終合成結果未通過整段內容、語速與音色驗證，未回傳音訊。") from error
+    except (RuntimeError, ValueError) as error:
+        raise gr.Error("品質驗證暫時無法完成，未回傳未驗證的語音。") from error
     print(
         format_cascade_evidence_log(
             cascade.diagnostics,
@@ -758,7 +875,9 @@ def _synthesize(
     )
     print(
         "[BlueMagpie] final output verified "
-        f"score={final_verification.score:.6f}"
+        f"turbo_score={final_verification.score:.6f} "
+        f"independent_score={independent_final_verification.score:.6f} "
+        f"independent_cache_entries={independent_cache.entry_count}"
     )
     return SR, waveform
 

quality_runtime.py

@@ -7,6 +7,7 @@ load the real models lazily while unit tests remain deterministic and offline.
 
 from __future__ import annotations
 
+import hashlib
 import json
 import math
 import operator
@@ -24,6 +25,8 @@ from production import AsrComparison, CandidateSequenceSelection, compare_asr_te
 
 WHISPER_MODEL_ID = "openai/whisper-large-v3-turbo"
 WHISPER_REVISION = "41f01f3fe87f28c78e2fbf8b568835947dd65ed9"
+VERIFICATION_WHISPER_MODEL_ID = "openai/whisper-large-v3"
+VERIFICATION_WHISPER_REVISION = "06f233fe06e710322aca913c1bc4249a0d71fce1"
 WHISPER_ATTENTION_IMPLEMENTATION = "eager"
 WHISPER_RETURN_ATTENTION_MASK = True
 WHISPER_SAMPLE_RATE = 16_000
@@ -424,13 +427,15 @@ class WhisperRuntime:
     dtype: torch.dtype
 
 
-def load_pinned_whisper_runtime(
+def _load_whisper_runtime(
+    model_id: str,
+    revision: str,
     *,
     device: str | torch.device | None = None,
     processor_factory: Any | None = None,
     model_factory: Any | None = None,
 ) -> WhisperRuntime:
-    """Load the exact ASR revision used by the quality gate.
+    """Load one exact ASR revision through the shared deterministic contract.
 
     Factory injection exists for offline tests.  The default imports
     ``transformers`` only when this function is first called.
@@ -447,12 +452,12 @@ def load_pinned_whisper_runtime(
     )
     dtype = torch.float16 if selected_device.type == "cuda" else torch.float32
     processor = processor_factory.from_pretrained(
-        WHISPER_MODEL_ID,
-        revision=WHISPER_REVISION,
+        model_id,
+        revision=revision,
     )
     model = model_factory.from_pretrained(
-        WHISPER_MODEL_ID,
-        revision=WHISPER_REVISION,
+        model_id,
+        revision=revision,
         attn_implementation=WHISPER_ATTENTION_IMPLEMENTATION,
         torch_dtype=dtype,
         low_cpu_mem_usage=True,
@@ -468,6 +473,40 @@ def load_pinned_whisper_runtime(
     )
 
 
+def load_pinned_whisper_runtime(
+    *,
+    device: str | torch.device | None = None,
+    processor_factory: Any | None = None,
+    model_factory: Any | None = None,
+) -> WhisperRuntime:
+    """Load the pinned turbo ASR used for local candidate screening."""
+
+    return _load_whisper_runtime(
+        WHISPER_MODEL_ID,
+        WHISPER_REVISION,
+        device=device,
+        processor_factory=processor_factory,
+        model_factory=model_factory,
+    )
+
+
+def load_pinned_verification_whisper_runtime(
+    *,
+    device: str | torch.device | None = None,
+    processor_factory: Any | None = None,
+    model_factory: Any | None = None,
+) -> WhisperRuntime:
+    """Load the independent full large-v3 ASR used only for whole outputs."""
+
+    return _load_whisper_runtime(
+        VERIFICATION_WHISPER_MODEL_ID,
+        VERIFICATION_WHISPER_REVISION,
+        device=device,
+        processor_factory=processor_factory,
+        model_factory=model_factory,
+    )
+
+
 class LazyWhisperASR:
     """Thread-safe one-shot lazy loader with an injectable runtime factory."""
 
@@ -487,6 +526,9 @@ class LazyWhisperASR:
 
 
 _DEFAULT_WHISPER = LazyWhisperASR()
+_DEFAULT_VERIFICATION_WHISPER = LazyWhisperASR(
+    load_pinned_verification_whisper_runtime
+)
 
 
 def _split_whisper_audio(
@@ -639,6 +681,32 @@ def transcribe_whisper(
     return " ".join(str(text).strip() for text in decoded if str(text).strip())
 
 
+def transcribe_verification_whisper(
+    audio: np.ndarray | Sequence[float],
+    sample_rate: int,
+    *,
+    lazy_asr: LazyWhisperASR | None = None,
+    runtime: WhisperRuntime | None = None,
+    language: str = "zh",
+    task: str = "transcribe",
+    max_new_tokens: int = 128,
+) -> str:
+    """Transcribe with the separately pinned full large-v3 final verifier."""
+
+    selected_lazy = lazy_asr
+    if runtime is None and selected_lazy is None:
+        selected_lazy = _DEFAULT_VERIFICATION_WHISPER
+    return transcribe_whisper(
+        audio,
+        sample_rate,
+        lazy_asr=selected_lazy,
+        runtime=runtime,
+        language=language,
+        task=task,
+        max_new_tokens=max_new_tokens,
+    )
+
+
 @dataclass(frozen=True)
 class PreparedCandidateAudio:
     """Validated candidate waveform and its ASR transcript."""
@@ -1140,6 +1208,7 @@ def verify_candidate(
     max_prefix_cer: float = 0.0,
     max_suffix_cer: float = 0.0,
     max_extra_tail_units: int = 0,
+    speaker_gate_enabled: bool = True,
     short_audio_seconds: float = 1.5,
     min_speaker_similarity: float = 0.10,
     max_boundary_speaker_drop: float = 0.03,
@@ -1162,6 +1231,9 @@ def verify_candidate(
         short_unit_limit = max(0, int(short_text_units))
     except (TypeError, ValueError, OverflowError):
         short_unit_limit = -1
+    speaker_enabled = isinstance(speaker_gate_enabled, (bool, np.bool_))
+    if speaker_enabled:
+        speaker_enabled = bool(speaker_gate_enabled)
 
     # First normalize with a permissive finite limit to determine target units.
     preliminary = compare_asr_text(
@@ -1207,23 +1279,31 @@ def verify_candidate(
         elif pace > max_pace:
             reasons.append("pace_too_fast")
 
-    valid_common_config = all(
-        value is not None
-        for value in (
-            short_duration_limit,
-            general_cer_limit,
-            exact_cer_limit,
-            min_similarity,
-            max_boundary,
-            speaker_cost_weight,
-            boundary_cost_weight,
+    valid_common_config = bool(
+        isinstance(speaker_gate_enabled, (bool, np.bool_))
+        and short_duration_limit is not None
+        and general_cer_limit is not None
+        and exact_cer_limit is not None
+        and short_unit_limit >= 0
+        and (
+            not speaker_enabled
+            or all(
+                value is not None
+                for value in (
+                    min_similarity,
+                    max_boundary,
+                    speaker_cost_weight,
+                    boundary_cost_weight,
+                )
+            )
         )
-    ) and short_unit_limit >= 0
+    )
     if not valid_common_config:
         reasons.append("invalid_gate_config")
 
     speaker_gate_applied = bool(
-        duration is not None
+        speaker_enabled
+        and duration is not None
         and short_duration_limit is not None
         and duration >= short_duration_limit
     )
@@ -1323,6 +1403,84 @@ def trajectory_gate_evidence(
     )
 
 
+def exact_waveform_sha256(
+    audio: np.ndarray | Sequence[float],
+    sample_rate: int,
+) -> str:
+    """Hash exact canonical float32 samples together with their sample rate."""
+
+    if isinstance(sample_rate, (bool, np.bool_)):
+        raise ValueError("sample_rate must be a positive integer")
+    try:
+        rate = operator.index(sample_rate)
+    except (TypeError, ValueError, OverflowError) as error:
+        raise ValueError("sample_rate must be a positive integer") from error
+    if rate <= 0:
+        raise ValueError("sample_rate must be a positive integer")
+    waveform = _mono_audio(audio)
+    canonical = np.ascontiguousarray(waveform, dtype=np.dtype("<f4"))
+    digest = hashlib.sha256()
+    digest.update(b"bluemagpie-whole-waveform-f32le-v1\0")
+    digest.update(int(rate).to_bytes(8, "little", signed=False))
+    digest.update(int(canonical.size).to_bytes(8, "little", signed=False))
+    digest.update(memoryview(canonical).cast("B"))
+    return digest.hexdigest()
+
+
+class WholeWaveformVerificationCache:
+    """Request-local cache keyed by exact audio, target and verifier profile.
+
+    The cache deliberately accepts no process-global state.  A caller must
+    instantiate it inside one synthesis request, and verifier calls that raise
+    are never cached.  Passed and rejected gate evidence are both deterministic
+    evidence and may be reused only for an exact key match.
+    """
+
+    def __init__(self) -> None:
+        self._entries: dict[
+            tuple[str, int, str, str],
+            TrajectoryGateResult,
+        ] = {}
+
+    @property
+    def entry_count(self) -> int:
+        return len(self._entries)
+
+    def verify(
+        self,
+        audio: np.ndarray | Sequence[float],
+        sample_rate: int,
+        target_text: str,
+        verifier_profile: str,
+        verifier: Callable[[np.ndarray, int, str], TrajectoryGateResult],
+    ) -> TrajectoryGateResult:
+        if not isinstance(target_text, str) or not target_text:
+            raise ValueError("target_text must be a non-empty string")
+        if not isinstance(verifier_profile, str) or not verifier_profile:
+            raise ValueError("verifier_profile must be a non-empty string")
+        if not callable(verifier):
+            raise ValueError("verifier must be callable")
+        waveform = _mono_audio(audio)
+        if isinstance(sample_rate, (bool, np.bool_)):
+            raise ValueError("sample_rate must be a positive integer")
+        try:
+            rate = operator.index(sample_rate)
+        except (TypeError, ValueError, OverflowError) as error:
+            raise ValueError("sample_rate must be a positive integer") from error
+        if rate <= 0:
+            raise ValueError("sample_rate must be a positive integer")
+        waveform_hash = exact_waveform_sha256(waveform, rate)
+        key = (waveform_hash, rate, target_text, verifier_profile)
+        cached = self._entries.get(key)
+        if cached is not None:
+            return cached
+        verification = verifier(waveform, rate, target_text)
+        if not isinstance(verification, TrajectoryGateResult):
+            raise RuntimeError("whole-waveform verifier returned an invalid result")
+        self._entries[key] = verification
+        return verification
+
+
 def verify_trajectory(
     observations: Sequence[CandidateObservation],
     *,

tests/test_quality_runtime.py

@@ -14,6 +14,8 @@ from quality_runtime import (
     CASCADE_EVIDENCE_LOG_PREFIX,
     CASCADE_EVIDENCE_SCHEMA_VERSION,
     SAFE_DURATION_GENERATION_POLICY,
+    VERIFICATION_WHISPER_MODEL_ID,
+    VERIFICATION_WHISPER_REVISION,
     WHISPER_ATTENTION_IMPLEMENTATION,
     WHISPER_MODEL_ID,
     WHISPER_RETURN_ATTENTION_MASK,
@@ -28,6 +30,7 @@ from quality_runtime import (
     RELEASE_SPEAKER_TRIGGER_SECONDS,
     SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP,
     TrajectoryGateResult,
+    WholeWaveformVerificationCache,
     _sequence_fallback_candidate_result,
     WhisperRuntime,
     _split_whisper_audio,
@@ -35,9 +38,11 @@ from quality_runtime import (
     candidate_chunk_transition_score,
     candidate_limit_for_chunk_budget,
     cosine_similarity,
+    exact_waveform_sha256,
     format_cascade_evidence_log,
     generation_policy_for_candidate_offset,
     load_pinned_whisper_runtime,
+    load_pinned_verification_whisper_runtime,
     prepare_candidate_audio,
     qualify_trajectory_with_joined_output,
     release_speaker_evidence_from_audio,
@@ -50,6 +55,7 @@ from quality_runtime import (
     speaker_evidence_from_audio,
     trim_release_speaker_activity,
     transcribe_whisper,
+    transcribe_verification_whisper,
     trajectory_gate_evidence,
     verify_candidate,
     verify_trajectory,
@@ -352,6 +358,34 @@ def test_pinned_whisper_loader_uses_exact_revision_without_global_download():
     assert WHISPER_RETURN_ATTENTION_MASK is True
 
 
+def test_pinned_verification_whisper_loader_is_separate_and_exact():
+    processor = _FakeProcessor()
+    model = _FakeWhisperModel()
+    processor_factory = _Factory(processor)
+    model_factory = _Factory(model)
+
+    runtime = load_pinned_verification_whisper_runtime(
+        device="cpu",
+        processor_factory=processor_factory,
+        model_factory=model_factory,
+    )
+
+    assert runtime.processor is processor
+    assert runtime.model is model
+    assert processor_factory.calls == [
+        (
+            (VERIFICATION_WHISPER_MODEL_ID,),
+            {"revision": VERIFICATION_WHISPER_REVISION},
+        )
+    ]
+    model_args, model_kwargs = model_factory.calls[0]
+    assert model_args == (VERIFICATION_WHISPER_MODEL_ID,)
+    assert model_kwargs["revision"] == VERIFICATION_WHISPER_REVISION
+    assert model_kwargs["attn_implementation"] == WHISPER_ATTENTION_IMPLEMENTATION
+    assert model_kwargs["torch_dtype"] == torch.float32
+    assert model_kwargs["use_safetensors"] is True
+
+
 def test_lazy_whisper_runtime_loads_once_and_transcription_is_deterministic():
     processor = _FakeProcessor()
     model = _FakeWhisperModel()
@@ -384,6 +418,25 @@ def test_lazy_whisper_runtime_loads_once_and_transcription_is_deterministic():
     }
 
 
+def test_verification_transcriber_uses_injected_full_runtime_without_turbo_state():
+    processor = _FakeProcessor()
+    model = _FakeWhisperModel()
+    runtime = WhisperRuntime(processor, model, torch.device("cpu"), torch.float32)
+    loads = []
+    lazy_asr = LazyWhisperASR(lambda: loads.append("verification") or runtime)
+
+    transcript = transcribe_verification_whisper(
+        _tone(seconds=0.5),
+        16_000,
+        lazy_asr=lazy_asr,
+        max_new_tokens=440,
+    )
+
+    assert transcript == "合成內容完整。"
+    assert loads == ["verification"]
+    assert model.generate_calls[0][1]["max_new_tokens"] == 440
+
+
 def test_long_whisper_audio_is_batched_below_thirty_second_limit():
     processor = _FakeProcessor()
     model = _FakeWhisperModel()
@@ -661,6 +714,42 @@ def test_short_candidate_uses_exact_semantics_but_bypasses_speaker_hard_gate():
     assert failed.rejection_reasons == ("semantic_gate",)
 
 
+def test_semantic_only_gate_never_requires_speaker_evidence_but_stays_strict():
+    long_target = "這是一段超過短音訊門檻而且內容完整的驗證文字"
+    passed = verify_candidate(
+        CandidateObservation(
+            target_text=long_target,
+            transcript_text=long_target,
+            audio_duration_seconds=10.0,
+        ),
+        speaker_gate_enabled=False,
+    )
+    failed = verify_candidate(
+        CandidateObservation(
+            target_text=long_target,
+            transcript_text=long_target[:-2],
+            audio_duration_seconds=10.0,
+        ),
+        speaker_gate_enabled=False,
+    )
+    invalid = verify_candidate(
+        CandidateObservation(
+            target_text=long_target,
+            transcript_text=long_target,
+            audio_duration_seconds=10.0,
+        ),
+        speaker_gate_enabled="no",
+    )
+
+    assert passed.passed
+    assert not passed.speaker_gate_applied
+    assert passed.score == 0.0
+    assert not failed.passed
+    assert failed.rejection_reasons == ("semantic_gate",)
+    assert not invalid.passed
+    assert "invalid_gate_config" in invalid.rejection_reasons
+
+
 def test_candidate_pace_gate_is_optional_and_fails_closed_when_enabled():
     base = CandidateObservation(
         target_text="內容完整",
@@ -774,6 +863,129 @@ def test_final_whole_output_gate_rejects_post_join_regression_without_fallback()
         require_verified_final_output(None)
 
 
+def test_whole_waveform_cache_reuses_only_exact_audio_target_rate_and_profile():
+    cache = WholeWaveformVerificationCache()
+    waveform = _tone(seconds=0.25)
+    calls = []
+
+    def verifier(exact_waveform, sample_rate, target_text):
+        calls.append((exact_waveform.copy(), sample_rate, target_text))
+        return _joined_verification(target_text, target_text)
+
+    first = cache.verify(waveform, 16_000, "內容完整", "large-v3@one", verifier)
+    second = cache.verify(
+        waveform.copy(),
+        16_000,
+        "內容完整",
+        "large-v3@one",
+        verifier,
+    )
+    changed = waveform.copy()
+    changed[-1] = np.nextafter(changed[-1], np.float32(1.0))
+    cache.verify(changed, 16_000, "內容完整", "large-v3@one", verifier)
+    cache.verify(waveform, 48_000, "內容完整", "large-v3@one", verifier)
+    cache.verify(waveform, 16_000, "另一內容", "large-v3@one", verifier)
+    cache.verify(waveform, 16_000, "內容完整", "large-v3@two", verifier)
+
+    assert first is second
+    assert len(calls) == 5
+    assert cache.entry_count == 5
+    assert exact_waveform_sha256(waveform, 16_000) == exact_waveform_sha256(
+        waveform.copy(),
+        16_000,
+    )
+    assert exact_waveform_sha256(waveform, 16_000) != exact_waveform_sha256(
+        waveform,
+        48_000,
+    )
+
+
+def test_whole_waveform_cache_caches_rejections_but_never_errors_or_bad_results():
+    waveform = _tone(seconds=0.25)
+    rejected = _joined_verification("完整內容", "錯誤內容")
+    cache = WholeWaveformVerificationCache()
+    rejection_calls = []
+
+    def reject_verifier(*args):
+        rejection_calls.append(args)
+        return rejected
+
+    assert not cache.verify(
+        waveform,
+        16_000,
+        "完整內容",
+        "large-v3@one",
+        reject_verifier,
+    ).passed
+    assert not cache.verify(
+        waveform.copy(),
+        16_000,
+        "完整內容",
+        "large-v3@one",
+        reject_verifier,
+    ).passed
+    assert len(rejection_calls) == 1
+    assert cache.entry_count == 1
+
+    error_cache = WholeWaveformVerificationCache()
+    error_calls = []
+
+    def fail_verifier(*args):
+        error_calls.append(args)
+        raise RuntimeError("verification OOM")
+
+    for _ in range(2):
+        with pytest.raises(RuntimeError, match="OOM"):
+            error_cache.verify(
+                waveform,
+                16_000,
+                "完整內容",
+                "large-v3@one",
+                fail_verifier,
+            )
+    assert len(error_calls) == 2
+    assert error_cache.entry_count == 0
+
+    with pytest.raises(RuntimeError, match="invalid result"):
+        error_cache.verify(
+            waveform,
+            16_000,
+            "完整內容",
+            "large-v3@one",
+            lambda *args: True,
+        )
+    assert error_cache.entry_count == 0
+
+
+@pytest.mark.parametrize(
+    ("audio", "sample_rate", "target", "profile", "match"),
+    [
+        ([0.0, math.nan], 16_000, "內容", "profile", "non-finite"),
+        ([0.0], 0, "內容", "profile", "sample_rate"),
+        ([0.0], True, "內容", "profile", "sample_rate"),
+        ([0.0], 16_000, "", "profile", "target_text"),
+        ([0.0], 16_000, "內容", "", "verifier_profile"),
+    ],
+)
+def test_whole_waveform_cache_rejects_malformed_keys(
+    audio,
+    sample_rate,
+    target,
+    profile,
+    match,
+):
+    cache = WholeWaveformVerificationCache()
+    with pytest.raises(ValueError, match=match):
+        cache.verify(
+            audio,
+            sample_rate,
+            target,
+            profile,
+            lambda *args: _joined_verification("內容", "內容"),
+        )
+    assert cache.entry_count == 0
+
+
 def test_joined_output_rejection_preserves_local_results_for_sequence_dp():
     observations = []
     artifacts = []
@@ -1513,6 +1725,33 @@ def test_adaptive_cascade_expands_to_five_and_selects_lowest_verified_score():
     assert result.attempted_seeds == (100, 101, 102, 103, 104)
 
 
+def test_hard_secondary_rejection_continues_cascade_to_later_candidate():
+    calls = []
+
+    def generator(chunks, seed):
+        calls.append(seed)
+        return (chunks[0], seed)
+
+    def dual_verifier(trajectory, chunks, seed):
+        # This models a turbo-passing candidate rejected by the independent
+        # whole-output ASR.  It remains a normal rejected gate result so the
+        # deterministic cascade must continue instead of returning it.
+        return _gate_result(seed == 101, score=0.0)
+
+    result = run_adaptive_cascade(
+        ["完整內容"],
+        100,
+        generator,
+        dual_verifier,
+        max_candidates=5,
+    )
+
+    assert calls == [100, 101, 102, 103, 104]
+    assert result.seed == 101
+    assert result.candidate_index == 1
+    assert result.attempted_seeds == (100, 101, 102, 103, 104)
+
+
 def test_adaptive_cascade_uses_ten_only_when_stage_five_has_no_verified_candidate():
     calls = []
 

tests/test_release_pins.py

@@ -72,9 +72,18 @@ def test_quality_asr_is_revision_pinned():
 
     assert constants["WHISPER_MODEL_ID"] == "openai/whisper-large-v3-turbo"
     assert constants["WHISPER_REVISION"] == "41f01f3fe87f28c78e2fbf8b568835947dd65ed9"
+    assert constants["VERIFICATION_WHISPER_MODEL_ID"] == "openai/whisper-large-v3"
+    assert (
+        constants["VERIFICATION_WHISPER_REVISION"]
+        == "06f233fe06e710322aca913c1bc4249a0d71fce1"
+    )
     assert constants["WHISPER_ATTENTION_IMPLEMENTATION"] == "eager"
     assert "WHISPER_RETURN_ATTENTION_MASK = True" in quality_source
     assert "snapshot_download(WHISPER_MODEL_ID, revision=WHISPER_REVISION)" in app_source
+    assert "snapshot_download(\n    VERIFICATION_WHISPER_MODEL_ID," in app_source
+    assert "revision=VERIFICATION_WHISPER_REVISION" in app_source
+    assert "load_pinned_verification_whisper_runtime" in quality_source
+    assert "transcribe_verification_whisper" in quality_source
 
 
 def test_quality_runtime_dependencies_are_version_pinned():
@@ -264,6 +273,103 @@ def test_whole_candidate_qualification_uses_the_exact_return_assembler_after_loc
     assert "require_verified_final_output(final_verification)" in synthesize_source
 
 
+def test_space_hard_intersects_dual_asr_only_on_exact_whole_waveforms():
+    source = (ROOT / "app.py").read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    functions = {
+        node.name: node
+        for node in tree.body
+        if isinstance(node, ast.FunctionDef)
+    }
+    verify_source = ast.get_source_segment(source, functions["_verify_trajectory_audio"])
+    independent_source = ast.get_source_segment(
+        source,
+        functions["_verify_independent_whole_audio"],
+    )
+    qualify_source = ast.get_source_segment(
+        source,
+        functions["_qualify_candidate_trajectory_audio"],
+    )
+    sequence_source = ast.get_source_segment(
+        source,
+        functions["_verify_sequence_trajectory_audio"],
+    )
+    synthesize_source = ast.get_source_segment(source, functions["_synthesize"])
+
+    assert all(
+        segment is not None
+        for segment in (
+            verify_source,
+            independent_source,
+            qualify_source,
+            sequence_source,
+            synthesize_source,
+        )
+    )
+    assert "transcriber=transcribe_whisper" in verify_source
+    assert "speaker_gate_enabled=not semantic_only" in verify_source
+    assert "transcriber=transcribe_verification_whisper" in independent_source
+    assert "semantic_only=True" in independent_source
+    assert "cache.verify(" in independent_source
+    assert "VERIFICATION_ASR_PROFILE" in independent_source
+
+    local_index = qualify_source.index("local_verification = _verify_trajectory_audio(")
+    local_fail_index = qualify_source.index("if not local_verification.passed:")
+    assemble_index = qualify_source.index("waveform = _assemble_trajectory_audio(")
+    turbo_joined_index = qualify_source.index("joined_verification = _verify_trajectory_audio(")
+    turbo_fail_index = qualify_source.index("if not qualified.passed:")
+    independent_index = qualify_source.index("_verify_independent_whole_audio(")
+    dual_index = qualify_source.index(
+        "dual_qualified = qualify_trajectory_with_joined_output("
+    )
+    assert (
+        local_index
+        < local_fail_index
+        < assemble_index
+        < turbo_joined_index
+        < turbo_fail_index
+        < independent_index
+        < dual_index
+    )
+
+    sequence_assemble = sequence_source.index("_assemble_trajectory_audio(")
+    sequence_turbo = sequence_source.index("turbo_verification = _verify_trajectory_audio(")
+    sequence_turbo_fail = sequence_source.index("if not turbo_verification.passed:")
+    sequence_independent = sequence_source.index("_verify_independent_whole_audio(")
+    assert sequence_assemble < sequence_turbo < sequence_turbo_fail < sequence_independent
+
+    cache_create = synthesize_source.index(
+        "independent_cache = WholeWaveformVerificationCache()"
+    )
+    cascade_index = synthesize_source.index("cascade = run_adaptive_cascade(")
+    final_assemble = synthesize_source.index(
+        "waveform = _assemble_trajectory_audio(cascade.trajectory, chunks, speed)"
+    )
+    final_turbo = synthesize_source.index("final_verification = _verify_trajectory_audio(")
+    final_turbo_require = synthesize_source.index(
+        "require_verified_final_output(final_verification)"
+    )
+    final_independent = synthesize_source.index(
+        "independent_final_verification = _verify_independent_whole_audio("
+    )
+    final_independent_require = synthesize_source.index(
+        "require_verified_final_output(independent_final_verification)"
+    )
+    return_index = synthesize_source.index("return SR, waveform")
+    assert (
+        cache_create
+        < cascade_index
+        < final_assemble
+        < final_turbo
+        < final_turbo_require
+        < final_independent
+        < final_independent_require
+        < return_index
+    )
+    assert synthesize_source.count("independent_cache,") >= 3
+    assert "except (RuntimeError, ValueError) as error:" in synthesize_source
+
+
 def test_space_wires_bounded_k_best_paths_to_exact_assembled_whole_gate():
     source = (ROOT / "app.py").read_text(encoding="utf-8")
     readme = (ROOT / "README.md").read_text(encoding="utf-8")