diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 9fed8b3b1b..3360d3f234 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -145,8 +145,16 @@ producer modules using the qualified CLI startup isolation above. Shadow diagnostics now have an independent read-only boundary: retained outbox disk codecs live in `local_authority_shadow_history.py`, and status/candidate -read transport in `local_authority_shadow_readback.py`. The public status CLI -does not load source producers; drain still uses its existing orchestration. +read transport in `local_authority_shadow_readback.py`. The public status and enabled +`coordination-shadow inspect`/`qualify`/`read-candidate` commands call that owner +without loading source producers. Full source and normal-wheel observations +retain matched/refused, default-off, source-drift and missing/corrupt-store +recovery behavior with all four producer modules absent. Historical imports +remain exact aliases; bootstrap, rollback and promotion retain their live adapter. +The existing TS qualifier now propagates provider read refusals before lineage +validation can mislabel corruption as `bootstrap_required`; valid/missing stores +retain their behavior. This is an explicit failure-diagnostic correction, with +no new RPC or migration authority; drain still uses its existing orchestration. Original entry identity, malformed evidence, cursor residue and missing-store behavior remain unchanged. Python retains historical filesystem decoding and presentation IO; TS retains history proof, delivery and cleanup authority. 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 c9ea8f831a..b0a01bb434 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 @@ -116,7 +116,14 @@ producer module。 shadow 诊断现在有独立只读边界：历史 outbox 磁盘 codec 位于 `local_authority_shadow_history.py`，status／candidate read transport 位于 -`local_authority_shadow_readback.py`。公开 status CLI 不加载源 producer；drain +`local_authority_shadow_readback.py`。公开 status 和启用的 +`coordination-shadow inspect`／`qualify`／`read-candidate` 直接调用保留 owner， +不加载源 producer。源码和正常 wheel 物理删除四个 producer 后，完整读回保持 +匹配／拒绝、默认关闭、源漂移、store 缺失／损坏及恢复结果；历史 import 保留 +同一函数 alias，bootstrap／rollback／promotion 的活跃 adapter 继续受支持。 +既有 TS qualifier 保留 provider 读取失败，避免把损坏 journal 误称尚未 +bootstrap；正常／缺失状态保留原结果。这是明确的失败诊断修复，无新 RPC 或 +迁移权限；drain 仍走现有 orchestration。原 entry 身份、损坏证据、cursor 残留和缺失 store 行为 保持。Python 保留历史文件解码及展示 IO；TS 保留历史证明、投递与清理权限。 旧 producer 的已有导入为支持调用方保留 alias，不复制实现。源码及正常 wheel diff --git a/docs/reference/authority-observation-retirement.md b/docs/reference/authority-observation-retirement.md index 050a928055..4a09018aee 100644 --- a/docs/reference/authority-observation-retirement.md +++ b/docs/reference/authority-observation-retirement.md @@ -71,6 +71,22 @@ counts are diagnostic hints: they do not authorize cleanup or prove delivery. A missing candidate stays missing; status never creates a replacement identity, repairs bytes, drains an outbox or grants a writer permission. +The enabled `coordination-shadow inspect`, `qualify` and `read-candidate` +commands likewise call `local_authority_shadow_readback.py` directly. Their +complete source snapshot, default-off gate, bounded qualification and refusal +results retain the same TS owners. Source and normal-wheel checks remove all +four old producers and compare byte-stable observations, insufficient coverage, +missing Todo, source drift, unreadable/missing store and subsequent recovery. +The historical `runtime_shadow.py` imports remain aliases to the same functions. +The shared TS qualifier now preserves `provider_protocol_violation` and +`provider_read_unavailable` rather than mislabelling a failed provider read as +`bootstrap_required`. Consequently a corrupt journal makes inspection fail, +instead of reporting a missing baseline; qualification and candidate reads +retain the provider refusal too. Missing and valid stores retain their existing +behavior. Recovery restores the original bytes, without creating a new lineage. +Bootstrap, rollback and promotion still load their supported management adapter; +this reader exit grants no writer cutoff, import or promotion authority. + The retained Python filesystem codecs preserve the existing wire and errors; candidate proof and drain effects remain in their TypeScript owners. Supported unpromoted writes and `authority-shadow drain` still need their current producer diff --git a/loopx/cli_commands/coordination_shadow.py b/loopx/cli_commands/coordination_shadow.py index ab698ee6c8..31f1f894fb 100644 --- a/loopx/cli_commands/coordination_shadow.py +++ b/loopx/cli_commands/coordination_shadow.py @@ -10,6 +10,7 @@ from ..control_plane.coordination.authority_source_capture import build_runtime_shadow_source_snapshot from ..control_plane.coordination.configuration import resolve_coordination_runtime_shadow_config +from ..control_plane.coordination import local_authority_shadow_readback as observation from ..control_plane.coordination.promotion_review import execute_reviewed_coordination_promotion from ..control_plane.coordination.shadow_goal_scope import shadow_goal_scope from ..control_plane.projects.registry_codec import load_project_registry @@ -344,8 +345,6 @@ def handle_coordination_shadow_command( } print_payload(payload, output_format(args), _render) return 1 - # Only explicit, enabled legacy shadow actions load the producer. - from ..control_plane.coordination import runtime_shadow as shadow _, _, state_path = resolve_goal_state(registry=registry, goal_id=args.goal_id, project_override=args.project, state_file_override=args.state_file) @@ -357,7 +356,7 @@ def handle_coordination_shadow_command( projection_version = _projection_version(projection) projected_todos = projection.get("todos") projected_leases = projection.get("leases") - inspection = {"status": "not_evaluated"} if args.coordination_shadow_command == "rollback" else shadow.inspect_coordination_runtime_shadow( + inspection = {"status": "not_evaluated"} if args.coordination_shadow_command == "rollback" else observation.inspect_coordination_runtime_shadow( goal=goal, runtime_root=runtime_root, goal_id=args.goal_id, @@ -388,6 +387,8 @@ def handle_coordination_shadow_command( "decision_read_from_shadow": False, } if args.coordination_shadow_command == "bootstrap" and args.execute: + from ..control_plane.coordination import runtime_shadow as shadow + from ..control_plane.coordination.shadow_management import read_shadow_management_state management = read_shadow_management_state(runtime_root, args.goal_id) if management is not None and management["status"] in {"bootstrapping", "active"}: @@ -412,7 +413,7 @@ def handle_coordination_shadow_command( payload["executed"] = True payload["bootstrap"] = bootstrap if bootstrap.get("status") in {"applied", "replayed", "recovered"}: - payload["inspection"] = shadow.inspect_coordination_runtime_shadow( + payload["inspection"] = observation.inspect_coordination_runtime_shadow( goal=goal, runtime_root=runtime_root, goal_id=args.goal_id, @@ -426,7 +427,7 @@ def handle_coordination_shadow_command( and final_inspection.get("status") == "matched" ) if args.coordination_shadow_command == "qualify": - qualification = shadow.qualify_coordination_runtime_shadow( + qualification = observation.qualify_coordination_runtime_shadow( goal=goal, runtime_root=runtime_root, goal_id=args.goal_id, @@ -438,7 +439,7 @@ def handle_coordination_shadow_command( payload["qualification"] = qualification payload["ok"] = qualification.get("status") == "qualified" if args.coordination_shadow_command == "read-candidate": - read_candidate = shadow.read_coordination_runtime_shadow_todo_candidate( + read_candidate = observation.read_coordination_runtime_shadow_todo_candidate( goal=goal, runtime_root=runtime_root, goal_id=args.goal_id, @@ -453,6 +454,8 @@ def handle_coordination_shadow_command( and read_candidate.get("decision_read_from_shadow") is False ) if args.coordination_shadow_command == "promote": + from ..control_plane.coordination import runtime_shadow as shadow + # A saved plan owns qualification policy, so the flag defaults to # `None`; normalize it only for the freshly reviewed path. minimum_operations = args.minimum_operations if args.minimum_operations is not None else 3 @@ -499,6 +502,8 @@ def handle_coordination_shadow_command( payload["expected_provider_revision"] = provider_revision payload["expected_bootstrap_operation_id"] = pending_bootstrap if args.execute: + from ..control_plane.coordination import runtime_shadow as shadow + rollback = shadow.rollback_coordination_runtime_shadow( goal=goal, runtime_root=runtime_root, goal_id=args.goal_id, operation_id=f"shadow-rollback:{args.goal_id}:{provider_revision or pending_bootstrap}", diff --git a/loopx/control_plane/coordination/local_authority_shadow_readback.py b/loopx/control_plane/coordination/local_authority_shadow_readback.py index c5a1214ba6..702413b009 100644 --- a/loopx/control_plane/coordination/local_authority_shadow_readback.py +++ b/loopx/control_plane/coordination/local_authority_shadow_readback.py @@ -5,6 +5,7 @@ """ from __future__ import annotations +from collections.abc import Callable, Mapping from dataclasses import asdict from pathlib import Path from typing import Any @@ -15,10 +16,18 @@ from ..effect_runtime import effect_runtime_result from . import local_authority_shadow_history as outbox from .coordination_state_contract_generated import ( + COORDINATION_RUNTIME_SHADOW_INSPECT_REQUEST_SCHEMA as RUNTIME_SHADOW_INSPECT_REQUEST_SCHEMA_VERSION, + COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA, + COORDINATION_RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA as RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA_VERSION, + COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA, + COORDINATION_RUNTIME_SHADOW_TODO_READ_REQUEST_SCHEMA as RUNTIME_SHADOW_TODO_READ_REQUEST_SCHEMA_VERSION, + COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA, LOCAL_AUTHORITY_SHADOW_READ_REQUEST_SCHEMA, LOCAL_AUTHORITY_SHADOW_READ_RESULT_SCHEMA, ) -from .local_authority_shadow_projection import head_digest +from .local_authority_shadow_projection import ( + head_digest, source_effect_runtime_result as source_effect_runtime_result, +) from .configuration import resolve_coordination_runtime_shadow_config, local_authority_shadow_summary from .shadow_management import read_shadow_capture_binding @@ -180,3 +189,166 @@ def local_authority_shadow_status( if store_bytes is not None else None, } + + +RuntimeInvoker = Callable[..., object] +RUNTIME_SHADOW_INSPECT_METHOD = "coordination.runtime_shadow.inspect" +RUNTIME_SHADOW_QUALIFY_METHOD = "coordination.runtime_shadow.qualify" +RUNTIME_SHADOW_TODO_READ_METHOD = "coordination.runtime_shadow.todo_read_candidate" + + +def inspect_coordination_runtime_shadow( + *, + goal: Mapping[str, Any] | None, + runtime_root: Path, + goal_id: str, + projection: Mapping[str, Any], + source_snapshot: Mapping[str, Any] | None = None, + runtime_invoker: RuntimeInvoker = source_effect_runtime_result, +) -> dict[str, object]: + """Read parity evidence without allowing the shadow to drive decisions.""" + + config = resolve_coordination_runtime_shadow_config(goal) + if not config.enabled: + return { + "schema_version": COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA, + "status": "disabled", + "reason_code": config.reason_code, + "parity_matches": False, + "bootstrap_required": False, + "decision_read_from_shadow": False, + } + request = { + "schema_version": RUNTIME_SHADOW_INSPECT_REQUEST_SCHEMA_VERSION, + "runtime_root": str(runtime_root.expanduser().absolute()), + "goal_id": goal_id, + "source_snapshot": dict(source_snapshot or {}), + "projection": dict(projection), + } + try: + result = runtime_invoker(RUNTIME_SHADOW_INSPECT_METHOD, request) + except Exception as exc: + return { + "schema_version": COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA, + "status": "failed", + "reason_code": "shadow_runtime_unavailable", + "reason": str(exc), + "parity_matches": False, + "bootstrap_required": False, + "decision_read_from_shadow": False, + } + if not isinstance(result, Mapping): + return { + "schema_version": COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA, + "status": "failed", + "reason_code": "shadow_runtime_result_invalid", + "parity_matches": False, + "bootstrap_required": False, + "decision_read_from_shadow": False, + } + return dict(result) + +def qualify_coordination_runtime_shadow( + *, + goal: Mapping[str, Any] | None, + runtime_root: Path, + goal_id: str, + projection: Mapping[str, Any], + minimum_operations: int, + required_event_kinds: list[str], + source_snapshot: Mapping[str, Any] | None = None, + runtime_invoker: RuntimeInvoker = source_effect_runtime_result, +) -> dict[str, object]: + """Qualify coverage across a shadow lineage without serving from it.""" + + config = resolve_coordination_runtime_shadow_config(goal) + if not config.enabled: + return { + "schema_version": COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA, + "status": "disabled", + "reason_code": config.reason_code, + "qualified": False, + "primary_writeback_preserved": True, + "decision_read_from_shadow": False, + } + request = { + "schema_version": RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA_VERSION, + "runtime_root": str(runtime_root.expanduser().absolute()), + "goal_id": goal_id, + "source_snapshot": dict(source_snapshot or {}), + "projection": dict(projection), + "minimum_operations": minimum_operations, + "required_event_kinds": list(required_event_kinds), + } + try: + result = runtime_invoker(RUNTIME_SHADOW_QUALIFY_METHOD, request) + except Exception as exc: + return { + "schema_version": COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA, + "status": "failed", + "reason_code": "shadow_qualification_runtime_unavailable", + "reason": str(exc), + "qualified": False, + "primary_writeback_preserved": True, + "decision_read_from_shadow": False, + } + if not isinstance(result, Mapping): + return { + "schema_version": COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA, + "status": "failed", + "reason_code": "shadow_qualification_runtime_result_invalid", + "qualified": False, + "primary_writeback_preserved": True, + "decision_read_from_shadow": False, + } + return dict(result) + +def read_coordination_runtime_shadow_todo_candidate( + *, + goal: Mapping[str, Any] | None, + runtime_root: Path, + goal_id: str, + todo_id: str, + projection: Mapping[str, Any], + source_snapshot: Mapping[str, Any] | None = None, + runtime_invoker: RuntimeInvoker = source_effect_runtime_result, +) -> dict[str, object]: + """Read one parity-matched file Todo as pre-promotion evidence only.""" + + config = resolve_coordination_runtime_shadow_config(goal) + if not config.enabled: + return { + "schema_version": COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA, + "status": "disabled", + "reason_code": config.reason_code, + "read_candidate_qualified": False, + "decision_read_from_shadow": False, + } + request = { + "schema_version": RUNTIME_SHADOW_TODO_READ_REQUEST_SCHEMA_VERSION, + "runtime_root": str(runtime_root.expanduser().absolute()), + "goal_id": goal_id, + "source_snapshot": dict(source_snapshot or {}), + "todo_id": todo_id, + "projection": dict(projection), + } + try: + result = runtime_invoker(RUNTIME_SHADOW_TODO_READ_METHOD, request) + except Exception as exc: + return { + "schema_version": COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA, + "status": "failed", + "reason_code": "shadow_todo_read_runtime_unavailable", + "reason": str(exc), + "read_candidate_qualified": False, + "decision_read_from_shadow": False, + } + if not isinstance(result, Mapping): + return { + "schema_version": COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA, + "status": "failed", + "reason_code": "shadow_todo_read_runtime_result_invalid", + "read_candidate_qualified": False, + "decision_read_from_shadow": False, + } + return dict(result) diff --git a/loopx/control_plane/coordination/runtime_shadow.py b/loopx/control_plane/coordination/runtime_shadow.py index 7cc83c4707..35a3e748d2 100644 --- a/loopx/control_plane/coordination/runtime_shadow.py +++ b/loopx/control_plane/coordination/runtime_shadow.py @@ -16,14 +16,8 @@ COORDINATION_RUNTIME_SHADOW_EXACT_BOOTSTRAP_REQUEST_SCHEMA, COORDINATION_RUNTIME_SHADOW_BOOTSTRAP_REQUEST_SCHEMA as RUNTIME_SHADOW_BOOTSTRAP_REQUEST_SCHEMA_VERSION, COORDINATION_RUNTIME_SHADOW_BOOTSTRAP_RESULT_SCHEMA, - COORDINATION_RUNTIME_SHADOW_INSPECT_REQUEST_SCHEMA as RUNTIME_SHADOW_INSPECT_REQUEST_SCHEMA_VERSION, - COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA, - COORDINATION_RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA as RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA_VERSION, - COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA, COORDINATION_RUNTIME_SHADOW_ROLLBACK_REQUEST_SCHEMA as RUNTIME_SHADOW_ROLLBACK_REQUEST_SCHEMA_VERSION, COORDINATION_RUNTIME_SHADOW_ROLLBACK_RESULT_SCHEMA, - COORDINATION_RUNTIME_SHADOW_TODO_READ_REQUEST_SCHEMA as RUNTIME_SHADOW_TODO_READ_REQUEST_SCHEMA_VERSION, - COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA, LOCAL_COORDINATION_PROMOTION_REVIEW_REQUEST_SCHEMA, LOCAL_COORDINATION_PROMOTION_REVIEW_RESULT_SCHEMA, ) @@ -52,11 +46,23 @@ ) -RUNTIME_SHADOW_INSPECT_METHOD = "coordination.runtime_shadow.inspect" +from .local_authority_shadow_readback import ( # noqa: F401 + RUNTIME_SHADOW_INSPECT_REQUEST_SCHEMA_VERSION, + COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA, + RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA_VERSION, + COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA, + RUNTIME_SHADOW_TODO_READ_REQUEST_SCHEMA_VERSION, + COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA, + inspect_coordination_runtime_shadow, + qualify_coordination_runtime_shadow, + read_coordination_runtime_shadow_todo_candidate, + RUNTIME_SHADOW_INSPECT_METHOD, + RUNTIME_SHADOW_QUALIFY_METHOD, + RUNTIME_SHADOW_TODO_READ_METHOD, +) + RUNTIME_SHADOW_BOOTSTRAP_METHOD = "coordination.runtime_shadow.bootstrap" RUNTIME_SHADOW_ROLLBACK_METHOD = "coordination.runtime_shadow.rollback" -RUNTIME_SHADOW_QUALIFY_METHOD = "coordination.runtime_shadow.qualify" -RUNTIME_SHADOW_TODO_READ_METHOD = "coordination.runtime_shadow.todo_read_candidate" LOCAL_AUTHORITY_PROMOTION_REVIEW_METHOD = "coordination.local_authority.promotion_review" @@ -193,114 +199,6 @@ def rollback_coordination_runtime_shadow( return dict(result) -def inspect_coordination_runtime_shadow( - *, - goal: Mapping[str, Any] | None, - runtime_root: Path, - goal_id: str, - projection: Mapping[str, Any], - source_snapshot: Mapping[str, Any] | None = None, - runtime_invoker: RuntimeInvoker = effect_runtime_result, -) -> dict[str, object]: - """Read parity evidence without allowing the shadow to drive decisions.""" - - config = resolve_coordination_runtime_shadow_config(goal) - if not config.enabled: - return { - "schema_version": COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA, - "status": "disabled", - "reason_code": config.reason_code, - "parity_matches": False, - "bootstrap_required": False, - "decision_read_from_shadow": False, - } - request = { - "schema_version": RUNTIME_SHADOW_INSPECT_REQUEST_SCHEMA_VERSION, - "runtime_root": str(runtime_root.expanduser().absolute()), - "goal_id": goal_id, - "source_snapshot": dict(source_snapshot or {}), - "projection": dict(projection), - } - try: - result = runtime_invoker(RUNTIME_SHADOW_INSPECT_METHOD, request) - except Exception as exc: - return { - "schema_version": COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA, - "status": "failed", - "reason_code": "shadow_runtime_unavailable", - "reason": str(exc), - "parity_matches": False, - "bootstrap_required": False, - "decision_read_from_shadow": False, - } - if not isinstance(result, Mapping): - return { - "schema_version": COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA, - "status": "failed", - "reason_code": "shadow_runtime_result_invalid", - "parity_matches": False, - "bootstrap_required": False, - "decision_read_from_shadow": False, - } - return dict(result) - - -def qualify_coordination_runtime_shadow( - *, - goal: Mapping[str, Any] | None, - runtime_root: Path, - goal_id: str, - projection: Mapping[str, Any], - minimum_operations: int, - required_event_kinds: list[str], - source_snapshot: Mapping[str, Any] | None = None, - runtime_invoker: RuntimeInvoker = effect_runtime_result, -) -> dict[str, object]: - """Qualify coverage across a shadow lineage without serving from it.""" - - config = resolve_coordination_runtime_shadow_config(goal) - if not config.enabled: - return { - "schema_version": COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA, - "status": "disabled", - "reason_code": config.reason_code, - "qualified": False, - "primary_writeback_preserved": True, - "decision_read_from_shadow": False, - } - request = { - "schema_version": RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA_VERSION, - "runtime_root": str(runtime_root.expanduser().absolute()), - "goal_id": goal_id, - "source_snapshot": dict(source_snapshot or {}), - "projection": dict(projection), - "minimum_operations": minimum_operations, - "required_event_kinds": list(required_event_kinds), - } - try: - result = runtime_invoker(RUNTIME_SHADOW_QUALIFY_METHOD, request) - except Exception as exc: - return { - "schema_version": COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA, - "status": "failed", - "reason_code": "shadow_qualification_runtime_unavailable", - "reason": str(exc), - "qualified": False, - "primary_writeback_preserved": True, - "decision_read_from_shadow": False, - } - if not isinstance(result, Mapping): - return { - "schema_version": COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA, - "status": "failed", - "reason_code": "shadow_qualification_runtime_result_invalid", - "qualified": False, - "primary_writeback_preserved": True, - "decision_read_from_shadow": False, - } - return dict(result) - - def review_local_coordination_authority_promotion( *, goal: Mapping[str, Any] | None, @@ -369,54 +267,3 @@ def review_local_coordination_authority_promotion( "legacy_fallback_used": False, } return dict(result) - - -def read_coordination_runtime_shadow_todo_candidate( - *, - goal: Mapping[str, Any] | None, - runtime_root: Path, - goal_id: str, - todo_id: str, - projection: Mapping[str, Any], - source_snapshot: Mapping[str, Any] | None = None, - runtime_invoker: RuntimeInvoker = effect_runtime_result, -) -> dict[str, object]: - """Read one parity-matched file Todo as pre-promotion evidence only.""" - - config = resolve_coordination_runtime_shadow_config(goal) - if not config.enabled: - return { - "schema_version": COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA, - "status": "disabled", - "reason_code": config.reason_code, - "read_candidate_qualified": False, - "decision_read_from_shadow": False, - } - request = { - "schema_version": RUNTIME_SHADOW_TODO_READ_REQUEST_SCHEMA_VERSION, - "runtime_root": str(runtime_root.expanduser().absolute()), - "goal_id": goal_id, - "source_snapshot": dict(source_snapshot or {}), - "todo_id": todo_id, - "projection": dict(projection), - } - try: - result = runtime_invoker(RUNTIME_SHADOW_TODO_READ_METHOD, request) - except Exception as exc: - return { - "schema_version": COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA, - "status": "failed", - "reason_code": "shadow_todo_read_runtime_unavailable", - "reason": str(exc), - "read_candidate_qualified": False, - "decision_read_from_shadow": False, - } - if not isinstance(result, Mapping): - return { - "schema_version": COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA, - "status": "failed", - "reason_code": "shadow_todo_read_runtime_result_invalid", - "read_candidate_qualified": False, - "decision_read_from_shadow": False, - } - return dict(result) diff --git a/loopx/control_plane/coordination/runtime_shadow.ts b/loopx/control_plane/coordination/runtime_shadow.ts index 1971164f72..7c2d2780ca 100644 --- a/loopx/control_plane/coordination/runtime_shadow.ts +++ b/loopx/control_plane/coordination/runtime_shadow.ts @@ -358,6 +358,11 @@ export async function qualifyCoordinationShadowLineageUnderLocks( const store = dependencies.createStore?.(join(request.runtime_root, "authority-shadow", "file-v0"), request.goal_id) ?? new FileAuthorityStore(join(request.runtime_root, "authority-shadow", "file-v0"), request.goal_id, { existingOnly: true }); const initial = await store.loadAuthority(); + // A failed read is not an absent baseline. Preserve the provider-owned + // refusal before lineage validation can mislabel it bootstrap_required. + if (initial.status === "failed" || initial.status === "unavailable") { + throw new ShadowLineageError(initial.reason_code); + } if (initial.status === "loaded" && initial.head.capture_profile !== "file_outbox_v1") throw new ShadowLineageError("legacy_lineage_ineligible"); const binding = await requireShadowCaptureBinding(request.runtime_root, request.goal_id); const lineage = await loadValidatedShadowLineage(store, request.runtime_root, request.goal_id, binding); diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json index e076e23f56..a7cc284380 100644 --- a/loopx/semantics/project_registry_io_manifest_v1.json +++ b/loopx/semantics/project_registry_io_manifest_v1.json @@ -671,7 +671,7 @@ }, { "site": "loopx/cli_commands/coordination_shadow.py::.handle_coordination_shadow_command::codec_read:load_project_registry#1", - "line": 245, + "line": 246, "column": 20, "kind": "codec_read", "api": "load_project_registry", @@ -1119,7 +1119,7 @@ }, { "site": "loopx/control_plane/coordination/local_authority_shadow_readback.py::.local_authority_shadow_status::codec_read:load_registry#1", - "line": 84, + "line": 93, "column": 16, "kind": "codec_read", "api": "load_registry", diff --git a/tests/control_plane/test_canonical_todo_writer_isolation.py b/tests/control_plane/test_canonical_todo_writer_isolation.py index 5c79e17cf8..546abad098 100644 --- a/tests/control_plane/test_canonical_todo_writer_isolation.py +++ b/tests/control_plane/test_canonical_todo_writer_isolation.py @@ -456,6 +456,7 @@ def test_explicit_source_writer_import_preserves_the_existing_seam(tmp_path): result = subprocess.run( [sys.executable, "-c", """ import sys +import sysconfig import loopx.todos as todos assert 'loopx.control_plane.todos.line_update' not in sys.modules assert 'loopx.control_plane.todos.legacy_mutation' not in sys.modules diff --git a/tests/control_plane/test_coordination_shadow_command.py b/tests/control_plane/test_coordination_shadow_command.py index 5d8078dab4..e55da2dc8f 100644 --- a/tests/control_plane/test_coordination_shadow_command.py +++ b/tests/control_plane/test_coordination_shadow_command.py @@ -5,6 +5,7 @@ from pathlib import Path from loopx.control_plane.coordination import runtime_shadow as shadow +from loopx.control_plane.coordination import local_authority_shadow_readback as observation from loopx.control_plane.coordination.authority_source_capture import build_todo_runtime_shadow_projection from loopx.cli import build_parser from loopx.cli_commands import coordination_shadow as command @@ -119,7 +120,7 @@ def test_coordination_shadow_inspect_is_read_only_and_compact( tmp_path: Path, ) -> None: monkeypatch.setattr( - shadow, + observation, "inspect_coordination_runtime_shadow", lambda **_kwargs: { "status": "missing", @@ -161,7 +162,7 @@ def test_coordination_shadow_bootstrap_requires_execute_and_reads_back_parity( ] ) monkeypatch.setattr( - shadow, + observation, "inspect_coordination_runtime_shadow", lambda **_kwargs: next(inspections), ) @@ -208,7 +209,7 @@ def test_coordination_shadow_bootstrap_uses_source_session_goal_ref( } bootstrap_request: dict[str, object] = {} monkeypatch.setattr( - shadow, + observation, "inspect_coordination_runtime_shadow", lambda **_kwargs: { "status": "matched", @@ -326,7 +327,7 @@ def test_coordination_shadow_reads_parity_matched_todo_candidate( tmp_path: Path, ) -> None: monkeypatch.setattr( - shadow, + observation, "inspect_coordination_runtime_shadow", lambda **_kwargs: { "status": "matched", @@ -347,7 +348,7 @@ def read_candidate(**kwargs) -> dict[str, object]: } monkeypatch.setattr( - shadow, + observation, "read_coordination_runtime_shadow_todo_candidate", read_candidate, ) @@ -374,7 +375,7 @@ def test_coordination_shadow_qualify_applies_coverage_policy( tmp_path: Path, ) -> None: monkeypatch.setattr( - shadow, + observation, "inspect_coordination_runtime_shadow", lambda **_kwargs: { "status": "matched", @@ -392,7 +393,7 @@ def qualify(**kwargs) -> dict[str, object]: "decision_read_from_shadow": False, } - monkeypatch.setattr(shadow, "qualify_coordination_runtime_shadow", qualify) + monkeypatch.setattr(observation, "qualify_coordination_runtime_shadow", qualify) result, payload = _run( monkeypatch, tmp_path, @@ -417,7 +418,7 @@ def test_coordination_shadow_promote_previews_and_applies_only_through_review_ow tmp_path: Path, ) -> None: monkeypatch.setattr( - shadow, + observation, "inspect_coordination_runtime_shadow", lambda **_kwargs: { "status": "matched", @@ -502,7 +503,7 @@ def test_coordination_shadow_rollback_passes_exact_selector_to_management_owner( ] ) monkeypatch.setattr( - shadow, + observation, "inspect_coordination_runtime_shadow", lambda **_kwargs: next(inspections), ) diff --git a/tests/control_plane/test_shadow_observation_retirement.py b/tests/control_plane/test_shadow_observation_retirement.py new file mode 100644 index 0000000000..1b748b0e39 --- /dev/null +++ b/tests/control_plane/test_shadow_observation_retirement.py @@ -0,0 +1,121 @@ +"""Retained migration observations must survive source-producer retirement.""" +from __future__ import annotations + +import json +from pathlib import Path +import subprocess + +import pytest + +from shadow_e2e_fixture import workspace +from test_canonical_todo_writer_isolation import ( + isolated_todo_distribution, retirement_wheel, # noqa: F401,F811 +) + + +def _bytes(root: Path): + return {str(path.relative_to(root)): path.read_bytes() + for path in root.rglob("*") if path.is_file() and path.suffix != ".lock"} + + +@pytest.mark.parametrize("producers_present", [True, False], ids=["present", "absent"]) +def test_complete_shadow_observation_without_producers( + tmp_path, isolated_todo_distribution, producers_present, # noqa: F811 -- imported pytest fixture +): + package, command = isolated_todo_distribution + w = workspace(tmp_path / "shadow", bootstrap=False) + w.command, w.cwd, w.package = tuple(command), tmp_path, package + assert w.cli("coordination-shadow", "bootstrap", "--execute")["bootstrap"]["status"] == "applied" + todo_id = w.add("Preserve the original source work")["todo_id"] + w.add("Second source mutation") + w.add("Third source mutation") + if not producers_present: + for name in ("runtime_shadow", "runtime_shadow_writer_adapter", + "local_authority_shadow_adapter", "local_authority_shadow_outbox"): + (package / f"control_plane/coordination/{name}.py").unlink() + check = subprocess.run([command[0], "-S", "-c", + "import importlib.util; assert importlib.util.find_spec(" + "'loopx.control_plane.coordination.runtime_shadow') is None"], + cwd=tmp_path, capture_output=True, text=True, timeout=45) + assert check.returncode == 0, check.stdout + check.stderr + + def observe(*args, success=True): + before = _bytes(w.registry.parent) + result = w.cli("coordination-shadow", *args, success=success) + assert _bytes(w.registry.parent) == before + assert result["executed"] is False + assert result["decision_read_from_shadow"] is False + return result + + inspected = observe("inspect") + assert inspected["inspection"]["status"] == "matched" + qualified = observe("qualify", "--require-event-kind", "todo_add") + assert qualified["qualification"]["status"] == "qualified" + insufficient = observe("qualify", "--minimum-operations", "4", success=False) + assert insufficient["ok"] is False + assert insufficient["qualification"]["qualified"] is False + candidate = observe("read-candidate", "--todo-id", todo_id) + assert candidate["read_candidate"]["read_candidate_qualified"] is True + missing = observe("read-candidate", "--todo-id", "todo_missing", success=False) + assert missing["ok"] is False + assert missing["read_candidate"]["read_candidate_qualified"] is False + + original = w.state.read_bytes() + w.state.write_text(w.state.read_text().replace("Third source mutation", "Changed source work")) + drifted = observe("inspect", success=False) + assert drifted["inspection"]["parity_matches"] is False + refused = observe("read-candidate", "--todo-id", todo_id, success=False) + assert refused["ok"] is False + assert refused["read_candidate"]["read_candidate_qualified"] is False + w.state.write_bytes(original) + assert observe("inspect")["inspection"]["status"] == "matched" + + store = next((w.runtime / "authority-shadow/file-v0").glob("authority-store-*.json")) + stored = store.read_bytes() + store.write_text("{unreadable historical store") + for args in [("inspect",), ("qualify",), ("read-candidate", "--todo-id", todo_id)]: + unavailable = observe(*args, success=False) + assert unavailable["ok"] is False + detail = unavailable[{"inspect": "inspection", "qualify": "qualification", + "read-candidate": "read_candidate"}[args[0]]] + assert detail["reason_code"] == "provider_protocol_violation" + store.write_bytes(stored) + assert observe("inspect")["inspection"]["status"] == "matched" + + store.rename(store.with_suffix(".retained")) + store.mkdir() + for args in [("inspect",), ("qualify",), ("read-candidate", "--todo-id", todo_id)]: + unavailable = observe(*args, success=False) + assert unavailable["ok"] is False + detail = unavailable[{"inspect": "inspection", "qualify": "qualification", + "read-candidate": "read_candidate"}[args[0]]] + assert detail["reason_code"] == "provider_read_unavailable" + store.rmdir() + store.with_suffix(".retained").rename(store) + assert observe("inspect")["inspection"]["status"] == "matched" + + directory = store.parent + retained = directory.with_name("retained-original") + directory.rename(retained) + assert observe("inspect", success=False)["inspection"]["parity_matches"] is False + for args in [("qualify",), ("read-candidate", "--todo-id", todo_id)]: + assert observe(*args, success=False)["ok"] is False + assert not directory.exists() + retained.rename(directory) + assert observe("inspect")["inspection"]["status"] == "matched" + + registry = json.loads(w.registry.read_text()) + registry["goals"][0]["coordination"]["runtime_shadow"]["enabled"] = False + w.registry.write_text(json.dumps(registry)) + for args in [("inspect",), ("qualify",), ("read-candidate", "--todo-id", todo_id)]: + disabled = observe(*args, success=False) + assert disabled["error_code"] == "coordination_shadow_not_enabled" + + +def test_historical_observation_imports_share_the_retained_owner(): + from loopx.control_plane.coordination import local_authority_shadow_readback as reader + from loopx.control_plane.coordination import runtime_shadow as old + + for name in ("inspect_coordination_runtime_shadow", "qualify_coordination_runtime_shadow", + "read_coordination_runtime_shadow_todo_candidate"): + assert getattr(old, name) is getattr(reader, name) diff --git a/tests/control_plane_ts/coordination_runtime_shadow.test.ts b/tests/control_plane_ts/coordination_runtime_shadow.test.ts index a73cda433f..dde01b6a34 100644 --- a/tests/control_plane_ts/coordination_runtime_shadow.test.ts +++ b/tests/control_plane_ts/coordination_runtime_shadow.test.ts @@ -28,6 +28,29 @@ async function qualify(f: ShadowFixture, head: JsonObject, extra: JsonObject = { minimum_operations: 3, required_event_kinds: ["todo_add"], ...extra }); } +test("retained observations refuse corrupt provider bytes without suggesting bootstrap", async (t) => { + const {f, head} = await qualifiedFixture(t); + const source = await sourceRequest(f, head); + const original = await readFile(f.store.path); + await writeFile(f.store.path, "{corrupt journal"); + for (const [read, schema, extra] of [ + [inspectCoordinationRuntimeShadow, schemas.COORDINATION_RUNTIME_SHADOW_INSPECT_REQUEST_SCHEMA, {}], + [qualifyCoordinationRuntimeShadow, schemas.COORDINATION_RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA, + {minimum_operations: 3, required_event_kinds: ["todo_add"]}], + [readCoordinationRuntimeShadowTodoCandidate, schemas.COORDINATION_RUNTIME_SHADOW_TODO_READ_REQUEST_SCHEMA, + {todo_id: "todo_1"}], + ] as const) { + const result = await read({...source, schema_version: schema, ...extra}); + assert.equal(result.status, "failed"); + assert.equal(result.reason_code, "provider_protocol_violation"); + assert.equal(result.qualified, false); + assert.equal(result.decision_read_from_shadow, false); + assert.equal(await readFile(f.store.path, "utf8"), "{corrupt journal"); + } + await writeFile(f.store.path, original); + assert.equal((await qualify(f, head)).status, "qualified"); +}); + test("bootstrap is exactly replayable, has no mutation receipt and cannot overwrite a different baseline", async (t) => { const f = await fixture(t); const request = { ...await sourceRequest(f, f.baseline), schema_version: schemas.COORDINATION_RUNTIME_SHADOW_BOOTSTRAP_REQUEST_SCHEMA,