Bind release evidence to verified text variants

app.py

@@ -154,6 +154,8 @@ MAX_TEXT_CHARS = 360
 QUALITY_MAX_GENERATED_CHUNKS = 32
 QUALITY_MAX_GENERATED_TEXT_UNITS = 800
 EMAIL_MAIL_FALLBACK_CANDIDATE_ORDINALS = frozenset((2, 5, 6))
+BASE_CHUNK_TEXT_VARIANT = "base"
+EMAIL_DOMAIN_MAIL_CHUNK_TEXT_VARIANT = "email_domain_mail_v1"
 QUALITY_FINAL_ASR_MAX_NEW_TOKENS = 440
 QUALITY_MAX_CER = 0.20
 QUALITY_MAX_PACE_CPS = 4.30
@@ -1327,7 +1329,26 @@ def _verify_sequence_trajectory_audio(
         f"chunk_candidates={sequence_result.chunk_candidate_indices} "
         f"{_verification_metric_log_fields(assembled_verification)}"
     )
-    return assembled_verification
+    if not assembled_verification.passed:
+        return assembled_verification
+    independent_verification = _verify_independent_whole_audio(
+        waveform,
+        whole_target_text,
+        anchor,
+        independent_cache,
+    )
+    intersected = qualify_trajectory_with_joined_output(
+        assembled_verification,
+        independent_verification,
+    )
+    independent_status = "verified" if independent_verification.passed else "rejected"
+    print(
+        f"[BlueMagpie] sequence path independent large-v3 {independent_status} "
+        f"rank={sequence_result.sequence_path_rank} "
+        f"chunk_candidates={sequence_result.chunk_candidate_indices} "
+        f"{_verification_metric_log_fields(independent_verification)}"
+    )
+    return intersected
 
 
 def _synthesize(
@@ -1451,6 +1472,48 @@ def _synthesize(
             )
         return tuple(transformed)
 
+    def candidate_generation_text_variants(
+        canonical_chunks: tuple[str, ...],
+        generation_context: CandidateGenerationContext,
+    ) -> tuple[str, ...]:
+        expected_canonical = canonical_chunks_for_context(generation_context)
+        if canonical_chunks != expected_canonical:
+            raise ValueError("candidate text variant lacks canonical provenance")
+        generated_chunks = candidate_generation_text_transform(
+            canonical_chunks,
+            generation_context,
+        )
+        variants: list[str] = []
+        for chunk_index, canonical, generated, candidate_ordinal in zip(
+            generation_context.chunk_indices,
+            canonical_chunks,
+            generated_chunks,
+            generation_context.chunk_candidate_ordinals,
+            strict=True,
+        ):
+            if generated == canonical:
+                variants.append(BASE_CHUNK_TEXT_VARIANT)
+                continue
+            if chunk_specs is None:
+                raise ValueError("candidate text variant lacks network proof")
+            spec = chunk_specs[chunk_index]
+            expected_variant = email_domain_mail_generation_variant(
+                canonical,
+                spec.network_fragment_proofs,
+            )
+            if (
+                candidate_ordinal not in EMAIL_MAIL_FALLBACK_CANDIDATE_ORDINALS
+                or generated != expected_variant
+                or generated == canonical
+                or not any(
+                    proof.identifier_kind == "email"
+                    for proof in spec.network_fragment_proofs
+                )
+            ):
+                raise ValueError("candidate text variant is not proof-bound")
+            variants.append(EMAIL_DOMAIN_MAIL_CHUNK_TEXT_VARIANT)
+        return tuple(variants)
+
     def generation_chunk_specs(
         seed: int,
         candidate_chunks: tuple[str, ...],
@@ -1615,6 +1678,17 @@ def _synthesize(
             candidate_chunks,
             generation_context=generation_context,
         )
+        canonical_chunks = canonical_chunks_for_context(generation_context)
+        expected_generation_chunks = candidate_generation_text_transform(
+            canonical_chunks,
+            generation_context,
+        )
+        if candidate_chunks != expected_generation_chunks:
+            raise ValueError("generation evidence text disagrees with its schedule")
+        text_variants = candidate_generation_text_variants(
+            canonical_chunks,
+            generation_context,
+        )
         rows = tuple(
             chunk_cfg_evidence(
                 chunk,
@@ -1637,6 +1711,7 @@ def _synthesize(
             floor_reasons=tuple(row[1] for row in rows),
             chunk_candidate_ordinals=candidate_ordinals,
             network_conditioned=network_flags,
+            chunk_text_variants=text_variants,
         )
 
     try:

quality_runtime.py

@@ -61,7 +61,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 = 4
+CASCADE_EVIDENCE_SCHEMA_VERSION = 5
 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]
@@ -1583,6 +1583,7 @@ class CandidateGenerationEvidence:
     floor_reasons: tuple[tuple[str, ...], ...]
     chunk_candidate_ordinals: tuple[int, ...] = ()
     network_conditioned: tuple[bool, ...] = ()
+    chunk_text_variants: tuple[str, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -1600,6 +1601,7 @@ class CandidateAttemptEvidence:
     chunk_text_units: tuple[int, ...]
     chunk_candidate_ordinals: tuple[int, ...]
     network_conditioned: tuple[bool, ...]
+    chunk_text_variants: tuple[str, ...]
     scheduled_cfg: float | None
     effective_cfgs: tuple[float, ...]
     floor_reasons: tuple[tuple[str, ...], ...]
@@ -2526,6 +2528,7 @@ def _candidate_attempt_evidence_payload(
         "chunk_text_units": list(evidence.chunk_text_units),
         "chunk_candidate_ordinals": list(candidate_ordinals),
         "chunk_policies": chunk_policies,
+        "chunk_text_variants": list(evidence.chunk_text_variants),
         "network_conditioned": list(evidence.network_conditioned),
         "scheduled_cfg": evidence.scheduled_cfg,
         "effective_cfgs": list(evidence.effective_cfgs),
@@ -2615,6 +2618,7 @@ def _selected_generation_evidence_payload(
     candidate_ordinals: list[int | None] = []
     policies: list[str | None] = []
     network_conditioned: list[bool | None] = []
+    chunk_text_variants: list[str | None] = []
     complete = True
     for chunk_index, candidate_index in enumerate(
         selection.chunk_candidate_indices
@@ -2628,6 +2632,7 @@ def _selected_generation_evidence_payload(
             candidate_ordinals.append(None)
             policies.append(None)
             network_conditioned.append(None)
+            chunk_text_variants.append(None)
             continue
         try:
             local_index = attempt.chunk_indices.index(chunk_index)
@@ -2641,6 +2646,7 @@ def _selected_generation_evidence_payload(
             candidate_ordinals.append(None)
             policies.append(None)
             network_conditioned.append(None)
+            chunk_text_variants.append(None)
             continue
         try:
             ordinal = attempt.chunk_candidate_ordinals[local_index]
@@ -2661,6 +2667,11 @@ def _selected_generation_evidence_payload(
         except IndexError:
             complete = False
             network_conditioned.append(None)
+        try:
+            chunk_text_variants.append(attempt.chunk_text_variants[local_index])
+        except IndexError:
+            complete = False
+            chunk_text_variants.append(None)
     return {
         "complete": complete,
         "chunk_scheduled_cfgs": scheduled_cfgs,
@@ -2669,6 +2680,7 @@ def _selected_generation_evidence_payload(
         "chunk_candidate_ordinals": candidate_ordinals,
         "chunk_policies": policies,
         "network_conditioned": network_conditioned,
+        "chunk_text_variants": chunk_text_variants,
     }
 
 
@@ -2753,6 +2765,7 @@ def format_cascade_evidence_log(
         and len(attempt.chunk_indices) == len(attempt.chunk_text_units)
         == len(attempt.chunk_candidate_ordinals)
         == len(attempt.network_conditioned)
+        == len(attempt.chunk_text_variants)
         == len(attempt.effective_cfgs)
         == len(attempt.floor_reasons)
         and bool(attempt.chunk_indices)
@@ -3298,6 +3311,21 @@ def _validated_candidate_generation_evidence(
         network_conditioned = value.network_conditioned
     else:
         network_conditioned = (False,) * len(chunks)
+    if value.chunk_text_variants:
+        if (
+            not isinstance(value.chunk_text_variants, tuple)
+            or len(value.chunk_text_variants) != len(chunks)
+            or any(
+                variant not in {"base", "email_domain_mail_v1"}
+                for variant in value.chunk_text_variants
+            )
+        ):
+            raise ValueError(
+                "generation evidence text variants do not match the attempt"
+            )
+        chunk_text_variants = value.chunk_text_variants
+    else:
+        chunk_text_variants = ("base",) * len(chunks)
     raw_ordinals = value.chunk_candidate_ordinals
     if not raw_ordinals:
         if require_explicit_candidate_ordinals:
@@ -3377,6 +3405,7 @@ def _validated_candidate_generation_evidence(
         floor_reasons=tuple(floor_reasons),
         chunk_candidate_ordinals=candidate_ordinals,
         network_conditioned=network_conditioned,
+        chunk_text_variants=chunk_text_variants,
     )
 
 
@@ -3415,6 +3444,9 @@ def _candidate_attempt_evidence(
         network_conditioned=(
             generation.network_conditioned if generation is not None else ()
         ),
+        chunk_text_variants=(
+            generation.chunk_text_variants 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 ()),

tests/test_coverage_adaptive.py

@@ -567,7 +567,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"] == 4
+    assert payload["schema_version"] == 5
     assert payload["generation_evidence_complete"] is True
     assert payload["request_chunk_count"] == 2
     assert payload["generated_chunk_count"] == 3
@@ -589,6 +589,7 @@ def test_refill_budget_accounts_exact_generated_chunks_and_text_units():
         "chunk_candidate_ordinals": [1, 0],
         "chunk_policies": ["safe_duration", "base"],
         "network_conditioned": [False, False],
+        "chunk_text_variants": ["base", "base"],
     }
 
 
@@ -1221,6 +1222,91 @@ def test_remaining_budget_prioritizes_request_endpoints_after_full_coverage():
     assert result.chunk_candidate_counts == (1, 1, 2)
 
 
+def test_exact_asr_speaker_near_miss_gets_priority_then_latest_result_reorders():
+    chunks = ("第一段", "第二段")
+    generated = []
+
+    def verification(candidate_chunks, states):
+        observations = []
+        artifacts = []
+        for chunk, state in zip(candidate_chunks, states, strict=True):
+            semantic_passed = state != "semantic_failure"
+            speaker_similarity = 0.05 if state == "speaker_near_miss" else 0.80
+            observations.append(
+                CandidateObservation(
+                    target_text=chunk,
+                    transcript_text=chunk if semantic_passed else "完全錯誤",
+                    audio_duration_seconds=2.0,
+                    speaker_similarity=speaker_similarity,
+                    begin_speaker_similarity=speaker_similarity,
+                    end_speaker_similarity=speaker_similarity,
+                    pace_cps=2.0,
+                )
+            )
+            artifacts.append(
+                ChunkCandidateArtifact(
+                    speaker_embedding=np.array([1.0, 0.0], dtype=np.float32),
+                    rms_db=-20.0,
+                )
+            )
+        return verify_trajectory(
+            observations,
+            chunk_artifacts=artifacts,
+            max_pace_cps=4.3,
+            min_speaker_similarity=0.10,
+        )
+
+    initial_local = verification(
+        chunks,
+        ("semantic_failure", "speaker_near_miss"),
+    )
+    assert initial_local.candidate_results[0].comparison.passed is False
+    assert initial_local.candidate_results[1].comparison.passed is True
+    assert "speaker_similarity" in initial_local.candidate_results[
+        1
+    ].rejection_reasons
+
+    def generator(candidate_chunks, seed):
+        generated.append((candidate_chunks, seed))
+        return tuple(f"{seed}:{chunk}" for chunk in candidate_chunks)
+
+    def refill_verifier(trajectory, candidate_chunks, seed):
+        state = (
+            "semantic_failure"
+            if candidate_chunks == ("第二段",) and seed == 101
+            else "passed"
+        )
+        return verification(candidate_chunks, (state,))
+
+    result = run_coverage_adaptive_cascade(
+        chunks,
+        100,
+        generator,
+        lambda trajectory, candidate_chunks, seed: TrajectoryGateResult(
+            passed=False,
+            candidate_results=initial_local.candidate_results,
+            score=math.inf,
+            rejection_reasons=("joined_output:semantic_gate",),
+            chunk_artifacts=initial_local.chunk_artifacts,
+        ),
+        refill_verifier,
+        sequence_final_verifier=lambda result, candidate_chunks: _exact_final(
+            candidate_chunks
+        ),
+        max_generated_chunks=5,
+        max_generated_text_units=100,
+        max_sequence_paths=1,
+    )
+
+    assert generated[1:] == [
+        (("第二段",), 101),
+        (("第一段",), 102),
+        (("第二段",), 103),
+    ]
+    assert result.chunk_candidate_indices == (2, 3)
+    assert result.chunk_candidate_counts == (1, 1)
+
+
 def test_ragged_dp_accepts_unequal_row_widths_and_stable_cost_order():
     ranked = select_culprit_diverse_candidate_sequences(
         [[0.0], [0.3, 0.1, 0.2], [0.4, 0.0]],

tests/test_production.py

@@ -1054,6 +1054,37 @@ def test_network_request_bounded_ordinary_envelope_splits_long_final_clause():
     assert "".join(spec.text for spec in specs) == target
 
 
+@pytest.mark.parametrize(
+    ("terminal_units", "expected_units"),
+    (
+        (36, (14, 36)),
+        (37, (14, 25, 12)),
+    ),
+)
+def test_network_request_terminal_ordinary_clause_splits_only_above_36_units(
+    terminal_units,
+    expected_units,
+):
+    raw = "請寄到 patrol@forestmail.tw。" + "甲" * terminal_units + "。"
+    target = normalize_spoken_forms(raw)
+
+    specs = plan_generation_chunks(raw, target, ordinary_max_units=36)
+
+    assert tuple(count_speech_units(spec.text) for spec in specs) == expected_units
+    assert specs[0].network_conditioned
+    assert all(not spec.network_conditioned for spec in specs[1:])
+    assert all(
+        left.source_end == right.source_start
+        for left, right in zip(specs, specs[1:], strict=False)
+    )
+    assert specs[0].source_start == 0
+    assert specs[-1].source_end == len(target)
+    assert all(spec.boundary_after == "semantic" for spec in specs[:-1])
+    assert specs[-1].boundary_after == "none"
+    assert max(count_speech_units(spec.text) for spec in specs) <= 36
+    assert "".join(spec.text for spec in specs) == target
+
+
 def test_generation_network_planner_prefers_context_before_identifier_start():
     raw = (
         "途中若發現落石或樹枝阻斷通行，請拍照並寄到 "

tests/test_quality_runtime.py

@@ -1675,7 +1675,7 @@ def test_canonical_attempt_evidence_distinguishes_local_v3_pass_and_not_run():
     )
     attempt = payload["attempts"][0]
 
-    assert payload["schema_version"] == 4
+    assert payload["schema_version"] == 5
     assert attempt["independent_local_evidence_complete"] is True
     assert attempt["independent_local_results"] == [
         {

tests/test_release_pins.py

@@ -3,6 +3,16 @@ import hashlib
 from pathlib import Path
 from types import SimpleNamespace
 
+import numpy as np
+
+from production import (
+    GenerationChunkSpec,
+    count_speech_units,
+    fade_variable_internal_edges,
+    join_audio_chunks_variable,
+    match_chunk_rms,
+    punctuation_pause_seconds,
+)
 from quality_runtime import LocalIndependentGateEvidence
 
 
@@ -51,6 +61,59 @@ def _literal_constants(path: Path) -> dict[str, object]:
     return values
 
 
+def _isolated_assemble_trajectory_audio():
+    """Execute only the production assembly function without loading the model."""
+
+    app_path = ROOT / "app.py"
+    tree = ast.parse(app_path.read_text(encoding="utf-8"))
+    function = next(
+        node
+        for node in tree.body
+        if isinstance(node, ast.FunctionDef)
+        and node.name == "_assemble_trajectory_audio"
+    )
+    module = ast.Module(
+        body=[
+            ast.ImportFrom(
+                module="__future__",
+                names=[ast.alias(name="annotations")],
+                level=0,
+            ),
+            function,
+        ],
+        type_ignores=[],
+    )
+    ast.fix_missing_locations(module)
+
+    constants = _literal_constants(app_path)
+    namespace = {
+        "np": np,
+        "GenerationChunkSpec": GenerationChunkSpec,
+        "SR": 1_000,
+        "CHUNK_RMS_MATCH_DB": constants["CHUNK_RMS_MATCH_DB"],
+        "SEMANTIC_CHUNK_MIN_SILENCE_MS": constants[
+            "SEMANTIC_CHUNK_MIN_SILENCE_MS"
+        ],
+        "NETWORK_REQUEST_SEMANTIC_CHUNK_MIN_SILENCE_MS": constants[
+            "NETWORK_REQUEST_SEMANTIC_CHUNK_MIN_SILENCE_MS"
+        ],
+        "NETWORK_INTERNAL_SILENCE_MS": constants["NETWORK_INTERNAL_SILENCE_MS"],
+        "NETWORK_INTERNAL_FADE_MS": constants["NETWORK_INTERNAL_FADE_MS"],
+        "CHUNK_EDGE_FADE_MS": constants["CHUNK_EDGE_FADE_MS"],
+        "CROSSFADE_MS": constants["CROSSFADE_MS"],
+        "match_chunk_rms": match_chunk_rms,
+        "punctuation_pause_seconds": punctuation_pause_seconds,
+        "fade_variable_internal_edges": fade_variable_internal_edges,
+        "join_audio_chunks_variable": join_audio_chunks_variable,
+        "apply_loudness_floor": lambda waveform, **_kwargs: waveform,
+        "_apply_speed": lambda waveform, _speed: waveform,
+        "count_speech_units": count_speech_units,
+        "finish_audio": lambda waveform, _sample_rate, **_kwargs: waveform,
+    }
+    exec(compile(module, str(app_path), "exec"), namespace)
+    return namespace["_assemble_trajectory_audio"]
+
+
 def _isolated_load_speakers(*, metadata: dict, speaker_ids: tuple[str, ...]):
     """Execute only ``_load_speakers`` without importing the GPU application."""
 
@@ -522,7 +585,8 @@ 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 = 4" in quality_source
+    assert "CASCADE_EVIDENCE_SCHEMA_VERSION = 5" 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
     assert "independent_local_results=" in refill_source
@@ -1023,8 +1087,19 @@ def test_space_hard_intersects_dual_asr_on_whole_and_proven_network_locals():
     sequence_large_v3_transcriber = sequence_source.index(
         "transcriber=transcribe_verification_whisper"
     )
-    assert sequence_assemble < sequence_large_v3 < sequence_large_v3_transcriber
-    assert "_verify_independent_whole_audio(" not in sequence_source
+    sequence_independent = sequence_source.index(
+        "independent_verification = _verify_independent_whole_audio("
+    )
+    sequence_intersection = sequence_source.index(
+        "intersected = qualify_trajectory_with_joined_output("
+    )
+    assert (
+        sequence_assemble
+        < sequence_large_v3
+        < sequence_large_v3_transcriber
+        < sequence_independent
+        < sequence_intersection
+    )
 
     cache_create = synthesize_source.index(
         "independent_cache = WholeWaveformVerificationCache()"
@@ -1306,6 +1381,88 @@ def test_space_locks_validated_nfe_and_bounds_total_generation_work():
     assert "boundary-only relaxation" in readme
 
 
+def test_pace_only_fallback_guard_starts_at_30_ordinary_units_not_29():
+    app_path = ROOT / "app.py"
+    constants = _literal_constants(app_path)
+    tree = ast.parse(app_path.read_text(encoding="utf-8"))
+    generate_chunk = next(
+        node
+        for node in tree.body
+        if isinstance(node, ast.FunctionDef) and node.name == "_generate_chunk"
+    )
+    guarded_ifs = [
+        node
+        for node in ast.walk(generate_chunk)
+        if isinstance(node, ast.If)
+        and any(
+            isinstance(name, ast.Name)
+            and name.id == "PACE_ONLY_FALLBACK_MIN_UNITS"
+            for name in ast.walk(node.test)
+        )
+    ]
+
+    assert constants["PACE_ONLY_FALLBACK_MIN_UNITS"] == 30
+    assert len(guarded_ifs) == 1
+    expression = ast.Expression(body=guarded_ifs[0].test)
+    ast.fix_missing_locations(expression)
+    guard = compile(expression, str(app_path), "eval")
+
+    def eligible(units, *, network_conditioned=False):
+        return eval(
+            guard,
+            {
+                "PACE_ONLY_FALLBACK_MIN_UNITS": constants[
+                    "PACE_ONLY_FALLBACK_MIN_UNITS"
+                ],
+                "count_speech_units": lambda _text: units,
+                "network_conditioned": network_conditioned,
+                "text": "測試",
+            },
+        )
+
+    assert eligible(29) is False
+    assert eligible(30) is True
+    assert eligible(30, network_conditioned=True) is False
+
+
+def test_assembled_waveform_uses_250ms_ordinary_and_350ms_network_request_pause():
+    assemble = _isolated_assemble_trajectory_audio()
+    chunks = ("第一段，", "第二段")
+    trajectory = (
+        np.ones(100, dtype=np.float32),
+        np.ones(100, dtype=np.float32),
+    )
+
+    ordinary = assemble(trajectory, chunks, 1.0)
+
+    assert ordinary.shape == (450,)
+    np.testing.assert_array_equal(ordinary[100:350], np.zeros(250, dtype=np.float32))
+
+    first_end = len(chunks[0])
+    network_specs = (
+        GenerationChunkSpec(
+            text=chunks[0],
+            source_start=0,
+            source_end=first_end,
+            boundary_after="semantic",
+        ),
+        GenerationChunkSpec(
+            text=chunks[1],
+            source_start=first_end,
+            source_end=first_end + len(chunks[1]),
+            network_span_indices=(0,),
+            boundary_after="none",
+        ),
+    )
+    network_request = assemble(trajectory, chunks, 1.0, network_specs)
+
+    assert network_request.shape == (550,)
+    np.testing.assert_array_equal(
+        network_request[100:450],
+        np.zeros(350, dtype=np.float32),
+    )
+
+
 def test_space_locks_waveform_only_bounded_whisper_segmentation():
     source = (ROOT / "quality_runtime.py").read_text(encoding="utf-8")
     readme = (ROOT / "README.md").read_text(encoding="utf-8")