"""Current native-life observations; import/calling without opt-in is inert.

The current UI is read through the shared bounded map/UI transport. Pixels and
receipts are supplied by the caller. No capture, input, game invocation, heap
scan, process write, helper launch or paid connector action belongs here.
"""
from __future__ import annotations
from mini_masukja.native_type_reference import type_definition_index

from copy import deepcopy
from datetime import datetime, timezone
from functools import lru_cache
from pathlib import Path
import json
import math
import hashlib
from .supply_selection import SupplySelector,collect_supply_candidates,validated_material_hold_request

from .paths import root
from .ui_layout_observer import CurrentUILayoutSession
from .vendor.map_memory import MapReadPending
from .vendor.map_character_context import _class


@lru_cache(maxsize=1)
def declarations():
    return json.loads((root()/'data/arbeit-life-observer/declarations.json').read_text('utf-8'))


@lru_cache(maxsize=1)
def selection_bindings():
    return json.loads((root()/'data/r3-life-selection/route-bindings.json').read_text('utf-8'))


def bind_native_gather_objective(node, snapshot, receipt=None):
    """Join one *current* typed target to a source route, never to its output name.

    This is a life-adapter intent, separate from the client's quest autoplay
    request and from native IsAchieved completion evidence. No raw GoalCount,
    table display name, inventory amount or saved route can produce success.
    """
    from .objective_binding import current_objective_occurrence
    from .quest_program import load_schema
    from .arbeit_operations import _fresh
    result = dict(ready=False, confirmed=False, missing=[], quest_objective=True,
                  semantics='life.gather', node_id=node.get('id'))
    def gap(reason):
        return dict(result, missing=['current_typed_gather_objective_to_native_route_binding:'+reason])
    if node.get('op') != 'life.gather':
        return gap('operation_not_life_gather')
    state = snapshot.get('quest_state_read') or {}
    current = state.get('current_quest') or snapshot.get('current_quest') or {}
    candidates = [p for p in snapshot.get('objective_programs', []) if isinstance(p, dict)
        and all(p.get(k) == node.get(k) for k in ('quest_id', 'step_id', 'objective_index'))]
    if len(candidates) != 1:
        return gap('unique_current_owned_program_required')
    program = candidates[0]
    if (program.get('owner_verified') is not True
            or program.get('source') != 'current-root-typed-quest-collections'
            or not current_objective_occurrence(program, snapshot, node, receipt)):
        return gap('fresh_same_accepted_occurrence_required')
    schema = load_schema()
    if any((program.get('source_version') or {}).get(k) != schema[k]
           for k in ('game_assembly_sha256', 'metadata_sha256')):
        return gap('current_program_source_revision_changed')
    data = program.get('data') or {}
    name = 'GatheringSuccessQuestObjectiveData'
    if data.get('type') != 'MM.QDBL.CodeGenerated.'+name:
        return gap('unsupported_objective_type_or_group_domain')
    spec = next(s for s in schema['classes'] if s['name'] == name)
    getter = next(g for g in spec['getters'] if g['name'] == 'get_Type')
    disc = data.get('discriminant') or {}
    if (data.get('available') is not True or data.get('missing')
            or data.get('source') != 'current_owned_typed_quest_program'
            or disc.get('verified') is not True or disc.get('value') != getter['constant']
            or disc.get('basis') != 'pinned_get_Type_constant_bytes'):
        return gap('current_objective_discriminant_unverified')
    target = (data.get('values') or {}).get('GatherableName') or {}
    if (target.get('interpreted') is not True or target.get('domain_mapping_verified') is not True
            or target.get('declared_type') != 'Shared.Symbols.GatherableSheetId'
            or target.get('table_domain') != 'GatherableSheet' or target.get('member') != 'Id'
            or type(target.get('value')) is not int or target['value'] == 0):
        return gap('exact_GatherableSheetId_domain_required')
    definitions = selection_bindings()
    if definitions['metadata_sha256'] != schema['metadata_sha256']:
        return gap('source_route_metadata_revision_changed')
    matches = [r for r in definitions['bindings'] if r.get('sheet_id_type') == 'GatherableSheetId'
               and r.get('selected_sheet_id') == target['value']]
    if len(matches) != 1:
        return gap('source_route_not_found' if not matches else 'source_route_ambiguous')
    binding = matches[0]
    owner = {k:deepcopy(program[k]) for k in ('quest_id', 'instance_id', 'accept_time',
        'step_id', 'step_index', 'objective_index', 'process_started_at', 'session_id',
        'character_id', 'source', 'source_version')}
    observed_at = min(datetime.fromisoformat(p['observed_at'].replace('Z','+00:00'))
                      for p in (program,current)).isoformat()
    paths = ('data/arbeit-quest-program/source-schema.json', 'data/r3-life-selection/route-bindings.json')
    refs = [dict(path=p, sha256=hashlib.sha256((root()/p).read_bytes()).hexdigest()) for p in paths]
    return dict(result, **owner, ready=True, confirmed=True, observed_at=observed_at,
        native_route_id=binding['route_id'], identity_binding='current_GatherableName_exact_GatherableSheetId_to_source_route',
        target=deepcopy(target), route_binding=deepcopy(binding), source_refs=refs+deepcopy(definitions['source_refs']),
        completion_basis='separate_current_quest_objective_native_IsAchieved',
        auto_play_substitution=False, current_route_eligibility_confirmed=False)


def _utc():
    return datetime.now(timezone.utc).isoformat()


