eric/audita
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Added an interface for validators, and implemented an initial deterministic validator to protect terms in the provided glossary

Browse Source
This commit is contained in:
Eric Rakestraw
2026-04-24 11:08:32 -05:00
parent f39de37974
commit bca2152971
15 changed files with 696 additions and 392 deletions
Download Patch File Download Diff File Expand all files Collapse all files

227
tests/test_framework_runner.py
View File

@@ -1,81 +1,56 @@
from pathlib import Path
from audita.core.config import AuditaConfig
from audita.core.schemas import parse_glossary_yaml, parse_transcript_json
from audita.framework.models import (
CorrectionProposal,
FilterDecision,
ModuleContext,
ModuleRunSpec,
ReviewDecision,
)
from audita.framework.models import CorrectionProposal, ModuleContext, ModuleRunSpec
from audita.framework.runner import PipelineRunner
from audita.validators import ProtectedGlossaryTermsValidator
from audita.validators.base import ValidationContext, ValidationDecision, ValidationResult
class AllowAllFilter:
name = "allow_all"
class RecordingValidator:
execution_kind = "deterministic"
def evaluate(self, proposal, transcript, glossary, config):
return FilterDecision(approved=True)
def __init__(self, name, recorder, approve=True):
self.name = name
self._recorder = recorder
self._approve = approve
def validate(self, context: ValidationContext) -> ValidationResult:
self._recorder.append((self.name, [proposal.corrected_text for proposal in context.proposals]))
return ValidationResult(
validator_name=self.name,
execution_kind=self.execution_kind,
decisions=[
ValidationDecision(
proposal_index=proposal.proposal_index,
approved=self._approve,
reason=None if self._approve else f"{self.name} rejected proposal",
)
for proposal in context.proposals
],
)
class RejectAllFilter:
name = "reject_all"
def evaluate(self, proposal, transcript, glossary, config):
return FilterDecision(approved=False, reason="filter rejected proposal")
class AllowAllReviewStage:
name = "allow_all_review"
def review(self, proposals, transcript, glossary, config, llm_client, run_dir):
return [ReviewDecision(proposal_index=proposal.proposal_index, approved=True) for proposal in proposals]
class RecordingLLMValidator(RecordingValidator):
execution_kind = "llm"
class RecordingModule:
replacement_policy = "require_unique"
def __init__(self, module_key, proposals, recorder):
def __init__(self, module_key, proposals, validators, recorder):
self.module_key = module_key
self._proposals = proposals
self._validators = validators
self._recorder = recorder
def deterministic_filters(self):
return [AllowAllFilter()]
def review_stages(self):
return [AllowAllReviewStage()]
def validators(self):
return list(self._validators)
def propose(self, transcript_section, context: ModuleContext):
self._recorder.append([segment.text for segment in transcript_section])
self._recorder.append(("propose", [segment.text for segment in transcript_section]))
return list(self._proposals)
class RejectedModule:
module_key = "rejected"
replacement_policy = "require_unique"
def deterministic_filters(self):
return [RejectAllFilter()]
def review_stages(self):
return []
def propose(self, transcript_section, context):
return [
CorrectionProposal(
proposal_index=0,
module_instance=context.run_spec.instance_name,
module_key=context.run_spec.module_key,
id=1,
original_text="Hello",
corrected_text="Goodbye",
confidence=0.9,
)
]
def test_pipeline_runner_applies_modules_sequentially(tmp_path):
transcript = parse_transcript_json(
"""
@@ -106,6 +81,7 @@ def test_pipeline_runner_applies_modules_sequentially(tmp_path):
confidence=0.9,
)
],
[RecordingValidator("first_validator", seen)],
seen,
)
second = RecordingModule(
@@ -121,6 +97,7 @@ def test_pipeline_runner_applies_modules_sequentially(tmp_path):
confidence=0.9,
)
],
[RecordingValidator("second_validator", seen)],
seen,
)
@@ -136,13 +113,15 @@ def test_pipeline_runner_applies_modules_sequentially(tmp_path):
run_dir=tmp_path / "run",
)
assert seen[0] == ["Alpha."]
assert seen[1] == ["Beta."]
assert seen[0] == ("propose", ["Alpha."])
assert seen[1] == ("first_validator", ["Beta"])
assert seen[2] == ("propose", ["Beta."])
assert seen[3] == ("second_validator", ["Gamma"])
assert result.transcript[0].text == "Gamma."
assert len(result.applied_changes) == 2
def test_pipeline_runner_reports_filter_rejections(tmp_path):
def test_pipeline_runner_validator_order_respects_survivors(tmp_path):
transcript = parse_transcript_json(
"""
[
@@ -158,16 +137,138 @@ def test_pipeline_runner_reports_filter_rejections(tmp_path):
summary: "Hello."
"""
)
seen = []
module = RecordingModule(
"mod",
[
CorrectionProposal(
proposal_index=0,
module_instance="mod",
module_key="mod",
id=1,
original_text="Hello",
corrected_text="Goodbye",
confidence=0.9,
)
],
[
RecordingValidator("first", seen, approve=False),
RecordingLLMValidator("second", seen, approve=True),
],
seen,
)
runner = PipelineRunner()
result = runner.run(
transcript=transcript,
glossary=glossary,
module_specs=[ModuleRunSpec(instance_name="rejected", module_key="rejected", module=RejectedModule())],
module_specs=[ModuleRunSpec(instance_name="mod", module_key="mod", module=module)],
config=AuditaConfig.from_sources(env={}),
run_dir=tmp_path / "run",
)
assert result.transcript[0].text == "Hello."
assert result.module_reports[0].skipped_count == 1
assert result.skipped_corrections[0].reason == "filter rejected proposal"
assert ("first", ["Goodbye"]) in seen
assert all(entry[0] != "second" for entry in seen)
assert result.module_reports[0].validators[0].rejected_count == 1
assert result.module_reports[0].validators[1].candidate_count == 0
assert result.skipped_corrections[0].source == "validator:first"
def test_pipeline_runner_supports_deterministic_and_llm_validators_in_one_chain(tmp_path):
transcript = parse_transcript_json(
"""
[
{"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "Alpha."}
]
"""
)
glossary = parse_glossary_yaml(
"""
glossary:
- name: "Alpha"
category: noun
summary: "Alpha."
"""
)
seen = []
module = RecordingModule(
"mixed",
[
CorrectionProposal(
proposal_index=0,
module_instance="mixed",
module_key="mixed",
id=1,
original_text="Alpha",
corrected_text="Beta",
confidence=0.9,
)
],
[
RecordingValidator("deterministic_guard", seen),
RecordingLLMValidator("llm_review", seen),
],
seen,
)
runner = PipelineRunner()
result = runner.run(
transcript=transcript,
glossary=glossary,
module_specs=[ModuleRunSpec(instance_name="mixed", module_key="mixed", module=module)],
config=AuditaConfig.from_sources(env={}),
run_dir=tmp_path / "run",
)
assert result.transcript[0].text == "Beta."
assert [report.execution_kind for report in result.module_reports[0].validators] == [
"deterministic",
"llm",
]
def test_pipeline_runner_uses_real_protected_glossary_validator(tmp_path):
transcript = parse_transcript_json(
"""
[
{"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "Hrank moves."}
]
"""
)
glossary = parse_glossary_yaml(
"""
glossary:
- name: "Hrank"
category: pc
summary: "Hrank is a player character."
"""
)
module = RecordingModule(
"protected",
[
CorrectionProposal(
proposal_index=0,
module_instance="protected",
module_key="protected",
id=1,
original_text="Hrank",
corrected_text="Frank",
confidence=0.9,
)
],
[ProtectedGlossaryTermsValidator("protected_glossary_guard")],
[],
)
runner = PipelineRunner()
result = runner.run(
transcript=transcript,
glossary=glossary,
module_specs=[ModuleRunSpec(instance_name="protected", module_key="protected", module=module)],
config=AuditaConfig.from_sources(env={}),
run_dir=tmp_path / "run",
)
assert result.transcript[0].text == "Hrank moves."
assert result.skipped_corrections[0].source == "validator:protected_glossary_guard"
assert result.skipped_corrections[0].reason == "correction changes protected glossary term usage"

36
tests/test_new_pipeline.py
View File

@@ -3,6 +3,7 @@ import json
from audita.core.config import AuditaConfig
from audita.core.io import write_report
from audita.core.schemas import parse_glossary_yaml, parse_source_transcript_json
from audita.modules import default_module_specs
from audita.pipeline import process_transcript, process_transcript_result
@@ -73,6 +74,11 @@ def test_process_transcript_result_writes_report_and_preserves_skips_per_policy(
assert result.report.totals["applied_change_count"] == 0
assert (result.run_dir / "report.json").exists()
assert (result.run_dir / "normalization" / "summary.json").exists()
assert [validator["name"] for validator in result.report.modules[0].to_dict()["validators"]] == [
"protected_glossary_guard",
"toward_glossary_term_review",
"context_support_review",
]
def test_external_report_can_be_written(tmp_path):
@@ -84,3 +90,33 @@ def test_external_report_can_be_written(tmp_path):
payload = json.loads(report_path.read_text(encoding="utf-8"))
assert payload["pipeline"][0] == "glossary_primary"
assert payload["totals"]["applied_change_count"] == 0
def test_default_module_specs_expose_final_validator_order():
specs = default_module_specs()
assert [validator.name for validator in specs[0].module.validators()] == [
"protected_glossary_guard",
"toward_glossary_term_review",
"context_support_review",
]
assert [validator.name for validator in specs[1].module.validators()] == [
"protected_glossary_guard",
"acoustic_similarity_review",
"contextual_plausibility_review",
"antonym_reversal_review",
]
assert [validator.name for validator in specs[2].module.validators()] == [
"protected_glossary_guard",
"toward_glossary_term_review",
"context_support_review",
]
assert [validator.name for validator in specs[3].module.validators()] == [
"protected_glossary_guard",
"spoken_marker_cleanup_review",
"meaning_preservation_review",
]
assert [validator.name for validator in specs[4].module.validators()] == [
"protected_glossary_guard",
"edited_text_readability_review",
]

201
tests/test_protected_validator.py Normal file
View File

@@ -0,0 +1,201 @@
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.base import ValidationContext
def _glossary():
return parse_glossary_yaml(
"""
glossary:
- name: "Hrank"
aliases:
- "Greenfield"
category: pc
summary: "Hrank Greenfield is a player character."
- 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."
- name: "Lyra"
category: npc
summary: "Lyra is an NPC."
- name: "Loviator"
category: deity
summary: "Loviator is a deity."
"""
)
def test_protected_vocabulary_blocks_replacing_protected_term():
vocabulary = ProtectedVocabulary.from_glossary(_glossary())
assert (
vocabulary.violation_reason("Hrank moves.", "Frank moves.")
== "correction changes protected glossary term usage"
)
def test_protected_vocabulary_blocks_noncanonical_capitalization():
vocabulary = ProtectedVocabulary.from_glossary(_glossary())
assert (
vocabulary.violation_reason("Popov moves.", "POPOV moves.")
== "correction changes protected glossary term capitalization"
)
assert (
vocabulary.violation_reason("gestures", "jesters")
== "correction changes protected glossary term capitalization"
)
def test_protected_vocabulary_allows_corrections_toward_protected_terms():
vocabulary = ProtectedVocabulary.from_glossary(_glossary())
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_protected_vocabulary_allows_unchanged_noncanonical_terms_and_quote_wrapping():
vocabulary = ProtectedVocabulary.from_glossary(_glossary())
assert vocabulary.violation_reason("jesters advance.", "jesters advance.") is None
before = (
"When you say that, Popov will say, when I was in that room with the jesters, "
"I just knew that Godfrey and Lyra came directly from Loviator herself."
)
after = (
'When you say that, Popov will say, "When I was in that room with the jesters, '
'I just knew that Godfrey and Lyra came directly from Loviator herself."'
)
assert vocabulary.violation_reason(before, after) is None
def test_protected_vocabulary_allows_inferred_and_explicit_plurals():
vocabulary = ProtectedVocabulary.from_glossary(_glossary())
explicit = ProtectedVocabulary.from_glossary(
parse_glossary_yaml(
"""
glossary:
- name: "Mox"
plural: "Moxen"
category: faction
summary: "The Mox are a faction."
"""
)
)
assert vocabulary.violation_reason("Godfrey's", "Godfreys") is None
assert vocabulary.violation_reason("gesture", "Jesters") is None
assert explicit.violation_reason("Mox's", "Moxen") is None
def test_protected_vocabulary_does_not_match_embedded_substrings():
vocabulary = ProtectedVocabulary.from_glossary(_glossary())
assert vocabulary.violation_reason("The shrank spell worked.", "The shrank spell works.") is None
def test_protected_glossary_terms_validator_returns_proposal_indexed_decisions():
validator = ProtectedGlossaryTermsValidator("protected_glossary_guard")
transcript = parse_transcript_json(
"""
[
{"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Frank moves."},
{"id": 2, "speaker": "Eric", "start": 1.0, "end": 2.0, "text": "Pawpaw waits."}
]
"""
)
proposals = [
CorrectionProposal(
proposal_index=0,
module_instance="glossary_primary",
module_key="glossary",
id=1,
original_text="Hrank",
corrected_text="Frank",
confidence=0.9,
),
CorrectionProposal(
proposal_index=1,
module_instance="glossary_primary",
module_key="glossary",
id=2,
original_text="Pawpaw",
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_primary", module_key="glossary", module=None), # type: ignore[arg-type]
run_dir=transcript[0].__class__.__module__ and __import__("pathlib").Path("."),
)
)
assert [decision.proposal_index for decision in result.decisions] == [0, 1]
assert result.decisions[0].approved is False
assert result.decisions[0].reason == "correction changes protected glossary term usage"
assert result.decisions[1].approved is True
def test_protected_glossary_terms_validator_uses_proposal_span_only():
validator = ProtectedGlossaryTermsValidator("protected_glossary_guard")
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."
"""
)
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."}
]
"""
)
proposals = [
CorrectionProposal(
proposal_index=0,
module_instance="spoken_word",
module_key="spoken_word",
id=1,
original_text="keep it bind",
corrected_text="keep in mind",
confidence=0.9,
)
]
result = validator.validate(
ValidationContext(
proposals=proposals,
transcript=transcript,
glossary=glossary,
config=None, # type: ignore[arg-type]
run_spec=ModuleRunSpec(instance_name="spoken_word", module_key="spoken_word", module=None), # type: ignore[arg-type]
run_dir=__import__("pathlib").Path("."),
)
)
assert result.decisions[0].approved is True