"""Explicit draft-3 free wire contracts. No authentication, I/O or permission grant.

Draft-1 policy and unversioned URL/HTTP/asset semantics are reused explicitly;
draft-2 envelopes and global version constants are never changed or relabelled.
"""
import re

from odexa_ref import contracts as c, crypto, policy
from . import delegation

VERSION = '1.2.0-draft.3'
BINDING = {'protocol_version', 'origin', 'service_id', 'issuer', 'delegation_id'}
MAX_DURATION = 31536000
ValidationError = c.ValidationError
need, keys, integer, array, text = c.need, c.keys, c.integer, c.array, c.text
timestamp, identifier, reference, hash_value = c.timestamp, c.identifier, c.reference, c.hash_value
load_json, json_bytes, digest = c.load_json, crypto.json_bytes, crypto.digest


def _bound(value, fields, *, max_bytes=131072):
    c.document(value, max_bytes)
    keys(value, BINDING | set(fields.split()), 'document')
    need(value['protocol_version'] == VERSION, 'protocol_version', 'explicit draft-3 envelope required')
    policy.origin(value['origin'])
    reference(value['service_id'], 'service_id', value['origin'])
    reference(value['issuer'], 'issuer')
    if value['delegation_id'] is not None:
        reference(value['delegation_id'], 'delegation_id', value['origin'])


def _same_binding(value, other):
    need(all(value[k] == other[k] for k in BINDING), 'binding', 'origin/service/issuer/delegation mismatch')


def validate_binding(value):
    _bound(value, '')
    return value


def _token(value, name, minimum=1):
    need(isinstance(value, str) and bool(re.fullmatch(r'[A-Za-z0-9_-]{'+str(minimum)+r',128}', value)), name, 'bounded token required')


def _nonce(value, name='nonce'):
    need(isinstance(value, str) and len(value) == 43 and len(crypto.unb64u(value)) == 32, name, 'canonical 32-byte base64url required')


def _seconds(value):
    integer(value, 'seconds', 1)
    need(value <= MAX_DURATION, 'seconds', 'duration exceeds profile')


def _windows(value):
    _seconds(value['access_seconds']); _seconds(value['use_seconds'])
    need(value['access_seconds'] <= value['use_seconds'], 'seconds', 'access exceeds use duration')


def validate_request(value, origin=None):
    keys(value, 'url actions purposes supported_obligations'.split(), 'request')
    reference(value['url'], 'request.url', origin, query=True)
    policy.enum_array(value['actions'], policy.ACTIONS, 'request.actions')
    policy.enum_array(value['purposes'], policy.PURPOSES, 'request.purposes')
    policy.enum_array(value['supported_obligations'], policy.DUTIES, 'request.supported_obligations', 0)
    return value


def validate_obligations(value):
    array(value, 'obligations', 0, 64)
    for duty in value:
        need(isinstance(duty, dict) and isinstance(duty.get('type'), str) and duty['type'] in policy.DUTIES, 'obligation', 'known obligation required')
        if duty['type'] == 'attribution':
            keys(duty, {'type', 'name', 'url'}, 'obligation')
            text(duty['name'], 'name', 256); policy.https_reference(duty['url'], 'url')
        elif duty['type'] == 'retention':
            keys(duty, {'type', 'max_seconds'}, 'obligation'); integer(duty['max_seconds'], 'max_seconds')
        else:
            keys(duty, {'type', 'endpoint', 'deadline_seconds'}, 'obligation')
            policy.https_reference(duty['endpoint'], 'endpoint')
            need(not policy.parse_url(duty['endpoint'])['has_query'], 'endpoint', 'report query forbidden')
            integer(duty['deadline_seconds'], 'deadline_seconds', 1)
    need(policy.combine_obligations([{'obligations': value}]) == value, 'obligations', 'duties must be canonical cumulative set')
    return value


