diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 4101766d1f..8e42dab950 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -438,12 +438,15 @@ shipped Stage 2B cutovers are in place: zero. - Todo completion: TypeScript owns completion identity, terminal replay fence, validation declaration/effect planning, validation-receipt reduction, - continuation/recovery, and completion metadata in one transaction. A Todo - without declared validation, including a replay, uses one reduction. A real - caller-approved validation command remains an explicit Python provider - between two reductions. A source snapshot is compared after the mutation - lock so a receipt for one declaration cannot authorize a changed Todo. - Materialized and event-projected writes consume the same typed result. + continuation/recovery, completion metadata, registered-agent admission, + successor ownership/exclusion, and existing-successor selection in one + transaction. Python projects registry and Todo-source facts without deciding + policy. A Todo without declared validation, including a replay, uses one + reduction. A real caller-approved validation command remains an explicit + Python provider between two reductions. Todo and policy-source snapshots are + compared after the mutation lock so a receipt for one declaration or agent + registry cannot authorize changed facts. Materialized and event-projected + writes consume the same typed result. - Scheduler heartbeat/state: TypeScript owns receipt freshness, ACK and host-failure validation, identity-aware progression, failure-cache retention/counting, replay and CAS fencing, preview reduction, the locked @@ -519,11 +522,12 @@ Until then Python supplies compact projection facts, clock/effect identity, result validation, and the shared legacy index lock. The Todo cutover removes the Python state-evaluation dataclass, local identity projection, replay helper, and public runtime handlers for those implementation -leaves. The remaining Python Todo facade owns transport, external command -execution, source compare-and-swap, legacy response projection, and the actual -Markdown/event write. It exits when those writers and the CLI move into the -native TS transaction. The remaining fine-grained Turn facade exits after -quota and host-adapter callers move to their own coarse transactions. The +leaves. The remaining Python Todo facade owns fact projection, transport, +external command execution, source compare-and-swap, legacy response +projection, and the actual Markdown/event write. It exits when those writers +and the CLI move into the native TS transaction. The remaining fine-grained +Turn facade exits after quota and host-adapter callers move to their own coarse +transactions. The task-lease semantic facade, atomic Python providers, settlement bridge operation, and lifecycle rule engine are deleted. Python retains compact source projection, one process transport, context-manager plumbing that carries the 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 be60746292..4eda63f82e 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 @@ -363,10 +363,12 @@ window 仍需 differential proof 时才保留 characterization corpus；引入 一次；空 candidate 的 short circuit 仍为零次。 - Todo completion：TypeScript 在一笔 transaction 中拥有 completion identity、 terminal replay fence、validation declaration/effect planning、validation receipt - reduction、continuation/recovery 与 completion metadata。没有声明 validation 的 - Todo（包括 replay）使用一次 reduction；真实 caller-approved validation command - 作为显式 Python provider，位于两次 reduction 之间。取得 mutation lock 后会比较 - source snapshot，确保一份 declaration 的 receipt 不能授权已经变化的 Todo。 + reduction、continuation/recovery、completion metadata、registered-agent admission、 + successor ownership/exclusion 与 existing-successor selection。Python 只投影 registry + 和 Todo source facts，不再决定 policy。没有声明 validation 的 Todo（包括 replay）使用 + 一次 reduction；真实 caller-approved validation command 作为显式 Python provider， + 位于两次 reduction 之间。取得 mutation lock 后会同时比较 Todo 与 policy-source + snapshot，确保一份 declaration 或 agent registry 的 receipt 不能授权已经变化的事实。 Materialized 与 event-projected 写入消费同一 typed result。 - Scheduler heartbeat/state 由 TypeScript 拥有 receipt freshness、ACK 与 host-failure validation、带 identity 的 progression、failure-cache @@ -428,10 +430,10 @@ facade 即可退出。在此之前，Python 只提供 compact projection facts identity、result validation 与共享 legacy index lock。Todo cutover 删除了 Python state-evaluation dataclass、local identity projection、 replay helper，以及这些 implementation leaf 的 public runtime handler。剩余 Python -Todo facade 只拥有 transport、external command execution、source compare-and-swap、 -legacy response projection 与实际 Markdown/event write；当 writer 与 CLI 进入 native -TS transaction 后即可退出。剩余细粒度 Turn facade 则在 quota 与 host-adapter -caller 进入各自 coarse transaction 后退出。Task-lease semantic facade、Python atomic +Todo facade 只拥有 fact projection、transport、external command execution、source +compare-and-swap、legacy response projection 与实际 Markdown/event write；当 writer 与 +CLI 进入 native TS transaction 后即可退出。剩余细粒度 Turn facade 则在 quota 与 +host-adapter caller 进入各自 coarse transaction 后退出。Task-lease semantic facade、Python atomic provider、settlement bridge operation 与 lifecycle rule engine 已经删除。Python 只保留 compact source projection、一次 process transport、携带 opaque fence token/receipt id 的 context-manager plumbing、legacy response projection，以及现有 Python caller 所需的 diff --git a/docs/development/control-plane-course/topic-long-horizon-convergence.md b/docs/development/control-plane-course/topic-long-horizon-convergence.md index 8ea12f0f16..999ec86a3c 100644 --- a/docs/development/control-plane-course/topic-long-horizon-convergence.md +++ b/docs/development/control-plane-course/topic-long-horizon-convergence.md @@ -1546,7 +1546,7 @@ LoopX 已经提供的通用机制包括： | Agent-facing packet | `loopx/control_plane/work_items/interaction_contract.py::build_interaction_contract` | selected work、gate、replan、terminal 是否完整投影 | | Goal frontier replan | `loopx/control_plane/goals/goal_frontier/replan_rules.py::select_goal_frontier_replan_rule` | runnable、gate、succession gap、monitor exhaustion 的优先级 | | Vision checkpoint | `loopx/state_refresh.py::build_vision_checkpoint` | material closeout 后如何防止局部目标替代长期方向 | -| Todo succession | `loopx/control_plane/todos/succession_warning.py::build_open_parent_successor_advisory`、`loopx/control_plane/todos/completion_policy.py::resolve_completion_policy` | successor 为什么只记录 lineage，open parent 为什么仍需显式 complete/defer | +| Todo succession | `loopx/control_plane/todos/succession_warning.py::build_open_parent_successor_advisory`、`loopx/control_plane/todos/completion_policy.ts::resolveTodoCompletionPolicy` | successor 为什么只记录 lineage，open parent 为什么仍需显式 complete/defer | | Turn transaction | `loopx/control_plane/turn_driver/executor.py::run_loopx_turn_once` | phase failure 怎样恢复，何时允许 commit | | Domain State seam | `loopx/domain_state.py::default_domain_state_file_path`、`upsert_domain_state_jsonl` | goal/pack 分区、稳定 key、原子 upsert 和 unchanged observation | | Issue lifecycle | `loopx/capabilities/issue_fix/pr_lifecycle.py::build_issue_fix_pr_lifecycle_monitor_packet` | 外部 PR observation 怎样变成有限 proposal | diff --git a/examples/control_plane/todo-continuation-policy-smoke.py b/examples/control_plane/todo-continuation-policy-smoke.py index 15291bae8e..2a5098a5ff 100644 --- a/examples/control_plane/todo-continuation-policy-smoke.py +++ b/examples/control_plane/todo-continuation-policy-smoke.py @@ -1,9 +1,8 @@ #!/usr/bin/env python3 -"""Exercise typed continuation policy for equal peers.""" +"""Exercise TS-owned Todo continuation policy through the public CLI.""" from __future__ import annotations -import json import sys import tempfile from pathlib import Path @@ -13,16 +12,12 @@ if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) -from loopx.control_plane.todos.completion_policy import ( # noqa: E402 - LinkedSuccessor, - resolve_completion_policy, +from loopx.control_plane.testing.canary_harness import ( # noqa: E402 + run_json_cli, + run_json_cli_result, + write_fixture_registry, ) -from loopx.control_plane.todos.contract import ( # noqa: E402 - TodoContinuationPolicy, - normalize_todo_continuation_policy, - resolve_todo_continuation_policy, -) -from loopx.todos import add_todo_to_lines # noqa: E402 +from loopx.status import parse_active_state_todos # noqa: E402 GOAL_ID = "continuation-policy-fixture" @@ -30,155 +25,167 @@ PEER_BETA = "codex-beta" -def write_registry(path: Path) -> None: - path.write_text( - json.dumps( - { - "goals": [ - { - "id": GOAL_ID, - "coordination": { - "agent_model": "peer_v1", - "registered_agents": [PEER_ALPHA, PEER_BETA], - }, - } - ] - } - ) - + "\n", +def write_fixture(root: Path) -> tuple[Path, Path]: + project = root / "project" + runtime = root / "runtime" + state_file = project / ".codex" / "goals" / GOAL_ID / "ACTIVE_GOAL_STATE.md" + registry_path = project / ".loopx" / "registry.json" + state_file.parent.mkdir(parents=True) + state_file.write_text( + "---\n" + "status: active\n" + "updated_at: 2026-01-01T00:00:00+00:00\n" + "---\n\n" + "# Active Goal State\n\n" + "## Agent Todo\n", encoding="utf-8", ) + write_fixture_registry( + project=project, + runtime_root=runtime, + registry_path=registry_path, + goal_id=GOAL_ID, + domain="continuation-policy-fixture", + adapter_kind="generic_project_goal_v0", + registered_agents=[PEER_ALPHA, PEER_BETA], + quota_allowed_slots=None, + ) + return registry_path, state_file + + +def add_todo(registry_path: Path, text: str) -> dict: + return run_json_cli( + "todo", + "add", + "--goal-id", + GOAL_ID, + "--role", + "agent", + "--text", + text, + "--task-class", + "advancement_task", + "--claimed-by", + PEER_ALPHA, + registry_path=registry_path, + ) + + +def agent_todo(state_file: Path, todo_id: str) -> dict: + items = parse_active_state_todos( + state_file.read_text(encoding="utf-8") + )["agent_todos"]["items"] + return next(item for item in items if item["todo_id"] == todo_id) def main() -> int: with tempfile.TemporaryDirectory(prefix="loopx-continuation-policy-") as tmp: - registry_path = Path(tmp) / "registry.json" - write_registry(registry_path) - - same_peer = resolve_completion_policy( + registry_path, state_file = write_fixture(Path(tmp)) + + same_peer = add_todo(registry_path, "Complete one bounded migration slice.") + same_peer_result = run_json_cli( + "todo", + "complete", + "--goal-id", + GOAL_ID, + "--todo-id", + same_peer["todo_id"], + "--claimed-by", + PEER_ALPHA, + "--agent-id", + PEER_ALPHA, + "--evidence", + "focused validation passed", + "--next-agent-todo", + "Continue a read-only validation lane.", + "--next-continuation-policy", + "same_agent_non_delivery", registry_path=registry_path, - goal_id=GOAL_ID, - claimed_by=PEER_ALPHA, - next_agent_todo="Continue a read-only validation lane.", - next_continuation_policy="same_agent_non_delivery", ) - assert same_peer.effective_next_claimed_by == PEER_ALPHA, same_peer - - independent_validation = resolve_completion_policy( + same_peer_successor = agent_todo( + state_file, same_peer_result["next_todos"][0]["todo_id"] + ) + assert same_peer_successor["claimed_by"] == PEER_ALPHA + assert ( + same_peer_successor["continuation_policy"] + == "same_agent_non_delivery" + ) + assert same_peer_successor["unblocks_todo_id"] == same_peer["todo_id"] + + independent = add_todo(registry_path, "Prepare an independent handoff.") + independent_result = run_json_cli( + "todo", + "complete", + "--goal-id", + GOAL_ID, + "--todo-id", + independent["todo_id"], + "--claimed-by", + PEER_ALPHA, + "--agent-id", + PEER_ALPHA, + "--evidence", + "handoff boundary validated", + "--next-agent-todo", + "Independently review the delivery.", + "--next-claimed-by", + PEER_BETA, + "--next-excluded-agent", + PEER_ALPHA, + "--next-continuation-policy", + "independent_handoff", + registry_path=registry_path, + ) + independent_successor = agent_todo( + state_file, independent_result["next_todos"][0]["todo_id"] + ) + assert independent_successor["claimed_by"] == PEER_BETA + assert independent_successor["excluded_agents"] == [PEER_ALPHA] + + conflict = add_todo(registry_path, "Reject an impossible assignment.") + returncode, rejected = run_json_cli_result( + "todo", + "complete", + "--goal-id", + GOAL_ID, + "--todo-id", + conflict["todo_id"], + "--claimed-by", + PEER_ALPHA, + "--agent-id", + PEER_ALPHA, + "--evidence", + "must remain unchanged", + "--next-agent-todo", + "Review your own delivery.", + "--next-claimed-by", + PEER_ALPHA, + "--next-excluded-agent", + PEER_ALPHA, registry_path=registry_path, - goal_id=GOAL_ID, - claimed_by=PEER_ALPHA, - next_claimed_by=PEER_BETA, - next_agent_todo="Independently review the delivery.", - next_continuation_policy="independent_handoff", - next_excluded_agents=[PEER_ALPHA], ) - assert independent_validation.effective_next_claimed_by == PEER_BETA - assert independent_validation.effective_next_excluded_agents == [PEER_ALPHA] - assert not hasattr(independent_validation, "primary_agent") - - try: - resolve_completion_policy( - registry_path=registry_path, - goal_id=GOAL_ID, - claimed_by=PEER_ALPHA, - next_claimed_by=PEER_ALPHA, - next_agent_todo="Review your own delivery.", - next_continuation_policy="independent_handoff", - next_excluded_agents=[PEER_ALPHA], - ) - except ValueError as exc: - assert "cannot also appear in next_excluded_agents" in str(exc), exc - else: - raise AssertionError("an excluded peer must not receive the successor claim") - - try: - resolve_completion_policy( - registry_path=registry_path, - goal_id=GOAL_ID, - claimed_by=PEER_ALPHA, - next_agent_todo="Continue a same-peer audit.", - next_continuation_policy="same_agent_non_delivery", - next_excluded_agents=[PEER_ALPHA], - ) - except ValueError as exc: - assert "cannot also appear in next_excluded_agents" in str(exc), exc - else: - raise AssertionError("inherited same-peer ownership must respect exclusions") - - try: - resolve_completion_policy( - registry_path=registry_path, - goal_id=GOAL_ID, - claimed_by=PEER_ALPHA, - self_merged=True, - ) - except ValueError as exc: - assert "--self-merged requires --evidence" in str(exc), exc - else: - raise AssertionError("self-merged completion requires evidence") - - merged = resolve_completion_policy( + assert returncode == 1, rejected + assert "cannot also appear in next_excluded_agents" in rejected["error"] + assert agent_todo(state_file, conflict["todo_id"])["status"] == "open" + + missing_evidence = add_todo(registry_path, "Require self-merge evidence.") + returncode, rejected = run_json_cli_result( + "todo", + "complete", + "--goal-id", + GOAL_ID, + "--todo-id", + missing_evidence["todo_id"], + "--claimed-by", + PEER_ALPHA, + "--agent-id", + PEER_ALPHA, + "--self-merged", registry_path=registry_path, - goal_id=GOAL_ID, - claimed_by=PEER_ALPHA, - self_merged=True, - evidence="commit and focused validation passed", - linked_successors=[ - LinkedSuccessor( - todo_id="todo_peer_successor", - role="agent", - status="open", - claimed_by=PEER_BETA, - ) - ], ) - assert merged.self_merged is True, merged - assert merged.linked_successor_id == "todo_peer_successor", merged - - assert normalize_todo_continuation_policy("review_handoff") is None - assert normalize_todo_continuation_policy("primary_review") is None - assert resolve_todo_continuation_policy( - None, - action_kind="primary_review_merge", - ) == TodoContinuationPolicy.INDEPENDENT_HANDOFF - - try: - add_todo_to_lines( - ["# Fixture"], - role="agent", - text="Use the deprecated agent gate field.", - blocks_agent=PEER_ALPHA, - ) - except ValueError as exc: - assert "blocks_agent is only valid for user gates" in str(exc), exc - else: - raise AssertionError("agent todos must use executor exclusions") - - try: - add_todo_to_lines( - ["# Fixture"], - role="agent", - text="Create an impossible executor assignment.", - claimed_by=PEER_ALPHA, - excluded_agents=[PEER_ALPHA], - ) - except ValueError as exc: - assert "cannot also appear in excluded_agents" in str(exc), exc - else: - raise AssertionError("claimed_by must not conflict with excluded_agents") - - try: - add_todo_to_lines( - ["# Fixture"], - role="user", - text="Misapply an executor exclusion to a user todo.", - excluded_agents=[PEER_ALPHA], - ) - except ValueError as exc: - assert "excluded_agents is only valid for agent todos" in str(exc), exc - else: - raise AssertionError("user todos must not accept executor exclusions") + assert returncode == 1, rejected + assert "--self-merged requires --evidence" in rejected["error"], rejected + assert agent_todo(state_file, missing_evidence["todo_id"])["status"] == "open" print("todo-continuation-policy-smoke ok") return 0 diff --git a/loopx/control_plane/coordination/todo_agents.ts b/loopx/control_plane/coordination/todo_agents.ts index e84de3a474..c402000405 100644 --- a/loopx/control_plane/coordination/todo_agents.ts +++ b/loopx/control_plane/coordination/todo_agents.ts @@ -16,6 +16,16 @@ const PYTHON_LEADING_TRAILING_WHITESPACE = new RegExp( ); const PYTHON_WHITESPACE_RUN = new RegExp(`${PYTHON_WHITESPACE_CLASS}+`, "gu"); +/** Strip exactly the characters removed by Python str.strip(). */ +export function stripPythonWhitespace(value: string): string { + return value.replace(PYTHON_LEADING_TRAILING_WHITESPACE, ""); +} + +/** Match Python's bool(str(value).strip()) presence contract. */ +export function hasPythonNonWhitespaceText(value: string): boolean { + return stripPythonWhitespace(value).length > 0; +} + export function normalizeTodoAgent(value: unknown, label: string): string { if (typeof value !== "string") { throw new AuthorityStoreProtocolError(`${label} must be a public-safe agent id`); @@ -24,7 +34,7 @@ export function normalizeTodoAgent(value: unknown, label: string): string { // typed with any Python whitespace (including U+0085 NEL, U+001C..U+001F, // tabs, and NBSP) fold exactly like the Python kernel's compact_todo_text path // (loopx/control_plane/todos/contract.py normalize_todo_claimed_by). - const stripped = value.replace(PYTHON_LEADING_TRAILING_WHITESPACE, ""); + const stripped = stripPythonWhitespace(value); const candidate = stripped.toLowerCase().replace(PYTHON_WHITESPACE_RUN, "-"); if (!/^[a-z][a-z0-9_.:@-]{0,79}$/u.test(candidate)) { throw new AuthorityStoreProtocolError(`${label} must be a public-safe agent id`); diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index e7229a066c..75ed0251e4 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -64,6 +64,7 @@ import { selectTodoCompletionContinuation, } from "./todos/completion_state.ts"; import { reduceTodoCompletionTransaction } from "./todos/completion_transaction.ts"; +import { resolveTodoCompletionPolicy } from "./todos/completion_policy.ts"; import { transitionTodoNextAction } from "./todos/next_action.ts"; import { evaluateTodoResumeConditions, @@ -374,6 +375,7 @@ export function createEffectRuntimeHandlers( ), ], ["todo.completion.reduce", reduceTodoCompletionTransaction], + ["todo.completion_policy.resolve", resolveTodoCompletionPolicy], ["todo.next_action.transition", transitionTodoNextAction], ["todo.resume_condition.normalize", normalizeTodoResumeWhen], ["todo.resume_condition.evaluate", evaluateTodoResumeConditions], diff --git a/loopx/control_plane/todos/completion_policy.py b/loopx/control_plane/todos/completion_policy.py index 802672c462..1b63a3ebb3 100644 --- a/loopx/control_plane/todos/completion_policy.py +++ b/loopx/control_plane/todos/completion_policy.py @@ -6,21 +6,21 @@ from ...agent_registry import ( load_goal_from_registry, - registered_agent_ids_from_registry, - require_registered_agent_id, + registered_agent_ids_for_goal, ) -from ..agents.runtime_model import agent_runtime_model_for_goal +from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result from .active_state_editing import find_todo_block from .contract import ( - TodoContinuationPolicy, normalize_todo_claimed_by, normalize_todo_continuation_policy, normalize_todo_id, - require_todo_excluded_agents, - resolve_todo_continuation_policy, ) +TODO_COMPLETION_POLICY_REQUEST_SCHEMA = "loopx_todo_completion_policy_request_v0" +TODO_COMPLETION_POLICY_RESULT_SCHEMA = "loopx_todo_completion_policy_result_v0" + + @dataclass(frozen=True) class LinkedSuccessor: todo_id: str @@ -89,22 +89,7 @@ def linked_successors_from_state( return successors -def _first_open_agent_successor( - successors: Iterable[LinkedSuccessor], -) -> str | None: - return next( - ( - successor.todo_id - for successor in successors - if successor.role == "agent" - and successor.todo_id - and (not successor.status or successor.status == "open") - ), - None, - ) - - -def resolve_completion_policy( +def build_completion_policy_request( *, registry_path: Path, goal_id: str, @@ -116,73 +101,110 @@ def resolve_completion_policy( next_excluded_agents: Iterable[str] = (), self_merged: bool = False, evidence: str | None = None, - no_followup: bool = False, linked_successors: Iterable[LinkedSuccessor] = (), - completion_todo: Mapping[str, Any] | None = None, -) -> CompletionPolicy: - del no_followup, completion_todo - effective_claimed_by = ( - require_registered_agent_id( - registry_path=registry_path, - goal_id=goal_id, - agent_id=claimed_by, - ) - if claimed_by - else None - ) - registered_agents = registered_agent_ids_from_registry(registry_path, goal_id) - agent_runtime_model_for_goal(load_goal_from_registry(registry_path, goal_id)) - effective_next_claimed_by = ( - require_registered_agent_id( - registry_path=registry_path, - goal_id=goal_id, - agent_id=next_claimed_by, - field="next_claimed_by", - ) - if next_claimed_by - else None - ) - effective_next_excluded_agents = sorted( - require_registered_agent_id( - registry_path=registry_path, - goal_id=goal_id, - agent_id=agent_id, - field="next_excluded_agents", - ) - for agent_id in require_todo_excluded_agents( - next_excluded_agents, - field="next_excluded_agents", - ) +) -> dict[str, Any]: + """Project registry and source facts without deciding successor authority.""" + + del next_action_kind + goal = load_goal_from_registry(registry_path, goal_id) + coordination = goal.get("coordination") if isinstance(goal, Mapping) else None + agent_model = ( + coordination.get("agent_model") if isinstance(coordination, Mapping) else None ) - if self_merged and not str(evidence or "").strip(): - raise ValueError( - "--self-merged requires --evidence with the merge, commit, and " - "validation summary" - ) - next_policy = resolve_todo_continuation_policy( - next_continuation_policy, - action_kind=next_action_kind, + if not agent_model and isinstance(goal, Mapping): + agent_model = goal.get("agent_model") + successor_rows = [ + { + "todo_id": successor.todo_id, + "role": successor.role, + "status": successor.status, + } + for successor in linked_successors + ] + return { + "schema_version": TODO_COMPLETION_POLICY_REQUEST_SCHEMA, + "goal_id": goal_id, + "agent_model": agent_model, + "claimed_by": claimed_by, + "registered_agents": registered_agent_ids_for_goal( + dict(goal) if isinstance(goal, Mapping) else None + ), + "next_claimed_by": next_claimed_by, + "next_agent_todo": next_agent_todo, + "next_continuation_policy": next_continuation_policy, + "next_excluded_agents": list(next_excluded_agents), + "self_merged": bool(self_merged), + "evidence": evidence, + "linked_successors": successor_rows, + } + + +def completion_policy_from_transaction( + transaction: Mapping[str, Any], +) -> CompletionPolicy: + """Adapt the TypeScript-owned completion policy into legacy Python fields.""" + + if transaction.get("decision") == "replay": + # These fields are dead on the event-projected replay path; the TS + # completion fence has already prohibited every write. + return CompletionPolicy(None, [], None, [], False) + policy = transaction.get("completion_policy") + if not isinstance(policy, Mapping) or ( + policy.get("schema_version") != TODO_COMPLETION_POLICY_RESULT_SCHEMA + ): + raise RuntimeError("TypeScript Todo completion policy result shape mismatch") + registered_agents = policy.get("registered_agents") + excluded_agents = policy.get("effective_next_excluded_agents") + claimed_by = policy.get("effective_claimed_by") + next_claimed_by = policy.get("effective_next_claimed_by") + linked_successor_id = policy.get("linked_successor_id") + scalar_shape_is_valid = all( + value is None or isinstance(value, str) + for value in (claimed_by, next_claimed_by, linked_successor_id) ) if ( - next_agent_todo - and not effective_next_claimed_by - and next_policy == TodoContinuationPolicy.SAME_AGENT_NON_DELIVERY + not isinstance(registered_agents, list) + or not all(isinstance(value, str) for value in registered_agents) + or not isinstance(excluded_agents, list) + or not all(isinstance(value, str) for value in excluded_agents) + or not scalar_shape_is_valid + or not isinstance(policy.get("self_merged"), bool) ): - effective_next_claimed_by = effective_claimed_by - if effective_next_claimed_by in effective_next_excluded_agents: - raise ValueError( - f"next_claimed_by={effective_next_claimed_by!r} cannot also appear in " - "next_excluded_agents" - ) - if effective_next_claimed_by and not next_agent_todo: - raise ValueError("--next-claimed-by requires --next-agent-todo") - if effective_next_excluded_agents and not next_agent_todo: - raise ValueError("--next-excluded-agent requires --next-agent-todo") + raise RuntimeError("TypeScript Todo completion policy result shape mismatch") return CompletionPolicy( - effective_claimed_by=effective_claimed_by, - registered_agents=registered_agents, - effective_next_claimed_by=effective_next_claimed_by, - effective_next_excluded_agents=effective_next_excluded_agents, - self_merged=bool(self_merged), - linked_successor_id=_first_open_agent_successor(linked_successors), + effective_claimed_by=claimed_by, + registered_agents=list(registered_agents), + effective_next_claimed_by=next_claimed_by, + effective_next_excluded_agents=list(excluded_agents), + self_merged=bool(policy["self_merged"]), + linked_successor_id=linked_successor_id, ) + + +def bind_completion_policy_to_transaction( + transaction: Mapping[str, Any], + completion_policy_request: Mapping[str, Any], +) -> dict[str, Any]: + """Attach the TS-owned policy after actor and lease admission. + + External validation and completion-state reduction stay single-shot. Only + the pure policy reducer runs under the locked authority boundary, which + preserves the legacy actor -> lease -> policy error priority. + """ + + bound = dict(transaction) + if bound.get("decision") != "commit": + return bound + try: + result = effect_runtime_result( + "todo.completion_policy.resolve", + dict(completion_policy_request), + ) + except EffectRuntimeRejected as exc: + raise ValueError(str(exc)) from None + if not isinstance(result, Mapping): + raise RuntimeError("TypeScript Todo completion policy result must be an object") + bound["completion_policy"] = dict(result) + # Reuse the public adapter as the exact result-shape guard. + completion_policy_from_transaction(bound) + return bound diff --git a/loopx/control_plane/todos/completion_policy.ts b/loopx/control_plane/todos/completion_policy.ts new file mode 100644 index 0000000000..bca5723765 --- /dev/null +++ b/loopx/control_plane/todos/completion_policy.ts @@ -0,0 +1,293 @@ +import type { JsonObject } from "../effect_program.ts"; +import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { AuthorityStoreProtocolError } from "../coordination/authority_store_codec.ts"; +import { + hasPythonNonWhitespaceText, + normalizeRegisteredTodoAgents, + normalizeTodoAgent, + stripPythonWhitespace, +} from "../coordination/todo_agents.ts"; +import { + requireBoolean, + requireJsonObject, + requireNonEmptyString, + requireStringArray, +} from "../runtime_decode.ts"; + +export const TODO_COMPLETION_POLICY_REQUEST_SCHEMA = + "loopx_todo_completion_policy_request_v0"; +export const TODO_COMPLETION_POLICY_RESULT_SCHEMA = + "loopx_todo_completion_policy_result_v0"; + +const CONTINUATION_POLICIES = [ + "independent_handoff", + "same_agent_non_delivery", +] as const; + +interface LinkedSuccessor { + readonly todo_id: string; + readonly role: string | null; + readonly status: string | null; +} + +interface CompletionPolicyRequest { + readonly goal_id: string; + readonly agent_model: string | null; + readonly claimed_by: unknown; + readonly registered_agents: readonly string[]; + readonly next_claimed_by: unknown; + readonly next_agent_todo: string | null; + readonly next_continuation_policy: string | null; + readonly next_excluded_agents: readonly unknown[]; + readonly self_merged: boolean; + readonly evidence: string | null; + readonly linked_successors: readonly LinkedSuccessor[]; +} + +export interface TodoCompletionPolicyResult extends JsonObject { + readonly schema_version: typeof TODO_COMPLETION_POLICY_RESULT_SCHEMA; + readonly effective_claimed_by: string | null; + readonly registered_agents: readonly string[]; + readonly effective_next_claimed_by: string | null; + readonly effective_next_excluded_agents: readonly string[]; + readonly self_merged: boolean; + readonly linked_successor_id: string | null; +} + +function optionalString(value: unknown, label: string): string | null { + if (value === null || value === undefined) return null; + if (typeof value !== "string") { + throw new EffectRuntimeRequestError(`${label} must be a string or null`); + } + return value; +} + +function normalizeAgentId(value: unknown): string | null { + try { + return normalizeTodoAgent(value, "agent_id"); + } catch (error) { + if (error instanceof AuthorityStoreProtocolError) return null; + throw error; + } +} + +function requireRegisteredAgents(value: unknown): string[] { + const agents = requireStringArray(value, "registered_agents"); + try { + return normalizeRegisteredTodoAgents(agents); + } catch (error) { + if (error instanceof AuthorityStoreProtocolError) { + throw new EffectRuntimeRequestError(error.message); + } + throw error; + } +} + +function requireRegisteredAgent( + value: unknown, + field: string, + request: CompletionPolicyRequest, +): string { + const normalized = normalizeAgentId(value); + if (normalized === null) { + throw new EffectRuntimeRequestError( + `${field} must be a public-safe registered agent id`, + ); + } + if (request.registered_agents.length === 0) { + throw new EffectRuntimeRequestError( + `${field}='${normalized}' cannot be used because goal '${request.goal_id}' ` + + "has no coordination.registered_agents list. Register this peer identity first: " + + `loopx configure-goal --goal-id ${request.goal_id} ` + + `--registered-agent ${normalized} --execute`, + ); + } + if (!request.registered_agents.includes(normalized)) { + throw new EffectRuntimeRequestError( + `${field}='${normalized}' is not registered for goal '${request.goal_id}'; ` + + `registered_agents=${request.registered_agents.join(", ")}`, + ); + } + return normalized; +} + +function decodeLinkedSuccessors(value: unknown): LinkedSuccessor[] { + if (!Array.isArray(value)) { + throw new EffectRuntimeRequestError("linked_successors must be an array"); + } + return value.map((raw, index) => { + const successor = requireJsonObject(raw, `linked_successors[${index}]`); + return { + todo_id: requireNonEmptyString( + successor.todo_id, + `linked_successors[${index}].todo_id`, + ), + role: optionalString(successor.role, `linked_successors[${index}].role`), + status: optionalString( + successor.status, + `linked_successors[${index}].status`, + ), + }; + }); +} + +function decodeRequest(value: unknown): CompletionPolicyRequest { + const request = requireJsonObject(value, "todo completion policy request"); + if (request.schema_version !== TODO_COMPLETION_POLICY_REQUEST_SCHEMA) { + throw new EffectRuntimeRequestError( + "Todo completion policy request schema mismatch", + ); + } + const registeredAgents = requireRegisteredAgents(request.registered_agents); + if (!Array.isArray(request.next_excluded_agents)) { + throw new EffectRuntimeRequestError( + "next_excluded_agents must be an array", + ); + } + return { + goal_id: requireNonEmptyString(request.goal_id, "goal_id"), + agent_model: optionalString(request.agent_model, "agent_model"), + claimed_by: request.claimed_by, + registered_agents: registeredAgents, + next_claimed_by: request.next_claimed_by, + next_agent_todo: optionalString(request.next_agent_todo, "next_agent_todo"), + next_continuation_policy: optionalString( + request.next_continuation_policy, + "next_continuation_policy", + ), + next_excluded_agents: request.next_excluded_agents, + self_merged: requireBoolean(request.self_merged, "self_merged"), + evidence: optionalString(request.evidence, "evidence"), + linked_successors: decodeLinkedSuccessors(request.linked_successors), + }; +} + +function requireExcludedAgents( + values: readonly unknown[], + request: CompletionPolicyRequest, +): string[] { + const normalized = new Set(); + for (const value of values) { + const agentId = normalizeAgentId(value); + if (agentId === null) { + throw new EffectRuntimeRequestError( + "next_excluded_agents must contain public-safe agent tokens such as " + + "codex-side-bypass", + ); + } + normalized.add(agentId); + } + return [...normalized] + // Agent ids are closed to public-safe ASCII. Code-point ordering therefore + // matches Python exactly and avoids locale-dependent collation drift. + .sort((left, right) => left < right ? -1 : left > right ? 1 : 0) + .map((agentId) => + requireRegisteredAgent(agentId, "next_excluded_agents", request) + ); +} + +function continuationPolicy( + value: string | null, +): typeof CONTINUATION_POLICIES[number] { + const candidate = stripPythonWhitespace(String(value ?? "")).toLowerCase(); + return CONTINUATION_POLICIES.includes( + candidate as typeof CONTINUATION_POLICIES[number], + ) + ? candidate as typeof CONTINUATION_POLICIES[number] + : "independent_handoff"; +} + +function firstOpenAgentSuccessor( + successors: readonly LinkedSuccessor[], +): string | null { + return successors.find((successor) => + successor.role === "agent" && + Boolean(successor.todo_id) && + (successor.status === null || successor.status === "" || + successor.status === "open") + )?.todo_id ?? null; +} + +/** Resolve successor ownership inside the coarse work-item completion transaction. */ +export function resolveTodoCompletionPolicy( + value: unknown, +): TodoCompletionPolicyResult { + const request = decodeRequest(value); + // Preserve Python bool(str) compatibility: null and the empty string mean + // "not supplied", while a non-empty (including whitespace-only) string is a + // caller-supplied work item. This matters for same-agent ownership and dependent + // --next-* validation. + const hasNextAgentTodo = request.next_agent_todo !== null && + request.next_agent_todo !== ""; + const effectiveClaimedBy = request.claimed_by === null || + request.claimed_by === undefined || request.claimed_by === "" + ? null + : requireRegisteredAgent(request.claimed_by, "claimed_by", request); + if ( + request.agent_model !== null && request.agent_model !== "" && + request.agent_model !== "peer_v1" && + request.agent_model !== "legacy_hierarchy" + ) { + throw new EffectRuntimeRequestError( + "coordination.agent_model must be peer_v1", + ); + } + let effectiveNextClaimedBy = request.next_claimed_by === null || + request.next_claimed_by === undefined || request.next_claimed_by === "" + ? null + : requireRegisteredAgent( + request.next_claimed_by, + "next_claimed_by", + request, + ); + const effectiveNextExcludedAgents = requireExcludedAgents( + request.next_excluded_agents, + request, + ); + if ( + request.self_merged && + !hasPythonNonWhitespaceText(String(request.evidence ?? "")) + ) { + throw new EffectRuntimeRequestError( + "--self-merged requires --evidence with the merge, commit, and " + + "validation summary", + ); + } + if ( + hasNextAgentTodo && effectiveNextClaimedBy === null && + continuationPolicy(request.next_continuation_policy) === + "same_agent_non_delivery" + ) { + effectiveNextClaimedBy = effectiveClaimedBy; + } + if ( + effectiveNextClaimedBy !== null && + effectiveNextExcludedAgents.includes(effectiveNextClaimedBy) + ) { + throw new EffectRuntimeRequestError( + `next_claimed_by='${effectiveNextClaimedBy}' cannot also appear in ` + + "next_excluded_agents", + ); + } + if (effectiveNextClaimedBy !== null && !hasNextAgentTodo) { + throw new EffectRuntimeRequestError( + "--next-claimed-by requires --next-agent-todo", + ); + } + if ( + effectiveNextExcludedAgents.length > 0 && !hasNextAgentTodo + ) { + throw new EffectRuntimeRequestError( + "--next-excluded-agent requires --next-agent-todo", + ); + } + return { + schema_version: TODO_COMPLETION_POLICY_RESULT_SCHEMA, + effective_claimed_by: effectiveClaimedBy, + registered_agents: [...request.registered_agents], + effective_next_claimed_by: effectiveNextClaimedBy, + effective_next_excluded_agents: effectiveNextExcludedAgents, + self_merged: request.self_merged, + linked_successor_id: firstOpenAgentSuccessor(request.linked_successors), + }; +} diff --git a/loopx/control_plane/todos/completion_transaction.py b/loopx/control_plane/todos/completion_transaction.py index a1083fb873..3e4962741c 100644 --- a/loopx/control_plane/todos/completion_transaction.py +++ b/loopx/control_plane/todos/completion_transaction.py @@ -36,6 +36,7 @@ "validation_label", "validation_timeout_seconds", ) +_COMPLETION_POLICY_RESULT_SCHEMA = "loopx_todo_completion_policy_result_v0" def _json_sequence(value: Any) -> Any: @@ -99,6 +100,7 @@ def locked_todo_completion_transaction( dry_run: bool, require_source_match: bool, missing_is_drift: bool, + current_completion_policy_source: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Bind a pre-lock transaction to the Todo observed under the write lock.""" @@ -135,6 +137,25 @@ def locked_todo_completion_transaction( ), ), } + expected_policy_source = validation_gate.get("completion_policy_source") + if current_completion_policy_source is not None or expected_policy_source is not None: + if ( + not isinstance(expected_policy_source, Mapping) + or current_completion_policy_source is None + or dict(expected_policy_source) != dict(current_completion_policy_source) + ): + return { + "transaction": None, + "failure": todo_completion_source_drift_failure( + goal_id=goal_id, + todo_id=todo_id, + dry_run=dry_run, + reason=( + "Todo completion policy source changed after validation " + "planning; retry against the current source" + ), + ), + } return {"transaction": dict(transaction), "failure": None} @@ -280,75 +301,110 @@ def _valid_common(result: Mapping[str, Any]) -> bool: ) -def _valid_result(result: Mapping[str, Any]) -> bool: - if not _valid_common(result): +def _valid_completion_policy(value: Any) -> bool: + if not isinstance(value, Mapping): return False - decision = result.get("decision") - if decision == "execute_validation": - effect = result.get("validation_effect") - return ( - isinstance(effect, Mapping) - and effect.get("kind") == "caller_validation" - and ( - effect.get("validation_command") is None - or isinstance(effect.get("validation_command"), str) - ) - and ( - effect.get("validation_argv") is None - or ( - isinstance(effect.get("validation_argv"), list) - and bool(effect.get("validation_argv")) - and all( - isinstance(item, str) and item - for item in effect.get("validation_argv") or [] - ) - ) - ) - and ( - (effect.get("validation_command") is None) - != (effect.get("validation_argv") is None) - ) - and ( - effect.get("validation_label") is None - or isinstance(effect.get("validation_label"), str) + required_fields = { + "schema_version", + "effective_claimed_by", + "registered_agents", + "effective_next_claimed_by", + "effective_next_excluded_agents", + "self_merged", + "linked_successor_id", + } + return ( + required_fields.issubset(value) + and value.get("schema_version") == _COMPLETION_POLICY_RESULT_SCHEMA + and isinstance(value.get("registered_agents"), list) + and all(isinstance(item, str) for item in value.get("registered_agents") or []) + and isinstance(value.get("effective_next_excluded_agents"), list) + and all( + isinstance(item, str) + for item in value.get("effective_next_excluded_agents") or [] + ) + and all( + item is None or isinstance(item, str) + for item in ( + value.get("effective_claimed_by"), + value.get("effective_next_claimed_by"), + value.get("linked_successor_id"), ) - and ( - effect.get("validation_timeout_seconds") is None - or ( - isinstance(effect.get("validation_timeout_seconds"), int) - and not isinstance( - effect.get("validation_timeout_seconds"), bool - ) - and 1 - <= int(effect["validation_timeout_seconds"]) - <= 29 + ) + and isinstance(value.get("self_merged"), bool) + ) + + +def _valid_execute_validation_result(result: Mapping[str, Any]) -> bool: + effect = result.get("validation_effect") + return ( + isinstance(effect, Mapping) + and effect.get("kind") == "caller_validation" + and ( + effect.get("validation_command") is None + or isinstance(effect.get("validation_command"), str) + ) + and ( + effect.get("validation_argv") is None + or ( + isinstance(effect.get("validation_argv"), list) + and bool(effect.get("validation_argv")) + and all( + isinstance(item, str) and item + for item in effect.get("validation_argv") or [] ) ) ) - if decision == "commit": - state = result.get("completion_state") - updates = result.get("metadata_updates") - receipt = result.get("validation_receipt") - return ( - isinstance(state, Mapping) - and state.get("continuation") in _CONTINUATIONS - and ( - state.get("recovery") is None - or state.get("recovery") in _RECOVERIES - ) - and isinstance(updates, Mapping) - and all( - key in {"completion_continuation", "completion_recovery"} - and isinstance(value, str) - for key, value in updates.items() + and ( + (effect.get("validation_command") is None) + != (effect.get("validation_argv") is None) + ) + and ( + effect.get("validation_label") is None + or isinstance(effect.get("validation_label"), str) + ) + and ( + effect.get("validation_timeout_seconds") is None + or ( + isinstance(effect.get("validation_timeout_seconds"), int) + and not isinstance(effect.get("validation_timeout_seconds"), bool) + and 1 <= int(effect["validation_timeout_seconds"]) <= 29 ) - and updates.get("completion_continuation") - == state.get("continuation") - and updates.get("completion_recovery") == state.get("recovery") - and (receipt is None or _valid_receipt(receipt)) ) - if decision == "replay": - return result["fence"].get("outcome") == "replay" + ) + + +def _valid_commit_result( + result: Mapping[str, Any], + *, + completion_policy_required: bool, +) -> bool: + state = result.get("completion_state") + updates = result.get("metadata_updates") + receipt = result.get("validation_receipt") + policy = result.get("completion_policy") + return ( + isinstance(state, Mapping) + and state.get("continuation") in _CONTINUATIONS + and (state.get("recovery") is None or state.get("recovery") in _RECOVERIES) + and isinstance(updates, Mapping) + and all( + key in {"completion_continuation", "completion_recovery"} + and isinstance(value, str) + for key, value in updates.items() + ) + and updates.get("completion_continuation") == state.get("continuation") + and updates.get("completion_recovery") == state.get("recovery") + and (receipt is None or _valid_receipt(receipt)) + and ( + _valid_completion_policy(policy) + if completion_policy_required or policy is not None + else True + ) + ) + + +def _valid_reject_result(result: Mapping[str, Any]) -> bool: failure = result.get("failure") return ( isinstance(failure, Mapping) @@ -360,6 +416,26 @@ def _valid_result(result: Mapping[str, Any]) -> bool: ) +def _valid_result( + result: Mapping[str, Any], + *, + completion_policy_required: bool, +) -> bool: + if not _valid_common(result): + return False + decision = result.get("decision") + if decision == "execute_validation": + return _valid_execute_validation_result(result) + if decision == "commit": + return _valid_commit_result( + result, + completion_policy_required=completion_policy_required, + ) + if decision == "replay": + return result["fence"].get("outcome") == "replay" + return _valid_reject_result(result) + + def reduce_todo_completion_transaction( *, todo: Mapping[str, Any], @@ -372,34 +448,44 @@ def reduce_todo_completion_transaction( requested_has_successor: bool, dry_run: bool, validation_receipt: Mapping[str, Any] | None = None, + completion_policy_request: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Reduce one coarse completion request through the TS authority.""" try: - result = effect_runtime_result( - "todo.completion.reduce", - { - "schema_version": TODO_COMPLETION_TRANSACTION_REQUEST_SCHEMA, - "goal_id": goal_id, - "todo_id": todo_id, - "projection_source": projection_source, - "todo": todo_completion_source_snapshot(todo), - "requested_no_followup": no_followup, - "requested_completion_turn_key": completion_turn_key, - "requested_completion_identity_source": completion_identity_source, - "requested_has_successor": requested_has_successor, - "dry_run": dry_run, - "validation_receipt": ( - dict(validation_receipt) - if isinstance(validation_receipt, Mapping) - else None - ), - }, - ) + params: dict[str, Any] = { + "schema_version": TODO_COMPLETION_TRANSACTION_REQUEST_SCHEMA, + "goal_id": goal_id, + "todo_id": todo_id, + "projection_source": projection_source, + "todo": todo_completion_source_snapshot(todo), + "requested_no_followup": no_followup, + "requested_completion_turn_key": completion_turn_key, + "requested_completion_identity_source": completion_identity_source, + "requested_has_successor": requested_has_successor, + "dry_run": dry_run, + "validation_receipt": ( + dict(validation_receipt) + if isinstance(validation_receipt, Mapping) + else None + ), + } + if completion_policy_request is not None: + params["completion_policy_request"] = dict(completion_policy_request) + result = effect_runtime_result("todo.completion.reduce", params) except EffectRuntimeRejected as exc: raise ValueError(str(exc)) from None if not isinstance(result, Mapping): - raise RuntimeError("TypeScript Todo completion transaction result must be an object") - if not _valid_result(result): - raise RuntimeError("TypeScript Todo completion transaction result shape mismatch") + raise RuntimeError( + "TypeScript Todo completion transaction result must be an object" + ) + if not _valid_result( + result, + completion_policy_required=( + completion_policy_request is not None and result.get("decision") == "commit" + ), + ): + raise RuntimeError( + "TypeScript Todo completion transaction result shape mismatch" + ) return dict(result) diff --git a/loopx/control_plane/todos/completion_transaction.ts b/loopx/control_plane/todos/completion_transaction.ts index 0b09bc61bb..b96798a246 100644 --- a/loopx/control_plane/todos/completion_transaction.ts +++ b/loopx/control_plane/todos/completion_transaction.ts @@ -29,6 +29,10 @@ import { type TodoCompletionValidationPlanResult, TODO_COMPLETION_VALIDATION_PLAN_REQUEST_SCHEMA, } from "./completion_validation_plan.ts"; +import { + resolveTodoCompletionPolicy, + type TodoCompletionPolicyResult, +} from "./completion_policy.ts"; export const TODO_COMPLETION_TRANSACTION_REQUEST_SCHEMA = "loopx_todo_completion_transaction_v0"; @@ -86,6 +90,7 @@ export interface TodoCompletionCommit extends CompletionTransactionBase { completion_state: CompletionStateProjection; metadata_updates: JsonObject; validation_receipt: CallerValidationReceipt | null; + completion_policy?: TodoCompletionPolicyResult; } export interface TodoCompletionReplay extends CompletionTransactionBase { @@ -118,6 +123,7 @@ interface DecodedTransactionRequest { requested_has_successor: boolean; dry_run: boolean; validation_receipt: CallerValidationReceipt | null; + completion_policy_request: JsonObject | null; } function optionalIdentitySource( @@ -232,6 +238,14 @@ function decodeRequest(value: unknown): DecodedTransactionRequest { ), dry_run: requireBoolean(request.dry_run, "dry_run"), validation_receipt: decodeValidationReceipt(request.validation_receipt), + completion_policy_request: + request.completion_policy_request === null || + request.completion_policy_request === undefined + ? null + : requireJsonObject( + request.completion_policy_request, + "completion_policy_request", + ), }; } @@ -400,6 +414,9 @@ export function reduceTodoCompletionTransaction( metadataResult.updates, "completion metadata updates", ); + const completionPolicy = request.completion_policy_request === null + ? null + : resolveTodoCompletionPolicy(request.completion_policy_request); return { ...base, decision: "commit", @@ -409,5 +426,8 @@ export function reduceTodoCompletionTransaction( }, metadata_updates: updates, validation_receipt: request.validation_receipt, + ...(completionPolicy === null + ? {} + : { completion_policy: completionPolicy }), }; } diff --git a/loopx/control_plane/todos/completion_validation.py b/loopx/control_plane/todos/completion_validation.py index 9375310cd4..66e7e7901e 100644 --- a/loopx/control_plane/todos/completion_validation.py +++ b/loopx/control_plane/todos/completion_validation.py @@ -3,7 +3,7 @@ import subprocess from json import loads as json_loads from pathlib import Path -from typing import Any +from typing import Any, Mapping from ...history import load_registry from ...materials import find_registry_goal, goal_repo @@ -18,6 +18,10 @@ reduce_todo_completion_transaction, todo_completion_source_snapshot, ) +from .completion_policy import ( + build_completion_policy_request, + linked_successors_from_state, +) # Kept safely under the 30s outer CLI/MCP subprocess budget so a timed-out @@ -188,6 +192,8 @@ def run_completion_validation_gate_with_source( completion_turn_key: str | None = None, completion_identity_source: str | None = None, requested_has_successor: bool = False, + completion_policy_facts: Mapping[str, Any] | None = None, + requested_successor_todo_ids: list[str] | None = None, ) -> dict[str, Any]: """Run the caller-approved completion validation gate, OUTSIDE the mutation lock. @@ -201,6 +207,7 @@ def run_completion_validation_gate_with_source( """ projection_source = "materialized" source_authority: dict[str, Any] | None = None + event_context: dict[str, Any] | None = None todo = _materialized_todo_item(state_file=state_file, todo_id=todo_id, role=role) if todo is None: event_context = event_projection_todo_context( @@ -222,6 +229,22 @@ def run_completion_validation_gate_with_source( todo["role"] = event_context["role"] source_authority = event_projection_source_authority(event_context) source_snapshot = todo_completion_source_snapshot(todo) + completion_policy_source = None + if completion_policy_facts is not None: + try: + lines = state_file.read_text(encoding="utf-8").splitlines() + except FileNotFoundError: + lines = [] + completion_policy_source = completion_policy_source_from_state( + registry_path=registry_path, + goal_id=goal_id, + lines=lines, + successor_todo_ids=requested_successor_todo_ids or [], + event_fields=( + event_context.get("fields") if event_context is not None else None + ), + facts=completion_policy_facts, + ) transaction = reduce_todo_completion_transaction( todo=todo, projection_source=projection_source, @@ -233,6 +256,10 @@ def run_completion_validation_gate_with_source( todo_id=todo_id, requested_has_successor=requested_has_successor, validation_receipt=None, + # Preserve the legacy error priority: policy admission is evaluated + # only after actor authority and the task-lease fence are established + # under the write lock. The source is still captured here for CAS. + completion_policy_request=None, ) completion_validation = None if transaction["decision"] == "execute_validation": @@ -275,12 +302,14 @@ def run_completion_validation_gate_with_source( todo_id=todo_id, requested_has_successor=requested_has_successor, validation_receipt=completion_validation, + completion_policy_request=None, ) if transaction["decision"] != "reject": return { "failure": None, "source_authority": source_authority, "source_snapshot": source_snapshot, + "completion_policy_source": completion_policy_source, "transaction": transaction, } failure_payload = transaction["failure"] @@ -299,10 +328,39 @@ def run_completion_validation_gate_with_source( "failure": failure, "source_authority": source_authority, "source_snapshot": source_snapshot, + "completion_policy_source": completion_policy_source, "transaction": transaction, } +def completion_policy_source_from_state( + *, + registry_path: Path, + goal_id: str, + lines: list[str], + successor_todo_ids: list[str], + event_fields: Mapping[str, Any] | None, + facts: Mapping[str, Any], +) -> dict[str, Any]: + """Project lock-comparable facts for the TS completion policy.""" + + return build_completion_policy_request( + registry_path=registry_path, + goal_id=goal_id, + claimed_by=facts.get("claimed_by"), + next_claimed_by=facts.get("next_claimed_by"), + next_agent_todo=facts.get("next_agent_todo"), + next_action_kind=facts.get("next_action_kind"), + next_continuation_policy=facts.get("next_continuation_policy"), + next_excluded_agents=facts.get("next_excluded_agents") or [], + self_merged=bool(facts.get("self_merged")), + evidence=facts.get("evidence"), + linked_successors=linked_successors_from_state( + lines=lines, + successor_todo_ids=successor_todo_ids, + event_fields=event_fields, + ), + ) def prepare_user_todo_update_completion( *, status: str | None, diff --git a/loopx/todos.py b/loopx/todos.py index d452e36813..546874e530 100644 --- a/loopx/todos.py +++ b/loopx/todos.py @@ -62,8 +62,8 @@ from .control_plane.todos.addition import matching_todo_block, require_replan_successor_rebinding, require_replan_successor_scope from .control_plane.todos.completed_archive import archive_completed_todo_lines from .control_plane.todos.completion_policy import ( - linked_successors_from_state, - resolve_completion_policy, + bind_completion_policy_to_transaction, + completion_policy_from_transaction, ) from .control_plane.todos.completion_transaction import ( locked_todo_completion_transaction, @@ -1628,6 +1628,13 @@ def complete_goal_todo( next_user_todo, next_user_task_class, ) + completion_policy_facts = { + "claimed_by": claimed_by, "next_claimed_by": next_claimed_by, + "next_agent_todo": next_agent_todo, "next_action_kind": next_action_kind, + "next_continuation_policy": next_continuation_policy, + "next_excluded_agents": next_excluded_agents or [], + "self_merged": self_merged, "evidence": evidence, + } resolved_project, resolved_state_file = resolve_todo_state_path( registry_path=registry_path, goal_id=goal_id, @@ -1649,6 +1656,8 @@ def complete_goal_todo( requested_has_successor=bool( normalized_successor_todo_ids or next_agent_todo or next_user_todo ), + completion_policy_facts=completion_policy_facts, + requested_successor_todo_ids=normalized_successor_todo_ids, ) validation_failure = validation_gate.get("failure") if validation_failure is not None: @@ -1686,6 +1695,16 @@ def complete_goal_todo( raise ValueError( f"todo_id {normalized_todo_id!r} was not found in active user or agent todos" ) + locked_completion_policy_source = ( + completion_validation_module.completion_policy_source_from_state( + registry_path=registry_path, + goal_id=goal_id, + lines=lines, + successor_todo_ids=normalized_successor_todo_ids, + event_fields=event_context.get("fields") if event_context else None, + facts=completion_policy_facts, + ) + ) locked_completion = locked_todo_completion_transaction( validation_gate=validation_gate, todo=completion_todo, @@ -1694,6 +1713,7 @@ def complete_goal_todo( dry_run=dry_run, require_source_match=bool(completion_match), missing_is_drift=True, + current_completion_policy_source=locked_completion_policy_source, ) if locked_completion["failure"] is not None: return locked_completion["failure"] @@ -1716,7 +1736,6 @@ def complete_goal_todo( decision_target=decision_target, ) completion_handoff = resolve_todo_completion_handoff(state_text=original, mutation_authority=mutation_authority) - completion_fence = completion_transaction["fence"] terminal_replay = materialized_todo_completion_replay( transaction=completion_transaction, todo=completion_todo, @@ -1749,27 +1768,12 @@ def complete_goal_todo( runtime_root=shadow_runtime_root, ) ) - completion_state = completion_transaction.get("completion_state") - linked_successors = linked_successors_from_state( - lines=lines, - successor_todo_ids=normalized_successor_todo_ids, - event_fields=event_context.get("fields") if event_context else None, - ) - completion_policy = resolve_completion_policy( - registry_path=registry_path, - goal_id=goal_id, - claimed_by=claimed_by, - next_claimed_by=next_claimed_by, - next_agent_todo=next_agent_todo, - next_action_kind=next_action_kind, - next_continuation_policy=next_continuation_policy, - next_excluded_agents=next_excluded_agents or [], - self_merged=self_merged, - evidence=evidence, - no_followup=no_followup, - linked_successors=linked_successors, - completion_todo=completion_todo, + completion_transaction = bind_completion_policy_to_transaction( + completion_transaction, locked_completion_policy_source ) + completion_fence = completion_transaction["fence"] + completion_state = completion_transaction.get("completion_state") + completion_policy = completion_policy_from_transaction(completion_transaction) effective_claimed_by = completion_policy.effective_claimed_by registered_agents = completion_policy.registered_agents effective_next_claimed_by = completion_policy.effective_next_claimed_by diff --git a/tests/control_plane/test_todo_completion_transaction_runtime.py b/tests/control_plane/test_todo_completion_transaction_runtime.py index 7d5bf7e538..760e0f0d93 100644 --- a/tests/control_plane/test_todo_completion_transaction_runtime.py +++ b/tests/control_plane/test_todo_completion_transaction_runtime.py @@ -30,6 +30,19 @@ def _commit_result(**overrides: object) -> dict[str, object]: } +def _completion_policy_result(**overrides: object) -> dict[str, object]: + return { + "schema_version": "loopx_todo_completion_policy_result_v0", + "effective_claimed_by": "agent-a", + "registered_agents": ["agent-a"], + "effective_next_claimed_by": "agent-a", + "effective_next_excluded_agents": [], + "self_merged": False, + "linked_successor_id": None, + **overrides, + } + + def test_python_adapter_sends_one_coarse_transaction(monkeypatch) -> None: captured: dict[str, object] = {} @@ -97,6 +110,156 @@ def test_source_snapshot_detects_completion_authority_drift() -> None: ) +def test_python_adapter_requires_requested_completion_policy_projection( + monkeypatch, +) -> None: + captured: dict[str, object] = {} + policy_request = { + "schema_version": "loopx_todo_completion_policy_request_v0", + "goal_id": "goal-example", + } + + def call(method: str, params: dict[str, object]) -> dict[str, object]: + captured["method"] = method + captured["params"] = params + return _commit_result(completion_policy=_completion_policy_result()) + + monkeypatch.setattr(completion_transaction, "effect_runtime_result", call) + result = completion_transaction.reduce_todo_completion_transaction( + todo={"status": "open"}, + projection_source="materialized", + goal_id="goal-example", + todo_id="todo_example001", + completion_turn_key=None, + completion_identity_source=None, + no_followup=False, + requested_has_successor=True, + dry_run=False, + completion_policy_request=policy_request, + ) + + assert result["completion_policy"] == _completion_policy_result() + assert captured["params"]["completion_policy_request"] == policy_request + + monkeypatch.setattr( + completion_transaction, + "effect_runtime_result", + lambda _method, _params: _commit_result(), + ) + with pytest.raises(RuntimeError, match="result shape mismatch"): + completion_transaction.reduce_todo_completion_transaction( + todo={"status": "open"}, + projection_source="materialized", + goal_id="goal-example", + todo_id="todo_example001", + completion_turn_key=None, + completion_identity_source=None, + no_followup=False, + requested_has_successor=True, + dry_run=False, + completion_policy_request=policy_request, + ) + + +def test_python_adapter_and_typescript_runtime_share_agent_identity_semantics() -> None: + policy_request = { + "schema_version": "loopx_todo_completion_policy_request_v0", + "goal_id": "goal-example", + "agent_model": "peer_v1", + "claimed_by": "agent\u0085a", + "registered_agents": ["agent\u0085a", "agent-b"], + "next_claimed_by": "agent\u0085b", + "next_agent_todo": "Continue the bounded migration.", + "next_continuation_policy": None, + "next_excluded_agents": ["agent\u0085a"], + "self_merged": False, + "evidence": None, + "linked_successors": [], + } + + result = completion_transaction.reduce_todo_completion_transaction( + todo={"status": "open"}, + projection_source="materialized", + goal_id="goal-example", + todo_id="todo_example001", + completion_turn_key=None, + completion_identity_source=None, + no_followup=False, + requested_has_successor=True, + dry_run=False, + completion_policy_request=policy_request, + ) + + assert result["completion_policy"] == _completion_policy_result( + registered_agents=["agent-a", "agent-b"], + effective_next_claimed_by="agent-b", + effective_next_excluded_agents=["agent-a"], + ) + + continuation_result = completion_transaction.reduce_todo_completion_transaction( + todo={"status": "open"}, + projection_source="materialized", + goal_id="goal-example", + todo_id="todo_example001", + completion_turn_key=None, + completion_identity_source=None, + no_followup=False, + requested_has_successor=True, + dry_run=False, + completion_policy_request={ + **policy_request, + "next_claimed_by": None, + "next_continuation_policy": "\u0085same_agent_non_delivery\u0085", + "next_excluded_agents": [], + "self_merged": True, + "evidence": "\ufeff", + }, + ) + assert continuation_result["completion_policy"] == _completion_policy_result( + registered_agents=["agent-a", "agent-b"], + effective_next_claimed_by="agent-a", + self_merged=True, + ) + + with pytest.raises(ValueError, match="--self-merged requires --evidence"): + completion_transaction.reduce_todo_completion_transaction( + todo={"status": "open"}, + projection_source="materialized", + goal_id="goal-example", + todo_id="todo_example001", + completion_turn_key=None, + completion_identity_source=None, + no_followup=False, + requested_has_successor=False, + dry_run=False, + completion_policy_request={ + **policy_request, + "next_claimed_by": None, + "next_agent_todo": None, + "next_excluded_agents": [], + "self_merged": True, + "evidence": "\u0085", + }, + ) + + with pytest.raises(ValueError, match="public-safe registered agent id"): + completion_transaction.reduce_todo_completion_transaction( + todo={"status": "open"}, + projection_source="materialized", + goal_id="goal-example", + todo_id="todo_example001", + completion_turn_key=None, + completion_identity_source=None, + no_followup=False, + requested_has_successor=True, + dry_run=False, + completion_policy_request={ + **policy_request, + "claimed_by": "\ufeffagent-a", + }, + ) + + @pytest.mark.parametrize( "malformed", [ diff --git a/tests/control_plane/test_todo_completion_validation.py b/tests/control_plane/test_todo_completion_validation.py index 520e7c62ae..3b000222d4 100644 --- a/tests/control_plane/test_todo_completion_validation.py +++ b/tests/control_plane/test_todo_completion_validation.py @@ -315,6 +315,49 @@ def validate_then_drift(**kwargs): # type: ignore[no-untyped-def] assert _agent_todo(state, str(todo["todo_id"]))["status"] == "open" +def test_validation_receipt_cannot_commit_changed_completion_policy_source( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + registry, state = _write_fixture(tmp_path) + todo = _add_todo( + registry, + validation_command=_PASS_COMMAND, + validation_label="caller-declared smoke", + ) + original_validation = ( + completion_validation_module._run_declared_completion_validation + ) + + def validate_then_change_registry(**kwargs): # type: ignore[no-untyped-def] + receipt = original_validation(**kwargs) + payload = json.loads(registry.read_text(encoding="utf-8")) + payload["goals"][0]["coordination"]["registered_agents"].append( + "codex-new-peer" + ) + registry.write_text(json.dumps(payload), encoding="utf-8") + return receipt + + monkeypatch.setattr( + completion_validation_module, + "_run_declared_completion_validation", + validate_then_change_registry, + ) + + result = complete_goal_todo( + registry_path=registry, + goal_id=GOAL_ID, + todo_id=str(todo["todo_id"]), + agent_id=AGENT, + evidence="stale registry projection", + ) + + assert result["ok"] is False + assert result["completion_source_drift"] is True + assert result["changed"] is False + assert _agent_todo(state, str(todo["todo_id"]))["status"] == "open" + + def test_validation_timeout_blocks_completion( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/control_plane/test_todo_mutation_authority.py b/tests/control_plane/test_todo_mutation_authority.py index c48faeb950..90079ed3ba 100644 --- a/tests/control_plane/test_todo_mutation_authority.py +++ b/tests/control_plane/test_todo_mutation_authority.py @@ -117,6 +117,7 @@ def _agent_todo(state: Path, todo_id: str) -> dict: def _add_agent_todo( registry: Path, *, + text: str = "Deliver one bounded control-plane change.", claimed_by: str | None = AUTHOR_AGENT, excluded_agents: list[str] | None = None, ) -> dict: @@ -124,7 +125,7 @@ def _add_agent_todo( registry_path=registry, goal_id=GOAL_ID, role="agent", - text="Deliver one bounded control-plane change.", + text=text, task_class="advancement_task", claimed_by=claimed_by, excluded_agents=excluded_agents, @@ -339,6 +340,113 @@ def test_non_owner_cannot_mutate_claimed_todo( assert state.read_text(encoding="utf-8") == before +def test_completion_policy_unicode_parity_through_public_facade( + tmp_path: Path, +) -> None: + registry, state = _write_fixture(tmp_path) + todo = _add_agent_todo(registry, text="Continue with Python whitespace parity.") + + result = complete_goal_todo( + registry_path=registry, + goal_id=GOAL_ID, + todo_id=todo["todo_id"], + agent_id=AUTHOR_AGENT, + claimed_by=AUTHOR_AGENT, + evidence="validated parity", + next_agent_todo="Run the next bounded parity check.", + next_continuation_policy="\u0085same_agent_non_delivery\u0085", + ) + + successor = _agent_todo(state, result["next_todos"][0]["todo_id"]) + assert successor["claimed_by"] == AUTHOR_AGENT + + blank_evidence = _add_agent_todo( + registry, + text="Reject Python-blank self-merge evidence.", + ) + before = state.read_text(encoding="utf-8") + with pytest.raises(ValueError, match="--self-merged requires --evidence"): + complete_goal_todo( + registry_path=registry, + goal_id=GOAL_ID, + todo_id=blank_evidence["todo_id"], + agent_id=AUTHOR_AGENT, + self_merged=True, + evidence="\u0085", + no_followup=True, + ) + assert state.read_text(encoding="utf-8") == before + + bom_evidence = _add_agent_todo( + registry, + text="Accept evidence retained by Python strip.", + ) + accepted = complete_goal_todo( + registry_path=registry, + goal_id=GOAL_ID, + todo_id=bom_evidence["todo_id"], + agent_id=AUTHOR_AGENT, + self_merged=True, + evidence="\ufeff", + no_followup=True, + ) + assert accepted["self_merged"] is True + + +def test_completion_policy_errors_do_not_preempt_actor_or_lease_fences( + tmp_path: Path, +) -> None: + registry, state = _write_fixture(tmp_path) + claimed_by_other = _add_agent_todo( + registry, + text="Keep actor authority ahead of policy diagnostics.", + claimed_by=REVIEW_AGENT, + ) + before = state.read_text(encoding="utf-8") + + with pytest.raises(ValueError, match="claimed_by='codex-review'"): + complete_goal_todo( + registry_path=registry, + goal_id=GOAL_ID, + todo_id=claimed_by_other["todo_id"], + agent_id=AUTHOR_AGENT, + self_merged=True, + evidence="\u0085", + no_followup=True, + ) + assert state.read_text(encoding="utf-8") == before + + leased = _add_agent_todo( + registry, + text="Keep the task lease ahead of policy diagnostics.", + ) + lease_key = "policy-priority-instance" + acquire_task_lease( + registry_path=registry, + runtime_root=tmp_path / "runtime", + goal_id=GOAL_ID, + todo_id=leased["todo_id"], + owner=AUTHOR_AGENT, + idempotency_key=lease_key, + ttl_seconds=600, + ) + before = state.read_text(encoding="utf-8") + with pytest.raises(TaskLeaseError) as stale_lease: + complete_goal_todo( + registry_path=registry, + goal_id=GOAL_ID, + todo_id=leased["todo_id"], + agent_id=AUTHOR_AGENT, + task_lease_idempotency_key=lease_key, + task_lease_expected_version=0, + self_merged=True, + evidence="\u0085", + no_followup=True, + ) + assert stale_lease.value.code == "version_mismatch" + assert state.read_text(encoding="utf-8") == before + + def test_owner_actor_update_returns_typed_receipt(tmp_path: Path) -> None: registry, state = _write_fixture(tmp_path) todo = _add_agent_todo(registry) diff --git a/tests/control_plane_ts/todo_completion_policy.test.ts b/tests/control_plane_ts/todo_completion_policy.test.ts new file mode 100644 index 0000000000..e89c347dd4 --- /dev/null +++ b/tests/control_plane_ts/todo_completion_policy.test.ts @@ -0,0 +1,184 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + resolveTodoCompletionPolicy, + TODO_COMPLETION_POLICY_REQUEST_SCHEMA, +} from "../../loopx/control_plane/todos/completion_policy.ts"; + +function request(overrides: Record = {}) { + return { + schema_version: TODO_COMPLETION_POLICY_REQUEST_SCHEMA, + goal_id: "goal-example", + agent_model: "peer_v1", + claimed_by: "agent-a", + registered_agents: ["agent-a", "agent-b"], + next_claimed_by: null, + next_agent_todo: null, + next_continuation_policy: null, + next_excluded_agents: [], + self_merged: false, + evidence: null, + linked_successors: [], + ...overrides, + }; +} + +test("same-agent continuation and linked-successor selection are TS-owned", () => { + const result = resolveTodoCompletionPolicy( + request({ + next_agent_todo: "Continue the bounded migration.", + next_continuation_policy: "same_agent_non_delivery", + next_excluded_agents: ["AGENT-B", "agent-b"], + linked_successors: [ + { todo_id: "todo_user", role: "user", status: "open" }, + { todo_id: "todo_done", role: "agent", status: "done" }, + { todo_id: "todo_open", role: "agent", status: "open" }, + { todo_id: "todo_later", role: "agent", status: null }, + ], + }), + ); + + assert.deepEqual(result, { + schema_version: "loopx_todo_completion_policy_result_v0", + effective_claimed_by: "agent-a", + registered_agents: ["agent-a", "agent-b"], + effective_next_claimed_by: "agent-a", + effective_next_excluded_agents: ["agent-b"], + self_merged: false, + linked_successor_id: "todo_open", + }); +}); +test("registration, exclusion, and self-merge invariants fail closed", () => { + assert.throws( + () => resolveTodoCompletionPolicy(request({ claimed_by: "agent-c" })), + /claimed_by='agent-c' is not registered for goal 'goal-example'; registered_agents=agent-a, agent-b/, + ); + assert.throws( + () => + resolveTodoCompletionPolicy( + request({ + next_agent_todo: "Continue.", + next_claimed_by: "agent-b", + next_excluded_agents: ["agent-b"], + }), + ), + /next_claimed_by='agent-b' cannot also appear in next_excluded_agents/, + ); + assert.throws( + () => resolveTodoCompletionPolicy(request({ self_merged: true })), + /--self-merged requires --evidence/, + ); + assert.throws( + () => + resolveTodoCompletionPolicy( + request({ next_claimed_by: "agent-b" }), + ), + /--next-claimed-by requires --next-agent-todo/, + ); + assert.throws( + () => + resolveTodoCompletionPolicy( + request({ agent_model: "hierarchy_v2" }), + ), + /coordination.agent_model must be peer_v1/, + ); +}); + +test("empty next_agent_todo preserves legacy absence semantics", () => { + const empty = resolveTodoCompletionPolicy( + request({ + next_agent_todo: "", + next_continuation_policy: "same_agent_non_delivery", + }), + ); + assert.equal(empty.effective_next_claimed_by, null); + + assert.throws( + () => + resolveTodoCompletionPolicy( + request({ next_agent_todo: "", next_claimed_by: "agent-b" }), + ), + /--next-claimed-by requires --next-agent-todo/, + ); + assert.throws( + () => + resolveTodoCompletionPolicy( + request({ next_agent_todo: "", next_excluded_agents: ["agent-b"] }), + ), + /--next-excluded-agent requires --next-agent-todo/, + ); + + const whitespaceOnly = resolveTodoCompletionPolicy( + request({ + next_agent_todo: " ", + next_continuation_policy: "same_agent_non_delivery", + }), + ); + assert.equal(whitespaceOnly.effective_next_claimed_by, "agent-a"); +}); + +test("agent identity normalization matches the Python Unicode contract", () => { + const pythonWhitespaceCodePoints = [ + 0x0009, 0x000a, 0x000b, 0x000c, 0x000d, 0x001c, 0x001d, 0x001e, + 0x001f, 0x0020, 0x0085, 0x00a0, 0x1680, 0x2000, 0x2001, 0x2002, + 0x2003, 0x2004, 0x2005, 0x2006, 0x2007, 0x2008, 0x2009, 0x200a, + 0x2028, 0x2029, 0x202f, 0x205f, 0x3000, + ]; + for (const codePoint of pythonWhitespaceCodePoints) { + const separator = String.fromCodePoint(codePoint); + const result = resolveTodoCompletionPolicy( + request({ + claimed_by: `agent${separator}a`, + registered_agents: [`agent${separator}a`, "agent-b"], + next_agent_todo: "Continue.", + next_claimed_by: `agent${separator}b`, + next_excluded_agents: [`agent${separator}a`], + }), + ); + assert.equal(result.effective_claimed_by, "agent-a"); + assert.deepEqual(result.registered_agents, ["agent-a", "agent-b"]); + assert.equal(result.effective_next_claimed_by, "agent-b"); + assert.deepEqual(result.effective_next_excluded_agents, ["agent-a"]); + + const continuation = resolveTodoCompletionPolicy( + request({ + next_agent_todo: "Continue.", + next_continuation_policy: + `${separator}same_agent_non_delivery${separator}`, + }), + ); + assert.equal(continuation.effective_next_claimed_by, "agent-a"); + assert.throws( + () => + resolveTodoCompletionPolicy( + request({ self_merged: true, evidence: separator }), + ), + /--self-merged requires --evidence/, + ); + } + + for (const field of [ + { claimed_by: "\ufeffagent-a" }, + { registered_agents: ["\ufeffagent-a", "agent-b"] }, + { next_agent_todo: "Continue.", next_claimed_by: "\ufeffagent-b" }, + { next_agent_todo: "Continue.", next_excluded_agents: ["\ufeffagent-a"] }, + ]) { + assert.throws( + () => resolveTodoCompletionPolicy(request(field)), + /must be a public-safe (?:registered )?agent id|must contain public-safe agent tokens/, + ); + } + + const bomContinuation = resolveTodoCompletionPolicy( + request({ + next_agent_todo: "Continue.", + next_continuation_policy: "\ufeffsame_agent_non_delivery\ufeff", + }), + ); + assert.equal(bomContinuation.effective_next_claimed_by, null); + const bomEvidence = resolveTodoCompletionPolicy( + request({ self_merged: true, evidence: "\ufeff" }), + ); + assert.equal(bomEvidence.self_merged, true); +}); diff --git a/tests/control_plane_ts/todo_completion_transaction.test.ts b/tests/control_plane_ts/todo_completion_transaction.test.ts index add7728065..b63e98d90d 100644 --- a/tests/control_plane_ts/todo_completion_transaction.test.ts +++ b/tests/control_plane_ts/todo_completion_transaction.test.ts @@ -35,6 +35,21 @@ function request(overrides: Record = {}) { }; } +const completionPolicyRequest = { + schema_version: "loopx_todo_completion_policy_request_v0", + goal_id: "goal-example", + agent_model: "peer_v1", + claimed_by: "agent-a", + registered_agents: ["agent-a"], + next_claimed_by: null, + next_agent_todo: "Continue the bounded migration.", + next_continuation_policy: "same_agent_non_delivery", + next_excluded_agents: [], + self_merged: false, + evidence: "focused validation passed", + linked_successors: [], +}; + test("open completion commits in one reduction with a stable local identity", () => { const result = reduceTodoCompletionTransaction(request()); @@ -132,6 +147,37 @@ test("declared validation is one external effect between two reductions", () => assert.equal(rejected.failure.validation_receipt.passed, false); }); +test("completion policy joins the coarse transaction only at commit", () => { + const pending = reduceTodoCompletionTransaction( + request({ + todo: { + ...baseTodo, + validation_command: "true", + }, + completion_policy_request: { + ...completionPolicyRequest, + claimed_by: "not registered", + }, + }), + ); + assert.equal(pending.decision, "execute_validation"); + assert.equal("completion_policy" in pending, false); + + const committed = reduceTodoCompletionTransaction( + request({ completion_policy_request: completionPolicyRequest }), + ); + assert.equal(committed.decision, "commit"); + assert.deepEqual(committed.completion_policy, { + schema_version: "loopx_todo_completion_policy_result_v0", + effective_claimed_by: "agent-a", + registered_agents: ["agent-a"], + effective_next_claimed_by: "agent-a", + effective_next_excluded_agents: [], + self_merged: false, + linked_successor_id: null, + }); +}); + test("terminal replay bypasses a stale validation declaration", () => { const result = reduceTodoCompletionTransaction( request({ diff --git a/tests/test_todo_write_serialization.py b/tests/test_todo_write_serialization.py index 9c891a6b1c..602ca9e73f 100644 --- a/tests/test_todo_write_serialization.py +++ b/tests/test_todo_write_serialization.py @@ -102,7 +102,7 @@ def deterministic_lock( monkeypatch.setattr( legacy_writer_fence, - "exclusive_file_lock", + "exclusive_cross_runtime_file_lock", deterministic_lock, )