From 5ca1c5f3f05e172402f6cdbc8df9e3645d9721ca Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Fri, 26 Jun 2026 06:45:05 +0800 Subject: [PATCH] Add ensure_state: idempotent read-compare-act-verify for control state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Acting unconditionally double-toggles an already-checked box or re-enters an already-correct field and can't be safely re-run. ensure_state reads first and only acts (then re-reads to verify) when the state differs; ensure_toggle is the boolean flip specialization. A control already in the desired state is left untouched, so the call is idempotent. Distinct from idempotency (request-key cache) — this converges device state. --- WHATS_NEW.md | 6 + .../doc/new_features/v212_features_doc.rst | 52 ++++++++ .../Zh/doc/new_features/v212_features_doc.rst | 42 +++++++ je_auto_control/__init__.py | 3 + .../gui/script_builder/command_schema.py | 16 +++ .../utils/ensure_state/__init__.py | 6 + .../utils/ensure_state/ensure_state.py | 66 +++++++++++ .../utils/executor/action_executor.py | 19 +++ .../utils/mcp_server/tools/_factories.py | 17 +++ .../utils/mcp_server/tools/_handlers.py | 9 ++ .../headless/test_ensure_state_batch.py | 112 ++++++++++++++++++ 11 files changed, 348 insertions(+) create mode 100644 docs/source/Eng/doc/new_features/v212_features_doc.rst create mode 100644 docs/source/Zh/doc/new_features/v212_features_doc.rst create mode 100644 je_auto_control/utils/ensure_state/__init__.py create mode 100644 je_auto_control/utils/ensure_state/ensure_state.py create mode 100644 test/unit_test/headless/test_ensure_state_batch.py diff --git a/WHATS_NEW.md b/WHATS_NEW.md index 531ded96..8ea46692 100644 --- a/WHATS_NEW.md +++ b/WHATS_NEW.md @@ -2,6 +2,12 @@ ## What's new (2026-06-26) +### 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). + +- **`ensure_state` / `ensure_toggle`** (`AC_ensure_field_value`): automation that acts *unconditionally* double-toggles a box that was already checked or re-enters an already-correct field, and can't be safely re-run. The robust shape is read-compare-act-verify. `ensure_state` reads via an injectable `reader`, and only if it doesn't equal `desired` applies `setter` and re-reads (up to `attempts`); `ensure_toggle` is the boolean specialization that calls `toggle` only while the state differs. A control already in the desired state is left untouched (`changed=False`) — idempotent and safe to re-run, distinct from `idempotency` (a request-key replay cache) since this converges *device state*. The executor's `AC_ensure_field_value` idempotently sets a native control's value via the accessibility backend. Fourth feature of the ROUND-15 input-fidelity lane. No `PySide6`. + ### Adaptive Timeout from Observed Durations Stop guessing wait timeouts — learn them from how long the step actually takes. Full reference: [`docs/source/Eng/doc/new_features/v211_features_doc.rst`](docs/source/Eng/doc/new_features/v211_features_doc.rst). diff --git a/docs/source/Eng/doc/new_features/v212_features_doc.rst b/docs/source/Eng/doc/new_features/v212_features_doc.rst new file mode 100644 index 00000000..e1833d8a --- /dev/null +++ b/docs/source/Eng/doc/new_features/v212_features_doc.rst @@ -0,0 +1,52 @@ +Ensure a Control Is in the Desired State (Idempotent) +===================================================== + +Automation that *acts unconditionally* — "click the checkbox", "type the value" +— double-toggles a box that was already checked, or re-enters a field that was +already correct, and can't be safely re-run. The robust shape is +read-compare-act-verify: look at the current state, do nothing if it already +matches, otherwise apply the change and confirm it took. ``ensure_state`` is +that primitive. + +* :func:`ensure_state` — generic: read via ``reader``, and if it doesn't equal + ``desired`` apply ``setter`` and re-read, up to ``attempts`` times. +* :func:`ensure_toggle` — the boolean specialization for a stateless flip: read + ``is_on`` and call ``toggle`` only while it differs from ``desired``. + +A control already in the desired state is left untouched (``changed=False``), so +the call is idempotent and safe to re-run. This is distinct from +:mod:`idempotency` (a request-key replay cache) — ``ensure_state`` converges +*device state*, not call results. The reader / setter / toggle seams are +injectable, so the logic is fully testable without a real control. Imports no +``PySide6``. + +Headless API +------------ + +.. code-block:: python + + from je_auto_control import ensure_state, ensure_toggle + + # Idempotently make a setting "on" — no write if it already is + ensure_state("on", reader=read_combo, setter=write_combo) + # -> {'ok': True, 'changed': False, 'value': 'on', 'attempts': 0} + + # Flip a checkbox to checked only if it isn't already + ensure_toggle(True, is_on=is_checked, toggle=click_checkbox) + +Both return ``{ok, changed, value, attempts}``: ``changed`` tells you whether an +action was actually performed (useful for "did I have to fix this?" reporting), +and ``ok`` whether the desired state was reached within ``attempts``. Pass a +custom ``equals`` to :func:`ensure_state` for case-insensitive or normalized +comparisons. + +Executor commands +----------------- + +``AC_ensure_field_value`` (``desired`` + ``name`` / ``role`` / ``app_name`` / +``automation_id`` / ``attempts`` → ``{ok, changed, value, attempts}``) +idempotently sets a native control's value through the accessibility backend — +reading first and doing nothing if it already matches. It is the matching +``ac_ensure_field_value`` MCP tool and a Script Builder command under **Flow**. +:func:`ensure_state` / :func:`ensure_toggle` (which take arbitrary callables) are +the Python-API surface. diff --git a/docs/source/Zh/doc/new_features/v212_features_doc.rst b/docs/source/Zh/doc/new_features/v212_features_doc.rst new file mode 100644 index 00000000..65bf8a34 --- /dev/null +++ b/docs/source/Zh/doc/new_features/v212_features_doc.rst @@ -0,0 +1,42 @@ +確保控制項處於目標狀態(冪等) +============================== + +*無條件採取行動*的自動化——「點選核取方塊」、「輸入該值」——會把已勾選的方塊再次切換,或對已正確的 +欄位重新輸入,且無法安全地重跑。穩健的型態是讀取-比較-行動-驗證:看目前狀態,若已相符就什麼都不做, +否則套用變更並確認生效。``ensure_state`` 正是此原語。 + +* :func:`ensure_state` ——通用:透過 ``reader`` 讀取,若不等於 ``desired`` 就套用 ``setter`` 並 + 重讀,最多 ``attempts`` 次。 +* :func:`ensure_toggle` ——針對無狀態翻轉的布林特化:讀取 ``is_on``,僅在與 ``desired`` 不同時 + 呼叫 ``toggle``。 + +已處於目標狀態的控制項會保持不動(``changed=False``),故此呼叫是冪等且可安全重跑。這有別於 +:mod:`idempotency`(請求鍵重放快取)——``ensure_state`` 收斂的是*裝置狀態*,而非呼叫結果。 +reader / setter / toggle 接縫皆可注入,故邏輯能在沒有真實控制項的情況下完整測試。不匯入 ``PySide6``。 + +無頭 API +-------- + +.. code-block:: python + + from je_auto_control import ensure_state, ensure_toggle + + # 冪等地把某設定設為 "on" ——若已是則不寫入 + ensure_state("on", reader=read_combo, setter=write_combo) + # -> {'ok': True, 'changed': False, 'value': 'on', 'attempts': 0} + + # 僅在尚未勾選時把核取方塊翻為勾選 + ensure_toggle(True, is_on=is_checked, toggle=click_checkbox) + +兩者皆回傳 ``{ok, changed, value, attempts}``:``changed`` 告訴你是否實際執行了動作 +(對「我是否得修正它?」的報告很有用),``ok`` 則是是否在 ``attempts`` 內達到目標狀態。 +對 :func:`ensure_state` 傳入自訂 ``equals`` 可做不分大小寫或正規化比較。 + +執行器指令 +---------- + +``AC_ensure_field_value``(``desired`` 加上 ``name`` / ``role`` / ``app_name`` / +``automation_id`` / ``attempts`` → ``{ok, changed, value, attempts}``)透過無障礙後端 +冪等地設定原生控制項的值——先讀取,若已相符則不做任何事。以對應的 ``ac_ensure_field_value`` +MCP 工具及 Script Builder 指令(位於 **Flow** 分類下)形式提供。:func:`ensure_state` / +:func:`ensure_toggle`(接受任意 callable)則是 Python API 介面。 diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index 8e202f2f..3225251c 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -127,6 +127,8 @@ from je_auto_control.utils.adaptive_timeout import ( recommend_timeout, timeout_stats, ) +# Idempotently bring a control / setting to a desired state +from je_auto_control.utils.ensure_state import ensure_state, ensure_toggle # 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, @@ -1749,6 +1751,7 @@ def start_autocontrol_gui(*args, **kwargs): "RetryBudget", "run_with_budget", "backoff_delay", "jittered_delay", "compare_field_value", "verify_field_value", "fill_and_verify", "recommend_timeout", "timeout_stats", + "ensure_state", "ensure_toggle", "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 e1d51a07..20ccfcc0 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -4484,6 +4484,22 @@ def _add_work_queue_specs(specs: List[CommandSpec]) -> None: ), description="Timeout recommendation plus percentiles and clamp flags.", )) + specs.append(CommandSpec( + "AC_ensure_field_value", "Flow", "Ensure Field Value", + fields=( + FieldSpec("desired", FieldType.STRING, placeholder="desired value"), + FieldSpec("name", FieldType.STRING, optional=True, + placeholder="control name"), + FieldSpec("role", FieldType.STRING, optional=True, + placeholder="control role"), + FieldSpec("app_name", FieldType.STRING, optional=True, + placeholder="app name"), + FieldSpec("automation_id", FieldType.STRING, optional=True, + placeholder="automation id"), + FieldSpec("attempts", FieldType.INT, optional=True, default=2), + ), + description="Idempotently set a control's value (no-op if already set).", + )) specs.append(CommandSpec( "AC_normalize_ext", "Shell", "Normalize Extension", fields=( diff --git a/je_auto_control/utils/ensure_state/__init__.py b/je_auto_control/utils/ensure_state/__init__.py new file mode 100644 index 00000000..68137be8 --- /dev/null +++ b/je_auto_control/utils/ensure_state/__init__.py @@ -0,0 +1,6 @@ +"""Idempotently bring a control or setting to a desired state.""" +from je_auto_control.utils.ensure_state.ensure_state import ( + ensure_state, ensure_toggle, +) + +__all__ = ["ensure_state", "ensure_toggle"] diff --git a/je_auto_control/utils/ensure_state/ensure_state.py b/je_auto_control/utils/ensure_state/ensure_state.py new file mode 100644 index 00000000..0a703590 --- /dev/null +++ b/je_auto_control/utils/ensure_state/ensure_state.py @@ -0,0 +1,66 @@ +"""Idempotently bring a control or setting to a desired state: read, act, verify. + +Automation that *acts unconditionally* — "click the checkbox", "type the value" +— double-toggles a box that was already checked, or re-enters a field that was +already correct, and can't be safely re-run. The robust shape is +read-compare-act-verify: look at the current state, do nothing if it already +matches, otherwise apply the change and confirm it took. ``ensure_state`` is +that primitive. + +* :func:`ensure_state` — generic: read via ``reader``, and if it doesn't equal + ``desired`` apply ``setter`` and re-read, up to ``attempts`` times. +* :func:`ensure_toggle` — the boolean specialization for a stateless flip: read + ``is_on`` and call ``toggle`` only while it differs from ``desired``. + +A control already in the desired state is left untouched (``changed=False``), +so the call is idempotent and safe to re-run. The reader / setter / toggle seams +are injectable, so the logic is fully testable without a real control. Distinct +from :mod:`idempotency` (a request-key replay cache) — this converges *device +state*, not call results. Imports no ``PySide6``. +""" +from typing import Any, Callable, Dict + +StateReader = Callable[[], Any] +StateSetter = Callable[[Any], None] +StateEquals = Callable[[Any, Any], bool] + + +def _default_equals(left: Any, right: Any) -> bool: + """Default equality used to decide whether the state already matches.""" + return left == right + + +def ensure_state(desired: Any, *, reader: StateReader, setter: StateSetter, + equals: StateEquals = _default_equals, + attempts: int = 2) -> Dict[str, Any]: + """Bring the state read by ``reader`` to ``desired`` via ``setter`` (idempotent). + + Reads the current state; if ``equals(current, desired)`` it returns + immediately with ``changed=False``. Otherwise it applies ``setter(desired)`` + and re-reads, up to ``attempts`` times. Returns + ``{ok, changed, value, attempts}``. + """ + current = reader() + if equals(current, desired): + return {"ok": True, "changed": False, "value": current, "attempts": 0} + used = 0 + for used in range(1, max(1, int(attempts)) + 1): + setter(desired) + current = reader() + if equals(current, desired): + return {"ok": True, "changed": True, "value": current, + "attempts": used} + return {"ok": False, "changed": True, "value": current, "attempts": used} + + +def ensure_toggle(desired: bool, *, is_on: Callable[[], bool], + toggle: Callable[[], None], + attempts: int = 2) -> Dict[str, Any]: + """Bring a boolean toggle to ``desired`` by flipping only while it differs. + + ``is_on`` reads the current boolean; ``toggle`` performs a stateless flip and + is called only when the state does not already match ``desired``. Returns + ``{ok, changed, value, attempts}``. + """ + return ensure_state(bool(desired), reader=lambda: bool(is_on()), + setter=lambda _desired: toggle(), attempts=attempts) diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index 4183e25a..411872a6 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -2805,6 +2805,24 @@ def _timeout_stats(durations: Any, percentile_q: Any = 95.0, factor: Any = 1.5, max_s=float(max_s)) +def _ensure_field_value(desired: Any, name: Optional[str] = None, + role: Optional[str] = None, + app_name: Optional[str] = None, + automation_id: Optional[str] = None, + attempts: Any = 2) -> Dict[str, Any]: + """Adapter: idempotently set a native control's value and verify (read-act).""" + from je_auto_control.utils.ensure_state import ensure_state + return ensure_state( + str(desired), + reader=lambda: _control_get_value(name=name, role=role, + app_name=app_name, + automation_id=automation_id), + setter=lambda value: _control_set_value(value, name=name, role=role, + app_name=app_name, + automation_id=automation_id), + attempts=int(attempts)) + + 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 @@ -6833,6 +6851,7 @@ def __init__(self): "AC_verify_field_value": _verify_field_value, "AC_adaptive_timeout": _adaptive_timeout, "AC_timeout_stats": _timeout_stats, + "AC_ensure_field_value": _ensure_field_value, "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 74f7c236..5b64f920 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -1857,6 +1857,23 @@ def smart_wait_tools() -> List[MCPTool]: handler=h.timeout_stats, annotations=READ_ONLY, ), + MCPTool( + name="ac_ensure_field_value", + description=("Idempotently set a native control's value to " + "'desired' and verify: reads first and does nothing if " + "it already matches. Identify the control by name / " + "role / app_name / automation_id. Returns {ok, " + "changed, value, attempts}."), + input_schema=schema({"desired": {"type": "string"}, + "name": {"type": "string"}, + "role": {"type": "string"}, + "app_name": {"type": "string"}, + "automation_id": {"type": "string"}, + "attempts": {"type": "integer"}}, + required=["desired"]), + handler=h.ensure_field_value, + annotations=SIDE_EFFECT_ONLY, + ), ] diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index 96138a70..4a278bd9 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -712,6 +712,15 @@ def timeout_stats(durations, percentile_q=95.0, factor=1.5, min_s=1.0, return _timeout_stats(durations, percentile_q, factor, min_s, max_s) +def ensure_field_value(desired, name=None, role=None, app_name=None, + automation_id=None, attempts=2): + from je_auto_control.utils.executor.action_executor import ( + _ensure_field_value, + ) + return _ensure_field_value(desired, name, role, app_name, automation_id, + attempts) + + 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_ensure_state_batch.py b/test/unit_test/headless/test_ensure_state_batch.py new file mode 100644 index 00000000..487ce62b --- /dev/null +++ b/test/unit_test/headless/test_ensure_state_batch.py @@ -0,0 +1,112 @@ +"""Headless tests for ensure_state (injected reader / setter / toggle).""" +import je_auto_control as ac +from je_auto_control.utils.ensure_state import ensure_state, ensure_toggle + + +class Cell: + """A tiny mutable state holder for reader / setter seams.""" + + def __init__(self, value): + self.value = value + self.sets = 0 + + def read(self): + return self.value + + def write(self, new): + self.sets += 1 + self.value = new + + +# --- ensure_state --------------------------------------------------------- + +def test_ensure_state_noop_when_already_matches(): + cell = Cell("on") + result = ensure_state("on", reader=cell.read, setter=cell.write) + assert result == {"ok": True, "changed": False, "value": "on", + "attempts": 0} + assert cell.sets == 0 # idempotent: setter never called + + +def test_ensure_state_sets_when_different(): + cell = Cell("off") + result = ensure_state("on", reader=cell.read, setter=cell.write) + assert result["ok"] is True + assert result["changed"] is True + assert result["value"] == "on" + assert result["attempts"] == 1 + assert cell.sets == 1 + + +def test_ensure_state_gives_up_when_setter_ineffective(): + # setter that never actually changes the value + reads = {"n": 0} + + def stubborn_set(_value): + reads["n"] += 1 + + result = ensure_state("on", reader=lambda: "off", setter=stubborn_set, + attempts=3) + assert result["ok"] is False + assert result["changed"] is True + assert result["attempts"] == 3 + assert reads["n"] == 3 + + +def test_ensure_state_custom_equals(): + cell = Cell("ON") + # case-insensitive equality means "on" already matches "ON" + result = ensure_state("on", reader=cell.read, setter=cell.write, + equals=lambda a, b: a.lower() == b.lower()) + assert result["changed"] is False + assert cell.sets == 0 + + +# --- ensure_toggle -------------------------------------------------------- + +def test_ensure_toggle_noop_when_already_desired(): + state = {"on": True} + flips = {"n": 0} + + def toggle(): + flips["n"] += 1 + state["on"] = not state["on"] + + result = ensure_toggle(True, is_on=lambda: state["on"], toggle=toggle) + assert result["changed"] is False + assert flips["n"] == 0 + + +def test_ensure_toggle_flips_once_to_reach_desired(): + state = {"on": False} + flips = {"n": 0} + + def toggle(): + flips["n"] += 1 + state["on"] = not state["on"] + + result = ensure_toggle(True, is_on=lambda: state["on"], toggle=toggle) + assert result["ok"] is True + assert result["changed"] is True + assert result["value"] is True + assert flips["n"] == 1 + + +# --- wiring --------------------------------------------------------------- + +def test_wiring(): + known = set(ac.executor.known_commands()) + assert "AC_ensure_field_value" in 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_ensure_field_value" in names + from je_auto_control.gui.script_builder.command_schema import _build_specs + specs = {s.command for s in _build_specs()} + assert "AC_ensure_field_value" in specs + + +def test_facade_exports(): + for name in ("ensure_state", "ensure_toggle"): + assert hasattr(ac, name) and name in ac.__all__