def validate_offer_request(value):
    _bound(value, 'principal_id request access_seconds use_seconds')
    reference(value['principal_id'], 'principal_id'); validate_request(value['request'], value['origin']); _windows(value)
    return value


def validate_offer(value, request=None):
    _bound(value, 'offer_id client_id principal_id policy_id policy_revision policy_digest authority_revision authority_digest terms_digest request obligations context nonce issued_at expires_at access_seconds use_seconds revocation payment')
    identifier(value['offer_id'], 'offer_id'); _token(value['client_id'], 'client_id')
    reference(value['principal_id'], 'principal_id'); reference(value['policy_id'], 'policy_id', value['origin'])
    for field in ('policy_revision', 'authority_revision'): integer(value[field], field, 1)
    for field in ('policy_digest', 'authority_digest', 'terms_digest'): hash_value(value[field], field)
    validate_request(value['request'], value['origin']); validate_obligations(value['obligations'])
    need({o['type'] for o in value['obligations']} <= set(value['request']['supported_obligations']), 'obligations', 'unsupported duty in offer')
    _nonce(value['nonce']); _windows(value)
    issued, expires = timestamp(value['issued_at'], 'issued_at'), timestamp(value['expires_at'], 'expires_at')
    need(0 < (expires-issued).total_seconds() <= 600, 'expires_at', 'offer validity must be 1..600 seconds')
    need(value['revocation'] == 'future_access_only', 'revocation', 'unsupported revocation semantics')
    keys(value['payment'], {'required'}, 'payment'); need(value['payment']['required'] is False, 'payment', 'free profile only')
    array(value['context'], 'context', 3, 3)
    seen = set()
    expected = {'policy': ('policy_digest', 'application/json'), 'human_terms': ('terms_digest', 'text/plain'), 'origin_authority': ('authority_digest', 'application/json')}
    for context in value['context']:
        keys(context, {'role', 'url', 'digest', 'media_type'}, 'context')
        need(isinstance(context['role'], str) and context['role'] in expected and context['role'] not in seen, 'context.role', 'exact unique context roles required')
        seen.add(context['role']); reference(context['url'], 'context.url')
        digest_field, media = expected[context['role']]
        need(context['digest'] == value[digest_field] and context['media_type'] == media, 'context', 'context digest/media mismatch')
    if request is not None:
        validate_offer_request(request); _same_binding(value, request)
        for field in ('principal_id', 'request', 'access_seconds', 'use_seconds'):
            need(value[field] == request[field], field, 'offer differs from requested exchange')
    return value


