diff --git a/MyIA.AI.Notebooks/GenAI/Audio/04-Applications/v4/p5_tts.py b/MyIA.AI.Notebooks/GenAI/Audio/04-Applications/v4/p5_tts.py index eb2fc9beaa3..4402d51d7b4 100644 --- a/MyIA.AI.Notebooks/GenAI/Audio/04-Applications/v4/p5_tts.py +++ b/MyIA.AI.Notebooks/GenAI/Audio/04-Applications/v4/p5_tts.py @@ -17,6 +17,8 @@ from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed import re +import contextlib +import tempfile logger = logging.getLogger(__name__) @@ -91,8 +93,12 @@ # text (tags included) is unchanged, so its hash -- the other cache key -- # cannot see the fix; only this sentinel re-renders the glued narrators # (measured: 16/20 sampled segments carried omissions traced to glue -# points like "route,[short pause] quand" -> "route,quand"). -_CV3_NARRATOR_REFERENCE_ID: str = "cosyvoice3-zeroshot-narrator-fr-v3" +# points like "route,[short pause] quand" -> "route,quand"). v4 = the +# attack guard (#19739): a chunk that passes the duration floor can still +# start mid-text (onset-drop), and the guard re-rolls it -- segments the +# pre-guard run shipped with a dropped attack must not stay cached as +# "generated". +_CV3_NARRATOR_REFERENCE_ID: str = "cosyvoice3-zeroshot-narrator-fr-v4" # Engine -> reference_id sentinel. The narrator cache is keyed on it # (#19692): an MP3 rendered by one engine is never served for another. @@ -1371,15 +1377,77 @@ class NarratorCosyVoice3Unavailable(RuntimeError): # bit-identically and only a degenerate one is perturbed. _CV3_RETRY_SEED_STRIDE = 104_729 # prime, unrelated to seed + seg_index -# Per-chunk attack guard (#19739, phase 2 of the #19692 narrator routing). -# The duration floor above is a GROSS backstop and can never see an onset-drop: -# a render that is a plausible length but simply started mid-text. Only the ASR -# attack guard (v4.attack_guard — the per-chunk extension of p7's omission -# organ) catches that class, so it is wired into the render loop as a second -# acceptance gate: a chunk that clears the floor is re-rolled when its ASR -# shows the attack missing. Off by default — it costs one Whisper call per -# chunk and needs the ASR server, so it is opt-in via the env flag. -_CV3_ATTACK_GUARD: bool = os.getenv("NARRATOR_CV3_ATTACK_GUARD", "0") == "1" +# Attack guard (#19739, phase 2): a chunk that passes the duration floor can +# still start mid-text -- the onset-drop measured in phase 1 (PR #19699) is +# seed-dependent and invisible to any duration check (the chars/s bands of +# clean and omitting renders overlap fully), so only the ASR can see it. The +# verdict organ is v4.attack_guard (delivered by #21459); this loop is where +# it plugs in: a floor-passing render is accepted only if the ASR hears the +# chunk's first words, otherwise the same spaced-seed re-roll that answers a +# degenerate render answers the dropped attack. V4_ATTACK_GUARD=off restores +# the pre-guard acceptance (one ASR call less) -- that is the A/B arm the +# measured comparison needs, not a production mode: a run with the guard off +# ships the onset-drop the issue exists to close. +_ATTACK_GUARD_ENV = "V4_ATTACK_GUARD" +_ATTACK_GUARD_ASR_BASE_URL_ENV = "V4_ASR_BASE_URL" +# Same endpoint and model as the omission organ p7 (localhost:8190, +# large-v3-turbo): the guard must measure exactly what p7 measures, so an +# "attack dropped" verdict and a p7 omission span cannot diverge. +_ATTACK_GUARD_ASR_BASE_URL = "http://localhost:8190" + + +def _attack_guard_enabled() -> bool: + """V4_ATTACK_GUARD reads on, off, 0/1 -- default ON (the correction is the + phase-2 deliverable, not an opt-in).""" + return os.getenv(_ATTACK_GUARD_ENV, "on").strip().lower() not in { + "0", "off", "no", "false", + } + + +def _resolve_attack_asr(): + """wav path -> transcript, on the organ's http client (asr_http). + + The callable is resolved per segment, not at import: tests inject their + own, and the run can point V4_ASR_BASE_URL at the Whisper server of the + machine that carries the services. + """ + from .attack_guard import asr_http + + base_url = os.getenv( + _ATTACK_GUARD_ASR_BASE_URL_ENV, _ATTACK_GUARD_ASR_BASE_URL + ) + + def _transcribe(wav_path) -> str: + return asr_http({"audio": str(wav_path)}, base_url=base_url) + + return _transcribe + + +def _cv3_attack_verdict(candidate, sr, chunk, chunk_index, chunk_seed, asr_fn, tmp_dir): + """Floor passed -- does the ASR hear the chunk's first words? (#19739) + + Writes the candidate to a scratch wav (the organ's clients read a file), + transcribes it with the injected asr_fn and delegates the verdict to + attack_guard.evaluate_chunk. Returns the organ's verdict dict; the caller + decides what a dropped attack costs. + """ + import torchaudio + + from .attack_guard import evaluate_chunk + + wav_path = Path(tmp_dir) / f"chunk_{chunk_index}_attack.wav" + torchaudio.save(str(wav_path), candidate, sr, format="wav") + transcript = asr_fn(wav_path) + return evaluate_chunk( + { + "chunk_index": chunk_index, + "text": chunk, + "seed": chunk_seed, + "engine": "cosyvoice3", + "audio": str(wav_path), + }, + transcript, + ) # CosyVoice3 AutoModel: ~3.4 GB VRAM, tens of seconds to load — one instance # per process, not per segment. @@ -1600,44 +1668,6 @@ def _cv3_min_expected_s(text: str) -> float: return max(_CV3_MIN_ABS_S, len(text) / _CV3_MIN_CHARS_PER_SEC) -def _cv3_chunk_asr(wav_path: str) -> str: - """Transcribe one rendered chunk via the p7 Whisper server. - - Organ-first: this is `attack_guard.asr_http` (the same - `localhost:8190/v1/audio/transcriptions` path p7 uses), wrapped so the - render loop passes a bare path. It is injected into - `_cv3_chunk_attack_dropped` so hermetic tests substitute a stub and the - loop never reaches for the network. - """ - from . import attack_guard - - return attack_guard.asr_http({"audio": wav_path}) - - -def _cv3_chunk_attack_dropped(wav, sr: int, chunk_text: str, asr_fn) -> bool: - """Inline per-chunk attack guard (#19739). - - Saves the candidate to a temp WAV, transcribes it with `asr_fn`, and - reports whether the attack is missing. The verdict is - `attack_guard.detect_missing_attack` — the positional head-loss detector - built on p7's `_missing_word_spans` — invoked organ-first, not - re-implemented here. `asr_fn` is injected so the loop stays testable - without a server or a GPU. - """ - import tempfile - import torchaudio - - from . import attack_guard - - with tempfile.TemporaryDirectory(prefix="cv3_attack_") as tmp: - wav_path = str(Path(tmp) / "chunk.wav") - torchaudio.save(wav_path, wav, sr) - transcript = asr_fn(wav_path) - return bool( - attack_guard.detect_missing_attack(chunk_text, transcript)["attack_dropped"] - ) - - def _synthesize_narrator_cosyvoice3( seg: AnnotatedSegment, fishaudio_text: str, @@ -1679,56 +1709,75 @@ def _synthesize_narrator_cosyvoice3( prompt_text = cv3_client.PROMPT_TEXT_ZH + cv3_client.ENDOFPROMPT sr = int(model.sample_rate) + # Attack guard (#19739): on by default -- see _attack_guard_enabled. The + # ASR callable is resolved once per segment; the scratch dir holds the + # candidate wavs the organ's clients transcribe (system temp, never the + # repo, and it goes away with the segment). + attack_asr = _resolve_attack_asr() if _attack_guard_enabled() else None + tmp_ctx = ( + tempfile.TemporaryDirectory(prefix="v4_attack_guard_") + if attack_asr is not None + else contextlib.nullcontext(None) + ) + t0 = time.time() wavs = [] attempts = 0 - for i, chunk in enumerate(chunks): - floor_s = _cv3_min_expected_s(chunk) - wav = None - last_dur = 0.0 - last_exc: Exception | None = None - for attempt in range(_CV3_RENDER_ATTEMPTS): - attempts += 1 - # attempt 0 keeps the pre-guard seed, so a healthy chunk is - # unchanged and only a degenerate one is perturbed. - torch.manual_seed(seed + i + attempt * _CV3_RETRY_SEED_STRIDE) - attack_dropped = False - try: - gen = model.inference_zero_shot( - chunk, prompt_text, str(asset_wav), stream=False - ) - parts = [j["tts_speech"] for j in gen] - candidate = torch.cat(parts, dim=-1) if len(parts) > 1 else parts[0] - except Exception as exc: # noqa: BLE001 - # The crash and the silent 0.04 s are the same event: too few - # speech tokens. Here the mel was short enough to break the f0 - # predictor's conv1d before anything was yielded, so it is the - # same re-roll that answers it. - last_exc, last_dur = exc, 0.0 - else: - last_dur = candidate.shape[-1] / sr - # A render that clears the duration floor is accepted only when - # the ASR attack guard also passes: the floor is a gross - # backstop that cannot see an onset-drop (a plausible-length - # render that started late), so the guard is the fidelity organ - # for that class (#19739). Off by default, so a plain run keeps - # the pre-guard behaviour and no Whisper dependency. - attack_dropped = ( - _CV3_ATTACK_GUARD - and last_dur >= floor_s - and _cv3_chunk_attack_dropped(candidate, sr, chunk, _cv3_chunk_asr) - ) - if last_dur >= floor_s and not attack_dropped: - wav = candidate - break - if attack_dropped: - logger.warning( - "[CV3] seg %s chunk %d/%d attempt %d/%d: onset-drop " - "(attack not spoken per ASR guard) - re-rolling seed", - seg.seg_index, i + 1, len(chunks), attempt + 1, - _CV3_RENDER_ATTEMPTS, - ) - else: + attack_checks = 0 + attack_rerolls = 0 + with tmp_ctx as attack_tmp: + for i, chunk in enumerate(chunks): + floor_s = _cv3_min_expected_s(chunk) + wav = None + last_dur = 0.0 + last_exc: Exception | None = None + last_attack: dict | None = None + for attempt in range(_CV3_RENDER_ATTEMPTS): + attempts += 1 + # attempt 0 keeps the pre-guard seed, so a healthy chunk is + # unchanged and only a degenerate one is perturbed. The same + # formula answers a dropped attack (attack_guard.spaced_seed). + chunk_seed = seed + i + attempt * _CV3_RETRY_SEED_STRIDE + torch.manual_seed(chunk_seed) + try: + gen = model.inference_zero_shot( + chunk, prompt_text, str(asset_wav), stream=False + ) + parts = [j["tts_speech"] for j in gen] + candidate = ( + torch.cat(parts, dim=-1) if len(parts) > 1 else parts[0] + ) + except Exception as exc: # noqa: BLE001 + # The crash and the silent 0.04 s are the same event: too few + # speech tokens. Here the mel was short enough to break the f0 + # predictor's conv1d before anything was yielded, so it is the + # same re-roll that answers it. + last_exc, last_dur = exc, 0.0 + else: + last_dur = candidate.shape[-1] / sr + if last_dur >= floor_s: + if attack_asr is None: + wav = candidate + break + attack_checks += 1 + verdict = _cv3_attack_verdict( + candidate, sr, chunk, i, chunk_seed, + attack_asr, attack_tmp, + ) + if verdict["attack_dropped"]: + attack_rerolls += 1 + last_attack = verdict + logger.warning( + "[CV3] seg %s chunk %d/%d attempt %d/%d: attack " + "dropped (%s, %d leading word(s) lost) - re-" + "rolling seed (#19739)", + seg.seg_index, i + 1, len(chunks), attempt + 1, + _CV3_RENDER_ATTEMPTS, verdict["reason"], + verdict["n_missing_leading"], + ) + continue + wav = candidate + break logger.warning( "[CV3] seg %s chunk %d/%d attempt %d/%d: degenerate render " "(%.2fs for %d chars, floor %.2fs%s) - re-rolling seed", @@ -1736,17 +1785,30 @@ def _synthesize_narrator_cosyvoice3( _CV3_RENDER_ATTEMPTS, last_dur, len(chunk), floor_s, f", {type(last_exc).__name__}" if last_exc else "", ) - if wav is None: - # Hard failure, never a silent short MP3: the run counts it in - # Failed, which is honest, where a 0.10 s file is not. - raise NarratorCosyVoice3Unavailable( - f"seg {seg.seg_index}: chunk {i + 1}/{len(chunks)} still " - f"degenerate after {_CV3_RENDER_ATTEMPTS} attempts (last " - f"{last_dur:.2f}s for {len(chunk)} chars, floor " - f"{floor_s:.2f}s): {chunk[:60]!r}" - ) from last_exc - wavs.append(wav) + if wav is None: + # Hard failure, never a silent short MP3: the run counts it in + # Failed, which is honest, where a 0.10 s file is not. + if last_attack is not None and last_dur >= floor_s: + raise NarratorCosyVoice3Unavailable( + f"seg {seg.seg_index}: chunk {i + 1}/{len(chunks)} " + f"attack still dropped after {_CV3_RENDER_ATTEMPTS} " + f"attempts (last verdict: {last_attack['reason']}, " + f"{last_attack['n_missing_leading']} leading word(s) " + f"lost): {chunk[:60]!r} (#19739)" + ) + raise NarratorCosyVoice3Unavailable( + f"seg {seg.seg_index}: chunk {i + 1}/{len(chunks)} still " + f"degenerate after {_CV3_RENDER_ATTEMPTS} attempts (last " + f"{last_dur:.2f}s for {len(chunk)} chars, floor " + f"{floor_s:.2f}s): {chunk[:60]!r}" + ) from last_exc + wavs.append(wav) elapsed = time.time() - t0 + if attack_asr is not None: + logger.info( + "[CV3] seg %s: attack guard checked %d render(s), re-rolled %d (#19739)", + seg.seg_index, attack_checks, attack_rerolls, + ) pause = torch.zeros(1, int(_CV3_PAUSE_S * sr)) parts_audio: list[torch.Tensor] = [] diff --git a/MyIA.AI.Notebooks/GenAI/Audio/04-Applications/v4/tests/test_p5_cv3_attack_guard.py b/MyIA.AI.Notebooks/GenAI/Audio/04-Applications/v4/tests/test_p5_cv3_attack_guard.py new file mode 100644 index 00000000000..25e773c9ae5 --- /dev/null +++ b/MyIA.AI.Notebooks/GenAI/Audio/04-Applications/v4/tests/test_p5_cv3_attack_guard.py @@ -0,0 +1,346 @@ +"""Attack guard wired in the CV3 render loop (#19739, phase 2) — offline tests. + +The organ (verdict + re-roll plan) was delivered by #21459 and has its own +suite (test_attack_guard.py); what these tests guard is the WIRING inside +`_synthesize_narrator_cosyvoice3`: a floor-passing render is accepted only +when the ASR hears the chunk's first words, a dropped attack consumes the +same spaced-seed re-roll as a degenerate render, and V4_ATTACK_GUARD=off +restores the pre-guard acceptance for the A/B arm. + +Run from the v4 directory: + python -m pytest tests/test_p5_cv3_attack_guard.py -v +""" +from __future__ import annotations + +import sys +import types +from pathlib import Path + +_V4 = Path(__file__).resolve().parent.parent +if str(_V4.parent) not in sys.path: + sys.path.insert(0, str(_V4.parent)) + + +# Long enough that several words can be lost from the head (the organ's +# threshold is 3 leading words) and that the chunker keeps it whole. +_CHUNK = ( + "Il marcha lentement sur le quai désert en regardant les bateaux " + "quitter le port une dernière fois." +) +# Same text minus its first four words: the operational onset-drop. +_CHUNK_SANS_ATTAQUE = ( + "sur le quai désert en regardant les bateaux quitter le port une " + "dernière fois." +) +# Two-sentence text over 280 chars: the chunker (greedy, <= 280) must split +# it in two, so the per-chunk seed indexing is exercised for real. +_S1 = ( + "La vieille dame traversa la place déserte au petit matin en tenant " + "son parapluie fermé comme une arme oubliée, et personne dans le " + "village ne se souvenait de l'avoir vue arriver un jour quelconque." +) +_S2 = ( + "Le boulanger lui fit un signe de tête mesuré derrière la vitre " + "embuée de sa boutique, puis reprit son travail sans un mot de plus " + "car il savait mieux que quiconque qu'il ne fallait jamais déranger." +) +_TWO_CHUNKS = f"{_S1} {_S2}" + + +def _build_seg(*, text: str = "x"): + from v4.schemas import AnnotatedSegment + + return AnnotatedSegment( + seg_index=1, + speaker="narrateur", + type="narration", + text=text, + annotated_text=text, + ) + + +class _FakeTensor: + def __init__(self, n: int): + self.n = n + self.shape = (1, n) + + +class _FakeTorch: + def __init__(self): + self.seeds: list[int] = [] + + def manual_seed(self, value: int): + self.seeds.append(value) + + def cat(self, tensors, dim=-1): + return _FakeTensor(sum(t.n for t in tensors)) + + def zeros(self, *shape): + return _FakeTensor(shape[-1]) + + +class _FakeModel: + """Yields a WAV of the scripted duration per call (all above the floor).""" + + sample_rate = 24000 + + def __init__(self, script): + self.script = list(script) + self.calls: list[str] = [] + + def inference_zero_shot(self, text, prompt_text, wav_path, stream=False): + dur = self.script[len(self.calls)] + self.calls.append(text) + yield {"tts_speech": _FakeTensor(int(dur * self.sample_rate))} + + +class _FakeASR: + """Returns the scripted transcript per call and records what it served.""" + + def __init__(self, transcripts): + self.script = list(transcripts) + self.calls: list[str] = [] + + def __call__(self, wav_path): + self.calls.append(str(wav_path)) + idx = min(len(self.calls), len(self.script)) - 1 + return self.script[idx] + + +def _install_fake_runtime(monkeypatch, model, asr): + """Fake torch/torchaudio/pydub + stub client + the injected ASR (#19739). + + Same stand-ins as test_p5_cv3_narrator (kept local: test modules are not + importable organs), with torchaudio.save accepting BOTH the scratch wav + path the attack verdict writes and the BytesIO the MP3 export uses. + """ + import sys + + fake_torch = _FakeTorch() + monkeypatch.setitem(sys.modules, "torch", fake_torch) + + torchaudio = types.ModuleType("torchaudio") + + def _save(dest, wav, sr, format=None): + if hasattr(dest, "write"): + dest.write(b"WAV") + + torchaudio.save = _save + monkeypatch.setitem(sys.modules, "torchaudio", torchaudio) + + pydub = types.ModuleType("pydub") + + class _FakeAudio: + @staticmethod + def from_file(buf, format=None): + return _FakeAudio() + + def export(self, buf, format=None, bitrate=None): + buf.write(b"MP3") + + pydub.AudioSegment = _FakeAudio + monkeypatch.setitem(sys.modules, "pydub", pydub) + + client = types.SimpleNamespace( + _bootstrap_paths=lambda: (None, None, "asset.wav"), + PROMPT_TEXT_ZH="prompt", + ENDOFPROMPT="<|endofprompt|>", + ) + monkeypatch.setattr("v4.p5_tts._get_cv3_client", lambda: client) + monkeypatch.setattr("v4.p5_tts._load_cv3_model", lambda: model) + monkeypatch.setattr("v4.p5_tts.audio_duration_mp3", lambda b: 1.5) + # The real resolver builds on attack_guard.asr_http (network); the wiring + # under test is the loop's use of the callable, so the callable is fake. + monkeypatch.setattr("v4.p5_tts._resolve_attack_asr", lambda: asr) + return fake_torch + + +def _synthesize(tmp_path, monkeypatch, model, asr, text=_CHUNK, guard=None): + from v4.p5_tts import _synthesize_narrator_cosyvoice3 + + if guard is not None: + monkeypatch.setenv("V4_ATTACK_GUARD", guard) + else: + monkeypatch.delenv("V4_ATTACK_GUARD", raising=False) + fake_torch = _install_fake_runtime(monkeypatch, model, asr) + out = _synthesize_narrator_cosyvoice3( + seg=_build_seg(text=text), + fishaudio_text=text, + mp3_path=tmp_path / "seg_0001_narrateur.mp3", + seed=42, + text_hash="h", + ) + return out, fake_torch + + +# Durations scripted above the floor of any chunk used here (~150 chars -> +# floor ~3.9 s at 38.9 chars/s), so the floor never fires in this suite +# except where a test says it does. +_DUR = 6.0 + + +def test_guard_accepts_clean_chunk_on_first_attempt(tmp_path, monkeypatch): + """Healthy chunk: floor passes, ASR hears the attack -> accepted at + attempt 0, original seed, exactly one ASR call. A clean chunk must be + untouched by the guard (the pre-registered protocol's identity arm).""" + model = _FakeModel([_DUR]) + asr = _FakeASR([_CHUNK]) + out, fake_torch = _synthesize(tmp_path, monkeypatch, model, asr) + + assert out.status == "generated" + assert out.attempts == 1 + assert fake_torch.seeds == [42], "attempt 0 must keep the pre-guard seed" + assert len(asr.calls) == 1 + assert len(model.calls) == 1 + + +def test_guard_re_rolls_dropped_attack_with_spaced_seed(tmp_path, monkeypatch): + """The onset-drop: floor passes but the transcript lost the first words. + The chunk is re-rolled at the spaced seed -- the same formula the + degenerate path uses -- and accepted once the attack is heard.""" + model = _FakeModel([_DUR, _DUR]) + asr = _FakeASR([_CHUNK_SANS_ATTAQUE, _CHUNK]) + out, fake_torch = _synthesize(tmp_path, monkeypatch, model, asr) + + assert out.status == "generated" + assert out.attempts == 2 + assert fake_torch.seeds == [42, 42 + 104_729] + assert len(asr.calls) == 2 + assert len(set(model.calls)) == 1, "a re-roll must not rewrite the text" + + +def test_guard_fails_loudly_when_attack_never_recovers(tmp_path, monkeypatch): + """An attack that never comes back is a hard failure naming the guard, + never a shipped MP3 whose first words are missing (#19739 is what the + run counts in Failed -- an unguarded accept would hide the drop).""" + from v4.p5_tts import _CV3_RENDER_ATTEMPTS, NarratorCosyVoice3Unavailable + + model = _FakeModel([_DUR] * _CV3_RENDER_ATTEMPTS) + asr = _FakeASR([_CHUNK_SANS_ATTAQUE]) + mp3 = tmp_path / "seg_0001_narrateur.mp3" + monkeypatch.delenv("V4_ATTACK_GUARD", raising=False) + fake_torch = _install_fake_runtime(monkeypatch, model, asr) + + from v4.p5_tts import _synthesize_narrator_cosyvoice3 + + try: + _synthesize_narrator_cosyvoice3( + seg=_build_seg(text=_CHUNK), fishaudio_text=_CHUNK, + mp3_path=mp3, seed=42, text_hash="h", + ) + except NarratorCosyVoice3Unavailable as exc: + assert "attack" in str(exc) + assert "#19739" in str(exc) + assert not mp3.exists(), "a failed render must not leave a partial MP3" + assert len(asr.calls) == _CV3_RENDER_ATTEMPTS + assert fake_torch.seeds[0] == 42 + assert fake_torch.seeds[-1] == 42 + (_CV3_RENDER_ATTEMPTS - 1) * 104_729 + return + raise AssertionError("an all-dropped-attack chunk must raise, not return") + + +def test_guard_off_restores_pre_guard_acceptance(tmp_path, monkeypatch): + """V4_ATTACK_GUARD=off is the A/B arm of the measured comparison: the + floor alone decides and the ASR is never called.""" + model = _FakeModel([_DUR]) + asr = _FakeASR([]) # any call would be recorded and fail the assert + out, fake_torch = _synthesize( + tmp_path, monkeypatch, model, asr, guard="off" + ) + + assert out.status == "generated" + assert out.attempts == 1 + assert asr.calls == [], "the guard being off must mean zero ASR calls" + assert fake_torch.seeds == [42] + + +def test_asr_runs_only_after_the_floor_passes(tmp_path, monkeypatch): + """A degenerate render is re-rolled by the floor BEFORE any ASR call -- + transcribing a 0.04 s file would waste the ASR and say nothing.""" + model = _FakeModel([0.04, _DUR]) + asr = _FakeASR([_CHUNK]) + out, _ = _synthesize(tmp_path, monkeypatch, model, asr) + + assert out.status == "generated" + assert out.attempts == 2 + assert len(asr.calls) == 1, "the degenerate render must not be transcribed" + + +def test_guard_is_per_chunk_in_a_multi_chunk_segment(tmp_path, monkeypatch): + """Two chunks; only the second drops its attack on its first attempt. + The re-roll seed must carry THAT chunk's index (seed + 1 + stride), not + the segment's -- this is the attack_guard.spaced_seed convention p5 + already uses for degenerate renders.""" + from v4.p5_tts import _chunk_narration + + chunks = _chunk_narration(_TWO_CHUNKS) + assert len(chunks) == 2, f"fixture must split in two, got {len(chunks)}" + + model = _FakeModel([_DUR, _DUR, _DUR]) + asr = _FakeASR([chunks[0], _headless(chunks[1]), chunks[1]]) + out, fake_torch = _synthesize( + tmp_path, monkeypatch, model, asr, text=_TWO_CHUNKS + ) + + assert out.status == "generated" + assert out.attempts == 3 + assert fake_torch.seeds == [42, 43, 43 + 104_729] + assert len(asr.calls) == 3 + + +def _headless(text: str) -> str: + """The text minus its first four words -- a buildable onset-drop.""" + words = text.split() + return " ".join(words[4:]) + + +def test_guard_default_is_on_and_off_values_are_recognized(monkeypatch): + """The correction ships ON by default; the off-spellings are the A/B + switch. Anything else (typo, empty) fails safe: the guard stays on.""" + from v4 import p5_tts + + monkeypatch.delenv("V4_ATTACK_GUARD", raising=False) + assert p5_tts._attack_guard_enabled() is True + for value in ("0", "off", "no", "false", "OFF", "No"): + monkeypatch.setenv("V4_ATTACK_GUARD", value) + assert p5_tts._attack_guard_enabled() is False, value + for value in ("1", "on", "", "tru"): # unknown spellings fail safe: ON + monkeypatch.setenv("V4_ATTACK_GUARD", value) + assert p5_tts._attack_guard_enabled() is True, value + + +def test_reference_id_bumped_for_the_attack_guard(): + """The batch cache sentinel must move: a segment the pre-guard run + shipped with a dropped attack is stale under the guarded acceptance, + and the sentinel is what forces its re-render (#19692 convention).""" + from v4.p5_tts import _CV3_NARRATOR_REFERENCE_ID + + assert _CV3_NARRATOR_REFERENCE_ID == "cosyvoice3-zeroshot-narrator-fr-v4" + + +def test_attack_asr_reuses_the_organ_http_client(monkeypatch): + """The resolver must build on attack_guard.asr_http -- same endpoint and + model as p7 -- and honour V4_ASR_BASE_URL, not carry a second client.""" + from v4 import p5_tts + import v4.attack_guard as ag + + captured = {} + + def _fake_asr_http(chunk, model="large-v3-turbo", base_url="http://localhost:8190", language="fr"): + captured["base_url"] = base_url + captured["audio"] = chunk["audio"] + return "transcrit" + + monkeypatch.setattr(ag, "asr_http", _fake_asr_http) + + monkeypatch.setenv("V4_ASR_BASE_URL", "http://elsewhere:9000") + asr_fn = p5_tts._resolve_attack_asr() + assert asr_fn(Path("x.wav")) == "transcrit" + assert captured["base_url"] == "http://elsewhere:9000" + assert captured["audio"].endswith("x.wav") + + # The default (no env) points at the p7 endpoint. + monkeypatch.delenv("V4_ASR_BASE_URL", raising=False) + asr_default = p5_tts._resolve_attack_asr() + asr_default(Path("y.wav")) + assert captured["base_url"] == "http://localhost:8190" diff --git a/MyIA.AI.Notebooks/GenAI/Audio/04-Applications/v4/tests/test_p5_cv3_narrator.py b/MyIA.AI.Notebooks/GenAI/Audio/04-Applications/v4/tests/test_p5_cv3_narrator.py index 4de31221d50..61e790f764a 100644 --- a/MyIA.AI.Notebooks/GenAI/Audio/04-Applications/v4/tests/test_p5_cv3_narrator.py +++ b/MyIA.AI.Notebooks/GenAI/Audio/04-Applications/v4/tests/test_p5_cv3_narrator.py @@ -267,9 +267,15 @@ def inference_zero_shot(self, text, prompt_text, wav_path, stream=False): yield {"tts_speech": _FakeTensor(int(dur * self.sample_rate))} -def _install_fake_runtime(monkeypatch, model): +def _install_fake_runtime(monkeypatch, model, guard_stub=True): """Wire the fake torch/torchaudio/pydub and the stub client, so the retry - loop is exercised hermetically -- no GPU, no CosyVoice install.""" + loop is exercised hermetically -- no GPU, no CosyVoice install. + + `guard_stub=True` (the default) stubs the attack verdict to a clean one, + so the tests that are NOT about the guard keep their focus. A test that + drives the guard passes `guard_stub=False` and lets the real verdict + path run against its own fake ASR. + """ import sys import types @@ -303,6 +309,22 @@ def export(self, buf, format=None, bitrate=None): monkeypatch.setattr("v4.p5_tts._get_cv3_client", lambda: client) monkeypatch.setattr("v4.p5_tts._load_cv3_model", lambda: model) monkeypatch.setattr("v4.p5_tts.audio_duration_mp3", lambda b: 1.5) + # The attack guard (#19739) is ON by default in the loop these tests + # drive; its verdict organ has its own suites (test_attack_guard.py, + # test_p5_cv3_attack_guard.py). Here a clean verdict is stubbed so the + # re-roll loop stays the only thing under test -- no ASR, no organ. + # + # The stub is a DEFAULT, not a wall: a test that drives the guard + # itself (the onset-drop re-roll) must be able to replace it, otherwise + # the stub would answer the very question the test asks. `guard_stub=False` + # is how such a test asks for the real verdict path. + if guard_stub: + monkeypatch.setattr( + "v4.p5_tts._cv3_attack_verdict", + lambda *a, **k: { + "attack_dropped": False, "reason": "none", "n_missing_leading": 0, + }, + ) return fake_torch @@ -375,7 +397,9 @@ def test_cv3_attack_guard_re_rolls_an_onset_drop(tmp_path, monkeypatch): from v4.p5_tts import _synthesize_narrator_cosyvoice3 model = _FakeModel([2.0, 2.0]) # both clear the duration floor - fake_torch = _install_fake_runtime(monkeypatch, model) + # guard_stub=False: this test drives the guard itself, so the real + # verdict path must run (a stubbed verdict would answer the question). + fake_torch = _install_fake_runtime(monkeypatch, model, guard_stub=False) # The ASR "hears" the attack dropped on the first render (its transcript # is short on the 3 leading words), complete on the re-roll. @@ -389,8 +413,8 @@ def fake_asr(path): asr_calls.append(path) return transcripts[min(len(asr_calls) - 1, len(transcripts) - 1)] - monkeypatch.setattr(p5_tts, "_cv3_chunk_asr", fake_asr) - monkeypatch.setattr(p5_tts, "_CV3_ATTACK_GUARD", True) + monkeypatch.setattr(p5_tts, "_resolve_attack_asr", lambda: fake_asr) + monkeypatch.setenv("V4_ATTACK_GUARD", "on") text = "Le vieux marin regardait la mer sans rien dire." seg = _build_seg(speaker="narrateur", text=text) @@ -412,13 +436,13 @@ def test_cv3_attack_guard_accepts_a_clean_render(tmp_path, monkeypatch): from v4.p5_tts import _synthesize_narrator_cosyvoice3 model = _FakeModel([2.0]) - _install_fake_runtime(monkeypatch, model) + _install_fake_runtime(monkeypatch, model, guard_stub=False) def fake_asr(path): return "Le vieux marin regardait la mer sans rien dire" # complete - monkeypatch.setattr(p5_tts, "_cv3_chunk_asr", fake_asr) - monkeypatch.setattr(p5_tts, "_CV3_ATTACK_GUARD", True) + monkeypatch.setattr(p5_tts, "_resolve_attack_asr", lambda: fake_asr) + monkeypatch.setenv("V4_ATTACK_GUARD", "on") text = "Le vieux marin regardait la mer sans rien dire." seg = _build_seg(speaker="narrateur", text=text) @@ -432,9 +456,14 @@ def fake_asr(path): def test_cv3_attack_guard_off_never_calls_asr(tmp_path, monkeypatch): - """Off by default: with the flag unset the render loop must not touch the - ASR at all -- the guard is opt-in, so a plain run has no Whisper - dependency (this is why the pre-existing tests need no ASR stub).""" + """`V4_ATTACK_GUARD=off` is the A/B arm of the measured comparison: the + render loop must not touch the ASR at all, so a run can reproduce the + pre-guard acceptance without a Whisper server beside it. + + The toggle is the one #23486 standardised for the Qwen narrator paths + (`ATTACK_GUARD_ENV = "V4_ATTACK_GUARD"`, default on): the CV3 path now + reads the same variable, so one switch governs the whole v4 pipeline. + """ from v4 import p5_tts from v4.p5_tts import _synthesize_narrator_cosyvoice3 @@ -444,8 +473,8 @@ def test_cv3_attack_guard_off_never_calls_asr(tmp_path, monkeypatch): def boom(path): raise AssertionError("the ASR must not be called when the guard is off") - monkeypatch.setattr(p5_tts, "_cv3_chunk_asr", boom) - # _CV3_ATTACK_GUARD left at its default (False) + monkeypatch.setattr(p5_tts, "_resolve_attack_asr", lambda: boom) + monkeypatch.setenv("V4_ATTACK_GUARD", "off") text = "Le vieux marin regardait la mer sans rien dire." seg = _build_seg(speaker="narrateur", text=text)