"""Pure Arbeit operation library. Plans source-bound input; never sends it.

Only current scoped observations establish effects. A dispatched or uncertain
receipt prevents replay until an effect or explicit not-sent evidence arrives.
Definitions, inventory ownership and action counters are separate facts.
"""
from __future__ import annotations
from copy import deepcopy
from collections import Counter
from datetime import datetime, timezone
import hashlib
import json
import math

from .paths import root


_SPECS = {
    'context.identify': ('identity', ['current_character_identity', 'current_field_hud']),
    'quest.offer': ('offer', ['current_exact_offer', 'current_acceptance_conditions']),
    'move.target': ('movement', ['current_map_target', 'current_map_pose', 'arrival_evidence']),
    'dialogue.accept': ('dialogue', ['current_quest_npc', 'current_accept_form', 'accept_effect']),
    'quest.accepted': ('state', ['current_accepted_quest']),
    'inventory.hold': ('inventory', ['fresh_exact_bag_item', 'locked_and_reserved_quantity']),
    'objective.deliver': ('delivery', ['current_npc', 'current_submission_form', 'exact_item_quantity', 'objective_effect']),
    'dialogue.talk': ('dialogue', ['current_npc', 'current_dialogue_branch', 'objective_effect']),
    'dialogue.report': ('dialogue', ['current_npc', 'current_report_branch', 'objective_effect']),
    'objective.move': ('movement', ['current_map_target', 'current_map_pose', 'objective_effect']),
    'objective.inspect': ('interaction', ['typed_interaction_target', 'current_interaction_branch', 'objective_effect']),
    'life.craft': ('craft', ['exact_current_recipe', 'current_materials', 'current_craft_form', 'objective_effect']),
    'life.alter': ('alter', ['exact_current_recipe', 'current_facility_slots', 'current_materials', 'owned_work_receipt', 'objective_effect']),
    'objective.dungeon_clear': ('dungeon', ['typed_dungeon_identity', 'current_entry_conditions', 'same_instance_clear', 'objective_effect']),
    'objective.read_story': ('story', ['typed_story_identity', 'current_story_form', 'objective_effect']),
    'objective.use_quest_item': ('quest_item', ['typed_quest_fake_item_identity', 'current_exact_item_card', 'objective_effect']),
    'life.gather': ('gather', ['typed_gatherable_identity', 'current_unlock_conditions', 'current_native_life_state', 'objective_effect']),
    'objective.unresolved': ('unresolved', ['typed_objective_program', 'current_objective_branch', 'objective_effect']),
    'quest.step_complete': ('state', ['current_exact_step_completion']),
    'reward.claim': ('reward', ['current_exact_quest_reward', 'current_reward_form', 'completion_effect']),
    'quest.completed': ('state', ['current_exact_quest_completion']),
}


def operation_support(node):
    op = node.get('op') if isinstance(node, dict) else str(node)
    spec = _SPECS.get(op)
    if not spec:
        return dict(op=op, implemented=False, adapter_connected=False,
                    definition_missing=['unknown_operation'], live_test_ready=False)
    concrete = op in ('context.identify', 'inventory.hold', 'move.target', 'objective.move', 'life.gather', 'life.alter', 'life.craft')
    from .arbeit_coverage import PROFILES, RUNTIME_PROFILES
    profile=(PROFILES | RUNTIME_PROFILES)[op]
    conditional=profile[0]=='conditional_planner'
    detail = objective_requirements(node if isinstance(node,dict) else {'op':op})
    return dict(op=op, implemented=True, adapter_connected=concrete or conditional, concrete_adapter=concrete,
                conditional_adapter=conditional, family=spec[0],
                implementation='masukja2.arbeit_operations.plan_operation',
                input_adapter=profile[2],
                generic_proposal_bridge=True,
                observer_implemented=True,
                typed_state_consumer=op in ('quest.accepted', 'quest.step_complete', 'quest.completed'),
                official_reader_connected=op in ('inventory.hold','life.alter'),
                planner_implemented=concrete or conditional,
                current_provider_required=op not in ('context.identify', 'inventory.hold'),
                end_to_end_live_adapter_verified=False,
                implementation_scope=profile[1]+'; requires '+profile[4],
                required_evidence=list(spec[1]),
                definition_missing=detail.get('missing_definition', []),
                objective_diagnostic=detail,
                live_test_ready=False, game_test_performed=False)


def objective_requirements(node):
    """Separate missing structures per source objective; hints never actuate."""
    params=node.get('params',{})
    obj=params.get('objective',{})
    kinds={t.get('kind') for t in params.get('targets',obj.get('targets',[]))}
    op=node.get('op')
    if op=='objective.unresolved':
        text=obj.get('text','')
        if 'npc' in kinds:
            family='npc_exchange_or_dialogue'
            missing=['quest_dialogue_branch_id','quest_exchange_payload','current_npc_binding']
        elif 'gatherable' in kinds:
            family='quest_gatherable_interaction'
            missing=['typed_gatherable_sheet_id','quest_action_counter_semantics','current_interaction_target']
        elif 'fakeitem' in kinds:
            family='quest_local_item_or_story'
            missing=['quest_local_item_definition','typed_use_or_story_action','current_action_target']
        elif 'item' in kinds:
            family='item_acquisition_or_transform'
            missing=['typed_acquire_vs_transform_predicate','current_recipe_or_interaction_target','required_action_count_semantics']
        else:
            family='world_object_interaction'
            missing=['quest_actor_or_interaction_object_id','objective_location_and_activation_predicate','native_interaction_action']
        return dict(family_hint=family,text=text,executes_from_hint=False,missing_definition=missing)
    if op=='life.craft':
        return dict(family_hint='equipment_craft' if 'equipment' in kinds else 'consumable_craft',
                    executes_from_hint=False,missing_definition=[],
                    source_definition='data/arbeit-craft/objective-bindings.json; recipes.json',
                    current_evidence_required=['exact_current_recipe','current_recipe_materials_and_output','current_form_and_owned_effect'])
    if op=='objective.dungeon_clear':
        return dict(family_hint='quest_dungeon_clear',executes_from_hint=False,
                    missing_definition=['quest_battlefield_or_dungeon_id','entry_branch_and_instance_binding'])
    if op=='objective.inspect':
        return dict(family_hint='quest_interaction',executes_from_hint=False,
                    missing_definition=['typed_interaction_object_or_dialogue_branch','current_interaction_ui_form'])
    if op=='objective.read_story':
        return dict(family_hint='quest_story',executes_from_hint=False,
                    missing_definition=['quest_local_item_story_id','story_open_and_read_completion_predicate'])
    return {}


def operation_registry():
    return {op: operation_support(op) for op in _SPECS}


def _stamp(value):
    parsed = datetime.fromisoformat(str(value).replace('Z', '+00:00'))
    if parsed.tzinfo is None:
        raise ValueError('timezone_required')
    return parsed


def _fresh(evidence, snapshot):
    if not isinstance(evidence, dict):
        return False
    if evidence.get('fresh') is False or evidence.get('certainty') in ('unknown', 'uncertain', 'conflicting'):
        return False
    if not evidence.get('source', evidence.get('observation_source')):
        return False
    try:
        now = _stamp(snapshot.get('now', datetime.now(timezone.utc).isoformat()))
        captured = _stamp(evidence.get('observed_at', evidence.get('captured_at')))
        if not -.5 <= (now - captured).total_seconds() <= 5:
            return False
    except (ValueError, TypeError):
        return False
    for key in ('session_id', 'character_id'):
        if not snapshot.get(key) or evidence.get(key) != snapshot[key]:
            return False
    return True


def _node_key(node):
    return node.get('id') or ':'.join(str(node.get(k, '')) for k in ('quest_id', 'step_id', 'objective_index', 'op'))


def _scope(evidence, node, *, objective=False):
    keys = ['quest_id']
    if node.get('step_id') is not None:
        keys.append('step_id')
    if objective and node.get('objective_index') is not None:
        keys.append('objective_index')
    return all(evidence.get(k) is not None and str(evidence[k]) == str(node[k]) for k in keys)


def _waiting(node, reason, missing, receipt=None, **extra):
    return dict(status='waiting', reason=reason, wait_reason=reason,
                missing=list(missing), action=None, proposal=None,
                effect_evidence=[], next_receipt=deepcopy(receipt),
                operation=_node_key(node), sent_to_game=False, **extra)


def _complete(node, evidence, receipt=None):
    return dict(status='complete', reason='current_scoped_effect_confirmed', missing=[],
                action=None, proposal=None, effect_evidence=[deepcopy(evidence)],
                next_receipt=deepcopy(receipt), operation=_node_key(node), sent_to_game=False)


def _proofs(snapshot, key):
    value = snapshot.get(key, [])
    if isinstance(value, dict):
        return list(value.values())
    return value if isinstance(value, list) else []


def _objective_effect(node, snapshot, receipt=None):
    if node.get('objective_index') is None:
        return None
    matches = [p for p in _proofs(snapshot, 'objective_evidence')
               if _fresh(p, snapshot) and _scope(p, node, objective=True)]
    if len(matches) != 1:
        return None
    p = matches[0]
    # Count >= goal does not substitute for the client's completed flag.
    from .objective_binding import current_objective_occurrence
    return p if p.get('complete') is True and current_objective_occurrence(p,snapshot,node,receipt) else None


def _state_effect(node, snapshot):
    expected = {'quest.accepted': 'accepted', 'dialogue.accept': 'accepted',
                'quest.step_complete': 'step_complete', 'reward.claim': 'completed',
                'quest.completed': 'completed'}.get(node['op'])
    if not expected:
        return None
    matches = [p for p in _proofs(snapshot, 'state_proofs')
               if _fresh(p, snapshot) and _scope(p, node)
               and p.get('state') == expected and p.get('confirmed') is True
               and p.get('basis') not in ('absence', 'button_click', 'inventory_delta', 'synthetic')]
    return matches[0] if len(matches) == 1 else None


def _gather_idle_evidence(snapshot, receipt):
    if (receipt or {}).get('phase') not in ('gather.native_activity','gather.stop'):
        return []
    rows=[]
    for proof in (snapshot.get('native_life') or {}).get('idle_observations',[]):
        if not _fresh(proof,snapshot) or proof.get('idle') is not True or proof.get('route_id')!=(receipt.get('child_receipt') or {}).get('route_id'):
            continue
        try:
            if _stamp(proof['observed_at'])>_stamp(receipt['observed_at']):
                rows.append(proof)
        except (KeyError,TypeError,ValueError):
            continue
    return rows if len({p['observed_at'] for p in rows})>=2 else []


