98 lines
3.8 KiB
Python
98 lines
3.8 KiB
Python
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from agent_framework.guardrails.base import RailDecision
|
|
from agent_framework.guardrails.output_supervisor import OutputSupervisor
|
|
from agent_framework.guardrails.rail_action import RailAction
|
|
from agent_framework.observability.code_mapper import create_observability_code_mapper
|
|
|
|
|
|
def _settings(**kwargs):
|
|
defaults = dict(
|
|
OBSERVABILITY_DEFAULT_MAPPING_ENABLED=True,
|
|
OBSERVABILITY_DEFAULT_MAPPING_PATH=None,
|
|
OBSERVABILITY_CODE_MAPPING_ENABLED=False,
|
|
OBSERVABILITY_CODE_MAPPING_PATH=None,
|
|
)
|
|
defaults.update(kwargs)
|
|
return SimpleNamespace(**defaults)
|
|
|
|
|
|
def test_old_agent_without_mapping_gets_framework_default_registry():
|
|
mapper = create_observability_code_mapper(_settings())
|
|
assert mapper.map("guardrail.output_supervisor.started") == "GRL.001"
|
|
assert mapper.map("guardrail.result.block") == "GRL.004"
|
|
assert mapper.map("guardrail.output_supervisor.completed") == "GRL.009"
|
|
assert mapper.action_for("REVPREC") == "retry"
|
|
assert mapper.action_for("CMP") == "retry"
|
|
assert mapper.action_for("SCO") == "retry"
|
|
assert mapper.action_for("GND") == "retry"
|
|
assert mapper.action_for("ATH") == "handover"
|
|
assert mapper.map("GRL.004") == "GRL.004"
|
|
|
|
|
|
def test_agent_overlay_overrides_only_declared_entries(tmp_path: Path):
|
|
overlay = tmp_path / "observability_mapping.yaml"
|
|
overlay.write_text(
|
|
"""version: \"2\"\nmappings:\n guardrail.dlex_in:\n label: GRL.004\n aliases: [DLEX_IN]\n guardrail.tox:\n label: GRL.005\n aliases: [TOX]\n""",
|
|
encoding="utf-8",
|
|
)
|
|
mapper = create_observability_code_mapper(_settings(
|
|
OBSERVABILITY_CODE_MAPPING_ENABLED=True,
|
|
OBSERVABILITY_CODE_MAPPING_PATH=str(overlay),
|
|
))
|
|
# Agent-specific contract overrides framework named GRL.DLEX_IN / GRL.TOX.
|
|
assert mapper.map("guardrail.dlex_in") == "GRL.004"
|
|
assert mapper.map("DLEX_IN") == "GRL.004"
|
|
assert mapper.map("TOX") == "GRL.005"
|
|
# Unrelated framework defaults remain intact.
|
|
assert mapper.map("guardrail.result.retry") == "GRL.005"
|
|
assert mapper.action_for("REVPREC") == "retry"
|
|
assert mapper.map("guardrail.pinj") == "GRL.PINJ"
|
|
|
|
|
|
def test_agent_overlay_can_override_default_action(tmp_path: Path):
|
|
overlay = tmp_path / "observability_mapping.yaml"
|
|
overlay.write_text(
|
|
"""mappings:\n guardrail.revprec:\n action: handover\n aliases: [REVPREC]\n""",
|
|
encoding="utf-8",
|
|
)
|
|
mapper = create_observability_code_mapper(_settings(
|
|
OBSERVABILITY_CODE_MAPPING_ENABLED=True,
|
|
OBSERVABILITY_CODE_MAPPING_PATH=str(overlay),
|
|
))
|
|
assert mapper.action_for("REVPREC") == "handover"
|
|
# Other historical defaults are still inherited.
|
|
assert mapper.action_for("CMP") == "retry"
|
|
|
|
|
|
def test_framework_default_carries_legacy_phraseology_rewrite_policy():
|
|
mapper = create_observability_code_mapper(_settings())
|
|
remediation = mapper.remediation_for("FRASEOLOGIA")
|
|
assert remediation is not None
|
|
assert remediation["type"] == "rewrite"
|
|
assert remediation["max_attempts"] == 1
|
|
assert remediation["generation_name"] == "guardrail.fraseologia.rewrite"
|
|
|
|
|
|
class _DeniedLegacyRail:
|
|
code = "REVPREC"
|
|
|
|
async def evaluate(self, text, context):
|
|
return RailDecision(code=self.code, allowed=False, reason="premature")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_old_legacy_rail_keeps_retry_with_new_framework_and_no_agent_mapping():
|
|
mapper = create_observability_code_mapper(_settings())
|
|
supervisor = OutputSupervisor(
|
|
rails=[_DeniedLegacyRail()],
|
|
enable_parallel=False,
|
|
observability_mapper=mapper,
|
|
)
|
|
decision = await supervisor.evaluate("candidate", {})
|
|
assert decision.action == RailAction.RETRY
|
|
assert decision.results[0].metadata.get("action_source") == "observability_mapping"
|