Spaces:
Sleeping
Sleeping
File size: 4,432 Bytes
6ec7f56 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """Scripted oracle agent — the harness's own reference implementation.
It parses the natural-language instruction (never the oracle API) and emits
ordinary computer-use actions, but it locates click targets via the DOM
(`browser.center`) instead of pixels. Its purpose is to validate the full
pipeline — env → screenshot loop → action executor → server state → verify —
not to be a fair baseline. Model adapters must not receive the browser handle.
"""
from __future__ import annotations
import re
from typing import Iterator
from ..browser import Browser
from .base import Observation
# must match server/seeding.py TIMEZONES order (oracle knowledge, documented)
TIMEZONES = [
"UTC", "America/Vancouver", "America/New_York", "Europe/Paris",
"Europe/Berlin", "Asia/Tokyo", "Asia/Singapore", "Australia/Sydney",
]
class ScriptedAgent:
def __init__(self, browser: Browser, task_id: str):
self.browser = browser
self.task_id = task_id
self._plan: Iterator[dict] | None = None
def begin(self, instruction: str) -> None:
plans = {
"email-triage": self._email_plan,
"settings-panel": self._settings_plan,
"checkout-form": self._checkout_plan,
}
if self.task_id not in plans:
raise ValueError(f"scripted agent has no plan for '{self.task_id}'")
self._plan = plans[self.task_id](instruction)
def step(self, obs: Observation) -> dict:
assert self._plan is not None, "begin() not called"
try:
return next(self._plan)
except StopIteration:
return {"action": "done"}
# ------------------------------------------------------------------ plans
def _click(self, selector: str, has_text: str | None = None) -> dict:
x, y = self.browser.center(selector, has_text)
return {"action": "left_click", "coordinate": [round(x), round(y)]}
def _email_plan(self, instruction: str) -> Iterator[dict]:
m = re.search(r"archive every email from (.+?) and star the email "
r"with the subject “(.+?)”", instruction)
assert m, f"could not parse instruction: {instruction!r}"
sender, subject = m.group(1), m.group(2)
while self.browser.exists("tr.email-row", has_text=sender):
row = f"tr.email-row:has(td.sender:text-is('{sender}'))"
x, y = self.browser.center(f"{row} .archive-btn")
yield {"action": "left_click", "coordinate": [round(x), round(y)]}
yield {"action": "wait", "duration": 0.3}
star_row = f"tr.email-row:has-text('{subject}')"
yield self._click(f"{star_row} .star-btn")
def _settings_plan(self, instruction: str) -> Iterator[dict]:
m = re.search(r"set the timezone to (\S+?),", instruction)
assert m, f"could not parse instruction: {instruction!r}"
target_tz = m.group(1)
yield self._click("#nav button", has_text="Appearance")
yield self._click("input[data-field='dark_mode']")
yield self._click("#nav button", has_text="General")
# focus the timezone <select>, then arrow down from UTC (index 0)
yield self._click("select[data-field='timezone']")
for _ in range(TIMEZONES.index(target_tz)):
yield {"action": "key", "text": "Down"}
yield {"action": "key", "text": "Return"} # close the dropdown
yield self._click("#save")
yield {"action": "wait", "duration": 0.3}
def _checkout_plan(self, instruction: str) -> Iterator[dict]:
m = re.search(r"checkout for (.+?) \(email: (.+?)\)\. "
r"Ship to (.+?), (.+?), ZIP (\d{5})", instruction)
assert m, f"could not parse instruction: {instruction!r}"
name, email, street, city, zip_code = m.groups()
for field, value in (("#name", name), ("#email", email)):
yield self._click(field)
yield {"action": "type", "text": value}
yield self._click("#next1")
yield {"action": "wait", "duration": 0.3}
for field, value in (("#street", street), ("#city", city), ("#zip", zip_code)):
yield self._click(field)
yield {"action": "type", "text": value}
yield self._click("#next2")
yield {"action": "wait", "duration": 0.3}
yield self._click("#submit")
yield {"action": "wait", "duration": 0.3}
|