fix(audio): detect transcript-wide hallucinations

tests/test_api.py

@@ -638,6 +638,9 @@ def test_transcript_coverage_pagination_and_manual_reindex(tmp_path: Path, monke
             assert payload["items"][0]["start_ms"] == 10_000
             assert payload["detected_language"] == "zh"
             assert payload["quality_state"] == "ready"
+            assert payload["aggregate_risk"] is False
+            assert payload["repeated_phrases"] == []
+            assert payload["quality_score_semantics"] == "rule_check"
             assert payload["quality_flags"] == []
 
             speech = await client.get("/api/v1/speech/config", headers=headers)

tests/test_audio_index.py

@@ -120,6 +120,42 @@ def test_audio_indexer_writes_timed_fts_entries_and_reconciles(tmp_path: Path, m
     assert json.loads(job["payload_json"]) == {"video_id": "video"}
 
 
+def test_repeated_hallucinations_are_saved_but_not_searchable(tmp_path: Path, monkeypatch):
+    app = _app(tmp_path)
+    service = app.state.services
+    monkeypatch.setattr(service.media, "input_for", lambda _video: MediaInput("movie.mp4"))
+    monkeypatch.setattr(
+        service.media,
+        "probe",
+        lambda _media: {"raw": {"streams": [{"codec_type": "audio"}]}, "duration_ms": 30_000},
+    )
+    monkeypatch.setattr(
+        service.speech,
+        "transcribe",
+        lambda *_args, **_kwargs: [
+            {"text": "拜拜", "start_ms": index * 1000, "end_ms": index * 1000 + 800}
+            for index in range(8)
+        ],
+    )
+    job_id = service.jobs.enqueue("transcribe_audio", {"video_id": "video"}, dedupe_key="audio:video")
+    service.audio_indexer.index(job_id, "video", quality_repair=True)
+    with service.db.read() as conn:
+        entries = conn.execute(
+            "SELECT count(*) FROM text_entries WHERE video_id='video' AND kind='audio'"
+        ).fetchone()[0]
+        searchable = conn.execute(
+            "SELECT count(*) FROM text_fts WHERE entry_id IN "
+            "(SELECT id FROM text_entries WHERE video_id='video' AND kind='audio')"
+        ).fetchone()[0]
+        video = conn.execute(
+            "SELECT audio_quality_score,audio_quality_flags_json FROM videos WHERE id='video'"
+        ).fetchone()
+    assert entries == 8
+    assert searchable == 0
+    assert video["audio_quality_score"] < 0.7
+    assert "whole_ending_hallucination" in json.loads(video["audio_quality_flags_json"])
+
+
 def test_audio_indexer_retries_medium_memory_pressure(tmp_path: Path, monkeypatch):
     app = _app(tmp_path)
     service = app.state.services

tests/test_speech_quality.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from types import SimpleNamespace
 
 from imagefind.speech_quality import (
+    aggregate_transcript_quality,
     normalize_language,
     pcm16_voiced_regions,
     select_language,
@@ -10,6 +11,32 @@ from imagefind.speech_quality import (
 )
 
 
+def test_aggregate_quality_rejects_repeated_short_hallucinations() -> None:
+    segments = ([{"text": "好"}] * 11) + ([{"text": "拜拜"}] * 7) + ([{"text": "正常对话内容"}] * 4)
+    result = aggregate_transcript_quality(segments, base_score=1.0)
+    assert result.score < 0.7
+    assert "whole_repeated_phrase" in result.flags
+    assert "whole_short_segment_dominance" in result.flags
+    assert "whole_ending_hallucination" in result.flags
+    assert ("拜拜", 7) in result.repeated_phrases
+
+
+def test_sports_samples_patterns_cannot_remain_ready() -> None:
+    sports_2 = ([{"text": "好"}] * 11) + ([{"text": "拜拜"}] * 7) + ([{"text": "嗯"}] * 5)
+    sports_2 += [{"text": f"正常对话{i}"} for i in range(19)]
+    sports_3 = ([{"text": "拜拜"}] * 8) + ([{"text": "啊"}] * 7) + ([{"text": "嗯"}] * 7)
+    sports_3 += ([{"text": "谢谢大家收看"}] * 3) + [{"text": f"正常内容{i}"} for i in range(32)]
+    for segments in (sports_2, sports_3):
+        result = aggregate_transcript_quality(segments, base_score=1.0)
+        assert result.score < 0.7
+        assert "whole_repeated_phrase" in result.flags
+
+
+def test_traditional_ending_phrase_is_flagged() -> None:
+    result = aggregate_transcript_quality([{"text": "謝謝大家收看"}] * 3, base_score=1.0)
+    assert "whole_ending_hallucination" in result.flags
+
+
 def test_language_policy_treats_stream_metadata_as_weak_hint() -> None:
     assert select_language(["en", "zh"], policy="zh_priority") == "zh"
     assert select_language(["en", "zh"], policy="auto") == "en"