Skip to content

refactor(goals): converge Vision wait projection in TypeScript #7298

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hhyykk wants to merge 1 commit into
base: main
Choose a base branch
from
+276 −304
Open

refactor(goals): converge Vision wait projection in TypeScript #7298

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions docs/architecture/rfcs/typescript-control-plane-migration-v0.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
5 changes: 4 additions & 1 deletion docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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。

Expand Down
1 change: 1 addition & 0 deletions loopx/control_plane/effect_runtime_handlers.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)],
Expand Down
313 changes: 43 additions & 270 deletions loopx/control_plane/goals/goal_vision_wait.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")

@loopx-agent loopx-agent Oct 25, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] 保留 blocker 的 priority

当同一 Agent 的 canonical blocker 为 P0、带非空 reason 并覆盖当前 Vision gap 时，这个白名单丢弃了 priority。下游 waitStatePayload 仍读取 selectedTodo.priority，再把 null 字段移除，因此公开 quota/status 等待投影由 selected_todo_priority: P0 变为缺键。已在不可变 base/head 的真实 File、SQLite CLI，以及隔离 wheel、sdist 中复现；没有把 reason 的已披露 typed-state 修正当成问题。请传递 priority，并通过真实 CLI 断言字段存在及值。

The new blocker-row codec drops a canonical priority required by the existing public wait payload; preserve it and cover the ordinary CLI path.

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]:
Expand Down Expand Up @@ -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,
Expand All @@ -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
Loading
Loading