fix(quota): keep autonomous replan executable across quiet monitor gate

docs/quota-allocation.md

@@ -407,8 +407,14 @@ wait state: quota uses the typed `future_monitor_wait` rule, returns
 synthetic replan. A due monitor remains executable through
 `due_monitor_execution`. Missing or invalid schedules, no-change streaks,
 vision/succession gaps, user gates, and real blockers still enter their
-higher-priority repair or replan rules. Projected ACKs from a different agent
-lane remain diagnostic only and cannot clear a current-lane obligation.
+higher-priority repair or replan rules. If an autonomous replan is selected
+while only a future-due monitor and an unrelated scoped user gate remain,
+`effective_action=autonomous_replan_required` keeps the agent's execution
+obligation active even when the user-gate notification is also shown. The
+quiet monitor lane cannot turn that replan into a no-op; the scheduler keeps
+`run_now` until the replan produces a typed outcome. Projected ACKs from a
+different agent lane remain diagnostic only and cannot clear a current-lane
+obligation.
 
 Executable todos can also declare explicit write-scope requirements through
 todo metadata, for example `required_write_scopes=runner%2F%2A%2A` or the CLI

loopx/control_plane/quota/should_run_packet.py

@@ -1204,7 +1204,11 @@ def _build_active_quota_payload(
     execution_obligation = _execution_obligation(
         should_run=route.should_run,
         effective_action=route.effective_action,
-        heartbeat_recommendation=route.heartbeat_recommendation,
+        heartbeat_recommendation=(
+            {**route.heartbeat_recommendation, "replan_obligation": prepared.replan_obligation}
+            if route.replan_decision_allowed
+            else route.heartbeat_recommendation
+        ),
         work_lane_contract=route.payload_work_lane_contract,
         external_evidence_observation=route.external_evidence_observation,
         user_gate_owns_frontier=_user_action_owns_empty_agent_lane(

loopx/control_plane/work_items/execution_obligation.py

@@ -73,7 +73,10 @@ def build_execution_obligation(
                 "source is unchanged and no concrete safe handoff exists"
             ),
         }
-    if should_run and recommended_mode == "autonomous_replan_required":
+    if should_run and (
+        effective_action == "autonomous_replan_required"
+        or recommended_mode == "autonomous_replan_required"
+    ):
         raw_replan_obligation = heartbeat_recommendation.get("replan_obligation")
         replan_obligation = (
             cast(dict[str, Any], raw_replan_obligation)

tests/control_plane/test_replan_host_context_projection.py

@@ -16,7 +16,11 @@
 from loopx.control_plane.scheduler.execution_context import (
     GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT,
 )
-from loopx.control_plane.testing.quota_fixtures import quota_status_payload
+from loopx.control_plane.testing.quota_fixtures import (
+    quota_status_payload,
+    quota_todo_item,
+    quota_todo_summary,
+)
 from loopx.control_plane.work_items.autonomous_replan_obligation import (
     build_autonomous_replan_obligation_payload,
 )
@@ -146,6 +150,60 @@ def test_quota_delivers_coverage_context_and_minimal_replan_action() -> None:
     ] == "produce one typed outcome from the host-projected replan action packet"
 
 
+def test_replan_obligation_outlives_quiet_monitor_and_scoped_user_gate() -> None:
+    monitor = quota_todo_item(
+        todo_id="todo_waiting_monitor",
+        title="Watch a pending review.",
+        task_class="continuous_monitor",
+        claimed_by=AGENT_ID,
+        target_key="github:example/project:pr:123",
+        cadence="30m",
+        next_due_at="2099-01-01T00:00:00+00:00",
+    )
+    gate = quota_todo_item(
+        todo_id="todo_test_scope_gate",
+        role="user",
+        title="Provide a non-production test scope.",
+        task_class="user_gate",
+        blocks_agent=AGENT_ID,
+    )
+    status = quota_status_payload(
+        goal_id=GOAL_ID,
+        status="active",
+        quota_state="operator_gate",
+        safe_bypass=True,
+        recommended_action="Wait for the test scope.",
+        agent_todos=quota_todo_summary([monitor], role="agent"),
+        user_todos=quota_todo_summary([gate], role="user"),
+        project_asset_extra={"autonomous_replan_obligation": _obligation()},
+        coordination={
+            "agent_model": "peer_v1",
+            "registered_agents": [AGENT_ID],
+        },
+    )
+
+    packet = build_quota_should_run(
+        status,
+        goal_id=GOAL_ID,
+        agent_id=AGENT_ID,
+        scheduler_execution_context=GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT,
+    )
+
+    assert packet["decision"] == "autonomous_replan_required"
+    assert packet["work_lane_contract"]["obligation"] == "quiet_until_material_monitor_transition"
+    assert packet["heartbeat_recommendation"]["recommended_mode"] == "ask_operator_gate"
+    assert packet["execution_obligation"]["must_attempt_work"] is True
+    assert packet["execution_obligation"]["kind"] == "autonomous_replan_required"
+    assert packet["execution_obligation"]["stall_threshold"] == 2
+    assert packet["interaction_contract"]["mode"] == "autonomous_replan"
+    assert packet["interaction_contract"]["user_channel"]["notify"] == "NOTIFY"
+    assert packet["interaction_contract"]["user_channel"]["actions"]
+    assert packet["interaction_contract"]["agent_channel"]["must_attempt"] is True
+    assert packet["selected_todo"] is None
+    assert packet["scheduler_hint"]["action"] == "run_now"
+    assert packet["scheduler_hint"].get("consistency_error") is None
+
+
 def test_manual_evidence_read_receipt_cannot_close_replan() -> None:
     payload = _quota_payload(include_legacy_receipt=True)