"""Single-owner continuous graph driver, shared by CLI and the Tk workbench.

The driver advances the real PreparationSession. It never fabricates frames,
restarts a waiting graph, arms input, or turns elapsed time into task success.
"""
from copy import deepcopy
import queue
import threading


def observation_can_progress(report):
    execution = report.get('execution', report)
    if execution.get('status') == 'running':
        return True
    if execution.get('status') != 'waiting':
        return False
    pending = execution.get('active_pending') or execution.get('pending') or {}
    # A dry proposal has no possible game effect. Do not poll for one.
    if (pending.get('input_receipt') or {}).get('status') == 'dry_run':
        return False
    reason = execution.get('reason', '')
    if execution.get('wait_contract') is not None:
        from .wait_contract import can_observe_owned_effect
        return can_observe_owned_effect(execution['wait_contract'],pending)
    # New typed movement/processing boundaries retain an already owned intent.
    # Their exact reason may contain "changed" (the current projection moved),
    # which is distinct from a changed owner/source. Never enable these polls
    # from the reason alone or turn a missing capability into an endless retry.
    receipt = pending.get('planner_receipt') or {}
    if receipt.get('phase')=='quest_item.acquire':
        material=receipt.get('child_receipt') or {}
        inner=material.get('inner_receipt') or {}
        if ((material.get('material_identity') or {}).get('node_id')!=pending.get('node_id')
                or not inner or inner.get('phase')=='quest_item.acquire'
                or any(inner.get(k)!=receipt.get(k) for k in ('node_id','session_id','character_id'))):
            return False
        # Poll the already-owned inner boundary; keep the actual outer ledger.
        receipt=inner
    phase = receipt.get('phase') or ''
    child = receipt.get('child_receipt') or {}
    owned = bool(pending.get('operation_id') and pending.get('node_id') and
        receipt.get('node_id') == pending['node_id'] and receipt.get('session_id') and receipt.get('character_id') and
        pending.get('status') in ('sent_unconfirmed','unknown','intent_persisted') and
        (pending.get('input_receipt') or {}).get('attempted') is not False)
    processing_waits = {
        'processing_navigation_actual_owned_successor_required': {'alter.field','alter.hub_tabs','alter.facility_list','alter.recipe_list','alter.remote_facility','alter.recipe_scroll'},
        'processing_navigation_scroll_effect_required': {'alter.recipe_scroll'},
        'processing_collection_current_successor_surface_required': {'alter.collect'},
    }
    card_waits={'native_map_card_autoplay_sent_accept_effect_unconfirmed_no_resend': 'autoplay_sent',
        'map_arbeit_category_effect_unconfirmed_no_resend':'category_sent',
        'map_quest_card_open_effect_unconfirmed_no_resend':'open_sent'}
    if reason in card_waits:
        return bool(owned and phase=='map_quest_card.'+card_waits[reason]
            and child.get('flow')=='map_quest_card.native_autoplay' and child.get('offer_scope'))
    if reason in processing_waits:
        return owned and phase in processing_waits[reason]
    view_phases = {'open_sent','world_sent','region_sent','pan_sent','zoom_sent'}
    point_phases = {'select_sent','move_sent','pan_sent','zoom_sent'}
    if reason in ('pending_region_point_navigation_unresolved','region_map_transition_fresh_successor_required'):
        stage = 'point' if reason == 'pending_region_point_navigation_unresolved' else 'view'
        return bool(owned and phase.startswith('movement.') and child.get('region_request_key') and
            child.get('region_stage') == stage and child.get('phase') in (point_phases if stage == 'point' else view_phases))
    if reason in ('current_spawned_target_moved_pending_navigation_no_resend',
            'pending_world_target_point_projection_changed_no_resend','pending_world_target_view_fresh_effect_required'):
        stage = 'view' if reason == 'pending_world_target_view_fresh_effect_required' else 'point'
        target_key=child.get('world_target_key') or child.get('group_npc_key')
        target_owner=child.get('world_target_owner') or child.get('group_npc_owner')
        target_stage=child.get('world_target_stage') or child.get('group_npc_stage')
        return bool(owned and phase.startswith('movement.') and target_key and
            target_owner and target_stage == stage and
            child.get('phase') in (point_phases if stage == 'point' else view_phases))
    if any(part in reason for part in ('unavailable', 'unbound', 'unsupported',
            'changed', 'missing', 'disconnected', 'not_authorized', 'mode_mismatch',
            'provider_disabled', 'process_read_disabled', 'source_mismatch')):
        return False
    # These are authored observation boundaries after an owned action, rather
    # than reasons to keep retrying an absent capability or unavailable client.
    return reason in {
        'observation_clock_tick_unresolved',
        'input_requires_effect_observation', 'pending_input_effect_unconfirmed',
        'material_return_input_requires_effect_observation','owned_close_effect_not_observed',
        'current_exact_probe_hit_not_observed', 'new_current_surface_observation_required',
        'post_close_current_HUD_required',
        'pending_input_requires_current_effect', 'native_gather_effect_pending',
        'gather_goal_reached_owned_activity_not_ended',
        'gather_inventory_reached_owned_activity_not_ended',
        'owned_processing_work_in_progress', 'processing_collection_effect_pending',
        'owned_processing_work_receipt_pending',
        'processing_pointer_probe_effect_pending',
        'processing_collection_exact_owned_work_removal_pending',
        'processing_collection_exact_output_delta_pending',
        'movement_in_flight_or_effect_unconfirmed_no_resend',
        'map_closed_or_open_effect_unconfirmed_no_resend',
        'map_pan_or_zoom_effect_unconfirmed_no_repeat',
        'region_transition_unconfirmed_no_resend',
        'region_selection_effect_unconfirmed_no_resend',
        'target_selection_effect_unconfirmed_no_resend',
        'map_transition_or_data_not_ready',
        'selected_move_panel_animation_or_scroll_unsettled',
        'processing_navigation_effect_pending', 'recovery_input_requires_effect_observation',
        'pending_recovery_input_requires_current_effect',
        'quest_native_autoplay_dispatched_wait_for_typed_effect_no_resend',
        'quest_native_autoplay_already_active_wait_for_typed_effect',
        'owned_gather_reconciliation_observing',
        'processing_scroll_effect_pending',
        'craft_submission_awaiting_current_effect_or_result',
        'craft_result_ack_awaiting_typed_empty_UI_and_HUD',
        'craft_result_probe_effect_not_observed',
        'craft_batch_owned_UI_cleanup_required',
        'craft_scroll_effect_not_observed',
        'craft_scroll_probe_effect_not_observed',
        'craft_owned_transition_not_observed',
        'current_quantity_effect_not_observed',
        'quest_map_open_or_effect_unconfirmed_no_resend',
        'quest_marker_selection_effect_unconfirmed_no_resend',
        'quest_map_pan_or_zoom_effect_unconfirmed_no_resend',
        'arrival_observed_quest_effect_pending',
        'destination_arrived_current_interaction_required',
    }


class ContinuousRunner:
    def __init__(self, session, *, interval_seconds=1.0):
        if type(interval_seconds) not in (int, float) or interval_seconds <= 0:
            raise ValueError('Positive observation interval required')
        self.session = session
        self.interval_seconds = interval_seconds
        self.commands, self.results = queue.Queue(), queue.Queue()
        self.closed = threading.Event()
        self._request_lock = threading.RLock()
        self._closing = False
        self._start_reserved = False
        self._requested_generation = None
        prepare=getattr(self.session,'prepare_stop_monitor',None)
        self.stop_monitor_status=prepare() if callable(prepare) else None
        self.worker = threading.Thread(target=self._run, name='arbeit-executor', daemon=True)
        self.worker.start()

    def start(self, quest_id, *, enable_input=False):
        if type(enable_input) is not bool:
            raise ValueError('Explicit input choice must be boolean')
        with self._request_lock:
            generation=getattr(self.session,'stop_generation',None)
            if self._closing or (self._start_reserved and generation==self._requested_generation):
                return False
            self._start_reserved=True
            self._requested_generation=generation
            self.commands.put(('start', (quest_id,enable_input,generation)))
            return True

    def export_logs(self):
        with self._request_lock:
            self.session.signal_stop()
            self.commands.put(('export_logs', None))

    def pause(self):
        with self._request_lock:
            self.session.signal_stop()
            self.commands.put(('pause', None))

    def stop(self):
        with self._request_lock:
            self.session.signal_stop()
            self.commands.put(('stop', None))

    def close(self):
        with self._request_lock:
            if self._closing:return
            self._closing=True
            getattr(self.session,'signal_close',self.session.signal_stop)()
            self.commands.put(('close', None))

    def _publish(self, result, active):
        value = deepcopy(result)
        value['continuous_runner'] = {'observing':active,
            'state':'observing' if active else 'attention_or_terminal',
            'interval_seconds':self.interval_seconds,
            'input_armed_by_driver':False}
        self.results.put(value)

    def _run(self):
        active = False
        active_generation = None
        try:
            while True:
                try:
                    command, argument = self.commands.get(timeout=self.interval_seconds if active else None)
                except queue.Empty:
                    command, argument = 'advance', None
                try:
                    if command == 'close':
                        break
                    if command == 'start':
                        quest_id, enable_input, stop_generation = argument
                        active_generation=stop_generation
                        result = self.session.request_start(quest_id, enable_input=enable_input,
                            expected_stop_generation=stop_generation)
                    elif command == 'pause':
                        result = getattr(self.session,'finish_pause',self.session.pause)()
                    elif command == 'stop':
                        result = {'execution':self.session.stop(), 'input_enabled':False}
                    elif command == 'export_logs':
                        from .support_bundle import export_support_bundle
                        stopped = self.session.stop()
                        result = {'execution':stopped,'input_enabled':False,
                            'support_bundle':export_support_bundle(self.session.storage)}
                    else:
                        result = self.session.advance()
                    active = observation_can_progress(result)
                    with self._request_lock:
                        if not active and active_generation==self._requested_generation:
                            self._start_reserved=False
                    self._publish(result, active)
                except Exception as exc:
                    active = False
                    # A failed provider/driver cannot leave input permission on.
                    result = self.session.pause('driver_error')
                    with self._request_lock:
                        if active_generation==self._requested_generation:
                            self._start_reserved=False
                    result['driver_error'] = type(exc).__name__ + ':' + str(exc)
                    self._publish(result, False)
        finally:
            try:
                self.session.close()
            finally:
                self.closed.set()
