"""Durable recipe-material children, separate from native quest hold objectives.

Pure APIs (no process, UI, file write, timer completion or input dispatch):
  begin_material_supply(parent_node, snapshot, parent_pending, ancestry=())
      -> {status: ready|not_needed|waiting, state?, missing, reason}
  advance_material_supply(state, snapshot, field_return_evidence=None,
                          child_result=None)
      -> {status: return_to_field|child|resume_parent|waiting, state,
          node?, material_request?, resume?, missing, reason}
  record_material_child_pending(state, pending) -> new state
  validate_material_request(node, snapshot, request) -> bool
  make_material_hold_request(child_node, snapshot) -> separate request (ready)
  reconcile_material_pass(state, snapshot, field_return_evidence=None)
      -> {status: child|resume_parent|return_to_field|waiting, state, ...}

The caller journals every returned state BEFORE dispatch. state.parent_pending
is an immutable copy of the original whole graph pending record. A child input
is stored separately. Resuming clears only resolved navigation pending; it never
restores the old navigation phase after the UI has returned to the field.

For status=child, observe the returned node, then supply material_request under
snapshot.material_hold_requests. It has requirement_origin=source_recipe_material
and quest_objective=False; it must NOT be relabelled as an observed native quest
objective, or carry a fabricated native IsAchieved flag. Its quantity is the
total usable bag amount required by this recipe plus separately validated other
native goal commitments, not the remaining deficit. source_required_quantity
retains the unchanged recipe amount. Inventory quantities are not rewritten.
The caller uses the existing inventory.hold planner and ordinary input port.

Nested children pass state.ancestry to begin_material_supply. A repeated recipe
or a material that is already an ancestor output is a named dependency cycle.
Occurrence/step changes retain all journals and require parent reconciliation.
"""
from copy import deepcopy
from datetime import datetime, timezone
import hashlib
import json

SCHEMA='masukja.material-supply.v1'
SCOPE=('quest_id','instance_id','accept_time','step_id','step_index','process_started_at')
OWNER=('session_id','character_id')
BLOCKED_PHASES={'alter.recipe_detail','alter.collect','craft.submit'}
MAX_DEPTH=16


class MaterialSupplyUnavailable(ValueError):
    pass


def _digest(value):
    return hashlib.sha256(json.dumps(value,sort_keys=True,ensure_ascii=False,separators=(',',':')).encode()).hexdigest()


def _stamp(value):
    value=datetime.fromisoformat(value.replace('Z','+00:00'))
    if value.tzinfo is None:raise MaterialSupplyUnavailable('material_time_zone_missing')
    return value


def _fresh(value,snapshot):
    from .arbeit_operations import _fresh as fresh
    return fresh(value,snapshot)


def _int(value,*,positive=False):
    if type(value) is not int or not (1 if positive else 0)<=value<=2147483647:
        raise MaterialSupplyUnavailable('material_quantity_must_be_nonnegative_integer_units')
    return value


def _current_scope(node,snapshot):
    current=snapshot.get('current_quest') or (snapshot.get('quest_state_read') or {}).get('current_quest') or {}
    if (not _fresh(current,snapshot) or current.get('confirmed') is not True
            or current.get('source')!='current-root-typed-quest-collections'
            or any(current.get(k) is None for k in SCOPE)):
        raise MaterialSupplyUnavailable('material_current_parent_quest_occurrence_required')
    if current.get('quest_id')!=node.get('quest_id') or current.get('step_id')!=node.get('step_id'):
        raise MaterialSupplyUnavailable('material_parent_goal_reconciliation_required')
    if type(current['accept_time']) is not int or current['accept_time']<=0:
        raise MaterialSupplyUnavailable('material_current_accepted_occurrence_required')
    inherited=node.get('params',{}).get('material_requirement',{}).get('parent_scope')
    if inherited and any(current.get(k)!=inherited.get(k) for k in SCOPE):
        raise MaterialSupplyUnavailable('material_parent_goal_reconciliation_required')
    identity=snapshot.get('identity') or {}
    if identity.get('process_started_at') is not None and identity['process_started_at']!=current['process_started_at']:
        raise MaterialSupplyUnavailable('material_parent_process_changed')
    return {**{k:current[k] for k in SCOPE},**{k:snapshot[k] for k in OWNER},
            'objective_index':node.get('objective_index'),'source':current['source'],
            'observed_at':current['observed_at']}


def _item_key(item):
    if item.get('table_domain') not in ('ConsumableSheet','QuestFakeItemSheet') or type(item.get('item_id')) is not int or not item.get('name'):
        raise MaterialSupplyUnavailable('material_exact_consumable_source_identity_required')
    return item['table_domain']+':'+str(item['item_id'])


