From 27dfa55d6b2245dc324ab243fa549a0f0e7ddc73 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 6 Nov 2026 00:32:40 +0800
Subject: [PATCH] Honor scoped Todo resume conditions across external and
 managed schedulers

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../protocols/typed-date-resume-trigger-v0.md | 11 ++-
 loopx/control_plane/agents/agent_scope.py     | 12 +--
 .../goals/vision_wait_coverage.ts             | 77 +----------------
 loopx/control_plane/scheduler/arbitration.py  | 23 +++--
 loopx/control_plane/scheduler/hint.ts         |  9 +-
 loopx/control_plane/todos/automatic_resume.ts | 79 +++++++++++++++++
 loopx/control_plane/todos/resume_planning.ts  |  8 +-
 .../work_items/interaction_contract.ts        | 31 ++++---
 scripts/external_scheduler_worker.py          | 17 ++--
 .../test_dated_vision_wait_cli.py             | 86 ++++++++++++++++++-
 .../control_plane_ts/resume_planning.test.ts  | 21 +++++
 .../scheduler_arbitration.test.ts             | 15 ++++
 12 files changed, 275 insertions(+), 114 deletions(-)
 create mode 100644 loopx/control_plane/todos/automatic_resume.ts

diff --git a/docs/reference/protocols/typed-date-resume-trigger-v0.md b/docs/reference/protocols/typed-date-resume-trigger-v0.md
index 88ba123bb5..ef33184bad 100644
--- a/docs/reference/protocols/typed-date-resume-trigger-v0.md
+++ b/docs/reference/protocols/typed-date-resume-trigger-v0.md
@@ -64,16 +64,19 @@ same receipt and never increment the generation beyond `1`.
 - Once due, quota returns the existing `successor_replan_required` lifecycle
   action. A receipt is proof of the condition transition, not execution
   authority and not an implicit reopen.
-- Managed Turn interprets the same quota packet. Heartbeat runs do not own a
-  separate timer or readiness cache.
+- Managed Turn and the external CLI worker take their next check from the
+  scheduler's projection of the scoped Todo. A verified future wait keeps the
+  worker alive across unchanged checks, including when the Goal has no Vision
+  record. The worker sleeps for the shorter of its observation interval and
+  the remaining time, then requests fresh admission. This check spends no quota.
+  A stop, user gate or reassignment still controls the next decision.
 - The frontend extends the existing Todo defer/details surface with pending,
   ready, and receipt information. It does not add another configuration owner.
 - Lark requires no condition-specific state owner: the Goal Channel consumes
   the same status/interaction projection and may render the receipt through its
   normal presentation sink.
 
-This protocol does not alter periodic-report policy, scheduler cadence, or the
-user's notification preference.
+The periodic-report and notification policies govern user-facing notices.
 
 ## Acceptance
 
diff --git a/loopx/control_plane/agents/agent_scope.py b/loopx/control_plane/agents/agent_scope.py
index 4df716f5da..4293fb2c1a 100644
--- a/loopx/control_plane/agents/agent_scope.py
+++ b/loopx/control_plane/agents/agent_scope.py
@@ -1096,10 +1096,8 @@ def _deferred_resume_frontier(
 def _blocked_successor_wait_frontier(
     context: _AgentScopeNoCandidateContext,
 ) -> dict[str, Any] | None:
-    candidates = project_todo_resume_planning(
-        context.summary,
-        agent_id=context.agent_id,
-    )["blocked_successor_items"]
+    planning = project_todo_resume_planning(context.summary, agent_id=context.agent_id)
+    candidates = planning["blocked_successor_items"]
     if not candidates:
         return None
     first = candidates[0]
@@ -1124,7 +1122,11 @@ def _blocked_successor_wait_frontier(
             **context.candidate_counts,
             "blocked_successor_wait_count": len(candidates),
         },
-        extra_fields={"blocked_successor_wait_candidates": candidates[:3]},
+        extra_fields={
+            "blocked_successor_wait_candidates": candidates[:3],
+            **({"automatic_resume_todo": planning["automatic_resume_todo"]}
+               if "automatic_resume_todo" in planning else {}),
+        },
     )
 
 
diff --git a/loopx/control_plane/goals/vision_wait_coverage.ts b/loopx/control_plane/goals/vision_wait_coverage.ts
index c4c2190fed..128e9cb9a8 100644
--- a/loopx/control_plane/goals/vision_wait_coverage.ts
+++ b/loopx/control_plane/goals/vision_wait_coverage.ts
@@ -2,42 +2,13 @@ import type { JsonObject } from "../effect_program.ts";
 import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
 import { jsonObject, optionalNonEmptyString, requireInteger, requireJsonObject, requireStringArray } from "../runtime_decode.ts";
 import { projectTodoResumePlanning } from "../todos/resume_planning.ts";
