"""Optional independent decoder for local Oak SQLite v1 committed trees.

No Oak code or executable is imported or invoked. Optional blake3/zstandard

packages belong to a pinned reader environment, not the base harness. Unknown

formats and missing instruments are unavailable evidence; corrupt objects fail.

Evidence contract (2026-09-27.1):

- 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 must exist, and the HEAD commit

  preimage plus root tree serialization must verify. A mismatch there is an

  unrecognized format (Unavailable); once the root verifies, the same

  serialization is proven, so deeper hash mismatches are Invalid.

- In-scope non-regular entries (symlinks, unknown kinds) are reported in

  ``nonregular_paths`` instead of being refused, so the oracle can fail them

  exactly as Git's mode check does.

"""

from __future__ import annotations

import hashlib

import importlib.metadata

import json

from pathlib import Path

import re

import sqlite3

import sys

import time

import tempfile

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

# 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

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'},

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

}

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 read_committed_tree(repo: Path, destination: Path, scope=None, max_blob_bytes=None) -> dict:

    scope = normalize_scope(scope)

    try:

        import blake3

        import zstandard

    except ImportError as exc:

        raise Unavailable('reader dependencies unavailable: blake3 and zstandard required') from exc

    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.exists() or destination.is_symlink():

        raise Unavailable('reader destination must not exist')

    deadline = time.monotonic() + 60

    destination.mkdir(parents=True)

    def hashed(raw, expected):

        if not isinstance(expected, str) or not re.fullmatch('[0-9a-f]{64}', expected):

            raise Invalid('invalid object hash')

        if blake3.blake3(raw).hexdigest() != expected:

            raise Invalid('object hash mismatch')

        return raw

    def decode(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')

        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')

            raw = 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

        if len(raw) > MAX_OBJECT:

            raise Unavailable('decompressed object exceeds reader bound')

        return raw

    def database_snapshot():

        if any(Path(str(db_path) + suffix).exists() 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()

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

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

        snapshot_path.write_bytes(snapshot)

        del snapshot

        return _read_snapshot(snapshot_path, database_sha256, database_snapshot,

                              destination, deadline, hashed, decode, blake3, scope, max_blob_bytes)

def _read_snapshot(snapshot_path, database_sha256, database_snapshot, destination,

                   deadline, hashed, decode, blake3, scope=None, max_blob_bytes=None):

    total = 0

    entries_seen = 0

    nonregular = []

    oversized = []

    collisions = []

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

        db.execute('PRAGMA query_only=ON')

        db.execute('PRAGMA trusted_schema=OFF')

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

        db.execute('BEGIN')

        for table, columns in REQUIRED_COLUMNS.items():

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

            if not columns <= present:

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

        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 = 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', (head,)).fetchone()

        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.

        try:

            hashed('\n'.join(v or '' for v in commit).encode(), head)

        except Invalid as exc:

            raise Unavailable('commit preimage format unrecognized: ' + str(exc)) from exc

        root_hash = commit[3]

        # 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 = db.execute('SELECT hash,content,codec FROM blobs WHERE content IS NOT NULL AND length(content)<=? '

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

        if candidates:

            pinned = False

            for object_hash, content, codec in candidates:

                try:

                    hashed(decode(content, codec), object_hash)

                    pinned = True

                    break

                except (Invalid, Unavailable):

                    continue

            if not pinned:

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

        def blob(object_hash):

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

            if row is None:

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

                    raise Unavailable('stored blob exceeds reader bound')

                # Chunk-only hydration is not yet qualified by this reader.

                raise Unavailable('complete blob unavailable; chunk-only storage unsupported')

            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 = decode(content, codec)

            if len(raw) != size:

                raise Invalid('blob size mismatch')

            return hashed(raw, object_hash)

        def visit(tree_hash, relative, ancestors):

            nonlocal total, entries_seen

            if tree_hash in ancestors:

                raise Invalid('tree cycle')

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

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

            if tree_hash == blake3.blake3(b'').hexdigest():

                return

            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 None:

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

                raw = '\n'.join('\t'.join((kind, name, obj, 'tree' if kind == 'tree' else mode)) for kind, name, obj, mode in stored).encode()

            else:

                raw = decode(row[0], 1)

            if not ancestors:

                try:

                    hashed(raw, tree_hash)

                except Invalid as exc:

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

            else:

                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 (not name or name in ('.', '..', '.oak', '.git') or any(c in name for c in '/\\\x00\n\r')

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

                    raise Invalid('unsafe or unordered tree path')

                previous = name

                path = relative / name

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

                if not visit_entry:

                    continue

                entries_seen += 1

                if entries_seen > MAX_ENTRIES:

                    raise Unavailable('tree entry bound exceeded')

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

                    # Git fails the path set for colliding committed paths;

                    # report it so the oracle fails the same way.

                    collisions.append(path.as_posix())

                    continue

                if kind == 'tree' and mode == 'tree':

                    (destination / path).mkdir()

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

                elif kind == 'blob' and mode in ('regular', 'executable'):

                    if max_blob_bytes is not None:

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

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

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

                            # before any reader resource bound applies.

                            oversized.append(path.as_posix())

                            continue

                    content = blob(object_hash)

                    total += len(content)

                    if total > MAX_TOTAL:

                        raise Unavailable('total payload bound exceeded')

                    target = destination / path

                    target.write_bytes(content)

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

                elif scope is not None:

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

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

                    nonregular.append(path.as_posix())

                else:

                    raise Unavailable('unsupported tree entry kind or mode')

        visit(root_hash, Path(), set())

        db.commit()

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

            raise Unavailable('database changed during independent read')

    return {'head': head, 'database_sha256': database_sha256, 'reader_version': READER_VERSION,

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

            'oversized_paths': sorted(oversized), 'collision_paths': sorted(collisions),

            'schema_migrations_max': migrations,

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

            'blake3_version': importlib.metadata.version('blake3'),

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

def main():

    try:

        args = sys.argv[1:]

        scope = None

        max_blob_bytes = None

        usage = 'usage: oak_sqlite_reader.py REPOSITORY NEW_DESTINATION [--max-blob-bytes N] [--scope PATH...]'

        if len(args) < 2:

            raise Unavailable(usage)

        rest = args[2:]

        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)

        result = read_committed_tree(Path(args[0]), Path(args[1]), scope, max_blob_bytes)

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