"""Typed current-root quest reads; OFF unless explicitly enabled by the caller.

No heap scan, game invocation, capture, input, writes, elevation or helper launch.
Payload decoding is copied from recorded mini27 methods; live roots are followed
through the current singular context already bundled for map observations.
"""
from copy import deepcopy
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import struct
import threading

from .paths import root
from .vendor.map_readonly_reader import Reader, QuestReadCancelled, file_sha256
from .vendor.map_character_context import active_context, _class
from .vendor.map_character_identity import CharacterMemoryReader
from .vendor.map_resources import load_rule


class QuestPayloadReader(Reader):
    def scalar_fields(self, pointer, shape, allowed):
        result = {}
        for field in shape['fields']:
            name, offset, kind = field['name'],field['offset'],field['type_code']
            if name not in allowed or offset < 16:
                continue
            address = pointer + offset
            if kind == 2:
                value = self.read(address,1)[0]
                if value not in (0,1):
                    raise ValueError('Quest boolean invalid.')
                result[name] = bool(value)
            elif kind in (8,9):
                result[name] = self.i32(address) if kind == 8 else self.u32(address)
            elif kind in (10,11,17):
                # Selected ID/time wrapper fields are independently eight
                # bytes; enums are explicitly handled by their field names.
                result[name] = self.i32(address) if name in ('<BlockReason>k__BackingField',) else self.u64(address)
            elif kind == 14:
                result[name] = self.string(self.u64(address))
        return result

    def quest_list(self, pointer):
        if not pointer:
            return []
        shape = self.layout(self.u64(pointer))
        if shape['name'] == 'QuestObjective[]':
            size = self.u64(pointer+24)
            if not 0 <= size <= 1024 or shape['element_size'] != 8:
                raise ValueError('Direct quest objective array bounds invalid.')
            return list(struct.unpack('<'+'Q'*size,self.read(pointer+32,8*size))) if size else []
        if shape['name'] == 'List`1' and shape['namespace'] == 'System.Collections.Generic':
            fields = {f['name'].lstrip('_'):f for f in shape['fields']}
            if not {'items','size'} <= set(fields):
                raise ValueError('Quest objective generic list fields not verified.')
            size = self.i32(pointer+fields['size']['offset'])
            array = self.u64(pointer+fields['items']['offset'])
            if not 0 <= size <= 1024:
                raise ValueError('Quest objective generic list size invalid.')
            if not size:
                return []
            array_class = self.u64(array)
            if self.class_name(array_class)[0] != 'QuestObjective[]' or self.u32(array_class+0x100) != 8 or self.u64(array+24) < size:
                raise ValueError('Generic quest objective array not verified.')
            return list(struct.unpack('<'+'Q'*size,self.read(array+32,8*size)))
        if shape['namespace'] != 'Silvervine.ManualMemory' or not shape['name'].startswith(('MMList','MMReadOnlyList')):
            raise ValueError('Quest objective container unsupported: '+shape['namespace']+'.'+shape['name'])
        if not shape['fields']:
            shape = self.layout(int(shape['parent_class'],16))
        fields = {f['name']:f for f in shape['fields']}
        if not {'array','version','size'} <= set(fields):
            raise ValueError('Quest list fields not verified.')
        size = self.i32(pointer+fields['size']['offset'])
        if not 0 <= size <= 1024:
            raise ValueError('Quest objective list size invalid.')
        array = self.u64(pointer+fields['array']['offset'])
        if not size:
            return []
        array_class = self.u64(array)
        if self.class_name(array_class)[0] != 'QuestObjective[]' or self.u32(array_class+0x100) != 8 or self.u64(array+24) < size:
            raise ValueError('Quest objective array layout not verified.')
        return list(struct.unpack('<'+'Q'*size,self.read(array+32,8*size)))

    def decode_objectives(self, quest_pointer, field_pointer, expected_class, catalog_objectives):
        decoded = []
        for object_pointer in self.quest_list(field_pointer):
            if self.u64(object_pointer) != expected_class:
                raise ValueError('Quest objective object class differs from metadata.')
            shape = self.layout(expected_class)
            fields = {f['name']:f for f in shape['fields']}
            if self.u64(object_pointer+fields['<OwnerActiveQuest>k__BackingField']['offset']) != quest_pointer:
                raise ValueError('Quest objective belongs to another quest.')
            index = self.i32(object_pointer+fields['<Index>k__BackingField']['offset'])
            counter_field = fields['<Count>k__BackingField']
            count = self.i32(object_pointer+counter_field['offset']) if counter_field['type_code'] == 8 else self.unpack(object_pointer+counter_field['offset'],'<q')
            goal_field = fields['goal']
            goal = None
            if goal_field['type_code'] == 8:
                goal = self.i32(object_pointer+goal_field['offset'])
            elif goal_field['type_code'] == 10:
                goal = self.unpack(object_pointer+goal_field['offset'],'<q')
            elif goal_field['type_code'] == 21:
                # The cached Nullable<int> is decoded only for valid bool
                # tags. Empty cache is not evaluated by calling game code.
                tag = self.read(object_pointer+goal_field['offset'],1)[0]
                if tag == 1:
                    goal = self.i32(object_pointer+goal_field['offset']+4)
            static = catalog_objectives[index] if 0 <= index < len(catalog_objectives) else {}
            decoded.append(dict(index=index,count=count,goal_cached=goal,
                                goal_memory_raw=goal,
                                description_static=static.get('text'),goal_static=static.get('goal'),
                                kind_static=static.get('kind')))
        return decoded

    def decode_wait_fields(self, pointer):
        """Capture only scalar fields of typed conditions in this quest list.

        Do not invoke requirement checkers or traverse unrelated references.
        Unknown wrapper bits stay explicitly raw, not a guessed status.
        """
        if not pointer:
            return []
        shape = self.layout(self.u64(pointer))
        if shape['name'] == 'List`1' and shape['namespace'] == 'System.Collections.Generic':
            fields = {f['name'].lstrip('_'):f for f in shape['fields']}
            if not {'items','size'} <= set(fields):
                raise ValueError('Wait requirement list layout incomplete.')
            count = self.i32(pointer+fields['size']['offset'])
            array = self.u64(pointer+fields['items']['offset'])
        elif shape['name'].endswith('[]'):
            array,count = pointer,self.u64(pointer+24)
        else:
            raise ValueError('Wait requirement container not decoded: '+shape['name'])
        if not 0 <= count <= 1024:
            raise ValueError('Wait requirement count out of bounds.')
        if not count:
            return []
        array_class = self.u64(array)
        array_name,namespace = self.class_name(array_class)
        if not array_name.endswith('[]') or self.u64(array+24)<count or self.u32(array_class+0x100)!=8:
            raise ValueError('Wait requirement reference array not verified: '+array_name)
        pointers = struct.unpack('<'+'Q'*count,self.read(array+32,count*8))
        result = []
        for index,item_pointer in enumerate(pointers):
            info = self.layout(self.u64(item_pointer))
            if 'Requirement' not in info['name'] or not info['namespace'].startswith(('MM.Client.','Shared.')):
                raise ValueError('Wait array element is not an in-scope requirement: '+info['namespace']+'.'+info['name'])
            values = {}
            offsets = sorted({f['offset'] for f in info['fields'] if f['offset']>=16} | {info['instance_size']})
            for field in info['fields']:
                offset,kind = field['offset'],field['type_code']
                if offset<16:
                    continue
                address = item_pointer+offset
                name = field['name']
                if kind==2:
                    boolean=self.read(address,1)[0]
                    if boolean not in (0,1):
                        raise ValueError('Wait condition bool invalid.')
                    values[name]=bool(boolean)
                elif kind in (8,9):
                    values[name]=self.i32(address) if kind==8 else self.u32(address)
                elif kind in (10,11):
                    values[name]=self.unpack(address,'<q') if kind==10 else self.u64(address)
                elif kind==14:
                    values[name]=self.string(self.u64(address))
                elif kind==17:
                    extent=next(o for o in offsets if o>offset)-offset
                    if extent in (4,8):
                        values[name]=dict(raw_value_type_bits=self.u32(address) if extent==4 else self.u64(address),bytes=extent)
            result.append(dict(index=index,type=info['namespace']+'.'+info['name'],scalar_fields=values,
                               interpretation='raw requirement fields; checker not invoked'))
        return result

    def decode_quests(self, root, classes, catalog, status_layout=None, visibility=None):
        character_id = root.get('character_id')
        if not isinstance(character_id,int) or character_id <= 0:
            raise ValueError('Current quest root has no positive character ID.')
        result = dict(character_id=character_id,active=[],completed=[])
        before = {}
        for role, expected in (('active','ActiveQuest'),('completed','CompletedQuest')):
            dictionary = root[role]
            if dictionary['kind'] != 'manual_dense':
                raise ValueError('Only the verified game dictionary can be decoded.')
            arrays, count = dictionary['arrays'],dictionary['count']
            if arrays['values']['name'] != expected+'[]' or arrays['values']['stride'] != 8 or arrays['keys']['stride'] != 8:
                raise ValueError('Quest ID/value array types not verified.')
            address = int(dictionary['pointer'],16)
            before[role] = (self.i32(address+dictionary['version_offset']),self.i32(address+dictionary['count_offset']))
            keys = list(struct.unpack('<'+'Q'*count,self.read(int(arrays['keys']['pointer'],16)+32,8*count))) if count else []
            pointers = list(struct.unpack('<'+'Q'*count,self.read(int(arrays['values']['pointer'],16)+32,8*count))) if count else []
            if len(keys) != len(set(keys)):
                raise ValueError('Duplicate quest IDs in the current dictionary.')
            klass, shape = classes[expected]
            fields = {f['name']:f for f in shape['fields']}
            for item_index,(key, pointer) in enumerate(zip(keys,pointers)):
                if self.u64(pointer) != klass:
                    raise ValueError('Quest object class mismatch at '+role+':'+str(item_index)+':'+str(self.class_name(self.u64(pointer))))
                object_character_id = self.u64(pointer+fields['characterId']['offset'])
                if object_character_id not in (0,character_id):
                    shared = False
                    if role == 'active':
                        static_pointer = self.u64(pointer+fields['<QuestStaticData>k__BackingField']['offset'])
                        static_class,static_shape = classes['QuestStaticData']
                        if self.u64(static_pointer) == static_class:
                            static_shared = next(f for f in static_shape['fields'] if f['name']=='<IsAccountShared>k__BackingField')
                            shared = self.read(static_pointer+static_shared['offset'],1)[0] == 1
                    if not shared:
                        raise ValueError('Quest character mismatch at '+role+':'+str(item_index)+': object='+str(object_character_id)+', root='+str(character_id))
                if role == 'completed':
                    item = self.scalar_fields(pointer,shape,{'questStaticDataId','acceptTime','rewardedTime','nextRepeatTime','clearCount','rewardedCount','characterId'})
                    if key != item['questStaticDataId']:
                        raise ValueError('Completed quest key does not match its object.')
                    definition = catalog.get(key,{})
                    item.update(quest_id=key,title=definition.get('title'),name=definition.get('name'),type=definition.get('type'),
                                hidden=(visibility or {}).get(key))
                else:
                    instance = self.u64(pointer+fields['<InstanceId>k__BackingField']['offset'])
                    if key != instance:
                        raise ValueError('Active quest instance key does not match its object.')
                    static_pointer = self.u64(pointer+fields['<QuestStaticData>k__BackingField']['offset'])
                    static_class,static_shape = classes['QuestStaticData']
                    if self.u64(static_pointer) != static_class:
                        raise ValueError('Active quest static object class mismatch.')
                    static_fields = {f['name']:f for f in static_shape['fields']}
                    static_id = self.u64(static_pointer+static_fields['<Id>k__BackingField']['offset'])
                    definition = catalog.get(static_id,{})
                    static_values = self.scalar_fields(static_pointer,static_shape,{'<UniqueName>k__BackingField','<DisplayTitle>k__BackingField','<Description>k__BackingField','<IsHidden>k__BackingField','<IsAccountShared>k__BackingField','<IsAcceptAutomatically>k__BackingField','<IsRepeatedlyExecutable>k__BackingField','<IsDroppable>k__BackingField'})
                    raw = self.scalar_fields(pointer,shape,{'<AcceptTime>k__BackingField','<StepIndex>k__BackingField','<StepStartTime>k__BackingField','<IsRewarded>k__BackingField','<BlockingQuestStaticDataId>k__BackingField','<BlockingQuestStepId>k__BackingField','<BlockReason>k__BackingField','<BlockDescriptParam>k__BackingField','<CanAcceptableQuestDialogueTrigger>k__BackingField','<CurrentStackCount>k__BackingField','characterId'})
                    index = raw['<StepIndex>k__BackingField']
                    # Runtime StepIds are decoded below. Never bind a current
                    # index directly to a preload display-array position.
                    step = {}
                    item = dict(quest_id=static_id,instance_id=key,name=definition.get('name'),
                                title=static_values.get('<DisplayTitle>k__BackingField') or definition.get('title'),
                                type=definition.get('type'),step_index=index,step_title=step.get('title'),
                                static=static_values,state_fields=raw,objectives=[],wait_requirements=[],
                                ownership=('current_character' if object_character_id==character_id else
                                           'current_character_dictionary_with_unassigned_object_owner' if object_character_id==0 else
                                           'account_shared_in_current_character_dictionary'))
                    base_shape = self.layout(int(shape['parent_class'],16))
                    for _ in range(4):
                        if (base_shape['name'],base_shape['namespace'])==('ActiveQuestBase`2','Shared.GameLogic.Quests'):
                            break
                        base_shape = self.layout(int(base_shape['parent_class'],16))
                    if (base_shape['name'],base_shape['namespace']) != ('ActiveQuestBase`2','Shared.GameLogic.Quests'):
                        raise ValueError('Active quest base not verified.')
                    item['base_state_fields'] = self.scalar_fields(pointer,base_shape,
                        {'<IsAccountShared>k__BackingField','isRestrictedByCharacterRequirement'})
                    for list_name,field_name,static in (('objectives','<Objectives>k__BackingField',step.get('objectives',[])),
                                                       ('wait_requirements','<WaitRequirements>k__BackingField',[])):
                        try:
                            item[list_name] = self.decode_objectives(pointer,self.u64(pointer+fields[field_name]['offset']),classes['QuestObjective'][0],static)
                        except (OSError,ValueError,UnicodeError,KeyError) as exc:
                            if list_name=='wait_requirements':
                                try:
                                    item[list_name] = self.decode_wait_fields(self.u64(pointer+fields[field_name]['offset']))
                                except (OSError,ValueError,UnicodeError,KeyError) as nested:
                                    item[list_name+'_read_error'] = str(nested)
                            else:
                                item[list_name+'_read_error'] = str(exc)
                    item['status'] = classify_status(item,status_layout)
                result[role].append(item)
            if before[role] != (self.i32(address+dictionary['version_offset']),self.i32(address+dictionary['count_offset'])):
                raise ValueError('Quest collection changed during the snapshot; no complete list asserted.')
        for role in before:
            dictionary = root[role]
            address = int(dictionary['pointer'],16)
            if before[role] != (self.i32(address+dictionary['version_offset']),self.i32(address+dictionary['count_offset'])):
                raise ValueError('Quest snapshot no longer consistent.')
        result['complete_collection_snapshot'] = True
        return result

    def status_layout(self, base, classes, plan):
        """Verify ordinary getter rules without calling any client code."""
        rule = plan.get('status_rules',{})
        if not rule:
            return None
        try:
            info = self.u64(base + rule['method_slot_rva'])
            klass = self.u64(info+0x20)
            if self.class_name(klass) != ('ActiveQuestBase`2','Shared.GameLogic.Quests'):
                return None
            if self.ascii(self.u64(info+0x18)) != 'GetCurrentStatus' or self.u64(info)-base != rule['getter_rva']:
                return None
            context = self.u64(klass+0xc0)
            for offset,name,rva in rule['context_methods']:
                method = self.u64(context+offset)
                if self.ascii(self.u64(method+0x18))!=name or self.u64(method+8)-base!=rva:
                    return None
            active_class = classes['ActiveQuest'][0]
            for offset,name,rva in rule['virtual_methods']:
                method = self.u64(active_class+offset+8)
                if self.ascii(self.u64(method+0x18))!=name or self.u64(active_class+offset)-base!=rva:
                    return None
            wrapper = self.u64(base+rule['wrapper_slot_rva'])
            if self.class_name(wrapper)!=('ILFixDynamicMethodWrapper','IFix'):
                return None
            array = self.u64(self.u64(wrapper+0xb8))
            count = self.u64(array+24)
            if count>100000 or any(count>slot and self.u64(array+32+slot*8) for slot in rule['patch_slots']):
                return None
            step_type = self.u64(base+rule['step_constants_slot_rva'])
            if self.class_name(step_type)!=('QuestStepIndexConstants','Shared.GameLogic.Quests'):
                return None
            statics = self.u64(step_type+0xb8)
            waiting,completed = self.i32(statics),self.i32(statics+4)
            block_type = self.u64(base+rule['block_constants_slot_rva'])
            if self.class_name(block_type)!=('QuestStaticDataId','Shared.GameLogic.Quests'):
                return None
            invalid = self.u64(self.u64(block_type+0xb8))
            if (waiting,completed,invalid)!=(-1,2147483647,18446744073709551615):
                return None
            return dict(verified=True,waiting=waiting,completed=completed,invalid_block=invalid,
                        source='file-derived GetCurrentStatus, runtime metadata/constants verified; no method calls')
        except (OSError,ValueError,UnicodeError,KeyError):
            return None


