chenmingxuan/Artemis-Eapil
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 32 Wiki Activity

fix: guard transient controls and verify timeline endpoints

Browse Source
This commit is contained in:
chenmingxuan committed 2026-11-02 13:54:50 +08:00
1 parent 479a185192
commit 00e62a663a
29 files changed
+1150 -90
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

artemis/mcp/action_executor.py
+23 -5
View File
@@ -190,7 +190,7 @@ class McpActionExecutor:
raw_name: str,
previous_screenshot: str | None,
previous_elements: list[dict[str, Any]] | None = None,
) -> tuple[ObserveResult, bool]:
) -> tuple[ObserveResult, bool | None]:
"""Prefer a settled image/tree pair after actions that can change screens.
The Android helper captures its bitmap and hierarchy concurrently. During a
@@ -198,6 +198,11 @@ class McpActionExecutor:
describes the new one. Bounded re-observation waits for adjacent stable
frames and rejects a new hierarchy paired with the previous page's bitmap.
"""
if raw_name == "action_sequence":
# Keep transient controls in the recorded post-image. Repeated settling
# captures can outlast the popup and replace useful evidence with a
# later closed state. This capture does not assert screen stability.
return await session.observe(settle_ms=0), None
first = await session.observe(settle_ms=400)
if not first.ok or raw_name not in _SCREEN_CHANGING_ACTIONS:
return first, first.ok
@@ -440,7 +445,7 @@ class McpActionExecutor:
@staticmethod
def _observe_despite_failure(raw_name: str, res: ActionResult) -> bool:
"""Capture the screen after launch failures and wait_for_text timeouts.
"""Capture the screen after runtime failures, including stale indexed taps.
Argument and package lookup errors do not require an observation.
"""
@@ -450,6 +455,8 @@ class McpActionExecutor:
return res.code is ActionCode.TIMEOUT
if raw_name in {"drag", "action_sequence"}:
return res.code not in {ActionCode.INVALID_ARGS, ActionCode.UNSUPPORTED}
if raw_name == "click":
return res.metadata.get("execution") in {"not_started", "unknown"}
return False
# --- Argument translation --------------------------------------------------------
@@ -474,18 +481,24 @@ class McpActionExecutor:
"""
if raw_name in {"drag", "action_sequence"}:
from artemis.config.agent import load_agent_config
from artemis.mcp.gestures import DragRequest, SequenceRequest, plan_drag, plan_sequence
from artemis.mcp.gestures import (
DragRequest,
plan_drag,
plan_sequence,
validate_sequence,
)
try:
if raw_name == "drag":
request = DragRequest.model_validate(args)
planned = plan_drag(request, *self._screen_size())
return "drag", request.model_dump(mode="json"), None, planned
sequence_request = SequenceRequest.model_validate(args)
limit = load_agent_config().pro.execution.max_burst_actions
sequence_request = validate_sequence(args, limit)
plan_sequence(
sequence_request,
*self._screen_size(),
limit=load_agent_config().pro.execution.max_burst_actions,
limit=limit,
)
return "action_sequence", sequence_request.model_dump(mode="json"), None, {}
except ValueError as exc:
@@ -505,6 +518,11 @@ class McpActionExecutor:
"target": target,
"times": args.get("times", 1),
"delay_ms": args.get("delay_ms", 100),
**(
{"expected_resource_id": recorded["target_resource_id"]}
if recorded.get("target_resource_id")
else {}
),
},
None,
recorded,
artemis/mcp/action_names.py
+8
View File
@@ -101,6 +101,14 @@ def to_canonical_call(
"target": [pair[0], pair[1]],
"times": action_item.get("times") or action_item.get("click_times") or 1,
"delay_ms": action_item.get("delay_ms") or action_item.get("delay") or 100,
**(
{
"expected_resource_id": action_item.get("resource_id")
or action_item.get("target_resource_id")
}
if action_item.get("resource_id") or action_item.get("target_resource_id")
else {}
),
}
if verb == "long_press_on":
artemis/mcp/action_specs.py
+72 -16
View File
@@ -186,8 +186,11 @@ _TARGET_DESCRIPTION = (
async def _wire_click(actuator: Any, a: dict[str, Any]) -> ActionResult:
guard = (
{"expected_resource_id": a["expected_resource_id"]} if a.get("expected_resource_id") else {}
)
return await actuator.click(
int(a["target"][0]), int(a["target"][1]), times=a["times"], delay_ms=a["delay_ms"]
int(a["target"][0]), int(a["target"][1]), times=a["times"], delay_ms=a["delay_ms"], **guard
)
@@ -259,15 +262,30 @@ _SPECS: tuple[ActionSpec, ...] = (
operator=OperatorDialect(
description=(
"[ACTION] Execute a short sequence of click, wait and drag steps in ONE call, "
"in order, without model turns or screenshots between members. Use for "
"in order, without model turns between members; explicit visibility guards "
"may read the live UI locally. Use for "
"temporary overlays: open the control, wait briefly, then operate it before "
"it disappears. Use observed coordinates, or positions learned from an "
"earlier observation of the same restored layout; element indices are not "
"accepted. Each click/drag needs target_description. All members are "
"it disappears. For clicks prefer target=the exact observed resource_id string: "
"live bounds are resolved locally, refusing absent or ambiguous targets. "
"Otherwise use observed [x,y] in the declared coordinate_space; do not confuse "
"original screenshot pixels with 0..1000 normalized coordinates. Element indices "
"are not accepted. Each click/drag needs target_description. All members are "
"validated before any input. The configured fast-action limit applies. "
"The first failure stops the remainder; receipts identify completed and "
"unexecuted members. Never replay the whole sequence after a partial result. "
"Verify the final screen before claiming the intended result."
"Every wait counts as a member. "
"click.times (1..10) performs consecutive taps within ONE member, like the "
"ordinary click tool; prefer this over expanding repeated +/- taps into many "
"members/model turns. Each resource-targeted tap is revalidated locally; "
"missing targets stop remaining taps and receipts preserve completed_taps. "
"For maximum/minimum prefer observed native +/- controls, confirming that "
"further same-direction adjustment stops changing handle/ticks. An estimated "
"drag endpoint can be imprecise. For every auto-hiding adjustment, including "
"the reverse direction, keep opener/wait/operation in one call. "
"A prior screenshot does not prove the panel is still open. Use click.unless_visible "
"and drag.require_visible with an observed panel resource_id to avoid toggling an "
"open panel closed or dragging its underlying content. Never invent resource IDs. "
"Verify actual ticks/handle endpoints before claiming maximum or minimum."
),
params=(
ParamSpec("steps", list[GestureStep], "Ordered click, wait and drag members."),
@@ -286,8 +304,12 @@ _SPECS: tuple[ActionSpec, ...] = (
"[ACTION] Precisely drag a local slider or timeline. Use explicit start/end "
"or calibration from TWO observed numeric scale anchors. coordinate_space "
"is normalized (0..1000) or pixel (the current screenshot's actual pixels); "
"pixel avoids quantization on a zoomed scale. Default duration_ms=300: "
"slower is not always more precise and may trigger long-press recognition. "
"pixel avoids quantization on a zoomed scale. Default duration_ms=300. "
"If a panned timeline coasts past the release position, try a slower pan "
"(e.g. 1500ms), then wait for its ticks to stop before measuring the residual. "
"A transient release-time bubble can disagree with the final fixed-pointer "
"scale; do not treat it as final position proof. A permanently visible "
"timeline can be panned independently of its zoom popup. "
"Do not overshoot the endpoint automatically. For calibration, axis=x/y; "
"mode=thumb moves the handle to target_value, mode=pan shifts the scale "
"under indicator_position. Use one numeric unit, e.g. seconds since midnight, "
@@ -317,6 +339,12 @@ _SPECS: tuple[ActionSpec, ...] = (
"Observed numeric anchors that compute the endpoint. When supplied, end MUST be null; otherwise set calibration null.",
required=False,
),
ParamSpec(
"require_visible",
str | None,
"Previously observed panel resource_id; refuse the drag if it is no longer visible.",
required=False,
),
),
),
wire=WireDialect(
@@ -330,6 +358,7 @@ _SPECS: tuple[ActionSpec, ...] = (
),
ParamSpec("duration_ms", int, required=False, default=300),
ParamSpec("calibration", DragCalibration | None, required=False),
ParamSpec("require_visible", str | None, required=False),
),
bind=_wire_drag,
),
@@ -339,7 +368,10 @@ _SPECS: tuple[ActionSpec, ...] = (
operator=OperatorDialect(
description=(
"[ACTION] Click on the target location on the screen (supports element"
" index or absolute normalized coordinates)."
" index or absolute normalized coordinates). Indexed resource targets are checked "
"against live bounds before input; missing/moved targets refuse the tap. For "
"floating or auto-hiding panels, use a guarded action_sequence with opener and "
"target in the same call; an open panel in a screenshot may already have expired."
),
params=(
ParamSpec(
@@ -379,6 +411,12 @@ _SPECS: tuple[ActionSpec, ...] = (
ParamSpec("target", list[int]),
ParamSpec("times", int, required=False, default=1),
ParamSpec("delay_ms", int, required=False, default=100),
ParamSpec(
"expected_resource_id",
str | None,
"Observed index resource ID; refuse a stale tap outside its live bounds.",
required=False,
),
),
bind=_wire_click,
),
@@ -386,7 +424,8 @@ _SPECS: tuple[ActionSpec, ...] = (
"target: the agent dialect accepts an element index or a coordinate pair"
" (an index is resolved client-side against the indexed element list); the"
" wire takes coordinates only. target_description: recorded, never sent"
" to the wire; required only for coordinate targets."
" to the wire; required only for coordinate targets. Indexed resource identity is "
"sent as expected_resource_id for live bounds validation; coordinate-only calls omit it."
),
),
ActionSpec(
@@ -809,7 +848,7 @@ def exception_prefix(action: str) -> str:
@cache
def _operator_args_model(name: str):
def _operator_args_model(name: str, sequence_limit: int | None = None):
"""Builds (once) the pydantic argument model for an operator shell."""
dialect = ACTION_SPECS[name].operator
if dialect is None:
@@ -817,7 +856,12 @@ def _operator_args_model(name: str):
fields: dict[str, Any] = {}
for p in dialect.params:
if p.required:
fields[p.name] = (p.annotation, Field(description=p.description))
bounds = (
{"min_length": 1, "max_length": sequence_limit or 10}
if name == "action_sequence"
else {}
)
fields[p.name] = (p.annotation, Field(description=p.description, **bounds))
else:
fields[p.name] = (
p.annotation,
@@ -826,7 +870,7 @@ def _operator_args_model(name: str):
return create_model(name, **fields)
def operator_shell_tool(name: str) -> StructuredTool:
def operator_shell_tool(name: str, *, sequence_limit: int | None = None) -> StructuredTool:
"""Builds the Operator's declaration-only shell for one action.
The body is never executed by the Operator loop -- action calls are translated
@@ -838,8 +882,13 @@ def operator_shell_tool(name: str) -> StructuredTool:
raise ValueError(f"Action '{name}' has no operator dialect.")
return StructuredTool(
name=name,
description=dialect.description,
args_schema=_operator_args_model(name),
description=dialect.description
+ (
f" This task allows at most {sequence_limit} members per call, including waits."
if name == "action_sequence" and sequence_limit is not None
else ""
),
args_schema=_operator_args_model(name, sequence_limit),
func=lambda **kwargs: "Action Recorded",
)
@@ -900,7 +949,7 @@ def _projected_parameters(dialect: OperatorDialect, name: str) -> dict[str, Any]
@cache
def tool_declaration(name: str) -> ToolDeclaration:
def tool_declaration(name: str, *, sequence_limit: int | None = None) -> ToolDeclaration:
"""Builds the Flash ``ToolDeclaration`` for one action.
A shared action is the operator shell projected onto JSON (same description,
@@ -908,6 +957,13 @@ def tool_declaration(name: str) -> ToolDeclaration:
``DeclarationDialect``.
"""
spec = ACTION_SPECS[name]
if name == "action_sequence" and sequence_limit is not None:
shell = operator_shell_tool(name, sequence_limit=sequence_limit)
return ToolDeclaration(
name=name,
description=shell.description,
parameters=convert_to_openai_tool(shell)["function"]["parameters"],
)
if spec.declaration is not None:
return ToolDeclaration(
name=name,
artemis/mcp/actuators/adb.py
+208 -12
View File
@@ -143,9 +143,26 @@ class AdbActuator:
return await self._dispatch_drag(planned)
async def _dispatch_drag(self, planned: dict) -> ActionResult:
error = await self.controller.swipe_coords(*planned["coordinates"], planned["duration_ms"])
metadata = dict(planned)
metadata.pop("action")
if resource_id := planned.get("require_visible"):
try:
guard = await self._visibility_guard(resource_id)
except (OSError, RuntimeError) as exc:
return ActionResult.failure(
"drag",
f"Visibility check failed; no input executed: {exc}",
metadata={**metadata, "execution": "not_started", "side_effect": "none"},
)
metadata["visibility_guard"] = guard
if not guard["visible"]:
return ActionResult.failure(
"drag",
"TARGET_NOT_FOUND: required panel control has disappeared; no drag executed. Re-establish the panel and operate it in one guarded action_sequence.",
code=ActionCode.TARGET_NOT_FOUND,
metadata={**metadata, "execution": "not_started", "side_effect": "none"},
)
error = await self.controller.swipe_coords(*planned["coordinates"], planned["duration_ms"])
if error:
return ActionResult.failure("drag", str(error), metadata=metadata)
return ActionResult.success(
@@ -156,20 +173,104 @@ class AdbActuator:
metadata=metadata,
)
async def _live_ui_elements(self) -> list[dict]:
elements = await self.controller.get_ui_elements()
if not isinstance(elements, list) or not elements:
raise RuntimeError("No live UI hierarchy available for the visibility guard.")
return elements
async def _visibility_guard(self, resource_id: str, elements: list[dict] | None = None) -> dict:
"""Read live UI locally; never treat failed/empty hierarchy as absence."""
if elements is None:
elements = await self._live_ui_elements()
visible = any(
(e.get("resource_id") or e.get("resource-id")) == resource_id
and e.get("visible-to-user", True) not in (False, "false")
for e in elements
if isinstance(e, dict)
)
return {"resource_id": resource_id, "visible": visible, "checked_at": time.time()}
def _live_target_coordinates(
self, resource_id: str, elements: list[dict], expected_point: tuple[int, int] | None = None
) -> list[int]:
from artemis.utils.visualization import parse_bounds
width, height = self._dims()
matches = []
for element in elements:
if (
element.get("resource_id") or element.get("resource-id")
) != resource_id or element.get("visible-to-user", True) in (False, "false"):
continue
raw_bounds = element.get("bounds")
bounds = parse_bounds(raw_bounds) if isinstance(raw_bounds, str) else raw_bounds
if not (
isinstance(bounds, (list, tuple))
and len(bounds) == 4
and all(isinstance(v, int) and not isinstance(v, bool) for v in bounds)
):
continue
if bounds:
left, top, right, bottom = bounds
left, top = max(0, left), max(0, top)
right, bottom = min(width, right), min(height, bottom)
if expected_point and not (
left <= expected_point[0] < right and top <= expected_point[1] < bottom
):
continue
if left < right and top < bottom:
matches.append([int((left + right) / 2), int((top + bottom) / 2)])
if len(matches) != 1:
raise LookupError(
f"TARGET_NOT_FOUND: {resource_id} has {len(matches)} visible matches; no further tap executed. Use a unique observed resource_id or re-observed coordinates."
)
return matches[0]
async def _tap_native_target(
self,
resource_id: str,
receipt: dict,
*,
times: int = 1,
delay_ms: int = 100,
elements: list[dict] | None = None,
expected_point: tuple[int, int] | None = None,
) -> str | None:
"""Revalidate every repeated native tap, retaining partial input receipts."""
receipt.update(completed_taps=0, execution="not_started")
for index in range(times):
if index:
await asyncio.sleep(max(0, delay_ms) / 1000)
elements = None
if elements is None:
elements = await self._live_ui_elements()
point = self._live_target_coordinates(resource_id, elements, expected_point)
coordinate_key = "pixel_coordinates" if expected_point is not None else "coordinates"
point = list(expected_point) if expected_point is not None else point
receipt[coordinate_key] = point
receipt["execution"] = "unknown"
result = await self.controller.tap_at(*point)
if error := getattr(result, "error", None):
return str(error)
receipt["completed_taps"] += 1
return None
async def action_sequence(self, steps: list[Any]) -> ActionResult:
"""Validate the entire sequence, then execute once with no intermediate observation."""
"""Execute once; optional visibility guards read live UI without model turns."""
import time
from artemis.config.agent import load_agent_config
from artemis.data_engine.trace import TraceSpan
from artemis.mcp.gestures import SequenceRequest, plan_sequence
from artemis.mcp.gestures import plan_sequence, validate_sequence
from artemis.runtime.task_control import TaskControl, has_pending_instruction
try:
limit = load_agent_config().pro.execution.max_burst_actions
planned = plan_sequence(
SequenceRequest.model_validate({"steps": steps}),
validate_sequence({"steps": steps}, limit),
*self._dims(),
limit=load_agent_config().pro.execution.max_burst_actions,
limit=limit,
)
except ValueError as exc:
return ActionResult.failure("action_sequence", str(exc), code=ActionCode.INVALID_ARGS)
@@ -178,6 +279,7 @@ class AdbActuator:
{**step, "index": i + 1, "execution": "not_started"} for i, step in enumerate(planned)
]
failed = None
failed_code = ActionCode.DEVICE_ERROR
for step, receipt in zip(planned, receipts, strict=True):
control = getattr(self.ctx, "task_control", None)
if isinstance(control, TaskControl) and (
@@ -192,17 +294,65 @@ class AdbActuator:
span.payload = {"args": step}
receipt["started_ms"] = round((time.monotonic() - started) * 1000, 3)
receipt["execution"] = "unknown"
member_execution = "completed"
try:
if step["action"] == "click":
result = await self.controller.tap_at(*step["coordinates"])
error = getattr(result, "error", None)
guard = None
elements = None
error = None
if step.get("unless_visible") or step.get("target_resource_id"):
receipt["execution"] = "not_started"
elements = await self._live_ui_elements()
if resource_id := step.get("unless_visible"):
guard = await self._visibility_guard(resource_id, elements)
receipt["visibility_guard"] = guard
if guard and guard["visible"]:
member_execution = "skipped"
error = None
else:
if resource_id := step.get("target_resource_id"):
try:
error = await self._tap_native_target(
resource_id,
receipt,
times=step.get("times", 1),
delay_ms=step.get("delay_ms", 100),
elements=elements,
)
except LookupError as exc:
error = str(exc)
member_execution = receipt["execution"]
failed_code = ActionCode.TARGET_NOT_FOUND
else:
receipt["execution"] = "unknown"
repeats = (
{"times": step["times"], "delay_ms": step["delay_ms"]}
if "times" in step
else {}
)
result = await self.controller.tap_at(
*receipt["coordinates"], **repeats
)
error = getattr(result, "error", None)
elif step["action"] == "drag":
result = await self._dispatch_drag(step)
error = None if result.ok else result.message
if not result.ok:
member_execution = result.metadata.get("execution", "unknown")
failed_code = result.code
if guard_result := result.metadata.get("visibility_guard"):
receipt["visibility_guard"] = guard_result
else:
await asyncio.sleep(step["time_in_ms"] / 1000)
error = None
receipt.update(execution="unknown" if error else "completed", error=error)
receipt.update(
execution=(
member_execution
if member_execution in {"skipped", "not_started"} or not error
else "unknown"
),
error=error,
)
if error:
failed = str(error)
span.status = "failed"
@@ -211,7 +361,7 @@ class AdbActuator:
receipt.update(execution="unknown", error="Input cancelled; effect uncertain.")
raise
except (OSError, RuntimeError) as exc:
receipt.update(execution="unknown", error=str(exc))
receipt.update(error=str(exc))
failed = str(exc)
span.status = "failed"
span.error = failed
@@ -222,13 +372,17 @@ class AdbActuator:
if failed:
break
completed = sum(r["execution"] == "completed" for r in receipts)
attempted = any(r["action"] != "wait" and r["execution"] != "not_started" for r in receipts)
skipped = sum(r["execution"] == "skipped" for r in receipts)
attempted = any(
r["action"] != "wait" and r["execution"] in {"completed", "unknown"} for r in receipts
)
execution = "completed" if not failed else ("partial" if completed else "unknown")
if failed and not attempted:
execution = "not_started"
metadata = {
"sequence_results": receipts,
"completed_steps": completed,
"skipped_steps": skipped,
"execution": execution,
"side_effect": "possible" if attempted else "none",
}
@@ -237,11 +391,12 @@ class AdbActuator:
"action_sequence",
f"Sequence stopped after {completed}/{len(planned)} completed steps: {failed}"
" Remaining steps were not executed; do not replay completed inputs.",
code=failed_code,
metadata=metadata,
)
return ActionResult.success(
"action_sequence",
f"Executed {completed} gesture steps consecutively. Verify the final screen.",
f"Executed {completed} gesture steps consecutively; skipped {skipped} unnecessary opener clicks. Verify actual ticks/handle endpoints in the final screen.",
metadata=metadata,
)
@@ -270,8 +425,49 @@ class AdbActuator:
# --- Optional actions ------------------------------------------------------------
async def click(self, nx: int, ny: int, times: int = 1, delay_ms: int = 100) -> ActionResult:
async def click(
self,
nx: int,
ny: int,
times: int = 1,
delay_ms: int = 100,
*,
expected_resource_id: str | None = None,
) -> ActionResult:
x, y = self._to_px(nx, ny)
if expected_resource_id:
receipt: dict = {}
try:
error = await self._tap_native_target(
expected_resource_id,
receipt,
times=times,
delay_ms=delay_ms,
expected_point=(x, y),
)
except (LookupError, OSError, RuntimeError) as exc:
return ActionResult.failure(
"click",
f"Observed target could not be confirmed for the next tap; completed {receipt.get('completed_taps', 0)} taps: {exc}. Re-observe. For an auto-hiding panel use one guarded action_sequence containing its opener and target, using observed resource IDs.",
code=ActionCode.TARGET_NOT_FOUND
if isinstance(exc, LookupError)
else ActionCode.DEVICE_ERROR,
metadata={
**receipt,
"side_effect": "none"
if receipt.get("execution") == "not_started"
else "possible",
"expected_resource_id": expected_resource_id,
},
)
if error:
return ActionResult.failure("click", error, metadata=receipt)
return ActionResult.success(
"click",
f"Tapped at [{nx}, {ny}] (normalized).",
normalized_coordinates=[int(nx), int(ny)],
metadata={**receipt, "execution": "completed"},
)
result = await self.controller.tap_at(x, y, times=times, delay_ms=delay_ms)
err = getattr(result, "error", None) if result else None
if err is not None:
artemis/mcp/actuators/base.py
+7 -1
View File
@@ -64,7 +64,13 @@ class Actuator(Protocol):
# --- Optional actions ------------------------------------------------------------
async def click(
self, nx: int, ny: int, times: int = 1, delay_ms: int = 100
self,
nx: int,
ny: int,
times: int = 1,
delay_ms: int = 100,
*,
expected_resource_id: str | None = None,
) -> ActionResult: ...
async def long_press(self, nx: int, ny: int, duration_ms: int = 1000) -> ActionResult: ...
artemis/mcp/gestures.py
+48 -2
View File
@@ -51,6 +51,11 @@ class DragRequest(GestureModel):
calibration: DragCalibration | None = Field(
default=None, description="Computes the endpoint from observed anchors; then set end=null."
)
require_visible: str | None = Field(
default=None,
min_length=1,
description="Exact previously observed resource_id of a panel control. Check live visibility before dragging; refuse input if absent. Use for auto-hiding controls.",
)
@model_validator(mode="after")
def endpoint_or_calibration(self):
@@ -63,9 +68,24 @@ class DragRequest(GestureModel):
class ClickStep(GestureModel):
action: Literal["click"]
target: Point
target: Point | Annotated[str, Field(min_length=1)] = Field(
description="Prefer an exact observed resource_id: resolve its live bounds before tapping, refusing absent/ambiguous targets. Otherwise use [x,y] in coordinate_space; never mix screenshot pixels with normalized coordinates."
)
target_description: str = Field(min_length=1)
coordinate_space: CoordinateSpace = "normalized"
times: int = Field(
default=1,
strict=True,
ge=1,
le=10,
description="Consecutive taps within this ONE member, e.g. 10 for a scale +/- button; no model turns between taps.",
)
delay_ms: int = Field(default=100, strict=True, ge=0, le=1000)
unless_visible: str | None = Field(
default=None,
min_length=1,
description="Exact previously observed resource_id inside a toggled panel (not the opener). Skip this opener click when that control is already visible; otherwise click. Avoid closing a still-open panel.",
)
class DragStep(DragRequest):
@@ -84,6 +104,17 @@ class SequenceRequest(GestureModel):
steps: list[GestureStep] = Field(min_length=1, max_length=10)
def validate_sequence(payload: dict, limit: int) -> SequenceRequest:
steps = payload.get("steps")
if isinstance(steps, list) and len(steps) > limit:
raise ValueError(
f"Sequence has {len(steps)} members; this task allows at most {limit}, including waits. "
"No input executed. Split into complete short open/wait/operate groups; "
"do not fall back to separate model turns for auto-hiding targets."
)
return SequenceRequest.model_validate(payload)
def pixel_point(point: list[int], space: CoordinateSpace, width: int, height: int) -> list[int]:
if width <= 0 or height <= 0:
raise ValueError("Screen dimensions must be positive.")
@@ -162,6 +193,7 @@ def plan_drag(request: DragRequest, width: int, height: int) -> dict:
],
"duration_ms": request.duration_ms,
"target_description": request.target_description,
**({"require_visible": request.require_visible} if request.require_visible else {}),
**({"calibration": calibration} if calibration else {}),
}
@@ -177,9 +209,23 @@ def plan_sequence(request: SequenceRequest, width: int, height: int, limit: int)
planned.append(
{
"action": "click",
"coordinates": pixel_point(step.target, step.coordinate_space, width, height),
**(
{"target_resource_id": step.target}
if isinstance(step.target, str)
else {
"coordinates": pixel_point(
step.target, step.coordinate_space, width, height
)
}
),
"coordinate_space": "pixel",
"target_description": step.target_description,
**(
{"times": step.times, "delay_ms": step.delay_ms}
if step.times != 1 or step.delay_ms != 100
else {}
),
**({"unless_visible": step.unless_visible} if step.unless_visible else {}),
}
)
else: