refactor(goals): converge Vision wait projection in TypeScript

Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com>

Author: hhyykk

docs/architecture/rfcs/typescript-control-plane-migration-v0.md

@@ -191,6 +191,10 @@ Python retains registry, document/material IO, independent Vision-wait consumers
 and compatibility exports. Its replaced summary, filter and result coordination
 is retired. Persisted completion spellings use shared token data and compatibility codecs
 without scalar RPCs;
+Vision-wait state projection now composes typed resume planning, current-agent
+blocker eligibility, causal lineage coverage and result assembly in one TypeScript
+effect. Python retains source-snapshot encoding, result validation, historical
+wait-state decoding and independent consumers.
 `todo.summary.project`, `todo.completion_state.normalize` and
 `todo.completion_state.require_metadata` are retired after their production
 callers exit. Existing semantic owners execute in process inside the transaction.

docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md

@@ -155,7 +155,10 @@ Host fence 和真实 provider IO。有界扫描只提供读取提示，写入锁
 依赖与闭合判断仍使用完整源。
 Python 保留 registry、文档／材料 IO、独立的 Vision-wait 消费者及兼容导出。
 被替代的摘要、筛选与结果协调已退役。持久化 completion 拼写使用共享 token 数据与兼容 codec，
-不再逐标量调用 RPC；实际生产调用者退出后，退役 `todo.summary.project`、
+不再逐标量调用 RPC；
+Vision-wait 状态投影现在由一次 TypeScript effect 组合类型化 resume planning、当前 Agent blocker
+资格、因果血缘覆盖和结果组装。Python 保留源快照编码、结果校验、历史等待态解码及独立消费者。
+实际生产调用者退出后，退役 `todo.summary.project`、
 `todo.completion_state.normalize` 与 `todo.completion_state.require_metadata`。
 事务在进程内执行既有语义 owner。
 

loopx/control_plane/effect_runtime_handlers.ts

