"""Independent decoder for local Oak SQLite v1 committed state (stdlib-only).

No Oak code or executable is imported or invoked. BLAKE3 and zstd are decoded

by this repository's own pure-Python implementations (``pureblake3``,

``purezstd``), written from the public specifications, so a verified object is

checked by code that shares nothing with the Rust ``blake3`` crate or libzstd

that Oak links. ``OAK_BENCH_READER_BACKEND=native`` (or ``backend='native'``)

opts into the optional ``blake3``/``zstandard`` packages for speed; they wrap

the same upstream libraries Oak uses, so rows record which backend ran.

Evidence contract (2026-09-28.2):

- Scope: when scope paths are given, only entries on those paths (and the

  trees leading to them) are decoded. This is the same scope the Git

  committed-payload check reads with ``git ls-tree HEAD -- <scope>``, so the

  two readers can share an evidence class. Out-of-scope entries are never

  decoded, so repo-wide conditions the task does not own cannot fail Oak.

- Reader resource and time budgets are the harness's limits, not properties

  of Oak's stored content: exceeding one is Unavailable (exit 77), never

  Invalid. Invalid (exit 1) is reserved for positively identified bad bytes.

- Format pin: required tables/columns, a known schema migration and hash

  format, and at least one stored object of any kind (commit, tree, blob)

  verifying under the v1 rules. With the format pinned, every hash mismatch -

  HEAD commit, root tree, nested tree, blob, chunk - is Invalid (corruption).

  Only a store in which nothing verifies is an unrecognized format

  (Unavailable), indistinguishable from a future serialization.

- Every materialized object is re-hashed: blobs (``blake3(plaintext)``),

  chunks and their blob reassembly, trees (canonical ``kind\\tname\\thash\\tmode``

  lines), and commits (v1 7-field preimage; legacy 6-field and the server's

  microsecond-truncated timestamps are accepted exactly as Oak verifies them).

- Tree structure is validated independently of the hash: strictly ascending

  unique names, Oak's entry-name rules (no empty, ``.``/``..``, ``/``, NUL or

  control characters; ``.oak``/``.git`` are never stored), lowercase 64-hex

  object ids, and kind/mode pairs ``tree/tree`` and

  ``blob/{regular,executable,symlink}``.

- Symlinks: unscoped reads materialize them (after every regular file, so no

  write passes through a created link); scoped reads report them in

  ``nonregular_paths`` so the workflow oracle fails them exactly as Git's

  100644 mode check does.

- Chunked storage: a blob with no complete ``blobs`` row is reassembled from

  ``blob_chunks``/``chunks`` (contiguous offsets, per-chunk hash, total size,

  whole-blob hash). ``deep=True`` also cross-checks the chunk mapping of

  complete blobs and walks the commit chain. ``complete=True`` makes absent

  content Invalid (a fresh full clone must hold every committed object);

  otherwise absent content is Unavailable (it may be legitimately unhydrated).

"""

from __future__ import annotations

import hashlib

import importlib.metadata

import json

import os

from pathlib import Path

import re

import sqlite3

import sys

import tempfile

import time

try:  # Package import (harness) or standalone ``python -I reader.py``.

    from oakbench import pureblake3, purezstd

except ImportError:  # pragma: no cover - exercised by the subprocess tests

    sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

    from oakbench import pureblake3, purezstd

READER_VERSION = 'oak-sqlite-v1-2026-09-29.1'

# Newest Oak schema migration this reader is qualified against (oak 0.105.0).

# A newer migration is an unrecognized format: unmeasured, never corrupt.

KNOWN_MAX_MIGRATION = '0014_change_captures'

MAX_OBJECT = 32 * 1024 * 1024

MAX_TOTAL = 128 * 1024 * 1024

MAX_ENTRIES = 8192

MAX_DATABASE = 512 * 1024 * 1024

MAX_HISTORY = 4096

DEADLINE_S = 60

# The deadline grows with the bytes actually verified, so large (or chunk-only)

# payloads get headroom proportional to their size: pure BLAKE3 hashes at

# ~1.7 MB/s and zstd decodes at ~30 MB/s on an M4, so 1.5 s/MiB is ~2.5x margin.

DEADLINE_S_PER_MIB = 1.5

HEX64 = re.compile('[0-9a-f]{64}')

EMPTY_HASH = 'af1349b9f5f9a1a6a0404dea36dcc9499bcb25c9adc112b7cc9a93cae41f3262'

