Harden endpoint audio delivery

README.md

@@ -92,6 +92,8 @@ the runtime additionally verifies the exact weight SHA-256 before deserializatio
 每第四個 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。
+唯一例外是 completion-headroom 的一至兩單位短句：其 sparse retry 將 hard cap 下限提高到 5 steps，
+讓第一個 eligible weak-stop 有一次真正的 threshold 判定機會；所有額外內容仍須通過雙 ASR 與 tail=0。
 選中完整 trajectory 或 coverage-sequence-DP 混合路徑後，log 會記錄每個 chunk 的 candidate、seed、policy、實際 CFG，
 並記錄每個已嘗試 row-local ordinal 的 schedule CFG。Ordinal 0 與正偶數使用 primary CFG 3.0，
 正奇數使用 alternate CFG 2.0；含 URL/email component 的 chunk 仍有 CFG 3.0 的內容安全下限。
@@ -103,8 +105,9 @@ speech-unit counter；只有 URL/email-conditioned native endpoint estimate 對
 採用除以 2 的保守 counter。這個 counter 刻意同時決定 adaptive weak-stop 的
 `expected_steps` progress 與 `max_len` hard cap；兩者屬於同一份 native endpoint plan，
 但不會進入播放 pace、chunk planning 或 32/800 work accounting，`min_len` 仍固定為 2。
-終端 outcome 另輸出 content-free canonical evidence schema v5，包含 original chunk indices、
+終端 outcome 另輸出 content-free canonical evidence schema v6，包含 original chunk indices、
 row-local ordinal、network provenance、CFG contract、32/800 預算用量與 selected-path 交叉檢查，
+每個 chunk 的 stop reason、pre-fade endpoint energy、generated/hard-stop steps，
 每個 local row 的 independent-large-v3 attempted/pass/proof-count/result attestation，以及
 local、joined、independent-large-v3 與 final gate 的 bounded SQUIM scalar，
 不包含 target/transcript/audio/embedding。
@@ -115,8 +118,15 @@ local、joined、independent-large-v3 與 final gate 的 bounded SQUIM scalar，
 不會執行 refill 或 DP。若失敗，之後只做單 chunk refill，不再生成第二條完整 trajectory。
 Ragged DP 可混合已通過 hard gate 的 chunks，
 以及下述唯一 boundary-only、drop 不超過 0.15 的 DP-only chunks。切段保留標點，並依逗號、分號或句末標點
-插入不同長度的停頓。輸出最後會套用保守的 RMS floor 與 peak limit，通過所有 verifier
-後才明確量化為 PCM16；不讓 Gradio 對 float ndarray 再做 peak normalization，以免額外放大底噪。
+插入不同長度的停頓。已通過 hard gate 的候選再加入 bounded endpoint soft cost：
+任何到達 hard cap（即使 threshold 恰於同一步 crossing）皆加 0.05，再加上 `0.02 ×`
+最後 5 ms RMS/全波形 peak。只有在目前 preferred 候選到達 hard cap 時，才允許真正於 cap
+前自然停止的候選使用最多 STOI 0.02／PESQ 0.03 的 preferred-tier slack；speaker 與 boundary
+在既有長度 gate 適用時仍須完整達到 preferred 門檻，所有 hard gate 亦維持不變。
+輸出最後會套用保守的 RMS floor 與 peak limit、固定 5 ms cosine-squared 收尾與 180 ms silence，
+接著先量化到 PCM16 並以 WAV decoder 等價波形執行 joined/final verifier。公開 wrapper 再編碼時
+逐 sample 冪等，因此驗證的就是使用者收到的 PCM codes；同時不讓 Gradio 對 float ndarray
+再做 peak normalization，以免額外放大底噪。
 完成 RMS matching、edge fade、pause、crossfade 與使用者 speed 後，服務會再對最終整段 waveform
 執行 normalized target 的 prefix/whole/suffix/tail、pace 與 speaker anchor/boundary gate，並要求
 turbo 與 full large-v3 的語意 hard intersection；這個 post-join gate 失敗時不會回傳先前已通過的

app.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from dataclasses import replace
+from dataclasses import dataclass, replace
 from importlib import metadata as importlib_metadata
 import inspect
 import json