def _item(node):
    params = node.get('params', {})
    direct = params.get('item')
    if isinstance(direct, dict):
        return direct
    targets = params.get('targets', params.get('objective', {}).get('targets', []))
    items = [t for t in targets if t.get('kind') == 'item']
    if len(items) != 1:
        return None
    t = items[0]
    table = t.get('table_data', {})
    return dict(item_id=table.get('raw_id'), name=table.get('display_name'),
                category=t.get('category'), symbol=(t.get('arguments') or [None])[0],
                source_ref=t.get('source_ref'))


def supply_candidates(item):
    """Historical supply paths, not current eligibility or quantity proof."""
    if item and item.get('table_domain') not in (None,'ConsumableSheet'):
        return []
    path = root() / 'data/arbeit-operations.json'
    if not item or not path.exists():
        return []
    data = json.loads(path.read_text(encoding='utf-8'))
    return deepcopy(data.get('item_supply', {}).get(item.get('name', ''), []))


def _inventory(node, snapshot, receipt):
    params = node.get('params', {})
    item = _item(node)
    if 'material_requirement' in params:
        from .material_supply import validate_material_request
        current = [p for p in _proofs(snapshot, 'material_hold_requests')
            if validate_material_request(node,snapshot,p)]
        if len(current)!=1:
            return _waiting(node,'current_source_material_requirement_unavailable',
                ['validated_source_recipe_requirement_and_current_bag'],receipt)
    else:
        current = [p for p in _proofs(snapshot, 'inventory_hold_requests') if _fresh(p,snapshot)
        and _scope(p,node,objective=True) and p.get('node_id')==_node_key(node)
        and p.get('ready') is True and p.get('confirmed') is True
        and p.get('semantics')=='inventory.hold' and p.get('identity_binding')]
    if not current:
        current = [p for p in _proofs(snapshot, 'objective_evidence') if _fresh(p, snapshot)
               and _scope(p, node, objective=True) and p.get('semantics') == 'inventory.hold'
               and p.get('item_name') == (item or {}).get('name') and p.get('identity_binding')]
    if len(current) != 1:
        return _waiting(node, 'current_hold_semantics_unavailable',
                        ['fresh_exact_current_hold_objective', 'current_required_item_quantity'], receipt)
    quantity = current[0].get('required_quantity')
    if isinstance(current[0].get('item'),dict):
        item = deepcopy(current[0]['item'])
    if current[0].get('item_category'):
        item = dict(item, category=current[0]['item_category'])
    if not item or not item.get('name') or type(quantity) is not int or quantity <= 0:
        return _waiting(node, 'hold_item_or_quantity_unresolved', ['exact_item_identity', 'positive_hold_quantity'], receipt)
    if item.get('table_domain')=='QuestFakeItemSheet':
        if 'material_requirement' not in params:
            return _waiting(node,'quest_fake_item_native_goal_required',['same_current_typed_native_goal'],receipt)
        # Reconcile owned world input before stock can finish the material
        # child. Acquisition counts and usable inventory remain separate facts.
        from .quest_item_acquisition import plan_quest_item_acquisition
        return plan_quest_item_acquisition(node,snapshot,receipt)
    if current[0].get('inventory_count_completion_allowed') is False:
        count=current[0].get('observed_objective_count')
        if type(count) is not int or count<0:
            return _waiting(node,'current_typed_hold_count_unavailable',['native_current_hold_count'],receipt)
        if count>=quantity:
            return _waiting(node,'current_typed_hold_conditions_incomplete',['native_current_hold_IsAchieved'],receipt)
        return _waiting(node,'inventory_shortage',['current_supply_route_eligibility'],receipt,
            shortage=quantity-count,item=item,supply_candidates=supply_candidates(item),
            typed_hold_request=deepcopy(current[0]),consumable_inventory_does_not_prove_equipment=True)
    inv = snapshot.get('inventory', {})
    if not _fresh(inv, snapshot) or inv.get('source') not in ('get_items', 'official:get_items', 'official-cli:get_items'):
        return _waiting(node, 'current_inventory_unavailable', ['fresh_get_items'], receipt)
    if inv.get('complete') is not True:
        return _waiting(node, 'inventory_scope_incomplete', ['complete_bag_inventory'], receipt)
    rows = []
    for row in inv.get('items', []):
        # Official schemas without raw IDs use exact name/category only. Never
        # attach the catalog ID to the observed row as if the game returned it.
        if row.get('item_id') is not None and item.get('item_id') is not None:
            match = str(row['item_id']) == str(item['item_id'])
        else:
            match = row.get('name') == item['name']
        if match and (not item.get('category') or row.get('category') == item['category']):
            rows.append(row)
    if rows and len({r.get('category') for r in rows}) != 1:
        return _waiting(node, 'ambiguous_item_category', ['exact_item_category'], receipt)
    have = 0
    for row in rows:
        if row.get('scope', 'bag') != 'bag':
            continue
        n, locked, reserved = row.get('quantity'), row.get('locked_quantity'), row.get('reserved_quantity', 0)
        if any(type(x) is not int or x < 0 for x in (n, locked, reserved)) or locked + reserved > n:
            return _waiting(node, 'inventory_protection_unresolved', ['quantity', 'locked_quantity', 'reserved_quantity'], receipt)
        have += n - locked - reserved
    if have >= quantity:
        return _complete(node, dict(inv, matched_item=item, usable_quantity=have, required_quantity=quantity), receipt)
    return _waiting(node, 'inventory_shortage', ['current_supply_route_eligibility'], receipt,
                    shortage=quantity-have, item=item, supply_candidates=supply_candidates(item))


def _authored_action(node, snapshot, actions, evidence, phase, receipt=None, *, child=None, prior_effect=False, observed_at=None):
    context = {k:v for k,v in snapshot.get('input_context', {}).items()
               if k in ('source_revision','source','observed_at','session_id','character_id')}
    observed_at = observed_at or snapshot.get('observed_at')
    ownership=(snapshot.get('native_life') or {}).get('life_ownership')
    if phase.startswith('gather.') and ownership:
        child=dict(child or {},life_ownership=deepcopy(ownership))
    proposal = dict(status='ready', actions=actions, source_refs=evidence,
                    observed_at=observed_at, session_id=snapshot['session_id'],
                    current_binding={'node_id':_node_key(node), 'phase':phase},
                    expected_effect=phase, context=dict(context, observed_at=observed_at))
    if isinstance(child, dict) and child.get('owner'):
        proposal['next_receipt'] = deepcopy(child)
    saved = dict(node_id=_node_key(node), quest_id=node.get('quest_id'),
                 session_id=snapshot['session_id'], character_id=snapshot['character_id'],
                 observed_at=observed_at, dispatch_state='prepared', phase=phase,
                 child_receipt=deepcopy(child))
    current=snapshot.get('current_quest') or (snapshot.get('quest_state_read') or {}).get('current_quest') or {}
    if (receipt or {}).get('goal_scope'):
        saved['goal_scope']=deepcopy(receipt['goal_scope'])
    elif _fresh(current,snapshot) and current.get('confirmed') is True and _scope(current,node):
        saved['goal_scope']={k:current.get(k) for k in
            ('quest_id','instance_id','accept_time','step_id','step_index','process_started_at')}
    return dict(status='action', proposal=proposal, action=None, reason='source_planner_input',
                next_receipt=saved, effect_evidence=[], missing=[], sent_to_game=False,
                prior_effect_confirmed=prior_effect)


def _movement(node, snapshot, receipt):
    from .arbeit_movement import bind_movement_goal, bind_offer_movement_goal, plan_movement
    raw = snapshot.get('map') or {}
    current = dict(raw.get('snapshot', raw))
    current['runtime_descriptors'] = snapshot.get('runtime_descriptors', [])
    arrival_only=(receipt or {}).get('phase')=='movement.move_sent'
    if raw.get('ok') is False or (raw.get('planner_ready') is False and not arrival_only) or not current:
        return _waiting(node, 'current_map_provider_unavailable', raw.get('missing', ['current_map_observation']), receipt)
    if current.get('session_id') != snapshot.get('session_id') or (current.get('identity') or {}).get('character_id') != snapshot.get('character_id'):
        return _waiting(node, 'current_map_owner_changed', ['current_map_owner'], receipt)
    params = node.get('params', {})
    normalized=params.get('typed_operands') or {}
    if normalized.get('ready') and 'npc_group_tag' in normalized.get('operands',{}):
        from .group_npc_movement import plan_group_npc_movement
        planned=plan_group_npc_movement(node,snapshot,(receipt or {}).get('child_receipt'),
            now=_stamp(snapshot.get('now',datetime.now(timezone.utc).isoformat())))
        return _movement_result(node,snapshot,current,planned,receipt)
    if normalized.get('operation')=='objective.move' and node['op']=='objective.move':
        from .typed_region_movement import plan_typed_region_movement
        planned=plan_typed_region_movement(normalized,current,(receipt or {}).get('child_receipt'),
            now=_stamp(snapshot.get('now',datetime.now(timezone.utc).isoformat())))
        return _movement_result(node,snapshot,current,planned,receipt)
    if normalized.get('operation')=='objective.inspect' and node['op']=='move.target':
        from .world_target_observer import plan_world_target_movement
        planned=plan_world_target_movement(dict(node,op='objective.inspect'),snapshot,(receipt or {}).get('child_receipt'),
            now=_stamp(snapshot.get('now',datetime.now(timezone.utc).isoformat())))
        return _movement_result(node,snapshot,current,planned,receipt)
    targets = params.get('targets', [])
    offer_binding=None
    if params.get('target_binding') == 'current_offer.giver':
        offers = [r for r in _proofs(snapshot, 'offers') if _fresh(r, snapshot) and _scope(r, node)
                  and isinstance(r.get('giver'), dict)]
        if len(offers) != 1:
            return _waiting(node, 'current_offer_giver_unavailable', ['current_exact_offer.giver'], receipt)
        targets = [offers[0]['giver']]
        offer_binding=bind_offer_movement_goal(offers[0], current,
            now=_stamp(snapshot.get('now',datetime.now(timezone.utc).isoformat())))
    current_quest = (snapshot.get('quest_state_read') or {}).get('current_quest') or {}
    expected = None
    if _fresh(current_quest, snapshot) and _scope(current_quest, node):
        expected = {k:current_quest.get(k) for k in ('quest_id','instance_id','accept_time','step_id','step_index')}
        expected['objective_index'] = node.get('objective_index')
    candidates = [offer_binding] if offer_binding is not None else [bind_movement_goal(t, current_snapshot=current, expected_objective=expected) for t in targets]
    ready = [r for r in candidates if r.get('ready')]
    if len(ready) != 1:
        return _waiting(node, 'movement_destination_binding_unavailable',
                        sorted({m for r in candidates for m in r.get('missing', [])}) or ['unique_source_target'], receipt,
                        target_bindings=candidates)
    planned = plan_movement(current, ready[0]['goal'], (receipt or {}).get('child_receipt'),
                            now=_stamp(snapshot.get('now', datetime.now(timezone.utc).isoformat())))
    return _movement_result(node,snapshot,current,planned,receipt,ready[0])


def _movement_result(node,snapshot,current,planned,receipt,binding=None):
    binding=binding or {}
    if planned['status']=='transition':
        child=planned.get('next_receipt') or {}
        evidence=planned.get('evidence') or {}
        if (not receipt or child.get('phase')!='replan_ready' or
                (child.get('world_target_stage')!='replan' and child.get('group_npc_stage')!='replan')
                or child.get('resolved_navigation_receipt')!=receipt.get('child_receipt')
                or not evidence.get('original_goal_arrival')):
            return _waiting(node,'movement_replan_original_arrival_evidence_required',
                            ['owned_original_navigation_effect_before_replan'],receipt)
        successor=deepcopy(receipt)
        successor.update(phase='movement.replan_ready',dispatch_state='effect_confirmed',
            observed_at=current['captured_at'],child_receipt=deepcopy(child))
        return dict(status='transition',reason='original_movement_arrived_current_target_replan_required',
            next_receipt=successor,effect_evidence=[deepcopy(evidence)],proposal=None,
            not_objective_completion=True)
    if planned['status'] == 'arrived':
        if node['op'] == 'move.target':
            return _complete(node, dict(source=current.get('observation_source'), evidence=planned.get('evidence'),
                                       goal=binding.get('goal'), observed_at=current['captured_at']), receipt)
        return _waiting(node, 'arrival_observed_quest_effect_pending', ['current_objective_completion'], receipt)
    if planned['status'] == 'action':
        action=deepcopy(planned['action'])
        if action['kind'] in ('scroll','map_drag','map_long_press'):
            action['safe_rect']=list(current['safe_rect'])
        return _authored_action(node, snapshot, [action],
            binding.get('source_refs') or [planned.get('evidence')],
            'movement.'+planned['next_receipt']['phase'], receipt,
            child=planned['next_receipt'], prior_effect=bool(receipt), observed_at=current['captured_at'])
    from .wait_contract import effect_wait_contract
    contract=(effect_wait_contract(receipt,family='group_npc_navigation')
        if planned.get('wait_kind')=='owned_input_effect' and planned.get('wait_family')=='group_npc_navigation' else None)
    return _waiting(node, planned.get('wait_reason', 'movement_effect_pending'),
                    planned.get('missing', ['current_map_transition_effect']), receipt,wait_contract=contract)


def _native_gather(node, snapshot, receipt, candidates=None):
    """Current exact life binding -> historical selection tape -> owned link.

    Route identity is provided by a current provider and must belong to the
    source output candidates. We do not choose a potentially wrong byproduct
    route by item-name heuristics or count inventory gain as gather actions.
    """
    from .arbeit_movement import prepare_gather_tape, gather_nearby_candidate, movement_definition, MovementEvidenceUnavailable
    life = snapshot.get('native_life') or {}
    if not _fresh(life, snapshot) or life.get('confirmed') is not True:
        return _waiting(node, 'current_native_life_binding_unavailable', ['current_native_life'], receipt)
    route = life.get('route_id')
    if candidates is not None and route not in [c['route_id'] for c in candidates if c['kind']=='native_gather']:
        return _waiting(node, 'gather_output_route_mismatch', ['current_route_in_source_output_candidates'], receipt)
    if candidates is None and (life.get('node_id') != _node_key(node) or life.get('semantics') != 'life.gather'):
        return _waiting(node, 'gather_objective_semantics_unavailable', ['current_typed_gather_objective_binding'], receipt)
    if candidates is None:
        from .life_observer import bind_native_gather_objective
        binding=bind_native_gather_objective(node,snapshot,receipt)
        if (binding.get('ready') is not True or binding.get('native_route_id')!=route
                or life.get('objective_binding')!=binding):
            return _waiting(node,'current_typed_gather_objective_to_route_unavailable',
                binding.get('missing') or ['same_current_typed_gatherable_source_route'],receipt)
    width, height = life.get('viewport', [None, None])
    if life.get('profile_confirmed') is not True:
        if life.get('reconciliation_only') is True:
            return _waiting(node,'owned_gather_reconciliation_observing',
                ['two_current_idle_observations_and_current_field_layout'],receipt)
        return _waiting(node, 'gather_profile_unconfirmed', ['current_profile'], receipt)
    phase = (receipt or {}).get('phase')
    reconciliation=None
    if life.get('selection_reconciled') is True:
        proof=life.get('selection_reconciliation_evidence') or {}
        pending=snapshot.get('pending_input') or {}
        if (not _fresh(proof,snapshot) or proof.get('confirmed') is not True
                or proof.get('original_operation_id')!=pending.get('operation_id')
                or not pending.get('operation_id')
                or proof.get('goal_scope')!=(receipt or {}).get('goal_scope')
                or proof.get('input_epoch')!=(snapshot.get('input_owner') or {}).get('epoch')
                or not proof.get('current_typed') or not proof.get('source_refs')
                or proof.get('original_input_effect_inferred') is not False
                or _stamp(proof['observed_at'])<=_stamp(receipt['observed_at'])):
            return _waiting(node,'gather_selection_reconciliation_unconfirmed',
                ['current_typed_selection_or_empty_panels_same_goal_new_lease'],receipt)
        reconciliation=proof
    def audited(result):
        if reconciliation:
            result.setdefault('effect_evidence',[]).append(deepcopy(reconciliation))
            result['proposal'].setdefault('prior_effect_evidence',[]).append(deepcopy(reconciliation))
            result['proposal']['source_refs'].extend(deepcopy(reconciliation['source_refs']))
        return result
    try:
        refs = movement_definition()['sources']
        if candidates is None:refs=refs+deepcopy(binding['source_refs'])
        if (phase=='gather.selection' and reconciliation and life.get('checkpoint')=='field_idle'
                and life.get('selection_reconciliation_resume_ready') is True):
            tape=prepare_gather_tape(route,width,height,ui_percent=life['ui_percent'],safe_rect=life['safe_rect'])
            for action in tape['actions']:
                if action['kind']=='scroll':action['safe_rect']=list(life['safe_rect'])
            return audited(_authored_action(node,snapshot,tape['actions'],refs,'gather.selection',receipt,
                child={'route_id':route},prior_effect=True,observed_at=life['observed_at']))
        if phase=='gather.selection' and life.get('checkpoint')=='locked_detail':
            proof=life.get('restriction_evidence') or {}
            action=life.get('selection_reset_action') or {}
            if (not _fresh(proof,snapshot) or not proof.get('identity_binding')
                    or life.get('current_unlock_confirmed') is not False
                    or action.get('kind')!='key' or action.get('key')!='ESC'
                    or not life.get('next_route_id')):
                return _waiting(node,'locked_gather_detail_reset_unproven',['current_owned_locked_detail'],receipt)
            return _authored_action(node,snapshot,[action],life.get('source_refs') or refs,
                'gather.selection_reset',receipt,child={'route_id':route,'next_route_id':life['next_route_id']},
                prior_effect=True,observed_at=life['observed_at'])
        if phase=='gather.selection_reset' and life.get('checkpoint')=='field_idle' and life.get('selection_reset_confirmed') is True:
            if route!=(receipt.get('child_receipt') or {}).get('next_route_id'):
                return _waiting(node,'gather_reset_successor_route_changed',['same_selected_next_candidate'],receipt)
            tape=prepare_gather_tape(route,width,height,ui_percent=life['ui_percent'],safe_rect=life['safe_rect'])
            for action in tape['actions']:
                if action['kind']=='scroll':action['safe_rect']=list(life['safe_rect'])
            return audited(_authored_action(node,snapshot,tape['actions'],refs,'gather.selection',receipt,
                child={'route_id':route},prior_effect=True,observed_at=life['observed_at']))
        if phase in ('gather.native_activity','gather.stop'):
            idle=_gather_idle_evidence(snapshot,receipt)
            if idle:
                if (receipt.get('child_receipt') or {}).get('route_id')!=route:
                    return _waiting(node,'gather_resume_route_changed',['original_owned_gather_route'],receipt)
                hud=snapshot.get('field_hud') or {}
                context=snapshot.get('current_context') or {}
                layout=snapshot.get('layout') or {}
                # The real provider retains checkpoint=native_activity while
                # observing its own pending input. A fresh HUD/layout witness
                # independently establishes the field entry after bag recovery.
                if not _fresh(hud,snapshot) or hud.get('confirmed') is not True:
                    return _waiting(node,'gather_resume_field_hud_unconfirmed',['current_field_hud_after_native_end'],receipt)
                if (context.get('ok') is not True or context.get('field_hud') is not True or
                        context.get('profile',{}).get('visual_confirmed') is not True or
                        context.get('profile',{}).get('percent')!=life.get('ui_percent') or
                        layout.get('observed_at')!=hud.get('observed_at') or
                        layout.get('viewport')!=life.get('viewport') or
                        layout.get('safe_rect')!=life.get('safe_rect') or
                        layout.get('canvas_scale')!=life.get('canvas_scale')):
                    return _waiting(node,'gather_resume_current_layout_unconfirmed',
                                    ['current_source_layout_matches_owned_profile'],receipt)
                tape=prepare_gather_tape(route,width,height,ui_percent=life['ui_percent'],safe_rect=life['safe_rect'])
                for action in tape['actions']:
                    if action['kind']=='scroll':action['safe_rect']=list(life['safe_rect'])
                result=_authored_action(node,snapshot,tape['actions'],refs,'gather.selection',receipt,
                    child={'route_id':route},prior_effect=True,observed_at=hud['observed_at'])
                result['reason']='owned_gather_ended_resume_remaining_goal'
                result['effect_evidence']=deepcopy(idle)
                result['proposal']['prior_effect_evidence']=deepcopy(idle)
                result['proposal']['current_binding']['resume_basis']='same_route_two_distinct_idle_and_current_field_layout'
                return result
        if not receipt and life.get('checkpoint') == 'field_idle':
            tape = prepare_gather_tape(route, width, height, ui_percent=life.get('ui_percent'),safe_rect=life['safe_rect'])
            for action in tape['actions']:
                if action['kind']=='scroll':action['safe_rect']=list(life['safe_rect'])
            return _authored_action(node, snapshot, tape['actions'], refs, 'gather.selection',
                child={'route_id':route}, observed_at=life['observed_at'])
        if phase == 'gather.selection' and life.get('checkpoint') == 'selected_detail':
            if life.get('current_unlock_confirmed') is not True:
                return _waiting(node, 'gather_current_unlock_unconfirmed', ['current_gatherable_unlock'], receipt)
            if (receipt.get('child_receipt') or {}).get('route_id') != route:
                return _waiting(node, 'owned_gather_selection_changed', ['same_owned_gather_route'], receipt)
            action = gather_nearby_candidate(route, width=width, height=height,
                ui_percent=life['ui_percent'], canvas_scale=life['canvas_scale'],
                safe_rect=life['safe_rect'], detail=life['detail'])
            return audited(_authored_action(node, snapshot, [action], refs, 'gather.native_activity', receipt,
                child={'route_id':route}, prior_effect=True, observed_at=life['observed_at']))
    except (MovementEvidenceUnavailable, KeyError, TypeError, ValueError) as exc:
        return _waiting(node, 'native_gather_geometry_unavailable', [str(exc)], receipt)
    return _waiting(node, 'native_gather_effect_pending',
                    ['current_inventory_goal', 'owned_native_stop_or_two_distinct_idle_observations'], receipt)