def classify_status(item, rule):
    if not rule or not rule.get('verified'):
        return 'unknown'
    state = item['state_fields']
    step = item['step_index']
    if step==rule['completed']:
        return 'wait_complete' if state['<IsRewarded>k__BackingField'] else 'wait_reward'
    if state['<BlockingQuestStaticDataId>k__BackingField']!=rule['invalid_block']:
        return 'blocked'
    base = item['base_state_fields']
    if base['<IsAccountShared>k__BackingField'] and base['isRestrictedByCharacterRequirement']:
        return 'restricted'
    if step!=rule['waiting']:
        return 'in_progress'
    if item.get('wait_requirements_read_error'):
        return 'unknown'
    flags = [condition.get('scalar_fields',{}).get('<IsSatisfied>k__BackingField')
             for condition in item['wait_requirements']]
    if any(type(flag) is not bool for flag in flags):
        return 'unknown'
    return 'acceptable' if all(flags) else 'waiting'


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


class QuestStateUnavailable(ValueError):
    pass


def _runtime_step_ids(reader, pointer):
    """Bounded current collection; unknown element wrappers stay explicit gaps.

    The original metadata identifies StepIds as a generic reference. We inspect
    the actual named collection/array types rather than invent its element ABI.
    """
    if not pointer:
        raise QuestStateUnavailable("current_StepIds_collection_missing")
    shape=reader.layout(reader.u64(pointer))
    name,namespace=shape['name'],shape['namespace']
    version_address=None
    if name.endswith('[]'):
        array=pointer
        count=reader.u64(array+24)
    else:
        if name=='List`1' and namespace=='System.Collections.Generic':
            fields={f['name'].lstrip('_'):f['offset'] for f in shape['fields']}
            array=reader.u64(pointer+fields['items'])
            count=reader.i32(pointer+fields['size'])
            version_address=pointer+fields['version']
        elif name.startswith(('MMList','MMReadOnlyList')) and namespace=='Silvervine.ManualMemory':
            for _ in range(4):
                fields={f['name']:f['offset'] for f in shape['fields']}
                if {'array','size','version'}<=fields.keys():break
                shape=reader.layout(int(shape['parent_class'],16))
            if not {'array','size','version'}<=fields.keys():
                raise QuestStateUnavailable('current_StepIds_list_layout_unresolved')
            array=reader.u64(pointer+fields['array']);count=reader.i32(pointer+fields['size'])
            version_address=pointer+fields['version']
        else:
            raise QuestStateUnavailable('current_StepIds_container_unresolved:'+namespace+'.'+name)
    if type(count) is not int or not 0<=count<=1024:
        raise QuestStateUnavailable('current_StepIds_count_out_of_bounds')
    array_shape=reader.layout(reader.u64(array))
    array_name=array_shape['name']
    if array_name not in ('Int32[]','UInt32[]') or array_shape['namespace']!='System' or array_shape['element_size']!=4:
        raise QuestStateUnavailable('current_StepIds_element_unresolved:'+array_shape['namespace']+'.'+array_name)
    capacity=reader.u64(array+24)
    if not count<=capacity<=4096:
        raise QuestStateUnavailable('current_StepIds_capacity_invalid')
    version=reader.i32(version_address) if version_address is not None else None
    values=list(struct.unpack('<'+('i' if array_name=='Int32[]' else 'I')*count,reader.read(array+32,count*4))) if count else []
    if len(set(values))!=len(values):
        raise QuestStateUnavailable('current_StepIds_not_unique')
    if reader.u64(array+24)!=capacity or (version_address is not None and reader.i32(version_address)!=version):
        raise QuestStateUnavailable('current_StepIds_changed_during_read')
    return values


