fix(control-plane): withhold auxiliary polls on replan-bound turns

examples/control_plane/interaction-contract-state-machine-smoke.py

@@ -596,20 +596,16 @@ def assert_required_reads_are_mirrored_into_execution_channels() -> None:
         work_lane=advancement_lane(),
         heartbeat_mode="steering_audit_then_one_step",
     )
-    payload["required_reads"] = [
+    expected = [
         {
             "kind": "agent_scoped_evidence_log",
             "command": "  loopx evidence-log --goal-id interaction-state-machine-goal  ",
         }
     ]
+    # Reads are mirrored losslessly; the transport does not rewrite commands.
+    payload["required_reads"] = expected
     payload = finalize(payload)
     contract = payload["interaction_contract"]
-    expected = [
-        {
-            "kind": "agent_scoped_evidence_log",
-            "command": "loopx evidence-log --goal-id interaction-state-machine-goal",
-        }
-    ]
     assert contract["agent_channel"]["required_reads"] == expected, contract
     assert contract["cli_channel"]["required_reads"] == expected, contract
     assert "required_reads" not in contract["user_channel"], contract

loopx/control_plane/work_items/interaction_contract.py

@@ -1375,6 +1375,11 @@ def _build_interaction_cli_channel(
                     },
                 },
             }
+            replan_binding = (
+                replan_settlement_contract.get("settlement_binding")
+                if isinstance(replan_settlement_contract, Mapping)
+                else None
+            )
             if _auxiliary_monitor_receipt_binding_required(payload):
                 auxiliary_projection.update(
                     {
@@ -1395,6 +1400,17 @@ def _build_interaction_cli_channel(
                         "reason_code": "auxiliary_monitor_turn_instance_id_missing",
                     }
                 )
+            elif isinstance(replan_binding, Mapping) and replan_binding.get("kind") == "autonomous_replan":
+                auxiliary_projection.update(
+                    {
+                        "availability": "receipt_binding_required",
+                        "reason_code": "auxiliary_monitor_replan_receipt_binding",
+                        "next_step": (
+                            "settle the current autonomous replan Turn, then "
+                            "observe the due monitor from a Todo-bound Turn"
+                        ),
+                    }
+                )
             elif (payload.get("requires_user_action") is True and
                   (not isinstance(auxiliary_monitor.get("gate_scope"), Mapping) or
                    auxiliary_monitor["gate_scope"].get("state") != "independent")):

tests/control_plane/test_auxiliary_monitor_poll_availability.py

@@ -5,11 +5,20 @@
 from loopx.control_plane.work_items.interaction_contract import (
     build_interaction_contract,
 )
+from loopx.control_plane.scheduler.execution_context import (
+    SchedulerRuntimeProfile,
+    scheduler_execution_context_for_runtime_profile,
+)
 
 DUE_MONITOR_TODO_ID = "todo_due_monitor_fixture"
 
 
-def _payload(*, portfolio_requires_explicit_binding: bool) -> dict[str, object]:
+def _payload(
+    *,
+    portfolio_requires_explicit_binding: bool,
+    replan_obligation_id: str | None = None,
+    selected_todo_id: str | None = None,
+) -> dict[str, object]:
     payload: dict[str, object] = {
         "goal_id": "auxiliary-monitor-availability-fixture",
         "effective_action": "normal_run",
@@ -48,13 +57,20 @@ def _payload(*, portfolio_requires_explicit_binding: bool) -> dict[str, object]:
                 "requires_explicit_turn_binding": True,
             },
         }
+    if replan_obligation_id:
+        payload["replan_action_packet"] = {"obligation_id": replan_obligation_id}
+    if selected_todo_id:
+        payload["selected_todo"] = {"todo_id": selected_todo_id}
     return payload
 
 
 def _poll_projection(payload: dict[str, object]) -> dict[str, object]:
     contract = build_interaction_contract(
         payload,
         available_capabilities=["network", "external_evidence_poll"],
+        scheduler_execution_context=scheduler_execution_context_for_runtime_profile(
+            SchedulerRuntimeProfile.CODEX_APP_HEARTBEAT
+        ),
         turn_instance_id="turn-auxiliary-monitor-availability",
         runtime_root="/tmp/auxiliary-monitor-availability-runtime",
     )
@@ -81,11 +97,48 @@ def test_identity_less_turn_names_the_missing_receipt_binding() -> None:
 def test_turn_with_a_settlement_binding_still_offers_the_poll_command() -> None:
     """Parity: a Turn that can carry the observation keeps the ready command."""
 
-    projection = _poll_projection(_payload(portfolio_requires_explicit_binding=False))
+    projection = _poll_projection(
+        _payload(
+            portfolio_requires_explicit_binding=False,
+            selected_todo_id="todo_selected_fixture",
+        )
+    )
 
     assert projection["availability"] == "ready"
     assert projection["turn_instance_id"] == "turn-auxiliary-monitor-availability"
     assert "quota monitor-poll" in str(projection["command"])
     assert DUE_MONITOR_TODO_ID in str(projection["command"])
     assert "--use-current-task-lease" in str(projection["command"])
     assert projection["input_contract"]["task_lease_proof"]["source"] == "canonical_lease_or_same_turn_receipt"
+
+
+def test_replan_bound_turn_does_not_offer_a_monitor_poll_command() -> None:
+    """A replan receipt cannot also bind the due monitor observation."""
+
+    projection = _poll_projection(
+        _payload(
+            portfolio_requires_explicit_binding=False,
+            replan_obligation_id="replan-0000000000000001",
+        )
+    )
+
+    assert projection["availability"] == "receipt_binding_required"
+    assert projection["reason_code"] == "auxiliary_monitor_replan_receipt_binding"
+    assert "command" not in projection
+    assert "material_change_command" not in projection
+    assert "Todo-bound Turn" in str(projection["next_step"])
+
+
+def test_todo_bound_replan_still_offers_the_poll_command() -> None:
+    """The typed replan settlement owner may bind a selected Todo instead."""
+
+    projection = _poll_projection(
+        _payload(
+            portfolio_requires_explicit_binding=False,
+            replan_obligation_id="replan-0000000000000001",
+            selected_todo_id="todo_selected_fixture",
+        )
+    )
+
+    assert projection["availability"] == "ready"
+    assert "quota monitor-poll" in str(projection["command"])