"""Owned files and bounded numeric parsing for the fixed Lab22 adapter.

Hash maps establish identity relative to a supplied reference, not authenticity.
No input names select code, array schemas, arbitrary imports or subprocesses.
"""
import contextlib,fcntl,hashlib,io,json,math,os,re,stat,struct,time,uuid,zipfile
from datetime import datetime,timezone
from pathlib import Path

SCHEMA='llm-microscope/1'
RAW_CAP=50_000_000
PAYLOAD_CAP=40_000_000
MEMBER_CAP=20_000_000
HEADER_CAP=10_000

def stamp():return datetime.now(timezone.utc).isoformat()
def digest(data):return hashlib.sha256(data).hexdigest()
def encode(value):return (json.dumps(value,indent=2,sort_keys=True,allow_nan=False)+'\n').encode()
def unique(items):
    result={}
    for k,v in items:
        if k in result:raise ValueError('Duplicate JSON key: '+k)
        result[k]=v
    return result

def finite_float(text):
    value=float(text)
    if not math.isfinite(value):raise ValueError('Nonfinite/overflow JSON number')
    return value

def decode(data):
    return json.loads(data,object_pairs_hook=unique,parse_float=finite_float,
        parse_constant=lambda x:(_ for _ in ()).throw(ValueError('Nonfinite JSON: '+x)))

def safe(path):
    p=Path(path).absolute()
    if '..' in p.parts or any(x.is_symlink() for x in (p,*p.parents)):
        raise ValueError('Traversal or symlink path')
    return p

def relative(root,name):
    if not isinstance(name,str) or not name or Path(name).is_absolute() or '\\' in name or any(x in ('..','.','') for x in name.split('/')):
        raise ValueError('Require a relative inventoried path')
    p=safe(safe(root)/name)
    if not p.is_relative_to(safe(root)):raise ValueError('Path escapes selected root')
    return p

def bounded(path,limit=RAW_CAP):
    p=safe(path)
    if not p.is_file() or not stat.S_ISREG(p.stat().st_mode) or p.stat().st_size>limit:
        raise ValueError('Missing, nonregular or oversized file: '+p.name)
    with p.open('rb') as f:data=f.read(limit+1)
    if len(data)>limit:raise ValueError('Read cap exceeded')
    return data

def read_json(path,limit=2_000_000):
    return decode(bounded(path,limit))

def inventory(root):
    root=safe(root);result={}
    for p in sorted(root.rglob('*')):
        safe(p)
        try:record=p.stat()
        except FileNotFoundError:
            if re.search(r'\.writing-[0-9a-f]{32}$',p.name):continue
            raise
        if stat.S_ISDIR(record.st_mode):continue
        if not stat.S_ISREG(record.st_mode):raise ValueError('Nonregular evidence')
        result[p.relative_to(root).as_posix()]=record.st_size
    return result

def size(root):return sum(inventory(root).values())
def hashes(root,exclude=()):return {n:digest(bounded(relative(root,n))) for n in inventory(root) if n not in exclude}
def check_hashes(root,expected,exclude=()):
    if not isinstance(expected,dict) or any(not isinstance(v,str) or len(v)!=64 for v in expected.values()):raise ValueError('Invalid hash map')
    if hashes(root,exclude)!=expected:raise ValueError('Evidence inventory/hash mismatch')

def separate(out,*roots):
    out=safe(out)
    for root in roots:
        root=safe(root)
        if out.is_relative_to(root) or root.is_relative_to(out):raise ValueError('Source/output overlap')
    return out

class Store:
    def __init__(self,root,cap=RAW_CAP,terminal=False):self.root=safe(root);self.cap=cap;self.terminal=terminal
    def admission(self,extra):
        ceiling=self.cap if self.terminal else self.cap-200_000
        if size(self.root)+extra>ceiling:raise ValueError('Storage cap/terminal reserve exceeded')
    @contextlib.contextmanager
    def write_lock(self,deadline=None):
        # Lock the existing directory inode: parent logs and child evidence
        # share one admission/write transaction without another evidence file.
        fd=os.open(self.root,os.O_RDONLY)
        try:
            until=min(time.monotonic()+.1,deadline) if deadline is not None else time.monotonic()+.1
            while True:
                if time.monotonic()>=until:raise TimeoutError('Bounded storage admission lock')
                try:fcntl.flock(fd,fcntl.LOCK_EX|fcntl.LOCK_NB);break
                except BlockingIOError:
                    if time.monotonic()>=until:raise TimeoutError('Bounded storage admission lock')
                    time.sleep(min(.005,max(0,until-time.monotonic())))
            yield
        finally:os.close(fd)
    def put(self,name,data,replace=False):
        p=relative(self.root,name)
        with self.write_lock():
            if p.exists() and not replace:raise ValueError('Existing evidence file')
            self.admission(len(data))
            p.parent.mkdir(parents=True,exist_ok=True)
            temp=p.with_name(p.name+'.writing-'+uuid.uuid4().hex)
            with temp.open('xb') as f:f.write(data);f.flush();os.fsync(f.fileno())
            os.replace(temp,p)
    def json(self,name,value,replace=False):self.put(name,encode(value),replace)
    def append(self,stream,data,deadline=None):
        if safe(stream.name).parent!=self.root:raise ValueError('Log must be in owned root')
        with self.write_lock(deadline):
            self.admission(len(data))
            stream.write(data);stream.flush();os.fsync(stream.fileno())
    def log(self,event,name='attempts.jsonl'):
        data=(json.dumps(event,allow_nan=False)+'\n').encode()
        with relative(self.root,name).open('ab') as stream:self.append(stream,data)
    def arrays(self,name,values):
        import numpy as np
        stream=io.BytesIO();np.savez_compressed(stream,**values);self.put(name,stream.getvalue(),replace=True)

