"""Closed draft-3 paid-evidence snapshots and inert verified archives.

Public keys and origin documents carried by an exporter never bootstrap trust.
ExternalTrust is trusted in-process configuration, not part of the wire bundle.
"""
from dataclasses import dataclass, asdict
import copy
import datetime as dt
import json
import os
from pathlib import Path
import sqlite3
import stat
import uuid

from odexa_ref import crypto, policy
from . import delegation as d, free_contracts as f, paid_contracts as pc, payments as pay
from .authority_transport import AuthoritySnapshot

PROFILE = 'odexa-paid-evidence-snapshot-1'
TYPE = 'odexa-evidence-export+jws'
VERSION = '1.2.0-draft.3'
MAX_BUNDLE = 16777216
MAX_ITEMS = 2048
KINDS = {'agreement', 'offer', 'acceptance', 'acceptance_jws', 'receipt_jws',
         'status_jws', 'intake_jws', 'admission', 'decision', 'document', 'client_key',
         'payment_quote', 'payment_accepted', 'payment_state', 'payer_proof', 'payment_audit'}


class PortabilityError(ValueError): pass


def need(ok, message):
    if not ok: raise PortabilityError(message)


def stamp(clock):
    need(callable(clock), 'trusted callable clock required')
    result = clock()
    f.timestamp(result, 'now')
    return result


def utc_now(): return dt.datetime.now(dt.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
def load(raw): return crypto.strict_json(raw, MAX_BUNDLE)
def raw(value):
    encoded = json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(',', ':'), allow_nan=False).encode('utf-8')
    crypto.strict_json(encoded, MAX_BUNDLE)
    return encoded
def digest(value): return crypto.digest(value)
def fields(value, names): policy.keys(value, names.split(), 'portable')


def payment_scalar(state):
    return raw({k:v for k,v in asdict(state).items() if k not in {'quote_bytes','acceptance_bytes','mandate_bytes'}})


def history_documents(history, origin, wanted):
    """Exact previously observed bytes, not an external verifier trust source."""
    found, seen, cursor = {}, set(), 0
    while wanted - set(found):
        page = history.observations(origin, after_sequence=cursor, limit=1000)
        if not page: break
        for observation in page:
            cursor = observation['sequence']
            pair = (observation['authority_revision'],observation['policy_revision'])
            if pair in seen: continue
            seen.add(pair)
            need(len(seen) <= MAX_ITEMS, 'history lookup bound exceeded')
            snapshot = history.history(origin,*pair)
            for name in ('authority_bytes','policy_bytes'):
                body = bytes(snapshot[name]); sha = digest(body)
                if sha in wanted: found[sha] = body
    need(wanted <= set(found), 'independently retained verifier/context history missing')
    return found


def validate_paid_context(offer, pol_raw, meta_raw, terms_raw):
    pc.validate_offer(offer)
    need(all(digest(body) == offer[field] for body,field in [(pol_raw,'policy_digest'),
        (meta_raw,'authority_digest'),(terms_raw,'terms_digest')]), 'paid context exact bytes differ')
    need(0 < len(terms_raw) <= 131072, 'bounded human terms required'); terms_raw.decode('utf-8')
    pol=policy.validate_policy(load(pol_raw));meta=d.validate_authority(load(meta_raw),pol)
    need(pol['policy_id']==offer['policy_id'] and pol['revision']==offer['policy_revision'] and
         meta['revision']==offer['authority_revision'], 'paid context revision mismatch')
    service=next(s for s in meta['services'] if s['id']==offer['service_id'])
    allowed={offer['origin'],policy.parse_url(service['base_url'])['origin']}
    need(all(policy.parse_url(x['url'])['origin'] in allowed for x in offer['context']), 'unappointed context origin')
    decision=policy.evaluate(pol,offer['request'],offer['issued_at'])
    need(decision['decision'] in {'permit','require_agreement'} and decision['obligations']==offer['obligations'],
         'paid offer changes policy permission or duties')
    need(offer['issuer']==service['issuer'], 'paid offer issuer mismatch')
    return pol,meta


@dataclass(frozen=True)
class PaidExternalTrust:
    """Pinned data acquired independently of the candidate bundle.

    current_snapshot comes from the trusted origin transport. pinned_documents
    maps historical policy/authority digests to exact independently retained
    bytes. client_keys maps kid to exact public registry descriptor plus
    revoked:boolean; it must not be derived from exporter-carried key records.
    """
    current_snapshot: object
    pinned_documents: dict
    client_keys: dict
    payer_keys: dict


def _binding(value, binding):
    need(all(value[k] == binding[k] for k in f.BINDING), 'artifact changes agreement service binding')


def _key_descriptor(row):
    return {'client_id': row['client_id'], 'role': row['role'], 'principals': load(bytes(row['principals'])),
            'reporter_id': row['reporter_id'], 'key_id': row['key_id'], 'public_jwk': load(bytes(row['jwk']))}


def _public_key(value):
    fields(value, 'client_id role principals reporter_id key_id public_jwk')
    need(value['role'] in {'agent', 'gateway'}, 'unsupported public reporter role')
    f._token(value['client_id'], 'client_id')
    f.reference(value['reporter_id'], 'reporter_id'); f.reference(value['key_id'], 'key_id')
    policy.array(value['principals'], 'principals', 1, 256)
    need(len(set(value['principals'])) == len(value['principals']), 'duplicate principal')
    for principal in value['principals']: f.reference(principal, 'principal')
    crypto.load_public_jwk(value['public_jwk'])
    return value


def _operation_matches(operation, offer, agreement_id):
    d.validate_operation(operation)
    need(operation['kind'] == 'export_evidence' and operation['agreement_id'] == agreement_id,
         'explicit agreement-scoped export operation required')
    need(operation['request'] == offer['request'], 'export operation scope differs from accepted request')
    need(all(operation[k] == offer[k] for k in ('origin', 'service_id', 'issuer', 'delegation_id')), 'export binding mismatch')


def build_bundle(store, *, agreement_id, snapshot, history, export_operation, private_key, key_id, now=utc_now):
    """Read one consistent source snapshot and sign its declared paid scope.

    The caller authenticates the request and permission to export this agreement.
    No financial assertion, external asset or cross-snapshot event reference is omitted.
    Unsupported or unclosed references fail during validation/import.
    """
    f.identifier(agreement_id, 'agreement_id')
    need(getattr(store, 'storage_version', None) == b'paid-agreement-2', 'explicit paid-agreement-2 source required')
    blobs, inventory = {}, []
    def add(kind, ident, body, media='application/json'):
        need(isinstance(body, bytes) and 0 < len(body) <= 1048576, 'bounded exact artifact bytes required')
        sha = digest(body)
        need(not any(x['kind'] == kind and x['id'] == ident for x in inventory), 'duplicate logical artifact')
        blobs[sha] = body
        inventory.append({'kind': kind, 'id': ident, 'digest': sha, 'size': len(body), 'media_type': media})
        return sha
    def document(body, media='application/json'):
        sha = digest(body)
        if not any(x['kind'] == 'document' and x['id'] == sha for x in inventory):
            add('document', sha, body, media)
        return sha
    with store.connect() as db:
        db.execute('BEGIN')
        row = db.execute('SELECT * FROM agreements WHERE id=?', (agreement_id,)).fetchone()
        need(row is not None, 'unknown agreement')
        need(row['state'] in {'pending_payment', 'active', 'revoked', 'expired'}, 'unsupported paid agreement state')
        offer_raw = bytes(db.execute('SELECT body FROM offers WHERE id=?', (row['offer_id'],)).fetchone()[0])
        offer = pc.validate_offer(load(offer_raw))
        accepted = pc.validate_acceptance(load(bytes(row['acceptance'])), offer_raw)
        receipt_payload = load(crypto.split_jws(row['receipt'])[3])
        pc.validate_receipt(receipt_payload, offer_bytes=offer_raw, acceptance_bytes=bytes(row['acceptance']))
        _operation_matches(export_operation, offer, agreement_id)
        need(store.identity == {k: offer[k] for k in ('origin', 'service_id', 'issuer')}, 'store identity mismatch')
        add('offer', row['offer_id'], offer_raw)
        add('acceptance', agreement_id, bytes(row['acceptance']))
        add('acceptance_jws', agreement_id, row['acceptance_jws'].encode('ascii'), 'application/jose')
        add('receipt_jws', agreement_id, row['receipt'].encode('ascii'), 'application/jose')
        state_fields = 'id offer_id client_id principal_id state status_version effective_at reason access_expires_at use_expires_at'.split()
        add('agreement', agreement_id, raw({k: row[k] for k in state_fields}))
        payment = db.execute('SELECT * FROM payment_states WHERE agreement_id=?',(agreement_id,)).fetchone()
        need(payment is not None, 'paid receipt has no retained payment state')
        quote_raw = bytes(payment['quote']); quote = pc.bind_quote(offer,quote_raw)
        registered = db.execute('SELECT * FROM payment_quotes WHERE agreement_id=?',(agreement_id,)).fetchone()
        need(registered is not None and bytes(registered['raw']) == quote_raw and
             registered['provider_id'] == quote['provider_id'] and registered['quote_id'] == quote['quote_id'],
             'immutable quote registry mismatch')
        add('payment_quote',quote['quote_id'],quote_raw)
        add('payment_accepted',agreement_id,bytes(payment['accepted']))
        add('payment_state',agreement_id,bytes(payment['state']))
        payer_proofs = db.execute('SELECT * FROM payer_proofs WHERE agreement_id=?',(agreement_id,)).fetchall()
        need(len(payer_proofs) == (1 if payment['mandate'] is not None else 0), 'mandate proof closure mismatch')
        for proof in payer_proofs:
            need(bytes(proof['payload']) == bytes(payment['mandate']), 'retained payer mandate differs')
            add('payer_proof',agreement_id,raw({'payer_client_id':proof['payer_client_id'],
                'mandate_jws':proof['mandate_jws'],'payload_b64url':crypto.b64u(bytes(proof['payload'])),
                'public_identity_b64url':crypto.b64u(bytes(proof['public_identity']))}))
        audits = db.execute('SELECT * FROM payment_audit WHERE agreement_id=? ORDER BY sequence',(agreement_id,)).fetchall()
        payment_history = set()
        for audit in audits:
            add('payment_audit',str(audit['sequence']),raw({k:audit[k] for k in ('sequence','kind','recorded_at','state_digest')} |
                {k+'_b64url':crypto.b64u(bytes(audit[k])) if audit[k] is not None else None for k in ('input','context')}))
            if audit['kind'] == 'verification':
                context = load(bytes(audit['context']))
                payment_history.update((context['authority_digest'],context['policy_digest']))
        for context in offer['context']:
            found = db.execute('SELECT body,media_type FROM documents WHERE digest=?', (context['digest'],)).fetchone()
            need(found is not None and found['media_type'] == context['media_type'], 'missing exact offer context')
            need(document(bytes(found['body']), found['media_type']) == context['digest'], 'changed stored context')
        statuses = db.execute('SELECT version,jws FROM status_history WHERE agreement_id=? ORDER BY version', (agreement_id,)).fetchall()
        attestations = db.execute('SELECT digest,version,jws FROM status_attestations WHERE agreement_id=? ORDER BY rowid', (agreement_id,)).fetchall()
        status_refs = []
        for status in list(statuses) + list(attestations):
            wire = status['jws'].encode('ascii'); sha = digest(wire)
            if not any(x['kind'] == 'status_jws' and x['id'] == sha for x in inventory):
                add('status_jws', sha, wire, 'application/jose')
        for status in statuses:
            status_refs.append({'version': status['version'], 'digest': digest(status['jws'].encode('ascii'))})
        records = db.execute('SELECT * FROM records WHERE agreement_id=? ORDER BY seq', (agreement_id,)).fetchall()
        clients = {row['client_id']}
        source_reports = {}
        required_authorities = set()
        for status in list(statuses) + list(attestations):
            required_authorities.add(load(crypto.split_jws(status['jws'])[3])['authority_digest'])
        for record in records:
            need(record['client_id'] is not None, 'unverified intake is unsupported in this profile')
            intake = f.validate_intake(load(crypto.split_jws(record['jws'])[3]))
            event_raw = crypto.unb64u(intake['payload_b64url']); event = f.validate_report(load(event_raw))
            need(event_raw == bytes(record['payload']), 'stored report/intake exact bytes disagree')
            need(event['asset_ref'] is None and not event['derived_from'], 'asset-manifest closure unsupported; cannot export incomplete references')
            source_reports[(event['reporter_id'],event['event_id'])] = (event,digest(event_raw))
            required_authorities.add(intake['authority']['digest'])
            clients.add(record['client_id'])
            add('intake_jws', record['record_id'], record['jws'].encode('ascii'), 'application/jose')
        for event, _ in source_reports.values():
            for related in event['related_events']:
                related_key = (related['reporter_id'],related['event_id'])
                need(related_key in source_reports and source_reports[related_key][1] == related['payload_digest'], 'related event lies outside selected closure')
        admissions = db.execute('SELECT * FROM admissions WHERE agreement_id=? ORDER BY rowid', (agreement_id,)).fetchall()
        for admission in admissions:
            clients.add(admission['gateway_client_id'])
            required_authorities.add(load(bytes(admission['payload']))['admission']['authority_digest'])
            add('admission', admission['request_id'], raw({'gateway_client_id': admission['gateway_client_id'],
                'terminal_event_id': admission['terminal_event_id'], 'response_b64url': crypto.b64u(bytes(admission['payload']))}))
        decisions = db.execute('SELECT * FROM decisions WHERE agreement_id=? ORDER BY rowid', (agreement_id,)).fetchall()
        for decision in decisions:
            ar, pr = bytes(decision['authority_bytes']), bytes(decision['policy_bytes'])
            document(ar); document(pr)
            add('decision', decision['operation_id'], raw({'operation_b64url': crypto.b64u(bytes(decision['operation'])),
                'decision_b64url': crypto.b64u(bytes(decision['decision'])), 'observation_b64url': crypto.b64u(bytes(decision['observation'])),
                'authority_digest': digest(ar), 'policy_digest': digest(pr)}))
        for client_id in sorted(clients):
            client = db.execute('SELECT * FROM clients WHERE client_id=?', (client_id,)).fetchone()
            need(client is not None and client['jwk'] is not None, 'missing public reporter registry')
            descriptor = _public_key(_key_descriptor(client))
            add('client_key', descriptor['key_id'], raw(descriptor))
        cut = {'records_max_seq': max((r['seq'] for r in records), default=0), 'records_count': len(records),
               'status_history': status_refs, 'status_attestations_count': len(attestations),
               'admissions_count': len(admissions), 'decisions_count': len(decisions), 'public_keys_count': len(clients),
               'payment_audit_count':len(audits),'payment_audit_max_sequence':max((r['sequence'] for r in audits),default=0),
               'payer_proofs_count':len(payer_proofs)}
    for body in history_documents(history,offer['origin'],(required_authorities|payment_history)-set(blobs)).values():
        document(body)
    # Current authority is a separate observation, not part of the SQLite cut.
    history.observe(snapshot, now=now)
    document(snapshot.authority_bytes); document(snapshot.policy_bytes)
    history.require_current(snapshot)
    created = stamp(now)
    decision = d.evaluate_authority(snapshot.authority_bytes, snapshot.policy_bytes, export_operation,
                                   now=created, observation=snapshot.observation)
    need(decision['decision'] == 'allow', 'current export authority denied')
    need(export_operation['key_id'] == key_id and export_operation['key_use'] == TYPE, 'wrong export signing role')
    metadata = load(snapshot.authority_bytes)
    service = next(s for s in metadata['services'] if s['id'] == offer['service_id'])
    key = next(k for k in service['signing_keys'] if k['kid'] == key_id)
    need(crypto.public_jwk(private_key) == key['public_jwk'], 'export private key does not match origin appointment')
    manifest = {**{k: offer[k] for k in f.BINDING}, 'profile': PROFILE, 'bundle_id': str(uuid.uuid4()),
        'created_at': created, 'agreement_id': agreement_id, 'cut': cut,
        'source_scope': 'all_retained_paid_material_for_one_agreement_at_sqlite_snapshot',
        'global_completeness_claimed': False, 'reactivates_access': False,
        'operation': export_operation, 'authority_digest': digest(snapshot.authority_bytes),
        'policy_digest': digest(snapshot.policy_bytes), 'inventory': sorted(inventory, key=lambda x: (x['kind'], x['id']))}
    need(len(inventory) <= MAX_ITEMS, 'snapshot exceeds item bound; never truncate')
    manifest_jws = crypto.sign_jws(raw(manifest), private_key, key_id, TYPE, allowed_types={TYPE})
    result = raw({'protocol_version': VERSION, 'profile': PROFILE, 'manifest_jws': manifest_jws,
                  'blobs': [{'digest': sha, 'body_b64url': crypto.b64u(body)} for sha, body in sorted(blobs.items())]})
    need(len(result) <= MAX_BUNDLE, 'snapshot exceeds byte bound; never truncate')
    history.require_current(snapshot)
    final = d.evaluate_authority(snapshot.authority_bytes, snapshot.policy_bytes, export_operation,
        now=stamp(now), observation=snapshot.observation)
    need(final['decision'] == 'allow', 'export authority expired before bundle emission')
    return result


def _verify_bundle(bundle_bytes, *, trust, now=utc_now):
    """Verify closure, exact bytes, authority, signatures and free correlations.

    Current compromised material is rejected even for an old signature. Success
    is historical evidence verification, never reactivation or proof of use.
    """
    need(isinstance(trust, PaidExternalTrust), 'external trust configuration required')
    bundle = load(bundle_bytes); fields(bundle, 'protocol_version profile manifest_jws blobs')
    need(bundle['protocol_version'] == VERSION and bundle['profile'] == PROFILE, 'unsupported archive profile')
    _, _, header, manifest_raw, _ = crypto.split_jws(bundle['manifest_jws'], allowed_types={TYPE})
    manifest = load(manifest_raw)
    fields(manifest, 'protocol_version origin service_id issuer delegation_id profile bundle_id created_at agreement_id cut source_scope global_completeness_claimed reactivates_access operation authority_digest policy_digest inventory')
    f.validate_binding({k: manifest[k] for k in f.BINDING})
    f.identifier(manifest['bundle_id'], 'bundle_id'); f.identifier(manifest['agreement_id'], 'agreement_id')
    need(manifest['profile'] == PROFILE and manifest['source_scope'] == 'all_retained_paid_material_for_one_agreement_at_sqlite_snapshot'
         and manifest['global_completeness_claimed'] is False and manifest['reactivates_access'] is False, 'unsupported completeness or access claim')
    instant = stamp(now)
    need(f.timestamp(manifest['created_at'], 'created_at') <= f.timestamp(instant, 'now'), 'future snapshot')
    policy.array(manifest['inventory'], 'inventory', 1, MAX_ITEMS)
    policy.array(bundle['blobs'], 'blobs', 1, MAX_ITEMS)
    blobs = {}
    for entry in bundle['blobs']:
        fields(entry, 'digest body_b64url'); f.hash_value(entry['digest'], 'digest')
        body = crypto.unb64u(entry['body_b64url'])
        need(0 < len(body) <= 1048576 and digest(body) == entry['digest'], 'blob hash/size mismatch')
        need(entry['digest'] not in blobs, 'duplicate blob')
        blobs[entry['digest']] = body
    items = {}
    for entry in manifest['inventory']:
        fields(entry, 'kind id digest size media_type')
        need(entry['kind'] in KINDS, 'unsupported artifact kind')
        policy.text(entry['id'], 'id'); f.hash_value(entry['digest'], 'digest'); policy.integer(entry['size'], 'size', 1)
        expected_media = 'application/jose' if entry['kind'].endswith('_jws') else 'application/json'
        need(entry['media_type'] == expected_media or (entry['kind'] == 'document' and entry['media_type'] == 'text/plain'), 'unsupported artifact media type')
        key = (entry['kind'], entry['id'])
        need(key not in items and entry['digest'] in blobs and len(blobs[entry['digest']]) == entry['size'], 'inventory incomplete/duplicate')
        if entry['kind'] == 'document': need(entry['id'] == entry['digest'], 'document identifier mismatch')
        items[key] = blobs[entry['digest']]
    need(set(blobs) == {x['digest'] for x in manifest['inventory']}, 'unlisted extra blob')
    def rows(kind): return {ident: value for (typ, ident), value in items.items() if typ == kind}
    def one(kind, ident=None):
        values = rows(kind)
        need(len(values) == 1 and (ident is None or ident in values), 'single required artifact missing')
        return next(iter(values.values()))
    documents = rows('document')
    trusted_documents = copy.deepcopy(trust.pinned_documents)
    provided = trust.current_snapshot
    snapshot = AuthoritySnapshot(provided.authority_bytes,provided.policy_bytes,copy.deepcopy(provided.observation))
    trusted_clients = copy.deepcopy(trust.client_keys)
    for value in (snapshot.authority_bytes, snapshot.policy_bytes): trusted_documents[digest(value)] = value
    for sha, body in trusted_documents.items(): need(digest(body) == sha, 'invalid external document pin')
    def pinned(sha):
        need(sha in documents and trusted_documents.get(sha) == documents[sha], 'document lacks independent origin pin')
        return documents[sha]
    current_meta = d.validate_authority(load(snapshot.authority_bytes), load(snapshot.policy_bytes))
    need(current_meta['origin'] == manifest['origin'], 'external trust belongs to another origin')
    revoked = {raw(k['public_jwk']) for s in current_meta['services'] for k in s['signing_keys'] if k['state'] == 'revoked'}
    # A previously observed compromise must not disappear merely because current
    # metadata omits the revoked key. Only independently pinned history counts.
    pinned_policies = []
    for candidate in trusted_documents.values():
        try:
            pol = policy.validate_policy(load(candidate))
            if pol['origin'] == manifest['origin']: pinned_policies.append(pol)
        except (ValueError, KeyError, TypeError): pass
    for candidate in trusted_documents.values():
        for pol in pinned_policies:
            try:
                historical = d.validate_authority(load(candidate), pol)
                revoked.update(raw(k['public_jwk']) for s in historical['services']
                               for k in s['signing_keys'] if k['state'] == 'revoked')
                break
            except (ValueError, KeyError, TypeError): pass
    for descriptor in trusted_clients.values():
        if descriptor.get('revoked') is True: revoked.add(raw(descriptor['public_jwk']))
    offer_raw = one('offer')
    offer = pc.validate_offer(load(offer_raw)); _binding(offer, manifest)
    need(set(rows('offer')) == {offer['offer_id']}, 'offer logical identifier mismatch')
    agreement_id = manifest['agreement_id']
    _operation_matches(manifest['operation'], offer, agreement_id)
    need(header['kid'] == manifest['operation']['key_id'], 'manifest signing key differs from operation')
    historical_meta = d.validate_authority(load(pinned(manifest['authority_digest'])), load(pinned(manifest['policy_digest'])))
    historical_observation = dict(source='https', origin=manifest['origin'], authority_url=manifest['origin']+'/odexa-service.json',
        policy_url=manifest['origin']+'/odexa.json', authority_digest=manifest['authority_digest'], policy_digest=manifest['policy_digest'],
        checked_at=manifest['created_at'], available=True, tls_verified=True, redirected=False, revalidated=True)
    historical_decision = d.evaluate_authority(pinned(manifest['authority_digest']), pinned(manifest['policy_digest']),
        manifest['operation'], now=manifest['created_at'], observation=historical_observation)
    need(historical_decision['decision'] == 'allow', 'manifest lacked export authority at claimed time')
    current = d.evaluate_authority(snapshot.authority_bytes, snapshot.policy_bytes, manifest['operation'], now=instant, observation=snapshot.observation)
    need(current['decision'] == 'allow', 'export signer no longer authorized by current origin')
    def service_key(meta, kid, typ, at):
        service = next((s for s in meta['services'] if s['id'] == manifest['service_id']), None)
        need(service is not None and service['issuer'] == manifest['issuer'], 'service issuer not historically appointed')
        key = next((k for k in service['signing_keys'] if k['kid'] == kid), None)
        need(key is not None and typ in key['uses'] and key['state'] == 'active', 'signature key/type was not active')
        when = f.timestamp(at, 'signed_at')
        need(f.timestamp(meta['issued_at'], 'issued_at') <= when < f.timestamp(meta['expires_at'], 'expires_at'), 'signature outside authority window')
        need(f.timestamp(key['not_before'], 'not_before') <= when < f.timestamp(key['not_after'], 'not_after'), 'signature outside key window')
        need(raw(key['public_jwk']) not in revoked, 'known compromised material cannot validate history')
        return key['public_jwk']
    manifest_key = service_key(historical_meta, header['kid'], TYPE, manifest['created_at'])
    crypto.verify_jws(bundle['manifest_jws'], manifest_key, TYPE, header['kid'], allowed_types={TYPE})
    current_service = next(s for s in current_meta['services'] if s['id'] == manifest['service_id'])
    current_key = next(k for k in current_service['signing_keys'] if k['kid'] == header['kid'])
    need(current_key['public_jwk'] == manifest_key, 'current key ID changed public material')
    for context in offer['context']:
        listed = next((x for x in manifest['inventory'] if x['kind'] == 'document' and x['id'] == context['digest']), None)
        need(listed is not None and listed['media_type'] == context['media_type'], 'context media type differs from pinned offer')
    f.validate_context_bundle(offer, policy_bytes=pinned(offer['policy_digest']), authority_bytes=pinned(offer['authority_digest']),
                             terms_bytes=documents.get(offer['terms_digest'], b''))
    needed_docs = {offer['policy_digest'], offer['authority_digest'], offer['terms_digest'],
                   manifest['authority_digest'], manifest['policy_digest']}
    def authority(sha):
        value = load(pinned(sha))
        # Every historical authority is validated against at least one independently
        # pinned policy in this closed inventory with the same origin/lineage.
        candidates = []
        for psha, candidate in documents.items():
            if trusted_documents.get(psha) != candidate: continue
            try:
                pol = policy.validate_policy(load(candidate))
                d.validate_authority(value, pol); candidates.append(pol)
            except (ValueError, KeyError, TypeError): pass
        need(bool(candidates), 'historical authority has no pinned matching policy')
        needed_docs.add(sha)
        return value
    def signed(wire, typ, sha, at):
        _, _, h, payload, _ = crypto.split_jws(wire)
        meta = authority(sha)
        key = service_key(meta, h['kid'], typ, at)
        service = next(s for s in meta['services'] if s['id'] == manifest['service_id'])
        kind = {'odexa-receipt+jws': 'accept', 'odexa-status+jws': 'status', 'odexa-event-record+jws': 'report'}[typ]
        op = dict(operation_id='00000000-0000-4000-8000-000000000001', payment_request_digest=None, kind=kind,
                  origin=manifest['origin'], service_id=manifest['service_id'], issuer=manifest['issuer'],
                  endpoint=service['base_url'] + ({'accept':'agreements', 'report':'events'}.get(kind) or 'agreements/'+agreement_id+'/status'),
                  delegation_id=manifest['delegation_id'], key_id=h['kid'], key_use=typ, request=offer['request'],
                  offer_seconds=1 if kind == 'accept' else 0, access_seconds=offer['access_seconds'] if kind == 'accept' else 0,
                  use_seconds=offer['use_seconds'] if kind == 'accept' else 0, payment_mode='none', asset_id=None, version_id=None)
        if kind == 'status': op['agreement_id'] = agreement_id
        authorized = False
        for psha, candidate in documents.items():
            if trusted_documents.get(psha) != candidate: continue
            observation = dict(historical_observation, authority_digest=sha, policy_digest=psha, checked_at=at)
            if d.evaluate_authority(pinned(sha), candidate, op, now=at, observation=observation)['decision'] == 'allow':
                authorized = True
                break
        need(authorized, 'historical signature exceeded scoped origin delegation')
        body, _ = crypto.verify_jws(wire, key, typ, h['kid'])
        need(body == payload, 'signature payload mismatch')
        value = load(body); _binding(value, manifest)
        if typ == 'odexa-status+jws':
            need(value['authority_revision'] == meta['revision'], 'status authority revision/digest mismatch')
        elif typ == 'odexa-event-record+jws':
            need(value['authority']['revision'] == meta['revision'], 'intake authority revision/digest mismatch')
        return value
    registry = {}
    for kid, encoded in rows('client_key').items():
        value = _public_key(load(encoded))
        need(value['key_id'] == kid, 'registry key ID mismatch')
        external = trusted_clients.get(kid)
        need(isinstance(external, dict) and set(external) == set(value) | {'revoked'}, 'external client key pin required')
        need(type(external['revoked']) is bool and not external['revoked'] and
             {k: external[k] for k in value} == value and raw(value['public_jwk']) not in revoked, 'client registry not independently trusted')
        registry[kid] = value
    def client_signature(wire, typ):
        _, _, h, _, _ = crypto.split_jws(wire)
        descriptor = registry.get(h['kid'])
        need(descriptor is not None, 'reporter key missing from closed external registry')
        payload, _ = crypto.verify_jws(wire, descriptor['public_jwk'], typ, h['kid'])
        return payload, descriptor
    acceptance_raw = one('acceptance', agreement_id)
    accepted_payload, owner = client_signature(one('acceptance_jws', agreement_id), 'odexa-acceptance+jws')
    need(accepted_payload == acceptance_raw and owner['role'] == 'agent', 'exact assent bytes or role mismatch')
    acceptance = pc.validate_acceptance(load(acceptance_raw), offer_raw, manifest['created_at'])
    need(owner['client_id'] == offer['client_id'] and offer['principal_id'] in owner['principals'], 'agent key is not authorized for accepted principal')
    receipt_wire = one('receipt_jws', agreement_id)
    receipt_payload = load(crypto.split_jws(receipt_wire)[3])
    receipt = signed(receipt_wire, 'odexa-receipt+jws', offer['authority_digest'], receipt_payload['recorded_at'])
    pc.validate_receipt(receipt, offer_bytes=offer_raw, acceptance_bytes=acceptance_raw)
    need(receipt['agreement_id'] == agreement_id, 'receipt outside snapshot scope')
    need(receipt['recorded_at'] <= manifest['created_at'], 'receipt exceeds snapshot time')
    state = load(one('agreement', agreement_id))
    fields(state, 'id offer_id client_id principal_id state status_version effective_at reason access_expires_at use_expires_at')
    need(state['id'] == agreement_id and state['state'] in {'active', 'revoked', 'expired'}, 'unsupported paid or mismatched indexed agreement')
    for key in ('offer_id', 'client_id', 'principal_id', 'access_expires_at', 'use_expires_at'):
        need(state[key] == receipt[key], 'indexed agreement changes immutable receipt')
    policy.integer(state['status_version'], 'status_version', 1)
    f.timestamp(state['effective_at'], 'effective_at')
    policy.text(state['reason'], 'reason', 256)
    need(receipt['recorded_at'] <= state['effective_at'] <= manifest['created_at'], 'indexed transition outside receipt/snapshot interval')
    initial = ('active', receipt['recorded_at'], 'accepted')
    if state['state'] == 'active':
        need(state['status_version'] == 1 and (state['state'],state['effective_at'],state['reason']) == initial,
             'free initial indexed state differs from receipt')
    else:
        need(state['status_version'] == 2, 'free terminal transition must be lifecycle version two')
        if state['state'] == 'expired':
            need(state['effective_at'] == receipt['access_expires_at'] and state['reason'] == 'access_expired',
                 'indexed expiry differs from accepted access deadline')
        else:
            need(state['effective_at'] < receipt['access_expires_at'], 'revocation cannot follow access expiry')
    status_values = {}
    for sha, wire in rows('status_jws').items():
        need(sha == digest(wire), 'status artifact digest ID mismatch')
        value = load(crypto.split_jws(wire)[3])
        value = signed(wire, 'odexa-status+jws', value['authority_digest'], value['issued_at'])
        pc.validate_status(value, receipt)
        need(value['status_version'] <= state['status_version'] and value['issued_at'] <= manifest['created_at'], 'status exceeds snapshot')
        status_values[sha] = value
    states = {}
    for value in status_values.values():
        ver = value['status_version']
        meaning = (value['state'], value['effective_at'], value['reason'])
        need((ver == 1 and meaning == initial) or
             (ver == 2 and value['state'] in {'revoked','expired'}), 'unsupported free signed lifecycle')
        need(ver not in states or states[ver] == meaning, 'conflicting status meanings at one lifecycle version')
        states[ver] = meaning
    terminal = False
    for ver, meaning in sorted(states.items()):
        need(not terminal or meaning[0] != 'active', 'status history reactivates terminal access')
        terminal = terminal or meaning[0] in {'revoked', 'expired'}
    if state['status_version'] in states:
        need(states[state['status_version']] == (state['state'], state['effective_at'], state['reason']), 'snapshot state contradicts signed status')
    if state['state'] == 'revoked':
        need(2 in states, 'indexed revocation lacks original signed terminal status')
    if state['state'] != 'active':
        need(all(value['state'] != 'active' or value['issued_at'] <= state['effective_at']
                 for value in status_values.values()), 'active attestation follows terminal transition')
    need(not terminal or state['state'] != 'active', 'snapshot state reactivates historical terminal status')
    indexed_assurance = ('signed_status_correlated' if state['status_version'] in states else
                         'receipt_correlated_initial_state' if state['state'] == 'active' else
                         'exporter_asserted_lazy_expiry_at_receipt_deadline')
    admissions = {}
    for request_id, encoded in rows('admission').items():
        entry = load(encoded); fields(entry, 'gateway_client_id terminal_event_id response_b64url')
        response = load(crypto.unb64u(entry['response_b64url']))
        pc.validate_introspection_response(response, receipt=receipt); _binding(response, manifest)
        need(response['permitted'] and response['request_id'] == request_id, 'unbound or denied admission')
        admission = response['admission']
        need(receipt['recorded_at'] <= admission['checked_at'] <= manifest['created_at'],
             'admission outside receipt/snapshot interval')
        need(admission['status_version'] == 1, 'free admission must name initial active lifecycle')
        if state['state'] != 'active':
            need(admission['checked_at'] <= state['effective_at'], 'admission follows terminal transition')
        needed_docs.add(response['admission']['authority_digest']); pinned(response['admission']['authority_digest'])
        admission_meta = authority(response['admission']['authority_digest'])
        need(response['admission']['authority_revision'] == admission_meta['revision'], 'admission authority revision mismatch')
        need(response['admission']['policy_digest'] == offer['policy_digest']
             and response['admission']['policy_id'] == offer['policy_id']
             and response['admission']['policy_revision'] == offer['policy_revision'], 'admission changed accepted policy')
        need(response['admission']['resource_url'] == offer['request']['url']
             and set(response['admission']['actions']) <= set(offer['request']['actions'])
             and set(response['admission']['purposes']) <= set(offer['request']['purposes']), 'admission expands accepted scope')
        admission_service = next(s for s in admission_meta['services'] if s['id'] == manifest['service_id'])
        admission_operation = dict(operation_id=admission['decision_id'], payment_request_digest=None,
            kind='introspect', origin=manifest['origin'], service_id=manifest['service_id'], issuer=manifest['issuer'],
            endpoint=admission_service['base_url']+'introspect', delegation_id=manifest['delegation_id'],
            key_id=None, key_use=None, request=offer['request'], offer_seconds=0,
            access_seconds=offer['access_seconds'], use_seconds=offer['use_seconds'],
            payment_mode='none', asset_id=None, version_id=None)
        admission_observation = dict(historical_observation, authority_digest=admission['authority_digest'],
            policy_digest=offer['policy_digest'], checked_at=admission['authority_checked_at'])
        need(d.evaluate_authority(pinned(admission['authority_digest']), pinned(offer['policy_digest']),
            admission_operation, now=admission['checked_at'], observation=admission_observation)['decision'] == 'allow',
            'admission exceeds pinned historical authority scope')
        admissions[request_id] = (entry, response)
    reports, used_keys = {}, {owner['key_id']}
    terminal_reports = {}
    for record_id, wire in rows('intake_jws').items():
        value = load(crypto.split_jws(wire)[3])
        value = signed(wire, 'odexa-event-record+jws', value['authority']['digest'], value['received_at'])
        f.validate_intake(value)
        need(value['record_id'] == record_id and value['received_at'] <= manifest['created_at'], 'intake outside snapshot')
        need(value['assurance'] in {'client_key_verified', 'origin_key_verified'}, 'unsupported unverified/service report')
        event_raw, reporter = client_signature(value['reporter_jws'], 'odexa-event+jws'); used_keys.add(reporter['key_id'])
        need(event_raw == crypto.unb64u(value['payload_b64url']), 'collector changed reporter bytes')
        event = f.validate_report(load(event_raw), role=reporter['role'], reporter_id=reporter['reporter_id'])
        _binding(event, manifest)
        need(event['agreement_id'] == agreement_id and event['resource_url'] == offer['request']['url']
             and event['policy_id'] == offer['policy_id'] and event['policy_revision'] == offer['policy_revision'], 'report agreement/context mismatch')
        need(set(event['actions']) <= set(offer['request']['actions']) and set(event['purposes']) <= set(offer['request']['purposes']), 'report expands accepted scope')
        need(event['asset_ref'] is None and not event['derived_from'], 'asset-manifest closure unsupported; cannot omit referenced evidence')
        need(event['occurred_at'] <= value['received_at'], 'collector intake predates reported event')
        if reporter['role'] == 'agent': need(reporter['client_id'] == receipt['client_id'], 'foreign agent report')
        else:
            request_id = event['http']['delivery_id']
            need(request_id in admissions, 'delivery admission missing')
            admission, response = admissions[request_id]
            f.validate_report(event, role='gateway', admission=response)
            need(admission['gateway_client_id'] == reporter['client_id'] and admission['terminal_event_id'] == event['event_id'], 'gateway journal binding mismatch')
            need(request_id not in terminal_reports, 'multiple terminal reports for one delivery')
            terminal_reports[request_id] = event['event_id']
        key = (event['reporter_id'], event['event_id'])
        need(key not in reports, 'duplicate reporter event')
        reports[key] = (event, digest(event_raw))
    for event, _ in reports.values():
        for related in event['related_events']:
            key = (related['reporter_id'], related['event_id'])
            need(key in reports and reports[key][1] == related['payload_digest'], 'related event missing from closure')
    for _, (entry, response) in admissions.items():
        if entry['terminal_event_id'] is not None:
            need(response['request_id'] in terminal_reports, 'terminal admission lacks exact report')
        # Gateway public registry is required even for an unresolved admission.
        candidates = [v for v in registry.values() if v['client_id'] == entry['gateway_client_id'] and v['role'] == 'gateway']
        need(len(candidates) == 1, 'gateway registry missing/ambiguous'); used_keys.add(candidates[0]['key_id'])
    need(set(registry) == used_keys, 'unrelated public key outside declared closure')
    for operation_id, encoded in rows('decision').items():
        entry = load(encoded); fields(entry, 'operation_b64url decision_b64url observation_b64url authority_digest policy_digest')
        ar, pr = pinned(entry['authority_digest']), pinned(entry['policy_digest'])
        needed_docs.update((entry['authority_digest'], entry['policy_digest']))
        op, outcome, observation = [load(crypto.unb64u(entry[name])) for name in ('operation_b64url', 'decision_b64url', 'observation_b64url')]
        need(op['operation_id'] == operation_id and outcome['decision'] == 'allow', 'unbound historical decision')
        need(op['request'] == offer['request'] and op['origin'] == manifest['origin'] and op['service_id'] == manifest['service_id'], 'historical decision outside agreement scope')
        if 'agreement_id' in op: need(op['agreement_id'] == agreement_id, 'decision names another agreement')
        # The existing decision stores acquisition time, not an exact execution
        # timestamp. Verify there exists an authorized point in its declared
        # <=5-second observation window; do not invent stronger timing evidence.
        checked = f.timestamp(outcome['authority']['checked_at'], 'checked_at')
        matches = False
        for delta in range(6):
            at = (checked + dt.timedelta(seconds=delta)).strftime('%Y-%m-%dT%H:%M:%SZ')
            if at > manifest['created_at']: continue
            if d.evaluate_authority(ar, pr, op, now=at, observation=observation) == outcome:
                matches = True
                break
        need(matches, 'historical decision correlation invalid')
    need(set(documents) == needed_docs, 'missing or unrelated context/history document')
    cut = manifest['cut']
    fields(cut, 'records_max_seq records_count status_history status_attestations_count admissions_count decisions_count public_keys_count')
    for key in ('records_max_seq','records_count','status_attestations_count','admissions_count','decisions_count','public_keys_count'): policy.integer(cut[key], key)
    need(cut['records_count'] == len(reports) and cut['admissions_count'] == len(admissions)
         and cut['decisions_count'] == len(rows('decision')) and cut['public_keys_count'] == len(registry), 'signed snapshot count mismatch')
    need((cut['records_count'] == 0) == (cut['records_max_seq'] == 0) and cut['records_max_seq'] >= cut['records_count'], 'invalid record cut')
    seen_versions = set()
    for item in cut['status_history']:
        fields(item, 'version digest'); policy.integer(item['version'], 'version', 1)
        need(item['version'] not in seen_versions and item['digest'] in status_values
             and status_values[item['digest']]['status_version'] == item['version'], 'state history closure mismatch')
        seen_versions.add(item['version'])
    need(cut['status_attestations_count'] <= len(status_values)
         and len(status_values) <= cut['status_attestations_count'] + len(seen_versions), 'status attestation count mismatch')
    final_time = stamp(now)
    final_authority = d.evaluate_authority(snapshot.authority_bytes, snapshot.policy_bytes, manifest['operation'],
        now=final_time, observation=snapshot.observation)
    need(final_authority['decision'] == 'allow', 'export authority expired during verification')
    return {'profile': PROFILE, 'origin': manifest['origin'], 'bundle_id': manifest['bundle_id'],
            'agreement_id': agreement_id, 'manifest': manifest, 'bundle_digest': digest(bundle_bytes),
            'verified_at': final_time, 'archive_only': True, 'reactivates_access': False,
            'assurance': 'signatures_and_declared_snapshot_closure_verified',
            'lifecycle': {'indexed_state': state['state'], 'indexed_status_version': state['status_version'],
                          'indexed_effective_at': state['effective_at'], 'indexed_state_assurance': indexed_assurance,
                          'signed_status_versions': sorted(states),
                          'access_at_snapshot': ('expired' if state['state'] == 'active' and
                              manifest['created_at'] >= receipt['access_expires_at'] else state['state']),
                          'equal_second_transition_order_verified': False},
            'global_completeness_verified': False, 'downstream_use_verified': False}


