diff --git a/loopx/control_plane/goals/goal_frontier/__init__.py b/loopx/control_plane/goals/goal_frontier/__init__.py index f0e60dedf3..ead42f1f24 100644 --- a/loopx/control_plane/goals/goal_frontier/__init__.py +++ b/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. ), ) diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py index a49d2dce5e..a402050247 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/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, diff --git a/loopx/control_plane/quota/unsettled_host_turn.py b/loopx/control_plane/quota/unsettled_host_turn.py index d27baf207b..139c642ed7 100644 --- a/loopx/control_plane/quota/unsettled_host_turn.py +++ b/loopx/control_plane/quota/unsettled_host_turn.py @@ -103,8 +103,9 @@ 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 @@ -112,6 +113,23 @@ def restore_receipt_bound_todo_source( """ 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 {} diff --git a/tests/control_plane/test_lifecycle_guard_reentry.py b/tests/control_plane/test_lifecycle_guard_reentry.py new file mode 100644 index 0000000000..ffae88862c --- /dev/null +++ b/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 diff --git a/tests/control_plane/test_replan_successor_guard_reentry.py b/tests/control_plane/test_replan_successor_guard_reentry.py index 7b6317c5f3..4b9293f24e 100644 --- a/tests/control_plane/test_replan_successor_guard_reentry.py +++ b/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: