#!/usr/bin/env python3
"""Build or verify deterministic shards for a verified corpus work index.

This command reads only inventory metadata. It does not read source snapshots,
extract claims, create proposals, admit knowledge, or write to a database.
"""

from __future__ import annotations

import argparse
import copy
import hashlib
import json
import os
import re
import stat
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Any

try:
    from . import run_local_corpus_knowledge_rebuild as inventory
except ImportError:  # pragma: no cover - direct script execution
    import run_local_corpus_knowledge_rebuild as inventory


SHARD_SCHEMA = "livingip.localCorpusExtractionWorkShard.v1"
MANIFEST_SCHEMA = "livingip.localCorpusExtractionShardManifest.v1"
RECEIPT_SCHEMA = "livingip.localCorpusExtractionShardReceipt.v1"
STATUS_SCHEMA = "livingip.localCorpusExtractionShardStatus.v1"
INVENTORY_BINDING_SCHEMA = "livingip.localCorpusInventoryBinding.v1"

MODE = "semantic-extraction-work-sharding"
ASSIGNMENT_ALGORITHM = "sha256_utf8_work_item_id_modulo_v1"
SHARDS_DIR = "shards"
MANIFEST_PATH = "shard-manifest.json"
RECEIPT_PATH = "receipt.json"
INCOMPLETE_MARKER = ".incomplete"
MIN_SHARD_COUNT = 1
MAX_SHARD_COUNT = 1024
MAX_METADATA_BYTES = inventory.MAX_METADATA_FILE_BYTES
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")

NO_WRITE_FLAGS = dict(inventory.NO_WRITE_FLAGS)
WORK_ITEM_KEYS = {
    "admission_performed",
    "artifact_sha256",
    "database_write_performed",
    "locator",
    "relative_path",
    "semantic_claim_extraction_performed",
    "snapshot_path",
    "work_item_id",
    "work_type",
}
LOCATOR_KEYS = {
    "line_end",
    "line_start",
    "quote_sha256",
    "section_ordinal",
    "utf8_byte_end",
    "utf8_byte_start",
}


class ShardError(RuntimeError):
    """Fail-closed sharding error with a stable, non-sensitive code."""

    def __init__(self, code: str, safe_message: str) -> None:
        super().__init__(safe_message)
        self.code = code
        self.safe_message = safe_message


@dataclass(frozen=True)
class InventoryView:
    root: Path
    binding: dict[str, Any]
    items: list[dict[str, Any]]


@dataclass(frozen=True)
class ReservedOutput:
    path: Path
    name: str
    parent_fd: int
    root_fd: int
    device: int
    inode: int


def canonical_json_bytes(value: Any) -> bytes:
    try:
        return json.dumps(
            value,
            allow_nan=False,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("utf-8")
    except (TypeError, ValueError) as exc:
        raise ShardError("invalid_json_value", "shard data is not deterministic JSON") from exc


def canonical_sha256(value: Any) -> str:
    return hashlib.sha256(canonical_json_bytes(value)).hexdigest()


def rendered_json_bytes(value: Any) -> bytes:
    try:
        rendered = json.dumps(
            value,
            allow_nan=False,
            ensure_ascii=True,
            indent=2,
            sort_keys=True,
        )
    except (TypeError, ValueError) as exc:
        raise ShardError("invalid_json_value", "shard data is not deterministic JSON") from exc
    return (rendered + "\n").encode("utf-8")


def sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def _absolute_path(path: Path) -> Path:
    expanded = path.expanduser()
    return expanded if expanded.is_absolute() else Path.cwd() / expanded


def _directory_flags() -> int:
    return os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)


def _assert_exact_keys(value: Mapping[str, Any], expected: set[str], *, label: str) -> None:
    if set(value) != expected:
        raise ShardError("invalid_shard_schema", f"{label} fields do not match the supported schema")


def _assert_no_write_flags(value: Mapping[str, Any], *, label: str) -> None:
    for key, expected in NO_WRITE_FLAGS.items():
        if value.get(key) is not expected:
            raise ShardError("false_sharding_claim", f"{label} contains an unsupported side-effect claim")


def _assert_no_absolute_strings(value: Any) -> None:
    if isinstance(value, dict):
        for child in value.values():
            _assert_no_absolute_strings(child)
    elif isinstance(value, list):
        for child in value:
            _assert_no_absolute_strings(child)
    elif isinstance(value, str) and (value.startswith(("/", "\\")) or re.match(r"^[A-Za-z]:[\\/]", value)):
        raise ShardError("absolute_path_disclosure", "shard metadata must not contain absolute paths")