def _step_predicate_guard(reader, base, plan):
    """Verify the exact current-root branch without calling any game method.

    This predicate lives in a different IFix wrapper from GetCurrentStatus.
    Its own wrapper slot is pinned by its IsPatched helper, not reused by name.
    """
    try:
        for check in plan['code_checks']:
            expected=bytes.fromhex(check['bytes_hex'])
            if reader.read(base+check['rva'],len(expected))!=expected:
                return {'verified':False,'missing':'current_step_predicate_code_changed:'+check['name']}
        wrapper=reader.u64(base+plan['wrapper_slot_rva'])
        if reader.class_name(wrapper)!=('ILFixDynamicMethodWrapper','IFix'):
            return {'verified':False,'missing':'current_step_predicate_IFix_wrapper_changed'}
        array=reader.u64(reader.u64(wrapper+0xb8))
        count=reader.u64(array+24)
        if count>100000 or any(count>slot and reader.u64(array+32+slot*8) for slot in plan['patch_slots']):
            return {'verified':False,'missing':'current_step_predicate_IFix_patch_present_or_unresolved'}
        return {'verified':True,'source':'native_IsQuestStepCompletedBefore_current_active_branch',
            'wrapper_slot_rva':plan['wrapper_slot_rva'],'patch_slots':plan['patch_slots']}
    except (OSError,ValueError,KeyError,UnicodeError) as exc:
        return {'verified':False,'missing':'current_step_predicate_guard_unavailable:'+str(exc)}


def _descriptor_code_guard(reader,base,plan):
    for check in plan['code_checks']:
        expected=bytes.fromhex(check['bytes_hex'])
        if reader.read(base+check['rva'],len(expected))!=expected:
            raise QuestStateUnavailable('current_descriptor_getter_changed:'+check['name'])


def _read_qdbl_optional(reader,pointer,klass,spec,value):
    """Read closed Optional<T> named fields; never infer tag position from width."""
    declaration=value.get('declared_type_metadata') or {}
    if (declaration.get('type_code')!=21 or declaration.get('name')!='Optional`1' or
            declaration.get('namespace')!='MM.QDBL.Definitions.ValueType' or len(declaration.get('arguments',[]))!=1):
        raise QuestStateUnavailable('current_optional_source_declaration_mismatch')
    argument=declaration['arguments'][0]
    if argument.get('type_code')==14:
        payload_width=8;payload_kind=14;value_type='System.String';domain=None
    elif argument.get('type_code')==10:
        payload_width=8;payload_kind=10;value_type='System.Int64';domain=None
    elif (argument.get('type_code')==17 and argument.get('namespace')=='Shared.Symbols' and
            argument.get('name') in ('ConsumableSheetId','QuestFakeItemSheetId','ChannelSheetId')):
        payload_width=4;payload_kind=17;value_type='Shared.Symbols.'+argument['name'];domain=argument['name'][:-2]
    else:
        raise QuestStateUnavailable('current_optional_source_argument_unsupported')
    table=reader.u64(klass+0x80)
    field_index=spec['fields'].index(value['field'])
    type_address=table+field_index*32+8
    type_pointer=reader.u64(type_address)
    if not type_pointer or ((reader.u32(type_pointer+8)>>16)&255)!=21:
        raise QuestStateUnavailable('current_optional_field_type_changed')
    generic=reader.u64(type_pointer)
    if not generic:raise QuestStateUnavailable('current_optional_generic_missing')
    closed_address=generic+24
    closed_class=reader.u64(closed_address)
    if not closed_class:raise QuestStateUnavailable('current_optional_closed_class_not_initialized')
    if reader.class_name(closed_class)!=('Optional`1','MM.QDBL.Definitions.ValueType'):
        raise QuestStateUnavailable('current_optional_closed_class_mismatch')
    shape=reader.layout(closed_class,['value','<HasValue>k__BackingField'])
    parent=int(shape['parent_class'],16)
    if not parent or reader.class_name(parent)!=('ValueType','System'):
        raise QuestStateUnavailable('current_optional_boxed_value_type_parent_unverified')
    payload,presence=shape['fields']
    if (payload['type_code'],presence['type_code'])!=(payload_kind,2):
        raise QuestStateUnavailable('current_optional_closed_field_types_changed')
    width=value['width'];header=16
    if shape['instance_size']!=header+width:
        raise QuestStateUnavailable('current_optional_boxed_size_differs_from_source_getter_span')
    vo,ho=payload['offset']-header,presence['offset']-header
    if not (0<=vo<=width-payload_width and 0<=ho<width and not vo<=ho<vo+payload_width):
        raise QuestStateUnavailable('current_optional_named_offsets_outside_source_span')
    address=pointer+value['offset'];raw=reader.read(address,width)
    tag=raw[ho]
    if tag not in (0,1):raise QuestStateUnavailable('current_optional_presence_tag_invalid')
    decoded=None
    if tag:
        if payload_kind==14:
            reference=struct.unpack_from('<Q',raw,vo)[0]
            decoded=reader.string(reference) if reference else None
        else:
            decoded=struct.unpack_from('<q' if payload_width==8 else '<i',raw,vo)[0]
    if (reader.read(address,width)!=raw or reader.u64(closed_address)!=closed_class or
            reader.u64(type_address)!=type_pointer or reader.u64(type_pointer)!=generic):
        raise QuestStateUnavailable('current_optional_value_or_layout_changed_during_read')
    result={'has_value':bool(tag),'value':decoded,'storage':'QDBL.Optional','interpreted':True,
        'declared_type':'MM.QDBL.Definitions.ValueType.Optional<'+value_type+'>','value_type':value_type,
        'underlying_type':'System.String' if payload_kind==14 else 'System.Int64' if payload_width==8 else 'System.Int32',
        'layout_basis':'current_cached_closed_class_named_value_and_HasValue_fields',
        'source_inline_width':width,'value_offset':vo,'has_value_offset':ho}
    if domain:
        result.update(table_domain=domain,id_field='Id',domain_mapping_verified=True,
            symbol_name=None,symbol_name_resolved=False)
    return result


def _read_descriptor(reader,pointer,plan):
    """Decode only pinned generated types and their file-derived getter widths."""
    if not pointer:
        return {'available':False,'missing':['current_descriptor_reference_null']}
    klass=reader.u64(pointer);name,namespace=reader.class_name(klass)
    spec=next((r for r in plan['classes'] if (r['name'],r['namespace'])==(name,namespace)),None)
    result={'type':namespace+'.'+name,'values':{},'missing':[],
        'is_current_reference':True,'activation_confirmed':False,'executable_target_confirmed':False}
    if spec is None:
        return result|{'available':False,'missing':['current_descriptor_type_not_supported:'+namespace+'.'+name]}
    shape=reader.layout(klass,spec['fields']);fields={f['name']:f for f in shape['fields']}
    for value in spec['values']:
        f=fields[value['field']]
        if f['offset']!=value['offset'] or f['offset']+value['width']>shape['instance_size']:
            raise QuestStateUnavailable('current_descriptor_field_layout_changed:'+value['name'])
        address=pointer+f['offset'];codec=value['codec']
        if codec=='string_reference':
            if f['type_code']!=14:raise QuestStateUnavailable('current_descriptor_string_type_changed')
            string=reader.u64(address)
            decoded={'value':reader.string(string) if string else None,'storage':'System.String','interpreted':True}
        elif codec=='native_i32_identifier':
            if f['type_code']!=17:raise QuestStateUnavailable('current_descriptor_identifier_type_changed')
            declared=value.get('declared_type_metadata') or {}
            if (declared.get('type_code')!=17 or declared.get('namespace')!='Shared.Symbols' or
                    declared.get('name') not in ('InstantPropSheetId','ChannelSheetId') or
                    value.get('id_field')!='Id' or value.get('underlying_type')!='System.Int32'):
                raise QuestStateUnavailable('descriptor_sheet_id_declaration_unverified')
            decoded={'value':reader.i32(address),'storage':'native_32bit_identifier','interpreted':True,
                'declared_type':value['declared_type'],'table_domain':value['table_domain'],
                'id_field':'Id','underlying_type':'System.Int32',
                'symbol_name':None,'symbol_name_resolved':False,'domain_mapping_verified':True}
        elif codec=='qdbl_optional':
            if f['type_code']!=21:raise QuestStateUnavailable('current_descriptor_wrapper_type_changed')
            try:
                decoded=_read_qdbl_optional(reader,pointer,klass,spec,value)
            except (OSError,ValueError,KeyError,UnicodeError) as exc:
                reason='current_optional_layout_unavailable:'+value['name']+':'+str(exc)
                decoded={'bytes_hex':reader.read(address,value['width']).hex(),'storage':'unresolved_QDBL.Optional',
                    'interpreted':False,'reason':reason,'declared_type_metadata':deepcopy(value.get('declared_type_metadata'))}
                result['missing'].append(reason)
        else:
            raise QuestStateUnavailable('current_descriptor_codec_not_supported:'+str(codec))
        result['values'][value['name']]=decoded|{'metadata_type_index':value['metadata_type_index']}
    if reader.u64(pointer)!=klass:raise QuestStateUnavailable('current_descriptor_class_changed_during_read')
    return result|{'available':True,'field_layout_verified':True}


