from __future__ import annotations import argparse import json import pathlib import re import sys from typing import Any RESULT_KEYS = ("results", "cases", "records", "items", "diff_table") SUPPORTED_SUFFIXES = {".json", ".jsonl", ".md", ".markdown"} NO_CALL_VALUES = {"", "none", "null", "no_call", "no-call", "no call", "nocall", "no decision"} def main(argv: list[str] | None = None) -> int: args = parse_args(argv) targeted_cases, targeted_errors = load_targeted_cases(args.targeted_case, args.targeted_cases_file) before_cases, before_errors = load_case_map( args.before_results, args.case_id_field, args.route_field, args.expected_route_field, "before", ) after_cases, after_errors = load_case_map( args.after_results, args.case_id_field, args.route_field, args.expected_route_field, "after", ) blocking_errors = [*targeted_errors, *before_errors, *after_errors] if blocking_errors: summary = empty_summary(blocking_errors) write_reports(args.output_dir, summary) print(f"routing-behavior-diff-audit: {len(blocking_errors)} blocking error(s)", file=sys.stderr) return 1 summary = compare_cases(before_cases, after_cases, targeted_cases) write_reports(args.output_dir, summary) print(str(args.output_dir / "routing-behavior-diff.json")) return 0 def parse_args(argv: list[str] | None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Compare before/after rules-based routing outputs and write deterministic diff reports.", ) parser.add_argument("--before-results", required=True, type=pathlib.Path) parser.add_argument("--after-results", required=True, type=pathlib.Path) parser.add_argument("--output-dir", required=True, type=pathlib.Path) parser.add_argument("--case-id-field", default="case_id") parser.add_argument("--route-field", default="route") parser.add_argument("--expected-route-field", default=None) parser.add_argument("--targeted-case", action="append", default=[]) parser.add_argument("--targeted-cases-file", type=pathlib.Path) return parser.parse_args(argv) def load_targeted_cases(raw_cases: list[str], cases_file: pathlib.Path | None) -> tuple[set[str], list[str]]: targeted = {str(case_id) for case_id in raw_cases} errors: list[str] = [] if not cases_file: return targeted, errors if not cases_file.exists(): return targeted, [f"Targeted cases file not found: {cases_file}"] try: loaded = parse_file(cases_file) except ValueError as exc: return targeted, [f"Unable to parse targeted cases file {cases_file}: {exc}"] candidate: Any = loaded if isinstance(candidate, dict): for key in ("targeted_cases", "case_ids", "cases"): if key in candidate: candidate = candidate[key] break if not isinstance(candidate, list): return targeted, [f"Targeted cases file must contain a list or object with targeted_cases/case_ids/cases: {cases_file}"] for item in candidate: if isinstance(item, dict): for key in ("case_id", "id"): if key in item: targeted.add(str(item[key])) break else: targeted.add(str(item)) return targeted, errors def load_case_map( root: pathlib.Path, case_id_field: str, route_field: str, expected_route_field: str | None, side: str, ) -> tuple[dict[str, dict[str, Any]], list[str]]: errors: list[str] = [] if not root.exists(): return {}, [f"{side} results path not found: {root}"] files = collect_input_files(root) if not files: return {}, [f"{side} results path contains no supported files: {root}"] cases: dict[str, dict[str, Any]] = {} for path in files: try: parsed = parse_file(path) except ValueError as exc: errors.append(f"Unable to parse {side} file {path}: {exc}") continue records = extract_records(parsed) if not records: errors.append(f"No records found in {side} file: {path}") continue for record in records: if not isinstance(record, dict): errors.append(f"Record in {side} file is not an object: {path}") continue if case_id_field not in record: errors.append(f"Record missing case id field '{case_id_field}' in {side} file: {path}") continue case_id = stable_string(record[case_id_field]) if case_id in cases: errors.append(f"Duplicate case id '{case_id}' in {side} inputs") continue cases[case_id] = { "case_id": case_id, "route": normalize_value(record.get(route_field)), "expected_route": normalize_value(record.get(expected_route_field)) if expected_route_field else None, "source_file": str(path), } return cases, errors def collect_input_files(root: pathlib.Path) -> list[pathlib.Path]: if root.is_file(): return [root] if root.suffix.lower() in SUPPORTED_SUFFIXES else [] return sorted(path for path in root.rglob("*") if path.is_file() and path.suffix.lower() in SUPPORTED_SUFFIXES) def parse_file(path: pathlib.Path) -> Any: suffix = path.suffix.lower() text = path.read_text(encoding="utf-8") if suffix == ".json": return json.loads(text) if suffix == ".jsonl": records = [] for line_number, line in enumerate(text.splitlines(), start=1): stripped = line.strip() if not stripped: continue try: records.append(json.loads(stripped)) except json.JSONDecodeError as exc: raise ValueError(f"invalid JSONL on line {line_number}: {exc}") from exc return records if suffix in {".md", ".markdown"}: return parse_markdown_summary(text, path) raise ValueError(f"unsupported suffix: {suffix}") def parse_markdown_summary(text: str, path: pathlib.Path) -> Any: fences = re.findall(r"```(?:json|yaml|yml)?\s*\n(.*?)\n```", text, flags=re.DOTALL | re.IGNORECASE) errors: list[str] = [] for fence in fences: try: return json.loads(fence) except json.JSONDecodeError as exc: errors.append(f"json: {exc}") yaml_loaded = try_load_yaml(fence) if yaml_loaded is not None: return yaml_loaded raise ValueError(f"no parseable JSON/YAML fenced block found in Markdown summary {path}; {'; '.join(errors)}") def try_load_yaml(text: str) -> Any: try: import yaml # type: ignore[import-not-found] except ImportError: return None try: return yaml.safe_load(text) except Exception: return None def extract_records(parsed: Any) -> list[Any]: if isinstance(parsed, list): return parsed if isinstance(parsed, dict): for key in RESULT_KEYS: value = parsed.get(key) if isinstance(value, list): return value if "case_id" in parsed or "id" in parsed: return [parsed] return [] def compare_cases(before_cases: dict[str, dict[str, Any]], after_cases: dict[str, dict[str, Any]], targeted_cases: set[str]) -> dict[str, Any]: before_ids = set(before_cases) after_ids = set(after_cases) common_ids = sorted(before_ids & after_ids) missing_before = sorted(after_ids - before_ids) missing_after = sorted(before_ids - after_ids) unchanged: list[str] = [] changed: list[str] = [] targeted_changes: list[str] = [] non_target_changes: list[str] = [] new_failures: list[str] = [] resolved_failures: list[str] = [] diff_table: list[dict[str, Any]] = [] for case_id in common_ids: before = before_cases[case_id] after = after_cases[case_id] before_failure = is_failure(before) after_failure = is_failure(after) route_changed = before["route"] != after["route"] expected_changed = before["expected_route"] != after["expected_route"] failure_status_changed = before_failure != after_failure no_call_changed = is_no_call(before["route"]) != is_no_call(after["route"]) row_changed = route_changed or expected_changed or failure_status_changed or no_call_changed change_type = build_change_type(route_changed, expected_changed, failure_status_changed, no_call_changed) if row_changed: changed.append(case_id) if case_id in targeted_cases: targeted_changes.append(case_id) else: non_target_changes.append(case_id) else: unchanged.append(case_id) if not before_failure and after_failure: new_failures.append(case_id) if before_failure and not after_failure: resolved_failures.append(case_id) diff_table.append( { "case_id": case_id, "change_type": change_type, "targeted": case_id in targeted_cases, "before_route": before["route"], "after_route": after["route"], "before_expected_route": before["expected_route"], "after_expected_route": after["expected_route"], "route_changed": route_changed, "expected_route_changed": expected_changed, "before_failure": before_failure, "after_failure": after_failure, "failure_status_changed": failure_status_changed, "no_call_changed": no_call_changed, "before_source_file": before["source_file"], "after_source_file": after["source_file"], } ) for case_id in missing_before: after = after_cases[case_id] diff_table.append( { "case_id": case_id, "change_type": "missing_before", "targeted": case_id in targeted_cases, "before_route": None, "after_route": after["route"], "before_expected_route": None, "after_expected_route": after["expected_route"], "route_changed": True, "expected_route_changed": after["expected_route"] is not None, "before_failure": None, "after_failure": is_failure(after), "failure_status_changed": None, "no_call_changed": None, "before_source_file": None, "after_source_file": after["source_file"], } ) for case_id in missing_after: before = before_cases[case_id] diff_table.append( { "case_id": case_id, "change_type": "missing_after", "targeted": case_id in targeted_cases, "before_route": before["route"], "after_route": None, "before_expected_route": before["expected_route"], "after_expected_route": None, "route_changed": True, "expected_route_changed": before["expected_route"] is not None, "before_failure": is_failure(before), "after_failure": None, "failure_status_changed": None, "no_call_changed": None, "before_source_file": before["source_file"], "after_source_file": None, } ) status = "ATTENTION" if non_target_changes or missing_before or missing_after or new_failures else "PASS" return { "total_cases_compared": len(common_ids), "unchanged_cases": unchanged, "changed_cases": changed, "targeted_changes": targeted_changes, "non_target_changes": non_target_changes, "new_failures": new_failures, "resolved_failures": resolved_failures, "missing_before_cases": missing_before, "missing_after_cases": missing_after, "diff_table": sorted(diff_table, key=lambda row: row["case_id"]), "blocking_errors": [], "machine_readable_summary": { "status": status, "total_cases_compared": len(common_ids), "changed_count": len(changed), "targeted_change_count": len(targeted_changes), "non_target_change_count": len(non_target_changes), "new_failure_count": len(new_failures), "resolved_failure_count": len(resolved_failures), "missing_before_count": len(missing_before), "missing_after_count": len(missing_after), }, } def build_change_type(route_changed: bool, expected_changed: bool, failure_status_changed: bool, no_call_changed: bool) -> str: labels: list[str] = [] if route_changed: labels.append("route_changed") if expected_changed: labels.append("expected_route_changed") if failure_status_changed: labels.append("failure_status_changed") if no_call_changed: labels.append("no_call_changed") return "+".join(labels) if labels else "unchanged" def is_failure(case: dict[str, Any]) -> bool: expected = case.get("expected_route") if expected is None: return False return case.get("route") != expected def is_no_call(value: Any) -> bool: if value is None: return True return str(value).strip().casefold() in NO_CALL_VALUES def normalize_value(value: Any) -> str | None: if value is None: return None return stable_string(value) def stable_string(value: Any) -> str: if isinstance(value, str): return value return json.dumps(value, ensure_ascii=False, sort_keys=True) def empty_summary(blocking_errors: list[str]) -> dict[str, Any]: return { "total_cases_compared": 0, "unchanged_cases": [], "changed_cases": [], "targeted_changes": [], "non_target_changes": [], "new_failures": [], "resolved_failures": [], "missing_before_cases": [], "missing_after_cases": [], "diff_table": [], "blocking_errors": blocking_errors, "machine_readable_summary": { "status": "FAIL", "blocking_error_count": len(blocking_errors), "total_cases_compared": 0, "changed_count": 0, "targeted_change_count": 0, "non_target_change_count": 0, "new_failure_count": 0, "resolved_failure_count": 0, "missing_before_count": 0, "missing_after_count": 0, }, } def write_reports(output_dir: pathlib.Path, summary: dict[str, Any]) -> None: output_dir.mkdir(parents=True, exist_ok=True) (output_dir / "routing-behavior-diff.json").write_text( json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) (output_dir / "routing-behavior-diff.md").write_text(render_markdown(summary), encoding="utf-8") def render_markdown(summary: dict[str, Any]) -> str: machine = summary["machine_readable_summary"] lines = [ "# Routing Behavior Diff", "", f"Status: `{machine['status']}`", "", "## Summary", "", f"- Total cases compared: {summary['total_cases_compared']}", f"- Unchanged cases: {len(summary['unchanged_cases'])}", f"- Changed cases: {len(summary['changed_cases'])}", f"- Targeted changes: {len(summary['targeted_changes'])}", f"- Non-target changes: {len(summary['non_target_changes'])}", f"- New failures: {len(summary['new_failures'])}", f"- Resolved failures: {len(summary['resolved_failures'])}", f"- Missing before cases: {len(summary['missing_before_cases'])}", f"- Missing after cases: {len(summary['missing_after_cases'])}", "", ] if summary["blocking_errors"]: lines.extend(["## Blocking Errors", ""]) lines.extend(f"- {error}" for error in summary["blocking_errors"]) lines.append("") lines.extend( [ "## Diff Table", "", "| Case ID | Change Type | Targeted | Before Route | After Route | Before Failure | After Failure | No-Call Changed |", "| --- | --- | --- | --- | --- | --- | --- | --- |", ] ) if summary["diff_table"]: for row in summary["diff_table"]: lines.append( "| " + " | ".join( markdown_cell(row[key]) for key in ( "case_id", "change_type", "targeted", "before_route", "after_route", "before_failure", "after_failure", "no_call_changed", ) ) + " |" ) else: lines.append("| _none_ | _none_ | _none_ | _none_ | _none_ | _none_ | _none_ | _none_ |") lines.extend( [ "", "## Machine Readable Summary", "", "```json", json.dumps(machine, ensure_ascii=False, indent=2), "```", "", ] ) return "\n".join(lines) def markdown_cell(value: Any) -> str: if value is None: return "" return str(value).replace("|", "\\|").replace("\n", " ") if __name__ == "__main__": raise SystemExit(main())