ZSTD_MAGIC = b'\x28\xb5\x2f\xfd'

BLOB_MODES = ('regular', 'executable', 'symlink')

RFC3339 = re.compile(r'(\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d)(?:\.(\d{1,9}))?(Z|[+-]\d\d:\d\d)')

class Unavailable(Exception):

    pass

class Invalid(Exception):

    pass

REQUIRED_COLUMNS = {

    'metadata': {'key', 'value'},

    'commits': {'hash', 'branch_name', 'parent_hash', 'merge_parent_hash', 'manifest_hash', 'author', 'message', 'timestamp'},

    'blobs': {'hash', 'content', 'size', 'codec'},

    'trees': {'hash', 'content'},

}

# Legacy per-entry tree rows. `oak maintenance compact` backfills every

# trees.content from them and then DROPS the table by design, so it is

# optional: when absent, every tree must carry content (a NULL is Invalid).

LEGACY_TREE_COLUMNS = {'tree_entries': {'tree_hash', 'name', 'kind', 'hash', 'mode'}}

CHUNK_COLUMNS = {

    'blob_chunks': {'blob_hash', 'chunk_hash', 'chunk_index', 'offset', 'size'},

    'chunks': {'hash', 'content', 'size'},

}

class Backend:

    """Hash and codec implementation actually used, recorded in every result."""

    def __init__(self, name=None):

        name = name or os.environ.get('OAK_BENCH_READER_BACKEND') or 'pure'

        if name not in ('pure', 'native'):

            raise Unavailable('unknown reader backend: ' + name)

        self.name = name

        if name == 'native':

            try:

                import blake3

                import zstandard

            except ImportError as exc:

                raise Unavailable('native reader backend requires blake3 and zstandard') from exc

            self._blake3, self._zstd = blake3, zstandard

            self.versions = {'blake3_version': importlib.metadata.version('blake3'),

                             'zstandard_version': importlib.metadata.version('zstandard')}

        else:

            self.versions = {'blake3_version': None, 'zstandard_version': None}

    def hexdigest(self, raw: bytes) -> str:

        if self.name == 'native':

            return self._blake3.blake3(raw).hexdigest()

        return pureblake3.hexdigest(raw)

    def decompress(self, content: bytes) -> bytes:

        if self.name == 'native':

            zstandard = self._zstd

            try:

                size = zstandard.frame_content_size(content)

                if zstandard.get_frame_parameters(content).window_size > MAX_OBJECT:

                    raise Unavailable('decompression window exceeds reader bound')

                if size not in (-1, zstandard.CONTENTSIZE_UNKNOWN) and not 0 <= size <= MAX_OBJECT:

                    raise Unavailable('decompressed object exceeds reader bound')

                return zstandard.ZstdDecompressor(max_window_size=MAX_OBJECT).decompress(

                    content, max_output_size=MAX_OBJECT, allow_extra_data=False)

            except zstandard.ZstdError as exc:

                raise Invalid('invalid compressed object') from exc

        try:

            return purezstd.decompress(content, max_output=MAX_OBJECT, max_window=MAX_OBJECT)

        except purezstd.ZstdBoundError as exc:

            raise Unavailable(str(exc).replace('bound', 'reader bound')) from exc

        except purezstd.ZstdError as exc:

            raise Invalid('invalid compressed object: ' + str(exc)) from exc

def _nonempty(path: Path) -> bool:

    try:

        return path.stat().st_size > 0

    except FileNotFoundError:

        return False

def codec_sha256() -> str:

    """Identity of the pure codec modules the reader verified with."""

    digest = hashlib.sha256()

    for module in (pureblake3, purezstd):

        digest.update(Path(module.__file__).read_bytes())

    return digest.hexdigest()

def normalize_scope(scope):

    """Validated repo-relative POSIX scope paths, or None for the whole tree."""

    if scope is None:

        return None

    result = []

    for item in scope:

        parts = item.split('/')

        if (not item or item.startswith('/') or any(part in ('', '.', '..', '.oak', '.git') for part in parts)

                or any(c in item for c in '\x00\n\r')):

            raise Unavailable('invalid reader scope path')

        result.append(item)

    return tuple(sorted(set(result)))

def in_scope(relative: str, scope) -> tuple[bool, bool]:

    """(visit, materialize) for a repo-relative path under the scope."""

    if scope is None:

        return True, True

    for item in scope:

        if relative == item or relative.startswith(item + '/'):

            return True, True

        if item.startswith(relative + '/'):

            return True, False

    return False, False

