From a90c7c67a433a597f0c41f64e2ba835e832c626f Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Tue, 23 Jun 2026 06:38:16 +0800 Subject: [PATCH] Add actionability gate (visible + stable + enabled + not-occluded) --- README/WHATS_NEW_zh-CN.md | 6 + README/WHATS_NEW_zh-TW.md | 6 + WHATS_NEW.md | 6 + .../doc/new_features/v137_features_doc.rst | 48 ++++++ docs/source/Eng/eng_index.rst | 1 + .../Zh/doc/new_features/v137_features_doc.rst | 42 +++++ docs/source/Zh/zh_index.rst | 1 + je_auto_control/__init__.py | 8 + .../gui/script_builder/command_schema.py | 13 ++ .../utils/actionability/__init__.py | 7 + .../utils/actionability/actionability.py | 156 ++++++++++++++++++ .../utils/executor/action_executor.py | 29 ++++ .../utils/mcp_server/tools/_factories.py | 25 ++- .../utils/mcp_server/tools/_handlers.py | 6 + .../headless/test_actionability_batch.py | 109 ++++++++++++ 15 files changed, 462 insertions(+), 1 deletion(-) create mode 100644 docs/source/Eng/doc/new_features/v137_features_doc.rst create mode 100644 docs/source/Zh/doc/new_features/v137_features_doc.rst create mode 100644 je_auto_control/utils/actionability/__init__.py create mode 100644 je_auto_control/utils/actionability/actionability.py create mode 100644 test/unit_test/headless/test_actionability_batch.py diff --git a/README/WHATS_NEW_zh-CN.md b/README/WHATS_NEW_zh-CN.md index 2b0a8374..758136c0 100644 --- a/README/WHATS_NEW_zh-CN.md +++ b/README/WHATS_NEW_zh-CN.md @@ -1,5 +1,11 @@ # 本次更新 — AutoControl +## 本次更新 (2026-06-23) — 可操作性闸门(操作前先等待就绪) + +目标真正就绪前不要点击。完整参考:[`docs/source/Zh/doc/new_features/v137_features_doc.rst`](../docs/source/Zh/doc/new_features/v137_features_doc.rst)。 + +- **`wait_actionable` / `act_when_ready`**(`AC_wait_actionable`):Playwright/Cypress 在每次点击前都会做可操作性检查——存在 + 已停止移动 + 启用 + 未被遮盖——但 AutoControl 先前没有(`self_heal_click` 立即点击;`wait_until_screen_stable` 观察整个画面)。本功能把这四项合成单一闸门,返回 `ActionabilityReport`(各项检查布尔值、目标 `point`、`reason` = 第一个失败的检查)。每个信号都是可注入 callable(`bbox_provider` / `region_sampler` / `enabled_probe` / `hit_tester`)再加可注入 `clock`/`sleep`,因此完全确定性且可无头测试。执行器命令以模板图像把关。 + ## 本次更新 (2026-06-23) — 多显示器 / 虚拟桌面几何 在多台显示器间正确摆放窗口与坐标。完整参考:[`docs/source/Zh/doc/new_features/v136_features_doc.rst`](../docs/source/Zh/doc/new_features/v136_features_doc.rst)。 diff --git a/README/WHATS_NEW_zh-TW.md b/README/WHATS_NEW_zh-TW.md index 7d731ab3..97d830c2 100644 --- a/README/WHATS_NEW_zh-TW.md +++ b/README/WHATS_NEW_zh-TW.md @@ -1,5 +1,11 @@ # 本次更新 — AutoControl +## 本次更新 (2026-06-23) — 可操作性閘門(操作前先等待就緒) + +目標真正就緒前不要點擊。完整參考:[`docs/source/Zh/doc/new_features/v137_features_doc.rst`](../docs/source/Zh/doc/new_features/v137_features_doc.rst)。 + +- **`wait_actionable` / `act_when_ready`**(`AC_wait_actionable`):Playwright/Cypress 在每次點擊前都會做可操作性檢查——存在 + 已停止移動 + 啟用 + 未被遮蓋——但 AutoControl 先前沒有(`self_heal_click` 立即點擊;`wait_until_screen_stable` 觀察整個畫面)。本功能把這四項合成單一閘門,回傳 `ActionabilityReport`(各項檢查布林值、目標 `point`、`reason` = 第一個失敗的檢查)。每個訊號都是可注入 callable(`bbox_provider` / `region_sampler` / `enabled_probe` / `hit_tester`)再加可注入 `clock`/`sleep`,因此完全決定性且可無頭測試。執行器命令以模板影像把關。 + ## 本次更新 (2026-06-23) — 多螢幕 / 虛擬桌面幾何 在多台顯示器間正確擺放視窗與座標。完整參考:[`docs/source/Zh/doc/new_features/v136_features_doc.rst`](../docs/source/Zh/doc/new_features/v136_features_doc.rst)。 diff --git a/WHATS_NEW.md b/WHATS_NEW.md index 597329a1..d3b82083 100644 --- a/WHATS_NEW.md +++ b/WHATS_NEW.md @@ -1,5 +1,11 @@ # What's New — AutoControl +## What's new (2026-06-23) — Actionability Gate (Wait Until Ready Before Acting) + +Don't click until the target is genuinely ready. Full reference: [`docs/source/Eng/doc/new_features/v137_features_doc.rst`](docs/source/Eng/doc/new_features/v137_features_doc.rst). + +- **`wait_actionable` / `act_when_ready`** (`AC_wait_actionable`): Playwright/Cypress run an actionability check before every click — present + stopped moving + enabled + not covered — but AutoControl had none (`self_heal_click` clicks immediately; `wait_until_screen_stable` watches the whole frame). This composes the four checks into one gate and returns an `ActionabilityReport` (per-check booleans, target `point`, `reason` = first failing check). Every signal is an injectable callable (`bbox_provider` / `region_sampler` / `enabled_probe` / `hit_tester`) plus an injectable `clock`/`sleep`, so it's fully deterministic and headless-testable. The executor command gates on a template image. + ## What's new (2026-06-23) — Multi-Monitor / Virtual-Desktop Geometry Place windows and points correctly across several displays. Full reference: [`docs/source/Eng/doc/new_features/v136_features_doc.rst`](docs/source/Eng/doc/new_features/v136_features_doc.rst). diff --git a/docs/source/Eng/doc/new_features/v137_features_doc.rst b/docs/source/Eng/doc/new_features/v137_features_doc.rst new file mode 100644 index 00000000..526f1d7b --- /dev/null +++ b/docs/source/Eng/doc/new_features/v137_features_doc.rst @@ -0,0 +1,48 @@ +Actionability Gate — Wait Until Ready Before Acting +=================================================== + +Modern UI frameworks (Playwright, Cypress, WebdriverIO) run an *actionability* check +before every click: the target must be present, have stopped moving, be enabled, and +actually receive the event (not be covered). AutoControl had no equivalent — +``self_heal_click`` locates and clicks immediately, and ``wait_until_screen_stable`` +only watches the *whole* frame. ``wait_actionable`` composes the four checks into one +gate, so a click lands on a button that is genuinely ready rather than mid-animation, +disabled, or behind a dialog. + +Every signal is an injectable callable — ``bbox_provider`` (locate the target), +``region_sampler`` (pixel-stability token), ``enabled_probe``, ``hit_tester`` — plus +an injectable ``clock`` / ``sleep`` via :class:`GateConfig`, so the gate is fully +deterministic and headless-testable. Imports no ``PySide6``. + +Headless API +------------ + +.. code-block:: python + + from je_auto_control import wait_actionable, act_when_ready, GateConfig + + report = wait_actionable( + bbox_provider=lambda: locate_button(), # () -> (x, y, w, h) or None + enabled_probe=lambda: not is_greyed_out(), + config=GateConfig(timeout_s=8.0, stable_for_s=0.4)) + if report.actionable: + click(*report.point) + else: + print("blocked:", report.reason) # not visible / not stable / … + + # Or gate + act in one call (raises if it never becomes actionable): + act_when_ready(lambda point: click(*point), bbox_provider=locate_button) + +``wait_actionable`` returns an :class:`ActionabilityReport` with ``actionable`` plus +the per-check booleans (``visible`` / ``stable`` / ``enabled`` / ``receives_events``), +the target ``point``, ``waited_s`` and a ``reason`` (the first failing check). +``act_when_ready`` waits and then calls ``action(center_point)``, raising +``AutoControlActionException`` on timeout. + +Executor command +---------------- + +``AC_wait_actionable`` binds the gate to a ``template`` image (located each poll) and +samples that region's pixels for stability: ``timeout_s`` / ``stable_for_s`` / +``min_score`` / ``region`` → the report dict. It is exposed as the MCP tool +``ac_wait_actionable`` and as a Script Builder command under **Flow**. diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst index 2c7e2e60..30e3bb55 100644 --- a/docs/source/Eng/eng_index.rst +++ b/docs/source/Eng/eng_index.rst @@ -159,6 +159,7 @@ Comprehensive guides for all AutoControl features. doc/new_features/v134_features_doc doc/new_features/v135_features_doc doc/new_features/v136_features_doc + doc/new_features/v137_features_doc doc/ocr_backends/ocr_backends_doc doc/observability/observability_doc doc/operations_layer/operations_layer_doc diff --git a/docs/source/Zh/doc/new_features/v137_features_doc.rst b/docs/source/Zh/doc/new_features/v137_features_doc.rst new file mode 100644 index 00000000..bb40e254 --- /dev/null +++ b/docs/source/Zh/doc/new_features/v137_features_doc.rst @@ -0,0 +1,42 @@ +可操作性閘門——在操作前先等待就緒 +================================== + +現代 UI 框架(Playwright、Cypress、WebdriverIO)在每次點擊前都會執行*可操作性*檢查:目標必須存在、已停止移動、 +為啟用狀態,且確實能接收事件(未被遮蓋)。AutoControl 先前沒有對應功能——``self_heal_click`` 定位後立即點擊, +``wait_until_screen_stable`` 只觀察*整個*畫面。``wait_actionable`` 把這四項檢查合成單一閘門,讓點擊落在真正就緒的 +按鈕上,而非動畫中、停用、或被對話框擋住的狀態。 + +每個訊號都是可注入的 callable——``bbox_provider``(定位目標)、``region_sampler``(像素穩定 token)、 +``enabled_probe``、``hit_tester``——再加上透過 :class:`GateConfig` 注入的 ``clock`` / ``sleep``,因此閘門完全 +決定性且可無頭測試。不匯入 ``PySide6``。 + +無頭 API +-------- + +.. code-block:: python + + from je_auto_control import wait_actionable, act_when_ready, GateConfig + + report = wait_actionable( + bbox_provider=lambda: locate_button(), # () -> (x, y, w, h) 或 None + enabled_probe=lambda: not is_greyed_out(), + config=GateConfig(timeout_s=8.0, stable_for_s=0.4)) + if report.actionable: + click(*report.point) + else: + print("blocked:", report.reason) # not visible / not stable / … + + # 或一次完成「等待 + 操作」(若始終未就緒則丟例外): + act_when_ready(lambda point: click(*point), bbox_provider=locate_button) + +``wait_actionable`` 回傳 :class:`ActionabilityReport`,含 ``actionable`` 以及各項檢查布林值 +(``visible`` / ``stable`` / ``enabled`` / ``receives_events``)、目標 ``point``、``waited_s`` 與 ``reason`` +(第一個失敗的檢查)。``act_when_ready`` 等待後呼叫 ``action(center_point)``,逾時則丟出 +``AutoControlActionException``。 + +執行器命令 +---------- + +``AC_wait_actionable`` 將閘門綁定到 ``template`` 影像(每次輪詢時定位),並取樣該區域像素以判斷穩定: +``timeout_s`` / ``stable_for_s`` / ``min_score`` / ``region`` → 回傳報告字典。它以 MCP 工具 +``ac_wait_actionable`` 以及 Script Builder 中 **Flow** 分類下的命令提供。 diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst index c8d59775..2fcb5579 100644 --- a/docs/source/Zh/zh_index.rst +++ b/docs/source/Zh/zh_index.rst @@ -159,6 +159,7 @@ AutoControl 所有功能的完整使用指南。 doc/new_features/v134_features_doc doc/new_features/v135_features_doc doc/new_features/v136_features_doc + doc/new_features/v137_features_doc doc/ocr_backends/ocr_backends_doc doc/observability/observability_doc doc/operations_layer/operations_layer_doc diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index f879ef45..cf1890a9 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -308,6 +308,10 @@ Monitor, enumerate_monitors, monitor_at_point, monitor_for_window, primary_monitor, remap_point, to_local, to_virtual, virtual_bounds, ) +# Pre-action readiness gate (visible + stable + enabled + not-occluded) +from je_auto_control.utils.actionability import ( + ActionabilityReport, GateConfig, act_when_ready, wait_actionable, +) # CI workflow annotations (GitHub Actions) from je_auto_control.utils.ci_annotations import ( emit_annotations, format_annotation, @@ -1133,6 +1137,10 @@ def start_autocontrol_gui(*args, **kwargs): "to_local", "to_virtual", "virtual_bounds", + "wait_actionable", + "act_when_ready", + "ActionabilityReport", + "GateConfig", "emit_annotations", "format_annotation", "ClipboardHistory", "default_clipboard_history", "analyze_heal_log", "heal_stats", "scan_secrets", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index 5d70a9dd..c1b69f35 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -654,6 +654,19 @@ def _add_flow_specs(specs: List[CommandSpec]) -> None: min_value=0.01), ), )) + specs.append(CommandSpec( + "AC_wait_actionable", "Flow", "Wait Until Actionable", + fields=( + FieldSpec("template", FieldType.FILE_PATH), + FieldSpec("timeout_s", FieldType.FLOAT, optional=True, default=5.0), + FieldSpec("stable_for_s", FieldType.FLOAT, optional=True, default=0.3), + FieldSpec("min_score", FieldType.FLOAT, optional=True, default=0.8, + min_value=0.0, max_value=1.0), + FieldSpec("region", FieldType.STRING, optional=True, + placeholder="[left, top, right, bottom]"), + ), + description="Wait until a target is visible + stable before acting.", + )) specs.append(CommandSpec( "AC_wait_pixel", "Flow", "Wait for Pixel", fields=( diff --git a/je_auto_control/utils/actionability/__init__.py b/je_auto_control/utils/actionability/__init__.py new file mode 100644 index 00000000..453ca1ff --- /dev/null +++ b/je_auto_control/utils/actionability/__init__.py @@ -0,0 +1,7 @@ +"""Pre-action readiness gate (visible + stable + enabled + not-occluded).""" +from je_auto_control.utils.actionability.actionability import ( + ActionabilityReport, GateConfig, act_when_ready, wait_actionable, +) + +__all__ = ["ActionabilityReport", "GateConfig", "act_when_ready", + "wait_actionable"] diff --git a/je_auto_control/utils/actionability/actionability.py b/je_auto_control/utils/actionability/actionability.py new file mode 100644 index 00000000..3389d8ac --- /dev/null +++ b/je_auto_control/utils/actionability/actionability.py @@ -0,0 +1,156 @@ +"""Pre-action readiness gate — visible + stable + enabled + not-occluded before acting. + +Modern UI frameworks (Playwright, Cypress, WebdriverIO) auto-run an *actionability* +check before every click: the target must be present, have stopped moving, be enabled, +and actually receive the event (not be covered). AutoControl had none — ``self_heal_click`` +locates and clicks immediately, and ``wait_until_screen_stable`` only watches the *whole* +frame. This composes the four checks into one gate. + +Every signal is an injectable callable — ``bbox_provider`` (locate the target), +``region_sampler`` (pixel-stability token), ``enabled_probe`` (is it enabled?), +``hit_tester`` (is the click point on top?) — plus an injectable ``clock`` / ``sleep``, +so the gate is fully deterministic and headless-testable. Imports no ``PySide6``. +""" +import time +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, List, Optional, Tuple + +from je_auto_control.utils.exception.exceptions import AutoControlActionException + +Bbox = Tuple[int, int, int, int] +BboxProvider = Callable[[], Optional[Bbox]] + + +@dataclass +class GateConfig: + """Timing knobs and test seams for the actionability gate.""" + + timeout_s: float = 5.0 + stable_for_s: float = 0.3 + poll_interval_s: float = 0.1 + clock: Callable[[], float] = time.monotonic + sleep: Callable[[float], None] = time.sleep + + +@dataclass(frozen=True) +class ActionabilityReport: + """The outcome of a gate: per-check booleans, the target point, and timing.""" + + actionable: bool + visible: bool + stable: bool + enabled: bool + receives_events: bool + waited_s: float + reason: str + point: Optional[List[int]] = None + checks: Dict[str, bool] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + """Return the report as a plain dict.""" + return {"actionable": self.actionable, "visible": self.visible, + "stable": self.stable, "enabled": self.enabled, + "receives_events": self.receives_events, "waited_s": self.waited_s, + "reason": self.reason, "point": self.point, "checks": self.checks} + + +def _center(bbox: Bbox) -> List[int]: + return [bbox[0] + bbox[2] // 2, bbox[1] + bbox[3] // 2] + + +def _reason(visible: bool, stable: bool, enabled: bool, receives: bool) -> str: + if not visible: + return "not visible" + if not stable: + return "not stable" + if not enabled: + return "disabled" + if not receives: + return "occluded" + return "actionable" + + +class _StabilityTracker: + """Tracks how long the bbox (and optional pixel sample) has stayed unchanged.""" + + def __init__(self, sampler, stable_for_s: float): + self._sampler = sampler + self._stable_for_s = stable_for_s + self._prev: Any = object() + self._since: Optional[float] = None + + def update(self, bbox: Optional[Bbox], now: float) -> bool: + if bbox is None: + self._since = None + self._prev = object() + return False + token = (bbox, self._sampler(bbox) if self._sampler else None) + if token != self._prev: + self._prev = token + self._since = now + return False + return (now - self._since) >= self._stable_for_s + + +def _evaluate(bbox, tracker, now, enabled_probe, hit_tester): + """Return ``(visible, stable, enabled, receives, point)`` for one poll.""" + visible = bbox is not None + stable = tracker.update(bbox, now) + enabled = enabled_probe is None or bool(enabled_probe()) + point = _center(bbox) if visible else None + receives = hit_tester is None or point is None or bool(hit_tester(point)) + return visible, stable, enabled, receives, point + + +def _report(visible, stable, enabled, receives, point, waited): + actionable = visible and stable and enabled and receives + return ActionabilityReport( + actionable, visible, stable, enabled, receives, round(waited, 4), + _reason(visible, stable, enabled, receives), point, + {"visible": visible, "stable": stable, "enabled": enabled, + "receives_events": receives}) + + +def wait_actionable(bbox_provider: BboxProvider, *, + region_sampler: Optional[Callable[[Bbox], Any]] = None, + enabled_probe: Optional[Callable[[], Optional[bool]]] = None, + hit_tester: Optional[Callable[[List[int]], bool]] = None, + config: Optional[GateConfig] = None) -> ActionabilityReport: + """Poll until the target is visible + stable + enabled + unoccluded, or timeout. + + ``bbox_provider`` returns the target ``(x, y, w, h)`` or ``None``. The optional + ``region_sampler`` returns a pixel token over the bbox (movement/animation + settling); ``enabled_probe`` reports enabled state; ``hit_tester`` reports + whether the centre point is on top. Returns an :class:`ActionabilityReport`. + """ + cfg = config or GateConfig() + tracker = _StabilityTracker(region_sampler, cfg.stable_for_s) + start = cfg.clock() + deadline = start + cfg.timeout_s + while True: + now = cfg.clock() + signals = _evaluate(bbox_provider(), tracker, now, enabled_probe, + hit_tester) + report = _report(*signals, now - start) + if report.actionable or now >= deadline: + return report + cfg.sleep(cfg.poll_interval_s) + + +def act_when_ready(action: Callable[[List[int]], Any], bbox_provider: BboxProvider, + *, region_sampler: Optional[Callable[[Bbox], Any]] = None, + enabled_probe: Optional[Callable[[], Optional[bool]]] = None, + hit_tester: Optional[Callable[[List[int]], bool]] = None, + config: Optional[GateConfig] = None) -> Any: + """Wait for the target to be actionable, then call ``action(center_point)``. + + Raises ``AutoControlActionException`` (with the failing check) if the gate times + out before the target becomes actionable. + """ + report = wait_actionable(bbox_provider, region_sampler=region_sampler, + enabled_probe=enabled_probe, hit_tester=hit_tester, + config=config) + if not report.actionable: + raise AutoControlActionException( + f"target not actionable ({report.reason}) after {report.waited_s}s") + return action(report.point) diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index 20cec92e..d606248c 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -3451,6 +3451,34 @@ def _monitor_at_point(x: Any, y: Any) -> Dict[str, Any]: "monitor": monitor.to_dict() if monitor else None} +def _region_pixel_token(bbox): + """Stability token: a hash of the bbox region's pixels (changes on movement).""" + from je_auto_control.utils.cv2_utils.screenshot import pil_screenshot + left, top, width, height = bbox + image = pil_screenshot(screen_region=[left, top, left + width, top + height]) + return hash(image.tobytes()) + + +def _wait_actionable(template: str, timeout_s: Any = 5.0, stable_for_s: Any = 0.3, + min_score: Any = 0.8, region: Any = None) -> Dict[str, Any]: + """Adapter: wait until a template is visible + stable before acting.""" + import json + from je_auto_control.utils.actionability import GateConfig, wait_actionable + from je_auto_control.utils.visual_match import match_template + if isinstance(region, str): + region = json.loads(region) if region.strip() else None + + def locate(): + match = match_template(template, region=region, min_score=float(min_score)) + return (match.x, match.y, match.width, match.height) if match else None + + report = wait_actionable( + locate, region_sampler=_region_pixel_token, + config=GateConfig(timeout_s=float(timeout_s), + stable_for_s=float(stable_for_s))) + return report.to_dict() + + def _with_modifiers(modifiers: Any, actions: Any) -> Dict[str, Any]: """Adapter: run nested actions while modifier keys are held down.""" import json @@ -5180,6 +5208,7 @@ def __init__(self): "AC_preprocess_image": _preprocess_image, "AC_enumerate_monitors": _enumerate_monitors, "AC_monitor_at_point": _monitor_at_point, + "AC_wait_actionable": _wait_actionable, "AC_tile_rect": _tile_rect, "AC_grid_rects": _grid_rects, "AC_cascade_rects": _cascade_rects, diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index a2965e2b..3dacc1c3 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -2844,6 +2844,29 @@ def monitor_layout_tools() -> List[MCPTool]: ] +def actionability_tools() -> List[MCPTool]: + return [ + MCPTool( + name="ac_wait_actionable", + description=("Wait until 'template' (image path) is VISIBLE and STABLE " + "(stopped moving / animating) on screen before you act — " + "the Playwright-style actionability gate. Returns {actionable, " + "visible, stable, enabled, receives_events, point, reason, " + "waited_s}. 'timeout_s', 'stable_for_s', 'min_score', " + "'region'."), + input_schema=schema({ + "template": {"type": "string"}, + "timeout_s": {"type": "number"}, + "stable_for_s": {"type": "number"}, + "min_score": {"type": "number"}, + "region": {"type": "array", "items": {"type": "integer"}}}, + required=["template"]), + handler=h.wait_actionable, + annotations=READ_ONLY, + ), + ] + + def ssim_tools() -> List[MCPTool]: return [ MCPTool( @@ -6347,7 +6370,7 @@ def media_assert_tools() -> List[MCPTool]: modifier_state_tools, grid_locator_tools, visual_match_tools, color_region_tools, ssim_tools, feature_match_tools, shape_locator_tools, window_layout_tools, window_arrange_tools, preprocess_tools, - monitor_layout_tools, plugin_sdk_tools, governance_tools, + monitor_layout_tools, actionability_tools, plugin_sdk_tools, governance_tools, credential_lease_tools, egress_tools, approval_testing_tools, trajectory_eval_tools, compliance_tools, agent_trace_tools, video_report_tools, fuzzy_tools, artifact_store_tools, image_dedup_tools, diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index 02499205..8b39ef7c 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -2172,6 +2172,12 @@ def monitor_at_point(x, y): return _monitor_at_point(x, y) +def wait_actionable(template, timeout_s=5.0, stable_for_s=0.3, min_score=0.8, + region=None): + from je_auto_control.utils.executor.action_executor import _wait_actionable + return _wait_actionable(template, timeout_s, stable_for_s, min_score, region) + + def detect_drift(reference, current, threshold=0.25, bins=10): from je_auto_control.utils.executor.action_executor import _detect_drift return _detect_drift(reference, current, threshold, bins) diff --git a/test/unit_test/headless/test_actionability_batch.py b/test/unit_test/headless/test_actionability_batch.py new file mode 100644 index 00000000..177699ec --- /dev/null +++ b/test/unit_test/headless/test_actionability_batch.py @@ -0,0 +1,109 @@ +"""Headless tests for the pre-action actionability gate. No Qt; clock is injected.""" +import pytest + +import je_auto_control as ac +from je_auto_control.utils.actionability import ( + GateConfig, act_when_ready, wait_actionable, +) + + +class _Clock: + """A fake clock that only advances when the gate sleeps (deterministic).""" + + def __init__(self): + self.t = 0.0 + + def now(self): + return self.t + + def sleep(self, seconds): + self.t += seconds + + +def _config(): + clock = _Clock() + return GateConfig(timeout_s=5.0, stable_for_s=0.3, poll_interval_s=0.1, + clock=clock.now, sleep=clock.sleep) + + +def test_becomes_actionable_once_visible_and_stable(): + calls = {"n": 0} + + def bbox(): + calls["n"] += 1 + return (10, 10, 40, 20) if calls["n"] >= 2 else None + + report = wait_actionable(bbox, config=_config()) + assert report.actionable is True and report.reason == "actionable" + assert report.point == [30, 20] + + +def test_never_visible_times_out(): + report = wait_actionable(lambda: None, config=_config()) + assert report.actionable is False and report.reason == "not visible" + + +def test_moving_target_is_not_stable(): + counter = {"n": 0} + + def moving(): + counter["n"] += 1 + return (counter["n"], 10, 40, 20) # x shifts every poll + + report = wait_actionable(moving, config=_config()) + assert report.actionable is False and report.reason == "not stable" + + +def test_disabled_blocks(): + report = wait_actionable(lambda: (0, 0, 10, 10), enabled_probe=lambda: False, + config=_config()) + assert report.actionable is False and report.reason == "disabled" + + +def test_occluded_blocks(): + report = wait_actionable(lambda: (0, 0, 10, 10), hit_tester=lambda point: False, + config=_config()) + assert report.actionable is False and report.reason == "occluded" + + +def test_region_sampler_must_settle(): + samples = {"n": 0} + + def churning(_bbox): + samples["n"] += 1 + return samples["n"] # pixel token keeps changing + + report = wait_actionable(lambda: (0, 0, 10, 10), region_sampler=churning, + config=_config()) + assert report.actionable is False and report.reason == "not stable" + + +def test_act_when_ready_calls_action_with_center(): + clicked = [] + result = act_when_ready(lambda point: clicked.append(point) or "ok", + lambda: (10, 10, 40, 20), config=_config()) + assert result == "ok" and clicked == [[30, 20]] + + +def test_act_when_ready_raises_on_timeout(): + from je_auto_control.utils.exception.exceptions import AutoControlActionException + with pytest.raises(AutoControlActionException): + act_when_ready(lambda point: "never", lambda: None, config=_config()) + + +# --- wiring --------------------------------------------------------------- + +def test_wiring(): + assert "AC_wait_actionable" in set(ac.executor.known_commands()) + from je_auto_control.utils.mcp_server.tools import build_default_tool_registry + names = {t.name for t in build_default_tool_registry()} + assert "ac_wait_actionable" in names + from je_auto_control.gui.script_builder.command_schema import _build_specs + specs = {s.command for s in _build_specs()} + assert "AC_wait_actionable" in specs + + +def test_facade_exports(): + for attr in ("wait_actionable", "act_when_ready", "ActionabilityReport", + "GateConfig"): + assert hasattr(ac, attr) and attr in ac.__all__