fix(ci): reconcile qualification invariant owners

docs/architecture.md

@@ -175,17 +175,28 @@ claim, gate, monitor, quota, writeback, recovery, and terminal closure. See the
 The executable boundary policy lives in
 [`test_control_plane_import_boundaries.py`](../tests/architecture/test_control_plane_import_boundaries.py):
 
-- `test_control_plane_does_not_gain_outward_dependencies` rejects control-plane
-  imports of presentation, CLI, capability, or benchmark-adapter layers.
+- `test_control_plane_has_no_unreviewed_outward_dependencies` rejects
+  control-plane imports of presentation, CLI, capability, or benchmark-adapter
+  layers unless the existing maintainability policy records the exact edge,
+  its review reason, and its retirement plan.
 - `test_status_has_no_forbidden_outward_dependencies` rejects status imports of
   benchmark-adapter or presentation layers.
 - `test_quota_markdown_is_owned_by_the_presentation_layer` protects the renderer
   ownership and CLI composition boundary.
 
-These are zero-exception checks. The former quota-Markdown and status
-verifier-bootstrap edges have been removed; there is no remaining debt
-allowlist for these checks. Run the linked architecture test file to inspect
-current violations rather than maintaining an edge inventory in prose.
+The control-plane check shares
+[`REVIEWED_MAINTAINABILITY_EXCEPTIONS`](../loopx/canary/maintainability_ratchet.py)
+with the maintainability ratchet. Its retained legacy Todo writer bridge keeps
+the existing Explore completion-evidence guard until that writer retires.
+Exceptions match complete source/target identities; neighboring modules and
+targets remain rejected, and a retired edge makes its exception stale. This
+replaces the former unconditional rejection that contradicted the already
+reviewed migration debt. It adds no exception or import to the runtime.
+
+The status and quota-Markdown checks remain zero-exception: their former
+verifier-bootstrap and renderer edges have been removed. Run the linked
+architecture test file for current violations and the ratchet for reviewed
+debt, invalid or stale exceptions, and growth beyond unchanged ceilings.
 
 Adapter-specific enrichment belongs behind application/plugin composition
 rather than in the status core. Move an edge only after characterization

loopx/control_plane/goals/goal_frontier/__init__.py

@@ -4,7 +4,7 @@
 
 from ...agents.agent_scope import (
     agent_scope_blocking_handoff_gates,
-    agent_scope_count_advancement_items,
+    agent_scope_count_advancement_items as _count_advancement_items,
     agent_scope_item_claimed_by,
     agent_scope_item_claimed_by_agent_or_unclaimed,
 )
@@ -26,6 +26,7 @@
     TODO_SUCCESSION_WARNING_REASON_CODE,
     todo_succession_gap_items,
 )
+from ...todos.user_gate import open_todo_count as _open_todo_count
 from ...work_items.autonomous_replan_ack import (
     autonomous_replan_ack_matches_agent,
     autonomous_replan_ack_matches_frontier,
@@ -550,12 +551,6 @@ def build_vision_gap_judge(
     }
 
 
-def _open_todo_count(summary: dict[str, Any] | None) -> int:
-    if not isinstance(summary, dict):
-        return 0
-    return safe_non_negative_int(summary.get("open_count"))
-
-
 def _blocking_user_open_count(summary: dict[str, Any] | None) -> int:
     if not isinstance(summary, dict):
         return 0
@@ -581,10 +576,6 @@ def _todo_task_class(item: dict[str, Any]) -> str:
     return str(item.get("task_class") or "").strip()
 
 
-def _count_advancement_items(items: Any, *, claimed_by: str | None = None) -> int:
-    return agent_scope_count_advancement_items(items, claimed_by=claimed_by)
-
-
 def _summary_task_counts(summary: dict[str, Any] | None) -> dict[str, int]:
     open_count = _open_todo_count(summary)
     if not isinstance(summary, dict):

loopx/control_plane/work_items/stride_policy_codec.py

@@ -13,7 +13,7 @@
 
 @lru_cache(maxsize=1)
 def stride_strategy_options() -> tuple[str, ...]:
-    contract = json.loads(files("loopx.control_plane").joinpath("stride_strategy_schema.json").read_text())
+    contract = json.loads(files("loopx.control_plane").joinpath("stride_strategy_schema.json").read_text(encoding="utf-8"))
     return tuple(contract["enum"])
 
 

tests/architecture/test_control_plane_import_boundaries.py

@@ -7,6 +7,12 @@
 import subprocess
 import sys
 
+import pytest
+
+from loopx.canary.maintainability_ratchet import (
+    REVIEWED_MAINTAINABILITY_EXCEPTIONS,
+    evaluate_maintainability_findings,
+)
 
 REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
 PACKAGE_ROOT = REPOSITORY_ROOT / "loopx"
@@ -201,7 +207,19 @@ def _public_contract_evidence() -> dict[tuple[str, str], set[str]]:
     return evidence
 
 
-def test_control_plane_does_not_gain_outward_dependencies() -> None:
+def _review_outward_dependencies(edges, *, reviewed=REVIEWED_MAINTAINABILITY_EXCEPTIONS):
+    # Use the existing exact debt owner: the quarantined legacy writer retains
+    # its Explore completion guard until retirement. This is not a prefix waiver.
+    findings = [
+        {"id": f"dependency_debt:{source}->{target}", "category": "dependency_debt"}
+        for source, target in sorted(edges)
+    ]
+    return evaluate_maintainability_findings(findings, reviewed_exceptions={
+        key: value for key, value in reviewed.items() if key.startswith("dependency_debt:")
+    })
+
+
+def test_control_plane_has_no_unreviewed_outward_dependencies() -> None:
     outward_dependencies = {
         (_module_name(path), dependency)
         for path in CONTROL_PLANE_ROOT.rglob("*.py")
@@ -212,12 +230,45 @@ def test_control_plane_does_not_gain_outward_dependencies() -> None:
         )
     }
 