def _typed_reference_collection(reader,pointer,array_name,namespace,limit=1024):
    """Known managed arrays/lists only; unfamiliar dictionaries remain a gap."""
    if not pointer:return []
    shape=reader.layout(reader.u64(pointer));name=shape['name'];version_address=None
    if name==array_name:
        array=pointer;count=reader.u64(array+24)
    elif (name=='List`1' and shape['namespace']=='System.Collections.Generic') or (
            name.startswith(('MMList','MMReadOnlyList')) and shape['namespace']=='Silvervine.ManualMemory'):
        for _ in range(4):
            fields={f['name'].lstrip('_'):f['offset'] for f in shape['fields']}
            if {'size','version'}<=fields.keys() and ('items'in fields or 'array'in fields):break
            shape=reader.layout(int(shape['parent_class'],16))
        if not {'size','version'}<=fields.keys() or not ('items'in fields or 'array'in fields):
            raise QuestStateUnavailable('descriptor_reference_list_layout_unresolved')
        array=reader.u64(pointer+fields.get('items',fields.get('array')))
        count=reader.i32(pointer+fields['size']);version_address=pointer+fields['version']
    else:
        raise QuestStateUnavailable('descriptor_reference_collection_unsupported:'+shape['namespace']+'.'+name)
    if not 0<=count<=limit:raise QuestStateUnavailable('descriptor_reference_collection_bounds')
    array_class=reader.u64(array);shape=reader.layout(array_class)
    if (shape['name'],shape['namespace'],shape['element_size'])!=(array_name,namespace,8):
        raise QuestStateUnavailable('descriptor_reference_array_type_changed')
    capacity=reader.u64(array+24)
    if not count<=capacity<=4096:raise QuestStateUnavailable('descriptor_reference_array_capacity')
    version=reader.i32(version_address) if version_address is not None else None
    values=list(struct.unpack('<'+'Q'*count,reader.read(array+32,count*8))) if count else []
    if (reader.u64(array+24)!=capacity or
            version_address is not None and reader.i32(version_address)!=version):
        raise QuestStateUnavailable('descriptor_reference_collection_changed')
    return values


def read_current_quest_descriptors(reader,active,static,classes,plan,*,step_id=None):
    """No scanning: follow only the already validated current quest's fields."""
    result={'objectives':[],'traits':[],'missing':[]}
    active_class,active_shape=classes['ActiveQuest'];objective_class,objective_shape=classes['QuestObjective']
    if reader.u64(active)!=active_class or reader.u64(static)!=classes['QuestStaticData'][0]:
        raise QuestStateUnavailable('descriptor_current_quest_owner_class_changed')
    af={f['name']:f for f in active_shape['fields']};of={f['name']:f for f in objective_shape['fields']}
    for accessor in plan['objective_accessors']:
        if of[accessor['field']]['offset']!=accessor['offset']:
            raise QuestStateUnavailable('descriptor_objective_accessor_layout_changed:'+accessor['field'])
    objective_list=reader.u64(active+af['<Objectives>k__BackingField']['offset'])
    pointers=reader.quest_list(objective_list);indexes=set()
    for pointer in pointers:
        if reader.u64(pointer)!=objective_class or reader.u64(pointer+of['<OwnerActiveQuest>k__BackingField']['offset'])!=active:
            raise QuestStateUnavailable('descriptor_objective_belongs_to_other_quest')
        index=reader.i32(pointer+of['<Index>k__BackingField']['offset'])
        if index<0 or index>=1024 or index in indexes:raise QuestStateUnavailable('descriptor_objective_index_ambiguous')
        indexes.add(index);entry={'objective_index':index,'data':None,'auto_play':None}
        for key,field in (('data','<Data>k__BackingField'),('auto_play','<AutoPlayDescriptor>k__BackingField')):
            target=reader.u64(pointer+of[field]['offset'])
            try:
                entry[key]=_read_descriptor(reader,target,plan)
            except (OSError,ValueError,KeyError,UnicodeError) as exc:
                entry[key]={'available':False,'missing':[str(exc)]}
            if reader.u64(pointer+of[field]['offset'])!=target:
                raise QuestStateUnavailable('descriptor_reference_changed_during_read')
        result['objectives'].append(entry)
    if reader.u64(active+af['<Objectives>k__BackingField']['offset'])!=objective_list:
        raise QuestStateUnavailable('descriptor_objectives_reference_changed')

    traversal={r['name']:r for r in plan['traversal_classes']}
    def traits_at(owner,shape,scope):
        fields={f['name']:f for f in shape['fields']};field=fields['<Traits>k__BackingField']
        collection=reader.u64(owner+field['offset']);spec=traversal['QuestTraitStaticData']
        for ptr in _typed_reference_collection(reader,collection,'QuestTraitStaticData[]','Shared.QDBLs.QDBLXData'):
            klass=reader.u64(ptr)
            if reader.class_name(klass)!=(spec['name'],spec['namespace']):raise QuestStateUnavailable('quest_trait_class_changed')
            trait_shape=reader.layout(klass,spec['fields']);tf={f['name']:f for f in trait_shape['fields']}
            target=reader.u64(ptr+tf['<Trait>k__BackingField']['offset'])
            descriptor=_read_descriptor(reader,target,plan)
            result['traits'].append({'scope':scope,'descriptor':descriptor,'activation_conditions_verified':False})
        if reader.u64(owner+field['offset'])!=collection:raise QuestStateUnavailable('quest_trait_list_changed')
    try:traits_at(static,classes['QuestStaticData'][1],'quest')
    except (OSError,ValueError,KeyError,UnicodeError) as exc:result['missing'].append('current_quest_traits:'+str(exc))
    if step_id is not None:
        try:
            sf={f['name']:f for f in classes['QuestStaticData'][1]['fields']}
            steps_pointer=reader.u64(static+sf['<QuestSteps>k__BackingField']['offset'])
            # A dictionary layout has not been established for this particular
            # field; supported typed lists are read without assuming ordering.
            steps=_typed_reference_collection(reader,steps_pointer,'QuestStepStaticData[]','Shared.QDBLs.QDBLXData')
            matches=[];spec=traversal['QuestStepStaticData']
            for ptr in steps:
                klass=reader.u64(ptr)
                if reader.class_name(klass)!=(spec['name'],spec['namespace']):raise QuestStateUnavailable('quest_step_static_class_changed')
                shape=reader.layout(klass,spec['fields']);fields={f['name']:f for f in shape['fields']}
                if reader.i32(ptr+fields['<StepId>k__BackingField']['offset'])==step_id:matches.append((ptr,shape))
            if len(matches)!=1:raise QuestStateUnavailable('current_step_trait_owner_not_unique')
            traits_at(*matches[0],'current_step')
        except (OSError,ValueError,KeyError,UnicodeError) as exc:result['missing'].append('current_step_traits:'+str(exc))
    result['missing'].append('trait_activation_conditions_and_current_presence_not_evaluated')
    return result


