fix(control-plane): preserve monitor settlement binding

loopx/control_plane/agents/agent_lane_recommendation.py

@@ -13,6 +13,7 @@
 from ..todos.summary_item import compact_todo_summary_item
 from ..work_items.primary_action import protocol_action_text
 from ..work_items.work_lane import (
+    ReceiptBoundMonitorPhase,
     work_lane_contract_is_due_monitor_attempt,
     work_lane_contract_is_lark_inbox_reply_due,
 )
@@ -42,7 +43,12 @@ def build_receipt_bound_monitor_next_action(
     available_capabilities: Any,
     receipt_bound_todo_id: str | None,
 ) -> dict[str, Any] | None:
-    """Recover an exact due monitor omitted by priority-limited hot lanes."""
+    """Recover an exact receipt-bound monitor omitted by compact hot lanes.
+
+    A monitor remains the same-turn settlement identity after a successful poll
+    reschedules it into the future.  That replay must preserve the parent Todo
+    without asking the agent to observe the target again.
+    """
 
     if not isinstance(agent_identity, dict):
         return None
@@ -53,13 +59,16 @@ def build_receipt_bound_monitor_next_action(
     for item in agent_todo_items:
         if normalize_todo_id(item.get("todo_id")) != todo_id:
             continue
+        monitor_due = todo_item_is_due_monitor(item)
         if (
             not _todo_item_is_actionable_open(item)
             or _todo_task_class(item) != TODO_TASK_CLASS_MONITOR
-            or not todo_item_is_due_monitor(item)
-            or missing_required_capabilities(
-                item,
-                available_capabilities=available_capabilities,
+            or (
+                monitor_due
+                and missing_required_capabilities(
+                    item,
+                    available_capabilities=available_capabilities,
+                )
             )
             or not agent_scope_item_claimed_by_agent_or_unclaimed(
                 item,
@@ -80,6 +89,11 @@ def build_receipt_bound_monitor_next_action(
                 "confidence": "selected",
                 "preserves_goal_next_action": True,
                 "selection_binding": "heartbeat_receipt",
+                "receipt_bound_monitor_phase": (
+                    ReceiptBoundMonitorPhase.POLL_DUE
+                    if monitor_due
+                    else ReceiptBoundMonitorPhase.SETTLEMENT_PENDING
+                ).value,
             }
         )
         if not agent_scope_item_claimed_by(item):

loopx/control_plane/work_items/work_lane.py

@@ -1,22 +1,36 @@
 from __future__ import annotations
 
+from enum import StrEnum
 from typing import Any
 
 from ..todos.contract import TODO_TASK_CLASS_MONITOR, normalize_todo_id
 from ..todos.projection import todo_priority_label, todo_priority_rank
 
 
 WORK_LANE_CONTRACT_SCHEMA_VERSION = "work_lane_contract_v1"
+WORK_LANE_RECEIPT_BOUND_MONITOR_SETTLEMENT_OBLIGATION = (
+    "settle_receipt_bound_monitor"
+)
 WORK_LANE_CURRENT_AGENT_MONITOR_REPAIR_OBLIGATIONS = {
     "attempt_due_monitor",
     "repair_monitor_schedule_metadata",
     "repair_resume_gate_or_close_standing_monitor",
+    WORK_LANE_RECEIPT_BOUND_MONITOR_SETTLEMENT_OBLIGATION,
 }
 WORK_LANE_EXTERNAL_EVIDENCE_OBSERVATION_OBLIGATION = (
     "observe_external_evidence_or_blocker"
 )
 WORK_LANE_LARK_INBOX_REPLY_DUE_OBLIGATION = "drain_lark_inbox_reply_due"
 WORK_LANE_TODO_MONITOR_DUE_KIND = "todo_monitor_due"
+
+
+class ReceiptBoundMonitorPhase(StrEnum):
+    """Same-turn phase of a monitor selected by a heartbeat receipt."""
+
+    POLL_DUE = "poll_due"
+    SETTLEMENT_PENDING = "settlement_pending"
+
+
 WORK_LANE_TODO_ITEM_FIELDS = (
     "index",
     "text",
@@ -119,16 +133,49 @@ def preserve_heartbeat_receipt_bound_work_lane(
 ) -> dict[str, Any] | None:
     """Keep a committed same-turn Todo binding ahead of a newly due monitor."""
 
-    if not isinstance(contract, dict) or not work_lane_contract_is_due_monitor_attempt(
-        contract
-    ):
+    if not isinstance(contract, dict):
         return contract
     if not isinstance(selected_todo, dict):
         return contract
     todo_id = normalize_todo_id(selected_todo.get("todo_id"))
     if not todo_id or selected_todo.get("selection_binding") != "heartbeat_receipt":
         return contract
     if selected_todo.get("task_class") == TODO_TASK_CLASS_MONITOR:
+        try:
+            monitor_phase = ReceiptBoundMonitorPhase(
+                selected_todo.get("receipt_bound_monitor_phase")
+            )
+        except (TypeError, ValueError) as exc:
+            raise ValueError(
+                "receipt-bound monitor selection requires an explicit monitor phase"
+            ) from exc
+        if monitor_phase is ReceiptBoundMonitorPhase.SETTLEMENT_PENDING:
+            return {
+                **contract,
+                "schema_version": WORK_LANE_CONTRACT_SCHEMA_VERSION,
+                "lane": "continuous_monitor",
+                "obligation": (
+                    WORK_LANE_RECEIPT_BOUND_MONITOR_SETTLEMENT_OBLIGATION
+                ),
+                "must_attempt_work": True,
+                "selection_binding": "heartbeat_receipt",
+                "selected_todo_id": todo_id,
+                "monitor_due_count": 0,
+                "monitor_due_items": [],
+                "receipt_bound_monitor_item": selected_todo,
+                "reason_codes": [
+                    "heartbeat_receipt_bound_replay",
+                    "monitor_poll_already_recorded",
+                    "same_turn_settlement_identity",
+                ],
+                "monitor_policy": "settle_receipt_bound_monitor_without_repoll",
+                "deferred_work_lane": contract,
+                "action": (
+                    "finish refresh and quota settlement for the already-polled "
+                    "monitor bound to this heartbeat turn; do not poll again; "
+                    "reconsider its successor on the next turn"
+                ),
+            }
         return {
             **contract,
             "schema_version": WORK_LANE_CONTRACT_SCHEMA_VERSION,
@@ -150,6 +197,8 @@ def preserve_heartbeat_receipt_bound_work_lane(
                 "reconsider newly runnable monitor priority on the next turn"
             ),
         }
+    if not work_lane_contract_is_due_monitor_attempt(contract):
+        return contract
     return {
         "schema_version": WORK_LANE_CONTRACT_SCHEMA_VERSION,
         "lane": "advancement_task",

loopx/state_projection.py

@@ -390,16 +390,29 @@ def _section_lines(state_text: str, heading: str) -> list[str]:
     return lines
 
 
-def _section_entries(lines: list[str]) -> list[str]:
+def _section_entries(
+    lines: list[str],
+    *,
+    text_limit: int | None = 220,
+) -> list[str]:
     entries: list[str] = []
     current: list[str] = []
+
+    def append_entry(parts: list[str]) -> None:
+        text = compact_todo_text(" ".join(parts))
+        if not text:
+            return
+        entries.append(
+            text if text_limit is None else _compact_text(text, limit=text_limit)
+        )
+
     for line in lines:
         if line.strip().startswith("<!--"):
             continue
         bullet = BULLET_PATTERN.match(line)
         if bullet:
             if current:
-                entries.append(_compact_text(" ".join(current)))
+                append_entry(current)
             current = [bullet.group(1)]
             continue
         if current and line.startswith((" ", "\t")):
@@ -408,22 +421,26 @@ def _section_entries(lines: list[str]) -> list[str]:
                 current.append(continuation)
             continue
         if current:
-            entries.append(_compact_text(" ".join(current)))
+            append_entry(current)
             current = []
         stripped = line.strip()
         if stripped:
-            entries.append(_compact_text(stripped))
+            append_entry([stripped])
     if current:
-        entries.append(_compact_text(" ".join(current)))
+        append_entry(current)
     return [entry for entry in entries if entry]
 
 
 def active_state_next_action_entries(
     state_text: str,
     *,
     limit: int | None = 3,
+    text_limit: int | None = 220,
 ) -> list[str]:
-    entries = _section_entries(_section_lines(state_text, "Next Action"))
+    entries = _section_entries(
+        _section_lines(state_text, "Next Action"),
+        text_limit=text_limit,
+    )
     if limit is None:
         return entries
     return entries[: max(0, limit)]

loopx/state_refresh.py

@@ -482,7 +482,11 @@ def build_state_refresh_record(
     settlement_identity: SettlementIdentity | None = None,
 ) -> dict[str, Any]:
     frontmatter = parse_frontmatter(state_text)
-    next_action = active_state_next_action_entries(state_text, limit=8)
+    next_action = active_state_next_action_entries(
+        state_text,
+        limit=8,
+        text_limit=None,
+    )
     recent_feedback = extract_section_lines(state_text, "Recent User Feedback", limit=5)
     progress = extract_section_lines(state_text, "Progress Ledger", limit=5)
     digest = hashlib.sha256(state_text.encode("utf-8")).hexdigest()[:16]

tests/control_plane/test_monitor_followthrough_contract.py

@@ -17,6 +17,9 @@
     write_fixture_registry,
 )
 from loopx.control_plane.todos.contract import resolve_next_user_task_class
+from loopx.control_plane.work_items.work_lane import (
+    preserve_heartbeat_receipt_bound_work_lane,
+)
 from loopx.quota import build_quota_should_run, record_quota_monitor_poll
 from loopx.status import collect_status
 from loopx.todos import add_goal_todo, list_goal_todos, update_goal_todo
@@ -332,6 +335,23 @@ def test_turn_scoped_monitor_poll_preserves_receipt_todo_after_capability_reentr
     assert guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] == admitted[
         "todo_id"
     ]
+    guard_replay = run_json_cli(
+        "quota",
+        "should-run",
+        "--goal-id",
+        GOAL_ID,
+        "--agent-id",
+        AGENT_ID,
+        "--runtime-profile",
+        "generic_cli",
+        "--turn-instance-id",
+        turn_id,
+        registry_path=registry,
+        runtime_root=runtime,
+    )
+    assert guard_replay["agent_lane_next_action"][
+        "receipt_bound_monitor_phase"
+    ] == "poll_due"
 
     command = (
         "quota",
@@ -407,6 +427,133 @@ def test_turn_scoped_monitor_poll_preserves_receipt_todo_after_capability_reentr
     assert admitted["todo_id"] in conflict["reason"]
 
 
+def test_same_turn_should_run_settles_polled_monitor_before_successor_reselection(
+    tmp_path: Path,
+) -> None:
+    registry, runtime, _state = _write_fixture(tmp_path)
+    admitted = _add_monitor(
+        registry,
+        text="[P1] Poll the admitted public release target.",
+        target_key="public-release:receipt-settlement",
+        next_due_at="2000-01-01T00:00:00+00:00",
+    )
+    turn_id = "2026-08-21T09:48:02.405Z"
+    guard_args = (
+        "quota",
+        "should-run",
+        "--goal-id",
+        GOAL_ID,
+        "--agent-id",
+        AGENT_ID,
+        "--runtime-profile",
+        "generic_cli",
+        "--turn-instance-id",
+        turn_id,
+        "--available-capability",
+        "network",
+        "--available-capability",
+        "external_evidence_poll",
+    )
+
+    guard = run_json_cli(
+        *guard_args,
+        registry_path=registry,
+        runtime_root=runtime,
+    )
+    assert guard["selected_todo"]["todo_id"] == admitted["todo_id"]
+
+    poll = run_json_cli(
+        "quota",
+        "monitor-poll",
+        "--goal-id",
+        GOAL_ID,
+        "--agent-id",
+        AGENT_ID,
+        "--runtime-profile",
+        "generic_cli",
+        "--turn-instance-id",
+        turn_id,
+        "--available-capability",
+        "network",
+        "--available-capability",
+        "external_evidence_poll",
+        "--todo-id",
+        admitted["todo_id"],
+        "--target-key",
+        "public-release:receipt-settlement",
+        "--result-hash",
+        "merged-receipt-settlement",
+        "--material-change",
+        "--next-agent-todo",
+        "Validate the exact merged release head.",
+        "--next-action-kind",
+        "validate_release_head",
+        "--next-task-repository",
+        "git:github.com/huangruiteng/loopx",
+        "--next-required-capability",
+        "network",
+        "--next-continuation-policy",
+        "same_agent_non_delivery",
+        "--next-target-key",
+        "release-head:receipt-settlement",
+        "--next-claimed-by",
+        AGENT_ID,
+        "--execute",
+        registry_path=registry,
+        runtime_root=runtime,
+    )
+    successor_id = poll["successor_todo_ids"][0]
+    assert poll["after"]["selected_todo"]["todo_id"] == admitted["todo_id"]
+
+    replay = run_json_cli(
+        *guard_args,
+        registry_path=registry,
+        runtime_root=runtime,
+    )
+    assert replay["selected_todo"]["todo_id"] == admitted["todo_id"]
+    assert replay["selected_todo"]["selection_binding"] == "heartbeat_receipt"
+    assert replay["agent_lane_next_action"]["receipt_bound_monitor_phase"] == (
+        "settlement_pending"
+    )
+    assert replay["work_lane_contract"]["obligation"] == (
+        "settle_receipt_bound_monitor"
+    )
+    assert replay["work_lane_contract"]["selected_todo_id"] == admitted["todo_id"]
+    assert replay["work_lane_contract"]["deferred_work_lane"]["lane"] == (
+        "advancement_task"
+    )
+    assert "do not poll again" in replay["work_lane_contract"]["action"]
+    assert replay["heartbeat_receipt"]["status"] == "replayed"
+
+    next_turn_args = list(guard_args)
+    next_turn_args[next_turn_args.index(turn_id)] = "2026-08-21T09:51:02.405Z"
+    next_turn = run_json_cli(
+        *next_turn_args,
+        registry_path=registry,
+        runtime_root=runtime,
+    )
+    assert next_turn["selected_todo"]["todo_id"] == successor_id
+
+
+def test_receipt_bound_monitor_replay_requires_an_explicit_phase() -> None:
+    with pytest.raises(
+        ValueError,
+        match="receipt-bound monitor selection requires an explicit monitor phase",
+    ):
+        preserve_heartbeat_receipt_bound_work_lane(
+            {
+                "schema_version": "work_lane_contract_v1",
+                "lane": "continuous_monitor",
+                "must_attempt_work": True,
+            },
+            selected_todo={
+                "todo_id": "todo_monitor_without_phase",
+                "task_class": "continuous_monitor",
+                "selection_binding": "heartbeat_receipt",
+            },
+        )
+
+
 def test_turn_scoped_monitor_poll_requires_committed_receipt(tmp_path: Path) -> None:
     registry, runtime, _state = _write_fixture(tmp_path)
     monitor = _add_monitor(