def _gather_stop(node,snapshot,receipt):
    life=snapshot.get('native_life') or {}
    proposed=life.get('stop_proposal') or {}
    phase=(receipt or {}).get('phase')
    retry=phase=='gather.stop' and life.get('stop_retry_no_effect') is True and bool(life.get('stop_retry_evidence'))
    if (phase!='gather.native_activity' and not retry) or not _fresh(life,snapshot) or not _fresh(proposed,snapshot):
        return None
    pending=snapshot.get('pending_input') or {}
    native=(receipt.get('child_receipt') or {}).get('native_pending') if retry else pending
    if (proposed.get('kind')!='official_stop' or proposed.get('method')!='stop_action'
            or proposed.get('params') is not None or not proposed.get('source_refs')
            or not proposed.get('goal_evidence')
            or proposed.get('owned_operation_id')!=(native or {}).get('operation_id')):
        return None
    child=dict(receipt.get('child_receipt') or {},native_pending=deepcopy(native))
    if not child.get('goal_scope') and receipt.get('goal_scope'):
        child['goal_scope']=deepcopy(receipt['goal_scope'])
    result=_authored_action(node,snapshot,[proposed],proposed['source_refs'],'gather.stop',receipt,
        child=child,prior_effect=True,observed_at=proposed['observed_at'])
    result['reason']='owned_gather_goal_reached_stop_then_observe_idle'
    return result


def _quest_autoplay(node, snapshot, receipt, *, offer=False):
    from .quest_autoplay import plan_quest_autoplay, plan_offer_autoplay
    if receipt and not str(receipt.get('phase','')).startswith('quest_autoplay.'):
        return None
    current = snapshot.get('map') or {}
    raw = current.get('snapshot', current)
    if current.get('ok') is False or current.get('planner_ready') is False:
        return None
    rows = [r for r in _proofs(snapshot, 'offers' if offer else 'action_requests')
        if _fresh(r, snapshot) and _scope(r, node, objective=not offer)]
    if len(rows) != 1:
        return None
    planner = plan_offer_autoplay if offer else plan_quest_autoplay
    result = planner(raw, rows[0], (receipt or {}).get('child_receipt'),
        now=_stamp(snapshot.get('now', datetime.now(timezone.utc).isoformat())))
    if result['status'] != 'action':
        direct_npc=(node['op'] in ('move.target','objective.move','dialogue.talk','dialogue.report','objective.deliver','objective.inspect')
            and ((offer and node.get('params',{}).get('target_binding')=='current_offer.giver') or
                 len([t for t in node.get('params',{}).get('targets',[]) if t.get('kind')=='npc'])==1))
        native_unavailable=result.get('wait_reason') in (
            'current_same_quest_instance_StartAutoPlay_marker_missing','current_quest_autoplay_CTA_disabled',
            'quest_autoplay_evidence_unavailable:native_auto_play_explicitly_not_supported')
        normalized=node.get('params',{}).get('typed_operands') or {}
        direct_typed=(normalized.get('ready') is True and (normalized.get('operation') in ('objective.move','objective.inspect')
            or 'npc_group_tag' in normalized.get('operands',{})))
        if (direct_npc or direct_typed) and native_unavailable:
            if receipt is None:
                return None
            # Only map-open has an observed, harmless successor here. A
            # selected marker or native autoplay intent can never be discarded.
            if (result.get('wait_reason')=='current_same_quest_instance_StartAutoPlay_marker_missing'
                    and receipt.get('phase')=='quest_autoplay.quest_map_open_sent'
                    and raw.get('map_visible') is True and raw.get('map_ready') is True
                    and _stamp(raw['captured_at'])>_stamp(receipt['observed_at'])):
                successor=deepcopy(receipt)
                successor.update(phase='movement.approach_ready',dispatch_state='effect_confirmed',
                    observed_at=raw['captured_at'],child_receipt=None)
                evidence=dict(result.get('evidence') or {},source=raw['observation_source'],
                    observed_at=raw['captured_at'],map_open_observed=True,native_autoplay_dispatched=False)
                return dict(status='transition',reason='current_open_map_has_no_native_quest_marker_use_direct_target',
                    next_receipt=successor,effect_evidence=[evidence],proposal=None,not_objective_completion=True)
        return _waiting(node, result.get('wait_reason','quest_native_autoplay_effect_pending'),
            result.get('missing', ['current_native_quest_callback_or_effect']), receipt,
            native_action_scope='quest_instance', current_request=rows[0])
    action = deepcopy(result['action'])
    if action['kind'] in ('scroll','map_drag','map_long_press'):
        action['safe_rect'] = raw['safe_rect']
    child = dict(result['next_receipt'], action_scope='quest_instance',
        requested_objective={k:rows[0].get(k) for k in ('quest_id','instance_id','accept_time','step_id','step_index','objective_index','process_started_at')})
    return _authored_action(node, snapshot, [action], [result['evidence']],
        'quest_autoplay.'+child['phase'], receipt, child=child,
        prior_effect=bool(receipt), observed_at=raw['captured_at'])


def _map_card_accept(node,snapshot,receipt):
    """Preserve the exact waiting instance across native map-card autoplay."""
    from .map_quest_accept import plan_map_quest_accept
    phase=(receipt or {}).get('phase') or ''
    if receipt and not phase.startswith('map_quest_card.'):
        return None
    if not receipt and snapshot.get('action_proposals',{}).get(_node_key(node)):
        return None
    if not receipt and not snapshot.get('map_quest_card_frame'):
        return None
    raw=snapshot.get('map') or {}
    raw=raw.get('snapshot',raw)
    planned=plan_map_quest_accept(node,dict(snapshot,map=raw),snapshot.get('map_quest_card_frame') or {},
        (receipt or {}).get('child_receipt'),now=_stamp(snapshot.get('now',datetime.now(timezone.utc).isoformat())))
    if planned['status']=='accepted':
        return _complete(node,planned['evidence'],receipt)
    if planned['status']!='action':
        from .wait_contract import effect_wait_contract
        contract=(effect_wait_contract(receipt,family='map_quest_card')
            if planned.get('wait_kind')=='owned_input_effect' and planned.get('wait_family')=='map_quest_card' else None)
        return _waiting(node,planned.get('wait_reason','current_map_card_effect_required'),
            ['current_exact_waiting_instance_card_or_native_acceptance_effect'],receipt,
            direct_accept_supported=False,map_card_evidence=planned.get('evidence'),wait_contract=contract)
    child=planned['next_receipt']
    observed=(snapshot.get('current_map_accept') or {}).get('observed_at') or raw.get('captured_at')
    return _authored_action(node,snapshot,[planned['action']],[planned['evidence']],
        'map_quest_card.'+child['stage'],receipt,child=child,prior_effect=bool(receipt),observed_at=observed)


