BlueMagpie-TTS-Demo / quality_runtime.py
voidful's picture
Use Breeze25 for efficient semantic verification
86b8987 verified
Raw
History Blame Contribute Delete
244 kB
"""Fail-closed quality runtime helpers for the BlueMagpie-TTS Space.
The module deliberately has no import-time model downloads. Breeze ASR 25 and
ECAPA are supplied through small injectable boundaries so the online Space can
load the real models lazily while unit tests remain deterministic and offline.
"""
from __future__ import annotations
import hashlib
import inspect
import json
import math
import operator
import threading
from dataclasses import dataclass, replace
from math import gcd
from pathlib import Path
from typing import Any, Callable, Sequence
import numpy as np
import torch
import torch.nn.functional as torch_functional
from latency_timing import timed_latency_stage
from production import (
AsrComparison,
CandidateSequenceSelection,
compare_asr_text,
count_speech_units,
)
BREEZE25_MODEL_ID = "MediaTek-Research/Breeze-ASR-25"
BREEZE25_REVISION = "cffe7ccb404d025296a00758d0a33468bec3a9d0"
BREEZE25_WEIGHT_SHA256 = (
"c5d952b3bc03ea277209aff0ef5b5c4c055d74449ff794c02d8f4e315fdef6b6"
)
BREEZE25_ATTENTION_IMPLEMENTATION = "eager"
BREEZE25_RETURN_ATTENTION_MASK = True
BREEZE25_LANGUAGE = "zh"
BREEZE25_TASK = "transcribe"
BREEZE25_SAMPLE_RATE = 16_000
BREEZE25_MAX_SEGMENT_SECONDS = 28.0
BREEZE25_HARD_MAX_SEGMENT_SECONDS = 30.0
BREEZE25_MIN_SEGMENT_SECONDS = 1.25
BREEZE25_MIN_PAUSE_SECONDS = 0.25
BREEZE25_MAX_VERIFICATION_SEGMENTS = 12
GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS = 160
BREEZE25_MAX_MICROBATCH_SEGMENTS = 6
# Compatibility names remain available while app/evaluator callers migrate.
# They deliberately resolve to the one pinned Breeze ASR 25 contract and never
# refer to the retired OpenAI verifier models.
WHISPER_MODEL_ID = BREEZE25_MODEL_ID
WHISPER_REVISION = BREEZE25_REVISION
VERIFICATION_WHISPER_MODEL_ID = BREEZE25_MODEL_ID
VERIFICATION_WHISPER_REVISION = BREEZE25_REVISION
WHISPER_ATTENTION_IMPLEMENTATION = BREEZE25_ATTENTION_IMPLEMENTATION
WHISPER_RETURN_ATTENTION_MASK = BREEZE25_RETURN_ATTENTION_MASK
WHISPER_SAMPLE_RATE = BREEZE25_SAMPLE_RATE
WHISPER_MAX_SEGMENT_SECONDS = BREEZE25_MAX_SEGMENT_SECONDS
WHISPER_HARD_MAX_SEGMENT_SECONDS = BREEZE25_HARD_MAX_SEGMENT_SECONDS
WHISPER_MIN_SEGMENT_SECONDS = BREEZE25_MIN_SEGMENT_SECONDS
WHISPER_MIN_PAUSE_SECONDS = BREEZE25_MIN_PAUSE_SECONDS
WHISPER_MAX_VERIFICATION_SEGMENTS = BREEZE25_MAX_VERIFICATION_SEGMENTS
WHISPER_MAX_MICROBATCH_SEGMENTS = BREEZE25_MAX_MICROBATCH_SEGMENTS
SQUIM_OBJECTIVE_ASSET_PATH = "models/squim_objective_dns2020.pth"
SQUIM_OBJECTIVE_WEIGHT_SHA256 = (
"2c54586fea83fb5eb5394d710038ee89f55cab7011a5bf730bebed4c8777e828"
)
SQUIM_OBJECTIVE_SAMPLE_RATE = 16_000
SQUIM_OBJECTIVE_MAX_WEIGHT_BYTES = 67_108_864
SQUIM_OBJECTIVE_WINDOW_SECONDS = 10.0
SQUIM_OBJECTIVE_MAX_WINDOWS = 3
ADAPTIVE_CASCADE_STAGE_LIMITS = (1, 5, 10, 15, 20, 24, 28, 32)
REQUEST_SEED_LIMIT = 2_147_483_648
ACTIVE_VOICE_TOP_DB = 35.0
ACTIVE_VOICE_FRAME_MS = 25.0
ACTIVE_VOICE_HOP_MS = 10.0
ACTIVE_VOICE_MIN_RMS = 1.0e-4
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
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]
CASCADE_EVIDENCE_MAX_REASONS = 8
CASCADE_EVIDENCE_MAX_SEQUENCE_PATHS = 3
CASCADE_EVIDENCE_MAX_TEXT_UNITS = 800
_CASCADE_EVIDENCE_OUTCOMES = frozenset(
{"returned", "no_qualified_candidate", "final_output_rejected"}
)
_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",
"empty_trajectory",
"invalid_audio_duration",
"invalid_chunk_artifacts",
"invalid_gate_config",
"invalid_joined_verification",
"malformed_observation",
"missing_pace_evidence",
"missing_squim_evidence",
"missing_speaker_evidence",
"network_protected_span_mismatch",
"nonfinite_score",
"nonfinite_trajectory_score",
"pace_too_fast",
"semantic_gate",
"speaker_similarity",
"squim_pesq_too_low",
"squim_stoi_too_low",
"truncated",
}
)
_CANDIDATE_GATE_OVERRIDE_KEYS = frozenset(
{
"max_prefix_cer",
"max_suffix_cer",
}
)
@dataclass(frozen=True)
class GenerationPolicy:
"""Candidate-specific endpoint duration estimate used by the Space."""
name: str
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)
class CandidateGenerationContext:
"""Immutable global identity and row-local schedule for one generation.
``candidate_index`` and ``seed`` retain the request-global attempt identity.
``chunk_candidate_ordinals`` is independent: zero denotes the initial
trajectory and positive values count refill attempts within each source
chunk. Context-aware callbacks can therefore rotate generation policies
per chunk without changing the canonical seed schedule.
"""
candidate_index: int
seed: int
chunk_indices: tuple[int, ...]
chunk_candidate_ordinals: tuple[int, ...]
BASE_GENERATION_POLICY = GenerationPolicy(
name="base",
cjk_cps=5.2,
ascii_cps=4.6,
hard_stop_margin_steps=1,
)
SAFE_DURATION_GENERATION_POLICY = GenerationPolicy(
name="safe_duration",
cjk_cps=4.6,
ascii_cps=4.0,
hard_stop_margin_steps=1,
)
COMPLETION_HEADROOM_GENERATION_POLICY = GenerationPolicy(
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,
)
MIXED_CFG_SCHEDULE = "row_ordinal_zero_and_even_primary_odd_alternate"
MIXED_CFG_PRIMARY = 3.0
MIXED_CFG_ALTERNATE = 2.0
MIXED_CFG_SHORT_TEXT_MAX_UNITS = 6
MIXED_CFG_SHORT_TEXT_MIN = 3.0
MIXED_CFG_NETWORK_MIN = 3.0
def generation_policy_for_candidate_offset(candidate_offset: int) -> GenerationPolicy:
"""Map offsets to base, safe, and sparse completion-headroom estimates.
The policy changes only the native-duration endpoint estimate. It is not a
minimum-length policy and therefore never holds the generation loop open to
enforce playback pace. Every fourth retry receives modest completion
headroom; the adaptive stop decision and all semantic/tail gates remain
unchanged.
"""
if isinstance(candidate_offset, (bool, np.bool_)):
raise ValueError("candidate_offset must be a non-negative integer")
try:
offset = operator.index(candidate_offset)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("candidate_offset must be a non-negative integer") from error
if offset < 0:
raise ValueError("candidate_offset must be a non-negative integer")
if offset == 0:
return BASE_GENERATION_POLICY
if offset % 4 == 0:
return COMPLETION_HEADROOM_GENERATION_POLICY
return SAFE_DURATION_GENERATION_POLICY
def generation_cfg_for_candidate_offset(
candidate_offset: int,
*,
primary_cfg: float = MIXED_CFG_PRIMARY,
alternate_cfg: float = MIXED_CFG_ALTERNATE,
) -> float:
"""Return the frozen interleaved CFG for one candidate offset.
Offset zero and every positive even offset use the primary CFG. Positive
odd offsets use the alternate CFG. This preserves the candidate seeds,
endpoint policies, and 32-generation budget while adding the independently
observed CFG diversity; no unvalidated schedule is exposed at runtime.
"""
if isinstance(candidate_offset, (bool, np.bool_)):
raise ValueError("candidate_offset must be a non-negative integer")
try:
offset = operator.index(candidate_offset)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("candidate_offset must be a non-negative integer") from error
if offset < 0:
raise ValueError("candidate_offset must be a non-negative integer")
try:
primary = float(primary_cfg)
alternate = float(alternate_cfg)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("mixed CFG values must be finite values in [1.0, 4.0]") from error
if not all(
math.isfinite(value) and 1.0 <= value <= 4.0
for value in (primary, alternate)
):
raise ValueError("mixed CFG values must be finite values in [1.0, 4.0]")
return primary if offset % 2 == 0 else alternate
def resolve_request_seed(
request_seed: int | None,
random_seed_factory: Callable[[int], int],
) -> int:
"""Return a validated root seed, drawing randomness only for ``None``."""
candidate = (
random_seed_factory(REQUEST_SEED_LIMIT)
if request_seed is None
else request_seed
)
if isinstance(candidate, (bool, np.bool_)):
raise ValueError(f"request_seed must be an integer in [0, {REQUEST_SEED_LIMIT})")
try:
# ``operator.index`` semantics reject floats and numeric strings while
# accepting Python and NumPy integer scalars.
seed = operator.index(candidate)
except (AttributeError, TypeError, ValueError, OverflowError) as error:
raise ValueError(
f"request_seed must be an integer in [0, {REQUEST_SEED_LIMIT})"
) from error
seed = int(seed)
if not 0 <= seed < REQUEST_SEED_LIMIT:
raise ValueError(f"request_seed must be an integer in [0, {REQUEST_SEED_LIMIT})")
return seed
def _finite_float(
value: Any,
*,
minimum: float | None = None,
maximum: float | None = None,
) -> float | None:
if isinstance(value, (bool, np.bool_)):
return None
try:
result = float(value)
except (TypeError, ValueError, OverflowError):
return None
if not math.isfinite(result):
return None
if minimum is not None and result < minimum:
return None
if maximum is not None and result > maximum:
return None
return result
def release_speaker_measurement_required(active_duration_seconds: float) -> bool:
"""Trigger online ECAPA slightly before the external 1.50 s hard gate.
The 20 ms margin covers a bounded RMS-frame shift caused by WAV
serialization while leaving the published external eligibility threshold
unchanged.
"""
duration = _finite_float(active_duration_seconds, minimum=0.0)
return duration is not None and duration >= RELEASE_SPEAKER_TRIGGER_SECONDS
def _mono_audio(audio: np.ndarray | Sequence[float]) -> np.ndarray:
"""Return contiguous mono float32 audio, rejecting ambiguous/bad inputs."""
waveform = np.asarray(audio)
if waveform.ndim == 1:
pass
elif waveform.ndim == 2:
first, second = waveform.shape
if first <= 8 and second > first:
waveform = waveform.mean(axis=0)
elif second <= 8 and first > second:
waveform = waveform.mean(axis=1)
else:
raise ValueError("2-D audio must have an identifiable channel axis (at most 8 channels)")
else:
raise ValueError("audio must be a one- or two-dimensional array")
waveform = np.asarray(waveform, dtype=np.float32).reshape(-1)
if waveform.size == 0:
raise ValueError("audio is empty")
if not np.isfinite(waveform).all():
raise ValueError("audio contains non-finite samples")
return np.ascontiguousarray(waveform)
def _resample_audio(audio: np.ndarray, sample_rate: int, target_rate: int) -> np.ndarray:
try:
source_rate = int(sample_rate)
destination_rate = int(target_rate)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("sample rates must be positive integers") from error
if source_rate <= 0 or destination_rate <= 0:
raise ValueError("sample rates must be positive integers")
if source_rate == destination_rate:
return np.ascontiguousarray(audio, dtype=np.float32)
# SpeechBrain already depends on SciPy. ``resample_poly`` avoids an
# undeclared optional ``librosa`` resampler dependency in the Space image.
from scipy.signal import resample_poly
common_divisor = gcd(source_rate, destination_rate)
output = resample_poly(
np.asarray(audio, dtype=np.float32),
destination_rate // common_divisor,
source_rate // common_divisor,
)
output = np.asarray(output, dtype=np.float32).reshape(-1)
if output.size == 0 or not np.isfinite(output).all():
raise ValueError("resampling produced invalid audio")
return np.ascontiguousarray(output)
@dataclass(frozen=True)
class SquimObjectiveRuntime:
"""Pinned CPU SQUIM objective model and its verified weight identity."""
model: Any
device: torch.device
sample_rate: int
weight_sha256: str
@dataclass(frozen=True)
class SquimObjectiveEvidence:
"""Finite no-reference acoustic metrics for one bounded waveform view."""
stoi: float
pesq: float
si_sdr: float
window_count: int
def _bounded_file_sha256(path: str | Path) -> str:
selected = Path(path)
try:
size = selected.stat().st_size
except OSError as error:
raise RuntimeError("pinned SQUIM weight is unavailable") from error
if not selected.is_file() or not 0 < size <= SQUIM_OBJECTIVE_MAX_WEIGHT_BYTES:
raise RuntimeError("pinned SQUIM weight has an invalid size")
digest = hashlib.sha256()
total = 0
try:
with selected.open("rb") as handle:
while True:
block = handle.read(1024 * 1024)
if not block:
break
total += len(block)
if total > SQUIM_OBJECTIVE_MAX_WEIGHT_BYTES:
raise RuntimeError("pinned SQUIM weight exceeds the size limit")
digest.update(block)
except OSError as error:
raise RuntimeError("pinned SQUIM weight cannot be read") from error
if total != size:
raise RuntimeError("pinned SQUIM weight changed while hashing")
return digest.hexdigest()
def load_pinned_squim_objective_runtime(
*,
asset_fetcher: Callable[[str], str | Path] | None = None,
model_factory: Callable[[], Any] | None = None,
state_loader: Callable[..., Any] | None = None,
file_hasher: Callable[[str | Path], str] | None = None,
) -> SquimObjectiveRuntime:
"""Load SQUIM on CPU only after exact full-file SHA-256 verification."""
import torchaudio
fetcher = asset_fetcher or torchaudio.utils._download_asset
factory = model_factory or torchaudio.models.squim_objective_base
loader = state_loader or torch.load
hasher = file_hasher or _bounded_file_sha256
try:
weight_path = fetcher(SQUIM_OBJECTIVE_ASSET_PATH)
except Exception as error:
raise RuntimeError("pinned SQUIM weight download failed") from error
try:
digest = str(hasher(weight_path)).casefold()
except RuntimeError:
raise
except Exception as error:
raise RuntimeError("pinned SQUIM weight hash failed") from error
if digest != SQUIM_OBJECTIVE_WEIGHT_SHA256:
raise RuntimeError("pinned SQUIM weight SHA-256 mismatch")
# The hash check deliberately precedes deserialization. ``weights_only``
# further constrains the trusted, pinned state-dict load boundary.
try:
state_dict = loader(weight_path, map_location="cpu", weights_only=True)
model = factory()
model.load_state_dict(state_dict, strict=True)
model = model.to(torch.device("cpu"))
model.eval()
except Exception as error:
raise RuntimeError("pinned SQUIM model initialization failed") from error
return SquimObjectiveRuntime(
model=model,
device=torch.device("cpu"),
sample_rate=SQUIM_OBJECTIVE_SAMPLE_RATE,
weight_sha256=SQUIM_OBJECTIVE_WEIGHT_SHA256,
)
class LazySquimObjective:
"""Thread-safe lazy CPU loader for the pinned no-reference metric model."""
def __init__(
self,
runtime_loader: Callable[[], SquimObjectiveRuntime] | None = None,
) -> None:
self._runtime_loader = runtime_loader or load_pinned_squim_objective_runtime
self._runtime: SquimObjectiveRuntime | None = None
self._load_lock = threading.Lock()
self._inference_lock = threading.Lock()
def get_runtime(self) -> SquimObjectiveRuntime:
runtime = self._runtime
if runtime is not None:
return runtime
with self._load_lock:
if self._runtime is None:
loaded = self._runtime_loader()
if not isinstance(loaded, SquimObjectiveRuntime):
raise RuntimeError("SQUIM loader returned an invalid runtime")
if (
loaded.device != torch.device("cpu")
or loaded.sample_rate != SQUIM_OBJECTIVE_SAMPLE_RATE
or loaded.weight_sha256 != SQUIM_OBJECTIVE_WEIGHT_SHA256
):
raise RuntimeError("SQUIM runtime violates the pinned contract")
self._runtime = loaded
return self._runtime
@property
def inference_lock(self) -> threading.Lock:
return self._inference_lock
_DEFAULT_SQUIM_OBJECTIVE = LazySquimObjective()
def _bounded_squim_windows(waveform: np.ndarray) -> tuple[np.ndarray, ...]:
window_samples = int(
round(SQUIM_OBJECTIVE_WINDOW_SECONDS * SQUIM_OBJECTIVE_SAMPLE_RATE)
)
if waveform.size <= window_samples:
return (np.ascontiguousarray(waveform, dtype=np.float32),)
starts = np.linspace(
0,
waveform.size - window_samples,
num=SQUIM_OBJECTIVE_MAX_WINDOWS,
).round().astype(int)
windows = tuple(
np.ascontiguousarray(
waveform[start : start + window_samples],
dtype=np.float32,
)
for start in dict.fromkeys(starts.tolist())
)
if not windows or len(windows) > SQUIM_OBJECTIVE_MAX_WINDOWS:
raise ValueError("SQUIM window selection failed")
return windows
@torch.inference_mode()
@timed_latency_stage("squim")
def squim_objective_evidence_from_audio(
audio: np.ndarray | Sequence[float],
sample_rate: int,
*,
lazy_runtime: LazySquimObjective | None = None,
runtime: SquimObjectiveRuntime | None = None,
) -> SquimObjectiveEvidence:
"""Measure bounded SQUIM evidence with conservative long-audio pooling."""
if runtime is not None and lazy_runtime is not None:
raise ValueError("pass either runtime or lazy_runtime, not both")
waveform = _mono_audio(audio)
waveform = _resample_audio(
waveform,
int(sample_rate),
SQUIM_OBJECTIVE_SAMPLE_RATE,
)
windows = _bounded_squim_windows(waveform)
selected_lazy = lazy_runtime or _DEFAULT_SQUIM_OBJECTIVE
selected_runtime = runtime or selected_lazy.get_runtime()
if (
not isinstance(selected_runtime, SquimObjectiveRuntime)
or selected_runtime.device != torch.device("cpu")
or selected_runtime.sample_rate != SQUIM_OBJECTIVE_SAMPLE_RATE
or selected_runtime.weight_sha256 != SQUIM_OBJECTIVE_WEIGHT_SHA256
):
raise RuntimeError("SQUIM runtime violates the pinned contract")
tensor = torch.from_numpy(np.stack(windows)).to(
selected_runtime.device,
dtype=torch.float32,
)
lock = selected_lazy.inference_lock if runtime is None else threading.Lock()
try:
with lock:
outputs = selected_runtime.model(tensor)
except Exception as error:
raise RuntimeError("pinned SQUIM inference failed") from error
if not isinstance(outputs, (tuple, list)) or len(outputs) != 3:
raise RuntimeError("pinned SQUIM returned an invalid result")
rows: list[np.ndarray] = []
for output in outputs:
values = torch.as_tensor(output).detach().float().cpu().numpy().reshape(-1)
if values.size != len(windows) or not np.isfinite(values).all():
raise ValueError("SQUIM evidence is missing or non-finite")
rows.append(values.astype(np.float64, copy=False))
stoi_values, pesq_values, si_sdr_values = rows
return SquimObjectiveEvidence(
# A single degraded long-form window must not be hidden by clean prose.
stoi=float(np.min(stoi_values)),
pesq=float(np.min(pesq_values)),
si_sdr=float(np.median(si_sdr_values)),
window_count=len(windows),
)
def _trim_active_speech(
audio: np.ndarray,
*,
top_db: float = 35.0,
frame_length: int = 512,
hop_length: int = 128,
) -> np.ndarray:
threshold_db = _finite_float(top_db, minimum=0.0)
if threshold_db is None:
raise ValueError("top_db must be finite and non-negative")
if float(np.max(np.abs(audio))) <= 1.0e-7:
raise ValueError("audio contains no active speech")
import librosa
intervals = librosa.effects.split(
audio,
top_db=threshold_db,
frame_length=max(32, int(frame_length)),
hop_length=max(1, int(hop_length)),
)
if intervals.size == 0:
raise ValueError("audio contains no active speech")
start = int(intervals[0, 0])
stop = int(intervals[-1, 1])
active = np.asarray(audio[start:stop], dtype=np.float32)
if active.size == 0 or float(np.max(np.abs(active))) <= 1.0e-7:
raise ValueError("audio contains no active speech")
return np.ascontiguousarray(active)
def trim_release_speaker_activity(
audio: np.ndarray | Sequence[float],
sample_rate: int,
*,
top_db: float = 35.0,
margin_seconds: float = 0.05,
) -> np.ndarray:
"""Trim speaker audio with the independent release-gate contract.
This deliberately mirrors ``evaluate_space_profile.trim_speaker_activity``:
25 ms RMS frames, 10 ms hop, peak-minus-35 dB with a 1e-4 floor, and a
50 ms outer margin. It remains separate from the interval-union duration
used for pace and speaker eligibility.
"""
signal = _mono_audio(audio)
try:
rate = operator.index(sample_rate)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("sample rate must be positive") from error
relative_db = _finite_float(top_db, minimum=0.0)
margin_duration = _finite_float(margin_seconds, minimum=0.0)
if rate <= 0 or relative_db is None or margin_duration is None:
raise ValueError("release speaker trim settings are invalid")
frame = max(160, int(round(0.025 * rate)))
hop = max(80, int(round(0.010 * rate)))
if signal.size < frame:
active = signal
else:
starts = np.arange(0, signal.size - frame + 1, hop, dtype=np.int64)
rms = np.asarray(
[
float(
np.sqrt(
np.mean(
np.square(signal[start : start + frame], dtype=np.float64)
)
)
)
for start in starts
],
dtype=np.float64,
)
peak = float(rms.max(initial=0.0))
if peak <= 0.0:
active = signal
else:
threshold = max(1.0e-4, peak * (10.0 ** (-relative_db / 20.0)))
active_frames = np.flatnonzero(rms >= threshold)
if active_frames.size == 0:
active = signal
else:
margin = max(0, int(round(margin_duration * rate)))
begin = max(0, int(starts[int(active_frames[0])]) - margin)
end = min(
signal.size,
int(starts[int(active_frames[-1])]) + frame + margin,
)
active = signal[begin:end] if end > begin else signal
active = np.ascontiguousarray(active, dtype=np.float32)
if active.size == 0 or float(np.max(np.abs(active))) <= 1.0e-7:
raise ValueError("audio contains no active speech")
return active
def active_voiced_intervals(
audio: np.ndarray | Sequence[float],
sample_rate: int,
*,
top_db: float = ACTIVE_VOICE_TOP_DB,
frame_ms: float = ACTIVE_VOICE_FRAME_MS,
hop_ms: float = ACTIVE_VOICE_HOP_MS,
min_rms: float = ACTIVE_VOICE_MIN_RMS,
) -> tuple[tuple[int, int], ...]:
"""Return the deterministic union of active RMS-frame intervals.
This is the same 25 ms / 10 ms, peak-minus-35 dB, 1e-4 floor contract
used by the independent hosted evaluator. Unlike first-to-last trimming,
the interval union excludes internal punctuation and joining pauses from
both online pace evidence and the speaker-gate duration threshold.
"""
signal = _mono_audio(audio)
if isinstance(sample_rate, (bool, np.bool_)):
raise ValueError("sample rate must be positive")
try:
source_rate = operator.index(sample_rate)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("sample rate must be positive") from error
if source_rate <= 0:
raise ValueError("sample rate must be positive")
frame_duration = _finite_float(frame_ms, minimum=0.0)
hop_duration = _finite_float(hop_ms, minimum=0.0)
rms_floor = _finite_float(min_rms, minimum=0.0)
relative_db = _finite_float(top_db, minimum=0.0)
if (
frame_duration is None
or frame_duration <= 0.0
or hop_duration is None
or hop_duration <= 0.0
or rms_floor is None
or rms_floor <= 0.0
or relative_db is None
):
raise ValueError("active-voice detector settings are invalid")
frame = max(1, int(round(frame_duration * source_rate / 1000.0)))
hop = max(1, int(round(hop_duration * source_rate / 1000.0)))
if signal.size <= frame:
starts = np.asarray([0], dtype=np.int64)
else:
starts = np.arange(0, signal.size - frame + 1, hop, dtype=np.int64)
final_start = signal.size - frame
if int(starts[-1]) != final_start:
starts = np.append(starts, final_start)
rms = np.asarray(
[
float(
np.sqrt(
np.mean(
np.square(
signal[int(start) : int(start) + frame],
dtype=np.float64,
)
)
)
)
for start in starts
],
dtype=np.float64,
)
peak = float(rms.max(initial=0.0))
threshold = max(rms_floor, peak * 10.0 ** (-relative_db / 20.0))
active_starts = starts[rms >= threshold]
intervals: list[list[int]] = []
for raw_start in active_starts:
start = int(raw_start)
end = min(signal.size, start + frame)
if intervals and start <= intervals[-1][1]:
intervals[-1][1] = max(intervals[-1][1], end)
else:
intervals.append([start, end])
return tuple((start, end) for start, end in intervals)
def active_voiced_duration_seconds(
audio: np.ndarray | Sequence[float],
sample_rate: int,
**detector_kwargs: Any,
) -> float:
"""Measure active interval-union duration under the hosted gate contract."""
intervals = active_voiced_intervals(audio, sample_rate, **detector_kwargs)
active_samples = sum(end - start for start, end in intervals)
return active_samples / float(operator.index(sample_rate))
@dataclass(frozen=True)
class Breeze25Runtime:
"""Loaded processor/model pair for deterministic Breeze ASR 25."""
processor: Any
model: Any
device: torch.device
dtype: torch.dtype
def _load_breeze25_runtime(
model_id: str,
revision: str,
*,
device: str | torch.device | None = None,
processor_factory: Any | None = None,
model_factory: Any | None = None,
) -> Breeze25Runtime:
"""Load one exact ASR revision through the shared deterministic contract.
Factory injection exists for offline tests. The default imports
``transformers`` only when this function is first called.
"""
if processor_factory is None or model_factory is None:
from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor
processor_factory = processor_factory or AutoProcessor
model_factory = model_factory or AutoModelForSpeechSeq2Seq
selected_device = torch.device(
device if device is not None else ("cuda" if torch.cuda.is_available() else "cpu")
)
dtype = torch.float16 if selected_device.type == "cuda" else torch.float32
processor = processor_factory.from_pretrained(
model_id,
revision=revision,
)
model = model_factory.from_pretrained(
model_id,
revision=revision,
attn_implementation=BREEZE25_ATTENTION_IMPLEMENTATION,
torch_dtype=dtype,
low_cpu_mem_usage=True,
use_safetensors=True,
)
model = model.to(selected_device)
model.eval()
return Breeze25Runtime(
processor=processor,
model=model,
device=selected_device,
dtype=dtype,
)
def load_pinned_breeze25_runtime(
*,
device: str | torch.device | None = None,
processor_factory: Any | None = None,
model_factory: Any | None = None,
) -> Breeze25Runtime:
"""Load the sole pinned content verifier used by every semantic gate."""
return _load_breeze25_runtime(
BREEZE25_MODEL_ID,
BREEZE25_REVISION,
device=device,
processor_factory=processor_factory,
model_factory=model_factory,
)
class LazyBreeze25ASR:
"""Thread-safe one-shot lazy loader with an injectable runtime factory."""
def __init__(
self,
runtime_loader: Callable[[], Breeze25Runtime] | None = None,
) -> None:
self._runtime_loader = runtime_loader or load_pinned_breeze25_runtime
self._runtime: Breeze25Runtime | None = None
self._lock = threading.Lock()
def get_runtime(self) -> Breeze25Runtime:
runtime = self._runtime
if runtime is not None:
return runtime
with self._lock:
if self._runtime is None:
self._runtime = self._runtime_loader()
return self._runtime
_DEFAULT_BREEZE25 = LazyBreeze25ASR()
def _qualified_breeze25_pause_ranges(
waveform: np.ndarray,
*,
sample_rate: int,
minimum_pause_seconds: float = 0.25,
) -> tuple[tuple[int, int], ...]:
"""Find waveform-only pause atoms with active evidence on both sides.
A robust percentile, rather than the loudest frame, defines the local
energy scale. Every pause also needs active evidence on both sides, so a
silent waveform or one loud transient cannot manufacture pause atoms.
"""
rate = int(sample_rate)
if rate <= 0:
raise ValueError("sample_rate must be positive")
probe_radius = max(1, int(round(0.01 * rate)))
probe_hop = max(1, int(round(0.01 * rate)))
if waveform.size < 2 * probe_radius:
return ()
probes = np.arange(
probe_radius,
waveform.size - probe_radius + 1,
probe_hop,
dtype=np.int64,
)
if probes.size == 0:
return ()
squared = np.square(waveform, dtype=np.float64)
integral = np.concatenate((np.zeros(1, dtype=np.float64), np.cumsum(squared)))
window_energy = integral[probes + probe_radius] - integral[probes - probe_radius]
rms = np.sqrt(window_energy / float(2 * probe_radius))
robust_index = max(0, min(rms.size - 1, int(math.ceil(0.90 * rms.size)) - 1))
robust_peak = float(np.partition(rms, robust_index)[robust_index])
quiet_threshold = max(1.0e-5, robust_peak * 10.0 ** (-35.0 / 20.0))
active_threshold = max(2.0e-5, quiet_threshold * 4.0, robust_peak * 0.25)
quiet_positions = probes[rms <= quiet_threshold]
quiet_runs: list[tuple[int, int]] = []
for position in quiet_positions.tolist():
if quiet_runs and position <= quiet_runs[-1][1] + probe_hop:
quiet_runs[-1] = (quiet_runs[-1][0], position)
else:
quiet_runs.append((position, position))
minimum_pause = max(1, int(round(minimum_pause_seconds * rate)))
evidence_window = max(probe_hop, int(round(0.75 * rate)))
minimum_active_frames = max(1, int(math.ceil(0.10 * rate / probe_hop)))
qualified: list[tuple[int, int]] = []
for begin, end in quiet_runs:
run_start = max(0, begin - probe_radius)
run_stop = min(waveform.size, end + probe_radius)
if run_stop - run_start < minimum_pause:
continue
left_active = np.count_nonzero(
(probes >= max(0, run_start - evidence_window))
& (probes < run_start)
& (rms >= active_threshold)
)
right_active = np.count_nonzero(
(probes > run_stop)
& (probes <= min(waveform.size, run_stop + evidence_window))
& (rms >= active_threshold)
)
if left_active < minimum_active_frames or right_active < minimum_active_frames:
continue
qualified.append((run_start, run_stop))
return tuple(qualified)
def _split_short_breeze25_audio_at_sustained_pause(
waveform: np.ndarray,
*,
sample_rate: int,
minimum_pause_seconds: float = 0.25,
minimum_segment_seconds: float = 1.25,
) -> tuple[np.ndarray, ...]:
"""Expose genuine pause atoms as deterministic adjacent audio segments."""
rate = int(sample_rate)
if rate <= 0:
raise ValueError("sample_rate must be positive")
minimum_segment = max(1, int(round(minimum_segment_seconds * rate)))
if waveform.size < 2 * minimum_segment:
return (waveform,)
pause_ranges = _qualified_breeze25_pause_ranges(
waveform,
sample_rate=rate,
minimum_pause_seconds=minimum_pause_seconds,
)
boundaries = [0]
for begin, end in pause_ranges:
boundary = (begin + end) // 2
if (
boundary - boundaries[-1] >= minimum_segment
and waveform.size - boundary >= minimum_segment
):
boundaries.append(boundary)
boundaries.append(waveform.size)
return tuple(
np.ascontiguousarray(waveform[start:stop], dtype=np.float32)
for start, stop in zip(boundaries, boundaries[1:])
)
def _split_breeze25_audio(
waveform: np.ndarray,
*,
sample_rate: int = BREEZE25_SAMPLE_RATE,
max_segment_seconds: float = BREEZE25_MAX_SEGMENT_SECONDS,
boundary_search_seconds: float = 1.5,
max_verification_segments: int = BREEZE25_MAX_VERIFICATION_SEGMENTS,
) -> tuple[np.ndarray, ...]:
"""Create bounded verification segments without cutting voiced audio.
Every qualified pause is retained when doing so keeps all segments below
the target cap. This prevents a long multi-chunk decode from swallowing
several already-separated clauses in one Breeze25 context. If the natural
pause atoms cannot bound the contexts, the coarse latest-pause fallback is
used; a long voiced span without such a pause still fails closed.
"""
maximum_seconds = _finite_float(max_segment_seconds, minimum=1.0)
search_seconds = _finite_float(boundary_search_seconds, minimum=0.0)
if (
type(max_verification_segments) is not int
or not 1
<= max_verification_segments
<= GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS
):
raise ValueError(
"Breeze25 segment budget must be an exact integer in [1, 160]"
)
if maximum_seconds is None or search_seconds is None or sample_rate <= 0:
raise ValueError("invalid Breeze25 segmentation settings")
maximum_samples = max(1, int(round(maximum_seconds * sample_rate)))
hard_cap_samples = int(round(BREEZE25_HARD_MAX_SEGMENT_SECONDS * sample_rate))
if maximum_samples > hard_cap_samples:
raise ValueError("Breeze25 target segment cap exceeds the 30-second hard limit")
minimum_segment_samples = max(
1,
int(round(BREEZE25_MIN_SEGMENT_SECONDS * sample_rate)),
)
if waveform.size > hard_cap_samples * max_verification_segments:
raise ValueError("verification audio exceeds the bounded ASR segment budget")
pause_ranges = _qualified_breeze25_pause_ranges(
waveform,
sample_rate=sample_rate,
minimum_pause_seconds=BREEZE25_MIN_PAUSE_SECONDS,
)
pause_segments = _split_short_breeze25_audio_at_sustained_pause(
waveform,
sample_rate=sample_rate,
minimum_pause_seconds=BREEZE25_MIN_PAUSE_SECONDS,
minimum_segment_seconds=BREEZE25_MIN_SEGMENT_SECONDS,
)
if len(pause_segments) > max_verification_segments:
raise ValueError("verification audio exceeds the pause segment cap")
if all(segment.size <= maximum_samples for segment in pause_segments):
return pause_segments
coarse_minimum_samples = min(
maximum_samples // 2,
max(minimum_segment_samples, int(round(6.0 * sample_rate))),
)
boundaries = [0]
while waveform.size - boundaries[-1] > maximum_samples:
segment_start = boundaries[-1]
lower = segment_start + coarse_minimum_samples
upper = min(waveform.size, segment_start + maximum_samples)
boundary = None
for pause_start, pause_stop in reversed(pause_ranges):
clipped_start = max(lower, pause_start)
clipped_stop = min(upper, pause_stop)
if clipped_start > clipped_stop:
continue
midpoint = (pause_start + pause_stop) // 2
candidate = min(clipped_stop, max(clipped_start, midpoint))
boundary = candidate
break
if boundary is None:
raise ValueError(
"long verification audio has no qualified 250ms pause before the cap"
)
boundaries.append(boundary)
if len(boundaries) > max_verification_segments:
raise ValueError("verification audio exceeds the pause segment cap")
boundaries.append(waveform.size)
segments = tuple(
np.ascontiguousarray(waveform[start:stop], dtype=np.float32)
for start, stop in zip(boundaries, boundaries[1:])
)
if len(segments) >= 2 and segments[-1].size < minimum_segment_samples:
merged_samples = segments[-2].size + segments[-1].size
if merged_samples > hard_cap_samples:
raise ValueError("terminal verifier tail cannot be merged below 30 seconds")
segments = (
*segments[:-2],
np.ascontiguousarray(
np.concatenate((segments[-2], segments[-1])),
dtype=np.float32,
),
)
if (
not segments
or any(segment.size == 0 for segment in segments)
or len(segments) > max_verification_segments
or any(segment.size > hard_cap_samples for segment in segments)
or (
waveform.size >= minimum_segment_samples
and any(segment.size < minimum_segment_samples for segment in segments)
)
or sum(segment.size for segment in segments) != waveform.size
):
raise ValueError("failed to split audio within Breeze25's segment limit")
return segments
@torch.inference_mode()
def _transcribe_breeze25(
audio: np.ndarray | Sequence[float],
sample_rate: int,
*,
lazy_asr: LazyBreeze25ASR | None = None,
runtime: Breeze25Runtime | None = None,
language: str | None = BREEZE25_LANGUAGE,
task: str = BREEZE25_TASK,
max_new_tokens: int = 128,
max_verification_segments: int = BREEZE25_MAX_VERIFICATION_SEGMENTS,
) -> str:
"""Transcribe ndarray audio using the sole pinned Breeze25 decoder.
``runtime`` and ``lazy_asr`` are mutually exclusive injection points. An
empty decoded string is returned as-is; the semantic verifier will reject
it rather than accepting an arbitrary TTS fallback. Production fixes the
shared mixed-language embedding with ``language="zh"``. Passing ``None``
is supported only as an explicit evaluation injection and omits the
language argument so the model performs its own language detection.
"""
if runtime is not None and lazy_asr is not None:
raise ValueError("pass either runtime or lazy_asr, not both")
token_limit = int(max_new_tokens)
if token_limit <= 0:
raise ValueError("max_new_tokens must be positive")
waveform = _resample_audio(
_mono_audio(audio),
int(sample_rate),
BREEZE25_SAMPLE_RATE,
)
segments = _split_breeze25_audio(
waveform,
max_verification_segments=max_verification_segments,
)
selected_runtime = runtime or (lazy_asr or _DEFAULT_BREEZE25).get_runtime()
decoded_segments: list[str] = []
for start in range(0, len(segments), BREEZE25_MAX_MICROBATCH_SEGMENTS):
microbatch = segments[start : start + BREEZE25_MAX_MICROBATCH_SEGMENTS]
processor_input: np.ndarray | list[np.ndarray]
processor_input = microbatch[0] if len(microbatch) == 1 else list(microbatch)
processor_output = selected_runtime.processor(
processor_input,
sampling_rate=BREEZE25_SAMPLE_RATE,
return_tensors="pt",
return_attention_mask=BREEZE25_RETURN_ATTENTION_MASK,
)
features = processor_output.input_features.to(
device=selected_runtime.device,
dtype=selected_runtime.dtype,
)
attention_mask = processor_output.attention_mask.to(
device=selected_runtime.device
)
generation_kwargs = {
"attention_mask": attention_mask,
"task": task,
"do_sample": False,
"num_beams": 1,
"max_new_tokens": token_limit,
}
if language is not None:
generation_kwargs["language"] = language
token_ids = selected_runtime.model.generate(features, **generation_kwargs)
decoded = selected_runtime.processor.batch_decode(
token_ids,
skip_special_tokens=True,
)
if not decoded or len(decoded) != len(microbatch):
return ""
decoded_segments.extend(str(text).strip() for text in decoded)
return " ".join(text for text in decoded_segments if text)
@timed_latency_stage("breeze25_asr")
def transcribe_breeze25(
audio: np.ndarray | Sequence[float],
sample_rate: int,
*,
lazy_asr: LazyBreeze25ASR | None = None,
runtime: Breeze25Runtime | None = None,
language: str | None = BREEZE25_LANGUAGE,
task: str = BREEZE25_TASK,
max_new_tokens: int = 128,
max_verification_segments: int = BREEZE25_MAX_VERIFICATION_SEGMENTS,
) -> str:
"""Transcribe with the sole pinned Breeze ASR 25 content verifier."""
return _transcribe_breeze25(
audio,
sample_rate,
lazy_asr=lazy_asr,
runtime=runtime,
language=language,
task=task,
max_new_tokens=max_new_tokens,
max_verification_segments=max_verification_segments,
)
# Compatibility aliases are direct references, not decorated wrappers. All
# callers therefore share one lazy model and each decode is timed exactly once.
WhisperRuntime = Breeze25Runtime
LazyWhisperASR = LazyBreeze25ASR
load_pinned_whisper_runtime = load_pinned_breeze25_runtime
load_pinned_verification_whisper_runtime = load_pinned_breeze25_runtime
_DEFAULT_WHISPER = _DEFAULT_BREEZE25
_DEFAULT_VERIFICATION_WHISPER = _DEFAULT_BREEZE25
_qualified_whisper_pause_ranges = _qualified_breeze25_pause_ranges
_split_short_whisper_audio_at_sustained_pause = (
_split_short_breeze25_audio_at_sustained_pause
)
_split_whisper_audio = _split_breeze25_audio
_transcribe_whisper = _transcribe_breeze25
transcribe_whisper = transcribe_breeze25
transcribe_verification_whisper = transcribe_breeze25
@dataclass(frozen=True)
class PreparedCandidateAudio:
"""Validated candidate waveform and its ASR transcript."""
waveform: np.ndarray
duration_seconds: float
transcript_text: str
def prepare_candidate_audio(
audio: np.ndarray | Sequence[float],
sample_rate: int,
*,
transcriber: Callable[[np.ndarray, int], str] | None = None,
) -> PreparedCandidateAudio | None:
"""Prepare one candidate, returning ``None`` for candidate-data errors.
Invalid/empty/non-finite audio and a transcriber's ``ValueError`` describe
an unusable candidate, not a service outage. They therefore become a
normal gate rejection so the cascade can try the next seed. Runtime and
I/O failures deliberately propagate and abort the request fail-closed.
"""
try:
selected_sample_rate = int(sample_rate)
if selected_sample_rate <= 0:
raise ValueError("sample_rate must be positive")
waveform = _mono_audio(audio)
transcript = (transcriber or transcribe_breeze25)(
waveform,
selected_sample_rate,
)
if not isinstance(transcript, str):
raise ValueError("ASR transcript must be a string")
except (TypeError, ValueError, OverflowError):
return None
return PreparedCandidateAudio(
waveform=waveform,
duration_seconds=waveform.size / float(selected_sample_rate),
transcript_text=transcript.strip(),
)
@torch.inference_mode()
def _encode_speaker_segments(
segments: Sequence[np.ndarray],
encoder: Any,
*,
device: str | torch.device,
) -> np.ndarray:
"""Encode a variable-length segment batch in one ECAPA forward pass."""
waveforms = tuple(np.asarray(segment, dtype=np.float32).reshape(-1) for segment in segments)
if not waveforms or any(
waveform.size == 0 or not np.isfinite(waveform).all()
for waveform in waveforms
):
raise ValueError("speaker segments must be non-empty and finite")
maximum_length = max(waveform.size for waveform in waveforms)
batch = np.zeros((len(waveforms), maximum_length), dtype=np.float32)
relative_lengths = np.empty(len(waveforms), dtype=np.float32)
for index, waveform in enumerate(waveforms):
batch[index, : waveform.size] = waveform
relative_lengths[index] = waveform.size / float(maximum_length)
selected_device = torch.device(device)
tensor = torch.from_numpy(batch).to(selected_device)
wav_lens = torch.from_numpy(relative_lengths).to(selected_device)
embeddings = encoder.encode_batch(tensor, wav_lens=wav_lens)
embeddings = torch.as_tensor(embeddings).detach().float()
if embeddings.ndim == 0 or embeddings.shape[0] != len(waveforms):
raise ValueError("speaker encoder returned an invalid batch size")
embeddings = embeddings.reshape(len(waveforms), -1)
if embeddings.shape[1] == 0 or not torch.isfinite(embeddings).all():
raise ValueError("speaker encoder returned an invalid embedding")
norms = torch.linalg.vector_norm(embeddings, dim=1)
if not torch.isfinite(norms).all() or bool(torch.any(norms <= 1.0e-8)):
raise ValueError("speaker encoder returned a zero-norm embedding")
normalized = torch_functional.normalize(embeddings, dim=1).cpu().numpy().astype(np.float32)
if not np.isfinite(normalized).all():
raise ValueError("speaker encoder returned a non-finite embedding")
return normalized
@torch.inference_mode()
def speaker_embedding_from_audio(
audio: np.ndarray | Sequence[float],
sample_rate: int,
encoder: Any,
*,
device: str | torch.device = "cpu",
target_sample_rate: int = 16_000,
active_top_db: float = 35.0,
) -> np.ndarray:
"""Extract one normalized ECAPA embedding from active ndarray speech."""
waveform = _mono_audio(audio)
waveform = _resample_audio(waveform, int(sample_rate), int(target_sample_rate))
waveform = _trim_active_speech(waveform, top_db=active_top_db)
return _encode_speaker_segments((waveform,), encoder, device=device)[0]
def cosine_similarity(left: np.ndarray | Sequence[float], right: np.ndarray | Sequence[float]) -> float:
"""Return a finite cosine similarity, raising on unusable embeddings."""
left_array = np.asarray(left, dtype=np.float64).reshape(-1)
right_array = np.asarray(right, dtype=np.float64).reshape(-1)
if left_array.size == 0 or left_array.shape != right_array.shape:
raise ValueError("speaker embeddings must have equal non-empty shapes")
if not np.isfinite(left_array).all() or not np.isfinite(right_array).all():
raise ValueError("speaker embeddings must be finite")
denominator = float(np.linalg.norm(left_array) * np.linalg.norm(right_array))
if not math.isfinite(denominator) or denominator <= 1.0e-12:
raise ValueError("speaker embeddings must have non-zero norm")
similarity = float(np.dot(left_array, right_array) / denominator)
if not math.isfinite(similarity):
raise ValueError("speaker cosine similarity is non-finite")
return float(np.clip(similarity, -1.0, 1.0))
@dataclass(frozen=True)
class SpeakerEvidence:
similarity: float
begin_similarity: float
end_similarity: float
boundary_drop: float
active_duration_seconds: float
speaker_embedding: np.ndarray
active_rms_db: float
@timed_latency_stage("ecapa")
def speaker_evidence_from_audio(
audio: np.ndarray | Sequence[float],
sample_rate: int,
encoder: Any,
anchor_embedding: np.ndarray | Sequence[float],
*,
device: str | torch.device = "cpu",
edge_seconds: float = 1.5,
whole_window_seconds: float = 3.0,
whole_max_windows: int = 4,
active_top_db: float = 35.0,
) -> SpeakerEvidence:
"""Measure whole/begin/end anchor similarity on trimmed active speech."""
edge_duration = _finite_float(edge_seconds, minimum=0.01)
window_duration = _finite_float(whole_window_seconds, minimum=0.01)
try:
max_windows = int(whole_max_windows)
except (TypeError, ValueError, OverflowError):
max_windows = 0
if edge_duration is None or window_duration is None or max_windows <= 0:
raise ValueError("speaker window settings must be finite and positive")
waveform = _resample_audio(_mono_audio(audio), int(sample_rate), 16_000)
active_duration_seconds = active_voiced_duration_seconds(
waveform,
16_000,
top_db=active_top_db,
)
active = _trim_active_speech(waveform, top_db=active_top_db)
edge_samples = max(1, int(round(edge_duration * 16_000)))
whole_window_samples = max(1, int(round(window_duration * 16_000)))
begin = active[:edge_samples]
end = active[-edge_samples:]
if active.size <= whole_window_samples:
whole_segments = [active]
else:
starts = np.linspace(
0,
active.size - whole_window_samples,
num=max_windows,
).round().astype(int)
whole_segments = [
active[start : start + whole_window_samples]
for start in dict.fromkeys(starts.tolist())
]
embeddings = _encode_speaker_segments(
(*whole_segments, begin, end),
encoder,
device=device,
)
whole_embeddings = embeddings[: len(whole_segments)]
whole_embedding = np.mean(whole_embeddings, axis=0, dtype=np.float64)
whole_norm = float(np.linalg.norm(whole_embedding))
if not math.isfinite(whole_norm) or whole_norm <= 1.0e-8:
raise ValueError("speaker windows produced a zero-norm embedding")
whole_embedding = np.asarray(whole_embedding / whole_norm, dtype=np.float32)
begin_embedding, end_embedding = embeddings[-2:]
similarity = cosine_similarity(whole_embedding, anchor_embedding)
begin_similarity = cosine_similarity(begin_embedding, anchor_embedding)
end_similarity = cosine_similarity(end_embedding, anchor_embedding)
boundary_drop = max(0.0, begin_similarity - end_similarity)
active_rms = float(np.sqrt(np.mean(np.square(active, dtype=np.float64))))
if not math.isfinite(active_rms) or active_rms <= 1.0e-8:
raise ValueError("active speech has invalid RMS")
return SpeakerEvidence(
similarity=similarity,
begin_similarity=begin_similarity,
end_similarity=end_similarity,
boundary_drop=boundary_drop,
active_duration_seconds=active_duration_seconds,
speaker_embedding=whole_embedding.copy(),
active_rms_db=20.0 * math.log10(active_rms),
)
@timed_latency_stage("ecapa")
def release_speaker_evidence_from_audio(
audio: np.ndarray | Sequence[float],
sample_rate: int,
encoder: Any,
anchor_embedding: np.ndarray | Sequence[float],
*,
device: str | torch.device = "cpu",
active_top_db: float = 35.0,
) -> SpeakerEvidence:
"""Measure the exact full/third speaker evidence used for promotion.
Candidate-local scoring keeps the bounded window metric for latency. A
returned waveform is additionally checked with this independent-aligned
metric so a low online boundary drop cannot hide a release-gate failure.
"""
try:
rate = operator.index(sample_rate)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("sample rate must be positive") from error
if rate <= 0:
raise ValueError("sample rate must be positive")
waveform = _mono_audio(audio)
active_duration_seconds = active_voiced_duration_seconds(
waveform,
rate,
top_db=active_top_db,
)
active = trim_release_speaker_activity(
waveform,
rate,
top_db=active_top_db,
)
source_segments = (active, *tuple(np.array_split(active, 3)))
if any(segment.size == 0 for segment in source_segments):
raise ValueError("release speaker segments must be non-empty")
import librosa
segments_16k: list[np.ndarray] = []
for segment in source_segments:
resampled = (
segment
if rate == 16_000
else librosa.resample(segment, orig_sr=rate, target_sr=16_000)
)
segments_16k.append(np.ascontiguousarray(resampled, dtype=np.float32))
# Encode one segment at a time to match the independent verifier rather
# than allowing padding/batching to perturb short boundary embeddings.
selected_device = torch.device(device)
encoded: list[np.ndarray] = []
for segment in segments_16k:
tensor = torch.from_numpy(segment).unsqueeze(0).to(selected_device)
embedding = (
torch.as_tensor(encoder.encode_batch(tensor))
.detach()
.float()
.reshape(-1)
)
if embedding.numel() == 0 or not bool(torch.isfinite(embedding).all()):
raise ValueError("speaker encoder returned an invalid embedding")
norm = torch.linalg.vector_norm(embedding)
if not bool(torch.isfinite(norm)) or float(norm) <= 1.0e-8:
raise ValueError("speaker encoder returned a zero-norm embedding")
encoded.append((embedding / norm).cpu().numpy().astype(np.float32))
embeddings = np.stack(encoded, axis=0)
whole_embedding = embeddings[0]
begin_embedding = embeddings[1]
end_embedding = embeddings[-1]
similarity = cosine_similarity(whole_embedding, anchor_embedding)
begin_similarity = cosine_similarity(begin_embedding, anchor_embedding)
end_similarity = cosine_similarity(end_embedding, anchor_embedding)
active_rms = float(np.sqrt(np.mean(np.square(active, dtype=np.float64))))
if not math.isfinite(active_rms) or active_rms <= 1.0e-8:
raise ValueError("active speech has invalid RMS")
return SpeakerEvidence(
similarity=similarity,
begin_similarity=begin_similarity,
end_similarity=end_similarity,
boundary_drop=max(0.0, begin_similarity - end_similarity),
active_duration_seconds=active_duration_seconds,
speaker_embedding=whole_embedding.copy(),
active_rms_db=20.0 * math.log10(active_rms),
)
def active_audio_rms_db(audio: np.ndarray | Sequence[float], *, top_db: float = 35.0) -> float:
"""Measure finite RMS dB on the active region of candidate audio."""
active = _trim_active_speech(_mono_audio(audio), top_db=top_db)
rms = float(np.sqrt(np.mean(np.square(active, dtype=np.float64))))
if not math.isfinite(rms) or rms <= 1.0e-8:
raise ValueError("active speech has invalid RMS")
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))
@timed_latency_stage("f0")
def active_audio_median_f0_hz(
audio: np.ndarray | Sequence[float],
sample_rate: int,
*,
fmin_hz: float = 65.0,
fmax_hz: float = 500.0,
) -> float | None:
"""Return a robust voiced-pitch median for soft sequence transitions."""
waveform = _mono_audio(audio)
if sample_rate <= 0 or waveform.size < 1024:
return None
try:
import librosa
f0, voiced, _ = librosa.pyin(
waveform,
fmin=float(fmin_hz),
fmax=float(fmax_hz),
sr=int(sample_rate),
frame_length=1024,
hop_length=256,
)
except (ImportError, FloatingPointError, TypeError, ValueError):
return None
valid = np.asarray(f0, dtype=np.float64)
if voiced is not None:
valid = valid[np.asarray(voiced, dtype=bool)]
valid = valid[np.isfinite(valid)]
if valid.size == 0:
return None
median = float(np.median(valid))
return median if math.isfinite(median) and median > 0.0 else None
@dataclass(frozen=True)
class CandidateObservation:
target_text: str
transcript_text: str
audio_duration_seconds: float
speaker_similarity: float | None = None
begin_speaker_similarity: float | None = None
end_speaker_similarity: float | None = None
pace_cps: float | None = None
squim_stoi: float | None = None
squim_pesq: float | None = None
squim_si_sdr: float | None = None
truncated: bool = False
@dataclass(frozen=True)
class CandidateGateResult:
passed: bool
comparison: AsrComparison
audio_duration_seconds: float | None
speaker_gate_applied: bool
speaker_similarity: float | None
boundary_speaker_drop: float | None
pace_cps: float | None
squim_gate_applied: bool
squim_stoi: float | None
squim_pesq: float | None
squim_si_sdr: float | None
squim_quality_cost: float | None
score: float
rejection_reasons: tuple[str, ...]
@dataclass(frozen=True)
class ChunkCandidateArtifact:
"""Acoustic evidence retained for sequence-level candidate selection."""
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)
class CandidateGateEvidence:
"""Bounded, content-free diagnostic snapshot of one hard-gate result."""
passed: bool
target_units: int
hypothesis_units: int
edit_distance: int
cer: float | None
prefix_cer: float | None
suffix_cer: float | None
prefix_deletions: int
suffix_deletions: int
extra_tail_units: int
audio_duration_seconds: float | None
speaker_gate_applied: bool
speaker_similarity: float | None
boundary_speaker_drop: float | None
pace_cps: float | None
squim_gate_applied: bool
squim_stoi: float | None
squim_pesq: float | None
squim_si_sdr: float | None
squim_quality_cost: float | None
score: float | None
rejection_reasons: tuple[str, ...]
@dataclass(frozen=True)
class TrajectoryGateEvidence:
"""Content-free evidence for a joined, sequence, or final waveform gate."""
passed: bool
result_count: int
score: float | None
rejection_reasons: tuple[str, ...]
result: CandidateGateEvidence | None
@dataclass(frozen=True)
class LocalIndependentGateEvidence:
"""Bounded per-row attestation for independent local semantic ASR."""
attempted: bool
passed: bool | None
proof_count: int
result: CandidateGateEvidence | None
@dataclass(frozen=True)
class CandidateGenerationEvidence:
"""Content-free CFG and source-row evidence for one generation call."""
chunk_indices: tuple[int, ...]
chunk_text_units: tuple[int, ...]
scheduled_cfg: float
effective_cfgs: tuple[float, ...]
floor_reasons: tuple[tuple[str, ...], ...]
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)
class CandidateAttemptEvidence:
"""Diagnostics for one generated trajectory without waveform/text payloads."""
candidate_index: int
seed: int
trajectory_passed: bool
trajectory_score: float | None
trajectory_rejection_reasons: tuple[str, ...]
local_result_count: int
local_results: tuple[CandidateGateEvidence, ...]
chunk_indices: tuple[int, ...]
chunk_text_units: tuple[int, ...]
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, ...], ...]
independent_local_results: tuple[LocalIndependentGateEvidence, ...] = ()
joined_output: TrajectoryGateEvidence | None = None
independent_output: TrajectoryGateEvidence | None = None
@dataclass(frozen=True)
class SequencePathEvidence:
"""Final-verifier evidence in bounded check order across expanding lattices."""
rank: int
chunk_candidate_indices: tuple[int, ...]
chunk_seeds: tuple[int, ...]
final_output: TrajectoryGateEvidence
@dataclass(frozen=True)
class SequenceSearchEvidence:
"""Finite-lattice diagnostics and bounded exact-verifier search evidence.
``ranked_path_count`` counts verifier-order paths accumulated across the
incremental and completed-lattice searches; it is not a global rank count
recomputed against only the final lattice.
"""
eligible_candidate_counts: tuple[int, ...]
finite_transition_counts: tuple[int, ...]
ranked_path_count: int
checked_paths: tuple[SequencePathEvidence, ...] = ()
@dataclass(frozen=True)
class CascadeDiagnostics:
"""Bounded diagnostics retained on both success and fail-closed outcomes."""
attempts: tuple[CandidateAttemptEvidence, ...] = ()
sequence_search: SequenceSearchEvidence | None = None
def _evidence_float(value: Any) -> float | None:
return _finite_float(value)
def _evidence_int(value: Any, *, maximum: int = 1_000_000) -> int:
if isinstance(value, (bool, np.bool_)):
return 0
try:
integer = operator.index(value)
except (TypeError, ValueError, OverflowError):
return 0
return min(max(0, int(integer)), maximum)
def _sanitize_rejection_reason(reason: Any) -> str:
"""Return an allow-listed reason code without forwarding arbitrary text."""
if not isinstance(reason, str) or len(reason) > 96:
return "unknown_rejection"
tokens = reason.split(":")
if (
not tokens
or len(tokens) > 3
or tokens[-1] not in _CASCADE_EVIDENCE_REJECTION_CODES
):
return "unknown_rejection"
for token in tokens[:-1]:
if token == "joined_output":
continue
if token.startswith("chunk_"):
chunk_index = token[6:]
if (
chunk_index.isdigit()
and len(chunk_index) <= 2
and int(chunk_index) < CASCADE_EVIDENCE_MAX_LOCAL_RESULTS
):
continue
return "unknown_rejection"
return ":".join(tokens)
def _bounded_rejection_reasons(reasons: Any) -> tuple[str, ...]:
try:
values = tuple(reasons)
except TypeError:
values = ()
return tuple(
_sanitize_rejection_reason(reason)
for reason in values[:CASCADE_EVIDENCE_MAX_REASONS]
)
def candidate_gate_evidence(result: CandidateGateResult) -> CandidateGateEvidence:
"""Project a gate result to finite scalar/count evidence only."""
if not isinstance(result, CandidateGateResult):
raise TypeError("result must be a CandidateGateResult")
comparison = result.comparison
return CandidateGateEvidence(
passed=result.passed is True,
target_units=_evidence_int(len(comparison.target_text)),
hypothesis_units=_evidence_int(len(comparison.transcript_text)),
edit_distance=_evidence_int(comparison.edit_distance),
cer=_evidence_float(comparison.cer),
prefix_cer=_evidence_float(comparison.prefix_cer),
suffix_cer=_evidence_float(comparison.suffix_cer),
prefix_deletions=_evidence_int(comparison.prefix_deletions),
suffix_deletions=_evidence_int(comparison.suffix_deletions),
extra_tail_units=_evidence_int(comparison.extra_tail_units),
audio_duration_seconds=_evidence_float(result.audio_duration_seconds),
speaker_gate_applied=result.speaker_gate_applied is True,
speaker_similarity=_evidence_float(result.speaker_similarity),
boundary_speaker_drop=_evidence_float(result.boundary_speaker_drop),
pace_cps=_evidence_float(result.pace_cps),
squim_gate_applied=result.squim_gate_applied is True,
squim_stoi=_evidence_float(result.squim_stoi),
squim_pesq=_evidence_float(result.squim_pesq),
squim_si_sdr=_evidence_float(result.squim_si_sdr),
squim_quality_cost=_evidence_float(result.squim_quality_cost),
score=_evidence_float(result.score),
rejection_reasons=_bounded_rejection_reasons(result.rejection_reasons),
)
def verify_candidate(
observation: CandidateObservation,
*,
locale: str = "zh-TW",
short_text_units: int = 6,
short_text_max_cer: float = 0.0,
max_cer: float = 0.20,
prefix_units: int = 6,
suffix_units: int = 6,
max_prefix_cer: float = 0.0,
max_suffix_cer: float = 0.0,
max_prefix_deletions: int | None = None,
max_suffix_deletions: int | None = None,
max_extra_tail_units: int = 0,
speaker_gate_enabled: bool = True,
short_audio_seconds: float = 1.5,
min_speaker_similarity: float = 0.10,
max_boundary_speaker_drop: float = 0.03,
max_pace_cps: float | None = None,
squim_gate_enabled: bool = False,
min_squim_stoi: float = 0.60,
min_squim_pesq: float = 1.12,
squim_stoi_weight: float = 0.03,
squim_pesq_weight: float = 0.02,
squim_si_sdr_weight: float = 0.01,
speaker_weight: float = 0.05,
boundary_weight: float = 0.10,
skip_acoustic_on_hard_failure: bool = False,
) -> CandidateGateResult:
"""Apply strict semantic and duration-aware speaker gates to a candidate."""
duration = _finite_float(observation.audio_duration_seconds, minimum=0.0)
short_duration_limit = _finite_float(short_audio_seconds, minimum=0.0)
general_cer_limit = _finite_float(max_cer, minimum=0.0)
exact_cer_limit = _finite_float(short_text_max_cer, minimum=0.0)
min_similarity = _finite_float(min_speaker_similarity, minimum=-1.0, maximum=1.0)
max_boundary = _finite_float(max_boundary_speaker_drop, minimum=0.0)
max_pace = None if max_pace_cps is None else _finite_float(max_pace_cps, minimum=0.0)
min_stoi = _finite_float(min_squim_stoi, minimum=0.0, maximum=1.0)
min_pesq = _finite_float(min_squim_pesq, minimum=0.0, maximum=5.0)
stoi_cost_weight = _finite_float(squim_stoi_weight, minimum=0.0)
pesq_cost_weight = _finite_float(squim_pesq_weight, minimum=0.0)
si_sdr_cost_weight = _finite_float(squim_si_sdr_weight, minimum=0.0)
speaker_cost_weight = _finite_float(speaker_weight, minimum=0.0)
boundary_cost_weight = _finite_float(boundary_weight, minimum=0.0)
try:
short_unit_limit = max(0, int(short_text_units))
except (TypeError, ValueError, OverflowError):
short_unit_limit = -1
deletion_limits: list[int | None] = []
deletion_config_valid = True
for value in (max_prefix_deletions, max_suffix_deletions):
if value is None:
deletion_limits.append(None)
continue
if isinstance(value, (bool, np.bool_)):
deletion_limits.append(None)
deletion_config_valid = False
continue
try:
limit = operator.index(value)
except (TypeError, ValueError, OverflowError):
deletion_limits.append(None)
deletion_config_valid = False
continue
deletion_limits.append(int(limit))
if limit < 0:
deletion_config_valid = False
speaker_enabled = isinstance(speaker_gate_enabled, (bool, np.bool_))
if speaker_enabled:
speaker_enabled = bool(speaker_gate_enabled)
squim_enabled = isinstance(squim_gate_enabled, (bool, np.bool_))
if squim_enabled:
squim_enabled = bool(squim_gate_enabled)
skip_acoustic_enabled = isinstance(
skip_acoustic_on_hard_failure,
(bool, np.bool_),
)
if skip_acoustic_enabled:
skip_acoustic_enabled = bool(skip_acoustic_on_hard_failure)
# First normalize with a permissive finite limit to determine target units.
preliminary = compare_asr_text(
observation.target_text,
observation.transcript_text,
locale=locale,
prefix_units=prefix_units,
suffix_units=suffix_units,
max_cer=general_cer_limit if general_cer_limit is not None else math.nan,
max_prefix_cer=max_prefix_cer,
max_suffix_cer=max_suffix_cer,
max_extra_tail_units=max_extra_tail_units,
)
selected_cer_limit = general_cer_limit
if short_unit_limit >= 0 and len(preliminary.target_text) <= short_unit_limit:
selected_cer_limit = exact_cer_limit
comparison = compare_asr_text(
observation.target_text,
observation.transcript_text,
locale=locale,
prefix_units=prefix_units,
suffix_units=suffix_units,
max_cer=selected_cer_limit if selected_cer_limit is not None else math.nan,
max_prefix_cer=max_prefix_cer,
max_suffix_cer=max_suffix_cer,
max_extra_tail_units=max_extra_tail_units,
)
reasons: list[str] = []
if duration is None or duration <= 0.0:
reasons.append("invalid_audio_duration")
if observation.truncated is not False:
reasons.append("truncated")
deletion_gate_passed = bool(
deletion_config_valid
and (
deletion_limits[0] is None
or comparison.prefix_deletions <= deletion_limits[0]
)
and (
deletion_limits[1] is None
or comparison.suffix_deletions <= deletion_limits[1]
)
)
if not comparison.passed or (
deletion_config_valid and not deletion_gate_passed
):
reasons.append("semantic_gate")
if (
comparison.network_protected_spans > 0
and comparison.network_protected_spans_passed is not True
):
reasons.append("network_protected_span_mismatch")
pace = None
if max_pace_cps is not None:
pace = _finite_float(observation.pace_cps, minimum=0.0)
if max_pace is None:
reasons.append("invalid_gate_config")
elif pace is None:
reasons.append("missing_pace_evidence")
elif pace > max_pace:
reasons.append("pace_too_fast")
hard_failure_before_acoustics = bool(reasons)
skip_acoustic = bool(
skip_acoustic_enabled and hard_failure_before_acoustics
)
squim_gate_applied = bool(squim_enabled and not skip_acoustic)
squim_stoi: float | None = None
squim_pesq: float | None = None
squim_si_sdr: float | None = None
squim_quality_cost: float | None = None
if squim_gate_applied:
squim_stoi = _finite_float(
observation.squim_stoi,
minimum=0.0,
maximum=1.0,
)
squim_pesq = _finite_float(
observation.squim_pesq,
minimum=0.0,
maximum=5.0,
)
squim_si_sdr = _finite_float(observation.squim_si_sdr)
if (
squim_stoi is None
or squim_pesq is None
or squim_si_sdr is None
):
reasons.append("missing_squim_evidence")
else:
if min_stoi is None or squim_stoi < min_stoi:
reasons.append("squim_stoi_too_low")
if min_pesq is None or squim_pesq < min_pesq:
reasons.append("squim_pesq_too_low")
if all(
weight is not None
for weight in (
stoi_cost_weight,
pesq_cost_weight,
si_sdr_cost_weight,
)
):
assert squim_stoi is not None
assert squim_pesq is not None
assert squim_si_sdr is not None
squim_quality_cost = (
stoi_cost_weight * float(np.clip(1.0 - squim_stoi, 0.0, 1.0))
+ pesq_cost_weight
* float(np.clip((4.5 - squim_pesq) / 3.5, 0.0, 1.0))
+ si_sdr_cost_weight
* float(np.clip((20.0 - squim_si_sdr) / 40.0, 0.0, 1.0))
)
valid_common_config = bool(
isinstance(speaker_gate_enabled, (bool, np.bool_))
and short_duration_limit is not None
and general_cer_limit is not None
and exact_cer_limit is not None
and short_unit_limit >= 0
and deletion_config_valid
and isinstance(squim_gate_enabled, (bool, np.bool_))
and isinstance(skip_acoustic_on_hard_failure, (bool, np.bool_))
and (
not squim_enabled
or all(
value is not None
for value in (
min_stoi,
min_pesq,
stoi_cost_weight,
pesq_cost_weight,
si_sdr_cost_weight,
)
)
)
and (
not speaker_enabled
or all(
value is not None
for value in (
min_similarity,
max_boundary,
speaker_cost_weight,
boundary_cost_weight,
)
)
)
)
if not valid_common_config:
reasons.append("invalid_gate_config")
speaker_gate_applied = bool(
speaker_enabled
and not skip_acoustic
and duration is not None
and short_duration_limit is not None
and duration >= short_duration_limit
)
similarity: float | None = None
boundary_drop: float | None = None
if speaker_gate_applied:
similarity = _finite_float(
observation.speaker_similarity,
minimum=-1.0,
maximum=1.0,
)
begin_similarity = _finite_float(
observation.begin_speaker_similarity,
minimum=-1.0,
maximum=1.0,
)
end_similarity = _finite_float(
observation.end_speaker_similarity,
minimum=-1.0,
maximum=1.0,
)
if similarity is None or begin_similarity is None or end_similarity is None:
reasons.append("missing_speaker_evidence")
else:
boundary_drop = max(0.0, begin_similarity - end_similarity)
if min_similarity is None or similarity < min_similarity:
reasons.append("speaker_similarity")
if max_boundary is None or boundary_drop > max_boundary:
reasons.append("boundary_speaker_drop")
score = math.inf
if not reasons:
score = comparison.cer
if speaker_gate_applied:
assert similarity is not None and boundary_drop is not None
assert speaker_cost_weight is not None and boundary_cost_weight is not None
score += speaker_cost_weight * (1.0 - similarity)
score += boundary_cost_weight * boundary_drop
if squim_gate_applied:
assert squim_quality_cost is not None
score += squim_quality_cost
if not math.isfinite(score) or score < 0.0:
reasons.append("nonfinite_score")
score = math.inf
return CandidateGateResult(
passed=not reasons,
comparison=comparison,
audio_duration_seconds=duration,
speaker_gate_applied=speaker_gate_applied,
speaker_similarity=similarity,
boundary_speaker_drop=boundary_drop,
pace_cps=pace,
squim_gate_applied=squim_gate_applied,
squim_stoi=squim_stoi,
squim_pesq=squim_pesq,
squim_si_sdr=squim_si_sdr,
squim_quality_cost=squim_quality_cost,
score=score,
rejection_reasons=tuple(reasons),
)
@dataclass(frozen=True)
class TrajectoryGateResult:
passed: bool
candidate_results: tuple[CandidateGateResult, ...]
score: float
rejection_reasons: tuple[str, ...]
chunk_artifacts: tuple[ChunkCandidateArtifact, ...] = ()
@dataclass(frozen=True)
class CandidateVerification:
"""Selection result plus content-free whole-output gate evidence."""
verification: TrajectoryGateResult
independent_local_results: tuple[LocalIndependentGateEvidence, ...] = ()
joined_output: TrajectoryGateEvidence | None = None
independent_output: TrajectoryGateEvidence | None = None
def trajectory_gate_evidence(
verification: TrajectoryGateResult,
) -> TrajectoryGateEvidence:
"""Project a joined/final verification without retaining recognized text."""
if not isinstance(verification, TrajectoryGateResult):
raise TypeError("verification must be a TrajectoryGateResult")
sole_result = (
verification.candidate_results[0]
if len(verification.candidate_results) == 1
else None
)
result = (
candidate_gate_evidence(sole_result)
if isinstance(sole_result, CandidateGateResult)
else None
)
return TrajectoryGateEvidence(
passed=verification.passed is True,
result_count=_evidence_int(len(verification.candidate_results)),
score=_evidence_float(verification.score),
rejection_reasons=_bounded_rejection_reasons(
verification.rejection_reasons
),
result=result,
)
def exact_waveform_sha256(
audio: np.ndarray | Sequence[float],
sample_rate: int,
) -> str:
"""Hash exact canonical float32 samples together with their sample rate."""
if isinstance(sample_rate, (bool, np.bool_)):
raise ValueError("sample_rate must be a positive integer")
try:
rate = operator.index(sample_rate)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("sample_rate must be a positive integer") from error
if rate <= 0:
raise ValueError("sample_rate must be a positive integer")
waveform = _mono_audio(audio)
canonical = np.ascontiguousarray(waveform, dtype=np.dtype("<f4"))
digest = hashlib.sha256()
digest.update(b"bluemagpie-whole-waveform-f32le-v1\0")
digest.update(int(rate).to_bytes(8, "little", signed=False))
digest.update(int(canonical.size).to_bytes(8, "little", signed=False))
digest.update(memoryview(canonical).cast("B"))
return digest.hexdigest()
class WholeWaveformVerificationCache:
"""Request-local cache keyed by exact audio, target and verifier profile.
The cache deliberately accepts no process-global state. A caller must
instantiate it inside one synthesis request, and verifier calls that raise
are never cached. Passed and rejected gate evidence are both deterministic
evidence and may be reused only for an exact key match.
"""
def __init__(self) -> None:
self._entries: dict[
tuple[str, int, str, str],
TrajectoryGateResult,
] = {}
@property
def entry_count(self) -> int:
return len(self._entries)
def verify(
self,
audio: np.ndarray | Sequence[float],
sample_rate: int,
target_text: str,
verifier_profile: str,
verifier: Callable[[np.ndarray, int, str], TrajectoryGateResult],
) -> TrajectoryGateResult:
if not isinstance(target_text, str) or not target_text:
raise ValueError("target_text must be a non-empty string")
if not isinstance(verifier_profile, str) or not verifier_profile:
raise ValueError("verifier_profile must be a non-empty string")
if not callable(verifier):
raise ValueError("verifier must be callable")
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)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("sample_rate must be a positive integer") from error
if rate <= 0:
raise ValueError("sample_rate must be a positive integer")
waveform_hash = exact_waveform_sha256(waveform, rate)
key = (waveform_hash, rate, target_text, verifier_profile)
cached = self._entries.get(key)
if cached is not None:
return cached
verification = verifier(waveform, rate, target_text)
if not isinstance(verification, TrajectoryGateResult):
raise RuntimeError("whole-waveform verifier returned an invalid result")
self._entries[key] = verification
return verification
class WholeWaveformTranscriptCache:
"""Request-local ASR cache for one exact waveform and decoder profile.
Targets are intentionally absent from the key: transcription depends on
the immutable audio and pinned decoder contract, not on the text used by a
later semantic comparison. Exceptions and non-string results are never
cached.
"""
def __init__(self) -> None:
self._entries: dict[tuple[str, int, str], str] = {}
@property
def entry_count(self) -> int:
return len(self._entries)
def transcribe(
self,
audio: np.ndarray | Sequence[float],
sample_rate: int,
decoder_profile: str,
transcriber: Callable[[np.ndarray, int], str],
) -> str:
if not isinstance(decoder_profile, str) or not decoder_profile:
raise ValueError("decoder_profile must be a non-empty string")
if not callable(transcriber):
raise ValueError("transcriber must be callable")
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)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("sample_rate must be a positive integer") from error
if rate <= 0:
raise ValueError("sample_rate must be a positive integer")
key = (
exact_waveform_sha256(waveform, rate),
rate,
decoder_profile,
)
cached = self._entries.get(key)
if cached is not None:
return cached
transcript = transcriber(waveform, rate)
if not isinstance(transcript, str):
raise RuntimeError("whole-waveform transcriber returned an invalid result")
self._entries[key] = transcript
return transcript
def verify_trajectory(
observations: Sequence[CandidateObservation],
*,
chunk_artifacts: Sequence[ChunkCandidateArtifact] = (),
candidate_gate_kwargs_by_index: Sequence[dict[str, Any]] | None = None,
**candidate_gate_kwargs: Any,
) -> TrajectoryGateResult:
"""Require every chunk in a non-empty trajectory to pass all hard gates."""
try:
candidates = tuple(observations)
except TypeError:
candidates = ()
if not candidates:
return TrajectoryGateResult(False, (), math.inf, ("empty_trajectory",))
try:
artifacts = tuple(chunk_artifacts)
except TypeError:
artifacts = ()
if artifacts and (
len(artifacts) != len(candidates)
or any(not isinstance(artifact, ChunkCandidateArtifact) for artifact in artifacts)
):
return TrajectoryGateResult(
False,
(),
math.inf,
("invalid_chunk_artifacts",),
)
if candidate_gate_kwargs_by_index is None:
indexed_gate_kwargs = ({},) * len(candidates)
else:
raw_indexed_gate_kwargs: tuple[Any, ...] = ()
try:
raw_indexed_gate_kwargs = tuple(candidate_gate_kwargs_by_index)
indexed_gate_kwargs = tuple(
dict(kwargs)
for kwargs in raw_indexed_gate_kwargs
if type(kwargs) is dict
and set(kwargs).issubset(_CANDIDATE_GATE_OVERRIDE_KEYS)
)
except (TypeError, ValueError):
indexed_gate_kwargs = ()
if (
len(raw_indexed_gate_kwargs) != len(candidates)
or len(indexed_gate_kwargs) != len(candidates)
):
return TrajectoryGateResult(
False,
(),
math.inf,
("invalid_candidate_gate_overrides",),
)
results: list[CandidateGateResult] = []
rejection_reasons: list[str] = []
for index, (observation, indexed_kwargs) in enumerate(
zip(candidates, indexed_gate_kwargs, strict=True)
):
try:
result = verify_candidate(
observation,
**{**candidate_gate_kwargs, **indexed_kwargs},
)
except (TypeError, ValueError, OverflowError):
# A malformed observation must reject the entire trajectory.
comparison = compare_asr_text("", "")
result = CandidateGateResult(
passed=False,
comparison=comparison,
audio_duration_seconds=None,
speaker_gate_applied=False,
speaker_similarity=None,
boundary_speaker_drop=None,
pace_cps=None,
squim_gate_applied=False,
squim_stoi=None,
squim_pesq=None,
squim_si_sdr=None,
squim_quality_cost=None,
score=math.inf,
rejection_reasons=("malformed_observation",),
)
results.append(result)
rejection_reasons.extend(f"chunk_{index}:{reason}" for reason in result.rejection_reasons)
passed = bool(results) and all(result.passed for result in results)
score = sum(result.score for result in results) if passed else math.inf
if not math.isfinite(score):
passed = False
score = math.inf
if not rejection_reasons:
rejection_reasons.append("nonfinite_trajectory_score")
return TrajectoryGateResult(
passed=passed,
candidate_results=tuple(results),
score=score,
rejection_reasons=tuple(rejection_reasons),
chunk_artifacts=artifacts,
)
def qualify_trajectory_with_joined_output(
local_verification: TrajectoryGateResult,
joined_verification: TrajectoryGateResult,
) -> TrajectoryGateResult:
"""Require joined-output safety without discarding DP-local evidence.
``candidate_results`` and ``chunk_artifacts`` always stay local to the
generated chunks. This lets sequence DP reuse individually safe chunks
when RMS matching, fades, pauses or crossfade make the same-seed joined
waveform fail its whole-output gate.
"""
if not isinstance(local_verification, TrajectoryGateResult):
raise TypeError("local_verification must be a TrajectoryGateResult")
if local_verification.passed is not True or not math.isfinite(
local_verification.score
):
return local_verification
if not isinstance(joined_verification, TrajectoryGateResult):
raise TypeError("joined_verification must be a TrajectoryGateResult")
joined_passed = bool(
joined_verification.passed is True
and math.isfinite(joined_verification.score)
and len(joined_verification.candidate_results) == 1
)
if joined_passed:
return local_verification
joined_reasons = joined_verification.rejection_reasons or (
"invalid_joined_verification",
)
return TrajectoryGateResult(
passed=False,
candidate_results=local_verification.candidate_results,
score=math.inf,
rejection_reasons=tuple(
f"joined_output:{reason}" for reason in joined_reasons
),
chunk_artifacts=local_verification.chunk_artifacts,
)
def intersect_local_semantic_verification(
primary_verification: TrajectoryGateResult,
independent_verification: TrajectoryGateResult,
primary_indices: Sequence[int],
) -> TrajectoryGateResult:
"""Hard-intersect selected local rows without replacing acoustic evidence.
``primary_verification`` owns the Breeze25, speaker, SQUIM, pace and
boundary-proxy evidence consumed by the coverage selector.
``independent_verification`` contains a semantic-only projection for the
selected network-conditioned rows in the order given by
``primary_indices``. A secondary rejection is projected onto the
corresponding primary row with allow-listed semantic reason codes, making
it impossible for either the strict pool or the boundary-only proxy to
retain that local candidate. Production reuses the exact Breeze25
transcript cache, so this projection does not decode twice. Passing
intersections return the original primary object unchanged.
"""
if not isinstance(primary_verification, TrajectoryGateResult):
raise TypeError("primary_verification must be a TrajectoryGateResult")
if not isinstance(independent_verification, TrajectoryGateResult):
raise TypeError(
"independent_verification must be a TrajectoryGateResult"
)
try:
selected_indices = tuple(primary_indices)
except TypeError as error:
raise ValueError("primary_indices must be an ordered index sequence") from error
if (
not selected_indices
or any(
isinstance(index, (bool, np.bool_))
or not isinstance(index, int)
for index in selected_indices
)
or selected_indices != tuple(sorted(set(selected_indices)))
or any(
index < 0
or index >= len(primary_verification.candidate_results)
for index in selected_indices
)
or len(independent_verification.candidate_results)
!= len(selected_indices)
or any(
not isinstance(result, CandidateGateResult)
for result in independent_verification.candidate_results
)
):
raise ValueError(
"independent local verification rows do not match primary indices"
)
failed_secondary_rows = {
local_index
for local_index, result in enumerate(
independent_verification.candidate_results
)
if (
result.passed is not True
or not math.isfinite(result.score)
or bool(result.rejection_reasons)
)
}
if independent_verification.passed is not True and not failed_secondary_rows:
# A malformed/non-finite trajectory-level result must not allow any of
# its apparently passing local projections into the coverage pool.
failed_secondary_rows.update(range(len(selected_indices)))
if not failed_secondary_rows:
return primary_verification
combined_results = list(primary_verification.candidate_results)
for local_index in sorted(failed_secondary_rows):
primary_index = selected_indices[local_index]
primary_result = combined_results[primary_index]
independent_result = independent_verification.candidate_results[
local_index
]
semantic_reasons = ["semantic_gate"]
if (
"network_protected_span_mismatch"
in independent_result.rejection_reasons
):
semantic_reasons.append("network_protected_span_mismatch")
combined_results[primary_index] = replace(
primary_result,
passed=False,
score=math.inf,
rejection_reasons=tuple(
dict.fromkeys(
(*primary_result.rejection_reasons, *semantic_reasons)
)
),
)
rejection_reasons = tuple(
f"chunk_{index}:{reason}"
for index, result in enumerate(combined_results)
for reason in result.rejection_reasons
)
return TrajectoryGateResult(
passed=False,
candidate_results=tuple(combined_results),
score=math.inf,
rejection_reasons=rejection_reasons,
chunk_artifacts=primary_verification.chunk_artifacts,
)
class NoQualifiedCandidateError(RuntimeError):
"""Raised when the full adaptive cascade has no verified trajectory."""
def __init__(
self,
message: str,
*,
diagnostics: CascadeDiagnostics | None = None,
) -> None:
super().__init__(message)
self.diagnostics = diagnostics or CascadeDiagnostics()
class FinalOutputRejectedError(RuntimeError):
"""Raised when post-join output fails the final whole-waveform gate."""
def require_verified_final_output(
verification: TrajectoryGateResult,
) -> TrajectoryGateResult:
"""Return verified final evidence or reject without an audio fallback."""
if not isinstance(verification, TrajectoryGateResult):
raise FinalOutputRejectedError("final verifier returned an invalid result")
if (
verification.passed is not True
or not verification.candidate_results
or not all(result.passed for result in verification.candidate_results)
or not math.isfinite(verification.score)
):
reasons = ",".join(verification.rejection_reasons) or "unsafe_final_output"
raise FinalOutputRejectedError(f"final output rejected: {reasons}")
return verification
@dataclass(frozen=True)
class CascadeResult:
"""One selected trajectory and its bounded verification evidence.
``sequence_path_rank`` is the exact-verifier order across the progressively
expanded request-local lattice. It is not a global rank recomputed after
every later refill.
"""
trajectory: Any
verification: TrajectoryGateResult
seed: int | None
candidate_index: int | None
attempted_seeds: tuple[int, ...]
chunk_candidate_indices: tuple[int, ...] = ()
chunk_seeds: tuple[int, ...] = ()
selection_mode: str = "whole_trajectory"
sequence_path_rank: int | None = None
sequence_paths_checked: int = 0
diagnostics: CascadeDiagnostics = CascadeDiagnostics()
generated_chunk_count: int = 0
generated_text_units: int = 0
chunk_candidate_counts: tuple[int, ...] = ()
def _candidate_gate_evidence_payload(
evidence: CandidateGateEvidence,
) -> dict[str, Any]:
return {
"passed": evidence.passed,
"target_units": evidence.target_units,
"hypothesis_units": evidence.hypothesis_units,
"edit_distance": evidence.edit_distance,
"cer": evidence.cer,
"prefix_cer": evidence.prefix_cer,
"suffix_cer": evidence.suffix_cer,
"prefix_deletions": evidence.prefix_deletions,
"suffix_deletions": evidence.suffix_deletions,
"extra_tail_units": evidence.extra_tail_units,
"audio_duration_seconds": evidence.audio_duration_seconds,
"speaker_gate_applied": evidence.speaker_gate_applied,
"speaker_similarity": evidence.speaker_similarity,
"boundary_speaker_drop": evidence.boundary_speaker_drop,
"pace_cps": evidence.pace_cps,
"squim_gate_applied": evidence.squim_gate_applied,
"squim_stoi": evidence.squim_stoi,
"squim_pesq": evidence.squim_pesq,
"squim_si_sdr": evidence.squim_si_sdr,
"squim_quality_cost": evidence.squim_quality_cost,
"score": evidence.score,
"reasons": list(evidence.rejection_reasons),
}
def _trajectory_gate_evidence_payload(
evidence: TrajectoryGateEvidence | None,
) -> dict[str, Any] | None:
if evidence is None:
return None
return {
"passed": evidence.passed,
"result_count": evidence.result_count,
"score": evidence.score,
"reasons": list(evidence.rejection_reasons),
"result": (
None
if evidence.result is None
else _candidate_gate_evidence_payload(evidence.result)
),
}
def _local_independent_gate_evidence_payload(
evidence: LocalIndependentGateEvidence,
) -> dict[str, Any]:
return {
"attempted": evidence.attempted,
"passed": evidence.passed,
"proof_count": evidence.proof_count,
"result": (
None
if evidence.result is None
else _candidate_gate_evidence_payload(evidence.result)
),
}
def _candidate_attempt_evidence_payload(
evidence: CandidateAttemptEvidence,
) -> dict[str, Any]:
candidate_ordinals = evidence.chunk_candidate_ordinals
chunk_policies = [
generation_policy_for_candidate_offset(ordinal).name
for ordinal in candidate_ordinals
]
policy = (
chunk_policies[0]
if chunk_policies and len(set(chunk_policies)) == 1
else generation_policy_for_candidate_offset(evidence.candidate_index).name
)
return {
"candidate_index": evidence.candidate_index,
"seed": evidence.seed,
"policy": policy,
"trajectory_passed": evidence.trajectory_passed,
"trajectory_score": evidence.trajectory_score,
"trajectory_reasons": list(
evidence.trajectory_rejection_reasons
),
"chunk_indices": list(evidence.chunk_indices),
"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),
"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],
"local_result_count": evidence.local_result_count,
"local_results": [
_candidate_gate_evidence_payload(result)
for result in evidence.local_results
],
"independent_local_evidence_complete": (
len(evidence.independent_local_results)
== evidence.local_result_count
),
"independent_local_results": [
_local_independent_gate_evidence_payload(result)
for result in evidence.independent_local_results
],
"joined_output": _trajectory_gate_evidence_payload(
evidence.joined_output
),
"independent_output": _trajectory_gate_evidence_payload(
evidence.independent_output
),
}
def _sequence_search_evidence_payload(
evidence: SequenceSearchEvidence | None,
) -> dict[str, Any] | None:
if evidence is None:
return None
eligible_counts = evidence.eligible_candidate_counts[
:CASCADE_EVIDENCE_MAX_LOCAL_RESULTS
]
transition_counts = evidence.finite_transition_counts[
: max(0, CASCADE_EVIDENCE_MAX_LOCAL_RESULTS - 1)
]
return {
"row_count": len(evidence.eligible_candidate_counts),
"eligible_candidate_counts": list(eligible_counts),
"zero_eligible_rows": [
index
for index, count in enumerate(eligible_counts)
if count == 0
],
"transition_boundary_count": len(evidence.finite_transition_counts),
"finite_transition_counts": list(transition_counts),
"zero_transition_edges": [
index
for index, count in enumerate(transition_counts)
if count == 0
],
"ranked_path_count": evidence.ranked_path_count,
"checked_path_count": len(evidence.checked_paths),
"checked_paths": [
{
"rank": path.rank,
"chunk_candidates": list(
path.chunk_candidate_indices[
:CASCADE_EVIDENCE_MAX_LOCAL_RESULTS
]
),
"chunk_seeds": list(
path.chunk_seeds[:CASCADE_EVIDENCE_MAX_LOCAL_RESULTS]
),
"final_output": _trajectory_gate_evidence_payload(
path.final_output
),
}
for path in evidence.checked_paths[
:CASCADE_EVIDENCE_MAX_SEQUENCE_PATHS
]
],
}
def _selected_generation_evidence_payload(
diagnostics: CascadeDiagnostics,
selection: CascadeResult,
) -> dict[str, Any]:
attempts = {
attempt.candidate_index: attempt for attempt in diagnostics.attempts
}
scheduled_cfgs: list[float | None] = []
effective_cfgs: list[float | None] = []
floor_reasons: list[list[str]] = []
candidate_ordinals: list[int | None] = []
policies: list[str | None] = []
network_conditioned: list[bool | None] = []
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
):
attempt = attempts.get(candidate_index)
if attempt is None or attempt.scheduled_cfg is None:
complete = False
scheduled_cfgs.append(None)
effective_cfgs.append(None)
floor_reasons.append([])
candidate_ordinals.append(None)
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)
effective = attempt.effective_cfgs[local_index]
reasons = attempt.floor_reasons[local_index]
except (IndexError, ValueError):
complete = False
scheduled_cfgs.append(attempt.scheduled_cfg)
effective_cfgs.append(None)
floor_reasons.append([])
candidate_ordinals.append(None)
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]
except IndexError:
complete = False
ordinal = None
scheduled_cfgs.append(attempt.scheduled_cfg)
effective_cfgs.append(effective)
floor_reasons.append(list(reasons))
candidate_ordinals.append(ordinal)
policies.append(
None
if ordinal is None
else generation_policy_for_candidate_offset(ordinal).name
)
try:
network_conditioned.append(attempt.network_conditioned[local_index])
except IndexError:
complete = False
network_conditioned.append(None)
try:
chunk_text_variants.append(attempt.chunk_text_variants[local_index])
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,
"chunk_candidate_ordinals": candidate_ordinals,
"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,
*,
outcome: str,
generated_chunk_limit: int,
generated_text_unit_limit: int = CASCADE_EVIDENCE_MAX_TEXT_UNITS,
selection: CascadeResult | None = None,
final_output: TrajectoryGateEvidence | None = None,
independent_final_output: TrajectoryGateEvidence | None = None,
) -> str:
"""Return one canonical JSON log line containing only bounded evidence."""
if not isinstance(diagnostics, CascadeDiagnostics):
raise TypeError("diagnostics must be CascadeDiagnostics")
if outcome not in _CASCADE_EVIDENCE_OUTCOMES:
raise ValueError("invalid cascade evidence outcome")
if isinstance(generated_chunk_limit, (bool, np.bool_)):
raise ValueError("generated_chunk_limit must be an integer between 1 and 32")
try:
limit = operator.index(generated_chunk_limit)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError(
"generated_chunk_limit must be an integer between 1 and 32"
) from error
if not 1 <= limit <= CASCADE_EVIDENCE_MAX_ATTEMPTS:
raise ValueError("generated_chunk_limit must be an integer between 1 and 32")
if isinstance(generated_text_unit_limit, (bool, np.bool_)):
raise ValueError("generated_text_unit_limit must be an integer between 1 and 800")
try:
text_unit_limit = operator.index(generated_text_unit_limit)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError(
"generated_text_unit_limit must be an integer between 1 and 800"
) from error
if not 1 <= text_unit_limit <= CASCADE_EVIDENCE_MAX_TEXT_UNITS:
raise ValueError(
"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):
raise TypeError("selection must be a CascadeResult")
mode = (
selection.selection_mode
if selection.selection_mode in _CASCADE_EVIDENCE_SELECTION_MODES
else None
)
selected = {
"selection": mode,
"candidate_index": selection.candidate_index,
"seed": selection.seed,
"chunk_candidates": list(
selection.chunk_candidate_indices[
:CASCADE_EVIDENCE_MAX_LOCAL_RESULTS
]
),
"chunk_seeds": list(
selection.chunk_seeds[:CASCADE_EVIDENCE_MAX_LOCAL_RESULTS]
),
"sequence_rank": selection.sequence_path_rank,
"sequence_paths_checked": selection.sequence_paths_checked,
"generated_chunks": selection.generated_chunk_count,
"generated_text_units": selection.generated_text_units,
"chunk_candidate_counts": list(
selection.chunk_candidate_counts[
:CASCADE_EVIDENCE_MAX_LOCAL_RESULTS
]
),
"generation": _selected_generation_evidence_payload(
diagnostics,
selection,
),
}
generation_evidence_complete = bool(attempts) and all(
attempt.scheduled_cfg is not None
and len(attempt.chunk_indices) == len(attempt.chunk_text_units)
== len(attempt.chunk_candidate_ordinals)
== len(attempt.network_conditioned)
== len(attempt.chunk_text_variants)
== len(attempt.effective_cfgs)
== len(attempt.floor_reasons)
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
)
request_chunk_count = (
len(attempts[0].chunk_indices) if attempts else 0
)
if selection is not None and generation_evidence_complete:
if (
selection.generated_chunk_count != generated_chunk_count
or selection.generated_text_units != generated_text_units
):
raise ValueError(
"canonical generation evidence disagrees with cascade budgets"
)
payload = {
"schema_version": CASCADE_EVIDENCE_SCHEMA_VERSION,
"outcome": outcome,
"request_chunk_count": request_chunk_count,
"limits": {
"generated_chunks": int(limit),
"generated_text_units": int(text_unit_limit),
},
"cfg_contract": {
"schedule": MIXED_CFG_SCHEDULE,
"primary": MIXED_CFG_PRIMARY,
"alternate": MIXED_CFG_ALTERNATE,
"short_text_max_units": MIXED_CFG_SHORT_TEXT_MAX_UNITS,
"short_text_min": MIXED_CFG_SHORT_TEXT_MIN,
"network_min": MIXED_CFG_NETWORK_MIN,
},
"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": [
_candidate_attempt_evidence_payload(attempt)
for attempt in attempts
],
"sequence_search": _sequence_search_evidence_payload(
diagnostics.sequence_search
),
"selection": selected,
"final_output": _trajectory_gate_evidence_payload(final_output),
"independent_final_output": _trajectory_gate_evidence_payload(
independent_final_output
),
}
return CASCADE_EVIDENCE_LOG_PREFIX + json.dumps(
payload,
allow_nan=False,
ensure_ascii=True,
separators=(",", ":"),
sort_keys=True,
)
def select_k_candidate_sequences(
local_scores: Sequence[Sequence[float]],
transition_scores: Sequence[Sequence[Sequence[float]]] = (),
*,
max_paths: int = 3,
) -> tuple[CandidateSequenceSelection, ...]:
"""Return up to three distinct finite paths in stable cost order.
Each DP state retains only its ``max_paths`` best prefixes. This keeps the
search bounded at ``O(N K² max_paths)`` while still producing exact k-best
paths for the requested small bound. Single-chunk requests intentionally
return no sequence fallback.
"""
if isinstance(max_paths, (bool, np.bool_)):
raise ValueError("max_paths must be an integer between 1 and 3")
try:
path_limit = operator.index(max_paths)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("max_paths must be an integer between 1 and 3") from error
if not 1 <= path_limit <= 3:
raise ValueError("max_paths must be an integer between 1 and 3")
try:
raw_local = [list(row) for row in local_scores]
except TypeError:
return ()
if len(raw_local) <= 1 or any(not row for row in raw_local):
return ()
safe_local = [
[
score if score is not None else math.inf
for score in (_finite_float(value, minimum=0.0) for value in row)
]
for row in raw_local
]
try:
raw_transitions = [
[list(row) for row in matrix]
for matrix in transition_scores
]
except TypeError:
return ()
if len(raw_transitions) != len(safe_local) - 1:
return ()
safe_transitions: list[list[list[float]]] = []
for step, matrix in enumerate(raw_transitions):
previous_count = len(safe_local[step])
current_count = len(safe_local[step + 1])
if len(matrix) != previous_count or any(
len(row) != current_count for row in matrix
):
return ()
safe_transitions.append(
[
[
score if score is not None else math.inf
for score in (
_finite_float(value, minimum=0.0)
for value in row
)
]
for row in matrix
]
)
# One list of (cost, path) prefixes for each current candidate position.
states: list[list[tuple[float, tuple[int, ...]]]] = []
for candidate_index, score in enumerate(safe_local[0]):
states.append(
[(score, (candidate_index,))] if math.isfinite(score) else []
)
for step in range(1, len(safe_local)):
next_states: list[list[tuple[float, tuple[int, ...]]]] = []
for current_index, local_score in enumerate(safe_local[step]):
options: dict[tuple[int, ...], float] = {}
if math.isfinite(local_score):
for previous_index, prefixes in enumerate(states):
edge_score = safe_transitions[step - 1][previous_index][
current_index
]
if not math.isfinite(edge_score):
continue
for previous_score, prefix in prefixes:
total = previous_score + edge_score + local_score
path = prefix + (current_index,)
if math.isfinite(total):
old_score = options.get(path, math.inf)
if total < old_score:
options[path] = total
ranked = sorted(
((score, path) for path, score in options.items()),
key=lambda item: (item[0], item[1]),
)[:path_limit]
next_states.append(ranked)
states = next_states
complete: dict[tuple[int, ...], float] = {}
for prefixes in states:
for score, path in prefixes:
old_score = complete.get(path, math.inf)
if score < old_score:
complete[path] = score
ranked_complete = sorted(
((score, path) for path, score in complete.items()),
key=lambda item: (item[0], item[1]),
)[:path_limit]
return tuple(
CandidateSequenceSelection(candidate_indices=path, total_score=score)
for score, path in ranked_complete
)
def select_culprit_diverse_candidate_sequences(
local_scores: Sequence[Sequence[float]],
transition_scores: Sequence[Sequence[Sequence[float]]] = (),
*,
culprit_indices: Sequence[int] = (),
excluded_paths: Sequence[Sequence[int]] = (),
max_paths: int = 3,
) -> tuple[CandidateSequenceSelection, ...]:
"""Select bounded low-cost paths with distinct culprit projections.
The regular k-best result can spend all three exact-final checks on paths
that differ only in already-stable chunks. This helper adds the cheapest
path forced through every finite row candidate, then prefers previously
unseen assignments at the supplied culprit chunks. It remains bounded by
the total ragged candidate count (at most the generation-chunk budget).
"""
if isinstance(max_paths, (bool, np.bool_)):
raise ValueError("max_paths must be an integer between 1 and 3")
try:
path_limit = operator.index(max_paths)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("max_paths must be an integer between 1 and 3") from error
if not 1 <= path_limit <= 3:
raise ValueError("max_paths must be an integer between 1 and 3")
try:
raw_local = [list(row) for row in local_scores]
except TypeError:
return ()
if not raw_local or any(not row for row in raw_local):
return ()
if len(raw_local) == 1:
try:
if list(transition_scores):
return ()
except TypeError:
return ()
culprit_rows: list[int] = []
try:
for value in culprit_indices:
if isinstance(value, (bool, np.bool_)):
return ()
index = operator.index(value)
if index < 0 or index >= len(raw_local):
return ()
if index not in culprit_rows:
culprit_rows.append(index)
except (TypeError, ValueError, OverflowError):
return ()
projection_rows = tuple(culprit_rows) or tuple(range(len(raw_local)))
excluded: set[tuple[int, ...]] = set()
try:
for raw_path in excluded_paths:
path_values = list(raw_path)
if any(isinstance(value, (bool, np.bool_)) for value in path_values):
return ()
path = tuple(operator.index(value) for value in path_values)
if len(path) != len(raw_local):
return ()
if any(
index < 0 or index >= len(raw_local[row])
for row, index in enumerate(path)
):
return ()
excluded.add(path)
except (TypeError, ValueError, OverflowError):
return ()
candidates: dict[tuple[int, ...], float] = {}
def retain(selection: CandidateSequenceSelection) -> None:
path = tuple(selection.candidate_indices)
score = _finite_float(selection.total_score, minimum=0.0)
if len(path) != len(raw_local) or path in excluded or score is None:
return
old_score = candidates.get(path, math.inf)
if score < old_score:
candidates[path] = score
if len(raw_local) == 1:
for candidate_index, value in enumerate(raw_local[0]):
score = _finite_float(value, minimum=0.0)
if score is not None:
retain(
CandidateSequenceSelection(
candidate_indices=(candidate_index,),
total_score=score,
)
)
else:
for selection in select_k_candidate_sequences(
raw_local,
transition_scores,
max_paths=path_limit,
):
retain(selection)
for row_index, row in enumerate(raw_local):
for candidate_index, value in enumerate(row):
if _finite_float(value, minimum=0.0) is None:
continue
forced = [list(scores) for scores in raw_local]
forced[row_index] = [
score if index == candidate_index else math.inf
for index, score in enumerate(forced[row_index])
]
best = select_k_candidate_sequences(
forced,
transition_scores,
max_paths=1,
)
if best:
retain(best[0])
ranked = sorted(
(
CandidateSequenceSelection(candidate_indices=path, total_score=score)
for path, score in candidates.items()
),
key=lambda selection: (
selection.total_score,
selection.candidate_indices,
),
)
if not ranked:
return ()
selected = [ranked.pop(0)]
seen_projections = {
tuple(selected[0].candidate_indices[index] for index in projection_rows)
}
while ranked and len(selected) < path_limit:
diverse_index = next(
(
index
for index, selection in enumerate(ranked)
if tuple(
selection.candidate_indices[row] for row in projection_rows
)
not in seen_projections
),
None,
)
selected_index = 0 if diverse_index is None else diverse_index
selection = ranked.pop(selected_index)
selected.append(selection)
seen_projections.add(
tuple(selection.candidate_indices[index] for index in projection_rows)
)
return tuple(selected)
def candidate_chunk_transition_score(
previous_result: CandidateGateResult,
previous_artifact: ChunkCandidateArtifact,
current_result: CandidateGateResult,
current_artifact: ChunkCandidateArtifact,
*,
speaker_weight: float = 1.0,
rms_db_weight: float = 0.05,
median_f0_weight: float = 0.10,
) -> float:
"""Return a finite adjacent-chunk cost or ``inf`` for an unsafe edge."""
if previous_result.passed is not True or current_result.passed is not True:
return math.inf
speaker_w = _finite_float(speaker_weight, minimum=0.0)
rms_w = _finite_float(rms_db_weight, minimum=0.0)
f0_w = _finite_float(median_f0_weight, minimum=0.0)
previous_rms = _finite_float(previous_artifact.rms_db)
current_rms = _finite_float(current_artifact.rms_db)
if None in (speaker_w, rms_w, f0_w, previous_rms, current_rms):
return math.inf
previous_embedding = previous_artifact.speaker_embedding
current_embedding = current_artifact.speaker_embedding
speaker_cost = 0.0
if previous_result.speaker_gate_applied and previous_embedding is None:
return math.inf
if current_result.speaker_gate_applied and current_embedding is None:
return math.inf
if previous_embedding is not None and current_embedding is not None:
try:
speaker_cost = 1.0 - cosine_similarity(previous_embedding, current_embedding)
except ValueError:
return math.inf
assert previous_rms is not None and current_rms is not None
rms_cost = abs(previous_rms - current_rms)
f0_cost = 0.0
previous_f0 = previous_artifact.median_f0_hz
current_f0 = current_artifact.median_f0_hz
if previous_f0 is not None and current_f0 is not None:
previous_pitch = _finite_float(previous_f0, minimum=1.0)
current_pitch = _finite_float(current_f0, minimum=1.0)
if previous_pitch is None or current_pitch is None:
return math.inf
f0_cost = abs(math.log2(current_pitch / previous_pitch))
assert speaker_w is not None and rms_w is not None and f0_w is not None
score = speaker_w * speaker_cost + rms_w * rms_cost + f0_w * f0_cost
return score if math.isfinite(score) and score >= 0.0 else math.inf
def candidate_limit_for_chunk_budget(
chunk_count: int,
*,
max_candidates: int = 32,
max_generated_chunks: int = 32,
total_text_units: int | None = None,
max_generated_text_units: int | None = None,
) -> int:
"""Return a cap bounded by chunk count and optional text-generation work."""
if isinstance(chunk_count, (bool, np.bool_)):
raise ValueError("chunk_count must be a positive integer")
try:
chunks = int(chunk_count)
candidates = int(max_candidates)
generated_chunks = int(max_generated_chunks)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("candidate budget values must be integers") from error
if chunks <= 0:
raise ValueError("chunk_count must be a positive integer")
if candidates <= 0 or candidates > ADAPTIVE_CASCADE_STAGE_LIMITS[-1]:
raise ValueError("max_candidates must be between 1 and 32")
if generated_chunks <= 0:
raise ValueError("max_generated_chunks must be positive")
if chunks > generated_chunks:
raise ValueError("one trajectory exceeds the generated-chunk budget")
limit = min(candidates, generated_chunks // chunks)
if total_text_units is None and max_generated_text_units is None:
return limit
if total_text_units is None or max_generated_text_units is None:
raise ValueError("text-unit budget fields must be provided together")
if isinstance(total_text_units, (bool, np.bool_)) or isinstance(
max_generated_text_units,
(bool, np.bool_),
):
raise ValueError("text-unit budget values must be integers")
try:
units = int(total_text_units)
generated_units = int(max_generated_text_units)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("text-unit budget values must be integers") from error
if units <= 0 or generated_units <= 0:
raise ValueError("text-unit budget values must be positive")
if units > generated_units:
raise ValueError("one trajectory exceeds the generated-text-unit budget")
return min(limit, generated_units // units)
@dataclass(frozen=True)
class _VerifiedTrajectoryCandidate:
candidate_index: int
seed: int
trajectory: Any
verification: TrajectoryGateResult
independent_local_results: tuple[LocalIndependentGateEvidence, ...] = ()
joined_output: TrajectoryGateEvidence | None = None
independent_output: TrajectoryGateEvidence | None = None
generation_evidence: CandidateGenerationEvidence | None = None
def _unwrap_candidate_verification(
value: Any,
) -> tuple[
TrajectoryGateResult,
tuple[LocalIndependentGateEvidence, ...],
TrajectoryGateEvidence | None,
TrajectoryGateEvidence | None,
]:
if isinstance(value, CandidateVerification):
verification = value.verification
independent_local_results = value.independent_local_results
joined_output = value.joined_output
independent_output = value.independent_output
else:
verification = value
independent_local_results = ()
joined_output = None
independent_output = None
if not isinstance(verification, TrajectoryGateResult):
raise TypeError("candidate_verifier must return TrajectoryGateResult")
if not isinstance(independent_local_results, tuple) or (
independent_local_results
and (
len(independent_local_results)
!= len(verification.candidate_results)
or any(
not isinstance(result, LocalIndependentGateEvidence)
or type(result.attempted) is not bool
or type(result.proof_count) is not int
or result.proof_count > CASCADE_EVIDENCE_MAX_LOCAL_RESULTS
or (
result.attempted
and (
type(result.passed) is not bool
or result.proof_count <= 0
or not isinstance(result.result, CandidateGateEvidence)
or result.result.passed is not result.passed
)
)
or (
not result.attempted
and (
result.passed is not None
or result.proof_count < 0
or result.result is not None
)
)
for result in independent_local_results
)
)
):
raise TypeError("candidate independent-local evidence is invalid")
if joined_output is not None and not isinstance(
joined_output,
TrajectoryGateEvidence,
):
raise TypeError("candidate joined-output evidence is invalid")
if independent_output is not None and not isinstance(
independent_output,
TrajectoryGateEvidence,
):
raise TypeError("candidate independent-output evidence is invalid")
return (
verification,
independent_local_results,
joined_output,
independent_output,
)
def _validated_candidate_generation_evidence(
value: Any,
*,
chunk_indices: tuple[int, ...],
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."""
if not isinstance(value, CandidateGenerationEvidence):
raise TypeError(
"generation evidence factory must return CandidateGenerationEvidence"
)
if value.chunk_indices != chunk_indices:
raise ValueError("generation evidence chunk indices do not match the attempt")
expected_units = tuple(count_speech_units(chunk) for chunk in chunks)
if value.chunk_text_units != expected_units or any(unit <= 0 for unit in expected_units):
raise ValueError("generation evidence text units do not match the attempt")
if len(value.effective_cfgs) != len(chunks) or len(value.floor_reasons) != len(chunks):
raise ValueError("generation evidence CFG rows do not match the attempt")
if value.network_conditioned:
if (
not isinstance(value.network_conditioned, tuple)
or len(value.network_conditioned) != len(chunks)
or any(type(flag) is not bool for flag in value.network_conditioned)
):
raise ValueError(
"generation evidence network provenance does not match the attempt"
)
network_conditioned = value.network_conditioned
else:
network_conditioned = (False,) * len(chunks)
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)
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:
raise ValueError(
"context-aware generation evidence must provide chunk candidate ordinals"
)
candidate_ordinals = expected_candidate_ordinals
else:
if not isinstance(raw_ordinals, tuple) or len(raw_ordinals) != len(chunks):
raise ValueError("generation evidence candidate ordinals do not match the attempt")
candidate_ordinals_list: list[int] = []
for raw_ordinal in raw_ordinals:
if isinstance(raw_ordinal, (bool, np.bool_)):
raise ValueError("generation evidence candidate ordinal is invalid")
try:
ordinal = operator.index(raw_ordinal)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError(
"generation evidence candidate ordinal is invalid"
) from error
if ordinal < 0:
raise ValueError("generation evidence candidate ordinal is invalid")
candidate_ordinals_list.append(int(ordinal))
candidate_ordinals = tuple(candidate_ordinals_list)
if candidate_ordinals != expected_candidate_ordinals:
raise ValueError(
"generation evidence candidate ordinals do not match the schedule"
)
if len(set(candidate_ordinals)) != 1:
raise ValueError("one generation call must use one candidate schedule ordinal")
scheduled = _finite_float(value.scheduled_cfg, minimum=1.0, maximum=4.0)
if scheduled is None:
raise ValueError("generation evidence scheduled CFG is invalid")
expected_scheduled = generation_cfg_for_candidate_offset(candidate_ordinals[0])
if scheduled != expected_scheduled:
raise ValueError("generation evidence scheduled CFG disagrees with the schedule")
effective_cfgs: list[float] = []
floor_reasons: list[tuple[str, ...]] = []
for effective_value, raw_reasons, is_network in zip(
value.effective_cfgs,
value.floor_reasons,
network_conditioned,
strict=True,
):
effective = _finite_float(effective_value, minimum=1.0, maximum=4.0)
if effective is None or effective < scheduled:
raise ValueError("generation evidence effective CFG is invalid")
if not isinstance(raw_reasons, tuple) or any(
reason not in _CFG_FLOOR_REASONS for reason in raw_reasons
):
raise ValueError("generation evidence floor reasons are invalid")
reasons = tuple(dict.fromkeys(raw_reasons))
if len(reasons) != len(raw_reasons):
raise ValueError("generation evidence floor reasons must be unique")
if (not reasons and effective != scheduled) or (
reasons and effective <= scheduled
):
raise ValueError("generation evidence floor reasons disagree with CFG")
network_floor_required = bool(
is_network and scheduled < MIXED_CFG_NETWORK_MIN
)
if ("network" in reasons) != network_floor_required:
raise ValueError(
"generation evidence network floor disagrees with provenance"
)
if is_network and effective < MIXED_CFG_NETWORK_MIN:
raise ValueError(
"generation evidence network CFG is below the frozen minimum"
)
effective_cfgs.append(effective)
floor_reasons.append(reasons)
return CandidateGenerationEvidence(
chunk_indices=chunk_indices,
chunk_text_units=expected_units,
scheduled_cfg=scheduled,
effective_cfgs=tuple(effective_cfgs),
floor_reasons=tuple(floor_reasons),
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),
)
def _candidate_attempt_evidence(
candidate: _VerifiedTrajectoryCandidate,
) -> CandidateAttemptEvidence:
verification = candidate.verification
generation = candidate.generation_evidence
local_results = verification.candidate_results[
:CASCADE_EVIDENCE_MAX_LOCAL_RESULTS
]
return CandidateAttemptEvidence(
candidate_index=_evidence_int(candidate.candidate_index),
seed=_evidence_int(
candidate.seed,
maximum=REQUEST_SEED_LIMIT + CASCADE_EVIDENCE_MAX_ATTEMPTS,
),
trajectory_passed=verification.passed is True,
trajectory_score=_evidence_float(verification.score),
trajectory_rejection_reasons=_bounded_rejection_reasons(
verification.rejection_reasons
),
local_result_count=_evidence_int(len(verification.candidate_results)),
local_results=tuple(
candidate_gate_evidence(result)
for result in local_results
if isinstance(result, CandidateGateResult)
),
chunk_indices=(generation.chunk_indices if generation is not None else ()),
chunk_text_units=(
generation.chunk_text_units if generation is not None else ()
),
chunk_candidate_ordinals=(
generation.chunk_candidate_ordinals if generation is not None else ()
),
network_conditioned=(
generation.network_conditioned if generation is not None else ()
),
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 ()),
independent_local_results=candidate.independent_local_results,
joined_output=candidate.joined_output,
independent_output=candidate.independent_output,
)
def _cascade_diagnostics(
candidates: Sequence[_VerifiedTrajectoryCandidate],
sequence_search: SequenceSearchEvidence | None = None,
) -> CascadeDiagnostics:
return CascadeDiagnostics(
attempts=tuple(
_candidate_attempt_evidence(candidate)
for candidate in tuple(candidates)[:CASCADE_EVIDENCE_MAX_ATTEMPTS]
),
sequence_search=sequence_search,
)
@dataclass(frozen=True)
class _RaggedChunkCandidate:
"""One locally gated chunk retained in a coverage-adaptive pool."""
candidate_index: int
seed: int
audio: Any
result: CandidateGateResult
artifact: ChunkCandidateArtifact
def _whole_trajectory_result(
candidate: _VerifiedTrajectoryCandidate,
attempted_seeds: Sequence[int],
chunk_count: int,
*,
diagnostics: CascadeDiagnostics,
) -> CascadeResult:
return CascadeResult(
trajectory=candidate.trajectory,
verification=candidate.verification,
seed=candidate.seed,
candidate_index=candidate.candidate_index,
attempted_seeds=tuple(attempted_seeds),
chunk_candidate_indices=(candidate.candidate_index,) * chunk_count,
chunk_seeds=(candidate.seed,) * chunk_count,
selection_mode="whole_trajectory",
diagnostics=diagnostics,
)
@dataclass(frozen=True)
class _SequenceFallbackSearchResult:
results: tuple[CascadeResult, ...]
evidence: SequenceSearchEvidence
def _sequence_fallback_search(
candidates: Sequence[_VerifiedTrajectoryCandidate],
attempted_seeds: Sequence[int],
chunk_count: int,
*,
max_paths: int,
max_local_boundary_speaker_drop: float | None = None,
) -> _SequenceFallbackSearchResult:
"""Rank mixed-seed paths using safe or narrowly recoverable local chunks.
A boundary-only local rejection may be made eligible up to the supplied
fallback cap. This is intentionally an internal DP representation: the
caller must still verify the exactly assembled mixed path with the stricter
joined/final gate before any audio can be returned.
"""
if not candidates:
return _SequenceFallbackSearchResult(
results=(),
evidence=SequenceSearchEvidence(
eligible_candidate_counts=(0,) * max(0, chunk_count),
finite_transition_counts=(0,) * max(0, chunk_count - 1),
ranked_path_count=0,
),
)
if chunk_count <= 1:
eligible = 0
if chunk_count == 1:
for candidate in candidates:
results = candidate.verification.candidate_results
if len(results) != 1:
continue
result = results[0]
if (
isinstance(result, CandidateGateResult)
and result.passed is True
and math.isfinite(result.score)
and result.score >= 0.0
):
eligible += 1
return _SequenceFallbackSearchResult(
results=(),
evidence=SequenceSearchEvidence(
eligible_candidate_counts=((eligible,) if chunk_count == 1 else ()),
finite_transition_counts=(),
ranked_path_count=0,
),
)
local_scores: list[list[float]] = [[] for _ in range(chunk_count)]
usable: list[bool] = []
sequence_results: list[tuple[CandidateGateResult, ...]] = []
for candidate in candidates:
verification = candidate.verification
try:
trajectory_length = len(candidate.trajectory)
except TypeError:
trajectory_length = -1
candidate_usable = bool(
trajectory_length == chunk_count
and len(verification.candidate_results) == chunk_count
and len(verification.chunk_artifacts) == chunk_count
)
usable.append(candidate_usable)
adjusted_results: list[CandidateGateResult] = []
for chunk_index in range(chunk_count):
score = math.inf
if candidate_usable:
result = verification.candidate_results[chunk_index]
artifact = verification.chunk_artifacts[chunk_index]
adjusted_result = _sequence_fallback_candidate_result(
result,
max_local_boundary_speaker_drop=max_local_boundary_speaker_drop,
)
adjusted_results.append(adjusted_result)
if (
adjusted_result.passed
and math.isfinite(adjusted_result.score)
and adjusted_result.score >= 0.0
):
speaker_artifact_valid = True
if adjusted_result.speaker_gate_applied:
try:
speaker_artifact_valid = bool(
artifact.speaker_embedding is not None
and cosine_similarity(
artifact.speaker_embedding,
artifact.speaker_embedding,
) >= 1.0 - 1.0e-6
)
except ValueError:
speaker_artifact_valid = False
if speaker_artifact_valid:
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))
transitions: list[list[list[float]]] = []
for chunk_index in range(1, chunk_count):
matrix: list[list[float]] = []
for previous_position, previous_candidate in enumerate(candidates):
row: list[float] = []
for current_position, current_candidate in enumerate(candidates):
score = math.inf
if usable[previous_position] and usable[current_position]:
score = candidate_chunk_transition_score(
sequence_results[previous_position][chunk_index - 1],
previous_candidate.verification.chunk_artifacts[chunk_index - 1],
sequence_results[current_position][chunk_index],
current_candidate.verification.chunk_artifacts[chunk_index],
)
row.append(score)
matrix.append(row)
transitions.append(matrix)
selections = select_k_candidate_sequences(
local_scores,
transitions,
max_paths=max_paths,
)
output: list[CascadeResult] = []
for rank, selection in enumerate(selections, 1):
selected_candidates = tuple(
candidates[position]
for position in selection.candidate_indices
)
selected_trajectory = tuple(
candidate.trajectory[chunk_index]
for chunk_index, candidate in enumerate(selected_candidates)
)
selected_results = tuple(
sequence_results[position][chunk_index]
for chunk_index, position in enumerate(selection.candidate_indices)
)
selected_artifacts = tuple(
candidate.verification.chunk_artifacts[chunk_index]
for chunk_index, candidate in enumerate(selected_candidates)
)
verification = TrajectoryGateResult(
passed=True,
candidate_results=selected_results,
score=selection.total_score,
rejection_reasons=(),
chunk_artifacts=selected_artifacts,
)
output.append(
CascadeResult(
trajectory=selected_trajectory,
verification=verification,
seed=None,
candidate_index=None,
attempted_seeds=tuple(attempted_seeds),
chunk_candidate_indices=tuple(
candidate.candidate_index for candidate in selected_candidates
),
chunk_seeds=tuple(
candidate.seed for candidate in selected_candidates
),
selection_mode="sequence_dp",
sequence_path_rank=rank,
)
)
results = tuple(output)
evidence = SequenceSearchEvidence(
eligible_candidate_counts=tuple(
sum(math.isfinite(score) for score in row)
for row in local_scores
),
finite_transition_counts=tuple(
sum(
math.isfinite(score)
for row in matrix
for score in row
)
for matrix in transitions
),
ranked_path_count=len(results),
)
return _SequenceFallbackSearchResult(results=results, evidence=evidence)
def _sequence_fallback_results(
candidates: Sequence[_VerifiedTrajectoryCandidate],
attempted_seeds: Sequence[int],
chunk_count: int,
*,
max_paths: int,
max_local_boundary_speaker_drop: float | None = None,
) -> tuple[CascadeResult, ...]:
"""Compatibility wrapper returning only ranked sequence results."""
return _sequence_fallback_search(
candidates,
attempted_seeds,
chunk_count,
max_paths=max_paths,
max_local_boundary_speaker_drop=max_local_boundary_speaker_drop,
).results
def _sequence_fallback_candidate_result(
result: CandidateGateResult,
*,
max_local_boundary_speaker_drop: float | None,
) -> CandidateGateResult:
"""Return a DP-safe view of one local result or an unchanged hard reject."""
if result.passed is True and math.isfinite(result.score) and result.score >= 0.0:
return result
limit = (
None
if max_local_boundary_speaker_drop is None
else _finite_float(
max_local_boundary_speaker_drop,
minimum=0.0,
maximum=1.0,
)
)
if (
limit is None
or result.rejection_reasons != ("boundary_speaker_drop",)
or result.speaker_gate_applied is not True
or result.comparison.passed is not True
):
return result
similarity = _finite_float(
result.speaker_similarity,
minimum=-1.0,
maximum=1.0,
)
boundary_drop = _finite_float(result.boundary_speaker_drop, minimum=0.0)
cer = _finite_float(result.comparison.cer, minimum=0.0)
if (
similarity is None
or boundary_drop is None
or boundary_drop > limit + 1.0e-12
or cer is None
):
return result
score = (
cer
+ SEQUENCE_FALLBACK_SPEAKER_WEIGHT * (1.0 - similarity)
+ SEQUENCE_FALLBACK_BOUNDARY_WEIGHT * boundary_drop
)
if not math.isfinite(score) or score < 0.0:
return result
return replace(
result,
passed=True,
score=score,
rejection_reasons=(),
)
def local_candidate_has_coverage_eligibility(
result: CandidateGateResult,
*,
max_local_boundary_speaker_drop: float | None,
) -> bool:
"""Return whether a primary local result could enter either safe pool."""
if not isinstance(result, CandidateGateResult):
return False
eligible = _sequence_fallback_candidate_result(
result,
max_local_boundary_speaker_drop=max_local_boundary_speaker_drop,
)
return bool(
eligible.passed is True
and math.isfinite(eligible.score)
and eligible.score >= 0.0
and not eligible.rejection_reasons
)
def _preferred_speaker_verification(
verification: TrajectoryGateResult,
*,
min_similarity: float,
max_boundary_drop: float,
) -> bool:
if verification.passed is not True or not math.isfinite(verification.score):
return False
for result in verification.candidate_results:
if not result.speaker_gate_applied:
continue
similarity = _finite_float(result.speaker_similarity, minimum=-1.0, maximum=1.0)
boundary_drop = _finite_float(result.boundary_speaker_drop, minimum=0.0)
if (
similarity is None
or boundary_drop is None
or similarity < min_similarity
or boundary_drop > max_boundary_drop
):
return False
return True
def _preferred_squim_verification(
verification: TrajectoryGateResult,
*,
min_stoi: float,
min_pesq: float,
min_audio_duration_seconds: float = 0.0,
) -> bool:
"""Require long-enough hard-gated chunks to meet the preferred tier."""
preferred_stoi = _finite_float(min_stoi, minimum=0.0, maximum=1.0)
preferred_pesq = _finite_float(min_pesq, minimum=0.0, maximum=5.0)
duration_floor = _finite_float(min_audio_duration_seconds, minimum=0.0)
if (
preferred_stoi is None
or preferred_pesq is None
or duration_floor is None
or verification.passed is not True
or not math.isfinite(verification.score)
):
return False
for result in verification.candidate_results:
duration = _finite_float(result.audio_duration_seconds, minimum=0.0)
if duration is None:
return False
if duration < duration_floor:
continue
if result.squim_gate_applied is not True:
return False
stoi = _finite_float(result.squim_stoi, minimum=0.0, maximum=1.0)
pesq = _finite_float(result.squim_pesq, minimum=0.0, maximum=5.0)
if (
stoi is None
or pesq is None
or stoi < preferred_stoi
or pesq < preferred_pesq
):
return False
return True
def _preferred_release_verification(
verification: TrajectoryGateResult,
*,
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 = 0.0,
) -> bool:
"""Combine independently optional preferred speaker and quality tiers."""
if min_speaker_similarity is not None:
assert max_boundary_speaker_drop is not None
if not _preferred_speaker_verification(
verification,
min_similarity=min_speaker_similarity,
max_boundary_drop=max_boundary_speaker_drop,
):
return False
if min_squim_stoi is not None:
assert min_squim_pesq is not None
if not _preferred_squim_verification(
verification,
min_stoi=min_squim_stoi,
min_pesq=min_squim_pesq,
min_audio_duration_seconds=min_squim_audio_duration_seconds,
):
return False
return verification.passed is True and math.isfinite(verification.score)
def _preferred_release_evidence(
evidence: TrajectoryGateEvidence | None,
*,
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 = 0.0,
) -> bool:
"""Apply the preferred tier to one exact joined-waveform snapshot."""
if (
not isinstance(evidence, TrajectoryGateEvidence)
or evidence.passed is not True
or evidence.result_count != 1
or evidence.result is None
or evidence.result.passed is not True
or _finite_float(evidence.score, minimum=0.0) is None
):
return False
result = evidence.result
if min_speaker_similarity is not None:
assert max_boundary_speaker_drop is not None
if result.speaker_gate_applied:
similarity = _finite_float(
result.speaker_similarity,
minimum=-1.0,
maximum=1.0,
)
boundary_drop = _finite_float(
result.boundary_speaker_drop,
minimum=0.0,
)
if (
similarity is None
or boundary_drop is None
or similarity < min_speaker_similarity
or boundary_drop > max_boundary_speaker_drop
):
return False
if min_squim_stoi is not None:
assert min_squim_pesq is not None
duration = _finite_float(
result.audio_duration_seconds,
minimum=0.0,
)
duration_floor = _finite_float(
min_squim_audio_duration_seconds,
minimum=0.0,
)
if duration is None or duration_floor is None:
return False
if duration >= duration_floor:
stoi = _finite_float(
result.squim_stoi,
minimum=0.0,
maximum=1.0,
)
pesq = _finite_float(
result.squim_pesq,
minimum=0.0,
maximum=5.0,
)
if (
result.squim_gate_applied is not True
or stoi is None
or pesq is None
or stoi < min_squim_stoi
or pesq < min_squim_pesq
):
return False
return True
def _single_exact_fast_release_evidence(
release_evidence: TrajectoryGateEvidence | None,
independent_evidence: TrajectoryGateEvidence | None,
*,
min_speaker_similarity: float,
max_boundary_speaker_drop: float,
min_squim_stoi: float,
min_squim_pesq: float,
min_audio_duration_seconds: float,
) -> bool:
"""Accept only a fully measured exact waveform at the fast quality tier.
Unlike the general preferred helper, this latency shortcut never treats a
skipped short-audio speaker or SQUIM gate as success. It is deliberately
limited to exact joined-waveform evidence; local chunk proxies cannot
trigger it.
"""
if (
not isinstance(release_evidence, TrajectoryGateEvidence)
or release_evidence.result is None
or release_evidence.result.speaker_gate_applied is not True
or release_evidence.result.squim_gate_applied is not True
or not _preferred_release_evidence(
independent_evidence,
min_speaker_similarity=None,
max_boundary_speaker_drop=None,
min_squim_stoi=None,
min_squim_pesq=None,
)
):
return False
duration = _finite_float(
release_evidence.result.audio_duration_seconds,
minimum=0.0,
)
if duration is None or duration < min_audio_duration_seconds:
return False
return _preferred_release_evidence(
release_evidence,
min_speaker_similarity=min_speaker_similarity,
max_boundary_speaker_drop=max_boundary_speaker_drop,
min_squim_stoi=min_squim_stoi,
min_squim_pesq=min_squim_pesq,
min_squim_audio_duration_seconds=min_audio_duration_seconds,
)
def _coverage_trajectory_tuple(trajectory: Any, expected_chunks: int) -> tuple[Any, ...]:
"""Return one generator result as an exact, non-string trajectory."""
if isinstance(trajectory, (str, bytes, bytearray, np.ndarray)):
raise RuntimeError("candidate generator returned a malformed trajectory")
try:
normalized = tuple(trajectory)
except TypeError as error:
raise RuntimeError("candidate generator returned a malformed trajectory") from error
if len(normalized) != expected_chunks:
raise RuntimeError("candidate generator returned a misaligned trajectory")
return normalized
def _coverage_score_is_valid(value: Any, *, finite: bool) -> bool:
"""Return whether a gate score is non-negative and non-NaN."""
if isinstance(value, (bool, np.bool_)):
return False
try:
score = float(value)
except (TypeError, ValueError, OverflowError):
return False
if math.isnan(score) or score < 0.0:
return False
return math.isfinite(score) if finite else score != -math.inf
def _validate_coverage_evidence(
verification: Any,
expected_chunks: int,
*,
verifier_name: str,
) -> TrajectoryGateResult:
"""Validate structural evidence before retaining any local chunk."""
if not isinstance(verification, TrajectoryGateResult):
raise RuntimeError(f"{verifier_name} returned an invalid result")
if not isinstance(verification.passed, (bool, np.bool_)):
raise RuntimeError(f"{verifier_name} returned an invalid pass flag")
if not all(
isinstance(value, tuple)
for value in (
verification.candidate_results,
verification.chunk_artifacts,
verification.rejection_reasons,
)
):
raise RuntimeError(f"{verifier_name} returned malformed local evidence")
try:
results = tuple(verification.candidate_results)
artifacts = tuple(verification.chunk_artifacts)
reasons = tuple(verification.rejection_reasons)
except TypeError as error:
raise RuntimeError(
f"{verifier_name} returned malformed local evidence"
) from error
if (
len(results) != expected_chunks
or len(artifacts) != expected_chunks
or any(
not isinstance(result, CandidateGateResult)
or not isinstance(result.comparison, AsrComparison)
or not isinstance(result.passed, (bool, np.bool_))
or not isinstance(result.speaker_gate_applied, (bool, np.bool_))
or not isinstance(result.comparison.passed, (bool, np.bool_))
or not _coverage_score_is_valid(result.score, finite=result.passed is True)
or not isinstance(result.rejection_reasons, tuple)
or any(
not isinstance(reason, str)
for reason in result.rejection_reasons
)
for result in results
)
or any(
not isinstance(artifact, ChunkCandidateArtifact)
for artifact in artifacts
)
or any(not isinstance(reason, str) for reason in reasons)
or not _coverage_score_is_valid(
verification.score,
finite=verification.passed is True,
)
):
raise RuntimeError(f"{verifier_name} returned malformed local evidence")
if verification.passed is True and (
reasons
or any(
result.passed is not True
or result.rejection_reasons
or result.comparison.passed is not True
for result in results
)
):
raise RuntimeError(f"{verifier_name} returned inconsistent passing evidence")
if verification.passed is True and any(
not _coverage_artifact_is_valid(result, artifact)
for result, artifact in zip(
results,
artifacts,
strict=True,
)
):
raise RuntimeError(f"{verifier_name} passed without valid acoustic 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,
) -> bool:
"""Require finite transition evidence and ECAPA evidence when gated."""
if result.comparison.passed is not True:
return False
if result.squim_gate_applied and any(
value is None
for value in (
_finite_float(result.squim_stoi, minimum=0.0, maximum=1.0),
_finite_float(result.squim_pesq, minimum=0.0, maximum=5.0),
_finite_float(result.squim_si_sdr),
_finite_float(result.squim_quality_cost, minimum=0.0),
)
):
return False
if _finite_float(artifact.rms_db) is None:
return False
if artifact.median_f0_hz is not None and _finite_float(
artifact.median_f0_hz,
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
if embedding is not None:
try:
if cosine_similarity(embedding, embedding) < 1.0 - 1.0e-6:
return False
except ValueError:
return False
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.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,
seed: int,
audio: Any,
result: CandidateGateResult,
artifact: ChunkCandidateArtifact,
max_local_boundary_speaker_drop: float | None,
) -> _RaggedChunkCandidate | None:
"""Retain only a strict pass or the frozen boundary-only DP exception."""
adjusted = _sequence_fallback_candidate_result(
result,
max_local_boundary_speaker_drop=max_local_boundary_speaker_drop,
)
if (
adjusted.passed is not True
or not math.isfinite(adjusted.score)
or adjusted.score < 0.0
or adjusted.rejection_reasons
):
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,
audio=audio,
result=adjusted,
artifact=artifact,
)
def _coverage_final_passed(verification: Any) -> bool:
"""Accept only one internally consistent exact whole-waveform result."""
if not isinstance(verification, TrajectoryGateResult):
raise RuntimeError("sequence final verifier returned an invalid result")
if not all(
isinstance(value, tuple)
for value in (
verification.candidate_results,
verification.rejection_reasons,
)
):
raise RuntimeError("sequence final verifier returned malformed evidence")
if (
len(verification.candidate_results) != 1
or not isinstance(verification.candidate_results[0], CandidateGateResult)
or not isinstance(verification.candidate_results[0].comparison, AsrComparison)
or not isinstance(verification.passed, (bool, np.bool_))
or any(not isinstance(reason, str) for reason in verification.rejection_reasons)
):
raise RuntimeError("sequence final verifier returned malformed evidence")
result = verification.candidate_results[0]
if (
not isinstance(result.passed, (bool, np.bool_))
or not isinstance(result.comparison.passed, (bool, np.bool_))
or not isinstance(result.rejection_reasons, tuple)
or any(not isinstance(reason, str) for reason in result.rejection_reasons)
or not _coverage_score_is_valid(
verification.score,
finite=verification.passed is True,
)
or not _coverage_score_is_valid(result.score, finite=result.passed is True)
):
raise RuntimeError("sequence final verifier returned malformed evidence")
return bool(
verification.passed is True
and not verification.rejection_reasons
and result.passed is True
and not result.rejection_reasons
and result.comparison.passed is True
)
def run_coverage_adaptive_cascade(
chunks: Sequence[str],
root_seed: int,
candidate_generator: Callable[..., Any],
whole_candidate_verifier: (
Callable[
[Any, tuple[str, ...], int],
TrajectoryGateResult | CandidateVerification,
]
),
refill_candidate_verifier: (
Callable[
[Any, tuple[str, ...], int],
TrajectoryGateResult | CandidateVerification,
]
),
*,
sequence_final_verifier: Callable[
[CascadeResult, tuple[str, ...]],
TrajectoryGateResult,
],
generation_evidence_factory: Callable[..., CandidateGenerationEvidence] | None = None,
candidate_generation_text_transform: (
Callable[
[tuple[str, ...], CandidateGenerationContext],
Sequence[str],
]
| None
) = None,
transition_artifact_enricher: (
Callable[[Any, ChunkCandidateArtifact], ChunkCandidateArtifact] | None
) = None,
max_generated_chunks: int = 32,
max_generated_text_units: int = 800,
max_sequence_paths: int = 3,
sequence_fallback_max_local_boundary_speaker_drop: float | None = None,
preferred_min_speaker_similarity: float | None = None,
preferred_max_boundary_speaker_drop: float | None = None,
preferred_min_squim_stoi: float | None = None,
preferred_min_squim_pesq: float | None = None,
preferred_min_squim_audio_duration_seconds: float = 0.0,
single_exact_min_speaker_similarity: float | None = None,
single_exact_max_boundary_speaker_drop: float | None = None,
require_endpoint_evidence: bool = False,
) -> CascadeResult:
"""Run one whole trajectory, then deterministic low-coverage refills.
The initial same-seed trajectory is the only full-trajectory generation.
If its exact whole-output checks fail, every valid local observation is
retained. Later seeds generate exactly one low-coverage chunk under hard
generated-chunk and generated-text-unit budgets. Ragged DP paths are
never returned without the supplied exact whole-waveform verifier.
Existing two-argument generation callbacks remain unchanged. A callback
that explicitly accepts the keyword-only ``generation_context`` opts into
per-chunk refill scheduling. Its evidence factory must accept the same
keyword and report the supplied row-local candidate ordinals.
``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.
``transition_artifact_enricher`` may lazily attach expensive soft
transition evidence to retained candidates. It runs only when the ragged
lattice contains more than one selectable path. A one-path lattice has no
ranking decision, so enriching it cannot affect the selected waveform and
is deliberately skipped.
"""
if not all(
callable(callback)
for callback in (
candidate_generator,
whole_candidate_verifier,
refill_candidate_verifier,
sequence_final_verifier,
)
):
raise ValueError("coverage-adaptive callbacks must be callable")
if generation_evidence_factory is not None and not callable(
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
):
raise ValueError("candidate_generation_text_transform must be callable")
if (
transition_artifact_enricher is not None
and not callable(transition_artifact_enricher)
):
raise ValueError("transition artifact enricher must be callable")
def accepts_generation_context(callback: Callable[..., Any]) -> bool:
try:
parameters = inspect.signature(callback).parameters.values()
except (TypeError, ValueError):
return False
return any(
parameter.kind is inspect.Parameter.VAR_KEYWORD
or (
parameter.name == "generation_context"
and parameter.kind
in (
inspect.Parameter.POSITIONAL_OR_KEYWORD,
inspect.Parameter.KEYWORD_ONLY,
)
)
for parameter in parameters
)
context_aware_generator = accepts_generation_context(candidate_generator)
context_aware_evidence = (
generation_evidence_factory is not None
and accepts_generation_context(generation_evidence_factory)
)
if context_aware_generator != context_aware_evidence:
raise ValueError(
"context-aware generation and evidence callbacks must opt in together"
)
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:
raise ValueError("coverage-adaptive cascade requires text chunks") from error
if not chunk_tuple or any(not chunk for chunk in chunk_tuple):
raise ValueError("coverage-adaptive cascade requires non-empty text chunks")
if isinstance(root_seed, (bool, np.bool_)):
raise ValueError("root_seed must be an integer")
try:
base_seed = operator.index(root_seed)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("root_seed must be an integer") from error
def positive_budget(value: Any, name: str) -> int:
if isinstance(value, (bool, np.bool_)):
raise ValueError(f"{name} must be a positive integer")
try:
normalized = operator.index(value)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError(f"{name} must be a positive integer") from error
if normalized <= 0:
raise ValueError(f"{name} must be a positive integer")
return normalized
chunk_budget = positive_budget(max_generated_chunks, "max_generated_chunks")
text_budget = positive_budget(
max_generated_text_units,
"max_generated_text_units",
)
if chunk_budget > ADAPTIVE_CASCADE_STAGE_LIMITS[-1]:
raise ValueError("max_generated_chunks cannot exceed the frozen cap of 32")
if text_budget > 800:
raise ValueError(
"max_generated_text_units cannot exceed the frozen cap of 800"
)
path_limit = positive_budget(max_sequence_paths, "max_sequence_paths")
if path_limit > 3:
raise ValueError("max_sequence_paths must be between 1 and 3")
boundary_limit = None
if sequence_fallback_max_local_boundary_speaker_drop is not None:
boundary_limit = _finite_float(
sequence_fallback_max_local_boundary_speaker_drop,
minimum=0.0,
maximum=SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP,
)
if boundary_limit is None:
raise ValueError(
"sequence fallback boundary threshold must be finite and no greater than 0.15"
)
preferred_speaker_enabled = (
preferred_min_speaker_similarity is not None
or preferred_max_boundary_speaker_drop is not None
)
if preferred_speaker_enabled:
preferred_similarity = _finite_float(
preferred_min_speaker_similarity,
minimum=-1.0,
maximum=1.0,
)
preferred_boundary = _finite_float(
preferred_max_boundary_speaker_drop,
minimum=0.0,
maximum=1.0,
)
if preferred_similarity is None or preferred_boundary is None:
raise ValueError(
"preferred speaker thresholds must be supplied together and finite"
)
else:
preferred_similarity = None
preferred_boundary = None
preferred_squim_enabled = (
preferred_min_squim_stoi is not None
or preferred_min_squim_pesq is not None
)
if preferred_squim_enabled:
preferred_stoi = _finite_float(
preferred_min_squim_stoi,
minimum=0.0,
maximum=1.0,
)
preferred_pesq = _finite_float(
preferred_min_squim_pesq,
minimum=0.0,
maximum=5.0,
)
preferred_squim_duration = _finite_float(
preferred_min_squim_audio_duration_seconds,
minimum=0.0,
)
if (
preferred_stoi is None
or preferred_pesq is None
or preferred_squim_duration is None
):
raise ValueError(
"preferred SQUIM thresholds and duration must be finite"
)
else:
preferred_stoi = None
preferred_pesq = None
preferred_squim_duration = 0.0
single_exact_fast_enabled = (
single_exact_min_speaker_similarity is not None
or single_exact_max_boundary_speaker_drop is not None
)
if single_exact_fast_enabled:
single_exact_similarity = _finite_float(
single_exact_min_speaker_similarity,
minimum=-1.0,
maximum=1.0,
)
single_exact_boundary = _finite_float(
single_exact_max_boundary_speaker_drop,
minimum=0.0,
maximum=1.0,
)
if single_exact_similarity is None or single_exact_boundary is None:
raise ValueError(
"single exact speaker thresholds must be supplied together and finite"
)
if not preferred_squim_enabled:
raise ValueError(
"single exact fast tier requires preferred SQUIM thresholds"
)
else:
single_exact_similarity = None
single_exact_boundary = None
preferred_single_search = bool(
preferred_speaker_enabled or preferred_squim_enabled
)
chunk_units = tuple(count_speech_units(chunk) for chunk in chunk_tuple)
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
# used by multi-chunk DP; the published release boundary remains unchanged.
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")
attempted_seeds: list[int] = []
diagnostic_candidates: list[_VerifiedTrajectoryCandidate] = []
generated_chunks = chunk_count
generated_units = initial_units
def generation_evidence(
context: CandidateGenerationContext,
candidate_chunks: tuple[str, ...],
) -> CandidateGenerationEvidence | None:
if generation_evidence_factory is None:
return None
try:
if context_aware_evidence:
raw_evidence = generation_evidence_factory(
context.candidate_index,
context.seed,
context.chunk_indices,
candidate_chunks,
generation_context=context,
)
expected_ordinals = context.chunk_candidate_ordinals
else:
raw_evidence = generation_evidence_factory(
context.candidate_index,
context.seed,
context.chunk_indices,
candidate_chunks,
)
expected_ordinals = (context.candidate_index,) * len(
candidate_chunks
)
return _validated_candidate_generation_evidence(
raw_evidence,
chunk_indices=context.chunk_indices,
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
try:
if context_aware_generator:
raw_initial_trajectory = candidate_generator(
initial_generation_chunks,
base_seed,
generation_context=initial_context,
)
else:
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(
raw_initial_trajectory,
chunk_count,
)
initial_generation_evidence = generation_evidence(
initial_context,
initial_generation_chunks,
)
attempted_seeds.append(base_seed)
try:
raw_initial_verification = whole_candidate_verifier(
initial_trajectory,
chunk_tuple,
base_seed,
)
except Exception as error:
raise RuntimeError("whole trajectory verification failed") from error
try:
(
initial_verification,
initial_independent_local_results,
initial_joined_output,
initial_independent_output,
) = (
_unwrap_candidate_verification(raw_initial_verification)
)
except TypeError as error:
raise RuntimeError(
"whole candidate verifier returned an invalid result"
) from error
initial_verification = _validate_coverage_evidence(
initial_verification,
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,
seed=base_seed,
trajectory=initial_trajectory,
verification=initial_verification,
independent_local_results=initial_independent_local_results,
joined_output=initial_joined_output,
independent_output=initial_independent_output,
generation_evidence=initial_generation_evidence,
)
)
initial_preferred = (
_preferred_release_verification(
initial_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_release_evidence(
initial_joined_output,
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,
)
)
initial_fast_exact = bool(
chunk_count == 1
and single_exact_fast_enabled
and _single_exact_fast_release_evidence(
initial_joined_output,
initial_independent_output,
min_speaker_similarity=single_exact_similarity,
max_boundary_speaker_drop=single_exact_boundary,
min_squim_stoi=preferred_stoi,
min_squim_pesq=preferred_pesq,
min_audio_duration_seconds=preferred_squim_duration,
)
)
if initial_verification.passed is True and (
chunk_count > 1
or not preferred_single_search
or initial_preferred
or initial_fast_exact
):
return CascadeResult(
trajectory=initial_trajectory,
verification=initial_verification,
seed=base_seed,
candidate_index=0,
attempted_seeds=(base_seed,),
chunk_candidate_indices=(0,) * chunk_count,
chunk_seeds=(base_seed,) * chunk_count,
selection_mode="whole_trajectory",
diagnostics=_cascade_diagnostics(diagnostic_candidates),
generated_chunk_count=generated_chunks,
generated_text_units=generated_units,
chunk_candidate_counts=(1,) * chunk_count,
)
pools: list[list[_RaggedChunkCandidate]] = [
[] for _ in range(chunk_count)
]
initial_culprits: list[int] = []
initial_all_strict = True
for chunk_index, (audio, result, artifact) in enumerate(
zip(
initial_trajectory,
initial_verification.candidate_results,
initial_verification.chunk_artifacts,
strict=True,
)
):
strict_pass = bool(
result.passed is True
and math.isfinite(result.score)
and result.score >= 0.0
and not result.rejection_reasons
)
initial_all_strict = initial_all_strict and strict_pass
if not strict_pass:
initial_culprits.append(chunk_index)
retained = None
if not (
chunk_count == 1
and initial_joined_output is not None
and initial_verification.passed is not True
):
retained = _coverage_local_candidate(
candidate_index=0,
seed=base_seed,
audio=audio,
result=result,
artifact=artifact,
max_local_boundary_speaker_drop=dp_boundary_limit,
)
if retained is not None:
pools[chunk_index].append(retained)
refill_attempts = [0] * chunk_count
latest_local_results = list(initial_verification.candidate_results)
next_candidate_index = 1
sequence_path_ledger: list[SequencePathEvidence] = []
checked_sequence_path_identities: set[tuple[int, ...]] = set()
eager_hard_pass_result: CascadeResult | None = None
enriched_transition_candidates: set[tuple[int, int]] = set()
# Keep one exact-path check in reserve for the completed lattice. With the
# production K=3 contract this permits the original first complete-path
# probe plus one incremental preferred-tier rescue after a new candidate
# arrives, without increasing the final-verifier budget.
incremental_probe_limit = max(1, path_limit - 1)
def enrich_transition_artifacts_for_ranking() -> None:
"""Attach soft transition evidence once, immediately before ranking."""
if transition_artifact_enricher is None:
return
for chunk_index, pool in enumerate(pools):
for position, candidate in enumerate(pool):
identity = (chunk_index, candidate.candidate_index)
if identity in enriched_transition_candidates:
continue
try:
artifact = transition_artifact_enricher(
candidate.audio,
candidate.artifact,
)
except Exception as error:
raise RuntimeError(
"transition artifact enrichment failed"
) from error
if not isinstance(artifact, ChunkCandidateArtifact):
raise RuntimeError(
"transition artifact enricher returned invalid evidence"
)
raw_median_f0_hz = artifact.median_f0_hz
median_f0_hz: float | None = None
if raw_median_f0_hz is not None:
median_f0_hz = _finite_float(
raw_median_f0_hz,
minimum=1.0,
)
if median_f0_hz is None:
raise RuntimeError(
"transition artifact enricher returned invalid evidence"
)
# The callback may supply only the soft F0 statistic. Keep all
# semantic, speaker, RMS and endpoint provenance from the
# already validated retained row; it must never be able to
# rewrite evidence that was bound before ranking.
enriched_artifact = replace(
candidate.artifact,
median_f0_hz=median_f0_hz,
)
if not _coverage_artifact_is_valid(
candidate.result,
enriched_artifact,
):
raise RuntimeError(
"transition artifact enricher returned invalid evidence"
)
pool[position] = replace(
candidate,
artifact=enriched_artifact,
)
enriched_transition_candidates.add(identity)
def eager_rank_one_sequence_probe() -> CascadeResult | None:
"""Verify a bounded new rank-one path before filling every row to K=3.
One final-path slot remains reserved for the completed lattice. Each
incremental probe excludes paths already checked. A hard failure keeps
the remaining verifier budget for the completed lattice; only a hard
pass that misses the preferred tier may trigger one early alternative.
The probe returns only when both the unchanged hard final gate and
preferred tier pass.
"""
nonlocal eager_hard_pass_result
if (
len(sequence_path_ledger) >= incremental_probe_limit
or (
sequence_path_ledger
and eager_hard_pass_result is None
)
or chunk_count <= 1
or not preferred_single_search
or any(not pool for pool in pools)
):
return None
# With one candidate per row there is only one possible waveform.
# Soft transition evidence cannot change its rank and is intentionally
# deferred. Once any row has an alternative, enrich the full lattice
# before computing even the first transition score.
if any(len(pool) > 1 for pool in pools):
enrich_transition_artifacts_for_ranking()
local_scores = [
[candidate.result.score for candidate in pool]
for pool in pools
]
transitions = [
[
[
candidate_chunk_transition_score(
previous.result,
previous.artifact,
current.result,
current.artifact,
)
for current in pools[chunk_index]
]
for previous in pools[chunk_index - 1]
]
for chunk_index in range(1, chunk_count)
]
excluded_path_list: list[tuple[int, ...]] = []
if (
initial_verification.passed is not True
and initial_all_strict
and all(pool[0].candidate_index == 0 for pool in pools)
):
excluded_path_list.append((0,) * chunk_count)
for checked_identity in sorted(checked_sequence_path_identities):
checked_positions = tuple(
next(
position
for position, candidate in enumerate(pools[chunk_index])
if candidate.candidate_index == candidate_index
)
for chunk_index, candidate_index in enumerate(
checked_identity
)
)
if checked_positions not in excluded_path_list:
excluded_path_list.append(checked_positions)
selections = select_culprit_diverse_candidate_sequences(
local_scores,
transitions,
culprit_indices=tuple(initial_culprits),
excluded_paths=tuple(excluded_path_list),
max_paths=1,
)
# An all-strict initial assembly may be the only path and is excluded
# because its exact output was already rejected. Keep the probe armed
# until the first genuinely new rank-one path becomes available.
if not selections:
return None
selection = selections[0]
selected = tuple(
pools[chunk_index][position]
for chunk_index, position in enumerate(selection.candidate_indices)
)
local_verification = TrajectoryGateResult(
passed=True,
candidate_results=tuple(candidate.result for candidate in selected),
score=selection.total_score,
rejection_reasons=(),
chunk_artifacts=tuple(candidate.artifact for candidate in selected),
)
eligible_counts = tuple(len(pool) for pool in pools)
finite_transition_counts = tuple(
sum(
math.isfinite(score)
for row in matrix
for score in row
)
for matrix in transitions
)
sequence_result = CascadeResult(
trajectory=tuple(candidate.audio for candidate in selected),
verification=local_verification,
seed=None,
candidate_index=None,
attempted_seeds=tuple(attempted_seeds),
chunk_candidate_indices=tuple(
candidate.candidate_index for candidate in selected
),
chunk_seeds=tuple(candidate.seed for candidate in selected),
selection_mode="coverage_sequence_dp",
sequence_path_rank=len(sequence_path_ledger) + 1,
diagnostics=_cascade_diagnostics(
diagnostic_candidates,
SequenceSearchEvidence(
eligible_candidate_counts=eligible_counts,
finite_transition_counts=finite_transition_counts,
ranked_path_count=len(sequence_path_ledger) + 1,
),
),
generated_chunk_count=generated_chunks,
generated_text_units=generated_units,
chunk_candidate_counts=eligible_counts,
)
try:
final_verification = sequence_final_verifier(
sequence_result,
chunk_tuple,
)
except Exception as error:
raise RuntimeError(
"sequence final verification failed; refusing unverified audio"
) from error
final_passed = _coverage_final_passed(final_verification)
final_evidence = trajectory_gate_evidence(final_verification)
path_identity = sequence_result.chunk_candidate_indices
checked_sequence_path_identities.add(path_identity)
sequence_path_ledger.append(
SequencePathEvidence(
rank=len(sequence_path_ledger) + 1,
chunk_candidate_indices=path_identity,
chunk_seeds=sequence_result.chunk_seeds,
final_output=final_evidence,
)
)
preferred_passed = _preferred_release_verification(
final_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,
)
if final_passed and eager_hard_pass_result is None:
eager_hard_pass_result = sequence_result
if not (final_passed and preferred_passed):
return None
sequence_search = SequenceSearchEvidence(
eligible_candidate_counts=eligible_counts,
finite_transition_counts=finite_transition_counts,
ranked_path_count=len(sequence_path_ledger),
checked_paths=tuple(sequence_path_ledger),
)
return replace(
sequence_result,
sequence_paths_checked=len(sequence_path_ledger),
diagnostics=_cascade_diagnostics(
diagnostic_candidates,
sequence_search,
),
)
eager_result = eager_rank_one_sequence_probe()
if eager_result is not None:
return eager_result
while generated_chunks < chunk_budget:
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)
):
break
has_noninitial_coverage = any(
candidate.candidate_index > 0
for pool in pools
for candidate in pool
)
needs_first_alternative = initial_all_strict and not has_noninitial_coverage
structurally_eligible = [
index
for index, pool in enumerate(pools)
if (
(chunk_count == 1 and preferred_single_search)
or len(pool) < path_limit
or needs_first_alternative
)
]
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
all_rows_covered = all(pools)
def refill_priority(index: int) -> tuple[int, int, int, int, int]:
# Once every row has safe local coverage, spend any remaining
# bounded budget on request endpoints first. The exact final
# speaker gate compares the beginning and ending thirds, so an
# extra interior candidate cannot repair endpoint identity drift.
endpoint_rank = (
0
if all_rows_covered and index in {0, chunk_count - 1}
else 1
)
primary_exact_retry_rank = int(
bool(pools[index])
or latest_local_results[index].comparison.passed is not True
)
return (
endpoint_rank,
len(pools[index]),
primary_exact_retry_rank,
refill_attempts[index],
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],)
if generated_chunks + len(refill_generation_chunks) > chunk_budget:
break
try:
if context_aware_generator:
raw_refill_trajectory = candidate_generator(
refill_generation_chunks,
seed,
generation_context=refill_context,
)
else:
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_generation_chunks,
)
attempted_seeds.append(seed)
generated_chunks += 1
generated_units += refill_units
refill_attempts[chunk_index] += 1
try:
raw_refill_verification = refill_candidate_verifier(
refill_trajectory,
refill_chunks,
seed,
)
except Exception as error:
raise RuntimeError("chunk refill verification failed") from error
try:
(
refill_verification,
refill_independent_local_results,
refill_joined_output,
refill_independent_output,
) = (
_unwrap_candidate_verification(raw_refill_verification)
)
except TypeError as error:
raise RuntimeError(
"refill candidate verifier returned an invalid result"
) from error
refill_verification = _validate_coverage_evidence(
refill_verification,
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(
candidate_index=next_candidate_index,
seed=seed,
trajectory=refill_trajectory,
verification=refill_verification,
independent_local_results=refill_independent_local_results,
joined_output=refill_joined_output,
independent_output=refill_independent_output,
generation_evidence=refill_generation_evidence,
)
)
retained = None
if not (
chunk_count == 1
and refill_joined_output is not None
and refill_verification.passed is not True
):
retained = _coverage_local_candidate(
candidate_index=next_candidate_index,
seed=seed,
audio=refill_trajectory[0],
result=refill_verification.candidate_results[0],
artifact=refill_verification.chunk_artifacts[0],
max_local_boundary_speaker_drop=dp_boundary_limit,
)
if retained is not None:
pools[chunk_index].append(retained)
eager_result = eager_rank_one_sequence_probe()
if eager_result is not None:
return eager_result
refill_preferred = (
_preferred_release_verification(
refill_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_release_evidence(
refill_joined_output,
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,
)
)
refill_fast_exact = bool(
chunk_count == 1
and single_exact_fast_enabled
and _single_exact_fast_release_evidence(
refill_joined_output,
refill_independent_output,
min_speaker_similarity=single_exact_similarity,
max_boundary_speaker_drop=single_exact_boundary,
min_squim_stoi=preferred_stoi,
min_squim_pesq=preferred_pesq,
min_audio_duration_seconds=preferred_squim_duration,
)
)
if (
chunk_count == 1
and preferred_single_search
and refill_verification.passed is True
and (refill_preferred or refill_fast_exact)
):
preferred_candidate = diagnostic_candidates[-1]
preferred_artifact = preferred_candidate.verification.chunk_artifacts[0]
preferred_reached_hard_cap = (
_endpoint_artifact_reached_hard_cap(preferred_artifact)
)
allow_natural_endpoint_waiver = bool(
refill_preferred and preferred_reached_hard_cap
)
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_release_evidence(
candidate.joined_output,
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 (
single_exact_fast_enabled
and _single_exact_fast_release_evidence(
candidate.joined_output,
candidate.independent_output,
min_speaker_similarity=single_exact_similarity,
max_boundary_speaker_drop=single_exact_boundary,
min_squim_stoi=preferred_stoi,
min_squim_pesq=preferred_pesq,
min_audio_duration_seconds=(
preferred_squim_duration
),
)
)
or (
allow_natural_endpoint_waiver
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=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=(selected_candidate.candidate_index,),
chunk_seeds=(selected_candidate.seed,),
selection_mode="whole_trajectory",
diagnostics=_cascade_diagnostics(diagnostic_candidates),
generated_chunk_count=generated_chunks,
generated_text_units=generated_units,
chunk_candidate_counts=(len(pools[0]),),
)
next_candidate_index += 1
if any(not pool for pool in pools):
zero_coverage_transition_counts = tuple(
sum(
math.isfinite(
candidate_chunk_transition_score(
previous.result,
previous.artifact,
current.result,
current.artifact,
)
)
for previous in pools[chunk_index - 1]
for current in pools[chunk_index]
)
for chunk_index in range(1, chunk_count)
)
sequence_search = SequenceSearchEvidence(
eligible_candidate_counts=tuple(len(pool) for pool in pools),
finite_transition_counts=zero_coverage_transition_counts,
ranked_path_count=0,
)
raise NoQualifiedCandidateError(
"no exact-final TTS path: at least one chunk has zero safe coverage "
f"after {generated_chunks} generated chunks",
diagnostics=_cascade_diagnostics(
diagnostic_candidates,
sequence_search,
),
)
if chunk_count > 1 and any(len(pool) > 1 for pool in pools):
enrich_transition_artifacts_for_ranking()
local_scores = [
[candidate.result.score for candidate in pool]
for pool in pools
]
transitions: list[list[list[float]]] = []
for chunk_index in range(1, chunk_count):
transitions.append(
[
[
candidate_chunk_transition_score(
previous.result,
previous.artifact,
current.result,
current.artifact,
)
for current in pools[chunk_index]
]
for previous in pools[chunk_index - 1]
]
)
# If every initial local chunk was strict, the initial exact assembly was
# already rejected by Breeze25. Do not spend another final check
# on that identical waveform. Boundary-recovered initial paths are not
# excluded because the exact whole checks have not run for them yet.
excluded_path_list: list[tuple[int, ...]] = []
if (
initial_verification.passed is not True
and initial_all_strict
and all(pool[0].candidate_index == 0 for pool in pools)
):
excluded_path_list.append((0,) * chunk_count)
for checked_identity in sorted(checked_sequence_path_identities):
checked_positions = tuple(
next(
position
for position, candidate in enumerate(pools[chunk_index])
if candidate.candidate_index == candidate_index
)
for chunk_index, candidate_index in enumerate(checked_identity)
)
if checked_positions not in excluded_path_list:
excluded_path_list.append(checked_positions)
excluded_paths = tuple(excluded_path_list)
culprit_indices = tuple(initial_culprits)
remaining_path_budget = path_limit - len(sequence_path_ledger)
selections = (
select_culprit_diverse_candidate_sequences(
local_scores,
transitions,
culprit_indices=culprit_indices,
excluded_paths=excluded_paths,
max_paths=remaining_path_budget,
)
if remaining_path_budget > 0
else ()
)
eligible_counts = tuple(len(pool) for pool in pools)
finite_transition_counts = tuple(
sum(
math.isfinite(score)
for row in matrix
for score in row
)
for matrix in transitions
)
checked_paths = sequence_path_ledger
checked = len(checked_paths)
ranked_path_count = checked + len(selections)
for selection in selections:
rank = len(checked_paths) + 1
selected = tuple(
pools[chunk_index][position]
for chunk_index, position in enumerate(selection.candidate_indices)
)
local_verification = TrajectoryGateResult(
passed=True,
candidate_results=tuple(candidate.result for candidate in selected),
score=selection.total_score,
rejection_reasons=(),
chunk_artifacts=tuple(candidate.artifact for candidate in selected),
)
sequence_result = CascadeResult(
trajectory=tuple(candidate.audio for candidate in selected),
verification=local_verification,
seed=None,
candidate_index=None,
attempted_seeds=tuple(attempted_seeds),
chunk_candidate_indices=tuple(
candidate.candidate_index for candidate in selected
),
chunk_seeds=tuple(candidate.seed for candidate in selected),
selection_mode="coverage_sequence_dp",
sequence_path_rank=rank,
diagnostics=_cascade_diagnostics(
diagnostic_candidates,
SequenceSearchEvidence(
eligible_candidate_counts=eligible_counts,
finite_transition_counts=finite_transition_counts,
ranked_path_count=ranked_path_count,
checked_paths=tuple(checked_paths),
),
),
generated_chunk_count=generated_chunks,
generated_text_units=generated_units,
chunk_candidate_counts=tuple(len(pool) for pool in pools),
)
checked += 1
try:
final_verification = sequence_final_verifier(
sequence_result,
chunk_tuple,
)
except Exception as error:
raise RuntimeError(
"sequence final verification failed; refusing unverified audio"
) from error
final_passed = _coverage_final_passed(final_verification)
final_evidence = trajectory_gate_evidence(final_verification)
checked_sequence_path_identities.add(
sequence_result.chunk_candidate_indices
)
checked_paths.append(
SequencePathEvidence(
rank=rank,
chunk_candidate_indices=(
sequence_result.chunk_candidate_indices
),
chunk_seeds=sequence_result.chunk_seeds,
final_output=final_evidence,
)
)
sequence_search = SequenceSearchEvidence(
eligible_candidate_counts=eligible_counts,
finite_transition_counts=finite_transition_counts,
ranked_path_count=ranked_path_count,
checked_paths=tuple(checked_paths),
)
diagnostics = _cascade_diagnostics(
diagnostic_candidates,
sequence_search,
)
if final_passed:
return replace(
sequence_result,
sequence_paths_checked=checked,
diagnostics=diagnostics,
)
final_sequence_search = SequenceSearchEvidence(
eligible_candidate_counts=eligible_counts,
finite_transition_counts=finite_transition_counts,
ranked_path_count=ranked_path_count,
checked_paths=tuple(checked_paths),
)
if eager_hard_pass_result is not None:
return replace(
eager_hard_pass_result,
attempted_seeds=tuple(attempted_seeds),
sequence_paths_checked=checked,
diagnostics=_cascade_diagnostics(
diagnostic_candidates,
final_sequence_search,
),
generated_chunk_count=generated_chunks,
generated_text_units=generated_units,
chunk_candidate_counts=tuple(len(pool) for pool in pools),
)
raise NoQualifiedCandidateError(
"no exact-final TTS path after "
f"{generated_chunks} generated chunks and {checked} assembled paths",
diagnostics=_cascade_diagnostics(
diagnostic_candidates,
final_sequence_search,
),
)
def run_adaptive_cascade(
chunks: Sequence[str],
root_seed: int,
candidate_generator: Callable[[tuple[str, ...], int], Any],
candidate_verifier: Callable[
[Any, tuple[str, ...], int],
TrajectoryGateResult | CandidateVerification,
],
*,
initial_candidates: int = 1,
max_candidates: int = 5,
preferred_min_speaker_similarity: float = 0.25,
preferred_max_boundary_speaker_drop: float = 0.05,
sequence_final_verifier: (
Callable[[CascadeResult, tuple[str, ...]], TrajectoryGateResult] | None
) = None,
max_sequence_paths: int = 3,
sequence_fallback_max_local_boundary_speaker_drop: float | None = None,
) -> CascadeResult:
"""Run the frozen deterministic fail-closed cascade through 32 candidates.
The generator is called once per trajectory with ``root_seed + offset``.
It receives all chunks in one call, making the shared per-trajectory seed
contract explicit and preventing accidental per-chunk seed drift.
"""
try:
chunk_tuple = tuple(str(chunk) for chunk in chunks)
base_seed = int(root_seed)
first_stage = int(initial_candidates)
limit = int(max_candidates)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError("invalid adaptive cascade arguments") from error
if not chunk_tuple or any(not chunk for chunk in chunk_tuple):
raise ValueError("adaptive cascade requires non-empty text chunks")
if first_stage != 1:
raise ValueError("online adaptive cascade must start with exactly one candidate")
if limit < first_stage or limit > ADAPTIVE_CASCADE_STAGE_LIMITS[-1]:
raise ValueError("adaptive cascade supports between 1 and 32 candidates")
if isinstance(max_sequence_paths, (bool, np.bool_)):
raise ValueError("max_sequence_paths must be an integer between 1 and 3")
try:
sequence_path_limit = operator.index(max_sequence_paths)
except (TypeError, ValueError, OverflowError) as error:
raise ValueError(
"max_sequence_paths must be an integer between 1 and 3"
) from error
if not 1 <= sequence_path_limit <= 3:
raise ValueError("max_sequence_paths must be an integer between 1 and 3")
if sequence_final_verifier is not None and not callable(sequence_final_verifier):
raise ValueError("sequence_final_verifier must be callable")
sequence_fallback_boundary = None
if sequence_fallback_max_local_boundary_speaker_drop is not None:
sequence_fallback_boundary = _finite_float(
sequence_fallback_max_local_boundary_speaker_drop,
minimum=0.0,
maximum=1.0,
)
if sequence_fallback_boundary is None:
raise ValueError("sequence fallback boundary threshold must be finite")
if sequence_final_verifier is None:
raise ValueError(
"sequence fallback boundary relaxation requires a final verifier"
)
preferred_similarity = _finite_float(
preferred_min_speaker_similarity,
minimum=-1.0,
maximum=1.0,
)
preferred_boundary = _finite_float(
preferred_max_boundary_speaker_drop,
minimum=0.0,
)
if preferred_similarity is None or preferred_boundary is None:
raise ValueError("preferred speaker thresholds must be finite")
attempted_seeds: list[int] = []
candidates: list[_VerifiedTrajectoryCandidate] = []
sequence_search_evidence: SequenceSearchEvidence | None = None
first_seed = base_seed
first_trajectory = candidate_generator(chunk_tuple, first_seed)
(
first_verification,
first_independent_local_results,
first_joined_output,
first_independent_output,
) = _unwrap_candidate_verification(
candidate_verifier(first_trajectory, chunk_tuple, first_seed)
)
attempted_seeds.append(first_seed)
first_candidate = _VerifiedTrajectoryCandidate(
candidate_index=0,
seed=first_seed,
trajectory=first_trajectory,
verification=first_verification,
independent_local_results=first_independent_local_results,
joined_output=first_joined_output,
independent_output=first_independent_output,
)
candidates.append(first_candidate)
if (
first_verification.passed
and math.isfinite(first_verification.score)
and (
limit == 1
or _preferred_speaker_verification(
first_verification,
min_similarity=preferred_similarity,
max_boundary_drop=preferred_boundary,
)
)
):
return _whole_trajectory_result(
first_candidate,
attempted_seeds,
len(chunk_tuple),
diagnostics=_cascade_diagnostics(candidates),
)
stages = list(
dict.fromkeys(
min(stage_limit, limit)
for stage_limit in ADAPTIVE_CASCADE_STAGE_LIMITS[1:]
)
)
next_candidate = first_stage
for stage_size in stages:
for candidate_index in range(next_candidate, stage_size):
seed = base_seed + candidate_index
trajectory = candidate_generator(chunk_tuple, seed)
(
verification,
independent_local_results,
joined_output,
independent_output,
) = _unwrap_candidate_verification(
candidate_verifier(trajectory, chunk_tuple, seed)
)
attempted_seeds.append(seed)
candidates.append(
_VerifiedTrajectoryCandidate(
candidate_index=candidate_index,
seed=seed,
trajectory=trajectory,
verification=verification,
independent_local_results=independent_local_results,
joined_output=joined_output,
independent_output=independent_output,
)
)
qualified = [
candidate
for candidate in candidates
if candidate.verification.passed
and math.isfinite(candidate.verification.score)
]
final_stage = stage_size == limit
if qualified:
preferred_qualified = [
candidate
for candidate in qualified
if _preferred_speaker_verification(
candidate.verification,
min_similarity=preferred_similarity,
max_boundary_drop=preferred_boundary,
)
]
selectable = qualified if final_stage else preferred_qualified
if selectable:
selected_whole = min(
selectable,
key=lambda candidate: (
candidate.verification.score,
candidate.candidate_index,
),
)
return _whole_trajectory_result(
selected_whole,
attempted_seeds,
len(chunk_tuple),
diagnostics=_cascade_diagnostics(candidates),
)
next_candidate = stage_size
continue
# With a final-aware callback, defer DP until all whole-trajectory
# candidates in the request budget have been exhausted. This keeps
# whole trajectories globally preferred and bounds joined checks to
# at most ``max_sequence_paths`` once per request.
if sequence_final_verifier is not None and not final_stage:
next_candidate = stage_size
continue
sequence_search = _sequence_fallback_search(
candidates,
attempted_seeds,
len(chunk_tuple),
max_paths=(sequence_path_limit if sequence_final_verifier else 1),
max_local_boundary_speaker_drop=sequence_fallback_boundary,
)
sequence_results = sequence_search.results
sequence_search_evidence = sequence_search.evidence
if sequence_final_verifier is None:
sequence_result = sequence_results[0] if sequence_results else None
if sequence_result is not None and (
final_stage
or _preferred_speaker_verification(
sequence_result.verification,
min_similarity=preferred_similarity,
max_boundary_drop=preferred_boundary,
)
):
return replace(
sequence_result,
diagnostics=_cascade_diagnostics(
candidates,
sequence_search_evidence,
),
)
else:
checked_paths: list[SequencePathEvidence] = []
for checked_count, sequence_result in enumerate(sequence_results, 1):
try:
final_verification = sequence_final_verifier(
sequence_result,
chunk_tuple,
)
except Exception as error:
raise RuntimeError(
"sequence final verification failed; refusing unverified audio"
) from error
if not isinstance(final_verification, TrajectoryGateResult):
raise RuntimeError(
"sequence final verifier returned an invalid result"
)
checked_paths.append(
SequencePathEvidence(
rank=_evidence_int(sequence_result.sequence_path_rank),
chunk_candidate_indices=tuple(
_evidence_int(index)
for index in sequence_result.chunk_candidate_indices
),
chunk_seeds=tuple(
_evidence_int(
seed,
maximum=(
REQUEST_SEED_LIMIT
+ CASCADE_EVIDENCE_MAX_ATTEMPTS
),
)
for seed in sequence_result.chunk_seeds
),
final_output=trajectory_gate_evidence(
final_verification
),
)
)
sequence_search_evidence = replace(
sequence_search.evidence,
checked_paths=tuple(checked_paths),
)
if (
final_verification.passed is True
and math.isfinite(final_verification.score)
and len(final_verification.candidate_results) == 1
and isinstance(
final_verification.candidate_results[0],
CandidateGateResult,
)
and final_verification.candidate_results[0].passed is True
and math.isfinite(
final_verification.candidate_results[0].score
)
and not final_verification.rejection_reasons
):
return CascadeResult(
trajectory=sequence_result.trajectory,
verification=sequence_result.verification,
seed=sequence_result.seed,
candidate_index=sequence_result.candidate_index,
attempted_seeds=sequence_result.attempted_seeds,
chunk_candidate_indices=(
sequence_result.chunk_candidate_indices
),
chunk_seeds=sequence_result.chunk_seeds,
selection_mode=sequence_result.selection_mode,
sequence_path_rank=sequence_result.sequence_path_rank,
sequence_paths_checked=checked_count,
diagnostics=_cascade_diagnostics(
candidates,
sequence_search_evidence,
),
)
next_candidate = stage_size
raise NoQualifiedCandidateError(
f"no verified TTS trajectory after {len(attempted_seeds)} candidates",
diagnostics=_cascade_diagnostics(
candidates,
sequence_search_evidence,
),
)