"""Current owned quest data -> typed operation requests, never game execution.

Source schemas come from the pinned copied client. Catalog text, goal hints,
inventory changes and arbitrary elapsed time cannot select a program or prove
its effect. The caller supplies a read-only reader after its existing OFF gate.
"""
from copy import deepcopy
from datetime import datetime
from functools import lru_cache
import json
import math
import struct

from .paths import root


class ProgramUnavailable(ValueError):
    pass


@lru_cache(maxsize=1)
def load_schema():
    return json.loads((root() / 'data/arbeit-quest-program/source-schema.json').read_text(encoding='utf8'))


@lru_cache(maxsize=1)
def load_effect_rules():
    return json.loads((root() / 'data/arbeit-quest-program/objective-effect-rules.json').read_text(encoding='utf8'))


def _full(t):
    return t.get('namespace', '') + '.' + t.get('name', '')


def _label(name):
    return name[1:name.index('>')] if name.startswith('<') and '>' in name else name


class QuestProgramReader:
    """Bounded generated data references only; no discovery, scan or invocation."""
    def __init__(self, reader, base, schema=None):
        self.r = reader
        self.base = base
        self.schema = schema or load_schema()
        self.by_name = {_full(s): s for s in self.schema['classes']}
        self.by_id = {s['definition_index']: s for s in self.schema['classes']}
        self.watched = {}
        self.nodes = 0

    def watch(self, address, length):
        data = self.r.read(address, length)
        key = (address, length)
        if key in self.watched and self.watched[key] != data:
            raise ProgramUnavailable('current_program_changed_during_read')
        self.watched[key] = data
        return data

    def finish(self):
        if any(self.r.read(a, n) != value for (a, n), value in self.watched.items()):
            raise ProgramUnavailable('current_program_changed_during_read')

    def width(self, t):
        code = t['type_code']
        if code in (2, 4, 5): return 1
        if code in (3, 6, 7): return 2
        if code in (8, 9, 12): return 4
        if code in (10, 11, 13, 14, 18, 29): return 8
        if code == 17:
            spec = self.by_id.get(t.get('definition_index'))
            fields = [f for f in (spec or {}).get('fields', []) if not f['type'].get('attrs', 0) & 16]
            if len(fields) == 1 and fields[0]['offset'] == 16:
                return self.width(fields[0]['type'])
        if code == 21 and t.get('namespace') == 'System.Collections.Generic': return 8
        raise ProgramUnavailable('source_value_width_unresolved:' + _full(t))

    def atom(self, address, t, depth=0):
        if depth > 5: raise ProgramUnavailable('current_program_depth_limit')
        code = t['type_code']
        formats = {2:'B', 3:'H', 4:'b', 5:'B', 6:'h', 7:'H', 8:'i', 9:'I', 10:'q', 11:'Q', 12:'f', 13:'d'}
        if code in formats:
            fmt = '<' + formats[code]
            value = struct.unpack(fmt, self.watch(address, struct.calcsize(fmt)))[0]
            if code == 2:
                if value not in (0, 1): raise ProgramUnavailable('current_program_boolean_invalid')
                value = bool(value)
            if isinstance(value, float) and not math.isfinite(value):
                raise ProgramUnavailable('current_program_float_nonfinite')
            return {'value': value, 'type_code': code, 'interpreted': True}
        if code == 14:
            pointer = struct.unpack('<Q', self.watch(address, 8))[0]
            return {'value': self.r.string(pointer) if pointer else None, 'declared_type': 'System.String', 'interpreted': True}
        if code == 17:
            spec = self.by_id.get(t.get('definition_index'))
            fields = [f for f in (spec or {}).get('fields', []) if not f['type'].get('attrs', 0) & 16]
            if len(fields) != 1 or fields[0]['offset'] != 16:
                raise ProgramUnavailable('source_value_type_layout_unresolved:' + _full(t))
            decoded = self.atom(address, fields[0]['type'], depth + 1)
            decoded.update(declared_type=_full(t), member=fields[0]['name'], enum=bool(spec.get('enum')))
            if t.get('namespace') == 'Shared.Symbols' and fields[0]['name'] == 'Id':
                decoded.update(table_domain=t['name'][:-2], domain_mapping_verified=True)
            return decoded
        raise ProgramUnavailable('current_program_atom_type_unsupported:' + _full(t))

    def optional(self, pointer, klass, field, shape, depth):
        t = field['type']
        if (_full(t), len(t.get('arguments', []))) != ('MM.QDBL.Definitions.ValueType.Optional`1', 1):
            raise ProgramUnavailable('current_program_generic_value_unsupported:' + _full(t))
        table = self.r.u64(klass + 0x80)
        tp = struct.unpack('<Q', self.watch(table + field['index']*32 + 8, 8))[0]
        if ((self.r.u32(tp + 8) >> 16) & 255) != 21:
            raise ProgramUnavailable('current_program_optional_type_changed')
        generic = struct.unpack('<Q', self.watch(tp, 8))[0]
        closed = struct.unpack('<Q', self.watch(generic + 24, 8))[0] if generic else 0
        if not closed: raise ProgramUnavailable('current_program_optional_closed_class_not_initialized')
        if self.r.class_name(closed) != ('Optional`1', 'MM.QDBL.Definitions.ValueType'):
            raise ProgramUnavailable('current_program_optional_closed_class_mismatch')
        layout = self.r.layout(closed, ['value', '<HasValue>k__BackingField'])
        if self.r.class_name(int(layout['parent_class'], 16)) != ('ValueType', 'System'):
            raise ProgramUnavailable('current_program_optional_value_parent_mismatch')
        vf, hf = layout['fields']; arg = t['arguments'][0]
        if (vf['type_code'], hf['type_code']) != (arg['type_code'], 2):
            raise ProgramUnavailable('current_program_optional_field_types_changed')
        width = layout['instance_size'] - 16
        next_offset = min([f['offset'] for f in shape['fields'] if f['offset'] > field['offset']] + [shape['instance_size']])
        vo, ho = vf['offset'] - 16, hf['offset'] - 16
        payload_width = self.width(arg)
        if not (0 < width <= next_offset-field['offset'] and 0 <= vo <= width-payload_width and 0 <= ho < width and not vo <= ho < vo+payload_width):
            raise ProgramUnavailable('current_program_optional_named_offsets_invalid')
        address = pointer + field['offset']
        self.watch(address, width)
        tag = self.watch(address + ho, 1)[0]
        if tag not in (0, 1): raise ProgramUnavailable('current_program_optional_tag_invalid')
        inner = self.atom(address + vo, arg, depth+1) if tag else {'value': None, 'interpreted': True}
        return dict(inner, has_value=bool(tag), optional=True, optional_argument_type=deepcopy(arg))

    def collection(self, pointer, element, depth):
        if not pointer: return []
        r = self.r; klass = r.u64(pointer); shape = r.layout(klass)
        name, ns = shape['name'], shape['namespace']
        version = None
        if name.endswith('[]'):
            array = pointer; count = r.u64(array+24)
        else:
            for _ in range(4):
                fs = {f['name'].lstrip('_'): f for f in shape['fields']}
                if {'size', 'version'} <= fs.keys() and ('items' in fs or 'array' in fs): break
                parent = int(shape['parent_class'], 16)
                if not parent: break
                shape = r.layout(parent)
            if not ((name == 'List`1' and ns == 'System.Collections.Generic') or (name.startswith(('MMList', 'MMReadOnlyList')) and ns == 'Silvervine.ManualMemory')):
                raise ProgramUnavailable('current_program_collection_class_unsupported:'+ns+'.'+name)
            if not {'size', 'version'} <= fs.keys() or not ('items' in fs or 'array' in fs):
                raise ProgramUnavailable('current_program_collection_fields_missing')
            af = fs.get('items', fs.get('array'))
            array = struct.unpack('<Q', self.watch(pointer+af['offset'], 8))[0]
            count = struct.unpack('<i', self.watch(pointer+fs['size']['offset'], 4))[0]
            version = self.watch(pointer+fs['version']['offset'], 4)
        if not 0 <= count <= 128: raise ProgramUnavailable('current_program_collection_count_invalid')
        ashape = r.layout(r.u64(array)); capacity = r.u64(array+24)
        width = self.width(element)
        expected = element.get('name')
        if not expected:
            expected = {8:'Int32', 10:'Int64', 14:'String'}.get(element['type_code'])
        if not expected or ashape['name'] != expected+'[]' or ashape['element_size'] != width or not count <= capacity <= 4096:
            raise ProgramUnavailable('current_program_collection_element_layout_mismatch')
        if element.get('namespace') and ashape['namespace'] != element['namespace']:
            raise ProgramUnavailable('current_program_collection_element_namespace_mismatch')
        self.watch(array+24, 8); self.watch(array+32, width*count)
        result = []
        for i in range(count):
            address = array+32+i*width
            if element['type_code'] == 18:
                result.append(self.decode(r.u64(address), depth=depth+1))
            else: result.append(self.atom(address, element, depth+1))
        return result

    def decode(self, pointer, *, depth=0, selected_fields=None):
        if not pointer: return {'available': False, 'missing': ['current_program_reference_null']}
        self.nodes += 1
        if depth > 5 or self.nodes > 256: raise ProgramUnavailable('current_program_reference_budget_exceeded')
        r = self.r; klass = struct.unpack('<Q', self.watch(pointer, 8))[0]
        name, ns = r.class_name(klass); spec = self.by_name.get(ns+'.'+name)
        result = {'type': ns+'.'+name, 'values': {}, 'missing': [], 'source': 'current_owned_typed_quest_program', 'available': False}
        if spec is None:
            result['missing'].append('current_program_type_not_in_source_schema'); return result
        shape = r.layout(klass, [f['name'] for f in spec['fields']])
        disc = next((g for g in spec.get('getters', []) if g['name'] == 'get_Type' and 'constant' in g), None)
        if disc:
            expected = bytes.fromhex(disc['verified_bytes'])
            if self.watch(self.base+disc['rva'], len(expected)) != expected:
                raise ProgramUnavailable('current_program_type_getter_changed:'+name)
            result['discriminant'] = {'value': disc['constant'], 'declared_type': _full(disc['return_type']), 'basis': 'pinned_get_Type_constant_bytes', 'verified': True}
        for source, current in zip(spec['fields'], shape['fields']):
            if source['type'].get('attrs', 0) & 16: continue
            key = _label(source['name'])
            if selected_fields is not None and key not in selected_fields: continue
            if (source['offset'], source['type']['type_code']) != (current['offset'], current['type_code']):
                raise ProgramUnavailable('current_program_field_layout_changed:'+name+'.'+key)
            address = pointer+source['offset']; t = source['type']; code = t['type_code']
            try:
                if code == 21 and _full(t) == 'MM.QDBL.Definitions.ValueType.Optional`1':
                    value = self.optional(pointer, klass, source, shape, depth)
                elif code == 21 and t.get('namespace') == 'System.Collections.Generic' and len(t.get('arguments', [])) == 1:
                    value = {'items': self.collection(struct.unpack('<Q', self.watch(address, 8))[0], t['arguments'][0], depth), 'interpreted': True}
                elif code == 29:
                    value = {'items': self.collection(struct.unpack('<Q', self.watch(address, 8))[0], t['element'], depth), 'interpreted': True}
                elif code == 18:
                    value = self.decode(struct.unpack('<Q', self.watch(address, 8))[0], depth=depth+1)
                else:
                    width = self.width(t)
                    if source['offset'] < 16 or source['offset']+width > shape['instance_size']:
                        raise ProgramUnavailable('current_program_field_outside_instance')
                    value = self.atom(address, t, depth+1)
                result['values'][key] = value
            except (OSError, ValueError, KeyError, UnicodeError) as exc:
                result['values'][key] = {'interpreted': False, 'missing': str(exc), 'declared_type': deepcopy(t)}
                result['missing'].append(key+':'+str(exc))
        result['available'] = True
        # A successfully parsed parent/list is not a fully interpreted program
        # when a referenced child retained an unsupported operand.
        from .typed_objective_operands import collect_operand_missing
        result['missing'] = list(dict.fromkeys(result['missing'] + collect_operand_missing(
            {'values': result['values']})))
        return result