def _objective_travel(node, snapshot, receipt):
    """Same objective owns approach/return; arrival never completes its task."""
    if node['op'] not in ('dialogue.talk', 'dialogue.report', 'objective.deliver', 'objective.inspect'):
        return None
    phase = (receipt or {}).get('phase', '') or ''
    proposed=snapshot.get('action_proposals', {}).get(_node_key(node))
    if proposed and not phase.startswith('movement.'):
        return None
    if receipt and not phase.startswith('movement.') and phase != 'objective.arrived':
        return None  # Never replace an unresolved interaction with navigation.
    if phase == 'objective.arrived':
        return _waiting(node, 'destination_arrived_current_interaction_required',
                        ['current_owned_NPC_or_quest_interaction_control'], receipt)
    targets = [t for t in node.get('params', {}).get('targets', []) if t.get('kind') == 'npc']
    normalized=node.get('params',{}).get('typed_operands') or {}
    world_target=(node['op']=='objective.inspect' and normalized.get('ready') is True
                  and normalized.get('operation')=='objective.inspect')
    group_target=(normalized.get('ready') is True and 'npc_group_tag' in normalized.get('operands',{}))
    if len(targets) != 1 and not world_target and not group_target:
        return None
    if phase.startswith('movement.'):
        from .arbeit_movement import bind_movement_goal
        target_binding=bind_movement_goal(targets[0]) if not world_target and not group_target else {}
        npc_id=(target_binding.get('goal') or {}).get('npc_id') if target_binding.get('ready') else None
        scope=receipt.get('goal_scope') or {}
        child=receipt.get('child_receipt') or {}
        def same_target(p):
            if group_target:
                return (child.get('group_npc_key') is not None and p.get('entity_key')==child['group_npc_key']
                    and p.get('npc_id')==child.get('group_npc_id'))
            if world_target:
                return (child.get('world_target_key') is not None and p.get('entity_key')==child['world_target_key']
                    and p.get('instant_prop_id')==child.get('world_target_definition_id'))
            return type(p.get('npc_id')) is int and type(npc_id) is int and (p['npc_id']&0xffffffff)==(npc_id&0xffffffff)
        proofs=[p for p in snapshot.get('interaction_ready',[]) if _fresh(p,snapshot)
            and p.get('kind')=='owned_objective_interaction' and p.get('confirmed') is True
            and p.get('ownership_verified') is True and p.get('node_id')==node['id']
            and all(scope.get(k) is not None and p.get(k)==scope[k]
                    for k in ('quest_id','instance_id','accept_time','step_id','process_started_at'))
            and _scope(p,node,objective=True)
            and same_target(p)
            and _stamp(p['observed_at'])>_stamp(receipt['observed_at'])]
        if len(proofs)==1:
            successor=deepcopy(receipt)
            successor.update(phase='objective.arrived',dispatch_state='effect_confirmed',
                observed_at=proofs[0]['observed_at'])
            return dict(status='transition',reason='current_owned_interaction_resolved_target_approach',
                next_receipt=successor,effect_evidence=proofs,proposal=None,
                geometric_arrival_verified=False,not_objective_completion=True)
    approach = dict(node, op='move.target', params=dict(node.get('params', {}), targets=targets))
    result = _movement(approach, snapshot, receipt)
    if result['status'] != 'complete':
        result['parent_operation'] = node['op']
        return result
    successor = dict(receipt or {}, node_id=node['id'], quest_id=node['quest_id'],
        session_id=snapshot['session_id'], character_id=snapshot['character_id'],
        observed_at=snapshot['observed_at'], phase='objective.arrived', dispatch_state='effect_confirmed')
    return dict(status='transition', reason='objective_destination_arrived', next_receipt=successor,
        effect_evidence=result['effect_evidence'], proposal=None, not_objective_completion=True)


def processing_target(stage, state):
    """Source equations for six facilities/all recipes, with current parents.

    Ported from mini27 source_list_layouts recipe_card and source_life_routes
    MMML889 registration / MMML911 facility travel. No catalog-order fallback,
    portrait-anchor assumption, or inferred flowing-content height is allowed.
    """
    from .vendor.ui_layout import Rect
    if state.get('input_target_ready') is False:
        raise ValueError('current_processing_input_target_not_ready:'+','.join(state.get('input_missing',[])))
    target = state.get('current_target')
    if state.get('native_target') is True and isinstance(target, dict):
        if state.get('confirmed') is not True or state.get('source_bound') is not True:
            raise ValueError('current_native_processing_source_required')
        if target.get('kind') == 'key' and stage == 'field' and target.get('key') == 'K':
            return deepcopy(target)
        if state.get('settled') is not True or target.get('kind') not in ('click','scroll','pointer_probe','pointer_click','pointer_scroll'):
            raise ValueError('current_native_processing_target_unsettled')
        bounds = target.get('source_rect')
        if not bounds or len(bounds) != 4:
            raise ValueError('current_native_processing_bounds_required')
        x,y,right,bottom = bounds
        rect = Rect(x,y,right-x,bottom-y)
        safe = Rect(*state['safe_rect'])
        if rect.width<=0 or rect.height<=0 or not safe.contains(rect):
            raise ValueError('current_native_processing_target_clipped')
        return deepcopy(target)
    s = state.get('canvas_scale')
    if type(s) not in (int, float) or not math.isfinite(s) or s <= 0:
        raise ValueError('current_canvas_scale')
    safe = Rect(*state['safe_rect'])
    width, height = state['viewport']
    if not Rect(0, 0, width, height).contains(safe):
        raise ValueError('current_safe_rect')
    if state.get('source_bound') is not True or state.get('settled') is not True:
        raise ValueError('current_source_parent_and_settled_state')
    landscape = state.get('orientation') == 'landscape'
    if state.get('orientation') not in ('landscape', 'portrait'):
        raise ValueError('current_orientation')
    if stage == 'recipe_list':
        if state.get('family') != 'FacilityUI911/RecipeItem906':
            raise ValueError('current_recipe_list_family')
        ids = state['ordered_ids']
        if not ids or len(ids) != len(set(ids)) or state['recipe_id'] not in ids:
            raise ValueError('current_exact_recipe_array')
        parent = Rect(*state['parent_rect'])
        columns = state['current_flow_columns']
        if type(columns) is not int or columns < 1:
            raise ValueError('observed_flow_columns')
        expected = {math.floor((parent.width-padding*s+8*s)/(164*s+8*s)) for padding in (0,40)}
        if columns not in expected:
            raise ValueError('current_columns_disagree_with_source')
        sx, sy = state['scroll_x'], state['scroll_y']
        row, col = divmod(ids.index(state['recipe_id']), columns)
        rect = Rect(parent.left+20*s+col*172*s-sx, parent.top+row*244*s-sy,164*s,232*s)
        if not parent.contains(rect):
            raise ValueError('recipe_card_clipped_requires_current_scroll')
        role = 'processing.recipe.'+state['recipe_id']
    elif stage == 'recipe_detail':
        if state.get('family') != 'AlteringPanelUI889' or state.get('local_detail') is not True or state.get('register_enabled') is not True:
            raise ValueError('current_local_altering_panel_889')
        left = safe.right-752*s if landscape else safe.center[0]-360*s
        # An actual content bottom is mandatory, even for content that fits.
        bottom, sy = state['content_bottom'], state['scroll_y']
        if sy < 0 or not all(math.isfinite(v) for v in (bottom,sy)):
            raise ValueError('current_content_bottom_and_scroll')
        rect = Rect(left+120*s,bottom-80*s-sy,552*s,80*s)
        role = 'processing.register'
    elif stage == 'remote_facility':
        if state.get('family') != 'FacilityUI911/FacilityMainButton908' or state.get('remote') is not True or state.get('move_visible') is not True:
            raise ValueError('current_remote_facility_911_908')
        if state.get('source_default_flow_origin_confirmed') is not True or type(state.get('upgrade_visible')) is not bool:
            raise ValueError('current_source_flow_origin_and_upgrade_branch')
        left = safe.left+(96 if landscape else 32)*s
        flow_width = 490*s if landscape else safe.width-302*s
        minw,maxw,gap = 180*s,280*s,12*s
        shrink, w = 1.0,minw
        if state['upgrade_visible']:
            if flow_width<=0:
                raise ValueError('nonpositive_button_flow_width')
            shrink=min(1.0,flow_width/(2*maxw+gap))
            adjusted=left+(maxw+gap)*min(1.0,flow_width/(maxw+gap+minw))
            w=left+min(flow_width,2*minw+gap)-adjusted
            left=adjusted
        earliest=safe.top+((120 if landscape else 158)+110)*s
        latest=safe.top+((120 if landscape else 158)+124)*s
        rect=Rect(left,latest,w,earliest+48*s*shrink-latest)
        role='processing.facility.travel'
    elif stage == 'collect':
        if not landscape or state.get('family')!='FacilityUI911/ProgressRing912':
            raise ValueError('current_landscape_facility_queue_family')
        if any(state.get(k) is not True for k in ('facility_ready','collect_enabled','full_safe_area')):
            raise ValueError('current_owned_local_collect_enabled')
        if safe.left!=0 or safe.right!=width or safe.bottom!=height:
            raise ValueError('collect_source_requires_unchanged_screen_left_right_bottom')
        count=state['queue_count']
        if type(count) is not int or not 1<=count<=7:
            raise ValueError('current_generated_slot_capacity_not_work_count')
        rows=math.ceil(count/4)
        holder_height=(46+16+rows*100+(rows-1)*10)*s
        rect=Rect(96*s,height-45*s-holder_height,120*s,46*s)
        role='processing.collect_all'
    else:
        raise ValueError('processing_stage_requires_additional_source_binding:'+str(stage))
    if rect.width <= 0 or rect.height <= 0 or not safe.contains(rect):
        raise ValueError('processing_target_not_fully_visible')
    return {'kind':'click','role':role,'point':list(rect.center),'source_rect':list(rect.bounds),
            'geometry_source':'mini27 source_life_routes.py/source_list_layouts.py',
            'game_tested':False}


def _processing_probe_resume(node, snapshot, receipt):
    from .ui_effects import current_ui_effect
    current=snapshot.get('native_processing') or {}
    child=receipt.get('child_receipt') or {}
    proof=current_ui_effect(node,snapshot,receipt)
    if (proof is None or not _fresh(current,snapshot) or current.get('confirmed') is not True
            or current.get('stage')!=child.get('processing_stage')
            or current.get('recipe_id')!=child.get('recipe_id')
            or (current.get('current_target') or {}).get('kind') not in ('pointer_click','pointer_scroll')):
        return _waiting(node,'processing_pointer_probe_effect_pending',
            ['fresh_same_processing_target_final_pointer_hit'],receipt)
    # The probe has no production effect. Re-evaluate the genuine pre-probe
    # intent and every material/work predicate using the new observation.
    before=deepcopy(child.get('before_probe_pending') or {})
    original=before.get('planner_receipt')
    if original and original.get('phase')=='alter.pointer_probe':
        return _waiting(node,'processing_pointer_probe_original_intent_invalid',[],receipt)
    resumed=dict(snapshot,pending_input=before or None)
    result=plan_operation(node,resumed,original)
    reason=result.get('reason') or ''
    if result['status']=='waiting' and (reason in ('processing_material_shortage','processing_material_or_slot_unconfirmed')
            or reason.startswith('processing_material_goal_reserved_shortage:')):
        return dict(status='restore_pending',reason='processing_probe_resolved_restore_material_navigation',
            original_pending=before or None,effect_evidence=[proof],prior_effect_confirmed=True,
            proposal=None,missing=[],sent_to_game=False,blocked_successor=deepcopy(result))
    if result['status'] in ('action','transition','complete'):
        result['prior_effect_confirmed']=True
        result.setdefault('effect_evidence',[]).append(proof)
        if result.get('proposal'):
            result['proposal'].setdefault('prior_effect_evidence',[]).append(proof)
    return result


