#!/usr/bin/env python3 """Validate and atomically write the Copilot diagnosis routing artifact.""" from __future__ import annotations import argparse import json import sys from pathlib import Path from typing import Any ROUTING_PATH = Path("evidences") / "agent-runs" / "diagnosis-routing.json" STANDARD_ROUTES = { "sql_plan_or_query", "transaction_or_lock_contention", "tikv_hotspot_or_coprocessor", "storage_or_io_bottleneck", "memory_or_oom_pressure", "gc_or_allocator_pressure", "resource_control_or_capacity", "pd_tso_or_scheduling", "cdc_or_replication_lag", "network_or_connectivity", "component_failure_or_bug", } ALLOWED_TOP_LEVEL_FIELDS = { "schema_version", "status", "scope", "candidates", "recommended_route", "selection", } ALLOWED_CANDIDATE_FIELDS = { "route", "label", "routing_confidence", "evidence_for", "evidence_against", } def _fail(message: str) -> None: raise ValueError(message) def _non_empty_string(value: Any, field: str) -> str: if not isinstance(value, str) or not value.strip(): _fail(f"{field} must be a non-empty string") return value.strip() def _standard_route(value: Any, field: str) -> str: route = _non_empty_string(value, field) if route not in STANDARD_ROUTES: _fail( f"{field} must use the standard taxonomy; got {route!r}. " f"Allowed routes: {', '.join(sorted(STANDARD_ROUTES))}" ) return route def _string_list(value: Any, field: str) -> list[str]: if not isinstance(value, list) or any(not isinstance(item, str) for item in value): _fail(f"{field} must be an array of strings") return value def validate_routing(payload: Any) -> dict[str, Any]: if not isinstance(payload, dict): _fail("routing artifact must be a JSON object") unknown = sorted(set(payload) - ALLOWED_TOP_LEVEL_FIELDS) if unknown: _fail(f"unknown routing field(s): {', '.join(unknown)}") missing = sorted(ALLOWED_TOP_LEVEL_FIELDS - set(payload)) if missing: _fail(f"missing routing field(s): {', '.join(missing)}") if payload.get("schema_version") != "2.0": _fail("schema_version must be exactly '2.0'") status = payload.get("status") if status not in {"awaiting_user", "selected"}: _fail("status must be 'awaiting_user' or 'selected'") scope = payload.get("scope") if not isinstance(scope, dict): _fail("scope must be an object") _non_empty_string(scope.get("cluster_id"), "scope.cluster_id") time_range = scope.get("time_range") if not isinstance(time_range, dict): _fail("scope.time_range must be an object") _non_empty_string(time_range.get("start"), "scope.time_range.start") _non_empty_string(time_range.get("end"), "scope.time_range.end") candidates = payload.get("candidates") if not isinstance(candidates, list) or not 1 <= len(candidates) <= 3: _fail("candidates must contain one to three entries") candidate_routes: list[str] = [] for index, candidate in enumerate(candidates): field = f"candidates[{index}]" if not isinstance(candidate, dict): _fail(f"{field} must be an object") unknown_candidate = sorted(set(candidate) - ALLOWED_CANDIDATE_FIELDS) if unknown_candidate: _fail(f"{field} has unknown field(s): {', '.join(unknown_candidate)}") missing_candidate = sorted(ALLOWED_CANDIDATE_FIELDS - set(candidate)) if missing_candidate: _fail(f"{field} is missing field(s): {', '.join(missing_candidate)}") candidate_routes.append(_standard_route(candidate.get("route"), f"{field}.route")) _non_empty_string(candidate.get("label"), f"{field}.label") confidence = candidate.get("routing_confidence") if not isinstance(confidence, (int, float)) or isinstance(confidence, bool) or not 0 <= confidence <= 1: _fail(f"{field}.routing_confidence must be between 0 and 1") _string_list(candidate.get("evidence_for"), f"{field}.evidence_for") _string_list(candidate.get("evidence_against"), f"{field}.evidence_against") if len(set(candidate_routes)) != len(candidate_routes): _fail("candidate routes must be unique") recommended = _standard_route(payload.get("recommended_route"), "recommended_route") if recommended not in candidate_routes: _fail("recommended_route must be one of the candidate routes") selection = payload.get("selection") if status == "awaiting_user": if selection is not None: _fail("selection must be null while status is 'awaiting_user'") else: if not isinstance(selection, dict): _fail("selection must be an object while status is 'selected'") if set(selection) != {"routes", "source"}: _fail("selection must contain exactly routes and source") routes = selection.get("routes") if not isinstance(routes, list) or not 1 <= len(routes) <= 3: _fail("selection.routes must contain one to three routes") normalized_routes = [ _standard_route(route, f"selection.routes[{index}]") for index, route in enumerate(routes) ] if len(set(normalized_routes)) != len(normalized_routes): _fail("selection.routes must be unique") if selection.get("source") not in { "user", "preferred_route", "auto_continue", "feedback", }: _fail("selection.source is invalid") return payload def _load(path: Path) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except OSError as exc: _fail(f"cannot read {path}: {exc}") except json.JSONDecodeError as exc: _fail(f"{path} is not valid JSON: {exc}") def _write_atomic(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text( json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) temporary.replace(path) def main() -> int: parser = argparse.ArgumentParser( description="Validate and write evidences/agent-runs/diagnosis-routing.json" ) parser.add_argument( "--from-stdin", action="store_true", help="Read the complete routing JSON from stdin and atomically replace the artifact.", ) args = parser.parse_args() try: if args.from_stdin: payload = json.loads(sys.stdin.read()) validate_routing(payload) _write_atomic(ROUTING_PATH, payload) print(f"diagnosis-routing.json written: status={payload['status']}") else: payload = _load(ROUTING_PATH) validate_routing(payload) print(f"diagnosis-routing.json valid: status={payload['status']}") except (ValueError, json.JSONDecodeError) as exc: print(f"error: {exc}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())