def read_owned_programs(reader, base, active, classes, *, step_id):
    """Called inside the existing twice-read owned ActiveQuest snapshot."""
    program = QuestProgramReader(reader, base)
    af = {f['name']: f['offset'] for f in classes['ActiveQuest'][1]['fields']}
    of = {f['name']: f['offset'] for f in classes['QuestObjective'][1]['fields']}
    listing = reader.u64(active+af['<Objectives>k__BackingField'])
    rows = []
    for pointer in reader.quest_list(listing):
        if reader.u64(pointer) != classes['QuestObjective'][0] or reader.u64(pointer+of['<OwnerActiveQuest>k__BackingField']) != active:
            raise ProgramUnavailable('current_program_objective_owner_mismatch')
        index = reader.i32(pointer+of['<Index>k__BackingField'])
        if index < 0 or index >= 1024 or any(x['objective_index'] == index for x in rows):
            raise ProgramUnavailable('current_program_objective_index_invalid')
        row = {'objective_index': index, 'step_id': step_id, 'owner_verified': True}
        for key, field in [('data', '<Data>k__BackingField'), ('auto_play', '<AutoPlayDescriptor>k__BackingField')]:
            address = pointer+of[field]
            target = struct.unpack('<Q', program.watch(address, 8))[0]
            row[key] = program.decode(target)
        try:
            row['typed_effect']=read_objective_effect(program,pointer,active,classes,row['data'])
        except (OSError,ValueError,KeyError,UnicodeError) as exc:
            row['typed_effect']={'confirmed':False,'complete':None,'missing':[str(exc)]}
        rows.append(row)
    if reader.u64(active+af['<Objectives>k__BackingField']) != listing:
        raise ProgramUnavailable('current_program_objective_collection_changed')
    program.finish()
    return rows


