"""Integration boundary between official reads, route planners and actual input.

This revision deliberately does not turn an imported JSON frame into live proof.
Provider capabilities are explicit and cannot be armed by command-line flags.
"""
from dataclasses import asdict
from pathlib import Path
import json
import uuid
import threading
from .arbeit_graph import compile_all, compile_quest, load_catalog
from .graph_executor import GraphExecutor
from .paths import root, user_data


PHASE_BINDINGS = {
    "identify": {"observation": "current character + session + field HUD", "action": None},
    "travel_to_npc": {"observation": "current map pose/marker/panel + NPC arrival", "action": "map planner"},
    "accept": {"observation": "same NPC + exact current job/conditions + owned acceptance UI", "action": "dialogue.accept"},
    "objective": {"observation": "current typed objective program + inventory + owned action state", "action": "source-bound objective planner and material child graph"},
    "return_to_npc": {"observation": "same quest/NPC + current map/tracker + arrival", "action": "map planner"},
    "report": {"observation": "same NPC + exact active job + report choice", "action": "dialogue.report"},
    "deliver": {"observation": "current item/required/selected quantities + delivery form", "action": "delivery.submit"},
    "reward": {"observation": "exact quest clear/reward panel", "action": "reward.confirm"},
    "completion": {"observation": "owned exact quest completion proof + confirmed reward dismissal", "action": None},
}

PROVIDER_GAPS = [
    {"id": "current_character", "phases": ["identify"], "detail": "Official CLI was disconnected during development; current character/session could not be read."},
    {"id": "semantic_screen_provider", "phases": ["identify", "accept", "report", "deliver", "reward", "completion"],
     "detail": "Current dialogue/prompt/keyword, delivery item/quantity, and reward occurrence suppliers are connected. Current branch, enabled target and scoped effects must be observed; source linkage tests are not game execution."},
    {"id": "dynamic_ui_provider", "phases": ["accept", "report", "deliver", "reward"],
     "detail": "Current MMML parent/active child/layout bindings are read by a pinned supplier. Current modal/raycast occlusion and unsupported form families remain gaps; none have been observed on the game."},
    {"id": "map_frame_provider", "phases": ["travel_to_npc", "return_to_npc"],
     "detail": "A version-pinned read adapter is implemented but not exercised on the game. Current map centre/zoom, markers, selected panel and canvas layout remain unobserved."},
    {"id": "gather_lifecycle", "phases": ["objective"],
     "detail": "Preserved gather tapes, current typed life detail, route selection, owned stop and post-pause reconciliation are connected. Current eligibility/activity and exact route ownership remain required; unsupported supply domains stay explicit."},
    {"id": "bag_recovery", "phases": ["objective"],
     "detail": "Current source-region pixels, explicit material policy, preserved cleanup tape and child recovery journal are connected. Recovery needs an explicit policy/input interval and current scoped obstruction evidence; ring-clear observation supports retained 16:9/UI70 only."},
    {"id": "runtime_objective_program", "phases": ["all_objectives"],
     "detail": "Current typed predicates, indexed goal counts and quest-instance autoplay requests are connected. Missing current descriptors, unsupported native action programs and server eligibility remain unknown; display text never selects an action."},
    {"id": "current_offer_giver", "phases": ["accept"],
     "detail": "Source-condition verified current offers supply their exact giver, eligibility and occurrence. A current acceptance branch and its observed accepted effect are still required."},
    {"id":"material_and_crafting_boundary","phases":["objective"],
     "detail":"Recipe material subgraphs preserve parent input and protect requirements across child recovery. Current owned menu close, result acknowledgement, constrained equipment output and non-consumable material supply remain separately tracked boundaries."},
]