def validate_context_bundle(offer, *, policy_bytes, authority_bytes, terms_bytes, now=None):
    """Check retained bytes/semantics. Does not authenticate their origin or freshness."""
    validate_offer(offer)
    for raw, name in ((policy_bytes, 'policy_digest'), (authority_bytes, 'authority_digest'), (terms_bytes, 'terms_digest')):
        need(isinstance(raw, bytes) and 0 < len(raw) <= 131072 and digest(raw) == offer[name], name, 'exact context bytes mismatch')
    try: terms_bytes.decode('utf-8', errors='strict')
    except UnicodeDecodeError as exc: raise ValidationError('terms', 'UTF-8 terms required') from exc
    pol = policy.validate_policy(load_json(policy_bytes)); meta = delegation.validate_authority(load_json(authority_bytes), pol)
    need(pol['origin'] == offer['origin'] and pol['policy_id'] == offer['policy_id'] and pol['revision'] == offer['policy_revision'], 'policy', 'pinned policy identity mismatch')
    need(meta['revision'] == offer['authority_revision'], 'authority', 'pinned revision mismatch')
    service = next((s for s in meta['services'] if s['id'] == offer['service_id']), None)
    need(service is not None and service['issuer'] == offer['issuer'], 'service', 'offer issuer not in pinned context')
    grant = None
    if offer['delegation_id'] is not None:
        grant = next((g for g in meta['delegations'] if g['id'] == offer['delegation_id']), None)
        need(grant is not None and grant['delegate_service_id'] == offer['service_id'], 'delegation', 'offer delegation mismatch')
    else:
        need(policy.parse_url(service['base_url'])['origin'] == offer['origin'], 'delegation', 'external provider requires delegation')
    allowed_context_origins = {offer['origin'], policy.parse_url(service['base_url'])['origin']}
    need(all(policy.parse_url(x['url'])['origin'] in allowed_context_origins for x in offer['context']), 'context.url', 'context host outside publisher/provider')
    decision = policy.evaluate(pol, offer['request'], now or offer['issued_at'])
    need(decision['decision'] in {'permit', 'require_agreement'} and decision['obligations'] == offer['obligations'], 'policy', 'offer weakens or changes policy decision/duties')
    issued, expires = timestamp(offer['issued_at'], 'issued_at'), timestamp(offer['expires_at'], 'expires_at')
    for document in [pol, meta] + ([grant] if grant else []):
        need(timestamp(document['issued_at'], 'issued_at') <= issued < expires <= timestamp(document['expires_at'], 'expires_at'), 'offer', 'offer outside pinned authority/policy interval')
    for scope, capabilities, limits in [(service['scope'], service['capabilities'], service['limits'])] + ([(grant, grant['capabilities'], grant)] if grant else []):
        need('issue_agreements' in capabilities, 'capability', 'pinned service cannot issue agreement')
        need(set(decision['matched_resources']) <= set(scope['resource_ids']) and set(offer['request']['actions']) <= set(scope['actions']) and set(offer['request']['purposes']) <= set(scope['purposes']), 'scope', 'offer exceeds pinned authority')
        need(offer['access_seconds'] <= limits['max_access_seconds'] and offer['use_seconds'] <= limits['max_use_seconds'], 'duration', 'offer exceeds pinned duration ceilings')
    need((expires-issued).total_seconds() <= service['limits']['max_offer_seconds'], 'offer', 'offer exceeds pinned lifetime ceiling')
    return {'policy': pol, 'authority': meta, 'policy_decision': decision, 'origin_authenticated': False}


def validate_acceptance(value, offer_bytes=None, now=None):
    _bound(value, 'offer_id offer_digest client_id principal_id intent nonce accepted_at idempotency_key')
    identifier(value['offer_id'], 'offer_id'); hash_value(value['offer_digest'], 'offer_digest')
    _token(value['client_id'], 'client_id'); _token(value['idempotency_key'], 'idempotency_key', 16)
    reference(value['principal_id'], 'principal_id'); _nonce(value['nonce'])
    accepted = timestamp(value['accepted_at'], 'accepted_at')
    need(value['intent'] == 'accept', 'intent', 'explicit assent required')
    if now is not None: need(accepted <= timestamp(now, 'now'), 'accepted_at', 'future assent')
    if offer_bytes is not None:
        offer = validate_offer(load_json(offer_bytes)); _same_binding(value, offer)
        need(digest(offer_bytes) == value['offer_digest'], 'offer_digest', 'acceptance does not bind exact offer bytes')
        for field in ('offer_id', 'client_id', 'principal_id', 'nonce'):
            need(value[field] == offer[field], field, 'acceptance context mismatch')
        need(timestamp(offer['issued_at'], 'issued_at') <= accepted < timestamp(offer['expires_at'], 'expires_at'), 'accepted_at', 'assent outside offer interval')
    return value


def validate_receipt(value, offer_bytes=None, acceptance_bytes=None):
    _bound(value, 'record_type agreement_id client_id principal_id offer_id offer_digest acceptance_digest recorded_at access_expires_at use_expires_at state_at_issue payment_required access_credential')
    need(value['record_type'] == 'agreement.accepted', 'record_type', 'wrong receipt type')
    identifier(value['agreement_id'], 'agreement_id'); identifier(value['offer_id'], 'offer_id')
    _token(value['client_id'], 'client_id'); reference(value['principal_id'], 'principal_id')
    hash_value(value['offer_digest'], 'offer_digest'); hash_value(value['acceptance_digest'], 'acceptance_digest')
    recorded = timestamp(value['recorded_at'], 'recorded_at')
    access = timestamp(value['access_expires_at'], 'access_expires_at'); use = timestamp(value['use_expires_at'], 'use_expires_at')
    need(recorded < access <= use and (use-recorded).total_seconds() <= MAX_DURATION, 'receipt', 'invalid access/use windows')
    need(value['state_at_issue'] == 'active' and value['payment_required'] is False and value['access_credential'] is False, 'receipt', 'free evidence receipt required')
    if offer_bytes is not None:
        offer = validate_offer(load_json(offer_bytes)); _same_binding(value, offer)
        need(value['offer_digest'] == digest(offer_bytes), 'offer_digest', 'exact offer mismatch')
        for field in ('offer_id', 'client_id', 'principal_id'): need(value[field] == offer[field], field, 'receipt offer mismatch')
        need(timestamp(offer['issued_at'], 'issued_at') <= recorded < timestamp(offer['expires_at'], 'expires_at'), 'recorded_at', 'new commit outside offer validity')
        need((access-recorded).total_seconds() == offer['access_seconds'] and (use-recorded).total_seconds() == offer['use_seconds'], 'receipt', 'receipt durations differ from accepted offer')
    if acceptance_bytes is not None:
        acceptance = validate_acceptance(load_json(acceptance_bytes), offer_bytes, value['recorded_at']); _same_binding(value, acceptance)
        need(value['acceptance_digest'] == digest(acceptance_bytes), 'acceptance_digest', 'digest covers exact decoded JWS payload bytes')
        for field in ('offer_id', 'offer_digest', 'client_id', 'principal_id'): need(value[field] == acceptance[field], field, 'receipt acceptance mismatch')
    return value


def validate_status(value, receipt=None):
    _bound(value, 'record_type agreement_id state effective_at reason status_version issued_at authority_revision authority_digest authority_checked_at')
    need(value['record_type'] == 'agreement.status', 'record_type', 'wrong status type')
    identifier(value['agreement_id'], 'agreement_id'); integer(value['status_version'], 'status_version', 1)
    integer(value['authority_revision'], 'authority_revision', 1); hash_value(value['authority_digest'], 'authority_digest')
    text(value['reason'], 'reason', 256)
    need(isinstance(value['state'], str) and value['state'] in {'active', 'revoked', 'expired'}, 'state', 'free access state required')
    issued = timestamp(value['issued_at'], 'issued_at'); effective = timestamp(value['effective_at'], 'effective_at')
    checked = timestamp(value['authority_checked_at'], 'authority_checked_at')
    need(effective <= issued and 0 <= (issued-checked).total_seconds() <= 5, 'status', 'invalid status observation interval')
    if receipt is not None:
        validate_receipt(receipt); _same_binding(value, receipt)
        need(value['agreement_id'] == receipt['agreement_id'], 'agreement_id', 'status for another agreement')
        need(timestamp(receipt['recorded_at'], 'recorded_at') <= effective, 'effective_at', 'status before acceptance')
        if value['state'] == 'active': need(issued < timestamp(receipt['access_expires_at'], 'access_expires_at'), 'state', 'expired active status')
        if value['state'] == 'expired': need(value['effective_at'] == receipt['access_expires_at'], 'effective_at', 'expiry must match accepted access deadline')
    return value


def validate_revoke_request(value, agreement_id=None):
    _bound(value, 'agreement_id reason idempotency_key')
    identifier(value['agreement_id'], 'agreement_id'); _token(value['idempotency_key'], 'idempotency_key', 16)
    text(value['reason'], 'reason', 256)
    if agreement_id is not None: need(value['agreement_id'] == agreement_id, 'agreement_id', 'path/body mismatch')
    return value