def read_offer_definition(reader, base, static):
    program = QuestProgramReader(reader, base)
    result = program.decode(static, selected_fields={'Id', 'UniqueName', 'Sender', 'SenderNPCProfile', 'IsAcceptAutomatically'})
    if result.get('type') != 'Shared.QDBLs.QDBLXData.QuestStaticData':
        raise ProgramUnavailable('current_offer_static_class_mismatch')
    program.finish()
    values = result['values']
    sender = values.get('Sender', {})
    unique = values.get('UniqueName', {})
    return {'quest_name': None,
        'quest_name_id': unique.get('value') if unique.get('interpreted') else None,
        'quest_name_type': unique.get('declared_type'),
        'giver_definition': {'kind': 'npc', 'npc_id': sender.get('value'), 'declared_type': sender.get('declared_type'),
            'source': 'current_QuestStaticData.Sender', 'current_dialogue_owner_verified': False} if sender.get('interpreted') else None,
        'sender_profile': values.get('SenderNPCProfile'), 'accept_automatically': values.get('IsAcceptAutomatically'),
        'source': result['source'], 'missing': result['missing']}


def read_objective_effect(program, pointer, active, classes, data):
    """Interpret pinned IsAchieved/GetGoal semantics; no getter invocation."""
    r=program.r;rules=load_effect_rules();base=program.base
    if any(rules[k]!=program.schema[k] for k in ('game_assembly_sha256','metadata_sha256')):
        raise ProgramUnavailable('current_objective_effect_source_fingerprint_mismatch')
    for check in rules['code_checks']:
        raw=bytes.fromhex(check['bytes_hex'])
        if program.watch(base+check['rva'],len(raw))!=raw:
            raise ProgramUnavailable('current_objective_effect_code_changed:'+check['name'])
    klass=classes['QuestObjective'][0]
    if r.u64(pointer)!=klass:
        raise ProgramUnavailable('current_objective_effect_class_changed')
    for name,slot in rules['vtable'].items():
        if struct.unpack('<Q',program.watch(klass+slot['offset'],8))[0]!=base+slot['rva']:
            raise ProgramUnavailable('current_objective_effect_virtual_method_changed:'+name)
    optional=r.u64(base+rules['optional_multiple_class_slot_rva'])
    if r.class_name(optional)!=('OptionalMultipleObjectiveQuestObjectiveData','MM.QDBL.CodeGenerated'):
        raise ProgramUnavailable('current_objective_optional_type_slot_unresolved')
    fields={f['name']:f for f in classes['QuestObjective'][1]['fields']}
    for field,offset,code in [('<Count>k__BackingField',48,10),('goal',72,10),('<OwnerActiveQuest>k__BackingField',64,18)]:
        if (fields[field]['offset'],fields[field]['type_code'])!=(offset,code):
            raise ProgramUnavailable('current_objective_effect_named_layout_changed:'+field)
    if struct.unpack('<Q',program.watch(pointer+64,8))[0]!=active:
        raise ProgramUnavailable('current_objective_effect_owner_changed')
    disc=data.get('discriminant') or {}
    if not disc.get('verified') or disc.get('declared_type')!='MM.QDBL.CodeGenerated.NonLocale.QuestObjectiveType':
        raise ProgramUnavailable('current_objective_effect_type_discriminant_required')
    count=struct.unpack('<q',program.watch(pointer+48,8))[0]
    raw_goal=struct.unpack('<q',program.watch(pointer+72,8))[0]
    goal=raw_goal;stack=None
    if disc['value'] in rules['stackable_discriminants']:
        af={f['name']:f for f in classes['ActiveQuest'][1]['fields']}
        f=af['<CurrentStackCount>k__BackingField']
        if (f['offset'],f['type_code'])!=(164,8):
            raise ProgramUnavailable('current_objective_stack_count_layout_changed')
        stack=struct.unpack('<i',program.watch(active+164,4))[0]
        goal=((raw_goal*stack+(1<<63))%(1<<64))-(1<<63)
    optional_multiple=data['type']=='MM.QDBL.CodeGenerated.OptionalMultipleObjectiveQuestObjectiveData'
    return {'confirmed':True,'complete':True if optional_multiple else goal<=count,
        'count':count,'goal':goal,'raw_goal':raw_goal,'stack_count':stack,
        'semantics':'current_quest_objective_native_IsAchieved',
        'basis':'pinned_IsAchieved_GetGoal_IsStackableObjective_and_current_vtable_data_owner',
        'optional_multiple_native_bypass':optional_multiple}