def readiness(base=None):
    base = Path(base) if base else root()
    files = ("arbeit-catalog.json", "arbeit-ui-map.json", "arbeit-movement-map.json")
    resources = {name: (base / "data" / name).is_file() for name in files}
    bundle = compile_all()
    from collections import Counter
    counts = Counter(n["op"] for g in bundle["graphs"] for n in g["nodes"])
    from .arbeit_coverage import coverage
    implementation = coverage()
    from . import REVISION
    return {"version": "2.0.0-dev.2", "revision": REVISION, "scope": "all_observed_arbeit_definitions", "denominator": bundle["denominator"],
        "compiled_graphs": len(bundle["graphs"]), "compiled_nodes": sum(counts.values()), "operator_nodes": dict(counts), "live_test_ready": False,
        "implementation_groups": implementation["groups"], "whole_live_routes_ready": 0,
        "implementation_inventory": "data/arbeit-implementation-coverage.json",
        "game_execution_performed": False, "input_default": "dry_run", "resources": resources,
        "user_test_start_available": True, "user_test_start_requires_current_evidence": True,
        "phase_bindings": PHASE_BINDINGS, "blockers": PROVIDER_GAPS,
        "supported": ["official read-only CLI adapter", "Win32 guarded input port (not exercised on game)",
                      "durable phase/effect executor", "pure source-bound UI/map planners", "offline contract verification"],
        "completion_policy": "Missing HUD entries, clicks sent, inventory deltas alone and elapsed time never prove quest completion."}


def normalize_quest_read(envelope):
    """Preserve actual public HUD schema; never synthesize QuestId/ownership."""
    result = {"source": envelope.get("source"), "observed_at": envelope.get("observed_at"),
              "scope": "current_HUD_quests_only", "uncertainty": [], "quests": []}
    if not envelope.get("ok"):
        result["uncertainty"].append(envelope.get("error") or "read_failed")
        return result
    rows = envelope.get("body")
    if isinstance(rows, dict):
        rows = rows.get("quests")
    if not isinstance(rows, list):
        result["uncertainty"].append("unknown_quest_response_schema")
        return result
    for row in rows:
        if not isinstance(row, dict) or not isinstance(row.get("QuestTitle"), str) or not isinstance(row.get("Source"), str) or not isinstance(row.get("Objectives"), list):
            result["uncertainty"].append("invalid_quest_row")
            continue
        objectives = row["Objectives"]
        if any(not isinstance(x, dict) or not isinstance(x.get("Description"), str) or type(x.get("IsCompleted")) is not bool or type(x.get("Count")) is not int or type(x.get("Goal")) is not int for x in objectives):
            result["uncertainty"].append("invalid_objective_row")
            continue
        result["quests"].append(row)
    return result


