SILVER/NODEDC_1C
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Добавить реестр детекторов надежности агента

Browse Source
This commit is contained in:
dctouch
2026-05-24 14:17:16 +03:00
parent d29d885399
commit 34e854f030
4 changed files with 471 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/agent_reliability_contract_healthcheck.py
+122 -2
View File
@@ -11,8 +11,11 @@ from typing import Any
REPO_ROOT = Path(__file__).resolve().parent.parent
SCHEMA_DIR = REPO_ROOT / "docs" / "orchestration" / "schemas"
ISSUE_CATALOG_PATH = REPO_ROOT / "docs" / "orchestration" / "issue_catalog.json"
DETECTOR_REGISTRY_PATH = REPO_ROOT / "docs" / "orchestration" / "detector_registry.json"
CONTRACTS_DIR = REPO_ROOT / "docs" / "orchestration" / "contracts"
EXPECTED_SCHEMA_FILES = {
"agent_issue_catalog.schema.json": "Agent Issue Catalog",
"agent_detector_registry.schema.json": "Agent Detector Registry",
"auto_coder_gate.schema.json": "Auto-Coder Gate",
"business_audit_contract.schema.json": "Business Audit Contract",
"domain_loop_lead_coder_handoff.schema.json": "Domain Loop Lead Coder Handoff",
@@ -68,6 +71,43 @@ def has_answer_contract(issue: dict[str, Any]) -> bool:
return bool(normalize_string_list(acceptance.get("must_have")) or normalize_string_list(acceptance.get("must_not_have")))
def read_json_object_or_empty(path: Path) -> dict[str, Any]:
try:
payload = read_json(path)
except (json.JSONDecodeError, OSError):
return {}
return payload if isinstance(payload, dict) else {}
def collect_issue_detector_links(issue_catalog: dict[str, Any]) -> dict[str, set[str]]:
issues = issue_catalog.get("issues") if isinstance(issue_catalog.get("issues"), dict) else {}
links: dict[str, set[str]] = {}
for issue_code, issue in issues.items():
if not isinstance(issue, dict):
continue
for detector in normalize_string_list(issue.get("detectors")):
links.setdefault(detector, set()).add(str(issue_code))
return links
def collect_contract_detector_refs(contracts_dir: Path) -> tuple[dict[str, list[str]], list[str]]:
refs: dict[str, list[str]] = {}
warnings: list[str] = []
if not contracts_dir.exists():
return refs, warnings
for path in sorted(contracts_dir.glob("*.json")):
try:
payload = read_json(path)
except json.JSONDecodeError as error:
warnings.append(f"contract_detector_scan_invalid_json:{display_path(path)}:{error.msg}")
continue
if not isinstance(payload, dict):
continue
for detector in normalize_string_list(payload.get("detectors")):
refs.setdefault(detector, []).append(display_path(path))
return refs, warnings
def is_broad_patch_target(value: str) -> bool:
normalized = value.strip().replace("\\", "/").lower()
broad_targets = {
@@ -192,17 +232,97 @@ def check_issue_catalog(path: Path) -> tuple[dict[str, Any], list[str], list[str
return summary, failures, warnings
def check_detector_registry(
path: Path,
issue_catalog: dict[str, Any] | None = None,
*,
include_contracts: bool = True,
) -> tuple[dict[str, Any], list[str], list[str]]:
failures: list[str] = []
warnings: list[str] = []
if not path.exists():
return {"path": display_path(path), "exists": False}, ["missing_detector_registry"], warnings
try:
payload = read_json(path)
except json.JSONDecodeError as error:
return {"path": display_path(path), "exists": True}, [f"invalid_detector_registry_json:{error.msg}"], warnings
detectors = payload.get("detectors") if isinstance(payload.get("detectors"), dict) else {}
catalog = issue_catalog if isinstance(issue_catalog, dict) else {}
issues = catalog.get("issues") if isinstance(catalog.get("issues"), dict) else {}
known_issue_codes = set(str(issue_code) for issue_code in issues)
detector_links = collect_issue_detector_links(catalog)
contract_refs, contract_warnings = collect_contract_detector_refs(CONTRACTS_DIR) if include_contracts else ({}, [])
warnings.extend(contract_warnings)
summary = {
"path": display_path(path),
"exists": True,
"schema_version": payload.get("schema_version"),
"detector_count": len(detectors),
"catalog_referenced_detector_count": len(detector_links),
"contract_referenced_detector_count": len(contract_refs),
}
if payload.get("schema_version") != "agent_detector_registry_v1":
failures.append("detector_registry_schema_version_mismatch")
if not detectors:
failures.append("detector_registry_empty")
for detector_name, issue_codes in sorted(detector_links.items()):
if detector_name not in detectors:
for issue_code in sorted(issue_codes):
failures.append(f"detector_registry_missing_catalog_detector:{issue_code}:{detector_name}")
for detector_name, paths in sorted(contract_refs.items()):
if detector_name not in detectors:
for contract_path in paths:
failures.append(f"detector_registry_missing_contract_detector:{contract_path}:{detector_name}")
for detector_name, detector in sorted(detectors.items()):
if not isinstance(detector, dict):
failures.append(f"detector_registry_detector_not_object:{detector_name}")
continue
for field_name in ("kind", "automation_level", "description"):
if not str(detector.get(field_name) or "").strip():
failures.append(f"detector_registry_missing_field:{detector_name}:{field_name}")
issue_codes = normalize_string_list(detector.get("issue_codes"))
inputs = normalize_string_list(detector.get("inputs"))
check = detector.get("check")
if not issue_codes:
failures.append(f"detector_registry_empty_issue_codes:{detector_name}")
if not inputs:
failures.append(f"detector_registry_empty_inputs:{detector_name}")
if not isinstance(check, dict) or not check:
failures.append(f"detector_registry_empty_check:{detector_name}")
if known_issue_codes:
for issue_code in issue_codes:
if issue_code not in known_issue_codes:
failures.append(f"detector_registry_unknown_issue_code:{detector_name}:{issue_code}")
for issue_code in sorted(detector_links.get(detector_name, set())):
if issue_code not in issue_codes:
failures.append(f"detector_registry_missing_issue_link:{detector_name}:{issue_code}")
if not isinstance(catalog, dict) or not issues:
warnings.append("detector_registry_issue_catalog_unavailable")
return summary, failures, warnings
def build_healthcheck() -> dict[str, Any]:
schema_files, schema_failures = check_schema_files(SCHEMA_DIR)
issue_catalog, catalog_failures, catalog_warnings = check_issue_catalog(ISSUE_CATALOG_PATH)
failures = schema_failures + catalog_failures
warnings = catalog_warnings
issue_catalog_payload = read_json_object_or_empty(ISSUE_CATALOG_PATH)
detector_registry, detector_failures, detector_warnings = check_detector_registry(
DETECTOR_REGISTRY_PATH,
issue_catalog_payload,
)
failures = schema_failures + catalog_failures + detector_failures
warnings = catalog_warnings + detector_warnings
return {
"schema_version": "agent_reliability_contract_health_v1",
"status": "pass" if not failures else "fail",
"checked_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
"schema_files": schema_files,
"issue_catalog": issue_catalog,
"detector_registry": detector_registry,
"failures": failures,
"warnings": warnings,
}
scripts/test_agent_reliability_contract_healthcheck.py
+86
View File
@@ -130,6 +130,92 @@ class AgentReliabilityContractHealthcheckTests(unittest.TestCase):
failures,
)
def test_detector_registry_blocks_missing_catalog_detector(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
registry_path = Path(tmp) / "detector_registry.json"
registry_path.write_text(
json.dumps(
{
"schema_version": "agent_detector_registry_v1",
"detectors": {
"known_detector": {
"kind": "answer_text_shape",
"automation_level": "semi_automatic",
"description": "Known detector.",
"issue_codes": ["business_direct_answer_missing"],
"inputs": ["output.md"],
"check": {"first_line_should_be": "business_answer"},
}
},
}
),
encoding="utf-8",
)
issue_catalog = {
"schema_version": "agent_issue_catalog_v1",
"issues": {
"business_direct_answer_missing": {
"detectors": ["missing_detector"],
}
},
}
_, failures, _ = health.check_detector_registry(
registry_path,
issue_catalog,
include_contracts=False,
)
self.assertIn(
"detector_registry_missing_catalog_detector:business_direct_answer_missing:missing_detector",
failures,
)
def test_detector_registry_blocks_unknown_issue_link(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
registry_path = Path(tmp) / "detector_registry.json"
registry_path.write_text(
json.dumps(
{
"schema_version": "agent_detector_registry_v1",
"detectors": {
"first_line_not_direct_answer": {
"kind": "answer_text_shape",
"automation_level": "semi_automatic",
"description": "Direct answer detector.",
"issue_codes": ["unknown_issue_code"],
"inputs": ["output.md"],
"check": {"first_line_should_be": "business_answer"},
}
},
}
),
encoding="utf-8",
)
issue_catalog = {
"schema_version": "agent_issue_catalog_v1",
"issues": {
"business_direct_answer_missing": {
"detectors": ["first_line_not_direct_answer"],
}
},
}
_, failures, _ = health.check_detector_registry(
registry_path,
issue_catalog,
include_contracts=False,
)
self.assertIn(
"detector_registry_unknown_issue_code:first_line_not_direct_answer:unknown_issue_code",
failures,
)
self.assertIn(
"detector_registry_missing_issue_link:first_line_not_direct_answer:business_direct_answer_missing",
failures,
)
if __name__ == "__main__":
unittest.main()