fix(scheduler): keep automatic dated successor waits alive

docs/quota-allocation.md

@@ -1503,8 +1503,19 @@ That hint is not a delivery permission. It is the cross-runtime wait policy:
 `run_now` keeps the active cadence for required work; `backoff_waiting_for_user`
 slows Codex App and stops CLI/Claude loops after repeated unchanged polls;
 `backoff_until_reassigned` handles peer reassignment waits without dropping
-agent-to-agent handoff cadence too quickly;
-`backoff_until_material_transition` handles monitor-only quiet polls; and
+agent-to-agent handoff cadence too quickly.
+
+`backoff_until_state_change` also keeps a verified, automatically resumable
+`resume_at` successor alive through quiet rechecks. The scheduler bounds its
+backoff by the remaining scheduled wait, subject to host cadence floors, and
+does not apply the ordinary three-unchanged-poll stop to this clock-backed wait.
+This requires the current agent's exact pending successor and an unbound Turn;
+missing or invalid clock evidence, another agent's claim, and manual/dependency
+waits retain their existing supervision. A fresh due-time guard still owns
+admission; keeping the host alive grants no model wake, quota spend, new time
+budget or permission. Explicit stop, pause and user gates still take precedence.
+
+`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

loopx/control_plane/scheduler/arbitration.py

@@ -5,6 +5,8 @@
 from enum import Enum
 from typing import Any
 
+from ..goals.goal_vision_wait import exact_blocked_successor_wait_state
+
 SCHEDULER_ARBITRATION_SCHEMA_VERSION = "scheduler_arbitration_v0"
 
 
@@ -28,6 +30,7 @@ class SchedulerArbitration:
     reason_code: str
     mode: str
     errors: tuple[str, ...] = ()
+    scheduled_recheck_after_seconds: int | None = None
 
     @property
     def ok(self) -> bool:
@@ -54,12 +57,16 @@ def from_projection(cls, projection: Any) -> SchedulerArbitration:
                 or not isinstance(projection.get("errors"), list)
                 or any(not isinstance(error, str) for error in projection["errors"])):
             raise TypeError("TypeScript scheduler arbitration shape mismatch")
+        scheduled = projection.get("scheduled_recheck_after_seconds")
+        if scheduled is not None and (type(scheduled) is not int or scheduled <= 0):
+            raise TypeError("TypeScript scheduler scheduled recheck must be positive integer seconds")
         return cls(SchedulerDisposition(projection["disposition"]), projection["reason_code"],
-                   projection["mode"], tuple(projection["errors"]))
+                   projection["mode"], tuple(projection["errors"]), scheduled)
 
 
 def _scheduler_arbitration_facts(
     contract: Any, agent_scope_frontier_actions: Collection[str] = (),
+    *, payload: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Keep legacy Python scalar/Mapping codecs; transport only consumed facts."""
     contract = contract if isinstance(contract, Mapping) else {}
@@ -79,6 +86,14 @@ def _scheduler_arbitration_facts(
     wait = agent.get("wait") if isinstance(agent, Mapping) else None
     source = wait.get("source") if isinstance(wait, Mapping) else None
     channels["agent_channel"]["wait_source"] = source if isinstance(source, str) and len(source) <= 64 else None
-    return {"schema_version": contract.get("schema_version"),
+    facts = {"schema_version": contract.get("schema_version"),
             "mode": str(contract.get("mode") or "").strip(), **channels,
             "agent_scope_frontier_actions": [str(action) for action in agent_scope_frontier_actions]}
+    wait = exact_blocked_successor_wait_state(payload)
+    if wait:
+        identity = payload.get("agent_identity")
+        facts.update({"effective_action": payload.get("effective_action"),
+                      "agent_id": identity.get("agent_id") if isinstance(identity, Mapping) else None,
+                      "vision_wait_state": wait,
+                      "heartbeat_receipt": payload.get("heartbeat_receipt")})
+    return facts

loopx/control_plane/scheduler/scheduler_hint.py

@@ -1601,6 +1601,22 @@ def build_scheduler_hint(
         )
 
     if arbitration.disposition == SchedulerDisposition.AGENT_SCOPE_WAIT:
+        if arbitration.scheduled_recheck_after_seconds is not None:
+            horizon = max(1, math.ceil(arbitration.scheduled_recheck_after_seconds / 60))
+            progression = sorted({min(bucket, horizon) for bucket in (10, 20, 30, 60)})
+            return builder.build(
+                action="backoff_until_state_change",
+                cadence_class="agent_scope_wait",
+                reason=(
+                    "an automatic clock-backed successor is pending; keep quiet rechecks "
+                    "until its scheduled instant, then follow fresh admission gates"
+                ),
+                codex_interval=progression[0],
+                codex_max=progression[-1],
+                cli_limit=None,
+                claude_limit=None,
+                cadence_progression_override=progression,
+            )
         return builder.build(
             action="backoff_until_reassigned",
             cadence_class="agent_scope_wait",

loopx/control_plane/work_items/interaction_contract.py

@@ -1102,7 +1102,7 @@ def _interaction_reads_and_arbitration(
     selected = payload.get("selected_todo") or {}
     orchestration = payload.get("task_orchestration_contract") or {}
     projection_params = {
-        "arbitration_facts": _scheduler_arbitration_facts(contract, agent_scope_frontier_actions),
+        "arbitration_facts": _scheduler_arbitration_facts(contract, agent_scope_frontier_actions, payload=payload),
         "required_reads": [dict(item) for item in reads if isinstance(item, dict) and item.get("command")]
             if isinstance(reads, list) else [],
         "goal_id": payload.get("goal_id"),

loopx/control_plane/work_items/interaction_contract.ts

@@ -17,34 +17,44 @@ import type { JsonObject } from "../effect_program.ts";
 import { AgentScopeFrontierAction } from "../agents/agent_scope_frontier.generated.ts";
 import { resumeConditionHasKnownPendingTarget } from "../todos/resume_condition.ts";
 
-/** A clock-backed wait needs no observation before its scheduled instant.
- * This read model never relaxes an existing Turn binding or settlement fence.
- * Dependency waits retain their bounded observation/replan policy. */
-export function projectBlockedSuccessorWaitObservation(request: JsonObject): JsonObject {
+function exactBlockedSuccessorWait(request: JsonObject): JsonObject | null {
   const wait = jsonObject(request.vision_wait_state);
   const exact = request.effective_action === AgentScopeFrontierAction.AGENT_SCOPE_WAIT &&
     wait?.schema_version === "goal_vision_wait_state_v0" && wait.state === "waiting" &&
     wait.reason_code === "exact_blocked_successor" && wait.automatic_resume === true &&
     typeof wait.selected_todo_id === "string" && !!wait.selected_todo_id &&
     typeof wait.resume_when === "string" && !!wait.resume_when;
-  if (!exact || !wait) return {required: false};
+  return exact && wait ? wait : null;
+}
+
+/** Share the existing clock proof; a Turn binding always retains supervision. */
+function pendingDatedSuccessor(request: JsonObject, wait: JsonObject): JsonObject | null {
   const receipt = jsonObject(request.heartbeat_receipt);
   const unbound = request.heartbeat_receipt == null || (!!receipt &&
     ["committed", "replayed"].includes(String(receipt.status)) &&
     !("settlement_identity" in receipt) && !("pending_action_selection" in receipt));
   const condition = jsonObject(wait.resume_condition);
-  if (unbound && wait.agent_id === request.agent_id && wait.waiting_todo_count === 1 &&
+  if (unbound && typeof request.agent_id === "string" && !!request.agent_id &&
+      wait.agent_id === request.agent_id && wait.waiting_todo_count === 1 &&
       ["open", "deferred"].includes(String(wait.selected_todo_status)) &&
       condition?.kind === "resume_at") {
     try {
       if (resumeConditionHasKnownPendingTarget(condition, {
         todo_id: wait.selected_todo_id, resume_when: wait.resume_when,
-      })) return {required: false};
+      })) return condition;
     } catch {
       // An incomplete or malformed clock proof cannot relax the contract.
     }
   }
-  return {required: true};
+  return null;
+}
+
+/** A clock-backed wait needs no observation before its scheduled instant.
+ * This read model never relaxes an existing Turn binding or settlement fence.
+ * Dependency waits retain their bounded observation/replan policy. */
+export function projectBlockedSuccessorWaitObservation(request: JsonObject): JsonObject {
+  const wait = exactBlockedSuccessorWait(request);
+  return {required: !!wait && pendingDatedSuccessor(request, wait) === null};
 }
 
 export const INTERACTION_CONTRACT_SCHEMA_VERSION =
@@ -171,9 +181,21 @@ export function projectSchedulerArbitration(value: unknown): JsonObject {
   errors.push(...interactionConsistencyIssues(facts).map(issue => issue.code));
   const [disposition, reason] = schedulerDisposition(facts, modes,
     user.non_gating_notice === true && agent.wait_source === "turn_settlement");
+  const wait = exactBlockedSuccessorWait(request);
+  const condition = !errors.length && disposition === "agent_scope_wait" &&
+    !facts.must_attempt && !facts.user_required && facts.quiet_noop_allowed &&
+    !facts.spend_allowed_now && !facts.spend_after_validation && wait &&
+    (wait.selected_todo_claimed_by == null || wait.selected_todo_claimed_by === "" ||
+      wait.selected_todo_claimed_by === request.agent_id)
+    ? pendingDatedSuccessor(request, wait) : null;
+  // The resume-condition owner has verified both timestamps and a future target.
+  // This private duration bounds host rechecks; it grants no execution/spend.
+  const scheduled = condition ? {scheduled_recheck_after_seconds: Math.ceil(
+    (Date.parse(String(condition.scheduled_for)) - Date.parse(String(condition.evaluated_at))) / 1000,
+  )} : {};
   return {mode, errors: [...new Set(errors)],
     disposition: errors.length ? "consistency_repair" : disposition,
-    reason_code: errors.length ? "scheduler_interaction_contract_inconsistent" : reason};
+    reason_code: errors.length ? "scheduler_interaction_contract_inconsistent" : reason, ...scheduled};
 }
 
 function decodeUserChannel(value: unknown): UserInteractionChannel {

tests/control_plane/scheduler_arbitration_fixture.py

@@ -19,7 +19,7 @@ def project_native_arbitration(
         "delivery_allowed": False, "selection_required": False,
         "has_replan": False, "settlement_only": False,
         "arbitration_facts": _scheduler_arbitration_facts(
-            payload.get("interaction_contract"), agent_scope_frontier_actions,
+            payload.get("interaction_contract"), agent_scope_frontier_actions, payload=payload,
         ),
     })
     return SchedulerArbitration.from_projection(result["arbitration"])

tests/control_plane/test_dated_vision_wait_cli.py

@@ -151,9 +151,74 @@ def test_due_vision_wait_restores_explicit_resume(tmp_path, isolated_effect):
     assert any("todo update" in action and "--clear-resume-when" in action
                for action in guard["interaction_contract"]["cli_channel"]["next_cli_actions"])
     assert "vision_wait_state" not in guard
+    assert guard["scheduler_hint"]["action"] == "run_now"
+    assert all(limit is None for limit in guard["scheduler_hint"]["unchanged_poll"]["limits"].values())
     assert state.read_bytes() == original
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_real_cli_worker_keeps_automatic_wait_alive_and_honors_explicit_stop(
+    tmp_path, isolated_effect, monkeypatch, provider,
+):
+    from argparse import Namespace
+    from canonical_authority_fixture import initialize_canonical_authority
+    from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+
+    monkeypatch.syspath_prepend(str(REPO / "scripts"))
+    import external_scheduler_worker as worker
+
+    registry, runtime, state = fixture(tmp_path)
+    if provider == "sqlite":
+        monkeypatch.setenv("NODE_OPTIONS", os.environ.get("NODE_OPTIONS", "") + " --experimental-sqlite")
+    rc, listed = cli(registry, runtime, "todo", "list", "--goal-id", GOAL)
+    assert rc == 0, listed
+    initialize_canonical_authority(runtime, GOAL, build_todo_runtime_shadow_projection(
+        goal_id=GOAL, handoff_mode="soft_claim", todos=listed["todos"],
+    ), state_path=state, provider=provider)
+    original_state = state.read_bytes()
+    packets, sleeps = [], []
+
+    def real_probe(command, **kwargs):
+        # Adapt process transport only: every reply is produced by the real CLI
+        # against the isolated authority, with a fresh Turn identity.
+        rc, packet = cli(registry, runtime, *command[command.index("quota"):],
+                         "--turn-instance-id", f"scheduled-worker-{len(packets)}")
+        assert rc == 0, packet
+        packets.append(packet)
+        return packet
+
+    def advance_host(seconds):
+        sleeps.append(seconds)
+        assert len(sleeps) <= 4, "explicit stop must end the actual worker"
+        if len(sleeps) == 4:
+            stopped = json.loads(registry.read_text())
+            stopped["goals"][0]["activation_state"] = "stopped"
+            registry.write_text(json.dumps(stopped))
+
+    monkeypatch.setattr(worker, "run_quota_should_run", real_probe)
+    monkeypatch.setattr(worker, "_run_wake", lambda *a, **kw: pytest.fail("a pending wait cannot wake the model"))
+    args = Namespace(cli_bin="loopx", registry=str(registry), runtime_root=str(runtime),
+        runtime_profile="generic_cli", goal_id=GOAL, agent_id=AGENT,
+        state_file=str(tmp_path / "scheduler-state.json"), once=False, wake_cmd="unused",
+        error_backoff_seconds=5, quota_timeout_seconds=90, wake_timeout_seconds=None)
+    assert worker.run_worker(args, sleep=advance_host) == 0
+    assert sleeps == [600, 1200, 1800, 3600]
+    assert len(packets) == 5
+    for packet in packets[:-1]:
+        assert packet["should_run"] is False
+        assert packet["vision_wait_state"]["automatic_resume"] is True
+        hint = packet["scheduler_hint"]
+        assert hint["action"] == "backoff_until_state_change"
+        assert all(limit is None for limit in hint["unchanged_poll"]["limits"].values())
+        assert set(hint["unchanged_poll"]["after_limits"].values()) == {"continue"}
+        assert hint["unchanged_poll"]["final_quota_replan_check_enabled"] is False
+    assert packets[-1]["scheduler_hint"]["action"] == "stop_until_explicit_resume"
+    assert packets[-1]["scheduler_hint"]["cadence_class"] == "goal_stopped"
+    assert state.read_bytes() == original_state
+    runs = [json.loads(line) for line in (runtime / f"goals/{GOAL}/runs/index.jsonl").read_text().splitlines()]
+    assert not any(run.get("classification") in ("quota_monitor_poll", "quota_slot_spent") for run in runs)
+
+
 def test_independent_work_is_not_silenced_by_scheduled_wait(tmp_path, isolated_effect):
     registry, runtime, state = fixture(tmp_path)
     with state.open("a", encoding="utf-8") as output:

tests/control_plane/test_scheduler_arbitration_transport.py

@@ -25,6 +25,14 @@ def contract() -> dict:
         "cli_channel": {"spend_allowed_now": False, "spend_after_validation": False}}
 
 
+@pytest.mark.parametrize("invalid", [True, False, 0, -1, 1.5, "1800", []])
+def test_scheduled_recheck_codec_rejects_nonpositive_or_coerced_seconds(invalid):
+    with pytest.raises(TypeError, match="scheduled recheck"):
+        SchedulerArbitration.from_projection({"mode": "agent_scope_wait", "errors": [],
+            "disposition": "agent_scope_wait", "reason_code": "interaction_agent_scope_wait",
+            "scheduled_recheck_after_seconds": invalid})
+
+
 def encoded_size(value: dict) -> int:
     return len(json.dumps(value, separators=(",", ":")).encode())
 

tests/control_plane/test_scheduler_interaction_arbitration.py

@@ -564,3 +564,37 @@ def test_python_source_codec_does_not_coerce_non_boolean_authority(flag) -> None
     assert arbitration.disposition is SchedulerDisposition.CONSISTENCY_REPAIR
     assert arbitration.errors == ("interaction_contract.user_channel.action_required_must_be_boolean",)
     assert payload == original
+
+
+@pytest.mark.parametrize("minutes,progression", [
+    (1, [1]), (11, [10, 11]), (25, [10, 20, 25]), (61, [10, 20, 30, 60]),
+])
+def test_automatic_dated_wait_bounds_rechecks_without_finite_stop(minutes, progression):
+    payload = _payload(mode="agent_scope_wait", should_run=False, user_required=False,
+        must_attempt=False, delivery_allowed=False, quiet_noop_allowed=True)
+    evaluated = datetime(2099, 1, 1, tzinfo=timezone.utc)
+    due = (evaluated + timedelta(minutes=minutes)).isoformat().replace("+00:00", "Z")
+    resume = f"resume_at:{due}"
+    payload["vision_wait_state"] = {
+        "schema_version": "goal_vision_wait_state_v0", "state": "waiting",
+        "reason_code": "exact_blocked_successor", "automatic_resume": True,
+        "agent_id": payload["agent_identity"]["agent_id"], "waiting_todo_count": 1,
+        "selected_todo_id": "todo_scheduled", "selected_todo_status": "deferred",
+        "resume_when": resume, "resume_condition": {
+            "schema_version": "todo_resume_condition_v0", "kind": "resume_at",
+            "resume_when": resume, "target": due, "satisfied": False,
+            "scheduled_for": due, "evaluated_at": "2099-01-01T00:00:00Z",
+            "clock_provider": "runtime_clock", "material_change": False,
+            "material_change_generation": 0, "resume_receipt": None,
+        },
+    }
+    original = deepcopy(payload)
+    hint = build_scheduler_hint(payload, agent_scope_frontier_actions=AGENT_SCOPE_ACTIONS,
+        scheduler_execution_context=scheduler_execution_context_for_runtime_profile("generic_cli"),
+        include_detail=True)
+    local = hint["cold_path_detail"]["local_scheduler"]
+    assert hint["action"] == "backoff_until_state_change"
+    assert local["example_progression_minutes"] == progression
+    assert all(limit is None for limit in hint["unchanged_poll"]["limits"].values())
+    assert set(hint["unchanged_poll"]["after_limits"].values()) == {"continue"}
+    assert payload == original