fix: unblock safe recovery and canonical migration for legacy Goals

docs/reference/configuration-backup.md

@@ -35,6 +35,21 @@ select several. Full `backup-state` captures its global discovery registry by
 default, or the selected project registry with `--current-project-only`. The private
 archive manifest reports configuration verification and presence separately.
 
+For a reviewed cohort on a host with unreachable historical source registries,
+repeat `backup-state --configuration-goal-id example` to select the embedded
+configuration checkpoint. This uses the same strict source-owner capture as
+`configuration-backup export --goal-id`; a missing selected source still aborts.
+Omitting the flag retains all-Goal capture and its failure behavior. Physical
+state discovery remains unchanged: the archive retains reachable state and raw
+registry/history bytes, while `configuration_goal_ids`, `missing` and the
+verified configuration component expose the coverage limit. A selected cohort
+does not prove configuration recovery for omitted Goals.
+
+中文：历史源 registry 失联时，可显式重复 `--configuration-goal-id` 选择已审查的
+配置备份范围；所选源仍须严格校验，默认全量配置捕获不变。完整状态发现不收缩，
+原始 registry、历史和可达状态照常保存，清单明确记录配置范围及缺失项；不能把
+成功归档解释为已恢复未选 Goal 的配置。
+
 With `--current-project-only`, physical discovery also includes the selected
 project's registered `state_file` and `source_registry` routes, including custom
 paths outside conventional Goal directories. Relative routes resolve against

docs/reference/local-authority-provider-selection.md

@@ -404,6 +404,12 @@ original evidence and settled lease history. The trusted Python tar adapter
 reads actual archive members, including older backups without a member list;
 the TypeScript owner verifies coverage of the coordination source bytes and
 pins both saved artifacts. A missing source path in the backup refuses import.
+CLI and App retain the full backup and validate every archive member, but
+transport only member digests covering the registered Goal's source files,
+registry, evidence and lease directory. Unrelated backup inventory does not
+inflate the bounded import request; unsafe members and changed complete
+artifacts still reject. This does not reduce imported Todo records or qualify
+whole-Goal reactivation.
 Prepare may select an empty target and persist its identity and immutable plan,
 but does not fence the writer, import state or grant execution authority.
 

docs/reference/protocols/quota-blocked-causal-closeout-v0.md

@@ -95,8 +95,8 @@ closeout. Unsupported PR waits now name this recovery route explicitly.
 
 ### Unavailable Monitor observation
 
-An admitted open Agent `continuous_monitor` on canonical authority may discover
-that its observation cannot be obtained. Close the original Goal/Agent/Todo/Turn
+An admitted open Agent `continuous_monitor` on supported Markdown or canonical
+authority may discover that its observation cannot be obtained. Close the original Goal/Agent/Todo/Turn
 with `refresh-state --delivery-outcome outcome_gap --progress-result-class
 blocked`, a stable blocker ID, evidence IDs and a normal vision decision. Do not
 invent a `monitor-poll` hash, record control metadata as an account observation,
@@ -116,7 +116,7 @@ Exact replay returns the frozen receipt, with no duplicate writeback or debit.
 After closing the original effect, reenter the current recovery host Turn to
 select independently eligible work; do not create a third identity or rebind the
 old one. Missing capability admission or evidence and wrong bindings still fail.
-File and SQLite CLI acceptance covers this recovery, preserved observation
+Markdown, File and SQLite CLI acceptance covers this recovery, preserved observation
 fields, capability restoration, committed-poll rejection and zero debit.
 PostgreSQL and live host adoption remain separate qualifications. Before rollback,
 reconcile these receipts using a runtime that recognizes the new proof.
@@ -196,7 +196,7 @@ excluded/bound 限制仍拒绝。恢复后，模式要求执行 fencing 时获
 
 ### Monitor 观察不可取得
 
-canonical authority 上已准入、开放的 Agent `continuous_monitor`，可能在尝试中
+受支持的 Markdown 或 canonical authority 上已准入、开放的 Agent `continuous_monitor`，可能在尝试中
 发现无法取得观察。用原 Goal/Agent/Todo/Turn 调用 `refresh-state`，提供
 `outcome_gap`、typed `blocked`、稳定 blocker、证据 ID 与正常 vision 决策，沿既有
 计划关闭本次尝试。不得编造 poll hash、把控制元数据记成账户观察或扣额。

