"""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())