Harden network endpoint and local verification

README.md

@@ -56,10 +56,10 @@ Barbet 另固定在 `6fcd7ce4aa37f2250a3242995bef0fbc3b026ba8`，
 | Email / URL frontend | hybrid lexical labels; scheme, all-uppercase atoms and short suffixes use explicit ASCII letter tokens; digits are read one by one and separators remain audible |
 | Stop policy | 0.50 → 0.05 from 75% to 95% predicted progress, 1 hit |
 | Endpoint cue | append terminal punctuation for model input when missing, except very short text |
-| Hard stop | native-pace target steps, independent of playback pace |
+| Hard stop | native-pace target steps, independent of playback pace; URL/email chunks use a separate conservative endpoint-duration counter (ASCII alnum run ÷ 2) |
 | Pace correction | derive total/active pace from the completed raw waveform, then one pitch-preserving STFT stretch (ordinary `1536/384`; URL/email code-switch `2048/512`); no active-interval slicing |
 | Pace / speaker eligibility duration | union of active 25 ms RMS frames at 10 ms hop; internal pauses excluded |
-| Semantic verification | orthographic whole CER + tone-aware acoustic first/last 6 exact + no lexical tail; turbo/full-large-v3 hard intersection; fail closed |
+| Semantic verification | orthographic whole CER + tone-aware acoustic first/last 6 exact + no lexical tail; range-proven URL/email locals and every exact whole output require turbo/full-large-v3 hard intersection; fail closed |
 | Local endpoint verification | internal chunk first/last 6 may contain at most one substitution and zero deletion; absolute request onset/ending remain exact |
 | Acoustic quality gate | pinned CPU SQUIM objective SHA-256 `2c54586fea83fb5eb5394d710038ee89f55cab7011a5bf730bebed4c8777e828`; hard STOI ≥ 0.60 / PESQ ≥ 1.12; at ≥1.50 s, single-chunk early return additionally requires preferred STOI ≥ 0.72 / PESQ ≥ 1.20 and speaker 0.25 / 0.05; SI-SDR is bounded soft evidence only |
 | Local speaker verification | frozen request centroid + bounded window begin/end directional drift |
@@ -97,10 +97,16 @@ margin 固定 +1 latent step、`min_len` 固定為 2，不會為了播放目標
 正奇數使用 alternate CFG 2.0；含 URL/email component 的 chunk 仍有 CFG 3.0 的內容安全下限。
 同一列 log 也記錄實際 generated chunks/text units 與各 row 的有效候選數。
 每次生成前另外記錄 seed、local chunk index、duration policy、scheduled CFG，
-以及 network/short-text floor 後的 effective CFG；因此即使最後 fail closed 也能重建嘗試軌跡。
+endpoint-duration units/counter，以及 network/short-text floor 後的 effective CFG；因此即使最後
+fail closed 也能重建嘗試軌跡。公開 pace、chunk planner 與 32/800 work budget 仍使用原本的
+speech-unit counter；只有 URL/email-conditioned native endpoint estimate 對連續 ASCII alnum run
+採用除以 2 的保守 counter。這個 counter 刻意同時決定 adaptive weak-stop 的
+`expected_steps` progress 與 `max_len` hard cap；兩者屬於同一份 native endpoint plan，
+但不會進入播放 pace、chunk planning 或 32/800 work accounting，`min_len` 仍固定為 2。
 終端 outcome 另輸出 content-free canonical evidence schema v4，包含 original chunk indices、
 row-local ordinal、network provenance、CFG contract、32/800 預算用量與 selected-path 交叉檢查，
-以及 local、joined、independent-large-v3 與 final gate 的 bounded SQUIM scalar，
+每個 local row 的 independent-large-v3 attempted/pass/proof-count/result attestation，以及
+local、joined、independent-large-v3 與 final gate 的 bounded SQUIM scalar，
 不包含 target/transcript/audio/embedding。
 為了讓 release contract 與正式量測一致，UI 不提供其他 primary CFG。
 內部 hosted evaluator 可注入 `[0, 2^31)` 的固定 root seed 以重現結果；UI 不暴露這個參數，
@@ -114,9 +120,13 @@ Ragged DP 可混合已通過 hard gate 的 chunks，
 完成 RMS matching、edge fade、pause、crossfade 與使用者 speed 後，服務會再對最終整段 waveform
 執行 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；initial whole rejection 會啟動低覆蓋 refill，sequence rejection 則只會
-前進到下一條已排序的 exact path。
+chunk 音訊。除此之外，每個帶有 planner range proof 的 URL/email local chunk 在進入 coverage pool
+以前，也必須用完全相同的 `NetworkFragmentProof` 同時通過 turbo 與 pinned full large-v3；
+large-v3 reject 會成為不可被 boundary proxy 放寬的 semantic hard failure。一般 local chunk
+維持 turbo gate；已被 turbo 的非 boundary hard gate 拒絕、原本就不可能進 pool 的 network row
+不額外耗用 full large-v3。Production assembler 產生的 exact whole candidate 或 exact sequence path 仍全部
+使用 dual-ASR。Initial whole rejection 會啟動低覆蓋 refill，sequence rejection 則只會前進到
+下一條已排序的 exact path。
 此外，初始 same-seed trajectory 只有在每個 local chunk 都通過後，才會先用同一個
 `_assemble_trajectory_audio` 組成實際播放版本並跑整段 semantic/pace/speaker gate；joined gate
 失敗會取消該 whole trajectory 的資格，但保留已通過的 local evidence 供 sequence DP 使用。

app.py