def verify_bundle(bundle_bytes, *, trust, now=utc_now):
    """Public verifier: invalid input yields one fail-closed exception family."""
    try:
        return _verify_bundle(bundle_bytes, trust=trust, now=now)
    except (ValueError, TypeError, KeyError, OverflowError, AttributeError, StopIteration) as exc:
        if isinstance(exc, PortabilityError): raise
        raise PortabilityError('invalid portable evidence: ' + str(exc)) from exc


class PaidEvidenceArchive:
    """An inert private destination; no operational import or credential table."""
    def __init__(self, path):
        self.path = Path(path).absolute(); parent = self.path.parent
        parent.mkdir(mode=0o700, parents=True, exist_ok=True)
        info = parent.lstat()
        need(stat.S_ISDIR(info.st_mode) and info.st_uid == os.getuid() and not info.st_mode & 0o077, 'private owned archive directory required')
        fd = os.open(self.path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
        try:
            info = os.fstat(fd)
            need(stat.S_ISREG(info.st_mode) and info.st_uid == os.getuid() and not info.st_mode & 0o077 and info.st_nlink == 1, 'private owned archive file required')
        finally: os.close(fd)
        db = sqlite3.connect(self.path, isolation_level=None)
        try:
            db.execute('PRAGMA synchronous=FULL'); db.execute('BEGIN IMMEDIATE')
            tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")}
            need(tables <= {'archives'}, 'archive destination must be separate from operational databases')
            db.execute('CREATE TABLE IF NOT EXISTS archives(origin TEXT NOT NULL,bundle_id TEXT NOT NULL,digest TEXT NOT NULL,body BLOB NOT NULL,report BLOB NOT NULL,PRIMARY KEY(origin,bundle_id))')
            for action in ('UPDATE','DELETE'):
                db.execute(f"CREATE TRIGGER IF NOT EXISTS immutable_archive_{action} BEFORE {action} ON archives BEGIN SELECT RAISE(ABORT,'immutable archive'); END")
            db.commit()
        except BaseException:
            db.rollback(); raise
        finally: db.close()

    def import_bundle(self, bundle_bytes, *, trust, now=utc_now):
        report = verify_bundle(bundle_bytes, trust=trust, now=now)
        db = sqlite3.connect(self.path, timeout=10, isolation_level=None)
        try:
            db.execute('PRAGMA synchronous=FULL'); db.execute('PRAGMA recursive_triggers=ON'); db.execute('BEGIN IMMEDIATE')
            # Acquiring an archive lock may wait beyond the observation window.
            # Reverify with a post-lock trusted clock, never a pre-lock timestamp.
            report = verify_bundle(bundle_bytes, trust=trust, now=now)
            old = db.execute('SELECT digest,body,report FROM archives WHERE origin=? AND bundle_id=?', (report['origin'], report['bundle_id'])).fetchone()
            if old:
                need(old[0] == report['bundle_digest'] and old[1] == bundle_bytes, 'bundle identity conflicts with retained exact archive')
                result = load(old[2])
            else:
                db.execute('INSERT INTO archives VALUES (?,?,?,?,?)', (report['origin'], report['bundle_id'], report['bundle_digest'], bundle_bytes, raw(report)))
                result = report
            db.commit()
            return result
        except BaseException:
            db.rollback(); raise
        finally: db.close()

    def get(self, origin, bundle_id):
        policy.origin(origin); f.identifier(bundle_id, 'bundle_id')
        db = sqlite3.connect(self.path)
        try:
            row = db.execute('SELECT body,report FROM archives WHERE origin=? AND bundle_id=?', (origin, bundle_id)).fetchone()
        finally: db.close()
        need(row is not None, 'unknown archive')
        return {'bundle_bytes': row[0], 'import_report': load(row[1]), 'authority_status': 'historical_only', 'reactivates_access': False}