def _native_alter(node, snapshot, receipt):
    from .processing_completion import prepare_processing_collection, plan_processing_completion
    if (receipt or {}).get('phase') in ('alter.collect','alter.cleanup'):
        return plan_processing_completion(node,snapshot,receipt)
    current=snapshot.get('native_processing') or {}
    if not _fresh(current,snapshot) or current.get('confirmed') is not True:
        return _waiting(node,'current_processing_binding_unavailable',['current_native_processing'],receipt)
    item=_item(node)
    candidates=[r for r in supply_candidates(item) if r['kind']=='native_alter' and r['route_id']==current.get('recipe_id')]
    if len(candidates)!=1 or not (current.get('exact_recipe_confirmed') is True or current.get('recipe_intent_bound') is True):
        return _waiting(node,'current_recipe_identity_unconfirmed',['current_exact_recipe_source_join'],receipt)
    recipe=candidates[0]
    if recipe['group']!=current.get('facility_id'):
        return _waiting(node,'current_recipe_facility_mismatch',['same_recipe_facility'],receipt)
    stage=current.get('stage')
    prior=(receipt or {}).get('phase')
    # Scene change is proof for navigation, not proof of production. Once a
    # registration may have been sent only its work receipt can advance it.
    if prior=='alter.recipe_detail':
        baseline=(receipt.get('child_receipt') or {}).get('registration_baseline') or {}
        proof=current.get('works_observation') or {}
        before=Counter({(r['name'],r['facility']):r['count'] for r in baseline.get('multiset',[])})
        after=Counter({(r['name'],r['facility']):r['count'] for r in current.get('work_multiset',[])})
        exact=(baseline.get('recipe_name'),baseline.get('facility_name'))
        unchanged_owner=(current.get('input_epoch') is not None and current.get('input_epoch')==(receipt.get('child_receipt') or {}).get('input_epoch'))
        newer=_fresh(proof,snapshot) and _stamp(proof['observed_at'])>_stamp(receipt['observed_at'])
        if newer and baseline.get('same_recipe_count')==0 and unchanged_owner and after==before:
            return _waiting(node,'owned_processing_work_receipt_pending',
                            ['current_work_row_after_owned_registration'],receipt)
        if not newer or baseline.get('same_recipe_count')!=0 or not unchanged_owner or after-before!=Counter({exact:1}) or before-after:
            return _waiting(node,'owned_processing_work_receipt_required',
                            ['exact_work_multiset_increased_by_one','no_existing_same_recipe_work','same_input_owner_epoch'],receipt)
        own=[r for r in current.get('current_recipe_works',[]) if (r['name'],r['facility'])==exact]
        if len(own)!=1 or own[0].get('complete') is not True:
            return _waiting(node,'owned_processing_work_in_progress',['current_work_completion'],receipt)
        completed=[r for r in current.get('current_facility_works',[]) if r['complete']]
        if stage!='collect' or len(completed)!=1 or completed[0]!=own[0]:
            return _waiting(node,'owned_processing_collection_scope_unconfirmed',
                            ['collect_all_contains_only_owned_completed_work','source_collect_target'],receipt)
    elif stage=='collect':
        return _waiting(node,'owned_processing_registration_receipt_required',
                        ['owned_registration_before_collect_all'],receipt)
    allowed={'alter.field':('hub_tabs','facility_list'), 'alter.hub_tabs':('facility_list',),
        'alter.facility_list':('recipe_list','recipe_scroll','remote_facility','collect'),
        'alter.recipe_list':('recipe_detail',),'alter.remote_facility':('recipe_list','recipe_scroll','collect'),
        'alter.recipe_scroll':('recipe_scroll','recipe_list'),
        'alter.collection_received':('recipe_list','recipe_scroll','facility_list')}
    if receipt and prior!='alter.recipe_detail' and stage not in allowed.get(prior, ()):
        return _waiting(node,'processing_navigation_effect_pending',['current_owned_processing_successor'],receipt)
    if prior=='alter.recipe_scroll':
        previous=(receipt.get('child_receipt') or {}).get('scroll_observation') or {}
        current_scroll=current.get('scroll_observation') or {}
        if not _fresh(current_scroll,snapshot) or not previous.get('digest') or current_scroll.get('digest')==previous['digest']:
            return _waiting(node,'processing_scroll_effect_pending',['current_scroll_content_changed'],receipt)
    if stage=='recipe_detail':
        if current.get('exact_recipe_confirmed') is not True:
            return _waiting(node,'current_recipe_detail_identity_unconfirmed',['current_exact_typed_recipe_detail'],receipt)
        if current.get('materials_confirmed') is not True or current.get('free_slot_confirmed') is not True:
            return _waiting(node,'processing_material_or_slot_unconfirmed',
                            ['fresh_exact_unlocked_unreserved_materials','current_free_work_slot'],receipt)
        if current.get('uses_storage') is not False:
            return _waiting(node,'processing_storage_cost_branch_unbound',['explicit_bag_only_material_binding'],receipt)
        baseline=current.get('registration_baseline') or {}
        if not _fresh(baseline,snapshot) or not isinstance(baseline.get('multiset'),list) or baseline.get('same_recipe_count')!=0 or current.get('input_epoch') is None:
            return _waiting(node,'processing_work_baseline_unavailable',
                            ['current_work_multiset_before_registration','no_existing_same_recipe_work','current_input_epoch'],receipt)
        requirements=current.get('current_materials')
        if not isinstance(requirements,list) or requirements!=recipe['recipe']['materials']:
            return _waiting(node,'current_recipe_materials_changed',['current_recipe_material_definition_agreement'],receipt)
        inventory=snapshot.get('inventory') or {}
        if not _fresh(inventory,snapshot) or inventory.get('source')!='get_items' or inventory.get('complete') is not True:
            return _waiting(node,'processing_material_inventory_unavailable',['fresh_get_items_complete_bag'],receipt)
        for requirement in requirements:
            rows=[r for r in inventory['items'] if r.get('name')==requirement['name'] and r.get('scope','bag')=='bag']
            if len({r.get('category') for r in rows})!=1:
                return _waiting(node,'processing_material_identity_unresolved',['exact_category:'+requirement['name']],receipt)
            available=0
            for row in rows:
                values=[row.get('quantity'),row.get('locked_quantity'),row.get('reserved_quantity',0)]
                if any(type(v) is not int or v<0 for v in values) or values[1]+values[2]>values[0]:
                    return _waiting(node,'processing_material_protection_unknown',['lock_and_reserve:'+requirement['name']],receipt)
                available+=values[0]-values[1]-values[2]
            if available<requirement['required']:
                return _waiting(node,'processing_material_shortage',['unlocked_unreserved_material:'+requirement['name']],receipt)
            if snapshot.get('goal_reservations') or snapshot.get('goal_reservation_gaps'):
                from .processing_observer import processing_recipe_bindings
                from .goal_reservations import goal_commitment_for_item
                definitions=[r for r in processing_recipe_bindings()['bindings'] if r['route_id']==current['recipe_id']]
                materials=[m for r in definitions for m in r['current']['materials']
                    if m['name']==requirement['name'] and m['required']==requirement['required']]
                if len(materials)!=1:
                    return _waiting(node,'processing_material_identity_unresolved',['current_source_material_raw_ID'],receipt)
                try:
                    commitment=goal_commitment_for_item(snapshot,
                        dict(name=requirement['name'],item_id=materials[0]['raw_id'],table_domain='ConsumableSheet'),
                        exclude_node_id=node['id'])
                except ValueError as exc:
                    return _waiting(node,'processing_material_protection_unknown',[str(exc)],receipt)
                if available-commitment['required_quantity']<requirement['required']:
                    return _waiting(node,'processing_material_goal_reserved_shortage:'+requirement['name'],
                        ['current_other_goal_material_commitment'],receipt)
    try:
        action=processing_target(stage,current)
    except (KeyError,TypeError,ValueError) as exc:
        return _waiting(node,'processing_geometry_unavailable',[str(exc)],receipt)
    geometry_sources=json.loads((root()/'data/arbeit-operations.json').read_text(encoding='utf-8'))['processing_geometry_sources']
    child={'recipe_id':current['recipe_id'],'registration_baseline':current.get('registration_baseline'),
           'scroll_observation':current.get('scroll_observation'),
           'input_epoch':current.get('input_epoch'),'work_identity_kind':'multiset_not_stable_id'}
    if action.get('kind')=='pointer_probe':
        expected=(current.get('pointer_effect') or {}).get('expected_effect')
        if not isinstance(expected,dict) or expected.get('kind')!='ui_state':
            return _waiting(node,'processing_pointer_effect_contract_unavailable',[],receipt)
        child.update(processing_stage=stage,before_probe_pending=deepcopy(snapshot.get('pending_input')))
        result=_authored_action(node,snapshot,[action],recipe['source_refs']+geometry_sources+current.get('recipe_source_refs',[]),
            'alter.pointer_probe',receipt,child=child,prior_effect=bool(receipt),observed_at=current['observed_at'])
        result['next_receipt']['expected_effect']=deepcopy(expected)
        result['proposal']['expected_effect']=deepcopy(expected)
        return result
    if stage=='collect':
        prepared=prepare_processing_collection(node,snapshot,receipt)
        if prepared['status']!='ready':
            return _waiting(node,prepared['reason'],prepared.get('missing',[]),receipt)
        child['collection_baseline']=prepared['collection_baseline']
    return _authored_action(node,snapshot,[action],recipe['source_refs']+geometry_sources+current.get('recipe_source_refs',[]),'alter.'+stage,receipt,
                            child=child,prior_effect=bool(receipt),observed_at=current['observed_at'])


