fix: separate Cua keyboard targeting and verify browser actions
This commit is contained in:
chenmingxuan committed
1 parent
1c1d700e9b
commit
9c98b6bf8a
10 files changed
+591
-69
No files matched your search
@@ -0,0 +1,463 @@
|
||||
"""Opt-in Cua browser regression: real keyboard, form, scrolling and evidence.
|
||||
|
||||
Uses a disposable browser profile and a loopback event oracle, no CDP/DOM control.
|
||||
Only the launched browser process tree is cleaned up. No user tabs are changed.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import uuid
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
import jsonschema
|
||||
import psutil
|
||||
from artemis.plugins.models import ServerConfig
|
||||
from artemis.plugins.session import PluginSession
|
||||
|
||||
|
||||
async def model_task(base, pid, url, output, events, traces, model):
|
||||
"""Opt-in paid model run through the user's existing idle ARTEMIS service."""
|
||||
sid = str(uuid.uuid4())
|
||||
submitted = done = False
|
||||
async with httpx.AsyncClient(base_url=base, trust_env=False, timeout=30) as client:
|
||||
|
||||
async def api(path, body=None):
|
||||
r = (
|
||||
await client.get(path)
|
||||
if body is None
|
||||
else await client.post(path, json=body)
|
||||
)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
status = await api("/api/status")
|
||||
assert not status.get("active_tasks") and not status.get("queue"), (
|
||||
"ARTEMIS is busy"
|
||||
)
|
||||
body = {
|
||||
"goal": f"仅操作 PID={pid} 的独立 Brave 测试窗口，不操作其他浏览器窗口或手机。新建一个标签页并通过地址栏打开 {url}/form。将 Test parameter 输入框填写为 ARTEMIS 中文参数，点击 Apply parameter 一次，核对页面显示 Saved: ARTEMIS 中文参数。然后向下滚动到页面底部，点击 Finish scrolling 一次，确认显示 Finished。把最终窗口截图保存到任务 trace，文件名 browser-model-proof.png。以实际页面证据判定结果。",
|
||||
"session_id": sid,
|
||||
"profile": "flash",
|
||||
"ingress": "client",
|
||||
"runtime_overrides": {
|
||||
"execution_target": "computer",
|
||||
"plugins": [
|
||||
{
|
||||
"id": "cua-driver",
|
||||
"profile": "benchmark",
|
||||
"version": "0.36.0+artemis.2",
|
||||
}
|
||||
],
|
||||
"model": model,
|
||||
"reasoning_effort": "high",
|
||||
"timeout_seconds": 300,
|
||||
"request_timeout_seconds": 90,
|
||||
"max_retries": 1,
|
||||
"record_video": False,
|
||||
"analyze_video": False,
|
||||
},
|
||||
}
|
||||
(output / "request.json").write_text(
|
||||
json.dumps(body, ensure_ascii=False, indent=2), "utf-8"
|
||||
)
|
||||
try:
|
||||
reply = await api("/api/run", body)
|
||||
assert reply.get("status") not in {"rejected", "error"}, reply
|
||||
submitted = True
|
||||
print(json.dumps({"trace_id": sid, "submission": reply}), flush=True)
|
||||
async with asyncio.timeout(420):
|
||||
while True:
|
||||
await asyncio.sleep(10)
|
||||
status = await api(f"/api/sessions/{sid}")
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"trace_id": sid,
|
||||
"status": status["status"],
|
||||
"saved": any(e["path"] == "/save" for e in events),
|
||||
"finished": any(e["path"] == "/finish" for e in events),
|
||||
}
|
||||
),
|
||||
flush=True,
|
||||
)
|
||||
if status["status"] not in {
|
||||
"running",
|
||||
"pending",
|
||||
"queued",
|
||||
"stopping",
|
||||
"starting",
|
||||
}:
|
||||
done = True
|
||||
break
|
||||
proof = (
|
||||
traces / sid / "artifacts/plugins/cua-driver/browser-model-proof.png"
|
||||
)
|
||||
report = {
|
||||
"trace_id": sid,
|
||||
"status": status["status"],
|
||||
"events": list(events),
|
||||
"actual_pass": (
|
||||
[e["body"] for e in events if e["path"] == "/save"]
|
||||
== ["ARTEMIS 中文参数"]
|
||||
and len([e for e in events if e["path"] == "/finish"]) == 1
|
||||
and any(
|
||||
e["path"] == "/scroll" and float(e["body"]) > 1000
|
||||
for e in events
|
||||
)
|
||||
and proof.is_file()
|
||||
),
|
||||
}
|
||||
(output / "model-report.json").write_text(
|
||||
json.dumps(report, ensure_ascii=False, indent=2), "utf-8"
|
||||
)
|
||||
assert report["actual_pass"], report
|
||||
print(json.dumps({"trace_id": sid, "actual_pass": True}), flush=True)
|
||||
finally:
|
||||
if submitted and not done:
|
||||
await api("/api/stop", {"session_id": sid})
|
||||
|
||||
|
||||
async def run(
|
||||
executable: Path,
|
||||
browser: Path,
|
||||
output: Path,
|
||||
base=None,
|
||||
traces=None,
|
||||
model="gpt-6.1-sol",
|
||||
):
|
||||
output.mkdir(parents=True, exist_ok=True)
|
||||
events = []
|
||||
calls = []
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def log_message(self, *args):
|
||||
pass
|
||||
|
||||
def do_GET(self):
|
||||
events.append({"path": self.path})
|
||||
if self.path == "/form":
|
||||
body = """<!doctype html><meta charset="utf-8">
|
||||
<title>ARTEMIS Cua Browser Form</title>
|
||||
<style>body{font:24px sans-serif;margin:40px}input,button{font:inherit;padding:12px;display:block;margin:20px 0}</style>
|
||||
<h1>ARTEMIS Cua Browser Form</h1>
|
||||
<label>Test parameter<input aria-label="Test parameter" id="value"></label>
|
||||
<button aria-label="Apply parameter" onclick="fetch('/save',{method:'POST',body:document.getElementById('value').value}).then(()=>{document.getElementById('result').textContent='Saved: '+document.getElementById('value').value})">Apply parameter</button>
|
||||
<div id="result" aria-live="polite">Pending</div>
|
||||
<div style="height:1600px">Scroll down to finish</div>
|
||||
<button aria-label="Finish scrolling" onclick="fetch('/finish',{method:'POST',body:String(window.scrollY)}).then(()=>{this.textContent='Finished';this.setAttribute('aria-label','Finished');document.body.style.backgroundColor='#c4f9d0'})">Finish scrolling</button>
|
||||
<script>addEventListener('scroll',()=>fetch('/scroll',{method:'POST',body:String(window.scrollY)}))</script>
|
||||
"""
|
||||
else:
|
||||
body = "<title>ARTEMIS Cua Browser Start</title><h1>Isolated browser test</h1>"
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||||
self.end_headers()
|
||||
self.wfile.write(body.encode())
|
||||
|
||||
def do_POST(self):
|
||||
events.append(
|
||||
{
|
||||
"path": self.path,
|
||||
"body": self.rfile.read(
|
||||
int(self.headers["Content-Length"])
|
||||
).decode(),
|
||||
}
|
||||
)
|
||||
self.send_response(200)
|
||||
self.end_headers()
|
||||
|
||||
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
||||
threading.Thread(target=server.serve_forever, daemon=True).start()
|
||||
url = f"http://127.0.0.1:{server.server_port}"
|
||||
session = None
|
||||
process = None
|
||||
try:
|
||||
with tempfile.TemporaryDirectory(prefix="artemis-cua-browser-") as profile:
|
||||
try:
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
str(browser),
|
||||
"--user-data-dir=" + profile,
|
||||
"--no-first-run",
|
||||
"--no-default-browser-check",
|
||||
"--disable-extensions",
|
||||
"--disable-background-mode",
|
||||
"--force-renderer-accessibility",
|
||||
"--window-size=1000,800",
|
||||
"--window-position=100,100",
|
||||
url + "/start",
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
creationflags=subprocess.CREATE_NO_WINDOW,
|
||||
)
|
||||
if base:
|
||||
await model_task(
|
||||
base, process.pid, url, output, events, traces, model
|
||||
)
|
||||
return
|
||||
session = await PluginSession(
|
||||
ServerConfig(
|
||||
command=str(executable),
|
||||
env={"ARTEMIS_ARTIFACTS_DIR": str(output / "attachments")},
|
||||
),
|
||||
executable.parent,
|
||||
).start()
|
||||
schemas = {t.name: t.inputSchema for t in session.tools}
|
||||
|
||||
async def call(name, args, *, invalid=False):
|
||||
if not invalid:
|
||||
jsonschema.validate(args, schemas[name])
|
||||
started = time.monotonic()
|
||||
result = await session.call(name, args)
|
||||
data = result.structuredContent or {}
|
||||
calls.append(
|
||||
{
|
||||
"tool": name,
|
||||
"arguments": args,
|
||||
"seconds": round(time.monotonic() - started, 3),
|
||||
"is_error": result.isError,
|
||||
"result": data,
|
||||
"text": [
|
||||
b.text for b in result.content if b.type == "text"
|
||||
],
|
||||
}
|
||||
)
|
||||
(output / "calls.json").write_text(
|
||||
json.dumps(calls, ensure_ascii=False, indent=2), "utf-8"
|
||||
)
|
||||
if not invalid:
|
||||
assert not result.isError, calls[-1]
|
||||
return data
|
||||
|
||||
async with asyncio.timeout(20):
|
||||
while True:
|
||||
listing = await call("list_windows", {"pid": process.pid})
|
||||
own = listing["windows"]
|
||||
if own:
|
||||
break
|
||||
await asyncio.sleep(0.2)
|
||||
target = {"pid": process.pid, "window_id": own[0]["window_id"]}
|
||||
|
||||
async def observe():
|
||||
return await call("get_window_state", target)
|
||||
|
||||
async def wait_tree(text):
|
||||
async with asyncio.timeout(10):
|
||||
while True:
|
||||
state = await observe()
|
||||
if text in state["tree_markdown"]:
|
||||
return state
|
||||
await asyncio.sleep(0.2)
|
||||
|
||||
async def key(key, modifiers=None):
|
||||
return await call(
|
||||
"press_key",
|
||||
{
|
||||
**target,
|
||||
"key": key,
|
||||
"modifiers": modifiers or [],
|
||||
"delivery_mode": "foreground",
|
||||
},
|
||||
)
|
||||
|
||||
def token(state, label, role="Button"):
|
||||
match = re.search(
|
||||
r"\[(\d+)\] " + re.escape(role) + ' "' + re.escape(label) + '"',
|
||||
state["tree_markdown"],
|
||||
)
|
||||
assert match, state["tree_markdown"]
|
||||
return state["snapshot_id"] + ":" + match[1]
|
||||
|
||||
await wait_tree("ARTEMIS Cua Browser Start")
|
||||
# Reproduce task f1928... invalid form before any keyboard dispatch.
|
||||
bad = await call(
|
||||
"press_key",
|
||||
{
|
||||
**target,
|
||||
"key": "t",
|
||||
"modifiers": ["ctrl"],
|
||||
"element_token": "s00000001:9",
|
||||
"x": 0,
|
||||
"y": 0,
|
||||
},
|
||||
invalid=True,
|
||||
)
|
||||
assert bad["artemis"]["execution"] == "not_started"
|
||||
assert bad["artemis"]["side_effect"] == "none"
|
||||
assert "observation" not in bad
|
||||
await key("t", ["ctrl"])
|
||||
state = await observe()
|
||||
assert "ARTEMIS Cua Browser Start" not in state.get("window_title", "")
|
||||
await key("l", ["ctrl"])
|
||||
await call(
|
||||
"type_text",
|
||||
{**target, "text": url + "/form", "delivery_mode": "foreground"},
|
||||
)
|
||||
await key("return")
|
||||
state = await wait_tree("Test parameter")
|
||||
# Direct UIA replacement is a separate, unambiguous tool.
|
||||
await call(
|
||||
"set_value",
|
||||
{
|
||||
**target,
|
||||
"element_token": token(state, "Test parameter", "Edit"),
|
||||
"value": "seed",
|
||||
},
|
||||
)
|
||||
state = await observe()
|
||||
await call(
|
||||
"click",
|
||||
{
|
||||
**target,
|
||||
"element_token": token(state, "Test parameter", "Edit"),
|
||||
"delivery_mode": "foreground",
|
||||
},
|
||||
)
|
||||
await key("a", ["ctrl"])
|
||||
await call(
|
||||
"type_text",
|
||||
{
|
||||
**target,
|
||||
"text": "ARTEMIS 中文参数",
|
||||
"delivery_mode": "foreground",
|
||||
},
|
||||
)
|
||||
state = await wait_tree("ARTEMIS 中文参数")
|
||||
await call(
|
||||
"click",
|
||||
{**target, "element_token": token(state, "Apply parameter")},
|
||||
)
|
||||
state = await wait_tree("Saved: ARTEMIS 中文参数")
|
||||
assert [e["body"] for e in events if e["path"] == "/save"] == [
|
||||
"ARTEMIS 中文参数"
|
||||
]
|
||||
source = state["artemis"]["source"]
|
||||
# Wheel ticks do not guarantee an exact pixel/page distance.
|
||||
# Observe each actual movement before another bounded scroll.
|
||||
for _ in range(3):
|
||||
before = sum(e["path"] == "/scroll" for e in events)
|
||||
await call(
|
||||
"scroll",
|
||||
{
|
||||
**target,
|
||||
"direction": "down",
|
||||
"amount": 8,
|
||||
"by": "page",
|
||||
"x": source["width"] // 2,
|
||||
"y": source["height"] // 2,
|
||||
"delivery_mode": "foreground",
|
||||
},
|
||||
)
|
||||
async with asyncio.timeout(5):
|
||||
while sum(e["path"] == "/scroll" for e in events) <= before:
|
||||
await asyncio.sleep(0.1)
|
||||
state = await observe()
|
||||
source = state["artemis"]["source"]
|
||||
if any(
|
||||
e["path"] == "/scroll" and float(e["body"]) > 1000
|
||||
for e in events
|
||||
):
|
||||
break
|
||||
assert any(
|
||||
e["path"] == "/scroll" and float(e["body"]) > 1000 for e in events
|
||||
)
|
||||
state = await observe()
|
||||
await call(
|
||||
"click",
|
||||
{
|
||||
**target,
|
||||
"element_token": token(state, "Finish scrolling"),
|
||||
"delivery_mode": "foreground",
|
||||
},
|
||||
)
|
||||
await wait_tree("Finished")
|
||||
assert len([e for e in events if e["path"] == "/finish"]) == 1
|
||||
saved = await call(
|
||||
"save_screenshot", {**target, "filename": "browser-proof.png"}
|
||||
)
|
||||
attachment = saved["artemis"]["attachments"][0]
|
||||
assert (
|
||||
hashlib.sha256(
|
||||
(output / "attachments" / attachment["name"]).read_bytes()
|
||||
).hexdigest()
|
||||
== attachment["sha256"]
|
||||
)
|
||||
report = {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
"mixed parameters refused before dispatch",
|
||||
"Ctrl+T new tab",
|
||||
"Ctrl+L URL navigation",
|
||||
"UIA set_value",
|
||||
"focused Unicode text input",
|
||||
"single form submission readback",
|
||||
"actual page scroll",
|
||||
"finish button",
|
||||
"PNG hash",
|
||||
],
|
||||
"events": events,
|
||||
"calls": len(calls),
|
||||
}
|
||||
(output / "report.json").write_text(
|
||||
json.dumps(report, ensure_ascii=False, indent=2), "utf-8"
|
||||
)
|
||||
print(json.dumps(report, ensure_ascii=False))
|
||||
finally:
|
||||
(output / "events.json").write_text(
|
||||
json.dumps(events, ensure_ascii=False, indent=2), "utf-8"
|
||||
)
|
||||
if session:
|
||||
await session.aclose()
|
||||
if process:
|
||||
try:
|
||||
parent = psutil.Process(process.pid)
|
||||
owned = [*parent.children(recursive=True), parent]
|
||||
for proc in owned:
|
||||
try:
|
||||
proc.terminate()
|
||||
except psutil.NoSuchProcess:
|
||||
pass
|
||||
_, alive = psutil.wait_procs(owned, timeout=5)
|
||||
for proc in alive:
|
||||
proc.kill()
|
||||
psutil.wait_procs(alive, timeout=5)
|
||||
except psutil.NoSuchProcess:
|
||||
pass
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(__doc__)
|
||||
parser.add_argument("executable", type=Path)
|
||||
parser.add_argument("--browser", type=Path, required=True)
|
||||
parser.add_argument("--output", type=Path, required=True)
|
||||
parser.add_argument(
|
||||
"--model-base", help="Opt-in paid model test through an existing ARTEMIS server"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--traces", type=Path, help="That server's traces directory (model test only)"
|
||||
)
|
||||
parser.add_argument("--model", default="gpt-6.1-sol")
|
||||
args = parser.parse_args()
|
||||
if args.model_base and not args.traces:
|
||||
parser.error("--model-base requires --traces")
|
||||
asyncio.run(
|
||||
run(
|
||||
args.executable.resolve(),
|
||||
args.browser,
|
||||
args.output.resolve(),
|
||||
args.model_base,
|
||||
args.traces,
|
||||
args.model,
|
||||
)
|
||||
)
|
||||
Reference in new issue
Block a user