def _bag(item,snapshot):
    inventory=snapshot.get('inventory') or {}
    if (not _fresh(inventory,snapshot) or inventory.get('complete') is not True
            or inventory.get('source') not in ('get_items','official:get_items','official-cli:get_items')
            or inventory.get('scope')!='official_consumable_items_only' or not isinstance(inventory.get('items'),list)):
        raise MaterialSupplyUnavailable('material_fresh_complete_consumable_bag_required')
    _item_key(item)
    rows=[]
    for row in inventory['items']:
        if row.get('scope')!='bag':continue
        if row.get('name')!=item['name'] and row.get('item_id')!=item['item_id']:continue
        if row.get('name')!=item['name'] or (row.get('item_id') is not None and row['item_id']!=item['item_id']):
            raise MaterialSupplyUnavailable('material_observed_identity_conflict:'+item['name'])
        rows.append(row)
    categories={r.get('category') for r in rows}
    if rows and (len(categories)!=1 or None in categories or '' in categories):
        raise MaterialSupplyUnavailable('material_bag_category_ambiguous:'+item['name'])
    available=0
    for row in rows:
        quantity,locked,reserved=(_int(row.get('quantity')),_int(row.get('locked_quantity')),
                                  _int(row.get('reserved_quantity',0)))
        if locked+reserved>quantity:raise MaterialSupplyUnavailable('material_protection_exceeds_quantity')
        available+=quantity-locked-reserved
    return dict(source=inventory['source'],observed_at=inventory['observed_at'],
        **{k:snapshot[k] for k in OWNER},available_quantity=available,
        item=dict(deepcopy(item),category=next(iter(categories)) if categories else None),
        observed_items=deepcopy(rows),inventory_scope='unlocked_unreserved_bag',
        identity_binding='exact_source_name_and_unique_observed_category; source_id_is_not_an_observed_ID',
        response_sha256=inventory.get('response_sha256'),complete_inventory=True)


def material_inventory(item,snapshot,*,parent_node_id):
    """Actual usable bag plus separate other-goal protection, never double-count.

    get_items locked/reserved quantities are already handled by _bag. Recipe
    material_protections are cleanup policy, not another stock deduction.
    """
    if item.get('table_domain')=='QuestFakeItemSheet':
        from .objective_inventory_material import fake_material_inventory
        return fake_material_inventory(item,snapshot,parent_node_id=parent_node_id)
    from .goal_reservations import goal_commitment_for_item
    proof=_bag(item,snapshot)
    commitment=goal_commitment_for_item(snapshot,item,exclude_node_id=parent_node_id)
    protected=_int(commitment['required_quantity'])
    return dict(proof,goal_commitment=commitment,
        uncommitted_quantity=max(0,proof['available_quantity']-protected),
        inventory_quantity_modified=False)


def _supply_quantity(required,inventory):
    return _int(required+inventory['goal_commitment']['required_quantity'],positive=True)


def _processing_requirements(scene):
    from .processing_observer import processing_recipe_bindings
    catalog=processing_recipe_bindings()
    rows=[r for r in catalog['bindings'] if r['route_id']==scene.get('recipe_id')]
    if len(rows)!=1 or scene.get('recipe_definition')!=rows[0]:
        raise MaterialSupplyUnavailable('material_current_altering_source_recipe_binding_required')
    row=rows[0];definition=row['current']
    if (scene.get('stage')!='recipe_detail' or scene.get('source_bound') is not True
            or scene.get('exact_recipe_confirmed') is not True or scene.get('recipe_intent_bound') is not True
            or not scene.get('recipe_source_refs')):
        raise MaterialSupplyUnavailable('material_current_owned_altering_detail_required')
    expected=[dict(name=r['name'],required=r['required']) for r in definition['materials']]
    if scene.get('current_materials')!=expected:
        raise MaterialSupplyUnavailable('material_altering_requirement_definition_changed')
    requirements=[dict(item={'table_domain':'ConsumableSheet','item_id':r['raw_id'],'name':r['name'],
        'source_ref':definition['source_ref']},required_quantity=_int(r['required'],positive=True),
        quantity_unit='consumable_item_count',batches=1,per_batch=r['required']) for r in definition['materials']]
    recipe=dict(kind='native_alter',recipe_id=row['route_id'],table_domain='AlteringSheet',table_id=row['altering_sheet_id'],
        name=definition['display_name'],batches=1,definition_digest=_digest(row),
        source_refs=deepcopy(scene['recipe_source_refs']),source_ref=deepcopy(definition['source_ref']),
        source_revision=catalog['client_revision_sha256'])
    output_keys=['ConsumableSheet:'+str(definition['output_item_id'])]
    return recipe,requirements,output_keys


