From 282562d0418168edd9fbba2a4067cd8f3bb91882 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 1 Oct 2026 16:51:21 +0800 Subject: [PATCH 1/2] fix(scheduler): preserve host cadence on settled Turn replay Signed-off-by: huangruiteng --- loopx/control_plane/scheduler/arbitration.py | 5 +- .../scheduler/execution_context.py | 4 +- .../control_plane/scheduler/scheduler_hint.py | 8 ++- .../scheduler/state_transition_rules.py | 13 +++++ .../scheduler/state_transition_rules.ts | 54 ++++++++++++++++++- .../test_replan_successor_guard_reentry.py | 10 +++- .../test_settled_replay_construction.py | 48 +++++++++++++++++ .../scheduler_state_transition_rules.test.ts | 27 ++++++++++ 8 files changed, 164 insertions(+), 5 deletions(-) diff --git a/loopx/control_plane/scheduler/arbitration.py b/loopx/control_plane/scheduler/arbitration.py index 74be6750a4..af73f8812e 100644 --- a/loopx/control_plane/scheduler/arbitration.py +++ b/loopx/control_plane/scheduler/arbitration.py @@ -16,6 +16,7 @@ class SchedulerDisposition(str, Enum): PEER_COORDINATION_WAIT = "peer_coordination_wait" AGENT_MONITOR_ONLY_WAIT = "agent_monitor_only_wait" ACTIVE_WORK = "active_work" + SETTLED_REPLAY = "settled_replay" AGENT_SCOPE_WAIT = "agent_scope_wait" CONSISTENCY_REPAIR = "consistency_repair" HUMAN_GATE = "human_gate" @@ -88,7 +89,7 @@ def _classify_disposition( if mode == "monitor_quiet_skip": return SchedulerDisposition.MONITOR_WAIT, "interaction_monitor_quiet_wait" if mode == "heartbeat_settled_skip": - return SchedulerDisposition.QUIET_WAIT, "interaction_heartbeat_settled_wait" + return SchedulerDisposition.SETTLED_REPLAY, "interaction_heartbeat_settled_replay" if mode == "successor_replan_required" and must_attempt: return SchedulerDisposition.ACTIVE_WORK, "interaction_successor_replan_required" if mode in agent_scope_modes: @@ -168,6 +169,8 @@ def build_scheduler_arbitration( user_required or must_attempt or delivery_allowed or not quiet_noop_allowed ): errors.append("interaction_contract.terminal_conflicts_with_open_action") + if mode == "heartbeat_settled_skip" and (user_required or must_attempt or delivery_allowed): + errors.append("interaction_contract.settled_conflicts_with_open_action") disposition, reason_code = _classify_disposition( mode=mode, user_required=user_required, diff --git a/loopx/control_plane/scheduler/execution_context.py b/loopx/control_plane/scheduler/execution_context.py index 8337c16fef..20153251bf 100644 --- a/loopx/control_plane/scheduler/execution_context.py +++ b/loopx/control_plane/scheduler/execution_context.py @@ -512,7 +512,9 @@ def build_goal_runtime_continuation( frontier_recheck_after_seconds: int | None = None, ) -> dict[str, Any]: action = str(scheduler_hint.get("action") or "") - if action == "run_now": + if action in {"run_now", "preserve_current_schedule"}: + # Replay does not complete or block a native Goal. Its next iteration + # must consult quota with a fresh identity, not repeat this closed Turn. disposition = GoalRuntimeContinuationDisposition.CONTINUE_NOW elif action == "stop_until_explicit_resume": disposition = GoalRuntimeContinuationDisposition.COMPLETE diff --git a/loopx/control_plane/scheduler/scheduler_hint.py b/loopx/control_plane/scheduler/scheduler_hint.py index 390c74ca92..ba9d82a7f7 100644 --- a/loopx/control_plane/scheduler/scheduler_hint.py +++ b/loopx/control_plane/scheduler/scheduler_hint.py @@ -47,7 +47,7 @@ rrule_for_minutes, scheduler_rrule_interval_minutes, ) -from .state_transition_rules import decide_scheduler_backoff_state +from .state_transition_rules import decide_scheduler_backoff_state, project_settled_replay_schedule from .time import parse_scheduler_timestamp SCHEDULER_HINT_SCHEMA_VERSION = "scheduler_hint_v0" @@ -1336,6 +1336,12 @@ def build_scheduler_hint( unchanged_spend_policy="no quota spend for terminal loop stop", ) + if arbitration.disposition == SchedulerDisposition.SETTLED_REPLAY: + return apply_scheduler_execution_context( + project_settled_replay_schedule(observed_host_rrule=codex_app_current_rrule), + execution_context, + ) + builder = _SchedulerHintBuilder( payload=payload, execution_context=execution_context, diff --git a/loopx/control_plane/scheduler/state_transition_rules.py b/loopx/control_plane/scheduler/state_transition_rules.py index 516e5230f7..baafcf8942 100644 --- a/loopx/control_plane/scheduler/state_transition_rules.py +++ b/loopx/control_plane/scheduler/state_transition_rules.py @@ -42,6 +42,19 @@ class MonitorScheduleProjection: cadence_seconds: int | None +def project_settled_replay_schedule(*, observed_host_rrule: Any = None) -> dict[str, Any]: + """Adapt the typed scheduler owner's hold-only replay packet.""" + result = _runtime_result({ + "schema_version": SCHEDULER_STATE_TRANSITION_REQUEST_SCHEMA, + "operation": "settled_replay", + "observed_host_rrule": observed_host_rrule, + }) + hint = result.get("hint") + if result.get("operation") != "settled_replay" or not isinstance(hint, dict): + raise RuntimeError("TypeScript settled scheduler result shape mismatch") + return hint + + def project_monitor_todo_schedule( *, generated_at: str, diff --git a/loopx/control_plane/scheduler/state_transition_rules.ts b/loopx/control_plane/scheduler/state_transition_rules.ts index 7ca2debb75..79d9f8ceaf 100644 --- a/loopx/control_plane/scheduler/state_transition_rules.ts +++ b/loopx/control_plane/scheduler/state_transition_rules.ts @@ -87,11 +87,62 @@ export interface SchedulerMonitorScheduleResult extends JsonObject { cadence_seconds: number | null; } +export interface SchedulerSettledReplayResult extends JsonObject { + schema_version: typeof SCHEDULER_STATE_TRANSITION_RESULT_SCHEMA; + operation: "settled_replay"; + hint: JsonObject & { + action: "preserve_current_schedule"; + cadence_class: "settled_turn"; + next_trigger: "fresh_turn_identity"; + app_automation: JsonObject & { + apply: "none"; + host_action: "none"; + ack_required: false; + observed_host_rrule: string | null; + }; + }; +} + export type SchedulerStateTransitionResult = | SchedulerCadenceTransitionResult | SchedulerHostTransitionResult | SchedulerBackoffTransitionResult - | SchedulerMonitorScheduleResult; + | SchedulerMonitorScheduleResult + | SchedulerSettledReplayResult; + +function projectSettledReplay(request: JsonObject): SchedulerSettledReplayResult { + const observedRrule = optionalRequestString(request.observed_host_rrule, "observed_host_rrule"); + // A closed receipt owns this Turn only. It cannot classify the live frontier + // as waiting, reset a scheduler identity, or authorize another host effect. + return { + schema_version: SCHEDULER_STATE_TRANSITION_RESULT_SCHEMA, + operation: "settled_replay", + hint: { + schema_version: "scheduler_hint_v0", + source: "quota.should-run", + action: "preserve_current_schedule", + cadence_class: "settled_turn", + reason_code: "interaction_heartbeat_settled_replay", + reason: "This Turn is settled, not the Goal frontier. Preserve the existing schedule and evaluate work under a fresh Turn identity.", + spend_policy: "no quota spend for settled Turn replay", + next_trigger: "fresh_turn_identity", + app_automation: { + apply: "none", + host_action: "none", + ack_required: false, + observed_host_rrule: observedRrule || null, + no_spend_for_cadence_change: true, + }, + unchanged_poll: { + limits: { local_scheduler: null, codex_cli_tui: null, claude_code_loop: null }, + after_limits: { local_scheduler: "continue", codex_cli_tui: "continue", claude_code_loop: "continue" }, + final_quota_replan_check_enabled: false, + final_quota_replan_check_action: null, + spend_policy: "no quota spend for settled Turn replay", + }, + }, + }; +} const MONITOR_CADENCE_PATTERN = /^\s*(?[1-9][0-9]{0,4})\s*(?s|sec|secs|second|seconds|m|min|mins|minute|minutes|h|hr|hrs|hour|hours|d|day|days)\s*$/i; @@ -482,5 +533,6 @@ export function evaluateSchedulerStateTransition( if (request.operation === "monitor_schedule") { return evaluateMonitorSchedule(request); } + if (request.operation === "settled_replay") return projectSettledReplay(request); throw new EffectRuntimeRequestError("Scheduler state transition operation is unsupported"); } diff --git a/tests/control_plane/test_replan_successor_guard_reentry.py b/tests/control_plane/test_replan_successor_guard_reentry.py index b6a499504e..1df1ed107d 100644 --- a/tests/control_plane/test_replan_successor_guard_reentry.py +++ b/tests/control_plane/test_replan_successor_guard_reentry.py @@ -68,7 +68,8 @@ def call(*args: str, expected_code: int = 0) -> dict: def _guard(call, turn: str = TURN): return call("quota", "should-run", "--codex-app", "--goal-id", GOAL, - "--agent-id", AGENT, "--turn-instance-id", turn) + "--agent-id", AGENT, "--turn-instance-id", turn, + "--codex-app-current-rrule", "FREQ=MINUTELY;INTERVAL=3") def _add(call, obligation_id: str): @@ -137,12 +138,19 @@ def test_successor_guard_returns_original_settlement_not_repeated_planning( settled = _guard(call) assert settled["effective_action"] == "heartbeat_settled_skip" assert settled.get("selected_todo") is None + hint = settled["scheduler_hint"] + assert hint["action"] == "preserve_current_schedule" + assert hint["app_automation"]["host_action"] == "none" + assert "recommended_rrule" not in hint["app_automation"] + assert "ack_hint" not in hint["app_automation"] rows = [json.loads(line) for line in index.read_text().splitlines()] assert sum(row.get("classification") == "quota_slot_spent" for row in rows) == 1 todo = call("todo", "list", "--goal-id", GOAL, "--todo-id", added["todo_id"])["todo"] assert todo["status"] == "open" fresh = _guard(call, "turn-independent-vision-review") assert fresh["effective_action"] != "heartbeat_settled_skip" + assert fresh["scheduler_hint"]["action"] == "run_now" + assert fresh["scheduler_hint"]["app_automation"]["recommended_interval_minutes"] == 3 assert fresh["goal_frontier_projection"]["acceptance_gaps"] assert fresh["goal_frontier_projection"]["vision_continuation_audit"]["decision"] == "acceptance_gap_open" diff --git a/tests/control_plane/test_settled_replay_construction.py b/tests/control_plane/test_settled_replay_construction.py index 14a9f4f33e..50d4a25500 100644 --- a/tests/control_plane/test_settled_replay_construction.py +++ b/tests/control_plane/test_settled_replay_construction.py @@ -8,6 +8,54 @@ from loopx.control_plane.quota.should_run import build_quota_should_run from loopx.control_plane.testing.quota_fixtures import quota_status_payload from loopx.presentation.renderers.quota_markdown import render_quota_should_run_markdown +from loopx.control_plane.scheduler.execution_context import scheduler_execution_context_for_runtime_profile +from loopx.control_plane.scheduler.arbitration import SchedulerDisposition, build_scheduler_arbitration + + +@pytest.mark.parametrize("profile", ["codex_app_heartbeat", "trae_app", "ark_managed_agent_goal"]) +@pytest.mark.parametrize("observed_rrule", [None, "FREQ=MINUTELY;INTERVAL=3", "FREQ=MINUTELY;INTERVAL=60"]) +def test_settled_replay_preserves_schedule_without_new_host_effects(profile, observed_rrule): + status = quota_status_payload( + goal_id="settled-fixture", status="active", recommended_action="Advance successor", + agent_todo_items=[{"todo_id": "todo_successor", "index": 1, "text": "Advance successor", + "role": "agent", "status": "open", "task_class": "advancement_task"}], + ) + payload = build_quota_should_run( + status, goal_id="settled-fixture", receipt_bound_replay_phase=ReceiptBoundReplayPhase.SETTLED, + scheduler_execution_context=scheduler_execution_context_for_runtime_profile(profile), + codex_app_current_rrule=observed_rrule, + ) + hint = payload["scheduler_hint"] + assert hint["action"] == "preserve_current_schedule" + assert hint["next_trigger"] == "fresh_turn_identity" + assert "reset_policy" not in hint + for key in ("app_automation", "codex_app"): + if key not in hint: + continue + host = hint[key] + assert host.get("host_action") == "none" + assert host.get("ack_required") is False + assert not {"recommended_rrule", "recommended_interval_minutes", "ack_hint", "failure_hint", + "fallback_hint", "stateful_backoff"}.intersection(host) + assert payload["interaction_contract"]["agent_channel"]["must_attempt"] is False + assert payload["interaction_contract"]["cli_channel"]["spend_after_validation"] is False + if profile == "ark_managed_agent_goal": + assert hint["goal_runtime_continuation"]["disposition"] == "continue_now" + + +@pytest.mark.parametrize("channel,field", [("agent_channel", "must_attempt"), + ("agent_channel", "delivery_allowed"), + ("user_channel", "action_required")]) +def test_settled_contract_cannot_carry_open_execution(channel, field): + payload = build_quota_should_run( + quota_status_payload(goal_id="settled-fixture", status="active", recommended_action="Continue"), + goal_id="settled-fixture", receipt_bound_replay_phase=ReceiptBoundReplayPhase.SETTLED, + ) + payload["interaction_contract"][channel][field] = True + result = build_scheduler_arbitration(payload) + assert result.disposition is SchedulerDisposition.CONSISTENCY_REPAIR + assert "interaction_contract.settled_conflicts_with_open_action" in result.errors + @pytest.mark.parametrize("quota_state", ["eligible", "operator_gate", "waiting_external", "exhausted"]) diff --git a/tests/control_plane_ts/scheduler_state_transition_rules.test.ts b/tests/control_plane_ts/scheduler_state_transition_rules.test.ts index 6cd103110d..73cff165ed 100644 --- a/tests/control_plane_ts/scheduler_state_transition_rules.test.ts +++ b/tests/control_plane_ts/scheduler_state_transition_rules.test.ts @@ -26,6 +26,33 @@ const fixture = JSON.parse( }>; }; +test("settled replay preserves observed cadence without deciding a future schedule", () => { + for (const observed of [null, "FREQ=MINUTELY;INTERVAL=3", "FREQ=MINUTELY;INTERVAL=60"]) { + const request = { schema_version: SCHEDULER_STATE_TRANSITION_REQUEST_SCHEMA, + operation: "settled_replay", observed_host_rrule: observed }; + const before = structuredClone(request); + const result = evaluateSchedulerStateTransition(request); + assert.equal(result.operation, "settled_replay"); + const hint = result.hint as Record; + assert.equal(hint.action, "preserve_current_schedule"); + assert.equal(hint.next_trigger, "fresh_turn_identity"); + assert.equal(hint.reset_policy, undefined); + const host = hint.app_automation as Record; + assert.equal(host.observed_host_rrule, observed); + assert.equal(host.host_action, "none"); + assert.equal(host.ack_required, false); + for (const field of ["recommended_rrule", "recommended_interval_minutes", "stateful_backoff", + "ack_hint", "failure_hint", "fallback_hint"]) { + assert.equal(host[field], undefined, field); + } + assert.deepEqual(request, before); + } + assert.throws(() => evaluateSchedulerStateTransition({ + schema_version: SCHEDULER_STATE_TRANSITION_REQUEST_SCHEMA, + operation: "settled_replay", observed_host_rrule: 3, + }), /observed_host_rrule must be a string/); +}); + test("pinned Python scheduler transition characterization remains exact", () => { assert.equal( fixture.schema_version, From 5d43394ca0fc17176c5b1e2be4135b23cc5db54e Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 1 Oct 2026 16:51:40 +0800 Subject: [PATCH 2/2] docs(quota): distinguish settled replay from live waiting Signed-off-by: huangruiteng --- docs/quota-allocation.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 409b9907a1..4553a32202 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -1193,6 +1193,14 @@ slows Codex App and stops CLI/Claude loops after repeated unchanged polls; agent-to-agent handoff cadence too quickly; `backoff_until_material_transition` handles monitor-only quiet polls; and `backoff_until_fresh_evidence` handles mapped or post-handoff no-op waits. +`preserve_current_schedule` handles `heartbeat_settled_skip`: settlement closes +one Turn, not the Goal or its remaining Todos. Replay keeps the installed host +cadence and emits no target interval/RRULE, scheduler reset, ACK, fallback or +host-update instruction, even when no host cadence has been observed. The next +trigger uses a fresh Turn identity to evaluate the live frontier; it never +re-executes or spends for the closed Turn. Native Goal runtimes continue to that +fresh guard instead of blocking the Goal. Explicit Goal stop or quota pause +still takes precedence. Actual fresh-Turn waits retain their existing backoff. For Codex App and local schedulers, `recommended_interval_minutes` is the next target interval. For Codex App heartbeats, `recommended_rrule` is emitted only when `app_automation.stateful_backoff.apply_needed=true`; if the desired RRULE is