From 520c0f9f7cf2e8e617a75d344def7c9a4cab7b58 Mon Sep 17 00:00:00 2001 From: QianCyrus <101633534+QianCyrus@users.noreply.github.com> Date: Mon, 28 Sep 2026 00:50:29 +0800 Subject: [PATCH] feat(desktop): add verified text input --- docs/agents.md | 14 +- evals/desktop/README.md | 31 +++ evals/desktop/build_fixture.sh | 19 ++ evals/desktop/fixture.swift | 76 ++++++++ evals/desktop/run_local.sh | 20 ++ s1a/agents/desktop.py | 96 +++++++-- s1a/desktop/driver.py | 84 ++++++-- s1a/desktop/env.py | 140 ++++++++++++-- tests/test_agents_desktop.py | 69 ++++++- tests/test_desktop_actions.py | 342 +++++++++++++++++++++++++++++++++ tests/test_desktop_driver.py | 112 ++++++++++- tests/test_desktop_env.py | 44 +++-- 12 files changed, 975 insertions(+), 72 deletions(-) create mode 100644 evals/desktop/README.md create mode 100644 evals/desktop/build_fixture.sh create mode 100644 evals/desktop/fixture.swift create mode 100644 evals/desktop/run_local.sh create mode 100644 tests/test_desktop_actions.py diff --git a/docs/agents.md b/docs/agents.md index c7ff39a..e9654ec 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -30,9 +30,21 @@ models behind them: [architecture.md](architecture.md#models). `s1a run --help` lists every flag with its default. Beyond the shared ones: `flights` and `allrecipes` take `--goal`, `--batch on|off`, `--prefetch on|off`, `--goal-values on|off`, `--profile-out` and `--logs-dir`; `desktop` takes -`--app`, `--goal`, `--expect`, `--execute`, `--plan` and `--clear`; `ticket_router` takes `--dataset` and +`--app`, `--app-path`, `--window-title`, `--goal`, `--expect`, `--execute`, `--plan`, `--clear`, `--text`, +`--text-target`, `--text-mode` and `--verify-file`; `ticket_router` takes `--dataset` and `--batch-size`; `injection_guard` takes `--labelled-set`. The four games take no flag of their own. +### Desktop text input + +`--text` supplies the content for a `type:*` action. `--text-target` selects an exact field label or native +identifier. The default `--text-mode insert` inserts at the current selection; `replace` sets the whole field. +Both require a fresh readback before the input is recorded as successful. Chinese and multiline text are +supported through native field replacement. `--window-title` selects the document when an app has several windows. + +`--verify-file` requires the expected window state and a file changed during this episode whose UTF-8 content +matches `--text`. Without `--execute`, the first decision is only recorded as a plan. +The [macOS fixture](../evals/desktop/README.md) provides a local Laya demo and a fixed-plan execution check. + ## Allrecipes `allrecipes` is the first Allrecipes task of the [WebVoyager](https://github.com/MinorJerry/WebVoyager) task set diff --git a/evals/desktop/README.md b/evals/desktop/README.md new file mode 100644 index 0000000..b1b1dbe --- /dev/null +++ b/evals/desktop/README.md @@ -0,0 +1,31 @@ +# Desktop text input on macOS + +The native fixture has a Body field and Save/Clear buttons. Save writes the current text to +`/tmp/s1a-desktop-fixture.txt`. The agent checks both the visible result and the saved file. + +Install Cua Driver and grant macOS Accessibility and Screen Recording permissions. Keep the Mac unlocked +and avoid interacting with the test window during a run. + +## Local Laya + +```bash +bash evals/desktop/run_local.sh --episodes 3 +``` + +This builds the fixture if missing and uses local Laya to choose the input and Save actions. Set `LAYA_MODEL` +to a downloaded checkpoint directory, `S1A` to an existing CLI executable, or `CUA_DRIVER_BIN` to the installed +driver. The script's default command is `uv run --extra laya s1a`. No Jev or chat-model key is needed. + +## Fixed-plan execution check + +```bash +bash evals/desktop/build_fixture.sh /tmp/S1ADocumentFixture.app +uv run s1a run desktop --model rule --rethink off --episodes 1 \ + --app S1ADocumentFixture --app-path /tmp/S1ADocumentFixture.app \ + --window-title "S1A Document Fixture" --goal "Enter the text in Body and save it" --expect Saved \ + --text "本地输入测试" --text-target Body --text-mode replace \ + --verify-file /tmp/s1a-desktop-fixture.txt --clear Clear --plan 'type:Body,Save' --execute +``` + +Omit `--execute` to preview one decision. Close the fixture before rebuilding it. Run outputs stay in the +ignored `evals/results/` directory; they are not source files. diff --git a/evals/desktop/build_fixture.sh b/evals/desktop/build_fixture.sh new file mode 100644 index 0000000..b50a98e --- /dev/null +++ b/evals/desktop/build_fixture.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +set -euo pipefail + +app="${1:-/tmp/S1ADocumentFixture.app}" +root="$(cd "$(dirname "$0")/../.." && pwd)" +mkdir -p "$app/Contents/MacOS" +swiftc -O "$root/evals/desktop/fixture.swift" -framework AppKit -o "$app/Contents/MacOS/S1ADocumentFixture" +cat > "$app/Contents/Info.plist" <<'PLIST' + + + + CFBundleExecutableS1ADocumentFixture + CFBundleIdentifierorg.thinkflowlab.s1a.document-fixture + CFBundleNameS1ADocumentFixture + CFBundlePackageTypeAPPL + LSMinimumSystemVersion14.0 + +PLIST +printf '%s\n' "$app" diff --git a/evals/desktop/fixture.swift b/evals/desktop/fixture.swift new file mode 100644 index 0000000..1abf587 --- /dev/null +++ b/evals/desktop/fixture.swift @@ -0,0 +1,76 @@ +import AppKit +import Foundation + +// A small local app for the desktop agent's macOS end-to-end test. The file is +// written only when the app's Save button is pressed through the GUI. +private let output = URL(fileURLWithPath: "/tmp/s1a-desktop-fixture.txt") + +final class DocumentFixture: NSObject, NSApplicationDelegate { + private let window = NSWindow( + contentRect: NSRect(x: 300, y: 300, width: 620, height: 440), + styleMask: [.titled, .closable, .miniaturizable, .resizable], + backing: .buffered, + defer: false + ) + private let editor = NSTextView(frame: NSRect(x: 0, y: 0, width: 580, height: 320)) + private let status = NSTextView(frame: NSRect(x: 210, y: 25, width: 360, height: 22)) + + func applicationDidFinishLaunching(_ notification: Notification) { + window.title = "S1A Document Fixture" + let content = NSView(frame: window.contentView!.bounds) + content.autoresizingMask = [.width, .height] + + let scroll = NSScrollView(frame: NSRect(x: 20, y: 75, width: 580, height: 340)) + scroll.hasVerticalScroller = true + scroll.autoresizingMask = [.width, .height] + editor.isEditable = true + editor.isSelectable = true + editor.setAccessibilityLabel("Body") + scroll.documentView = editor + content.addSubview(scroll) + + let save = NSButton(title: "Save", target: self, action: #selector(saveDocument)) + save.frame = NSRect(x: 20, y: 20, width: 80, height: 30) + content.addSubview(save) + + let clear = NSButton(title: "Clear", target: self, action: #selector(clearDocument)) + clear.frame = NSRect(x: 110, y: 20, width: 80, height: 30) + content.addSubview(clear) + + status.frame = NSRect(x: 210, y: 25, width: 360, height: 22) + status.isEditable = false + status.isSelectable = false + status.drawsBackground = false + status.string = "Editing" + status.setAccessibilityLabel("Status") + content.addSubview(status) + + window.contentView = content + window.makeKeyAndOrderFront(nil) + NSApp.activate(ignoringOtherApps: true) + } + + @objc private func saveDocument() { + do { + try editor.string.write(to: output, atomically: true, encoding: .utf8) + status.string = "Saved" + } catch { + status.string = "Save failed" + } + } + + @objc private func clearDocument() { + editor.string = "" + status.string = "Editing" + try? FileManager.default.removeItem(at: output) + } + + func applicationShouldTerminateAfterLastWindowClosed(_ sender: NSApplication) -> Bool { + true + } +} + +let app = NSApplication.shared +let delegate = DocumentFixture() +app.delegate = delegate +app.run() diff --git a/evals/desktop/run_local.sh b/evals/desktop/run_local.sh new file mode 100644 index 0000000..a377821 --- /dev/null +++ b/evals/desktop/run_local.sh @@ -0,0 +1,20 @@ +#!/usr/bin/env bash +# Runs the native document demo with local Laya; no chat or Jev calls. +set -euo pipefail +root="$(cd "$(dirname "$0")/../.." && pwd)" +cd "$root" +export PYTHONPATH="$root${PYTHONPATH:+:$PYTHONPATH}" +app="${S1A_FIXTURE_APP:-/tmp/S1ADocumentFixture.app}" +if [[ ! -x "$app/Contents/MacOS/S1ADocumentFixture" ]]; then + bash evals/desktop/build_fixture.sh "$app" +fi +if [[ -n "${S1A:-}" ]]; then + s1a_command=("$S1A") +else + s1a_command=(uv run --extra laya s1a) +fi +exec "${s1a_command[@]}" run desktop --model laya --rethink off --episodes 1 \ + --app S1ADocumentFixture --app-path "$app" --window-title "S1A Document Fixture" \ + --goal "Enter the task text in Body and save it" --expect Saved \ + --text "Local Laya agent demo" --text-target Body --text-mode replace --verify-file /tmp/s1a-desktop-fixture.txt \ + --clear Clear --execute --log "$@" diff --git a/s1a/agents/desktop.py b/s1a/agents/desktop.py index 1727a03..fa4ba0e 100644 --- a/s1a/agents/desktop.py +++ b/s1a/agents/desktop.py @@ -19,17 +19,20 @@ import argparse import asyncio import sys +import uuid from contextlib import asynccontextmanager +from pathlib import Path from typing import Any, AsyncIterator -from s1a.desktop.driver import CuaDriver, Snapshot, driver_from_env, opened -from s1a.desktop.env import ABSTAIN, DONE, WindowEnv, clickable +from s1a.desktop.driver import CuaDriver, DriverError, Snapshot, driver_from_env, opened +from s1a.desktop.env import ABSTAIN, WindowEnv, clickable from s1a.spec import Budget, Series, ToolAgentSpec RULES = ( - "A desktop app window. goal says what to do; elements lists the window's controls with their labels and values; " - "presses lists what was clicked so far. Click the one control that moves the goal forward, one click per turn. " - "When the window shows the goal's result, pick done. Pick abstain only when no offered click helps." + "A desktop app window. goal says what to do; elements lists current controls and values; presses lists past " + "actions. Pick one offered click or type action that moves the goal forward. Type uses the task's " + "supplied text, never text invented by this decision. After each action the window is observed again. " + "The environment stops automatically when the result is verified. Pick abstain only when no offered action helps." ) @@ -47,13 +50,14 @@ def parse_plan(text: str) -> tuple[tuple[str, ...], ...]: def plan_rule(plan: tuple[tuple[str, ...], ...]) -> Any: - """The baseline: the next button of the plan by how many presses were made, then done.""" + """The baseline: the next action of the plan by how many actions were made.""" def rule(state: dict[str, Any], candidates: dict[str, str]) -> str: step = len(state["presses"]) if step >= len(plan): - return DONE - return next((f"click:{label}" for label in plan[step] if f"click:{label}" in candidates), ABSTAIN) + return ABSTAIN + keys = (label if label.startswith(("click:", "type:")) else f"click:{label}" for label in plan[step]) + return next((key for key in keys if key in candidates), ABSTAIN) return rule @@ -71,26 +75,65 @@ async def launch_app(app: str, driver: CuaDriver) -> None: @asynccontextmanager -async def _session(driver: CuaDriver, app: str) -> AsyncIterator[None]: +async def _session(driver: CuaDriver, app: str, *, owner: str, title: str) -> AsyncIterator[None]: async with opened(driver): await launch_app(app, driver) + for attempt in range(20): + try: + await driver.find_window(owner, title) if title else await driver.find_window(owner) + break + except DriverError as exc: + if not str(exc).startswith("list_windows: 0 on-screen") or attempt == 19: + raise + await asyncio.sleep(0.25) yield def make_series(flags: argparse.Namespace) -> Series: - driver = driver_from_env("s1a-desktop") # raises before the series starts when the driver is missing + if flags.app_path and sys.platform == "win32": + raise ValueError("--app-path currently accepts macOS .app bundles only") plan = parse_plan(flags.plan) if flags.plan else () - return Series( - seeds=range(flags.seed, flags.seed + flags.episodes), - env_for=lambda seed: WindowEnv( + if flags.verify_file and not flags.text: + raise ValueError("--verify-file requires --text") + + def file_version() -> tuple[int, int, int] | None: + try: + stat = Path(flags.verify_file).stat() + return stat.st_ino, stat.st_mtime_ns, stat.st_ctime_ns + except OSError: + return None + + def finished(snapshot: Snapshot, initial_file: tuple[int, int, int] | None) -> bool: + if not shows(snapshot, flags.expect): + return False + if not flags.verify_file: + return True + try: + return file_version() != initial_file and Path(flags.verify_file).read_text(encoding="utf-8") == flags.text + except (OSError, UnicodeError): + return False + + driver = driver_from_env(f"s1a-desktop-{uuid.uuid4().hex[:12]}") # one public session per run/transport + + def env_for(seed: int) -> WindowEnv: + initial_file = file_version() if flags.verify_file else None + return WindowEnv( driver, app_name=flags.app, + window_title=flags.window_title, goal=flags.goal, - done_when=lambda snapshot: shows(snapshot, flags.expect), + done_when=lambda snapshot: finished(snapshot, initial_file), execute=flags.execute, clear_labels=tuple(v.strip() for v in flags.clear.split(",") if v.strip()), - ), - session=_session(driver, flags.app), + text=flags.text, + text_target=flags.text_target, + text_mode=flags.text_mode, + ) + + return Series( + seeds=range(flags.seed, flags.seed + flags.episodes), + env_for=env_for, + session=_session(driver, flags.app_path or flags.app, owner=flags.app, title=flags.window_title), baseline=("plan", plan_rule(plan)) if plan else None, annotate=lambda env, episode: None, ) @@ -98,16 +141,31 @@ def make_series(flags: argparse.Namespace) -> Series: def flags(parser: argparse.ArgumentParser) -> None: parser.add_argument("--app", required=True, help="app name (Windows Calculator / Calculator), or a Windows AUMID") + parser.add_argument("--window-title", default="", help="exact window title when the app has multiple windows") + parser.add_argument( + "--app-path", default="", help="optional macOS .app bundle path; --app remains the window owner name" + ) parser.add_argument("--goal", required=True, help="what to do in the window, read by the model on every turn") parser.add_argument("--expect", required=True, help="the text a display or label shows when the goal is met") - parser.add_argument("--execute", action="store_true", help="click for real; without it one decision is planned") - parser.add_argument("--plan", default="", help="the rule baseline: button labels in order, | between variants") + parser.add_argument("--execute", action="store_true", help="act for real; without it one decision is planned") + parser.add_argument( + "--plan", default="", help="the rule baseline: button labels or action keys in order, | between variants" + ) parser.add_argument("--clear", default="", help="button labels pressed on reset when the window has one") + parser.add_argument("--text", default="", help="task-supplied text to enter and verify in an editable element") + parser.add_argument("--text-target", default="", help="restrict --text to this exact field label or identifier") + parser.add_argument( + "--text-mode", + choices=("insert", "replace"), + default="insert", + help="insert at selection, or replace a native field's entire value; both require fresh readback", + ) + parser.add_argument("--verify-file", default="", help="also require this file's UTF-8 content to equal --text") SPEC = ToolAgentSpec( name="desktop", - description="A Windows or macOS app window through Cua Driver: click controls toward --goal until --expect appears.", + description="A Windows or macOS app window through Cua Driver: choose grounded click and type actions.", rules=RULES, budget=Budget(max_steps=12, timeout_s=90, stall_after=0), flags=flags, diff --git a/s1a/desktop/driver.py b/s1a/desktop/driver.py index f160e89..a197a99 100644 --- a/s1a/desktop/driver.py +++ b/s1a/desktop/driver.py @@ -1,5 +1,5 @@ # coding: utf-8 -"""Cua Driver over MCP stdio: one ``cua-driver mcp`` process per series, exact windows, snapshot-bound clicks. +"""Cua Driver over MCP stdio: one process per series, exact windows, snapshot-bound element actions. The driver runs in ``standard`` permission mode; every action names the pid and window id the env was bound to. """ @@ -8,6 +8,7 @@ import json import os +import re import shutil from contextlib import AsyncExitStack, asynccontextmanager from dataclasses import dataclass, field @@ -55,6 +56,8 @@ class Element: value: str token: str | None actions: tuple[str, ...] + enabled: bool = True + identifier: str = "" @dataclass(frozen=True) @@ -66,11 +69,13 @@ class Snapshot: class Driver(Protocol): - """What a desktop env needs: find the window, read it, click in it.""" + """Window-scoped observation and input used by the desktop environment.""" - async def find_window(self, app_name: str) -> Window: ... + async def find_window(self, app_name: str, window_title: str = "") -> Window: ... async def window_state(self, window: Window) -> Snapshot: ... async def click(self, window: Window, token: str) -> Json: ... + async def type_text(self, window: Window, token: str, text: str) -> Json: ... + async def set_value(self, window: Window, token: str, text: str) -> Json: ... class CuaDriver: @@ -164,8 +169,8 @@ async def _launch_target(self, app_name: str) -> Json: return {"aumid": path[len(prefix) :]} return {"launch_path": path} if path else {"name": app_name} - async def find_window(self, app_name: str) -> Window: - """Find the launched window, or the one on-screen window named ``app_name``; never guess among matches.""" + async def find_window(self, app_name: str, window_title: str = "") -> Window: + """Find one on-screen app window, using an exact title when several exist.""" listed = await self.call("list_windows", on_screen_only=True) windows = listed.get("windows") if not isinstance(windows, list): @@ -182,9 +187,12 @@ async def find_window(self, app_name: str) -> Window: matches = [ w for w in windows if str(w.get("app_name") or "").casefold() == wanted and w.get("is_on_screen", True) ] + if window_title: + matches = [w for w in matches if str(w.get("title") or "") == window_title] if len(matches) != 1: titles = [str(w.get("title")) for w in matches] - raise DriverError(f"list_windows: {len(matches)} on-screen window(s) of {app_name!r}: {titles}") + choice = f" titled {window_title!r}" if window_title else "" + raise DriverError(f"list_windows: {len(matches)} on-screen window(s) of {app_name!r}{choice}: {titles}") window = matches[0] owner = str(window.get("app_name") or app_name) if pinned is not None else app_name return Window(int(window["pid"]), int(window["window_id"]), owner, str(window.get("title") or "")) @@ -199,23 +207,75 @@ async def window_state(self, window: Window) -> Snapshot: include_screenshot=False, ) raw_elements = state.get("elements") + if str(state.get("degraded_reason", "")).startswith("ax_window_unresolved"): + raise DriverError(f"get_window_state: {state['degraded_reason']}") if not isinstance(raw_elements, list): raise DriverError(f"get_window_state: no elements in the snapshot ({state.get('degradation')!r})") - elements = tuple(_element(raw) for raw in raw_elements) + elements = tuple(_element(raw, tree=str(state.get("tree_markdown") or "")) for raw in raw_elements) snapshot_id = state.get("snapshot_id") return Snapshot(window, str(snapshot_id) if snapshot_id else None, elements, state) async def click(self, window: Window, token: str) -> Json: - """One background click on a snapshot-bound element; a refused action is an error.""" - result = await self.call( + """Click a snapshot-bound element (background by default); a refused action is an error.""" + return await self._input( "click", target=window.target, element_token=token, delivery_mode="background", session=self._label ) + + async def type_text(self, window: Window, token: str, text: str) -> Json: + """Insert task-supplied text into one element from the latest snapshot.""" + return await self._input( + "type_text", + target=window.target, + element_token=token, + text=text, + delivery_mode="background", + session=self._label, + ) + + async def set_value(self, window: Window, token: str, text: str) -> Json: + """Replace a native field through AXValue, with the driver's exact-window admission and readback.""" + return await self._input( + "set_value", + pid=window.pid, + window_id=window.window_id, + element_token=token, + value=text, + session=self._label, + ) + + async def _input(self, tool: str, **args: Any) -> Json: + result = await self.call(tool, **args) if result.get("effect") == "refused": - raise DriverError(f"click: refused ({result.get('escalation')!r})") + raise DriverError(f"{tool}: refused ({result.get('escalation')!r})") return result -def _element(raw: Any) -> Element: +def _tree_identifier(raw: Json, tree: str) -> str: + """Cua Driver 0.30.1 omits AXIdentifier from JSON but includes it in indexed tree rows. + + Consume the known label/value first: text inside a field must never be parsed as metadata. + Ambiguous or unfamiliar rows get no fallback identifier. New drivers can supply it directly. + """ + label, value = str(raw.get("label") or ""), str(raw.get("value") or "") + prefix = rf"^[ \t]*- \[{int(raw['element_index'])}\] {re.escape(str(raw.get('role') or ''))}(?= |$)" + rows = list(re.finditer(prefix, tree, re.MULTILINE)) + if len(rows) != 1: + return "" + rest = tree[rows[0].end() :] + if label and rest.startswith(f' "{label}"'): + rest = rest[len(label) + 3 :] + if rest.startswith(f' = "{value}"'): + rest = rest[len(value) + 5 :] + elif value: + return "" + if label and rest.startswith(f" ({label})"): + rest = rest[len(label) + 3 :] + rest = rest.split("\n", 1)[0] + match = re.fullmatch(r' \[id=([^\[\]\r\n]+?)(?: help="[^"\n]*")?(?: actions=\[[\w, ]*\])?\]', rest) + return match[1] if match else "" + + +def _element(raw: Any, *, tree: str = "") -> Element: if not isinstance(raw, dict) or "element_index" not in raw: raise DriverError(f"get_window_state: an element without element_index: {str(raw)[:120]!r}") token = raw.get("element_token") @@ -223,9 +283,11 @@ def _element(raw: Any) -> Element: index=int(raw["element_index"]), role=str(raw.get("role") or ""), label=str(raw.get("label") or ""), + identifier=str(raw.get("identifier") or "") or _tree_identifier(raw, tree), value=str(raw.get("value") or ""), token=str(token) if token else None, actions=tuple(str(action) for action in raw.get("actions") or ()), + enabled=raw.get("enabled") is not False, ) diff --git a/s1a/desktop/env.py b/s1a/desktop/env.py index 6328a92..4235061 100644 --- a/s1a/desktop/env.py +++ b/s1a/desktop/env.py @@ -1,33 +1,52 @@ # coding: utf-8 -"""One native window as an ``Env``: its elements are the state, one click per clickable element is a candidate.""" +"""One native window as an ``Env``: offer only actions grounded in its current snapshot.""" from __future__ import annotations from typing import Any, Callable -from s1a.desktop.driver import Driver, Element, Snapshot, Window +from s1a.desktop.driver import Driver, DriverError, Element, Snapshot, Window DONE = "done" ABSTAIN = "abstain" RESERVED = { - DONE: "The window shows the task finished; stop here.", - ABSTAIN: "No offered click moves the task forward; stop without acting.", + ABSTAIN: "No offered action moves the task forward; stop without acting.", } _CLICKABLE_ROLES = ("button", "checkbox", "radiobutton", "menubutton", "link", "popupbutton", "disclosuretriangle") +_EDITABLE_ROLES = ("textfield", "textarea", "textview", "searchfield", "combobox", "editabletext") +_MENU_ROLES = frozenset({"menu", "menubar", "menubaritem", "menuitem"}) def clickable(element: Element) -> bool: role = element.role.casefold().removeprefix("ax") - return element.token is not None and ( - role in _CLICKABLE_ROLES or any("press" in a.casefold() or "click" in a.casefold() for a in element.actions) + if not element.label.strip() or role in _MENU_ROLES: + return False + return ( + element.enabled + and element.token is not None + and ( + role in _CLICKABLE_ROLES or any("press" in a.casefold() or "click" in a.casefold() for a in element.actions) + ) ) +def editable(element: Element) -> bool: + return ( + element.enabled and element.token is not None and element.role.casefold().removeprefix("ax") in _EDITABLE_ROLES + ) + + +def observable(element: Element) -> bool: + """Keep the named controls and visible values; omit macOS menu trees from the decision state.""" + role = element.role.casefold().removeprefix("ax") + return role not in _MENU_ROLES and bool(element.label.strip() or element.value.strip()) + + class WindowEnv: - """The mechanics: bind to the app's window, offer its clickable elements, click one, re-read the window. + """Bind to one app window, offer grounded actions, then re-read it after each action. ``goal`` rides in every observation for the model. ``done_when`` reads a snapshot and says whether the task is finished; it is also the score. Without ``execute`` - the first chosen click is recorded as ``planned`` and the episode ends, so a dry run shows one decision and acts on + the first chosen action is recorded as ``planned`` and the episode ends, so a dry run shows one decision and acts on nothing. ``clear_labels`` names a button pressed on ``reset`` when the window has one (a calculator's All Clear). """ @@ -40,23 +59,43 @@ def __init__( done_when: Callable[[Snapshot], bool], execute: bool, clear_labels: tuple[str, ...], + window_title: str = "", + text: str = "", + text_target: str = "", + text_mode: str = "insert", ) -> None: self._driver = driver self._app_name = app_name + self._window_title = window_title self._goal = goal self._done_when = done_when self._execute = execute self._clear_labels = clear_labels + self._text = text + self._text_target = text_target + if text_mode not in {"insert", "replace"}: + raise ValueError("text_mode must be insert or replace") + self._text_mode = text_mode + self._verified_value: str | None = None + self._text_field: tuple[int, int, str, str] | None = None self._window: Window | None = None self._snapshot: Snapshot | None = None self._keys: dict[str, Element] = {} + self._text_keys: dict[str, Element] = {} self._presses: list[str] = [] + self._typed = False self._planned: dict[str, Any] | None = None self._ended: str | None = None async def reset(self) -> None: - self._window = await self._driver.find_window(self._app_name) - self._presses, self._planned, self._ended = [], None, None + self._window = ( + await self._driver.find_window(self._app_name, self._window_title) + if self._window_title + else await self._driver.find_window(self._app_name) + ) + self._verified_value = None + self._text_field = None + self._presses, self._planned, self._ended, self._typed = [], None, None, False await self._refresh() clear = next((e for e in self._keys.values() if e.label in self._clear_labels), None) if clear is not None and self._execute: @@ -70,21 +109,49 @@ async def reset(self) -> None: async def _refresh(self) -> None: assert self._window is not None self._snapshot = await self._driver.window_state(self._window) + self._update_candidates() + + def _update_candidates(self) -> None: self._keys = {} - for element in self._snapshot.elements: + self._text_keys = {} + for element in self._require_snapshot().elements: if clickable(element): key = f"click:{element.label or element.role}" self._keys[key if key not in self._keys else f"{key}#{element.index}"] = element + if not editable(element): + continue + locator = self._locator(element) + if locator == self._text_field: + self._typed = element.value == self._verified_value + if ( + self._text + and not self._typed + and (not self._text_target or self._text_target in (element.label, element.identifier)) + and (self._text_field is None or locator == self._text_field) + ): + name = self._text_target or element.identifier or element.label or element.role + key = f"type:{name}" + self._text_keys[key if key not in self._text_keys else f"{key}#{element.index}"] = element async def observe(self) -> dict[str, Any]: snapshot = self._require_snapshot() - values = [e.value for e in snapshot.elements if e.value and not clickable(e)] + visible = [e for e in snapshot.elements if observable(e)] + values = [e.value for e in visible if e.value and not clickable(e)] state: dict[str, Any] = { "goal": self._goal, "app": self._app_name, "title": snapshot.window.title, - "elements": [{"role": e.role, "label": e.label, "value": e.value} for e in snapshot.elements], + "elements": [ + { + "role": e.role, + "label": e.label, + "value": e.value, + **({"identifier": e.identifier} if e.identifier else {}), + } + for e in visible + ], "presses": list(self._presses), + "text_pending": bool(self._text and not self._typed), "progress": {"values": values, "presses": len(self._presses)}, } if self._planned is not None: @@ -95,20 +162,53 @@ async def candidates(self) -> dict[str, str]: if self.done: return {} offered = {key: f'{e.role} "{e.label}"' + (f" = {e.value}" if e.value else "") for key, e in self._keys.items()} + for key, element in self._text_keys.items(): + offered[key] = f'Type the task text into {element.role} "{key.removeprefix("type:")}"' return {**offered, **RESERVED} async def step(self, key: str) -> None: - if key in RESERVED: + if key == ABSTAIN: self._ended = key return - element = self._keys[key] window = self._require_snapshot().window + element = self._keys.get(key) or self._text_keys.get(key) + if element is None and key not in await self.candidates(): + raise KeyError(key) if not self._execute: - self._planned = {"key": key, "role": element.role, "label": element.label, "token": element.token} + self._planned = { + "key": key, + "role": element.role if element else None, + "label": element.label if element else None, + "token": element.token if element else None, + } self._ended = "planned" return - await self._driver.click(window, self._token(element)) - self._presses.append(element.label) + if key in self._keys: + assert element is not None + await self._driver.click(window, self._token(element)) + self._presses.append(element.label) + elif key in self._text_keys: + assert element is not None + text = self._text + locator = self._locator(element) + operation = self._driver.set_value if self._text_mode == "replace" else self._driver.type_text + result = await operation(window, self._token(element), text) + await self._refresh() + current = [e for e in self._require_snapshot().elements if self._locator(e) == locator] + if result.get("effect") != "confirmed": + raise DriverError(f"text verification failed: driver effect {result.get('effect', 'missing')}") + value = current[0].value if len(current) == 1 else None + verified = value == text + if value is not None and self._text_mode == "insert" and element.value: + verified |= value.count(text) > element.value.count(text) + if not verified: + raise DriverError("text verification failed: current field does not contain the requested content") + self._verified_value = current[0].value + self._text_field = locator + self._typed = True + self._presses.append(key) + self._update_candidates() + return await self._refresh() @property @@ -128,3 +228,7 @@ def _require_snapshot(self) -> Snapshot: if self._snapshot is None: raise RuntimeError("the window was never observed: call reset first") return self._snapshot + + def _locator(self, element: Element) -> tuple[int, int, str, str]: + window = self._require_snapshot().window + return (window.pid, window.window_id, element.role, element.identifier or element.label) diff --git a/tests/test_agents_desktop.py b/tests/test_agents_desktop.py index ba6a20c..d2149fc 100644 --- a/tests/test_agents_desktop.py +++ b/tests/test_agents_desktop.py @@ -45,16 +45,29 @@ def test_shows_reads_displays_and_labels_of_non_clickable_elements_only(self) -> self.assertTrue( desktop.shows(Snapshot(window, None, (Element(2, "AXStaticText", "PASS", "", None, ()),), {}), "PASS") ) + self.assertTrue( + desktop.shows( + Snapshot(window, None, (Element(3, "AXButton", "Saved", "", "t", (), enabled=False),), {}), "Saved" + ) + ) - def test_the_plan_rule_follows_the_labels_with_variants_then_says_done_or_abstains(self) -> None: + def test_the_plan_rule_follows_the_labels_with_variants_then_abstains(self) -> None: rule = desktop.plan_rule(desktop.parse_plan("1, 2,Multiply|×,7,Equals|=")) offered = {f"click:{label}": "" for label in ("1", "2", "×", "7", "=")} self.assertEqual( [rule({"presses": ["x"] * n}, offered) for n in range(6)], - ["click:1", "click:2", "click:×", "click:7", "click:=", "done"], + ["click:1", "click:2", "click:×", "click:7", "click:=", "abstain"], ) self.assertEqual(rule({"presses": []}, {"click:9": ""}), "abstain") + def test_the_plan_can_name_text_actions(self) -> None: + rule = desktop.plan_rule(desktop.parse_plan("type:Body,Save")) + offered = {"type:Body": "", "click:Save": ""} + self.assertEqual( + [rule({"presses": ["x"] * n}, offered) for n in range(3)], + ["type:Body", "click:Save", "abstain"], + ) + def test_the_flags_require_app_goal_and_expect_and_default_to_a_dry_run(self) -> None: args = series.parser(desktop.SPEC).parse_args( ["--model", "jev", "--rethink", "off", "--episodes", "1", *CALCULATOR] @@ -70,6 +83,58 @@ def test_without_a_plan_there_is_no_rule_baseline(self) -> None: with patch.object(desktop, "driver_from_env", lambda label: FakeCalculator()): self.assertIsNone(desktop.make_series(args).baseline) + def test_each_series_uses_a_fresh_driver_session_label(self) -> None: + args = series.parser(desktop.SPEC).parse_args( + ["--model", "rule", "--rethink", "off", "--episodes", "1", *CALCULATOR] + ) + labels: list[str] = [] + + def driver(label: str) -> FakeCalculator: + labels.append(label) + return FakeCalculator() + + with patch.object(desktop, "driver_from_env", driver): + desktop.make_series(args) + desktop.make_series(args) + self.assertEqual(len(labels), 2) + self.assertNotEqual(labels[0], labels[1]) + self.assertTrue(all(label.startswith("s1a-desktop-") for label in labels)) + + def test_app_path_is_mac_only(self) -> None: + args = series.parser(desktop.SPEC).parse_args( + ["--model", "rule", "--rethink", "off", "--episodes", "1", *CALCULATOR, "--app-path", "/tmp/Fixture.app"] + ) + with patch("sys.platform", "win32"), self.assertRaisesRegex(ValueError, "macOS"): + desktop.make_series(args) + + def test_file_verification_requires_the_visible_result_and_exact_content(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "saved.txt" + args = series.parser(desktop.SPEC).parse_args( + [ + "--model", + "rule", + "--rethink", + "off", + "--episodes", + "1", + *CALCULATOR, + "--text", + "hello", + "--verify-file", + str(path), + ] + ) + with patch.object(desktop, "driver_from_env", lambda label: FakeCalculator()): + env = desktop.make_series(args).env_for(0) + window = Window(42, 7, "Calculator", "Calculator") + shown = Snapshot(window, "s1", (Element(0, "AXStaticText", "", "84", None, ()),), {}) + self.assertFalse(env._done_when(shown)) + path.write_text("wrong", encoding="utf-8") + self.assertFalse(env._done_when(shown)) + path.write_text("hello", encoding="utf-8") + self.assertTrue(env._done_when(shown)) + async def _noop(app: str, driver: object) -> None: return None diff --git a/tests/test_desktop_actions.py b/tests/test_desktop_actions.py new file mode 100644 index 0000000..3684b76 --- /dev/null +++ b/tests/test_desktop_actions.py @@ -0,0 +1,342 @@ +# coding: utf-8 +"""Desktop input actions through a snapshot-bound fake document window.""" + +from __future__ import annotations + +import tempfile +from pathlib import Path +from typing import Any +from unittest import IsolatedAsyncioTestCase +from unittest.mock import AsyncMock, patch + +from s1a.agents import desktop +from s1a.desktop.driver import DriverError, Element, Snapshot, Window +from s1a.desktop.env import WindowEnv +from s1a.run import started_runner +from s1a.tool import loop, series + + +class FakeDocument: + def __init__(self, path: Path) -> None: + self.path = path + self.window = Window(91, 3, "Document", "Document") + self.snapshots = 0 + self.text = "" + self.status = "Editing" + self.actions: list[tuple[str, Any]] = [] + self.opened = 0 + + async def open(self) -> None: + self.opened += 1 + + async def close(self) -> None: + self.opened -= 1 + + async def find_window(self, app_name: str) -> Window: + assert app_name == "Document" + return self.window + + async def window_state(self, window: Window) -> Snapshot: + assert window == self.window + self.snapshots += 1 + n = self.snapshots + return Snapshot( + window, + f"snap-{n}", + ( + Element(1, "AXTextArea", "Body", self.text, f"body-{n}", ("AXSetValue",)), + Element(2, "AXButton", "Save", "", f"save-{n}", ("AXPress",)), + Element(3, "AXStaticText", "Status", self.status, None, ()), + ), + {}, + ) + + async def click(self, window: Window, token: str) -> dict[str, Any]: + self._check(window, token) + self.actions.append(("click", token)) + if token.startswith("save-"): + self.path.write_text(self.text, encoding="utf-8") + self.status = "Saved" + return {"effect": "confirmed"} + + async def type_text(self, window: Window, token: str, text: str) -> dict[str, Any]: + self._check(window, token) + self.actions.append(("type", token)) + self.text = text + return {"effect": "confirmed"} + + async def set_value(self, window: Window, token: str, text: str) -> dict[str, Any]: + self._check(window, token) + self.actions.append(("replace", token)) + self.text = text + return {"effect": "confirmed", "verified": True} + + def _check(self, window: Window, token: str) -> None: + if window != self.window or not token.endswith(f"-{self.snapshots}"): + raise DriverError(f"stale token {token}") + + +class TestDocumentActions(IsolatedAsyncioTestCase): + async def test_claimed_success_with_incomplete_text_is_not_recorded_as_complete(self) -> None: + async def truncated(window: Window, token: str, text: str) -> dict: + self.fake.text = text[:2] + return {"effect": "confirmed"} + + self.fake.type_text = truncated + env = self.env() + await env.reset() + with self.assertRaisesRegex(DriverError, "text verification failed"): + await env.step("type:Body") + self.assertEqual((await env.observe())["presses"], []) + self.assertTrue((await env.observe())["text_pending"]) + + async def test_insert_does_not_verify_against_a_payload_already_in_the_field(self) -> None: + self.fake.text = "hello world" + + async def partial(window: Window, token: str, text: str) -> dict: + self.fake.text = "hhello world" + return {"effect": "confirmed"} + + self.fake.type_text = partial + env = self.env() + await env.reset() + with self.assertRaisesRegex(DriverError, "text verification failed"): + await env.step("type:Body") + + async def test_replace_handles_existing_chinese_and_multiline_text_and_detects_later_edits(self) -> None: + self.fake.text = "old default" + payload = "本地桌面代理\n第二行：验证保存。" + env = WindowEnv( + self.fake, + app_name="Document", + goal="write", + done_when=lambda s: False, + execute=True, + clear_labels=(), + text=payload, + text_target="Body", + text_mode="replace", + ) + await env.reset() + await env.step("type:Body") + self.assertEqual(self.fake.text, payload) + self.assertEqual(self.fake.actions[0][0], "replace") + self.assertNotIn("type:Body", await env.candidates()) + self.fake.text = "changed by app" + await env.step("click:Save") + self.assertIn("type:Body", await env.candidates()) + self.assertTrue((await env.observe())["text_pending"]) + + async def test_unverifiable_replace_is_not_accepted_even_if_ax_value_matches(self) -> None: + original = self.fake.set_value + + async def unverifiable(window: Window, token: str, text: str) -> dict: + await original(window, token, text) + return {"effect": "unverifiable"} + + self.fake.set_value = unverifiable + env = WindowEnv( + self.fake, + app_name="Document", + goal="write", + done_when=lambda s: False, + execute=True, + clear_labels=(), + text="hello", + text_target="Body", + text_mode="replace", + ) + await env.reset() + with self.assertRaisesRegex(DriverError, "unverifiable"): + await env.step("type:Body") + self.assertEqual((await env.observe())["presses"], []) + + async def test_session_waits_for_new_window_but_does_not_retry_ambiguity(self) -> None: + driver = AsyncMock() + driver.find_window.side_effect = [DriverError("list_windows: 0 on-screen window(s)"), self.fake.window] + with patch.object(desktop, "launch_app", AsyncMock()), patch.object(desktop.asyncio, "sleep", AsyncMock()): + async with desktop._session(driver, "Document", owner="Document", title="Document"): + pass + self.assertEqual(driver.find_window.await_count, 2) + driver.find_window.reset_mock(side_effect=True) + driver.find_window.side_effect = DriverError("list_windows: 2 on-screen window(s)") + with patch.object(desktop, "launch_app", AsyncMock()), self.assertRaises(DriverError): + async with desktop._session(driver, "Document", owner="Document", title="Document"): + pass + self.assertEqual(driver.find_window.await_count, 1) + + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.path = Path(self.tmp.name) / "document.txt" + self.fake = FakeDocument(self.path) + + def env(self, *, execute: bool = True) -> WindowEnv: + return WindowEnv( + self.fake, + app_name="Document", + goal="write and save the document", + done_when=lambda snapshot: snapshot.elements[2].value == "Saved" and self.path.read_text() == "hello", + execute=execute, + clear_labels=(), + text="hello", + ) + + async def test_typing_and_save_refresh_the_snapshot_and_verify_the_file(self) -> None: + env = self.env() + await env.reset() + self.assertIn("type:Body", await env.candidates()) + await env.step("type:Body") + self.assertNotIn("type:Body", await env.candidates()) + await env.step("click:Save") + self.assertEqual(self.path.read_text(), "hello") + self.assertEqual((env.done, env.score), (True, 1.0)) + self.assertEqual(self.fake.snapshots, 3) + self.assertEqual([name for name, _ in self.fake.actions], ["type", "click"]) + self.assertEqual((await env.observe())["presses"], ["type:Body", "Save"]) + + async def test_dry_run_records_the_selected_action_without_mutation(self) -> None: + env = self.env(execute=False) + await env.reset() + await env.step("type:Body") + self.assertEqual((env.done, env.score, self.fake.actions, self.path.exists()), (True, 0.0, [], False)) + self.assertEqual((await env.observe())["planned"]["key"], "type:Body") + + async def test_explicit_text_target_excludes_other_text_areas(self) -> None: + original = self.fake.window_state + + async def with_status(window: Window) -> Snapshot: + snapshot = await original(window) + return Snapshot( + window, + snapshot.snapshot_id, + (*snapshot.elements, Element(4, "AXTextArea", "Status", "Editing", "status", ())), + {}, + ) + + self.fake.window_state = with_status + env = WindowEnv( + self.fake, + app_name="Document", + goal="write", + done_when=lambda s: False, + execute=True, + clear_labels=(), + text="hello", + text_target="Body", + ) + await env.reset() + self.assertIn("type:Body", await env.candidates()) + self.assertNotIn("type:Status", await env.candidates()) + await env.step("type:Body") + self.assertEqual(self.fake.text, "hello") + + async def test_stale_target_and_driver_failure_do_not_count_as_a_successful_step(self) -> None: + env = self.env() + await env.reset() + await self.fake.window_state(self.fake.window) # invalidates the token given to the model + with self.assertRaises(DriverError): + await env.step("type:Body") + self.assertEqual((await env.observe())["presses"], []) + self.assertFalse(self.path.exists()) + + async def test_full_agent_rule_run_writes_and_checks_the_document(self) -> None: + args = series.parser(desktop.SPEC).parse_args( + [ + "--model", + "rule", + "--rethink", + "off", + "--episodes", + "1", + "--app", + "Document", + "--goal", + "write hello in the Body and save it", + "--expect", + "Saved", + "--text", + "hello", + "--verify-file", + str(self.path), + "--plan", + "type:Body,Save", + "--execute", + ] + ) + with ( + patch.object(loop, "WORKSPACE", Path(self.tmp.name) / "ws"), + patch.object(series, "optional_chat_model", lambda: None), + patch.object(desktop, "driver_from_env", lambda label: self.fake), + patch.object(desktop, "launch_app", AsyncMock()), + ): + async with started_runner(): + result = await series.play(desktop.SPEC, args, results_dir=Path(self.tmp.name) / "results") + self.assertEqual((result["mean_score"], result["errors"], self.fake.opened), (1.0, 0, 0)) + self.assertEqual(self.path.read_text(encoding="utf-8"), "hello") + + async def test_file_verification_does_not_trust_saved_label(self) -> None: + args = series.parser(desktop.SPEC).parse_args( + [ + "--model", + "rule", + "--rethink", + "off", + "--episodes", + "1", + "--app", + "Document", + "--goal", + "save hello", + "--expect", + "Saved", + "--text", + "hello", + "--verify-file", + str(self.path), + "--execute", + ] + ) + with patch.object(desktop, "driver_from_env", lambda label: self.fake): + env = desktop.make_series(args).env_for(0) + await env.reset() + self.assertFalse(env.done) + self.fake.status = "Saved" + self.path.write_text("wrong contents", encoding="utf-8") + await env.step("type:Body") + self.assertFalse(env.done) + await env.step("click:Save") + self.assertTrue(env.done) + + async def test_existing_matching_file_does_not_count_as_a_new_save(self) -> None: + self.path.write_text("hello", encoding="utf-8") + args = series.parser(desktop.SPEC).parse_args( + [ + "--model", + "rule", + "--rethink", + "off", + "--episodes", + "1", + "--app", + "Document", + "--goal", + "save hello", + "--expect", + "Saved", + "--text", + "hello", + "--verify-file", + str(self.path), + "--execute", + ] + ) + with patch.object(desktop, "driver_from_env", lambda label: self.fake): + env = desktop.make_series(args).env_for(0) + await env.reset() + self.fake.status = "Saved" # stale status and bytes from a previous trial + await env._refresh() + self.assertFalse(env.done) + await env.step("type:Body") + await env.step("click:Save") + self.assertTrue(env.done) diff --git a/tests/test_desktop_driver.py b/tests/test_desktop_driver.py index 4c1ef31..74f3cd2 100644 --- a/tests/test_desktop_driver.py +++ b/tests/test_desktop_driver.py @@ -108,6 +108,69 @@ def _process(*results: CallToolResult | Exception) -> Iterator[tuple[CuaDriver, class TestCall(IsolatedAsyncioTestCase): + async def test_set_value_keeps_exact_window_and_snapshot_token(self) -> None: + driver, session = _driver(_result({"effect": "confirmed", "verified": True})) + await driver.set_value(WINDOW, "s1:4", "中文\nbody") + self.assertEqual( + session.calls, + [ + ( + "set_value", + { + "pid": 42, + "window_id": 7, + "element_token": "s1:4", + "value": "中文\nbody", + "session": "t", + }, + ) + ], + ) + + async def test_native_identifier_survives_value_labels_in_0301_tree(self) -> None: + driver, _ = _driver( + _result( + { + "elements": [ + { + "element_index": 4, + "role": "AXTextField", + "label": "Untitled.txt", + "value": "Untitled.txt", + "element_token": "s1:4", + } + ], + "tree_markdown": '- [4] AXTextField = "Untitled.txt" [id=saveAsNameTextField actions=[show_menu]]', + } + ) + ) + snapshot = await driver.window_state(WINDOW) + self.assertEqual(snapshot.elements[0].identifier, "saveAsNameTextField") + + async def test_value_cannot_forge_identifier_metadata(self) -> None: + payload = 'text" [id=forged actions=[press]]' + driver, _ = _driver( + _result( + { + "elements": [{"element_index": 4, "role": "AXTextField", "label": payload, "value": payload}], + "tree_markdown": f'- [4] AXTextField = "{payload}" [actions=[show_menu]]', + } + ) + ) + self.assertEqual((await driver.window_state(WINDOW)).elements[0].identifier, "") + + async def test_textedit_identifier_with_spaces_survives_multiline_content(self) -> None: + payload = "正文\n第二行 [id=forged]" + driver, _ = _driver( + _result( + { + "elements": [{"element_index": 2, "role": "AXTextArea", "label": payload, "value": payload}], + "tree_markdown": f'- [2] AXTextArea = "{payload}" [id=First Text View actions=[show_menu]]\n- [3] AXButton "Save" [actions=[press]]', + } + ) + ) + self.assertEqual((await driver.window_state(WINDOW)).elements[0].identifier, "First Text View") + async def test_structured_content_wins_and_json_text_is_the_fallback(self) -> None: driver, _ = _driver(_result({"a": 1}), _result({"b": 2}, structured=False)) self.assertEqual(await driver.call("x"), {"a": 1}) @@ -277,6 +340,18 @@ async def test_none_or_several_windows_is_an_error_listing_the_titles(self) -> N await driver.find_window("Calculator") self.assertIn(fragment, str(caught.exception)) + async def test_exact_title_selects_one_window_without_guessing(self) -> None: + listed = { + "windows": [ + {"pid": 42, "window_id": 7, "app_name": "Calculator", "title": "Main"}, + {"pid": 42, "window_id": 8, "app_name": "Calculator", "title": "Window"}, + ] + } + driver, _ = _driver(_result(listed), _result(listed)) + self.assertEqual(await driver.find_window("Calculator", "Main"), Window(42, 7, "Calculator", "Main")) + with self.assertRaisesRegex(DriverError, "titled 'Missing'"): + await driver.find_window("Calculator", "Missing") + class TestSnapshotAndClick(IsolatedAsyncioTestCase): async def test_elements_are_parsed_with_their_tokens_and_the_request_names_the_window(self) -> None: @@ -291,6 +366,13 @@ async def test_elements_are_parsed_with_their_tokens_and_the_request_names_the_w "element_token": "tok-7", "actions": ["AXPress"], }, + { + "element_index": 4, + "role": "AXButton", + "label": "Saved", + "enabled": False, + "element_token": "tok-status", + }, ], } driver, session = _driver(_result(state)) @@ -298,7 +380,11 @@ async def test_elements_are_parsed_with_their_tokens_and_the_request_names_the_w self.assertEqual(snapshot.snapshot_id, "snap-1") self.assertEqual( snapshot.elements, - (Element(0, "AXStaticText", "", "84", None, ()), Element(3, "AXButton", "7", "", "tok-7", ("AXPress",))), + ( + Element(0, "AXStaticText", "", "84", None, ()), + Element(3, "AXButton", "7", "", "tok-7", ("AXPress",)), + Element(4, "AXButton", "Saved", "", "tok-status", (), enabled=False), + ), ) ((name, args),) = session.calls self.assertEqual(name, "get_window_state") @@ -338,6 +424,30 @@ async def test_a_click_is_background_on_the_token_and_a_refusal_is_an_error(self await driver.click(WINDOW, "tok-7") self.assertIn("refused", str(caught.exception)) + async def test_text_input_keeps_the_exact_window_and_session(self) -> None: + driver, session = _driver(_result({"effect": "confirmed"})) + await driver.type_text(WINDOW, "tok-field", "hello") + self.assertEqual( + session.calls, + [ + ( + "type_text", + { + "target": WINDOW.target, + "element_token": "tok-field", + "text": "hello", + "delivery_mode": "background", + "session": "t", + }, + ) + ], + ) + + async def test_a_refused_text_action_fails(self) -> None: + driver, _ = _driver(_result({"effect": "refused", "escalation": {"recommended": "foreground"}})) + with self.assertRaises(DriverError): + await driver.type_text(WINDOW, "tok-field", "hello") + class TestFromEnv(TestCase): def test_no_binary_names_the_installer(self) -> None: diff --git a/tests/test_desktop_env.py b/tests/test_desktop_env.py index d5135b4..6912da6 100644 --- a/tests/test_desktop_env.py +++ b/tests/test_desktop_env.py @@ -1,12 +1,12 @@ # coding: utf-8 -"""``WindowEnv`` over the fake calculator: candidates, snapshot-bound clicks, done and score, dry run, reserved keys.""" +"""``WindowEnv`` over the fake calculator: candidates, snapshot-bound clicks, done and score, dry run.""" from __future__ import annotations from unittest import IsolatedAsyncioTestCase -from s1a.desktop.driver import Snapshot -from s1a.desktop.env import ABSTAIN, DONE, WindowEnv +from s1a.desktop.driver import Element, Snapshot +from s1a.desktop.env import ABSTAIN, DONE, WindowEnv, clickable, observable from support_desktop import FakeCalculator @@ -26,20 +26,26 @@ def _env(fake: FakeCalculator, *, execute: bool = True) -> WindowEnv: class TestWindowEnv(IsolatedAsyncioTestCase): - async def test_candidates_are_the_clickable_elements_plus_done_and_abstain(self) -> None: + async def test_menu_items_and_unlabelled_window_controls_are_not_candidates(self) -> None: + self.assertFalse(clickable(Element(1, "AXMenuItem", "Shut Down", "", "token", ("AXPress",)))) + self.assertFalse(clickable(Element(2, "AXButton", "", "", "token", ("AXPress",)))) + self.assertFalse(clickable(Element(4, "AXButton", "Saved", "", "token", ("AXPress",), enabled=False))) + self.assertTrue(clickable(Element(3, "AXButton", "Save", "", "token", ("AXPress",)))) + self.assertFalse(observable(Element(1, "AXMenuItem", "Shut Down", "", "token", ("AXPress",)))) + self.assertFalse(observable(Element(2, "AXButton", "", "", "token", ("AXPress",)))) + self.assertTrue(observable(Element(4, "AXTextArea", "Status", "Saved", "token", ()))) + + async def test_candidates_are_the_clickable_elements_plus_abstain_until_success(self) -> None: env = _env(FakeCalculator()) await env.reset() offered = await env.candidates() self.assertEqual(offered["click:7"], 'AXButton "7"') - self.assertEqual( - (offered[DONE], offered[ABSTAIN]), - ( - "The window shows the task finished; stop here.", - "No offered click moves the task forward; stop without acting.", - ), - ) + self.assertNotIn(DONE, offered) # a premature done cannot be a valid model choice + self.assertEqual(offered[ABSTAIN], "No offered action moves the task forward; stop without acting.") + with self.assertRaises(KeyError): + await env.step(DONE) self.assertNotIn("click:keypad", offered) # no press action - self.assertEqual(len(offered), 13 + 2) + self.assertEqual(len(offered), 13 + 1) async def test_a_step_clicks_the_token_of_the_latest_snapshot_in_the_bound_window_and_re_reads(self) -> None: fake = FakeCalculator() @@ -60,14 +66,12 @@ async def test_a_step_clicks_the_token_of_the_latest_snapshot_in_the_bound_windo await env.step("click:Equals") self.assertEqual((env.done, env.score, await env.candidates()), (True, 1.0, {})) - async def test_done_and_abstain_end_the_episode_without_a_click(self) -> None: - for key in (DONE, ABSTAIN): - with self.subTest(key=key): - fake = FakeCalculator() - env = _env(fake) - await env.reset() - await env.step(key) - self.assertEqual((env.done, env.score, len(fake.clicks)), (True, 0.0, 1)) + async def test_abstain_ends_the_episode_without_a_click(self) -> None: + fake = FakeCalculator() + env = _env(fake) + await env.reset() + await env.step(ABSTAIN) + self.assertEqual((env.done, env.score, len(fake.clicks)), (True, 0.0, 1)) async def test_a_dry_run_plans_one_click_and_ends_without_touching_the_window(self) -> None: fake = FakeCalculator()