diff --git a/src/audita/corrections.py b/src/audita/corrections.py index 5fa6cab..b6d7a0e 100644 --- a/src/audita/corrections.py +++ b/src/audita/corrections.py @@ -60,7 +60,7 @@ def apply_corrections( segment = revised[position] revised_text = segment.text.replace(correction.original_text, correction.corrected_text) if correction_guard is not None: - reason = correction_guard(segment.text, revised_text) + reason = correction_guard(correction.original_text, correction.corrected_text) if reason is not None: skipped.append(_skip(correction, reason, actual_text=segment.text)) continue diff --git a/src/audita/prompts.py b/src/audita/prompts.py index 0506dd0..7e949c4 100644 --- a/src/audita/prompts.py +++ b/src/audita/prompts.py @@ -13,7 +13,7 @@ def build_glossary_correction_messages( glossary: Glossary, retry_pass: bool = False, ) -> List[Message]: - glossary_json = json.dumps(glossary.model_dump(mode="json"), ensure_ascii=False, indent=2) + glossary_json = json.dumps(glossary.model_dump(mode="json", exclude_none=True), ensure_ascii=False, indent=2) section_json = json.dumps(section.prompt_payload(), ensure_ascii=False, indent=2) system = ( @@ -67,7 +67,7 @@ def build_grammar_correction_messages( glossary: Glossary, retry_pass: bool = False, ) -> List[Message]: - glossary_json = json.dumps(glossary.model_dump(mode="json"), ensure_ascii=False, indent=2) + glossary_json = json.dumps(glossary.model_dump(mode="json", exclude_none=True), ensure_ascii=False, indent=2) section_json = json.dumps(section.prompt_payload(), ensure_ascii=False, indent=2) system = ( diff --git a/src/audita/protection.py b/src/audita/protection.py index 8dcc5b4..78a6a77 100644 --- a/src/audita/protection.py +++ b/src/audita/protection.py @@ -7,39 +7,40 @@ from .schemas import Glossary @dataclass(frozen=True) class ProtectedVocabulary: - canonical_by_folded: Dict[str, str] + terms_by_folded: Dict[str, "_ProtectedTermDefinition"] pattern: Optional[Pattern[str]] @classmethod def from_glossary(cls, glossary: Glossary) -> "ProtectedVocabulary": - canonical_by_folded: Dict[str, str] = {} - for entry in glossary.glossary: - _add_term(canonical_by_folded, entry.name) - for alias in entry.aliases: - _add_term(canonical_by_folded, alias) + terms_by_folded: Dict[str, _ProtectedTermDefinition] = {} + for identity, entry in enumerate(glossary.glossary): + entry_terms = [entry.name, *entry.aliases] + for term in entry_terms: + _add_term(terms_by_folded, term, identity) + _add_term(terms_by_folded, f"{term}s", identity) + if entry.plural is not None: + _add_term(terms_by_folded, entry.plural, identity) - terms = list(canonical_by_folded.values()) + terms = [definition.canonical for definition in terms_by_folded.values()] if not terms: - return cls(canonical_by_folded=canonical_by_folded, pattern=None) + return cls(terms_by_folded=terms_by_folded, pattern=None) alternatives = sorted((re.escape(term) for term in terms), key=len, reverse=True) pattern = re.compile(r"(? Optional[str]: before_terms = self._terms(before) after_terms = self._terms(after) - if not before_terms: - return None - after_folded = [term.folded for term in after_terms] - for before_term in before_terms: - if before_term.folded not in after_folded: - return "correction changes protected glossary term usage" + if before_terms: + after_identities = [term.identity for term in after_terms] + for before_term in before_terms: + if before_term.identity not in after_identities: + return "correction changes protected glossary term usage" for after_term in after_terms: - canonical = self.canonical_by_folded[after_term.folded] - if after_term.text != canonical: + if after_term.text != after_term.canonical: return "correction changes protected glossary term capitalization" return None @@ -47,20 +48,42 @@ class ProtectedVocabulary: def _terms(self, text: str) -> List["_ProtectedTerm"]: if self.pattern is None: return [] - return [ - _ProtectedTerm(text=match.group(0), folded=match.group(0).casefold()) - for match in self.pattern.finditer(text) - ] + terms = [] + for match in self.pattern.finditer(text): + matched_text = match.group(0) + definition = self.terms_by_folded[matched_text.casefold()] + terms.append( + _ProtectedTerm( + text=matched_text, + identity=definition.identity, + canonical=definition.canonical, + ) + ) + return terms @dataclass(frozen=True) class _ProtectedTerm: text: str - folded: str + identity: int + canonical: str -def _add_term(canonical_by_folded: Dict[str, str], term: str) -> None: +@dataclass(frozen=True) +class _ProtectedTermDefinition: + identity: int + canonical: str + + +def _add_term( + terms_by_folded: Dict[str, _ProtectedTermDefinition], + term: str, + identity: int, +) -> None: stripped = term.strip() if not stripped: return - canonical_by_folded.setdefault(stripped.casefold(), stripped) + terms_by_folded.setdefault( + stripped.casefold(), + _ProtectedTermDefinition(identity=identity, canonical=stripped), + ) diff --git a/src/audita/schemas.py b/src/audita/schemas.py index a9a8297..9998be7 100644 --- a/src/audita/schemas.py +++ b/src/audita/schemas.py @@ -99,6 +99,7 @@ class GlossaryEntry(BaseModel): name: StrictStr aliases: List[StrictStr] = Field(default_factory=list) + plural: Optional[StrictStr] = None category: StrictStr summary: StrictStr @@ -117,6 +118,13 @@ class GlossaryEntry(BaseModel): raise ValueError("aliases must not contain empty strings") return aliases + @field_validator("plural") + @classmethod + def require_non_empty_plural(cls, plural: Optional[str]) -> Optional[str]: + if plural is not None and not plural.strip(): + raise ValueError("plural must not be empty") + return plural + class Glossary(BaseModel): model_config = ConfigDict(extra="forbid") diff --git a/tests/test_corrections.py b/tests/test_corrections.py index c064609..0fd8d77 100644 --- a/tests/test_corrections.py +++ b/tests/test_corrections.py @@ -2,7 +2,8 @@ import pytest from audita.corrections import apply_corrections from audita.errors import AuditaValidationError -from audita.schemas import CorrectionCandidate, parse_transcript_json +from audita.protection import ProtectedVocabulary +from audita.schemas import CorrectionCandidate, parse_glossary_yaml, parse_transcript_json def _transcript(): @@ -200,6 +201,44 @@ def test_apply_corrections_skips_when_guard_rejects_replacement(): assert result.skipped[0].actual_text == "Hrank moves." +def test_apply_corrections_guards_only_replacement_span(): + transcript = parse_transcript_json( + """ + [ + {"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "You have to keep it bind. Svend sees the jesters."} + ] + """ + ) + glossary = parse_glossary_yaml( + """ + glossary: + - name: "Svend" + category: pc + summary: "Svend is a player character." + - name: "Jesters" + category: faction + summary: "The Jesters are a faction." + """ + ) + correction = CorrectionCandidate( + id=1, + original_text="keep it bind", + corrected_text="keep in mind", + confidence=0.8, + ) + + result = apply_corrections( + transcript, + [correction], + confidence_threshold=0.8, + replacement_mode="require_unique", + correction_guard=ProtectedVocabulary.from_glossary(glossary).violation_reason, + ) + + assert result.transcript[0].text == "You have to keep in mind. Svend sees the jesters." + assert result.skipped == [] + + def test_apply_corrections_without_guard_remains_permissive(): transcript = parse_transcript_json( """ diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index b4b8fae..6157512 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -119,6 +119,49 @@ def test_glossary_stage_can_correct_toward_protected_term(tmp_path): assert list((tmp_path / "work").iterdir()) == [] +def test_glossary_guard_ignores_unrelated_protected_terms_elsewhere_in_segment(tmp_path): + transcript = parse_source_transcript_json( + """ + [ + {"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "The gestures are near lyra."} + ] + """ + ) + glossary = parse_glossary_yaml( + """ + glossary: + - name: "Jesters" + category: faction + summary: "The Jesters are a faction." + - name: "Lyra" + category: npc + summary: "Lyra is an NPC." + """ + ) + glossary_correction = CorrectionCandidate( + id=1, + original_text="gestures", + corrected_text="Jesters", + confidence=0.95, + ) + fake_client = FakeLLMClient( + [ + CorrectionSet(corrections=[glossary_correction]), + CorrectionSet(corrections=[]), + ] + ) + + revised = process_transcript( + transcript, + glossary, + _config(tmp_path), + llm_client=fake_client, + ) + + assert revised[0].text == "The Jesters are near lyra." + assert list((tmp_path / "work").iterdir()) == [] + + def test_glossary_stage_cannot_change_away_from_protected_term(tmp_path): transcript = parse_source_transcript_json( """ diff --git a/tests/test_prompts.py b/tests/test_prompts.py index 7679a6d..96a8646 100644 --- a/tests/test_prompts.py +++ b/tests/test_prompts.py @@ -70,6 +70,44 @@ def test_prompt_uses_simplified_segment_payload(): assert "end" not in prompt_segments[0] +def test_prompts_do_not_include_inferred_plurals(): + transcript = parse_transcript_json( + """ + [ + {"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "The gestures are nearby."} + ] + """ + ) + glossary = parse_glossary_yaml( + """ + glossary: + - name: "Godfrey" + aliases: + - "Jester" + category: npc + summary: "Godfrey is an NPC." + """ + ) + section = chunk_transcript(transcript, max_section_tokens=16000)[0] + + glossary_messages = build_glossary_correction_messages(section, glossary) + glossary_json = glossary_messages[1]["content"].split("Glossary:\n", maxsplit=1)[1].split( + "\n\nTranscript section:", + maxsplit=1, + )[0] + grammar_messages = build_grammar_correction_messages(section, glossary) + grammar_json = grammar_messages[1]["content"].split("Protected glossary/context:\n", maxsplit=1)[1].split( + "\n\nTranscript section:", + maxsplit=1, + )[0] + + for prompt_glossary in (json.loads(glossary_json), json.loads(grammar_json)): + entry = prompt_glossary["glossary"][0] + assert "plural" not in entry + assert "Godfreys" not in json.dumps(prompt_glossary) + assert "Jesters" not in json.dumps(prompt_glossary) + + def test_grammar_prompt_limits_readability_corrections_and_protects_glossary(): transcript = parse_transcript_json( """ diff --git a/tests/test_protection.py b/tests/test_protection.py index 73ddb03..1ca20de 100644 --- a/tests/test_protection.py +++ b/tests/test_protection.py @@ -14,6 +14,17 @@ def _vocabulary(): - name: "Popov" category: npc summary: "Popov is an allied NPC." + - name: "Jesters" + aliases: + - "Jester" + category: faction + summary: "The Jesters are a faction." + - name: "Svend" + category: pc + summary: "Svend is a player character." + - name: "Godfrey" + category: npc + summary: "Godfrey is an NPC." """ ) return ProtectedVocabulary.from_glossary(glossary) @@ -59,6 +70,9 @@ def test_protection_allows_correction_toward_protected_term(): vocabulary = _vocabulary() assert vocabulary.violation_reason("Pawpaw moves.", "Popov moves.") is None + assert vocabulary.violation_reason("gestures", "Jesters") is None + assert vocabulary.violation_reason("rank", "Hrank") is None + assert vocabulary.violation_reason("spend", "Svend") is None def test_protection_allows_possessive_correction_toward_protected_term(): @@ -67,6 +81,54 @@ def test_protection_allows_possessive_correction_toward_protected_term(): assert vocabulary.violation_reason("Pawpaw's exhausted.", "Popov's exhausted.") is None +def test_protection_blocks_noncanonical_introduced_protected_term(): + vocabulary = _vocabulary() + + assert ( + vocabulary.violation_reason("gestures", "jesters") + == "correction changes protected glossary term capitalization" + ) + assert ( + vocabulary.violation_reason("rank", "hrank") + == "correction changes protected glossary term capitalization" + ) + assert ( + vocabulary.violation_reason("spend", "svend") + == "correction changes protected glossary term capitalization" + ) + + +def test_protection_allows_inferred_name_plural(): + vocabulary = _vocabulary() + + assert vocabulary.violation_reason("Godfrey's", "Godfreys") is None + + +def test_protection_allows_inferred_alias_plural(): + vocabulary = _vocabulary() + + assert vocabulary.violation_reason("gesture", "Jesters") is None + + +def test_protection_allows_explicit_plural(): + glossary = parse_glossary_yaml( + """ + glossary: + - name: "Mox" + plural: "Moxen" + category: faction + summary: "The Mox are a faction." + """ + ) + vocabulary = ProtectedVocabulary.from_glossary(glossary) + + assert vocabulary.violation_reason("Mox's", "Moxen") is None + assert ( + vocabulary.violation_reason("Mox's", "moxen") + == "correction changes protected glossary term capitalization" + ) + + def test_protection_allows_punctuation_around_protected_term(): vocabulary = _vocabulary() diff --git a/tests/test_validation.py b/tests/test_validation.py index 7974384..17142b4 100644 --- a/tests/test_validation.py +++ b/tests/test_validation.py @@ -177,6 +177,34 @@ def test_valid_glossary_parses(): ) assert glossary.glossary[0].name == "Lyra" + assert glossary.glossary[0].plural is None + + +def test_glossary_accepts_optional_plural(): + glossary = parse_glossary_yaml( + """ + glossary: + - name: "Godfrey" + plural: "Godfreys" + category: npc + summary: "Godfrey is an NPC." + """ + ) + + assert glossary.glossary[0].plural == "Godfreys" + + +def test_glossary_rejects_empty_plural(): + with pytest.raises(AuditaValidationError): + parse_glossary_yaml( + """ + glossary: + - name: "Godfrey" + plural: "" + category: npc + summary: "Godfrey is an NPC." + """ + ) def test_glossary_rejects_empty_entries():