Keep external and managed schedulers aligned on deferred Todo wakeups

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
 

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 {}),
+        },
     )
 
 

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,28 +134,16 @@ 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",
     agent_id: agentId, waiting_todo_count: waitItems.length,
     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

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

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);

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};
+}