fix(quota): preserve gated successor replans

examples/control_plane/quota-replan-decision-plane-smoke.py

@@ -629,6 +629,8 @@ def assert_replan_beats_monitor_quiet_skip() -> None:
     assert guard["goal_frontier_projection"]["deferred_successors"] == {
         "ready_count": 0,
         "blocked_count": 0,
+        "unknown_count": 0,
+        "counts_complete": True,
         "current_agent_ready_count": 0,
         "ready_todo_ids": [],
     }, guard

loopx/control_plane/agents/agent_lane_recommendation.py

@@ -496,6 +496,11 @@ def build_agent_lane_next_action(
                 if not claimed_by:
                     payload["claim_required_before_work"] = True
                 return payload
+            # The scoped gate selector is authoritative for this crossing. A
+            # ready deferred successor intentionally yields no delivery action
+            # so the frontier can project successor_replan_required. Falling
+            # through here could reselect the exact Todo covered by the gate.
+            return None
 
     candidate_sources: list[tuple[str, list[Any]]] = []
     # An empty projected list is authoritative; falling back would bypass the gate.

loopx/control_plane/quota/should_run.py

@@ -14,6 +14,7 @@
 from ..agents.agent_scope import (
     _attach_agent_identity_contracts,
 )
+from ..agents.agent_scope_frontier import AgentScopeFrontierAction
 from ..agents.identity import (
     build_quota_agent_identity,
 )
@@ -82,7 +83,11 @@ def _apply_selected_todo_guards(
 
     # Replan owns this Turn without selecting a delivery Todo. Both guard
     # helpers can fall back to a recommendation when given no selection.
-    if route.replan_decision_allowed:
+    if (
+        route.replan_decision_allowed
+        or route.effective_action
+        == AgentScopeFrontierAction.SUCCESSOR_REPLAN_REQUIRED.value
+    ):
         return route
 
     selected_todo = selected_todo_projection(

loopx/control_plane/quota/should_run_packet.py

@@ -33,7 +33,10 @@
     _attach_agent_identity_contracts,
 )
 from ..agents.capability_gate import missing_required_capabilities
-from ..agents.agent_scope_frontier import read_frontier_action
+from ..agents.agent_scope_frontier import (
+    build_agent_scope_frontier_payload,
+    read_frontier_action,
+)
 from ..goals.goal_frontier import (
     AUTONOMOUS_REPLAN_REQUIRED_MODE,
 )
@@ -694,6 +697,54 @@ def _resolve_agent_lane_delivery_route(
     return selected_action
 
 
+def _scoped_gate_deferred_replan_frontier(
+    prepared: _QuotaDecisionPreparation,
+) -> dict[str, Any] | None:
+    """Keep an authoritative gate fallback in the replan lane.
+
+    The fallback selector may choose a ready deferred successor while an open
+    advancement remains covered by the gate. That choice is a lifecycle replan,
+    not permission to fall through to the gated advancement candidate.
+    """
+
+    fallback = prepared.scoped_user_gate_fallback
+    selected = (
+        fallback.get("selected_executable")
+        if isinstance(fallback, dict)
+        else None
+    )
+    agent_id = normalize_todo_claimed_by(
+        (prepared.agent_identity or {}).get("agent_id")
+    )
+    if (
+        not agent_id
+        or not isinstance(selected, dict)
+        or selected.get("fallback_kind") != "deferred_successor_replan"
+    ):
+        return None
+    todo_id = normalize_todo_id(selected.get("todo_id")) or "<todo_id>"
+    return build_agent_scope_frontier_payload(
+        agent_id=agent_id,
+        action=AgentScopeFrontierAction.SUCCESSOR_REPLAN_REQUIRED,
+        quiet_noop_allowed=False,
+        spend_policy="spend once after validated successor replan/todo writeback",
+        reason=(
+            f"scoped user gate fallback selected ready deferred successor {todo_id}; "
+            "resolve its lifecycle before selecting another advancement"
+        ),
+        recommended_action=(
+            "Run a bounded successor replan: reopen, supersede, or record a "
+            f"no-follow-up rationale for {todo_id}."
+        ),
+        requires_replan=True,
+        candidate_counts={"deferred_resume_candidate_count": 1},
+        extra_fields={
+            "deferred_resume_candidates": [selected],
+            "scoped_user_gate_fallback": True,
+        },
+    )
+
+
 @dataclass(slots=True)
 class _QuotaDecisionRoute:
     normal_delivery_allowed: bool
@@ -1089,15 +1140,18 @@ def _resolve_quota_should_run_route(
             selected_recommended_action,
             agent_lane_next_action=agent_lane_next_action,
         )
-        agent_scope_frontier = _agent_scope_no_candidate_frontier(
-            agent_identity=prepared.agent_identity,
-            agent_todo_summary=prepared.agent_todo_summary,
-            agent_lane_next_action=agent_lane_next_action,
-            work_lane_contract=prepared.work_lane_contract,
-            candidate_should_run=bool(
-                (should_run and normal_delivery_allowed)
-                or ready_deferred_resume_candidates
-            ),
+        agent_scope_frontier = (
+            _scoped_gate_deferred_replan_frontier(prepared)
+            or _agent_scope_no_candidate_frontier(
+                agent_identity=prepared.agent_identity,
+                agent_todo_summary=prepared.agent_todo_summary,
+                agent_lane_next_action=agent_lane_next_action,
+                work_lane_contract=prepared.work_lane_contract,
+                candidate_should_run=bool(
+                    (should_run and normal_delivery_allowed)
+                    or ready_deferred_resume_candidates
+                ),
+            )
         )
         if (
             isinstance(agent_scope_frontier, dict)

loopx/control_plane/work_items/work_lane.py

@@ -61,6 +61,11 @@ def observe_work_lane(
 WORK_LANE_TODO_ITEM_FIELDS = (
     "index",
     "text",
+    # Preserve the source identity across lane rewrites. The selected-work
+    # readback compares this revision with the canonical Todo before granting
+    # delivery; dropping it makes an otherwise readable due Monitor fail
+    # closed as context_source_unavailable.
+    "content_revision",
     "todo_id",
     "status",
     "priority",

loopx/semantics/vocabulary_v0.json

@@ -362,7 +362,8 @@
         "loopx/control_plane/agents/agent_scope.py::_monitor_blocked_resume_frontier",
         "loopx/control_plane/agents/agent_scope.py::_other_agent_or_exhausted_frontier",
         "loopx/control_plane/agents/agent_scope.py::_route_continuation_frontier",
-        "loopx/control_plane/agents/agent_scope.py::_selected_candidate_priority_frontier"
+        "loopx/control_plane/agents/agent_scope.py::_selected_candidate_priority_frontier",
+        "loopx/control_plane/quota/should_run_packet.py::_scoped_gate_deferred_replan_frontier"
       ],
       "call_producers": {
         "loopx/control_plane/agents/agent_scope_frontier.py::build_agent_scope_frontier_payload": [

tests/control_plane/test_user_gate_lane_progress.py

@@ -59,7 +59,12 @@ def test_user_gate_fallback_preserves_repair_and_replan_precedence():
     ) is original
 
 
-def _status_payload(*, gate_action_kind: str, blocks_deferred: bool = False) -> dict:
+def _status_payload(
+    *,
+    gate_action_kind: str,
+    blocks_deferred: bool = False,
+    include_gated_agent: bool = False,
+) -> dict:
     completed = quota_todo_item(
         todo_id="todo_prerequisite",
         status="done",
@@ -75,8 +80,16 @@ def _status_payload(*, gate_action_kind: str, blocks_deferred: bool = False) ->
         resume_when="todo_done:todo_prerequisite",
         done=True,
     )
+    gated_agent = quota_todo_item(
+        todo_id="todo_first_screen",
+        status="open",
+        text="[P0] Publish the gated product first screen.",
+        claimed_by=AGENT_ID,
+        action_kind="publish_product_first_screen",
+        required_write_scopes=["docs/**"],
+    )
     agent_todos = quota_todo_summary(
-        [completed, deferred],
+        [completed, deferred, *([gated_agent] if include_gated_agent else [])],
         role="agent",
         claim_scope_agent_id=AGENT_ID,
     )
@@ -139,7 +152,10 @@ def _quality_vision_run() -> dict:
 
 def test_unrelated_user_gate_allows_ready_deferred_successor_replan() -> None:
     payload = build_quota_should_run(
-        _status_payload(gate_action_kind="approve_product_first_screen"),
+        _status_payload(
+            gate_action_kind="approve_product_first_screen",
+            include_gated_agent=True,
+        ),
         goal_id=GOAL_ID,
         agent_id=AGENT_ID,
         scheduler_execution_context=APP_CONTEXT,
@@ -149,6 +165,9 @@ def test_unrelated_user_gate_allows_ready_deferred_successor_replan() -> None:
     selected = fallback["selected_executable"]
     assert selected["todo_id"] == "todo_ready_deferred"
     assert selected["fallback_kind"] == "deferred_successor_replan"
+    assert payload["decision"] == "successor_replan_required"
+    assert payload["effective_action"] == "successor_replan_required"
+    assert payload["selected_todo"]["todo_id"] == "todo_ready_deferred"
     assert payload["interaction_contract"]["mode"] == "scoped_user_gate_fallback"
     assert payload["interaction_contract"]["agent_channel"]["must_attempt"] is True
     assert payload["interaction_contract"]["agent_channel"]["delivery_allowed"] is True
@@ -175,6 +194,45 @@ def test_unrelated_user_gate_allows_ready_deferred_successor_replan() -> None:
     }
 
 
+def test_ready_deferred_successor_replan_preempts_workspace_guard(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    guarded_todo_ids: list[str | None] = []
+
+    def _record_workspace_guard_selection(
+        *args: object,
+        selected_todo: dict | None = None,
+        **kwargs: object,
+    ) -> dict:
+        guarded_todo_ids.append((selected_todo or {}).get("todo_id"))
+        return {
+            "schema_version": "agent_workspace_guard_v1",
+            "reason": "delivery would require another workspace",
+            "required_action": "move to the delivery workspace",
+        }
+
+    monkeypatch.setattr(
+        "loopx.control_plane.quota.should_run.build_agent_workspace_guard",
+        _record_workspace_guard_selection,
+    )
+
+    payload = build_quota_should_run(
+        _status_payload(
+            gate_action_kind="approve_product_first_screen",
+            include_gated_agent=True,
+        ),
+        goal_id=GOAL_ID,
+        agent_id=AGENT_ID,
+        scheduler_execution_context=APP_CONTEXT,
+    )
+
+    assert payload["decision"] == "successor_replan_required"
+    assert payload["effective_action"] == "successor_replan_required"
+    assert payload["selected_todo"]["todo_id"] == "todo_ready_deferred"
+    assert "workspace_guard" not in payload
+    assert guarded_todo_ids == []
+
+
 def test_consumed_review_gate_exposes_quality_vision_replan() -> None:
     completed = quota_todo_item(
         todo_id="todo_completed_design",

tests/control_plane/test_work_lane_contract_core.py

@@ -3,8 +3,13 @@
 import pytest
 
 from loopx.control_plane.testing.quota_fixtures import quota_status_payload
+from loopx.control_plane.quota.selected_todo_projection import (
+    selected_todo_projection,
+)
+from loopx.control_plane.todos.summary_item import todo_text_content_revision
 from loopx.control_plane.work_items.work_lane import (
     preserve_heartbeat_receipt_bound_work_lane,
+    scoped_user_gate_due_monitor_contract,
 )
 from loopx.control_plane.work_items.work_lane_context import (
     item_progress_scope,
@@ -17,6 +22,32 @@
 PAST_DUE_AT = "2000-01-01T00:00:00+00:00"
 
 
+def test_scoped_gate_monitor_keeps_canonical_revision_for_selected_work_readback() -> None:
+    revision = "sha256:" + "a" * 64
+    lane = scoped_user_gate_due_monitor_contract(
+        {
+            "selected_executable": {
+                "todo_id": "todo_monitor",
+                "text": "Observe the independent target.",
+                "content_revision": revision,
+                "status": "open",
+                "task_class": "continuous_monitor",
+                "claimed_by": "agent-a",
+            }
+        },
+        current_contract={"lane": "advancement_task"},
+    )
+
+    assert lane is not None
+    assert lane["monitor_due_items"][0]["content_revision"] == revision
+    selected = selected_todo_projection(
+        agent_lane_next_action=None,
+        work_lane_contract=lane,
+    )
+    assert selected is not None
+    assert selected["content_revision"] == revision
+
+
 def _status(
     *,
     agent_todo_items: list[dict],
@@ -133,6 +164,9 @@ def test_due_watch_only_monitor_is_an_auxiliary_no_spend_route() -> None:
             {
                 "index": 1,
                 "text": "[P0] Monitor one overdue dependency.",
+                "content_revision": todo_text_content_revision(
+                    "[P0] Monitor one overdue dependency."
+                ),
                 "todo_id": "todo_watch_due",
                 "status": "open",
                 "priority": "P0",