"""Single-writer native debugger operations; no policy or fixed fight plan."""
import hashlib
import json
import math
import re
import secrets
import time

from . import native
from . import experiments as lab
from . import debug_conditions


class NativeRejected(RuntimeError):
    """An acknowledged native rejection, not a missing acknowledgement.

    Only these audited guards precede gameplay input/advance dispatch. Other
    negative replies retain their raw diagnostic without a no-effects claim.
    """
    NO_ACTION_GUARDS = {
        'input': {'player_input_not_ready', 'gameplay_input_obstructed',
                  'requires_devkit_pause', 'input_already_pending',
                  'input_tick_limit', 'unsupported_button',
                  'crossbow_requires_witcher','crossbow_requires_no_save_lock',
                  'crossbow_requires_on_foot','crossbow_not_equipped',
                  'crossbow_not_selected','crossbow_entity_unavailable','crossbow_release_pending'},
        'advance': {'tick_count_out_of_range', 'requires_devkit_pause'},
        'quickslot': {'quickslot_argument_count','stale_world_generation',
                      'quickslot_requires_alive_paused_nosave','quickslot_input_not_ready',
                      'quickslot_release_pending','quickslot_item_action_busy','quickslot_receipt_mismatch',
                      'quickslot_prior_item_changed','quickslot_action_invalid','quickslot_selection_active',
                      'quickslot_identity_exhausted','crossbow_not_equipped','quickslot_prior_not_restorable'},
    }

    def __init__(self, command, args, reply):
        super().__init__('native_rejected: ' + json.dumps(reply))
        self.command = command
        self.no_action_dispatch = (isinstance(reply.get('data'), str)
                                   and reply['data'] in self.NO_ACTION_GUARDS.get(command, set()))
        self.action_before = None
        self.evidence = dict(native_rejection=dict(command=command, args=args, reply=reply,
                                                   audited_no_action_guard=self.no_action_dispatch))


class OperationUncertain(RuntimeError):
    """Explicit compound-operation uncertainty, independent of message wording."""
    outcome_uncertain = True

    def __init__(self, message, evidence=None):
        super().__init__(message)
        self.evidence = evidence or {}


def validate(kind, args):
    allowed = {
        'inspect': {'view', 'reference'}, 'events': {'cursor', 'generation', 'limit'},
        'control': {'mode'}, 'heal': set(),
        'overlay': {'action','reference','title','profile','rows','accent','x','y'},
        'step': {'ticks', 'input'}, 'run': {'ticks'}, 'sample': {'ticks','input','every','actors'}, 'capture': set(),
        'observe-teleport': {'reference', 'ticks', 'input'},
        'request-algorithm': {'actors', 'ticks', 'algorithm', 'mode', 'scores'},
        'request-algorithm-stop': {'generation', 'window_id'},
        'quickslot': {'generation','action','selection_id'},
        'until': {'ticks', 'step', 'condition'}, 'pause': set(),
        'load': {'name', 'sha256'}, 'teleport': {'position'},
        'spawn': {'template', 'level', 'count', 'distance'},
        'cleanup-actors': set(), 'controls': {'payload','timescale'}, 'close': set(), 'handoff': set(),
        'ui-tree': {'root'}, 'ui-find': {'root','field','value','timeout'},
        'ui-dispatch': {'root','field','value','timeout','expect_label'},
        'record-start': set(), 'record-stop': set(), 'skip-intro': set(),
    }
    if kind not in allowed or set(args)-allowed[kind]:
        raise ValueError('invalid_operation: unknown operation or arguments')
    if kind == 'control' and args.get('mode') not in ('status','human','agent'):
        raise ValueError('control mode must be status, human or agent')
    if kind == 'overlay':
        from . import overlay
        overlay.validate(args)
    if kind in ('step', 'run', 'until','sample','observe-teleport','request-algorithm'):
        ticks = args.get('ticks')
        if type(ticks) is not int or not 1 <= ticks <= 600:
            raise ValueError('ticks must be an integer in 1..600')
        if 'input' in args:
            native.validate('input', args['input'])
            if int(args['input'].split('|')[-1]) != ticks:
                raise ValueError('input ticks must equal step ticks; input accepts 2..120')
    if kind == 'sample':
        native.validate('sample_arm',f"0|owned|{args.get('every',6)}|{args['ticks']}")
        if len(args.get('actors',[]))>24: raise ValueError('at most 24 selected actors')
        for ref in args.get('actors',[]): native.validate('actor',ref)
    if kind == 'until':
        if type(args.get('step', 1)) is not int or not 1 <= args.get('step', 1) <= 120:
            raise ValueError('until step must be 1..120')
        validate_condition(args.get('condition'))
    if kind == 'load':
        native.validate('load', args['name'])
        if not re.fullmatch('[0-9a-f]{64}', args['sha256']): raise ValueError('Expected lowercase SHA256')
    if kind == 'inspect' and args.get('view', 'world') not in ('world','actor','combat','profile','ui','controls','capabilities','water','teleport-observer','request-algorithm','crossbow'):
        raise ValueError('unknown inspect view')
    if kind == 'inspect' and args.get('view') in ('teleport-observer','request-algorithm','crossbow') and args.get('reference') is not None:
        raise ValueError('observer status accepts no reference')
    if kind == 'observe-teleport':
        native.validate('teleport_observer_arm', f"{args.get('reference','')}|{args['ticks']}")
    if kind == 'request-algorithm':
        from .request_algorithm import validate as validate_request
        validate_request(args)
    if kind == 'request-algorithm-stop':
        native.validate('request_algorithm_stop', f"{args.get('generation','')}|{args.get('window_id','')}")
    if kind == 'quickslot':
        native.validate('quickslot',f"{args.get('generation','')}|{args.get('action','')}|{args.get('selection_id',0)}")
        if ((args['action']=='restore' and not 1 <= args.get('selection_id',0) <= 2147483646)
                or (args['action']!='restore' and args.get('selection_id',0)!=0)):
            raise ValueError('quickslot restore requires a positive selection-id; selection/clear accept none')
    if kind == 'inspect' and args.get('view') == 'water' and args.get('reference') is not None:
        native.validate('water_depth_actor', args['reference'])
    if kind == 'inspect' and args.get('view') in ('profile','actor') and not args.get('reference'): raise ValueError('reference generation|actor_id is required for actor/profile views')
    if kind == 'inspect' and args.get('view') in ('profile','actor','combat') and args.get('reference') is not None: native.validate('profile', args['reference'])
    if kind == 'controls':
        native.validate('lab_config', args['payload'])
        if 'timescale' in args: native.validate('timescale',str(args['timescale']))
    if kind == 'teleport': native.validate('teleport', args['position'])
    if kind == 'spawn': native.validate('spawn', f"{args['template']}|{args.get('count',1)}|{args['level']}|{args.get('distance',3)}")
    if kind.startswith('ui-'):
        native.validate('ui_scan',args.get('root',''))
        if kind!='ui-tree':
            if args.get('field') not in ('label','text','name','class','path') or not isinstance(args.get('value'),str):
                raise ValueError('UI selector requires an exact field/value')
            if not math.isfinite(args.get('timeout',10)) or not 0<args.get('timeout',10)<=60: raise ValueError('UI timeout must be >0..60')
    if kind == 'events':
        if args.get('cursor',0) and 'generation' not in args: raise ValueError('generation required with nonzero cursor')
        if 'generation' in args and (type(args['generation']) is not int or not 0<=args['generation']<=2147483646): raise ValueError('generation must be a nonnegative native integer')
        if type(args.get('cursor',0)) is not int or args.get('cursor',0)<0: raise ValueError('cursor must be nonnegative integer')
        if not 1 <= args.get('limit',128) <= 128: raise ValueError('limit must be 1..128')


def validate_condition(c):
    return debug_conditions.validate(c)


def water_depth_snapshot(reply, reference=None):
    """Preserve the native receipt; the 10000 sentinel is unknown, never dry."""
    data = reply.get('data')
    depth = data.get('raw_depth') if isinstance(data, dict) else None
    if type(depth) not in (int, float) or not math.isfinite(depth):
        raise RuntimeError('water_depth_invalid_reply: finite raw_depth required')
    if reference is not None:
        generation, actor_id = map(int, reference.split('|'))
        if (type(reply.get('generation')) is not int or reply['generation'] != generation
                or type(data.get('actor_id')) is not int or data['actor_id'] != actor_id):
            raise RuntimeError('water_depth_identity_mismatch: no player fallback')
    available = depth != 10000
    return dict(reply, data=dict(data, available=available,
                                depth_m=depth if available else None,
                                reason=None if available else 'native_10000_sentinel',
                                scope='player_position_only' if reference is None else 'actor_position_only',
                                teleport_suitability='not_evaluated'))


class Runtime:
    def __init__(self, journal, cancelled):
        self.journal, self.cancelled = journal, cancelled
        self.deadline = None
        self.cached = None
        self.cached_controls = None
        self.work_deadline = None
        self.cleaning = False
        self.window_id = None
        self.mutation_dispatched = False
        self.crossbow_mapping = None
        self.rpc_count = self.response_bytes = 0
        self.rpc_parts = 0
        for op in journal.operations.values():
            if op.get('status')=='completed' and op.get('kind')=='record-start': self.window_id=op['result']['window_id']
            elif op.get('status')=='completed' and op.get('kind') in ('record-stop','close'): self.window_id=None

    def call(self, command, args='', deadline=None):
        remaining = min(self.deadline,deadline or self.deadline)-time.monotonic()
        cleanup=self.cleaning or command=='exit'
        if not cleanup and (self.rpc_count>=4096 or self.response_bytes>=16*1024*1024):
            raise RuntimeError('operation_observation_budget: native call or response byte limit reached')
        if remaining <= 0: raise TimeoutError('operation wall deadline reached; inspect operation/state; never replay')
        self.journal.append('native_dispatch',dict(command=command,args=args),cleanup=cleanup)
        self.rpc_count+=1
        if command not in native.READ_ONLY: self.mutation_dispatched=True
        reply = native.call(command, args, timeout=min(5, remaining))
        self.response_bytes+=len(json.dumps(reply).encode())
        self.rpc_parts+=native.state().get('last_response_parts',0)
        self.journal.append('native_reply', reply, cleanup=cleanup)
        if not reply.get('ok'): raise NativeRejected(command, args, reply)
        if command == 'snapshot': self.cached = reply
        if command == 'lab_state': self.cached_controls = reply
        return reply

    def paused(self):
        s = self.call('snapshot')
        if s['data'].get('control') == 'human':
            raise ValueError('agent_control_required: use debug control agent explicitly')
        if not s['data'].get('ready') or not s['data']['world']['devkit_paused']:
            raise ValueError('paused_world_required: load a fixture or pause before this operation')
        if s['data'].get('remaining_ticks',0) or s['data'].get('input_ticks',0):
            raise ValueError('pending_native_work: pause/release before a new operation')
        return s

    def halt(self):
        previous=self.cleaning; self.cleaning=True
        try:
            self.call('release'); self.call('pause')
            s = self.call('snapshot')
            if s['data'].get('input_ticks',0) or s['data'].get('remaining_ticks',0) or (s['data'].get('ready') and not s['data']['world']['devkit_paused']):
                raise RuntimeError('pause_release_unverified')
            return s
        finally: self.cleaning=previous

    def events(self, cursor, generation, limit=128):
        rows, gaps, target = [], [], None
        while len(rows)<limit:
            reply = self.call('event_page', f'{generation}|{cursor}|{min(4,limit-len(rows))}')
            if reply['generation'] != generation: raise RuntimeError('generation_changed')
            p = reply['data']
            if target is None: target=p['latest']
            if cursor>target: raise ValueError('cursor_ahead_of_stream')
            if cursor+1<p['oldest']:
                gaps.append(dict(first=cursor+1,last=p['oldest']-1)); cursor=p['oldest']-1
            selected=[r for r in p['events'] if r['sequence']<=target]
            if not selected: break
            for r in selected:
                if r['sequence']!=cursor+1: gaps.append(dict(first=cursor+1,last=r['sequence']-1))
                cursor=r['sequence']; rows.append(dict(r,engine_time=r.get('time')))
            if cursor>=target: break
        return dict(generation=generation,events=rows,gaps=gaps,next_cursor=cursor,target_sequence=target,complete=not gaps and cursor>=target)

    def safe_events(self, cursor, generation):
        try: return self.events(cursor,generation)
        except (TimeoutError,ConnectionError,RuntimeError,ValueError) as error:
            if 'generation' in str(error): raise
            # A confirmed paused action remains completed if later event reads
            # fail. Keep collection failure separate from input delivery.
            return dict(generation=generation,events=[],gaps=[],errors=[str(error)],
                        next_cursor=cursor,target_sequence=None,complete=False,collection_failed=True)

    def block(self, ticks, payload=None, drain=None):
        before=self.paused()
        if before['data']['world']['player']['alive'] is False:
            return dict(ok=True,before=before,after=before,stop_reason='player_death',input_delivered=False,
                        input_release_verified=True,effect_verified=False,ticks_observed=0,ticks_requested=ticks)
        if self.cancelled.is_set() or time.monotonic()>=self.work_deadline:
            reason='cancelled' if self.cancelled.is_set() else 'wall_budget'
            return dict(ok=True,before=before,after=before,stop_reason=reason,input_delivered=False,ticks_observed=0)
        try:
            ack=self.call('input' if payload else 'advance',payload or str(ticks))
        except NativeRejected as error:
            # This context proves the rejection was the initial action request,
            # not a later observation after an accepted input/advance.
            error.action_before = before
            error.evidence['action_before'] = before
            raise
        next_drain=time.monotonic()+.5
        while True:
            reason=None
            if time.monotonic()>=self.work_deadline:
                after=self.halt(); reason='wall_budget'
            else:
                try: after=self.call('snapshot',deadline=self.work_deadline)
                except TimeoutError:
                    # Only the read is retried. Native mutation uncertainty still
                    # blocks halt; its flag is never cleared here.
                    if time.monotonic()<self.work_deadline: continue
                    after=self.halt(); reason='wall_budget'
            if after['generation']!=before['generation']: raise RuntimeError('generation_changed')
            d=after['data']
            if not d.get('ready'): raise RuntimeError('world_unavailable_during_step')
            if not d['world']['player']['alive']: reason='player_death'
            elif self.cancelled.is_set(): reason='cancelled'
            elif reason is None and not d['remaining_ticks'] and d['world']['devkit_paused']: reason='tick_budget'
            if reason:
                if reason not in ('tick_budget','wall_budget'): after=self.halt()
                observed=after['data']['ticks']-before['data']['ticks']
                if reason=='tick_budget' and (observed!=ticks or after['data'].get('input_ticks',0)):
                    raise RuntimeError('step_completion_mismatch')
                value = dict(ok=True,before=before,after=after,acknowledgement=ack,stop_reason=reason,
                            ticks_observed=observed,ticks_requested=ticks,input_delivered=bool(payload),
                            effect_verified=False,input_release_verified=not after['data'].get('input_ticks',0))
                if payload and payload.split('|')[4]=='Guard':
                    # A paired handler release can be ignored by native action
                    # locks. Never equate dispatch/physical release with cleanup.
                    value['guard_release_observed'] = None
                    value['guard_cleanup_status'] = 'unknown'
                    try:
                        state = self.call('crossbow_state')
                        value['guard_after'] = state
                        if state.get('generation') != after['generation']:
                            raise ValueError('guard_readback_generation_changed')
                        data = state.get('data')
                        if not isinstance(data,dict):
                            raise ValueError('guard_readback_invalid')
                        guard = data.get('guard')
                        flags = [guard.get(key) for key in ('guarded_raw','parry_enabled_raw')] if isinstance(guard,dict) else []
                        if data and data.get('available') and len(flags)==2 and all(type(flag) is bool for flag in flags):
                            released = not any(flags)
                            value['guard_release_observed'] = released
                            value['guard_cleanup_status'] = 'clear' if released else 'retained'
                    except (OSError,RuntimeError,ValueError) as error:
                        value['guard_after_unavailable'] = str(error)
                if payload and payload.split('|')[4]=='Crossbow':
                    value['input_binding'] = self.crossbow_mapping
                    try:
                        state=self.call('crossbow_state')
                        value['crossbow_after']=state
                        if state.get('generation') != after['generation']:
                            raise ValueError('crossbow_readback_generation_changed')
                        data=state.get('data')
                        if not isinstance(data,dict):
                            raise ValueError('crossbow_readback_invalid')
                        keys=[data.get('throw_item_input_value'),data.get('throw_item_hold_input_value')]
                        value['engine_action_release_observed'] = all(key==0 for key in keys) if (
                            data.get('available') and all(type(key) in (int,float) and math.isfinite(key) for key in keys)) else None
                    except (OSError,RuntimeError,ValueError) as error:
                        value['crossbow_after_unavailable']=str(error)
                        value['engine_action_release_observed']=None
                return value
            if drain and time.monotonic()>=next_drain and time.monotonic()+1<self.work_deadline:
                if not drain():
                    after=self.halt()
                    return dict(ok=True,before=before,after=after,acknowledgement=ack,stop_reason='observation_gap',
                                ticks_observed=after['data']['ticks']-before['data']['ticks'],input_delivered=bool(payload),
                                input_release_verified=not after['data'].get('input_ticks',0),effect_verified=False)
                next_drain=time.monotonic()+.5
            if self.rpc_count>=4090 or self.response_bytes>=15*1024*1024:
                after=self.halt()
                return dict(ok=True,before=before,after=after,acknowledgement=ack,stop_reason='observation_budget',
                            ticks_observed=after['data']['ticks']-before['data']['ticks'],input_delivered=bool(payload),effect_verified=False)
            time.sleep(.02)

    def matches(self, c, s, events):
        return debug_conditions.evaluate(c,s,events,self.call)

    def execute(self, kind, args, wall_seconds):
        validate(kind,args)
        if kind in ('observe-teleport','request-algorithm') and not 0 < wall_seconds <= 60:
            raise ValueError('observer wall_seconds must be >0..60, including cleanup')
        self.deadline=time.monotonic()+wall_seconds
        self.work_deadline=self.deadline-min(2,wall_seconds*.25)
        self.rpc_count=self.response_bytes=self.rpc_parts=0
        self.mutation_dispatched=False
        self.crossbow_mapping=None
        if kind == 'overlay':
            from . import overlay
            return overlay.run(self,args)
        if args.get('input') and args['input'].split('|')[4]=='Crossbow':
            # Precheck before sampler/observer arm; Native.call repeats at dispatch.
            from .crossbow_input import verify_bindings
            self.crossbow_mapping=verify_bindings(native.state().get('input_settings_sha256'))
        if kind == 'control':
            mode = args['mode']
            if mode == 'status':
                return self.call('snapshot')
            acknowledgement = self.call('control', mode)
            snapshot = self.call('snapshot')
            data = snapshot['data']
            paused = (data.get('world') or {}).get('devkit_paused', False)
            if (data.get('control') != mode
                    or data.get('physical_controller_passthrough') is not (mode == 'human')
                    or data.get('input_ticks') != 0 or data.get('remaining_ticks') != 0
                    or (mode == 'human' and paused)
                    or (mode == 'agent' and data.get('ready') and not paused)):
                raise OperationUncertain('control_transition_unverified',
                                         dict(acknowledgement=acknowledgement,snapshot=snapshot))
            return dict(acknowledgement=acknowledgement,snapshot=snapshot)
        if kind == 'heal':
            before = self.paused()
            world = before['data']['world']
            player = world['player']
            health = (player.get('hp'), player.get('max_hp'))
            if (player.get('alive') is not True or world.get('saves_locked') is not True
                    or any(type(v) not in (int,float) or not math.isfinite(v) for v in health)
                    or not 0 < health[0] <= health[1]):
                raise ValueError('heal_requires_alive_valid_health_and_no_save_lock')
            acknowledgement = self.call('heal')
            snapshot = None
            try:
                snapshot = self.call('snapshot')
                data = snapshot['data']
                actual = (data.get('world') or {}).get('player') or {}
                receipt = acknowledgement.get('data')
                if (not isinstance(receipt,dict)
                        or any(type(v) not in (int,float) or not math.isfinite(v) for v in (
                            receipt.get('before'),receipt.get('after'),receipt.get('max_before'),
                            receipt.get('max_after'),actual.get('hp'),actual.get('max_hp')))
                        or receipt.get('full_health_verified') is not True
                        or receipt.get('before') != health[0] or receipt.get('max_before') != health[1]
                        or receipt.get('after') != health[1] or receipt.get('max_after') != health[1]
                        or actual.get('hp') != health[1] or actual.get('max_hp') != health[1]
                        or actual.get('alive') is not True
                        or acknowledgement.get('generation') != before['generation']
                        or snapshot.get('generation') != before['generation']
                        or data.get('ticks') != before['data']['ticks']
                        or data.get('input_ticks') != 0 or data.get('remaining_ticks') != 0
                        or data.get('control') != 'agent'
                        or data.get('world',{}).get('devkit_paused') is not True
                        or data.get('world',{}).get('saves_locked') is not True):
                    raise ValueError('health, identity or paused boundary mismatch')
            except Exception as error:
                raise OperationUncertain('heal_postcondition_unverified',
                    dict(before=before,acknowledgement=acknowledgement,
                         snapshot=snapshot,readback_error=str(error))) from error
            return dict(acknowledgement=acknowledgement,snapshot=snapshot,full_health_verified=True)
        if kind=='quickslot':
            self.paused()
            try:
                result=self.call('quickslot',f"{args['generation']}|{args['action']}|{args.get('selection_id',0)}")
            except NativeRejected as error:
                if error.evidence['native_rejection']['reply'].get('data')=='quickslot_postcondition_unverified':
                    raise OperationUncertain('quickslot_postcondition_unverified',error.evidence) from error
                raise
            if args['action']=='restore' and result.get('data',{}).get('prior_selection_restored') is not True:
                raise OperationUncertain('quickslot_restoration_unverified',dict(reply=result))
            return result
        if kind=='observe-teleport':
            from .teleport_observation import observe
            return observe(self,args)
        if kind=='request-algorithm':
            from .request_algorithm import run
            return run(self,args)
        if kind=='request-algorithm-stop':
            result=self.call('request_algorithm_stop', f"{args['generation']}|{args['window_id']}")
            data=result.get('data',{})
            if (data.get('generation') != args['generation'] or data.get('window_id') != args['window_id']
                    or data.get('armed') is not False or data.get('active') is not False):
                raise OperationUncertain('request_stop_unverified', dict(reply=result))
            return result
        if kind=='inspect':
            view=args.get('view','world')
            if view=='capabilities':
                from .debug_capabilities import registry
                return registry()
            if view=='combat':
                s=self.paused(); return self.call('combat_state',args.get('reference') or f"{s['generation']}|owned")
            if view=='water':
                reference=args.get('reference')
                return water_depth_snapshot(self.call('water_depth_actor',reference) if reference is not None
                                            else self.call('water_depth'), reference)
            if view=='teleport-observer':
                return self.call('teleport_observer_status')
            if view=='crossbow':
                self.paused()
                reply=self.call('crossbow_state')
                from .crossbow_input import verify_bindings
                try:
                    binding=dict(available=True,**verify_bindings(native.state().get('input_settings_sha256')))
                except (OSError,ValueError) as error:
                    binding=dict(available=False,reason=str(error))
                return dict(reply,data=dict(reply['data'],host_binding=binding))
            if view=='request-algorithm':
                reply=self.call('request_algorithm_status')
                status=reply['data']
                if status.get('armed') is False and status.get('window_id',0)>0:
                    from .request_algorithm import collect
                    return dict(reply,data=dict(status,collection=collect(
                        self,status['generation'],status['window_id'],status)))
                return reply
            return self.call({'world':'snapshot','actor':'actor','profile':'profile','ui':'ui','controls':'lab_state'}[view],args.get('reference',''))
        if kind.startswith('ui-'):
            from . import ui_driver
            token=ui_driver.OWNER_CALL.set(self.call)
            try:
                if kind=='ui-tree': return ui_driver.tree(args.get('root',''))
                if kind=='ui-find': return ui_driver.find(args['field'],args['value'],args.get('root',''),args.get('timeout',10))
                return ui_driver.dispatch(args['field'],args['value'],args.get('root',''),args.get('timeout',10),args.get('expect_label'))
            finally: ui_driver.OWNER_CALL.reset(token)
        if kind=='sample':
            from . import encounters
            call_token=lab.OWNER_CALL.set(self.call)
            step_token=lab.OWNER_STEP.set(lambda command,payload:self.block(int(payload.split('|')[-1]),payload if command=='input' else None))
            try:
                return encounters.sampled_action('input' if args.get('input') else 'advance',
                    args.get('input') or str(args['ticks']),args.get('every',6),args.get('actors',[]))
            finally:
                lab.OWNER_CALL.reset(call_token); lab.OWNER_STEP.reset(step_token)
        if kind=='capture':
            snapshot=self.call('snapshot')
            path=self.journal.folder/(str(time.time_ns())+'.png')
            native.legacy.run('spectacle','-b','-n','-f','-o',str(path),timeout=max(.1,min(10,self.deadline-time.monotonic())))
            return dict(path=str(path),snapshot=snapshot,capture_scope='desktop',same_render_frame=False)
        if kind=='skip-intro': return self.call('skip_intro')
        if kind=='pause':
            if self.call('snapshot')['data'].get('control') == 'human':
                raise ValueError('agent_control_required: human mode does not pause or step')
            return self.halt()
        if kind=='handoff':
            return dict(snapshot=self.paused(),handoff_ready=True)
        if kind=='close':
            result=native.stop(); result['closed_window_id']=self.window_id; self.window_id=None
            return result
        if kind=='load':
            files=list((native.legacy.manage.DOCS/'gamesaves').iterdir())
            matches=[p for p in files if p.is_file() and (p.name.lower()==args['name'].lower() or (p.suffix.lower()=='.sav' and p.stem.lower()==args['name'].lower()))]
            if len(matches)!=1: raise ValueError('fixture_not_unique: exact case-insensitive save name required')
            path=matches[0]
            if hashlib.sha256(path.read_bytes()).hexdigest()!=args['sha256']: raise ValueError('fixture_hash_mismatch')
            # The actual window and script bridge can appear during the first
            # rendered startup tick, before ListSavedGames populates. Read-only
            # readiness polling must precede the single load transmission.
            save_deadline=min(self.deadline,time.monotonic()+10)
            while True:
                names=self.call('saves')['data']
                names=[n for n in names if n.lower().removesuffix('.sav')==path.stem.lower()]
                if len(names)==1: break
                if len(names)>1 or time.monotonic()>=save_deadline:
                    raise ValueError('fixture_missing_or_ambiguous_in_native_save_list')
                time.sleep(.2)
            before=self.call('snapshot'); self.call('load',names[0])
            while True:
                try:
                    s=self.call('snapshot')
                except (TimeoutError, ConnectionError, OSError):
                    # Loading can block the script receiver for several seconds.
                    # Poll only a fresh read; never repeat the acknowledged load.
                    if time.monotonic()>=self.deadline: raise
                    time.sleep(.1)
                    continue
                if s['data'].get('ready') and s['generation']!=before['generation']:
                    return dict(snapshot=s if s['data'].get('control') == 'human' else self.halt(),
                                fixture=dict(name=path.name,sha256=args['sha256']),
                                stop_reason='cancelled' if self.cancelled.is_set() else 'loaded')
                time.sleep(.1)
        if kind=='events':
            s=self.call('snapshot')
            gen=args.get('generation',s['generation'])
            if gen!=s['generation']: raise ValueError('stale_generation')
            return self.events(args.get('cursor',0),gen,args.get('limit',128))
        if kind=='record-start':
            if self.window_id: raise ValueError('recording_window_already_open')
            self.window_id=secrets.token_hex(12)
            return dict(window_id=self.window_id,recording=True,operation_journal='always_on')
        if kind=='record-stop':
            if not self.window_id: raise ValueError('no_open_recording_window')
            window=self.window_id; self.window_id=None
            return dict(window_id=window,recording=False,operation_journal='always_on',
                        manifest_path=str(self.journal.folder/'manifest.json'))
        if kind in ('teleport','spawn','cleanup-actors','controls'):
            snapshot = self.call('snapshot')
            if snapshot['data'].get('control') == 'human' and kind in ('spawn','cleanup-actors'):
                acknowledgement = self.call('spawn',
                    f"{args['template']}|{args.get('count',1)}|{args['level']}|{args.get('distance',3)}"
                ) if kind == 'spawn' else self.call('cleanup')
                return dict(acknowledgement=acknowledgement,snapshot=self.call('snapshot'),
                            initialization_verified=False,simulation_controlled=False)
            self.paused()
            if kind=='cleanup-actors': return dict(acknowledgement=self.call('cleanup'),snapshot=self.halt())
            if kind=='controls':
                ack=self.call('lab_config',args['payload'])
                scale=self.call('timescale',str(args['timescale'])) if 'timescale' in args else None
                return dict(acknowledgement=ack,timescale_acknowledgement=scale,actual=self.call('lab_state'),snapshot=self.call('snapshot'))
            def step(command,payload):
                result=self.block(int(payload.split('|')[-1]),payload if command=='input' else None)
                if result['stop_reason']!='tick_budget': raise RuntimeError('setup_stopped: '+result['stop_reason'])
                return result
            call_token=lab.OWNER_CALL.set(self.call); step_token=lab.OWNER_STEP.set(step)
            try:
                if kind=='teleport': return lab.teleport(args['position'])
                return lab.spawn(args['template'],args['level'],args.get('count',1),args.get('distance',3))
            finally:
                lab.OWNER_CALL.reset(call_token); lab.OWNER_STEP.reset(step_token)
        before=self.paused(); gen=before['generation']
        page=self.call('event_page',f'{gen}|0|1')['data']; cursor=page['latest']
        if kind in ('step','run'):
            pages=[]
            def drain():
                nonlocal cursor
                page=self.safe_events(cursor,gen); pages.append(page); cursor=page['next_cursor']
                return page['complete']
            value=self.block(args['ticks'],args.get('input'),drain if kind=='run' else None)
            drain()
            value['events']=dict(pages[-1],events=[e for p in pages for e in p['events']],
                                 gaps=[g for p in pages for g in p['gaps']],errors=[e for p in pages for e in p.get('errors',[])],complete=all(p['complete'] for p in pages))
            if not value['events']['complete']: value['stop_reason']='observation_gap'
            return value
        c=args['condition']; total=0; chunks=[]
        def condition(s, events):
            if 'generation' in c and c['generation']!=s['generation']: raise ValueError('generation_changed: condition has stale identity')
            if not s['data']['world']['player']['alive']: return False,'player_death'
            if 'actor_id' in c:
                actor=self.call('actor',f"{c['generation']}|{c['actor_id']}")['data']
                if actor is None: raise ValueError('actor_unavailable: disappearance is not death')
                if actor['alive'] is False: return False,'actor_death'
            match=self.matches(c,s,events)
            return match,'condition' if match else None
        found,terminal=condition(before,[])
        reason=terminal or 'tick_budget'; after=before
        while terminal is None and total<args['ticks']:
            if self.rpc_count>=4080 or self.response_bytes>=15*1024*1024:
                reason='observation_budget'; break
            block=self.block(min(args.get('step',1),args['ticks']-total))
            total+=block['ticks_observed']; after=block['after']
            events=self.safe_events(cursor,gen); cursor=events['next_cursor']
            chunks.append(dict(step=block,events=events))
            if not events['complete']: reason='observation_gap'; break
            if block['stop_reason']=='player_death': reason='player_death'; break
            found,terminal=condition(after,events['events'])
            if terminal: reason=terminal; break
            if block['stop_reason'] in ('cancelled','wall_budget','observation_budget'): reason=block['stop_reason']; break
        return dict(stop_reason=reason,ticks_observed=total,before=before,after=after,
                    condition_match=found,event_tick=found.get('tick') if isinstance(found,dict) else None,detection_tick=after['data']['ticks'],pause_tick=after['data']['ticks'],
                    backend='host_step',step_ticks=args.get('step',1),chunks=chunks)