fix(scheduler): reject superseded heartbeat receipts on the Python follow-up route

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,
     )

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(

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,13 +322,16 @@ 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,
         agent_id=agent_id,
         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,
     )
 
 

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,
     )
 
 

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,