def make_runtime_pointer_guard(frame_provider):
    """Construct rooted scene readers without opening a process or arming input."""
    from .quest_ui_actions import make_pointer_guard
    from .craft_observer import CurrentCraftSession
    from .material_close import material_close_scene_reader
    from .processing_observer import CurrentProcessingSession, processing_pointer_scene
    from .recipe_scroll import recipe_scroll_scene
    from .world_object_observer import WorldObjectSession, world_object_scene_reader
    from .direct_dialogue_entry import DirectDialogueEntrySession, direct_dialogue_entry_scene
    from .direct_dialogue_input_guard import DirectDialogueInputSession, direct_dialogue_request_scene
    from .story_ui_observer import StoryUISession, story_ui_scene_reader
    from .map_quest_accept_observer import MapQuestCardSession, map_quest_card_scene_reader
    quest=make_pointer_guard(frame_provider)
    material=make_pointer_guard(frame_provider,scene_reader=material_close_scene_reader)
    craft=make_pointer_guard(frame_provider,session_factory=CurrentCraftSession,
        scene_reader=lambda session,binding,frame: session.current_result_pointer_scene(binding,frame))
    processing=make_pointer_guard(frame_provider,session_factory=CurrentProcessingSession,scene_reader=processing_pointer_scene)
    processing_scroll=make_pointer_guard(frame_provider,session_factory=CurrentProcessingSession,scene_reader=recipe_scroll_scene)
    craft_scroll=make_pointer_guard(frame_provider,session_factory=CurrentCraftSession,scene_reader=recipe_scroll_scene)
    world_object=make_pointer_guard(frame_provider,session_factory=WorldObjectSession,scene_reader=world_object_scene_reader)
    direct_dialogue=make_pointer_guard(frame_provider,session_factory=DirectDialogueEntrySession,scene_reader=direct_dialogue_entry_scene)
    direct_request=make_pointer_guard(frame_provider,session_factory=DirectDialogueInputSession,scene_reader=direct_dialogue_request_scene)
    story=make_pointer_guard(frame_provider,session_factory=StoryUISession,scene_reader=story_ui_scene_reader)
    map_card=make_pointer_guard(frame_provider,session_factory=MapQuestCardSession,scene_reader=map_quest_card_scene_reader)
    def read(binding,context,phase):
        target=binding.get('target_id') or ''
        if binding.get('source_guard_kind')=='map_quest_card':
            return map_card(binding,context,phase)
        if binding.get('source_guard_kind')=='story-ui':
            return story(binding,context,phase)
        if binding.get('source_guard_kind')=='direct-dialogue-entry':
            return direct_dialogue(binding,context,phase)
        if binding.get('source_guard_kind')=='direct-dialogue-request':
            return direct_request(binding,context,phase)
        if binding.get('source_guard_kind')=='world-object-interaction':
            return world_object(binding,context,phase)
        if binding.get('source_guard_kind')=='recipe-scroll':
            if binding.get('processing_request'):
                return processing_scroll(binding,context,phase)
            if binding.get('craft_scroll_request'):
                return craft_scroll(binding,context,phase)
            return {'ok':False,'reason':'current_recipe_scroll_family_required'}
        if binding.get('source_guard_kind')=='processing-input':
            return processing(binding,context,phase)
        if target.startswith('material_return.close.'):
            return material(binding,context,phase)
        if target=='craft.result.confirm' and binding.get('ui_type')=='CraftResultUIInstance':
            return craft(binding,context,phase)
        return quest(binding,context,phase)
    return read


def make_runtime_escape_guard(frame_provider):
    """Open a read session only when the separately armed port needs it."""
    def read(binding,context):
        from copy import deepcopy
        from datetime import datetime
        from .quest_ui_actions import QuestUIActionSession
        from .material_escape import read_escape_receiver
        frame=deepcopy(frame_provider())
        window=context['window']
        if (not isinstance(frame,dict) or frame.get('pid')!=window.pid
                or frame.get('process_started_at')!=context.get('observed_process_started_at')
                or frame.get('window_signature')!=binding.get('window_signature')
                or frame.get('frame_id')!=binding.get('frame_id')
                or (frame.get('input_owner') or {}).get('epoch')!=binding.get('input_epoch')):
            raise ValueError('current_escape_observer_owner_or_frame_changed')
        with QuestUIActionSession(window.pid) as session:
            started=datetime.fromisoformat(frame['process_started_at'].replace('Z','+00:00'))
            if session.reader.pid!=window.pid or session.reader.started_at!=started:
                raise ValueError('current_escape_process_occurrence_changed')
            return read_escape_receiver(session,binding,frame)
    return read


def probe(cli_path, evidence_dir):
    from .live_io import CliReadOnlyClient
    client = CliReadOnlyClient(cli_path, evidence_dir=Path(evidence_dir))
    results = {"connection": client.connect()}
    # Explicit one-shot observation, no infinite poll or auto-reconnection.
    for method in ("get_my_info", "get_quests", "get_items", "get_current_environment", "get_activity"):
        results[method] = client.query(method)
        if not results[method].get("ok"):
            results["stopped_reading_after"] = method
            break
    if "get_quests" in results:
        results["normalized_quests"] = normalize_quest_read(results["get_quests"])
    results["readiness"] = readiness()
    return results


def plan_phase(phase, *, goal=None, map_snapshot=None, viewport=None, current_ui=None, receipt=None):
    """Real module integration, blocked results retain exact missing fields."""
    if phase in ("travel_to_npc", "return_to_npc"):
        from .arbeit_movement import plan_npc_movement
        if not goal:
            return {"status": "waiting", "reason": "exact_current_movement_goal_required"}
        return plan_npc_movement(map_snapshot or {}, goal, receipt)
    target = PHASE_BINDINGS.get(phase, {}).get("action")
    if target in ("dialogue.accept", "dialogue.report", "delivery.submit", "reward.confirm"):
        from .arbeit_targets import build_target
        return build_target(target, viewport or {"width": 1920, "height": 1080, "origin": [0, 0]}, current_ui or {})
    return {"status": "waiting", "reason": "observation_or_objective_provider_unbound", "phase": phase}


class PreparationSession:
    """Explicit user test start; each action still needs current source evidence."""
    def __init__(self, storage=None, *, allow_native_observation=True, allow_process_read=False,
                 bag_policy=None, input_interval_seconds=None, foreground_idle_seconds=None, stop_monitor_factory=None, selected_quest=None, game_path=None, cli_path=None):
        self.storage = Path(storage) if storage else user_data()
        self.engine = None
        self.allow_native_observation = allow_native_observation
        self.allow_process_read = allow_process_read
        self._configured_process_read = allow_process_read
        self.input_adapter = None
        self.observer = None
        self.port = None
        self.selected_quest = dict(selected_quest or {})
        self.game_path = game_path or r"C:\Nexon\MabinogiMobile\MabinogiMobile.exe"
        self.cli_path = cli_path or r"C:\Nexon\MabinogiMobile\MabinogiMobile_CLI.exe"
        self.bag_policy = bag_policy
        self.input_interval_seconds = input_interval_seconds
        self.foreground_idle_seconds = foreground_idle_seconds
        self.last_report = None
        self.stop_generation = 0
        self._stop_lock = threading.RLock()
        self._monitor_init_lock = threading.Lock()
        self._closing = False
        self.stop_monitor = None
        self._stop_monitor_factory = stop_monitor_factory

    def prepare_stop_monitor(self):
        """Called before UI start controls exist; never arms game input."""
        if not self.allow_native_observation or self._closing:
            return {'ok':False,'reason':'stop_monitor_not_enabled_or_session_closed'}
        with self._monitor_init_lock:
            try:
                if self.stop_monitor is None:
                    from .live_io import NativeF8StopMonitor
                    factory=self._stop_monitor_factory or NativeF8StopMonitor
                    self.stop_monitor=factory(self.signal_stop)
                if not self.stop_monitor.available:
                    return {'ok':False,'reason':'F8_stop_monitor_unavailable'}
                if self.stop_monitor.f8_pressed():
                    self.signal_stop('F8')
                    return {'ok':False,'reason':'F8_pressed'}
                return {'ok':True,'source':'session_F8_stop_monitor'}
            except (OSError,RuntimeError,ValueError) as exc:
                return {'ok':False,'reason':'F8_stop_monitor_unavailable:'+str(exc)}

    def _authorize_start(self, generation, action):
        # Serialize a lease mutation or one ordinary foreground request with
        # cancellation. No game observation or polling runs under this lock.
        with self._stop_lock:
            if self._closing or generation!=self.stop_generation:
                raise RuntimeError('start_cancelled_by_user_stop')
            if self.stop_monitor is None or not self.stop_monitor.available:
                raise RuntimeError('F8_stop_monitor_unavailable')
            return action()

    def _reset_observers(self):
        # Life boundary state belongs to this explicitly started run. Keep it
        # between ticks, but never across a stop, quest change or lost window.
        life = getattr(self.observer, "life_provider", None)
        if life is not None:
            life.invalidate()
        self.observer = None
        self.port = None

    def _explicit_input_start(self, preflight, generation):
        """One normal foreground request, then the existing monitored input lease."""
        from .live_io import WindowSnapshot
        blockers=[b for b in preflight.get('blockers',[]) if b!='game_not_foreground']
        if blockers or not preflight.get('selected'):
            return dict(preflight,ok=False,blockers=blockers or ['current_selected_window_required'])
        try:
            authorize=lambda action:self._authorize_start(generation,action)
            self.input_adapter.bind(WindowSnapshot(**preflight['selected']))
            focus=self.input_adapter.focus_for_explicit_start(explicit_user_start=True,activation_guard=authorize)
            if not focus.get('ok'):
                self.input_adapter.stop('explicit_start_focus_failed')
                return dict(preflight,ok=False,blockers=[focus.get('reason','foreground_required')],focus=focus)
            checked=self.input_adapter.preflight()
            if not checked.get('ok'):return checked
            self.input_adapter.arm(explicit_user_start=True,activation_guard=authorize)
            return dict(checked,focus=focus,user_test_start=True)
        except (OSError,RuntimeError,ValueError) as exc:
            self.input_adapter.stop('explicit_start_blocked:'+str(exc))
            return dict(preflight,ok=False,blockers=[str(exc)])

    def _input_status(self, report):
        from .continuous_runner import observation_can_progress
        adapter=self.input_adapter
        if adapter is not None and getattr(adapter,'armed',False) is True and not observation_can_progress(report):
            adapter.stop('attention_or_terminal_boundary')
        return bool(adapter is not None and getattr(adapter,'dry_run',True) is False
            and getattr(adapter,'armed',False) is True and not adapter.cancelled.is_set())

    def request_start(self, quest_id=None, *, enable_input=False, expected_stop_generation=None):
        if type(enable_input) is not bool:
            raise ValueError('Explicit input choice must be boolean')
        generation=self.stop_generation if expected_stop_generation is None else expected_stop_generation
        if self._closing or generation!=self.stop_generation:
            return {'execution':{'status':'paused','reason':'queued_start_cancelled_by_user_stop'},'input_enabled':False}
        if enable_input:
            stop_watch=self.prepare_stop_monitor()
            if not stop_watch['ok']:
                return {'execution':{'status':'blocked','reason':stop_watch['reason']},'input_enabled':False}
            if self._closing or generation!=self.stop_generation:
                return {'execution':{'status':'paused','reason':'queued_start_cancelled_by_user_stop'},'input_enabled':False}
        from .source_integrity import check_sources
        import hashlib
        integrity = check_sources()
        if self._closing or generation!=self.stop_generation:
            return {'execution':{'status':'paused','reason':'queued_start_cancelled_by_user_stop'},'input_enabled':False}
        catalog = load_catalog()
        quest_id = quest_id if quest_id is not None else catalog["quests"][0]["id"]
        matches = [q for q in catalog["quests"] if q["id"] == int(quest_id)]
        if len(matches) != 1:
            raise ValueError("Exact catalog quest ID required")
        # Input Start gets a fresh lease. Repeated dry checks keep their current
        # observer until stop/window loss; pending journal intents always stay.
        if self.input_adapter is not None and (enable_input
                or getattr(self.input_adapter,'dry_run',True) is False
                or self.input_adapter.cancelled.is_set()):
            self.input_adapter.stop()
            self._reset_observers()
            self.input_adapter.close()
            self.input_adapter=None
        self.allow_process_read=bool(enable_input or self._configured_process_read)
        if self.engine and self.engine.graph["quest_id"] != int(quest_id):
            self.stop()
            self.engine.close()
            self.engine = None
        if self.engine is None:
            graph = compile_quest(matches[0])
            graph["runtime_source_lock"] = hashlib.sha256((root()/"data/arbeit-source-lock.json").read_bytes()).hexdigest()
            self.storage.mkdir(parents=True, exist_ok=True)
            pointer = self.storage / f"arbeit-q{quest_id}-last.txt"
            resume = pointer.exists()
            name = pointer.read_text(encoding="utf8") if resume else "arbeit-" + uuid.uuid4().hex + ".json"
            if Path(name).name != name or not name.startswith("arbeit-") or not name.endswith(".json"):
                raise ValueError("Invalid Arbeit checkpoint pointer")
            self.engine = GraphExecutor(graph, self.storage / name, mode="live", resume=resume)
            pointer.write_text(name, encoding="utf8")
        preflight = None
        if self.allow_native_observation:
            from .live_io import WindowsInputAdapter, CliReadOnlyClient
            from .game_observer import GameObserver
            from .ui_observer import SourceUiObserver
            from .action_port import SourceActionPort
            from .map_observer import MapObservationProvider
            from .current_context import CurrentContextReader
            from .ui_layout_observer import CurrentUILayoutReader
            from .quest_ui_actions import QuestUIActionSession
            from .quest_state_observer import QuestStateObservationProvider
            from .life_observer import LifeObservationProvider
            from .processing_observer import ProcessingObserver, CurrentProcessingSceneReader
            from .bag_recovery import BagRecoveryProvider
            from .craft_observer import CraftObservationProvider
            from .objective_inventory import ObjectiveInventoryProvider
            from .world_target_observer import WorldTargetObservationProvider
            from .world_object_observer import WorldObjectUIProvider
            if self._closing or generation!=self.stop_generation:
                return {'execution':self.engine.stop('start_cancelled_by_user_stop'),'input_enabled':False}
            if self.input_adapter is None:
                self.input_adapter = WindowsInputAdapter(self.storage / "input-ledger.sqlite", self.game_path, dry_run=not enable_input,
                    interrupt_callback=self.signal_stop)
            preflight = self.input_adapter.preflight()
            if generation!=self.stop_generation:
                preflight=dict(preflight,ok=False,blockers=['start_cancelled_by_user_stop'])
            elif enable_input:
                preflight=self._explicit_input_start(preflight,generation)
            if preflight.get("ok") and self.observer is None:
                # Reader activation belongs to an explicit user test start.
                # Constructing the app or checking static coverage does not read.
                def identity(window):
                    reader = MapObservationProvider(window.pid, session_id=self.engine.data["run_id"])
                    observed = reader.read(allow_process_read=self.allow_process_read, include_map=False)
                    if not observed.get("ok"):
                        return {"ok": False, "reason": observed["reason"], "source": observed["source"]}
                    return {"ok": True, "character_id": observed["snapshot"]["identity"]["character_id"],
                        "current_autoplay": observed['snapshot'].get('current_autoplay'),
                        "process_started_at": observed["snapshot"]["process_started_at"],
                        "source": observed["source"], "observed_at": observed["observed_at"], "source_version": observed["source_version"]}
                def map_state(*, window, node, current_ui, character_id, session_id, layout, process_started_at,region_requests=None):
                    reader = MapObservationProvider(window.pid, session_id=session_id)
                    return reader.read(allow_process_read=self.allow_process_read, include_map=True, layout=layout,region_requests=region_requests,
                        expected_character_id=character_id, expected_process_started_at=process_started_at,
                        viewport=[window.client_rect[2]-window.client_rect[0], window.client_rect[3]-window.client_rect[1]],
                        window_owner={"pid": window.pid, "hwnd": window.hwnd, "observed_at": window.observed_at,
                            "source": "current_win32_client_geometry", "certainty": "confirmed", "process_started_at": process_started_at})
                def quest_state(*, window, graph, node, identity, session_id):
                    reader = QuestStateObservationProvider(window.pid, session_id=session_id)
                    observed = reader.read(allow_process_read=self.allow_process_read,
                        expected_character_id=identity["character_id"], expected_process_started_at=identity.get("process_started_at"),
                        quest_id=graph["quest_id"], expected_accept_time=self.engine.data.get("quest_accept_time"))
                    observed["step_effects"] = [dict(proof, complete=True) for proof in observed.get("state_proofs", [])
                        if proof.get("state") == "step_complete"]
                    return observed
                cli = CliReadOnlyClient(self.cli_path, self.storage / "readonly-evidence")
                from .supply_selection import SupplySelector
                from .group_npc_observer import GroupNPCObservationProvider
                from .direct_dialogue_entry import DirectDialogueEntryProvider
                from .story_ui_observer import StoryUIProvider
                from .map_quest_accept_observer import MapQuestAcceptObservationProvider
                def map_quest_cards(*,window,node,snapshot,frame,image=None):
                    return MapQuestAcceptObservationProvider(window.pid,session_id=snapshot['session_id']).read(
                        node=node,snapshot=snapshot,frame=frame,allow_process_read=self.allow_process_read)
                supplies=SupplySelector()
                life = LifeObservationProvider(allow_process_read=self.allow_process_read,
                    supply_selector=supplies,
                    pending_provider=lambda: self.engine.observation_pending(), activity_reader=lambda: cli.query("get_activity"),
                    cancelled=self.input_adapter.cancelled.is_set)
                processing = ProcessingObserver(cli, scene_reader=CurrentProcessingSceneReader(allow_process_read=self.allow_process_read))
                def processing_state(*, node, snapshot, image=None, frame=None):
                    pending = self.engine.observation_pending() or {}
                    return processing.read(node, snapshot, receipt=pending.get("planner_receipt"), image=image, frame=frame)
                def input_continuity():
                    adapter = self.input_adapter
                    if adapter is None or adapter.dry_run or not adapter.armed:
                        return False
                    try:
                        adapter._guard()
                    except (OSError, RuntimeError, ValueError):
                        return False
                    return True
                bag = BagRecoveryProvider(self.bag_policy,
                    receipt_reader=lambda: (self.engine.observation_recovery() or {}).get("child_pending"),
                    continuity_reader=input_continuity,
                    input_interval_reader=lambda: self.input_interval_seconds)
                self.observer = GameObserver(self.engine.graph, cli,
                    self.input_adapter, selected_quest=self.selected_quest, identity_provider=identity,
                    ui_provider=SourceUiObserver(current_layout_reader=CurrentUILayoutReader(enabled=self.allow_process_read,
                        session_factory=QuestUIActionSession)),
                    context_reader=CurrentContextReader(), map_provider=map_state, quest_provider=quest_state, life_provider=life,
                    processing_provider=processing_state, bag_provider=bag,
                    craft_provider=CraftObservationProvider(allow_process_read=self.allow_process_read),
                    objective_inventory_provider=ObjectiveInventoryProvider(allow_process_read=self.allow_process_read),
                    world_target_provider=WorldTargetObservationProvider(allow_process_read=self.allow_process_read),
                    world_ui_provider=WorldObjectUIProvider(enabled=self.allow_process_read),
                    group_npc_provider=GroupNPCObservationProvider(allow_process_read=self.allow_process_read),
                    direct_dialogue_provider=DirectDialogueEntryProvider(enabled=self.allow_process_read),
                    story_provider=StoryUIProvider(enabled=self.allow_process_read),
                    map_quest_card_provider=map_quest_cards,
                    supply_selector=supplies,
                    recovery_reader=lambda: self.engine.observation_recovery(),
                    pending_reader=lambda: self.engine.observation_pending(),
                    material_context_reader=lambda: self.engine.material_observation_context(),
                    reservation_reader=lambda: self.engine.data.get('goal_reservations',[]),
                    input_owner_provider=self.input_adapter.input_owner,
                    initial_sequence=self.engine.data["last_sequence"])
                from .official_stop import GuardedOfficialStop
                self.port = SourceActionPort(self.input_adapter, lambda: self.observer.last_input_context,
                    official_stop=GuardedOfficialStop(self.input_adapter,cli),
                    pointer_guard=make_runtime_pointer_guard(lambda: self.observer.last_frame),
                    escape_guard=make_runtime_escape_guard(lambda: self.observer.last_frame))
            if preflight.get("ok"):
                if generation!=self.stop_generation:
                    self.input_adapter.stop('start_cancelled_by_user_stop')
                    report=self.engine.stop('start_cancelled_by_user_stop')
                else:
                    self.engine.start(explicit=True)
                    report = self.engine.tick(self.observer, self.port)
            else:
                self._reset_observers()
                report = self.engine.start(explicit=True, blockers=preflight.get("blockers", ["window_unavailable"]))
        else:
            report = self.engine.start(explicit=True, blockers=[b["id"] for b in readiness()["blockers"]])
        self.last_report = {"execution": report, "actual_window_preflight": preflight, "readiness": readiness(),
            "input_enabled": self._input_status(report), "source_integrity": integrity, "journal": str(self.engine.journal.path),
            "user_test_requested":bool(enable_input),"process_read_enabled":self.allow_process_read and self.allow_native_observation}
        return self.last_report

    def advance(self):
        """Continue the same started graph; never create/restart a run here."""
        if self.engine is None:
            return {'execution': {'status':'paused','reason':'explicit_start_required'}, 'input_enabled':False}
        if self.input_adapter is not None and self.input_adapter.cancelled.is_set():
            # Cancellation already invalidated the old start generation. This
            # worker is only recording it; a newer explicit Start must survive.
            return self.finish_pause('input_owner_cancelled')
        if self.engine.data['status'] not in ('running', 'waiting'):
            report = self.engine.report()
        elif self.observer is None or self.port is None:
            report = self.engine.boundary('current_observer_or_port_unavailable', 'blocked')
        else:
            adapter = self.input_adapter
            recover = getattr(adapter, "recover_foreground", None)
            if (callable(recover) and not adapter.dry_run and adapter.armed):
                focus = recover(idle_seconds=getattr(self, "foreground_idle_seconds", None))
                if not focus["ok"]:
                    if adapter.cancelled.is_set():
                        return self.finish_pause("input_owner_cancelled")
                    report = self.engine.report()
                    self.last_report = {"execution": report, "input_enabled": self._input_status(report),
                        "journal": str(self.engine.journal.path), "foreground_recovery": focus}
                    return self.last_report
            report = self.engine.tick(self.observer, self.port)
        self.last_report = {'execution':report, 'input_enabled':self._input_status(report),
            'journal':str(self.engine.journal.path)}
        return self.last_report

    def signal_stop(self, reason='user_stop'):
        """Immediate input cancellation; journal mutation stays on its owner."""
        with self._stop_lock:
            self.stop_generation += 1
            if self.input_adapter is not None:
                self.input_adapter.stop(reason)

    def signal_close(self):
        with self._stop_lock:
            self._closing=True
            self.signal_stop('application_closed')

    def pause(self, reason='user_pause'):
        self.signal_stop()
        return self.finish_pause(reason)

    def finish_pause(self, reason='user_pause'):
        """Worker-side journal update after a synchronous cancellation signal."""
        report = self.engine.stop(reason) if self.engine else {'status':'paused','reason':'no_run_started'}
        self.last_report = {'execution':report, 'input_enabled':False}
        return self.last_report

    def stop(self):
        if self.port:
            self.port.stop()
        if self.input_adapter:
            self.input_adapter.stop()
        self._reset_observers()
        return self.engine.stop() if self.engine else {"status": "paused", "reason": "no_run_started"}

    def close(self):
        self.signal_close()
        self._reset_observers()
        if self.engine:
            self.engine.stop("application_closed")
            self.engine.close()
        if self.input_adapter:
            self.input_adapter.close()
        if self.stop_monitor is not None:
            self.stop_monitor.close()