-import { resumeConditionHasKnownPendingTarget } from "../todos/resume_condition.ts";
-import { parseTodoTimestampMicros } from "../runtime_timestamp.ts";
-import { monitorIsActive } from "../todos/summary_lanes.ts";
+import { automaticResumeFields, normalizedTodoId, selectPendingResume } from "../todos/automatic_resume.ts";
 
 const WAIT_STATE_SCHEMA = "goal_vision_wait_state_v0";
 const VISION_GAP = "vision_acceptance_gap";
 const CAUSAL_ID_KEYS = ["todo_id", "current_todo_id", "blocked_todo_id", "completed_todo_id", "successor_todo_id",
   "vision_todo_ids", "lineage_todo_ids", "successor_todo_ids", "completed_todo_ids"] as const;
 
-function hasScheduledMonitorSource(condition: JsonObject): boolean {
-  const time = (value: unknown): number | null => {
-    const micros = typeof value === "string" ? parseTodoTimestampMicros(value) : null;
-    return micros === null ? null : Number(micros) / 1_000_000;
-  };
-  const observed = time(condition.evaluated_at), due = time(condition.target_next_due_at);
-  const expiry = time(condition.target_expires_at);
-  if (condition.clock_provider !== "runtime_clock" || observed === null || due === null ||
-      !(condition.target_archive_state === null || condition.target_archive_state === "active") ||
-      condition.target_expires_at !== null && expiry === null) return false;
-  return monitorIsActive({expiresAt: expiry}, observed) && monitorIsActive({expiresAt: expiry}, due);
-}
-
-/** Qualify each current source before its wait is compacted for a consumer. */
-export function pendingAutomaticResumeCondition(todo: JsonObject, agentId: string): JsonObject | null {
-  const condition = jsonObject(todo.resume_condition);
-  if (!normalizedTodoId(todo.todo_id) || !["open", "deferred"].includes(String(todo.status)) ||
-      !(todo.claimed_by == null || todo.claimed_by === "" || todo.claimed_by === agentId) ||
-      !condition || !["resume_at", "todo_done", "monitor_changed"].includes(String(condition.kind))) return null;
-  try {
-    return resumeConditionHasKnownPendingTarget(condition, todo) &&
-      (condition.kind !== "monitor_changed" || hasScheduledMonitorSource(condition)) ? condition : null;
-  } catch {
-    return null;
-  }
-}
-
 /** Every causal obligation needs its own witness; shared prerequisites do not cover sibling work. */
 export function projectVisionWaitCoverage(requestPayload: unknown): JsonObject {
   const request = requireJsonObject(requestPayload, "vision wait coverage");
@@ -72,12 +43,6 @@ export function projectVisionWaitCoverage(requestPayload: unknown): JsonObject {
     witness_todo_ids: [...witnesses].sort(), uncovered_todo_ids: [...uncovered].sort()};
 }
 