def _craft_requirements(node,scene,snapshot):
    from .craft_observer import craft_sources,craft_request_node,recipe_candidates
    bound,_=craft_request_node(node,snapshot)
    rows=[r for r in recipe_candidates(bound) if r['id']==scene.get('recipe_id')]
    if len(rows)!=1:raise MaterialSupplyUnavailable('material_current_craft_source_recipe_required')
    recipe=rows[0]
    if (scene.get('stage')!='detail' or scene.get('selected_recipe_id')!=recipe['id']
            or scene.get('current_binding_kind')!='typed current native recipe'
            or recipe['source_ref'] not in scene.get('source_refs',[])):
        raise MaterialSupplyUnavailable('material_current_owned_craft_detail_required')
    batches=_int((scene.get('quantity') or {}).get('value'),positive=True)
    wanted=snapshot.get('craft_request')
    if wanted is not None:
        if (not _fresh(wanted,snapshot) or wanted.get('node_id')!=node['id']
                or wanted.get('recipe_id')!=recipe['id'] or wanted.get('batches')!=batches):
            raise MaterialSupplyUnavailable('material_current_craft_batch_intent_changed')
    elif batches!=1:
        # The existing craft planner resets to one in the absence of a current
        # batch request. Do not gather for a stale manually entered quantity.
        raise MaterialSupplyUnavailable('material_craft_quantity_not_current_requested_batches')
    expected=[dict(r,domain='ConsumableSheet') for r in recipe['consumable_ingredients']]+[
        dict(r,name=r['display_name'],domain='QuestFakeItemSheet') for r in recipe['fake_item_ingredients']]
    materials=scene.get('materials')
    if not isinstance(materials,list) or len(materials)!=len(expected):
        raise MaterialSupplyUnavailable('material_craft_current_requirement_row_count')
    requirements=[];used=set()
    for need in expected:
        matches=[(i,r) for i,r in enumerate(materials) if r.get('item',{}).get('displayName')==need['name']]
        if len(matches)!=1 or matches[0][0] in used:raise MaterialSupplyUnavailable('material_craft_label_ambiguous')
        index,row=matches[0];used.add(index)
        required=_int(_int(need['amount'],positive=True)*batches,positive=True)
        if _int(row.get('requiredCount'))!=required:raise MaterialSupplyUnavailable('material_craft_source_quantity_changed')
        ui_count=_int(row.get('inventoryCount'))
        item={'table_domain':need['domain'],'item_id':need['raw_id'],'name':need['name'],'source_ref':recipe['source_ref']}
        unit='consumable_item_count'
        if need['domain']=='QuestFakeItemSheet':
            item['source_material_binding']={'recipe_id':recipe['id'],'source_ref':deepcopy(recipe['source_ref'])}
            observed=material_inventory(item,snapshot,parent_node_id=node['id'])
            if ui_count!=observed['quantity']:raise MaterialSupplyUnavailable('material_fake_current_Amount_and_UI_disagree')
            unit='quest_fake_item_count'
        requirements.append(dict(item=item,required_quantity=required,quantity_unit=unit,
            batches=batches,per_batch=need['amount'],ui_inventory_count=ui_count,
            ui_inventory_count_scope='not_assumed_to_be_unlocked_bag'))
    sources=craft_sources()
    description=dict(kind='native_craft',recipe_id=recipe['id'],table_domain='CraftingRecipeSheet',table_id=recipe['id'],
        name=recipe['display_name'],batches=batches,definition_digest=_digest(recipe),
        source_ref=deepcopy(recipe['source_ref']),source_refs=deepcopy(scene['source_refs']),
        source_revision=sources['declarations']['metadata_sha256'])
    return description,requirements,['ConsumableSheet:'+str(r['raw_id']) for r in recipe['consumable_results']]


