From df1904eb53137a2223cd029717bc4a35d16067cb Mon Sep 17 00:00:00 2001 From: "lusendong.6789" Date: Mon, 7 Sep 2026 22:04:18 +0800 Subject: [PATCH 1/3] fix(replan): derive complete causal wait coverage before compaction Signed-off-by: lusendong.6789 --- .../control_plane/effect_runtime_handlers.ts | 2 + .../goals/goal_frontier/__init__.py | 9 + loopx/control_plane/goals/goal_vision_wait.py | 195 ++++++---- .../goals/goal_vision_wait_projection.py | 55 +++ .../goals/vision_wait_coverage.ts | 46 +++ .../control_plane/quota/should_run_prepare.py | 6 +- .../todos/active_state_todo_parser.py | 1 + loopx/control_plane/todos/quota_summary.py | 15 +- loopx/control_plane/todos/todo_summary.py | 6 + .../work_items/semantic_replan_writeback.py | 10 +- ...test_goal_frontier_fallback_disposition.py | 14 +- .../test_vision_wait_coverage.py | 355 ++++++++++++++++++ .../vision_wait_coverage.test.ts | 49 +++ tsconfig.control-plane.json | 1 + 14 files changed, 671 insertions(+), 93 deletions(-) create mode 100644 loopx/control_plane/goals/goal_vision_wait_projection.py create mode 100644 loopx/control_plane/goals/vision_wait_coverage.ts create mode 100644 tests/control_plane/test_vision_wait_coverage.py create mode 100644 tests/control_plane_ts/vision_wait_coverage.test.ts diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 75ed0251e4..2e8230aae5 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/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], diff --git a/loopx/control_plane/goals/goal_frontier/__init__.py b/loopx/control_plane/goals/goal_frontier/__init__.py index 44080125be..884877f3b2 100644 --- a/loopx/control_plane/goals/goal_frontier/__init__.py +++ b/loopx/control_plane/goals/goal_frontier/__init__.py @@ -1046,6 +1046,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 +1605,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=( diff --git a/loopx/control_plane/goals/goal_vision_wait.py b/loopx/control_plane/goals/goal_vision_wait.py index 482dd3ec0e..f0388aa427 100644 --- a/loopx/control_plane/goals/goal_vision_wait.py +++ b/loopx/control_plane/goals/goal_vision_wait.py @@ -4,6 +4,8 @@ import json from typing import Any +from ..effect_runtime import effect_runtime_result +from ..todos.projection import todo_item_excludes_agent from ..todos.contract import ( TODO_TASK_CLASS_BLOCKER, normalize_todo_claimed_by, @@ -113,64 +115,108 @@ def _acceptance_gap_causal_todo_ids( ): values = gap.get(key) if isinstance(gap.get(key), list) else [] todo_ids.update( - todo_id - for value in values - if (todo_id := normalize_todo_id(value)) + todo_id for value in values if (todo_id := normalize_todo_id(value)) ) return todo_ids -def _causal_blocked_successor_items( - candidates: list[dict[str, Any]], +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], -) -> list[dict[str, Any]]: - """Keep only waits in the active vision's explicit Todo lineage.""" + 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, + } - if not causal_todo_ids: - return [] - lineage_ids = set(causal_todo_ids) - changed = True - while changed: - changed = False - for item in candidates: - todo_id = normalize_todo_id(item.get("todo_id")) - condition = ( - item.get("resume_condition") - if isinstance(item.get("resume_condition"), dict) - else {} - ) - target_todo_id = normalize_todo_id( - condition.get("target_todo_id") or condition.get("target") - ) - successor_todo_ids = { - successor_todo_id - for value in ( - item.get("successor_todo_ids") - if isinstance(item.get("successor_todo_ids"), list) - else [] - ) - if (successor_todo_id := normalize_todo_id(value)) - } - if not ( - (todo_id and todo_id in lineage_ids) - or (target_todo_id and target_todo_id in lineage_ids) - or successor_todo_ids.intersection(lineage_ids) - ): - continue - expanded = { - value - for value in (todo_id, target_todo_id, *successor_todo_ids) - if value - } - if not expanded.issubset(lineage_ids): - lineage_ids.update(expanded) - changed = True - return [ + blocker_items = ( + agent_todo_summary.get("current_agent_blocker_items") + if isinstance(agent_todo_summary, dict) + and isinstance(agent_todo_summary.get("current_agent_blocker_items"), list) + else [] + ) + safe_agent_id = normalize_todo_claimed_by(agent_id) + blocker_items = [ item - for item in candidates - if normalize_todo_id(item.get("todo_id")) in lineage_ids + 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 = todo_summary_blocked_successor_items( + agent_todo_summary or {}, agent_id=agent_id + ) + 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( @@ -179,6 +225,8 @@ def build_goal_vision_wait_state( agent_id: str | None, acceptance_gaps: list[dict[str, Any]] | None, selectable_advancement_count: int, + source_items: list[dict[str, Any]] | None = None, + lineage_source_items: list[dict[str, Any]] | None = None, ) -> dict[str, Any] | None: """Project a temporary vision wait over an authoritative blocked frontier. @@ -193,36 +241,27 @@ def build_goal_vision_wait_state( 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) - - blocker_items = ( - agent_todo_summary.get("current_agent_blocker_items") - if isinstance(agent_todo_summary, dict) - and isinstance(agent_todo_summary.get("current_agent_blocker_items"), list) - else [] + 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, ) - 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 normalize_todo_id(item.get("todo_id")) in causal_todo_ids - ] - candidates = [] - if not blocker_items: - candidates = _causal_blocked_successor_items( - todo_summary_blocked_successor_items( - agent_todo_summary or {}, - agent_id=agent_id, - ), - causal_todo_ids=causal_todo_ids, - ) if candidates: selected = candidates[0] waiting_todo_ids = [ @@ -247,6 +286,7 @@ def build_goal_vision_wait_state( "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, @@ -279,6 +319,7 @@ def build_goal_vision_wait_state( "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, diff --git a/loopx/control_plane/goals/goal_vision_wait_projection.py b/loopx/control_plane/goals/goal_vision_wait_projection.py new file mode 100644 index 0000000000..e5340adaf7 --- /dev/null +++ b/loopx/control_plane/goals/goal_vision_wait_projection.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +from typing import Any + +from ..todos.projection import agent_scoped_selectable_advancement_todo_ids +from .goal_vision_wait import build_goal_vision_wait_state + + +def attach_active_vision_waits( + summary: dict[str, Any], + runs: list[dict[str, Any]] | None, + *, + role: str | None, + items: list[dict[str, Any]], + lineage_items: list[dict[str, Any]] | None, +) -> None: + """Derive bounded wait witnesses before the canonical Todo rows are sliced. + + The ordinary refresh-state/Turn vision and Todo lifecycles are the only + producers. This read model is rebuilt on every source read, never persisted + or authored separately. Only positive proofs cross the presentation seam. + """ + if role != "agent" or not runs: + return + from .goal_frontier import acceptance_gaps_from_agent_vision + from .goal_frontier.semantic_history import latest_agent_vision_from_runs + + agent_ids = { + str( + (run.get("agent_vision") or {}).get("agent_id") or run.get("agent_id") or "" + ) + for run in runs + if isinstance(run, dict) and isinstance(run.get("agent_vision"), dict) + } - {""} + proofs = [] + for agent_id in sorted(agent_ids): + vision = latest_agent_vision_from_runs(runs, goal_id="", agent_id=agent_id) + gaps = acceptance_gaps_from_agent_vision(vision) + wait = build_goal_vision_wait_state( + agent_todo_summary=None, + agent_id=agent_id, + acceptance_gaps=gaps, + selectable_advancement_count=len( + agent_scoped_selectable_advancement_todo_ids( + {"executable_backlog_items": items}, + agent_id=agent_id, + ) + ), + source_items=items, + lineage_source_items=[*(lineage_items or []), *items], + ) + if wait: + proofs.append(wait) + if proofs: + summary["vision_wait_states"] = proofs diff --git a/loopx/control_plane/goals/vision_wait_coverage.ts b/loopx/control_plane/goals/vision_wait_coverage.ts new file mode 100644 index 0000000000..c34a39ae34 --- /dev/null +++ b/loopx/control_plane/goals/vision_wait_coverage.ts @@ -0,0 +1,46 @@ +import type { JsonObject } from "../effect_program.ts"; +import { requireJsonObject, requireStringArray } from "../runtime_decode.ts"; +import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; + +/** Every causal obligation needs its own wait witness; shared prerequisites + * do not make sibling work interchangeable. Inputs are evaluated Todo facts, + * never model-maintained alternate-route declarations. */ +export function projectVisionWaitCoverage(value: unknown): JsonObject { + const request = requireJsonObject(value, "vision wait coverage"); + const roots = requireStringArray(request.causal_todo_ids, "causal_todo_ids"); + const waits = new Set(requireStringArray(request.waiting_todo_ids, "waiting_todo_ids")); + const blockers = new Set(requireStringArray(request.blocker_todo_ids, "blocker_todo_ids")); + if (!Array.isArray(request.edges)) throw new EffectRuntimeRequestError("edges must be an array"); + const graph = new Map>(); + for (const edge of request.edges) { + const pair = requireStringArray(edge, "lineage edge"); + if (pair.length !== 2) throw new EffectRuntimeRequestError("lineage edge must have two ids"); + const targets = graph.get(pair[0]) ?? new Set(); + targets.add(pair[1]); + graph.set(pair[0], targets); + } + const witnesses = new Set(); + const uncovered = new Set(); + for (const root of roots) { + const seen = new Set(); + const pending = [root]; + let covered = false; + while (pending.length) { + const id = pending.pop()!; + if (seen.has(id)) continue; + seen.add(id); + if (waits.has(id) || blockers.has(id)) { + witnesses.add(id); + covered = true; + } + for (const next of graph.get(id) ?? []) pending.push(next); + } + if (!covered) uncovered.add(root); + } + return { + schema_version: "vision_wait_coverage_v0", + covered: roots.length > 0 && uncovered.size === 0, + witness_todo_ids: [...witnesses].sort(), + uncovered_todo_ids: [...uncovered].sort(), + }; +} diff --git a/loopx/control_plane/quota/should_run_prepare.py b/loopx/control_plane/quota/should_run_prepare.py index f075066714..632d7053cf 100644 --- a/loopx/control_plane/quota/should_run_prepare.py +++ b/loopx/control_plane/quota/should_run_prepare.py @@ -748,7 +748,11 @@ def _prepare_quota_should_run_item( project_asset=project_asset, user_todo_summary=user_todo_summary, agent_todo_summary=agent_todo_summary, - agent_todo_source_items=agent_todo_source_items, + agent_todo_source_items=select_planning_inventory_source_items( + item.get("agent_todos"), + project_asset.get("agent_todos") if project_asset else None, + include_terminal=True, + ), work_lane_contract=work_lane_contract, neutral_replan_ack_classifications=AUTONOMOUS_REPLAN_ACK_NEUTRAL_CLASSIFICATIONS, registered_agent_ids=registered_agent_ids, diff --git a/loopx/control_plane/todos/active_state_todo_parser.py b/loopx/control_plane/todos/active_state_todo_parser.py index 6b2ce0cdbf..aac7dbda06 100644 --- a/loopx/control_plane/todos/active_state_todo_parser.py +++ b/loopx/control_plane/todos/active_state_todo_parser.py @@ -141,6 +141,7 @@ def parse_active_state_todos( available_capabilities=available_capabilities, item_limit=item_limit, include_task_orchestration_authority=include_task_orchestration_authority, + vision_runs=(goal or {}).get("latest_runs"), ) archived_advancement_done_count = count_advancement_todos( [item for item in archive_items if item.get("done") is True] diff --git a/loopx/control_plane/todos/quota_summary.py b/loopx/control_plane/todos/quota_summary.py index 96b2c957b4..8761fbc3dc 100644 --- a/loopx/control_plane/todos/quota_summary.py +++ b/loopx/control_plane/todos/quota_summary.py @@ -569,6 +569,11 @@ def summarize_user_todos_for_quota( summary["convergence_open_count"] = value.get("convergence_open_count") if recent_completed_advancement_items: summary["recent_completed_advancement_items"] = recent_completed_advancement_items + if isinstance(value.get("vision_wait_states"), list): + summary["vision_wait_states"] = [ + proof for proof in value["vision_wait_states"] + if isinstance(proof, dict) and proof.get("agent_id") == agent_id + ] if blocker_items: summary["blocker_open_count"] = len(blocker_items) if current_agent_blocker_items: @@ -1114,7 +1119,7 @@ def select_quota_todo_source_items( return project_asset_items if project_asset_items is not None else canonical_items or [] -def _planning_inventory_source_items(value: Any) -> list[dict[str, Any]] | None: +def _planning_inventory_source_items(value: Any, *, include_terminal: bool = False) -> list[dict[str, Any]] | None: """Return canonical Todo rows before presentation-lane expansion. Planning consumers share domain-state rows instead of rebuilding a larger, @@ -1123,17 +1128,19 @@ def _planning_inventory_source_items(value: Any) -> list[dict[str, Any]] | None: if not isinstance(value, dict): return None - return todo_planning_source_items(value) + return todo_planning_source_items(value, include_terminal=include_terminal) def select_planning_inventory_source_items( canonical_value: Any, project_asset_value: Any, + *, + include_terminal: bool = False, ) -> list[dict[str, Any]]: """Select the canonical non-terminal rows shared by planning read models.""" - canonical_items = _planning_inventory_source_items(canonical_value) - project_asset_items = _planning_inventory_source_items(project_asset_value) + canonical_items = _planning_inventory_source_items(canonical_value, include_terminal=include_terminal) + project_asset_items = _planning_inventory_source_items(project_asset_value, include_terminal=include_terminal) if is_canonical_attention_todo_summary(canonical_value): return canonical_items if canonical_items is not None else project_asset_items or [] return project_asset_items if project_asset_items is not None else canonical_items or [] diff --git a/loopx/control_plane/todos/todo_summary.py b/loopx/control_plane/todos/todo_summary.py index 800a17439c..065ac305df 100644 --- a/loopx/control_plane/todos/todo_summary.py +++ b/loopx/control_plane/todos/todo_summary.py @@ -5,6 +5,7 @@ import re from typing import Any, Callable, Optional +from ..goals.goal_vision_wait_projection import attach_active_vision_waits from .contract import ( TODO_RESUME_KIND_TODO_DONE, TODO_STATUS_DONE, @@ -1064,6 +1065,7 @@ def compact_todo_group( available_capabilities: Any = None, item_limit: int | None = MAX_STATUS_TODOS_PER_ROLE, include_task_orchestration_authority: bool = False, + vision_runs: list[dict[str, Any]] | None = None, ) -> dict[str, Any] | None: if not items and not include_empty_source: return None @@ -1204,6 +1206,10 @@ def compact_todo_group( ][:MAX_DEFERRED_TODO_VISIBILITY_ITEMS], "items": lanes.budgeted_items if item_limit is None else lanes.budgeted_items[:item_limit], } + attach_active_vision_waits( + summary, vision_runs, role=role, items=items, + lineage_items=resume_source_items, + ) if watch_only_monitor_items: summary["watch_only_monitor_count"] = len(watch_only_monitor_items) summary["watch_only_monitor_due_count"] = len(watch_only_monitor_due_items) diff --git a/loopx/control_plane/work_items/semantic_replan_writeback.py b/loopx/control_plane/work_items/semantic_replan_writeback.py index cd6fe36395..48db421e62 100644 --- a/loopx/control_plane/work_items/semantic_replan_writeback.py +++ b/loopx/control_plane/work_items/semantic_replan_writeback.py @@ -19,7 +19,7 @@ ) from ..todos.active_state_todo_parser import parse_active_state_todos from ..todos.quota_summary import ( - select_quota_todo_source_items, + select_planning_inventory_source_items, select_quota_todo_summary, ) from ..todos.succession_warning import todo_succession_gap_items @@ -204,7 +204,10 @@ def qualify_replan_writeback( safe_agent_id = str(agent_id or "").strip() if not safe_agent_id: return None, None - todo_projection = parse_active_state_todos(state_text, item_limit=None) + todo_projection = parse_active_state_todos( + state_text, item_limit=None, + goal={**(registry_goal or {}), "latest_runs": list(newest_first_runs or [])}, + ) registered_agent_ids = registered_agent_ids_for_goal(registry_goal) agent_identity = ( build_quota_agent_identity(registry_goal, agent_id=safe_agent_id) @@ -227,9 +230,10 @@ def qualify_replan_writeback( None, agent_identity=agent_identity, ) - agent_todo_source_items = select_quota_todo_source_items( + agent_todo_source_items = select_planning_inventory_source_items( raw_agent_todos, None, + include_terminal=True, ) agent_todo_completion_items = ( [ diff --git a/tests/control_plane/test_goal_frontier_fallback_disposition.py b/tests/control_plane/test_goal_frontier_fallback_disposition.py index c02190e43c..ac2f8480f7 100644 --- a/tests/control_plane/test_goal_frontier_fallback_disposition.py +++ b/tests/control_plane/test_goal_frontier_fallback_disposition.py @@ -210,7 +210,7 @@ def test_declared_fallback_survives_prepare_compact_and_readback() -> None: assert readback["fallback_declarations"] == vision["fallback_declarations"] -def test_declared_fallback_without_resolution_projects_single_gap() -> None: +def test_uncovered_causal_todo_requires_replan_independently_of_fallback_advice() -> None: # The structured declaration links the fallback direction to a Todo id, # but no runnable Todo with that id exists on this agent's frontier. payload = _status_payload( @@ -224,12 +224,10 @@ def test_declared_fallback_without_resolution_projects_single_gap() -> None: frontier = _frontier_projection(payload) - # The blocked-successor wait state clears ordinary acceptance gaps; the - # declared fallback would disappear silently without the dedicated field. - assert frontier["acceptance_gaps"] == [] - wait = frontier["vision_wait_state"] - assert wait["reason_code"] == "exact_blocked_successor" - assert wait["selected_todo_id"] == PRIMARY_WAIT_ID + # The ordinary acceptance links both routes. A wait for the primary alone + # cannot cover the other causal Todo; the fallback field stays advisory. + assert [gap["kind"] for gap in frontier["acceptance_gaps"]] == ["vision_acceptance_gap"] + assert "vision_wait_state" not in frontier gaps = frontier["fallback_gaps"] assert len(gaps) == 1 gap = gaps[0] @@ -239,7 +237,7 @@ def test_declared_fallback_without_resolution_projects_single_gap() -> None: assert gap["unresolved_todo_ids"] == [FALLBACK_ID] assert "fallback" in gap["recommended_action"] assert "do not invent a user gate" in gap["recommended_action"] - assert frontier["replan_required"] is False + assert frontier["replan_required"] is True decision = build_quota_should_run( payload, diff --git a/tests/control_plane/test_vision_wait_coverage.py b/tests/control_plane/test_vision_wait_coverage.py new file mode 100644 index 0000000000..0be6961dfb --- /dev/null +++ b/tests/control_plane/test_vision_wait_coverage.py @@ -0,0 +1,355 @@ +"""Declared alternate paths must survive local failure, wait, and replan writeback.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from loopx.cli import main as cli_main +from loopx.control_plane.goals.goal_vision import normalize_goal_vision_packet +from loopx.control_plane.scheduler.execution_context import ( + scheduler_execution_context_for_runtime_profile, +) +from loopx.control_plane.testing.quota_fixtures import quota_status_payload +from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos +from loopx.control_plane.work_items.semantic_replan_writeback import ( + ReplanWritebackRejected, + enforce_open_replan_writeback, +) +from loopx.quota import build_quota_should_run + +GOAL = "fallback-wait-fixture" +AGENT = "worker" +COORDINATION = {"agent_model": "peer_v1", "registered_agents": [AGENT, "observer"]} +STATE = """# Active Goal State + +## Agent Todo + +- [ ] [P0] Observe recovery of source A. + +- [ ] [P0] Read source A after recovery. + +- [ ] [P2] Check source recovery on schedule. + +""" + + +def vision_run(*, declared: bool = True) -> dict: + vision = normalize_goal_vision_packet( + { + "state": "vision_drift_detected", + "vision_patch": { + "acceptance_summary": "Complete the bounded source check.", + "replan_trigger_summary": "Source A is unavailable.", + }, + "todo_delta": ["retain:todo_source_a", "create:todo_source_b"], + "fallback_declarations": ( + [{"declaration_id": "source-b", "target_todo_id": "todo_source_b"}] + if declared + else [] + ), + }, + goal_id=GOAL, + agent_id=AGENT, + ) + return { + "classification": "source_path_checkpoint", + "agent_id": AGENT, + "generated_at": "2026-09-01T00:00:00+00:00", + "agent_vision": vision, + } + + +def decision( + *, + state: str = STATE, + runs: list[dict] | None = None, + profile: str = "outer_controller", +) -> dict: + parsed = parse_active_state_todos(state, item_limit=None) + payload = quota_status_payload( + goal_id=GOAL, + status="active", + recommended_action="Wait for source A recovery.", + agent_todos=parsed["agent_todos"], + user_todos=parsed.get("user_todos"), + coordination=COORDINATION, + latest_runs=runs if runs is not None else [vision_run()], + ) + return build_quota_should_run( + payload, + goal_id=GOAL, + agent_id=AGENT, + scheduler_execution_context=scheduler_execution_context_for_runtime_profile( + profile + ), + ) + + +def fallback_state(metadata: str) -> str: + return ( + STATE + + f""" +- [ ] [P1] Read authorized source B. + +""" + ) + + +@pytest.mark.parametrize( + "extra_count,reverse", [(0, False), (8, False), (40, False), (40, True)] +) +@pytest.mark.parametrize("declared", [False, True]) +def test_cli_wait_coverage_is_independent_of_display_size( + tmp_path: Path, capsys, extra_count: int, reverse: bool, declared: bool +) -> None: + state_file = tmp_path / "ACTIVE_GOAL_STATE.md" + extra = "".join( + f"\n- [ ] [P1] Unrelated external wait {i}.\n \n" + for i in range(extra_count) + ) + + def state(metadata: str | None) -> str: + b = fallback_state(metadata)[len(STATE) :] if metadata else "" + return STATE + (b + extra if reverse else extra + b) + + state_file.write_text(state(None)) + registry = tmp_path / "registry.json" + registry.write_text( + json.dumps( + { + "goals": [ + { + "id": GOAL, + "status": "active", + "domain": "engineering", + "waiting_on": "codex", + "state_file": str(state_file), + "repo": str(tmp_path), + "adapter": { + "kind": "fixture_connected_delivery_v0", + "status": "connected-delivery", + }, + "quota": { + "compute": 1.0, + "window_hours": 24, + "slot_minutes": 1, + }, + "coordination": COORDINATION, + } + ] + } + ) + ) + runtime = tmp_path / "runtime" + runs = runtime / "goals" / GOAL / "runs" + runs.mkdir(parents=True) + if not declared: + assert ( + cli_main( + [ + "--format", + "json", + "--registry", + str(registry), + "--runtime-root", + str(runtime), + "refresh-state", + "--goal-id", + GOAL, + "--agent-id", + AGENT, + "--progress-scope", + "agent_lane", + "--delivery-outcome", + "surface_only", + "--vision-state", + "vision_drift_detected", + "--vision-acceptance", + "Complete both bounded source checks.", + "--vision-replan-trigger", + "Both source checks still require evidence.", + "--vision-todo-delta", + "retain:todo_source_a", + "--vision-todo-delta", + "create:todo_source_b", + "--suppress-external-sinks", + ] + ) + == 0 + ) + authored = json.loads(capsys.readouterr().out) + assert authored["ok"] is True + else: + run = vision_run(declared=declared) + json_path = runs / "source-checkpoint.json" + markdown_path = runs / "source-checkpoint.md" + json_path.write_text(json.dumps(run) + "\n") + markdown_path.write_text("# Source checkpoint\n") + (runs / "index.jsonl").write_text( + json.dumps( + { + **run, + "json_path": str(json_path), + "markdown_path": str(markdown_path), + } + ) + + "\n" + ) + args = [ + "--format", + "json", + "--registry", + str(registry), + "--runtime-root", + str(runtime), + "quota", + "should-run", + "--verbose", + "--goal-id", + GOAL, + "--agent-id", + AGENT, + "--runtime-profile", + "ark_managed_agent_goal", + ] + assert cli_main(args) == 0 + missing = json.loads(capsys.readouterr().out) + assert missing["effective_action"] == "autonomous_replan_required" + state_file.write_text(state("status=open claimed_by=worker")) + assert cli_main(args) == 0 + continued = json.loads(capsys.readouterr().out) + assert continued["selected_todo"]["todo_id"] == "todo_source_b" + state_file.write_text( + state("status=deferred claimed_by=worker resume_when=todo_done:todo_recovery") + ) + assert cli_main(args) == 0 + waiting = json.loads(capsys.readouterr().out) + assert waiting["execution_obligation"]["must_attempt_work"] is False + assert ( + waiting["scheduler_hint"]["goal_runtime_continuation"]["disposition"] == "defer" + ) + + +@pytest.mark.parametrize( + "metadata", + [ + "status=open claimed_by=observer", + "status=open claimed_by=worker excluded_agents=worker", + "status=deferred claimed_by=worker", + "status=done claimed_by=worker no_followup=true", + ], +) +def test_uncovered_causal_todo_does_not_borrow_source_a_wait(metadata: str) -> None: + result = decision(state=fallback_state(metadata), runs=[vision_run(declared=False)]) + assert (result.get("selected_todo") or {}).get("todo_id") != "todo_source_b" + assert result["goal_frontier_projection"].get("vision_wait_state") is None + assert result["effective_action"] == "autonomous_replan_required" + + +def test_missing_link_does_not_borrow_an_unrelated_blocker() -> None: + state = ( + STATE + + """ +- [ ] [P1] Await unrelated owner input. + +""" + ) + result = decision(state=state, runs=[vision_run(declared=False)]) + assert result["goal_frontier_projection"].get("vision_wait_state") is None + assert result["goal_frontier_projection"]["acceptance_gaps"] + + +@pytest.mark.parametrize("declared", [False, True]) +def test_explicit_terminal_vision_keeps_existing_lifecycle_semantics( + declared: bool, +) -> None: + run = vision_run(declared=declared) + run["agent_vision"]["state"] = "no_followup" + assert not decision(runs=[run])["goal_frontier_projection"]["acceptance_gaps"] + + +def test_real_wait_clears_readback_and_writeback_obligation_without_declaration() -> ( + None +): + state = fallback_state( + "status=deferred claimed_by=worker resume_when=todo_done:todo_recovery" + ) + assert ( + enforce_open_replan_writeback( + newest_first_runs=[vision_run(declared=False)], + state_text=state, + agent_id=AGENT, + goal_id=GOAL, + registry_goal={"coordination": COORDINATION}, + ) + is None + ) + + +def test_empty_writeback_cannot_settle_uncovered_acceptance() -> None: + with pytest.raises(ReplanWritebackRejected): + enforce_open_replan_writeback( + newest_first_runs=[vision_run(declared=False)], + state_text=STATE, + agent_id=AGENT, + goal_id=GOAL, + registry_goal={"coordination": COORDINATION}, + ) + + +def test_single_authorized_path_may_wait_without_inventing_an_alternative() -> None: + run = vision_run(declared=False) + run["agent_vision"]["todo_delta"] = ["retain:todo_source_a"] + result = decision(runs=[run]) + assert result["execution_obligation"]["must_attempt_work"] is False + assert ( + result["goal_frontier_projection"]["vision_wait_state"]["selected_todo_id"] + == "todo_source_a" + ) + + +@pytest.mark.parametrize("archived", [False, True]) +def test_completed_predecessor_needs_a_real_waiting_successor(archived: bool) -> None: + run = vision_run(declared=False) + run["agent_vision"]["todo_delta"] = ["retain:todo_old_route"] + source = fallback_state( + "status=deferred claimed_by=worker resume_when=todo_done:todo_recovery" + ) + if archived: + source += "\n## Completed Work Archive\n" + source += """ +- [x] [P1] Replaced source route. + +""" + parsed = parse_active_state_todos( + source, item_limit=None, goal={"latest_runs": [run]} + ) + proof = parsed["agent_todos"]["vision_wait_states"][0] + assert proof["selected_todo_id"] == "todo_source_b" + assert enforce_open_replan_writeback( + newest_first_runs=[run], state_text=source, agent_id=AGENT, goal_id=GOAL, + registry_goal={"coordination": COORDINATION}, + ) is None + # Finishing a predecessor alone does not close an open acceptance. + without_successor = source.replace(" successor_todo_ids=todo_source_b", "") + unproven = parse_active_state_todos( + without_successor, item_limit=None, goal={"latest_runs": [run]} + ) + assert not unproven["agent_todos"].get("vision_wait_states") + + +def test_monitor_and_excluded_blocker_cannot_cover_causal_work() -> None: + for metadata in [ + "status=open task_class=continuous_monitor claimed_by=worker cadence=1h next_due_at=2099-01-01T00:00:00+00:00", + "status=blocked task_class=blocker claimed_by=worker excluded_agents=worker reason=External%20wait", + ]: + source = ( + STATE + + f"\n- [ ] [P1] Source B observation.\n \n" + ) + result = decision(state=source, runs=[vision_run(declared=False)]) + assert not result["goal_frontier_projection"].get("vision_wait_state") + assert result["goal_frontier_projection"]["acceptance_gaps"] diff --git a/tests/control_plane_ts/vision_wait_coverage.test.ts b/tests/control_plane_ts/vision_wait_coverage.test.ts new file mode 100644 index 0000000000..81c74eb197 --- /dev/null +++ b/tests/control_plane_ts/vision_wait_coverage.test.ts @@ -0,0 +1,49 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { projectVisionWaitCoverage } from "../../loopx/control_plane/goals/vision_wait_coverage.ts"; + +const facts = (extra: Record = {}) => ({ + causal_todo_ids: ["todo_a", "todo_b"], waiting_todo_ids: ["todo_a"], + blocker_todo_ids: [], edges: [], ...extra, +}); + +test("one waiting path cannot cover another acceptance obligation", () => { + assert.deepEqual(projectVisionWaitCoverage(facts()).uncovered_todo_ids, ["todo_b"]); + assert.equal(projectVisionWaitCoverage(facts({ waiting_todo_ids: ["todo_a", "todo_b"] })).covered, true); +}); + +test("shared prerequisites do not turn sibling paths into substitutes", () => { + assert.equal(projectVisionWaitCoverage(facts({ + causal_todo_ids: ["todo_b"], edges: [["todo_recovery", "todo_a"], ["todo_recovery", "todo_b"]], + })).covered, false); + assert.equal(projectVisionWaitCoverage(facts({ + causal_todo_ids: ["todo_recovery"], edges: [["todo_recovery", "todo_a"]], + })).covered, true); +}); + +test("explicit successors carry lineage through replaced or completed predecessors", () => { + assert.equal(projectVisionWaitCoverage(facts({ + causal_todo_ids: ["todo_old"], edges: [["todo_old", "todo_next"], ["todo_next", "todo_a"]], + })).covered, true); + assert.equal(projectVisionWaitCoverage(facts({ causal_todo_ids: ["todo_done"] })).covered, false); +}); + +test("every root needs coverage even when one has a concrete blocker", () => { + assert.equal(projectVisionWaitCoverage(facts({ waiting_todo_ids: [], blocker_todo_ids: ["todo_a"] })).covered, false); + assert.equal(projectVisionWaitCoverage(facts({ waiting_todo_ids: ["todo_b"], blocker_todo_ids: ["todo_a"] })).covered, true); +}); + +test("cycles terminate and unrelated edges and input order do not change coverage", () => { + const edges = [["todo_a", "todo_b"], ["todo_b", "todo_a"], ["todo_other", "todo_remote"]]; + const a = projectVisionWaitCoverage(facts({ edges })); + const b = projectVisionWaitCoverage(facts({ edges: [...edges].reverse(), causal_todo_ids: ["todo_b", "todo_a"] })); + assert.deepEqual(a, b); + assert.equal(projectVisionWaitCoverage(facts({ edges, waiting_todo_ids: [] })).covered, false); +}); + +test("unbound acceptance and malformed facts cannot prove a wait", () => { + assert.equal(projectVisionWaitCoverage(facts({ causal_todo_ids: [] })).covered, false); + for (const patch of [{ edges: [["todo_a"]] }, { edges: null }, { waiting_todo_ids: [null] }]) { + assert.throws(() => projectVisionWaitCoverage(facts(patch))); + } +}); diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json index 53016ff5f4..313ab8ac75 100644 --- a/tsconfig.control-plane.json +++ b/tsconfig.control-plane.json @@ -106,6 +106,7 @@ "tests/control_plane_ts/turn_journal.test.ts", "tests/control_plane_ts/turn_journal_effects.test.ts", "tests/control_plane_ts/vision_checkpoint.test.ts", + "tests/control_plane_ts/vision_wait_coverage.test.ts", "tests/control_plane_ts/shared_goal_alignment.test.ts", "tests/control_plane_ts/goal_amendment_proposal.test.ts" ] From 59b0336db2e3e6fd4510718fa4eaa7f96e2ec2fd Mon Sep 17 00:00:00 2001 From: "lusendong.6789" Date: Mon, 7 Sep 2026 22:04:45 +0800 Subject: [PATCH 2/3] docs(replan): define causal wait coverage and authoring lifecycle Signed-off-by: lusendong.6789 --- .../goal-vision-replan-contract-v0.md | 41 +++++++++++++++++-- .../references/repair-patterns.md | 1 + 2 files changed, 38 insertions(+), 4 deletions(-) diff --git a/docs/reference/protocols/goal-vision-replan-contract-v0.md b/docs/reference/protocols/goal-vision-replan-contract-v0.md index 31ef0b00a6..7700f9b742 100644 --- a/docs/reference/protocols/goal-vision-replan-contract-v0.md +++ b/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 :` 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 diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md index fc3159ea4e..01d412162d 100644 --- a/skills/loopx-self-repair/references/repair-patterns.md +++ b/skills/loopx-self-repair/references/repair-patterns.md @@ -5,6 +5,7 @@ teaches a reusable control-plane lesson. | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair | | --- | --- | --- | --- | --- | +| `causal_wait_coverage_gap` | One waiting route hides another open acceptance binding, or adding unrelated waiting Todos changes defer into replan. | Replay the same canonical Todo/vision facts through the real CLI with absent/runnable/waiting successors, varied display sizes, and reordered unrelated rows. | A union of related waits was treated as coverage of every causal binding, or compact display absence was treated as canonical absence. | Derive per-binding wait coverage from canonical Todo rows before display slicing; reuse resume/ownership evaluators and directed successor lineage. Keep incomplete evidence distinct from proven absence, and prove ordinary authoring and lifecycle paths before adding separately maintained declarations. | | `narrative_delivery_authority_leak` | Rewording a run label turns resolved work into a blocker or preparation into a required follow-through; untyped history accumulates delivery debt. | Explicit outcome/kind/scale, scoped progress observation, compact history, and status/quota decisions before and after narrative-only mutations. | Keyword inference and unknown-as-failure streaks promoted narrative into machine authority. | Use explicit typed semantics and the existing scoped blocker predicate; unknown breaks evidence streaks. Delete keyword fallbacks, preserve labels for display, and test both negative narrative mutations and positive typed obligations through the CLI. | | `periodic_report_todo_log_editorial_gap` | A stage report is generated and frozen, but the artifact is an English Todo chronology, report-building work displaces business findings, or the page lacks a clear overview-to-depth analysis mainline. | Frozen artifact and language, projected completed/open Todo facts, report-meta action kinds, editorial source and section order, cause/boundary details, fact references, and approval-gate creation time. | The governed consumer treated durable Todo prose as audience-ready copy and let a renderer infer narrative structure from timestamps or generic content kinds. Generation correctness therefore proved persistence and authority, but not editorial quality. | Freeze a compact public-safe fact request for the exact intent, require Agent-authored Chinese editorial with a typed `overview -> problem map -> causal analysis -> coverage/actions -> next actions` contract, validate language density and fact lineage in the consumer, and create no frozen artifact or approval gate until the editorial response passes. Keep report-meta work outside the business narrative and retain publication as a later exact-payload authority gate. | | `periodic_report_stage_boundary_premature_promotion` | A periodic report is generated after setup work, several ordinary Todo completions, or a replan even though the intended analysis or delivery vision remains open. | Goal-vision revision, material outcome checkpoint and evidence refs, replan trigger kind, accepted semantic delta, successor frontier ownership or terminal Goal, report trigger receipt, and rendered-artifact count. | Runtime aggregation treated Todo volume or any replan as proof of a meaningful stage boundary instead of distinguishing success-path vision closure from recovery and continuation planning. | Derive stage completion as the strict successful sibling of autonomous replan: require an evidence-linked closed vision plus a durably settled successor frontier or terminal Goal, deduplicate by closed-vision/frontier identity, and reject ordinary completion plus blocker, stall, long-chain, and monitor replans. Keep generation, publication, and announcement as separate authority layers. | From 3a6c5ca9555f057c27259baf4432ccd970aa9ec2 Mon Sep 17 00:00:00 2001 From: "lusendong.6789" Date: Mon, 7 Sep 2026 22:14:46 +0800 Subject: [PATCH 3/3] refactor(vision): share narrow read model with Todo projection Signed-off-by: lusendong.6789 --- .../goals/goal_frontier/__init__.py | 99 +--------- .../goal_frontier/fallback_disposition.py | 30 +-- .../goals/goal_frontier/semantic_history.py | 48 +---- .../goals/goal_vision_read_model.py | 176 ++++++++++++++++++ loopx/control_plane/goals/goal_vision_wait.py | 32 +--- .../goals/goal_vision_wait_projection.py | 6 +- 6 files changed, 204 insertions(+), 187 deletions(-) create mode 100644 loopx/control_plane/goals/goal_vision_read_model.py diff --git a/loopx/control_plane/goals/goal_frontier/__init__.py b/loopx/control_plane/goals/goal_frontier/__init__.py index 884877f3b2..6127ea80ce 100644 --- a/loopx/control_plane/goals/goal_frontier/__init__.py +++ b/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, *, diff --git a/loopx/control_plane/goals/goal_frontier/fallback_disposition.py b/loopx/control_plane/goals/goal_frontier/fallback_disposition.py index c27cf0f28c..5382e49282 100644 --- a/loopx/control_plane/goals/goal_frontier/fallback_disposition.py +++ b/loopx/control_plane/goals/goal_frontier/fallback_disposition.py @@ -10,13 +10,13 @@ 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. @@ -24,7 +24,6 @@ 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]: diff --git a/loopx/control_plane/goals/goal_frontier/semantic_history.py b/loopx/control_plane/goals/goal_frontier/semantic_history.py index 796b7c49f0..0beef33c7f 100644 --- a/loopx/control_plane/goals/goal_frontier/semantic_history.py +++ b/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]], *, diff --git a/loopx/control_plane/goals/goal_vision_read_model.py b/loopx/control_plane/goals/goal_vision_read_model.py new file mode 100644 index 0000000000..4d6e6a58fb --- /dev/null +++ b/loopx/control_plane/goals/goal_vision_read_model.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +from typing import Any + +from .goal_vision_state import ( + goal_vision_state_is_closed, + goal_vision_state_requires_successor, +) + +# Shared read contract for both Todo wait projection and frontier decisions. +VISION_FRONTIER_TODO_DELTA_ACTIONS = frozenset( + {"activate", "create", "reopen", "resume", "retain"} +) +VISION_TODO_DELTA_ID_LIMIT = 120 +VISION_ACCEPTANCE_GAP_TRIGGER = "vision_acceptance_gap" +VISION_SUCCESSOR_GAP_TRIGGER = "vision_successor_required" + + +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 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_projection_text(raw_todo_id, limit=VISION_TODO_DELTA_ID_LIMIT) + or "" + ) + 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 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: dict[str, Any] = { + "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 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 [] + raw_patch = agent_vision.get("vision_patch") + patch = raw_patch if isinstance(raw_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] diff --git a/loopx/control_plane/goals/goal_vision_wait.py b/loopx/control_plane/goals/goal_vision_wait.py index f0388aa427..ef4f29e4f4 100644 --- a/loopx/control_plane/goals/goal_vision_wait.py +++ b/loopx/control_plane/goals/goal_vision_wait.py @@ -23,22 +23,13 @@ def exact_blocked_successor_wait_state(value: Any) -> dict[str, Any]: return {} candidate = value if value.get("schema_version") != GOAL_VISION_WAIT_STATE_SCHEMA_VERSION: - candidate = ( - value.get("vision_wait_state") - if isinstance(value.get("vision_wait_state"), dict) - else {} - ) + raw_candidate = value.get("vision_wait_state") + candidate = raw_candidate if isinstance(raw_candidate, dict) else {} if not candidate: - projection = ( - value.get("goal_frontier_projection") - if isinstance(value.get("goal_frontier_projection"), dict) - else {} - ) - candidate = ( - projection.get("vision_wait_state") - if isinstance(projection.get("vision_wait_state"), dict) - else {} - ) + raw_projection = value.get("goal_frontier_projection") + projection = raw_projection if isinstance(raw_projection, dict) else {} + raw_candidate = projection.get("vision_wait_state") + candidate = raw_candidate if isinstance(raw_candidate, dict) else {} if ( candidate.get("schema_version") != GOAL_VISION_WAIT_STATE_SCHEMA_VERSION or candidate.get("state") != "waiting" @@ -113,7 +104,8 @@ def _acceptance_gap_causal_todo_ids( "successor_todo_ids", "completed_todo_ids", ): - values = gap.get(key) if isinstance(gap.get(key), list) else [] + 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)) ) @@ -164,12 +156,8 @@ def _covered_wait_items( "current_agent_blocker_items": source_items, } - blocker_items = ( - agent_todo_summary.get("current_agent_blocker_items") - if isinstance(agent_todo_summary, dict) - and isinstance(agent_todo_summary.get("current_agent_blocker_items"), list) - else [] - ) + 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 diff --git a/loopx/control_plane/goals/goal_vision_wait_projection.py b/loopx/control_plane/goals/goal_vision_wait_projection.py index e5340adaf7..7a3b62dd2c 100644 --- a/loopx/control_plane/goals/goal_vision_wait_projection.py +++ b/loopx/control_plane/goals/goal_vision_wait_projection.py @@ -3,6 +3,10 @@ from typing import Any from ..todos.projection import agent_scoped_selectable_advancement_todo_ids +from .goal_vision_read_model import ( + acceptance_gaps_from_agent_vision, + latest_agent_vision_from_runs, +) from .goal_vision_wait import build_goal_vision_wait_state @@ -22,8 +26,6 @@ def attach_active_vision_waits( """ if role != "agent" or not runs: return - from .goal_frontier import acceptance_gaps_from_agent_vision - from .goal_frontier.semantic_history import latest_agent_vision_from_runs agent_ids = { str(