From e22459db46b4cd100290e929d268d098048e83df Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 15 Oct 2026 20:24:32 +0800 Subject: [PATCH 1/2] fix(quota): recover receipt-bound todos omitted from compact views Signed-off-by: huangruiteng --- loopx/control_plane/quota/live_decision.py | 6 ++ .../quota/unsettled_host_turn.py | 37 +++++++ .../test_deferred_original_turn_binding.py | 96 +++++++++++++++++++ 3 files changed, 139 insertions(+) create mode 100644 tests/control_plane/test_deferred_original_turn_binding.py diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py index a9762136c2..f15e1ed30c 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/loopx/control_plane/quota/live_decision.py @@ -35,6 +35,7 @@ from .unsettled_host_turn import ( apply_unsettled_host_turn_recovery_if_required, apply_receipt_bound_wait_recovery, + restore_receipt_bound_todo_source, ) @@ -612,6 +613,11 @@ def build_live_quota_should_run_decision( for item in queue.get("items") or [] ], } + if receipt_bound_todo_id and receipt_bound_replay_phase is not ReceiptBoundReplayPhase.SETTLED: + decision_status_payload = restore_receipt_bound_todo_source( + decision_status_payload, registry_path=registry_path, runtime_root=runtime_root, + goal_id=goal_id, todo_id=receipt_bound_todo_id, + ) decision_status_payload = overlay_active_turn_retries( decision_status_payload, goal_id=goal_id, diff --git a/loopx/control_plane/quota/unsettled_host_turn.py b/loopx/control_plane/quota/unsettled_host_turn.py index 954b80ba71..5d2ed3f3f8 100644 --- a/loopx/control_plane/quota/unsettled_host_turn.py +++ b/loopx/control_plane/quota/unsettled_host_turn.py @@ -99,6 +99,43 @@ 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, +) -> dict[str, Any]: + """Complete a display snapshot from the original binding's exact 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 + + queue = payload.get("attention_queue") or {} + rows = list(queue.get("items") or []) + for index, row in enumerate(rows): + if row.get("goal_id") != goal_id: + continue + asset = row.get("project_asset") or {} + source = select_planning_inventory_source_items( + row.get("agent_todos"), asset.get("agent_todos"), include_terminal=True, + ) + if any(item.get("todo_id") == todo_id for item in source): + return payload + bound = _bound_todo_item(registry_path=registry_path, runtime_root=runtime_root, + goal_id=goal_id, todo_id=todo_id) + if bound is None: + raise ValueError("The original receipt-bound Todo cannot be read; recover its lifecycle before selecting or settling work") + def restored(summary: Mapping[str, Any] | None) -> dict[str, Any]: + current = dict(summary or {}) + current["items"] = [*(current.get("items") or []), bound] + return current + rows[index] = {**row, "agent_todos": restored(row.get("agent_todos")), + "project_asset": {**asset, "agent_todos": restored(asset.get("agent_todos"))}} + return {**payload, "attention_queue": {**queue, "items": rows}} + return payload + + def _prior_closeout_preflight( *, runtime_root: Path, diff --git a/tests/control_plane/test_deferred_original_turn_binding.py b/tests/control_plane/test_deferred_original_turn_binding.py new file mode 100644 index 0000000000..fdc2e3a13d --- /dev/null +++ b/tests/control_plane/test_deferred_original_turn_binding.py @@ -0,0 +1,96 @@ +"""A compact candidate view cannot replace an admitted Turn's identity.""" +import copy +import json +from pathlib import Path + +import pytest + +from tests.control_plane.test_quota_settlement_cli import ( + AGENT_ID, + GOAL_ID, + TODO_ID, + _configure_autonomous_replan_fixture, + _run_cli, + _write_fixture, +) + + +@pytest.mark.parametrize("hidden_count", [0, 60]) +@pytest.mark.parametrize("checked", [False, True]) +@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"]) +def test_deferred_replay_keeps_original_binding_during_replan( + tmp_path: Path, hidden_count: int, checked: bool, provider: str, + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture, +) -> None: + if provider == "sqlite": + from canonical_authority_fixture import isolate_sqlite_runtime + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture(tmp_path) + turn = "turn-original-deferred" + guard = ("quota", "should-run", "--runtime-profile", "codex_app_ssh_goal", + "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--turn-instance-id", turn, "--todo-id", TODO_ID) + rc, original = _run_cli(registry, runtime, *guard) + assert rc == 0, original + identity = original["heartbeat_receipt"]["settlement_identity"] + assert identity["todo_id"] == TODO_ID + + state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" + text = state.read_text().replace("status: active-read-only", "status: active") + # A new replan is legitimately due, but does not own the old Turn. + _configure_autonomous_replan_fixture(project, runtime, registry) + text = text.replace(f"todo_id={TODO_ID} status=open", + f"todo_id={TODO_ID} status=deferred resume_when=todo_done:todo_prior") + if checked: + text = text.replace("- [ ] [P1] Validate", "- [x] [P1] Validate") + text += ("\n- [x] [P1] Prior dependency is complete.\n" + f" \n") + rows = "".join(f"- [ ] [P0] Independent waiting candidate {i}.\n" + f" \n" + for i in range(hidden_count)) + text = text.replace("## Agent Todo\n\n", "## Agent Todo\n\n" + rows) + state.write_text(text) + if provider != "legacy": + from canonical_authority_fixture import initialize_canonical_authority + from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection + from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos + + todos = parse_active_state_todos(text, item_limit=None)["agent_todos"]["items"] + initialize_canonical_authority(runtime, GOAL_ID, + build_todo_runtime_shadow_projection(goal_id=GOAL_ID, + todos=todos, handoff_mode="soft_claim", leases=[]), + state_path=state, provider=provider) + + from loopx.cli import main + from loopx.cli_commands import quota + + live = quota.build_live_quota_should_run_decision + + def compact_status(status, **kwargs): + # The display snapshot can omit the original row; the provider still + # owns it. Exercise the real reader, router, and receipt replay. + def compact(value): + if isinstance(value, list): + return [compact(row) for row in value + if not isinstance(row, dict) or row.get("todo_id") != TODO_ID] + if isinstance(value, dict): + return {key: compact(row) for key, row in value.items()} + return value + return live(compact(copy.deepcopy(status)) if hidden_count else status, **kwargs) + + monkeypatch.setattr(quota, "build_live_quota_should_run_decision", compact_status) + capsys.readouterr() + rc = main(["--format", "json", "--registry", str(registry), + "--runtime-root", str(runtime), *guard, "--scan-path", str(project)]) + replay = json.loads(capsys.readouterr().out) + assert rc == 0, replay + assert replay["heartbeat_receipt"]["settlement_identity"] == identity + channel = replay["interaction_contract"]["cli_channel"] + plan = channel.get("settlement_plan") + assert plan is None or plan["identity"]["todo_id"] == TODO_ID, channel + assert replay["should_run"] is False, replay.get("effective_action") + assert replay["normal_delivery_allowed"] is False + assert channel["spend_after_validation"] is False From 505290c2cbc020a4ce1675c03b1810c7a07d0a7c Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 15 Oct 2026 20:52:00 +0800 Subject: [PATCH 2/2] refactor(quota): keep receipt recovery guards at the source boundary Signed-off-by: huangruiteng --- loopx/control_plane/quota/live_decision.py | 10 +++++----- loopx/control_plane/quota/unsettled_host_turn.py | 5 ++++- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py index f15e1ed30c..a49d2dce5e 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/loopx/control_plane/quota/live_decision.py @@ -613,11 +613,11 @@ def build_live_quota_should_run_decision( for item in queue.get("items") or [] ], } - if receipt_bound_todo_id and receipt_bound_replay_phase is not ReceiptBoundReplayPhase.SETTLED: - decision_status_payload = restore_receipt_bound_todo_source( - decision_status_payload, registry_path=registry_path, runtime_root=runtime_root, - goal_id=goal_id, todo_id=receipt_bound_todo_id, - ) + decision_status_payload = restore_receipt_bound_todo_source( + 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, + ) decision_status_payload = overlay_active_turn_retries( decision_status_payload, goal_id=goal_id, diff --git a/loopx/control_plane/quota/unsettled_host_turn.py b/loopx/control_plane/quota/unsettled_host_turn.py index 5d2ed3f3f8..d27baf207b 100644 --- a/loopx/control_plane/quota/unsettled_host_turn.py +++ b/loopx/control_plane/quota/unsettled_host_turn.py @@ -29,6 +29,7 @@ HeartbeatReceiptIdentityConflictError, ) from .accounting_admission import quota_accounting_admission +from .effect_program import ReceiptBoundReplayPhase UNSETTLED_HOST_TURN_RECOVERY_SCHEMA_VERSION = "unsettled_host_turn_recovery_v0" @@ -101,7 +102,7 @@ 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, + goal_id: str, todo_id: str | None, replay_phase: ReceiptBoundReplayPhase | None, ) -> dict[str, Any]: """Complete a display snapshot from the original binding's exact reader. @@ -111,6 +112,8 @@ def restore_receipt_bound_todo_source( """ from ..todos.quota_summary import select_planning_inventory_source_items + if not todo_id or replay_phase is ReceiptBoundReplayPhase.SETTLED: + return payload queue = payload.get("attention_queue") or {} rows = list(queue.get("items") or []) for index, row in enumerate(rows):