import pytest from audita.core.chunking import TokenBatch from audita.core.config import AuditaConfig from audita.core.errors import AuditaLLMError from audita.core.schemas import parse_glossary_yaml, parse_transcript_json from audita.framework.models import CorrectionProposal, ModuleRunSpec from audita.validators import ( GrammarOnlyValidator, IdenticalTextValidator, NonEmptySegmentValidator, OriginalTextPresentValidator, ) from audita.validators.base import ValidationContext from audita.validators.llm import MeaningReversalValidator, SpokenFormPlausibilityValidator, SpokenWordValidator from audita.validators.prompts import ( build_meaning_reversal_messages, build_spoken_form_plausibility_messages, build_spoken_word_messages, ) import audita.validators.llm as llm_module class _Module: def __init__(self, replacement_policy: str) -> None: self.replacement_policy = replacement_policy class FakeStructuredLLMClient: def __init__(self, responses): self._responses = list(responses) self.calls = [] def run_structured(self, *, stage_name, messages, response_model, config): self.calls.append( { "stage_name": stage_name, "messages": list(messages), "response_model": response_model, } ) if not self._responses: raise AuditaLLMError("FakeStructuredLLMClient received more calls than expected.") return response_model.model_validate(self._responses.pop(0)) def _glossary(): return parse_glossary_yaml( """ glossary: - name: "Jesters" category: faction summary: "The Jesters are a faction." - name: "Hrank" category: pc summary: "Hrank is a player character." - name: "Lyra" category: npc summary: "Lyra is an NPC." """ ) def _context( *, proposals, transcript, llm_client, tmp_path, replacement_policy="require_unique", ): return ValidationContext( proposals=proposals, transcript=transcript, glossary=_glossary(), config=AuditaConfig.from_sources(env={"OPENROUTER_API_KEY": "test-key"}), run_spec=ModuleRunSpec( instance_name="homophones", module_key="homophones", module=_Module(replacement_policy), ), run_dir=tmp_path, llm_client=llm_client, ) def test_spoken_form_plausibility_validator_approves_plausible_and_rejects_implausible_corrections(tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "There were gestures at the temple."}, {"id": 2, "speaker": "A", "start": 1.0, "end": 2.0, "text": "Lyra moved first."} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="homophones", module_key="homophones", id=1, original_text="gestures", corrected_text="Jesters", confidence=0.95, ), CorrectionProposal( proposal_index=1, module_instance="homophones", module_key="homophones", id=2, original_text="Lyra", corrected_text="Jesters", confidence=0.95, ), ] client = FakeStructuredLLMClient( [ { "validations": [ { "correction_index": 0, "approved": True, "confidence": 0.97, "reason": "Likely spoken-form correction in context.", }, { "correction_index": 1, "approved": False, "confidence": 0.99, "reason": "Not plausibly related by homophone or mistranscription.", }, ] } ] ) result = SpokenFormPlausibilityValidator("spoken_form_plausibility_review").validate( _context(proposals=proposals, transcript=transcript, llm_client=client, tmp_path=tmp_path) ) assert [(decision.proposal_index, decision.approved) for decision in result.decisions] == [ (0, True), (1, False), ] prompt_text = client.calls[0]["messages"][1]["content"] assert '"original_segment_text"' in prompt_text assert "There were Jesters at the temple." in prompt_text assert "Lyra moved first." in prompt_text def test_meaning_reversal_validator_rejects_reversal_and_approves_nonreversal(tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "ChatGPT still can't really do that with a dam, but Claude actually can."}, {"id": 2, "speaker": "A", "start": 1.0, "end": 2.0, "text": "The figure became visible in the doorway."} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="homophones", module_key="homophones", id=1, original_text="dam", corrected_text="damn", confidence=0.95, ), CorrectionProposal( proposal_index=1, module_instance="homophones", module_key="homophones", id=2, original_text="visible", corrected_text="invisible", confidence=0.95, ), ] client = FakeStructuredLLMClient( [ { "validations": [ { "correction_index": 0, "approved": True, "confidence": 0.94, "reason": "Does not reverse the segment meaning.", }, { "correction_index": 1, "approved": False, "confidence": 1.0, "reason": "Changes visible to invisible and reverses the segment meaning.", }, ] } ] ) result = MeaningReversalValidator("meaning_reversal_review").validate( _context(proposals=proposals, transcript=transcript, llm_client=client, tmp_path=tmp_path) ) assert [(decision.proposal_index, decision.approved) for decision in result.decisions] == [ (0, True), (1, False), ] prompt_text = client.calls[0]["messages"][1]["content"] assert "The figure became visible in the doorway." in prompt_text assert "The figure became invisible in the doorway." in prompt_text def test_spoken_word_validator_approves_cleanup_and_rejects_rewrite(tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "I, uh, I think we should go."}, {"id": 2, "speaker": "A", "start": 1.0, "end": 2.0, "text": "We should maybe proceed carefully."} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="spoken_word", module_key="spoken_word", id=1, original_text="I, uh, I think", corrected_text="I think", confidence=0.95, ), CorrectionProposal( proposal_index=1, module_instance="spoken_word", module_key="spoken_word", id=2, original_text="maybe proceed carefully", corrected_text="go now", confidence=0.95, ), ] client = FakeStructuredLLMClient( [ { "validations": [ { "correction_index": 0, "approved": True, "confidence": 0.97, "reason": "Reasonable dysfluency cleanup that preserves meaning.", }, { "correction_index": 1, "approved": False, "confidence": 0.99, "reason": "This changes the substance of the segment rather than cleaning a dysfluency.", }, ] } ] ) result = SpokenWordValidator("spoken_word_review").validate( _context(proposals=proposals, transcript=transcript, llm_client=client, tmp_path=tmp_path) ) assert [(decision.proposal_index, decision.approved) for decision in result.decisions] == [ (0, True), (1, False), ] prompt_text = client.calls[0]["messages"][1]["content"] assert "I think we should go." in prompt_text assert "go now" in prompt_text def test_spoken_word_validator_allows_punctuation_cleanup_tied_to_dysfluency(tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "Well ... I think we should go."} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="spoken_word", module_key="spoken_word", id=1, original_text="Well ... ", corrected_text="", confidence=0.95, ) ] client = FakeStructuredLLMClient( [ { "validations": [ { "correction_index": 0, "approved": True, "confidence": 0.95, "reason": "Removes a hesitation artifact without changing substantive meaning.", } ] } ] ) result = SpokenWordValidator("spoken_word_review").validate( _context(proposals=proposals, transcript=transcript, llm_client=client, tmp_path=tmp_path) ) assert [(decision.proposal_index, decision.approved) for decision in result.decisions] == [(0, True)] def test_grammar_only_validator_allows_formatting_only_changes(tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "hello there"}, {"id": 2, "speaker": "A", "start": 1.0, "end": 2.0, "text": "cant we go"}, {"id": 3, "speaker": "A", "start": 2.0, "end": 3.0, "text": "Hello,world"} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="grammar", module_key="grammar", id=1, original_text="hello there", corrected_text="Hello there.", confidence=0.95, ), CorrectionProposal( proposal_index=1, module_instance="grammar", module_key="grammar", id=2, original_text="cant", corrected_text="can't", confidence=0.95, ), CorrectionProposal( proposal_index=2, module_instance="grammar", module_key="grammar", id=3, original_text="Hello,world", corrected_text="Hello, world", confidence=0.95, ), ] result = GrammarOnlyValidator("grammar_only_guard").validate( _context(proposals=proposals, transcript=transcript, llm_client=None, tmp_path=tmp_path) ) assert [(decision.proposal_index, decision.approved) for decision in result.decisions] == [ (0, True), (1, True), (2, True), ] def test_grammar_only_validator_allows_indefinite_article_changes(tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "a intelligence saving throw"}, {"id": 2, "speaker": "A", "start": 1.0, "end": 2.0, "text": "an owl"} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="grammar", module_key="grammar", id=1, original_text="a intelligence saving throw", corrected_text="an intelligence saving throw", confidence=0.95, ), CorrectionProposal( proposal_index=1, module_instance="grammar", module_key="grammar", id=2, original_text="an owl", corrected_text="a owl", confidence=0.95, ), ] result = GrammarOnlyValidator("grammar_only_guard").validate( _context(proposals=proposals, transcript=transcript, llm_client=None, tmp_path=tmp_path) ) assert [(decision.proposal_index, decision.approved) for decision in result.decisions] == [ (0, True), (1, True), ] def test_grammar_only_validator_rejects_word_level_changes(tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "their plan"}, {"id": 2, "speaker": "A", "start": 1.0, "end": 2.0, "text": "dam"}, {"id": 3, "speaker": "A", "start": 2.0, "end": 3.0, "text": "uh"}, {"id": 4, "speaker": "A", "start": 3.0, "end": 4.0, "text": "I I agree"}, {"id": 5, "speaker": "A", "start": 4.0, "end": 5.0, "text": "Eric"} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="grammar", module_key="grammar", id=1, original_text="their", corrected_text="there", confidence=0.95, ), CorrectionProposal( proposal_index=1, module_instance="grammar", module_key="grammar", id=2, original_text="dam", corrected_text="damn", confidence=0.95, ), CorrectionProposal( proposal_index=2, module_instance="grammar", module_key="grammar", id=3, original_text="uh", corrected_text="", confidence=0.95, ), CorrectionProposal( proposal_index=3, module_instance="grammar", module_key="grammar", id=4, original_text="I I", corrected_text="I", confidence=0.95, ), CorrectionProposal( proposal_index=4, module_instance="grammar", module_key="grammar", id=5, original_text="Eric", corrected_text="Eric's", confidence=0.95, ), ] result = GrammarOnlyValidator("grammar_only_guard").validate( _context(proposals=proposals, transcript=transcript, llm_client=None, tmp_path=tmp_path) ) assert [decision.approved for decision in result.decisions] == [False, False, False, False, False] assert all( decision.reason == "correction is not limited to punctuation, capitalization, and spacing" for decision in result.decisions ) def test_non_empty_segment_validator_rejects_empty_and_whitespace_only_segments(tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "uh"}, {"id": 2, "speaker": "A", "start": 1.0, "end": 2.0, "text": "um"} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="spoken_word", module_key="spoken_word", id=1, original_text="uh", corrected_text="", confidence=0.95, ), CorrectionProposal( proposal_index=1, module_instance="spoken_word", module_key="spoken_word", id=2, original_text="um", corrected_text=" ", confidence=0.95, ), ] result = NonEmptySegmentValidator("non_empty_segment_guard").validate( _context(proposals=proposals, transcript=transcript, llm_client=None, tmp_path=tmp_path) ) assert [decision.approved for decision in result.decisions] == [False, False] assert all(decision.reason == "correction would leave the segment empty" for decision in result.decisions) def test_non_empty_segment_validator_allows_punctuation_only_and_unpreviewable_proposals(tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "uh"}, {"id": 2, "speaker": "A", "start": 1.0, "end": 2.0, "text": "There were gestures at the temple."} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="spoken_word", module_key="spoken_word", id=1, original_text="uh", corrected_text=".", confidence=0.95, ), CorrectionProposal( proposal_index=1, module_instance="spoken_word", module_key="spoken_word", id=2, original_text="rank", corrected_text="Hrank", confidence=0.95, ), ] result = NonEmptySegmentValidator("non_empty_segment_guard").validate( _context(proposals=proposals, transcript=transcript, llm_client=None, tmp_path=tmp_path) ) assert [(decision.proposal_index, decision.approved, decision.reason) for decision in result.decisions] == [ (0, True, None), (1, True, None), ] def test_identical_text_validator_rejects_exact_noops_only(tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "hello"}, {"id": 2, "speaker": "A", "start": 1.0, "end": 2.0, "text": "hello"} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="grammar", module_key="grammar", id=1, original_text="hello", corrected_text="hello", confidence=0.95, ), CorrectionProposal( proposal_index=1, module_instance="grammar", module_key="grammar", id=2, original_text="hello", corrected_text="Hello", confidence=0.95, ), ] result = IdenticalTextValidator("identical_text_guard").validate( _context(proposals=proposals, transcript=transcript, llm_client=None, tmp_path=tmp_path) ) assert [(decision.proposal_index, decision.approved, decision.reason) for decision in result.decisions] == [ (0, False, "proposal original_text and corrected_text are identical"), (1, True, None), ] def test_original_text_present_validator_rejects_missing_spans_and_allows_present_or_unknown_segments(tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "hello hello"}, {"id": 2, "speaker": "A", "start": 1.0, "end": 2.0, "text": "goodbye"} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="homophones", module_key="homophones", id=1, original_text="hello", corrected_text="hi", confidence=0.95, ), CorrectionProposal( proposal_index=1, module_instance="homophones", module_key="homophones", id=2, original_text="rank", corrected_text="Hrank", confidence=0.95, ), CorrectionProposal( proposal_index=2, module_instance="homophones", module_key="homophones", id=99, original_text="missing", corrected_text="present", confidence=0.95, ), ] result = OriginalTextPresentValidator("original_text_present_guard").validate( _context(proposals=proposals, transcript=transcript, llm_client=None, tmp_path=tmp_path) ) assert [(decision.proposal_index, decision.approved, decision.reason) for decision in result.decisions] == [ (0, True, None), (1, False, "proposal original_text does not match segment text"), (2, True, None), ] @pytest.mark.parametrize( ("validator", "payload", "message_fragment"), [ ( SpokenFormPlausibilityValidator("spoken_form_plausibility_review"), { "validations": [ { "correction_index": 0, "approved": True, "confidence": 0.9, "reason": "ok", }, { "correction_index": 0, "approved": True, "confidence": 0.9, "reason": "duplicate", }, ] }, "duplicate correction_index values", ), ( MeaningReversalValidator("meaning_reversal_review"), {"validations": []}, "omitted correction_index values", ), ( SpokenFormPlausibilityValidator("spoken_form_plausibility_review"), { "validations": [ { "correction_index": 99, "approved": True, "confidence": 0.9, "reason": "unknown", } ] }, "unknown correction_index", ), ( SpokenWordValidator("spoken_word_review"), { "validations": [ { "correction_index": 99, "approved": True, "confidence": 0.9, "reason": "unknown", } ] }, "unknown correction_index", ), ], ) def test_llm_validators_reject_bad_correction_indexes(tmp_path, validator, payload, message_fragment): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "There were gestures at the temple."} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="homophones", module_key="homophones", id=1, original_text="gestures", corrected_text="Jesters", confidence=0.95, ) ] client = FakeStructuredLLMClient([payload]) with pytest.raises(AuditaLLMError, match=message_fragment): validator.validate(_context(proposals=proposals, transcript=transcript, llm_client=client, tmp_path=tmp_path)) def test_llm_validator_rejects_unpreviewable_proposals_without_calling_llm(tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "There were gestures at the temple."} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="homophones", module_key="homophones", id=1, original_text="rank", corrected_text="Hrank", confidence=0.95, ) ] client = FakeStructuredLLMClient([]) result = SpokenFormPlausibilityValidator("spoken_form_plausibility_review").validate( _context(proposals=proposals, transcript=transcript, llm_client=client, tmp_path=tmp_path) ) assert result.decisions[0].approved is False assert result.decisions[0].reason == "proposal original_text does not match segment text" assert client.calls == [] def test_llm_validator_skips_empty_candidate_sets_without_calling_llm(tmp_path): client = FakeStructuredLLMClient([]) result = MeaningReversalValidator("meaning_reversal_review").validate( _context(proposals=[], transcript=[], llm_client=client, tmp_path=tmp_path) ) assert result.decisions == [] assert client.calls == [] def test_spoken_form_prompt_emphasizes_acoustic_plausibility(): messages = build_spoken_form_plausibility_messages( [ { "correction_index": 0, "id": 1, "original_segment_text": "There were gestures at the temple.", "corrected_segment_text": "There were Jesters at the temple.", "original_text": "gestures", "corrected_text": "Jesters", } ] ) combined = messages[0]["content"] + messages[1]["content"] assert "homophone" in combined assert "gestures" in combined assert "Lyra" in combined assert '"original_segment_text"' in messages[1]["content"] def test_meaning_reversal_prompt_emphasizes_antonyms_and_segment_context(): messages = build_meaning_reversal_messages( [ { "correction_index": 0, "id": 1, "original_segment_text": "The figure became visible in the doorway.", "corrected_segment_text": "The figure became invisible in the doorway.", "original_text": "visible", "corrected_text": "invisible", } ] ) combined = messages[0]["content"] + messages[1]["content"] assert "antonym" in combined assert "visible" in combined assert "up" in combined assert "original_segment_text" in messages[1]["content"] def test_spoken_word_prompt_emphasizes_dysfluency_cleanup(): messages = build_spoken_word_messages( [ { "correction_index": 0, "id": 1, "original_segment_text": "I, uh, I think we should go.", "corrected_segment_text": "I think we should go.", "original_text": "I, uh, I think", "corrected_text": "I think", } ] ) combined = messages[0]["content"] + messages[1]["content"] assert "dysfluencies" in combined assert "intentional emphasis" in combined assert "Stop! Stop! Stop!" in combined assert "punctuation" in combined assert "substantive meaning" in combined assert "original_segment_text" in messages[1]["content"] def test_llm_validators_use_shared_token_batching_helper(monkeypatch, tmp_path): transcript = parse_transcript_json( """ [ {"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "There were gestures at the temple."}, {"id": 2, "speaker": "A", "start": 1.0, "end": 2.0, "text": "We saw rank near the gate."}, {"id": 3, "speaker": "A", "start": 2.0, "end": 3.0, "text": "ChatGPT still can't do that with a dam."} ] """ ) proposals = [ CorrectionProposal( proposal_index=0, module_instance="homophones", module_key="homophones", id=1, original_text="gestures", corrected_text="Jesters", confidence=0.95, ), CorrectionProposal( proposal_index=1, module_instance="homophones", module_key="homophones", id=2, original_text="rank", corrected_text="Hrank", confidence=0.95, ), CorrectionProposal( proposal_index=2, module_instance="homophones", module_key="homophones", id=3, original_text="dam", corrected_text="damn", confidence=0.95, ), ] client = FakeStructuredLLMClient( [ { "validations": [ { "correction_index": 0, "approved": True, "confidence": 0.9, "reason": "ok", } ] }, { "validations": [ { "correction_index": 1, "approved": True, "confidence": 0.9, "reason": "ok", }, { "correction_index": 2, "approved": True, "confidence": 0.9, "reason": "ok", }, ] }, ] ) chunk_calls = [] def fake_chunk_payload_items(items, max_tokens, payload_fn, empty_error_message): chunk_calls.append( { "max_tokens": max_tokens, "payloads": [payload_fn(item) for item in items], "empty_error_message": empty_error_message, } ) return [ TokenBatch(batch_index=0, items=list(items[:1]), token_count=1), TokenBatch(batch_index=1, items=list(items[1:]), token_count=1), ] monkeypatch.setattr(llm_module, "chunk_payload_items", fake_chunk_payload_items) result = SpokenFormPlausibilityValidator("spoken_form_plausibility_review").validate( _context(proposals=proposals, transcript=transcript, llm_client=client, tmp_path=tmp_path) ) assert [decision.approved for decision in result.decisions] == [True, True, True] assert len(client.calls) == 2 assert len(chunk_calls) == 1 assert chunk_calls[0]["max_tokens"] == AuditaConfig.from_sources(env={"OPENROUTER_API_KEY": "test-key"}).max_section_tokens assert all("corrected_segment_text" in payload for payload in chunk_calls[0]["payloads"])