Harden network generation and row-local recovery

README.md

@@ -45,7 +45,7 @@ Barbet 另固定在 `6fcd7ce4aa37f2250a3242995bef0fbc3b026ba8`，
 
 | 設定 | 值 |
 |---|---:|
-| CFG | fixed candidate schedule: offset 0/even = 3.0, positive odd = 2.0; UI primary CFG is locked |
+| CFG | fixed per-chunk schedule: row ordinal 0/even = 3.0, positive odd = 2.0; UI primary CFG is locked |
 | NFE steps | fixed at the validated value 10 |
 | Target pace | 4.0 speech units/sec |
 | Initial whole-trajectory policy | base: 5.2 CJK / 4.6 ASCII-mixed units/sec + 1 latent step |
@@ -69,9 +69,9 @@ Barbet 另固定在 `6fcd7ce4aa37f2250a3242995bef0fbc3b026ba8`，
 | Sequence fallback | ragged DP with up to 3 culprit-diverse paths; boundary-only local rejects may enter through a frozen 0.15 cap, then every exact assembly must pass turbo + full large-v3 and the stricter 0.105/0.095 whole gate |
 | 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 | 80 speech units |
+| 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 |
-| Crossfade / internal edge fade | 80 ms / 80 ms |
+| 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 |
 | Generated-audio retry | disabled |
@@ -80,18 +80,19 @@ Barbet 另固定在 `6fcd7ce4aa37f2250a3242995bef0fbc3b026ba8`，
 每個請求會取得新的隨機 root seed。服務先且只先生成一次完整 trajectory；其中所有長文切段
 共用 root seed，並使用 base policy（5.2 CJK / 4.6 ASCII）。如果這條 exact whole path 未通過，
 服務會保留每一段的 local evidence，再依 `(coverage, refill attempts, chunk index)` 穩定順序，
-只對 zero/low-coverage chunk 逐一生成 refill。大多數 offset 使用 safe-duration（4.6 CJK / 4.0 ASCII），
+只對 zero/low-coverage chunk 逐一生成 refill。每個 chunk 依自己的 refill ordinal 輪替 policy；大多數 refill 使用 safe-duration（4.6 CJK / 4.0 ASCII），
 每第四個 retry 使用 completion-headroom（4.2 CJK / 3.6 ASCII）。Refill 使用
 `root seed + global generation offset`，不會重新生成其他已覆蓋 chunks。三種 policy 都只改 native-duration endpoint estimate，
 margin 固定 +1 latent step、`min_len` 固定為 2，不會為了播放目標語速硬撐 generation loop。
 選中完整 trajectory 或 coverage-sequence-DP 混合路徑後，log 會記錄每個 chunk 的 candidate、seed、policy、實際 CFG，
-並記錄每個已嘗試 offset 的 schedule CFG。Offset 0 與偶數候選使用 primary CFG 3.0，
-正奇數候選使用 alternate CFG 2.0；含 URL/email 的 chunk 仍有 CFG 3.0 的內容安全下限。
+並記錄每個已嘗試 row-local ordinal 的 schedule CFG。Ordinal 0 與正偶數使用 primary CFG 3.0，
+正奇數使用 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 也能重建嘗試軌跡。
-終端 outcome 另輸出 content-free canonical evidence schema v2，包含 original chunk indices、
-CFG contract、20/800 預算用量與 selected-path 交叉檢查，不包含 target/transcript/audio/embedding。
+終端 outcome 另輸出 content-free canonical evidence schema v3，包含 original chunk indices、
+row-local ordinal、network provenance、CFG contract、20/800 預算用量與 selected-path 交叉檢查，
+不包含 target/transcript/audio/embedding。
 為了讓 release contract 與正式量測一致，UI 不提供其他 primary CFG。
 內部 hosted evaluator 可注入 `[0, 2^31)` 的固定 root seed 以重現結果；UI 不暴露這個參數，
 一般請求仍只在未注入 seed 時使用系統亂數。
@@ -139,6 +140,10 @@ zh-TW spoken form，例如 `2026/07/16`、`15:30`、`12.5%` 與 `10 km`。ASR
 Email 與 URL 會以可辨識的語義讀法展開：scheme、local/domain/path 的 opaque ASCII
 labels 逐字使用台灣華語字母名，數字逐位朗讀，分隔符明確朗讀（例如 `.tw`
 讀成「點、踢、達不溜」）。
+完整 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。
 這能降低���型把不常見 TLD 自動補成 `.com` 的風險；一般英文句子不會套用這個規則。
 URL 與後續英文 prose 應以空白或中文標點分隔；未分隔的 RFC path punctuation 會視為 URL
 本身的一部分並納入 exact gate。Quoted email local-part 暫不支援，輸入時會直接 fail closed。

app.py

