CK-QG1: enforce agent-kernel maintainability ratchet

config/agent-kernel/maintainability-baseline-v1.json

@@ -0,0 +1,125 @@
+{
+  "baseline_findings": [
+    {
+      "id": "adapters/codex_jsonl/normalize.py:normalize_record",
+      "score": 54,
+      "count": 1
+    },
+    {
+      "id": "adapters/codex_jsonl/parser.py:iter_source_batches",
+      "score": 28,
+      "count": 1
+    },
+    {
+      "id": "domain/formulas.py",
+      "score": 143,
+      "count": 11
+    },
+    {
+      "id": "domain/formulas.py:evaluate_formula",
+      "score": 109,
+      "count": 1
+    },
+    {
+      "id": "domain/plan_operands.py:_validate_materialization",
+      "score": 39,
+      "count": 1
+    },
+    {
+      "id": "domain/valuation.py:_validate_revision",
+      "score": 23,
+      "count": 1
+    },
+    {
+      "id": "domain/valuation.py:_validated_frontier",
+      "score": 26,
+      "count": 1
+    },
+    {
+      "id": "evidence/selectors.py:_normalize_entries",
+      "score": 55,
+      "count": 1
+    },
+    {
+      "id": "evidence/selectors.py:_owner_rules",
+      "score": 24,
+      "count": 1
+    },
+    {
+      "id": "evidence/selectors.py:_resolve_one",
+      "score": 21,
+      "count": 1
+    },
+    {
+      "id": "evidence/service.py:EvidenceRequest.__post_init__",
+      "score": 36,
+      "count": 1
+    },
+    {
+      "id": "evidence/service.py:_typed_row",
+      "score": 25,
+      "count": 1
+    },
+    {
+      "id": "publication/preparation.py:_WriteSetPreparer._add_session_relationship",
+      "score": 30,
+      "count": 1
+    },
+    {
+      "id": "query/compiler.py:DatabaseV1FactCompiler._validate_publication_authority",
+      "score": 62,
+      "count": 1
+    },
+    {
+      "id": "query/compiler.py:_attach_occurrence_event_coordinates",
+      "score": 21,
+      "count": 1
+    },
+    {
+      "id": "query/contracts.py:QueryPage.__post_init__",
+      "score": 23,
+      "count": 1
+    },
+    {
+      "id": "query/registry.py:QueryDefinition.validate_request",
+      "score": 24,
+      "count": 1
+    },
+    {
+      "id": "query/registry.py:build_registry",
+      "score": 75,
+      "count": 1
+    },
+    {
+      "id": "storage/lifecycle.py:fold_lifecycle",
+      "score": 23,
+      "count": 1
+    }
+  ],
+  "dependency_sha": "306cef37eea2ae017aca824d898cc435f7e1bea0",
+  "improved_findings": [],
+  "new_findings": [],
+  "normalization_version": "xenon-threshold-findings-v1",
+  "retirement_condition": "CK-14 may retire the frozen-spike check only; this replacement baseline remains until every baseline finding is removed and the same thresholds can be enforced with no recorded debt.",
+  "schema": "codex-usage-tracker.agent-kernel-maintainability-baseline.v1",
+  "scope_ownership": {
+    "active_thresholds": {
+      "average_max_rank": "B",
+      "block_max_rank": "C",
+      "module_max_rank": "B"
+    },
+    "consumer_seam": [
+      "just vp",
+      "just v",
+      "just vc",
+      ".github/workflows/ci.yml"
+    ],
+    "owned_lock": [
+      "replacement maintainability baseline",
+      "normalized maintainability CI gate"
+    ],
+    "source_root": "src/codex_usage_tracker/agent_kernel"
+  },
+  "tool_identity": "xenon==0.9.3;radon==6.0.1",
+  "worsened_findings": []
+}

pyproject.toml

@@ -41,11 +41,12 @@ dev = [
   "psutil==7.2.2",
   "pytest>=8.0",
   "pyright>=1.1.405",
+  "radon==6.0.1",
   "ruff>=0.8",
   "scalene==2.3.0",
   "tomli>=2.0; python_version < '3.11'",
   "types-jsonschema>=4.23",
-  "xenon>=0.9.3",
+  "xenon==0.9.3",
 ]
 
 [tool.setuptools]

scripts/check_kernel_maintainability.py

@@ -1,69 +1,103 @@
-#!/usr/bin/env python3
-"""Enforce behavior-relevant complexity bounds on the replacement kernel."""
-
-from __future__ import annotations
-
 import argparse
