"""Read-only official processing observations joined to a current UI provider.

No game writes, process reads, capture, implicit authentication, paid API calls,
or default UI geometry. Official work rows are a multiset, never stable IDs.
"""
from __future__ import annotations
from mini_masukja.native_type_reference import type_definition_index
from collections import Counter
from copy import deepcopy
from datetime import datetime, timezone
import math
import json
import struct
import hashlib
import re
from pathlib import Path
from functools import lru_cache

from .arbeit_operations import _fresh, _item, supply_candidates
from .paths import root
from .ui_layout_observer import CurrentUILayoutSession, XForestView
from .vendor.map_character_context import _class
from .vendor.map_memory import MapReadPending


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


@lru_cache(maxsize=1)
def processing_current_sources():
    return json.loads((root()/'data/r3-processing/source-map.json').read_text('utf-8'))


@lru_cache(maxsize=1)
def processing_recipe_bindings():
    """Current-version static definitions, independent of all old run paths."""
    result=json.loads((root()/'data/arbeit-processing-observer/recipe-bindings.json').read_text('utf-8'))
    if result['metadata_sha256']!=processing_declarations()['metadata_sha256']:
        raise ValueError('processing_recipe_metadata_revision_changed')
    ids=[r['altering_sheet_id'] for r in result['bindings']]
    routes=[r['route_id'] for r in result['bindings']]
    if len(ids)!=len(set(ids)) or len(routes)!=len(set(routes)):
        raise ValueError('processing_recipe_binding_not_unique')
    return result


def recipe_binding_for_selected_id(raw_id,metadata_sha256):
    if type(raw_id) is not int or metadata_sha256!=processing_declarations()['metadata_sha256']:
        return None
    # The sheet ID is an Int32 wrapper. Compare its actual signed value with
    # the exact four stored bytes, without inventing an official API item ID.
    stored_id=raw_id & 0xffffffff
    rows=[r for r in processing_recipe_bindings()['bindings'] if r['altering_sheet_id']==stored_id]
    return deepcopy(rows[0]) if len(rows)==1 else None


def verify_processing_sheet_revision(client_root,checks=None):
    """Hash only the retained Starter index and three exact ordinary ranges.

    Tables can change without a metadata/assembly update. No new package scan,
    decompression, process operation or fallback to historical bytes is used.
    """
    checks=processing_recipe_bindings()['client_revision_checks'] if checks is None else checks
    client_root=Path(client_root).resolve();before={}
    for check in checks:
        path=(client_root/check['path']).resolve()
        if not path.is_relative_to(client_root) or not check['path'].startswith('ResourcesPackage/Starter/'):
            raise ValueError('processing_sheet_revision_path_scope')
        if type(check['offset']) is not int or type(check['length']) is not int or not 0<=check['offset'] or not 0<check['length']<=8*1024*1024:
            raise ValueError('processing_sheet_revision_range_scope')
        stat=path.stat();signature=(stat.st_size,stat.st_mtime_ns)
        if path in before and before[path]!=signature:raise MapReadPending('processing_sheet_changed_during_read')
        before[path]=signature
        if check['whole_file'] and stat.st_size!=check['length']:raise MapReadPending('processing_sheet_index_revision_changed')
        with path.open('rb') as stream:stream.seek(check['offset']);raw=stream.read(check['length'])
        if len(raw)!=check['length'] or hashlib.sha256(raw).hexdigest()!=check['sha256']:
            raise MapReadPending('processing_sheet_revision_changed:'+check.get('asset',check['path']))
    for path,signature in before.items():
        stat=path.stat()
        if (stat.st_size,stat.st_mtime_ns)!=signature:raise MapReadPending('processing_sheet_changed_during_read')
    return hashlib.sha256(json.dumps(checks,sort_keys=True,separators=(',',':')).encode()).hexdigest()


