chore(quality): enforce incremental static analysis

.github/workflows/ci.yml

@@ -106,10 +106,14 @@ jobs:
           # CRITICAL: Python 3.10 coverage reads pyproject.toml only when the
           # TOML extra is present; do not downgrade this back to plain coverage.
           python -m pip install -r requirements.txt
+          python -m pip install -r requirements-quality.txt
           python -m pip install numpy pillow aiohttp "coverage[toml]"
       - name: R120 preflight
         run: |
           python scripts/preflight_check.py --strict
+      - name: Static-analysis policy
+        run: |
+          python scripts/verify_static_analysis_policy.py
       - name: Run MAE hard-guarantee suites
         env:
           MOLTBOT_STATE_DIR: ${{ github.workspace }}/moltbot_state/_ci_mae

.github/workflows/pre-commit.yml

@@ -31,6 +31,10 @@ jobs:
           # Install black/isort explicitly so CI doesn't fail due to missing tools
           # if a hook is configured to run via system python.
           pip install pre-commit black==24.1.1 isort==5.13.2
+          pip install -r requirements-quality.txt
+
+      - name: Verify static-analysis policy directly
+        run: python scripts/verify_static_analysis_policy.py
 
       - name: Run all pre-commit hooks
         run: pre-commit run --all-files --show-diff-on-failure

.pre-commit-config.yaml

@@ -41,6 +41,16 @@ repos:
         language: python
         pass_filenames: false
         always_run: true
+      - id: static-analysis-policy
+        name: incremental Ruff/Mypy static-analysis policy
+        # Keep isolated pins aligned with requirements-quality.txt and policy JSON.
+        entry: python -B scripts/verify_static_analysis_policy.py
+        language: python
+        additional_dependencies:
+          - ruff==0.15.20
+          - mypy==2.2.0
+        pass_filenames: false
+        always_run: true
 
   # Secret detection
   - repo: https://github.com/Yelp/detect-secrets

pyproject.toml

@@ -71,9 +71,12 @@ target-version = ["py310"]
 [tool.mypy]
 python_version = "3.10"
 warn_return_any = true
-warn_unused_configs = true
+warn_unused_configs = false
 disallow_untyped_defs = false  # Start lenient, can tighten later
 disallow_any_unimported = false
+explicit_package_bases = true
+no_site_packages = true
+ignore_missing_imports = true
 no_implicit_optional = true
 warn_redundant_casts = true
 warn_unused_ignores = true
@@ -81,8 +84,17 @@ warn_no_return = true
 check_untyped_defs = true
 strict_equality = true
 
-# Paths to check
+# Production paths governed by tests/static_analysis_policy.json.
-files = ["*.py", "tests/**/*.py"]
+files = [
+    "__init__.py",
+    "config.py",
+    "api",
+    "connector",
+    "models",
+    "nodes",
+    "services",
+    "scripts",
+]
 
 # Ignore missing imports for ComfyUI and external packages
 [[tool.mypy.overrides]]

requirements-quality.txt

@@ -0,0 +1,4 @@
+# Development/test-only static-analysis toolchain.
+# Keep exact pins aligned with tests/static_analysis_policy.json.
+ruff==0.15.20
+mypy==2.2.0

scripts/run_full_tests_linux.sh

@@ -123,6 +123,12 @@ if ! "$VENV_PY" -c "import cryptography" >/dev/null 2>&1; then
   echo "[tests] Installing cryptography into project venv ($VENV_DIR) ..."
   pip_install_or_fail "required for S57 secrets-at-rest encryption tests" cryptography
 fi
+
+if ! "$VENV_PY" -c "import json, sys; from importlib.metadata import version; p=json.load(open('tests/static_analysis_policy.json', encoding='utf-8')); sys.exit(0 if all(version(name)==cfg['version'] for name,cfg in p['tools'].items()) else 1)" >/dev/null 2>&1; then
+  echo "[tests] Installing pinned Ruff/Mypy into project venv ($VENV_DIR) ..."
+  pip_install_or_fail "required for static-analysis policy" -r requirements-quality.txt
+fi
+
 if ! "$VENV_PY" -c "import defusedxml" >/dev/null 2>&1; then
   # IMPORTANT: keep local full-test bootstrap aligned with requirements.txt.
   echo "[tests] Installing defusedxml into project venv ($VENV_DIR) ..."
@@ -194,6 +200,9 @@ echo "[tests] 0/11 supply-chain hardening check"
 
 ensure_npm_deps
 
+echo "[tests] 0.5/11 static analysis policy"
+"$VENV_PY" scripts/verify_static_analysis_policy.py
+
 echo "[tests] 0/9 R120 dependency preflight"
 "$VENV_PY" scripts/preflight_check.py --strict
 

scripts/run_full_tests_windows.ps1

@@ -195,6 +195,16 @@ if (-not $hasCoverageTomlSupport) {
   Invoke-Checked "pip install coverage[toml]" { & $venvPython -m pip install "coverage[toml]" }
 }
 