def _proposal(node, snapshot, receipt):
    key = _node_key(node)
    proposed = snapshot.get('action_proposals', {}).get(key)
    if not isinstance(proposed, dict) or proposed.get('status') not in ('ready', 'action'):
        return None
    context = proposed.get('context', {})
    evidence = dict(context, source=proposed.get('observation_source') or context.get('source') or context.get('source_revision'))
    if not _fresh(evidence, snapshot) or not proposed.get('source_refs', proposed.get('evidence')):
        return _waiting(node, 'action_proposal_evidence_unavailable', ['fresh_source_bound_action_proposal'], receipt)
    if not proposed.get('current_binding') or not proposed.get('expected_effect'):
        return _waiting(node, 'action_proposal_binding_unavailable', ['current_binding', 'expected_effect'], receipt)
    binding = proposed['current_binding']
    if isinstance(binding, dict) and binding.get('node_id', key) != key:
        return _waiting(node, 'action_proposal_node_mismatch', ['exact_node_binding'], receipt)
    if not any(proposed.get(k) for k in ('action', 'actions', 'target_id')):
        return _waiting(node, 'action_proposal_has_no_input', ['source_derived_input'], receipt)
    intermediate = None
    if receipt and receipt.get('dispatch_state') not in ('not_sent', 'effect_confirmed'):
        from .ui_effects import current_ui_effect
        child=receipt.get('child_receipt') or {}
        scope=child.get('requested_objective') or {}
        if child.get('action_scope')=='quest_instance':
            proofs=[p for p in snapshot.get('interaction_ready',[]) if _fresh(p,snapshot)
                and _scope(p,node,objective=True) and p.get('node_id')==key
                and p.get('kind')=='owned_objective_interaction' and p.get('confirmed') is True
                and p.get('instance_id')==scope.get('instance_id') and p.get('accept_time')==scope.get('accept_time')
                and _stamp(p['observed_at'])>_stamp(receipt['observed_at'])]
            if len(proofs)==1:intermediate=proofs[0]
        if intermediate is None and child.get('action_scope')=='quest_instance':
            from .story_native_handoff import current_story_adoption
            intermediate=current_story_adoption(node,snapshot,receipt,proposed)
        if intermediate is None:
            intermediate = current_ui_effect(node, snapshot, receipt)
        if intermediate is None:
            return _waiting(node, 'pending_input_effect_unconfirmed', ['fresh_owned_intermediate_UI_effect'], receipt)
    action_key = hashlib.sha256(json.dumps({k:proposed.get(k) for k in ('action','actions','target_id','current_binding')},
        sort_keys=True, default=str, ensure_ascii=False).encode()).hexdigest()
    if intermediate and receipt.get('action_fingerprint') == action_key:
        return _waiting(node, 'observed_effect_has_no_successor_action', ['distinct_current_successor_binding'], receipt)
    fingerprint = hashlib.sha256(json.dumps(proposed, sort_keys=True, default=str, ensure_ascii=False).encode()).hexdigest()
    saved = dict(node_id=key, quest_id=node.get('quest_id'),
                 session_id=snapshot['session_id'], character_id=snapshot['character_id'],
                 proposal_digest=fingerprint, observed_at=context['observed_at'],
                 expected_effect=deepcopy(proposed['expected_effect']), dispatch_state='prepared', action_fingerprint=action_key,
                 phase=(binding.get('phase') if isinstance(binding,dict) else None) or
                     ((proposed.get('expected_effect') or {}).get('phase') if isinstance(proposed.get('expected_effect'),dict) else None) or
                     'ui.'+str(proposed.get('target_id') or node['op']))
    if proposed.get('request_binding'):
        saved['request_binding']=deepcopy(proposed['request_binding'])
    current=snapshot.get('current_quest') or (snapshot.get('quest_state_read') or {}).get('current_quest') or {}
    if _fresh(current,snapshot) and current.get('confirmed') is True and _scope(current,node):
        saved['goal_scope']={k:current.get(k) for k in
            ('quest_id','instance_id','accept_time','step_id','step_index','process_started_at')}
    return dict(status='action', action=deepcopy(proposed.get('action')), proposal=deepcopy(proposed),
                reason='source_bound_input_proposed', missing=[], effect_evidence=[],
                next_receipt=saved, operation=key, sent_to_game=False,
                prior_effect_confirmed=bool(intermediate or (receipt and receipt.get('dispatch_state')=='effect_confirmed')),
                intermediate_effect_evidence=[intermediate] if intermediate else [])


def plan_operation(node, snapshot, receipt=None):
    # Reconcile a sent direct selection before generic state completion or a
    # changed current-step binding can replace the original occurrence.
    if (receipt or {}).get('request_binding'):
        from .direct_dialogue_request import reconcile_direct_dialogue_request
        from .wait_contract import effect_wait_contract
        pending=snapshot.get('pending_input') or {}
        if (any(receipt.get(k)!=snapshot.get(k) for k in ('session_id','character_id'))
                or receipt.get('node_id')!=_node_key(node)
                or (pending.get('planner_receipt') or {}).get('request_binding')!=receipt['request_binding']):
            return _waiting(node,'direct_dialogue_pending_receipt_owner_changed',['original_owned_request_receipt'],receipt)
        reconciled=reconcile_direct_dialogue_request(node,snapshot,pending)
        if reconciled['status']=='effect_observed':
            result=_complete(node,reconciled['effect_evidence'],receipt)
            result['direct_request_reconciliation']=reconciled
            if reconciled['effect_evidence'].get('state')=='step_complete':
                result.update(individual_objective_success_verified=False,not_objective_success=True,
                    completion_basis='original native quest step terminated after direct request')
            return result
        if reconciled['status']=='probe_only':
            result=_proposal(node,snapshot,receipt)
            if result is not None:return result
            from .direct_dialogue_input_guard import request_identity
            from .quest_ui_actions import _digest
            progress=snapshot.get('direct_dialogue_probe_progress') or (snapshot.get('current_ui') or {}).get('direct_dialogue_probe_progress') or {}
            try:
                rows=(pending.get('input_receipt') or {}).get('receipts') or []
                valid=(progress.get('kind')=='direct_dialogue_probe_cache_pending'
                    and progress.get('confirmed') is True and _fresh(progress,snapshot)
                    and progress.get('input_effect_confirmed') is False
                    and progress.get('quest_effect_owner_verified') is False
                    and progress.get('request_identity')==request_identity(receipt['request_binding'])
                    and progress.get('source_revision')==receipt['request_binding']['source_revision']
                    and progress.get('receipt_fingerprint')==_digest(pending['planner_receipt'])
                    and progress.get('operation_id')==pending['operation_id']
                    and len(rows)==1 and progress.get('probe_operation_id')==rows[0].get('operation_id')
                    and progress.get('input_epoch')==(snapshot.get('input_owner') or {}).get('epoch')
                    and all(progress.get(k)==receipt.get(k) for k in ('node_id','quest_id','session_id','character_id'))
                    and progress.get('process_started_at')==snapshot.get('process_started_at'))
            except (KeyError,TypeError,ValueError):valid=False
            if valid:
                return _waiting(node,'direct_dialogue_owned_probe_cache_pending',['later_owned_pointer_cache'],receipt,
                    wait_contract=effect_wait_contract(receipt,family='direct_dialogue_request'),
                    intermediate_observation=deepcopy(progress))
        result=_waiting(node,reconciled['reason'],['later_exact_typed_goal_effect'],receipt,
            direct_request_reconciliation=reconciled)
        if reconciled['status']=='pending' and reconciled['reason']=='direct_dialogue_request_awaiting_exact_typed_goal_effect':
            result['wait_contract']=effect_wait_contract(receipt,family='direct_dialogue_request')
        return result
    from .operation_binding import bind_operation, same_operation
    current=bind_operation(node,snapshot,receipt)
    previous=(receipt or {}).get('operation_binding')
    if current['status']=='invalid':
        return _waiting(node,'current_typed_operation_binding_invalid',current['missing'],receipt)
    if previous and (current['status'] not in ('bound','reconciling_terminated_craft','reconciling_processing_cleanup','reconciling_processing_termination','reconciling_terminated_gather') or not same_operation(previous,current['binding'])):
        return _waiting(node,'pending_typed_operation_changed_or_unobserved',
            ['same_current_type_operands_and_occurrence'],receipt)
    result=_plan_operation(current['node'],snapshot,receipt)
    if current['binding']:
        result['operation_binding']=deepcopy(current['binding'])
        if result.get('status')=='action' and isinstance(result.get('next_receipt'),dict):
            result['next_receipt']=dict(result['next_receipt'],operation_binding=deepcopy(current['binding']))
    return result


