From 0379886ae2733c8d613faa1c1b4ad4f35b4c87f1 Mon Sep 17 00:00:00 2001 From: gcl-coder Date: Thu, 12 Nov 2026 13:00:00 +0000 Subject: [PATCH 1/4] fix(scheduler): fence Python follow-ups on superseded heartbeat receipts Signed-off-by: gcl-coder --- .../cli_commands/quota_scheduler_followup.py | 89 ++++--- .../control_plane/quota/heartbeat_receipt.py | 62 +++++ loopx/control_plane/quota/scheduler_ack.py | 9 +- loopx/quota.py | 2 + .../test_scheduler_heartbeat_receipt_fence.py | 230 ++++++++++++++++++ 5 files changed, 362 insertions(+), 30 deletions(-) create mode 100644 tests/control_plane/test_scheduler_heartbeat_receipt_fence.py diff --git a/loopx/cli_commands/quota_scheduler_followup.py b/loopx/cli_commands/quota_scheduler_followup.py index a8ebbe64a1..eeeb53ee05 100644 --- a/loopx/cli_commands/quota_scheduler_followup.py +++ b/loopx/cli_commands/quota_scheduler_followup.py @@ -12,9 +12,12 @@ repository_delivery_interaction_hook, ) from ..control_plane.quota.heartbeat_receipt import ( + HEARTBEAT_RECEIPT_STATUS_FRESH, + HEARTBEAT_RECEIPT_STATUS_STALE, find_heartbeat_receipt, heartbeat_receipt_settlement_replan_obligation_id, heartbeat_receipt_settlement_todo_id, + heartbeat_receipt_status, ) from ..control_plane.quota.live_decision import build_live_quota_should_run_decision from ..control_plane.quota.scheduler_ack import ( @@ -105,34 +108,62 @@ def build_scheduler_followup_payload( if turn_instance_id else None ) - if turn_instance_id and heartbeat_receipt is None: - return { - "ok": False, - "schema_version": "quota_scheduler_followup_receipt_failure_v0", - "mode": args.quota_command, - "goal_id": args.goal_id, - "agent_id": args.agent_id, - "turn_instance_id": turn_instance_id, - "decision": "skip", - "should_run": False, - "status": "heartbeat_receipt_missing", - "state": "blocked_receipt", - "error_code": "SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_MISSING", - "reason": ( - "scheduler follow-up requires the originating heartbeat receipt; " - "refusing to rebuild authority from the current live frontier" - ), - "retry_guidance": ( - "retry with the exact Turn id from a committed quota should-run " - "heartbeat receipt" - ), - "write_performed": False, - "appended": False, - "registry_mutated": False, - "scheduler_state_mutated": False, - "quota_spend_performed": False, - "delivery_outcome": "surface_only", - } + if turn_instance_id: + heartbeat_status = heartbeat_receipt_status( + runtime_root, + goal_id=args.goal_id, + agent_id=args.agent_id, + turn_instance_id=turn_instance_id, + goal_ref=goal_ref, + ) + if heartbeat_status != HEARTBEAT_RECEIPT_STATUS_FRESH: + superseded = heartbeat_status == HEARTBEAT_RECEIPT_STATUS_STALE + return { + "ok": False, + "schema_version": "quota_scheduler_followup_receipt_failure_v0", + "mode": args.quota_command, + "goal_id": args.goal_id, + "agent_id": args.agent_id, + "turn_instance_id": turn_instance_id, + "decision": "skip", + "should_run": False, + "status": ( + "heartbeat_receipt_stale" + if superseded + else "heartbeat_receipt_missing" + ), + "state": "blocked_receipt", + "error_code": ( + "SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_STALE" + if superseded + else "SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_MISSING" + ), + "reason": ( + "scheduler follow-up Turn was superseded by a newer quota " + "should-run receipt; refusing stale host writeback" + if superseded + else ( + "scheduler follow-up requires the originating heartbeat " + "receipt; refusing to rebuild authority from the current " + "live frontier" + ) + ), + "retry_guidance": ( + "rerun quota should-run and use the ACK/failure command emitted " + "by the newest Turn" + if superseded + else ( + "retry with the exact Turn id from a committed quota " + "should-run heartbeat receipt" + ) + ), + "write_performed": False, + "appended": False, + "registry_mutated": False, + "scheduler_state_mutated": False, + "quota_spend_performed": False, + "delivery_outcome": "surface_only", + } observed_rrule = str( getattr(args, "app_automation_current_rrule", None) @@ -228,6 +259,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 +280,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/heartbeat_receipt.py b/loopx/control_plane/quota/heartbeat_receipt.py index 2c3d7b36d4..565f49c26d 100644 --- a/loopx/control_plane/quota/heartbeat_receipt.py +++ b/loopx/control_plane/quota/heartbeat_receipt.py @@ -197,6 +197,68 @@ def find_heartbeat_receipt( ) +HEARTBEAT_RECEIPT_STATUS_MISSING = "missing" +HEARTBEAT_RECEIPT_STATUS_STALE = "stale" +HEARTBEAT_RECEIPT_STATUS_FRESH = "fresh" + + +def heartbeat_receipt_status( + runtime_root: Path, + *, + goal_id: str, + agent_id: str, + turn_instance_id: str, + goal_ref: Mapping[str, object] | None = None, +) -> str: + """Report whether this Turn's receipt is missing, fresh, or superseded. + + The native scheduler fence rejects a follow-up once a later receipt for the + same Goal and Agent exists, or once a later committed normal-run + should-run audit proves the Turn was superseded. The Python compatibility + route reads the same rollout log, so it must answer the same question + instead of accepting any receipt it can find. + """ + + events = load_rollout_events(rollout_event_log_path(runtime_root, goal_id)) + receipts = _heartbeat_receipt_events( + events, + goal_id=goal_id, + agent_id=agent_id, + turn_instance_id=turn_instance_id, + goal_ref=goal_ref, + ) + if not receipts: + return HEARTBEAT_RECEIPT_STATUS_MISSING + start = next( + (index for index, event in enumerate(events) if event is receipts[0]), + None, + ) + if start is None: + return HEARTBEAT_RECEIPT_STATUS_MISSING + for event in events[start + 1:]: + if ( + event.get("event_kind") != "quota_should_run" + or str(event.get("goal_id") or "") != goal_id + or str(event.get("agent_id") or "") != agent_id + ): + continue + raw_run_id = event.get("run_id") + run_id = raw_run_id.strip() if isinstance(raw_run_id, str) else "" + if run_id: + if run_id != turn_instance_id: + return HEARTBEAT_RECEIPT_STATUS_STALE + continue + details = event.get("details") + if ( + event.get("status") == "normal_run" + and isinstance(details, Mapping) + and details.get("ok") is True + and details.get("should_run") is True + ): + return HEARTBEAT_RECEIPT_STATUS_STALE + return HEARTBEAT_RECEIPT_STATUS_FRESH + + def ensure_turn_heartbeat_settlement_receipt( runtime_root: Path, identity: SettlementIdentity, diff --git a/loopx/control_plane/quota/scheduler_ack.py b/loopx/control_plane/quota/scheduler_ack.py index b9bf45a87d..f0aa3fb937 100644 --- a/loopx/control_plane/quota/scheduler_ack.py +++ b/loopx/control_plane/quota/scheduler_ack.py @@ -184,14 +184,15 @@ 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.""" 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": bool(turn_instance_id), "before": before, "host_facts": host_facts, "use_current_hint": use_current_hint, @@ -298,6 +299,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 +369,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 +399,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 +447,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_heartbeat_receipt_fence.py b/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py new file mode 100644 index 0000000000..d788492792 --- /dev/null +++ b/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py @@ -0,0 +1,230 @@ +from __future__ import annotations + +import json +from argparse import Namespace +from pathlib import Path + +import pytest + +from loopx.cli_commands import quota_scheduler_followup +from loopx.control_plane.quota import heartbeat_receipt +from loopx.control_plane.quota.scheduler_ack import _run_native_followup +from loopx.rollout_event_log import build_rollout_event, rollout_event_log_path + + +GOAL_ID = "receipt-fence" +AGENT_ID = "agent-a" + + +def _append(runtime_root: Path, **fields: object) -> None: + path = rollout_event_log_path(runtime_root, GOAL_ID) + path.parent.mkdir(parents=True, exist_ok=True) + event = build_rollout_event(goal_id=GOAL_ID, agent_id=AGENT_ID, **fields) + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(event, ensure_ascii=False) + "\n") + + +def _status(runtime_root: Path, turn: str) -> str: + return heartbeat_receipt.heartbeat_receipt_status( + runtime_root, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + turn_instance_id=turn, + ) + + +def test_receipt_status_distinguishes_missing_fresh_and_superseded( + tmp_path: Path, +) -> None: + runtime_root = tmp_path / "runtime" + assert _status(runtime_root, "turn-1") == "missing" + + _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") + assert _status(runtime_root, "turn-1") == "fresh" + assert _status(runtime_root, "turn-2") == "missing" + + _append(runtime_root, event_kind="quota_should_run", run_id="turn-2") + assert _status(runtime_root, "turn-1") == "stale" + assert _status(runtime_root, "turn-2") == "fresh" + + +def test_later_normal_run_audit_supersedes_the_receipt(tmp_path: Path) -> None: + runtime_root = tmp_path / "runtime" + _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") + _append( + runtime_root, + event_kind="quota_should_run", + status="normal_run", + details={"ok": True, "should_run": True}, + ) + assert _status(runtime_root, "turn-1") == "stale" + + +def test_unrelated_normal_run_audit_does_not_supersede(tmp_path: Path) -> None: + runtime_root = tmp_path / "runtime" + _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") + _append( + runtime_root, + event_kind="quota_should_run", + status="normal_run", + details={"ok": True, "should_run": False}, + ) + assert _status(runtime_root, "turn-1") == "fresh" + + +def test_native_followup_request_carries_the_turn_fence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + captured: dict[str, object] = {} + + class _Completed: + returncode = 0 + stdout = '{"ok": true}' + stderr = "" + + class _Probe: + returncode = 0 + stdout = "v24.19.0" + stderr = "" + + def _run(command, **kwargs): + if "input" not in kwargs: + return _Probe() + captured["request"] = json.loads(kwargs["input"]) + return _Completed() + + monkeypatch.setattr( + "loopx.control_plane.quota.scheduler_ack.subprocess.run", _run + ) + _run_native_followup( + runtime_root=tmp_path, + before={}, + host_facts={}, + use_current_hint=False, + reason_summary=None, + turn_instance_id="turn-1", + ) + request = captured["request"] + assert request["turn_instance_id"] == "turn-1" + assert request["require_heartbeat_receipt"] is True + + _run_native_followup( + runtime_root=tmp_path, + before={}, + host_facts={}, + use_current_hint=False, + reason_summary=None, + ) + request = captured["request"] + assert request["turn_instance_id"] is None + assert request["require_heartbeat_receipt"] is False + + +def test_ack_route_forwards_the_turn_fence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from loopx import quota + + captured: dict[str, object] = {} + + def _capture(before, **kwargs): + captured.update(kwargs) + return {"ok": True} + + monkeypatch.setattr( + quota, "record_quota_scheduler_ack_for_decision", _capture + ) + quota.record_quota_scheduler_ack( + {"runtime_root": str(tmp_path)}, + goal_id="goals", + agent_id="agent-a", + before_decision={"decision": {}}, + turn_instance_id="turn-1", + ) + assert captured["turn_instance_id"] == "turn-1" + + +def test_superseded_receipt_blocks_before_any_scheduler_mutation( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + runtime_root = tmp_path / "runtime" + _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") + _append(runtime_root, event_kind="quota_should_run", run_id="turn-2") + calls: list[dict[str, object]] = [] + + def _ack(_status, **kwargs): + calls.append(kwargs) + return {"ok": True} + + monkeypatch.setattr( + quota_scheduler_followup, "record_quota_scheduler_ack", _ack + ) + args = Namespace( + quota_command="scheduler-ack-current", + goal_id=GOAL_ID, + agent_id=AGENT_ID, + ) + payload = quota_scheduler_followup.build_scheduler_followup_payload( + {}, + args, + registry_path=tmp_path / "registry.json", + runtime_root=runtime_root, + turn_instance_id="turn-1", + scheduler_context=None, + operator_inbox_urgency_projector=lambda **_kwargs: {}, + ) + + assert payload["ok"] is False + assert payload["status"] == "heartbeat_receipt_stale" + assert payload["error_code"] == "SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_STALE" + assert payload["scheduler_state_mutated"] is False + assert calls == [] + + +def test_fresh_receipt_still_reaches_the_ack_with_its_turn_fence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + runtime_root = tmp_path / "runtime" + _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") + calls: list[dict[str, object]] = [] + + def _ack(_status, **kwargs): + calls.append(kwargs) + return {"ok": True} + + monkeypatch.setattr( + quota_scheduler_followup, "record_quota_scheduler_ack", _ack + ) + monkeypatch.setattr( + quota_scheduler_followup, + "_build_scheduler_followup_decision", + lambda *_args, **_kwargs: {"effective_action": "execute_todo"}, + ) + args = Namespace( + quota_command="scheduler-ack-current", + goal_id=GOAL_ID, + agent_id=AGENT_ID, + codex_app_current_rrule=None, + applied_rrule="FREQ=MINUTELY;INTERVAL=3", + host_match_observed=False, + execute=True, + surface="trae_app", + state_key=None, + available_capabilities=[], + reset_token=None, + identity_signature=None, + reason_summary=None, + use_current_hint=False, + ) + payload = quota_scheduler_followup.build_scheduler_followup_payload( + {}, + args, + registry_path=tmp_path / "registry.json", + runtime_root=runtime_root, + turn_instance_id="turn-1", + scheduler_context=None, + operator_inbox_urgency_projector=lambda **_kwargs: {}, + ) + + assert payload == {"ok": True} + assert calls and calls[0]["turn_instance_id"] == "turn-1" From 346cfa19109eaa32ea499558e8868ae7f523608b Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Fri, 13 Nov 2026 00:55:29 +0000 Subject: [PATCH 2/4] fix(scheduler): let the shared receipt reader own follow-up freshness Strict UTF-8 decoding in the shared receipt-log snapshot keeps every replacement-decoded row from becoming a different, valid receipt. Damaged complete rows and unterminated tails are recorded as strict errors and skipped by the tolerant receipt reader, while settlement readback still fails closed. Delete the Python compatibility route's duplicate freshness scan so the native lock-held fence answers missing/stale once, and keep passing the originating Turn id through the ACK and host-failure transport. Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- .../cli_commands/quota_scheduler_followup.py | 60 -------- .../control_plane/quota/heartbeat_receipt.py | 62 -------- .../runtime/receipt_log_snapshot.ts | 27 +++- .../test_scheduler_heartbeat_receipt_fence.py | 137 +++++++++--------- .../receipt_log_snapshot.test.ts | 36 +++++ .../scheduler_heartbeat_followup.test.ts | 72 +++++++++ 6 files changed, 204 insertions(+), 190 deletions(-) diff --git a/loopx/cli_commands/quota_scheduler_followup.py b/loopx/cli_commands/quota_scheduler_followup.py index eeeb53ee05..c0f20d6444 100644 --- a/loopx/cli_commands/quota_scheduler_followup.py +++ b/loopx/cli_commands/quota_scheduler_followup.py @@ -12,12 +12,9 @@ repository_delivery_interaction_hook, ) from ..control_plane.quota.heartbeat_receipt import ( - HEARTBEAT_RECEIPT_STATUS_FRESH, - HEARTBEAT_RECEIPT_STATUS_STALE, find_heartbeat_receipt, heartbeat_receipt_settlement_replan_obligation_id, heartbeat_receipt_settlement_todo_id, - heartbeat_receipt_status, ) from ..control_plane.quota.live_decision import build_live_quota_should_run_decision from ..control_plane.quota.scheduler_ack import ( @@ -108,63 +105,6 @@ def build_scheduler_followup_payload( if turn_instance_id else None ) - if turn_instance_id: - heartbeat_status = heartbeat_receipt_status( - runtime_root, - goal_id=args.goal_id, - agent_id=args.agent_id, - turn_instance_id=turn_instance_id, - goal_ref=goal_ref, - ) - if heartbeat_status != HEARTBEAT_RECEIPT_STATUS_FRESH: - superseded = heartbeat_status == HEARTBEAT_RECEIPT_STATUS_STALE - return { - "ok": False, - "schema_version": "quota_scheduler_followup_receipt_failure_v0", - "mode": args.quota_command, - "goal_id": args.goal_id, - "agent_id": args.agent_id, - "turn_instance_id": turn_instance_id, - "decision": "skip", - "should_run": False, - "status": ( - "heartbeat_receipt_stale" - if superseded - else "heartbeat_receipt_missing" - ), - "state": "blocked_receipt", - "error_code": ( - "SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_STALE" - if superseded - else "SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_MISSING" - ), - "reason": ( - "scheduler follow-up Turn was superseded by a newer quota " - "should-run receipt; refusing stale host writeback" - if superseded - else ( - "scheduler follow-up requires the originating heartbeat " - "receipt; refusing to rebuild authority from the current " - "live frontier" - ) - ), - "retry_guidance": ( - "rerun quota should-run and use the ACK/failure command emitted " - "by the newest Turn" - if superseded - else ( - "retry with the exact Turn id from a committed quota " - "should-run heartbeat receipt" - ) - ), - "write_performed": False, - "appended": False, - "registry_mutated": False, - "scheduler_state_mutated": False, - "quota_spend_performed": False, - "delivery_outcome": "surface_only", - } - observed_rrule = str( getattr(args, "app_automation_current_rrule", None) or getattr(args, "codex_app_current_rrule", None) diff --git a/loopx/control_plane/quota/heartbeat_receipt.py b/loopx/control_plane/quota/heartbeat_receipt.py index 565f49c26d..2c3d7b36d4 100644 --- a/loopx/control_plane/quota/heartbeat_receipt.py +++ b/loopx/control_plane/quota/heartbeat_receipt.py @@ -197,68 +197,6 @@ def find_heartbeat_receipt( ) -HEARTBEAT_RECEIPT_STATUS_MISSING = "missing" -HEARTBEAT_RECEIPT_STATUS_STALE = "stale" -HEARTBEAT_RECEIPT_STATUS_FRESH = "fresh" - - -def heartbeat_receipt_status( - runtime_root: Path, - *, - goal_id: str, - agent_id: str, - turn_instance_id: str, - goal_ref: Mapping[str, object] | None = None, -) -> str: - """Report whether this Turn's receipt is missing, fresh, or superseded. - - The native scheduler fence rejects a follow-up once a later receipt for the - same Goal and Agent exists, or once a later committed normal-run - should-run audit proves the Turn was superseded. The Python compatibility - route reads the same rollout log, so it must answer the same question - instead of accepting any receipt it can find. - """ - - events = load_rollout_events(rollout_event_log_path(runtime_root, goal_id)) - receipts = _heartbeat_receipt_events( - events, - goal_id=goal_id, - agent_id=agent_id, - turn_instance_id=turn_instance_id, - goal_ref=goal_ref, - ) - if not receipts: - return HEARTBEAT_RECEIPT_STATUS_MISSING - start = next( - (index for index, event in enumerate(events) if event is receipts[0]), - None, - ) - if start is None: - return HEARTBEAT_RECEIPT_STATUS_MISSING - for event in events[start + 1:]: - if ( - event.get("event_kind") != "quota_should_run" - or str(event.get("goal_id") or "") != goal_id - or str(event.get("agent_id") or "") != agent_id - ): - continue - raw_run_id = event.get("run_id") - run_id = raw_run_id.strip() if isinstance(raw_run_id, str) else "" - if run_id: - if run_id != turn_instance_id: - return HEARTBEAT_RECEIPT_STATUS_STALE - continue - details = event.get("details") - if ( - event.get("status") == "normal_run" - and isinstance(details, Mapping) - and details.get("ok") is True - and details.get("should_run") is True - ): - return HEARTBEAT_RECEIPT_STATUS_STALE - return HEARTBEAT_RECEIPT_STATUS_FRESH - - def ensure_turn_heartbeat_settlement_receipt( runtime_root: Path, identity: SettlementIdentity, diff --git a/loopx/control_plane/runtime/receipt_log_snapshot.ts b/loopx/control_plane/runtime/receipt_log_snapshot.ts index da0495f76e..d5e9f18ba8 100644 --- a/loopx/control_plane/runtime/receipt_log_snapshot.ts +++ b/loopx/control_plane/runtime/receipt_log_snapshot.ts @@ -59,6 +59,19 @@ function freezeRecord(record: JsonObject): JsonObject { return record; } +// One strict decoder for the whole read: replacement characters must never +// become a different, valid-looking receipt identity. `ignoreBOM` matches +// Buffer#toString, so a leading BOM still fails JSON parsing as before. +const STRICT_UTF8 = new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }); + +function strictUtf8(bytes: Uint8Array): string | null { + try { + return STRICT_UTF8.decode(bytes); + } catch { + return null; + } +} + function decodeLine(line: string, schemaVersion?: string): JsonObject { const record = jsonObject(JSON.parse(line)); if (record === null || (schemaVersion !== undefined && @@ -106,9 +119,11 @@ export async function readReceiptLogSnapshot( let batchStart = offset; while (offset < completeBytes) { const end = bytes.indexOf(10, offset); - const line = bytes.subarray(offset, end).toString("utf8"); + const line = strictUtf8(bytes.subarray(offset, end)); lines += 1; - if (line.trim()) { + if (line === null) { + firstErrorLine ??= lines; + } else if (line.trim()) { try { records.push(decodeLine(line, schemaVersion)); } catch { @@ -128,7 +143,13 @@ export async function readReceiptLogSnapshot( }); } remember(key, prefix); - const tail = bytes.subarray(completeBytes).toString("utf8"); + const tail = strictUtf8(bytes.subarray(completeBytes)); + if (tail === null) { + return Object.freeze({ + records: prefix.records, + firstErrorLine: prefix.firstErrorLine ?? prefix.lines + 1, + }); + } if (!tail.trim()) return prefix; try { return Object.freeze({ diff --git a/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py b/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py index d788492792..bf56f55bcd 100644 --- a/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py +++ b/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py @@ -7,7 +7,6 @@ import pytest from loopx.cli_commands import quota_scheduler_followup -from loopx.control_plane.quota import heartbeat_receipt from loopx.control_plane.quota.scheduler_ack import _run_native_followup from loopx.rollout_event_log import build_rollout_event, rollout_event_log_path @@ -24,54 +23,6 @@ def _append(runtime_root: Path, **fields: object) -> None: handle.write(json.dumps(event, ensure_ascii=False) + "\n") -def _status(runtime_root: Path, turn: str) -> str: - return heartbeat_receipt.heartbeat_receipt_status( - runtime_root, - goal_id=GOAL_ID, - agent_id=AGENT_ID, - turn_instance_id=turn, - ) - - -def test_receipt_status_distinguishes_missing_fresh_and_superseded( - tmp_path: Path, -) -> None: - runtime_root = tmp_path / "runtime" - assert _status(runtime_root, "turn-1") == "missing" - - _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") - assert _status(runtime_root, "turn-1") == "fresh" - assert _status(runtime_root, "turn-2") == "missing" - - _append(runtime_root, event_kind="quota_should_run", run_id="turn-2") - assert _status(runtime_root, "turn-1") == "stale" - assert _status(runtime_root, "turn-2") == "fresh" - - -def test_later_normal_run_audit_supersedes_the_receipt(tmp_path: Path) -> None: - runtime_root = tmp_path / "runtime" - _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") - _append( - runtime_root, - event_kind="quota_should_run", - status="normal_run", - details={"ok": True, "should_run": True}, - ) - assert _status(runtime_root, "turn-1") == "stale" - - -def test_unrelated_normal_run_audit_does_not_supersede(tmp_path: Path) -> None: - runtime_root = tmp_path / "runtime" - _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") - _append( - runtime_root, - event_kind="quota_should_run", - status="normal_run", - details={"ok": True, "should_run": False}, - ) - assert _status(runtime_root, "turn-1") == "fresh" - - def test_native_followup_request_carries_the_turn_fence( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -144,12 +95,28 @@ def _capture(before, **kwargs): assert captured["turn_instance_id"] == "turn-1" -def test_superseded_receipt_blocks_before_any_scheduler_mutation( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - runtime_root = tmp_path / "runtime" - _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") - _append(runtime_root, event_kind="quota_should_run", run_id="turn-2") +def _followup_args(**overrides: object) -> Namespace: + values: dict[str, object] = { + "quota_command": "scheduler-ack-current", + "goal_id": GOAL_ID, + "agent_id": AGENT_ID, + "codex_app_current_rrule": None, + "applied_rrule": "FREQ=MINUTELY;INTERVAL=3", + "host_match_observed": False, + "execute": True, + "surface": "trae_app", + "state_key": None, + "available_capabilities": [], + "reset_token": None, + "identity_signature": None, + "reason_summary": None, + "use_current_hint": False, + } + values.update(overrides) + return Namespace(**values) + + +def _capture_ack(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, object]]: calls: list[dict[str, object]] = [] def _ack(_status, **kwargs): @@ -159,14 +126,25 @@ def _ack(_status, **kwargs): monkeypatch.setattr( quota_scheduler_followup, "record_quota_scheduler_ack", _ack ) - args = Namespace( - quota_command="scheduler-ack-current", - goal_id=GOAL_ID, - agent_id=AGENT_ID, + monkeypatch.setattr( + quota_scheduler_followup, + "_build_scheduler_followup_decision", + lambda *_args, **_kwargs: {"effective_action": "execute_todo"}, ) + return calls + + +def test_superseded_receipt_is_decided_by_the_native_fence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + runtime_root = tmp_path / "runtime" + _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") + _append(runtime_root, event_kind="quota_should_run", run_id="turn-2") + calls = _capture_ack(monkeypatch) + payload = quota_scheduler_followup.build_scheduler_followup_payload( {}, - args, + _followup_args(), registry_path=tmp_path / "registry.json", runtime_root=runtime_root, turn_instance_id="turn-1", @@ -174,11 +152,40 @@ def _ack(_status, **kwargs): operator_inbox_urgency_projector=lambda **_kwargs: {}, ) - assert payload["ok"] is False - assert payload["status"] == "heartbeat_receipt_stale" - assert payload["error_code"] == "SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_STALE" - assert payload["scheduler_state_mutated"] is False - assert calls == [] + # The Python route keeps no second freshness opinion: it binds the + # originating Turn and lets the native fence answer stale/missing. + assert payload == {"ok": True} + assert calls and calls[0]["turn_instance_id"] == "turn-1" + + +def test_malformed_log_line_does_not_block_the_ack_route( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + runtime_root = tmp_path / "runtime" + _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") + path = rollout_event_log_path(runtime_root, GOAL_ID) + with path.open("ab") as handle: + handle.write( + b'{"schema_version":"loopx_rollout_event_v0",' + b'"event_kind":"quota_should_run","goal_id":"receipt-fence",' + b'"agent_id":"agent-a","run_id":"turn-bad","summary":"' + b"\xff" + b'"}\n' + ) + calls = _capture_ack(monkeypatch) + + payload = quota_scheduler_followup.build_scheduler_followup_payload( + {}, + _followup_args(), + registry_path=tmp_path / "registry.json", + runtime_root=runtime_root, + turn_instance_id="turn-1", + scheduler_context=None, + operator_inbox_urgency_projector=lambda **_kwargs: {}, + ) + + assert payload == {"ok": True} + assert calls and calls[0]["turn_instance_id"] == "turn-1" def test_fresh_receipt_still_reaches_the_ack_with_its_turn_fence( diff --git a/tests/control_plane_ts/receipt_log_snapshot.test.ts b/tests/control_plane_ts/receipt_log_snapshot.test.ts index a5119950d2..ccd792882c 100644 --- a/tests/control_plane_ts/receipt_log_snapshot.test.ts +++ b/tests/control_plane_ts/receipt_log_snapshot.test.ts @@ -95,6 +95,42 @@ test("schema, malformed line numbers and tolerant valid rows survive reuse", asy }); }); +test("invalid UTF-8 rows are malformed, never receipts", async () => { + await fixture(async (path) => { + const valid = Buffer.from('{"run_id":"turn-1"}\n'); + const bad = Buffer.concat([ + Buffer.from('{"run_id":"turn-2","summary":"'), + Buffer.from([0xff]), + Buffer.from('"}\n'), + ]); + await writeFile(path, Buffer.concat([valid, bad])); + const snapshot = (await read(path))!; + assert.deepEqual(snapshot.records.map((r) => r.run_id), ["turn-1"]); + assert.equal(snapshot.firstErrorLine, 2); + // Cached prefix reuse must not resurrect or forget the bad row. + const reused = (await read(path))!; + assert.equal(reused.firstErrorLine, 2); + assert.deepEqual(reused.records.map((r) => r.run_id), ["turn-1"]); + // An unterminated tail with invalid bytes is malformed, never a record. + await writeFile(path, Buffer.concat([ + valid, + Buffer.from('{"run_id":"turn-3","summary":"'), + Buffer.from([0xff]), + ])); + const tail = (await read(path))!; + assert.deepEqual(tail.records.map((r) => r.run_id), ["turn-1"]); + assert.equal(tail.firstErrorLine, 2); + // Terminating the damaged tail keeps it malformed but never hides a later row. + await appendFile(path, Buffer.from('\n{"run_id":"turn-4"}\n')); + const completed = (await read(path))!; + assert.deepEqual( + completed.records.map((r) => r.run_id), + ["turn-1", "turn-4"], + ); + assert.equal(completed.firstErrorLine, 2); + }); +}); + test("eviction and concurrent cold readers affect cost, never evidence", async () => { await fixture(async (path) => { const row = JSON.stringify({payload: "x".repeat(2048)}) + "\n"; diff --git a/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts b/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts index a636ce3df3..9ec659bf74 100644 --- a/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts +++ b/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts @@ -1,5 +1,6 @@ import assert from "node:assert/strict"; import { + appendFile, mkdir, mkdtemp, readFile, @@ -226,6 +227,77 @@ test("missing or superseded heartbeat receipt rejects before scheduler mutation" assert.equal(JSON.parse(await readFile(legacyPath, "utf8")).goal_id, scope.goal_id); }); +async function appendMalformedReceipt( + runtimeRoot: string, + turnId: string, +): Promise { + const path = receiptPath(runtimeRoot); + await mkdir(dirname(path), { recursive: true }); + // Structurally a newer receipt, but the summary carries one raw 0xff byte. + const head = Buffer.from( + `{"schema_version":"loopx_rollout_event_v0","event_kind":"quota_should_run",` + + `"goal_id":"${scope.goal_id}","agent_id":"${scope.agent_id}",` + + `"run_id":"${turnId}","summary":"`, + ); + await appendFile(path, Buffer.concat([head, Buffer.from([0xff]), Buffer.from('"}\n')])); +} + +test("a malformed newer log row cannot block a still-valid ACK or failure", async (t) => { + const runtimeRoot = await tempRuntime(t); + await appendReceipt(runtimeRoot, "turn-followup-1"); + const base = request(runtimeRoot); + const statePath = schedulerStatePath(runtimeRoot, { + goalId: scope.goal_id, + agentId: scope.agent_id, + surface: scope.surface, + stateKey: scope.state_key, + }); + + await appendMalformedReceipt(runtimeRoot, "turn-bad-bytes"); + const snapshot = await readGoalRolloutEventSnapshot(runtimeRoot, scope.goal_id); + assert.equal(snapshot!.firstStrictErrorLine, 2); + assert.deepEqual( + goalHeartbeatReceiptsFromSnapshot(snapshot, scope.goal_id)!.map((r) => r.run_id), + ["turn-followup-1"], + ); + // Strict settlement readback still reports the damaged row. + assert.throws(() => strictGoalRolloutEvents(snapshot), { + code: "malformed_settlement_state", + }); + + const ack = await evaluateSchedulerHeartbeatFollowup(base); + assert.equal(ack.ok, true); + assert.equal((ack.scheduler_commit as Record).status, "written"); + + const failure = await evaluateSchedulerHeartbeatFollowup({ + ...base, + host_facts: { + ...(base.host_facts as Record), + operation: "host_failure", + applied_rrule: "FREQ=MINUTELY;INTERVAL=3", + observed_host_rrule: "FREQ=MINUTELY;INTERVAL=3", + failure_kind: "timeout", + source: "quota_scheduler_host_update_failure", + }, + }); + assert.equal(failure.ok, true); + assert.equal(failure.mode, "scheduler-fail-current"); + + // A genuinely newer receipt still fences both writebacks. + await appendReceipt(runtimeRoot, "turn-followup-2"); + const staleAck = await evaluateSchedulerHeartbeatFollowup({ + ...base, + host_facts: { + ...(base.host_facts as Record), + generated_at: "2026-08-27T06:35:00Z", + }, + }); + assert.equal(staleAck.ok, false); + assert.equal(staleAck.status, "heartbeat_receipt_stale"); + assert.equal(staleAck.scheduler_state_mutated, false); + assert.equal(JSON.parse(await readFile(statePath, "utf8")).goal_id, scope.goal_id); +}); + test("receipt-bound ACK rechecks freshness after waiting for the scheduler lock", async (t) => { const runtimeRoot = await tempRuntime(t); await appendReceipt(runtimeRoot, "turn-followup-1"); From fff4d61423c06cf212e08cb01eb5e141b3b8fc43 Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Fri, 13 Nov 2026 02:36:43 +0000 Subject: [PATCH 3/4] fix(scheduler): use the tolerant settlement view for follow-up returns Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- loopx/cli_commands/quota_context.py | 7 ++ .../cli_commands/quota_scheduler_followup.py | 5 + loopx/control_plane/quota/live_decision.py | 2 + loopx/control_plane/quota/settlement.py | 6 ++ .../quota/settlement_readback.ts | 34 ++++++- loopx/control_plane/status/collection.py | 6 ++ .../work_items/attention_queue.py | 10 +- .../work_items/replan_history_codec.py | 6 ++ .../work_items/replan_history_settlement.ts | 19 +++- loopx/status.py | 19 +++- .../test_scheduler_heartbeat_receipt_fence.py | 95 +++++++++++++++++++ .../quota_settlement_readback.test.ts | 25 +++++ 12 files changed, 224 insertions(+), 10 deletions(-) diff --git a/loopx/cli_commands/quota_context.py b/loopx/cli_commands/quota_context.py index e7b84085f9..276ce094aa 100644 --- a/loopx/cli_commands/quota_context.py +++ b/loopx/cli_commands/quota_context.py @@ -362,6 +362,13 @@ def prepare_quota_command_context( available_capabilities=args.available_capabilities, agent_lane_id=args.agent_id, include_presentation_projections=False, + # A scheduler ACK/failure return writes cadence, not settlement + # facts. One damaged audit row must not fail the whole status read + # before the tolerant native fence can answer it. + tolerate_malformed_settlement_history=command in { + "scheduler-ack-current", + "scheduler-fail-current", + }, ) if bool(getattr(args, "write_projection_cache", False)): cache_metadata = write_status_projection_cache( diff --git a/loopx/cli_commands/quota_scheduler_followup.py b/loopx/cli_commands/quota_scheduler_followup.py index c0f20d6444..08c1b5c7d1 100644 --- a/loopx/cli_commands/quota_scheduler_followup.py +++ b/loopx/cli_commands/quota_scheduler_followup.py @@ -76,6 +76,11 @@ def _build_scheduler_followup_decision( repository_delivery_interaction_hook(repo_path=Path.cwd()), ), goal_ref=goal_ref, + # The compatibility ACK/failure route writes scheduler cadence, not + # settlement facts: one damaged audit row must not turn a valid return + # into a failure before the tolerant native fence runs. Exact + # settlement branches keep the strict readback. + tolerate_malformed_settlement_history=True, ) diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py index 4aa1728da3..c28d3287f1 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/loopx/control_plane/quota/live_decision.py @@ -665,6 +665,7 @@ def build_live_quota_should_run_decision( workspace_path: Path | None = None, goal_ref: Mapping[str, object] | None = None, inline_goal_source: bool = False, + tolerate_malformed_settlement_history: bool = False, ) -> dict[str, Any]: """Build one live CLI decision while keeping host observation injectable.""" resolved_context = resolve_scheduler_execution_context(scheduler_execution_context) @@ -722,6 +723,7 @@ def build_live_quota_should_run_decision( ), registry_path=registry_path, goal_ref=goal_ref, + tolerate_malformed_history=tolerate_malformed_settlement_history, ) receipt_bound_monitor_phase = ( settlement_readback.monitor_phase if settlement_readback else None diff --git a/loopx/control_plane/quota/settlement.py b/loopx/control_plane/quota/settlement.py index 4ea55d529d..e7fab3daa2 100644 --- a/loopx/control_plane/quota/settlement.py +++ b/loopx/control_plane/quota/settlement.py @@ -450,6 +450,7 @@ def read_heartbeat_settlement( borrow_source_admission: bool = False, heartbeat_reentry_guard: dict[str, Any] | None = None, include_record_bodies: bool = True, + tolerate_malformed_history: bool = False, ) -> QuotaSettlementReadback | None: """Read one complete heartbeat settlement through the TS domain owner.""" @@ -468,6 +469,11 @@ def read_heartbeat_settlement( if include_record_bodies is not True else {} ), + **( + {"tolerate_malformed_history": True} + if tolerate_malformed_history + else {} + ), **({"resolve_original_binding": True} if resolve_original_binding else {}), **({"refresh_retry": refresh_retry} if refresh_retry is not None else {}), **({"borrow_source_admission": True} if borrow_source_admission else {}), diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts index d77a381bc3..4002a55d4f 100644 --- a/loopx/control_plane/quota/settlement_readback.ts +++ b/loopx/control_plane/quota/settlement_readback.ts @@ -110,6 +110,7 @@ interface ReadbackRequest { resolve_original_binding: boolean; borrow_source_admission: boolean; include_record_bodies: boolean; + tolerate_malformed_history: boolean; refresh_retry: RefreshRetryRequest | null; heartbeat_reentry_guard: JsonObject | null; owner: QuotaAccountingOwner; @@ -224,6 +225,14 @@ function decodeRequest( if (request.include_record_bodies !== undefined && typeof request.include_record_bodies !== "boolean") { throw new EffectRuntimeRequestError("include_record_bodies must be a boolean"); } + if ( + request.tolerate_malformed_history !== undefined && + typeof request.tolerate_malformed_history !== "boolean" + ) { + throw new EffectRuntimeRequestError( + "tolerate_malformed_history must be a boolean", + ); + } if (request.resolve_original_binding === true && request.infer_turn_instance_id) { throw new EffectRuntimeRequestError("original binding requires an explicit Turn identity"); } @@ -283,6 +292,7 @@ function decodeRequest( resolve_original_binding: request.resolve_original_binding === true, borrow_source_admission: request.borrow_source_admission === true, include_record_bodies: request.include_record_bodies !== false, + tolerate_malformed_history: request.tolerate_malformed_history === true, refresh_retry: decodeRefreshRetry(request.refresh_retry), heartbeat_reentry_guard: request.heartbeat_reentry_guard == null ? null : requireJsonObject(request.heartbeat_reentry_guard, "heartbeat_reentry_guard"), @@ -290,9 +300,12 @@ function decodeRequest( }; } -async function readRunReceipts(path: string): Promise { +async function readRunReceipts( + path: string, + tolerateMalformedHistory: boolean = false, +): Promise { const snapshot = await readReceiptLogSnapshot(path); - if (snapshot?.firstErrorLine != null) { + if (!tolerateMalformedHistory && snapshot?.firstErrorLine != null) { throw new EffectRuntimeRequestError( `settlement readback line ${snapshot.firstErrorLine} is malformed`, "malformed_settlement_state", @@ -373,6 +386,7 @@ export async function readQuotaSettlementSnapshot( runtimeRoot: string, goalId: string, rolloutSnapshot?: GoalRolloutEventSnapshot | null, + tolerateMalformedHistory: boolean = false, ): Promise { settlementScope(runtimeRoot, goalId); if ( @@ -393,12 +407,17 @@ export async function readQuotaSettlementSnapshot( rolloutSnapshot === undefined ? readGoalRolloutEventSnapshot(runtimeRoot, goalId) : Promise.resolve(rolloutSnapshot), - readRunReceipts(join(goalRoot, "runs", "index.jsonl")), + readRunReceipts(join(goalRoot, "runs", "index.jsonl"), tolerateMalformedHistory), ]); + // Settlement-authority callers keep failing closed on the first damaged row. + // The scheduler follow-up route only needs the receipts it can actually read, + // so it opts into the same tolerant view the heartbeat reader already uses. return indexSettlementSnapshot( runtimeRoot, goalId, - strictGoalRolloutEvents(snapshot), + tolerateMalformedHistory + ? snapshot?.events ?? [] + : strictGoalRolloutEvents(snapshot), runs, ); } @@ -1446,7 +1465,12 @@ export async function readQuotaSettlement(value: unknown): Promise { request.owner, async () => { const result = readQuotaSettlementFromRequest(request, - await readQuotaSettlementSnapshot(request.runtime_root, request.goal_id)); + await readQuotaSettlementSnapshot( + request.runtime_root, + request.goal_id, + undefined, + request.tolerate_malformed_history, + )); const identity = jsonObject(jsonObject(jsonObject(result.identity)?.result)?.value); if (identity && request.refresh_retry && jsonObject(result.refresh_recovery)?.decision !== "reject" && diff --git a/loopx/control_plane/status/collection.py b/loopx/control_plane/status/collection.py index 7ddd538a4b..3ad3f1973b 100644 --- a/loopx/control_plane/status/collection.py +++ b/loopx/control_plane/status/collection.py @@ -71,6 +71,7 @@ class StatusCollectionContext: build_goal_channel_notification_projection: StatusCallback status_control_plane_context_limit: int max_todo_index_items: int + tolerate_malformed_settlement_history: bool = False def collect_status( @@ -89,6 +90,7 @@ def collect_status( agent_lane_id: str | None = None, allow_missing_registry: bool = False, include_presentation_projections: bool = True, + tolerate_malformed_settlement_history: bool = False, ) -> dict[str, Any]: """Collect the status read model. @@ -187,6 +189,10 @@ def canonical_runtime_for_goal(current_goal_id: str) -> Path: current_registry=registry, registry_path=registry_path, include_goal_channel_projection=include_presentation_projections, + tolerate_malformed_settlement_history=( + tolerate_malformed_settlement_history + or context.tolerate_malformed_settlement_history + ), ) # No later projection consumes the shared status snapshot. del todo_snapshot diff --git a/loopx/control_plane/work_items/attention_queue.py b/loopx/control_plane/work_items/attention_queue.py index fb9976b9b9..fb7a76a115 100644 --- a/loopx/control_plane/work_items/attention_queue.py +++ b/loopx/control_plane/work_items/attention_queue.py @@ -50,6 +50,7 @@ class AttentionQueueContext: monitor_signal_waiting_on: str registry_path: Path | None = None external_progress_review_context: Optional[Callable[..., dict[str, Any] | None]] = None + tolerate_malformed_settlement_history: bool = False def merge_global_registry_findings( @@ -335,7 +336,14 @@ def build_attention_queue( next_action_projection_warning=context.next_action_projection_warning, autonomous_replan_obligation_from_runs=context.autonomous_replan_obligation_from_runs, external_progress_review=external_progress_review, - effective_turn_cadence=effective_turn_cadence_context(goal, runtime_root, registry_path=context.registry_path), + effective_turn_cadence=effective_turn_cadence_context( + goal, + runtime_root, + registry_path=context.registry_path, + tolerate_malformed_history=( + context.tolerate_malformed_settlement_history + ), + ), ) item["quota"] = context.quota_status( goal, diff --git a/loopx/control_plane/work_items/replan_history_codec.py b/loopx/control_plane/work_items/replan_history_codec.py index c2f2d446d9..5d15b051cd 100644 --- a/loopx/control_plane/work_items/replan_history_codec.py +++ b/loopx/control_plane/work_items/replan_history_codec.py @@ -35,6 +35,7 @@ def effective_turn_cadence_context( registry_path: Path | None = None, goal_ref: dict[str, str] | None = None, source_admission: dict[str, Any] | None = None, + tolerate_malformed_history: bool = False, ) -> dict[str, Any] | None: """Encode a Goal whose live cadence was resolved by the caller.""" profile = goal.get("execution_profile") or {} @@ -55,6 +56,11 @@ def effective_turn_cadence_context( "settlement_source": { "runtime_root": str(runtime_root.resolve()), "goal_id": goal["id"], + **( + {"tolerate_malformed_history": True} + if tolerate_malformed_history + else {} + ), }, } diff --git a/loopx/control_plane/work_items/replan_history_settlement.ts b/loopx/control_plane/work_items/replan_history_settlement.ts index b09652f49e..1327730403 100644 --- a/loopx/control_plane/work_items/replan_history_settlement.ts +++ b/loopx/control_plane/work_items/replan_history_settlement.ts @@ -22,6 +22,11 @@ export async function projectSettledReplanHistory(value: unknown): Promise quotaOwnerOwnsProjection(owner, jsonObject(raw)?.goal_ref)); const rolloutSnapshot = await readGoalRolloutEventSnapshot(runtimeRoot, goalId); - const events = strictGoalRolloutEvents(rolloutSnapshot); + // Scheduler writeback returns only need the receipts they can read; a + // damaged row is never a receipt. Settlement-authority callers keep the + // strict view. + const events = tolerateMalformedHistory === true + ? rolloutSnapshot?.events ?? [] + : strictGoalRolloutEvents(rolloutSnapshot); const receiptKeys = new Set(events.filter(event => event.event_kind === "quota_should_run" && event.goal_id === goalId && typeof event.agent_id === "string" && typeof event.run_id === "string" && @@ -49,7 +59,12 @@ export async function projectSettledReplanHistory(value: unknown): Promise(); const seen = new Set(); for (const raw of settlementRuns) { diff --git a/loopx/status.py b/loopx/status.py index c64de931c4..07be2e4da3 100644 --- a/loopx/status.py +++ b/loopx/status.py @@ -1093,6 +1093,7 @@ def build_attention_queue( current_registry: dict[str, Any] | None = None, registry_path: Path | None = None, include_goal_channel_projection: bool = True, + tolerate_malformed_settlement_history: bool = False, ) -> dict[str, Any]: def request_active_state_todo_fields( goal: dict[str, Any], @@ -1155,6 +1156,9 @@ def request_active_state_todo_fields( monitor_signal_waiting_on=MONITOR_SIGNAL_WAITING_ON, external_progress_review_context=external_progress_review_context, registry_path=registry_path, + tolerate_malformed_settlement_history=( + tolerate_malformed_settlement_history + ), ), runtime_root=runtime_root, include_task_graph=include_task_graph, @@ -1325,7 +1329,10 @@ def build_status_runtime_summaries( ) -def build_status_collection_context() -> StatusCollectionContext: +def build_status_collection_context( + *, + tolerate_malformed_settlement_history: bool = False, +) -> StatusCollectionContext: return StatusCollectionContext( load_registry=load_registry, resolve_runtime_root=resolve_runtime_root, @@ -1341,6 +1348,9 @@ def build_status_collection_context() -> StatusCollectionContext: build_goal_channel_notification_projection=build_goal_channel_notification_projection, status_control_plane_context_limit=STATUS_CONTROL_PLANE_CONTEXT_LIMIT, max_todo_index_items=MAX_TODO_INDEX_ITEMS, + tolerate_malformed_settlement_history=( + tolerate_malformed_settlement_history + ), ) @@ -1359,6 +1369,7 @@ def collect_status( agent_lane_id: str | None = None, allow_missing_registry: bool = False, include_presentation_projections: bool = True, + tolerate_malformed_settlement_history: bool = False, ) -> dict[str, Any]: with effect_runtime_request_scope(): return _collect_status_read_model( @@ -1377,5 +1388,9 @@ def collect_status( activation_state_filter=activation_state_filter, agent_lane_id=agent_lane_id, include_presentation_projections=include_presentation_projections, - context=build_status_collection_context(), + context=build_status_collection_context( + tolerate_malformed_settlement_history=( + tolerate_malformed_settlement_history + ), + ), ) diff --git a/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py b/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py index bf56f55bcd..2dd1f75051 100644 --- a/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py +++ b/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py @@ -6,8 +6,13 @@ import pytest +from examples.control_plane.quota_plan_fixtures import ( + SCOPED_AGENT_ID, + write_cli_fixture, +) from loopx.cli_commands import quota_scheduler_followup from loopx.control_plane.quota.scheduler_ack import _run_native_followup +from loopx.control_plane.testing.canary_harness import run_json_cli_result from loopx.rollout_event_log import build_rollout_event, rollout_event_log_path @@ -158,6 +163,34 @@ def test_superseded_receipt_is_decided_by_the_native_fence( assert calls and calls[0]["turn_instance_id"] == "turn-1" +def test_scheduler_followup_builds_from_the_tolerant_settlement_view( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + captured: dict[str, object] = {} + + def _decision(*_args, **kwargs): + captured.update(kwargs) + return {"effective_action": "execute_todo"} + + monkeypatch.setattr( + quota_scheduler_followup, + "build_live_quota_should_run_decision", + _decision, + ) + quota_scheduler_followup._build_scheduler_followup_decision( + {}, + _followup_args(), + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", + heartbeat_receipt=None, + turn_instance_id=None, + codex_app_current_rrule=None, + scheduler_context=None, + operator_inbox_urgency_projector=lambda **_kwargs: {}, + ) + assert captured["tolerate_malformed_settlement_history"] is True + + def test_malformed_log_line_does_not_block_the_ack_route( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -235,3 +268,65 @@ def _ack(_status, **kwargs): assert payload == {"ok": True} assert calls and calls[0]["turn_instance_id"] == "turn-1" + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("command", ["scheduler-ack-current", "scheduler-fail-current"]) +def test_real_compatibility_cli_survives_a_damaged_audit_row( + tmp_path: Path, provider: str, command: str +) -> None: + registry, runtime_root, project = write_cli_fixture( + tmp_path / provider, scoped_agents=True, provider=provider, + ) + goal_id = "needs-operator" + turn_id = "turn-damaged-history" + code, guard = run_json_cli_result( + "quota", + "should-run", + "--goal-id", + goal_id, + "--agent-id", + SCOPED_AGENT_ID, + "--trae_app", + "--app-automation-current-rrule", + "FREQ=MINUTELY;INTERVAL=3", + "--turn-instance-id", + turn_id, + registry_path=registry, + runtime_root=runtime_root, + cwd=project, + ) + assert code == 0, guard + + log_path = rollout_event_log_path(runtime_root, goal_id) + with log_path.open("ab") as handle: + handle.write( + b'{"schema_version":"loopx_rollout_event_v0",' + b'"event_kind":"quota_should_run","goal_id":"needs-operator",' + b'"agent_id":"' + SCOPED_AGENT_ID.encode() + b'",' + b'"run_id":"later-turn","summary":"' + b"\xff" + b'"}\n' + ) + + # The ACK derives its applied RRULE from the current hint; the failure + # return records the hint target. Both are ordinary scheduler writebacks. + operation_args = () + code, result = run_json_cli_result( + "quota", + command, + "--goal-id", + goal_id, + "--agent-id", + SCOPED_AGENT_ID, + "--trae_app", + "--turn-instance-id", + turn_id, + "--execute", + *operation_args, + registry_path=registry, + runtime_root=runtime_root, + cwd=project, + ) + assert code == 0, result + assert result["ok"] is True, result + assert result["scheduler_state_mutated"] is True diff --git a/tests/control_plane_ts/quota_settlement_readback.test.ts b/tests/control_plane_ts/quota_settlement_readback.test.ts index bcea7eb4c7..4ac27c75ce 100644 --- a/tests/control_plane_ts/quota_settlement_readback.test.ts +++ b/tests/control_plane_ts/quota_settlement_readback.test.ts @@ -558,6 +558,31 @@ test("scoped supersede settles only its exact Turn and never terminal acceptance }); }); +test("only the tolerant readback route survives a damaged audit row", async () => { + const root = await fixture({writeback: true, spend: true}); + try { + const clean = await readQuotaSettlement(request(root)); + const path = join(root, "goals", goalId, "rollout-event-log.jsonl"); + await appendFile(path, Buffer.concat([ + Buffer.from('{"schema_version":"loopx_rollout_event_v0","event_kind":"quota_should_run","goal_id":"'), + Buffer.from(goalId), + Buffer.from('","agent_id":"'), + Buffer.from(agentId), + Buffer.from('","run_id":"later-turn","summary":"'), + Buffer.from([0xff]), + Buffer.from('"}\n'), + ])); + await assert.rejects(readQuotaSettlement(request(root)), { + code: "malformed_settlement_state", + }); + const tolerant = await readQuotaSettlement(request(root, { + tolerate_malformed_history: true, + })); + assert.deepEqual(tolerant, clean, + "a damaged row must not change the settlement facts a valid return can read"); + } finally { await rm(root, {recursive: true, force: true}); } +}); + test("settlement progress requires both effects and exact receipts", async t => { const cases = [ { guard: false, state: "identity_required", next: "validation" }, From 6bf625a4ece9bc315a98dcdc36af81d05a79e5e8 Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Fri, 13 Nov 2026 05:50:04 +0000 Subject: [PATCH 4/4] fix(scheduler): scope follow-up receipts and keep strict caches intact Keep the exact Goal instance on the scheduler follow-up request so a same-name receipt from another lifetime can neither authorize a return nor supersede the exact instance's receipt at preflight or inside the commit lock. Share the tolerant scheduler read with the manual compatibility ACK, and refuse to publish that partial view into the projection cache strict status readers consume. Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- loopx/cli_commands/quota_context.py | 54 ++++++--- .../cli_commands/quota_scheduler_followup.py | 2 + loopx/control_plane/quota/scheduler_ack.py | 7 ++ loopx/control_plane/rollout_receipt_log.ts | 19 ++- .../scheduler/heartbeat_commit.ts | 24 ++++ .../scheduler/heartbeat_followup.ts | 28 +++++ loopx/quota.py | 2 + .../test_scheduler_heartbeat_receipt_fence.py | 114 +++++++++++++++++- .../scheduler_heartbeat_followup.test.ts | 88 +++++++++++++- 9 files changed, 313 insertions(+), 25 deletions(-) diff --git a/loopx/cli_commands/quota_context.py b/loopx/cli_commands/quota_context.py index 276ce094aa..a26fe1a6b9 100644 --- a/loopx/cli_commands/quota_context.py +++ b/loopx/cli_commands/quota_context.py @@ -13,6 +13,7 @@ cached_goal_run_index_is_current, load_status_projection_cache, resolve_status_projection_cache_runtime_root, + status_projection_cache_metadata, write_status_projection_cache, ) from ..control_plane.scheduler.execution_context import ( @@ -365,24 +366,45 @@ def prepare_quota_command_context( # A scheduler ACK/failure return writes cadence, not settlement # facts. One damaged audit row must not fail the whole status read # before the tolerant native fence can answer it. - tolerate_malformed_settlement_history=command in { - "scheduler-ack-current", - "scheduler-fail-current", - }, + tolerate_malformed_settlement_history=( + command in QUOTA_SCHEDULER_FOLLOWUP_COMMANDS + ), ) if bool(getattr(args, "write_projection_cache", False)): - cache_metadata = write_status_projection_cache( - registry_path=registry_path, - runtime_root=runtime_root, - limit=status_limit, - include_task_graph=False, - goal_id=status_goal_id, - payload=status_payload, - max_age_seconds=projection_cache_ttl_seconds, - available_capabilities=args.available_capabilities, - agent_lane_id=args.agent_id, - include_presentation_projections=False, - ) + if command in QUOTA_SCHEDULER_FOLLOWUP_COMMANDS: + # The follow-up collector tolerates malformed history so a + # valid scheduler return can still be recorded. That partial + # view must never become the cache consumed by strict status + # readers under the same key. + cache_metadata = status_projection_cache_metadata( + registry_path=registry_path, + runtime_root=runtime_root, + limit=status_limit, + include_task_graph=False, + goal_id=status_goal_id, + max_age_seconds=projection_cache_ttl_seconds, + available_capabilities=args.available_capabilities, + agent_lane_id=args.agent_id, + include_presentation_projections=False, + ) + cache_metadata.update({ + "hit": False, + "written": False, + "skipped_reason": "tolerant_scheduler_followup_view", + }) + else: + cache_metadata = write_status_projection_cache( + registry_path=registry_path, + runtime_root=runtime_root, + limit=status_limit, + include_task_graph=False, + goal_id=status_goal_id, + payload=status_payload, + max_age_seconds=projection_cache_ttl_seconds, + available_capabilities=args.available_capabilities, + agent_lane_id=args.agent_id, + include_presentation_projections=False, + ) elif isinstance(status_payload.get("projection_cache"), dict): cache_metadata = dict(status_payload["projection_cache"]) diff --git a/loopx/cli_commands/quota_scheduler_followup.py b/loopx/cli_commands/quota_scheduler_followup.py index 08c1b5c7d1..f960c828bc 100644 --- a/loopx/cli_commands/quota_scheduler_followup.py +++ b/loopx/cli_commands/quota_scheduler_followup.py @@ -205,6 +205,7 @@ def build_scheduler_followup_payload( observed_host_rrule=observed_rrule, failure_kind=args.failure_kind, turn_instance_id=turn_instance_id, + goal_ref=goal_ref, ) return record_quota_scheduler_ack( status_payload, @@ -226,4 +227,5 @@ def build_scheduler_followup_payload( operator_inbox_urgency_projector=operator_inbox_urgency_projector, before_decision=before_decision, turn_instance_id=turn_instance_id, + goal_ref=goal_ref, ) diff --git a/loopx/control_plane/quota/scheduler_ack.py b/loopx/control_plane/quota/scheduler_ack.py index f0aa3fb937..34cb5f4b12 100644 --- a/loopx/control_plane/quota/scheduler_ack.py +++ b/loopx/control_plane/quota/scheduler_ack.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Mapping from typing import Any, cast import json @@ -185,12 +186,14 @@ def _run_native_followup( use_current_hint: bool, reason_summary: str | None, turn_instance_id: str | None = None, + goal_ref: Mapping[str, object] | None = None, ) -> dict[str, Any]: """Decision-free compatibility transport for in-process Python callers.""" request = { "schema_version": _FOLLOWUP_REQUEST_SCHEMA, "runtime_root": str(runtime_root), + "goal_ref": dict(goal_ref) if goal_ref is not None else None, "turn_instance_id": turn_instance_id, "require_heartbeat_receipt": bool(turn_instance_id), "before": before, @@ -300,6 +303,7 @@ def record_quota_scheduler_ack_for_decision( use_current_hint: bool = False, host_match_observed: bool = False, turn_instance_id: str | None = None, + goal_ref: Mapping[str, object] | 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) @@ -370,6 +374,7 @@ def record_quota_scheduler_ack_for_decision( use_current_hint=use_current_hint, reason_summary=reason_summary, turn_instance_id=turn_instance_id, + goal_ref=goal_ref, ) except ValueError as exc: return _failure_payload( @@ -400,6 +405,7 @@ def record_quota_scheduler_failure_for_decision( failure_kind: str = "host_tool_failure", generated_at: str | None = None, turn_instance_id: str | None = None, + goal_ref: Mapping[str, object] | 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) @@ -448,6 +454,7 @@ def record_quota_scheduler_failure_for_decision( use_current_hint=False, reason_summary=None, turn_instance_id=turn_instance_id, + goal_ref=goal_ref, ) except ValueError as exc: return _failure_payload( diff --git a/loopx/control_plane/rollout_receipt_log.ts b/loopx/control_plane/rollout_receipt_log.ts index a02c8464cc..494b1a317c 100644 --- a/loopx/control_plane/rollout_receipt_log.ts +++ b/loopx/control_plane/rollout_receipt_log.ts @@ -11,7 +11,7 @@ import { relative, resolve, sep } from "node:path"; import type { JsonObject } from "./effect_program.ts"; import { EffectRuntimeRequestError } from "./effect_runtime_errors.ts"; -import { requireNonEmptyString } from "./runtime_decode.ts"; +import { jsonObject, requireNonEmptyString } from "./runtime_decode.ts"; import { readReceiptLogSnapshot } from "./runtime/receipt_log_snapshot.ts"; export const ROLLOUT_EVENT_SCHEMA_VERSION = "loopx_rollout_event_v0"; @@ -154,12 +154,23 @@ export async function readGoalHeartbeatReceipts( ); } +/** A receipt without an exact Goal instance belongs to the legacy goal scope. */ +function receiptMatchesGoalRef(event: JsonObject, goalRef: JsonObject | null): boolean { + if (goalRef === null) return !Object.hasOwn(event, "goal_ref"); + const eventGoalRef = jsonObject(event.goal_ref); + return eventGoalRef !== null && + eventGoalRef.goal_id === goalRef.goal_id && + eventGoalRef.goal_instance_id === goalRef.goal_instance_id; +} + export async function heartbeatReceiptStatus(params: { runtimeRoot: string; goalId: string; agentId: string; turnInstanceId: string; + goalRef?: JsonObject | null; }): Promise { + const goalRef = params.goalRef ?? null; const snapshot = await readGoalRolloutEventSnapshot( params.runtimeRoot, params.goalId, @@ -171,7 +182,8 @@ export async function heartbeatReceiptStatus(params: { ); if (snapshot === null || receipts === null) return "missing"; const firstReceipt = receipts.find( - (event) => event.run_id === params.turnInstanceId, + (event) => event.run_id === params.turnInstanceId && + receiptMatchesGoalRef(event, goalRef), ); const firstMatch = firstReceipt === undefined ? -1 @@ -181,7 +193,8 @@ export async function heartbeatReceiptStatus(params: { if ( event.event_kind !== HEARTBEAT_RECEIPT_EVENT_KIND || event.goal_id !== params.goalId || - event.agent_id !== params.agentId + event.agent_id !== params.agentId || + !receiptMatchesGoalRef(event, goalRef) ) return false; if (typeof event.run_id === "string" && event.run_id.trim().length > 0) { return event.run_id !== params.turnInstanceId; diff --git a/loopx/control_plane/scheduler/heartbeat_commit.ts b/loopx/control_plane/scheduler/heartbeat_commit.ts index 2017c07143..f58b7c8650 100644 --- a/loopx/control_plane/scheduler/heartbeat_commit.ts +++ b/loopx/control_plane/scheduler/heartbeat_commit.ts @@ -3,6 +3,7 @@ import { readFile } from "node:fs/promises"; import type { JsonObject } from "../effect_program.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { parseExactGoalRef } from "../goals/goal_instance_identity.ts"; import { atomicWriteJson, withFileMutationLock, @@ -171,6 +172,7 @@ export interface SchedulerHeartbeatCommitResult extends JsonObject { export interface SchedulerHeartbeatCommitOptions { readonly heartbeatReceiptTurnInstanceId?: string | null; + readonly heartbeatReceiptGoalRef?: JsonObject | null; } function stableValue(value: unknown): unknown { @@ -831,6 +833,27 @@ export async function evaluateSchedulerHeartbeatCommit( options.heartbeatReceiptTurnInstanceId, "heartbeatReceiptTurnInstanceId", ).trim(); + let receiptGoalRef: JsonObject | null = null; + if (options.heartbeatReceiptGoalRef !== undefined && + options.heartbeatReceiptGoalRef !== null) { + const parsed = parseExactGoalRef(options.heartbeatReceiptGoalRef); + if (parsed.kind === "invalid") { + throw new EffectRuntimeRequestError( + `heartbeat receipt goal_ref is invalid: ${parsed.issue}`, + "heartbeat_receipt_goal_ref_invalid", + ); + } + receiptGoalRef = { + goal_id: parsed.value.goalId.value, + goal_instance_id: parsed.value.goalInstanceId.value, + }; + if (receiptGoalRef.goal_id !== request.scope.goalId) { + throw new EffectRuntimeRequestError( + "heartbeat receipt goal_ref does not match the scheduler scope", + "heartbeat_receipt_goal_ref_mismatch", + ); + } + } const path = schedulerStatePath(request.runtime_root, request.scope); const fingerprint = requestDigest(request); // Legacy state stays read-only until the commit holds every applicable @@ -853,6 +876,7 @@ export async function evaluateSchedulerHeartbeatCommit( goalId: request.scope.goalId, agentId: request.scope.agentId, turnInstanceId: receiptTurnInstanceId, + goalRef: receiptGoalRef, }); if (status !== "fresh") { return result( diff --git a/loopx/control_plane/scheduler/heartbeat_followup.ts b/loopx/control_plane/scheduler/heartbeat_followup.ts index d1047f0454..5b0cb72efa 100644 --- a/loopx/control_plane/scheduler/heartbeat_followup.ts +++ b/loopx/control_plane/scheduler/heartbeat_followup.ts @@ -1,5 +1,6 @@ import type { JsonObject } from "../effect_program.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { parseExactGoalRef } from "../goals/goal_instance_identity.ts"; import { goalPathSegment, heartbeatReceiptStatus, @@ -32,6 +33,7 @@ const FAILURE_CLASSIFICATION = "quota_scheduler_host_update_failure"; export interface SchedulerHeartbeatFollowupRequest extends JsonObject { schema_version: typeof SCHEDULER_HEARTBEAT_FOLLOWUP_REQUEST_SCHEMA; runtime_root: string; + goal_ref: JsonObject | null; turn_instance_id: string | null; require_heartbeat_receipt: boolean; before: JsonObject; @@ -81,6 +83,27 @@ function requestObject(value: unknown): SchedulerHeartbeatFollowupRequest { goalPathSegment(hostFacts.goal_id); requireNonEmptyString(hostFacts.agent_id, "agent_id"); followupOperation(hostFacts); + const rawGoalRef = input.goal_ref; + let goalRef: JsonObject | null = null; + if (rawGoalRef !== undefined && rawGoalRef !== null) { + const parsed = parseExactGoalRef(rawGoalRef); + if (parsed.kind === "invalid") { + throw new EffectRuntimeRequestError( + `scheduler follow-up goal_ref is invalid: ${parsed.issue}`, + "scheduler_followup_goal_ref_invalid", + ); + } + goalRef = { + goal_id: parsed.value.goalId.value, + goal_instance_id: parsed.value.goalInstanceId.value, + }; + if (goalRef.goal_id !== String(hostFacts.goal_id)) { + throw new EffectRuntimeRequestError( + "scheduler follow-up goal_ref does not match goal_id", + "scheduler_followup_goal_ref_mismatch", + ); + } + } const turnInstanceId = optionalText(input.turn_instance_id); const requireReceipt = input.require_heartbeat_receipt === undefined ? turnInstanceId !== null @@ -94,6 +117,7 @@ function requestObject(value: unknown): SchedulerHeartbeatFollowupRequest { return { schema_version: SCHEDULER_HEARTBEAT_FOLLOWUP_REQUEST_SCHEMA, runtime_root: runtimeRoot, + goal_ref: goalRef, turn_instance_id: turnInstanceId, require_heartbeat_receipt: requireReceipt, before: requireJsonObject(input.before ?? {}, "before"), @@ -341,6 +365,7 @@ export async function evaluateSchedulerHeartbeatFollowup( goalId: String(facts.goal_id), agentId: String(facts.agent_id), turnInstanceId: String(request.turn_instance_id), + goalRef: request.goal_ref, }); if (status !== "fresh") return receiptFailure(request, status); } @@ -351,6 +376,9 @@ export async function evaluateSchedulerHeartbeatFollowup( heartbeatReceiptTurnInstanceId: request.require_heartbeat_receipt ? request.turn_instance_id : null, + heartbeatReceiptGoalRef: request.require_heartbeat_receipt + ? request.goal_ref + : null, }); const receiptStatus = commitReceiptFailureStatus(commit); if (receiptStatus !== null) return receiptFailure(request, receiptStatus); diff --git a/loopx/quota.py b/loopx/quota.py index 12cc5bf22c..9cb458e208 100644 --- a/loopx/quota.py +++ b/loopx/quota.py @@ -1206,6 +1206,7 @@ def record_quota_scheduler_ack( operator_inbox_urgency_projector: Callable[..., dict[str, Any]] | None = None, before_decision: Mapping[str, Any] | None = None, turn_instance_id: str | None = None, + goal_ref: Mapping[str, Any] | 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) @@ -1241,6 +1242,7 @@ def record_quota_scheduler_ack( use_current_hint=use_current_hint, host_match_observed=host_match_observed, turn_instance_id=turn_instance_id, + goal_ref=goal_ref, ) diff --git a/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py b/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py index 2dd1f75051..d23c018270 100644 --- a/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py +++ b/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py @@ -52,6 +52,10 @@ def _run(command, **kwargs): monkeypatch.setattr( "loopx.control_plane.quota.scheduler_ack.subprocess.run", _run ) + goal_ref = { + "goal_id": "receipt-fence", + "goal_instance_id": "ginst_" + "a" * 32, + } _run_native_followup( runtime_root=tmp_path, before={}, @@ -59,9 +63,11 @@ def _run(command, **kwargs): use_current_hint=False, reason_summary=None, turn_instance_id="turn-1", + goal_ref=goal_ref, ) request = captured["request"] assert request["turn_instance_id"] == "turn-1" + assert request["goal_ref"] == goal_ref assert request["require_heartbeat_receipt"] is True _run_native_followup( @@ -73,6 +79,7 @@ def _run(command, **kwargs): ) request = captured["request"] assert request["turn_instance_id"] is None + assert request["goal_ref"] is None assert request["require_heartbeat_receipt"] is False @@ -90,14 +97,20 @@ def _capture(before, **kwargs): monkeypatch.setattr( quota, "record_quota_scheduler_ack_for_decision", _capture ) + goal_ref = { + "goal_id": "goals", + "goal_instance_id": "ginst_" + "b" * 32, + } quota.record_quota_scheduler_ack( {"runtime_root": str(tmp_path)}, goal_id="goals", agent_id="agent-a", before_decision={"decision": {}}, turn_instance_id="turn-1", + goal_ref=goal_ref, ) assert captured["turn_instance_id"] == "turn-1" + assert captured["goal_ref"] == goal_ref def _followup_args(**overrides: object) -> Namespace: @@ -270,9 +283,17 @@ def _ack(_status, **kwargs): assert calls and calls[0]["turn_instance_id"] == "turn-1" @pytest.mark.parametrize("provider", ["file", "sqlite"]) -@pytest.mark.parametrize("command", ["scheduler-ack-current", "scheduler-fail-current"]) +@pytest.mark.parametrize( + ("command", "include_turn"), + [ + ("scheduler-ack-current", True), + ("scheduler-fail-current", True), + ("scheduler-ack", True), + ("scheduler-ack", False), + ], +) def test_real_compatibility_cli_survives_a_damaged_audit_row( - tmp_path: Path, provider: str, command: str + tmp_path: Path, provider: str, command: str, include_turn: bool ) -> None: registry, runtime_root, project = write_cli_fixture( tmp_path / provider, scoped_agents=True, provider=provider, @@ -310,7 +331,8 @@ def test_real_compatibility_cli_survives_a_damaged_audit_row( # The ACK derives its applied RRULE from the current hint; the failure # return records the hint target. Both are ordinary scheduler writebacks. - operation_args = () + operation_args = ("--use-current-hint",) if command == "scheduler-ack" else () + turn_args = ("--turn-instance-id", turn_id) if include_turn else () code, result = run_json_cli_result( "quota", command, @@ -319,8 +341,7 @@ def test_real_compatibility_cli_survives_a_damaged_audit_row( "--agent-id", SCOPED_AGENT_ID, "--trae_app", - "--turn-instance-id", - turn_id, + *turn_args, "--execute", *operation_args, registry_path=registry, @@ -330,3 +351,86 @@ def test_real_compatibility_cli_survives_a_damaged_audit_row( assert code == 0, result assert result["ok"] is True, result assert result["scheduler_state_mutated"] is True + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_scheduler_followup_cache_preview_cannot_poison_strict_status( + tmp_path: Path, provider: str +) -> None: + registry, runtime_root, project = write_cli_fixture( + tmp_path / provider, scoped_agents=True, provider=provider, + ) + goal_id = "needs-operator" + turn_id = "turn-cache-preview" + code, guard = run_json_cli_result( + "quota", + "should-run", + "--goal-id", + goal_id, + "--agent-id", + SCOPED_AGENT_ID, + "--trae_app", + "--app-automation-current-rrule", + "FREQ=MINUTELY;INTERVAL=3", + "--turn-instance-id", + turn_id, + registry_path=registry, + runtime_root=runtime_root, + cwd=project, + ) + assert code == 0, guard + + log_path = rollout_event_log_path(runtime_root, goal_id) + with log_path.open("ab") as handle: + handle.write( + b'{"schema_version":"loopx_rollout_event_v0",' + b'"event_kind":"quota_should_run","goal_id":"needs-operator",' + b'"agent_id":"' + SCOPED_AGENT_ID.encode() + b'",' + b'"run_id":"later-turn","summary":"' + b"\xff" + b'"}\n' + ) + + before_code, before_status = run_json_cli_result( + "quota", + "status", + "--goal-id", + goal_id, + registry_path=registry, + runtime_root=runtime_root, + cwd=project, + ) + code, preview = run_json_cli_result( + "quota", + "scheduler-ack-current", + "--goal-id", + goal_id, + "--agent-id", + SCOPED_AGENT_ID, + "--trae_app", + "--turn-instance-id", + turn_id, + "--execute", + "--write-projection-cache", + registry_path=registry, + runtime_root=runtime_root, + cwd=project, + ) + assert code == 0, preview + cache = preview["status_projection_cache"] + assert cache["written"] is False + assert cache["skipped_reason"] == "tolerant_scheduler_followup_view" + assert not Path(cache["path"]).exists() + + after_code, after_status = run_json_cli_result( + "quota", + "status", + "--goal-id", + goal_id, + "--use-projection-cache", + registry_path=registry, + runtime_root=runtime_root, + cwd=project, + ) + assert after_code == before_code + assert after_status.get("status_projection_cache", {}).get("hit") is not True diff --git a/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts b/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts index 9ec659bf74..19208ad53d 100644 --- a/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts +++ b/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts @@ -69,7 +69,11 @@ function receiptPath(runtimeRoot: string): string { return join(runtimeRoot, "goals", scope.goal_id, "rollout-event-log.jsonl"); } -async function appendReceipt(runtimeRoot: string, turnId: string): Promise { +async function appendReceipt( + runtimeRoot: string, + turnId: string, + goalRef?: Record, +): Promise { const path = receiptPath(runtimeRoot); await mkdir(dirname(path), { recursive: true }); await writeFile(path, `${JSON.stringify({ @@ -78,6 +82,7 @@ async function appendReceipt(runtimeRoot: string, turnId: string): Promise goal_id: scope.goal_id, agent_id: scope.agent_id, run_id: turnId, + ...(goalRef === undefined ? {} : { goal_ref: goalRef }), })}\n`, { encoding: "utf8", flag: "a" }); } @@ -325,6 +330,87 @@ test("receipt-bound ACK rechecks freshness after waiting for the scheduler lock" await assert.rejects(readFile(path, "utf8"), { code: "ENOENT" }); }); +test("another Goal instance cannot authorize or supersede an exact-instance receipt", async (t) => { + const runtimeRoot = await tempRuntime(t); + const goalRefA = { + goal_id: scope.goal_id, + goal_instance_id: `ginst_${"a".repeat(32)}`, + }; + const goalRefB = { + goal_id: scope.goal_id, + goal_instance_id: `ginst_${"b".repeat(32)}`, + }; + const statePath = schedulerStatePath(runtimeRoot, { + goalId: scope.goal_id, + agentId: scope.agent_id, + surface: scope.surface, + stateKey: scope.state_key, + }); + await appendReceipt(runtimeRoot, "turn-followup-1", goalRefB); + + const foreign = await evaluateSchedulerHeartbeatFollowup(request(runtimeRoot, { + goal_ref: goalRefA, + })); + assert.equal(foreign.ok, false); + assert.equal(foreign.status, "heartbeat_receipt_missing"); + assert.equal(foreign.scheduler_state_mutated, false); + await assert.rejects(readFile(statePath, "utf8"), { code: "ENOENT" }); + + await appendReceipt(runtimeRoot, "turn-followup-1", goalRefA); + const fresh = await evaluateSchedulerHeartbeatFollowup(request(runtimeRoot, { + goal_ref: goalRefA, + })); + assert.equal(fresh.ok, true); + assert.equal((fresh.scheduler_commit as Record).status, "written"); + + // A newer receipt for another Goal lifetime is outside this transaction's + // freshness scope and cannot turn the exact instance's receipt stale. + await appendReceipt(runtimeRoot, "turn-followup-2", goalRefB); + const stillFresh = await evaluateSchedulerHeartbeatFollowup(request(runtimeRoot, { + goal_ref: goalRefA, + host_facts: { + ...((request(runtimeRoot).host_facts) as Record), + generated_at: "2026-08-27T06:35:00Z", + }, + })); + assert.equal(stillFresh.ok, true); +}); + +test("the scheduler lock rechecks freshness inside the exact Goal instance", async (t) => { + const runtimeRoot = await tempRuntime(t); + const goalRefA = { + goal_id: scope.goal_id, + goal_instance_id: `ginst_${"c".repeat(32)}`, + }; + const goalRefB = { + goal_id: scope.goal_id, + goal_instance_id: `ginst_${"d".repeat(32)}`, + }; + const exactRequest = request(runtimeRoot, { goal_ref: goalRefA }); + await appendReceipt(runtimeRoot, "turn-followup-1", goalRefA); + const path = schedulerStatePath(runtimeRoot, { + goalId: scope.goal_id, + agentId: scope.agent_id, + surface: scope.surface, + stateKey: scope.state_key, + }); + const held = await acquireFileMutationLock(path); + let released = false; + t.after(async () => { + if (!released) await releaseFileMutationLock(held.targetPath, held.token); + }); + + const inFlight = evaluateSchedulerHeartbeatFollowup(exactRequest); + await delay(250); + await appendReceipt(runtimeRoot, "turn-followup-2", goalRefB); + released = await releaseFileMutationLock(held.targetPath, held.token); + + const written = await inFlight; + assert.equal(written.ok, true); + assert.equal((written.scheduler_commit as Record).status, "written"); + assert.equal(JSON.parse(await readFile(path, "utf8")).goal_id, scope.goal_id); +}); + test("decision-free compatibility requests can omit a heartbeat receipt", async (t) => { const runtimeRoot = await tempRuntime(t); const result = await evaluateSchedulerHeartbeatFollowup(request(runtimeRoot, {