GSQR-256 / fixed-guidance column recovery over GF(256)

gsqr_256c.py

#!/usr/bin/env python3
"""GSQR-256C: concurrent fixed-guidance column recovery over GF(256).

GSQR - Guided Syndrome Quotient Recovery. Each independently
compressed piece is striped into eight byte-wide data columns and protected
by two equal-width parity columns:

    P0 = D0 XOR D1 XOR ... XOR D7
    P1 = g0*D0 XOR g1*D1 XOR ... XOR g7*D7

Multiplication and division are in GF(2^8), using primitive polynomial 0x11D.
The eight nonzero guidance values ``g0..g7`` are fixed before encoding.  For a
single damaged data column, the quotient ``S1 / S0`` matches exactly one fixed
guidance value and therefore identifies the column before reconstruction.

The ``C`` suffix means pieces may be processed concurrently.  Concurrency is
strictly bounded and results are committed in piece-index order, so worker
timing cannot change archive bytes or decoded order.  Workers never share a
writable shard or manifest.

Recovery guarantee
------------------

* one parity-inferred, digest-confirmed bad column; or
* any two columns already identified as erasures by their stored digests.

Damage beyond that evidence is rejected.  The archive protects against
accidental loss and corruption, not hostile modification: its manifest is not
authenticated by a secret key.

This module is standalone, uses only the Python standard library, and does not
import or alter TDS, Qiskit, the prior GF(257) reference, or its tests.
"""

from __future__ import annotations

import argparse
from collections import deque
from concurrent.futures import Future, ProcessPoolExecutor
from dataclasses import dataclass
import hashlib
import json
import os
from pathlib import Path
import random
import shutil
import stat
import struct
import sys
import tempfile
from typing import Any, Callable, Iterable, Iterator, Sequence, TypeVar
import zlib


# ---------------------------------------------------------------------------
# Frozen GSQR-256 format constants
# ---------------------------------------------------------------------------

IMPLEMENTATION_NAME = "GSQR-256C"
FORMAT_NAME = "gsqr-256"
FORMAT_VERSION = 1

# 0x11D represents x^8 + x^4 + x^3 + x^2 + 1.  Unlike ordinary modulo-256
# arithmetic, this polynomial makes every nonzero byte multiplicatively
# invertible.  Generator 0x02 traverses all 255 nonzero field elements.
FIELD_POLYNOMIAL = 0x11D
PRIMITIVE_POLYNOMIAL = FIELD_POLYNOMIAL
FIELD_GENERATOR = 0x02

DATA_COLUMNS = 8
PARITY_COLUMNS = 2
PIECE_BYTES = 64 * 1024
COMPRESSION_LEVEL = 9
DEFAULT_WORKERS = max(1, min(4, os.cpu_count() or 1))
MAX_WORKERS = 64
IN_FLIGHT_PER_WORKER = 2

FRAME_MAGIC = b"GSQ256C1"
FRAME_HEADER = struct.Struct(">8sIQQ32s32s32s")
MANIFEST_NAME = "manifest.json"
MAX_MANIFEST_BYTES = 64 * 1024 * 1024


class CodecError(Exception):
    """Base class for expected GSQR codec failures."""


class FormatError(CodecError):
    """The archive structure or frozen guidance is invalid."""


class UnrecoverableError(CodecError):
    """Observed damage exceeds GSQR's declared correction evidence."""


def _canonical_json(value: Any) -> bytes:
    """Serialize guidance deterministically for its stable identity hash."""

    return json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=True,
    ).encode("utf-8")


def _pretty_json(value: Any) -> bytes:
    """Serialize a deterministic but human-readable archive manifest."""

    return (json.dumps(value, sort_keys=True, indent=2) + "\n").encode("utf-8")


def _bounded_manifest_bytes(value: Any) -> bytes:
    """Serialize without creating an archive its own reader must reject."""

    raw = _pretty_json(value)
    if len(raw) > MAX_MANIFEST_BYTES:
        raise CodecError(
            f"manifest exceeds the {MAX_MANIFEST_BYTES}-byte format bound"
        )
    return raw


# ---------------------------------------------------------------------------
# GF(256) arithmetic
# ---------------------------------------------------------------------------

def _build_field_tables() -> tuple[tuple[int, ...], tuple[int, ...]]:
    """Build immutable exponent/log tables and verify the primitive cycle.

    The log entry for zero remains -1 and is never consulted by multiplication
    or division.  Duplicating the exponent table avoids a modulo operation in
    the hot multiplication path.
    """

    exponent = [0] * 512
    logarithm = [-1] * 256
    value = 1
    for power in range(255):
        if logarithm[value] != -1:
            raise RuntimeError("0x02 is not primitive under polynomial 0x11D")
        exponent[power] = value
        logarithm[value] = power
        value <<= 1
        if value & 0x100:
            value ^= FIELD_POLYNOMIAL
    if value != 1 or any(logarithm[value] < 0 for value in range(1, 256)):
        raise RuntimeError("GF(256) table did not cover every nonzero byte")
    for power in range(255, len(exponent)):
        exponent[power] = exponent[power - 255]
    return tuple(exponent), tuple(logarithm)


GF_EXP, GF_LOG = _build_field_tables()


def _check_byte(value: int, label: str = "field value") -> None:
    if not isinstance(value, int) or isinstance(value, bool) or not 0 <= value <= 0xFF:
        raise ValueError(f"{label} must be an integer byte")


def gf_mul(left: int, right: int) -> int:
    """Multiply two byte values in GF(256)."""

    _check_byte(left, "left operand")
    _check_byte(right, "right operand")
    if left == 0 or right == 0:
        return 0
    return GF_EXP[GF_LOG[left] + GF_LOG[right]]


def gf_inv(value: int) -> int:
    """Return the multiplicative inverse of a nonzero GF(256) byte."""

    _check_byte(value)
    if value == 0:
        raise ZeroDivisionError("zero has no multiplicative inverse in GF(256)")
    return GF_EXP[255 - GF_LOG[value]]


