Harden hybrid network inference recovery

README.md

@@ -53,7 +53,7 @@ Barbet 另固定在 `6fcd7ce4aa37f2250a3242995bef0fbc3b026ba8`，
 | Sparse completion-headroom policy | every fourth retry: 4.2 CJK / 3.6 ASCII-mixed units/sec + 1 latent step |
 | Short-text guidance | minimum CFG 3.0 at no more than 6 speech units |
 | Generation guidance | fixed mixed-CFG assignment within the bounded cascade; public NFE fixed at 10 |
-| Email / URL frontend | explicit Taiwan-Mandarin letter names for scheme and opaque labels, digit-by-digit numbers, and audible separators |
+| 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 |
@@ -70,7 +70,7 @@ Barbet 另固定在 `6fcd7ce4aa37f2250a3242995bef0fbc3b026ba8`，
 | Final output gate | re-verify joined/faded/RMS-matched/speed-adjusted whole waveform; fail closed |
 | Runtime budget | at most 20 generated TTS chunks and 800 generated speech units per request; NFE fixed at 10 |
 | Maximum chunk | ordinary text 80 speech units; URL/email-bearing generation chunks 36 units |
-| Minimum chunk | 12 speech units where feasible; genuine short requests/natural short sentence boundaries are preserved |
+| Minimum chunk | ordinary text 12 speech units; URL/email-bearing chunks 8 units; genuine short non-network requests/natural short sentence boundaries are preserved |
 | Crossfade / internal edge fade | semantic boundary 80 ms / 80 ms; proven URL/email internal boundary 5 ms / 5 ms with no inserted pause |
 | Chunk RMS adjustment | at most 4 dB |
 | Generated continuation context | 0 sec |
@@ -137,13 +137,19 @@ transcript。
 日期、24 小時制時間、百分比、常見單位與大寫 acronym/model code 會先轉成保守的
 zh-TW spoken form，例如 `2026/07/16`、`15:30`、`12.5%` 與 `10 km`。ASR 會先統一
 繁簡字形再評分，避免把正確的台灣華語輸出誤判為內容錯誤。
-Email 與 URL 會以可辨識的語義讀法展開：scheme、local/domain/path 的 opaque ASCII
-labels 逐字使用台灣華語字母名，數字逐位朗讀，分隔符明確朗讀（例如 `.tw`
-讀成「點、踢、達不溜」）。
+Email 與 URL 會以混合式可辨識讀法展開：一般 local/domain/path label 保留 lexical ASCII
+（例如 `tour`、`help`），scheme、全大寫 atom 與短 suffix 使用明確 ASCII letter tokens，數字逐位
+朗讀，分隔符也明確朗讀（例如 `.tw` 讀成「點、T、W」）。
 完整 identifier 仍保留為 joined、full-large-v3 與 final ASR 的 exact protected target；只有模型
 generation 會在由原始 ASCII grammar 證明的 scheme/domain/path/query/email component 邊界切段，
-以 32 units 為目標、36 units 為硬上限。這些 identifier 內部邊界不插入 pause，只使用 5 ms
-fade/crossfade；component proof 無法完整重建、非 ASCII IRI 或單一不可拆 component 超限時會 fail closed。
+以 32 units 為目標、36 units 為硬上限、8 units 為 network minimum。Email 第一輪必須在 `小老鼠`
+之前切開；有 scheme 的 URL 第一輪必須在 `冒號 斜線 斜線` 之後切開，第一輪 DP 不得跨越這兩類
+grammar boundary。只有像 `a@b.co`、`https://a.tw` 這類 identifier 的某一側本身不足 8 units 且
+第一輪無解時，才放寬該短 identifier 的 preferred cut 並重跑相同 component-safe DP；其他強制切點
+不變。Identifier 內部邊界不插入 pause，只使用 5 ms fade/crossfade；component proof 無法完整
+重建、非 ASCII IRI 或單一不可拆 component 超限時會 fail closed。
+Local fragment gate 只接受 range-bound exact proof，並檢查所有最佳 edit alignment；重複的普通文字
+不能借用受保護片段的 proof。Joined、whole 與 final gate 不套用這項 local-only canonicalization。
 這能降低模型把不常見 TLD 自動補成 `.com` 的風險；一般英文句子不會套用這個規則。
 URL 與後續英文 prose 應以空白或中文標點分隔；未分隔的 RFC path punctuation 會視為 URL
 本身的一部分並納入 exact gate。Quoted email local-part 暫不支援，輸入時會直接 fail closed。

app.py