class QuestStateMemorySession:
    """Current singular quest root only; construction is behind provider gate."""
    def __init__(self,pid,*,cancel=None,reader_factory=QuestPayloadReader):
        self.reader=reader_factory(pid,cancel)
        try:
            self.base,self.module=self.reader.module_base()
            self.plan=json.loads((root()/'data/arbeit-quest-state/quest-memory-layout.json').read_text(encoding='utf8'))
            self.step_plan=json.loads((root()/'data/arbeit-quest-state/step-predicate.json').read_text(encoding='utf8'))
            self.descriptor_plan=json.loads((root()/'data/arbeit-quest-state/current-descriptors.json').read_text(encoding='utf8'))
            from .quest_program import load_schema
            self.program_schema=load_schema()
            self.character_plan=load_rule('character-memory-layout')
            metadata=Path(self.reader.image).parent/'MabinogiMobile_Data/il2cpp_data/Metadata/global-metadata.dat'
            for path,key in ((self.module,'game_assembly_sha256'),(metadata,'metadata_sha256')):
                if (file_sha256(path)!=self.plan[key] or self.plan[key]!=self.character_plan[key]
                        or self.plan[key]!=self.step_plan[key] or self.plan[key]!=self.descriptor_plan[key]
                        or self.plan[key]!=self.program_schema[key]):
                    raise QuestStateUnavailable('quest_and_current_character_client_fingerprint_mismatch')
            self.classes={}
            namespaces={'QuestComponent':'MM.Client.Game.Logic.Components','ActiveQuest':'MM.Client.PresentationContracts.ComponentInterfaces.Types',
                'CompletedQuest':'MM.Client.PresentationContracts.ComponentInterfaces.Types','QuestStaticData':'Shared.QDBLs.QDBLXData',
                'QuestObjective':'MM.Client.PresentationContracts.ComponentInterfaces.Types'}
            for entry in self.plan['classes']:
                if namespaces.get(entry['name'])!=entry['namespace']:continue
                klass=_class(self.reader,self.base,entry)
                self.classes[entry['name']]=(klass,self.reader.layout(klass,entry['fields']))
            if set(self.classes)!=set(namespaces):
                raise QuestStateUnavailable('required_quest_payload_classes_missing')
            identity=next(row for row in self.character_plan['classes'] if row['name']=='ControlObjectIdentityComponent')
            self.identity_class=_class(self.reader,self.base,identity)
            self.identity_shape=self.reader.layout(self.identity_class,identity['fields'])
            raw=json.loads((root()/'data/arbeit-catalog.json').read_text(encoding='utf8'))
            self.catalog={q['id']:q for q in raw['quests']}
        except BaseException:
            self.reader.close()
            raise

    def close(self):self.reader.close()
    def __enter__(self):return self
    def __exit__(self,*_):self.close()

    def _identity(self,context):
        if self.reader.u64(context['identity_pointer'])!=self.identity_class:
            raise QuestStateUnavailable('current_quest_character_identity_class_changed')
        return CharacterMemoryReader._value(self.reader,context['identity_pointer'],self.identity_class,self.identity_shape)

    def _root(self,context,identity):
        pointer=context['quest_pointer'];klass,shape=self.classes['QuestComponent']
        if not pointer or self.reader.u64(pointer)!=klass:
            raise QuestStateUnavailable('current_QuestComponent_not_ready')
        fields={f['name']:f['offset'] for f in shape['fields']}
        return {'character_id':int(identity['character_id']),
            'active':self.reader.dictionary(self.reader.u64(pointer+fields['<ActiveQuests>k__BackingField'])),
            'completed':self.reader.dictionary(self.reader.u64(pointer+fields['<CompletedQuests>k__BackingField']))}

    def _decode(self,current,rule):
        result=self.reader.decode_quests(current,self.classes,self.catalog,rule)
        dictionary=current['active'];count=dictionary['count']
        array=int(dictionary['arrays']['values']['pointer'],16)
        active_fields={f['name']:f['offset'] for f in self.classes['ActiveQuest'][1]['fields']}
        static_fields={f['name']:f['offset'] for f in self.classes['QuestStaticData'][1]['fields']}
        for index,row in enumerate(result['active']):
            try:
                pointer=self.reader.u64(array+32+index*8)
                if self.reader.u64(pointer+active_fields['<InstanceId>k__BackingField'])!=row['instance_id']:
                    raise QuestStateUnavailable('current_quest_instance_changed_before_StepIds')
                static=self.reader.u64(pointer+active_fields['<QuestStaticData>k__BackingField'])
                if self.reader.u64(static)!=self.classes['QuestStaticData'][0]:
                    raise QuestStateUnavailable('current_quest_static_class_changed')
                from .quest_program import read_offer_definition, read_owned_programs
                try:
                    row['offer_definition']=read_offer_definition(self.reader,self.base,static)
                except (OSError,ValueError,KeyError,UnicodeError) as exc:
                    row['offer_definition']={'missing':[str(exc)]}
                row['runtime_step_ids']=_runtime_step_ids(self.reader,self.reader.u64(static+static_fields['StepIds']))
                step_index=row['step_index'];ids=row['runtime_step_ids']
                if 0<=step_index<len(ids):
                    steps=[s for s in self.catalog.get(row['quest_id'],{}).get('steps',[]) if s.get('id')==ids[step_index]]
                    if len(steps)==1:
                        display=steps[0];row['step_title']=display.get('title')
                        for objective in row.get('objectives',[]):
                            oi=objective['index'];items=display.get('objectives',[])
                            if 0<=oi<len(items):
                                objective.update(description_static=items[oi].get('text'),goal_static=items[oi].get('goal'),
                                    kind_static=items[oi].get('kind'),static_binding='preload_display_candidate_only; not runtime objective program')
                try:
                    if not 0<=step_index<len(ids):
                        raise QuestStateUnavailable('current_step_binding_required_for_descriptors')
                    if self.descriptor_guard_error:
                        raise QuestStateUnavailable(self.descriptor_guard_error)
                    row['runtime_descriptors']=read_current_quest_descriptors(self.reader,pointer,static,self.classes,self.descriptor_plan,
                        step_id=ids[step_index])
                except (OSError,ValueError,KeyError,UnicodeError) as exc:
                    row['runtime_descriptors']={'objectives':[],'traits':[],'missing':[str(exc)]}
                if 0<=step_index<len(ids):
                    try:
                        row['runtime_programs']=read_owned_programs(self.reader,self.base,pointer,self.classes,step_id=ids[step_index])
                    except (OSError,ValueError,KeyError,UnicodeError) as exc:
                        row['runtime_programs_missing']=str(exc)
            except (OSError,ValueError,KeyError,UnicodeError) as exc:
                row['runtime_step_ids_missing']=str(exc)
        return result

    def read(self):
        r=self.reader
        context=active_context(r,self.base,self.character_plan)
        identity=self._identity(context)
        rule=r.status_layout(self.base,self.classes,self.plan)
        step_rule=_step_predicate_guard(r,self.base,self.step_plan)
        self.descriptor_guard_error=None
        try:_descriptor_code_guard(r,self.base,self.descriptor_plan)
        except (OSError,ValueError,KeyError,UnicodeError) as exc:self.descriptor_guard_error=str(exc)
        first=self._decode(self._root(context,identity),rule)
        second=self._decode(self._root(context,identity),rule)
        descriptor_guard_after=None
        try:_descriptor_code_guard(r,self.base,self.descriptor_plan)
        except (OSError,ValueError,KeyError,UnicodeError) as exc:descriptor_guard_after=str(exc)
        if (first!=second or context!=active_context(r,self.base,self.character_plan) or identity!=self._identity(context)
                or step_rule!=_step_predicate_guard(r,self.base,self.step_plan)
                or descriptor_guard_after!=self.descriptor_guard_error
                or rule!=r.status_layout(self.base,self.classes,self.plan)):
            raise QuestStateUnavailable('current_quest_state_or_owner_changed_during_read')
        second.update(pid=r.pid,process_started_at=r.started_at.isoformat(),captured_at=_now(),identity=identity,
            current_context_verified=True,status_rules_verified=rule is not None,status_rules=rule,
            intermediate_step_rule=step_rule,
            source_version={key:self.plan[key] for key in ('game_assembly_sha256','metadata_sha256')},
            root_binding='current SingularEntity.IQuestComponent; no historical heap owner selection')
        return second