def gf_div(numerator: int, denominator: int) -> int:
    """Divide two bytes in GF(256), rejecting a zero denominator."""

    _check_byte(numerator, "numerator")
    _check_byte(denominator, "denominator")
    if denominator == 0:
        raise ZeroDivisionError("division by zero in GF(256)")
    if numerator == 0:
        return 0
    return GF_EXP[(GF_LOG[numerator] - GF_LOG[denominator]) % 255]


def inverse_oracle(numerator: int, denominator: int) -> int:
    """GSQR's nonlinear quotient oracle, ``numerator / denominator``."""

    return gf_div(numerator, denominator)


# Consecutive powers of the primitive generator are fixed column locators.
# None is zero and all are distinct, which makes every pair of parity-check
# columns linearly independent.
GUIDANCE_CONSTANTS = tuple(GF_EXP[index] for index in range(DATA_COLUMNS))

# A 256-entry multiplication row per guidance constant removes log-table work
# from parity and syndrome loops.  Tuples are immutable and safe to share with
# forked workers (spawned workers deterministically rebuild the same tables).
GUIDANCE_MUL_TABLES = tuple(
    bytes(gf_mul(guide, value) for value in range(256))
    for guide in GUIDANCE_CONSTANTS
)


@dataclass(frozen=True)
class Guidance:
    """The complete format guidance fixed before any source bytes are read."""

    field: str = "GF(256)"
    field_polynomial: int = FIELD_POLYNOMIAL
    primitive_generator: int = FIELD_GENERATOR
    data_columns: int = DATA_COLUMNS
    parity_columns: int = PARITY_COLUMNS
    constants: tuple[int, ...] = GUIDANCE_CONSTANTS
    piece_bytes: int = PIECE_BYTES
    compression: str = "zlib"
    compression_level: int = COMPRESSION_LEVEL

    def validate(self) -> None:
        if self.field != "GF(256)":
            raise FormatError("GSQR-256C requires GF(256)")
        if self.field_polynomial != 0x11D or self.primitive_generator != 0x02:
            raise FormatError("GF(256) polynomial or generator mismatch")
        if self.data_columns != len(self.constants) or self.data_columns != 8:
            raise FormatError("GSQR-256C requires eight data-column locators")
        if self.parity_columns != 2:
            raise FormatError("GSQR-256C requires two parity columns")
        if len(set(self.constants)) != len(self.constants):
            raise FormatError("guidance constants must be distinct")
        if any(not 1 <= value <= 0xFF for value in self.constants):
            raise FormatError("guidance constants must be nonzero field units")
        if self.piece_bytes <= 0:
            raise FormatError("piece size must be positive")
        if self.compression != "zlib" or not 0 <= self.compression_level <= 9:
            raise FormatError("unsupported compression guidance")

    def descriptor(self) -> dict[str, Any]:
        return {
            "field": self.field,
            "field_polynomial": f"0x{self.field_polynomial:03x}",
            "primitive_generator": self.primitive_generator,
            "data_columns": self.data_columns,
            "parity_columns": self.parity_columns,
            "constants": list(self.constants),
            "piece_bytes": self.piece_bytes,
            "compression": self.compression,
            "compression_level": self.compression_level,
        }

    @property
    def identifier(self) -> str:
        return hashlib.sha256(_canonical_json(self.descriptor())).hexdigest()


FIXED_GUIDANCE = Guidance()
FIXED_GUIDANCE.validate()


def _zlib_compress_bound(source_bytes: int) -> int:
    """Return zlib's documented conservative ``compressBound`` calculation."""

    return (
        source_bytes
        + (source_bytes >> 12)
        + (source_bytes >> 14)
        + (source_bytes >> 25)
        + 13
    )


MAX_FRAME_BYTES = FRAME_HEADER.size + _zlib_compress_bound(PIECE_BYTES)
MAX_ROWS = (MAX_FRAME_BYTES + DATA_COLUMNS - 1) // DATA_COLUMNS


@dataclass(frozen=True)
class ShardState:
    name: str
    values: bytes
    healthy: bool
    reason: str | None = None


@dataclass(frozen=True)
class PieceReport:
    index: int
    repaired_columns: tuple[str, ...]
    inferred_column: str | None
    oracle_rows: int


@dataclass(frozen=True)
class PieceResult:
    index: int
    data_columns: tuple[bytes, ...]
    parity_columns: tuple[bytes, bytes]
    original: bytes
    repaired_columns: tuple[str, ...]
    inferred_column: str | None
    oracle_rows: int

    def report(self) -> PieceReport:
        return PieceReport(
            index=self.index,
            repaired_columns=self.repaired_columns,
            inferred_column=self.inferred_column,
            oracle_rows=self.oracle_rows,
        )


# ---------------------------------------------------------------------------
# Piece framing and parity construction
# ---------------------------------------------------------------------------

def _sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _data_name(column: int) -> str:
    return f"d{column:02d}.bin"


def _parity_name(column: int) -> str:
    return f"p{column}.bin"


def _piece_name(index: int) -> str:
    return f"piece-{index:08d}"


def _make_parity(data_columns: Sequence[bytes]) -> tuple[bytes, bytes]:
    """Build equal-width P0/P1 parity for eight byte columns."""

    if len(data_columns) != DATA_COLUMNS:
        raise ValueError("wrong data-column count")
    rows = len(data_columns[0])
    if any(len(column) != rows for column in data_columns):
        raise ValueError("data columns have unequal lengths")

    p0 = bytearray(rows)
    p1 = bytearray(rows)
    for column_index, column in enumerate(data_columns):
        multiply = GUIDANCE_MUL_TABLES[column_index]
        for row, symbol in enumerate(column):
            p0[row] ^= symbol
            p1[row] ^= multiply[symbol]
    return bytes(p0), bytes(p1)


