fix(scheduler): preserve cadence after settled Turn replay

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

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,

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

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,

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,

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*(?<count>[1-9][0-9]{0,4})\s*(?<unit>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");
 }

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"
 

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"])

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<string, unknown>;
+    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<string, unknown>;
+    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,