@@ -19,8 +19,10 @@ from bluemagpie import BlueMagpieModel
 from production import (
     StopHysteresisController,
     GenerationChunkSpec,
+    NetworkFragmentProof,
     active_pace_correction_speed,
     apply_loudness_floor,
+    canonicalize_asr_network_fragments,
     coalesce_text_chunks,
     contains_network_identifier,
     count_speech_units,
@@ -115,6 +117,7 @@ MIN_CHUNK_CHARS = 12
 CROSSFADE_MS = 80.0
 CHUNK_EDGE_FADE_MS = 80.0
 CHUNK_RMS_MATCH_DB = 4.0
+NETWORK_GENERATION_MIN_UNITS = 8
 NETWORK_GENERATION_TARGET_UNITS = 32
 NETWORK_GENERATION_MAX_UNITS = 36
 NETWORK_INTERNAL_FADE_MS = 5.0
@@ -442,13 +445,26 @@ def _verify_trajectory_audio(
     release_speaker_gate: bool = False,
     transcriber=transcribe_whisper,
     semantic_only: bool = False,
+    network_fragment_proofs: (
+        tuple[tuple[NetworkFragmentProof, ...], ...] | None
+    ) = None,
 ):
     if len(trajectory) != len(chunks):
         return verify_trajectory(())
+    if network_fragment_proofs is not None and len(network_fragment_proofs) != len(
+        chunks
+    ):
+        raise ValueError("network fragment proofs must align with local chunks")
     encoder = None
     observations: list[CandidateObservation] = []
     artifacts: list[ChunkCandidateArtifact] = []
-    for chunk, audio in zip(chunks, trajectory, strict=True):
+    proof_rows = network_fragment_proofs or ((),) * len(chunks)
+    for chunk, audio, fragment_proofs in zip(
+        chunks,
+        trajectory,
+        proof_rows,
+        strict=True,
+    ):
         prepared = prepare_candidate_audio(
             audio,
             SR,
@@ -475,6 +491,19 @@ def _verify_trajectory_audio(
         except ValueError:
             duration = 0.0
         transcript = prepared.transcript_text
+        if fragment_proofs:
+            fragment_evidence = canonicalize_asr_network_fragments(
+                transcript,
+                chunk,
+                fragment_proofs,
+            )
+            # A range-bound identifier mismatch is a hard local semantic
+            # failure even when its contribution to whole-chunk CER is small.
+            transcript = (
+                fragment_evidence.transcript_text
+                if fragment_evidence.passed
+                else ""
+            )
         speaker_similarity = None
         begin_similarity = None
         end_similarity = None
@@ -611,6 +640,37 @@ def _verify_independent_whole_audio(
     )
 
 
+def _network_fragment_proof_rows(
+    chunks: tuple[str, ...],
+    chunk_specs: tuple[GenerationChunkSpec, ...] | None,
+) -> tuple[tuple[NetworkFragmentProof, ...], ...] | None:
+    """Validate explicit planner provenance before any local ASR relaxation."""
+
+    if chunk_specs is None:
+        return None
+    if len(chunk_specs) != len(chunks):
+        raise ValueError("network fragment provenance must align with chunks")
+    rows: list[tuple[NetworkFragmentProof, ...]] = []
+    for chunk, spec in zip(chunks, chunk_specs, strict=True):
+        proofs = spec.network_fragment_proofs
+        if spec.text != chunk:
+            raise ValueError("network fragment provenance text does not match")
+        if spec.network_conditioned:
+            if (
+                not proofs
+                or len(proofs) != len(spec.network_span_indices)
+                or tuple(proof.span_index for proof in proofs)
+                != spec.network_span_indices
+                or tuple(proof.full_spoken_proof for proof in proofs)
+                != spec.network_full_spoken_proofs
+            ):
+                raise ValueError("network-conditioned chunk lacks exact fragment proof")
+        elif proofs or spec.network_full_spoken_proofs:
+            raise ValueError("ordinary chunk carries network fragment proof")
+        rows.append(proofs)
+    return tuple(rows)
+
+
 def _assemble_trajectory_audio(
     trajectory: tuple[np.ndarray, ...],
     chunks: tuple[str, ...],
@@ -742,6 +802,7 @@ def _qualify_candidate_trajectory_audio(
         chunks,
         anchor,
         playback_speed,
+        network_fragment_proofs=_network_fragment_proof_rows(chunks, chunk_specs),
     )
     if not local_verification.passed:
         return CandidateVerification(local_verification)
@@ -802,6 +863,7 @@ def _verify_refill_candidate_trajectory_audio(
     chunks: tuple[str, ...],
     anchor: np.ndarray,
     playback_speed: float,
+    chunk_specs: tuple[GenerationChunkSpec, ...] | None = None,
 ):
     """Apply the unchanged strict local gates to one safe-duration refill."""
 
@@ -812,6 +874,7 @@ def _verify_refill_candidate_trajectory_audio(
         chunks,
         anchor,
         playback_speed,
+        network_fragment_proofs=_network_fragment_proof_rows(chunks, chunk_specs),
     )
 
 
@@ -922,6 +985,7 @@ def _synthesize(
                 raw_text,
                 text,
                 min_units=MIN_CHUNK_CHARS,
+                network_min_units=NETWORK_GENERATION_MIN_UNITS,
                 target_units=NETWORK_GENERATION_TARGET_UNITS,
                 network_max_units=NETWORK_GENERATION_MAX_UNITS,
                 ordinary_max_units=CHUNK_CHARS,
@@ -948,6 +1012,30 @@ def _synthesize(
     request_seed = resolve_request_seed(request_seed, secrets.randbelow)
     anchor = _speaker_anchor_array(centroid)
     independent_cache = WholeWaveformVerificationCache()
+    generation_context_by_seed: dict[int, CandidateGenerationContext] = {}
+
+    def generation_chunk_specs(
+        seed: int,
+        candidate_chunks: tuple[str, ...],
+    ) -> tuple[GenerationChunkSpec, ...] | None:
+        if chunk_specs is None:
+            return None
+        context = generation_context_by_seed.get(seed)
+        if context is None or len(context.chunk_indices) != len(candidate_chunks):
+            raise ValueError("candidate verification lacks generation provenance")
+        selected: list[GenerationChunkSpec] = []
+        for chunk_index, chunk in zip(
+            context.chunk_indices,
+            candidate_chunks,
+            strict=True,
+        ):
+            if not 0 <= chunk_index < len(chunk_specs):
+                raise ValueError("candidate verification provenance is out of range")
+            spec = chunk_specs[chunk_index]
+            if spec.text != chunk:
+                raise ValueError("candidate verification provenance text does not match")
+            selected.append(spec)
+        return tuple(selected)
 
     def candidate_cfg(candidate_ordinal: int) -> float:
         return generation_cfg_for_candidate_offset(
@@ -1032,6 +1120,10 @@ def _synthesize(
     ) -> tuple[np.ndarray, ...]:
         if generation_context.seed != seed:
             raise ValueError("generation context seed does not match the request seed")
+        previous_context = generation_context_by_seed.get(seed)
+        if previous_context is not None and previous_context != generation_context:
+            raise ValueError("one candidate seed cannot carry two generation contexts")
+        generation_context_by_seed[seed] = generation_context
         ordinals = generation_context.chunk_candidate_ordinals
         if not ordinals or len(set(ordinals)) != 1:
             raise ValueError("one generation call must use one candidate ordinal")
@@ -1113,7 +1205,7 @@ def _synthesize(
                     speed,
                     independent_cache,
                     candidate_seed=seed,
-                    chunk_specs=chunk_specs,
+                    chunk_specs=generation_chunk_specs(seed, candidate_chunks),
                 ),
                 lambda trajectory, candidate_chunks, seed: (
                     _qualify_candidate_trajectory_audio(
@@ -1124,7 +1216,7 @@ def _synthesize(
                         speed,
                         independent_cache,
                         candidate_seed=seed,
-                        chunk_specs=chunk_specs,
+                        chunk_specs=generation_chunk_specs(seed, candidate_chunks),
                     )
                     if len(chunks) == 1
                     else _verify_refill_candidate_trajectory_audio(
@@ -1132,6 +1224,7 @@ def _synthesize(
                         candidate_chunks,
                         anchor,
                         speed,
+                        generation_chunk_specs(seed, candidate_chunks),
                     )
                 ),
                 sequence_final_verifier=lambda sequence_result, candidate_chunks: (

production.py

@@ -1030,7 +1030,7 @@ def _canonicalize_target_proven_frequency_scales(
 
 
 def _zh_network_text(value: str) -> str:
-    """Spell opaque network labels and make their separators audible."""
+    """Read lexical labels naturally while keeping typed atoms explicit."""
 
     output: list[str] = []
     buffer: list[str] = []
@@ -1042,7 +1042,10 @@ def _zh_network_text(value: str) -> str:
         if token.isdigit():
             output.append(_zh_digit_sequence(token))
         elif token.isascii() and token.isalpha():
-            output.append(_zh_network_letters(token))
+            if token.casefold() == "www" or (len(token) > 1 and token.isupper()):
+                output.append(_zh_network_letters(token))
+            else:
+                output.append(token)
         else:
             output.append(token)
         buffer.clear()
@@ -1070,10 +1073,10 @@ def _zh_network_text(value: str) -> str:
 
 
 def _zh_network_letters(value: str) -> str:
-    """Return explicit Taiwan-Mandarin names for opaque ASCII letters."""
+    """Return model-native spaced uppercase letters for typed ASCII atoms."""
 
     return " ".join(
-        _ZH_NETWORK_LETTER_READINGS.get(character.upper(), character)
+        character.upper()
         for character in value
         if not character.isspace()
     )
@@ -2660,6 +2663,25 @@ def split_text_for_tts(text: str, max_chars: int = 80, min_chunk_chars: int = 12
     return chunks or [text]
 
 
+@dataclass(frozen=True)
+class NetworkFragmentProof:
+    """Range-bound local slice of one planner-proven network identifier.
+
+    ``chunk_start``/``chunk_end`` are offsets in the exact generation chunk.
+    ``parent_start``/``parent_end`` are offsets in ``full_spoken_proof``.  The
+    two slices must be byte-for-byte equal after NFC normalization.  Keeping
+    both coordinate systems prevents a repeated label elsewhere in the chunk
+    from borrowing a parent URL/email proof.
+    """
+
+    span_index: int
+    chunk_start: int
+    chunk_end: int
+    parent_start: int
+    parent_end: int
+    full_spoken_proof: str
+
+
 @dataclass(frozen=True)
 class GenerationChunkSpec:
     """One generation-only slice with immutable full-identifier provenance."""
@@ -2670,6 +2692,7 @@ class GenerationChunkSpec:
     network_span_indices: tuple[int, ...] = ()
     network_component_indices: tuple[tuple[int, int], ...] = ()
     network_full_spoken_proofs: tuple[str, ...] = ()
+    network_fragment_proofs: tuple[NetworkFragmentProof, ...] = ()
     boundary_after: str = "none"
 
     @property
@@ -2683,6 +2706,7 @@ class _NetworkGenerationSpan:
     end: int
     spoken_proof: str
     component_ranges: tuple[tuple[int, int], ...]
+    mandatory_cut_offsets: tuple[int, ...]
 
 
 def _raw_network_generation_matches(
@@ -2762,6 +2786,50 @@ def _network_component_ranges(
     return tuple(ranges)
 
 
+def _network_mandatory_cut_offsets(
+    spoken: str,
+    *,
+    kind: str,
+    absolute_start: int,
+) -> tuple[int, ...]:
+    """Return grammar boundaries that no generation chunk may cross.
+
+    Hybrid lexical labels are substantially easier for the model than spelling
+    every letter, but a whole email/URL still asks one trajectory to preserve a
+    long opaque identifier.  Split email local/domain parts and URL
+    scheme/remainder parts independently so coverage selection can combine
+    exact fragments without relaxing whole-output verification.
+    """
+
+    token_matches = tuple(re.finditer(r"\S+", spoken))
+    tokens = tuple(
+        _SIMPLIFIED_NETWORK_SYMBOL_READINGS.get(match.group(0), match.group(0))
+        for match in token_matches
+    )
+    offsets: list[int] = []
+    if kind == "email":
+        separators = [
+            match.start()
+            for token, match in zip(tokens, token_matches, strict=True)
+            if token == "小老鼠"
+        ]
+        if len(separators) != 1:
+            raise ValueError("email generation proof lacks one separator")
+        offsets.append(separators[0])
+    elif kind == "url":
+        scheme_ends = [
+            token_matches[index].end()
+            for index in range(2, len(tokens))
+            if tokens[index - 2 : index + 1] == ("冒號", "斜線", "斜線")
+        ]
+        if len(scheme_ends) > 1:
+            raise ValueError("URL generation proof has ambiguous scheme separators")
+        offsets.extend(scheme_ends)
+    else:
+        raise ValueError("network generation proof has an invalid kind")
+    return tuple(absolute_start + offset for offset in offsets)
+
+
 def _network_generation_spans(
     raw_text: str,
     normalized_text: str,
@@ -2810,12 +2878,28 @@ def _network_generation_spans(
             absolute_start=start,
             hard_max_units=hard_max_units,
         )
+        mandatory_cuts = _network_mandatory_cut_offsets(
+            spoken,
+            kind=kind,
+            absolute_start=start,
+        )
+        component_boundaries = {
+            boundary
+            for component in components
+            for boundary in component
+        }
+        if any(
+            cut <= start or cut >= end or cut not in component_boundaries
+            for cut in mandatory_cuts
+        ):
+            raise ValueError("mandatory network cut lacks component provenance")
         spans.append(
             _NetworkGenerationSpan(
                 start=start,
                 end=end,
                 spoken_proof=spoken,
                 component_ranges=components,
+                mandatory_cut_offsets=mandatory_cuts,
             )
         )
         cursor = end
@@ -2832,6 +2916,7 @@ def _network_generation_group_plan(
     original_boundaries: Sequence[int],
     spans: Sequence[_NetworkGenerationSpan],
     minimum: int,
+    network_minimum: int,
     target: int,
     network_maximum: int,
     ordinary_maximum: int,
@@ -2862,6 +2947,13 @@ def _network_generation_group_plan(
         for offset in component
     }
     preferred_cuts.update(component_cuts)
+    mandatory_cuts = {
+        offset
+        for span in group_spans
+        for offset in span.mandatory_cut_offsets
+    }
+    if not mandatory_cuts.issubset(component_cuts):
+        raise ValueError("mandatory network cuts lack component provenance")
     network_interiors = {
         offset
         for span in group_spans
@@ -2881,39 +2973,68 @@ def _network_generation_group_plan(
     }
 
     ordered = sorted(offset for offset in cuts if group_start <= offset <= group_end)
-    best: dict[int, tuple[int, int, int, int, tuple[int, ...]]] = {
-        group_end: (0, 0, 0, 0, ())
-    }
-    for start in reversed(ordered[:-1]):
-        selected: tuple[int, int, int, int, tuple[int, ...]] | None = None
-        for end in ordered:
-            if end <= start:
-                continue
-            chunk = text[start:end]
-            units = count_speech_units(chunk)
-            network_conditioned = any(
-                span.start < end and span.end > start for span in group_spans
-            )
-            maximum = network_maximum if network_conditioned else ordinary_maximum
-            if units > maximum:
-                continue
-            if units < minimum and not (start == group_start and end == group_end):
-                continue
-            remainder = best.get(end)
-            if remainder is None:
-                continue
-            candidate = (
-                1 + remainder[0],
-                int(end != group_end and end not in preferred_cuts) + remainder[1],
-                max(units, remainder[2]),
-                abs(target - units) + remainder[3],
-                (end,) + remainder[4],
-            )
-            if selected is None or candidate < selected:
-                selected = candidate
-        if selected is not None:
-            best[start] = selected
-    plan = best.get(group_start)
+    def solve(
+        active_mandatory_cuts: set[int],
+    ) -> tuple[int, int, int, int, tuple[int, ...]] | None:
+        best: dict[int, tuple[int, int, int, int, tuple[int, ...]]] = {
+            group_end: (0, 0, 0, 0, ())
+        }
+        for start in reversed(ordered[:-1]):
+            selected: tuple[int, int, int, int, tuple[int, ...]] | None = None
+            for end in ordered:
+                if end <= start:
+                    continue
+                if any(start < cut < end for cut in active_mandatory_cuts):
+                    continue
+                chunk = text[start:end]
+                units = count_speech_units(chunk)
+                network_conditioned = any(
+                    span.start < end and span.end > start for span in group_spans
+                )
+                maximum = (
+                    network_maximum if network_conditioned else ordinary_maximum
+                )
+                effective_minimum = (
+                    network_minimum if network_conditioned else minimum
+                )
+                if units > maximum:
+                    continue
+                if units < effective_minimum and not (
+                    start == group_start and end == group_end
+                ):
+                    continue
+                remainder = best.get(end)
+                if remainder is None:
+                    continue
+                candidate = (
+                    1 + remainder[0],
+                    int(end != group_end and end not in preferred_cuts)
+                    + remainder[1],
+                    max(units, remainder[2]),
+                    abs(target - units) + remainder[3],
+                    (end,) + remainder[4],
+                )
+                if selected is None or candidate < selected:
+                    selected = candidate
+            if selected is not None:
+                best[start] = selected
+        return best.get(group_start)
+
+    plan = solve(mandatory_cuts)
+    if plan is None:
+        # A tiny common identifier such as ``https://a.tw`` cannot place both
+        # sides of the preferred scheme cut above the network minimum. Retry
+        # only after proving which identifier-local side is intrinsically short;
+        # all other mandatory cuts and every grammar component boundary remain.
+        short_identifier_cuts = {
+            cut
+            for span in group_spans
+            for cut in span.mandatory_cut_offsets
+            if count_speech_units(text[span.start:cut]) < network_minimum
+            or count_speech_units(text[cut:span.end]) < network_minimum
+        }
+        if short_identifier_cuts:
+            plan = solve(mandatory_cuts - short_identifier_cuts)
     if plan is None:
         raise ValueError("network-bearing sentence cannot satisfy generation limits")
     output: list[tuple[int, int]] = []
@@ -3019,6 +3140,7 @@ def plan_generation_chunks(
     normalized_text: str,
     *,
     min_units: int = 12,
+    network_min_units: int = 8,
     target_units: int = 32,
     network_max_units: int = 36,
     ordinary_max_units: int = 80,
@@ -3034,10 +3156,17 @@ def plan_generation_chunks(
     if network_identifier_has_ambiguous_iri(raw_text):
         raise ValueError("network generation does not support non-ASCII IRI")
     minimum = max(1, int(min_units))
+    network_minimum = max(1, int(network_min_units))
     target = max(minimum, int(target_units))
     network_maximum = max(minimum, int(network_max_units))
     ordinary_maximum = max(minimum, int(ordinary_max_units))
-    if not minimum <= target <= network_maximum <= ordinary_maximum:
+    if not (
+        network_minimum
+        <= minimum
+        <= target
+        <= network_maximum
+        <= ordinary_maximum
+    ):
         raise ValueError("generation chunk limits are inconsistent")
 
     normalized = normalize_tts_text(normalized_text)
@@ -3088,6 +3217,7 @@ def plan_generation_chunks(
                     original_boundaries=(),
                     spans=spans,
                     minimum=minimum,
+                    network_minimum=network_minimum,
                     target=target,
                     network_maximum=network_maximum,
                     ordinary_maximum=ordinary_maximum,
@@ -3141,6 +3271,19 @@ def plan_generation_chunks(
             )
             if left < end and right > start
         )
+        fragment_proofs = tuple(
+            NetworkFragmentProof(
+                span_index=span_index,
+                chunk_start=max(start, spans[span_index].start) - start,
+                chunk_end=min(end, spans[span_index].end) - start,
+                parent_start=max(start, spans[span_index].start)
+                - spans[span_index].start,
+                parent_end=min(end, spans[span_index].end)
+                - spans[span_index].start,
+                full_spoken_proof=spans[span_index].spoken_proof,
+            )
+            for span_index in overlapping
+        )
         specs.append(
             GenerationChunkSpec(
                 text=chunk,
@@ -3151,6 +3294,7 @@ def plan_generation_chunks(
                 network_full_spoken_proofs=tuple(
                     spans[span_index].spoken_proof for span_index in overlapping
                 ),
+                network_fragment_proofs=fragment_proofs,
                 boundary_after=(
                     "network_internal"
                     if end in internal_boundaries
@@ -3166,6 +3310,63 @@ def plan_generation_chunks(
         for spec in specs
     ):
         raise ValueError("generation chunk exceeds its provenance-specific limit")
+    for spec in specs:
+        proofs = spec.network_fragment_proofs
+        if spec.network_conditioned:
+            if not (
+                len(proofs)
+                == len(spec.network_span_indices)
+                == len(spec.network_full_spoken_proofs)
+            ):
+                raise ValueError("network fragment provenance is incomplete")
+        elif proofs or spec.network_full_spoken_proofs:
+            raise ValueError("ordinary chunk carries network fragment provenance")
+        previous_chunk_end = 0
+        for span_index, full_proof, proof in zip(
+            spec.network_span_indices,
+            spec.network_full_spoken_proofs,
+            proofs,
+            strict=True,
+        ):
+            if (
+                proof.span_index != span_index
+                or proof.full_spoken_proof != full_proof
+                or not 0 <= proof.chunk_start < proof.chunk_end <= len(spec.text)
+                or not 0 <= proof.parent_start < proof.parent_end <= len(full_proof)
+                or proof.chunk_start < previous_chunk_end
+                or spec.text[proof.chunk_start : proof.chunk_end]
+                != full_proof[proof.parent_start : proof.parent_end]
+            ):
+                raise ValueError("network fragment provenance is inconsistent")
+            previous_chunk_end = proof.chunk_end
+
+    # The local fragments are not merely hints: taken across the plan they must
+    # form an exact, non-overlapping partition of every parent identifier.
+    for span_index, span in enumerate(spans):
+        partition = sorted(
+            (
+                proof.parent_start,
+                proof.parent_end,
+                proof.full_spoken_proof,
+            )
+            for spec in specs
+            for proof in spec.network_fragment_proofs
+            if proof.span_index == span_index
+        )
+        if (
+            not partition
+            or partition[0][0] != 0
+            or partition[-1][1] != len(span.spoken_proof)
+            or any(
+                left_end != right_start
+                for (_, left_end, _), (right_start, _, _) in zip(
+                    partition,
+                    partition[1:],
+                )
+            )
+            or any(full_proof != span.spoken_proof for _, _, full_proof in partition)
+        ):
+            raise ValueError("network fragment provenance does not partition its parent")
     return tuple(specs)
 
 
@@ -3710,6 +3911,344 @@ def _levenshtein_alignment(source: str, hypothesis: str) -> list[tuple[str, int,
     return operations
 
 
+def _protected_ranges_are_exact_in_all_optimal_alignments(
+    source: str,
+    hypothesis: str,
+    protected_ranges: Sequence[tuple[int, int]],
+) -> bool:
+    """Require protected source ranges to survive every optimal edit path.
+
+    A deterministic Levenshtein backtrace is insufficient when ordinary text
+    duplicates a protected fragment: one optimal path can match the protected
+    occurrence while another deletes it and matches the ordinary occurrence.
+    Prefix/suffix distances let us inspect every edge that belongs to at least
+    one globally optimal path.  Protected characters may only traverse equal
+    diagonal edges, and insertions may not be anchored inside the closed range
+    (including immediately before its first or after its last character).
+    """
+
+    ranges = tuple(protected_ranges)
+    if not ranges:
+        return True
+    if any(
+        isinstance(start, bool)
+        or isinstance(end, bool)
+        or not isinstance(start, int)
+        or not isinstance(end, int)
+        or not 0 <= start < end <= len(source)
+        for start, end in ranges
+    ):
+        raise ValueError("protected alignment range is invalid")
+
+    source_protected = [False] * len(source)
+    insertion_protected = [False] * (len(source) + 1)
+    for start, end in ranges:
+        for source_index in range(start, end):
+            source_protected[source_index] = True
+        for source_offset in range(start, end + 1):
+            insertion_protected[source_offset] = True
+
+    source_length = len(source)
+    hypothesis_length = len(hypothesis)
+    forward = [
+        [0] * (hypothesis_length + 1) for _ in range(source_length + 1)
+    ]
+    for source_index in range(source_length + 1):
+        forward[source_index][0] = source_index
+    for hypothesis_index in range(hypothesis_length + 1):
+        forward[0][hypothesis_index] = hypothesis_index
+    for source_index in range(1, source_length + 1):
+        for hypothesis_index in range(1, hypothesis_length + 1):
+            forward[source_index][hypothesis_index] = min(
+                forward[source_index - 1][hypothesis_index] + 1,
+                forward[source_index][hypothesis_index - 1] + 1,
+                forward[source_index - 1][hypothesis_index - 1]
+                + (
+                    source[source_index - 1]
+                    != hypothesis[hypothesis_index - 1]
+                ),
+            )
+
+    backward = [
+        [0] * (hypothesis_length + 1) for _ in range(source_length + 1)
+    ]
+    for source_index in range(source_length + 1):
+        backward[source_index][hypothesis_length] = source_length - source_index
+    for hypothesis_index in range(hypothesis_length + 1):
+        backward[source_length][hypothesis_index] = (
+            hypothesis_length - hypothesis_index
+        )
+    for source_index in range(source_length - 1, -1, -1):
+        for hypothesis_index in range(hypothesis_length - 1, -1, -1):
+            backward[source_index][hypothesis_index] = min(
+                backward[source_index + 1][hypothesis_index] + 1,
+                backward[source_index][hypothesis_index + 1] + 1,
+                backward[source_index + 1][hypothesis_index + 1]
+                + (source[source_index] != hypothesis[hypothesis_index]),
+            )
+
+    optimal_distance = forward[source_length][hypothesis_length]
+    for source_index in range(source_length + 1):
+        for hypothesis_index in range(hypothesis_length + 1):
+            prefix_distance = forward[source_index][hypothesis_index]
+            if (
+                source_index < source_length
+                and source_protected[source_index]
+                and prefix_distance
+                + 1
+                + backward[source_index + 1][hypothesis_index]
+                == optimal_distance
+            ):
+                return False
+            if (
+                source_index < source_length
+                and hypothesis_index < hypothesis_length
+                and source_protected[source_index]
+                and source[source_index] != hypothesis[hypothesis_index]
+                and prefix_distance
+                + 1
+                + backward[source_index + 1][hypothesis_index + 1]
+                == optimal_distance
+            ):
+                return False
+            if (
+                hypothesis_index < hypothesis_length
+                and insertion_protected[source_index]
+                and prefix_distance
+                + 1
+                + backward[source_index][hypothesis_index + 1]
+                == optimal_distance
+            ):
+                return False
+    return True
+
+
+@dataclass(frozen=True)
+class NetworkFragmentAsrEvidence:
+    """Result of exact, range-bound local network-fragment verification."""
+
+    transcript_text: str
+    protected_range_count: int
+    passed: bool
+
+
+_NETWORK_FRAGMENT_IDENTIFIER_CHARACTERS = (
+    "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
+    + "".join(_ZH_NETWORK_SYMBOL_READINGS)
+)
+_NETWORK_FRAGMENT_READING_TO_ASCII = {
+    reading: symbol for symbol, reading in _ZH_NETWORK_SYMBOL_READINGS.items()
+}
+
+
+def _network_fragment_token_ascii(token: str) -> str | None:
+    """Invert one typed frontend network atom without prose heuristics."""
+
+    canonical = _SIMPLIFIED_NETWORK_SYMBOL_READINGS.get(token, token)
+    letter = _network_letter_token(canonical)
+    if letter is not None:
+        return letter
+    if canonical.isascii() and canonical.isalnum():
+        # A hybrid frontend may intentionally retain a planner-bound opaque
+        # label (``tour``/``islandmuseum``) while spelling only its separators
+        # and short suffix. It is still an exact typed atom, never prose: the
+        # range object below binds its precise occurrence to the parent URL.
+        return canonical.casefold()
+    if canonical and all(character in _ZH_DIGITS for character in canonical):
+        return "".join(str(_ZH_DIGITS.index(character)) for character in canonical)
+    return _NETWORK_FRAGMENT_READING_TO_ASCII.get(canonical)
+
+
+def _network_fragment_ascii(spoken_fragment: str) -> str | None:
+    """Return the exact ASCII atom sequence for one planner-bound fragment."""
+
+    output: list[str] = []
+    for token in normalize_tts_text(spoken_fragment).split():
+        value = _network_fragment_token_ascii(token)
+        if value is None:
+            return None
+        output.append(value)
+    rendered = "".join(output)
+    return rendered if rendered and any(character.isalnum() for character in rendered) else None
+
+
+def _network_fragment_alignment_text(value: str) -> str:
+    """Composable orthographic key used only by the exact fragment gate."""
+
+    normalized = _T2S_CONVERTER.convert(
+        normalize_tts_eval_text(unicodedata.normalize("NFC", str(value or ""))).casefold()
+    )
+    return "".join(character for character in normalized if character.isalnum())
+
+
+def canonicalize_asr_network_fragments(
+    text: str,
+    target_text: str,
+    fragment_proofs: Sequence[NetworkFragmentProof],
+) -> NetworkFragmentAsrEvidence:
+    """Canonicalize and exact-gate local ASCII URL/email fragments.
+
+    This helper is deliberately separate from :func:`compare_asr_text`.  It is
+    valid only for generation chunks carrying the range objects emitted by
+    :func:`plan_generation_chunks`; whole/joined/final verification must not
+    call it.  Each literal replacement must equal the *entire* chunk-local
+    parent slice.  Substrings of a full URL cannot borrow its proof.
+    """
+
+    raw_text = unicodedata.normalize("NFC", str(text or ""))
+    target = unicodedata.normalize("NFC", str(target_text or ""))
+    if isinstance(fragment_proofs, (str, bytes)):
+        raise ValueError("network fragment proofs must be range objects")
+    try:
+        proofs = tuple(fragment_proofs)
+    except TypeError as error:
+        raise ValueError("network fragment proofs must be iterable") from error
+    if not proofs:
+        return NetworkFragmentAsrEvidence(raw_text, 0, True)
+    if not target:
+        raise ValueError("network fragment target must be non-empty")
+
+    previous_chunk_end = 0
+    typed_fragments: dict[str, str] = {}
+    for proof in proofs:
+        if not isinstance(proof, NetworkFragmentProof):
+            raise ValueError("network fragment proof has an invalid type")
+        integer_fields = (
+            proof.span_index,
+            proof.chunk_start,
+            proof.chunk_end,
+            proof.parent_start,
+            proof.parent_end,
+        )
+        if any(isinstance(value, bool) or not isinstance(value, int) for value in integer_fields):
+            raise ValueError("network fragment proof offsets must be integers")
+        full_proof = unicodedata.normalize("NFC", str(proof.full_spoken_proof or ""))
+        if (
+            proof.span_index < 0
+            or not 0 <= proof.chunk_start < proof.chunk_end <= len(target)
+            or not 0 <= proof.parent_start < proof.parent_end <= len(full_proof)
+            or proof.chunk_start < previous_chunk_end
+            or target[proof.chunk_start : proof.chunk_end]
+            != full_proof[proof.parent_start : proof.parent_end]
+            or full_proof not in network_protected_spoken_spans(full_proof)
+        ):
+            raise ValueError("network fragment proof does not bind its target range")
+        previous_chunk_end = proof.chunk_end
+        fragment = target[proof.chunk_start : proof.chunk_end]
+        ascii_fragment = _network_fragment_ascii(fragment)
+        if ascii_fragment is None:
+            # Unmapped atoms remain eligible only in their canonical spoken
+            # form. A literal approximation is never inferred.
+            continue
+        old_fragment = typed_fragments.get(ascii_fragment.casefold())
+        if old_fragment is not None and old_fragment != fragment:
+            raise ValueError("network fragment ASCII proof is ambiguous")
+        typed_fragments[ascii_fragment.casefold()] = fragment
+
+    # Protect longer exact atoms first, normalize surrounding transcript
+    # punctuation, then restore their planner-proven spoken forms. A typed atom
+    # is eligible only at a conservative prose boundary. In particular, an
+    # unknown adjacent symbol (backslash, angle bracket, emoji, and so on) must
+    # remain visible rather than letting a global replacement erase its
+    # provenance before the exact range alignment.
+    canonicalized = raw_text
+    placeholder_prefix = "藍鵲片段保護佔位符"
+    while placeholder_prefix in raw_text or placeholder_prefix in target:
+        placeholder_prefix += "號"
+    fragment_placeholders: list[tuple[str, str]] = []
+    unsafe_typed_literal = False
+    ascii_sentence_boundaries = frozenset(",.;:!?\"'()[]{}")
+
+    def literal_boundary_kind(value: str, offset: int) -> str:
+        if offset < 0 or offset >= len(value):
+            return "safe"
+        character = value[offset]
+        if character.isascii() and character.isalnum():
+            return "identifier"
+        if (
+            character.isspace()
+            or _is_cjk(character)
+            or character in _NETWORK_NONASCII_BOUNDARY_PUNCTUATION
+            or character in ascii_sentence_boundaries
+        ):
+            return "safe"
+        return "unsafe"
+
+    for ascii_fragment, spoken_fragment in sorted(
+        typed_fragments.items(),
+        key=lambda item: (-len(item[0]), item[0]),
+    ):
+        character_pattern = r"\s*".join(
+            re.escape(character) for character in ascii_fragment
+        )
+        pattern = re.compile(
+            character_pattern,
+            flags=re.IGNORECASE | re.ASCII,
+        )
+
+        def protect_literal(match: re.Match[str]) -> str:
+            nonlocal unsafe_typed_literal
+            boundary_kinds = (
+                literal_boundary_kind(match.string, match.start() - 1),
+                literal_boundary_kind(match.string, match.end()),
+            )
+            if "identifier" in boundary_kinds:
+                return match.group(0)
+            if "unsafe" in boundary_kinds:
+                unsafe_typed_literal = True
+                return match.group(0)
+            placeholder = (
+                f"{placeholder_prefix}"
+                f"{_zh_integer(len(fragment_placeholders) + 1)}結束"
+            )
+            fragment_placeholders.append((placeholder, spoken_fragment))
+            return placeholder
+
+        canonicalized = pattern.sub(protect_literal, canonicalized)
+
+    canonicalized = normalize_tts_text(canonicalized)
+    for placeholder, spoken_fragment in fragment_placeholders:
+        canonicalized = canonicalized.replace(placeholder, spoken_fragment)
+
+    target_key = _network_fragment_alignment_text(target)
+    transcript_key = _network_fragment_alignment_text(canonicalized)
+    protected_ranges: list[tuple[int, int]] = []
+    rebuilt_target_key: list[str] = []
+    cursor = 0
+    key_cursor = 0
+    for proof in proofs:
+        prefix_key = _network_fragment_alignment_text(
+            target[cursor : proof.chunk_start]
+        )
+        fragment_key = _network_fragment_alignment_text(
+            target[proof.chunk_start : proof.chunk_end]
+        )
+        rebuilt_target_key.extend((prefix_key, fragment_key))
+        key_cursor += len(prefix_key)
+        if not fragment_key:
+            raise ValueError("network fragment proof has no comparable content")
+        protected_ranges.append((key_cursor, key_cursor + len(fragment_key)))
+        key_cursor += len(fragment_key)
+        cursor = proof.chunk_end
+    rebuilt_target_key.append(_network_fragment_alignment_text(target[cursor:]))
+    if "".join(rebuilt_target_key) != target_key:
+        raise ValueError("network fragment comparison ranges are not composable")
+
+    passed = (
+        not unsafe_typed_literal
+        and _protected_ranges_are_exact_in_all_optimal_alignments(
+            target_key,
+            transcript_key,
+            protected_ranges,
+        )
+    )
+    return NetworkFragmentAsrEvidence(
+        transcript_text=canonicalized,
+        protected_range_count=len(protected_ranges),
+        passed=bool(passed),
+    )
+
+
 @dataclass(frozen=True)
 class AsrComparison:
     """Orthographic whole-text and acoustic endpoint evidence for one candidate."""

tests/test_production.py

@@ -6,10 +6,12 @@ import torch
 from torch import nn
 
 from production import (
+    NetworkFragmentProof,
     StopHysteresisController,
     active_pace_correction_speed,
     candidate_local_score,
     candidate_transition_score,
+    canonicalize_asr_network_fragments,
     coalesce_text_chunks,
     compare_asr_text,
     contains_network_identifier,
@@ -40,8 +42,8 @@ from production import (
 )
 
 
-_SPOKEN_HTTPS = "艾取 踢 踢 批 艾斯"
-_SPOKEN_PATH = "批 欸 踢 艾取"
+_SPOKEN_HTTPS = "H T T P S"
+_SPOKEN_PATH = "path"
 
 
 def test_eval_only_pronoun_homophones_do_not_change_model_input_text():
@@ -629,14 +631,14 @@ def test_non_ascii_iri_is_explicitly_unsupported_and_fails_closed(transcript):
     (
         (
             "前方甲乙X，H T T P S 冒號 斜線 斜線 orange 點 "
-            "伊 艾克斯 欸 艾姆 批 艾爾 伊 點 踢 達不溜 斜線 road，"
+            "example 點 T W 斜線 road，"
             "後方丙丁。",
             1.0 / 6.0,
             0.0,
         ),
         (
             "前方甲乙，H T T P S 冒號 斜線 斜線 orange 點 "
-            "伊 艾克斯 欸 艾姆 批 艾爾 伊 點 踢 達不溜 斜線 road，"
+            "example 點 T W 斜線 road，"
             "X後方丙丁。",
             0.0,
             1.0 / 6.0,
@@ -875,20 +877,18 @@ def test_frontend_clause_split_matrix_for_structured_and_network_holdouts():
         "H06": (
             "若要更換導覽場次，請寄信到 tour.help@islandmuseum.tw。",
             (
-                "若要更換導覽場次，請寄信到，",
-                "踢 歐 優 阿爾 點 艾取 伊 艾爾 批 小老鼠 愛 艾斯 "
-                "艾爾 欸 恩 迪 艾姆 優 艾斯 伊 優 艾姆 點 踢 達不溜。",
+                "若要更換導覽場次，請寄信到，tour 點 help 小老鼠 "
+                "islandmuseum 點 T W。",
             ),
-            (12, 37),
+            (24,),
         ),
         "H07": (
             "潮汐預報可查詢 https://coastwatch.example.tw/tide。",
             (
-                "潮汐預報可查詢，艾取 踢 踢 批 艾斯 冒號 斜線 斜線 西 歐 "
-                "欸 艾斯 踢 達不溜 欸 踢 西 艾取 點 伊 艾克斯 欸 艾姆 "
-                "批 艾爾 伊 點 踢 達不溜 斜線 踢 愛 迪 伊。",
+                "潮汐預報可查詢，H T T P S 冒號 斜線 斜線 coastwatch 點 "
+                "example 點 T W 斜線 tide。",
             ),
-            (57,),
+            (30,),
         ),
         "H11": (
             raw_h11,
@@ -897,18 +897,17 @@ def test_frontend_clause_split_matrix_for_structured_and_network_holdouts():
                 "七點十五分 集合，",
                 "先用定位器 A X 五二零 核對座標，再分組檢查木棧道、"
                 "里程牌與飲水站。",
-                "若氣象網站，艾取 踢 踢 批 艾斯 冒號 斜線 斜線 踢 阿爾 "
-                "欸 愛 艾爾 達不溜 伊 欸 踢 艾取 伊 阿爾 點 伊 艾克斯 "
-                "欸 艾姆 批 艾爾 伊 點 踢 達不溜，",
+                "若氣象網站，H T T P S 冒號 斜線 斜線 trailweather 點 "
+                "example 點 T W，",
                 "顯示降雨機率超過 百分之六十五，",
                 "領隊就取消高海拔路線，改走較短的林間環線。",
-                "途中若發現落石或樹枝阻斷通行，請拍照並寄到，",
-                "批 欸 踢 阿爾 歐 艾爾 小老鼠 艾夫 歐 阿爾 伊 艾斯 "
-                "踢 艾姆 欸 愛 艾爾 點 踢 達不溜，不要自行搬動大型障礙物。",
+                "途中若發現落石或樹枝阻斷通行，",
+                "請拍照並寄到，patrol 小老鼠 forestmail 點 T W，"
+                "不要自行搬動大型障礙物。",
                 "所有隊員回到登山口後，還要清點無線電與急救包，"
                 "確認沒有任何人落單，才結束當天的巡查。",
             ),
-            (30, 29, 53, 14, 19, 20, 42, 38),
+            (30, 29, 25, 14, 19, 14, 28, 38),
         ),
     }
 
@@ -944,16 +943,22 @@ def test_generation_only_network_component_planner_holdout_matrix():
     matrix = {
         "H06": (
             "若要更換導覽場次，請寄信到 tour.help@islandmuseum.tw。",
-            (24, 25),
+            (15, 9),
+            (0,),
         ),
         "H07": (
             "潮汐預報可查詢 https://coastwatch.example.tw/tide。",
-            (34, 23),
+            (18, 12),
+            (0,),
+        ),
+        "H11": (
+            raw_h11,
+            (30, 29, 16, 23, 19, 22, 20, 38),
+            (2, 5),
         ),
-        "H11": (raw_h11, (30, 29, 18, 35, 33, 28, 34, 38)),
     }
 
-    for raw, expected_units in matrix.values():
+    for raw, expected_units, mandatory_after in matrix.values():
         normalized = normalize_spoken_forms(raw)
         specs = plan_generation_chunks(raw, normalized)
 
@@ -964,14 +969,384 @@ def test_generation_only_network_component_planner_holdout_matrix():
             for spec, following in zip(specs, specs[1:])
         )
         assert all(
-            12 <= count_speech_units(spec.text) <= (36 if spec.network_conditioned else 80)
+            (8 if spec.network_conditioned else 12)
+            <= count_speech_units(spec.text)
+            <= (36 if spec.network_conditioned else 80)
             for spec in specs
         )
-        assert any(spec.boundary_after == "network_internal" for spec in specs)
+        assert tuple(
+            index
+            for index, spec in enumerate(specs)
+            if spec.boundary_after == "network_internal"
+        ) == mandatory_after
         for spec in specs:
             assert len(spec.network_span_indices) == len(
                 spec.network_full_spoken_proofs
             )
+            assert len(spec.network_span_indices) == len(
+                spec.network_fragment_proofs
+            )
+            for proof in spec.network_fragment_proofs:
+                assert (
+                    spec.text[proof.chunk_start : proof.chunk_end]
+                    == proof.full_spoken_proof[
+                        proof.parent_start : proof.parent_end
+                    ]
+                )
+
+        for span_index in {
+            index for spec in specs for index in spec.network_span_indices
+        }:
+            ranges = sorted(
+                (proof.parent_start, proof.parent_end, proof.full_spoken_proof)
+                for spec in specs
+                for proof in spec.network_fragment_proofs
+                if proof.span_index == span_index
+            )
+            assert ranges[0][0] == 0
+            assert ranges[-1][1] == len(ranges[0][2])
+            assert all(
+                left_end == right_start
+                for (_, left_end, _), (right_start, _, _) in zip(
+                    ranges,
+                    ranges[1:],
+                )
+            )
+
+
+def _tour_help_fragment_case(prefix: str = "請核對，", suffix: str = "，完成。"):
+    fragment = "tour 點 help"
+    full_proof = normalize_spoken_forms("tour.help@islandmuseum.tw")
+    parent_start = full_proof.index(fragment)
+    target = f"{prefix}{fragment}{suffix}"
+    chunk_start = len(prefix)
+    return target, NetworkFragmentProof(
+        span_index=0,
+        chunk_start=chunk_start,
+        chunk_end=chunk_start + len(fragment),
+        parent_start=parent_start,
+        parent_end=parent_start + len(fragment),
+        full_spoken_proof=full_proof,
+    )
+
+
+@pytest.mark.parametrize("literal", ("tour.help", "Tour.Help", "t o u r . h e l p"))
+def test_local_network_fragment_canonicalizes_only_exact_typed_ascii(literal):
+    target, proof = _tour_help_fragment_case()
+    transcript = f"請核對，{literal}，完成。"
+
+    evidence = canonicalize_asr_network_fragments(transcript, target, (proof,))
+
+    assert evidence.passed is True
+    assert evidence.protected_range_count == 1
+    assert "tour 點 help" in evidence.transcript_text
+    assert compare_asr_text(target, evidence.transcript_text, max_cer=0.20).passed
+
+
+def test_local_network_fragment_normalizes_ascii_sentence_boundaries_first():
+    target, proof = _tour_help_fragment_case()
+
+    evidence = canonicalize_asr_network_fragments(
+        "請核對,tour.help,完成.",
+        target,
+        (proof,),
+    )
+
+    assert evidence.transcript_text == target
+    assert evidence.passed is True
+
+
+@pytest.mark.parametrize(
+    "literal",
+    (
+        "tour.herp",       # wrong content
+        "tourhelp",        # missing separator
+        "tour.hel",        # missing suffix letter
+        "help.tour",       # reordered labels
+        "xtour.help",      # extra prefix content
+        "tour.help/path",  # extra suffix content
+        "tour.help.help",  # repeated content
+    ),
+)
+def test_local_network_fragment_rejects_wrong_missing_reordered_or_extra_content(
+    literal,
+):
+    target, proof = _tour_help_fragment_case()
+    transcript = f"請核對，{literal}，完成。"
+
+    evidence = canonicalize_asr_network_fragments(transcript, target, (proof,))
+
+    assert evidence.passed is False
+    assert evidence.protected_range_count == 1
+
+
+@pytest.mark.parametrize("symbol", ("\\", "<", ">", "§", "🙂"))
+@pytest.mark.parametrize("side", ("prefix", "suffix"))
+def test_local_network_fragment_rejects_adjacent_unknown_symbols(symbol, side):
+    target, proof = _tour_help_fragment_case()
+    literal = f"{symbol}tour.help" if side == "prefix" else f"tour.help{symbol}"
+
+    evidence = canonicalize_asr_network_fragments(
+        f"請核對，{literal}，完成。",
+        target,
+        (proof,),
+    )
+
+    assert evidence.protected_range_count == 1
+    assert evidence.passed is False
+
+
+def _pure_label_fragment_case(label: str):
+    fragment = f"{label} "
+    full_proof = normalize_spoken_forms(f"{label}@islandmuseum.tw")
+    parent_start = full_proof.index(fragment)
+    prefix = "請核對，"
+    target = f"{prefix}{fragment}，完成。"
+    chunk_start = len(prefix)
+    return target, NetworkFragmentProof(
+        span_index=0,
+        chunk_start=chunk_start,
+        chunk_end=chunk_start + len(fragment),
+        parent_start=parent_start,
+        parent_end=parent_start + len(fragment),
+        full_spoken_proof=full_proof,
+    )
+
+
+@pytest.mark.parametrize("label", ("tour", "a"))
+@pytest.mark.parametrize("side", ("prefix", "suffix"))
+@pytest.mark.parametrize(
+    "symbol",
+    (
+        "/",
+        "\\",
+        "@",
+        "-",
+        "_",
+        "%",
+        "$",
+        "~",
+        "|",
+        "=",
+        "&",
+        "#",
+        "+",
+        "*",
+        "<",
+        ">",
+        "§",
+        "🙂",
+    ),
+)
+def test_pure_label_network_fragment_rejects_adjacent_identifier_symbols(
+    label,
+    side,
+    symbol,
+):
+    target, proof = _pure_label_fragment_case(label)
+    literal = f"{symbol}{label}" if side == "prefix" else f"{label}{symbol}"
+
+    evidence = canonicalize_asr_network_fragments(
+        f"請核對，{literal}，完成。",
+        target,
+        (proof,),
+    )
+
+    assert evidence.protected_range_count == 1
+    assert evidence.passed is False
+
+
+@pytest.mark.parametrize("label", ("tour", "a"))
+def test_pure_label_network_fragment_accepts_ascii_sentence_punctuation(label):
+    target, proof = _pure_label_fragment_case(label)
+
+    evidence = canonicalize_asr_network_fragments(
+        f"請核對,{label},完成.",
+        target,
+        (proof,),
+    )
+
+    assert evidence.transcript_text == target
+    assert evidence.passed is True
+
+
+def test_local_network_fragment_exact_gate_is_stricter_than_whole_chunk_cer():
+    prefix = "甲" * 30 + "，"
+    suffix = "，" + "乙" * 30 + "。"
+    target, proof = _tour_help_fragment_case(prefix, suffix)
+    wrong_spoken = target.replace("help", "herp", 1)
+
+    # One protected middle error is below the ordinary whole-chunk CER limit
+    # and outside the prefix/suffix windows, but the range gate still rejects.
+    assert compare_asr_text(target, wrong_spoken, max_cer=0.20).passed
+    evidence = canonicalize_asr_network_fragments(
+        wrong_spoken,
+        target,
+        (proof,),
+    )
+    assert evidence.passed is False
+
+
+def test_local_network_fragment_provenance_is_range_bound_and_fails_closed():
+    target, proof = _tour_help_fragment_case()
+    forged = NetworkFragmentProof(
+        span_index=proof.span_index,
+        chunk_start=proof.chunk_start + 1,
+        chunk_end=proof.chunk_end,
+        parent_start=proof.parent_start,
+        parent_end=proof.parent_end - 1,
+        full_spoken_proof=proof.full_spoken_proof,
+    )
+
+    with pytest.raises(ValueError, match="does not bind"):
+        canonicalize_asr_network_fragments("tour.help", target, (forged,))
+    with pytest.raises(ValueError, match="invalid type"):
+        canonicalize_asr_network_fragments("tour.help", target, (object(),))
+
+
+def test_local_network_fragment_cannot_borrow_repeated_plain_spoken_text():
+    fragment = "tour 點 help"
+    prefix = f"先照字面念{fragment}，再核對，"
+    target, proof = _tour_help_fragment_case(prefix, "。")
+    transcript = "先照字面念 tour.help，再核對，tour.herp。"
+
+    evidence = canonicalize_asr_network_fragments(transcript, target, (proof,))
+
+    assert evidence.passed is False
+
+
+def _hybrid_tour_help_fragment_case(prefix: str = "請核對，"):
+    fragment = "tour 點 help"
+    full_proof = normalize_spoken_forms("tour.help@islandmuseum.tw")
+    parent_start = full_proof.index(fragment)
+    target = f"{prefix}{fragment}，完成。"
+    chunk_start = len(prefix)
+    return target, NetworkFragmentProof(
+        span_index=0,
+        chunk_start=chunk_start,
+        chunk_end=chunk_start + len(fragment),
+        parent_start=parent_start,
+        parent_end=parent_start + len(fragment),
+        full_spoken_proof=full_proof,
+    )
+
+
+def test_local_network_fragment_accepts_exact_range_bound_hybrid_lexical_atoms():
+    target, proof = _hybrid_tour_help_fragment_case()
+
+    evidence = canonicalize_asr_network_fragments(
+        "請核對，Tour.Help，完成。",
+        target,
+        (proof,),
+    )
+
+    assert evidence.passed is True
+    assert evidence.transcript_text == target
+
+
+@pytest.mark.parametrize(
+    "literal",
+    (
+        "tours.help",
+        "tour.helps",
+        "our.help",
+        "tour.hepl",
+        "xtour.help",
+        "tour.help/x",
+    ),
+)
+def test_hybrid_lexical_network_atoms_reject_substrings_and_mutations(literal):
+    target, proof = _hybrid_tour_help_fragment_case()
+
+    evidence = canonicalize_asr_network_fragments(
+        f"請核對，{literal}，完成。",
+        target,
+        (proof,),
+    )
+
+    assert evidence.passed is False
+
+
+def _duplicate_network_fragment_case(*, protected_first: bool):
+    fragment = "點 欸"
+    full_proof = "H T T P S 冒號 斜線 斜線 tour 點 欸"
+    prefix = "甲" * 20
+    suffix = "乙" * 20 + "。"
+    target = f"{prefix}{fragment}，{fragment}{suffix}"
+    if protected_first:
+        chunk_start = len(prefix)
+    else:
+        chunk_start = len(prefix) + len(fragment) + 1
+    parent_start = full_proof.index(fragment)
+    return target, prefix, suffix, NetworkFragmentProof(
+        span_index=0,
+        chunk_start=chunk_start,
+        chunk_end=chunk_start + len(fragment),
+        parent_start=parent_start,
+        parent_end=parent_start + len(fragment),
+        full_spoken_proof=full_proof,
+    )
+
+
+@pytest.mark.parametrize("protected_first", (False, True))
+def test_network_fragment_duplicate_cannot_borrow_any_optimal_alignment(
+    protected_first,
+):
+    target, prefix, suffix, proof = _duplicate_network_fragment_case(
+        protected_first=protected_first
+    )
+
+    # The lone literal can align equally well to either duplicate. The local
+    # proof must fail because at least one optimal path deletes the protected
+    # occurrence, regardless of deterministic backtrace tie-breaking.
+    evidence = canonicalize_asr_network_fragments(
+        f"{prefix}.a{suffix}",
+        target,
+        (proof,),
+    )
+
+    assert compare_asr_text(
+        target,
+        evidence.transcript_text,
+        max_cer=0.20,
+    ).passed
+    assert evidence.passed is False
+
+
+@pytest.mark.parametrize("protected_first", (False, True))
+def test_network_fragment_duplicate_exact_two_occurrences_remain_valid(
+    protected_first,
+):
+    target, prefix, suffix, proof = _duplicate_network_fragment_case(
+        protected_first=protected_first
+    )
+
+    evidence = canonicalize_asr_network_fragments(
+        f"{prefix}.a，.a{suffix}",
+        target,
+        (proof,),
+    )
+
+    assert evidence.transcript_text == target
+    assert evidence.passed is True
+
+
+@pytest.mark.parametrize("boundary", ("start", "end"))
+def test_network_fragment_rejects_insertions_on_closed_range_boundary(boundary):
+    prefix = "甲" * 30 + "，"
+    suffix = "，" + "乙" * 30 + "。"
+    target, proof = _tour_help_fragment_case(prefix, suffix)
+    offset = proof.chunk_start if boundary == "start" else proof.chunk_end
+    transcript = target[:offset] + "錯" + target[offset:]
+
+    evidence = canonicalize_asr_network_fragments(
+        transcript,
+        target,
+        (proof,),
+    )
+
+    assert compare_asr_text(target, transcript, max_cer=0.20).passed
+    assert evidence.passed is False
 
 
 def test_generation_network_planner_keeps_ordinary_semantic_chunks_unchanged():
@@ -1007,25 +1382,124 @@ def test_generation_network_planner_fails_closed_on_invalid_provenance():
     with pytest.raises(ValueError, match="non-ASCII IRI"):
         plan_generation_chunks(emoji_iri, normalize_spoken_forms(emoji_iri))
 
-    oversized = "請查詢 https://" + "a" * 50 + ".tw。"
+    oversized = "請查詢 https://" + "a" * 150 + ".tw。"
     with pytest.raises(ValueError, match="indivisible network component"):
         plan_generation_chunks(oversized, normalize_spoken_forms(oversized))
 
+    with pytest.raises(ValueError, match="chunk limits are inconsistent"):
+        plan_generation_chunks(
+            "請查 https://example.tw。",
+            normalize_spoken_forms("請查 https://example.tw。"),
+            min_units=12,
+            network_min_units=13,
+        )
 
-def test_generation_network_planner_binds_duplicate_spoken_proof_to_raw_identifier():
-    spoken = (
-        "艾取 踢 踢 批 艾斯 冒號 斜線 斜線 "
-        "伊 艾克斯 欸 艾姆 批 艾爾 伊 點 踢 達不溜"
+
+@pytest.mark.parametrize("raw", ("a@b.co", "https://a.tw"))
+def test_short_network_identifier_relaxes_only_infeasible_preferred_cut(raw):
+    normalized = normalize_spoken_forms(raw)
+
+    specs = plan_generation_chunks(raw, normalized)
+
+    assert len(specs) == 1
+    spec = specs[0]
+    assert spec.text == normalized
+    assert spec.source_start == 0
+    assert spec.source_end == len(normalized)
+    assert spec.network_conditioned
+    assert spec.boundary_after == "none"
+    assert spec.network_span_indices == (0,)
+    assert spec.network_full_spoken_proofs == (normalized,)
+    assert len(spec.network_fragment_proofs) == 1
+    proof = spec.network_fragment_proofs[0]
+    assert (proof.chunk_start, proof.chunk_end) == (0, len(normalized))
+    assert (proof.parent_start, proof.parent_end) == (0, len(normalized))
+    assert proof.full_spoken_proof == normalized
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected_texts", "expected_units"),
+    (
+        (
+            "若要更換導覽場次，請寄信到 tour.help@islandmuseum.tw。",
+            (
+                "若要更換導覽場次，請寄信到，tour 點 help ",
+                "小老鼠 islandmuseum 點 T W。",
+            ),
+            (15, 9),
+        ),
+        (
+            "潮汐預報可查詢 https://coastwatch.example.tw/tide。",
+            (
+                "潮汐預報可查詢，H T T P S 冒號 斜線 斜線",
+                " coastwatch 點 example 點 T W 斜線 tide。",
+            ),
+            (18, 12),
+        ),
+    ),
+)
+def test_generation_planner_cannot_cross_mandatory_network_grammar_cuts(
+    raw,
+    expected_texts,
+    expected_units,
+):
+    normalized = normalize_spoken_forms(raw)
+    specs = plan_generation_chunks(
+        raw,
+        normalized,
+        target_units=80,
+        network_max_units=80,
     )
+
+    assert tuple(spec.text for spec in specs) == expected_texts
+    assert tuple(count_speech_units(spec.text) for spec in specs) == expected_units
+    assert specs[0].boundary_after == "network_internal"
+    assert specs[1].boundary_after == "none"
+    mandatory_cut = specs[0].source_end
+    assert all(
+        not (spec.source_start < mandatory_cut < spec.source_end)
+        for spec in specs
+    )
+    assert specs[0].network_fragment_proofs[0].parent_end == (
+        specs[1].network_fragment_proofs[0].parent_start
+    )
+
+
+def test_hybrid_network_frontend_keeps_labels_lexical_and_typed_atoms_explicit():
+    assert normalize_spoken_forms(
+        "若要更換導覽場次，請寄信到 tour.help@islandmuseum.tw。"
+    ) == (
+        "若要更換導覽場次，請寄信到，tour 點 help 小老鼠 "
+        "islandmuseum 點 T W。"
+    )
+    assert normalize_spoken_forms(
+        "潮汐預報可查詢 https://coastwatch.example.tw/tide。"
+    ) == (
+        "潮汐預報可查詢，H T T P S 冒號 斜線 斜線 coastwatch 點 "
+        "example 點 T W 斜線 tide。"
+    )
+    assert normalize_spoken_forms(
+        "HTTPS://WWW.Example.COM/API/v1?ID=RTX5090"
+    ) == (
+        "H T T P S 冒號 斜線 斜線 W W W 點 Example 點 C O M 斜線 "
+        "A P I 斜線 v 一 問號 I D 等於 R T X 五零九零"
+    )
+
+
+def test_generation_network_planner_binds_duplicate_spoken_proof_to_raw_identifier():
+    spoken = "H T T P S 冒號 斜線 斜線 example 點 T W"
     raw = f"先念 {spoken}，再查 https://example.tw。"
     normalized = normalize_spoken_forms(raw)
 
     specs = plan_generation_chunks(raw, normalized)
 
     assert "".join(spec.text for spec in specs) == normalized
-    assert [spec.network_conditioned for spec in specs] == [False, True]
-    assert spoken in specs[0].text
-    assert specs[1].network_full_spoken_proofs == (spoken,)
+    assert [spec.network_conditioned for spec in specs] == [True]
+    assert specs[0].text.count(spoken) == 2
+    assert specs[0].network_full_spoken_proofs == (spoken,)
+    proof = specs[0].network_fragment_proofs[0]
+    assert proof.chunk_start == specs[0].text.rindex(spoken)
+    assert specs[0].text[proof.chunk_start : proof.chunk_end] == spoken
     assert all(
         normalized[spec.source_start : spec.source_end] == spec.text
         for spec in specs
@@ -1277,10 +1751,10 @@ def test_ascii_text_receives_a_wider_generation_window():
     assert select_generation_cps("AI TTS 測試", cjk_cps=5.2, ascii_cps=4.6) == 4.6
 
 
-def test_explicit_letter_network_text_keeps_the_ascii_generation_window():
+def test_hybrid_network_text_keeps_the_ascii_generation_window():
     normalized = normalize_spoken_forms("https://example.tw/path")
 
-    assert not any(character.isascii() and character.isalnum() for character in normalized)
+    assert any(character.isascii() and character.isalnum() for character in normalized)
     assert network_protected_spoken_spans(normalized)
     assert select_generation_cps(normalized, cjk_cps=5.2, ascii_cps=4.6) == 4.6
 
@@ -1426,9 +1900,8 @@ def test_spoken_form_normalizer_expands_clock_time_and_spells_urls():
         "無效 24點30分、15點60分。"
     )
     assert normalize_spoken_forms("IP 192.168.1.1，網址 https://example.test:30/path") == (
-        "I P 192.168.1.1，網址，艾取 踢 踢 批 艾斯 冒號 斜線 斜線 "
-        "伊 艾克斯 欸 艾姆 批 艾爾 伊 點 踢 伊 艾斯 踢 "
-        "冒號 三零 斜線 批 欸 踢 艾取"
+        "I P 192.168.1.1，網址，H T T P S 冒號 斜線 斜線 "
+        "example 點 test 冒號 三零 斜線 path"
     )
     assert normalize_spoken_forms(
         "音量15點05分貝，區間15點30分鐘，比分15點05分。"
@@ -1864,14 +2337,12 @@ def test_spoken_form_normalizer_expands_network_identifiers_and_versions():
     assert normalize_spoken_forms(
         "網址 https://api.example.com/v1/items?q=RTX-5090&n=2。"
     ) == (
-        "網址，艾取 踢 踢 批 艾斯 冒號 斜線 斜線 欸 批 愛 點 "
-        "伊 艾克斯 欸 艾姆 批 艾爾 伊 點 西 歐 艾姆 斜線 維 一 "
-        "斜線 愛 踢 伊 艾姆 艾斯 問號 丘 等於 阿爾 踢 艾克斯 "
-        "橫線 五零九零 和 恩 等於 二。"
+        "網址，H T T P S 冒號 斜線 斜線 api 點 example 點 C O M "
+        "斜線 v 一 斜線 items 問號 q 等於 R T X 橫線 五零九零 "
+        "和 n 等於 二。"
     )
     assert normalize_spoken_forms("信箱 USER.name+tts@example.com。") == (
-        "信箱，優 艾斯 伊 阿爾 點 恩 欸 艾姆 伊 加號 踢 踢 艾斯 "
-        "小老鼠 伊 艾克斯 欸 艾姆 批 艾爾 伊 點 西 歐 艾姆。"
+        "信箱，U S E R 點 name 加號 tts 小老鼠 example 點 C O M。"
     )
     assert normalize_spoken_forms(
         "版本 v1.2.3，候選 version 10.4.0-beta.1+build.5。"
@@ -1916,7 +2387,7 @@ def test_expanded_url_rejects_raw_symbol_inserted_inside_path(symbol):
     target = normalize_spoken_forms("https://example.tw/path")
     transcript = target.replace(
         _SPOKEN_PATH,
-        f"批 欸{symbol}踢 艾取",
+        f"pa{symbol}th",
     )
     comparison = compare_asr_text(
         target,
@@ -1935,8 +2406,8 @@ def test_expanded_url_rejects_raw_symbol_inserted_inside_path(symbol):
 def test_expanded_email_rejects_raw_symbol_inserted_inside_local_part(symbol):
     target = normalize_spoken_forms("museum@example.tw")
     transcript = target.replace(
-        "艾姆 優 艾斯 伊 優 艾姆",
-        f"艾姆 優{symbol}艾斯 伊 優 艾姆",
+        "museum",
+        f"mu{symbol}seum",
     )
     comparison = compare_asr_text(
         target,
@@ -2093,7 +2564,7 @@ def test_expanded_www_url_keeps_an_exact_protected_span():
     )
     comparison = compare_asr_text(
         target,
-        target.replace(_SPOKEN_PATH, "批 欸 踢"),
+        target.replace(_SPOKEN_PATH, "pat"),
         max_cer=0.20,
         max_prefix_cer=1.0,
         max_suffix_cer=1.0,

tests/test_quality_runtime.py

@@ -811,8 +811,8 @@ def test_quality_gate_rejects_inexact_network_span_below_whole_cer_limit():
         "請先閱讀 https://museum.example.tw/path，確認展覽時間與集合位置後再回覆。"
     )
     transcript = target.replace(
-        "艾姆 優 艾斯 伊 優 艾姆",
-        "艾姆 優 艾斯 伊 優 艾姆 欸 恩",
+        "museum",
+        "museuman",
     )
     result = verify_candidate(
         CandidateObservation(

tests/test_release_pins.py

@@ -229,6 +229,11 @@ def test_readme_describes_coverage_refill_and_sequence_transition_scores():
     assert "median-F0 軟成本" in readme
     assert "5.2 CJK / 4.6 ASCII" in readme
     assert "4.6 CJK / 4.0 ASCII" in readme
+    assert "URL/email-bearing chunks 8 units" in readme
+    assert "Email 第一輪必須在 `小老鼠`" in readme
+    assert "第一輪 DP 不得跨越這兩類" in readme
+    assert "grammar boundary" in readme
+    assert "所有最佳 edit alignment" in readme
     assert "boundary-only local rejects" in readme
     assert "drop 不超過 0.15" in readme
     assert "整段仍必須通過 similarity 0.105 與 boundary drop 0.095" in readme
@@ -352,6 +357,7 @@ def test_app_and_quality_runtime_pin_the_same_mixed_cfg_contract():
 
 def test_app_rejects_silent_text_and_coalesces_before_runtime_budgeting():
     source = (ROOT / "app.py").read_text(encoding="utf-8")
+    production_source = (ROOT / "production.py").read_text(encoding="utf-8")
     tree = ast.parse(source)
     functions = {
         node.name: node
@@ -372,8 +378,10 @@ def test_app_rejects_silent_text_and_coalesces_before_runtime_budgeting():
     assert "chunks = tuple(spec.text for spec in chunk_specs)" in synthesize_source
     assert "max_chunks=QUALITY_MAX_GENERATED_CHUNKS" in synthesize_source
     assert "pre_faded_edges=True" in assemble_source
+    assert "NETWORK_GENERATION_MIN_UNITS = 8" in source
     assert "NETWORK_GENERATION_TARGET_UNITS = 32" in source
     assert "NETWORK_GENERATION_MAX_UNITS = 36" in source
+    assert "network_min_units=NETWORK_GENERATION_MIN_UNITS" in synthesize_source
     assert "NETWORK_INTERNAL_FADE_MS = 5.0" in source
     assert 'chunk_specs[index].boundary_after == "network_internal"' in (
         assemble_source
@@ -381,6 +389,16 @@ def test_app_rejects_silent_text_and_coalesces_before_runtime_budgeting():
     assert "join_audio_chunks_variable(" in assemble_source
     assert "network_conditioned=network_flags" in synthesize_source
     assert "network_conditioned=network_flag" in synthesize_source
+    assert "mandatory_cut_offsets" in production_source
+    assert "any(start < cut < end for cut in active_mandatory_cuts)" in (
+        production_source
+    )
+    assert "plan = solve(mandatory_cuts - short_identifier_cuts)" in (
+        production_source
+    )
+    assert "_protected_ranges_are_exact_in_all_optimal_alignments(" in (
+        production_source
+    )
 
 
 def test_app_emits_one_canonical_content_free_evidence_line_per_terminal_outcome():
@@ -679,6 +697,69 @@ def test_space_hard_intersects_dual_asr_only_on_exact_whole_waveforms():
     assert "except (RuntimeError, ValueError) as error:" in synthesize_source
 
 
+def test_network_fragment_relaxation_is_range_bound_and_local_only():
+    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"])
+    proof_source = ast.get_source_segment(
+        source,
+        functions["_network_fragment_proof_rows"],
+    )
+    qualify_source = ast.get_source_segment(
+        source,
+        functions["_qualify_candidate_trajectory_audio"],
+    )
+    refill_source = ast.get_source_segment(
+        source,
+        functions["_verify_refill_candidate_trajectory_audio"],
+    )
+    independent_source = ast.get_source_segment(
+        source,
+        functions["_verify_independent_whole_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,
+            proof_source,
+            qualify_source,
+            refill_source,
+            independent_source,
+            sequence_source,
+            synthesize_source,
+        )
+    )
+    assert "canonicalize_asr_network_fragments(" in verify_source
+    assert "if fragment_evidence.passed" in verify_source
+    assert "network-conditioned chunk lacks exact fragment proof" in proof_source
+    assert "proof.span_index for proof in proofs" in proof_source
+    assert "proof.full_spoken_proof for proof in proofs" in proof_source
+
+    local_index = qualify_source.index("local_verification = _verify_trajectory_audio(")
+    joined_index = qualify_source.index("joined_verification = _verify_trajectory_audio(")
+    assert "network_fragment_proofs=" in qualify_source[local_index:joined_index]
+    assert "network_fragment_proofs=" not in qualify_source[joined_index:]
+    assert "network_fragment_proofs=" in refill_source
+    assert "network_fragment_proofs=" not in independent_source
+    assert "network_fragment_proofs=" not in sequence_source
+
+    final_index = synthesize_source.index("final_verification = _verify_trajectory_audio(")
+    assert "network_fragment_proofs=" not in synthesize_source[final_index:]
+    assert "generation_context_by_seed" in synthesize_source
+    assert "generation_chunk_specs(seed, candidate_chunks)" 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")