-function normalizedTodoId(candidate: unknown): string | null {
-  if (typeof candidate !== "string") return null;
-  const id = candidate.trim().toLowerCase();
-  return /^todo_[a-z0-9_-]{3,64}$/.test(id) ? id : null;
-}
-
 function visionGapRows(value: unknown): JsonObject[] {
   return Array.isArray(value) ? value.filter((gap): gap is JsonObject =>
     typeof gap === "object" && gap !== null && !Array.isArray(gap)) : [];
@@ -161,28 +126,6 @@ function selectedWaitItems(blockers: JsonObject[], candidates: JsonObject[], wit
   return {waitItems: selectedCandidates.length ? selectedCandidates : selectedBlockers, automatic: selectedCandidates.length > 0};
 }
 
-function compactResumeCondition(resumeCondition: unknown): JsonObject {
-  const condition = typeof resumeCondition === "object" && resumeCondition !== null && !Array.isArray(resumeCondition)
-    ? resumeCondition as JsonObject : {};
-  const keys = ["schema_version", "resume_when", "satisfied", "kind", "target", "target_todo_id", "target_status",
-    "target_archive_state", "target_source_section", "target_task_class", "target_claimed_by", "target_next_due_at",
-    "capability", "provider", "scheduled_for", "evaluated_at", "clock_provider", "material_change",
-    "material_change_generation", "baseline_generation", "invalid_state", "target_expires_at"];
-  const compact: JsonObject = Object.fromEntries(keys.filter(key => key in condition).map(key => [key, condition[key]]));
-  if (condition.kind === "resume_at") compact.resume_receipt = condition.resume_receipt ?? null;
-  compact.satisfied = false;
-  return compact;
-}
-
-function automaticResumeFields(selected: JsonObject): JsonObject {
-  const fields: JsonObject = {};
-  for (const key of ["resume_when", "resume_monitor_generation"] as const) {
-    if (selected[key] !== undefined && selected[key] !== null) fields[key] = selected[key];
-  }
-  fields.resume_condition = compactResumeCondition(selected.resume_condition);
-  return fields;
-}
-
 function blockerReasonField(selectedTodo: JsonObject): JsonObject {
   return typeof selectedTodo.reason === "string" && selectedTodo.reason.trim()
     ? {blocker_reason: selectedTodo.reason.trim()} : {};
@@ -191,20 +134,8 @@ function blockerReasonField(selectedTodo: JsonObject): JsonObject {
 function waitStatePayload(input: {causalIds: string[]; gapCount: number; selectedTodo: JsonObject; waitItems: JsonObject[];
   agentId: string | null; automatic: boolean}): JsonObject {
   const {causalIds, gapCount, selectedTodo, waitItems, agentId, automatic} = input;
-  const pending = automatic && agentId ? waitItems.map(todo => ({todo,
-    condition: pendingAutomaticResumeCondition(todo, agentId)})).filter(
-      (source): source is {todo: JsonObject; condition: JsonObject} => source.condition !== null) : [];
-  const pendingCount = pending.length;
-  let selected = selectedTodo;
-  if (pendingCount === waitItems.length && pendingCount > 0) {
-    // A dependency needs periodic source rechecks. For a clock-only frontier,
-    // use its earliest deadline rather than the first priority-sorted row.
-    selected = (pending.find(source => source.condition.kind !== "resume_at") ?? pending.reduce((first, source) => {
-      const due = parseTodoTimestampMicros(String(source.condition.scheduled_for));
-      const prior = parseTodoTimestampMicros(String(first.condition.scheduled_for));
-      return due !== null && prior !== null && due < prior ? source : first;
-    })).todo;
-  }
+  const pending = selectPendingResume(waitItems, automatic ? agentId : null);
+  const selected = pending.todo ?? selectedTodo;
   const payload: JsonObject = {
     schema_version: WAIT_STATE_SCHEMA, state: "waiting",
     reason_code: automatic ? "exact_blocked_successor" : "current_agent_blocker",
@@ -212,7 +143,7 @@ function waitStatePayload(input: {causalIds: string[]; gapCount: number; selecte
     waiting_todo_ids: waitItems.slice(0, 5).map(candidate => normalizedTodoId(candidate.todo_id)).filter((id): id is string => id !== null),
     selected_todo_id: normalizedTodoId(selected.todo_id), selected_todo_status: selected.status ?? null,
     selected_todo_priority: selected.priority ?? null, selected_todo_claimed_by: selected.claimed_by ?? null,
-    ...(automatic ? {...automaticResumeFields(selected), pending_resume_count: pendingCount} : blockerReasonField(selected)),
+    ...(automatic ? {...automaticResumeFields(selected), pending_resume_count: pending.count} : blockerReasonField(selected)),
     causal_todo_ids: causalIds, deferred_acceptance_gap_count: gapCount,
     deferred_acceptance_gap_kinds: [VISION_GAP], automatic_resume: automatic,
     resume_behavior: automatic
diff --git a/loopx/control_plane/scheduler/arbitration.py b/loopx/control_plane/scheduler/arbitration.py
index 1794145baf..d9ee86f085 100644
--- a/loopx/control_plane/scheduler/arbitration.py
+++ b/loopx/control_plane/scheduler/arbitration.py
@@ -1,11 +1,11 @@
 from __future__ import annotations
 
-from typing import Any, cast
-
 from collections.abc import Collection, Mapping
 from dataclasses import dataclass
 from enum import Enum
+from typing import Any
 
+from ..agents.agent_scope_frontier import AGENT_SCOPE_FRONTIER_SCHEMA_VERSION, AgentScopeFrontierAction
 from ..goals.goal_vision_wait import exact_blocked_successor_wait_state
 
 SCHEDULER_ARBITRATION_SCHEMA_VERSION = "scheduler_arbitration_v0"
@@ -94,12 +94,19 @@ def _scheduler_arbitration_facts(
     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]}
+    if not payload:
+        return facts
     wait = exact_blocked_successor_wait_state(payload)
-    if wait:
-        payload = cast(dict[str, Any], payload)
-        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,
+    frontier = payload.get("agent_scope_frontier")
+    identity = payload.get("agent_identity")
+    agent_id = identity.get("agent_id") if isinstance(identity, Mapping) else None
+    scoped_wait = None
+    if (isinstance(frontier, Mapping) and frontier.get("schema_version") == AGENT_SCOPE_FRONTIER_SCHEMA_VERSION
+            and frontier.get("action") == AgentScopeFrontierAction.AGENT_SCOPE_WAIT.value
+            and frontier.get("agent_id") == agent_id):
+        scoped_wait = frontier.get("automatic_resume_todo")
+    if wait or scoped_wait:
+        facts.update({"effective_action": payload.get("effective_action"), "agent_id": agent_id,
+                      "vision_wait_state": wait or None, "scope_wait_todo": scoped_wait,
                       "heartbeat_receipt": payload.get("heartbeat_receipt")})
     return facts
diff --git a/loopx/control_plane/scheduler/hint.ts b/loopx/control_plane/scheduler/hint.ts
index 31a7958620..f472872568 100644
--- a/loopx/control_plane/scheduler/hint.ts
+++ b/loopx/control_plane/scheduler/hint.ts
@@ -299,10 +299,15 @@ function completeCadenceProjection(request: CompleteHintRequest, profile: HintPr
   const cooldown = notificationCooldown(profile, decision, request);
   if (cooldown) hint.user_gate_notification_cooldown = cooldown;
   const frontier = hintFrontier(request);
+  const frontierSeconds = frontier?.frontier_recheck_after_seconds;
+  const scheduled = arbitration.scheduled_recheck_after_seconds;
+  const recheck = typeof scheduled === "number"
+    ? Math.min(scheduled, typeof frontierSeconds === "number" ? frontierSeconds : scheduled) : frontierSeconds;
+  if (typeof scheduled === "number") (loops.local_scheduler as JsonObject).recheck_after_seconds = recheck;
   if (request.include_detail) hint.cold_path_detail = {schema_version: "scheduler_hint_detail_v0", source: "quota.should-run", ...loops,
     final_quota_replan_check: replan, reset_policy_detail: resetDetail(profile, identity, cadence, keys), stateful_backoff_detail: backoffDetail(profile, cadence, decision),
     ...(profile.detail ? {cadence_context: profile.detail} : {}), ...(frontier ? {frontier_recheck: frontier} : {})};
-  return applyHintExecutionContext(hint, context, frontier?.frontier_recheck_after_seconds);
+  return applyHintExecutionContext(hint, context, recheck);
 }
 
 const CADENCE_PROFILES: Record<string, HintProfile> = {
@@ -454,7 +459,7 @@ function selectedProfile(request: JsonObject, arbitration: JsonObject): HintProf
   const disposition = String(arbitration.disposition);
   let profile = CADENCE_PROFILES[disposition];
   if (!profile) throw new EffectRuntimeRequestError(`unhandled scheduler disposition: ${disposition}`);
-  if (disposition === "monitor_wait" && arbitration.pending_dependency_wait === true) {
+  if (disposition === "monitor_wait" && (arbitration.pending_dependency_wait === true || arbitration.scheduled_recheck_after_seconds != null)) {
     profile = {...profile, cli_limit: null, claude_limit: null};
   }
   if (disposition === "human_gate" || disposition === "monitor_wait") profile = boundedMonitorProfile(profile, payload, request);
diff --git a/loopx/control_plane/todos/automatic_resume.ts b/loopx/control_plane/todos/automatic_resume.ts
new file mode 100644
index 0000000000..3b4ef45509
--- /dev/null
+++ b/loopx/control_plane/todos/automatic_resume.ts
@@ -0,0 +1,79 @@
+import type { JsonObject } from "../runtime_decode.ts";
+import { jsonObject } from "../runtime_decode.ts";
+import { resumeConditionHasKnownPendingTarget } from "./resume_condition.ts";
+import { parseTodoTimestampMicros } from "../runtime_timestamp.ts";
+import { monitorIsActive } from "./summary_lanes.ts";
+
+export function normalizedTodoId(candidate: unknown): string | null {
+  if (typeof candidate !== "string") return null;
+  const id = candidate.trim().toLowerCase();
+  return /^todo_[a-z0-9_-]{3,64}$/.test(id) ? id : null;
+}
+
+function hasScheduledMonitorSource(condition: JsonObject): boolean {
+  const time = (value: unknown): number | null => {
+    const micros = typeof value === "string" ? parseTodoTimestampMicros(value) : null;
+    return micros === null ? null : Number(micros) / 1_000_000;
+  };
+  const observed = time(condition.evaluated_at), due = time(condition.target_next_due_at);
+  const expiry = time(condition.target_expires_at);
+  if (condition.clock_provider !== "runtime_clock" || observed === null || due === null ||
+      !(condition.target_archive_state === null || condition.target_archive_state === "active") ||
+      condition.target_expires_at !== null && expiry === null) return false;
+  return monitorIsActive({expiresAt: expiry}, observed) && monitorIsActive({expiresAt: expiry}, due);
+}
+
+/** Qualify each current source before its wait is compacted for a consumer. */
+export function pendingAutomaticResumeCondition(todo: JsonObject, agentId: string): JsonObject | null {
+  const condition = jsonObject(todo.resume_condition);
+  if (!normalizedTodoId(todo.todo_id) || !["open", "deferred"].includes(String(todo.status)) ||
+      !(todo.claimed_by == null || todo.claimed_by === "" || todo.claimed_by === agentId) ||
+      !condition || !["resume_at", "todo_done", "monitor_changed"].includes(String(condition.kind))) return null;
+  try {
+    return resumeConditionHasKnownPendingTarget(condition, todo) &&
+      (condition.kind !== "monitor_changed" || hasScheduledMonitorSource(condition)) ? condition : null;
+  } catch {
+    return null;
+  }
+}
+
+function compactResumeCondition(resumeCondition: unknown): JsonObject {
+  const condition = typeof resumeCondition === "object" && resumeCondition !== null && !Array.isArray(resumeCondition)
+    ? resumeCondition as JsonObject : {};
+  const keys = ["schema_version", "resume_when", "satisfied", "kind", "target", "target_todo_id", "target_status",
+    "target_archive_state", "target_source_section", "target_task_class", "target_claimed_by", "target_next_due_at",
+    "capability", "provider", "scheduled_for", "evaluated_at", "clock_provider", "material_change",
+    "material_change_generation", "baseline_generation", "invalid_state", "target_expires_at"];
+  const compact: JsonObject = Object.fromEntries(keys.filter(key => key in condition).map(key => [key, condition[key]]));
+  if (condition.kind === "resume_at") compact.resume_receipt = condition.resume_receipt ?? null;
+  compact.satisfied = false;
+  return compact;
+}
+
+export function automaticResumeFields(selected: JsonObject): JsonObject {
+  const fields: JsonObject = {};
+  for (const key of ["resume_when", "resume_monitor_generation"] as const) {
+    if (selected[key] !== undefined && selected[key] !== null) fields[key] = selected[key];
+  }
+  fields.resume_condition = compactResumeCondition(selected.resume_condition);
+  return fields;
+}
+
+/** Reduce the entire scoped source before display limits hide its first deadline. */
+export function selectPendingResume(waitItems: JsonObject[], agentId: string | null) {
+  const pending = agentId ? waitItems.map(todo => ({todo,
+    condition: pendingAutomaticResumeCondition(todo, agentId)})).filter(
+      (source): source is {todo: JsonObject; condition: JsonObject} => source.condition !== null) : [];
+  const pendingCount = pending.length;
+  let selected: JsonObject | null = null;
+  if (pendingCount === waitItems.length && pendingCount > 0) {
+    // A dependency needs periodic source rechecks. For a clock-only frontier,
+    // use its earliest deadline rather than the first priority-sorted row.
+    selected = (pending.find(source => source.condition.kind !== "resume_at") ?? pending.reduce((first, source) => {
+      const due = parseTodoTimestampMicros(String(source.condition.scheduled_for));
+      const prior = parseTodoTimestampMicros(String(first.condition.scheduled_for));
+      return due !== null && prior !== null && due < prior ? source : first;
+    })).todo;
+  }
+  return {count: pendingCount, todo: selected};
+}
diff --git a/loopx/control_plane/todos/resume_planning.ts b/loopx/control_plane/todos/resume_planning.ts
index 12bb62ff2e..797a64a228 100644
--- a/loopx/control_plane/todos/resume_planning.ts
+++ b/loopx/control_plane/todos/resume_planning.ts
@@ -11,6 +11,8 @@ import {
   resumeConditionHasKnownPendingTarget,
 } from "./resume_condition.ts";
 
+import { automaticResumeFields, selectPendingResume } from "./automatic_resume.ts";
+
 export const RESUME_PLANNING_REQUEST = "todo_resume_planning_request_v0";
 export const REFERENCED_RESUME_PLANNING_REQUEST = "todo_resume_planning_request_v1";
 export const RESUME_PLANNING_RESULT = "todo_resume_planning_v0";
@@ -349,13 +351,17 @@ export function projectTodoResumePlanning(value: unknown): JsonObject {
   const resumeBlocked = blocked(sources.resume_blocked_items.length ? sources.resume_blocked_items
     : [...sources.items, ...sources.backlog_items, ...sources.first_open_items]);
   const monitorItems = monitorBlocked(resumeBlocked);
+  const waits = payloads(successorWaits(resumeBlocked, deferredItems, display.agent));
+  const pending = selectPendingResume(waits, display.agent).todo;
   return {
     schema_version: RESUME_PLANNING_RESULT,
     deferred_lanes: {...deferredVisibility(deferredItems, candidates, display), deferred_work_counts: counts},
     resume_blocked_lanes: blockedVisibility(resumeBlocked, monitorItems, display),
     capacity_fields: capacityFields,
     deferred_items: payloads(deferredItems), monitor_blocked_items: payloads(monitorItems),
-    blocked_successor_items: payloads(successorWaits(resumeBlocked, deferredItems, display.agent)),
+    blocked_successor_items: waits,
+    ...(pending ? {automatic_resume_todo: {todo_id: pending.todo_id, status: pending.status,
+      claimed_by: pending.claimed_by ?? null, ...automaticResumeFields(pending)}} : {}),
     dependency_cycle_replan: dependencyCycleReplan([...resumeBlocked, ...deferredItems], display.agent),
   };
 }
diff --git a/loopx/control_plane/work_items/interaction_contract.ts b/loopx/control_plane/work_items/interaction_contract.ts
index 4f4450eeb9..8e79eb3016 100644
--- a/loopx/control_plane/work_items/interaction_contract.ts
+++ b/loopx/control_plane/work_items/interaction_contract.ts
@@ -13,12 +13,13 @@ import {
   optionalNonEmptyString,
   jsonObject,
 } from "../runtime_decode.ts";
+import { parseTodoTimestampMicros } from "../runtime_timestamp.ts";
 import { EffectiveAction } from "../quota/effective_action.generated.ts";
 import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts";
 
 import type { JsonObject } from "../effect_program.ts";
 import { AGENT_SCOPE_FRONTIER_ACTIONS, AgentScopeFrontierAction } from "../agents/agent_scope_frontier.generated.ts";
-import { pendingAutomaticResumeCondition } from "../goals/vision_wait_coverage.ts";
+import { pendingAutomaticResumeCondition } from "../todos/automatic_resume.ts";
 import {compactTodoContextSource} from "../quota/context_delivery.ts";
 import {compactPythonWhitespace} from "../coordination/python_whitespace.ts";
 
@@ -34,22 +35,24 @@ function exactBlockedSuccessorWait(request: JsonObject, monitorWait = false): Js
 }
 
 /** Share the resume owner's source proof; a Turn binding retains supervision. */
-function pendingAutomaticSuccessor(request: JsonObject, wait: JsonObject): JsonObject | null {
+function pendingAutomaticSuccessor(request: JsonObject, wait: JsonObject | null): 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));
+  if (!unbound || typeof request.agent_id !== "string" || !request.agent_id) return null;
+  if (!wait) {
+    const scoped = jsonObject(request.scope_wait_todo);
+    return scoped ? pendingAutomaticResumeCondition(scoped, request.agent_id) : null;
+  }
   const count = wait.waiting_todo_count;
   const complete = typeof count === "number" && Number.isSafeInteger(count) && count > 0 &&
     (wait.pending_resume_count === undefined ? count === 1 : wait.pending_resume_count === count);
-  if (unbound && typeof request.agent_id === "string" && !!request.agent_id &&
-      wait.agent_id === request.agent_id && complete) {
-    return pendingAutomaticResumeCondition({todo_id: wait.selected_todo_id,
-      status: wait.selected_todo_status, claimed_by: wait.selected_todo_claimed_by,
-      resume_when: wait.resume_when, resume_monitor_generation: wait.resume_monitor_generation,
-      resume_condition: wait.resume_condition}, request.agent_id);
-  }
-  return null;
+  if (!complete || wait.agent_id !== request.agent_id) return null;
+  return pendingAutomaticResumeCondition({todo_id: wait.selected_todo_id,
+    status: wait.selected_todo_status, claimed_by: wait.selected_todo_claimed_by,
+    resume_when: wait.resume_when, resume_monitor_generation: wait.resume_monitor_generation,
+    resume_condition: wait.resume_condition}, request.agent_id);
 }
 
 /** A proven pending automatic successor needs only fresh source rechecks.
@@ -604,13 +607,13 @@ export function projectSchedulerArbitration(value: unknown): JsonObject {
     disposition === "monitor_wait" && request.effective_action === EffectiveAction.MONITOR_QUIET_SKIP;
   const condition = !errors.length && waiting &&
     !facts.must_attempt && !facts.user_required && facts.quiet_noop_allowed &&
-    !facts.spend_allowed_now && !facts.spend_after_validation && wait
+    !facts.spend_allowed_now && !facts.spend_after_validation
     ? pendingAutomaticSuccessor(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 = disposition === "agent_scope_wait" && condition?.kind === "resume_at" ? {scheduled_recheck_after_seconds: Math.ceil(
-    (Date.parse(String(condition.scheduled_for)) - Date.parse(String(condition.evaluated_at))) / 1000,
-  )} : {};
+  const scheduled = condition?.kind === "resume_at" ? {scheduled_recheck_after_seconds: Math.ceil(Number(
+    parseTodoTimestampMicros(String(condition.scheduled_for))! - parseTodoTimestampMicros(String(condition.evaluated_at))!
+  ) / 1_000_000)} : {};
   const dependency = condition && ["todo_done", "monitor_changed"].includes(String(condition.kind))
     ? {pending_dependency_wait: true} : {};
   return {mode, errors: [...new Set(errors)],
diff --git a/scripts/external_scheduler_worker.py b/scripts/external_scheduler_worker.py
index 61feb64de3..42ced4e060 100644
--- a/scripts/external_scheduler_worker.py
+++ b/scripts/external_scheduler_worker.py
@@ -74,10 +74,7 @@ class TickDecision:
     final_probe_action: str
     reset_token: str
     terminal: bool
-
-    @property
-    def sleep_seconds(self) -> int:
-        return max(5, int(self.interval_minutes) * 60)
+    recheck_after_seconds: int | None = None
 
 
 def _mapping(value: Any) -> dict[str, Any]:
@@ -150,6 +147,10 @@ def parse_tick(payload: dict[str, Any]) -> TickDecision:
         raise ValueError(f"{admission_field} must be a boolean; refusing model wake")
     local = _extract_local_scheduler(payload)
 
+    recheck = local.get("recheck_after_seconds")
+    if recheck is not None and (type(recheck) is not int or recheck <= 0):
+        raise ValueError("local_scheduler.recheck_after_seconds must be a positive integer")
+
     progression_values = local.get("example_progression_minutes")
     progression: tuple[int, ...] = ()
     if isinstance(progression_values, (list, tuple)):
@@ -194,6 +195,7 @@ def parse_tick(payload: dict[str, Any]) -> TickDecision:
         final_probe_action=str(final_probe.get("action") or "").strip(),
         reset_token=_extract_reset_token(payload),
         terminal=False,
+        recheck_after_seconds=recheck,
     )
 
 
@@ -819,6 +821,9 @@ def save(fields: dict[str, Any]) -> None:
             interval_minutes, phase = select_interval(
                 decision, unchanged_count=unchanged_count
             )
+            interval_seconds = interval_minutes * 60
+            if not decision.should_run and decision.recheck_after_seconds is not None:
+                interval_seconds = min(interval_seconds, decision.recheck_after_seconds)
 
             effective_action = str(payload.get("effective_action") or "").strip()
             selected = _mapping(payload.get("selected_todo"))
@@ -1009,7 +1014,7 @@ def save(fields: dict[str, Any]) -> None:
                 "status=waiting "
                 f"phase={phase} class={decision.cadence_class} "
                 f"effective_action={effective_action or '-'} "
-                f"next_check_minutes={interval_minutes} "
+                f"next_check_minutes={interval_seconds / 60:g} "
                 f"unchanged_count={unchanged_count}"
                 + (
                     f" unchanged_limit={decision.unchanged_limit}"
@@ -1086,7 +1091,7 @@ def save(fields: dict[str, Any]) -> None:
         # still owns continuation; waiting/observe-only ticks retain their
         # cadence and failures use the independent backoff path above.
         if not wake_completed:
-            sleep(max(5, interval_minutes * 60))
+            sleep(interval_seconds)
 
 
 def main(argv: Sequence[str] | None = None) -> int:
diff --git a/tests/control_plane/test_dated_vision_wait_cli.py b/tests/control_plane/test_dated_vision_wait_cli.py
index e0c6a7eb95..bef18481d0 100644
--- a/tests/control_plane/test_dated_vision_wait_cli.py
+++ b/tests/control_plane/test_dated_vision_wait_cli.py
@@ -31,7 +31,7 @@ def isolated_effect(tmp_path: Path, monkeypatch):
     restart_effect_runtime()
 
 
-def fixture(root: Path, resume: str = FUTURE) -> tuple[Path, Path, Path]:
+def fixture(root: Path, resume: str = FUTURE, *, with_vision: bool = True) -> tuple[Path, Path, Path]:
     project, runtime = root / "project", root / "runtime"
     state = project / f".codex/goals/{GOAL}/ACTIVE_GOAL_STATE.md"
     state.parent.mkdir(parents=True)
@@ -58,6 +58,8 @@ def fixture(root: Path, resume: str = FUTURE) -> tuple[Path, Path, Path]:
                    "coordination": {"registered_agents": [AGENT], "agent_model": "peer_v1"},
                    "quota": {"compute": 1.0, "window_hours": 24, "allowed_slots": 20}}],
     }), encoding="utf-8")
+    if not with_vision:
+        return registry, runtime, state
     runs = runtime / f"goals/{GOAL}/runs"
     runs.mkdir(parents=True)
     run = {"goal_id": GOAL, "agent_id": AGENT,
@@ -292,6 +294,88 @@ def advance_host(seconds):
     assert not any(run.get("classification") in ("quota_monitor_poll", "quota_slot_spent") for run in runs)
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_deferred_todo_without_vision_wakes_original_agent_at_deadline(tmp_path, isolated_effect, monkeypatch, provider):
+    from argparse import Namespace
+    from contextlib import redirect_stdout
+    from datetime import datetime, timedelta, timezone
+    from io import StringIO
+    from loopx import entrypoint
+    from loopx.control_plane.runtime import time as runtime_time
+
+    monkeypatch.syspath_prepend(str(REPO / "scripts"))
+    import external_scheduler_worker as worker
+
+    now = datetime.now(timezone.utc).replace(microsecond=0)
+    due = now + timedelta(hours=2, seconds=17)
+
+    class FixtureDatetime(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return now.astimezone(tz) if tz else now.replace(tzinfo=None)
+
+    monkeypatch.setattr(runtime_time, "datetime", FixtureDatetime)
+    registry, runtime, state = fixture(tmp_path, f"resume_at:{due.isoformat()}", with_vision=False)
+    initialize_canonical_source_fixture(runtime, GOAL, state_path=state, provider=provider)
+    observations = []
+    wakes = []
+
+    def command(*args):
+        out = StringIO()
+        with redirect_stdout(out):
+            rc = entrypoint.main(["--format", "json", "--registry", str(registry),
+                                  "--runtime-root", str(runtime), *args])
+        data = json.loads(out.getvalue())
+        assert rc == 0, data
+        return data
+
+    def probe(argv, **kwargs):
+        guard = command(*argv[argv.index("quota"):])
+        if not observations:
+            # The managed runtime and CLI host consume the same scoped due time.
+            managed = command("quota", "should-run", "--goal-id", GOAL, "--agent-id", AGENT,
+                              "--runtime-profile", "ark_managed_agent_goal", "--context-delivery", "off")
+            continuation = managed["scheduler_hint"]["goal_runtime_continuation"]
+            assert continuation["disposition"] == "defer"
+            local = guard["scheduler_hint"]["cold_path_detail"]["local_scheduler"]
+            assert continuation["recheck_after_seconds"] == local["recheck_after_seconds"] == 7217
+        observations.append(guard)
+        return guard
+
+    def sleep(seconds):
+        nonlocal now
+        assert 0 < seconds <= (due - now).total_seconds()
+        now += timedelta(seconds=seconds)
+        assert len(observations) < 20
+
+    def wake(*args, **kwargs):
+        guard = observations[-1]
+        assert now == due and not wakes
+        assert guard["effective_action"] == "successor_replan_required"
+        assert guard["selected_todo"]["todo_id"] == TODO
+        assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] is False
+        updated = command("todo", "update", "--goal-id", GOAL, "--agent-id", AGENT,
+                          "--todo-id", TODO, "--status", "open", "--clear-resume-when",
+                          "--reason", "Scheduled condition satisfied")
+        assert updated.get("ok"), updated
+        readback = command("todo", "list", "--goal-id", GOAL, "--todo-id", TODO)
+        assert readback["todo"]["status"] == "open"
+        assert readback["todo"]["todo_id"] == TODO
+        wakes.append(TODO)
+        return worker.CappedProcessResult(75, b"", None)
+
+    monkeypatch.setattr(worker, "run_quota_should_run", probe)
+    monkeypatch.setattr(worker, "_run_wake", wake)
+    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 / "worker.json"), once=False, wake_cmd="fixture",
+                     wake_stop_exit_code=75, error_backoff_seconds=5, quota_timeout_seconds=30,
+                     wake_timeout_seconds=None)
+    assert worker.run_worker(args, sleep=sleep) == 0
+    assert wakes == [TODO] and len(observations) > 3
+    assert all(not item["interaction_contract"]["agent_channel"]["must_attempt"] for item in observations[:-1])
+
+
 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:
diff --git a/tests/control_plane_ts/resume_planning.test.ts b/tests/control_plane_ts/resume_planning.test.ts
index 019b0bc4a8..c2a8d4bd18 100644
--- a/tests/control_plane_ts/resume_planning.test.ts
+++ b/tests/control_plane_ts/resume_planning.test.ts
@@ -54,6 +54,27 @@ test("only the actual pending Monitor generation can cover an in-scope successor
   assert.equal(project({...waiting, excluded: ["agent-a"]}).length, 0);
 });
 
+test("automatic resume selects the earliest scoped clock before the display limit", () => {
+  const dates = ["06", "05", "04", "03", "02", "01"];
+  const rows = dates.map((day, index) => {
+    const id = "todo_clock_" + day, target = `2099-01-${day}T00:00:00Z`;
+    const todo = {todo_id: id, text: "Wait for the scheduled input", status: "deferred",
+      task_class: "advancement_task", resume_when: "resume_at:" + target};
+    const conditions = evaluateTodoResumeConditions({
+      schema_version: "todo_resume_evaluation_request_v0", items: [todo], source_items: [todo],
+      evaluated_at: "2098-12-01T00:00:00Z",
+    }).conditions as JsonObject[];
+    const claim = index === dates.length - 1 ? "peer" : "agent-a";
+    return fact(id, {index, claim, resume: todo.resume_when,
+      payload: {...todo, index, claimed_by: claim, resume_ready: false, resume_condition: conditions[0].condition}});
+  });
+  const input = request({item_limit: 1});
+  (input.sources as JsonObject).deferred_items = rows;
+  const result = projectTodoResumePlanning(input);
+  assert.equal((result.automatic_resume_todo as JsonObject).todo_id, "todo_clock_02");
+  assert.equal(((result.deferred_lanes as JsonObject).deferred_items as JsonObject[]).length, 1);
+});
+
 test("capacity evaluation and claim lanes use one snapshot without mutating it", () => {
   const input = request({ available_capabilities: ["compiler"] });
   (input.sources as JsonObject).items = [fact("todo_capacity", { claim: "agent-a" }),
diff --git a/tests/control_plane_ts/scheduler_arbitration.test.ts b/tests/control_plane_ts/scheduler_arbitration.test.ts
index b0a60a0d3f..894ab496c3 100644
--- a/tests/control_plane_ts/scheduler_arbitration.test.ts
+++ b/tests/control_plane_ts/scheduler_arbitration.test.ts
@@ -172,6 +172,21 @@ test("only an unbound pending automatic dated successor carries a scheduled rech
   }
 });
 
+test("a scoped Todo clock survives missing Vision but cannot override a bound Turn or another owner", () => {
+  const input = datedWait();
+  const wait = input.vision_wait_state as JsonObject;
+  const todo = {todo_id: wait.selected_todo_id, status: wait.selected_todo_status,
+    claimed_by: input.agent_id, resume_when: wait.resume_when, resume_condition: wait.resume_condition};
+  const scoped = {...input, vision_wait_state: null, scope_wait_todo: todo};
+  assert.equal(projectSchedulerArbitration(scoped).scheduled_recheck_after_seconds, 1800);
+  for (const patch of [{scope_wait_todo: {...todo, claimed_by: "other-agent"}},
+    {scope_wait_todo: {...todo, status: "done"}}, {scope_wait_todo: {...todo, resume_condition: null}},
+    {heartbeat_receipt: {status: "pending"}}, {heartbeat_receipt: {status: "committed", settlement_identity: {}}},
+    facts("user_gate", true, false, false, false), facts("terminal_no_followup", false, false, false, true)]) {
+    assert.equal(projectSchedulerArbitration({...scoped, ...patch}).scheduled_recheck_after_seconds, undefined);
+  }
+});
+
 test("unproven waits cannot relax finite scheduler supervision", () => {
   const input = datedWait();
   const wait = input.vision_wait_state as JsonObject;