From fdcf2ebf6e9ea5be568e007ce130b377c235c8ab Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Fri, 19 Jun 2026 07:51:27 +0800 Subject: [PATCH] Add native-UI element repository and flow step debugger --- README.md | 8 ++ README/README_zh-CN.md | 8 ++ README/README_zh-TW.md | 8 ++ .../Eng/doc/new_features/v12_features_doc.rst | 66 +++++++++ docs/source/Eng/eng_index.rst | 1 + .../Zh/doc/new_features/v12_features_doc.rst | 61 ++++++++ docs/source/Zh/zh_index.rst | 1 + je_auto_control/__init__.py | 6 + .../gui/script_builder/command_schema.py | 40 ++++++ .../utils/element_repository/__init__.py | 6 + .../element_repository/element_repository.py | 99 +++++++++++++ .../utils/executor/action_executor.py | 45 ++++++ .../utils/flow_debugger/__init__.py | 6 + .../utils/flow_debugger/flow_debugger.py | 130 ++++++++++++++++++ .../utils/mcp_server/tools/_factories.py | 66 +++++++++ .../utils/mcp_server/tools/_handlers.py | 31 +++++ test/unit_test/headless/test_native_batch.py | 128 +++++++++++++++++ 17 files changed, 710 insertions(+) create mode 100644 docs/source/Eng/doc/new_features/v12_features_doc.rst create mode 100644 docs/source/Zh/doc/new_features/v12_features_doc.rst create mode 100644 je_auto_control/utils/element_repository/__init__.py create mode 100644 je_auto_control/utils/element_repository/element_repository.py create mode 100644 je_auto_control/utils/flow_debugger/__init__.py create mode 100644 je_auto_control/utils/flow_debugger/flow_debugger.py create mode 100644 test/unit_test/headless/test_native_batch.py diff --git a/README.md b/README.md index 59c338b8..391fe198 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ ## Table of Contents +- [What's new (2026-06-19) — Authoring & Debugging](#whats-new-2026-06-19--authoring--debugging) - [What's new (2026-06-19) — Test & Tooling Batch](#whats-new-2026-06-19--test--tooling-batch) - [What's new (2026-06-19) — Transactional Queue](#whats-new-2026-06-19--transactional-queue) - [What's new (2026-06-19) — Unattended Reliability](#whats-new-2026-06-19--unattended-reliability) @@ -64,6 +65,13 @@ --- +## What's new (2026-06-19) — Authoring & Debugging + +Two pure-stdlib authoring-time tools, full stack (facade, `AC_*`, MCP, Script Builder). Full reference: [`docs/source/Eng/doc/new_features/v12_features_doc.rst`](docs/source/Eng/doc/new_features/v12_features_doc.rst). + +- **Element repository** — `ElementRepository` (`AC_element_save` / `AC_element_find` / `AC_element_click` / `AC_element_remove` / `AC_element_list`, `ac_element_*`): save native-UI locators under friendly names (object repository) and reuse them — `repo.click("login.submit")` instead of repeating name/role everywhere; a UI change is fixed in one place. +- **Step debugger / tracer** — `FlowDebugger` (breakpoints, `step`/`continue_`/`run_to_end`, live `variables()`) and `trace_actions` (`AC_debug_trace`, `ac_debug_trace`): step through an action list one command at a time with variables persisting across steps, or get a per-step `{index, command, result}` trace (with `dry_run` to plan without running). + ## What's new (2026-06-19) — Test & Tooling Batch Three pure-stdlib quality-of-life tools, full stack (facade, `AC_*`, MCP, Script Builder). Full reference: [`docs/source/Eng/doc/new_features/v11_features_doc.rst`](docs/source/Eng/doc/new_features/v11_features_doc.rst). diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md index e2ad4710..6388be0b 100644 --- a/README/README_zh-CN.md +++ b/README/README_zh-CN.md @@ -12,6 +12,7 @@ ## 目录 +- [本次更新 (2026-06-19) — 编写与调试](#本次更新-2026-06-19--编写与调试) - [本次更新 (2026-06-19) — 测试与工具三件套](#本次更新-2026-06-19--测试与工具三件套) - [本次更新 (2026-06-19) — 事务式工作队列](#本次更新-2026-06-19--事务式工作队列) - [本次更新 (2026-06-19) — 无人值守可靠性](#本次更新-2026-06-19--无人值守可靠性) @@ -63,6 +64,13 @@ --- +## 本次更新 (2026-06-19) — 编写与调试 + +两项纯标准库的编写期工具,走完整五层(facade、`AC_*`、MCP、Script Builder)。完整参考:[`docs/source/Zh/doc/new_features/v12_features_doc.rst`](../docs/source/Zh/doc/new_features/v12_features_doc.rst)。 + +- **元素库** — `ElementRepository`(`AC_element_save` / `AC_element_find` / `AC_element_click` / `AC_element_remove` / `AC_element_list`、`ac_element_*`):把原生 UI 定位器以友好名称存起来(object repository)重用——用 `repo.click("login.submit")` 取代到处重复 name/role;UI 变动只需改一处。 +- **步进调试器 / 追踪器** — `FlowDebugger`(断点、`step`/`continue_`/`run_to_end`、实时 `variables()`)与 `trace_actions`(`AC_debug_trace`、`ac_debug_trace`):把动作列表一次跑一个指令、变量跨步保留,或获取每步 `{index, command, result}` 追踪(用 `dry_run` 只规划不执行)。 + ## 本次更新 (2026-06-19) — 测试与工具三件套 三项纯标准库的生产力工具,走完整五层(facade、`AC_*`、MCP、Script Builder)。完整参考:[`docs/source/Zh/doc/new_features/v11_features_doc.rst`](../docs/source/Zh/doc/new_features/v11_features_doc.rst)。 diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md index 59724870..cf169a8c 100644 --- a/README/README_zh-TW.md +++ b/README/README_zh-TW.md @@ -12,6 +12,7 @@ ## 目錄 +- [本次更新 (2026-06-19) — 編寫與除錯](#本次更新-2026-06-19--編寫與除錯) - [本次更新 (2026-06-19) — 測試與工具三件套](#本次更新-2026-06-19--測試與工具三件套) - [本次更新 (2026-06-19) — 交易式工作佇列](#本次更新-2026-06-19--交易式工作佇列) - [本次更新 (2026-06-19) — 無人值守可靠性](#本次更新-2026-06-19--無人值守可靠性) @@ -63,6 +64,13 @@ --- +## 本次更新 (2026-06-19) — 編寫與除錯 + +兩項純標準庫的編寫期工具,走完整五層(facade、`AC_*`、MCP、Script Builder)。完整參考:[`docs/source/Zh/doc/new_features/v12_features_doc.rst`](../docs/source/Zh/doc/new_features/v12_features_doc.rst)。 + +- **元素庫** — `ElementRepository`(`AC_element_save` / `AC_element_find` / `AC_element_click` / `AC_element_remove` / `AC_element_list`、`ac_element_*`):把原生 UI 定位器以友善名稱存起來(object repository)重用——用 `repo.click("login.submit")` 取代到處重複 name/role;UI 變動只需改一處。 +- **步進除錯器 / 追蹤器** — `FlowDebugger`(中斷點、`step`/`continue_`/`run_to_end`、即時 `variables()`)與 `trace_actions`(`AC_debug_trace`、`ac_debug_trace`):把動作清單一次跑一個指令、變數跨步保留,或取得每步 `{index, command, result}` 追蹤(用 `dry_run` 只規劃不執行)。 + ## 本次更新 (2026-06-19) — 測試與工具三件套 三項純標準庫的生產力工具,走完整五層(facade、`AC_*`、MCP、Script Builder)。完整參考:[`docs/source/Zh/doc/new_features/v11_features_doc.rst`](../docs/source/Zh/doc/new_features/v11_features_doc.rst)。 diff --git a/docs/source/Eng/doc/new_features/v12_features_doc.rst b/docs/source/Eng/doc/new_features/v12_features_doc.rst new file mode 100644 index 00000000..bbc473c5 --- /dev/null +++ b/docs/source/Eng/doc/new_features/v12_features_doc.rst @@ -0,0 +1,66 @@ +==================================================== +New Features (2026-06-19) — Authoring & Debugging +==================================================== + +Two authoring-time tools, pure standard library and wired through the +full stack (facade, ``AC_*`` executor commands, MCP tools, Script +Builder): a native-UI **element repository** (object repository) and a +**step debugger / tracer** for action lists. + +.. contents:: + :local: + :depth: 2 + + +Element repository +================= + +Save native-UI locators under friendly names once, reuse them everywhere +— the classic RPA *object repository*. A flow references +``"login.submit"`` instead of repeating ``name="Submit", role="button"`` +at every call site, and a UI change is fixed in one place:: + + from je_auto_control import ElementRepository + + repo = ElementRepository("app.objects.json") + repo.save("login.submit", name="Submit", role="button") + repo.save("login.user", role="edit", app_name="MyApp") + + repo.click("login.submit") # resolve + click the live element + info = repo.find_info("login.user") # {found, name, role, center} + +A locator is a small set of accessibility filters (``name`` / ``role`` / +``app_name``); resolving finds the live element through the accessibility +backend. Storage is a JSON file and works on any platform; resolution +needs a platform accessibility backend. + +Executor / MCP commands: ``AC_element_save`` / ``AC_element_find`` / +``AC_element_click`` / ``AC_element_remove`` / ``AC_element_list`` (and +the matching ``ac_element_*`` MCP tools). + + +Step debugger and tracer +======================= + +Run an action list one command at a time with breakpoints, single-step, +and live variable inspection. Stepping reuses one executor instance, so +script variables (``${name}`` interpolation, ``AC_set_var`` …) persist +across steps exactly as in a normal run:: + + from je_auto_control import FlowDebugger + + dbg = FlowDebugger(actions, breakpoints=[3]) + dbg.continue_() # run until the breakpoint + dbg.variables() # inspect live variables + dbg.step() # one command at a time + dbg.run_to_end() + +The stateless one-shot form, :func:`trace_actions`, runs a list (or +``dry_run`` to only plan it) and returns a per-step trace of +``{index, command, result}`` — exposed as ``AC_debug_trace`` / +``ac_debug_trace``:: + + from je_auto_control import trace_actions + + plan = trace_actions(actions, dry_run=True) # plan without running + trace = trace_actions(actions) # run and trace diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst index 99807e44..8f5f8a02 100644 --- a/docs/source/Eng/eng_index.rst +++ b/docs/source/Eng/eng_index.rst @@ -34,6 +34,7 @@ Comprehensive guides for all AutoControl features. doc/new_features/v9_features_doc doc/new_features/v10_features_doc doc/new_features/v11_features_doc + doc/new_features/v12_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/v12_features_doc.rst b/docs/source/Zh/doc/new_features/v12_features_doc.rst new file mode 100644 index 00000000..472e75be --- /dev/null +++ b/docs/source/Zh/doc/new_features/v12_features_doc.rst @@ -0,0 +1,61 @@ +======================================== +新功能 (2026-06-19) — 編寫與除錯 +======================================== + +兩項編寫期工具,皆為純標準庫,並走完整五層(facade、``AC_*`` 執行器 +指令、MCP 工具、Script Builder):原生 UI 的**元素庫**(object +repository)與動作清單的**步進除錯器 / 追蹤器**。 + +.. contents:: + :local: + :depth: 2 + + +元素庫 +====== + +把原生 UI 定位器以友善名稱存一次、到處重用——這就是 RPA 經典的 +*object repository*。流程改以 ``"login.submit"`` 引用,而不必在每個 +呼叫點重複 ``name="Submit", role="button"``;UI 變動只需改一處:: + + from je_auto_control import ElementRepository + + repo = ElementRepository("app.objects.json") + repo.save("login.submit", name="Submit", role="button") + repo.save("login.user", role="edit", app_name="MyApp") + + repo.click("login.submit") # 解析並點擊實際元素 + info = repo.find_info("login.user") # {found, name, role, center} + +定位器是一組小的 accessibility 過濾條件(``name`` / ``role`` / +``app_name``);解析時透過 accessibility 後端找到實際元素。儲存為 JSON +檔、跨平台可用;解析需要平台的 accessibility 後端。 + +執行器 / MCP 指令:``AC_element_save`` / ``AC_element_find`` / +``AC_element_click`` / ``AC_element_remove`` / ``AC_element_list`` +(以及對應的 ``ac_element_*`` MCP 工具)。 + + +步進除錯器與追蹤器 +================== + +把動作清單一次跑一個指令,具備中斷點、單步與即時變數檢視。步進時重用 +同一個執行器實例,因此腳本變數(``${name}`` 插值、``AC_set_var`` …) +會像正常執行一樣在各步之間保留:: + + from je_auto_control import FlowDebugger + + dbg = FlowDebugger(actions, breakpoints=[3]) + dbg.continue_() # 跑到中斷點 + dbg.variables() # 檢視即時變數 + dbg.step() # 一次一個指令 + dbg.run_to_end() + +無狀態的一次性形式 :func:`trace_actions` 會跑完清單(或用 ``dry_run`` +只做規劃),回傳每步的 ``{index, command, result}`` 追蹤——對應 +``AC_debug_trace`` / ``ac_debug_trace``:: + + from je_auto_control import trace_actions + + plan = trace_actions(actions, dry_run=True) # 只規劃不執行 + trace = trace_actions(actions) # 執行並追蹤 diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst index d4be94af..b806ce3e 100644 --- a/docs/source/Zh/zh_index.rst +++ b/docs/source/Zh/zh_index.rst @@ -34,6 +34,7 @@ AutoControl 所有功能的完整使用指南。 doc/new_features/v9_features_doc doc/new_features/v10_features_doc doc/new_features/v11_features_doc + doc/new_features/v12_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 76fe4f00..9d929386 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -121,6 +121,10 @@ from je_auto_control.utils.mcp_registry import ( build_server_manifest, write_server_manifest, ) +# Named locator repository (object repository) for native UI +from je_auto_control.utils.element_repository import ElementRepository +# Step-through debugger / tracer for action lists +from je_auto_control.utils.flow_debugger import FlowDebugger, trace_actions # Background popup/interrupt watchdog (unattended automation) from je_auto_control.utils.watchdog import ( PopupWatchdog, WatchdogRule, default_popup_watchdog, @@ -525,6 +529,8 @@ def start_autocontrol_gui(*args, **kwargs): "generate_rows", "write_dataset", "rank_flows", "select_flows", "build_server_manifest", "write_server_manifest", + "ElementRepository", + "FlowDebugger", "trace_actions", # MCP server "AuditLogger", "HttpMCPServer", "MCPContent", "MCPPrompt", "MCPPromptArgument", "MCPResource", "MCPServer", "MCPTool", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index 514215f9..db22d608 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -654,6 +654,46 @@ def _add_misc_specs(specs: List[CommandSpec]) -> None: )) _add_work_queue_specs(specs) _add_tooling_specs(specs) + _add_authoring_specs(specs) + + +def _add_authoring_specs(specs: List[CommandSpec]) -> None: + path = FieldSpec("path", FieldType.FILE_PATH) + key = FieldSpec("key", FieldType.STRING) + specs.append(CommandSpec( + "AC_element_save", "Native UI", "Element: Save Locator", + fields=(path, key, + FieldSpec("name", FieldType.STRING, optional=True), + FieldSpec("role", FieldType.STRING, optional=True), + FieldSpec("app_name", FieldType.STRING, optional=True)), + description="Save a named native-UI locator (object repository).", + )) + specs.append(CommandSpec( + "AC_element_find", "Native UI", "Element: Find Saved", + fields=(path, key), + description="Resolve a saved locator to a live element summary.", + )) + specs.append(CommandSpec( + "AC_element_click", "Native UI", "Element: Click Saved", + fields=(path, key), + description="Click the element behind a saved locator.", + )) + specs.append(CommandSpec( + "AC_element_remove", "Native UI", "Element: Remove Saved", + fields=(path, key), + description="Delete a saved locator.", + )) + specs.append(CommandSpec( + "AC_element_list", "Native UI", "Element: List Saved", + fields=(path,), + description="List saved locator names in a repository file.", + )) + specs.append(CommandSpec( + "AC_debug_trace", "Flow", "Debug: Trace Actions", + fields=(FieldSpec("dry_run", FieldType.BOOL, optional=True, + default=False),), + description="Run 'actions' (JSON view) and return a per-step trace.", + )) def _add_tooling_specs(specs: List[CommandSpec]) -> None: diff --git a/je_auto_control/utils/element_repository/__init__.py b/je_auto_control/utils/element_repository/__init__.py new file mode 100644 index 00000000..93148968 --- /dev/null +++ b/je_auto_control/utils/element_repository/__init__.py @@ -0,0 +1,6 @@ +"""Named locator repository (object repository) for native-UI elements.""" +from je_auto_control.utils.element_repository.element_repository import ( + ElementRepository, +) + +__all__ = ["ElementRepository"] diff --git a/je_auto_control/utils/element_repository/element_repository.py b/je_auto_control/utils/element_repository/element_repository.py new file mode 100644 index 00000000..6826e7e7 --- /dev/null +++ b/je_auto_control/utils/element_repository/element_repository.py @@ -0,0 +1,99 @@ +"""Named locator repository (the classic RPA *object repository*). + +Save native-UI locators under friendly names once, reuse them everywhere +— so flows reference ``"login.submit"`` instead of repeating +``name="Submit", role="button"`` at every call site, and a UI change is +fixed in one place. A locator is a small set of accessibility filters +(``name`` / ``role`` / ``app_name``); resolving it finds the live element +through the accessibility backend. + +Pure standard library (JSON file storage); imports no ``PySide6``. The +accessibility backend is imported lazily so storage works on any platform. +""" +import json +from pathlib import Path +from typing import Any, Dict, List, Optional + +_FILTER_FIELDS = ("name", "role", "app_name") + + +class ElementRepository: + """A JSON-backed map of friendly name -> accessibility locator.""" + + def __init__(self, path: str) -> None: + self._path = Path(path) + self._items: Dict[str, Dict[str, str]] = self._load() + + def _load(self) -> Dict[str, Dict[str, str]]: + if not self._path.exists(): + return {} + data = json.loads(self._path.read_text(encoding="utf-8")) + if not isinstance(data, dict): + raise ValueError(f"{self._path} is not a locator map") + return {str(k): dict(v) for k, v in data.items()} + + def _flush(self) -> None: + self._path.write_text( + json.dumps(self._items, indent=2, ensure_ascii=False), + encoding="utf-8") + + def save(self, key: str, *, name: Optional[str] = None, + role: Optional[str] = None, + app_name: Optional[str] = None) -> Dict[str, str]: + """Store a locator under ``key``; needs at least one filter.""" + locator = {field: value for field, value in + (("name", name), ("role", role), ("app_name", app_name)) + if value is not None} + if not locator: + raise ValueError("a locator needs at least one of name/role/" + "app_name") + self._items[str(key)] = locator + self._flush() + return dict(locator) + + def get(self, key: str) -> Optional[Dict[str, str]]: + """Return the stored locator for ``key`` (a copy) or ``None``.""" + locator = self._items.get(str(key)) + return dict(locator) if locator is not None else None + + def remove(self, key: str) -> bool: + """Delete a locator; return whether it existed.""" + existed = str(key) in self._items + if existed: + del self._items[str(key)] + self._flush() + return existed + + def keys(self) -> List[str]: + """Return the saved locator names, sorted.""" + return sorted(self._items) + + def all(self) -> Dict[str, Dict[str, str]]: + """Return a copy of the whole repository.""" + return {key: dict(value) for key, value in self._items.items()} + + def _require(self, key: str) -> Dict[str, str]: + locator = self.get(key) + if locator is None: + raise KeyError(f"no locator named {key!r}") + return locator + + def resolve(self, key: str) -> Any: + """Find the live element for ``key`` (or ``None`` if not present).""" + from je_auto_control.utils.accessibility import ( + find_accessibility_element) + return find_accessibility_element(**self._require(key)) + + def find_info(self, key: str) -> Dict[str, Any]: + """Resolve ``key`` and return a serialisable summary.""" + element = self.resolve(key) + if element is None: + return {"found": False} + return {"found": True, "name": element.name, "role": element.role, + "center": list(element.center)} + + def click(self, key: str) -> bool: + """Click the element for ``key``; return whether it matched.""" + from je_auto_control.utils.accessibility import ( + click_accessibility_element) + return click_accessibility_element(**self._require(key)) diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index d9506479..ca9d59af 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -2388,6 +2388,45 @@ def _select_tests(flows: List[str], k: Optional[int] = None, window=int(window))} +def _element_repo(path: str): + from je_auto_control.utils.element_repository import ElementRepository + return ElementRepository(path) + + +def _element_save(path: str, key: str, name: Optional[str] = None, + role: Optional[str] = None, + app_name: Optional[str] = None) -> Dict[str, Any]: + """Adapter: save a named native-UI locator (object repository).""" + return {"locator": _element_repo(path).save( + key, name=name, role=role, app_name=app_name)} + + +def _element_find(path: str, key: str) -> Dict[str, Any]: + """Adapter: resolve a saved locator to a live element summary.""" + return _element_repo(path).find_info(key) + + +def _element_click(path: str, key: str) -> Dict[str, Any]: + """Adapter: click the element behind a saved locator.""" + return {"clicked": _element_repo(path).click(key)} + + +def _element_remove(path: str, key: str) -> Dict[str, Any]: + """Adapter: delete a saved locator.""" + return {"removed": _element_repo(path).remove(key)} + + +def _element_list(path: str) -> Dict[str, Any]: + """Adapter: list saved locator names.""" + return {"keys": _element_repo(path).keys()} + + +def _debug_trace(actions: List[Any], dry_run: bool = False) -> Dict[str, Any]: + """Adapter: run an action list and return a per-step trace.""" + from je_auto_control.utils.flow_debugger import trace_actions + return {"trace": trace_actions(actions, dry_run=bool(dry_run))} + + class Executor: """ Executor @@ -2555,6 +2594,12 @@ def __init__(self): "AC_mcp_manifest": _mcp_manifest, "AC_rank_tests": _rank_tests, "AC_select_tests": _select_tests, + "AC_element_save": _element_save, + "AC_element_find": _element_find, + "AC_element_click": _element_click, + "AC_element_remove": _element_remove, + "AC_element_list": _element_list, + "AC_debug_trace": _debug_trace, "AC_a11y_record_start": _a11y_record_start, "AC_a11y_record_stop": _a11y_record_stop, "AC_a11y_record_events": _a11y_record_events, diff --git a/je_auto_control/utils/flow_debugger/__init__.py b/je_auto_control/utils/flow_debugger/__init__.py new file mode 100644 index 00000000..2d1931cb --- /dev/null +++ b/je_auto_control/utils/flow_debugger/__init__.py @@ -0,0 +1,6 @@ +"""Step-through debugger and tracer for action lists.""" +from je_auto_control.utils.flow_debugger.flow_debugger import ( + FlowDebugger, trace_actions, +) + +__all__ = ["FlowDebugger", "trace_actions"] diff --git a/je_auto_control/utils/flow_debugger/flow_debugger.py b/je_auto_control/utils/flow_debugger/flow_debugger.py new file mode 100644 index 00000000..2f7c4365 --- /dev/null +++ b/je_auto_control/utils/flow_debugger/flow_debugger.py @@ -0,0 +1,130 @@ +"""Step-through debugger and tracer for action lists. + +Run an action list one command at a time with breakpoints, single-step, +and live variable inspection — the missing developer affordance for +authoring flows. Stepping reuses one :class:`Executor` instance so script +variables (``${name}`` interpolation, ``AC_set_var`` …) persist across +steps exactly as they would in a normal run. + +:func:`trace_actions` is the stateless one-shot form: run a list (or +``dry_run`` to only plan it) and get a per-step trace of command and +result — wired into the executor / MCP / Script Builder. + +Pure standard library; imports no ``PySide6``. +""" +from typing import Any, Dict, List, Optional + + +def _new_executor() -> Any: + """Return a fresh, isolated executor instance.""" + from je_auto_control.utils.executor.action_executor import Executor + return Executor() + + +def _command_of(action: Any) -> Optional[str]: + if isinstance(action, (list, tuple)) and action: + return action[0] if isinstance(action[0], str) else None + return None + + +class FlowDebugger: + """Step through an action list with breakpoints and variable inspection.""" + + def __init__(self, actions: List[Any], *, + breakpoints: Optional[List[int]] = None, + executor: Any = None) -> None: + self._actions = list(actions) + self._breakpoints = {int(b) for b in (breakpoints or ())} + self._executor = executor + self._index = 0 + self._record: Dict[str, Any] = {} + + def _exec(self) -> Any: + if self._executor is None: + self._executor = _new_executor() + return self._executor + + @property + def index(self) -> int: + """Index of the next action to run.""" + return self._index + + @property + def finished(self) -> bool: + """True once every action has run.""" + return self._index >= len(self._actions) + + @property + def record(self) -> Dict[str, Any]: + """A copy of the accumulated execution record.""" + return dict(self._record) + + def variables(self) -> Dict[str, Any]: + """Snapshot of the live script variables.""" + if self._executor is None: + return {} + return self._executor.variables.as_dict() + + def peek(self) -> Optional[Any]: + """Return the next action without running it (or ``None``).""" + return None if self.finished else self._actions[self._index] + + def set_breakpoint(self, index: int) -> None: + """Pause before the action at ``index``.""" + self._breakpoints.add(int(index)) + + def clear_breakpoint(self, index: int) -> None: + """Remove a breakpoint if present.""" + self._breakpoints.discard(int(index)) + + def step(self) -> Optional[Dict[str, Any]]: + """Run exactly one action; return ``{index, command, result}``.""" + if self.finished: + return None + current = self._index + action = self._actions[current] + result = self._exec().execute_action([action]) + self._record.update(result) + self._index += 1 + return {"index": current, "command": _command_of(action), + "result": next(iter(result.values()), None)} + + def continue_(self, max_steps: int = 100000) -> List[Dict[str, Any]]: + """Run until the next breakpoint or the end.""" + executed: List[Dict[str, Any]] = [] + while not self.finished and len(executed) < max_steps: + if self._index in self._breakpoints and executed: + break + executed.append(self.step()) + return executed + + def run_to_end(self) -> List[Dict[str, Any]]: + """Run every remaining action, ignoring breakpoints.""" + executed: List[Dict[str, Any]] = [] + while not self.finished: + executed.append(self.step()) + return executed + + def reset(self) -> None: + """Rewind to the start and clear record and variables.""" + self._index = 0 + self._record = {} + if self._executor is not None: + self._executor.variables.clear() + + +def trace_actions(actions: List[Any], *, dry_run: bool = False, + executor: Any = None) -> List[Dict[str, Any]]: + """Run ``actions`` and return a per-step trace. + + Each entry is ``{index, command, result}``. With ``dry_run`` the + actions are planned but not executed. + """ + runner = executor or _new_executor() + record = runner.execute_action(list(actions), dry_run=dry_run) + values = list(record.values()) + trace: List[Dict[str, Any]] = [] + for i, action in enumerate(actions): + trace.append({"index": i, "command": _command_of(action), + "result": values[i] if i < len(values) else None}) + return trace diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index 3a0f0aa6..f3e7281b 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -1795,6 +1795,71 @@ def test_selection_tools() -> List[MCPTool]: ] +def element_repository_tools() -> List[MCPTool]: + _R = {"path": {"type": "string"}, "key": {"type": "string"}} + return [ + MCPTool( + name="ac_element_save", + description=("Save a named native-UI locator (object repository): " + "store name/role/app under a friendly 'key' for " + "reuse. Needs at least one of name/role/app_name."), + input_schema=schema({ + "name": {"type": "string"}, "role": {"type": "string"}, + "app_name": {"type": "string"}, **_R}, + required=["path", "key"]), + handler=h.element_save, + annotations=SIDE_EFFECT_ONLY, + ), + MCPTool( + name="ac_element_find", + description=("Resolve a saved locator to a live element; returns " + "{found, name, role, center}."), + input_schema=schema(dict(_R), required=["path", "key"]), + handler=h.element_find, + annotations=READ_ONLY, + ), + MCPTool( + name="ac_element_click", + description="Click the element behind a saved locator.", + input_schema=schema(dict(_R), required=["path", "key"]), + handler=h.element_click, + annotations=SIDE_EFFECT_ONLY, + ), + MCPTool( + name="ac_element_remove", + description="Delete a saved locator; returns {removed}.", + input_schema=schema(dict(_R), required=["path", "key"]), + handler=h.element_remove, + annotations=SIDE_EFFECT_ONLY, + ), + MCPTool( + name="ac_element_list", + description="List saved locator names in a repository file.", + input_schema=schema({"path": {"type": "string"}}, + required=["path"]), + handler=h.element_list, + annotations=READ_ONLY, + ), + ] + + +def flow_debugger_tools() -> List[MCPTool]: + return [ + MCPTool( + name="ac_debug_trace", + description=("Run an action list and return a per-step trace " + "({index, command, result}). With dry_run=true the " + "actions are planned but not executed."), + input_schema=schema({ + "actions": {"type": "array"}, + "dry_run": {"type": "boolean"}}, + required=["actions"]), + handler=h.debug_trace, + annotations=SIDE_EFFECT_ONLY, + ), + ] + + def unattended_tools() -> List[MCPTool]: return [ MCPTool( @@ -2826,6 +2891,7 @@ def media_assert_tools() -> List[MCPTool]: scheduler_tools, trigger_tools, hotkey_tools, watchdog_tools, unattended_tools, work_queue_tools, synthetic_data_tools, mcp_registry_tools, test_selection_tools, + element_repository_tools, flow_debugger_tools, screen_record_tools, process_and_shell_tools, remote_desktop_tools, gamepad_tools, usb_passthrough_tools, assertion_tools, data_source_tools, diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index 10f02af2..f8015c33 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -857,6 +857,37 @@ def select_tests(flows, k=None, threshold=None, history_path=None, window=10): window=int(window))} +def _element_repo(path): + from je_auto_control.utils.element_repository import ElementRepository + return ElementRepository(path) + + +def element_save(path, key, name=None, role=None, app_name=None): + return {"locator": _element_repo(path).save( + key, name=name, role=role, app_name=app_name)} + + +def element_find(path, key): + return _element_repo(path).find_info(key) + + +def element_click(path, key): + return {"clicked": _element_repo(path).click(key)} + + +def element_remove(path, key): + return {"removed": _element_repo(path).remove(key)} + + +def element_list(path): + return {"keys": _element_repo(path).keys()} + + +def debug_trace(actions, dry_run=False): + from je_auto_control.utils.flow_debugger import trace_actions + return {"trace": trace_actions(actions, dry_run=bool(dry_run))} + + def vlm_locate(description: str, screen_region: Optional[List[int]] = None, model: Optional[str] = None) -> Optional[List[int]]: diff --git a/test/unit_test/headless/test_native_batch.py b/test/unit_test/headless/test_native_batch.py new file mode 100644 index 00000000..3d36e76d --- /dev/null +++ b/test/unit_test/headless/test_native_batch.py @@ -0,0 +1,128 @@ +"""Headless tests for the native-authoring batch: element repository +(object repository) and the flow step debugger / tracer. Pure stdlib; +no Qt imports.""" +from types import SimpleNamespace + +import pytest + +import je_auto_control as ac +from je_auto_control.utils.element_repository import ElementRepository +from je_auto_control.utils.flow_debugger import FlowDebugger, trace_actions + + +# --- element repository --------------------------------------------------- + +def test_repo_crud_and_persistence(tmp_path): + path = str(tmp_path / "r.json") + repo = ElementRepository(path) + repo.save("login.submit", name="Submit", role="button") + assert repo.get("login.submit") == {"name": "Submit", "role": "button"} + assert repo.keys() == ["login.submit"] + # reload from disk in a fresh instance + again = ElementRepository(path) + assert again.get("login.submit") == {"name": "Submit", "role": "button"} + assert again.remove("login.submit") is True + assert again.remove("login.submit") is False + assert again.keys() == [] + + +def test_repo_requires_a_filter(tmp_path): + repo = ElementRepository(str(tmp_path / "r.json")) + with pytest.raises(ValueError): + repo.save("empty") + + +def test_repo_resolve_and_click(tmp_path, monkeypatch): + import je_auto_control.utils.accessibility as a11y + repo = ElementRepository(str(tmp_path / "r.json")) + repo.save("btn", name="OK", role="button") + element = SimpleNamespace(name="OK", role="button", center=(10, 20)) + monkeypatch.setattr(a11y, "find_accessibility_element", + lambda **kw: element if kw.get("name") == "OK" else None) + monkeypatch.setattr(a11y, "click_accessibility_element", + lambda **kw: kw.get("name") == "OK") + assert repo.find_info("btn") == { + "found": True, "name": "OK", "role": "button", "center": [10, 20]} + assert repo.click("btn") is True + with pytest.raises(KeyError): + repo.find_info("missing") + + +# --- flow debugger -------------------------------------------------------- + +def _vars_program(): + return [["AC_set_var", {"name": "a", "value": 1}], + ["AC_set_var", {"name": "b", "value": 2}], + ["AC_inc_var", {"name": "a", "by": 10}]] + + +def test_debugger_step_and_variables(): + dbg = FlowDebugger(_vars_program()) + step = dbg.step() + assert step["index"] == 0 and step["command"] == "AC_set_var" + assert dbg.variables()["a"] == 1 + dbg.run_to_end() + assert dbg.finished + assert dbg.variables() == {"a": 11, "b": 2} + + +def test_debugger_breakpoint_pauses_then_resumes(): + dbg = FlowDebugger(_vars_program(), breakpoints=[2]) + first = dbg.continue_() + assert [s["index"] for s in first] == [0, 1] # paused before index 2 + assert dbg.index == 2 and not dbg.finished + rest = dbg.continue_() + assert [s["index"] for s in rest] == [2] + assert dbg.finished + + +def test_debugger_reset(): + dbg = FlowDebugger(_vars_program()) + dbg.run_to_end() + dbg.reset() + assert dbg.index == 0 and dbg.variables() == {} + + +def test_trace_actions_dry_run_and_real(): + real = trace_actions(_vars_program()) + assert [t["command"] for t in real] == \ + ["AC_set_var", "AC_set_var", "AC_inc_var"] + planned = trace_actions(_vars_program(), dry_run=True) + assert len(planned) == 3 + assert all("not executed" in str(t["result"]) for t in planned) + + +# --- wiring --------------------------------------------------------------- + +def test_executor_wiring(tmp_path): + path = str(tmp_path / "repo.json") + ac.execute_action([["AC_element_save", + {"path": path, "key": "ok", "name": "OK"}]]) + rec = ac.execute_action([["AC_element_list", {"path": path}]]) + assert any("ok" in str(v) for v in rec.values()) + trace_rec = ac.execute_action( + [["AC_debug_trace", + {"actions": [["AC_set_var", {"name": "x", "value": 1}]], + "dry_run": True}]]) + assert any("trace" in str(v) for v in trace_rec.values()) + known = ac.executor.known_commands() + assert {"AC_element_save", "AC_element_find", "AC_element_click", + "AC_element_remove", "AC_element_list", "AC_debug_trace"} <= known + + +def test_mcp_and_builder_wiring(): + 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_element_save", "ac_element_find", "ac_element_click", + "ac_element_remove", "ac_element_list", "ac_debug_trace"} <= names + from je_auto_control.gui.script_builder.command_schema import _build_specs + cmds = {s.command for s in _build_specs()} + assert {"AC_element_save", "AC_element_find", "AC_element_click", + "AC_element_remove", "AC_element_list", "AC_debug_trace"} <= cmds + + +def test_facade_exports(): + for attr in ("ElementRepository", "FlowDebugger", "trace_actions"): + assert hasattr(ac, attr) + assert attr in ac.__all__