def state_proofs_from_snapshot(snapshot,*,session_id,quest_id=None,expected_accept_time=None):
    """Current typed state only. Historical completion never completes a new run."""
    result={'state_proofs':[],'offers':[],'accepted_instances':[],'completed_records':[],
        'objective_evidence':[],'current_quests':[],'current_quest':None,
        'current_descriptors':[],'current_trait_descriptors':[],'objective_programs':[],'action_requests':[],'missing':[]}
    if not snapshot.get('current_context_verified') or not snapshot.get('complete_collection_snapshot'):
        result['missing'].append('current_character_complete_quest_collections_required')
        return result
    if not snapshot.get('status_rules_verified') or not (snapshot.get('status_rules') or {}).get('verified'):
        result['missing'].append('native_status_metadata_constants_and_patch_guards_required')
        return result
    character=str(snapshot['identity']['character_id'])
    if character!=str(snapshot['character_id']):
        result['missing'].append('quest_collection_identity_mismatch')
        return result
    stamp=snapshot['captured_at'];rule=snapshot['status_rules']
    base={'confirmed':True,'source':'current-root-typed-quest-collections','observed_at':stamp,'session_id':session_id,
        'character_id':character,'source_version':snapshot['source_version'],'process_started_at':snapshot['process_started_at']}
    active=[row for row in snapshot['active'] if quest_id is None or row['quest_id']==quest_id]
    counts={q:sum(row['quest_id']==q for row in active) for q in {row['quest_id'] for row in active}}
    for row in active:
        q=row['quest_id']
        if counts[q]!=1:
            result['missing'].append('current_quest_instance_ambiguous:'+str(q));continue
        index=row['step_index'];raw=row['state_fields'];accepted=index!=rule['waiting']
        if not accepted:
            definition=row.get('offer_definition') or {}
            giver=definition.get('giver_definition')
            valid_giver=(isinstance(giver,dict) and type(giver.get('npc_id')) is int and giver['npc_id']!=0
                and giver.get('declared_type')=='Shared.Symbols.NPCSheetId')
            eligible=row['status']=='acceptable'
            result['offers'].append(base|{'quest_id':q,'instance_id':row['instance_id'],'status':row['status'],
                'eligible':row['status']=='acceptable','source':base['source'],'observed_at':stamp,
                'quest_name':definition.get('quest_name'),'conditions_met':eligible,'eligibility_verified':row['status']!='unknown',
                'quest_name_id':definition.get('quest_name_id'),'quest_name_type':definition.get('quest_name_type'),
                'available_candidate':eligible and valid_giver,'offered':False,
                'giver_source':'current_QuestStaticData.Sender' if valid_giver else None,
                'giver':deepcopy(giver) if valid_giver else None,'giver_definition':deepcopy(giver),
                'npc_id':giver['npc_id'] if valid_giver else None,
                'definition_missing':definition.get('missing',[]),
                'basis':'verified_current_waiting_ActiveQuest_status_and_current_QuestStaticData_Sender',
                'wait_requirements':row.get('wait_requirements',[]),'dialogue_giver_binding':None,
                'not_current_npc_offer':True})
            continue
        accepted_proof=base|{'quest_id':q,'state':'accepted','instance_id':row['instance_id'],
            'accept_time':raw.get('<AcceptTime>k__BackingField'),'step_index':index,
            'basis':'native_IsAccepted_with_current_owned_ActiveQuest_and_verified_waiting_sentinel'}
        result['state_proofs'].append(accepted_proof)
        result['accepted_instances'].append({'quest_id':q,'instance_id':row['instance_id'],
            'accept_time':accepted_proof['accept_time'],'step_index':index,'status':row['status'],
            'runtime_step_ids':row.get('runtime_step_ids'),'objectives':row.get('objectives',[])})
        ids=row.get('runtime_step_ids')
        current_step_id=ids[index] if ids is not None and 0<=index<len(ids) else None
        binding=base|{'quest_id':q,'instance_id':row['instance_id'],
            'accept_time':accepted_proof['accept_time'],'step_index':index,'step_id':current_step_id,
            'title':row.get('title'),'name':row.get('name'),'runtime_step_ids':ids,
            'basis':'current_owned_ActiveQuest_and_runtime_StepIds',
            'objectives':row.get('objectives',[]),'objective_success_verified':False,
            'runtime_descriptors':deepcopy(row.get('runtime_descriptors'))}
        result['current_quests'].append(binding)
        from .quest_program import build_program
        for program_row in row.get('runtime_programs',[]):
            current_program=base|{'quest_id':q,'instance_id':row['instance_id'],
                'accept_time':accepted_proof['accept_time'],'step_id':current_step_id,'step_index':index,
                'objective_index':program_row['objective_index'],'owner_verified':program_row.get('owner_verified') is True,
                'data':deepcopy(program_row.get('data')),'auto_play':deepcopy(program_row.get('auto_play')),
                'typed_effect':deepcopy(program_row.get('typed_effect'))}
            built=build_program(current_program)
            result['objective_programs'].append(current_program|{'program':built})
            if built.get('action_request') is not None:result['action_requests'].append(built['action_request'])
        if row.get('runtime_programs_missing'):
            result['missing'].append('quest_program:'+str(q)+':'+row['runtime_programs_missing'])
        runtime=row.get('runtime_descriptors') or {}
        result['missing'].extend('quest_descriptor:'+str(q)+':'+reason for reason in runtime.get('missing',[]))
        for descriptor in runtime.get('objectives',[]):
            result['current_descriptors'].append(base|{'quest_id':q,'instance_id':row['instance_id'],
                'accept_time':accepted_proof['accept_time'],'step_id':current_step_id,'step_index':index,
                'objective_index':descriptor['objective_index'],'data':deepcopy(descriptor.get('data')),
                'auto_play':deepcopy(descriptor.get('auto_play')),'step_binding_confirmed':current_step_id is not None,
                'source':'current-root-typed-quest-descriptor',
                'basis':'current_owned_QuestObjective_Data_and_AutoPlayDescriptor_references',
                'objective_success_verified':False,'executable_target_confirmed':False})
        for trait in runtime.get('traits',[]):
            result['current_trait_descriptors'].append(base|{'quest_id':q,'instance_id':row['instance_id'],
                'accept_time':accepted_proof['accept_time'],'step_id':current_step_id if trait['scope']=='current_step' else None,
                'scope':trait['scope'],'descriptor':deepcopy(trait['descriptor']),
                'source':'current-root-typed-quest-descriptor','activation_conditions_verified':False,
                'executable_target_confirmed':False})
        if current_step_id is not None:
            for objective in row.get('objectives',[]):
                effect=next((p.get('typed_effect') for p in row.get('runtime_programs',[]) if p['objective_index']==objective['index']),None)
                typed=effect if isinstance(effect,dict) and effect.get('confirmed') is True and type(effect.get('complete')) is bool else None
                result['objective_evidence'].append({**binding,'objectives':None,'objective_index':objective['index'],
                    'raw_count':objective.get('count'),'raw_goal':objective.get('goal_memory_raw'),
                    'confirmed':typed is not None,'basis':typed['basis'] if typed else 'raw_typed_objective_fields_not_effect_proof',
                    'count':typed['count'] if typed else None,'goal':typed['goal'] if typed else None,
                    'semantics':typed['semantics'] if typed else None,'typed_effect':deepcopy(effect),
                    'complete':typed['complete'] if typed else None})
        verified_effect_indices={p['objective_index'] for p in row.get('runtime_programs',[]) if (p.get('typed_effect') or {}).get('confirmed')}
        if any(o.get('index') not in verified_effect_indices for o in row.get('objectives',[])):
            result['missing'].append('objective_native_effective_goal_and_optional_type_rule_required:'+str(q))
        if index==rule['completed']:
            ids=row.get('runtime_step_ids')
            if ids is None:
                result['missing'].append('runtime_StepIds_required_for_step_proofs:'+str(q)+':'+row.get('runtime_step_ids_missing','unknown'))
            else:
                for step_id in ids:
                    result['state_proofs'].append(base|{'quest_id':q,'step_id':step_id,'state':'step_complete',
                        'instance_id':row['instance_id'],'accept_time':accepted_proof['accept_time'],
                        'basis':'native_IsAllStepsCompleted_and_current_runtime_StepIds',
                        'scope':'aggregate_game_step_termination; skipped branches are not individual objective success'})
        elif index>=0:
            step_rule=snapshot.get('intermediate_step_rule') or {}
            if not step_rule.get('verified'):
                result['missing'].append('intermediate_step_predicate_not_verified:'+str(q)+':'+step_rule.get('missing','guard_required'))
            elif ids is None:
                result['missing'].append('runtime_StepIds_required_for_intermediate_step_proofs:'+str(q))
            elif index>=len(ids):
                result['missing'].append('current_step_index_outside_runtime_StepIds:'+str(q))
            else:
                for step_id in ids[:index]:
                    result['state_proofs'].append(base|{'quest_id':q,'step_id':step_id,'state':'step_complete',
                        'instance_id':row['instance_id'],'accept_time':accepted_proof['accept_time'],
                        'basis':'native_IsQuestStepCompletedBefore_with_current_runtime_StepIds',
                        'scope':'game_step_termination; no individual objective success inferred'})
    if len(result['current_quests'])==1:
        result['current_quest']=deepcopy(result['current_quests'][0])
    for row in snapshot['completed']:
        q=row['quest_id']
        if quest_id is not None and q!=quest_id:continue
        result['completed_records'].append({key:row.get(key) for key in ('quest_id','acceptTime','rewardedTime','nextRepeatTime','clearCount','rewardedCount')})
        expected=expected_accept_time.get(q,expected_accept_time.get(str(q))) if isinstance(expected_accept_time,dict) else expected_accept_time if quest_id==q else None
        if type(expected) is not int or expected<=0 or row.get('acceptTime')!=expected:
            result['missing'].append('current_run_accepted_time_binding_for_completed_record:'+str(q));continue
        if any(a['quest_id']==q and a['state_fields'].get('<AcceptTime>k__BackingField')==expected and
               a['step_index'] not in (rule['waiting'],rule['completed']) for a in active):
            result['missing'].append('active_completed_occurrence_conflict:'+str(q));continue
        if (type(row.get('rewardedTime')) is not int or row['rewardedTime']<expected or row['rewardedTime']<=0 or
                type(row.get('clearCount')) is not int or row['clearCount']<1 or
                type(row.get('rewardedCount')) is not int or row['rewardedCount']<1):
            result['missing'].append('current_run_completed_reward_record_incomplete:'+str(q));continue
        result['state_proofs'].append(base|{'quest_id':q,'state':'completed','accept_time':expected,
            'basis':'current_CompletedQuest_record_bound_to_this_accepted_occurrence',
            'completed_record':{key:row.get(key) for key in ('acceptTime','rewardedTime','clearCount','rewardedCount')}})
    return result


