"""Read-only official CLI and guarded ordinary Windows input.

No game process memory, private protocol, authentication or elevation API is used.
Win32 boundary adapted from the source snapshots documented in LIVE-IO.md.
SendInput acknowledgement is *not* evidence that a game action succeeded.
"""
from __future__ import annotations

import base64
import ctypes as ct
from contextlib import contextmanager
from ctypes import wintypes as wt
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
import hashlib
import json
import math
import os
from pathlib import Path
import secrets
import sqlite3
import subprocess
import threading
import time
import uuid
from typing import Any


READS = frozenset({"get_items", "get_inventory", "get_activity", "get_currencies",
    "get_gatherable_items", "get_craftable_items", "get_alterable_items",
    "get_altering_works", "get_quests", "get_current_environment", "get_my_info"})
DISCOVERY = frozenset({"status", "capabilities"})
KEYS = {"C": 0x43, "I": 0x49, "J": 0x4A, "K": 0x4B,
        "M": 0x4D, "ESC": 0x1B, "SPACE": 0x20, "T": 0x54, "ENTER": 0x0D}


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def digest(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def canonical(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


class CliReadOnlyClient:
    """Only the official, previously identified read commands are callable.

    connect() saves a fresh capability inventory; query() itself preserves errors
    and accepts discovery independently so an inaccessible connector is diagnosable.
    There is no action(), authentication setup, retry or CLI launch of the game.
    """
    def __init__(self, cli_path: str | Path, evidence_dir: str | Path | None = None,
                 *, runner=subprocess.run, timeout: float = 20):
        self.cli_path = Path(cli_path)
        self.evidence_dir = Path(evidence_dir) if evidence_dir else None
        self.runner, self.timeout = runner, timeout
        self.commands: dict[str, dict] = {}

    def query(self, method: str, params=None) -> dict:
        if method not in READS | DISCOVERY:
            raise ValueError("read_only_method_required: " + str(method))
        if self.cli_path.name.casefold() != "mabinogimobile_cli.exe" or not self.cli_path.is_absolute():
            raise ValueError("official_cli_absolute_path_required")
        result = dict(method=method, params=params, source="official_cli", observed_at=utc_now(),
                      cli_path=str(self.cli_path), ok=False, game_input_sent=False)
        args = [str(self.cli_path), method]
        if params is not None:
            args.append("base64:" + base64.b64encode(canonical(params).encode("utf-8")).decode("ascii"))
        try:
            result["cli_sha256"] = digest(self.cli_path.read_bytes())
            proc = self.runner(args, shell=False, stdin=subprocess.DEVNULL,
                stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=self.timeout,
                creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0)
            raw = proc.stdout.decode("utf-8-sig") if isinstance(proc.stdout, bytes) else proc.stdout
            err = proc.stderr.decode("utf-8", "replace") if isinstance(proc.stderr, bytes) else proc.stderr
            result.update(returncode=proc.returncode, raw_response=raw, stderr=err,
                          response_sha256=digest(raw.encode("utf-8")))
            body = json.loads(raw)
            if not isinstance(body, (dict, list)):
                raise ValueError("unexpected_response_shape")
            if isinstance(body, dict) and "status" in body:
                status = body["status"]
                nested = body.get("body", {})
                body = json.loads(nested) if isinstance(nested, str) else nested
                if status != "accepted":
                    result["error"] = "connector_rejected:" + str(status)
            result["body"] = body
            if proc.returncode != 0:
                result["error"] = "connector_exit:" + str(proc.returncode)
            elif isinstance(body, dict) and (body.get("error") or body.get("pipe") == "disconnected" or body.get("loading")):
                result["error"] = str(body.get("error") or body.get("reason") or "connector_unavailable")
            result["ok"] = "error" not in result
        except subprocess.TimeoutExpired:
            result["error"] = "readonly_query_timeout_game_outcome_unknown"
        except (OSError, UnicodeError, ValueError, TypeError) as exc:
            result["error"] = type(exc).__name__ + ":" + str(exc)
        if self.evidence_dir:
            self.evidence_dir.mkdir(parents=True, exist_ok=True)
            filename = result["observed_at"].replace(":", "-") + "-" + method + ".json"
            (self.evidence_dir / filename).write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
        return result

    def connect(self) -> dict:
        status = self.query("status")
        capabilities = self.query("capabilities")
        body = capabilities.get("body", {})
        rows = body.get("commands", []) if isinstance(body, dict) else []
        valid = isinstance(rows, list) and bool(rows) and all(isinstance(r, dict) and isinstance(r.get("Command"), str) for r in rows)
        self.commands = {r["Command"]: r for r in rows} if valid else {}
        status_body = status.get("body")
        return {"ok": bool(status["ok"] and isinstance(status_body, dict) and status_body.get("pipe") == "connected" and capabilities["ok"] and valid),
                "status": status, "capabilities": capabilities, "commands": self.commands}


@dataclass(frozen=True)
class WindowSnapshot:
    hwnd: int
    pid: int
    exe_path: str
    title: str
    client_rect: tuple[int, int, int, int]
    foreground: bool
    access_blocker: str | None = None
    observed_at: str = ""
    process_started_at: str = ""

    @property
    def identity(self):
        return self.hwnd, self.pid, self.process_started_at, os.path.normcase(self.exe_path), self.title


@dataclass(frozen=True)
class InputCommand:
    kind: str
    target_id: str
    evidence_ref: str
    expected_rect: tuple[int, int, int, int]
    x: float | None = None
    y: float | None = None
    key: str | None = None
    end_x: float | None = None
    end_y: float | None = None
    delta: int | None = None
    duration: float = .25
    safe_rect: tuple[float, float, float, float] | None = None  # physical client XYWH
    hold_seconds: float | None = None
    text: str | None = None
    text_scope: str | None = None
    # Explicit two-stage native target inputs; ordinary source tapes are unchanged.
    source_rect: tuple[float, float, float, float] | None = None  # client LTRB
    input_epoch: str | None = None
    probe_context: dict | None = None
    pointer_verified: dict | None = None
    # Only the new typed material-close path uses a current ESC receiver guard.
    escape_binding: dict | None = None


class InputLedger:
    """Commit intention before dispatch; a crash leaves unknown, never resend."""
    def __init__(self, path: str | Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        with self.connection() as db:
            db.execute("CREATE TABLE IF NOT EXISTS inputs (operation_id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, receipt TEXT NOT NULL)")

    @contextmanager
    def connection(self):
        db = sqlite3.connect(self.path)
        try:
            with db:
                yield db
        finally:
            db.close()

    def get(self, operation_id):
        with self.connection() as db:
            row = db.execute("SELECT fingerprint,receipt FROM inputs WHERE operation_id=?", (operation_id,)).fetchone()
        return (row[0], json.loads(row[1])) if row else None

    def reserve(self, operation_id, fingerprint, receipt):
        with self.connection() as db:
            db.execute("INSERT INTO inputs VALUES (?,?,?)", (operation_id, fingerprint, canonical(receipt)))

    def finish(self, operation_id, receipt):
        with self.connection() as db:
            db.execute("UPDATE inputs SET receipt=? WHERE operation_id=?", (canonical(receipt), operation_id))


class WindowsInputAdapter:
    """No input until bind + explicit arm. Every return preserves uncertainty.

    The caller owns target recognition and route effects. evidence_ref identifies
    the actual UI-target calculation; it is not an invitation to invent coordinates.
    """
    def __init__(self, ledger_path: str | Path, expected_exe_path: str | Path,
                 *, transport=None, dry_run=True, interrupt_callback=None):
        self.expected_exe_path = str(Path(expected_exe_path))
        self.transport = transport or NativeWindowsTransport()
        self.ledger = InputLedger(ledger_path)
        self.dry_run, self.armed = bool(dry_run), False
        self.bound: WindowSnapshot | None = None
        self.cancelled = threading.Event()
        self.stop_reason = "not_armed"
        self._focus_attempted_for_binding = False
        self.lock = threading.RLock()
        # Only cancellation versus a single process-creation handoff uses this
        # fence. Never hold it for observation, SQLite or process completion.
        self.dispatch_fence = threading.RLock()
        self.input_epoch = uuid.uuid4().hex
        self._last_input_operation_id = None
        self.interrupt_callback = interrupt_callback

    def preflight(self) -> dict:
        """Window/OS metadata only; no focus changes, hooks, captures or inputs."""
        try:
            windows = self.transport.enumerate_windows(self.expected_exe_path)
            blockers = []
            pids = {w.pid for w in windows}
            if len(pids) != 1:
                blockers.append("game_identity_missing_or_ambiguous")
            if not windows:
                return dict(ok=False, windows=[], blockers=blockers, dry_run=self.dry_run, input_sent=False)
            best = next((w for w in windows if w.foreground), windows[0])
            if best.access_blocker:
                blockers.append(best.access_blocker)
            if not best.foreground:
                blockers.append("game_not_foreground")
            if best.client_rect[2] <= best.client_rect[0] or best.client_rect[3] <= best.client_rect[1]:
                blockers.append("invalid_client_geometry")
            return dict(ok=not blockers, windows=[asdict(w) for w in windows], selected=asdict(best),
                        blockers=blockers, dry_run=self.dry_run, input_sent=False,
                        hotkey="F8 and external input stop monitoring starts only on explicit arm")
        except (OSError, RuntimeError, ValueError) as exc:
            return dict(ok=False, windows=[], blockers=[str(exc)], dry_run=self.dry_run, input_sent=False)

    def bind(self, window: WindowSnapshot):
        if self.armed:
            raise RuntimeError("cannot_rebind_armed_input")
        if os.path.normcase(window.exe_path) != os.path.normcase(self.expected_exe_path):
            raise ValueError("target_executable_mismatch")
        if not window.process_started_at:
            raise ValueError("process_creation_identity_required")
        self.bound = window
        self.input_epoch = uuid.uuid4().hex
        self._focus_attempted_for_binding = False

    def focus_for_explicit_start(self, *, explicit_user_start=False, activation_guard=None) -> dict:
        """One ordinary OS focus request per fresh binding and explicit Start.

        No preflight or send path calls this. A refused request never retries,
        uses AttachThreadInput, raises privilege, or restores/minimizes windows.
        The caller must bind fresh evidence for another user-started attempt.
        This prepares foreground only; arm remains a separate explicit action.
        """
        with self.lock:
            result = dict(ok=False, focus_requested=False, input_sent=False, observed_at=utc_now())
            try:
                if not explicit_user_start:
                    raise ValueError("explicit_user_start_required")
                if self.dry_run:
                    raise ValueError("dry_run_cannot_focus")
                if self.armed:
                    raise ValueError("cannot_refocus_during_armed_run")
                if self._focus_attempted_for_binding:
                    raise ValueError("focus_already_attempted_rebind_for_new_explicit_start")
                # A fresh explicit Start may clear a previous pause. F8/identity/
                # access are immediately rechecked before any focus request.
                self._begin_activation(activation_guard)
                current = self._guard(require_monitor=False, require_foreground=False)
                if current.foreground:
                    return result | {"ok":True,"reason":"already_foreground","window":asdict(current)}
                def request_focus():
                    if self.cancelled.is_set():
                        raise RuntimeError(self.stop_reason or 'start_cancelled')
                    self._focus_attempted_for_binding = True
                    result['focus_requested'] = True
                    return self.transport.focus_window(current.hwnd)
                if not (activation_guard or (lambda action: action()))(request_focus):
                    raise RuntimeError("os_foreground_request_refused")
                current = self._guard(require_monitor=False)
                return result | {"ok":True,"reason":"foreground_confirmed","window":asdict(current)}
            except (OSError, RuntimeError, ValueError) as exc:
                return result | {"reason":str(exc)}

    def _begin_activation(self, activation_guard):
        def reset():
            # A session ticket cannot erase a stop that arrived while preparing.
            if activation_guard is not None and self.cancelled.is_set():
                raise RuntimeError(self.stop_reason or 'start_cancelled')
            self.cancelled.clear()
            self.stop_reason = ''
        (activation_guard or (lambda action: action()))(reset)

    def _monitor_interrupted(self, reason):
        try:
            if self.interrupt_callback is not None:
                self.interrupt_callback(reason)
        finally:
            self.stop(reason)

    def arm(self, *, explicit_user_start=False, activation_guard=None):
        if not explicit_user_start:
            raise ValueError("explicit_user_start_required")
        if self.dry_run:
            raise ValueError("dry_run_cannot_arm")
        if not self.bound:
            raise ValueError("target_not_bound")
        self._begin_activation(activation_guard)
        self._guard(require_monitor=False)
        self.transport.start_monitor(self._monitor_interrupted)
        if not self.transport.monitor_available:
            self.stop("input_monitor_unavailable")
            raise RuntimeError(self.stop_reason)
        self._guard()
        def activate():
            if self.cancelled.is_set():
                raise RuntimeError(self.stop_reason or 'start_cancelled')
            self.armed = True
            self.input_epoch = uuid.uuid4().hex
        (activation_guard or (lambda action: action()))(activate)

    def input_owner(self, *, character_id, session_id):
        """Current uninterrupted arm lease, never proof of a game action effect."""
        with self.lock:
            result = dict(epoch=self.input_epoch, continuous=False, observed_at=utc_now(),
                source="guarded_input_monitor_lease", character_id=character_id, session_id=session_id)
            if self.bound is not None:
                result['owner'] = dict(pid=self.bound.pid, process_started_at=self.bound.process_started_at,
                    character_id=character_id, session_id=session_id)
            if self.dry_run or not self.armed:
                return result | {"reason": "input_not_armed"}
            try:
                self._guard()
            except (OSError, RuntimeError, ValueError) as exc:
                self.stop("input_continuity_lost:" + str(exc))
                return result | {"epoch": self.input_epoch, "reason": self.stop_reason}
            return result | {"continuous": True}


    def recover_foreground(self, *, idle_seconds=None):
        """Recover an active lease before observation, never re-arm a stopped run."""
        with self.lock:
            result = dict(ok=False, input_sent=False, reason="foreground_wait")
            if self.dry_run or not self.armed or self.cancelled.is_set():
                return result | {"reason": self.stop_reason or "input_not_armed"}
            try:
                try:
                    current = self._guard(require_foreground=False)
                except RuntimeError as exc:
                    if str(exc) != "window_missing_or_minimized":
                        raise
                    current = None  # Its process identity is checked before restore.
                if current is not None and current.foreground:
                    return result | {"ok": True, "reason": "already_foreground"}
                from mini_masukja.windows import observe_input_idle, last_input_stamp
                now = time.monotonic()
                if now < getattr(self, "_foreground_retry_at", 0.):
                    return result
                observed = observe_input_idle(idle_seconds)
                if not observed["ready"]:
                    return result | {"reason": observed["reason"]}
                stamp = last_input_stamp()
                def ready():
                    if self.cancelled.is_set() or not self.armed:
                        return False
                    if self.transport.f8_pressed():
                        self._monitor_interrupted("F8")
                        return False
                    return (self.transport.monitor_available and last_input_stamp() == stamp
                            and stamp is not None and observe_input_idle(idle_seconds)["ready"]
                            and last_input_stamp() == stamp)
                if not ready():
                    return result
                self._foreground_retry_at = now + .25
                if not self.transport.focus_bound_window(self.bound, ready=ready):
                    return result | {"reason": "os_foreground_request_pending"}
                self._guard()
                return result | {"ok": True, "reason": "foreground_confirmed"}
            except (OSError, RuntimeError, ValueError) as exc:
                reason = str(exc)
                # Only lost foreground/minimized geometry is a recoverable
                # focus wait. Preserve input_owner's existing stop boundary for
                # a changed process/geometry/access/monitor, rather than hiding
                # that failure in an active foreground-wait loop.
                if reason not in ("game_not_foreground", "window_missing_or_minimized"):
                    if not self.cancelled.is_set():
                        self.stop("input_continuity_lost:" + reason)
                return result | {"reason": self.stop_reason if self.cancelled.is_set() else reason}

    def _guard(self, *, require_monitor=True, require_foreground=True):
        if self.cancelled.is_set():
            raise RuntimeError(self.stop_reason or "stopped")
        if not self.bound:
            raise RuntimeError("target_not_bound")
        current = self.transport.inspect(self.bound.hwnd)
        if current.identity != self.bound.identity:
            raise RuntimeError("window_identity_changed")
        if tuple(current.client_rect) != tuple(self.bound.client_rect):
            raise RuntimeError("client_geometry_changed")
        if current.access_blocker:
            raise RuntimeError(current.access_blocker)
        if require_foreground and not current.foreground:
            raise RuntimeError("game_not_foreground")
        if self.transport.f8_pressed():
            self._monitor_interrupted("F8")
            raise RuntimeError("F8")
        if require_monitor and not self.dry_run and not self.transport.monitor_available:
            raise RuntimeError("input_monitor_unavailable")
        if self.cancelled.is_set():
            raise RuntimeError(self.stop_reason or 'stopped')
        return current

    def _pointer_guard(self, command, current, reader):
        """A move receipt is not raycast proof. Re-read the exact current scene.

        `reader` is supplied by the source port and reads current target/root/
        callback identity. No scene reader means no move, including in dry-run.
        """
        binding = command.probe_context
        if not isinstance(binding, dict) or not command.input_epoch or command.input_epoch != self.input_epoch:
            raise ValueError('pointer_input_epoch_missing_or_changed')
        if binding.get('input_epoch') != command.input_epoch:
            raise ValueError('pointer_binding_epoch_mismatch')
        for name in ('target_ui_pointer', 'control_pointer', 'root_object_pointer', 'window_id'):
            if type(binding.get(name)) is not int or binding[name] <= 0:
                raise ValueError('pointer_exact_identity_required:' + name)
        if type(binding.get('target_index')) is not int or binding['target_index'] < 0:
            raise ValueError('pointer_exact_target_index_required')
        if type(binding.get('model_event_id')) is not int or binding['model_event_id'] < 0:
            raise ValueError('pointer_exact_callback_required')
        if not all(binding.get(key) for key in ('scene_signature', 'source_hashes', 'frame_id', 'window_signature', 'before_results_hash')):
            raise ValueError('pointer_source_scene_binding_required')
        if binding.get('point') not in ([command.x, command.y], (command.x, command.y)):
            raise ValueError('pointer_binding_point_mismatch')
        for name in ('point_screen', 'before_position'):
            values = binding.get(name)
            if not isinstance(values, (list, tuple)) or len(values) != 2 or any(
                    type(v) not in (int, float) or not math.isfinite(v) for v in values):
                raise ValueError('pointer_current_raycast_position_required:' + name)
        if tuple(binding['point_screen']) == tuple(binding['before_position']):
            raise ValueError('pointer_probe_requires_distinct_current_position')
        rect = command.source_rect
        if not isinstance(rect, (tuple, list)) or len(rect) != 4 or any(
                type(v) not in (int, float) or not math.isfinite(v) for v in rect):
            raise ValueError('pointer_confirmed_source_rect_required')
        x1, y1, x2, y2 = rect
        width, height = current.client_rect[2]-current.client_rect[0], current.client_rect[3]-current.client_rect[1]
        if not (0 <= x1 < x2 <= width and 0 <= y1 < y2 <= height and
                x1 <= round(command.x) < x2 and y1 <= round(command.y) < y2):
            raise ValueError('pointer_point_outside_confirmed_source_rect')
        if reader is None:
            raise ValueError('pointer_current_scene_reader_required')
        observed = reader(command, current)
        if not isinstance(observed, dict) or observed.get('confirmed') is not True:
            raise ValueError('pointer_current_scene_not_confirmed')
        if observed.get('input_epoch') != command.input_epoch or observed.get('scene_signature') != binding['scene_signature']:
            raise ValueError('pointer_current_scene_or_epoch_changed')
        if not observed.get('source_refs'):
            raise ValueError('pointer_current_scene_source_required')
        self._pointer_observed_time(observed.get('observed_at'))
        if command.kind in ('pointer_click','pointer_scroll'):
            intended=binding.get('pointer_action','pointer_click')
            if intended!=command.kind:
                raise ValueError('pointer_action_differs_from_source_probe')
            if command.kind=='pointer_scroll' and (type(command.delta) is not int or command.delta not in (-120,120) or binding.get('scroll_delta')!=command.delta):
                raise ValueError('pointer_scroll_requires_source_bound_single_notch')
            proof = command.pointer_verified
            if not isinstance(proof, dict) or proof.get('confirmed') is not True or not proof.get('source_refs'):
                raise ValueError('pointer_current_raycast_proof_required')
            if proof.get('input_epoch') != command.input_epoch or proof.get('probe_context') != binding:
                raise ValueError('pointer_raycast_binding_mismatch')
            observed_at = self._pointer_observed_time(proof.get('observed_at'))
            saved = self.ledger.get(proof.get('probe_operation_id', ''))
            previous = saved[1] if saved else {}
            if self._last_input_operation_id != proof.get('probe_operation_id'):
                raise ValueError('pointer_probe_not_latest_owned_input')
            if (previous.get('status') != 'sent_unconfirmed' or previous.get('attempted') is not True or
                    previous.get('transport', {}).get('complete') is not True):
                raise ValueError('pointer_real_probe_receipt_required')
            sent = previous.get('command', {})
            if (sent.get('kind') != 'pointer_probe' or sent.get('probe_context') != binding or
                    sent.get('input_epoch') != command.input_epoch or
                    tuple(sent.get('source_rect') or ()) != tuple(rect) or
                    tuple(sent.get('safe_rect') or ()) != tuple(command.safe_rect or ()) or
                    tuple(sent.get('expected_rect') or ()) != tuple(command.expected_rect) or
                    (sent.get('x'), sent.get('y')) != (command.x, command.y)):
                raise ValueError('pointer_probe_receipt_target_mismatch')
            try:
                finished = datetime.fromisoformat(previous['dispatch_finished_at'].replace('Z', '+00:00'))
                if finished.tzinfo is None or observed_at <= finished:
                    raise ValueError('pointer_raycast_must_follow_probe')
            except (KeyError, TypeError):
                raise ValueError('pointer_probe_dispatch_time_required')
            live_proof = observed.get('pointer_verified')
            if (not isinstance(live_proof, dict) or live_proof.get('confirmed') is not True or
                    not live_proof.get('source_refs') or any(live_proof.get(key) != proof.get(key)
                        for key in ('probe_operation_id', 'input_epoch', 'probe_context'))):
                raise ValueError('pointer_current_raycast_guard_required')
            if self._pointer_observed_time(live_proof.get('observed_at')) < observed_at:
                raise ValueError('pointer_current_raycast_guard_older_than_proof')
            position_reader = getattr(self.transport, 'cursor_position', None)
            if not callable(position_reader):
                raise ValueError('pointer_current_position_reader_required')
            cursor = position_reader()
            if tuple(cursor) != (current.client_rect[0]+round(command.x), current.client_rect[1]+round(command.y)):
                raise ValueError('pointer_cursor_no_longer_at_verified_target')

    @staticmethod
    def _pointer_observed_time(value):
        try:
            parsed = datetime.fromisoformat(value.replace('Z', '+00:00'))
        except (AttributeError, TypeError, ValueError):
            raise ValueError('pointer_current_observation_time_required')
        if parsed.tzinfo is None or not -.5 <= (datetime.now(timezone.utc)-parsed).total_seconds() <= 5:
            raise ValueError('pointer_current_observation_stale')
        return parsed

    def _escape_guard(self, command, current, reader):
        """Verify one source-owned close receiver without changing legacy ESC tapes."""
        if command.key!='ESC' or command.target_id!='material_return.close.detail':
            raise ValueError('owned_escape_requires_material_detail_close')
        original=command.escape_binding
        if not isinstance(original,dict):raise ValueError('owned_escape_receiver_binding_required')
        binding=json.loads(canonical(original))
        if (binding.get('target_id')!=command.target_id or binding.get('source_guard_kind')!='material-escape'):
            raise ValueError('owned_escape_source_scope_mismatch')
        if (not command.input_epoch or command.input_epoch!=self.input_epoch or
                binding.get('input_epoch')!=command.input_epoch):
            raise ValueError('owned_escape_input_epoch_changed')
        identities=('target_ui_pointer','handler_pointer','callback_pointer','root_object_pointer','target_game_object')
        if any(type(binding.get(k)) is not int or binding[k]<=0 for k in identities):
            raise ValueError('owned_escape_exact_receiver_identity_required')
        keys=identities+('target_id','source_guard_kind','source_revision','scene_signature','input_epoch','frame_id','window_signature')
        refs=binding.get('source_refs')
        if (not all(binding.get(k) for k in keys) or not isinstance(refs,list) or not refs or
                any(not isinstance(v,str) or not v for v in refs)):
            raise ValueError('owned_escape_source_scene_evidence_required')
        before=self._pointer_observed_time(binding.get('observed_at'))
        if reader is None:raise ValueError('owned_escape_current_receiver_reader_required')
        observed=reader(command,current)
        if (not isinstance(observed,dict) or observed.get('confirmed') is not True or
                any(observed.get(k)!=binding[k] for k in keys) or
                any(type(observed.get(k)) is not int for k in identities)):
            raise ValueError('owned_escape_current_receiver_or_scene_changed')
        actual_refs=observed.get('source_refs')
        if (not isinstance(actual_refs,list) or not actual_refs or any(not isinstance(v,str) or not v for v in actual_refs)
                or not set(refs).issubset(actual_refs)):
            raise ValueError('owned_escape_current_receiver_source_changed')
        if self._pointer_observed_time(observed.get('observed_at'))<before:
            raise ValueError('owned_escape_receiver_proof_predates_binding')
        if canonical(command.escape_binding)!=canonical(binding):
            raise ValueError('owned_escape_binding_mutated_during_read')

    from mini_masukja.game_input_scope import serialized_input as _serialized_input

    @_serialized_input
    def send(self, command: InputCommand, operation_id: str, *, pointer_guard=None, escape_guard=None) -> dict:
        with self.lock:
            receipt = dict(operation_id=operation_id, command=asdict(command), observed_at=utc_now(),
                           status="blocked", attempted=False, effect_confirmed=False, game_completed=False)
            try:
                fingerprint = digest(canonical(asdict(command)).encode("utf-8"))
            except UnicodeEncodeError:
                return receipt | {"reason": "invalid_unicode_input"}
            old = self.ledger.get(operation_id)
            if old:
                if old[0] != fingerprint:
                    return receipt | {"reason": "operation_id_command_conflict"}
                return old[1] | {"duplicate_prevented": True}
            try:
                if not operation_id or not command.target_id or not command.evidence_ref:
                    raise ValueError("operation_and_target_evidence_required")
                if command.kind not in ("click", "key", "scroll", "map_drag", "map_long_press", "bag_text", "craft_text", "pointer_probe", "pointer_click", "pointer_scroll", "owned_escape"):
                    raise ValueError("unsupported_input_kind")
                if command.kind == "bag_text":
                    value = command.text
                    if (command.text_scope not in ("bag.search_name", "bag.quantity") or
                            command.target_id != command.text_scope or type(value) is not str or not value.strip() or
                            any(ord(c) < 32 or ord(c) == 127 or 0xD800 <= ord(c) <= 0xDFFF for c in value)):
                        raise ValueError("owned_bag_text_field_and_value_required")
                    if command.text_scope == "bag.search_name" and not 1 <= len(value.encode("utf-16-le"))//2 <= 50:
                        raise ValueError("bag_search_name_utf16_limit_50")
                    if command.text_scope == "bag.quantity" and (not value.isascii() or not value.isdecimal() or
                            len(value) > 10 or not 1 <= int(value) <= 2147483647):
                        raise ValueError("bag_quantity_positive_int32_required")
                if command.kind == 'craft_text':
                    value = command.text
                    if (command.text_scope != 'craft.quantity' or command.target_id != 'craft.quantity.text'
                            or type(value) is not str or not value.isascii() or not value.isdecimal()
                            or len(value)>10 or not 1<=int(value)<=2147483647):
                        raise ValueError('owned_craft_quantity_positive_int32_required')
                if command.kind == "key" and command.key not in KEYS:
                    raise ValueError("unsupported_interface_key")
                if command.kind == "key" and command.key == "ENTER" and command.target_id not in ("bag.search_focus", 'craft.quantity.complete'):
                    raise ValueError("enter_requires_owned_bag_search_focus")
                current = self._guard(require_monitor=not self.dry_run)
                if tuple(command.expected_rect) != tuple(current.client_rect):
                    raise ValueError("target_geometry_evidence_stale")
                screen = None
                if command.kind in ("click", "scroll", "map_drag", "map_long_press", "pointer_probe", "pointer_click", "pointer_scroll"):
                    left, top, right, bottom = current.client_rect
                    if any(type(v) not in (float, int) or not math.isfinite(v) for v in (command.x, command.y)):
                        raise ValueError("finite_client_pixel_required")
                    if not 0 <= command.x < right-left or not 0 <= command.y < bottom-top:
                        raise ValueError("client_pixel_out_of_bounds")
                    screen = (left + int(round(command.x)), top + int(round(command.y)))
                    if not left <= screen[0] < right or not top <= screen[1] < bottom:
                        raise ValueError("rounded_client_pixel_out_of_bounds")
                    if not self.transport.point_owned_by(screen, current.hwnd):
                        raise ValueError("click_target_occluded")
                    if command.kind in ("scroll", "map_drag", "map_long_press", "pointer_probe", "pointer_click", "pointer_scroll"):
                        if not command.safe_rect or len(command.safe_rect) != 4 or any(
                                type(v) not in (float, int) or not math.isfinite(v) for v in command.safe_rect):
                            raise ValueError("source_bound_safe_rect_required")
                        sx, sy, sw, sh = command.safe_rect
                        if not (0 <= sx < sx+sw <= right-left and 0 <= sy < sy+sh <= bottom-top):
                            raise ValueError("safe_rect_outside_client")
                        points = [(command.x, command.y)]
                        if command.kind == "scroll":
                            if type(command.delta) is not int or not command.delta or command.delta % 120 or abs(command.delta) > 600:
                                raise ValueError("scroll_requires_one_to_five_120_unit_notches")
                        elif command.kind == "map_long_press":
                            if type(command.hold_seconds) not in (int,float) or not math.isfinite(command.hold_seconds) or not .05 <= command.hold_seconds <= 5:
                                raise ValueError("source_hold_threshold_required_between_0.05_and_5_seconds")
                        elif command.kind == 'map_drag':
                            if type(command.duration) not in (int,float) or not math.isfinite(command.duration) or not .08 <= command.duration <= .4:
                                raise ValueError("map_drag_duration_out_of_bounds")
                            if any(type(v) not in (int,float) or not math.isfinite(v) for v in (command.end_x, command.end_y)):
                                raise ValueError("map_drag_endpoint_required")
                            points.append((command.end_x, command.end_y))
                            if math.dist(*points) <= 12:
                                raise ValueError("map_drag_must_not_be_a_click")
                            steps = max(4, math.ceil(command.duration/.025))
                            points = [(command.x+(command.end_x-command.x)*i/steps,
                                       command.y+(command.end_y-command.y)*i/steps) for i in range(steps+1)]
                        for px, py in points:
                            if not (sx <= round(px) < sx+sw and sy <= round(py) < sy+sh):
                                raise ValueError("map_input_leaves_source_safe_rect")
                            if not self.transport.point_owned_by((left+round(px),top+round(py)),current.hwnd):
                                raise ValueError("map_segment_occluded")
                if command.kind in ('pointer_probe', 'pointer_click', 'pointer_scroll'):
                    self._pointer_guard(command, current, pointer_guard)
                    receipt.update(input_epoch=command.input_epoch, target_binding=command.probe_context,
                        probe_operation_id=operation_id if command.kind == 'pointer_probe' else command.pointer_verified['probe_operation_id'])
                if command.kind=='owned_escape':
                    self._escape_guard(command,current,escape_guard)
                    self._guard(require_monitor=not self.dry_run)
                    if command.input_epoch!=self.input_epoch:
                        raise RuntimeError('owned_escape_epoch_changed_during_receiver_read')
                    receipt.update(input_epoch=command.input_epoch,target_binding=json.loads(canonical(command.escape_binding)))
                receipt.update(window=asdict(current), screen_point=screen)
                if self.dry_run:
                    return receipt | {"status": "dry_run", "reason": "input_disabled"}
                if not self.armed:
                    raise RuntimeError("explicit_user_start_required")
                self._guard()
                # Persistence failure prevents all dispatch. Unknown intent also
                # survives a crash between reserve and SendInput acknowledgement.
                receipt.update(status="unknown", attempted=True, reason="dispatch_intent_persisted")
                self.ledger.reserve(operation_id, fingerprint, receipt)
                try:
                    self._guard()
                except (OSError, RuntimeError, ValueError) as exc:
                    receipt.update(status="blocked", attempted=False, reason="stopped_before_dispatch:"+str(exc))
                    self.ledger.finish(operation_id, receipt)
                    return receipt
                def check(point=None):
                    latest = self._guard()
                    if point is not None and not self.transport.point_owned_by(point, latest.hwnd):
                        raise RuntimeError("input_path_became_occluded")
                    if command.kind in ('pointer_probe', 'pointer_click', 'pointer_scroll'):
                        self._pointer_guard(command, latest, pointer_guard)
                        # A current scene read may span a user stop/focus loss.
                        # Recheck after that read, immediately before the prefix.
                        latest = self._guard()
                        if command.input_epoch != self.input_epoch:
                            raise RuntimeError('pointer_input_epoch_changed_during_scene_read')
                        if point is not None and not self.transport.point_owned_by(point, latest.hwnd):
                            raise RuntimeError('input_path_became_occluded')
                    if command.kind=='owned_escape':
                        self._escape_guard(command,latest,escape_guard)
                        self._guard()
                        if command.input_epoch!=self.input_epoch:
                            raise RuntimeError('owned_escape_epoch_changed_during_receiver_read')
                result = self.transport.send(command, screen, check=check, wait=self.cancelled.wait)
                self._last_input_operation_id = operation_id
                receipt.update(transport=result, status="sent_unconfirmed" if result.get("complete") else "unknown",
                               reason="game_effect_requires_observation", dispatch_finished_at=utc_now())
                self.ledger.finish(operation_id, receipt)
                return receipt
            except (OSError, RuntimeError, ValueError, sqlite3.Error) as exc:
                # If reservation exists, retain unknown even if send/finish raised.
                reserved = self.ledger.get(operation_id)
                if reserved:
                    self._last_input_operation_id = operation_id
                    return reserved[1] | {"reason": "dispatch_or_receipt_error:" + str(exc)}
                return receipt | {"status": "blocked", "attempted": False, "reason": str(exc)}

    def stop(self, reason="user_stop"):
        with self.dispatch_fence:
            self.input_epoch = uuid.uuid4().hex
            self.stop_reason = str(reason)
            self.cancelled.set()
            self.armed = False

    def close(self):
        self.stop("closed")
        self.transport.close_monitor()

    def capture_image(self):
        """Return client pixels and provenance in RAM. Never creates a file.

        sha256 binds pixel mode, dimensions and raw bytes, not a PNG encoding.
        Capturing does not arm input, focus the target, or install a monitor.
        """
        current = self._guard(require_monitor=False)
        from PIL import ImageGrab
        started_at = utc_now()
        image = ImageGrab.grab(bbox=current.client_rect, all_screens=True)
        self._guard(require_monitor=False)
        expected = (current.client_rect[2]-current.client_rect[0], current.client_rect[3]-current.client_rect[1])
        if image.size != expected:
            raise RuntimeError("captured_client_geometry_mismatch")
        header = canonical({"mode":image.mode,"size":list(image.size)}).encode("utf-8")
        metadata = dict(sha256=digest(header+b"\0"+image.tobytes()),
                        hash_format="image-mode-size-raw-pixels-v1", observed_at=started_at,
                        capture_finished_at=utc_now(), window=asdict(current),
                        source="visible_client_pixels", storage="ram-only")
        return image, metadata

    def capture(self, destination: str | Path):
        """Explicit diagnostic file save; normal observer uses capture_image()."""
        image, metadata = self.capture_image()
        destination = Path(destination)
        destination.parent.mkdir(parents=True, exist_ok=True)
        image.save(destination)
        return metadata | dict(path=str(destination), pixel_sha256=metadata["sha256"],
                               sha256=digest(destination.read_bytes()), hash_format="file-bytes",
                               storage="explicit-diagnostic-file")


class NativeWindowsTransport:
    """Ordinary user32 input; missing rights remain blockers without escalation."""
    def __init__(self):
        self.monitor = None
        self.tag = secrets.randbelow(0x7FFFFFFF) + 1
        self.user = self.kernel = None
        if os.name != "nt":
            return
        self.user = ct.WinDLL("user32", use_last_error=True)
        self.kernel = ct.WinDLL("kernel32", use_last_error=True)
        self.user.IsWindow.argtypes = [wt.HWND]
        self.user.IsWindow.restype = wt.BOOL
        self.user.GetForegroundWindow.restype = wt.HWND
        self.user.SetForegroundWindow.argtypes = [wt.HWND]
        self.user.SetForegroundWindow.restype = wt.BOOL
        self.user.GetWindowThreadProcessId.argtypes = [wt.HWND, ct.POINTER(wt.DWORD)]
        self.user.IsWindowVisible.argtypes = [wt.HWND]
        self.user.IsIconic.argtypes = [wt.HWND]
        self.user.GetClientRect.argtypes = [wt.HWND, ct.POINTER(wt.RECT)]
        self.user.ClientToScreen.argtypes = [wt.HWND, ct.POINTER(wt.POINT)]
        self.user.GetWindowTextLengthW.argtypes = [wt.HWND]
        self.user.GetWindowTextW.argtypes = [wt.HWND, wt.LPWSTR, ct.c_int]
        self.user.WindowFromPoint.argtypes = [wt.POINT]
        self.user.WindowFromPoint.restype = wt.HWND
        self.user.GetAncestor.argtypes = [wt.HWND, wt.UINT]
        self.user.GetAncestor.restype = wt.HWND
        self.user.GetAsyncKeyState.argtypes = [ct.c_int]
        self.user.GetAsyncKeyState.restype = ct.c_short
        self.kernel.OpenProcess.argtypes = [wt.DWORD, wt.BOOL, wt.DWORD]
        self.kernel.OpenProcess.restype = wt.HANDLE
        self.kernel.CloseHandle.argtypes = [wt.HANDLE]
        self.kernel.QueryFullProcessImageNameW.argtypes = [wt.HANDLE, wt.DWORD, wt.LPWSTR, ct.POINTER(wt.DWORD)]
        self.kernel.GetProcessTimes.argtypes = [wt.HANDLE] + [ct.POINTER(wt.FILETIME)] * 4
        self.user.SetThreadDpiAwarenessContext.argtypes = [ct.c_void_p]
        self.user.SetThreadDpiAwarenessContext.restype = ct.c_void_p

    def _require_windows(self):
        if self.user is None:
            raise RuntimeError("windows_required")

    def process_identity(self, pid):
        process = self.kernel.OpenProcess(0x1000, False, pid)
        if not process:
            raise OSError(ct.get_last_error(), "process_identity_query_denied")
        try:
            buf, size = ct.create_unicode_buffer(32768), wt.DWORD(32768)
            if not self.kernel.QueryFullProcessImageNameW(process, 0, buf, ct.byref(size)):
                raise OSError(ct.get_last_error(), "process_identity_query_failed")
            created, exited, cpu_kernel, cpu_user = (wt.FILETIME() for _ in range(4))
            if not self.kernel.GetProcessTimes(process, ct.byref(created), ct.byref(exited), ct.byref(cpu_kernel), ct.byref(cpu_user)):
                raise OSError(ct.get_last_error(), "process_creation_query_failed")
            token = (created.dwHighDateTime << 32) | created.dwLowDateTime
            if token <= 0:
                raise RuntimeError("process_creation_identity_unknown")
            return buf.value, "windows-filetime:" + str(token)
        finally:
            self.kernel.CloseHandle(process)

    def inspect(self, hwnd):
        self._require_windows()
        old = self.user.SetThreadDpiAwarenessContext(ct.c_void_p(-4))
        try:
            rect, point, pid = wt.RECT(), wt.POINT(), wt.DWORD()
            if not self.user.IsWindowVisible(hwnd) or self.user.IsIconic(hwnd) or not self.user.GetClientRect(hwnd, ct.byref(rect)):
                raise RuntimeError("window_missing_or_minimized")
            if not self.user.ClientToScreen(hwnd, ct.byref(point)):
                raise RuntimeError("client_origin_unavailable")
            self.user.GetWindowThreadProcessId(hwnd, ct.byref(pid))
            title = ct.create_unicode_buffer(self.user.GetWindowTextLengthW(hwnd) + 1)
            self.user.GetWindowTextW(hwnd, title, len(title))
            exe_path, process_started_at = self.process_identity(pid.value)
            return WindowSnapshot(int(hwnd), pid.value, exe_path, title.value,
                (point.x, point.y, point.x + rect.right, point.y + rect.bottom),
                self.user.GetForegroundWindow() == hwnd, self.access_blocker(pid.value), utc_now(), process_started_at)
        finally:
            if old:
                self.user.SetThreadDpiAwarenessContext(old)

    def enumerate_windows(self, expected_exe):
        self._require_windows()
        windows = []
        callback_type = ct.WINFUNCTYPE(wt.BOOL, wt.HWND, wt.LPARAM)
        @callback_type
        def callback(hwnd, _):
            try:
                snap = self.inspect(hwnd)
                if os.path.normcase(snap.exe_path) == os.path.normcase(expected_exe):
                    windows.append(snap)
            except (OSError, RuntimeError):
                pass
            return True
        self.user.EnumWindows(callback, 0)
        return windows

    def access_blocker(self, pid):
        # Snapshot of existing access.py policy: ordinary desktop + token query.
        # OpenProcess uses QUERY_LIMITED_INFORMATION; never reads process memory.
        kernel, user = self.kernel, self.user
        adv = ct.WinDLL("advapi32", use_last_error=True)
        kernel.GetCurrentProcess.restype = wt.HANDLE
        kernel.GetCurrentThreadId.restype = wt.DWORD
        adv.OpenProcessToken.argtypes = [wt.HANDLE, wt.DWORD, ct.POINTER(wt.HANDLE)]
        adv.GetTokenInformation.argtypes = [wt.HANDLE, ct.c_int, ct.c_void_p, wt.DWORD, ct.POINTER(wt.DWORD)]
        adv.GetSidSubAuthorityCount.argtypes = [ct.c_void_p]
        adv.GetSidSubAuthorityCount.restype = ct.POINTER(ct.c_ubyte)
        adv.GetSidSubAuthority.argtypes = [ct.c_void_p, wt.DWORD]
        adv.GetSidSubAuthority.restype = ct.POINTER(wt.DWORD)
        user.GetThreadDesktop.argtypes = [wt.DWORD]
        user.GetThreadDesktop.restype = wt.HANDLE
        user.OpenInputDesktop.argtypes = [wt.DWORD, wt.BOOL, wt.DWORD]
        user.OpenInputDesktop.restype = wt.HANDLE
        user.CloseDesktop.argtypes = [wt.HANDLE]
        user.GetUserObjectInformationW.argtypes = [wt.HANDLE, ct.c_int, ct.c_void_p, wt.DWORD, ct.POINTER(wt.DWORD)]
        def name(handle):
            buf, needed = ct.create_unicode_buffer(512), wt.DWORD()
            return buf.value if handle and user.GetUserObjectInformationW(handle, 2, buf, ct.sizeof(buf), ct.byref(needed)) else None
        desktop = name(user.GetThreadDesktop(kernel.GetCurrentThreadId()))
        handle = user.OpenInputDesktop(0, False, 1)
        try:
            active = name(handle)
        finally:
            if handle:
                user.CloseDesktop(handle)
        if not desktop or not active:
            return "desktop_unknown"
        if active.casefold() != "default":
            return "protected_desktop"
        if desktop != active:
            return "isolated_desktop"
        def integrity(target):
            process = kernel.GetCurrentProcess() if target is None else kernel.OpenProcess(0x1000, False, target)
            token = wt.HANDLE()
            try:
                if not process or not adv.OpenProcessToken(process, 8, ct.byref(token)):
                    return None
                needed = wt.DWORD()
                adv.GetTokenInformation(token, 25, None, 0, ct.byref(needed))
                if not 0 < needed.value <= 65536:
                    return None
                buf = ct.create_string_buffer(needed.value)
                if not adv.GetTokenInformation(token, 25, buf, len(buf), ct.byref(needed)):
                    return None
                sid = ct.cast(buf, ct.POINTER(ct.c_void_p))[0]
                count = adv.GetSidSubAuthorityCount(sid)[0]
                return int(adv.GetSidSubAuthority(sid, count - 1)[0]) if count else None
            finally:
                if token:
                    kernel.CloseHandle(token)
                if target is not None and process:
                    kernel.CloseHandle(process)
        own, target = integrity(None), integrity(pid)
        if own is None or target is None:
            return "integrity_unknown"
        return "lower_integrity" if own < target else None

    def point_owned_by(self, point, hwnd):
        self._require_windows()
        old = self.user.SetThreadDpiAwarenessContext(ct.c_void_p(-4))
        try:
            return self.user.GetAncestor(self.user.WindowFromPoint(wt.POINT(*point)), 2) == hwnd
        finally:
            if old:
                self.user.SetThreadDpiAwarenessContext(old)

    def cursor_position(self):
        """Current physical pointer; no move, capture, or process read."""
        self._require_windows()
        old = self.user.SetThreadDpiAwarenessContext(ct.c_void_p(-4))
        try:
            point = wt.POINT()
            if not self.user.GetCursorPos(ct.byref(point)):
                raise OSError('current_pointer_position_unavailable')
            return point.x, point.y
        finally:
            if old:
                self.user.SetThreadDpiAwarenessContext(old)


    def focus_bound_window(self, bound, *, ready):
        """Focus-only metadata checks also work while client geometry is minimized."""
        self._require_windows()
        from mini_masukja.windows import request_game_foreground
        def owned():
            if not ready():
                return False
            if not self.user.IsWindow(bound.hwnd):
                raise RuntimeError("window_identity_changed")
            if not self.user.IsWindowVisible(bound.hwnd):
                return False
            pid = wt.DWORD()
            if (not self.user.GetWindowThreadProcessId(bound.hwnd, ct.byref(pid))
                    or pid.value != bound.pid):
                raise RuntimeError("window_identity_changed")
            try:
                exe, started = self.process_identity(pid.value)
                blocker = self.access_blocker(pid.value)
            except OSError as exc:
                raise RuntimeError("foreground_owner_query_failed") from exc
            title = ct.create_unicode_buffer(self.user.GetWindowTextLengthW(bound.hwnd)+1)
            self.user.GetWindowTextW(bound.hwnd, title, len(title))
            if (started != bound.process_started_at
                    or os.path.normcase(exe) != os.path.normcase(bound.exe_path)
                    or title.value != bound.title):
                raise RuntimeError("window_identity_changed")
            if blocker:
                raise RuntimeError(blocker)
            return ready()
        return request_game_foreground(bound.hwnd, expected_pid=bound.pid, ready=owned)

    def focus_window(self, hwnd):
        """Single OS request; no fallback, activation loop or permission change."""
        self._require_windows()
        return bool(self.user.SetForegroundWindow(hwnd))

    def f8_pressed(self):
        return bool(self.user.GetAsyncKeyState(0x77) & 0x8000)

    @property
    def monitor_available(self):
        return bool(self.monitor and self.monitor.available and self.monitor.thread.is_alive())

    def start_monitor(self, stop):
        self.close_monitor()
        self.monitor = _InputMonitor(self.tag, stop)

    def close_monitor(self):
        if self.monitor:
            self.monitor.close()
            self.monitor = None

    def _send_events(self, specs):
        """ABI seam: offline tests override this, so no OS input is involved."""
        self._require_windows()
        class Mouse(ct.Structure):
            _fields_ = [("dx", wt.LONG), ("dy", wt.LONG), ("mouseData", wt.DWORD),
                        ("dwFlags", wt.DWORD), ("time", wt.DWORD), ("dwExtraInfo", ct.c_size_t)]
        class Key(ct.Structure):
            _fields_ = [("wVk", wt.WORD), ("wScan", wt.WORD), ("dwFlags", wt.DWORD),
                        ("time", wt.DWORD), ("dwExtraInfo", ct.c_size_t)]
        class Union(ct.Union):
            _fields_ = [("mi", Mouse), ("ki", Key)]
        class Input(ct.Structure):
            _fields_ = [("type", wt.DWORD), ("u", Union)]
        events = (Input * len(specs))()
        for event, spec in zip(events, specs):
            if spec["kind"] == "key":
                event.type, event.u.ki.wVk, event.u.ki.dwExtraInfo = 1, spec.get("key", 0), self.tag
                event.u.ki.wScan = spec.get("scan", 0)
                event.u.ki.dwFlags = spec["flags"]
            else:
                event.u.mi.dwFlags, event.u.mi.dwExtraInfo = spec["flags"], self.tag
                event.u.mi.dx, event.u.mi.dy = spec.get("x", 0), spec.get("y", 0)
                event.u.mi.mouseData = spec.get("delta", 0) & 0xFFFFFFFF
        self.user.SendInput.argtypes = [wt.UINT, ct.POINTER(Input), ct.c_int]
        self.user.SendInput.restype = wt.UINT
        return int(self.user.SendInput(len(events), events, ct.sizeof(Input)))

    def _move_spec(self, point):
        # inspect()/cursor_position() use physical pixels. The virtual desktop
        # must be read in the same DPI context, including when hosted by an
        # unaware UI thread; applying a DPI factor to the point would be twice.
        old = self.user.SetThreadDpiAwarenessContext(ct.c_void_p(-4))
        if not old:
            raise OSError(ct.get_last_error(), "physical_virtual_screen_context_unavailable")
        try:
            left, top = self.user.GetSystemMetrics(76), self.user.GetSystemMetrics(77)
            width, height = self.user.GetSystemMetrics(78), self.user.GetSystemMetrics(79)
        finally:
            self.user.SetThreadDpiAwarenessContext(old)
        if not (1 < width <= 32768 and 1 < height <= 32768 and left <= point[0] < left+width and top <= point[1] < top+height):
            raise ValueError("virtual_screen_geometry_invalid")
        return {"kind": "mouse", "flags": 0xC001,
            "x": min(65535,max(0,round((point[0]-left+.5)*65536/width))),
            "y": min(65535,max(0,round((point[1]-top+.5)*65536/height)))}

    def send(self, command, screen, *, check=lambda point=None: None, wait=time.sleep):
        """Dispatch bounded source-derived segments; release after every prefix.

        Caller records intent before entry. Even cursor movement is a dispatched
        prefix, so interrupted scroll/drag never becomes safe to replay.
        """
        check(screen)
        if command.kind in ("bag_text", "craft_text"):
            return self._replace_bag_text(command.text, check=check, wait=wait)
        if command.kind == 'pointer_probe':
            # One move only. The caller must later observe the game's own hit
            # target before proposing pointer_click; no timer promotes this.
            sent = self._send_events([self._move_spec(screen)])
            return dict(expected=1, sent=sent, complete=sent == 1)
        if command.kind == 'pointer_scroll':
            # The actual scroll receiver was freshly verified at this position.
            # No cursor move, button DOWN/UP, or repeated wheel event is added.
            sent=self._send_events([{'kind':'mouse','flags':0x0800,'delta':command.delta}])
            return dict(expected=1,sent=sent,complete=sent==1)
        if command.kind == 'pointer_click':
            # The probe already positioned the pointer. Moving here would make
            # the supplied game raycast stale before DOWN.
            release = {'kind': 'mouse', 'flags': 4}
            specs = [{'kind': 'mouse', 'flags': 2}, release]
            sent = None
            try:
                sent = self._send_events(specs)
            finally:
                if sent != len(specs):
                    self._send_events([release])
            return dict(expected=2, sent=sent, complete=sent == 2)
        if command.kind in ("click", "key", "owned_escape"):
            if command.kind == "click":
                release = {"kind":"mouse", "flags":4}
                specs = [self._move_spec(screen), {"kind":"mouse","flags":2}, release]
            else:
                release = {"kind":"key","key":KEYS[command.key],"flags":2}
                specs = [{"kind":"key","key":KEYS[command.key],"flags":0}, release]
            sent = None
            try:
                sent = self._send_events(specs)
            finally:
                # Exception or partial ABI result must not strand a pressed key.
                if sent != len(specs):
                    self._send_events([release])
            return dict(expected=len(specs), sent=sent, complete=sent==len(specs))
        if command.kind == "scroll":
            sent = self._send_events([self._move_spec(screen)])
            if sent != 1:
                return dict(expected=1+abs(command.delta)//120,sent=sent,complete=False)
            notches = abs(command.delta)//120
            for index in range(notches):
                if index:
                    wait(.08)
                check(screen)
                count = self._send_events([{"kind":"mouse","flags":0x0800,"delta":120 if command.delta>0 else -120}])
                sent += count
                if count != 1:
                    return dict(expected=1+notches,sent=sent,complete=False)
            return dict(expected=1+notches,sent=sent,complete=True)
        if command.kind == "map_long_press":
            # The source's threshold is mandatory; no guessed hold duration.
            # Split that exact supplied wait only to make cancellation responsive.
            steps = max(1, math.ceil(command.hold_seconds/.025))
            release_attempted = False
            summary = dict(expected=3,sent=0,complete=False,release_attempted=False,release_sent=0,
                           source_hold_seconds=command.hold_seconds,scheduled_wait_seconds=0.)
            try:
                count = self._send_events([self._move_spec(screen)])
                summary["sent"] += count
                if count != 1:
                    return summary
                check(screen)
                release_attempted = True
                count = self._send_events([{"kind":"mouse","flags":2}])
                summary["sent"] += count
                if count != 1:
                    return summary
                for _ in range(steps):
                    check(screen)
                    interval = command.hold_seconds/steps
                    wait(interval)
                    summary["scheduled_wait_seconds"] += interval
                check(screen)
                summary["complete"] = True
            except (OSError, RuntimeError, ValueError) as exc:
                summary["interrupted"] = type(exc).__name__ + ":" + str(exc)
            finally:
                if release_attempted:
                    summary["release_attempted"] = True
                    try:
                        count = self._send_events([{"kind":"mouse","flags":4}])
                        summary["sent"] += count
                        summary["release_sent"] = count
                        if count != 1:
                            summary["complete"] = False
                    except OSError as exc:
                        summary.update(complete=False,release_error=str(exc))
            return summary
        if command.kind != "map_drag":
            raise ValueError("unsupported_input_kind")
        steps = max(4, math.ceil(command.duration/.025))
        end = (screen[0]+round(command.end_x)-round(command.x),
               screen[1]+round(command.end_y)-round(command.y))
        points = [(round(screen[0]+(end[0]-screen[0])*i/steps),
                   round(screen[1]+(end[1]-screen[1])*i/steps)) for i in range(steps+1)]
        # Validate the prepared entire segment before DOWN; validate ownership
        # again for each actual step, with no focus recovery or new coordinates.
        moves = [self._move_spec(point) for point in points]
        for point in points:
            check(point)
        release_attempted = False
        summary = dict(expected=steps+3,sent=0,complete=False,release_attempted=False,release_sent=0)
        try:
            count = self._send_events([moves[0]])
            summary["sent"] += count
            if count != 1:
                return summary
            check(points[0])
            release_attempted = True  # A DOWN exception is still uncertain.
            count = self._send_events([{"kind":"mouse","flags":2}])
            summary["sent"] += count
            if count != 1:
                return summary
            for point, move in zip(points[1:], moves[1:]):
                check(point)
                count = self._send_events([move])
                summary["sent"] += count
                if count != 1:
                    return summary
                wait(command.duration/steps)
            check(points[-1])
            summary["complete"] = True
        except (OSError, RuntimeError, ValueError) as exc:
            summary["interrupted"] = type(exc).__name__ + ":" + str(exc)
        finally:
            if release_attempted:
                # No guard on release: F8/foreground loss cannot strand DOWN.
                summary["release_attempted"] = True
                try:
                    release_sent = self._send_events([{"kind":"mouse","flags":4}])
                    summary["sent"] += release_sent
                    summary["release_sent"] = release_sent
                    if release_sent != 1:
                        summary["complete"] = False
                except OSError as exc:
                    summary.update(complete=False,release_error=str(exc))
        return summary

    @staticmethod
    def _text_hold():
        # mini26 timing.INPUT_DELAYS['hold']; no clipboard and no focus recovery.
        import random
        return random.SystemRandom().triangular(.055, .13, .085)

    def _replace_bag_text(self, value, *, check, wait):
        """Retained Ctrl+A then UTF-16 SendInput in an already owned bag field.

        Every partial prefix stays unknown. A stopped run releases held keys
        only; it never refocuses, reselects or resumes remaining characters.
        """
        encoded = value.encode("utf-16-le")
        units = [int.from_bytes(encoded[i:i+2], "little") for i in range(0,len(encoded),2)]
        result = dict(expected=4+2*len(units), sent=0, complete=False, release_attempts=0)
        def emit(spec):
            count = self._send_events([spec]); result["sent"] += count
            if count != 1:
                raise RuntimeError("partial_bag_text_input")
        def release(spec):
            result["release_attempts"] += 1
            try:
                emit(spec)
            except (OSError, RuntimeError, ValueError) as exc:
                result["release_error"] = str(exc)
        try:
            check()
            try:
                emit({"kind":"key","key":0x11,"flags":0})
                check()
                emit({"kind":"key","key":0x41,"flags":0})
                wait(self._text_hold())
            finally:
                release({"kind":"key","key":0x41,"flags":2})
                release({"kind":"key","key":0x11,"flags":2})
            if result.get("release_error"):
                return result
            for unit in units:
                check()
                try:
                    emit({"kind":"key","scan":unit,"flags":4})
                    wait(self._text_hold())
                finally:
                    release({"kind":"key","scan":unit,"flags":6})
                if result.get("release_error"):
                    return result
            check()
            result["complete"] = True
        except (OSError, RuntimeError, ValueError) as exc:
            result["interrupted"] = type(exc).__name__ + ":" + str(exc)
        return result


class _InputMonitor:
    """Non-suppressing local hooks. F8/user/foreign automation stop future input.

    The F8-only session listener precedes start controls; the broader input
    lease listener is installed on explicit arm. Neither suppresses events.
    """
    def __init__(self, tag, stop, *, f8_only=False):
        self.tag, self.stop = tag, stop
        self.f8_only = f8_only
        self.available = False
        self.ready, self.closed = threading.Event(), threading.Event()
        self.thread = threading.Thread(target=self._run, daemon=True)
        self.thread.start()
        self.ready.wait(2)

    def _run(self):
        user, kernel = ct.WinDLL("user32", use_last_error=True), ct.WinDLL("kernel32", use_last_error=True)
        class Keyboard(ct.Structure):
            _fields_ = [("vkCode", wt.DWORD), ("scanCode", wt.DWORD), ("flags", wt.DWORD),
                        ("time", wt.DWORD), ("dwExtraInfo", ct.c_size_t)]
        class Mouse(ct.Structure):
            _fields_ = [("pt", wt.POINT), ("mouseData", wt.DWORD), ("flags", wt.DWORD),
                        ("time", wt.DWORD), ("dwExtraInfo", ct.c_size_t)]
        callback_type = ct.WINFUNCTYPE(ct.c_ssize_t, ct.c_int, wt.WPARAM, wt.LPARAM)
        user.SetWindowsHookExW.argtypes = [ct.c_int, callback_type, wt.HINSTANCE, wt.DWORD]
        user.SetWindowsHookExW.restype = wt.HANDLE
        user.CallNextHookEx.argtypes = [wt.HANDLE, ct.c_int, wt.WPARAM, wt.LPARAM]
        user.CallNextHookEx.restype = ct.c_ssize_t
        user.UnhookWindowsHookEx.argtypes = [wt.HANDLE]
        kernel.GetModuleHandleW.argtypes = [wt.LPCWSTR]
        kernel.GetModuleHandleW.restype = wt.HMODULE
        callbacks, hooks = [], []
        def build(structure, keyboard):
            @callback_type
            def callback(code, message, ptr):
                if code >= 0:
                    value = ct.cast(ptr, ct.POINTER(structure)).contents
                    own = value.dwExtraInfo == self.tag and bool(value.flags & (0x10 if keyboard else 1))
                    if not own and (not self.f8_only or keyboard and value.vkCode == 0x77):
                        self.stop("F8" if keyboard and value.vkCode == 0x77 else "external_input")
                return user.CallNextHookEx(None, code, message, ptr)
            return callback
        try:
            kinds=((13, Keyboard, True),) if self.f8_only else ((13, Keyboard, True), (14, Mouse, False))
            for kind, structure, keyboard in kinds:
                callback = build(structure, keyboard)
                callbacks.append(callback)
                hook = user.SetWindowsHookExW(kind, callback, kernel.GetModuleHandleW(None), 0)
                if not hook:
                    self.stop("input_monitor_install_failed:" + str(ct.get_last_error()))
                    return
                hooks.append(hook)
            self.available = True
            self.ready.set()
            msg = wt.MSG()
            while not self.closed.wait(.01):
                while user.PeekMessageW(ct.byref(msg), None, 0, 0, 1):
                    user.TranslateMessage(ct.byref(msg))
                    user.DispatchMessageW(ct.byref(msg))
        finally:
            self.available = False
            self.ready.set()
            for hook in hooks:
                user.UnhookWindowsHookEx(hook)

    def close(self):
        self.closed.set()
        if threading.current_thread() is not self.thread:
            self.thread.join(timeout=2)


class NativeF8StopMonitor:
    """Session-lifetime, non-suppressing F8 observation; no game access/input."""
    def __init__(self, callback):
        if os.name != 'nt':
            raise RuntimeError('F8_stop_monitor_requires_Windows')
        self.transport = NativeWindowsTransport()
        self.monitor = _InputMonitor(self.transport.tag, callback, f8_only=True)

    @property
    def available(self):
        return bool(self.monitor.available and self.monitor.thread.is_alive())

    def f8_pressed(self):
        return self.transport.f8_pressed()

    def close(self):
        self.monitor.close()
