test(authority): rehearse source outbox and lease recovery before cold import

tests/control_plane/test_backup_registered_source_retention.py

@@ -3,6 +3,7 @@
 import hashlib
 import os
 import sqlite3
+import shutil
 import subprocess
 import sys
 import tarfile
@@ -12,11 +13,23 @@
 import loopx
 
 from loopx.control_plane.effect_runtime import restart_effect_runtime
+from loopx.control_plane.coordination.local_authority_shadow_adapter import read_local_authority_shadow
 from loopx.state_backup import build_state_backup_plan, execute_state_backup_plan
 from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
 from tests.control_plane.shadow_e2e_fixture import workspace
 
 
+def extract_verified_backup(archive_path, destination):
+    """Check the saved member witness independently before using copied bytes."""
+    with tarfile.open(archive_path) as archive:
+        archive.extractall(destination, filter="data")
+    manifest = json.loads((destination / "manifest.json").read_bytes())
+    for member in manifest["execution"]["file_members"]:
+        data = (destination / member["archive_path"]).read_bytes()
+        assert len(data) == member["size_bytes"]
+        assert hashlib.sha256(data).hexdigest() == member["sha256"]
+
+
 @pytest.fixture(autouse=True)
 def isolated_runtime(tmp_path, monkeypatch):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
@@ -229,13 +242,7 @@ def authority(root):
     assert saved["ok"] is True
     copy = tmp_path / "independent-empty-recovery"
     assert not copy.exists()
-    with tarfile.open(saved["archive_path"]) as archive:
-        archive.extractall(copy, filter="data")
-    manifest = json.loads((copy / "manifest.json").read_bytes())
-    for member in manifest["execution"]["file_members"]:
-        data = (copy / member["archive_path"]).read_bytes()
-        assert len(data) == member["size_bytes"]
-        assert hashlib.sha256(data).hexdigest() == member["sha256"]
+    extract_verified_backup(saved["archive_path"], copy)
     routes = {row["key"]: row["archive_path"] for row in saved["included"]}
     assert (copy / routes["registry_active_state:goal-e2e"]).read_bytes() == post_import_state
     assert (copy / routes["configuration_source_registry"]).read_bytes() == fixture.registry.read_bytes()
@@ -275,3 +282,319 @@ def authority(root):
     assert not selected_directory.exists()
     if provider == "sqlite":
         assert not (restored_runtime / "authority/file-v0").exists()
