From 755fd98ef876ac1c70a227e1b1dbdbd76fc7fe17 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 23:39:16 +0800 Subject: [PATCH] fix(backup): reject registry drift after source discovery Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/complete-state-recovery-v0.md | 11 + .../rfcs/complete-state-recovery-v0.zh-CN.md | 8 + docs/reference/configuration-backup.md | 20 ++ loopx/capabilities/configuration_backup.py | 25 +- .../control_plane/projects/registry_codec.py | 20 +- .../runtime/runtime_projection_route.py | 7 +- loopx/state_backup.py | 86 ++++--- .../test_backup_discovery_capture.py | 213 ++++++++++++++++++ 8 files changed, 342 insertions(+), 48 deletions(-) create mode 100644 tests/control_plane/test_backup_discovery_capture.py diff --git a/docs/architecture/rfcs/complete-state-recovery-v0.md b/docs/architecture/rfcs/complete-state-recovery-v0.md index 4240c01cc7..448fb333f5 100644 --- a/docs/architecture/rfcs/complete-state-recovery-v0.md +++ b/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. diff --git a/docs/architecture/rfcs/complete-state-recovery-v0.zh-CN.md b/docs/architecture/rfcs/complete-state-recovery-v0.zh-CN.md index e57e032bcc..92007926d7 100644 --- a/docs/architecture/rfcs/complete-state-recovery-v0.zh-CN.md +++ b/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 来源拒绝捕获，不被标成不存在。自包含的本地链接复用归档内 diff --git a/docs/reference/configuration-backup.md b/docs/reference/configuration-backup.md index 2ecb0e1943..2629cdf136 100644 --- a/docs/reference/configuration-backup.md +++ b/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, diff --git a/loopx/capabilities/configuration_backup.py b/loopx/capabilities/configuration_backup.py index 2391b35a43..54d6d2a295 100644 --- a/loopx/capabilities/configuration_backup.py +++ b/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]}) diff --git a/loopx/control_plane/projects/registry_codec.py b/loopx/control_plane/projects/registry_codec.py index 67ce229ea2..f05952425a 100644 --- a/loopx/control_plane/projects/registry_codec.py +++ b/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()) diff --git a/loopx/control_plane/runtime/runtime_projection_route.py b/loopx/control_plane/runtime/runtime_projection_route.py index 29e1d81721..2e83d3d9d2 100644 --- a/loopx/control_plane/runtime/runtime_projection_route.py +++ b/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" diff --git a/loopx/state_backup.py b/loopx/state_backup.py index 09580c7caf..a927dad2a5 100644 --- a/loopx/state_backup.py +++ b/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 diff --git a/tests/control_plane/test_backup_discovery_capture.py b/tests/control_plane/test_backup_discovery_capture.py new file mode 100644 index 0000000000..57e5563bfe --- /dev/null +++ b/tests/control_plane/test_backup_discovery_capture.py @@ -0,0 +1,213 @@ +"""The registry that selected backup dependencies must agree with copied bytes.""" +import json +import os +import subprocess +import sys +import tarfile + +import pytest + +from loopx.control_plane.effect_runtime import restart_effect_runtime +from loopx.state_backup import build_state_backup_plan, execute_state_backup_plan +from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture + + +@pytest.fixture(autouse=True) +def isolated_runtime(tmp_path, monkeypatch): + isolate_sqlite_runtime(tmp_path, monkeypatch) + yield + restart_effect_runtime() + + +def _declare(registry, name): + data = json.loads(registry.read_text()) + data["goals"][0]["authority_sources"] = [{"kind": "goal_doc", "path": name}] + registry.write_text(json.dumps(data)) + + +def _plan(project, runtime, registry, output): + return build_state_backup_plan(project=project, runtime_root=runtime, registry_path=registry, + output_dir=output, backup_id="capture", include_registry_projects=False, + include_skills=False, include_automations=False) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_moved_required_document_refuses_publication_then_replans_after_source_loss(tmp_path, provider): + registry, runtime, state = promoted_create_fixture(tmp_path, provider=provider) + project = state.parents[3] + (project / "INPUT.md").write_text("Original requirement") + later = project / "LATER.md" + later_bytes = "# New required material\n\n完整验收 🧭\n".encode() + later.write_bytes(later_bytes) + _declare(registry, "INPUT.md") + output = tmp_path / "backups" + frozen = _plan(project, runtime, registry, output) + # Keep the exact JSON plan: private raw registry bytes are not a plan API. + frozen = json.loads(json.dumps(frozen)) + _declare(registry, "LATER.md") + with pytest.raises(ValueError, match="discovery source differs from copied bytes"): + execute_state_backup_plan(frozen) + assert not list(output.iterdir()), "a stale dependency plan must publish neither output" + saved = execute_state_backup_plan(_plan(project, runtime, registry, output)) + route = next(row["archive_path"] for row in saved["included"] if row["source_path"] == str(later)) + for path in (project, runtime): + path.rename(path.with_name("unavailable-" + path.name)) + destination = tmp_path / "inert" + result = subprocess.run([sys.executable, "-m", "loopx.cli", "--format", "json", + "backup-state", "verify", "--archive", saved["archive_path"], "--manifest", saved["manifest_path"], + "--destination", str(destination), "--execute"], capture_output=True, text=True, timeout=60) + assert result.returncode == 0, result.stdout + result.stderr + audit = json.loads(result.stdout) + assert (destination / "data" / route).read_bytes() == later_bytes + assert audit["capture_profile"] == "legacy_best_effort" and audit["status"] == "held" + assert audit["activation_eligible"] is False and audit["execution_authority_granted"] is False + + +def test_discovery_witness_is_from_decoded_bytes_not_a_later_path_read(tmp_path, monkeypatch): + from loopx import state_backup + + project, runtime = tmp_path / "project", tmp_path / "runtime" + registry = project / ".loopx/registry.json" + registry.parent.mkdir(parents=True) + (project / "INPUT.md").write_text("original requirement") + (project / "LATER.md").write_text("later requirement") + registry.write_text(json.dumps({"goals": [{"id": "fixture", "repo": str(project)}]})) + _declare(registry, "INPUT.md") + original_load = state_backup.load_registry + reads = [] + + def change_after_decode(path, **kwargs): + data = original_load(path, **kwargs) + reads.append(path) + _declare(registry, "LATER.md") + return data + + monkeypatch.setattr(state_backup, "load_registry", change_after_decode) + plan = _plan(project, runtime, registry, tmp_path / "backups") + assert reads == [registry] + assert any(row["source_path"] == str(project / "INPUT.md") for row in plan["included"]) + with pytest.raises(ValueError, match="discovery source differs from copied bytes"): + execute_state_backup_plan(plan) + assert not list((tmp_path / "backups").iterdir()) + + +def test_global_source_route_and_declarations_share_the_same_registry_read(tmp_path, monkeypatch): + from loopx.control_plane.projects import registry_codec + + project, runtime, other_runtime = tmp_path / "project", tmp_path / "runtime", tmp_path / "other-runtime" + source = project / ".loopx/registry.json" + source.parent.mkdir(parents=True) + runtime.mkdir() + other_runtime.mkdir() + for name in ("INPUT.md", "LATER.md"): + (project / name).write_text(name) + source.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [ + {"id": "fixture", "repo": str(project)}]})) + _declare(source, "INPUT.md") + global_registry = runtime / "registry.global.json" + global_registry.write_text(json.dumps({"registry_role": "global-local", "goals": [ + {"id": "fixture", "repo": str(project), "source_registry": str(source)}]})) + original_decode = registry_codec.decode_registry_snapshot + decodes = [] + + def change_after_route_read(path, raw): + data = original_decode(path, raw) + if path == source and not decodes: + decodes.append(True) + changed = json.loads(source.read_text()) + changed["common_runtime_root"] = str(other_runtime) + changed["goals"][0]["authority_sources"][0]["path"] = "LATER.md" + source.write_text(json.dumps(changed)) + return data + + monkeypatch.setattr(registry_codec, "decode_registry_snapshot", change_after_route_read) + plan = build_state_backup_plan(project=project, runtime_root=runtime, + output_dir=tmp_path / "backups", include_skills=False, include_automations=False) + assert decodes == [True] + assert any(row["source_path"] == str(project / "INPUT.md") for row in plan["included"]) + assert not any(row["source_path"] == str(other_runtime) for row in plan["included"]) + with pytest.raises(ValueError, match="discovery source differs from copied bytes"): + execute_state_backup_plan(plan) + assert not list((tmp_path / "backups").iterdir()) + + +@pytest.mark.skipif(os.name == "nt", reason="native Windows symlink privilege is not qualified") +def test_frozen_archive_routes_cannot_be_rebound_to_another_registry_alias(tmp_path): + project, runtime = tmp_path / "project", tmp_path / "runtime" + directory = project / ".loopx" + directory.mkdir(parents=True) + first, second, registry = directory / "first.json", directory / "second.json", directory / "registry.json" + for path, name in ((first, "INPUT.md"), (second, "LATER.md")): + (project / name).write_text(name) + path.write_text(json.dumps({"goals": [{"id": "fixture", "repo": str(project), + "authority_sources": [{"kind": "goal_doc", "path": name}]}]})) + registry.symlink_to(first.name) + plan = _plan(project, runtime, registry, tmp_path / "backups") + registry.unlink() + registry.symlink_to(second.name) + # Both regular files are captured independently; a fresh path resolution + # would incorrectly let the unchanged first file witness the new alias. + with pytest.raises(ValueError, match="discovery source differs from copied bytes"): + execute_state_backup_plan(plan) + assert not list((tmp_path / "backups").iterdir()) + + +def test_registry_absence_cannot_be_replaced_by_a_new_definition(tmp_path): + project, runtime = tmp_path / "project", tmp_path / "runtime" + registry = project / ".loopx/registry.json" + registry.parent.mkdir(parents=True) + plan = _plan(project, runtime, registry, tmp_path / "backups") + registry.write_text(json.dumps({"goals": []})) + with pytest.raises(ValueError, match="discovery source absence differs from copied bytes"): + execute_state_backup_plan(plan) + assert not list((tmp_path / "backups").iterdir()) + + +def test_real_cli_reports_changed_discovery_and_preserves_output_directory(tmp_path, monkeypatch, capsys): + from loopx.cli import main + from loopx.cli_commands import support_control_backup + + project, runtime = tmp_path / "project", tmp_path / "runtime" + registry = project / ".loopx/registry.json" + registry.parent.mkdir(parents=True) + registry.write_text(json.dumps({"goals": [{"id": "fixture", "repo": str(project)}]})) + for name in ("INPUT.md", "LATER.md"): + (project / name).write_text(name) + _declare(registry, "INPUT.md") + original_plan = support_control_backup.build_state_backup_plan + + def change_after_planning(**kwargs): + plan = original_plan(**kwargs) + _declare(registry, "LATER.md") + return plan + + monkeypatch.setattr(support_control_backup, "build_state_backup_plan", change_after_planning) + output = tmp_path / "backups" + assert main(["--format", "json", "--runtime-root", str(runtime), "backup-state", + "--project", str(project), "--output-dir", str(output), "--current-project-only", + "--no-skills", "--no-automations", "--execute"]) == 1 + refusal = json.loads(capsys.readouterr().out) + assert refusal["ok"] is False + assert refusal["error"] == "backup discovery source differs from copied bytes; retry the backup" + assert "preserve existing archive" in refusal["recommended_action"] + assert not list(output.iterdir()) + + +@pytest.mark.parametrize("historical_plan", [False, True]) +def test_unchanged_definition_does_not_freeze_unrelated_material_bytes(tmp_path, historical_plan): + project, runtime = tmp_path / "project", tmp_path / "runtime" + registry = project / ".loopx/registry.json" + registry.parent.mkdir(parents=True) + registry.write_text(json.dumps({"goals": [{"id": "fixture", "repo": str(project)}]})) + _declare(registry, "INPUT.md") + document = project / "INPUT.md" + document.write_text("first content") + plan = _plan(project, runtime, registry, tmp_path / "backups") + if historical_plan: + plan["registry_discovery"].pop("source_capture") + document.write_text("content present when copied") + saved = execute_state_backup_plan(plan) + route = next(row["archive_path"] for row in saved["included"] if row["source_path"] == str(document)) + with tarfile.open(saved["archive_path"]) as archive: + assert archive.extractfile(route).read() == b"content present when copied" + assert bool(saved["execution"].get("discovery_source_revisions")) is not historical_plan