fix(backup): refuse registry drift after dependency discovery

docs/architecture/rfcs/complete-state-recovery-v0.md

@@ -552,6 +552,17 @@ publication when they differ or lack a witness. File/SQLite tests cover both a
 persistent change and A → B → A; a later live-path reread cannot prove this case.
 Routed sources and missing witnesses have separate negative coverage.
 
+The discovery-to-copy interval has a separate refusal: a registry declaration
+could previously change after planning, leaving the old selected document beside
+a new raw registry and a matching new configuration checkpoint. New full-backup
+plans retain the exact registry reads and freeze their archive routes; copied
+members must agree with those reads as well. Routing, discovery and configuration
+reuse the existing registry codec's immutable-read option. File/SQLite fixtures
+prove refusal, fresh-plan recovery and source-loss inert readback. This is a
+conservative byte-level check (even unrelated registry changes require retry),
+not a required-material export, shared owner epoch or M2 profile qualification.
+Historical plans and the scoped migration producer retain their prior contracts.
+
 The same rule covers the existing stored machine-configuration owner. Its exact
 read supplies the checkpoint and witness together, including confirmed absence;
 changed, newly created or removed configuration cannot contradict the raw copy.

docs/architecture/rfcs/complete-state-recovery-v0.zh-CN.md

@@ -466,6 +466,14 @@ Goal 文档，同时维持 inert/held。M2 仍需完整 required-owner closure
 覆盖持续变化与 A → B → A 回退，事后重读 live 路径不能证明这类情况；路由到
 外部的源 registry 与缺少见证各有负向覆盖。
 
+发现到复制之间另有拒绝边界：旧计划生成后 registry 声明可能改变，导致归档
+保留旧文档、新原始 registry 和与新 registry 匹配的配置 checkpoint。新完整备份
+计划保留实际 registry 读取及固定归档路径，已复制成员也必须与发现时字节一致。
+路由、发现和配置复用既有 registry codec 的不可变读取选项。File/SQLite 验证
+拒绝、重新规划恢复及失源后的 inert 读回。这是保守的字节检查（无关 registry
+变更也要求重试），不等于必要材料 export、共享 owner epoch 或 M2 profile 合格；
+历史计划和有界迁移 producer 保持原契约。
+
 同一规则也覆盖既有 machine 配置 store owner：checkpoint 和见证使用同一次
 实际读取，包括确认不存在。变更、新建或删除配置都不能与已复制文件矛盾；
 不可用的 machine 来源拒绝捕获，不被标成不存在。自包含的本地链接复用归档内

docs/reference/configuration-backup.md

@@ -217,6 +217,26 @@ source owner 的见证契约。私有 `execution.configuration_capture` 记录
 区间与匹配成员 revision，包括外部 registry 和存在的 machine 配置；它不证明
 与 authority、effect 等其他 owner 的跨源同时刻，也不代表 M2 `online` 合格。
 
+New full-backup plans also bind the registry bytes used to discover documents,
+source runtimes and lifecycle journals. The private
+`registry_discovery.source_capture` freezes their read interval, hashes and
+archive routes; publication checks those hashes against the saved members.
+Changing a declaration after planning now reports
+`backup discovery source differs from copied bytes; retry the backup` instead of
+saving old dependencies beside a new registry. Rebuild the plan and retry with
+fresh output names. Any registry byte change can require a retry, including
+unrelated metadata. Routing and configuration decoding reuse the same immutable
+registry reads. Historical plans without this evidence remain readable, and
+migration-only backups retain their existing contract. This IO agreement does
+not freeze document contents or establish a complete recovery-set profile.
+
+中文：新完整备份计划还绑定发现文档、源 runtime 和生命周期日志时实际读取的
+registry 字节。私有 `registry_discovery.source_capture` 固定读取区间、哈希与
+归档路径，并在发布前核对已保存成员。计划后声明改变会拒绝发布，避免保存
+“旧依赖 + 新 registry”；应重新规划并以新输出名称重试。任意 registry 字节变化
+（包括无关 metadata）都可能要求重试；路由与配置解码复用同一不可变读取。
+历史计划和迁移专用备份保持原契约。这不冻结文档内容，也不使完整恢复集合格。
+
 Full-backup discovery also retains declared local `authority_sources` entries
 with `kind=goal_doc`, including documents outside conventional state directories.
 The existing material resolver confines reads to the selected Goal's repository,