@@ -366,6 +366,7 @@ export function createEffectRuntimeHandlers(
     ["goal.checkpoint_read_context.commit", lazyHandler(() => import("./goals/checkpoint_commit.ts"), ({commitCheckpoint}) => commitCheckpoint)],
     ["goal.checkpoint_read_context.inspect_replay", lazyHandler(() => import("./goals/checkpoint_commit.ts"), ({inspectCheckpointReplay}) => inspectCheckpointReplay)],
     ["goal.vision_wait.coverage", lazyHandler(() => import("./goals/vision_wait_coverage.ts"), ({projectVisionWaitCoverage}) => projectVisionWaitCoverage)],
+    ["goal.vision_wait.project", lazyHandler(() => import("./goals/vision_wait_coverage.ts"), ({projectGoalVisionWaitState}) => projectGoalVisionWaitState)],
     ["goal.shared_goal_alignment.project", lazyHandler(() => import("./goals/shared_goal_alignment.ts"), ({projectSharedGoalAlignment}) => projectSharedGoalAlignment)],
     ["goal.operator_actions.project", lazyHandler(() => import("./goals/operator_actions.ts"), ({projectGoalOperatorActions}) => projectGoalOperatorActions)],
     ["goal.amendment.execute", lazyHandler(() => import("./goals/goal_amendment_authority.ts"), ({goalAmendmentLocal}) => goalAmendmentLocal)],

loopx/control_plane/goals/goal_vision_wait.py

@@ -5,17 +5,22 @@
 from typing import Any
 
 from ..effect_runtime import effect_runtime_result
-from ..todos.todo_semantics import todo_item_excludes_agent
-from ..todos.contract import (
-    TODO_TASK_CLASS_BLOCKER,
-    normalize_todo_claimed_by,
-    normalize_todo_id,
-    normalize_todo_status,
-)
-from ..todos.resume_planning import project_todo_resume_planning
+from ..todos.contract import normalize_todo_id
+from ..todos.resume_planning import build_todo_resume_planning_request
 
 GOAL_VISION_WAIT_STATE_SCHEMA_VERSION = "goal_vision_wait_state_v0"
-VISION_ACCEPTANCE_GAP_KIND = "vision_acceptance_gap"
+
+
+def _vision_wait_blocker_rows(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    fields = ("todo_id", "task_class", "status", "claimed_by", "excluded_agents", "reason")
+    return [{key: item[key] for key in fields if key in item} for item in items if isinstance(item, dict)]
+
+
+def _vision_wait_lineage_rows(items: list[dict[str, Any]] | None) -> list[dict[str, Any]] | None:
+    if items is None:
+        return None
+    fields = ("todo_id", "successor_todo_ids", "resume_condition")
+    return [{key: item[key] for key in fields if key in item} for item in items if isinstance(item, dict)]
 
 
 def exact_blocked_successor_wait_state(value: Any) -> dict[str, Any]:
@@ -57,174 +62,6 @@ def exact_blocked_successor_frontier_identity(value: Any) -> str | None:
     ).hexdigest()[:16]
 
 
-def _compact_resume_condition(value: Any) -> dict[str, Any]:
-    if not isinstance(value, dict):
-        return {}
-    compact = {
-        key: value.get(key)
-        for key in (
-            "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",
-        )
-        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.
-        compact["resume_receipt"] = value.get("resume_receipt")
-    compact["satisfied"] = False
-    return compact
-
-
-def _acceptance_gap_causal_todo_ids(
-    gaps: list[dict[str, Any]],
-) -> set[str]:
-    todo_ids: set[str] = set()
-    for gap in gaps:
-        for key in (
-            "todo_id",
-            "current_todo_id",
-            "blocked_todo_id",
-            "completed_todo_id",
-            "successor_todo_id",
-        ):
-            todo_id = normalize_todo_id(gap.get(key))
-            if todo_id:
-                todo_ids.add(todo_id)
-        for key in (
-            "vision_todo_ids",
-            "lineage_todo_ids",
-            "successor_todo_ids",
-            "completed_todo_ids",
-        ):
-            raw_values = gap.get(key)
-            values = raw_values if isinstance(raw_values, list) else []
-            todo_ids.update(
-                todo_id for value in values if (todo_id := normalize_todo_id(value))
-            )
-    return todo_ids
-
-
-def _wait_lineage_edges(items: list[dict[str, Any]]) -> list[list[str]]:
-    edges: set[tuple[str, str]] = set()
-    for item in items:
-        todo_id = normalize_todo_id(item.get("todo_id"))
-        if not todo_id:
-            continue
-        for value in item.get("successor_todo_ids") or []:
-            successor = normalize_todo_id(value)
-            if successor:
-                edges.add((todo_id, successor))
-        condition = item.get("resume_condition")
-        if isinstance(condition, dict) and condition.get("satisfied") is False:
-            target = normalize_todo_id(condition.get("target_todo_id"))
-            if target:
-                # A prerequisite can explain its waiting successor, never the
-                # reverse: shared prerequisites do not cover unrelated siblings.
-                edges.add((target, todo_id))
-    return [list(edge) for edge in sorted(edges)]
-
-
-def _covered_wait_items(
-    *,
-    agent_todo_summary: dict[str, Any] | None,
-    agent_id: str | None,
-    causal_todo_ids: set[str],
-    source_items: list[dict[str, Any]] | None,
-    lineage_source_items: list[dict[str, Any]] | None,
-) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
-    if source_items is not None:
-        # Source rows have already passed the canonical resume evaluator. Never
-        # rebuild them from the quota's bounded deferred/backlog display lanes.
-        agent_todo_summary = {
-            "items": source_items,
-            "deferred_items": [
-                i for i in source_items if i.get("status") == "deferred"
-            ],
-            "resume_blocked_items": [
-                i
-                for i in source_items
-                if i.get("resume_ready") is False and i.get("status") == "open"
-            ],
-            "current_agent_blocker_items": source_items,
-        }
-
-    raw_blockers = (agent_todo_summary or {}).get("current_agent_blocker_items")
-    blocker_items = raw_blockers if isinstance(raw_blockers, list) else []
-    safe_agent_id = normalize_todo_claimed_by(agent_id)
-    blocker_items = [
-        item
-        for item in blocker_items
-        if isinstance(item, dict)
-        and safe_agent_id is not None
-        and item.get("task_class") == TODO_TASK_CLASS_BLOCKER
-        and normalize_todo_status(item.get("status")) == "blocked"
-        and str(item.get("reason") or "").strip()
-        and normalize_todo_claimed_by(item.get("claimed_by")) == safe_agent_id
-        and not todo_item_excludes_agent(item, agent_id=safe_agent_id)
-    ]
-    candidates = project_todo_resume_planning(
-        agent_todo_summary or {}, agent_id=agent_id
-    )["blocked_successor_items"]
-    coverage = effect_runtime_result(
-        "goal.vision_wait.coverage",
-        {
-            "causal_todo_ids": sorted(causal_todo_ids),
-            "waiting_todo_ids": [i["todo_id"] for i in candidates if i.get("todo_id")],
-            "blocker_todo_ids": [
-                i["todo_id"] for i in blocker_items if i.get("todo_id")
-            ],
-            "edges": _wait_lineage_edges(
-                lineage_source_items
-                if lineage_source_items is not None
-                else source_items
-                if source_items is not None
-                else candidates
-            ),
-        },
-    )
-    if (
-        not isinstance(coverage, dict)
-        or coverage.get("schema_version") != "vision_wait_coverage_v0"
-    ):
-        raise RuntimeError("TypeScript vision wait coverage shape mismatch")
-    if coverage.get("covered") is not True:
-        return [], []
-    witnesses = set(coverage["witness_todo_ids"])
-    blocker_items = [i for i in blocker_items if i.get("todo_id") in witnesses]
-    candidates = (
-        []
-        if blocker_items
-        else [i for i in candidates if i.get("todo_id") in witnesses]
-    )
-    return blocker_items, candidates
-
-
 def build_goal_vision_wait_state(
     *,
     agent_todo_summary: dict[str, Any] | None,
@@ -241,98 +78,34 @@ def build_goal_vision_wait_state(
     checkpoints and closed-stage successor requirements remain strict.
     """
 
-    gaps = [gap for gap in (acceptance_gaps or []) if isinstance(gap, dict)]
-    if not gaps or any(gap.get("kind") != VISION_ACCEPTANCE_GAP_KIND for gap in gaps):
-        return None
-    if selectable_advancement_count > 0:
-        return None
-
-    # A gap with no causal link cannot borrow another gap's wait witness.
-    if any(not _acceptance_gap_causal_todo_ids([gap]) for gap in gaps):
-        return None
-    causal_todo_ids = _acceptance_gap_causal_todo_ids(gaps)
-    if agent_todo_summary:
-        for proof in (agent_todo_summary or {}).get("vision_wait_states") or []:
-            if (
-                isinstance(proof, dict)
-                and proof.get("schema_version") == GOAL_VISION_WAIT_STATE_SCHEMA_VERSION
-                and proof.get("state") == "waiting"
-                and proof.get("agent_id") == agent_id
-                and proof.get("causal_todo_ids") == sorted(causal_todo_ids)
-            ):
-                return dict(proof)
-    blocker_items, candidates = _covered_wait_items(
-        agent_todo_summary=agent_todo_summary,
-        agent_id=agent_id,
-        causal_todo_ids=causal_todo_ids,
-        source_items=source_items,
-        lineage_source_items=lineage_source_items,
-    )
-    if candidates:
-        selected = candidates[0]
-        waiting_todo_ids = [
-            todo_id
-            for todo_id in (
-                normalize_todo_id(item.get("todo_id")) for item in candidates[:5]
-            )
-            if todo_id
-        ]
-        payload: dict[str, Any] = {
-            "schema_version": GOAL_VISION_WAIT_STATE_SCHEMA_VERSION,
-            "state": "waiting",
-            "reason_code": "exact_blocked_successor",
-            "agent_id": agent_id,
-            "waiting_todo_count": len(candidates),
-            "waiting_todo_ids": waiting_todo_ids,
-            "selected_todo_id": normalize_todo_id(selected.get("todo_id")),
-            "selected_todo_status": selected.get("status"),
-            "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")
-            ),
-            "causal_todo_ids": sorted(causal_todo_ids),
-            "deferred_acceptance_gap_count": len(gaps),
-            "deferred_acceptance_gap_kinds": [VISION_ACCEPTANCE_GAP_KIND],
-            "automatic_resume": True,
-            "resume_behavior": (
-                "when resume_ready becomes true, restore ordinary open-todo or "
-                "deferred-successor routing without closing the active vision"
-            ),
+    summary = agent_todo_summary if isinstance(agent_todo_summary, dict) else {}
+    planning_input = summary
+    if source_items is not None:
+        planning_input = {
+            "items": source_items,
+            "deferred_items": [item for item in source_items if item.get("status") == "deferred"],
+            "resume_blocked_items": [
+                item for item in source_items
+                if item.get("resume_ready") is False and item.get("status") == "open"
+            ],
+            "current_agent_blocker_items": source_items,
         }
-        return {key: value for key, value in payload.items() if value is not None}
-
-    if not blocker_items:
-        return None
-    selected = blocker_items[0]
-    waiting_todo_ids = [
-        todo_id
-        for todo_id in (
-            normalize_todo_id(item.get("todo_id")) for item in blocker_items[:5]
-        )
-        if todo_id
-    ]
-    payload = {
-        "schema_version": GOAL_VISION_WAIT_STATE_SCHEMA_VERSION,
-        "state": "waiting",
-        "reason_code": "current_agent_blocker",
+    blocker_items = source_items if source_items is not None else summary.get("current_agent_blocker_items")
+    wait_summary = {"vision_wait_states": summary.get("vision_wait_states"),
+                    "current_agent_blocker_items": _vision_wait_blocker_rows(blocker_items or [])}
+    lineage_items = lineage_source_items if lineage_source_items is not None else source_items
+    result = effect_runtime_result("goal.vision_wait.project", {
+        "agent_todo_summary": wait_summary,
         "agent_id": agent_id,
-        "waiting_todo_count": len(blocker_items),
-        "waiting_todo_ids": waiting_todo_ids,
-        "selected_todo_id": normalize_todo_id(selected.get("todo_id")),
-        "selected_todo_status": selected.get("status"),
-        "selected_todo_priority": selected.get("priority"),
-        "selected_todo_claimed_by": selected.get("claimed_by"),
-        "blocker_reason": str(selected.get("reason") or "").strip(),
-        "causal_todo_ids": sorted(causal_todo_ids),
-        "deferred_acceptance_gap_count": len(gaps),
-        "deferred_acceptance_gap_kinds": [VISION_ACCEPTANCE_GAP_KIND],
-        "automatic_resume": False,
-        "resume_behavior": (
-            "re-evaluate the active vision when the blocker is resolved or "
-            "superseded, or when runnable scoped advancement appears"
-        ),
-    }
-    return {key: value for key, value in payload.items() if value is not None}
+        "acceptance_gaps": acceptance_gaps or [],
+        "selectable_advancement_count": selectable_advancement_count,
+        "lineage_source_items": _vision_wait_lineage_rows(lineage_items),
+        "resume_planning_request": build_todo_resume_planning_request(planning_input, agent_id=agent_id),
+    })
+    if result is not None and (
+        not isinstance(result, dict)
+        or result.get("schema_version") != GOAL_VISION_WAIT_STATE_SCHEMA_VERSION
+        or result.get("state") != "waiting"
+    ):
+        raise RuntimeError("TypeScript vision wait state shape mismatch")
+    return result