504 lines
22 KiB
Python
504 lines
22 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
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")
|
||
|
||
|
||
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_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_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()
|