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..2fc8939725 100644 --- a/loopx/control_plane/quota/scheduler_ack.py +++ b/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( diff --git a/loopx/quota.py b/loopx/quota.py index 1ce7e8fa87..12cc5bf22c 100644 --- a/loopx/quota.py +++ b/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, ) diff --git a/tests/control_plane/test_scheduler_stale_receipt_gate.py b/tests/control_plane/test_scheduler_stale_receipt_gate.py new file mode 100644 index 0000000000..4196b3f940 --- /dev/null +++ b/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"]