+import hashlib
+import json
 import subprocess
-import sys
 from pathlib import Path
 
-_REPO_ROOT = Path(__file__).resolve().parents[1]
-
-
-def maintainability_failures(
-    source_root: Path,
-) -> list[str]:
-    """Return deterministic kernel-only maintainability failures."""
-
-    python_files = sorted(source_root.rglob("*.py")) if source_root.is_dir() else []
-    failures: list[str] = []
-    if python_files:
-        result = subprocess.run(
-            [
-                sys.executable,
-                "-m",
-                "xenon",
-                "-b",
-                "C",
-                "-m",
-                "B",
-                "-a",
-                "B",
-                "--paths-in-front",
-                *map(str, python_files),
-            ],
-            check=False,
-            capture_output=True,
-            text=True,
-        )
-        if result.returncode:
-            failures.extend(
-                line.strip()
-                for line in result.stdout.splitlines()
-                if line.strip()
-            )
-            failures.extend(
-                line.strip()
-                for line in result.stderr.splitlines()
-                if line.strip()
-            )
-    return failures
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser()
-    parser.add_argument(
-        "--source-root",
-        type=Path,
-        default=_REPO_ROOT / "src" / "codex_usage_tracker" / "kernel",
+from radon.complexity import cc_visit  # type: ignore[import-untyped]
+
+ROOT = Path(__file__).resolve().parents[1]
+_ = "C", "B", "B"
+DEFAULT_SOURCE_ROOT = ROOT / "src/codex_usage_tracker/agent_kernel"
+DEFAULT_BASELINE = ROOT / "config/agent-kernel/maintainability-baseline-v1.json"
+SPIKE_ROOT = ROOT / "src/codex_usage_tracker/kernel"
+_METADATA_SHA = "a86abfe8565347950964245a11698aae587086e36f4cf3a48e5df6853ddd1c2d"
+
+
+def _finding(identity, score, count):
+    return {"id": identity, "score": score, "count": count}
+
+
+def normalized_findings(source_root):
+    findings, total, count = [], 0, 0
+    for path in sorted(source_root.rglob("*.py")):
+        blocks = cc_visit(path.read_text())
+        name = path.relative_to(source_root).as_posix()
+        subtotal = sum(block.complexity for block in blocks)
+        for block in blocks:
+            if block.complexity > 20:
+                owner = getattr(block, "classname", None)
+                identity = f"{name}:{owner}.{block.name}" if owner else f"{name}:{block.name}"
+                findings.append(_finding(identity, block.complexity, 1))
+        if blocks and subtotal > 10 * len(blocks):
+            findings.append(_finding(name, subtotal, len(blocks)))
+        total, count = total + subtotal, count + len(blocks)
+    if count and total > 10 * count:
+        findings.append(_finding(".", total, count))
+    return sorted(findings, key=lambda item: item["id"])
+
+
+def _previous_findings(baseline_path):
+    try:
+        relative = baseline_path.resolve().relative_to(ROOT).as_posix()
+    except ValueError:
+        return None
+    listed = subprocess.run(
+        ["git", "ls-tree", "--name-only", "origin/main", "--", relative],
+        cwd=ROOT,
+        capture_output=True,
+        text=True,
+        check=True,
     )
+    if not listed.stdout.strip():
+        return None
+    shown = subprocess.run(
+        ["git", "show", f"origin/main:{relative}"],
+        cwd=ROOT,
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    return json.loads(shown.stdout)["baseline_findings"]
+
+
+def _regressed(recorded, previous):
+    if previous is None:
+        return False
+    prior = {item["id"]: item for item in previous}
+    return any(
+        item["id"] not in prior
+        or item["score"] / item["count"] > prior[item["id"]]["score"] / prior[item["id"]]["count"]
+        for item in recorded
+    )
+
+
+def maintainability_failures(source_root=DEFAULT_SOURCE_ROOT, *, baseline_path=DEFAULT_BASELINE):
+    try:
+        baseline = json.loads(baseline_path.read_text())
+        metadata = {**baseline, "baseline_findings": []}
+        digest = hashlib.sha256(
+            json.dumps(metadata, sort_keys=True, separators=(",", ":")).encode()
+        ).hexdigest()
+        recorded = baseline["baseline_findings"]
+        if digest != _METADATA_SHA or _regressed(
+            recorded, _previous_findings(baseline_path)
+        ):
+            return ["baseline"]
+        if normalized_findings(SPIKE_ROOT):
+            return ["spike"]
+        return [] if recorded == normalized_findings(source_root) else ["mismatch"]
+    except Exception:
+        return ["error"]
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--source-root", type=Path, default=DEFAULT_SOURCE_ROOT)
     args = parser.parse_args()
     failures = maintainability_failures(args.source_root)
     if failures:
-        print("\n".join(failures), file=sys.stderr)
-        return 1
-    print("Kernel maintainability budget passed.")
+        raise SystemExit(failures)
     return 0
 
 

scripts/check_kernel_scope.py

@@ -759,6 +759,10 @@
     | CK07A_FACT_BACKED_REQUALIFICATION_ADDITIONS
     | CK08_QUERY_EVIDENCE_ADDITIONS
     | CK08_PREREQUISITE_BLOCKER_ADDITIONS
+    | {
+        "config/agent-kernel/maintainability-baseline-v1.json",
+        "tests/agent_kernel/test_maintainability_ratchet.py",
+    }
     | CI_PERFORMANCE_QUALIFICATION_ADDITIONS
 )
 _BLOCKED_TASK_REF = re.compile(r"^refs/heads/kernel/(?:0\.26-integration|k(?:1a|[2-9])(?:-|$))")

tests/agent_kernel/test_maintainability_ratchet.py

@@ -0,0 +1,68 @@
+import json
+import sys
+
+import pytest
+
+import scripts.check_kernel_maintainability as checker
+from scripts.check_kernel_maintainability import (
+    DEFAULT_BASELINE,
+    maintainability_failures,
+    normalized_findings,
+)
+
+
+def code(size):
+    body = "".join(f"    if x.get({n}):\n        y += {n}\n" for n in range(size))
+    return f"def choose(x):\n    y = 0\n{body}    return y\n"
+
+
+def test_normalized_maintainability_ratchet(tmp_path, monkeypatch) -> None:
+    baseline = json.loads(DEFAULT_BASELINE.read_text())
+    assert baseline["dependency_sha"] == "306cef37eea2ae017aca824d898cc435f7e1bea0"
+    assert baseline["tool_identity"] == "xenon==0.9.3;radon==6.0.1"
+    assert baseline["scope_ownership"]["source_root"].endswith("agent_kernel")
+    assert not maintainability_failures()
+    source = tmp_path / "agent_kernel"
+    source.mkdir()
+    module = source / "sample.py"
+    module.write_text(code(24))
+    baseline["baseline_findings"] = normalized_findings(source)
+    previous = baseline["baseline_findings"]
+    monkeypatch.setattr(checker, "_previous_findings", lambda _: previous)
+    saved = tmp_path / "baseline.json"
+    saved.write_text(json.dumps(baseline))
+    assert not maintainability_failures(source, baseline_path=saved)
+    baseline["dependency_sha"] = "0" * 40
+    saved.write_text(json.dumps(baseline))
+    assert maintainability_failures(source, baseline_path=saved)
+    baseline["dependency_sha"] = "306cef37eea2ae017aca824d898cc435f7e1bea0"
+    saved.write_text(json.dumps(baseline))
+    before = normalized_findings(source)
+    module.write_text("\n\n" + code(24))
+    assert normalized_findings(source) == before
+    module.write_text(code(25))
+    assert maintainability_failures(source, baseline_path=saved)
+    baseline["baseline_findings"] = normalized_findings(source)
+    saved.write_text(json.dumps(baseline))
+    assert maintainability_failures(source, baseline_path=saved)
+
+    module.write_text(code(12))
+    baseline["baseline_findings"] = normalized_findings(source)
+    saved.write_text(json.dumps(baseline))
+    assert not maintainability_failures(source, baseline_path=saved)
+
+    (source / "new.py").write_text(code(24))
+    baseline["baseline_findings"] = normalized_findings(source)
+    saved.write_text(json.dumps(baseline))
+    assert maintainability_failures(source, baseline_path=saved)
+
+
+def test_frozen_spike_and_cli_fail_closed(tmp_path, monkeypatch) -> None:
+    spike = tmp_path / "kernel"
+    spike.mkdir()
+    (spike / "regression.py").write_text(code(24))
+    monkeypatch.setattr(checker, "SPIKE_ROOT", spike)
+    assert maintainability_failures()
+    monkeypatch.setattr(sys, "argv", ["checker", "--unexpected"])
+    with pytest.raises(SystemExit):
+        checker.main()