def validate_token_request(value):
    _bound(value, 'agreement_id'); identifier(value['agreement_id'], 'agreement_id')
    return value


def validate_token_response(value, receipt=None, now=None):
    _bound(value, 'token_type access_token expires_at agreement_id')
    need(value['token_type'] == 'Bearer', 'token_type', 'opaque bearer token required')
    _nonce(value['access_token'], 'access_token'); identifier(value['agreement_id'], 'agreement_id')
    expiry = timestamp(value['expires_at'], 'expires_at')
    if now is not None:
        instant = timestamp(now, 'now'); need(0 < (expiry-instant).total_seconds() <= 300, 'expires_at', 'token lifetime outside 1..300 seconds')
    if receipt is not None:
        validate_receipt(receipt); _same_binding(value, receipt)
        need(value['agreement_id'] == receipt['agreement_id'] and expiry <= timestamp(receipt['access_expires_at'], 'access_expires_at'), 'token', 'token exceeds agreement access')
    return value


def validate_introspection_request(value):
    _bound(value, 'token resource_url method actions purposes request_id')
    _token(value['token'], 'token'); identifier(value['request_id'], 'request_id')
    reference(value['resource_url'], 'resource_url', value['origin'], query=True)
    need(isinstance(value['method'], str) and value['method'] in {'GET', 'HEAD'}, 'method', 'delivery methods only')
    policy.enum_array(value['actions'], policy.ACTIONS, 'actions'); policy.enum_array(value['purposes'], policy.PURPOSES, 'purposes')
    need('retrieve' in value['actions'], 'actions', 'retrieval action required for delivery')
    return value


def validate_introspection_response(value, request=None, receipt=None, token_response=None):
    _bound(value, 'request_id active permitted reason admission')
    identifier(value['request_id'], 'request_id')
    need(type(value['active']) is bool and type(value['permitted']) is bool, 'decision', 'strict booleans required')
    if request is not None:
        validate_introspection_request(request); _same_binding(value, request)
        need(value['request_id'] == request['request_id'], 'request_id', 'wrong introspection response')
    if not value['permitted']:
        need(value['admission'] is None, 'admission', 'denial cannot assert admission')
        expected = {'scope_mismatch'} if value['active'] else {'invalid_token', 'inactive_agreement'}
        need(isinstance(value['reason'], str) and value['reason'] in expected, 'reason', 'inconsistent denial reason')
        return value
    need(value['active'] and value['reason'] is None, 'decision', 'permission requires active token')
    a = value['admission']
    keys(a, 'decision_id agreement_id client_id principal_id resource_url method actions purposes expires_at checked_at status_version authority_revision authority_digest authority_checked_at policy_id policy_revision policy_digest'.split(), 'admission')
    for field in ('decision_id', 'agreement_id'): identifier(a[field], field)
    _token(a['client_id'], 'client_id'); reference(a['principal_id'], 'principal_id')
    reference(a['resource_url'], 'resource_url', value['origin'], query=True); reference(a['policy_id'], 'policy_id', value['origin'])
    need(isinstance(a['method'], str) and a['method'] in {'GET', 'HEAD'}, 'method', 'delivery methods only')
    policy.enum_array(a['actions'], policy.ACTIONS, 'actions'); policy.enum_array(a['purposes'], policy.PURPOSES, 'purposes')
    need('retrieve' in a['actions'], 'actions', 'retrieval action required')
    for field in ('status_version', 'authority_revision', 'policy_revision'): integer(a[field], field, 1)
    for field in ('authority_digest', 'policy_digest'): hash_value(a[field], field)
    checked = timestamp(a['checked_at'], 'checked_at'); authority_checked = timestamp(a['authority_checked_at'], 'authority_checked_at')
    expires = timestamp(a['expires_at'], 'expires_at')
    need(0 <= (checked-authority_checked).total_seconds() <= 5 and checked < expires, 'admission', 'stale or expired decision')
    if request is not None:
        for field in ('resource_url', 'method', 'actions', 'purposes'): need(a[field] == request[field], field, 'admission for another request')
    if receipt is not None:
        validate_receipt(receipt); _same_binding(value, receipt)
        for field in ('agreement_id', 'client_id', 'principal_id'): need(a[field] == receipt[field], field, 'admission principal/agreement mismatch')
        need(expires <= timestamp(receipt['access_expires_at'], 'access_expires_at'), 'expires_at', 'admission exceeds agreement')
    if token_response is not None:
        validate_token_response(token_response, receipt); _same_binding(value, token_response)
        need(a['agreement_id'] == token_response['agreement_id'] and expires <= timestamp(token_response['expires_at'], 'expires_at'), 'expires_at', 'admission exceeds token')
    return value