class CurrentLifeSession(CurrentUILayoutSession):
    """The existing session transport, with six exact metadata declarations."""
    def declarations(self):
        ui = super().declarations()
        life = declarations()
        if life['metadata_sha256'] != ui['metadata_sha256']:
            raise ValueError('Life metadata and current reader revisions differ')
        return dict(life, types=ui['types']+life['types'])

    def active_detail(self, *, allow_empty=False):
        r = self.reader
        entry = self.schema('PresentationSingleton')
        klass = _class(r, self.base, entry)
        fields = self.fields(klass, entry)
        instance = r.u64(r.u64(klass+0xb8)+fields['instance'])
        if not instance or r.u64(instance) != klass:
            raise MapReadPending('Current presentation singleton unavailable')
        service = self.ref(instance, fields, '<UIInstanceService>k__BackingField')
        fields = self.object_fields(service, 'UIInstanceService')
        table = self.ref(service, fields, 'handleTable')
        fields = self.object_fields(table, 'UIInstanceHandleTable')
        table = self.ref(table, fields, 'handleTable')
        wanted = next(t for t in declarations()['types'] if t['name']=='DetailCardUIInstance')
        active = []
        self._active_life_panels = []
        for _, handle in self.dictionary_values(table, limit=self.plan['limits']['instances']):
            if not handle or r.class_name(r.u64(handle)) != ('UIInstanceHandle', ''):
                continue
            shape = r.layout(r.u64(handle))
            fields = {f['name']:f for f in shape['fields']}
            status = fields.get('<Status>k__BackingField', {})
            target = fields.get('<Instance>k__BackingField', {})
            if status.get('type_code') not in (8,17) or target.get('type_code') not in (18,21,28):
                raise ValueError('Current UI handle declaration changed')
            self.watch(handle+status['offset'],4)
            if r.i32(handle+status['offset']) != 2:
                continue
            self.watch(handle+target['offset'],8)
            pointer = r.u64(handle+target['offset'])
            if pointer:
                name, namespace = r.class_name(r.u64(pointer))
                if (namespace.startswith('MM.Client.Presentation.UI.Instances.StageScene') and
                    name in ('CollectionBookUIInstance', 'CollectionSubUIInstance', 'CollectionUIInstance',
                             'LivingSkillUIInstance', 'DetailCardUIInstance')):
                    self._active_life_panels.append(dict(type=name,namespace=namespace,pointer=hex(pointer)))
            if pointer and r.class_name(r.u64(pointer)) == (wanted['name'], wanted['namespace']):
                active.append(pointer)
        if allow_empty and not active:
            return None
        if len(active) != 1:
            raise MapReadPending('Exact active DetailCardUIInstance count: '+str(len(active)))
        return active[0]

    def _boolean(self, pointer, field):
        if field['type_code'] != 2 or field['offset'] < 16:
            raise ValueError('Current detail boolean type changed')
        self.watch(pointer+field['offset'],1)
        return self.bool(pointer+field['offset'])

    def selected_sheet_id(self, pointer, name, field_name, wrapper_name):
        """Decode a verified single Int32 sequential value wrapper, never a ref.

        Native Il2CppType.data must identify the exact metadata definition.
        A different runtime type representation remains an explicit read gap.
        """
        fields = self.declared_fields(pointer, name)
        field = fields[field_name]
        if field['type_code'] != 17 or field['offset'] < 16:
            raise ValueError('Selected life ID is not the declared value type')
        declaration = next(t for t in declarations()['types'] if t['name']==name)
        wrapper = next(t for t in declarations()['types'] if t['name']==wrapper_name)
        if wrapper['flags'] & 0x18 != 8 or wrapper['fields'] != [{'name':'Id','type_index':136776}]:
            raise ValueError('Selected life ID wrapper is not single sequential System.Int32')
        index = [f['name'] for f in declaration['fields']].index(field_name)
        klass = self.reader.u64(pointer)
        field_table = self.reader.u64(klass+0x80)
        type_pointer = self.reader.u64(field_table+32*index+8)
        if type_definition_index(self, type_pointer) != wrapper['definition_index']:
            raise MapReadPending('Selected life ID runtime type definition unavailable')
        self.watch(pointer+field['offset'],4)
        return self.reader.i32(pointer+field['offset'])

    def current_life_detail(self):
        self._guards = []
        pointer = self.active_detail()
        fields = self.declared_fields(pointer, 'DetailCardUIInstance')
        names = ('isActivated','isClosing','isPanelOpened','isResultWaiting','isMotioning','isInitialized')
        flags = {name:self._boolean(pointer,fields[name]) for name in names}
        card = self.declared_ref(pointer, fields, 'cardInfo')
        if not card:
            raise MapReadPending('Current detail cardInfo unavailable')
        card_name, card_namespace = self.reader.class_name(self.reader.u64(card))
        wanted = {'GatherableSheetIdCardUIInfo':('GatherableSheetId','GatherableSheetId','nearby'),
                  'FisheryCardUIInfo':('SheetId','FisherySheetId','fishing')}
        if card_name not in wanted:
            raise MapReadPending('Current detail is not a native gathering/fishing card')
        # Exact declaration is checked even when inline ID decoding is missing.
        self.declared_fields(card,card_name)
        field_name, wrapper, link_kind = wanted[card_name]
        gaps = []
        try:
            selected_id = self.selected_sheet_id(card,card_name,field_name,wrapper)
        except MapReadPending as exc:
            selected_id = None
            gaps.append(str(exc))
        if any(self.reader.read(address,len(raw)) != raw for address,raw in self._guards):
            raise MapReadPending('Current life UI changed during read')
        # Waiting for a modal result can be normal while the card is open.
        # Preserve that field without inventing a busy/disabled interpretation.
        settled = all(flags[k] for k in ('isActivated','isPanelOpened','isInitialized')) and not any(
            flags[k] for k in ('isClosing','isMotioning'))
        return {'source':'current typed UIInstanceService -> DetailCardUIInstance.cardInfo',
                'observed_at':_utc(), 'metadata_sha256':declarations()['metadata_sha256'],
                'ui_pointer':hex(pointer),'card_pointer':hex(card), 'card_type':card_name,
                'card_namespace':card_namespace,'selected_sheet_id':selected_id,
                'sheet_id_type':wrapper,'link_kind':link_kind,'flags':flags,
                'settled':settled,'stable':True,'missing':gaps,'game_inputs':0}



    def current_life_identity(self):
        """Exact selected typed ID plus independently pinned current table files.

        Used only when reconciling an interrupted selection. This neither
        replays its tape nor promotes a saved selected-detail flag to current.
        """
        from .processing_observer import verify_processing_sheet_revision
        revision = verify_processing_sheet_revision(
            Path(self.reader.image).parent, selection_bindings()['client_revision_checks'])
        value = self.current_life_detail()
        value['table_revision_sha256'] = revision
        return value

    def current_life_field(self):
        """Actual empty owned-life panel enumeration; HUD is checked separately."""
        self._guards=[]
        self.active_detail(allow_empty=True)
        if any(self.reader.read(address,len(raw)) != raw for address,raw in self._guards):
            raise MapReadPending('Current life panels changed during empty-panel read')
        return dict(source='current UIInstanceHandleTable active owned life-panel enumeration',
            observed_at=_utc(),metadata_sha256=declarations()['metadata_sha256'],stable=True,
            active_owned_panels=deepcopy(self._active_life_panels),
            panels_absent_confirmed=not self._active_life_panels)


def detail_regions(route_id, layout, prepared):
    """Reuse the requirement ROI prepared with this exact selection route."""
    if prepared.get('node_id') != route_id or prepared.get('viewport') != list(layout['viewport']):
        raise ValueError('prepared_detail_region_route_changed')
    return {key:deepcopy(prepared[key]) for key in ('frame','requirements','exclusions','source','parent_kind')}




def requirement_pixels(image, regions):
    """Reuse mini27 red lock threshold/mask on the supplied image; no OCR."""
    import numpy as np
    bounds = tuple(round(v) for v in regions['requirements'])
    if not (0 <= bounds[0] < bounds[2] <= image.width and 0 <= bounds[1] < bounds[3] <= image.height):
        raise ValueError('requirement_region_outside_supplied_frame')
    pixels = np.asarray(image.crop(bounds).convert('RGB')).astype('int16')
    eligible = np.ones(pixels.shape[:2],dtype=bool)
    for region in regions['exclusions']:
        left,top,right,bottom = map(round,region)
        left,top,right,bottom = max(bounds[0],left),max(bounds[1],top),min(bounds[2],right),min(bounds[3],bottom)
        if left < right and top < bottom:
            eligible[top-bounds[1]:bottom-bounds[1],left-bounds[0]:right-bounds[0]] = False
    count = int(eligible.sum())
    if not count:
        raise ValueError('requirement_pixel_envelope_empty')
    red,green,blue = pixels[:,:,0],pixels[:,:,1],pixels[:,:,2]
    locks = (red > 95) & (green < red*.58) & (blue < red*.65) & eligible
    fraction = float(locks.sum()/count)
    return {'requirement_met':fraction <= .008,'red_fraction':fraction,'threshold':.008,
            'eligible_pixels':count,'source':'mini27 prepared_requirement_locked + execution-default',
            'scope':'current red restriction display only; not server unlock API', 'regions':regions}


def activity_idle(envelope):
    """Decode only known official activity flags. Absence never means false."""
    if envelope.get('ok') is not True or envelope.get('method') != 'get_activity':
        return None
    body = envelope.get('body')
    if not isinstance(body,dict):
        return None
    def flag(name,group):
        nested = body.get(group,{})
        return body.get(name,nested.get(name) if isinstance(nested,dict) else None)
    values = [flag('IsDead','combatState'),flag('IsInCombat','combatState'),
              flag('IsAutoPlaying','autoPlay'),flag('IsAutoTraveling','autoTravel')]
    dungeon = body.get('Dungeon',{})
    mode = body.get('Mode',{})
    if any(type(v) is not bool for v in values) or not isinstance(dungeon,dict) or not isinstance(mode,dict):
        return None
    if 'State' not in dungeon or 'MainButtonState' not in mode:
        return None
    if any(flag(name,'dialogue') is True for name in ('IsDialoguePlaying','IsWaitingForSelection')):
        return False
    if body.get('IsAbyssResultSequencePlaying') is True:
        return False
    return (dungeon['State']=='NotInDungeon' and not any(values) and
            mode['MainButtonState'] in ('Compass','Interaction','Combat','Fishing'))


def activity_stoppable(envelope):
    """Existing mini27 _field_blockers + exact Stop state; no absent-false guess."""
    if envelope.get('ok') is not True or envelope.get('method')!='get_activity':return False
    body=envelope.get('body')
    if not isinstance(body,dict):return False
    def flag(name,group):return body.get(name,(body.get(group) or {}).get(name))
    return (body.get('Mode',{}).get('MainButtonState')=='Stop' and body.get('Dungeon',{}).get('State')=='NotInDungeon'
        and all(flag(name,'combatState') is False for name in ('IsDead','IsInCombat'))
        and all(flag(name,'dialogue') is False for name in ('IsDialoguePlaying','IsWaitingForSelection'))
        and body.get('IsAbyssResultSequencePlaying',False) is False)



def gather_termination_evidence(node,snapshot,receipt):
    """Reconcile an owned activity after native step termination, never re-route.

    The saved typed program and selection tape retain their original observation
    times. Only the native step predicate and same current occurrence are fresh.
    This proof does not assert that the individual objective succeeded.
    """
    from .arbeit_operations import _fresh,_stamp
    from .quest_program import load_schema
    from .arbeit_movement import prepare_gather_tape
    try:
        if not receipt or receipt.get('phase') not in ('gather.native_activity','gather.stop'):return None
        if any(all(p.get(k)==node.get(k) for k in ('quest_id','step_id','objective_index'))
               for p in snapshot.get('objective_programs',[]) if isinstance(p,dict)):return None
        previous=receipt.get('operation_binding') or {};child=receipt.get('child_receipt') or {}
        scope=receipt.get('goal_scope') or {};owned=child.get('life_ownership') or {}
        if (previous.get('confirmed') is not True or previous.get('source')!='current-root-typed-quest-collections'
                or previous.get('dispatch_op')!=node.get('op')
                or node.get('op') not in ('life.gather','inventory.hold')
                or receipt.get('node_id')!=node.get('id')):return None
        identity=('quest_id','instance_id','accept_time','step_id','process_started_at')
        if any(scope.get(k) is None or scope[k]!=previous.get(k) for k in identity):return None
        if any(previous.get(k)!=node.get(k) for k in ('quest_id','step_id','objective_index')):return None
        if any(previous.get(k)!=snapshot.get(k) or receipt.get(k)!=snapshot.get(k)
               for k in ('session_id','character_id')):return None
        schema=load_schema()
        if any((previous.get('source_version') or {}).get(k)!=schema[k]
               for k in ('metadata_sha256','game_assembly_sha256')):return None
        declarations=[c for c in schema['classes'] if c['namespace']+'.'+c['name']==previous.get('data_type')]
        getters=[g for c in declarations for g in c.get('getters',[]) if g['name']=='get_Type' and 'constant' in g]
        if (len(getters)!=1 or (previous.get('discriminant') or {}).get('verified') is not True
                or previous['discriminant'].get('value')!=getters[0]['constant']):return None
        quest=snapshot.get('current_quest') or {}
        if (quest.get('source')!='current-root-typed-quest-collections' or quest.get('confirmed') is not True
                or not _fresh(quest,snapshot) or any(quest.get(k)!=scope[k] for k in identity if k!='step_id')):return None
        route=child.get('route_id');layout=owned.get('layout') or {};owner=owned.get('owner') or {}
        if (not route or owned.get('route_id')!=route or any(owner.get(k)!=snapshot.get(k)
                for k in ('session_id','character_id'))):return None
        tape=prepare_gather_tape(route,*layout['viewport'],ui_percent=layout['ui_percent'],safe_rect=layout['safe_rect'])
        if tape['actions']!=owned.get('selection_actions'):return None
        if node['op']=='life.gather':
            target=(previous.get('operands') or {}).get('GatherableName') or {}
            matches=[r for r in selection_bindings()['bindings'] if r['route_id']==route]
            if (previous.get('data_type')!='MM.QDBL.CodeGenerated.GatheringSuccessQuestObjectiveData'
                    or target.get('table_domain')!='GatherableSheet' or target.get('domain_mapping_verified') is not True
                    or target.get('interpreted') is not True or len(matches)!=1
                    or matches[0]['selected_sheet_id']!=target.get('value')):return None
        rows=[p for p in snapshot.get('state_proofs',[]) if p.get('source')=='current-root-typed-quest-collections'
            and p.get('confirmed') is True and p.get('state')=='step_complete' and _fresh(p,snapshot)
            and all(p.get(k)==scope[k] for k in identity)
            and all((p.get('source_version') or {}).get(k)==schema[k] for k in ('metadata_sha256','game_assembly_sha256'))
            and p.get('basis') in ('native_IsQuestStepCompletedBefore_with_current_runtime_StepIds',
                                 'native_IsAllStepsCompleted_and_current_runtime_StepIds')
            and _stamp(p['observed_at'])>_stamp(receipt['observed_at'])]
        if len(rows)!=1:return None
        return dict(deepcopy(rows[0]),semantics='owned_native_step_termination',route_id=route,
            node_id=node['id'],individual_objective_success_inferred=False,
            original_binding_observed_at=previous['observed_at'])
    except (ValueError,TypeError,KeyError):return None


def validate_gather_termination_stop(command,context,*,now):
    """Extra dispatch gate only for the explicit terminated-activity Stop path."""
    goals=command.get('goal_evidence') or []
    terminated=any(isinstance(p,dict) and p.get('semantics')=='owned_native_step_termination'
        for p in (goals if isinstance(goals,list) else [goals]))
    if not command.get('cleanup_only') and not terminated:return
    current=deepcopy(command.get('termination_context') or {})
    node=current.get('node') or {};snapshot=current.get('snapshot') or {};pending=current.get('pending') or {}
    receipt=pending.get('planner_receipt') or {};snapshot['now']=now
    proof=gather_termination_evidence(node,snapshot,receipt)
    process=getattr(context.get('window'),'process_started_at',None)
    owner=snapshot.get('input_owner') or {};owned=(receipt.get('child_receipt') or {}).get('life_ownership') or {}
    from .arbeit_operations import _fresh
    sent=pending.get('input_receipt') or {};rows=sent.get('receipts') or []
    actual=(rows[0].get('command') or {}) if len(rows)==1 else {}
    wanted=(owned.get('selected') or {}).get('link_action') or {}
    native_input_matches=(receipt.get('phase')=='gather.native_activity' and pending.get('status')=='sent_unconfirmed'
        and sent.get('status')=='sent_unconfirmed' and len(rows)==1 and rows[0].get('status')=='sent_unconfirmed'
        and actual.get('kind')=='click' and actual.get('target_id')==wanted.get('role')
        and [actual.get('x'),actual.get('y')]==wanted.get('point')
        and actual.get('expected_rect')==(owned.get('owner') or {}).get('client_rect'))
    if (command.get('cleanup_only') is not True or proof is None or goals!=[proof]
            or not native_input_matches or pending.get('operation_id')!=command.get('owned_operation_id')
            or any(snapshot.get(k)!=context.get(k) or command.get(k)!=context.get(k) for k in ('session_id','character_id'))
            or not process or process!=proof.get('process_started_at')
            or owner.get('continuous') is not True or not _fresh(owner,snapshot)
            or command.get('input_epoch')!=owner.get('epoch') or owner.get('epoch')!=owned.get('input_epoch')
            or command.get('source_refs')!=stop_sources()):
        raise ValueError('owned_native_step_termination_stop_context_changed')