def _duplicate_free_json_object(raw: bytes, *, label: str) -> dict[str, Any]:
    def pairs_hook(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
        value: dict[str, Any] = {}
        for key, child in pairs:
            if key in value:
                raise ShardError("duplicate_json_key", f"{label} contains a duplicate JSON key")
            value[key] = child
        return value

    def reject_constant(_value: str) -> None:
        raise ShardError("invalid_json_value", f"{label} contains a non-finite JSON number")

    try:
        parsed = json.loads(raw.decode("utf-8"), object_pairs_hook=pairs_hook, parse_constant=reject_constant)
    except ShardError:
        raise
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise ShardError("invalid_json", f"{label} is not valid duplicate-free UTF-8 JSON") from exc
    if not isinstance(parsed, dict):
        raise ShardError("invalid_json", f"{label} must be a JSON object")
    return parsed


def _read_regular_bytes(path: Path, *, label: str) -> bytes:
    try:
        before = path.lstat()
    except OSError as exc:
        raise ShardError("missing_bundle_entry", f"{label} is missing") from exc
    if stat.S_ISLNK(before.st_mode) or not stat.S_ISREG(before.st_mode):
        raise ShardError("unsafe_bundle_entry", f"{label} must be a regular non-symlink file")
    if before.st_size > MAX_METADATA_BYTES:
        raise ShardError("metadata_limit_exceeded", f"{label} exceeds the metadata byte limit")
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
    try:
        descriptor = os.open(path, flags)
    except OSError as exc:
        raise ShardError("unsafe_bundle_entry", f"{label} could not be opened safely") from exc
    try:
        opened = os.fstat(descriptor)
        if not stat.S_ISREG(opened.st_mode) or opened.st_dev != before.st_dev or opened.st_ino != before.st_ino:
            raise ShardError("bundle_entry_changed", f"{label} changed while it was opened")
        chunks: list[bytes] = []
        total = 0
        while True:
            chunk = os.read(descriptor, 1024 * 1024)
            if not chunk:
                break
            total += len(chunk)
            if total > MAX_METADATA_BYTES:
                raise ShardError("metadata_limit_exceeded", f"{label} exceeds the metadata byte limit")
            chunks.append(chunk)
        after = os.fstat(descriptor)
        before_signature = (before.st_size, before.st_mtime_ns, before.st_ctime_ns)
        after_signature = (after.st_size, after.st_mtime_ns, after.st_ctime_ns)
        if before_signature != after_signature or total != after.st_size:
            raise ShardError("bundle_entry_changed", f"{label} changed while it was read")
        return b"".join(chunks)
    finally:
        os.close(descriptor)


def _inventory_status_signature(value: Mapping[str, Any]) -> tuple[Any, ...]:
    return (
        str(value.get("bundle_path")),
        value.get("bundle_tree_sha256"),
        value.get("receipt_file_sha256"),
        value.get("source_count"),
        value.get("unique_snapshot_count"),
        value.get("work_item_count"),
    )


def _load_verified_inventory(bundle: Path) -> InventoryView:
    try:
        first = inventory.verify_bundle(bundle)
    except inventory.RebuildError as exc:
        raise ShardError("inventory_verification_failed", exc.safe_message) from exc
    root = Path(first["bundle_path"])
    source_raw = _read_regular_bytes(root / inventory.SOURCE_MANIFEST_PATH, label="inventory source manifest")
    work_raw = _read_regular_bytes(
        root / inventory.EXTRACTION_WORK_INDEX_PATH,
        label="inventory extraction-work index",
    )
    receipt_raw = _read_regular_bytes(root / inventory.RECEIPT_PATH, label="inventory receipt")
    source_manifest = _duplicate_free_json_object(source_raw, label="inventory source manifest")
    work_index = _duplicate_free_json_object(work_raw, label="inventory extraction-work index")
    receipt = _duplicate_free_json_object(receipt_raw, label="inventory receipt")
    try:
        second = inventory.verify_bundle(root)
    except inventory.RebuildError as exc:
        raise ShardError("inventory_verification_failed", exc.safe_message) from exc
    if _inventory_status_signature(first) != _inventory_status_signature(second):
        raise ShardError("inventory_changed", "inventory metadata changed during verification")

    items = work_index.get("items")
    if not isinstance(items, list) or not items:
        raise ShardError("invalid_inventory_work", "inventory must contain at least one extraction work item")
    for item in items:
        _validate_work_item(item, label="inventory work item")

    source_hash = sha256_bytes(source_raw)
    work_hash = sha256_bytes(work_raw)
    receipt_hash = sha256_bytes(receipt_raw)
    receipt_hashes = receipt.get("hashes")
    if not isinstance(receipt_hashes, dict):
        raise ShardError("inventory_hash_mismatch", "inventory receipt hashes are malformed")
    if source_hash != receipt_hashes.get("source_manifest_file_sha256"):
        raise ShardError("inventory_hash_mismatch", "inventory source-manifest hash does not match")
    if work_hash != receipt_hashes.get("extraction_work_index_file_sha256"):
        raise ShardError("inventory_hash_mismatch", "inventory extraction-work hash does not match")
    if receipt_hash != first.get("receipt_file_sha256"):
        raise ShardError("inventory_hash_mismatch", "inventory receipt file hash does not match")

    binding_core = {
        "schema": INVENTORY_BINDING_SCHEMA,
        "mode": inventory.MODE_SOURCE_INVENTORY_ONLY,
        "work_type": inventory.WORK_TYPE,
        "source_manifest_schema": source_manifest.get("schema"),
        "extraction_work_index_schema": work_index.get("schema"),
        "inventory_receipt_schema": receipt.get("schema"),
        "source_count": first.get("source_count"),
        "work_item_count": first.get("work_item_count"),
        "inventory_bundle_tree_sha256": first.get("bundle_tree_sha256"),
        "source_manifest_file_sha256": source_hash,
        "extraction_work_index_file_sha256": work_hash,
        "inventory_receipt_file_sha256": receipt_hash,
    }
    binding = {
        **binding_core,
        "inventory_bundle_binding_sha256": canonical_sha256(
            {
                "inventory_bundle_tree_sha256": binding_core["inventory_bundle_tree_sha256"],
                "inventory_receipt_file_sha256": receipt_hash,
            }
        ),
    }
    binding["inventory_metadata_sha256"] = canonical_sha256(binding)
    _validate_inventory_binding(binding, expected=binding)
    return InventoryView(root=root, binding=binding, items=items)


def _validate_inventory_binding(value: Any, *, expected: Mapping[str, Any]) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ShardError("inventory_mismatch", "inventory binding must be an object")
    expected_keys = {
        "schema",
        "mode",
        "work_type",
        "source_manifest_schema",
        "extraction_work_index_schema",
        "inventory_receipt_schema",
        "source_count",
        "work_item_count",
        "inventory_bundle_tree_sha256",
        "source_manifest_file_sha256",
        "extraction_work_index_file_sha256",
        "inventory_receipt_file_sha256",
        "inventory_bundle_binding_sha256",
        "inventory_metadata_sha256",
    }
    _assert_exact_keys(value, expected_keys, label="inventory binding")
    preimage = dict(value)
    metadata_hash = preimage.pop("inventory_metadata_sha256", None)
    if metadata_hash != canonical_sha256(preimage):
        raise ShardError("inventory_mismatch", "inventory metadata hash does not verify")
    bundle_binding = canonical_sha256(
        {
            "inventory_bundle_tree_sha256": value.get("inventory_bundle_tree_sha256"),
            "inventory_receipt_file_sha256": value.get("inventory_receipt_file_sha256"),
        }
    )
    if value.get("inventory_bundle_binding_sha256") != bundle_binding:
        raise ShardError("inventory_mismatch", "inventory bundle binding hash does not verify")
    if value != expected:
        raise ShardError("inventory_mismatch", "shards do not bind the supplied inventory bundle")
    return value


def _validate_work_item(value: Any, *, label: str) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ShardError("invalid_work_item", f"{label} must be an object")
    _assert_exact_keys(value, WORK_ITEM_KEYS, label=label)
    _assert_no_write_flags(value, label=label)
    if value.get("work_type") != inventory.WORK_TYPE:
        raise ShardError("invalid_work_item", f"{label} has an unsupported work type")
    work_id = value.get("work_item_id")
    artifact_hash = value.get("artifact_sha256")
    if not isinstance(work_id, str) or not SHA256_RE.fullmatch(work_id):
        raise ShardError("invalid_work_item", f"{label} has a malformed work-item ID")
    if not isinstance(artifact_hash, str) or not SHA256_RE.fullmatch(artifact_hash):
        raise ShardError("invalid_work_item", f"{label} has a malformed artifact hash")
    locator = value.get("locator")
    if not isinstance(locator, dict):
        raise ShardError("invalid_work_item", f"{label} has a malformed locator")
    _assert_exact_keys(locator, LOCATOR_KEYS, label=f"{label} locator")
    if not isinstance(locator.get("quote_sha256"), str) or not SHA256_RE.fullmatch(locator["quote_sha256"]):
        raise ShardError("invalid_work_item", f"{label} has a malformed quote hash")
    relative_path = value.get("relative_path")
    snapshot_path = value.get("snapshot_path")
    for candidate in (relative_path, snapshot_path):
        if not isinstance(candidate, str):
            raise ShardError("invalid_work_item", f"{label} has a malformed relative path")
        path = PurePosixPath(candidate)
        if path.is_absolute() or not path.parts or any(part in {"", ".", ".."} for part in path.parts):
            raise ShardError("absolute_path_disclosure", f"{label} must contain only safe relative paths")
    return value


def assignment_for(work_item_id: str, shard_count: int) -> int:
    digest = hashlib.sha256(work_item_id.encode("utf-8")).digest()
    return int.from_bytes(digest, "big") % shard_count


def _assignment_contract(shard_count: int) -> dict[str, Any]:
    return {
        "algorithm": ASSIGNMENT_ALGORITHM,
        "hash": "sha256",
        "input": "utf8(work_item_id)",
        "reduction": "unsigned_big_endian_modulo_shard_count",
        "shard_count": shard_count,
    }


def _validate_shard_count(value: Any, *, work_item_count: int) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        raise ShardError("invalid_shard_count", "shard count must be an integer")
    if not MIN_SHARD_COUNT <= value <= MAX_SHARD_COUNT:
        raise ShardError("invalid_shard_count", "shard count must be between 1 and 1024")
    if value > work_item_count:
        raise ShardError("invalid_shard_count", "shard count must not exceed the work-item count")
    return value


def _prepare_output(output: Path, *, inventory_root: Path) -> Path:
    absolute = _absolute_path(output)
    if absolute.name in {"", ".", ".."}:
        raise ShardError("unsafe_output", "output-dir must name a new directory")
    try:
        parent = absolute.parent.resolve(strict=True)
    except OSError as exc:
        raise ShardError("unsafe_output_parent", "output-dir parent must be an existing directory") from exc
    try:
        parent_metadata = parent.lstat()
    except OSError as exc:
        raise ShardError("unsafe_output_parent", "output-dir parent could not be inspected") from exc
    if stat.S_ISLNK(parent_metadata.st_mode) or not stat.S_ISDIR(parent_metadata.st_mode):
        raise ShardError("unsafe_output_parent", "output-dir parent must be a real directory")
    prepared = parent / absolute.name
    try:
        prepared.lstat()
    except FileNotFoundError:
        pass
    except OSError as exc:
        raise ShardError("unsafe_output", "output-dir could not be inspected safely") from exc
    else:
        raise ShardError("output_exists", "output-dir must not already exist")
    inventory_resolved = inventory_root.resolve()
    common = Path(os.path.commonpath((str(prepared), str(inventory_resolved))))
    if common in {prepared, inventory_resolved}:
        raise ShardError("overlapping_paths", "output-dir and inventory-bundle must not overlap")
    return prepared


def _write_bytes_at(directory_fd: int, name: str, payload: bytes) -> None:
    if not name or "/" in name or name in {".", ".."}:
        raise ShardError("unsafe_output_entry", "output artifact name is unsafe")
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
    try:
        descriptor = os.open(name, flags, 0o600, dir_fd=directory_fd)
    except OSError as exc:
        raise ShardError("output_write_failed", "output artifact could not be created exclusively") from exc
    try:
        view = memoryview(payload)
        while view:
            written = os.write(descriptor, view)
            if written <= 0:
                raise ShardError("output_write_failed", "output artifact write did not complete")
            view = view[written:]
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def _write_json_at(directory_fd: int, name: str, value: Any) -> bytes:
    payload = rendered_json_bytes(value)
    _write_bytes_at(directory_fd, name, payload)
    return payload


def _reserve_output(output: Path) -> ReservedOutput:
    try:
        parent_fd = os.open(output.parent, _directory_flags())
    except OSError as exc:
        raise ShardError("output_reservation_failed", "output parent could not be opened safely") from exc
    try:
        os.mkdir(output.name, mode=0o700, dir_fd=parent_fd)
    except FileExistsError as exc:
        os.close(parent_fd)
        raise ShardError("output_exists", "output-dir must not already exist") from exc
    except OSError as exc:
        os.close(parent_fd)
        raise ShardError("output_reservation_failed", "output-dir could not be reserved") from exc
    root_fd: int | None = None
    try:
        root_fd = os.open(output.name, _directory_flags(), dir_fd=parent_fd)
        metadata = os.fstat(root_fd)
        reserved = ReservedOutput(
            path=output,
            name=output.name,
            parent_fd=parent_fd,
            root_fd=root_fd,
            device=metadata.st_dev,
            inode=metadata.st_ino,
        )
        _write_bytes_at(root_fd, INCOMPLETE_MARKER, b"corpus extraction sharding incomplete\n")
        return reserved
    except Exception:
        if root_fd is not None:
            try:
                _remove_tree_contents(root_fd)
            except OSError:
                pass
            os.close(root_fd)
        try:
            os.rmdir(output.name, dir_fd=parent_fd)
        except OSError:
            pass
        os.close(parent_fd)
        raise


def _assert_output_binding(reserved: ReservedOutput) -> None:
    try:
        opened = os.fstat(reserved.root_fd)
        linked = os.stat(reserved.name, dir_fd=reserved.parent_fd, follow_symlinks=False)
    except OSError as exc:
        raise ShardError("output_binding_changed", "reserved output directory is no longer available") from exc
    for metadata in (opened, linked):
        if (
            not stat.S_ISDIR(metadata.st_mode)
            or stat.S_ISLNK(metadata.st_mode)
            or metadata.st_dev != reserved.device
            or metadata.st_ino != reserved.inode
        ):
            raise ShardError("output_binding_changed", "reserved output directory identity changed")
    if stat.S_IMODE(opened.st_mode) != 0o700:
        raise ShardError("unsafe_private_permissions", "reserved output directory permissions changed")


def _remove_tree_contents(directory_fd: int) -> None:
    for name in os.listdir(directory_fd):
        try:
            metadata = os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
        except FileNotFoundError:
            continue
        if stat.S_ISDIR(metadata.st_mode) and not stat.S_ISLNK(metadata.st_mode):
            child_fd = os.open(name, _directory_flags(), dir_fd=directory_fd)
            try:
                _remove_tree_contents(child_fd)
            finally:
                os.close(child_fd)
            os.rmdir(name, dir_fd=directory_fd)
        else:
            os.unlink(name, dir_fd=directory_fd)


def _release_reserved(reserved: ReservedOutput) -> None:
    os.close(reserved.root_fd)
    os.close(reserved.parent_fd)


def _cleanup_reserved(reserved: ReservedOutput) -> None:
    try:
        _remove_tree_contents(reserved.root_fd)
    except OSError:
        pass
    try:
        linked = os.stat(reserved.name, dir_fd=reserved.parent_fd, follow_symlinks=False)
        if (
            stat.S_ISDIR(linked.st_mode)
            and not stat.S_ISLNK(linked.st_mode)
            and linked.st_dev == reserved.device
            and linked.st_ino == reserved.inode
        ):
            os.rmdir(reserved.name, dir_fd=reserved.parent_fd)
    except OSError:
        pass
    _release_reserved(reserved)


def _mkdir_at(directory_fd: int, name: str) -> int:
    try:
        os.mkdir(name, mode=0o700, dir_fd=directory_fd)
        return os.open(name, _directory_flags(), dir_fd=directory_fd)
    except OSError as exc:
        raise ShardError("output_write_failed", "private shard directory could not be created") from exc


def _shard_path(index: int, shard_count: int) -> str:
    return f"{SHARDS_DIR}/shard-{index:04d}-of-{shard_count:04d}.json"


def _artifact_records(root: Path, paths: Sequence[str]) -> list[dict[str, Any]]:
    records: list[dict[str, Any]] = []
    for relative in sorted(paths, key=lambda value: value.encode("utf-8")):
        posix = PurePosixPath(relative)
        if posix.is_absolute() or any(part in {"", ".", ".."} for part in posix.parts):
            raise ShardError("unsafe_bundle_entry", "bundle artifact path is unsafe")
        raw = _read_regular_bytes(root / Path(relative), label="shard artifact")
        records.append({"bytes": len(raw), "path": relative, "sha256": sha256_bytes(raw)})
    return records


def _after_output_reservation(_reserved: ReservedOutput) -> None:
    """Test seam after exclusive reservation; production does nothing."""


def _before_final_publication(_inventory_root: Path) -> None:
    """Test seam before final inventory revalidation; production does nothing."""


def _before_receipt_publication(_reserved: ReservedOutput) -> None:
    """Test seam while the incomplete marker must still exist."""


def build_shard_bundle(inventory_bundle: Path, output_dir: Path, shard_count: int) -> dict[str, Any]:
    view = _load_verified_inventory(inventory_bundle)
    checked_count = _validate_shard_count(shard_count, work_item_count=len(view.items))
    output = _prepare_output(output_dir, inventory_root=view.root)
    reserved = _reserve_output(output)
    success = False
    try:
        _after_output_reservation(reserved)
        _assert_output_binding(reserved)
        assignments: list[list[dict[str, Any]]] = [[] for _ in range(checked_count)]
        for item in sorted(view.items, key=lambda row: row["work_item_id"].encode("utf-8")):
            assignments[assignment_for(item["work_item_id"], checked_count)].append(item)

        shards_fd = _mkdir_at(reserved.root_fd, SHARDS_DIR)
        shard_records: list[dict[str, Any]] = []
        try:
            for index, items in enumerate(assignments):
                _assert_output_binding(reserved)
                work_ids = [item["work_item_id"] for item in items]
                shard_document = {
                    "schema": SHARD_SCHEMA,
                    "mode": MODE,
                    "deterministic": True,
                    **NO_WRITE_FLAGS,
                    "work_type": inventory.WORK_TYPE,
                    "inventory": view.binding,
                    "assignment": {**_assignment_contract(checked_count), "shard_index": index},
                    "work_item_count": len(items),
                    "work_item_ids_sha256": canonical_sha256(work_ids),
                    "items": items,
                }
                filename = Path(_shard_path(index, checked_count)).name
                payload = _write_json_at(shards_fd, filename, shard_document)
                shard_records.append(
                    {
                        "bytes": len(payload),
                        "index": index,
                        "path": _shard_path(index, checked_count),
                        "sha256": sha256_bytes(payload),
                        "work_item_count": len(items),
                        "work_item_ids_sha256": canonical_sha256(work_ids),
                    }
                )
        finally:
            os.close(shards_fd)

        assignment_contract = _assignment_contract(checked_count)
        manifest = {
            "schema": MANIFEST_SCHEMA,
            "mode": MODE,
            "deterministic": True,
            **NO_WRITE_FLAGS,
            "work_type": inventory.WORK_TYPE,
            "inventory": view.binding,
            "assignment": assignment_contract,
            "shard_count": checked_count,
            "work_item_count": len(view.items),
            "shards": shard_records,
        }
        _assert_output_binding(reserved)
        _write_json_at(reserved.root_fd, MANIFEST_PATH, manifest)

        _before_final_publication(view.root)
        fresh_view = _load_verified_inventory(view.root)
        if fresh_view.binding != view.binding or fresh_view.items != view.items:
            raise ShardError("inventory_changed", "inventory changed before shard publication")

        artifact_paths = [MANIFEST_PATH, *(record["path"] for record in shard_records)]
        artifacts = _artifact_records(output, artifact_paths)
        receipt = {
            "schema": RECEIPT_SCHEMA,
            "status": "pass",
            "result": "extraction_work_shards_ready",
            "mode": MODE,
            "deterministic": True,
            **NO_WRITE_FLAGS,
            "work_type": inventory.WORK_TYPE,
            "inventory": view.binding,
            "assignment": assignment_contract,
            "shard_count": checked_count,
            "work_item_count": len(view.items),
            "outputs": {
                "artifact_count": len(artifacts),
                "artifacts": artifacts,
                "manifest_path": MANIFEST_PATH,
            },
            "hashes": {
                "bundle_tree_sha256": canonical_sha256(artifacts),
                "manifest_file_sha256": next(
                    record["sha256"] for record in artifacts if record["path"] == MANIFEST_PATH
                ),
            },
        }
        receipt["hashes"]["receipt_content_sha256"] = canonical_sha256(receipt)
        _before_receipt_publication(reserved)
        _assert_output_binding(reserved)
        _write_json_at(reserved.root_fd, RECEIPT_PATH, receipt)
        os.fsync(reserved.root_fd)
        os.unlink(INCOMPLETE_MARKER, dir_fd=reserved.root_fd)
        os.fsync(reserved.root_fd)
        _assert_output_binding(reserved)
        verified = verify_shard_bundle(view.root, output)
        _assert_output_binding(reserved)
        success = True
        return {"bundle_path": output, "verification": verified}
    finally:
        if success:
            _release_reserved(reserved)
        else:
            _cleanup_reserved(reserved)


def _validate_shard_root(bundle: Path) -> Path:
    absolute = _absolute_path(bundle)
    try:
        metadata = absolute.lstat()
    except OSError as exc:
        raise ShardError("missing_shard_bundle", "shard bundle must be an existing directory") from exc
    if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
        raise ShardError("unsafe_shard_bundle", "shard bundle must be a real non-symlink directory")
    root = absolute.resolve()
    if stat.S_IMODE(metadata.st_mode) != 0o700:
        raise ShardError("unsafe_private_permissions", "shard bundle directory permissions must be 0700")
    marker = root / INCOMPLETE_MARKER
    if marker.exists() or marker.is_symlink():
        raise ShardError("incomplete_shard_bundle", "shard bundle retains its incomplete marker")
    return root


def _scan_bundle_tree(root: Path) -> tuple[set[str], set[str]]:
    files: set[str] = set()
    directories: set[str] = set()

    def visit(directory: Path, relative_parent: PurePosixPath) -> None:
        for child in sorted(directory.iterdir(), key=lambda path: path.name.encode("utf-8")):
            relative = (relative_parent / child.name).as_posix()
            metadata = child.lstat()
            if stat.S_ISLNK(metadata.st_mode):
                raise ShardError("unsafe_bundle_entry", "shard bundle contains a symlink")
            if stat.S_ISDIR(metadata.st_mode):
                if stat.S_IMODE(metadata.st_mode) != 0o700:
                    raise ShardError("unsafe_private_permissions", "shard directory permissions must be 0700")
                directories.add(relative)
                visit(child, relative_parent / child.name)
            elif stat.S_ISREG(metadata.st_mode):
                if stat.S_IMODE(metadata.st_mode) != 0o600:
                    raise ShardError("unsafe_private_permissions", "shard file permissions must be 0600")
                files.add(relative)
            else:
                raise ShardError("unsafe_bundle_entry", "shard bundle contains a special file")

    visit(root, PurePosixPath())
    return files, directories


def _validate_common_document(
    value: Any,
    *,
    expected_keys: set[str],
    label: str,
    expected_inventory: Mapping[str, Any],
) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ShardError("invalid_shard_schema", f"{label} must be an object")
    _assert_exact_keys(value, expected_keys, label=label)
    _assert_no_write_flags(value, label=label)
    if value.get("mode") != MODE or value.get("deterministic") is not True:
        raise ShardError("invalid_shard_schema", f"{label} mode or deterministic flag is unsupported")
    if value.get("work_type") != inventory.WORK_TYPE:
        raise ShardError("invalid_shard_schema", f"{label} work type is unsupported")
    _validate_inventory_binding(value.get("inventory"), expected=expected_inventory)
    _assert_no_absolute_strings(value)
    return value


def verify_shard_bundle(inventory_bundle: Path, shard_bundle: Path) -> dict[str, Any]:
    view = _load_verified_inventory(inventory_bundle)
    root = _validate_shard_root(shard_bundle)
    manifest_raw = _read_regular_bytes(root / MANIFEST_PATH, label="shard manifest")
    receipt_raw = _read_regular_bytes(root / RECEIPT_PATH, label="shard receipt")
    manifest = _duplicate_free_json_object(manifest_raw, label="shard manifest")
    receipt = _duplicate_free_json_object(receipt_raw, label="shard receipt")

    manifest_keys = {
        "schema",
        "mode",
        "deterministic",
        *NO_WRITE_FLAGS,
        "work_type",
        "inventory",
        "assignment",
        "shard_count",
        "work_item_count",
        "shards",
    }
    receipt_keys = {
        "schema",
        "status",
        "result",
        "mode",
        "deterministic",
        *NO_WRITE_FLAGS,
        "work_type",
        "inventory",
        "assignment",
        "shard_count",
        "work_item_count",
        "outputs",
        "hashes",
    }
    _validate_common_document(
        manifest,
        expected_keys=manifest_keys,
        label="shard manifest",
        expected_inventory=view.binding,
    )
    _validate_common_document(
        receipt,
        expected_keys=receipt_keys,
        label="shard receipt",
        expected_inventory=view.binding,
    )
    if manifest.get("schema") != MANIFEST_SCHEMA or receipt.get("schema") != RECEIPT_SCHEMA:
        raise ShardError("invalid_shard_schema", "shard manifest or receipt schema is unsupported")
    if receipt.get("status") != "pass" or receipt.get("result") != "extraction_work_shards_ready":
        raise ShardError("invalid_shard_schema", "receipt does not represent a completed shard bundle")

    shard_count = _validate_shard_count(manifest.get("shard_count"), work_item_count=len(view.items))
    if receipt.get("shard_count") != shard_count:
        raise ShardError("invalid_shard_cardinality", "receipt shard count differs from the manifest")
    assignment = _assignment_contract(shard_count)
    if manifest.get("assignment") != assignment or receipt.get("assignment") != assignment:
        raise ShardError("invalid_assignment", "shard assignment contract is unsupported")
    if manifest.get("work_item_count") != len(view.items) or receipt.get("work_item_count") != len(view.items):
        raise ShardError("invalid_shard_cardinality", "shard work-item count differs from the inventory")

    shard_records = manifest.get("shards")
    if not isinstance(shard_records, list) or len(shard_records) != shard_count:
        raise ShardError("invalid_shard_cardinality", "manifest must contain one record per shard")
    expected_files = {MANIFEST_PATH, RECEIPT_PATH}
    expected_files.update(_shard_path(index, shard_count) for index in range(shard_count))
    actual_files, actual_directories = _scan_bundle_tree(root)
    if actual_directories != {SHARDS_DIR} or actual_files != expected_files:
        raise ShardError("unexpected_bundle_entry", "shard bundle contains missing or unexpected entries")

    expected_by_id = {item["work_item_id"]: item for item in view.items}
    seen: set[str] = set()
    expected_shard_records: list[dict[str, Any]] = []
    shard_paths: list[str] = []
    for index, record in enumerate(shard_records):
        if not isinstance(record, dict):
            raise ShardError("invalid_shard_schema", "shard record must be an object")
        _assert_exact_keys(
            record,
            {"bytes", "index", "path", "sha256", "work_item_count", "work_item_ids_sha256"},
            label="shard record",
        )
        path = _shard_path(index, shard_count)
        if record.get("index") != index or record.get("path") != path:
            raise ShardError("invalid_shard_order", "shard records are not in deterministic index order")
        raw = _read_regular_bytes(root / Path(path), label="shard file")
        document = _duplicate_free_json_object(raw, label="shard file")
        shard_keys = {
            "schema",
            "mode",
            "deterministic",
            *NO_WRITE_FLAGS,
            "work_type",
            "inventory",
            "assignment",
            "work_item_count",
            "work_item_ids_sha256",
            "items",
        }
        _validate_common_document(
            document,
            expected_keys=shard_keys,
            label="shard file",
            expected_inventory=view.binding,
        )
        if document.get("schema") != SHARD_SCHEMA:
            raise ShardError("invalid_shard_schema", "shard file schema is unsupported")
        expected_assignment = {**assignment, "shard_index": index}
        if document.get("assignment") != expected_assignment:
            raise ShardError("invalid_assignment", "shard file assignment metadata is unsupported")
        items = document.get("items")
        if not isinstance(items, list):
            raise ShardError("invalid_shard_schema", "shard items must be an array")
        ids: list[str] = []
        previous_id: str | None = None
        for item in items:
            _validate_work_item(item, label="shard work item")
            work_id = item["work_item_id"]
            if previous_id is not None and work_id.encode("utf-8") <= previous_id.encode("utf-8"):
                raise ShardError("invalid_shard_order", "shard work items are not in deterministic order")
            previous_id = work_id
            if work_id in seen:
                raise ShardError("duplicate_work_item", "a work-item ID appears more than once across shards")
            if work_id not in expected_by_id:
                raise ShardError("unknown_work_item", "a shard contains a work-item ID absent from the inventory")
            if item != expected_by_id[work_id]:
                raise ShardError("inventory_mismatch", "a shard work item differs from the verified inventory row")
            if assignment_for(work_id, shard_count) != index:
                raise ShardError("invalid_assignment", "a work item is assigned to the wrong hash shard")
            seen.add(work_id)
            ids.append(work_id)
        ids_hash = canonical_sha256(ids)
        if document.get("work_item_count") != len(items) or document.get("work_item_ids_sha256") != ids_hash:
            raise ShardError("invalid_shard_cardinality", "shard item count or ID hash is incorrect")
        expected_record = {
            "bytes": len(raw),
            "index": index,
            "path": path,
            "sha256": sha256_bytes(raw),
            "work_item_count": len(items),
            "work_item_ids_sha256": ids_hash,
        }
        if record != expected_record:
            raise ShardError("invalid_shard_hash", "manifest shard record does not match shard bytes")
        expected_shard_records.append(expected_record)
        shard_paths.append(path)

    missing = set(expected_by_id) - seen
    if missing:
        raise ShardError("missing_work_item", "one or more inventory work items are absent from all shards")
    if shard_records != expected_shard_records:
        raise ShardError("invalid_shard_hash", "manifest shard records are not reproducible")

    artifacts = _artifact_records(root, [MANIFEST_PATH, *shard_paths])
    outputs = receipt.get("outputs")
    hashes = receipt.get("hashes")
    if not isinstance(outputs, dict) or not isinstance(hashes, dict):
        raise ShardError("invalid_shard_schema", "receipt outputs or hashes are malformed")
    _assert_exact_keys(outputs, {"artifact_count", "artifacts", "manifest_path"}, label="receipt outputs")
    _assert_exact_keys(
        hashes,
        {"bundle_tree_sha256", "manifest_file_sha256", "receipt_content_sha256"},
        label="receipt hashes",
    )
    if outputs.get("manifest_path") != MANIFEST_PATH:
        raise ShardError("invalid_shard_schema", "receipt manifest path is unsupported")
    if outputs.get("artifacts") != artifacts or outputs.get("artifact_count") != len(artifacts):
        raise ShardError("invalid_shard_hash", "receipt artifact inventory does not match bundle bytes")
    expected_hashes = {
        "bundle_tree_sha256": canonical_sha256(artifacts),
        "manifest_file_sha256": sha256_bytes(manifest_raw),
    }
    for key, expected in expected_hashes.items():
        if hashes.get(key) != expected:
            raise ShardError("invalid_shard_hash", f"receipt {key} does not match bundle bytes")
    receipt_preimage = copy.deepcopy(receipt)
    preimage_hashes = receipt_preimage.get("hashes")
    if not isinstance(preimage_hashes, dict):
        raise ShardError("invalid_shard_schema", "receipt hash object is malformed")
    recorded_receipt_hash = preimage_hashes.pop("receipt_content_sha256", None)
    if recorded_receipt_hash != canonical_sha256(receipt_preimage):
        raise ShardError("invalid_receipt_hash", "receipt content hash does not verify")

    return {
        "bundle_path": root,
        "bundle_tree_sha256": expected_hashes["bundle_tree_sha256"],
        "inventory_bundle_binding_sha256": view.binding["inventory_bundle_binding_sha256"],
        "receipt_file_sha256": sha256_bytes(receipt_raw),
        "shard_count": shard_count,
        "status": "pass",
        "work_item_count": len(seen),
    }


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--inventory-bundle", type=Path, required=True)
    parser.add_argument("--output-dir", type=Path, help="new directory for deterministic shard artifacts")
    parser.add_argument("--shard-count", type=int, help="number of stable hash shards, from 1 through 1024")
    parser.add_argument("--verify-bundle", type=Path, help="read-only validation of an existing shard bundle")
    return parser.parse_args(argv)


def _print_status(
    status: str,
    *,
    operation: str,
    reason: str | None = None,
    message: str | None = None,
    result: Mapping[str, Any] | None = None,
) -> None:
    payload: dict[str, Any] = {
        "schema": STATUS_SCHEMA,
        "status": status,
        "operation": operation,
        "mode": MODE,
        **NO_WRITE_FLAGS,
    }
    if reason is not None:
        payload["reason"] = reason
    if message is not None:
        payload["message"] = message
    if result is not None:
        payload.update(
            {
                "bundle_tree_sha256": result["bundle_tree_sha256"],
                "inventory_bundle_binding_sha256": result["inventory_bundle_binding_sha256"],
                "receipt_file_sha256": result["receipt_file_sha256"],
                "shard_count": result["shard_count"],
                "work_item_count": result["work_item_count"],
            }
        )
    print(json.dumps(payload, ensure_ascii=True, sort_keys=True))


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    operation = "verify" if args.verify_bundle is not None else "build"
    try:
        if args.verify_bundle is not None:
            if args.output_dir is not None or args.shard_count is not None:
                raise ShardError("invalid_arguments", "verify mode accepts only inventory-bundle and verify-bundle")
            result = verify_shard_bundle(args.inventory_bundle, args.verify_bundle)
        else:
            if args.output_dir is None or args.shard_count is None:
                raise ShardError("invalid_arguments", "build mode requires output-dir and shard-count")
            built = build_shard_bundle(args.inventory_bundle, args.output_dir, args.shard_count)
            result = built["verification"]
    except ShardError as exc:
        _print_status("rejected", operation=operation, reason=exc.code, message=exc.safe_message)
        return 2
    except OSError:
        _print_status(
            "rejected",
            operation=operation,
            reason="filesystem_operation_failed",
            message="a fail-closed filesystem operation did not complete",
        )
        return 2
    _print_status("pass", operation=operation, result=result)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())