Add prompt-safe router state diagnostics wrapper

scripts/diagnose_router_state.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+import sys
+
+ROOT_DIR = Path(__file__).resolve().parents[1]
+if str(ROOT_DIR) not in sys.path:
+    sys.path.insert(0, str(ROOT_DIR))
+
+try:
+    from router.config import load_settings
+    from scripts.check_route_error_budget import BudgetConfig, check_budget, format_budget_result
+    from scripts.router_log_summary import (
+        ParseDiagnostics,
+        format_summary,
+        parse_route_records,
+        summarize_records,
+    )
+except ModuleNotFoundError:
+    from router.config import load_settings
+    from check_route_error_budget import BudgetConfig, check_budget, format_budget_result
+    from router_log_summary import (
+        ParseDiagnostics,
+        format_summary,
+        parse_route_records,
+        summarize_records,
+    )
+
+
+def render_config_section(routes_path: str) -> str:
+    settings = load_settings(routes_path)
+    route_targets = {
+        route_id: route_spec.target_model or route_id
+        for route_id, route_spec in sorted(settings.routes.items())
+    }
+    hard_route_ids = sorted({rule.route_id for rule in settings.hard_rules})
+
+    lines = [
+        "[router_config]",
+        f"entry_model: {settings.entry_model}",
+        f"fallback_route_id: {settings.fallback_route_id}",
+        "route_targets:",
+    ]
+    lines.extend([f"  {route_id}: {target}" for route_id, target in route_targets.items()])
+    lines.append(
+        "hard_rule_route_ids: "
+        + (", ".join(hard_route_ids) if hard_route_ids else "none")
+    )
+    return "\n".join(lines)
+
+
+def render_logs_sections(logs_path: str) -> str:
+    diagnostics = ParseDiagnostics()
+    records = list(parse_route_records(Path(logs_path).read_text(encoding="utf-8").splitlines(), diagnostics=diagnostics))
+    route_summary = summarize_records(records, parse_diagnostics=diagnostics)
+    budget_result = check_budget(
+        records,
+        BudgetConfig(
+            min_total=0,
+            max_error_rate=1.0,
+            max_target_error_rate=1.0,
+            max_route_error_rate=1.0,
+        ),
+        parse_diagnostics=diagnostics,
+    )
+    return "\n".join(
+        [
+            "[route_summary]",
+            format_summary(route_summary),
+            "[route_error_budget]",
+            format_budget_result(budget_result),
+        ]
+    )
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Print a prompt-safe router diagnostics summary from config and logs."
+    )
+    parser.add_argument("--routes", required=True, help="Path to config/routes.yaml")
+    parser.add_argument("--logs", help="Optional path to structured router log file")
+    args = parser.parse_args(argv)
+
+    sections = [render_config_section(args.routes)]
+    if args.logs:
+        sections.append(render_logs_sections(args.logs))
+
+    sys.stdout.write("\n\n".join(sections) + "\n")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_diagnose_router_state.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import subprocess
+import sys
+
+from scripts.diagnose_router_state import main
+
+
+def test_cli_emits_stable_config_section(capsys):
+    exit_code = main(["--routes", "config/routes.yaml"])
+
+    output = capsys.readouterr().out
+    assert exit_code == 0
+    assert "[router_config]" in output
+    assert "entry_model: semantic-router" in output
+    assert "fallback_route_id: fast" in output
+    assert "route_targets:" in output
+    assert "  fast: cheap-router" in output
+    assert "  strong: pro-router" in output
+    assert "hard_rule_route_ids: strong" in output
+
+
+def test_cli_with_logs_includes_summary_and_budget_and_redacts_payload(tmp_path, capsys):
+    logs_path = tmp_path / "router.ndjson"
+    logs_path.write_text(
+        "\n".join(
+            [
+                '{"event":"route_complete","route_id":"fast","target_model":"cheap-router","reason":"embedding","stream":false}',
+                '{"event":"route_error","route_id":"strong","target_model":"pro-router","reason":"hard_rule:debug","stream":true,"error_type":"TimeoutError","upstream_status":503,"messages":"secret-prompt","authorization":"Bearer abc"}',
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    exit_code = main(["--routes", "config/routes.yaml", "--logs", str(logs_path)])
+
+    output = capsys.readouterr().out
+    assert exit_code == 0
+    assert "[route_summary]" in output
+    assert "[route_error_budget]" in output
+    assert "routes: fast=1, strong=1" in output
+    assert "error_types: TimeoutError=1" in output
+    assert "secret-prompt" not in output
+    assert "Bearer abc" not in output
+    assert "authorization" not in output.lower()
+    assert "messages" not in output.lower()
+
+
+def test_script_file_execution_from_repo_root():
+    completed = subprocess.run(
+        [
+            sys.executable,
+            "scripts/diagnose_router_state.py",
+            "--routes",
+            "config/routes.yaml",
+            "--logs",
+            "tests/samples/router_logs.ndjson",
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert completed.returncode == 0
+    assert "[router_config]" in completed.stdout
+    assert "[route_summary]" in completed.stdout
+    assert "[route_error_budget]" in completed.stdout