test: reconcile release regressions with typed planning owners

examples/control_plane/quota-cleared-blocker-successor-gate-smoke.py

@@ -17,10 +17,10 @@
 )
 from loopx.status import compact_todo_group  # noqa: E402
 from loopx.control_plane.todos.handoff_gate import (  # noqa: E402
-    build_todo_handoff_gate_lanes,
     build_todo_handoff_gate_states,
     handoff_ready_successor_todo_ids,
 )
+from loopx.control_plane.todos.quota_selection import project_quota_planning  # noqa: E402
 
 
 GOAL_ID = "cleared-blocker-successor-gate-fixture"
@@ -122,11 +122,15 @@ def assert_handoff_gate_lanes_and_ready_successors_are_todo_context_helpers() ->
     summary = agent_todo_summary([review_gate, successor, blocked_gate])
 
     assert handoff_ready_successor_todo_ids(summary) == {"todo_followup_slice"}, summary
-    lanes = build_todo_handoff_gate_lanes(
+    open_items = [item for item in summary["items"] if not item["done"]]
+    lanes = project_quota_planning(
         summary,
+        all_open_items=open_items,
+        source_open_count=len(open_items),
         agent_identity={"agent_id": BLOCKED_AGENT},
-        item_limit=10,
-    )
+        filter_user_gate_blocks_agent=False,
+        available_capabilities=[],
+    )["handoff_lanes"]
     assert lanes["handoff_gate_count"] == 2, lanes
     assert lanes["current_agent_handoff_gate_count"] == 2, lanes
     assert lanes["current_agent_cleared_without_successor_handoff_count"] == 1, lanes

examples/control_plane/quota-todo-summary-readmodel-smoke.py

@@ -14,11 +14,11 @@
 
 from loopx.control_plane.todos.quota_summary import (  # noqa: E402
     compact_quota_todo_summary_for_payload,
-    is_user_gate_todo_item,
     select_quota_todo_summary,
     summarize_project_asset_todos_for_quota,
     summarize_user_todos_for_quota,
 )
+from loopx.control_plane.todos.user_gate import is_user_gate_todo_item  # noqa: E402
 
 
 AGENT_IDENTITY = {"agent_id": "codex-product-capability"}

examples/control_plane/todo-route-continuation-lanes-smoke.py

@@ -11,10 +11,7 @@
 if str(REPO_ROOT) not in sys.path:
     sys.path.insert(0, str(REPO_ROOT))
 
-from loopx.control_plane.todos.route_continuation import (  # noqa: E402
-    TODO_ROUTE_CONTINUATION_SELECTION_POLICY,
-    build_todo_route_continuation_lanes,
-)
+from loopx.control_plane.todos.quota_selection import project_quota_planning  # noqa: E402
 
 
 def assert_route_continuation_lanes_filter_current_unclaimed_and_other_agents() -> None:
@@ -79,11 +76,14 @@ def assert_route_continuation_lanes_filter_current_unclaimed_and_other_agents()
         ],
     }
 
