#!/usr/bin/env python3
"""Bounded decimal-wire permutation queries with two explicit provider models.

Seeded HMAC bits are deterministic and carry no statistical mixing/security
claim. A local POSIX append-only store serializes cooperating workers and
persists each new OS-entropy bit; actual independent fairness is not verified.
Test tapes use their own visibly distinct plan kind. No damaged file is repaired.
"""
import argparse
import fcntl
import hashlib
import hmac
import json
import os
import re
import secrets
import time
from contextlib import contextmanager
from pathlib import Path

from queryable_permutation_reference import integer, pair_id

MAX_BYTES = 8_000_000
MAX_ROWS = 100_000
MAX_WORK = 4_000_000
KINDS = ('seeded_hmac_sha256', 'append_os_entropy', 'append_test_bits')


def encoded(record):
    return json.dumps(record, sort_keys=True, separators=(',', ':'), ensure_ascii=True).encode()


def digest(record):
    return hashlib.sha256(encoded(record)).hexdigest()


def decimal(value, upper, name):
    if not isinstance(value, str) or len(value) > 20 or not re.fullmatch(r'0|[1-9][0-9]*', value):
        raise ValueError('Require canonical decimal string for ' + name)
    result = int(value)
    if not 0 <= result <= upper:
        raise ValueError('Out of range ' + name)
    return result


def validate_plan(plan):
    fields = {'version', 'plan_id', 'epoch', 'dataset_revision', 'dimension', 'sweeps', 'target_size', 'provider_kind', 'seed_commitment'}
    if not isinstance(plan, dict) or set(plan) != fields or type(plan['version']) is not int or plan['version'] != 1:
        raise ValueError('Invalid plan schema')
    for name in ('plan_id', 'epoch', 'dataset_revision'):
        if not isinstance(plan[name], str) or not re.fullmatch(r'[A-Za-z0-9_.-]{1,128}', plan[name]):
            raise ValueError('Invalid plan namespace')
    d = integer(plan['dimension'], 1, 63, 'dimension')
    integer(plan['sweeps'], 0, 64, 'sweeps')
    m = decimal(plan['target_size'], 1 << d, 'target size')
    if m == 0 or plan['provider_kind'] not in KINDS:
        raise ValueError('Invalid target/provider')
    commitment = plan['seed_commitment']
    if plan['provider_kind'] == KINDS[0]:
        if not isinstance(commitment, str) or not re.fullmatch(r'[0-9a-f]{64}', commitment):
            raise ValueError('Require seed commitment')
    elif commitment is not None:
        raise ValueError('Append store does not use a seed')
    return d, m


def key_record(key):
    return dict(sweep=key[0], coordinate=key[1], pair=str(key[2]))


class SeededBits:
    def __init__(self, plan, seed):
        validate_plan(plan)
        if plan['provider_kind'] != KINDS[0] or not isinstance(seed, bytes) or len(seed) != 32:
            raise ValueError('Require seeded plan and 32-byte seed')
        if hashlib.sha256(seed).hexdigest() != plan['seed_commitment']:
            raise ValueError('Seed differs from immutable plan')
        self.seed = seed
        self.plan_hash = digest(plan)
        self.lookups = 0

    def get(self, key):
        self.lookups += 1
        message = encoded(dict(plan_sha256=self.plan_hash, switch=key_record(key)))
        return hmac.new(self.seed, message, hashlib.sha256).digest()[0] >> 7

    def summary(self):
        return dict(kind=KINDS[0], retained_switch_rows=0, retained_store_bytes=0,
                    lookups=self.lookups, randomness_status='Deterministic 256-bit seeded mapping; source independent-bit law not inherited',
                    security_evaluation='not_run')