loopx/cli_commands/coordination_shadow.py

@@ -264,7 +264,11 @@ def handle_coordination_shadow_command(
                     include_all_archived_todos=True)
                 request.update(action="prepare", projection=projection, source_snapshot=snapshot,
                     target_handoff_mode=args.target_handoff_mode, target_provider=args.provider,
-                    source_backup=read_cold_source_backup(args.backup_manifest))
+                    source_backup=read_cold_source_backup(args.backup_manifest, source_paths=[
+                        Path(snapshot["state_path"]), Path(snapshot["registry_source"]["path"]),
+                        *(Path(item["path"]) for item in snapshot["evidence_files"]),
+                        runtime_root / "goals" / args.goal_id / "task-leases",
+                    ]))
             else:
                 request.update(action="apply" if args.coordination_shadow_command == "apply-import" else "recover",
                     expected_plan_sha256=args.plan_sha256)

loopx/cli_commands/support_control_backup.py

@@ -62,6 +62,12 @@ def register_backup_state_command(
         action="store_true",
         help="Do not discover additional project state from the global registry.",
     )
+    parser.add_argument(
+        "--configuration-goal-id",
+        action="append",
+        help=("Select source-owned Goal configurations for the embedded checkpoint; "
+              "repeat for a reviewed cohort. Raw state discovery is unchanged."),
+    )
     parser.add_argument(
         "--execute",
         action="store_true",
@@ -91,6 +97,7 @@ def handle_backup_state_command(
             include_skills=not bool(args.no_skills),
             include_registry_projects=not bool(args.current_project_only),
             registry_path=(Path(args.project).expanduser() / ".loopx/registry.json") if args.current_project_only else None,
+            configuration_goal_ids=getattr(args, "configuration_goal_id", None),
         )
         if args.execute:
             payload = execute_state_backup_plan(payload)

loopx/control_plane/coordination/cold_source_backup.py

@@ -20,7 +20,9 @@ def _sha(handle: BinaryIO) -> str:
     return digest.hexdigest()
 
 
-def read_cold_source_backup(manifest_path: Path) -> dict[str, Any]:
+def read_cold_source_backup(
+    manifest_path: Path, *, source_paths: list[Path] | None = None,
+) -> dict[str, Any]:
     """Witness the existing backup format, including pre-member-list archives.
 
     A checksum of a manifest alone is insufficient. Read the same open archive,
@@ -38,6 +40,21 @@ def read_cold_source_backup(manifest_path: Path) -> dict[str, Any]:
     execution = manifest.get("execution")
     if not isinstance(execution, dict):
         raise ValueError("cold_import_backup_not_executed")
+    # Transport only the requested Host source coverage. Still validate/hash
+    # every actual member and both complete artifacts; TS owns whether this
+    # subset proves all required source bytes. No backup bytes are removed.
+    selected: list[str] | None = None
+    if source_paths is not None:
+        selected = []
+        for source in source_paths:
+            source = source.expanduser().absolute()
+            for item in manifest["included"]:
+                try:
+                    suffix = source.relative_to(Path(item["source_path"]))
+                except ValueError:
+                    continue
+                prefix = PurePosixPath(item["archive_path"])
+                selected.append(str(prefix / suffix.as_posix()) if suffix.parts else str(prefix))
     archive_path = Path(execution["archive_path"]).expanduser().absolute()
     if not stat.S_ISREG(archive_path.lstat().st_mode):
         raise ValueError("cold_import_backup_archive_unsafe")
@@ -71,7 +88,9 @@ def read_cold_source_backup(manifest_path: Path) -> dict[str, Any]:
                         raise ValueError("cold_import_backup_member_unreadable")
                     with stream:
                         digest = _sha(stream)
-                    members.append({"archive_path": name, "sha256": digest})
+                    if selected is None or any(name == prefix or name.startswith(prefix + "/")
+                                               for prefix in selected):
+                        members.append({"archive_path": name, "sha256": digest})
             if not isinstance(embedded, dict) or any(embedded.get(key) != manifest.get(key)
                 for key in ("schema_version", "runtime_root", "project", "included", "configuration_source_registry")):
                 raise ValueError("cold_import_backup_source_map_changed")

loopx/control_plane/quota/blocked_wait.ts

@@ -124,10 +124,13 @@ export function prepareBlockedWait(value: unknown): JsonObject {
   if (observed === null) reject("has an invalid timestamp");
   if (todo.task_class === "continuous_monitor") {
     if (todo.role !== "agent" || todo.status !== "open" ||
-        (todo.archive_state != null && todo.archive_state !== "active") ||
-        request.allow_turn_settlement_retry !== true) {
-      reject("requires an admitted open Monitor attempt on canonical authority");
+        (todo.archive_state != null && todo.archive_state !== "active")) {
+      reject("requires an admitted open Monitor attempt");
     }
+    // Monitor failure closeout has no retry timer or Todo mutation. The
+    // settlement owner verifies its original admission and evidence for both
+    // supported Markdown and canonical sources. Keep canonical-only implicit
+    // advancement retry scheduling in the separate branch below.
     const fields = ["claimed_by", "last_checked_at", "next_due_at", "result_hash",
       "material_change", "material_change_generation", "cadence", "expires_at", "watch_only"];
     return {schema_version: MONITOR_UNAVAILABLE_SCHEMA, source: "turn_settlement",

loopx/presentation/goal_storage_api.py

@@ -160,7 +160,11 @@ def _storage_import(self, *, action: str) -> None:
                     registry_path=registry_path))
                 request.update(projection=projection, source_snapshot=snapshot,
                     target_provider=body["provider"], target_handoff_mode=body["handoff_mode"],
-                    source_backup=read_cold_source_backup(Path(backup["manifest_path"])))
+                    source_backup=read_cold_source_backup(Path(backup["manifest_path"]), source_paths=[
+                        Path(snapshot["state_path"]), Path(snapshot["registry_source"]["path"]),
+                        *(Path(item["path"]) for item in snapshot["evidence_files"]),
+                        root / "goals" / goal_id / "task-leases",
+                    ]))
             else:
                 request["expected_plan_sha256"] = _token(body, "plan_sha256", 64)
                 if action == "apply":

loopx/semantics/project_registry_io_manifest_v1.json

@@ -919,7 +919,7 @@
     },
     {
       "site": "loopx/cli_commands/support_control_backup.py::<module>.handle_backup_state_command::codec_read:load_registry#1",
-      "line": 84,
+      "line": 90,
       "column": 49,
       "kind": "codec_read",
       "api": "load_registry",

loopx/state_backup.py

@@ -392,6 +392,7 @@ def build_state_backup_plan(
     include_skills: bool = True,
     include_registry_projects: bool = True,
     registry_path: Path | None = None,
+    configuration_goal_ids: list[str] | None = None,
 ) -> dict[str, Any]:
     resolved_project = _resolved(Path(project))
     resolved_runtime_root = _resolved(Path(runtime_root).expanduser() if runtime_root else select_default_runtime_root())
@@ -425,6 +426,7 @@ def build_state_backup_plan(
         "project": str(resolved_project),
         "runtime_root": str(resolved_runtime_root),
         "configuration_source_registry": str(configuration_source_registry),
+        "configuration_goal_ids": configuration_goal_ids,
         "registry_discovery": registry_discovery,
         "codex_home": str(_codex_home()),
         "output_dir": str(resolved_output_dir),
@@ -616,6 +618,7 @@ def execute_state_backup_plan(payload: dict[str, Any]) -> dict[str, Any]:
             configuration = capture_configuration_backup(
                 registry_path=Path(payload["configuration_source_registry"]),
                 runtime_root=Path(payload["runtime_root"]),
+                goal_ids=payload.get("configuration_goal_ids"),
             )
             configuration_bytes = json.dumps(configuration, ensure_ascii=False, indent=2).encode("utf-8")
             info = tarfile.TarInfo("configuration-backup.json")

loopx/state_refresh.py

@@ -1205,6 +1205,13 @@ def refresh_state_run(
             turn_scoped_settlement_qualified
             and workspace_requirement != "not_required"
             and not checkpoint_supplement
+            # The TS settlement owner has qualified only an unavailable
+            # observation, with no delivery, Todo mutation or quota debit.
+            # Recording that failed attempt needs no producing git worktree.
+            and not (
+                isinstance(blocked_retry, dict)
+                and blocked_retry.get("schema_version") == "quota_monitor_unavailable_v0"
+            )
         ):
             delivery_workspace = capture_delivery_workspace(
                 current_path=delivery_workspace_path,

tests/control_plane/test_cold_source_backup_scope.py

@@ -0,0 +1,67 @@
+"""Large unrelated backup inventories do not inflate cold-import transport."""
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+from pathlib import Path
+import tarfile
+
+import pytest
+
+from loopx.control_plane.coordination.cold_source_backup import read_cold_source_backup
+
+
+def backup(tmp_path: Path, unrelated: int = 0, unsafe: bool = False) -> Path:
+    archive = tmp_path / "state.tar.gz"
+    manifest = {
+        "schema_version": "loopx_state_backup_v0", "dry_run": False,
+        "execute_requested": True, "runtime_root": str(tmp_path / "runtime"),
+        "project": str(tmp_path / "project"), "configuration_source_registry": None,
+        "included": [{"source_path": str(tmp_path / "runtime"), "archive_path": "runtime"}],
+    }
+    with tarfile.open(archive, "w:gz") as tar:
+        def add(name: str, data: bytes = b"") -> None:
+            member = tarfile.TarInfo(name)
+            member.size = len(data)
+            tar.addfile(member, io.BytesIO(data))
+        add("manifest.json", json.dumps(manifest).encode())
+        add("runtime/goals/example/state.md", b"retained source")
+        add("runtime/goals/example/task-leases/settled.json", b"retained lease")
+        for index in range(unrelated):
+            # Tar names need no extraction. Each segment remains portable;
+            # many unrelated deep paths reproduce an oversized member proof.
+            add(f"runtime/other/{index}/" + "segment/" * 512 + "file")
+        if unsafe:
+            add("runtime/other/../escape")
+    manifest["execution"] = {
+        "archive_path": str(archive),
+        "archive_sha256": hashlib.sha256(archive.read_bytes()).hexdigest(),
+    }
+    path = tmp_path / "manifest.json"
+    path.write_text(json.dumps(manifest))
+    return path
+
+
+def test_scoped_proof_preserves_complete_artifact_checksums_below_transport_limit(tmp_path):
+    manifest = backup(tmp_path, unrelated=4500)
+    before = manifest.read_bytes()
+    full = read_cold_source_backup(manifest)
+    assert len(json.dumps(full).encode()) > 16 * 1024 * 1024
+    scoped = read_cold_source_backup(manifest, source_paths=[tmp_path / "runtime/goals/example"])
+    assert len(json.dumps(scoped).encode()) < 4096
+    assert scoped == {**full, "members": full["members"][:2]}
+    assert manifest.read_bytes() == before
+    exact = read_cold_source_backup(manifest, source_paths=[tmp_path / "runtime/goals/example/state.md"])
+    assert exact == {**full, "members": full["members"][:1]}
+
+
+def test_scoping_does_not_hide_unsafe_unrelated_members_or_changed_artifact(tmp_path):
+    manifest = backup(tmp_path, unsafe=True)
+    with pytest.raises(ValueError, match="member_unsafe"):
+        read_cold_source_backup(manifest, source_paths=[tmp_path / "runtime/goals/example"])
+    manifest = backup(tmp_path)
+    archive = Path(json.loads(manifest.read_text())["execution"]["archive_path"])
+    archive.write_bytes(archive.read_bytes() + b"changed")
+    with pytest.raises(ValueError, match="archive_changed"):
+        read_cold_source_backup(manifest, source_paths=[tmp_path / "runtime/goals/example"])