def _frame_piece(index: int, original: bytes) -> bytes:
    compressed = zlib.compress(original, COMPRESSION_LEVEL)
    header = FRAME_HEADER.pack(
        FRAME_MAGIC,
        index,
        len(original),
        len(compressed),
        bytes.fromhex(FIXED_GUIDANCE.identifier),
        hashlib.sha256(original).digest(),
        hashlib.sha256(compressed).digest(),
    )
    return header + compressed


def _columnize(frame: bytes) -> tuple[tuple[bytes, ...], tuple[bytes, bytes]]:
    rows = (len(frame) + DATA_COLUMNS - 1) // DATA_COLUMNS
    padded = frame + bytes(rows * DATA_COLUMNS - len(frame))
    data = tuple(padded[column::DATA_COLUMNS] for column in range(DATA_COLUMNS))
    return data, _make_parity(data)


def _assemble(data_columns: Sequence[bytes]) -> bytes:
    if len(data_columns) != DATA_COLUMNS:
        raise FormatError("wrong number of data columns")
    rows = len(data_columns[0])
    if any(len(column) != rows for column in data_columns):
        raise FormatError("data columns have unequal lengths")
    padded = bytearray(rows * DATA_COLUMNS)
    for column, values in enumerate(data_columns):
        padded[column::DATA_COLUMNS] = values
    return bytes(padded)


def _parse_frame(padded: bytes, expected_index: int) -> bytes:
    """Validate, decompress, and authenticate one parity-protected frame."""

    if len(padded) < FRAME_HEADER.size:
        raise UnrecoverableError(f"piece {expected_index}: frame is too short")
    try:
        (
            magic,
            index,
            original_bytes,
            compressed_bytes,
            guidance_digest,
            original_digest,
            compressed_digest,
        ) = FRAME_HEADER.unpack_from(padded)
    except struct.error as exc:
        raise UnrecoverableError(f"piece {expected_index}: invalid frame header") from exc

    if magic != FRAME_MAGIC:
        raise UnrecoverableError(f"piece {expected_index}: frame magic mismatch")
    if index != expected_index:
        raise UnrecoverableError(
            f"piece {expected_index}: embedded piece index is {index}"
        )
    if guidance_digest.hex() != FIXED_GUIDANCE.identifier:
        raise UnrecoverableError(f"piece {expected_index}: embedded guidance mismatch")
    if original_bytes > PIECE_BYTES:
        raise UnrecoverableError(f"piece {expected_index}: impossible original length")

    frame_bytes = FRAME_HEADER.size + compressed_bytes
    if frame_bytes > len(padded):
        raise UnrecoverableError(f"piece {expected_index}: truncated compressed frame")
    if any(padded[frame_bytes:]):
        raise UnrecoverableError(f"piece {expected_index}: nonzero frame padding")

    compressed = padded[FRAME_HEADER.size:frame_bytes]
    if hashlib.sha256(compressed).digest() != compressed_digest:
        raise UnrecoverableError(f"piece {expected_index}: compressed digest mismatch")

    # The max_length cap prevents a crafted compressed frame from expanding
    # beyond the fixed piece boundary before its embedded length is checked.
    try:
        decompressor = zlib.decompressobj()
        original = decompressor.decompress(compressed, PIECE_BYTES + 1)
    except zlib.error as exc:
        raise UnrecoverableError(f"piece {expected_index}: invalid zlib stream") from exc
    if len(original) > PIECE_BYTES or decompressor.unconsumed_tail:
        raise UnrecoverableError(f"piece {expected_index}: decompressed piece exceeds limit")
    if not decompressor.eof or decompressor.unused_data:
        raise UnrecoverableError(f"piece {expected_index}: incomplete or trailing zlib stream")
    if len(original) != original_bytes:
        raise UnrecoverableError(f"piece {expected_index}: original length mismatch")
    if hashlib.sha256(original).digest() != original_digest:
        raise UnrecoverableError(f"piece {expected_index}: original digest mismatch")
    return original


# ---------------------------------------------------------------------------
# Bounded I/O and deterministic archive metadata
# ---------------------------------------------------------------------------