-    assert not outward_dependencies, (
-        "control-plane code must not depend on presentation, CLI, capability, or "
-        f"benchmark-adapter layers; unexpected edges: {sorted(outward_dependencies)}"
+    report = _review_outward_dependencies(outward_dependencies)
+    assert report["ok"], (
+        "control-plane outward dependencies require an exact reviewed exception "
+        f"with its retirement plan: {report}"
     )
 
 
+@pytest.mark.parametrize("edge", [
+    ("loopx.control_plane.todos.provider_update", "loopx.capabilities.explore.research_frontier"),
+    ("loopx.control_plane.todos.legacy_mutation_copy", "loopx.capabilities.explore.research_frontier"),
+    ("loopx.control_plane.todos.legacy_mutation", "loopx.capabilities.explore.research_frontier_extra"),
+    ("loopx.control_plane.todos.legacy_mutation", "loopx.capabilities.explore.research_frontier.child"),
+    ("loopx.control_plane.todos.legacy_mutation", "loopx.capabilities.issue_fix"),
+])
+def test_reviewed_legacy_bridge_never_allows_another_outward_edge(edge):
+    legacy = ("loopx.control_plane.todos.legacy_mutation", "loopx.capabilities.explore.research_frontier")
+    key = f"dependency_debt:{legacy[0]}->{legacy[1]}"
+    reviewed = {key: REVIEWED_MAINTAINABILITY_EXCEPTIONS[key]}
+    assert _review_outward_dependencies({legacy}, reviewed=reviewed)["ok"]
+    report = _review_outward_dependencies({legacy, edge}, reviewed=reviewed)
+    assert not report["ok"] and report["unreviewed_count"] == 1
+    assert report["unreviewed_findings"][0]["id"] == f"dependency_debt:{edge[0]}->{edge[1]}"
+
+
+@pytest.mark.parametrize("field", ["reason", "retirement_plan"])
+def test_outward_dependency_exception_requires_its_existing_review_evidence(field):
+    legacy = ("loopx.control_plane.todos.legacy_mutation", "loopx.capabilities.explore.research_frontier")
+    key = f"dependency_debt:{legacy[0]}->{legacy[1]}"
+    reviewed = {key: {**REVIEWED_MAINTAINABILITY_EXCEPTIONS[key], field: ""}}
+    report = _review_outward_dependencies({legacy}, reviewed=reviewed)
+    assert not report["ok"] and report["invalid_exception_count"] == 1
+
+
+def test_retired_outward_dependency_cannot_keep_a_stale_exception():
+    key = "dependency_debt:loopx.control_plane.todos.legacy_mutation->loopx.capabilities.explore.research_frontier"
+    report = _review_outward_dependencies(set(), reviewed={key: REVIEWED_MAINTAINABILITY_EXCEPTIONS[key]})
+    assert not report["ok"] and report["stale_exception_count"] == 1
+
+
 def test_periodic_report_hook_does_not_break_cold_imports() -> None:
     periodic_report_facade = (
         PACKAGE_ROOT / "capabilities" / "periodic_report" / "__init__.py"

tests/control_plane/test_advancement_frontier_classification.py

@@ -16,6 +16,37 @@ def item(identity, **fields):
     return {"todo_id": identity, "status": "open", "task_class": "advancement_task", **fields}
 
 
+@pytest.mark.parametrize("summary, expected", [
+    (None, 0), ([], 0), ({}, 0), ({"open_count": None}, 0),
+    ({"open_count": -2}, 0), ({"open_count": "bad"}, 0),
+    ({"open_count": []}, 0), ({"open_count": "12"}, 12),
+    ({"open_count": True}, 1), ({"open_count": 2.9}, 2),
+    ({"open_count": 9007199254740993}, 9007199254740993),
+])
+def test_frontier_open_counts_keep_legacy_read_compatibility(summary, expected):
+    from copy import deepcopy
+    from loopx.control_plane.goals import goal_frontier
+
+    before = deepcopy(summary)
+    assert goal_frontier._open_todo_count(summary) == expected
+    assert summary == before
+
+
+@pytest.mark.parametrize("claim, expected", [(None, 2), ("worker-a", 1),
+    ("worker-b", 0), ("__unclaimed__", 1)])
+def test_frontier_advancement_counts_keep_claim_and_terminal_boundaries(claim, expected):
+    from copy import deepcopy
+    from loopx.control_plane.goals import goal_frontier
+
+    rows = [item("todo_own", claimed_by="worker-a"), item("todo_free"),
+            item("todo_closed", status="done"), item("todo_marked_done", done=True),
+            item("todo_monitor", task_class="continuous_monitor"),
+            item("todo_deferred", status="deferred"), None]
+    before = deepcopy(rows)
+    assert goal_frontier._count_advancement_items(rows, claimed_by=claim) == expected
+    assert rows == before
+
+
 @pytest.mark.parametrize("agent", ["worker-a", None])
 def test_executable_slot_wins_and_peer_visibility_never_grants_selection(agent):
     source = [item("todo_own", claimed_by="worker-a"), item("todo_free"),