"""Recoverable post-commit retirement of one legacy directory tree. The canonical publisher commit and legacy retirement are deliberately separate transactions. This coordinator may run only after the direct stable-path publisher can be verified read-only as IDEMPOTENT_COMMITTED. Its official receipt directory is the durable ownership anchor; directory rename is the only formal legacy mutation. """ from __future__ import annotations import argparse import csv import hashlib import io import json import os import re import sys import uuid from dataclasses import dataclass from pathlib import Path from types import SimpleNamespace from typing import Any from . import publisher as base from .direct_stable_path import _validate_direct_config, _verify_replay SCHEMA = "SHARED_CONTENT_POSTCOMMIT_ARCHIVE_CONFIG_V1" TERMINAL_SCHEMA = "SHARED_CONTENT_POSTCOMMIT_ARCHIVE_TERMINAL_V1" ANCHOR_SCHEMA = "SHARED_CONTENT_POSTCOMMIT_ARCHIVE_ANCHOR_V1" ROLLBACK_SCHEMA = "SHARED_CONTENT_POSTCOMMIT_ARCHIVE_ROLLBACK_CONFIG_V1" ROLLBACK_TERMINAL_SCHEMA = "SHARED_CONTENT_POSTCOMMIT_ARCHIVE_ROLLBACK_TERMINAL_V1" SHA_RE = re.compile(r"^[0-9A-F]{64}$") MAP_HEADER = ( "legacy_path", "canonical_target", "archive_path", "legacy_bytes", "legacy_sha256", "canonical_bytes", "canonical_sha256", "prior_acceptance_mode", "prior_acceptance_reference", "prior_acceptance_audit_id", "migration_gate_audit_id", "expected_final_status", "recovery_basis", ) class ArchiveError(Exception): def __init__(self, code: str, detail: str, exit_code: int = 12): super().__init__(f"{code}:{detail}") self.code = code self.detail = detail self.exit_code = exit_code class InjectedStop(ArchiveError): pass @dataclass(frozen=True) class Row: legacy_path: str canonical_target: str archive_path: str legacy_bytes: int legacy_sha256: str canonical_bytes: int canonical_sha256: str @dataclass(frozen=True) class FormalRow: formal_relative_path: str bytes: int sha256: str @dataclass(frozen=True) class Plan: config: dict[str, Any] config_path: Path config_data: bytes config_sha256: str root: Path identity: dict[str, str] publisher_config: Path publisher_config_data: bytes source_root: Path target_root: Path receipt_root: Path map_path: Path rows: tuple[Row, ...] source_set_sha256: str canonical_set_sha256: str publisher_terminal: dict[str, Any] publisher_formal_rows: tuple[FormalRow, ...] publisher_formal_set_sha256: str publisher_lock_path: Path publisher_ledger_path: Path publisher_attempt_receipt_root: Path publisher_value: dict[str, Any] test_fault: str def _read(path: Path) -> bytes: try: return base._read_bytes(path) except base.PublisherError as exc: raise ArchiveError(exc.code, exc.detail, exc.exit_code) from exc def _sha(data: bytes) -> str: return hashlib.sha256(data).hexdigest().upper() def _canonical(value: Any) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n" def _need_object(value: Any, label: str) -> dict[str, Any]: if not isinstance(value, dict): raise ArchiveError("CONFIG_TYPE", label) return value def _keys(value: dict[str, Any], exact: set[str], label: str) -> None: if set(value) != exact: raise ArchiveError("CONFIG_KEYS", f"{label}:missing={sorted(exact-set(value))};extra={sorted(set(value)-exact)}") def _string(value: Any, label: str) -> str: if not isinstance(value, str) or not value or value != value.strip(): raise ArchiveError("CONFIG_STRING", label) return value def _sha_value(value: Any, label: str) -> str: text = _string(value, label) if not SHA_RE.fullmatch(text): raise ArchiveError("CONFIG_SHA256", label) return text def _integer(value: Any, label: str, *, positive: bool = False) -> int: if isinstance(value, bool) or not isinstance(value, int) or (positive and value <= 0): raise ArchiveError("CONFIG_INTEGER", label) return value def _safe_relative(value: Any, label: str) -> str: text = _string(value, label).replace("\\", "/") path = Path(text) if path.is_absolute() or not path.parts or any(part in ("", ".", "..") for part in path.parts): raise ArchiveError("CONFIG_PATH", label) return "/".join(path.parts) def _inside(root: Path, relative: str) -> Path: path = root / Path(relative) root_text = os.path.abspath(root).casefold().rstrip("\\/") path_text = os.path.abspath(path).casefold() if path_text == root_text or not path_text.startswith(root_text + os.sep.casefold()): raise ArchiveError("PATH_ESCAPE", relative) return path def _relative(path: Path, root: Path) -> str: return os.path.relpath(path, root).replace("\\", "/") def _identity(path: Path) -> tuple[int, str]: data = _read(path) return len(data), _sha(data) def _ordinary_dir(path: Path, label: str) -> None: try: state = os.lstat(base._long(path)) except OSError as exc: raise ArchiveError("DIRECTORY_UNREADABLE", f"{label}:{exc}") from exc attrs = getattr(state, "st_file_attributes", 0) reparse = bool(attrs & getattr(__import__("stat"), "FILE_ATTRIBUTE_REPARSE_POINT", 0x400)) if not __import__("stat").S_ISDIR(state.st_mode) or reparse: raise ArchiveError("DIRECTORY_NOT_ORDINARY", label) def _ancestor_gate(root: Path, path: Path, label: str, *, include_leaf: bool = True) -> None: root_abs = Path(os.path.abspath(root)) path_abs = Path(os.path.abspath(path)) current = path_abs if include_leaf else path_abs.parent chain: list[Path] = [] while True: chain.append(current) if os.path.normcase(str(current)) == os.path.normcase(str(root_abs)): break if current.parent == current: raise ArchiveError("PATH_OUTSIDE_ROOT", label) current = current.parent for item in reversed(chain): if base._exists(item): _ordinary_dir(item, label) def _walk(root: Path) -> tuple[str, ...]: try: return tuple(sorted(base._walk_files(root), key=str.casefold)) except base.PublisherError as exc: raise ArchiveError(exc.code, exc.detail, exc.exit_code) from exc def _set_digest(rows: tuple[Row, ...], side: str) -> str: values = [] for row in rows: if side == "legacy": values.append({"bytes": row.legacy_bytes, "path": row.legacy_path, "sha256": row.legacy_sha256}) elif side == "archive": values.append({"bytes": row.legacy_bytes, "path": row.archive_path, "sha256": row.legacy_sha256}) else: values.append({"bytes": row.canonical_bytes, "path": row.canonical_target, "sha256": row.canonical_sha256}) return _sha(_canonical(values)) def _load_rows(plan_root: Path, map_path: Path, source_root: Path, target_root: Path) -> tuple[Row, ...]: try: reader = csv.DictReader(io.StringIO(_read(map_path).decode("utf-8"), newline="")) except Exception as exc: raise ArchiveError("MAP_PARSE", str(exc)) from exc if tuple(reader.fieldnames or ()) != MAP_HEADER: raise ArchiveError("MAP_HEADER", str(reader.fieldnames)) rows: list[Row] = [] seen_legacy: set[str] = set() seen_archive: set[str] = set() seen_canonical: set[str] = set() source_prefix = _relative(source_root, plan_root).rstrip("/") + "/" target_prefix = _relative(target_root, plan_root).rstrip("/") + "/" for number, raw in enumerate(reader, start=2): if set(raw) != set(MAP_HEADER) or any(value is None for value in raw.values()): raise ArchiveError("MAP_ROW_SHAPE", str(number)) legacy = _safe_relative(raw["legacy_path"], f"map[{number}].legacy_path") archive = _safe_relative(raw["archive_path"], f"map[{number}].archive_path") canonical = _safe_relative(raw["canonical_target"], f"map[{number}].canonical_target") if not legacy.casefold().startswith(source_prefix.casefold()) or not archive.casefold().startswith(target_prefix.casefold()): raise ArchiveError("MAP_ROOT_BINDING", str(number)) if legacy.casefold() in seen_legacy or archive.casefold() in seen_archive or canonical.casefold() in seen_canonical: raise ArchiveError("MAP_DUPLICATE", str(number)) seen_legacy.add(legacy.casefold()); seen_archive.add(archive.casefold()); seen_canonical.add(canonical.casefold()) try: legacy_bytes = int(raw["legacy_bytes"]); canonical_bytes = int(raw["canonical_bytes"]) except ValueError as exc: raise ArchiveError("MAP_INTEGER", str(number)) from exc legacy_sha = _sha_value(raw["legacy_sha256"], f"map[{number}].legacy_sha256") canonical_sha = _sha_value(raw["canonical_sha256"], f"map[{number}].canonical_sha256") if legacy_bytes <= 0 or canonical_bytes <= 0: raise ArchiveError("MAP_BYTES", str(number)) if raw["expected_final_status"] != "ARCHIVED_AFTER_PUBLISHER_COMMITTED_READBACK": raise ArchiveError("MAP_FINAL_STATUS", str(number)) if not raw["recovery_basis"]: raise ArchiveError("MAP_RECOVERY_BASIS", str(number)) rows.append(Row(legacy, canonical, archive, legacy_bytes, legacy_sha, canonical_bytes, canonical_sha)) if not rows: raise ArchiveError("MAP_EMPTY", "no rows") return tuple(sorted(rows, key=lambda row: row.legacy_path.casefold())) def _verify_tree(plan: Plan, side: str) -> None: root = plan.source_root if side == "legacy" else plan.target_root expected = { _relative(_inside(plan.root, row.legacy_path if side == "legacy" else row.archive_path), root): row for row in plan.rows } actual = _walk(root) if tuple(sorted(expected, key=str.casefold)) != actual: raise ArchiveError("ARCHIVE_EXACT_SET", f"{side}:expected={len(expected)};actual={len(actual)}", 30) for relative, row in expected.items(): path = root / Path(relative) if _identity(path) != (row.legacy_bytes, row.legacy_sha256): raise ArchiveError("ARCHIVE_IDENTITY", f"{side}:{relative}", 30) def _verify_canonical(plan: Plan) -> None: for row in plan.rows: if _identity(_inside(plan.root, row.canonical_target)) != (row.canonical_bytes, row.canonical_sha256): raise ArchiveError("CANONICAL_IDENTITY", row.canonical_target, 30) def _publisher_formal_rows(value: dict[str, Any]) -> tuple[FormalRow, ...]: try: raw_rows = value["expected"]["candidate"]["rows"] except (KeyError, TypeError) as exc: raise ArchiveError("PUBLISHER_FORMAL_ROWS", "missing", 30) from exc if not isinstance(raw_rows, list) or not raw_rows: raise ArchiveError("PUBLISHER_FORMAL_ROWS", "not nonempty list", 30) rows: list[FormalRow] = [] seen: set[str] = set() for number, raw in enumerate(raw_rows, 1): if not isinstance(raw, dict): raise ArchiveError("PUBLISHER_FORMAL_ROW", str(number), 30) formal = _safe_relative(raw.get("formal_relative_path"), f"publisher.rows[{number}].formal_relative_path") folded = formal.casefold() if folded in seen: raise ArchiveError("PUBLISHER_FORMAL_DUPLICATE", formal, 30) seen.add(folded) rows.append(FormalRow( formal, _integer(raw.get("bytes"), f"publisher.rows[{number}].bytes", positive=True), _sha_value(raw.get("sha256"), f"publisher.rows[{number}].sha256"), )) return tuple(sorted(rows, key=lambda row: row.formal_relative_path.casefold())) def _publisher_formal_digest(rows: tuple[FormalRow, ...]) -> str: return _sha(_canonical([ {"bytes": row.bytes, "path": row.formal_relative_path, "sha256": row.sha256} for row in rows ])) def _verify_publisher_formal(plan: Plan) -> None: if base._exists(plan.publisher_lock_path): raise ArchiveError("PUBLISHER_LOCK_PRESENT", str(plan.publisher_lock_path), 30) for row in plan.publisher_formal_rows: if _identity(_inside(plan.root, row.formal_relative_path)) != (row.bytes, row.sha256): raise ArchiveError("PUBLISHER_FORMAL_IDENTITY", row.formal_relative_path, 30) if _publisher_formal_digest(plan.publisher_formal_rows) != plan.publisher_formal_set_sha256: raise ArchiveError("PUBLISHER_FORMAL_SET", "digest", 30) terminal = plan.publisher_terminal states = tuple(terminal.get("event_states", ())) processes = terminal.get("event_process_identities") expected = plan.publisher_value["expected"] ledger_expected = expected["ledger"] ledger_data = _read(plan.publisher_ledger_path) seed_bytes = _integer(ledger_expected["bytes"], "publisher.expected.ledger.bytes", positive=True) if len(ledger_data) <= seed_bytes: raise ArchiveError("PUBLISHER_EVENT_CHAIN", "ledger has no attempt rows", 30) ledger_base_data = ledger_data[:seed_bytes] if _sha(ledger_base_data) != _sha_value(ledger_expected["sha256"], "publisher.expected.ledger.sha256"): raise ArchiveError("PUBLISHER_EVENT_CHAIN", "ledger prefix", 30) if not isinstance(processes, list): raise ArchiveError("PUBLISHER_EVENT_CHAIN", "processes", 30) proxy = SimpleNamespace( operation_root=plan.root, paths={ "ledger_path": plan.publisher_ledger_path, "attempt_receipt_root": plan.publisher_attempt_receipt_root, }, ledger_base_data=ledger_base_data, next_event_seq=_integer(ledger_expected["next_event_seq"], "publisher.expected.ledger.next_event_seq", positive=True), next_attempt_seq=_integer(ledger_expected["next_attempt_seq"], "publisher.expected.ledger.next_attempt_seq", positive=True), prior_release_set=plan.publisher_value["expected"]["prior"]["release_set_sha256"], identity=plan.publisher_value["identity"], ) try: digest, start, end = base._validate_attempt_event_chain( proxy, states, tuple(processes), expected_digest=terminal["attempt_event_chain_sha256"], expected_time_start=terminal["event_time_start_utc"], expected_time_end=terminal["event_time_end_utc"], ) except (base.PublisherError, KeyError, TypeError) as exc: raise ArchiveError("PUBLISHER_EVENT_CHAIN", str(exc), 30) from exc if (digest, start, end) != ( terminal["attempt_event_chain_sha256"], terminal["event_time_start_utc"], terminal["event_time_end_utc"], ): raise ArchiveError("PUBLISHER_EVENT_CHAIN", "terminal mismatch", 30) if base._exists(plan.publisher_lock_path): raise ArchiveError("PUBLISHER_LOCK_PRESENT", str(plan.publisher_lock_path), 30) def _fault(plan: Plan, name: str) -> None: if plan.test_fault == name: marker = plan.receipt_root.parent / (plan.receipt_root.name + ".test-fault-" + name) if base._exists(marker): return try: base._create_exclusive(marker, b"consumed\n") except base.PublisherError as exc: raise ArchiveError(exc.code, exc.detail, exc.exit_code) from exc raise InjectedStop("INJECTED_PROCESS_RESTART", name, 20) def _create_exact(path: Path, data: bytes) -> None: if base._exists(path): if _read(path) != data: raise ArchiveError("RECOVERY_REQUIRED_RECEIPT_PARTIAL", _relative(path, path.parent), 30) return try: base._create_exclusive(path, data) except base.PublisherError as exc: raise ArchiveError(exc.code, exc.detail, exc.exit_code) from exc def _anchor_values(plan: Plan) -> tuple[bytes, bytes, bytes]: contract = _canonical({ "archive_id": plan.identity["archive_id"], "attempt_id": plan.identity["attempt_id"], "batch_id": plan.identity["batch_id"], "canonical_set_sha256": plan.canonical_set_sha256, "case_id": plan.identity["case_id"], "config_bytes": len(plan.config_data), "config_sha256": plan.config_sha256, "legacy_map_bytes": len(_read(plan.map_path)), "legacy_map_sha256": _sha(_read(plan.map_path)), "publisher_config_bytes": len(plan.publisher_config_data), "publisher_config_sha256": _sha(plan.publisher_config_data), "publisher_event_chain_sha256": plan.publisher_terminal["attempt_event_chain_sha256"], "publisher_formal_row_count": len(plan.publisher_formal_rows), "publisher_formal_set_sha256": plan.publisher_formal_set_sha256, "publisher_terminal_bytes": _identity(Path(plan.publisher_terminal["_path"]))[0], "publisher_terminal_sha256": _identity(Path(plan.publisher_terminal["_path"]))[1], "row_count": len(plan.rows), "run_id": plan.identity["run_id"], "schema_version": ANCHOR_SCHEMA, "source_root": _relative(plan.source_root, plan.root), "source_set_sha256": plan.source_set_sha256, "target_root": _relative(plan.target_root, plan.root), "task_id": plan.identity["task_id"], }) map_data = _read(plan.map_path) manifest = _canonical({ "members": [ {"bytes": len(contract), "name": "archive_contract.json", "sha256": _sha(contract)}, {"bytes": len(map_data), "name": "legacy_case_path_map.csv", "sha256": _sha(map_data)}, ], "schema_version": ANCHOR_SCHEMA, }) return contract, map_data, manifest def _prepare_anchor(plan: Plan) -> None: if not base._exists(plan.receipt_root): try: os.mkdir(base._long(plan.receipt_root)) except FileExistsError: pass except OSError as exc: raise ArchiveError("RECEIPT_ROOT_CREATE", str(exc)) from exc _ordinary_dir(plan.receipt_root, "receipt_root") _fault(plan, "INTERRUPT_AFTER_RECEIPT_ROOT") contract, map_data, manifest = _anchor_values(plan) _create_exact(plan.receipt_root / "archive_contract.json", contract) _fault(plan, "INTERRUPT_AFTER_CONTRACT") _create_exact(plan.receipt_root / "legacy_case_path_map.csv", map_data) _fault(plan, "INTERRUPT_AFTER_MAP") _create_exact(plan.receipt_root / "anchor_manifest.json", manifest) _fault(plan, "INTERRUPT_AFTER_ANCHOR") def _verify_anchor(plan: Plan, *, allow_terminal: bool) -> None: contract, map_data, manifest = _anchor_values(plan) expected = { "archive_contract.json": contract, "legacy_case_path_map.csv": map_data, "anchor_manifest.json": manifest, } allowed = set(expected) if allow_terminal: allowed |= {"terminal.tmp.json", "terminal.json"} actual = set(_walk(plan.receipt_root)) if not set(expected).issubset(actual) or not actual.issubset(allowed): raise ArchiveError("RECOVERY_REQUIRED_RECEIPT_EXACT_SET", f"actual={sorted(actual)}", 30) for name, data in expected.items(): if _read(plan.receipt_root / name) != data: raise ArchiveError("RECOVERY_REQUIRED_RECEIPT_BINDING", name, 30) def _terminal_value(plan: Plan) -> dict[str, Any]: return { "archive_id": plan.identity["archive_id"], "archive_set_sha256": _set_digest(plan.rows, "archive"), "attempt_id": plan.identity["attempt_id"], "batch_id": plan.identity["batch_id"], "canonical_set_sha256": plan.canonical_set_sha256, "case_id": plan.identity["case_id"], "config_bytes": len(plan.config_data), "config_sha256": plan.config_sha256, "exit_code": 0, "legacy_source_present": False, "publisher_event_chain_sha256": plan.publisher_terminal["attempt_event_chain_sha256"], "publisher_formal_row_count": len(plan.publisher_formal_rows), "publisher_formal_set_sha256": plan.publisher_formal_set_sha256, "publisher_status": "COMMITTED", "row_count": len(plan.rows), "run_id": plan.identity["run_id"], "schema_version": TERMINAL_SCHEMA, "status": "ARCHIVED", "target_present": True, "task_id": plan.identity["task_id"], } def _publish_terminal(plan: Plan) -> dict[str, Any]: expected = _canonical(_terminal_value(plan)) temporary = plan.receipt_root / "terminal.tmp.json" terminal = plan.receipt_root / "terminal.json" if base._exists(terminal): if _read(terminal) != expected or base._exists(temporary): raise ArchiveError("RECOVERY_REQUIRED_TERMINAL_BINDING", "terminal/temp", 30) return _terminal_value(plan) _create_exact(temporary, expected) _fault(plan, "INTERRUPT_AFTER_TERMINAL_TEMP") try: os.rename(base._long(temporary), base._long(terminal)) except FileExistsError: if _read(terminal) != expected: raise ArchiveError("RECOVERY_REQUIRED_TERMINAL_RACE", "drift", 30) except OSError as exc: raise ArchiveError("TERMINAL_ATOMIC_RENAME", str(exc), 30) from exc if _read(terminal) != expected or base._exists(temporary): raise ArchiveError("RECOVERY_REQUIRED_TERMINAL_READBACK", "mismatch", 30) return _terminal_value(plan) def _load_plan(config_path: Path) -> Plan: config_data = _read(config_path) try: config = json.loads(config_data.decode("utf-8"), object_pairs_hook=base._pairs_no_duplicates) except Exception as exc: raise ArchiveError("CONFIG_PARSE", str(exc)) from exc config = _need_object(config, "root") _keys(config, {"schema_version", "identity", "roots", "expected", "test_control"}, "root") if config["schema_version"] != SCHEMA: raise ArchiveError("CONFIG_SCHEMA", str(config["schema_version"])) identity = _need_object(config["identity"], "identity") identity_keys = {"task_id", "case_id", "batch_id", "run_id", "attempt_id", "archive_id", "operator", "archive_gate_audit_id"} _keys(identity, identity_keys, "identity") identity = {key: _string(value, f"identity.{key}") for key, value in identity.items()} roots = _need_object(config["roots"], "roots") root_keys = {"operation_root", "resolved_root", "publisher_config_path", "source_root", "target_root", "receipt_root", "legacy_map_path"} _keys(roots, root_keys, "roots") root = Path(_string(roots["operation_root"], "roots.operation_root")) if not root.is_absolute() or os.path.normcase(os.path.abspath(root)) != os.path.normcase(_string(roots["resolved_root"], "roots.resolved_root")): raise ArchiveError("ROOT_IDENTITY", str(root)) _ordinary_dir(root, "operation_root") resolved = {key: _inside(root, _safe_relative(roots[key], f"roots.{key}")) for key in root_keys - {"operation_root", "resolved_root"}} source_root = resolved["source_root"]; target_root = resolved["target_root"] receipt_root = resolved["receipt_root"]; map_path = resolved["legacy_map_path"] for left, right in ((source_root, target_root), (source_root, receipt_root), (target_root, receipt_root)): a = os.path.abspath(left).casefold().rstrip("\\/"); b = os.path.abspath(right).casefold().rstrip("\\/") if a == b or a.startswith(b + os.sep.casefold()) or b.startswith(a + os.sep.casefold()): raise ArchiveError("PATH_OVERLAP", f"{left}|{right}") _ancestor_gate(root, source_root, "source_root", include_leaf=base._exists(source_root)) _ancestor_gate(root, target_root, "target_root", include_leaf=base._exists(target_root)) _ancestor_gate(root, receipt_root, "receipt_root", include_leaf=base._exists(receipt_root)) if base._volume(source_root.parent).upper() != base._volume(target_root.parent).upper(): raise ArchiveError("CROSS_VOLUME", f"{source_root}|{target_root}") expected = _need_object(config["expected"], "expected") expected_keys = {"publisher_config_bytes", "publisher_config_sha256", "legacy_map_bytes", "legacy_map_sha256", "row_count", "source_set_sha256", "canonical_set_sha256"} _keys(expected, expected_keys, "expected") publisher_config = resolved["publisher_config_path"] publisher_data = _read(publisher_config) if (len(publisher_data), _sha(publisher_data)) != ( _integer(expected["publisher_config_bytes"], "expected.publisher_config_bytes", positive=True), _sha_value(expected["publisher_config_sha256"], "expected.publisher_config_sha256"), ): raise ArchiveError("PUBLISHER_CONFIG_IDENTITY", str(publisher_config)) if _identity(map_path) != ( _integer(expected["legacy_map_bytes"], "expected.legacy_map_bytes", positive=True), _sha_value(expected["legacy_map_sha256"], "expected.legacy_map_sha256"), ): raise ArchiveError("MAP_IDENTITY", str(map_path)) rows = _load_rows(root, map_path, source_root, target_root) if len(rows) != _integer(expected["row_count"], "expected.row_count", positive=True): raise ArchiveError("ROW_COUNT", str(len(rows))) source_digest = _set_digest(rows, "legacy") canonical_digest = _set_digest(rows, "canonical") if source_digest != _sha_value(expected["source_set_sha256"], "expected.source_set_sha256"): raise ArchiveError("SOURCE_SET_DIGEST", source_digest) if canonical_digest != _sha_value(expected["canonical_set_sha256"], "expected.canonical_set_sha256"): raise ArchiveError("CANONICAL_SET_DIGEST", canonical_digest) test_control = _need_object(config["test_control"], "test_control") _keys(test_control, {"environment", "fault"}, "test_control") environment = _string(test_control["environment"], "test_control.environment") fault = _string(test_control["fault"], "test_control.fault") faults = {"NONE", "INTERRUPT_AFTER_RECEIPT_ROOT", "INTERRUPT_AFTER_CONTRACT", "INTERRUPT_AFTER_MAP", "INTERRUPT_AFTER_ANCHOR", "INTERRUPT_BEFORE_RENAME", "INTERRUPT_AFTER_RENAME", "INTERRUPT_AFTER_TERMINAL_TEMP"} if environment not in {"PRODUCTION", "ISOLATED_TEST"} or fault not in faults or (environment == "PRODUCTION" and fault != "NONE"): raise ArchiveError("TEST_CONTROL", f"{environment}:{fault}") try: publisher_config_value = json.loads(publisher_data.decode("utf-8"), object_pairs_hook=base._pairs_no_duplicates) except Exception as exc: raise ArchiveError("PUBLISHER_CONFIG_PARSE", str(exc), 30) from exc if any(publisher_config_value["identity"].get(key) != identity[key] for key in ("task_id", "case_id", "batch_id", "run_id", "attempt_id")): raise ArchiveError("PUBLISHER_IDENTITY_BINDING", "task/case/batch/run/attempt") formal_rows = _publisher_formal_rows(publisher_config_value) formal_set_sha256 = _publisher_formal_digest(formal_rows) publisher_lock_path = _inside( root, _safe_relative(publisher_config_value["roots"]["lock_path"], "publisher.roots.lock_path"), ) publisher_ledger_path = _inside( root, _safe_relative(publisher_config_value["roots"]["ledger_path"], "publisher.roots.ledger_path"), ) publisher_attempt_receipt_root = _inside( root, _safe_relative( publisher_config_value["roots"]["attempt_receipt_root"], "publisher.roots.attempt_receipt_root", ), ) terminal_path = _inside( root, _safe_relative( publisher_config_value["roots"]["attempt_receipt_root"], "publisher.roots.attempt_receipt_root", ), ) / "terminal.json" try: raw_terminal = json.loads(_read(terminal_path).decode("utf-8"), object_pairs_hook=base._pairs_no_duplicates) except Exception as exc: raise ArchiveError("PUBLISHER_TERMINAL_PARSE", str(exc), 30) from exc if ( not isinstance(raw_terminal, dict) or raw_terminal.get("status") != "COMMITTED" or raw_terminal.get("exit_code") != 0 or raw_terminal.get("config_sha256") != _sha(publisher_data) or not SHA_RE.fullmatch(str(raw_terminal.get("attempt_event_chain_sha256", ""))) ): raise ArchiveError("PUBLISHER_TERMINAL_BINDING", "status/config/event-chain", 30) publisher_terminal = dict(raw_terminal); publisher_terminal["_path"] = str(terminal_path) return Plan( config, config_path, config_data, _sha(config_data), root, identity, publisher_config, publisher_data, source_root, target_root, receipt_root, map_path, rows, source_digest, canonical_digest, publisher_terminal, formal_rows, formal_set_sha256, publisher_lock_path, publisher_ledger_path, publisher_attempt_receipt_root, publisher_config_value, fault, ) def run(config_path: Path) -> dict[str, Any]: plan = _load_plan(config_path) rollback_root = plan.receipt_root.parent / "rollback" / plan.identity["archive_id"] if base._exists(rollback_root): raise ArchiveError("RECOVERY_REQUIRED_ARCHIVE_ROLLED_BACK", str(rollback_root), 30) source_exists = base._exists(plan.source_root) target_exists = base._exists(plan.target_root) if source_exists and target_exists: raise ArchiveError("RECOVERY_REQUIRED_SOURCE_AND_TARGET", "both present", 30) if not source_exists and not target_exists: raise ArchiveError("RECOVERY_REQUIRED_SOURCE_AND_TARGET", "both absent", 30) if source_exists: try: direct_plan = _validate_direct_config(plan.publisher_value, plan.publisher_config, plan.publisher_config_data) replay = _verify_replay(direct_plan) except base.PublisherError as exc: raise ArchiveError("PUBLISHER_NOT_COMMITTED", f"{exc.code}:{exc.detail}", 30) from exc if replay.get("status") != "IDEMPOTENT_COMMITTED" or replay.get("exit_code") != 0: raise ArchiveError("PUBLISHER_NOT_COMMITTED", str(replay.get("status")), 30) _verify_publisher_formal(plan) _verify_canonical(plan) if source_exists: _verify_tree(plan, "legacy") else: _verify_tree(plan, "archive") _prepare_anchor(plan) _verify_anchor(plan, allow_terminal=True) if source_exists: _fault(plan, "INTERRUPT_BEFORE_RENAME") try: os.rename(base._long(plan.source_root), base._long(plan.target_root)) except OSError as exc: raise ArchiveError("ARCHIVE_ATOMIC_RENAME", str(exc), 30) from exc _fault(plan, "INTERRUPT_AFTER_RENAME") if base._exists(plan.source_root) or not base._exists(plan.target_root): raise ArchiveError("RECOVERY_REQUIRED_RENAME_TRUTH", "source/target", 30) _verify_tree(plan, "archive") _verify_publisher_formal(plan) _verify_canonical(plan) value = _publish_terminal(plan) _verify_anchor(plan, allow_terminal=True) if set(_walk(plan.receipt_root)) != {"anchor_manifest.json", "archive_contract.json", "legacy_case_path_map.csv", "terminal.json"}: raise ArchiveError("RECOVERY_REQUIRED_FINAL_RECEIPT_SET", "mismatch", 30) if value["status"] == "ARCHIVED" and base._exists(plan.receipt_root / "terminal.json"): value = dict(value) value["status"] = "IDEMPOTENT_ARCHIVED" if not source_exists else "ARCHIVED" return value def _rollback_fault(receipt_root: Path, fault: str, name: str) -> None: if fault != name: return marker = receipt_root.parent / (receipt_root.name + ".test-fault-" + name) if base._exists(marker): return try: base._create_exclusive(marker, b"consumed\n") except base.PublisherError as exc: raise ArchiveError(exc.code, exc.detail, exc.exit_code) from exc raise InjectedStop("INJECTED_ROLLBACK_PROCESS_RESTART", name, 20) def _rollback_config(config_path: Path) -> tuple[Plan, dict[str, Any], bytes, Path, str]: data = _read(config_path) try: value = json.loads(data.decode("utf-8"), object_pairs_hook=base._pairs_no_duplicates) except Exception as exc: raise ArchiveError("ROLLBACK_CONFIG_PARSE", str(exc), 30) from exc value = _need_object(value, "rollback") _keys(value, {"schema_version", "identity", "roots", "binding", "test_control"}, "rollback") if value["schema_version"] != ROLLBACK_SCHEMA: raise ArchiveError("ROLLBACK_CONFIG_SCHEMA", str(value["schema_version"]), 30) identity = _need_object(value["identity"], "rollback.identity") identity_keys = { "task_id", "case_id", "batch_id", "run_id", "attempt_id", "archive_id", "rollback_id", "operator", "rollback_authorization_id", "rollback_audit_id", } _keys(identity, identity_keys, "rollback.identity") identity = {key: _string(item, f"rollback.identity.{key}") for key, item in identity.items()} if not identity["rollback_authorization_id"].startswith("AUTH-") or not identity["rollback_audit_id"].startswith("AUDIT-"): raise ArchiveError("ROLLBACK_AUTHORIZATION_BINDING", "AUTH/AUDIT domain", 30) roots = _need_object(value["roots"], "rollback.roots") _keys(roots, {"operation_root", "resolved_root", "archive_config_path"}, "rollback.roots") root = Path(_string(roots["operation_root"], "rollback.roots.operation_root")) if not root.is_absolute() or os.path.normcase(os.path.abspath(root)) != os.path.normcase(_string(roots["resolved_root"], "rollback.roots.resolved_root")): raise ArchiveError("ROLLBACK_ROOT_IDENTITY", str(root), 30) archive_config = _inside(root, _safe_relative(roots["archive_config_path"], "rollback.roots.archive_config_path")) binding = _need_object(value["binding"], "rollback.binding") _keys(binding, {"archive_config_bytes", "archive_config_sha256", "archive_terminal_bytes", "archive_terminal_sha256"}, "rollback.binding") if _identity(archive_config) != ( _integer(binding["archive_config_bytes"], "rollback.binding.archive_config_bytes", positive=True), _sha_value(binding["archive_config_sha256"], "rollback.binding.archive_config_sha256"), ): raise ArchiveError("ROLLBACK_ARCHIVE_CONFIG_IDENTITY", str(archive_config), 30) plan = _load_plan(archive_config) for key in ("task_id", "case_id", "batch_id", "run_id", "attempt_id", "archive_id"): if identity[key] != plan.identity[key]: raise ArchiveError("ROLLBACK_IDENTITY_BINDING", key, 30) archive_terminal = plan.receipt_root / "terminal.json" if _identity(archive_terminal) != ( _integer(binding["archive_terminal_bytes"], "rollback.binding.archive_terminal_bytes", positive=True), _sha_value(binding["archive_terminal_sha256"], "rollback.binding.archive_terminal_sha256"), ): raise ArchiveError("ROLLBACK_ARCHIVE_TERMINAL_IDENTITY", str(archive_terminal), 30) control = _need_object(value["test_control"], "rollback.test_control") _keys(control, {"environment", "fault"}, "rollback.test_control") environment = _string(control["environment"], "rollback.test_control.environment") fault = _string(control["fault"], "rollback.test_control.fault") faults = {"NONE", "INTERRUPT_AFTER_ROLLBACK_ANCHOR", "INTERRUPT_BEFORE_ROLLBACK_RENAME", "INTERRUPT_AFTER_ROLLBACK_RENAME", "INTERRUPT_AFTER_ROLLBACK_TERMINAL_TEMP"} if environment not in {"PRODUCTION", "ISOLATED_TEST"} or fault not in faults or (environment == "PRODUCTION" and fault != "NONE"): raise ArchiveError("ROLLBACK_TEST_CONTROL", f"{environment}:{fault}", 30) receipt_root = plan.receipt_root.parent / "rollback" / plan.identity["archive_id"] _ancestor_gate(plan.root, receipt_root, "rollback_receipt_root", include_leaf=base._exists(receipt_root)) return plan, identity, data, receipt_root, fault def _rollback_values(plan: Plan, identity: dict[str, str], data: bytes) -> tuple[bytes, bytes]: archive_terminal = _read(plan.receipt_root / "terminal.json") contract = _canonical({ "archive_config_bytes": len(plan.config_data), "archive_config_sha256": plan.config_sha256, "archive_id": identity["archive_id"], "archive_terminal_bytes": len(archive_terminal), "archive_terminal_sha256": _sha(archive_terminal), "authorization_id": identity["rollback_authorization_id"], "publisher_formal_row_count": len(plan.publisher_formal_rows), "publisher_formal_set_sha256": plan.publisher_formal_set_sha256, "rollback_audit_id": identity["rollback_audit_id"], "rollback_config_bytes": len(data), "rollback_config_sha256": _sha(data), "rollback_id": identity["rollback_id"], "schema_version": ROLLBACK_TERMINAL_SCHEMA, "source_root": _relative(plan.source_root, plan.root), "source_set_sha256": plan.source_set_sha256, "target_root": _relative(plan.target_root, plan.root), }) manifest = _canonical({ "members": [ {"bytes": len(contract), "name": "rollback_contract.json", "sha256": _sha(contract)}, {"bytes": len(archive_terminal), "name": "archive_terminal.json", "sha256": _sha(archive_terminal)}, ], "schema_version": ROLLBACK_TERMINAL_SCHEMA, }) return contract, manifest def _verify_rollback_anchor(plan: Plan, identity: dict[str, str], data: bytes, receipt_root: Path, *, allow_terminal: bool) -> None: contract, manifest = _rollback_values(plan, identity, data) expected = { "archive_terminal.json": _read(plan.receipt_root / "terminal.json"), "rollback_contract.json": contract, "rollback_manifest.json": manifest, } allowed = set(expected) | ({"terminal.json", "terminal.tmp.json"} if allow_terminal else set()) actual = set(_walk(receipt_root)) if not set(expected).issubset(actual) or not actual.issubset(allowed): raise ArchiveError("RECOVERY_REQUIRED_ROLLBACK_RECEIPT_SET", str(sorted(actual)), 30) for name, payload in expected.items(): if _read(receipt_root / name) != payload: raise ArchiveError("RECOVERY_REQUIRED_ROLLBACK_RECEIPT_BINDING", name, 30) def _rollback_terminal_value(plan: Plan, identity: dict[str, str], data: bytes) -> dict[str, Any]: return { "archive_id": identity["archive_id"], "authorization_id": identity["rollback_authorization_id"], "exit_code": 0, "legacy_source_present": True, "publisher_formal_set_sha256": plan.publisher_formal_set_sha256, "rollback_config_bytes": len(data), "rollback_config_sha256": _sha(data), "rollback_id": identity["rollback_id"], "schema_version": ROLLBACK_TERMINAL_SCHEMA, "status": "LEGACY_RESTORED", "target_present": False, } def rollback(config_path: Path) -> dict[str, Any]: plan, identity, data, receipt_root, fault = _rollback_config(config_path) source_exists = base._exists(plan.source_root) target_exists = base._exists(plan.target_root) if source_exists == target_exists: raise ArchiveError("RECOVERY_REQUIRED_ROLLBACK_SOURCE_AND_TARGET", f"source={source_exists};target={target_exists}", 30) if not base._exists(receipt_root): try: os.mkdir(base._long(receipt_root)) except OSError as exc: raise ArchiveError("ROLLBACK_RECEIPT_ROOT_CREATE", str(exc), 30) from exc contract, manifest = _rollback_values(plan, identity, data) _create_exact(receipt_root / "rollback_contract.json", contract) _create_exact(receipt_root / "archive_terminal.json", _read(plan.receipt_root / "terminal.json")) _create_exact(receipt_root / "rollback_manifest.json", manifest) _verify_rollback_anchor(plan, identity, data, receipt_root, allow_terminal=True) _rollback_fault(receipt_root, fault, "INTERRUPT_AFTER_ROLLBACK_ANCHOR") if target_exists: _verify_anchor(plan, allow_terminal=True) _verify_tree(plan, "archive") _verify_publisher_formal(plan) _rollback_fault(receipt_root, fault, "INTERRUPT_BEFORE_ROLLBACK_RENAME") try: os.rename(base._long(plan.target_root), base._long(plan.source_root)) except OSError as exc: raise ArchiveError("ROLLBACK_ATOMIC_RENAME", str(exc), 30) from exc _rollback_fault(receipt_root, fault, "INTERRUPT_AFTER_ROLLBACK_RENAME") if not base._exists(plan.source_root) or base._exists(plan.target_root): raise ArchiveError("RECOVERY_REQUIRED_ROLLBACK_RENAME_TRUTH", "source/target", 30) # A restart after the atomic rollback rename must prove the same complete # publisher/archive preimage as the first rollback attempt before it may # create or replay a PASS-like terminal. The archive directory is gone at # this point, so the durable archive anchor, the 648 formal publisher rows, # the 17-column event chain and the canonical 645-row mapping are the # authoritative bridge between the archived and restored tree states. _verify_anchor(plan, allow_terminal=True) _verify_publisher_formal(plan) _verify_canonical(plan) _verify_tree(plan, "legacy") if base._exists(plan.publisher_lock_path): raise ArchiveError("PUBLISHER_LOCK_PRESENT", str(plan.publisher_lock_path), 30) terminal_value = _rollback_terminal_value(plan, identity, data) terminal_data = _canonical(terminal_value) temporary = receipt_root / "terminal.tmp.json" terminal = receipt_root / "terminal.json" if base._exists(terminal): if _read(terminal) != terminal_data or base._exists(temporary): raise ArchiveError("RECOVERY_REQUIRED_ROLLBACK_TERMINAL", "binding", 30) else: _create_exact(temporary, terminal_data) _rollback_fault(receipt_root, fault, "INTERRUPT_AFTER_ROLLBACK_TERMINAL_TEMP") try: os.rename(base._long(temporary), base._long(terminal)) except OSError as exc: raise ArchiveError("ROLLBACK_TERMINAL_RENAME", str(exc), 30) from exc _verify_rollback_anchor(plan, identity, data, receipt_root, allow_terminal=True) if set(_walk(receipt_root)) != {"archive_terminal.json", "rollback_contract.json", "rollback_manifest.json", "terminal.json"}: raise ArchiveError("RECOVERY_REQUIRED_ROLLBACK_FINAL_SET", "mismatch", 30) result = dict(terminal_value) result["status"] = "IDEMPOTENT_LEGACY_RESTORED" if source_exists else "LEGACY_RESTORED" return result def _failure(exc: ArchiveError, config_path: Path) -> dict[str, Any]: return { "config_path": str(config_path), "detail": exc.detail, "error_code": exc.code, "exit_code": exc.exit_code, "schema_version": TERMINAL_SCHEMA, "status": "RECOVERY_REQUIRED" if exc.exit_code == 30 else "FAIL_CLOSED", } def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Recoverable legacy postcommit archive coordinator") group = parser.add_mutually_exclusive_group(required=True) group.add_argument("--config") group.add_argument("--rollback-config") args = parser.parse_args(argv) config_path = Path(args.config or args.rollback_config) try: value = run(config_path) if args.config else rollback(config_path) except ArchiveError as exc: value = _failure(exc, config_path) except Exception as exc: value = _failure(ArchiveError("UNEXPECTED", f"{type(exc).__name__}:{exc}", 30), config_path) sys.stdout.write(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n") return int(value["exit_code"]) if __name__ == "__main__": raise SystemExit(main())