diff --git a/WHATS_NEW.md b/WHATS_NEW.md index 8ea46692..2732b174 100644 --- a/WHATS_NEW.md +++ b/WHATS_NEW.md @@ -2,6 +2,12 @@ ## What's new (2026-06-26) +### Wait Until the App Is Idle + +Hold off the next click until the busy/wait cursor settles — don't act mid-churn. Full reference: [`docs/source/Eng/doc/new_features/v213_features_doc.rst`](docs/source/Eng/doc/new_features/v213_features_doc.rst). + +- **`wait_until_app_idle` / `idle_point`** (`AC_wait_until_app_idle`, `AC_idle_point`): a click fired while the app is still churning (busy cursor up, dialog mid-paint, long handler running) is dropped or mis-targeted. `smart_waits` watches *pixels* settle; this watches the app's *busy signal* settle, which is cheaper and survives animated-but-idle UI. It reuses `settle_detector.SettleTracker` — each poll feeds 1.0 when busy / 0.0 when idle, and returns once the app has read idle for `quiet_samples` polls in a row (a busy spike resets the run). `wait_until_app_idle` polls an injectable `busy_probe` (default = Windows busy/app-starting cursor) with injectable `clock`/`sleep`; `idle_point` is the pure analyser over a recorded busy/idle trace. Fully testable without an app. Fifth feature of the ROUND-15 input-fidelity lane. No `PySide6`. + ### Ensure a Control Is in the Desired State (Idempotent) Read-compare-act-verify instead of acting blind — don't double-toggle an already-checked box. Full reference: [`docs/source/Eng/doc/new_features/v212_features_doc.rst`](docs/source/Eng/doc/new_features/v212_features_doc.rst). diff --git a/docs/source/Eng/doc/new_features/v213_features_doc.rst b/docs/source/Eng/doc/new_features/v213_features_doc.rst new file mode 100644 index 00000000..b2f17e7e --- /dev/null +++ b/docs/source/Eng/doc/new_features/v213_features_doc.rst @@ -0,0 +1,48 @@ +Wait Until the App Is Idle +========================== + +A click fired while the app is still churning — the busy / wait cursor is up, a +dialog is mid-paint, a long handler is running — is dropped or mis-targeted. +``smart_waits`` watches *pixels* settling; ``app_idle`` watches the app's *busy +signal* settle instead, which is cheaper and survives animated-but-idle UI. It +reuses :class:`settle_detector.SettleTracker`: each poll feeds ``1.0`` when busy +and ``0.0`` when idle, and the wait returns once the app has read idle for +``quiet_samples`` polls in a row (a fresh busy spike resets the run). + +* :func:`wait_until_app_idle` — poll a ``busy_probe`` until the app settles idle + or a timeout, with injectable ``clock`` / ``sleep`` / ``busy_probe``. +* :func:`idle_point` — pure: the index in a recorded busy/idle sample series at + which it first becomes settled-idle. + +The default probe reports the Windows busy / app-starting cursor; every wait and +settle decision runs through the injectable seam, so it is fully testable +without an app. Imports no ``PySide6``. + +Headless API +------------ + +.. code-block:: python + + from je_auto_control import wait_until_app_idle, idle_point + + # Launch something, then wait for its busy cursor to settle before clicking + start_exe("setup.exe") + if wait_until_app_idle(quiet_samples=3, timeout_s=30)["idle"]: + click_next() + + # Pure: analyse a recorded busy/idle trace + idle_point([True, True, False, False, False], quiet_samples=3) # 4 + +``wait_until_app_idle`` returns ``{idle, polls, quiet_run, elapsed_s}``. Pass a +custom ``busy_probe`` (a ``() -> bool``) to gate on any busy signal — a spinner +image match, a process-CPU threshold, an accessibility "busy" flag — not just the +cursor. + +Executor commands +----------------- + +``AC_wait_until_app_idle`` (``quiet_samples`` / ``timeout`` / ``interval`` → +``{idle, polls, quiet_run, elapsed_s}``, using the Windows busy cursor) and +``AC_idle_point`` (``busy_samples`` JSON list + ``quiet_samples`` → ``{index}``, +pure). They are the matching read-only ``ac_*`` MCP tools and Script Builder +commands under **Flow**. diff --git a/docs/source/Zh/doc/new_features/v213_features_doc.rst b/docs/source/Zh/doc/new_features/v213_features_doc.rst new file mode 100644 index 00000000..29b00cba --- /dev/null +++ b/docs/source/Zh/doc/new_features/v213_features_doc.rst @@ -0,0 +1,42 @@ +等待應用程式閒置 +================ + +在應用程式仍在忙碌時觸發的點擊——忙碌 / 等待游標出現、對話框正在繪製、長處理程序正在執行—— +會被丟棄或誤點。``smart_waits`` 看*像素*安定;``app_idle`` 改看應用程式的*忙碌訊號*安定, +這更省成本且能在「有動畫但已閒置」的 UI 下運作。它重用 :class:`settle_detector.SettleTracker`: +每次輪詢在忙碌時餵入 ``1.0``、閒置時餵入 ``0.0``,當應用程式連續 ``quiet_samples`` 次讀到閒置 +即返回(新的忙碌尖峰會重置該連續計數)。 + +* :func:`wait_until_app_idle` ——輪詢 ``busy_probe`` 直到應用程式安定閒置或逾時,``clock`` / + ``sleep`` / ``busy_probe`` 皆可注入。 +* :func:`idle_point` ——純函式:在已記錄的忙碌/閒置取樣序列中,首次變為安定閒置的索引。 + +預設 probe 回報 Windows 的忙碌 / 應用程式啟動游標;每個等待與安定決策都透過可注入接縫執行, +故能在沒有應用程式的情況下完整測試。不匯入 ``PySide6``。 + +無頭 API +-------- + +.. code-block:: python + + from je_auto_control import wait_until_app_idle, idle_point + + # 啟動某程式,點擊前先等其忙碌游標安定 + start_exe("setup.exe") + if wait_until_app_idle(quiet_samples=3, timeout_s=30)["idle"]: + click_next() + + # 純函式:分析已記錄的忙碌/閒置軌跡 + idle_point([True, True, False, False, False], quiet_samples=3) # 4 + +``wait_until_app_idle`` 回傳 ``{idle, polls, quiet_run, elapsed_s}``。傳入自訂 ``busy_probe`` +(一個 ``() -> bool``)可對任何忙碌訊號設閘——spinner 影像比對、行程 CPU 門檻、無障礙「忙碌」旗標—— +不限於游標。 + +執行器指令 +---------- + +``AC_wait_until_app_idle``(``quiet_samples`` / ``timeout`` / ``interval`` → +``{idle, polls, quiet_run, elapsed_s}``,使用 Windows 忙碌游標)與 +``AC_idle_point``(``busy_samples`` JSON 清單加上 ``quiet_samples`` → ``{index}``,純函式)。 +皆以對應的唯讀 ``ac_*`` MCP 工具及 Script Builder 指令(位於 **Flow** 分類下)形式提供。 diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index 3225251c..21dd9129 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -129,6 +129,8 @@ ) # Idempotently bring a control / setting to a desired state from je_auto_control.utils.ensure_state import ensure_state, ensure_toggle +# Wait until an application stops being busy before the next step +from je_auto_control.utils.app_idle import idle_point, wait_until_app_idle # Rich clipboard formats — RTF + CSV/TSV codecs and Windows get / set from je_auto_control.utils.clipboard_rich_formats import ( build_rtf, csv_to_rows, get_clipboard_csv, get_clipboard_rtf, rows_to_csv, @@ -1752,6 +1754,7 @@ def start_autocontrol_gui(*args, **kwargs): "compare_field_value", "verify_field_value", "fill_and_verify", "recommend_timeout", "timeout_stats", "ensure_state", "ensure_toggle", + "wait_until_app_idle", "idle_point", "build_rtf", "rtf_to_text", "rows_to_csv", "csv_to_rows", "set_clipboard_rtf", "get_clipboard_rtf", "set_clipboard_csv", "get_clipboard_csv", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index 20ccfcc0..96092eb9 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -4500,6 +4500,28 @@ def _add_work_queue_specs(specs: List[CommandSpec]) -> None: ), description="Idempotently set a control's value (no-op if already set).", )) + specs.append(CommandSpec( + "AC_wait_until_app_idle", "Flow", "Wait Until App Idle", + fields=( + FieldSpec("quiet_samples", FieldType.INT, optional=True, + default=3, placeholder="consecutive idle polls"), + FieldSpec("timeout", FieldType.FLOAT, optional=True, default=10.0, + placeholder="timeout seconds"), + FieldSpec("interval", FieldType.FLOAT, optional=True, default=0.1, + placeholder="poll interval seconds"), + ), + description="Block until the app's busy cursor settles idle or timeout.", + )) + specs.append(CommandSpec( + "AC_idle_point", "Flow", "Idle Point", + fields=( + FieldSpec("busy_samples", FieldType.STRING, + placeholder="JSON list of busy booleans"), + FieldSpec("quiet_samples", FieldType.INT, optional=True, + default=3), + ), + description="Index where a busy/idle series first settles idle.", + )) specs.append(CommandSpec( "AC_normalize_ext", "Shell", "Normalize Extension", fields=( diff --git a/je_auto_control/utils/app_idle/__init__.py b/je_auto_control/utils/app_idle/__init__.py new file mode 100644 index 00000000..b6756650 --- /dev/null +++ b/je_auto_control/utils/app_idle/__init__.py @@ -0,0 +1,6 @@ +"""Wait until an application stops being busy before driving the next step.""" +from je_auto_control.utils.app_idle.app_idle import ( + idle_point, wait_until_app_idle, +) + +__all__ = ["wait_until_app_idle", "idle_point"] diff --git a/je_auto_control/utils/app_idle/app_idle.py b/je_auto_control/utils/app_idle/app_idle.py new file mode 100644 index 00000000..e42528ec --- /dev/null +++ b/je_auto_control/utils/app_idle/app_idle.py @@ -0,0 +1,102 @@ +"""Wait until an application stops being busy before driving the next step. + +A click fired while the app is still churning — the busy / wait cursor is up, a +dialog is mid-paint, a long handler is running — is dropped or mis-targeted. +``smart_waits`` watches *pixels* settling; this watches the app's *busy signal* +settle instead, which is cheaper and survives animated-but-idle UI. It reuses +:class:`settle_detector.SettleTracker`: each poll feeds ``1.0`` when busy and +``0.0`` when idle, and the wait returns once the app has read idle for +``quiet_samples`` polls in a row (a fresh busy spike resets the run). + +* :func:`wait_until_app_idle` — poll a ``busy_probe`` until the app settles idle + or a timeout, with injectable ``clock`` / ``sleep`` / ``busy_probe``. +* :func:`idle_point` — pure: the index in a recorded busy/idle sample series at + which it first becomes settled-idle. + +The default probe reports the Windows busy / app-starting cursor; every wait and +settle decision runs through the injectable seam, so it is fully testable +without an app. Imports no ``PySide6``. +""" +import sys +import time +from typing import Any, Callable, Dict, Optional, Sequence + +from je_auto_control.utils.settle_detector import SettleTracker, settle_point + +# A busy probe returns truthy while the application is busy. +BusyProbe = Callable[[], bool] + +# Windows busy cursors (IDC_WAIT / IDC_APPSTARTING). +_IDC_WAIT = 32514 +_IDC_APPSTARTING = 32650 + + +def idle_point(busy_samples: Sequence[Any], *, + quiet_samples: int = 3) -> Optional[int]: + """Index at which a busy/idle sample series first settles idle (pure). + + Each truthy sample is "busy"; the series settles once it has read idle for + ``quiet_samples`` in a row. Returns ``None`` if it never settles. + """ + churns = [1.0 if bool(sample) else 0.0 for sample in busy_samples] + return settle_point(churns, quiet_samples=int(quiet_samples), max_churn=0.0) + + +def wait_until_app_idle(*, busy_probe: Optional[BusyProbe] = None, + quiet_samples: int = 3, timeout_s: float = 10.0, + interval_s: float = 0.1, + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep + ) -> Dict[str, Any]: + """Block until the app reads idle for ``quiet_samples`` polls, or timeout. + + Returns ``{idle, polls, quiet_run, elapsed_s}``. ``clock`` / ``sleep`` / + ``busy_probe`` are injectable for deterministic tests; the default probe + reports the Windows busy cursor. + """ + probe = busy_probe if busy_probe is not None else _default_busy_probe + tracker = SettleTracker(quiet_samples=int(quiet_samples), max_churn=0.0) + start = clock() + deadline = start + float(timeout_s) + polls = 0 + quiet_run = 0 + while True: + polls += 1 + state = tracker.update(1.0 if bool(probe()) else 0.0) + quiet_run = state.quiet_run + if state.settled: + return {"idle": True, "polls": polls, "quiet_run": quiet_run, + "elapsed_s": round(clock() - start, 4)} + if clock() >= deadline: + return {"idle": False, "polls": polls, "quiet_run": quiet_run, + "elapsed_s": round(clock() - start, 4)} + sleep(float(interval_s)) + + +def _cursor_is_busy() -> bool: + """Return True when the Windows global cursor is a busy / app-starting one.""" + import ctypes + + class _POINT(ctypes.Structure): + _fields_ = [("x", ctypes.c_long), ("y", ctypes.c_long)] + + class _CURSORINFO(ctypes.Structure): + _fields_ = [("cbSize", ctypes.c_uint), ("flags", ctypes.c_uint), + ("hCursor", ctypes.c_void_p), ("ptScreenPos", _POINT)] + + user32 = ctypes.windll.user32 + info = _CURSORINFO() + info.cbSize = ctypes.sizeof(_CURSORINFO) + if not user32.GetCursorInfo(ctypes.byref(info)): + return False + busy = {user32.LoadCursorW(None, _IDC_WAIT), + user32.LoadCursorW(None, _IDC_APPSTARTING)} + return info.hCursor in busy + + +def _default_busy_probe() -> bool: + """Default busy probe: the Windows busy cursor; raise on other platforms.""" + if not sys.platform.startswith("win"): + raise RuntimeError( + "app-idle has no OS busy probe on this platform; pass busy_probe=") + return _cursor_is_busy() diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index 411872a6..56f4305b 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -2823,6 +2823,22 @@ def _ensure_field_value(desired: Any, name: Optional[str] = None, attempts=int(attempts)) +def _wait_until_app_idle(quiet_samples: Any = 3, timeout: Any = 10.0, + interval: Any = 0.1) -> Dict[str, Any]: + """Adapter: block until the foreground app's busy cursor settles or timeout.""" + from je_auto_control.utils.app_idle import wait_until_app_idle + return wait_until_app_idle(quiet_samples=int(quiet_samples), + timeout_s=float(timeout), + interval_s=float(interval)) + + +def _idle_point(busy_samples: Any, quiet_samples: Any = 3) -> Dict[str, Any]: + """Adapter: index where a busy/idle sample series first settles idle (pure).""" + from je_auto_control.utils.app_idle import idle_point + samples = _coerce_list(busy_samples) if busy_samples else [] + return {"index": idle_point(samples, quiet_samples=int(quiet_samples))} + + def _normalize_ext(target: str) -> Dict[str, Any]: """Adapter: the lowercased extension of a path / bare ext (pure).""" from je_auto_control.utils.file_assoc import normalize_ext @@ -6852,6 +6868,8 @@ def __init__(self): "AC_adaptive_timeout": _adaptive_timeout, "AC_timeout_stats": _timeout_stats, "AC_ensure_field_value": _ensure_field_value, + "AC_wait_until_app_idle": _wait_until_app_idle, + "AC_idle_point": _idle_point, "AC_normalize_ext": _normalize_ext, "AC_file_association": _file_association, "AC_get_control_text": _get_control_text, diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index 5b64f920..91e56b8b 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -1874,6 +1874,30 @@ def smart_wait_tools() -> List[MCPTool]: handler=h.ensure_field_value, annotations=SIDE_EFFECT_ONLY, ), + MCPTool( + name="ac_wait_until_app_idle", + description=("Block until the foreground app's busy / wait cursor " + "settles idle for 'quiet_samples' polls (or 'timeout' " + "seconds). Returns {idle, polls, quiet_run, " + "elapsed_s} (Windows)."), + input_schema=schema({"quiet_samples": {"type": "integer"}, + "timeout": {"type": "number"}, + "interval": {"type": "number"}}), + handler=h.wait_until_app_idle, + annotations=READ_ONLY, + ), + MCPTool( + name="ac_idle_point", + description=("Index in a recorded busy/idle 'busy_samples' series " + "at which it first settles idle for 'quiet_samples' in " + "a row (pure). Returns {index} (null if never)."), + input_schema=schema({"busy_samples": {"type": "array", + "items": {"type": "boolean"}}, + "quiet_samples": {"type": "integer"}}, + required=["busy_samples"]), + handler=h.idle_point, + annotations=READ_ONLY, + ), ] diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index 4a278bd9..ef5debeb 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -721,6 +721,18 @@ def ensure_field_value(desired, name=None, role=None, app_name=None, attempts) +def wait_until_app_idle(quiet_samples=3, timeout=10.0, interval=0.1): + from je_auto_control.utils.executor.action_executor import ( + _wait_until_app_idle, + ) + return _wait_until_app_idle(quiet_samples, timeout, interval) + + +def idle_point(busy_samples, quiet_samples=3): + from je_auto_control.utils.executor.action_executor import _idle_point + return _idle_point(busy_samples, quiet_samples) + + def normalize_ext(target): from je_auto_control.utils.executor.action_executor import _normalize_ext return _normalize_ext(target) diff --git a/test/unit_test/headless/test_app_idle_batch.py b/test/unit_test/headless/test_app_idle_batch.py new file mode 100644 index 00000000..5db32bd0 --- /dev/null +++ b/test/unit_test/headless/test_app_idle_batch.py @@ -0,0 +1,88 @@ +"""Headless tests for app_idle (settle gate over an injected busy probe).""" +import je_auto_control as ac +from je_auto_control.utils.app_idle import idle_point, wait_until_app_idle + + +# --- pure idle_point ------------------------------------------------------ + +def test_idle_point_after_busy_run(): + # busy, busy, idle, idle, idle -> settles at index 4 (3 quiet in a row) + samples = [True, True, False, False, False] + assert idle_point(samples, quiet_samples=3) == 4 + + +def test_idle_point_resets_on_spike(): + # idle, idle, BUSY, idle, idle, idle -> the spike resets the quiet run + samples = [False, False, True, False, False, False] + assert idle_point(samples, quiet_samples=3) == 5 + + +def test_idle_point_never_settles(): + assert idle_point([True, True, True], quiet_samples=2) is None + + +def test_idle_point_immediate_when_quiet_one(): + assert idle_point([False], quiet_samples=1) == 0 + + +# --- wait_until_app_idle -------------------------------------------------- + +def _probe_sequence(values): + state = {"i": 0} + + def probe(): + i = min(state["i"], len(values) - 1) + state["i"] += 1 + return values[i] + + return probe + + +def test_wait_until_app_idle_settles(): + # busy then idle x3 -> idle=True + probe = _probe_sequence([True, False, False, False]) + clock = iter([0.0, 0.0, 1.0, 2.0, 3.0, 4.0]) + result = wait_until_app_idle(busy_probe=probe, quiet_samples=3, + timeout_s=100.0, interval_s=1.0, + clock=lambda: next(clock), + sleep=lambda _s: None) + assert result["idle"] is True + assert result["quiet_run"] >= 3 + + +def test_wait_until_app_idle_times_out_when_busy(): + probe = _probe_sequence([True]) # always busy + ticks = iter([0.0, 1.0, 2.0, 6.0, 7.0]) + result = wait_until_app_idle(busy_probe=probe, quiet_samples=3, + timeout_s=5.0, interval_s=1.0, + clock=lambda: next(ticks), + sleep=lambda _s: None) + assert result["idle"] is False + + +# --- wiring --------------------------------------------------------------- + +def test_executor_pure_idle_point(): + from je_auto_control.utils.executor.action_executor import _idle_point + out = _idle_point([True, False, False, False], 3) + assert out["index"] == 3 + # accepts a JSON-list string (Script Builder text field) + assert _idle_point("[false, false]", 2)["index"] == 1 + + +def test_wiring(): + known = set(ac.executor.known_commands()) + assert {"AC_wait_until_app_idle", "AC_idle_point"} <= known + 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_until_app_idle", "ac_idle_point"} <= names + from je_auto_control.gui.script_builder.command_schema import _build_specs + specs = {s.command for s in _build_specs()} + assert {"AC_wait_until_app_idle", "AC_idle_point"} <= specs + + +def test_facade_exports(): + for name in ("wait_until_app_idle", "idle_point"): + assert hasattr(ac, name) and name in ac.__all__