+
+
+@pytest.mark.parametrize(
+    "partition,policy,window",
+    [
+        ("todos", policy, window)
+        for policy in ("soft_claim", "hard_lease")
+        for window in (
+            "before_replace",
+            "before_marker",
+            "before_commit",
+            "after_commit",
+        )
+    ]
+    + [
+        ("leases", "hard_lease", window) for window in ("before_commit", "after_commit")
+    ],
+)
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_restored_source_outbox_resumes_once_before_cold_import(
+    tmp_path, partition, policy, window, provider
+):
+    """Restore only the owned runtime at its original binding before import.
+
+    Original source paths remain; this does not reactivate a whole Goal.
+    """
+    fixture = workspace(tmp_path / "project", bootstrap=False)
+    registry = fixture.state.parent / ".loopx/registry.json"
+    registry.parent.mkdir()
+    fixture.registry.rename(registry)
+    fixture.registry = registry
+    fixture.state.write_text(
+        fixture.state.read_text().replace(
+            "handoff_mode: hard_lease", f"handoff_mode: {policy}"
+        )
+    )
+    fixture.cli("coordination-shadow", "bootstrap", "--execute")
+    first = fixture.add("Keep the original requirement")["todo_id"]
+    source_before = fixture.state.read_bytes()
+    lease_identity = (
+        "--owner",
+        "agent-a",
+        "--idempotency-key",
+        "original-source-lease",
+    )
+    interrupted = (
+        (
+            "task-lease",
+            "acquire",
+            "--todo-id",
+            first,
+            *lease_identity,
+            "--ttl-seconds",
+            "3600",
+        )
+        if partition == "leases"
+        else (
+            "todo",
+            "add",
+            "--role",
+            "agent",
+            "--text",
+            "Retain the interrupted original operation",
+        )
+    )
+    fixture.crash(window, *interrupted)
+    source_after = fixture.state.read_bytes()
+    source_unchanged = partition == "leases" or window == "before_replace"
+    assert (source_after == source_before) == source_unchanged
+    lease_path = (
+        fixture.runtime / "goals" / fixture.goal / "task-leases" / (first + ".json")
+    )
+    lease_bytes = lease_path.read_bytes() if partition == "leases" else None
+    original_lease = json.loads(lease_bytes) if lease_bytes is not None else None
+    outbox = fixture.runtime / "authority-shadow/outbox" / fixture.goal / partition
+    prepared = next(outbox.glob("*.prepared.json"))
+    operation = json.loads(prepared.read_bytes())["entry_id"]
+    original = {
+        str(path.relative_to(fixture.runtime)): path.read_bytes()
+        for path in fixture.runtime.rglob("*")
+        if path.is_file() and not path.name.endswith(".lock")
+    }
+    registry_bytes = fixture.registry.read_bytes()
+    inventory = fixture.cli("coordination-shadow", "inspect-source")["source_inventory"]
+    assert (
+        inventory["outbox_reconciliation_verified"] is False
+        and inventory["writer_stop_verified"] is False
+    )
+    assert inventory["capture"]["management_state"]["status"] == "active"
+    assert original == {
+        str(path.relative_to(fixture.runtime)): path.read_bytes()
+        for path in fixture.runtime.rglob("*")
+        if path.is_file() and not path.name.endswith(".lock")
+    }
+
+    def backup(name):
+        child = subprocess.run(
+            [
+                *fixture.command,
+                "--registry",
+                str(fixture.registry),
+                "--runtime-root",
+                str(fixture.runtime),
+                "--format",
+                "json",
+                "backup-state",
+                "--project",
+                str(fixture.state.parent),
+                "--output-dir",
+                str(tmp_path / "saved"),
+                "--backup-id",
+                name,
+                "--current-project-only",
+                "--no-skills",
+                "--no-automations",
+                "--execute",
+            ],
+            cwd=fixture.cwd,
+            capture_output=True,
+            text=True,
+            timeout=60,
+        )
+        assert child.returncode == 0, child.stdout + child.stderr
+        return json.loads(child.stdout)
+
+    saved = backup("interrupted-source")
+    copy = tmp_path / "independent-copy"
+    extract_verified_backup(saved["archive_path"], copy)
+    for relative, data in original.items():
+        assert (copy / "runtime-root" / relative).read_bytes() == data
+    routes = {row["key"]: row["archive_path"] for row in saved["included"]}
+    assert (
+        copy / routes["registry_active_state:goal-e2e"]
+    ).read_bytes() == source_after
+    assert (
+        copy / routes["configuration_source_registry"]
+    ).read_bytes() == registry_bytes
+
+    # The killed writer is owned by this fixture. Stop its remaining transport
+    # before replacing only its disposable runtime at the original binding path.
+    restart_effect_runtime()
+    fixture.runtime.rename(tmp_path / "quarantined-original-runtime")
+    shutil.copytree(copy / "runtime-root", fixture.runtime)
+    assert (
+        fixture.state.read_bytes() == source_after
+        and fixture.registry.read_bytes() == registry_bytes
+    )
+    resumed = fixture.drain()
+    assert resumed["ok"] is True and resumed["outcome"] == "drained", resumed
+    assert resumed["pending_after"] == 0 and resumed["prepared_only_after"] == 0
+    assert resumed["no_op"] == (1 if window == "before_replace" else 0)
+    assert not list(outbox.glob("*.prepared.json")) and not list(
+        outbox.glob("*.committed.json")
+    )
+    proof = read_local_authority_shadow(
+        runtime_root=fixture.runtime, goal_id=fixture.goal, scan_limit=100
+    )["proof"]
+    transactions = proof["transactions"]
+    # Bootstrap, original requirement and exactly one interrupted position.
+    assert len(transactions) == 3, transactions
+    assert transactions[-1]["receipts"][0]["entry_id"] == operation
+    assert transactions[-1]["receipts"][0]["no_op"] == (window == "before_replace")
+    head = transactions[-1]["projection"]
+    todos = {row["todo_id"]: row for row in head["todos"]}
+    assert first in todos and len(todos) == (1 if source_unchanged else 2)
+    if partition == "leases":
+        assert head["leases"] == [original_lease]
+        assert lease_path.read_bytes() == lease_bytes
+        recovered_lease = fixture.cli(
+            "task-lease",
+            "acquire",
+            "--todo-id",
+            first,
+            *lease_identity,
+            "--ttl-seconds",
+            "3600",
+        )
+        assert (
+            recovered_lease["idempotent"] and recovered_lease["lease"] == original_lease
+        )
+        assert lease_path.read_bytes() == lease_bytes
+    assert fixture.state.read_bytes() == source_after
+    revision = fixture.cli("coordination-shadow", "inspect")["inspection"][
+        "provider_revision"
+    ]
+    again = fixture.drain()
+    assert again["ok"] and again["outcome"] == "nothing_pending"
+    assert (
+        read_local_authority_shadow(
+            runtime_root=fixture.runtime, goal_id=fixture.goal, scan_limit=100
+        )["proof"]["transactions"]
+        == transactions
+    )
+
+    rollback = fixture.cli(
+        "coordination-shadow", "rollback", "--provider-revision", revision, "--execute"
+    )["rollback"]
+    assert rollback["status"] == "applied"
+    if partition == "leases":
+        # Recovering an outbox neither releases nor renews its execution grant.
+        # A separate native owner settlement is required before cold import.
+        unsettled = backup("unsettled-lease")
+        refused = fixture.cli(
+            "coordination-shadow",
+            "prepare-import",
+            "--operation-id",
+            "unsettled-import",
+            "--backup-manifest",
+            unsettled["manifest_path"],
+            "--provider",
+            provider,
+            "--target-handoff-mode",
+            policy,
+            success=False,
+        )["cold_import"]
+        assert refused["reason_code"] == "cold_import_lease_requires_settlement"
+        assert not list(fixture.runtime.rglob("writer-fence.json"))
+        assert lease_path.read_bytes() == lease_bytes
+        released = fixture.cli(
+            "task-lease",
+            "release",
+            "--todo-id",
+            first,
+            *lease_identity,
+            "--expected-version",
+            str(original_lease["version"]),
+        )["lease"]
+        assert released["status"] == "released"
+    settled = backup("settled-source")
+    plan = fixture.cli(
+        "coordination-shadow",
+        "prepare-import",
+        "--operation-id",
+        "original-reviewed-import",
+        "--backup-manifest",
+        settled["manifest_path"],
+        "--provider",
+        provider,
+        "--target-handoff-mode",
+        policy,
+    )["cold_import"]
+    assert (
+        plan["status"] == "prepared" and plan["complete_goal_backup_verified"] is False
+    )
+    digest = plan["plan_sha256"]
+    imported = fixture.cli(
+        "coordination-shadow",
+        "apply-import",
+        "--operation-id",
+        "original-reviewed-import",
+        "--plan-sha256",
+        digest,
+        "--writers-stopped",
+        "--execute",
+    )["cold_import"]
+    assert (
+        imported["status"] == "applied"
+        and imported["execution_authority_granted"] is False
+    )
+    if partition == "leases":
+        assert (
+            fixture.cli("task-lease", "inspect", "--todo-id", first)["lease"]
+            == released
+        )
+        fresh_lease = fixture.cli(
+            "task-lease",
+            "acquire",
+            "--todo-id",
+            first,
+            "--owner",
+            "agent-a",
+            "--idempotency-key",
+            "fresh-canonical-lease",
+            "--expected-version",
+            str(released["version"]),
+            "--ttl-seconds",
+            "3600",
+        )["lease"]
+        assert fresh_lease["version"] == released["version"] + 1
+        assert fresh_lease["lease_epoch"] == released["lease_epoch"] + 1
+    created = fixture.cli(
+        "todo",
+        "add",
+        "--role",
+        "agent",
+        "--text",
+        "New canonical data after recovered source",
+        "--operation-id",
+        "later-new-data",
+        "--claimed-by",
+        "agent-a",
+    )
+    assert created["ok"]
+    before = fixture.cli("todo", "list")["todos"]
+    recovered = fixture.cli(
+        "coordination-shadow",
+        "recover-import",
+        "--operation-id",
+        "original-reviewed-import",
+        "--plan-sha256",
+        digest,
+        "--execute",
+    )["cold_import"]
+    assert (
+        recovered["status"] == "replayed"
+        and recovered["execution_authority_granted"] is False
+    )
+    if partition == "leases":
+        assert (
+            fixture.cli("task-lease", "inspect", "--todo-id", first)["lease"]
+            == fresh_lease
+        )
+    assert fixture.cli("todo", "list")["todos"] == before
+    assert first in {row["todo_id"] for row in before} and created["todo_id"] in {
+        row["todo_id"] for row in before
+    }