Implemented a custom validator for the glossary module that allows changes from one protected glossary term to another

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"),

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",

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

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"]],

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",

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(