class AppendBits:
    def __init__(self, stream, plan, source, row_limit, byte_limit, created):
        self.stream, self.plan, self.source = stream, json.loads(encoded(plan)), source
        self.plan_hash = digest(self.plan)
        self.row_limit, self.byte_limit = row_limit, byte_limit
        self.bits, self.lookups, self.new_rows = {}, 0, 0
        header = dict(type='permutation_switch_store_v1', plan=plan, plan_sha256=digest(plan))
        stream.seek(0)
        file_size = os.fstat(stream.fileno()).st_size
        if file_size > byte_limit:
            raise ValueError('Existing store exceeds byte cap; left unchanged')
        # Reading the cap itself made a tiny store reserve an 8 MB buffer.
        # The lock fixes the cooperating-writer view; still refuse growth.
        raw = stream.read(file_size + 1)
        if len(raw) > file_size:
            raise ValueError('Store changed outside the cooperating lock; left unchanged')
        if created:
            if raw:
                raise ValueError('Fresh store unexpectedly contains bytes')
            self.append(header)
            raw = encoded(header) + b'\n'
        if not raw or not raw.endswith(b'\n'):
            raise ValueError('Empty or incomplete store; left unchanged')
        lines = raw.splitlines()
        if len(lines) > row_limit + 1 or lines[0] != encoded(header):
            raise ValueError('Store plan/header/cap mismatch; left unchanged')
        self.chain = digest(header)
        d, _ = validate_plan(plan)
        for line in lines[1:]:
            row = json.loads(line)
            if not isinstance(row, dict) or set(row) != {'switch', 'bit', 'previous_sha256'} or line != encoded(row):
                raise ValueError('Invalid/noncanonical store row; left unchanged')
            switch = row['switch']
            if not isinstance(switch, dict) or set(switch) != {'sweep', 'coordinate', 'pair'}:
                raise ValueError('Invalid store switch; left unchanged')
            key = (integer(switch['sweep'], 0, plan['sweeps'] - 1, 'sweep'),
                   integer(switch['coordinate'], 0, d - 1, 'coordinate'),
                   decimal(switch['pair'], (1 << d) // 2 - 1, 'pair'))
            bit = integer(row['bit'], 0, 1, 'bit')
            if key in self.bits or row['previous_sha256'] != self.chain:
                raise ValueError('Duplicate or changed store row; left unchanged')
            self.bits[key] = bit
            self.chain = digest(row)
        self.rows_at_open = len(self.bits)

    def append(self, record):
        body = encoded(record) + b'\n'
        self.stream.seek(0, os.SEEK_END)
        if self.stream.tell() + len(body) > self.byte_limit:
            raise OverflowError('Persistent store byte cap exhausted')
        self.stream.write(body)
        self.stream.flush()
        os.fsync(self.stream.fileno())

    def get(self, key):
        self.lookups += 1
        if key in self.bits:
            return self.bits[key]
        if len(self.bits) >= self.row_limit:
            raise OverflowError('Persistent store row cap exhausted')
        try:
            value = self.source()
        except StopIteration:
            raise LookupError('Explicit test tape exhausted') from None
        bit = integer(value, 0, 1, 'new provider bit')
        row = dict(switch=key_record(key), bit=bit, previous_sha256=self.chain)
        self.append(row)
        self.chain = digest(row)
        self.bits[key] = bit
        self.new_rows += 1
        return bit

    def summary(self):
        return dict(kind=self.plan['provider_kind'], retained_switch_rows=len(self.bits),
                    retained_store_bytes=os.fstat(self.stream.fileno()).st_size,
                    rows_at_open=self.rows_at_open, new_rows=self.new_rows, lookups=self.lookups,
                    randomness_status=('OS entropy calls; actual independent fairness unverified' if self.plan['provider_kind'] == KINDS[1]
                                       else 'Explicit deterministic test tape; no random law established'),
                    consistency_scope='Cooperating processes on a local POSIX file with flock; no network/distributed/power-loss guarantee',
                    integrity_scope='Canonical rows and hash chaining detect malformed/changed interior records; unanchored tail removal is not detected')


@contextmanager
def store(path, plan, test_source=None, row_limit=MAX_ROWS, byte_limit=MAX_BYTES, lock_timeout=1.0):
    validate_plan(plan)
    integer(row_limit, 1, MAX_ROWS, 'store rows')
    integer(byte_limit, 1024, MAX_BYTES, 'store bytes')
    if plan['provider_kind'] not in KINDS[1:] or type(lock_timeout) not in (int, float) or not 0 <= lock_timeout <= 5:
        raise ValueError('Invalid local store options')
    if plan['provider_kind'] == KINDS[2]:
        if not callable(test_source):
            raise ValueError('Test provider needs an explicit test tape')
        source = test_source
    else:
        if test_source is not None:
            raise ValueError('OS provider cannot be replaced with a test tape')
        source = lambda: secrets.randbits(1)
    try:
        descriptor = os.open(path, os.O_RDWR | os.O_CREAT | os.O_EXCL, 0o600)
        created = True
    except FileExistsError:
        descriptor = os.open(path, os.O_RDWR)
        created = False
    with os.fdopen(descriptor, 'r+b') as stream:
        deadline = time.monotonic() + lock_timeout
        while True:
            try:
                fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                break
            except BlockingIOError:
                if time.monotonic() >= deadline:
                    raise TimeoutError('Cooperating store writer busy; no query served') from None
                time.sleep(0.01)
        try:
            yield AppendBits(stream, plan, source, row_limit, byte_limit, created)
        finally:
            fcntl.flock(stream.fileno(), fcntl.LOCK_UN)


def queries(plan, requests, provider, work_budget=MAX_WORK):
    d, m = validate_plan(plan)
    integer(work_budget, 1, MAX_WORK, 'work budget')
    if not isinstance(requests, list) or len(requests) > 256:
        raise ValueError('At most 256 queries per locked batch')
    for request in requests:
        if not isinstance(request, dict) or set(request) != {'index', 'direction'} or request['direction'] not in ('forward', 'inverse'):
            raise ValueError('Invalid query schema')
        decimal(request['index'], m - 1, 'query index')
    if getattr(provider, 'plan_hash', digest(getattr(provider, 'plan', {}))) != digest(plan):
        raise ValueError('Provider plan mismatch')
    work, results = 0, []
    for request in requests:
        position, hops, before = int(request['index']), 0, work
        direction = request['direction']
        try:
            while True:
                hops += 1
                layers = ((s, c) for s in range(plan['sweeps']) for c in range(d)) if direction == 'forward' else (
                    (s, c) for s in range(plan['sweeps'] - 1, -1, -1) for c in range(d - 1, -1, -1))
                for sweep, coordinate in layers:
                    if work >= work_budget:
                        raise OverflowError('Switch lookup budget exhausted')
                    work += 1
                    bit_position = d - 1 - coordinate
                    key = (sweep, coordinate, pair_id(position, bit_position))
                    if provider.get(key):
                        position ^= 1 << bit_position
                if position < m:
                    break
                if hops > (1 << d) - m:
                    raise ArithmeticError('Cycle restriction bound violated')
        except (OverflowError, LookupError) as exc:
            results.append(dict(query=request, status='unknown', output=None, reason=str(exc), switch_lookups=work - before))
        else:
            results.append(dict(query=request, status='complete', output=str(position), cycle_hops=hops, switch_lookups=work - before))
    return dict(status='complete' if all(r['status'] == 'complete' for r in results) else 'partial',
                plan_sha256=digest(plan), dimension=d, target_size=str(m), outputs=results, work_units=work,
                provider=provider.summary(), source_proof_verification='not_run', program_formal_verification='not_run',
                source_mixing_guarantee='not_established', numerical_universal_sweep_threshold='not_selected',
                wire_scope='Canonical decimal strings parsed by Python exact integers; no JavaScript implementation tested',
                limits=dict(queries=256, switch_lookups=work_budget, store_rows=MAX_ROWS, store_bytes=MAX_BYTES, lock_wait_seconds_max=5))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('input', type=Path)
    parser.add_argument('--store', type=Path)
    parser.add_argument('--seed-hex')
    parser.add_argument('--work-budget', type=int, default=MAX_WORK)
    parser.add_argument('--output', type=Path)
    args = parser.parse_args()
    record = json.loads(args.input.read_text())
    if not isinstance(record, dict) or set(record) != {'plan', 'queries'}:
        raise ValueError('Require plan and queries')
    plan = record['plan']
    validate_plan(plan)
    if plan['provider_kind'] == KINDS[0]:
        if args.store or not args.seed_hex or not re.fullmatch(r'[0-9a-f]{64}', args.seed_hex):
            raise ValueError('Seeded mode requires exactly 64 lowercase seed hex characters and no store')
        result = queries(plan, record['queries'], SeededBits(plan, bytes.fromhex(args.seed_hex)), args.work_budget)
    elif plan['provider_kind'] == KINDS[1]:
        if not args.store or args.seed_hex:
            raise ValueError('OS entropy mode requires store and no seed')
        with store(args.store, plan) as provider:
            result = queries(plan, record['queries'], provider, args.work_budget)
    else:
        raise ValueError('Test tapes are available to explicit validation code only')
    body = json.dumps(result, indent=2, allow_nan=False) + '\n'
    if args.output:
        with args.output.open('x') as stream:
            stream.write(body)
    else:
        print(body, end='')


if __name__ == '__main__':
    main()