diff --git a/README/WHATS_NEW_zh-CN.md b/README/WHATS_NEW_zh-CN.md index 342fe7de..d5fabbb6 100644 --- a/README/WHATS_NEW_zh-CN.md +++ b/README/WHATS_NEW_zh-CN.md @@ -1,5 +1,11 @@ # 本次更新 — AutoControl +## 本次更新 (2026-06-24) — 失败 / 无效果动作的修复策略引擎 + +当动作没效果时选择下一个修复战术——并驱动重试循环。完整参考:[`docs/source/Zh/doc/new_features/v170_features_doc.rst`](../docs/source/Zh/doc/new_features/v170_features_doc.rst)。 + +- **`plan_repair` / `next_tactic` / `run_with_repair`**(`AC_plan_repair`):`self_healing`/`locator_repair` 只修复*无法解析*的定位器;`loop_guard` 只*检测*卡住循环而无战术选择。本功能消费效果判定(例如来自 `action_effect`)并返回有序战术——`wait_retry` / `relocate` / `nudge` / `scroll_into_view` / `escalate`——接着 `run_with_repair` 以注入的 `act` / `verify` / `apply_tactic` / `verdict_for` / `sleep` 接缝驱动有界重试循环,返回 `RepairOutcome`。纯标准库状态机;不导入 `PySide6`。与 `action_effect` + `postcondition` 完成自我修正三件套。 + ## 本次更新 (2026-06-24) — 声明式动作后置条件 以 JSON 规格断言动作的预期结果,并对照 before 帧做差异。完整参考:[`docs/source/Zh/doc/new_features/v169_features_doc.rst`](../docs/source/Zh/doc/new_features/v169_features_doc.rst)。 diff --git a/README/WHATS_NEW_zh-TW.md b/README/WHATS_NEW_zh-TW.md index 2f531b98..4fe1e509 100644 --- a/README/WHATS_NEW_zh-TW.md +++ b/README/WHATS_NEW_zh-TW.md @@ -1,5 +1,11 @@ # 本次更新 — AutoControl +## 本次更新 (2026-06-24) — 失敗 / 無效果動作的修復策略引擎 + +當動作沒效果時選擇下一個修復戰術——並驅動重試迴圈。完整參考:[`docs/source/Zh/doc/new_features/v170_features_doc.rst`](../docs/source/Zh/doc/new_features/v170_features_doc.rst)。 + +- **`plan_repair` / `next_tactic` / `run_with_repair`**(`AC_plan_repair`):`self_healing`/`locator_repair` 只修復*無法解析*的定位器;`loop_guard` 只*偵測*卡住迴圈而無戰術選擇。本功能消費效果判定(例如來自 `action_effect`)並回傳有序戰術——`wait_retry` / `relocate` / `nudge` / `scroll_into_view` / `escalate`——接著 `run_with_repair` 以注入的 `act` / `verify` / `apply_tactic` / `verdict_for` / `sleep` 接縫驅動有界重試迴圈,回傳 `RepairOutcome`。純標準函式庫狀態機;不匯入 `PySide6`。與 `action_effect` + `postcondition` 完成自我修正三件套。 + ## 本次更新 (2026-06-24) — 宣告式動作後置條件 以 JSON 規格斷言動作的預期結果,並對照 before 幀做差異。完整參考:[`docs/source/Zh/doc/new_features/v169_features_doc.rst`](../docs/source/Zh/doc/new_features/v169_features_doc.rst)。 diff --git a/WHATS_NEW.md b/WHATS_NEW.md index a8721c9e..d1b4ebd6 100644 --- a/WHATS_NEW.md +++ b/WHATS_NEW.md @@ -1,5 +1,11 @@ # What's New — AutoControl +## What's new (2026-06-24) — Repair-Tactic Policy for Failed / No-Effect Actions + +Pick the next repair tactic when an action does nothing — and drive the retry loop. Full reference: [`docs/source/Eng/doc/new_features/v170_features_doc.rst`](docs/source/Eng/doc/new_features/v170_features_doc.rst). + +- **`plan_repair` / `next_tactic` / `run_with_repair`** (`AC_plan_repair`): `self_healing`/`locator_repair` only fix a locator that *didn't resolve*; `loop_guard` only *detects* a stuck loop with no tactic selection. This consumes an effect verdict (e.g. from `action_effect`) and returns the ordered tactics to try — `wait_retry` / `relocate` / `nudge` / `scroll_into_view` / `escalate` — then `run_with_repair` drives a bounded retry loop with injected `act` / `verify` / `apply_tactic` / `verdict_for` / `sleep` seams, returning a `RepairOutcome`. Pure-stdlib state machine; no `PySide6`. Completes the self-correction trio with `action_effect` + `postcondition`. + ## What's new (2026-06-24) — Declarative Action Postconditions Assert an action's expected outcome as a JSON spec, diffed against the before-frame. Full reference: [`docs/source/Eng/doc/new_features/v169_features_doc.rst`](docs/source/Eng/doc/new_features/v169_features_doc.rst). diff --git a/docs/source/Eng/doc/new_features/v170_features_doc.rst b/docs/source/Eng/doc/new_features/v170_features_doc.rst new file mode 100644 index 00000000..90bd65ff --- /dev/null +++ b/docs/source/Eng/doc/new_features/v170_features_doc.rst @@ -0,0 +1,50 @@ +Repair-Tactic Policy for Failed / No-Effect Actions +=================================================== + +When an action does nothing or lands wrong, the agent needs a *policy* for what to try next — +re-locate and retry, nudge the coordinate, scroll the target into view, wait and retry, or give +up and escalate. ``self_healing`` / ``locator_repair`` only repair a locator that *did not +resolve* (element not found); they do nothing when the element was found and clicked but had no +effect. ``loop_guard`` only *detects* a stuck loop — it has no tactic selection or backoff. +``step_repair`` is that missing controller: it consumes an effect verdict (e.g. from +``action_effect``) and drives a bounded retry loop, choosing the next untried tactic each round. + +Pure-stdlib state machine; every side effect — performing the action, verifying it, applying a +tactic, sleeping — is an injected callable, so the loop is fully deterministic and +unit-testable with no device. Imports no ``PySide6``. + +Headless API +------------ + +.. code-block:: python + + from je_auto_control import (plan_repair, run_with_repair, RepairPolicy, + classify_effect) + + # just the plan + plan_repair("no_op") # ['wait_retry', 'relocate', 'nudge'] + plan_repair("changed_elsewhere") # ['escalate'] + + # drive the loop with injected seams + outcome = run_with_repair( + act=lambda: click(*target), + verify=lambda: not is_no_op(before(), after()), + apply_tactic=apply, # e.g. relocate / nudge the target + verdict_for=lambda: classify_effect(before(), after(), action).effect, + policy=RepairPolicy(max_attempts=3)) + print(outcome.ok, outcome.tactics_used) + +``plan_repair`` returns the ordered tactics for a verdict (a string like ``no_op`` / +``changed_elsewhere`` or an ``EffectVerdict`` dict), capped at ``max_attempts``; +``next_tactic`` returns the next untried one. ``run_with_repair`` runs ``act`` then ``verify``; +on failure it applies tactics until success or exhaustion, returning a ``RepairOutcome`` +(``ok`` / ``attempts`` / ``tactics_used`` / ``detail``). ``RepairPolicy`` caps attempts and +lists the allowed tactics. + +Executor command +---------------- + +``AC_plan_repair`` (``verdict`` / ``max_attempts`` → ``{count, tactics}``) is exposed as the +MCP tool ``ac_plan_repair`` (read-only) and as the Script Builder command **Plan Repair +Tactics** under **Native UI**. (The live ``run_with_repair`` loop is driven from Python, since +it takes injected callables.) diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst index cc5733ed..e46aa880 100644 --- a/docs/source/Eng/eng_index.rst +++ b/docs/source/Eng/eng_index.rst @@ -192,6 +192,7 @@ Comprehensive guides for all AutoControl features. doc/new_features/v167_features_doc doc/new_features/v168_features_doc doc/new_features/v169_features_doc + doc/new_features/v170_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/v170_features_doc.rst b/docs/source/Zh/doc/new_features/v170_features_doc.rst new file mode 100644 index 00000000..7438d776 --- /dev/null +++ b/docs/source/Zh/doc/new_features/v170_features_doc.rst @@ -0,0 +1,45 @@ +失敗 / 無效果動作的修復策略引擎 +================================ + +當動作沒有效果或落點錯誤時,代理需要一個*策略*決定下一步該試什麼——重新定位重試、微調座標、 +把目標捲入視野、等待重試,或放棄並升級。``self_healing`` / ``locator_repair`` 只修復*無法 +解析*的定位器(找不到元素);當元素找到並點擊了卻無效果時,它們無能為力。``loop_guard`` 只 +*偵測*卡住的迴圈——沒有戰術選擇或退避。``step_repair`` 正是缺少的控制器:它消費一個效果判定 +(例如來自 ``action_effect``),並驅動有界的重試迴圈,每輪選擇下一個尚未嘗試的戰術。 + +純標準函式庫狀態機;每個副作用——執行動作、驗證、套用戰術、睡眠——都是注入的可呼叫物件, +因此迴圈完全確定性、可在無裝置下單元測試。不匯入 ``PySide6``。 + +無頭 API +-------- + +.. code-block:: python + + from je_auto_control import (plan_repair, run_with_repair, RepairPolicy, + classify_effect) + + # 只要規劃 + plan_repair("no_op") # ['wait_retry', 'relocate', 'nudge'] + plan_repair("changed_elsewhere") # ['escalate'] + + # 以注入接縫驅動迴圈 + outcome = run_with_repair( + act=lambda: click(*target), + verify=lambda: not is_no_op(before(), after()), + apply_tactic=apply, # 例如 relocate / nudge 目標 + verdict_for=lambda: classify_effect(before(), after(), action).effect, + policy=RepairPolicy(max_attempts=3)) + print(outcome.ok, outcome.tactics_used) + +``plan_repair`` 回傳某判定(字串如 ``no_op`` / ``changed_elsewhere`` 或 ``EffectVerdict`` +字典)的有序戰術,截到 ``max_attempts``;``next_tactic`` 回傳下一個尚未試過的。 +``run_with_repair`` 執行 ``act`` 然後 ``verify``;失敗時套用戰術直到成功或耗盡,回傳 +``RepairOutcome``(``ok`` / ``attempts`` / ``tactics_used`` / ``detail``)。``RepairPolicy`` +限制嘗試次數並列出允許的戰術。 + +執行器指令 +---------- + +``AC_plan_repair``(``verdict`` / ``max_attempts`` → ``{count, tactics}``)以 MCP 工具 +``ac_plan_repair``(唯讀)及 Script Builder 指令 **Plan Repair Tactics**(位於 **Native UI** +分類下)形式提供。(實際的 ``run_with_repair`` 迴圈因接受注入可呼叫物件,由 Python 驅動。) diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst index 24df00fd..a8364cd9 100644 --- a/docs/source/Zh/zh_index.rst +++ b/docs/source/Zh/zh_index.rst @@ -192,6 +192,7 @@ AutoControl 所有功能的完整使用指南。 doc/new_features/v167_features_doc doc/new_features/v168_features_doc doc/new_features/v169_features_doc + doc/new_features/v170_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 56b0c9c5..c01e1065 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -323,6 +323,10 @@ from je_auto_control.utils.postcondition import ( PostconditionReport, check_postcondition, compile_postcondition, ) +# Repair-tactic policy for failed / no-effect actions (self-correction loop) +from je_auto_control.utils.step_repair import ( + RepairOutcome, RepairPolicy, next_tactic, plan_repair, run_with_repair, +) # Locate on-screen regions by colour (mask + connected components) from je_auto_control.utils.color_region import ( find_color_region, find_color_regions, @@ -1267,6 +1271,11 @@ def start_autocontrol_gui(*args, **kwargs): "PostconditionReport", "check_postcondition", "compile_postcondition", + "RepairPolicy", + "RepairOutcome", + "plan_repair", + "next_tactic", + "run_with_repair", "find_color_region", "find_color_regions", "ssim_compare", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index a7114075..ba6bb37a 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -3205,6 +3205,15 @@ def _add_set_of_marks_specs(specs: List[CommandSpec]) -> None: ), description="Check expected outcome clauses against after/before frames.", )) + specs.append(CommandSpec( + "AC_plan_repair", "Native UI", "Plan Repair Tactics", + fields=( + FieldSpec("verdict", FieldType.STRING, + placeholder="no_op / changed_elsewhere / changed"), + FieldSpec("max_attempts", FieldType.INT, optional=True, default=3), + ), + description="Ordered repair tactics for a failed/no-effect action verdict.", + )) specs.append(CommandSpec( "AC_validate_action", "Native UI", "Validate / Snap Action", fields=( diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index 06feb5a2..b49081c6 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -4144,6 +4144,17 @@ def _check_postcondition(after: Any, spec: Any, before: Any = None) -> Dict[str, return check_postcondition(after, spec, before=before).to_dict() +def _plan_repair(verdict: Any, max_attempts: Any = 3) -> Dict[str, Any]: + """Adapter: ordered repair tactics for an effect verdict (no_op / changed_…).""" + import json + from je_auto_control.utils.step_repair import RepairPolicy, plan_repair + if isinstance(verdict, str) and verdict.strip().startswith("{"): + verdict = json.loads(verdict) + tactics = plan_repair(verdict, + policy=RepairPolicy(max_attempts=int(max_attempts))) + return {"count": len(tactics), "tactics": tactics} + + def _validate_action(action: Any, screen: Any = None, targets: Any = None) -> Dict[str, Any]: """Adapter: validate a coordinate action (bounds + optional snap-to-target).""" @@ -6025,6 +6036,7 @@ def __init__(self): "AC_classify_effect": _classify_effect, "AC_effect_near_point": _effect_near_point, "AC_check_postcondition": _check_postcondition, + "AC_plan_repair": _plan_repair, "AC_validate_action": _validate_action, "AC_replay_trace": _replay_trace, "AC_match_elements": _match_elements, diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index eb97cbbf..e76c5451 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -3395,6 +3395,20 @@ def observation_tools() -> List[MCPTool]: handler=h.check_postcondition, annotations=READ_ONLY, ), + MCPTool( + name="ac_plan_repair", + description=("Given an effect 'verdict' (string like 'no_op' / " + "'changed_elsewhere', or an EffectVerdict dict), return the " + "ordered repair tactics to try — wait_retry / relocate / " + "nudge / scroll_into_view / escalate — capped at " + "'max_attempts'. Returns {count, tactics}."), + input_schema=schema({ + "verdict": {"type": "string"}, + "max_attempts": {"type": "integer"}}, + required=["verdict"]), + handler=h.plan_repair, + 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 e846e290..bec7c7fd 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -2438,6 +2438,11 @@ def check_postcondition(after, spec, before=None): return _check_postcondition(after, spec, before) +def plan_repair(verdict, max_attempts=3): + from je_auto_control.utils.executor.action_executor import _plan_repair + return _plan_repair(verdict, max_attempts) + + def validate_action(action, screen=None, targets=None): from je_auto_control.utils.executor.action_executor import _validate_action return _validate_action(action, screen, targets) diff --git a/je_auto_control/utils/step_repair/__init__.py b/je_auto_control/utils/step_repair/__init__.py new file mode 100644 index 00000000..09875ea7 --- /dev/null +++ b/je_auto_control/utils/step_repair/__init__.py @@ -0,0 +1,9 @@ +"""Repair-tactic policy for failed / no-effect actions (self-correction loop).""" +from je_auto_control.utils.step_repair.step_repair import ( + RepairOutcome, RepairPolicy, next_tactic, plan_repair, run_with_repair, +) + +__all__ = [ + "RepairPolicy", "RepairOutcome", + "plan_repair", "next_tactic", "run_with_repair", +] diff --git a/je_auto_control/utils/step_repair/step_repair.py b/je_auto_control/utils/step_repair/step_repair.py new file mode 100644 index 00000000..66d06b49 --- /dev/null +++ b/je_auto_control/utils/step_repair/step_repair.py @@ -0,0 +1,105 @@ +"""Repair-tactic policy for failed / no-effect actions (self-correction loop). + +When an action does nothing or lands wrong, the agent needs a *policy* for what to try next — +re-locate and retry, nudge the coordinate, scroll the target into view, wait and retry, or give +up and escalate. ``self_healing`` / ``locator_repair`` only repair a locator that *did not +resolve* (element not found); they do nothing when the element was found and clicked but had no +effect. ``loop_guard`` only *detects* a stuck loop — it has no tactic selection or backoff. +``step_repair`` is that missing controller: it consumes an effect verdict (e.g. from +``action_effect``) and drives a bounded retry loop, choosing the next untried tactic each round. + +Pure-stdlib state machine; every side effect — performing the action, verifying it, applying a +tactic, sleeping — is an injected callable, so the loop is fully deterministic and +unit-testable with no device. Imports no ``PySide6``. +""" +from dataclasses import asdict, dataclass, field +from typing import Any, Callable, Dict, List, Optional, Tuple + +_DEFAULT_TACTICS = ("wait_retry", "relocate", "nudge", "scroll_into_view", "escalate") + +# which tactics make sense for a given effect verdict, best-first +_VERDICT_TACTICS: Dict[str, Tuple[str, ...]] = { + "no_op": ("wait_retry", "relocate", "nudge", "scroll_into_view"), + "changed_elsewhere": ("escalate",), + "changed": ("wait_retry", "relocate"), +} + + +@dataclass(frozen=True) +class RepairPolicy: + """How hard to try: a cap on attempts and the allowed tactics, in priority order.""" + + max_attempts: int = 3 + tactics: Tuple[str, ...] = _DEFAULT_TACTICS + + +@dataclass(frozen=True) +class RepairOutcome: + """The result of a repair loop: success, attempt count, tactics used, detail.""" + + ok: bool + attempts: int + tactics_used: List[str] = field(default_factory=list) + detail: str = "" + + def to_dict(self) -> Dict[str, Any]: + """Return the outcome as a plain dict.""" + return asdict(self) + + +def _effect_of(verdict: Any) -> str: + """Extract the effect string from a dict / EffectVerdict / bare string.""" + if isinstance(verdict, dict): + return str(verdict.get("effect", "no_op")) + return str(getattr(verdict, "effect", verdict)) + + +def plan_repair(verdict: Any, *, policy: Optional[RepairPolicy] = None) -> List[str]: + """Return the ordered repair tactics to try for ``verdict``, capped at ``max_attempts``.""" + policy = policy or RepairPolicy() + preferred = _VERDICT_TACTICS.get(_effect_of(verdict), policy.tactics) + ordered = [tactic for tactic in preferred if tactic in policy.tactics] + return (ordered or list(policy.tactics))[:int(policy.max_attempts)] + + +def next_tactic(verdict: Any, used: List[str], *, + policy: Optional[RepairPolicy] = None) -> Optional[str]: + """Return the next untried tactic for ``verdict``, or ``None`` when exhausted.""" + for tactic in plan_repair(verdict, policy=policy): + if tactic not in used: + return tactic + return None + + +def run_with_repair(act: Callable[[], Any], verify: Callable[[], bool], *, + policy: Optional[RepairPolicy] = None, + apply_tactic: Optional[Callable[[str], Any]] = None, + verdict_for: Optional[Callable[[], Any]] = None, + sleep: Optional[Callable[[float], Any]] = None) -> RepairOutcome: + """Run ``act`` then ``verify``; on failure apply repair tactics until ok or exhausted. + + Every effect is injected: ``act`` performs the action, ``verify`` returns success, + ``apply_tactic`` mutates state for a named tactic, ``verdict_for`` supplies the current + effect verdict, ``sleep`` backs off. Returns a :class:`RepairOutcome`. + """ + policy = policy or RepairPolicy() + sleeper = sleep or (lambda _seconds: None) + act() + if verify(): + return RepairOutcome(True, 1, [], "ok on first try") + used: List[str] = [] + while len(used) < int(policy.max_attempts): + tactic = next_tactic(verdict_for() if verdict_for else "no_op", used, + policy=policy) + if tactic is None: + break + used.append(tactic) + if apply_tactic is not None: + apply_tactic(tactic) + act() + if verify(): + return RepairOutcome(True, len(used) + 1, list(used), + f"recovered via {tactic}") + sleeper(0) + return RepairOutcome(False, len(used) + 1, list(used), + "exhausted repair tactics") diff --git a/test/unit_test/headless/test_step_repair_batch.py b/test/unit_test/headless/test_step_repair_batch.py new file mode 100644 index 00000000..eadadfe9 --- /dev/null +++ b/test/unit_test/headless/test_step_repair_batch.py @@ -0,0 +1,68 @@ +"""Headless tests for the repair-tactic policy / loop (pure stdlib, injected seams).""" +import je_auto_control as ac +from je_auto_control.utils.step_repair import ( + RepairPolicy, next_tactic, plan_repair, run_with_repair, +) + + +def test_plan_repair_orders_tactics_for_no_op(): + plan = plan_repair("no_op", policy=RepairPolicy(max_attempts=3)) + assert plan == ["wait_retry", "relocate", "nudge"] + + +def test_plan_repair_escalates_on_changed_elsewhere(): + assert plan_repair("changed_elsewhere") == ["escalate"] + + +def test_plan_repair_accepts_effect_verdict_dict(): + assert plan_repair({"effect": "changed_elsewhere"}) == ["escalate"] + + +def test_next_tactic_skips_used(): + assert next_tactic("no_op", ["wait_retry"]) == "relocate" + assert next_tactic("changed_elsewhere", ["escalate"]) is None + + +def test_run_with_repair_recovers_after_a_tactic(): + calls = {"act": 0} + # verify succeeds only on the 3rd act (i.e. after two repair tactics) + def act(): + calls["act"] += 1 + def verify(): + return calls["act"] >= 3 + used = [] + outcome = run_with_repair(act, verify, apply_tactic=used.append) + assert outcome.ok is True + assert outcome.attempts == 3 + assert used == ["wait_retry", "relocate"] + + +def test_run_with_repair_exhausts_and_fails(): + outcome = run_with_repair(lambda: None, lambda: False, + policy=RepairPolicy(max_attempts=2)) + assert outcome.ok is False + assert outcome.attempts == 3 and len(outcome.tactics_used) == 2 + + +def test_run_with_repair_ok_first_try(): + outcome = run_with_repair(lambda: None, lambda: True) + assert outcome.ok is True and outcome.attempts == 1 + assert outcome.tactics_used == [] + + +# --- wiring --------------------------------------------------------------- + +def test_wiring(): + assert "AC_plan_repair" 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_plan_repair" in names + from je_auto_control.gui.script_builder.command_schema import _build_specs + specs = {s.command for s in _build_specs()} + assert "AC_plan_repair" in specs + + +def test_facade_exports(): + for name in ("RepairPolicy", "RepairOutcome", "plan_repair", "next_tactic", + "run_with_repair"): + assert hasattr(ac, name) and name in ac.__all__