"""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"],
|
}
|