class QuestStateObservationProvider:
    def __init__(self,pid,*,session_id,cancel=None,session_factory=None,reader_factory=None):
        if type(pid) is not int or pid<=0 or not isinstance(session_id,str) or not session_id:
            raise ValueError('explicit_pid_and_session_required')
        self.pid,self.session_id=pid,session_id
        self.cancel=cancel or threading.Event()
        self.session_factory=session_factory or QuestStateMemorySession
        self.reader_factory=reader_factory
        self.injected_transport=session_factory is not None or reader_factory is not None

    def read(self,*,allow_process_read=False,expected_character_id=None,expected_process_started_at=None,
             quest_id=None,expected_accept_time=None):
        result={'ok':False,'source':'current-root-typed-quest-collections','observed_at':_now(),
            'read_attempted':False,'memory_writes':0,'game_inputs':0,'heap_scans':0,'client_functions_called':0,
            'privilege_adjustments':0,'live_tested_in_dev2':False,'state_proofs':[],
            'mode':'offline-contract' if self.injected_transport else 'live'}
        if allow_process_read is not True:
            return result|{'reason':'process_read_not_explicitly_enabled'}
        if self.cancel.is_set():return result|{'reason':'cancelled_before_process_open'}
        result['read_attempted']=True
        try:
            options={'cancel':self.cancel}
            if self.reader_factory is not None:options['reader_factory']=self.reader_factory
            with self.session_factory(self.pid,**options) as session:
                snapshot=session.read()
            if self.cancel.is_set():return result|{'reason':'cancelled_after_read'}
            if snapshot.get('pid')!=self.pid:raise QuestStateUnavailable('quest_reader_pid_changed')
            if expected_character_id is not None and str(expected_character_id)!=str(snapshot['identity']['character_id']):
                raise QuestStateUnavailable('quest_reader_character_changed')
            if expected_process_started_at is not None and expected_process_started_at!=snapshot['process_started_at']:
                raise QuestStateUnavailable('quest_reader_process_restarted')
            evidence=state_proofs_from_snapshot(snapshot,session_id=self.session_id,quest_id=quest_id,expected_accept_time=expected_accept_time)
            return result|{'ok':True,'snapshot':snapshot,'observed_at':snapshot['captured_at'],
                'source_version':snapshot['source_version'],**evidence,
                'snapshot_sha256':hashlib.sha256(json.dumps(snapshot,sort_keys=True,ensure_ascii=False,allow_nan=False).encode()).hexdigest()}
        except (OSError,RuntimeError,ValueError,KeyError,TypeError) as exc:
            return result|{'reason':'quest_state_read_unavailable:'+str(exc),'error_type':type(exc).__name__}