loopx/capabilities/configuration_backup.py

@@ -4,7 +4,7 @@
 
 from ..control_plane.effect_runtime import effect_runtime_result
 from ..control_plane.runtime.runtime_projection_route import resolve_goal_source_runtime_route
-from ..control_plane.projects.registry_codec import decode_registry_snapshot
+from ..control_plane.projects.registry_codec import load_registry
 from ..registry import registry_goals
 from .machine_configuration.store import read_stored_machine_configuration
 
@@ -18,30 +18,17 @@ def capture_configuration_backup(
     # cannot witness the bytes used here (the source may change A -> B -> A).
     sources = source_snapshots if source_snapshots is not None else {}
 
-    def read_registry(path: Path) -> dict[str, Any]:
-        source = path.expanduser().resolve()
-        if source not in sources:
-            try:
-                sources[source] = source.read_bytes()
-            except FileNotFoundError:
-                sources[source] = None
-        raw = sources[source]
-        if raw is None:
-            raise FileNotFoundError(path)
-        return decode_registry_snapshot(path, raw)
-
-    try:
-        registry = read_registry(registry_path)
-    except FileNotFoundError:
-        registry = {"goals": []}
+    registry = load_registry(registry_path, source_snapshots=sources)
     selected = goal_ids if goal_ids is not None else [str(goal["id"]) for goal in registry_goals(registry)]
     if len(selected) != len(set(selected)):
         raise ValueError("configuration backup Goal ids must be unique")
     snapshots = []
     for goal_id in selected:
-        route = resolve_goal_source_runtime_route(registry_path=registry_path, goal_id=goal_id, registry=registry)
+        route = resolve_goal_source_runtime_route(registry_path=registry_path, goal_id=goal_id,
+            registry=registry, source_snapshots=sources)
         source = Path(route["source_registry"])
-        matches = [goal for goal in registry_goals(read_registry(source)) if goal.get("id") == goal_id]
+        matches = [goal for goal in registry_goals(load_registry(source, source_snapshots=sources))
+                   if goal.get("id") == goal_id]
         if len(matches) != 1:
             raise ValueError("configuration backup requires exactly one source-owned Goal")
         snapshots.append({"goal_id": goal_id, "goal_configuration": matches[0]})

loopx/control_plane/projects/registry_codec.py

@@ -211,10 +211,28 @@ def decode_registry_snapshot(path: Path, raw_bytes: bytes) -> dict[str, Any]:
     return payload
 
 
-def load_registry(path: Path) -> dict[str, Any]:
+def load_registry(
+    path: Path, *, source_snapshots: dict[Path, bytes | None] | None = None,
+) -> dict[str, Any]:
     """Load a project registry or an object-only global registry."""
 
     expanded = path.expanduser()
+    if source_snapshots is not None:
+        # Capture callers must decode the same bytes that supply their IO
+        # witness. Retain absence too; rereading could conceal A -> B -> A.
+        route = expanded.absolute()
+        if route not in source_snapshots:
+            source = expanded.resolve()
+            if source not in source_snapshots:
+                try:
+                    source_snapshots[source] = source.read_bytes()
+                except FileNotFoundError:
+                    source_snapshots[source] = None
+            # Witness the invoked alias too. Its saved link can change while
+            # an independent copy of the original regular file still agrees.
+            source_snapshots[route] = source_snapshots[source]
+        raw = source_snapshots[route]
+        return {} if raw is None else decode_registry_snapshot(expanded, raw)
     if not expanded.exists():
         return {}
     return decode_registry_snapshot(expanded, expanded.read_bytes())

loopx/control_plane/runtime/runtime_projection_route.py

@@ -113,6 +113,7 @@ def resolve_goal_source_runtime_route(
     goal_id: str,
     runtime_root_override: str | None = None,
     registry: dict[str, Any] | None = None,
+    source_snapshots: dict[Path, bytes | None] | None = None,
 ) -> dict[str, Any]:
     """Resolve goal-scoped writes and reads to the canonical source runtime.
 
@@ -124,7 +125,9 @@ def resolve_goal_source_runtime_route(
     """
 
     invoked_registry = registry_path.expanduser().resolve()
-    registry_payload = registry if registry is not None else load_registry(invoked_registry)
+    registry_payload = registry if registry is not None else load_registry(
+        invoked_registry, source_snapshots=source_snapshots,
+    )
     invoked_runtime = _registry_runtime_root(
         registry_payload,
         registry_path=invoked_registry,
@@ -185,7 +188,7 @@ def resolve_goal_source_runtime_route(
             f"goal {goal_id!r} source_registry is missing; refusing to use the shared runtime as source"
         )
     try:
-        source_payload = load_registry(source_registry)
+        source_payload = load_registry(source_registry, source_snapshots=source_snapshots)
     except (OSError, ValueError, json.JSONDecodeError) as exc:
         raise ValueError(
             f"goal {goal_id!r} source_registry is unreadable; refusing to use the shared runtime as source"

loopx/state_backup.py

@@ -135,6 +135,7 @@ def _discover_targets(
     include_skills: bool,
     include_registry_projects: bool,
     configuration_source_registry: Path,
+    source_snapshots: dict[Path, bytes | None],
 ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[str], dict[str, Any]]:
     targets: list[dict[str, Any]] = []
     missing: list[dict[str, Any]] = []
@@ -185,13 +186,13 @@ def add(key: str, source_path: Path, archive_path: str) -> str | None:
     registry_source_registry_count = 0
     registry_source_registry_included_count = 0
     goal_runtime_roots: dict[str, str | None] = {}
-    if include_registry_projects or discovery_registry.exists():
-        try:
-            registry = load_registry(discovery_registry)
-            goals = registry_goals(registry)
-        except (OSError, ValueError, json.JSONDecodeError) as exc:
-            goals = []
-            warnings.append(f"could not discover registry projects from {discovery_registry}: {exc}")
+    try:
+        registry = load_registry(discovery_registry, source_snapshots=source_snapshots)
+        goals = registry_goals(registry)
+    except (OSError, ValueError, json.JSONDecodeError) as exc:
+        goals = []
+        warnings.append(f"could not discover registry projects from {discovery_registry}: {exc}")
+    if goals:
         if not include_registry_projects:
             goals = [
                 goal for goal in goals
@@ -215,9 +216,10 @@ def add(key: str, source_path: Path, archive_path: str) -> str | None:
                     source_registry = _resolved_source_registry(goal, registry_path=discovery_registry)
                     route = resolve_goal_source_runtime_route(
                         registry_path=discovery_registry, goal_id=goal_id, registry=registry,
+                        source_snapshots=source_snapshots,
                     )
                     source = Path(route["source_registry"])
-                    source_payload = registry if source == _resolved(discovery_registry) else load_registry(source)
+                    source_payload = load_registry(source, source_snapshots=source_snapshots)
                     matches = [row for row in registry_goals(source_payload) if row.get("id") == goal_id]
                     if len(matches) != 1:
                         raise ValueError("backup requires exactly one source-owned Goal")
@@ -490,6 +492,8 @@ def build_state_backup_plan(
         resolved_runtime_root / "registry.global.json" if include_registry_projects
         else resolved_project / ".loopx/registry.json"
     )
+    source_snapshots: dict[Path, bytes | None] = {}
+    discovery_started_at = datetime.now(timezone.utc).isoformat()
     targets, missing, warnings, registry_discovery = _discover_targets(
         project=resolved_project,
         runtime_root=resolved_runtime_root,
@@ -498,7 +502,13 @@ def build_state_backup_plan(
         include_skills=include_skills,
         include_registry_projects=include_registry_projects,
         configuration_source_registry=configuration_source_registry,
+        source_snapshots=source_snapshots,
     )
+    registry_discovery["source_capture"] = {
+        "started_at": discovery_started_at,
+        "completed_at": datetime.now(timezone.utc).isoformat(),
+        "source_revisions": _source_revisions(source_snapshots, targets),
+    }
     total_stats = _sum_target_stats(targets)
     logical_source_bytes = total_stats["bytes"]
     return {
@@ -698,9 +708,29 @@ def _sha256_file(path: Path) -> str:
     return digest.hexdigest()
 
 
-def _configuration_source_witnesses(
-    sources: dict[Path, bytes | None], included: list[dict[str, Any]], members: list[dict[str, Any]],
-    archive_members: list[tarfile.TarInfo],
+def _source_revisions(
+    sources: dict[Path, bytes | None], included: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Freeze source hashes and their archive routes at the actual read boundary."""
+    result = []
+    for source, raw in sources.items():
+        paths = set()
+        for target in included:
+            path = Path(target["source_path"]).expanduser()
+            for root in {path.absolute(), _resolved(path)}:
+                try:
+                    relative = source.relative_to(root)
+                except ValueError:
+                    continue
+                paths.add(str(PurePosixPath(target["archive_path"]) / relative.as_posix()))
+        result.append({"source_path": str(source), "sha256": hashlib.sha256(raw).hexdigest()
+                       if raw is not None else None, "archive_paths": sorted(paths)})
+    return result
+
+
+def _copied_source_witnesses(
+    revisions: list[dict[str, Any]], members: list[dict[str, Any]],
+    archive_members: list[tarfile.TarInfo], *, source_kind: str,
 ) -> list[dict[str, Any]]:
     """Compare the actual decoded source bytes with every regular tar copy.
 
@@ -719,19 +749,14 @@ def _configuration_source_witnesses(
         else:
             continue
         link_targets.setdefault(member.name, set()).add(linked_path)
-    for source, raw in sources.items():
-        paths = set()
-        for target in included:
-            try:
-                relative = source.relative_to(_resolved(Path(target["source_path"])))
-            except ValueError:
-                continue
-            paths.add(str(PurePosixPath(target["archive_path"]) / relative.as_posix()))
-        if raw is None:
+    for revision in revisions:
+        paths = set(revision["archive_paths"])
+        sha256 = revision["sha256"]
+        if sha256 is None:
             # A retained link is still a captured source, even though it has no
             # separate regular-file byte witness of its own.
             if any(member.name in paths for member in archive_members):
-                raise ValueError("backup configuration source absence differs from copied bytes; retry the backup")
+                raise ValueError(f"backup {source_kind} source absence differs from copied bytes; retry the backup")
             continue
         pending = list(paths)
         while pending:
@@ -740,11 +765,10 @@ def _configuration_source_witnesses(
                     paths.add(target_path)
                     pending.append(target_path)
         witnesses = [row for row in members if row["archive_path"] in paths]
-        sha256 = hashlib.sha256(raw).hexdigest()
         if not witnesses:
-            raise ValueError("backup configuration source has no copied bytes witness; include its regular source file")
+            raise ValueError(f"backup {source_kind} source has no copied bytes witness; include its regular source file")
         if any(row["sha256"] != sha256 for row in witnesses):
-            raise ValueError("backup configuration source differs from copied bytes; retry the backup")
+            raise ValueError(f"backup {source_kind} source differs from copied bytes; retry the backup")
         result.extend(witnesses)
     return result
 
@@ -834,9 +858,19 @@ def execute_state_backup_plan(payload: dict[str, Any]) -> dict[str, Any]:
                 updated["execution"]["configuration_capture"] = {
                     "started_at": configuration_started_at,
                     "completed_at": datetime.now(timezone.utc).isoformat(),
-                    "source_revisions": _configuration_source_witnesses(
-                        source_snapshots, included, tar.file_members, tar.getmembers()),
+                    "source_revisions": _copied_source_witnesses(
+                        _source_revisions(source_snapshots, included), tar.file_members,
+                        tar.getmembers(), source_kind="configuration"),
                 }
+                # New ordinary plans bind the registry reads that selected
+                # documents, runtimes and journals. A later matching config
+                # capture cannot certify a stale discovery plan. Historical
+                # plans and explicit migration backups remain best-effort.
+                discovery_capture = payload.get("registry_discovery", {}).get("source_capture")
+                if discovery_capture is not None:
+                    updated["execution"]["discovery_source_revisions"] = _copied_source_witnesses(
+                        discovery_capture["source_revisions"], tar.file_members,
+                        tar.getmembers(), source_kind="discovery")
             configuration_bytes = json.dumps(configuration, ensure_ascii=False, indent=2).encode("utf-8")
             info = tarfile.TarInfo("configuration-backup.json")
             info.size, info.mode = len(configuration_bytes), 0o600