Add candidate-local generation text transforms

quality_runtime.py

@@ -4083,6 +4083,13 @@ def run_coverage_adaptive_cascade(
         TrajectoryGateResult,
     ],
     generation_evidence_factory: Callable[..., CandidateGenerationEvidence] | None = None,
+    candidate_generation_text_transform: (
+        Callable[
+            [tuple[str, ...], CandidateGenerationContext],
+            Sequence[str],
+        ]
+        | None
+    ) = None,
     max_generated_chunks: int = 32,
     max_generated_text_units: int = 800,
     max_sequence_paths: int = 3,
@@ -4105,6 +4112,13 @@ def run_coverage_adaptive_cascade(
     that explicitly accepts the keyword-only ``generation_context`` opts into
     per-chunk refill scheduling.  Its evidence factory must accept the same
     keyword and report the supplied row-local candidate ordinals.
+
+    ``candidate_generation_text_transform`` may derive candidate-local text
+    from the canonical verifier targets and immutable generation context.  The
+    transformed chunks are sent only to the generator and generation-evidence
+    factory.  Semantic verifiers continue to receive the canonical chunks.
+    Generated-text-unit accounting uses the transformed text and is checked
+    before every generator call.
     """
 
     if not all(
@@ -4121,6 +4135,10 @@ def run_coverage_adaptive_cascade(
         generation_evidence_factory
     ):
         raise ValueError("generation_evidence_factory must be callable")
+    if candidate_generation_text_transform is not None and not callable(
+        candidate_generation_text_transform
+    ):
+        raise ValueError("candidate_generation_text_transform must be callable")
 
     def accepts_generation_context(callback: Callable[..., Any]) -> bool:
         try:
@@ -4149,6 +4167,46 @@ def run_coverage_adaptive_cascade(
         raise ValueError(
             "context-aware generation and evidence callbacks must opt in together"
         )
+
+    def generation_chunks(
+        canonical_chunks: tuple[str, ...],
+        context: CandidateGenerationContext,
+    ) -> tuple[str, ...]:
+        if candidate_generation_text_transform is None:
+            return canonical_chunks
+        try:
+            transformed = candidate_generation_text_transform(
+                canonical_chunks,
+                context,
+            )
+        except Exception as error:
+            raise RuntimeError(
+                "candidate generation text transform failed"
+            ) from error
+        if isinstance(transformed, (str, bytes, bytearray, np.ndarray)):
+            raise RuntimeError(
+                "candidate generation text transform returned invalid chunks"
+            )
+        try:
+            normalized = tuple(transformed)
+        except TypeError as error:
+            raise RuntimeError(
+                "candidate generation text transform returned invalid chunks"
+            ) from error
+        if (
+            len(normalized) != len(canonical_chunks)
+            or any(
+                not isinstance(chunk, str)
+                or not chunk
+                or count_speech_units(chunk) <= 0
+                for chunk in normalized
+            )
+        ):
+            raise RuntimeError(
+                "candidate generation text transform returned invalid chunks"
+            )
+        return normalized
+
     try:
         chunk_tuple = tuple(str(chunk) for chunk in chunks)
     except TypeError as error:
@@ -4257,7 +4315,6 @@ def run_coverage_adaptive_cascade(
     )
 
     chunk_units = tuple(count_speech_units(chunk) for chunk in chunk_tuple)
-    initial_units = sum(chunk_units)
     chunk_count = len(chunk_tuple)
     # A single local chunk still receives a fresh exact whole-waveform speaker
     # measurement before return.  Allow the same narrowly bounded local proxy
@@ -4265,6 +4322,19 @@ def run_coverage_adaptive_cascade(
     dp_boundary_limit = boundary_limit
     if chunk_count > chunk_budget:
         raise ValueError("initial trajectory exceeds the generated-chunk budget")
+    initial_context = CandidateGenerationContext(
+        candidate_index=0,
+        seed=base_seed,
+        chunk_indices=tuple(range(chunk_count)),
+        chunk_candidate_ordinals=(0,) * chunk_count,
+    )
+    initial_generation_chunks = generation_chunks(
+        chunk_tuple,
+        initial_context,
+    )
+    initial_units = sum(
+        count_speech_units(chunk) for chunk in initial_generation_chunks
+    )
     if initial_units > text_budget:
         raise ValueError("initial trajectory exceeds the generated-text-unit budget")
 
@@ -4309,21 +4379,18 @@ def run_coverage_adaptive_cascade(
         except Exception as error:
             raise RuntimeError("candidate generation evidence is invalid") from error
 
-    initial_context = CandidateGenerationContext(
-        candidate_index=0,
-        seed=base_seed,
-        chunk_indices=tuple(range(chunk_count)),
-        chunk_candidate_ordinals=(0,) * chunk_count,
-    )
     try:
         if context_aware_generator:
             raw_initial_trajectory = candidate_generator(
-                chunk_tuple,
+                initial_generation_chunks,
                 base_seed,
                 generation_context=initial_context,
             )
         else:
-            raw_initial_trajectory = candidate_generator(chunk_tuple, base_seed)
+            raw_initial_trajectory = candidate_generator(
+                initial_generation_chunks,
+                base_seed,
+            )
     except Exception as error:
         raise RuntimeError("initial trajectory generation failed") from error
     initial_trajectory = _coverage_trajectory_tuple(
@@ -4332,7 +4399,7 @@ def run_coverage_adaptive_cascade(
     )
     initial_generation_evidence = generation_evidence(
         initial_context,
-        chunk_tuple,
+        initial_generation_chunks,
     )
     attempted_seeds.append(base_seed)
     try:
@@ -4441,7 +4508,7 @@ def run_coverage_adaptive_cascade(
     refill_attempts = [0] * chunk_count
     next_candidate_index = 1
     while generated_chunks < chunk_budget:
-        if any(
+        if candidate_generation_text_transform is None and any(
             not pool and generated_units + chunk_units[index] > text_budget
             for index, pool in enumerate(pools)
         ):
@@ -4452,7 +4519,7 @@ def run_coverage_adaptive_cascade(
             for candidate in pool
         )
         needs_first_alternative = initial_all_strict and not has_noninitial_coverage
-        eligible = [
+        structurally_eligible = [
             index
             for index, pool in enumerate(pools)
             if (
@@ -4460,45 +4527,88 @@ def run_coverage_adaptive_cascade(
                 or len(pool) < path_limit
                 or needs_first_alternative
             )
-            and generated_units + chunk_units[index] <= text_budget
         ]
+        if candidate_generation_text_transform is None:
+            eligible = [
+                index
+                for index in structurally_eligible
+                if generated_units + chunk_units[index] <= text_budget
+            ]
+        else:
+            eligible = structurally_eligible
         if not eligible:
             break
-        chunk_index = min(
-            eligible,
-            key=lambda index: (
+
+        def refill_priority(index: int) -> tuple[int, int, int]:
+            return (
                 len(pools[index]),
                 refill_attempts[index],
                 index,
-            ),
-        )
-        seed = base_seed + next_candidate_index
+            )
+
+        refill_proposal = None
+        for proposed_index in sorted(eligible, key=refill_priority):
+            proposed_seed = base_seed + next_candidate_index
+            proposed_canonical_chunks = (chunk_tuple[proposed_index],)
+            proposed_context = CandidateGenerationContext(
+                candidate_index=next_candidate_index,
+                seed=proposed_seed,
+                chunk_indices=(proposed_index,),
+                chunk_candidate_ordinals=(
+                    refill_attempts[proposed_index] + 1,
+                ),
+            )
+            proposed_generation_chunks = generation_chunks(
+                proposed_canonical_chunks,
+                proposed_context,
+            )
+            proposed_units = sum(
+                count_speech_units(chunk)
+                for chunk in proposed_generation_chunks
+            )
+            if generated_units + proposed_units <= text_budget:
+                refill_proposal = (
+                    proposed_index,
+                    proposed_seed,
+                    proposed_generation_chunks,
+                    proposed_context,
+                    proposed_units,
+                )
+                break
+        if refill_proposal is None:
+            break
+        (
+            chunk_index,
+            seed,
+            refill_generation_chunks,
+            refill_context,
+            refill_units,
+        ) = refill_proposal
         refill_chunks = (chunk_tuple[chunk_index],)
-        refill_context = CandidateGenerationContext(
-            candidate_index=next_candidate_index,
-            seed=seed,
-            chunk_indices=(chunk_index,),
-            chunk_candidate_ordinals=(refill_attempts[chunk_index] + 1,),
-        )
+        if generated_chunks + len(refill_generation_chunks) > chunk_budget:
+            break
         try:
             if context_aware_generator:
                 raw_refill_trajectory = candidate_generator(
-                    refill_chunks,
+                    refill_generation_chunks,
                     seed,
                     generation_context=refill_context,
                 )
             else:
-                raw_refill_trajectory = candidate_generator(refill_chunks, seed)
+                raw_refill_trajectory = candidate_generator(
+                    refill_generation_chunks,
+                    seed,
+                )
         except Exception as error:
             raise RuntimeError("chunk refill generation failed") from error
         refill_trajectory = _coverage_trajectory_tuple(raw_refill_trajectory, 1)
         refill_generation_evidence = generation_evidence(
             refill_context,
-            refill_chunks,
+            refill_generation_chunks,
         )
         attempted_seeds.append(seed)
         generated_chunks += 1
-        generated_units += chunk_units[chunk_index]
+        generated_units += refill_units
         refill_attempts[chunk_index] += 1
         try:
             raw_refill_verification = refill_candidate_verifier(

tests/test_coverage_adaptive.py

@@ -4,6 +4,7 @@ import math
 import numpy as np
 import pytest
 
+from production import count_speech_units
 from quality_runtime import (
     CASCADE_EVIDENCE_LOG_PREFIX,
     CandidateGenerationContext,
@@ -861,6 +862,166 @@ def test_context_aware_refills_rotate_cfg_and_endpoint_policy_per_chunk():
     ]
 
 
+def test_candidate_local_generation_text_is_separate_from_canonical_verification():
+    chunks = ("甲", "乙")
+    transformed = []
+    generated = []
+    evidenced = []
+    verified = []
+
+    def transform(candidate_chunks, generation_context):
+        assert isinstance(generation_context, CandidateGenerationContext)
+        transformed.append((candidate_chunks, generation_context))
+        if generation_context.candidate_index == 0:
+            return ("甲甲", "乙乙")
+        if generation_context.chunk_indices == (0,):
+            return ("甲甲甲甲",)
+        return ("乙乙乙",)
+
+    def generator(candidate_chunks, seed, *, generation_context):
+        generated.append((candidate_chunks, seed, generation_context))
+        return candidate_chunks
+
+    def generation_evidence(
+        candidate_index,
+        seed,
+        chunk_indices,
+        candidate_chunks,
+        *,
+        generation_context,
+    ):
+        evidenced.append((candidate_chunks, generation_context))
+        ordinal = generation_context.chunk_candidate_ordinals[0]
+        assert all(
+            value == ordinal
+            for value in generation_context.chunk_candidate_ordinals
+        )
+        scheduled = generation_cfg_for_candidate_offset(ordinal)
+        return CandidateGenerationEvidence(
+            chunk_indices=chunk_indices,
+            chunk_text_units=tuple(
+                count_speech_units(chunk) for chunk in candidate_chunks
+            ),
+            scheduled_cfg=scheduled,
+            effective_cfgs=tuple(scheduled for _ in candidate_chunks),
+            floor_reasons=tuple(() for _ in candidate_chunks),
+            chunk_candidate_ordinals=(
+                generation_context.chunk_candidate_ordinals
+            ),
+        )
+
+    def whole_verifier(trajectory, candidate_chunks, seed):
+        verified.append(("whole", candidate_chunks))
+        return _joined_rejected_local(candidate_chunks)
+
+    def refill_verifier(trajectory, candidate_chunks, seed):
+        verified.append(("refill", candidate_chunks))
+        return _local_verification(candidate_chunks)
+
+    def final_verifier(result, candidate_chunks):
+        verified.append(("final", candidate_chunks))
+        return _exact_final(candidate_chunks)
+
+    result = run_coverage_adaptive_cascade(
+        chunks,
+        10,
+        generator,
+        whole_verifier,
+        refill_verifier,
+        sequence_final_verifier=final_verifier,
+        generation_evidence_factory=generation_evidence,
+        candidate_generation_text_transform=transform,
+        max_generated_chunks=3,
+        max_generated_text_units=7,
+        max_sequence_paths=1,
+    )
+
+    assert [(values, seed) for values, seed, _ in generated] == [
+        (("甲甲", "乙乙"), 10),
+        (("乙乙乙",), 11),
+    ]
+    assert [values for values, _ in evidenced] == [
+        ("甲甲", "乙乙"),
+        ("乙乙乙",),
+    ]
+    assert verified == [
+        ("whole", chunks),
+        ("refill", ("乙",)),
+        ("final", chunks),
+    ]
+    assert [context.chunk_indices for _, context in transformed] == [
+        (0, 1),
+        (0,),
+        (1,),
+    ]
+    assert result.trajectory == ("甲甲", "乙乙乙")
+    assert result.chunk_candidate_indices == (0, 1)
+    assert result.generated_chunk_count == 3
+    assert result.generated_text_units == 7
+    assert result.diagnostics.attempts[0].chunk_text_units == (2, 2)
+    assert result.diagnostics.attempts[1].chunk_text_units == (3,)
+
+
+def test_generation_evidence_must_report_transformed_text_units():
+    def evidence(candidate_index, seed, chunk_indices, candidate_chunks):
+        return CandidateGenerationEvidence(
+            chunk_indices=chunk_indices,
+            chunk_text_units=(1,),
+            scheduled_cfg=3.0,
+            effective_cfgs=(3.0,),
+            floor_reasons=((),),
+        )
+
+    with pytest.raises(
+        RuntimeError,
+        match="candidate generation evidence is invalid",
+    ):
+        run_coverage_adaptive_cascade(
+            ("甲",),
+            10,
+            lambda candidate_chunks, seed: candidate_chunks,
+            lambda trajectory, candidate_chunks, seed: _local_verification(
+                candidate_chunks
+            ),
+            lambda trajectory, candidate_chunks, seed: _local_verification(
+                candidate_chunks
+            ),
+            sequence_final_verifier=lambda result, candidate_chunks: (
+                _exact_final(candidate_chunks)
+            ),
+            generation_evidence_factory=evidence,
+            candidate_generation_text_transform=(
+                lambda candidate_chunks, context: ("甲甲",)
+            ),
+        )
+
+
+@pytest.mark.parametrize(
+    "transform",
+    [
+        lambda chunks, context: "甲",
+        lambda chunks, context: ("甲", "乙"),
+        lambda chunks, context: ("",),
+        lambda chunks, context: (object(),),
+    ],
+)
+def test_candidate_generation_text_transform_must_return_aligned_speech_chunks(
+    transform,
+):
+    generated = []
+    with pytest.raises(RuntimeError, match="transform returned invalid chunks"):
+        run_coverage_adaptive_cascade(
+            ("甲",),
+            10,
+            lambda chunks, seed: generated.append((chunks, seed)),
+            lambda trajectory, chunks, seed: _local_verification(chunks),
+            lambda trajectory, chunks, seed: _local_verification(chunks),
+            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
+            candidate_generation_text_transform=transform,
+        )
+    assert generated == []
+
+
 def test_context_generation_and_evidence_must_opt_in_together_and_match():
     def context_generator(chunks, seed, *, generation_context):
         return tuple(chunks)

tests/test_quality_runtime.py

@@ -3070,6 +3070,40 @@ def test_network_endpoint_units_do_not_change_k32_or_800_work_budget():
     assert result.diagnostics.attempts[0].chunk_text_units == (public_units,)
 
 
+def test_transformed_initial_text_budget_is_checked_before_generation():
+    generated = []
+    verified = []
+    contexts = []
+
+    def transform(chunks, generation_context):
+        contexts.append(generation_context)
+        return ("甲" * 801,)
+
+    with pytest.raises(
+        ValueError,
+        match="initial trajectory exceeds the generated-text-unit budget",
+    ):
+        run_coverage_adaptive_cascade(
+            ("甲",),
+            902,
+            lambda chunks, seed: generated.append((chunks, seed)),
+            lambda *args: verified.append(("whole", args)),
+            lambda *args: verified.append(("refill", args)),
+            sequence_final_verifier=lambda *args: verified.append(
+                ("final", args)
+            ),
+            candidate_generation_text_transform=transform,
+            max_generated_chunks=32,
+            max_generated_text_units=800,
+        )
+
+    assert len(contexts) == 1
+    assert contexts[0].candidate_index == 0
+    assert contexts[0].chunk_indices == (0,)
+    assert generated == []
+    assert verified == []
+
+
 def test_candidate_limit_requires_complete_positive_text_unit_budget():
     with pytest.raises(ValueError, match="provided together"):
         candidate_limit_for_chunk_budget(1, total_text_units=40)