class CurrentProcessingSession(CurrentUILayoutSession):
    """Read only two exact active processing panels and their existing models.

    Active handle -> MMML window -> stable LayoutVM outputs and ViewModelRoot
    are the same pinned paths used by the current dialogue/map readers. Model
    values are observations; no source formula is executed in the game and no
    layout rectangle becomes an input target merely because it is present.
    """
    PANEL_NAMES = ('AlteringFacilityPopupUIInstance','AlteringPanelPopupUIInstance','CraftingHubPopupUIInstance')

    def declarations(self):
        original=super().declarations();processing=processing_declarations()
        if original['metadata_sha256']!=processing['metadata_sha256']:
            raise ValueError('processing_metadata_revision_changed')
        added=processing_current_sources()
        if added['metadata_sha256']!=processing['metadata_sha256']:raise ValueError('r3_processing_metadata_revision_changed')
        unique={}
        for declaration in original['types']+processing['types']+added['types']:
            key=(declaration['name'],declaration['namespace'])
            if key in unique and any(unique[key].get(k)!=declaration.get(k) for k in ('definition_index','flags','fields')):
                raise ValueError('conflicting_processing_runtime_declaration:'+'.'.join(key))
            unique.setdefault(key,declaration)
        return dict(processing,types=list(unique.values()))

    def active_processing_panels(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')
        service=self.ref(instance,fields,'<UIInstanceService>k__BackingField')
        f=self.object_fields(service,'UIInstanceService');table=self.ref(service,f,'handleTable')
        f=self.object_fields(table,'UIInstanceHandleTable');table=self.ref(table,f,'handleTable')
        wanted={t['name']:t['namespace'] for t in self.declarations()['types'] if t['name'] in self.PANEL_NAMES}
        found={}
        for _,handle in self.dictionary_values(table,limit=self.plan['limits']['instances']):
            if not handle or r.class_name(r.u64(handle))!=('UIInstanceHandle',''):continue
            fields={f['name']:f for f in r.layout(r.u64(handle))['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('processing_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 not pointer:continue
            name,namespace=r.class_name(r.u64(pointer))
            if name not in wanted or namespace!=wanted[name]:continue
            if name in found:raise MapReadPending('ambiguous_current_processing_panel:'+name)
            self.declared_fields(pointer,name)
            found[name]=pointer
        if not found and not allow_empty:raise MapReadPending('current_altering_facility_or_panel_ui_required')
        return found

    def selected_recipe_id(self,pointer):
        """Single verified sequential Int32 wrapper, never a guessed field size."""
        name='AlteringPanelPopupUIInstance';f=self.declared_fields(pointer,name)['alteringSheetId']
        declarations=processing_declarations()['types'];decl=next(t for t in declarations if t['name']==name)
        wrapper=next(t for t in declarations if t['name']=='AlteringSheetId')
        if f['type_code']!=17 or f['offset']<16 or wrapper['flags']&0x18!=8 or wrapper['fields']!=[{'name':'Id','type_index':136776}]:
            raise ValueError('current_altering_sheet_id_type_changed')
        index=[f['name'] for f in decl['fields']].index('alteringSheetId')
        table=self.reader.u64(self.reader.u64(pointer)+0x80)
        type_pointer=self.reader.u64(table+32*index+8)
        if type_definition_index(self, type_pointer)!=wrapper['definition_index']:
            raise MapReadPending('current_altering_sheet_id_runtime_definition')
        return struct.unpack('<i',self.watch(pointer+f['offset'],4))[0]

    def processing_model(self,handle):
        r=self.reader;h=self.object_fields(handle,'MMMLUIHandle')
        # MMMLUIHandle.ViewModelRoot is a source-pinned inline XElem. Verify
        # the element shape before reading Forest/Id with unboxed offsets.
        shape=self.source_shape('XElem');klass=_class(r,self.base,shape);offsets=self.fields(klass,shape)
        if offsets['Forest']!=16 or offsets['Id']!=24:raise ValueError('current_processing_xelem_shape_changed')
        address=handle+h['<ViewModelRoot>k__BackingField']
        forest=struct.unpack('<Q',self.watch(address,8))[0]
        element=struct.unpack('<i',self.watch(address+8,4))[0]
        if not forest:raise MapReadPending('current_processing_model_root')
        view=XForestView(self,forest)
        names=set(processing_current_sources()['model_attributes'])
        # Traverse only fields declared by the retained processing writers.
        # Array child counts remain model observations, not API work counts.
        queue=[element];seen=set();rows=[]
        while queue:
            current=queue.pop(0)
            if current in seen:continue
            if len(seen)>=256:raise MapReadPending('current_processing_model_bound_exceeded')
            seen.add(current);values=view.attributes(current,names);children=view.children(current)
            rows.append(dict(element_id=current,values=values,children=children))
            queue.extend(children)
            for value in values.values():
                if isinstance(value,dict) and 'element_id' in value:
                    if value['forest']!=forest:raise ValueError('processing_model_cross_forest_reference')
                    queue.append(value['element_id'])
        for row in rows:
            for value in row['values'].values():
                if isinstance(value,dict):value.pop('forest',None)
        return dict(root_element_id=element,rows=rows)

    def processing_layout(self,handle):
        r=self.reader;h=self.object_fields(handle,'MMMLUIHandle')
        self.watch(handle+h['shouldActive'],1)
        if not self.bool(handle+h['shouldActive']):raise MapReadPending('current_processing_mmml_handle_inactive')
        window_id=self.window_id(handle);service=self.ref(handle,h,'service')
        sf=self.object_fields(service,'MMMLService');entries=self.ref(service,sf,'entryMap')
        windows=[p for key,p in self.dictionary_values(entries,key_type='Int32',value_type='UnityMMMLWindow',limit=4096) if key==window_id]
        if len(windows)!=1:raise MapReadPending('current_processing_mmml_window_ambiguous')
        window=windows[0];w=self.object_fields(window,'UnityMMMLWindow')
        if r.i32(window+w['windowId'])!=window_id:raise MapReadPending('current_processing_window_identity_changed')
        base=self.fields(r.u64(r.u64(window)+0x58),self.schema('MMMLWindow'))
        for field in ('closing','needCheckLayoutDirty'):self.watch(window+base[field],1)
        if self.bool(window+base['closing']) or self.bool(window+base['needCheckLayoutDirty']):
            raise MapReadPending('current_processing_layout_closing_or_dirty')
        context=self.ref(window,base,'mainLayoutContext');c=self.object_fields(context,'LayoutVMContext')
        self.watch(context+c['NeedToCompile'],1)
        if self.bool(context+c['NeedToCompile']):raise MapReadPending('current_processing_layout_needs_compile')
        vm=self.ref(context,c,'VM');v=self.object_fields(vm,'LayoutVM')
        program=self.ref(vm,v,'Program');p=self.object_fields(program,'LayoutProgram')
        infos=self.ref(program,p,'ElementBasicInfos');values=self.ref(vm,v,'values')
        count,_=self.array(infos,name='ElementBasicInfo[]',stride=8,limit=self.plan['limits']['layout_elements'])
        value_count,_=self.array(values,name='Single[]',stride=4,limit=self.plan['limits']['layout_elements']*8+10000)
        if count*8>value_count:raise ValueError('current_processing_layout_values_truncated')
        candidates=[]
        for index in range(count):
            tag_id,name_id=struct.unpack('<ii',self.watch(infos+32+index*8,8))
            raw=struct.unpack('<8f',self.watch(values+32+index*32,32))
            rect=[raw[self.plan['layout_values'][k]] for k in ('left','top','width','height')]
            if all(math.isfinite(x) for x in rect) and rect[2]>0 and rect[3]>0:
                candidates.append(dict(index=index,tag=self.name_text(tag_id),name=self.name_text(name_id),rect_ui=rect))
        for field,size in (('lastReferenceResolutionScale',4),('global_ScreenSize',8)):self.watch(service+sf[field],size)
        scale=self.float(service+sf['lastReferenceResolutionScale']);screen=self.pair(service+sf['global_ScreenSize'])
        if scale<=0:raise ValueError('current_processing_mmml_scale')
        if self.ref(vm,v,'Program')!=program or self.ref(vm,v,'values')!=values:
            raise MapReadPending('current_processing_layout_program_changed')
        result=dict(window_id=window_id,scale=scale,screen_ui=list(screen),candidates=candidates,
                    rect_source='current-mmml-output',candidate_only=True)
        from .quest_ui_actions import quest_ui_session
        try:quest_ui_session(self).bind_runtime_callbacks(result,handle=handle)
        except (OSError,RuntimeError,ValueError) as exc:
            result['callback_missing']=[str(exc)]
        from .material_close import decorate_material_close_layout
        return decorate_material_close_layout(self,handle,result)

    def current_processing(self):
        self._guards=[];table_revision=self.current_table_revision()
        mmml_revision=self.current_mmml_revision()
        panels=self.active_processing_panels();out={};missing=[]
        for name,pointer in panels.items():
            fields=self.declared_fields(pointer,name);handle=self.declared_ref(pointer,fields,'handle')
            row=dict(ui_type=name,ui_pointer=hex(pointer),handle_pointer=handle,layout={})
            try:row['layout']=self.processing_layout(handle)
            except MapReadPending as exc:row['layout_missing']=[str(exc)]
            row['layout']['ui_pointer']=hex(pointer)
            if name=='AlteringPanelPopupUIInstance':
                try:row['altering_sheet_id']=self.selected_recipe_id(pointer)
                except MapReadPending as exc:missing.append(str(exc))
                row['production_amount']=self.declared_scalar(pointer,fields,'productionAmount',codes=(8,))
            try:row['model']=self.processing_model(handle)
            except MapReadPending as exc:missing.append(str(exc))
            out[name]=row
        if any(self.reader.read(a,len(raw))!=raw for a,raw in self._guards):
            raise MapReadPending('current_processing_ui_changed_during_read')
        result=dict(source='current typed UIInstanceService -> Altering panels -> MMML model/layout',
                    observed_at=datetime.now(timezone.utc).isoformat(),metadata_sha256=processing_declarations()['metadata_sha256'],
                    table_revision_sha256=table_revision,mmml_revision_sha256=mmml_revision,
                    stable=True,panels=out,missing=missing,game_inputs=0,memory_writes=0)
        result['return_ui']=processing_return_ui(result)
        return result

    def current_table_revision(self):
        return verify_processing_sheet_revision(Path(self.reader.image).parent)

    def current_mmml_revision(self):
        return verify_processing_sheet_revision(Path(self.reader.image).parent,processing_current_sources()['mmml_checks'])


def processing_scene_from_current(current,snapshot):
    """Preserve actual current facts without manufacturing unresolved joins."""
    scene=dict(source=current.get('source'),observed_at=current.get('observed_at'),
               session_id=snapshot.get('session_id'),character_id=snapshot.get('character_id'),
               confirmed=False,source_bound=current.get('stable') is True,typed_current=current,
               missing=list(current.get('missing',[])),game_inputs=0,live_verified=False)
    panels=current.get('panels',{});detail=panels.get('AlteringPanelPopupUIInstance');facility=panels.get('AlteringFacilityPopupUIInstance')
    if detail:
        scene.update(stage='recipe_detail',selected_altering_sheet_id=detail.get('altering_sheet_id'),production_amount=detail.get('production_amount'))
        table_current=current.get('table_revision_sha256')==processing_recipe_bindings()['client_revision_sha256']
        binding=recipe_binding_for_selected_id(detail.get('altering_sheet_id'),current.get('metadata_sha256')) if current.get('stable') is True and table_current else None
        if not table_current:scene['missing'].append('current_processing_client_sheet_revision')
        if binding:
            definition=binding['current']
            scene.update(recipe_id=binding['route_id'],facility_id=binding['group'],
                         current_materials=[dict(name=m['name'],required=m['required']) for m in definition['materials']],
                         recipe_definition=binding,
                         recipe_definition_basis='source-pinned client table selected by current typed AlteringSheetId; not current inventory or unlock',
                         recipe_source_refs=processing_recipe_bindings()['source_refs'])
        else:
            scene['missing']+=['current_altering_sheet_id_to_source_recipe_join','current_recipe_materials_identity_and_quantity']
        scene['missing']+=['current_detail_local_register_enabled','current_recipe_content_bottom_and_scroll']
    else:
        scene['missing']+=['current_facility_recipe_list_or_collect_intent','current_recipe_array_identity_order_and_scroll',
                           'current_recipe_list_parent_and_flow_columns']
    if facility:
        model=facility.get('model',{});rows={r['element_id']:r for r in model.get('rows',[])}
        values=rows.get(model.get('root_element_id'),{}).get('values',{})
        if isinstance(values.get('currentFacilityName'),str):scene['facility_name']=values['currentFacilityName']
        for source,key in (('isMoveToFacilityVisible','move_visible'),('isUpgradeVisible','upgrade_visible')):
            if type(values.get(source)) is bool:scene[key]=values[source]
        queue=values.get('processingQueue')
        if isinstance(queue,dict) and queue.get('element_id') in rows:
            scene['observed_queue_model_child_count']=len(rows[queue['element_id']]['children'])
            # Native child count is retained, but no level/slot-capacity or
            # generated-slot semantic join has been established from it yet.
    selected=detail or facility
    layout=selected.get('layout',{}) if selected else {}
    viewport=(snapshot.get('layout') or {}).get('viewport')
    if not viewport or len(viewport)!=2 or len(layout.get('screen_ui',[]))!=2 or any(
            abs(layout['screen_ui'][i]*layout.get('scale',0)-viewport[i])>2 for i in range(2)):
        scene['missing'].append('current_processing_mmml_viewport_matches_window')
        scene['source_bound']=False
    scene['missing']+=['current_generated_slot_capacity','current_processing_action_owner_and_input_epoch',
                       'current_processing_target_enabled_and_unoccluded']
    scene['missing']=list(dict.fromkeys(scene['missing']))
    return scene


class ProcessingModel:
    """Identity-preserving view of the source-selected XForest nodes."""
    def __init__(self,panel):
        model=panel.get('model',{});self.rows={r['element_id']:r for r in model.get('rows',[])}
        self.root=self.rows.get(model.get('root_element_id'),{}).get('values',{})
    def values(self,reference):
        return self.rows.get(reference.get('element_id'),{}).get('values',{}) if isinstance(reference,dict) else {}
    def children(self,reference):
        row=self.rows.get(reference.get('element_id'),{}) if isinstance(reference,dict) else {}
        return [(key,self.rows.get(key,{}).get('values',{})) for key in row.get('children',[])]
    def body(self):
        return self.values(self.root.get('uiData')) or self.root


def processing_return_ui(current):
    """Current close/back descriptor, separate from the active key receiver.

    An event proves a callback exists, not that an unrelated overlay will let
    Escape reach it. The return planner must bind its current input receiver.
    """
    source=processing_current_sources();panels=current.get('panels',{})
    selected=next(((surface,panels[name]) for surface,name in (
        ('detail','AlteringPanelPopupUIInstance'),('facility','AlteringFacilityPopupUIInstance'),('hub','CraftingHubPopupUIInstance')) if name in panels),None)
    result=dict(source='current processing model close/back callback + current-version MMML',source_family='native_alter',observed_at=current.get('observed_at'),
        source_refs=source['source_refs'],key_receiver_confirmed=False,missing=[])
    if not selected:return dict(result,missing=['current_processing_return_surface'])
    surface,panel=selected;binding=source['return_bindings'][surface];model=ProcessingModel(panel);data=model.root
    event=data.get(binding['callback']);branch=binding.get('target','onclose')
    result.update(surface=surface,panel_id=panel.get('ui_pointer'),ui_pointer=panel.get('ui_pointer'),
                  current_layout=deepcopy(panel.get('layout',{})),window_id=panel.get('layout',{}).get('window_id'))
    if current.get('stable') is not True or current.get('mmml_revision_sha256')!=source['mmml_revision_sha256']:
        return dict(result,missing=['current_processing_return_source_revision'])
    if surface=='facility':
        if type(data.get('canGoBack')) is not bool:return dict(result,missing=['current_facility_canGoBack'])
        branch=binding['true_branch'] if data['canGoBack'] else binding['false_branch'];result['can_go_back']=data['canGoBack']
    if not isinstance(event,dict) or event.get('event_present') is not True:
        return dict(result,missing=['current_processing_return_callback_event'])
    result['close']=dict(callback_event=deepcopy(event),callback=deepcopy(event),callback_name=binding['callback'],branch=branch,
        control=branch.rsplit('.',1)[-1],enabled=True,closable=binding.get('closable_constant'),key='ESC',source_binding=binding,
        static_source_pinned=binding,source_refs=source['source_refs'],
        basis='source callback presence; actual key receiver is a separate current-state requirement')
    return result


def bind_processing_return_ui(current,node,snapshot,frame):
    result=processing_return_ui(current)
    result.update(session_id=snapshot.get('session_id'),character_id=snapshot.get('character_id'),
        process_started_at=snapshot.get('window_preflight',{}).get('selected',{}).get('process_started_at'),
        source_revision=processing_current_sources()['mmml_source_sha256'],
        source_checks_revision=current.get('mmml_revision_sha256'),
        stable=current.get('stable') is True,confirmed=False,frame=deepcopy(frame))
    try:
        owner=_processing_owner(snapshot,frame);intent=_processing_intent(node,snapshot)
        result.update(recipe_id=intent['route_id'],input_epoch=owner['epoch'],owner_proof=owner,
                      recipe_source_refs=processing_recipe_bindings()['source_refs'])
        detail=current.get('panels',{}).get('AlteringPanelPopupUIInstance')
        if detail:
            actual=recipe_binding_for_selected_id(detail.get('altering_sheet_id'),current.get('metadata_sha256'))
            if not actual or actual['route_id']!=intent['route_id']:
                raise MapReadPending('current_return_recipe_identity_mismatch')
        result['confirmed']=bool(result.get('close') and not result['missing'])
    except (ValueError,RuntimeError) as exc:result['missing'].append(str(exc))
    return result


def _processing_prior_pending(snapshot):
    """Keep the genuine registration/navigation intent across a move-only probe."""
    pending=snapshot.get('pending_input') or {}
    receipt=pending.get('planner_receipt') or {}
    if receipt.get('phase')=='alter.pointer_probe':
        return (receipt.get('child_receipt') or {}).get('before_probe_pending') or {}
    return pending


def _processing_intent(node,snapshot):
    from .supply_selection import collect_supply_candidates
    candidates=[c for c in collect_supply_candidates(node,snapshot) if c['kind']=='native_alter']
    pending=_processing_prior_pending(snapshot);receipt=pending.get('planner_receipt') or {}
    selected=(snapshot.get('supply_selection') or {}).get('candidate') or {}
    requested=(snapshot.get('processing_route_id') or node.get('params',{}).get('recipe_id') or
               (receipt.get('child_receipt') or {}).get('recipe_id') or
               (selected.get('route_id') if selected.get('kind')=='native_alter' else None))
    if requested:candidates=[c for c in candidates if c['route_id']==requested]
    if len(candidates)!=1:raise MapReadPending('unique_or_explicit_current_processing_supply_route')
    matches=[b for b in processing_recipe_bindings()['bindings'] if b['route_id']==candidates[0]['route_id']]
    if len(matches)!=1:raise MapReadPending('source_processing_route_definition')
    return matches[0]


def _processing_owner(snapshot,frame):
    proof=snapshot.get('input_owner') or {};owner=proof.get('owner') or {}
    meta=dict(proof,session_id=owner.get('session_id'),character_id=owner.get('character_id'))
    if proof.get('continuous') is not True or proof.get('epoch') is None or not _fresh(meta,snapshot):
        raise MapReadPending('current_continuous_processing_input_epoch')
    if any(owner.get(k)!=snapshot.get(k) or frame.get(k)!=snapshot.get(k) for k in ('session_id','character_id')):
        raise MapReadPending('current_processing_input_owner_changed')
    selected=snapshot.get('window_preflight',{}).get('selected',{})
    if owner.get('pid')!=selected.get('pid') or owner.get('process_started_at')!=selected.get('process_started_at'):
        raise MapReadPending('current_processing_process_owner_changed')
    if not _fresh(dict(frame,source=frame.get('source','current supplied frame'),observed_at=frame.get('captured_at')),snapshot):
        raise MapReadPending('fresh_processing_frame_owner')
    pending=snapshot.get('pending_input') or {};receipt=pending.get('planner_receipt') or {}
    if receipt:
        child=receipt.get('child_receipt') or {}
        if child.get('input_epoch') is not None and child['input_epoch']!=proof['epoch']:
            raise MapReadPending('pending_processing_epoch_changed')
    return proof


def _physical_rect(candidate,scale):
    x,y,w,h=candidate['rect_ui']
    return [x*scale,y*scale,w*scale,h*scale]


def _processing_region_texts(image,ocr,regions,viewport,safe):
    """OCR only known current native regions; coordinates always come from VM."""
    if image is None or tuple(image.size)!=tuple(viewport):
        raise MapReadPending('current_processing_boundary_frame')
    if ocr is None:
        from .arbeit_targets import OfflineOCR
        ocr=OfflineOCR()
    sx,sy,sw,sh=safe;texts=[];seen_regions=set();seen_texts=set()
    for x,y,w,h in regions:
        box=(max(0,math.ceil(sx),math.ceil(x)),max(0,math.ceil(sy),math.ceil(y)),
             min(viewport[0],math.floor(sx+sw),math.floor(x+w)),
             min(viewport[1],math.floor(sy+sh),math.floor(y+h)))
        if box in seen_regions or box[0]>=box[2] or box[1]>=box[3]:continue
        if box==(0,0,*viewport):raise MapReadPending('current_processing_ocr_specific_native_region_required')
        seen_regions.add(box)
        for row in ocr.read(image.crop(box)):
            coords=row.get('box')
            if not isinstance(coords,(list,tuple)) or len(coords)!=4:continue
            if any(type(v) not in (int,float) or not math.isfinite(v) for v in coords):continue
            if not 0<=coords[0]<=coords[2]<=box[2]-box[0] or not 0<=coords[1]<=coords[3]<=box[3]-box[1]:continue
            shifted=[coords[0]+box[0],coords[1]+box[1],coords[2]+box[0],coords[3]+box[1]]
            key=(row.get('text'),*shifted)
            if key in seen_texts:continue
            seen_texts.add(key);texts.append(dict(row,box=shifted,source_crop=list(box)))
    return texts


def _recipe_scroll_observation(candidates,scale,safe,captions,texts):
    """Current visible rows adjudicate direction; wheel units are not pixels.

    MMML911 PCs2202..2388 contains one virtual vscroll with indexed children.
    Require one native scroll container enclosing at least one correctly
    rendered current row. Every subsequent notch requires a changed observed
    row position/identity in the executor, never a computed scroll offset.
    """
    sx,sy,sw,sh=safe;parents=[];norm=lambda text:re.sub(r'\s+','',text)
    for parent in candidates:
        if parent.get('tag')!='vscroll':continue
        x,y,w,h=_physical_rect(parent,scale)
        left,top,right,bottom=max(x,sx),max(y,sy),min(x+w,sx+sw),min(y+h,sy+sh)
        if right-left<=40*scale or bottom-top<=40*scale:continue
        # The retained virtual-scroll maskPadding=-20 excludes edge captions.
        left+=20*scale;right-=20*scale;top+=20*scale;bottom-=20*scale
        visible=[]
        for candidate in candidates:
            name=candidate.get('name','')
            if candidate.get('tag')!='__placeholder' or not name.isdecimal() or int(name) not in captions:continue
            rx,ry,rw,rh=_physical_rect(candidate,scale)
            labels=[t for t in texts if t.get('score',0)>=.85 and norm(t.get('text',''))==norm(captions[int(name)])
                and max(left,rx)<=t['box'][0]<=t['box'][2]<=min(right,rx+rw)
                and max(top,ry)<=t['box'][1]<=t['box'][3]<=min(bottom,ry+rh)]
            if len(labels)==1:visible.append(dict(index=int(name),rect_ui=candidate['rect_ui'],caption=labels[0]))
        if visible:parents.append((parent,[left,top,right-left,bottom-top],visible))
    if len(parents)!=1:return None
    parent,rect,visible=parents[0]
    signature=hashlib.sha256(json.dumps(visible,sort_keys=True,ensure_ascii=False).encode()).hexdigest()
    return dict(parent=parent,physical_rect=rect,visible=visible,digest=signature,
                basis='current MMML911 indexed rows and native vscroll; current caption containment')


def resolve_processing_current_scene(current,*,node,snapshot,frame,image=None,ocr=None,occlusion_reader=None,geometry_reader=None,scroll_reader=None):
    """Compose actual native geometry, current descriptor and input ownership.

    No stored order/scroll default is used. Current layout rectangles carry
    their native index/window ID so the common occlusion reader can adjudicate
    the actual point. Confirmed source facts and input readiness are separate.
    Exact source names/events take precedence over optional region-only OCR.
    """
    intent=_processing_intent(node,snapshot);owner=_processing_owner(snapshot,frame)
    definition=intent['current'];stage=None;panels=current.get('panels',{})
    scene=dict(source=current.get('source'),observed_at=current.get('observed_at'),session_id=snapshot['session_id'],
        character_id=snapshot['character_id'],confirmed=False,source_bound=False,recipe_id=intent['route_id'],facility_id=intent['group'],
        recipe_intent_bound=True,exact_recipe_confirmed=False,recipe_definition=intent,input_epoch=owner['epoch'],
        current_materials=[dict(name=m['name'],required=m['required']) for m in definition['materials']],
        recipe_source_refs=processing_current_sources()['source_refs']+processing_recipe_bindings()['source_refs'],
        missing=list(current.get('missing',[])),input_missing=[],slot_missing=[],input_target_ready=False,
        current_target=None,native_target=False,typed_current=current,game_inputs=0,live_verified=False)
    if current.get('stable') is not True or current.get('mmml_revision_sha256')!=processing_current_sources()['mmml_revision_sha256']:
        raise MapReadPending('current_processing_mmml_revision_and_stable_model')
    if current.get('table_revision_sha256')!=processing_recipe_bindings()['client_revision_sha256']:
        raise MapReadPending('current_processing_client_sheet_revision')
    layout=snapshot.get('layout') or {};viewport=layout.get('viewport');safe=layout.get('safe_rect')
    if not viewport or not safe:scene['input_missing'].append('current_processing_viewport_and_safe_area')
    scene.update(viewport=viewport,safe_rect=safe,orientation=layout.get('orientation'),canvas_scale=layout.get('canvas_scale'))
    facility=panels.get('AlteringFacilityPopupUIInstance');detail=panels.get('AlteringPanelPopupUIInstance');hub=panels.get('CraftingHubPopupUIInstance')
    panel=detail or facility or hub
    if not panel:raise MapReadPending('current_processing_panel')
    native_layout=panel.get('layout',{});scale=native_layout.get('scale');screen=native_layout.get('screen_ui',[])
    if not viewport or not scale or len(screen)!=2 or any(abs(screen[i]*scale-viewport[i])>2 for i in range(2)):
        scene['input_missing'].append('current_processing_mmml_viewport_matches_window')
    scene['input_missing'].extend(panel.get('layout_missing',[]))
    candidates=native_layout.get('candidates',[]);desired=[];caption=None;role=None;descriptor=None
    if facility:
        model=ProcessingModel(facility);body=model.body();scene['facility_name']=body.get('currentFacilityName')
        queue=model.children(body.get('processingQueue'))
        rings=[c for c in facility.get('layout',{}).get('candidates',[]) if c.get('tag')=='t-ProgressRing']
        if not 1<=len(queue)<=7 or len(rings)!=len(queue):
            scene['slot_missing'].append('current_generated_processing_queue_and_rendered_rings')
        else:scene.update(queue_count=len(queue),queue_capacity_basis='current processingQueue children each rendered once by MMML911/912')
        scene.update(move_visible=body.get('isMoveToFacilityVisible'),upgrade_visible=body.get('isUpgradeVisible'))
    else:
        body={};model=None
    if detail:
        selected=recipe_binding_for_selected_id(detail.get('altering_sheet_id'),current.get('metadata_sha256'))
        if not selected or selected['route_id']!=intent['route_id']:raise MapReadPending('current_selected_processing_recipe_differs_from_intent')
        desc=ProcessingModel(detail);data=desc.body();button=desc.values(data.get('confirmButton'))
        register_enabled=(button.get('text')=='가공하기' and button.get('disabled') is False
                          and button.get('onclick',{}).get('event_present') is True)
        if not register_enabled:scene['input_missing'].append('current_local_register_descriptor_enabled')
        stage='recipe_detail';role='processing.register';caption=button.get('text')
        descriptor=button
        # MMML889 PCs1420..1537: actual Area_ButtonCTA -> grid 56px 16px 1fr,
        # confirm occupies col3. Actual area already includes current layout.
        for candidate in candidates:
            if candidate.get('name')=='Area_ButtonCTA':
                c=deepcopy(candidate);x,y,w,h=c['rect_ui'];c['rect_ui']=[x+72,y,w-72,h];desired.append(c)
        scene.update(local_detail=True,register_enabled=register_enabled,exact_recipe_confirmed=True,
                     selected_altering_sheet_id=detail['altering_sheet_id'],production_amount=detail.get('production_amount'),
                     register_descriptor=deepcopy(button))
    elif facility:
        if body.get('isMoveToFacilityVisible') is True:
            button=model.values(body.get('moveToFacilityButton'))
            if not button.get('onclick',{}).get('event_present') or button.get('isActive') is not True:
                raise MapReadPending('current_facility_travel_enabled')
            stage='remote_facility';caption=button.get('text');role='processing.facility.travel'
            descriptor=button
            desired=[c for c in candidates if c.get('tag')=='t-UpgradeButton'];scene['remote']=True
        else:
            pending=_processing_prior_pending(snapshot);prior=(pending.get('planner_receipt') or {}).get('phase')
            if prior in ('alter.recipe_detail','alter.collect') and not (prior=='alter.collect' and body.get('isGetAll') is False):
                button=model.values(body.get('getAllButton'))
                if body.get('isMoveToFacilityVisible') is not False or body.get('isGetAll') is not True or not button.get('onclick',{}).get('event_present'):
                    raise MapReadPending('current_local_collect_descriptor_enabled')
                stage='collect';caption=button.get('text');role='processing.collect_all'
                descriptor=button
                desired=[c for c in candidates if c.get('name')=='progressBtn'];scene.update(facility_ready=True,collect_enabled=True)
            else:
                rows=model.children(body.get('itemRecipeList'));matches=[];ordered=[];captions={}
                for index,(_,row) in enumerate(rows):
                    item=model.values(row.get('itemData'));name=item.get('displayName');level=row.get('requiredFacilityLevelText')
                    if isinstance(name,str) and name:captions[index]=name
                    choices=[b for b in processing_recipe_bindings()['bindings'] if b['group']==intent['group'] and b['current']['display_name']==name]
                    if len(choices)>1 and isinstance(level,str):
                        match=re.fullmatch(r'(?:Lv\.?|레벨)\s*(\d+)',level.strip(),re.I)
                        if match:choices=[b for b in choices if b['current']['required_facility_level']==int(match.group(1))]
                    ordered.append(choices[0]['route_id'] if len(choices)==1 else None)
                    if len(choices)==1 and choices[0]['route_id']==intent['route_id']:
                        if row.get('isFacilityLevelValid') is not True or row.get('isSkillLevelValid') is not True or not row.get('onclick',{}).get('event_present'):
                            raise MapReadPending('current_recipe_selection_requirement')
                        matches.append((index,name,row))
                if len(matches)!=1:raise MapReadPending('current_recipe_array_exact_identity')
                index,caption,descriptor=matches[0];role='processing.recipe.'+intent['route_id'];stage='recipe_list'
                desired=[c for c in candidates if c.get('tag')=='__placeholder' and c.get('name')==str(index)]
                scene.update(ordered_ids=ordered,selected_array_index=index,recipe_row_captions=captions,recipe_array_basis='current itemRecipeList; unknown other entries preserved')
    else:
        model=ProcessingModel(hub);data=model.body();category=model.values(data.get('topCategoryGroup'))
        buttons=model.children(category.get('buttons'));selected=category.get('currentButton')
        if selected!='Altering':
            buttons=[r for _,r in buttons if r.get('name')=='Altering' and r.get('onclick',{}).get('event_present')]
            if len(buttons)!=1:raise MapReadPending('current_altering_tab_descriptor')
            caption=buttons[0].get('displayText');stage='hub_tabs';role='processing.tab.Altering'
            descriptor=buttons[0]
            desired=[c for c in candidates if c.get('tag')=='t-ButtonPreset_MainTab']
        else:
            tab=model.values(data.get('alteringTabData'));groups={'metal':'금속 가공','wood':'목재 가공','cloth':'옷감 가공','leather':'가죽 가공','medicine':'약품 가공','food':'식재료 가공'}
            facilities=[r for _,r in model.children(tab.get('facilities')) if r.get('displayName')==groups[intent['group']] and r.get('onclick',{}).get('event_present')]
            if len(facilities)!=1:raise MapReadPending('current_exact_facility_descriptor')
            caption=facilities[0]['displayName'];stage='facility_list';role='processing.facility.'+intent['group']
            descriptor=facilities[0]
            desired=[c for c in candidates if c.get('tag')=='t-FacilitySlot']
    scene.update(stage=stage,source_bound=True,settled=True,confirmed=not scene['missing'],facts_confirmed=not scene['missing'],
                 frame_id=frame.get('frame_id',frame.get('observation_id')),owner_proof=owner,
                 current_action_descriptor=deepcopy(descriptor))
    if scene['input_missing']:return scene
    try:
        return _bind_processing_input(scene,native_layout,panel,candidates,desired,caption,role,stage,
                                      snapshot,frame,image,ocr,occlusion_reader,geometry_reader,node,scroll_reader)
    except (MapReadPending,OSError,ValueError) as exc:
        # A missing input target cannot erase current selected recipe/material
        # facts. Planners can start material recovery, but cannot submit input.
        scene['input_missing'].append(str(exc))
        return scene


def _bind_processing_input(scene,native_layout,panel,candidates,desired,caption,role,stage,
                           snapshot,frame,image,ocr,occlusion_reader,geometry_reader=None,node=None,scroll_reader=None):
    scale=native_layout['scale'];viewport=scene['viewport'];safe=scene['safe_rect']
    current=scene['typed_current']
    if stage=='recipe_list':
        try:
            from .recipe_scroll import observe_recipe_rows
            observe_recipe_rows(scene,panel,frame,target_index=scene['selected_array_index'],
                                row_tags=['__placeholder'],source_function=943)
        except (ImportError,MapReadPending,ValueError,KeyError) as exc:
            # A missing current row observation is not a wheel effect. Existing
            # legacy boundary observation may still provide it below.
            scene.setdefault('scroll_observation_missing',[]).append(str(exc))
    if geometry_reader is not None:
        # The model callback selects the exact current runtime button. Its
        # native Graphic supplies geometry; a placeholder/CTA area is not a
        # substitute when the current source has not produced that control.
        event=(scene.get('current_action_descriptor') or {}).get('onclick') or {}
        if event.get('event_present') is not True or type(event.get('event_id')) is not int:
            raise MapReadPending('current_processing_exact_model_event_id')
        exact=[]
        for candidate in candidates:
            attrs=candidate.get('runtime_attributes') or {};actual=attrs.get('onclick') or {}
            if (candidate.get('callback_source')=='current LayoutVM.elementIdFromElementIndex/runtimeForest'
                and candidate.get('model_event_id')==event['event_id'] and actual.get('event_present') is True
                and actual.get('event_id')==event['event_id']
                and candidate.get('button_pointer') and attrs.get('disabled',False) is False):
                exact.append(candidate)
        if len(exact)==1:
            try:
                return geometry_reader(scene,deepcopy(exact[0]),panel,node,snapshot,frame,role)
            except MapReadPending as exc:
                if stage!='recipe_list' or str(exc)!='current_processing_native_button_clipped' or scroll_reader is None:raise
                return scroll_reader(scene,panel,node,snapshot,frame)
        if stage=='recipe_list' and not exact and scroll_reader is not None:
            return scroll_reader(scene,panel,node,snapshot,frame)
        # A non-rendered recipe row can still expose the source scroll boundary.
        # Scroll clearance needs its own actual supplier; no button proof is
        # borrowed for the scroll container. Other stages fail closed here.
        if stage!='recipe_list' or len(exact)>1:
            raise MapReadPending('current_processing_unique_runtime_button_callback')
    if not isinstance(caption,str) or not caption:raise MapReadPending('current_processing_action_caption')
    # Exact MMML names have a pinned single control binding. Generic repeated
    # tags still require current callback identity or a caption within that
    # known native control; OCR never supplies an alternate rectangle.
    source_named=stage in ('recipe_detail','collect') and len(desired)==1
    texts=[]
    if not source_named:
        regions=[_physical_rect(c,scale) for c in desired]
        if stage=='recipe_list':
            regions.extend(_physical_rect(c,scale) for c in candidates if c.get('tag')=='vscroll')
        texts=_processing_region_texts(image,ocr,regions,viewport,safe)
    normalize=lambda t:re.sub(r'\s+','',t)
    scroll=_recipe_scroll_observation(candidates,scale,safe,scene.get('recipe_row_captions',{}),texts) if stage=='recipe_list' else None
    if scroll:
        scene['scroll_observation']=dict(source='current native virtual recipe scroll',observed_at=current['observed_at'],
            character_id=snapshot['character_id'],session_id=snapshot['session_id'],window_id=native_layout['window_id'],
            digest=scroll['digest'],visible_indices=[v['index'] for v in scroll['visible']],rows=scroll['visible'])
    actual=[r for r in texts if r.get('score',0)>=.85 and normalize(r.get('text',''))==normalize(caption)]
    matched=[]
    for candidate in desired:
        x,y,w,h=_physical_rect(candidate,scale);sx,sy,sw,sh=safe
        if w<=0 or h<=0 or not (sx<=x<x+w<=sx+sw and sy<=y<y+h<=sy+sh):continue
        labels=[r for r in actual if x<=r['box'][0]<=r['box'][2]<=x+w and y<=r['box'][1]<=r['box'][3]<=y+h]
        if source_named:
            matched.append((candidate,[x,y,w,h],dict(basis='current source-named MMML control and current callback descriptor',
                name=candidate.get('name'),text=caption,ocr_used=False)))
        elif len(labels)==1:matched.append((candidate,[x,y,w,h],labels[0]))
    delta=None
    if len(matched)!=1:
        if not matched and scroll:
            indices=[v['index'] for v in scroll['visible']];target=scene['selected_array_index']
            delta=-120 if target>max(indices) else 120 if target<min(indices) else None
        if delta is None:raise MapReadPending('current_processing_caption_inside_unique_native_target')
        candidate=deepcopy(scroll['parent']);rect=scroll['physical_rect'];candidate['rect_ui']=[n/scale for n in rect]
        label=dict(basis=scroll['basis'],visible_indices=indices);stage='recipe_scroll';role='processing.recipes.scroll'
    else:candidate,rect,label=matched[0]
    if occlusion_reader is None:raise MapReadPending('current_processing_target_raycast_occlusion_supplier')
    occlusion=occlusion_reader(window_id=native_layout['window_id'],rect_ui=candidate['rect_ui'],target_ui_pointer=panel.get('ui_pointer'))
    if occlusion.get('occluded') is not False or not occlusion.get('source'):
        raise MapReadPending('current_processing_target_not_proven_clear:'+str(occlusion.get('reason')))
    scene.update(stage=stage,input_target_ready=True,native_target=True,
        current_target=dict(kind='click',role=role,point=[rect[0]+rect[2]/2,rect[1]+rect[3]/2],source_rect=[rect[0],rect[1],rect[0]+rect[2],rect[1]+rect[3]],
            window_id=native_layout['window_id'],layout_index=candidate['index'],rect_ui=candidate['rect_ui'],ui_pointer=panel.get('ui_pointer'),
            safe_rect=safe,source='current MMML processing model/layout',caption_evidence=label,occlusion=occlusion),
        frame_id=frame.get('observation_id'))
    if delta is not None:scene['current_target'].update(kind='scroll',delta=delta)
    return scene


def _processing_pointer_view(session,scene,candidate,panel,frame,role):
    """Actual current callback and ButtonControl, sharing the caller's reader."""
    from .quest_ui_actions import quest_ui_session
    pointer=panel['ui_pointer'];pointer=int(pointer,16) if isinstance(pointer,str) else pointer
    current=scene['typed_current'];layout=panel['layout']
    view=dict(layout,candidates=[candidate],ui_pointer=pointer,ui_type=panel.get('ui_type'),
        source_hash=processing_current_sources()['mmml_source_sha256'],target_id=role,
        pointer_semantics=dict(source_family='native_alter',recipe_id=scene['recipe_id'],stage=scene['stage'],
            role=role,panel_id=pointer,descriptor=deepcopy(scene['current_action_descriptor']),
            table_revision=current['table_revision_sha256'],mmml_revision=current['mmml_revision_sha256']))
    proxy=quest_ui_session(session)
    proxy.attach_current_geometry(view,frame)
    candidate=view['candidates'][0]
    if candidate.get('rect_source')!='current-native-output':
        raise MapReadPending('current_processing_exact_native_button_geometry')
    x,y,w,h=candidate['rect_client'];sx,sy,sw,sh=scene['safe_rect']
    if not (sx<=x<x+w<=sx+sw and sy<=y<y+h<=sy+sh):
        raise MapReadPending('current_processing_native_button_clipped')
    return candidate,view


def _processing_pointer_action(session,scene,candidate,panel,node,snapshot,frame,role):
    from .quest_ui_actions import prepare_pointer_proposal,pointer_effect_contract
    actual_frame=dict(frame,input_owner=snapshot['input_owner'])
    for key in ('pending_input','pointer_probe_receipt'):
        if key not in actual_frame and key in snapshot:actual_frame[key]=snapshot[key]
    candidate,view=_processing_pointer_view(session,scene,candidate,panel,actual_frame,role)
    return _finish_processing_pointer_action(scene,candidate,view,panel,node,snapshot,actual_frame,role)


def _finish_processing_pointer_action(scene,candidate,view,panel,node,snapshot,actual_frame,role):
    from .quest_ui_actions import prepare_pointer_proposal,pointer_effect_contract
    frame=actual_frame
    pointer=prepare_pointer_proposal(candidate,view,actual_frame)
    context=pointer['probe_context']
    if pointer['input_kind']=='pointer_probe':
        # Preserve the genuine pre-probe receipt, not a synthesized registration.
        context.update(source_guard_kind='recipe-scroll' if scene['stage']=='recipe_scroll' else 'processing-input',processing_request=dict(
            node=deepcopy(node),recipe_id=scene['recipe_id'],
            before_probe_pending=deepcopy(_processing_prior_pending(snapshot))))
    owner=frame.get('current_quest') or snapshot.get('current_quest') or {}
    if (owner.get('confirmed') is not True or owner.get('quest_id')!=node.get('quest_id')
            or not _fresh(owner,snapshot)):
        raise MapReadPending('current_processing_pointer_effect_quest_occurrence')
    effect=pointer_effect_contract(pointer,owner,node,actual_frame,view['pointer_state']['observed_at'])
    source=('current source-bound processing scroll handler and exact scroll processor receiver'
            if scene['stage']=='recipe_scroll' else 'current source-bound processing button and final game pointer targets')
    effect['effect_proofs']=[dict(p,source=source)
                            for p in effect['effect_proofs']]
    x,y,w,h=candidate['rect_client']
    target=dict(pointer,kind=pointer['input_kind'],role=role,point=pointer['point_client'],
        source_rect=[x,y,x+w,y+h],safe_rect=scene['safe_rect'],rect_ui=candidate['rect_ui'],
        rect_source=candidate['rect_source'],native_geometry_proof=candidate['native_geometry_proof'],
        window_id=candidate['window_id'],layout_index=candidate['index'],ui_pointer=candidate['ui_pointer'],
        input_epoch=scene['input_epoch'],source_guard_kind=context['source_guard_kind'],
        source=source,
        source_refs=context['source_hashes'],caption_evidence=dict(ocr_used=False,basis='exact current model/runtime callback identity'))
    scene.update(input_target_ready=True,native_target=True,current_target=target,pointer_effect=effect,
        source_revision=context['source_revision'],source_guard_kind=context['source_guard_kind'])
    return scene


def _processing_scroll_action(session,scene,panel,node,snapshot,frame):
    try:
        from .recipe_scroll import processing_scroll_view
    except ImportError as exc:
        raise MapReadPending('current_recipe_scroll_source_supplier_unavailable') from exc
    actual_frame=dict(frame,input_owner=snapshot['input_owner'])
    for key in ('pending_input','pointer_probe_receipt'):
        if key not in actual_frame and key in snapshot:actual_frame[key]=snapshot[key]
    candidate,view,observation=processing_scroll_view(session,scene,panel,actual_frame)
    scene.update(stage='recipe_scroll',scroll_observation=observation)
    return _finish_processing_pointer_action(scene,candidate,view,panel,node,snapshot,actual_frame,'processing.recipes.scroll')


def processing_pointer_scene(session,binding,frame):
    """Last-moment scene_reader for make_pointer_guard, using its open session.

    ``session_factory=CurrentProcessingSession`` is required. Rebuild all current
    model/callback/geometry facts; retained request data selects the original
    recipe and phase only. No saved descriptor is treated as a current fact.
    """
    request=binding.get('processing_request') or {}
    if binding.get('source_guard_kind')!='processing-input' or not request.get('node'):
        raise ValueError('current processing pointer request identity required')
    physical=frame['physical_client_rect'];viewport=[physical[2]-physical[0],physical[3]-physical[1]]
    safe=(frame.get('current_context') or {}).get('safe_rect')
    if not isinstance(safe,(list,tuple)) or len(safe)!=4:
        raise MapReadPending('current_processing_dispatch_safe_area')
    # Frame safe_rect is LTRB; the processing planner consumes LTWH.
    safe=[safe[0],safe[1],safe[2]-safe[0],safe[3]-safe[1]]
    owner=frame['input_owner'];identity=owner['owner']
    snapshot=dict(now=frame['captured_at'],session_id=frame['session_id'],character_id=frame['character_id'],
        window_preflight={'selected':{'pid':identity['pid'],'process_started_at':identity['process_started_at']}},
        input_owner=owner,layout={'viewport':viewport,'safe_rect':safe},processing_route_id=request['recipe_id'],
        pending_input=request.get('before_probe_pending'),current_quest=frame.get('current_quest'))
    current=session.current_processing();collected=[]
    def collect(scene,candidate,panel,node,snapshot,frame,role):
        pair=_processing_pointer_view(session,scene,candidate,panel,frame,role)
        collected.append(pair)
        return scene
    result=resolve_processing_current_scene(current,node=request['node'],snapshot=snapshot,frame=frame,
        geometry_reader=collect)
    if result.get('input_missing') or len(collected)!=1:
        raise MapReadPending('current_processing_dispatch_target_changed:'+str(result.get('input_missing')))
    return collected[0]


class CurrentProcessingSceneReader:
    """Concrete read-only supplier. Import/default construction opens nothing.

    Pass ``allow_process_read=...`` only from the app's existing opt-in. This
    supplier confirms source facts independently of ``input_target_ready``.
    Source/owner mismatches remain unconfirmed; missing geometry or disabled
    controls do not erase the current recipe. CLI success is never UI readiness.
    """
    def __init__(self,*,allow_process_read=False,session_factory=CurrentProcessingSession,ocr=None,occlusion_reader=None):
        self.allow_process_read=allow_process_read;self.session_factory=session_factory
        self.ocr=ocr;self.occlusion_reader=occlusion_reader

    def __call__(self,*,node,snapshot,session=None,image=None,frame=None):
        empty=dict(source='current typed processing UI',observed_at=None,confirmed=False,source_bound=False,
                   session_id=snapshot.get('session_id'),character_id=snapshot.get('character_id'),game_inputs=0,missing=[])
        if not self.allow_process_read:return dict(empty,missing=['read_only_processing_process_access_disabled'])
        window=snapshot.get('window_preflight',{}).get('selected',{});pid=window.get('pid')
        if type(pid) is not int or pid<=0 or not all(snapshot.get(k) for k in ('session_id','character_id')):
            return dict(empty,missing=['current_processing_window_and_character_owner'])
        try:
            if frame is not None:
                owner=_processing_owner(snapshot,frame)
                hud=snapshot.get('field_hud') or {};context=frame.get('current_context') or snapshot.get('current_context') or {}
                if _fresh(hud,snapshot) and hud.get('confirmed') is True and context.get('field_hud') is True and context.get('profile',{}).get('visual_confirmed') is True:
                    intent=_processing_intent(node,snapshot)
                    result=dict(empty,source='SourceLife.visit field entry + current HUD/owner',observed_at=frame['captured_at'],
                        confirmed=True,source_bound=True,native_target=True,stage='field',recipe_id=intent['route_id'],facility_id=intent['group'],
                        recipe_definition=intent,recipe_intent_bound=True,exact_recipe_confirmed=False,input_epoch=owner['epoch'],
                        owner_proof=owner,current_target=dict(kind='key',role='processing.menu',key='K'),missing=[],
                        recipe_source_refs=processing_recipe_bindings()['source_refs'])
                    if snapshot.get('material_return_requested') is True or snapshot.get('processing_cleanup_requested') is True:
                        def absent(opened):
                            opened._guards=[];panels=opened.active_processing_panels(allow_empty=True)
                            if panels:raise MapReadPending('processing_return_owned_panels_still_active')
                            if any(opened.reader.read(a,len(raw))!=raw for a,raw in opened._guards):
                                raise MapReadPending('processing_return_active_panels_changed_during_read')
                        if session is not None:absent(session)
                        else:
                            with self.session_factory(pid) as opened:absent(opened)
                        result['return_ui']=dict(source='current active UIInstanceService processing-panel enumeration + current HUD',
                            source_family='native_alter',source_revision=processing_current_sources()['mmml_source_sha256'],
                            source_checks_revision=processing_current_sources()['mmml_revision_sha256'],observed_at=datetime.now(timezone.utc).isoformat(),
                            source_refs=processing_current_sources()['source_refs'],session_id=snapshot['session_id'],character_id=snapshot['character_id'],
                            process_started_at=window.get('process_started_at'),input_epoch=owner['epoch'],owner_proof=owner,recipe_id=intent['route_id'],
                            surface='hud',panel_id=None,stable=True,confirmed=True,active_owned_panels=[],panels_absent_confirmed=True,
                            key_receiver_confirmed=False,missing=[])
                        result.update(native_target=False,current_target=None)
                    return result
            def resolve(opened):
                current=opened.current_processing()
                if frame is None:return processing_scene_from_current(current,snapshot)
                occlusion=self.occlusion_reader or getattr(opened,'read_target_occlusion',None)
                return_ui=bind_processing_return_ui(current,node,snapshot,frame)
                from .material_close import attach_material_close_geometry
                attach_material_close_geometry(opened,return_ui,frame)
                try:
                    result=resolve_processing_current_scene(current,node=node,snapshot=snapshot,frame=frame,image=image,ocr=self.ocr,occlusion_reader=occlusion,
                        geometry_reader=lambda *args:_processing_pointer_action(opened,*args),
                        scroll_reader=lambda *args:_processing_scroll_action(opened,*args))
                except MapReadPending as exc:
                    result=processing_scene_from_current(current,snapshot)
                    result['missing'].append(str(exc))
                result['return_ui']=return_ui
                if any(opened.reader.read(a,len(raw))!=raw for a,raw in opened._guards):
                    raise MapReadPending('current_processing_ui_changed_during_boundary_read')
                return result
            if session is not None:return resolve(session)
            with self.session_factory(pid) as opened:return resolve(opened)
        except (OSError,ValueError,RuntimeError) as exc:
            return dict(empty,missing=[str(exc)])


_NAMES = {
    'medicine-light-ghost':'광휘의 결정(유령 반딧불이)',
    'medicine-fire':'불꽃의 결정(석양나비)',
    'medicine-ice':'얼음의 결정(흰얼음풍뎅이)',
    'medicine-lightning':'전기의 결정(낙엽나방)',
    'medicine-seal':'봉인의 결정(황혼잠자리)',
    'medicine-light-stag':'광휘의 결정(달무늬 사슴벌레)',
}


def parse_works(envelope):
    if not envelope.get('ok') or not isinstance(envelope.get('body'),dict):
        raise ValueError('official_altering_works_unavailable')
    body=envelope['body']
    if not isinstance(body.get('works'),list):
        raise ValueError('official_altering_works_schema')
    result=[]
    for raw in body['works']:
        if not isinstance(raw,dict) or any(not isinstance(raw.get(k),str) or not raw[k] for k in ('DisplayName','FacilityName')):
            raise ValueError('official_work_name_or_facility')
        state=raw.get('State');complete=raw.get('IsCompleted');remaining=raw.get('RemainingSeconds')
        if state not in ('NotStarted','InProgress','Completed') or type(complete) is not bool or complete!=(state=='Completed'):
            raise ValueError('official_work_completion_disagrees')
        if type(remaining) not in (int,float) or not math.isfinite(remaining) or remaining<0 or (complete and remaining!=0):
            raise ValueError('official_work_remaining_time_invalid')
        result.append(dict(name=raw['DisplayName'],facility=raw['FacilityName'],state=state,
                           complete=complete,remaining_seconds=remaining))
    if type(body.get('completedCount')) is not int or body['completedCount']!=sum(r['complete'] for r in result):
        raise ValueError('official_work_completed_count_disagrees')
    return result


def parse_recipes(envelope):
    if not envelope.get('ok') or not isinstance(envelope.get('body'),dict) or not isinstance(envelope['body'].get('items'),list):
        raise ValueError('official_alterable_items_unavailable')
    rows=envelope['body']['items']
    if any(not isinstance(r,dict) or not isinstance(r.get('DisplayName'),str) for r in rows):
        raise ValueError('official_alterable_items_schema')
    return deepcopy(rows)


def multiset(rows):
    counts=Counter((r['name'],r['facility']) for r in rows)
    return [dict(name=name,facility=facility,count=count) for (name,facility),count in sorted(counts.items())]


def _meta(envelope,snapshot,source):
    return dict(source=source,observed_at=envelope.get('observed_at'),
                session_id=snapshot['session_id'],character_id=snapshot['character_id'],
                response_sha256=envelope.get('response_sha256'))


class ProcessingObserver:
    """Queries only existing READS; scene_reader must supply actual UI evidence.

    read(node, snapshot, receipt=None, scene=None) returns parsed current data
    even when a scene provider is absent. Missing geometry stays unconfirmed.
    A supplied scene is trusted only as the output of the app's typed current
    UI reader, with its own source/time/owner; historical fixtures are tests.
    """
    def __init__(self,cli,scene_reader=None):
        self.cli,self.scene_reader=cli,scene_reader

    def read(self,node,snapshot,receipt=None,scene=None,*,image=None,frame=None):
        envelopes={name:self.cli.query(name) for name in
                   ('get_alterable_items','get_altering_works','get_items','get_activity')}
        missing=[]
        try:
            recipes=parse_recipes(envelopes['get_alterable_items'])
            works=parse_works(envelopes['get_altering_works'])
        except ValueError as exc:
            return dict(ok=False,native_processing=None,missing=[str(exc)],reads=envelopes)
        from .game_observer import inventory_from_read
        inventory=inventory_from_read(envelopes['get_items'],snapshot['character_id'],snapshot['session_id'])
        metas={name:_meta(env,snapshot,name) for name,env in envelopes.items()}
        for name,meta in metas.items():
            if not envelopes[name].get('ok') or not _fresh(meta,snapshot):
                missing.append('fresh_'+name)
        if not inventory:
            missing.append('current_complete_consumable_inventory')
        if scene is None and self.scene_reader is not None:
            scene=self.scene_reader(node=node,snapshot=snapshot,image=image,frame=frame)
        scene=scene or {}
        missing.extend(scene.get('missing',[]))
        from .supply_selection import collect_supply_candidates
        candidates=[c for c in collect_supply_candidates(node,snapshot) if c['kind']=='native_alter']
        # The user's/current UI exact recipe selects a variant. Never choose a
        # material variant merely because it is first in a historical catalog.
        candidates=[c for c in candidates if c['route_id']==scene.get('recipe_id')]
        if len(candidates)!=1:
            missing.append('current_exact_recipe_ui_binding')
        if not _fresh(scene,snapshot) or scene.get('confirmed') is not True or scene.get('source_bound') is not True:
            missing.append('current_processing_ui_scene')
        if scene.get('stage') not in ('field','hub_tabs','facility_list','recipe_list','recipe_scroll','recipe_detail','remote_facility','collect'):
            missing.append('current_processing_ui_stage')
        native=dict(scene,source=scene.get('source','current_processing_ui_unavailable'),
                    session_id=snapshot['session_id'],character_id=snapshot['character_id'],
                    observed_at=scene.get('observed_at'),confirmed=False,
                    works_observation=metas['get_altering_works'],works=works,work_multiset=multiset(works),
                    stable_work_ids_available=False,recipe_read=metas['get_alterable_items'],
                    activity_read=metas['get_activity'],inventory_observation=inventory)
        if len(candidates)==1:
            candidate=candidates[0];recipe=candidate['recipe']
            definition=scene.get('recipe_definition',{}).get('current',{})
            cli_name=definition.get('display_name') or _NAMES.get(recipe['id'],recipe['name'])
            matches=[r for r in recipes if r['DisplayName']==cli_name]
            if len(matches)!=1 or matches[0].get('ProducedPerWork')!=recipe['output_per_batch']:
                missing.append('current_exact_recipe_variant_and_output_join')
            else:
                native.update(recipe_id=recipe['id'],facility_id=recipe['group'],
                    recipe_intent_bound=True,exact_recipe_confirmed=scene.get('exact_recipe_confirmed') is True,current_recipe=matches[0],
                    recipe_identity_binding='exact CLI DisplayName/ProducedPerWork joined to source route; route ID is not API data')
            # Current-version table materials are admissible only with the
            # selected typed recipe binding above, or an explicit UI supplier.
            # They remain a static requirement definition, not inventory proof.
            consumes=scene.get('stage') in ('recipe_detail','collect')
            if consumes and scene.get('current_materials')!=recipe['materials']:
                missing.append('current_recipe_materials_definition')
            if consumes and (type(scene.get('queue_count')) is not int or not 1<=scene.get('queue_count',0)<=7):
                native.setdefault('slot_missing',[]).append('current_generated_slot_capacity')
            current_facility=scene.get('facility_name')
            facility_works=[r for r in works if r['facility']==current_facility] if current_facility else []
            if consumes and not current_facility:
                native.setdefault('slot_missing',[]).append('current_exact_facility_name')
            native['free_slot_confirmed']=bool(current_facility and type(scene.get('queue_count')) is int
                                              and len(facility_works)<scene['queue_count'])
            native['current_recipe_works']=[r for r in facility_works if r['name']==cli_name]
            native['current_facility_works']=facility_works
            # Official schema cannot name stable physical slots. Preserve the
            # complete multiset baseline for later intent-scoped reconciliation.
            native['registration_baseline']=dict(metas['get_altering_works'],
                multiset=multiset(works),recipe_name=cli_name,facility_name=current_facility,
                same_recipe_count=sum(r['name']==cli_name for r in facility_works),
                basis='official_work_multiset_not_stable_ids')
            if inventory and _fresh(inventory,snapshot):
                enough=True;material_observations=[]
                for material in recipe['materials']:
                    selected=[r for r in inventory['items'] if r['name']==material['name'] and r['scope']=='bag']
                    categories={r['category'] for r in selected};identity_resolved=len(categories)==1 or not selected
                    available=sum(r['quantity']-r['locked_quantity']-r.get('reserved_quantity',0) for r in selected) if identity_resolved else None
                    material_observations.append(dict(requirement=deepcopy(material),item=dict(name=material['name'],category=next(iter(categories)) if len(categories)==1 else None),
                        observed_items=deepcopy(selected),available_quantity=available,
                        shortage=max(0,material['required']-available) if available is not None else None,
                        identity_resolved=identity_resolved,inventory_source=inventory['source'],inventory_observed_at=inventory['observed_at'],
                        source_refs=scene.get('recipe_source_refs',[])))
                    if available is None:
                        enough=False;continue
                    if available<material['required']:
                        enough=False
                native['materials_confirmed']=enough
                native['material_observations']=material_observations
                native['material_shortages']=[r for r in material_observations if r['shortage'] is None or r['shortage']>0]
                if enough and scene.get('stage')=='recipe_detail':
                    native['uses_storage']=False
                    native['materials_scope_basis']='fresh get_items unlocked unreserved bag alone covers source-bound recipe requirements'
            else:
                native['materials_confirmed']=False
        native['confirmed']=not missing
        return dict(ok=not missing,native_processing=native,inventory=inventory,
                    missing=list(dict.fromkeys(missing)),reads=envelopes,
                    game_input_sent=False)