def _navigation(node,snapshot,pending,scene,recipe):
    if not isinstance(pending,dict) or not pending.get('operation_id'):
        raise MaterialSupplyUnavailable('material_owned_parent_navigation_pending_required')
    saved=pending.get('planner_receipt') or {};child=saved.get('child_receipt') or {};phase=saved.get('phase')
    if phase in BLOCKED_PHASES:raise MaterialSupplyUnavailable('material_consumption_pending_must_not_be_suspended:'+phase)
    allowed={'native_alter':{'alter.recipe_list'},'native_craft':{'craft.open','craft.category','craft.recipe','craft.quantity'}}
    if phase not in allowed[recipe['kind']]:raise MaterialSupplyUnavailable('material_pending_is_not_detail_navigation')
    goal_scope=saved.get('goal_scope') or child.get('goal_scope')
    current_scope=_current_scope(node,snapshot)
    if not isinstance(goal_scope,dict) or any(goal_scope.get(k) is None for k in SCOPE):
        raise MaterialSupplyUnavailable('material_parent_navigation_occurrence_unbound')
    if any(goal_scope.get(k)!=current_scope[k] for k in SCOPE):
        raise MaterialSupplyUnavailable('material_parent_goal_reconciliation_required')
    if (saved.get('node_id')!=node['id'] or any(saved.get(k)!=snapshot.get(k) for k in OWNER)
            or child.get('recipe_id')!=recipe['recipe_id'] or not saved.get('observed_at')
            or _stamp(scene['observed_at'])<=_stamp(saved['observed_at'])):
        raise MaterialSupplyUnavailable('material_parent_navigation_scope_or_effect_time_changed')
    if pending.get('status') not in ('sent_unconfirmed','unknown','effect_confirmed'):
        raise MaterialSupplyUnavailable('material_parent_navigation_dispatch_not_owned')
    owner=snapshot.get('input_owner') or {}
    if (owner.get('continuous') is not True or owner.get('epoch') is None or not _fresh(owner,snapshot)
            or (owner.get('owner') or {}).get('process_started_at')!=current_scope['process_started_at']):
        raise MaterialSupplyUnavailable('material_current_continuous_input_owner_required')
    if recipe['kind']=='native_alter':
        if scene.get('input_epoch') is None or child.get('input_epoch')!=scene['input_epoch'] or owner['epoch']!=scene['input_epoch']:
            raise MaterialSupplyUnavailable('material_altering_input_epoch_changed')
    else:
        expected={k:owner.get(k) for k in ('epoch','character_id','session_id')}
        if (owner.get('continuous') is not True or owner.get('epoch') is None
                or scene.get('input_owner')!=owner or child.get('craft_owner')!=expected
                or child.get('window_signature')!=scene.get('window_signature')):
            raise MaterialSupplyUnavailable('material_craft_input_owner_changed')
    return dict(source=scene['source'],observed_at=scene['observed_at'],**{k:snapshot[k] for k in OWNER},
        parent_node_id=node['id'],recipe_id=recipe['recipe_id'],operation_id=pending['operation_id'],
        confirmed=True,effect='current_same_recipe_detail_resolves_navigation_only',consumption_confirmed=False)


def _wait(reason,state=None):
    return dict(status='waiting',reason=reason,missing=[reason],state=deepcopy(state),game_inputs=0)


def begin_material_supply(parent_node,snapshot,parent_pending,*,ancestry=()):
    """Snapshot parent intent and current source requirements before leaving UI."""
    try:
        scope=_current_scope(parent_node,snapshot)
        scenes=[s for key in ('native_processing','crafting') if (s:=snapshot.get(key))
                and s.get('stage') in ('recipe_detail','detail')]
        if len(scenes)!=1:raise MaterialSupplyUnavailable('material_unique_current_recipe_detail_required')
        scene=scenes[0]
        if scene.get('confirmed') is not True or not _fresh(scene,snapshot):
            raise MaterialSupplyUnavailable('material_fresh_confirmed_recipe_detail_required')
        if scene is snapshot.get('native_processing'):
            recipe,requirements,outputs=_processing_requirements(scene)
        else:recipe,requirements,outputs=_craft_requirements(parent_node,scene,snapshot)
        navigation=_navigation(parent_node,snapshot,parent_pending,scene,recipe)
        if not requirements:return dict(status='not_needed',reason='no_supported_consumable_material_requirement',missing=[])
        if len({_item_key(r['item']) for r in requirements})!=len(requirements):
            raise MaterialSupplyUnavailable('material_duplicate_source_identity_requires_quantity_rule')
        proofs=[material_inventory(r['item'],snapshot,parent_node_id=parent_node['id']) for r in requirements]
        shortages=[dict(deepcopy(r),initial_inventory=proof,initial_shortage=max(0,_supply_quantity(r['required_quantity'],proof)-proof['available_quantity']))
            for r,proof in zip(requirements,proofs) if proof['uncommitted_quantity']<r['required_quantity']]
        if recipe['kind']=='native_alter':
            provided=scene.get('material_shortages')
            if not isinstance(provided,list):raise MaterialSupplyUnavailable('material_processing_shortage_observations_required')
            for need in shortages:
                # The processing UI/provider reports raw unlocked bag shortage.
                # Native goal protection is a separately verified commitment.
                raw_shortage=max(0,need['required_quantity']-need['initial_inventory']['available_quantity'])
                if not raw_shortage:continue
                matches=[r for r in provided if r.get('requirement')=={'name':need['item']['name'],'required':need['required_quantity']}]
                if (len(matches)!=1 or matches[0].get('identity_resolved') is not True
                        or matches[0].get('shortage')!=raw_shortage
                        or matches[0].get('available_quantity')!=need['initial_inventory']['available_quantity']
                        or matches[0].get('inventory_observed_at')!=need['initial_inventory']['observed_at']):
                    raise MaterialSupplyUnavailable('material_processing_shortage_definition_or_bag_disagrees')
        if not shortages:return dict(status='not_needed',reason='current_bag_covers_source_recipe',missing=[],material_evidence=proofs)
        recipe_key=recipe['table_domain']+':'+str(recipe['table_id'])
        inherited=parent_node.get('params',{}).get('material_requirement',{}).get('ancestry')
        if inherited is not None and ancestry and list(ancestry)!=inherited:
            raise MaterialSupplyUnavailable('material_nested_ancestry_changed')
        lineage=deepcopy(inherited if inherited is not None else list(ancestry))
        if len(lineage)>=MAX_DEPTH:raise MaterialSupplyUnavailable('material_dependency_depth_limit')
        if any(r.get('recipe_key')==recipe_key for r in lineage):raise MaterialSupplyUnavailable('material_dependency_recipe_cycle')
        forbidden=set(outputs)
        for row in lineage:forbidden.update(row.get('output_item_keys',[]))
        if any(_item_key(r['item']) in forbidden for r in shortages):raise MaterialSupplyUnavailable('material_dependency_item_cycle')
        lineage.append(dict(recipe_key=recipe_key,output_item_keys=outputs,parent_node_id=parent_node['id']))
        identity=dict(parent_node_id=parent_node['id'],parent_scope=scope,source_recipe=recipe)
        state=dict(schema=SCHEMA,id=_digest(identity),phase='return_to_field',created_at=scene['observed_at'],
            parent_node=deepcopy(parent_node),parent_pending=deepcopy(parent_pending),parent_scope=scope,
            source_recipe=recipe,requirements=deepcopy(requirements),goals=shortages,index=0,
            child_pending=None,child_completions=[],ancestry=lineage,navigation_resolution_evidence=navigation,
            field_return_evidence=None,requirement_origin='source_recipe_material',quest_objective=False)
        state.update(pass_index=0,pass_history=[],pass_started_at=scene['observed_at'])
        state['immutable_digest']=_immutable(state)
        return dict(status='ready',reason='current_source_recipe_material_shortage',missing=[],state=state,game_inputs=0)
    except (KeyError,TypeError,ValueError,AttributeError,OverflowError) as exc:return _wait(str(exc))