def _write_bytes(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("wb") as handle:
        handle.write(data)


def _read_regular_file_bounded(path: Path, maximum_bytes: int) -> bytes:
    """Read through a bounded descriptor, refusing special/oversized files."""

    flags = os.O_RDONLY
    flags |= getattr(os, "O_CLOEXEC", 0)
    flags |= getattr(os, "O_NOFOLLOW", 0)
    flags |= getattr(os, "O_BINARY", 0)
    descriptor = os.open(path, flags)
    try:
        metadata = os.fstat(descriptor)
        if not stat.S_ISREG(metadata.st_mode):
            raise OSError(f"not a regular file: {path}")
        if metadata.st_size > maximum_bytes:
            raise OSError(f"file exceeds {maximum_bytes}-byte bound: {path}")
        with os.fdopen(descriptor, "rb") as handle:
            descriptor = -1
            raw = handle.read(maximum_bytes + 1)
        if len(raw) > maximum_bytes:
            raise OSError(f"file grew beyond its bound while reading: {path}")
        return raw
    finally:
        if descriptor >= 0:
            os.close(descriptor)


def _piece_metadata(
    index: int,
    data_columns: Sequence[bytes],
    parity_columns: Sequence[bytes],
) -> dict[str, Any]:
    rows = len(data_columns[0])
    return {
        "index": index,
        "rows": rows,
        "data": [
            {
                "name": _data_name(column),
                "bytes": rows,
                "sha256": _sha256(data_columns[column]),
            }
            for column in range(DATA_COLUMNS)
        ],
        "parity": [
            {
                "name": _parity_name(column),
                "bytes": rows,
                "sha256": _sha256(parity_columns[column]),
            }
            for column in range(PARITY_COLUMNS)
        ],
    }


def _write_piece(
    archive: Path,
    index: int,
    data_columns: Sequence[bytes],
    parity_columns: Sequence[bytes],
) -> dict[str, Any]:
    piece_dir = archive / _piece_name(index)
    piece_dir.mkdir(parents=True, exist_ok=False)
    for column, values in enumerate(data_columns):
        _write_bytes(piece_dir / _data_name(column), values)
    for column, values in enumerate(parity_columns):
        _write_bytes(piece_dir / _parity_name(column), values)
    return _piece_metadata(index, data_columns, parity_columns)


def _atomic_directory(target: Path, builder: Callable[[Path], None]) -> None:
    """Build privately, then publish by one rename after full validation."""

    target = target.resolve()
    if target.exists():
        raise CodecError(f"destination already exists: {target}")
    target.parent.mkdir(parents=True, exist_ok=True)
    temporary = Path(
        tempfile.mkdtemp(prefix=f".{target.name}.tmp-", dir=str(target.parent))
    )
    try:
        builder(temporary)
        os.replace(temporary, target)
    except BaseException:
        shutil.rmtree(temporary, ignore_errors=True)
        raise


def _require_int(
    value: Any,
    label: str,
    minimum: int = 0,
    maximum: int | None = None,
) -> int:
    if not isinstance(value, int) or isinstance(value, bool) or value < minimum:
        raise FormatError(f"manifest field {label!r} is invalid")
    if maximum is not None and value > maximum:
        raise FormatError(f"manifest field {label!r} exceeds its bound")
    return value


def _require_digest(value: Any, label: str) -> str:
    if (
        not isinstance(value, str)
        or len(value) != 64
        or value != value.lower()
    ):
        raise FormatError(f"manifest digest {label!r} is invalid")
    try:
        bytes.fromhex(value)
    except ValueError as exc:
        raise FormatError(f"manifest digest {label!r} is invalid") from exc
    return value


def _load_manifest(archive: Path | str) -> tuple[Path, dict[str, Any]]:
    archive_path = Path(archive).resolve()
    manifest_path = archive_path / MANIFEST_NAME
    try:
        raw = _read_regular_file_bounded(manifest_path, MAX_MANIFEST_BYTES)
    except OSError as exc:
        raise FormatError(f"cannot read {manifest_path}: {exc}") from exc
    try:
        manifest = json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise FormatError("manifest is not valid UTF-8 JSON") from exc
    if not isinstance(manifest, dict):
        raise FormatError("manifest root must be an object")
    if manifest.get("format") != FORMAT_NAME:
        raise FormatError("archive format name mismatch")
    if manifest.get("implementation") != IMPLEMENTATION_NAME:
        raise FormatError("archive implementation name mismatch")
    if manifest.get("version") != FORMAT_VERSION:
        raise FormatError("archive format version mismatch")
    if manifest.get("guidance_id") != FIXED_GUIDANCE.identifier:
        raise FormatError("archive guidance ID does not match fixed guidance")
    if manifest.get("guidance") != FIXED_GUIDANCE.descriptor():
        raise FormatError("archive guidance descriptor does not match fixed guidance")

    source = manifest.get("source")
    if not isinstance(source, dict):
        raise FormatError("manifest source must be an object")
    source_bytes = _require_int(source.get("bytes"), "source.bytes")
    _require_digest(source.get("sha256"), "source.sha256")
    if not isinstance(source.get("name"), str):
        raise FormatError("manifest source.name is invalid")

    pieces = manifest.get("pieces")
    if not isinstance(pieces, list):
        raise FormatError("manifest pieces must be an array")
    if _require_int(manifest.get("piece_count"), "piece_count") != len(pieces):
        raise FormatError("manifest piece count mismatch")
    expected_count = (source_bytes + PIECE_BYTES - 1) // PIECE_BYTES
    if len(pieces) != expected_count:
        raise FormatError("piece count is inconsistent with fixed piece size")

    for expected_index, piece in enumerate(pieces):
        if not isinstance(piece, dict) or piece.get("index") != expected_index:
            raise FormatError(f"piece {expected_index}: invalid index or metadata")
        rows = _require_int(
            piece.get("rows"),
            f"piece {expected_index}.rows",
            minimum=1,
            maximum=MAX_ROWS,
        )
        data_meta = piece.get("data")
        parity_meta = piece.get("parity")
        if not isinstance(data_meta, list) or len(data_meta) != DATA_COLUMNS:
            raise FormatError(f"piece {expected_index}: data metadata count mismatch")
        if not isinstance(parity_meta, list) or len(parity_meta) != PARITY_COLUMNS:
            raise FormatError(f"piece {expected_index}: parity metadata count mismatch")
        for column, item in enumerate(data_meta):
            if not isinstance(item, dict):
                raise FormatError(f"piece {expected_index}: invalid data metadata")
            if item.get("name") != _data_name(column) or item.get("bytes") != rows:
                raise FormatError(f"piece {expected_index}: data metadata mismatch")
            _require_digest(item.get("sha256"), f"piece {expected_index}.d{column}")
        for column, item in enumerate(parity_meta):
            if not isinstance(item, dict):
                raise FormatError(f"piece {expected_index}: invalid parity metadata")
            if item.get("name") != _parity_name(column) or item.get("bytes") != rows:
                raise FormatError(f"piece {expected_index}: parity metadata mismatch")
            _require_digest(item.get("sha256"), f"piece {expected_index}.p{column}")
    return archive_path, manifest


# ---------------------------------------------------------------------------
# Bounded deterministic process concurrency
# ---------------------------------------------------------------------------

T = TypeVar("T")
R = TypeVar("R")


def _worker_count(workers: int | None, piece_count: int) -> int:
    requested = DEFAULT_WORKERS if workers is None else workers
    if not isinstance(requested, int) or isinstance(requested, bool):
        raise CodecError("worker count must be an integer")
    if not 1 <= requested <= MAX_WORKERS:
        raise CodecError(f"worker count must be between 1 and {MAX_WORKERS}")
    # A zero-piece archive needs no pool; returning one keeps the serial path.
    return max(1, min(requested, max(1, piece_count)))


def _ordered_process_map(
    worker: Callable[[T], R],
    tasks: Iterable[T],
    workers: int,
) -> Iterator[R]:
    """Run bounded work concurrently and yield results in submission order.

    Only ``workers * IN_FLIGHT_PER_WORKER`` tasks may be resident.  Waiting for
    the oldest submitted future can temporarily sacrifice throughput, but it
    is the simple invariant that makes archive publication deterministic and
    keeps memory bounded by a small multiple of one piece.
    """

    if workers == 1:
        for task in tasks:
            yield worker(task)
        return

    iterator = iter(tasks)
    pending: deque[Future[R]] = deque()
    executor = ProcessPoolExecutor(max_workers=workers)
    try:
        for _ in range(workers * IN_FLIGHT_PER_WORKER):
            try:
                pending.append(executor.submit(worker, next(iterator)))
            except StopIteration:
                break
        while pending:
            yield pending.popleft().result()
            try:
                pending.append(executor.submit(worker, next(iterator)))
            except StopIteration:
                pass
    except BaseException:
        for future in pending:
            future.cancel()
        raise
    finally:
        executor.shutdown(wait=True, cancel_futures=True)


def _encode_piece_worker(task: tuple[int, bytes]) -> tuple[int, tuple[bytes, ...], tuple[bytes, bytes]]:
    index, original = task
    data, parity = _columnize(_frame_piece(index, original))
    return index, data, parity


def _repair_piece_worker(task: tuple[str, dict[str, Any]]) -> PieceResult:
    archive, piece_meta = task
    return _repair_piece(Path(archive), piece_meta)


# ---------------------------------------------------------------------------
# Encoding
# ---------------------------------------------------------------------------

def encode_file(
    source: Path | str,
    archive: Path | str,
    workers: int | None = DEFAULT_WORKERS,
) -> dict[str, Any]:
    """Compress and encode ``source`` into a deterministic GSQR archive."""

    source_path = Path(source).resolve()
    archive_path = Path(archive)
    if not source_path.is_file():
        raise CodecError(f"input is not a regular file: {source_path}")
    estimated_pieces = (
        source_path.stat().st_size + PIECE_BYTES - 1
    ) // PIECE_BYTES
    effective_workers = _worker_count(workers, estimated_pieces)
    manifest: dict[str, Any] = {}

    def build(temporary: Path) -> None:
        source_hash = hashlib.sha256()
        source_size = 0
        pieces: list[dict[str, Any]] = []

        def tasks() -> Iterator[tuple[int, bytes]]:
            nonlocal source_size
            with source_path.open("rb") as handle:
                index = 0
                while True:
                    original = handle.read(PIECE_BYTES)
                    if not original:
                        break
                    source_hash.update(original)
                    source_size += len(original)
                    yield index, original
                    index += 1

        for expected_index, result in enumerate(
            _ordered_process_map(_encode_piece_worker, tasks(), effective_workers)
        ):
            index, data, parity = result
            if index != expected_index:
                raise RuntimeError("worker results escaped deterministic piece order")
            pieces.append(_write_piece(temporary, index, data, parity))

        manifest.update(
            {
                "format": FORMAT_NAME,
                "implementation": IMPLEMENTATION_NAME,
                "version": FORMAT_VERSION,
                "guidance_id": FIXED_GUIDANCE.identifier,
                "guidance": FIXED_GUIDANCE.descriptor(),
                "source": {
                    "name": source_path.name,
                    "bytes": source_size,
                    "sha256": source_hash.hexdigest(),
                },
                "piece_count": len(pieces),
                "pieces": pieces,
            }
        )
        # Never publish an archive that its own bounded reader would reject.
        # A future large-object format can replace this JSON manifest with a
        # compact index without weakening the current read bound.
        _write_bytes(temporary / MANIFEST_NAME, _bounded_manifest_bytes(manifest))

    _atomic_directory(archive_path, build)
    return manifest


# ---------------------------------------------------------------------------
# Syndrome inference and recovery
# ---------------------------------------------------------------------------

def _read_shard(
    path: Path,
    name: str,
    expected_bytes: int,
    expected_digest: str,
    rows: int,
) -> ShardState:
    try:
        raw = _read_regular_file_bounded(path, expected_bytes)
    except OSError as exc:
        return ShardState(name, bytes(rows), False, str(exc))
    if len(raw) != expected_bytes:
        return ShardState(
            name,
            bytes(rows),
            False,
            f"length {len(raw)} != {expected_bytes}",
        )
    if _sha256(raw) != expected_digest:
        # Retaining aligned damaged bytes lets parity independently infer the
        # column; multi-erasure equations later ignore every unhealthy shard.
        return ShardState(name, raw, False, "SHA-256 mismatch")
    return ShardState(name, raw, True)


def _syndromes(
    data: Sequence[bytes],
    p0: bytes,
    p1: bytes,
) -> tuple[bytes, bytes]:
    rows = len(p0)
    s0 = bytearray(p0)
    s1 = bytearray(p1)
    for column_index, column in enumerate(data):
        multiply = GUIDANCE_MUL_TABLES[column_index]
        for row, symbol in enumerate(column):
            # Addition and subtraction are both XOR in characteristic two.
            s0[row] ^= symbol
            s1[row] ^= multiply[symbol]
    return bytes(s0), bytes(s1)


def _infer_single_column(s0: bytes, s1: bytes) -> tuple[str | None, int]:
    """Match the complete syndrome-vector direction to fixed guidance.

    A single pivot proposes the quotient, but every row must have exactly the
    same projective direction.  Reducing each vector to one aggregate scalar
    would permit XOR cancellation and is deliberately forbidden.
    """

    nonzero0 = [index for index, value in enumerate(s0) if value]
    nonzero1 = [index for index, value in enumerate(s1) if value]
    if not nonzero0 and not nonzero1:
        return None, 0
    if nonzero0 and not nonzero1:
        return _parity_name(0), len(nonzero0)
    if not nonzero0 and nonzero1:
        return _parity_name(1), len(nonzero1)

    pivot = nonzero0[0]
    candidate = inverse_oracle(s1[pivot], s0[pivot])
    try:
        column = GUIDANCE_CONSTANTS.index(candidate)
    except ValueError as exc:
        raise UnrecoverableError(
            f"syndrome quotient 0x{candidate:02x} does not match fixed guidance"
        ) from exc
    multiply = GUIDANCE_MUL_TABLES[column]
    if any(s1[row] != multiply[s0[row]] for row in range(len(s0))):
        raise UnrecoverableError(
            "syndrome vectors do not match one fixed guidance column"
        )
    return _data_name(column), len(nonzero0)


def _recover_known_erasures(
    data_states: Sequence[ShardState],
    parity_states: Sequence[ShardState],
    bad_data: Sequence[int],
    bad_parity: Sequence[int],
) -> tuple[tuple[bytes, ...], tuple[bytes, bytes]]:
    """Solve every supported one- or two-erasure pattern byte by byte."""

    data = [bytearray(state.values) for state in data_states]
    parity = [bytearray(state.values) for state in parity_states]
    rows = len(data[0])

    if len(bad_data) + len(bad_parity) > 2:
        names = [data_states[index].name for index in bad_data] + [
            parity_states[index].name for index in bad_parity
        ]
        raise UnrecoverableError(
            "more than two known erasures: " + ", ".join(names)
        )

    if not bad_data:
        clean_data = tuple(bytes(column) for column in data)
        return clean_data, _make_parity(clean_data)

    if len(bad_data) == 1:
        missing = bad_data[0]
        if 0 not in bad_parity:
            # P0 directly gives the XOR of the missing data and all survivors.
            for row in range(rows):
                value = parity[0][row]
                for column in range(DATA_COLUMNS):
                    if column != missing:
                        value ^= data[column][row]
                data[missing][row] = value
        elif 1 not in bad_parity:
            # With only P1, divide the weighted remainder by the fixed locator.
            guide = GUIDANCE_CONSTANTS[missing]
            for row in range(rows):
                weighted = parity[1][row]
                for column in range(DATA_COLUMNS):
                    if column != missing:
                        weighted ^= GUIDANCE_MUL_TABLES[column][data[column][row]]
                data[missing][row] = gf_div(weighted, guide)
        else:
            raise UnrecoverableError("data erasure has no surviving parity column")

    elif len(bad_data) == 2:
        if bad_parity:
            raise UnrecoverableError("two data erasures require both parity columns")
        first, second = bad_data
        g_first = GUIDANCE_CONSTANTS[first]
        g_second = GUIDANCE_CONSTANTS[second]
        denominator = g_first ^ g_second
        for row in range(rows):
            remainder0 = parity[0][row]
            remainder1 = parity[1][row]
            for column in range(DATA_COLUMNS):
                if column in (first, second):
                    continue
                remainder0 ^= data[column][row]
                remainder1 ^= GUIDANCE_MUL_TABLES[column][data[column][row]]
            first_value = gf_div(
                remainder1 ^ gf_mul(g_second, remainder0),
                denominator,
            )
            data[first][row] = first_value
            data[second][row] = remainder0 ^ first_value

    clean_data = tuple(bytes(column) for column in data)
    return clean_data, _make_parity(clean_data)


def _repair_piece(archive: Path, piece_meta: dict[str, Any]) -> PieceResult:
    index = piece_meta["index"]
    rows = piece_meta["rows"]
    piece_dir = archive / _piece_name(index)
    data_states = [
        _read_shard(
            piece_dir / item["name"],
            item["name"],
            item["bytes"],
            item["sha256"],
            rows,
        )
        for item in piece_meta["data"]
    ]
    parity_states = [
        _read_shard(
            piece_dir / item["name"],
            item["name"],
            item["bytes"],
            item["sha256"],
            rows,
        )
        for item in piece_meta["parity"]
    ]
    bad_data = [position for position, state in enumerate(data_states) if not state.healthy]
    bad_parity = [
        position for position, state in enumerate(parity_states) if not state.healthy
    ]
    repaired_names = tuple(
        [data_states[position].name for position in bad_data]
        + [parity_states[position].name for position in bad_parity]
    )

    s0, s1 = _syndromes(
        [state.values for state in data_states],
        parity_states[0].values,
        parity_states[1].values,
    )
    inferred: str | None = None
    oracle_rows = 0
    if len(repaired_names) <= 1:
        inferred, oracle_rows = _infer_single_column(s0, s1)
        if repaired_names and inferred is not None and inferred != repaired_names[0]:
            raise UnrecoverableError(
                f"piece {index}: digest identifies {repaired_names[0]}, "
                f"but GSQR guidance identifies {inferred}"
            )
        if not repaired_names and inferred is not None:
            raise UnrecoverableError(
                f"piece {index}: parity identifies {inferred}, but all digests pass"
            )

    try:
        data, parity = _recover_known_erasures(
            data_states,
            parity_states,
            bad_data,
            bad_parity,
        )
    except UnrecoverableError as exc:
        raise UnrecoverableError(f"piece {index}: {exc}") from exc

    # A successful correction must reproduce every original shard digest,
    # both parity equations, the framed compressed digest, and the raw digest.
    for column, values in enumerate(data):
        if _sha256(values) != piece_meta["data"][column]["sha256"]:
            raise UnrecoverableError(
                f"piece {index}: restored {_data_name(column)} digest mismatch"
            )
    for column, values in enumerate(parity):
        if _sha256(values) != piece_meta["parity"][column]["sha256"]:
            raise UnrecoverableError(
                f"piece {index}: restored {_parity_name(column)} digest mismatch"
            )
    original = _parse_frame(_assemble(data), index)
    return PieceResult(
        index=index,
        data_columns=data,
        parity_columns=parity,
        original=original,
        repaired_columns=repaired_names,
        inferred_column=inferred,
        oracle_rows=oracle_rows,
    )


def _validate_source_summary(
    manifest: dict[str, Any], source_hash: Any, source_bytes: int
) -> None:
    if source_bytes != manifest["source"]["bytes"]:
        raise UnrecoverableError("reconstructed source length does not match manifest")
    if source_hash.hexdigest() != manifest["source"]["sha256"]:
        raise UnrecoverableError("reconstructed source digest does not match manifest")


def _repair_tasks(
    archive: Path,
    manifest: dict[str, Any],
) -> Iterator[tuple[str, dict[str, Any]]]:
    for piece in manifest["pieces"]:
        yield str(archive), piece


def inspect_archive(
    archive: Path | str,
    workers: int | None = DEFAULT_WORKERS,
) -> tuple[dict[str, Any], list[PieceReport]]:
    """Verify and simulate healing without modifying the archive."""

    archive_path, manifest = _load_manifest(archive)
    effective_workers = _worker_count(workers, manifest["piece_count"])
    reports: list[PieceReport] = []
    source_hash = hashlib.sha256()
    source_bytes = 0
    for result in _ordered_process_map(
        _repair_piece_worker,
        _repair_tasks(archive_path, manifest),
        effective_workers,
    ):
        source_hash.update(result.original)
        source_bytes += len(result.original)
        reports.append(result.report())
    _validate_source_summary(manifest, source_hash, source_bytes)
    return manifest, reports


def heal_archive(
    source: Path | str,
    destination: Path | str,
    workers: int | None = DEFAULT_WORKERS,
) -> list[PieceReport]:
    """Write a canonical healed archive without mutating its source."""

    source_path, manifest = _load_manifest(source)
    destination_path = Path(destination).resolve()
    if destination_path == source_path or source_path in destination_path.parents:
        raise CodecError("healed archive destination must be outside the source archive")
    effective_workers = _worker_count(workers, manifest["piece_count"])
    reports: list[PieceReport] = []

    def build(temporary: Path) -> None:
        source_hash = hashlib.sha256()
        source_bytes = 0
        for result in _ordered_process_map(
            _repair_piece_worker,
            _repair_tasks(source_path, manifest),
            effective_workers,
        ):
            _write_piece(
                temporary,
                result.index,
                result.data_columns,
                result.parity_columns,
            )
            source_hash.update(result.original)
            source_bytes += len(result.original)
            reports.append(result.report())
        _validate_source_summary(manifest, source_hash, source_bytes)
        _write_bytes(temporary / MANIFEST_NAME, _bounded_manifest_bytes(manifest))

    _atomic_directory(destination_path, build)
    return reports


def decode_archive(
    archive: Path | str,
    destination: Path | str,
    workers: int | None = DEFAULT_WORKERS,
) -> list[PieceReport]:
    """Decode in piece order, applying recoverable repairs only in memory."""

    archive_path, manifest = _load_manifest(archive)
    destination_path = Path(destination).resolve()
    if destination_path.exists():
        raise CodecError(f"destination already exists: {destination_path}")
    if archive_path in destination_path.parents:
        raise CodecError("decoded-file destination must be outside the archive")
    effective_workers = _worker_count(workers, manifest["piece_count"])
    destination_path.parent.mkdir(parents=True, exist_ok=True)
    reports: list[PieceReport] = []
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{destination_path.name}.tmp-",
        dir=str(destination_path.parent),
    )
    try:
        with os.fdopen(descriptor, "wb") as handle:
            source_hash = hashlib.sha256()
            source_bytes = 0
            for result in _ordered_process_map(
                _repair_piece_worker,
                _repair_tasks(archive_path, manifest),
                effective_workers,
            ):
                handle.write(result.original)
                source_hash.update(result.original)
                source_bytes += len(result.original)
                reports.append(result.report())
            handle.flush()
            os.fsync(handle.fileno())
        _validate_source_summary(manifest, source_hash, source_bytes)
        os.replace(temporary_name, destination_path)
    except BaseException:
        try:
            os.unlink(temporary_name)
        except FileNotFoundError:
            pass
        raise
    return reports