# These are local operation selectors, not invented game APIs. Exact generated
# type and pinned get_Type discriminant select them; catalog kinds never do.
_OPERATIONS = {
    'TalkQuestObjectiveData': 'dialogue.talk', 'TalkGroupTagNPCQuestObjectiveData': 'dialogue.talk',
    'InteractInstantPropQuestObjectiveData': 'objective.inspect',
    'GainFakeItemByInteractInstantPropQuestObjectiveData': 'objective.inspect',
    'ReadStoryBoxQuestObjectiveData': 'objective.read_story',
    'GatheringSuccessQuestObjectiveData': 'life.gather', 'GatheringSuccessGroupTagGatherableQuestObjectiveData': 'life.gather',
    'CraftingSuccessQuestObjectiveData': 'life.craft', 'AlteringSuccessQuestObjectiveData': 'life.alter',
    'HoldConsumableQuestObjectiveData': 'inventory.hold', 'HoldEquipmentQuestObjectiveData': 'inventory.hold',
    'HoldGroupTagConsumableQuestObjectiveData': 'inventory.hold',
    'DeliverConsumablesToNPCQuestObjectiveData': 'objective.deliver', 'DeliverEquipmentsToNPCQuestObjectiveData': 'objective.deliver',
    'DeliverFakeItemsToNPCQuestObjectiveData': 'objective.deliver', 'DeliverMultipleItemsToNPCQuestObjectiveData': 'objective.deliver',
    'DeliverMultipleFakeItemsToNPCQuestObjectiveData': 'objective.deliver', 'DeliverMultipleConsumablesToNPCQuestObjectiveData': 'objective.deliver',
    'DeliverMultipleItemsToGroupTagNPCQuestObjectiveData': 'objective.deliver',
    'DeliverCostumesToNPCQuestObjectiveData': 'objective.deliver', 'DeliverMultipleCostumesToNPCQuestObjectiveData': 'objective.deliver',
    'DeliverRidableToNPCQuestObjectiveData': 'objective.deliver',
    'DeliverGroupTagConsumablesToNPCQuestObjectiveData': 'objective.deliver',
    'ArriveAtRegionQuestObjectiveData': 'objective.move', 'MoveToPermanentChannelQuestObjectiveData': 'objective.move',
    'ClearDungeonQuestObjectiveData': 'objective.dungeon_clear', 'ClearDungeonCategoryQuestObjectiveData': 'objective.dungeon_clear',
    'DungeonFinishObjectiveClearQuestObjectiveData': 'objective.dungeon_clear',
    'UseQuestFakeItemQuestObjectiveData': 'objective.use_quest_item', 'UseConsumableQuestObjectiveData': 'objective.use_item',
    'KillMonsterQuestObjectiveData': 'objective.combat', 'KillGroupTagMonsterQuestObjectiveData': 'objective.combat',
    'GainFakeItemByKillMonsterQuestObjectiveData': 'objective.combat',
    'PlayCutsceneQuestObjectiveData': 'objective.cutscene', 'WaitRealTimeQuestObjectiveData': 'objective.wait',
    'CompleteQuestStepQuestObjectiveData': 'quest.subgoal', 'CompleteQuestQuestObjectiveData': 'quest.subgoal',
    'MultipleObjectiveQuestObjectiveData': 'objective.composite', 'OptionalMultipleObjectiveQuestObjectiveData': 'objective.composite',
    'OrQuestObjectiveData': 'objective.composite',
}
_SCOPE = ('quest_id', 'instance_id', 'accept_time', 'step_id', 'step_index', 'objective_index')


