From 542570c6ac890839ec431e946eeb36fd03021522 Mon Sep 17 00:00:00 2001 From: now-ing Date: Sun, 6 Sep 2026 11:15:25 +0800 Subject: [PATCH 1/7] fix(goal): surface declared vision fallback gaps in frontier projection A fallback direction declared in the goal vision used to disappear silently once the primary path blocked: the blocked-successor wait state clears acceptance gaps, so the fallback text stayed in the vision while the scheduler had no selectable successor for it. Add an advisory read-path check that keeps each declared fallback accounted for at projection time. When the primary lane is blocked and none of the three dispositions holds (a runnable open Todo referenced via todo_delta, a bounded create/reopen successor declaration, or an explicit terminal no-follow-up disposition via a closed-family state or path_delta.outcome=stop), exactly one vision_fallback_unresolved gap is projected into the independent fallback_gaps field, which the wait state never clears. The field is advisory only: it does not enter the acceptance-gap replan stream and does not render into CLI output surfaces. Focused fixtures cover the three acceptance checks from the issue: blocked primary plus a runnable fallback stays selectable, an unresolved declared fallback projects exactly one actionable gap, and a valid terminal disposition closes the gap without regenerating it. Signed-off-by: now-ing --- .../goals/goal_frontier/__init__.py | 21 ++ .../goal_frontier/fallback_disposition.py | 201 +++++++++++++++ ...test_goal_frontier_fallback_disposition.py | 235 ++++++++++++++++++ 3 files changed, 457 insertions(+) create mode 100644 loopx/control_plane/goals/goal_frontier/fallback_disposition.py create mode 100644 tests/control_plane/test_goal_frontier_fallback_disposition.py diff --git a/loopx/control_plane/goals/goal_frontier/__init__.py b/loopx/control_plane/goals/goal_frontier/__init__.py index f71765ccff..4075b56405 100644 --- a/loopx/control_plane/goals/goal_frontier/__init__.py +++ b/loopx/control_plane/goals/goal_frontier/__init__.py @@ -47,6 +47,7 @@ autonomous_replan_ack_satisfies_obligation, replan_successor_transition_ack, ) +from .fallback_disposition import declared_fallback_gap_from_agent_vision from .long_todo_chain import ( LONG_TODO_CHAIN_TRIGGER, classify_long_todo_chain_ack, @@ -1416,6 +1417,7 @@ def build_goal_frontier_projection_from_summaries( replan_obligation: dict[str, Any] | None, acceptance_gaps: list[dict[str, Any]] | None = None, vision_wait_state: dict[str, Any] | None = None, + fallback_gaps: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: user_counts = _summary_task_counts(user_todo_summary) agent_counts = _summary_task_counts(agent_todo_summary) @@ -1453,6 +1455,7 @@ def build_goal_frontier_projection_from_summaries( replan_obligation=replan_obligation, acceptance_gaps=acceptance_gaps, vision_wait_state=vision_wait_state, + fallback_gaps=fallback_gaps, deferred_successors=_deferred_successors( agent_todo_summary, agent_id=agent_id, @@ -1602,6 +1605,17 @@ def build_goal_frontier_projection_context_from_status( ), ) acceptance_gaps = [] if vision_wait_state else source_acceptance_gaps + declared_fallback_gaps = [ + gap + for gap in ( + declared_fallback_gap_from_agent_vision( + latest_agent_vision, + agent_todo_summary=agent_todo_summary, + agent_id=agent_id, + ), + ) + if isinstance(gap, dict) + ] projected_replan_ack = projected_autonomous_replan_ack_for_agent( item, project_asset, @@ -1716,6 +1730,7 @@ def build_goal_frontier_projection_context_from_status( replan_obligation=replan_obligation, acceptance_gaps=acceptance_gaps, vision_wait_state=vision_wait_state, + fallback_gaps=declared_fallback_gaps, ) if latest_replan_ack_feedback: goal_frontier_projection["replan_ack_feedback"] = ( @@ -1821,6 +1836,7 @@ def build_goal_frontier_projection( acceptance_gaps: list[dict[str, Any]] | None = None, deferred_successors: dict[str, Any] | None = None, vision_wait_state: dict[str, Any] | None = None, + fallback_gaps: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: replan_required = autonomous_replan_is_required(replan_obligation) blockers: list[str] = [] @@ -1883,6 +1899,11 @@ def build_goal_frontier_projection( } if vision_continuation_audit: projection["vision_continuation_audit"] = vision_continuation_audit + # Advisory-only field: unlike acceptance_gaps it is never cleared by the + # blocked-successor wait state, which is exactly when a declared fallback + # would otherwise disappear silently. + if fallback_gaps: + projection["fallback_gaps"] = fallback_gaps[:1] if isinstance(vision_wait_state, dict): projection["vision_wait_state"] = vision_wait_state if replan_required and isinstance(replan_obligation, dict): diff --git a/loopx/control_plane/goals/goal_frontier/fallback_disposition.py b/loopx/control_plane/goals/goal_frontier/fallback_disposition.py new file mode 100644 index 0000000000..298b76c7f2 --- /dev/null +++ b/loopx/control_plane/goals/goal_frontier/fallback_disposition.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +import re +from typing import Any + +from ...todos.contract import ( + TODO_TASK_CLASS_ADVANCEMENT, + normalize_todo_id, + normalize_todo_task_class, +) +from ...todos.deferred_resume import todo_summary_blocked_successor_items +from ..goal_vision_state import goal_vision_state_is_closed + +VISION_FALLBACK_GAP_TRIGGER = "vision_fallback_unresolved" +VISION_FALLBACK_GAP_SCHEMA_VERSION = "goal_fallback_gap_v0" +VISION_FALLBACK_GAP_REASON_CODE = "declared_fallback_without_runnable_or_terminal" +DECLARED_FALLBACK_PATTERN = re.compile(r"\bfallback\b", re.IGNORECASE) +VISION_FALLBACK_TERMINAL_PATH_OUTCOME = "stop" +# Mirror of VISION_FRONTIER_TODO_DELTA_ACTIONS in goal_frontier.__init__ +# (kept local to avoid a circular import from the package root). +VISION_FALLBACK_TODO_DELTA_ACTIONS = frozenset( + {"activate", "create", "reopen", "resume", "retain"} +) +VISION_FALLBACK_SUCCESSOR_DELTA_ACTIONS = frozenset({"create", "reopen"}) +VISION_FALLBACK_RUNNABLE_ITEM_SLOTS = ( + "executable_backlog_items", + "first_executable_items", + "backlog_items", + "unclaimed_priority_open_items", + "claimed_advancement_open_items", +) +VISION_FALLBACK_RUNNABLE_ITEM_LIMIT = 3 +VISION_FALLBACK_RECOMMENDED_ACTION = ( + "resolve the declared fallback direction: link or retain a runnable " + "successor Todo referencing it, declare a bounded create/reopen " + "successor, or record an explicit terminal no-follow-up disposition; " + "do not invent a user gate" +) + + +def _compact_text(value: Any, *, limit: int) -> str: + return " ".join(str(value or "").strip().split())[:limit] + + +def _declares_fallback_direction(agent_vision: dict[str, Any]) -> bool: + patch = agent_vision.get("vision_patch") + patch = patch if isinstance(patch, dict) else {} + return any( + DECLARED_FALLBACK_PATTERN.search(str(field or "")) + for field in ( + patch.get("acceptance_summary"), + patch.get("vision_summary"), + agent_vision.get("vision_summary"), + ) + ) + + +def _vision_todo_delta(agent_vision: dict[str, Any]) -> list[tuple[str, str]]: + delta: list[tuple[str, str]] = [] + for value in agent_vision.get("todo_delta") or []: + if not isinstance(value, str): + continue + action, separator, raw_todo_id = value.strip().partition(":") + todo_id = _compact_text(raw_todo_id, limit=120) + if separator and todo_id and action.strip().lower() in ( + VISION_FALLBACK_TODO_DELTA_ACTIONS + ): + delta.append((action.strip().lower(), todo_id)) + return delta + + +def _item_is_runnable_open_advancement(item: dict[str, Any]) -> bool: + if item.get("done") is True: + return False + status = str(item.get("status") or "open").strip().lower() + if status not in {"", "open", "todo", "active", "pending"}: + return False + text = " ".join( + str(value or "") + for value in (item.get("title"), item.get("text")) + if str(value or "").strip() + ) + task_class = normalize_todo_task_class( + item.get("task_class"), + text=text, + action_kind=item.get("action_kind"), + ) + return task_class == TODO_TASK_CLASS_ADVANCEMENT + + +def _summary_runnable_open_todo_ids( + agent_todo_summary: dict[str, Any] | None, +) -> set[str]: + """Collect open advancement Todo ids the frontier can still select.""" + + if not isinstance(agent_todo_summary, dict): + return set() + runnable: set[str] = set() + for slot in VISION_FALLBACK_RUNNABLE_ITEM_SLOTS: + items = agent_todo_summary.get(slot) + if not isinstance(items, list): + continue + for item in items: + if ( + isinstance(item, dict) + and _item_is_runnable_open_advancement(item) + and (todo_id := normalize_todo_id(item.get("todo_id"))) + ): + runnable.add(todo_id) + return runnable + + +def _blocked_primary_waiting( + agent_todo_summary: dict[str, Any] | None, + *, + agent_id: str | None, +) -> bool: + """Reuse the blocked-successor wait scope as the primary-blocked signal.""" + + if not isinstance(agent_todo_summary, dict): + return False + blocker_items = agent_todo_summary.get("current_agent_blocker_items") + if isinstance(blocker_items, list) and blocker_items: + return True + return bool( + todo_summary_blocked_successor_items( + agent_todo_summary, + agent_id=agent_id, + ) + ) + + +def _vision_has_terminal_disposition(agent_vision: dict[str, Any]) -> bool: + """Terminal evidence: closed-family state or path_delta.outcome=stop.""" + + if goal_vision_state_is_closed(agent_vision.get("state")): + return True + path_delta = agent_vision.get("path_delta") + path_delta = path_delta if isinstance(path_delta, dict) else {} + return ( + str(path_delta.get("outcome") or "").strip().lower() + == VISION_FALLBACK_TERMINAL_PATH_OUTCOME + ) + + +def declared_fallback_gap_from_agent_vision( + agent_vision: dict[str, Any] | None, + *, + agent_todo_summary: dict[str, Any] | None, + agent_id: str | None, +) -> dict[str, Any] | None: + """Project one advisory gap for an unresolved declared fallback. + + A fallback direction declared in the goal vision must resolve to one of: + a runnable open Todo referencing it via todo_delta, a bounded successor + declaration (create/reopen action), or an explicit terminal disposition + (closed-family state or path_delta.outcome=stop). When the primary path + is blocked and none of the three holds, the fallback would otherwise + disappear silently behind the blocked-successor wait state, which clears + the ordinary acceptance gaps. This advisory gap stays in the independent + ``fallback_gaps`` projection field and never enters the acceptance-gap + replan stream. + """ + + if not isinstance(agent_vision, dict): + return None + if _vision_has_terminal_disposition(agent_vision): + return None + if not _declares_fallback_direction(agent_vision): + return None + if not _blocked_primary_waiting( + agent_todo_summary, + agent_id=agent_id, + ): + return None + todo_delta = _vision_todo_delta(agent_vision) + referenced_todo_ids = {todo_id for _, todo_id in todo_delta} + if referenced_todo_ids & _summary_runnable_open_todo_ids(agent_todo_summary): + return None + if {action for action, _ in todo_delta} & VISION_FALLBACK_SUCCESSOR_DELTA_ACTIONS: + return None + gap: dict[str, Any] = { + "schema_version": VISION_FALLBACK_GAP_SCHEMA_VERSION, + "kind": VISION_FALLBACK_GAP_TRIGGER, + "source": "latest_agent_vision", + "agent_id": agent_vision.get("agent_id"), + "state": agent_vision.get("state"), + "reason_code": VISION_FALLBACK_GAP_REASON_CODE, + "recommended_action": VISION_FALLBACK_RECOMMENDED_ACTION, + } + unresolved_todo_ids = [ + todo_id + for todo_id in sorted(referenced_todo_ids) + if todo_id + ][:VISION_FALLBACK_RUNNABLE_ITEM_LIMIT] + if unresolved_todo_ids: + gap["unresolved_todo_ids"] = unresolved_todo_ids + generated_at = _compact_text(agent_vision.get("generated_at"), limit=80) + if generated_at: + gap["generated_at"] = generated_at + return {key: value for key, value in gap.items() if value is not None} diff --git a/tests/control_plane/test_goal_frontier_fallback_disposition.py b/tests/control_plane/test_goal_frontier_fallback_disposition.py new file mode 100644 index 0000000000..365014b78e --- /dev/null +++ b/tests/control_plane/test_goal_frontier_fallback_disposition.py @@ -0,0 +1,235 @@ +from __future__ import annotations + +import pytest + +from loopx.control_plane.goals.goal_frontier import ( + build_goal_frontier_projection_context_from_status, +) +from loopx.control_plane.scheduler.execution_context import ( + GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT, +) +from loopx.control_plane.testing.quota_fixtures import ( + quota_status_payload, + quota_todo_item, + quota_todo_summary, +) +from loopx.quota import build_quota_should_run + +GOAL_ID = "vision-fallback-disposition-fixture" +AGENT_ID = "codex-fallback-agent" +PRIMARY_AGENT = "codex-primary-agent" +PREREQ_ID = "todo_primary_prereq" +PRIMARY_WAIT_ID = "todo_primary_successor" +FALLBACK_ID = "todo_declared_fallback" +DECLARED_FALLBACK_ACCEPTANCE = ( + "Deliver the primary successor; if the primary stays blocked, " + "deliver the declared fallback direction instead." +) + + +def _fallback_vision_run( + *, + state: str = "vision_drift_detected", + todo_delta: list[str] | None = None, + acceptance_summary: str = DECLARED_FALLBACK_ACCEPTANCE, + path_outcome: str | None = None, +) -> dict: + agent_vision: dict = { + "schema_version": "goal_vision_replan_contract_v0", + "agent_id": AGENT_ID, + "state": state, + "todo_delta": todo_delta + if todo_delta is not None + else [f"retain:{PRIMARY_WAIT_ID}"], + "vision_patch": { + "acceptance_summary": acceptance_summary, + "replan_trigger_summary": "The primary acceptance remains open.", + "advancement_policy": "repeat_until_closed", + }, + } + if path_outcome is not None: + agent_vision["path_delta"] = {"outcome": path_outcome} + return { + "classification": "vision_fallback_disposition_fixture", + "generated_at": "2026-09-05T00:00:00+00:00", + "agent_id": AGENT_ID, + "progress_scope": "agent_lane", + "agent_vision": agent_vision, + } + + +def _agent_todos(*, fallback_runnable: bool) -> dict: + prereq = quota_todo_item( + todo_id=PREREQ_ID, + index=1, + text="[P0] Complete the primary prerequisite.", + claimed_by=PRIMARY_AGENT, + ) + waiting = quota_todo_item( + todo_id=PRIMARY_WAIT_ID, + index=2, + text="[P0] Resume the primary successor.", + status="deferred", + claimed_by=AGENT_ID, + resume_when=f"todo_done:{PREREQ_ID}", + ) + items = [prereq, waiting] + if fallback_runnable: + items.append( + quota_todo_item( + todo_id=FALLBACK_ID, + index=3, + text="[P1] Deliver the declared fallback direction.", + claimed_by=AGENT_ID, + ) + ) + return quota_todo_summary(items, role="agent") + + +def _status_payload( + *, + fallback_runnable: bool, + latest_runs: list[dict], +) -> dict: + return quota_status_payload( + goal_id=GOAL_ID, + status="active", + recommended_action="Resolve the declared fallback direction.", + agent_todos=_agent_todos(fallback_runnable=fallback_runnable), + coordination={ + "agent_model": "peer_v1", + "registered_agents": [PRIMARY_AGENT, AGENT_ID], + }, + latest_runs=latest_runs, + ) + + +def _frontier_projection(payload: dict) -> dict: + item = payload["attention_queue"]["items"][0] + context = build_goal_frontier_projection_context_from_status( + goal_id=GOAL_ID, + agent_id=AGENT_ID, + status_payload=payload, + item=item, + project_asset=item["project_asset"], + user_todo_summary=item["user_todos"], + agent_todo_summary=item["agent_todos"], + work_lane_contract=None, + neutral_replan_ack_classifications=set(), + registered_agent_ids=[PRIMARY_AGENT, AGENT_ID], + goal_status="active", + ) + return context["goal_frontier_projection"] + + +def test_blocked_primary_with_runnable_fallback_projects_todo_selectable() -> None: + payload = _status_payload( + fallback_runnable=True, + latest_runs=[ + _fallback_vision_run(todo_delta=[f"retain:{FALLBACK_ID}"]) + ], + ) + + frontier = _frontier_projection(payload) + assert "fallback_gaps" not in frontier + assert "vision_wait_state" not in frontier + remaining = frontier["remaining_advancement_frontier"] + assert remaining["current_agent_claimed_advancement_count"] == 1 + assert remaining["unclaimed_advancement_count"] == 0 + + decision = build_quota_should_run( + payload, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + scheduler_execution_context=( + GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT + ), + ) + assert decision["decision"] == "run" + assert decision["should_run"] is True + assert decision["selected_todo"]["todo_id"] == FALLBACK_ID + assert "fallback_gaps" not in decision["goal_frontier_projection"] + + +def test_declared_fallback_without_resolution_projects_single_gap() -> None: + payload = _status_payload( + fallback_runnable=False, + latest_runs=[_fallback_vision_run()], + ) + + 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 + gaps = frontier["fallback_gaps"] + assert len(gaps) == 1 + gap = gaps[0] + assert gap["kind"] == "vision_fallback_unresolved" + assert gap["reason_code"] == "declared_fallback_without_runnable_or_terminal" + assert gap["agent_id"] == AGENT_ID + assert gap["unresolved_todo_ids"] == [PRIMARY_WAIT_ID] + assert "fallback" in gap["recommended_action"] + assert "do not invent a user gate" in gap["recommended_action"] + assert frontier["replan_required"] is False + + +@pytest.mark.parametrize( + ("state", "path_outcome"), + [ + ("no_followup", None), + ("vision_drift_detected", "stop"), + ], + ids=["closed-state", "terminal-path-outcome"], +) +def test_terminal_disposition_closes_fallback_gap_without_regenerating( + state: str, + path_outcome: str | None, +) -> None: + payload = _status_payload( + fallback_runnable=False, + latest_runs=[ + _fallback_vision_run(state=state, path_outcome=path_outcome) + ], + ) + + first = _frontier_projection(payload) + assert "fallback_gaps" not in first + assert first["acceptance_gaps"] == [] + + second = _frontier_projection(payload) + assert "fallback_gaps" not in second + assert second["acceptance_gaps"] == [] + + +def test_declared_bounded_successor_delta_resolves_the_gap() -> None: + payload = _status_payload( + fallback_runnable=False, + latest_runs=[ + _fallback_vision_run(todo_delta=[f"create:{FALLBACK_ID}"]) + ], + ) + + frontier = _frontier_projection(payload) + + assert "fallback_gaps" not in frontier + + +def test_vision_without_declared_fallback_word_projects_no_gap() -> None: + payload = _status_payload( + fallback_runnable=False, + latest_runs=[ + _fallback_vision_run( + acceptance_summary=( + "Deliver the primary successor after its prerequisite clears." + ) + ) + ], + ) + + frontier = _frontier_projection(payload) + + assert "fallback_gaps" not in frontier From 022263c218189c2ef5140b6d56bfd9ba68c001b9 Mon Sep 17 00:00:00 2001 From: now-ing Date: Sun, 6 Sep 2026 12:52:45 +0800 Subject: [PATCH 2/7] fix(goal): align the fallback gap shape with sibling gap records Drop the schema_version field from the vision fallback gap: the sibling acceptance-gap records in the same projection carry only their typed kind, and the repo's versioned schemas all live under the loopx_ prefix, so a standalone goal_fallback_gap_v0 constant deviated twice for no consumer. Signed-off-by: now-ing --- loopx/control_plane/goals/goal_frontier/fallback_disposition.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/loopx/control_plane/goals/goal_frontier/fallback_disposition.py b/loopx/control_plane/goals/goal_frontier/fallback_disposition.py index 298b76c7f2..0215b7dfe8 100644 --- a/loopx/control_plane/goals/goal_frontier/fallback_disposition.py +++ b/loopx/control_plane/goals/goal_frontier/fallback_disposition.py @@ -12,7 +12,6 @@ from ..goal_vision_state import goal_vision_state_is_closed VISION_FALLBACK_GAP_TRIGGER = "vision_fallback_unresolved" -VISION_FALLBACK_GAP_SCHEMA_VERSION = "goal_fallback_gap_v0" VISION_FALLBACK_GAP_REASON_CODE = "declared_fallback_without_runnable_or_terminal" DECLARED_FALLBACK_PATTERN = re.compile(r"\bfallback\b", re.IGNORECASE) VISION_FALLBACK_TERMINAL_PATH_OUTCOME = "stop" @@ -180,7 +179,6 @@ def declared_fallback_gap_from_agent_vision( if {action for action, _ in todo_delta} & VISION_FALLBACK_SUCCESSOR_DELTA_ACTIONS: return None gap: dict[str, Any] = { - "schema_version": VISION_FALLBACK_GAP_SCHEMA_VERSION, "kind": VISION_FALLBACK_GAP_TRIGGER, "source": "latest_agent_vision", "agent_id": agent_vision.get("agent_id"), From d7985ff38735363531ecb51ea640706731570a6a Mon Sep 17 00:00:00 2001 From: now-ing Date: Sun, 6 Sep 2026 18:00:25 +0800 Subject: [PATCH 3/7] fix(goal): read fallback declarations from structured todo_delta linkage The prose fallback-word matcher invented a gap for explicit negations (No fallback is authorized) and missed non-English declarations. A declaration now requires a structured todo_delta linkage entry (activate/resume/retain) outside the blocked-successor wait scope; prose text alone never projects a gap. Completion evidence now reuses the authoritative agent-scoped selectable advancement frontier (slot order and item predicates of todo_advancement_frontier_counts), so a peer-claimed primary prerequisite retained in todo_delta no longer suppresses the missing-fallback warning, and linked non-advancement work does not count as resolution. Converge the todo_delta action constants and the entry parser into fallback_disposition as the single owner, re-exported by the package root, removing the duplicated sibling constant sets. Signed-off-by: now-ing --- .../goals/goal_frontier/__init__.py | 22 +- .../goal_frontier/fallback_disposition.py | 210 +++++++++++------- ...test_goal_frontier_fallback_disposition.py | 167 +++++++++++--- 3 files changed, 274 insertions(+), 125 deletions(-) diff --git a/loopx/control_plane/goals/goal_frontier/__init__.py b/loopx/control_plane/goals/goal_frontier/__init__.py index 4075b56405..e7470309b9 100644 --- a/loopx/control_plane/goals/goal_frontier/__init__.py +++ b/loopx/control_plane/goals/goal_frontier/__init__.py @@ -47,7 +47,12 @@ autonomous_replan_ack_satisfies_obligation, replan_successor_transition_ack, ) -from .fallback_disposition import declared_fallback_gap_from_agent_vision +from .fallback_disposition import ( + VISION_FRONTIER_TODO_DELTA_ACTIONS, # noqa: F401 + agent_scoped_selectable_advancement_todo_ids, # noqa: F401 + declared_fallback_gap_from_agent_vision, + parse_vision_todo_delta_entries, +) from .long_todo_chain import ( LONG_TODO_CHAIN_TRIGGER, classify_long_todo_chain_ack, @@ -92,13 +97,6 @@ TODO_SUCCESSION_GAP_TRIGGER = TODO_SUCCESSION_WARNING_REASON_CODE TODO_TASK_CLASS_ADVANCEMENT = "advancement_task" TODO_TASK_CLASS_MONITOR = "continuous_monitor" -VISION_FRONTIER_TODO_DELTA_ACTIONS = { - "activate", - "create", - "reopen", - "resume", - "retain", -} def safe_non_negative_int(value: Any) -> int: @@ -394,11 +392,9 @@ def acceptance_gaps_from_agent_vision( gap["acceptance_summary"] = acceptance vision_todo_ids = [ todo_id - for value in (agent_vision.get("todo_delta") or []) - if isinstance(value, str) - and (parts := value.strip().partition(":"))[1] - and parts[0].strip().lower() in VISION_FRONTIER_TODO_DELTA_ACTIONS - and (todo_id := _compact_projection_text(parts[2], limit=120)) + 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)) diff --git a/loopx/control_plane/goals/goal_frontier/fallback_disposition.py b/loopx/control_plane/goals/goal_frontier/fallback_disposition.py index 0215b7dfe8..e5fad28944 100644 --- a/loopx/control_plane/goals/goal_frontier/fallback_disposition.py +++ b/loopx/control_plane/goals/goal_frontier/fallback_disposition.py @@ -1,33 +1,36 @@ from __future__ import annotations -import re from typing import Any from ...todos.contract import ( TODO_TASK_CLASS_ADVANCEMENT, + normalize_todo_claimed_by, normalize_todo_id, - normalize_todo_task_class, ) from ...todos.deferred_resume import todo_summary_blocked_successor_items +from ...todos.projection import ( + todo_item_excludes_agent, + todo_item_is_actionable_open, + todo_item_task_class, +) from ..goal_vision_state import goal_vision_state_is_closed -VISION_FALLBACK_GAP_TRIGGER = "vision_fallback_unresolved" -VISION_FALLBACK_GAP_REASON_CODE = "declared_fallback_without_runnable_or_terminal" -DECLARED_FALLBACK_PATTERN = re.compile(r"\bfallback\b", re.IGNORECASE) -VISION_FALLBACK_TERMINAL_PATH_OUTCOME = "stop" -# Mirror of VISION_FRONTIER_TODO_DELTA_ACTIONS in goal_frontier.__init__ -# (kept local to avoid a circular import from the package root). -VISION_FALLBACK_TODO_DELTA_ACTIONS = frozenset( +# 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"} ) -VISION_FALLBACK_SUCCESSOR_DELTA_ACTIONS = frozenset({"create", "reopen"}) -VISION_FALLBACK_RUNNABLE_ITEM_SLOTS = ( - "executable_backlog_items", - "first_executable_items", - "backlog_items", - "unclaimed_priority_open_items", - "claimed_advancement_open_items", +# create/reopen entries are bounded successor declarations and resolve the +# fallback disposition on their own; activate/resume/retain entries only link +# the vision to existing Todos and still need a selectable frontier match. +VISION_TODO_DELTA_SUCCESSOR_ACTIONS = frozenset({"create", "reopen"}) +VISION_TODO_DELTA_LINKAGE_ACTIONS = frozenset( + VISION_FRONTIER_TODO_DELTA_ACTIONS - VISION_TODO_DELTA_SUCCESSOR_ACTIONS ) +VISION_TODO_DELTA_ID_LIMIT = 120 +VISION_FALLBACK_GAP_TRIGGER = "vision_fallback_unresolved" +VISION_FALLBACK_GAP_REASON_CODE = "declared_fallback_without_runnable_or_terminal" +VISION_FALLBACK_TERMINAL_PATH_OUTCOME = "stop" VISION_FALLBACK_RUNNABLE_ITEM_LIMIT = 3 VISION_FALLBACK_RECOMMENDED_ACTION = ( "resolve the declared fallback direction: link or retain a runnable " @@ -41,72 +44,92 @@ def _compact_text(value: Any, *, limit: int) -> str: return " ".join(str(value or "").strip().split())[:limit] -def _declares_fallback_direction(agent_vision: dict[str, Any]) -> bool: - patch = agent_vision.get("vision_patch") - patch = patch if isinstance(patch, dict) else {} - return any( - DECLARED_FALLBACK_PATTERN.search(str(field or "")) - for field in ( - patch.get("acceptance_summary"), - patch.get("vision_summary"), - agent_vision.get("vision_summary"), - ) - ) +def parse_vision_todo_delta_entries(entries: Any) -> list[tuple[str, str]]: + """Parse ``action:todo_id`` vision todo_delta entries once for consumers.""" - -def _vision_todo_delta(agent_vision: dict[str, Any]) -> list[tuple[str, str]]: - delta: list[tuple[str, str]] = [] - for value in agent_vision.get("todo_delta") or []: + 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=120) - if separator and todo_id and action.strip().lower() in ( - VISION_FALLBACK_TODO_DELTA_ACTIONS + 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) ): - delta.append((action.strip().lower(), todo_id)) - return delta - - -def _item_is_runnable_open_advancement(item: dict[str, Any]) -> bool: - if item.get("done") is True: - return False - status = str(item.get("status") or "open").strip().lower() - if status not in {"", "open", "todo", "active", "pending"}: - return False - text = " ".join( - str(value or "") - for value in (item.get("title"), item.get("text")) - if str(value or "").strip() - ) - task_class = normalize_todo_task_class( - item.get("task_class"), - text=text, - action_kind=item.get("action_kind"), - ) - return task_class == TODO_TASK_CLASS_ADVANCEMENT + parsed.append((normalized_action, todo_id)) + return parsed -def _summary_runnable_open_todo_ids( +def agent_scoped_selectable_advancement_todo_ids( agent_todo_summary: dict[str, Any] | None, + *, + agent_id: str | None, ) -> set[str]: - """Collect open advancement Todo ids the frontier can still select.""" + """Return the ids the agent-scoped selectable advancement frontier holds. + + Mirrors the slot order and item predicates of the authoritative + ``todo_advancement_frontier_counts`` counter (executable backlog first, + unclaimed priority plus agent-claimed advancement items as the slotless + fallback), including claim ownership: peer-claimed advancement work is + intentionally not part of this agent's selectable frontier. + """ if not isinstance(agent_todo_summary, dict): return set() - runnable: set[str] = set() - for slot in VISION_FALLBACK_RUNNABLE_ITEM_SLOTS: - items = agent_todo_summary.get(slot) + normalized_agent_id = normalize_todo_claimed_by(agent_id) + executable_items = agent_todo_summary.get("executable_backlog_items") + if isinstance(executable_items, list): + slots: tuple[Any, ...] = (executable_items,) + else: + slots = ( + agent_todo_summary.get("unclaimed_priority_open_items"), + agent_todo_summary.get("claimed_advancement_open_items"), + ) + selectable: set[str] = set() + for items in slots: if not isinstance(items, list): continue for item in items: - if ( - isinstance(item, dict) - and _item_is_runnable_open_advancement(item) - and (todo_id := normalize_todo_id(item.get("todo_id"))) - ): - runnable.add(todo_id) - return runnable + if not isinstance(item, dict): + continue + if not todo_item_is_actionable_open(item): + continue + if todo_item_task_class(item) != TODO_TASK_CLASS_ADVANCEMENT: + continue + claimed_by = normalize_todo_claimed_by(item.get("claimed_by")) + if normalized_agent_id and claimed_by and claimed_by != normalized_agent_id: + continue + if todo_item_excludes_agent(item, agent_id=normalized_agent_id): + continue + if todo_id := normalize_todo_id(item.get("todo_id")): + selectable.add(todo_id) + return selectable + + +def _blocked_successor_todo_ids( + agent_todo_summary: dict[str, Any] | None, + *, + agent_id: str | None, +) -> set[str]: + """Ids the blocked-successor wait state itself is waiting on.""" + + if not isinstance(agent_todo_summary, dict): + return set() + return { + todo_id + for todo_id in ( + normalize_todo_id(item.get("todo_id")) + for item in todo_summary_blocked_successor_items( + agent_todo_summary, + agent_id=agent_id, + ) + if isinstance(item, dict) + ) + if todo_id + } def _blocked_primary_waiting( @@ -150,33 +173,52 @@ def declared_fallback_gap_from_agent_vision( ) -> dict[str, Any] | None: """Project one advisory gap for an unresolved declared fallback. - A fallback direction declared in the goal vision must resolve to one of: - a runnable open Todo referencing it via todo_delta, a bounded successor - declaration (create/reopen action), or an explicit terminal disposition - (closed-family state or path_delta.outcome=stop). When the primary path - is blocked and none of the three holds, the fallback would otherwise - disappear silently behind the blocked-successor wait state, which clears - the ordinary acceptance gaps. This advisory gap stays in the independent - ``fallback_gaps`` projection field and never enters the acceptance-gap - replan stream. + A fallback direction is declared structurally: the vision's todo_delta + links it to existing Todos via activate/resume/retain entries. Prose + mentions never declare a fallback, so negated or non-English vision text + cannot invent a gap. The declared direction is resolved when one of: + a linked Todo sits on the authoritative agent-scoped selectable + advancement frontier (peer-claimed primary-path Todos do not), the + todo_delta declares a bounded create/reopen successor, or the vision + records an explicit terminal disposition (closed-family state or + path_delta.outcome=stop). The primary path's own blocked successors are + the wait state itself, not fallback declarations, and are excluded. + + When the primary path is blocked and none of the resolutions holds, the + declared fallback would otherwise disappear silently behind the + blocked-successor wait state, which clears the ordinary acceptance gaps. + This advisory gap stays in the independent ``fallback_gaps`` projection + field and never enters the acceptance-gap replan stream. """ if not isinstance(agent_vision, dict): return None if _vision_has_terminal_disposition(agent_vision): return None - if not _declares_fallback_direction(agent_vision): - return None if not _blocked_primary_waiting( agent_todo_summary, agent_id=agent_id, ): return None - todo_delta = _vision_todo_delta(agent_vision) - referenced_todo_ids = {todo_id for _, todo_id in todo_delta} - if referenced_todo_ids & _summary_runnable_open_todo_ids(agent_todo_summary): + todo_delta = parse_vision_todo_delta_entries(agent_vision.get("todo_delta")) + if {action for action, _ in todo_delta} & VISION_TODO_DELTA_SUCCESSOR_ACTIONS: return None - if {action for action, _ in todo_delta} & VISION_FALLBACK_SUCCESSOR_DELTA_ACTIONS: + waiting_todo_ids = _blocked_successor_todo_ids( + agent_todo_summary, + agent_id=agent_id, + ) + declared_linkage_todo_ids = { + todo_id + for action, todo_id in todo_delta + if action in VISION_TODO_DELTA_LINKAGE_ACTIONS + and todo_id not in waiting_todo_ids + } + if not declared_linkage_todo_ids: + return None + if declared_linkage_todo_ids & agent_scoped_selectable_advancement_todo_ids( + agent_todo_summary, + agent_id=agent_id, + ): return None gap: dict[str, Any] = { "kind": VISION_FALLBACK_GAP_TRIGGER, @@ -187,9 +229,7 @@ def declared_fallback_gap_from_agent_vision( "recommended_action": VISION_FALLBACK_RECOMMENDED_ACTION, } unresolved_todo_ids = [ - todo_id - for todo_id in sorted(referenced_todo_ids) - if todo_id + todo_id for todo_id in sorted(declared_linkage_todo_ids) if todo_id ][:VISION_FALLBACK_RUNNABLE_ITEM_LIMIT] if unresolved_todo_ids: gap["unresolved_todo_ids"] = unresolved_todo_ids diff --git a/tests/control_plane/test_goal_frontier_fallback_disposition.py b/tests/control_plane/test_goal_frontier_fallback_disposition.py index 365014b78e..3ca89613b4 100644 --- a/tests/control_plane/test_goal_frontier_fallback_disposition.py +++ b/tests/control_plane/test_goal_frontier_fallback_disposition.py @@ -3,6 +3,8 @@ import pytest from loopx.control_plane.goals.goal_frontier import ( + VISION_FRONTIER_TODO_DELTA_ACTIONS, + agent_scoped_selectable_advancement_todo_ids, build_goal_frontier_projection_context_from_status, ) from loopx.control_plane.scheduler.execution_context import ( @@ -13,6 +15,7 @@ quota_todo_item, quota_todo_summary, ) +from loopx.control_plane.todos.projection import todo_advancement_frontier_counts from loopx.quota import build_quota_should_run GOAL_ID = "vision-fallback-disposition-fixture" @@ -40,7 +43,7 @@ def _fallback_vision_run( "state": state, "todo_delta": todo_delta if todo_delta is not None - else [f"retain:{PRIMARY_WAIT_ID}"], + else [f"retain:{PRIMARY_WAIT_ID}", f"retain:{FALLBACK_ID}"], "vision_patch": { "acceptance_summary": acceptance_summary, "replan_trigger_summary": "The primary acceptance remains open.", @@ -125,9 +128,7 @@ def _frontier_projection(payload: dict) -> dict: def test_blocked_primary_with_runnable_fallback_projects_todo_selectable() -> None: payload = _status_payload( fallback_runnable=True, - latest_runs=[ - _fallback_vision_run(todo_delta=[f"retain:{FALLBACK_ID}"]) - ], + latest_runs=[_fallback_vision_run(todo_delta=[f"retain:{FALLBACK_ID}"])], ) frontier = _frontier_projection(payload) @@ -141,9 +142,7 @@ def test_blocked_primary_with_runnable_fallback_projects_todo_selectable() -> No payload, goal_id=GOAL_ID, agent_id=AGENT_ID, - scheduler_execution_context=( - GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT - ), + scheduler_execution_context=(GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT), ) assert decision["decision"] == "run" assert decision["should_run"] is True @@ -152,9 +151,15 @@ def test_blocked_primary_with_runnable_fallback_projects_todo_selectable() -> No def test_declared_fallback_without_resolution_projects_single_gap() -> 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( fallback_runnable=False, - latest_runs=[_fallback_vision_run()], + latest_runs=[ + _fallback_vision_run( + todo_delta=[f"retain:{PRIMARY_WAIT_ID}", f"retain:{FALLBACK_ID}"] + ) + ], ) frontier = _frontier_projection(payload) @@ -171,12 +176,106 @@ def test_declared_fallback_without_resolution_projects_single_gap() -> None: assert gap["kind"] == "vision_fallback_unresolved" assert gap["reason_code"] == "declared_fallback_without_runnable_or_terminal" assert gap["agent_id"] == AGENT_ID - assert gap["unresolved_todo_ids"] == [PRIMARY_WAIT_ID] + 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 +def test_retaining_only_the_blocked_primary_successor_is_no_declaration() -> None: + # The primary successor is the wait state's own object; retaining it does + # not declare a fallback direction, so no gap is invented. + payload = _status_payload( + fallback_runnable=False, + latest_runs=[_fallback_vision_run(todo_delta=[f"retain:{PRIMARY_WAIT_ID}"])], + ) + + frontier = _frontier_projection(payload) + + assert "fallback_gaps" not in frontier + + +@pytest.mark.parametrize( + "acceptance_summary", + [ + # Owner probe 1: explicit negation must not project a fallback gap. + "No fallback is authorized; wait for the primary prerequisite.", + # Owner probe 2: a non-English prose declaration has no structured + # declaration channel, so the conservative projection yields no gap. + "主路径阻塞时，执行已声明的备用方案。", + # English prose mentioning a fallback is equally non-declarative. + "Deliver the primary successor after its prerequisite clears; " + "the fallback wording lives in prose only.", + ], + ids=["negated-english", "chinese-prose", "english-prose"], +) +def test_prose_text_alone_never_declares_a_fallback( + acceptance_summary: str, +) -> None: + payload = _status_payload( + fallback_runnable=False, + latest_runs=[ + _fallback_vision_run( + acceptance_summary=acceptance_summary, + todo_delta=[f"retain:{PRIMARY_WAIT_ID}"], + ) + ], + ) + + frontier = _frontier_projection(payload) + + assert "fallback_gaps" not in frontier + + +def test_other_agent_primary_todo_does_not_resolve_the_gap() -> None: + # Owner probe 3: the vision retains the deferred primary successor and + # the peer-held primary prerequisite, and there is no fallback Todo at + # all. The peer-claimed prerequisite is not on this agent's selectable + # frontier, so the missing-fallback gap must survive. + payload = _status_payload( + fallback_runnable=False, + latest_runs=[ + _fallback_vision_run( + todo_delta=[f"retain:{PRIMARY_WAIT_ID}", f"retain:{PREREQ_ID}"] + ) + ], + ) + + frontier = _frontier_projection(payload) + + assert frontier["acceptance_gaps"] == [] + gaps = frontier["fallback_gaps"] + assert len(gaps) == 1 + assert gaps[0]["unresolved_todo_ids"] == [PREREQ_ID] + + +def test_declared_fallback_linked_to_monitor_todo_keeps_the_gap() -> None: + # A linked Todo that is not advancement work is not a runnable fallback + # successor, so the declaration stays unresolved. + payload = _status_payload( + fallback_runnable=False, + latest_runs=[_fallback_vision_run(todo_delta=[f"retain:{FALLBACK_ID}"])], + ) + item = payload["attention_queue"]["items"][0] + monitor = quota_todo_item( + todo_id=FALLBACK_ID, + index=3, + text="[P2] Watch the declared fallback direction.", + claimed_by=AGENT_ID, + ) + monitor["task_class"] = "continuous_monitor" + summary = _agent_todos(fallback_runnable=False) + for slot in ("executable_backlog_items", "backlog_items"): + summary[slot] = list(summary.get(slot) or []) + [monitor] + item["agent_todos"] = summary + + frontier = _frontier_projection(payload) + + gaps = frontier["fallback_gaps"] + assert len(gaps) == 1 + assert gaps[0]["unresolved_todo_ids"] == [FALLBACK_ID] + + @pytest.mark.parametrize( ("state", "path_outcome"), [ @@ -191,9 +290,7 @@ def test_terminal_disposition_closes_fallback_gap_without_regenerating( ) -> None: payload = _status_payload( fallback_runnable=False, - latest_runs=[ - _fallback_vision_run(state=state, path_outcome=path_outcome) - ], + latest_runs=[_fallback_vision_run(state=state, path_outcome=path_outcome)], ) first = _frontier_projection(payload) @@ -208,9 +305,7 @@ def test_terminal_disposition_closes_fallback_gap_without_regenerating( def test_declared_bounded_successor_delta_resolves_the_gap() -> None: payload = _status_payload( fallback_runnable=False, - latest_runs=[ - _fallback_vision_run(todo_delta=[f"create:{FALLBACK_ID}"]) - ], + latest_runs=[_fallback_vision_run(todo_delta=[f"create:{FALLBACK_ID}"])], ) frontier = _frontier_projection(payload) @@ -218,18 +313,36 @@ def test_declared_bounded_successor_delta_resolves_the_gap() -> None: assert "fallback_gaps" not in frontier -def test_vision_without_declared_fallback_word_projects_no_gap() -> None: - payload = _status_payload( - fallback_runnable=False, - latest_runs=[ - _fallback_vision_run( - acceptance_summary=( - "Deliver the primary successor after its prerequisite clears." - ) - ) - ], +def test_selectable_frontier_ids_mirror_the_authoritative_counts() -> None: + # The completion-evidence id set must be the same agent-scoped frontier + # the authoritative advancement counter projects. + summary = _agent_todos(fallback_runnable=True) + peer_only = quota_todo_item( + todo_id="todo_peer_owned_direction", + index=4, + text="[P1] Advance the peer-owned direction.", + claimed_by=PRIMARY_AGENT, ) + for slot in ("executable_backlog_items", "backlog_items"): + summary[slot] = list(summary.get(slot) or []) + [peer_only] - frontier = _frontier_projection(payload) + selectable_ids = agent_scoped_selectable_advancement_todo_ids( + summary, + agent_id=AGENT_ID, + ) + counts = todo_advancement_frontier_counts(summary, agent_id=AGENT_ID) - assert "fallback_gaps" not in frontier + assert selectable_ids == {FALLBACK_ID} + assert counts["current_agent_claimed_advancement_count"] == 1 + assert counts["unclaimed_advancement_count"] == 0 + assert counts["other_agent_claimed_advancement_count"] == 2 + assert PREREQ_ID not in selectable_ids + assert PRIMARY_WAIT_ID not in selectable_ids + + +def test_vision_todo_delta_actions_contract_stays_the_shared_owner() -> None: + # Both the acceptance-gap projection and the fallback disposition must + # consume one action contract; create/reopen stay the successor subset. + assert VISION_FRONTIER_TODO_DELTA_ACTIONS == frozenset( + {"activate", "create", "reopen", "resume", "retain"} + ) From 1e0e20da0684fef09518ba842eb4e385dcb1176e Mon Sep 17 00:00:00 2001 From: now-ing Date: Sun, 6 Sep 2026 23:48:19 +0800 Subject: [PATCH 4/7] fix(goal): type fallback declaration-to-successor relations and share frontier items Add a structured fallback declaration-to-successor contract to resolve fallback directions without guessing from generic todo_delta actions. An unrelated create/reopen action no longer suppresses declared fallback gaps, and generic primary-path retain actions no longer trigger false positive fallback warnings. Derive both agent_scoped_selectable_advancement_todo_ids and todo_advancement_frontier_counts from a single authoritative helper todo_advancement_frontier_items in projection.py, eliminating duplicate precedence and predicate logic. Signed-off-by: now-ing --- .../goals/goal_frontier/__init__.py | 2 + .../goal_frontier/fallback_disposition.py | 274 +++++++++++++----- .../goals/goal_frontier/semantic_history.py | 4 + loopx/control_plane/todos/projection.py | 191 ++++++++---- ...test_goal_frontier_fallback_disposition.py | 138 ++++++++- 5 files changed, 475 insertions(+), 134 deletions(-) diff --git a/loopx/control_plane/goals/goal_frontier/__init__.py b/loopx/control_plane/goals/goal_frontier/__init__.py index e7470309b9..74c7146a7c 100644 --- a/loopx/control_plane/goals/goal_frontier/__init__.py +++ b/loopx/control_plane/goals/goal_frontier/__init__.py @@ -49,8 +49,10 @@ ) from .fallback_disposition import ( VISION_FRONTIER_TODO_DELTA_ACTIONS, # noqa: F401 + FallbackDeclaration, # noqa: F401 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 ( diff --git a/loopx/control_plane/goals/goal_frontier/fallback_disposition.py b/loopx/control_plane/goals/goal_frontier/fallback_disposition.py index e5fad28944..7bf6b382d8 100644 --- a/loopx/control_plane/goals/goal_frontier/fallback_disposition.py +++ b/loopx/control_plane/goals/goal_frontier/fallback_disposition.py @@ -1,17 +1,14 @@ from __future__ import annotations +from dataclasses import dataclass from typing import Any from ...todos.contract import ( - TODO_TASK_CLASS_ADVANCEMENT, - normalize_todo_claimed_by, normalize_todo_id, ) from ...todos.deferred_resume import todo_summary_blocked_successor_items from ...todos.projection import ( - todo_item_excludes_agent, - todo_item_is_actionable_open, - todo_item_task_class, + agent_scoped_selectable_advancement_todo_ids, ) from ..goal_vision_state import goal_vision_state_is_closed @@ -40,6 +37,31 @@ ) +@dataclass(frozen=True) +class FallbackDeclaration: + """Structured declaration of a fallback direction and its associated work.""" + + declaration_id: str + target_todo_id: str | None = None + successor_todo_id: str | None = None + + @property + def candidate_todo_ids(self) -> set[str]: + return { + todo_id + for todo_id in ( + self.target_todo_id, + self.successor_todo_id, + self.declaration_id, + ) + if todo_id + } + + @property + def unresolved_todo_id(self) -> str: + return self.target_todo_id or self.declaration_id + + def _compact_text(value: Any, *, limit: int) -> str: return " ".join(str(value or "").strip().split())[:limit] @@ -63,50 +85,134 @@ def parse_vision_todo_delta_entries(entries: Any) -> list[tuple[str, str]]: return parsed -def agent_scoped_selectable_advancement_todo_ids( - agent_todo_summary: dict[str, Any] | None, - *, - agent_id: str | None, -) -> set[str]: - """Return the ids the agent-scoped selectable advancement frontier holds. +def parse_fallback_declarations( + agent_vision: dict[str, Any] | None, + agent_todo_summary: dict[str, Any] | None = None, +) -> list[FallbackDeclaration]: + """Extract structured fallback declarations from vision and linkage contracts.""" - Mirrors the slot order and item predicates of the authoritative - ``todo_advancement_frontier_counts`` counter (executable backlog first, - unclaimed priority plus agent-claimed advancement items as the slotless - fallback), including claim ownership: peer-claimed advancement work is - intentionally not part of this agent's selectable frontier. - """ + declarations: list[FallbackDeclaration] = [] + seen: set[tuple[str, str | None, str | None]] = set() - if not isinstance(agent_todo_summary, dict): - return set() - normalized_agent_id = normalize_todo_claimed_by(agent_id) - executable_items = agent_todo_summary.get("executable_backlog_items") - if isinstance(executable_items, list): - slots: tuple[Any, ...] = (executable_items,) - else: - slots = ( - agent_todo_summary.get("unclaimed_priority_open_items"), - agent_todo_summary.get("claimed_advancement_open_items"), - ) - selectable: set[str] = set() - for items in slots: - if not isinstance(items, list): - continue - for item in items: - if not isinstance(item, dict): - continue - if not todo_item_is_actionable_open(item): - continue - if todo_item_task_class(item) != TODO_TASK_CLASS_ADVANCEMENT: - continue - claimed_by = normalize_todo_claimed_by(item.get("claimed_by")) - if normalized_agent_id and claimed_by and claimed_by != normalized_agent_id: + def add_declaration( + declaration_id: str, + target_todo_id: str | None = None, + successor_todo_id: str | None = None, + ) -> None: + key = (declaration_id, target_todo_id, successor_todo_id) + if key not in seen: + seen.add(key) + declarations.append( + FallbackDeclaration( + declaration_id=declaration_id, + target_todo_id=target_todo_id, + successor_todo_id=successor_todo_id, + ) + ) + + if isinstance(agent_vision, dict): + patch = agent_vision.get("vision_patch") + patch = patch if isinstance(patch, dict) else {} + for source in ( + agent_vision.get("fallback_declarations"), + agent_vision.get("fallback_relationships"), + agent_vision.get("fallbacks"), + patch.get("fallback_declarations"), + patch.get("fallback_relationships"), + patch.get("fallbacks"), + ): + if not isinstance(source, list): continue - if todo_item_excludes_agent(item, agent_id=normalized_agent_id): + for raw in source: + if isinstance(raw, dict): + raw_decl_id = ( + raw.get("declaration_id") + or raw.get("fallback_id") + or raw.get("id") + or raw.get("name") + or raw.get("fallback_todo_id") + or raw.get("todo_id") + ) + declaration_id = _compact_text( + raw_decl_id, + limit=VISION_TODO_DELTA_ID_LIMIT, + ) + target_todo_id = normalize_todo_id( + raw.get("target_todo_id") + or raw.get("fallback_todo_id") + or raw.get("todo_id") + ) + successor_todo_id = normalize_todo_id( + raw.get("successor_todo_id") or raw.get("successor_id") + ) + if not declaration_id and target_todo_id: + declaration_id = target_todo_id + if not target_todo_id and not successor_todo_id and declaration_id: + target_todo_id = normalize_todo_id(declaration_id) + if declaration_id: + add_declaration( + declaration_id=declaration_id, + target_todo_id=target_todo_id, + successor_todo_id=successor_todo_id, + ) + elif isinstance(raw, str): + text = _compact_text(raw, limit=VISION_TODO_DELTA_ID_LIMIT) + if not text: + continue + if "->" in text: + decl, _, succ = text.partition("->") + d_id = decl.strip() + s_id = normalize_todo_id(succ.strip()) + add_declaration( + declaration_id=d_id, + target_todo_id=normalize_todo_id(d_id), + successor_todo_id=s_id, + ) + elif ":" in text and not text.startswith(("todo_", "task_")): + decl, _, succ = text.partition(":") + d_id = decl.strip() + s_id = normalize_todo_id(succ.strip()) + add_declaration( + declaration_id=d_id, + target_todo_id=normalize_todo_id(d_id), + successor_todo_id=s_id, + ) + else: + t_id = normalize_todo_id(text) or text + add_declaration( + declaration_id=text, + target_todo_id=t_id, + successor_todo_id=t_id, + ) + + # Check linkage contract on blocked successor items in agent_todo_summary + if isinstance(agent_todo_summary, dict): + for slot in ( + "deferred_items", + "deferred_resume_candidates", + "current_agent_blocker_items", + ): + items = agent_todo_summary.get(slot) + if not isinstance(items, list): continue - if todo_id := normalize_todo_id(item.get("todo_id")): - selectable.add(todo_id) - return selectable + for item in items: + if not isinstance(item, dict): + continue + if fallback_todo_id := normalize_todo_id(item.get("fallback_todo_id")): + add_declaration( + declaration_id=fallback_todo_id, + target_todo_id=fallback_todo_id, + successor_todo_id=fallback_todo_id, + ) + if fallback_succ_id := normalize_todo_id( + item.get("fallback_successor_todo_id") + ): + add_declaration( + declaration_id=fallback_succ_id, + successor_todo_id=fallback_succ_id, + ) + + return declarations def _blocked_successor_todo_ids( @@ -173,16 +279,19 @@ def declared_fallback_gap_from_agent_vision( ) -> dict[str, Any] | None: """Project one advisory gap for an unresolved declared fallback. - A fallback direction is declared structurally: the vision's todo_delta - links it to existing Todos via activate/resume/retain entries. Prose - mentions never declare a fallback, so negated or non-English vision text - cannot invent a gap. The declared direction is resolved when one of: - a linked Todo sits on the authoritative agent-scoped selectable - advancement frontier (peer-claimed primary-path Todos do not), the - todo_delta declares a bounded create/reopen successor, or the vision - records an explicit terminal disposition (closed-family state or - path_delta.outcome=stop). The primary path's own blocked successors are - the wait state itself, not fallback declarations, and are excluded. + A fallback direction is declared structurally via the agent vision's + ``fallback_declarations`` / ``fallback_relationships`` contract or the + task linkage contract on blocked successor items. Prose mentions never + declare a fallback, and generic ``todo_delta`` actions are not fallback + declarations on their own. + + The declared direction is resolved when one of: + 1. A linked Todo sits on the authoritative agent-scoped selectable + advancement frontier (peer-claimed primary-path Todos do not); + 2. A bounded successor Todo is created or reopened specifically for + this fallback direction; or + 3. The vision records an explicit terminal disposition (closed-family state + or path_delta.outcome=stop). When the primary path is blocked and none of the resolutions holds, the declared fallback would otherwise disappear silently behind the @@ -200,26 +309,53 @@ def declared_fallback_gap_from_agent_vision( agent_id=agent_id, ): return None - todo_delta = parse_vision_todo_delta_entries(agent_vision.get("todo_delta")) - if {action for action, _ in todo_delta} & VISION_TODO_DELTA_SUCCESSOR_ACTIONS: + + declarations = parse_fallback_declarations( + agent_vision, + agent_todo_summary=agent_todo_summary, + ) + if not declarations: return None + + selectable_ids = agent_scoped_selectable_advancement_todo_ids( + agent_todo_summary, + agent_id=agent_id, + ) waiting_todo_ids = _blocked_successor_todo_ids( agent_todo_summary, agent_id=agent_id, ) - declared_linkage_todo_ids = { + todo_delta = parse_vision_todo_delta_entries(agent_vision.get("todo_delta")) + created_or_reopened_ids = { todo_id for action, todo_id in todo_delta - if action in VISION_TODO_DELTA_LINKAGE_ACTIONS - and todo_id not in waiting_todo_ids + if action in VISION_TODO_DELTA_SUCCESSOR_ACTIONS } - if not declared_linkage_todo_ids: - return None - if declared_linkage_todo_ids & agent_scoped_selectable_advancement_todo_ids( - agent_todo_summary, - agent_id=agent_id, - ): + + unresolved_ids: set[str] = set() + for declaration in declarations: + candidate_ids = declaration.candidate_todo_ids - waiting_todo_ids + if not candidate_ids: + continue + # Disposition 1: Runnable on authoritative selectable advancement frontier + if candidate_ids & selectable_ids: + continue + # Disposition 2: Bounded successor created/reopened specifically for this fallback + if candidate_ids & created_or_reopened_ids: + continue + if ( + declaration.successor_todo_id + and declaration.successor_todo_id in created_or_reopened_ids + ): + continue + + unresolved_id = declaration.unresolved_todo_id + if unresolved_id not in waiting_todo_ids: + unresolved_ids.add(unresolved_id) + + if not unresolved_ids: return None + gap: dict[str, Any] = { "kind": VISION_FALLBACK_GAP_TRIGGER, "source": "latest_agent_vision", @@ -228,9 +364,9 @@ def declared_fallback_gap_from_agent_vision( "reason_code": VISION_FALLBACK_GAP_REASON_CODE, "recommended_action": VISION_FALLBACK_RECOMMENDED_ACTION, } - unresolved_todo_ids = [ - todo_id for todo_id in sorted(declared_linkage_todo_ids) if todo_id - ][:VISION_FALLBACK_RUNNABLE_ITEM_LIMIT] + unresolved_todo_ids = [todo_id for todo_id in sorted(unresolved_ids) if todo_id][ + :VISION_FALLBACK_RUNNABLE_ITEM_LIMIT + ] if unresolved_todo_ids: gap["unresolved_todo_ids"] = unresolved_todo_ids generated_at = _compact_text(agent_vision.get("generated_at"), limit=80) diff --git a/loopx/control_plane/goals/goal_frontier/semantic_history.py b/loopx/control_plane/goals/goal_frontier/semantic_history.py index bf92e520ca..66004a81ec 100644 --- a/loopx/control_plane/goals/goal_frontier/semantic_history.py +++ b/loopx/control_plane/goals/goal_frontier/semantic_history.py @@ -167,6 +167,10 @@ def latest_agent_vision_from_runs( } 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"] + if isinstance(vision.get("fallback_relationships"), list): + result["fallback_relationships"] = vision["fallback_relationships"] return result return None diff --git a/loopx/control_plane/todos/projection.py b/loopx/control_plane/todos/projection.py index c6b570ed26..048c0fa04e 100644 --- a/loopx/control_plane/todos/projection.py +++ b/loopx/control_plane/todos/projection.py @@ -149,9 +149,9 @@ def todo_claimed_visibility_items( if len(selected) >= limit: break - return sorted(selected, key=lambda item: original_index.get(id(item), TODO_MISSING_INDEX))[ - :limit - ] + return sorted( + selected, key=lambda item: original_index.get(id(item), TODO_MISSING_INDEX) + )[:limit] def todo_item_task_text( @@ -160,9 +160,7 @@ def todo_item_task_text( keys: tuple[str, ...] = ("title", "text"), ) -> str: return " ".join( - str(item.get(key) or "") - for key in keys - if str(item.get(key) or "").strip() + str(item.get(key) or "") for key in keys if str(item.get(key) or "").strip() ) @@ -197,7 +195,9 @@ def todo_item_expires_at(item: dict[str, Any]) -> datetime | None: return monitor_todo_expires_at(item) -def todo_item_is_expired_monitor(item: dict[str, Any], *, now: datetime | None = None) -> bool: +def todo_item_is_expired_monitor( + item: dict[str, Any], *, now: datetime | None = None +) -> bool: return monitor_todo_is_expired(item, now=now) @@ -245,38 +245,33 @@ def todo_item_claimed_by_agent_or_unclaimed( return not claimed_by or claimed_by == normalized_agent_id -def todo_advancement_frontier_counts( +def todo_advancement_frontier_items( summary: dict[str, Any] | None, *, agent_id: str | None, -) -> dict[str, int]: - """Classify the durable advancement frontier by exact claim ownership.""" - +) -> dict[str, list[dict[str, Any]]]: + """Return the authoritative advancement frontier items grouped by claim ownership. + + Preserves the slot precedence of executable backlog first, falling back to + unclaimed priority and claimed advancement open items when the executable backlog + is omitted. Peer-claimed items are tracked separately and excluded from the current + agent's selectable advancement frontier. + """ + + empty: dict[str, list[dict[str, Any]]] = { + "current_agent_claimed_items": [], + "unclaimed_items": [], + "other_agent_claimed_items": [], + } if not isinstance(summary, dict): - return { - "current_agent_claimed_advancement_count": 0, - "unclaimed_advancement_count": 0, - "other_agent_claimed_advancement_count": 0, - } + return empty + normalized_agent_id = normalize_todo_claimed_by(agent_id) - claim_scope = summary.get("claim_scope") - other_items = ( - claim_scope.get("other_agent_claimed_items") - if isinstance(claim_scope, dict) - else [] - ) - diagnostic_other_count = sum( - 1 - for value in other_items or [] - if isinstance(value, dict) - and todo_item_is_actionable_open(value) - and todo_item_task_class(value) == TODO_TASK_CLASS_ADVANCEMENT - ) executable_items = summary.get("executable_backlog_items") if isinstance(executable_items, list): - current_count = 0 - unclaimed_count = 0 - other_count = 0 + current_items: list[dict[str, Any]] = [] + unclaimed_items: list[dict[str, Any]] = [] + other_items: list[dict[str, Any]] = [] for value in executable_items: if not isinstance(value, dict): continue @@ -287,44 +282,123 @@ def todo_advancement_frontier_counts( claimed_by = normalize_todo_claimed_by(value.get("claimed_by")) if claimed_by: if normalized_agent_id and claimed_by == normalized_agent_id: - if not todo_item_excludes_agent(value, agent_id=normalized_agent_id): - current_count += 1 + if not todo_item_excludes_agent( + value, agent_id=normalized_agent_id + ): + current_items.append(value) elif normalized_agent_id: - other_count += 1 + other_items.append(value) else: - current_count += 1 + current_items.append(value) continue if not todo_item_excludes_agent(value, agent_id=normalized_agent_id): - unclaimed_count += 1 + unclaimed_items.append(value) return { - "current_agent_claimed_advancement_count": max( - current_count, - _positive_int(summary.get("current_agent_claimed_advancement_count")), - ), - "unclaimed_advancement_count": unclaimed_count, - # Agent-scoped executable backlogs intentionally omit peer-owned - # work. Preserve that diagnostic lane from claim_scope without - # letting it contribute to the current Agent's selectable count. - "other_agent_claimed_advancement_count": max( - other_count, - diagnostic_other_count, - ), + "current_agent_claimed_items": current_items, + "unclaimed_items": unclaimed_items, + "other_agent_claimed_items": other_items, } - unclaimed_count = sum( - 1 + unclaimed_items = [ + value for value in summary.get("unclaimed_priority_open_items") or [] if isinstance(value, dict) and todo_item_is_actionable_open(value) and todo_item_task_class(value) == TODO_TASK_CLASS_ADVANCEMENT and not todo_item_excludes_agent(value, agent_id=normalized_agent_id) + ] + current_items = [ + value + for value in summary.get("claimed_advancement_open_items") or [] + if isinstance(value, dict) + and todo_item_is_actionable_open(value) + and todo_item_task_class(value) == TODO_TASK_CLASS_ADVANCEMENT + and ( + not normalized_agent_id + or normalize_todo_claimed_by(value.get("claimed_by")) == normalized_agent_id + ) + and not todo_item_excludes_agent(value, agent_id=normalized_agent_id) + ] + other_items = [ + value + for value in summary.get("claimed_advancement_open_items") or [] + if isinstance(value, dict) + and todo_item_is_actionable_open(value) + and todo_item_task_class(value) == TODO_TASK_CLASS_ADVANCEMENT + and normalized_agent_id + and normalize_todo_claimed_by(value.get("claimed_by")) + and normalize_todo_claimed_by(value.get("claimed_by")) != normalized_agent_id + ] + return { + "current_agent_claimed_items": current_items, + "unclaimed_items": unclaimed_items, + "other_agent_claimed_items": other_items, + } + + +def agent_scoped_selectable_advancement_todo_ids( + agent_todo_summary: dict[str, Any] | None, + *, + agent_id: str | None, +) -> set[str]: + """Return the ids the agent-scoped selectable advancement frontier holds. + + Derived directly from the authoritative ``todo_advancement_frontier_items`` + helper so that slot precedence and claim ownership predicates never diverge + from the frontier counter. + """ + + frontier_items = todo_advancement_frontier_items( + agent_todo_summary, + agent_id=agent_id, + ) + selectable: set[str] = set() + for item in ( + frontier_items["current_agent_claimed_items"] + + frontier_items["unclaimed_items"] + ): + if todo_id := normalize_todo_id(item.get("todo_id")): + selectable.add(todo_id) + return selectable + + +def todo_advancement_frontier_counts( + summary: dict[str, Any] | None, + *, + agent_id: str | None, +) -> dict[str, int]: + """Classify the durable advancement frontier by exact claim ownership.""" + + if not isinstance(summary, dict): + return { + "current_agent_claimed_advancement_count": 0, + "unclaimed_advancement_count": 0, + "other_agent_claimed_advancement_count": 0, + } + frontier_items = todo_advancement_frontier_items(summary, agent_id=agent_id) + claim_scope = summary.get("claim_scope") + other_items = ( + claim_scope.get("other_agent_claimed_items") + if isinstance(claim_scope, dict) + else [] + ) + diagnostic_other_count = sum( + 1 + for value in other_items or [] + if isinstance(value, dict) + and todo_item_is_actionable_open(value) + and todo_item_task_class(value) == TODO_TASK_CLASS_ADVANCEMENT ) return { - "current_agent_claimed_advancement_count": _positive_int( - summary.get("current_agent_claimed_advancement_count") + "current_agent_claimed_advancement_count": max( + len(frontier_items["current_agent_claimed_items"]), + _positive_int(summary.get("current_agent_claimed_advancement_count")), + ), + "unclaimed_advancement_count": len(frontier_items["unclaimed_items"]), + "other_agent_claimed_advancement_count": max( + len(frontier_items["other_agent_claimed_items"]), + diagnostic_other_count, ), - "unclaimed_advancement_count": unclaimed_count, - "other_agent_claimed_advancement_count": diagnostic_other_count, } @@ -344,7 +418,8 @@ def todo_item_excludes_agent( normalized_agent_id = normalize_todo_claimed_by(agent_id) return bool( normalized_agent_id - and normalized_agent_id in normalize_todo_excluded_agents(item.get("excluded_agents")) + and normalized_agent_id + in normalize_todo_excluded_agents(item.get("excluded_agents")) ) @@ -674,7 +749,9 @@ def todo_summary_open_task_counts(summary: dict[str, Any] | None) -> dict[str, i } -def todo_summary_has_only_future_scoped_monitor_work(summary: dict[str, Any] | None) -> bool: +def todo_summary_has_only_future_scoped_monitor_work( + summary: dict[str, Any] | None, +) -> bool: """Return true when the scoped agent has only non-due monitor work left.""" agent_id = todo_summary_claim_scope_agent_id(summary) diff --git a/tests/control_plane/test_goal_frontier_fallback_disposition.py b/tests/control_plane/test_goal_frontier_fallback_disposition.py index 3ca89613b4..ed09c5c9ea 100644 --- a/tests/control_plane/test_goal_frontier_fallback_disposition.py +++ b/tests/control_plane/test_goal_frontier_fallback_disposition.py @@ -1,5 +1,7 @@ from __future__ import annotations +from typing import Any + import pytest from loopx.control_plane.goals.goal_frontier import ( @@ -36,6 +38,7 @@ def _fallback_vision_run( todo_delta: list[str] | None = None, acceptance_summary: str = DECLARED_FALLBACK_ACCEPTANCE, path_outcome: str | None = None, + fallback_declarations: list[Any] | None = None, ) -> dict: agent_vision: dict = { "schema_version": "goal_vision_replan_contract_v0", @@ -43,13 +46,23 @@ def _fallback_vision_run( "state": state, "todo_delta": todo_delta if todo_delta is not None - else [f"retain:{PRIMARY_WAIT_ID}", f"retain:{FALLBACK_ID}"], + else [f"retain:{PRIMARY_WAIT_ID}"], "vision_patch": { "acceptance_summary": acceptance_summary, "replan_trigger_summary": "The primary acceptance remains open.", "advancement_policy": "repeat_until_closed", }, } + if fallback_declarations is not None: + agent_vision["fallback_declarations"] = fallback_declarations + elif acceptance_summary == DECLARED_FALLBACK_ACCEPTANCE: + agent_vision["fallback_declarations"] = [ + { + "declaration_id": "declared_fallback_direction", + "target_todo_id": FALLBACK_ID, + "successor_todo_id": FALLBACK_ID, + } + ] if path_outcome is not None: agent_vision["path_delta"] = {"outcome": path_outcome} return { @@ -187,7 +200,12 @@ def test_retaining_only_the_blocked_primary_successor_is_no_declaration() -> Non # not declare a fallback direction, so no gap is invented. payload = _status_payload( fallback_runnable=False, - latest_runs=[_fallback_vision_run(todo_delta=[f"retain:{PRIMARY_WAIT_ID}"])], + latest_runs=[ + _fallback_vision_run( + todo_delta=[f"retain:{PRIMARY_WAIT_ID}"], + fallback_declarations=[], + ) + ], ) frontier = _frontier_projection(payload) @@ -218,6 +236,27 @@ def test_prose_text_alone_never_declares_a_fallback( _fallback_vision_run( acceptance_summary=acceptance_summary, todo_delta=[f"retain:{PRIMARY_WAIT_ID}"], + fallback_declarations=[], + ) + ], + ) + + frontier = _frontier_projection(payload) + + assert "fallback_gaps" not in frontier + + +def test_other_agent_primary_todo_without_fallback_declaration_generates_no_gap() -> ( + None +): + # Maintainer blocker 2: generic primary-path retain (e.g. peer-held prerequisite) + # is not a fallback declaration; without a structured declaration, no gap is invented. + payload = _status_payload( + fallback_runnable=False, + latest_runs=[ + _fallback_vision_run( + todo_delta=[f"retain:{PRIMARY_WAIT_ID}", f"retain:{PREREQ_ID}"], + fallback_declarations=[], ) ], ) @@ -228,15 +267,21 @@ def test_prose_text_alone_never_declares_a_fallback( def test_other_agent_primary_todo_does_not_resolve_the_gap() -> None: - # Owner probe 3: the vision retains the deferred primary successor and - # the peer-held primary prerequisite, and there is no fallback Todo at - # all. The peer-claimed prerequisite is not on this agent's selectable - # frontier, so the missing-fallback gap must survive. + # A peer-claimed prerequisite retained on the primary path is not on this + # agent's selectable frontier and is not linked to the declared fallback, + # so the declared fallback gap survives for the declared fallback Todo. payload = _status_payload( fallback_runnable=False, latest_runs=[ _fallback_vision_run( - todo_delta=[f"retain:{PRIMARY_WAIT_ID}", f"retain:{PREREQ_ID}"] + todo_delta=[f"retain:{PRIMARY_WAIT_ID}", f"retain:{PREREQ_ID}"], + fallback_declarations=[ + { + "declaration_id": "declared_fallback_direction", + "target_todo_id": FALLBACK_ID, + "successor_todo_id": FALLBACK_ID, + } + ], ) ], ) @@ -246,7 +291,7 @@ def test_other_agent_primary_todo_does_not_resolve_the_gap() -> None: assert frontier["acceptance_gaps"] == [] gaps = frontier["fallback_gaps"] assert len(gaps) == 1 - assert gaps[0]["unresolved_todo_ids"] == [PREREQ_ID] + assert gaps[0]["unresolved_todo_ids"] == [FALLBACK_ID] def test_declared_fallback_linked_to_monitor_todo_keeps_the_gap() -> None: @@ -313,6 +358,83 @@ def test_declared_bounded_successor_delta_resolves_the_gap() -> None: assert "fallback_gaps" not in frontier +def test_unrelated_create_does_not_resolve_fallback_gap() -> None: + # Maintainer blocker 1: An unrelated create/reopen action must not + # resolve the declared fallback direction. + payload = _status_payload( + fallback_runnable=False, + latest_runs=[ + _fallback_vision_run( + todo_delta=[ + f"retain:{PRIMARY_WAIT_ID}", + "create:todo_unrelated_maintenance", + ], + fallback_declarations=[ + { + "declaration_id": "fallback_direction", + "target_todo_id": FALLBACK_ID, + "successor_todo_id": FALLBACK_ID, + } + ], + ) + ], + ) + + frontier = _frontier_projection(payload) + + gaps = frontier["fallback_gaps"] + assert len(gaps) == 1 + assert gaps[0]["unresolved_todo_ids"] == [FALLBACK_ID] + + +def test_fallback_declared_via_todo_linkage_contract_projects_gap() -> None: + # A fallback declared on a blocked successor item via the task linkage + # contract projects an advisory gap when unresolved. + payload = _status_payload( + fallback_runnable=False, + latest_runs=[ + _fallback_vision_run( + todo_delta=[f"retain:{PRIMARY_WAIT_ID}"], + fallback_declarations=[], + ) + ], + ) + item = payload["attention_queue"]["items"][0] + agent_todos = item["agent_todos"] + for deferred in agent_todos.get("deferred_items") or []: + if isinstance(deferred, dict): + deferred["fallback_todo_id"] = FALLBACK_ID + + frontier = _frontier_projection(payload) + + gaps = frontier["fallback_gaps"] + assert len(gaps) == 1 + assert gaps[0]["unresolved_todo_ids"] == [FALLBACK_ID] + + +def test_typed_declaration_successor_relation_resolves_gap() -> None: + # A typed declaration-to-successor relation resolves when its declared + # bounded successor is created in todo_delta. + payload = _status_payload( + fallback_runnable=False, + latest_runs=[ + _fallback_vision_run( + todo_delta=[f"retain:{PRIMARY_WAIT_ID}", "create:todo_typed_successor"], + fallback_declarations=[ + { + "declaration_id": "fallback_direction", + "successor_todo_id": "todo_typed_successor", + } + ], + ) + ], + ) + + frontier = _frontier_projection(payload) + + assert "fallback_gaps" not in frontier + + def test_selectable_frontier_ids_mirror_the_authoritative_counts() -> None: # The completion-evidence id set must be the same agent-scoped frontier # the authoritative advancement counter projects. From 54519bd0089adde8fdaa19940bcf6aeb11e46f37 Mon Sep 17 00:00:00 2001 From: now-ing Date: Mon, 7 Sep 2026 06:09:44 +0800 Subject: [PATCH 5/7] fix(goal): persist typed fallback declarations through the vision write contract The real caller entry dropped the round-3 fallback_declarations: the TS-owned goal.vision_checkpoint prepare whitelisted them away, and the status/shared-runtime compact read model mirrored that loss, so only tests that hand-filled raw run history could project a fallback gap. - prepare now validates and carries one bounded shape: at most four entries, unique non-empty declaration_id, optional target_todo_id and successor_todo_id, each public-safe and 120-char bounded; over-wide, duplicated, unsafe, or id-less packets are rejected explicitly instead of silently losing the declared fallback direction - compact_goal_vision_packet mirrors the same bounds so the persisted declaration survives status and shared-runtime read-model compaction - the frontier parser now consumes only the contract-written shape; the fallback_relationships/fallbacks aliases, arrow/colon string forms, renamed-key fallbacks, and the un-authored todo-summary linkage fields are removed rather than kept as reader-only compatibility - tests exercise the production chain end to end: caller packet through the real TS prepare and compact read model into run history, an independent semantic-history readback, and the quota/frontier projections; alias shapes stay non-declarative Signed-off-by: now-ing --- .../goal_frontier/fallback_disposition.py | 166 +++++------------- .../goals/goal_frontier/semantic_history.py | 2 - loopx/control_plane/goals/goal_vision.py | 50 +++++- .../control_plane/goals/vision_checkpoint.ts | 74 ++++++++ ...test_goal_frontier_fallback_disposition.py | 151 ++++++++++++---- .../vision_checkpoint.test.ts | 115 ++++++++++++ 6 files changed, 395 insertions(+), 163 deletions(-) diff --git a/loopx/control_plane/goals/goal_frontier/fallback_disposition.py b/loopx/control_plane/goals/goal_frontier/fallback_disposition.py index 7bf6b382d8..c27cf0f28c 100644 --- a/loopx/control_plane/goals/goal_frontier/fallback_disposition.py +++ b/loopx/control_plane/goals/goal_frontier/fallback_disposition.py @@ -25,6 +25,8 @@ 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" VISION_FALLBACK_GAP_REASON_CODE = "declared_fallback_without_runnable_or_terminal" VISION_FALLBACK_TERMINAL_PATH_OUTCOME = "stop" @@ -87,131 +89,46 @@ def parse_vision_todo_delta_entries(entries: Any) -> list[tuple[str, str]]: def parse_fallback_declarations( agent_vision: dict[str, Any] | None, - agent_todo_summary: dict[str, Any] | None = None, ) -> list[FallbackDeclaration]: - """Extract structured fallback declarations from vision and linkage contracts.""" + """Parse typed fallback declarations written by the TS Vision contract. + + The only supported authoring path is ``agent_vision.fallback_declarations`` + as validated and persisted by the TS-owned ``goal.vision_checkpoint`` + prepare (and mirrored through the status/shared-runtime compact read + model). Prose mentions, generic ``todo_delta`` actions, and legacy alias + shapes are not declarations. + """ declarations: list[FallbackDeclaration] = [] - seen: set[tuple[str, str | None, str | None]] = set() + if not isinstance(agent_vision, dict): + return declarations + source = agent_vision.get("fallback_declarations") + if not isinstance(source, list): + return declarations - def add_declaration( - declaration_id: str, - target_todo_id: str | None = None, - successor_todo_id: str | None = None, - ) -> None: + seen: set[tuple[str, str | None, str | None]] = set() + for raw in source[:VISION_FALLBACK_DECLARATION_ENTRY_LIMIT]: + if not isinstance(raw, dict): + continue + declaration_id = _compact_text( + raw.get("declaration_id"), + limit=VISION_TODO_DELTA_ID_LIMIT, + ) + if not declaration_id: + continue + target_todo_id = normalize_todo_id(raw.get("target_todo_id")) + successor_todo_id = normalize_todo_id(raw.get("successor_todo_id")) key = (declaration_id, target_todo_id, successor_todo_id) - if key not in seen: - seen.add(key) - declarations.append( - FallbackDeclaration( - declaration_id=declaration_id, - target_todo_id=target_todo_id, - successor_todo_id=successor_todo_id, - ) + if key in seen: + continue + seen.add(key) + declarations.append( + FallbackDeclaration( + declaration_id=declaration_id, + target_todo_id=target_todo_id, + successor_todo_id=successor_todo_id, ) - - if isinstance(agent_vision, dict): - patch = agent_vision.get("vision_patch") - patch = patch if isinstance(patch, dict) else {} - for source in ( - agent_vision.get("fallback_declarations"), - agent_vision.get("fallback_relationships"), - agent_vision.get("fallbacks"), - patch.get("fallback_declarations"), - patch.get("fallback_relationships"), - patch.get("fallbacks"), - ): - if not isinstance(source, list): - continue - for raw in source: - if isinstance(raw, dict): - raw_decl_id = ( - raw.get("declaration_id") - or raw.get("fallback_id") - or raw.get("id") - or raw.get("name") - or raw.get("fallback_todo_id") - or raw.get("todo_id") - ) - declaration_id = _compact_text( - raw_decl_id, - limit=VISION_TODO_DELTA_ID_LIMIT, - ) - target_todo_id = normalize_todo_id( - raw.get("target_todo_id") - or raw.get("fallback_todo_id") - or raw.get("todo_id") - ) - successor_todo_id = normalize_todo_id( - raw.get("successor_todo_id") or raw.get("successor_id") - ) - if not declaration_id and target_todo_id: - declaration_id = target_todo_id - if not target_todo_id and not successor_todo_id and declaration_id: - target_todo_id = normalize_todo_id(declaration_id) - if declaration_id: - add_declaration( - declaration_id=declaration_id, - target_todo_id=target_todo_id, - successor_todo_id=successor_todo_id, - ) - elif isinstance(raw, str): - text = _compact_text(raw, limit=VISION_TODO_DELTA_ID_LIMIT) - if not text: - continue - if "->" in text: - decl, _, succ = text.partition("->") - d_id = decl.strip() - s_id = normalize_todo_id(succ.strip()) - add_declaration( - declaration_id=d_id, - target_todo_id=normalize_todo_id(d_id), - successor_todo_id=s_id, - ) - elif ":" in text and not text.startswith(("todo_", "task_")): - decl, _, succ = text.partition(":") - d_id = decl.strip() - s_id = normalize_todo_id(succ.strip()) - add_declaration( - declaration_id=d_id, - target_todo_id=normalize_todo_id(d_id), - successor_todo_id=s_id, - ) - else: - t_id = normalize_todo_id(text) or text - add_declaration( - declaration_id=text, - target_todo_id=t_id, - successor_todo_id=t_id, - ) - - # Check linkage contract on blocked successor items in agent_todo_summary - if isinstance(agent_todo_summary, dict): - for slot in ( - "deferred_items", - "deferred_resume_candidates", - "current_agent_blocker_items", - ): - items = agent_todo_summary.get(slot) - if not isinstance(items, list): - continue - for item in items: - if not isinstance(item, dict): - continue - if fallback_todo_id := normalize_todo_id(item.get("fallback_todo_id")): - add_declaration( - declaration_id=fallback_todo_id, - target_todo_id=fallback_todo_id, - successor_todo_id=fallback_todo_id, - ) - if fallback_succ_id := normalize_todo_id( - item.get("fallback_successor_todo_id") - ): - add_declaration( - declaration_id=fallback_succ_id, - successor_todo_id=fallback_succ_id, - ) - + ) return declarations @@ -280,9 +197,9 @@ def declared_fallback_gap_from_agent_vision( """Project one advisory gap for an unresolved declared fallback. A fallback direction is declared structurally via the agent vision's - ``fallback_declarations`` / ``fallback_relationships`` contract or the - task linkage contract on blocked successor items. Prose mentions never - declare a fallback, and generic ``todo_delta`` actions are not fallback + typed ``fallback_declarations`` contract, which the TS-owned Vision + prepare validates and persists. Prose mentions never declare a + fallback, and generic ``todo_delta`` actions are not fallback declarations on their own. The declared direction is resolved when one of: @@ -310,10 +227,7 @@ def declared_fallback_gap_from_agent_vision( ): return None - declarations = parse_fallback_declarations( - agent_vision, - agent_todo_summary=agent_todo_summary, - ) + declarations = parse_fallback_declarations(agent_vision) if not declarations: return None diff --git a/loopx/control_plane/goals/goal_frontier/semantic_history.py b/loopx/control_plane/goals/goal_frontier/semantic_history.py index 66004a81ec..796b7c49f0 100644 --- a/loopx/control_plane/goals/goal_frontier/semantic_history.py +++ b/loopx/control_plane/goals/goal_frontier/semantic_history.py @@ -169,8 +169,6 @@ def latest_agent_vision_from_runs( result["path_delta"] = vision["path_delta"] if isinstance(vision.get("fallback_declarations"), list): result["fallback_declarations"] = vision["fallback_declarations"] - if isinstance(vision.get("fallback_relationships"), list): - result["fallback_relationships"] = vision["fallback_relationships"] return result return None diff --git a/loopx/control_plane/goals/goal_vision.py b/loopx/control_plane/goals/goal_vision.py index b44235a46a..1e8e33af52 100644 --- a/loopx/control_plane/goals/goal_vision.py +++ b/loopx/control_plane/goals/goal_vision.py @@ -44,6 +44,11 @@ "total_limit", "total_usage", ) +# Mirrors the TS-owned prepare contract for bounded typed fallback +# declarations so compaction cannot drop a declared fallback direction. +GOAL_VISION_FALLBACK_DECLARATION_ENTRY_LIMIT = 4 +GOAL_VISION_FALLBACK_DECLARATION_ID_LIMIT = 120 +GOAL_VISION_FALLBACK_DECLARATION_FIELDS = ("target_todo_id", "successor_todo_id") def _compact_public_text(value: Any, *, limit: int) -> str | None: @@ -82,6 +87,33 @@ def _compact_goal_path_delta(value: Any) -> dict[str, Any] | None: return compact if len(compact) > 1 else None +def _compact_fallback_declarations(value: Any) -> list[dict[str, str]]: + if not isinstance(value, list): + return [] + declarations: list[dict[str, str]] = [] + seen: set[str] = set() + for raw in value[:GOAL_VISION_FALLBACK_DECLARATION_ENTRY_LIMIT]: + if not isinstance(raw, dict): + continue + declaration_id = _compact_public_text( + raw.get("declaration_id"), + limit=GOAL_VISION_FALLBACK_DECLARATION_ID_LIMIT, + ) + if not declaration_id or declaration_id in seen: + continue + seen.add(declaration_id) + entry = {"declaration_id": declaration_id} + for field in GOAL_VISION_FALLBACK_DECLARATION_FIELDS: + text = _compact_public_text( + raw.get(field), + limit=GOAL_VISION_FALLBACK_DECLARATION_ID_LIMIT, + ) + if text: + entry[field] = text + declarations.append(entry) + return declarations + + def compact_goal_vision_packet(value: Any) -> dict[str, Any] | None: """Return the public read-path shape of an agent goal-vision packet.""" @@ -93,7 +125,9 @@ def compact_goal_vision_packet(value: Any) -> dict[str, Any] | None: if text: compact[field] = text - patch = value.get("vision_patch") if isinstance(value.get("vision_patch"), dict) else {} + patch = ( + value.get("vision_patch") if isinstance(value.get("vision_patch"), dict) else {} + ) compact_patch: dict[str, str] = {} for field, limit in GOAL_VISION_FIELD_LIMITS.items(): text = _compact_public_text(patch.get(field), limit=limit) @@ -116,7 +150,15 @@ def compact_goal_vision_packet(value: Any) -> dict[str, Any] | None: if todo_delta: compact["todo_delta"] = todo_delta - budget = value.get("vision_budget") if isinstance(value.get("vision_budget"), dict) else {} + declarations = _compact_fallback_declarations(value.get("fallback_declarations")) + if declarations: + compact["fallback_declarations"] = declarations + + budget = ( + value.get("vision_budget") + if isinstance(value.get("vision_budget"), dict) + else {} + ) compact_budget = { field: budget[field] for field in GOAL_VISION_BUDGET_COMPACT_FIELDS @@ -125,7 +167,9 @@ def compact_goal_vision_packet(value: Any) -> dict[str, Any] | None: if compact_budget: compact["vision_budget"] = compact_budget - validation = value.get("validation") if isinstance(value.get("validation"), dict) else {} + validation = ( + value.get("validation") if isinstance(value.get("validation"), dict) else {} + ) compact_validation = { field: validation[field] for field in ("budget_checked", "budget_status", "write_correctness_checked") diff --git a/loopx/control_plane/goals/vision_checkpoint.ts b/loopx/control_plane/goals/vision_checkpoint.ts index 043a1c9aba..e175a22271 100644 --- a/loopx/control_plane/goals/vision_checkpoint.ts +++ b/loopx/control_plane/goals/vision_checkpoint.ts @@ -58,6 +58,14 @@ const GOAL_PATH_DELTA_LIST_LIMITS = { unresolved_questions: [2, 140], evidence_refs: [4, 140], } as const; +// Bounded typed fallback declarations survive prepare unchanged so the +// declared direction cannot disappear behind later read-model compaction. +const VISION_FALLBACK_DECLARATION_ENTRY_LIMIT = 4; +const VISION_FALLBACK_DECLARATION_ID_LIMIT = 120; +const VISION_FALLBACK_DECLARATION_FIELDS = [ + "target_todo_id", + "successor_todo_id", +] as const; const GOAL_VISION_STATE_ALIASES: Readonly> = { closed: "vision_closed", satisfied: "vision_closed", @@ -359,6 +367,63 @@ function normalizeGoalPathDelta( return [normalized, fieldUsage]; } +function normalizeFallbackDeclarations( + value: unknown, +): [JsonObject[], Record] | null { + if (value === null || value === undefined) return null; + if (!Array.isArray(value)) { + throw new EffectRuntimeRequestError( + "agent_vision.fallback_declarations must be a JSON array", + ); + } + if (value.length === 0) return null; + if (value.length > VISION_FALLBACK_DECLARATION_ENTRY_LIMIT) { + throw new EffectRuntimeRequestError( + `agent_vision.fallback_declarations has ${value.length} items; limit is ${VISION_FALLBACK_DECLARATION_ENTRY_LIMIT}`, + ); + } + const declarations: JsonObject[] = []; + const seenIds = new Set(); + const fieldUsage: Record = {}; + value.forEach((raw, index) => { + const entry = requiredObject( + raw, + `agent_vision.fallback_declarations[${index}]`, + ); + const declarationId = boundedPublicText( + `fallback_declarations[${index}].declaration_id`, + entry.declaration_id ?? null, + VISION_FALLBACK_DECLARATION_ID_LIMIT, + ); + if (!declarationId) { + throw new EffectRuntimeRequestError( + `agent_vision.fallback_declarations[${index}] requires a non-empty declaration_id`, + ); + } + if (seenIds.has(declarationId)) { + throw new EffectRuntimeRequestError( + `agent_vision.fallback_declarations repeats declaration_id ${JSON.stringify(declarationId)}`, + ); + } + seenIds.add(declarationId); + const declaration: JsonObject = { declaration_id: declarationId }; + fieldUsage[`fallback_declarations[${index}].declaration_id`] = + declarationId.length; + for (const field of VISION_FALLBACK_DECLARATION_FIELDS) { + const text = boundedPublicText( + `fallback_declarations[${index}].${field}`, + entry[field], + VISION_FALLBACK_DECLARATION_ID_LIMIT, + ); + if (!text) continue; + declaration[field] = text; + fieldUsage[`fallback_declarations[${index}].${field}`] = text.length; + } + declarations.push(declaration); + }); + return [declarations, fieldUsage]; +} + function decodePrepareRequest(request: JsonObject): VisionRefreshPrepareRequest { return { phase: "prepare", @@ -438,6 +503,10 @@ function prepareVisionRefresh(request: VisionRefreshPrepareRequest): JsonObject const [pathDelta, pathDeltaUsage] = normalizeGoalPathDelta(updatePacket.path_delta); Object.assign(fieldUsage, pathDeltaUsage); + const [fallbackDeclarations, fallbackUsage] = normalizeFallbackDeclarations( + updatePacket.fallback_declarations, + ) ?? [null, {}]; + Object.assign(fieldUsage, fallbackUsage); const totalUsage = Object.values(fieldUsage).reduce((total, used) => total + used, 0); if (totalUsage > GOAL_VISION_TOTAL_LIMIT) { throw visionBudgetError( @@ -475,6 +544,8 @@ function prepareVisionRefresh(request: VisionRefreshPrepareRequest): JsonObject for (const [field, [, itemLimit]] of Object.entries(GOAL_PATH_DELTA_LIST_LIMITS)) { fieldLimits[`path_delta.${field}[]`] = itemLimit; } + fieldLimits["fallback_declarations"] = VISION_FALLBACK_DECLARATION_ENTRY_LIMIT; + fieldLimits["fallback_declarations[]"] = VISION_FALLBACK_DECLARATION_ID_LIMIT; const agentVision: JsonObject = { schema_version: GOAL_VISION_REPLAN_SCHEMA_VERSION, @@ -494,6 +565,9 @@ function prepareVisionRefresh(request: VisionRefreshPrepareRequest): JsonObject validation, }; if (pathDelta !== null) agentVision.path_delta = pathDelta; + if (fallbackDeclarations !== null) { + agentVision.fallback_declarations = fallbackDeclarations; + } if ( request.require_path_delta_for_durable_change && diff --git a/tests/control_plane/test_goal_frontier_fallback_disposition.py b/tests/control_plane/test_goal_frontier_fallback_disposition.py index ed09c5c9ea..c02190e43c 100644 --- a/tests/control_plane/test_goal_frontier_fallback_disposition.py +++ b/tests/control_plane/test_goal_frontier_fallback_disposition.py @@ -9,6 +9,10 @@ agent_scoped_selectable_advancement_todo_ids, build_goal_frontier_projection_context_from_status, ) +from loopx.control_plane.goals.goal_vision import ( + compact_goal_vision_packet, + normalize_goal_vision_packet, +) from loopx.control_plane.scheduler.execution_context import ( GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT, ) @@ -40,8 +44,17 @@ def _fallback_vision_run( path_outcome: str | None = None, fallback_declarations: list[Any] | None = None, ) -> dict: - agent_vision: dict = { - "schema_version": "goal_vision_replan_contract_v0", + """Persist a caller packet through the production write/readback chain. + + The caller packet goes through the real TS ``goal.vision_checkpoint`` + prepare (the executor entry) and the compact read-model projection (the + status/shared-runtime entry) before it becomes a run-history record, so + the tests can only pass when the typed declaration survives the same + chain a real caller uses. + """ + + packet: dict = { + "goal_id": GOAL_ID, "agent_id": AGENT_ID, "state": state, "todo_delta": todo_delta @@ -54,9 +67,9 @@ def _fallback_vision_run( }, } if fallback_declarations is not None: - agent_vision["fallback_declarations"] = fallback_declarations + packet["fallback_declarations"] = fallback_declarations elif acceptance_summary == DECLARED_FALLBACK_ACCEPTANCE: - agent_vision["fallback_declarations"] = [ + packet["fallback_declarations"] = [ { "declaration_id": "declared_fallback_direction", "target_todo_id": FALLBACK_ID, @@ -64,13 +77,23 @@ def _fallback_vision_run( } ] if path_outcome is not None: - agent_vision["path_delta"] = {"outcome": path_outcome} + packet["path_delta"] = { + "outcome": path_outcome, + "prior_assumption": "The primary route stays viable after the " + "prerequisite clears.", + "observed_reality": "The declared fallback direction remains the " + "bounded alternative path.", + "stopped": ["Continue only the primary route."], + } + prepared = normalize_goal_vision_packet(packet, goal_id=GOAL_ID, agent_id=AGENT_ID) + compact = compact_goal_vision_packet(prepared) + assert compact is not None return { "classification": "vision_fallback_disposition_fixture", "generated_at": "2026-09-05T00:00:00+00:00", "agent_id": AGENT_ID, "progress_scope": "agent_lane", - "agent_vision": agent_vision, + "agent_vision": compact, } @@ -163,6 +186,30 @@ def test_blocked_primary_with_runnable_fallback_projects_todo_selectable() -> No assert "fallback_gaps" not in decision["goal_frontier_projection"] +def test_declared_fallback_survives_prepare_compact_and_readback() -> None: + # The typed declaration must survive the real TS prepare, the compact + # read model, and the history readback before any gap can be projected. + run = _fallback_vision_run( + todo_delta=[f"retain:{PRIMARY_WAIT_ID}", f"retain:{FALLBACK_ID}"] + ) + vision = run["agent_vision"] + + assert vision["fallback_declarations"] == [ + { + "declaration_id": "declared_fallback_direction", + "target_todo_id": FALLBACK_ID, + "successor_todo_id": FALLBACK_ID, + } + ] + from loopx.control_plane.goals.goal_frontier.semantic_history import ( + latest_agent_vision_from_runs, + ) + + readback = latest_agent_vision_from_runs([run], goal_id=GOAL_ID, agent_id=AGENT_ID) + assert readback is not None + assert readback["fallback_declarations"] == vision["fallback_declarations"] + + def test_declared_fallback_without_resolution_projects_single_gap() -> 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. @@ -194,6 +241,15 @@ def test_declared_fallback_without_resolution_projects_single_gap() -> None: assert "do not invent a user gate" in gap["recommended_action"] assert frontier["replan_required"] is False + decision = build_quota_should_run( + payload, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + scheduler_execution_context=(GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT), + ) + quota_projection = decision["goal_frontier_projection"] + assert quota_projection["fallback_gaps"][0]["unresolved_todo_ids"] == [FALLBACK_ID] + def test_retaining_only_the_blocked_primary_successor_is_no_declaration() -> None: # The primary successor is the wait state's own object; retaining it does @@ -209,7 +265,6 @@ def test_retaining_only_the_blocked_primary_successor_is_no_declaration() -> Non ) frontier = _frontier_projection(payload) - assert "fallback_gaps" not in frontier @@ -387,31 +442,6 @@ def test_unrelated_create_does_not_resolve_fallback_gap() -> None: assert gaps[0]["unresolved_todo_ids"] == [FALLBACK_ID] -def test_fallback_declared_via_todo_linkage_contract_projects_gap() -> None: - # A fallback declared on a blocked successor item via the task linkage - # contract projects an advisory gap when unresolved. - payload = _status_payload( - fallback_runnable=False, - latest_runs=[ - _fallback_vision_run( - todo_delta=[f"retain:{PRIMARY_WAIT_ID}"], - fallback_declarations=[], - ) - ], - ) - item = payload["attention_queue"]["items"][0] - agent_todos = item["agent_todos"] - for deferred in agent_todos.get("deferred_items") or []: - if isinstance(deferred, dict): - deferred["fallback_todo_id"] = FALLBACK_ID - - frontier = _frontier_projection(payload) - - gaps = frontier["fallback_gaps"] - assert len(gaps) == 1 - assert gaps[0]["unresolved_todo_ids"] == [FALLBACK_ID] - - def test_typed_declaration_successor_relation_resolves_gap() -> None: # A typed declaration-to-successor relation resolves when its declared # bounded successor is created in todo_delta. @@ -435,6 +465,63 @@ def test_typed_declaration_successor_relation_resolves_gap() -> None: assert "fallback_gaps" not in frontier +def test_legacy_alias_shapes_do_not_declare_a_fallback() -> None: + # Unsupported alias shapes (string arrows, patch-level fields, renamed + # keys) have no production author; the reader must not resurrect them. + run = _fallback_vision_run( + todo_delta=[f"retain:{PRIMARY_WAIT_ID}"], + fallback_declarations=[], + ) + vision = dict(run["agent_vision"]) + vision["fallback_relationships"] = [ + {"fallback_id": FALLBACK_ID, "successor_id": FALLBACK_ID} + ] + vision["vision_patch"] = { + **vision["vision_patch"], + "fallback_declarations": [f"{FALLBACK_ID}->{FALLBACK_ID}"], + } + payload = _status_payload( + fallback_runnable=False, + latest_runs=[{**run, "agent_vision": vision}], + ) + + frontier = _frontier_projection(payload) + + assert "fallback_gaps" not in frontier + + +def test_compact_read_model_mirrors_the_bounded_declaration_contract() -> None: + # Compaction is the defensive mirror of the TS prepare contract: at most + # four entries, one per unique declaration_id, typed fields only, and + # anything past the bound is truncated exactly like the write contract + # rejects it. + compact = compact_goal_vision_packet( + { + "schema_version": "goal_vision_replan_contract_v0", + "goal_id": GOAL_ID, + "agent_id": AGENT_ID, + "state": "vision_drift_detected", + "vision_patch": {"vision_summary": "Bounded route."}, + "fallback_declarations": [ + { + "declaration_id": "first_direction", + "target_todo_id": FALLBACK_ID, + "legacy_field": "dropped", + }, + {"declaration_id": "first_direction", "target_todo_id": "todo_dup"}, + {"target_todo_id": "todo_missing_id"}, + "not-an-object", + {"declaration_id": "truncated_direction"}, + ], + } + ) + + assert compact is not None + assert compact["fallback_declarations"] == [ + {"declaration_id": "first_direction", "target_todo_id": FALLBACK_ID}, + ] + + def test_selectable_frontier_ids_mirror_the_authoritative_counts() -> None: # The completion-evidence id set must be the same agent-scoped frontier # the authoritative advancement counter projects. diff --git a/tests/control_plane_ts/vision_checkpoint.test.ts b/tests/control_plane_ts/vision_checkpoint.test.ts index 51f2f82240..b9cd27499c 100644 --- a/tests/control_plane_ts/vision_checkpoint.test.ts +++ b/tests/control_plane_ts/vision_checkpoint.test.ts @@ -224,6 +224,121 @@ test("prepare rejects incomplete and over-wide path deltas", () => { ); }); +test("prepare validates and carries bounded unique fallback declarations", () => { + const result = buildVisionCheckpoint(prepareRequest({ + agent_vision_packet: { + vision_summary: "Ship one bounded route.", + fallback_declarations: [ + { + declaration_id: " declared_fallback_direction ", + target_todo_id: "todo_declared_fallback", + successor_todo_id: null, + }, + { declaration_id: "typed_successor", successor_todo_id: "todo_successor" }, + ], + }, + })); + + const vision = result.agent_vision as Record; + assert.deepEqual(vision.fallback_declarations, [ + { + declaration_id: "declared_fallback_direction", + target_todo_id: "todo_declared_fallback", + }, + { declaration_id: "typed_successor", successor_todo_id: "todo_successor" }, + ]); + const budget = vision.vision_budget as Record; + const fieldLimits = budget.field_limits as Record; + const fieldUsage = budget.field_usage as Record; + assert.equal(fieldLimits["fallback_declarations"], 4); + assert.equal(fieldLimits["fallback_declarations[]"], 120); + assert.equal( + fieldUsage["fallback_declarations[0].declaration_id"], + "declared_fallback_direction".length, + ); + + const empty = buildVisionCheckpoint(prepareRequest({ + agent_vision_packet: { + vision_summary: "Ship one bounded route.", + fallback_declarations: [], + }, + })) as Record; + assert.equal( + (empty.agent_vision as Record).fallback_declarations, + undefined, + ); +}); + +test("prepare rejects unbounded, duplicated, and unsafe fallback declarations", () => { + const declaration = (id: string) => ({ declaration_id: id }); + assert.throws( + () => buildVisionCheckpoint(prepareRequest({ + agent_vision_packet: { + vision_summary: "Ship one bounded route.", + fallback_declarations: "declared_fallback_direction", + }, + })), + /fallback_declarations must be a JSON array/, + ); + assert.throws( + () => buildVisionCheckpoint(prepareRequest({ + agent_vision_packet: { + vision_summary: "Ship one bounded route.", + fallback_declarations: [1, 2, 3, 4, 5].map(() => declaration("one")), + }, + })), + /fallback_declarations has 5 items; limit is 4/, + ); + assert.throws( + () => buildVisionCheckpoint(prepareRequest({ + agent_vision_packet: { + vision_summary: "Ship one bounded route.", + fallback_declarations: [ + declaration("declared_fallback_direction"), + declaration("declared_fallback_direction"), + ], + }, + })), + /repeats declaration_id "declared_fallback_direction"/, + ); + assert.throws( + () => buildVisionCheckpoint(prepareRequest({ + agent_vision_packet: { + vision_summary: "Ship one bounded route.", + fallback_declarations: [{ target_todo_id: "todo_declared_fallback" }], + }, + })), + /requires a non-empty declaration_id/, + ); + assert.throws( + () => buildVisionCheckpoint(prepareRequest({ + agent_vision_packet: { + vision_summary: "Ship one bounded route.", + fallback_declarations: [ + { declaration_id: "route", target_todo_id: "todo_with_fallback" }, + { declaration_id: "token = leaked", successor_todo_id: "todo_x" }, + ], + }, + })), + /contains a private-looking value/, + ); + assert.throws( + () => buildVisionCheckpoint(prepareRequest({ + agent_vision_packet: { + vision_summary: "Ship one bounded route.", + fallback_declarations: [ + { declaration_id: "x".repeat(121) }, + ], + }, + })), + (error: unknown) => { + const rejection = error as { code?: string; message?: string }; + assert.equal(rejection.code, "vision_budget_exceeded"); + return true; + }, + ); +}); + test("pure prepare and finalize reductions replay deterministically", () => { const prepare = prepareRequest(); assert.deepEqual( From 08e8220cd58d22c5867a8aaf9459138b5702158c Mon Sep 17 00:00:00 2001 From: now-ing Date: Mon, 7 Sep 2026 07:26:56 +0800 Subject: [PATCH 6/7] fix(state-refresh): keep fallback declarations in the run index allowlist The state-refresh run index still wrote agent_vision through the old field allowlist, so a successfully prepared and persisted fallback declaration disappeared from runs/index.jsonl and every later semantic-history read missed the frontier reminder the declaration exists to provide. The allowlist now carries fallback_declarations, pinned by a builder-level index regression that feeds a real typed declaration through and asserts it survives the index projection. Signed-off-by: now-ing --- loopx/state_refresh.py | 287 +++++++++++++++--------- tests/test_state_refresh_projections.py | 68 +++++- 2 files changed, 245 insertions(+), 110 deletions(-) diff --git a/loopx/state_refresh.py b/loopx/state_refresh.py index 4630d393cf..1c1832c673 100644 --- a/loopx/state_refresh.py +++ b/loopx/state_refresh.py @@ -29,7 +29,9 @@ render_refresh_recovery_markdown, settlement_result_payload, ) -from .control_plane.quota.settlement_workspace_causality import resolve_settlement_workspace_requirement +from .control_plane.quota.settlement_workspace_causality import ( + resolve_settlement_workspace_requirement, +) from .control_plane.quota.codex_session_usage import ( book_codex_session_usage, usage_booking_lock_target, @@ -75,7 +77,9 @@ reserve_unique_run_paths, unique_run_paths, ) -from .control_plane.runtime.local_state_write_correctness import build_local_state_write_correctness_dry_run_packet +from .control_plane.runtime.local_state_write_correctness import ( + build_local_state_write_correctness_dry_run_packet, +) from .paths import resolve_runtime_root from .control_plane.goals.vision_checkpoint import ( build_vision_checkpoint, @@ -206,7 +210,9 @@ def registered_agents_for_goal(registry_goal: dict[str, Any] | None) -> list[str if registry_goal and isinstance(registry_goal.get("coordination"), dict) else {} ) - registered_raw = coordination.get("registered_agents") if isinstance(coordination, dict) else [] + registered_raw = ( + coordination.get("registered_agents") if isinstance(coordination, dict) else [] + ) registered_values = registered_raw if isinstance(registered_raw, list) else [] registered_agents: list[str] = [] for value in registered_values: @@ -330,7 +336,10 @@ def resolve_goal_state( project_override: Path | None, state_file_override: Path | None, ) -> tuple[dict[str, Any] | None, Path | None, Path]: - goal = next((item for item in registry_goals(registry) if str(item.get("id")) == goal_id), None) + goal = next( + (item for item in registry_goals(registry) if str(item.get("id")) == goal_id), + None, + ) project = project_override.expanduser().resolve() if project_override else None if project is None and goal and goal.get("repo"): project = Path(str(goal.get("repo"))).expanduser() @@ -344,7 +353,9 @@ def resolve_goal_state( if state_file is None and goal: state_file = registered_state_file if state_file is None: - raise ValueError("state file is required when the goal is not resolvable from registry") + raise ValueError( + "state file is required when the goal is not resolvable from registry" + ) if not state_file.is_absolute(): if project is None: raise ValueError("relative state file requires --project or registry repo") @@ -352,9 +363,13 @@ def resolve_goal_state( state_file = state_file.resolve() if state_file_override is not None: if project is None: - raise ValueError("--state-file override requires --project or a registry goal with repo") + raise ValueError( + "--state-file override requires --project or a registry goal with repo" + ) registered_resolved = ( - registered_state_file.resolve() if registered_state_file is not None else None + registered_state_file.resolve() + if registered_state_file is not None + else None ) if state_file != registered_resolved and not state_file.is_relative_to(project): raise ValueError( @@ -444,9 +459,7 @@ def build_state_refresh_record( if settlement_identity.todo_id: record["todo_id"] = settlement_identity.todo_id if settlement_identity.replan_obligation_id: - record["replan_obligation_id"] = ( - settlement_identity.replan_obligation_id - ) + record["replan_obligation_id"] = settlement_identity.replan_obligation_id if autonomous_replan_recorded: record["autonomous_replan_ack"] = { "schema_version": "autonomous_replan_ack_v0", @@ -456,9 +469,7 @@ def build_state_refresh_record( if repair_delta_contract: record["autonomous_replan_ack"]["delta_contract"] = repair_delta_contract if replan_semantic_delta: - record["autonomous_replan_ack"]["semantic_delta"] = ( - replan_semantic_delta - ) + record["autonomous_replan_ack"]["semantic_delta"] = replan_semantic_delta if autonomous_replan_frontier_identity: record["autonomous_replan_ack"]["frontier_identity"] = ( autonomous_replan_frontier_identity @@ -497,19 +508,25 @@ def _build_state_refresh_output_projections( index_record = { field: record[field] for field in ( - "generated_at", "goal_id", "classification", "recommended_action", - "recommended_action_source", "health_check", + "generated_at", + "goal_id", + "classification", + "recommended_action", + "recommended_action_source", + "health_check", ) } - index_record.update({ - "json_path": str(json_path), - "markdown_path": str(markdown_path), - "state": { - "sha256_16": record_state.get("sha256_16"), - "frontmatter": {"updated_at": record_frontmatter.get("updated_at")}, - }, - "runtime_projection_route": record["runtime_projection_route"], - }) + index_record.update( + { + "json_path": str(json_path), + "markdown_path": str(markdown_path), + "state": { + "sha256_16": record_state.get("sha256_16"), + "frontmatter": {"updated_at": record_frontmatter.get("updated_at")}, + }, + "runtime_projection_route": record["runtime_projection_route"], + } + ) for field in ( "recommended_action_resolution", "delivery_batch_scale", @@ -528,15 +545,22 @@ def _build_state_refresh_output_projections( if autonomous_replan_recorded_requested or replan_ack.get("recorded") is True: index_record["autonomous_replan_ack"] = replan_ack if replan_ack.get("requested_classification"): - index_record["requested_classification"] = replan_ack["requested_classification"] + index_record["requested_classification"] = replan_ack[ + "requested_classification" + ] agent_vision = record.get("agent_vision") if isinstance(agent_vision, dict): index_record["agent_vision"] = { field: agent_vision.get(field) for field in ( - "schema_version", "agent_id", "state", "vision_patch", - "todo_delta", "vision_budget", + "schema_version", + "agent_id", + "state", + "vision_patch", + "todo_delta", + "fallback_declarations", + "vision_budget", ) } if isinstance(agent_vision.get("path_delta"), dict): @@ -560,26 +584,42 @@ def _build_state_refresh_output_projections( "runtime_root": str(runtime_root), "project": str(project) if project else None, } - payload.update({ - field: record.get(field) - for field in ("goal_id", "classification", "progress_scope", "agent_id", "agent_lane") - }) - payload.update({ - "autonomous_replan_recorded": bool(replan_ack.get("recorded")), - "autonomous_replan_recorded_requested": autonomous_replan_recorded_requested, - "repair_delta_contract": replan_ack.get("delta_contract"), - "json_path": str(json_path), - "markdown_path": str(markdown_path), - "index_path": str(index_path), - }) - payload.update({ - field: record.get(field) - for field in ( - "agent_vision", "vision_checkpoint", "recommended_action", - "recommended_action_source", "active_state_next_action_update", - "generated_at", "health_check", - ) - }) + payload.update( + { + field: record.get(field) + for field in ( + "goal_id", + "classification", + "progress_scope", + "agent_id", + "agent_lane", + ) + } + ) + payload.update( + { + "autonomous_replan_recorded": bool(replan_ack.get("recorded")), + "autonomous_replan_recorded_requested": autonomous_replan_recorded_requested, + "repair_delta_contract": replan_ack.get("delta_contract"), + "json_path": str(json_path), + "markdown_path": str(markdown_path), + "index_path": str(index_path), + } + ) + payload.update( + { + field: record.get(field) + for field in ( + "agent_vision", + "vision_checkpoint", + "recommended_action", + "recommended_action_source", + "active_state_next_action_update", + "generated_at", + "health_check", + ) + } + ) payload.update(record) return index_record, payload @@ -589,7 +629,9 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str: if recovery_markdown is not None: return recovery_markdown state = payload.get("state") if isinstance(payload.get("state"), dict) else {} - frontmatter = state.get("frontmatter") if isinstance(state.get("frontmatter"), dict) else {} + frontmatter = ( + state.get("frontmatter") if isinstance(state.get("frontmatter"), dict) else {} + ) lines = [ "# LoopX State Refresh", "", @@ -696,7 +738,9 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str: f"target_roles={','.join(projection_gap.get('target_roles') or [])}" ) if projection_gap.get("recommended_action"): - lines.append(f"- state_projection_gap_action: {projection_gap.get('recommended_action')}") + lines.append( + f"- state_projection_gap_action: {projection_gap.get('recommended_action')}" + ) next_action_update = ( payload.get("active_state_next_action_update") @@ -721,8 +765,16 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str: else {} ) if write_correctness: - intent = write_correctness.get("write_intent") if isinstance(write_correctness.get("write_intent"), dict) else {} - preview = write_correctness.get("preview") if isinstance(write_correctness.get("preview"), dict) else {} + intent = ( + write_correctness.get("write_intent") + if isinstance(write_correctness.get("write_intent"), dict) + else {} + ) + preview = ( + write_correctness.get("preview") + if isinstance(write_correctness.get("preview"), dict) + else {} + ) apply_result = ( write_correctness.get("apply_result") if isinstance(write_correctness.get("apply_result"), dict) @@ -736,7 +788,11 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str: f"non_destructive={preview.get('non_destructive')}" ) - global_sync = payload.get("global_sync") if isinstance(payload.get("global_sync"), dict) else {} + global_sync = ( + payload.get("global_sync") + if isinstance(payload.get("global_sync"), dict) + else {} + ) if global_sync: lines.extend( [ @@ -765,9 +821,7 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str: f"`{recommendation_resolution.get('settlement_alignment')}`" ) if recommendation_resolution.get("todo_id"): - lines.append( - f"- todo_id: `{recommendation_resolution.get('todo_id')}`" - ) + lines.append(f"- todo_id: `{recommendation_resolution.get('todo_id')}`") lines.append(str(payload.get("recommended_action") or "")) for heading, key in ( ("Next Action", "next_action"), @@ -836,7 +890,9 @@ def refresh_state_run( raise ValueError("--agent-lane requires --agent-id so the lane has an owner") normalized_progress_scope = normalize_progress_scope(progress_scope) normalized_delivery_batch_scale = ( - require_delivery_batch_scale(delivery_batch_scale).value if delivery_batch_scale else None + require_delivery_batch_scale(delivery_batch_scale).value + if delivery_batch_scale + else None ) normalized_delivery_outcome = ( require_delivery_outcome(delivery_outcome).value if delivery_outcome else None @@ -904,7 +960,9 @@ def refresh_state_run( }, ) if settlement_readback is None: - raise RuntimeError("exact settlement readback unexpectedly returned not-found") + raise RuntimeError( + "exact settlement readback unexpectedly returned not-found" + ) settlement_result = settlement_readback.identity if settlement_result.failure is not None: raise ValueError(settlement_result.failure.reason) @@ -943,7 +1001,8 @@ def refresh_state_run( ) return payload settlement_workspace_requirement = resolve_settlement_workspace_requirement( - delivery_workspace_causality, settlement_binding_kind=settlement_identity.binding_kind.value + delivery_workspace_causality, + settlement_binding_kind=settlement_identity.binding_kind.value, ) runtime_projection_route = resolve_runtime_projection_route( registry_path=registry_path, @@ -979,7 +1038,9 @@ def refresh_state_run( todo_id=(settlement_identity.todo_id if settlement_identity else None), agent_id=normalized_agent_id or None, ) - normalized_next_action = normalize_next_action_text(next_action) if next_action else None + normalized_next_action = ( + normalize_next_action_text(next_action) if next_action else None + ) registered_agents = registered_agents_for_goal(registry_goal) known_agents = {agent for agent in registered_agents if agent} multi_agent_goal = len(known_agents) > 1 @@ -1018,7 +1079,9 @@ def refresh_state_run( if normalized_progress_scope == GOAL_PROGRESS_SCOPE: if normalized_agent_lane: raise ValueError("--agent-lane requires --progress-scope agent_lane") - if (agent_vision_packet is not None or vision_unchanged_reason) and not normalized_agent_id: + if ( + agent_vision_packet is not None or vision_unchanged_reason + ) and not normalized_agent_id: raise ValueError("vision writeback requires --agent-id") agent_vision: dict[str, Any] | None = None existing_agent_vision: dict[str, Any] | None = None @@ -1107,15 +1170,13 @@ def refresh_state_run( existing_agent_vision=existing_agent_vision, agent_id=normalized_agent_id, dry_run=dry_run, - settlement_todo_id=(settlement_identity.todo_id if settlement_identity else None), - settlement_guard_scoped=( - settlement_replan_guard.get("scope") == "turn_guard" + settlement_todo_id=( + settlement_identity.todo_id if settlement_identity else None ), + settlement_guard_scoped=(settlement_replan_guard.get("scope") == "turn_guard"), settlement_guard_semantic_replan_obligation_id=( settlement_replan_guard.get("selected_obligation_id") - if isinstance( - settlement_replan_guard.get("selected_obligation_id"), str - ) + if isinstance(settlement_replan_guard.get("selected_obligation_id"), str) else None ), newest_first_runs=newest_first_runs, @@ -1182,13 +1243,9 @@ def refresh_state_run( else None ), ) - if ( - peer_independent_worktree_required - and ( - delivery_workspace is None - or delivery_workspace.get("workspace_kind") - != "independent_git_worktree" - ) + if peer_independent_worktree_required and ( + delivery_workspace is None + or delivery_workspace.get("workspace_kind") != "independent_git_worktree" ): raise ValueError( "accountable peer delivery must be refreshed from the independent " @@ -1255,9 +1312,7 @@ def refresh_state_run( if refresh_recovery: record["refresh_recovery"] = refresh_recovery if settlement_workspace_requirement: - record["settlement_workspace_requirement"] = ( - settlement_workspace_requirement - ) + record["settlement_workspace_requirement"] = settlement_workspace_requirement if autonomous_replan_recorded: if "autonomous_replan_ack" not in record: record["autonomous_replan_ack"] = { @@ -1272,7 +1327,9 @@ def refresh_state_run( autonomous_replan_frontier_identity ) if requested_classification != classification: - record["autonomous_replan_ack"]["requested_classification"] = requested_classification + record["autonomous_replan_ack"]["requested_classification"] = ( + requested_classification + ) record["autonomous_replan_noop"] = { "schema_version": REPAIR_NOOP_SCHEMA_VERSION, "classification": classification, @@ -1288,9 +1345,7 @@ def refresh_state_run( "source": "refresh_state_semantic_delta", }, ) - record["autonomous_replan_ack"]["semantic_delta"] = ( - replan_semantic_delta - ) + record["autonomous_replan_ack"]["semantic_delta"] = replan_semantic_delta if active_state_next_action_update: record["active_state_next_action_update"] = active_state_next_action_update compact_route = compact_runtime_projection_route(runtime_projection_route) @@ -1339,7 +1394,9 @@ def refresh_state_run( payload["usage"] = dict(record["usage"]) if dry_run: expected_write_scopes = ["runtime_history"] - if active_state_next_action_update and active_state_next_action_update.get("would_update"): + if active_state_next_action_update and active_state_next_action_update.get( + "would_update" + ): expected_write_scopes.insert(0, "active_state") if sync_global and route_status in {"resolved", "single_runtime"}: expected_write_scopes.append("global_registry") @@ -1354,31 +1411,40 @@ def refresh_state_run( if sync_global and route_status in {"resolved", "single_runtime"}: patch_parts.append("sync public-safe registry projection") elif sync_global: - patch_parts.append(f"block global sync on {route_status} runtime projection route") + patch_parts.append( + f"block global sync on {route_status} runtime projection route" + ) if shared_runtime_root: - patch_parts.append("project compact refresh to registered shared runtime") - payload["local_state_write_correctness"] = build_local_state_write_correctness_dry_run_packet( - goal_id=safe_goal_id, - writer_id=normalized_agent_id or "loopx.refresh-state", - write_class="refresh_state", - state_text=expected_write_state_text, - target_refs={ - "state_file_ref": "registry.goal.state_file", - "run_history_ref": "runtime.goal.runs", - "index_ref": "runtime.goal.runs.index", - "global_registry_ref": ( - "runtime.registry.global" - if sync_global and route_status in {"resolved", "single_runtime"} - else None - ), - "shared_runtime_projection_ref": ( - "shared_runtime.goal.runs.index" if shared_runtime_root else None - ), - }, - patch_summary="; ".join(patch_parts), - expected_write_scopes=expected_write_scopes, - lease_ref=None, - projection_status_surface=f"refresh-state dry-run: {classification}", + patch_parts.append( + "project compact refresh to registered shared runtime" + ) + payload["local_state_write_correctness"] = ( + build_local_state_write_correctness_dry_run_packet( + goal_id=safe_goal_id, + writer_id=normalized_agent_id or "loopx.refresh-state", + write_class="refresh_state", + state_text=expected_write_state_text, + target_refs={ + "state_file_ref": "registry.goal.state_file", + "run_history_ref": "runtime.goal.runs", + "index_ref": "runtime.goal.runs.index", + "global_registry_ref": ( + "runtime.registry.global" + if sync_global + and route_status in {"resolved", "single_runtime"} + else None + ), + "shared_runtime_projection_ref": ( + "shared_runtime.goal.runs.index" + if shared_runtime_root + else None + ), + }, + patch_summary="; ".join(patch_parts), + expected_write_scopes=expected_write_scopes, + lease_ref=None, + projection_status_surface=f"refresh-state dry-run: {classification}", + ) ) if not dry_run: runs_dir.mkdir(parents=True, exist_ok=True) @@ -1388,12 +1454,17 @@ def refresh_state_run( payload["json_path"] = str(json_path) payload["markdown_path"] = str(markdown_path) json_path.write_text( - json.dumps(record, ensure_ascii=False, indent=2, allow_nan=False) + "\n", + json.dumps(record, ensure_ascii=False, indent=2, allow_nan=False) + + "\n", encoding="utf-8", ) - markdown_path.write_text(render_state_refresh_markdown(payload) + "\n", encoding="utf-8") + markdown_path.write_text( + render_state_refresh_markdown(payload) + "\n", encoding="utf-8" + ) with index_path.open("a", encoding="utf-8") as f: - f.write(json.dumps(index_record, ensure_ascii=False, allow_nan=False) + "\n") + f.write( + json.dumps(index_record, ensure_ascii=False, allow_nan=False) + "\n" + ) if sync_global and route_status in {"missing", "ambiguous"}: payload["ok"] = False payload["partial_write"] = not dry_run diff --git a/tests/test_state_refresh_projections.py b/tests/test_state_refresh_projections.py index 6e77217389..b411ac344d 100644 --- a/tests/test_state_refresh_projections.py +++ b/tests/test_state_refresh_projections.py @@ -4,7 +4,10 @@ def test_refresh_output_projections_preserve_optional_run_contracts() -> None: - delta_contract = {"schema_version": "repair_delta_contract_v0", "delta_present": True} + delta_contract = { + "schema_version": "repair_delta_contract_v0", + "delta_present": True, + } agent_vision = { "schema_version": "goal_vision_v0", "agent_id": "quality-agent", @@ -12,6 +15,7 @@ def test_refresh_output_projections_preserve_optional_run_contracts() -> None: "vision_patch": {"acceptance_summary": "Keep simplification measurable."}, "todo_delta": ["Simplify the projection owner."], "vision_budget": {"status": "within_budget"}, + "fallback_declarations": None, "path_delta": {"changed": ["projection assembly"]}, "validation": {"budget_checked": True}, } @@ -25,7 +29,10 @@ def test_refresh_output_projections_preserve_optional_run_contracts() -> None: "state": { "path": "/ignored/project/state.md", "sha256_16": "0123456789abcdef", - "frontmatter": {"updated_at": "2026-07-30T23:59:00+00:00", "status": "active"}, + "frontmatter": { + "updated_at": "2026-07-30T23:59:00+00:00", + "status": "active", + }, "next_action": ["continue"], }, "runtime_projection_route": {"status": "resolved", "route_id": "route-1"}, @@ -71,6 +78,7 @@ def test_refresh_output_projections_preserve_optional_run_contracts() -> None: "state", "vision_patch", "todo_delta", + "fallback_declarations", "vision_budget", "path_delta", ) @@ -95,3 +103,59 @@ def test_refresh_output_projections_preserve_optional_run_contracts() -> None: assert payload["active_state_next_action_update"] == {"updated": True} assert payload["agent_vision"] is agent_vision assert payload["state"] is record["state"] + + +def test_run_index_agent_vision_keeps_fallback_declarations(tmp_path): + """The run index must persist typed fallback declarations for later reads.""" + + from loopx.state_refresh import _build_state_refresh_output_projections + + record = { + "schema_version": "loopx_goal_run_v0", + "generated_at": "2026-09-07T00:00:00+00:00", + "goal_id": "goal-1", + "agent_id": "agent-a", + "turn_instance_id": "turn-1", + "observed_at": "2026-09-07T00:00:00Z", + "classification": "validated_progress", + "recommended_action": "continue", + "recommended_action_source": "explicit_arg", + "health_check": "state_file 1/1", + "state": {"frontmatter": {"updated_at": "2026-09-06T23:59:00+00:00"}}, + "runtime_projection_route": {"status": "resolved", "route_id": "route-1"}, + "delivery_batch_scale": "single_surface", + "delivery_outcome": "outcome_progress", + "delivery_workspace": {"workspace_kind": "independent_git_worktree"}, + "agent_vision": { + "schema_version": "loopx_goal_vision_packet_v0", + "agent_id": "agent-a", + "state": "active", + "vision_patch": "Primary path; fallback direction declared.", + "todo_delta": [], + "fallback_declarations": [ + {"declaration_id": "decl_fallback_1", "target_todo_id": "todo_fb1"} + ], + "vision_budget": {}, + }, + } + registry = tmp_path / "registry.global.json" + registry.write_text("{}", encoding="utf-8") + runtime_root = tmp_path / "runtime" + json_path = runtime_root / "goals" / "goal-1" / "runs" / "run.json" + markdown_path = runtime_root / "goals" / "goal-1" / "runs" / "run.md" + index_path = runtime_root / "goals" / "goal-1" / "runs" / "index.jsonl" + _record, index_record = _build_state_refresh_output_projections( + record=record, + registry_path=registry, + runtime_root=runtime_root, + project=tmp_path, + json_path=json_path, + markdown_path=markdown_path, + index_path=index_path, + dry_run=True, + autonomous_replan_recorded_requested=False, + ) + indexed_vision = index_record.get("agent_vision") or {} + assert indexed_vision.get("fallback_declarations") == [ + {"declaration_id": "decl_fallback_1", "target_todo_id": "todo_fb1"} + ] From b765958076d5ed655261bab040696a762ff322b7 Mon Sep 17 00:00:00 2001 From: now-ing Date: Mon, 7 Sep 2026 07:28:48 +0800 Subject: [PATCH 7/7] fix(state-refresh): keep the run index allowlist edit surgical The previous formatting pass expanded historical compact imports and pushed the module past the maintainability ratchet line budget. Restore the original layout and reapply only the fallback_declarations allowlist entry, keeping the module at its previous 1479 lines. Signed-off-by: now-ing --- loopx/state_refresh.py | 287 ++++++++++++++++------------------------- 1 file changed, 108 insertions(+), 179 deletions(-) diff --git a/loopx/state_refresh.py b/loopx/state_refresh.py index 1c1832c673..11e61e95c3 100644 --- a/loopx/state_refresh.py +++ b/loopx/state_refresh.py @@ -29,9 +29,7 @@ render_refresh_recovery_markdown, settlement_result_payload, ) -from .control_plane.quota.settlement_workspace_causality import ( - resolve_settlement_workspace_requirement, -) +from .control_plane.quota.settlement_workspace_causality import resolve_settlement_workspace_requirement from .control_plane.quota.codex_session_usage import ( book_codex_session_usage, usage_booking_lock_target, @@ -77,9 +75,7 @@ reserve_unique_run_paths, unique_run_paths, ) -from .control_plane.runtime.local_state_write_correctness import ( - build_local_state_write_correctness_dry_run_packet, -) +from .control_plane.runtime.local_state_write_correctness import build_local_state_write_correctness_dry_run_packet from .paths import resolve_runtime_root from .control_plane.goals.vision_checkpoint import ( build_vision_checkpoint, @@ -210,9 +206,7 @@ def registered_agents_for_goal(registry_goal: dict[str, Any] | None) -> list[str if registry_goal and isinstance(registry_goal.get("coordination"), dict) else {} ) - registered_raw = ( - coordination.get("registered_agents") if isinstance(coordination, dict) else [] - ) + registered_raw = coordination.get("registered_agents") if isinstance(coordination, dict) else [] registered_values = registered_raw if isinstance(registered_raw, list) else [] registered_agents: list[str] = [] for value in registered_values: @@ -336,10 +330,7 @@ def resolve_goal_state( project_override: Path | None, state_file_override: Path | None, ) -> tuple[dict[str, Any] | None, Path | None, Path]: - goal = next( - (item for item in registry_goals(registry) if str(item.get("id")) == goal_id), - None, - ) + goal = next((item for item in registry_goals(registry) if str(item.get("id")) == goal_id), None) project = project_override.expanduser().resolve() if project_override else None if project is None and goal and goal.get("repo"): project = Path(str(goal.get("repo"))).expanduser() @@ -353,9 +344,7 @@ def resolve_goal_state( if state_file is None and goal: state_file = registered_state_file if state_file is None: - raise ValueError( - "state file is required when the goal is not resolvable from registry" - ) + raise ValueError("state file is required when the goal is not resolvable from registry") if not state_file.is_absolute(): if project is None: raise ValueError("relative state file requires --project or registry repo") @@ -363,13 +352,9 @@ def resolve_goal_state( state_file = state_file.resolve() if state_file_override is not None: if project is None: - raise ValueError( - "--state-file override requires --project or a registry goal with repo" - ) + raise ValueError("--state-file override requires --project or a registry goal with repo") registered_resolved = ( - registered_state_file.resolve() - if registered_state_file is not None - else None + registered_state_file.resolve() if registered_state_file is not None else None ) if state_file != registered_resolved and not state_file.is_relative_to(project): raise ValueError( @@ -459,7 +444,9 @@ def build_state_refresh_record( if settlement_identity.todo_id: record["todo_id"] = settlement_identity.todo_id if settlement_identity.replan_obligation_id: - record["replan_obligation_id"] = settlement_identity.replan_obligation_id + record["replan_obligation_id"] = ( + settlement_identity.replan_obligation_id + ) if autonomous_replan_recorded: record["autonomous_replan_ack"] = { "schema_version": "autonomous_replan_ack_v0", @@ -469,7 +456,9 @@ def build_state_refresh_record( if repair_delta_contract: record["autonomous_replan_ack"]["delta_contract"] = repair_delta_contract if replan_semantic_delta: - record["autonomous_replan_ack"]["semantic_delta"] = replan_semantic_delta + record["autonomous_replan_ack"]["semantic_delta"] = ( + replan_semantic_delta + ) if autonomous_replan_frontier_identity: record["autonomous_replan_ack"]["frontier_identity"] = ( autonomous_replan_frontier_identity @@ -508,25 +497,19 @@ def _build_state_refresh_output_projections( index_record = { field: record[field] for field in ( - "generated_at", - "goal_id", - "classification", - "recommended_action", - "recommended_action_source", - "health_check", + "generated_at", "goal_id", "classification", "recommended_action", + "recommended_action_source", "health_check", ) } - index_record.update( - { - "json_path": str(json_path), - "markdown_path": str(markdown_path), - "state": { - "sha256_16": record_state.get("sha256_16"), - "frontmatter": {"updated_at": record_frontmatter.get("updated_at")}, - }, - "runtime_projection_route": record["runtime_projection_route"], - } - ) + index_record.update({ + "json_path": str(json_path), + "markdown_path": str(markdown_path), + "state": { + "sha256_16": record_state.get("sha256_16"), + "frontmatter": {"updated_at": record_frontmatter.get("updated_at")}, + }, + "runtime_projection_route": record["runtime_projection_route"], + }) for field in ( "recommended_action_resolution", "delivery_batch_scale", @@ -545,22 +528,15 @@ def _build_state_refresh_output_projections( if autonomous_replan_recorded_requested or replan_ack.get("recorded") is True: index_record["autonomous_replan_ack"] = replan_ack if replan_ack.get("requested_classification"): - index_record["requested_classification"] = replan_ack[ - "requested_classification" - ] + index_record["requested_classification"] = replan_ack["requested_classification"] agent_vision = record.get("agent_vision") if isinstance(agent_vision, dict): index_record["agent_vision"] = { field: agent_vision.get(field) for field in ( - "schema_version", - "agent_id", - "state", - "vision_patch", - "todo_delta", - "fallback_declarations", - "vision_budget", + "schema_version", "agent_id", "state", "vision_patch", + "todo_delta", "fallback_declarations", "vision_budget", ) } if isinstance(agent_vision.get("path_delta"), dict): @@ -584,42 +560,26 @@ def _build_state_refresh_output_projections( "runtime_root": str(runtime_root), "project": str(project) if project else None, } - payload.update( - { - field: record.get(field) - for field in ( - "goal_id", - "classification", - "progress_scope", - "agent_id", - "agent_lane", - ) - } - ) - payload.update( - { - "autonomous_replan_recorded": bool(replan_ack.get("recorded")), - "autonomous_replan_recorded_requested": autonomous_replan_recorded_requested, - "repair_delta_contract": replan_ack.get("delta_contract"), - "json_path": str(json_path), - "markdown_path": str(markdown_path), - "index_path": str(index_path), - } - ) - payload.update( - { - field: record.get(field) - for field in ( - "agent_vision", - "vision_checkpoint", - "recommended_action", - "recommended_action_source", - "active_state_next_action_update", - "generated_at", - "health_check", - ) - } - ) + payload.update({ + field: record.get(field) + for field in ("goal_id", "classification", "progress_scope", "agent_id", "agent_lane") + }) + payload.update({ + "autonomous_replan_recorded": bool(replan_ack.get("recorded")), + "autonomous_replan_recorded_requested": autonomous_replan_recorded_requested, + "repair_delta_contract": replan_ack.get("delta_contract"), + "json_path": str(json_path), + "markdown_path": str(markdown_path), + "index_path": str(index_path), + }) + payload.update({ + field: record.get(field) + for field in ( + "agent_vision", "vision_checkpoint", "recommended_action", + "recommended_action_source", "active_state_next_action_update", + "generated_at", "health_check", + ) + }) payload.update(record) return index_record, payload @@ -629,9 +589,7 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str: if recovery_markdown is not None: return recovery_markdown state = payload.get("state") if isinstance(payload.get("state"), dict) else {} - frontmatter = ( - state.get("frontmatter") if isinstance(state.get("frontmatter"), dict) else {} - ) + frontmatter = state.get("frontmatter") if isinstance(state.get("frontmatter"), dict) else {} lines = [ "# LoopX State Refresh", "", @@ -738,9 +696,7 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str: f"target_roles={','.join(projection_gap.get('target_roles') or [])}" ) if projection_gap.get("recommended_action"): - lines.append( - f"- state_projection_gap_action: {projection_gap.get('recommended_action')}" - ) + lines.append(f"- state_projection_gap_action: {projection_gap.get('recommended_action')}") next_action_update = ( payload.get("active_state_next_action_update") @@ -765,16 +721,8 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str: else {} ) if write_correctness: - intent = ( - write_correctness.get("write_intent") - if isinstance(write_correctness.get("write_intent"), dict) - else {} - ) - preview = ( - write_correctness.get("preview") - if isinstance(write_correctness.get("preview"), dict) - else {} - ) + intent = write_correctness.get("write_intent") if isinstance(write_correctness.get("write_intent"), dict) else {} + preview = write_correctness.get("preview") if isinstance(write_correctness.get("preview"), dict) else {} apply_result = ( write_correctness.get("apply_result") if isinstance(write_correctness.get("apply_result"), dict) @@ -788,11 +736,7 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str: f"non_destructive={preview.get('non_destructive')}" ) - global_sync = ( - payload.get("global_sync") - if isinstance(payload.get("global_sync"), dict) - else {} - ) + global_sync = payload.get("global_sync") if isinstance(payload.get("global_sync"), dict) else {} if global_sync: lines.extend( [ @@ -821,7 +765,9 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str: f"`{recommendation_resolution.get('settlement_alignment')}`" ) if recommendation_resolution.get("todo_id"): - lines.append(f"- todo_id: `{recommendation_resolution.get('todo_id')}`") + lines.append( + f"- todo_id: `{recommendation_resolution.get('todo_id')}`" + ) lines.append(str(payload.get("recommended_action") or "")) for heading, key in ( ("Next Action", "next_action"), @@ -890,9 +836,7 @@ def refresh_state_run( raise ValueError("--agent-lane requires --agent-id so the lane has an owner") normalized_progress_scope = normalize_progress_scope(progress_scope) normalized_delivery_batch_scale = ( - require_delivery_batch_scale(delivery_batch_scale).value - if delivery_batch_scale - else None + require_delivery_batch_scale(delivery_batch_scale).value if delivery_batch_scale else None ) normalized_delivery_outcome = ( require_delivery_outcome(delivery_outcome).value if delivery_outcome else None @@ -960,9 +904,7 @@ def refresh_state_run( }, ) if settlement_readback is None: - raise RuntimeError( - "exact settlement readback unexpectedly returned not-found" - ) + raise RuntimeError("exact settlement readback unexpectedly returned not-found") settlement_result = settlement_readback.identity if settlement_result.failure is not None: raise ValueError(settlement_result.failure.reason) @@ -1001,8 +943,7 @@ def refresh_state_run( ) return payload settlement_workspace_requirement = resolve_settlement_workspace_requirement( - delivery_workspace_causality, - settlement_binding_kind=settlement_identity.binding_kind.value, + delivery_workspace_causality, settlement_binding_kind=settlement_identity.binding_kind.value ) runtime_projection_route = resolve_runtime_projection_route( registry_path=registry_path, @@ -1038,9 +979,7 @@ def refresh_state_run( todo_id=(settlement_identity.todo_id if settlement_identity else None), agent_id=normalized_agent_id or None, ) - normalized_next_action = ( - normalize_next_action_text(next_action) if next_action else None - ) + normalized_next_action = normalize_next_action_text(next_action) if next_action else None registered_agents = registered_agents_for_goal(registry_goal) known_agents = {agent for agent in registered_agents if agent} multi_agent_goal = len(known_agents) > 1 @@ -1079,9 +1018,7 @@ def refresh_state_run( if normalized_progress_scope == GOAL_PROGRESS_SCOPE: if normalized_agent_lane: raise ValueError("--agent-lane requires --progress-scope agent_lane") - if ( - agent_vision_packet is not None or vision_unchanged_reason - ) and not normalized_agent_id: + if (agent_vision_packet is not None or vision_unchanged_reason) and not normalized_agent_id: raise ValueError("vision writeback requires --agent-id") agent_vision: dict[str, Any] | None = None existing_agent_vision: dict[str, Any] | None = None @@ -1170,13 +1107,15 @@ def refresh_state_run( existing_agent_vision=existing_agent_vision, agent_id=normalized_agent_id, dry_run=dry_run, - settlement_todo_id=( - settlement_identity.todo_id if settlement_identity else None + settlement_todo_id=(settlement_identity.todo_id if settlement_identity else None), + settlement_guard_scoped=( + settlement_replan_guard.get("scope") == "turn_guard" ), - settlement_guard_scoped=(settlement_replan_guard.get("scope") == "turn_guard"), settlement_guard_semantic_replan_obligation_id=( settlement_replan_guard.get("selected_obligation_id") - if isinstance(settlement_replan_guard.get("selected_obligation_id"), str) + if isinstance( + settlement_replan_guard.get("selected_obligation_id"), str + ) else None ), newest_first_runs=newest_first_runs, @@ -1243,9 +1182,13 @@ def refresh_state_run( else None ), ) - if peer_independent_worktree_required and ( - delivery_workspace is None - or delivery_workspace.get("workspace_kind") != "independent_git_worktree" + if ( + peer_independent_worktree_required + and ( + delivery_workspace is None + or delivery_workspace.get("workspace_kind") + != "independent_git_worktree" + ) ): raise ValueError( "accountable peer delivery must be refreshed from the independent " @@ -1312,7 +1255,9 @@ def refresh_state_run( if refresh_recovery: record["refresh_recovery"] = refresh_recovery if settlement_workspace_requirement: - record["settlement_workspace_requirement"] = settlement_workspace_requirement + record["settlement_workspace_requirement"] = ( + settlement_workspace_requirement + ) if autonomous_replan_recorded: if "autonomous_replan_ack" not in record: record["autonomous_replan_ack"] = { @@ -1327,9 +1272,7 @@ def refresh_state_run( autonomous_replan_frontier_identity ) if requested_classification != classification: - record["autonomous_replan_ack"]["requested_classification"] = ( - requested_classification - ) + record["autonomous_replan_ack"]["requested_classification"] = requested_classification record["autonomous_replan_noop"] = { "schema_version": REPAIR_NOOP_SCHEMA_VERSION, "classification": classification, @@ -1345,7 +1288,9 @@ def refresh_state_run( "source": "refresh_state_semantic_delta", }, ) - record["autonomous_replan_ack"]["semantic_delta"] = replan_semantic_delta + record["autonomous_replan_ack"]["semantic_delta"] = ( + replan_semantic_delta + ) if active_state_next_action_update: record["active_state_next_action_update"] = active_state_next_action_update compact_route = compact_runtime_projection_route(runtime_projection_route) @@ -1394,9 +1339,7 @@ def refresh_state_run( payload["usage"] = dict(record["usage"]) if dry_run: expected_write_scopes = ["runtime_history"] - if active_state_next_action_update and active_state_next_action_update.get( - "would_update" - ): + if active_state_next_action_update and active_state_next_action_update.get("would_update"): expected_write_scopes.insert(0, "active_state") if sync_global and route_status in {"resolved", "single_runtime"}: expected_write_scopes.append("global_registry") @@ -1411,40 +1354,31 @@ def refresh_state_run( if sync_global and route_status in {"resolved", "single_runtime"}: patch_parts.append("sync public-safe registry projection") elif sync_global: - patch_parts.append( - f"block global sync on {route_status} runtime projection route" - ) + patch_parts.append(f"block global sync on {route_status} runtime projection route") if shared_runtime_root: - patch_parts.append( - "project compact refresh to registered shared runtime" - ) - payload["local_state_write_correctness"] = ( - build_local_state_write_correctness_dry_run_packet( - goal_id=safe_goal_id, - writer_id=normalized_agent_id or "loopx.refresh-state", - write_class="refresh_state", - state_text=expected_write_state_text, - target_refs={ - "state_file_ref": "registry.goal.state_file", - "run_history_ref": "runtime.goal.runs", - "index_ref": "runtime.goal.runs.index", - "global_registry_ref": ( - "runtime.registry.global" - if sync_global - and route_status in {"resolved", "single_runtime"} - else None - ), - "shared_runtime_projection_ref": ( - "shared_runtime.goal.runs.index" - if shared_runtime_root - else None - ), - }, - patch_summary="; ".join(patch_parts), - expected_write_scopes=expected_write_scopes, - lease_ref=None, - projection_status_surface=f"refresh-state dry-run: {classification}", - ) + patch_parts.append("project compact refresh to registered shared runtime") + payload["local_state_write_correctness"] = build_local_state_write_correctness_dry_run_packet( + goal_id=safe_goal_id, + writer_id=normalized_agent_id or "loopx.refresh-state", + write_class="refresh_state", + state_text=expected_write_state_text, + target_refs={ + "state_file_ref": "registry.goal.state_file", + "run_history_ref": "runtime.goal.runs", + "index_ref": "runtime.goal.runs.index", + "global_registry_ref": ( + "runtime.registry.global" + if sync_global and route_status in {"resolved", "single_runtime"} + else None + ), + "shared_runtime_projection_ref": ( + "shared_runtime.goal.runs.index" if shared_runtime_root else None + ), + }, + patch_summary="; ".join(patch_parts), + expected_write_scopes=expected_write_scopes, + lease_ref=None, + projection_status_surface=f"refresh-state dry-run: {classification}", ) if not dry_run: runs_dir.mkdir(parents=True, exist_ok=True) @@ -1454,17 +1388,12 @@ def refresh_state_run( payload["json_path"] = str(json_path) payload["markdown_path"] = str(markdown_path) json_path.write_text( - json.dumps(record, ensure_ascii=False, indent=2, allow_nan=False) - + "\n", + json.dumps(record, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8", ) - markdown_path.write_text( - render_state_refresh_markdown(payload) + "\n", encoding="utf-8" - ) + markdown_path.write_text(render_state_refresh_markdown(payload) + "\n", encoding="utf-8") with index_path.open("a", encoding="utf-8") as f: - f.write( - json.dumps(index_record, ensure_ascii=False, allow_nan=False) + "\n" - ) + f.write(json.dumps(index_record, ensure_ascii=False, allow_nan=False) + "\n") if sync_global and route_status in {"missing", "ambiguous"}: payload["ok"] = False payload["partial_write"] = not dry_run