From 13bc71861e2bede66a4900b89e4a35f8e6539513 Mon Sep 17 00:00:00 2001 From: jsboige Date: Fri, 30 Oct 2026 07:33:21 +0100 Subject: [PATCH] Feat(audio,#19739): garde d'attaque CV3 en ligne dans la boucle p5 -- re-roll cible sur perte de tete Co-Authored-By: Claude Haiku 5.5 --- .../GenAI/Audio/04-Applications/v4/p5_tts.py | 220 ++++++++--- .../v4/tests/test_p5_cv3_attack_guard.py | 346 ++++++++++++++++++ .../v4/tests/test_p5_cv3_narrator.py | 10 + 3 files changed, 531 insertions(+), 45 deletions(-) create mode 100644 MyIA.AI.Notebooks/GenAI/Audio/04-Applications/v4/tests/test_p5_cv3_attack_guard.py 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 3f8d365e531..3dbf01b61da 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. @@ -1347,6 +1353,78 @@ class NarratorCosyVoice3Unavailable(RuntimeError): # bit-identically and only a degenerate one is perturbed. _CV3_RETRY_SEED_STRIDE = 104_729 # prime, unrelated to seed + seg_index +# 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. _CV3_MODEL_CACHE: dict[str, object] = {} @@ -1607,54 +1685,106 @@ 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) - try: - gen = model.inference_zero_shot( - chunk, prompt_text, str(asset_wav), stream=False + 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", + seg.seg_index, i + 1, len(chunks), attempt + 1, + _CV3_RENDER_ATTEMPTS, last_dur, len(chunk), floor_s, + f", {type(last_exc).__name__}" if last_exc else "", ) - 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: - 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", - seg.seg_index, i + 1, len(chunks), attempt + 1, - _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 6aa574d5ef0..68ddd2dbe19 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 @@ -301,6 +301,16 @@ 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. + monkeypatch.setattr( + "v4.p5_tts._cv3_attack_verdict", + lambda *a, **k: { + "attack_dropped": False, "reason": "none", "n_missing_leading": 0, + }, + ) return fake_torch