@@ -103,6 +103,7 @@ from production import (
     normalize_spoken_forms,
     punctuation_pause_seconds,
     pcm16_audio_output,
+    pcm16_verification_waveform,
     plan_generation_chunks,
     select_generation_cps,
     set_generation_seed,
@@ -133,6 +134,7 @@ from quality_runtime import (
     candidate_gate_evidence,
     active_audio_rms_db,
     active_audio_median_f0_hz,
+    endpoint_tail_energy_ratio,
     format_cascade_evidence_log,
     generation_cfg_for_candidate_offset,
     generation_policy_for_candidate_offset,
@@ -191,6 +193,7 @@ QUALITY_MEDIUM_CHUNK_UNITS = 24
 CROSSFADE_MS = 80.0
 CHUNK_EDGE_FADE_MS = 80.0
 CHUNK_RMS_MATCH_DB = 4.0
+FINAL_ENDPOINT_FADE_MS = 5.0
 NETWORK_GENERATION_MIN_UNITS = 8
 NETWORK_GENERATION_TARGET_UNITS = 32
 NETWORK_GENERATION_MAX_UNITS = 36
@@ -375,6 +378,15 @@ def _apply_speed(
     )
 
 
+@dataclass(frozen=True)
+class _ChunkGenerationOutcome:
+    audio: np.ndarray
+    stop_reason: str
+    endpoint_energy_ratio: float
+    generated_steps: int
+    hard_stop_steps: int
+
+
 def _generate_chunk(
     text: str,
     centroid: torch.Tensor,
@@ -384,7 +396,7 @@ def _generate_chunk(
     request_seed: int,
     policy: GenerationPolicy,
     network_conditioned: bool = False,
-) -> np.ndarray:
+) -> _ChunkGenerationOutcome:
     generation_cps = (
         policy.ascii_cps
         if network_conditioned
@@ -407,6 +419,19 @@ def _generate_chunk(
         add_terminal_punctuation=count_speech_units(text) >= MIN_ENDPOINT_CUE_UNITS,
         duration_units=endpoint_duration_units,
     )
+    public_text_units = count_speech_units(text)
+    short_headroom_max_units = int(
+        getattr(policy, "short_headroom_max_units", 0)
+    )
+    short_hard_stop_floor_steps = int(
+        getattr(policy, "short_hard_stop_floor_steps", 0)
+    )
+    short_headroom_floor_applied = bool(
+        0 < public_text_units <= short_headroom_max_units
+        and hard_stop_steps < short_hard_stop_floor_steps
+    )
+    if short_headroom_floor_applied:
+        hard_stop_steps = short_hard_stop_floor_steps
     # Do not hold generation open to enforce pace. The model can finish the
     # requested text early; extending its latent sequence creates tail speech.
     min_len = 2
@@ -445,6 +470,16 @@ def _generate_chunk(
     finally:
         if _STOP_CONTROLLER is not None:
             _STOP_CONTROLLER.end()
+    stop_reason = (
+        _STOP_CONTROLLER.last_stop_reason
+        if _STOP_CONTROLLER is not None
+        else "native_stop"
+    )
+    generated_steps = (
+        _STOP_CONTROLLER.last_generated_steps
+        if _STOP_CONTROLLER is not None
+        else hard_stop_steps
+    )
     if _STOP_CONTROLLER is not None:
         print(
             "[BlueMagpie] endpoint "
@@ -458,9 +493,21 @@ def _generate_chunk(
         f"duration_units={endpoint_duration_units} "
         f"duration_counter={'network_conservative' if network_conditioned else 'public'} "
         f"expected_steps={expected_steps} hard_stop_steps={hard_stop_steps} "
-        f"min_len={min_len}"
+        f"min_len={min_len} "
+        f"short_headroom_floor_applied={short_headroom_floor_applied}"
     )
     audio = audio.detach().float().cpu().numpy().reshape(-1)
+
+    def outcome(waveform: np.ndarray) -> _ChunkGenerationOutcome:
+        normalized = np.asarray(waveform, dtype=np.float32).reshape(-1)
+        return _ChunkGenerationOutcome(
+            audio=normalized,
+            stop_reason=stop_reason,
+            endpoint_energy_ratio=endpoint_tail_energy_ratio(normalized, SR),
+            generated_steps=generated_steps,
+            hard_stop_steps=hard_stop_steps,
+        )
+
     pace_speed = target_pace_speed(
         audio.size,
         SR,
@@ -498,7 +545,7 @@ def _generate_chunk(
     try:
         corrected_active_duration = active_voiced_duration_seconds(corrected, SR)
     except (TypeError, ValueError, RuntimeError, OverflowError):
-        return corrected
+        return outcome(corrected)
     rerender_speed = active_pace_correction_speed(
         corrected_active_duration,
         text,
@@ -535,7 +582,7 @@ def _generate_chunk(
         try:
             final_active_duration = active_voiced_duration_seconds(final_audio, SR)
         except (TypeError, ValueError, RuntimeError, OverflowError):
-            return final_audio
+            return outcome(final_audio)
         observed_cps = count_speech_units(text) / final_active_duration
         if observed_cps > QUALITY_MAX_PACE_CPS:
             fallback_residual = active_pace_correction_speed(
@@ -555,12 +602,14 @@ def _generate_chunk(
                     f"residual_rate={fallback_residual:.6f} "
                     f"total_rate={fallback_speed:.6f}"
                 )
-                return _apply_speed(
-                    audio,
-                    fallback_speed,
-                    network_conditioned=False,
+                return outcome(
+                    _apply_speed(
+                        audio,
+                        fallback_speed,
+                        network_conditioned=False,
+                    )
                 )
-    return final_audio
+    return outcome(final_audio)
 
 
 def _speaker_anchor_array(centroid: torch.Tensor) -> np.ndarray:
@@ -583,9 +632,10 @@ def _generate_trajectory(
     policy: GenerationPolicy,
     network_cfg_min: float = NETWORK_TEXT_CFG_MIN,
     network_conditioned: tuple[bool, ...] | None = None,
-) -> tuple[np.ndarray, ...]:
+) -> tuple[tuple[np.ndarray, ...], tuple[_ChunkGenerationOutcome, ...]]:
     scheduled_cfg = float(cfg)
     trajectory: list[np.ndarray] = []
+    endpoint_outcomes: list[_ChunkGenerationOutcome] = []
     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):
@@ -617,18 +667,18 @@ def _generate_trajectory(
             f"network_floor_applied={network_floor_applied} "
             f"short_floor_applied={short_floor_applied}"
         )
-        trajectory.append(
-            _generate_chunk(
-                chunk,
-                centroid,
-                cfg=network_adjusted_cfg,
-                steps=steps,
-                request_seed=request_seed,
-                policy=policy,
-                network_conditioned=network_chunk,
-            )
+        outcome = _generate_chunk(
+            chunk,
+            centroid,
+            cfg=network_adjusted_cfg,
+            steps=steps,
+            request_seed=request_seed,
+            policy=policy,
+            network_conditioned=network_chunk,
         )
-    return tuple(trajectory)
+        trajectory.append(outcome.audio)
+        endpoint_outcomes.append(outcome)
+    return tuple(trajectory), tuple(endpoint_outcomes)
 
 
 def _attach_transition_f0(
@@ -1099,8 +1149,14 @@ def _assemble_trajectory_audio(
         max_gain=3.0,
     )
     waveform = _apply_speed(waveform, playback_speed)
-    finish_fade_ms = 0.0 if count_speech_units("".join(chunks)) <= 6 else 5.0
-    return finish_audio(waveform, SR, fade_ms=finish_fade_ms)
+    waveform = finish_audio(
+        waveform,
+        SR,
+        fade_ms=FINAL_ENDPOINT_FADE_MS,
+    )
+    # Joined/final ASR, SQUIM, and speaker gates must inspect the exact
+    # decoder-equivalent waveform represented by the public PCM16 response.
+    return pcm16_verification_waveform(waveform)
 
 
 def _verification_metric_log_fields(verification) -> str:
@@ -1504,6 +1560,10 @@ def _synthesize(
     anchor = _speaker_anchor_array(centroid)
     independent_cache = WholeWaveformVerificationCache()
     generation_context_by_seed: dict[int, CandidateGenerationContext] = {}
+    generation_endpoint_evidence_by_seed: dict[
+        int,
+        tuple[_ChunkGenerationOutcome, ...],
+    ] = {}
 
     def canonical_chunks_for_context(
         generation_context: CandidateGenerationContext,
@@ -1714,7 +1774,7 @@ def _synthesize(
             candidate_chunks,
             generation_context=generation_context,
         )
-        return _generate_trajectory(
+        trajectory, endpoint_outcomes = _generate_trajectory(
             candidate_chunks,
             centroid,
             cfg=candidate_cfg(candidate_ordinal),
@@ -1723,6 +1783,10 @@ def _synthesize(
             policy=generation_policy_for_candidate_offset(candidate_ordinal),
             network_conditioned=network_flags,
         )
+        if len(endpoint_outcomes) != len(candidate_chunks):
+            raise ValueError("generation endpoint evidence is incomplete")
+        generation_endpoint_evidence_by_seed[seed] = endpoint_outcomes
+        return trajectory
 
     def candidate_generation_evidence(
         candidate_index: int,
@@ -1773,6 +1837,11 @@ def _synthesize(
                 strict=True,
             )
         )
+        endpoint_outcomes = generation_endpoint_evidence_by_seed.get(seed)
+        if endpoint_outcomes is None or len(endpoint_outcomes) != len(
+            candidate_chunks
+        ):
+            raise ValueError("generation endpoint evidence is missing")
         return CandidateGenerationEvidence(
             chunk_indices=chunk_indices,
             chunk_text_units=tuple(
@@ -1784,6 +1853,18 @@ def _synthesize(
             chunk_candidate_ordinals=candidate_ordinals,
             network_conditioned=network_flags,
             chunk_text_variants=text_variants,
+            chunk_stop_reasons=tuple(
+                outcome.stop_reason for outcome in endpoint_outcomes
+            ),
+            chunk_endpoint_energy_ratios=tuple(
+                outcome.endpoint_energy_ratio for outcome in endpoint_outcomes
+            ),
+            chunk_generated_steps=tuple(
+                outcome.generated_steps for outcome in endpoint_outcomes
+            ),
+            chunk_hard_stop_steps=tuple(
+                outcome.hard_stop_steps for outcome in endpoint_outcomes
+            ),
         )
 
     try:
@@ -1855,6 +1936,7 @@ def _synthesize(
                 preferred_min_squim_audio_duration_seconds=(
                     QUALITY_PREFERRED_SQUIM_MIN_DURATION_SECONDS
                 ),
+                require_endpoint_evidence=True,
             )
     except NoQualifiedCandidateError as error:
         print(

production.py

@@ -419,8 +419,9 @@ class StopHysteresisController(nn.Module):
         should_stop = eligible and self._hits >= self.consecutive
         stop_reason = "stop_threshold" if should_stop else "running"
         if self._hard_stop_steps > 0 and generated_steps >= self._hard_stop_steps:
-            should_stop = True
-            stop_reason = "hard_stop"
+            if not should_stop:
+                should_stop = True
+                stop_reason = "hard_stop"
         self._step += 1
         self.last_generated_steps = generated_steps
         self.last_stop_reason = stop_reason
@@ -3116,7 +3117,15 @@ def finish_audio(
         max(0, int(round(float(fade_ms) * int(sample_rate) / 1000.0))),
     )
     if fade_samples > 0:
-        signal[-fade_samples:] *= np.linspace(1.0, 0.0, fade_samples, dtype=np.float32)
+        phase = np.linspace(
+            0.0,
+            np.pi / 2.0,
+            fade_samples,
+            dtype=np.float32,
+        )
+        envelope = np.square(np.cos(phase))
+        envelope[-1] = 0.0
+        signal[-fade_samples:] *= envelope
     silence_samples = max(
         0,
         int(round(float(trailing_silence_ms) * int(sample_rate) / 1000.0)),
@@ -4602,17 +4611,51 @@ def _peak_rms(audio: np.ndarray) -> tuple[float, float]:
     return peak, rms
 
 
+def encode_pcm16(audio: np.ndarray) -> np.ndarray:
+    """Encode finite mono audio with the standard signed PCM16 scale."""
+
+    values = np.asarray(audio)
+    if values.ndim != 1 or values.size <= 0 or not np.issubdtype(
+        values.dtype,
+        np.floating,
+    ):
+        raise ValueError("audio must be a non-empty mono floating-point array")
+    finite = values.astype(np.float64, copy=False)
+    if not np.isfinite(finite).all():
+        raise ValueError("audio must contain only finite samples")
+    if float(np.max(np.abs(finite))) > 1.0:
+        raise ValueError("audio samples must be bounded to [-1, 1]")
+
+    scaled = np.rint(finite * 32768.0)
+    return np.clip(scaled, -32768.0, 32767.0).astype(np.int16)
+
+
+def decode_pcm16(pcm: np.ndarray) -> np.ndarray:
+    """Decode one mono PCM16 buffer exactly as a WAV reader does."""
+
+    values = np.asarray(pcm)
+    if values.ndim != 1 or values.size <= 0 or values.dtype != np.int16:
+        raise ValueError("pcm must be a non-empty mono int16 array")
+    return values.astype(np.float32) / np.float32(32768.0)
+
+
+def pcm16_verification_waveform(audio: np.ndarray) -> np.ndarray:
+    """Return the decoder-equivalent waveform for the exact public PCM codes."""
+
+    return decode_pcm16(encode_pcm16(audio))
+
+
 def pcm16_audio_output(
     sample_rate: int,
     audio: np.ndarray,
 ) -> tuple[int, np.ndarray]:
-    """Serialize finite mono float audio as PCM16 without peak normalization.
+    """Serialize verified mono float audio as PCM16 without peak normalization.
 
     Gradio peak-normalizes floating-point ndarray outputs before writing them.
     Returning explicit PCM16 preserves the level already selected by the
     production loudness/peak limiter, including its intended noise headroom.
-    This helper is deliberately a serialization boundary: quality evaluation
-    and speaker verification continue to consume the original float waveform.
+    Callers should verify ``pcm16_verification_waveform(audio)``; encoding that
+    canonical waveform here is sample-for-sample idempotent.
     """
 
     if isinstance(sample_rate, (bool, np.bool_)):
@@ -4623,23 +4666,7 @@ def pcm16_audio_output(
         raise ValueError("sample_rate must be a positive integer") from error
     if rate_hz <= 0:
         raise ValueError("sample_rate must be a positive integer")
-
-    values = np.asarray(audio)
-    if values.ndim != 1 or values.size <= 0 or not np.issubdtype(
-        values.dtype,
-        np.floating,
-    ):
-        raise ValueError("audio must be a non-empty mono floating-point array")
-    finite = values.astype(np.float64, copy=False)
-    if not np.isfinite(finite).all():
-        raise ValueError("audio must contain only finite samples")
-    if float(np.max(np.abs(finite))) > 1.0:
-        raise ValueError("audio samples must be bounded to [-1, 1]")
-
-    # 32767 is intentional: it gives a symmetric mapping and, unlike Gradio's
-    # float conversion, never divides by the waveform peak.
-    pcm = np.rint(finite * 32767.0).astype(np.int16)
-    return int(rate_hz), pcm
+    return int(rate_hz), encode_pcm16(audio)
 
 
 def match_chunk_rms(reference: np.ndarray, chunk: np.ndarray, max_adjust_db: float = 4.0) -> np.ndarray:

quality_runtime.py

@@ -61,7 +61,12 @@ 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 = 5
+ENDPOINT_TAIL_WINDOW_MS = 5.0
+ENDPOINT_HARD_STOP_PENALTY = 0.05
+ENDPOINT_ENERGY_WEIGHT = 0.02
+ENDPOINT_PREFERRED_SQUIM_STOI_SLACK = 0.02
+ENDPOINT_PREFERRED_SQUIM_PESQ_SLACK = 0.03
+CASCADE_EVIDENCE_SCHEMA_VERSION = 6
 CASCADE_EVIDENCE_LOG_PREFIX = "[BlueMagpie] cascade evidence "
 CASCADE_EVIDENCE_MAX_ATTEMPTS = ADAPTIVE_CASCADE_STAGE_LIMITS[-1]
 CASCADE_EVIDENCE_MAX_LOCAL_RESULTS = ADAPTIVE_CASCADE_STAGE_LIMITS[-1]
@@ -76,6 +81,9 @@ _CASCADE_EVIDENCE_SELECTION_MODES = frozenset(
     {"whole_trajectory", "sequence_dp", "coverage_sequence_dp"}
 )
 _CFG_FLOOR_REASONS = frozenset({"network", "short_text"})
+_GENERATION_STOP_REASONS = frozenset(
+    {"hard_stop", "native_stop", "stop_threshold"}
+)
 _CASCADE_EVIDENCE_REJECTION_CODES = frozenset(
     {
         "boundary_speaker_drop",
@@ -115,6 +123,8 @@ class GenerationPolicy:
     cjk_cps: float
     ascii_cps: float
     hard_stop_margin_steps: int
+    short_headroom_max_units: int = 0
+    short_hard_stop_floor_steps: int = 0
 
 
 @dataclass(frozen=True)
@@ -151,6 +161,8 @@ COMPLETION_HEADROOM_GENERATION_POLICY = GenerationPolicy(
     cjk_cps=4.2,
     ascii_cps=3.6,
     hard_stop_margin_steps=1,
+    short_headroom_max_units=2,
+    short_hard_stop_floor_steps=5,
 )
 MIXED_CFG_SCHEDULE = "row_ordinal_zero_and_even_primary_odd_alternate"
 MIXED_CFG_PRIMARY = 3.0
@@ -1446,6 +1458,40 @@ def active_audio_rms_db(audio: np.ndarray | Sequence[float], *, top_db: float =
     return 20.0 * math.log10(rms)
 
 
+def endpoint_tail_energy_ratio(
+    audio: np.ndarray | Sequence[float],
+    sample_rate: int,
+    *,
+    window_ms: float = ENDPOINT_TAIL_WINDOW_MS,
+) -> float:
+    """Measure pre-fade endpoint RMS relative to the waveform peak.
+
+    The ratio is intentionally scale-independent and is measured before final
+    fade/padding. A strongly voiced forced endpoint approaches one, while a
+    naturally quiet endpoint approaches zero.
+    """
+
+    waveform = _mono_audio(audio)
+    if isinstance(sample_rate, (bool, np.bool_)):
+        raise ValueError("sample_rate must be a positive integer")
+    try:
+        rate = operator.index(sample_rate)
+        duration_ms = float(window_ms)
+    except (TypeError, ValueError, OverflowError) as error:
+        raise ValueError("endpoint window must be finite and positive") from error
+    if rate <= 0 or not math.isfinite(duration_ms) or duration_ms <= 0.0:
+        raise ValueError("endpoint window must be finite and positive")
+    window = max(1, int(round(duration_ms * int(rate) / 1000.0)))
+    tail = waveform[-min(waveform.size, window) :]
+    peak = float(np.max(np.abs(waveform)))
+    if not math.isfinite(peak) or peak <= 1.0e-8:
+        return 0.0
+    tail_rms = float(np.sqrt(np.mean(np.square(tail, dtype=np.float64))))
+    if not math.isfinite(tail_rms) or tail_rms < 0.0:
+        raise ValueError("endpoint energy is invalid")
+    return float(np.clip(tail_rms / peak, 0.0, 1.0))
+
+
 def active_audio_median_f0_hz(
     audio: np.ndarray | Sequence[float],
     sample_rate: int,
@@ -1521,6 +1567,10 @@ class ChunkCandidateArtifact:
     speaker_embedding: np.ndarray | None = None
     rms_db: float | None = None
     median_f0_hz: float | None = None
+    stop_reason: str | None = None
+    endpoint_energy_ratio: float | None = None
+    generated_steps: int | None = None
+    hard_stop_steps: int | None = None
 
 
 @dataclass(frozen=True)
@@ -1584,6 +1634,10 @@ class CandidateGenerationEvidence:
     chunk_candidate_ordinals: tuple[int, ...] = ()
     network_conditioned: tuple[bool, ...] = ()
     chunk_text_variants: tuple[str, ...] = ()
+    chunk_stop_reasons: tuple[str, ...] = ()
+    chunk_endpoint_energy_ratios: tuple[float, ...] = ()
+    chunk_generated_steps: tuple[int, ...] = ()
+    chunk_hard_stop_steps: tuple[int, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -1602,6 +1656,10 @@ class CandidateAttemptEvidence:
     chunk_candidate_ordinals: tuple[int, ...]
     network_conditioned: tuple[bool, ...]
     chunk_text_variants: tuple[str, ...]
+    chunk_stop_reasons: tuple[str, ...]
+    chunk_endpoint_energy_ratios: tuple[float, ...]
+    chunk_generated_steps: tuple[int, ...]
+    chunk_hard_stop_steps: tuple[int, ...]
     scheduled_cfg: float | None
     effective_cfgs: tuple[float, ...]
     floor_reasons: tuple[tuple[str, ...], ...]
@@ -2530,6 +2588,12 @@ def _candidate_attempt_evidence_payload(
         "chunk_policies": chunk_policies,
         "chunk_text_variants": list(evidence.chunk_text_variants),
         "network_conditioned": list(evidence.network_conditioned),
+        "chunk_stop_reasons": list(evidence.chunk_stop_reasons),
+        "chunk_endpoint_energy_ratios": list(
+            evidence.chunk_endpoint_energy_ratios
+        ),
+        "chunk_generated_steps": list(evidence.chunk_generated_steps),
+        "chunk_hard_stop_steps": list(evidence.chunk_hard_stop_steps),
         "scheduled_cfg": evidence.scheduled_cfg,
         "effective_cfgs": list(evidence.effective_cfgs),
         "floor_reasons": [list(reasons) for reasons in evidence.floor_reasons],
@@ -2619,7 +2683,12 @@ def _selected_generation_evidence_payload(
     policies: list[str | None] = []
     network_conditioned: list[bool | None] = []
     chunk_text_variants: list[str | None] = []
+    stop_reasons: list[str | None] = []
+    endpoint_energy_ratios: list[float | None] = []
+    generated_steps: list[int | None] = []
+    hard_stop_steps: list[int | None] = []
     complete = True
+    endpoint_complete = True
     for chunk_index, candidate_index in enumerate(
         selection.chunk_candidate_indices
     ):
@@ -2633,6 +2702,11 @@ def _selected_generation_evidence_payload(
             policies.append(None)
             network_conditioned.append(None)
             chunk_text_variants.append(None)
+            stop_reasons.append(None)
+            endpoint_energy_ratios.append(None)
+            generated_steps.append(None)
+            hard_stop_steps.append(None)
+            endpoint_complete = False
             continue
         try:
             local_index = attempt.chunk_indices.index(chunk_index)
@@ -2647,6 +2721,11 @@ def _selected_generation_evidence_payload(
             policies.append(None)
             network_conditioned.append(None)
             chunk_text_variants.append(None)
+            stop_reasons.append(None)
+            endpoint_energy_ratios.append(None)
+            generated_steps.append(None)
+            hard_stop_steps.append(None)
+            endpoint_complete = False
             continue
         try:
             ordinal = attempt.chunk_candidate_ordinals[local_index]
@@ -2672,8 +2751,41 @@ def _selected_generation_evidence_payload(
         except IndexError:
             complete = False
             chunk_text_variants.append(None)
+        try:
+            stop_reason = attempt.chunk_stop_reasons[local_index]
+            endpoint_energy_ratio = (
+                attempt.chunk_endpoint_energy_ratios[local_index]
+            )
+            generated_step_count = attempt.chunk_generated_steps[local_index]
+            hard_stop_step_count = attempt.chunk_hard_stop_steps[local_index]
+        except IndexError:
+            endpoint_complete = False
+            stop_reasons.append(None)
+            endpoint_energy_ratios.append(None)
+            generated_steps.append(None)
+            hard_stop_steps.append(None)
+        else:
+            endpoint_artifact = ChunkCandidateArtifact(
+                stop_reason=stop_reason,
+                endpoint_energy_ratio=endpoint_energy_ratio,
+                generated_steps=generated_step_count,
+                hard_stop_steps=hard_stop_step_count,
+            )
+            if (
+                stop_reason is None
+                or endpoint_energy_ratio is None
+                or generated_step_count is None
+                or hard_stop_step_count is None
+                or not _endpoint_artifact_is_valid(endpoint_artifact)
+            ):
+                endpoint_complete = False
+            stop_reasons.append(stop_reason)
+            endpoint_energy_ratios.append(endpoint_energy_ratio)
+            generated_steps.append(generated_step_count)
+            hard_stop_steps.append(hard_stop_step_count)
     return {
         "complete": complete,
+        "endpoint_complete": endpoint_complete,
         "chunk_scheduled_cfgs": scheduled_cfgs,
         "chunk_effective_cfgs": effective_cfgs,
         "chunk_floor_reasons": floor_reasons,
@@ -2681,9 +2793,47 @@ def _selected_generation_evidence_payload(
         "chunk_policies": policies,
         "network_conditioned": network_conditioned,
         "chunk_text_variants": chunk_text_variants,
+        "chunk_stop_reasons": stop_reasons,
+        "chunk_endpoint_energy_ratios": endpoint_energy_ratios,
+        "chunk_generated_steps": generated_steps,
+        "chunk_hard_stop_steps": hard_stop_steps,
     }
 
 
+def _attempt_endpoint_evidence_is_complete(
+    attempt: CandidateAttemptEvidence,
+) -> bool:
+    """Validate every bounded endpoint row before declaring it complete."""
+
+    row_count = len(attempt.chunk_indices)
+    endpoint_rows = (
+        attempt.chunk_stop_reasons,
+        attempt.chunk_endpoint_energy_ratios,
+        attempt.chunk_generated_steps,
+        attempt.chunk_hard_stop_steps,
+    )
+    if row_count == 0 or any(len(row) != row_count for row in endpoint_rows):
+        return False
+    return all(
+        reason is not None
+        and energy is not None
+        and generated is not None
+        and hard_stop is not None
+        and _endpoint_artifact_is_valid(
+            ChunkCandidateArtifact(
+                stop_reason=reason,
+                endpoint_energy_ratio=energy,
+                generated_steps=generated,
+                hard_stop_steps=hard_stop,
+            )
+        )
+        for reason, energy, generated, hard_stop in zip(
+            *endpoint_rows,
+            strict=True,
+        )
+    )
+
+
 def format_cascade_evidence_log(
     diagnostics: CascadeDiagnostics,
     *,
@@ -2723,6 +2873,19 @@ def format_cascade_evidence_log(
             "generated_text_unit_limit must be an integer between 1 and 800"
         )
 
+    attempts = diagnostics.attempts[:CASCADE_EVIDENCE_MAX_ATTEMPTS]
+    for attempt in attempts:
+        endpoint_rows = (
+            attempt.chunk_stop_reasons,
+            attempt.chunk_endpoint_energy_ratios,
+            attempt.chunk_generated_steps,
+            attempt.chunk_hard_stop_steps,
+        )
+        if any(endpoint_rows) and not _attempt_endpoint_evidence_is_complete(
+            attempt
+        ):
+            raise ValueError("canonical endpoint evidence is invalid")
+
     selected: dict[str, Any] | None = None
     if selection is not None:
         if not isinstance(selection, CascadeResult):
@@ -2759,7 +2922,6 @@ def format_cascade_evidence_log(
             ),
         }
 
-    attempts = diagnostics.attempts[:CASCADE_EVIDENCE_MAX_ATTEMPTS]
     generation_evidence_complete = bool(attempts) and all(
         attempt.scheduled_cfg is not None
         and len(attempt.chunk_indices) == len(attempt.chunk_text_units)
@@ -2771,6 +2933,10 @@ def format_cascade_evidence_log(
         and bool(attempt.chunk_indices)
         for attempt in attempts
     )
+    endpoint_evidence_complete = bool(attempts) and all(
+        _attempt_endpoint_evidence_is_complete(attempt)
+        for attempt in attempts
+    )
     generated_chunk_count = sum(len(attempt.chunk_indices) for attempt in attempts)
     generated_text_units = sum(
         sum(attempt.chunk_text_units) for attempt in attempts
@@ -2804,6 +2970,7 @@ def format_cascade_evidence_log(
         },
         "attempt_count": len(diagnostics.attempts),
         "generation_evidence_complete": generation_evidence_complete,
+        "endpoint_evidence_complete": endpoint_evidence_complete,
         "generated_chunk_count": generated_chunk_count,
         "generated_text_units": generated_text_units,
         "attempts": [
@@ -3285,6 +3452,7 @@ def _validated_candidate_generation_evidence(
     chunks: tuple[str, ...],
     expected_candidate_ordinals: tuple[int, ...],
     require_explicit_candidate_ordinals: bool,
+    require_endpoint_evidence: bool,
 ) -> CandidateGenerationEvidence:
     """Validate deterministic CFG evidence supplied by the hosted app."""
 
@@ -3326,6 +3494,55 @@ def _validated_candidate_generation_evidence(
         chunk_text_variants = value.chunk_text_variants
     else:
         chunk_text_variants = ("base",) * len(chunks)
+    endpoint_rows = (
+        value.chunk_stop_reasons,
+        value.chunk_endpoint_energy_ratios,
+        value.chunk_generated_steps,
+        value.chunk_hard_stop_steps,
+    )
+    endpoint_evidence_present = any(endpoint_rows)
+    if require_endpoint_evidence and not all(endpoint_rows):
+        raise ValueError("generation endpoint evidence is required")
+    if endpoint_evidence_present and not all(
+        isinstance(row, tuple) and len(row) == len(chunks)
+        for row in endpoint_rows
+    ):
+        raise ValueError("generation endpoint evidence does not match the attempt")
+    stop_reasons: list[str] = []
+    endpoint_energy_ratios: list[float] = []
+    generated_steps: list[int] = []
+    hard_stop_steps: list[int] = []
+    if endpoint_evidence_present:
+        for raw_reason, raw_energy, raw_generated, raw_hard_stop in zip(
+            *endpoint_rows,
+            strict=True,
+        ):
+            if raw_reason not in _GENERATION_STOP_REASONS:
+                raise ValueError("generation stop reason is invalid")
+            energy = _finite_float(raw_energy, minimum=0.0, maximum=1.0)
+            if energy is None:
+                raise ValueError("generation endpoint energy is invalid")
+            if any(
+                isinstance(value, (bool, np.bool_))
+                for value in (raw_generated, raw_hard_stop)
+            ):
+                raise ValueError("generation endpoint steps are invalid")
+            try:
+                generated = operator.index(raw_generated)
+                hard_stop = operator.index(raw_hard_stop)
+            except (TypeError, ValueError, OverflowError) as error:
+                raise ValueError("generation endpoint steps are invalid") from error
+            if (
+                not 1 <= generated <= 2_000
+                or not 1 <= hard_stop <= 2_000
+                or generated > hard_stop
+                or (raw_reason == "hard_stop" and generated != hard_stop)
+            ):
+                raise ValueError("generation endpoint steps are invalid")
+            stop_reasons.append(raw_reason)
+            endpoint_energy_ratios.append(energy)
+            generated_steps.append(int(generated))
+            hard_stop_steps.append(int(hard_stop))
     raw_ordinals = value.chunk_candidate_ordinals
     if not raw_ordinals:
         if require_explicit_candidate_ordinals:
@@ -3406,6 +3623,10 @@ def _validated_candidate_generation_evidence(
         chunk_candidate_ordinals=candidate_ordinals,
         network_conditioned=network_conditioned,
         chunk_text_variants=chunk_text_variants,
+        chunk_stop_reasons=tuple(stop_reasons),
+        chunk_endpoint_energy_ratios=tuple(endpoint_energy_ratios),
+        chunk_generated_steps=tuple(generated_steps),
+        chunk_hard_stop_steps=tuple(hard_stop_steps),
     )
 
 
@@ -3447,6 +3668,20 @@ def _candidate_attempt_evidence(
         chunk_text_variants=(
             generation.chunk_text_variants if generation is not None else ()
         ),
+        chunk_stop_reasons=(
+            generation.chunk_stop_reasons if generation is not None else ()
+        ),
+        chunk_endpoint_energy_ratios=(
+            generation.chunk_endpoint_energy_ratios
+            if generation is not None
+            else ()
+        ),
+        chunk_generated_steps=(
+            generation.chunk_generated_steps if generation is not None else ()
+        ),
+        chunk_hard_stop_steps=(
+            generation.chunk_hard_stop_steps 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 ()),
@@ -3598,7 +3833,17 @@ def _sequence_fallback_search(
                         except ValueError:
                             speaker_artifact_valid = False
                     if speaker_artifact_valid:
-                        score = adjusted_result.score
+                        endpoint_cost = candidate_endpoint_selection_cost(
+                            artifact
+                        )
+                        adjusted_score = adjusted_result.score + endpoint_cost
+                        if math.isfinite(adjusted_score):
+                            adjusted_result = replace(
+                                adjusted_result,
+                                score=adjusted_score,
+                            )
+                            adjusted_results[-1] = adjusted_result
+                            score = adjusted_score
             local_scores[chunk_index].append(score)
         sequence_results.append(tuple(adjusted_results))
 
@@ -3981,6 +4226,50 @@ def _validate_coverage_evidence(
     return verification
 
 
+def _endpoint_artifact_is_valid(artifact: ChunkCandidateArtifact) -> bool:
+    """Validate an optional all-or-none endpoint provenance row."""
+
+    endpoint_values = (
+        artifact.stop_reason,
+        artifact.endpoint_energy_ratio,
+        artifact.generated_steps,
+        artifact.hard_stop_steps,
+    )
+    if all(value is None for value in endpoint_values):
+        return True
+    if any(value is not None for value in endpoint_values):
+        if any(value is None for value in endpoint_values):
+            return False
+        if artifact.stop_reason not in _GENERATION_STOP_REASONS:
+            return False
+        energy = _finite_float(
+            artifact.endpoint_energy_ratio,
+            minimum=0.0,
+            maximum=1.0,
+        )
+        if energy is None or any(
+            isinstance(value, (bool, np.bool_))
+            for value in (artifact.generated_steps, artifact.hard_stop_steps)
+        ):
+            return False
+        try:
+            generated = operator.index(artifact.generated_steps)
+            hard_stop = operator.index(artifact.hard_stop_steps)
+        except (TypeError, ValueError, OverflowError):
+            return False
+        if (
+            not 1 <= generated <= 2_000
+            or not 1 <= hard_stop <= 2_000
+            or generated > hard_stop
+            or (
+                artifact.stop_reason == "hard_stop"
+                and generated != hard_stop
+            )
+        ):
+            return False
+    return True
+
+
 def _coverage_artifact_is_valid(
     result: CandidateGateResult,
     artifact: ChunkCandidateArtifact,
@@ -4006,6 +4295,8 @@ def _coverage_artifact_is_valid(
         minimum=1.0,
     ) is None:
         return False
+    if not _endpoint_artifact_is_valid(artifact):
+        return False
     embedding = artifact.speaker_embedding
     if result.speaker_gate_applied is True and embedding is None:
         return False
@@ -4018,6 +4309,127 @@ def _coverage_artifact_is_valid(
     return True
 
 
+def _bind_generation_endpoint_evidence(
+    verification: TrajectoryGateResult,
+    generation: CandidateGenerationEvidence | None,
+    *,
+    required: bool,
+) -> TrajectoryGateResult:
+    """Attach validated generation endpoint provenance to chunk artifacts."""
+
+    if generation is None or not generation.chunk_stop_reasons:
+        if required:
+            raise RuntimeError("candidate endpoint evidence is missing")
+        return verification
+    row_count = len(verification.chunk_artifacts)
+    endpoint_rows = (
+        generation.chunk_stop_reasons,
+        generation.chunk_endpoint_energy_ratios,
+        generation.chunk_generated_steps,
+        generation.chunk_hard_stop_steps,
+    )
+    if any(len(row) != row_count for row in endpoint_rows):
+        raise RuntimeError("candidate endpoint evidence is misaligned")
+    artifacts = tuple(
+        replace(
+            artifact,
+            stop_reason=reason,
+            endpoint_energy_ratio=energy,
+            generated_steps=generated,
+            hard_stop_steps=hard_stop,
+        )
+        for artifact, reason, energy, generated, hard_stop in zip(
+            verification.chunk_artifacts,
+            *endpoint_rows,
+            strict=True,
+        )
+    )
+    if any(
+        not _coverage_artifact_is_valid(result, artifact)
+        for result, artifact in zip(
+            verification.candidate_results,
+            artifacts,
+            strict=True,
+        )
+        if result.comparison.passed is True
+    ):
+        raise RuntimeError("candidate endpoint evidence is invalid")
+    return replace(verification, chunk_artifacts=artifacts)
+
+
+def candidate_endpoint_selection_cost(
+    artifact: ChunkCandidateArtifact,
+) -> float:
+    """Return a bounded soft preference for natural, quiet endpoints."""
+
+    if artifact.stop_reason is None and artifact.endpoint_energy_ratio is None:
+        return 0.0
+    if not _endpoint_artifact_is_valid(artifact):
+        return math.inf
+    energy = float(artifact.endpoint_energy_ratio)
+    hard_stop = (
+        ENDPOINT_HARD_STOP_PENALTY
+        if _endpoint_artifact_reached_hard_cap(artifact)
+        else 0.0
+    )
+    return hard_stop + ENDPOINT_ENERGY_WEIGHT * energy
+
+
+def _endpoint_artifact_reached_hard_cap(
+    artifact: ChunkCandidateArtifact,
+) -> bool:
+    """Return whether endpoint evidence reached the configured hard cap.
+
+    A threshold crossing can coincide with the final permitted generation
+    step.  The stop reason correctly records that the model crossed the
+    threshold, but endpoint selection must still treat it as cap-reached so it
+    cannot evade the bounded hard-cap cost or natural-endpoint waiver rules.
+    """
+
+    if not _endpoint_artifact_is_valid(artifact):
+        return False
+    if artifact.generated_steps is None or artifact.hard_stop_steps is None:
+        return False
+    return artifact.generated_steps >= artifact.hard_stop_steps
+
+
+def _preferred_natural_endpoint_waiver(
+    verification: TrajectoryGateResult,
+    artifact: ChunkCandidateArtifact,
+    *,
+    min_speaker_similarity: float | None,
+    max_boundary_speaker_drop: float | None,
+    min_squim_stoi: float | None,
+    min_squim_pesq: float | None,
+    min_squim_audio_duration_seconds: float,
+) -> bool:
+    """Allow only a tightly bounded quality trade for a true pre-cap stop."""
+
+    if (
+        artifact.stop_reason != "stop_threshold"
+        or _endpoint_artifact_reached_hard_cap(artifact)
+    ):
+        return False
+    relaxed_stoi = (
+        None
+        if min_squim_stoi is None
+        else max(0.0, min_squim_stoi - ENDPOINT_PREFERRED_SQUIM_STOI_SLACK)
+    )
+    relaxed_pesq = (
+        None
+        if min_squim_pesq is None
+        else max(0.0, min_squim_pesq - ENDPOINT_PREFERRED_SQUIM_PESQ_SLACK)
+    )
+    return _preferred_release_verification(
+        verification,
+        min_speaker_similarity=min_speaker_similarity,
+        max_boundary_speaker_drop=max_boundary_speaker_drop,
+        min_squim_stoi=relaxed_stoi,
+        min_squim_pesq=relaxed_pesq,
+        min_squim_audio_duration_seconds=min_squim_audio_duration_seconds,
+    )
+
+
 def _coverage_local_candidate(
     *,
     candidate_index: int,
@@ -4042,6 +4454,11 @@ def _coverage_local_candidate(
         return None
     if not _coverage_artifact_is_valid(adjusted, artifact):
         raise RuntimeError("local verifier passed without valid acoustic evidence")
+    endpoint_cost = candidate_endpoint_selection_cost(artifact)
+    adjusted_score = adjusted.score + endpoint_cost
+    if not math.isfinite(adjusted_score) or adjusted_score < 0.0:
+        raise RuntimeError("local verifier passed without valid endpoint evidence")
+    adjusted = replace(adjusted, score=adjusted_score)
     return _RaggedChunkCandidate(
         candidate_index=candidate_index,
         seed=seed,
@@ -4132,6 +4549,7 @@ def run_coverage_adaptive_cascade(
     preferred_min_squim_stoi: float | None = None,
     preferred_min_squim_pesq: float | None = None,
     preferred_min_squim_audio_duration_seconds: float = 0.0,
+    require_endpoint_evidence: bool = False,
 ) -> CascadeResult:
     """Run one whole trajectory, then deterministic low-coverage refills.
 
@@ -4168,6 +4586,12 @@ def run_coverage_adaptive_cascade(
         generation_evidence_factory
     ):
         raise ValueError("generation_evidence_factory must be callable")
+    if type(require_endpoint_evidence) is not bool:
+        raise ValueError("require_endpoint_evidence must be a boolean")
+    if require_endpoint_evidence and generation_evidence_factory is None:
+        raise ValueError(
+            "endpoint evidence requires a generation evidence factory"
+        )
     if candidate_generation_text_transform is not None and not callable(
         candidate_generation_text_transform
     ):
@@ -4408,6 +4832,7 @@ def run_coverage_adaptive_cascade(
                 chunks=candidate_chunks,
                 expected_candidate_ordinals=expected_ordinals,
                 require_explicit_candidate_ordinals=context_aware_evidence,
+                require_endpoint_evidence=require_endpoint_evidence,
             )
         except Exception as error:
             raise RuntimeError("candidate generation evidence is invalid") from error
@@ -4461,6 +4886,11 @@ def run_coverage_adaptive_cascade(
         chunk_count,
         verifier_name="whole candidate verifier",
     )
+    initial_verification = _bind_generation_endpoint_evidence(
+        initial_verification,
+        initial_generation_evidence,
+        required=require_endpoint_evidence,
+    )
     diagnostic_candidates.append(
         _VerifiedTrajectoryCandidate(
             candidate_index=0,
@@ -4687,6 +5117,11 @@ def run_coverage_adaptive_cascade(
             1,
             verifier_name="refill candidate verifier",
         )
+        refill_verification = _bind_generation_endpoint_evidence(
+            refill_verification,
+            refill_generation_evidence,
+            required=require_endpoint_evidence,
+        )
         latest_local_results[chunk_index] = refill_verification.candidate_results[0]
         diagnostic_candidates.append(
             _VerifiedTrajectoryCandidate(
@@ -4729,14 +5164,67 @@ def run_coverage_adaptive_cascade(
                 min_squim_audio_duration_seconds=preferred_squim_duration,
             )
         ):
+            preferred_candidate = diagnostic_candidates[-1]
+            preferred_artifact = preferred_candidate.verification.chunk_artifacts[0]
+            preferred_reached_hard_cap = (
+                _endpoint_artifact_reached_hard_cap(preferred_artifact)
+            )
+            selectable = [
+                candidate
+                for candidate in diagnostic_candidates
+                if candidate.verification.passed is True
+                and math.isfinite(candidate.verification.score)
+                and len(candidate.verification.chunk_artifacts) == 1
+                and (
+                    _preferred_release_verification(
+                        candidate.verification,
+                        min_speaker_similarity=preferred_similarity,
+                        max_boundary_speaker_drop=preferred_boundary,
+                        min_squim_stoi=preferred_stoi,
+                        min_squim_pesq=preferred_pesq,
+                        min_squim_audio_duration_seconds=(
+                            preferred_squim_duration
+                        ),
+                    )
+                    or (
+                        preferred_reached_hard_cap
+                        and _preferred_natural_endpoint_waiver(
+                            candidate.verification,
+                            candidate.verification.chunk_artifacts[0],
+                            min_speaker_similarity=preferred_similarity,
+                            max_boundary_speaker_drop=preferred_boundary,
+                            min_squim_stoi=preferred_stoi,
+                            min_squim_pesq=preferred_pesq,
+                            min_squim_audio_duration_seconds=(
+                                preferred_squim_duration
+                            ),
+                        )
+                    )
+                )
+                and math.isfinite(
+                    candidate_endpoint_selection_cost(
+                        candidate.verification.chunk_artifacts[0]
+                    )
+                )
+            ]
+            selected_candidate = min(
+                selectable,
+                key=lambda candidate: (
+                    candidate.verification.score
+                    + candidate_endpoint_selection_cost(
+                        candidate.verification.chunk_artifacts[0]
+                    ),
+                    candidate.candidate_index,
+                ),
+            )
             return CascadeResult(
-                trajectory=refill_trajectory,
-                verification=refill_verification,
-                seed=seed,
-                candidate_index=next_candidate_index,
+                trajectory=selected_candidate.trajectory,
+                verification=selected_candidate.verification,
+                seed=selected_candidate.seed,
+                candidate_index=selected_candidate.candidate_index,
                 attempted_seeds=tuple(attempted_seeds),
-                chunk_candidate_indices=(next_candidate_index,),
-                chunk_seeds=(seed,),
+                chunk_candidate_indices=(selected_candidate.candidate_index,),
+                chunk_seeds=(selected_candidate.seed,),
                 selection_mode="whole_trajectory",
                 diagnostics=_cascade_diagnostics(diagnostic_candidates),
                 generated_chunk_count=generated_chunks,

tests/test_coverage_adaptive.py

@@ -1,3 +1,4 @@
+from dataclasses import replace
 import json
 import math
 
@@ -368,6 +369,222 @@ def test_single_chunk_requires_joint_preferred_quality_and_speaker_for_early_ret
     assert payload["selection"]["generated_chunks"] == 3
 
 
+def test_preferred_refill_selects_prior_natural_endpoint_over_forced_candidate():
+    chunk = ("完整內容",)
+    generated = []
+    profiles = {
+        10: (0.65, 1.15),
+        11: (0.71, 1.19),
+        12: (0.73, 1.21),
+    }
+
+    def verifier(trajectory, chunks, seed):
+        stoi, pesq = profiles[seed]
+        return _squim_verification(
+            chunks,
+            stoi=stoi,
+            pesq=pesq,
+            speaker_similarity=0.30,
+        )
+
+    def generation_evidence(candidate_index, seed, chunk_indices, candidate_chunks):
+        scheduled = generation_cfg_for_candidate_offset(candidate_index)
+        effective = max(3.0, scheduled)
+        threshold_stop = candidate_index == 1
+        return CandidateGenerationEvidence(
+            chunk_indices=chunk_indices,
+            chunk_text_units=(4,),
+            scheduled_cfg=scheduled,
+            effective_cfgs=(effective,),
+            floor_reasons=(("short_text",),) if effective != scheduled else ((),),
+            chunk_stop_reasons=(
+                ("stop_threshold",) if threshold_stop else ("hard_stop",)
+            ),
+            chunk_endpoint_energy_ratios=(
+                (0.10,) if threshold_stop else (0.90,)
+            ),
+            chunk_generated_steps=((4,) if threshold_stop else (5,)),
+            chunk_hard_stop_steps=(5,),
+        )
+
+    result = run_coverage_adaptive_cascade(
+        chunk,
+        10,
+        lambda chunks, seed: generated.append(seed) or (f"audio-{seed}",),
+        verifier,
+        verifier,
+        sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
+        generation_evidence_factory=generation_evidence,
+        max_generated_chunks=5,
+        max_generated_text_units=100,
+        preferred_min_speaker_similarity=0.25,
+        preferred_max_boundary_speaker_drop=0.05,
+        preferred_min_squim_stoi=0.72,
+        preferred_min_squim_pesq=1.20,
+        require_endpoint_evidence=True,
+    )
+
+    assert generated == [10, 11, 12]
+    assert result.candidate_index == 1
+    assert result.seed == 11
+    artifact = result.verification.chunk_artifacts[0]
+    assert artifact.stop_reason == "stop_threshold"
+    assert artifact.endpoint_energy_ratio == pytest.approx(0.10)
+    payload = json.loads(
+        format_cascade_evidence_log(
+            result.diagnostics,
+            outcome="returned",
+            generated_chunk_limit=5,
+            generated_text_unit_limit=100,
+            selection=result,
+        ).removeprefix(CASCADE_EVIDENCE_LOG_PREFIX)
+    )
+    assert payload["endpoint_evidence_complete"] is True
+    assert payload["selection"]["generation"]["endpoint_complete"] is True
+    assert payload["selection"]["generation"]["chunk_stop_reasons"] == [
+        "stop_threshold"
+    ]
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("chunk_stop_reasons", ("unknown",)),
+        ("chunk_endpoint_energy_ratios", (math.nan,)),
+        ("chunk_generated_steps", (True,)),
+        ("chunk_generated_steps", (6,)),
+    ],
+)
+def test_canonical_log_rejects_invalid_endpoint_rows(field, value):
+    chunk = ("完整內容",)
+
+    def evidence(candidate_index, seed, chunk_indices, candidate_chunks):
+        return CandidateGenerationEvidence(
+            chunk_indices=chunk_indices,
+            chunk_text_units=(4,),
+            scheduled_cfg=3.0,
+            effective_cfgs=(3.0,),
+            floor_reasons=((),),
+            chunk_stop_reasons=("stop_threshold",),
+            chunk_endpoint_energy_ratios=(0.10,),
+            chunk_generated_steps=(4,),
+            chunk_hard_stop_steps=(5,),
+        )
+
+    result = run_coverage_adaptive_cascade(
+        chunk,
+        10,
+        lambda chunks, seed: (f"audio-{seed}",),
+        lambda trajectory, chunks, seed: _squim_verification(
+            chunks,
+            stoi=0.73,
+            pesq=1.21,
+            speaker_similarity=0.30,
+        ),
+        lambda trajectory, chunks, seed: _squim_verification(
+            chunks,
+            stoi=0.73,
+            pesq=1.21,
+            speaker_similarity=0.30,
+        ),
+        sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
+        generation_evidence_factory=evidence,
+        max_generated_chunks=5,
+        max_generated_text_units=100,
+        preferred_min_speaker_similarity=0.25,
+        preferred_max_boundary_speaker_drop=0.05,
+        preferred_min_squim_stoi=0.72,
+        preferred_min_squim_pesq=1.20,
+        require_endpoint_evidence=True,
+    )
+
+    invalid_attempt = replace(
+        result.diagnostics.attempts[0],
+        **{field: value},
+    )
+    invalid_diagnostics = replace(
+        result.diagnostics,
+        attempts=(invalid_attempt,),
+    )
+    with pytest.raises(ValueError, match="canonical endpoint evidence is invalid"):
+        format_cascade_evidence_log(
+            invalid_diagnostics,
+            outcome="returned",
+            generated_chunk_limit=5,
+            generated_text_unit_limit=100,
+            selection=result,
+        )
+
+
+@pytest.mark.parametrize(
+    ("natural_stoi", "natural_pesq", "natural_speaker", "natural_steps"),
+    [
+        (0.69, 1.19, 0.30, 4),
+        (0.71, 1.16, 0.30, 4),
+        (0.71, 1.19, 0.24, 4),
+        (0.71, 1.19, 0.30, 5),
+    ],
+)
+def test_natural_endpoint_waiver_cannot_bypass_bounded_preferred_tier(
+    natural_stoi,
+    natural_pesq,
+    natural_speaker,
+    natural_steps,
+):
+    chunk = ("完整內容",)
+    profiles = {
+        10: (0.65, 1.15, 0.30),
+        11: (natural_stoi, natural_pesq, natural_speaker),
+        12: (0.73, 1.21, 0.30),
+    }
+
+    def verifier(trajectory, chunks, seed):
+        stoi, pesq, speaker = profiles[seed]
+        return _squim_verification(
+            chunks,
+            stoi=stoi,
+            pesq=pesq,
+            speaker_similarity=speaker,
+        )
+
+    def generation_evidence(candidate_index, seed, chunk_indices, candidate_chunks):
+        natural = candidate_index == 1
+        scheduled = generation_cfg_for_candidate_offset(candidate_index)
+        return CandidateGenerationEvidence(
+            chunk_indices=chunk_indices,
+            chunk_text_units=(4,),
+            scheduled_cfg=scheduled,
+            effective_cfgs=(scheduled,),
+            floor_reasons=((),),
+            chunk_stop_reasons=(
+                ("stop_threshold",) if natural else ("hard_stop",)
+            ),
+            chunk_endpoint_energy_ratios=((0.10,) if natural else (0.90,)),
+            chunk_generated_steps=((natural_steps,) if natural else (5,)),
+            chunk_hard_stop_steps=(5,),
+        )
+
+    result = run_coverage_adaptive_cascade(
+        chunk,
+        10,
+        lambda chunks, seed: (f"audio-{seed}",),
+        verifier,
+        verifier,
+        sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
+        generation_evidence_factory=generation_evidence,
+        max_generated_chunks=5,
+        max_generated_text_units=100,
+        preferred_min_speaker_similarity=0.25,
+        preferred_max_boundary_speaker_drop=0.05,
+        preferred_min_squim_stoi=0.72,
+        preferred_min_squim_pesq=1.20,
+        require_endpoint_evidence=True,
+    )
+
+    assert result.candidate_index == 2
+    assert result.seed == 12
+
+
 def test_single_chunk_final_cap_selects_lowest_bounded_soft_quality_cost():
     chunk = ("完整內容",)
     profiles = {
@@ -567,7 +784,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"] == 5
+    assert payload["schema_version"] == 6
     assert payload["generation_evidence_complete"] is True
     assert payload["request_chunk_count"] == 2
     assert payload["generated_chunk_count"] == 3
@@ -590,6 +807,11 @@ def test_refill_budget_accounts_exact_generated_chunks_and_text_units():
         "chunk_policies": ["safe_duration", "base"],
         "network_conditioned": [False, False],
         "chunk_text_variants": ["base", "base"],
+        "endpoint_complete": False,
+        "chunk_stop_reasons": [None, None],
+        "chunk_endpoint_energy_ratios": [None, None],
+        "chunk_generated_steps": [None, None],
+        "chunk_hard_stop_steps": [None, None],
     }
 
 

tests/test_inference_cleanup.py

@@ -73,6 +73,8 @@ def test_closed_loop_pace_rerenders_once_from_untouched_model_waveform():
         "target_pace_speed": lambda *args, **kwargs: 0.90,
         "active_voiced_duration_seconds": active_duration,
         "active_pace_correction_speed": active_correction,
+        "endpoint_tail_energy_ratio": lambda audio, sample_rate: 0.5,
+        "_ChunkGenerationOutcome": SimpleNamespace,
         "_apply_speed": apply_speed,
         "_NATIVE_STOP_POLICY": False,
         "_GENERATE_PARAMETERS": set(),
@@ -113,7 +115,14 @@ def test_closed_loop_pace_rerenders_once_from_untouched_model_waveform():
     assert apply_calls[0][1] == pytest.approx(0.90)
     assert apply_calls[1][1] == pytest.approx(0.90 * 0.95)
     assert all(call[2] is True for call in apply_calls)
-    assert np.array_equal(output, np.ones(16, dtype=np.float32) * np.float32(0.855))
+    assert np.array_equal(
+        output.audio,
+        np.ones(16, dtype=np.float32) * np.float32(0.855),
+    )
+    assert output.stop_reason == "native_stop"
+    assert output.endpoint_energy_ratio == 0.5
+    assert output.generated_steps == 12
+    assert output.hard_stop_steps == 12
     assert len(active_calls) == 2
     assert np.array_equal(active_calls[0][0], np.ones(16, dtype=np.float32))
     assert np.array_equal(active_calls[1][0], np.ones(16, dtype=np.float32) * 0.90)
@@ -124,6 +133,84 @@ def test_closed_loop_pace_rerenders_once_from_untouched_model_waveform():
     }
 
 
+def test_sparse_completion_policy_adds_only_bounded_one_two_unit_headroom():
+    def run(units, policy):
+        generation_calls = []
+        namespace = {
+            "np": np,
+            "torch": torch,
+            "model": SimpleNamespace(
+                generate=lambda **kwargs: (
+                    generation_calls.append(kwargs) or torch.ones(64)
+                )
+            ),
+            "select_generation_cps": lambda *args, **kwargs: 5.0,
+            "count_network_endpoint_duration_units": lambda text: units,
+            "count_speech_units": lambda text: units,
+            "endpoint_generation_plan": (
+                lambda *args, **kwargs: ("短句。", 3, 4)
+            ),
+            "effective_generation_cfg": lambda text, cfg, **kwargs: cfg,
+            "set_generation_seed": lambda seed: None,
+            "target_pace_speed": lambda *args, **kwargs: 1.0,
+            "active_voiced_duration_seconds": (
+                lambda *args, **kwargs: (_ for _ in ()).throw(ValueError())
+            ),
+            "active_pace_correction_speed": lambda *args, **kwargs: 1.0,
+            "endpoint_tail_energy_ratio": lambda audio, sample_rate: 0.25,
+            "_apply_speed": lambda audio, speed, **kwargs: np.asarray(audio),
+            "_ChunkGenerationOutcome": SimpleNamespace,
+            "_NATIVE_STOP_POLICY": False,
+            "_GENERATE_PARAMETERS": set(),
+            "_STOP_CONTROLLER": None,
+            "SR": 48_000,
+            "STEP_SECONDS": 0.16,
+            "MIN_ENDPOINT_CUE_UNITS": 6,
+            "SHORT_TEXT_CFG_UNITS": 6,
+            "SHORT_TEXT_CFG_MIN": 3.0,
+            "TARGET_CPS": 4.0,
+            "ACTIVE_PACE_TARGET_CPS": 4.0,
+            "CLOSED_LOOP_ACTIVE_PACE_TARGET_CPS": 3.95,
+            "MIN_PACE_SPEED": 0.80,
+            "STOP_THRESHOLD": 0.50,
+            "STOP_CONSECUTIVE": 1,
+        }
+        generate = _isolated_app_function("_generate_chunk", namespace)
+        outcome = generate(
+            "短句",
+            torch.ones(4),
+            cfg=3.0,
+            steps=10,
+            request_seed=123,
+            policy=policy,
+            network_conditioned=True,
+        )
+        return generation_calls[0]["max_len"], outcome
+
+    base = SimpleNamespace(
+        name="base",
+        cjk_cps=5.2,
+        ascii_cps=4.6,
+        hard_stop_margin_steps=1,
+        short_headroom_max_units=0,
+        short_hard_stop_floor_steps=0,
+    )
+    completion = SimpleNamespace(
+        name="completion_headroom",
+        cjk_cps=4.2,
+        ascii_cps=3.6,
+        hard_stop_margin_steps=1,
+        short_headroom_max_units=2,
+        short_hard_stop_floor_steps=5,
+    )
+
+    assert run(2, base)[0] == 4
+    headroom_cap, headroom = run(2, completion)
+    assert headroom_cap == 5
+    assert headroom.hard_stop_steps == 5
+    assert run(3, completion)[0] == 4
+
+
 def test_lazy_transition_f0_measures_only_dp_eligible_rows():
     observations = (
         CandidateObservation("第一段完整", "第一段完整", 2.0),

tests/test_production.py

@@ -20,7 +20,9 @@ from production import (
     count_network_endpoint_duration_units,
     count_speech_units,
     duration_hard_stop_steps,
+    decode_pcm16,
     email_domain_mail_generation_variant,
+    encode_pcm16,
     effective_generation_cfg,
     endpoint_generation_plan,
     ensure_terminal_punctuation,
@@ -39,6 +41,7 @@ from production import (
     normalize_tts_text,
     plan_generation_chunks,
     pcm16_audio_output,
+    pcm16_verification_waveform,
     select_candidate_sequence,
     select_generation_cps,
     split_leading_clause,
@@ -834,6 +837,24 @@ def test_stop_controller_can_stop_on_first_confident_hit():
     assert decisions == [0, 0, 1]
 
 
+def test_stop_controller_preserves_threshold_reason_at_the_hard_cap():
+    controller = StopHysteresisController(
+        _SequenceStopHead([0.01, 0.01, 0.01, 0.9]),
+        threshold=0.50,
+        late_threshold=0.05,
+        consecutive=1,
+    )
+    controller.begin(min_len=2, expected_steps=3, hard_stop_steps=4)
+
+    decisions = [
+        controller(torch.zeros(1, 2)).argmax(dim=-1).item()
+        for _ in range(4)
+    ]
+
+    assert decisions == [0, 0, 0, 1]
+    assert controller.last_stop_reason == "stop_threshold"
+
+
 def test_stop_controller_relaxes_near_endpoint_and_enforces_hard_stop():
     relaxed = StopHysteresisController(
         _SequenceStopHead([0.6] * 10),
@@ -3234,12 +3255,25 @@ def test_pcm16_output_preserves_production_level_without_peak_normalizing():
 
     assert sample_rate == 48_000
     assert pcm.dtype == np.int16
-    assert float(np.max(np.abs(pcm.astype(np.float64))) / 32767.0) == pytest.approx(
+    assert float(np.max(np.abs(pcm.astype(np.float64))) / 32768.0) == pytest.approx(
         0.62,
-        abs=1.0 / 32767.0,
+        abs=1.0 / 32768.0,
+    )
+    restored = decode_pcm16(pcm)
+    np.testing.assert_allclose(restored, waveform, atol=1.0 / 32768.0)
+
+
+def test_pcm16_verification_roundtrip_is_exhaustively_idempotent():
+    codes = np.arange(-32768, 32768, dtype=np.int32).astype(np.int16)
+
+    decoded = decode_pcm16(codes)
+    reencoded = encode_pcm16(decoded)
+
+    np.testing.assert_array_equal(reencoded, codes)
+    np.testing.assert_array_equal(
+        encode_pcm16(pcm16_verification_waveform(decoded)),
+        codes,
     )
-    restored = pcm.astype(np.float64) / 32767.0
-    np.testing.assert_allclose(restored, waveform, atol=1.0 / 32767.0)
 
 
 @pytest.mark.parametrize(

tests/test_quality_runtime.py

@@ -19,6 +19,10 @@ from quality_runtime import (
     COMPLETION_HEADROOM_GENERATION_POLICY,
     CASCADE_EVIDENCE_LOG_PREFIX,
     CASCADE_EVIDENCE_SCHEMA_VERSION,
+    ENDPOINT_ENERGY_WEIGHT,
+    ENDPOINT_HARD_STOP_PENALTY,
+    ENDPOINT_PREFERRED_SQUIM_PESQ_SLACK,
+    ENDPOINT_PREFERRED_SQUIM_STOI_SLACK,
     SAFE_DURATION_GENERATION_POLICY,
     SQUIM_OBJECTIVE_ASSET_PATH,
     SQUIM_OBJECTIVE_SAMPLE_RATE,
@@ -51,9 +55,11 @@ from quality_runtime import (
     active_voiced_duration_seconds,
     candidate_gate_evidence,
     candidate_chunk_transition_score,
+    candidate_endpoint_selection_cost,
     candidate_limit_for_chunk_budget,
     cosine_similarity,
     exact_waveform_sha256,
+    endpoint_tail_energy_ratio,
     format_cascade_evidence_log,
     generation_cfg_for_candidate_offset,
     generation_policy_for_candidate_offset,
@@ -295,11 +301,20 @@ def test_candidate_generation_policy_mapping_uses_sparse_completion_headroom():
     assert ADAPTIVE_CASCADE_STAGE_LIMITS == (1, 5, 10, 15, 20, 24, 28, 32)
     base = generation_policy_for_candidate_offset(0)
     assert base is BASE_GENERATION_POLICY
-    assert (base.name, base.cjk_cps, base.ascii_cps, base.hard_stop_margin_steps) == (
+    assert (
+        base.name,
+        base.cjk_cps,
+        base.ascii_cps,
+        base.hard_stop_margin_steps,
+        base.short_headroom_max_units,
+        base.short_hard_stop_floor_steps,
+    ) == (
         "base",
         5.2,
         4.6,
         1,
+        0,
+        0,
     )
 
     for offset in (1, 2, 5, 9):
@@ -310,9 +325,11 @@ def test_candidate_generation_policy_mapping_uses_sparse_completion_headroom():
             safe.cjk_cps,
             safe.ascii_cps,
             safe.hard_stop_margin_steps,
-        ) == ("safe_duration", 4.6, 4.0, 1)
+            safe.short_headroom_max_units,
+            safe.short_hard_stop_floor_steps,
+        ) == ("safe_duration", 4.6, 4.0, 1, 0, 0)
 
-    for offset in (4, 8, 12, 16):
+    for offset in (4, 8, 12, 16, 20, 24, 28):
         headroom = generation_policy_for_candidate_offset(offset)
         assert headroom is COMPLETION_HEADROOM_GENERATION_POLICY
         assert (
@@ -320,7 +337,54 @@ def test_candidate_generation_policy_mapping_uses_sparse_completion_headroom():
             headroom.cjk_cps,
             headroom.ascii_cps,
             headroom.hard_stop_margin_steps,
-        ) == ("completion_headroom", 4.2, 3.6, 1)
+            headroom.short_headroom_max_units,
+            headroom.short_hard_stop_floor_steps,
+        ) == ("completion_headroom", 4.2, 3.6, 1, 2, 5)
+
+
+def test_endpoint_tail_energy_and_selection_cost_prefer_natural_quiet_endings():
+    waveform = np.concatenate(
+        (
+            np.ones(950, dtype=np.float32),
+            np.full(50, 0.25, dtype=np.float32),
+        )
+    )
+    energy = endpoint_tail_energy_ratio(
+        waveform,
+        10_000,
+        window_ms=5.0,
+    )
+    threshold = ChunkCandidateArtifact(
+        stop_reason="stop_threshold",
+        endpoint_energy_ratio=energy,
+        generated_steps=4,
+        hard_stop_steps=5,
+    )
+    threshold_at_cap = ChunkCandidateArtifact(
+        stop_reason="stop_threshold",
+        endpoint_energy_ratio=energy,
+        generated_steps=5,
+        hard_stop_steps=5,
+    )
+    forced = ChunkCandidateArtifact(
+        stop_reason="hard_stop",
+        endpoint_energy_ratio=energy,
+        generated_steps=5,
+        hard_stop_steps=5,
+    )
+
+    assert energy == pytest.approx(0.25)
+    assert candidate_endpoint_selection_cost(threshold) == pytest.approx(
+        ENDPOINT_ENERGY_WEIGHT * 0.25
+    )
+    assert candidate_endpoint_selection_cost(threshold_at_cap) == pytest.approx(
+        ENDPOINT_HARD_STOP_PENALTY + ENDPOINT_ENERGY_WEIGHT * 0.25
+    )
+    assert candidate_endpoint_selection_cost(forced) == pytest.approx(
+        ENDPOINT_HARD_STOP_PENALTY + ENDPOINT_ENERGY_WEIGHT * 0.25
+    )
+    assert ENDPOINT_PREFERRED_SQUIM_STOI_SLACK == pytest.approx(0.02)
+    assert ENDPOINT_PREFERRED_SQUIM_PESQ_SLACK == pytest.approx(0.03)
 
 
 @pytest.mark.parametrize("offset", [-1, 0.5, 1.0, True, None, "one"])
@@ -1675,7 +1739,7 @@ def test_canonical_attempt_evidence_distinguishes_local_v3_pass_and_not_run():
     )
     attempt = payload["attempts"][0]
 
-    assert payload["schema_version"] == 5
+    assert payload["schema_version"] == 6
     assert attempt["independent_local_evidence_complete"] is True
     assert attempt["independent_local_results"] == [
         {

tests/test_release_pins.py

@@ -12,6 +12,7 @@ from production import (
     join_audio_chunks_variable,
     match_chunk_rms,
     normalize_spoken_forms,
+    pcm16_verification_waveform,
     plan_generation_chunks,
     punctuation_pause_seconds,
 )
@@ -103,6 +104,7 @@ def _isolated_assemble_trajectory_audio():
         "NETWORK_INTERNAL_FADE_MS": constants["NETWORK_INTERNAL_FADE_MS"],
         "CHUNK_EDGE_FADE_MS": constants["CHUNK_EDGE_FADE_MS"],
         "CROSSFADE_MS": constants["CROSSFADE_MS"],
+        "FINAL_ENDPOINT_FADE_MS": constants["FINAL_ENDPOINT_FADE_MS"],
         "match_chunk_rms": match_chunk_rms,
         "punctuation_pause_seconds": punctuation_pause_seconds,
         "fade_variable_internal_edges": fade_variable_internal_edges,
@@ -111,6 +113,7 @@ def _isolated_assemble_trajectory_audio():
         "_apply_speed": lambda waveform, _speed: waveform,
         "count_speech_units": count_speech_units,
         "finish_audio": lambda waveform, _sample_rate, **_kwargs: waveform,
+        "pcm16_verification_waveform": pcm16_verification_waveform,
     }
     exec(compile(module, str(app_path), "exec"), namespace)
     return namespace["_assemble_trajectory_audio"]
@@ -363,6 +366,18 @@ def test_app_wires_row_local_candidate_ordinal_to_generation_policy_and_logs_it(
     assert "effective_cfg={effective_cfg:.2f}" in source
     assert "network_floor_applied={network_floor_applied}" in source
     assert "short_floor_applied={short_floor_applied}" in source
+    assert "short_headroom_floor_applied={short_headroom_floor_applied}" in source
+    assert "generation_endpoint_evidence_by_seed" in source
+    assert "chunk_stop_reasons=tuple(" in source
+    assert "chunk_endpoint_energy_ratios=tuple(" in source
+    assert "require_endpoint_evidence=True" in source
+    quality_source = (ROOT / "quality_runtime.py").read_text(encoding="utf-8")
+    assert "short_headroom_max_units=2" in quality_source
+    assert "short_hard_stop_floor_steps=5" in quality_source
+    assert "ENDPOINT_HARD_STOP_PENALTY = 0.05" in quality_source
+    assert "ENDPOINT_ENERGY_WEIGHT = 0.02" in quality_source
+    assert "ENDPOINT_PREFERRED_SQUIM_STOI_SLACK = 0.02" in quality_source
+    assert "ENDPOINT_PREFERRED_SQUIM_PESQ_SLACK = 0.03" in quality_source
 
 
 def test_app_rejects_ambiguous_iri_before_frontend_normalization():
@@ -646,7 +661,7 @@ def test_network_local_dual_asr_capability_is_range_bound_for_initial_and_refill
     assert "chunk_artifacts=primary_verification.chunk_artifacts" in (
         intersection_source
     )
-    assert "CASCADE_EVIDENCE_SCHEMA_VERSION = 5" in quality_source
+    assert "CASCADE_EVIDENCE_SCHEMA_VERSION = 6" in quality_source
     assert '"chunk_text_variants"' in quality_source
     assert "local_candidate_has_coverage_eligibility(" in helper_source
     assert "independent_local_results=" in initial_source
@@ -861,15 +876,14 @@ def test_app_reverifies_the_post_join_speed_adjusted_whole_waveform():
     speed_index = assemble_source.index(
         "waveform = _apply_speed(waveform, playback_speed)"
     )
-    finish_index = assemble_source.index(
-        "return finish_audio(waveform, SR, fade_ms=finish_fade_ms)"
-    )
-    assert speed_index < finish_index
-    assert (
-        'finish_fade_ms = 0.0 if count_speech_units("".join(chunks)) <= 6 else 5.0'
-        in assemble_source
+    finish_index = assemble_source.index("waveform = finish_audio(")
+    pcm_index = assemble_source.index(
+        "return pcm16_verification_waveform(waveform)"
     )
-    assert "else 60.0" not in assemble_source
+    assert speed_index < finish_index < pcm_index
+    assert "fade_ms=FINAL_ENDPOINT_FADE_MS" in assemble_source
+    assert "FINAL_ENDPOINT_FADE_MS = 5.0" in source
+    assert "finish_fade_ms" not in assemble_source
     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(
@@ -951,7 +965,7 @@ def test_chunk_generation_closed_loop_rerenders_final_audio_from_raw_once():
     assert "fallback_speed = final_speed * fallback_residual" in generate_source
     assert "_apply_speed(\n        corrected" not in generate_source
     assert "_apply_speed(\n            audio,\n            final_speed" in generate_source
-    assert "_apply_speed(\n                    audio,\n                    fallback_speed" in generate_source
+    assert "audio,\n                        fallback_speed" in generate_source
     assert "PACE_STRETCH_N_FFT = 1536" in source
     assert "PACE_STRETCH_HOP_LENGTH = 384" in source
     assert "NETWORK_PACE_STRETCH_N_FFT = 2048" in source
@@ -962,7 +976,7 @@ def test_chunk_generation_closed_loop_rerenders_final_audio_from_raw_once():
     assert "hop_length=hop_length" in apply_speed_source
 
 
-def test_public_tts_wrappers_serialize_pcm16_after_float_verification():
+def test_public_tts_wrappers_serialize_the_same_pcm16_grid_used_for_verification():
     source = (ROOT / "app.py").read_text(encoding="utf-8")
     tree = ast.parse(source)
     functions = {
@@ -971,8 +985,14 @@ def test_public_tts_wrappers_serialize_pcm16_after_float_verification():
         if isinstance(node, ast.FunctionDef)
     }
     synthesize_source = ast.get_source_segment(source, functions["_synthesize"])
+    assemble_source = ast.get_source_segment(
+        source,
+        functions["_assemble_trajectory_audio"],
+    )
 
     assert synthesize_source is not None
+    assert assemble_source is not None
+    assert "return pcm16_verification_waveform(waveform)" in assemble_source
     assert "pcm16_audio_output" not in synthesize_source
     assert "return SR, waveform" in synthesize_source
     for wrapper_name in ("tts_speaker", "tts_reference", "tts_longform"):