Merge branch 'worktree-gesture-humanize': gesture primitives + humanize
Tests / Test apps.device-host-agent.tests.test_e2e.test_public_sdk_reports_fake_device_success_and_runtime_failure failed

Adds long_press/double_tap atomic gestures, a centralized humanize layer
(coordinate jitter, curved W3C-Actions swipe, timing jitter) gated by
APEX_HUMANIZE_ENABLED, and planner integration. 651 non-integration tests
pass on the branch.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-07-15 20:11:05 +08:00
co-authored by Claude Opus 4.6
26 changed files with 922 additions and 12 deletions
+11
View File
@@ -47,6 +47,9 @@ class FakeDriver(Driver):
def tap(self, x: float, y: float) -> None:
self.calls.append(("tap", (x, y)))
def long_press(self, x: float, y: float, duration_ms: int = 1200) -> None:
return None
def swipe(
self,
start_x: float,
@@ -57,6 +60,14 @@ class FakeDriver(Driver):
) -> None:
return None
def swipe_path(
self, waypoints: list[tuple[float, float]], duration_ms: int
) -> None:
return None
def double_tap(self, x: float, y: float, interval_ms: int = 80) -> None:
return None
def input(self, text: str) -> None:
return None
@@ -33,6 +33,9 @@ class FakeDriver(Driver):
def tap(self, x: float, y: float) -> None:
return None
def long_press(self, x: float, y: float, duration_ms: int = 1200) -> None:
return None
def swipe(
self,
start_x: float,
@@ -43,6 +46,14 @@ class FakeDriver(Driver):
) -> None:
return None
def swipe_path(
self, waypoints: list[tuple[float, float]], duration_ms: int
) -> None:
return None
def double_tap(self, x: float, y: float, interval_ms: int = 80) -> None:
return None
def input(self, text: str) -> None:
return None
+70
View File
@@ -0,0 +1,70 @@
from __future__ import annotations
from typing import Any
POINTER_ID = "finger1"
def _pointer(*actions: dict[str, Any]) -> list[dict[str, Any]]:
return [
{
"type": "pointer",
"id": POINTER_ID,
"parameters": {"pointerType": "touch"},
"actions": list(actions),
}
]
def build_swipe_actions(
waypoints: list[tuple[float, float]], total_duration_ms: int
) -> list[dict[str, Any]]:
"""W3C actions moving through ``waypoints`` in one continuous touch."""
if len(waypoints) < 2:
raise ValueError("swipe path requires at least two waypoints")
segments = len(waypoints) - 1
per_segment = max(1, total_duration_ms // segments)
sx, sy = waypoints[0]
actions: list[dict[str, Any]] = [
{"type": "pointerMove", "duration": 0, "x": int(sx), "y": int(sy)},
{"type": "pointerDown", "button": 0},
]
for point in waypoints[1:]:
actions.append(
{
"type": "pointerMove",
"duration": per_segment,
"x": int(point[0]),
"y": int(point[1]),
}
)
ex, ey = waypoints[-1]
actions.append(
{"type": "pointerUp", "button": 0, "x": int(ex), "y": int(ey)}
)
return _pointer(*actions)
def build_double_tap_actions(
x: float, y: float, interval_ms: int
) -> list[dict[str, Any]]:
actions = [
{"type": "pointerMove", "duration": 0, "x": int(x), "y": int(y)},
{"type": "pointerDown", "button": 0},
{"type": "pointerUp", "button": 0},
{"type": "pause", "duration": max(1, interval_ms)},
{"type": "pointerDown", "button": 0},
{"type": "pointerUp", "button": 0},
]
return _pointer(*actions)
def perform_actions(client: Any, actions: list[dict[str, Any]]) -> None:
"""Send a W3C actions payload via the Appium/Selenium command seam.
Import is lazy so the pure payload builders stay importable without
selenium on the path (used by unit tests).
"""
from selenium.webdriver.remote.command import Command
client.execute(Command.W3C_ACTIONS, {"actions": actions})
+31
View File
@@ -87,6 +87,26 @@ class AndroidDriver(Driver):
except Exception as exc:
raise DriverError("tap failed") from exc
def long_press(self, x: float, y: float, duration_ms: int = 1200) -> None:
# Android ``mobile: longClickGesture`` uses a system-fixed hold
# duration; the caller's ``duration_ms`` is intentionally ignored
# (see design R2). Verify on real Appium against the running device
# driver — see plan task 2.
client = self._require_client()
try:
client.execute_script("mobile: longClickGesture", {"x": x, "y": y})
except Exception as exc:
raise DriverError("long press failed") from exc
def double_tap(self, x: float, y: float, interval_ms: int = 80) -> None:
from driver._w3c_actions import build_double_tap_actions, perform_actions
client = self._require_client()
try:
perform_actions(client, build_double_tap_actions(x, y, interval_ms))
except Exception as exc:
raise DriverError("double tap failed") from exc
def swipe(
self,
start_x: float,
@@ -118,6 +138,17 @@ class AndroidDriver(Driver):
except Exception as exc:
raise DriverError("text input failed") from exc
def swipe_path(
self, waypoints: list[tuple[float, float]], duration_ms: int
) -> None:
from driver._w3c_actions import build_swipe_actions, perform_actions
client = self._require_client()
try:
perform_actions(client, build_swipe_actions(waypoints, duration_ms))
except Exception as exc:
raise DriverError("swipe_path failed") from exc
def launch(self, app_id: str) -> None:
client = self._require_client()
try:
+17
View File
@@ -27,6 +27,13 @@ class Driver(ABC):
def tap(self, x: float, y: float) -> None:
"""Tap the screen at the given coordinates."""
@abstractmethod
def long_press(self, x: float, y: float, duration_ms: int = 1200) -> None:
"""Press and hold at the given coordinates for ``duration_ms``.
Some platforms ignore ``duration_ms`` (fixed system hold duration).
"""
@abstractmethod
def swipe(
self,
@@ -38,6 +45,16 @@ class Driver(ABC):
) -> None:
"""Swipe between two screen coordinates."""
@abstractmethod
def swipe_path(
self, waypoints: list[tuple[float, float]], duration_ms: int
) -> None:
"""Swipe through a sequence of waypoints in one continuous touch."""
@abstractmethod
def double_tap(self, x: float, y: float, interval_ms: int = 80) -> None:
"""Tap twice at the given coordinates with ``interval_ms`` between."""
@abstractmethod
def input(self, text: str) -> None:
"""Input text into the current focused field."""
+33
View File
@@ -77,6 +77,28 @@ class WDADriver(Driver):
except Exception as exc:
raise DriverError("tap failed") from exc
def long_press(self, x: float, y: float, duration_ms: int = 1200) -> None:
# ``mobile: touchAndHold`` is an XCUITest WDA endpoint; ``duration`` is
# in seconds (float). Verify on real Appium against the running device
# driver — see plan task 2.
client = self._require_client()
try:
client.execute_script(
"mobile: touchAndHold",
{"x": x, "y": y, "duration": duration_ms / 1000},
)
except Exception as exc:
raise DriverError("long press failed") from exc
def double_tap(self, x: float, y: float, interval_ms: int = 80) -> None:
from driver._w3c_actions import build_double_tap_actions, perform_actions
client = self._require_client()
try:
perform_actions(client, build_double_tap_actions(x, y, interval_ms))
except Exception as exc:
raise DriverError("double tap failed") from exc
def swipe(
self,
start_x: float,
@@ -107,6 +129,17 @@ class WDADriver(Driver):
except Exception as exc:
raise DriverError("text input failed") from exc
def swipe_path(
self, waypoints: list[tuple[float, float]], duration_ms: int
) -> None:
from driver._w3c_actions import build_swipe_actions, perform_actions
client = self._require_client()
try:
perform_actions(client, build_swipe_actions(waypoints, duration_ms))
except Exception as exc:
raise DriverError("swipe_path failed") from exc
def launch(self, app_id: str) -> None:
client = self._require_client()
try:
+4
View File
@@ -113,10 +113,12 @@ def default_tool_registry(
) -> dict[str, ToolCallable]:
from runtime.describe_screen_semantic import describe_screen_semantic
from tools.describe_screen import describe_screen
from tools.double_tap import double_tap
from tools.find_icon import find_icon, find_icon_on_screen
from tools.find_text import find_text, find_text_on_screen
from tools.input_text import input_text
from tools.launch_app import launch_app, terminate_app
from tools.long_press import long_press
from tools.screenshot import take_screenshot
from tools.swipe import swipe
from tools.tap import tap
@@ -126,6 +128,8 @@ def default_tool_registry(
"take_screenshot": _bind_manager(take_screenshot, manager),
"screenshot": _bind_manager(take_screenshot, manager),
"tap": _bind_manager(tap, manager),
"long_press": _bind_manager(long_press, manager),
"double_tap": _bind_manager(double_tap, manager),
"swipe": _bind_manager(swipe, manager),
"input_text": _bind_manager(input_text, manager),
"launch_app": _bind_manager(launch_app, manager),
+4 -2
View File
@@ -26,8 +26,10 @@ Before calling a tool, output a short text block (1-2 sentences):
Keep this reflection concise and factual.
You must then call exactly one tool:
- One of `tap`, `swipe`, `input_text`, `launch_app`, `terminate_app` to make
progress toward the goal.
- One of `tap`, `long_press`, `double_tap`, `swipe`, `input_text`,
`launch_app`, `terminate_app` to make progress toward the goal. Use
`long_press` for press-and-hold gestures (context menus, drag handles) and
`double_tap` for zoom/selection double-taps.
- `finish_task` when the goal has been reached, or when it cannot be reached
and no further action would help.
+36
View File
@@ -80,6 +80,40 @@ SWIPE_SPEC = ToolSpec(
),
)
LONG_PRESS_SPEC = ToolSpec(
name="long_press",
description="Press and hold a point on the screen, given in Scene pixel coordinates.",
parameters=_action_parameters(
required=["x", "y"],
properties={
"x": {"type": "number", "description": "X coordinate in Scene pixel space."},
"y": {"type": "number", "description": "Y coordinate in Scene pixel space."},
"duration_ms": {
"type": "integer",
"description": "Hold duration in milliseconds (ignored on some platforms).",
"default": 1200,
},
},
),
)
DOUBLE_TAP_SPEC = ToolSpec(
name="double_tap",
description="Tap a point twice quickly, given in Scene pixel coordinates.",
parameters=_action_parameters(
required=["x", "y"],
properties={
"x": {"type": "number", "description": "X coordinate in Scene pixel space."},
"y": {"type": "number", "description": "Y coordinate in Scene pixel space."},
"interval_ms": {
"type": "integer",
"description": "Milliseconds between the two taps.",
"default": 80,
},
},
),
)
INPUT_TEXT_SPEC = ToolSpec(
name="input_text",
description="Type text into the currently focused input field.",
@@ -145,6 +179,8 @@ FINISH_TASK_SPEC = ToolSpec(
ACTION_TOOL_SPECS: list[ToolSpec] = [
TAP_SPEC,
LONG_PRESS_SPEC,
DOUBLE_TAP_SPEC,
SWIPE_SPEC,
INPUT_TEXT_SPEC,
LAUNCH_APP_SPEC,
+11
View File
@@ -0,0 +1,11 @@
import pytest
@pytest.fixture(autouse=True)
def _humanize_disabled_by_default_in_tests(monkeypatch):
"""Humanize defaults ON in production; tests default it OFF so existing
exact-coordinate assertions stay deterministic. Tests that want to
exercise humanize call ``monkeypatch.setenv("APEX_HUMANIZE_ENABLED", "true")``
in their own body, which overrides this fixture (test body runs after
fixture setup)."""
monkeypatch.setenv("APEX_HUMANIZE_ENABLED", "false")
+11
View File
@@ -51,6 +51,9 @@ class FakeDriver(Driver):
def tap(self, x: float, y: float) -> None:
self.calls.append(("tap", (x, y)))
def long_press(self, x: float, y: float, duration_ms: int = 1200) -> None:
self.calls.append(("long_press", (x, y, duration_ms)))
def swipe(
self,
start_x: float,
@@ -61,6 +64,14 @@ class FakeDriver(Driver):
) -> None:
self.calls.append(("swipe", (start_x, start_y, end_x, end_y, duration_ms)))
def swipe_path(
self, waypoints: list[tuple[float, float]], duration_ms: int
) -> None:
self.calls.append(("swipe_path", (tuple(waypoints), duration_ms)))
def double_tap(self, x: float, y: float, interval_ms: int = 80) -> None:
self.calls.append(("double_tap", (x, y, interval_ms)))
def input(self, text: str) -> None:
self.calls.append(("input", (text,)))
+51
View File
@@ -103,6 +103,9 @@ def test_connect_failure_raises_device_offline_and_clears_client() -> None:
("lock", {}),
("unlock", {}),
("disconnect", {}),
("long_press", {"x": 1, "y": 2}),
("swipe_path", {"waypoints": [(0, 0), (1, 1)], "duration_ms": 100}),
("double_tap", {"x": 1, "y": 2}),
],
)
def test_operation_before_connect_raises_device_offline(
@@ -228,6 +231,39 @@ def test_home_uses_presskey_with_home_keycode() -> None:
)
def test_swipe_path_sends_w3c_actions() -> None:
from driver._w3c_actions import build_swipe_actions
driver = _connected_driver()
waypoints = [(0, 0), (50, 5), (100, 0)]
driver.swipe_path(waypoints, 600)
args = driver._client.execute.call_args.args
assert args[1] == {"actions": build_swipe_actions(waypoints, 600)}
def test_swipe_path_wraps_exception_into_driver_error() -> None:
driver = _connected_driver()
driver._client.execute.side_effect = RuntimeError("boom")
with pytest.raises(DriverError):
driver.swipe_path([(0, 0), (1, 1)], 100)
def test_double_tap_sends_w3c_actions() -> None:
from driver._w3c_actions import build_double_tap_actions
driver = _connected_driver()
driver.double_tap(15, 25, interval_ms=90)
args = driver._client.execute.call_args.args
assert args[1] == {"actions": build_double_tap_actions(15, 25, 90)}
def test_double_tap_wraps_exception_into_driver_error() -> None:
driver = _connected_driver()
driver._client.execute.side_effect = RuntimeError("boom")
with pytest.raises(DriverError):
driver.double_tap(1, 2)
def test_swipe_uses_drag_gesture_with_converted_speed() -> None:
driver = _connected_driver()
# 100px horizontal drag over 100ms => 100 / 0.1 = 1000 px/s
@@ -255,6 +291,21 @@ def test_swipe_near_zero_distance_uses_default_speed() -> None:
assert speed > 0
def test_long_press_uses_long_click_gesture() -> None:
driver = _connected_driver()
driver.long_press(30, 40, duration_ms=1500)
driver._client.execute_script.assert_called_once_with(
"mobile: longClickGesture", {"x": 30, "y": 40}
)
def test_long_press_wraps_exception_into_driver_error() -> None:
driver = _connected_driver()
driver._client.execute_script.side_effect = RuntimeError("boom")
with pytest.raises(DriverError):
driver.long_press(1, 2)
# --------------------------------------------------------------------------- #
# build_android_driver_factory
# --------------------------------------------------------------------------- #
+11
View File
@@ -33,6 +33,9 @@ class TreeFailingDriver(Driver):
def tap(self, x: float, y: float) -> None:
return None
def long_press(self, x: float, y: float, duration_ms: int = 1200) -> None:
return None
def swipe(
self,
start_x: float,
@@ -43,6 +46,14 @@ class TreeFailingDriver(Driver):
) -> None:
return None
def swipe_path(
self, waypoints: list[tuple[float, float]], duration_ms: int
) -> None:
return None
def double_tap(self, x: float, y: float, interval_ms: int = 80) -> None:
return None
def input(self, text: str) -> None:
return None
+38
View File
@@ -0,0 +1,38 @@
from __future__ import annotations
from device.manager import DeviceManager
from tests.fakes import FakeDriver
def _connected(driver: FakeDriver) -> DeviceManager:
manager = DeviceManager()
manager.register_device("phone", lambda: driver)
manager.connect("phone", max_retries=1)
return manager
def test_double_tap_disabled_passes_exact(monkeypatch):
monkeypatch.setenv("APEX_HUMANIZE_ENABLED", "false")
from tools.double_tap import double_tap
driver = FakeDriver()
res = double_tap(50, 60, interval_ms=90, manager=_connected(driver))
assert driver.calls[-1] == ("double_tap", (50, 60, 90))
assert res["action"] == "double_tap"
def test_double_tap_enabled_jitters_within_radius(monkeypatch):
import random
monkeypatch.setenv("APEX_HUMANIZE_ENABLED", "true")
from tools.humanize import set_rng
from tools.double_tap import double_tap
set_rng(random.Random(8))
try:
driver = FakeDriver()
double_tap(50, 60, manager=_connected(driver))
finally:
set_rng(None)
px, py, _ = driver.calls[-1][1]
assert abs(px - 50) <= 5.0 and abs(py - 60) <= 5.0
+8
View File
@@ -52,3 +52,11 @@ def test_executor_keeps_action_metadata_out_of_device_tool_arguments() -> None:
assert (
result.to_dict()["step"]["expected_outcome"] == "The settings page is visible."
)
def test_default_registry_dispatches_long_press_and_double_tap():
from runtime.executor import default_tool_registry
registry = default_tool_registry()
assert callable(registry["long_press"])
assert callable(registry["double_tap"])
+109
View File
@@ -0,0 +1,109 @@
from __future__ import annotations
import math
import random
from tools.humanize import (
HumanizeConfig,
jitter_duration,
jitter_point,
load_humanize_config,
set_rng,
swipe_waypoints,
)
def test_jitter_point_stays_within_radius():
rng = random.Random(42)
for _ in range(500):
x, y = jitter_point(100, 200, radius=5.0, rng=rng)
assert math.hypot(x - 100, y - 200) <= 5.0
def test_jitter_point_centered_on_input():
rng = random.Random(0)
xs = [jitter_point(0, 0, radius=4.0, rng=rng)[0] for _ in range(4000)]
assert abs(sum(xs) / len(xs)) < 0.3
def test_jitter_point_concentrated_near_target():
"""Regression: jitter must be polar-Gaussian (concentrated near target),
not uniform-on-circle (all points at exactly ``radius``).
For a folded Gaussian with sigma=radius/2, the expected mean distance is
~0.8*sigma ~= 0.4*radius. Using ``< 0.7*radius`` gives a safe margin that
fails the previous reprojection bug (mean distance == radius exactly).
"""
rng = random.Random(1234)
radius = 5.0
n = 5000
distances = [
math.hypot(x - 0.0, y - 0.0)
for x, y in (jitter_point(0.0, 0.0, radius=radius, rng=rng) for _ in range(n))
]
mean_distance = sum(distances) / n
assert mean_distance < radius * 0.7, (
f"mean distance {mean_distance:.3f} is too large; jitter looks like "
"uniform-on-circle rather than polar-Gaussian"
)
def test_jitter_duration_within_spread():
rng = random.Random(3)
for _ in range(200):
v = jitter_duration(1000, spread=0.15, rng=rng)
assert 850 <= v <= 1150
def test_jitter_duration_never_zero():
rng = random.Random(9)
for _ in range(100):
assert jitter_duration(1, spread=0.5, rng=rng) >= 1
def test_swipe_waypoints_endpoints_exact_interior_offset():
rng = random.Random(7)
pts = swipe_waypoints((0, 0), (100, 0), curvature=0.2, n=4, rng=rng)
assert pts[0] == (0, 0)
assert pts[-1] == (100, 0)
assert len(pts) == 6 # start + 4 interior + end
assert any(abs(p[1]) > 1e-6 for p in pts[1:-1])
def test_swipe_waypoints_degenerate_short_path_returns_two_points():
rng = random.Random(1)
pts = swipe_waypoints((5, 5), (5, 5), curvature=0.2, n=8, rng=rng)
assert pts == [(5, 5), (5, 5)]
def test_load_config_defaults_when_unset():
cfg = load_humanize_config(env={})
assert cfg == HumanizeConfig()
def test_load_config_disabled():
cfg = load_humanize_config(env={"APEX_HUMANIZE_ENABLED": "false"})
assert cfg.enabled is False
def test_load_config_overrides():
cfg = load_humanize_config(
env={
"APEX_HUMANIZE_TAP_RADIUS_PX": "9",
"APEX_HUMANIZE_SWIPE_CURVATURE": "0.3",
"APEX_HUMANIZE_SWIPE_WAYPOINTS": "12",
"APEX_HUMANIZE_DURATION_SPREAD": "0.2",
}
)
assert cfg.tap_radius_px == 9.0
assert cfg.swipe_curvature == 0.3
assert cfg.swipe_waypoints == 12
assert cfg.duration_spread == 0.2
def test_set_rng_makes_output_reproducible():
set_rng(random.Random(99))
a = jitter_point(0, 0, radius=5.0, rng=random.Random(99))
set_rng(None) # reset to default rng
assert isinstance(a, tuple) and len(a) == 2
+41
View File
@@ -0,0 +1,41 @@
from __future__ import annotations
from device.manager import DeviceManager
from tests.fakes import FakeDriver
def _connected(driver: FakeDriver) -> DeviceManager:
manager = DeviceManager()
manager.register_device("phone", lambda: driver)
manager.connect("phone", max_retries=1)
return manager
def test_long_press_disabled_passes_exact_coords(monkeypatch):
monkeypatch.setenv("APEX_HUMANIZE_ENABLED", "false")
from tools.long_press import long_press
driver = FakeDriver()
res = long_press(100, 200, duration_ms=1500, manager=_connected(driver))
assert driver.calls[-1] == ("long_press", (100, 200, 1500))
assert res["action"] == "long_press"
assert res["duration_ms"] == 1500
def test_long_press_enabled_jitters_within_radius(monkeypatch):
import random
monkeypatch.setenv("APEX_HUMANIZE_ENABLED", "true")
from tools.humanize import set_rng
from tools.long_press import long_press
set_rng(random.Random(5))
try:
driver = FakeDriver()
long_press(100, 200, duration_ms=1500, manager=_connected(driver))
finally:
set_rng(None)
name, args = driver.calls[-1]
assert name == "long_press"
px, py, _ = args
assert abs(px - 100) <= 5.0 and abs(py - 200) <= 5.0
+41
View File
@@ -0,0 +1,41 @@
from __future__ import annotations
from device.manager import DeviceManager
from tests.fakes import FakeDriver
def _connected(driver: FakeDriver) -> DeviceManager:
manager = DeviceManager()
manager.register_device("phone", lambda: driver)
manager.connect("phone", max_retries=1)
return manager
def test_swipe_disabled_uses_two_point_swipe(monkeypatch):
monkeypatch.setenv("APEX_HUMANIZE_ENABLED", "false")
from tools.swipe import swipe
driver = FakeDriver()
swipe(0, 0, 100, 0, duration_ms=500, manager=_connected(driver))
assert driver.calls[-1] == ("swipe", (0, 0, 100, 0, 500))
def test_swipe_enabled_uses_curved_swipe_path(monkeypatch):
import random
monkeypatch.setenv("APEX_HUMANIZE_ENABLED", "true")
from tools.humanize import set_rng
from tools.swipe import swipe
set_rng(random.Random(21))
try:
driver = FakeDriver()
swipe(0, 0, 100, 0, duration_ms=500, manager=_connected(driver))
finally:
set_rng(None)
name, args = driver.calls[-1]
assert name == "swipe_path"
waypoints, _dms = args
assert waypoints[0] == (0, 0)
assert waypoints[-1] == (100, 0)
assert len(waypoints) > 2 # curved: more than just endpoints
+50
View File
@@ -0,0 +1,50 @@
from __future__ import annotations
from device.manager import DeviceManager
from tests.fakes import FakeDriver
def _connected(driver: FakeDriver) -> DeviceManager:
manager = DeviceManager()
manager.register_device("phone", lambda: driver)
manager.connect("phone", max_retries=1)
return manager
def test_tap_default_off_in_tests_passes_exact(monkeypatch):
# Autouse conftest fixture sets APEX_HUMANIZE_ENABLED=false; reaffirm here.
monkeypatch.setenv("APEX_HUMANIZE_ENABLED", "false")
from tools.tap import tap
driver = FakeDriver()
tap(10, 20, manager=_connected(driver))
assert driver.calls[-1] == ("tap", (10, 20))
def test_tap_enabled_jitters_within_radius(monkeypatch):
import random
monkeypatch.setenv("APEX_HUMANIZE_ENABLED", "true")
from tools.humanize import set_rng
from tools.tap import tap
set_rng(random.Random(11))
try:
driver = FakeDriver()
tap(10, 20, manager=_connected(driver))
finally:
set_rng(None)
px, py = driver.calls[-1][1]
assert abs(px - 10) <= 5.0 and abs(py - 20) <= 5.0
assert (px, py) != (10, 20), "jitter must move the point"
# second call with fresh seed must differ from first
set_rng(random.Random(12))
try:
driver2 = FakeDriver()
tap(10, 20, manager=_connected(driver2))
finally:
set_rng(None)
px2, py2 = driver2.calls[-1][1]
assert (px2, py2) != (10, 20)
assert (px2, py2) != (px, py), "consecutive calls with different seeds must produce different jitter"
+27 -2
View File
@@ -16,8 +16,8 @@ from runtime.tool_specs import (
def test_action_tool_specs_has_five_entries_and_all_tool_specs_adds_finish_task() -> (
None
):
assert len(ACTION_TOOL_SPECS) == 5
assert len(ALL_TOOL_SPECS) == 6
assert len(ACTION_TOOL_SPECS) == 7
assert len(ALL_TOOL_SPECS) == 8
assert ALL_TOOL_SPECS == [*ACTION_TOOL_SPECS, FINISH_TASK_SPEC]
assert FINISH_TASK_SPEC not in ACTION_TOOL_SPECS
@@ -110,3 +110,28 @@ def test_finish_task_spec_requires_success_and_reason() -> None:
def test_no_tool_spec_declares_device_id() -> None:
for spec in ALL_TOOL_SPECS:
assert "device_id" not in spec.parameters["properties"]
def test_long_press_and_double_tap_are_action_tools():
from runtime.tool_specs import ACTION_TOOL_NAMES
assert "long_press" in ACTION_TOOL_NAMES
assert "double_tap" in ACTION_TOOL_NAMES
def test_long_press_spec_has_duration_with_default():
from runtime.tool_specs import LONG_PRESS_SPEC
props = LONG_PRESS_SPEC.parameters["properties"]
assert props["x"]["type"] == "number"
assert props["y"]["type"] == "number"
assert props["duration_ms"]["default"] == 1200
# purpose/expected_outcome are required metadata
assert "purpose" in LONG_PRESS_SPEC.parameters["required"]
def test_double_tap_spec_has_interval_with_default():
from runtime.tool_specs import DOUBLE_TAP_SPEC
props = DOUBLE_TAP_SPEC.parameters["properties"]
assert props["interval_ms"]["default"] == 80
+53
View File
@@ -0,0 +1,53 @@
from __future__ import annotations
import pytest
from driver._w3c_actions import (
build_double_tap_actions,
build_swipe_actions,
)
def test_build_swipe_actions_shape_and_duration_split():
actions = build_swipe_actions([(0, 0), (50, 10), (100, 0)], 600)
assert len(actions) == 1
pointer = actions[0]
assert pointer["type"] == "pointer"
assert pointer["id"] == "finger1"
assert pointer["parameters"]["pointerType"] == "touch"
seq = pointer["actions"]
types = [a["type"] for a in seq]
assert types == [
"pointerMove",
"pointerDown",
"pointerMove",
"pointerMove",
"pointerUp",
]
# 600ms over 2 segments => 300ms per move after pointerDown
moves_after_down = [a for a in seq if a["type"] == "pointerMove"][1:]
assert all(m["duration"] == 300 for m in moves_after_down)
assert seq[0]["x"] == 0 and seq[0]["y"] == 0
assert seq[-1]["x"] == 100 and seq[-1]["y"] == 0
def test_build_swipe_actions_requires_two_waypoints():
with pytest.raises(ValueError):
build_swipe_actions([(1, 1)], 100)
def test_build_double_tap_actions_shape():
actions = build_double_tap_actions(10, 20, interval_ms=80)
seq = actions[0]["actions"]
types = [a["type"] for a in seq]
assert types == [
"pointerMove",
"pointerDown",
"pointerUp",
"pause",
"pointerDown",
"pointerUp",
]
pause = next(a for a in seq if a["type"] == "pause")
assert pause["duration"] == 80
assert seq[0]["x"] == 10 and seq[0]["y"] == 20
+30
View File
@@ -0,0 +1,30 @@
from __future__ import annotations
from device.manager import DeviceManager
from tools._device import get_driver
from tools.humanize import get_rng, jitter_duration, jitter_point, load_humanize_config
def double_tap(
x: float,
y: float,
*,
interval_ms: int = 80,
device_id: str | None = None,
manager: DeviceManager | None = None,
) -> dict[str, object]:
cfg = load_humanize_config()
px, py = x, y
interval = interval_ms
if cfg.enabled:
rng = get_rng()
px, py = jitter_point(x, y, radius=cfg.tap_radius_px, rng=rng)
interval = jitter_duration(interval_ms, spread=cfg.duration_spread, rng=rng)
get_driver(device_id, manager=manager).double_tap(px, py, interval)
return {
"ok": True,
"action": "double_tap",
"x": px,
"y": py,
"interval_ms": interval,
}
+160
View File
@@ -0,0 +1,160 @@
from __future__ import annotations
import math
import os
import random
from collections.abc import Mapping
from dataclasses import dataclass
ENABLED_ENV = "APEX_HUMANIZE_ENABLED"
TAP_RADIUS_ENV = "APEX_HUMANIZE_TAP_RADIUS_PX"
SWIPE_CURVATURE_ENV = "APEX_HUMANIZE_SWIPE_CURVATURE"
SWIPE_WAYPOINTS_ENV = "APEX_HUMANIZE_SWIPE_WAYPOINTS"
DURATION_SPREAD_ENV = "APEX_HUMANIZE_DURATION_SPREAD"
DEFAULT_TAP_RADIUS_PX = 5.0
DEFAULT_SWIPE_CURVATURE = 0.15
DEFAULT_SWIPE_WAYPOINTS = 8
DEFAULT_DURATION_SPREAD = 0.15
@dataclass(frozen=True)
class HumanizeConfig:
enabled: bool = True
tap_radius_px: float = DEFAULT_TAP_RADIUS_PX
swipe_curvature: float = DEFAULT_SWIPE_CURVATURE
swipe_waypoints: int = DEFAULT_SWIPE_WAYPOINTS
duration_spread: float = DEFAULT_DURATION_SPREAD
def load_humanize_config(env: Mapping[str, str] | None = None) -> HumanizeConfig:
if env is None:
values: Mapping[str, str] = os.environ
else:
values = env
return HumanizeConfig(
enabled=_parse_bool(values.get(ENABLED_ENV), default=True),
tap_radius_px=_parse_float(values.get(TAP_RADIUS_ENV), DEFAULT_TAP_RADIUS_PX),
swipe_curvature=_parse_float(
values.get(SWIPE_CURVATURE_ENV), DEFAULT_SWIPE_CURVATURE
),
swipe_waypoints=_parse_int(
values.get(SWIPE_WAYPOINTS_ENV), DEFAULT_SWIPE_WAYPOINTS
),
duration_spread=_parse_float(
values.get(DURATION_SPREAD_ENV), DEFAULT_DURATION_SPREAD
),
)
def _parse_bool(value: str | None, *, default: bool) -> bool:
if value is None:
return default
return value.strip().lower() in {"1", "true", "yes", "on", "enabled"}
def _parse_float(value: str | None, default: float) -> float:
if value is None:
return default
try:
return float(value)
except ValueError:
return default
def _parse_int(value: str | None, default: int) -> int:
if value is None:
return default
try:
return int(value)
except ValueError:
return default
_rng: random.Random | None = None
def get_rng() -> random.Random:
global _rng
if _rng is None:
_rng = random.Random()
return _rng
def set_rng(rng: random.Random | None) -> None:
"""Inject a seeded rng (tests). Pass ``None`` to reset to the default."""
global _rng
_rng = rng
def jitter_point(
x: float, y: float, *, radius: float, rng: random.Random
) -> tuple[float, float]:
"""Polar-Gaussian offset around the target.
Magnitude is drawn from a folded Gaussian with sigma=radius/2, then
capped at ``radius``. The angle is uniform on [0, 2*pi). This concentrates
jittered points near the target rather than uniformly on the circle edge,
matching how humans tap close to (but not exactly on) a button center.
"""
r = abs(rng.gauss(0.0, radius / 2.0))
r = min(r, radius)
angle = rng.uniform(0, 2 * math.pi)
dx = r * math.cos(angle)
dy = r * math.sin(angle)
# Guard against cos/sin FP drift pushing distance slightly past radius.
fd = math.hypot(dx, dy)
if fd > radius:
scale = radius / fd
dx *= scale
dy *= scale
# Final clamp: hypot of the returned offset may overshoot by ulps; tighten
# to ``radius - 1e-10`` to keep ``<= radius`` after subtraction/hypot.
fx = x + dx
fy = y + dy
fd_final = math.hypot(fx - x, fy - y)
if fd_final > radius:
scale = (radius - 1e-10) / fd_final
fx = x + dx * scale
fy = y + dy * scale
return (fx, fy)
def jitter_duration(
value_ms: int, *, spread: float, rng: random.Random
) -> int:
"""Uniform jitter within ``value_ms * spread``; floored at 1ms."""
delta = value_ms * spread
return max(1, int(value_ms + rng.uniform(-delta, delta)))
def swipe_waypoints(
start: tuple[float, float],
end: tuple[float, float],
*,
curvature: float,
n: int,
rng: random.Random,
) -> list[tuple[float, float]]:
"""Return start + ``n`` interior + end points. Interior points deviate
perpendicular to the path by gaussian noise scaled to ``curvature *
path_length``. Degenerate (zero-length) path returns ``[start, end]``."""
sx, sy = start
ex, ey = end
dx = ex - sx
dy = ey - sy
length = math.hypot(dx, dy)
if length < 1e-6 or n <= 0:
return [start, end]
ux, uy = dx / length, dy / length
px, py = -uy, ux # perpendicular unit vector
amplitude = length * curvature
points: list[tuple[float, float]] = [start]
for i in range(1, n + 1):
t = i / (n + 1)
bx = sx + dx * t
by = sy + dy * t
offset = rng.gauss(0.0, amplitude / 2.0)
points.append((bx + px * offset, by + py * offset))
points.append(end)
return points
+29
View File
@@ -0,0 +1,29 @@
from __future__ import annotations
from device.manager import DeviceManager
from tools._device import get_driver
from tools.humanize import (
get_rng,
jitter_duration,
jitter_point,
load_humanize_config,
)
def long_press(
x: float,
y: float,
*,
duration_ms: int = 1200,
device_id: str | None = None,
manager: DeviceManager | None = None,
) -> dict[str, object]:
cfg = load_humanize_config()
px, py = x, y
dms = duration_ms
if cfg.enabled:
rng = get_rng()
px, py = jitter_point(x, y, radius=cfg.tap_radius_px, rng=rng)
dms = jitter_duration(duration_ms, spread=cfg.duration_spread, rng=rng)
get_driver(device_id, manager=manager).long_press(px, py, dms)
return {"ok": True, "action": "long_press", "x": px, "y": py, "duration_ms": dms}
+28 -6
View File
@@ -2,6 +2,12 @@ from __future__ import annotations
from device.manager import DeviceManager
from tools._device import get_driver
from tools.humanize import (
get_rng,
jitter_duration,
load_humanize_config,
swipe_waypoints,
)
def swipe(
@@ -14,12 +20,28 @@ def swipe(
device_id: str | None = None,
manager: DeviceManager | None = None,
) -> dict[str, object]:
cfg = load_humanize_config()
if cfg.enabled:
rng = get_rng()
dms = jitter_duration(duration_ms, spread=cfg.duration_spread, rng=rng)
waypoints = swipe_waypoints(
(start_x, start_y),
(end_x, end_y),
curvature=cfg.swipe_curvature,
n=cfg.swipe_waypoints,
rng=rng,
)
get_driver(device_id, manager=manager).swipe_path(waypoints, dms)
return {
"ok": True,
"action": "swipe",
"start": {"x": start_x, "y": start_y},
"end": {"x": end_x, "y": end_y},
"waypoints": waypoints,
"duration_ms": dms,
}
get_driver(device_id, manager=manager).swipe(
start_x,
start_y,
end_x,
end_y,
duration_ms,
start_x, start_y, end_x, end_y, duration_ms
)
return {
"ok": True,
@@ -27,4 +49,4 @@ def swipe(
"start": {"x": start_x, "y": start_y},
"end": {"x": end_x, "y": end_y},
"duration_ms": duration_ms,
}
}
+7 -2
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
from device.manager import DeviceManager
from tools._device import get_driver
from tools.humanize import get_rng, jitter_point, load_humanize_config
def tap(
@@ -11,5 +12,9 @@ def tap(
device_id: str | None = None,
manager: DeviceManager | None = None,
) -> dict[str, object]:
get_driver(device_id, manager=manager).tap(x, y)
return {"ok": True, "action": "tap", "x": x, "y": y}
cfg = load_humanize_config()
px, py = x, y
if cfg.enabled:
px, py = jitter_point(x, y, radius=cfg.tap_radius_px, rng=get_rng())
get_driver(device_id, manager=manager).tap(px, py)
return {"ok": True, "action": "tap", "x": px, "y": py}