def _valid_scope(current, expected):
    if not current.get('confirmed') or not current.get('owner_verified'):
        return False
    if not all(current.get(k) is not None for k in _SCOPE): return False
    if any(current.get(k) != expected.get(k) for k in _SCOPE if k in expected): return False
    if not all(current.get(k) for k in ('session_id', 'character_id', 'observed_at', 'source')): return False
    try:
        if datetime.fromisoformat(current['observed_at'].replace('Z', '+00:00')).tzinfo is None: return False
    except (ValueError, TypeError): return False
    return True


def build_program(current, *, expected=None):
    """Return a concrete local request, still requiring the physical UI port.

    A request does not mean a click target exists or an effect was observed.
    The executor must match the same occurrence and current UI target.
    """
    result = {'ready': False, 'classification': 'B_current_observation_required', 'missing': [], 'action_request': None}
    if not _valid_scope(current, expected or {}):
        result['missing'] = ['current_exact_owned_objective_scope']; return result
    data, auto = current.get('data') or {}, current.get('auto_play') or {}
    type_name = data.get('type', '').removeprefix('MM.QDBL.CodeGenerated.')
    if not data.get('available') or not (data.get('discriminant') or {}).get('verified'):
        result['missing'] = ['current_typed_objective_discriminant']; return result
    schemas={_full(s):s for s in load_schema()['classes']}
    for label, payload, suffix in [('objective',data,'QuestObjectiveData'),('auto_play',auto,'AutoPlayDescriptorData')]:
        spec=schemas.get(payload.get('type'))
        proof=next((g for g in (spec or {}).get('getters',[]) if g['name']=='get_Type' and 'constant' in g),None)
        if not spec or not spec['name'].endswith(suffix) or not proof or (payload.get('discriminant') or {}).get('value')!=proof['constant']:
            result['missing']=['current_'+label+'_discriminant_source_mismatch'];return result
    # Other source-declared objectives keep their exact semantics and delegate
    # only to the client's already current objective autoplay control. No
    # fabricated per-type input sequence or goal predicate is constructed.
    op = _OPERATIONS.get(type_name,'objective.native')
    scope = {k: current[k] for k in _SCOPE}
    basis = {k: current[k] for k in ('source', 'observed_at', 'session_id', 'character_id')}
    basis.update(process_started_at=current.get('process_started_at'), source_version=current.get('source_version'))
    semantic = {'operation': op, 'data_type': data['type'], 'discriminant': deepcopy(data['discriminant']),
        'operands': deepcopy(data.get('values', {})), 'auto_play': deepcopy(auto),
        'goal_semantics': 'current_typed_objective; not inventory_delta or catalog_text',
        'effect': dict(scope, state='objective_complete', basis='current_objective_success_or_verified_current_step_predicate')}
    # The client's unsupported descriptor is an explicit refusal, not an action.
    if auto.get('type') == 'MM.QDBL.CodeGenerated.NotSupportedAutoPlayDescriptorData':
        return dict(result, classification='A_source_handler_required', missing=['native_auto_play_explicitly_not_supported'], semantics=semantic)
    if not auto.get('available') or not (auto.get('discriminant') or {}).get('verified'):
        return dict(result, missing=['current_typed_auto_play_descriptor'], semantics=semantic)
    # Preserve all unresolved operands; no default recipe, NPC, region, quantity
    # or ordinary gathering route is substituted for a current quest target.
    from .typed_objective_operands import collect_operand_missing
    gaps = ['objective:'+m for m in collect_operand_missing(data)] + ['auto_play:'+m for m in collect_operand_missing(auto)]
    if gaps:
        return dict(result, missing=gaps, semantics=semantic)
    request = dict(basis, **scope, kind='quest.objective.autoplay', operation=op,
        semantics=semantic, expected_effect=semantic['effect'],
        requires=['current_same_objective_auto_play_control', 'source_guarded_physical_input_port'],
        current_target_required=True, dispatch_state='not_sent')
    return dict(result, ready=True, classification='A_source_implemented_B_current_UI_required',
                semantics=semantic, action_request=request)