def _immutable(state):
    return _digest({k:state[k] for k in ('schema','id','parent_node','parent_pending','parent_scope','source_recipe','requirements','ancestry')})


def _state_scope(state,snapshot):
    if state.get('schema')!=SCHEMA or state.get('immutable_digest')!=_immutable(state):
        raise MaterialSupplyUnavailable('material_durable_parent_state_changed')
    requirements={_item_key(r['item']):r for r in state['requirements']}
    for goal in state['goals']:
        expected=requirements.get(_item_key(goal['item']))
        if not expected or any(goal.get(k)!=expected[k] for k in ('item','required_quantity','quantity_unit','batches','per_batch')):
            raise MaterialSupplyUnavailable('material_child_requirement_changed')
    current=_current_scope(state['parent_node'],snapshot)
    if any(current.get(k)!=state['parent_scope'].get(k) for k in SCOPE+OWNER):
        raise MaterialSupplyUnavailable('material_parent_goal_reconciliation_required')
    recipe=state['source_recipe']
    if recipe['kind']=='native_alter':
        from .processing_observer import processing_recipe_bindings
        rows=[r for r in processing_recipe_bindings()['bindings'] if r['route_id']==recipe['recipe_id']]
    else:
        from .craft_observer import craft_sources
        rows=[r for r in craft_sources()['recipes']['recipes'] if r['id']==recipe['recipe_id']]
    if len(rows)!=1 or _digest(rows[0])!=recipe['definition_digest']:
        raise MaterialSupplyUnavailable('material_source_recipe_updated_reconciliation_required')


def _field_proof(state,snapshot,proof,after):
    if not isinstance(proof,dict) or not _fresh(proof,snapshot):return False
    scope=state['parent_scope']
    if (proof.get('confirmed') is not True or proof.get('field_hud') is not True
            or proof.get('parent_node_id')!=state['parent_node']['id']
            or proof.get('recipe_id')!=state['source_recipe']['recipe_id']
            or any(proof.get(k)!=scope.get(k) for k in ('instance_id','accept_time','step_id','objective_index','process_started_at'))
            or _stamp(proof['observed_at'])<=_stamp(after)):
        return False
    operation=proof.get('parent_pending_operation_id')
    return operation is None or operation==state['parent_pending']['operation_id']


def _child(state):
    goal=state['goals'][state['index']];scope=state['parent_scope']
    requirement=dict(deepcopy(goal),parent_scope=deepcopy(scope),source_recipe=deepcopy(state['source_recipe']),
        material_state_id=state['id'],parent_node_id=state['parent_node']['id'],
        material_pass_index=state.get('pass_index',0),
        source_required_at=state['created_at'],
        ancestry=deepcopy(state['ancestry']),
        requirement_origin='source_recipe_material',quest_objective=False)
    suffix='/pass/'+str(state.get('pass_index',0)) if state.get('pass_index',0) else ''
    return dict(id=state['parent_node']['id']+'/material/'+_digest([_item_key(goal['item']),state['id']])[:16]+suffix,
        op='inventory.hold',quest_id=scope['quest_id'],step_id=scope['step_id'],objective_index=None,
        params=dict(item=deepcopy(goal['item']),material_requirement=requirement))