-    lanes = build_todo_route_continuation_lanes(
+    lanes = project_quota_planning(
         summary,
+        all_open_items=[],
+        source_open_count=0,
         agent_identity={"agent_id": "codex-side-bypass"},
-        item_limit=10,
-    )
+        filter_user_gate_blocks_agent=False,
+        available_capabilities=[],
+    )["route_lanes"]
     assert lanes["route_continuation_replan_count"] == 4, lanes
     assert [
         item.get("route_id") or item.get("route_key")
@@ -113,7 +113,8 @@ def assert_route_continuation_lanes_filter_current_unclaimed_and_other_agents()
     assert current["decision_scope"]["scope_key"] == "route", current
     assert (
         lanes["route_continuation_replan_selection_policy"]
-        == TODO_ROUTE_CONTINUATION_SELECTION_POLICY
+        == "quota may wake the current peer for route continuation replan "
+        "candidates claimed by that agent or unclaimed; other-agent route candidates remain diagnostic visibility"
     ), lanes
 
 

examples/control_plane/todo-user-gate-readmodel-smoke.py

@@ -12,12 +12,12 @@
     sys.path.insert(0, str(REPO_ROOT))
 
 from loopx.control_plane.todos.quota_summary import (  # noqa: E402
-    is_user_gate_todo_item,
     summarize_user_todos_for_quota,
 )
 from loopx.control_plane.todos.user_gate import (  # noqa: E402
     build_gate_prompt,
     has_open_user_gate_todo,
+    is_user_gate_todo_item,
     open_todo_count,
     open_user_gate_todo_items,
     should_notify_user_on_open_todo,

tests/control_plane/test_cli_output_budget.py

@@ -1196,10 +1196,11 @@ def test_quota_cli_keeps_required_authoring_inline_and_diagnostic_audit_cold(
 
     assert default_exit_code == 0, default_text
     assert detail_exit_code == 0, detail_text
-    # Same 36-Todo / 12-run command: 41,176 -> 44,225 chars (+7.4%).
-    # Required authoring replaces a second diagnostic read; preserve its rules
-    # and field limits. This fixed allowance does not relax other lane budgets.
-    assert len(default_text) <= 45_000
+    # Same stable 36-Todo / 12-run command: 44,894 -> 45,249 chars. The added
+    # automation-cadence readback exposes the owning eligibility condition;
+    # retain it and inline authoring with 751 chars of measured headroom. This
+    # fixed 46,000-char allowance does not relax other lane/scale budgets.
+    assert len(default_text) <= 46_000
     default_payload = json.loads(default_text)
     detail_payload = json.loads(detail_text)
     compact_audit = default_payload["vision_continuation_audit"]

tests/control_plane/test_delivery_response.py

@@ -93,7 +93,6 @@ def test_status_compaction_preserves_binding_and_all_consumers_defer_to_current_
 @pytest.mark.parametrize("resume, patch", [
     ("monitor_changed:todo_dependency", {"baseline_generation": 1}),
     ("capacity_available:network", {"capability": "other"}),
-    ("pr_merged:#1", {"pr_number": 2}),
     # The scheduled instant is one day ahead of the injected evaluation clock,
     # so the condition stays unsatisfied no matter when CI runs.
     (f"resume_at:{RESUME_AT}", {"clock_provider": "other"}),
@@ -117,6 +116,30 @@ def test_real_resume_projection_identity_survives_python_transport(resume, patch
     assert project_delivery_response(blocked_run(), summary)["reason"] == "history_supervision"
 
 
+@pytest.mark.parametrize("resume", ["pr_merged:#1", "pr_merged:example/project#1"])
+def test_unobserved_pr_binding_cannot_prove_a_pending_delivery_wait(resume):
+    todo = {**waiting_todo(), "resume_when": resume,
+            "task_repository": "git:github.com/example/project"}
+    condition = evaluate_todo_resume_conditions(
+        [todo], source_items=[], available_capabilities=[],
+        evaluated_at=EVALUATED_AT,
+    )[todo["todo_id"]]
+    assert condition["pr_repo"] == "example/project"
+    assert condition["pr_number"] == 1
+    assert condition["satisfied"] is False
+    summary = quota_todo_summary([todo], claim_scope_agent_id="agent-a",
+                                evaluated_at=EVALUATED_AT)
+    for key in TODO_PLANNING_SOURCE_KEYS:
+        for row in summary.get(key, []):
+            if row["todo_id"] == todo["todo_id"]:
+                row["resume_condition"] = condition
+    # Repo/number identify what to observe. Missing merge evidence does not
+    # establish that the PR is pending or relax stalled-delivery supervision.
+    assert project_delivery_response(blocked_run(), summary)["reason"] == "history_supervision"
+    condition.update({"pr_repo": "example/other", "pr_number": 2})
+    assert project_delivery_response(blocked_run(), summary)["reason"] == "history_supervision"
+
+
 def _resume_at_projection(evaluated_at: str, tamper_clock: bool = False) -> tuple[bool, str]:
     """Evaluate one `resume_at` wait at an explicit instant and project it."""
 

tests/control_plane/test_monitor_observation_admission.py

@@ -378,7 +378,15 @@ def test_completed_advancement_retains_auxiliary_monitor_admission(
     assert resume["identity"]["turn_instance_id"] == turn
     assert resume["grants_new_delivery"] is False
     assert resume["progress_ref"] == "$.settlement_progress"
-    assert len(json.dumps(resume, ensure_ascii=False, indent=2)) < 2_000
+    # Same stable fixture: v1.3.1 writeback was 1,831 chars; the bounded vision
+    # authoring contract and unavailable-monitor closeout now require 5,328.
+    # Keep useful decision guidance, with 672 chars of measured headroom. Spend
+    # and settled packets retain the original 2,000-character regression budget.
+    resume_budget = (
+        6_000 if (resume["next_step"] or {}).get("kind") == "durable_writeback"
+        else 2_000
+    )
+    assert len(json.dumps(resume, ensure_ascii=False, indent=2)) < resume_budget
     after_cli = poll["after"]["interaction_contract"]["cli_channel"]
     assert after_cli["settlement_resume_ref"] == "$.settlement_resume"
     assert after_cli["spend_after_validation"] is False

tests/control_plane/test_public_safety_credential_caller_faces.py

@@ -256,10 +256,6 @@ def test_the_named_policy_lists_both_credential_categories() -> None:
     # verdict: direction 2 asks for an explicit, tested short-assignment policy and
     # does not say which way, so this face stays where it is until that is decided.
     "loopx/extensions/presentation.py": "short-assignment verdict undecided",
-    # Beyond the labels this rule also names vendor forms the owner does not
-    # (`ak`/`sk`, `access_key_id`, `secret_key`). Route those to the owner first,
-    # then this site is a plain policy call.
-    "loopx/control_plane/todos/handoff_note.py": "names vendor forms the owner lacks",
     # The public-boundary contract is the publication tier: it refuses a label with
     # no value at all -- a trailing scheme word, an assignment operator with an
     # empty right side -- which is exactly what the owner's publication helper
@@ -318,8 +314,8 @@ def _credential_alternations_outside_the_owner() -> list[str]:
     The criterion is a credential alternation that also reaches for whitespace --
     three or more labels joined by ``|``, plus ``\\s``. A rule with no ``\\s`` is
     matching an environment-variable name or a field name, which is a different
-    question with a different owner; three such rules exist today and are named in
-    the PR body rather than silently excluded.
+    question with a different owner; two such rules remain explicitly declared
+    rather than silently excluded.
     """
 
     offenders: list[str] = []