def bind_current_offer(offer, dialogue):
    """Merge only an independently proven current accept action owner.

    A dialogue condition mentioning a quest or the static Sender is not such
    a proof. No field is synthesized from eligible alone.
    """
    result = deepcopy(offer)
    result.update(offered=False, dialogue_giver_binding=None, not_current_npc_offer=True)
    if not isinstance(dialogue, dict): return result
    if not (offer.get('eligible') is True and offer.get('conditions_met') is True): return result
    if not (dialogue.get('confirmed') is True and dialogue.get('action_owner_verified') is True and dialogue.get('action') == 'accept'):
        return result
    required = ('quest_id', 'instance_id', 'session_id', 'character_id', 'process_started_at')
    if any(offer.get(k) is None or dialogue.get(k) != offer[k] for k in required): return result
    if offer.get('quest_name') is not None and dialogue.get('quest_name') != offer['quest_name']: return result
    if not dialogue.get('source') or not dialogue.get('observed_at') or not dialogue.get('ownership_basis'): return result
    npc = dialogue.get('npc_id'); giver = dialogue.get('giver')
    if type(npc) is not int or not isinstance(giver, dict) or giver.get('npc_id') != npc: return result
    result.update(offered=True, npc_id=npc, giver=deepcopy(giver), dialogue_giver_binding=deepcopy(dialogue), not_current_npc_offer=False)
    # The join's freshness is bounded by both observations, not restamped.
    try:
        stamps=[datetime.fromisoformat(row['observed_at'].replace('Z','+00:00')) for row in (offer,dialogue)]
        if any(s.tzinfo is None for s in stamps):return deepcopy(offer)
        result['observed_at']=min(stamps).isoformat()
    except (ValueError,TypeError,KeyError):return deepcopy(offer)
    return result
