diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md index db1e310ed0..0902a84bc7 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md @@ -2489,9 +2489,19 @@ The next complete stage packages are: and reads against actual callers. Status/attention now joins `todo list` in reading canonical Todo summaries after promotion, without requiring the Markdown file. Missing providers fail closed and empty canonical collections - never revive legacy Todos. This is consumer progress, not promotion proof: - Turn, quota, planning, standing decisions, leases and monitor writeback still - need their own parity inventory. Read authority does not grant writeback. + never revive legacy Todos. Refresh recommendation, repair/replan qualification, + completion-validation accountability, Todo-add replan binding and guided-start + frontier now share that canonical source. A refresh reads one snapshot and + passes it through its decisions rather than rereading a changing provider or + Markdown at each gate. Provider failure aborts; an empty snapshot is not a + fallback signal. This is consumer progress, not promotion proof: Turn/quota, + standing decisions, leases, monitor writeback, shared-goal alignment and + amendment revision bases still need their own parity inventory. Read authority + does not grant writeback. Source/display independence is tested with the + shared production-scale fixture and real FileAuthorityStore; these reads do + not establish freshness/CAS for a later business commit or change provider + defaults. Next Action narrative remains independent of Todo authority. + Lifecycle admission and the preauthorized terminal fence now share the TS owner across legacy writers and native terminal transactions; the replaced Python rules are removed without changing provider defaults or promotion. diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md index 02a83e38a3..9d9eb764c7 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md @@ -1975,8 +1975,15 @@ backend、实时双向同步或按命令拆开的权威；晋升后不支持的 terminal/successor/archive 路径。按实际 caller 盘点剩余公开 mutation 和 read。 status/attention 现在与 `todo list` 一样，在 promotion 后读 canonical Todo summary， 不要求 Markdown 文件存在；provider 缺失 fail closed，canonical 空集合不能复活旧 - Todo。这是 consumer 进展，不是 promotion 证明：Turn、quota、planning、standing - decision、lease、monitor writeback 仍需各自的 parity 清单。读权威不授予写回能力。 + Todo。Refresh 推荐、repair/replan 验收、completion-validation 问责、Todo-add replan + 绑定和 guided-start frontier 现已复用该 canonical 来源。一次 refresh 读取一份快照， + 传给各项决策，不在每个门禁重新读取变化中的 provider 或 Markdown；provider 故障 + 直接中止，空快照不是 fallback 信号。这是 consumer 进展，不是 promotion 证明： + Turn/quota、standing decision、lease、monitor writeback、shared-goal alignment 与 + amendment revision basis 仍需各自 parity 清单。读权威不授予写回能力。共用的复杂 + fixture 和真实 FileAuthorityStore 验证 source/display 独立性，但不证明后续业务 + commit 的 freshness/CAS，也不改变 provider 默认值。Next Action 正文仍独立于 Todo 权威。 + Lifecycle 准入及预授权 terminal fence 现由 legacy writer 与 native terminal transaction 共用 TS owner；删除对应 Python 规则，不改变 provider 默认或 promotion。 这不是完整 native 字段编辑：在 update 的字段、ownership、validation 和 monitor/resume diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index d057cb7784..17b5b66fdb 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -225,10 +225,18 @@ than extending these adapters field by field. TS owner and delete the replaced decisions in the same slice. Reuse the canonical Todo summary for reads: `todo list` and status/attention must not select stale Markdown or event Todos after promotion, even when the display - is missing or the canonical collection is empty. Audit Turn, quota, planning, - Dashboard and standing-decision consumers separately; fixing one does not - qualify all consumers. Prove real-entrypoint parity and unavailable-provider - rejection, not just transport snapshots. + is missing or the canonical collection is empty. Refresh now loads one + unbounded canonical Todo snapshot for recommendation, repair/replan + qualification and completion-validation accountability; Todo-add's replan + binding and guided-start's existing frontier use the same source adapter. + Their existing decision reducers remain owners: no second planning store or + permission rule is introduced. Legacy callers retain their parser contracts. + Audit Turn/quota, Dashboard, standing decisions, shared-goal alignment and + amendment revision bases separately; this closes the named planning callers, + not every consumer. Prove real-entrypoint parity and unavailable-provider + rejection, not just transport snapshots. Independently authored Next Action + remains narrative, not a Todo import. Missing display does not authorize + reconstructing narrative or weaken an accountable completion fence. 2. **Make the display a recoverable one-way projection.** Reuse the canonical journal/outbox and Todo-section renderer. Keep human narrative, source revision, idempotent delivery and actionable pending repair. A failed render diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index c4d8055549..607a15e1ab 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -174,8 +174,14 @@ crossing 一起折叠进完整事务，不能沿着 adapter 逐字段继续加 字段编辑、monitor、lease、event caller，把规则迁入既有 TS owner，并在同一切片 删除被替代的 decision。读取复用 canonical Todo summary：promotion 后，`todo list` 和 status/attention 不得选择陈旧 Markdown/event Todo；投影缺失、canonical 集合为空 - 也不例外。另行审计 Turn、quota、planning、Dashboard、standing-decision consumer； - 修好一条不等于全部合格。通过真实入口验证 parity 和 provider 故障拒绝，不能只比传输快照。 + 也不例外。Refresh 现在只读一次无截断 canonical Todo 快照，供推荐、repair/replan + 验收和 completion-validation 问责共同使用；Todo-add 的 replan 绑定和 guided-start + 的既有 frontier 也复用同一来源适配器。既有 decision reducer 仍是规则 owner，不增加 + 第二份 planning store 或权限规则；旧模式保留原 parser 合同。Turn/quota、Dashboard、 + standing decision、shared-goal alignment 与 amendment revision basis 另行审计， + 不能将这些具名调用链的闭合等同于全部 consumer 合格。通过真实入口验证 parity 和 + provider 故障拒绝。独立维护的 Next Action 仍是正文，不导入 Todo；展示缺失不授权 + 重建丢失正文，也不能削弱完成验收门禁。 2. **把展示闭合为可恢复的单向投影。** 复用 canonical journal/outbox 与 Todo-section renderer，保留人工叙述、来源 revision、幂等交付和可操作的 pending repair。 渲染失败不能撤销已提交事务，也不能授权 Markdown fallback；恢复投影不能重跑业务操作。 diff --git a/loopx/bootstrap_command_pack.py b/loopx/bootstrap_command_pack.py index 10b8112bfe..6f31ea9a32 100644 --- a/loopx/bootstrap_command_pack.py +++ b/loopx/bootstrap_command_pack.py @@ -1527,6 +1527,7 @@ def rerun_start_goal(selected_agent_id: str) -> str: project_connection, resolved_goal_id=str(command_pack.get("goal_id") or ""), effective_agent_id=str(command_pack.get("agent_id") or "") or None, + runtime_root_arg=runtime_root_arg, ) if isinstance(project_connection, dict) and not isinstance(identity_selection_gate, dict) diff --git a/loopx/cli_commands/todo.py b/loopx/cli_commands/todo.py index b8837beba4..1fb999d3e6 100644 --- a/loopx/cli_commands/todo.py +++ b/loopx/cli_commands/todo.py @@ -4,6 +4,7 @@ from collections.abc import Callable, Sequence from pathlib import Path +from ..control_plane.coordination.local_authority import read_canonical_todo_fields_if_promoted from ..control_plane.todos.contract import ( TODO_TASK_CLASS_ADVANCEMENT, normalize_todo_continuation_policy, @@ -148,11 +149,16 @@ def _validated_replan_successor_obligation( ) registry = load_registry(registry_path) runtime_root = resolve_runtime_root(registry, runtime_root_arg) - _, _, state_text, _ = resolve_todo_state( - registry_path=registry_path, - goal_id=args.goal_id, - **_todo_path_args(args), + todo_fields = read_canonical_todo_fields_if_promoted( + runtime_root=runtime_root, goal_id=args.goal_id, ) + state_text = "" + if todo_fields is None: + _, _, state_text, _ = resolve_todo_state( + registry_path=registry_path, + goal_id=args.goal_id, + **_todo_path_args(args), + ) existing_runs, _ = load_index( runtime_root / "goals" / args.goal_id / "runs" / "index.jsonl" ) @@ -176,6 +182,7 @@ def _validated_replan_successor_obligation( None, ) obligation, _ = qualify_replan_writeback( + todo_fields=todo_fields, newest_first_runs=newest_first_runs, state_text=state_text, agent_id=args.claimed_by, diff --git a/loopx/control_plane/coordination/local_authority.py b/loopx/control_plane/coordination/local_authority.py index eb991722b4..0fbef2ef01 100644 --- a/loopx/control_plane/coordination/local_authority.py +++ b/loopx/control_plane/coordination/local_authority.py @@ -13,7 +13,7 @@ from uuid import uuid4 from ...agent_registry import registered_agent_ids_from_registry -from ...state_refresh import now_local +from ..runtime.time import now_local_iso as now_local from ..effect_runtime import effect_runtime_result from .coordination_state_contract import ( TODO_CANONICAL_READ_RECORD_SCHEMA_VERSION, @@ -233,6 +233,23 @@ def read_canonical_todos_if_promoted( return payload +def read_canonical_todo_fields_if_promoted( + *, runtime_root: Path, goal_id: str, + rollout_events: list[dict[str, Any]] | None = None, +) -> dict[str, Any] | None: + """Read one unbounded planning snapshot; None alone permits legacy parsing. + + Empty canonical state is authoritative. Provider failures propagate; this + read neither repairs Markdown nor grants mutation/promotion authority. + Callers pass the same fields to all decisions in one planning operation. + """ + canonical = read_canonical_todos_if_promoted(runtime_root=runtime_root, goal_id=goal_id) + return ( + canonical_todo_summary_fields(canonical["todos"], rollout_events=rollout_events) + if canonical is not None else None + ) + + def canonical_todo_summary_fields( todos: list[dict[str, Any]], *, diff --git a/loopx/control_plane/goals/start_goal_todo_delta.py b/loopx/control_plane/goals/start_goal_todo_delta.py index 6a5b7ddd90..2bbb0d5206 100644 --- a/loopx/control_plane/goals/start_goal_todo_delta.py +++ b/loopx/control_plane/goals/start_goal_todo_delta.py @@ -15,6 +15,8 @@ from pathlib import Path from typing import Any +from ..coordination.local_authority import read_canonical_todo_fields_if_promoted +from ...paths import resolve_runtime_root from ...control_plane.todos.active_state_todo_parser import parse_active_state_todos from ...control_plane.todos.contract import ( TODO_TASK_CLASS_ADVANCEMENT, @@ -33,6 +35,7 @@ def existing_runnable_agent_frontier( *, resolved_goal_id: str, effective_agent_id: str | None, + runtime_root_arg: str | None = None, ) -> list[dict[str, Any]] | None: """Runnable advancement agent Todos already present in the goal's state. @@ -41,9 +44,11 @@ def existing_runnable_agent_frontier( whose resume condition is satisfied (or absent) enter the frontier. Blocked, deferred, monitor, blocker, resume-blocked, or peer-claimed Todos never enter the frontier. Returns ``None`` whenever the frontier - cannot be proven (not connected, unknown goal, missing or unreadable - state file, or nothing runnable), so callers keep the unconditional - planning contract — fail-closed. + cannot be proven (not connected, unknown goal, missing legacy state, or + nothing runnable), so callers keep the unconditional planning contract. + After cutover only canonical records count; provider unavailability raises + instead of being mistaken for an empty frontier. A missing display is safe + to ignore, not permission to reconstruct its non-Todo narrative. """ if inspection.get("connection_state") != "connected": return None @@ -63,18 +68,22 @@ def existing_runnable_agent_frontier( Path(str(inspection.get("project") or "")), registry_goal.get("state_file"), ) - if state_file is None or not state_file.is_file(): - return None - try: - state_text = state_file.read_text(encoding="utf-8") - except OSError: - return None - parsed = parse_active_state_todos( - state_text, - goal=registry_goal, - state_path=state_file, - item_limit=None, + parsed = read_canonical_todo_fields_if_promoted( + runtime_root=resolve_runtime_root( + registry_payload or {}, runtime_root_arg, registry_path=registry_path, + ), + goal_id=resolved_goal_id, ) + if parsed is None: + if state_file is None or not state_file.is_file(): + return None + try: + state_text = state_file.read_text(encoding="utf-8") + except OSError: + return None + parsed = parse_active_state_todos( + state_text, goal=registry_goal, state_path=state_file, item_limit=None, + ) agent_summary = parsed.get("agent_todos") if isinstance(parsed, dict) else None items = ( agent_summary.get("items", []) if isinstance(agent_summary, dict) else [] @@ -86,19 +95,12 @@ def existing_runnable_agent_frontier( and todo_item_is_actionable_open(item) and item.get("task_class") == TODO_TASK_CLASS_ADVANCEMENT ] - if effective_agent_id: - runnable = [ - item - for item in runnable - if not item.get("claimed_by") - or str(item.get("claimed_by")) == effective_agent_id - ] - else: - runnable = [ - item - for item in runnable - if not item.get("claimed_by") - ] + runnable = [ + item + for item in runnable + if not item.get("claimed_by") + or (effective_agent_id and str(item.get("claimed_by")) == effective_agent_id) + ] return runnable or None diff --git a/loopx/control_plane/todos/completion_validation_accountability.py b/loopx/control_plane/todos/completion_validation_accountability.py index 99eefd448c..a37c4a68c9 100644 --- a/loopx/control_plane/todos/completion_validation_accountability.py +++ b/loopx/control_plane/todos/completion_validation_accountability.py @@ -2,6 +2,8 @@ from __future__ import annotations +from typing import Any + from .active_state_todo_parser import parse_active_state_todos from .completion_validation_projection import pending_completion_validation_todo @@ -11,10 +13,12 @@ def require_accountable_completion_validation( *, todo_id: str | None, agent_id: str | None, + todo_fields: dict[str, Any] | None = None, ) -> None: """Reject accountable evidence while its exact validation Todo is open.""" - summary = parse_active_state_todos(state_text, item_limit=None).get("agent_todos") + fields = todo_fields if todo_fields is not None else parse_active_state_todos(state_text, item_limit=None) + summary = fields.get("agent_todos") pending = pending_completion_validation_todo( summary, todo_id=todo_id, diff --git a/loopx/control_plane/work_items/refresh_recommendation.py b/loopx/control_plane/work_items/refresh_recommendation.py index 0ee8653cf3..f3ec0810cb 100644 --- a/loopx/control_plane/work_items/refresh_recommendation.py +++ b/loopx/control_plane/work_items/refresh_recommendation.py @@ -5,11 +5,13 @@ from typing import Any from ..agents.agent_lane_recommendation import build_agent_lane_next_action +from ..coordination.local_authority import read_canonical_todo_fields_if_promoted from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result from ..todos.active_state_todo_parser import parse_active_state_todos from ..todos.contract import normalize_todo_id from ...feedback import validate_local_control_text from ...state_projection import active_state_next_action_entries +from ...rollout_event_log import load_rollout_events, rollout_event_log_path REFRESH_RECOMMENDATION_REQUEST_SCHEMA_VERSION = "refresh_recommendation_request_v0" REFRESH_RECOMMENDATION_SCHEMA_VERSION = "refresh_recommendation_v0" @@ -24,6 +26,31 @@ RECOMMENDED_ACTION_SOURCE_DEFAULT = "default_refresh_action" +def load_refresh_planning_source( + runtime_root: Path, + goal_id: str, + state_path: Path, + *, + require_display: bool, +) -> tuple[str, list[dict[str, Any]], dict[str, Any] | None]: + """Read one shared planning snapshot without repairing its display. + + Canonical Todo availability permits observation without Markdown, not an + edit of missing Next Action narrative. Provider failures propagate. + """ + events = load_rollout_events(rollout_event_log_path(runtime_root, goal_id)) + fields = read_canonical_todo_fields_if_promoted( + runtime_root=runtime_root, goal_id=goal_id, rollout_events=events, + ) + try: + text = state_path.read_text(encoding="utf-8") + except FileNotFoundError: + if fields is None or require_display: + raise FileNotFoundError(f"state file does not exist: {state_path}") from None + text = "" + return text, events, fields + + def _first_valid_action(values: list[str]) -> str | None: for value in values: try: @@ -41,9 +68,10 @@ def _agent_todo_summary( state_path: Path | None, settlement_todo_id: str | None, rollout_events: list[dict[str, Any]] | None, + todo_fields: dict[str, Any] | None, ) -> dict[str, Any] | None: preferred = {settlement_todo_id} if settlement_todo_id else None - parsed = parse_active_state_todos( + parsed = todo_fields if todo_fields is not None else parse_active_state_todos( state_text, goal=registry_goal, state_path=state_path, @@ -95,6 +123,7 @@ def resolve_refresh_recommendation( registry_goal: dict[str, Any] | None = None, state_path: Path | None = None, rollout_events: list[dict[str, Any]] | None = None, + todo_fields: dict[str, Any] | None = None, ) -> dict[str, Any]: """Adapt canonical Todo facts into the TS-owned refresh read reducer.""" @@ -121,6 +150,7 @@ def resolve_refresh_recommendation( state_path=state_path, settlement_todo_id=settlement_todo_id, rollout_events=rollout_events, + todo_fields=todo_fields, ) lane_candidate = build_agent_lane_next_action( agent_identity={"agent_id": agent_id} if agent_id else None, diff --git a/loopx/control_plane/work_items/semantic_replan_writeback.py b/loopx/control_plane/work_items/semantic_replan_writeback.py index 48db421e62..0b7b225a8c 100644 --- a/loopx/control_plane/work_items/semantic_replan_writeback.py +++ b/loopx/control_plane/work_items/semantic_replan_writeback.py @@ -193,6 +193,7 @@ def qualify_replan_writeback( agent_vision: dict[str, Any] | None = None, completion_todo_id: str | None = None, completion_turn_key: str | None = None, + todo_fields: dict[str, Any] | None = None, ) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: """Return the shared open obligation and the writeback's typed delta. @@ -204,7 +205,7 @@ def qualify_replan_writeback( safe_agent_id = str(agent_id or "").strip() if not safe_agent_id: return None, None - todo_projection = parse_active_state_todos( + todo_projection = todo_fields if todo_fields is not None else parse_active_state_todos( state_text, item_limit=None, goal={**(registry_goal or {}), "latest_runs": list(newest_first_runs or [])}, ) @@ -337,6 +338,7 @@ def enforce_open_replan_writeback( completion_turn_key: str | None = None, guard_scoped: bool = False, guard_semantic_replan_obligation_id: str | None = None, + todo_fields: dict[str, Any] | None = None, ) -> dict[str, Any] | None: """Fail closed unless concrete typed evidence satisfies the selected replan. @@ -357,6 +359,7 @@ def enforce_open_replan_writeback( agent_vision=agent_vision, completion_todo_id=completion_todo_id, completion_turn_key=completion_turn_key, + todo_fields=todo_fields, ) if not obligation: return None @@ -406,6 +409,7 @@ def qualify_refresh_replan_writeback( completion_turn_key: str | None, classification: str, delivery_outcome: str | None, + todo_fields: dict[str, Any] | None = None, ) -> RefreshReplanQualification: """Qualify one refresh's replan delta and accountable settlement outcome.""" @@ -420,8 +424,9 @@ def qualify_refresh_replan_writeback( active_state_next_action_update=active_state_next_action_update, agent_vision=agent_vision, existing_agent_vision=existing_agent_vision, - agent_todo_summary=parse_active_state_todos( - state_text, item_limit=None + agent_todo_summary=( + todo_fields if todo_fields is not None + else parse_active_state_todos(state_text, item_limit=None) ).get("agent_todos"), agent_id=agent_id or None, dry_run=dry_run, @@ -466,6 +471,7 @@ def qualify_refresh_replan_writeback( completion_todo_id=completion_todo_id, completion_turn_key=completion_turn_key, guard_scoped=settlement_guard_scoped, + todo_fields=todo_fields, guard_semantic_replan_obligation_id=( settlement_guard_semantic_replan_obligation_id ), diff --git a/loopx/state_refresh.py b/loopx/state_refresh.py index 6ea7e092d2..a9060498c3 100644 --- a/loopx/state_refresh.py +++ b/loopx/state_refresh.py @@ -57,6 +57,7 @@ RECOMMENDED_ACTION_SOURCE_SETTLEMENT_BOUND_TODO as RECOMMENDED_ACTION_SOURCE_SETTLEMENT_BOUND_TODO, derive_recommended_action as derive_recommended_action, derive_recommended_action_with_source as derive_recommended_action_with_source, + load_refresh_planning_source, resolve_refresh_recommendation, ) from .control_plane.runtime.shared_runtime_refresh_projection import ( @@ -98,7 +99,6 @@ from .control_plane.todos.completion_validation_accountability import ( require_accountable_completion_validation, ) -from .rollout_event_log import load_rollout_events, rollout_event_log_path DEFAULT_REFRESH_CLASSIFICATION = "state_refreshed" GOAL_PROGRESS_SCOPE = "goal" @@ -973,13 +973,14 @@ def refresh_state_run( project_override=project, state_file_override=state_file, ) - if not resolved_state_file.exists(): - raise FileNotFoundError(f"state file does not exist: {resolved_state_file}") - state_text = resolved_state_file.read_text(encoding="utf-8") + state_text, planning_events, todo_fields = load_refresh_planning_source( + runtime_root, safe_goal_id, resolved_state_file, require_display=bool(next_action) + ) expected_write_state_text = state_text if normalized_delivery_outcome in ACCOUNTABLE_DELIVERY_OUTCOMES: require_accountable_completion_validation( state_text, + todo_fields=todo_fields, todo_id=(settlement_identity.todo_id if settlement_identity else None), agent_id=normalized_agent_id or None, ) @@ -1078,6 +1079,7 @@ def refresh_state_run( recommendation_resolution = resolve_refresh_recommendation( state_text, + todo_fields=todo_fields, explicit_action=recommended_action, agent_id=normalized_agent_id or None, settlement_identity=( @@ -1086,10 +1088,7 @@ def refresh_state_run( registry_goal=registry_goal, state_path=resolved_state_file, rollout_events=( - load_rollout_events(rollout_event_log_path(runtime_root, safe_goal_id)) - if not recommended_action - and (normalized_agent_id or settlement_identity is not None) - else None + planning_events if normalized_agent_id or settlement_identity is not None else None ), ) action = str(recommendation_resolution["recommended_action"]) @@ -1104,6 +1103,7 @@ def refresh_state_run( else {} ) replan_qualification = qualify_refresh_replan_writeback( + todo_fields=todo_fields, autonomous_replan_recorded=autonomous_replan_recorded, requested_delta_kinds=normalized_repair_delta_kinds, active_state_next_action_update=active_state_next_action_update, diff --git a/tests/control_plane/test_canonical_planning_consumers.py b/tests/control_plane/test_canonical_planning_consumers.py new file mode 100644 index 0000000000..1b70cfb857 --- /dev/null +++ b/tests/control_plane/test_canonical_planning_consumers.py @@ -0,0 +1,525 @@ +"""Planning consumers must use provider state, not its stale display copy.""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +import pytest +from canonical_authority_fixture import initialize_canonical_authority + +from loopx.control_plane.coordination.local_authority import ( + LocalCoordinationAuthorityUnavailable, + canonical_todo_summary_fields, + read_canonical_todos_if_promoted, +) +from loopx.control_plane.coordination.runtime_shadow import ( + build_todo_runtime_shadow_projection, +) +from loopx.control_plane.goals.start_goal_todo_delta import ( + existing_runnable_agent_frontier, +) +from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos +from loopx.control_plane.todos.completion_validation_accountability import ( + require_accountable_completion_validation, +) +from loopx.control_plane.work_items.refresh_recommendation import ( + resolve_refresh_recommendation, +) +from loopx.control_plane.work_items.semantic_replan_writeback import ( + qualify_replan_writeback, +) +from loopx.state_refresh import refresh_state_run + + +def _refresh(registry: Path, **overrides: object) -> dict: + return refresh_state_run( + **{ + "registry_path": registry, + "runtime_root_override": str(registry.parent / "runtime"), + "goal_id": "goal-a", + "project": registry.parent, + "state_file": None, + "classification": "state_refreshed", + "recommended_action": None, + "delivery_batch_scale": "single_surface", + "delivery_outcome": "surface_only", + "agent_id": "agent-a", + "dry_run": True, + "sync_global": False, + **overrides, + } + ) + + +def _state(*, done: bool = False, text: str = "Canonical work") -> str: + return ( + "# Goal\n\n## Agent Todo\n\n" + f"- [{'x' if done else ' '}] {text}\n" + " \n" + ) + + +def _fixture(root: Path, *, state: str | None = None) -> tuple[Path, Path, dict]: + root.mkdir(parents=True, exist_ok=True) + state_path = root / "ACTIVE_GOAL_STATE.md" + state_path.write_text(state if state is not None else _state(), encoding="utf-8") + goal = { + "id": "goal-a", + "repo": str(root), + "state_file": str(state_path), + "coordination": {"registered_agents": ["agent-a", "agent-b"]}, + } + runtime = root / "runtime" + registry = root / "registry.json" + registry.write_text( + json.dumps( + { + "schema_version": 1, + "common_runtime_root": str(runtime), + "goals": [goal], + } + ), + encoding="utf-8", + ) + return registry, state_path, goal + + +def _promote(registry: Path, path: Path, goal: dict) -> dict: + fields = parse_active_state_todos(path.read_text(), goal=goal, item_limit=None) + todos = [ + item + for key in ("user_todos", "agent_todos") + for item in fields.get(key, {}).get("items", []) + ] + runtime = registry.parent / "runtime" + initialize_canonical_authority( + runtime, + "goal-a", + build_todo_runtime_shadow_projection( + goal_id="goal-a", + todos=todos, + handoff_mode="soft_claim", + ), + state_path=path, + ) + loaded = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a") + assert loaded is not None + return canonical_todo_summary_fields(loaded["todos"]) + + +@pytest.mark.parametrize("display", ["stale", "missing", "empty"]) +def test_start_frontier_uses_real_provider_and_never_repairs_display( + tmp_path: Path, display: str +) -> None: + registry, path, goal = _fixture(tmp_path) + _promote(registry, path, goal) + if display == "missing": + path.unlink() + else: + path.write_text(_state(done=True, text="Stale display") if display == "stale" else "") + before = path.read_bytes() if path.exists() else None + frontier = existing_runnable_agent_frontier( + { + "connection_state": "connected", + "registry": str(registry), + "project": str(tmp_path), + }, + resolved_goal_id="goal-a", + effective_agent_id="agent-a", + ) + assert frontier is not None + assert [(item["todo_id"], item["text"]) for item in frontier] == [ + ("todo_selected", "Canonical work") + ] + assert (path.read_bytes() if path.exists() else None) == before + + +def test_empty_canonical_frontier_does_not_resurrect_markdown(tmp_path: Path) -> None: + registry, path, goal = _fixture(tmp_path, state="# Goal\n") + _promote(registry, path, goal) + path.write_text(_state()) + assert ( + existing_runnable_agent_frontier( + { + "connection_state": "connected", + "registry": str(registry), + "project": str(tmp_path), + }, + resolved_goal_id="goal-a", + effective_agent_id="agent-a", + ) + is None + ) + + +def test_provider_failure_is_not_permission_to_plan_again(tmp_path: Path) -> None: + registry, path, goal = _fixture(tmp_path) + _promote(registry, path, goal) + # Preserve the cutover fence but remove only this disposable provider head. + head = next((tmp_path / "runtime/authority/file-v0").glob("authority-store-*.json"), None) + assert head is not None + head.rename(head.with_suffix(".unavailable")) + with pytest.raises(LocalCoordinationAuthorityUnavailable): + existing_runnable_agent_frontier( + { + "connection_state": "connected", + "registry": str(registry), + "project": str(tmp_path), + }, + resolved_goal_id="goal-a", + effective_agent_id="agent-a", + ) + with pytest.raises(LocalCoordinationAuthorityUnavailable): + _refresh(registry, recommended_action="An explicit action is not a provider fallback") + + +def test_recommendation_uses_canonical_claim_and_receipt_binding( + tmp_path: Path, +) -> None: + registry, path, goal = _fixture(tmp_path) + fields = _promote(registry, path, goal) + identity = { + "goal_id": "goal-a", + "agent_id": "agent-a", + "todo_id": "todo_selected", + "turn_instance_id": "turn-a", + "effect_id": "goal-a:agent-a:todo_selected:turn-a", + } + result = resolve_refresh_recommendation( + _state(done=True, text="Stale copy"), + agent_id="agent-a", + registry_goal=goal, + settlement_identity=identity, + todo_fields=fields, + ) + assert result["recommended_action"] == "Canonical work" + assert result["recommended_action_source"] == "settlement_bound_todo" + assert result["settlement_alignment"] == "exact" + empty = resolve_refresh_recommendation(_state(), agent_id="agent-a", todo_fields={}) + assert empty["recommended_action_source"] == "default_refresh_action" + + +def test_completion_fence_uses_canonical_validation_not_display(tmp_path: Path) -> None: + state = _state().replace("status=open", "status=open validation_command=pytest") + registry, path, goal = _fixture(tmp_path, state=state) + fields = _promote(registry, path, goal) + with pytest.raises(ValueError, match="completion validation"): + require_accountable_completion_validation( + "", + todo_id="todo_selected", + agent_id="agent-a", + todo_fields=fields, + ) + require_accountable_completion_validation( + state, + todo_id="todo_selected", + agent_id="agent-a", + todo_fields={}, + ) + + +def _completed_chain() -> str: + state = "## Agent Todo\n" + for index in range(3): + state += ( + f"- [x] [P1] Completed slice {index}\n" + f" \n" + ) + state += "- [x] [P1] Lineage anchor\n \n" + return state + + +def test_replan_completion_cadence_cannot_be_hidden_by_stale_display( + tmp_path: Path, +) -> None: + state = _completed_chain() + registry, path, goal = _fixture(tmp_path, state=state) + goal["execution_profile"] = {"replan_after_completed_todos": 3} + fields = _promote(registry, path, goal) + arguments = { + "newest_first_runs": [], + "agent_id": "agent-a", + "goal_id": "goal-a", + "registry_goal": goal, + } + obligation, _ = qualify_replan_writeback(state_text="", todo_fields=fields, **arguments) + assert obligation is not None + assert any( + trigger["kind"] == "vision_outcome_checkpoint_required" + for trigger in obligation["triggers"] + ) + empty, _ = qualify_replan_writeback(state_text=state, todo_fields={}, **arguments) + assert empty is None + + +def test_todo_add_replan_guard_binds_canonical_obligation_without_display( + tmp_path: Path, +) -> None: + from argparse import Namespace + from loopx.cli_commands.todo import _validated_replan_successor_obligation + + registry, path, goal = _fixture(tmp_path, state=_completed_chain()) + goal["execution_profile"] = {"replan_after_completed_todos": 3} + payload = json.loads(registry.read_text()) + payload["goals"] = [goal] + registry.write_text(json.dumps(payload)) + fields = _promote(registry, path, goal) + obligation, _ = qualify_replan_writeback( + state_text="", + newest_first_runs=[], + agent_id="agent-a", + goal_id="goal-a", + registry_goal=goal, + todo_fields=fields, + ) + assert obligation is not None + args = Namespace( + replan_obligation_id=obligation["obligation_id"], + role="agent", + task_class="advancement_task", + claimed_by="agent-a", + action_kind="implementation", + monitor_target_key="feature-a", + explore_result_node_refs=[], + goal_id="goal-a", + ) + path.unlink() + assert ( + _validated_replan_successor_obligation(args, registry_path=registry, runtime_root_arg=None) + == obligation["obligation_id"] + ) + args.replan_obligation_id = "obsolete-obligation" + with pytest.raises(ValueError, match="does not match"): + _validated_replan_successor_obligation(args, registry_path=registry, runtime_root_arg=None) + assert not path.exists() + + +@pytest.mark.parametrize("promoted", [False, True]) +def test_public_refresh_retains_legacy_parity_and_uses_one_provider_read( + tmp_path: Path, monkeypatch, promoted: bool +) -> None: + from loopx.control_plane.coordination import local_authority + + registry, path, goal = _fixture(tmp_path) + if promoted: + _promote(registry, path, goal) + path.write_text(_state(done=True, text="Stale work")) + original = local_authority.read_canonical_todos_if_promoted + reads = [] + + def read(**kwargs): + result = original(**kwargs) + reads.append(result) + return result + + monkeypatch.setattr(local_authority, "read_canonical_todos_if_promoted", read) + result = _refresh(registry) + assert "Canonical work" in json.dumps(result) + assert len(reads) == 1 + assert (reads[0] is not None) == promoted + + +@pytest.mark.parametrize("promoted", [False, True]) +def test_refresh_todo_text_is_record_content_not_an_artifact_path( + tmp_path: Path, promoted: bool, +) -> None: + text = "../../escape.json" + registry, path, goal = _fixture(tmp_path, state=_state(text=text)) + if promoted: + _promote(registry, path, goal) + before = path.read_bytes() + result = _refresh(registry, dry_run=False) + runs = tmp_path / "runtime/goals/goal-a/runs" + for key in ("json_path", "markdown_path", "index_path"): + artifact = Path(result[key]) + assert artifact.parent == runs + assert artifact.is_file() + record = json.loads(Path(result["json_path"]).read_text()) + assert text in json.dumps(record) + assert not list(tmp_path.rglob("escape.json")) + assert path.read_bytes() == before + + +@pytest.mark.parametrize("promoted", [False, True]) +def test_public_refresh_missing_projection_is_readable_not_implicitly_rebuilt( + tmp_path: Path, promoted: bool, +) -> None: + registry, path, goal = _fixture(tmp_path) + if promoted: + _promote(registry, path, goal) + path.unlink() + if promoted: + result = _refresh(registry) + assert "Canonical work" in json.dumps(result) + else: + with pytest.raises(FileNotFoundError): + _refresh(registry) + assert not path.exists() + with pytest.raises(FileNotFoundError): + _refresh(registry, next_action="Replace the missing narrative", progress_scope="goal") + assert not path.exists() + + +def test_committed_refresh_records_canonical_recommendation_without_rewriting_display( + tmp_path: Path, +) -> None: + registry, path, goal = _fixture(tmp_path) + _promote(registry, path, goal) + path.write_text(_state(done=True, text="Stale display")) + before = read_canonical_todos_if_promoted(runtime_root=tmp_path / "runtime", goal_id="goal-a") + display = path.read_bytes() + _refresh(registry, dry_run=False) + runs = (tmp_path / "runtime/goals/goal-a/runs/index.jsonl").read_text() + assert "Canonical work" in runs + assert path.read_bytes() == display + assert ( + read_canonical_todos_if_promoted(runtime_root=tmp_path / "runtime", goal_id="goal-a") + == before + ) + + +def test_public_refresh_keeps_validation_gate_even_with_explicit_recommendation( + tmp_path: Path, +) -> None: + state = _state().replace("status=open", "status=open validation_command=pytest") + registry, path, goal = _fixture(tmp_path, state=state) + _promote(registry, path, goal) + path.write_text(_state(done=True)) + with pytest.raises(ValueError, match="completion validation"): + _refresh( + registry, + recommended_action="Explicit recommendation", + delivery_outcome="outcome_progress", + ) + + +def test_production_scale_snapshot_reaches_all_planning_consumers( + tmp_path: Path, +) -> None: + generated = subprocess.run( + [ + "node", + "--no-warnings", + "--experimental-strip-types", + "--input-type=module", + "-e", + "import {productionScaleCoordinationFixture} from './tests/control_plane_ts/production_scale_coordination_fixture.ts';" + "process.stdout.write(JSON.stringify(productionScaleCoordinationFixture('goal-a')));", + ], + cwd=Path(__file__).resolve().parents[2], + capture_output=True, + text=True, + check=True, + timeout=30, + ) + fixture = json.loads(generated.stdout) + registry, path, goal = _fixture(tmp_path) + initialize_canonical_authority( + tmp_path / "runtime", "goal-a", fixture["projection"], state_path=path + ) + before = read_canonical_todos_if_promoted(runtime_root=tmp_path / "runtime", goal_id="goal-a") + assert before is not None + fields = canonical_todo_summary_fields(before["todos"]) + assert len(fields["agent_todos"]["items"]) == 256 + assert len(fields["user_todos"]["items"]) == 208 + path.unlink() + for agent, selected_id in [ + ("agent-a", fixture["completion_todo_id"]), + ("agent-b", fixture["supersede_todo_id"]), + ]: + identity = { + "goal_id": "goal-a", + "agent_id": agent, + "todo_id": selected_id, + "turn_instance_id": "turn-a", + "effect_id": f"goal-a:{agent}:{selected_id}:turn-a", + } + result = resolve_refresh_recommendation( + "", agent_id=agent, settlement_identity=identity, todo_fields=fields + ) + assert result["todo_id"] == selected_id + assert result["settlement_alignment"] == "exact" + frontier = existing_runnable_agent_frontier( + { + "connection_state": "connected", + "registry": str(registry), + "project": str(tmp_path), + }, + resolved_goal_id="goal-a", + effective_agent_id=agent, + ) + assert frontier + assert selected_id in {item["todo_id"] for item in frontier} + assert all( + item.get("claimed_by") in (None, agent) + and item["status"] == "open" + and item["task_class"] == "advancement_task" + for item in frontier + ) + # Replan consumes the complete same snapshot, including terminal records + # and standing user decisions. Arbitrary display Todo edits have no say. + baseline = qualify_replan_writeback( + state_text="", + newest_first_runs=[], + agent_id=agent, + goal_id="goal-a", + registry_goal=goal, + todo_fields=fields, + ) + assert ( + qualify_replan_writeback( + state_text=_state(), + newest_first_runs=[], + agent_id=agent, + goal_id="goal-a", + registry_goal=goal, + todo_fields=fields, + ) + == baseline + ) + with pytest.raises(ValueError, match="completion validation"): + require_accountable_completion_validation( + "", + todo_id=fixture["completion_todo_id"], + agent_id="agent-a", + todo_fields=fields, + ) + assert ( + read_canonical_todos_if_promoted(runtime_root=tmp_path / "runtime", goal_id="goal-a") + == before + ) + assert not path.exists() + + +def test_start_goal_packet_honors_runtime_override_and_canonical_frontier( + tmp_path: Path, +) -> None: + from loopx.bootstrap_command_pack import build_start_goal_guided_packet + + registry, path, goal = _fixture(tmp_path) + _promote(registry, path, goal) + project_registry = tmp_path / ".loopx/registry.json" + project_registry.parent.mkdir() + contents = json.loads(registry.read_text()) + contents["common_runtime_root"] = str(tmp_path / "unused-runtime") + project_registry.write_text(json.dumps(contents)) + path.write_text(_state(done=True, text="Stale copy")) + result = build_start_goal_guided_packet( + project=tmp_path, + goal_id="goal-a", + agent_id="agent-a", + cli_bin="loopx", + host_surface="codex-app", + goal_text="Continue bounded work", + available_capabilities=["network"], + runtime_root_arg=str(tmp_path / "runtime"), + ) + steps = result["guided_transaction"]["ordered_steps"] + delta = next(step for step in steps if step["id"] == "apply_todo_delta") + assert "Canonical work" in json.dumps(delta) + assert "Stale copy" not in json.dumps(delta)