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/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/loopx/quota.py b/loopx/quota.py index 3836b4b82d..4dac35fb65 100644 --- a/loopx/quota.py +++ b/loopx/quota.py @@ -1244,6 +1244,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) @@ -1278,6 +1279,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..bbed41e852 100644 --- a/tests/control_plane/test_scheduler_ack_current_host_binding.py +++ b/tests/control_plane/test_scheduler_ack_current_host_binding.py @@ -442,6 +442,134 @@ def test_scheduler_fail_current_rejects_missing_turn_receipt_without_state_write assert not state_path.exists() +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")) + ] + + +@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, 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) == [] + + 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( + 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,