action_executor.py

import types
from typing import Any, Callable, Dict, List, Optional, Union

from je_auto_control.utils.exception.exception_tags import (
    action_is_null_error_message, add_command_exception_error_message,
    executor_list_error_message, cant_execute_action_error_message
)
from je_auto_control.utils.exception.exceptions import (
    AutoControlActionException, AutoControlAddCommandException,
    AutoControlActionNullException
)
from je_auto_control.utils.accessibility.accessibility_api import (
    click_accessibility_element, find_accessibility_element,
)
from je_auto_control.utils.self_healing import (
    default_heal_log,
    self_heal_click as _self_heal_click_impl,
    self_heal_locate as _self_heal_locate_impl,
)
from je_auto_control.utils.vision.vlm_api import (
    click_by_description, locate_by_description,
)
from je_auto_control.utils.clipboard.clipboard import (
    get_clipboard, set_clipboard,
)
from je_auto_control.utils.executor.action_schema import validate_actions
from je_auto_control.utils.executor.flow_control import (
    BLOCK_COMMANDS, LoopBreak, LoopContinue,
)
from je_auto_control.utils.executor.mouse_aliases import MOUSE_BUTTON_COMMANDS
from je_auto_control.utils.llm.planner import (
    plan_actions as llm_plan_actions,
    run_from_description as llm_run_from_description,
)
from je_auto_control.utils.remote_desktop.registry import (
    registry as remote_desktop_registry,
)
from je_auto_control.utils.rest_api.rest_registry import (
    rest_api_registry,
)
from je_auto_control.utils.admin.admin_client import (
    default_admin_console,
)
from je_auto_control.utils.ocr.ocr_engine import (
    click_text as ocr_click_text,
    find_text_regex as ocr_find_text_regex,
    locate_text_center as ocr_locate_text_center,
    read_text_in_region as ocr_read_text_in_region,
    wait_for_text as ocr_wait_for_text,
)
from je_auto_control.utils.profiler.profiler import default_profiler
from je_auto_control.utils.run_history.history_store import default_history_store
from je_auto_control.utils.secrets import default_secret_manager
from je_auto_control.utils.script_vars.interpolate import (
    interpolate_actions, interpolate_value,
)
from je_auto_control.utils.script_vars.scope import VariableScope
from je_auto_control.utils.http_client.http_client import http_request
from je_auto_control.utils.generate_report.generate_html_report import generate_html, generate_html_report
from je_auto_control.utils.generate_report.generate_json_report import generate_json, generate_json_report
from je_auto_control.utils.generate_report.generate_xml_report import generate_xml, generate_xml_report
from je_auto_control.utils.json.json_file import read_action_json
from je_auto_control.utils.logging.logging_instance import autocontrol_logger
from je_auto_control.utils.mcp_server.http_transport import start_mcp_http_server
from je_auto_control.utils.mcp_server.server import start_mcp_stdio_server
from je_auto_control.utils.package_manager.package_manager_class import package_manager
from je_auto_control.utils.project.create_project_structure import create_project_dir
from je_auto_control.utils.shell_process.shell_exec import default_shell_manager
from je_auto_control.utils.start_exe.start_another_process import start_exe
from je_auto_control.utils.test_record.record_test_class import record_action_to_list, test_record_instance
from je_auto_control.wrapper.auto_control_image import locate_all_image, locate_and_click, locate_image_center
from je_auto_control.wrapper.auto_control_keyboard import (
    check_key_is_press, get_keyboard_keys_table,
    press_keyboard_key, release_keyboard_key, hotkey, type_keyboard, write
)
from je_auto_control.wrapper.auto_control_mouse import (
    get_mouse_position, press_mouse, release_mouse, click_mouse,
    mouse_scroll, get_mouse_table, set_mouse_position
)
from je_auto_control.wrapper.auto_control_record import record, stop_record
from je_auto_control.wrapper.auto_control_screen import screenshot, screen_size
from je_auto_control.wrapper.auto_control_window import (
    close_window_by_title, focus_window, list_windows, wait_for_window,
)


def _a11y_list_as_dicts(app_name: Optional[str] = None,
                        max_results: int = 200) -> List[dict]:
    """Executor adapter: list accessibility elements as plain dicts."""
    from je_auto_control.utils.accessibility.accessibility_api import (
        list_accessibility_elements,
    )
    return [
        element.to_dict()
        for element in list_accessibility_elements(
            app_name=app_name, max_results=int(max_results),
        )
    ]


def _a11y_find_as_dict(name: Optional[str] = None,
                       role: Optional[str] = None,
                       app_name: Optional[str] = None) -> Optional[dict]:
    """Executor adapter: find an accessibility element, return its dict."""
    element = find_accessibility_element(
        name=name, role=role, app_name=app_name,
    )
    return None if element is None else element.to_dict()


def _vlm_locate_as_list(description: str,
                        screen_region: Optional[List[int]] = None,
                        model: Optional[str] = None) -> Optional[List[int]]:
    """Executor adapter: return VLM-located coords as a JSON-safe list."""
    coords = locate_by_description(
        description, screen_region=screen_region, model=model,
    )
    return None if coords is None else [coords[0], coords[1]]


def _self_heal_locate(template_path: Optional[str] = None,
                      description: Optional[str] = None,
                      detect_threshold: float = 0.9,
                      screen_region: Optional[List[int]] = None,
                      model: Optional[str] = None,
                      raise_on_miss: bool = False) -> Dict[str, Any]:
    """Executor adapter: template-first locate with VLM fallback."""
    outcome = _self_heal_locate_impl(
        template_path=template_path, description=description,
        detect_threshold=float(detect_threshold),
        screen_region=screen_region, model=model,
        raise_on_miss=bool(raise_on_miss),
    )
    return outcome.to_dict()


def _self_heal_click(template_path: Optional[str] = None,
                     description: Optional[str] = None,
                     mouse_keycode: str = "mouse_left",
                     detect_threshold: float = 0.9,
                     screen_region: Optional[List[int]] = None,
                     model: Optional[str] = None,
                     raise_on_miss: bool = False) -> Dict[str, Any]:
    """Executor adapter: locate with self-heal, then click."""
    outcome = _self_heal_click_impl(
        template_path=template_path, description=description,
        mouse_keycode=mouse_keycode,
        detect_threshold=float(detect_threshold),
        screen_region=screen_region, model=model,
        raise_on_miss=bool(raise_on_miss),
    )
    return outcome.to_dict()


def _self_heal_log_list(limit: int = 50) -> List[Dict[str, Any]]:
    """Executor adapter: return the recent self-healing events."""
    return [event.to_dict()
            for event in default_heal_log.list_events(limit=int(limit))]


def _self_heal_log_clear() -> Dict[str, Any]:
    default_heal_log.clear()
    return {"cleared": True, "path": str(default_heal_log.path)}


def _run_dag(definition: Dict[str, Any],
             max_parallel: int = 4) -> Dict[str, Any]:
    """Executor adapter: run a cross-host DAG definition."""
    from je_auto_control.utils.dag import run_dag
    return run_dag(definition, max_parallel=int(max_parallel)).to_dict()


_AX_RECORDER_SINGLETON = None


def _a11y_dump(app_name: Optional[str] = None,
                max_results: int = 500) -> Dict[str, Any]:
    """Executor adapter: dump the accessibility tree as nested dict."""
    from je_auto_control.utils.accessibility import dump_accessibility_tree
    return dump_accessibility_tree(
        app_name=app_name, max_results=int(max_results),
    ).to_dict()


def _a11y_record_start(app_name: Optional[str] = None,
                        poll_interval_s: float = 0.25,
                        min_movement_px: int = 8) -> Dict[str, Any]:
    """Executor adapter: start the singleton accessibility recorder."""
    from je_auto_control.utils.accessibility import AccessibilityRecorder
    global _AX_RECORDER_SINGLETON
    if (_AX_RECORDER_SINGLETON is not None
            and _AX_RECORDER_SINGLETON.is_running):
        return {"running": True, "already": True}
    _AX_RECORDER_SINGLETON = AccessibilityRecorder(
        app_name=app_name,
        poll_interval_s=float(poll_interval_s),
        min_movement_px=int(min_movement_px),
    )
    _AX_RECORDER_SINGLETON.start()
    return {"running": True, "already": False}


def _a11y_record_stop() -> List[Dict[str, Any]]:
    """Executor adapter: stop the recorder and return the captured events."""
    global _AX_RECORDER_SINGLETON
    if _AX_RECORDER_SINGLETON is None:
        return []
    events = _AX_RECORDER_SINGLETON.stop()
    _AX_RECORDER_SINGLETON = None
    return [event.to_dict() for event in events]


def _a11y_record_events() -> List[Dict[str, Any]]:
    """Executor adapter: peek at events without stopping the recorder."""
    if _AX_RECORDER_SINGLETON is None:
        return []
    return [event.to_dict() for event in _AX_RECORDER_SINGLETON.events()]


def _ab_locate(target_id: str,
               strategies: Dict[str, Dict[str, Any]],
               max_parallel: int = 4,
               record: bool = True) -> Dict[str, Any]:
    """Executor adapter: race N locator strategies for the same target."""
    from je_auto_control.utils.ab_locator import ab_locate
    from je_auto_control.utils.anchor_locator import (
        Locator as AnchorLocator,
    )
    locators = {name: AnchorLocator(**spec)
                for name, spec in strategies.items()}
    return ab_locate(
        target_id=target_id, strategies=locators,
        max_parallel=int(max_parallel), record=bool(record),
    ).to_dict()


def _ab_report(target_id: str) -> Dict[str, Any]:
    from je_auto_control.utils.ab_locator import ab_report_for
    return ab_report_for(target_id).to_dict()


def _ab_best_strategy(target_id: str) -> Dict[str, Any]:
    from je_auto_control.utils.ab_locator import ab_best_strategy
    return {"target_id": target_id,
            "strategy": ab_best_strategy(target_id)}


def _ab_clear() -> Dict[str, Any]:
    from je_auto_control.utils.ab_locator import default_ab_store
    default_ab_store.clear()
    return {"cleared": True}


def _failure_hook_fire(source: str, source_id: str,
                       error_text: str = "",
                       script_path: Optional[str] = None,
                       screenshot_path: Optional[str] = None,
                       log_tail: str = "",
                       metadata: Optional[Dict[str, Any]] = None,
                       ) -> List[Dict[str, Any]]:
    """Executor adapter: file a ticket through every registered backend."""
    from je_auto_control.utils.failure_hooks import (
        FailureReport, default_failure_hook_manager,
    )
    report = FailureReport(
        source=source, source_id=source_id, error_text=error_text,
        script_path=script_path, screenshot_path=screenshot_path,
        log_tail=log_tail, metadata=dict(metadata or {}),
    )
    return [result.to_dict()
            for result in default_failure_hook_manager.fire(report)]


def _failure_hook_list() -> List[Dict[str, Any]]:
    from je_auto_control.utils.failure_hooks import default_failure_hook_manager
    return default_failure_hook_manager.list_backends()


def _failure_hook_clear() -> Dict[str, Any]:
    from je_auto_control.utils.failure_hooks import default_failure_hook_manager
    default_failure_hook_manager.clear()
    return {"cleared": True}


def _costs_record(provider: str, model: str,
                  input_tokens: int, output_tokens: int,
                  label: Optional[str] = None,
                  run_id: Optional[str] = None,
                  user: Optional[str] = None) -> Dict[str, Any]:
    """Executor adapter: append one LLM call to the cost-telemetry log."""
    from je_auto_control.utils.cost_telemetry import record_llm_call
    event = record_llm_call(
        provider=provider, model=model,
        input_tokens=int(input_tokens),
        output_tokens=int(output_tokens),
        label=label, run_id=run_id, user=user,
    )
    return event.to_dict()


def _costs_summary(limit: int = 10000) -> Dict[str, Any]:
    """Executor adapter: aggregate cost events by model / provider / day."""
    from je_auto_control.utils.cost_telemetry import (
        default_cost_store, summarise_llm_costs,
    )
    events = default_cost_store.list_events(limit=int(limit))
    return summarise_llm_costs(events).to_dict()


def _costs_list(limit: int = 100) -> List[Dict[str, Any]]:
    from je_auto_control.utils.cost_telemetry import default_cost_store
    return [event.to_dict()
            for event in default_cost_store.list_events(limit=int(limit))]


def _costs_clear() -> Dict[str, Any]:
    from je_auto_control.utils.cost_telemetry import default_cost_store
    default_cost_store.clear()
    return {"cleared": True, "path": str(default_cost_store.path)}


def _wait_screen_stable(region: Optional[List[int]] = None,
                        timeout_s: float = 10.0,
                        poll_interval_s: float = 0.2,
                        stable_for_s: float = 0.5,
                        max_pixel_diff: int = 0) -> Dict[str, Any]:
    """Executor adapter: smart wait for the screen to stop moving."""
    from je_auto_control.utils.smart_waits import wait_until_screen_stable
    return wait_until_screen_stable(
        region=region, timeout_s=float(timeout_s),
        poll_interval_s=float(poll_interval_s),
        stable_for_s=float(stable_for_s),
        max_pixel_diff=int(max_pixel_diff),
    ).to_dict()


def _wait_pixel_changes(x: int, y: int,
                         timeout_s: float = 10.0,
                         poll_interval_s: float = 0.1,
                         rgb_tolerance: int = 5) -> Dict[str, Any]:
    """Executor adapter: smart wait for one pixel to change colour."""
    from je_auto_control.utils.smart_waits import wait_until_pixel_changes
    return wait_until_pixel_changes(
        x=int(x), y=int(y),
        timeout_s=float(timeout_s),
        poll_interval_s=float(poll_interval_s),
        rgb_tolerance=int(rgb_tolerance),
    ).to_dict()


def _wait_clipboard_change(baseline: Optional[str] = None,
                           target: Optional[str] = None,
                           contains: bool = False,
                           timeout_s: float = 10.0,
                           poll_interval_s: float = 0.2) -> Dict[str, Any]:
    """Executor adapter: wait until the clipboard changes (or matches target)."""
    from je_auto_control.utils.smart_waits import wait_until_clipboard_changes
    return wait_until_clipboard_changes(
        baseline=baseline, target=target, contains=bool(contains),
        timeout_s=float(timeout_s), poll_interval_s=float(poll_interval_s),
    ).to_dict()


def _wait_window_closed(title: str, case_sensitive: bool = False,
                        timeout_s: float = 10.0,
                        poll_interval_s: float = 0.2) -> Dict[str, Any]:
    """Executor adapter: wait until a window matching ``title`` disappears."""
    from je_auto_control.utils.smart_waits import wait_until_window_closed
    return wait_until_window_closed(
        title, case_sensitive=bool(case_sensitive),
        timeout_s=float(timeout_s), poll_interval_s=float(poll_interval_s),
    ).to_dict()


def _wait_region_idle(region: List[int],
                      timeout_s: float = 10.0,
                      poll_interval_s: float = 0.2,
                      stable_for_s: float = 0.5,
                      max_pixel_diff: int = 0) -> Dict[str, Any]:
    """Executor adapter: smart wait for a sub-region to stop moving."""
    from je_auto_control.utils.smart_waits import wait_until_region_idle
    return wait_until_region_idle(
        region=region, timeout_s=float(timeout_s),
        poll_interval_s=float(poll_interval_s),
        stable_for_s=float(stable_for_s),
        max_pixel_diff=int(max_pixel_diff),
    ).to_dict()


def _wait_for_file(path: str, timeout_s: float = 30.0,
                   poll_interval_s: float = 0.25,
                   stable_for_s: float = 1.0,
                   min_size: int = 1) -> Dict[str, Any]:
    """Executor adapter: wait until a file exists and finishes being written."""
    from je_auto_control.utils.smart_waits import wait_until_file
    return wait_until_file(
        path, timeout_s=float(timeout_s),
        poll_interval_s=float(poll_interval_s),
        stable_for_s=float(stable_for_s), min_size=int(min_size),
    ).to_dict()


def _wait_for_port(host: str, port: int, timeout_s: float = 30.0,
                   poll_interval_s: float = 0.25,
                   connect_timeout_s: float = 1.0) -> Dict[str, Any]:
    """Executor adapter: wait until a TCP port accepts connections."""
    from je_auto_control.utils.smart_waits import wait_until_port
    return wait_until_port(
        host, int(port), timeout_s=float(timeout_s),
        poll_interval_s=float(poll_interval_s),
        connect_timeout_s=float(connect_timeout_s),
    ).to_dict()


def _wait_for_process(name: str, present: bool = True, timeout_s: float = 30.0,
                      poll_interval_s: float = 0.25) -> Dict[str, Any]:
    """Executor adapter: wait until a process appears or exits."""
    from je_auto_control.utils.smart_waits import wait_until_process
    return wait_until_process(
        name, present=bool(present), timeout_s=float(timeout_s),
        poll_interval_s=float(poll_interval_s),
    ).to_dict()


def _ocr_read_structure(region: Optional[List[int]] = None,
                        lang: str = "eng",
                        min_confidence: float = 60.0,
                        ) -> Dict[str, Any]:
    """Executor adapter: structured OCR (rows / tables / form fields)."""
    from je_auto_control.utils.ocr.structure import read_structure
    structured = read_structure(
        region=region, lang=lang,
        min_confidence=float(min_confidence),
    )
    return structured.to_dict()


def _anchor_locate(anchor: Dict[str, Any], target: Dict[str, Any],
                   relation: str = "near",
                   max_distance_px: float = 200.0) -> Dict[str, Any]:
    """Executor adapter: anchor-based spatial locator."""
    from je_auto_control.utils.anchor_locator import (
        Locator, anchor_locate,
    )
    anchor_loc = Locator(**anchor)
    target_loc = Locator(**target)
    outcome = anchor_locate(
        anchor=anchor_loc, target=target_loc,
        relation=relation, max_distance_px=float(max_distance_px),
    )
    return outcome.to_dict()


def _anchor_click(anchor: Dict[str, Any], target: Dict[str, Any],
                  mouse_keycode: str = "mouse_left",
                  relation: str = "near",
                  max_distance_px: float = 200.0) -> Dict[str, Any]:
    """Executor adapter: anchor-locate + click."""
    outcome = _anchor_locate(anchor, target, relation, max_distance_px)
    if outcome.get("found") and outcome.get("target_coords"):
        cx, cy = outcome["target_coords"]
        from je_auto_control.wrapper.auto_control_mouse import (
            click_mouse, set_mouse_position,
        )
        set_mouse_position(int(cx), int(cy))
        click_mouse(mouse_keycode, int(cx), int(cy))
    return outcome


def _chatops_dispatch(message: str,
                      context: Optional[Dict[str, Any]] = None,
                      script_root: Optional[str] = None) -> Dict[str, Any]:
    """Executor adapter: route one chat message through the default router."""
    from je_auto_control.utils.chatops import (
        CommandRouter, register_chatops_default_commands,
    )
    router = CommandRouter()
    register_chatops_default_commands(router)
    merged_context: Dict[str, Any] = dict(context or {})
    if script_root is not None:
        merged_context.setdefault("script_root", script_root)
    result = router.dispatch(message, context=merged_context)
    return {"matched": False} if result is None else {
        "matched": True, **result.to_dict(),
    }


def _presence_register(viewer_id: str, label: str = "",
                       role: str = "observer") -> Dict[str, Any]:
    from je_auto_control.utils.remote_desktop.presence import (
        default_presence_registry,
    )
    return default_presence_registry().register(
        viewer_id, label, role=role,
    ).to_dict()


def _presence_unregister(viewer_id: str) -> Dict[str, Any]:
    from je_auto_control.utils.remote_desktop.presence import (
        default_presence_registry,
    )
    removed = default_presence_registry().unregister(viewer_id)
    return {"viewer_id": viewer_id, "removed": removed}


def _presence_update_cursor(viewer_id: str, x: int, y: int) -> Dict[str, Any]:
    from je_auto_control.utils.remote_desktop.presence import (
        default_presence_registry,
    )
    return default_presence_registry().update_cursor(
        viewer_id, int(x), int(y),
    ).to_dict()


def _presence_set_role(viewer_id: str, role: str) -> Dict[str, Any]:
    from je_auto_control.utils.remote_desktop.presence import (
        default_presence_registry,
    )
    return default_presence_registry().update_role(viewer_id, role).to_dict()


def _presence_list() -> List[Dict[str, Any]]:
    from je_auto_control.utils.remote_desktop.presence import (
        default_presence_registry,
    )
    return [row.to_dict() for row in default_presence_registry().list()]


def _presence_clear() -> Dict[str, Any]:
    from je_auto_control.utils.remote_desktop.presence import (
        default_presence_registry,
    )
    default_presence_registry().clear()
    return {"cleared": True}


def _run_agent(goal: str,
               backend: str = "anthropic",
               max_steps: int = 25,
               wall_seconds: float = 300.0,
               model: Optional[str] = None,
               max_tokens: int = 1024) -> Dict[str, Any]:
    """Executor adapter: drive the closed-loop ``AgentLoop`` against ``goal``.

    ``backend`` selects between the production backends (Anthropic /
    OpenAI). The Anthropic computer-use raw path remains available
    via :func:`_computer_use` / ``AC_computer_use``.
    """
    from je_auto_control.utils.agent import AgentBudget, AgentLoop
    from je_auto_control.utils.agent.backends import (
        AnthropicAgentBackend, OpenAIAgentBackend,
    )
    from je_auto_control.utils.tool_use_schema import (
        export_anthropic_tools, export_openai_tools,
    )
    name = (backend or "anthropic").strip().lower()
    if name == "anthropic":
        tools = export_anthropic_tools()
        backend_obj = AnthropicAgentBackend(
            tools=tools,
            model=model or "claude-opus-4-7",
            max_tokens=int(max_tokens),
        )
    elif name == "openai":
        tools = export_openai_tools()
        # OpenAIAgentBackend does not accept max_tokens (Anthropic-only).
        backend_obj = OpenAIAgentBackend(
            tools=tools,
            model=model or "gpt-4o",
        )
    else:
        raise ValueError(f"unknown agent backend: {backend!r}")
    budget = AgentBudget(
        max_steps=int(max_steps), wall_seconds=float(wall_seconds),
    )
    result = AgentLoop(backend_obj, budget=budget).run(goal)
    return {
        "succeeded": bool(result.succeeded),
        "elapsed_s": float(result.elapsed_s),
        "final_message": result.final_message,
        "steps": [
            {
                "index": step.index,
                "tool": step.tool,
                "arguments": step.arguments,
                "error": step.error,
                "stop_reason": step.stop_reason,
            }
            for step in result.steps
        ],
    }


def _redact_screenshot(file_path: str,
                       output_path: Optional[str] = None,
                       policy: str = "moderate",
                       regions: Optional[List[List[int]]] = None,
                       accessibility: Optional[List[Dict[str, Any]]] = None,
                       ocr: Optional[List[Dict[str, Any]]] = None,
                       ) -> Dict[str, Any]:
    """Executor adapter: blur PII regions in a saved screenshot.

    Reads ``file_path``, applies the chosen redaction policy
    (optionally with caller-supplied accessibility / OCR context),
    and writes the result to ``output_path`` (or overwrites the
    source when omitted). Returns ``{output_path, boxes,
    detectors_used}`` for downstream audit.
    """
    from je_auto_control.utils.redaction import (
        RedactionEngine, policy_from_name,
    )
    target = output_path or file_path
    chosen = policy_from_name(policy)
    if regions:
        chosen = chosen.with_extra_regions(
            [tuple(int(v) for v in r) for r in regions],
        )
    engine = RedactionEngine(chosen)
    context: Dict[str, Any] = {}
    if accessibility is not None:
        context["accessibility"] = list(accessibility)
    if ocr is not None:
        context["ocr"] = [(item["text"], item["bbox"]) for item in ocr]
    with open(file_path, "rb") as src:
        png_bytes = src.read()
    redacted, result = engine.redact_bytes(png_bytes, context)
    with open(target, "wb") as dest:
        dest.write(redacted)
    return {
        "output_path": str(target),
        "boxes": [list(b) for b in result.boxes],
        "detectors_used": list(result.detectors_used),
    }


def _assert_text(text: str,
                 region: Optional[List[int]] = None,
                 lang: str = "eng",
                 regex: bool = False,
                 present: bool = True,
                 ignore_case: bool = True,
                 min_confidence: float = 60.0,
                 raise_on_fail: bool = True,
                 capture_on_fail: bool = False) -> Dict[str, Any]:
    """Executor adapter: assert OCR text is (not) on screen."""
    from je_auto_control.utils.assertion import assert_text
    return assert_text(
        text, region=region, lang=lang, regex=bool(regex),
        present=bool(present), ignore_case=bool(ignore_case),
        min_confidence=float(min_confidence),
        raise_on_fail=bool(raise_on_fail),
        capture_on_fail=bool(capture_on_fail),
    ).to_dict()


def _assert_image(template_path: str,
                  threshold: float = 0.9,
                  present: bool = True,
                  raise_on_fail: bool = True,
                  capture_on_fail: bool = False) -> Dict[str, Any]:
    """Executor adapter: assert a template image is (not) on screen."""
    from je_auto_control.utils.assertion import assert_image
    return assert_image(
        template_path, threshold=float(threshold), present=bool(present),
        raise_on_fail=bool(raise_on_fail),
        capture_on_fail=bool(capture_on_fail),
    ).to_dict()


def _assert_pixel(x: int, y: int, rgb: List[int],
                  tolerance: int = 0,
                  match: bool = True,
                  raise_on_fail: bool = True,
                  capture_on_fail: bool = False) -> Dict[str, Any]:
    """Executor adapter: assert a pixel matches (or differs from) ``rgb``."""
    from je_auto_control.utils.assertion import assert_pixel
    return assert_pixel(
        int(x), int(y), rgb, tolerance=int(tolerance), match=bool(match),
        raise_on_fail=bool(raise_on_fail),
        capture_on_fail=bool(capture_on_fail),
    ).to_dict()


def _assert_window(title: str,
                   exists: bool = True,
                   ignore_case: bool = True,
                   raise_on_fail: bool = True,
                   capture_on_fail: bool = False) -> Dict[str, Any]:
    """Executor adapter: assert a window matching ``title`` does (not) exist."""
    from je_auto_control.utils.assertion import assert_window
    return assert_window(
        title, exists=bool(exists), ignore_case=bool(ignore_case),
        raise_on_fail=bool(raise_on_fail),
        capture_on_fail=bool(capture_on_fail),
    ).to_dict()


def _assert_vlm(description: str,
                present: bool = True,
                screen_region: Optional[List[int]] = None,
                model: Optional[str] = None,
                raise_on_fail: bool = True,
                capture_on_fail: bool = False) -> Dict[str, Any]:
    """Executor adapter: assert the screen matches a description (VLM judged)."""
    from je_auto_control.utils.assertion import assert_by_description
    return assert_by_description(
        description, present=bool(present), screen_region=screen_region,
        model=model, raise_on_fail=bool(raise_on_fail),
        capture_on_fail=bool(capture_on_fail),
    ).to_dict()


def _assert_clipboard(text: str,
                      mode: str = "equals",
                      ignore_case: bool = False,
                      present: bool = True,
                      raise_on_fail: bool = True,
                      capture_on_fail: bool = False) -> Dict[str, Any]:
    """Executor adapter: assert clipboard text matches ``text``."""
    from je_auto_control.utils.assertion import assert_clipboard
    return assert_clipboard(
        text, mode=mode, ignore_case=bool(ignore_case),
        present=bool(present), raise_on_fail=bool(raise_on_fail),
        capture_on_fail=bool(capture_on_fail),
    ).to_dict()


def _assert_process(name: str,
                    running: bool = True,
                    raise_on_fail: bool = True,
                    capture_on_fail: bool = False) -> Dict[str, Any]:
    """Executor adapter: assert a process matching ``name`` is (not) running."""
    from je_auto_control.utils.assertion import assert_process
    return assert_process(
        name, running=bool(running), raise_on_fail=bool(raise_on_fail),
        capture_on_fail=bool(capture_on_fail),
    ).to_dict()


def _assert_file(path: str,
                 exists: bool = True,
                 contains: Optional[str] = None,
                 sha256: Optional[str] = None,
                 min_size: Optional[int] = None,
                 raise_on_fail: bool = True) -> Dict[str, Any]:
    """Executor adapter: assert a file's existence / content / hash / size."""
    from je_auto_control.utils.assertion import assert_file
    return assert_file(
        path, exists=bool(exists), contains=contains, sha256=sha256,
        min_size=None if min_size is None else int(min_size),
        raise_on_fail=bool(raise_on_fail),
    ).to_dict()


def _assert_http(url: str,
                 status: int = 200,
                 contains: Optional[str] = None,
                 timeout: float = 10.0,
                 method: str = "GET",
                 raise_on_fail: bool = True) -> Dict[str, Any]:
    """Executor adapter: assert an HTTP(S) endpoint status / body."""
    from je_auto_control.utils.assertion import assert_http
    return assert_http(
        url, status=int(status), contains=contains, timeout=float(timeout),
        method=method, raise_on_fail=bool(raise_on_fail),
    ).to_dict()


def _assert_all(specs: List[Dict[str, Any]],
                raise_on_fail: bool = True) -> Dict[str, Any]:
    """Executor adapter: run a batch of assertion specs (soft assertions)."""
    from je_auto_control.utils.assertion import assert_all
    return assert_all(specs, raise_on_fail=bool(raise_on_fail)).to_dict()


def _assert_any(specs: List[Dict[str, Any]],
                raise_on_fail: bool = True) -> Dict[str, Any]:
    """Executor adapter: pass when at least one assertion spec passes."""
    from je_auto_control.utils.assertion import assert_any
    return assert_any(specs, raise_on_fail=bool(raise_on_fail)).to_dict()


def _assert_eventually(spec: Dict[str, Any],
                       timeout: float = 5.0,
                       interval: float = 0.25,
                       raise_on_fail: bool = True) -> Dict[str, Any]:
    """Executor adapter: poll one assertion spec until it passes / times out."""
    from je_auto_control.utils.assertion import assert_eventually
    return assert_eventually(
        spec, timeout=float(timeout), interval=float(interval),
        raise_on_fail=bool(raise_on_fail),
    ).to_dict()


def _load_data(source: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Executor adapter: load tabular rows from a data source spec."""
    from je_auto_control.utils.data_source import load_rows
    return load_rows(source)


def _flaky_report(limit: int = 500,
                  min_runs: int = 2,
                  group_by: str = "script_path") -> Dict[str, Any]:
    """Executor adapter: score run-history flakiness per script / source."""
    from je_auto_control.utils.flakiness import analyze_flakiness
    return analyze_flakiness(
        limit=int(limit), min_runs=int(min_runs), group_by=group_by,
    ).to_dict()


def _run_suite(spec: Dict[str, Any],
               tags: Optional[List[str]] = None,
               respect_quarantine: bool = True,
               junit_path: Optional[str] = None,
               allure_dir: Optional[str] = None) -> Dict[str, Any]:
    """Executor adapter: run a QA suite and optionally write CI reports."""
    from je_auto_control.utils.test_suite import (
        run_suite, write_allure_results, write_junit_xml,
    )
    result = run_suite(
        spec, executor=executor, tags=tags,
        respect_quarantine=bool(respect_quarantine),
    )
    payload = result.to_dict()
    reports: Dict[str, Any] = {}
    if junit_path:
        reports["junit"] = write_junit_xml(result, junit_path)
    if allure_dir:
        reports["allure"] = write_allure_results(result, allure_dir)
    if reports:
        payload["reports"] = reports
    return payload


def _quarantine_add(name: str, reason: str = "") -> Dict[str, Any]:
    from je_auto_control.utils.quarantine import default_quarantine_store
    return default_quarantine_store().add(name, reason=reason).to_dict()


def _quarantine_remove(name: str) -> Dict[str, Any]:
    from je_auto_control.utils.quarantine import default_quarantine_store
    return {"name": name, "removed": default_quarantine_store().remove(name)}


def _quarantine_list() -> List[Dict[str, Any]]:
    from je_auto_control.utils.quarantine import default_quarantine_store
    return [entry.to_dict() for entry in default_quarantine_store().list()]


def _quarantine_clear() -> Dict[str, Any]:
    from je_auto_control.utils.quarantine import default_quarantine_store
    return {"cleared": default_quarantine_store().clear()}


def _quarantine_auto(flip_rate_threshold: float = 0.5,
                     min_runs: int = 3,
                     limit: int = 500,
                     group_by: str = "script_path") -> List[Dict[str, Any]]:
    from je_auto_control.utils.quarantine import auto_quarantine_from_flakiness
    return [
        entry.to_dict()
        for entry in auto_quarantine_from_flakiness(
            flip_rate_threshold=float(flip_rate_threshold),
            min_runs=int(min_runs), limit=int(limit), group_by=group_by,
        )
    ]


def _audit_accessibility(app_name: Optional[str] = None,
                         contrast_pairs: Optional[List[Dict[str, Any]]] = None,
                         texts: Optional[List[str]] = None,
                         min_ratio: float = 4.5,
                         max_results: int = 500) -> Dict[str, Any]:
    """Executor adapter: run the accessibility / i18n audit."""
    from je_auto_control.utils.a11y_audit import run_audit
    return run_audit(
        app_name=app_name, contrast_pairs=contrast_pairs, texts=texts,
        min_ratio=float(min_ratio), max_results=int(max_results),
    ).to_dict()


def _audit_contrast(foreground: List[int], background: List[int],
                    min_ratio: float = 4.5) -> Dict[str, Any]:
    """Executor adapter: WCAG contrast ratio for one colour pair."""
    from je_auto_control.utils.a11y_audit import contrast_ratio
    ratio = contrast_ratio(foreground, background)
    return {
        "ratio": round(ratio, 2),
        "passes_aa": ratio >= float(min_ratio),
        "foreground": list(foreground), "background": list(background),
    }


def _wcag_audit(app_name: Optional[str] = None,
                contrast_pairs: Optional[List[Dict[str, Any]]] = None,
                texts: Optional[List[str]] = None, level: str = "AA",
                min_target_px: int = 24, max_results: int = 500
                ) -> Dict[str, Any]:
    """Executor adapter: WCAG-tagged conformance audit (SC ids + levels)."""
    from je_auto_control.utils.a11y_audit import wcag_audit
    return wcag_audit(
        app_name=app_name, contrast_pairs=contrast_pairs, texts=texts,
        level=str(level), min_target_px=int(min_target_px),
        max_results=int(max_results))


def _run_device_matrix(actions: List[Any], devices: List[Dict[str, Any]],
                       max_parallel: int = 4,
                       var_name: str = "device") -> Dict[str, Any]:
    """Executor adapter: run an action list across many devices in parallel."""
    from je_auto_control.utils.device_matrix import run_on_devices
    return run_on_devices(
        actions, devices, max_parallel=int(max_parallel), var_name=var_name,
    ).to_dict()


def _assert_audio(duration_s: float = 1.0,
                  threshold: float = 0.01,
                  expect_sound: bool = True,
                  samplerate: int = 44100,
                  channels: int = 1,
                  raise_on_fail: bool = True) -> Dict[str, Any]:
    """Executor adapter: assert audio activity / silence."""
    from je_auto_control.utils.media_assert import assert_audio_activity
    return assert_audio_activity(
        duration_s=float(duration_s), threshold=float(threshold),
        expect_sound=bool(expect_sound), samplerate=int(samplerate),
        channels=int(channels), raise_on_fail=bool(raise_on_fail),
    ).to_dict()


def _assert_video_changes(video_path: str,
                          start_s: float = 0.0,
                          end_s: Optional[float] = None,
                          threshold: float = 1.0,
                          expect_motion: bool = True,
                          region: Optional[List[int]] = None,
                          raise_on_fail: bool = True) -> Dict[str, Any]:
    """Executor adapter: assert a video segment has motion / is static."""
    from je_auto_control.utils.media_assert import assert_video_changes
    return assert_video_changes(
        video_path, start_s=float(start_s),
        end_s=None if end_s is None else float(end_s),
        threshold=float(threshold), expect_motion=bool(expect_motion),
        region=region, raise_on_fail=bool(raise_on_fail),
    ).to_dict()


def _computer_use(goal: str,
                  display_width_px: Optional[int] = None,
                  display_height_px: Optional[int] = None,
                  display_number: Optional[int] = None,
                  max_steps: int = 25,
                  wall_seconds: float = 300.0,
                  model: str = "claude-opus-4-7",
                  max_tokens: int = 1024) -> Dict[str, Any]:
    """Executor adapter: run Anthropic Computer-Use to achieve ``goal``."""
    from je_auto_control.utils.agent.computer_use import (
        result_to_dict, run_computer_use,
    )
    result = run_computer_use(
        goal,
        display_width_px=display_width_px,
        display_height_px=display_height_px,
        display_number=display_number,
        max_steps=int(max_steps), wall_seconds=float(wall_seconds),
        model=model, max_tokens=int(max_tokens),
    )
    return result_to_dict(result)


def _remote_start_host(token: str,
                       bind: str = "127.0.0.1",
                       port: int = 0,
                       fps: float = 10.0,
                       quality: int = 70,
                       region: Optional[List[int]] = None,
                       max_clients: int = 4) -> Dict[str, Any]:
    """Executor adapter: start the singleton remote-desktop host."""
    return remote_desktop_registry.start_host(
        token=token, bind=bind, port=int(port),
        fps=float(fps), quality=int(quality),
        region=region, max_clients=int(max_clients),
    )


def _remote_stop_host() -> Dict[str, Any]:
    return remote_desktop_registry.stop_host()


def _remote_host_status() -> Dict[str, Any]:
    return remote_desktop_registry.host_status()


def _remote_connect(host: str, port: int, token: str,
                    timeout: float = 5.0) -> Dict[str, Any]:
    """Executor adapter: connect the singleton viewer."""
    return remote_desktop_registry.connect_viewer(
        host=host, port=int(port), token=token, timeout=float(timeout),
    )


def _remote_disconnect() -> Dict[str, Any]:
    return remote_desktop_registry.disconnect_viewer()


def _remote_viewer_status() -> Dict[str, Any]:
    return remote_desktop_registry.viewer_status()


def _remote_send_input(action: Dict[str, Any]) -> Dict[str, Any]:
    return remote_desktop_registry.send_input(action)


# --- WebSocket-transport remote desktop ------------------------------------

def _ws_start_host(token: str,
                   bind: str = "127.0.0.1",
                   port: int = 0,
                   fps: float = 10.0,
                   quality: int = 70,
                   region: Optional[List[int]] = None,
                   max_clients: int = 4) -> Dict[str, Any]:
    """Executor adapter: start the singleton WebSocket-transport host."""
    return remote_desktop_registry.start_ws_host(
        token=token, bind=bind, port=int(port),
        fps=float(fps), quality=int(quality),
        region=region, max_clients=int(max_clients),
    )


def _ws_stop_host() -> Dict[str, Any]:
    return remote_desktop_registry.stop_ws_host()


def _ws_host_status() -> Dict[str, Any]:
    return remote_desktop_registry.ws_host_status()


def _ws_connect(host: str, port: int, token: str,
                path: str = "/",
                timeout: float = 5.0) -> Dict[str, Any]:
    """Executor adapter: connect the singleton WS viewer."""
    return remote_desktop_registry.connect_ws_viewer(
        host=host, port=int(port), token=token,
        path=path, timeout=float(timeout),
    )


def _ws_disconnect() -> Dict[str, Any]:
    return remote_desktop_registry.disconnect_ws_viewer()


def _ws_viewer_status() -> Dict[str, Any]:
    return remote_desktop_registry.ws_viewer_status()


def _ws_send_input(action: Dict[str, Any]) -> Dict[str, Any]:
    return remote_desktop_registry.ws_send_input(action)


# --- WebRTC-transport remote desktop (manual SDP signaling) ----------------

def _webrtc_start_host(token: str,
                       read_only: bool = False) -> Dict[str, Any]:
    """Executor adapter: allocate the singleton WebRTC host.

    Follow up with ``AC_webrtc_create_offer`` then
    ``AC_webrtc_accept_answer`` once the viewer's answer SDP arrives.
    """
    return remote_desktop_registry.start_webrtc_host(
        token=token, read_only=bool(read_only),
    )


def _webrtc_create_offer(peer_label: str = "remote viewer") -> Dict[str, Any]:
    return remote_desktop_registry.webrtc_create_offer(peer_label=peer_label)


def _webrtc_accept_answer(answer_sdp: str) -> Dict[str, Any]:
    return remote_desktop_registry.webrtc_accept_answer(answer_sdp)


def _webrtc_stop_host() -> Dict[str, Any]:
    return remote_desktop_registry.stop_webrtc_host()


def _webrtc_host_status() -> Dict[str, Any]:
    return remote_desktop_registry.webrtc_host_status()


def _webrtc_start_viewer(token: str,
                         viewer_id: Optional[str] = None) -> Dict[str, Any]:
    """Executor adapter: allocate the singleton WebRTC viewer."""
    return remote_desktop_registry.start_webrtc_viewer(
        token=token, viewer_id=viewer_id,
    )


def _webrtc_process_offer(offer_sdp: str,
                          expected_dtls_fingerprint: Optional[str] = None,
                          ) -> Dict[str, Any]:
    return remote_desktop_registry.webrtc_process_offer(
        offer_sdp,
        expected_dtls_fingerprint=expected_dtls_fingerprint,
    )


def _webrtc_send_input(action: Dict[str, Any]) -> Dict[str, Any]:
    return remote_desktop_registry.webrtc_send_input(action)


def _webrtc_stop_viewer() -> Dict[str, Any]:
    return remote_desktop_registry.stop_webrtc_viewer()


def _webrtc_viewer_status() -> Dict[str, Any]:
    return remote_desktop_registry.webrtc_viewer_status()


# --- Virtual gamepad (ViGEm) -----------------------------------------------

def _gamepad_press(button: str) -> Dict[str, Any]:
    from je_auto_control.utils.gamepad import default_gamepad
    default_gamepad().press_button(button)
    return {"button": button, "state": "down"}


def _gamepad_release(button: str) -> Dict[str, Any]:
    from je_auto_control.utils.gamepad import default_gamepad
    default_gamepad().release_button(button)
    return {"button": button, "state": "up"}


def _gamepad_click(button: str) -> Dict[str, Any]:
    from je_auto_control.utils.gamepad import default_gamepad
    default_gamepad().click_button(button)
    return {"button": button, "state": "click"}


def _gamepad_dpad(direction: str) -> Dict[str, Any]:
    from je_auto_control.utils.gamepad import default_gamepad
    default_gamepad().set_dpad(direction)
    return {"dpad": direction}


def _gamepad_left_stick(x: int, y: int) -> Dict[str, Any]:
    from je_auto_control.utils.gamepad import default_gamepad
    default_gamepad().set_left_stick(int(x), int(y))
    return {"left_stick": [int(x), int(y)]}


def _gamepad_right_stick(x: int, y: int) -> Dict[str, Any]:
    from je_auto_control.utils.gamepad import default_gamepad
    default_gamepad().set_right_stick(int(x), int(y))
    return {"right_stick": [int(x), int(y)]}


def _gamepad_left_trigger(value: int) -> Dict[str, Any]:
    from je_auto_control.utils.gamepad import default_gamepad
    default_gamepad().set_left_trigger(int(value))
    return {"left_trigger": int(value)}


def _gamepad_right_trigger(value: int) -> Dict[str, Any]:
    from je_auto_control.utils.gamepad import default_gamepad
    default_gamepad().set_right_trigger(int(value))
    return {"right_trigger": int(value)}


def _gamepad_reset() -> Dict[str, Any]:
    from je_auto_control.utils.gamepad import default_gamepad
    default_gamepad().reset()
    return {"reset": True}


def _rest_api_start(host: str = "127.0.0.1",
                    port: int = 9939,
                    token: Optional[str] = None,
                    enable_audit: bool = True) -> Dict[str, Any]:
    """Executor adapter: start the singleton REST API server."""
    return rest_api_registry.start(
        host=host, port=int(port), token=token,
        enable_audit=bool(enable_audit),
    )


def _rest_api_stop() -> Dict[str, Any]:
    return rest_api_registry.stop()


def _rest_api_status() -> Dict[str, Any]:
    return rest_api_registry.status()


def _admin_add_host(label: str, base_url: str, token: str,
                    tags: Optional[List[str]] = None) -> Dict[str, Any]:
    """Executor adapter: register a remote AutoControl REST endpoint."""
    host = default_admin_console().add_host(
        label=label, base_url=base_url, token=token, tags=tags,
    )
    return {"label": host.label, "base_url": host.base_url, "tags": host.tags}


def _admin_remove_host(label: str) -> Dict[str, Any]:
    return {"removed": default_admin_console().remove_host(label)}


def _admin_list_hosts() -> List[Dict[str, Any]]:
    return [
        {"label": h.label, "base_url": h.base_url, "tags": list(h.tags)}
        for h in default_admin_console().list_hosts()
    ]


def _admin_poll(labels: Optional[List[str]] = None) -> List[Dict[str, Any]]:
    return [
        {
            "label": s.label, "base_url": s.base_url, "healthy": s.healthy,
            "latency_ms": s.latency_ms, "error": s.error,
            "sessions": s.sessions, "job_count": s.job_count,
        }
        for s in default_admin_console().poll_all(labels=labels)
    ]


def _admin_broadcast_execute(actions: List[Any],
                             labels: Optional[List[str]] = None,
                             ) -> List[Dict[str, Any]]:
    return default_admin_console().broadcast_execute(
        actions=actions, labels=labels,
    )


def _audit_log_list(event_type: Optional[str] = None,
                    host_id: Optional[str] = None,
                    limit: int = 200) -> List[Dict[str, Any]]:
    """Executor adapter: query the audit log."""
    from je_auto_control.utils.remote_desktop.audit_log import default_audit_log
    return default_audit_log().query(
        event_type=event_type, host_id=host_id, limit=int(limit),
    )


def _audit_log_verify() -> Dict[str, Any]:
    from je_auto_control.utils.remote_desktop.audit_log import default_audit_log
    result = default_audit_log().verify_chain()
    return {
        "ok": result.ok,
        "broken_at_id": result.broken_at_id,
        "total_rows": result.total_rows,
    }


def _audit_log_clear() -> Dict[str, Any]:
    from je_auto_control.utils.remote_desktop.audit_log import default_audit_log
    return {"deleted": default_audit_log().clear()}


def _inspector_recent(n: int = 60) -> List[Dict[str, Any]]:
    """Executor adapter: most recent N WebRTC stat samples."""
    from je_auto_control.utils.remote_desktop.webrtc_inspector import (
        default_webrtc_inspector,
    )
    return default_webrtc_inspector().recent(int(n))


def _inspector_summary() -> Dict[str, Any]:
    from je_auto_control.utils.remote_desktop.webrtc_inspector import (
        default_webrtc_inspector,
    )
    return default_webrtc_inspector().summary()


def _inspector_reset() -> Dict[str, Any]:
    from je_auto_control.utils.remote_desktop.webrtc_inspector import (
        default_webrtc_inspector,
    )
    return {"cleared": default_webrtc_inspector().reset()}


def _list_usb_devices() -> Dict[str, Any]:
    """Executor adapter: enumerate USB devices on this host."""
    from je_auto_control.utils.usb.usb_devices import list_usb_devices
    return list_usb_devices().to_dict()


def _diagnose() -> Dict[str, Any]:
    """Executor adapter: run system diagnostics and return the report."""
    from je_auto_control.utils.diagnostics.diagnostics import run_diagnostics
    return run_diagnostics().to_dict()


def _config_export() -> Dict[str, Any]:
    """Executor adapter: build the config bundle dict in-memory."""
    from je_auto_control.utils.config_bundle import export_config_bundle
    return export_config_bundle()


def _config_import(bundle: Dict[str, Any],
                   dry_run: bool = False) -> Dict[str, Any]:
    """Executor adapter: apply a config bundle dict to the user config root."""
    from je_auto_control.utils.config_bundle import import_config_bundle
    return import_config_bundle(bundle, dry_run=bool(dry_run)).to_dict()


def _usb_watch_start(poll_interval_s: float = 2.0) -> Dict[str, Any]:
    """Executor adapter: start the singleton USB hotplug watcher."""
    from je_auto_control.utils.usb.usb_watcher import default_usb_watcher
    watcher = default_usb_watcher()
    # poll_interval_s is consumed at watcher construction time only;
    # honor it on a fresh singleton, otherwise just (re-)start.
    watcher.start()
    return {"running": watcher.is_running, "interval_s": poll_interval_s}


def _usb_watch_stop() -> Dict[str, Any]:
    from je_auto_control.utils.usb.usb_watcher import default_usb_watcher
    watcher = default_usb_watcher()
    watcher.stop()
    return {"running": watcher.is_running}


def _usb_recent_events(since: int = 0,
                       limit: Optional[int] = None) -> List[Dict[str, Any]]:
    from je_auto_control.utils.usb.usb_watcher import default_usb_watcher
    return default_usb_watcher().recent_events(
        since=int(since),
        limit=int(limit) if limit is not None else None,
    )


# --- USB passthrough (Phase 2) — delegate to the shared command module --

def _usb_passthrough_enable(enabled: bool = True) -> Dict[str, Any]:
    from je_auto_control.utils.usb.passthrough import commands
    return commands.passthrough_enable(enabled)


def _usb_passthrough_status() -> Dict[str, Any]:
    from je_auto_control.utils.usb.passthrough import commands
    return commands.passthrough_status()


def _usb_acl_list() -> Dict[str, Any]:
    from je_auto_control.utils.usb.passthrough import commands
    return commands.acl_list()


def _usb_acl_add(vendor_id: str, product_id: str,
                 serial: Optional[str] = None, allow: bool = True,
                 prompt_on_open: bool = False, label: str = "") -> Dict[str, Any]:
    from je_auto_control.utils.usb.passthrough import commands
    return commands.acl_add(
        vendor_id, product_id, serial=serial, allow=allow,
        prompt_on_open=prompt_on_open, label=label,
    )


def _usb_acl_remove(vendor_id: str, product_id: str,
                    serial: Optional[str] = None) -> Dict[str, Any]:
    from je_auto_control.utils.usb.passthrough import commands
    return commands.acl_remove(vendor_id, product_id, serial=serial)


def _usb_acl_set_default(policy: str) -> Dict[str, Any]:
    from je_auto_control.utils.usb.passthrough import commands
    return commands.acl_set_default(policy)


def _usb_acl_export(path: str) -> Dict[str, Any]:
    from je_auto_control.utils.usb.passthrough import commands
    return commands.acl_export(path)


def _usb_acl_import(path: str, replace: bool = False) -> Dict[str, Any]:
    from je_auto_control.utils.usb.passthrough import commands
    return commands.acl_import(path, replace=replace)


def _usb_loopback_list() -> Dict[str, Any]:
    from je_auto_control.utils.usb.passthrough import commands
    return commands.loopback_list()


def _usb_loopback_open(vendor_id: str, product_id: str,
                       serial: Optional[str] = None) -> Dict[str, Any]:
    from je_auto_control.utils.usb.passthrough import commands
    return commands.loopback_open(vendor_id, product_id, serial=serial)


def _usb_remote_list() -> Dict[str, Any]:
    from je_auto_control.utils.usb.passthrough import commands
    return commands.remote_list()


def _usb_remote_open(vendor_id: str, product_id: str,
                     serial: Optional[str] = None) -> Dict[str, Any]:
    from je_auto_control.utils.usb.passthrough import commands
    return commands.remote_open(vendor_id, product_id, serial=serial)


def _ac_web_run(action: Optional[Dict[str, Any]] = None,
                **action_kwargs: Any) -> Any:
    """Bridge one WR_* action into the WebRunner executor (Phase 7.7).

    Accepts ``{"action": "WR_*", "params": {...}}`` either as a positional
    dict or unpacked kwargs so it composes with the existing AC_ schema.
    """
    from je_auto_control.utils.webrunner_bridge import run_webrunner_action
    payload = action if isinstance(action, dict) else action_kwargs
    return run_webrunner_action(payload)


def _ac_web_run_actions(actions: list) -> list:
    """Bridge a list of WR_* actions through the WebRunner executor."""
    from je_auto_control.utils.webrunner_bridge import run_webrunner_actions
    return run_webrunner_actions(actions)


def _ac_web_available() -> bool:
    """Return True when ``je_web_runner`` is importable."""
    from je_auto_control.utils.webrunner_bridge import is_webrunner_available
    return is_webrunner_available()


def _ac_web_list_commands() -> list:
    """Return every WR_* command the local WebRunner install exposes."""
    from je_auto_control.utils.webrunner_bridge import list_webrunner_commands
    return list_webrunner_commands()


def _ac_web_open(url: str, browser: str = "chrome",
                 **driver_kwargs: Any) -> Any:
    """Convenience executor: start a browser then navigate to ``url``."""
    from je_auto_control.utils.webrunner_bridge import web_open
    return web_open(url, browser=browser, **driver_kwargs)


def _ac_web_quit() -> Any:
    """Convenience executor: tear down WebRunner driver sessions."""
    from je_auto_control.utils.webrunner_bridge import web_quit
    return web_quit()


def _ac_web_screenshot(file_path: str) -> Any:
    """Convenience executor: save a screenshot of the active browser."""
    from je_auto_control.utils.webrunner_bridge import web_screenshot
    return web_screenshot(file_path)


def _ac_web_current_url() -> Any:
    """Convenience executor: return the active browser tab's URL."""
    from je_auto_control.utils.webrunner_bridge import web_current_url
    return web_current_url()


# --- Android via ADB (Phase 9.7) ---------------------------------------

_android_client_cache: Dict[Optional[str], Any] = {}


def _android_client(serial: Optional[str] = None,
                    adb_path: Optional[str] = None) -> Any:
    """Build (or return) a cached :class:`AdbClient` for ``serial``."""
    key = (serial, adb_path)
    cached = _android_client_cache.get(key)
    if cached is not None:
        return cached
    from je_auto_control.android import AdbClient
    cached = AdbClient(adb_path=adb_path, default_serial=serial)
    _android_client_cache[key] = cached
    return cached


def _ac_android_tap(x: int, y: int,
                    serial: Optional[str] = None,
                    adb_path: Optional[str] = None) -> None:
    """Send a single ``input tap`` to an Android device."""
    _android_client(serial, adb_path).tap(int(x), int(y))


def _ac_android_swipe(x1: int, y1: int, x2: int, y2: int,
                      duration_ms: int = 250,
                      serial: Optional[str] = None,
                      adb_path: Optional[str] = None) -> None:
    """Send a touch swipe via ``input swipe``."""
    _android_client(serial, adb_path).swipe(
        int(x1), int(y1), int(x2), int(y2),
        duration_ms=int(duration_ms),
    )


def _ac_android_key(key: str,
                    serial: Optional[str] = None,
                    adb_path: Optional[str] = None) -> None:
    """Send a keycode (``KEYCODE_HOME`` etc.) via ``input keyevent``."""
    _android_client(serial, adb_path).key_event(key)


def _ac_android_text(text: str,
                     serial: Optional[str] = None,
                     adb_path: Optional[str] = None) -> None:
    """Type a string via ``input text``."""
    _android_client(serial, adb_path).text(text)


def _ac_android_screenshot(file_path: str,
                           serial: Optional[str] = None,
                           adb_path: Optional[str] = None) -> str:
    """Capture the live Android screen and save it as PNG at ``file_path``."""
    path = _android_client(serial, adb_path).save_screenshot(file_path)
    return str(path)


def _ac_android_list_devices(adb_path: Optional[str] = None) -> list:
    """Return ``{serial, state, model, …}`` for every adb-attached device."""
    devices = _android_client(None, adb_path).list_devices()
    return [
        {"serial": d.serial, "state": d.state,
         "model": d.model, "product": d.product,
         "transport_id": d.transport_id}
        for d in devices
    ]


def _ac_android_shell(command: str,
                      serial: Optional[str] = None,
                      adb_path: Optional[str] = None) -> str:
    """Run an ``adb shell`` command and return its stdout."""
    return _android_client(serial, adb_path).shell(command)


def _ac_android_find_element(text: Optional[str] = None,
                              resource_id: Optional[str] = None,
                              description: Optional[str] = None,
                              class_name: Optional[str] = None,
                              timeout_s: float = 5.0,
                              serial: Optional[str] = None,
                              ) -> Dict[str, int]:
    """Find an Android widget via uiautomator2; return its bounding rect."""
    from je_auto_control.android import (
        UIAutomatorDevice, find_element,
    )
    device = UIAutomatorDevice(serial=serial)
    x1, y1, x2, y2 = find_element(
        text=text, resource_id=resource_id, description=description,
        class_name=class_name, timeout_s=float(timeout_s), device=device,
    )
    return {"x1": x1, "y1": y1, "x2": x2, "y2": y2}


def _ac_android_click_element(text: Optional[str] = None,
                               resource_id: Optional[str] = None,
                               description: Optional[str] = None,
                               class_name: Optional[str] = None,
                               timeout_s: float = 5.0,
                               serial: Optional[str] = None,
                               ) -> Dict[str, int]:
    """Tap the first widget matching the selectors; return click centre."""
    from je_auto_control.android import (
        UIAutomatorDevice, click_element,
    )
    device = UIAutomatorDevice(serial=serial)
    cx, cy = click_element(
        text=text, resource_id=resource_id, description=description,
        class_name=class_name, timeout_s=float(timeout_s), device=device,
    )
    return {"x": cx, "y": cy}


def _ac_android_dump_hierarchy(serial: Optional[str] = None) -> str:
    """Return the device's widget tree as an XML string."""
    from je_auto_control.android import UIAutomatorDevice, dump_hierarchy
    device = UIAutomatorDevice(serial=serial)
    return dump_hierarchy(device=device)


# === iOS executor adapters (WebDriverAgent / facebook-wda) ==================

def _ios_device(url: Optional[str]) -> Any:
    from je_auto_control.ios import IOSDevice
    return IOSDevice(url=url)


def _ac_ios_tap(x: int, y: int, url: Optional[str] = None) -> Dict[str, int]:
    from je_auto_control.ios import tap
    tap(int(x), int(y), device=_ios_device(url))
    return {"x": int(x), "y": int(y)}


def _ac_ios_swipe(x1: int, y1: int, x2: int, y2: int,
                  duration_s: float = 0.5,
                  url: Optional[str] = None) -> Dict[str, Any]:
    from je_auto_control.ios import swipe
    swipe(int(x1), int(y1), int(x2), int(y2),
          duration_s=float(duration_s), device=_ios_device(url))
    return {"x1": int(x1), "y1": int(y1),
            "x2": int(x2), "y2": int(y2)}


def _ac_ios_type(text: str, url: Optional[str] = None) -> str:
    from je_auto_control.ios import type_text
    type_text(text, device=_ios_device(url))
    return text


def _ac_ios_screenshot(file_path: str,
                       url: Optional[str] = None) -> str:
    from je_auto_control.ios import screenshot
    written = screenshot(file_path, device=_ios_device(url))
    if written is None:
        raise RuntimeError("screenshot returned no path")
    return written


def _ac_ios_find_element(name: Optional[str] = None,
                          class_name: Optional[str] = None,
                          predicate: Optional[str] = None,
                          timeout_s: float = 5.0,
                          url: Optional[str] = None) -> Dict[str, int]:
    from je_auto_control.ios import find_element
    x1, y1, x2, y2 = find_element(
        name=name, class_name=class_name, predicate=predicate,
        timeout_s=float(timeout_s), device=_ios_device(url),
    )
    return {"x1": x1, "y1": y1, "x2": x2, "y2": y2}


def _ac_ios_click_element(name: Optional[str] = None,
                           class_name: Optional[str] = None,
                           predicate: Optional[str] = None,
                           timeout_s: float = 5.0,
                           url: Optional[str] = None) -> Dict[str, int]:
    from je_auto_control.ios import click_element
    cx, cy = click_element(
        name=name, class_name=class_name, predicate=predicate,
        timeout_s=float(timeout_s), device=_ios_device(url),
    )
    return {"x": cx, "y": cy}


def _ac_ios_dump_source(url: Optional[str] = None) -> str:
    from je_auto_control.ios import dump_source
    return dump_source(device=_ios_device(url))


def _llm_plan_for_executor(description: str,
                           examples: Optional[list] = None,
                           model: Optional[str] = None,
                           max_tokens: int = 2048) -> list:
    """Executor adapter: plan without executing, using current command set."""
    return llm_plan_actions(
        description,
        known_commands=executor.known_commands(),
        examples=examples,
        model=model,
        max_tokens=int(max_tokens),
    )


def _llm_run_for_executor(description: str,
                          examples: Optional[list] = None,
                          model: Optional[str] = None,
                          max_tokens: int = 2048) -> Dict[str, Any]:
    """Executor adapter: plan and execute against the global executor."""
    return llm_run_from_description(
        description,
        executor=executor,
        examples=examples,
        model=model,
        max_tokens=int(max_tokens),
    )


def _ocr_read_region_as_dicts(region: Optional[List[int]] = None,
                              lang: str = "eng",
                              min_confidence: float = 60.0) -> List[dict]:
    """Executor adapter: dump OCR hits in a region as JSON-friendly dicts."""
    return [
        {
            "text": match.text, "x": match.x, "y": match.y,
            "width": match.width, "height": match.height,
            "confidence": match.confidence,
        }
        for match in ocr_read_text_in_region(
            region=region, lang=lang, min_confidence=float(min_confidence),
        )
    ]


def _ocr_find_regex_as_dicts(pattern: str,
                             lang: str = "eng",
                             region: Optional[List[int]] = None,
                             min_confidence: float = 60.0,
                             flags: int = 0) -> List[dict]:
    """Executor adapter: regex OCR search returning JSON-friendly dicts."""
    return [
        {
            "text": match.text, "x": match.x, "y": match.y,
            "width": match.width, "height": match.height,
            "confidence": match.confidence,
        }
        for match in ocr_find_text_regex(
            pattern, lang=lang, region=region,
            min_confidence=float(min_confidence), flags=int(flags),
        )
    ]


def _email_trigger_add(host: str, username: str, password: str,
                       script_path: str,
                       port: Optional[int] = None,
                       use_ssl: bool = True,
                       mailbox: str = "INBOX",
                       search_criteria: str = "UNSEEN",
                       mark_seen: bool = True,
                       poll_seconds: float = 60.0) -> Dict[str, Any]:
    """Executor adapter: register an IMAP poll trigger."""
    from je_auto_control.utils.triggers.email_trigger import (
        default_email_trigger_watcher,
    )
    trigger = default_email_trigger_watcher.add(
        host=host, username=username, password=password,
        script_path=script_path, port=port, use_ssl=bool(use_ssl),
        mailbox=mailbox, search_criteria=search_criteria,
        mark_seen=bool(mark_seen), poll_seconds=float(poll_seconds),
    )
    return {
        "id": trigger.trigger_id, "host": trigger.host,
        "username": trigger.username, "mailbox": trigger.mailbox,
        "search_criteria": trigger.search_criteria,
        "poll_seconds": trigger.poll_seconds,
    }


def _email_trigger_remove(trigger_id: str) -> Dict[str, Any]:
    from je_auto_control.utils.triggers.email_trigger import (
        default_email_trigger_watcher,
    )
    return {"removed": default_email_trigger_watcher.remove(trigger_id)}


def _email_trigger_list() -> List[Dict[str, Any]]:
    from je_auto_control.utils.triggers.email_trigger import (
        default_email_trigger_watcher,
    )
    rows: List[Dict[str, Any]] = []
    for trigger in default_email_trigger_watcher.list_triggers():
        rows.append({
            "id": trigger.trigger_id, "host": trigger.host,
            "username": trigger.username, "mailbox": trigger.mailbox,
            "script_path": trigger.script_path,
            "search_criteria": trigger.search_criteria,
            "poll_seconds": trigger.poll_seconds,
            "enabled": trigger.enabled, "fired": trigger.fired,
            "last_error": trigger.last_error,
        })
    return rows


def _email_trigger_start() -> Dict[str, Any]:
    from je_auto_control.utils.triggers.email_trigger import (
        default_email_trigger_watcher,
    )
    default_email_trigger_watcher.start()
    return {"running": default_email_trigger_watcher.is_running}


def _email_trigger_stop() -> Dict[str, Any]:
    from je_auto_control.utils.triggers.email_trigger import (
        default_email_trigger_watcher,
    )
    default_email_trigger_watcher.stop()
    return {"running": default_email_trigger_watcher.is_running}


def _email_trigger_poll_once() -> Dict[str, Any]:
    from je_auto_control.utils.triggers.email_trigger import (
        default_email_trigger_watcher,
    )
    return {"fired": default_email_trigger_watcher.poll_once()}


def _webhook_start(host: str = "127.0.0.1", port: int = 0) -> Dict[str, Any]:
    """Executor adapter: start the webhook HTTP server."""
    from je_auto_control.utils.triggers.webhook_server import (
        default_webhook_server,
    )
    bound_host, bound_port = default_webhook_server.start(host, int(port))
    return {"host": bound_host, "port": bound_port}


def _webhook_stop() -> Dict[str, Any]:
    from je_auto_control.utils.triggers.webhook_server import (
        default_webhook_server,
    )
    default_webhook_server.stop()
    return {"running": default_webhook_server.is_running}


def _webhook_add(path: str, script_path: str,
                 methods: Optional[List[str]] = None,
                 token: Optional[str] = None) -> Dict[str, Any]:
    from je_auto_control.utils.triggers.webhook_server import (
        default_webhook_server,
    )
    trigger = default_webhook_server.add(
        path=path, script_path=script_path,
        methods=methods, token=token,
    )
    return {
        "id": trigger.webhook_id, "path": trigger.path,
        "methods": list(trigger.methods),
        "script_path": trigger.script_path,
        "has_token": bool(trigger.token),
    }


def _webhook_remove(webhook_id: str) -> Dict[str, Any]:
    from je_auto_control.utils.triggers.webhook_server import (
        default_webhook_server,
    )
    return {"removed": default_webhook_server.remove(webhook_id)}


def _webhook_list() -> List[Dict[str, Any]]:
    from je_auto_control.utils.triggers.webhook_server import (
        default_webhook_server,
    )
    rows: List[Dict[str, Any]] = []
    for trigger in default_webhook_server.list_webhooks():
        rows.append({
            "id": trigger.webhook_id, "path": trigger.path,
            "methods": list(trigger.methods),
            "script_path": trigger.script_path,
            "enabled": trigger.enabled, "fired": trigger.fired,
            "has_token": bool(trigger.token),
        })
    return rows


def _webhook_status() -> Dict[str, Any]:
    from je_auto_control.utils.triggers.webhook_server import (
        default_webhook_server,
    )
    bound = default_webhook_server.bound_address
    return {
        "running": default_webhook_server.is_running,
        "host": bound[0] if bound else None,
        "port": bound[1] if bound else None,
        "registered": len(default_webhook_server.list_webhooks()),
    }


def _secret_initialize(passphrase: str) -> Dict[str, Any]:
    """Executor adapter: create a fresh vault under ``passphrase``."""
    default_secret_manager.initialize(passphrase)
    return {
        "initialized": True,
        "path": str(default_secret_manager.path),
        "unlocked": default_secret_manager.is_unlocked,
    }


def _secret_unlock(passphrase: str) -> Dict[str, Any]:
    return {"unlocked": default_secret_manager.unlock(passphrase)}


def _secret_lock() -> Dict[str, Any]:
    default_secret_manager.lock()
    return {"unlocked": default_secret_manager.is_unlocked}


def _secret_set(name: str, value: str) -> Dict[str, Any]:
    default_secret_manager.set(name, value)
    return {"name": name, "saved": True}


def _secret_remove(name: str) -> Dict[str, Any]:
    return {"name": name, "removed": default_secret_manager.remove(name)}


def _secret_list() -> List[str]:
    return default_secret_manager.list_names()


def _secret_status() -> Dict[str, Any]:
    return {
        "path": str(default_secret_manager.path),
        "initialized": default_secret_manager.is_initialized,
        "unlocked": default_secret_manager.is_unlocked,
    }


def _profiler_stats_as_dicts(limit: Optional[int] = None) -> List[dict]:
    """Executor adapter: dump profiler stats as JSON-friendly dicts."""
    rows = default_profiler.stats()
    if limit is not None:
        rows = rows[: max(0, int(limit))]
    return [row.to_dict() for row in rows]


def _profiler_hot_spots_as_dicts(limit: int = 10) -> List[dict]:
    """Executor adapter: top N actions by total time, as dicts."""
    return [row.to_dict() for row in default_profiler.hot_spots(int(limit))]


def _profiler_enable() -> Dict[str, Any]:
    default_profiler.enable()
    return {"enabled": default_profiler.enabled}


def _profiler_disable() -> Dict[str, Any]:
    default_profiler.disable()
    return {"enabled": default_profiler.enabled}


def _profiler_reset() -> Dict[str, Any]:
    default_profiler.reset()
    return {"reset": True}


def _history_list_as_dicts(limit: int = 100,
                           source_type: Optional[str] = None) -> List[dict]:
    """Executor adapter: list run history as plain dicts (JSON-friendly)."""
    rows = default_history_store.list_runs(
        limit=int(limit), source_type=source_type,
    )
    return [
        {
            "id": r.id, "source_type": r.source_type,
            "source_id": r.source_id, "script_path": r.script_path,
            "started_at": r.started_at, "finished_at": r.finished_at,
            "status": r.status, "error_text": r.error_text,
            "duration_seconds": r.duration_seconds,
        }
        for r in rows
    ]


_EXECUTOR_METRIC_CACHE: Dict[str, Any] = {}


def _executor_metrics():
    """Lazily register the action-executor Counter + Histogram (Phase 10.1)."""
    if "calls" in _EXECUTOR_METRIC_CACHE:
        return _EXECUTOR_METRIC_CACHE
    from je_auto_control.utils.observability import (
        Counter, Histogram, default_registry,
    )
    registry = default_registry()
    _EXECUTOR_METRIC_CACHE["calls"] = registry.register(Counter(
        "autocontrol_action_calls_total",
        "Number of AC_* actions executed, partitioned by name + outcome.",
        label_names=("action", "outcome"),
    ))
    _EXECUTOR_METRIC_CACHE["duration"] = registry.register(Histogram(
        "autocontrol_action_duration_seconds",
        "Wall-clock duration of each AC_* action call.",
        label_names=("action",),
    ))
    return _EXECUTOR_METRIC_CACHE


def _observe_executor_metrics(action: str, started_at: float,
                              *, error: Optional[BaseException]) -> None:
    """Emit Counter + Histogram samples for one action execution."""
    import time as _time
    try:
        metrics = _executor_metrics()
    except (ImportError, ValueError, RuntimeError):
        return
    duration = max(0.0, _time.monotonic() - started_at)
    outcome = "error" if error is not None else "ok"
    try:
        metrics["calls"].inc(labels={"action": action, "outcome": outcome})
        metrics["duration"].observe(duration, labels={"action": action})
    except ValueError:
        # Defensive: if the label set drifts (e.g. tests reset the registry)
        # we'd rather lose a sample than crash the executor.
        pass


def _human_move(x: int, y: int, duration_s: float = 0.4, curve: float = 0.2,
                overshoot: float = 0.0, jitter: float = 1.0,
                seed: Optional[int] = None) -> Dict[str, Any]:
    """Executor adapter: move the mouse to (x, y) along a human-like path."""
    from je_auto_control.utils.humanize.motion import (
        HumanizedMotion, move_mouse_humanized,
    )
    motion = HumanizedMotion(curve=float(curve), overshoot=float(overshoot),
                             jitter=float(jitter), seed=seed)
    path = move_mouse_humanized(int(x), int(y),
                                duration_s=float(duration_s), motion=motion)
    return {"x": int(x), "y": int(y), "waypoints": len(path)}


def _human_type(text: str, base_delay: float = 0.05, jitter: float = 0.04,
                pause_chance: float = 0.0,
                seed: Optional[int] = None) -> Dict[str, Any]:
    """Executor adapter: type text with humanized inter-key delays."""
    from je_auto_control.utils.humanize.typing import type_text_humanized
    delays = type_text_humanized(
        str(text), base_delay=float(base_delay), jitter=float(jitter),
        pause_chance=float(pause_chance), seed=seed,
    )
    return {"chars": len(str(text)), "total_delay_s": round(sum(delays), 3)}


def _sign_action_file(path: str, key: Optional[str] = None) -> Dict[str, Any]:
    """Executor adapter: write an HMAC-SHA256 signature sidecar for a file."""
    from je_auto_control.utils.action_signing import sign_action_file
    return {"signature_path": sign_action_file(path, key)}


def _verify_action_file(path: str, key: Optional[str] = None,
                        raise_on_fail: bool = False) -> Dict[str, Any]:
    """Executor adapter: verify an action file against its signature sidecar."""
    from je_auto_control.utils.action_signing import verify_action_file
    return verify_action_file(
        path, key, raise_on_fail=bool(raise_on_fail),
    ).to_dict()


def _encrypt_action_file(path: str, key: Optional[str] = None) -> Dict[str, Any]:
    """Executor adapter: Fernet-encrypt an action file to <path>.enc."""
    from je_auto_control.utils.action_signing import encrypt_action_file
    return {"encrypted_path": encrypt_action_file(path, key)}


def _decrypt_action_file(enc_path: str, key: Optional[str] = None,
                         output_path: Optional[str] = None) -> Dict[str, Any]:
    """Executor adapter: decrypt a Fernet-encrypted action file."""
    from je_auto_control.utils.action_signing import decrypt_action_file
    return {"output_path": decrypt_action_file(enc_path, key, output_path)}


def _annotate_screenshot(source: str,
                         annotations: Union[List[Dict[str, Any]], str],
                         output_path: str) -> Dict[str, Any]:
    """Executor adapter: draw annotations onto a screenshot and save it.

    ``annotations`` may be a list of annotation dicts, or a JSON string of
    the same (so the visual builder can pass it through a text field).
    """
    import json
    from je_auto_control.utils.annotate import annotate_screenshot
    if isinstance(annotations, str):
        annotations = json.loads(annotations) if annotations.strip() else []
    return {"output_path": annotate_screenshot(
        source, annotations, output_path)}


def _notify(title: str, message: str = "") -> Dict[str, Any]:
    """Executor adapter: show a cross-platform desktop notification."""
    from je_auto_control.utils.notify import notify
    return notify(str(title), str(message)).to_dict()


def _move_to_trash(path: str) -> Dict[str, Any]:
    """Executor adapter: move a file to the OS recycle bin (recoverable)."""
    from je_auto_control.utils.trash import move_to_trash
    return {"trashed": move_to_trash(path)}


def _read_qr(region: Optional[Union[List[int], str]] = None) -> Dict[str, Any]:
    """Executor adapter: decode QR codes in a screen region.

    ``region`` is ``[x1, y1, x2, y2]`` (or a JSON string for the builder);
    omit it to scan the whole screen.
    """
    import json
    import os
    import tempfile
    from je_auto_control.utils.qr import read_qr_codes
    from je_auto_control.wrapper.auto_control_screen import screenshot
    if isinstance(region, str):
        region = json.loads(region) if region.strip() else None
    handle, tmp = tempfile.mkstemp(prefix="qr_", suffix=".png")
    os.close(handle)
    try:
        screenshot(tmp, screen_region=region)
        return {"codes": read_qr_codes(tmp)}
    finally:
        try:
            os.unlink(tmp)
        except OSError:
            pass


def _scroll_to_find(target: str, kind: str = "image", direction: str = "down",
                    max_scrolls: int = 10,
                    scroll_amount: int = 3) -> Dict[str, Any]:
    """Executor adapter: scroll until a target image / text is visible."""
    from je_auto_control.utils.scroll_find import scroll_until_visible
    return scroll_until_visible(
        target, kind=kind, direction=direction,
        max_scrolls=int(max_scrolls), scroll_amount=int(scroll_amount),
    )


def _capture_window(title: str, output_path: str) -> Dict[str, Any]:
    """Executor adapter: screenshot the window matching ``title``."""
    from je_auto_control.utils.window_capture import capture_window
    return {"output_path": capture_window(title, output_path)}


def _snap_window(title: str, position: str = "left") -> Dict[str, Any]:
    """Executor adapter: snap a window to a screen region."""
    from je_auto_control.utils.window_capture import snap_window
    return {"moved": snap_window(title, position)}


def _save_window_layout(path: Optional[str] = None) -> Dict[str, Any]:
    """Executor adapter: snapshot every window's geometry (optionally to file)."""
    from je_auto_control.utils.window_capture import save_window_layout
    layout = save_window_layout(path)
    return {"count": len(layout), "path": path, "layout": layout}


def _restore_window_layout(layout: Union[List[Dict[str, Any]], str]
                           ) -> Dict[str, Any]:
    """Executor adapter: move windows back to a saved layout (list or path)."""
    from je_auto_control.utils.window_capture import restore_window_layout
    return {"restored": restore_window_layout(layout)}


def _region_color_stats(region: Optional[Union[List[int], str]] = None,
                        buckets: int = 8) -> Dict[str, Any]:
    """Executor adapter: average + dominant colour of a screen region.

    ``region`` is ``[x1, y1, x2, y2]`` (or a JSON string of it for the
    visual builder); omit it to analyse the whole screen.
    """
    import json
    import os
    import tempfile
    from je_auto_control.utils.color_stats import region_color_stats
    from je_auto_control.wrapper.auto_control_screen import screenshot
    if isinstance(region, str):
        region = json.loads(region) if region.strip() else None
    handle, tmp = tempfile.mkstemp(prefix="colorstats_", suffix=".png")
    os.close(handle)
    try:
        screenshot(tmp, screen_region=region)
        return region_color_stats(tmp, buckets=int(buckets)).to_dict()
    finally:
        try:
            os.unlink(tmp)
        except OSError:
            pass


def _generate_code(source: Any, output: Optional[str] = None,
                   target: str = "pytest", name: str = "recorded_flow",
                   style: str = "calls") -> str:
    """Render an action list/file as code, optionally writing a file."""
    from je_auto_control.utils.codegen.codegen import (
        generate_code, generate_code_file,
    )
    if output:
        return generate_code_file(source, output, target=target,
                                  name=name, style=style)
    actions = source if isinstance(source, list) else read_action_json(source)
    return generate_code(actions, target=target, name=name, style=style)


def _send_email(message: Any, smtp: Any) -> Dict[str, Any]:
    """Adapter: send an email via SMTP (message/smtp config dicts)."""
    from je_auto_control.utils.email_send.email_sender import send_email
    return send_email(message, smtp)


def _assert_pdf_text(path: str, text: str, present: bool = True,
                     page: Any = None, case_sensitive: bool = True,
                     raise_on_fail: bool = True) -> Dict[str, Any]:
    """Adapter: assert text is present/absent in a PDF document."""
    from je_auto_control.utils.pdf.pdf_reader import assert_pdf_text
    return assert_pdf_text(path, text, present=bool(present), page=page,
                           case_sensitive=bool(case_sensitive),
                           raise_on_fail=bool(raise_on_fail))


def _take_golden(path: str, region: Optional[List[int]] = None) -> str:
    """Adapter: capture and save a golden/baseline image."""
    from je_auto_control.utils.visual_regression import take_golden
    return str(take_golden(path, region=region))


def _assert_visual(golden_path: str, region: Optional[List[int]] = None,
                   tolerance: float = 0.0, per_pixel_threshold: int = 16,
                   diff_path: Optional[str] = None,
                   create_if_missing: bool = True,
                   raise_on_fail: bool = True) -> Dict[str, Any]:
    """Adapter: compare the screen to a golden image (first run creates it)."""
    import os
    from je_auto_control.utils.exception.exceptions import (
        AutoControlAssertionException,
    )
    from je_auto_control.utils.visual_regression import (
        compare_to_golden, take_golden,
    )
    if create_if_missing and not os.path.exists(
            os.path.expanduser(str(golden_path))):
        take_golden(golden_path, region=region)
        return {"created": True, "matched": True, "golden": str(golden_path)}
    result = compare_to_golden(
        golden_path, region=region, tolerance=float(tolerance),
        per_pixel_threshold=int(per_pixel_threshold))
    if diff_path and result.diff_image is not None:
        result.write_diff(diff_path)
    data = {"matched": result.matched, "diff_pct": result.diff_pct,
            "differing_pixels": result.differing_pixels,
            "total_pixels": result.total_pixels,
            "tolerance_pct": result.tolerance_pct}
    if not result.matched and raise_on_fail:
        raise AutoControlAssertionException(result.summary)
    return data


def _run_state_machine(spec: Any) -> Dict[str, Any]:
    """Adapter: run a finite-state-machine spec through the executor."""
    from je_auto_control.utils.state_machine import run_state_machine
    return run_state_machine(spec)


def _control_get_value(name: Optional[str] = None, role: Optional[str] = None,
                       app_name: Optional[str] = None,
                       automation_id: Optional[str] = None) -> Optional[str]:
    """Adapter: read a native control's value via the accessibility backend."""
    from je_auto_control.utils.accessibility import control_get_value
    return control_get_value(name=name, role=role, app_name=app_name,
                             automation_id=automation_id)


def _control_set_value(value: str, name: Optional[str] = None,
                       role: Optional[str] = None, app_name: Optional[str] = None,
                       automation_id: Optional[str] = None) -> bool:
    """Adapter: set a native control's value via the accessibility backend."""
    from je_auto_control.utils.accessibility import control_set_value
    return control_set_value(value, name=name, role=role, app_name=app_name,
                             automation_id=automation_id)


def _control_invoke(name: Optional[str] = None, role: Optional[str] = None,
                    app_name: Optional[str] = None,
                    automation_id: Optional[str] = None) -> bool:
    """Adapter: invoke a native control (e.g. press a button)."""
    from je_auto_control.utils.accessibility import control_invoke
    return control_invoke(name=name, role=role, app_name=app_name,
                          automation_id=automation_id)


def _control_toggle(name: Optional[str] = None, role: Optional[str] = None,
                    app_name: Optional[str] = None,
                    automation_id: Optional[str] = None) -> bool:
    """Adapter: toggle a native control (e.g. a checkbox)."""
    from je_auto_control.utils.accessibility import control_toggle
    return control_toggle(name=name, role=role, app_name=app_name,
                          automation_id=automation_id)


def _read_table(name: Optional[str] = None, role: Optional[str] = None,
                app_name: Optional[str] = None,
                automation_id: Optional[str] = None) -> List[List[str]]:
    """Adapter: read a grid/table/list control as rows of cell strings."""
    from je_auto_control.utils.accessibility import read_control_table
    return read_control_table(name=name, role=role, app_name=app_name,
                              automation_id=automation_id)


def _watchdog_add(title: str, action: str = "close",
                  case_sensitive: bool = False,
                  name: Optional[str] = None) -> Dict[str, Any]:
    """Adapter: register a popup-dismissal rule on the default watchdog."""
    from je_auto_control.utils.watchdog import default_popup_watchdog
    default_popup_watchdog.add_window_rule(
        title, action=str(action), case_sensitive=bool(case_sensitive),
        name=name)
    return {"rules": default_popup_watchdog.rule_names()}


def _watchdog_start() -> Dict[str, Any]:
    """Adapter: start the background popup watchdog."""
    from je_auto_control.utils.watchdog import default_popup_watchdog
    default_popup_watchdog.start()
    return {"running": True}


def _watchdog_stop() -> Dict[str, Any]:
    """Adapter: stop the background popup watchdog."""
    from je_auto_control.utils.watchdog import default_popup_watchdog
    default_popup_watchdog.stop()
    return {"running": False}


def _watchdog_list() -> Dict[str, Any]:
    """Adapter: report the watchdog's rules, run state and dismissals."""
    from je_auto_control.utils.watchdog import default_popup_watchdog
    w = default_popup_watchdog
    return {"running": w.running, "rules": w.rule_names(), "hits": w.hits}


def _handle_file_dialog(path: str, action: str = "open",
                        window_title: Optional[str] = None,
                        timeout_s: float = 10.0,
                        confirm_key: str = "enter") -> Dict[str, Any]:
    """Adapter: wait for a native file dialog, type the path, confirm."""
    from je_auto_control.utils.file_dialog import handle_file_dialog
    return handle_file_dialog(path, action=action, window_title=window_title,
                              timeout_s=float(timeout_s),
                              confirm_key=confirm_key)


def _assert_session_active() -> Dict[str, Any]:
    """Adapter: raise unless the session is interactive (not locked)."""
    from je_auto_control.utils.session_guard import ensure_interactive_session
    return {"interactive": ensure_interactive_session()}


def _queue(db: str, name: str):
    from je_auto_control.utils.work_queue import WorkQueue
    return WorkQueue(db, name)


def _queue_add(db: str, data: Any, reference: Optional[str] = None,
               name: str = "default") -> Dict[str, Any]:
    """Adapter: enqueue a work item (skips live duplicate references)."""
    return {"id": _queue(db, name).add(data, reference=reference)}


def _queue_next(db: str, name: str = "default") -> Optional[Dict[str, Any]]:
    """Adapter: atomically claim the next work item (or None)."""
    item = _queue(db, name).get_next()
    return None if item is None else {
        "id": item.id, "reference": item.reference, "data": item.data,
        "status": item.status, "retries": item.retries}


def _queue_complete(db: str, item_id: int, output: Any = None,
                    name: str = "default") -> Dict[str, Any]:
    """Adapter: mark a work item successful."""
    _queue(db, name).complete(int(item_id), output=output)
    return {"id": int(item_id), "status": "success"}


def _queue_fail(db: str, item_id: int, error: str,
                kind: str = "application", max_retries: int = 3,
                name: str = "default") -> Dict[str, Any]:
    """Adapter: fail a work item (application errors retry, business don't)."""
    status = _queue(db, name).fail(int(item_id), str(error), kind=str(kind),
                                   max_retries=int(max_retries))
    return {"id": int(item_id), "status": status}


def _queue_stats(db: str, name: str = "default") -> Dict[str, int]:
    """Adapter: return per-status counts for a work queue."""
    return _queue(db, name).stats()


def _generate_data(schema: Dict[str, Any], count: int = 10,
                   path: Optional[str] = None, fmt: Optional[str] = None,
                   seed: Optional[int] = None) -> Dict[str, Any]:
    """Adapter: generate synthetic rows; write to ``path`` when given."""
    from je_auto_control.utils.test_data import generate_rows, write_dataset
    rows = generate_rows(schema, int(count), seed=seed)
    if path:
        return {"path": write_dataset(rows, path, fmt), "count": len(rows)}
    return {"rows": rows, "count": len(rows)}


def _mcp_manifest(path: Optional[str] = None,
                  include_tools: bool = False) -> Dict[str, Any]:
    """Adapter: build (or write) the MCP registry server.json manifest."""
    from je_auto_control.utils.mcp_registry import (
        build_server_manifest, write_server_manifest)
    if path:
        return {"path": write_server_manifest(
            path, include_tools=bool(include_tools))}
    return {"manifest": build_server_manifest(
        include_tools=bool(include_tools))}


def _rank_tests(flows: List[str], history_path: Optional[str] = None,
                window: int = 10) -> Dict[str, Any]:
    """Adapter: score flows by risk (riskiest first)."""
    from je_auto_control.utils.test_select import rank_flows
    return {"ranked": rank_flows(flows, history_path=history_path,
                                 window=int(window))}


def _select_tests(flows: List[str], k: Optional[int] = None,
                  threshold: Optional[float] = None,
                  history_path: Optional[str] = None,
                  window: int = 10) -> Dict[str, Any]:
    """Adapter: pick the riskiest flows to run (top-k / threshold)."""
    from je_auto_control.utils.test_select import select_flows
    return {"selected": select_flows(
        flows, k=k, threshold=threshold, history_path=history_path,
        window=int(window))}


def _element_repo(path: str):
    from je_auto_control.utils.element_repository import ElementRepository
    return ElementRepository(path)


def _element_save(path: str, key: str, name: Optional[str] = None,
                  role: Optional[str] = None,
                  app_name: Optional[str] = None) -> Dict[str, Any]:
    """Adapter: save a named native-UI locator (object repository)."""
    return {"locator": _element_repo(path).save(
        key, name=name, role=role, app_name=app_name)}


def _element_find(path: str, key: str) -> Dict[str, Any]:
    """Adapter: resolve a saved locator to a live element summary."""
    return _element_repo(path).find_info(key)


def _element_click(path: str, key: str) -> Dict[str, Any]:
    """Adapter: click the element behind a saved locator."""
    return {"clicked": _element_repo(path).click(key)}


def _element_remove(path: str, key: str) -> Dict[str, Any]:
    """Adapter: delete a saved locator."""
    return {"removed": _element_repo(path).remove(key)}


def _element_list(path: str) -> Dict[str, Any]:
    """Adapter: list saved locator names."""
    return {"keys": _element_repo(path).keys()}


def _debug_trace(actions: List[Any], dry_run: bool = False) -> Dict[str, Any]:
    """Adapter: run an action list and return a per-step trace."""
    from je_auto_control.utils.flow_debugger import trace_actions
    return {"trace": trace_actions(actions, dry_run=bool(dry_run))}


def _skill_lib(path: str):
    from je_auto_control.utils.skill_library import SkillLibrary
    return SkillLibrary(path)


def _skill_save(path: str, name: str, actions: List[Any],
                description: str = "",
                tags: Optional[List[str]] = None) -> Dict[str, Any]:
    """Adapter: save a reusable action sequence (skill)."""
    skill = _skill_lib(path).save(name, actions, description=description,
                                  tags=tags)
    return {"name": skill.name, "tags": skill.tags}


def _skill_run(path: str, name: str) -> Dict[str, Any]:
    """Adapter: execute a stored skill's actions."""
    return {"record": _skill_lib(path).run(name)}


def _skill_list(path: str) -> Dict[str, Any]:
    """Adapter: list saved skill names."""
    return {"names": _skill_lib(path).names()}


def _skill_remove(path: str, name: str) -> Dict[str, Any]:
    """Adapter: delete a saved skill."""
    return {"removed": _skill_lib(path).remove(name)}


def _skill_search(path: str, query: str) -> Dict[str, Any]:
    """Adapter: search skills by name/description/tags."""
    return {"names": [s.name for s in _skill_lib(path).search(query)]}


def _guard_text(text: str, threshold: int = 2) -> Dict[str, Any]:
    """Adapter: assess text for prompt-injection patterns."""
    from je_auto_control.utils.guardrail import assess_text
    return assess_text(text, threshold=int(threshold))


def _agent_card(path: Optional[str] = None) -> Dict[str, Any]:
    """Adapter: build (or write) the A2A agent card."""
    from je_auto_control.utils.a2a import build_agent_card, write_agent_card
    if path:
        return {"path": write_agent_card(path)}
    return {"card": build_agent_card()}


def _read_workbook(path: str, sheet: str = "") -> Dict[str, Any]:
    """Adapter: read an .xlsx worksheet into rows."""
    from je_auto_control.utils.office import read_workbook
    return {"rows": read_workbook(path, sheet=sheet)}


def _write_workbook(path: str, rows: List[Dict[str, Any]],
                    sheet: str = "Sheet1") -> Dict[str, Any]:
    """Adapter: write rows to an .xlsx file."""
    from je_auto_control.utils.office import write_workbook
    return {"path": write_workbook(path, rows, sheet=sheet)}


def _read_document(path: str) -> Dict[str, Any]:
    """Adapter: read a .docx file's paragraphs."""
    from je_auto_control.utils.office import read_document
    return read_document(path)


def _write_document(path: str, paragraphs: List[str]) -> Dict[str, Any]:
    """Adapter: write paragraphs to a .docx file."""
    from je_auto_control.utils.office import write_document
    return {"path": write_document(path, paragraphs)}


def _read_presentation(path: str) -> Dict[str, Any]:
    """Adapter: read a .pptx file's per-slide text."""
    from je_auto_control.utils.office import read_presentation
    return read_presentation(path)


def _write_presentation(path: str, slides: List[Any]) -> Dict[str, Any]:
    """Adapter: write slides to a .pptx file."""
    from je_auto_control.utils.office import write_presentation
    return {"path": write_presentation(path, slides)}


def _memory(db: str):
    from je_auto_control.utils.agent_memory import AgentMemory
    return AgentMemory(db)


def _memory_remember(db: str, goal: str, steps: Optional[List[Any]] = None,
                     outcome: str = "",
                     tags: Optional[List[str]] = None) -> Dict[str, Any]:
    """Adapter: store an agent episode (goal/trajectory/outcome)."""
    return {"id": _memory(db).remember(goal, steps=steps, outcome=outcome,
                                       tags=tags)}


def _memory_recall(db: str, query: str, limit: int = 5) -> Dict[str, Any]:
    """Adapter: recall episodes most relevant to a query."""
    episodes = _memory(db).recall(query, limit=int(limit))
    return {"episodes": [_episode_to_dict(ep) for ep in episodes]}


def _memory_recent(db: str, limit: int = 10) -> Dict[str, Any]:
    """Adapter: list the most recent episodes."""
    episodes = _memory(db).recent(limit=int(limit))
    return {"episodes": [_episode_to_dict(ep) for ep in episodes]}


def _memory_forget(db: str, episode_id: int) -> Dict[str, Any]:
    """Adapter: delete an episode."""
    return {"removed": _memory(db).forget(int(episode_id))}


def _memory_stats(db: str) -> Dict[str, int]:
    """Adapter: episode count for a memory store."""
    return _memory(db).stats()


def _episode_to_dict(episode: Any) -> Dict[str, Any]:
    return {"id": episode.id, "goal": episode.goal, "steps": episode.steps,
            "outcome": episode.outcome, "tags": episode.tags,
            "score": episode.score}


def _seed_everything(seed: int = 0) -> Dict[str, Any]:
    """Adapter: seed all RNG run-wide for reproducible runs."""
    from je_auto_control.utils.deterministic import seed_everything
    return {"seed": seed_everything(int(seed))}


def _observe_handler(actions: List[Any]) -> Callable[[str, Any], None]:
    """Build an observer callback that runs an action list on each event."""
    def handler(_event: str, _value: Any) -> None:
        if actions:
            executor.execute_action(list(actions))
    return handler


def _observe_predicate(kind: str, params: Dict[str, Any]):
    from je_auto_control.utils.observer import (
        image_predicate, pixel_predicate, text_predicate)
    builders = {
        "image": lambda: image_predicate(params.get("image", ""),
                                         params.get("threshold", 0.8)),
        "text": lambda: text_predicate(params.get("text", "")),
        "pixel": lambda: pixel_predicate(int(params.get("x", 0)),
                                         int(params.get("y", 0))),
    }
    if kind not in builders:
        raise AutoControlActionException(f"unknown observe kind: {kind!r}")
    return builders[kind]()


def _observe_add(name: str, kind: str = "image", event: str = "appear",
                 actions: Optional[List[Any]] = None,
                 **params: Any) -> Dict[str, Any]:
    """Adapter: watch image/text/pixel; run ``actions`` on the event."""
    from je_auto_control.utils.observer import default_observer
    default_observer.add(name, _observe_predicate(kind, params),
                         _observe_handler(actions or []), events=(event,))
    return {"name": name, "kind": kind, "event": event}


def _observe_remove(name: str) -> Dict[str, Any]:
    """Adapter: remove a registered watch."""
    from je_auto_control.utils.observer import default_observer
    return {"removed": default_observer.remove(name)}


def _observe_list() -> Dict[str, Any]:
    """Adapter: list registered watch names."""
    from je_auto_control.utils.observer import default_observer
    return {"names": default_observer.names()}


def _observe_poll() -> Dict[str, Any]:
    """Adapter: evaluate all watches once; return fired events."""
    from je_auto_control.utils.observer import default_observer
    return {"fired": default_observer.poll_once()}


def _observe_start() -> Dict[str, Any]:
    """Adapter: start the background observer thread."""
    from je_auto_control.utils.observer import default_observer
    default_observer.start()
    return {"running": default_observer.running}


def _observe_stop() -> Dict[str, Any]:
    """Adapter: stop the background observer thread."""
    from je_auto_control.utils.observer import default_observer
    default_observer.stop()
    return {"running": default_observer.running}


def _generate_sbom(path: Optional[str] = None,
                   root: str = "je_auto_control") -> Dict[str, Any]:
    """Adapter: build (or write) a CycloneDX SBOM for the project."""
    from je_auto_control.utils.sbom import build_sbom, write_sbom
    root_arg = root or None
    if path:
        return {"path": write_sbom(path, root_arg)}
    return {"sbom": build_sbom(root_arg)}


def _shard_suite(flows: List[str], shards: int = 2,
                 history_path: Optional[str] = None,
                 window: int = 20) -> Dict[str, Any]:
    """Adapter: balance flows into duration-aware shards."""
    from je_auto_control.utils.test_shard import shard_flows
    return {"shards": shard_flows(flows, int(shards),
                                  history_path=history_path,
                                  window=int(window))}


def _merge_results(reports: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Adapter: merge per-shard report dicts into one report."""
    from je_auto_control.utils.test_shard import merge_results
    return merge_results(reports)


def _validate_rows(rows: List[Dict[str, Any]],
                   schema: Dict[str, Any]) -> Dict[str, Any]:
    """Adapter: validate rows against a declarative schema."""
    from je_auto_control.utils.data_quality import validate_rows
    return validate_rows(rows, schema)


def _extract_fields(text: str, fields: Optional[List[str]] = None,
                    patterns: Optional[Dict[str, str]] = None
                    ) -> Dict[str, Any]:
    """Adapter: extract structured fields from free text."""
    from je_auto_control.utils.data_quality import extract_fields
    return {"fields": extract_fields(text, fields=fields, patterns=patterns)}


def _mask_rows(rows: List[Dict[str, Any]],
               rules: Dict[str, str]) -> Dict[str, Any]:
    """Adapter: mask sensitive columns in rows."""
    from je_auto_control.utils.data_quality import mask_rows
    return {"rows": mask_rows(rows, rules)}


def _pseudo_localize(text: Optional[str] = None,
                     mapping: Optional[Dict[str, Any]] = None,
                     expansion: float = 0.4) -> Dict[str, Any]:
    """Adapter: pseudo-localize a string or a whole catalog mapping."""
    from je_auto_control.utils.i18n_test import (
        pseudo_localize, pseudo_localize_catalog)
    if mapping is not None:
        return {"catalog": pseudo_localize_catalog(
            mapping, expansion=float(expansion))}
    return {"text": pseudo_localize(text or "", expansion=float(expansion))}


def _check_overflow(elements: Optional[List[Any]] = None,
                    avg_char_px: float = 7.0,
                    app_name: Optional[str] = None) -> Dict[str, Any]:
    """Adapter: flag text wider than its widget (live a11y unless given)."""
    from je_auto_control.utils.i18n_test import check_overflow
    items = elements
    if items is None:
        from je_auto_control.utils.accessibility.accessibility_api import (
            list_accessibility_elements)
        items = list_accessibility_elements(app_name=app_name)
    return {"issues": check_overflow(items, avg_char_px=float(avg_char_px))}


def _check_catalog(base: Dict[str, Any],
                   target: Dict[str, Any]) -> Dict[str, Any]:
    """Adapter: diff a translation catalog against the base locale."""
    from je_auto_control.utils.i18n_test import check_catalog
    return check_catalog(base, target)


def _run_resumable(actions: List[Any], run_id: str, db: str,
                   variables: Optional[Dict[str, Any]] = None
                   ) -> Dict[str, Any]:
    """Adapter: run actions with checkpoint/resume keyed by run_id."""
    from je_auto_control.utils.checkpoint import CheckpointStore, run_resumable
    return run_resumable(actions, run_id=run_id,
                         store=CheckpointStore(db), variables=variables)


def _checkpoint_status(run_id: str, db: str) -> Dict[str, Any]:
    """Adapter: return the saved checkpoint for a run (or null)."""
    from je_auto_control.utils.checkpoint import CheckpointStore
    checkpoint = CheckpointStore(db).load(run_id)
    if checkpoint is None:
        return {"checkpoint": None}
    return {"checkpoint": {"run_id": checkpoint.run_id,
                           "step_index": checkpoint.step_index,
                           "variables": checkpoint.variables}}


def _checkpoint_clear(run_id: str, db: str) -> Dict[str, Any]:
    """Adapter: delete a run's checkpoint."""
    from je_auto_control.utils.checkpoint import CheckpointStore
    return {"cleared": CheckpointStore(db).clear(run_id)}


def _mark_screen(app_name: Optional[str] = None,
                 render_path: Optional[str] = None) -> Dict[str, Any]:
    """Adapter: number live UI elements (Set-of-Marks) for VLM grounding."""
    from je_auto_control.utils.set_of_marks import mark_screen
    return mark_screen(app_name=app_name, render_path=render_path)


def _mark_click(mark_id: int) -> Dict[str, Any]:
    """Adapter: click the element behind a numbered mark."""
    from je_auto_control.utils.set_of_marks import mark_click
    return {"clicked": mark_click(int(mark_id))}


def _screen_snapshot(app_name: Optional[str] = None) -> Dict[str, Any]:
    """Adapter: snapshot the live a11y tree as a diff baseline."""
    from je_auto_control.utils.screen_state import snapshot_screen
    return {"snapshot": snapshot_screen(app_name=app_name)}


def _screen_diff(before: List[Dict[str, Any]],
                 after: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Adapter: semantic diff between two snapshots."""
    from je_auto_control.utils.screen_state import diff_snapshots
    return diff_snapshots(before, after)


def _screen_changed(app_name: Optional[str] = None) -> Dict[str, Any]:
    """Adapter: diff the live screen against the last snapshot baseline."""
    from je_auto_control.utils.screen_state import screen_changed
    return screen_changed(app_name=app_name)


def _describe_screen(app_name: Optional[str] = None) -> Dict[str, Any]:
    """Adapter: structured 'where am I' description of the live screen."""
    from je_auto_control.utils.screen_state import describe_screen
    return describe_screen(app_name=app_name)


def _replay_timeline(events: List[Dict[str, Any]],
                     speed: float = 1.0) -> Dict[str, Any]:
    """Adapter: replay timed input events at a speed multiplier."""
    from je_auto_control.utils.input_macro import replay_timeline
    return {"played": replay_timeline(events, speed=float(speed))}


def _input_sequence(steps: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Adapter: run a declarative input sequence (press/hold/repeat/...)."""
    from je_auto_control.utils.input_macro import run_sequence
    return {"log": run_sequence(steps)}


_CIRCUIT_BREAKERS: Dict[str, Any] = {}


def _circuit_call(name: str, actions: List[Any], threshold: int = 5,
                  reset_s: float = 30.0) -> Dict[str, Any]:
    """Adapter: run an action list through a named circuit breaker."""
    from je_auto_control.utils.resilience import CircuitBreaker
    breaker = _CIRCUIT_BREAKERS.setdefault(
        name, CircuitBreaker(int(threshold), float(reset_s)))
    record = breaker.call(
        lambda: executor.execute_action(list(actions), raise_on_error=True))
    return {"state": breaker.state, "record": record}


def _ci_annotations(annotations: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Adapter: emit GitHub Actions annotations from result dicts."""
    from je_auto_control.utils.ci_annotations import emit_annotations
    return {"lines": emit_annotations(annotations)}


def _clip_history_capture() -> Dict[str, Any]:
    """Adapter: capture the live clipboard into history."""
    from je_auto_control.utils.clipboard_history import (
        default_clipboard_history)
    return {"added": default_clipboard_history.capture_once()}


def _clip_history_list() -> Dict[str, Any]:
    """Adapter: list clipboard history (newest first)."""
    from je_auto_control.utils.clipboard_history import (
        default_clipboard_history)
    return {"history": default_clipboard_history.snapshot()}


def _clip_history_search(query: str) -> Dict[str, Any]:
    """Adapter: search clipboard history."""
    from je_auto_control.utils.clipboard_history import (
        default_clipboard_history)
    return {"matches": default_clipboard_history.search(query)}


def _clip_history_start() -> Dict[str, Any]:
    """Adapter: start the background clipboard-history poller."""
    from je_auto_control.utils.clipboard_history import (
        default_clipboard_history)
    default_clipboard_history.start()
    return {"running": default_clipboard_history.running}


def _clip_history_stop() -> Dict[str, Any]:
    """Adapter: stop the background clipboard-history poller."""
    from je_auto_control.utils.clipboard_history import (
        default_clipboard_history)
    default_clipboard_history.stop()
    return {"running": default_clipboard_history.running}


def _heal_stats(limit: int = 200) -> Dict[str, Any]:
    """Adapter: aggregate the self-heal log into metrics."""
    from je_auto_control.utils.heal_analytics import analyze_heal_log
    return analyze_heal_log(limit=int(limit))


def _scan_secrets(data: Any) -> Dict[str, Any]:
    """Adapter: scan JSON/data for hardcoded secrets."""
    from je_auto_control.utils.secrets_scan import scan_secrets
    return {"findings": scan_secrets(data)}


def _generate_sop(actions: List[Any], title: str = "Automation Procedure",
                  path: Optional[str] = None) -> Dict[str, Any]:
    """Adapter: build (or write) a step-by-step SOP from an action list."""
    from je_auto_control.utils.process_doc import generate_sop, write_sop
    if path:
        return {"path": write_sop(actions, path, title=title)}
    return generate_sop(actions, title=title)


class Executor:
    """
    Executor
    指令執行器
    - 提供 event_dict 對應字串名稱到函式
    - 支援滑鼠、鍵盤、螢幕、影像辨識、報告生成等功能
    - 可執行 action list 或 action file
    - 支援流程控制指令 (AC_loop, AC_if_image_found 等)
    """

    # Args keys that hold nested action lists; runtime interpolation must
    # leave them untouched so each iteration re-reads current variable state.
    _DEFERRED_ARG_KEYS: frozenset = frozenset(
        {"body", "then", "else", "branches"})

    def __init__(self):
        self._block_commands = BLOCK_COMMANDS
        self.variables = VariableScope()
        # Named, parameterised macros registered via AC_define_macro.
        self.macros: Dict[str, Any] = {}
        # 事件字典，對應字串名稱到函式
        self.event_dict: dict = {
            # Mouse 滑鼠相關
            **MOUSE_BUTTON_COMMANDS,
            "AC_click_mouse": click_mouse,
            "AC_get_mouse_table": get_mouse_table,
            "AC_get_mouse_position": get_mouse_position,
            "AC_press_mouse": press_mouse,
            "AC_release_mouse": release_mouse,
            "AC_mouse_scroll": mouse_scroll,
            "AC_set_mouse_position": set_mouse_position,
            "AC_human_move": _human_move,
            "AC_human_type": _human_type,

            # Keyboard 鍵盤相關
            "AC_get_keyboard_keys_table": get_keyboard_keys_table,
            "AC_type_keyboard": type_keyboard,
            "AC_press_keyboard_key": press_keyboard_key,
            "AC_release_keyboard_key": release_keyboard_key,
            "AC_check_key_is_press": check_key_is_press,
            "AC_write": write,
            "AC_hotkey": hotkey,

            # Image 影像辨識
            "AC_locate_all_image": locate_all_image,
            "AC_locate_image_center": locate_image_center,
            "AC_locate_and_click": locate_and_click,

            # Screen 螢幕相關
            "AC_screen_size": screen_size,
            "AC_screenshot": screenshot,

            # Test record 測試紀錄
            "AC_set_record_enable": test_record_instance.set_record_enable,

            # Report 報告生成
            "AC_generate_html": generate_html,
            "AC_generate_json": generate_json,
            "AC_generate_xml": generate_xml,
            "AC_generate_html_report": generate_html_report,
            "AC_generate_json_report": generate_json_report,
            "AC_generate_xml_report": generate_xml_report,
            "AC_generate_code": _generate_code,
            "AC_send_email": _send_email,
            "AC_assert_pdf_text": _assert_pdf_text,
            "AC_take_golden": _take_golden,
            "AC_assert_visual": _assert_visual,
            "AC_run_state_machine": _run_state_machine,
            "AC_http_request": http_request,

            # Record 錄製
            "AC_record": record,
            "AC_stop_record": stop_record,

            # Executor 執行器
            "AC_execute_action": self.execute_action,
            "AC_execute_files": self.execute_files,
            "AC_add_package_to_executor": package_manager.add_package_to_executor,
            "AC_add_package_to_callback_executor": package_manager.add_package_to_callback_executor,

            # Project 專案
            "AC_create_project": create_project_dir,

            # Shell
            "AC_shell_command": default_shell_manager.exec_shell,

            # Process
            "AC_execute_process": start_exe,

            # OCR
            "AC_locate_text": ocr_locate_text_center,
            "AC_wait_text": ocr_wait_for_text,
            "AC_click_text": ocr_click_text,
            "AC_read_text_in_region": _ocr_read_region_as_dicts,
            "AC_find_text_regex": _ocr_find_regex_as_dicts,

            # Window management
            "AC_list_windows": list_windows,
            "AC_focus_window": focus_window,
            "AC_wait_window": wait_for_window,
            "AC_close_window": close_window_by_title,

            # Clipboard
            "AC_clipboard_get": get_clipboard,
            "AC_clipboard_set": set_clipboard,

            # Run history
            "AC_history_list": _history_list_as_dicts,
            "AC_history_clear": default_history_store.clear,

            # Profiler
            "AC_profiler_enable": _profiler_enable,
            "AC_profiler_disable": _profiler_disable,
            "AC_profiler_reset": _profiler_reset,
            "AC_profiler_stats": _profiler_stats_as_dicts,
            "AC_profiler_hot_spots": _profiler_hot_spots_as_dicts,

            # Webhook trigger (HTTP push triggers)
            "AC_webhook_start": _webhook_start,
            "AC_webhook_stop": _webhook_stop,
            "AC_webhook_add": _webhook_add,
            "AC_webhook_remove": _webhook_remove,
            "AC_webhook_list": _webhook_list,
            "AC_webhook_status": _webhook_status,

            # Email/IMAP poll trigger
            "AC_email_trigger_add": _email_trigger_add,
            "AC_email_trigger_remove": _email_trigger_remove,
            "AC_email_trigger_list": _email_trigger_list,
            "AC_email_trigger_start": _email_trigger_start,
            "AC_email_trigger_stop": _email_trigger_stop,
            "AC_email_trigger_poll_once": _email_trigger_poll_once,

            # Secret manager (encrypted vault for ${secrets.NAME})
            "AC_secret_init": _secret_initialize,
            "AC_secret_unlock": _secret_unlock,
            "AC_secret_lock": _secret_lock,
            "AC_secret_set": _secret_set,
            "AC_secret_remove": _secret_remove,
            "AC_secret_list": _secret_list,
            "AC_secret_status": _secret_status,

            # Accessibility-tree widget location
            "AC_a11y_list": _a11y_list_as_dicts,
            "AC_a11y_find": _a11y_find_as_dict,
            "AC_a11y_click": click_accessibility_element,
            "AC_a11y_dump": _a11y_dump,
            "AC_control_get_value": _control_get_value,
            "AC_control_set_value": _control_set_value,
            "AC_control_invoke": _control_invoke,
            "AC_control_toggle": _control_toggle,
            "AC_read_table": _read_table,
            "AC_watchdog_add": _watchdog_add,
            "AC_watchdog_start": _watchdog_start,
            "AC_watchdog_stop": _watchdog_stop,
            "AC_watchdog_list": _watchdog_list,
            "AC_handle_file_dialog": _handle_file_dialog,
            "AC_assert_session_active": _assert_session_active,
            "AC_queue_add": _queue_add,
            "AC_queue_next": _queue_next,
            "AC_queue_complete": _queue_complete,
            "AC_queue_fail": _queue_fail,
            "AC_queue_stats": _queue_stats,
            "AC_generate_data": _generate_data,
            "AC_mcp_manifest": _mcp_manifest,
            "AC_rank_tests": _rank_tests,
            "AC_select_tests": _select_tests,
            "AC_element_save": _element_save,
            "AC_element_find": _element_find,
            "AC_element_click": _element_click,
            "AC_element_remove": _element_remove,
            "AC_element_list": _element_list,
            "AC_debug_trace": _debug_trace,
            "AC_skill_save": _skill_save,
            "AC_skill_run": _skill_run,
            "AC_skill_list": _skill_list,
            "AC_skill_remove": _skill_remove,
            "AC_skill_search": _skill_search,
            "AC_guard_text": _guard_text,
            "AC_agent_card": _agent_card,
            "AC_read_workbook": _read_workbook,
            "AC_write_workbook": _write_workbook,
            "AC_read_document": _read_document,
            "AC_write_document": _write_document,
            "AC_read_presentation": _read_presentation,
            "AC_write_presentation": _write_presentation,
            "AC_memory_remember": _memory_remember,
            "AC_memory_recall": _memory_recall,
            "AC_memory_recent": _memory_recent,
            "AC_memory_forget": _memory_forget,
            "AC_memory_stats": _memory_stats,
            "AC_seed_everything": _seed_everything,
            "AC_observe_add": _observe_add,
            "AC_observe_remove": _observe_remove,
            "AC_observe_list": _observe_list,
            "AC_observe_poll": _observe_poll,
            "AC_observe_start": _observe_start,
            "AC_observe_stop": _observe_stop,
            "AC_generate_sbom": _generate_sbom,
            "AC_shard_suite": _shard_suite,
            "AC_merge_results": _merge_results,
            "AC_validate_rows": _validate_rows,
            "AC_extract_fields": _extract_fields,
            "AC_mask_rows": _mask_rows,
            "AC_pseudo_localize": _pseudo_localize,
            "AC_check_overflow": _check_overflow,
            "AC_check_catalog": _check_catalog,
            "AC_run_resumable": _run_resumable,
            "AC_checkpoint_status": _checkpoint_status,
            "AC_checkpoint_clear": _checkpoint_clear,
            "AC_mark_screen": _mark_screen,
            "AC_mark_click": _mark_click,
            "AC_screen_snapshot": _screen_snapshot,
            "AC_screen_diff": _screen_diff,
            "AC_screen_changed": _screen_changed,
            "AC_describe_screen": _describe_screen,
            "AC_replay_timeline": _replay_timeline,
            "AC_input_sequence": _input_sequence,
            "AC_circuit_call": _circuit_call,
            "AC_ci_annotations": _ci_annotations,
            "AC_clip_history_capture": _clip_history_capture,
            "AC_clip_history_list": _clip_history_list,
            "AC_clip_history_search": _clip_history_search,
            "AC_clip_history_start": _clip_history_start,
            "AC_clip_history_stop": _clip_history_stop,
            "AC_heal_stats": _heal_stats,
            "AC_scan_secrets": _scan_secrets,
            "AC_generate_sop": _generate_sop,
            "AC_a11y_record_start": _a11y_record_start,
            "AC_a11y_record_stop": _a11y_record_stop,
            "AC_a11y_record_events": _a11y_record_events,

            # VLM-based element locator
            "AC_vlm_locate": _vlm_locate_as_list,
            "AC_vlm_click": click_by_description,

            # Self-healing locator (template-first, VLM fallback, audit log)
            "AC_self_heal_locate": _self_heal_locate,
            "AC_self_heal_click": _self_heal_click,
            "AC_self_heal_log_list": _self_heal_log_list,
            "AC_self_heal_log_clear": _self_heal_log_clear,

            # Assertion DSL (verify screen state; raise on mismatch)
            "AC_assert_text": _assert_text,
            "AC_assert_image": _assert_image,
            "AC_assert_pixel": _assert_pixel,
            "AC_assert_window": _assert_window,
            "AC_assert_vlm": _assert_vlm,
            "AC_assert_clipboard": _assert_clipboard,
            "AC_assert_process": _assert_process,
            "AC_assert_file": _assert_file,
            "AC_assert_http": _assert_http,
            "AC_assert_all": _assert_all,
            "AC_assert_any": _assert_any,
            "AC_assert_eventually": _assert_eventually,

            # Action-file integrity (HMAC-SHA256 sign / verify)
            "AC_sign_action_file": _sign_action_file,
            "AC_verify_action_file": _verify_action_file,
            "AC_encrypt_action_file": _encrypt_action_file,
            "AC_decrypt_action_file": _decrypt_action_file,

            # Data-driven execution (load rows from CSV / JSON / SQLite / ...)
            "AC_load_data": _load_data,

            # Flaky-test detection (analytics over the run-history store)
            "AC_flaky_report": _flaky_report,

            # QA suite runner + CI report output (JUnit / Allure)
            "AC_run_suite": _run_suite,

            # Flaky quarantine (skip known-unstable cases in suites)
            "AC_quarantine_add": _quarantine_add,
            "AC_quarantine_remove": _quarantine_remove,
            "AC_quarantine_list": _quarantine_list,
            "AC_quarantine_clear": _quarantine_clear,
            "AC_quarantine_auto": _quarantine_auto,

            # Accessibility / i18n audit (missing labels, contrast, truncation)
            "AC_audit_accessibility": _audit_accessibility,
            "AC_audit_contrast": _audit_contrast,
            "AC_wcag_audit": _wcag_audit,

            # Mobile device matrix (parallel script across devices)
            "AC_run_device_matrix": _run_device_matrix,

            # Media assertions (audio activity, video motion)
            "AC_assert_audio": _assert_audio,
            "AC_assert_video_changes": _assert_video_changes,

            # Computer-use (Anthropic computer_20250124 closed-loop agent)
            "AC_computer_use": _computer_use,

            # Generic plan→act→verify→retry agent loop (Anthropic / OpenAI)
            "AC_run_agent": _run_agent,

            # Screenshot PII redaction (blur emails / credit cards /
            # password fields / explicit regions before upload).
            "AC_redact_screenshot": _redact_screenshot,

            # Cross-host DAG orchestrator
            "AC_run_dag": _run_dag,

            # Chat-ops slash-command router
            "AC_chatops_dispatch": _chatops_dispatch,

            # Anchor-based locator (spatial composition of locator backends)
            "AC_anchor_locate": _anchor_locate,
            "AC_anchor_click": _anchor_click,

            # Structured OCR (rows / tables / form fields)
            "AC_ocr_read_structure": _ocr_read_structure,

            # Smart waits (frame-diff replacements for time.sleep)
            "AC_wait_screen_stable": _wait_screen_stable,
            "AC_wait_pixel_changes": _wait_pixel_changes,
            "AC_wait_region_idle": _wait_region_idle,
            "AC_wait_for_file": _wait_for_file,
            "AC_wait_for_port": _wait_for_port,
            "AC_wait_for_process": _wait_for_process,
            "AC_wait_clipboard_change": _wait_clipboard_change,
            "AC_wait_window_closed": _wait_window_closed,

            # Cost telemetry (LLM token + USD tracking)
            "AC_costs_record": _costs_record,
            "AC_costs_summary": _costs_summary,
            "AC_costs_list": _costs_list,
            "AC_costs_clear": _costs_clear,

            # Failure → ticket automation (Jira / Linear / GitHub fan-out)
            "AC_failure_hook_fire": _failure_hook_fire,
            "AC_failure_hook_list": _failure_hook_list,
            "AC_failure_hook_clear": _failure_hook_clear,

            # A/B locator framework (race N locator strategies)
            "AC_ab_locate": _ab_locate,
            "AC_ab_report": _ab_report,
            "AC_ab_best_strategy": _ab_best_strategy,
            "AC_ab_clear": _ab_clear,

            # Multi-viewer presence roster (read-only / controller roles)
            "AC_presence_register": _presence_register,
            "AC_presence_unregister": _presence_unregister,
            "AC_presence_update_cursor": _presence_update_cursor,
            "AC_presence_set_role": _presence_set_role,
            "AC_presence_list": _presence_list,
            "AC_presence_clear": _presence_clear,

            # MCP server (Model Context Protocol stdio bridge)
            "AC_start_mcp_server": start_mcp_stdio_server,
            "AC_start_mcp_http_server": start_mcp_http_server,

            # WebRunner bridge (browser automation via je_web_runner)
            "AC_web_run": _ac_web_run,
            "AC_web_run_actions": _ac_web_run_actions,
            "AC_web_available": _ac_web_available,
            "AC_web_list_commands": _ac_web_list_commands,
            "AC_web_open": _ac_web_open,
            "AC_web_quit": _ac_web_quit,
            "AC_web_screenshot": _ac_web_screenshot,
            "AC_web_current_url": _ac_web_current_url,

            # Android via ADB (Phase 9.7)
            # uiautomator2 widget tree (find / click / dump)
            "AC_android_find_element": _ac_android_find_element,
            "AC_android_click_element": _ac_android_click_element,
            "AC_android_dump_hierarchy": _ac_android_dump_hierarchy,
            # iOS XCUITest (WebDriverAgent / facebook-wda)
            "AC_ios_tap": _ac_ios_tap,
            "AC_ios_swipe": _ac_ios_swipe,
            "AC_ios_type": _ac_ios_type,
            "AC_ios_screenshot": _ac_ios_screenshot,
            "AC_ios_find_element": _ac_ios_find_element,
            "AC_ios_click_element": _ac_ios_click_element,
            "AC_ios_dump_source": _ac_ios_dump_source,
            # Existing adb-based primitives
            "AC_android_tap": _ac_android_tap,
            "AC_android_swipe": _ac_android_swipe,
            "AC_android_key": _ac_android_key,
            "AC_android_text": _ac_android_text,
            "AC_android_screenshot": _ac_android_screenshot,
            "AC_android_list_devices": _ac_android_list_devices,
            "AC_android_shell": _ac_android_shell,

            # LLM action planner
            "AC_llm_plan": _llm_plan_for_executor,
            "AC_llm_run": _llm_run_for_executor,

            # Remote desktop host (this machine streams to others)
            "AC_start_remote_host": _remote_start_host,
            "AC_stop_remote_host": _remote_stop_host,
            "AC_remote_host_status": _remote_host_status,

            # Remote desktop viewer (this machine controls others)
            "AC_remote_connect": _remote_connect,
            "AC_remote_disconnect": _remote_disconnect,
            "AC_remote_viewer_status": _remote_viewer_status,
            "AC_remote_send_input": _remote_send_input,

            # WebSocket-transport remote desktop host
            "AC_start_ws_host": _ws_start_host,
            "AC_stop_ws_host": _ws_stop_host,
            "AC_ws_host_status": _ws_host_status,

            # WebSocket-transport remote desktop viewer
            "AC_ws_connect": _ws_connect,
            "AC_ws_disconnect": _ws_disconnect,
            "AC_ws_viewer_status": _ws_viewer_status,
            "AC_ws_send_input": _ws_send_input,

            # WebRTC-transport host (manual SDP exchange)
            "AC_start_webrtc_host": _webrtc_start_host,
            "AC_webrtc_create_offer": _webrtc_create_offer,
            "AC_webrtc_accept_answer": _webrtc_accept_answer,
            "AC_stop_webrtc_host": _webrtc_stop_host,
            "AC_webrtc_host_status": _webrtc_host_status,

            # WebRTC-transport viewer (manual SDP exchange)
            "AC_start_webrtc_viewer": _webrtc_start_viewer,
            "AC_webrtc_process_offer": _webrtc_process_offer,
            "AC_webrtc_send_input": _webrtc_send_input,
            "AC_stop_webrtc_viewer": _webrtc_stop_viewer,
            "AC_webrtc_viewer_status": _webrtc_viewer_status,

            # Virtual gamepad (ViGEm — drives games that ignore SendInput)
            "AC_gamepad_press": _gamepad_press,
            "AC_gamepad_release": _gamepad_release,
            "AC_gamepad_click": _gamepad_click,
            "AC_gamepad_dpad": _gamepad_dpad,
            "AC_gamepad_left_stick": _gamepad_left_stick,
            "AC_gamepad_right_stick": _gamepad_right_stick,
            "AC_gamepad_left_trigger": _gamepad_left_trigger,
            "AC_gamepad_right_trigger": _gamepad_right_trigger,
            "AC_gamepad_reset": _gamepad_reset,

            # REST API (HTTP front-end exposing the headless API)
            "AC_rest_api_start": _rest_api_start,
            "AC_rest_api_stop": _rest_api_stop,
            "AC_rest_api_status": _rest_api_status,

            # Admin console (manage many remote AutoControl REST hosts)
            "AC_admin_add_host": _admin_add_host,
            "AC_admin_remove_host": _admin_remove_host,
            "AC_admin_list_hosts": _admin_list_hosts,
            "AC_admin_poll": _admin_poll,
            "AC_admin_broadcast_execute": _admin_broadcast_execute,

            # Audit log (tamper-evident security log)
            "AC_audit_log_list": _audit_log_list,
            "AC_audit_log_verify": _audit_log_verify,
            "AC_audit_log_clear": _audit_log_clear,

            # WebRTC inspector (live stat history)
            "AC_inspector_recent": _inspector_recent,
            "AC_inspector_summary": _inspector_summary,
            "AC_inspector_reset": _inspector_reset,

            # USB device enumeration (read-only)
            "AC_list_usb_devices": _list_usb_devices,

            # USB hotplug watcher (Phase 1.5)
            "AC_usb_watch_start": _usb_watch_start,
            "AC_usb_watch_stop": _usb_watch_stop,
            "AC_usb_recent_events": _usb_recent_events,

            # USB passthrough (Phase 2) — flag, ACL, local + remote use
            "AC_usb_passthrough_enable": _usb_passthrough_enable,
            "AC_usb_passthrough_status": _usb_passthrough_status,
            "AC_usb_acl_list": _usb_acl_list,
            "AC_usb_acl_add": _usb_acl_add,
            "AC_usb_acl_remove": _usb_acl_remove,
            "AC_usb_acl_set_default": _usb_acl_set_default,
            "AC_usb_acl_export": _usb_acl_export,
            "AC_usb_acl_import": _usb_acl_import,
            "AC_usb_loopback_list": _usb_loopback_list,
            "AC_usb_loopback_open": _usb_loopback_open,
            "AC_usb_remote_list": _usb_remote_list,
            "AC_usb_remote_open": _usb_remote_open,

            # System diagnostics
            "AC_diagnose": _diagnose,

            # Config bundle export / import
            "AC_config_export": _config_export,
            "AC_config_import": _config_import,

            # Screenshot annotation (boxes / highlights / arrows / labels)
            "AC_annotate_screenshot": _annotate_screenshot,

            # Desktop notification
            "AC_notify": _notify,

            # Region colour statistics (dominant / average colour)
            "AC_region_color_stats": _region_color_stats,

            # Recoverable deletion (move a file to the OS recycle bin)
            "AC_move_to_trash": _move_to_trash,

            # QR code decoding from a screen region
            "AC_read_qr": _read_qr,

            # Scroll until a target image / text is visible
            "AC_scroll_to_find": _scroll_to_find,

            # Per-window capture + window-layout save / restore + snap
            "AC_capture_window": _capture_window,
            "AC_save_window_layout": _save_window_layout,
            "AC_restore_window_layout": _restore_window_layout,
            "AC_snap_window": _snap_window,
        }

    def known_commands(self) -> set:
        """Return the set of all command names the executor recognises."""
        return set(self.event_dict.keys()) | set(self._block_commands.keys())

    def _resolve_runtime_args(self, args: Any) -> Any:
        """Interpolate ``${var}`` placeholders against the current scope.

        Keys inside :attr:`_DEFERRED_ARG_KEYS` (``body``/``then``/``else``)
        are left as-is so nested action lists keep their placeholders for
        per-iteration evaluation.
        """
        if not self.variables:
            return args
        if isinstance(args, dict):
            resolved: Dict[str, Any] = {}
            for key, value in args.items():
                if key in self._DEFERRED_ARG_KEYS:
                    resolved[key] = value
                else:
                    resolved[key] = interpolate_value(value, self.variables)
            return resolved
        if isinstance(args, list):
            return [interpolate_value(item, self.variables) for item in args]
        return args

    def _execute_event(self, action: list) -> Any:
        """
        執行單一事件
        Execute a single event
        """
        name = action[0]
        block_handler = self._block_commands.get(name)
        if block_handler is not None:
            args = action[1] if len(action) == 2 else {}
            if not isinstance(args, dict):
                raise AutoControlActionException(
                    f"{name} requires a dict of arguments"
                )
            return block_handler(self, self._resolve_runtime_args(args))

        event = self.event_dict.get(name)
        if event is None:
            raise AutoControlActionException(f"Unknown action: {name}")

        if len(action) == 2:
            resolved = self._resolve_runtime_args(action[1])
            if isinstance(resolved, dict):
                return event(**resolved)
            return event(*resolved)
        if len(action) == 1:
            return event()
        raise AutoControlActionException(cant_execute_action_error_message + " " + str(action))

    def execute_action(self, action_list: Union[list, dict],
                       raise_on_error: bool = False,
                       _validated: bool = False,
                       dry_run: bool = False,
                       step_callback: Optional[Callable[[list], None]] = None,
                       ) -> Dict[str, str]:
        """
        執行 action list
        Execute all actions in action list

        :param action_list: list 或 dict (包含 auto_control key)
        :param raise_on_error: 若為 True，遇到錯誤立即拋出 (流程控制用)
        :param _validated: 內部用；子呼叫已驗證過時避免重複驗證
        :param dry_run: 若為 True，只記錄將執行的動作，不實際呼叫。
        :param step_callback: 每個 action 開始前呼叫此 hook（偵錯用）。
        :return: 執行紀錄字典
        """
        autocontrol_logger.info(f"execute_action, action_list: {action_list}")
        action_list = self._unwrap_action_list(action_list)
        if not _validated:
            validate_actions(action_list, self.known_commands())

        execute_record_dict: Dict[str, Any] = {}
        for action in action_list:
            if step_callback is not None:
                step_callback(action)
            if dry_run:
                execute_record_dict["dry-run: " + str(action)] = "(not executed)"
                continue
            self._run_one_action(action, execute_record_dict, raise_on_error)

        for key, value in execute_record_dict.items():
            autocontrol_logger.info("%s -> %s", key, value)
        return execute_record_dict

    @staticmethod
    def _unwrap_action_list(action_list: Union[list, dict]) -> list:
        """Normalise the ``action_list`` argument or raise on invalid input."""
        if isinstance(action_list, dict):
            action_list = action_list.get("auto_control")
            if action_list is None:
                raise AutoControlActionNullException(executor_list_error_message)
        if not isinstance(action_list, list) or len(action_list) == 0:
            raise AutoControlActionNullException(action_is_null_error_message)
        return action_list

    def _run_one_action(self, action: list, record: Dict[str, Any],
                        raise_on_error: bool) -> None:
        """Execute a single action, recording the result or raising."""
        import time as _time
        key = "execute: " + str(action)
        action_name = action[0] if action and isinstance(action[0], str) else "<invalid>"
        started = _time.monotonic()
        try:
            with default_profiler.measure(action_name):
                record[key] = self._execute_event(action)
            _observe_executor_metrics(action_name, started, error=None)
        except (LoopBreak, LoopContinue):
            raise
        except (AutoControlActionException, OSError, RuntimeError,
                AttributeError, TypeError, ValueError) as error:
            _observe_executor_metrics(action_name, started, error=error)
            if raise_on_error:
                raise
            autocontrol_logger.info(
                f"execute_action failed, action: {action}, error: {repr(error)}"
            )
            record_action_to_list("AC_execute_action", None, repr(error))
            record[key] = repr(error)

    def execute_files(self, execute_files_list: list) -> List[Dict[str, str]]:
        """
        執行 action files
        Execute actions from files

        :param execute_files_list: list of file paths
        :return: 每個檔案的執行結果
        """
        autocontrol_logger.info(f"execute_files, execute_files_list: {execute_files_list}")
        from je_auto_control.utils.action_signing import require_signed_actions
        execute_detail_list = []
        for file in execute_files_list:
            require_signed_actions(file)
            execute_detail_list.append(self.execute_action(read_action_json(file)))
        return execute_detail_list


# === 全域 Executor 實例 Global Executor Instance ===
executor = Executor()
package_manager.executor = executor


def add_command_to_executor(command_dict: dict) -> None:
    """
    新增自訂指令到 Executor
    Add custom commands to Executor

    :param command_dict: dict {command_name: function}
    """
    for command_name, command in command_dict.items():
        if isinstance(command, (types.MethodType, types.FunctionType)):
            executor.event_dict[command_name] = command
        else:
            raise AutoControlAddCommandException(add_command_exception_error_message)


def execute_action(action_list: list) -> Dict[str, str]:
    return executor.execute_action(action_list)


def execute_files(execute_files_list: list) -> List[Dict[str, str]]:
    return executor.execute_files(execute_files_list)


def execute_action_with_vars(action_list: list, variables: dict
                             ) -> Dict[str, str]:
    """Interpolate ``${name}`` placeholders with ``variables`` and execute.

    The same mapping seeds the runtime variable scope so flow-control
    commands (``AC_set_var``/``AC_if_var``/...) can read and mutate the
    same values during execution.
    """
    resolved = interpolate_actions(action_list, variables)
    executor.variables.update_many(variables)
    return executor.execute_action(resolved)