fix(replan): derive causal wait coverage before Todo compaction

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

@@ -404,10 +404,11 @@ goal.
 
 ### Exact blocked-successor wait
 
-An open agent vision does not need another replan when the lane already has an
-exact current-agent or unclaimed advancement successor whose supported
-`resume_when` condition is projected as `resume_ready=false`. When there is no
-other selectable advancement, quota/status expose
+An open agent vision can wait when every causal Todo binding of its ordinary
+acceptance gap has a related current-agent or unclaimed advancement successor
+whose supported `resume_when` condition is projected as `resume_ready=false`,
+or an exact current-agent blocker with a reason. One route's wait cannot cover
+another uncovered binding. When there is no other selectable advancement, quota/status expose
 `goal_vision_wait_state_v0` with the waiting todo id, `resume_when`, compact
 `resume_condition`, and `automatic_resume=true`. The ordinary
 `vision_acceptance_gap` is deferred while that read model is active, so the
@@ -420,6 +421,38 @@ It cannot suppress `vision_checkpoint_missing`, `vision_successor_required`, a
 resume condition that lacks exact projected evidence, or the dedicated repair
 for an advancement todo incorrectly gated by a standing continuous monitor.
 
+The normal `refresh-state --vision-todo-delta <action>:<todo_id>` and Turn vision
+write paths supply the causal bindings. Todo create/update/complete/supersede
+and resume evaluation supply their current facts. A planned `create/reopen`
+entry is a binding to inspect, not proof that its Todo exists. An agent changes
+the active bindings through the existing vision writeback contract when the
+plan changes; finishing a Todo alone does not prove its acceptance is closed.
+Explicit successor lineage can connect a completed or archived predecessor to
+a real waiting successor. Sharing a prerequisite does not make two sibling
+Todos interchangeable, and a terminal vision keeps its existing lifecycle rules.
+
+Wait witnesses are derived from canonical Todo rows and evaluated conditions
+before display compaction. `agent_todos.vision_wait_states` carries only those
+positive, agent-scoped results, bound to `causal_todo_ids`; each source read
+rebuilds them for the latest vision. It is not a stored or separately authored
+state. Quota and semantic writeback use the same coverage reducer. Display
+limits remain unchanged: extra unrelated Todos and reordering cannot change
+the wait decision. If a legacy/incomplete source cannot prove coverage, the
+existing acceptance gap stays open; missing display rows do not prove that
+canonical work is absent or that all alternatives are exhausted.
+
+This tightens the previous any-related-wait behavior. With bindings to A and B,
+A waiting and B unmaterialized requires replan when execution gates permit it;
+a runnable B continues, and related valid waits for both preserve defer. The
+rule uses existing acceptance/lineage facts regardless of the optional advisory
+`fallback_declarations`. It neither discovers alternatives nor invents AND/OR
+relationships, and it grants no additional authority. Ownership, exclusions,
+capabilities, user gates, and quota remain independent execution constraints.
+
+等待资格现在逐项检查已有 acceptance 的 Todo 关联，并在展示裁剪前从完整来源计算。
+A 的等待不能遮住尚未落实的 B；有可执行工作则继续，相关工作都具有合法等待证据才暂缓。
+无需另外维护 fallback 声明；无关 Todo 的数量和顺序不应改变决策。
+
 ## Replan Triggers
 
 A replan trigger is goal-level and should be evaluated before lane-local quiet

loopx/control_plane/effect_runtime_handlers.ts

@@ -78,6 +78,7 @@ import {
   writeSchedulerState,
 } from "./scheduler/state_store.ts";
 import { buildVisionCheckpoint } from "./goals/vision_checkpoint.ts";
