861 lines
38 KiB
Python
861 lines
38 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
|
||
import agent_detector_runner as runner
|
||
|
||
|
||
def write_json(path: Path, payload: object) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
|
||
def write_text(path: Path, text: str) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(text, encoding="utf-8")
|
||
|
||
|
||
def to_win1251_mojibake(value: str) -> str:
|
||
return value.encode("utf-8").decode("cp1251")
|
||
|
||
|
||
class AgentDetectorRunnerTests(unittest.TestCase):
|
||
def test_default_runner_fails_missing_effective_runtime(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
artifact_dir.mkdir()
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"missing_effective_runtime_json": {
|
||
"kind": "artifact_presence",
|
||
"automation_level": "automatic",
|
||
"description": "Manifest is required.",
|
||
"issue_codes": ["runtime_manifest_missing"],
|
||
"inputs": ["effective_runtime.json"],
|
||
"check": {"required_files": ["effective_runtime.json"]},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(
|
||
issue_catalog_path,
|
||
{
|
||
"schema_version": "agent_issue_catalog_v1",
|
||
"issues": {
|
||
"runtime_manifest_missing": {
|
||
"detectors": ["missing_effective_runtime_json"],
|
||
}
|
||
},
|
||
},
|
||
)
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "fail")
|
||
self.assertEqual(results["results"][0]["detector"], "missing_effective_runtime_json")
|
||
self.assertEqual(results["results"][0]["status"], "fail")
|
||
|
||
def test_default_runner_passes_when_effective_runtime_exists(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
write_json(artifact_dir / "effective_runtime.json", {"runner": "test"})
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"missing_effective_runtime_json": {
|
||
"kind": "artifact_presence",
|
||
"automation_level": "automatic",
|
||
"description": "Manifest is required.",
|
||
"issue_codes": ["runtime_manifest_missing"],
|
||
"inputs": ["effective_runtime.json"],
|
||
"check": {"required_files": ["effective_runtime.json"]},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "pass")
|
||
self.assertEqual(results["results"][0]["status"], "pass")
|
||
|
||
def test_limited_next_action_accepts_mojibake_what_to_check_next(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
answer = (
|
||
"Это не подтвержденная чистая прибыль.\n"
|
||
"Что проверить дальше: чистую прибыль через счета 90/91/99 и закрытие периода."
|
||
)
|
||
write_text(artifact_dir / "steps" / "s01" / "output.md", to_win1251_mojibake(answer))
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"limited_answer_without_next_action": {
|
||
"kind": "answer_text_required_when_limited",
|
||
"automation_level": "semi_automatic",
|
||
"description": "Limited answers need a next action.",
|
||
"issue_codes": ["business_next_step_missing"],
|
||
"inputs": ["output.md"],
|
||
"check": {
|
||
"limited_patterns": ["(?i)не подтвержден"],
|
||
"required_next_action_patterns_any": ["(?i)(можно|проверь)"],
|
||
},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(
|
||
issue_catalog_path,
|
||
{
|
||
"schema_version": "agent_issue_catalog_v1",
|
||
"issues": {
|
||
"business_next_step_missing": {
|
||
"detectors": ["limited_answer_without_next_action"],
|
||
}
|
||
},
|
||
},
|
||
)
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
issue_codes=["business_next_step_missing"],
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "pass")
|
||
self.assertEqual(results["results"][0]["message"], "limited answer includes a next action")
|
||
|
||
def test_limited_next_action_ignores_non_limited_off_domain_can_not_phrase(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
step_dir = artifact_dir / "scenarios" / "chat" / "steps" / "s01"
|
||
write_text(step_dir / "output.md", "Нет, пингвины не могут летать.")
|
||
write_json(
|
||
step_dir / "step_state.json",
|
||
{
|
||
"business_first_review": {
|
||
"limited_answer_detected": False,
|
||
"next_action_present": False,
|
||
}
|
||
},
|
||
)
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"limited_answer_without_next_action": {
|
||
"kind": "answer_text_required_when_limited",
|
||
"automation_level": "semi_automatic",
|
||
"description": "Limited answers need a next action.",
|
||
"issue_codes": ["business_next_step_missing"],
|
||
"inputs": ["output.md"],
|
||
"check": {
|
||
"limited_patterns": ["(?i)не могу"],
|
||
"required_next_action_patterns_any": ["(?i)что проверить дальше"],
|
||
},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(
|
||
issue_catalog_path,
|
||
{
|
||
"schema_version": "agent_issue_catalog_v1",
|
||
"issues": {
|
||
"business_next_step_missing": {
|
||
"detectors": ["limited_answer_without_next_action"],
|
||
}
|
||
},
|
||
},
|
||
)
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
issue_codes=["business_next_step_missing"],
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "pass")
|
||
self.assertEqual(results["results"][0]["message"], "answer is not limited; next action requirement not triggered")
|
||
|
||
def test_candidate_forbidden_regex_is_limited_to_evidence_path(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
write_text(artifact_dir / "scenarios" / "ok" / "steps" / "s01" / "output.md", "Маржа не подтверждена.")
|
||
write_text(
|
||
artifact_dir / "scenarios" / "bad" / "steps" / "s02" / "output.md",
|
||
"Это амортизация объекта ОС, но этот шаг не в evidence scope.",
|
||
)
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
candidates_path = root / "detector_candidates.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"forbidden_margin_terms": {
|
||
"kind": "answer_text_regex_forbidden",
|
||
"automation_level": "automatic",
|
||
"description": "No wrong-domain words.",
|
||
"issue_codes": ["margin_domain_leak_accounting_route"],
|
||
"inputs": ["output.md"],
|
||
"check": {"forbidden_patterns": ["(?i)(амортизац|объект\\s+ОС)"]},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(
|
||
issue_catalog_path,
|
||
{
|
||
"schema_version": "agent_issue_catalog_v1",
|
||
"issues": {
|
||
"margin_domain_leak_accounting_route": {
|
||
"detectors": ["forbidden_margin_terms"],
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(
|
||
candidates_path,
|
||
{
|
||
"schema_version": "detector_candidates_v1",
|
||
"candidates": [
|
||
{
|
||
"issue_code": "margin_domain_leak_accounting_route",
|
||
"detector": "forbidden_margin_terms",
|
||
"evidence_paths": ["scenarios/ok/steps/s01/output.md"],
|
||
}
|
||
],
|
||
},
|
||
)
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
detector_candidates_path=candidates_path,
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "pass")
|
||
self.assertEqual(results["results"][0]["status"], "pass")
|
||
|
||
def test_required_any_detector_can_exclude_selected_object_followup_paths(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
write_text(
|
||
artifact_dir / "scenarios" / "profit_cashflow_boundary" / "steps" / "step_01_margin_root" / "output.md",
|
||
"Top margin item: revenue 100, cogs 60, gross profit 40, margin 40%.",
|
||
)
|
||
write_text(
|
||
artifact_dir
|
||
/ "scenarios"
|
||
/ "margin_followup"
|
||
/ "steps"
|
||
/ "step_03_selected_item_supplier"
|
||
/ "output.md",
|
||
"Supplier for the selected item: Example LLC.",
|
||
)
|
||
write_text(
|
||
artifact_dir
|
||
/ "scenarios"
|
||
/ "business_health"
|
||
/ "steps"
|
||
/ "step_05_next_checks"
|
||
/ "output.md",
|
||
"Check profit, debt, inventory, taxes, and customer concentration next.",
|
||
)
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"missing_revenue_cogs_margin_fields": {
|
||
"kind": "answer_text_required_any",
|
||
"automation_level": "semi_automatic",
|
||
"description": "Root margin answers need margin fields.",
|
||
"issue_codes": ["margin_domain_leak_accounting_route"],
|
||
"inputs": ["output.md"],
|
||
"check": {
|
||
"required_patterns_any": ["(?i)(revenue|cogs|gross profit|margin)"],
|
||
"artifact_path_include_patterns": ["(?i)steps[\\\\/][^\\\\/]*(?:margin|profitability)"],
|
||
"artifact_path_exclude_patterns": ["(?i)selected_item"],
|
||
},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
detector_names=["missing_revenue_cogs_margin_fields"],
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "pass")
|
||
self.assertEqual(results["results"][0]["status"], "pass")
|
||
evidence_paths = [item["path"] for item in results["results"][0]["evidence"]]
|
||
self.assertEqual(len(evidence_paths), 1)
|
||
self.assertIn("step_01_margin_root", evidence_paths[0])
|
||
|
||
def test_counterparty_value_flow_required_surface_scopes_to_current_net_step(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
write_text(
|
||
artifact_dir
|
||
/ "scenarios"
|
||
/ "mixed_planner_counterparty_evidence_and_living_guard"
|
||
/ "steps"
|
||
/ "step_03_incoming_by_resolved_entity"
|
||
/ "output.md",
|
||
"Входящие денежные поступления по контрагенту Группа СВК за 2020: 12 093 465 руб.",
|
||
)
|
||
write_text(
|
||
artifact_dir
|
||
/ "scenarios"
|
||
/ "mixed_planner_counterparty_evidence_and_living_guard"
|
||
/ "steps"
|
||
/ "step_05_net_after_payout"
|
||
/ "output.md",
|
||
"По контрагенту Группа СВК за период 2020 получили 12 093 465 руб., заплатили 0 руб.; расчетное нетто в нашу сторону: 12 093 465 руб.",
|
||
)
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"counterparty_value_flow_required_surface": {
|
||
"kind": "answer_text_required_any",
|
||
"automation_level": "automatic",
|
||
"description": "Net value-flow answer must surface all money directions.",
|
||
"issue_codes": ["counterparty_value_flow_misrouted_to_company_profit"],
|
||
"inputs": ["output.md"],
|
||
"check": {
|
||
"artifact_path_include_patterns": ["(?i)step_05_net_after_payout"],
|
||
"required_patterns_any": [
|
||
"(?is)(?=.*(СВК|Группа\\s+СВК))(?=.*(входящ|получил|получено|получили))(?=.*(исходящ|заплатил|заплачено|заплатили|ушло))(?=.*(нетто|сальдо|разниц|чистый\\s+денежный))"
|
||
],
|
||
},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
detector_names=["counterparty_value_flow_required_surface"],
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "pass")
|
||
self.assertEqual(results["results"][0]["status"], "pass")
|
||
evidence_paths = [item["path"] for item in results["results"][0]["evidence"]]
|
||
self.assertEqual(len(evidence_paths), 1)
|
||
self.assertIn("step_05_net_after_payout", evidence_paths[0])
|
||
|
||
def test_answer_text_shape_uses_business_first_step_review(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
step_dir = artifact_dir / "scenarios" / "mixed" / "steps" / "step_01"
|
||
write_text(step_dir / "output.md", "Direct business answer first.")
|
||
write_json(step_dir / "step_state.json", {"business_first_review": {"direct_answer_first_ok": True}})
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"first_line_not_direct_answer": {
|
||
"kind": "answer_text_shape",
|
||
"automation_level": "semi_automatic",
|
||
"description": "First line should be direct.",
|
||
"issue_codes": ["business_direct_answer_missing"],
|
||
"inputs": ["output.md"],
|
||
"check": {"first_line_should_be": "business_answer_or_honest_boundary"},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
detector_names=["first_line_not_direct_answer"],
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "pass")
|
||
self.assertEqual(results["results"][0]["status"], "pass")
|
||
|
||
def test_answer_text_shape_fails_when_business_first_review_rejects_first_line(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
step_dir = artifact_dir / "scenarios" / "mixed" / "steps" / "step_01"
|
||
write_text(step_dir / "output.md", "Let me inspect the route first.")
|
||
write_json(step_dir / "step_state.json", {"business_first_review": {"direct_answer_first_ok": False}})
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"first_line_not_direct_answer": {
|
||
"kind": "answer_text_shape",
|
||
"automation_level": "semi_automatic",
|
||
"description": "First line should be direct.",
|
||
"issue_codes": ["business_direct_answer_missing"],
|
||
"inputs": ["output.md"],
|
||
"check": {"first_line_should_be": "business_answer_or_honest_boundary"},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
detector_names=["first_line_not_direct_answer"],
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "fail")
|
||
self.assertEqual(results["results"][0]["status"], "fail")
|
||
|
||
def test_step_state_lookup_handles_long_artifact_paths(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
step_dir = (
|
||
root
|
||
/ "run"
|
||
/ "scenarios"
|
||
/ ("mixed_planner_counterparty_evidence_and_living_guard_" + "s" * 48)
|
||
/ "steps"
|
||
/ ("step_04_payout_switch_by_resolved_entity_" + "x" * 80)
|
||
)
|
||
output_path = step_dir / "output.md"
|
||
step_state_path = step_dir / "step_state.json"
|
||
try:
|
||
os.makedirs(runner.path_for_io(step_dir), exist_ok=True)
|
||
with open(runner.path_for_io(output_path), "w", encoding="utf-8") as handle:
|
||
handle.write("Direct business answer first.")
|
||
runner.write_json(step_state_path, {"business_first_review": {"direct_answer_first_ok": True}})
|
||
|
||
self.assertGreater(len(str(step_state_path.absolute())), 260)
|
||
self.assertEqual(runner.output_step_state_path(output_path), step_state_path)
|
||
self.assertEqual(
|
||
runner.read_json_object(step_state_path).get("business_first_review"),
|
||
{"direct_answer_first_ok": True},
|
||
)
|
||
finally:
|
||
for path in (output_path, step_state_path):
|
||
try:
|
||
os.remove(runner.path_for_io(path))
|
||
except OSError:
|
||
pass
|
||
current = step_dir
|
||
while current != root:
|
||
try:
|
||
os.rmdir(runner.path_for_io(current))
|
||
except OSError:
|
||
pass
|
||
current = current.parent
|
||
|
||
def test_composite_detector_fails_after_child_detector_fails(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
write_text(artifact_dir / "steps" / "s01" / "output.md", "Маржа почему-то посчитана через амортизацию.")
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"forbidden_margin_terms": {
|
||
"kind": "answer_text_regex_forbidden",
|
||
"automation_level": "automatic",
|
||
"description": "No wrong-domain words.",
|
||
"issue_codes": ["margin_domain_leak_accounting_route"],
|
||
"inputs": ["output.md"],
|
||
"check": {"forbidden_patterns": ["(?i)амортизац"]},
|
||
},
|
||
"margin_domain_leak_accounting_route": {
|
||
"kind": "composite_detector",
|
||
"automation_level": "semi_automatic",
|
||
"description": "Composite leak detector.",
|
||
"issue_codes": ["margin_domain_leak_accounting_route"],
|
||
"inputs": ["output.md"],
|
||
"check": {"uses_detectors": ["forbidden_margin_terms"]},
|
||
},
|
||
},
|
||
},
|
||
)
|
||
write_json(
|
||
issue_catalog_path,
|
||
{
|
||
"schema_version": "agent_issue_catalog_v1",
|
||
"issues": {
|
||
"margin_domain_leak_accounting_route": {
|
||
"detectors": ["margin_domain_leak_accounting_route"],
|
||
}
|
||
},
|
||
},
|
||
)
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
detector_names=["margin_domain_leak_accounting_route"],
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
statuses = {item["detector"]: item["status"] for item in results["results"]}
|
||
self.assertEqual(statuses["forbidden_margin_terms"], "fail")
|
||
self.assertEqual(statuses["margin_domain_leak_accounting_route"], "fail")
|
||
|
||
def test_forbidden_regex_uses_current_assistant_answer_from_turn_json(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
step_dir = artifact_dir / "steps" / "s01"
|
||
write_text(
|
||
step_dir / "output.md",
|
||
"# Assistant conversation export\n\n"
|
||
"## 1. user\n"
|
||
"Can you compute margin from payment_document evidence?\n\n"
|
||
"## 2. assistant\n"
|
||
"Clean current answer.\n",
|
||
)
|
||
write_json(
|
||
step_dir / "turn.json",
|
||
{
|
||
"user_message": {"text": "Can you compute margin from payment_document evidence?"},
|
||
"assistant_message": {"text": "Clean current answer.", "reply_type": "factual"},
|
||
"technical_debug_payload": {"detected_intent": "inventory_margin_ranking_for_nomenclature"},
|
||
},
|
||
)
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"forbidden_margin_terms": {
|
||
"kind": "answer_text_regex_forbidden",
|
||
"automation_level": "automatic",
|
||
"description": "No wrong-domain words in the current answer.",
|
||
"issue_codes": ["margin_domain_leak_accounting_route"],
|
||
"inputs": ["output.md"],
|
||
"check": {"forbidden_patterns": ["payment_document"]},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
detector_names=["forbidden_margin_terms"],
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "pass")
|
||
self.assertEqual(results["results"][0]["status"], "pass")
|
||
|
||
def test_answer_text_outputs_ignore_aggregate_scenario_export(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
write_text(
|
||
artifact_dir / "scenarios" / "inventory" / "scenario_output.md",
|
||
"# Technical export\n\ncapability_id: address.inventory.debug\nruntime_debug: true\n",
|
||
)
|
||
step_output = artifact_dir / "scenarios" / "inventory" / "steps" / "s01" / "output.md"
|
||
write_text(step_output, "Clean user-facing answer.")
|
||
|
||
outputs = runner.collect_output_artifacts(artifact_dir)
|
||
|
||
artifact_paths = [item["artifact_path"].replace("\\", "/") for item in outputs]
|
||
self.assertEqual(artifact_paths, ["scenarios/inventory/steps/s01/output.md"])
|
||
|
||
def test_trace_guard_uses_current_route_fields_not_full_turn_history(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
step_dir = artifact_dir / "steps" / "s01"
|
||
write_json(
|
||
step_dir / "turn.json",
|
||
{
|
||
"user_message": {"text": "Ignore fixed_asset and answer selected item document."},
|
||
"assistant_message": {"text": "Inventory document answer.", "reply_type": "factual"},
|
||
"technical_debug_payload": {
|
||
"detected_intent": "inventory_purchase_documents_for_item",
|
||
"selected_recipe": "address_inventory_purchase_documents_for_item_v1",
|
||
"capability_id": "inventory_inventory_purchase_documents_for_item",
|
||
},
|
||
},
|
||
)
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"wrong_capability_family": {
|
||
"kind": "trace_value_guard",
|
||
"automation_level": "automatic",
|
||
"description": "No wrong current route family.",
|
||
"issue_codes": ["margin_domain_leak_accounting_route"],
|
||
"inputs": ["turn.json"],
|
||
"check": {"forbidden_trace_markers": ["fixed_asset"]},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
detector_names=["wrong_capability_family"],
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "pass")
|
||
self.assertEqual(results["results"][0]["status"], "pass")
|
||
|
||
def test_trace_guard_fails_when_current_route_field_has_forbidden_marker(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "run"
|
||
step_dir = artifact_dir / "steps" / "s01"
|
||
write_json(
|
||
step_dir / "turn.json",
|
||
{
|
||
"assistant_message": {"text": "Wrong route answer.", "reply_type": "factual"},
|
||
"technical_debug_payload": {
|
||
"detected_intent": "fixed_asset_amortization",
|
||
"selected_recipe": "address_fixed_asset_amortization_v1",
|
||
},
|
||
},
|
||
)
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"wrong_capability_family": {
|
||
"kind": "trace_value_guard",
|
||
"automation_level": "automatic",
|
||
"description": "No wrong current route family.",
|
||
"issue_codes": ["margin_domain_leak_accounting_route"],
|
||
"inputs": ["turn.json"],
|
||
"check": {"forbidden_trace_markers": ["fixed_asset"]},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
detector_names=["wrong_capability_family"],
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "fail")
|
||
self.assertEqual(results["results"][0]["status"], "fail")
|
||
|
||
def test_stage_review_signal_detector_fails_when_issue_code_is_present(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "review"
|
||
write_json(
|
||
artifact_dir / "run_review.json",
|
||
{
|
||
"summary": {"overall_business_status": "fail"},
|
||
"findings": [{"issue_codes": ["backend_error_response"]}],
|
||
},
|
||
)
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"backend_error_response_signal": {
|
||
"kind": "stage_review_signal",
|
||
"automation_level": "automatic",
|
||
"description": "Backend error issue in GUI review.",
|
||
"issue_codes": ["backend_error_response"],
|
||
"inputs": ["run_review.json"],
|
||
"check": {"target_status": "backend_error_response"},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(
|
||
issue_catalog_path,
|
||
{
|
||
"schema_version": "agent_issue_catalog_v1",
|
||
"issues": {
|
||
"backend_error_response": {
|
||
"detectors": ["backend_error_response_signal"],
|
||
}
|
||
},
|
||
},
|
||
)
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
issue_codes=["backend_error_response"],
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "fail")
|
||
self.assertEqual(results["results"][0]["detector"], "backend_error_response_signal")
|
||
self.assertEqual(results["results"][0]["status"], "fail")
|
||
|
||
def test_stage_review_signal_detector_ignores_possible_invariant_names(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
artifact_dir = root / "review"
|
||
write_json(
|
||
artifact_dir / "run_review.json",
|
||
{
|
||
"summary": {"overall_business_status": "pass", "issue_counts": {}},
|
||
"findings": [],
|
||
"repair_targets": [],
|
||
"step_states": [
|
||
{
|
||
"business_first_review": {
|
||
"invariant_severity": {
|
||
"backend_error_response": "P0"
|
||
}
|
||
}
|
||
}
|
||
],
|
||
},
|
||
)
|
||
registry_path = root / "detector_registry.json"
|
||
issue_catalog_path = root / "issue_catalog.json"
|
||
write_json(
|
||
registry_path,
|
||
{
|
||
"schema_version": "agent_detector_registry_v1",
|
||
"detectors": {
|
||
"backend_error_response_signal": {
|
||
"kind": "stage_review_signal",
|
||
"automation_level": "automatic",
|
||
"description": "Backend error issue in GUI review.",
|
||
"issue_codes": ["backend_error_response"],
|
||
"inputs": ["run_review.json"],
|
||
"check": {"target_status": "backend_error_response"},
|
||
}
|
||
},
|
||
},
|
||
)
|
||
write_json(
|
||
issue_catalog_path,
|
||
{
|
||
"schema_version": "agent_issue_catalog_v1",
|
||
"issues": {
|
||
"backend_error_response": {
|
||
"detectors": ["backend_error_response_signal"],
|
||
}
|
||
},
|
||
},
|
||
)
|
||
|
||
results = runner.build_detector_results(
|
||
artifact_dir,
|
||
issue_codes=["backend_error_response"],
|
||
registry_path=registry_path,
|
||
issue_catalog_path=issue_catalog_path,
|
||
include_default_global=False,
|
||
)
|
||
|
||
self.assertEqual(results["summary"]["status"], "pass")
|
||
self.assertEqual(results["results"][0]["status"], "pass")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|