def entry_name_problem(name: str):

    """Oak's own entry-name rule (core tree.rs validate_entry_name), re-stated."""

    if not name:

        return 'empty filename'

    if name in ('.', '..'):

        return 'reserved name'

    if name in ('.oak', '.git'):

        return 'VCS metadata name'

    if '/' in name:

        return 'slash in filename'

    if any(ord(c) < 0x20 or 0x7f <= ord(c) <= 0x9f for c in name):

        return 'control character in filename'

    return None

def rfc3339_tail_candidates(timestamp: str):

    """Oak's verifier restores 1..999 ns lost to microsecond storage; mirror it."""

    match = RFC3339.fullmatch(timestamp)

    if not match:

        return []

    base, fraction, zone = match.groups()

    nanos = int((fraction or '').ljust(9, '0') or '0')

    if nanos % 1000:

        return []

    return [f'{base}.{nanos + tail:09d}{zone}' for tail in range(1, 1000)]

def commit_verifies(backend, commit_hash, fields):

    """fields = (branch, parent, merge_parent, manifest, author, message, timestamp)."""

    branch, parent, merge, manifest, author, message, timestamp = [v or '' for v in fields]

    def seven(ts):

        return '\n'.join((branch, parent, merge, manifest, author, message, ts)).encode()

    if backend.hexdigest(seven(timestamp)) == commit_hash:

        return 'v1'

    if not merge and backend.hexdigest('\n'.join((branch, parent, manifest, author, message, timestamp)).encode()) == commit_hash:

        return 'v1-legacy-6-field'

    for candidate in rfc3339_tail_candidates(timestamp):

        if backend.hexdigest(seven(candidate)) == commit_hash:

            return 'v1-microsecond-truncated'

    return None

def read_committed_tree(repo: Path, destination, scope=None, max_blob_bytes=None, *,

                        manifest=False, deep=False, complete=False, backend=None) -> dict:

    """Materialize (``destination``) and/or describe (``manifest``) Oak's committed HEAD.

    ``destination=None`` materializes nothing; the manifest is then the only

    output, which avoids filesystem aliasing (case folding, Unicode

    normalization) when the oracle must see exact committed names.

    """

    scope = normalize_scope(scope)

    backend = backend if isinstance(backend, Backend) else Backend(backend)

    repo = Path(repo)

    db_path = repo / '.oak' / 'oak.db'

    # The repository is subject-controlled state: a missing or linked database

    # is a measured failure, exactly as Git's ls-tree on a repo whose .git was

    # removed is (git_head_tree_unreadable). Never "unmeasured".

    if (repo.is_symlink() or (repo / '.oak').is_symlink() or db_path.is_symlink()

            or not db_path.is_file()):

        raise Invalid('committed state unreadable: local Oak database missing or linked')

    if destination is not None:

        destination = Path(destination)

        if destination.exists() or destination.is_symlink():

            raise Unavailable('reader destination must not exist')

    deadline = time.monotonic() + DEADLINE_S

    def database_snapshot():

        # A zero-byte -wal/-journal holds no frames (left by a closed reader

        # connection); only a non-empty one can change what the database means.

        if any(_nonempty(Path(str(db_path) + suffix)) for suffix in ('-wal', '-journal')):

            raise Unavailable('active SQLite journal; quiescent database required')

        with db_path.open('rb') as stream:

            content = stream.read(MAX_DATABASE + 1)

        if len(content) > MAX_DATABASE:

            raise Unavailable('database exceeds independent reader bound')

        return content

    # Snapshot only quiescent databases. immutable=1 is safe on this private copy,

    # never on a live WAL database; the original bytes are checked again below.

    snapshot = database_snapshot()

    database_sha256 = hashlib.sha256(snapshot).hexdigest()

    if destination is not None:

        destination.mkdir(parents=True)

    with tempfile.TemporaryDirectory(prefix='oakbench-reader-db-') as temp:

        snapshot_path = Path(temp) / 'snapshot.db'

        snapshot_path.write_bytes(snapshot)

        del snapshot

        reader = _SnapshotReader(snapshot_path, backend, deadline, scope, max_blob_bytes,

                                 destination, manifest, deep, complete)

        result = reader.run()

        if hashlib.sha256(database_snapshot()).hexdigest() != database_sha256:

            raise Unavailable('database changed during independent read')

    result.update({'database_sha256': database_sha256, 'reader_version': READER_VERSION,

                   'scope': list(scope) if scope is not None else None,

                   'reader_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),

                   'reader_backend': backend.name, 'codec_sha256': codec_sha256(), **backend.versions})

    return result

