fix(robustness): harden exception boundaries

scripts/verify_exception_boundary_policy.py

@@ -12,6 +12,7 @@ import ast
 import json
 from collections import Counter
 from dataclasses import dataclass
+from datetime import date
 from pathlib import Path
 from typing import Any, Iterable
 
@@ -20,6 +21,7 @@ VALID_CLASSIFICATIONS = {
     "needs_narrowing",
     "needs_follow_up_test_coverage",
 }
+VALID_COVERAGE_MODES = {"all_broad_catches", "selected_scopes"}
 
 
 @dataclass(frozen=True)
@@ -93,11 +95,26 @@ def validate_exception_boundary_policy(
     policy: dict[str, Any],
 ) -> list[str]:
     failures: list[str] = []
+    if set(policy) != {"version", "selected_modules"}:
+        failures.append("policy root keys must match the version 2 schema")
+    if policy.get("version") != 2:
+        failures.append("policy version must equal 2")
     modules = policy.get("selected_modules")
     if not isinstance(modules, dict) or not modules:
         return ["policy selected_modules must be a non-empty object"]
 
     for rel_path, module_policy in sorted(modules.items()):
+        rel_path_obj = Path(rel_path)
+        if (
+            rel_path_obj.is_absolute()
+            or ".." in rel_path_obj.parts
+            or rel_path_obj.suffix != ".py"
+        ):
+            failures.append(f"{rel_path}: unsafe selected module path")
+            continue
+        if not isinstance(module_policy, dict):
+            failures.append(f"{rel_path}: module policy must be an object")
+            continue
         path = repo_root / rel_path
         if not path.is_file():
             failures.append(f"{rel_path}: selected module does not exist")
@@ -108,14 +125,60 @@ def validate_exception_boundary_policy(
             failures.append(f"{rel_path}: broad_catches must be a list")
             continue
 
+        coverage = module_policy.get("coverage")
+        if coverage not in VALID_COVERAGE_MODES:
+            failures.append(f"{rel_path}: invalid coverage mode {coverage!r}")
+            continue
+        expected_module_keys = {"coverage", "broad_catches"}
+        if coverage == "selected_scopes":
+            expected_module_keys.add("selected_scopes")
+        if set(module_policy) != expected_module_keys:
+            failures.append(f"{rel_path}: module keys must match coverage schema")
+        selected_scopes_raw = module_policy.get("selected_scopes", [])
+        if coverage == "selected_scopes":
+            if (
+                not isinstance(selected_scopes_raw, list)
+                or not selected_scopes_raw
+                or any(
+                    not isinstance(scope, str) or not scope
+                    for scope in selected_scopes_raw
+                )
+                or len(selected_scopes_raw) != len(set(selected_scopes_raw))
+            ):
+                failures.append(
+                    f"{rel_path}: selected_scopes must be a unique non-empty string list"
+                )
+                selected_scopes: set[str] = set()
+            else:
+                selected_scopes = set(selected_scopes_raw)
+        else:
+            if selected_scopes_raw:
+                failures.append(
+                    f"{rel_path}: all_broad_catches must not declare selected_scopes"
+                )
+            selected_scopes = set()
+
         entries_by_scope: dict[str, dict[str, Any]] = {}
         for index, entry in enumerate(allowed):
             if not isinstance(entry, dict):
                 failures.append(f"{rel_path}: broad_catches[{index}] must be an object")
                 continue
+            if set(entry) != {
+                "scope",
+                "expected_count",
+                "classification",
+                "reason",
+                "regression_owner",
+                "review_after",
+            }:
+                failures.append(
+                    f"{rel_path}: broad_catches[{index}] entry keys must match schema"
+                )
             scope = entry.get("scope")
             classification = entry.get("classification")
             reason = entry.get("reason")
+            regression_owner = entry.get("regression_owner")
+            review_after = entry.get("review_after")
             if not isinstance(scope, str) or not scope:
                 failures.append(f"{rel_path}: broad_catches[{index}] missing scope")
                 continue
@@ -128,10 +191,49 @@ def validate_exception_boundary_policy(
                 )
             if not isinstance(reason, str) or not reason.strip():
                 failures.append(f"{rel_path}:{scope}: missing reason")
+            if not isinstance(regression_owner, str) or not regression_owner.strip():
+                failures.append(f"{rel_path}:{scope}: missing regression_owner")
+            else:
+                owner_path = Path(regression_owner)
+                if (
+                    owner_path.is_absolute()
+                    or ".." in owner_path.parts
+                    or not owner_path.parts
+                    or owner_path.parts[0] != "tests"
+                    or owner_path.suffix != ".py"
+                ):
+                    failures.append(
+                        f"{rel_path}:{scope}: regression_owner must be a safe tests/*.py path"
+                    )
+                elif not (repo_root / owner_path).is_file():
+                    failures.append(
+                        f"{rel_path}:{scope}: regression_owner does not exist"
+                    )
+            try:
+                if not isinstance(review_after, str):
+                    raise TypeError
+                review_date = date.fromisoformat(review_after)
+            except (TypeError, ValueError):
+                failures.append(f"{rel_path}:{scope}: invalid review_after")
+            else:
+                if review_date < date.today():
+                    failures.append(f"{rel_path}:{scope}: review_after is expired")
+            if coverage == "selected_scopes" and scope not in selected_scopes:
+                failures.append(f"{rel_path}:{scope}: entry is outside selected_scopes")
 
         catches = tuple(iter_broad_catches(path))
-        counts = Counter(catch.scope for catch in catches)
-        for catch in catches:
+        governed_catches = (
+            catches
+            if coverage == "all_broad_catches"
+            else tuple(catch for catch in catches if catch.scope in selected_scopes)
+        )
+        counts = Counter(catch.scope for catch in governed_catches)
+        if coverage == "selected_scopes":
+            for stale_scope in sorted(selected_scopes - set(counts)):
+                failures.append(
+                    f"{rel_path}:{stale_scope}: selected scope has no broad catch"
+                )
+        for catch in governed_catches:
             if catch.scope not in entries_by_scope:
                 failures.append(
                     f"{rel_path}:{catch.line}: undocumented broad catch in {catch.scope}"