fix(scheduler): preserve qualified dependency waits and scoped admission

docs/reference/protocols/goal-vision-replan-contract-v0.md

@@ -1122,6 +1122,29 @@ another uncovered binding. When there is no other selectable advancement, quota/
 `vision_acceptance_gap` is deferred while that read model is active, so the
 agent can remain quiet instead of inventing duplicate successor work.
 
+For a single exact successor without a bound heartbeat Turn, the scheduler can
+keep inexpensive source rechecks alive beyond the unchanged-poll limit when
+the resume owner supplies positive pending evidence. Alongside the existing
+clock-backed `resume_at` case, `todo_done` requires an identified, unarchived
+source of the supported task class that is still open or deferred.
+`monitor_changed` requires the exact open Monitor, the successor's matching
+generation baseline, and a current generation that has not advanced. The
+Monitor source must also carry a valid next-due time, the runtime clock
+observation, and explicit expiry evidence: either no expiry or an expiry later
+than both the observation and the next due time. These source facts are
+projected before compaction, including explicit null values that distinguish
+known absence of expiry or archival from missing evidence.
+
+This quiet path grants no delivery or spend. Missing sources, incomplete proof,
+an unscheduled or expired Monitor, and an existing Turn or settlement binding
+retain their original supervision. Supplying a new Turn id to a quota command
+is a bound operation, not an unbound scheduler recheck. Monitor-only waits keep
+their existing schedule-based cadence caps; when a Monitor becomes due, fresh
+quota admission must select its work before quiet waiting. A material poll then
+advances the original generation fence so the successor can resume. Goal stop,
+user gates and independently runnable advancement retain their existing
+priority.
+
 This is a read model, not a stored vision or todo lifecycle state. When the
 condition becomes ready, normal open-todo or deferred-successor routing resumes
 automatically and the active vision remains available for acceptance auditing.

loopx/control_plane/goals/goal_vision_wait.py

@@ -74,17 +74,25 @@ def _compact_resume_condition(value: Any) -> dict[str, Any]:
             "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",
         )
         if value.get(key) is not None
     }
+    if "target_archive_state" in value and value["target_archive_state"] is None:
+        # An explicitly unarchived source is positive evidence. An omitted
+        # source field must remain omitted for the typed resume owner.
+        compact["target_archive_state"] = None
+    if "target_expires_at" in value:
+        compact["target_expires_at"] = value["target_expires_at"]
     if value.get("kind") == "resume_at":
         # Keep the complete positive clock proof for the typed interaction
         # owner, including the explicit absence of an already-triggered receipt.
@@ -281,6 +289,7 @@ def build_goal_vision_wait_state(
             "selected_todo_priority": selected.get("priority"),
             "selected_todo_claimed_by": selected.get("claimed_by"),
             "resume_when": selected.get("resume_when"),
+            "resume_monitor_generation": selected.get("resume_monitor_generation"),
             "resume_condition": _compact_resume_condition(
                 selected.get("resume_condition")
             ),

loopx/control_plane/quota/should_run_packet.py

@@ -470,29 +470,40 @@ def _resolve_agent_lane_delivery_route(
     delivery_preemptions: list[str],
     should_run: bool,
     normal_delivery_allowed: bool,
-) -> dict[str, Any] | None:
-    """Project candidates once, then let TypeScript own their delivery route."""
+) -> tuple[dict[str, Any] | None, bool]:
+    """Return the selected route with its typed admission decision."""
 
     if prepared.requested_action_todo_id is not None:
+        scoped_fallback = prepared.scoped_user_gate_fallback or {}
+        scoped_selected = scoped_fallback.get("selected_executable") or {}
         qualification = qualify_action_selection_from_inventory(
             requested_todo_id=prepared.requested_action_todo_id,
             candidate=prepared.requested_action_candidate,
             source_items=prepared.agent_todo_planning_source_items,
             should_run=should_run,
             normal_delivery_allowed=normal_delivery_allowed,
             delivery_preemptions=delivery_preemptions,
+            scoped_fallback_todo_id=normalize_todo_id(scoped_selected.get("todo_id")),
+            quota_state=prepared.state,
+            goal_health_ok=prepared.goal_health_ok,
         )
         prepared.action_selection_qualification = qualification
 
+    if (prepared.action_selection_qualification or {}).get("state") == "qualified":
+        # The typed selection owner may admit this exact scoped fallback while
+        # the broader user-gated delivery lane remains closed. Keep selected
+        # workspace/boundary guards active without reopening normal delivery.
+        should_run = True
+
     if isinstance(prepared.guarded_agent_lane_next_action, dict):
         # Workspace/scope guards can change admission after initial selection.
         # Keep the selected identity frozen, but report the final qualification.
-        return prepared.guarded_agent_lane_next_action
+        return prepared.guarded_agent_lane_next_action, should_run
 
     if prepared.requested_action_todo_id is not None:
         qualification = prepared.action_selection_qualification or {}
         if qualification.get("state") != "qualified":
-            return None
+            return None, should_run
         if not isinstance(prepared.requested_action_candidate, dict):
             raise RuntimeError(
                 "qualified action selection is missing its projected candidate"
@@ -501,7 +512,7 @@ def _resolve_agent_lane_delivery_route(
             **prepared.requested_action_candidate,
             "selected_by": "agent_action_selection",
             "selection_binding": "pending_action_selection",
-        }
+        }, should_run
 
     item = prepared.item
     fallback = prepared.receipt_bound_agent_next_action
@@ -548,7 +559,7 @@ def _resolve_agent_lane_delivery_route(
         (prepared.agent_identity or {}).get("agent_id")
     )
     if not delivery_agent_id:
-        return fallback
+        return fallback, should_run
 
     continuity_todo = prepared.delivery_continuity_todo
     delivery_anchor = prepared.delivery_continuity_anchor
@@ -567,7 +578,7 @@ def _resolve_agent_lane_delivery_route(
         and fallback is None
         and replan_candidate is None
     ):
-        return None
+        return None, should_run
     delivery_route = evaluate_delivery_route(
         agent_id=delivery_agent_id,
         previous_todo_id=(
@@ -668,7 +679,7 @@ def _resolve_agent_lane_delivery_route(
         # The router may find an independent successor after the bound Todo
         # becomes unavailable.  That successor cannot replace an already
         # committed settlement identity inside the same heartbeat Turn.
-        return None
+        return None, should_run
 
     boundary = delivery_route.get("boundary")
     if (
@@ -679,8 +690,8 @@ def _resolve_agent_lane_delivery_route(
         return {
             **selected_action,
             "delivery_boundary": boundary["delivery_boundary"],
-        }
-    return selected_action
+        }, should_run
+    return selected_action, should_run
 
 
 def _selected_deferred_replan_frontier(
@@ -1106,7 +1117,7 @@ def _resolve_quota_should_run_route(
         capability_repair_allowed=capability_repair_allowed,
         workspace_repair_allowed=workspace_repair_allowed,
     )
-    agent_lane_next_action = _resolve_agent_lane_delivery_route(
+    agent_lane_next_action, should_run = _resolve_agent_lane_delivery_route(
         prepared,
         due_monitor_attempt=due_monitor_attempt,
         receipt_bound_replan_decision=receipt_bound_replan_decision,
@@ -1141,10 +1152,15 @@ def _resolve_quota_should_run_route(
                 agent_todo_summary=prepared.agent_todo_summary,
                 agent_lane_next_action=agent_lane_next_action,
                 work_lane_contract=prepared.work_lane_contract,
-                candidate_should_run=bool(
-                    (should_run and normal_delivery_allowed)
-                    or ready_deferred_resume_candidates
-                ),
+                # A satisfied dependency is readiness, not admission. The
+                # existing scoped fallback above qualifies an independent
+                # successor under a user gate; an unqualified ready row must
+                # not resurrect delivery refused by the current quota owner.
+                candidate_should_run=bool(should_run and (
+                    normal_delivery_allowed
+                    or (ready_deferred_resume_candidates
+                        and prepared.goal_health_ok and state == "eligible")
+                )),
             )
         )
         if (

loopx/control_plane/scheduler/arbitration.py

@@ -31,6 +31,7 @@ class SchedulerArbitration:
     mode: str
     errors: tuple[str, ...] = ()
     scheduled_recheck_after_seconds: int | None = None
+    pending_dependency_wait: bool = False
 
     @property
     def ok(self) -> bool:
@@ -60,8 +61,11 @@ def from_projection(cls, projection: Any) -> SchedulerArbitration:
         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")
+        pending_dependency = projection.get("pending_dependency_wait", False)
+        if type(pending_dependency) is not bool:
+            raise TypeError("TypeScript scheduler pending dependency wait must be boolean")
         return cls(SchedulerDisposition(projection["disposition"]), projection["reason_code"],
-                   projection["mode"], tuple(projection["errors"]), scheduled)
+                   projection["mode"], tuple(projection["errors"]), scheduled, pending_dependency)
 
 
 def _scheduler_arbitration_facts(

loopx/control_plane/scheduler/scheduler_hint.py

@@ -1180,6 +1180,7 @@ def _monitor_bounded_wait_profile(
     cadence_class: str,
     default_interval_minutes: int,
     max_interval_minutes: int,
+    unchanged_poll_limit: int | None = 3,
 ) -> dict[str, Any]:
     """Cap a wait profile to the tightest continuous-monitor wakeup."""
 
@@ -1205,8 +1206,8 @@ def _monitor_bounded_wait_profile(
             "app_automation_initial_rrule": rrule_for_minutes(initial_interval),
             "app_automation_max_interval_minutes": max_interval_minutes,
             "unchanged_poll_backoff_multiplier": 2,
-            "local_scheduler_unchanged_poll_limit": 3,
-            "claude_code_loop_unchanged_poll_limit": 3,
+            "local_scheduler_unchanged_poll_limit": unchanged_poll_limit,
+            "claude_code_loop_unchanged_poll_limit": unchanged_poll_limit,
             **monitor_plan["reset_profile"],
         }
         if isinstance(monitor_plan, dict)
@@ -1617,6 +1618,20 @@ def build_scheduler_hint(
                 claude_limit=None,
                 cadence_progression_override=progression,
             )
+        if arbitration.pending_dependency_wait:
+            return builder.build(
+                action="backoff_until_state_change",
+                cadence_class="agent_scope_wait",
+                reason=(
+                    "an exact successor has a registered pending dependency; keep quiet "
+                    "source rechecks and follow fresh admission gates when it changes"
+                ),
+                codex_interval=10,
+                codex_max=60,
+                cli_limit=None,
+                claude_limit=None,
+                cadence_progression_override=[10, 20, 30, 60],
+            )
         return builder.build(
             action="backoff_until_reassigned",
             cadence_class="agent_scope_wait",
@@ -1634,11 +1649,13 @@ def build_scheduler_hint(
         )
 
     if arbitration.disposition == SchedulerDisposition.MONITOR_WAIT:
+        unchanged_limit = None if arbitration.pending_dependency_wait else 3
         monitor_profile = _monitor_bounded_wait_profile(
             payload,
             cadence_class="monitor_wait",
             default_interval_minutes=MONITOR_WAIT_HOST_FLOOR_MINUTES,
             max_interval_minutes=60,
+            unchanged_poll_limit=unchanged_limit,
         )
         return builder.build(
             action="backoff_until_material_transition",
@@ -1647,8 +1664,8 @@ def build_scheduler_hint(
                 "monitor-only quiet polls should remain alive but use a slower "
                 "cadence until material evidence, a blocker, or replan obligation appears"
             ),
-            cli_limit=3,
-            claude_limit=3,
+            cli_limit=unchanged_limit,
+            claude_limit=unchanged_limit,
             **monitor_profile,
         )
 

loopx/control_plane/todos/resume_condition.py

@@ -78,6 +78,8 @@
     "resume_ready",
     "resume_monitor_generation",
     "material_change_generation",
+    "next_due_at",
+    "expires_at",
 )
 
 

loopx/control_plane/todos/resume_condition.ts

@@ -159,6 +159,8 @@ function todoItem(value: unknown, label: string): TodoItem {
     "source_section",
     "claimed_by",
     "task_repository",
+    "next_due_at",
+    "expires_at",
   ] as const) {
     const normalized = optionalString(raw[field], `${label}.${field}`);
     if (normalized !== undefined) item[field] = normalized;
@@ -544,6 +546,17 @@ function conditionFor(
   condition.target_todo_id = spec.target;
   condition.target_status = monitor?.status ?? null;
   condition.target_task_class = monitor?.task_class ?? null;
+  if (monitor) {
+    // Source scheduling facts qualify quiet rechecks separately from the
+    // material-generation fence. They do not make the successor ready.
+    condition.target_archive_state = monitor.archive_state ?? null;
+    condition.target_next_due_at = monitor.next_due_at ?? null;
+    condition.target_expires_at = monitor.expires_at ?? null;
+    if (evaluatedAt) {
+      condition.evaluated_at = evaluatedAt.iso;
+      condition.clock_provider = "runtime_clock";
+    }
+  }
   condition.baseline_generation = baseline ?? null;
   condition.material_change_generation = generation;
   condition.generation_fence = "strictly_greater_than_baseline";
@@ -674,7 +687,7 @@ export function evaluateTodoResumeConditions(value: unknown): JsonObject {
     const spec = parseResumeWhen(item.resume_when);
     return spec?.kind === "resume_at" && (!requestedKinds || requestedKinds.has(spec.kind));
   });
-  const evaluatedAt = needsRuntimeClock ? resumeTimestamp(request.evaluated_at) : null;
+  const evaluatedAt = resumeTimestamp(request.evaluated_at);
   if (needsRuntimeClock && !evaluatedAt) {
     throw new EffectRuntimeRequestError(
       "todo_resume_evaluation_request.evaluated_at must be a timezone-aware RFC3339 timestamp",

loopx/control_plane/todos/resume_planning.ts

@@ -8,6 +8,7 @@ import {
 import {
   diagnoseTodoResumeCondition, type ResumeConditionDiagnosis,
   evaluateTodoResumeConditions, TODO_RESUME_EVALUATION_REQUEST_SCHEMA_VERSION,
+  resumeConditionHasKnownPendingTarget,
 } from "./resume_condition.ts";
 
 export const RESUME_PLANNING_REQUEST = "todo_resume_planning_request_v0";
@@ -135,10 +136,12 @@ function successorWaits(resumeBlocked: readonly Item[], deferredItems: readonly
   return ordered(unique([...resumeBlocked, ...deferredItems.filter((entry) => entry.payload.resume_ready === false)])
     .filter((entry) => {
       const lane = claimLane(entry, agent);
+      const condition = jsonObject(entry.payload.resume_condition);
       return entry.payload.task_class === "advancement_task" &&
         (lane === "current_agent" || lane === "unclaimed") && entry.resume !== null &&
-        jsonObject(entry.payload.resume_condition)?.satisfied === false &&
-        entry.diagnosis?.state === "pending" && entry.diagnosis.kind !== "monitor_changed";
+        condition?.satisfied === false && entry.diagnosis?.state === "pending" &&
+        (entry.diagnosis.kind !== "monitor_changed" ||
+          resumeConditionHasKnownPendingTarget(condition, entry.payload));
     }).map((entry) => ({ ...entry, payload: {
       ...entry.payload, resume_when: entry.resume, resume_ready: false,
     } }))).sort((a, b) => Number(b.claim === agent) - Number(a.claim === agent));

loopx/control_plane/work_items/action_portfolio.py

@@ -125,6 +125,9 @@ def qualify_action_selection(
     should_run: bool,
     normal_delivery_allowed: bool,
     delivery_preemptions: list[str],
+    scoped_fallback_todo_id: str | None = None,
+    quota_state: str | None = None,
+    goal_health_ok: bool = False,
 ) -> dict[str, Any]:
     """Adapt current Python projections into the TS-owned selection reducer."""
 
@@ -140,6 +143,9 @@ def qualify_action_selection(
                 "should_run": should_run,
                 "normal_delivery_allowed": normal_delivery_allowed,
                 "delivery_preemptions": delivery_preemptions,
+                "scoped_fallback_todo_id": scoped_fallback_todo_id,
+                "quota_state": quota_state,
+                "goal_health_ok": goal_health_ok,
             },
         )
     except EffectRuntimeRejected as exc:
@@ -159,6 +165,9 @@ def qualify_action_selection_from_inventory(
     should_run: bool,
     normal_delivery_allowed: bool,
     delivery_preemptions: list[str],
+    scoped_fallback_todo_id: str | None = None,
+    quota_state: str | None = None,
+    goal_health_ok: bool = False,
 ) -> dict[str, Any]:
     """Resolve the requested task class before invoking the typed reducer."""
 
@@ -186,6 +195,9 @@ def qualify_action_selection_from_inventory(
         should_run=should_run,
         normal_delivery_allowed=normal_delivery_allowed,
         delivery_preemptions=delivery_preemptions,
+        scoped_fallback_todo_id=scoped_fallback_todo_id,
+        quota_state=quota_state,
+        goal_health_ok=goal_health_ok,
     )