def gather_goal_evidence(node,snapshot):
    from .arbeit_operations import _inventory,_objective_effect
    if 'material_requirement' in node.get('params',{}):
        request=validated_material_hold_request(node,snapshot)
        if request is None:return None
        proof=request['current_inventory_evidence']
        if proof['available_quantity']<request['required_quantity']:return None
        # The validator recomputes this complete unlocked/unreserved bag proof
        # and verifies total recipe quantity. Never relabel it as quest state.
        return [dict(deepcopy(proof),node_id=node['id'],required_quantity=request['required_quantity'],
            quantity_unit=request['quantity_unit'],requirement_origin='source_recipe_material',
            quest_objective=False,native_quest_effect_supplied=False,
            material_state_id=request['material_state_id'],parent_node_id=request['parent_node_id'],
            parent_scope=deepcopy(request['parent_scope']),source_recipe=deepcopy(request['source_recipe']),
            source_refs=deepcopy(request['source_refs']))]
    if node['op']=='inventory.hold':
        result=_inventory(node,snapshot,None)
        return result.get('effect_evidence') if result['status']=='complete' else None
    proof=_objective_effect(node,snapshot)
    return [proof] if proof else None


def stop_sources():
    return json.loads((root()/'data/r3-life/source-map.json').read_text('utf-8'))['source_refs']