# ---------------------------------------------------------------------------
# Built-in validation and CLI
# ---------------------------------------------------------------------------

def _selftest_check(condition: bool, message: str) -> None:
    # Unlike ``assert``, this remains active under ``python -O``.
    if not condition:
        raise AssertionError(message)


def _damage_byte(path: Path, offset: int = 0) -> None:
    raw = bytearray(path.read_bytes())
    if not raw:
        raise AssertionError(f"cannot damage empty test shard {path}")
    raw[offset % len(raw)] ^= 0xA5
    path.write_bytes(raw)


def _tree_snapshot(root: Path) -> dict[str, bytes]:
    return {
        path.relative_to(root).as_posix(): path.read_bytes()
        for path in sorted(root.rglob("*"))
        if path.is_file()
    }


def run_selftest() -> None:
    """Run serial/concurrent round-trip and injected-damage checks."""

    # Verify the arithmetic independently of archive operations.
    known_products = ((0x02, 0x80, 0x1D), (0x53, 0xCA, 0x8F), (0x57, 0x83, 0x31))
    for left, right, expected in known_products:
        _selftest_check(gf_mul(left, right) == expected, "GF(256) product mismatch")
    for value in range(256):
        for divisor in (1, 2, 3, 0x53, 0x80, 0xFF):
            _selftest_check(
                gf_div(gf_mul(value, divisor), divisor) == value,
                "GF(256) division did not reverse multiplication",
            )

    rng = random.Random(0x256C)
    with tempfile.TemporaryDirectory(prefix="gsqr-256c-selftest-") as root_name:
        root = Path(root_name)
        source = root / "source.bin"
        source.write_bytes(
            bytes(rng.randrange(256) for _ in range(2 * PIECE_BYTES + 777))
        )
        serial = root / "serial.gsqr"
        concurrent = root / "concurrent.gsqr"
        encode_file(source, serial, workers=1)
        encode_file(source, concurrent, workers=3)
        _selftest_check(
            _tree_snapshot(serial) == _tree_snapshot(concurrent),
            "serial and concurrent archives differ",
        )

        damaged = root / "damaged.gsqr"
        shutil.copytree(concurrent, damaged)
        _damage_byte(damaged / _piece_name(0) / _data_name(5), 11)
        _, reports = inspect_archive(damaged, workers=2)
        _selftest_check(
            reports[0].inferred_column == _data_name(5),
            "syndrome quotient inferred the wrong column",
        )
        decoded = root / "decoded.bin"
        decode_archive(damaged, decoded, workers=3)
        _selftest_check(
            decoded.read_bytes() == source.read_bytes(),
            "concurrent repaired decode differs from source",
        )

        two_missing = root / "two-missing.gsqr"
        shutil.copytree(concurrent, two_missing)
        (two_missing / _piece_name(1) / _data_name(1)).unlink()
        (two_missing / _piece_name(1) / _parity_name(0)).unlink()
        healed = root / "healed.gsqr"
        heal_archive(two_missing, healed, workers=3)
        _selftest_check(
            _tree_snapshot(healed) == _tree_snapshot(concurrent),
            "two-erasure healing did not restore canonical archive bytes",
        )

        too_many = root / "too-many.gsqr"
        shutil.copytree(concurrent, too_many)
        for column in (0, 2, 6):
            (too_many / _piece_name(0) / _data_name(column)).unlink()
        refused = False
        try:
            inspect_archive(too_many, workers=2)
        except UnrecoverableError:
            refused = True
        _selftest_check(refused, "three erasures were not refused")


