fix(scheduler): apply the heartbeat receipt fence to Python follow-ups

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,21 @@ 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 Python callers.
+
+    A caller that already bound the follow-up to a Turn passes it here, so the
+    native transaction applies the same heartbeat-receipt fence (missing or
+    superseded) as the generated route. Callers without a Turn keep the
+    receipt-less compatibility request.
+    """
 
     request = {
         "schema_version": _FOLLOWUP_REQUEST_SCHEMA,
         "runtime_root": str(runtime_root),
-        "turn_instance_id": None,
-        "require_heartbeat_receipt": False,
+        "turn_instance_id": turn_instance_id,
+        "require_heartbeat_receipt": turn_instance_id is not None,
         "before": before,
         "host_facts": host_facts,
         "use_current_hint": use_current_hint,
@@ -298,6 +305,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 +375,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 +405,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 +453,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/quota.py

@@ -1205,6 +1205,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)
@@ -1239,6 +1240,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_stale_receipt_gate.py

@@ -0,0 +1,121 @@
+"""The Python follow-up route applies the same receipt fence as the native route.
+
+A Turn id binds the follow-up to that Turn's heartbeat receipt. The native
+transaction rejects a receipt that is missing *or* superseded; the compatibility
+route used to check only for absence, so an ACK for a Turn whose receipt a later
+`quota should-run` had superseded still rewrote scheduler state.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from examples.control_plane.quota_plan_fixtures import (
+    SCOPED_AGENT_ID,
+    write_cli_fixture,
+)
+from loopx.control_plane.scheduler.state import (
+    APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY as APP_KEY,
+    load_scheduler_state,
+    scheduler_state_path,
+)
+from loopx.control_plane.testing.canary_harness import run_json_cli_result
+from loopx.rollout_event_log import (
+    append_rollout_event,
+    build_rollout_event,
+    rollout_event_log_path,
+)
+
+GOAL = "needs-operator"
+TURN_1 = "stale-receipt-turn-1"
+TURN_2 = "stale-receipt-turn-2"
+
+
+def _fixture(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> tuple[Path, Path, Path]:
+    registry, runtime, project = write_cli_fixture(
+        tmp_path / "fixture", scoped_agents=True, provider="file"
+    )
+    monkeypatch.setenv("CODEX_HOME", str(tmp_path / "codex-home"))
+    monkeypatch.setenv("CODEX_THREAD_ID", "scheduler-fixture")
+    return registry, runtime, project
+
+
+def _should_run(
+    registry: Path, runtime: Path, project: Path, turn: str
+) -> dict:
+    code, result = run_json_cli_result(
+        "quota", "should-run", "--goal-id", GOAL,
+        "--agent-id", SCOPED_AGENT_ID, "--codex-app",
+        "--turn-instance-id", turn,
+        registry_path=registry, runtime_root=runtime, cwd=project,
+    )
+    assert code == 0, result
+    return result
+
+
+def _ack(
+    registry: Path, runtime: Path, project: Path, turn: str
+) -> tuple[int, dict]:
+    return run_json_cli_result(
+        "quota", "scheduler-ack-current", "--goal-id", GOAL,
+        "--agent-id", SCOPED_AGENT_ID, "--codex-app",
+        "--turn-instance-id", turn, "--execute",
+        registry_path=registry, runtime_root=runtime, cwd=project,
+    )
+
+
+def _supercede(runtime: Path, turn: str) -> None:
+    """Append a later receipt for ``turn`` on the same goal and agent."""
+
+    append_rollout_event(
+        rollout_event_log_path(runtime, GOAL),
+        build_rollout_event(
+            goal_id=GOAL,
+            event_kind="quota_should_run",
+            agent_id=SCOPED_AGENT_ID,
+            run_id=turn,
+            status="normal_run",
+            details={"ok": True, "should_run": True},
+        ),
+    )
+
+
+def test_superseded_receipt_cannot_commit_a_scheduler_ack(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    registry, runtime, project = _fixture(tmp_path, monkeypatch)
+    _should_run(registry, runtime, project, TURN_1)
+    _supercede(runtime, TURN_2)
+
+    code, result = _ack(registry, runtime, project, TURN_1)
+
+    assert code == 1, result
+    assert result["ok"] is False
+    assert result["status"] == "heartbeat_receipt_stale"
+    assert result["state"] == "blocked_receipt"
+    assert result["scheduler_state_mutated"] is False
+    assert not scheduler_state_path(
+        runtime, goal_id=GOAL, agent_id=SCOPED_AGENT_ID, state_key=APP_KEY
+    ).exists()
+
+
+def test_fresh_receipt_still_commits_the_ack(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    registry, runtime, project = _fixture(tmp_path, monkeypatch)
+    _should_run(registry, runtime, project, TURN_1)
+
+    code, result = _ack(registry, runtime, project, TURN_1)
+
+    assert code == 0, result
+    assert result["ok"] is True
+    assert result["scheduler_state_mutated"] is True
+    state = load_scheduler_state(
+        runtime, goal_id=GOAL, agent_id=SCOPED_AGENT_ID, state_key=APP_KEY
+    )
+    assert state is not None
+    assert state["last_applied_rrule"]