@@ -24,6 +24,7 @@ from production import (
     apply_loudness_floor,
     canonicalize_asr_network_fragments,
     coalesce_text_chunks,
+    count_network_endpoint_duration_units,
     contains_network_identifier,
     count_speech_units,
     effective_generation_cfg,
@@ -59,16 +60,20 @@ from quality_runtime import (
     ChunkCandidateArtifact,
     FinalOutputRejectedError,
     GenerationPolicy,
+    LocalIndependentGateEvidence,
     NoQualifiedCandidateError,
     RELEASE_SPEAKER_TRIGGER_SECONDS,
     SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP,
     WholeWaveformVerificationCache,
     active_voiced_duration_seconds,
+    candidate_gate_evidence,
     active_audio_rms_db,
     active_audio_median_f0_hz,
     format_cascade_evidence_log,
     generation_cfg_for_candidate_offset,
     generation_policy_for_candidate_offset,
+    intersect_local_semantic_verification,
+    local_candidate_has_coverage_eligibility,
     prepare_candidate_audio,
     qualify_trajectory_with_joined_output,
     require_verified_final_output,
@@ -313,12 +318,18 @@ def _generate_chunk(
             ascii_cps=policy.ascii_cps,
         )
     )
+    endpoint_duration_units = (
+        count_network_endpoint_duration_units(text)
+        if network_conditioned
+        else count_speech_units(text)
+    )
     model_text, expected_steps, hard_stop_steps = endpoint_generation_plan(
         text,
         generation_cps=generation_cps,
         step_seconds=STEP_SECONDS,
         margin_steps=policy.hard_stop_margin_steps,
         add_terminal_punctuation=count_speech_units(text) >= MIN_ENDPOINT_CUE_UNITS,
+        duration_units=endpoint_duration_units,
     )
     # Do not hold generation open to enforce pace. The model can finish the
     # requested text early; extending its latent sequence creates tail speech.
@@ -368,7 +379,10 @@ def _generate_chunk(
     print(
         "[BlueMagpie] generation policy "
         f"name={policy.name} seed={request_seed} generation_cps={generation_cps:.2f} "
-        f"expected_steps={expected_steps} hard_stop_steps={hard_stop_steps} min_len={min_len}"
+        f"duration_units={endpoint_duration_units} "
+        f"duration_counter={'network_conservative' if network_conditioned else 'public'} "
+        f"expected_steps={expected_steps} hard_stop_steps={hard_stop_steps} "
+        f"min_len={min_len}"
     )
     audio = audio.detach().float().cpu().numpy().reshape(-1)
     pace_speed = target_pace_speed(
@@ -886,6 +900,101 @@ def _verification_metric_log_fields(verification) -> str:
     )
 
 
+def _verify_network_local_asr_intersection(
+    turbo_verification,
+    trajectory: tuple[np.ndarray, ...],
+    chunks: tuple[str, ...],
+    anchor: np.ndarray,
+    proof_rows: tuple[tuple[NetworkFragmentProof, ...], ...] | None,
+    *,
+    candidate_seed: int,
+):
+    """Require pinned large-v3 semantics before retaining network locals.
+
+    Only proof-bearing rows are sent through the independent model.  Their
+    exact ``NetworkFragmentProof`` tuples are reused unchanged, so neither ASR
+    can borrow a whole URL/email proof for a different local range.  The
+    intersection helper keeps turbo speaker/SQUIM/pace/artifact evidence while
+    projecting every independent rejection into a hard semantic local failure.
+    """
+
+    if proof_rows is None:
+        return turbo_verification, tuple(
+            LocalIndependentGateEvidence(False, None, 0, None)
+            for _ in chunks
+        )
+    if (
+        len(proof_rows) != len(chunks)
+        or len(trajectory) != len(chunks)
+        or len(turbo_verification.candidate_results) != len(chunks)
+    ):
+        raise ValueError("network local verification provenance does not align")
+    independent_evidence = [
+        LocalIndependentGateEvidence(
+            attempted=False,
+            passed=None,
+            proof_count=len(proofs),
+            result=None,
+        )
+        for proofs in proof_rows
+    ]
+    selected_indices = tuple(
+        index
+        for index, proofs in enumerate(proof_rows)
+        if (
+            proofs
+            and local_candidate_has_coverage_eligibility(
+                turbo_verification.candidate_results[index],
+                max_local_boundary_speaker_drop=(
+                    SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP
+                ),
+            )
+        )
+    )
+    if not selected_indices:
+        return turbo_verification, tuple(independent_evidence)
+    independent_proof_rows = tuple(
+        proof_rows[index] for index in selected_indices
+    )
+    independent_verification = _verify_trajectory_audio(
+        tuple(trajectory[index] for index in selected_indices),
+        tuple(chunks[index] for index in selected_indices),
+        anchor,
+        1.0,
+        QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
+        transcriber=transcribe_verification_whisper,
+        semantic_only=True,
+        network_fragment_proofs=independent_proof_rows,
+    )
+    for local_index, primary_index in enumerate(selected_indices):
+        result = independent_verification.candidate_results[local_index]
+        independent_evidence[primary_index] = LocalIndependentGateEvidence(
+            attempted=True,
+            passed=result.passed is True,
+            proof_count=len(proof_rows[primary_index]),
+            result=candidate_gate_evidence(result),
+        )
+        comparison = result.comparison
+        print(
+            "[BlueMagpie] network local independent "
+            f"seed={candidate_seed} local_chunk_index={primary_index} "
+            f"proof_count={len(proof_rows[primary_index])} "
+            f"passed={result.passed} cer={comparison.cer:.6f} "
+            f"prefix_cer={comparison.prefix_cer:.6f} "
+            f"suffix_cer={comparison.suffix_cer:.6f} "
+            f"tail_units={comparison.extra_tail_units} "
+            f"reasons={result.rejection_reasons}"
+        )
+    return (
+        intersect_local_semantic_verification(
+            turbo_verification,
+            independent_verification,
+            selected_indices,
+        ),
+        tuple(independent_evidence),
+    )
+
+
 def _qualify_candidate_trajectory_audio(
     trajectory: tuple[np.ndarray, ...],
     chunks: tuple[str, ...],
@@ -899,16 +1008,30 @@ def _qualify_candidate_trajectory_audio(
 ):
     """Run whole-output qualification only after every local chunk passes."""
 
+    proof_rows = _network_fragment_proof_rows(chunks, chunk_specs)
     local_verification = _verify_trajectory_audio(
         trajectory,
         chunks,
         anchor,
         playback_speed,
-        network_fragment_proofs=_network_fragment_proof_rows(chunks, chunk_specs),
+        network_fragment_proofs=proof_rows,
         local_endpoint_roles=_local_endpoint_role_rows(chunks, chunk_specs),
     )
+    local_verification, independent_local_results = (
+        _verify_network_local_asr_intersection(
+            local_verification,
+            trajectory,
+            chunks,
+            anchor,
+            proof_rows,
+            candidate_seed=candidate_seed,
+        )
+    )
     if not local_verification.passed:
-        return CandidateVerification(local_verification)
+        return CandidateVerification(
+            local_verification,
+            independent_local_results=independent_local_results,
+        )
 
     waveform = _assemble_trajectory_audio(
         trajectory,
@@ -937,6 +1060,7 @@ def _qualify_candidate_trajectory_audio(
         )
         return CandidateVerification(
             qualified,
+            independent_local_results=independent_local_results,
             joined_output=joined_evidence,
         )
 
@@ -958,6 +1082,7 @@ def _qualify_candidate_trajectory_audio(
         )
     return CandidateVerification(
         dual_qualified,
+        independent_local_results=independent_local_results,
         joined_output=joined_evidence,
         independent_output=trajectory_gate_evidence(independent_verification),
     )
@@ -969,19 +1094,36 @@ def _verify_refill_candidate_trajectory_audio(
     anchor: np.ndarray,
     playback_speed: float,
     chunk_specs: tuple[GenerationChunkSpec, ...] | None = None,
+    *,
+    candidate_seed: int,
 ):
-    """Apply the unchanged strict local gates to one safe-duration refill."""
+    """Apply strict dual-ASR local gates to one safe-duration refill."""
 
     if len(trajectory) != 1 or len(chunks) != 1:
         return verify_trajectory(())
-    return _verify_trajectory_audio(
+    proof_rows = _network_fragment_proof_rows(chunks, chunk_specs)
+    local_verification = _verify_trajectory_audio(
         trajectory,
         chunks,
         anchor,
         playback_speed,
-        network_fragment_proofs=_network_fragment_proof_rows(chunks, chunk_specs),
+        network_fragment_proofs=proof_rows,
         local_endpoint_roles=_local_endpoint_role_rows(chunks, chunk_specs),
     )
+    local_verification, independent_local_results = (
+        _verify_network_local_asr_intersection(
+            local_verification,
+            trajectory,
+            chunks,
+            anchor,
+            proof_rows,
+            candidate_seed=candidate_seed,
+        )
+    )
+    return CandidateVerification(
+        local_verification,
+        independent_local_results=independent_local_results,
+    )
 
 
 def _verify_sequence_trajectory_audio(
@@ -1331,6 +1473,7 @@ def _synthesize(
                         anchor,
                         speed,
                         generation_chunk_specs(seed, candidate_chunks),
+                        candidate_seed=seed,
                     )
                 ),
                 sequence_final_verifier=lambda sequence_result, candidate_chunks: (

production.py

@@ -2118,6 +2118,39 @@ def count_speech_units(text: str) -> int:
     return units
 
 
+def count_network_endpoint_duration_units(text: str) -> int:
+    """Count conservative duration units for network-conditioned generation.
+
+    Public pace, chunk planning and generation-work budgets intentionally keep
+    using :func:`count_speech_units`.  The acoustic model needs more native
+    endpoint headroom for opaque URL/email labels, though, so this separate
+    counter compresses every contiguous ASCII alphanumeric run by two instead
+    of compressing alphabetic runs by four.  It is valid only for the
+    network-conditioned endpoint estimate and never imposes a minimum length.
+    """
+
+    units = 0
+    ascii_run_length = 0
+
+    def flush_ascii() -> None:
+        nonlocal units, ascii_run_length
+        if ascii_run_length <= 0:
+            return
+        units += max(1, math.ceil(ascii_run_length / 2))
+        ascii_run_length = 0
+
+    for char in normalize_tts_text(text).lower():
+        if char.isascii() and char.isalnum():
+            ascii_run_length += 1
+        elif _is_cjk(char):
+            flush_ascii()
+            units += 1
+        else:
+            flush_ascii()
+    flush_ascii()
+    return units
+
+
 def effective_generation_cfg(
     text: str,
     requested_cfg: float,
@@ -2268,6 +2301,7 @@ def endpoint_generation_plan(
     step_seconds: float | None,
     margin_steps: int = 1,
     add_terminal_punctuation: bool = True,
+    duration_units: int | None = None,
 ) -> tuple[str, int, int]:
     """Plan a native-pace generation cap independently of output playback pace."""
 
@@ -2276,7 +2310,38 @@ def endpoint_generation_plan(
         if add_terminal_punctuation
         else normalize_tts_text(text)
     )
-    expected_steps = target_cps_steps(model_text, generation_cps, step_seconds)
+    if duration_units is None:
+        expected_steps = target_cps_steps(
+            model_text,
+            generation_cps,
+            step_seconds,
+        )
+    else:
+        if isinstance(duration_units, bool):
+            raise ValueError("duration_units must be a non-negative integer")
+        try:
+            planned_units = operator.index(duration_units)
+        except (TypeError, ValueError, OverflowError) as error:
+            raise ValueError(
+                "duration_units must be a non-negative integer"
+            ) from error
+        if planned_units < 0:
+            raise ValueError("duration_units must be a non-negative integer")
+        expected_steps = (
+            0
+            if (
+                planned_units <= 0
+                or generation_cps <= 0.0
+                or not step_seconds
+                or step_seconds <= 0.0
+            )
+            else max(
+                1,
+                math.ceil(
+                    (planned_units / generation_cps) / step_seconds
+                ),
+            )
+        )
     hard_stop_steps = duration_hard_stop_steps(
         expected_steps,
         ratio=1.0,

quality_runtime.py

@@ -1561,6 +1561,16 @@ class TrajectoryGateEvidence:
     result: CandidateGateEvidence | None
 
 
+@dataclass(frozen=True)
+class LocalIndependentGateEvidence:
+    """Bounded per-row attestation for independent local semantic ASR."""
+
+    attempted: bool
+    passed: bool | None
+    proof_count: int
+    result: CandidateGateEvidence | None
+
+
 @dataclass(frozen=True)
 class CandidateGenerationEvidence:
     """Content-free CFG and source-row evidence for one generation call."""
@@ -1592,6 +1602,7 @@ class CandidateAttemptEvidence:
     scheduled_cfg: float | None
     effective_cfgs: tuple[float, ...]
     floor_reasons: tuple[tuple[str, ...], ...]
+    independent_local_results: tuple[LocalIndependentGateEvidence, ...] = ()
     joined_output: TrajectoryGateEvidence | None = None
     independent_output: TrajectoryGateEvidence | None = None
 
@@ -2008,6 +2019,7 @@ class CandidateVerification:
     """Selection result plus content-free whole-output gate evidence."""
 
     verification: TrajectoryGateResult
+    independent_local_results: tuple[LocalIndependentGateEvidence, ...] = ()
     joined_output: TrajectoryGateEvidence | None = None
     independent_output: TrajectoryGateEvidence | None = None
 
@@ -2264,6 +2276,114 @@ def qualify_trajectory_with_joined_output(
     )
 
 
+def intersect_local_semantic_verification(
+    primary_verification: TrajectoryGateResult,
+    independent_verification: TrajectoryGateResult,
+    primary_indices: Sequence[int],
+) -> TrajectoryGateResult:
+    """Hard-intersect selected local rows without replacing acoustic evidence.
+
+    ``primary_verification`` owns the turbo ASR, speaker, SQUIM, pace and
+    boundary-proxy evidence consumed by the coverage selector.
+    ``independent_verification`` contains semantic-only large-v3 results for
+    the selected network-conditioned rows in the order given by
+    ``primary_indices``.  A large-v3 rejection is projected onto the
+    corresponding primary row with allow-listed semantic reason codes, making
+    it impossible for either the strict pool or the boundary-only proxy to
+    retain that local candidate.  Passing intersections return the original
+    primary object unchanged.
+    """
+
+    if not isinstance(primary_verification, TrajectoryGateResult):
+        raise TypeError("primary_verification must be a TrajectoryGateResult")
+    if not isinstance(independent_verification, TrajectoryGateResult):
+        raise TypeError(
+            "independent_verification must be a TrajectoryGateResult"
+        )
+    try:
+        selected_indices = tuple(primary_indices)
+    except TypeError as error:
+        raise ValueError("primary_indices must be an ordered index sequence") from error
+    if (
+        not selected_indices
+        or any(
+            isinstance(index, (bool, np.bool_))
+            or not isinstance(index, int)
+            for index in selected_indices
+        )
+        or selected_indices != tuple(sorted(set(selected_indices)))
+        or any(
+            index < 0
+            or index >= len(primary_verification.candidate_results)
+            for index in selected_indices
+        )
+        or len(independent_verification.candidate_results)
+        != len(selected_indices)
+        or any(
+            not isinstance(result, CandidateGateResult)
+            for result in independent_verification.candidate_results
+        )
+    ):
+        raise ValueError(
+            "independent local verification rows do not match primary indices"
+        )
+
+    failed_secondary_rows = {
+        local_index
+        for local_index, result in enumerate(
+            independent_verification.candidate_results
+        )
+        if (
+            result.passed is not True
+            or not math.isfinite(result.score)
+            or bool(result.rejection_reasons)
+        )
+    }
+    if independent_verification.passed is not True and not failed_secondary_rows:
+        # A malformed/non-finite trajectory-level result must not allow any of
+        # its apparently passing local projections into the coverage pool.
+        failed_secondary_rows.update(range(len(selected_indices)))
+    if not failed_secondary_rows:
+        return primary_verification
+
+    combined_results = list(primary_verification.candidate_results)
+    for local_index in sorted(failed_secondary_rows):
+        primary_index = selected_indices[local_index]
+        primary_result = combined_results[primary_index]
+        independent_result = independent_verification.candidate_results[
+            local_index
+        ]
+        semantic_reasons = ["semantic_gate"]
+        if (
+            "network_protected_span_mismatch"
+            in independent_result.rejection_reasons
+        ):
+            semantic_reasons.append("network_protected_span_mismatch")
+        combined_results[primary_index] = replace(
+            primary_result,
+            passed=False,
+            score=math.inf,
+            rejection_reasons=tuple(
+                dict.fromkeys(
+                    (*primary_result.rejection_reasons, *semantic_reasons)
+                )
+            ),
+        )
+
+    rejection_reasons = tuple(
+        f"chunk_{index}:{reason}"
+        for index, result in enumerate(combined_results)
+        for reason in result.rejection_reasons
+    )
+    return TrajectoryGateResult(
+        passed=False,
+        candidate_results=tuple(combined_results),
+        score=math.inf,
+        rejection_reasons=rejection_reasons,
+        chunk_artifacts=primary_verification.chunk_artifacts,
+    )
+
+
 class NoQualifiedCandidateError(RuntimeError):
     """Raised when the full adaptive cascade has no verified trajectory."""
 
@@ -2364,6 +2484,21 @@ def _trajectory_gate_evidence_payload(
     }
 
 
+def _local_independent_gate_evidence_payload(
+    evidence: LocalIndependentGateEvidence,
+) -> dict[str, Any]:
+    return {
+        "attempted": evidence.attempted,
+        "passed": evidence.passed,
+        "proof_count": evidence.proof_count,
+        "result": (
+            None
+            if evidence.result is None
+            else _candidate_gate_evidence_payload(evidence.result)
+        ),
+    }
+
+
 def _candidate_attempt_evidence_payload(
     evidence: CandidateAttemptEvidence,
 ) -> dict[str, Any]:
@@ -2399,6 +2534,14 @@ def _candidate_attempt_evidence_payload(
             _candidate_gate_evidence_payload(result)
             for result in evidence.local_results
         ],
+        "independent_local_evidence_complete": (
+            len(evidence.independent_local_results)
+            == evidence.local_result_count
+        ),
+        "independent_local_results": [
+            _local_independent_gate_evidence_payload(result)
+            for result in evidence.independent_local_results
+        ],
         "joined_output": _trajectory_gate_evidence_payload(
             evidence.joined_output
         ),
@@ -3045,6 +3188,7 @@ class _VerifiedTrajectoryCandidate:
     seed: int
     trajectory: Any
     verification: TrajectoryGateResult
+    independent_local_results: tuple[LocalIndependentGateEvidence, ...] = ()
     joined_output: TrajectoryGateEvidence | None = None
     independent_output: TrajectoryGateEvidence | None = None
     generation_evidence: CandidateGenerationEvidence | None = None
@@ -3054,19 +3198,54 @@ def _unwrap_candidate_verification(
     value: Any,
 ) -> tuple[
     TrajectoryGateResult,
+    tuple[LocalIndependentGateEvidence, ...],
     TrajectoryGateEvidence | None,
     TrajectoryGateEvidence | None,
 ]:
     if isinstance(value, CandidateVerification):
         verification = value.verification
+        independent_local_results = value.independent_local_results
         joined_output = value.joined_output
         independent_output = value.independent_output
     else:
         verification = value
+        independent_local_results = ()
         joined_output = None
         independent_output = None
     if not isinstance(verification, TrajectoryGateResult):
         raise TypeError("candidate_verifier must return TrajectoryGateResult")
+    if not isinstance(independent_local_results, tuple) or (
+        independent_local_results
+        and (
+            len(independent_local_results)
+            != len(verification.candidate_results)
+            or any(
+                not isinstance(result, LocalIndependentGateEvidence)
+                or type(result.attempted) is not bool
+                or type(result.proof_count) is not int
+                or result.proof_count > CASCADE_EVIDENCE_MAX_LOCAL_RESULTS
+                or (
+                    result.attempted
+                    and (
+                        type(result.passed) is not bool
+                        or result.proof_count <= 0
+                        or not isinstance(result.result, CandidateGateEvidence)
+                        or result.result.passed is not result.passed
+                    )
+                )
+                or (
+                    not result.attempted
+                    and (
+                        result.passed is not None
+                        or result.proof_count < 0
+                        or result.result is not None
+                    )
+                )
+                for result in independent_local_results
+            )
+        )
+    ):
+        raise TypeError("candidate independent-local evidence is invalid")
     if joined_output is not None and not isinstance(
         joined_output,
         TrajectoryGateEvidence,
@@ -3077,7 +3256,12 @@ def _unwrap_candidate_verification(
         TrajectoryGateEvidence,
     ):
         raise TypeError("candidate independent-output evidence is invalid")
-    return verification, joined_output, independent_output
+    return (
+        verification,
+        independent_local_results,
+        joined_output,
+        independent_output,
+    )
 
 
 def _validated_candidate_generation_evidence(
@@ -3233,6 +3417,7 @@ def _candidate_attempt_evidence(
         scheduled_cfg=(generation.scheduled_cfg if generation is not None else None),
         effective_cfgs=(generation.effective_cfgs if generation is not None else ()),
         floor_reasons=(generation.floor_reasons if generation is not None else ()),
+        independent_local_results=candidate.independent_local_results,
         joined_output=candidate.joined_output,
         independent_output=candidate.independent_output,
     )
@@ -3541,6 +3726,27 @@ def _sequence_fallback_candidate_result(
     )
 
 
+def local_candidate_has_coverage_eligibility(
+    result: CandidateGateResult,
+    *,
+    max_local_boundary_speaker_drop: float | None,
+) -> bool:
+    """Return whether a primary local result could enter either safe pool."""
+
+    if not isinstance(result, CandidateGateResult):
+        return False
+    eligible = _sequence_fallback_candidate_result(
+        result,
+        max_local_boundary_speaker_drop=max_local_boundary_speaker_drop,
+    )
+    return bool(
+        eligible.passed is True
+        and math.isfinite(eligible.score)
+        and eligible.score >= 0.0
+        and not eligible.rejection_reasons
+    )
+
+
 def _preferred_speaker_verification(
     verification: TrajectoryGateResult,
     *,
@@ -4140,6 +4346,7 @@ def run_coverage_adaptive_cascade(
     try:
         (
             initial_verification,
+            initial_independent_local_results,
             initial_joined_output,
             initial_independent_output,
         ) = (
@@ -4160,6 +4367,7 @@ def run_coverage_adaptive_cascade(
             seed=base_seed,
             trajectory=initial_trajectory,
             verification=initial_verification,
+            independent_local_results=initial_independent_local_results,
             joined_output=initial_joined_output,
             independent_output=initial_independent_output,
             generation_evidence=initial_generation_evidence,
@@ -4303,6 +4511,7 @@ def run_coverage_adaptive_cascade(
         try:
             (
                 refill_verification,
+                refill_independent_local_results,
                 refill_joined_output,
                 refill_independent_output,
             ) = (
@@ -4323,6 +4532,7 @@ def run_coverage_adaptive_cascade(
                 seed=seed,
                 trajectory=refill_trajectory,
                 verification=refill_verification,
+                independent_local_results=refill_independent_local_results,
                 joined_output=refill_joined_output,
                 independent_output=refill_independent_output,
                 generation_evidence=refill_generation_evidence,
@@ -4630,6 +4840,7 @@ def run_adaptive_cascade(
     first_trajectory = candidate_generator(chunk_tuple, first_seed)
     (
         first_verification,
+        first_independent_local_results,
         first_joined_output,
         first_independent_output,
     ) = _unwrap_candidate_verification(
@@ -4641,6 +4852,7 @@ def run_adaptive_cascade(
         seed=first_seed,
         trajectory=first_trajectory,
         verification=first_verification,
+        independent_local_results=first_independent_local_results,
         joined_output=first_joined_output,
         independent_output=first_independent_output,
     )
@@ -4677,6 +4889,7 @@ def run_adaptive_cascade(
             trajectory = candidate_generator(chunk_tuple, seed)
             (
                 verification,
+                independent_local_results,
                 joined_output,
                 independent_output,
             ) = _unwrap_candidate_verification(
@@ -4689,6 +4902,7 @@ def run_adaptive_cascade(
                     seed=seed,
                     trajectory=trajectory,
                     verification=verification,
+                    independent_local_results=independent_local_results,
                     joined_output=joined_output,
                     independent_output=independent_output,
                 )

tests/test_coverage_adaptive.py

@@ -11,11 +11,14 @@ from quality_runtime import (
     CandidateObservation,
     CandidateVerification,
     ChunkCandidateArtifact,
+    LocalIndependentGateEvidence,
     NoQualifiedCandidateError,
     SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP,
     TrajectoryGateResult,
+    candidate_gate_evidence,
     format_cascade_evidence_log,
     generation_cfg_for_candidate_offset,
+    intersect_local_semantic_verification,
     run_coverage_adaptive_cascade,
     select_culprit_diverse_candidate_sequences,
     trajectory_gate_evidence,
@@ -161,6 +164,142 @@ def test_whole_exact_winner_returns_before_any_refill_or_sequence_check():
     assert result.diagnostics.attempts[0].joined_output.passed
 
 
+def test_mixed_network_local_v3_failure_cannot_enter_strict_or_proxy_pool():
+    chunks = ("普通內容完整", "網路內容完整")
+    generated = []
+
+    def primary(candidate_chunks):
+        observations = []
+        artifacts = []
+        for index, chunk in enumerate(candidate_chunks):
+            observations.append(
+                CandidateObservation(
+                    target_text=chunk,
+                    transcript_text=chunk,
+                    audio_duration_seconds=2.0,
+                    speaker_similarity=0.70,
+                    begin_speaker_similarity=(0.80 if index == len(candidate_chunks) - 1 else 0.60),
+                    end_speaker_similarity=(0.68 if index == len(candidate_chunks) - 1 else 0.60),
+                    pace_cps=2.0,
+                )
+            )
+            artifacts.append(
+                ChunkCandidateArtifact(
+                    speaker_embedding=np.array([1.0, 0.0], dtype=np.float32),
+                    rms_db=-20.0,
+                )
+            )
+        return verify_trajectory(
+            observations,
+            chunk_artifacts=artifacts,
+            max_pace_cps=4.3,
+            max_boundary_speaker_drop=0.10,
+        )
+
+    def independent_failure(target):
+        return verify_trajectory(
+            [CandidateObservation(target, "完全錯誤", 2.0)],
+            speaker_gate_enabled=False,
+        )
+
+    def combined(candidate_chunks):
+        primary_verification = primary(candidate_chunks)
+        network_local_index = len(candidate_chunks) - 1
+        independent = independent_failure(
+            candidate_chunks[network_local_index]
+        )
+        verification = intersect_local_semantic_verification(
+            primary_verification,
+            independent,
+            (network_local_index,),
+        )
+        local_evidence = tuple(
+            (
+                LocalIndependentGateEvidence(
+                    True,
+                    False,
+                    1,
+                    candidate_gate_evidence(
+                        independent.candidate_results[0]
+                    ),
+                )
+                if index == network_local_index
+                else LocalIndependentGateEvidence(False, None, 0, None)
+            )
+            for index in range(len(candidate_chunks))
+        )
+        return CandidateVerification(
+            verification,
+            independent_local_results=local_evidence,
+        )
+
+    def generator(candidate_chunks, seed):
+        generated.append((candidate_chunks, seed))
+        return tuple(f"{seed}:{chunk}" for chunk in candidate_chunks)
+
+    with pytest.raises(NoQualifiedCandidateError) as captured:
+        run_coverage_adaptive_cascade(
+            chunks,
+            800,
+            generator,
+            lambda trajectory, candidate_chunks, seed: combined(
+                candidate_chunks
+            ),
+            lambda trajectory, candidate_chunks, seed: combined(
+                candidate_chunks
+            ),
+            sequence_final_verifier=lambda result, candidate_chunks: pytest.fail(
+                "a full path must not exist"
+            ),
+            max_generated_chunks=3,
+            max_generated_text_units=100,
+            sequence_fallback_max_local_boundary_speaker_drop=(
+                SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP
+            ),
+        )
+
+    assert generated == [
+        (chunks, 800),
+        ((chunks[1],), 801),
+    ]
+    diagnostics = captured.value.diagnostics
+    assert len(diagnostics.attempts) == 2
+    assert diagnostics.sequence_search is not None
+    assert diagnostics.sequence_search.eligible_candidate_counts == (1, 0)
+    initial = diagnostics.attempts[0]
+    refill = diagnostics.attempts[1]
+    assert initial.local_results[0].passed is True
+    assert initial.local_results[1].passed is False
+    assert initial.local_results[1].rejection_reasons == (
+        "boundary_speaker_drop",
+        "semantic_gate",
+    )
+    assert initial.independent_local_results[0] == (
+        LocalIndependentGateEvidence(False, None, 0, None)
+    )
+    assert initial.independent_local_results[1].attempted is True
+    assert initial.independent_local_results[1].passed is False
+    assert refill.independent_local_results[0].attempted is True
+    assert refill.independent_local_results[0].passed is False
+    payload = json.loads(
+        format_cascade_evidence_log(
+            diagnostics,
+            outcome="no_qualified_candidate",
+            generated_chunk_limit=3,
+            generated_text_unit_limit=100,
+        ).removeprefix(CASCADE_EVIDENCE_LOG_PREFIX)
+    )
+    assert payload["attempts"][0]["independent_local_results"][0] == {
+        "attempted": False,
+        "passed": None,
+        "proof_count": 0,
+        "result": None,
+    }
+    assert payload["attempts"][0]["independent_local_results"][1][
+        "passed"
+    ] is False
+
+
 def test_single_chunk_requires_joint_preferred_quality_and_speaker_for_early_return():
     chunk = ("完整內容",)
     generated = []

tests/test_production.py

@@ -1,4 +1,5 @@
 import math
+from dataclasses import replace
 
 import numpy as np
 import pytest
@@ -15,6 +16,7 @@ from production import (
     coalesce_text_chunks,
     compare_asr_text,
     contains_network_identifier,
+    count_network_endpoint_duration_units,
     count_speech_units,
     duration_hard_stop_steps,
     effective_generation_cfg,
@@ -1245,6 +1247,44 @@ def test_local_network_fragment_accepts_exact_range_bound_hybrid_lexical_atoms()
     assert evidence.transcript_text == target
 
 
+def test_real_network_fragment_proofs_reject_repeat_wrong_range_and_reorder():
+    raw = "請查 https://a.tw 並寄信到 a@b.co。"
+    spec = plan_generation_chunks(raw, normalize_spoken_forms(raw))[0]
+    first, second = spec.network_fragment_proofs
+
+    exact = canonicalize_asr_network_fragments(
+        spec.text,
+        spec.text,
+        (first, second),
+    )
+
+    assert exact.passed is True
+    with pytest.raises(ValueError, match="bind|offset|range"):
+        canonicalize_asr_network_fragments(
+            spec.text,
+            spec.text,
+            (first, first),
+        )
+    with pytest.raises(ValueError, match="bind|offset|range"):
+        canonicalize_asr_network_fragments(
+            spec.text,
+            spec.text,
+            (second, first),
+        )
+    with pytest.raises(ValueError, match="bind|offset|range"):
+        canonicalize_asr_network_fragments(
+            spec.text,
+            spec.text,
+            (
+                replace(
+                    first,
+                    chunk_start=first.chunk_start + 1,
+                ),
+                second,
+            ),
+        )
+
+
 @pytest.mark.parametrize(
     "transcript",
     (
@@ -1929,6 +1969,7 @@ def test_prefaded_network_verifier_seam_preserves_exact_zero_silence():
 
 def test_duration_units_and_target_pace_min_len():
     assert count_speech_units("AI TTS 測試 1234") == 6
+    assert count_network_endpoint_duration_units("AI TTS 測試 1234") == 7
     assert target_cps_min_len(
         "一二三四五六七八",
         target_cps=4.0,
@@ -1941,6 +1982,60 @@ def test_duration_units_and_target_pace_min_len():
     assert duration_hard_stop_steps(0, ratio=1.08, margin_steps=3) == 2000
 
 
+def test_network_endpoint_duration_counter_is_separate_from_public_units():
+    text = " coastwatch 點 example 點 T W 斜線 tide。"
+
+    assert count_speech_units(text) == 12
+    assert count_network_endpoint_duration_units(text) == 17
+    assert count_speech_units("tour123") == 2
+    assert count_network_endpoint_duration_units("tour123") == 4
+
+
+def test_h07_safe_network_endpoint_uses_conservative_units_only_for_cap():
+    raw = "潮汐預報可查詢 https://coastwatch.example.tw/tide。"
+    normalized = normalize_spoken_forms(raw)
+    specs = plan_generation_chunks(raw, normalized)
+    row = specs[1]
+
+    assert row.network_conditioned
+    assert count_speech_units(row.text) == 12
+    duration_units = count_network_endpoint_duration_units(row.text)
+    model_text, expected_steps, hard_stop_steps = endpoint_generation_plan(
+        row.text,
+        generation_cps=4.0,
+        step_seconds=0.16,
+        margin_steps=1,
+        duration_units=duration_units,
+    )
+
+    assert model_text == normalize_tts_text(row.text)
+    assert duration_units == 17
+    assert expected_steps == 27
+    assert hard_stop_steps == 28
+    # Planner and public pace/work accounting remain on the frozen counter.
+    assert tuple(count_speech_units(spec.text) for spec in specs) == (18, 12)
+    assert target_pace_speed(
+        28_000,
+        10_000,
+        row.text,
+        target_cps=4.0,
+        min_speed=0.80,
+    ) == pytest.approx(2.8 / 3.0)
+
+
+@pytest.mark.parametrize("duration_units", [True, -1, 1.5, "17"])
+def test_endpoint_generation_plan_rejects_invalid_explicit_duration_units(
+    duration_units,
+):
+    with pytest.raises(ValueError, match="duration_units"):
+        endpoint_generation_plan(
+            "測試",
+            generation_cps=4.0,
+            step_seconds=0.16,
+            duration_units=duration_units,
+        )
+
+
 def test_terminal_punctuation_supplies_an_endpoint_cue():
     assert ensure_terminal_punctuation("內容已經說完") == "內容已經說完。"
     assert ensure_terminal_punctuation("內容已經說完！") == "內容已經說完！"

tests/test_quality_runtime.py

@@ -7,7 +7,12 @@ import pytest
 import soundfile as sf
 import torch
 
-from production import count_speech_units, normalize_spoken_forms, split_text_for_tts
+from production import (
+    count_network_endpoint_duration_units,
+    count_speech_units,
+    normalize_spoken_forms,
+    split_text_for_tts,
+)
 from quality_runtime import (
     ADAPTIVE_CASCADE_STAGE_LIMITS,
     BASE_GENERATION_POLICY,
@@ -24,6 +29,7 @@ from quality_runtime import (
     WHISPER_MODEL_ID,
     WHISPER_RETURN_ATTENTION_MASK,
     WHISPER_REVISION,
+    CandidateGenerationEvidence,
     CandidateObservation,
     CandidateVerification,
     CascadeResult,
@@ -31,6 +37,7 @@ from quality_runtime import (
     FinalOutputRejectedError,
     LazySquimObjective,
     LazyWhisperASR,
+    LocalIndependentGateEvidence,
     NoQualifiedCandidateError,
     RELEASE_SPEAKER_TRIGGER_SECONDS,
     SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP,
@@ -42,6 +49,7 @@ from quality_runtime import (
     _split_whisper_audio,
     active_audio_median_f0_hz,
     active_voiced_duration_seconds,
+    candidate_gate_evidence,
     candidate_chunk_transition_score,
     candidate_limit_for_chunk_budget,
     cosine_similarity,
@@ -49,6 +57,8 @@ from quality_runtime import (
     format_cascade_evidence_log,
     generation_cfg_for_candidate_offset,
     generation_policy_for_candidate_offset,
+    intersect_local_semantic_verification,
+    local_candidate_has_coverage_eligibility,
     load_pinned_whisper_runtime,
     load_pinned_verification_whisper_runtime,
     load_pinned_squim_objective_runtime,
@@ -59,6 +69,7 @@ from quality_runtime import (
     require_verified_final_output,
     resolve_request_seed,
     run_adaptive_cascade,
+    run_coverage_adaptive_cascade,
     select_k_candidate_sequences,
     speaker_embedding_from_audio,
     speaker_evidence_from_audio,
@@ -1446,6 +1457,271 @@ def test_joined_output_qualification_returns_local_result_only_when_joined_passe
     assert qualify_trajectory_with_joined_output(local, joined) is local
 
 
+def test_independent_local_semantic_reject_preserves_primary_acoustic_evidence():
+    target = normalize_spoken_forms(
+        "https://museum.example.tw/path"
+    )
+    artifact = ChunkCandidateArtifact(
+        speaker_embedding=np.asarray([1.0, 0.0], dtype=np.float32),
+        rms_db=-20.0,
+        median_f0_hz=180.0,
+    )
+    primary = verify_trajectory(
+        [
+            CandidateObservation(
+                target_text=target,
+                transcript_text=target,
+                audio_duration_seconds=2.0,
+                speaker_similarity=0.70,
+                begin_speaker_similarity=0.60,
+                end_speaker_similarity=0.55,
+                pace_cps=4.0,
+                squim_stoi=0.80,
+                squim_pesq=1.50,
+                squim_si_sdr=4.0,
+            )
+        ],
+        chunk_artifacts=(artifact,),
+        max_pace_cps=4.3,
+        squim_gate_enabled=True,
+        min_squim_stoi=0.60,
+        min_squim_pesq=1.12,
+        max_boundary_speaker_drop=0.10,
+    )
+    independent = verify_trajectory(
+        [
+            CandidateObservation(
+                target_text=target,
+                transcript_text=target.replace("museum", "museuman"),
+                audio_duration_seconds=2.0,
+            )
+        ],
+        speaker_gate_enabled=False,
+    )
+
+    combined = intersect_local_semantic_verification(
+        primary,
+        independent,
+        (0,),
+    )
+
+    assert primary.passed
+    assert independent.rejection_reasons == (
+        "chunk_0:semantic_gate",
+        "chunk_0:network_protected_span_mismatch",
+    )
+    assert not combined.passed
+    assert math.isinf(combined.score)
+    result = combined.candidate_results[0]
+    primary_result = primary.candidate_results[0]
+    assert result.comparison is primary_result.comparison
+    assert result.speaker_similarity == primary_result.speaker_similarity
+    assert result.boundary_speaker_drop == primary_result.boundary_speaker_drop
+    assert result.squim_stoi == primary_result.squim_stoi
+    assert result.squim_pesq == primary_result.squim_pesq
+    assert result.squim_si_sdr == primary_result.squim_si_sdr
+    assert result.rejection_reasons == (
+        "semantic_gate",
+        "network_protected_span_mismatch",
+    )
+    assert combined.chunk_artifacts is primary.chunk_artifacts
+    # Boundary-only fallback cannot erase an independent semantic rejection.
+    assert (
+        _sequence_fallback_candidate_result(
+            result,
+            max_local_boundary_speaker_drop=0.15,
+        )
+        is result
+    )
+
+
+def test_independent_local_semantic_pass_preserves_boundary_proxy_unchanged():
+    target = "這段網路片段內容完整"
+    primary = verify_trajectory(
+        [
+            CandidateObservation(
+                target_text=target,
+                transcript_text=target,
+                audio_duration_seconds=2.0,
+                speaker_similarity=0.70,
+                begin_speaker_similarity=0.80,
+                end_speaker_similarity=0.68,
+            )
+        ],
+    )
+    independent = verify_trajectory(
+        [CandidateObservation(target, target, 2.0)],
+        speaker_gate_enabled=False,
+    )
+
+    combined = intersect_local_semantic_verification(
+        primary,
+        independent,
+        (0,),
+    )
+
+    assert not primary.passed
+    assert primary.rejection_reasons == ("chunk_0:boundary_speaker_drop",)
+    assert independent.passed
+    assert combined is primary
+    relaxed = _sequence_fallback_candidate_result(
+        combined.candidate_results[0],
+        max_local_boundary_speaker_drop=0.15,
+    )
+    assert relaxed.passed
+    assert relaxed.speaker_similarity == primary.candidate_results[0].speaker_similarity
+
+
+@pytest.mark.parametrize(
+    "indices",
+    [(), (0, 0), (1, 0), (2,), (True,)],
+)
+def test_independent_local_semantic_intersection_rejects_unbound_indices(
+    indices,
+):
+    primary = verify_trajectory(
+        [
+            CandidateObservation("第一段", "第一段", 1.0),
+            CandidateObservation("第二段", "第二段", 1.0),
+        ],
+        speaker_gate_enabled=False,
+    )
+    independent = verify_trajectory(
+        [CandidateObservation("第一段", "第一段", 1.0)],
+        speaker_gate_enabled=False,
+    )
+
+    with pytest.raises(ValueError, match="primary indices"):
+        intersect_local_semantic_verification(
+            primary,
+            independent,
+            indices,
+        )
+
+
+def test_local_coverage_eligibility_matches_the_frozen_boundary_proxy():
+    target = "網路片段內容完整"
+    strict = verify_candidate(
+        CandidateObservation(target, target, 1.0),
+        speaker_gate_enabled=False,
+    )
+    boundary = verify_candidate(
+        CandidateObservation(
+            target,
+            target,
+            2.0,
+            speaker_similarity=0.70,
+            begin_speaker_similarity=0.80,
+            end_speaker_similarity=0.68,
+        )
+    )
+    semantic = verify_candidate(
+        CandidateObservation(target, target[:-1], 1.0),
+        speaker_gate_enabled=False,
+    )
+
+    assert local_candidate_has_coverage_eligibility(
+        strict,
+        max_local_boundary_speaker_drop=0.15,
+    )
+    assert local_candidate_has_coverage_eligibility(
+        boundary,
+        max_local_boundary_speaker_drop=(
+            SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP
+        ),
+    )
+    assert not local_candidate_has_coverage_eligibility(
+        boundary,
+        max_local_boundary_speaker_drop=0.10,
+    )
+    assert not local_candidate_has_coverage_eligibility(
+        semantic,
+        max_local_boundary_speaker_drop=0.15,
+    )
+
+
+def test_canonical_attempt_evidence_distinguishes_local_v3_pass_and_not_run():
+    chunks = ("普通內容完整", "網路內容完整")
+    primary = verify_trajectory(
+        [
+            CandidateObservation(chunks[0], chunks[0], 1.0),
+            CandidateObservation(chunks[1], chunks[1], 1.0),
+        ],
+        speaker_gate_enabled=False,
+    )
+    bounded_result = candidate_gate_evidence(primary.candidate_results[1])
+    local_independent = (
+        LocalIndependentGateEvidence(False, None, 0, None),
+        LocalIndependentGateEvidence(True, True, 1, bounded_result),
+    )
+
+    result = run_adaptive_cascade(
+        chunks,
+        900,
+        lambda candidate_chunks, seed: tuple(candidate_chunks),
+        lambda trajectory, candidate_chunks, seed: CandidateVerification(
+            primary,
+            independent_local_results=local_independent,
+        ),
+        max_candidates=1,
+    )
+    payload = json.loads(
+        format_cascade_evidence_log(
+            result.diagnostics,
+            outcome="returned",
+            generated_chunk_limit=32,
+            selection=result,
+        ).removeprefix(CASCADE_EVIDENCE_LOG_PREFIX)
+    )
+    attempt = payload["attempts"][0]
+
+    assert payload["schema_version"] == 4
+    assert attempt["independent_local_evidence_complete"] is True
+    assert attempt["independent_local_results"] == [
+        {
+            "attempted": False,
+            "passed": None,
+            "proof_count": 0,
+            "result": None,
+        },
+        {
+            "attempted": True,
+            "passed": True,
+            "proof_count": 1,
+            "result": attempt["local_results"][1],
+        },
+    ]
+
+
+@pytest.mark.parametrize(
+    "bad_evidence",
+    [
+        (LocalIndependentGateEvidence(False, None, 1.0, None),),
+        (LocalIndependentGateEvidence(False, None, 33, None),),
+        (LocalIndependentGateEvidence(True, None, 1, None),),
+    ],
+)
+def test_candidate_verification_rejects_malformed_local_v3_attestation(
+    bad_evidence,
+):
+    verification = verify_trajectory(
+        [CandidateObservation("內容", "內容", 1.0)],
+        speaker_gate_enabled=False,
+    )
+
+    with pytest.raises(TypeError, match="independent-local"):
+        run_adaptive_cascade(
+            ("內容",),
+            901,
+            lambda chunks, seed: tuple(chunks),
+            lambda trajectory, chunks, seed: CandidateVerification(
+                verification,
+                independent_local_results=bad_evidence,
+            ),
+            max_candidates=1,
+        )
+
+
 def test_cascade_evidence_success_is_content_free_canonical_and_behavior_golden():
     private_target = "PRIVATE_TARGET_ALPHA"
     private_transcript = "PRIVATE_TRANSCRIPT_OMEGA"
@@ -2749,6 +3025,51 @@ def test_candidate_limit_obeys_generated_text_unit_budget(text_units, expected_l
     assert limit * text_units <= 800
 
 
+def test_network_endpoint_units_do_not_change_k32_or_800_work_budget():
+    text = " coastwatch 點 example 點 T W 斜線 tide。"
+    public_units = count_speech_units(text)
+    endpoint_units = count_network_endpoint_duration_units(text)
+
+    assert (public_units, endpoint_units) == (12, 17)
+    assert candidate_limit_for_chunk_budget(
+        1,
+        total_text_units=public_units,
+        max_generated_text_units=800,
+    ) == 32
+    assert candidate_limit_for_chunk_budget(
+        1,
+        total_text_units=endpoint_units,
+        max_generated_text_units=800,
+    ) == 32
+    evidence = CandidateGenerationEvidence(
+        chunk_indices=(0,),
+        chunk_text_units=(public_units,),
+        scheduled_cfg=3.0,
+        effective_cfgs=(3.0,),
+        floor_reasons=((),),
+    )
+    result = run_coverage_adaptive_cascade(
+        (text,),
+        902,
+        lambda chunks, seed: tuple(chunks),
+        lambda trajectory, chunks, seed: verify_trajectory(
+            [CandidateObservation(text, text, 1.0)],
+            chunk_artifacts=[ChunkCandidateArtifact(rms_db=-20.0)],
+            speaker_gate_enabled=False,
+        ),
+        lambda trajectory, chunks, seed: pytest.fail("no refill expected"),
+        sequence_final_verifier=lambda result, chunks: pytest.fail(
+            "no sequence expected"
+        ),
+        generation_evidence_factory=(
+            lambda candidate_index, seed, chunk_indices, chunks: evidence
+        ),
+        max_generated_text_units=800,
+    )
+    assert result.generated_text_units == public_units
+    assert result.diagnostics.attempts[0].chunk_text_units == (public_units,)
+
+
 def test_candidate_limit_requires_complete_positive_text_unit_budget():
     with pytest.raises(ValueError, match="provided together"):
         candidate_limit_for_chunk_budget(1, total_text_units=40)

tests/test_release_pins.py

@@ -3,6 +3,8 @@ import hashlib
 from pathlib import Path
 from types import SimpleNamespace
 
+from quality_runtime import LocalIndependentGateEvidence
+
 
 ROOT = Path(__file__).resolve().parents[1]
 
@@ -409,6 +411,232 @@ def test_app_rejects_silent_text_and_coalesces_before_runtime_budgeting():
     )
 
 
+def test_network_endpoint_headroom_is_isolated_from_public_unit_contracts():
+    app_source = (ROOT / "app.py").read_text(encoding="utf-8")
+    production_source = (ROOT / "production.py").read_text(encoding="utf-8")
+    app_tree = ast.parse(app_source)
+    production_tree = ast.parse(production_source)
+    app_functions = {
+        node.name: node
+        for node in app_tree.body
+        if isinstance(node, ast.FunctionDef)
+    }
+    production_functions = {
+        node.name: node
+        for node in production_tree.body
+        if isinstance(node, ast.FunctionDef)
+    }
+    generate_source = ast.get_source_segment(
+        app_source,
+        app_functions["_generate_chunk"],
+    )
+    public_counter_source = ast.get_source_segment(
+        production_source,
+        production_functions["count_speech_units"],
+    )
+    network_counter_source = ast.get_source_segment(
+        production_source,
+        production_functions["count_network_endpoint_duration_units"],
+    )
+
+    assert generate_source is not None
+    assert public_counter_source is not None
+    assert network_counter_source is not None
+    assert "count_network_endpoint_duration_units(text)" in generate_source
+    assert "if network_conditioned" in generate_source
+    assert "duration_units=endpoint_duration_units" in generate_source
+    assert "min_len = 2" in generate_source
+    assert '"max_len": hard_stop_steps' in generate_source
+    assert "expected_steps=expected_steps" in generate_source
+    assert "hard_stop_steps=hard_stop_steps" in generate_source
+    assert "duration_counter=" in generate_source
+    assert "divisor = 2 if token.isdigit() else 4" in public_counter_source
+    assert "math.ceil(ascii_run_length / 2)" in network_counter_source
+    assert production_source.count(
+        "count_network_endpoint_duration_units("
+    ) == 1
+
+
+def test_network_local_dual_asr_capability_is_range_bound_for_initial_and_refill():
+    app_source = (ROOT / "app.py").read_text(encoding="utf-8")
+    quality_source = (ROOT / "quality_runtime.py").read_text(encoding="utf-8")
+    app_tree = ast.parse(app_source)
+    quality_tree = ast.parse(quality_source)
+    app_functions = {
+        node.name: node
+        for node in app_tree.body
+        if isinstance(node, ast.FunctionDef)
+    }
+    quality_functions = {
+        node.name: node
+        for node in quality_tree.body
+        if isinstance(node, ast.FunctionDef)
+    }
+    helper_source = ast.get_source_segment(
+        app_source,
+        app_functions["_verify_network_local_asr_intersection"],
+    )
+    initial_source = ast.get_source_segment(
+        app_source,
+        app_functions["_qualify_candidate_trajectory_audio"],
+    )
+    refill_source = ast.get_source_segment(
+        app_source,
+        app_functions["_verify_refill_candidate_trajectory_audio"],
+    )
+    intersection_source = ast.get_source_segment(
+        quality_source,
+        quality_functions["intersect_local_semantic_verification"],
+    )
+
+    assert all(
+        source is not None
+        for source in (
+            helper_source,
+            initial_source,
+            refill_source,
+            intersection_source,
+        )
+    )
+    assert "proof_rows[index] for index in selected_indices" in helper_source
+    assert "transcriber=transcribe_verification_whisper" in helper_source
+    assert "semantic_only=True" in helper_source
+    assert "network_fragment_proofs=independent_proof_rows" in helper_source
+    assert "intersect_local_semantic_verification(" in helper_source
+    assert "[BlueMagpie] network local independent " in helper_source
+    assert "proof_count=" in helper_source
+    assert "transcript_text" not in helper_source
+    for caller_source in (initial_source, refill_source):
+        turbo_index = caller_source.index(
+            "local_verification = _verify_trajectory_audio("
+        )
+        intersection_index = caller_source.index(
+            "_verify_network_local_asr_intersection("
+        )
+        assert turbo_index < intersection_index
+        assert "proof_rows = _network_fragment_proof_rows(" in caller_source
+        assert "network_fragment_proofs=proof_rows" in caller_source
+        assert "proof_rows," in caller_source[intersection_index:]
+        assert "candidate_seed=" in caller_source[intersection_index:]
+    assert "semantic_reasons = [\"semantic_gate\"]" in intersection_source
+    assert "\"network_protected_span_mismatch\"" in intersection_source
+    assert "chunk_artifacts=primary_verification.chunk_artifacts" in (
+        intersection_source
+    )
+    assert "CASCADE_EVIDENCE_SCHEMA_VERSION = 4" in quality_source
+    assert "local_candidate_has_coverage_eligibility(" in helper_source
+    assert "independent_local_results=" in initial_source
+    assert "independent_local_results=" in refill_source
+    assert '"independent_local_evidence_complete"' in quality_source
+    assert '"independent_local_results"' in quality_source
+
+
+def test_network_local_dual_asr_runtime_reuses_exact_proof_rows_and_logs_no_text(
+    capsys,
+):
+    source = (ROOT / "app.py").read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    function = next(
+        node
+        for node in tree.body
+        if (
+            isinstance(node, ast.FunctionDef)
+            and node.name == "_verify_network_local_asr_intersection"
+        )
+    )
+    module = ast.Module(
+        body=[
+            ast.ImportFrom(
+                module="__future__",
+                names=[ast.alias(name="annotations")],
+                level=0,
+            ),
+            function,
+        ],
+        type_ignores=[],
+    )
+    ast.fix_missing_locations(module)
+
+    calls = []
+    proof = object()
+    skipped_proof = object()
+    proof_rows = ((), (proof,), (skipped_proof,))
+    primary_results = (object(), object(), object())
+    turbo = SimpleNamespace(candidate_results=primary_results)
+    independent_result = SimpleNamespace(
+        passed=True,
+        rejection_reasons=(),
+        comparison=SimpleNamespace(
+            cer=0.0,
+            prefix_cer=0.0,
+            suffix_cer=0.0,
+            extra_tail_units=0,
+        ),
+    )
+    independent = SimpleNamespace(candidate_results=(independent_result,))
+    verification_transcriber = object()
+
+    def fake_verify(*args, **kwargs):
+        calls.append(("verify", args, kwargs))
+        return independent
+
+    def fake_intersect(*args):
+        calls.append(("intersect", args))
+        return "combined"
+
+    namespace = {
+        "_verify_trajectory_audio": fake_verify,
+        "QUALITY_FINAL_ASR_MAX_NEW_TOKENS": 440,
+        "SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP": 0.15,
+        "transcribe_verification_whisper": verification_transcriber,
+        "intersect_local_semantic_verification": fake_intersect,
+        "local_candidate_has_coverage_eligibility": (
+            lambda result, **_kwargs: result is primary_results[1]
+        ),
+        "LocalIndependentGateEvidence": LocalIndependentGateEvidence,
+        "candidate_gate_evidence": lambda result: ("bounded", result),
+    }
+    exec(compile(module, "<isolated-network-local>", "exec"), namespace)
+    result = namespace["_verify_network_local_asr_intersection"](
+        turbo,
+        ("ordinary-audio", "network-audio", "ordinary-audio-2"),
+        ("PRIVATE_ORDINARY_A", "PRIVATE_NETWORK_TEXT", "PRIVATE_ORDINARY_B"),
+        "anchor",
+        proof_rows,
+        candidate_seed=123,
+    )
+
+    assert result[0] == "combined"
+    verify_call = calls[0]
+    assert verify_call[0] == "verify"
+    assert verify_call[1][:4] == (
+        ("network-audio",),
+        ("PRIVATE_NETWORK_TEXT",),
+        "anchor",
+        1.0,
+    )
+    assert verify_call[1][4] == 440
+    assert verify_call[2]["transcriber"] is verification_transcriber
+    assert verify_call[2]["semantic_only"] is True
+    selected_proofs = verify_call[2]["network_fragment_proofs"]
+    assert selected_proofs == ((proof,),)
+    assert selected_proofs[0] is proof_rows[1]
+    assert calls[1] == (
+        "intersect",
+        (turbo, independent, (1,)),
+    )
+    evidence = result[1]
+    assert evidence[0] == LocalIndependentGateEvidence(False, None, 0, None)
+    assert evidence[1].attempted is True
+    assert evidence[1].passed is True
+    assert evidence[1].proof_count == 1
+    assert evidence[1].result == ("bounded", independent_result)
+    assert evidence[2] == LocalIndependentGateEvidence(False, None, 1, None)
+    log = capsys.readouterr().out
+    assert "seed=123 local_chunk_index=1 proof_count=1 passed=True" in log
+    assert "PRIVATE_" not in log
+
+
 def test_app_emits_one_canonical_content_free_evidence_line_per_terminal_outcome():
     source = (ROOT / "app.py").read_text(encoding="utf-8")
     tree = ast.parse(source)
@@ -441,7 +669,10 @@ def test_app_emits_one_canonical_content_free_evidence_line_per_terminal_outcome
     assert "error.diagnostics" in synthesize_source
     assert "cascade.diagnostics" in synthesize_source
     assert "final_output=final_evidence" in synthesize_source
-    assert "CandidateVerification(local_verification)" in qualifier_source
+    assert "CandidateVerification(" in qualifier_source
+    assert "independent_local_results=independent_local_results" in (
+        qualifier_source
+    )
     assert "joined_evidence = trajectory_gate_evidence(joined_verification)" in (
         qualifier_source
     )
@@ -689,7 +920,7 @@ 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():
+def test_space_hard_intersects_dual_asr_on_whole_and_proven_network_locals():
     source = (ROOT / "app.py").read_text(encoding="utf-8")
     tree = ast.parse(source)
     functions = {
@@ -714,6 +945,10 @@ def test_space_hard_intersects_dual_asr_only_on_exact_whole_waveforms():
         source,
         functions["_verify_refill_candidate_trajectory_audio"],
     )
+    network_local_source = ast.get_source_segment(
+        source,
+        functions["_verify_network_local_asr_intersection"],
+    )
     synthesize_source = ast.get_source_segment(source, functions["_synthesize"])
 
     assert all(
@@ -724,6 +959,7 @@ def test_space_hard_intersects_dual_asr_only_on_exact_whole_waveforms():
             qualify_source,
             sequence_source,
             refill_source,
+            network_local_source,
             synthesize_source,
         )
     )
@@ -733,8 +969,12 @@ def test_space_hard_intersects_dual_asr_only_on_exact_whole_waveforms():
     assert "semantic_only=True" in independent_source
     assert "cache.verify(" in independent_source
     assert "VERIFICATION_ASR_PROFILE" in independent_source
-    assert "return _verify_trajectory_audio(" in refill_source
+    assert "local_verification = _verify_trajectory_audio(" in refill_source
     assert "_verify_independent_whole_audio" not in refill_source
+    assert "transcriber=transcribe_verification_whisper" in network_local_source
+    assert "semantic_only=True" in network_local_source
+    assert "network_fragment_proofs=independent_proof_rows" in network_local_source
+    assert "intersect_local_semantic_verification(" in network_local_source
 
     local_index = qualify_source.index("local_verification = _verify_trajectory_audio(")
     local_fail_index = qualify_source.index("if not local_verification.passed:")