Add ensure_state: idempotent read-compare-act-verify for control state

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).

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.

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 介面。

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",

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=(

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"]

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)

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,

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,
+        ),
     ]
 
 

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)