From 619c615d211f1c0d71da6fc741cf818fafdfa784 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 25 Apr 2026 13:58:20 -0500 Subject: [PATCH] Implemented a custom validator for the glossary module that allows changes from one protected glossary term to another --- src/audita/modules/glossary.py | 4 +- src/audita/validators/__init__.py | 2 + src/audita/validators/deterministic.py | 24 +++++++++ src/audita/validators/protection.py | 20 +++++++ tests/test_new_pipeline.py | 13 +++-- tests/test_protected_validator.py | 74 +++++++++++++++++++++++++- 6 files changed, 131 insertions(+), 6 deletions(-) diff --git a/src/audita/modules/glossary.py b/src/audita/modules/glossary.py index b45330d..c2a106d 100644 --- a/src/audita/modules/glossary.py +++ b/src/audita/modules/glossary.py @@ -5,10 +5,10 @@ from audita.framework.models import CorrectionProposal, ModuleContext from audita.framework.proposal_generation import generate_llm_correction_proposals from audita.modules.prompts import build_glossary_proposal_messages from audita.validators import ( + GlossaryStageProtectedGlossaryTermsValidator, MeaningReversalValidator, NonEmptySegmentValidator, ProposalConfidenceValidator, - ProtectedGlossaryTermsValidator, SpokenFormPlausibilityValidator, Validator, ) @@ -21,7 +21,7 @@ class GlossaryModule: def validators(self) -> Sequence[Validator]: return [ ProposalConfidenceValidator("proposal_confidence_guard", "glossary_confidence_threshold"), - ProtectedGlossaryTermsValidator("protected_glossary_guard"), + GlossaryStageProtectedGlossaryTermsValidator("glossary_stage_protected_glossary_guard"), NonEmptySegmentValidator("non_empty_segment_guard"), SpokenFormPlausibilityValidator("spoken_form_plausibility_review"), MeaningReversalValidator("meaning_reversal_review"), diff --git a/src/audita/validators/__init__.py b/src/audita/validators/__init__.py index 66291ff..796f7fb 100644 --- a/src/audita/validators/__init__.py +++ b/src/audita/validators/__init__.py @@ -1,5 +1,6 @@ from .base import ValidationContext, ValidationDecision, ValidationResult, Validator from .deterministic import ( + GlossaryStageProtectedGlossaryTermsValidator, GrammarOnlyValidator, NonEmptySegmentValidator, ProposalConfidenceValidator, @@ -15,6 +16,7 @@ __all__ = [ "Validator", "ProposalConfidenceValidator", "ProtectedGlossaryTermsValidator", + "GlossaryStageProtectedGlossaryTermsValidator", "NonEmptySegmentValidator", "GrammarOnlyValidator", "ProtectedVocabulary", diff --git a/src/audita/validators/deterministic.py b/src/audita/validators/deterministic.py index ef2a34e..8634f09 100644 --- a/src/audita/validators/deterministic.py +++ b/src/audita/validators/deterministic.py @@ -53,6 +53,30 @@ class ProtectedGlossaryTermsValidator: ) +@dataclass(frozen=True) +class GlossaryStageProtectedGlossaryTermsValidator: + name: str + execution_kind: str = "deterministic" + + def validate(self, context: ValidationContext) -> ValidationResult: + vocabulary = ProtectedVocabulary.from_glossary(context.glossary) + return ValidationResult( + validator_name=self.name, + execution_kind=self.execution_kind, + decisions=[ + ValidationDecision( + proposal_index=proposal.proposal_index, + approved=(reason := vocabulary.glossary_stage_violation_reason( + proposal.original_text, + proposal.corrected_text, + )) is None, + reason=reason, + ) + for proposal in context.proposals + ], + ) + + @dataclass(frozen=True) class NonEmptySegmentValidator: name: str diff --git a/src/audita/validators/protection.py b/src/audita/validators/protection.py index 703f7d7..b1c18eb 100644 --- a/src/audita/validators/protection.py +++ b/src/audita/validators/protection.py @@ -38,6 +38,15 @@ class ProtectedVocabulary: return reason return self._validate_capitalization_transitions(before_occurrences, after_occurrences) + def glossary_stage_violation_reason(self, before: str, after: str) -> Optional[str]: + before_occurrences = self._occurrences_by_identity(before) + after_occurrences = self._occurrences_by_identity(after) + + reason = self._validate_glossary_stage_identity_preservation(before_occurrences, after_occurrences) + if reason is not None: + return reason + return self._validate_capitalization_transitions(before_occurrences, after_occurrences) + def _occurrences(self, text: str) -> List["_ProtectedOccurrence"]: if self.pattern is None: return [] @@ -70,6 +79,17 @@ class ProtectedVocabulary: return "correction changes protected glossary term usage" return None + def _validate_glossary_stage_identity_preservation( + self, + before_occurrences: Dict[int, List["_ProtectedOccurrence"]], + after_occurrences: Dict[int, List["_ProtectedOccurrence"]], + ) -> Optional[str]: + before_total = sum(len(items) for items in before_occurrences.values()) + after_total = sum(len(items) for items in after_occurrences.values()) + if after_total < before_total: + return "correction changes protected glossary term usage" + return None + def _validate_capitalization_transitions( self, before_occurrences: Dict[int, List["_ProtectedOccurrence"]], diff --git a/tests/test_new_pipeline.py b/tests/test_new_pipeline.py index 8116c22..ff71c4d 100644 --- a/tests/test_new_pipeline.py +++ b/tests/test_new_pipeline.py @@ -130,7 +130,14 @@ def test_process_transcript_result_writes_report_and_preserves_skips_per_policy( assert (result.run_dir / "normalization" / "summary.json").exists() assert [validator["name"] for validator in result.report.modules[0].to_dict()["validators"]] == [ "proposal_confidence_guard", - "protected_glossary_guard", + "glossary_stage_protected_glossary_guard", + "non_empty_segment_guard", + "spoken_form_plausibility_review", + "meaning_reversal_review", + ] + assert [validator["name"] for validator in result.report.modules[2].to_dict()["validators"]] == [ + "proposal_confidence_guard", + "glossary_stage_protected_glossary_guard", "non_empty_segment_guard", "spoken_form_plausibility_review", "meaning_reversal_review", @@ -184,7 +191,7 @@ def test_default_module_specs_expose_final_validator_order(): ] assert [validator.name for validator in specs[0].module.validators()] == [ "proposal_confidence_guard", - "protected_glossary_guard", + "glossary_stage_protected_glossary_guard", "non_empty_segment_guard", "spoken_form_plausibility_review", "meaning_reversal_review", @@ -198,7 +205,7 @@ def test_default_module_specs_expose_final_validator_order(): ] assert [validator.name for validator in specs[2].module.validators()] == [ "proposal_confidence_guard", - "protected_glossary_guard", + "glossary_stage_protected_glossary_guard", "non_empty_segment_guard", "spoken_form_plausibility_review", "meaning_reversal_review", diff --git a/tests/test_protected_validator.py b/tests/test_protected_validator.py index d33318f..2bf05e7 100644 --- a/tests/test_protected_validator.py +++ b/tests/test_protected_validator.py @@ -1,6 +1,10 @@ from audita.core.schemas import parse_glossary_yaml, parse_transcript_json from audita.framework.models import CorrectionProposal, ModuleRunSpec -from audita.validators import ProtectedGlossaryTermsValidator, ProtectedVocabulary +from audita.validators import ( + GlossaryStageProtectedGlossaryTermsValidator, + ProtectedGlossaryTermsValidator, + ProtectedVocabulary, +) from audita.validators.base import ValidationContext @@ -46,6 +50,22 @@ def test_protected_vocabulary_blocks_replacing_protected_term(): ) +def test_protected_vocabulary_glossary_stage_allows_glossary_to_glossary_changes(): + vocabulary = ProtectedVocabulary.from_glossary(_glossary()) + + assert vocabulary.violation_reason("Hrank moves.", "Popov moves.") == "correction changes protected glossary term usage" + assert vocabulary.glossary_stage_violation_reason("Hrank moves.", "Popov moves.") is None + + +def test_protected_vocabulary_glossary_stage_still_blocks_glossary_to_nonglossary_changes(): + vocabulary = ProtectedVocabulary.from_glossary(_glossary()) + + assert ( + vocabulary.glossary_stage_violation_reason("Hrank moves.", "Frank moves.") + == "correction changes protected glossary term usage" + ) + + def test_protected_vocabulary_blocks_noncanonical_capitalization(): vocabulary = ProtectedVocabulary.from_glossary(_glossary()) @@ -57,6 +77,10 @@ def test_protected_vocabulary_blocks_noncanonical_capitalization(): vocabulary.violation_reason("gestures", "jesters") == "correction changes protected glossary term capitalization" ) + assert ( + vocabulary.glossary_stage_violation_reason("Hrank moves.", "POPOV moves.") + == "correction changes protected glossary term capitalization" + ) def test_protected_vocabulary_allows_corrections_toward_protected_terms(): @@ -155,6 +179,54 @@ def test_protected_glossary_terms_validator_returns_proposal_indexed_decisions() assert result.decisions[1].approved is True +def test_glossary_stage_protected_glossary_terms_validator_allows_glossary_to_glossary_replacement(): + validator = GlossaryStageProtectedGlossaryTermsValidator("glossary_stage_protected_glossary_guard") + transcript = parse_transcript_json( + """ + [ + {"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Hrank moves."}, + {"id": 2, "speaker": "Eric", "start": 1.0, "end": 2.0, "text": "Hrank moves."} + ] + """ + ) + proposals = [ + CorrectionProposal( + proposal_index=0, + module_instance="glossary_1", + module_key="glossary", + id=1, + original_text="Hrank", + corrected_text="Popov", + confidence=0.9, + ), + CorrectionProposal( + proposal_index=1, + module_instance="glossary_1", + module_key="glossary", + id=2, + original_text="Hrank", + corrected_text="POPOV", + confidence=0.9, + ), + ] + + result = validator.validate( + ValidationContext( + proposals=proposals, + transcript=transcript, + glossary=_glossary(), + config=None, # type: ignore[arg-type] + run_spec=ModuleRunSpec(instance_name="glossary_1", module_key="glossary", module=None), # type: ignore[arg-type] + run_dir=__import__("pathlib").Path("."), + ) + ) + + assert [decision.proposal_index for decision in result.decisions] == [0, 1] + assert result.decisions[0].approved is True + assert result.decisions[1].approved is False + assert result.decisions[1].reason == "correction changes protected glossary term capitalization" + + def test_protected_glossary_terms_validator_uses_proposal_span_only(): validator = ProtectedGlossaryTermsValidator("protected_glossary_guard") glossary = parse_glossary_yaml(