def _plan_operation(node, snapshot, receipt=None):
    """Return complete, source input proposal, or a named evidence requirement.

    snapshot.state_proofs rows: quest_id, step_id when relevant, state in
    accepted/step_complete/completed, confirmed=True, basis, source, observed_at,
    session_id, character_id. Neither absent HUD rows nor click receipts prove
    completion. snapshot.identity and field_hud use the same provenance fields
    and confirmed=True. action_proposals are keyed by node.id.
    """
    op = node.get('op')
    if op not in _SPECS:
        return _waiting(node, 'unknown_operation', ['registered_operation'], receipt)
    if _node_key(node) in snapshot.get('conflicting_action_nodes',[]):
        return _waiting(node,'conflicting_current_source_proposals',
            ['unique_current_action_for_same_goal'],receipt)
    if receipt and (receipt.get('node_id') != _node_key(node) or
                    receipt.get('session_id') != snapshot.get('session_id') or
                    receipt.get('character_id') != snapshot.get('character_id')):
        return _waiting(node, 'pending_receipt_owner_changed', ['original_receipt_owner_reconciliation'], receipt)
    acceptance_move=op=='move.target' and node.get('params',{}).get('role')=='acceptance'
    acceptance_phase=str((receipt or {}).get('phase',''))
    if acceptance_move and receipt and (acceptance_phase.startswith(('map_quest_card.','acceptance_movement.'))
            or acceptance_phase=='quest_autoplay.quest_map_open_sent'):
        from .acceptance_movement import plan_acceptance_movement
        alternative=plan_acceptance_movement(node,snapshot,receipt)
        if alternative is not None:return alternative
    if op=='dialogue.accept' and acceptance_phase.startswith('map_quest_card.'):
        return _map_card_accept(node,snapshot,receipt)
    if node.get('operation_reconciliation')=='owned_processing_termination':
        phase=(receipt or {}).get('phase')
        if phase in ('alter.collect','alter.collection_received','alter.cleanup'):
            from .processing_completion import plan_processing_completion
            return plan_processing_completion(node,snapshot,receipt)
        if phase=='alter.pointer_probe' and (receipt.get('child_receipt') or {}).get('processing_stage')=='collect':
            return _processing_probe_resume(node,snapshot,receipt)
        if phase=='alter.recipe_detail' and (snapshot.get('native_processing') or {}).get('stage')=='collect':
            return _native_alter(node,snapshot,receipt)
        return _waiting(node,'terminated_processing_owned_work_cleanup_pending',
                        ['owned_completed_work_before_collection'],receipt)
    life=snapshot.get('native_life') or {}
    if life.get('cleanup_only') is True:
        from .arbeit_crafting import _step_termination
        child=(receipt or {}).get('child_receipt') or {}
        scope=child.get('goal_scope') or (receipt or {}).get('goal_scope') or {}
        termination=_step_termination(node,snapshot,dict(child,goal_scope=scope))
        evidence=life.get('termination_evidence') or {}
        if (op not in ('life.gather','inventory.hold') or (receipt or {}).get('phase') not in ('gather.native_activity','gather.stop')
                or not _fresh(life,snapshot) or not termination or not _fresh(evidence,snapshot)
                or evidence.get('semantics')!='owned_native_step_termination'
                or any(evidence.get(k)!=termination.get(k) for k in
                       ('quest_id','instance_id','accept_time','step_id','process_started_at','source','observed_at'))):
            return _waiting(node,'owned_gather_termination_cleanup_evidence_required',
                            ['same_native_step_termination_and_owned_activity'],receipt)
        idle=_gather_idle_evidence(snapshot,receipt)
        if idle:
            result=_complete(node,termination,receipt)
            result.update(completion_basis='native step terminated after owned activity ended',
                          individual_objective_success_verified=False)
            result['effect_evidence'].extend(idle)
            return result
        stop=_gather_stop(node,snapshot,receipt)
        if stop is not None:
            stop['reason']='owned_gather_step_terminated_stop_then_observe_idle'
            return stop
        return _waiting(node,'gather_goal_reached_owned_activity_not_ended',
                        ['owned_native_stop_or_two_distinct_idle_observations'],receipt)
    effect = _objective_effect(node, snapshot, receipt) or _state_effect(node, snapshot)
    processing_phase=(receipt or {}).get('phase') or ''
    if op in ('inventory.hold','life.craft') and processing_phase.startswith('craft.'):
        from .supply_selection import plan_supply_cleanup
        cleanup=plan_supply_cleanup(node,snapshot,receipt,effect=effect)
        if cleanup is not None:return cleanup
    if processing_phase=='alter.pointer_probe':
        return _processing_probe_resume(node,snapshot,receipt)
    if processing_phase in ('alter.collect','alter.cleanup') or (effect and processing_phase=='alter.collection_received'):
        from .processing_completion import plan_processing_completion
        return plan_processing_completion(node,snapshot,receipt,effect)
    if effect and processing_phase=='alter.recipe_detail':
        return _native_alter(node,snapshot,receipt)
    if effect and processing_phase.startswith('alter.'):
        from .processing_completion import plan_processing_navigation_cleanup
        return plan_processing_navigation_cleanup(node,snapshot,receipt,effect)
    if effect and (receipt or {}).get('phase') in ('gather.native_activity','gather.stop'):
        if not _gather_idle_evidence(snapshot,receipt):
            stop=_gather_stop(node,snapshot,receipt)
            if stop is not None:return stop
            return _waiting(node, 'gather_goal_reached_owned_activity_not_ended',
                            ['owned_native_stop_or_two_distinct_idle_observations'], receipt)
    if effect:
        if processing_phase.startswith('craft.'):
            from .arbeit_crafting import craft_completion_boundary
            return craft_completion_boundary(node,snapshot,receipt,effect)
        return _complete(node, effect, receipt)
    if op=='move.target' and node.get('params',{}).get('role')=='acceptance':
        scope=((receipt or {}).get('child_receipt') or {}).get('requested_objective') or {}
        opened=[p for p in snapshot.get('interaction_ready',[]) if _fresh(p,snapshot)
            and _scope(p,node) and p.get('node_id')==node['id']
            and p.get('kind')=='owned_accept_dialogue' and p.get('confirmed') is True
            and p.get('instance_id') is not None and p.get('npc_id') is not None
            and (not scope or p['instance_id']==scope.get('instance_id'))]
        if len(opened)==1:
            result=_complete(node,opened[0],receipt)
            result['reason']='acceptance_dialogue_opened_not_quest_accepted'
            return result
    if op == 'context.identify':
        identity, hud = snapshot.get('identity', {}), snapshot.get('field_hud', {})
        if _fresh(identity, snapshot) and identity.get('confirmed') is True and _fresh(hud, snapshot) and hud.get('confirmed') is True:
            return _complete(node, dict(identity=identity, field_hud=hud), receipt)
        return _waiting(node, 'current_character_or_hud_unconfirmed', _SPECS[op][1], receipt)
    if op == 'quest.offer':
        candidates = [r for r in _proofs(snapshot, 'offers') if _fresh(r, snapshot) and _scope(r, node)
                      and r.get('available_candidate') is True and r.get('eligibility_verified') is True
                      and r.get('conditions_met') is True and r.get('giver_source')
                      and isinstance(r.get('giver'),dict) and r['giver'].get('npc_id') is not None]
        if len(candidates) == 1:
            return _complete(node, candidates[0], receipt)
    if op == 'inventory.hold':
        result = _inventory(node, snapshot, receipt)
        if result['status']=='complete' and processing_phase in ('alter.collect','alter.cleanup','alter.collection_received'):
            from .processing_completion import plan_processing_completion
            return plan_processing_completion(node,snapshot,receipt,result['effect_evidence'][0])
        if result['status']=='complete' and processing_phase=='alter.recipe_detail':
            return _native_alter(node,snapshot,receipt)
        if result['status']=='complete' and processing_phase.startswith('alter.'):
            from .processing_completion import plan_processing_navigation_cleanup
            return plan_processing_navigation_cleanup(node,snapshot,receipt,result['effect_evidence'][0])
        if result['status']=='complete' and processing_phase.startswith('craft.'):
            from .arbeit_crafting import craft_completion_boundary
            return craft_completion_boundary(node,snapshot,receipt,result['effect_evidence'][0])
        if result['status'] == 'complete' and (receipt or {}).get('phase') in ('gather.native_activity','gather.stop'):
            idle=_gather_idle_evidence(snapshot,receipt)
            if idle:
                result['effect_evidence'].extend(idle)
                return result
            stop=_gather_stop(node,snapshot,receipt)
            if stop is not None:return stop
            return _waiting(node, 'gather_inventory_reached_owned_activity_not_ended',
                            ['owned_native_stop_or_two_distinct_idle_observations'], receipt)
        if result['status'] == 'complete' or result['reason'] != 'inventory_shortage':
            return result
        proposed = _proposal(node, snapshot, receipt)
        if proposed:
            proposed['supply'] = {k:result[k] for k in ('shortage', 'item', 'supply_candidates')}
            return proposed
        if snapshot.get('native_processing') and (not receipt or str(receipt.get('phase','')).startswith('alter.')):
            processing = snapshot['native_processing']
            if processing.get('confirmed') is True:
                return _native_alter(node, snapshot, receipt)
        if snapshot.get('crafting') and (not receipt or str(receipt.get('phase','')).startswith('craft.')):
            if snapshot['crafting'].get('confirmed') is True:
                from .arbeit_crafting import plan_crafting
                return plan_crafting(node, snapshot, receipt)
        if snapshot.get('native_life'):
            return _native_gather(node, snapshot, receipt, result['supply_candidates'])
        return result
    # Current typed requests are actual client autoplay programs. Their scope
    # is the quest instance, not an invented objective-specific command.
    native_life=snapshot.get('native_life') or {}
    if (op=='life.gather' and native_life.get('confirmed') is True
            and (native_life.get('objective_binding') or {}).get('ready') is True
            and (not receipt or processing_phase.startswith('gather.'))):
        return _native_gather(node,snapshot,receipt)
    if not snapshot.get('action_proposals', {}).get(_node_key(node)):
        if op == 'move.target' and node.get('params', {}).get('role') == 'acceptance':
            native = _quest_autoplay(node, snapshot, receipt, offer=True)
            if receipt is None:
                from .acceptance_movement import NATIVE_UNAVAILABLE,plan_acceptance_movement
                if native is None or (native.get('status')=='waiting' and native.get('reason') in NATIVE_UNAVAILABLE):
                    alternative=plan_acceptance_movement(node,snapshot)
                    if alternative is not None:return alternative
            if native is not None:
                return native
        elif node.get('objective_index') is not None and op not in ('life.alter','life.craft'):
            native = _quest_autoplay(node, snapshot, receipt)
            if native is not None:
                return native
    if op in ('move.target', 'objective.move') and snapshot.get('map'):
        return _movement(node, snapshot, receipt)
    if op == 'life.gather' and snapshot.get('native_life'):
        return _native_gather(node, snapshot, receipt)
    if op == 'life.alter' and snapshot.get('native_processing'):
        return _native_alter(node, snapshot, receipt)
    if op == 'life.craft' and snapshot.get('crafting'):
        from .arbeit_crafting import plan_crafting
        return plan_crafting(node, snapshot, receipt)
    if op=='dialogue.accept':
        card=_map_card_accept(node,snapshot,receipt)
        if card is not None:return card
    travel = _objective_travel(node, snapshot, receipt)
    if travel is not None:
        return travel
    proposed = _proposal(node, snapshot, receipt)
    if proposed:
        return proposed
    if receipt and receipt.get('dispatch_state') != 'not_sent':
        return _waiting(node, 'pending_input_effect_unconfirmed', ['fresh_effect_or_explicit_not_sent'], receipt)
    specific = objective_requirements(node).get('missing_definition', [])
    descriptors = [d for d in snapshot.get('runtime_descriptors', [])
                   if _fresh(d, snapshot) and _scope(d, node, objective=True)]
    return _waiting(node, 'current_operation_evidence_unavailable', list(_SPECS[op][1])+specific, receipt,
                    support=operation_support(node), current_runtime_descriptors=deepcopy(descriptors),
                    descriptor_dispatch_implemented=False if descriptors else None)