def validate_report(value, *, role=None, origin=None, reporter_id=None, admission=None):
    """Draft-3 event; source/HTTP semantics are explicit, never a version shim."""
    _bound(value, 'event_id event_type occurred_at reporter_id source resource_url policy_id policy_revision actions purposes agreement_id asset_ref operation http quantity unit related_events derived_from')
    identifier(value['event_id'], 'event_id'); timestamp(value['occurred_at'], 'occurred_at')
    if origin is not None: need(value['origin'] == origin, 'origin', 'origin mismatch')
    reference(value['resource_url'], 'resource_url', value['origin'], query=True); reference(value['reporter_id'], 'reporter_id')
    if reporter_id is not None: need(value['reporter_id'] == reporter_id, 'reporter_id', 'reporter mismatch')
    need(isinstance(value['source'], str) and isinstance(value['event_type'], str) and value['source'] in c.SOURCES and value['event_type'] in c.SOURCES[value['source']], 'source', 'unsupported source/event')
    if role is not None:
        mapping = {'agent':'client_reported', 'gateway':'origin_observed', 'service':'service_recorded'}
        need(role in mapping and mapping[role] == value['source'], 'source', 'authenticated role spoof')
    if value['policy_id'] is None:
        need(value['policy_revision'] is None and value['event_type'] in c.TRANSPORT, 'policy', 'policy required')
    else:
        reference(value['policy_id'], 'policy_id', value['origin']); integer(value['policy_revision'], 'policy_revision', 1)
    empty = value['event_type'] in c.TRANSPORT or value['event_type'] == 'policy.observed'
    policy.enum_array(value['actions'], policy.ACTIONS, 'actions', 0 if empty else 1)
    policy.enum_array(value['purposes'], policy.PURPOSES, 'purposes', 0 if empty else 1)
    need(bool(value['actions']) == bool(value['purposes']), 'intent', 'paired intent arrays required')
    if value['agreement_id'] is not None: identifier(value['agreement_id'], 'agreement_id')
    c.validate_asset_ref(value['asset_ref'], value['origin'])
    op = value['operation']
    if op is not None:
        keys(op, 'id started_at ended_at state'.split(), 'operation'); identifier(op['id'], 'operation.id')
        need(timestamp(op['started_at'], 'started_at') <= timestamp(value['occurred_at'], 'occurred_at'), 'operation', 'start after event')
        need(isinstance(op['state'], str) and op['state'] in {'in_progress','completed','failed'}, 'operation.state', 'unknown state')
        need(op['ended_at'] is None if op['state'] == 'in_progress' else op['ended_at'] == value['occurred_at'], 'ended_at', 'operation time mismatch')
    if value['event_type'] == 'use.reported': need(op is not None and len(value['actions']) == 1, 'operation', 'use report requires one action/operation')
    if value['event_type'] in c.TRANSPORT: c.validate_http(value['http'], value['event_type'], value['origin'])
    else: need(value['http'] is None, 'http', 'transport data on nontransport event')
    if value['quantity'] is None: need(value['unit'] is None, 'unit', 'paired quantity/unit required')
    else:
        integer(value['quantity'], 'quantity'); need(isinstance(value['unit'], str) and bool(re.fullmatch(r'[A-Za-z][A-Za-z0-9._:-]{0,63}', value['unit'])), 'unit', 'invalid unit')
    for field in ('related_events', 'derived_from'):
        array(value[field], field, 0, 64); seen = set()
        for item in value[field]:
            if field == 'related_events':
                keys(item, 'reporter_id event_id payload_digest'.split(), 'related_event')
                reference(item['reporter_id'], 'reporter_id'); identifier(item['event_id'], 'event_id'); hash_value(item['payload_digest'], 'payload_digest')
            else:
                need(item is not None, field, 'null asset reference'); c.validate_asset_ref(item)
            encoded = json_bytes(item); need(encoded not in seen, field, 'duplicate reference'); seen.add(encoded)
    if admission is not None:
        validate_introspection_response(admission)
        need(admission['permitted'] and value['event_type'] in c.TRANSPORT and value['source'] == 'origin_observed', 'admission', 'delivery binding requires allowed decision')
        _same_binding(value, admission); a = admission['admission']
        for field in ('agreement_id','resource_url','actions','purposes','policy_id','policy_revision'): need(value[field] == a[field], field, 'delivery differs from admission')
        need(value['http']['method'] == a['method'] and value['http']['delivery_id'] == admission['request_id'], 'http', 'wrong admitted delivery')
        need(timestamp(value['occurred_at'], 'occurred_at') >= timestamp(a['checked_at'], 'checked_at'), 'occurred_at', 'delivery before admission')
    return value