class _SnapshotReader:

    def __init__(self, path, backend, deadline, scope, max_blob_bytes, destination, manifest, deep, complete):

        self.path, self.backend, self.deadline = path, backend, deadline

        self.scope, self.max_blob_bytes = scope, max_blob_bytes

        self.destination, self.want_manifest = destination, manifest

        self.deep, self.complete = deep, complete

        self.total = 0

        self.entries_seen = 0

        self.nonregular, self.oversized, self.collisions = [], [], []

        self.entries = []

        self.symlinks = []

        self.chunked_blobs = 0

        self.chunk_crosschecks = 0

    def hashed(self, raw, expected):

        if not isinstance(expected, str) or not HEX64.fullmatch(expected):

            raise Invalid('invalid object hash')

        self.deadline += len(raw) / (1024 * 1024) * DEADLINE_S_PER_MIB

        if self.backend.hexdigest(raw) != expected:

            raise Invalid('object hash mismatch')

        return raw

    def decode(self, content, codec):

        if not isinstance(content, bytes):

            raise Invalid('stored object has invalid type')

        if len(content) > MAX_OBJECT:

            raise Unavailable('stored object exceeds reader bound')

        if codec == 0:

            return content

        if codec != 1:

            raise Unavailable('unsupported object codec')

        raw = self.backend.decompress(content)

        if len(raw) > MAX_OBJECT:

            raise Unavailable('decompressed object exceeds reader bound')

        return raw

    def run(self):

        with sqlite3.connect(self.path.as_uri() + '?mode=ro&immutable=1', uri=True, timeout=5) as db:

            self.db = db

            db.execute('PRAGMA query_only=ON')

            db.execute('PRAGMA trusted_schema=OFF')

            db.set_progress_handler(lambda: int(time.monotonic() > self.deadline), 1000)

            db.execute('BEGIN')

            try:

                return self._run()

            except sqlite3.OperationalError as exc:

                if 'interrupted' in str(exc):

                    raise Unavailable('reader deadline exceeded') from exc

                raise

            finally:

                db.commit()

    def _columns(self, table):

        return {row[1] for row in self.db.execute(f'PRAGMA table_info({table})')}

    def _run(self):

        db = self.db

        for table, columns in REQUIRED_COLUMNS.items():

            if not columns <= self._columns(table):

                raise Unavailable('unrecognized Oak storage schema: ' + table)

        self.has_chunks = all(columns <= self._columns(table) for table, columns in CHUNK_COLUMNS.items())

        self.has_tree_entries = bool(db.execute(

            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='tree_entries'").fetchone())

        if self.has_tree_entries and not LEGACY_TREE_COLUMNS['tree_entries'] <= self._columns('tree_entries'):

            raise Unavailable('unrecognized Oak storage schema: tree_entries')

        migrations = None

        if db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='schema_migrations'").fetchone():

            row = db.execute("SELECT max(version) FROM schema_migrations WHERE length(version)<=128").fetchone()

            migrations = row[0] if row else None

        if isinstance(migrations, str) and migrations > KNOWN_MAX_MIGRATION:

            raise Unavailable('unrecognized Oak schema migration: ' + migrations)

        def metadata(key):

            row = db.execute('SELECT value FROM metadata WHERE key=? AND length(value)<=4096', (key,)).fetchone()

            return row[0] if row else None

        if metadata('hash_format') not in (None, 'v1'):

            raise Unavailable('unsupported Oak hash format')

        head = metadata('head')

        branch = metadata('current_branch')

        if not head or not branch:

            # Nothing committed (or HEAD removed): subject state, measured failure.

            raise Invalid('no committed HEAD')

        commit = self._commit_row(head)

        if not commit:

            # The storage format is already pinned (schema columns and

            # migrations), so an Oak-written HEAD naming a missing commit row

            # is a dangling HEAD: an Oak failure, not unmeasured (QA R2-3).

            raise Invalid('committed HEAD names a missing commit row')

        if any(v is not None and (not isinstance(v, str) or len(v) > MAX_OBJECT) for v in commit):

            raise Invalid('committed HEAD object invalid')

        # Oak v1 stores the same RFC3339 timestamp text used in its hash preimage.

        # A preimage mismatch is an unrecognized format (fail closed), never a

        # published Oak correctness failure.

        if not HEX64.fullmatch(head):

            raise Invalid('invalid object hash')

        preimage = commit_verifies(self.backend, head, commit)

        if preimage is None:

            # Pinned format (another commit, tree or blob verifies): a HEAD

            # whose fields do not hash to its id is corruption, not drift.

            if self._format_recognized(exclude_commit=head):

                raise Invalid('committed HEAD commit hash mismatch')

            raise Unavailable('commit preimage format unrecognized: object hash mismatch')

        root_hash = commit[3]

        self._pin_blob_format()

        self.visit(root_hash, '', set())

        # Symlinks last: no regular write may pass through a created link.

        for relative, target in self.symlinks:

            os.symlink(target, self.destination / relative)

        history = self._history(head, commit) if self.deep else None

        result = {'head': head, 'branch': branch, 'root_tree': root_hash, 'commit_preimage': preimage,

                  'commit': {'parent': commit[1], 'merge_parent': commit[2], 'author': commit[4],

                             'message_sha256': hashlib.sha256((commit[5] or '').encode()).hexdigest(),

                             'timestamp': commit[6]},

                  'nonregular_paths': sorted(self.nonregular), 'oversized_paths': sorted(self.oversized),

                  'collision_paths': sorted(self.collisions), 'schema_migrations_max': migrations,

                  'chunked_blobs_reassembled': self.chunked_blobs,

                  'chunk_mappings_crosschecked': self.chunk_crosschecks}

        if history is not None:

            result['history'] = history

        if self.want_manifest:

            result['manifest'] = sorted(self.entries, key=lambda e: e['path'])

        return result

    def _commit_row(self, commit_hash):

        return self.db.execute(

            'SELECT branch_name,parent_hash,merge_parent_hash,manifest_hash,author,message,timestamp FROM commits '

            'WHERE hash=? AND length(CAST(branch_name AS BLOB))<=4096 AND coalesce(length(CAST(parent_hash AS BLOB)),0)<=64 '

            'AND coalesce(length(CAST(merge_parent_hash AS BLOB)),0)<=64 AND length(CAST(manifest_hash AS BLOB))=64 '

            'AND length(CAST(author AS BLOB))<=4096 AND coalesce(length(CAST(message AS BLOB)),0)<=1048576 '

            'AND length(CAST(timestamp AS BLOB))<=128', (commit_hash,)).fetchone()

    def _history(self, head, commit):

        """Walk first and merge parents; every present commit must hash-verify.

        A parent absent from the local store is a shallow boundary (reported),

        not corruption: Oak clones may be depth-limited.

        """

        seen, frontier, missing = {head}, [commit], []

        verified = 1

        while frontier:

            if time.monotonic() > self.deadline or verified > MAX_HISTORY:

                raise Unavailable('history walk bound exceeded')

            current = frontier.pop()

            for parent in (current[1], current[2]):

                if not parent or parent in seen:

                    continue

                seen.add(parent)

                if not HEX64.fullmatch(parent):

                    raise Invalid('invalid parent hash')

                row = self._commit_row(parent)

                if row is None:

                    missing.append(parent)

                    continue

                if commit_verifies(self.backend, parent, row) is None:

                    raise Invalid('ancestor commit hash mismatch: ' + parent)

                verified += 1

                frontier.append(row)

        return {'commits_verified': verified, 'missing_parents': sorted(missing)}

    def _pin_blob_format(self):

        # Blob-hash format pin: at least one of the smallest stored complete

        # blobs must verify under blake3(raw). If none does, the convention

        # changed (unmeasured); once pinned, in-scope blob mismatches are

        # corruption. (A repository whose every sampled blob is corrupt is

        # indistinguishable from a format change and fails closed.)

        candidates = self.db.execute('SELECT hash,content,codec FROM blobs WHERE content IS NOT NULL AND length(content)<=? '

                                     'ORDER BY size LIMIT 8', (MAX_OBJECT,)).fetchall()

        checked = 0

        for object_hash, content, codec in candidates:

            try:

                self.hashed(self.decode(content, codec), object_hash)

                return

            except Invalid:

                checked += 1

            except Unavailable:

                continue  # over a reader bound: says nothing about the format

        if checked and not (self._pin('commit') or self._pin('tree')):

            raise Unavailable('blob hash format unrecognized: no sampled blob verifies')

    # --- format pins -------------------------------------------------------

    # The storage format counts as recognized when the schema is pinned and at

    # least one stored object of ANY kind (commit, tree, blob) verifies under

    # the v1 rules, excluding the object under test. A mismatch on a

    # recognized format is corruption (Invalid); only a store in which nothing

    # verifies is indistinguishable from a format change (Unavailable).

    def _pin(self, kind, exclude=None):

        cache = self.__dict__.setdefault('_pins', {})

        key = (kind, exclude)

        if key in cache:

            return cache[key]

        ok = False

        if kind == 'commit':

            rows = self.db.execute('SELECT hash FROM commits WHERE hash IS NOT ? LIMIT 16', (exclude,)).fetchall()

            for (commit_hash,) in rows:

                row = self._commit_row(commit_hash)

                if row and isinstance(commit_hash, str) and commit_verifies(self.backend, commit_hash, row):

                    ok = True

                    break

        elif kind == 'tree':

            rows = self.db.execute('SELECT hash FROM trees WHERE hash IS NOT ? LIMIT 16', (exclude,)).fetchall()

            for (tree_hash,) in rows:

                try:

                    self.hashed(self._tree_bytes(tree_hash), tree_hash)

                    ok = True

                    break

                except (Invalid, Unavailable):

                    continue

        else:

            rows = self.db.execute('SELECT hash,content,codec FROM blobs WHERE content IS NOT NULL AND '

                                   'length(content)<=? ORDER BY size LIMIT 8', (MAX_OBJECT,)).fetchall()

            for object_hash, content, codec in rows:

                try:

                    self.hashed(self.decode(content, codec), object_hash)

                    ok = True

                    break

                except (Invalid, Unavailable):

                    continue

        cache[key] = ok

        return ok

    def _format_recognized(self, exclude_commit=None, exclude_tree=None):

        return (self._pin('commit', exclude_commit) or self._pin('tree', exclude_tree) or self._pin('blob'))

    def _chunks(self, object_hash):

        if not self.has_chunks:

            return None

        rows = self.db.execute('SELECT chunk_index,chunk_hash,offset,size FROM blob_chunks WHERE blob_hash=? '

                               'ORDER BY chunk_index LIMIT 100001', (object_hash,)).fetchall()

        return rows or None

    def _assemble(self, object_hash, rows, expected_size):

        out = bytearray()

        for position, (index, chunk_hash, offset, size) in enumerate(rows):

            if index != position or offset != len(out) or type(size) is not int or size < 0:

                raise Invalid('non-contiguous chunk mapping')

            if len(out) + size > MAX_OBJECT:

                raise Unavailable('chunked blob exceeds reader bound')

            row = self.db.execute('SELECT content,size FROM chunks WHERE hash=? AND length(content)<=?',

                                  (chunk_hash, MAX_OBJECT)).fetchone()

            if row is None:

                if self.db.execute('SELECT 1 FROM chunks WHERE hash=?', (chunk_hash,)).fetchone():

                    raise Unavailable('stored chunk exceeds reader bound')

                raise (Invalid if self.complete else Unavailable)('chunk content absent: ' + str(chunk_hash))

            content = row[0]

            if not isinstance(content, bytes):

                raise Invalid('stored chunk has invalid type')

            # Oak's chunk_decode: a zstd-looking chunk is plaintext only if the

            # decoded bytes hash to the chunk id; otherwise the raw bytes are.

            data = content

            if content[:4] == ZSTD_MAGIC and self.backend.hexdigest(content) != chunk_hash:

                data = self.backend.decompress(content)

            if len(data) != size:

                raise Invalid('chunk size mismatch')

            self.hashed(data, chunk_hash)

            out += data

        if expected_size is not None and len(out) != expected_size:

            raise Invalid('chunked blob size mismatch')

        return self.hashed(bytes(out), object_hash)

    def blob(self, object_hash):

        row = self.db.execute('SELECT content,size,codec FROM blobs WHERE hash=? AND length(content)<=?',

                              (object_hash, MAX_OBJECT)).fetchone()

        if row is None:

            if self.db.execute('SELECT 1 FROM blobs WHERE hash=?', (object_hash,)).fetchone():

                raise Unavailable('stored blob exceeds reader bound')

            if object_hash == EMPTY_HASH:

                return b''  # Oak implies the empty blob's bytes from its hash.

            rows = self._chunks(object_hash)

            if rows is None:

                raise (Invalid if self.complete else Unavailable)(

                    'complete blob unavailable and no chunk mapping: ' + object_hash)

            self.chunked_blobs += 1

            return self._assemble(object_hash, rows, None)

        content, size, codec = row

        if type(size) is not int or size < 0:

            raise Invalid('invalid stored blob size')

        if size > MAX_OBJECT:

            raise Unavailable('blob size exceeds reader bound')

        raw = self.decode(content, codec)

        if len(raw) != size:

            raise Invalid('blob size mismatch')

        raw = self.hashed(raw, object_hash)

        if self.deep:

            rows = self._chunks(object_hash)

            if rows is not None:

                self._assemble(object_hash, rows, size)

                self.chunk_crosschecks += 1

        return raw

    def _tree_bytes(self, tree_hash):

        db = self.db

        row = db.execute('SELECT content FROM trees WHERE hash=? AND (content IS NULL OR length(content)<=?)',

                         (tree_hash, MAX_OBJECT)).fetchone()

        if row is None:

            if db.execute('SELECT 1 FROM trees WHERE hash=?', (tree_hash,)).fetchone():

                raise Unavailable('committed tree exceeds reader bound')

            raise Invalid('committed tree missing')

        if row[0] is not None:

            return self.decode(row[0], 1)

        if not self.has_tree_entries:

            # Compacted store: content is the only representation of a tree.

            raise Invalid('committed tree content missing in compacted store')

        if db.execute('SELECT 1 FROM tree_entries WHERE tree_hash=? AND (length(CAST(name AS BLOB))>4096 OR '

                      'length(CAST(hash AS BLOB))!=64 OR length(CAST(kind AS BLOB))>4 OR length(CAST(mode AS BLOB))>10) '

                      'LIMIT 1', (tree_hash,)).fetchone():

            raise Unavailable('stored tree entry exceeds reader bound')

        stored = db.execute('SELECT kind,name,hash,mode FROM tree_entries WHERE tree_hash=? ORDER BY name LIMIT ?',

                            (tree_hash, MAX_ENTRIES + 1)).fetchall()

        return '\n'.join('\t'.join((kind, name, obj, 'tree' if kind == 'tree' else mode))

                         for kind, name, obj, mode in stored).encode()

    def _record(self, relative, kind, **fields):

        self.entries.append({'path': relative, 'kind': kind, **fields})

    def visit(self, tree_hash, relative, ancestors):

        if tree_hash in ancestors:

            raise Invalid('tree cycle')

        if len(ancestors) >= 64 or time.monotonic() > self.deadline:

            raise Unavailable('tree depth or time bound exceeded')

        if not isinstance(tree_hash, str) or not HEX64.fullmatch(tree_hash):

            raise Invalid('invalid object hash')

        if tree_hash == EMPTY_HASH:

            return

        raw = self._tree_bytes(tree_hash)

        if not ancestors:

            try:

                self.hashed(raw, tree_hash)

            except Invalid as exc:

                if self._format_recognized(exclude_tree=tree_hash):

                    raise Invalid('root tree hash mismatch: ' + str(exc)) from exc

                raise Unavailable('root tree serialization unrecognized: ' + str(exc)) from exc

        else:

            self.hashed(raw, tree_hash)

        try:

            lines = raw.decode('utf-8').split('\n') if raw else []

        except UnicodeError as exc:

            raise Invalid('invalid tree encoding') from exc

        previous = None

        for line in lines:

            parts = line.split('\t')

            if len(parts) != 4:

                raise Invalid('invalid tree entry')

            kind, name, object_hash, mode = parts

            if entry_name_problem(name) or (previous is not None and name <= previous):

                raise Invalid('unsafe or unordered tree path: ' + (entry_name_problem(name) or 'order'))

            previous = name

            if not HEX64.fullmatch(object_hash):

                raise Invalid('invalid object hash in tree entry')

            path = name if not relative else relative + '/' + name

            visit_entry, _ = in_scope(path, self.scope)

            if not visit_entry:

                continue

            if not ((kind == 'tree' and mode == 'tree') or (kind == 'blob' and mode in BLOB_MODES)):

                if self.scope is not None:

                    self.nonregular.append(path)

                    continue

                raise Unavailable('unsupported tree entry kind or mode: %s/%s' % (kind, mode))

            self.entries_seen += 1

            if self.entries_seen > MAX_ENTRIES:

                raise Unavailable('tree entry bound exceeded')

            target = self.destination / path if self.destination is not None else None

            if target is not None and os.path.lexists(target):

                # Git fails the path set for colliding committed paths;

                # report it so the oracle fails the same way.

                self.collisions.append(path)

                continue

            if kind == 'tree':

                if target is not None:

                    target.mkdir()

                if self.want_manifest:

                    self._record(path, 'dir')

                self.visit(object_hash, path, ancestors | {tree_hash})

                continue

            if mode == 'symlink' and self.scope is not None:

                # Git's scoped check fails any non-100644 entry; report

                # it so the oracle fails the payload the same way.

                self.nonregular.append(path)

                continue

            if self.max_blob_bytes is not None and mode != 'symlink':

                stored = self.db.execute('SELECT size FROM blobs WHERE hash=?', (object_hash,)).fetchone()

                if stored and type(stored[0]) is int and stored[0] > self.max_blob_bytes:

                    # The task's own size limit fails first, as in Git,

                    # before any reader resource bound applies.

                    self.oversized.append(path)

                    continue

            content = self.blob(object_hash)

            self.total += len(content)

            if self.total > MAX_TOTAL:

                raise Unavailable('total payload bound exceeded')

            if mode == 'symlink':

                if b'\x00' in content or not content:

                    raise Invalid('invalid symlink target')

                if self.want_manifest:

                    self._record(path, 'symlink', target_hex=content.hex(), blob=object_hash)

                if target is not None:

                    self.symlinks.append((path, content))

                continue

            if self.want_manifest:

                self._record(path, 'file', executable=mode == 'executable', size=len(content),

                             sha256=hashlib.sha256(content).hexdigest(), blob=object_hash)

            if target is not None:

                target.write_bytes(content)

                target.chmod(0o755 if mode == 'executable' else 0o644)

def committed_payloads(repo: Path, paths, *, verify_all: bool = False) -> tuple[dict, dict]:

    """{path: bytes} of committed regular files at HEAD for harness oracles.

    Paths absent from HEAD (or not regular files) are omitted. ``verify_all``

    first runs an unscoped deep read (every object and ancestor re-hashed,

    nothing materialized): Oak's missing ``fsck`` equivalent.

    Raises Invalid / Unavailable exactly like ``read_committed_tree``.

    """

    repo = Path(repo)

    meta = {}

    if verify_all:

        full = read_committed_tree(repo, None, deep=True)

        meta.update(head=full['head'], history=full.get('history'), verified_scope='all')

    scope = normalize_scope(list(paths))

    payloads = {}

    if scope:

        with tempfile.TemporaryDirectory(prefix='oakbench-committed-') as temporary:

            destination = Path(temporary) / 'tree'

            result = read_committed_tree(repo, destination, scope)

            meta.setdefault('head', result['head'])

            meta['reader_version'] = result['reader_version']

            meta['reader_backend'] = result['reader_backend']

            for relative in scope:

                target = destination / relative

                if target.is_file() and not target.is_symlink():

                    payloads[relative] = target.read_bytes()

    return payloads, meta

def main():

    try:

        args = sys.argv[1:]

        scope = None

        max_blob_bytes = None

        options = {'manifest': False, 'deep': False, 'complete': False}

        usage = ('usage: oak_sqlite_reader.py REPOSITORY NEW_DESTINATION|- [--manifest] [--deep] [--complete] '

                 '[--max-blob-bytes N] [--scope PATH...]')

        if len(args) < 2:

            raise Unavailable(usage)

        rest = args[2:]

        while rest[:1] in (['--manifest'], ['--deep'], ['--complete']):

            options[rest[0][2:]] = True

            rest = rest[1:]

        if rest[:1] == ['--max-blob-bytes']:

            if len(rest) < 2 or not rest[1].isdigit():

                raise Unavailable(usage)

            max_blob_bytes = int(rest[1])

            rest = rest[2:]

        if rest[:1] == ['--scope']:

            scope = rest[1:]

            if not scope:

                raise Unavailable('empty reader scope')

        elif rest:

            raise Unavailable(usage)

        destination = None if args[1] == '-' else Path(args[1])

        result = read_committed_tree(Path(args[0]), destination, scope, max_blob_bytes, **options)

        code = 0

    except Unavailable as exc:

        result, code = {'unavailable': str(exc)}, 77

    except Invalid as exc:

        result, code = {'corrupt': True, 'reason': str(exc)}, 1

    except (OSError, ValueError, TypeError, sqlite3.Error) as exc:

        result, code = {'unavailable': 'reader instrument error: ' + str(exc)}, 77

    print(json.dumps(result, sort_keys=True))

    return code

if __name__ == '__main__':

    raise SystemExit(main())