def _request(state,snapshot,child):
    result=make_material_hold_request(child,snapshot)
    if result.get('ready') is not True:raise MaterialSupplyUnavailable(result['missing'][0])
    return result


def make_material_hold_request(child_node,snapshot):
    """Refresh only stock/ownership; retain original source requirement/time."""
    try:
        requirement=child_node['params']['material_requirement'];scope=requirement['parent_scope']
        current=_current_scope(child_node,snapshot)
        if (child_node.get('op')!='inventory.hold' or child_node.get('objective_index') is not None
                or requirement.get('requirement_origin')!='source_recipe_material' or requirement.get('quest_objective') is not False
                or any(current[k]!=scope[k] for k in SCOPE+OWNER)):
            raise MaterialSupplyUnavailable('material_parent_goal_reconciliation_required')
        recipe=requirement['source_recipe']
        if recipe['kind']=='native_alter':
            from .processing_observer import processing_recipe_bindings
            rows=[r for r in processing_recipe_bindings()['bindings'] if r['route_id']==recipe['recipe_id']]
            materials=rows[0]['current']['materials'] if len(rows)==1 else []
            exact=[r for r in materials if r['raw_id']==requirement['item']['item_id'] and r['name']==requirement['item']['name']]
            amount=exact[0]['required'] if len(exact)==1 else None
        elif recipe['kind']=='native_craft':
            from .craft_observer import craft_sources
            rows=[r for r in craft_sources()['recipes']['recipes'] if r['id']==recipe['recipe_id']]
            if requirement['item'].get('table_domain')=='QuestFakeItemSheet':
                materials=[dict(r,name=r['display_name']) for r in rows[0]['fake_item_ingredients']] if len(rows)==1 else []
            else:materials=rows[0]['consumable_ingredients'] if len(rows)==1 else []
            exact=[r for r in materials if r['raw_id']==requirement['item']['item_id'] and r['name']==requirement['item']['name']]
            amount=exact[0]['amount']*recipe['batches'] if len(exact)==1 else None
        else:raise MaterialSupplyUnavailable('material_source_recipe_kind_unknown')
        unit='quest_fake_item_count' if requirement['item'].get('table_domain')=='QuestFakeItemSheet' else 'consumable_item_count'
        if (len(rows)!=1 or _digest(rows[0])!=recipe['definition_digest']
                or amount!=requirement['required_quantity'] or requirement.get('quantity_unit')!=unit):
            raise MaterialSupplyUnavailable('material_source_recipe_updated_reconciliation_required')
        _int(amount,positive=True);_stamp(requirement['source_required_at'])
        proof=material_inventory(requirement['item'],snapshot,parent_node_id=requirement['parent_node_id'])
        return dict(source='source_recipe_material_requirement_with_current_bag',observed_at=proof['observed_at'],
        **{k:scope[k] for k in SCOPE+OWNER},node_id=child_node['id'],objective_index=None,
        ready=True,confirmed=True,semantics='inventory.hold',requirement_origin='source_recipe_material',quest_objective=False,
        material_state_id=requirement['material_state_id'],parent_node_id=requirement['parent_node_id'],parent_scope=deepcopy(scope),
        item=proof['item'],item_name=proof['item']['name'],required_quantity=_supply_quantity(requirement['required_quantity'],proof),
        source_required_quantity=requirement['required_quantity'],other_goal_commitment=deepcopy(proof['goal_commitment']),
        quantity_basis='unchanged_source_recipe_requirement_plus_current_other_native_goal_commitments',
        quantity_unit=unit,identity_binding=proof['identity_binding'],
        source_recipe=deepcopy(recipe),source_refs=deepcopy(recipe['source_refs']),source_required_at=requirement['source_required_at'],
        inventory_count_completion_allowed=True,current_inventory_evidence=proof,
        native_quest_effect_supplied=False,missing=[])
    except (KeyError,TypeError,ValueError,AttributeError,OverflowError) as exc:
        return dict(ready=False,confirmed=False,requirement_origin='source_recipe_material',quest_objective=False,missing=[str(exc)])


def validate_material_request(node,snapshot,request):
    """Validate the separate material channel before the ordinary hold planner."""
    try:
        expected=node['params']['material_requirement'];scope=expected['parent_scope'];current=_current_scope(node,snapshot)
        if (node.get('op')!='inventory.hold' or node.get('objective_index') is not None or not _fresh(request,snapshot)
                or request.get('requirement_origin')!='source_recipe_material' or request.get('quest_objective') is not False
                or request.get('native_quest_effect_supplied') is not False or request.get('node_id')!=node['id']
                or request.get('confirmed') is not True or request.get('ready') is not True
                or request.get('parent_scope')!=scope or any(current[k]!=scope[k] for k in SCOPE+OWNER)
                or any(request.get(k)!=expected.get(k) for k in ('material_state_id','source_recipe','parent_node_id','quantity_unit'))
                or request.get('source_required_quantity')!=expected.get('required_quantity')
                or _item_key(request['item'])!=_item_key(expected['item']) or request['item']['name']!=expected['item']['name']):return False
        canonical=make_material_hold_request(node,snapshot)
        return canonical.get('ready') is True and all(request.get(k)==v for k,v in canonical.items())
    except (KeyError,TypeError,ValueError,AttributeError):return False