def validate_intake(value, *, report_validator=validate_report):
    _bound(value, 'record_id collector_id received_at assurance authenticated_reporter_id payload_digest payload_b64url reporter_jws authority', max_bytes=1048576)
    identifier(value['record_id'], 'record_id'); timestamp(value['received_at'], 'received_at')
    authority = value['authority']
    keys(authority, 'origin service_id revision url digest'.split(), 'authority')
    need(authority['origin'] == value['origin'], 'authority.origin', 'wrong publisher')
    reference(authority['service_id'], 'authority.service_id', value['origin'])
    integer(authority['revision'], 'authority.revision', 1)
    reference(authority['url'], 'authority.url'); hash_value(authority['digest'], 'authority.digest')
    need(value['collector_id'] == value['service_id'] == value['authority']['service_id'], 'collector_id', 'collector authority mismatch')
    raw = crypto.unb64u(value['payload_b64url']); need(0 < len(raw) <= 131072, 'payload', 'bounded report required')
    need(value['payload_digest'] == digest(raw), 'payload_digest', 'exact payload mismatch')
    # Explicit trusted adapter extension; the default remains the closed free
    # event contract. A wire field can never select an arbitrary validator.
    event = report_validator(load_json(raw)); _same_binding(value, event)
    need(isinstance(value['assurance'], str) and value['assurance'] in {'unverified','client_key_verified','origin_key_verified','service_key_verified'}, 'assurance', 'unknown assurance')
    if value['assurance'] == 'unverified':
        need(value['authenticated_reporter_id'] is None and value['reporter_jws'] is None and event['source'] == 'client_reported', 'assurance', 'unverified origin/service claim forbidden')
    else:
        need(value['authenticated_reporter_id'] == event['reporter_id'], 'reporter', 'authenticated reporter mismatch')
        expected = {'client_reported':'client_key_verified','origin_observed':'origin_key_verified','service_recorded':'service_key_verified'}
        need(value['assurance'] == expected[event['source']], 'assurance', 'wrong assurance for source')
        _, _, header, payload, _ = crypto.split_jws(value['reporter_jws'])
        need(header['typ'] == 'odexa-event+jws' and payload == raw, 'reporter_jws', 'signed payload/type mismatch; cryptographic verification remains separate')
    return value