diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index e896e74ba8..695ea76578 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -113,6 +113,16 @@ The existing delivery Todos should carry these concrete exits: This updates the existing R5/T4 frontier and release proposal; it creates no parallel migration owner, runtime gate or release activation. +Team-plan recovery prerequisite: an unpromoted plan that published its original +receipt can now recover through the Chat confirmation owner while shadow +bootstrap is interrupted. Registry/Todo/source locks still protect the read, +and the existing typed identity validator rejects changed intent, actor or +reviewed basis. Recovery verifies source durability without loading the capture +producer or admitting new writes; fresh plans still obey all source, management +and promotion guards. This repairs the existing 1.x recovery path and does not +activate the 2.0 cutoff or settle the remaining lease/outbox/Host obligations. + + The retained import-compatible field writer now composes resume normalization and contradictory-clear rejection in its existing typed field plan. Keep blank Python input omission and locked IO; retire the duplicate normalization crossing, diff --git a/loopx/control_plane/coordination/legacy_writer_fence.py b/loopx/control_plane/coordination/legacy_writer_fence.py index bab4cf543e..f540942769 100644 --- a/loopx/control_plane/coordination/legacy_writer_fence.py +++ b/loopx/control_plane/coordination/legacy_writer_fence.py @@ -339,6 +339,22 @@ def require_legacy_coordination_write_allowed( ) +def require_legacy_todo_write_allowed( + *, registry_path: Path, runtime_root: Path, goal_id: str, state_file: Path, +) -> None: + """Admit a fresh source mutation while its Todo/state locks are held. + + Historical receipt reads use the same locks but do not admit a new write. + Keep source binding, management and promotion checks together for writers. + """ + require_registry_source_write_allowed( + registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id, + state_file=state_file, + ) + require_shadow_primary_write_allowed(runtime_root, goal_id) + require_legacy_coordination_write_allowed(runtime_root=runtime_root, goal_id=goal_id) + + @contextmanager def legacy_todo_write_transaction( registry_path: Path, @@ -376,13 +392,8 @@ def legacy_todo_write_transaction( operation=operation, ): if not dry_run: - require_registry_source_write_allowed( - registry_path=registry_path, runtime_root=resolved_runtime_root, goal_id=goal_id, - state_file=state_file, - ) - require_shadow_primary_write_allowed(resolved_runtime_root, goal_id) - require_legacy_coordination_write_allowed( - runtime_root=resolved_runtime_root, - goal_id=goal_id, + require_legacy_todo_write_allowed( + registry_path=registry_path, runtime_root=resolved_runtime_root, + goal_id=goal_id, state_file=state_file, ) yield diff --git a/loopx/control_plane/work_items/team_plan_adapter.py b/loopx/control_plane/work_items/team_plan_adapter.py index 714cd603de..2316450d33 100644 --- a/loopx/control_plane/work_items/team_plan_adapter.py +++ b/loopx/control_plane/work_items/team_plan_adapter.py @@ -231,39 +231,52 @@ def previous_receipt(original: str) -> dict[str, Any] | None: result["projection_delivery"] = projected.get("projection_delivery") return result - # Only unpromoted Goals need the source writer and capture producer. Keep - # canonical replay and projection independent of this compatibility path. - from ..coordination.legacy_writer_fence import legacy_todo_write_transaction - from ..coordination.runtime_shadow_writer_adapter import ( - begin_todo_runtime_shadow_capture, settle_todo_runtime_shadow_capture, - ) + # Lock the original source before consulting its receipt. A management hold + # stops fresh writes, not readback of an already-published operation. Keep + # receipt identity validation in the shared typed planner. + from ..coordination.legacy_writer_fence import require_legacy_todo_write_allowed from ..runtime.document_io import verify_state_text_durable - from ..todos.active_state_editing import ( - insert_into_existing_section, insert_new_section, section_bounds, - replace_updated_at, - ) - from ..todos.contract import format_todo_metadata_line, parse_todo_metadata_line with exclusive_cross_runtime_file_lock(registry_path, operation="team_plan_commit"), \ - legacy_todo_write_transaction(registry_path, goal_id, state, agent_id, - "todo_add", False, runtime_root=runtime): + exclusive_cross_runtime_file_lock( + legacy_coordination_todo_lock_path(runtime_root=runtime, goal_id=goal_id), + agent_id=agent_id, operation="legacy_coordination_todo_write"), \ + exclusive_cross_runtime_file_lock(state, agent_id=agent_id, operation="todo_add"): read_source_request() original = state.read_text(encoding="utf-8") previous = previous_receipt(original) + if previous is not None: + # Promotion may have completed while these locks were acquired. + # An unavailable selected provider must still fail closed; this is + # original receipt recovery, never a Markdown authority fallback. + read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal_id) + replay = effect_runtime_result("work_items.team_plan.plan", { + **request, "previous_receipt": previous}) + verify_state_text_durable(state, original) + return dict(replay["result"]) + require_legacy_todo_write_allowed( + registry_path=registry_path, runtime_root=runtime, goal_id=goal_id, state_file=state) + # Recovery above needs neither the source producer nor Markdown editing. + from ..coordination.runtime_shadow_writer_adapter import ( + begin_todo_runtime_shadow_capture, settle_todo_runtime_shadow_capture, + ) + from ..todos.active_state_editing import ( + insert_into_existing_section, insert_new_section, section_bounds, + replace_updated_at, + ) + from ..todos.contract import format_todo_metadata_line, parse_todo_metadata_line + existing = [metadata for line in original.splitlines() if (metadata := parse_todo_metadata_line(line)) is not None] current_fingerprint = read_fingerprint() if read_fingerprint else None try: planned = effect_runtime_result("work_items.team_plan.plan", { - **request, "previous_receipt": previous, "todos": existing, + **request, "todos": existing, "current_state_fingerprint": current_fingerprint, }) except EffectRuntimeRejected as error: raise TeamPlanCommitError(error.diagnostic_code, str(error), current_fingerprint) from None result = dict(planned["result"]) - if planned["replayed"]: - verify_state_text_durable(state, original) - return result capture = begin_todo_runtime_shadow_capture(registry_path=registry_path, runtime_root=runtime, goal_id=goal_id, state_path=state, write_class="todo_add", original_text=original) diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json index 96b0a61889..9b2fd2f0c0 100644 --- a/loopx/semantics/project_registry_io_manifest_v1.json +++ b/loopx/semantics/project_registry_io_manifest_v1.json @@ -1119,7 +1119,7 @@ }, { "site": "loopx/control_plane/coordination/legacy_writer_fence.py::.legacy_todo_write_transaction::codec_read:load_registry#1", - "line": 362, + "line": 378, "column": 9, "kind": "codec_read", "api": "load_registry", diff --git a/tests/control_plane/test_team_plan_authority.py b/tests/control_plane/test_team_plan_authority.py index 7b55e5a1da..1d9b369166 100644 --- a/tests/control_plane/test_team_plan_authority.py +++ b/tests/control_plane/test_team_plan_authority.py @@ -478,3 +478,120 @@ def test_capture_free_chat_preserves_pre_promotion_receipt(tmp_path, monkeypatch assert replay["lane_todo_ids"] == applied["receipt"]["resource_ids"]["lane_todo_ids"] assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a") == current assert state.read_bytes() == display + + # Simulate promotion finishing after the first mode observation but before + # acquiring source locks; its selected backend subsequently goes offline. + from loopx.control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable + from loopx.control_plane.work_items import team_plan_adapter + read_authority = team_plan_adapter.read_canonical_todos_if_promoted + observed = [] + + def promotion_race(**kwargs): + observed.append(True) + return None if len(observed) == 1 else read_authority(**kwargs) + + backend = runtime / "authority" / current["source_authority"].replace("_", "-") + offline = backend.with_name(backend.name + "-offline") + backend.rename(offline) + try: + with monkeypatch.context() as race: + race.setattr(team_plan_adapter, "read_canonical_todos_if_promoted", promotion_race) + saved = service.store.load(preview["proposal_id"]) + with pytest.raises(LocalCoordinationAuthorityUnavailable): + team_plan_adapter.settle_team_plan( + registry_path=registry, goal_id="goal-a", agent_id=None, + proposal={**saved["normalized_parameters"]["plan"], "proposal_id": preview["proposal_id"]}, + expected_state_fingerprint=saved["expected_state_fingerprint"], basis_agent_id="alpha") + assert len(observed) == 2 + assert state.read_bytes() == display and not backend.exists() + finally: + offline.rename(backend) + assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a") == current + + +@pytest.mark.skipif(os.name != "posix", reason="real bootstrap crash barrier uses select on a pipe") +def test_legacy_chat_replay_survives_interrupted_shadow_bootstrap(tmp_path, monkeypatch): + """A stopped writer must not strand an already-published plan receipt.""" + import select + from loopx.control_plane.work_items import team_plan_adapter + from loopx.control_plane.coordination.runtime_shadow import build_runtime_shadow_source_snapshot + from loopx.control_plane.coordination.shadow_management import read_shadow_management_state + from loopx.control_plane.coordination.coordination_state_contract_generated import ( + COORDINATION_RUNTIME_SHADOW_BOOTSTRAP_REQUEST_SCHEMA, + ) + + runtime = tmp_path / "runtime" + state = tmp_path / "state.md" + state.write_text("# Goal\n\n## Agent Todo\n\n") + goal = {"id": "goal-a", "repo": str(tmp_path), "state_file": "state.md", + "coordination": {"registered_agents": ["alpha"], "agent_model": "peer_v1"}} + registry = tmp_path / "registry.json" + registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [goal]})) + service = ChatActionService(store=ChatActionStore(runtime / "chat/actions"), registry_path=registry) + preview = service.preview({"action_kind": "team.plan", "summary": "Assign work", "context": {}, + "normalized_parameters": {"goal_id": "goal-a", "plan": _agent_plan()}, + "idempotency_key": "lost-original-response"}) + write = team_plan_adapter.write_captured_todo_state + + def lose_response(*args, **kwargs): + write(*args, **kwargs) + raise OSError("response lost after publication") + + with monkeypatch.context() as failed_response: + failed_response.setattr(team_plan_adapter, "write_captured_todo_state", lose_response) + assert service.apply(preview["proposal_id"])["proposal"]["status"] == "failed" + assert state.read_text().count("loopx:team-plan:") == 1 + state.write_text(state.read_text() + "\nLater owner data must survive.\n") + original = state.read_bytes() + projection, snapshot = build_runtime_shadow_source_snapshot( + goal=goal, runtime_root=runtime, state_path=state, registry_path=registry) + request = {"schema_version": COORDINATION_RUNTIME_SHADOW_BOOTSTRAP_REQUEST_SCHEMA, + "runtime_root": str(runtime), "goal_id": "goal-a", "operation_id": "bootstrap:plan-recovery", + "source_version": "source:before-crash", "projection": projection, "source_snapshot": snapshot} + repo = Path(__file__).resolve().parents[2] + child = subprocess.Popen(["node", "--no-warnings", "--experimental-strip-types", + str(repo / "tests/control_plane_ts/fixtures/shadow_management_crash_worker.ts"), + "bootstrap-public", json.dumps(request), "bootstrap_prepared"], + cwd=repo, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + try: + ready, _, _ = select.select([child.stdout], [], [], 15) + assert ready, "bootstrap did not reach its real filesystem barrier" + assert child.stdout.readline().strip() == "ready:bootstrap_prepared" + finally: + child.kill() + child.communicate(timeout=10) + pending = read_shadow_management_state(runtime, "goal-a") + assert pending["status"] == "bootstrapping" + # Restart from an isolated package with the old capture producer absent. + invoke = _capture_free_chat((runtime, state, service, preview), tmp_path) + replay = invoke("historical") + assert replay["action"] == "reused" and replay["created_todo_ids"] == [] + recovered = service.apply(preview["proposal_id"])["proposal"] + assert recovered["status"] == "applied", recovered + assert replay["lane_todo_ids"] == recovered["receipt"]["resource_ids"]["lane_todo_ids"] + assert state.read_bytes() == original + assert read_shadow_management_state(runtime, "goal-a") == pending + + # Recovery is tied to the exact intent, actor and original reviewed basis. + saved = service.store.load(preview["proposal_id"]) + original_plan = {**saved["normalized_parameters"]["plan"], "proposal_id": preview["proposal_id"]} + for changed in ( + {"proposal": {**original_plan, "objective": "Changed intent"}}, + {"agent_id": "alpha"}, + {"expected_state_fingerprint": "different-basis"}, + ): + with pytest.raises(team_plan_adapter.EffectRuntimeRejected, match="identity mismatch"): + team_plan_adapter.settle_team_plan(**{ + "registry_path": registry, "goal_id": "goal-a", "agent_id": None, + "proposal": original_plan, "expected_state_fingerprint": saved["expected_state_fingerprint"], + "basis_agent_id": "alpha", **changed, + }) + assert state.read_bytes() == original + # A different plan still needs normal admission; recovery cannot restart work. + fresh = service.preview({"action_kind": "team.plan", "summary": "New work", "context": {}, + "normalized_parameters": {"goal_id": "goal-a", "plan": _agent_plan()}, + "idempotency_key": "new-request-during-hold"}) + refused = service.apply(fresh["proposal_id"])["proposal"] + assert refused["status"] == "failed", refused + assert state.read_bytes() == original + assert read_shadow_management_state(runtime, "goal-a") == pending