def _format_repairs(results: Iterable[PieceReport]) -> str:
    descriptions: list[str] = []
    for result in results:
        if not result.repaired_columns:
            continue
        detail = f"piece {result.index}: {', '.join(result.repaired_columns)}"
        if result.inferred_column:
            detail += (
                f"; guidance matched {result.inferred_column}"
                f" on {result.oracle_rows} row(s)"
            )
        descriptions.append(detail)
    return "\n".join(descriptions)


def _add_workers(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        "--workers",
        type=int,
        default=DEFAULT_WORKERS,
        help=f"bounded piece workers (default: {DEFAULT_WORKERS}, max: {MAX_WORKERS})",
    )


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="GSQR-256C concurrent fixed-guidance column recovery."
    )
    commands = parser.add_subparsers(dest="command", required=True)

    encode = commands.add_parser("encode", help="encode a file into a GSQR archive")
    encode.add_argument("input", type=Path)
    encode.add_argument("archive", type=Path)
    _add_workers(encode)

    verify = commands.add_parser(
        "verify", help="verify and simulate healing without changing the archive"
    )
    verify.add_argument("archive", type=Path)
    _add_workers(verify)

    heal = commands.add_parser(
        "heal", help="write a canonical healed copy without changing its source"
    )
    heal.add_argument("archive", type=Path)
    heal.add_argument("output_archive", type=Path)
    _add_workers(heal)

    decode = commands.add_parser(
        "decode", help="decode with in-memory healing into the original byte stream"
    )
    decode.add_argument("archive", type=Path)
    decode.add_argument("output", type=Path)
    _add_workers(decode)

    commands.add_parser("guidance", help="print the frozen GSQR guidance descriptor")
    commands.add_parser("selftest", help="run arithmetic, concurrency, and damage tests")
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    parser = _build_parser()
    args = parser.parse_args(argv)
    try:
        if args.command == "encode":
            manifest = encode_file(args.input, args.archive, workers=args.workers)
            print(
                f"encoded {manifest['source']['bytes']} bytes into "
                f"{manifest['piece_count']} piece(s): {Path(args.archive).resolve()}"
            )
            return 0
        if args.command == "verify":
            _, reports = inspect_archive(args.archive, workers=args.workers)
            repairs = _format_repairs(reports)
            if repairs:
                print("archive is recoverable but not clean")
                print(repairs)
                return 1
            print("archive is clean")
            return 0
        if args.command == "heal":
            reports = heal_archive(
                args.archive,
                args.output_archive,
                workers=args.workers,
            )
            repairs = _format_repairs(reports)
            print(f"healed archive written to {Path(args.output_archive).resolve()}")
            print(repairs if repairs else "source archive was already clean")
            return 0
        if args.command == "decode":
            reports = decode_archive(args.archive, args.output, workers=args.workers)
            repairs = _format_repairs(reports)
            print(f"decoded file written to {Path(args.output).resolve()}")
            if repairs:
                print("in-memory repairs:")
                print(repairs)
            return 0
        if args.command == "guidance":
            output = {
                "algorithm": IMPLEMENTATION_NAME,
                "guidance_id": FIXED_GUIDANCE.identifier,
                **FIXED_GUIDANCE.descriptor(),
                "locator_formula": "S1 / S0 in GF(256)",
                "guarantee": (
                    "one parity-inferred, digest-confirmed bad column or "
                    "any two digest-identified erasures"
                ),
                "default_workers": DEFAULT_WORKERS,
                "max_workers": MAX_WORKERS,
            }
            print(json.dumps(output, sort_keys=True, indent=2))
            return 0
        if args.command == "selftest":
            run_selftest()
            print("self-test passed")
            return 0
        parser.error(f"unknown command: {args.command}")
    except CodecError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    return 2


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