fix(canary): enforce module ceiling settlement in the same diff as growth

loopx/canary/maintainability_ratchet.py

@@ -612,6 +612,83 @@ def collect_module_metric_findings(
     return sorted(findings, key=lambda item: str(item["id"]))
 
 
+def _git_show_text(repository_root: Path, rev: str, path: str) -> str | None:
+    result = subprocess.run(
+        ["git", "show", f"{rev}:{path}"],
+        cwd=repository_root, capture_output=True,
+        text=True, encoding="utf-8", errors="replace", check=False,
+    )
+    return result.stdout if result.returncode == 0 else None
+
+
+def _rev_baseline_ceilings(repository_root: Path, rev: str) -> dict[str, dict[str, int]]:
+    """Mirror ``module_metric_baseline`` for a committed revision."""
+    text = _git_show_text(repository_root, rev, "loopx/canary/module_metric_baseline.json")
+    if text is None:
+        return {}
+    payload = json.loads(text)
+    ceilings = payload.get("module_metric_ceilings")
+    if not isinstance(ceilings, dict):
+        return {}
+    return {
+        str(path): {key: int(metrics[key]) for key in ("lines", "any_count", "dict_any_count") if key in metrics}
+        for path, metrics in ceilings.items()
+        if isinstance(metrics, dict)
+    }
+
+
+def diff_scoped_module_ceiling_violations(
+    repository_root: Path,
+    changed_files: Sequence[str],
+    *,
+    base_ref: str,
+) -> list[dict[str, Any]]:
+    """Module growth that crossed a reviewed ceiling must settle in the same diff.
+
+    A module may exceed its pre-diff ceiling without this check firing, as long
+    as this diff did not cause the crossing; and a module whose ceiling this diff
+    raises can stay silent. The single case this flags is the one that previously
+    merged and turned ``main`` red until a separate reconciliation PR refreshed
+    the ledger: this diff grew a module past the ceiling it inherited, without
+    settling that ceiling here.
+    """
+    head_ceilings = module_metric_baseline(
+        repository_root / "loopx" / "canary" / "module_metric_baseline.json"
+    )
+    base_ceilings = _rev_baseline_ceilings(repository_root, (base_ref or "origin/main").strip() or "origin/main")
+    violations: list[dict[str, Any]] = []
+    for changed in changed_files:
+        relative = str(changed)
+        if not relative.startswith("loopx/") or not relative.endswith(".py"):
+            continue
+        checkout_path = repository_root / relative
+        if not checkout_path.is_file():
+            continue
+        head_lines = module_metrics(checkout_path)["lines"]
+        base_lines = _git_show_text(repository_root, (base_ref or "origin/main").strip() or "origin/main", relative)
+        if base_lines is None:
+            base_ceiling = MODULE_LINE_LIMIT
+            was_within_budget = True
+        else:
+            base_ceiling = base_ceilings.get(relative, {}).get("lines", MODULE_LINE_LIMIT)
+            was_within_budget = len(base_lines.splitlines()) <= base_ceiling
+        head_ceiling = head_ceilings.get(relative, {}).get("lines", MODULE_LINE_LIMIT)
+        crossed_inherited_ceiling = was_within_budget and head_lines > base_ceiling
+        if crossed_inherited_ceiling and head_ceiling < head_lines:
+            violations.append(
+                {
+                    "id": _finding_id("module_metric_budget", relative),
+                    "category": "module_metric_budget",
+                    "path": relative,
+                    "base_lines": 0 if base_lines is None else len(base_lines.splitlines()),
+                    "base_ceiling": base_ceiling,
+                    "head_lines": head_lines,
+                    "head_ceiling": head_ceiling,
+                }
+            )
+    return sorted(violations, key=lambda item: str(item["id"]))
+
+
 def evaluate_maintainability_findings(
     findings: Sequence[Mapping[str, Any]],
     *,

loopx/canary/premerge.py

@@ -502,6 +502,55 @@ def _diff_hygiene_checks(
     return checks
 
 
+def _module_ceiling_colocation_check(
+    *,
+    changed_files: list[str],
+    base_ref: str,
+    execute: bool,
+    repo_root: Path = REPO_ROOT,
+) -> dict[str, Any] | None:
+    python_files = [
+        str(path) for path in changed_files
+        if str(path).startswith("loopx/")
+        and str(path).endswith(".py")
+        and (repo_root / str(path)).is_file()
+    ]
+    if not python_files:
+        return None
+    base = (base_ref or "origin/main").strip() or "origin/main"
+    if not execute:
+        return {
+            "id": "module_ceiling_colocation",
+            "kind": "direct_import",
+            "command": "python3 examples/control_plane/control-plane-maintainability-ratchet-smoke.py",
+            "reason": "module growth that crossed its reviewed ceiling must settle in the same diff",
+            "status": "ready",
+            "ok": True,
+        }
+    from loopx.canary.maintainability_ratchet import diff_scoped_module_ceiling_violations
+
+    violations = diff_scoped_module_ceiling_violations(
+        repo_root, python_files, base_ref=base
+    )
+    ok = not violations
+    check: dict[str, Any] = {
+        "id": "module_ceiling_colocation",
+        "kind": "direct_import",
+        "command": "python3 examples/control_plane/control-plane-maintainability-ratchet-smoke.py",
+        "reason": "module growth that crossed its reviewed ceiling must settle in the same diff",
+        "status": "passed" if ok else "failed",
+        "ok": ok,
+    }
+    if violations:
+        check["detail"] = [
+            f"{item['path']}: grew {item['base_lines']} -> {item['head_lines']} lines "
+            f"past its inherited ceiling {item['base_ceiling']}; settle the ceiling in "
+            f"this diff (loopx/canary/module_metric_baseline.json)"
+            for item in violations
+        ]
+    return check
+
+
 def _py_compile_check(
     *,
     python_files: list[str],
@@ -805,6 +854,15 @@ def build_premerge_validation_gate(
     if py_compile is not None:
         direct_checks.append(py_compile)
 
+    module_colocation = _module_ceiling_colocation_check(
+        changed_files=files,
+        base_ref=base_ref,
+        execute=execute,
+        repo_root=target_repo_root,
+    )
+    if module_colocation is not None:
+        direct_checks.append(module_colocation)
+
     if files:
         catalog_progress = _section_progress_callback(
             progress_callback,

tests/canary/test_maintainability_ratchet.py

@@ -1,13 +1,17 @@
 from __future__ import annotations
 
+import json
+import os
 from pathlib import Path
+import subprocess
 
 from loopx.canary.maintainability_ratchet import (
     MODULE_METRIC_BASELINE_SCHEMA_VERSION,
     build_control_plane_maintainability_report,
     collect_dependency_debt,
     collect_module_metric_findings,
     collect_oversized_decision_functions,
+    diff_scoped_module_ceiling_violations,
     evaluate_maintainability_findings,
     module_metrics,
     render_control_plane_maintainability_report,
@@ -337,3 +341,111 @@ def test_decision_ratchet_covers_quota_cli_orchestration(tmp_path: Path) -> None
             },
         }
     ]
+
+
+def _git(repo: Path, args: list[str]) -> None:
+    subprocess.run(
+        ["git", "-C", str(repo), *args],
+        check=True,
+        capture_output=True,
+        text=True,
+        env={
+            **os.environ,
+            "GIT_AUTHOR_NAME": "t",
+            "GIT_AUTHOR_EMAIL": "t@example.com",
+            "GIT_COMMITTER_NAME": "t",
+            "GIT_COMMITTER_EMAIL": "t@example.com",
+        },
+    )
+
+
+def _write_baseline(repo: Path, ceilings: dict[str, dict[str, int]]) -> None:
+    (repo / "loopx" / "canary" / "module_metric_baseline.json").write_text(
+        json.dumps(
+            {
+                "schema_version": MODULE_METRIC_BASELINE_SCHEMA_VERSION,
+                "default_limits": {"lines": 1500, "any_count": 300, "dict_any_count": 300},
+                "module_metric_ceilings": ceilings,
+            }
+        ),
+        encoding="utf-8",
+    )
+
+
+def test_module_ceiling_growth_must_settle_in_the_same_diff(tmp_path: Path) -> None:
+    repo = tmp_path / "repo"
+    (repo / "loopx" / "canary").mkdir(parents=True)
+    (repo / "loopx" / "grown.py").write_text("\n".join(["# x"] * 5) + "\n", encoding="utf-8")
+    _write_baseline(
+        repo,
+        {"loopx/grown.py": {"lines": 10, "any_count": 0, "dict_any_count": 0}},
+    )
+    _git(repo, ["init", "-q"])
+    _git(repo, ["add", "."])
+    _git(repo, ["commit", "-q", "-m", "base"])
+
+    # Grow past the inherited ceiling (10) without settling the ledger here.
+    (repo / "loopx" / "grown.py").write_text("\n".join(["# x"] * 20) + "\n", encoding="utf-8")
+    violations = diff_scoped_module_ceiling_violations(
+        repo, ["loopx/grown.py"], base_ref="HEAD"
+    )
+    assert [item["path"] for item in violations] == ["loopx/grown.py"]
+    assert violations[0]["base_lines"] == 5
+    assert violations[0]["base_ceiling"] == 10
+    assert violations[0]["head_lines"] == 20
+
+    # Settling the ceiling in the same tree clears the violation.
+    _write_baseline(
+        repo,
+        {"loopx/grown.py": {"lines": 20, "any_count": 0, "dict_any_count": 0}},
+    )
+    assert (
+        diff_scoped_module_ceiling_violations(
+            repo, ["loopx/grown.py"], base_ref="HEAD"
+        )
+        == []
+    )
+
+
+def test_module_ceiling_growth_below_ceiling_does_not_require_settlement(
+    tmp_path: Path,
+) -> None:
+    repo = tmp_path / "repo"
+    (repo / "loopx" / "canary").mkdir(parents=True)
+    (repo / "loopx" / "grown.py").write_text("\n".join(["# x"] * 5) + "\n", encoding="utf-8")
+    _write_baseline(
+        repo,
+        {"loopx/grown.py": {"lines": 100, "any_count": 0, "dict_any_count": 0}},
+    )
+    _git(repo, ["init", "-q"])
+    _git(repo, ["add", "."])
+    _git(repo, ["commit", "-q", "-m", "base"])
+
+    # Growth that stays under the inherited ceiling needs no ledger change.
+    (repo / "loopx" / "grown.py").write_text("\n".join(["# x"] * 50) + "\n", encoding="utf-8")
+    assert (
+        diff_scoped_module_ceiling_violations(
+            repo, ["loopx/grown.py"], base_ref="HEAD"
+        )
+        == []
+    )
+
+
+def test_module_ceiling_colocation_ignores_non_python_and_deleted_files(
+    tmp_path: Path,
+) -> None:
+    repo = tmp_path / "repo"
+    (repo / "loopx" / "canary").mkdir(parents=True)
+    _write_baseline(repo, {})
+    _git(repo, ["init", "-q"])
+    _git(repo, ["add", "."])
+    _git(repo, ["commit", "-q", "-m", "base"])
+
+    assert (
+        diff_scoped_module_ceiling_violations(
+            repo,
+            ["loopx/README.md", "loopx/removed.py", "docs/notes.py"],
+            base_ref="HEAD",
+        )
+        == []
+    )