@@ -18,6 +18,7 @@ from transformers import PreTrainedTokenizerFast
 from bluemagpie import BlueMagpieModel
 from production import (
     StopHysteresisController,
+    GenerationChunkSpec,
     active_pace_correction_speed,
     apply_loudness_floor,
     coalesce_text_chunks,
@@ -27,14 +28,15 @@ from production import (
     endpoint_generation_plan,
     estimate_step_seconds,
     extract_windowed_speaker_embedding,
-    fade_internal_edges,
+    fade_variable_internal_edges,
     finish_audio,
-    join_audio_chunks,
+    join_audio_chunks_variable,
     match_chunk_rms,
     network_identifier_has_ambiguous_iri,
     network_protected_spoken_spans,
     normalize_spoken_forms,
     punctuation_pause_seconds,
+    plan_generation_chunks,
     select_generation_cps,
     set_generation_seed,
     split_text_for_tts,
@@ -47,6 +49,7 @@ from quality_runtime import (
     VERIFICATION_WHISPER_REVISION,
     WHISPER_MODEL_ID,
     WHISPER_REVISION,
+    CandidateGenerationContext,
     CandidateGenerationEvidence,
     CandidateVerification,
     CandidateObservation,
@@ -101,7 +104,7 @@ TARGET_CPS = 4.0
 ACTIVE_PACE_TARGET_CPS = 4.00
 MIXED_CFG_PRIMARY = 3.0
 MIXED_CFG_ALTERNATE = 2.0
-MIXED_CFG_SCHEDULE = "offset_zero_and_even_primary_odd_alternate"
+MIXED_CFG_SCHEDULE = "row_ordinal_zero_and_even_primary_odd_alternate"
 MIN_ENDPOINT_CUE_UNITS = 6
 SHORT_TEXT_CFG_MIN = 3.0
 SHORT_TEXT_CFG_UNITS = 6
@@ -112,6 +115,9 @@ MIN_CHUNK_CHARS = 12
 CROSSFADE_MS = 80.0
 CHUNK_EDGE_FADE_MS = 80.0
 CHUNK_RMS_MATCH_DB = 4.0
+NETWORK_GENERATION_TARGET_UNITS = 32
+NETWORK_GENERATION_MAX_UNITS = 36
+NETWORK_INTERNAL_FADE_MS = 5.0
 STOP_THRESHOLD = 0.50
 STOP_LATE_THRESHOLD = 0.05
 STOP_LATE_START_RATIO = 0.75
@@ -259,11 +265,16 @@ def _generate_chunk(
     steps: int,
     request_seed: int,
     policy: GenerationPolicy,
+    network_conditioned: bool = False,
 ) -> np.ndarray:
-    generation_cps = select_generation_cps(
-        text,
-        cjk_cps=policy.cjk_cps,
-        ascii_cps=policy.ascii_cps,
+    generation_cps = (
+        policy.ascii_cps
+        if network_conditioned
+        else select_generation_cps(
+            text,
+            cjk_cps=policy.cjk_cps,
+            ascii_cps=policy.ascii_cps,
+        )
     )
     model_text, expected_steps, hard_stop_steps = endpoint_generation_plan(
         text,
@@ -372,11 +383,18 @@ def _generate_trajectory(
     request_seed: int,
     policy: GenerationPolicy,
     network_cfg_min: float = NETWORK_TEXT_CFG_MIN,
+    network_conditioned: tuple[bool, ...] | None = None,
 ) -> tuple[np.ndarray, ...]:
     scheduled_cfg = float(cfg)
     trajectory: list[np.ndarray] = []
+    if network_conditioned is not None and len(network_conditioned) != len(chunks):
+        raise ValueError("network provenance must align with generation chunks")
     for local_chunk_index, chunk in enumerate(chunks):
-        network_chunk = bool(network_protected_spoken_spans(chunk))
+        network_chunk = (
+            bool(network_conditioned[local_chunk_index])
+            if network_conditioned is not None
+            else bool(network_protected_spoken_spans(chunk))
+        )
         network_floor_applied = bool(
             network_chunk and scheduled_cfg < float(network_cfg_min)
         )
@@ -408,6 +426,7 @@ def _generate_trajectory(
                 steps=steps,
                 request_seed=request_seed,
                 policy=policy,
+                network_conditioned=network_chunk,
             )
         )
     return tuple(trajectory)
@@ -596,13 +615,49 @@ def _assemble_trajectory_audio(
     trajectory: tuple[np.ndarray, ...],
     chunks: tuple[str, ...],
     playback_speed: float,
+    chunk_specs: tuple[GenerationChunkSpec, ...] | None = None,
 ) -> np.ndarray:
     """Assemble chunks exactly as they will be returned to the listener."""
 
     if not trajectory or len(trajectory) != len(chunks):
         raise ValueError("trajectory and text chunks must be non-empty and aligned")
+    if chunk_specs is not None and (
+        len(chunk_specs) != len(chunks)
+        or any(spec.text != chunk for spec, chunk in zip(chunk_specs, chunks, strict=True))
+    ):
+        raise ValueError("generation chunk provenance does not align with text")
+    if chunk_specs is not None:
+        for index, spec in enumerate(chunk_specs):
+            if (
+                spec.source_start < 0
+                or spec.source_end <= spec.source_start
+                or spec.boundary_after
+                not in {"semantic", "network_internal", "none"}
+            ):
+                raise ValueError("generation chunk provenance is invalid")
+            is_last = index + 1 == len(chunk_specs)
+            if is_last:
+                if spec.boundary_after != "none":
+                    raise ValueError("final generation boundary must be none")
+                continue
+            following = chunk_specs[index + 1]
+            if (
+                spec.source_end != following.source_start
+                or spec.boundary_after == "none"
+            ):
+                raise ValueError("generation chunk boundaries are incomplete")
+            if spec.boundary_after == "network_internal" and (
+                not spec.network_conditioned
+                or not following.network_conditioned
+                or not set(spec.network_span_indices).intersection(
+                    following.network_span_indices
+                )
+            ):
+                raise ValueError("network boundary lacks shared identifier proof")
     audio_chunks = [np.asarray(audio, dtype=np.float32).copy() for audio in trajectory]
     pauses: list[int] = []
+    fades_ms: list[float] = []
+    crossfades_ms: list[float] = []
     for index, chunk in enumerate(chunks):
         if index > 0:
             audio_chunks[index] = match_chunk_rms(
@@ -611,13 +666,34 @@ def _assemble_trajectory_audio(
                 max_adjust_db=CHUNK_RMS_MATCH_DB,
             )
         if index + 1 < len(chunks):
-            pauses.append(int(round(punctuation_pause_seconds(chunk) * SR)))
+            network_internal = bool(
+                chunk_specs is not None
+                and chunk_specs[index].boundary_after == "network_internal"
+            )
+            pauses.append(
+                0
+                if network_internal
+                else int(round(punctuation_pause_seconds(chunk) * SR))
+            )
+            fades_ms.append(
+                NETWORK_INTERNAL_FADE_MS
+                if network_internal
+                else CHUNK_EDGE_FADE_MS
+            )
+            crossfades_ms.append(
+                NETWORK_INTERNAL_FADE_MS
+                if network_internal
+                else CROSSFADE_MS
+            )
 
-    audio_chunks = fade_internal_edges(audio_chunks, SR, fade_ms=CHUNK_EDGE_FADE_MS)
-    waveform = join_audio_chunks(
+    audio_chunks = fade_variable_internal_edges(audio_chunks, SR, fades_ms)
+    waveform = join_audio_chunks_variable(
         audio_chunks,
         pauses,
-        crossfade_samples=int(round(CROSSFADE_MS * SR / 1000.0)),
+        crossfade_samples_by_boundary=[
+            int(round(crossfade_ms * SR / 1000.0))
+            for crossfade_ms in crossfades_ms
+        ],
         pre_faded_edges=True,
     )
     waveform = apply_loudness_floor(
@@ -657,6 +733,7 @@ def _qualify_candidate_trajectory_audio(
     independent_cache: WholeWaveformVerificationCache,
     *,
     candidate_seed: int,
+    chunk_specs: tuple[GenerationChunkSpec, ...] | None = None,
 ):
     """Run whole-output qualification only after every local chunk passes."""
 
@@ -669,7 +746,12 @@ def _qualify_candidate_trajectory_audio(
     if not local_verification.passed:
         return CandidateVerification(local_verification)
 
-    waveform = _assemble_trajectory_audio(trajectory, chunks, playback_speed)
+    waveform = _assemble_trajectory_audio(
+        trajectory,
+        chunks,
+        playback_speed,
+        chunk_specs,
+    )
     joined_verification = _verify_trajectory_audio(
         (waveform,),
         (whole_target_text,),
@@ -740,6 +822,7 @@ def _verify_sequence_trajectory_audio(
     anchor: np.ndarray,
     playback_speed: float,
     independent_cache: WholeWaveformVerificationCache,
+    chunk_specs: tuple[GenerationChunkSpec, ...] | None = None,
 ):
     """Verify one ranked DP path after exact production assembly."""
 
@@ -747,6 +830,7 @@ def _verify_sequence_trajectory_audio(
         sequence_result.trajectory,
         chunks,
         playback_speed,
+        chunk_specs,
     )
     turbo_verification = _verify_trajectory_audio(
         (waveform,),
@@ -813,12 +897,13 @@ def _synthesize(
         raise gr.Error("CFG 必須介於 1.0 與 4.0。")
     if cfg_value != MIXED_CFG_PRIMARY:
         raise gr.Error(f"目前只支援已驗證的主 CFG {MIXED_CFG_PRIMARY:.1f}。")
-    if network_identifier_has_ambiguous_iri(text):
+    raw_text = str(text)
+    if network_identifier_has_ambiguous_iri(raw_text):
         raise gr.Error("網址目前只支援 ASCII 字元；非 ASCII IRI 會與字母讀音混淆。")
-    network_request = contains_network_identifier(text)
+    network_request = contains_network_identifier(raw_text)
     request_cfg = MIXED_CFG_PRIMARY
     try:
-        text = normalize_spoken_forms(text, locale="zh-TW")
+        text = normalize_spoken_forms(raw_text, locale="zh-TW")
     except ValueError as error:
         raise gr.Error(str(error)) from None
     if not text:
@@ -830,38 +915,66 @@ def _synthesize(
     if not np.isfinite(float(speed)) or not 0.85 <= float(speed) <= 1.05:
         raise gr.Error("後處理語速必須介於 0.85 與 1.05。")
 
-    chunks = split_text_for_tts(text, max_chars=CHUNK_CHARS, min_chunk_chars=MIN_CHUNK_CHARS)
-    chunks = coalesce_text_chunks(
-        chunks,
-        max_chunks=QUALITY_MAX_GENERATED_CHUNKS,
-        max_units=CHUNK_CHARS,
-    )
+    chunk_specs: tuple[GenerationChunkSpec, ...] | None = None
+    try:
+        if network_request:
+            chunk_specs = plan_generation_chunks(
+                raw_text,
+                text,
+                min_units=MIN_CHUNK_CHARS,
+                target_units=NETWORK_GENERATION_TARGET_UNITS,
+                network_max_units=NETWORK_GENERATION_MAX_UNITS,
+                ordinary_max_units=CHUNK_CHARS,
+            )
+            chunks = tuple(spec.text for spec in chunk_specs)
+            if len(chunks) > QUALITY_MAX_GENERATED_CHUNKS:
+                raise ValueError(
+                    "network request exceeds the generated-chunk budget"
+                )
+        else:
+            chunks = tuple(
+                coalesce_text_chunks(
+                    split_text_for_tts(
+                        text,
+                        max_chars=CHUNK_CHARS,
+                        min_chunk_chars=MIN_CHUNK_CHARS,
+                    ),
+                    max_chunks=QUALITY_MAX_GENERATED_CHUNKS,
+                    max_units=CHUNK_CHARS,
+                )
+            )
+    except ValueError as error:
+        raise gr.Error("文字無法在已驗證的生成限制內安全分段。") from error
     request_seed = resolve_request_seed(request_seed, secrets.randbelow)
     anchor = _speaker_anchor_array(centroid)
     independent_cache = WholeWaveformVerificationCache()
 
-    def candidate_cfg(seed: int) -> float:
+    def candidate_cfg(candidate_ordinal: int) -> float:
         return generation_cfg_for_candidate_offset(
-            seed - request_seed,
+            candidate_ordinal,
             primary_cfg=request_cfg,
             alternate_cfg=MIXED_CFG_ALTERNATE,
         )
 
     def chunk_cfg_evidence(
         chunk: str,
-        candidate_offset: int,
+        candidate_ordinal: int,
+        *,
+        network_conditioned: bool | None = None,
     ) -> tuple[float, tuple[str, ...]]:
         scheduled = generation_cfg_for_candidate_offset(
-            candidate_offset,
+            candidate_ordinal,
             primary_cfg=request_cfg,
             alternate_cfg=MIXED_CFG_ALTERNATE,
         )
         reasons: list[str] = []
         network_adjusted = scheduled
-        if (
-            network_protected_spoken_spans(chunk)
-            and scheduled < NETWORK_TEXT_CFG_MIN
-        ):
+        is_network = (
+            bool(network_protected_spoken_spans(chunk))
+            if network_conditioned is None
+            else bool(network_conditioned)
+        )
+        if is_network and scheduled < NETWORK_TEXT_CFG_MIN:
             network_adjusted = NETWORK_TEXT_CFG_MIN
             reasons.append("network")
         effective = effective_generation_cfg(
@@ -874,21 +987,105 @@ def _synthesize(
             reasons.append("short_text")
         return effective, tuple(reasons)
 
-    def effective_chunk_cfg(chunk: str, candidate_offset: int) -> float:
-        return chunk_cfg_evidence(chunk, candidate_offset)[0]
+    def generation_network_flags(
+        chunk_indices: tuple[int, ...],
+        candidate_chunks: tuple[str, ...],
+    ) -> tuple[bool, ...]:
+        if len(chunk_indices) != len(candidate_chunks):
+            raise ValueError("generation provenance does not align with chunks")
+        if chunk_specs is None:
+            return tuple(
+                bool(network_protected_spoken_spans(chunk))
+                for chunk in candidate_chunks
+            )
+        flags: list[bool] = []
+        for chunk_index, chunk in zip(
+            chunk_indices,
+            candidate_chunks,
+            strict=True,
+        ):
+            if not 0 <= chunk_index < len(chunk_specs):
+                raise ValueError("generation provenance index is out of range")
+            spec = chunk_specs[chunk_index]
+            if spec.text != chunk:
+                raise ValueError("generation provenance text does not match")
+            flags.append(spec.network_conditioned)
+        return tuple(flags)
+
+    def effective_chunk_cfg(
+        chunk: str,
+        candidate_ordinal: int,
+        *,
+        network_conditioned: bool | None = None,
+    ) -> float:
+        return chunk_cfg_evidence(
+            chunk,
+            candidate_ordinal,
+            network_conditioned=network_conditioned,
+        )[0]
+
+    def candidate_generation(
+        candidate_chunks: tuple[str, ...],
+        seed: int,
+        *,
+        generation_context: CandidateGenerationContext,
+    ) -> tuple[np.ndarray, ...]:
+        if generation_context.seed != seed:
+            raise ValueError("generation context seed does not match the request seed")
+        ordinals = generation_context.chunk_candidate_ordinals
+        if not ordinals or len(set(ordinals)) != 1:
+            raise ValueError("one generation call must use one candidate ordinal")
+        candidate_ordinal = ordinals[0]
+        network_flags = generation_network_flags(
+            generation_context.chunk_indices,
+            candidate_chunks,
+        )
+        return _generate_trajectory(
+            candidate_chunks,
+            centroid,
+            cfg=candidate_cfg(candidate_ordinal),
+            steps=steps,
+            request_seed=seed,
+            policy=generation_policy_for_candidate_offset(candidate_ordinal),
+            network_conditioned=network_flags,
+        )
 
     def candidate_generation_evidence(
         candidate_index: int,
         seed: int,
         chunk_indices: tuple[int, ...],
         candidate_chunks: tuple[str, ...],
+        *,
+        generation_context: CandidateGenerationContext,
     ) -> CandidateGenerationEvidence:
         if candidate_index != seed - request_seed:
             raise ValueError("candidate index does not match request seed offset")
-        scheduled = candidate_cfg(seed)
+        if (
+            generation_context.candidate_index != candidate_index
+            or generation_context.seed != seed
+            or generation_context.chunk_indices != chunk_indices
+        ):
+            raise ValueError("candidate generation context does not match the attempt")
+        candidate_ordinals = generation_context.chunk_candidate_ordinals
+        if not candidate_ordinals or len(set(candidate_ordinals)) != 1:
+            raise ValueError("one generation call must use one candidate ordinal")
+        candidate_ordinal = candidate_ordinals[0]
+        scheduled = candidate_cfg(candidate_ordinal)
+        network_flags = generation_network_flags(
+            chunk_indices,
+            candidate_chunks,
+        )
         rows = tuple(
-            chunk_cfg_evidence(chunk, candidate_index)
-            for chunk in candidate_chunks
+            chunk_cfg_evidence(
+                chunk,
+                candidate_ordinal,
+                network_conditioned=network_flag,
+            )
+            for chunk, network_flag in zip(
+                candidate_chunks,
+                network_flags,
+                strict=True,
+            )
         )
         return CandidateGenerationEvidence(
             chunk_indices=chunk_indices,
@@ -898,6 +1095,8 @@ def _synthesize(
             scheduled_cfg=scheduled,
             effective_cfgs=tuple(row[0] for row in rows),
             floor_reasons=tuple(row[1] for row in rows),
+            chunk_candidate_ordinals=candidate_ordinals,
+            network_conditioned=network_flags,
         )
 
     try:
@@ -905,14 +1104,7 @@ def _synthesize(
             cascade = run_coverage_adaptive_cascade(
                 chunks,
                 request_seed,
-                lambda candidate_chunks, seed: _generate_trajectory(
-                    candidate_chunks,
-                    centroid,
-                    cfg=candidate_cfg(seed),
-                    steps=steps,
-                    request_seed=seed,
-                    policy=generation_policy_for_candidate_offset(seed - request_seed),
-                ),
+                candidate_generation,
                 lambda trajectory, candidate_chunks, seed: _qualify_candidate_trajectory_audio(
                     trajectory,
                     candidate_chunks,
@@ -921,6 +1113,7 @@ def _synthesize(
                     speed,
                     independent_cache,
                     candidate_seed=seed,
+                    chunk_specs=chunk_specs,
                 ),
                 lambda trajectory, candidate_chunks, seed: (
                     _qualify_candidate_trajectory_audio(
@@ -931,6 +1124,7 @@ def _synthesize(
                         speed,
                         independent_cache,
                         candidate_seed=seed,
+                        chunk_specs=chunk_specs,
                     )
                     if len(chunks) == 1
                     else _verify_refill_candidate_trajectory_audio(
@@ -948,6 +1142,7 @@ def _synthesize(
                         anchor,
                         speed,
                         independent_cache,
+                        chunk_specs,
                     )
                 ),
                 generation_evidence_factory=candidate_generation_evidence,
@@ -971,24 +1166,49 @@ def _synthesize(
     except (RuntimeError, ValueError) as error:
         raise gr.Error("品質驗證暫時無法完成，未回傳未驗證的語音。") from error
 
+    attempts_by_index = {
+        attempt.candidate_index: attempt for attempt in cascade.diagnostics.attempts
+    }
+    selected_ordinals: list[int] = []
+    for chunk_index, candidate_index in enumerate(cascade.chunk_candidate_indices):
+        attempt = attempts_by_index.get(candidate_index)
+        if attempt is None or chunk_index not in attempt.chunk_indices:
+            raise gr.Error("品質驗證紀錄不完整，未回傳未驗證的語音。")
+        local_index = attempt.chunk_indices.index(chunk_index)
+        try:
+            selected_ordinals.append(attempt.chunk_candidate_ordinals[local_index])
+        except IndexError as error:
+            raise gr.Error("品質驗證紀錄不完整，未回傳未驗證的語音。") from error
     selected_policies = tuple(
-        generation_policy_for_candidate_offset(index).name
-        for index in cascade.chunk_candidate_indices
+        generation_policy_for_candidate_offset(ordinal).name
+        for ordinal in selected_ordinals
     )
     attempted_policies = tuple(
-        generation_policy_for_candidate_offset(seed - request_seed).name
-        for seed in cascade.attempted_seeds
+        generation_policy_for_candidate_offset(
+            attempt.chunk_candidate_ordinals[0]
+        ).name
+        for attempt in cascade.diagnostics.attempts
+    )
+    selected_network_flags = generation_network_flags(
+        tuple(range(len(chunks))),
+        chunks,
     )
     selected_cfgs = tuple(
-        effective_chunk_cfg(chunk, candidate_index)
-        for chunk, candidate_index in zip(
+        effective_chunk_cfg(
+            chunk,
+            candidate_ordinal,
+            network_conditioned=network_flag,
+        )
+        for chunk, candidate_ordinal, network_flag in zip(
             chunks,
-            cascade.chunk_candidate_indices,
+            selected_ordinals,
+            selected_network_flags,
             strict=True,
         )
     )
     attempted_schedule_cfgs = tuple(
-        candidate_cfg(seed) for seed in cascade.attempted_seeds
+        candidate_cfg(attempt.chunk_candidate_ordinals[0])
+        for attempt in cascade.diagnostics.attempts
     )
     print(
         "[BlueMagpie] quality cascade "
@@ -1009,7 +1229,12 @@ def _synthesize(
         f" cfg_schedule={MIXED_CFG_SCHEDULE}"
         f" network_request={network_request}"
     )
-    waveform = _assemble_trajectory_audio(cascade.trajectory, chunks, speed)
+    waveform = _assemble_trajectory_audio(
+        cascade.trajectory,
+        chunks,
+        speed,
+        chunk_specs,
+    )
     final_verification = _verify_trajectory_audio(
         (waveform,),
         (text,),

production.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import math
+import operator
 import random
 import re
 import unicodedata
@@ -1139,7 +1140,7 @@ def network_identifier_has_ambiguous_iri(text: str) -> bool:
 
     raw = unicodedata.normalize("NFC", str(text or ""))
     return any(
-        any(character.isalnum() and not character.isascii() for character in match.group(0))
+        not match.group(0).isascii()
         for match in _SPOKEN_URL_RE.finditer(raw)
     )
 
@@ -2332,6 +2333,10 @@ _PROTECTED_ASCII_SPAN_RE = re.compile(
 )
 _STRUCTURED_CLAUSE_BREAKS = frozenset("，,、：:")
 _STRUCTURED_CLAUSE_TARGET_UNITS = 32
+_NETWORK_GENERATION_COMPONENT_AFTER = frozenset()
+_NETWORK_GENERATION_COMPONENT_BEFORE = frozenset(
+    {"點", "小老鼠", "斜線", "問號", "井號", "冒號", "和"}
+)
 _NORMALIZED_ZH_NUMBER_PATTERN = (
     r"[正負]?[" + _ZH_DIGITS + r"十百千萬億兆]+"
     r"(?:點[" + _ZH_DIGITS + r"]+)?"
@@ -2655,6 +2660,515 @@ def split_text_for_tts(text: str, max_chars: int = 80, min_chunk_chars: int = 12
     return chunks or [text]
 
 
+@dataclass(frozen=True)
+class GenerationChunkSpec:
+    """One generation-only slice with immutable full-identifier provenance."""
+
+    text: str
+    source_start: int
+    source_end: int
+    network_span_indices: tuple[int, ...] = ()
+    network_component_indices: tuple[tuple[int, int], ...] = ()
+    network_full_spoken_proofs: tuple[str, ...] = ()
+    boundary_after: str = "none"
+
+    @property
+    def network_conditioned(self) -> bool:
+        return bool(self.network_span_indices)
+
+
+@dataclass(frozen=True)
+class _NetworkGenerationSpan:
+    start: int
+    end: int
+    spoken_proof: str
+    component_ranges: tuple[tuple[int, int], ...]
+
+
+def _raw_network_generation_matches(
+    text: str,
+) -> tuple[tuple[int, int, str, str], ...]:
+    """Return non-overlapping raw identifiers and their exact spoken proofs."""
+
+    raw = unicodedata.normalize("NFC", str(text or ""))
+    matches: list[tuple[int, int, str, str]] = []
+    for match in _SPOKEN_URL_RE.finditer(raw):
+        value = match.group(0)
+        if not value.isascii():
+            raise ValueError("network generation does not support non-ASCII IRI")
+        matches.append((match.start(), match.end(), _zh_url(value), "url"))
+    for match in _SPOKEN_EMAIL_RE.finditer(raw):
+        value = match.group(0)
+        if not value.isascii():
+            raise ValueError("network generation does not support non-ASCII email")
+        matches.append((match.start(), match.end(), _zh_email(value), "email"))
+
+    selected: list[tuple[int, int, str, str]] = []
+    for start, end, spoken, kind in sorted(
+        matches,
+        key=lambda item: (item[0], -item[1]),
+    ):
+        if any(
+            start < kept_end and end > kept_start
+            for kept_start, kept_end, _, _ in selected
+        ):
+            continue
+        selected.append((start, end, normalize_tts_text(spoken), kind))
+    return tuple(sorted(selected))
+
+
+def _network_component_ranges(
+    spoken: str,
+    *,
+    absolute_start: int,
+    hard_max_units: int,
+) -> tuple[tuple[int, int], ...]:
+    """Split one proven identifier only at audible grammar delimiters."""
+
+    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
+    )
+    cuts: set[int] = {0, len(spoken)}
+    scheme_slashes: set[int] = set()
+    for index in range(2, len(tokens)):
+        if tokens[index - 2 : index + 1] == ("冒號", "斜線", "斜線"):
+            scheme_slashes.update({index - 1, index})
+            cuts.add(token_matches[index].end())
+    for index, (token, match) in enumerate(zip(tokens, token_matches, strict=True)):
+        if token in _NETWORK_GENERATION_COMPONENT_AFTER:
+            cuts.add(match.end())
+        if (
+            token in _NETWORK_GENERATION_COMPONENT_BEFORE
+            and index not in scheme_slashes
+            and not (token == "冒號" and index + 2 < len(tokens) and tokens[index + 1 : index + 3] == ("斜線", "斜線"))
+        ):
+            cuts.add(match.start())
+
+    ordered = sorted(cuts)
+    ranges: list[tuple[int, int]] = []
+    for left, right in zip(ordered, ordered[1:]):
+        component = normalize_tts_text(spoken[left:right])
+        if not component:
+            continue
+        if count_speech_units(component) > hard_max_units:
+            raise ValueError("an indivisible network component exceeds the generation limit")
+        ranges.append((absolute_start + left, absolute_start + right))
+    if not ranges:
+        raise ValueError("network generation proof has no components")
+    if ranges[0][0] != absolute_start or ranges[-1][1] != absolute_start + len(spoken):
+        raise ValueError("network generation components do not cover their proof")
+    return tuple(ranges)
+
+
+def _network_generation_spans(
+    raw_text: str,
+    normalized_text: str,
+    *,
+    hard_max_units: int,
+) -> tuple[_NetworkGenerationSpan, ...]:
+    """Bind raw ASCII grammar to exact offsets in the normalized model text."""
+
+    normalized = normalize_tts_text(normalized_text)
+    expected = normalize_spoken_forms(raw_text)
+    if normalized != expected:
+        raise ValueError("normalized text does not match the raw network request")
+    raw = unicodedata.normalize("NFC", str(raw_text or ""))
+    raw_matches = _raw_network_generation_matches(raw)
+    spans: list[_NetworkGenerationSpan] = []
+    previous_raw_end: int | None = None
+    previous_kind: str | None = None
+    cursor = 0
+    for raw_start, raw_end, spoken, kind in raw_matches:
+        # Derive the offset from the exact raw prefix rather than searching for
+        # a possibly duplicated spoken phrase.  The frontend inserts a comma
+        # on each unpunctuated side of a network identifier; a preceding
+        # identifier at the end of the prefix needs its deferred suffix comma
+        # accounted for as well.
+        normalized_prefix = normalize_spoken_forms(raw[:raw_start])
+        boundary_count = 0
+        preceding = raw[:raw_start].rstrip()[-1:]
+        if preceding and preceding not in _NETWORK_BOUNDARY_PUNCTUATION:
+            boundary_count += 1
+        if (
+            previous_raw_end is not None
+            and previous_kind == kind
+            and not raw[previous_raw_end:raw_start].strip()
+        ):
+            following = raw[previous_raw_end:].lstrip()[:1]
+            if following and following not in _NETWORK_BOUNDARY_PUNCTUATION:
+                boundary_count += 1
+        start = len(normalized_prefix) + boundary_count
+        if start < cursor or normalized[start : start + len(spoken)] != spoken:
+            raise ValueError(
+                "network spoken proof offset does not match normalized text"
+            )
+        end = start + len(spoken)
+        components = _network_component_ranges(
+            spoken,
+            absolute_start=start,
+            hard_max_units=hard_max_units,
+        )
+        spans.append(
+            _NetworkGenerationSpan(
+                start=start,
+                end=end,
+                spoken_proof=spoken,
+                component_ranges=components,
+            )
+        )
+        cursor = end
+        previous_raw_end = raw_end
+        previous_kind = kind
+    return tuple(spans)
+
+
+def _network_generation_group_plan(
+    text: str,
+    *,
+    group_start: int,
+    group_end: int,
+    original_boundaries: Sequence[int],
+    spans: Sequence[_NetworkGenerationSpan],
+    minimum: int,
+    target: int,
+    network_maximum: int,
+    ordinary_maximum: int,
+) -> tuple[tuple[int, int], ...]:
+    """Deterministically pack one network-bearing sentence near the target."""
+
+    group_spans = tuple(
+        span for span in spans if span.start < group_end and span.end > group_start
+    )
+    if not group_spans:
+        return tuple(
+            zip(
+                (group_start, *original_boundaries),
+                (*original_boundaries, group_end),
+                strict=True,
+            )
+        )
+
+    preferred_cuts: set[int] = {
+        group_start,
+        group_end,
+        *original_boundaries,
+    }
+    component_cuts = {
+        offset
+        for span in group_spans
+        for component in span.component_ranges
+        for offset in component
+    }
+    preferred_cuts.update(component_cuts)
+    network_interiors = {
+        offset
+        for span in group_spans
+        for offset in range(span.start + 1, span.end)
+    }
+    for index in range(group_start, group_end):
+        end = index + 1
+        if text[index] in _STRUCTURED_CLAUSE_BREAKS and end not in network_interiors:
+            preferred_cuts.add(end)
+
+    # Every source offset outside an identifier remains an emergency prose
+    # split. Inside an identifier, only exact component boundaries are legal.
+    cuts = {
+        offset
+        for offset in range(group_start, group_end + 1)
+        if offset not in network_interiors or offset in component_cuts
+    }
+
+    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)
+    if plan is None:
+        raise ValueError("network-bearing sentence cannot satisfy generation limits")
+    output: list[tuple[int, int]] = []
+    start = group_start
+    for end in plan[4]:
+        output.append((start, end))
+        start = end
+    return tuple(output)
+
+
+def _semantic_sentence_ranges(text: str) -> tuple[tuple[int, int], ...]:
+    """Return exact, contiguous strong-sentence source ranges."""
+
+    if not text:
+        return ()
+    ranges: list[tuple[int, int]] = []
+    start = 0
+    index = 0
+    while index < len(text):
+        if text[index] not in _SEMANTIC_STRONG_BREAKS:
+            index += 1
+            continue
+        end = index + 1
+        while end < len(text) and text[end] in _SEMANTIC_TRAILING_CLOSERS:
+            end += 1
+        while end < len(text) and text[end].isspace():
+            end += 1
+        if count_speech_units(text[start:end]) > 0:
+            ranges.append((start, end))
+        start = end
+        index = end
+    if start < len(text) and count_speech_units(text[start:]) > 0:
+        ranges.append((start, len(text)))
+    if not ranges or ranges[0][0] != 0 or ranges[-1][1] != len(text):
+        raise ValueError("semantic source ranges do not cover normalized text")
+    if any(
+        left_end != right_start
+        for (_, left_end), (right_start, _) in zip(ranges, ranges[1:])
+    ):
+        raise ValueError("semantic source ranges are not contiguous")
+    return tuple(ranges)
+
+
+def _ordinary_generation_ranges(
+    text: str,
+    *,
+    start: int,
+    end: int,
+    minimum: int,
+    maximum: int,
+) -> tuple[tuple[int, int], ...]:
+    """Hard-split one exact non-network source range without re-normalizing it."""
+
+    source = text[start:end]
+    semantic_chunks = split_text_for_tts(
+        source,
+        max_chars=maximum,
+        min_chunk_chars=minimum,
+    )
+    if semantic_chunks and "".join(semantic_chunks) == source:
+        ranges: list[tuple[int, int]] = []
+        cursor = start
+        for chunk in semantic_chunks:
+            next_cursor = cursor + len(chunk)
+            ranges.append((cursor, next_cursor))
+            cursor = next_cursor
+        if cursor == end:
+            return tuple(ranges)
+    if count_speech_units(source) <= maximum:
+        return ((start, end),)
+    output: list[tuple[int, int]] = []
+    cursor = start
+    while count_speech_units(text[cursor:end]) > maximum:
+        local = text[cursor:end]
+        forbidden = _protected_split_offsets(local, maximum)
+        candidates: list[tuple[int, int, int]] = []
+        for local_offset in range(1, len(local)):
+            if local_offset in forbidden:
+                continue
+            head_units = count_speech_units(local[:local_offset])
+            tail_units = count_speech_units(local[local_offset:])
+            if not minimum <= head_units <= maximum:
+                continue
+            if 0 < tail_units < minimum:
+                continue
+            semantic = int(
+                local[local_offset - 1] in _SEMANTIC_SOFT_BREAKS
+                or local[local_offset].isspace()
+            )
+            candidates.append((semantic, head_units, -local_offset))
+        if not candidates:
+            raise ValueError("ordinary generation range cannot satisfy chunk limits")
+        _, _, negative_offset = max(candidates)
+        next_cursor = cursor - negative_offset
+        output.append((cursor, next_cursor))
+        cursor = next_cursor
+    output.append((cursor, end))
+    return tuple(output)
+
+
+def plan_generation_chunks(
+    raw_text: str,
+    normalized_text: str,
+    *,
+    min_units: int = 12,
+    target_units: int = 32,
+    network_max_units: int = 36,
+    ordinary_max_units: int = 80,
+) -> tuple[GenerationChunkSpec, ...]:
+    """Plan model chunks while retaining exact whole-network verification proof.
+
+    Ordinary semantic chunking is unchanged. Only sentences that contain a raw
+    ASCII URL/email are repacked at proven audible component delimiters. The
+    caller must continue to verify the untouched ``normalized_text`` as one
+    whole target before returning audio.
+    """
+
+    if network_identifier_has_ambiguous_iri(raw_text):
+        raise ValueError("network generation does not support non-ASCII IRI")
+    minimum = max(1, int(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:
+        raise ValueError("generation chunk limits are inconsistent")
+
+    normalized = normalize_tts_text(normalized_text)
+    if not contains_network_identifier(raw_text):
+        ordinary = split_text_for_tts(
+            normalized,
+            max_chars=ordinary_maximum,
+            min_chunk_chars=minimum,
+        )
+        specs: list[GenerationChunkSpec] = []
+        cursor = 0
+        for index, chunk in enumerate(ordinary):
+            start = normalized.find(chunk, cursor)
+            if start < 0:
+                raise ValueError("ordinary chunk is absent from normalized text")
+            end = start + len(chunk)
+            specs.append(
+                GenerationChunkSpec(
+                    text=chunk,
+                    source_start=start,
+                    source_end=end,
+                    boundary_after=("semantic" if index + 1 < len(ordinary) else "none"),
+                )
+            )
+            cursor = end
+        return tuple(specs)
+
+    spans = _network_generation_spans(
+        raw_text,
+        normalized,
+        hard_max_units=network_maximum,
+    )
+    if not spans:
+        raise ValueError("network request has no bound spoken proof")
+
+    planned_ranges: list[tuple[int, int]] = []
+    for sentence_start, sentence_end in _semantic_sentence_ranges(normalized):
+        sentence_has_network = any(
+            span.start < sentence_end and span.end > sentence_start
+            for span in spans
+        )
+        if sentence_has_network:
+            planned_ranges.extend(
+                _network_generation_group_plan(
+                    normalized,
+                    group_start=sentence_start,
+                    group_end=sentence_end,
+                    original_boundaries=(),
+                    spans=spans,
+                    minimum=minimum,
+                    target=target,
+                    network_maximum=network_maximum,
+                    ordinary_maximum=ordinary_maximum,
+                )
+            )
+        else:
+            planned_ranges.extend(
+                _ordinary_generation_ranges(
+                    normalized,
+                    start=sentence_start,
+                    end=sentence_end,
+                    minimum=minimum,
+                    maximum=ordinary_maximum,
+                )
+            )
+
+    if (
+        not planned_ranges
+        or planned_ranges[0][0] != 0
+        or planned_ranges[-1][1] != len(normalized)
+        or any(
+            left_end != right_start
+            for (_, left_end), (right_start, _) in zip(
+                planned_ranges,
+                planned_ranges[1:],
+            )
+        )
+    ):
+        raise ValueError("generation source ranges do not exactly cover the target")
+
+    specs = []
+    internal_boundaries = {
+        right
+        for span in spans
+        for _, right in span.component_ranges[:-1]
+    }
+    for start, end in planned_ranges:
+        chunk = normalized[start:end]
+        if not chunk or count_speech_units(chunk) <= 0:
+            raise ValueError("generation planner produced an empty chunk")
+        overlapping = tuple(
+            span_index
+            for span_index, span in enumerate(spans)
+            if span.start < end and span.end > start
+        )
+        component_indices = tuple(
+            (span_index, component_index)
+            for span_index in overlapping
+            for component_index, (left, right) in enumerate(
+                spans[span_index].component_ranges
+            )
+            if left < end and right > start
+        )
+        specs.append(
+            GenerationChunkSpec(
+                text=chunk,
+                source_start=start,
+                source_end=end,
+                network_span_indices=overlapping,
+                network_component_indices=component_indices,
+                network_full_spoken_proofs=tuple(
+                    spans[span_index].spoken_proof for span_index in overlapping
+                ),
+                boundary_after=(
+                    "network_internal"
+                    if end in internal_boundaries
+                    else ("semantic" if end < len(normalized) else "none")
+                ),
+            )
+        )
+    if "".join(spec.text for spec in specs) != normalized:
+        raise ValueError("generation chunks do not reconstruct the normalized target")
+    if any(
+        count_speech_units(spec.text)
+        > (network_maximum if spec.network_conditioned else ordinary_maximum)
+        for spec in specs
+    ):
+        raise ValueError("generation chunk exceeds its provenance-specific limit")
+    return tuple(specs)
+
+
 def coalesce_text_chunks(
     chunks: Sequence[str],
     *,
@@ -2761,6 +3275,57 @@ def fade_internal_edges(chunks: list[np.ndarray], sample_rate: int, fade_ms: flo
     return outputs
 
 
+def fade_variable_internal_edges(
+    chunks: Sequence[np.ndarray],
+    sample_rate: int,
+    fade_ms_by_boundary: Sequence[float],
+) -> list[np.ndarray]:
+    """Apply independently bounded fades at each adjacent chunk boundary."""
+
+    outputs = [np.asarray(chunk, dtype=np.float32).reshape(-1).copy() for chunk in chunks]
+    if len(fade_ms_by_boundary) != max(0, len(outputs) - 1):
+        raise ValueError("fade boundary count must equal chunk count minus one")
+    if sample_rate <= 0:
+        raise ValueError("sample_rate must be positive")
+    if any(output.size <= 0 for output in outputs):
+        raise ValueError("audio chunks must be non-empty")
+    widths: list[int] = []
+    for index, raw_fade_ms in enumerate(fade_ms_by_boundary):
+        try:
+            fade_ms = float(raw_fade_ms)
+        except (TypeError, ValueError, OverflowError) as error:
+            raise ValueError("fade duration must be finite and non-negative") from error
+        if not math.isfinite(fade_ms) or fade_ms < 0.0:
+            raise ValueError("fade duration must be finite and non-negative")
+        requested = int(round(fade_ms * sample_rate / 1000.0))
+        widths.append(
+            min(requested, outputs[index].size, outputs[index + 1].size)
+        )
+    if any(
+        widths[index - 1] + widths[index] > outputs[index].size
+        for index in range(1, len(outputs) - 1)
+    ):
+        raise ValueError("adjacent fades overlap inside an audio chunk")
+    for index, count in enumerate(widths):
+        if count <= 0:
+            continue
+        outputs[index][-count:] *= np.linspace(
+            1.0,
+            0.0,
+            count,
+            endpoint=True,
+            dtype=np.float32,
+        )
+        outputs[index + 1][:count] *= np.linspace(
+            0.0,
+            1.0,
+            count,
+            endpoint=True,
+            dtype=np.float32,
+        )
+    return outputs
+
+
 def join_audio_chunks(
     chunks: list[np.ndarray],
     pauses: list[int],
@@ -2792,6 +3357,105 @@ def join_audio_chunks(
     return output.astype(np.float32, copy=False)
 
 
+def join_audio_chunks_variable(
+    chunks: Sequence[np.ndarray],
+    pauses: Sequence[int],
+    crossfade_samples_by_boundary: Sequence[int],
+    *,
+    pre_faded_edges: bool = False,
+) -> np.ndarray:
+    """Join chunks with one validated crossfade width per boundary."""
+
+    if not chunks:
+        if pauses or crossfade_samples_by_boundary:
+            raise ValueError("empty chunks cannot carry boundary metadata")
+        return np.zeros(0, dtype=np.float32)
+    boundary_count = len(chunks) - 1
+    if len(pauses) != boundary_count or len(crossfade_samples_by_boundary) != boundary_count:
+        raise ValueError("pause/crossfade boundary counts must equal chunk count minus one")
+    prepared = [
+        np.asarray(chunk, dtype=np.float32).reshape(-1).copy()
+        for chunk in chunks
+    ]
+    if any(chunk.size <= 0 for chunk in prepared):
+        raise ValueError("audio chunks must be non-empty")
+    validated_pauses: list[int] = []
+    requested_crossfades: list[int] = []
+    for raw_pause, raw_crossfade in zip(
+        pauses,
+        crossfade_samples_by_boundary,
+        strict=True,
+    ):
+        if isinstance(raw_pause, (bool, np.bool_)) or isinstance(
+            raw_crossfade,
+            (bool, np.bool_),
+        ):
+            raise ValueError("pause and crossfade widths must be non-negative integers")
+        try:
+            pause = operator.index(raw_pause)
+            crossfade = operator.index(raw_crossfade)
+        except (TypeError, ValueError, OverflowError) as error:
+            raise ValueError(
+                "pause and crossfade widths must be non-negative integers"
+            ) from error
+        if pause < 0 or crossfade < 0:
+            raise ValueError("pause and crossfade widths must be non-negative integers")
+        validated_pauses.append(int(pause))
+        requested_crossfades.append(int(crossfade))
+    widths = [
+        min(requested, prepared[index].size, prepared[index + 1].size)
+        for index, requested in enumerate(requested_crossfades)
+    ]
+    if any(
+        widths[index - 1] + widths[index] > prepared[index].size
+        for index in range(1, len(prepared) - 1)
+    ):
+        raise ValueError("adjacent crossfades overlap inside an audio chunk")
+
+    output = prepared[0]
+    for index, next_chunk in enumerate(prepared[1:]):
+        pause = validated_pauses[index]
+        crossfade = widths[index]
+        if pause > 0:
+            if crossfade > 0 and not pre_faded_edges:
+                output[-crossfade:] *= np.linspace(
+                    1.0,
+                    0.0,
+                    crossfade,
+                    dtype=np.float32,
+                )
+                next_chunk[:crossfade] *= np.linspace(
+                    0.0,
+                    1.0,
+                    crossfade,
+                    dtype=np.float32,
+                )
+            output = np.concatenate(
+                (output, np.zeros(pause, dtype=np.float32), next_chunk)
+            )
+        elif crossfade > 0:
+            if pre_faded_edges:
+                overlap = output[-crossfade:] + next_chunk[:crossfade]
+            else:
+                fade_out = np.linspace(
+                    1.0,
+                    0.0,
+                    crossfade,
+                    endpoint=False,
+                    dtype=np.float32,
+                )
+                overlap = (
+                    output[-crossfade:] * fade_out
+                    + next_chunk[:crossfade] * (1.0 - fade_out)
+                )
+            output = np.concatenate(
+                (output[:-crossfade], overlap, next_chunk[crossfade:])
+            )
+        else:
+            output = np.concatenate((output, next_chunk))
+    return output.astype(np.float32, copy=False)
+
+
 def apply_loudness_floor(
     audio: np.ndarray,
     min_rms: float = 0.07,

quality_runtime.py

@@ -8,6 +8,7 @@ load the real models lazily while unit tests remain deterministic and offline.
 from __future__ import annotations
 
 import hashlib
+import inspect
 import json
 import math
 import operator
@@ -46,7 +47,7 @@ RELEASE_SPEAKER_TRIGGER_SECONDS = 1.48
 SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP = 0.15
 SEQUENCE_FALLBACK_SPEAKER_WEIGHT = 0.05
 SEQUENCE_FALLBACK_BOUNDARY_WEIGHT = 0.10
-CASCADE_EVIDENCE_SCHEMA_VERSION = 2
+CASCADE_EVIDENCE_SCHEMA_VERSION = 3
 CASCADE_EVIDENCE_LOG_PREFIX = "[BlueMagpie] cascade evidence "
 CASCADE_EVIDENCE_MAX_ATTEMPTS = ADAPTIVE_CASCADE_STAGE_LIMITS[-1]
 CASCADE_EVIDENCE_MAX_LOCAL_RESULTS = 20
@@ -93,6 +94,23 @@ class GenerationPolicy:
     hard_stop_margin_steps: int
 
 
+@dataclass(frozen=True)
+class CandidateGenerationContext:
+    """Immutable global identity and row-local schedule for one generation.
+
+    ``candidate_index`` and ``seed`` retain the request-global attempt identity.
+    ``chunk_candidate_ordinals`` is independent: zero denotes the initial
+    trajectory and positive values count refill attempts within each source
+    chunk.  Context-aware callbacks can therefore rotate generation policies
+    per chunk without changing the canonical seed schedule.
+    """
+
+    candidate_index: int
+    seed: int
+    chunk_indices: tuple[int, ...]
+    chunk_candidate_ordinals: tuple[int, ...]
+
+
 BASE_GENERATION_POLICY = GenerationPolicy(
     name="base",
     cjk_cps=5.2,
@@ -111,7 +129,7 @@ COMPLETION_HEADROOM_GENERATION_POLICY = GenerationPolicy(
     ascii_cps=3.6,
     hard_stop_margin_steps=1,
 )
-MIXED_CFG_SCHEDULE = "offset_zero_and_even_primary_odd_alternate"
+MIXED_CFG_SCHEDULE = "row_ordinal_zero_and_even_primary_odd_alternate"
 MIXED_CFG_PRIMARY = 3.0
 MIXED_CFG_ALTERNATE = 2.0
 MIXED_CFG_SHORT_TEXT_MAX_UNITS = 6
@@ -1181,6 +1199,8 @@ class CandidateGenerationEvidence:
     scheduled_cfg: float
     effective_cfgs: tuple[float, ...]
     floor_reasons: tuple[tuple[str, ...], ...]
+    chunk_candidate_ordinals: tuple[int, ...] = ()
+    network_conditioned: tuple[bool, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -1196,6 +1216,8 @@ class CandidateAttemptEvidence:
     local_results: tuple[CandidateGateEvidence, ...]
     chunk_indices: tuple[int, ...]
     chunk_text_units: tuple[int, ...]
+    chunk_candidate_ordinals: tuple[int, ...]
+    network_conditioned: tuple[bool, ...]
     scheduled_cfg: float | None
     effective_cfgs: tuple[float, ...]
     floor_reasons: tuple[tuple[str, ...], ...]
@@ -1804,12 +1826,20 @@ def _trajectory_gate_evidence_payload(
 def _candidate_attempt_evidence_payload(
     evidence: CandidateAttemptEvidence,
 ) -> dict[str, Any]:
+    candidate_ordinals = evidence.chunk_candidate_ordinals
+    chunk_policies = [
+        generation_policy_for_candidate_offset(ordinal).name
+        for ordinal in candidate_ordinals
+    ]
+    policy = (
+        chunk_policies[0]
+        if chunk_policies and len(set(chunk_policies)) == 1
+        else generation_policy_for_candidate_offset(evidence.candidate_index).name
+    )
     return {
         "candidate_index": evidence.candidate_index,
         "seed": evidence.seed,
-        "policy": generation_policy_for_candidate_offset(
-            evidence.candidate_index
-        ).name,
+        "policy": policy,
         "trajectory_passed": evidence.trajectory_passed,
         "trajectory_score": evidence.trajectory_score,
         "trajectory_reasons": list(
@@ -1817,6 +1847,9 @@ def _candidate_attempt_evidence_payload(
         ),
         "chunk_indices": list(evidence.chunk_indices),
         "chunk_text_units": list(evidence.chunk_text_units),
+        "chunk_candidate_ordinals": list(candidate_ordinals),
+        "chunk_policies": chunk_policies,
+        "network_conditioned": list(evidence.network_conditioned),
         "scheduled_cfg": evidence.scheduled_cfg,
         "effective_cfgs": list(evidence.effective_cfgs),
         "floor_reasons": [list(reasons) for reasons in evidence.floor_reasons],
@@ -1891,7 +1924,9 @@ def _selected_generation_evidence_payload(
     scheduled_cfgs: list[float | None] = []
     effective_cfgs: list[float | None] = []
     floor_reasons: list[list[str]] = []
+    candidate_ordinals: list[int | None] = []
     policies: list[str | None] = []
+    network_conditioned: list[bool | None] = []
     complete = True
     for chunk_index, candidate_index in enumerate(
         selection.chunk_candidate_indices
@@ -1902,7 +1937,9 @@ def _selected_generation_evidence_payload(
             scheduled_cfgs.append(None)
             effective_cfgs.append(None)
             floor_reasons.append([])
+            candidate_ordinals.append(None)
             policies.append(None)
+            network_conditioned.append(None)
             continue
         try:
             local_index = attempt.chunk_indices.index(chunk_index)
@@ -1913,20 +1950,37 @@ def _selected_generation_evidence_payload(
             scheduled_cfgs.append(attempt.scheduled_cfg)
             effective_cfgs.append(None)
             floor_reasons.append([])
+            candidate_ordinals.append(None)
             policies.append(None)
+            network_conditioned.append(None)
             continue
+        try:
+            ordinal = attempt.chunk_candidate_ordinals[local_index]
+        except IndexError:
+            complete = False
+            ordinal = None
         scheduled_cfgs.append(attempt.scheduled_cfg)
         effective_cfgs.append(effective)
         floor_reasons.append(list(reasons))
+        candidate_ordinals.append(ordinal)
         policies.append(
-            generation_policy_for_candidate_offset(candidate_index).name
+            None
+            if ordinal is None
+            else generation_policy_for_candidate_offset(ordinal).name
         )
+        try:
+            network_conditioned.append(attempt.network_conditioned[local_index])
+        except IndexError:
+            complete = False
+            network_conditioned.append(None)
     return {
         "complete": complete,
         "chunk_scheduled_cfgs": scheduled_cfgs,
         "chunk_effective_cfgs": effective_cfgs,
         "chunk_floor_reasons": floor_reasons,
+        "chunk_candidate_ordinals": candidate_ordinals,
         "chunk_policies": policies,
+        "network_conditioned": network_conditioned,
     }
 
 
@@ -2008,6 +2062,8 @@ def format_cascade_evidence_log(
     generation_evidence_complete = bool(attempts) and all(
         attempt.scheduled_cfg is not None
         and len(attempt.chunk_indices) == len(attempt.chunk_text_units)
+        == len(attempt.chunk_candidate_ordinals)
+        == len(attempt.network_conditioned)
         == len(attempt.effective_cfgs)
         == len(attempt.floor_reasons)
         and bool(attempt.chunk_indices)
@@ -2469,6 +2525,8 @@ def _validated_candidate_generation_evidence(
     *,
     chunk_indices: tuple[int, ...],
     chunks: tuple[str, ...],
+    expected_candidate_ordinals: tuple[int, ...],
+    require_explicit_candidate_ordinals: bool,
 ) -> CandidateGenerationEvidence:
     """Validate deterministic CFG evidence supplied by the hosted app."""
 
@@ -2483,14 +2541,60 @@ def _validated_candidate_generation_evidence(
         raise ValueError("generation evidence text units do not match the attempt")
     if len(value.effective_cfgs) != len(chunks) or len(value.floor_reasons) != len(chunks):
         raise ValueError("generation evidence CFG rows do not match the attempt")
+    if value.network_conditioned:
+        if (
+            not isinstance(value.network_conditioned, tuple)
+            or len(value.network_conditioned) != len(chunks)
+            or any(type(flag) is not bool for flag in value.network_conditioned)
+        ):
+            raise ValueError(
+                "generation evidence network provenance does not match the attempt"
+            )
+        network_conditioned = value.network_conditioned
+    else:
+        network_conditioned = (False,) * len(chunks)
+    raw_ordinals = value.chunk_candidate_ordinals
+    if not raw_ordinals:
+        if require_explicit_candidate_ordinals:
+            raise ValueError(
+                "context-aware generation evidence must provide chunk candidate ordinals"
+            )
+        candidate_ordinals = expected_candidate_ordinals
+    else:
+        if not isinstance(raw_ordinals, tuple) or len(raw_ordinals) != len(chunks):
+            raise ValueError("generation evidence candidate ordinals do not match the attempt")
+        candidate_ordinals_list: list[int] = []
+        for raw_ordinal in raw_ordinals:
+            if isinstance(raw_ordinal, (bool, np.bool_)):
+                raise ValueError("generation evidence candidate ordinal is invalid")
+            try:
+                ordinal = operator.index(raw_ordinal)
+            except (TypeError, ValueError, OverflowError) as error:
+                raise ValueError(
+                    "generation evidence candidate ordinal is invalid"
+                ) from error
+            if ordinal < 0:
+                raise ValueError("generation evidence candidate ordinal is invalid")
+            candidate_ordinals_list.append(int(ordinal))
+        candidate_ordinals = tuple(candidate_ordinals_list)
+        if candidate_ordinals != expected_candidate_ordinals:
+            raise ValueError(
+                "generation evidence candidate ordinals do not match the schedule"
+            )
+    if len(set(candidate_ordinals)) != 1:
+        raise ValueError("one generation call must use one candidate schedule ordinal")
     scheduled = _finite_float(value.scheduled_cfg, minimum=1.0, maximum=4.0)
     if scheduled is None:
         raise ValueError("generation evidence scheduled CFG is invalid")
+    expected_scheduled = generation_cfg_for_candidate_offset(candidate_ordinals[0])
+    if scheduled != expected_scheduled:
+        raise ValueError("generation evidence scheduled CFG disagrees with the schedule")
     effective_cfgs: list[float] = []
     floor_reasons: list[tuple[str, ...]] = []
-    for effective_value, raw_reasons in zip(
+    for effective_value, raw_reasons, is_network in zip(
         value.effective_cfgs,
         value.floor_reasons,
+        network_conditioned,
         strict=True,
     ):
         effective = _finite_float(effective_value, minimum=1.0, maximum=4.0)
@@ -2507,6 +2611,17 @@ def _validated_candidate_generation_evidence(
             reasons and effective <= scheduled
         ):
             raise ValueError("generation evidence floor reasons disagree with CFG")
+        network_floor_required = bool(
+            is_network and scheduled < MIXED_CFG_NETWORK_MIN
+        )
+        if ("network" in reasons) != network_floor_required:
+            raise ValueError(
+                "generation evidence network floor disagrees with provenance"
+            )
+        if is_network and effective < MIXED_CFG_NETWORK_MIN:
+            raise ValueError(
+                "generation evidence network CFG is below the frozen minimum"
+            )
         effective_cfgs.append(effective)
         floor_reasons.append(reasons)
     return CandidateGenerationEvidence(
@@ -2515,6 +2630,8 @@ def _validated_candidate_generation_evidence(
         scheduled_cfg=scheduled,
         effective_cfgs=tuple(effective_cfgs),
         floor_reasons=tuple(floor_reasons),
+        chunk_candidate_ordinals=candidate_ordinals,
+        network_conditioned=network_conditioned,
     )
 
 
@@ -2547,6 +2664,12 @@ def _candidate_attempt_evidence(
         chunk_text_units=(
             generation.chunk_text_units if generation is not None else ()
         ),
+        chunk_candidate_ordinals=(
+            generation.chunk_candidate_ordinals if generation is not None else ()
+        ),
+        network_conditioned=(
+            generation.network_conditioned if generation is not None else ()
+        ),
         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 ()),
@@ -3093,7 +3216,7 @@ def _coverage_final_passed(verification: Any) -> bool:
 def run_coverage_adaptive_cascade(
     chunks: Sequence[str],
     root_seed: int,
-    candidate_generator: Callable[[tuple[str, ...], int], Any],
+    candidate_generator: Callable[..., Any],
     whole_candidate_verifier: (
         Callable[
             [Any, tuple[str, ...], int],
@@ -3111,10 +3234,7 @@ def run_coverage_adaptive_cascade(
         [CascadeResult, tuple[str, ...]],
         TrajectoryGateResult,
     ],
-    generation_evidence_factory: Callable[
-        [int, int, tuple[int, ...], tuple[str, ...]],
-        CandidateGenerationEvidence,
-    ] | None = None,
+    generation_evidence_factory: Callable[..., CandidateGenerationEvidence] | None = None,
     max_generated_chunks: int = 20,
     max_generated_text_units: int = 800,
     max_sequence_paths: int = 3,
@@ -3127,6 +3247,11 @@ def run_coverage_adaptive_cascade(
     retained.  Later seeds generate exactly one low-coverage chunk under hard
     generated-chunk and generated-text-unit budgets.  Ragged DP paths are
     never returned without the supplied exact whole-waveform verifier.
+
+    Existing two-argument generation callbacks remain unchanged.  A callback
+    that explicitly accepts the keyword-only ``generation_context`` opts into
+    per-chunk refill scheduling.  Its evidence factory must accept the same
+    keyword and report the supplied row-local candidate ordinals.
     """
 
     if not all(
@@ -3143,6 +3268,34 @@ def run_coverage_adaptive_cascade(
         generation_evidence_factory
     ):
         raise ValueError("generation_evidence_factory must be callable")
+
+    def accepts_generation_context(callback: Callable[..., Any]) -> bool:
+        try:
+            parameters = inspect.signature(callback).parameters.values()
+        except (TypeError, ValueError):
+            return False
+        return any(
+            parameter.kind is inspect.Parameter.VAR_KEYWORD
+            or (
+                parameter.name == "generation_context"
+                and parameter.kind
+                in (
+                    inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                    inspect.Parameter.KEYWORD_ONLY,
+                )
+            )
+            for parameter in parameters
+        )
+
+    context_aware_generator = accepts_generation_context(candidate_generator)
+    context_aware_evidence = (
+        generation_evidence_factory is not None
+        and accepts_generation_context(generation_evidence_factory)
+    )
+    if context_aware_generator != context_aware_evidence:
+        raise ValueError(
+            "context-aware generation and evidence callbacks must opt in together"
+        )
     try:
         chunk_tuple = tuple(str(chunk) for chunk in chunks)
     except TypeError as error:
@@ -3211,30 +3364,56 @@ def run_coverage_adaptive_cascade(
     generated_units = initial_units
 
     def generation_evidence(
-        candidate_index: int,
-        seed: int,
-        source_chunk_indices: tuple[int, ...],
+        context: CandidateGenerationContext,
         candidate_chunks: tuple[str, ...],
     ) -> CandidateGenerationEvidence | None:
         if generation_evidence_factory is None:
             return None
         try:
-            raw_evidence = generation_evidence_factory(
-                candidate_index,
-                seed,
-                source_chunk_indices,
-                candidate_chunks,
-            )
+            if context_aware_evidence:
+                raw_evidence = generation_evidence_factory(
+                    context.candidate_index,
+                    context.seed,
+                    context.chunk_indices,
+                    candidate_chunks,
+                    generation_context=context,
+                )
+                expected_ordinals = context.chunk_candidate_ordinals
+            else:
+                raw_evidence = generation_evidence_factory(
+                    context.candidate_index,
+                    context.seed,
+                    context.chunk_indices,
+                    candidate_chunks,
+                )
+                expected_ordinals = (context.candidate_index,) * len(
+                    candidate_chunks
+                )
             return _validated_candidate_generation_evidence(
                 raw_evidence,
-                chunk_indices=source_chunk_indices,
+                chunk_indices=context.chunk_indices,
                 chunks=candidate_chunks,
+                expected_candidate_ordinals=expected_ordinals,
+                require_explicit_candidate_ordinals=context_aware_evidence,
             )
         except Exception as error:
             raise RuntimeError("candidate generation evidence is invalid") from error
 
+    initial_context = CandidateGenerationContext(
+        candidate_index=0,
+        seed=base_seed,
+        chunk_indices=tuple(range(chunk_count)),
+        chunk_candidate_ordinals=(0,) * chunk_count,
+    )
     try:
-        raw_initial_trajectory = candidate_generator(chunk_tuple, base_seed)
+        if context_aware_generator:
+            raw_initial_trajectory = candidate_generator(
+                chunk_tuple,
+                base_seed,
+                generation_context=initial_context,
+            )
+        else:
+            raw_initial_trajectory = candidate_generator(chunk_tuple, base_seed)
     except Exception as error:
         raise RuntimeError("initial trajectory generation failed") from error
     initial_trajectory = _coverage_trajectory_tuple(
@@ -3242,9 +3421,7 @@ def run_coverage_adaptive_cascade(
         chunk_count,
     )
     initial_generation_evidence = generation_evidence(
-        0,
-        base_seed,
-        tuple(range(chunk_count)),
+        initial_context,
         chunk_tuple,
     )
     attempted_seeds.append(base_seed)
@@ -3366,15 +3543,26 @@ def run_coverage_adaptive_cascade(
         )
         seed = base_seed + next_candidate_index
         refill_chunks = (chunk_tuple[chunk_index],)
+        refill_context = CandidateGenerationContext(
+            candidate_index=next_candidate_index,
+            seed=seed,
+            chunk_indices=(chunk_index,),
+            chunk_candidate_ordinals=(refill_attempts[chunk_index] + 1,),
+        )
         try:
-            raw_refill_trajectory = candidate_generator(refill_chunks, seed)
+            if context_aware_generator:
+                raw_refill_trajectory = candidate_generator(
+                    refill_chunks,
+                    seed,
+                    generation_context=refill_context,
+                )
+            else:
+                raw_refill_trajectory = candidate_generator(refill_chunks, seed)
         except Exception as error:
             raise RuntimeError("chunk refill generation failed") from error
         refill_trajectory = _coverage_trajectory_tuple(raw_refill_trajectory, 1)
         refill_generation_evidence = generation_evidence(
-            next_candidate_index,
-            seed,
-            (chunk_index,),
+            refill_context,
             refill_chunks,
         )
         attempted_seeds.append(seed)

tests/test_coverage_adaptive.py

@@ -6,6 +6,7 @@ import pytest
 
 from quality_runtime import (
     CASCADE_EVIDENCE_LOG_PREFIX,
+    CandidateGenerationContext,
     CandidateGenerationEvidence,
     CandidateObservation,
     CandidateVerification,
@@ -14,6 +15,7 @@ from quality_runtime import (
     SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP,
     TrajectoryGateResult,
     format_cascade_evidence_log,
+    generation_cfg_for_candidate_offset,
     run_coverage_adaptive_cascade,
     select_culprit_diverse_candidate_sequences,
     trajectory_gate_evidence,
@@ -180,7 +182,7 @@ def test_refill_budget_accounts_exact_generated_chunks_and_text_units():
         selection=result,
     )
     payload = json.loads(line.removeprefix(CASCADE_EVIDENCE_LOG_PREFIX))
-    assert payload["schema_version"] == 2
+    assert payload["schema_version"] == 3
     assert payload["generation_evidence_complete"] is True
     assert payload["request_chunk_count"] == 2
     assert payload["generated_chunk_count"] == 3
@@ -190,6 +192,7 @@ def test_refill_budget_accounts_exact_generated_chunks_and_text_units():
         "generated_text_units": 6,
     }
     assert payload["attempts"][1]["chunk_indices"] == [0]
+    assert payload["attempts"][1]["chunk_candidate_ordinals"] == [1]
     assert payload["attempts"][1]["scheduled_cfg"] == 2.0
     assert payload["attempts"][1]["effective_cfgs"] == [3.0]
     assert payload["attempts"][1]["floor_reasons"] == [["short_text"]]
@@ -198,7 +201,9 @@ def test_refill_budget_accounts_exact_generated_chunks_and_text_units():
         "chunk_scheduled_cfgs": [2.0, 3.0],
         "chunk_effective_cfgs": [3.0, 3.0],
         "chunk_floor_reasons": [["short_text"], []],
+        "chunk_candidate_ordinals": [1, 0],
         "chunk_policies": ["safe_duration", "base"],
+        "network_conditioned": [False, False],
     }
 
 
@@ -361,6 +366,240 @@ def test_refill_schedule_and_selected_path_are_deterministic():
     assert first.chunk_candidate_counts == second.chunk_candidate_counts == (2, 2, 2)
 
 
+def test_context_aware_refills_rotate_cfg_and_endpoint_policy_per_chunk():
+    chunks = ("第一段", "第二段")
+    contexts = []
+
+    def generator(candidate_chunks, seed, *, generation_context):
+        assert isinstance(generation_context, CandidateGenerationContext)
+        contexts.append(generation_context)
+        return generation_context.chunk_candidate_ordinals
+
+    def generation_evidence(
+        candidate_index,
+        seed,
+        chunk_indices,
+        candidate_chunks,
+        *,
+        generation_context,
+    ):
+        assert candidate_index == generation_context.candidate_index
+        assert seed == generation_context.seed
+        assert chunk_indices == generation_context.chunk_indices
+        ordinal = generation_context.chunk_candidate_ordinals[0]
+        assert all(
+            candidate_ordinal == ordinal
+            for candidate_ordinal in generation_context.chunk_candidate_ordinals
+        )
+        scheduled = generation_cfg_for_candidate_offset(ordinal)
+        return CandidateGenerationEvidence(
+            chunk_indices=chunk_indices,
+            chunk_text_units=tuple(3 for _ in candidate_chunks),
+            scheduled_cfg=scheduled,
+            effective_cfgs=tuple(scheduled for _ in candidate_chunks),
+            floor_reasons=tuple(() for _ in candidate_chunks),
+            chunk_candidate_ordinals=(ordinal,) * len(candidate_chunks),
+        )
+
+    result = run_coverage_adaptive_cascade(
+        chunks,
+        100,
+        generator,
+        lambda trajectory, candidate_chunks, seed: _joined_rejected_local(
+            candidate_chunks,
+            passing=(False, False),
+        ),
+        lambda trajectory, candidate_chunks, seed: _local_verification(
+            candidate_chunks,
+            passing=(trajectory[0] >= 4,),
+        ),
+        sequence_final_verifier=lambda result, candidate_chunks: _exact_final(
+            candidate_chunks
+        ),
+        generation_evidence_factory=generation_evidence,
+        max_generated_chunks=10,
+        max_generated_text_units=100,
+        max_sequence_paths=1,
+    )
+
+    assert [context.candidate_index for context in contexts] == list(range(9))
+    assert [context.seed for context in contexts] == list(range(100, 109))
+    assert [context.chunk_indices for context in contexts] == [
+        (0, 1),
+        (0,),
+        (1,),
+        (0,),
+        (1,),
+        (0,),
+        (1,),
+        (0,),
+        (1,),
+    ]
+    assert [context.chunk_candidate_ordinals for context in contexts] == [
+        (0, 0),
+        (1,),
+        (1,),
+        (2,),
+        (2,),
+        (3,),
+        (3,),
+        (4,),
+        (4,),
+    ]
+    assert result.attempted_seeds == tuple(range(100, 109))
+    assert result.chunk_candidate_indices == (7, 8)
+    assert result.chunk_seeds == (107, 108)
+
+    line = format_cascade_evidence_log(
+        result.diagnostics,
+        outcome="returned",
+        generated_chunk_limit=10,
+        generated_text_unit_limit=100,
+        selection=result,
+    )
+    payload = json.loads(line.removeprefix(CASCADE_EVIDENCE_LOG_PREFIX))
+    assert payload["generation_evidence_complete"] is True
+    assert payload["attempts"][7]["candidate_index"] == 7
+    assert payload["attempts"][7]["seed"] == 107
+    assert payload["attempts"][7]["chunk_candidate_ordinals"] == [4]
+    assert payload["attempts"][7]["scheduled_cfg"] == 3.0
+    assert payload["attempts"][7]["policy"] == "completion_headroom"
+    assert payload["attempts"][7]["chunk_policies"] == [
+        "completion_headroom"
+    ]
+    assert payload["selection"]["generation"]["chunk_candidate_ordinals"] == [
+        4,
+        4,
+    ]
+    assert payload["selection"]["generation"]["chunk_policies"] == [
+        "completion_headroom",
+        "completion_headroom",
+    ]
+
+
+def test_context_generation_and_evidence_must_opt_in_together_and_match():
+    def context_generator(chunks, seed, *, generation_context):
+        return tuple(chunks)
+
+    with pytest.raises(ValueError, match="must opt in together"):
+        run_coverage_adaptive_cascade(
+            ("完整內容",),
+            10,
+            context_generator,
+            lambda trajectory, chunks, seed: _local_verification(chunks),
+            lambda trajectory, chunks, seed: _local_verification(chunks),
+            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
+        )
+
+    def mismatched_evidence(
+        candidate_index,
+        seed,
+        chunk_indices,
+        candidate_chunks,
+        *,
+        generation_context,
+    ):
+        return CandidateGenerationEvidence(
+            chunk_indices=chunk_indices,
+            chunk_text_units=(4,),
+            scheduled_cfg=2.0,
+            effective_cfgs=(2.0,),
+            floor_reasons=((),),
+            chunk_candidate_ordinals=(1,),
+        )
+
+    with pytest.raises(RuntimeError, match="generation evidence is invalid"):
+        run_coverage_adaptive_cascade(
+            ("完整內容",),
+            10,
+            context_generator,
+            lambda trajectory, chunks, seed: _local_verification(chunks),
+            lambda trajectory, chunks, seed: _local_verification(chunks),
+            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
+            generation_evidence_factory=mismatched_evidence,
+        )
+
+
+def test_context_generation_records_and_enforces_network_cfg_provenance():
+    def generator(chunks, seed, *, generation_context):
+        return tuple(chunks)
+
+    def network_evidence(
+        candidate_index,
+        seed,
+        chunk_indices,
+        candidate_chunks,
+        *,
+        generation_context,
+    ):
+        ordinal = generation_context.chunk_candidate_ordinals[0]
+        scheduled = generation_cfg_for_candidate_offset(ordinal)
+        return CandidateGenerationEvidence(
+            chunk_indices=chunk_indices,
+            chunk_text_units=(4,),
+            scheduled_cfg=scheduled,
+            effective_cfgs=(3.0,),
+            floor_reasons=(("network",),) if scheduled < 3.0 else ((),),
+            chunk_candidate_ordinals=(ordinal,),
+            network_conditioned=(True,),
+        )
+
+    result = run_coverage_adaptive_cascade(
+        ("完整內容",),
+        10,
+        generator,
+        lambda trajectory, chunks, seed: _joined_rejected_local(chunks),
+        lambda trajectory, chunks, seed: _local_verification(chunks),
+        sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
+        generation_evidence_factory=network_evidence,
+        max_generated_chunks=2,
+        max_generated_text_units=8,
+        max_sequence_paths=1,
+    )
+
+    assert result.diagnostics.attempts[1].chunk_candidate_ordinals == (1,)
+    assert result.diagnostics.attempts[1].network_conditioned == (True,)
+    assert result.diagnostics.attempts[1].floor_reasons == (("network",),)
+    payload = json.loads(
+        format_cascade_evidence_log(
+            result.diagnostics,
+            outcome="returned",
+            generated_chunk_limit=2,
+            generated_text_unit_limit=8,
+            selection=result,
+        ).removeprefix(CASCADE_EVIDENCE_LOG_PREFIX)
+    )
+    assert payload["attempts"][1]["network_conditioned"] == [True]
+    assert payload["selection"]["generation"]["network_conditioned"] == [True]
+
+    def forged_network_evidence(*args, generation_context, **kwargs):
+        ordinal = generation_context.chunk_candidate_ordinals[0]
+        scheduled = generation_cfg_for_candidate_offset(ordinal)
+        return CandidateGenerationEvidence(
+            chunk_indices=generation_context.chunk_indices,
+            chunk_text_units=(4,),
+            scheduled_cfg=scheduled,
+            effective_cfgs=(3.0,),
+            floor_reasons=((),),
+            chunk_candidate_ordinals=(ordinal,),
+            network_conditioned=(True,),
+        )
+
+    with pytest.raises(RuntimeError, match="generation evidence is invalid"):
+        run_coverage_adaptive_cascade(
+            ("完整內容",),
+            10,
+            generator,
+            lambda trajectory, chunks, seed: _joined_rejected_local(chunks),
+            lambda trajectory, chunks, seed: _local_verification(chunks),
+            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
+            generation_evidence_factory=forged_network_evidence,
+            max_generated_chunks=2,
+            max_generated_text_units=8,
+            max_sequence_paths=1,
+        )
+
+
 def test_zero_coverage_rows_are_refilled_before_low_coverage_rows_round_robin():
     chunks = ("第一段", "第二段", "第三段")
     generated = []

tests/test_production.py

@@ -20,13 +20,16 @@ from production import (
     ensure_terminal_punctuation,
     finish_audio,
     fade_internal_edges,
+    fade_variable_internal_edges,
     join_audio_chunks,
+    join_audio_chunks_variable,
     normalize_asr_spoken_forms,
     mandarin_acoustic_units,
     network_protected_spoken_spans,
     normalize_spoken_forms,
     normalize_tts_eval_text,
     normalize_tts_text,
+    plan_generation_chunks,
     select_candidate_sequence,
     select_generation_cps,
     split_leading_clause,
@@ -927,6 +930,168 @@ def test_frontend_clause_split_matrix_for_structured_and_network_holdouts():
             ) == 4.6
 
 
+def test_generation_only_network_component_planner_holdout_matrix():
+    raw_h11 = (
+        "山區步道的志工預計在 2028/10/21 上午 07:15 集合，"
+        "先用定位器 AX-520 核對座標，再分組檢查木棧道、里程牌與飲水站。"
+        "若氣象網站 https://trailweather.example.tw 顯示降雨機率超過 65%，"
+        "領隊就取消高海拔路線，改走較短的林間環線。"
+        "途中若發現落石或樹枝阻斷通行，請拍照並寄到 "
+        "patrol@forestmail.tw，不要自行搬動大型障礙物。"
+        "所有隊員回到登山口後，還要清點無線電與急救包，"
+        "確認沒有任何人落單，才結束當天的巡查。"
+    )
+    matrix = {
+        "H06": (
+            "若要更換導覽場次，請寄信到 tour.help@islandmuseum.tw。",
+            (24, 25),
+        ),
+        "H07": (
+            "潮汐預報可查詢 https://coastwatch.example.tw/tide。",
+            (34, 23),
+        ),
+        "H11": (raw_h11, (30, 29, 18, 35, 33, 28, 34, 38)),
+    }
+
+    for raw, expected_units in matrix.values():
+        normalized = normalize_spoken_forms(raw)
+        specs = plan_generation_chunks(raw, normalized)
+
+        assert tuple(count_speech_units(spec.text) for spec in specs) == expected_units
+        assert normalize_tts_text(" ".join(spec.text for spec in specs)) == normalized
+        assert all(
+            spec.source_end == following.source_start
+            for spec, following in zip(specs, specs[1:])
+        )
+        assert all(
+            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)
+        for spec in specs:
+            assert len(spec.network_span_indices) == len(
+                spec.network_full_spoken_proofs
+            )
+
+
+def test_generation_network_planner_keeps_ordinary_semantic_chunks_unchanged():
+    raw = (
+        "清晨開館以前，水族館人員會先量測各池的水溫與鹽度，"
+        "再觀察魚群是否正常進食。確認設備都正常後才開門。"
+    )
+    normalized = normalize_spoken_forms(raw)
+
+    specs = plan_generation_chunks(raw, normalized)
+
+    assert [spec.text for spec in specs] == split_text_for_tts(
+        normalized,
+        max_chars=80,
+        min_chunk_chars=12,
+    )
+    assert all(not spec.network_conditioned for spec in specs)
+
+
+def test_generation_network_planner_fails_closed_on_invalid_provenance():
+    raw = "請寄到 tour.help@islandmuseum.tw。"
+    different = normalize_spoken_forms(
+        "請寄到 tour.help@differentmuseum.tw。"
+    )
+    with pytest.raises(ValueError, match="does not match"):
+        plan_generation_chunks(raw, different)
+
+    iri = "請查詢 https://example.tw/愛。"
+    with pytest.raises(ValueError, match="non-ASCII IRI"):
+        plan_generation_chunks(iri, normalize_spoken_forms(iri))
+
+    emoji_iri = "請查詢 https://example.tw/🐦。"
+    with pytest.raises(ValueError, match="non-ASCII IRI"):
+        plan_generation_chunks(emoji_iri, normalize_spoken_forms(emoji_iri))
+
+    oversized = "請查詢 https://" + "a" * 50 + ".tw。"
+    with pytest.raises(ValueError, match="indivisible network component"):
+        plan_generation_chunks(oversized, normalize_spoken_forms(oversized))
+
+
+def test_generation_network_planner_binds_duplicate_spoken_proof_to_raw_identifier():
+    spoken = (
+        "艾取 踢 踢 批 艾斯 冒號 斜線 斜線 "
+        "伊 艾克斯 欸 艾姆 批 艾爾 伊 點 踢 達不溜"
+    )
+    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 all(
+        normalized[spec.source_start : spec.source_end] == spec.text
+        for spec in specs
+    )
+
+
+def test_generation_network_planner_is_range_first_for_english_and_long_prose():
+    cases = (
+        "Please visit https://example.tw。",
+        "甲" * 35 + "請查 https://c.tw。",
+    )
+    for raw in cases:
+        normalized = normalize_spoken_forms(raw)
+        specs = plan_generation_chunks(raw, normalized)
+
+        assert "".join(spec.text for spec in specs) == normalized
+        assert all(
+            count_speech_units(spec.text)
+            <= (36 if spec.network_conditioned else 80)
+            for spec in specs
+        )
+    prose_specs = plan_generation_chunks(cases[1], normalize_spoken_forms(cases[1]))
+    assert any(
+        not spec.network_conditioned and count_speech_units(spec.text) > 36
+        for spec in prose_specs
+    )
+
+
+def test_generation_network_planner_splits_long_and_repeated_identifiers_by_proof():
+    cases = (
+        "請查 https://example.tw/a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p。",
+        "https://a.tw https://a.tw",
+    )
+    for raw in cases:
+        normalized = normalize_spoken_forms(raw)
+        specs = plan_generation_chunks(raw, normalized)
+
+        assert "".join(spec.text for spec in specs) == normalized
+        assert all(count_speech_units(spec.text) <= 36 for spec in specs)
+        assert all(spec.network_conditioned for spec in specs)
+    repeated = plan_generation_chunks(cases[1], normalize_spoken_forms(cases[1]))
+    assert {index for spec in repeated for index in spec.network_span_indices} == {0, 1}
+
+
+@pytest.mark.parametrize(
+    "raw",
+    (
+        "a@b.co https://c.de",
+        "https://a.co b@c.de",
+        "a@b.co b@c.de",
+        "https://a.co https://b.de",
+    ),
+)
+def test_generation_network_planner_maps_whitespace_adjacent_identifiers(raw):
+    normalized = normalize_spoken_forms(raw)
+
+    specs = plan_generation_chunks(raw, normalized)
+
+    assert "".join(spec.text for spec in specs) == normalized
+    assert {index for spec in specs for index in spec.network_span_indices} == {
+        0,
+        1,
+    }
+    assert all(spec.network_conditioned for spec in specs)
+
+
 def test_split_preserves_long_strong_sentences_and_is_deterministic():
     sentences = (
         "甲" * 59 + "。",
@@ -987,6 +1152,60 @@ def test_onset_split_requires_a_natural_boundary():
     ]
 
 
+def test_variable_network_boundary_uses_short_fade_without_losing_chunks():
+    sample_rate = 1_000
+    chunks = [np.ones(1_000, dtype=np.float32) for _ in range(3)]
+
+    faded = fade_variable_internal_edges(
+        chunks,
+        sample_rate,
+        fade_ms_by_boundary=[5.0, 80.0],
+    )
+    joined = join_audio_chunks_variable(
+        faded,
+        pauses=[0, 10],
+        crossfade_samples_by_boundary=[5, 80],
+        pre_faded_edges=True,
+    )
+
+    assert joined.shape == (3_005,)
+    assert np.isfinite(joined).all()
+    assert np.max(np.abs(joined)) <= 1.0
+    assert np.count_nonzero(faded[0]) >= 995
+    assert np.count_nonzero(faded[1]) >= 915
+    with pytest.raises(ValueError, match="boundary count"):
+        fade_variable_internal_edges(chunks, sample_rate, [5.0])
+    with pytest.raises(ValueError, match="boundary counts"):
+        join_audio_chunks_variable(chunks, [0, 0], [5])
+
+
+def test_variable_audio_boundaries_reject_empty_overlap_and_invalid_widths():
+    sample_rate = 1_000
+    empty_middle = [
+        np.ones(10, dtype=np.float32),
+        np.zeros(0, dtype=np.float32),
+        np.ones(10, dtype=np.float32),
+    ]
+    with pytest.raises(ValueError, match="non-empty"):
+        fade_variable_internal_edges(empty_middle, sample_rate, [5.0, 5.0])
+    with pytest.raises(ValueError, match="non-empty"):
+        join_audio_chunks_variable(empty_middle, [0, 0], [5, 5])
+
+    short_middle = [
+        np.ones(10, dtype=np.float32),
+        np.ones(6, dtype=np.float32),
+        np.ones(10, dtype=np.float32),
+    ]
+    with pytest.raises(ValueError, match="overlap"):
+        fade_variable_internal_edges(short_middle, sample_rate, [4.0, 4.0])
+    with pytest.raises(ValueError, match="overlap"):
+        join_audio_chunks_variable(short_middle, [0, 0], [4, 4])
+    with pytest.raises(ValueError, match="non-negative integers"):
+        join_audio_chunks_variable(short_middle, [-1, 0], [0, 0])
+    with pytest.raises(ValueError, match="non-negative integers"):
+        join_audio_chunks_variable(short_middle, [0, 0], [1.5, 0])
+
+
 def test_prefaded_join_does_not_apply_a_second_edge_ramp():
     sample_rate = 1_000
     chunks = fade_internal_edges(

tests/test_release_pins.py

@@ -237,11 +237,13 @@ def test_readme_describes_coverage_refill_and_sequence_transition_scores():
     assert "1→5→10→15→20" not in app_source
 
 
-def test_app_wires_candidate_offset_to_explicit_generation_policy_and_logs_it():
+def test_app_wires_row_local_candidate_ordinal_to_generation_policy_and_logs_it():
     source = (ROOT / "app.py").read_text(encoding="utf-8")
 
     assert source.count("policy: GenerationPolicy") == 2
-    assert "policy=generation_policy_for_candidate_offset(seed - request_seed)" in source
+    assert "generation_context.chunk_candidate_ordinals" in source
+    assert "policy=generation_policy_for_candidate_offset(candidate_ordinal)" in source
+    assert "generation_context.seed != seed" in source
     assert 'f"name={policy.name}' in source
     assert "chunk_policies={selected_policies}" in source
     assert '"min_len": min_len' in source
@@ -255,16 +257,17 @@ def test_app_wires_candidate_offset_to_explicit_generation_policy_and_logs_it():
 def test_app_rejects_ambiguous_iri_before_frontend_normalization():
     source = (ROOT / "app.py").read_text(encoding="utf-8")
     synthesize_start = source.index("def _synthesize(")
+    raw_text = source.index("raw_text = str(text)", synthesize_start)
     iri_guard = source.index(
-        "if network_identifier_has_ambiguous_iri(text):",
+        "if network_identifier_has_ambiguous_iri(raw_text):",
         synthesize_start,
     )
     normalization = source.index(
-        'text = normalize_spoken_forms(text, locale="zh-TW")',
+        'text = normalize_spoken_forms(raw_text, locale="zh-TW")',
         synthesize_start,
     )
 
-    assert synthesize_start < iri_guard < normalization
+    assert synthesize_start < raw_text < iri_guard < normalization
     assert "非 ASCII IRI 必須先轉成 ASCII/percent-encoded" in (
         ROOT / "README.md"
     ).read_text(encoding="utf-8")
@@ -302,13 +305,17 @@ def test_app_applies_fixed_mixed_cfg_schedule_after_global_quality_floor():
     assert "MIXED_CFG_PRIMARY = 3.0" in source
     assert "MIXED_CFG_ALTERNATE = 2.0" in source
     assert (
-        'MIXED_CFG_SCHEDULE = "offset_zero_and_even_primary_odd_alternate"'
+        'MIXED_CFG_SCHEDULE = "row_ordinal_zero_and_even_primary_odd_alternate"'
         in source
     )
-    assert "network_request = contains_network_identifier(text)" in synthesize_source
+    assert "network_request = contains_network_identifier(raw_text)" in synthesize_source
     assert "cfg_value != MIXED_CFG_PRIMARY" in synthesize_source
     assert "request_cfg = MIXED_CFG_PRIMARY" in synthesize_source
-    assert "cfg=candidate_cfg(seed)" in synthesize_source
+    assert "cfg=candidate_cfg(candidate_ordinal)" in synthesize_source
+    assert "candidate_ordinals = generation_context.chunk_candidate_ordinals" in (
+        synthesize_source
+    )
+    assert "chunk_candidate_ordinals=candidate_ordinals" in synthesize_source
     assert "generation_cfg_for_candidate_offset(" in synthesize_source
     assert "chunk_cfgs={selected_cfgs}" in synthesize_source
     assert "attempted_schedule_cfgs={attempted_schedule_cfgs}" in synthesize_source
@@ -360,9 +367,20 @@ def test_app_rejects_silent_text_and_coalesces_before_runtime_budgeting():
     assert synthesize_source is not None
     assert assemble_source is not None
     assert "if count_speech_units(text) <= 0" in synthesize_source
-    assert "chunks = coalesce_text_chunks(" in synthesize_source
+    assert "coalesce_text_chunks(" in synthesize_source
+    assert "chunk_specs = plan_generation_chunks(" in synthesize_source
+    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_TARGET_UNITS = 32" in source
+    assert "NETWORK_GENERATION_MAX_UNITS = 36" in source
+    assert "NETWORK_INTERNAL_FADE_MS = 5.0" in source
+    assert 'chunk_specs[index].boundary_after == "network_internal"' in (
+        assemble_source
+    )
+    assert "join_audio_chunks_variable(" in assemble_source
+    assert "network_conditioned=network_flags" in synthesize_source
+    assert "network_conditioned=network_flag" in synthesize_source
 
 
 def test_app_emits_one_canonical_content_free_evidence_line_per_terminal_outcome():
@@ -468,7 +486,7 @@ def test_app_reverifies_the_post_join_speed_adjusted_whole_waveform():
     production_source = (ROOT / "production.py").read_text(encoding="utf-8")
     assert "trailing_silence_ms: float = 180.0" in production_source
     assemble_index = synthesize_source.index(
-        "waveform = _assemble_trajectory_audio(cascade.trajectory, chunks, speed)"
+        "waveform = _assemble_trajectory_audio("
     )
     verify_index = synthesize_source.index("final_verification = _verify_trajectory_audio(")
     require_index = synthesize_source.index(
@@ -549,10 +567,9 @@ def test_whole_candidate_qualification_uses_the_exact_return_assembler_after_loc
     assert "QUALITY_FINAL_ASR_MAX_NEW_TOKENS" in qualify_source[joined_index:]
     assert "release_speaker_gate=True" in qualify_source[joined_index:]
     assert "qualify_trajectory_with_joined_output(" in qualify_source[joined_index:]
-    assert (
-        "waveform = _assemble_trajectory_audio(cascade.trajectory, chunks, speed)"
-        in synthesize_source
-    )
+    assert "waveform = _assemble_trajectory_audio(" in synthesize_source
+    assert "cascade.trajectory" in synthesize_source
+    assert "chunk_specs" in synthesize_source
     assert "require_verified_final_output(final_verification)" in synthesize_source
 
 
@@ -635,7 +652,7 @@ def test_space_hard_intersects_dual_asr_only_on_exact_whole_waveforms():
         "cascade = run_coverage_adaptive_cascade("
     )
     final_assemble = synthesize_source.index(
-        "waveform = _assemble_trajectory_audio(cascade.trajectory, chunks, speed)"
+        "waveform = _assemble_trajectory_audio("
     )
     final_turbo = synthesize_source.index("final_verification = _verify_trajectory_audio(")
     final_turbo_require = synthesize_source.index(