+import { projectVisionWaitCoverage } from "./goals/vision_wait_coverage.ts";
 import { admitGoalAmendmentProposal } from "./goals/goal_amendment_proposal.ts";
 import { projectSharedGoalAlignment } from "./goals/shared_goal_alignment.ts";
 import {
@@ -392,6 +393,7 @@ export function createEffectRuntimeHandlers(
     ["work_item.planning_inventory.detail", projectTodoPlanningInventoryDetail],
     ["work_item.refresh_recommendation.resolve", resolveRefreshRecommendation],
     ["goal.vision_checkpoint.evaluate", buildVisionCheckpoint],
+    ["goal.vision_wait.coverage", projectVisionWaitCoverage],
     ["goal.shared_goal_alignment.project", projectSharedGoalAlignment],
     ["goal.amendment_proposal.admit", admitGoalAmendmentProposal],
     ["agent.delivery_workspace.evaluate", evaluateDeliveryWorkspace],

loopx/control_plane/goals/goal_frontier/__init__.py

@@ -39,7 +39,13 @@
 )
 from ..goal_vision_state import (
     goal_vision_state_is_closed,
-    goal_vision_state_requires_successor,
+)
+from ..goal_vision_read_model import (
+    VISION_ACCEPTANCE_GAP_TRIGGER as VISION_ACCEPTANCE_GAP_TRIGGER,
+    VISION_SUCCESSOR_GAP_TRIGGER as VISION_SUCCESSOR_GAP_TRIGGER,
+    _compact_projection_text,
+    acceptance_gaps_from_agent_vision as acceptance_gaps_from_agent_vision,
+    parse_vision_todo_delta_entries as parse_vision_todo_delta_entries,
 )
 from ..goal_vision_wait import build_goal_vision_wait_state
 from . import outcome_continuity
@@ -53,7 +59,6 @@
     agent_scoped_selectable_advancement_todo_ids,  # noqa: F401
     declared_fallback_gap_from_agent_vision,
     parse_fallback_declarations,  # noqa: F401
-    parse_vision_todo_delta_entries,
 )
 from .long_todo_chain import (
     LONG_TODO_CHAIN_TRIGGER,
@@ -93,8 +98,6 @@
 AUTONOMOUS_REPLAN_REQUIRED_MODE = "autonomous_replan_required"
 FRONTIER_EXHAUSTED_MONITOR_TRIGGER = "frontier_exhausted_monitor_lane"
 MONITOR_NO_CHANGE_STREAK_TRIGGER = "monitor_no_change_streak"
-VISION_ACCEPTANCE_GAP_TRIGGER = "vision_acceptance_gap"
-VISION_SUCCESSOR_GAP_TRIGGER = "vision_successor_required"
 VISION_PROFILE_MISSING_TRIGGER = "required_agent_vision_missing"
 TODO_SUCCESSION_GAP_TRIGGER = TODO_SUCCESSION_WARNING_REASON_CODE
 TODO_TASK_CLASS_ADVANCEMENT = "advancement_task"
@@ -305,13 +308,6 @@ def autonomous_replan_scope_decision(
     return payload
 
 
-def _compact_projection_text(value: Any, *, limit: int = 360) -> str | None:
-    text = " ".join(str(value or "").strip().split())
-    if not text:
-        return None
-    return text[:limit]
-
-
 def projected_autonomous_replan_ack_for_agent(
     item: dict[str, Any],
     project_asset: dict[str, Any] | None,
@@ -331,87 +327,6 @@ def projected_autonomous_replan_ack_for_agent(
     return None
 
 
-def acceptance_gaps_from_agent_vision(
-    agent_vision: dict[str, Any] | None,
-    *,
-    goal_status: str | None = None,
-) -> list[dict[str, Any]]:
-    """Convert bounded vision replan triggers into goal-frontier gap records."""
-
-    if not isinstance(agent_vision, dict):
-        return []
-    patch = agent_vision.get("vision_patch") if isinstance(agent_vision.get("vision_patch"), dict) else {}
-    state = str(agent_vision.get("state") or "").strip()
-    if goal_vision_state_is_closed(state):
-        normalized_goal_status = str(goal_status or "").strip().lower()
-        active_goal = normalized_goal_status == "active" or normalized_goal_status.startswith(
-            "active-"
-        )
-        if goal_vision_state_requires_successor(state) and active_goal:
-            return [
-                {
-                    "kind": VISION_SUCCESSOR_GAP_TRIGGER,
-                    "source": "latest_agent_vision",
-                    "agent_id": agent_vision.get("agent_id"),
-                    "state": agent_vision.get("state"),
-                    "goal_status": normalized_goal_status,
-                    "replan_trigger_summary": (
-                        "the current stage vision is closed while the registry goal "
-                        "remains active; establish a successor vision before continuing"
-                    ),
-                    "acceptance_summary": (
-                        "Write the next bounded agent vision, or explicitly retire, "
-                        "supersede, or close the lane with no_followup."
-                    ),
-                    "advancement_policy": "repeat_until_closed",
-                    "generated_at": agent_vision.get("generated_at"),
-                }
-            ]
-        return []
-    acceptance = _compact_projection_text(patch.get("acceptance_summary"), limit=420)
-    explicit_trigger = _compact_projection_text(
-        patch.get("replan_trigger_summary"),
-        limit=240,
-    )
-    trigger = explicit_trigger
-    if not trigger and acceptance:
-        trigger = "active agent vision remains open with acceptance evidence still required"
-    if not trigger:
-        return []
-    gap: dict[str, Any] = {
-        "kind": VISION_ACCEPTANCE_GAP_TRIGGER,
-        "source": "latest_agent_vision",
-        "agent_id": agent_vision.get("agent_id"),
-        "state": agent_vision.get("state"),
-        "replan_trigger_summary": trigger,
-        "replan_trigger_source": (
-            "explicit_vision_trigger"
-            if explicit_trigger
-            else "implicit_open_acceptance"
-        ),
-    }
-    if acceptance:
-        gap["acceptance_summary"] = acceptance
-    vision_todo_ids = [
-        todo_id
-        for _, todo_id in parse_vision_todo_delta_entries(
-            agent_vision.get("todo_delta")
-        )
-    ]
-    if vision_todo_ids:
-        gap["vision_todo_ids"] = list(dict.fromkeys(vision_todo_ids))
-    advancement_policy = _compact_projection_text(
-        patch.get("advancement_policy"),
-        limit=32,
-    )
-    if advancement_policy:
-        gap["advancement_policy"] = advancement_policy
-    generated_at = _compact_projection_text(agent_vision.get("generated_at"), limit=80)
-    if generated_at:
-        gap["generated_at"] = generated_at
-    return [gap]
-
-
 def acceptance_gaps_from_agent_profile_requirement(
     agent_profile: dict[str, Any] | None,
     *,
@@ -1046,6 +961,14 @@ def derive_goal_frontier_replan_obligation_from_summaries(
     compact_acceptance_gaps = [
         item for item in (acceptance_gaps or []) if isinstance(item, dict)
     ]
+    if any(gap.get("vision_todo_ids") for gap in compact_acceptance_gaps):
+        # Diagnostic claim counts retain executor-excluded work. A causal
+        # acceptance obligation needs an actually selectable Todo identity.
+        selectable_frontier_advancement = len(
+            agent_scoped_selectable_advancement_todo_ids(
+                agent_todo_summary, agent_id=agent_id,
+            )
+        )
     successor_vision_required = any(
         item.get("kind")
         in {VISION_SUCCESSOR_GAP_TRIGGER, VISION_PROFILE_MISSING_TRIGGER}
@@ -1597,6 +1520,7 @@ def build_goal_frontier_projection_context_from_status(
     )
     vision_wait_state = build_goal_vision_wait_state(
         agent_todo_summary=agent_todo_summary,
+        source_items=agent_todo_source_items,
         agent_id=agent_id,
         acceptance_gaps=source_acceptance_gaps,
         selectable_advancement_count=(

loopx/control_plane/goals/goal_frontier/fallback_disposition.py

@@ -10,21 +10,20 @@
 from ...todos.projection import (
     agent_scoped_selectable_advancement_todo_ids,
 )
+from ..goal_vision_read_model import (
+    VISION_FRONTIER_TODO_DELTA_ACTIONS as VISION_FRONTIER_TODO_DELTA_ACTIONS,
+    VISION_TODO_DELTA_ID_LIMIT as VISION_TODO_DELTA_ID_LIMIT,
+    parse_vision_todo_delta_entries as parse_vision_todo_delta_entries,
+)
 from ..goal_vision_state import goal_vision_state_is_closed
 
-# Single owner of the vision todo_delta action contract shared by the
-# acceptance-gap projection and this module.
-VISION_FRONTIER_TODO_DELTA_ACTIONS = frozenset(
-    {"activate", "create", "reopen", "resume", "retain"}
-)
 # create/reopen entries are bounded successor declarations and resolve the
 # fallback disposition on their own; activate/resume/retain entries only link
 # the vision to existing Todos and still need a selectable frontier match.
 VISION_TODO_DELTA_SUCCESSOR_ACTIONS = frozenset({"create", "reopen"})
 VISION_TODO_DELTA_LINKAGE_ACTIONS = frozenset(
     VISION_FRONTIER_TODO_DELTA_ACTIONS - VISION_TODO_DELTA_SUCCESSOR_ACTIONS
 )
-VISION_TODO_DELTA_ID_LIMIT = 120
 VISION_FALLBACK_DECLARATION_ENTRY_LIMIT = 4
 VISION_FALLBACK_DECLARATION_FIELDS = ("target_todo_id", "successor_todo_id")
 VISION_FALLBACK_GAP_TRIGGER = "vision_fallback_unresolved"
@@ -68,25 +67,6 @@ def _compact_text(value: Any, *, limit: int) -> str:
     return " ".join(str(value or "").strip().split())[:limit]
 
 
-def parse_vision_todo_delta_entries(entries: Any) -> list[tuple[str, str]]:
-    """Parse ``action:todo_id`` vision todo_delta entries once for consumers."""
-
-    parsed: list[tuple[str, str]] = []
-    for value in entries or []:
-        if not isinstance(value, str):
-            continue
-        action, separator, raw_todo_id = value.strip().partition(":")
-        todo_id = _compact_text(raw_todo_id, limit=VISION_TODO_DELTA_ID_LIMIT)
-        normalized_action = action.strip().lower()
-        if (
-            separator
-            and todo_id
-            and normalized_action in (VISION_FRONTIER_TODO_DELTA_ACTIONS)
-        ):
-            parsed.append((normalized_action, todo_id))
-    return parsed
-
-
 def parse_fallback_declarations(
     agent_vision: dict[str, Any] | None,
 ) -> list[FallbackDeclaration]:

loopx/control_plane/goals/goal_frontier/semantic_history.py

@@ -2,6 +2,8 @@
 
 from typing import Any
 
+from ..goal_vision_read_model import latest_agent_vision_from_runs as latest_agent_vision_from_runs
+
 from ...work_items.autonomous_replan_ack import (
     latest_autonomous_replan_ack_for_projection,
 )
@@ -127,52 +129,6 @@ def latest_agent_vision_from_status_payload(
     )
 
 
-def latest_agent_vision_from_runs(
-    runs: list[dict[str, Any]],
-    *,
-    goal_id: str,
-    agent_id: str | None,
-) -> dict[str, Any] | None:
-    """Return the newest active vision from newest-first compact run records."""
-
-    for run in runs:
-        vision = run.get("agent_vision")
-        if not isinstance(vision, dict):
-            continue
-        vision_agent_id = str(
-            vision.get("agent_id") or run.get("agent_id") or ""
-        ).strip()
-        if agent_id and vision_agent_id and vision_agent_id != agent_id:
-            continue
-        patch = (
-            vision.get("vision_patch")
-            if isinstance(vision.get("vision_patch"), dict)
-            else {}
-        )
-        if not patch:
-            continue
-        result = {
-            "schema_version": vision.get("schema_version"),
-            "goal_id": goal_id,
-            "agent_id": vision_agent_id or agent_id,
-            "state": vision.get("state"),
-            "vision_patch": patch,
-            "todo_delta": vision.get("todo_delta")
-            if isinstance(vision.get("todo_delta"), list)
-            else [],
-            "vision_budget": vision.get("vision_budget")
-            if isinstance(vision.get("vision_budget"), dict)
-            else None,
-            "generated_at": run.get("generated_at"),
-        }
-        if isinstance(vision.get("path_delta"), dict):
-            result["path_delta"] = vision["path_delta"]
-        if isinstance(vision.get("fallback_declarations"), list):
-            result["fallback_declarations"] = vision["fallback_declarations"]
-        return result
-    return None
-
-
 def _latest_missing_vision_checkpoint_from_runs(
     runs: list[dict[str, Any]],
     *,