|
| 1 | +"""Canonical archives retain promoted history, not the complete old source. |
| 2 | +
|
| 3 | +This characterizes the existing qualified shadow route. It cannot qualify a |
| 4 | +cold Markdown import or authorize removal of historical source readers. |
| 5 | +""" |
| 6 | +from __future__ import annotations |
| 7 | + |
| 8 | +import json |
| 9 | +import subprocess |
| 10 | +import pytest |
| 11 | + |
| 12 | +from canonical_authority_fixture import isolate_sqlite_runtime |
| 13 | +from shadow_e2e_fixture import REPO, workspace |
| 14 | +from loopx.control_plane.effect_runtime import restart_effect_runtime |
| 15 | +from loopx.rollout_event_log import ROLLOUT_EVENT_SCHEMA_VERSION |
| 16 | + |
| 17 | + |
| 18 | +@pytest.mark.parametrize("provider", ["file", "sqlite"]) |
| 19 | +def test_promoted_archive_scope_and_later_write_recovery(tmp_path, monkeypatch, provider): |
| 20 | + isolate_sqlite_runtime(tmp_path, monkeypatch) |
| 21 | + ws = workspace(tmp_path / "source", bootstrap=False) |
| 22 | + ws.state.write_text( |
| 23 | + "---\ngoal_id: goal-e2e\nhandoff_mode: hard_lease\n---\n\n## Agent Todo\n" |
| 24 | + "- [ ] Continue work\n" |
| 25 | + " <!-- loopx:todo todo_id=todo_active task_class=advancement_task resume_when=todo_done:todo_prior -->\n" |
| 26 | + "- [x] Retained work\n" |
| 27 | + " <!-- loopx:todo todo_id=todo_retained task_class=advancement_task completion_receipt_id=receipt-retained -->\n" |
| 28 | + "\n## Completed Work Archive\n" |
| 29 | + "- [x] Referenced work\n" |
| 30 | + " <!-- loopx:todo todo_id=todo_prior role=agent task_class=advancement_task completion_receipt_id=receipt-prior -->\n" |
| 31 | + "- [x] Unreferenced historical result\n" |
| 32 | + " <!-- loopx:todo todo_id=todo_unrelated role=agent task_class=advancement_task completion_receipt_id=receipt-unrelated -->\n" |
| 33 | + ) |
| 34 | + terminal = { |
| 35 | + "schema_version": "task_lease_v0", "goal_id": ws.goal, |
| 36 | + "todo_id": "todo_retained", "owner": "agent-a", "version": 2, |
| 37 | + "lease_epoch": 1, "status": "released", |
| 38 | + "released_at": "2026-09-01T00:00:00Z", "idempotency_key": "old-retained", |
| 39 | + "future_extension": {"missing_value": None, "flag": False}, |
| 40 | + } |
| 41 | + orphan = {**terminal, "todo_id": "todo_unrelated", "idempotency_key": "old-orphan"} |
| 42 | + directory = ws.runtime / "goals" / ws.goal / "task-leases" |
| 43 | + directory.mkdir(parents=True) |
| 44 | + for record in (terminal, orphan): |
| 45 | + (directory / (record["todo_id"] + ".json")).write_text(json.dumps(record)) |
| 46 | + lease_bytes = {path: path.read_bytes() for path in directory.glob("*.json")} |
| 47 | + rollout = directory.parent / "rollout-event-log.jsonl" |
| 48 | + original_receipt = json.dumps({ |
| 49 | + "schema_version": ROLLOUT_EVENT_SCHEMA_VERSION, "kind": "todo_complete", |
| 50 | + "goal_id": ws.goal, "todo_id": "todo_unrelated", |
| 51 | + "timestamp": "2026-09-01T00:00:00Z", "receipt_id": "original-unrelated-receipt", |
| 52 | + }) + "\n" |
| 53 | + rollout.write_text(original_receipt) |
| 54 | + if provider == "sqlite": |
| 55 | + selected = subprocess.run([ |
| 56 | + "node", "--no-warnings", "--experimental-strip-types", "--experimental-sqlite", |
| 57 | + str(REPO / "loopx/control_plane/coordination/local_authority_provider.ts"), |
| 58 | + "--runtime-root", str(ws.runtime), "--goal-id", ws.goal, "--execute", |
| 59 | + ], cwd=REPO, capture_output=True, text=True, timeout=45, check=True) |
| 60 | + assert json.loads(selected.stdout)["provider"] == "sqlite" |
| 61 | + |
| 62 | + assert ws.cli("coordination-shadow", "bootstrap", "--execute")["bootstrap"]["status"] == "applied" |
| 63 | + cold_bytes = ws.state.read_bytes() |
| 64 | + refused = ws.cli("coordination-shadow", "promote", success=False) |
| 65 | + assert refused["ok"] is False |
| 66 | + assert refused["promotion"]["reason_code"] == "local_authority_shadow_not_qualified" |
| 67 | + assert ws.state.read_bytes() == cold_bytes |
| 68 | + |
| 69 | + # Real warm-profile operations are not evidence of a cold import. |
| 70 | + captured = {ws.add(f"Independent qualification work {index}")["todo_id"] for index in range(3)} |
| 71 | + assert ws.drain(budget_seconds="60")["ok"] is True |
| 72 | + preview = ws.cli("coordination-shadow", "promote") |
| 73 | + assert preview["promotion"]["status"] == "preview_ready" |
| 74 | + reviewed = tmp_path / "reviewed.json" |
| 75 | + reviewed.write_text(json.dumps(preview)) |
| 76 | + source_bytes = ws.state.read_bytes() |
| 77 | + promoted = ws.cli("coordination-shadow", "promote", "--reviewed-plan", str(reviewed), "--execute") |
| 78 | + assert promoted["promotion"]["canonical_authority"] == f"{provider}_v0" |
| 79 | + assert ws.state.read_bytes() == source_bytes |
| 80 | + assert {path: path.read_bytes() for path in lease_bytes} == lease_bytes |
| 81 | + assert rollout.read_text().startswith(original_receipt) |
| 82 | + assert ws.cli("task-lease", "inspect", "--todo-id", "todo_retained")["lease"] == terminal |
| 83 | + |
| 84 | + later = ws.add("Later canonical work") |
| 85 | + ws.state.unlink() # Only the disposable source: canonical recovery must not need it. |
| 86 | + assert restart_effect_runtime()["status"] in {"stopped", "not_running"} |
| 87 | + recovered = ws.cli("coordination-shadow", "recover-promotion", "--reviewed-plan", str(reviewed), "--execute") |
| 88 | + assert recovered["promotion"]["status"] == "replayed" |
| 89 | + current = ws.cli("todo", "list", "--todo-id", later["todo_id"]) |
| 90 | + assert current["todo"]["text"] == "Later canonical work" |
| 91 | + assert current["authority_read"]["legacy_fallback_used"] is False |
| 92 | + assert not ws.state.exists() |
| 93 | + |
| 94 | + archive = tmp_path / "authority.ndjson" |
| 95 | + exported = ws.cli("authority-archive", "export", "--archive", str(archive)) |
| 96 | + assert exported["status"] == "exported" |
| 97 | + assert "original-unrelated-receipt" not in archive.read_text() |
| 98 | + digest = exported["archive"]["archive_sha256"] |
| 99 | + for target in ("file", "sqlite"): |
| 100 | + destination = tmp_path / f"restore-{target}" |
| 101 | + restored = ws.cli("authority-archive", "restore", "--archive", str(archive), |
| 102 | + "--archive-sha256", digest, "--destination", str(destination), "--provider", target, "--execute") |
| 103 | + assert restored["status"] == "restored" |
| 104 | + assert restored["execution_authority_granted"] is False |
| 105 | + assert restored["requires_separate_authority_cutover"] is True |
| 106 | + audit = ws.cli("authority-archive", "audit", "--archive", str(archive), |
| 107 | + "--archive-sha256", digest, "--destination", str(destination)) |
| 108 | + assert audit["audit"]["status"] == "matched" |
| 109 | + |
| 110 | + # Independently reopen the real restored store, including the other provider. |
| 111 | + module = REPO / f"loopx/control_plane/coordination/{target}_authority_store.ts" |
| 112 | + owner = "FileAuthorityStore" if target == "file" else "SqliteAuthorityStore" |
| 113 | + script = ( |
| 114 | + f"import {{{owner}}} from {json.dumps(module.as_uri())};" |
| 115 | + f"const store=new {owner}({json.dumps(str(destination / 'store'))},{json.dumps(ws.goal)});" |
| 116 | + "process.stdout.write(JSON.stringify(await store.loadAuthority()));" |
| 117 | + ) |
| 118 | + process = subprocess.run([ |
| 119 | + "node", "--no-warnings", "--experimental-strip-types", "--experimental-sqlite", |
| 120 | + "--input-type=module", "-e", script, |
| 121 | + ], capture_output=True, text=True, timeout=45, check=True) |
| 122 | + head = json.loads(process.stdout)["head"] |
| 123 | + assert {row["todo_id"] for row in head["todos"]} == { |
| 124 | + "todo_active", "todo_retained", "todo_prior", later["todo_id"], |
| 125 | + } | captured |
| 126 | + prior = next(row for row in head["todos"] if row["todo_id"] == "todo_prior") |
| 127 | + assert prior["archive_state"] == "archive" |
| 128 | + assert prior["completion_receipt_id"] == "receipt-prior" |
| 129 | + assert head["leases"] == [terminal] |
0 commit comments