From 84377d76b456d443cbb868fe069511b426d4a263 Mon Sep 17 00:00:00 2001 From: Lihua <1017343802@qq.com> Date: Fri, 13 Nov 2026 16:32:45 -0800 Subject: [PATCH 1/2] fix(scheduler): reject superseded heartbeat receipts on the Python follow-up route The Python compatibility route for scheduler-ack-current/scheduler-fail-current checked only that the receipt for --turn-instance-id exists; a receipt already superseded by a newer heartbeat Turn still committed scheduler state, while the native TypeScript transaction rejects it as heartbeat_receipt_stale. Thread the Turn from build_scheduler_followup_payload through record_quota_scheduler_ack / record_quota_scheduler_ack_for_decision / record_quota_scheduler_failure_for_decision into _run_native_followup, which now sends turn_instance_id with require_heartbeat_receipt=true. The TypeScript transaction remains the single owner of the missing/stale/superseded fence (it also rechecks under the scheduler state lock); optional parameters keep the in-process API, which has no Turn, unchanged. Fixes #8605 Signed-off-by: Lihua <1017343802@qq.com> --- .../cli_commands/quota_scheduler_followup.py | 2 + loopx/control_plane/quota/scheduler_ack.py | 21 +- loopx/quota.py | 2 + ...test_scheduler_ack_current_host_binding.py | 200 ++++++++++++++++++ 4 files changed, 222 insertions(+), 3 deletions(-) diff --git a/loopx/cli_commands/quota_scheduler_followup.py b/loopx/cli_commands/quota_scheduler_followup.py index a8ebbe64a1..8bbe506bba 100644 --- a/loopx/cli_commands/quota_scheduler_followup.py +++ b/loopx/cli_commands/quota_scheduler_followup.py @@ -228,6 +228,7 @@ def build_scheduler_followup_payload( failed_rrule=args.failed_rrule, observed_host_rrule=observed_rrule, failure_kind=args.failure_kind, + turn_instance_id=turn_instance_id, ) return record_quota_scheduler_ack( status_payload, @@ -248,4 +249,5 @@ def build_scheduler_followup_payload( scheduler_execution_context=scheduler_context, operator_inbox_urgency_projector=operator_inbox_urgency_projector, before_decision=before_decision, + turn_instance_id=turn_instance_id, ) diff --git a/loopx/control_plane/quota/scheduler_ack.py b/loopx/control_plane/quota/scheduler_ack.py index b9bf45a87d..e973714e4d 100644 --- a/loopx/control_plane/quota/scheduler_ack.py +++ b/loopx/control_plane/quota/scheduler_ack.py @@ -184,14 +184,25 @@ def _run_native_followup( host_facts: dict[str, Any], use_current_hint: bool, reason_summary: str | None, + turn_instance_id: str | None = None, ) -> dict[str, Any]: - """Decision-free compatibility transport for in-process Python callers.""" + """Decision-free compatibility transport for in-process Python callers. + A receipt-bound Turn keeps the native transaction as the single owner of + the missing/stale/superseded receipt fence: it rejects before any + scheduler mutation instead of rebuilding a second staleness rule here. + """ + + receipt_turn_instance_id = ( + str(turn_instance_id).strip() or None + if turn_instance_id is not None + else None + ) request = { "schema_version": _FOLLOWUP_REQUEST_SCHEMA, "runtime_root": str(runtime_root), - "turn_instance_id": None, - "require_heartbeat_receipt": False, + "turn_instance_id": receipt_turn_instance_id, + "require_heartbeat_receipt": receipt_turn_instance_id is not None, "before": before, "host_facts": host_facts, "use_current_hint": use_current_hint, @@ -298,6 +309,7 @@ def record_quota_scheduler_ack_for_decision( generated_at: str | None = None, use_current_hint: bool = False, host_match_observed: bool = False, + turn_instance_id: str | None = None, ) -> dict[str, Any]: state_key = _followup_state_key(before, surface=surface, state_key=state_key) safe_agent_id = normalize_todo_claimed_by(agent_id) @@ -367,6 +379,7 @@ def record_quota_scheduler_ack_for_decision( host_facts=facts, use_current_hint=use_current_hint, reason_summary=reason_summary, + turn_instance_id=turn_instance_id, ) except ValueError as exc: return _failure_payload( @@ -396,6 +409,7 @@ def record_quota_scheduler_failure_for_decision( observed_host_rrule: str | None = None, failure_kind: str = "host_tool_failure", generated_at: str | None = None, + turn_instance_id: str | None = None, ) -> dict[str, Any]: state_key = _followup_state_key(before, surface=surface, state_key=state_key) safe_agent_id = normalize_todo_claimed_by(agent_id) @@ -443,6 +457,7 @@ def record_quota_scheduler_failure_for_decision( host_facts=facts, use_current_hint=False, reason_summary=None, + turn_instance_id=turn_instance_id, ) except ValueError as exc: return _failure_payload( diff --git a/loopx/quota.py b/loopx/quota.py index eded708cd4..060fe86163 100644 --- a/loopx/quota.py +++ b/loopx/quota.py @@ -1213,6 +1213,7 @@ def record_quota_scheduler_ack( | None = None, operator_inbox_urgency_projector: Callable[..., dict[str, Any]] | None = None, before_decision: Mapping[str, Any] | None = None, + turn_instance_id: str | None = None, ) -> dict[str, Any]: safe_goal_id = _validate_goal_id_path_segment(str(goal_id or "")) safe_agent_id = normalize_todo_claimed_by(agent_id) @@ -1247,6 +1248,7 @@ def record_quota_scheduler_ack( reason_summary=reason_summary, use_current_hint=use_current_hint, host_match_observed=host_match_observed, + turn_instance_id=turn_instance_id, ) diff --git a/tests/control_plane/test_scheduler_ack_current_host_binding.py b/tests/control_plane/test_scheduler_ack_current_host_binding.py index 000d5f9ab3..a688309d9b 100644 --- a/tests/control_plane/test_scheduler_ack_current_host_binding.py +++ b/tests/control_plane/test_scheduler_ack_current_host_binding.py @@ -24,6 +24,11 @@ scheduler_execution_context_for_runtime_profile, ) from loopx.control_plane.quota.turn_envelope import build_turn_envelope +from loopx.rollout_event_log import ( + _append_rollout_event_line, + build_rollout_event, + rollout_event_log_path, +) from loopx.status import AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK @@ -442,6 +447,201 @@ def test_scheduler_fail_current_rejects_missing_turn_receipt_without_state_write assert not state_path.exists() +def _append_later_heartbeat_receipt( + runtime_root: Path, + *, + turn_instance_id: str, +) -> None: + """Append a later committed receipt that supersedes earlier Turns.""" + + _append_rollout_event_line( + rollout_event_log_path(runtime_root, GOAL_ID), + build_rollout_event( + goal_id=GOAL_ID, + event_kind="quota_should_run", + agent_id=SCOPED_AGENT_ID, + run_id=turn_instance_id, + status="normal_run", + summary=( + "later heartbeat cycle supersedes earlier turns " + f"({turn_instance_id})" + ), + details={"ok": True, "should_run": True}, + ), + ) + + +def _scheduler_state_files(runtime_root: Path) -> list[Path]: + return [ + path + for goal_dir in sorted((runtime_root / "goals").glob(f"{GOAL_ID}*")) + for path in sorted((goal_dir / "scheduler-state").rglob("*.json")) + ] + + +def test_scheduler_ack_current_rejects_superseded_turn_receipt_without_state_write( + tmp_path: Path, +) -> None: + registry_path, runtime_root, project = write_cli_fixture( + tmp_path / "fixture", + scoped_agents=True, + ) + + seeded_rc, seeded = run_json_cli_result( + "quota", + "should-run", + "--goal-id", + GOAL_ID, + "--agent-id", + SCOPED_AGENT_ID, + "--codex-app", + "--turn-instance-id", + "heartbeat-superseded-ack", + registry_path=registry_path, + runtime_root=runtime_root, + cwd=project, + ) + assert seeded_rc == 0 + assert seeded["ok"] is True + _append_later_heartbeat_receipt( + runtime_root, + turn_instance_id="heartbeat-newer-ack", + ) + + returncode, payload = run_json_cli_result( + "quota", + "scheduler-ack-current", + "--goal-id", + GOAL_ID, + "--agent-id", + SCOPED_AGENT_ID, + "--codex-app", + "--turn-instance-id", + "heartbeat-superseded-ack", + "--execute", + registry_path=registry_path, + runtime_root=runtime_root, + cwd=project, + ) + + assert returncode == 1 + assert payload["ok"] is False + assert payload["status"] == "heartbeat_receipt_stale" + assert payload["state"] == "blocked_receipt" + assert payload["error_code"] == ("SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_STALE") + assert payload["write_performed"] is False + assert payload["scheduler_state_mutated"] is False + assert payload["quota_spend_performed"] is False + assert _scheduler_state_files(runtime_root) == [] + + +def test_scheduler_fail_current_rejects_superseded_turn_receipt_without_state_write( + tmp_path: Path, +) -> None: + registry_path, runtime_root, project = write_cli_fixture( + tmp_path / "fixture", + scoped_agents=True, + ) + + seeded_rc, seeded = run_json_cli_result( + "quota", + "should-run", + "--goal-id", + GOAL_ID, + "--agent-id", + SCOPED_AGENT_ID, + "--codex-app", + "--turn-instance-id", + "heartbeat-superseded-fail", + registry_path=registry_path, + runtime_root=runtime_root, + cwd=project, + ) + assert seeded_rc == 0 + assert seeded["ok"] is True + _append_later_heartbeat_receipt( + runtime_root, + turn_instance_id="heartbeat-newer-fail", + ) + + returncode, payload = run_json_cli_result( + "quota", + "scheduler-fail-current", + "--goal-id", + GOAL_ID, + "--agent-id", + SCOPED_AGENT_ID, + "--codex-app", + "--turn-instance-id", + "heartbeat-superseded-fail", + "--failed-rrule", + "FREQ=MINUTELY;INTERVAL=30", + "--state-key", + APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY, + "--execute", + registry_path=registry_path, + runtime_root=runtime_root, + cwd=project, + ) + + assert returncode == 1 + assert payload["ok"] is False + assert payload["status"] == "heartbeat_receipt_stale" + assert payload["state"] == "blocked_receipt" + assert payload["error_code"] == ("SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_STALE") + assert payload["write_performed"] is False + assert payload["scheduler_state_mutated"] is False + assert payload["quota_spend_performed"] is False + assert _scheduler_state_files(runtime_root) == [] + + +def test_scheduler_ack_current_routes_the_newest_turn_receipt( + tmp_path: Path, +) -> None: + registry_path, runtime_root, project = write_cli_fixture( + tmp_path / "fixture", + scoped_agents=True, + ) + + seeded_rc, seeded = run_json_cli_result( + "quota", + "should-run", + "--goal-id", + GOAL_ID, + "--agent-id", + SCOPED_AGENT_ID, + "--codex-app", + "--turn-instance-id", + "heartbeat-newest", + registry_path=registry_path, + runtime_root=runtime_root, + cwd=project, + ) + assert seeded_rc == 0 + assert seeded["ok"] is True + + returncode, payload = run_json_cli_result( + "quota", + "scheduler-ack-current", + "--goal-id", + GOAL_ID, + "--agent-id", + SCOPED_AGENT_ID, + "--codex-app", + "--turn-instance-id", + "heartbeat-newest", + "--execute", + registry_path=registry_path, + runtime_root=runtime_root, + cwd=project, + ) + + assert returncode == 0 + assert payload["ok"] is True + assert payload["scheduler_state_mutated"] is True + assert Path(payload["scheduler_state_path"]).exists() + + def test_trae_app_scheduler_failure_does_not_read_codex_automation_store( tmp_path: Path, monkeypatch, From 3f41913d4d2985d06305bc66da75221986e34c08 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 15 Nov 2026 22:13:51 +0800 Subject: [PATCH 2/2] Keep scheduler follow-up hint reads out of legacy migration Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/scheduler/state.py | 9 + ...test_scheduler_ack_current_host_binding.py | 206 ++++++------------ 2 files changed, 76 insertions(+), 139 deletions(-) diff --git a/loopx/control_plane/scheduler/state.py b/loopx/control_plane/scheduler/state.py index c00c5a6d84..cfeecac179 100644 --- a/loopx/control_plane/scheduler/state.py +++ b/loopx/control_plane/scheduler/state.py @@ -252,6 +252,7 @@ def _store_result( agent_id: str, surface: str, state_key: str, + migrate_legacy: bool | None = None, ) -> Mapping[str, Any]: params: dict[str, Any] = { "schema_version": SCHEDULER_STATE_STORE_REQUEST_SCHEMA, @@ -265,6 +266,8 @@ def _store_result( } if state is not None: params["state"] = state + if migrate_legacy is not None: + params["migrate_legacy"] = migrate_legacy try: result = effect_runtime_result(f"scheduler.state.{operation}", params) except EffectRuntimeRejected as exc: @@ -287,6 +290,7 @@ def load_scheduler_state( surface: str = CODEX_APP_SURFACE, state_key: str = CODEX_APP_STATEFUL_BACKOFF_STATE_KEY, read_observer: SchedulerStateReadObserver | None = None, + migrate_legacy: bool | None = None, ) -> dict[str, Any] | None: if not agent_id: return None @@ -298,6 +302,7 @@ def load_scheduler_state( agent_id=agent_id, surface=surface, state_key=state_key, + migrate_legacy=migrate_legacy, ) state = result.get("state") if state is not None and not isinstance(state, dict): @@ -317,6 +322,8 @@ def load_app_automation_scheduler_state( surface: str, read_observer: SchedulerStateReadObserver | None = None, ) -> dict[str, Any] | None: + # Hint reconstruction may precede a preview or receipt rejection. + # Legacy migration belongs to the authorized native commit under lock. current = load_scheduler_state( runtime_root, goal_id=goal_id, @@ -324,6 +331,7 @@ def load_app_automation_scheduler_state( surface=surface, state_key=APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY, read_observer=read_observer, + migrate_legacy=False, ) if current is not None or surface != CODEX_APP_SURFACE: return current @@ -337,6 +345,7 @@ def load_app_automation_scheduler_state( surface=surface, state_key=LEGACY_CODEX_APP_STATEFUL_BACKOFF_STATE_KEY, read_observer=read_observer, + migrate_legacy=False, ) diff --git a/tests/control_plane/test_scheduler_ack_current_host_binding.py b/tests/control_plane/test_scheduler_ack_current_host_binding.py index a688309d9b..bbed41e852 100644 --- a/tests/control_plane/test_scheduler_ack_current_host_binding.py +++ b/tests/control_plane/test_scheduler_ack_current_host_binding.py @@ -24,11 +24,6 @@ scheduler_execution_context_for_runtime_profile, ) from loopx.control_plane.quota.turn_envelope import build_turn_envelope -from loopx.rollout_event_log import ( - _append_rollout_event_line, - build_rollout_event, - rollout_event_log_path, -) from loopx.status import AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK @@ -447,30 +442,6 @@ def test_scheduler_fail_current_rejects_missing_turn_receipt_without_state_write assert not state_path.exists() -def _append_later_heartbeat_receipt( - runtime_root: Path, - *, - turn_instance_id: str, -) -> None: - """Append a later committed receipt that supersedes earlier Turns.""" - - _append_rollout_event_line( - rollout_event_log_path(runtime_root, GOAL_ID), - build_rollout_event( - goal_id=GOAL_ID, - event_kind="quota_should_run", - agent_id=SCOPED_AGENT_ID, - run_id=turn_instance_id, - status="normal_run", - summary=( - "later heartbeat cycle supersedes earlier turns " - f"({turn_instance_id})" - ), - details={"ok": True, "should_run": True}, - ), - ) - - def _scheduler_state_files(runtime_root: Path) -> list[Path]: return [ path @@ -479,120 +450,77 @@ def _scheduler_state_files(runtime_root: Path) -> list[Path]: ] -def test_scheduler_ack_current_rejects_superseded_turn_receipt_without_state_write( - tmp_path: Path, +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("operation", ["scheduler-ack-current", "scheduler-fail-current"]) +def test_scheduler_followup_gates_state_migration( + tmp_path: Path, provider: str, operation: str, ) -> None: registry_path, runtime_root, project = write_cli_fixture( - tmp_path / "fixture", - scoped_agents=True, - ) - - seeded_rc, seeded = run_json_cli_result( - "quota", - "should-run", - "--goal-id", - GOAL_ID, - "--agent-id", - SCOPED_AGENT_ID, - "--codex-app", - "--turn-instance-id", - "heartbeat-superseded-ack", - registry_path=registry_path, - runtime_root=runtime_root, - cwd=project, - ) - assert seeded_rc == 0 - assert seeded["ok"] is True - _append_later_heartbeat_receipt( - runtime_root, - turn_instance_id="heartbeat-newer-ack", - ) - - returncode, payload = run_json_cli_result( - "quota", - "scheduler-ack-current", - "--goal-id", - GOAL_ID, - "--agent-id", - SCOPED_AGENT_ID, - "--codex-app", - "--turn-instance-id", - "heartbeat-superseded-ack", - "--execute", - registry_path=registry_path, - runtime_root=runtime_root, - cwd=project, - ) - - assert returncode == 1 - assert payload["ok"] is False - assert payload["status"] == "heartbeat_receipt_stale" - assert payload["state"] == "blocked_receipt" - assert payload["error_code"] == ("SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_STALE") - assert payload["write_performed"] is False - assert payload["scheduler_state_mutated"] is False - assert payload["quota_spend_performed"] is False + tmp_path / "fixture", scoped_agents=True, provider=provider, + ) + + def call(command: str, turn: str, *flags: str): + return run_json_cli_result( + "quota", command, "--goal-id", GOAL_ID, "--agent-id", SCOPED_AGENT_ID, + "--codex-app", "--app-automation-current-rrule", "FREQ=MINUTELY;INTERVAL=5", + "--turn-instance-id", turn, *flags, + registry_path=registry_path, runtime_root=runtime_root, cwd=project, + ) + + stale, current = "heartbeat-older", "heartbeat-current" + for turn in (stale, current): + rc, guard = call("should-run", turn) + assert rc == 0 and guard["ok"] is True + assert guard["heartbeat_receipt"]["turn_instance_id"] == turn + app = guard["scheduler_hint"]["app_automation"] + key = APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY + flags = ["--state-key", key] + if operation == "scheduler-fail-current": + flags.extend(["--failed-rrule", app["recommended_rrule"]]) + + def reject_stale(*execution: str) -> None: + rc, payload = call(operation, stale, *flags, *execution) + assert rc == 1 and payload["ok"] is False + assert payload["status"] == "heartbeat_receipt_stale" + assert payload["state"] == "blocked_receipt" + assert payload["error_code"] == "SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_STALE" + assert payload["write_performed"] is False + assert payload["scheduler_state_mutated"] is False + assert payload["quota_spend_performed"] is False + + reject_stale("--execute") assert _scheduler_state_files(runtime_root) == [] - -def test_scheduler_fail_current_rejects_superseded_turn_receipt_without_state_write( - tmp_path: Path, -) -> None: - registry_path, runtime_root, project = write_cli_fixture( - tmp_path / "fixture", - scoped_agents=True, - ) - - seeded_rc, seeded = run_json_cli_result( - "quota", - "should-run", - "--goal-id", - GOAL_ID, - "--agent-id", - SCOPED_AGENT_ID, - "--codex-app", - "--turn-instance-id", - "heartbeat-superseded-fail", - registry_path=registry_path, - runtime_root=runtime_root, - cwd=project, - ) - assert seeded_rc == 0 - assert seeded["ok"] is True - _append_later_heartbeat_receipt( - runtime_root, - turn_instance_id="heartbeat-newer-fail", - ) - - returncode, payload = run_json_cli_result( - "quota", - "scheduler-fail-current", - "--goal-id", - GOAL_ID, - "--agent-id", - SCOPED_AGENT_ID, - "--codex-app", - "--turn-instance-id", - "heartbeat-superseded-fail", - "--failed-rrule", - "FREQ=MINUTELY;INTERVAL=30", - "--state-key", - APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY, - "--execute", - registry_path=registry_path, - runtime_root=runtime_root, - cwd=project, - ) - - assert returncode == 1 - assert payload["ok"] is False - assert payload["status"] == "heartbeat_receipt_stale" - assert payload["state"] == "blocked_receipt" - assert payload["error_code"] == ("SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_STALE") - assert payload["write_performed"] is False - assert payload["scheduler_state_mutated"] is False - assert payload["quota_spend_performed"] is False - assert _scheduler_state_files(runtime_root) == [] + canonical = scheduler_state_path( + runtime_root, goal_id=GOAL_ID, agent_id=SCOPED_AGENT_ID, state_key=key, + ) + legacy = ( + runtime_root / "goals" / GOAL_ID / "scheduler-state" + / SCOPED_AGENT_ID / "codex_app" / canonical.name + ) + assert legacy != canonical + legacy.parent.mkdir(parents=True, exist_ok=True) + legacy.write_text(json.dumps(build_scheduler_state( + goal_id=GOAL_ID, agent_id=SCOPED_AGENT_ID, state_key=key, + reset_token="earlier-upgrade", identity_signature="earlier-host", + progression_index=0, progression_minutes=[3, 10, 30], + last_applied_rrule="FREQ=MINUTELY;INTERVAL=5", + updated_at="2026-01-01T00:00:00Z", + )), encoding="utf-8") + original = legacy.read_bytes() + + for execution in ([], ["--execute"]): + reject_stale(*execution) + assert legacy.read_bytes() == original + assert not canonical.exists() + rc, preview = call(operation, current, *flags) + assert rc == 0 and preview["ok"] is True + assert legacy.read_bytes() == original and not canonical.exists() + + rc, committed = call(operation, current, *flags, "--execute") + assert rc == 0 and committed["ok"] is True + assert committed["scheduler_state_mutated"] is True + assert canonical.exists() and not legacy.exists() def test_scheduler_ack_current_routes_the_newest_turn_receipt(