From 277dab815eb8dba20ff1ca6256009589a1675a85 Mon Sep 17 00:00:00 2001 From: song Date: Mon, 12 Oct 2026 11:14:49 +0800 Subject: [PATCH] test(settlement): observe current Host before scheduler acknowledgement Signed-off-by: song --- .../test_quota_settlement_cli.py | 45 ++++++++++++++++++- 1 file changed, 43 insertions(+), 2 deletions(-) diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 792582f8f6..99ed64d67e 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -12,6 +12,7 @@ import pytest +from examples.control_plane.quota_plan_fixtures import write_codex_heartbeat_fixture from loopx.bootstrap_command_pack import build_start_goal_guided_packet from loopx.control_plane.work_items.delivery_outcome import ( PROGRESS_DELIVERY_OUTCOMES, @@ -47,6 +48,36 @@ SELECTED_REPLAN_TODO_ID = "todo_chain_000000000000" +@pytest.fixture +def scheduler_host_home(tmp_path: Path, monkeypatch) -> Path: + root = tmp_path / "codex-host" + monkeypatch.setenv("CODEX_HOME", str(root)) + monkeypatch.setenv("CODEX_THREAD_ID", "scheduler-fixture") + return root + + +def _prepare_scheduler_host_readback( + registry_path: Path, runtime: Path, host_root: Path, ack_args: list[str], +) -> None: + """Exercise the real Host reader over a synthetic manifest, not a live App.""" + requested = ack_args[ack_args.index("--applied-rrule") + 1] + wrong = "FREQ=MINUTELY;INTERVAL=60" + assert requested != wrong + for observed in (None, wrong): + if observed is not None: + write_codex_heartbeat_fixture(host_root, goal_id=GOAL_ID, agent_id=AGENT_ID, rrule=observed) + rc, refused = _run_cli(registry_path, runtime, *ack_args) + assert rc == 1, refused + assert refused["scheduler_commit"]["reason_code"] == "rrule_mismatch", refused + assert refused["scheduler_commit"]["written"] is False + assert refused["scheduler_commit"]["observed_host_rrule"] == (observed or "") + assert load_scheduler_state(runtime, goal_id=GOAL_ID, agent_id=AGENT_ID, + state_key=APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY) is None + assert _spend_run_count(runtime) == 1 + # Only change the isolated Host observation, not the original receipt/hint. + write_codex_heartbeat_fixture(host_root, goal_id=GOAL_ID, agent_id=AGENT_ID, rrule=requested) + + def _assert_action_selection_recovery_projections(payload: dict[str, Any]) -> None: from loopx.control_plane.quota.turn_envelope import build_turn_envelope from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority @@ -2697,7 +2728,7 @@ def test_prior_host_closeout_reads_archived_exact_todo( def test_standard_codex_app_settlement_is_receipted_and_idempotent( - tmp_path: Path, + tmp_path: Path, scheduler_host_home: Path, ) -> None: project, runtime, registry_path = _write_fixture(tmp_path) _configure_read_only_todo(project) @@ -2924,11 +2955,16 @@ def test_standard_codex_app_settlement_is_receipted_and_idempotent( assert fresh_turn_rc == 0, fresh_turn assert fresh_turn["selected_todo"]["todo_id"] == successor_id fresh_ack_args = fresh_turn["scheduler_hint"]["app_automation"]["ack_hint"]["cli_args"] + _prepare_scheduler_host_readback(registry_path, runtime, scheduler_host_home, fresh_ack_args) fresh_ack_rc, fresh_ack = _run_cli(registry_path, runtime, *fresh_ack_args) assert fresh_ack_rc == 0, fresh_ack assert fresh_ack["schema_version"] == "loopx_scheduler_host_followup_result_v0" assert fresh_ack["mode"] == "scheduler-ack" assert fresh_ack["scheduler_commit"]["written"] is True + assert load_scheduler_state(runtime, goal_id=GOAL_ID, agent_id=AGENT_ID, + state_key=APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY)["last_applied_rrule"] == ( + fresh_ack_args[fresh_ack_args.index("--applied-rrule") + 1] + ) scheduler_bytes = Path(fresh_ack["scheduler_state_path"]).read_bytes() replay_ack_rc, replay_ack = _run_cli(registry_path, runtime, *fresh_ack_args) assert replay_ack_rc == 0, replay_ack @@ -5589,7 +5625,7 @@ def test_pending_action_selection_does_not_commit_after_new_user_gate( @pytest.mark.parametrize("write_vision", [True, False], ids=["qualified-vision", "missing-vision"]) def test_todoless_autonomous_replan_settles_quota_refresh_spend_chain( - tmp_path: Path, write_vision: bool, + tmp_path: Path, write_vision: bool, scheduler_host_home: Path, ) -> None: project, runtime, registry_path = _write_fixture(tmp_path) _configure_autonomous_replan_fixture(project, runtime, registry_path) @@ -5761,6 +5797,7 @@ def test_todoless_autonomous_replan_settles_quota_refresh_spend_chain( ] == "autonomous_replan" assert _spend_run_count(runtime) == 1 + _prepare_scheduler_host_readback(registry_path, runtime, scheduler_host_home, original_scheduler_ack_args) ack_rc, ack = _run_cli( registry_path, runtime, @@ -5775,6 +5812,10 @@ def test_todoless_autonomous_replan_settles_quota_refresh_spend_chain( assert ack["scheduler_commit"]["written"] is True assert ack["scheduler_state_mutated"] is True assert ack["appended"] is False + assert load_scheduler_state(runtime, goal_id=GOAL_ID, agent_id=AGENT_ID, + state_key=APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY)["last_applied_rrule"] == ( + original_scheduler_ack_args[original_scheduler_ack_args.index("--applied-rrule") + 1] + ) scheduler_bytes = Path(ack["scheduler_state_path"]).read_bytes() ack_replay_rc, ack_replay = _run_cli( registry_path, runtime, *original_scheduler_ack_args