class LifeObservationProvider:
    """Source-aware lifecycle supplier; process reads require explicit opt-in.

    ``pending_provider`` returns the engine's persisted pending record including
    input_receipt. Prepared, partial, unknown and dry-run tapes cannot own UI.
    Keep this provider for the run; ``invalidate`` on foreign input/stop/resume.
    ``session`` allows an already-open shared CurrentLifeSession to be reused.
    """
    def __init__(self, *, allow_process_read=False, session_factory=CurrentLifeSession,
                 pending_provider=None, activity_reader=None, route_selector=None,
                 cancelled=None,supply_selector=None):
        self.allow_process_read = allow_process_read
        self.session_factory = session_factory
        self.pending_provider = pending_provider
        self.activity_reader = activity_reader
        self.route_selector = route_selector
        self._shared_selector=supply_selector
        self.cancelled = cancelled or (lambda:False)
        self.invalidate()

    def invalidate(self):
        self._pending = None
        self._prepared = None
        self._selected = None
        self._idle = []
        self._activity_operation = None
        self._last_activity = None
        self._selector=self._shared_selector if self._shared_selector is not None else SupplySelector()
        self._selector.invalidate()
        self._route_decision=None
        self._native_pending=None
        self._reset_pending=None
        self._selection_reconciled_operation=None
        self._stop_sent=None
        self._reconciliation_operation=None

    def set_receipt(self, pending):
        self._pending = deepcopy(pending)

    def _route(self,node,snapshot):
        from .arbeit_operations import _item, _fresh, supply_candidates
        from .arbeit_movement import movement_definition
        self._termination=None
        pending=self.pending_provider() if self.pending_provider else self._pending
        terminated=gather_termination_evidence(node,snapshot,(pending or {}).get('planner_receipt'))
        if terminated:
            self._termination=terminated
            self._objective_binding=None
            self._route_decision={'status':'owned_cleanup_only','route_id':terminated['route_id']}
            return terminated['route_id']
        if node['op']=='inventory.hold':
            rows = [c for c in collect_supply_candidates(node,snapshot) if c['kind']=='native_gather']
            candidates=list(dict.fromkeys(c['route_id'] for c in rows))
            pending=self.pending_provider() if self.pending_provider else self._pending
            if self.route_selector:route=self.route_selector(node,snapshot,candidates)
            else:
                self._route_decision=self._selector.choose(node,snapshot,rows,kind='native_gather',pending=pending)
                route=(self._route_decision.get('candidate') or {}).get('route_id')
                # A locked card keeps ownership until its close input and HUD
                # successor are proved. The next inspection is a separate step.
                if pending and self._prepared and (pending.get('planner_receipt') or {}).get('phase') in ('gather.selection','gather.selection_reset'):
                    route=self._prepared['route_id']
            if route not in candidates:
                raise ValueError('unique_or_explicit_native_output_route_required')
        elif node['op']=='life.gather':
            pending=self.pending_provider() if self.pending_provider else self._pending
            binding=bind_native_gather_objective(node,snapshot,(pending or {}).get('planner_receipt'))
            if not binding['ready']:
                raise ValueError(binding['missing'][0])
            self._objective_binding=binding
            route=binding['native_route_id']
        else:
            raise ValueError('native_life_observer_does_not_observe_craft_or_alter')
        if route not in movement_definition()['gather_routes']:
            raise ValueError('route_not_in_source_selection_tapes')
        return route

    def _owned(self,pending,node,snapshot,owner,route,phase,frame_at):
        if not self._prepared or self._prepared['owner'] != owner or self._prepared['route_id'] != route:
            raise ValueError('same_uninterrupted_owned_selection_context_required')
        receipt = pending.get('planner_receipt') or {}
        sent = pending.get('input_receipt') or {}
        if not pending.get('operation_id') or pending.get('status')!='sent_unconfirmed' or sent.get('status')!='sent_unconfirmed':
            raise ValueError('complete_own_sent_tape_receipt_required')
        if receipt.get('phase')!=phase or receipt.get('node_id')!=node['id'] or any(
                receipt.get(k)!=snapshot.get(k) for k in ('session_id','character_id')):
            raise ValueError('owned_life_receipt_scope_changed')
        if (receipt.get('child_receipt') or {}).get('route_id')!=route:
            raise ValueError('owned_life_receipt_route_changed')
        expected = self._prepared['actions'] if phase=='gather.selection' else [self._selected['link_action']] if self._selected else []
        rows = sent.get('receipts',[])
        if not expected or len(rows)!=len(expected):
            raise ValueError('owned_selection_tape_incomplete')
        for actual,wanted in zip(rows,expected):
            command = actual.get('command',{})
            sent_at = datetime.fromisoformat(actual['observed_at'].replace('Z','+00:00'))
            prepared_at = datetime.fromisoformat(self._prepared['observed_at'].replace('Z','+00:00'))
            captured_at = datetime.fromisoformat(frame_at.replace('Z','+00:00'))
            if any(t.tzinfo is None for t in (sent_at,prepared_at,captured_at)) or not prepared_at <= sent_at <= captured_at:
                raise ValueError('owned_selection_receipt_not_between_preparation_and_current_frame')
            if actual.get('status')!='sent_unconfirmed' or command.get('kind')!=wanted['kind']:
                raise ValueError('owned_selection_tape_step_unconfirmed')
            if wanted['kind']=='key':
                if command.get('key')!=wanted['key']:
                    raise ValueError('owned_selection_tape_key_changed')
            else:
                if command.get('target_id')!=wanted['role'] or [command.get('x'),command.get('y')]!=wanted['point']:
                    raise ValueError('owned_selection_tape_target_changed')
            if tuple(command.get('expected_rect',())) != owner[-1]:
                raise ValueError('owned_selection_tape_viewport_changed')
        return receipt

    def _reconcile_after_pause(self,pending,node,snapshot,owner,route,frame,image,base):
        """Read current quiescence after RAM/lease loss, never revive ownership.

        The durable intent identifies the requested goal. Only new real idle
        observations and a current HUD permit a fresh selection tape. Neither
        an old click nor a saved selected-detail flag authorizes another click
        or an official Stop after interrupted input ownership.
        """
        receipt=pending.get('planner_receipt') or {};sent=pending.get('input_receipt') or {}
        if receipt.get('node_id')!=node['id'] or any(receipt.get(k)!=snapshot.get(k) for k in ('session_id','character_id')):
            raise ValueError('durable_gather_reconciliation_owner_changed')
        if pending.get('status') not in ('sent_unconfirmed','unknown') or sent.get('status') not in ('sent_unconfirmed','unknown'):
            raise ValueError('durable_gather_reconciliation_dispatched_intent_required')
        route_before=(receipt.get('child_receipt') or {}).get('route_id')
        if route_before!=route or not pending.get('operation_id') or not sent.get('receipts'):
            raise ValueError('durable_gather_reconciliation_route_and_receipt_required')
        rows=sent['receipts'];phase=receipt.get('phase')
        if len(rows)!=1 or (rows[0].get('command') or {}).get('kind')!=('official_stop' if phase=='gather.stop' else 'click'):
            raise ValueError('durable_gather_reconciliation_native_input_shape')
        if phase!='gather.stop':
            old_rect=rows[0]['command'].get('expected_rect',[])
            if len(old_rect)!=4 or any(type(v) not in (int,float) for v in old_rect) or old_rect[2]<=old_rect[0] or old_rect[3]<=old_rect[1]:
                raise ValueError('durable_gather_reconciliation_original_viewport_required')
            base['original_input_viewport']=old_rect
            base['viewport_changed_after_intent']=tuple(old_rect)!=owner[-1]
        elif rows[0]['command'].get('method')!='stop_action' or rows[0]['command'].get('params') is not None:
            raise ValueError('durable_gather_reconciliation_exact_stop_method')
        if self._reconciliation_operation!=pending['operation_id']:
            self._reconciliation_operation=pending['operation_id'];self._idle=[];self._last_activity=None
        envelope=self.activity_reader() if self.activity_reader else snapshot.get('activity_read',{})
        idle=activity_idle(envelope)
        if idle is None:raise ValueError('current_official_activity_flags_unavailable')
        stamp=datetime.fromisoformat(envelope['observed_at'].replace('Z','+00:00'))
        original=datetime.fromisoformat(receipt['observed_at'].replace('Z','+00:00'))
        if stamp.tzinfo is None or original.tzinfo is None or stamp<=original:
            raise ValueError('reconciliation_activity_predates_durable_intent')
        if self._last_activity and (stamp<self._last_activity[0] or stamp==self._last_activity[0] and idle!=self._last_activity[1]):
            raise ValueError('reconciliation_activity_out_of_order_or_conflicting')
        self._last_activity=(stamp,idle)
        proof=dict(base,source='official-cli:get_activity',observed_at=envelope['observed_at'],idle=idle,
            operation_id=pending['operation_id'],basis='current idle after interrupted ownership; original action effect not claimed',
            response_sha256=envelope.get('response_sha256'))
        if not idle:self._idle=[]
        elif not any(r['observed_at']==proof['observed_at'] for r in self._idle):self._idle=(self._idle+[proof])[-2:]
        base.update(checkpoint='native_activity',current_unlock_confirmed=False,observed_at=envelope['observed_at'],
            idle_observations=deepcopy(self._idle),reconciliation_only=True,original_action_ownership_confirmed=False)
        context=frame.get('current_context') or snapshot.get('current_context') or {}
        if not base.get('cleanup_only') and len(self._idle)>=2 and context.get('ok') is True and context.get('field_hud') is True and context.get('profile',{}).get('visual_confirmed') is True:
            x,y,right,bottom=context['safe_rect']
            layout=dict(viewport=list(image.size),ui_percent=context['profile']['percent'],canvas_scale=context['canvas_scale'],
                safe_rect=[x,y,right-x,bottom-y],profile_confirmed=True)
            from .arbeit_movement import prepare_gather_tape
            tape=prepare_gather_tape(route,*image.size,ui_percent=layout['ui_percent'],safe_rect=layout['safe_rect'])
            self._prepared=dict(owner=owner,route_id=route,layout=layout,actions=tape['actions'],detail_geometry=tape['detail_geometry'],observed_at=frame['captured_at'],
                input_epoch=(snapshot.get('input_owner') or {}).get('epoch'))
            self._selected=None;base.update(layout,reconciliation_resume_ready=True)
        return dict(ok=True,native_life=base,activity_read=envelope,missing=[],source='current idle reconciliation of durable gather intent',
            game_inputs=0,process_reads_enabled=False,live_verified=False)

    def _reconcile_selection(self,pending,node,snapshot,owner,route,frame,image,base,window,session):
        """Resolve the current selection under a new lease, never replay C/tape."""
        from .arbeit_operations import _fresh
        receipt=pending.get('planner_receipt') or {};child=receipt.get('child_receipt') or {}
        sent=pending.get('input_receipt') or {};scope=receipt.get('goal_scope') or {}
        current_quest=snapshot.get('current_quest') or {};lease=snapshot.get('input_owner') or {};identity=lease.get('owner') or {}
        phase=receipt.get('phase')
        if (phase not in ('gather.selection','gather.selection_reset') or receipt.get('node_id')!=node['id'] or child.get('route_id')!=route
            or any(receipt.get(k)!=snapshot.get(k) for k in ('session_id','character_id'))):
            raise ValueError('selection_reconciliation_original_goal_route_owner_required')
        if phase=='gather.selection_reset':
            next_route=child.get('next_route_id')
            candidates=collect_supply_candidates(node,snapshot)
            if next_route not in [r['route_id'] for r in candidates if r['kind']=='native_gather']:
                raise ValueError('selection_reset_reconciliation_exact_next_supply_candidate_required')
            route=next_route;base['route_id']=route
        if (pending.get('status') not in ('sent_unconfirmed','unknown') or sent.get('status') not in ('sent_unconfirmed','unknown')
            or not pending.get('operation_id') or not sent.get('receipts') or sent.get('attempted') is False):
            raise ValueError('selection_reconciliation_dispatched_intent_required')
        rows=sent['receipts']
        if phase=='gather.selection_reset':
            if len(rows)!=1 or rows[0].get('command',{}).get('kind')!='key' or rows[0]['command'].get('key')!='ESC':
                raise ValueError('selection_reset_reconciliation_original_close_intent_required')
        else:
            from .arbeit_movement import prepare_gather_tape
            old_rect=rows[0].get('command',{}).get('expected_rect') or []
            if len(old_rect)!=4 or any(type(v) not in (int,float) or not math.isfinite(v) for v in old_rect):
                raise ValueError('selection_reconciliation_original_tape_geometry_required')
            saved_layout=(child.get('life_ownership') or {}).get('layout') or {}
            expected=prepare_gather_tape(route,old_rect[2]-old_rect[0],old_rect[3]-old_rect[1],
                ui_percent=saved_layout.get('ui_percent',70),safe_rect=saved_layout.get('safe_rect'))['actions']
            if not 0<len(rows)<=len(expected):
                raise ValueError('selection_reconciliation_original_source_tape_prefix_required')
            for row,action in zip(rows,expected):
                command=row.get('command') or {}
                if command.get('kind')!=action['kind'] or tuple(command.get('expected_rect') or ())!=tuple(old_rect):
                    raise ValueError('selection_reconciliation_original_source_input_changed')
                if action['kind']=='key':
                    if command.get('key')!=action['key']:raise ValueError('selection_reconciliation_original_source_key_changed')
                elif (command.get('target_id')!=action['role'] or [command.get('x'),command.get('y')]!=action['point']
                    or action['kind']=='scroll' and command.get('delta')!=action.get('delta')):
                    raise ValueError('selection_reconciliation_original_source_target_changed')
        if (not _fresh(current_quest,snapshot) or current_quest.get('confirmed') is not True
            or current_quest.get('source')!='current-root-typed-quest-collections'
            or any(scope.get(k) is None or scope[k]!=current_quest.get(k)
                for k in ('quest_id','instance_id','accept_time','step_id','process_started_at'))
            or scope['quest_id']!=node.get('quest_id')
            or node.get('step_id') is not None and scope['step_id']!=node['step_id']):
            raise ValueError('same_uninterrupted_selection_unavailable:selection_reconciliation_same_accepted_occurrence_required')
        lease_meta=dict(lease,session_id=identity.get('session_id'),character_id=identity.get('character_id'))
        if (lease.get('continuous') is not True or not lease.get('epoch') or not _fresh(lease_meta,snapshot)
            or identity.get('process_started_at')!=scope['process_started_at'] or identity.get('pid')!=window.pid):
            raise ValueError('selection_reconciliation_fresh_current_input_lease_required')
        context=frame.get('current_context') or snapshot.get('current_context') or {}
        if context.get('ok') is not True or context.get('profile',{}).get('visual_confirmed') is not True:
            raise ValueError('selection_reconciliation_current_source_layout_required')
        x,y,right,bottom=context['safe_rect']
        layout=dict(viewport=list(image.size),ui_percent=context['profile']['percent'],canvas_scale=context['canvas_scale'],
            safe_rect=[x,y,right-x,bottom-y],profile_confirmed=True)
        if not self.allow_process_read:
            raise ValueError('read_only_life_process_access_disabled')
        # An already confirmed close (or a user-closed selection after pause)
        # can return to source entry only when actual owned panels are absent.
        if context.get('field_hud') is True:
            if session is not None:field=session.current_life_field()
            else:
                with self.session_factory(window.pid) as opened:field=opened.current_life_field()
            if (isinstance(field,dict) and field.get('stable') is True and field.get('panels_absent_confirmed') is True
                and field.get('active_owned_panels')==[] and field.get('metadata_sha256')==declarations()['metadata_sha256']):
                if not _fresh(dict(field,session_id=snapshot['session_id'],character_id=snapshot['character_id']),snapshot):
                    raise ValueError('selection_reconciliation_current_empty_panel_observation_stale')
                stamp=datetime.fromisoformat(field['observed_at'].replace('Z','+00:00'))
                original=datetime.fromisoformat(receipt['observed_at'].replace('Z','+00:00'))
                if stamp.tzinfo is None or original.tzinfo is None or stamp<=original:
                    raise ValueError('selection_reconciliation_current_field_must_follow_intent')
                from .arbeit_movement import prepare_gather_tape
                tape=prepare_gather_tape(route,*image.size,ui_percent=layout['ui_percent'],safe_rect=layout['safe_rect'])
                self._prepared=dict(owner=owner,route_id=route,layout=layout,actions=tape['actions'],detail_geometry=tape['detail_geometry'],
                    observed_at=frame['captured_at'],input_epoch=lease['epoch'])
                self._selected=None;self._selection_reconciled_operation=pending['operation_id']
                evidence=dict(base,source=field['source'],observed_at=frame['captured_at'],confirmed=True,
                    input_epoch=lease['epoch'],original_operation_id=pending['operation_id'],goal_scope=deepcopy(scope),
                    current_typed=deepcopy(field),source_refs=deepcopy(selection_bindings()['source_refs']),
                    original_input_effect_inferred=False,field_hud=True)
                base.update(layout,checkpoint='field_idle',current_unlock_confirmed=False,
                    selection_reconciled=True,selection_reconciliation_evidence=evidence,
                    selection_reconciliation_resume_ready=True,prior_route_id=child.get('route_id'),
                    life_ownership=dict(route_id=route,layout=deepcopy(layout),selection_actions=deepcopy(tape['actions']),
                        prepared_at=frame['captured_at'],input_epoch=lease['epoch'],selected=None))
                if phase=='gather.selection_reset':base['selection_reset_confirmed']=True
                return dict(ok=True,native_life=base,missing=[],game_inputs=0,live_verified=False)
        if phase=='gather.selection_reset':
            raise ValueError('selection_reset_reconciliation_current_empty_owned_panels_and_HUD_required')
        if session is not None:current=session.current_life_identity()
        else:
            with self.session_factory(window.pid) as opened:current=opened.current_life_identity()
        definitions=selection_bindings()
        matches=[r for r in definitions['bindings'] if r['route_id']==route]
        if len(matches)!=1:
            raise ValueError('selection_reconciliation_route_typed_ID_join_unavailable')
        binding=matches[0]
        if (not isinstance(current,dict) or current.get('stable') is not True or current.get('settled') is not True
            or current.get('metadata_sha256')!=definitions['metadata_sha256']
            or current.get('table_revision_sha256')!=definitions['client_revision_sha256']):
            raise ValueError('selection_reconciliation_current_typed_detail_and_table_revision_required')
        if not _fresh(dict(current,session_id=snapshot['session_id'],character_id=snapshot['character_id']),snapshot):
            raise ValueError('selection_reconciliation_current_typed_detail_stale')
        if current.get('selected_sheet_id')!=binding['selected_sheet_id'] or current.get('sheet_id_type')!=binding['sheet_id_type']:
            raise ValueError('selection_reconciliation_current_selected_item_differs_from_requested_route')
        link='fishing' if route.startswith('fishing-') else 'nearby'
        if current.get('link_kind')!=link:
            raise ValueError('selection_reconciliation_current_link_kind_changed')
        current_at=datetime.fromisoformat(current['observed_at'].replace('Z','+00:00'))
        old_at=datetime.fromisoformat(receipt['observed_at'].replace('Z','+00:00'))
        if current_at.tzinfo is None or old_at.tzinfo is None or current_at<=old_at:
            raise ValueError('selection_reconciliation_observation_must_follow_intent')
        from .arbeit_movement import prepare_gather_tape
        tape=prepare_gather_tape(route,*image.size,ui_percent=layout['ui_percent'],safe_rect=layout['safe_rect'])
        geometry=tape['detail_geometry']
        pixels=requirement_pixels(image,detail_regions(route,layout,geometry))
        if pixels['requirement_met'] is not True:
            raise ValueError('selection_reconciliation_current_native_requirement_locked')
        from .arbeit_movement import gather_nearby_candidate
        detail=dict(node_id=route,link_kind=link,owned=True,requirement_met=True,link_confirmed=True,settled=True,
            selected_sheet_id=current['selected_sheet_id'],sheet_id_type=current['sheet_id_type'],
            identity_binding='same requested route/current exact typed ID/source table revision/new input lease',
            link_binding='prepared common gather.travel layout',prepared_link_rect=geometry['link_rect'])
        action=gather_nearby_candidate(route,width=image.width,height=image.height,ui_percent=layout['ui_percent'],
            canvas_scale=layout['canvas_scale'],safe_rect=layout['safe_rect'],detail=detail)
        self._prepared=dict(owner=owner,route_id=route,layout=layout,actions=tape['actions'],detail_geometry=tape['detail_geometry'],
            observed_at=frame['captured_at'],input_epoch=lease['epoch'])
        self._selected=dict(detail=detail,typed=current,link_action=action)
        self._selection_reconciled_operation=pending['operation_id']
        evidence=dict(base,source='current typed selection plus source-pinned Gatherable/Fishery route identity',
            observed_at=frame['captured_at'],confirmed=True,input_epoch=lease['epoch'],
            original_operation_id=pending['operation_id'],original_input_effect_inferred=False,
            goal_scope=deepcopy(scope),current_typed=deepcopy(current),binding=deepcopy(binding),
            source_refs=deepcopy(definitions['source_refs']))
        base.update(layout,checkpoint='selected_detail',current_unlock_confirmed=True,detail=detail,
            selection_reconciled=True,selection_reconciliation_evidence=evidence,
            life_ownership=dict(route_id=route,owner=dict(session_id=owner[0],character_id=owner[1],
                window_signature=owner[2],client_rect=list(owner[3])),layout=deepcopy(layout),
                selection_actions=deepcopy(tape['actions']),prepared_at=frame['captured_at'],
                input_epoch=lease['epoch'],selected=deepcopy(self._selected)))
        return dict(ok=True,native_life=base,typed_detail=current,restriction_pixels=pixels,missing=[],
            game_inputs=0,process_reads_enabled=self.allow_process_read,live_verified=False)

    def __call__(self, *, window, node, image, frame, snapshot, session=None):
        result = {'ok':False,'missing':[],'source':'source native-life observer', 'game_inputs':0,
                  'process_reads_enabled':self.allow_process_read,'live_verified':False}
        try:
            if self.cancelled():
                self.invalidate()
                raise ValueError('life_observation_cancelled')
            if not all(snapshot.get(k) and frame.get(k)==snapshot[k] for k in ('session_id','character_id')):
                raise ValueError('current_frame_owner_required')
            if not frame.get('captured_at') or not frame.get('window_signature'):
                raise ValueError('current_frame_time_and_window_signature_required')
            rect = tuple(window.client_rect)
            if tuple(image.size)!=(rect[2]-rect[0],rect[3]-rect[1]):
                raise ValueError('current_frame_viewport_changed')
            owner = (snapshot['session_id'],snapshot['character_id'],frame['window_signature'],rect)
            if self._prepared and self._prepared['owner'] != owner:
                # A change cannot be undone by returning to the former window.
                self._prepared = None
                self._selected = None
                self._idle = []
            if self._prepared and self._prepared.get('input_epoch') is not None and self._prepared['input_epoch']!=(snapshot.get('input_owner') or {}).get('epoch'):
                self._prepared=None;self._selected=None;self._idle=[]
            pending = self.pending_provider() if self.pending_provider else self._pending
            if pending:
                prior=pending.get('planner_receipt') or {}
                if any(prior.get(k)!=snapshot.get(k) for k in ('session_id','character_id')):
                    raise ValueError('durable_gather_reconciliation_owner_changed')
            route = self._route(node,snapshot)
            base = {k:snapshot[k] for k in ('session_id','character_id')}
            base.update(source='native_life:owned_source_tape+current_typed_card+restriction_pixels',
                        observed_at=frame['captured_at'],route_id=route,confirmed=True)
            base['supply_selection']=deepcopy(self._route_decision)
            if self._termination:
                base.update(cleanup_only=True,termination_evidence=deepcopy(self._termination))
            if node['op']=='life.gather':
                base.update(node_id=node['id'],semantics='life.gather',objective_binding=deepcopy(self._objective_binding))
            phase=(pending.get('planner_receipt') or {}).get('phase') if pending else None
            if phase in ('gather.selection','gather.selection_reset') and (not self._prepared or self._selection_reconciled_operation==pending.get('operation_id')):
                return self._reconcile_selection(pending,node,snapshot,owner,route,frame,image,base,window,session)
            if phase in ('gather.native_activity','gather.stop') and (not self._prepared or self._reconciliation_operation==pending.get('operation_id')):
                return self._reconcile_after_pause(pending,node,snapshot,owner,route,frame,image,base)
            if not pending:
                context = frame.get('current_context') or snapshot.get('current_context') or {}
                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:
                    raise ValueError('current_HUD_menu_and_M_source_scale_match')
                x,y,right,bottom = context['safe_rect']
                layout = dict(viewport=list(image.size),ui_percent=context['profile']['percent'],
                              canvas_scale=context['canvas_scale'],safe_rect=[x,y,right-x,bottom-y],profile_confirmed=True)
                from .arbeit_movement import prepare_gather_tape
                tape = prepare_gather_tape(route,*image.size,ui_percent=layout['ui_percent'],safe_rect=layout['safe_rect'])
                self._prepared = dict(owner=owner,route_id=route,layout=layout,actions=tape['actions'],detail_geometry=tape['detail_geometry'],observed_at=frame['captured_at'],
                    input_epoch=(snapshot.get('input_owner') or {}).get('epoch'))
                self._selected = None
                self._idle = []
                base.update(layout,checkpoint='field_idle',current_unlock_confirmed=False,
                            checkpoint_scope='initial HUD selection entry; no claim of idle activity or unlock')
            else:
                phase = (pending.get('planner_receipt') or {}).get('phase')
                if phase=='gather.selection_reset':
                    if not self._reset_pending or self._reset_pending.get('owner')!=owner:
                        raise ValueError('owned_locked_detail_reset_preparation_required')
                    sent=pending.get('input_receipt') or {};rows=sent.get('receipts',[])
                    if pending.get('status')!='sent_unconfirmed' or sent.get('status')!='sent_unconfirmed' or len(rows)!=1:
                        raise ValueError('owned_locked_detail_reset_sent_receipt_required')
                    command=rows[0].get('command') or {}
                    reset_receipt=pending.get('planner_receipt') or {}
                    if reset_receipt.get('node_id')!=node['id'] or any(reset_receipt.get(k)!=snapshot.get(k) for k in ('session_id','character_id')):
                        raise ValueError('owned_locked_detail_reset_scope_changed')
                    reset_at=datetime.fromisoformat(rows[0]['observed_at'].replace('Z','+00:00'))
                    prepared_at=datetime.fromisoformat(self._reset_pending['observed_at'].replace('Z','+00:00'))
                    frame_at=datetime.fromisoformat(frame['captured_at'].replace('Z','+00:00'))
                    if any(t.tzinfo is None for t in (reset_at,prepared_at,frame_at)) or not prepared_at<=reset_at<=frame_at:
                        raise ValueError('owned_locked_detail_reset_timestamp_changed')
                    if rows[0].get('status')!='sent_unconfirmed' or command.get('kind')!='key' or command.get('key')!='ESC' or tuple(command.get('expected_rect',[]))!=owner[-1]:
                        raise ValueError('owned_locked_detail_reset_command_changed')
                    context=frame.get('current_context') or snapshot.get('current_context') or {}
                    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:
                        raise ValueError('locked_detail_reset_current_field_HUD_required')
                    route=self._reset_pending['next_route_id'];layout=self._prepared['layout']
                    from .arbeit_movement import prepare_gather_tape
                    tape=prepare_gather_tape(route,*image.size,ui_percent=layout['ui_percent'],safe_rect=layout['safe_rect'])
                    self._prepared=dict(owner=owner,route_id=route,layout=layout,actions=tape['actions'],detail_geometry=tape['detail_geometry'],observed_at=frame['captured_at'],input_epoch=(snapshot.get('input_owner') or {}).get('epoch'))
                    self._selected=None;self._idle=[]
                    base.update(layout,route_id=route,checkpoint='field_idle',current_unlock_confirmed=False,
                        selection_reset_confirmed=True,prior_route_id=self._reset_pending['route_id'])
                    result.update(ok=True,native_life=base);return result
                if phase not in ('gather.selection','gather.native_activity','gather.stop'):
                    raise ValueError('pending_action_is_not_owned_native_gather')
                if phase=='gather.stop':
                    if not self._native_pending or not self._stop_sent:
                        raise ValueError('same_uninterrupted_native_and_stop_preparation_required')
                    receipt=pending.get('planner_receipt') or {};sent=pending.get('input_receipt') or {};rows=sent.get('receipts',[])
                    if receipt.get('node_id')!=node['id'] or any(receipt.get(k)!=snapshot.get(k) for k in ('session_id','character_id')):
                        raise ValueError('owned_stop_receipt_scope_changed')
                    if len(rows)!=1 or (rows[0].get('command') or {}).get('kind')!='official_stop':
                        raise ValueError('owned_stop_command_receipt_required')
                    command=rows[0]['command']
                    if any(command.get(k)!=self._stop_sent.get(k) for k in ('owned_operation_id','method','params','input_epoch','session_id','character_id')):
                        raise ValueError('owned_stop_command_identity_changed')
                    if (receipt.get('child_receipt') or {}).get('route_id')!=route:
                        raise ValueError('owned_stop_route_changed')
                    # No retry on an absent/uncertain response. The source path
                    # continues read-only observations until two real idles.
                    native_pending=self._native_pending
                    self._owned(native_pending,node,snapshot,owner,route,'gather.native_activity',frame['captured_at'])
                else:
                    receipt = self._owned(pending,node,snapshot,owner,route,phase,frame['captured_at'])
                    native_pending=pending
                base.update(self._prepared['layout'])
                if phase=='gather.selection':
                    if not self.allow_process_read:
                        raise ValueError('read_only_life_process_access_disabled')
                    if session is not None:
                        current = session.current_life_identity() if node['op']=='life.gather' else session.current_life_detail()
                    else:
                        with self.session_factory(window.pid) as opened:
                            current = opened.current_life_identity() if node['op']=='life.gather' else opened.current_life_detail()
                    result['typed_detail'] = current
                    if current.get('stable') is not True or current.get('settled') is not True:
                        raise ValueError('current_native_detail_not_settled')
                    if node['op']=='life.gather':
                        binding=self._objective_binding['route_binding'];definitions=selection_bindings()
                        stamp=dict(current,session_id=snapshot['session_id'],character_id=snapshot['character_id'])
                        from .arbeit_operations import _fresh
                        if (current.get('sheet_id_type')!=binding['sheet_id_type']
                                or current.get('selected_sheet_id')!=binding['selected_sheet_id']
                                or current.get('metadata_sha256')!=definitions['metadata_sha256']
                                or current.get('table_revision_sha256')!=definitions['client_revision_sha256']
                                or not _fresh(stamp,snapshot)
                                or datetime.fromisoformat(current['observed_at'].replace('Z','+00:00')) <=
                                   datetime.fromisoformat(receipt['observed_at'].replace('Z','+00:00'))):
                            raise ValueError('current_native_gather_quest_selected_exact_source_identity_required')
                    link = 'fishing' if route.startswith('fishing-') else 'nearby'
                    if current.get('link_kind')!=link:
                        raise ValueError('current_native_detail_kind_changed')
                    geometry = self._prepared['detail_geometry']
                    pixels = requirement_pixels(image,detail_regions(route,self._prepared['layout'],geometry))
                    result['restriction_pixels'] = pixels
                    if pixels['requirement_met'] is not True:
                        from .arbeit_operations import _item,supply_candidates
                        candidates=[c for c in collect_supply_candidates(node,snapshot) if c['kind']=='native_gather']
                        candidate=next((c for c in candidates if c['route_id']==route),None)
                        if candidate and len(candidates)>1:
                            evidence=dict(base,identity_binding='own exact sent selection tape and current typed detail',eligible=False,restriction_pixels=pixels)
                            self._selector.observe(node,snapshot,candidate,eligible=False,evidence=evidence)
                            choice=self._selector.choose(node,snapshot,candidates,kind='native_gather')
                            next_route=(choice.get('candidate') or {}).get('route_id')
                            if next_route and next_route!=route:
                                self._reset_pending=dict(owner=owner,route_id=route,next_route_id=next_route,observed_at=frame['captured_at'])
                                base.update(checkpoint='locked_detail',current_unlock_confirmed=False,restriction_evidence=evidence,next_route_id=next_route,
                                    selection_reset_action={'kind':'key','key':'ESC','role':'gather.locked_detail.close'},source_refs=stop_sources())
                                result.update(ok=True,native_life=base,missing=['current_native_detail_requirement_locked']);return result
                        raise ValueError('current_native_detail_requirement_locked')
                    detail = dict(node_id=route,link_kind=link,owned=True,requirement_met=True,link_confirmed=True,settled=True,
                                  identity_binding='same own sent selection tape, window, character and uninterrupted provider',
                                  link_binding='prepared common gather.travel layout',prepared_link_rect=geometry['link_rect'],
                                  selected_sheet_id=current.get('selected_sheet_id'),sheet_id_type=current.get('sheet_id_type'))
                    from .arbeit_movement import gather_nearby_candidate
                    layout = self._prepared['layout']
                    action = gather_nearby_candidate(route,width=image.width,height=image.height,ui_percent=layout['ui_percent'],
                        canvas_scale=layout['canvas_scale'],safe_rect=layout['safe_rect'],detail=detail)
                    self._selected = dict(detail=detail,typed=current,link_action=action)
                    self._reconciliation_operation=None
                    base.update(checkpoint='selected_detail',current_unlock_confirmed=True,detail=detail)
                else:
                    self._native_pending=deepcopy(native_pending)
                    if self._activity_operation != native_pending['operation_id']:
                        self._idle = []
                        self._last_activity = None
                        self._activity_operation = native_pending['operation_id']
                    envelope = self.activity_reader() if self.activity_reader else snapshot.get('activity_read',{})
                    result['activity_read'] = envelope
                    idle = activity_idle(envelope)
                    if idle is None or not envelope.get('observed_at'):
                        raise ValueError('current_official_activity_flags_unavailable')
                    stamp = datetime.fromisoformat(envelope['observed_at'].replace('Z','+00:00'))
                    sent_at = max(datetime.fromisoformat(row['observed_at'].replace('Z','+00:00')) for row in native_pending['input_receipt']['receipts'])
                    if stamp.tzinfo is None or stamp <= sent_at:
                        raise ValueError('activity_observation_predates_own_native_action')
                    if self._last_activity:
                        previous_at,previous_idle = self._last_activity
                        if stamp < previous_at or (stamp == previous_at and idle != previous_idle):
                            raise ValueError('activity_observation_out_of_order_or_conflicting')
                    self._last_activity = (stamp,idle)
                    proof = dict(base,source='official-cli:get_activity',observed_at=envelope['observed_at'],idle=idle,
                                 response_sha256=envelope.get('response_sha256'),operation_id=pending['operation_id'])
                    if not idle:
                        self._idle=[]
                    elif not any(row['observed_at']==proof['observed_at'] for row in self._idle):
                        self._idle=(self._idle+[proof])[-2:]
                    base.update(observed_at=envelope['observed_at'],checkpoint='native_activity',
                                current_unlock_confirmed=True,detail=self._selected['detail'],idle_observations=deepcopy(self._idle))
                    goal=[deepcopy(self._termination)] if self._termination else gather_goal_evidence(node,snapshot)
                    lease=snapshot.get('input_owner') or {}
                    from .arbeit_operations import _fresh
                    activity_meta=dict(base,source='official-cli:get_activity',observed_at=envelope['observed_at'])
                    stop_row=((pending.get('input_receipt') or {}).get('receipts') or [{}])[0] if phase=='gather.stop' else {}
                    retry=(stop_row.get('status')=='no_effect' and stop_row.get('explicit_no_effect') is True
                        and (stop_row.get('response') or {}).get('error')=='invalid_state'
                        and stop_row.get('observed_at') and stamp>datetime.fromisoformat(stop_row['observed_at'].replace('Z','+00:00')))
                    if (phase=='gather.native_activity' or retry) and goal and activity_stoppable(envelope) and _fresh(activity_meta,snapshot):
                        if lease.get('continuous') is not True or not lease.get('epoch') or lease['epoch']!=self._prepared.get('input_epoch') or not _fresh(lease,snapshot):
                            raise ValueError('same_current_input_epoch_required_for_owned_stop')
                        proposal=dict(kind='official_stop',method='stop_action',params=None,role='gather.owned.stop',
                            observed_at=envelope['observed_at'],session_id=snapshot['session_id'],character_id=snapshot['character_id'],input_epoch=lease['epoch'],
                            source='mini27 _stop_gather_at_goal -> connector.stop_action; current owned goal + Stop',
                            owned_operation_id=native_pending['operation_id'],current_activity=envelope,goal_evidence=goal,source_refs=stop_sources())
                        if self._termination:
                            stop_snapshot={k:deepcopy(snapshot.get(k)) for k in ('session_id','character_id','observed_at',
                                'current_quest','objective_programs','state_proofs','input_owner')}
                            proposal.update(cleanup_only=True,termination_context=dict(node=deepcopy(node),
                                snapshot=stop_snapshot,pending=deepcopy(native_pending)))
                        self._stop_sent=deepcopy(proposal);base['stop_proposal']=proposal
                        if retry:base.update(stop_retry_no_effect=True,stop_retry_evidence=deepcopy(stop_row))
            if self.cancelled():
                self.invalidate()
                raise ValueError('life_observation_cancelled')
            if self._prepared:
                base['life_ownership']=dict(route_id=route,owner=dict(session_id=owner[0],character_id=owner[1],window_signature=owner[2],client_rect=list(owner[3])),
                    layout=deepcopy(self._prepared['layout']),selection_actions=deepcopy(self._prepared['actions']),
                    prepared_at=self._prepared['observed_at'],input_epoch=self._prepared.get('input_epoch'),selected=deepcopy(self._selected))
            result.update(ok=True,native_life=base)
            return result
        except (OSError,ValueError,RuntimeError,KeyError,TypeError,AttributeError) as exc:
            result['missing'].append(str(exc))
            return result
