"""DIRECT_STABLE_PATH_SET_V1 transaction for the shared publisher. This module deliberately leaves the accepted release-directory publisher unchanged. It consumes the same strict candidate, prior, history and 17-column ledger contracts, but publishes each declared candidate member to its exact stable destination and switches the Markdown cases current index last. """ from __future__ import annotations import hashlib import ctypes import json import os import shutil import time from dataclasses import dataclass from pathlib import Path from typing import Any, Iterable from . import publisher as base DIRECT_SCHEMA = base.DIRECT_CONFIG_SCHEMA DIRECT_STRATEGY = base.DIRECT_COMMIT_STRATEGY DIRECT_TERMINAL_SCHEMA = "SHARED_CONTENT_PUBLISHER_TERMINAL_V4" DIRECT_PRIOR_SCHEMA = "SHARED_CONTENT_PUBLISHER_DIRECT_PRIOR_STATE_V1" DIRECT_COMMIT_ORDER = "PAYLOAD_THEN_METADATA_THEN_CASE_CURRENT_INDEX_V1" DIRECT_ANCHOR_CONTRACT_SCHEMA = "SHARED_CONTENT_PUBLISHER_RECOVERY_ANCHOR_CONTRACT_V1" DIRECT_ANCHOR_MANIFEST_SCHEMA = "SHARED_CONTENT_PUBLISHER_RECOVERY_ANCHOR_MANIFEST_V1" DIRECT_ANCHOR_MANIFEST = "recovery_anchor_manifest.json" DIRECT_ATTEMPT_CONTRACT = "attempt_contract.json" DIRECT_GENESIS_MODE = "ABSENT_GENESIS_V1" class DirectProcessInterrupted(base.PublisherError): """Isolated-test process boundary that deliberately leaves a valid prefix.""" def _is_absent_genesis(prior: Any) -> bool: """Return whether the strict prior contract represents a first publication.""" return isinstance(prior, dict) and prior.get("mode") == DIRECT_GENESIS_MODE @dataclass(frozen=True) class DestinationState: member_id: str formal_relative_path: str present: bool object_type: str reparse: bool bytes: int | None sha256: str | None def value(self) -> dict[str, Any]: return { "bytes": self.bytes, "formal_relative_path": self.formal_relative_path, "member_id": self.member_id, "object_type": self.object_type, "present": self.present, "reparse": self.reparse, "sha256": self.sha256, } @dataclass class DirectPlan: config: dict[str, Any] config_path: Path config_data: bytes config_sha256: str operation_root: Path identity: dict[str, str] paths: dict[str, Path] candidate_rows: tuple[base.ArtifactRow, ...] prior_rows: tuple[base.FormalRow, ...] history_rows: tuple[tuple[str, int, str], ...] link_checks: tuple[tuple[str, tuple[str, ...]], ...] evidence_checks: tuple[tuple[str, tuple[str, ...]], ...] ledger_base_data: bytes next_event_seq: int next_attempt_seq: int prior_index_data: bytes candidate_index_data: bytes target_release_relative: str target_rows: tuple[base.ArtifactRow, ...] manifest_row: base.ArtifactRow result_row: base.FormalRow candidate_result_row: base.ArtifactRow prior_release_set: str lifecycle: str destination_prior: tuple[DestinationState, ...] destination_paths: dict[str, Path] commit_rows: tuple[base.ArtifactRow, ...] activation_row: base.ArtifactRow prior_state_data: bytes scope_roots: dict[str, Path] prior_scope_sets: dict[str, tuple[str, ...]] def _parse_destination_states(value: Any, where: str) -> tuple[DestinationState, ...]: if not isinstance(value, list) or not value: raise base.PublisherError("CONFIG_TYPE", f"{where} must be nonempty array") expected_keys = { "member_id", "formal_relative_path", "present", "object_type", "reparse", "bytes", "sha256", } rows: list[DestinationState] = [] members: set[str] = set() paths: set[str] = set() for index, item in enumerate(value): if not isinstance(item, dict): raise base.PublisherError("CONFIG_TYPE", f"{where}[{index}]") base._expect_keys(item, expected_keys, f"{where}[{index}]") member = base._need_string(item["member_id"], f"{where}[{index}].member_id") formal = base._safe_relative(item["formal_relative_path"], f"{where}[{index}].formal_relative_path") present = item["present"] reparse = item["reparse"] if not isinstance(present, bool) or not isinstance(reparse, bool): raise base.PublisherError("CONFIG_TYPE", f"{where}[{index}].present/reparse") object_type = base._need_string(item["object_type"], f"{where}[{index}].object_type") size = item["bytes"] digest = item["sha256"] if present: if object_type != "FILE" or reparse: raise base.PublisherError("DESTINATION_PRIOR_TYPE", member) size = base._need_int(size, f"{where}[{index}].bytes", positive=True) digest = base._need_sha(digest, f"{where}[{index}].sha256") elif object_type != "ABSENT" or reparse or size is not None or digest is not None: raise base.PublisherError("DESTINATION_PRIOR_ABSENT", member) folded = formal.casefold() if member in members or folded in paths: raise base.PublisherError("CONFIG_DUPLICATE_ROW", f"{where}[{index}]") members.add(member) paths.add(folded) rows.append(DestinationState(member, formal, present, object_type, reparse, size, digest)) return tuple(rows) def _actual_state(path: Path, member: str, formal: str) -> DestinationState: if not base._exists(path): return DestinationState(member, formal, False, "ABSENT", False, None, None) try: st = base._lstat(path) except OSError as exc: raise base.PublisherError("DESTINATION_UNREADABLE", f"{formal}:{exc}") from exc if base._is_reparse(st): return DestinationState(member, formal, True, "REPARSE", True, None, None) if not os.path.isfile(base._long(path)): return DestinationState(member, formal, True, "OTHER", False, None, None) size, digest = base._identity(path) return DestinationState(member, formal, True, "FILE", False, size, digest) def _state_matches(actual: DestinationState, expected: DestinationState) -> bool: return actual == expected def _candidate_state(row: base.ArtifactRow) -> DestinationState: return DestinationState( row.member_id, row.formal_relative_path, True, "FILE", False, row.bytes, row.sha256, ) def _relative_to(path: Path, root: Path) -> str: return os.path.relpath(path, root).replace("\\", "/") def _under(path: Path, root: Path) -> bool: return os.path.commonpath([base._norm(root), base._norm(path)]) == base._norm(root) def _scope_files(root: Path, ignored: Iterable[Path]) -> tuple[str, ...]: ignored_norm = {base._norm(path) for path in ignored} return tuple(sorted( ( rel for rel in base._walk_files(root) if base._norm(base._inside(root, rel)) not in ignored_norm ), key=str.casefold, )) def _scope_for(plan: DirectPlan, path: Path) -> tuple[str, str]: matches = [(name, root) for name, root in plan.scope_roots.items() if _under(path, root)] if len(matches) != 1: raise base.PublisherError("DESTINATION_SCOPE", str(path)) name, root = matches[0] return name, _relative_to(path, root) def _prior_state_document( config_sha256: str, attempt_id: str, states: tuple[DestinationState, ...], scopes: dict[str, tuple[str, ...]], ) -> bytes: return base._canonical_json({ "attempt_id": attempt_id, "config_sha256": config_sha256, "destination_states": [row.value() for row in states], "scope_file_sets": {key: list(scopes[key]) for key in sorted(scopes)}, "schema_version": DIRECT_PRIOR_SCHEMA, }) def _destination_exact_set(rows: tuple[base.ArtifactRow, ...]) -> str: return base._sha_bytes(base._canonical_json([ { "bytes": row.bytes, "formal_relative_path": row.formal_relative_path, "member_id": row.member_id, "sha256": row.sha256, } for row in sorted(rows, key=lambda item: item.formal_relative_path.casefold()) ], newline=False)) def _commit_plan_data(plan: DirectPlan) -> bytes: return base._canonical_json({ "activation_member_id": plan.activation_row.member_id, "activation_position": len(plan.commit_rows), "attempt_id": plan.identity["attempt_id"], "commit_order": DIRECT_COMMIT_ORDER, "members": [ { "formal_relative_path": row.formal_relative_path, "member_id": row.member_id, "position": index, } for index, row in enumerate(plan.commit_rows, start=1) ], "schema_version": "SHARED_CONTENT_PUBLISHER_DIRECT_COMMIT_PLAN_V1", }) def _anchor_identity_data( identity: dict[str, str], config_sha256: str, receipt_relative: str, ) -> bytes: return base._canonical_json({ "attempt_id": identity["attempt_id"], "batch_id": identity["batch_id"], "case_id": identity["case_id"], "config_sha256": config_sha256, "official_anchor_relative_path": receipt_relative, "run_id": identity["run_id"], "schema_version": DIRECT_ANCHOR_CONTRACT_SCHEMA, "task_id": identity["task_id"], }) def _anchor_prepare_name(identity: dict[str, str], config_sha256: str, receipt_relative: str) -> str: digest = base._sha_bytes(_anchor_identity_data(identity, config_sha256, receipt_relative)) return f".recovery-anchor-{digest[:32]}.preparing" def _prepared_temp_relative(relative: str) -> str: path = Path(relative) digest = base._sha_bytes(relative.encode("utf-8"))[:16] return (path.parent / f".{path.name}.{digest}.preparing.tmp").as_posix() def _atomic_rename_no_replace(source: Path, target: Path) -> None: """Atomically publish one same-volume file/directory without replacement.""" if base._volume(source.parent).upper() != base._volume(target.parent).upper(): raise base.PublisherError("ATOMIC_RENAME_VOLUME", f"{source}->{target}", exit_code=27) if os.name == "nt": kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) move = kernel32.MoveFileExW move.argtypes = (ctypes.c_wchar_p, ctypes.c_wchar_p, ctypes.c_uint32) move.restype = ctypes.c_int # MOVEFILE_WRITE_THROUGH, deliberately without MOVEFILE_REPLACE_EXISTING. if not move(base._long(source), base._long(target), 0x00000008): code = ctypes.get_last_error() raise base.PublisherError( "ATOMIC_RENAME_NO_REPLACE", f"winerror={code}:{source}->{target}", exit_code=20, ) return libc = ctypes.CDLL(None, use_errno=True) renameat2 = getattr(libc, "renameat2", None) if renameat2 is None: raise base.PublisherError("ATOMIC_RENAME_UNSUPPORTED", os.name, exit_code=27) renameat2.argtypes = ( ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint, ) renameat2.restype = ctypes.c_int encoded_source = os.fsencode(os.path.abspath(source)) encoded_target = os.fsencode(os.path.abspath(target)) if renameat2(-100, encoded_source, -100, encoded_target, 1) != 0: # AT_FDCWD / RENAME_NOREPLACE code = ctypes.get_errno() raise base.PublisherError( "ATOMIC_RENAME_NO_REPLACE", f"errno={code}:{os.strerror(code)}:{source}->{target}", exit_code=20, ) @dataclass class _PreAnchorFault: kind: str target: int | None position: int = 0 enabled: bool = True def checkpoint(self, label: str, *, partial: tuple[Path, bytes] | None = None) -> None: self.position += 1 if not self.enabled or self.target != self.position: return if partial is not None: path, data = partial prefix = data[:max(1, len(data) // 2)] base._create_exclusive(path, prefix) detail = f"{self.position}:{label}" if self.kind == "DIRECT_INTERRUPT_PREANCHOR": raise DirectProcessInterrupted("INJECTED_PREANCHOR_RESTART", detail, exit_code=20) if self.kind == "DIRECT_IO_ERROR_PREANCHOR": raise base.PublisherError("INJECTED_PREANCHOR_IO", detail, exit_code=20) def _preanchor_checkpoint_count(plan: DirectPlan) -> int: prepared_member_count = sum(1 for state in plan.destination_prior if state.present) + 4 # Five checkpoints per prepared member and canonical manifest, then # before/during/after the one atomic anchor rename. return (prepared_member_count + 1) * 5 + 3 def _validate_direct_config(config: dict[str, Any], config_path: Path, config_data: bytes) -> DirectPlan: base._expect_keys(config, {"schema_version", "identity", "roots", "expected", "commit", "test_control"}, "root") if config["schema_version"] != DIRECT_SCHEMA: raise base.PublisherError("CONFIG_SCHEMA", str(config["schema_version"])) for name in ("identity", "roots", "expected", "commit", "test_control"): if not isinstance(config[name], dict): raise base.PublisherError("CONFIG_TYPE", name) identity = config["identity"] identity_keys = { "industry", "task_id", "case_id", "batch_id", "run_id", "attempt_id", "canonical_audit_id", "review_handoff_id", "release_id", "prior_release_id", "candidate_release_set_sha256", "operator", } base._expect_keys(identity, identity_keys, "identity") for key in identity_keys: base._need_string(identity[key], f"identity.{key}") if not identity["canonical_audit_id"].startswith("AUDIT-"): raise base.PublisherError("AUDIT_ID_DOMAIN", identity["canonical_audit_id"]) if not identity["review_handoff_id"].startswith("HANDOFF-"): raise base.PublisherError("HANDOFF_ID_DOMAIN", identity["review_handoff_id"]) if identity["canonical_audit_id"] == identity["review_handoff_id"]: raise base.PublisherError("AUDIT_HANDOFF_AMBIGUOUS", "equal identities") base._need_sha(identity["release_id"], "identity.release_id", lower=True) base._need_sha(identity["prior_release_id"], "identity.prior_release_id", lower=True) base._need_sha(identity["candidate_release_set_sha256"], "identity.candidate_release_set_sha256") preimage = "".join( f"{key}={value}\n" for key, value in ( ("industry", identity["industry"]), ("task_id", identity["task_id"]), ("case_id", identity["case_id"]), ("batch_id", identity["batch_id"]), ("run_id", identity["run_id"]), ("accepted_audit_id", identity["canonical_audit_id"]), ) ) if hashlib.sha256(preimage.encode("utf-8")).hexdigest() != identity["release_id"]: raise base.PublisherError("RELEASE_PREIMAGE", "release id mismatch") roots = config["roots"] raw_expected = config["expected"] raw_prior = raw_expected.get("prior") if isinstance(raw_expected, dict) else None absent_genesis = _is_absent_genesis(raw_prior) root_keys = { "operation_root", "resolved_root", "volume_identity", "candidate_root", "case_current_index_path", "result_current_index_path", "ledger_path", "lock_path", "attempt_receipt_root", "history_root", "current_state_relative_path", } base._expect_keys(roots, root_keys, "roots") operation_root = Path(os.path.abspath(base._need_string(roots["operation_root"], "roots.operation_root"))) base._physical_chain(operation_root, final_kind="dir") base._ordinary_dir(operation_root) if base._norm(operation_root) != base._norm(base._need_string(roots["resolved_root"], "roots.resolved_root")): raise base.PublisherError("ROOT_RESOLUTION", str(operation_root)) volume = base._need_string(roots["volume_identity"], "roots.volume_identity").upper() if base._volume(operation_root).upper() != volume: raise base.PublisherError("VOLUME_IDENTITY", volume) if not _under(config_path, operation_root): raise base.PublisherError("CONFIG_OUTSIDE_ROOT", str(config_path)) paths: dict[str, Path] = {} path_names = ( "candidate_root", "case_current_index_path", "result_current_index_path", "ledger_path", "lock_path", "attempt_receipt_root", "history_root", ) for key in path_names: paths[key] = base._inside(operation_root, base._safe_relative(roots[key], f"roots.{key}")) paths["current_state_path"] = base._inside( paths["history_root"], base._safe_relative(roots["current_state_relative_path"], "roots.current_state_relative_path"), ) for key in ("candidate_root", "history_root"): base._validate_parent_chain(operation_root, paths[key], final_kind="dir") base._ordinary_dir(paths[key]) if base._volume(paths[key]).upper() != volume: raise base.PublisherError("VOLUME_DRIFT", key) for key in ("case_current_index_path", "result_current_index_path", "current_state_path"): if absent_genesis: base._validate_parent_chain(operation_root, paths[key]) if base._exists(paths[key]): if key == "current_state_path" or not base._exists(paths["attempt_receipt_root"]): raise base.PublisherError("GENESIS_TARGET_PRESENT", key) base._ordinary_file(paths[key]) else: base._validate_parent_chain(operation_root, paths[key], final_kind="file") base._ordinary_file(paths[key]) base._validate_parent_chain(operation_root, paths["ledger_path"], final_kind="file") base._ordinary_file(paths["ledger_path"]) for key in ("lock_path", "attempt_receipt_root"): base._validate_parent_chain(operation_root, paths[key]) base._validate_parent_chain(operation_root, config_path, final_kind="file") commit = config["commit"] base._expect_keys(commit, {"strategy", "staging_relative_path"}, "commit") if commit["strategy"] != DIRECT_STRATEGY: raise base.PublisherError("COMMIT_STRATEGY", str(commit["strategy"])) stage_rel = base._safe_relative(commit["staging_relative_path"], "commit.staging_relative_path") paths["staging"] = base._inside(operation_root, stage_rel) paths["snapshot"] = paths["attempt_receipt_root"] / "prior_snapshot" receipt_relative = _relative_to(paths["attempt_receipt_root"], operation_root) paths["anchor_prepare"] = paths["attempt_receipt_root"].parent / _anchor_prepare_name( identity, base._sha_bytes(config_data), receipt_relative, ) for key in ("staging", "snapshot", "anchor_prepare"): base._validate_parent_chain(operation_root, paths[key]) graph = {key: paths[key] for key in ( "candidate_root", "case_current_index_path", "result_current_index_path", "ledger_path", "lock_path", "attempt_receipt_root", "history_root", "staging", "snapshot", "anchor_prepare", )} graph["config_path"] = config_path base._validate_path_graph(graph, {frozenset(("attempt_receipt_root", "snapshot"))}) if base._exists(paths["lock_path"]): raise base.PublisherError("LOCK_OCCUPIED", str(paths["lock_path"])) expected = config["expected"] base._expect_keys(expected, { "ledger", "candidate", "prior", "destination_prior", "history", "case_current_index", "result_current_index", "current_state", }, "expected") candidate = expected["candidate"] if not isinstance(candidate, dict): raise base.PublisherError("CONFIG_TYPE", "expected.candidate") candidate_keys = { "manifest_relative_path", "manifest_bytes", "manifest_sha256", "current_manifest_relative_path", "current_manifest_bytes", "current_manifest_sha256", "manifest_self_formal_relative_path", "rows", "link_checks", "evidence_checks", } base._expect_keys(candidate, candidate_keys, "expected.candidate") candidate_rows = base._parse_artifact_rows(candidate["rows"], "expected.candidate.rows") activation = tuple(row for row in candidate_rows if row.commit_role == "CASE_CURRENT_INDEX") if len(activation) != 1 or len(candidate_rows) < 2: raise base.PublisherError("CANDIDATE_COMMIT_ROLES", f"index={len(activation)}") activation_row = activation[0] case_formal = _relative_to(paths["case_current_index_path"], operation_root) if activation_row.formal_relative_path.casefold() != case_formal.casefold(): raise base.PublisherError("CASE_INDEX_FORMAL_PATH", activation_row.formal_relative_path) case_root = paths["case_current_index_path"].parent scope_roots = { "CORE": case_root / "核心文档", "MANIFEST": paths["ledger_path"].parent, "RESULT": paths["result_current_index_path"].parent, } for name, root in scope_roots.items(): base._validate_parent_chain(operation_root, root, final_kind="dir") base._ordinary_dir(root) if base._volume(root).upper() != volume: raise base.PublisherError("VOLUME_DRIFT", f"scope.{name}") for protected_name in ("candidate_root", "attempt_receipt_root", "history_root", "staging", "anchor_prepare"): if base._overlap(root, paths[protected_name]): raise base.PublisherError("SCOPE_PROTECTED_OVERLAP", f"{name}<->{protected_name}") destination_paths: dict[str, Path] = {} for row in candidate_rows: path = base._inside(operation_root, row.formal_relative_path) base._validate_parent_chain(operation_root, path) base._validate_parent_chain(operation_root, path.parent, final_kind="dir") base._ordinary_dir(path.parent) if row is not activation_row and not any(_under(path, root) for root in scope_roots.values()): raise base.PublisherError("DESTINATION_SCOPE", row.formal_relative_path) if row is activation_row and base._norm(path) != base._norm(paths["case_current_index_path"]): raise base.PublisherError("ACTIVATION_PATH", row.formal_relative_path) if base._volume(path.parent).upper() != volume: raise base.PublisherError("DESTINATION_VOLUME", row.formal_relative_path) destination_paths[row.member_id] = path target_items = list(destination_paths.items()) for index, (member, path) in enumerate(target_items): for other_member, other in target_items[index + 1:]: if base._overlap(path, other): raise base.PublisherError("DESTINATION_OVERLAP", f"{member}<->{other_member}") for protected_name in ("candidate_root", "ledger_path", "lock_path", "attempt_receipt_root", "history_root", "staging", "anchor_prepare"): if base._overlap(path, paths[protected_name]): # The declared current manifest may be in the manifest root, but # never at the ledger/lock path itself or beneath it. raise base.PublisherError("DESTINATION_PROTECTED_OVERLAP", f"{member}<->{protected_name}") manifest_rel = base._safe_relative(candidate["manifest_relative_path"], "expected.candidate.manifest_relative_path") manifest_data = base._read_bytes(base._inside(paths["candidate_root"], manifest_rel)) if (len(manifest_data), base._sha_bytes(manifest_data)) != ( base._need_int(candidate["manifest_bytes"], "expected.candidate.manifest_bytes", positive=True), base._need_sha(candidate["manifest_sha256"], "expected.candidate.manifest_sha256"), ): raise base.PublisherError("CANDIDATE_MANIFEST_IDENTITY", manifest_rel) manifest_actual = [ ( row["member_id"], row["candidate_relative_path"].removeprefix("candidate/").replace("\\", "/"), row["formal_relative_path"].replace("\\", "/"), row["artifact_type"], int(row["bytes"]), row["sha256"], ) for row in base._manifest_rows(manifest_data, candidate=True) ] manifest_expected = [ (row.member_id, row.relative_path, row.formal_relative_path, row.artifact_type, row.bytes, row.sha256) for row in candidate_rows ] if manifest_actual != manifest_expected: raise base.PublisherError("CANDIDATE_MANIFEST_BINDING", "ordered rows differ") exact_candidate = {manifest_rel} | {row.relative_path for row in candidate_rows} actual_candidate = set(base._walk_files(paths["candidate_root"])) if actual_candidate != exact_candidate: raise base.PublisherError("CANDIDATE_EXACT_SET", f"missing={sorted(exact_candidate-actual_candidate)};extra={sorted(actual_candidate-exact_candidate)}") for row in candidate_rows: base._verify_identity(base._inside(paths["candidate_root"], row.relative_path), row.bytes, row.sha256, "CANDIDATE_IDENTITY") current_manifest_rel = base._safe_relative(candidate["current_manifest_relative_path"], "expected.candidate.current_manifest_relative_path") manifest_matches = [row for row in candidate_rows if row.relative_path.casefold() == current_manifest_rel.casefold()] if len(manifest_matches) != 1: raise base.PublisherError("CURRENT_MANIFEST_ROW", str(len(manifest_matches))) manifest_row = manifest_matches[0] self_formal = base._safe_relative(candidate["manifest_self_formal_relative_path"], "expected.candidate.manifest_self_formal_relative_path") if self_formal.casefold() != manifest_row.formal_relative_path.casefold(): raise base.PublisherError("CURRENT_MANIFEST_SELF_PATH", f"{self_formal}!={manifest_row.formal_relative_path}") current_manifest_data = base._read_bytes(base._inside(paths["candidate_root"], current_manifest_rel)) if (len(current_manifest_data), base._sha_bytes(current_manifest_data)) != ( base._need_int(candidate["current_manifest_bytes"], "expected.candidate.current_manifest_bytes", positive=True), base._need_sha(candidate["current_manifest_sha256"], "expected.candidate.current_manifest_sha256"), ): raise base.PublisherError("CURRENT_MANIFEST_IDENTITY", current_manifest_rel) current_rows = base._manifest_rows(current_manifest_data, candidate=False) candidate_formal = {base._artifact_formal_tuple(row) for row in candidate_rows} current_set = {base._manifest_tuple(row) for row in current_rows} if not current_set or not current_set.issubset(candidate_formal): raise base.PublisherError("CURRENT_MANIFEST_COVERAGE", "not a nonempty candidate subset") required_manifest_paths = {activation_row.formal_relative_path.casefold(), paths["result_current_index_path"].relative_to(operation_root).as_posix().casefold()} actual_manifest_paths = {row[1].casefold() for row in current_set} core_candidate_paths = { row.formal_relative_path.casefold() for row in candidate_rows if _under(destination_paths[row.member_id], scope_roots["CORE"]) } if not required_manifest_paths.issubset(actual_manifest_paths) or not core_candidate_paths.issubset(actual_manifest_paths): raise base.PublisherError("CURRENT_MANIFEST_ACTIVATED_SET", "case/result/core coverage") candidate_set = base._release_set(current_manifest_data, self_formal) if candidate_set != identity["candidate_release_set_sha256"]: raise base.PublisherError("CANDIDATE_RELEASE_SET", candidate_set) link_checks = base._parse_checks(candidate["link_checks"], "expected.candidate.link_checks") evidence_checks = base._parse_checks(candidate["evidence_checks"], "expected.candidate.evidence_checks") base._verify_checks(paths["candidate_root"], link_checks, "LINK_CHECK_MISSING") base._verify_checks(paths["candidate_root"], evidence_checks, "EVIDENCE_CHECK_MISSING") prior = expected["prior"] if not isinstance(prior, dict): raise base.PublisherError("CONFIG_TYPE", "expected.prior") if absent_genesis: base._expect_keys( prior, {"mode", "release_id", "release_set_sha256", "rows"}, "expected.prior", ) if prior["mode"] != DIRECT_GENESIS_MODE or prior["rows"] != []: raise base.PublisherError("GENESIS_PRIOR_CONTRACT", "mode/rows") prior_rows: tuple[base.FormalRow, ...] = () else: base._expect_keys( prior, { "release_id", "manifest_formal_relative_path", "manifest_bytes", "manifest_sha256", "manifest_self_formal_relative_path", "release_set_sha256", "rows", }, "expected.prior", ) prior_rows = base._parse_formal_rows(prior.get("rows"), "expected.prior.rows") destination_prior = _parse_destination_states(expected["destination_prior"], "expected.destination_prior") if {(row.member_id, row.formal_relative_path.casefold()) for row in destination_prior} != { (row.member_id, row.formal_relative_path.casefold()) for row in candidate_rows }: raise base.PublisherError("DESTINATION_PRIOR_COVERAGE", "candidate set mismatch") prior_by_member = {row.member_id: row for row in destination_prior} if not {row.formal_relative_path.casefold() for row in prior_rows}.issubset( {row.formal_relative_path.casefold() for row in destination_prior} ): raise base.PublisherError("DESTINATION_PRIOR_ACTIVE_COVERAGE", "prior manifest rows") if absent_genesis: if any(state.present for state in destination_prior): raise base.PublisherError("GENESIS_DESTINATION_PRESENT", "destination_prior") prior_release_set = base._need_sha( prior["release_set_sha256"], "expected.prior.release_set_sha256" ) else: prior_manifest_formal = base._safe_relative( prior["manifest_formal_relative_path"], "expected.prior.manifest_formal_relative_path" ) prior_manifest_expected = ( base._need_int(prior["manifest_bytes"], "expected.prior.manifest_bytes", positive=True), base._need_sha(prior["manifest_sha256"], "expected.prior.manifest_sha256"), ) stable_prior_manifest = base._inside(operation_root, prior_manifest_formal) if base._exists(stable_prior_manifest) and base._identity(stable_prior_manifest) == prior_manifest_expected: prior_source_root = operation_root elif base._exists(paths["attempt_receipt_root"]): prior_source_root = paths["snapshot"] / "files" else: raise base.PublisherError("PRIOR_MANIFEST_IDENTITY", prior_manifest_formal) _, prior_release_set = base._verify_prior_manifest(prior_source_root, prior, prior_rows) if prior["release_id"] != identity["prior_release_id"]: raise base.PublisherError("PRIOR_RELEASE_ID", str(prior["release_id"])) result_expected = expected["result_current_index"] if not isinstance(result_expected, dict): raise base.PublisherError("CONFIG_TYPE", "expected.result_current_index") result_keys = ( {"member_id", "formal_relative_path", "artifact_type", "present", "required_utf8_substrings"} if absent_genesis else {"member_id", "formal_relative_path", "artifact_type", "bytes", "sha256", "required_utf8_substrings"} ) base._expect_keys(result_expected, result_keys, "expected.result_current_index") if absent_genesis and result_expected["present"] is not False: raise base.PublisherError("GENESIS_RESULT_CONTRACT", "present") candidate_result_binding = next( ( row for row in candidate_rows if row.formal_relative_path.casefold() == base._safe_relative( result_expected["formal_relative_path"], "expected.result_current_index.formal_relative_path", ).casefold() ), None, ) if absent_genesis and candidate_result_binding is None: raise base.PublisherError("RESULT_INDEX_CANDIDATE_BINDING", "0") result_row = base.FormalRow( base._need_string(result_expected["member_id"], "expected.result_current_index.member_id"), base._safe_relative(result_expected["formal_relative_path"], "expected.result_current_index.formal_relative_path"), base._need_string(result_expected["artifact_type"], "expected.result_current_index.artifact_type"), ( candidate_result_binding.bytes if absent_genesis and candidate_result_binding is not None else base._need_int(result_expected["bytes"], "expected.result_current_index.bytes", positive=True) ), ( candidate_result_binding.sha256 if absent_genesis and candidate_result_binding is not None else base._need_sha(result_expected["sha256"], "expected.result_current_index.sha256") ), ) if result_row.formal_relative_path.casefold() != _relative_to(paths["result_current_index_path"], operation_root).casefold(): raise base.PublisherError("RESULT_INDEX_FORMAL_PATH", result_row.formal_relative_path) candidate_result_matches = [ row for row in candidate_rows if row.formal_relative_path.casefold() == result_row.formal_relative_path.casefold() ] if len(candidate_result_matches) != 1 or candidate_result_matches[0].member_id != result_row.member_id: raise base.PublisherError("RESULT_INDEX_CANDIDATE_BINDING", str(len(candidate_result_matches))) candidate_result_row = candidate_result_matches[0] result_prior = prior_by_member[result_row.member_id] if absent_genesis: if result_prior.present: raise base.PublisherError("GENESIS_RESULT_PRESENT", result_row.member_id) elif not result_prior.present or (result_prior.bytes, result_prior.sha256) != (result_row.bytes, result_row.sha256): raise base.PublisherError("RESULT_INDEX_PRIOR_BINDING", result_row.member_id) case_expected = expected["case_current_index"] if not isinstance(case_expected, dict): raise base.PublisherError("CONFIG_TYPE", "expected.case_current_index") if absent_genesis: base._expect_keys(case_expected, {"present"}, "expected.case_current_index") if case_expected["present"] is not False: raise base.PublisherError("GENESIS_CASE_INDEX_CONTRACT", "present") prior_index_id = (0, base._sha_bytes(b"")) else: base._expect_keys(case_expected, {"bytes", "sha256", "required_utf8_substrings"}, "expected.case_current_index") prior_index_id = ( base._need_int(case_expected["bytes"], "expected.case_current_index.bytes", positive=True), base._need_sha(case_expected["sha256"], "expected.case_current_index.sha256"), ) case_prior = prior_by_member[activation_row.member_id] if absent_genesis: if case_prior.present: raise base.PublisherError("GENESIS_CASE_INDEX_PRESENT", activation_row.member_id) elif not case_prior.present or (case_prior.bytes, case_prior.sha256) != prior_index_id: raise base.PublisherError("CASE_INDEX_PRIOR_BINDING", activation_row.member_id) candidate_index_data = base._read_bytes(base._inside(paths["candidate_root"], activation_row.relative_path)) actual_states = { row.member_id: _actual_state(destination_paths[row.member_id], row.member_id, row.formal_relative_path) for row in candidate_rows } prior_match = all(_state_matches(actual_states[row.member_id], prior_by_member[row.member_id]) for row in candidate_rows) candidate_match = all(_state_matches(actual_states[row.member_id], _candidate_state(row)) for row in candidate_rows) receipt_exists = base._exists(paths["attempt_receipt_root"]) staging_exists = base._exists(paths["staging"]) terminal_names = ( "terminal.json", "recovery_terminal.json", ) rolled_terminal_exists = receipt_exists and any( base._exists(paths["attempt_receipt_root"] / name) for name in terminal_names ) if prior_match and not receipt_exists and not staging_exists: lifecycle = "FRESH" elif candidate_match and receipt_exists: lifecycle = "REPLAY" elif prior_match and receipt_exists: lifecycle = "REPLAY_PRIOR" if rolled_terminal_exists else "RECOVERY_INCOMPLETE" elif receipt_exists: lifecycle = "RECOVERY_MIXED" else: raise base.PublisherError("DESTINATION_STATE", f"prior={prior_match};candidate={candidate_match};receipt={receipt_exists};staging={staging_exists}") prior_index_data = b"" if absent_genesis else ( base._read_bytes(paths["case_current_index_path"]) if prior_match else base._read_bytes(base._inside(paths["snapshot"], "files/" + activation_row.formal_relative_path)) ) if (len(prior_index_data), base._sha_bytes(prior_index_data)) != prior_index_id: raise base.PublisherError("CASE_INDEX_PRIOR_IDENTITY", "mismatch") result_prior_data = b"" if absent_genesis else ( base._read_bytes(paths["result_current_index_path"]) if prior_match else base._read_bytes(base._inside(paths["snapshot"], "files/" + result_row.formal_relative_path)) ) if not absent_genesis and (len(result_prior_data), base._sha_bytes(result_prior_data)) != (result_row.bytes, result_row.sha256): raise base.PublisherError("RESULT_INDEX_PRIOR_IDENTITY", "mismatch") result_validation_data = ( base._read_bytes(base._inside(paths["candidate_root"], candidate_result_row.relative_path)) if absent_genesis else result_prior_data ) try: result_prior_text = result_validation_data.decode("utf-8") except UnicodeDecodeError as exc: raise base.PublisherError("RESULT_INDEX_UTF8", "prior") from exc required_result = result_expected["required_utf8_substrings"] if not isinstance(required_result, list) or not required_result: raise base.PublisherError("CONFIG_TYPE", "expected.result_current_index.required_utf8_substrings") for needle in required_result: if base._need_string(needle, "expected.result_current_index.required_utf8_substrings") not in result_prior_text: raise base.PublisherError("RESULT_INDEX_LINK", needle) try: prior_text = prior_index_data.decode("utf-8") candidate_text = candidate_index_data.decode("utf-8") except UnicodeDecodeError as exc: raise base.PublisherError("CASE_INDEX_UTF8", "prior/candidate") from exc if not absent_genesis: required_case = case_expected["required_utf8_substrings"] if not isinstance(required_case, list) or not required_case: raise base.PublisherError("CONFIG_TYPE", "expected.case_current_index.required_utf8_substrings") for needle in required_case: if base._need_string(needle, "expected.case_current_index.required_utf8_substrings") not in prior_text: raise base.PublisherError("CASE_INDEX_PRIOR_LINK", needle) if not candidate_text.strip(): raise base.PublisherError("CASE_INDEX_CANDIDATE_EMPTY", "empty") history = expected["history"] if not isinstance(history, dict): raise base.PublisherError("CONFIG_TYPE", "expected.history") base._expect_keys(history, {"rows", "minimum_long_paths", "long_path_threshold"}, "expected.history") history_rows = base._parse_history_rows(history["rows"], "expected.history.rows") base._verify_history( paths["history_root"], history_rows, base._need_int(history["long_path_threshold"], "expected.history.long_path_threshold", positive=True), base._need_int(history["minimum_long_paths"], "expected.history.minimum_long_paths"), ) state_expected = expected["current_state"] if not isinstance(state_expected, dict): raise base.PublisherError("CONFIG_TYPE", "expected.current_state") if absent_genesis: base._expect_keys(state_expected, {"present"}, "expected.current_state") if state_expected["present"] is not False or base._exists(paths["current_state_path"]): raise base.PublisherError("GENESIS_CURRENT_STATE_PRESENT", str(paths["current_state_path"])) else: base._expect_keys(state_expected, {"bytes", "sha256", "semantic_field", "semantic_value", "status_field", "status_value", "exit_code_field", "exit_code_value"}, "expected.current_state") base._verify_identity( paths["current_state_path"], base._need_int(state_expected["bytes"], "expected.current_state.bytes", positive=True), base._need_sha(state_expected["sha256"], "expected.current_state.sha256"), "CURRENT_STATE_IDENTITY", ) try: state_value = json.loads(base._read_bytes(paths["current_state_path"]).decode("utf-8")) except Exception as exc: raise base.PublisherError("CURRENT_STATE_PARSE", str(exc)) from exc for field_key, value_key in (("semantic_field", "semantic_value"), ("status_field", "status_value"), ("exit_code_field", "exit_code_value")): field = base._need_string(state_expected[field_key], f"expected.current_state.{field_key}") if state_value.get(field) != state_expected[value_key]: raise base.PublisherError("CURRENT_STATE_SEMANTIC", field) ledger_expected = expected["ledger"] if not isinstance(ledger_expected, dict): raise base.PublisherError("CONFIG_TYPE", "expected.ledger") base._expect_keys(ledger_expected, {"bytes", "sha256", "next_event_seq", "next_attempt_seq", "prior_attempt_id", "prior_terminal_state"}, "expected.ledger") base_size = base._need_int(ledger_expected["bytes"], "expected.ledger.bytes", positive=True) base_sha = base._need_sha(ledger_expected["sha256"], "expected.ledger.sha256") next_event = base._need_int(ledger_expected["next_event_seq"], "expected.ledger.next_event_seq", positive=True) next_attempt = base._need_int(ledger_expected["next_attempt_seq"], "expected.ledger.next_attempt_seq", positive=True) ledger_actual = base._read_bytes(paths["ledger_path"]) if lifecycle == "FRESH": if (len(ledger_actual), base._sha_bytes(ledger_actual)) != (base_size, base_sha): raise base.PublisherError("LEDGER_IDENTITY", str(paths["ledger_path"])) ledger_base = ledger_actual else: if len(ledger_actual) < base_size or base._sha_bytes(ledger_actual[:base_size]) != base_sha: raise base.PublisherError("LEDGER_BASE_PREFIX", "identity") ledger_base = ledger_actual[:base_size] base_rows = base._read_ledger(ledger_base) if base._derive_sequences(base_rows, identity) != (next_event, next_attempt): raise base.PublisherError("LEDGER_SEQUENCE", "derived mismatch") prior_attempt = base._need_string(ledger_expected["prior_attempt_id"], "expected.ledger.prior_attempt_id") prior_terminal = base._need_string(ledger_expected["prior_terminal_state"], "expected.ledger.prior_terminal_state") prior_matches = [row for row in base_rows if row["attempt_id"] == prior_attempt] if not prior_matches or prior_matches[-1]["state"] != prior_terminal: raise base.PublisherError("LEDGER_PRIOR_TERMINAL", prior_attempt) test_control = config["test_control"] base._expect_keys(test_control, {"environment", "fault", "race_marker_relative_path", "fault_commit_after"}, "test_control") environment = base._need_string(test_control["environment"], "test_control.environment") fault = base._need_string(test_control["fault"], "test_control.fault") marker = test_control["race_marker_relative_path"] after = test_control["fault_commit_after"] if not isinstance(marker, str) or (after is not None and (isinstance(after, bool) or not isinstance(after, int) or after <= 0)): raise base.PublisherError("CONFIG_TYPE", "test_control") allowed_faults = { "NONE", "PAUSE_AFTER_LOCK", "DIRECT_FAIL_AFTER_COMMIT", "DIRECT_EXTRA_TARGET_AFTER_COMMIT", "DIRECT_PAUSE_AFTER_COMMIT", "DIRECT_INTERRUPT_AFTER_SNAPSHOT", "DIRECT_INTERRUPT_AFTER_PREPARED", "DIRECT_INTERRUPT_AFTER_VERIFIED", "DIRECT_INTERRUPT_BEFORE_FIRST_DESTINATION", "DIRECT_INTERRUPT_AFTER_COMMIT", "DIRECT_INTERRUPT_PREANCHOR", "DIRECT_IO_ERROR_PREANCHOR", } if environment not in ("PRODUCTION", "ISOLATED_TEST") or fault not in allowed_faults: raise base.PublisherError("CONFIG_ENUM", "test_control") if environment == "PRODUCTION" and (fault != "NONE" or marker or after is not None): raise base.PublisherError("TEST_CONTROL_PRODUCTION", fault) if fault == "PAUSE_AFTER_LOCK" and not marker: raise base.PublisherError("CONFIG_TYPE", "race marker required") commit_faults = { "DIRECT_FAIL_AFTER_COMMIT", "DIRECT_EXTRA_TARGET_AFTER_COMMIT", "DIRECT_PAUSE_AFTER_COMMIT", "DIRECT_INTERRUPT_AFTER_COMMIT", } preanchor_faults = {"DIRECT_INTERRUPT_PREANCHOR", "DIRECT_IO_ERROR_PREANCHOR"} preanchor_count = (sum(1 for state in destination_prior if state.present) + 5) * 5 + 3 if fault in commit_faults and (after is None or after > len(candidate_rows)): raise base.PublisherError("CONFIG_TYPE", "fault_commit_after") if fault in preanchor_faults and (after is None or after > preanchor_count): raise base.PublisherError("CONFIG_TYPE", "preanchor fault boundary") if fault not in commit_faults | preanchor_faults and after is not None: raise base.PublisherError("CONFIG_TYPE", "unexpected fault_commit_after") if fault == "DIRECT_PAUSE_AFTER_COMMIT" and not marker: raise base.PublisherError("CONFIG_TYPE", "pause marker required") if marker: paths["race_marker"] = base._inside(operation_root, base._safe_relative(marker, "test_control.race_marker_relative_path")) commit_rows = tuple(sorted( candidate_rows, key=lambda row: ( 2 if row is activation_row else (0 if _under(destination_paths[row.member_id], scope_roots["CORE"]) else 1), row.formal_relative_path.casefold(), ), )) if commit_rows[-1] is not activation_row: raise base.PublisherError("ACTIVATION_ORDER", "CASE_CURRENT_INDEX not last") ignored = (paths["lock_path"],) actual_scope_sets = {name: _scope_files(root, ignored) for name, root in scope_roots.items()} if lifecycle == "FRESH": prior_scope_sets = actual_scope_sets else: prior_scope_sets = {} for name, values in actual_scope_sets.items(): mutable = set(values) for state in destination_prior: if state.present: continue path = destination_paths[state.member_id] if _under(path, scope_roots[name]): mutable.discard(_relative_to(path, scope_roots[name])) prior_scope_sets[name] = tuple(sorted(mutable, key=str.casefold)) prior_state_data = _prior_state_document( base._sha_bytes(config_data), identity["attempt_id"], destination_prior, prior_scope_sets, ) if lifecycle != "FRESH": state_path = paths["snapshot"] / "direct_prior_state.json" if base._read_bytes(state_path) != prior_state_data: raise base.PublisherError("DIRECT_PRIOR_STATE_BINDING", "snapshot mismatch") plan = DirectPlan( config=config, config_path=config_path, config_data=config_data, config_sha256=base._sha_bytes(config_data), operation_root=operation_root, identity=identity, paths=paths, candidate_rows=candidate_rows, prior_rows=prior_rows, history_rows=history_rows, link_checks=link_checks, evidence_checks=evidence_checks, ledger_base_data=ledger_base, next_event_seq=next_event, next_attempt_seq=next_attempt, prior_index_data=prior_index_data, candidate_index_data=candidate_index_data, target_release_relative="", target_rows=candidate_rows, manifest_row=manifest_row, result_row=result_row, candidate_result_row=candidate_result_row, prior_release_set=prior_release_set, lifecycle=lifecycle, destination_prior=destination_prior, destination_paths=destination_paths, commit_rows=commit_rows, activation_row=activation_row, prior_state_data=prior_state_data, scope_roots=scope_roots, prior_scope_sets=prior_scope_sets, ) _verify_destination_prior(plan, "DESTINATION_PRIOR") if lifecycle == "FRESH" else None return plan def _verify_destination_prior(plan: DirectPlan, code: str) -> None: for expected in plan.destination_prior: actual = _actual_state( plan.destination_paths[expected.member_id], expected.member_id, expected.formal_relative_path, ) if actual != expected: raise base.PublisherError(code, expected.member_id, exit_code=20 if code.startswith("ROLLBACK") else 12) def _verify_scope_sets(plan: DirectPlan, *, candidate: bool, code: str) -> None: ignored = (plan.paths["lock_path"],) for name, root in plan.scope_roots.items(): expected = set(plan.prior_scope_sets[name]) if candidate: for state in plan.destination_prior: if state.present: continue path = plan.destination_paths[state.member_id] if _under(path, root): expected.add(_relative_to(path, root)) actual = set(_scope_files(root, ignored)) if actual != expected: raise base.PublisherError(code, f"{name}:missing={sorted(expected-actual)};extra={sorted(actual-expected)}", exit_code=20) def _verify_candidate_destinations(plan: DirectPlan, code: str) -> None: for row in plan.candidate_rows: base._verify_identity(plan.destination_paths[row.member_id], row.bytes, row.sha256, code) if base._read_bytes(plan.paths["case_current_index_path"]) != plan.candidate_index_data: raise base.PublisherError(code, "activation index", exit_code=20) current_manifest = base._read_bytes(plan.destination_paths[plan.manifest_row.member_id]) candidate = plan.config["expected"]["candidate"] if (len(current_manifest), base._sha_bytes(current_manifest)) != ( candidate["current_manifest_bytes"], candidate["current_manifest_sha256"], ): raise base.PublisherError(code, "current manifest identity", exit_code=20) if base._release_set(current_manifest, plan.manifest_row.formal_relative_path) != plan.identity["candidate_release_set_sha256"]: raise base.PublisherError(code, "release set", exit_code=20) actual_rows = {base._manifest_tuple(row) for row in base._manifest_rows(current_manifest, candidate=False)} candidate_rows = {base._artifact_formal_tuple(row) for row in plan.candidate_rows} if not actual_rows or not actual_rows.issubset(candidate_rows): raise base.PublisherError(code, "manifest subset", exit_code=20) _verify_scope_sets(plan, candidate=True, code=code) def _locked_revalidate(plan: DirectPlan) -> None: if base._read_bytes(plan.config_path) != plan.config_data: raise base.PublisherError("CONFIG_IDENTITY_LOCKED", "changed after preflight") base._physical_chain(plan.operation_root, final_kind="dir") if base._volume(plan.operation_root).upper() != plan.config["roots"]["volume_identity"].upper(): raise base.PublisherError("VOLUME_IDENTITY_LOCKED", "changed after preflight") ledger = base._read_bytes(plan.paths["ledger_path"]) if ledger != plan.ledger_base_data: raise base.PublisherError("LEDGER_IDENTITY_LOCKED", "changed after preflight") if base._derive_sequences(base._read_ledger(ledger), plan.identity) != (plan.next_event_seq, plan.next_attempt_seq): raise base.PublisherError("LEDGER_SEQUENCE_LOCKED", "rederived mismatch") _verify_destination_prior(plan, "DESTINATION_PRIOR_LOCKED") _verify_scope_sets(plan, candidate=False, code="DESTINATION_SCOPE_LOCKED") for row in plan.prior_rows: base._verify_identity(base._inside(plan.operation_root, row.formal_relative_path), row.bytes, row.sha256, "PRIOR_LOCKED") prior = plan.config["expected"]["prior"] if _is_absent_genesis(prior): if any(state.present for state in plan.destination_prior): raise base.PublisherError("GENESIS_DESTINATION_PRESENT_LOCKED", "destination_prior") for key in ("case_current_index_path", "result_current_index_path", "current_state_path"): if base._exists(plan.paths[key]): raise base.PublisherError("GENESIS_TARGET_APPEARED_LOCKED", key) if base._need_sha(prior["release_set_sha256"], "expected.prior.release_set_sha256") != plan.prior_release_set: raise base.PublisherError("PRIOR_RELEASE_SET_LOCKED", "genesis mismatch") else: prior_manifest = base._inside(plan.operation_root, prior["manifest_formal_relative_path"]) base._verify_identity(prior_manifest, prior["manifest_bytes"], prior["manifest_sha256"], "PRIOR_MANIFEST_LOCKED") if base._release_set(base._read_bytes(prior_manifest), prior["manifest_self_formal_relative_path"]) != plan.prior_release_set: raise base.PublisherError("PRIOR_RELEASE_SET_LOCKED", "mismatch") history = plan.config["expected"]["history"] base._verify_history(plan.paths["history_root"], plan.history_rows, history["long_path_threshold"], history["minimum_long_paths"]) candidate = plan.config["expected"]["candidate"] for row in plan.candidate_rows: base._verify_identity(base._inside(plan.paths["candidate_root"], row.relative_path), row.bytes, row.sha256, "CANDIDATE_LOCKED") base._verify_identity( base._inside(plan.paths["candidate_root"], candidate["manifest_relative_path"]), candidate["manifest_bytes"], candidate["manifest_sha256"], "CANDIDATE_MANIFEST_LOCKED", ) if set(base._walk_files(plan.paths["candidate_root"])) != {candidate["manifest_relative_path"]} | {row.relative_path for row in plan.candidate_rows}: raise base.PublisherError("CANDIDATE_EXACT_SET_LOCKED", "mismatch") base._verify_checks(plan.paths["candidate_root"], plan.link_checks, "LINK_CHECK_LOCKED") base._verify_checks(plan.paths["candidate_root"], plan.evidence_checks, "EVIDENCE_CHECK_LOCKED") if base._exists(plan.paths["staging"]) or base._exists(plan.paths["attempt_receipt_root"]): raise base.PublisherError("TARGET_APPEARED_LOCKED", "staging/receipt") def _attempt_contract_data(plan: DirectPlan) -> bytes: return base._canonical_json({ "attempt_id": plan.identity["attempt_id"], "batch_id": plan.identity["batch_id"], "case_id": plan.identity["case_id"], "config_bytes": len(plan.config_data), "config_sha256": plan.config_sha256, "official_anchor_relative_path": _relative_to(plan.paths["attempt_receipt_root"], plan.operation_root), "resolved_root": os.path.abspath(plan.operation_root), "run_id": plan.identity["run_id"], "schema_version": DIRECT_ANCHOR_CONTRACT_SCHEMA, "task_id": plan.identity["task_id"], "volume_identity": base._volume(plan.operation_root).upper(), }) def _snapshot_receipt_data(plan: DirectPlan, created_at: str) -> bytes: return base._canonical_json({ "schema_version": DIRECT_TERMINAL_SCHEMA, "attempt_id": plan.identity["attempt_id"], "config_bytes": len(plan.config_data), "config_sha256": plan.config_sha256, "prior_release_set_sha256": plan.prior_release_set, "destination_count": len(plan.destination_prior), "prior_state_bytes": len(plan.prior_state_data), "prior_state_sha256": base._sha_bytes(plan.prior_state_data), "created_at": created_at, }) def _anchor_fixed_members(plan: DirectPlan) -> list[tuple[str, bytes]]: members: list[tuple[str, bytes]] = [] for state in sorted( (item for item in plan.destination_prior if item.present), key=lambda item: item.formal_relative_path.casefold(), ): members.append(( "prior_snapshot/files/" + state.formal_relative_path, base._read_bytes(plan.destination_paths[state.member_id]), )) members.extend(( ("prior_snapshot/direct_prior_state.json", plan.prior_state_data), ("commit_plan.json", _commit_plan_data(plan)), )) return members def _anchor_manifest_data(plan: DirectPlan, members: list[tuple[str, bytes]]) -> bytes: return base._canonical_json({ "attempt_id": plan.identity["attempt_id"], "batch_id": plan.identity["batch_id"], "case_id": plan.identity["case_id"], "config_sha256": plan.config_sha256, "members": [ {"bytes": len(data), "relative_path": relative, "sha256": base._sha_bytes(data)} for relative, data in members ], "run_id": plan.identity["run_id"], "schema_version": DIRECT_ANCHOR_MANIFEST_SCHEMA, "task_id": plan.identity["task_id"], }) def _prepared_write( plan: DirectPlan, preparation_root: Path, relative: str, data: bytes, fault: _PreAnchorFault, ) -> None: target = base._inside(preparation_root, relative) temp_relative = _prepared_temp_relative(relative) temp = base._inside(preparation_root, temp_relative) fault.checkpoint(f"BEFORE_WRITE:{relative}") fault.checkpoint(f"DURING_WRITE:{relative}", partial=(temp, data)) base._create_exclusive(temp, data) fault.checkpoint(f"AFTER_FLUSH_CLOSE:{relative}") if base._read_bytes(temp) != data: raise base.PublisherError("RECOVERY_PREPARATION_READBACK", relative, exit_code=27) fault.checkpoint(f"AFTER_REREAD:{relative}") _atomic_rename_no_replace(temp, target) fault.checkpoint(f"AFTER_MEMBER_RENAME:{relative}") def _parse_json_object(data: bytes, code: str) -> dict[str, Any]: try: value = json.loads(data.decode("utf-8"), object_pairs_hook=base._pairs_no_duplicates) except base.PublisherError: raise except Exception as exc: raise base.PublisherError(code, str(exc), exit_code=27) from exc if not isinstance(value, dict): raise base.PublisherError(code, "not object", exit_code=27) return value def _validate_snapshot_receipt(plan: DirectPlan, data: bytes) -> None: value = _parse_json_object(data, "RECOVERY_SNAPSHOT_RECEIPT_PARSE") expected_keys = { "schema_version", "attempt_id", "config_bytes", "config_sha256", "prior_release_set_sha256", "destination_count", "prior_state_bytes", "prior_state_sha256", "created_at", } base._expect_keys(value, expected_keys, "prior_snapshot_receipt") expected_values = { "schema_version": DIRECT_TERMINAL_SCHEMA, "attempt_id": plan.identity["attempt_id"], "config_bytes": len(plan.config_data), "config_sha256": plan.config_sha256, "prior_release_set_sha256": plan.prior_release_set, "destination_count": len(plan.destination_prior), "prior_state_bytes": len(plan.prior_state_data), "prior_state_sha256": base._sha_bytes(plan.prior_state_data), } for key, expected in expected_values.items(): if value.get(key) != expected: raise base.PublisherError("RECOVERY_SNAPSHOT_RECEIPT_BINDING", key, exit_code=27) base._parse_strict_utc(value.get("created_at"), "RECOVERY_SNAPSHOT_CREATED_AT") def _verify_snapshot_at(plan: DirectPlan, snapshot_root: Path) -> None: actual = set(base._walk_files(snapshot_root)) expected = _snapshot_expected(plan) if actual != expected: raise base.PublisherError("DIRECT_SNAPSHOT_EXACT_SET", f"missing={sorted(expected-actual)};extra={sorted(actual-expected)}", exit_code=27) if base._read_bytes(snapshot_root / "direct_prior_state.json") != plan.prior_state_data: raise base.PublisherError("DIRECT_SNAPSHOT_STATE", "mismatch", exit_code=27) for state in plan.destination_prior: if state.present: base._verify_identity( base._inside(snapshot_root, "files/" + state.formal_relative_path), int(state.bytes), str(state.sha256), "DIRECT_SNAPSHOT_IDENTITY", ) def _verify_anchor(plan: DirectPlan, root: Path, *, allowed_extras: set[str] | None = None) -> None: base._ordinary_dir(root) fixed_paths = [ "prior_snapshot/files/" + state.formal_relative_path for state in sorted( (item for item in plan.destination_prior if item.present), key=lambda item: item.formal_relative_path.casefold(), ) ] + ["prior_snapshot/direct_prior_state.json", "commit_plan.json"] expected_member_paths = fixed_paths + ["prior_snapshot.json", DIRECT_ATTEMPT_CONTRACT] actual = set(base._walk_files(root)) expected = set(expected_member_paths) | {DIRECT_ANCHOR_MANIFEST} | (allowed_extras or set()) if actual != expected: raise base.PublisherError( "RECOVERY_ANCHOR_EXACT_SET", f"missing={sorted(expected-actual)};extra={sorted(actual-expected)}", exit_code=27, ) expected_members = [ (relative, base._read_bytes(base._inside(root, relative))) for relative in fixed_paths ] if dict(expected_members)["prior_snapshot/direct_prior_state.json"] != plan.prior_state_data: raise base.PublisherError("RECOVERY_ANCHOR_MEMBER_IDENTITY", "direct_prior_state.json", exit_code=27) if dict(expected_members)["commit_plan.json"] != _commit_plan_data(plan): raise base.PublisherError("RECOVERY_ANCHOR_MEMBER_IDENTITY", "commit_plan.json", exit_code=27) snapshot_receipt = base._read_bytes(root / "prior_snapshot.json") _validate_snapshot_receipt(plan, snapshot_receipt) expected_members.extend(( ("prior_snapshot.json", snapshot_receipt), (DIRECT_ATTEMPT_CONTRACT, _attempt_contract_data(plan)), )) expected_manifest = _anchor_manifest_data(plan, expected_members) if base._read_bytes(root / DIRECT_ATTEMPT_CONTRACT) != _attempt_contract_data(plan): raise base.PublisherError("RECOVERY_ANCHOR_MEMBER_IDENTITY", DIRECT_ATTEMPT_CONTRACT, exit_code=27) if base._read_bytes(root / DIRECT_ANCHOR_MANIFEST) != expected_manifest: raise base.PublisherError("RECOVERY_ANCHOR_MANIFEST_IDENTITY", "mismatch", exit_code=27) _verify_snapshot_at(plan, root / "prior_snapshot") def _preparation_member_order(plan: DirectPlan) -> list[str]: return [relative for relative, _ in _anchor_fixed_members(plan)] + [ "prior_snapshot.json", DIRECT_ATTEMPT_CONTRACT, DIRECT_ANCHOR_MANIFEST, ] def _retire_clean_preparation(plan: DirectPlan) -> bool: preparation = plan.paths["anchor_prepare"] if not base._exists(preparation): return False if base._exists(plan.paths["attempt_receipt_root"]): raise base.PublisherError("RECOVERY_REQUIRED_ANCHOR_AMBIGUOUS", "official+preparation", exit_code=30) _verify_destination_prior(plan, "RECOVERY_PREPARATION_FORMAL_DRIFT") _verify_scope_sets(plan, candidate=False, code="RECOVERY_PREPARATION_SCOPE_DRIFT") actual = set(base._walk_files(preparation)) ordered = _preparation_member_order(plan) allowed_final = set(ordered) temp_for = {relative: _prepared_temp_relative(relative) for relative in ordered} allowed_temp = set(temp_for.values()) unexpected = actual - allowed_final - allowed_temp finals = [relative for relative in ordered if relative in actual] if unexpected or finals != ordered[:len(finals)]: raise base.PublisherError( "RECOVERY_REQUIRED_PREPARATION_AMBIGUOUS", f"unexpected={sorted(unexpected)};finals={finals}", exit_code=30, ) temps = [relative for relative in allowed_temp if relative in actual] expected_temp = [] if len(finals) == len(ordered) else [temp_for[ordered[len(finals)]]] if sorted(temps) not in ([], sorted(expected_temp)): raise base.PublisherError("RECOVERY_REQUIRED_PREPARATION_AMBIGUOUS", f"temps={sorted(temps)}", exit_code=30) fixed = dict(_anchor_fixed_members(plan)) fixed[DIRECT_ATTEMPT_CONTRACT] = _attempt_contract_data(plan) for relative in finals: if relative == "prior_snapshot.json": _validate_snapshot_receipt(plan, base._read_bytes(base._inside(preparation, relative))) elif relative == DIRECT_ANCHOR_MANIFEST: _verify_anchor(plan, preparation) elif base._read_bytes(base._inside(preparation, relative)) != fixed[relative]: raise base.PublisherError("RECOVERY_REQUIRED_PREPARATION_TAMPER", relative, exit_code=30) shutil.rmtree(base._long(preparation)) if base._exists(preparation): raise base.PublisherError("RECOVERY_REQUIRED_PREPARATION_RETIRE", str(preparation), exit_code=30) return True def _build_and_publish_anchor(plan: DirectPlan, fault: _PreAnchorFault) -> None: preparation = plan.paths["anchor_prepare"] os.makedirs(base._long(preparation), exist_ok=False) members = _anchor_fixed_members(plan) for relative, data in members: _prepared_write(plan, preparation, relative, data, fault) snapshot_receipt = _snapshot_receipt_data(plan, base._utc_now()) members.append(("prior_snapshot.json", snapshot_receipt)) _prepared_write(plan, preparation, "prior_snapshot.json", snapshot_receipt, fault) attempt_contract = _attempt_contract_data(plan) members.append((DIRECT_ATTEMPT_CONTRACT, attempt_contract)) _prepared_write(plan, preparation, DIRECT_ATTEMPT_CONTRACT, attempt_contract, fault) manifest = _anchor_manifest_data(plan, members) _prepared_write(plan, preparation, DIRECT_ANCHOR_MANIFEST, manifest, fault) _verify_anchor(plan, preparation) # Private preparation can take long enough for an external actor to drift # a frozen input. Rebind the complete locked preimage immediately before # the only operation that makes the recovery anchor authoritative. _locked_revalidate(plan) fault.checkpoint("BEFORE_ANCHOR_RENAME") fault.checkpoint("DURING_ANCHOR_RENAME") _atomic_rename_no_replace(preparation, plan.paths["attempt_receipt_root"]) def _snapshot_prior(plan: DirectPlan) -> None: # Kept as a narrow compatibility helper for callers that validate an # already-published official recovery anchor. _verify_snapshot_at(plan, plan.paths["snapshot"]) def _snapshot_expected(plan: DirectPlan) -> set[str]: return {"direct_prior_state.json"} | { "files/" + state.formal_relative_path for state in plan.destination_prior if state.present } def _verify_snapshot(plan: DirectPlan) -> None: _verify_snapshot_at(plan, plan.paths["snapshot"]) def _stage(plan: DirectPlan) -> None: os.makedirs(base._long(plan.paths["staging"]), exist_ok=False) for row in plan.candidate_rows: base._copy_file( base._inside(plan.paths["candidate_root"], row.relative_path), base._inside(plan.paths["staging"], "files/" + row.formal_relative_path), ) expected = {"files/" + row.formal_relative_path for row in plan.candidate_rows} actual = set(base._walk_files(plan.paths["staging"])) if actual != expected: raise base.PublisherError("DIRECT_STAGING_EXACT_SET", f"missing={sorted(expected-actual)};extra={sorted(actual-expected)}", exit_code=20) for row in plan.candidate_rows: base._verify_identity( base._inside(plan.paths["staging"], "files/" + row.formal_relative_path), row.bytes, row.sha256, "DIRECT_STAGING_IDENTITY", ) def _remove_staging(plan: DirectPlan, *, allow_partial: bool = False) -> None: if not base._exists(plan.paths["staging"]): return expected = {"files/" + row.formal_relative_path for row in plan.candidate_rows} actual = set(base._walk_files(plan.paths["staging"])) if (not allow_partial and actual != expected) or (allow_partial and not actual.issubset(expected)): raise base.PublisherError("DIRECT_STAGING_OWNERSHIP", "exact set mismatch", exit_code=27) by_formal = {"files/" + row.formal_relative_path: row for row in plan.candidate_rows} for relative in actual: row = by_formal[relative] base._verify_identity(base._inside(plan.paths["staging"], relative), row.bytes, row.sha256, "DIRECT_STAGING_OWNERSHIP") shutil.rmtree(base._long(plan.paths["staging"])) def _direct_terminal( plan: DirectPlan, status: str, exit_code: int, states: tuple[str, ...], process_identity: str, event_chain_sha256: str, event_time_start_utc: str, event_time_end_utc: str, event_process_identities: tuple[str, ...], *, committed: bool, rolled_back: bool, ) -> dict[str, Any]: return { "activation_formal_relative_path": plan.activation_row.formal_relative_path, "attempt_event_chain_row_count": len(states), "attempt_event_chain_schema": base.EVENT_CHAIN_SCHEMA, "attempt_event_chain_sha256": event_chain_sha256, "attempt_id": plan.identity["attempt_id"], "attempt_seq": plan.next_attempt_seq, "batch_id": plan.identity["batch_id"], "candidate_release_set_sha256": plan.identity["candidate_release_set_sha256"], "case_current_candidate_bytes": len(plan.candidate_index_data), "case_current_candidate_sha256": base._sha_bytes(plan.candidate_index_data), "case_current_prior_bytes": len(plan.prior_index_data), "case_current_prior_sha256": base._sha_bytes(plan.prior_index_data), "case_id": plan.identity["case_id"], "canonical_audit_id": plan.identity["canonical_audit_id"], "commit_order": DIRECT_COMMIT_ORDER, "commit_plan_sha256": base._sha_bytes(_commit_plan_data(plan)), "commit_strategy": DIRECT_STRATEGY, "committed": committed, "config_bytes": len(plan.config_data), "config_path": _relative_to(plan.config_path, plan.operation_root), "config_sha256": plan.config_sha256, "destination_count": len(plan.candidate_rows), "destination_set_sha256": _destination_exact_set(plan.candidate_rows), "event_process_identities": list(event_process_identities), "event_seq_end": plan.next_event_seq + len(states) - 1, "event_seq_start": plan.next_event_seq, "event_states": list(states), "event_time_end_utc": event_time_end_utc, "event_time_start_utc": event_time_start_utc, "exit_code": exit_code, "lock_absent_on_return": True, "prior_release_id": plan.identity["prior_release_id"], "prior_release_set_sha256": plan.prior_release_set, "prior_state_bytes": len(plan.prior_state_data), "prior_state_sha256": base._sha_bytes(plan.prior_state_data), "process_instance_identity": process_identity, "release_id": plan.identity["release_id"], "resolved_root": os.path.abspath(plan.operation_root), "result_current_bytes": plan.candidate_result_row.bytes, "result_current_sha256": plan.candidate_result_row.sha256, "review_handoff_id": plan.identity["review_handoff_id"], "rolled_back": rolled_back, "run_id": plan.identity["run_id"], "schema_version": DIRECT_TERMINAL_SCHEMA, "status": status, "task_id": plan.identity["task_id"], "volume_identity": base._volume(plan.operation_root).upper(), } def _receipt_expected(plan: DirectPlan, terminal_name: str) -> set[str]: return { "commit_plan.json", "prior_snapshot.json", DIRECT_ATTEMPT_CONTRACT, DIRECT_ANCHOR_MANIFEST, terminal_name, } | { "prior_snapshot/" + path for path in _snapshot_expected(plan) } def _prefix_receipt_expected(plan: DirectPlan) -> set[str]: return { "commit_plan.json", "prior_snapshot.json", DIRECT_ATTEMPT_CONTRACT, DIRECT_ANCHOR_MANIFEST, } | { "prior_snapshot/" + path for path in _snapshot_expected(plan) } def _validate_durable_prefix(plan: DirectPlan, *, allow_existing_terminal: bool) -> tuple[tuple[str, ...], str]: """Validate every durable byte before recovery is permitted to mutate.""" if base._exists(plan.paths["anchor_prepare"]): raise base.PublisherError("RECOVERY_REQUIRED_ANCHOR_AMBIGUOUS", "official+preparation", exit_code=30) allowed_extras: set[str] = set() if allow_existing_terminal: allowed_extras = { name for name in ("terminal.json", "recovery_terminal.json") if base._exists(plan.paths["attempt_receipt_root"] / name) } _verify_anchor(plan, plan.paths["attempt_receipt_root"], allowed_extras=allowed_extras) actual_receipts = set(base._walk_files(plan.paths["attempt_receipt_root"])) expected_receipts = _prefix_receipt_expected(plan) if allow_existing_terminal: for name in ("terminal.json", "recovery_terminal.json"): if name in actual_receipts: expected_receipts.add(name) if actual_receipts != expected_receipts: raise base.PublisherError( "RECOVERY_RECEIPT_EXACT_SET", f"missing={sorted(expected_receipts-actual_receipts)};extra={sorted(actual_receipts-expected_receipts)}", exit_code=27, ) rows = base._read_ledger(base._read_bytes(plan.paths["ledger_path"])) attempt_rows = base._event_rows_for_attempt(rows, plan.identity["attempt_id"]) observed_states = tuple(row["state"] for row in attempt_rows) legal_prefixes = ( (), ("PREPARED",), ("PREPARED", "VERIFIED"), base.SUCCESS_STATES[:3], base.SUCCESS_STATES, ) if observed_states not in legal_prefixes: raise base.PublisherError("RECOVERY_EVENT_STATES", str(list(observed_states)), exit_code=27) if attempt_rows: process_values = {row["process_identity"] for row in attempt_rows} if len(process_values) != 1: raise base.PublisherError("RECOVERY_EVENT_PROCESS_UNKNOWN", str(sorted(process_values)), exit_code=27) anchor = next(iter(process_values)) base._validate_attempt_event_chain(plan, observed_states, (anchor,) * len(observed_states)) else: anchor = "" for row in plan.candidate_rows: actual = _actual_state(plan.destination_paths[row.member_id], row.member_id, row.formal_relative_path) prior = next(item for item in plan.destination_prior if item.member_id == row.member_id) if actual not in (prior, _candidate_state(row)): raise base.PublisherError("RECOVERY_DESTINATION_UNKNOWN", row.member_id, exit_code=27) if base._exists(plan.paths["staging"]): expected_stage = {"files/" + row.formal_relative_path: row for row in plan.candidate_rows} actual_stage = set(base._walk_files(plan.paths["staging"])) if not actual_stage.issubset(expected_stage): raise base.PublisherError("RECOVERY_STAGING_UNKNOWN", str(sorted(actual_stage-set(expected_stage))), exit_code=27) for relative in actual_stage: row = expected_stage[relative] base._verify_identity(base._inside(plan.paths["staging"], relative), row.bytes, row.sha256, "RECOVERY_STAGING_IDENTITY") return observed_states, anchor def _verify_terminal(plan: DirectPlan, terminal: dict[str, Any]) -> None: process_identity = terminal.get("process_instance_identity") event_processes = terminal.get("event_process_identities") if not isinstance(process_identity, str) or not process_identity.startswith("PID-") or not isinstance(event_processes, list): raise base.PublisherError("REPLAY_TERMINAL_PROCESS", "invalid") states = tuple(terminal.get("event_states", ())) if states != base.SUCCESS_STATES: raise base.PublisherError("REPLAY_TERMINAL_STATES", str(states)) expected = _direct_terminal( plan, "COMMITTED", 0, states, process_identity, base._need_sha(terminal.get("attempt_event_chain_sha256"), "terminal.attempt_event_chain_sha256"), base._need_string(terminal.get("event_time_start_utc"), "terminal.event_time_start_utc"), base._need_string(terminal.get("event_time_end_utc"), "terminal.event_time_end_utc"), tuple(event_processes), committed=True, rolled_back=False, ) if terminal != expected: differing = sorted(key for key in set(expected) | set(terminal) if terminal.get(key) != expected.get(key)) raise base.PublisherError("REPLAY_TERMINAL_BINDING", str(differing)) base._validate_attempt_event_chain( plan, states, tuple(event_processes), expected_digest=terminal["attempt_event_chain_sha256"], expected_time_start=terminal["event_time_start_utc"], expected_time_end=terminal["event_time_end_utc"], ) if base._exists(plan.paths["lock_path"]): raise base.PublisherError("REPLAY_LOCK_PRESENT", str(plan.paths["lock_path"])) def _verify_replay(plan: DirectPlan) -> dict[str, Any]: if plan.lifecycle != "REPLAY": raise base.PublisherError("REPLAY_LIFECYCLE", plan.lifecycle) _verify_candidate_destinations(plan, "DIRECT_REPLAY_TARGET") _verify_anchor(plan, plan.paths["attempt_receipt_root"], allowed_extras={"terminal.json"}) terminal_path = plan.paths["attempt_receipt_root"] / "terminal.json" try: terminal = json.loads(base._read_bytes(terminal_path).decode("utf-8"), object_pairs_hook=base._pairs_no_duplicates) except base.PublisherError: raise except Exception as exc: raise base.PublisherError("REPLAY_TERMINAL_PARSE", str(exc)) from exc if not isinstance(terminal, dict): raise base.PublisherError("REPLAY_TERMINAL_TYPE", "not object") _verify_terminal(plan, terminal) actual_receipt = set(base._walk_files(plan.paths["attempt_receipt_root"])) expected_receipt = _receipt_expected(plan, "terminal.json") if actual_receipt != expected_receipt: raise base.PublisherError("REPLAY_RECEIPT_EXACT_SET", f"missing={sorted(expected_receipt-actual_receipt)};extra={sorted(actual_receipt-expected_receipt)}") value = dict(terminal) value["status"] = "IDEMPOTENT_COMMITTED" return value def _restore_prior(plan: DirectPlan, *, known_extra: Path | None = None) -> None: _verify_snapshot(plan) for row in reversed(plan.commit_rows): state = next(item for item in plan.destination_prior if item.member_id == row.member_id) path = plan.destination_paths[row.member_id] actual = _actual_state(path, row.member_id, row.formal_relative_path) if actual == state: continue if actual != _candidate_state(row): raise base.PublisherError("ROLLBACK_DESTINATION_UNKNOWN", row.member_id, exit_code=27) if state.present: prior_data = base._read_bytes(base._inside(plan.paths["snapshot"], "files/" + state.formal_relative_path)) base._atomic_replace(path, prior_data, plan.identity["attempt_id"] + "-rollback-" + row.member_id) else: os.unlink(base._long(path)) if known_extra is not None and base._exists(known_extra): if base._identity(known_extra) != (7, base._sha_bytes(b"attack\n")): raise base.PublisherError("ROLLBACK_EXTRA_UNKNOWN", str(known_extra), exit_code=27) os.unlink(base._long(known_extra)) _verify_destination_prior(plan, "ROLLBACK_READBACK") _verify_scope_sets(plan, candidate=False, code="ROLLBACK_SCOPE_READBACK") _remove_staging(plan, allow_partial=True) def _rollback( plan: DirectPlan, process_identity: str, detail: str, base_process_identity: str, *, terminal_name: str, known_extra: Path | None = None, base_states: tuple[str, ...] = base.SUCCESS_STATES[:3], ) -> dict[str, Any]: offset = len(base_states) base._append_event(plan, "ROLLING_BACK", offset, base._receipt_relative(plan, terminal_name), process_identity) _restore_prior(plan, known_extra=known_extra) base._append_event(plan, "ROLLED_BACK", offset + 1, base._receipt_relative(plan, terminal_name), process_identity) states = base_states + ("ROLLING_BACK", "ROLLED_BACK") event_processes = (base_process_identity,) * len(base_states) + (process_identity, process_identity) digest, start_time, end_time = base._validate_attempt_event_chain( plan, states, event_processes, rollback_terminal_name=terminal_name, ) value = _direct_terminal( plan, "ROLLED_BACK", 20, states, process_identity, digest, start_time, end_time, event_processes, committed=False, rolled_back=True, ) value["failure_detail"] = detail superseded = plan.paths["attempt_receipt_root"] / "terminal.json" if terminal_name == "recovery_terminal.json" and base._exists(superseded): size, digest_value = base._identity(superseded) value["superseded_terminal_bytes"] = size value["superseded_terminal_sha256"] = digest_value base._create_exclusive(plan.paths["attempt_receipt_root"] / terminal_name, base._canonical_json(value)) return value def _verify_rolled_back_replay(plan: DirectPlan) -> dict[str, Any]: if plan.lifecycle != "REPLAY_PRIOR": raise base.PublisherError("ROLLBACK_REPLAY_LIFECYCLE", plan.lifecycle) _verify_destination_prior(plan, "ROLLBACK_REPLAY_PRIOR") _verify_scope_sets(plan, candidate=False, code="ROLLBACK_REPLAY_SCOPE") terminal_name = ( "recovery_terminal.json" if base._exists(plan.paths["attempt_receipt_root"] / "recovery_terminal.json") else "terminal.json" ) anchor_extras = {terminal_name} if terminal_name == "recovery_terminal.json" and base._exists(plan.paths["attempt_receipt_root"] / "terminal.json"): anchor_extras.add("terminal.json") _verify_anchor(plan, plan.paths["attempt_receipt_root"], allowed_extras=anchor_extras) terminal_path = plan.paths["attempt_receipt_root"] / terminal_name try: terminal = json.loads(base._read_bytes(terminal_path).decode("utf-8"), object_pairs_hook=base._pairs_no_duplicates) except base.PublisherError: raise except Exception as exc: raise base.PublisherError("ROLLBACK_REPLAY_TERMINAL_PARSE", str(exc), exit_code=27) from exc if not isinstance(terminal, dict): raise base.PublisherError("ROLLBACK_REPLAY_TERMINAL_TYPE", "not object", exit_code=27) states = tuple(terminal.get("event_states", ())) legal_prefixes = ( (), ("PREPARED",), ("PREPARED", "VERIFIED"), base.SUCCESS_STATES[:3], base.SUCCESS_STATES, ) if len(states) < 2 or states[-2:] != ("ROLLING_BACK", "ROLLED_BACK") or states[:-2] not in legal_prefixes: raise base.PublisherError("ROLLBACK_REPLAY_STATES", str(states), exit_code=27) event_processes = terminal.get("event_process_identities") process_identity = terminal.get("process_instance_identity") failure_detail = terminal.get("failure_detail") if ( not isinstance(event_processes, list) or not isinstance(process_identity, str) or not process_identity.startswith("PID-") or not isinstance(failure_detail, str) or not failure_detail ): raise base.PublisherError("ROLLBACK_REPLAY_FIELDS", "process/failure", exit_code=27) expected = _direct_terminal( plan, "ROLLED_BACK", 20, states, process_identity, base._need_sha(terminal.get("attempt_event_chain_sha256"), "terminal.attempt_event_chain_sha256"), base._need_string(terminal.get("event_time_start_utc"), "terminal.event_time_start_utc"), base._need_string(terminal.get("event_time_end_utc"), "terminal.event_time_end_utc"), tuple(event_processes), committed=False, rolled_back=True, ) expected["failure_detail"] = failure_detail superseded = plan.paths["attempt_receipt_root"] / "terminal.json" if terminal_name == "recovery_terminal.json" and base._exists(superseded): size, digest_value = base._identity(superseded) expected["superseded_terminal_bytes"] = size expected["superseded_terminal_sha256"] = digest_value if terminal != expected: differing = sorted(key for key in set(expected) | set(terminal) if terminal.get(key) != expected.get(key)) raise base.PublisherError("ROLLBACK_REPLAY_TERMINAL_BINDING", str(differing), exit_code=27) base._validate_attempt_event_chain( plan, states, tuple(event_processes), rollback_terminal_name=terminal_name, expected_digest=terminal["attempt_event_chain_sha256"], expected_time_start=terminal["event_time_start_utc"], expected_time_end=terminal["event_time_end_utc"], ) expected_receipts = _prefix_receipt_expected(plan) | {terminal_name} if terminal_name == "recovery_terminal.json" and base._exists(superseded): expected_receipts.add("terminal.json") actual_receipts = set(base._walk_files(plan.paths["attempt_receipt_root"])) if actual_receipts != expected_receipts: raise base.PublisherError("ROLLBACK_REPLAY_RECEIPT_EXACT_SET", "mismatch", exit_code=27) if base._exists(plan.paths["lock_path"]) or base._exists(plan.paths["staging"]): raise base.PublisherError("ROLLBACK_REPLAY_ACTIVE_STATE", "lock/staging", exit_code=27) value = dict(terminal) value["status"] = "IDEMPOTENT_ROLLED_BACK" return value def _recover(plan: DirectPlan, reason: base.PublisherError) -> dict[str, Any]: lock_value = { "schema_version": DIRECT_TERMINAL_SCHEMA, "attempt_id": plan.identity["attempt_id"], "config_bytes": len(plan.config_data), "config_sha256": plan.config_sha256, "process_instance_identity": base._process_identity(), "acquired_at": base._utc_now(), } base._create_exclusive(plan.paths["lock_path"], base._canonical_json(lock_value)) process_identity = lock_value["process_instance_identity"] try: observed_states, anchor = _validate_durable_prefix( plan, allow_existing_terminal=plan.lifecycle == "REPLAY", ) return _rollback( plan, process_identity, f"RECOVERY:{reason.code}:{reason.detail}", anchor or process_identity, terminal_name="recovery_terminal.json", base_states=observed_states, ) finally: if base._exists(plan.paths["lock_path"]): os.unlink(base._long(plan.paths["lock_path"])) def _publish(plan: DirectPlan) -> dict[str, Any]: if plan.lifecycle != "FRESH": raise base.PublisherError("PUBLISH_LIFECYCLE", plan.lifecycle) process_identity = base._process_identity() lock_value = { "schema_version": DIRECT_TERMINAL_SCHEMA, "attempt_id": plan.identity["attempt_id"], "config_path": _relative_to(plan.config_path, plan.operation_root), "config_bytes": len(plan.config_data), "config_sha256": plan.config_sha256, "resolved_root": os.path.abspath(plan.operation_root), "volume_identity": base._volume(plan.operation_root).upper(), "process_instance_identity": process_identity, "acquired_at": base._utc_now(), } base._create_exclusive(plan.paths["lock_path"], base._canonical_json(lock_value)) committed = False committed_event = False durable_prefix = False known_extra: Path | None = None try: fault = plan.config["test_control"]["fault"] after = plan.config["test_control"]["fault_commit_after"] if fault == "PAUSE_AFTER_LOCK": marker = plan.paths["race_marker"] deadline = time.monotonic() + 5.0 while time.monotonic() < deadline and not base._exists(marker): time.sleep(0.01) if not base._exists(marker): raise base.PublisherError("TEST_RACE_MARKER_TIMEOUT", str(marker)) _locked_revalidate(plan) try: remnant_retired = _retire_clean_preparation(plan) except base.PublisherError as exc: if exc.code.startswith("RECOVERY_REQUIRED"): raise raise base.PublisherError( "RECOVERY_REQUIRED_PREPARATION_AMBIGUOUS", f"{exc.code}:{exc.detail}", exit_code=30, ) from exc preanchor_fault = _PreAnchorFault( fault, after, enabled=not remnant_retired and fault in { "DIRECT_INTERRUPT_PREANCHOR", "DIRECT_IO_ERROR_PREANCHOR", }, ) _build_and_publish_anchor(plan, preanchor_fault) durable_prefix = True preanchor_fault.checkpoint("AFTER_ANCHOR_RENAME") _verify_anchor(plan, plan.paths["attempt_receipt_root"]) if fault == "DIRECT_INTERRUPT_AFTER_SNAPSHOT": raise DirectProcessInterrupted("INJECTED_PROCESS_RESTART", "AFTER_SNAPSHOT", exit_code=20) base._append_event(plan, "PREPARED", 0, base._receipt_relative(plan, "prior_snapshot.json"), process_identity) if fault == "DIRECT_INTERRUPT_AFTER_PREPARED": raise DirectProcessInterrupted("INJECTED_PROCESS_RESTART", "AFTER_PREPARED", exit_code=20) _stage(plan) base._append_event(plan, "VERIFIED", 1, base._receipt_relative(plan, "prior_snapshot.json"), process_identity) if fault == "DIRECT_INTERRUPT_AFTER_VERIFIED": raise DirectProcessInterrupted("INJECTED_PROCESS_RESTART", "AFTER_VERIFIED", exit_code=20) base._append_event(plan, "COMMITTING", 2, base._receipt_relative(plan, "terminal.json"), process_identity) if fault == "DIRECT_INTERRUPT_BEFORE_FIRST_DESTINATION": raise DirectProcessInterrupted("INJECTED_PROCESS_RESTART", "BEFORE_FIRST_DESTINATION", exit_code=20) for index, row in enumerate(plan.commit_rows, start=1): source = base._inside(plan.paths["staging"], "files/" + row.formal_relative_path) target = plan.destination_paths[row.member_id] prior = next(item for item in plan.destination_prior if item.member_id == row.member_id) data = base._read_bytes(source) if prior.present: base._atomic_replace(target, data, plan.identity["attempt_id"] + f"-{index:03d}") else: base._create_exclusive(target, data) committed = True base._verify_identity(target, row.bytes, row.sha256, "DIRECT_COMMIT_READBACK") if fault == "DIRECT_EXTRA_TARGET_AFTER_COMMIT" and after == index: known_extra = plan.scope_roots["MANIFEST"] / f".publisher-attack-{plan.identity['attempt_id']}.txt" base._create_exclusive(known_extra, b"attack\n") raise base.PublisherError("INJECTED_EXTRA_TARGET", str(index), exit_code=20) if fault == "DIRECT_FAIL_AFTER_COMMIT" and after == index: raise base.PublisherError("INJECTED_COMMIT_BOUNDARY", str(index), exit_code=20) if fault == "DIRECT_INTERRUPT_AFTER_COMMIT" and after == index: raise DirectProcessInterrupted("INJECTED_PROCESS_RESTART", f"AFTER_COMMIT:{index}", exit_code=20) if fault == "DIRECT_PAUSE_AFTER_COMMIT" and after == index: marker = plan.paths["race_marker"] ready = Path(str(marker) + ".ready") proceed = Path(str(marker) + ".continue") base._create_exclusive(ready, b"ready\n") deadline = time.monotonic() + 10.0 while time.monotonic() < deadline and not base._exists(proceed): time.sleep(0.01) if not base._exists(proceed): raise base.PublisherError("TEST_OBSERVER_MARKER_TIMEOUT", str(proceed), exit_code=20) os.unlink(base._long(ready)) os.unlink(base._long(proceed)) _verify_candidate_destinations(plan, "DIRECT_POSTCOMMIT") _remove_staging(plan) base._append_event(plan, "COMMITTED", 3, base._receipt_relative(plan, "terminal.json"), process_identity) committed_event = True event_processes = (process_identity,) * len(base.SUCCESS_STATES) digest, start_time, end_time = base._validate_attempt_event_chain(plan, base.SUCCESS_STATES, event_processes) value = _direct_terminal( plan, "COMMITTED", 0, base.SUCCESS_STATES, process_identity, digest, start_time, end_time, event_processes, committed=True, rolled_back=False, ) base._create_exclusive(plan.paths["attempt_receipt_root"] / "terminal.json", base._canonical_json(value)) return value except DirectProcessInterrupted: # Isolated fixtures use this to model a new process resuming an exact # durable prefix. Production config forbids every non-NONE fault. raise except base.PublisherError as exc: if durable_prefix: observed_states, anchor = _validate_durable_prefix(plan, allow_existing_terminal=False) return _rollback( plan, process_identity, f"{exc.code}:{exc.detail}", anchor or process_identity, terminal_name="terminal.json", known_extra=known_extra, base_states=observed_states, ) raise except Exception as exc: changed = False for row in plan.candidate_rows: prior = next(item for item in plan.destination_prior if item.member_id == row.member_id) actual = _actual_state( plan.destination_paths[row.member_id], row.member_id, row.formal_relative_path, ) if actual != prior: changed = True break wrapped = base.PublisherError( "DIRECT_WRITE_EXCEPTION", f"{type(exc).__name__}:{exc}", exit_code=27, ) if durable_prefix: observed_states, anchor = _validate_durable_prefix(plan, allow_existing_terminal=False) return _rollback( plan, process_identity, f"{wrapped.code}:{wrapped.detail}", anchor or process_identity, terminal_name="terminal.json", known_extra=known_extra, base_states=observed_states, ) raise wrapped finally: if base._exists(plan.paths["lock_path"]): os.unlink(base._long(plan.paths["lock_path"])) def run_direct_config( config: dict[str, Any], config_path: Path, config_data: bytes, *, validate_only: bool = False, ) -> dict[str, Any]: plan = _validate_direct_config(config, config_path, config_data) if validate_only: return { "schema_version": DIRECT_TERMINAL_SCHEMA, "status": "VALIDATION_PASS", "exit_code": 0, "lifecycle": plan.lifecycle, "candidate_count": len(plan.candidate_rows), "target_count": len(plan.target_rows), "history_count": len(plan.history_rows), "next_event_seq": plan.next_event_seq, "next_attempt_seq": plan.next_attempt_seq, "config_bytes": len(plan.config_data), "config_sha256": plan.config_sha256, "commit_strategy": DIRECT_STRATEGY, } try: if plan.lifecycle == "REPLAY": try: return _verify_replay(plan) except base.PublisherError as exc: return _recover(plan, exc) if plan.lifecycle == "REPLAY_PRIOR": return _verify_rolled_back_replay(plan) if plan.lifecycle in ("RECOVERY_INCOMPLETE", "RECOVERY_MIXED"): return _recover(plan, base.PublisherError("MIXED_STABLE_PATH_STATE", "recovery required")) return _publish(plan) except base.PublisherError as exc: if not exc.code.startswith("RECOVERY_REQUIRED") and plan.lifecycle == "FRESH": raise error_code = ( exc.code if exc.code.startswith("RECOVERY_REQUIRED") else "RECOVERY_REQUIRED_OFFICIAL_ANCHOR" ) return { "attempt_id": plan.identity["attempt_id"], "batch_id": plan.identity["batch_id"], "case_id": plan.identity["case_id"], "config_bytes": len(plan.config_data), "config_sha256": plan.config_sha256, "detail": exc.detail, "error_code": error_code, "exit_code": 30, "formal_mutation_permitted": False, "run_id": plan.identity["run_id"], "schema_version": DIRECT_TERMINAL_SCHEMA, "status": "RECOVERY_REQUIRED", "task_id": plan.identity["task_id"], }