def record_material_child_pending(state,pending):
    """Journal intent before child dispatch; never erase a pending implicitly."""
    saved=deepcopy(state)
    if saved.get('phase')!='supplying':raise MaterialSupplyUnavailable('material_child_not_current_phase')
    if pending is None:raise MaterialSupplyUnavailable('material_pending_requires_observed_child_completion_to_clear')
    child=_child(saved);receipt=pending.get('planner_receipt') or {}
    if receipt.get('node_id')!=child['id'] or any(receipt.get(k)!=saved['parent_scope'].get(k) for k in OWNER):
        raise MaterialSupplyUnavailable('material_child_pending_owner_changed')
    prior=saved.get('child_pending')
    if prior and prior.get('operation_id')!=pending.get('operation_id'):
        raise MaterialSupplyUnavailable('material_existing_child_pending_requires_effect_transition')
    saved['child_pending']=deepcopy(pending)
    return saved


def advance_material_supply(state,snapshot,*,field_return_evidence=None,child_result=None):
    """Consume current proofs, emit next child or a navigation-only resume."""
    saved=deepcopy(state)
    try:
        _state_scope(saved,snapshot)
        if saved['phase']=='return_to_field':
            if not _field_proof(saved,snapshot,field_return_evidence,saved['created_at']):
                return dict(status='return_to_field',state=saved,node=deepcopy(saved['parent_node']),
                    reason='material_owned_field_return_required',missing=['current_owned_field_return_evidence'])
            saved.update(phase='supplying',field_return_evidence=deepcopy(field_return_evidence))
        if saved['phase']=='supplying':
            child=_child(saved);goal=saved['goals'][saved['index']]
            if child_result is not None:
                if child_result.get('status')!='complete' or not child_result.get('effect_evidence'):
                    return _wait('material_child_effect_not_confirmed',saved)
                inventory=material_inventory(goal['item'],snapshot,parent_node_id=saved['parent_node']['id'])
                if inventory['uncommitted_quantity']<goal['required_quantity']:
                    return _wait('material_child_completion_without_current_usable_bag',saved)
                pending=saved.get('child_pending')
                if pending:
                    stamp=(pending.get('planner_receipt') or {}).get('observed_at')
                    if not stamp or _stamp(inventory['observed_at'])<=_stamp(stamp):
                        return _wait('material_child_inventory_not_after_owned_input',saved)
                saved['child_completions'].append(dict(node=child,result=deepcopy(child_result),pending=deepcopy(pending),
                    inventory_evidence=inventory,observed_at=inventory['observed_at']))
                saved['child_pending']=None;saved['index']+=1
                if saved['index']>=len(saved['goals']):saved['phase']='return_to_parent'
                else:child=_child(saved)
            if saved['phase']=='supplying':
                return dict(status='child',state=saved,node=child,material_request=_request(saved,snapshot,child),
                    pending=deepcopy(saved['child_pending']),ancestry=deepcopy(saved['ancestry']),missing=[],reason='source_recipe_material_child')
        if saved['phase'] in ('return_to_parent','resume_parent'):
            return reconcile_material_pass(saved,snapshot,field_return_evidence=field_return_evidence)
        raise MaterialSupplyUnavailable('material_phase_unknown')
    except (KeyError,TypeError,ValueError,AttributeError,OverflowError) as exc:return _wait(str(exc),saved)


def _pass_supply_paths(completions):
    """Only saved source selections; no path is inferred from inventory gain."""
    paths=[]
    for row in completions:
        result=row.get('result') or {}
        receipt=result.get('supply_receipt') or ((result.get('preserved_input') or row.get('pending') or {}).get('planner_receipt')) or {}
        child=receipt.get('child_receipt') or {}
        paths.append({'item':_item_key(row['node']['params']['item']),
            'phase':receipt.get('phase'),'recipe_id':child.get('recipe_id'),
            'route_id':child.get('route_id'),
            'selected_source_verified':child.get('recipe_id') is not None or bool(child.get('route_id'))})
    return paths


