"""Declarative, fail-closed, project-level content publisher.
|
|
The only user-visible commit point is an existing Markdown cases current index.
|
Immutable release members are prepared first; the Markdown index is replaced
|
atomically last. The result-side Markdown entry remains an invariant thin link.
|
No industry, batch, task, run, release, or attempt identity is hard-coded here.
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
import csv
|
import hashlib
|
import io
|
import json
|
import os
|
import stat
|
import sys
|
import time
|
import uuid
|
from dataclasses import dataclass
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
from typing import Any, Iterable
|
|
|
CONFIG_SCHEMA = "SHARED_CONTENT_PUBLISHER_CONFIG_V2"
|
DIRECT_CONFIG_SCHEMA = "SHARED_CONTENT_PUBLISHER_CONFIG_V3"
|
TERMINAL_SCHEMA = "SHARED_CONTENT_PUBLISHER_TERMINAL_V3"
|
EVENT_CHAIN_SCHEMA = "SHARED_CONTENT_PUBLISHER_ATTEMPT_EVENT_CHAIN_V1"
|
COMMIT_STRATEGY = "MARKDOWN_CURRENT_INDEX_REPLACE_V1"
|
DIRECT_COMMIT_STRATEGY = "DIRECT_STABLE_PATH_SET_V1"
|
LEDGER_HEADER = [
|
"release_id", "event_seq", "state", "prior_release_id",
|
"prior_release_set_sha256", "candidate_release_set_sha256", "task_id",
|
"case_id", "batch_id", "run_id", "accepted_audit_id", "operator",
|
"process_identity", "event_time", "recovery_or_rollback_receipt",
|
"attempt_id", "attempt_seq",
|
]
|
SUCCESS_STATES = ("PREPARED", "VERIFIED", "COMMITTING", "COMMITTED")
|
HEX64 = set("0123456789ABCDEF")
|
|
|
class PublisherError(Exception):
|
"""A classified publisher failure."""
|
|
def __init__(self, code: str, detail: str, *, exit_code: int = 12) -> None:
|
super().__init__(f"{code}:{detail}")
|
self.code = code
|
self.detail = detail
|
self.exit_code = exit_code
|
|
|
def _pairs_no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
result: dict[str, Any] = {}
|
for key, value in pairs:
|
if key in result:
|
raise PublisherError("CONFIG_DUPLICATE_KEY", key)
|
result[key] = value
|
return result
|
|
|
def _load_config(path: Path) -> tuple[dict[str, Any], bytes]:
|
data = _read_bytes(path)
|
if data.startswith(b"\xef\xbb\xbf") or b"\x00" in data:
|
raise PublisherError("CONFIG_ENCODING", "BOM/NUL forbidden")
|
try:
|
value = json.loads(data.decode("utf-8"), object_pairs_hook=_pairs_no_duplicates)
|
except PublisherError:
|
raise
|
except Exception as exc:
|
raise PublisherError("CONFIG_PARSE", str(exc)) from exc
|
if not isinstance(value, dict):
|
raise PublisherError("CONFIG_TYPE", "root must be object")
|
return value, data
|
|
|
def _expect_keys(value: dict[str, Any], expected: Iterable[str], where: str) -> None:
|
expected_set = set(expected)
|
actual = set(value)
|
missing = sorted(expected_set - actual)
|
extra = sorted(actual - expected_set)
|
if missing or extra:
|
raise PublisherError("CONFIG_KEYS", f"{where}:missing={missing};extra={extra}")
|
|
|
def _need_string(value: Any, where: str) -> str:
|
if not isinstance(value, str) or not value:
|
raise PublisherError("CONFIG_TYPE", f"{where} must be nonempty string")
|
return value
|
|
|
def _need_int(value: Any, where: str, *, positive: bool = False) -> int:
|
if isinstance(value, bool) or not isinstance(value, int) or (positive and value <= 0):
|
raise PublisherError("CONFIG_TYPE", f"{where} must be integer")
|
return value
|
|
|
def _need_sha(value: Any, where: str, *, lower: bool = False) -> str:
|
text = _need_string(value, where)
|
expected = text.lower() if lower else text.upper()
|
alphabet = set("0123456789abcdef") if lower else HEX64
|
if text != expected or len(text) != 64 or any(ch not in alphabet for ch in text):
|
raise PublisherError("CONFIG_SHA256", where)
|
return text
|
|
|
def _long(path: Path | str) -> str:
|
text = os.path.abspath(os.fspath(path))
|
if os.name != "nt" or text.startswith("\\\\?\\"):
|
return text
|
if text.startswith("\\\\"):
|
return "\\\\?\\UNC\\" + text[2:]
|
return "\\\\?\\" + text
|
|
|
def _lstat(path: Path | str) -> os.stat_result:
|
return os.stat(_long(path), follow_symlinks=False)
|
|
|
def _exists(path: Path | str) -> bool:
|
try:
|
_lstat(path)
|
return True
|
except FileNotFoundError:
|
return False
|
|
|
def _is_reparse(st: os.stat_result) -> bool:
|
return bool(getattr(st, "st_file_attributes", 0) & 0x400) or stat.S_ISLNK(st.st_mode)
|
|
|
def _ordinary_file(path: Path | str) -> os.stat_result:
|
try:
|
st = _lstat(path)
|
except OSError as exc:
|
raise PublisherError("FILE_UNREADABLE", f"{path}:{exc}") from exc
|
if not stat.S_ISREG(st.st_mode) or _is_reparse(st):
|
raise PublisherError("FILE_NOT_ORDINARY", str(path))
|
return st
|
|
|
def _ordinary_dir(path: Path | str) -> os.stat_result:
|
try:
|
st = _lstat(path)
|
except OSError as exc:
|
raise PublisherError("DIR_UNREADABLE", f"{path}:{exc}") from exc
|
if not stat.S_ISDIR(st.st_mode) or _is_reparse(st):
|
raise PublisherError("DIR_NOT_ORDINARY", str(path))
|
return st
|
|
|
def _read_bytes(path: Path | str) -> bytes:
|
_ordinary_file(path)
|
try:
|
with open(_long(path), "rb") as handle:
|
return handle.read()
|
except OSError as exc:
|
raise PublisherError("FILE_UNREADABLE", f"{path}:{exc}") from exc
|
|
|
def _sha_bytes(data: bytes) -> str:
|
return hashlib.sha256(data).hexdigest().upper()
|
|
|
def _identity(path: Path | str) -> tuple[int, str]:
|
data = _read_bytes(path)
|
if not data:
|
raise PublisherError("FILE_EMPTY", str(path))
|
return len(data), _sha_bytes(data)
|
|
|
def _identity_independent(path: Path | str) -> tuple[int, str]:
|
_ordinary_file(path)
|
total = 0
|
digest = hashlib.sha256()
|
try:
|
with open(_long(path), "rb", buffering=0) as handle:
|
while True:
|
chunk = handle.read(131072)
|
if not chunk:
|
break
|
total += len(chunk)
|
digest.update(chunk)
|
except OSError as exc:
|
raise PublisherError("FILE_UNREADABLE", f"{path}:{exc}") from exc
|
if total <= 0:
|
raise PublisherError("FILE_EMPTY", str(path))
|
return total, digest.hexdigest().upper()
|
|
|
def _canonical_json(value: Any, *, newline: bool = True) -> bytes:
|
data = json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode("utf-8")
|
return data + (b"\n" if newline else b"")
|
|
|
def _utc_now() -> str:
|
return datetime.now(timezone.utc).isoformat(timespec="microseconds").replace("+00:00", "Z")
|
|
|
def _parse_strict_utc(value: str, code: str) -> datetime:
|
if not isinstance(value, str) or not value.endswith("Z"):
|
raise PublisherError(code, str(value), exit_code=27)
|
try:
|
parsed = datetime.fromisoformat(value[:-1] + "+00:00")
|
except ValueError as exc:
|
raise PublisherError(code, value, exit_code=27) from exc
|
canonical = parsed.astimezone(timezone.utc).isoformat(timespec="microseconds").replace("+00:00", "Z")
|
if parsed.utcoffset() != timezone.utc.utcoffset(parsed) or canonical != value:
|
raise PublisherError(code, value, exit_code=27)
|
return parsed
|
|
|
def _safe_relative(value: Any, where: str) -> str:
|
text = _need_string(value, where).replace("\\", "/")
|
if text.startswith("/") or Path(text).is_absolute() or any(part in ("", ".", "..") for part in text.split("/")):
|
raise PublisherError("PATH_RELATIVE", where)
|
return text
|
|
|
def _norm(path: Path | str) -> str:
|
return os.path.normcase(os.path.abspath(os.fspath(path))).casefold()
|
|
|
def _inside(root: Path, relative: str) -> Path:
|
candidate = Path(os.path.abspath(os.path.join(root, *relative.split("/"))))
|
if os.path.commonpath([_norm(root), _norm(candidate)]) != _norm(root):
|
raise PublisherError("PATH_ESCAPE", relative)
|
return candidate
|
|
|
def _volume(path: Path) -> str:
|
drive, _ = os.path.splitdrive(os.path.abspath(path))
|
return drive.upper() if drive else str(_lstat(path).st_dev)
|
|
|
def _physical_chain(path: Path, *, final_kind: str | None = None) -> None:
|
"""Reject every existing reparse/alias component from the volume root down."""
|
absolute = Path(os.path.abspath(path))
|
anchor = Path(absolute.anchor)
|
current = anchor
|
parts = absolute.parts[1:] if absolute.anchor else absolute.parts
|
if _exists(anchor):
|
_ordinary_dir(anchor)
|
for index, part in enumerate(parts):
|
current = current / part
|
if not _exists(current):
|
break
|
st = _lstat(current)
|
if _is_reparse(st):
|
raise PublisherError("PATH_REPARSE", str(current))
|
is_final = index == len(parts) - 1
|
if not is_final or final_kind == "dir":
|
if not stat.S_ISDIR(st.st_mode):
|
raise PublisherError("PATH_ANCESTOR_TYPE", str(current))
|
elif final_kind == "file" and not stat.S_ISREG(st.st_mode):
|
raise PublisherError("PATH_FINAL_TYPE", str(current))
|
if _norm(os.path.realpath(current)) != _norm(current):
|
raise PublisherError("PATH_ALIAS", str(current))
|
|
|
def _validate_parent_chain(root: Path, path: Path, *, final_kind: str | None = None) -> None:
|
if os.path.commonpath([_norm(root), _norm(path)]) != _norm(root):
|
raise PublisherError("PATH_OUTSIDE_ROOT", str(path))
|
_physical_chain(path, final_kind=final_kind)
|
|
|
def _overlap(first: Path, second: Path) -> bool:
|
a, b = _norm(first), _norm(second)
|
return os.path.commonpath([a, b]) in (a, b)
|
|
|
def _validate_path_graph(paths: dict[str, Path], allowed: set[frozenset[str]]) -> None:
|
names = sorted(paths)
|
for index, left in enumerate(names):
|
for right in names[index + 1:]:
|
if _overlap(paths[left], paths[right]) and frozenset((left, right)) not in allowed:
|
raise PublisherError("PATH_OVERLAP", f"{left}<->{right}")
|
|
|
def _walk_files(root: Path) -> list[str]:
|
_ordinary_dir(root)
|
found: list[str] = []
|
|
def visit(current: Path, relative: str) -> None:
|
try:
|
entries = sorted(os.scandir(_long(current)), key=lambda item: item.name.casefold())
|
except OSError as exc:
|
raise PublisherError("TREE_UNREADABLE", f"{current}:{exc}") from exc
|
for entry in entries:
|
rel = f"{relative}/{entry.name}" if relative else entry.name
|
st = entry.stat(follow_symlinks=False)
|
if _is_reparse(st):
|
raise PublisherError("TREE_REPARSE", rel)
|
if stat.S_ISDIR(st.st_mode):
|
visit(current / entry.name, rel)
|
elif stat.S_ISREG(st.st_mode):
|
found.append(rel.replace("\\", "/"))
|
else:
|
raise PublisherError("TREE_OBJECT_TYPE", rel)
|
|
visit(root, "")
|
return found
|
|
|
@dataclass(frozen=True)
|
class ArtifactRow:
|
member_id: str
|
relative_path: str
|
formal_relative_path: str
|
artifact_type: str
|
commit_role: str
|
bytes: int
|
sha256: str
|
|
|
@dataclass(frozen=True)
|
class FormalRow:
|
member_id: str
|
formal_relative_path: str
|
artifact_type: str
|
bytes: int
|
sha256: str
|
|
|
def _parse_artifact_rows(value: Any, where: str) -> tuple[ArtifactRow, ...]:
|
if not isinstance(value, list) or not value:
|
raise PublisherError("CONFIG_TYPE", f"{where} must be nonempty array")
|
rows: list[ArtifactRow] = []
|
member_ids: set[str] = set()
|
sources: set[str] = set()
|
formals: set[str] = set()
|
keys = {"member_id", "relative_path", "formal_relative_path", "artifact_type", "commit_role", "bytes", "sha256"}
|
for index, item in enumerate(value):
|
if not isinstance(item, dict):
|
raise PublisherError("CONFIG_TYPE", f"{where}[{index}]")
|
_expect_keys(item, keys, f"{where}[{index}]")
|
member = _need_string(item["member_id"], f"{where}[{index}].member_id")
|
source = _safe_relative(item["relative_path"], f"{where}[{index}].relative_path")
|
formal = _safe_relative(item["formal_relative_path"], f"{where}[{index}].formal_relative_path")
|
artifact = _need_string(item["artifact_type"], f"{where}[{index}].artifact_type")
|
role = _need_string(item["commit_role"], f"{where}[{index}].commit_role")
|
if role not in ("RELEASE_MEMBER", "CASE_CURRENT_INDEX"):
|
raise PublisherError("CONFIG_ENUM", f"{where}[{index}].commit_role")
|
size = _need_int(item["bytes"], f"{where}[{index}].bytes", positive=True)
|
digest = _need_sha(item["sha256"], f"{where}[{index}].sha256")
|
if member in member_ids or source in sources or formal.casefold() in formals:
|
raise PublisherError("CONFIG_DUPLICATE_ROW", f"{where}[{index}]")
|
member_ids.add(member)
|
sources.add(source)
|
formals.add(formal.casefold())
|
rows.append(ArtifactRow(member, source, formal, artifact, role, size, digest))
|
return tuple(rows)
|
|
|
def _parse_formal_rows(value: Any, where: str) -> tuple[FormalRow, ...]:
|
if not isinstance(value, list) or not value:
|
raise PublisherError("CONFIG_TYPE", f"{where} must be nonempty array")
|
rows: list[FormalRow] = []
|
seen_members: set[str] = set()
|
seen_paths: set[str] = set()
|
keys = {"member_id", "formal_relative_path", "artifact_type", "bytes", "sha256"}
|
for index, item in enumerate(value):
|
if not isinstance(item, dict):
|
raise PublisherError("CONFIG_TYPE", f"{where}[{index}]")
|
_expect_keys(item, keys, f"{where}[{index}]")
|
member = _need_string(item["member_id"], f"{where}[{index}].member_id")
|
formal = _safe_relative(item["formal_relative_path"], f"{where}[{index}].formal_relative_path")
|
artifact = _need_string(item["artifact_type"], f"{where}[{index}].artifact_type")
|
size = _need_int(item["bytes"], f"{where}[{index}].bytes", positive=True)
|
digest = _need_sha(item["sha256"], f"{where}[{index}].sha256")
|
if member in seen_members or formal.casefold() in seen_paths:
|
raise PublisherError("CONFIG_DUPLICATE_ROW", f"{where}[{index}]")
|
seen_members.add(member)
|
seen_paths.add(formal.casefold())
|
rows.append(FormalRow(member, formal, artifact, size, digest))
|
return tuple(rows)
|
|
|
def _parse_history_rows(value: Any, where: str) -> tuple[tuple[str, int, str], ...]:
|
if not isinstance(value, list) or not value:
|
raise PublisherError("CONFIG_TYPE", f"{where} must be nonempty array")
|
result: list[tuple[str, int, str]] = []
|
seen: set[str] = set()
|
for index, item in enumerate(value):
|
if not isinstance(item, dict):
|
raise PublisherError("CONFIG_TYPE", f"{where}[{index}]")
|
_expect_keys(item, {"relative_path", "bytes", "sha256"}, f"{where}[{index}]")
|
rel = _safe_relative(item["relative_path"], f"{where}[{index}].relative_path")
|
size = _need_int(item["bytes"], f"{where}[{index}].bytes", positive=True)
|
digest = _need_sha(item["sha256"], f"{where}[{index}].sha256")
|
if rel.casefold() in seen:
|
raise PublisherError("CONFIG_DUPLICATE_ROW", f"{where}[{index}]")
|
seen.add(rel.casefold())
|
result.append((rel, size, digest))
|
return tuple(result)
|
|
|
def _verify_identity(path: Path, size: int, digest: str, code: str) -> None:
|
first = _identity(path)
|
second = _identity_independent(path)
|
if first != second or first != (size, digest):
|
raise PublisherError(code, str(path))
|
|
|
def _verify_history(root: Path, rows: tuple[tuple[str, int, str], ...], threshold: int, minimum: int) -> None:
|
expected = {row[0] for row in rows}
|
actual = set(_walk_files(root))
|
if actual != expected:
|
raise PublisherError("HISTORY_EXACT_SET", f"missing={sorted(expected-actual)};extra={sorted(actual-expected)}")
|
long_count = 0
|
for rel, size, digest in rows:
|
path = _inside(root, rel)
|
_verify_identity(path, size, digest, "HISTORY_IDENTITY")
|
if len(os.path.abspath(path)) >= threshold:
|
long_count += 1
|
if long_count < minimum:
|
raise PublisherError("HISTORY_LONG_PATH_COUNT", f"{long_count}<{minimum}")
|
|
|
def _manifest_rows(data: bytes, *, candidate: bool) -> list[dict[str, str]]:
|
header = (["member_id", "candidate_relative_path", "formal_relative_path", "artifact_type", "bytes", "sha256"]
|
if candidate else ["member_id", "formal_relative_path", "artifact_type", "bytes", "sha256"])
|
try:
|
reader = csv.DictReader(io.StringIO(data.decode("utf-8-sig"), newline=""))
|
if reader.fieldnames != header:
|
raise PublisherError("MANIFEST_HEADER", str(reader.fieldnames))
|
rows = list(reader)
|
except PublisherError:
|
raise
|
except Exception as exc:
|
raise PublisherError("MANIFEST_PARSE", str(exc)) from exc
|
if not rows or any(None in row for row in rows):
|
raise PublisherError("MANIFEST_ROWS", "empty or malformed")
|
ids: set[str] = set()
|
paths: set[str] = set()
|
for row in rows:
|
member = row["member_id"]
|
formal = row["formal_relative_path"].replace("\\", "/").casefold()
|
if not member or member in ids or not formal or formal in paths:
|
raise PublisherError("MANIFEST_DUPLICATE", member or formal)
|
try:
|
if int(row["bytes"]) <= 0:
|
raise ValueError
|
except ValueError as exc:
|
raise PublisherError("MANIFEST_BYTES", member) from exc
|
_need_sha(row["sha256"], f"manifest.{member}.sha256")
|
ids.add(member)
|
paths.add(formal)
|
return rows
|
|
|
def _formal_tuple(row: FormalRow) -> tuple[str, str, str, int, str]:
|
return row.member_id, row.formal_relative_path, row.artifact_type, row.bytes, row.sha256
|
|
|
def _artifact_formal_tuple(row: ArtifactRow) -> tuple[str, str, str, int, str]:
|
return row.member_id, row.formal_relative_path, row.artifact_type, row.bytes, row.sha256
|
|
|
def _manifest_tuple(row: dict[str, str]) -> tuple[str, str, str, int, str]:
|
return row["member_id"], row["formal_relative_path"].replace("\\", "/"), row["artifact_type"], int(row["bytes"]), row["sha256"]
|
|
|
def _release_set(data: bytes, self_path: str) -> str:
|
descriptor = _canonical_json({"bytes": len(data), "formal_relative_path": self_path, "sha256": _sha_bytes(data)}, newline=False)
|
return _sha_bytes(data + descriptor)
|
|
|
def _parse_checks(value: Any, where: str) -> tuple[tuple[str, tuple[str, ...]], ...]:
|
if not isinstance(value, list) or not value:
|
raise PublisherError("CONFIG_TYPE", f"{where} must be nonempty array")
|
result: list[tuple[str, tuple[str, ...]]] = []
|
for index, item in enumerate(value):
|
if not isinstance(item, dict):
|
raise PublisherError("CONFIG_TYPE", f"{where}[{index}]")
|
_expect_keys(item, {"relative_path", "required_utf8_substrings"}, f"{where}[{index}]")
|
rel = _safe_relative(item["relative_path"], f"{where}[{index}].relative_path")
|
required = item["required_utf8_substrings"]
|
if not isinstance(required, list) or not required:
|
raise PublisherError("CONFIG_TYPE", f"{where}[{index}].required_utf8_substrings")
|
strings = tuple(_need_string(text, f"{where}[{index}].required_utf8_substrings") for text in required)
|
result.append((rel, strings))
|
return tuple(result)
|
|
|
def _verify_checks(root: Path, checks: tuple[tuple[str, tuple[str, ...]], ...], code: str) -> None:
|
for rel, required in checks:
|
try:
|
text = _read_bytes(_inside(root, rel)).decode("utf-8")
|
except UnicodeDecodeError as exc:
|
raise PublisherError(code, f"{rel}:UTF8") from exc
|
for needle in required:
|
if needle not in text:
|
raise PublisherError(code, f"{rel}:{needle}")
|
|
|
def _read_ledger(data: bytes) -> list[dict[str, str]]:
|
try:
|
reader = csv.DictReader(io.StringIO(data.decode("utf-8-sig"), newline=""))
|
if reader.fieldnames != LEDGER_HEADER:
|
raise PublisherError("LEDGER_HEADER", str(reader.fieldnames))
|
rows = list(reader)
|
except PublisherError:
|
raise
|
except Exception as exc:
|
raise PublisherError("LEDGER_PARSE", str(exc)) from exc
|
sequences: set[int] = set()
|
for row in rows:
|
if set(row) != set(LEDGER_HEADER) or any(not isinstance(row[key], str) for key in LEDGER_HEADER):
|
raise PublisherError("LEDGER_ROW_SHAPE", str(row))
|
try:
|
sequence = int(row["event_seq"])
|
attempt_seq = int(row["attempt_seq"])
|
except (TypeError, ValueError) as exc:
|
raise PublisherError("LEDGER_INTEGER", str(row)) from exc
|
if sequence <= 0 or attempt_seq <= 0 or sequence in sequences:
|
raise PublisherError("LEDGER_SEQUENCE_UNIQUE", str(sequence))
|
sequences.add(sequence)
|
return rows
|
|
|
def _derive_sequences(rows: list[dict[str, str]], identity: dict[str, str]) -> tuple[int, int]:
|
next_event = max((int(row["event_seq"]) for row in rows), default=0) + 1
|
matching = [int(row["attempt_seq"]) for row in rows if all(row[key] == identity[key] for key in ("task_id", "case_id", "batch_id", "run_id"))]
|
return next_event, max(matching, default=0) + 1
|
|
|
def _event_rows_for_attempt(rows: list[dict[str, str]], attempt: str) -> list[dict[str, str]]:
|
return sorted((row for row in rows if row["attempt_id"] == attempt), key=lambda row: int(row["event_seq"]))
|
|
|
def _event_chain_digest(rows: list[dict[str, str]]) -> str:
|
payload = {
|
"columns": LEDGER_HEADER,
|
"rows": [[row[column] for column in LEDGER_HEADER] for row in rows],
|
"schema_version": EVENT_CHAIN_SCHEMA,
|
}
|
return _sha_bytes(_canonical_json(payload, newline=False))
|
|
|
@dataclass
|
class Plan:
|
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[ArtifactRow, ...]
|
prior_rows: tuple[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[ArtifactRow, ...]
|
manifest_row: ArtifactRow
|
result_row: FormalRow
|
prior_release_set: str
|
lifecycle: str
|
|
|
def _verify_prior_manifest(plan_root: Path, prior: dict[str, Any], rows: tuple[FormalRow, ...]) -> tuple[bytes, str]:
|
_expect_keys(prior, {"release_id", "manifest_formal_relative_path", "manifest_bytes", "manifest_sha256", "manifest_self_formal_relative_path", "release_set_sha256", "rows"}, "expected.prior")
|
_need_sha(prior["release_id"], "expected.prior.release_id", lower=True)
|
manifest_rel = _safe_relative(prior["manifest_formal_relative_path"], "expected.prior.manifest_formal_relative_path")
|
self_rel = _safe_relative(prior["manifest_self_formal_relative_path"], "expected.prior.manifest_self_formal_relative_path")
|
if manifest_rel.casefold() != self_rel.casefold():
|
raise PublisherError("PRIOR_MANIFEST_SELF_PATH", f"{manifest_rel}!={self_rel}")
|
data = _read_bytes(_inside(plan_root, manifest_rel))
|
expected_id = (_need_int(prior["manifest_bytes"], "expected.prior.manifest_bytes", positive=True), _need_sha(prior["manifest_sha256"], "expected.prior.manifest_sha256"))
|
if (len(data), _sha_bytes(data)) != expected_id:
|
raise PublisherError("PRIOR_MANIFEST_IDENTITY", manifest_rel)
|
actual = {_manifest_tuple(row) for row in _manifest_rows(data, candidate=False)}
|
expected = {_formal_tuple(row) for row in rows}
|
if actual != expected or len(actual) != len(rows):
|
raise PublisherError("PRIOR_MANIFEST_COVERAGE", f"actual={len(actual)};expected={len(expected)}")
|
release_set = _release_set(data, self_rel)
|
if release_set != _need_sha(prior["release_set_sha256"], "expected.prior.release_set_sha256"):
|
raise PublisherError("PRIOR_RELEASE_SET", release_set)
|
return data, release_set
|
|
|
def _validate_config(config: dict[str, Any], config_path: Path, config_data: bytes) -> Plan:
|
_expect_keys(config, {"schema_version", "identity", "roots", "expected", "commit", "test_control"}, "root")
|
if config["schema_version"] != CONFIG_SCHEMA:
|
raise PublisherError("CONFIG_SCHEMA", str(config["schema_version"]))
|
for name in ("identity", "roots", "expected", "commit", "test_control"):
|
if not isinstance(config[name], dict):
|
raise 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"}
|
_expect_keys(identity, identity_keys, "identity")
|
for key in identity_keys:
|
_need_string(identity[key], f"identity.{key}")
|
if not identity["canonical_audit_id"].startswith("AUDIT-"):
|
raise PublisherError("AUDIT_ID_DOMAIN", identity["canonical_audit_id"])
|
if not identity["review_handoff_id"].startswith("HANDOFF-"):
|
raise PublisherError("HANDOFF_ID_DOMAIN", identity["review_handoff_id"])
|
if identity["canonical_audit_id"] == identity["review_handoff_id"]:
|
raise PublisherError("AUDIT_HANDOFF_AMBIGUOUS", "equal identities")
|
_need_sha(identity["release_id"], "identity.release_id", lower=True)
|
_need_sha(identity["prior_release_id"], "identity.prior_release_id", lower=True)
|
_need_sha(identity["candidate_release_set_sha256"], "identity.candidate_release_set_sha256")
|
preimage = "".join(f"{key}={identity[key]}\n" for key in ("industry", "task_id", "case_id", "batch_id", "run_id", "canonical_audit_id"))
|
if hashlib.sha256(preimage.encode("utf-8")).hexdigest() != identity["release_id"]:
|
raise PublisherError("RELEASE_PREIMAGE", "release id mismatch")
|
|
roots = config["roots"]
|
root_keys = {"operation_root", "resolved_root", "volume_identity", "candidate_root", "release_store_root", "case_current_index_path", "result_current_index_path", "ledger_path", "lock_path", "attempt_receipt_root", "history_root", "current_state_relative_path"}
|
_expect_keys(roots, root_keys, "roots")
|
operation_root = Path(os.path.abspath(_need_string(roots["operation_root"], "roots.operation_root")))
|
_physical_chain(operation_root, final_kind="dir")
|
_ordinary_dir(operation_root)
|
if _norm(operation_root) != _norm(_need_string(roots["resolved_root"], "roots.resolved_root")):
|
raise PublisherError("ROOT_RESOLUTION", str(operation_root))
|
volume = _need_string(roots["volume_identity"], "roots.volume_identity").upper()
|
if _volume(operation_root).upper() != volume:
|
raise PublisherError("VOLUME_IDENTITY", volume)
|
if os.path.commonpath([_norm(operation_root), _norm(config_path)]) != _norm(operation_root):
|
raise PublisherError("CONFIG_OUTSIDE_ROOT", str(config_path))
|
|
paths: dict[str, Path] = {}
|
for key in ("candidate_root", "release_store_root", "case_current_index_path", "result_current_index_path", "ledger_path", "lock_path", "attempt_receipt_root", "history_root"):
|
paths[key] = _inside(operation_root, _safe_relative(roots[key], f"roots.{key}"))
|
history_state_rel = _safe_relative(roots["current_state_relative_path"], "roots.current_state_relative_path")
|
paths["current_state_path"] = _inside(paths["history_root"], history_state_rel)
|
for key in ("candidate_root", "release_store_root", "history_root"):
|
_validate_parent_chain(operation_root, paths[key], final_kind="dir")
|
_ordinary_dir(paths[key])
|
if _volume(paths[key]).upper() != volume:
|
raise PublisherError("VOLUME_DRIFT", key)
|
for key in ("case_current_index_path", "result_current_index_path", "ledger_path", "current_state_path"):
|
_validate_parent_chain(operation_root, paths[key], final_kind="file")
|
_ordinary_file(paths[key])
|
for key in ("lock_path", "attempt_receipt_root"):
|
_validate_parent_chain(operation_root, paths[key])
|
_validate_parent_chain(operation_root, config_path, final_kind="file")
|
|
commit = config["commit"]
|
_expect_keys(commit, {"strategy", "target_release_relative_path", "staging_relative_path"}, "commit")
|
if commit["strategy"] != COMMIT_STRATEGY:
|
raise PublisherError("COMMIT_STRATEGY", str(commit["strategy"]))
|
target_rel = _safe_relative(commit["target_release_relative_path"], "commit.target_release_relative_path")
|
stage_rel = _safe_relative(commit["staging_relative_path"], "commit.staging_relative_path")
|
paths["target_release"] = _inside(operation_root, target_rel)
|
paths["staging"] = _inside(operation_root, stage_rel)
|
paths["snapshot"] = paths["attempt_receipt_root"] / "prior_snapshot"
|
for key in ("target_release", "staging", "snapshot"):
|
_validate_parent_chain(operation_root, paths[key])
|
if os.path.commonpath([_norm(paths["release_store_root"]), _norm(paths["target_release"])]) != _norm(paths["release_store_root"]):
|
raise PublisherError("TARGET_RELEASE_STORE", target_rel)
|
if os.path.commonpath([_norm(paths["release_store_root"]), _norm(paths["staging"])]) != _norm(paths["release_store_root"]):
|
raise PublisherError("STAGING_RELEASE_STORE", stage_rel)
|
path_graph = {key: paths[key] for key in ("candidate_root", "release_store_root", "case_current_index_path", "result_current_index_path", "ledger_path", "lock_path", "attempt_receipt_root", "history_root", "target_release", "staging", "snapshot")}
|
path_graph["config_path"] = config_path
|
_validate_path_graph(path_graph, {frozenset(("release_store_root", "target_release")), frozenset(("release_store_root", "staging")), frozenset(("attempt_receipt_root", "snapshot"))})
|
if _exists(paths["lock_path"]):
|
raise PublisherError("LOCK_OCCUPIED", str(paths["lock_path"]))
|
|
expected = config["expected"]
|
_expect_keys(expected, {"ledger", "candidate", "prior", "history", "case_current_index", "result_current_index", "current_state"}, "expected")
|
candidate = expected["candidate"]
|
if not isinstance(candidate, dict):
|
raise 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"}
|
_expect_keys(candidate, candidate_keys, "expected.candidate")
|
candidate_rows = _parse_artifact_rows(candidate["rows"], "expected.candidate.rows")
|
target_rows = tuple(row for row in candidate_rows if row.commit_role == "RELEASE_MEMBER")
|
index_rows = tuple(row for row in candidate_rows if row.commit_role == "CASE_CURRENT_INDEX")
|
if len(index_rows) != 1 or not target_rows:
|
raise PublisherError("CANDIDATE_COMMIT_ROLES", f"release={len(target_rows)};index={len(index_rows)}")
|
candidate_index = index_rows[0]
|
case_formal = os.path.relpath(paths["case_current_index_path"], operation_root).replace("\\", "/")
|
if candidate_index.formal_relative_path.casefold() != case_formal.casefold():
|
raise PublisherError("CASE_INDEX_FORMAL_PATH", candidate_index.formal_relative_path)
|
target_prefix = target_rel.rstrip("/") + "/"
|
for row in target_rows:
|
if not row.formal_relative_path.casefold().startswith(target_prefix.casefold()):
|
raise PublisherError("TARGET_MEMBER_PATH", row.formal_relative_path)
|
formal_tail = [row.formal_relative_path[len(target_prefix):].casefold() for row in target_rows]
|
if len(formal_tail) != len(set(formal_tail)) or any(not tail for tail in formal_tail):
|
raise PublisherError("TARGET_MEMBER_DUPLICATE", str(formal_tail))
|
|
manifest_rel = _safe_relative(candidate["manifest_relative_path"], "expected.candidate.manifest_relative_path")
|
manifest_data = _read_bytes(_inside(paths["candidate_root"], manifest_rel))
|
if (len(manifest_data), _sha_bytes(manifest_data)) != (_need_int(candidate["manifest_bytes"], "expected.candidate.manifest_bytes", positive=True), _need_sha(candidate["manifest_sha256"], "expected.candidate.manifest_sha256")):
|
raise PublisherError("CANDIDATE_MANIFEST_IDENTITY", manifest_rel)
|
manifest_actual = []
|
for row in _manifest_rows(manifest_data, candidate=True):
|
manifest_actual.append((row["member_id"], row["candidate_relative_path"].removeprefix("candidate/").replace("\\", "/"), row["formal_relative_path"].replace("\\", "/"), row["artifact_type"], int(row["bytes"]), row["sha256"]))
|
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 PublisherError("CANDIDATE_MANIFEST_BINDING", "ordered rows differ")
|
exact_candidate = {manifest_rel} | {row.relative_path for row in candidate_rows}
|
actual_candidate = set(_walk_files(paths["candidate_root"]))
|
if actual_candidate != exact_candidate:
|
raise PublisherError("CANDIDATE_EXACT_SET", f"missing={sorted(exact_candidate-actual_candidate)};extra={sorted(actual_candidate-exact_candidate)}")
|
for row in candidate_rows:
|
_verify_identity(_inside(paths["candidate_root"], row.relative_path), row.bytes, row.sha256, "CANDIDATE_IDENTITY")
|
|
current_manifest_rel = _safe_relative(candidate["current_manifest_relative_path"], "expected.candidate.current_manifest_relative_path")
|
manifest_matches = [row for row in target_rows if row.relative_path.casefold() == current_manifest_rel.casefold()]
|
if len(manifest_matches) != 1:
|
raise PublisherError("CURRENT_MANIFEST_ROW", str(len(manifest_matches)))
|
manifest_row = manifest_matches[0]
|
self_formal = _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 PublisherError("CURRENT_MANIFEST_SELF_PATH", f"{self_formal}!={manifest_row.formal_relative_path}")
|
current_manifest_data = _read_bytes(_inside(paths["candidate_root"], current_manifest_rel))
|
if (len(current_manifest_data), _sha_bytes(current_manifest_data)) != (_need_int(candidate["current_manifest_bytes"], "expected.candidate.current_manifest_bytes", positive=True), _need_sha(candidate["current_manifest_sha256"], "expected.candidate.current_manifest_sha256")):
|
raise PublisherError("CURRENT_MANIFEST_IDENTITY", current_manifest_rel)
|
|
result_expected = expected["result_current_index"]
|
if not isinstance(result_expected, dict):
|
raise PublisherError("CONFIG_TYPE", "expected.result_current_index")
|
_expect_keys(result_expected, {"member_id", "formal_relative_path", "artifact_type", "bytes", "sha256", "required_utf8_substrings"}, "expected.result_current_index")
|
result_row = FormalRow(
|
_need_string(result_expected["member_id"], "expected.result_current_index.member_id"),
|
_safe_relative(result_expected["formal_relative_path"], "expected.result_current_index.formal_relative_path"),
|
_need_string(result_expected["artifact_type"], "expected.result_current_index.artifact_type"),
|
_need_int(result_expected["bytes"], "expected.result_current_index.bytes", positive=True),
|
_need_sha(result_expected["sha256"], "expected.result_current_index.sha256"),
|
)
|
result_formal = os.path.relpath(paths["result_current_index_path"], operation_root).replace("\\", "/")
|
if result_row.formal_relative_path.casefold() != result_formal.casefold():
|
raise PublisherError("RESULT_INDEX_FORMAL_PATH", result_row.formal_relative_path)
|
_verify_identity(paths["result_current_index_path"], result_row.bytes, result_row.sha256, "RESULT_INDEX_IDENTITY")
|
required_result = result_expected["required_utf8_substrings"]
|
if not isinstance(required_result, list) or not required_result:
|
raise PublisherError("CONFIG_TYPE", "expected.result_current_index.required_utf8_substrings")
|
result_text = _read_bytes(paths["result_current_index_path"]).decode("utf-8")
|
for needle in required_result:
|
if _need_string(needle, "expected.result_current_index.required_utf8_substrings") not in result_text:
|
raise PublisherError("RESULT_INDEX_LINK", needle)
|
|
current_expected_set = {_artifact_formal_tuple(row) for row in candidate_rows if row is not manifest_row} | {_formal_tuple(result_row)}
|
current_actual_rows = _manifest_rows(current_manifest_data, candidate=False)
|
current_actual_set = {_manifest_tuple(row) for row in current_actual_rows}
|
if current_actual_set != current_expected_set or len(current_actual_rows) != len(current_expected_set):
|
raise PublisherError("CURRENT_MANIFEST_COVERAGE", f"actual={len(current_actual_set)};expected={len(current_expected_set)}")
|
candidate_set = _release_set(current_manifest_data, self_formal)
|
if candidate_set != identity["candidate_release_set_sha256"]:
|
raise PublisherError("CANDIDATE_RELEASE_SET", candidate_set)
|
|
link_checks = _parse_checks(candidate["link_checks"], "expected.candidate.link_checks")
|
evidence_checks = _parse_checks(candidate["evidence_checks"], "expected.candidate.evidence_checks")
|
_verify_checks(paths["candidate_root"], link_checks, "LINK_CHECK_MISSING")
|
_verify_checks(paths["candidate_root"], evidence_checks, "EVIDENCE_CHECK_MISSING")
|
|
prior = expected["prior"]
|
if not isinstance(prior, dict):
|
raise PublisherError("CONFIG_TYPE", "expected.prior")
|
prior_rows = _parse_formal_rows(prior.get("rows"), "expected.prior.rows")
|
prior_paths = {row.member_id: _inside(operation_root, row.formal_relative_path) for row in prior_rows}
|
for member, path in prior_paths.items():
|
_validate_parent_chain(operation_root, path, final_kind="file")
|
_ordinary_file(path)
|
required_prior_paths = {_norm(paths["case_current_index_path"]), _norm(paths["result_current_index_path"])}
|
if not required_prior_paths.issubset({_norm(path) for path in prior_paths.values()}):
|
raise PublisherError("PRIOR_ENTRY_COVERAGE", "cases/result current entries required")
|
protected = {
|
"candidate_root": paths["candidate_root"], "release_store_root": paths["release_store_root"],
|
"target_release": paths["target_release"], "staging": paths["staging"],
|
"attempt_receipt_root": paths["attempt_receipt_root"], "history_root": paths["history_root"],
|
"ledger_path": paths["ledger_path"], "lock_path": paths["lock_path"], "config_path": config_path,
|
}
|
prior_items = list(prior_paths.items())
|
for index, (member, path) in enumerate(prior_items):
|
for other_member, other_path in prior_items[index + 1:]:
|
if _overlap(path, other_path):
|
raise PublisherError("PRIOR_PATH_OVERLAP", f"{member}<->{other_member}")
|
if _norm(path) not in required_prior_paths:
|
for protected_name, protected_path in protected.items():
|
if _overlap(path, protected_path):
|
raise PublisherError("PRIOR_PATH_OVERLAP", f"{member}<->{protected_name}")
|
prior_manifest_path = _inside(operation_root, _safe_relative(prior["manifest_formal_relative_path"], "expected.prior.manifest_formal_relative_path"))
|
_validate_parent_chain(operation_root, prior_manifest_path, final_kind="file")
|
for protected_name, protected_path in protected.items():
|
if _overlap(prior_manifest_path, protected_path):
|
raise PublisherError("PRIOR_MANIFEST_PATH_OVERLAP", protected_name)
|
if any(_overlap(prior_manifest_path, path) for path in prior_paths.values()):
|
raise PublisherError("PRIOR_MANIFEST_PATH_OVERLAP", "prior row")
|
_, prior_release_set = _verify_prior_manifest(operation_root, prior, prior_rows)
|
if prior["release_id"] != identity["prior_release_id"]:
|
raise PublisherError("PRIOR_RELEASE_ID", str(prior["release_id"]))
|
|
case_expected = expected["case_current_index"]
|
if not isinstance(case_expected, dict):
|
raise PublisherError("CONFIG_TYPE", "expected.case_current_index")
|
_expect_keys(case_expected, {"bytes", "sha256", "required_utf8_substrings"}, "expected.case_current_index")
|
prior_index_id = (_need_int(case_expected["bytes"], "expected.case_current_index.bytes", positive=True), _need_sha(case_expected["sha256"], "expected.case_current_index.sha256"))
|
required_case = case_expected["required_utf8_substrings"]
|
if not isinstance(required_case, list) or not required_case:
|
raise PublisherError("CONFIG_TYPE", "expected.case_current_index.required_utf8_substrings")
|
prior_index_data: bytes
|
current_actual_id = _identity(paths["case_current_index_path"])
|
candidate_index_id = (candidate_index.bytes, candidate_index.sha256)
|
receipt_exists = _exists(paths["attempt_receipt_root"])
|
target_exists = _exists(paths["target_release"])
|
if current_actual_id == prior_index_id and not receipt_exists and not target_exists:
|
lifecycle = "FRESH"
|
prior_index_data = _read_bytes(paths["case_current_index_path"])
|
for row in prior_rows:
|
_verify_identity(_inside(operation_root, row.formal_relative_path), row.bytes, row.sha256, "PRIOR_IDENTITY")
|
elif current_actual_id == candidate_index_id and receipt_exists and target_exists:
|
lifecycle = "REPLAY"
|
snapshot_index = _inside(paths["snapshot"], case_formal)
|
_verify_identity(snapshot_index, *prior_index_id, "PRIOR_SNAPSHOT_INDEX")
|
prior_index_data = _read_bytes(snapshot_index)
|
elif current_actual_id == prior_index_id and receipt_exists:
|
lifecycle = "TERMINAL_PRIOR"
|
prior_index_data = _read_bytes(paths["case_current_index_path"])
|
else:
|
raise PublisherError("CURRENT_INDEX_STATE", f"current={current_actual_id};receipt={receipt_exists};target={target_exists}")
|
try:
|
prior_text = prior_index_data.decode("utf-8")
|
except UnicodeDecodeError as exc:
|
raise PublisherError("CASE_INDEX_UTF8", "prior") from exc
|
for needle in required_case:
|
if _need_string(needle, "expected.case_current_index.required_utf8_substrings") not in prior_text:
|
raise PublisherError("CASE_INDEX_LINK", needle)
|
|
history = expected["history"]
|
if not isinstance(history, dict):
|
raise PublisherError("CONFIG_TYPE", "expected.history")
|
_expect_keys(history, {"rows", "minimum_long_paths", "long_path_threshold"}, "expected.history")
|
history_rows = _parse_history_rows(history["rows"], "expected.history.rows")
|
threshold = _need_int(history["long_path_threshold"], "expected.history.long_path_threshold", positive=True)
|
minimum = _need_int(history["minimum_long_paths"], "expected.history.minimum_long_paths")
|
_verify_history(paths["history_root"], history_rows, threshold, minimum)
|
|
state_expected = expected["current_state"]
|
if not isinstance(state_expected, dict):
|
raise PublisherError("CONFIG_TYPE", "expected.current_state")
|
_expect_keys(state_expected, {"bytes", "sha256", "semantic_field", "semantic_value", "status_field", "status_value", "exit_code_field", "exit_code_value"}, "expected.current_state")
|
state_id = (_need_int(state_expected["bytes"], "expected.current_state.bytes", positive=True), _need_sha(state_expected["sha256"], "expected.current_state.sha256"))
|
_verify_identity(paths["current_state_path"], *state_id, "CURRENT_STATE_IDENTITY")
|
try:
|
state_value = json.loads(_read_bytes(paths["current_state_path"]).decode("utf-8"))
|
except Exception as exc:
|
raise 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 = _need_string(state_expected[field_key], f"expected.current_state.{field_key}")
|
if state_value.get(field) != state_expected[value_key]:
|
raise PublisherError("CURRENT_STATE_SEMANTIC", field)
|
|
ledger_expected = expected["ledger"]
|
if not isinstance(ledger_expected, dict):
|
raise PublisherError("CONFIG_TYPE", "expected.ledger")
|
_expect_keys(ledger_expected, {"bytes", "sha256", "next_event_seq", "next_attempt_seq", "prior_attempt_id", "prior_terminal_state"}, "expected.ledger")
|
base_size = _need_int(ledger_expected["bytes"], "expected.ledger.bytes", positive=True)
|
base_sha = _need_sha(ledger_expected["sha256"], "expected.ledger.sha256")
|
next_event = _need_int(ledger_expected["next_event_seq"], "expected.ledger.next_event_seq", positive=True)
|
next_attempt = _need_int(ledger_expected["next_attempt_seq"], "expected.ledger.next_attempt_seq", positive=True)
|
prior_attempt = _need_string(ledger_expected["prior_attempt_id"], "expected.ledger.prior_attempt_id")
|
prior_terminal = _need_string(ledger_expected["prior_terminal_state"], "expected.ledger.prior_terminal_state")
|
ledger_actual = _read_bytes(paths["ledger_path"])
|
if lifecycle == "FRESH":
|
if (len(ledger_actual), _sha_bytes(ledger_actual)) != (base_size, base_sha):
|
raise PublisherError("LEDGER_IDENTITY", str(paths["ledger_path"]))
|
ledger_base = ledger_actual
|
else:
|
if len(ledger_actual) < base_size:
|
raise PublisherError("LEDGER_BASE_PREFIX", "short")
|
ledger_base = ledger_actual[:base_size]
|
if _sha_bytes(ledger_base) != base_sha:
|
raise PublisherError("LEDGER_BASE_PREFIX", "hash")
|
base_rows = _read_ledger(ledger_base)
|
derived_event, derived_attempt = _derive_sequences(base_rows, identity)
|
if derived_event != next_event:
|
raise PublisherError("LEDGER_NEXT_EVENT", f"{derived_event}!={next_event}")
|
if derived_attempt != next_attempt:
|
raise PublisherError("LEDGER_NEXT_ATTEMPT", f"{derived_attempt}!={next_attempt}")
|
prior_matches = [row for row in base_rows if row["attempt_id"] == prior_attempt]
|
if len(prior_matches) == 0 or prior_matches[-1]["state"] != prior_terminal:
|
raise PublisherError("LEDGER_PRIOR_TERMINAL", prior_attempt)
|
|
test_control = config["test_control"]
|
_expect_keys(test_control, {"environment", "fault", "race_marker_relative_path"}, "test_control")
|
environment = _need_string(test_control["environment"], "test_control.environment")
|
fault = _need_string(test_control["fault"], "test_control.fault")
|
marker = test_control["race_marker_relative_path"]
|
if not isinstance(marker, str):
|
raise PublisherError("CONFIG_TYPE", "test_control.race_marker_relative_path")
|
allowed_faults = {"NONE", "POSTCOMMIT_READBACK_FAIL", "EXTRA_TARGET_AFTER_RENAME", "PAUSE_AFTER_LOCK"}
|
if environment == "PRODUCTION" and (fault != "NONE" or marker):
|
raise PublisherError("TEST_CONTROL_PRODUCTION", fault)
|
if environment not in ("PRODUCTION", "ISOLATED_TEST") or fault not in allowed_faults:
|
raise PublisherError("CONFIG_ENUM", "test_control")
|
if fault == "PAUSE_AFTER_LOCK" and not marker:
|
raise PublisherError("CONFIG_TYPE", "race marker required")
|
if marker:
|
paths["race_marker"] = _inside(operation_root, _safe_relative(marker, "test_control.race_marker_relative_path"))
|
|
return Plan(
|
config=config, config_path=config_path, config_data=config_data, config_sha256=_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=_read_bytes(_inside(paths["candidate_root"], candidate_index.relative_path)),
|
target_release_relative=target_rel, target_rows=target_rows, manifest_row=manifest_row,
|
result_row=result_row, prior_release_set=prior_release_set, lifecycle=lifecycle,
|
)
|
|
|
def _copy_file(source: Path, target: Path) -> None:
|
os.makedirs(_long(target.parent), exist_ok=True)
|
_ordinary_dir(target.parent)
|
if _exists(target):
|
raise PublisherError("TARGET_EXISTS", str(target), exit_code=20)
|
data = _read_bytes(source)
|
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0)
|
fd = os.open(_long(target), flags, 0o600)
|
try:
|
offset = 0
|
while offset < len(data):
|
offset += os.write(fd, data[offset:])
|
os.fsync(fd)
|
finally:
|
os.close(fd)
|
if _identity(target) != (len(data), _sha_bytes(data)):
|
raise PublisherError("COPY_READBACK", str(target), exit_code=20)
|
|
|
def _create_exclusive(path: Path, data: bytes) -> None:
|
if not _exists(path.parent):
|
os.makedirs(_long(path.parent), exist_ok=False)
|
_ordinary_dir(path.parent)
|
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0)
|
try:
|
fd = os.open(_long(path), flags, 0o600)
|
except FileExistsError as exc:
|
raise PublisherError("TARGET_EXISTS", str(path), exit_code=20) from exc
|
try:
|
offset = 0
|
while offset < len(data):
|
offset += os.write(fd, data[offset:])
|
os.fsync(fd)
|
finally:
|
os.close(fd)
|
|
|
def _atomic_replace(path: Path, data: bytes, attempt: str) -> None:
|
temp = path.with_name(f".{path.name}.{attempt}.tmp")
|
if _exists(temp):
|
raise PublisherError("TEMP_EXISTS", str(temp), exit_code=20)
|
_create_exclusive(temp, data)
|
try:
|
os.replace(_long(temp), _long(path))
|
except Exception:
|
if _exists(temp):
|
os.unlink(_long(temp))
|
raise
|
|
|
def _target_relative(plan: Plan, row: ArtifactRow) -> str:
|
prefix = plan.target_release_relative.rstrip("/") + "/"
|
return row.formal_relative_path[len(prefix):]
|
|
|
def _verify_target(plan: Plan) -> None:
|
expected = {_target_relative(plan, row) for row in plan.target_rows}
|
actual = set(_walk_files(plan.paths["target_release"]))
|
if actual != expected:
|
raise PublisherError("TARGET_EXACT_SET", f"missing={sorted(expected-actual)};extra={sorted(actual-expected)}", exit_code=20)
|
for row in plan.target_rows:
|
_verify_identity(_inside(plan.paths["target_release"], _target_relative(plan, row)), row.bytes, row.sha256, "TARGET_IDENTITY")
|
manifest_path = _inside(plan.paths["target_release"], _target_relative(plan, plan.manifest_row))
|
data = _read_bytes(manifest_path)
|
expected_set = {_artifact_formal_tuple(row) for row in plan.candidate_rows if row is not plan.manifest_row} | {_formal_tuple(plan.result_row)}
|
actual_rows = _manifest_rows(data, candidate=False)
|
actual_set = {_manifest_tuple(row) for row in actual_rows}
|
if actual_set != expected_set or len(actual_rows) != len(expected_set):
|
raise PublisherError("TARGET_MANIFEST_COVERAGE", f"actual={len(actual_set)};expected={len(expected_set)}", exit_code=20)
|
if _release_set(data, plan.manifest_row.formal_relative_path) != plan.identity["candidate_release_set_sha256"]:
|
raise PublisherError("TARGET_RELEASE_SET", "mismatch", exit_code=20)
|
|
|
def _event_line(plan: Plan, sequence: int, state: str, receipt: str, process_identity: str, event_time: str) -> bytes:
|
row = [
|
plan.identity["release_id"], str(sequence), state, plan.identity["prior_release_id"],
|
plan.prior_release_set, plan.identity["candidate_release_set_sha256"], plan.identity["task_id"],
|
plan.identity["case_id"], plan.identity["batch_id"], plan.identity["run_id"],
|
plan.identity["canonical_audit_id"], plan.identity["operator"], process_identity,
|
event_time, receipt, plan.identity["attempt_id"], str(plan.next_attempt_seq),
|
]
|
stream = io.StringIO(newline="")
|
csv.writer(stream, lineterminator="\n").writerow(row)
|
return stream.getvalue().encode("utf-8")
|
|
|
def _append_event(plan: Plan, state: str, offset: int, receipt: str, process_identity: str) -> None:
|
rows = _read_ledger(_read_bytes(plan.paths["ledger_path"]))
|
attempt_rows = _event_rows_for_attempt(rows, plan.identity["attempt_id"])
|
current = datetime.now(timezone.utc)
|
if attempt_rows:
|
previous = _parse_strict_utc(attempt_rows[-1]["event_time"], "EVENT_TIME_EXISTING")
|
if current <= previous:
|
current = previous + timedelta(microseconds=1)
|
event_time = current.isoformat(timespec="microseconds").replace("+00:00", "Z")
|
with open(_long(plan.paths["ledger_path"]), "ab", buffering=0) as handle:
|
handle.write(_event_line(plan, plan.next_event_seq + offset, state, receipt, process_identity, event_time))
|
os.fsync(handle.fileno())
|
|
|
def _process_identity() -> str:
|
return f"PID-{os.getpid()}-{uuid.uuid4().hex.upper()}"
|
|
|
def _receipt_relative(plan: Plan, name: str) -> str:
|
return os.path.relpath(plan.paths["attempt_receipt_root"] / name, plan.operation_root).replace("\\", "/")
|
|
|
def _locked_revalidate(plan: Plan) -> None:
|
if _read_bytes(plan.config_path) != plan.config_data:
|
raise PublisherError("CONFIG_IDENTITY_LOCKED", "changed after preflight")
|
_physical_chain(plan.operation_root, final_kind="dir")
|
if _volume(plan.operation_root).upper() != plan.config["roots"]["volume_identity"].upper():
|
raise PublisherError("VOLUME_IDENTITY_LOCKED", "changed after preflight")
|
ledger = _read_bytes(plan.paths["ledger_path"])
|
if ledger != plan.ledger_base_data:
|
raise PublisherError("LEDGER_IDENTITY_LOCKED", "changed after preflight")
|
rows = _read_ledger(ledger)
|
if _derive_sequences(rows, plan.identity) != (plan.next_event_seq, plan.next_attempt_seq):
|
raise PublisherError("LEDGER_SEQUENCE_LOCKED", "rederived sequence mismatch")
|
if _read_bytes(plan.paths["case_current_index_path"]) != plan.prior_index_data:
|
raise PublisherError("CURRENT_INDEX_LOCKED", "changed after preflight")
|
_verify_identity(plan.paths["result_current_index_path"], plan.result_row.bytes, plan.result_row.sha256, "RESULT_INDEX_LOCKED")
|
for row in plan.prior_rows:
|
_verify_identity(_inside(plan.operation_root, row.formal_relative_path), row.bytes, row.sha256, "PRIOR_LOCKED")
|
prior = plan.config["expected"]["prior"]
|
prior_manifest = _inside(plan.operation_root, prior["manifest_formal_relative_path"])
|
_verify_identity(prior_manifest, prior["manifest_bytes"], prior["manifest_sha256"], "PRIOR_MANIFEST_LOCKED")
|
if _release_set(_read_bytes(prior_manifest), prior["manifest_self_formal_relative_path"]) != plan.prior_release_set:
|
raise PublisherError("PRIOR_RELEASE_SET_LOCKED", "mismatch")
|
_verify_history(plan.paths["history_root"], plan.history_rows, plan.config["expected"]["history"]["long_path_threshold"], plan.config["expected"]["history"]["minimum_long_paths"])
|
for row in plan.candidate_rows:
|
_verify_identity(_inside(plan.paths["candidate_root"], row.relative_path), row.bytes, row.sha256, "CANDIDATE_LOCKED")
|
candidate = plan.config["expected"]["candidate"]
|
candidate_manifest = _inside(plan.paths["candidate_root"], candidate["manifest_relative_path"])
|
_verify_identity(candidate_manifest, candidate["manifest_bytes"], candidate["manifest_sha256"], "CANDIDATE_MANIFEST_LOCKED")
|
expected_candidate = {candidate["manifest_relative_path"]} | {row.relative_path for row in plan.candidate_rows}
|
actual_candidate = set(_walk_files(plan.paths["candidate_root"]))
|
if actual_candidate != expected_candidate:
|
raise PublisherError("CANDIDATE_EXACT_SET_LOCKED", f"missing={sorted(expected_candidate-actual_candidate)};extra={sorted(actual_candidate-expected_candidate)}")
|
_verify_checks(plan.paths["candidate_root"], plan.link_checks, "LINK_CHECK_LOCKED")
|
_verify_checks(plan.paths["candidate_root"], plan.evidence_checks, "EVIDENCE_CHECK_LOCKED")
|
if _exists(plan.paths["target_release"]) or _exists(plan.paths["staging"]) or _exists(plan.paths["attempt_receipt_root"]):
|
raise PublisherError("TARGET_APPEARED_LOCKED", "target/staging/receipt")
|
|
|
def _snapshot_prior(plan: Plan) -> None:
|
for row in plan.prior_rows:
|
source = _inside(plan.operation_root, row.formal_relative_path)
|
target = _inside(plan.paths["snapshot"], row.formal_relative_path)
|
_copy_file(source, target)
|
manifest_rel = plan.config["expected"]["prior"]["manifest_formal_relative_path"]
|
_copy_file(_inside(plan.operation_root, manifest_rel), _inside(plan.paths["snapshot"], manifest_rel))
|
expected = {row.formal_relative_path for row in plan.prior_rows} | {manifest_rel}
|
actual = set(_walk_files(plan.paths["snapshot"]))
|
if actual != expected:
|
raise PublisherError("SNAPSHOT_EXACT_SET", f"missing={sorted(expected-actual)};extra={sorted(actual-expected)}")
|
|
|
def _verify_snapshot(plan: Plan) -> None:
|
manifest_rel = plan.config["expected"]["prior"]["manifest_formal_relative_path"]
|
expected = {row.formal_relative_path for row in plan.prior_rows} | {manifest_rel}
|
actual = set(_walk_files(plan.paths["snapshot"]))
|
if actual != expected:
|
raise PublisherError("REPLAY_SNAPSHOT_EXACT_SET", f"missing={sorted(expected-actual)};extra={sorted(actual-expected)}")
|
for row in plan.prior_rows:
|
_verify_identity(_inside(plan.paths["snapshot"], row.formal_relative_path), row.bytes, row.sha256, "REPLAY_SNAPSHOT_IDENTITY")
|
prior = plan.config["expected"]["prior"]
|
manifest_path = _inside(plan.paths["snapshot"], manifest_rel)
|
_verify_identity(manifest_path, prior["manifest_bytes"], prior["manifest_sha256"], "REPLAY_SNAPSHOT_MANIFEST")
|
if _release_set(_read_bytes(manifest_path), prior["manifest_self_formal_relative_path"]) != plan.prior_release_set:
|
raise PublisherError("REPLAY_SNAPSHOT_RELEASE_SET", "mismatch")
|
|
|
def _terminal(
|
plan: Plan,
|
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]:
|
target_set = _sha_bytes(_canonical_json([
|
{"bytes": row.bytes, "formal_relative_path": row.formal_relative_path, "sha256": row.sha256}
|
for row in sorted(plan.target_rows, key=lambda item: item.formal_relative_path.casefold())
|
], newline=False))
|
return {
|
"schema_version": TERMINAL_SCHEMA,
|
"status": status,
|
"exit_code": exit_code,
|
"task_id": plan.identity["task_id"],
|
"case_id": plan.identity["case_id"],
|
"batch_id": plan.identity["batch_id"],
|
"run_id": plan.identity["run_id"],
|
"attempt_id": plan.identity["attempt_id"],
|
"release_id": plan.identity["release_id"],
|
"prior_release_id": plan.identity["prior_release_id"],
|
"canonical_audit_id": plan.identity["canonical_audit_id"],
|
"review_handoff_id": plan.identity["review_handoff_id"],
|
"config_path": os.path.relpath(plan.config_path, plan.operation_root).replace("\\", "/"),
|
"config_bytes": len(plan.config_data),
|
"config_sha256": plan.config_sha256,
|
"resolved_root": os.path.abspath(plan.operation_root),
|
"volume_identity": _volume(plan.operation_root).upper(),
|
"commit_strategy": COMMIT_STRATEGY,
|
"case_current_prior_bytes": len(plan.prior_index_data),
|
"case_current_prior_sha256": _sha_bytes(plan.prior_index_data),
|
"case_current_candidate_bytes": len(plan.candidate_index_data),
|
"case_current_candidate_sha256": _sha_bytes(plan.candidate_index_data),
|
"result_current_bytes": plan.result_row.bytes,
|
"result_current_sha256": plan.result_row.sha256,
|
"prior_release_set_sha256": plan.prior_release_set,
|
"candidate_release_set_sha256": plan.identity["candidate_release_set_sha256"],
|
"target_release_relative_path": plan.target_release_relative,
|
"target_set_sha256": target_set,
|
"event_seq_start": plan.next_event_seq,
|
"event_seq_end": plan.next_event_seq + len(states) - 1,
|
"event_states": list(states),
|
"attempt_event_chain_schema": EVENT_CHAIN_SCHEMA,
|
"attempt_event_chain_row_count": len(states),
|
"attempt_event_chain_sha256": event_chain_sha256,
|
"event_time_start_utc": event_time_start_utc,
|
"event_time_end_utc": event_time_end_utc,
|
"event_process_identities": list(event_process_identities),
|
"attempt_seq": plan.next_attempt_seq,
|
"process_instance_identity": process_identity,
|
"committed": committed,
|
"rolled_back": rolled_back,
|
"lock_absent_on_return": True,
|
}
|
|
|
def _write_terminal(plan: Plan, value: dict[str, Any]) -> None:
|
_create_exclusive(plan.paths["attempt_receipt_root"] / "terminal.json", _canonical_json(value))
|
|
|
def _expected_event_receipt(plan: Plan, state: str, rollback_terminal_name: str) -> str:
|
if state in ("PREPARED", "VERIFIED"):
|
return _receipt_relative(plan, "prior_snapshot.json")
|
if state in ("COMMITTING", "COMMITTED"):
|
return _receipt_relative(plan, "terminal.json")
|
if state in ("ROLLING_BACK", "ROLLED_BACK"):
|
return _receipt_relative(plan, rollback_terminal_name)
|
raise PublisherError("EVENT_STATE_UNSUPPORTED", state, exit_code=27)
|
|
|
def _validate_attempt_event_chain(
|
plan: Plan,
|
expected_states: tuple[str, ...],
|
process_identities: tuple[str, ...],
|
*,
|
rollback_terminal_name: str = "terminal.json",
|
expected_digest: str | None = None,
|
expected_time_start: str | None = None,
|
expected_time_end: str | None = None,
|
) -> tuple[str, str, str]:
|
if len(process_identities) != len(expected_states) or any(
|
not isinstance(value, str) or not value.startswith("PID-") for value in process_identities
|
):
|
raise PublisherError("EVENT_PROCESS_IDENTITIES", "invalid", exit_code=27)
|
rows = _read_ledger(_read_bytes(plan.paths["ledger_path"]))
|
base_count = len(_read_ledger(plan.ledger_base_data))
|
attempt_rows = rows[base_count:]
|
if len(attempt_rows) != len(expected_states):
|
raise PublisherError("EVENT_CHAIN_ROW_COUNT", f"{len(attempt_rows)}!={len(expected_states)}", exit_code=27)
|
expected_sequences = list(range(plan.next_event_seq, plan.next_event_seq + len(expected_states)))
|
previous_time: datetime | None = None
|
event_times: list[str] = []
|
for index, (row, state, sequence, process_identity) in enumerate(
|
zip(attempt_rows, expected_states, expected_sequences, process_identities, strict=True)
|
):
|
expected = {
|
"release_id": plan.identity["release_id"],
|
"event_seq": str(sequence),
|
"state": state,
|
"prior_release_id": plan.identity["prior_release_id"],
|
"prior_release_set_sha256": plan.prior_release_set,
|
"candidate_release_set_sha256": plan.identity["candidate_release_set_sha256"],
|
"task_id": plan.identity["task_id"],
|
"case_id": plan.identity["case_id"],
|
"batch_id": plan.identity["batch_id"],
|
"run_id": plan.identity["run_id"],
|
"accepted_audit_id": plan.identity["canonical_audit_id"],
|
"operator": plan.identity["operator"],
|
"process_identity": process_identity,
|
"recovery_or_rollback_receipt": _expected_event_receipt(plan, state, rollback_terminal_name),
|
"attempt_id": plan.identity["attempt_id"],
|
"attempt_seq": str(plan.next_attempt_seq),
|
}
|
for column, value in expected.items():
|
if row[column] != value:
|
raise PublisherError("EVENT_CHAIN_FIELD_BINDING", f"row={index};state={state};field={column}", exit_code=27)
|
parsed_time = _parse_strict_utc(row["event_time"], "EVENT_TIME_UTC")
|
if previous_time is not None and parsed_time <= previous_time:
|
raise PublisherError("EVENT_TIME_ORDER", f"row={index};state={state}", exit_code=27)
|
previous_time = parsed_time
|
event_times.append(row["event_time"])
|
digest = _event_chain_digest(attempt_rows)
|
if expected_digest is not None and digest != _need_sha(expected_digest, "terminal.attempt_event_chain_sha256"):
|
raise PublisherError("EVENT_CHAIN_DIGEST", "mismatch", exit_code=27)
|
if expected_time_start is not None and event_times[0] != expected_time_start:
|
raise PublisherError("EVENT_TIME_START_BINDING", "mismatch", exit_code=27)
|
if expected_time_end is not None and event_times[-1] != expected_time_end:
|
raise PublisherError("EVENT_TIME_END_BINDING", "mismatch", exit_code=27)
|
return digest, event_times[0], event_times[-1]
|
|
|
def _validate_event_chain(plan: Plan, terminal: dict[str, Any]) -> None:
|
expected_states = tuple(terminal["event_states"])
|
process_identities = terminal.get("event_process_identities")
|
if not isinstance(process_identities, list):
|
raise PublisherError("REPLAY_EVENT_PROCESS_IDENTITIES", "invalid")
|
digest, start_time, end_time = _validate_attempt_event_chain(
|
plan,
|
expected_states,
|
tuple(process_identities),
|
expected_digest=terminal["attempt_event_chain_sha256"],
|
expected_time_start=terminal["event_time_start_utc"],
|
expected_time_end=terminal["event_time_end_utc"],
|
)
|
if terminal["attempt_event_chain_schema"] != EVENT_CHAIN_SCHEMA:
|
raise PublisherError("REPLAY_EVENT_CHAIN_SCHEMA", str(terminal["attempt_event_chain_schema"]))
|
if terminal["attempt_event_chain_row_count"] != len(expected_states):
|
raise PublisherError("REPLAY_EVENT_CHAIN_COUNT", str(terminal["attempt_event_chain_row_count"]))
|
if terminal["event_seq_end"] != plan.next_event_seq + len(expected_states) - 1 or terminal["event_seq_start"] != plan.next_event_seq:
|
raise PublisherError("REPLAY_TERMINAL_EVENT_RANGE", "mismatch")
|
if (digest, start_time, end_time) != (
|
terminal["attempt_event_chain_sha256"], terminal["event_time_start_utc"], terminal["event_time_end_utc"]
|
):
|
raise PublisherError("REPLAY_EVENT_EVIDENCE", "mismatch")
|
|
|
def _verify_terminal_binding(plan: Plan, terminal: dict[str, Any]) -> None:
|
process_identity = terminal.get("process_instance_identity")
|
if not isinstance(process_identity, str) or not process_identity.startswith("PID-"):
|
raise PublisherError("REPLAY_TERMINAL_PROCESS", "invalid")
|
chain_digest = terminal.get("attempt_event_chain_sha256")
|
start_time = terminal.get("event_time_start_utc")
|
end_time = terminal.get("event_time_end_utc")
|
if not isinstance(chain_digest, str) or not isinstance(start_time, str) or not isinstance(end_time, str):
|
raise PublisherError("REPLAY_TERMINAL_EVENT_EVIDENCE", "invalid")
|
expected = _terminal(
|
plan,
|
"COMMITTED",
|
0,
|
SUCCESS_STATES,
|
process_identity,
|
_need_sha(chain_digest, "terminal.attempt_event_chain_sha256"),
|
start_time,
|
end_time,
|
(process_identity,) * len(SUCCESS_STATES),
|
committed=True,
|
rolled_back=False,
|
)
|
if set(terminal) != set(expected):
|
raise PublisherError("REPLAY_TERMINAL_KEYS", f"missing={sorted(set(expected)-set(terminal))};extra={sorted(set(terminal)-set(expected))}")
|
for key, value in expected.items():
|
if terminal.get(key) != value:
|
raise PublisherError("REPLAY_TERMINAL_BINDING", key)
|
if _exists(plan.paths["lock_path"]):
|
raise PublisherError("REPLAY_LOCK_PRESENT", str(plan.paths["lock_path"]))
|
|
|
def _verify_replay(plan: Plan) -> dict[str, Any]:
|
if plan.lifecycle != "REPLAY":
|
raise PublisherError("REPLAY_LIFECYCLE", plan.lifecycle)
|
if _read_bytes(plan.paths["case_current_index_path"]) != plan.candidate_index_data:
|
raise PublisherError("REPLAY_CURRENT_INDEX", "not candidate")
|
_verify_identity(plan.paths["result_current_index_path"], plan.result_row.bytes, plan.result_row.sha256, "REPLAY_RESULT_INDEX")
|
_verify_target(plan)
|
_verify_snapshot(plan)
|
terminal_path = plan.paths["attempt_receipt_root"] / "terminal.json"
|
try:
|
terminal = json.loads(_read_bytes(terminal_path).decode("utf-8"), object_pairs_hook=_pairs_no_duplicates)
|
except PublisherError:
|
raise
|
except Exception as exc:
|
raise PublisherError("REPLAY_TERMINAL_PARSE", str(exc)) from exc
|
if not isinstance(terminal, dict):
|
raise PublisherError("REPLAY_TERMINAL_TYPE", "not object")
|
_verify_terminal_binding(plan, terminal)
|
_validate_event_chain(plan, terminal)
|
expected_receipt = {
|
"prior_snapshot.json", "terminal.json",
|
*{f"prior_snapshot/{row.formal_relative_path}" for row in plan.prior_rows},
|
f"prior_snapshot/{plan.config['expected']['prior']['manifest_formal_relative_path']}",
|
}
|
actual_receipt = set(_walk_files(plan.paths["attempt_receipt_root"]))
|
if actual_receipt != expected_receipt:
|
raise 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 _rollback(
|
plan: Plan,
|
process_identity: str,
|
offset: int,
|
detail: str,
|
base_states: tuple[str, ...],
|
terminal_name: str,
|
*,
|
base_process_identity: str | None = None,
|
) -> dict[str, Any]:
|
_append_event(plan, "ROLLING_BACK", offset, _receipt_relative(plan, terminal_name), process_identity)
|
_atomic_replace(plan.paths["case_current_index_path"], plan.prior_index_data, plan.identity["attempt_id"] + "-rollback")
|
if _read_bytes(plan.paths["case_current_index_path"]) != plan.prior_index_data:
|
raise PublisherError("ROLLBACK_READBACK", detail, exit_code=27)
|
_append_event(plan, "ROLLED_BACK", offset + 1, _receipt_relative(plan, terminal_name), process_identity)
|
states = base_states + ("ROLLING_BACK", "ROLLED_BACK")
|
base_process = base_process_identity or process_identity
|
event_processes = (base_process,) * len(base_states) + (process_identity, process_identity)
|
chain_digest, start_time, end_time = _validate_attempt_event_chain(
|
plan, states, event_processes, rollback_terminal_name=terminal_name
|
)
|
value = _terminal(
|
plan,
|
"ROLLED_BACK",
|
20,
|
states,
|
process_identity,
|
chain_digest,
|
start_time,
|
end_time,
|
event_processes,
|
committed=False,
|
rolled_back=True,
|
)
|
value["failure_detail"] = detail
|
_create_exclusive(plan.paths["attempt_receipt_root"] / terminal_name, _canonical_json(value))
|
return value
|
|
|
def _recover_invalid_replay(plan: Plan, reason: PublisherError) -> dict[str, Any]:
|
lock_value = {
|
"schema_version": TERMINAL_SCHEMA, "attempt_id": plan.identity["attempt_id"],
|
"config_bytes": len(plan.config_data), "config_sha256": plan.config_sha256,
|
"process_instance_identity": _process_identity(), "acquired_at": _utc_now(),
|
}
|
_create_exclusive(plan.paths["lock_path"], _canonical_json(lock_value))
|
process_identity = lock_value["process_instance_identity"]
|
try:
|
if _read_bytes(plan.paths["case_current_index_path"]) != plan.candidate_index_data:
|
raise PublisherError("RECOVERY_CURRENT_CHANGED", "not candidate", exit_code=27)
|
_verify_snapshot(plan)
|
ledger_rows = _read_ledger(_read_bytes(plan.paths["ledger_path"]))
|
attempt_rows = _event_rows_for_attempt(ledger_rows, plan.identity["attempt_id"])
|
process_values = {row["process_identity"] for row in attempt_rows}
|
if len(process_values) != 1:
|
raise PublisherError("RECOVERY_EVENT_PROCESS_UNKNOWN", str(sorted(process_values)), exit_code=27)
|
process_anchor = next(iter(process_values))
|
digest_anchor: str | None = None
|
start_anchor: str | None = None
|
end_anchor: str | None = None
|
try:
|
terminal = json.loads(
|
_read_bytes(plan.paths["attempt_receipt_root"] / "terminal.json").decode("utf-8"),
|
object_pairs_hook=_pairs_no_duplicates,
|
)
|
if isinstance(terminal, dict):
|
if isinstance(terminal.get("process_instance_identity"), str):
|
process_anchor = terminal["process_instance_identity"]
|
if isinstance(terminal.get("attempt_event_chain_sha256"), str):
|
digest_anchor = terminal["attempt_event_chain_sha256"]
|
if isinstance(terminal.get("event_time_start_utc"), str):
|
start_anchor = terminal["event_time_start_utc"]
|
if isinstance(terminal.get("event_time_end_utc"), str):
|
end_anchor = terminal["event_time_end_utc"]
|
except Exception:
|
# A malformed terminal can still be recovered only when the complete
|
# ledger chain independently validates against the declarative plan.
|
pass
|
_validate_attempt_event_chain(
|
plan,
|
SUCCESS_STATES,
|
(process_anchor,) * len(SUCCESS_STATES),
|
expected_digest=digest_anchor,
|
expected_time_start=start_anchor,
|
expected_time_end=end_anchor,
|
)
|
value = _rollback(
|
plan,
|
process_identity,
|
len(SUCCESS_STATES),
|
f"REPLAY_INVALID:{reason.code}:{reason.detail}",
|
SUCCESS_STATES,
|
"recovery_terminal.json",
|
base_process_identity=process_anchor,
|
)
|
return value
|
finally:
|
if _exists(plan.paths["lock_path"]):
|
os.unlink(_long(plan.paths["lock_path"]))
|
|
|
def _publish(plan: Plan) -> dict[str, Any]:
|
if plan.lifecycle == "TERMINAL_PRIOR":
|
raise PublisherError("TERMINAL_REPEAT_BLOCKED", str(plan.paths["attempt_receipt_root"]))
|
if plan.lifecycle != "FRESH":
|
raise PublisherError("PUBLISH_LIFECYCLE", plan.lifecycle)
|
process_identity = _process_identity()
|
lock_value = {
|
"schema_version": TERMINAL_SCHEMA, "attempt_id": plan.identity["attempt_id"],
|
"config_path": os.path.relpath(plan.config_path, plan.operation_root).replace("\\", "/"),
|
"config_bytes": len(plan.config_data), "config_sha256": plan.config_sha256,
|
"resolved_root": os.path.abspath(plan.operation_root), "volume_identity": _volume(plan.operation_root).upper(),
|
"process_instance_identity": process_identity, "acquired_at": _utc_now(),
|
}
|
_create_exclusive(plan.paths["lock_path"], _canonical_json(lock_value))
|
switched = False
|
try:
|
fault = plan.config["test_control"]["fault"]
|
if fault == "PAUSE_AFTER_LOCK":
|
marker = plan.paths["race_marker"]
|
deadline = time.monotonic() + 5.0
|
while time.monotonic() < deadline and not _exists(marker):
|
time.sleep(0.01)
|
if not _exists(marker):
|
raise PublisherError("TEST_RACE_MARKER_TIMEOUT", str(marker))
|
_locked_revalidate(plan)
|
|
os.makedirs(_long(plan.paths["attempt_receipt_root"]), exist_ok=False)
|
_ordinary_dir(plan.paths["attempt_receipt_root"])
|
_snapshot_prior(plan)
|
snapshot_receipt = {
|
"schema_version": 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,
|
"row_count": len(plan.prior_rows), "created_at": _utc_now(),
|
}
|
_create_exclusive(plan.paths["attempt_receipt_root"] / "prior_snapshot.json", _canonical_json(snapshot_receipt))
|
_append_event(plan, "PREPARED", 0, _receipt_relative(plan, "prior_snapshot.json"), process_identity)
|
|
os.makedirs(_long(plan.paths["staging"]), exist_ok=False)
|
for row in plan.target_rows:
|
_copy_file(_inside(plan.paths["candidate_root"], row.relative_path), _inside(plan.paths["staging"], _target_relative(plan, row)))
|
expected_stage = {_target_relative(plan, row) for row in plan.target_rows}
|
if set(_walk_files(plan.paths["staging"])) != expected_stage:
|
raise PublisherError("STAGING_EXACT_SET", "mismatch", exit_code=20)
|
for row in plan.target_rows:
|
_verify_identity(_inside(plan.paths["staging"], _target_relative(plan, row)), row.bytes, row.sha256, "STAGING_IDENTITY")
|
_append_event(plan, "VERIFIED", 1, _receipt_relative(plan, "prior_snapshot.json"), process_identity)
|
_append_event(plan, "COMMITTING", 2, _receipt_relative(plan, "terminal.json"), process_identity)
|
|
os.rename(_long(plan.paths["staging"]), _long(plan.paths["target_release"]))
|
if fault == "EXTRA_TARGET_AFTER_RENAME":
|
_create_exclusive(plan.paths["target_release"] / "UNDECLARED.txt", b"attack\n")
|
_atomic_replace(plan.paths["case_current_index_path"], plan.candidate_index_data, plan.identity["attempt_id"])
|
switched = True
|
if fault == "POSTCOMMIT_READBACK_FAIL":
|
raise PublisherError("INJECTED_POSTCOMMIT", "isolated test", exit_code=20)
|
_verify_target(plan)
|
if _read_bytes(plan.paths["case_current_index_path"]) != plan.candidate_index_data:
|
raise PublisherError("POSTCOMMIT_CURRENT_INDEX", "mismatch", exit_code=20)
|
_verify_identity(plan.paths["result_current_index_path"], plan.result_row.bytes, plan.result_row.sha256, "POSTCOMMIT_RESULT_INDEX")
|
_append_event(plan, "COMMITTED", 3, _receipt_relative(plan, "terminal.json"), process_identity)
|
event_processes = (process_identity,) * len(SUCCESS_STATES)
|
chain_digest, start_time, end_time = _validate_attempt_event_chain(plan, SUCCESS_STATES, event_processes)
|
value = _terminal(
|
plan,
|
"COMMITTED",
|
0,
|
SUCCESS_STATES,
|
process_identity,
|
chain_digest,
|
start_time,
|
end_time,
|
event_processes,
|
committed=True,
|
rolled_back=False,
|
)
|
_write_terminal(plan, value)
|
return value
|
except PublisherError as exc:
|
if switched:
|
return _rollback(plan, process_identity, 3, f"{exc.code}:{exc.detail}", SUCCESS_STATES[:3], "terminal.json")
|
raise
|
finally:
|
if _exists(plan.paths["lock_path"]):
|
os.unlink(_long(plan.paths["lock_path"]))
|
|
|
def run_config(config_path: Path, *, validate_only: bool = False) -> dict[str, Any]:
|
config, config_data = _load_config(config_path)
|
if (
|
config.get("schema_version") == DIRECT_CONFIG_SCHEMA
|
and isinstance(config.get("commit"), dict)
|
and config["commit"].get("strategy") == DIRECT_COMMIT_STRATEGY
|
):
|
# The direct stable-path transaction is intentionally isolated from the
|
# accepted release-directory strategy. Both use the same public CLI,
|
# ledger and event-chain primitives, while retaining separate strict
|
# schemas and topology validators.
|
from .direct_stable_path import run_direct_config
|
|
return run_direct_config(
|
config,
|
Path(os.path.abspath(config_path)),
|
config_data,
|
validate_only=validate_only,
|
)
|
plan = _validate_config(config, Path(os.path.abspath(config_path)), config_data)
|
if validate_only:
|
return {
|
"schema_version": 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,
|
}
|
if plan.lifecycle == "REPLAY":
|
try:
|
return _verify_replay(plan)
|
except PublisherError as exc:
|
return _recover_invalid_replay(plan, exc)
|
return _publish(plan)
|
|
|
def main(argv: list[str] | None = None) -> int:
|
if hasattr(sys.stdout, "reconfigure"):
|
sys.stdout.reconfigure(encoding="utf-8", errors="strict")
|
parser = argparse.ArgumentParser(description=__doc__)
|
parser.add_argument("--config", required=True, type=Path)
|
parser.add_argument("--validate-only", action="store_true")
|
parser.add_argument("--read-direct-current", action="store_true")
|
args = parser.parse_args(argv)
|
try:
|
if args.validate_only and args.read_direct_current:
|
raise PublisherError("CLI_MODE", "validate-only and read-direct-current are mutually exclusive")
|
if args.read_direct_current:
|
from .authoritative_reader import read_direct_current
|
|
value = read_direct_current(args.config)
|
else:
|
value = run_config(args.config, validate_only=args.validate_only)
|
print(json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True))
|
return int(value.get("exit_code", 0))
|
except PublisherError as exc:
|
value = {"schema_version": TERMINAL_SCHEMA, "status": "FAIL_CLOSED", "exit_code": exc.exit_code, "error_code": exc.code, "detail": exc.detail}
|
print(json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True))
|
return exc.exit_code
|
except Exception as exc: # fail closed without inventing state
|
value = {"schema_version": TERMINAL_SCHEMA, "status": "FAIL_CLOSED", "exit_code": 27, "error_code": "UNEXPECTED", "detail": f"{type(exc).__name__}:{exc}"}
|
print(json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True))
|
return 27
|
|
|
if __name__ == "__main__":
|
sys.exit(main())
|
