diff --git a/priv/python/albedo_plugins/browser/page.py b/priv/python/albedo_plugins/browser/page.py
index 98163fd0..75380f2e 100644
--- a/priv/python/albedo_plugins/browser/page.py
+++ b/priv/python/albedo_plugins/browser/page.py
@@ -682,7 +682,7 @@ class Page:
frame_id: str | None = None,
timeout: float | None = None,
) -> Any:
- """Evaluate an expression or a zero-argument function; await promises; return data.
+ """Evaluate an expression, function, or statement body; await promises; return data.
This is a raw escape hatch: it may mutate the page and bypasses action locking.
Explicitly invoke functions taking arguments in the supplied expression.
@@ -706,6 +706,14 @@ class Page:
)
params["contextId"] = world["executionContextId"]
result = await cdp.send("Runtime.evaluate", params, timeout=timeout)
+ detail = result.get("exceptionDetails", {})
+ # Parse failures have no stack; a runtime SyntaxError must not run twice.
+ if (
+ detail.get("exception", {}).get("className") == "SyntaxError"
+ and "stackTrace" not in detail
+ ):
+ params["expression"] = f"(async () => {{\n{expression}\n}})()"
+ result = await cdp.send("Runtime.evaluate", params, timeout=timeout)
return self._value(result)
@reported("browser.goto", {"url": "url"})
@@ -875,11 +883,17 @@ class Page:
return x, y
async def _click_point(
- self, record: Reference, state: FrameSession, object_id: str
+ self,
+ record: Reference,
+ state: FrameSession,
+ object_id: str,
+ *,
+ scroll: bool = True,
) -> tuple[float, float]:
- await state.cdp.send(
- "DOM.scrollIntoViewIfNeeded", {"backendNodeId": record.backend_id}
- )
+ if scroll:
+ await state.cdp.send(
+ "DOM.scrollIntoViewIfNeeded", {"backendNodeId": record.backend_id}
+ )
quads = (
await state.cdp.send(
"DOM.getContentQuads", {"backendNodeId": record.backend_id}
@@ -959,34 +973,64 @@ class Page:
"name": str(node.get("name", {}).get("value", ""))[:80],
}
- @reported("browser.click", {"ref": "ref"})
- async def click(self, ref: str | Node, *, button: str = "left") -> JSON:
- if button not in {"left", "right", "middle"}:
- raise ValueError("button must be left, right, or middle")
- async with self._action("click", ref) as receipt:
- async with self._resolved(ref) as (key, record, state, obj, status):
- if status["disabled"]:
- raise NotActionable("Element is disabled.", ref=key)
- x, y = await self._click_point(record, state, obj)
- self._ref(key)
- receipt["may_have_executed"] = True
- pressed = False
- try:
- await self.cdp.send(
- "Input.dispatchMouseEvent",
- {"type": "mouseMoved", "x": x, "y": y},
- )
- pressed = True
- await self.cdp.send(
- "Input.dispatchMouseEvent",
- {
- "type": "mousePressed",
- "x": x,
- "y": y,
- "button": button,
- "clickCount": 1,
- },
- )
+ async def _point(
+ self, target: str | Node | tuple[float, float], *, scroll: bool = True
+ ) -> tuple[float, float]:
+ if isinstance(target, tuple):
+ if len(target) != 2 or not all(
+ isinstance(v, (int, float)) and math.isfinite(v) for v in target
+ ):
+ raise ValueError("point must be a finite (x, y) pair")
+ return target
+ async with self._resolved(target) as (key, record, state, obj, status):
+ if status["disabled"]:
+ raise NotActionable("Element is disabled.", ref=key)
+ point = await self._click_point(record, state, obj, scroll=scroll)
+ self._ref(key)
+ return point
+
+ async def _mouse_gesture(
+ self, points: list[tuple[float, float]], button: str, receipt: JSON
+ ) -> None:
+ x, y = points[0]
+ pressed = False
+ failed = True
+ receipt["may_have_executed"] = True
+ try:
+ await self.cdp.send(
+ "Input.dispatchMouseEvent", {"type": "mouseMoved", "x": x, "y": y}
+ )
+ pressed = True
+ await self.cdp.send(
+ "Input.dispatchMouseEvent",
+ {
+ "type": "mousePressed",
+ "x": x,
+ "y": y,
+ "button": button,
+ "clickCount": 1,
+ },
+ )
+ for x, y in points[1:]:
+ await self.cdp.send(
+ "Input.dispatchMouseEvent",
+ {
+ "type": "mouseMoved",
+ "x": x,
+ "y": y,
+ "button": button,
+ "buttons": {"left": 1, "right": 2, "middle": 4}[button],
+ },
+ )
+ failed = False
+ finally:
+ if pressed and not self.browser._connection.closed:
+ # Release normally too; cleanup must not replace a move's error.
+ with (
+ contextlib.suppress(Exception)
+ if failed
+ else contextlib.nullcontext()
+ ):
await self.cdp.send(
"Input.dispatchMouseEvent",
{
@@ -996,22 +1040,63 @@ class Page:
"button": button,
"clickCount": 1,
},
+ timeout=1,
)
- pressed = False
- finally:
- if pressed and not self.browser._connection.closed:
- with contextlib.suppress(Exception):
- await self.cdp.send(
- "Input.dispatchMouseEvent",
- {
- "type": "mouseReleased",
- "x": x,
- "y": y,
- "button": button,
- "clickCount": 1,
- },
- timeout=1,
- )
+
+ @reported("browser.click", {"ref": "ref"})
+ async def click(
+ self, ref: str | Node | tuple[float, float], *, button: str = "left"
+ ) -> JSON:
+ """Click an observed node or a viewport (x, y) point."""
+ if button not in {"left", "right", "middle"}:
+ raise ValueError("button must be left, right, or middle")
+ async with self._action(
+ "click", None if isinstance(ref, tuple) else ref
+ ) as receipt:
+ await self._mouse_gesture([await self._point(ref)], button, receipt)
+ return copy.deepcopy(receipt)
+
+ async def click_text(self, text: str) -> JSON:
+ """Click the unique button or link whose accessible name contains text."""
+ if not isinstance(text, str) or not text:
+ raise ValueError("text must be a nonempty string")
+ observation = await self.observe()
+ matches = [
+ node
+ for node in find(observation, name=text, exact=False)
+ if node["role"] in {"button", "link"}
+ ]
+ if len(matches) != 1:
+ raise SelectionError(
+ f"Expected one button or link containing {text!r}, found {len(matches)}.",
+ matches=matches[:10],
+ observation_complete=observation["coverage"].get("complete", False),
+ )
+ return await self.click(matches[0])
+
+ @reported("browser.drag", {"start": "start", "end": "end"})
+ async def drag(
+ self,
+ start: str | Node | tuple[float, float],
+ end: str | Node | tuple[float, float],
+ *,
+ steps: int = 20,
+ ) -> JSON:
+ """Drag between nodes or viewport points with the left button held."""
+ if (
+ isinstance(steps, bool)
+ or not isinstance(steps, int)
+ or not 1 <= steps <= 1000
+ ):
+ raise ValueError("steps must be an integer from 1 to 1000")
+ async with self._action("drag") as receipt:
+ x0, y0 = await self._point(start)
+ x1, y1 = await self._point(end, scroll=False)
+ points = [
+ (x0 + (x1 - x0) * i / steps, y0 + (y1 - y0) * i / steps)
+ for i in range(steps + 1)
+ ]
+ await self._mouse_gesture(points, "left", receipt)
return copy.deepcopy(receipt)
@reported("browser.fill", {"ref": "ref", "text": "text"})
diff --git a/robot-docs/kernel.md b/robot-docs/kernel.md
index eea16eb6..a37c6503 100644
--- a/robot-docs/kernel.md
+++ b/robot-docs/kernel.md
@@ -50,6 +50,23 @@ file path into memory.
Invalid, cut-short, or over-edge attachments still raise. No image decoder is
required, so overflow is not tiled.
+## browser gestures
+
+`priv/python/albedo_plugins/browser/page.py` owns page actions and their lock.
+Use `await page.click((x, y))` for viewport coordinates and
+`await page.drag(start, end, steps=20)` for left-button drags. Endpoints may be
+observed nodes, references, or viewport `(x, y)` points; node endpoints must
+both fit in the viewport. Drag steps are 1–1000, and mouse release runs even
+if a move fails. Document-level mouse/pointer listeners do not need capture.
+Use `await page.click_text(text)` to click a unique button or link whose
+accessible name contains the case-sensitive text. Zero or multiple matches
+raise `SelectionError` rather than choosing one.
+
+`await page.evaluate(source)` accepts expressions, zero-argument functions, and
+statement bodies such as `localStorage.clear(); return 1`. It selects the syntax
+before running the source, so a runtime exception never causes a retry.
+Promises are awaited. This escape hatch still bypasses action locking.
+
## spawned browsers
`priv/python/albedo_plugins/browser/launch.py` owns browser discovery and spawned
diff --git a/src/albedo/harness/extensions/browser/extension.gleam b/src/albedo/harness/extensions/browser/extension.gleam
index 56c4ec8c..4518e42e 100644
--- a/src/albedo/harness/extensions/browser/extension.gleam
+++ b/src/albedo/harness/extensions/browser/extension.gleam
@@ -14,6 +14,9 @@ pub fn extension() -> extension.Extension {
<> " await page.select(node, value), await page.hover(node), await page.press(key),"
<> " await page.wait_for(...), and await page.screenshot()."
<> " browser.find(snapshot, ...) and browser.one(snapshot, ...) locate nodes in an observation."
+ <> " Use await page.click((x, y)) for coordinates, await page.drag(start, end, steps=20) for drags (observed nodes or viewport points),"
+ <> " and await page.click_text(text) for a unique button/link whose accessible name contains text; none or several raises SelectionError."
+ <> " await page.evaluate(source) accepts expressions, zero-argument functions, or statement bodies such as 'localStorage.clear(); return 1'."
<> " Finish with await b.close(). Failures raise BrowserError.",
["python"],
)
diff --git a/test/e2e/browser_test.py b/test/e2e/browser_test.py
index 49b9f72d..323f8225 100644
--- a/test/e2e/browser_test.py
+++ b/test/e2e/browser_test.py
@@ -1,6 +1,10 @@
"""Browser discovery through the real kernel, without starting a GUI app."""
import unittest
+import http.server
+import threading
+import shutil
+from pathlib import Path
from harness import Albedo, Provider, exclusive, python, text
@@ -81,3 +85,120 @@ Path({str(app.workspace / "passed")!r}).write_text('passed')
app.prompt(session, "try the installed browsers").close()
app.idle(session)
self.assertTrue((app.workspace / "passed").exists(), app.history(session))
+
+
+class BrowserGestureTests(unittest.TestCase):
+ def test_gestures_and_statement_bodies_in_a_real_browser(self):
+ if not (
+ shutil.which("chromium")
+ or shutil.which("google-chrome")
+ or shutil.which("chrome-headless-shell")
+ or Path(
+ "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
+ ).exists()
+ or Path("/Applications/Chromium.app/Contents/MacOS/Chromium").exists()
+ ):
+ self.skipTest("No Chrome or Chromium installed on this system")
+
+ class Handler(http.server.BaseHTTPRequestHandler):
+ def do_GET(self):
+ body = b"CSP test"
+ self.send_response(200)
+ self.send_header("Content-Type", "text/html")
+ self.send_header("Content-Length", str(len(body)))
+ self.send_header("Content-Security-Policy", "script-src 'self'")
+ self.end_headers()
+ self.wfile.write(body)
+
+ def log_message(self, format, *args):
+ pass
+
+ server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+ thread = threading.Thread(target=server.serve_forever, daemon=True)
+ thread.start()
+ self.addCleanup(server.server_close)
+ self.addCleanup(thread.join)
+ self.addCleanup(server.shutdown)
+ code = """
+from albedo_plugins.browser import SelectionError, JavaScriptError
+b = await browser.spawn()
+try:
+ page = await b.new_page()
+ await page.goto('data:text/html,Other link')
+ assert await page.evaluate('1 + 2') == 3
+ assert await page.evaluate('() => 4') == 4
+ assert await page.evaluate('window.runs = 0; return 5') == 5
+ try:
+ await page.evaluate("window.runs++; throw new SyntaxError('runtime')")
+ except JavaScriptError:
+ pass
+ else:
+ raise AssertionError('runtime error was swallowed')
+ assert await page.evaluate('window.runs') == 1
+ try:
+ await page.evaluate("(() => { window.runs++; throw new SyntaxError('runtime'); })()")
+ except JavaScriptError:
+ pass
+ else:
+ raise AssertionError('expression runtime error was swallowed')
+ assert await page.evaluate('window.runs') == 2
+ await page.click((30, 30))
+ await page.click_text('Drag handle')
+ assert await page.evaluate('window.clicks') == 2
+ for label in ('missing', 'Other'):
+ try:
+ await page.click_text(label)
+ except SelectionError as error:
+ assert ('found 0' if label == 'missing' else 'found 2') in str(error)
+ else:
+ raise AssertionError('ambiguous or missing text was clicked')
+ await page.evaluate('''
+ window.moves = []; window.releases = 0;
+ document.addEventListener('pointermove', e => {if(e.buttons === 1) window.moves.push([e.clientX,e.clientY]);});
+ document.addEventListener('pointerup', () => window.releases++);
+ return true;
+ ''')
+ snapshot = await page.observe()
+ node = browser.one(snapshot, role='button', name='Drag handle')
+ await page.drag(node, (200, 150), steps=5)
+ assert await page.evaluate('window.moves.length') == 5
+ assert await page.evaluate('window.moves.at(-1)') == [200, 150]
+ assert await page.evaluate('window.releases') == 1
+ await page.drag((200, 150), node, steps=3)
+ assert await page.evaluate('window.moves.length') == 8
+ assert await page.evaluate('window.releases') == 2
+ print('document drag: 8 held-button moves, 2 releases; clicks: 2')
+ await page.goto(CSP_URL)
+ try:
+ await page.evaluate("new Function('return 1')()")
+ except JavaScriptError as error:
+ assert 'unsafe-eval' in str(error.details), error
+ else:
+ raise AssertionError('CSP must block Function compilation')
+ assert await page.evaluate('1 + 2') == 3
+ assert await page.evaluate('localStorage.clear(); return 5') == 5
+ print('CSP expression and statement body passed')
+finally:
+ await b.close()
+"""
+
+ def reply(request):
+ if request["messages"][-1].get("role") == "user":
+ return python(
+ code.replace(
+ "CSP_URL", repr(f"http://127.0.0.1:{server.server_port}/")
+ )
+ )
+ return text("done")
+
+ provider = Provider(reply)
+ self.addCleanup(provider.close)
+ with Albedo(provider) as app:
+ session = app.session()
+ app.prompt(session, "exercise browser gestures").close()
+ app.idle(session)
+ history = str(app.history(session))
+ self.assertIn("CSP expression and statement body passed", history)
+ self.assertIn(
+ "document drag: 8 held-button moves, 2 releases; clicks: 2", history
+ )