def reconcile_material_pass(state,snapshot,*,field_return_evidence=None):
    """Recheck all unchanged recipe requirements after owned child UI cleanup.

    A new pass is a new child goal and journal, never a replay of a completed or
    uncertain input. The old pass records, including opaque dispatch receipts,
    remain in pass_history. No elapsed-time or retry-count terminal is used.
    """
    saved=deepcopy(state)
    try:
        _state_scope(saved,snapshot)
        if saved['phase'] not in ('return_to_parent','resume_parent'):
            raise MaterialSupplyUnavailable('material_pass_requires_completed_child_goals')
        if saved.get('child_pending') is not None:
            raise MaterialSupplyUnavailable('material_pass_unresolved_child_pending')
        if saved['index']!=len(saved['goals']) or len(saved['child_completions'])!=len(saved['goals']):
            raise MaterialSupplyUnavailable('material_pass_child_completion_records_incomplete')
        pass_index=saved.get('pass_index',0);history=saved.get('pass_history',[])
        if type(pass_index) is not int or pass_index<0 or not isinstance(history,list) or len(history)!=pass_index:
            raise MaterialSupplyUnavailable('material_pass_history_changed')
        after=saved['child_completions'][-1]['observed_at'] if saved['child_completions'] else saved['created_at']
        if not _field_proof(saved,snapshot,field_return_evidence,after):
            return dict(status='return_to_field',state=saved,node=deepcopy(saved['parent_node']),
                reason='material_current_field_required_before_parent_replan',missing=['new_current_owned_field_return_evidence'])
        proofs=[material_inventory(r['item'],snapshot,parent_node_id=saved['parent_node']['id']) for r in saved['requirements']]
        if any(_stamp(p['observed_at'])<=_stamp(after) for p in proofs):
            raise MaterialSupplyUnavailable('material_pass_inventory_not_after_child_completion')
        lacking=[(r,p) for r,p in zip(saved['requirements'],proofs) if p['uncommitted_quantity']<r['required_quantity']]
        if lacking:
            paths=_pass_supply_paths(saved['child_completions'])
            vector=[dict(item=_item_key(r['item']),required_quantity=r['required_quantity'],
                available_quantity=p['available_quantity'],uncommitted_quantity=p['uncommitted_quantity'],
                other_goal_quantity=p['goal_commitment']['required_quantity']) for r,p in zip(saved['requirements'],proofs)]
            signature=_digest({'source_recipe':saved['source_recipe']['definition_digest'],'stock':vector,'supply_paths':paths})
            if any(row.get('reconciliation_signature')==signature for row in history):
                reason=('material_supply_consumption_cycle_requires_reconciliation' if paths and
                    all(p['selected_source_verified'] for p in paths) else 'material_repeated_stock_state_requires_reconciliation')
                return _wait(reason,saved)
            forbidden={key for row in saved['ancestry'] for key in row.get('output_item_keys',[])}
            if any(_item_key(r['item']) in forbidden for r,p in lacking):
                raise MaterialSupplyUnavailable('material_dependency_item_cycle')
            history.append(dict(pass_index=pass_index,goals=deepcopy(saved['goals']),
                child_completions=deepcopy(saved['child_completions']),child_pending=deepcopy(saved.get('child_pending')),
                return_pending=deepcopy(saved.get('return_pending')),child_graph=deepcopy(saved.get('child_graph')),
                field_return_evidence=deepcopy(field_return_evidence),inventory_evidence=deepcopy(proofs),
                stock_vector=vector,supply_paths=paths,reconciliation_signature=signature))
            saved.update(pass_index=pass_index+1,pass_history=history,pass_started_at=min(p['observed_at'] for p in proofs),
                phase='supplying',goals=[dict(deepcopy(r),initial_inventory=deepcopy(p),
                    initial_shortage=_supply_quantity(r['required_quantity'],p)-p['available_quantity']) for r,p in lacking],
                index=0,child_completions=[],child_pending=None,return_pending=None,
                field_return_evidence=deepcopy(field_return_evidence))
            saved.pop('child_graph',None)
            child=_child(saved)
            return dict(status='child',state=saved,node=child,material_request=_request(saved,snapshot,child),
                pending=None,ancestry=deepcopy(saved['ancestry']),missing=[],reason='source_recipe_material_replenishment_pass')
        saved.update(phase='resume_parent',field_return_evidence=deepcopy(field_return_evidence),material_evidence=proofs)
        resume=dict(parent_node=deepcopy(saved['parent_node']),clear_navigation_pending=True,restore_parent_pending=False,
            preserved_parent_pending=deepcopy(saved['parent_pending']),source_recipe=deepcopy(saved['source_recipe']),
            navigation_resolution_evidence=deepcopy(saved['navigation_resolution_evidence']),
            material_evidence=proofs,field_return_evidence=deepcopy(field_return_evidence),
            parent_goal_complete=False,replan_from_current_field=True)
        return dict(status='resume_parent',state=saved,resume=resume,reason='materials_observed_replan_parent_from_field',missing=[])
    except (KeyError,TypeError,ValueError,AttributeError,OverflowError) as exc:return _wait(str(exc),saved)
