From bd8273d1e978c012f87f3d1b4eadecee0dd5fe3a Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 04:38:28 +0800 Subject: [PATCH] fix(quota): preserve deferred replan before execution preparation Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../quota/execution_preparation.py | 3 + .../test_quota_settlement_cli.py | 57 +++++++++++++++++-- .../test_recovery_due_successor_identity.py | 11 +++- 3 files changed, 64 insertions(+), 7 deletions(-) diff --git a/loopx/control_plane/quota/execution_preparation.py b/loopx/control_plane/quota/execution_preparation.py index 792c508937..3d9d840902 100644 --- a/loopx/control_plane/quota/execution_preparation.py +++ b/loopx/control_plane/quota/execution_preparation.py @@ -109,6 +109,9 @@ def apply_selected_todo_execution_preparation( if ( not isinstance(selected, dict) or payload.get("should_run") is not True + # Planning and recovery can retain a Todo without authorizing its + # business execution. Preserve their existing typed route. + or payload.get("normal_delivery_allowed") is not True or action_portfolio_requires_explicit_selection(payload) ): return diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 5c044f8e67..3be80d5da8 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -5362,6 +5362,13 @@ def test_same_turn_bound_p0_does_not_project_p1_after_p0_becomes_deferred( "--agent-id", AGENT_ID, "--claimed-by", AGENT_ID, "--todo-id", TODO_ID, ) assert claim_rc == 0 and claim["execution_managed"] is True, json.dumps(claim) + prepared_rc, prepared = _run_cli(registry_path, runtime, *guard, "--todo-id", TODO_ID) + assert prepared_rc == 0, prepared + assert prepared["execution_preparation"]["state"] == "ready" + assert prepared["normal_delivery_allowed"] is True + assert prepared["heartbeat_receipt"]["settlement_identity"] == ( + first["heartbeat_receipt"]["settlement_identity"] + ) # This result exists before the lifecycle wait; recovery must account for # it without executing more work or rebinding the original Turn. artifact = project / "verified-before-wait.txt" @@ -5409,7 +5416,8 @@ def test_same_turn_bound_p0_does_not_project_p1_after_p0_becomes_deferred( ) plan = interaction["cli_channel"].get("settlement_plan") assert plan["identity"] == first["heartbeat_receipt"]["settlement_identity"] - assert _heartbeat_receipt_count(runtime, turn_id) == 1 + # Preparation and positive work admission share one settlement identity. + assert _heartbeat_receipt_count(runtime, turn_id) == 2 assert _spend_run_count(runtime) == 0 conflict_rc, conflict = _run_cli( @@ -5420,7 +5428,7 @@ def test_same_turn_bound_p0_does_not_project_p1_after_p0_becomes_deferred( "heartbeat_receipt_identity_conflict", "quota_action_selection_rejected", } - assert _heartbeat_receipt_count(runtime, turn_id) == 1 + assert _heartbeat_receipt_count(runtime, turn_id) == 2 vision = tmp_path / "before-wait-vision.json" vision.write_text(json.dumps({ @@ -5464,9 +5472,10 @@ def test_same_turn_bound_p0_does_not_project_p1_after_p0_becomes_deferred( assert next_turn["selected_todo"]["todo_id"] == ALTERNATIVE_TODO_ID +@pytest.mark.parametrize("handoff_mode", ["soft_claim", "hard_lease"]) @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_due_successor_replan_replays_until_the_same_task_is_reopened( - tmp_path: Path, provider: str, monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, provider: str, handoff_mode: str, monkeypatch: pytest.MonkeyPatch, ) -> None: from canonical_authority_fixture import isolate_sqlite_runtime @@ -5478,7 +5487,7 @@ def test_due_successor_replan_replays_until_the_same_task_is_reopened( f"todo_id={TODO_ID} status=open", f"todo_id={TODO_ID} status=deferred resume_when=resume_at:2000-01-01T00:00:00Z", )) - _initialize_fixture_authority(project, runtime, provider=provider) + _initialize_fixture_authority(project, runtime, provider=provider, handoff_mode=handoff_mode) guard = ( "quota", "should-run", "--codex-app", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--turn-instance-id", "due-successor-replay", @@ -5486,6 +5495,10 @@ def test_due_successor_replan_replays_until_the_same_task_is_reopened( rc, first = _run_cli(registry_path, runtime, *guard) assert rc == 0, first assert first["effective_action"] == "successor_replan_required" + assert first["should_run"] is True + assert first["normal_delivery_allowed"] is False + assert first["interaction_contract"]["agent_channel"]["must_attempt"] is True + assert first["interaction_contract"]["agent_channel"]["quiet_noop_allowed"] is False identity = first["heartbeat_receipt"]["settlement_identity"] rc, replay = _run_cli(registry_path, runtime, *guard) assert rc == 0, replay @@ -5495,12 +5508,35 @@ def test_due_successor_replan_replays_until_the_same_task_is_reopened( assert replay["heartbeat_receipt"]["settlement_identity"] == identity assert _spend_run_count(runtime) == 0 - rc, deferred = _run_cli( - registry_path, runtime, "todo", "update", "--goal-id", GOAL_ID, + defer_args = ( + "todo", "update", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--status", "deferred", "--resume-when", "resume_at:2099-01-01T00:00:00Z", "--reason", "A new dependency is not ready yet.", ) + rc, deferred = _run_cli(registry_path, runtime, *defer_args) + if handoff_mode == "hard_lease": + assert rc == 1 and deferred["error_code"] == "handoff_mode_requires_lease", deferred + recovery = deferred["recovery"] + assert recovery["execution_authority_granted"] is False + assert recovery["action"] == "resolve_lifecycle_edit" + # Follow the returned lifecycle recovery, without acquiring execution + # for a deferred Todo or changing its declaration. + rc, current = _run_cli(registry_path, runtime, "todo", "list", + "--goal-id", GOAL_ID, "--todo-id", TODO_ID) + assert rc == 0, current + command = shlex.join(["loopx", *recovery["retry"]["cli_args"], + "--reason", "Review the wait before rescheduling it.", + "--update-operation-id", "due-successor-reviewed-resume", + "--update-expected-provider-revision", current["authority_read"]["provider_revision"]]) + rc, resumed_wait = _run_generated_cli(command, registry_path=registry_path) + assert rc == 0, resumed_wait + rc, current = _run_cli(registry_path, runtime, "todo", "list", + "--goal-id", GOAL_ID, "--todo-id", TODO_ID) + assert rc == 0, current + rc, deferred = _run_cli(registry_path, runtime, *defer_args, + "--update-operation-id", "due-successor-reviewed-deferral", + "--update-expected-provider-revision", current["authority_read"]["provider_revision"]) assert rc == 0, deferred rc, paused = _run_cli(registry_path, runtime, *guard) assert rc == 0, paused @@ -5518,6 +5554,15 @@ def test_due_successor_replan_replays_until_the_same_task_is_reopened( assert rc == 0, reopened rc, resumed = _run_cli(registry_path, runtime, *guard, "--todo-id", TODO_ID) assert rc == 0, resumed + if handoff_mode == "hard_lease": + assert resumed["normal_delivery_allowed"] is False + assert resumed["execution_preparation"]["state"] == "claim_required" + assert resumed["heartbeat_receipt"]["settlement_identity"] == identity + rc, claimed = _run_cli(registry_path, runtime, "todo", "claim", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--claimed-by", AGENT_ID, "--todo-id", TODO_ID) + assert rc == 0 and claimed["execution_managed"] is True, claimed + rc, resumed = _run_cli(registry_path, runtime, *guard, "--todo-id", TODO_ID) + assert rc == 0 and resumed["execution_preparation"]["state"] == "ready", resumed assert resumed["normal_delivery_allowed"] is True assert resumed["heartbeat_receipt"]["settlement_identity"] == identity assert _spend_run_count(runtime) == 0 diff --git a/tests/control_plane/test_recovery_due_successor_identity.py b/tests/control_plane/test_recovery_due_successor_identity.py index bb32cc76f7..dcbf8f4987 100644 --- a/tests/control_plane/test_recovery_due_successor_identity.py +++ b/tests/control_plane/test_recovery_due_successor_identity.py @@ -12,6 +12,7 @@ isolate_sqlite_runtime, ) from loopx.cli import main as cli_main +from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload from loopx.control_plane.coordination.runtime_shadow import ( build_todo_runtime_shadow_projection, ) @@ -67,7 +68,7 @@ def cli(*args: str) -> tuple[int, dict]: *args, ] ) - return rc, json.loads(capsys.readouterr().out) + return rc, read_cli_policy_payload(capsys.readouterr().out, returncode=rc) initialize_canonical_source_fixture(runtime, GOAL_ID, state_path=state, provider=provider) rc, vision = cli( @@ -133,6 +134,9 @@ def cli(*args: str) -> tuple[int, dict]: "--turn-instance-id", old_turn, ) + rc, claimed = cli("todo", "claim", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--claimed-by", AGENT_ID, "--todo-id", TODO_ID) + assert rc == 0 and claimed["execution_managed"] is True, claimed rc, admitted = cli( "quota", "should-run", @@ -295,8 +299,13 @@ def cli(*args: str) -> tuple[int, dict]: "--status", "open", "--clear-resume-when", + "--reason", + "Resume the verified due successor.", ) assert rc == 0, json.dumps(resumed, indent=2) + rc, claimed = cli("todo", "claim", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--claimed-by", AGENT_ID, "--todo-id", due_id) + assert rc == 0 and claimed["execution_managed"] is True, claimed rc, selected = cli( "quota", "should-run",