+$qualityToolsReady = $true
+& $venvPython -c "import json, sys; from importlib.metadata import version; p=json.load(open('tests/static_analysis_policy.json', encoding='utf-8')); sys.exit(0 if all(version(name)==cfg['version'] for name,cfg in p['tools'].items()) else 1)" | Out-Null
+if ($LASTEXITCODE -ne 0) {
+  $qualityToolsReady = $false
+}
+if (-not $qualityToolsReady) {
+  Write-Host "[tests] Installing pinned Ruff/Mypy into project venv ..."
+  Invoke-Checked "pip install quality tools" { & $venvPython -m pip install -r requirements-quality.txt }
+}
+
 # Ensure Node >= 18
 $nodeMajor = [int]((& node -p "process.versions.node.split('.')[0]").Trim())
 if ($nodeMajor -lt 18) {
@@ -267,6 +277,11 @@ Invoke-Checked "supply-chain hardening check" {
 }
 Ensure-NpmDeps
 
+Write-Host "[tests] 0.5/11 static analysis policy"
+Invoke-Checked "static analysis policy" {
+  & $venvPython scripts/verify_static_analysis_policy.py
+}
+
 Write-Host "[tests] 0/8 R120 dependency preflight"
 Invoke-Checked "preflight_check" { & $venvPython scripts\preflight_check.py --strict }
 

scripts/verify_static_analysis_policy.py

@@ -0,0 +1,604 @@
+"""Incremental Ruff/Mypy debt-ratchet verifier.
+
+The policy owns source paths and normalized diagnostic counts. Line numbers are
+intentionally excluded so harmless edits do not churn the baseline; count changes
+still fail and require an explicit reviewed baseline refresh.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import subprocess
+import sys
+from collections import Counter
+from collections.abc import Callable, Iterable, Mapping
+from copy import deepcopy
+from dataclasses import dataclass
+from datetime import date
+from pathlib import Path
+from typing import Any, Protocol
+
+
+@dataclass(frozen=True, order=True)
+class Diagnostic:
+    tool: str
+    path: str
+    code: str
+    message: str
+
+
+class _ProcessResult(Protocol):
+    returncode: int
+    stdout: str
+    stderr: str
+
+
+class ToolExecutionError(RuntimeError):
+    """Raised with content-free context when a quality tool cannot run."""
+
+
+def _normalize_message(message: Any, repo_root: Path | None = None) -> str:
+    normalized = " ".join(str(message or "").split())
+    if repo_root is not None:
+        variants = {
+            str(repo_root.resolve()),
+            str(repo_root.resolve()).replace("\\", "/"),
+        }
+        for variant in sorted(variants, key=len, reverse=True):
+            if variant:
+                normalized = normalized.replace(variant, "<repo>")
+    return normalized
+
+
+def _is_safe_relative_path(value: Any) -> bool:
+    if not isinstance(value, str) or not value.strip():
+        return False
+    candidate = Path(value)
+    return not candidate.is_absolute() and ".." not in candidate.parts
+
+
+def _repo_relative_path(path_value: Any, repo_root: Path) -> str:
+    path = Path(str(path_value))
+    if not path.is_absolute():
+        path = repo_root / path
+    try:
+        relative = path.resolve().relative_to(repo_root.resolve())
+    except ValueError as exc:
+        raise ValueError(f"diagnostic path is outside repository: {path_value}") from exc
+    return relative.as_posix()
+
+
+def _path_within(path: str, root: str) -> bool:
+    path_obj = Path(path)
+    root_obj = Path(root)
+    return path_obj == root_obj or root_obj in path_obj.parents
+
+
+def _excluded_path_values(policy: Mapping[str, Any]) -> tuple[str, ...]:
+    entries = policy.get("excluded_paths", [])
+    if not isinstance(entries, list):
+        return ()
+    return tuple(
+        str(entry.get("path", ""))
+        for entry in entries
+        if isinstance(entry, dict) and entry.get("path")
+    )
+
+
+def discover_owned_python_files(
+    repo_root: Path, policy: Mapping[str, Any]
+) -> tuple[str, ...]:
+    excluded = _excluded_path_values(policy)
+    discovered: set[str] = set()
+    for root_value in policy.get("production_roots", []):
+        root_path = repo_root / str(root_value)
+        candidates: Iterable[Path]
+        if root_path.is_file():
+            candidates = (root_path,) if root_path.suffix == ".py" else ()
+        elif root_path.is_dir():
+            candidates = root_path.rglob("*.py")
+        else:
+            continue
+        for candidate in candidates:
+            relative = _repo_relative_path(candidate, repo_root)
+            if any(_path_within(relative, excluded_path) for excluded_path in excluded):
+                continue
+            discovered.add(relative)
+    return tuple(sorted(discovered))
+
+
+def _baseline_counter(policy: Mapping[str, Any]) -> Counter[Diagnostic]:
+    baseline: Counter[Diagnostic] = Counter()
+    entries = policy.get("baseline", [])
+    if not isinstance(entries, list):
+        return baseline
+    for entry in entries:
+        if not isinstance(entry, dict):
+            continue
+        diagnostic = Diagnostic(
+            tool=str(entry.get("tool", "")),
+            path=str(entry.get("path", "")),
+            code=str(entry.get("code", "")),
+            message=_normalize_message(entry.get("message", "")),
+        )
+        count = entry.get("count", 0)
+        if isinstance(count, int) and count > 0:
+            baseline[diagnostic] += count
+    return baseline
+
+
+def validate_policy(repo_root: Path, policy: Mapping[str, Any]) -> list[str]:
+    failures: list[str] = []
+    if policy.get("schema_version") != 1:
+        failures.append("schema_version must be 1")
+
+    review = policy.get("review")
+    if not isinstance(review, dict):
+        failures.append("review must be an object")
+    else:
+        owner = review.get("owner")
+        if not isinstance(owner, str) or not owner.strip():
+            failures.append("review.owner must be a non-empty string")
+        parsed_dates: dict[str, date] = {}
+        for key in ("reviewed_at", "next_review_by"):
+            try:
+                parsed_dates[key] = date.fromisoformat(str(review.get(key, "")))
+            except ValueError:
+                failures.append(f"review.{key} must be an ISO date")
+        if (
+            len(parsed_dates) == 2
+            and parsed_dates["next_review_by"] < parsed_dates["reviewed_at"]
+        ):
+            failures.append("review.next_review_by must not precede reviewed_at")
+
+    tools = policy.get("tools")
+    if not isinstance(tools, dict) or set(tools) != {"ruff", "mypy"}:
+        failures.append("tools must contain exactly ruff and mypy")
+        tools = {}
+    for tool_name in ("ruff", "mypy"):
+        config = tools.get(tool_name)
+        if (
+            not isinstance(config, dict)
+            or not isinstance(config.get("version"), str)
+            or not config["version"].strip()
+        ):
+            failures.append(f"tools.{tool_name}.version must be a non-empty string")
+
+    roots = policy.get("production_roots")
+    if not isinstance(roots, list) or not roots:
+        failures.append("production_roots must be a non-empty list")
+        roots = []
+    seen_roots: set[str] = set()
+    valid_roots: list[str] = []
+    for index, value in enumerate(roots):
+        if not _is_safe_relative_path(value):
+            failures.append(f"production_roots[{index}] is unsafe: {value!r}")
+            continue
+        value = str(value)
+        if value in seen_roots:
+            failures.append(f"duplicate production root: {value}")
+            continue
+        seen_roots.add(value)
+        valid_roots.append(value)
+        if not (repo_root / value).exists():
+            failures.append(f"production root is missing: {value}")
+
+    excluded_entries = policy.get("excluded_paths")
+    if not isinstance(excluded_entries, list):
+        failures.append("excluded_paths must be a list")
+        excluded_entries = []
+    seen_excluded: set[str] = set()
+    for index, entry in enumerate(excluded_entries):
+        if not isinstance(entry, dict):
+            failures.append(f"excluded_paths[{index}] must be an object")
+            continue
+        value = entry.get("path")
+        reason = entry.get("reason")
+        if not _is_safe_relative_path(value):
+            failures.append(f"excluded_paths[{index}] is unsafe: {value!r}")
+            continue
+        value = str(value)
+        if value in seen_excluded:
+            failures.append(f"duplicate excluded path: {value}")
+        seen_excluded.add(value)
+        if not isinstance(reason, str) or not reason.strip():
+            failures.append(f"excluded path {value} is missing a reason")
+        if not any(_path_within(value, root) for root in valid_roots):
+            failures.append(f"excluded path is outside owned roots: {value}")
+        if not (repo_root / value).exists():
+            failures.append(f"excluded path is missing: {value}")
+
+    owned_files = set(discover_owned_python_files(repo_root, policy))
+    strict_paths = policy.get("strict_paths")
+    if not isinstance(strict_paths, list):
+        failures.append("strict_paths must be a list")
+        strict_paths = []
+    seen_strict: set[str] = set()
+    for index, value in enumerate(strict_paths):
+        if not _is_safe_relative_path(value):
+            failures.append(f"strict_paths[{index}] is unsafe: {value!r}")
+            continue
+        value = str(value)
+        if value in seen_strict:
+            failures.append(f"duplicate strict path: {value}")
+        seen_strict.add(value)
+        if not any(_path_within(path, value) for path in owned_files):
+            failures.append(f"strict path has no owned Python files: {value}")
+
+    baseline_entries = policy.get("baseline")
+    if not isinstance(baseline_entries, list):
+        failures.append("baseline must be a list")
+        baseline_entries = []
+    seen_diagnostics: set[Diagnostic] = set()
+    for index, entry in enumerate(baseline_entries):
+        if not isinstance(entry, dict):
+            failures.append(f"baseline[{index}] must be an object")
+            continue
+        diagnostic = Diagnostic(
+            tool=str(entry.get("tool", "")),
+            path=str(entry.get("path", "")),
+            code=str(entry.get("code", "")),
+            message=_normalize_message(entry.get("message", "")),
+        )
+        if diagnostic.tool not in {"ruff", "mypy"}:
+            failures.append(f"baseline[{index}] has unknown tool {diagnostic.tool!r}")
+        if diagnostic.path not in owned_files:
+            failures.append(
+                f"baseline[{index}] path is not an owned Python file: {diagnostic.path}"
+            )
+        if not diagnostic.code or not diagnostic.message:
+            failures.append(f"baseline[{index}] code/message must be non-empty")
+        count = entry.get("count")
+        if not isinstance(count, int) or isinstance(count, bool) or count < 1:
+            failures.append(f"baseline[{index}] count must be a positive integer")
+        if diagnostic in seen_diagnostics:
+            failures.append(f"duplicate baseline diagnostic: {diagnostic}")
+        seen_diagnostics.add(diagnostic)
+
+    if baseline_entries != serialize_baseline(_baseline_counter(policy)):
+        failures.append("baseline must use canonical sorted serialization")
+
+    return failures
+
+
+def validate_tool_versions(
+    policy: Mapping[str, Any], detected_versions: Mapping[str, str]
+) -> list[str]:
+    failures: list[str] = []
+    tools = policy.get("tools", {})
+    for tool_name in ("ruff", "mypy"):
+        config = tools.get(tool_name, {}) if isinstance(tools, dict) else {}
+        expected = config.get("version") if isinstance(config, dict) else None
+        found = detected_versions.get(tool_name, "missing")
+        if expected != found:
+            failures.append(
+                f"{tool_name} version drift: expected {expected}, found {found}"
+            )
+    return failures
+
+
+def validate_requirement_pins(
+    policy: Mapping[str, Any], requirement_lines: Iterable[str]
+) -> list[str]:
+    requirements: dict[str, str] = {}
+    for raw_line in requirement_lines:
+        line = raw_line.split("#", 1)[0].strip()
+        if not line:
+            continue
+        match = re.match(r"^(ruff|mypy)(?:\[.*\])?(.*)$", line, re.IGNORECASE)
+        if match:
+            requirements[match.group(1).lower()] = (
+                match.group(1).lower() + match.group(2).strip()
+            )
+
+    failures: list[str] = []
+    tools = policy.get("tools", {})
+    for tool_name in sorted(("ruff", "mypy")):
+        config = tools.get(tool_name, {}) if isinstance(tools, dict) else {}
+        version = config.get("version") if isinstance(config, dict) else None
+        expected = f"{tool_name}=={version}"
+        found = requirements.get(tool_name, "missing")
+        if found != expected:
+            failures.append(
+                f"{tool_name} requirement drift: expected {expected}, found {found}"
+            )
+    return failures
+
+
+def _run_command(
+    runner: Callable[..., _ProcessResult], command: list[str], repo_root: Path
+) -> _ProcessResult:
+    return runner(
+        command,
+        cwd=repo_root,
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        check=False,
+        shell=False,
+    )
+
+
+def _parse_tool_version(tool_name: str, output: str) -> str:
+    match = re.search(
+        rf"\b{re.escape(tool_name)}\s+([0-9]+(?:\.[0-9]+)+)", output
+    )
+    if not match:
+        raise ToolExecutionError(f"{tool_name} version output was not recognized")
+    return match.group(1)
+
+
+def run_static_analysis(
+    repo_root: Path,
+    policy: Mapping[str, Any],
+    *,
+    runner: Callable[..., _ProcessResult] = subprocess.run,
+) -> tuple[dict[str, str], Counter[Diagnostic]]:
+    owned_files = list(discover_owned_python_files(repo_root, policy))
+    versions: dict[str, str] = {}
+
+    for tool_name in ("ruff", "mypy"):
+        command = [sys.executable, "-m", tool_name, "--version"]
+        result = _run_command(runner, command, repo_root)
+        if result.returncode != 0:
+            raise ToolExecutionError(
+                f"{tool_name} version check failed with exit code {result.returncode}"
+            )
+        versions[tool_name] = _parse_tool_version(tool_name, result.stdout)
+
+    commands = (
+        (
+            "ruff",
+            [
+                sys.executable,
+                "-m",
+                "ruff",
+                "check",
+                "--output-format",
+                "json",
+                "--no-cache",
+                *owned_files,
+            ],
+        ),
+        (
+            "mypy",
+            [
+                sys.executable,
+                "-m",
+                "mypy",
+                "--output",
+                "json",
+                "--no-incremental",
+                "--explicit-package-bases",
+                "--no-warn-unused-configs",
+                "--no-error-summary",
+                "--no-site-packages",
+                "--ignore-missing-imports",
+                *owned_files,
+            ],
+        ),
+    )
+
+    diagnostics: Counter[Diagnostic] = Counter()
+    for tool_name, command in commands:
+        result = _run_command(runner, command, repo_root)
+        if result.returncode not in {0, 1}:
+            # SECURITY: stderr may contain private host paths or source content.
+            # Keep the public/loggable failure content-free and deterministic.
+            raise ToolExecutionError(
+                f"{tool_name} execution failed with exit code {result.returncode}"
+            )
+        if tool_name == "ruff":
+            diagnostics.update(parse_ruff_output(result.stdout, repo_root))
+        else:
+            diagnostics.update(parse_mypy_output(result.stdout, repo_root))
+    return versions, diagnostics
+
+
+def compare_diagnostics(
+    policy: Mapping[str, Any], current: Counter[Diagnostic]
+) -> list[str]:
+    failures: list[str] = []
+    baseline = _baseline_counter(policy)
+    strict_paths = tuple(str(value) for value in policy.get("strict_paths", []))
+
+    for diagnostic, count in sorted(current.items()):
+        if any(_path_within(diagnostic.path, path) for path in strict_paths):
+            failures.append(
+                "strict path diagnostic: "
+                f"{diagnostic.tool}:{diagnostic.path}:{diagnostic.code} x{count}"
+            )
+
+    for diagnostic in sorted(set(baseline) | set(current)):
+        expected = baseline.get(diagnostic, 0)
+        found = current.get(diagnostic, 0)
+        label = (
+            f"{diagnostic.tool}:{diagnostic.path}:{diagnostic.code}:"
+            f"{diagnostic.message}"
+        )
+        if found > expected:
+            failures.append(f"new debt: {label} expected {expected}, found {found}")
+        elif found < expected:
+            failures.append(
+                f"stale baseline: {label} expected {expected}, found {found}"
+            )
+    return failures
+
+
+def _diagnostic_from_payload(
+    *, tool: str, payload: Mapping[str, Any], path_key: str, repo_root: Path
+) -> Diagnostic:
+    return Diagnostic(
+        tool=tool,
+        path=_repo_relative_path(payload.get(path_key, ""), repo_root),
+        code=str(payload.get("code") or "unknown"),
+        message=_normalize_message(payload.get("message", ""), repo_root),
+    )
+
+
+def parse_ruff_output(raw: str, repo_root: Path) -> Counter[Diagnostic]:
+    payload = json.loads(raw or "[]")
+    if not isinstance(payload, list):
+        raise ValueError("Ruff JSON output must be a list")
+    diagnostics: Counter[Diagnostic] = Counter()
+    for entry in payload:
+        if not isinstance(entry, dict):
+            raise ValueError("Ruff JSON diagnostic must be an object")
+        diagnostics[
+            _diagnostic_from_payload(
+                tool="ruff", payload=entry, path_key="filename", repo_root=repo_root
+            )
+        ] += 1
+    return diagnostics
+
+
+def parse_mypy_output(raw: str, repo_root: Path) -> Counter[Diagnostic]:
+    stripped = raw.strip()
+    if not stripped:
+        return Counter()
+    if stripped.startswith("["):
+        payloads = json.loads(stripped)
+    else:
+        payloads = [json.loads(line) for line in stripped.splitlines() if line.strip()]
+    if not isinstance(payloads, list):
+        raise ValueError("Mypy JSON output must be a list or JSON lines")
+    diagnostics: Counter[Diagnostic] = Counter()
+    for entry in payloads:
+        if not isinstance(entry, dict):
+            raise ValueError("Mypy JSON diagnostic must be an object")
+        if entry.get("severity", "error") != "error":
+            continue
+        diagnostics[
+            _diagnostic_from_payload(
+                tool="mypy", payload=entry, path_key="file", repo_root=repo_root
+            )
+        ] += 1
+    return diagnostics
+
+
+def serialize_baseline(diagnostics: Counter[Diagnostic]) -> list[dict[str, Any]]:
+    return [
+        {
+            "tool": diagnostic.tool,
+            "path": diagnostic.path,
+            "code": diagnostic.code,
+            "message": diagnostic.message,
+            "count": count,
+        }
+        for diagnostic, count in sorted(diagnostics.items())
+        if count > 0
+    ]
+
+
+def with_updated_baseline(
+    policy: Mapping[str, Any], diagnostics: Counter[Diagnostic]
+) -> dict[str, Any]:
+    updated = deepcopy(dict(policy))
+    updated["baseline"] = serialize_baseline(diagnostics)
+    return updated
+
+
+def evaluate_policy(
+    repo_root: Path,
+    policy: Mapping[str, Any],
+    *,
+    requirement_lines: Iterable[str],
+    runner: Callable[..., _ProcessResult] = subprocess.run,
+) -> tuple[list[str], Counter[Diagnostic]]:
+    failures = validate_policy(repo_root, policy)
+    if failures:
+        return failures, Counter()
+    failures.extend(validate_requirement_pins(policy, requirement_lines))
+
+    versions, diagnostics = run_static_analysis(repo_root, policy, runner=runner)
+    failures.extend(validate_tool_versions(policy, versions))
+    failures.extend(compare_diagnostics(policy, diagnostics))
+    return failures, diagnostics
+
+
+def _write_policy(path: Path, policy: Mapping[str, Any]) -> None:
+    path.write_text(
+        json.dumps(policy, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
+    )
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Verify the incremental Ruff/Mypy static-analysis debt policy."
+    )
+    parser.add_argument("--repo-root", default=".")
+    parser.add_argument("--policy", default="tests/static_analysis_policy.json")
+    parser.add_argument("--requirements", default="requirements-quality.txt")
+    parser.add_argument(
+        "--write-baseline",
+        action="store_true",
+        help="Explicitly replace the accepted diagnostic baseline after review.",
+    )
+    args = parser.parse_args(argv)
+
+    repo_root = Path(args.repo_root).resolve()
+    policy_path = repo_root / args.policy
+    requirements_path = repo_root / args.requirements
+    try:
+        policy = json.loads(policy_path.read_text(encoding="utf-8"))
+        requirement_lines = requirements_path.read_text(encoding="utf-8").splitlines()
+        structural_failures = validate_policy(repo_root, policy)
+        structural_failures.extend(
+            validate_requirement_pins(policy, requirement_lines)
+        )
+        if structural_failures:
+            for failure in structural_failures:
+                print(f"STATIC-ANALYSIS-FAIL: {failure}")
+            return 1
+
+        versions, diagnostics = run_static_analysis(repo_root, policy)
+        failures = validate_tool_versions(policy, versions)
+        if args.write_baseline:
+            strict_policy = with_updated_baseline(policy, diagnostics)
+            strict_failures = [
+                failure
+                for failure in compare_diagnostics(strict_policy, diagnostics)
+                if failure.startswith("strict path diagnostic:")
+            ]
+            if strict_failures:
+                for failure in strict_failures:
+                    print(f"STATIC-ANALYSIS-FAIL: {failure}")
+                return 1
+            if failures:
+                for failure in failures:
+                    print(f"STATIC-ANALYSIS-FAIL: {failure}")
+                return 1
+            _write_policy(policy_path, strict_policy)
+            print(
+                "STATIC-ANALYSIS-BASELINE-WRITTEN: "
+                f"{len(diagnostics)} fingerprints, {sum(diagnostics.values())} findings"
+            )
+            return 0
+
+        failures.extend(compare_diagnostics(policy, diagnostics))
+        if failures:
+            for failure in failures:
+                print(f"STATIC-ANALYSIS-FAIL: {failure}")
+            return 1
+        print(
+            "STATIC-ANALYSIS-PASS: "
+            f"{len(diagnostics)} fingerprints, {sum(diagnostics.values())} governed findings"
+        )
+        return 0
+    except (OSError, ValueError, json.JSONDecodeError) as exc:
+        print(f"STATIC-ANALYSIS-FAIL: invalid policy or tool output ({type(exc).__name__})")
+        return 1
+    except ToolExecutionError as exc:
+        print(f"STATIC-ANALYSIS-FAIL: {exc}")
+        print(
+            "STATIC-ANALYSIS-REMEDIATION: use the project-local Python to install "
+            "requirements-quality.txt"
+        )
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_r217_static_analysis_policy.py

@@ -0,0 +1,521 @@
+import json
+import sys
+import tempfile
+import unittest
+from collections import Counter
+from pathlib import Path
+from types import SimpleNamespace
+
+from scripts import verify_static_analysis_policy as policy_module
+
+
+class TestStaticAnalysisPolicy(unittest.TestCase):
+    def _policy(self) -> dict:
+        return {
+            "schema_version": 1,
+            "review": {
+                "owner": "maintainers",
+                "reviewed_at": "2026-07-11",
+                "next_review_by": "2026-10-11",
+            },
+            "tools": {
+                "ruff": {"version": "0.15.20"},
+                "mypy": {"version": "2.2.0"},
+            },
+            "production_roots": ["config.py", "pkg"],
+            "excluded_paths": [
+                {
+                    "path": "pkg/generated.py",
+                    "reason": "generated fixture excluded from source ownership",
+                }
+            ],
+            "strict_paths": ["pkg/clean.py"],
+            "baseline": [
+                {
+                    "tool": "ruff",
+                    "path": "pkg/owned.py",
+                    "code": "F401",
+                    "message": "unused import",
+                    "count": 1,
+                }
+            ],
+        }
+
+    def _create_repo(self, root: Path) -> None:
+        (root / "config.py").write_text("VALUE = 1\n", encoding="utf-8")
+        (root / "pkg" / "nested").mkdir(parents=True)
+        (root / "pkg" / "owned.py").write_text("import os\n", encoding="utf-8")
+        (root / "pkg" / "clean.py").write_text("VALUE: int = 1\n", encoding="utf-8")
+        (root / "pkg" / "generated.py").write_text("generated = True\n", encoding="utf-8")
+        (root / "pkg" / "nested" / "child.py").write_text(
+            "CHILD = True\n", encoding="utf-8"
+        )
+
+    def test_policy_validation_rejects_missing_and_unsafe_paths(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            self._create_repo(root)
+            policy = self._policy()
+            policy["production_roots"].extend(["missing", "../outside"])
+            policy["strict_paths"].append("pkg/missing.py")
+            policy["excluded_paths"].append(
+                {"path": "reference/untrusted.py", "reason": "not inside an owned root"}
+            )
+
+            failures = policy_module.validate_policy(root, policy)
+
+        self.assertTrue(any("missing" in failure for failure in failures))
+        self.assertTrue(any("unsafe" in failure for failure in failures))
+        self.assertTrue(any("strict path" in failure for failure in failures))
+        self.assertTrue(any("excluded path" in failure for failure in failures))
+
+    def test_owned_file_discovery_is_recursive_and_excludes_only_declared_paths(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            self._create_repo(root)
+
+            files = policy_module.discover_owned_python_files(root, self._policy())
+
+        self.assertEqual(
+            files,
+            (
+                "config.py",
+                "pkg/clean.py",
+                "pkg/nested/child.py",
+                "pkg/owned.py",
+            ),
+        )
+
+    def test_exact_baseline_passes(self):
+        policy = self._policy()
+        current = Counter(
+            {
+                policy_module.Diagnostic(
+                    tool="ruff",
+                    path="pkg/owned.py",
+                    code="F401",
+                    message="unused import",
+                ): 1
+            }
+        )
+
+        self.assertEqual(policy_module.compare_diagnostics(policy, current), [])
+
+    def test_new_and_resolved_debt_both_fail_the_ratchet(self):
+        policy = self._policy()
+        key = policy_module.Diagnostic(
+            tool="ruff",
+            path="pkg/owned.py",
+            code="F401",
+            message="unused import",
+        )
+
+        increased = policy_module.compare_diagnostics(policy, Counter({key: 2}))
+        resolved = policy_module.compare_diagnostics(policy, Counter())
+
+        self.assertTrue(any("new debt" in failure for failure in increased))
+        self.assertTrue(any("stale baseline" in failure for failure in resolved))
+
+    def test_strict_path_rejects_even_baselined_diagnostic(self):
+        policy = self._policy()
+        policy["baseline"].append(
+            {
+                "tool": "mypy",
+                "path": "pkg/clean.py",
+                "code": "assignment",
+                "message": "incompatible assignment",
+                "count": 1,
+            }
+        )
+        current = Counter(
+            {
+                policy_module.Diagnostic(
+                    tool="ruff",
+                    path="pkg/owned.py",
+                    code="F401",
+                    message="unused import",
+                ): 1,
+                policy_module.Diagnostic(
+                    tool="mypy",
+                    path="pkg/clean.py",
+                    code="assignment",
+                    message="incompatible assignment",
+                ): 1,
+            }
+        )
+
+        failures = policy_module.compare_diagnostics(policy, current)
+
+        self.assertTrue(any("strict path" in failure for failure in failures))
+
+    def test_tool_version_drift_is_rejected(self):
+        failures = policy_module.validate_tool_versions(
+            self._policy(), {"ruff": "0.15.19", "mypy": "2.2.0"}
+        )
+
+        self.assertEqual(
+            failures,
+            ["ruff version drift: expected 0.15.20, found 0.15.19"],
+        )
+
+    def test_policy_validation_rejects_noncanonical_baseline_order(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            self._create_repo(root)
+            policy = self._policy()
+            policy["baseline"] = [
+                {
+                    "tool": "ruff",
+                    "path": "pkg/owned.py",
+                    "code": "F401",
+                    "message": "unused import",
+                    "count": 1,
+                },
+                {
+                    "tool": "mypy",
+                    "path": "pkg/owned.py",
+                    "code": "assignment",
+                    "message": "bad assignment",
+                    "count": 1,
+                },
+            ]
+
+            failures = policy_module.validate_policy(root, policy)
+
+        self.assertIn("baseline must use canonical sorted serialization", failures)
+
+    def test_ruff_json_is_normalized_to_repo_relative_diagnostics(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            source = root / "pkg" / "owned.py"
+            raw = json.dumps(
+                [
+                    {
+                        "filename": str(source),
+                        "code": "F401",
+                        "message": "  unused   import  ",
+                        "location": {"row": 1, "column": 1},
+                    }
+                ]
+            )
+
+            result = policy_module.parse_ruff_output(raw, root)
+
+        self.assertEqual(
+            result,
+            Counter(
+                {
+                    policy_module.Diagnostic(
+                        tool="ruff",
+                        path="pkg/owned.py",
+                        code="F401",
+                        message="unused import",
+                    ): 1
+                }
+            ),
+        )
+
+    def test_mypy_json_lines_are_normalized_without_line_numbers(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            source = root / "pkg" / "owned.py"
+            raw = json.dumps(
+                {
+                    "file": str(source),
+                    "line": 4,
+                    "column": 2,
+                    "message": " Name  is not defined ",
+                    "hint": None,
+                    "code": "name-defined",
+                    "severity": "error",
+                }
+            )
+
+            result = policy_module.parse_mypy_output(raw, root)
+
+        self.assertEqual(
+            result,
+            Counter(
+                {
+                    policy_module.Diagnostic(
+                        tool="mypy",
+                        path="pkg/owned.py",
+                        code="name-defined",
+                        message="Name is not defined",
+                    ): 1
+                }
+            ),
+        )
+
+    def test_serialized_baseline_is_deterministic_and_sorted(self):
+        diagnostics = Counter(
+            {
+                policy_module.Diagnostic("ruff", "z.py", "F401", "unused"): 2,
+                policy_module.Diagnostic("mypy", "a.py", "assignment", "bad"): 1,
+            }
+        )
+
+        self.assertEqual(
+            policy_module.serialize_baseline(diagnostics),
+            [
+                {
+                    "tool": "mypy",
+                    "path": "a.py",
+                    "code": "assignment",
+                    "message": "bad",
+                    "count": 1,
+                },
+                {
+                    "tool": "ruff",
+                    "path": "z.py",
+                    "code": "F401",
+                    "message": "unused",
+                    "count": 2,
+                },
+            ],
+        )
+
+    def test_quality_requirement_pins_must_match_policy(self):
+        policy = self._policy()
+
+        self.assertEqual(
+            policy_module.validate_requirement_pins(
+                policy, ["ruff==0.15.20", "mypy==2.2.0"]
+            ),
+            [],
+        )
+        self.assertEqual(
+            policy_module.validate_requirement_pins(
+                policy, ["ruff>=0.15.20", "mypy==2.1.0"]
+            ),
+            [
+                "mypy requirement drift: expected mypy==2.2.0, found mypy==2.1.0",
+                "ruff requirement drift: expected ruff==0.15.20, found ruff>=0.15.20",
+            ],
+        )
+
+    def test_tool_runner_uses_current_interpreter_and_combines_diagnostics(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            self._create_repo(root)
+            commands: list[tuple[str, ...]] = []
+
+            def runner(command, **kwargs):
+                self.assertEqual(kwargs["cwd"], root)
+                self.assertFalse(kwargs.get("shell", False))
+                commands.append(tuple(command))
+                if command[-1] == "--version":
+                    name = command[2]
+                    output = (
+                        "ruff 0.15.20\n"
+                        if name == "ruff"
+                        else "mypy 2.2.0 (compiled: yes)\n"
+                    )
+                    return SimpleNamespace(returncode=0, stdout=output, stderr="")
+                if command[2] == "ruff":
+                    return SimpleNamespace(
+                        returncode=1,
+                        stdout=json.dumps(
+                            [
+                                {
+                                    "filename": str(root / "pkg" / "owned.py"),
+                                    "code": "F401",
+                                    "message": "unused import",
+                                }
+                            ]
+                        ),
+                        stderr="",
+                    )
+                return SimpleNamespace(
+                    returncode=1,
+                    stdout=json.dumps(
+                        {
+                            "file": str(root / "pkg" / "nested" / "child.py"),
+                            "line": 1,
+                            "column": 1,
+                            "message": "missing annotation",
+                            "code": "no-untyped-def",
+                            "severity": "error",
+                        }
+                    ),
+                    stderr="",
+                )
+
+            versions, diagnostics = policy_module.run_static_analysis(
+                root, self._policy(), runner=runner
+            )
+
+        self.assertEqual(versions, {"ruff": "0.15.20", "mypy": "2.2.0"})
+        self.assertEqual(sum(diagnostics.values()), 2)
+        self.assertEqual(len(commands), 4)
+        for command in commands:
+            self.assertEqual(command[:2], (sys.executable, "-m"))
+            self.assertIn(command[2], {"ruff", "mypy"})
+        ruff_command = next(command for command in commands if "check" in command)
+        mypy_command = next(command for command in commands if "--output" in command)
+        self.assertIn("--no-cache", ruff_command)
+        self.assertIn("json", ruff_command)
+        self.assertIn("--no-incremental", mypy_command)
+        self.assertIn("--explicit-package-bases", mypy_command)
+        self.assertIn("--no-warn-unused-configs", mypy_command)
+        self.assertIn("--no-error-summary", mypy_command)
+        self.assertIn("--no-site-packages", mypy_command)
+        self.assertIn("--ignore-missing-imports", mypy_command)
+        self.assertIn("json", mypy_command)
+
+    def test_tool_runner_rejects_abnormal_tool_failure_without_echoing_stderr(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            self._create_repo(root)
+
+            def runner(command, **_kwargs):
+                if command[-1] == "--version":
+                    name = command[2]
+                    return SimpleNamespace(
+                        returncode=0,
+                        stdout=f"{name} 0.15.20\n"
+                        if name == "ruff"
+                        else "mypy 2.2.0\n",
+                        stderr="",
+                    )
+                return SimpleNamespace(
+                    returncode=2,
+                    stdout="",
+                    stderr="private C:\\Users\\name\\secret tool failure",
+                )
+
+            with self.assertRaises(policy_module.ToolExecutionError) as ctx:
+                policy_module.run_static_analysis(root, self._policy(), runner=runner)
+
+        self.assertEqual(str(ctx.exception), "ruff execution failed with exit code 2")
+        self.assertNotIn("private", str(ctx.exception))
+
+    def test_updated_policy_baseline_does_not_mutate_input(self):
+        policy = self._policy()
+        diagnostics = Counter(
+            {policy_module.Diagnostic("mypy", "pkg/owned.py", "assignment", "bad"): 1}
+        )
+
+        updated = policy_module.with_updated_baseline(policy, diagnostics)
+
+        self.assertNotEqual(updated["baseline"], policy["baseline"])
+        self.assertEqual(policy["baseline"][0]["tool"], "ruff")
+        self.assertEqual(updated["baseline"][0]["tool"], "mypy")
+
+    def test_evaluate_policy_reports_requirement_and_installed_version_drift(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            self._create_repo(root)
+            policy = self._policy()
+            policy["baseline"] = []
+
+            def runner(command, **_kwargs):
+                if command[-1] == "--version":
+                    return SimpleNamespace(
+                        returncode=0,
+                        stdout=(
+                            "ruff 0.15.19\n"
+                            if command[2] == "ruff"
+                            else "mypy 2.2.0\n"
+                        ),
+                        stderr="",
+                    )
+                return SimpleNamespace(
+                    returncode=0,
+                    stdout="[]" if command[2] == "ruff" else "",
+                    stderr="",
+                )
+
+            failures, diagnostics = policy_module.evaluate_policy(
+                root,
+                policy,
+                requirement_lines=["ruff>=0.15.20", "mypy==2.2.0"],
+                runner=runner,
+            )
+
+        self.assertEqual(diagnostics, Counter())
+        self.assertIn(
+            "ruff requirement drift: expected ruff==0.15.20, found ruff>=0.15.20",
+            failures,
+        )
+        self.assertIn(
+            "ruff version drift: expected 0.15.20, found 0.15.19", failures
+        )
+
+
+class TestRepositoryStaticAnalysisPolicy(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.repo_root = Path(__file__).resolve().parents[1]
+        cls.policy_path = cls.repo_root / "tests" / "static_analysis_policy.json"
+        cls.requirements_path = cls.repo_root / "requirements-quality.txt"
+
+    def test_repository_policy_and_quality_pins_are_valid(self):
+        policy = json.loads(self.policy_path.read_text(encoding="utf-8"))
+        requirement_lines = self.requirements_path.read_text(encoding="utf-8").splitlines()
+
+        self.assertEqual(policy_module.validate_policy(self.repo_root, policy), [])
+        self.assertEqual(
+            policy_module.validate_requirement_pins(policy, requirement_lines), []
+        )
+        self.assertEqual(
+            policy["production_roots"],
+            [
+                "__init__.py",
+                "config.py",
+                "api",
+                "connector",
+                "models",
+                "nodes",
+                "services",
+                "scripts",
+            ],
+        )
+        self.assertGreater(len(policy_module.discover_owned_python_files(self.repo_root, policy)), 200)
+        self.assertGreaterEqual(len(policy["strict_paths"]), 1)
+
+    def test_repository_execution_surfaces_use_shared_verifier(self):
+        expected = "scripts/verify_static_analysis_policy.py"
+        surfaces = {
+            ".pre-commit-config.yaml": self.repo_root / ".pre-commit-config.yaml",
+            "windows full gate": self.repo_root / "scripts" / "run_full_tests_windows.ps1",
+            "linux full gate": self.repo_root / "scripts" / "run_full_tests_linux.sh",
+            "pre-commit CI": self.repo_root / ".github" / "workflows" / "pre-commit.yml",
+            "unit CI": self.repo_root / ".github" / "workflows" / "ci.yml",
+        }
+        for label, path in surfaces.items():
+            with self.subTest(surface=label):
+                content = path.read_text(encoding="utf-8")
+                self.assertIn(expected, content)
+                self.assertIn("requirements-quality.txt", content)
+
+    def test_precommit_hook_has_an_isolated_pinned_tool_environment(self):
+        content = (self.repo_root / ".pre-commit-config.yaml").read_text(
+            encoding="utf-8"
+        )
+        hook = content.split("- id: static-analysis-policy", 1)[1].split(
+            "# Secret detection", 1
+        )[0]
+
+        self.assertIn("language: python", hook)
+        self.assertIn("ruff==0.15.20", hook)
+        self.assertIn("mypy==2.2.0", hook)
+
+    def test_quality_tools_are_not_runtime_requirements(self):
+        runtime_requirements = (self.repo_root / "requirements.txt").read_text(
+            encoding="utf-8"
+        )
+        project_config = (self.repo_root / "pyproject.toml").read_text(encoding="utf-8")
+
+        self.assertNotIn("ruff", runtime_requirements.lower())
+        self.assertNotIn("mypy", runtime_requirements.lower())
+        dependencies_block = project_config.split("[project.urls]", 1)[0]
+        self.assertNotIn("ruff", dependencies_block.lower())
+        self.assertNotIn("mypy", dependencies_block.lower())
+        self.assertIn("explicit_package_bases = true", project_config)
+        self.assertIn("no_site_packages = true", project_config)
+        self.assertIn("ignore_missing_imports = true", project_config)
+
+
+if __name__ == "__main__":
+    unittest.main()