fix(control-plane): restore current quota contracts

loopx/control_plane/quota/should_run_prepare.py

@@ -30,6 +30,7 @@
 from ..goals.goal_frontier import (
     build_goal_frontier_projection_context_from_status,
 )
+from ..quota.blocked_transition_notice import build_blocked_transition_notice
 from ..quota.error_codes import HeartbeatReceiptIdentityConflictError
 from ..agents.capability_memory import resolve_agent_capabilities
 from ..quota.goal_boundary import (
@@ -229,6 +230,7 @@ def _blocked_priority_fallback(
         return None
 
     blocked_items: list[dict[str, Any]] = []
+    transition_notices: list[dict[str, Any]] = []
     owner_visible_blocker = False
     for item in first_open:
         if not isinstance(item, dict):
@@ -270,6 +272,12 @@ def _blocked_priority_fallback(
             status == TODO_STATUS_BLOCKED or resume_condition_pending
         ):
             owner_visible_blocker = True
+            notice = build_blocked_transition_notice(
+                item,
+                selected_executable=selected,
+            )
+            if notice is not None:
+                transition_notices.append(notice)
 
     if not blocked_items:
         return None
@@ -294,6 +302,7 @@ def _blocked_priority_fallback(
             )
         ),
         "blocked_items": blocked_items[:3],
+        "blocked_transition_notices": transition_notices[:3],
         "selected_executable": selected_item,
         "recommended_action": (
             "Keep the blocked core todo visible in status while selecting fallback; "

tests/control_plane/test_selection_replan_reentry.py

@@ -1,4 +1,5 @@
 """A deferred selection must expose a runnable recovery before settlement."""
+import json
 import shlex
 
 import pytest
@@ -89,6 +90,45 @@ def test_deferred_selection_recovers_same_turn_and_settles_once(tmp_path, bindin
     cli = resumed["interaction_contract"]["cli_channel"]
     assert cli["settlement_plan"]["identity"] == identity
     refresh = next(c for c in cli["next_cli_actions"] if "refresh-state" in c)
+    if binding == "todo":
+        decision = tmp_path / "selection-replan-vision.json"
+        decision.write_text(
+            json.dumps(
+                {
+                    "schema_version": "goal_vision_replan_contract_v0",
+                    "state": "vision_patch_proposed",
+                    "vision_patch": {
+                        "vision_summary": (
+                            "Validate the existing bounded slices in dependency order."
+                        ),
+                        "acceptance_summary": (
+                            "Each slice has independent validation before dependent "
+                            "work proceeds."
+                        ),
+                        "advancement_policy": "as_needed",
+                    },
+                    "path_delta": {
+                        "schema_version": "goal_path_delta_v0",
+                        "outcome": "replan",
+                        "prior_assumption": (
+                            "The long chain needed a bounded review."
+                        ),
+                        "observed_reality": (
+                            "The reviewed chain has a runnable validation slice."
+                        ),
+                        "retained": ["Existing acceptance boundaries"],
+                        "changed": ["Proceed with the first validation slice"],
+                        "evidence_refs": ["evidence:selection-replan"],
+                    },
+                }
+            )
+            + "\n",
+            encoding="utf-8",
+        )
+        refresh = refresh.replace(
+            "<path-to-evidence-linked-goal-vision-replan-contract-v0.json>",
+            str(decision),
+        )
     for key, value in {"<advanced|blocked|exploration_exhausted|no_followup>": "advanced",
                        "<surface-id>": "accepted-artifact", "<hypothesis-id>": "adoption",
                        "<probe-kind>": "acceptance", "<evidence-id>": "evidence:readback"}.items():

tests/control_plane/test_settled_replay_construction.py

@@ -44,11 +44,16 @@ def unexpected(*args, **kwargs):
         assert payload[flag] is False, flag
     assert payload["execution_obligation"]["must_attempt_work"] is False
     assert payload["heartbeat_recommendation"]["agent_must_attempt"] is False
-    for field in ("selected_todo", "replan_action_packet", "autonomous_replan_obligation", "action_portfolio"):
+    for field in (
+        "selected_todo",
+        "replan_action_packet",
+        "autonomous_replan_obligation",
+        "action_portfolio",
+        "protocol_action_packet",
+    ):
         assert field not in payload
     assert payload["interaction_contract"]["agent_channel"]["must_attempt"] is False
     assert payload["interaction_contract"]["cli_channel"]["spend_after_validation"] is False
-    assert payload["protocol_action_packet"]["summary"]
 
 
 def test_pause_still_precedes_settled_replay() -> None: