Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions WHATS_NEW.md
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down
48 changes: 48 additions & 0 deletions docs/source/Eng/doc/new_features/v213_features_doc.rst
Original file line number Diff line number Diff line change
@@ -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**.
42 changes: 42 additions & 0 deletions docs/source/Zh/doc/new_features/v213_features_doc.rst
Original file line number Diff line number Diff line change
@@ -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** 分類下)形式提供。
3 changes: 3 additions & 0 deletions je_auto_control/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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",
Expand Down
22 changes: 22 additions & 0 deletions je_auto_control/gui/script_builder/command_schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -4359,9 +4359,9 @@
"AC_wait_for_unlock", "Shell", "Wait for Unlock",
fields=(
FieldSpec("timeout", FieldType.FLOAT, optional=True, default=30.0,
placeholder="timeout seconds"),

Check failure on line 4362 in je_auto_control/gui/script_builder/command_schema.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Define a constant instead of duplicating this literal "timeout seconds" 3 times.

See more on https://sonarcloud.io/project/issues?id=Integration-Automation_AutoControlGUI&issues=AZ8BBnhFd6uwXB75T_7i&open=AZ8BBnhFd6uwXB75T_7i&pullRequest=440
FieldSpec("interval", FieldType.FLOAT, optional=True, default=0.5,
placeholder="poll interval seconds"),

Check failure on line 4364 in je_auto_control/gui/script_builder/command_schema.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Define a constant instead of duplicating this literal "poll interval seconds" 3 times.

See more on https://sonarcloud.io/project/issues?id=Integration-Automation_AutoControlGUI&issues=AZ8BBnhFd6uwXB75T_7j&open=AZ8BBnhFd6uwXB75T_7j&pullRequest=440
),
description="Block until the session is unlocked or timeout.",
))
Expand Down Expand Up @@ -4500,6 +4500,28 @@
),
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=(
Expand Down
6 changes: 6 additions & 0 deletions je_auto_control/utils/app_idle/__init__.py
Original file line number Diff line number Diff line change
@@ -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"]
102 changes: 102 additions & 0 deletions je_auto_control/utils/app_idle/app_idle.py
Original file line number Diff line number Diff line change
@@ -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()
18 changes: 18 additions & 0 deletions je_auto_control/utils/executor/action_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down
24 changes: 24 additions & 0 deletions je_auto_control/utils/mcp_server/tools/_factories.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
),
]


Expand Down
12 changes: 12 additions & 0 deletions je_auto_control/utils/mcp_server/tools/_handlers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Loading
Loading