"""Structured, receipt-owned observation waits; diagnostic text is not policy.

Only supported producers opt into owned_input_effect. Missing evidence,
unsupported behavior and user intervention keep their existing stop boundary.
A wait contract grants observation only, never input replay or completion.
"""
from copy import deepcopy
import hashlib
import json

SCHEMA = 'masukja.wait.v1'
OWNER = ('node_id', 'quest_id', 'session_id', 'character_id', 'phase')
VIEW_PHASES = {'open_sent', 'world_sent', 'region_sent', 'pan_sent', 'zoom_sent'}
POINT_PHASES = {'select_sent', 'move_sent', 'pan_sent', 'zoom_sent'}


def _fingerprint(receipt):
    stable = {k: v for k, v in receipt.items() if k not in ('dispatch_state', 'input_receipt')}
    return hashlib.sha256(json.dumps(stable, sort_keys=True, ensure_ascii=False,
        separators=(',', ':'), allow_nan=False).encode()).hexdigest()


def _family_owned(receipt, family):
    child = receipt.get('child_receipt') or {}
    phase = receipt.get('phase', '')
    if any(receipt.get(k) is None for k in OWNER) or not receipt.get('observed_at'):
        return False
    if family == 'group_npc_navigation':
        stage = child.get('group_npc_stage')
        allowed = VIEW_PHASES if stage == 'view' else POINT_PHASES if stage == 'point' else set()
        return bool(child.get('group_npc_owner') and child.get('group_npc_key')
            and child.get('phase') in allowed and phase == 'movement.' + child['phase'])
    if family == 'quest_item_acquisition':
        inner = child.get('inner_receipt') or {}
        return bool(phase == 'quest_item.acquire' and child.get('acquisition_binding')
            and (child.get('material_identity') or {}).get('node_id') == receipt['node_id']
            and inner.get('phase') and inner['phase'] != 'quest_item.acquire'
            and all(inner.get(k) == receipt[k] for k in ('node_id', 'session_id', 'character_id')))
    if family == 'direct_dialogue_request':
        request=receipt.get('request_binding') or {}
        return bool(phase in ('direct_dialogue.request_sent','direct_dialogue.request_probe')
            and request.get('current_dialogue_input_owned') is True
            and request.get('quest_effect_owner_verified') is False
            and request.get('transaction_id') and request.get('before_scope')
            and request.get('process_started_at') and request.get('source_revision')
            and all(request.get(k)==receipt[k] for k in ('node_id','quest_id','session_id','character_id')))
    if family == 'map_quest_card':
        scope = child.get('offer_scope') or {}
        stages = {'open_sent', 'category_sent', 'unfold_sent', 'scroll_sent', 'autoplay_sent',
            'category_probe_sent', 'unfold_probe_sent', 'scroll_probe_sent', 'autoplay_probe_sent'}
        return bool(child.get('flow') == 'map_quest_card.native_autoplay'
            and child.get('stage') in stages and phase == 'map_quest_card.' + child['stage']
            and type(scope.get('instance_id')) is int and scope['instance_id'] > 0
            and scope.get('process_started_at')
            and all(scope.get(k) == receipt[k] for k in ('quest_id', 'session_id', 'character_id')))
    return False


def effect_wait_contract(receipt, *, family):
    if not isinstance(receipt, dict) or not _family_owned(receipt, family):
        return None
    return dict(schema=SCHEMA, kind='owned_input_effect', family=family,
        owner={k: deepcopy(receipt[k]) for k in OWNER}, receipt_sha256=_fingerprint(receipt))


def can_observe_owned_effect(contract, pending):
    """Reject stale/replaced/unattempted/dry contracts before scheduling a read."""
    try:
        if (not isinstance(contract, dict) or contract.get('schema') != SCHEMA
                or contract.get('kind') != 'owned_input_effect' or not pending.get('operation_id')
                or pending.get('status') not in ('sent_unconfirmed', 'unknown', 'intent_persisted')):
            return False
        sent = pending.get('input_receipt') or {}
        receipt = pending.get('planner_receipt') or {}
        if (sent.get('status') in ('dry_run', 'not_sent') or sent.get('attempted') is False
                or receipt.get('dispatch_state') == 'not_sent'
                or receipt.get('node_id') != pending.get('node_id')):
            return False
        expected = effect_wait_contract(receipt, family=contract.get('family'))
        return expected is not None and contract == expected
    except (KeyError, TypeError, ValueError, AttributeError):
        return False
