fix(quota): recover canonical sources for lifecycle-bound replans

loopx/control_plane/goals/goal_frontier/__init__.py

@@ -1743,10 +1743,13 @@ def build_goal_frontier_projection_context_from_status(
             guard_recorded_at=receipt_bound_guard_recorded_at, agent_id=agent_id,
             items=agent_todo_source_items or [],
             derive=lambda summary: derive_goal_frontier_replan_obligation_from_summaries(
-                user_todo_summary=user_todo_summary, agent_todo_summary=summary,
+                user_todo_summary=None, agent_todo_summary=summary,
                 work_lane_contract=work_lane_contract, agent_id=agent_id,
                 existing_replan_obligation=None, agent_todo_source_items=agent_todo_source_items,
-                _frontier_counts=frontier_counts,
+                # Reconstruct the admitted closed-row source, not the current
+                # frontier: independent agent work or user gates may have arrived
+                # after admission. This candidate still requires the original
+                # obligation digest; live gates retain authority over new work.
             ),
         )
 

loopx/control_plane/quota/live_decision.py

@@ -587,6 +587,9 @@ def build_live_quota_should_run_decision(
     receipt_bound_replay_phase = (
         settlement_readback.replay_phase if settlement_readback else None
     )
+    receipt_bound_guard_fields = _receipt_bound_guard_fields(
+        settlement_readback, receipt_bound_replan_obligation_id,
+    )
     fresh_operator_inbox_read = _fresh_operator_inbox_read_required(
         turn_start_hook_dispatch
     )
@@ -617,6 +620,7 @@ def build_live_quota_should_run_decision(
         decision_status_payload, registry_path=registry_path, runtime_root=runtime_root,
         goal_id=goal_id, todo_id=receipt_bound_todo_id,
         replay_phase=receipt_bound_replay_phase,
+        replan_guard_scoped=receipt_bound_guard_fields["receipt_bound_replan_guard_scoped"],
     )
     decision_status_payload = overlay_active_turn_retries(
         decision_status_payload,
@@ -654,9 +658,7 @@ def build_live_quota_should_run_decision(
         receipt_bound_monitor_phase=receipt_bound_monitor_phase,
         receipt_bound_replay_phase=receipt_bound_replay_phase,
         receipt_bound_replan_obligation_id=receipt_bound_replan_obligation_id,
-        **_receipt_bound_guard_fields(
-            settlement_readback, receipt_bound_replan_obligation_id,
-        ),
+        **receipt_bound_guard_fields,
         turn_instance_id=turn_instance_id,
         runtime_root=runtime_root,
         workspace_path=workspace_path,

loopx/control_plane/quota/unsettled_host_turn.py

@@ -103,15 +103,33 @@ def _todo_binding_facts(item: Mapping[str, Any] | None) -> dict[str, Any] | None
 def restore_receipt_bound_todo_source(
     payload: dict[str, Any], *, registry_path: Path, runtime_root: Path,
     goal_id: str, todo_id: str | None, replay_phase: ReceiptBoundReplayPhase | None,
+    replan_guard_scoped: bool = False,
 ) -> dict[str, Any]:
-    """Complete a display snapshot from the original binding's exact reader.
+    """Complete a display snapshot from the original binding's provider reader.
 
     Display lane caps cannot decide whether a committed Todo still exists.
     This only transports provider facts to the existing decision owner; it
     neither changes lifecycle nor substitutes a new settlement binding.
     """
     from ..todos.quota_summary import select_planning_inventory_source_items
 
+    if replan_guard_scoped and replay_phase is ReceiptBoundReplayPhase.OPEN:
+        # A replan receipt names an obligation, not a Todo. Its lifecycle
+        # source may be absent from every bounded display lane; the typed
+        # owner still validates the original source identity and timestamps.
+        from ..todos.list_readback import list_goal_todos
+
+        source = list_goal_todos(
+            registry_path=registry_path, runtime_root_arg=str(runtime_root), goal_id=goal_id,
+        )
+        todo_fields = {key: source[key] for key in ("user_todos", "agent_todos")}
+        queue = payload.get("attention_queue") or {}
+        return {**payload, "attention_queue": {**queue, "items": [
+            {**row, **todo_fields,
+                "project_asset": {**(row.get("project_asset") or {}), **todo_fields}}
+            if row.get("goal_id") == goal_id else row
+            for row in queue.get("items") or []
+        ]}}
     if not todo_id or replay_phase is ReceiptBoundReplayPhase.SETTLED:
         return payload
     queue = payload.get("attention_queue") or {}

tests/control_plane/test_lifecycle_guard_reentry.py

@@ -0,0 +1,110 @@
+"""A bounded display must not strand an admitted lifecycle recovery."""
+from __future__ import annotations
+
+import json
+from hashlib import sha256
+from pathlib import Path
+
+import pytest
+
+from test_replan_successor_guard_reentry import AGENT, GOAL, _fixture, _guard
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("later_user_gate", [False, True])
+def test_hidden_completed_source_recovers_its_original_guard_after_new_work(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, later_user_gate: bool,
+) -> None:
+    original_id = "todo_original_lifecycle"
+    completion_key = "local_completion_" + sha256(
+        f"loopx-local-todo-completion-v0\0{GOAL}\0{original_id}".encode()
+    ).hexdigest()[:32]
+    completed = {
+        "schema_version": "todo_item_v0", "archive_state": "active", "source_section": "Agent Todo",
+        "todo_id": original_id, "role": "agent", "index": 1,
+        "text": "Record the independently accepted maintenance outcome.",
+        "status": "done", "done": True, "task_class": "advancement_task",
+        "claimed_by": AGENT, "last_actor_agent_id": AGENT,
+        "note": "The finite maintenance outcome was independently accepted.",
+        "completed_at": "2026-08-01T00:00:00Z", "updated_at": "2026-08-01T00:00:00Z",
+        "completion_turn_key": completion_key, "completion_continuation": "active_goal",
+    }
+    # Enough newer terminal rows to move the original outside display budgets.
+    recent = [{**completed, "todo_id": f"todo_later_{i}", "index": i + 2,
+        "text": f"Accepted independent source {i}.", "no_followup": True,
+        "completion_continuation": "no_followup", "completion_turn_key": f"later-{i}",
+        "completed_at": "2026-08-02T00:00:00Z", "updated_at": "2026-08-02T00:00:00Z"}
+        for i in range(24)]
+    watch = {"schema_version": "todo_item_v0", "archive_state": "active", "source_section": "Agent Todo",
+        "todo_id": "todo_future_watch", "role": "agent", "index": 30,
+        "text": "Observe the pending independent source.", "status": "open", "done": False,
+        "task_class": "continuous_monitor", "claimed_by": AGENT,
+        "action_kind": "monitor", "target_key": "independent-source",
+        "cadence": "30m", "next_due_at": "2099-01-01T00:00:00Z",
+        "expires_at": "2099-01-02T00:00:00Z"}
+    call, runtime, index = _fixture(tmp_path, monkeypatch, provider,
+        initial_todos=[completed, *recent, watch])
+    index.write_text("")
+    turn = "turn-admitted-lifecycle-recovery"
+    admitted = _guard(call, turn)
+    identity = admitted["heartbeat_receipt"]["settlement_identity"]
+    obligation = admitted["autonomous_replan_obligation"]
+    assert obligation["resolution_mode"] == "todo_lifecycle_settlement"
+    assert [row["todo_id"] for row in obligation["triggers"]] == [original_id]
+
+    closed = call("todo", "complete", "--goal-id", GOAL, "--todo-id", original_id,
+        "--agent-id", AGENT, "--completion-identity-key", completion_key,
+        "--no-follow-up", "--note", "The finite maintenance outcome was independently accepted.")
+    assert closed["completion_recovery"] == "lifecycle_reentry_terminal_closeout"
+    # A later human request owns a different live frontier, not this old review.
+    later = call("todo", "add", "--goal-id", GOAL, "--role", "agent",
+        "--claimed-by", AGENT, "--text", "Validate the newly requested independent source.",
+        "--task-class", "advancement_task", "--action-kind", "validate",
+        "--target-key", "new-independent-source")
+    user_gate = None
+    if later_user_gate:
+        user_gate = call("todo", "add", "--goal-id", GOAL, "--role", "user",
+            "--agent-id", AGENT, "--task-class", "user_gate", "--global-gate",
+            "--text", "Authorize the separate newly requested source.")
+
+    recovered = _guard(call, turn)
+    assert recovered["heartbeat_receipt"]["settlement_identity"] == identity
+    assert recovered.get("selected_todo") is None
+    assert recovered["autonomous_replan_obligation"]["resolution_mode"] == "receipt_bound_replan_settlement"
+    assert recovered["replan_action_packet"]["decision"] == "vision_outcome_checkpoint_required"
+    assert recovered["replan_action_packet"]["semantic_delta"] is None
+    denied = call("quota", "spend-slot", "--goal-id", GOAL, "--agent-id", AGENT,
+        "--codex-app", "--turn-instance-id", turn, "--replan-obligation-id",
+        obligation["obligation_id"], "--source", "heartbeat", "--execute", expected_code=1)
+    assert denied.get("appended") is not True
+
+    vision = tmp_path / "accepted-review.json"
+    vision.write_text(json.dumps({"schema_version": "goal_vision_replan_contract_v0",
+        "state": "vision_patch_proposed", "vision_patch": {
+            "acceptance_summary": "Finite maintenance is verified; the independent new source remains open."},
+        "path_delta": {"schema_version": "goal_path_delta_v0", "outcome": "continue",
+            "prior_assumption": "Finite maintenance needed a terminal disposition.",
+            "observed_reality": "Its original outcome and no-follow-up were verified; new work remains independent.",
+            "retained": ["The newly requested source requires its own admission."],
+            "evidence_refs": ["evidence:accepted-maintenance"]}}))
+    refreshed = call("refresh-state", "--goal-id", GOAL, "--agent-id", AGENT,
+        "--turn-instance-id", turn, "--replan-obligation-id", obligation["obligation_id"],
+        "--classification", "verified_lifecycle_outcome", "--delivery-batch-scale", "single_surface",
+        "--delivery-outcome", "outcome_progress", "--agent-vision-json", str(vision),
+        "--no-global-sync", "--suppress-external-sinks")
+    assert refreshed["settlement_progress"]["state"] == "spend_required"
+    spend = ("quota", "spend-slot", "--goal-id", GOAL, "--agent-id", AGENT,
+        "--codex-app", "--turn-instance-id", turn, "--replan-obligation-id",
+        obligation["obligation_id"], "--source", "heartbeat", "--execute")
+    assert call(*spend)["settlement_progress"]["state"] == "settled"
+    assert call(*spend)["appended"] is False
+    assert call("todo", "list", "--goal-id", GOAL, "--todo-id", later["todo_id"])["todo"]["status"] == "open"
+    if user_gate:
+        assert call("todo", "list", "--goal-id", GOAL, "--todo-id", user_gate["todo_id"])["todo"]["status"] == "open"
+        fresh = _guard(call, "turn-independent-source-after-lifecycle")
+        interaction = fresh["interaction_contract"]
+        assert interaction["agent_channel"]["delivery_allowed"] is False
+        assert interaction["agent_channel"]["must_attempt"] is False
+        assert interaction["response_plan"]["decision"] == "ask_user"
+    assert len([row for row in index.read_text().splitlines()
+        if json.loads(row).get("classification") == "quota_slot_spent"]) == 1

tests/control_plane/test_replan_successor_guard_reentry.py

@@ -24,7 +24,7 @@
 ROOT = Path(loopx.__file__).resolve().parents[1]
 
 
-def _fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, *, later_vision: bool = False, future_monitor: bool = True):
+def _fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, *, later_vision: bool = False, future_monitor: bool = True, initial_todos: list[dict] | None = None):
     if provider == "sqlite":
         isolate_sqlite_runtime(tmp_path, monkeypatch)
     project, runtime = tmp_path / "project", tmp_path / "runtime"
@@ -67,7 +67,7 @@ def _fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, *,
         "coordination": {"agent_model": "peer_v1", "registered_agents": [AGENT, "different-agent"]},
     }]}))
     initialize_canonical_authority(runtime, GOAL, build_todo_runtime_shadow_projection(
-        goal_id=GOAL, todos=[], handoff_mode="soft_claim", leases=[],
+        goal_id=GOAL, todos=initial_todos or [], handoff_mode="soft_claim", leases=[],
     ), state_path=state, provider=provider)
 
     def call(*args: str, expected_code: int = 0) -> dict: