fix(quota): bound blocked-frontier Vision baseline initialization

loopx/control_plane/effect_runtime_handlers.ts

@@ -385,7 +385,7 @@ export function createEffectRuntimeHandlers(
       lazyHandler(() => import("./quota/unsettled_host_turn_recovery.ts"), ({reduceUnsettledHostTurnRecovery}) => reduceUnsettledHostTurnRecovery),
     ],
     ["quota.turn_envelope.evaluate", lazyHandler(() => import("./quota/turn_envelope.ts"), ({evaluateTurnEnvelope}) => evaluateTurnEnvelope)],
-    ["quota.run_decision", lazyHandler(() => import("./quota/run_decision.ts"), ({resolveQuotaRunDecision}) => resolveQuotaRunDecision)],
+    ["quota.run_decision", lazyHandler(() => import("./quota/run_decision.ts"), ({admitQuotaRunDecision}) => admitQuotaRunDecision)],
     ["quota.scoped_override.project", lazyHandler(() => import("./quota/scoped_override.ts"), ({projectScopedOverride}) => projectScopedOverride)],
     ["task_lease.acquire.native", lazyHandler(() => import("./work_items/task_lease_acquire.ts"), ({executeTaskLeaseAcquire}) => executeTaskLeaseAcquire)],
     ["task_lease.inspect.native", lazyHandler(() => import("./work_items/task_lease_inspection.ts"), ({inspectTaskLease}) => inspectTaskLease)],

loopx/control_plane/quota/decision_summary.py

@@ -151,6 +151,7 @@ class QuotaRunDecision:
     quota: dict[str, Any]
     replan_decision_allowed: bool
     goal_route_hint_override: dict[str, Any] | None = None
+    agent_scope_frontier: dict[str, Any] | None = None
 
 
 def refine_quota_recommended_action(
@@ -263,6 +264,7 @@ def resolve_quota_run_decision(
     registered_agent_ids: list[str],
     goal_frontier_projection: dict[str, Any] | None,
     task_orchestration_contract: dict[str, Any] | None,
+    vision_initialization_context: dict[str, Any] | None = None,
 ) -> QuotaRunDecision:
     """Apply the ordered final guards for one quota run decision."""
 
@@ -294,6 +296,7 @@ def resolve_quota_run_decision(
         "automation_prompt_upgrade_required": automation_prompt_upgrade_required,
         "replan_required": replan_required,
         "replan_applies": bool(scope.get("applies")),
+        "vision_initialization_context": vision_initialization_context,
         "goal_health_ok": goal_health_ok,
         "inbox_reply_due": inbox_reply_due,
         "inbox_material_review_due": inbox_material_review_due,

loopx/control_plane/quota/run_decision.ts

@@ -2,6 +2,7 @@ import type {JsonObject} from "../effect_program.ts";
 import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
 import {requireBoolean, requireJsonObject} from "../runtime_decode.ts";
 import {EffectiveAction} from "./effective_action.generated.ts";
+import {boundVisionBaselineAdmission} from "./vision_baseline_admission.ts";
 
 interface RunFacts {
   normal_delivery_allowed: boolean;
@@ -39,6 +40,7 @@ interface RunDecision extends JsonObject {
   quota_patch: JsonObject;
   replan_decision_allowed: boolean;
   goal_route_hint_override: JsonObject | null;
+  agent_scope_frontier: JsonObject | null;
 }
 
 function text(value: unknown, label: string): string {
@@ -214,6 +216,7 @@ export function resolveQuotaRunDecision(params: JsonObject): JsonObject {
     should_run: false, effective_action: EffectiveAction.QUOTA_SKIP,
     reason: facts.reason, state: facts.state, quota_patch: {}, replan_decision_allowed: false,
     goal_route_hint_override: null,
+    agent_scope_frontier: null,
   };
   applyDeliveryGuards(facts, decision);
   decision.effective_action = quotaEffectiveAction(facts, decision);
@@ -224,3 +227,7 @@ export function resolveQuotaRunDecision(params: JsonObject): JsonObject {
     facts.task_orchestration_contract, facts.task_orchestration_lane_count);
   return decision;
 }
+
+export async function admitQuotaRunDecision(params: JsonObject): Promise<JsonObject> {
+  return boundVisionBaselineAdmission(params, resolveQuotaRunDecision(params));
+}

loopx/control_plane/quota/should_run.py

@@ -670,6 +670,16 @@ def build_quota_should_run(
             receipt_bound_guard_turn_instance_id=receipt_bound_guard_turn_instance_id,
             receipt_bound_replan_review=receipt_bound_replan_review,
         )
+        if runtime_root is not None and turn_instance_id and prepared.agent_frontier_id:
+            prepared.vision_initialization_context = {
+                "runtime_root": str(runtime_root), "goal_id": safe_goal_id,
+                "agent_id": prepared.agent_frontier_id, "turn_instance_id": turn_instance_id,
+                "receipt_bound": bool(receipt_bound_todo_id or receipt_bound_replan_obligation_id),
+                "todo_summary": prepared.agent_todo_summary,
+                "todo_items": prepared.agent_todo_planning_source_items,
+                "replan_obligation": prepared.replan_obligation,
+                "acceptance_gaps": prepared.goal_frontier_projection.get("acceptance_gaps"),
+            }
         route = _resolve_quota_should_run_route(prepared)
         route = _apply_selected_todo_guards(
             prepared,

loopx/control_plane/quota/should_run_packet.py

@@ -935,6 +935,7 @@ def _resolve_quota_should_run_route(
         registered_agent_ids=prepared.registered_agent_ids,
         goal_frontier_projection=prepared.goal_frontier_projection,
         task_orchestration_contract=prepared.task_orchestration_contract,
+        vision_initialization_context=prepared.vision_initialization_context,
     )
     if prepared.receipt_bound_replay_phase is ReceiptBoundReplayPhase.SETTLED:
         fields = settled_replay_fields()
@@ -1179,7 +1180,8 @@ def _resolve_quota_should_run_route(
             agent_lane_next_action=agent_lane_next_action,
         )
         agent_scope_frontier = (
-            _selected_deferred_replan_frontier(prepared)
+            run_decision.agent_scope_frontier
+            or _selected_deferred_replan_frontier(prepared)
             or _agent_scope_no_candidate_frontier(
                 agent_identity=prepared.agent_identity,
                 agent_todo_summary=prepared.agent_todo_summary,

loopx/control_plane/quota/should_run_prepare.py

@@ -185,6 +185,7 @@ class _QuotaDecisionPreparation:
     resolved_scheduler_context: SchedulerExecutionContextResolution
     delivery_continuity_anchor: dict[str, Any] | None
     delivery_continuity_todo: dict[str, Any] | None
+    vision_initialization_context: dict[str, Any] | None = None
 
     @property
     def inbox_priority_due(self) -> bool:

loopx/control_plane/quota/vision_baseline_admission.ts

@@ -0,0 +1,85 @@
+import {createHash} from "node:crypto";
+import {readFile} from "node:fs/promises";
+import {join} from "node:path";
+import type {JsonObject} from "../effect_program.ts";
+import {durableWriteJson, withFileMutationLock} from "../effect_runtime_io.ts";
+import {jsonObject, requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts";
+import {AgentScopeFrontierAction} from "../agents/agent_scope_frontier.generated.ts";
+import {readAgent, readExcludedAgents} from "../todos/read_codec.ts";
+
+const BASELINE_ADMISSION_SCHEMA = "vision_baseline_admission_v0";
+
+interface VisionBaselineAdmission extends JsonObject {
+  schema_version: typeof BASELINE_ADMISSION_SCHEMA;
+  goal_id: string;
+  agent_id: string;
+  turn_instance_id: string;
+  replan_obligation_id: string | null;
+}
+
+/** One initialization Turn per runtime/Goal/Agent; missing checkpoints cannot rearm it. */
+export async function boundVisionBaselineAdmission(params: JsonObject, decision: JsonObject): Promise<JsonObject> {
+  const context = params.vision_initialization_context == null ? null
+    : requireJsonObject(params.vision_initialization_context, "vision_initialization_context");
+  if (!context || decision.replan_decision_allowed !== true || context.receipt_bound === true ||
+      decision.self_repair_allowed === true || decision.capability_repair_allowed === true ||
+      decision.workspace_repair_allowed === true) return decision;
+  const agent = requireNonEmptyString(context.agent_id, "agent_id");
+  const summary = requireJsonObject(context.todo_summary, "todo_summary");
+  const counts = jsonObject(summary.work_counts);
+  // Unknown or incomplete projections cannot prove that all real work is blocked.
+  if (counts?.complete !== true || counts.agent_id !== agent || counts.advancement !== 0 ||
+      summary.monitor_due_count !== 0 || Number(summary.monitor_schedule_gap_count ?? 0) > 0) return decision;
+  const rows = context.todo_items;
+  if (!Array.isArray(rows)) throw new TypeError("todo_items must be an array");
+  const blocked = rows.map(row => requireJsonObject(row, "Todo")).filter(row =>
+    row.status === "blocked" && row.done !== true && row.resume_ready !== true &&
+    (row.task_class === "advancement_task" || row.task_class === "blocker") &&
+    (!readAgent(row.claimed_by) || readAgent(row.claimed_by) === agent) &&
+    !readExcludedAgents(row.excluded_agents).includes(agent));
+  if (!blocked.length) return decision;
+  const obligation = requireJsonObject(context.replan_obligation, "replan_obligation");
+  const triggers = obligation.triggers;
+  const gaps = context.acceptance_gaps;
+  if (!Array.isArray(triggers) || !triggers.length || !Array.isArray(gaps) || !gaps.length ||
+      !triggers.every(raw => jsonObject(raw)?.kind === "vision_checkpoint_missing") ||
+      !gaps.every(raw => {
+        const gap = jsonObject(raw);
+        return gap?.kind === "vision_checkpoint_missing" && gap.missing_baseline === true && gap.agent_id === agent;
+      })) return decision;
+  const root = requireNonEmptyString(context.runtime_root, "runtime_root");
+  const scope = {goal_id: requireNonEmptyString(context.goal_id, "goal_id"), agent_id: agent};
+  const turn = requireNonEmptyString(context.turn_instance_id, "turn_instance_id");
+  const path = join(root, "vision-baseline-admissions",
+    `${createHash("sha256").update(JSON.stringify(scope)).digest("hex")}.json`);
+  return withFileMutationLock(path, async () => {
+    let previous: JsonObject | null = null;
+    try { previous = requireJsonObject(JSON.parse(await readFile(path, "utf8")), "baseline admission"); }
+    catch (error) {
+      if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
+    }
+    if (previous && (previous.schema_version !== BASELINE_ADMISSION_SCHEMA ||
+        previous.goal_id !== scope.goal_id || previous.agent_id !== agent)) {
+      throw new TypeError("Vision baseline admission scope/schema mismatch");
+    }
+    if (!previous) {
+      const admission: VisionBaselineAdmission = {schema_version: BASELINE_ADMISSION_SCHEMA, ...scope,
+        turn_instance_id: turn, replan_obligation_id: requireNonEmptyString(obligation.obligation_id, "obligation_id")};
+      await durableWriteJson(path, admission);
+      return decision;
+    }
+    // Reentry of the admitted Turn must retain its original work and settlement identity.
+    if (requireNonEmptyString(previous.turn_instance_id, "admitted Turn") === turn) return decision;
+    const reason = "Vision baseline initialization was already admitted; the scoped frontier remains blocked";
+    const frontier = {schema_version: "agent_scope_frontier_v1", agent_id: agent,
+      action: AgentScopeFrontierAction.AGENT_SCOPE_WAIT, blocks_delivery: true,
+      quiet_noop_allowed: true, spend_policy: "no quota spend while waiting for runnable work or an owner change",
+      reason, recommended_action: "Wait for executable work or an explicit owner/replan change; retain the pending Vision baseline duty.",
+      candidate_counts: {current_agent_claimed_advancement_count: 0, unclaimed_advancement_count: 0,
+        blocked_wait_count: blocked.length},
+      vision_baseline_admission: previous};
+    return {...decision, normal_delivery_allowed: false, recovery_delivery_allowed: false,
+      should_run: false, replan_decision_allowed: false,
+      effective_action: AgentScopeFrontierAction.AGENT_SCOPE_WAIT, reason, agent_scope_frontier: frontier};
+  });
+}

tests/control_plane/test_missing_vision_baseline_wait.py

@@ -0,0 +1,120 @@
+"""A retained missing baseline cannot repeatedly wake a blocked-only Agent."""
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+
+import pytest
+
+from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+from loopx.control_plane.testing.quota_fixtures import quota_status_payload, quota_todo_item
+from loopx.control_plane.scheduler.execution_context import GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT
+from loopx.quota import build_quota_should_run
+
+GOAL = "baseline-wait-fixture"
+AGENT = "agent-baseline"
+
+
+def status(*, goal=GOAL, agent=AGENT, extra=()):
+    checkpoint = {"schema_version": "vision_checkpoint_v0", "agent_id": agent,
+                  "required": True, "satisfied": False, "decision": "missing_required",
+                  "missing_baseline": True,
+                  "triggers": [{"kind": "material_delivery_outcome", "delivery_outcome": "outcome_progress"}]}
+    return quota_status_payload(
+        goal_id=goal, status="active", recommended_action="Wait for external input.",
+        agent_todo_items=[quota_todo_item(todo_id="todo_blocked", title="External input pending.",
+                                        status="blocked", claimed_by=agent), *extra],
+        latest_runs=[{"classification": "state_refreshed", "generated_at": "2026-01-01T00:00:00Z",
+                      "agent_id": agent, "progress_scope": "agent_lane", "vision_checkpoint": checkpoint}],
+        coordination={"agent_model": "peer_v1", "registered_agents": [agent]},
+        quota_extra={"allowed_slots": 12, "spent_slots": 1})
+
+
+def guard(payload, runtime, turn, *, goal=GOAL, agent=AGENT, **kwargs):
+    return build_quota_should_run(
+        payload, goal_id=goal, agent_id=agent, runtime_root=runtime, turn_instance_id=turn,
+        scheduler_execution_context=GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT,
+        include_scheduler_detail=True, **kwargs)
+
+
+def assert_wait(result):
+    assert result["effective_action"] == "agent_scope_wait"
+    assert result["should_run"] is False
+    assert result["normal_delivery_allowed"] is False
+    assert result["execution_obligation"]["must_attempt_work"] is False
+    channel = result["interaction_contract"]["agent_channel"]
+    assert channel["delivery_allowed"] is False
+    assert channel["quiet_noop_allowed"] is True
+    assert channel["must_attempt"] is False
+    assert result["scheduler_hint"]["action"] == "backoff_until_reassigned"
+    assert result["autonomous_replan_obligation"]["required"] is True
+
+
+def test_first_initialization_then_fresh_turns_and_process_readback_wait(tmp_path):
+    payload = status()
+    first = guard(payload, tmp_path, "initialization")
+    assert first["effective_action"] == "autonomous_replan_required"
+    assert first["should_run"] is True
+    assert guard(payload, tmp_path, "initialization")["should_run"] is True
+    for turn in ("poll-1", "poll-2"):
+        assert_wait(guard(payload, tmp_path, turn))
+    script = (
+        "import json,sys; from pathlib import Path; "
+        "from tests.control_plane.test_missing_vision_baseline_wait import guard,assert_wait; "
+        "result=guard(json.load(sys.stdin),Path(sys.argv[1]),'process-readback'); "
+        "assert_wait(result); print(result['effective_action'])"
+    )
+    completed = subprocess.run([sys.executable, "-c", script, str(tmp_path)], input=json.dumps(payload),
+                               text=True, capture_output=True, check=True,
+                               env={**os.environ, "LOOPX_USAGE_PING": "0"})
+    assert completed.stdout.strip() == "agent_scope_wait"
+    assert len(list((tmp_path / "vision-baseline-admissions").glob("*.json"))) == 1
+
+
+@pytest.mark.parametrize("task_class", ["advancement_task", "continuous_monitor"])
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_real_work_and_due_monitor_resume_after_wait(tmp_path, task_class, provider):
+    from canonical_authority_fixture import initialize_canonical_authority
+
+    guard(status(), tmp_path, "initialization")
+    assert_wait(guard(status(), tmp_path, "blocked"))
+    work = quota_todo_item(todo_id="todo_independent", title="Perform independent work.",
+                               task_class=task_class, claimed_by=AGENT, archive_state="active",
+                           **({"cadence": "30m", "next_due_at": "2000-01-01T00:00:00Z"}
+                              if task_class == "continuous_monitor" else {}))
+    state = tmp_path / "state.md"
+    state.write_text("# Synthetic fixture\n\n## Agent Todo\n", encoding="utf-8")
+    initialize_canonical_authority(
+        tmp_path, GOAL, build_todo_runtime_shadow_projection(
+            goal_id=GOAL, todos=[work], handoff_mode="soft_claim"),
+        state_path=state, provider=provider)
+    result = guard(status(extra=[work]), tmp_path, "ready")
+    assert result["should_run"] is True
+    assert result["effective_action"] != "agent_scope_wait"
+    assert_wait(guard(status(), tmp_path, "blocked-again"))
+
+
+def test_admission_is_scoped_to_goal_and_agent(tmp_path):
+    guard(status(), tmp_path, "initialization")
+    for goal, agent in (("another-goal", AGENT), (GOAL, "another-agent")):
+        result = guard(status(goal=goal, agent=agent), tmp_path, "initialization", goal=goal, agent=agent)
+        assert result["effective_action"] == "autonomous_replan_required"
+    assert len(list((tmp_path / "vision-baseline-admissions").glob("*.json"))) == 3
+
+
+def test_receipt_bound_replan_retains_original_settlement(tmp_path):
+    first = guard(status(), tmp_path, "initialization")
+    obligation = first["autonomous_replan_obligation"]["obligation_id"]
+    result = guard(status(), tmp_path, "original-owed-turn",
+                   receipt_bound_replan_obligation_id=obligation)
+    assert result["should_run"] is True
+    assert result["autonomous_replan_obligation"]["obligation_id"] == obligation
+
+
+def test_changed_checkpoint_timestamp_does_not_rearm_baseline(tmp_path):
+    guard(status(), tmp_path, "initialization")
+    updated = status()
+    updated["run_history"]["goals"][0]["latest_runs"][0]["generated_at"] = "2026-02-01T00:00:00Z"
+    assert_wait(guard(updated, tmp_path, "changed-checkpoint"))

tests/control_plane_ts/fixtures/quota_run_decision.json

@@ -33,7 +33,8 @@
     "state": "eligible",
     "quota_patch": {},
     "replan_decision_allowed": false,
-    "goal_route_hint_override": null
+    "goal_route_hint_override": null,
+    "agent_scope_frontier": null
   },
   "cases": [
     {