def events(path):
    return [decode(line) for line in bounded(path,1_000_000).splitlines()]

def npz(path,expected,optional=False):
    """Inspect every authorized header before array allocation; stream actual caps.

    expected maps exact keys to (shape, dtype), supplied by adapter code, not data.
    Returns closed, copied arrays and the authorized numerical payload byte count.
    """
    import numpy as np
    if not isinstance(expected,dict):raise TypeError('Adapter schema required')
    arrays={};payload=0
    with zipfile.ZipFile(io.BytesIO(bounded(path))) as archive:
        members=archive.infolist();names=[m.filename for m in members]
        permitted={k+'.npy' for k in expected}
        if len(names)!=len(set(names)) or not set(names).issubset(permitted) or (not optional and set(names)!=permitted):raise ValueError('NPZ exact key inventory')
        for member in members:
            if member.file_size>MEMBER_CAP or member.flag_bits&1:raise ValueError('Oversized/encrypted NPZ member')
            key=member.filename[:-4];shape,dtype=expected[key]
            dtype=np.dtype(dtype)
            if dtype.hasobject or dtype.kind not in 'fiu' or any(type(n) is not int or n<0 for n in shape):raise ValueError('Unsupported adapter array schema')
            count=1
            for n in shape:count*=n
            needed=count*dtype.itemsize
            payload+=needed
            if needed>MEMBER_CAP or payload>PAYLOAD_CAP:raise ValueError('Numeric payload cap')
            with archive.open(member) as stream:
                version=np.lib.format.read_magic(stream)
                if version not in ((1,0),(2,0)):raise ValueError('Unsupported NPY header version')
                length_bytes=stream.read(2 if version==(1,0) else 4)
                if len(length_bytes)!=(2 if version==(1,0) else 4):raise ValueError('Truncated NPY header length')
                header_length=struct.unpack('<H' if version==(1,0) else '<I',length_bytes)[0]
                if header_length>HEADER_CAP:raise ValueError('NPY header exceeds bound before parsing/allocation')
                header=stream.read(header_length)
                if len(header)!=header_length:raise ValueError('Truncated NPY header')
                reader=np.lib.format.read_array_header_1_0 if version==(1,0) else np.lib.format.read_array_header_2_0
                actual,fortran,actual_dtype=reader(io.BytesIO(length_bytes+header),max_header_size=HEADER_CAP)
                consumed=stream.tell()
                if tuple(actual)!=tuple(shape) or actual_dtype!=dtype or actual_dtype.hasobject or fortran:
                    raise ValueError('NPY shape/dtype/order mismatch')
                if consumed+needed!=member.file_size:raise ValueError('Advertised NPY byte count mismatch')
                data=bytearray()
                while True:
                    block=stream.read(min(65536,needed-len(data)+1))
                    if not block:break
                    data.extend(block)
                    if len(data)>needed or consumed+len(data)>MEMBER_CAP:raise ValueError('Actual decompression cap')
                if len(data)!=needed:raise ValueError('Truncated NPY payload')
                value=np.frombuffer(data,dtype=dtype).reshape(shape).copy()
                if not np.isfinite(value).all():raise ValueError('Nonfinite numeric measurement')
                arrays[key]=value
    return arrays,payload

def finalize(store):
    if relative(store.root,'files.sha256.json').exists():raise ValueError('Already finalized')
    store.json('files.sha256.json',hashes(store.root))

def integrity_files(root):
    value=read_json(relative(root,'files.sha256.json'))
    if value.get('schema')=='llm-microscope-integrity/1':
        if set(value)!={'schema','files','processing_receipt'}:raise ValueError('Unknown integrity metadata')
        receipt=value['processing_receipt'];observed=receipt.get('actual_parent_observation')
        if type(observed) not in (int,float) or not math.isfinite(observed) or observed<0 or observed>receipt.get('charged_seconds_upper_bound',-1):
            raise ValueError('Unclosed/inconsistent terminal processing receipt')
        return value['files']
    return value

def finalized(root):
    expected=integrity_files(root)
    check_hashes(root,expected,('files.sha256.json',))
    return digest(bounded(relative(root,'files.sha256.json')))