From c76c9bb7e8d774145756e70652d7d7d65eecd14e Mon Sep 17 00:00:00 2001 From: somew1nd Date: Tue, 1 Sep 2026 17:57:15 -0700 Subject: [PATCH 1/2] refactor(errors): replace silent exception handling in console and server --- apps/admin_console/database/connection.py | 11 ++- .../repositories/session_repository.py | 14 +++- .../database/repositories/step_repository.py | 12 ++- .../database/repositories/trace_repository.py | 6 +- apps/admin_console/replay_manager.py | 56 +++++++++---- apps/admin_console/routers/sessions.py | 9 ++- apps/admin_console/routers/steps.py | 3 +- apps/admin_console/routers/tasks.py | 3 +- apps/admin_console/services/ipc_service.py | 18 ++++- apps/admin_console/services/media_service.py | 27 +++++-- apps/admin_console/services/model_service.py | 15 +++- .../services/task_queue_service.py | 81 ++++++++++++++----- mcp_server/background/task_runner.py | 2 +- mcp_server/base.py | 2 + mcp_server/notifiers/agentapi.py | 17 ++-- mcp_server/server.py | 5 +- mcp_server/tools/inspect_trace.py | 21 +++-- mcp_server/tools/task_manager.py | 56 ++++++++----- mcp_server/tools/task_runner.py | 13 ++- mcp_server/utils/device_utils.py | 3 +- 20 files changed, 267 insertions(+), 107 deletions(-) diff --git a/apps/admin_console/database/connection.py b/apps/admin_console/database/connection.py index 2590b35..83eeedd 100644 --- a/apps/admin_console/database/connection.py +++ b/apps/admin_console/database/connection.py @@ -13,11 +13,13 @@ # limitations under the License. from contextlib import contextmanager +import logging from pathlib import Path import sqlite3 from artemis.config import DB_PATH +logger = logging.getLogger(__name__) _initialized_dbs = set() @@ -33,14 +35,17 @@ def get_db(db_path: Path | None = None) -> sqlite3.Connection: StorageManager(db_path=path, base_trace_dir=path.parent) _initialized_dbs.add(path_key) - except Exception: - pass + except (ImportError, OSError, sqlite3.Error) as exc: + # Schema bootstrap is best-effort here; queries against a missing + # schema will surface their own errors, but log the root cause. + logger.warning("Schema bootstrap failed for %s: %s", path, exc) conn = sqlite3.connect(path, timeout=10.0) conn.row_factory = sqlite3.Row try: conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=10000") - except Exception: + except sqlite3.Error: + # Optional performance PRAGMAs; the connection works without them. pass return conn diff --git a/apps/admin_console/database/repositories/session_repository.py b/apps/admin_console/database/repositories/session_repository.py index b51d512..a3937e7 100644 --- a/apps/admin_console/database/repositories/session_repository.py +++ b/apps/admin_console/database/repositories/session_repository.py @@ -12,11 +12,14 @@ # See the License for the specific language governing permissions and # limitations under the License. +import logging import time from typing import Any from artemis.runtime import trace_store +logger = logging.getLogger(__name__) + try: from admin_console.database.connection import db_session except ImportError: @@ -317,8 +320,15 @@ class SessionRepository: "failed", error="Process terminated prior to server startup.", ) - except Exception: - pass + except OSError as exc: + # read_status itself never raises; this guards the + # lock/write side of update_trace_status. + logger.warning( + "Could not mark trace %s failed during orphan " + "reconciliation: %s", + row["session_id"], + exc, + ) conn.commit() return count except Exception: diff --git a/apps/admin_console/database/repositories/step_repository.py b/apps/admin_console/database/repositories/step_repository.py index 3d18126..515a987 100644 --- a/apps/admin_console/database/repositories/step_repository.py +++ b/apps/admin_console/database/repositories/step_repository.py @@ -106,7 +106,8 @@ class StepRepository: if isinstance(payload_raw, str): try: payload_obj = json.loads(payload_raw) - except Exception: + except (ValueError, TypeError): + # Non-JSON payload: the isinstance(dict) guard below bails out. pass if not isinstance(payload_obj, dict): return 0, 0, 0 @@ -175,7 +176,8 @@ class StepRepository: if isinstance(payload_raw, str): try: payload_obj = json.loads(payload_raw) - except Exception: + except (ValueError, TypeError): + # Non-JSON payload: the isinstance(dict) guard below bails out. pass if not isinstance(payload_obj, dict): @@ -308,7 +310,8 @@ class StepRepository: if step_dict.get(json_col): try: step_dict[json_col] = json.loads(step_dict[json_col]) - except Exception: + except (ValueError, TypeError): + # Non-JSON column value: keep the raw string. pass # If post_image_name is identical to pre_image_name, normalize to None @@ -471,7 +474,8 @@ class StepRepository: p_raw_th.append(str(th_text).strip()) elif thr["type"] == "thinking" and th_text: p_nat_th.append(str(th_text).strip()) - except Exception: + except (ValueError, TypeError, AttributeError): + # Unparseable or non-dict thinking payload: skip it. pass virtual_step = { diff --git a/apps/admin_console/database/repositories/trace_repository.py b/apps/admin_console/database/repositories/trace_repository.py index 95225c5..7bc0dbc 100644 --- a/apps/admin_console/database/repositories/trace_repository.py +++ b/apps/admin_console/database/repositories/trace_repository.py @@ -52,7 +52,8 @@ class TraceRepository: if row_dict.get("payload"): try: row_dict["payload"] = json.loads(row_dict["payload"]) - except Exception: + except (ValueError, TypeError): + # Non-JSON payload: keep the raw string for display. pass nodes[row_dict["trace_id"]] = {**row_dict, "children": []} @@ -94,7 +95,8 @@ class TraceRepository: if row_dict.get("payload"): try: row_dict["payload"] = json.loads(row_dict["payload"]) - except Exception: + except (ValueError, TypeError): + # Non-JSON payload: keep the raw string for display. pass nodes[row_dict["trace_id"]] = {**row_dict, "children": []} diff --git a/apps/admin_console/replay_manager.py b/apps/admin_console/replay_manager.py index 6202ea8..7174d64 100644 --- a/apps/admin_console/replay_manager.py +++ b/apps/admin_console/replay_manager.py @@ -485,7 +485,8 @@ class ReplayManager: if td.get("payload"): try: td["payload"] = json.loads(td["payload"]) - except Exception: + except (ValueError, TypeError): + # Non-JSON payload: keep the raw string. pass traces.append(td) conn.close() @@ -721,8 +722,11 @@ class ReplayManager: with open(traces_json_path, encoding="utf-8") as f: traces = json.load(f) return {str(t.get("trace_id")).lower() for t in traces if t.get("trace_id")} - except Exception: - pass + except (OSError, ValueError) as e: + print( + f"Warning: Failed to read preloaded trace ids from {traces_json_path}: {e}", + file=sys.stderr, + ) return set() def load_traces_from_db(self, db_path: Path, step_num: int, preloaded_trace_ids: set) -> list: @@ -747,7 +751,8 @@ class ReplayManager: if td.get("payload"): try: td["payload"] = json.loads(td["payload"]) - except Exception: + except (ValueError, TypeError): + # Non-JSON payload: normalized to {} just below. pass if not isinstance(td.get("payload"), dict): td["payload"] = {} @@ -808,7 +813,8 @@ class ReplayManager: session_id = None try: session_id = temp_db_path.parents[1].name.split("_")[0] - except Exception: + except IndexError: + # Path too shallow to carry a session dir: fall back to the DB probe. pass if not session_id: @@ -820,7 +826,8 @@ class ReplayManager: if row: session_id = str(row[0]) conn.close() - except Exception: + except sqlite3.Error: + # Unreadable sandbox DB: fall back to single-step trace loading. pass if not session_id: @@ -890,8 +897,11 @@ class ReplayManager: with open(step_json_path, encoding="utf-8") as f: step_data = json.load(f) current_step_id = step_data.get("step_id") - except Exception: - pass + except (OSError, ValueError) as e: + print( + f"Warning: Failed to read step id from {step_json_path}: {e}", + file=sys.stderr, + ) current_step_id_str = str(current_step_id).lower() if current_step_id else None @@ -995,7 +1005,9 @@ class ReplayManager: sid = step_data.get("step_id") if sid: step_dirs[str(sid).lower()] = subdir - except Exception: + except (OSError, ValueError): + # Unreadable step.json: screenshots for this step + # simply are not linked into the trace payloads. pass for td in all_traces: @@ -1043,7 +1055,8 @@ class ReplayManager: return ( f"file://{abs_path}" if abs_path.startswith("/") else f"file:///{abs_path}" ) - except Exception: + except (OSError, ValueError): + # Not a resolvable filesystem path: return the value verbatim. pass return val @@ -1104,7 +1117,8 @@ class ReplayManager: if r["payload"]: try: payload = json.loads(r["payload"]) - except Exception: + except (ValueError, TypeError): + # Non-JSON payload: treated as an empty tool payload. pass response_val = payload.get("result") @@ -1116,7 +1130,8 @@ class ReplayManager: ): try: response_val = json.loads(response_val) - except Exception: + except ValueError: + # Looks like JSON but is not: keep the raw string. pass args = payload.get("args", {}) @@ -1682,7 +1697,8 @@ class ReplayManager: if row and row[0]: try: device_info = json.loads(row[0]) - except Exception: + except (ValueError, TypeError): + # Malformed device_info: simulate with the defaults below. pass sim_device_id = device_info.get("device_id", "replay-device") @@ -2294,8 +2310,11 @@ class ReplayManager: res_data = json.load(f) replay_outcome = res_data.get("result") replay_id = res_data.get("replay_id") - except Exception: - pass + except (OSError, ValueError) as e: + print( + f"Warning: Failed to read replay result from {result_json_path}: {e}", + file=sys.stderr, + ) # Update virtual tool trace result with the actual formatted outcome if available if replay_outcome is not None: @@ -2464,8 +2483,11 @@ class ReplayManager: llm_calls_log = res_data.get("llm_calls") or [] replay_outcome = res_data.get("result") replay_id = res_data.get("replay_id") - except Exception: - pass + except (OSError, ValueError) as e: + print( + f"Warning: Failed to read replay result from {result_json_path}: {e}", + file=sys.stderr, + ) # Update virtual tool trace result with the actual formatted outcome if available if replay_outcome is not None: diff --git a/apps/admin_console/routers/sessions.py b/apps/admin_console/routers/sessions.py index 3ad3e4b..14ccc06 100644 --- a/apps/admin_console/routers/sessions.py +++ b/apps/admin_console/routers/sessions.py @@ -83,7 +83,8 @@ def _list_sessions_sync(): d_info = json.loads(d_info_raw) if isinstance(d_info_raw, str) else d_info_raw if isinstance(d_info, dict): device_id = d_info.get("device_id") or d_info.get("device_serial") - except Exception: + except (ValueError, TypeError): + # Malformed device_info JSON: fall back to the recording's device_id. pass if not device_id and recording: device_id = recording.get("device_id") @@ -157,8 +158,10 @@ def _list_sessions_sync(): "failed", error="Process terminated unexpectedly (auto-harvested).", ) - except Exception: - pass + except OSError as e: + # read_status never raises; this guards the lock/write side + # of update_trace_status. + print(f"[list_sessions] Could not mark harvested traces failed: {e}") except Exception as e: print(f"[list_sessions] Failed to update orphaned sessions: {e}") diff --git a/apps/admin_console/routers/steps.py b/apps/admin_console/routers/steps.py index 4e4c31f..d682396 100644 --- a/apps/admin_console/routers/steps.py +++ b/apps/admin_console/routers/steps.py @@ -78,7 +78,8 @@ async def get_trace(trace_id: str, session_id: str = None, step_number: int = No try: payload_obj = json.loads(trace_dict["payload"]) trace_dict["payload"] = media_service.unwrap_payload(payload_obj) - except Exception: + except (ValueError, TypeError, KeyError, AttributeError): + # Non-JSON or unexpectedly shaped payload: serve it raw. pass return trace_dict diff --git a/apps/admin_console/routers/tasks.py b/apps/admin_console/routers/tasks.py index fd18f35..d732958 100644 --- a/apps/admin_console/routers/tasks.py +++ b/apps/admin_console/routers/tasks.py @@ -209,7 +209,8 @@ async def stop_task( target_sid = str(body["session_id"]) if body.get("device_id"): target_dev = str(body["device_id"]) - except Exception: + except ValueError: + # Empty or non-JSON body: fall back to the query parameters. pass stopped = task_queue_service.stop_tasks( diff --git a/apps/admin_console/services/ipc_service.py b/apps/admin_console/services/ipc_service.py index f93accb..712b1d8 100644 --- a/apps/admin_console/services/ipc_service.py +++ b/apps/admin_console/services/ipc_service.py @@ -15,8 +15,11 @@ import asyncio from contextlib import suppress import json +import logging from typing import Any +logger = logging.getLogger(__name__) + try: from admin_console.core.state import state except ImportError: @@ -76,7 +79,8 @@ class IPCService: if isinstance(payload, str): try: payload = json.loads(payload) - except Exception: + except ValueError: + # Non-JSON payload string: cleaned as-is below. pass if tr_type in ("tool", "action") and isinstance(payload, dict): @@ -102,7 +106,8 @@ class IPCService: if isinstance(pl, str): try: pl = json.loads(pl) - except Exception: + except ValueError: + # Non-JSON payload string: cleaned as-is below. pass if tr_dict.get("type") == "llm_call": @@ -190,7 +195,14 @@ class IPCService: try: cb(event_type, data) except Exception: - pass + # One broken subscriber must not block the + # others, but a silent drop hides it entirely. + logger.warning( + "IPC subscriber %r failed for event %s", + cb, + event_type, + exc_info=True, + ) except (ValueError, UnicodeDecodeError, TypeError) as e: print(f"IPC parse error: {e}") except asyncio.CancelledError: diff --git a/apps/admin_console/services/media_service.py b/apps/admin_console/services/media_service.py index 96eb877..7b8940c 100644 --- a/apps/admin_console/services/media_service.py +++ b/apps/admin_console/services/media_service.py @@ -15,6 +15,7 @@ import base64 import hashlib import json +import logging import os from pathlib import Path import subprocess @@ -25,6 +26,8 @@ from fastapi import HTTPException from artemis.config import IMAGES_DIR, TRACES_PATH, WORKSPACE_ROOT +logger = logging.getLogger(__name__) + class MediaService: """Service handling media indexing, local file security, payload unwrapping and notes/plans.""" @@ -76,7 +79,9 @@ class MediaService: str(result), ) return result - except Exception: + except OSError: + # Filesystem probe failed (missing/locked file): serve the + # original path and let the media endpoint report the error. pass return p @@ -167,8 +172,9 @@ class MediaService: url = cls.path_to_video_url(item) idx[item.stem] = url idx[item.name] = url - except Exception: - pass + except OSError as exc: + # Unreadable directory entry: index the rest of the tree. + logger.warning("Video index scan failed under %s: %s", d, exc) return idx @classmethod @@ -219,7 +225,8 @@ class MediaService: item = cls.ensure_browser_playable_video(item) v_url = cls.path_to_video_url(item) break - except Exception: + except OSError: + # Unreadable session dir: no recording to resolve. pass if not v_url and row_dict.get("initial_goal"): @@ -274,7 +281,8 @@ class MediaService: obj.startswith("[") and obj.endswith("]") ): return MediaService.unwrap_payload(json.loads(obj)) - except Exception: + except ValueError: + # Looks like JSON but is not: treat it as a plain string. pass # Check if the string is a Base64 image @@ -284,7 +292,8 @@ class MediaService: try: _, base64_data = obj.split(";base64,", 1) is_base64_img = True - except Exception: + except ValueError: + # Malformed data URI: fall through to plain-string handling. pass elif obj.startswith("iVBORw0KGgo") or obj.startswith("/9j/"): is_base64_img = True @@ -301,8 +310,10 @@ class MediaService: image_path.write_bytes(image_bytes) return f"image://{image_hash}" - except Exception: - pass + except (ValueError, OSError) as exc: + # Invalid base64 or image cache write failure: fall back + # to the truncated raw string below. + logger.debug("Could not persist inline image: %s", exc) if len(obj) > 2000: preview_start = obj[:100].replace("\n", " ") diff --git a/apps/admin_console/services/model_service.py b/apps/admin_console/services/model_service.py index 4206ca7..0ccc0bc 100644 --- a/apps/admin_console/services/model_service.py +++ b/apps/admin_console/services/model_service.py @@ -13,9 +13,12 @@ # limitations under the License. import json +import logging import time from typing import Any +logger = logging.getLogger(__name__) + class ModelService: """Service handling agent architecture profile detection and metadata formatting.""" @@ -44,8 +47,10 @@ class ModelService: elif llm_cfg and llm_cfg.default: provider = str(llm_cfg.default.provider) model_id = str(llm_cfg.default.model) - except Exception: - pass + except Exception as exc: + # Unreadable/invalid LLM config: fall back to display defaults + # (cached for the TTL, so this logs at most once per window). + logger.warning("Could not resolve LLM config for display: %s", exc) cls._llm_info_cache = (now, provider, model_id) return provider, model_id @@ -94,7 +99,8 @@ class ModelService: return "pro" elif "flash" in p_str: return "flash" - except Exception: + except (ValueError, TypeError): + # Malformed device_info JSON: try the other profile sources. pass if (row_dict.get("status") == "running") and running_profile: @@ -132,7 +138,8 @@ class ModelService: for x in ("planner", "validator", "operator", "checker") ): return "pro" - except Exception: + except (ValueError, TypeError): + # Malformed trace payload JSON: skip this trace. pass return None diff --git a/apps/admin_console/services/task_queue_service.py b/apps/admin_console/services/task_queue_service.py index b10bf08..9881052 100644 --- a/apps/admin_console/services/task_queue_service.py +++ b/apps/admin_console/services/task_queue_service.py @@ -15,6 +15,7 @@ import asyncio import codecs from datetime import datetime +import logging import os import psutil import shutil @@ -48,6 +49,8 @@ from artemis.runtime import ( trace_store, ) +logger = logging.getLogger(__name__) + class TaskQueueService: """Service managing FIFO task execution, background worker, subprocess lifecycle, @@ -76,7 +79,14 @@ class TaskQueueService: try: cb(event_type, data) except Exception: - pass + # One broken subscriber must not block the others, but a + # silent drop hides it entirely. + logger.warning( + "Event subscriber %r failed for event %s", + cb, + event_type, + exc_info=True, + ) @classmethod def _broadcast_startup_progress( @@ -194,7 +204,9 @@ class TaskQueueService: if log_file is not None: try: log_file.write(text) - except Exception: + except (OSError, ValueError): + # Best-effort tee into stdout.log; console output above + # already carried the text. pass try: @@ -214,7 +226,8 @@ class TaskQueueService: if log_file is not None: try: log_file.close() - except Exception: + except OSError: + # Flush-on-close of the best-effort tee failed; nothing to do. pass @staticmethod @@ -265,7 +278,8 @@ class TaskQueueService: return proc.returncode if proc.returncode is not None else -15 except (psutil.NoSuchProcess, ProcessLookupError): return proc.returncode if proc.returncode is not None else -15 - except Exception: + except psutil.Error: + # Transient probe failure (e.g. AccessDenied): keep waiting. pass else: return proc.returncode if proc.returncode is not None else -15 @@ -279,7 +293,9 @@ class TaskQueueService: state.worker_task = loop.create_task(cls.queue_worker()) try: state.wake_event.set() - except Exception: + except AttributeError: + # Partially initialized state (unit tests): the worker + # loop polls anyway. pass except RuntimeError: pass @@ -515,8 +531,11 @@ class TaskQueueService: if st: st["pid"] = proc.pid trace_store.write_status(str(sess_id), st) - except Exception: - pass + except OSError as exc: + # Status probes fall back to DB PID bookkeeping, but note it. + print( + f"[QueueWorker] Could not record worker pid in status.json for {sess_id}: {exc}" + ) cls._broadcast_startup_progress( sess_id, "process_ready", "Execution process started" ) @@ -583,8 +602,12 @@ class TaskQueueService: str(sess_id), canonical_mcp_status, ) - except Exception: - pass + except OSError as exc: + # A stale status.json would leave MCP pollers seeing + # "running" forever, so surface the failure. + print( + f"[QueueWorker] Could not persist terminal MCP status for {sess_id}: {exc}" + ) rec_info = session_repo.get_video_recording_for_session(sess_id) rec_status = (rec_info or {}).get("status") recovered_video_path = None @@ -848,8 +871,9 @@ class TaskQueueService: active_owners: dict[str, Any] = {} try: active_owners = DeviceExecutionLock.get_active_owners() - except Exception: - pass + except OSError as exc: + # Unreadable lock dir: fall back to the single-owner probe below. + print(f"[stop_tasks] Could not enumerate device owners: {exc}") fallback = DeviceExecutionLock.get_active_owner() if fallback and not active_owners: active_owners["default"] = fallback @@ -872,8 +896,10 @@ class TaskQueueService: "cancelled", error="Task stopped from the Artemis frontend.", ) - except Exception: - pass + except OSError as exc: + print( + f"[stop_tasks] Could not persist cancelled MCP status for {sid}: {exc}" + ) cls._broadcast_event( "session_ended", { @@ -895,7 +921,8 @@ class TaskQueueService: if run_proc is not None and run_proc.returncode is None: try: run_proc.kill() - except Exception: + except (ProcessLookupError, OSError): + # Process already exited between the check and the kill. pass # active_runs entries are popped by each run's finalizer once the # process exit is observed; clearing them here would free the device @@ -903,7 +930,8 @@ class TaskQueueService: if state.current_process: try: state.current_process.kill() - except Exception: + except (ProcessLookupError, OSError): + # Process already exited between the check and the kill. pass state.current_process = None @@ -915,7 +943,9 @@ class TaskQueueService: if PAUSE_FILE.exists(): try: PAUSE_FILE.unlink() - except Exception: + except OSError: + # Best-effort cleanup of the pause marker; a leftover file + # only pauses until the next resume request. pass cls.ensure_worker_running() @@ -926,8 +956,9 @@ class TaskQueueService: active_owners = {} try: active_owners = DeviceExecutionLock.get_active_owners() - except Exception: - pass + except OSError as exc: + # Unreadable lock dir: the local-process fallbacks below still apply. + print(f"[stop_tasks] Could not enumerate device owners: {exc}") owner = None if target_sid: @@ -1066,7 +1097,8 @@ class TaskQueueService: try: local_proc.kill() stopped = True - except Exception: + except (ProcessLookupError, OSError): + # Process already exited; terminate_tree above may have got it. pass stopped = True is_local_owner = True @@ -1097,8 +1129,11 @@ class TaskQueueService: if row_pid and session_repo.process_is_alive(row_pid): try: process_supervisor.terminate_tree(int(row_pid)) - except Exception: - pass + except Exception as exc: + # Report it: a surviving worker keeps the device busy. + print( + f"[stop_tasks] Could not terminate worker pid {row_pid}: {exc}" + ) DeviceExecutionLock.cleanup_stale_locks() stopped = True @@ -1176,7 +1211,9 @@ class TaskQueueService: if PAUSE_FILE.exists(): try: PAUSE_FILE.unlink() - except Exception: + except OSError: + # Best-effort cleanup of the pause marker; a leftover file + # only pauses until the next resume request. pass cls.ensure_worker_running() diff --git a/mcp_server/background/task_runner.py b/mcp_server/background/task_runner.py index 13f6665..4d52733 100644 --- a/mcp_server/background/task_runner.py +++ b/mcp_server/background/task_runner.py @@ -93,7 +93,7 @@ def resolve_profile_file() -> str | None: os.path.join(app_dir, "llm-config.json"), ] ) - except Exception: + except (ImportError, OSError): pass candidates.extend( diff --git a/mcp_server/base.py b/mcp_server/base.py index 1d015b7..242761b 100644 --- a/mcp_server/base.py +++ b/mcp_server/base.py @@ -21,6 +21,8 @@ try: FastMCPSettings.model_rebuild() except Exception: + # Version-compat shim: older/newer FastMCP releases may not expose + # Settings or need the rebuild; the server works without it. pass # Define the shared FastMCP instance for external IDE and agent clients. diff --git a/mcp_server/notifiers/agentapi.py b/mcp_server/notifiers/agentapi.py index d0489db..8ca6a15 100644 --- a/mcp_server/notifiers/agentapi.py +++ b/mcp_server/notifiers/agentapi.py @@ -37,6 +37,8 @@ class AgentApiNotifier(BaseNotifier): if addr and token: self._save_shared_env(addr, token) except Exception: + # Environment recovery is opportunistic; notifier construction + # must never fail because of it. pass @property @@ -59,7 +61,8 @@ class AgentApiNotifier(BaseNotifier): for cand in candidates: if os.path.exists(cand) and os.access(cand, os.X_OK): return cand - except Exception: + except OSError: + # Filesystem probe failed: report the binary as absent. pass return None @@ -70,7 +73,8 @@ class AgentApiNotifier(BaseNotifier): return True if "ANTIGRAVITY_LS_ADDRESS" in os.environ: return True - except Exception: + except OSError: + # Filesystem probe failed: report the notifier as unavailable. pass return False @@ -85,7 +89,8 @@ class AgentApiNotifier(BaseNotifier): os.path.dirname(os.path.dirname(os.path.abspath(__file__))) ) shared_candidates.append(os.path.join(parent_dir, ".jetski_env")) - except Exception: + except OSError: + # Repo-relative candidate is optional; the home-dir ones remain. pass data = { @@ -128,7 +133,8 @@ class AgentApiNotifier(BaseNotifier): os.path.dirname(os.path.dirname(os.path.abspath(__file__))) ) shared_candidates.append(os.path.join(parent_dir, ".jetski_env")) - except Exception: + except OSError: + # Repo-relative candidate is optional; the home-dir ones remain. pass for shared_file in shared_candidates: @@ -190,7 +196,8 @@ class AgentApiNotifier(BaseNotifier): ) if local_addr and local_token: proc_matches.append((mtime, local_addr, local_token)) - except Exception: + except OSError: + # /proc entries vanish or deny access as processes exit. pass # Sort matches descending by creation timestamp so active/newest sessions come first diff --git a/mcp_server/server.py b/mcp_server/server.py index 2c1da04..8353c8b 100644 --- a/mcp_server/server.py +++ b/mcp_server/server.py @@ -36,7 +36,8 @@ def _rotate_log(path: str, max_bytes: int = _LOG_MAX_BYTES) -> None: if os.path.exists(old_path): os.remove(old_path) os.replace(path, old_path) - except Exception: + except OSError: + # Best-effort rotation; a locked or vanished log file is tolerable. pass @@ -74,6 +75,8 @@ try: sys.stderr = StderrTee(sys.stderr, os.path.join(_mcp_log_dir, "mcp_stderr.log")) except Exception: + # Debug logging and the stderr tee are optional diagnostics; they must + # never prevent the MCP stdio server from starting. pass # 1. Import shared FastMCP instance diff --git a/mcp_server/tools/inspect_trace.py b/mcp_server/tools/inspect_trace.py index 6ee5d88..3e8358f 100644 --- a/mcp_server/tools/inspect_trace.py +++ b/mcp_server/tools/inspect_trace.py @@ -121,7 +121,8 @@ async def mobile_inspect_trace(action: str, trace_id: str, step_number: int | No try: with open(plan_path, encoding="utf-8") as f: plan_content = f.read() - except Exception: + except OSError: + # Unreadable plan file: the summary just omits the plan. pass try: @@ -149,7 +150,8 @@ async def mobile_inspect_trace(action: str, trace_id: str, step_number: int | No if step_dict.get(col): try: step_dict[col] = json.loads(step_dict[col]) - except Exception: + except (ValueError, TypeError): + # Non-JSON column value: keep the raw string. pass cursor.execute( @@ -163,7 +165,8 @@ async def mobile_inspect_trace(action: str, trace_id: str, step_number: int | No if t_dict.get("payload"): try: t_dict["payload"] = json.loads(t_dict["payload"]) - except Exception: + except (ValueError, TypeError): + # Non-JSON payload: keep the raw string. pass events.append(t_dict) step_dict["interleaved_events"] = events @@ -200,10 +203,12 @@ async def mobile_inspect_trace(action: str, trace_id: str, step_number: int | No d_info = json.loads(row_dev["device_info"]) if isinstance(d_info, dict) and d_info.get("device_id"): device_serial = d_info["device_id"] - except Exception: + except (ValueError, TypeError): + # Malformed device_info JSON: leave the serial unset. pass conn_dev.close() - except Exception: + except sqlite3.Error: + # The serial enriches the summary header only; skip it. pass device_info_str = f" | **Device Serial:** `{device_serial}`" if device_serial else "" @@ -314,7 +319,8 @@ async def mobile_inspect_trace(action: str, trace_id: str, step_number: int | No if step_dict.get(col): try: step_dict[col] = json.loads(step_dict[col]) - except Exception: + except (ValueError, TypeError): + # Non-JSON column value: keep the raw string. pass cursor.execute( @@ -335,7 +341,8 @@ async def mobile_inspect_trace(action: str, trace_id: str, step_number: int | No if t_dict.get("payload"): try: t_dict["payload"] = json.loads(t_dict["payload"]) - except Exception: + except (ValueError, TypeError): + # Non-JSON payload: keep the raw string. pass events.append(t_dict) step_dict["interleaved_events"] = events diff --git a/mcp_server/tools/task_manager.py b/mcp_server/tools/task_manager.py index 41ac4c9..d38be80 100644 --- a/mcp_server/tools/task_manager.py +++ b/mcp_server/tools/task_manager.py @@ -64,14 +64,16 @@ def _session_tracked_by_lock(trace_id: str) -> bool: for q_item in DeviceExecutionLock.get_queued_tasks(): if str(q_item.get("session_id")) == trace_id: return True - except Exception: - pass + except Exception as exc: + # An unreadable queue may hide a live session; the caller then leans + # on DB/pid evidence, so surface the probe failure. + print(f"Could not read device queue for {trace_id}: {exc}", file=sys.stderr) try: for owner in DeviceExecutionLock.get_active_owners().values(): if owner.session_id and str(owner.session_id) == trace_id: return True - except Exception: - pass + except Exception as exc: + print(f"Could not read device owners for {trace_id}: {exc}", file=sys.stderr) return False @@ -115,8 +117,9 @@ def _reconcile_task_state( if row: db_status = row["status"] db_pid = row["pid"] - except Exception: - pass + except sqlite3.Error as exc: + # Reconciliation then relies on status.json and lock evidence only. + print(f"Could not read DB session row for {trace_id}: {exc}", file=sys.stderr) if db_pid and db_pid != pid: pid = db_pid @@ -188,6 +191,8 @@ def _mark_liveness_failure(trace_id: str, status_data: dict[str, Any]) -> None: payload={"trace_id": trace_id, "error": _LIVENESS_FAILURE_ERROR}, ) except Exception: + # Notification dispatch is best-effort; the failure verdict is + # already persisted in status.json. pass @@ -278,10 +283,13 @@ def mobile_manage_task( device_serial = d_info["device_id"] status_data["device_serial"] = device_serial trace_store.write_status(trace_id, status_data) - except Exception: + except (ValueError, TypeError, OSError): + # Malformed device_info or failed cache write: the + # serial simply stays unknown for this poll. pass conn_dev.close() - except Exception: + except sqlite3.Error: + # DB probe for the serial is optional enrichment. pass if not device_serial and pid: @@ -293,7 +301,8 @@ def mobile_manage_task( status_data["device_serial"] = device_serial trace_store.write_status(trace_id, status_data) break - except Exception: + except OSError: + # Lock-owner probe / cache write failed: serial stays unknown. pass response: dict[str, Any] = { @@ -329,7 +338,8 @@ def mobile_manage_task( "task_status": run_outcome.get("task_status"), **tests, } - except Exception: + except (OSError, ValueError, TypeError): + # Unreadable or malformed run_outcome.json: omit the summary. pass break @@ -394,7 +404,8 @@ def mobile_manage_task( progress["latest_thought"] = p_obj["response"][0].get( "text", "" ) - except Exception: + except (ValueError, TypeError, AttributeError): + # Unparseable thought payload: omit it. pass cur2.execute( @@ -411,7 +422,9 @@ def mobile_manage_task( act_payload = json.loads(row_action["payload"]).get( "args", {} ) - except Exception: + except (ValueError, TypeError, AttributeError): + # Unparseable action payload: show the + # bare action name. pass progress["latest_action"] = ( f"{act_name}({act_payload})" if act_payload else act_name @@ -433,7 +446,8 @@ def mobile_manage_task( try: with open(plan_path, encoding="utf-8") as f: plan_content = f.read() - except Exception: + except OSError: + # Unreadable plan file: progress omits the plan. pass progress["task_plan"] = plan_content else: @@ -458,16 +472,18 @@ def mobile_manage_task( try: if is_daemon_running(): stopped_via_daemon = stop_task_on_daemon(trace_id) - except Exception: - pass + except Exception as exc: + # Fall through to direct process termination below. + print(f"Daemon stop attempt failed for {trace_id}: {exc}", file=sys.stderr) # 2. Cancel queue reservation if present queue_ticket = status_data.get("queue_ticket") if queue_ticket: try: DeviceExecutionLock.cancel_reservation(queue_ticket) - except Exception: - pass + except OSError as exc: + # A stale reservation expires on its own; note the failure. + print(f"Could not cancel reservation {queue_ticket}: {exc}", file=sys.stderr) # 3. If process PID is missing, try to resolve from active device owners if not pid: @@ -479,8 +495,10 @@ def mobile_manage_task( status_data["pid"] = pid trace_store.write_status(trace_id, status_data) break - except Exception: - pass + except OSError as exc: + # Without a resolved pid the stop falls back to the + # daemon/queue outcome, so surface the probe failure. + print(f"Could not resolve owner pid for {trace_id}: {exc}", file=sys.stderr) # 4. If process PID could not be determined and not stopped via daemon or queue if not pid and not stopped_via_daemon and not queue_ticket: diff --git a/mcp_server/tools/task_runner.py b/mcp_server/tools/task_runner.py index 195163e..90f66a1 100644 --- a/mcp_server/tools/task_runner.py +++ b/mcp_server/tools/task_runner.py @@ -117,6 +117,8 @@ def _watch_spawn( payload={"trace_id": trace_id, "error": error_text}, ) except Exception: + # Notification dispatch is best-effort; the failure verdict is + # already persisted in status.json. pass return False @@ -321,8 +323,12 @@ def mobile_run_task( "stdout_log": str(trace_store.get_trace_stdout_log_path(trace_id)), "stderr_log": str(trace_store.get_trace_stderr_log_path(trace_id)), } - except Exception: - pass + except Exception as exc: + # The queue probe is a rescue path for a slow daemon reply; + # its failure downgrades the verdict to "failed" below. + logging.getLogger("mcp_server").warning( + f"Could not verify daemon queue for '{trace_id}': {exc}" + ) # When Daemon is running, refuse to launch a conflicting standalone runner on the same device logging.getLogger("mcp_server").warning( @@ -392,7 +398,8 @@ def mobile_run_task( ipc_port = read_ipc_port() if ipc_port: env["ARTEMIS_IPC_PORT"] = str(ipc_port) - except Exception: + except (OSError, ValueError): + # No readable IPC port file: the runner works without live telemetry. pass proc_kwargs = env_utils.get_detached_process_kwargs() diff --git a/mcp_server/utils/device_utils.py b/mcp_server/utils/device_utils.py index d2ca5a7..5782a8d 100644 --- a/mcp_server/utils/device_utils.py +++ b/mcp_server/utils/device_utils.py @@ -167,7 +167,8 @@ def ensure_emulator( if "1" in res.stdout: time.sleep(2) # Stabilization return True - except Exception: + except (subprocess.SubprocessError, OSError): + # adb not ready yet while the emulator boots: keep polling. pass time.sleep(2) From a4c28865f2fcb6d20f732727a11c2a91690bcf0b Mon Sep 17 00:00:00 2001 From: somew1nd Date: Tue, 1 Sep 2026 18:07:45 -0700 Subject: [PATCH 2/2] refactor(core): consolidate action and plan definitions --- artemis/context.py | 1 - artemis/llm/__init__.py | 6 +- artemis/llm/router.py | 78 -- artemis/services/llm.py | 12 +- artemis/tools/mobile/exec_tools.py | 683 ------------------ artemis/utils/notes.py | 31 +- artemis/utils/task_tree.py | 6 +- .../action_surfaces/adb_server_manifest.json | 223 ++++++ tests/unit/mcp/test_action_specs.py | 11 +- tests/unit/mcp/test_adb_server_contract.py | 227 ++++++ ..._model_router.py => test_llm_grounding.py} | 59 +- tests/unit/test_tools.py | 70 -- tests/unit/tools/test_click_tool.py | 148 ---- tests/unit/tools/test_input_text_tool.py | 172 ----- tests/unit/tools/test_long_press_tool.py | 152 ---- tests/unit/tools/test_press_key_tool.py | 164 ----- tests/unit/tools/test_swipe_tool.py | 316 -------- 17 files changed, 481 insertions(+), 1878 deletions(-) delete mode 100644 artemis/tools/mobile/exec_tools.py create mode 100644 tests/fixtures/action_surfaces/adb_server_manifest.json create mode 100644 tests/unit/mcp/test_adb_server_contract.py rename tests/unit/{test_model_router.py => test_llm_grounding.py} (75%) delete mode 100644 tests/unit/test_tools.py delete mode 100644 tests/unit/tools/test_click_tool.py delete mode 100644 tests/unit/tools/test_input_text_tool.py delete mode 100644 tests/unit/tools/test_long_press_tool.py delete mode 100644 tests/unit/tools/test_press_key_tool.py delete mode 100644 tests/unit/tools/test_swipe_tool.py diff --git a/artemis/context.py b/artemis/context.py index a82d297..996b63d 100644 --- a/artemis/context.py +++ b/artemis/context.py @@ -179,7 +179,6 @@ class ArtemisContext(BaseModel): device: DeviceContext llm_config: LLMConfig | None = None - model_router: Any | None = None agent_config: Any = None adb_client: Any | None = None ui_adb_client: Any | None = None diff --git a/artemis/llm/__init__.py b/artemis/llm/__init__.py index 626456d..6cd97ee 100644 --- a/artemis/llm/__init__.py +++ b/artemis/llm/__init__.py @@ -12,18 +12,14 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Dynamic LLM Routing and Provider Integration layer.""" +"""LLM provider integration layer (endpoint descriptors and model factory).""" from artemis.llm.router import ( ModelEndpoint, ModelProvider, - ModelRouter, - default_router, ) __all__ = [ - "ModelRouter", "ModelEndpoint", "ModelProvider", - "default_router", ] diff --git a/artemis/llm/router.py b/artemis/llm/router.py index 471f820..5a48a6e 100644 --- a/artemis/llm/router.py +++ b/artemis/llm/router.py @@ -367,81 +367,3 @@ class ModelFactory: raise ValueError(f"Unsupported model provider: {provider}") -class ModelRouter: - """Dynamic router mapping agent roles to primary and fallback model endpoints.""" - - def __init__(self, default_model: str = "gemini-2.5-flash"): - self.default_endpoint = ModelEndpoint( - provider=ModelProvider.GOOGLE, - model_name=default_model, - ) - self._role_routes: dict[str, ModelEndpoint] = {} - self._fallback_routes: dict[str, list[ModelEndpoint]] = {} - self._setup_default_routes() - - def _setup_default_routes(self) -> None: - """Initializes recommended role mappings.""" - self._role_routes["planner"] = ModelEndpoint( - provider=ModelProvider.GOOGLE, - model_name="gemini-2.5-pro", - temperature=0.2, - ) - self._role_routes["operator"] = ModelEndpoint( - provider=ModelProvider.GOOGLE, - model_name="gemini-2.5-flash", - temperature=0.0, - ) - self._role_routes["validator"] = ModelEndpoint( - provider=ModelProvider.GOOGLE, - model_name="gemini-2.5-pro", - temperature=0.1, - ) - self._role_routes["summarizer"] = ModelEndpoint( - provider=ModelProvider.GOOGLE, - model_name="gemini-2.5-flash", - temperature=0.0, - ) - self._fallback_routes["planner"] = [ - ModelEndpoint(provider=ModelProvider.GOOGLE, model_name="gemini-2.5-flash"), - ] - - def set_role_route( - self, - role: str, - endpoint: ModelEndpoint | str, - fallbacks: list[ModelEndpoint] | None = None, - ) -> None: - """Configures or overrides a route for a specific agent role.""" - if isinstance(endpoint, str): - endpoint = ModelEndpoint(model_name=endpoint) - self._role_routes[role.lower()] = endpoint - if fallbacks is not None: - self._fallback_routes[role.lower()] = fallbacks - - def get_endpoint(self, role: str) -> ModelEndpoint: - """Retrieves the configured primary model endpoint for a role.""" - return self._role_routes.get(role.lower(), self.default_endpoint) - - def get_fallbacks(self, role: str) -> list[ModelEndpoint]: - """Retrieves fallback endpoints for a role.""" - return self._fallback_routes.get(role.lower(), []) - - def instantiate_model(self, endpoint: ModelEndpoint) -> BaseChatModel: - """Instantiates a concrete LangChain chat model via the unified ModelFactory.""" - return ModelFactory.create_model(endpoint) - - def create_chat_model(self, role: str) -> BaseChatModel: - """Creates a chat model with configured fallbacks for the specified role.""" - primary_endpoint = self.get_endpoint(role) - primary_model = self.instantiate_model(primary_endpoint) - - fallbacks = self.get_fallbacks(role) - if fallbacks: - fallback_models = [self.instantiate_model(ep) for ep in fallbacks] - return primary_model.with_fallbacks(fallback_models) - - return primary_model - - -# Global singleton router instance -default_router = ModelRouter() diff --git a/artemis/services/llm.py b/artemis/services/llm.py index 59664c3..a89d87f 100644 --- a/artemis/services/llm.py +++ b/artemis/services/llm.py @@ -1025,17 +1025,7 @@ def _resolve_endpoint( is_utils: bool = False, use_fallback: bool = False, ) -> ModelEndpoint: - """Cleanly resolves a ModelEndpoint from context router or llm_config.""" - # 1. Prefer dynamic ModelRouter if attached to context - if getattr(ctx, "model_router", None) is not None: - role_name = str(name).lower() - if use_fallback: - fallbacks = ctx.model_router.get_fallbacks(role_name) - if fallbacks: - return fallbacks[0] - return ctx.model_router.get_endpoint(role_name) - - # 2. Fall back to context llm_config + """Cleanly resolves a ModelEndpoint from context llm_config.""" if getattr(ctx, "llm_config", None) is None: try: ctx.llm_config = get_default_llm_config() diff --git a/artemis/tools/mobile/exec_tools.py b/artemis/tools/mobile/exec_tools.py deleted file mode 100644 index 1c06858..0000000 --- a/artemis/tools/mobile/exec_tools.py +++ /dev/null @@ -1,683 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import asyncio -from typing import Any, Literal - -from langchain_core.messages import ToolMessage -from langchain_core.tools import BaseTool -from pydantic import BaseModel, Field - -from artemis.context import ArtemisContext -from artemis.controllers.unified_controller import UnifiedMobileController -from artemis.data_engine.trace import trace_langchain_tool -from artemis.drivers.base import BaseDeviceDriver -from artemis.graph.state import State -from artemis.tools.base import ArtemisTool, ToolCategory, ToolRegistry -from artemis.tools.tool_wrapper import ToolWrapper -from artemis.tools.types import CyFunctionDetector -from artemis.utils.coordinates import ( - compute_smart_swipe_coordinates, - parse_swipe_parameters, -) -from artemis.utils.logger import get_logger - -logger = get_logger(__name__) - - -class ClickArgs(BaseModel): - """Arguments schema for clicking on normalized coordinates.""" - - model_config = {"ignored_types": (CyFunctionDetector,)} - target: list[int] = Field( - ..., - description=("Click target normalized coordinates [x, y] in 0-1000 scale."), - ) - - -class LongPressArgs(BaseModel): - """Arguments schema for long pressing on normalized coordinates.""" - - model_config = {"ignored_types": (CyFunctionDetector,)} - target: list[int] = Field( - ..., - description=("Long press target normalized coordinates [x, y] in 0-1000 scale."), - ) - duration: int = Field(1000, description="Long press duration in milliseconds (default 1000).") - - -class InputTextArgs(BaseModel): - """Arguments schema for inputting text into normalized coordinates.""" - - model_config = {"ignored_types": (CyFunctionDetector,)} - text: str = Field( - ..., - description="The text content to input. Supports multi-line strings with '\\n'.", - ) - target: list[int] = Field( - ..., - description=("Input target field normalized coordinates [x, y] in 0-1000 scale."), - ) - clear_exist: bool = Field( - True, - description=( - "Whether to clear existing text before typing (True to replace all" - " text, False to append at the end of existing content)." - ), - ) - - -class SwipeArgs(BaseModel): - """Arguments schema for performing swipe gestures.""" - - model_config = {"ignored_types": (CyFunctionDetector,)} - direction: Literal["up", "down", "left", "right"] | None = Field( - None, - description=( - "Direction for scrolling and swiping: 'up' (drags bottom-to-top, scrolling down to reveal content below)," - " 'down' (drags top-to-bottom, scrolling up to reveal content above)," - " 'left' (drags right-to-left, scrolling right)," - " 'right' (drags left-to-right, scrolling left)." - ), - ) - start: list[int] | None = Field( - None, - description=( - "Start normalized coordinates [start_x, start_y] in 0-1000 scale for precise," - " local interactions (e.g. adjusting sliders, SeekBars, fine range selection, or drag-and-drop)." - ), - ) - end: list[int] | None = Field( - None, - description=( - "End normalized coordinates [end_x, end_y] in 0-1000 scale for precise," - " local interactions (e.g. adjusting sliders, SeekBars, fine range selection, or drag-and-drop)." - ), - ) - target: int | list[int] | str | None = Field( - None, - description=( - "Optional target element index (e.g. 2) or container bounds [left, top, right, bottom] to scope the directional swipe within." - ), - ) - action: Literal["up", "down", "left", "right"] | list[int] | None = Field( - None, - description=( - "Backward-compatible swipe gesture: smart direction string ('up', 'down', 'left', 'right')" - " OR precise custom coordinates [start_x, start_y, end_x, end_y] in 0-1000 scale." - ), - ) - duration: int = Field( - 800, - description=( - "Optional swipe/drag duration in milliseconds (default 800). For drag-and-drop," - " list reordering, or sliding/adjusting sliders (e.g., volume, brightness, SeekBars)," - " set duration >= 1000 (e.g. 1500). If omitted for directional swipe, duration is computed automatically." - ), - ) - - -class PressKeyArgs(BaseModel): - """Arguments schema for pressing system keys.""" - - model_config = {"ignored_types": (CyFunctionDetector,)} - key: Literal["ENTER", "BACK", "HOME", "APP_SWITCH"] = Field( - ..., - description=("Standard Android system button name (ENTER, BACK, HOME, APP_SWITCH)."), - ) - - -CLICK_DOCSTRING = "[ACTION] Click on the target normalized coordinates on the screen." - - -class ClickTool(ArtemisTool): - """Universal tool for clicking on target normalized coordinates on the screen.""" - - def __init__(self, category: ToolCategory = "action"): - super().__init__( - name="click", - description=CLICK_DOCSTRING, - args_schema=ClickArgs, - category=category, - ) - - # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals - async def execute( - self, - driver: BaseDeviceDriver | None = None, - ctx: ArtemisContext | None = None, - target: list[int] | None = None, - tool_call_id: str | None = None, - state: State | None = None, - **kwargs: Any, - ) -> Any: - tgt = target if target is not None else (kwargs.get("target") or kwargs.get("Target") or []) - tcid = tool_call_id if tool_call_id is not None else kwargs.get("tool_call_id") - st = state if state is not None else kwargs.get("state") - - width = 1080 - height = 2400 - if ctx and hasattr(ctx, "device") and ctx.device: - width = getattr(ctx.device, "device_width", 1080) - height = getattr(ctx.device, "device_height", 2400) - elif driver and hasattr(driver, "screen_size"): - width, height = driver.screen_size - - try: - if not tgt or len(tgt) < 2: - raise ValueError(f"Invalid target coordinates: {tgt}") - nx, ny = tgt[0], tgt[1] - x = int(max(0, min(width - 1, nx * width / 1000))) - y = int(max(0, min(height - 1, ny * height / 1000))) - - if driver is not None and hasattr(driver, "tap"): - success = await driver.tap(x, y) - outcome = ( - f"Clicked at [{nx}, {ny}] (normalized) successfully." - if success - else f"Click failed at {tgt}" - ) - elif ctx is not None: - controller = UnifiedMobileController(ctx) - result = await controller.tap_at(x, y) - success = result.error is None - outcome = ( - f"Clicked at [{nx}, {ny}] (normalized) successfully." - if success - else f"Failed to click: {result.error}" - ) - else: - success = False - outcome = "Error during click: No driver or context provided." - except Exception as e: # pylint: disable=broad-exception-caught - success = False - outcome = f"Error during click: {e}" - - if st is not None: - return ToolMessage( - tool_call_id=tcid or "", - content=outcome, - status="success" if success else "error", - ) - - return outcome - - -# Universal tool instance & aliases -click = ClickTool() -Click = ClickTool -ToolRegistry.register(click) - - -def get_click_tool(ctx: ArtemisContext) -> BaseTool: - """Exports click as a LangChain BaseTool.""" - return trace_langchain_tool(click.to_langchain_tool(ctx), ctx) - - -LONG_PRESS_DOCSTRING = "[ACTION] Long press on the target normalized coordinates on the screen." - - -class LongPressTool(ArtemisTool): - """Universal tool for long pressing on target normalized coordinates on the screen.""" - - def __init__(self, category: ToolCategory = "action"): - super().__init__( - name="long_press", - description=LONG_PRESS_DOCSTRING, - args_schema=LongPressArgs, - category=category, - ) - - # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals - async def execute( - self, - driver: BaseDeviceDriver | None = None, - ctx: ArtemisContext | None = None, - target: list[int] | None = None, - duration: int = 1000, - tool_call_id: str | None = None, - state: State | None = None, - **kwargs: Any, - ) -> Any: - tgt = target if target is not None else (kwargs.get("target") or kwargs.get("Target") or []) - dur = ( - duration - if duration is not None - else (kwargs.get("duration") or kwargs.get("Duration") or 1000) - ) - tcid = tool_call_id if tool_call_id is not None else kwargs.get("tool_call_id") - st = state if state is not None else kwargs.get("state") - - width = 1080 - height = 2400 - if ctx and hasattr(ctx, "device") and ctx.device: - width = getattr(ctx.device, "device_width", 1080) - height = getattr(ctx.device, "device_height", 2400) - elif driver and hasattr(driver, "screen_size"): - width, height = driver.screen_size - - try: - if not tgt or len(tgt) < 2: - raise ValueError(f"Invalid target coordinates: {tgt}") - nx, ny = tgt[0], tgt[1] - x = int(max(0, min(width - 1, nx * width / 1000))) - y = int(max(0, min(height - 1, ny * height / 1000))) - - if driver is not None and hasattr(driver, "long_press"): - success = await driver.long_press(x, y, duration_ms=dur) - outcome = ( - f"Long pressed at [{nx}, {ny}] (normalized) successfully." - if success - else f"Failed to long press: Long press failed at {tgt}" - ) - elif ctx is not None: - controller = UnifiedMobileController(ctx) - result = await controller.tap_at(x, y, long_press=True, long_press_duration=dur) - success = result.error is None - outcome = ( - f"Long pressed at [{nx}, {ny}] (normalized) successfully." - if success - else f"Failed to long press: {result.error}" - ) - else: - success = False - outcome = "Error during long press: No driver or context provided." - except Exception as e: # pylint: disable=broad-exception-caught - success = False - outcome = f"Error during long press: {e}" - - if st is not None: - return ToolMessage( - tool_call_id=tcid or "", - content=outcome, - status="success" if success else "error", - ) - - return outcome - - -# Universal tool instance & aliases -long_press = LongPressTool() -LongPress = LongPressTool -ToolRegistry.register(long_press) - - -def get_long_press_tool(ctx: ArtemisContext) -> BaseTool: - """Exports long_press as a LangChain BaseTool.""" - return trace_langchain_tool(long_press.to_langchain_tool(ctx), ctx) - - -INPUT_TEXT_DOCSTRING = "[ACTION] Type text into the target input field coordinates." - - -class InputTextTool(ArtemisTool): - """Universal tool for typing text into target input field coordinates.""" - - def __init__(self, category: ToolCategory = "action"): - super().__init__( - name="input_text", - description=INPUT_TEXT_DOCSTRING, - args_schema=InputTextArgs, - category=category, - ) - - # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals - async def execute( - self, - driver: BaseDeviceDriver | None = None, - ctx: ArtemisContext | None = None, - text: str | None = None, - target: list[int] | None = None, - clear_exist: bool = True, - tool_call_id: str | None = None, - state: State | None = None, - **kwargs: Any, - ) -> Any: - txt = text if text is not None else (kwargs.get("text") or kwargs.get("Text") or "") - tgt = target if target is not None else (kwargs.get("target") or kwargs.get("Target") or []) - clr = ( - clear_exist - if clear_exist is not None - else kwargs.get("clear_exist", kwargs.get("ClearExist", True)) - ) - tcid = tool_call_id if tool_call_id is not None else kwargs.get("tool_call_id") - st = state if state is not None else kwargs.get("state") - - width = 1080 - height = 2400 - if ctx and hasattr(ctx, "device") and ctx.device: - width = getattr(ctx.device, "device_width", 1080) - height = getattr(ctx.device, "device_height", 2400) - elif driver and hasattr(driver, "screen_size"): - width, height = driver.screen_size - - try: - if not tgt or len(tgt) < 2: - raise ValueError(f"Invalid target coordinates: {tgt}") - nx, ny = tgt[0], tgt[1] - x = int(max(0, min(width - 1, nx * width / 1000))) - y = int(max(0, min(height - 1, ny * height / 1000))) - - if driver is not None and hasattr(driver, "input_text"): - await driver.tap(x, y) - await asyncio.sleep(0.2) - success = await driver.input_text(txt, clear_existing=clr) - outcome = ( - f"Typed '{txt}' successfully." - if success - else f"Failed to input text into {tgt}" - ) - elif ctx is not None: - controller = UnifiedMobileController(ctx) - tap_result = await controller.tap_at(x, y) - if tap_result.error: - raise RuntimeError(f"Failed to focus input field: {tap_result.error}") - - if clr: - await controller.erase_text() - else: - await controller.press_key("123") # Move to end - - success = await controller.type_text(txt, clear_existing=False) - outcome = f"Typed '{txt}' successfully." if success else "Failed to type text." - else: - success = False - outcome = "Error during input text: No driver or context provided." - except Exception as e: # pylint: disable=broad-exception-caught - success = False - outcome = f"Error during input text: {e}" - - if st is not None: - return ToolMessage( - tool_call_id=tcid or "", - content=outcome, - status="success" if success else "error", - ) - - return outcome - - -# Universal tool instance & aliases -input_text = InputTextTool() -InputText = InputTextTool -ToolRegistry.register(input_text) - - -def get_input_text_tool(ctx: ArtemisContext) -> BaseTool: - """Exports input_text as a LangChain BaseTool.""" - return trace_langchain_tool(input_text.to_langchain_tool(ctx), ctx) - - -SWIPE_DOCSTRING = ( - "[ACTION] Perform a swipe, drag, or slider-adjustment gesture on the" - " screen.\n\n" - "• Directional Scrolling ('direction'): Recommended for general" - " browsing and standard page scrolling in most scenarios. Automatically" - " computes safe swipe vectors and adaptive duration, retains a ~40% visual overlap" - " anchor for zero-omission traversal, and prevents inertial flings. Supports" - " scoping to a sub-container via 'target'. If it fails on certain custom layouts," - " fall back to specifying exact coordinates ('start' and 'end') directly.\n" - "• Precise Coordinate Gestures ('start', 'end'): Best for local," - " fine-grained interactions such as adjusting sliders/SeekBars (e.g.," - " volume, brightness, progress bars), drag-and-drop / list reordering," - " or as a reliable fallback when directional scrolling fails on specific" - " containers. Always drag slightly PAST the target position to overcome" - " touch slop and reliably trigger the update. When setting a slider to" - " Maximum (100%) or Minimum (0%), swipe fully to the extreme boundary." -) - - -class SwipeTool(ArtemisTool): - """Universal tool for performing swipe gestures on the screen.""" - - def __init__(self, category: ToolCategory = "action"): - super().__init__( - name="swipe", - description=SWIPE_DOCSTRING, - args_schema=SwipeArgs, - category=category, - ) - - # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-branches,too-many-locals,too-many-statements - async def execute( - self, - driver: BaseDeviceDriver | None = None, - ctx: ArtemisContext | None = None, - action: Literal["up", "down", "left", "right"] | list[int] | str | None = None, - direction: Literal["up", "down", "left", "right"] | str | None = None, - start: list[int] | None = None, - end: list[int] | None = None, - duration: int = 800, - tool_call_id: str | None = None, - state: State | None = None, - **kwargs: Any, - ) -> Any: - params = dict(kwargs) - if action is not None: - params["action"] = action - if direction is not None: - params["direction"] = direction - if start is not None: - params["start"] = start - if end is not None: - params["end"] = end - if duration is not None: - params["duration"] = duration - - kind, target, dur = parse_swipe_parameters(params, default_duration=duration or 800) - - tcid = tool_call_id if tool_call_id is not None else kwargs.get("tool_call_id") - st = state if state is not None else kwargs.get("state") - - width = 1080 - height = 2400 - if ctx and hasattr(ctx, "device") and ctx.device: - width = getattr(ctx.device, "device_width", 1080) - height = getattr(ctx.device, "device_height", 2400) - elif driver and hasattr(driver, "screen_size"): - width, height = driver.screen_size - - try: - if kind is None or target is None: - raise ValueError( - f"Invalid swipe action parameter: {action or direction or (start, end) or kwargs}" - ) - - if ( - driver is not None - and hasattr(driver, "swipe_direction") - and hasattr(driver, "swipe") - ): - if kind == "direction" and isinstance(target, str): - success = await driver.swipe_direction(target, duration_ms=dur) - outcome = ( - "Swipe completed successfully." if success else f"Swipe {target} failed" - ) - elif kind == "coords" and isinstance(target, list) and len(target) == 4: - start_x = int(max(0, min(width - 1, target[0] * width / 1000))) - start_y = int(max(0, min(height - 1, target[1] * height / 1000))) - end_x = int(max(0, min(width - 1, target[2] * width / 1000))) - end_y = int(max(0, min(height - 1, target[3] * height / 1000))) - success = await driver.swipe(start_x, start_y, end_x, end_y, duration_ms=dur) - outcome = ( - "Swipe completed successfully." - if success - else f"Swipe coordinates {target} failed" - ) - else: - raise ValueError(f"Invalid swipe action parameter: {target}") - elif ctx is not None: - controller = UnifiedMobileController(ctx) - - if kind == "direction" and isinstance(target, str): - indexed_elems = getattr(state, "indexed_elements", None) if state else None - ui_hier = getattr(state, "latest_ui_hierarchy", None) if state else None - start_x, start_y, end_x, end_y, smart_dur = compute_smart_swipe_coordinates( - direction=target, - target=params.get("target"), - indexed_elements=indexed_elems, - ui_hierarchy=ui_hier, - width=width, - height=height, - duration=dur, - ) - dur = smart_dur - elif kind == "coords" and isinstance(target, list) and len(target) == 4: - nx1, ny1, nx2, ny2 = target - start_x = int(max(0, min(width - 1, nx1 * width / 1000))) - start_y = int(max(0, min(height - 1, ny1 * height / 1000))) - end_x = int(max(0, min(width - 1, nx2 * width / 1000))) - end_y = int(max(0, min(height - 1, ny2 * height / 1000))) - else: - raise ValueError(f"Invalid swipe action parameter: {target}") - - error = await controller.swipe_coords(start_x, start_y, end_x, end_y, dur) - success = error is None - outcome = "Swipe completed successfully." if success else f"Swipe failed: {error}" - else: - success = False - outcome = "Error during swipe: No driver or context provided." - except Exception as e: # pylint: disable=broad-exception-caught - success = False - outcome = f"Error during swipe: {e}" - - if st is not None: - return ToolMessage( - tool_call_id=tcid or "", - content=outcome, - status="success" if success else "error", - ) - - return outcome - - -# Universal tool instance & aliases -swipe = SwipeTool() -Swipe = SwipeTool -ToolRegistry.register(swipe) - - -def get_swipe_tool(ctx: ArtemisContext) -> BaseTool: - """Exports swipe as a LangChain BaseTool.""" - return trace_langchain_tool(swipe.to_langchain_tool(ctx), ctx) - - -PRESS_KEY_DOCSTRING = "[ACTION] Press a physical or virtual system button." - - -class PressKeyTool(ArtemisTool): - """Universal tool for pressing a physical or virtual system button.""" - - def __init__(self, category: ToolCategory = "action"): - super().__init__( - name="press_key", - description=PRESS_KEY_DOCSTRING, - args_schema=PressKeyArgs, - category=category, - ) - - # pylint: disable=too-many-arguments,too-many-positional-arguments - async def execute( - self, - driver: BaseDeviceDriver | None = None, - ctx: ArtemisContext | None = None, - key: Literal["ENTER", "BACK", "HOME", "APP_SWITCH"] | str | None = None, - tool_call_id: str | None = None, - state: State | None = None, - **kwargs: Any, - ) -> Any: - k = key if key is not None else (kwargs.get("key") or kwargs.get("Key") or "") - tcid = tool_call_id if tool_call_id is not None else kwargs.get("tool_call_id") - st = state if state is not None else kwargs.get("state") - - try: - if not k: - raise ValueError("Key parameter is required.") - - if driver is not None and hasattr(driver, "press_key"): - success = await driver.press_key(k) - outcome = ( - f"Pressed key '{k}' successfully." if success else f"Failed to press key '{k}'." - ) - elif ctx is not None: - controller = UnifiedMobileController(ctx) - success = False - if k == "ENTER": - success = await controller.press_enter() - elif k == "BACK": - success = await controller.go_back() - elif k == "HOME": - success = await controller.go_home() - elif k == "APP_SWITCH": - success = await controller.press_key("KEYCODE_APP_SWITCH") - else: - success = await controller.press_key(k) - - outcome = ( - f"Pressed key '{k}' successfully." if success else f"Failed to press key '{k}'." - ) - else: - success = False - outcome = "Error during press key: No driver or context provided." - except Exception as e: # pylint: disable=broad-exception-caught - success = False - outcome = f"Error during press key: {e}" - - if st is not None: - return ToolMessage( - tool_call_id=tcid or "", - content=outcome, - status="success" if success else "error", - ) - - return outcome - - -# Universal tool instance & aliases -press_key = PressKeyTool() -PressKey = PressKeyTool -ToolRegistry.register(press_key) - - -def get_press_key_tool(ctx: ArtemisContext) -> BaseTool: - """Exports press_key as a LangChain BaseTool.""" - return trace_langchain_tool(press_key.to_langchain_tool(ctx), ctx) - - -# Wrappers to expose -click_wrapper = ToolWrapper( - tool_fn_getter=get_click_tool, - on_success_fn=lambda *args, **kwargs: "Success", - on_failure_fn=lambda *args, **kwargs: "Failure", -) -long_press_wrapper = ToolWrapper( - tool_fn_getter=get_long_press_tool, - on_success_fn=lambda *args, **kwargs: "Success", - on_failure_fn=lambda *args, **kwargs: "Failure", -) -input_text_wrapper = ToolWrapper( - tool_fn_getter=get_input_text_tool, - on_success_fn=lambda *args, **kwargs: "Success", - on_failure_fn=lambda *args, **kwargs: "Failure", -) -swipe_wrapper = ToolWrapper( - tool_fn_getter=get_swipe_tool, - on_success_fn=lambda *args, **kwargs: "Success", - on_failure_fn=lambda *args, **kwargs: "Failure", -) -press_key_wrapper = ToolWrapper( - tool_fn_getter=get_press_key_tool, - on_success_fn=lambda *args, **kwargs: "Success", - on_failure_fn=lambda *args, **kwargs: "Failure", -) diff --git a/artemis/utils/notes.py b/artemis/utils/notes.py index 71b0e5f..69aa8e9 100644 --- a/artemis/utils/notes.py +++ b/artemis/utils/notes.py @@ -13,7 +13,6 @@ # limitations under the License. import difflib -import hashlib import json import os from pathlib import Path @@ -21,6 +20,7 @@ import re from artemis.data_engine.engine import _CURRENT_DATA_ENGINE from artemis.utils.logger import get_logger +from artemis.utils.plan_grammar import parse_plan logger = get_logger(__name__) @@ -411,23 +411,24 @@ def record_subgoal_hash_chain( if not content_before or not content_after: return - # Helper to get the active subgoal hash from task_plan string + # Helper to get the active subgoal hash from task_plan string. + # Parsing goes through the plan-grammar single source; the hash is + # ``PlanItem.key`` (= ``plan_grammar.subgoal_hash``, md5 of the stripped + # subgoal text) -- byte-identical to the historical inline md5, so anchors + # already persisted in subgoal_hash_chain.json stay valid. def get_active_hash(task_plan: str) -> str | None: try: - lines = task_plan.splitlines() - for line in reversed(lines): - if line.strip().startswith("- [/]"): - text = line.strip()[5:].strip() - return hashlib.md5(text.encode("utf-8")).hexdigest() - - # Fallback to first pending [ ] if no [/] - for line in lines: - if line.strip().startswith("- [ ]"): - text = line.strip()[5:].strip() - return hashlib.md5(text.encode("utf-8")).hexdigest() + snapshot = parse_plan(task_plan) + # Last active [/] item wins (its own key, not the parent's) -- + # matches the historical reversed-scan behavior of this helper. + item = snapshot.last_active() + if item is None: + # Fallback to the first pending [ ] item if no [/]. + item = next((i for i in snapshot.items if i.is_pending), None) + return item.key if item is not None else None except Exception as exc: - # Pure string/hash operations; a failure here is unexpected and - # silently disables subgoal rename tracking. + # A failure here is unexpected and silently disables subgoal + # rename tracking. logger.warning("Failed to compute active subgoal hash: %s", exc) return None diff --git a/artemis/utils/task_tree.py b/artemis/utils/task_tree.py index 7a8bddc..1326f6a 100644 --- a/artemis/utils/task_tree.py +++ b/artemis/utils/task_tree.py @@ -820,10 +820,8 @@ def build_plan_and_history( # 1. Plan Checklist Section output_parts = [] - # Check if task_plan has actual subgoals (e.g. starts with '- [') - has_subgoals = task_plan and any( - line.strip().startswith("- [") for line in task_plan.split("\n") - ) + # Check if task_plan has actual subgoals (single-source checkbox grammar). + has_subgoals = bool(task_plan) and bool(parse_plan(task_plan).items) if has_subgoals: output_parts.append("--- Task Plan ---") diff --git a/tests/fixtures/action_surfaces/adb_server_manifest.json b/tests/fixtures/action_surfaces/adb_server_manifest.json new file mode 100644 index 0000000..119d75e --- /dev/null +++ b/tests/fixtures/action_surfaces/adb_server_manifest.json @@ -0,0 +1,223 @@ +{ + "tap": { + "description": "Taps on the screen at coordinates.\n\n'coordinates' is a list [x, y].\n'times' is the number of consecutive clicks (default 1).\n'delay_ms' is the delay in milliseconds between consecutive clicks (default\n100).\n", + "inputSchema": { + "properties": { + "coordinates": { + "items": { + "type": "integer" + }, + "title": "Coordinates", + "type": "array" + }, + "times": { + "default": 1, + "title": "Times", + "type": "integer" + }, + "delay_ms": { + "default": 100, + "title": "Delay Ms", + "type": "integer" + } + }, + "required": [ + "coordinates" + ], + "title": "tapArguments", + "type": "object" + } + }, + "long_press_on": { + "description": "Long presses on the screen at coordinates.\n\n'coordinates' is a list [x, y].\n'duration' is in milliseconds (default 1000).\n", + "inputSchema": { + "properties": { + "coordinates": { + "items": { + "type": "integer" + }, + "title": "Coordinates", + "type": "array" + }, + "duration": { + "default": 1000, + "title": "Duration", + "type": "integer" + } + }, + "required": [ + "coordinates" + ], + "title": "long_press_onArguments", + "type": "object" + } + }, + "swipe": { + "description": "Swipes from start coordinates to end coordinates.\n\n'coordinates' is a list [start_x, start_y, end_x, end_y].\n'duration' is in milliseconds (default 400).\n\nSet duration >= 1000 to drag-and-drop. Drag slightly past the target\nposition to trigger reordering.\n", + "inputSchema": { + "properties": { + "coordinates": { + "items": { + "type": "integer" + }, + "title": "Coordinates", + "type": "array" + }, + "duration": { + "default": 400, + "title": "Duration", + "type": "integer" + } + }, + "required": [ + "coordinates" + ], + "title": "swipeArguments", + "type": "object" + } + }, + "back": { + "description": "Simulates pressing the system back button.", + "inputSchema": { + "properties": {}, + "title": "backArguments", + "type": "object" + } + }, + "launch_app": { + "description": "Launches an application by its Android package name with retries and smart polling.", + "inputSchema": { + "properties": { + "package_name": { + "title": "Package Name", + "type": "string" + } + }, + "required": [ + "package_name" + ], + "title": "launch_appArguments", + "type": "object" + } + }, + "stop_app": { + "description": "Force stops an application by its Android package name.", + "inputSchema": { + "properties": { + "package_name": { + "title": "Package Name", + "type": "string" + } + }, + "required": [ + "package_name" + ], + "title": "stop_appArguments", + "type": "object" + } + }, + "open_link": { + "description": "Opens a URL or deep link on the device.", + "inputSchema": { + "properties": { + "url": { + "title": "Url", + "type": "string" + } + }, + "required": [ + "url" + ], + "title": "open_linkArguments", + "type": "object" + } + }, + "focus_and_input_text": { + "description": "Focuses on a UI element at coordinates and inputs text.\n\n'coordinates' is a list [x, y] to tap first to gain focus.\n'clear_before_input' if True, clears all existing text before typing. If False, appends text at the end of existing content.\n'text' supports multi-line content with '\\n'.\n", + "inputSchema": { + "properties": { + "coordinates": { + "items": { + "type": "integer" + }, + "title": "Coordinates", + "type": "array" + }, + "text": { + "title": "Text", + "type": "string" + }, + "clear_before_input": { + "default": false, + "title": "Clear Before Input", + "type": "boolean" + } + }, + "required": [ + "coordinates", + "text" + ], + "title": "focus_and_input_textArguments", + "type": "object" + } + }, + "focus_and_clear_text": { + "description": "Focuses on a UI element at coordinates and clears its text content.\n\n'coordinates' is a list [x, y] to tap first to gain focus.\n", + "inputSchema": { + "properties": { + "coordinates": { + "items": { + "type": "integer" + }, + "title": "Coordinates", + "type": "array" + } + }, + "required": [ + "coordinates" + ], + "title": "focus_and_clear_textArguments", + "type": "object" + } + }, + "erase_one_char": { + "description": "Erases a single character (simulates Backspace).", + "inputSchema": { + "properties": {}, + "title": "erase_one_charArguments", + "type": "object" + } + }, + "press_key": { + "description": "Presses a specific Android key event (e.g., KEYCODE_ENTER, KEYCODE_HOME).", + "inputSchema": { + "properties": { + "keycode": { + "title": "Keycode", + "type": "string" + } + }, + "required": [ + "keycode" + ], + "title": "press_keyArguments", + "type": "object" + } + }, + "take_screenshot": { + "description": "Takes a screenshot of the device screen.\n\nReturns the screenshot as a base64 encoded JPEG string.\n", + "inputSchema": { + "properties": {}, + "title": "take_screenshotArguments", + "type": "object" + } + }, + "get_ui_hierarchy": { + "description": "Retrieves the current UI elements hierarchy from the device.", + "inputSchema": { + "properties": {}, + "title": "get_ui_hierarchyArguments", + "type": "object" + } + } +} diff --git a/tests/unit/mcp/test_action_specs.py b/tests/unit/mcp/test_action_specs.py index 9fead48..5ad23c4 100644 --- a/tests/unit/mcp/test_action_specs.py +++ b/tests/unit/mcp/test_action_specs.py @@ -210,15 +210,14 @@ def test_operator_verbs_lower_onto_manifest_actions(): def test_registry_wait_for_delay_seconds_drift_stays_dead(): - """Guards the removal of the duplicate device-action registration. + """Guards against re-growing a shadow device-action registration. ``artemis/tools/actions/device_actions.py`` used to shadow-register a second - ``wait_for_delay`` taking *seconds* while every prompt teaches milliseconds. The - registry (a direct-execution SDK surface, artemis/tools/mobile/exec_tools.py) - must never again carry a ``wait_for_delay`` whose unit disagrees with the - manifest's ``time_in_ms``. + ``wait_for_delay`` taking *seconds* while every prompt teaches milliseconds + (the last such surface, ``artemis/tools/mobile/exec_tools.py``, is deleted). + The ToolRegistry must never again carry a ``wait_for_delay`` whose unit + disagrees with the manifest's ``time_in_ms``. """ - import artemis.tools.mobile.exec_tools # noqa: F401 - registration side effect from artemis.tools.base import ToolRegistry tool = ToolRegistry.get("wait_for_delay") diff --git a/tests/unit/mcp/test_adb_server_contract.py b/tests/unit/mcp/test_adb_server_contract.py new file mode 100644 index 0000000..286748b --- /dev/null +++ b/tests/unit/mcp/test_adb_server_contract.py @@ -0,0 +1,227 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Consistency contract between the stdio ADB MCP server and the action manifest. + +``artemis/mcp/adb_server.py`` deliberately keeps its historical *pixel-space* +contract for external MCP clients (absolute pixel coordinates, combined +``coordinates`` list parameters, ``KEYCODE_*`` vocabulary), so its FastMCP tools +are hand-written rather than generated from ``artemis.mcp.action_specs``. This +suite is the compensating contract: + +* **Fixture pin** -- the full generated schema of every adb_server tool is + frozen; any signature or docstring change must regenerate the fixture in the + same commit (external clients see this schema directly). +* **Projection check** -- every device-action tool is mapped onto its canonical + ``ActionSpec`` wire dialect. Parameters that mirror a wire parameter must + agree on JSON type, requiredness, and default. Deliberate legacy divergences + are declared in ``DECLARED_DIVERGENCES`` and are *ratcheted*: if the two + surfaces converge, the stale declaration fails too, so the table always + matches reality. + +Together the two guards make silent drift between the second declaration +surface and the canonical manifest impossible in either direction. +""" + +import json +from pathlib import Path + +import pytest + +from artemis.mcp.action_specs import ACTION_SPECS, ParamSpec + +FIXTURES = Path(__file__).resolve().parents[2] / "fixtures" / "action_surfaces" + +FIXTURE_HINT = ( + "The generated schema differs from the pinned fixture. If this change is" + " intentional, regenerate the fixture in the same commit and explain the schema" + " change in the commit message; external MCP clients see this schema directly." +) + + +# --- The declared projection ---------------------------------------------------------- + +#: adb_server tool name -> (canonical action name, {adb param -> wire param}). +#: Params absent from the mapping are legacy packaging with no one-to-one wire +#: counterpart (the combined ``coordinates`` lists); their shape is asserted +#: separately below and pinned by the fixture. +ADB_TOOL_PROJECTION: dict[str, tuple[str, dict[str, str]]] = { + "tap": ("click", {"times": "times", "delay_ms": "delay_ms"}), + "long_press_on": ("long_press", {"duration": "duration_ms"}), + "swipe": ("swipe", {"duration": "duration_ms"}), + "focus_and_input_text": ("input_text", {"text": "text", "clear_before_input": "clear_exist"}), + "press_key": ("press_key", {"keycode": "key"}), + "open_link": ("open_link", {"url": "url"}), + "erase_one_char": ("erase_one_char", {}), + "focus_and_clear_text": ("focus_and_clear_text", {}), +} + +#: Combined-coordinates legacy packaging: adb tool -> (marker in docstring, arity). +#: ``coordinates`` packs the wire dialect's coordinate pair(s) into one list. +COORDINATE_PACKING: dict[str, tuple[str, int]] = { + "tap": ("[x, y]", 2), + "long_press_on": ("[x, y]", 2), + "swipe": ("[start_x, start_y, end_x, end_y]", 4), + "focus_and_input_text": ("[x, y]", 2), + "focus_and_clear_text": ("[x, y]", 2), +} + +#: Deliberate legacy divergences, (adb tool, adb param) -> reason. Each entry is +#: ratcheted: the test fails if the surfaces actually agree, so this table can +#: never go stale. +DECLARED_DIVERGENCES: dict[tuple[str, str], str] = { + ("swipe", "duration"): ( + "legacy stdio default is 400ms; the canonical wire duration_ms defaults" + " to 800ms (adaptive-duration smart swipes only exist in agent dialects)" + ), + ("focus_and_input_text", "clear_before_input"): ( + "legacy stdio appends by default (False); the canonical wire clear_exist" + " defaults to True (replace)" + ), +} + +_JSON_TYPE: dict[type, str] = {int: "integer", str: "string", bool: "boolean"} + + +def _wire_param(action: str, name: str) -> ParamSpec: + spec = ACTION_SPECS[action] + assert spec.wire is not None, f"'{action}' has no wire dialect." + for p in spec.wire.params: + if p.name == name: + return p + raise AssertionError(f"wire dialect of '{action}' has no param '{name}'") + + +async def _adb_server_tools() -> dict: + from artemis.mcp import adb_server + + tools = await adb_server.mcp.list_tools() + return {t.name: t for t in tools} + + +def _prop(tool, name: str) -> dict: + props = tool.inputSchema.get("properties", {}) + assert name in props, f"tool '{tool.name}' lost parameter '{name}'" + return props[name] + + +# --- Fixture pin ----------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_adb_server_manifest_matches_fixture(): + expected = json.loads( + (FIXTURES / "adb_server_manifest.json").read_text(encoding="utf-8") + ) + tools = await _adb_server_tools() + generated = { + name: {"description": t.description, "inputSchema": t.inputSchema} + for name, t in tools.items() + } + assert set(generated) == set(expected), FIXTURE_HINT + for name in expected: + assert generated[name] == expected[name], f"adb_server tool '{name}': {FIXTURE_HINT}" + + +# --- Projection consistency ------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_projection_covers_all_wire_backed_adb_tools(): + """Every adb_server tool that mirrors a canonical action is in the projection.""" + tools = await _adb_server_tools() + missing = set(ADB_TOOL_PROJECTION) - set(tools) + assert not missing, f"projection references adb_server tools that no longer exist: {missing}" + for adb_name, (canonical, _) in ADB_TOOL_PROJECTION.items(): + assert canonical in ACTION_SPECS, ( + f"adb_server tool '{adb_name}' maps to '{canonical}', which the canonical" + " manifest does not know." + ) + assert ACTION_SPECS[canonical].wire is not None, ( + f"adb_server tool '{adb_name}' maps to '{canonical}', which has no wire dialect." + ) + + +@pytest.mark.asyncio +async def test_mirrored_params_agree_with_wire_dialect(): + """Type/requiredness/default of every mirrored parameter matches the wire spec, + except where a divergence is declared -- and declared divergences must be real.""" + tools = await _adb_server_tools() + for adb_name, (canonical, param_map) in ADB_TOOL_PROJECTION.items(): + tool = tools[adb_name] + required = set(tool.inputSchema.get("required", [])) + for adb_param, wire_name in param_map.items(): + wire = _wire_param(canonical, wire_name) + prop = _prop(tool, adb_param) + expected_type = _JSON_TYPE.get(wire.annotation) + if expected_type is not None: + assert prop.get("type") == expected_type, ( + f"'{adb_name}.{adb_param}': schema type {prop.get('type')!r} !=" + f" canonical wire '{canonical}.{wire_name}' type {expected_type!r}" + ) + adb_required = adb_param in required + adb_default = prop.get("default") + diverged_key = (adb_name, adb_param) + agrees = adb_required == wire.required and ( + wire.required or adb_default == wire.default + ) + if diverged_key in DECLARED_DIVERGENCES: + assert not agrees, ( + f"'{adb_name}.{adb_param}' is declared divergent from" + f" '{canonical}.{wire_name}' but the surfaces now agree -- remove" + f" the stale entry from DECLARED_DIVERGENCES:" + f" {DECLARED_DIVERGENCES[diverged_key]}" + ) + else: + assert agrees, ( + f"'{adb_name}.{adb_param}' drifted from canonical wire" + f" '{canonical}.{wire_name}': required {adb_required} vs" + f" {wire.required}, default {adb_default!r} vs {wire.default!r}." + " Either re-align, or declare the divergence in" + " DECLARED_DIVERGENCES with a reason." + ) + + +@pytest.mark.asyncio +async def test_coordinate_packing_shape_is_documented(): + """The combined `coordinates` legacy parameter keeps its documented arity.""" + tools = await _adb_server_tools() + for adb_name, (marker, _arity) in COORDINATE_PACKING.items(): + tool = tools[adb_name] + prop = _prop(tool, "coordinates") + assert prop.get("type") == "array", f"'{adb_name}.coordinates' must be an array" + assert prop.get("items", {}).get("type") == "integer", ( + f"'{adb_name}.coordinates' must be a list of integers" + ) + assert "coordinates" in set(tool.inputSchema.get("required", [])), ( + f"'{adb_name}.coordinates' must stay required" + ) + assert marker in (tool.description or ""), ( + f"'{adb_name}' docstring no longer documents the coordinate shape" + f" '{marker}'; external clients rely on this wording." + ) + + +@pytest.mark.asyncio +async def test_press_key_vocabulary_divergence_is_documented(): + """adb_server's press_key takes raw KEYCODE_* strings; the canonical wire takes + bare lowercase words. The canonical spec documents that divergence in prose.""" + tools = await _adb_server_tools() + desc = tools["press_key"].description or "" + assert "KEYCODE_" in desc, ( + "adb_server press_key docstring must keep teaching KEYCODE_* vocabulary" + ) + assert "key" in ACTION_SPECS["press_key"].differences, ( + "the canonical manifest no longer documents the press_key vocabulary split" + ) diff --git a/tests/unit/test_model_router.py b/tests/unit/test_llm_grounding.py similarity index 75% rename from tests/unit/test_model_router.py rename to tests/unit/test_llm_grounding.py index b1097b8..c99b01d 100644 --- a/tests/unit/test_model_router.py +++ b/tests/unit/test_llm_grounding.py @@ -12,80 +12,33 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Unit tests for Dynamic Model Router.""" +"""Unit tests for ModelFactory instantiation and grounding tool wiring.""" -from artemis.llm.router import ModelEndpoint, ModelProvider, ModelRouter +from artemis.llm.router import ModelEndpoint, ModelFactory, ModelProvider -def test_model_router_default_routes(): - """Verify default role routing returns expected models.""" - router = ModelRouter() - - # Planner route - planner_ep = router.get_endpoint("planner") - assert planner_ep.provider == ModelProvider.GEMINI - assert "pro" in planner_ep.model_name - - # Operator route - operator_ep = router.get_endpoint("operator") - assert operator_ep.provider == ModelProvider.GEMINI - assert "flash" in operator_ep.model_name - - -def test_model_router_custom_route_override(): - """Verify dynamic role route override.""" - router = ModelRouter() - custom_ep = ModelEndpoint( - provider=ModelProvider.OPENAI, - model_name="gpt-4o", - api_base="https://custom.api.endpoint/v1", - ) - router.set_role_route("operator", custom_ep) - - ep = router.get_endpoint("operator") - assert ep.provider == ModelProvider.OPENAI - assert ep.model_name == "gpt-4o" - assert ep.api_base == "https://custom.api.endpoint/v1" - - -def test_model_router_fallbacks(): - """Verify fallback endpoint configuration.""" - router = ModelRouter() - fallbacks = [ - ModelEndpoint(provider=ModelProvider.GEMINI, model_name="gemini-2.5-flash"), - ModelEndpoint(provider=ModelProvider.OPENAI, model_name="gpt-4o-mini"), - ] - router.set_role_route("custom_worker", "gemini-2.5-pro", fallbacks=fallbacks) - fb = router.get_fallbacks("custom_worker") - assert len(fb) == 2 - assert fb[0].model_name == "gemini-2.5-flash" - assert fb[1].model_name == "gpt-4o-mini" - - -def test_model_router_anthropic_instantiation(): +def test_model_factory_anthropic_instantiation(): """Verify Anthropic Claude model instantiation with thinking budget.""" - router = ModelRouter() anth_ep = ModelEndpoint( provider=ModelProvider.ANTHROPIC, model_name="claude-3-7-sonnet-20250219", api_key="sk-ant-test-key", reasoning_effort="high", ) - model = router.instantiate_model(anth_ep) + model = ModelFactory.create_model(anth_ep) assert model.model == "claude-3-7-sonnet-20250219" assert getattr(model, "thinking", None) == {"type": "enabled", "budget_tokens": 32768} -def test_model_router_openai_instantiation(): +def test_model_factory_openai_instantiation(): """Verify OpenAI model instantiation with reasoning effort.""" - router = ModelRouter() oai_ep = ModelEndpoint( provider=ModelProvider.OPENAI, model_name="o3-mini", api_key="sk-test-key", reasoning_effort="medium", ) - model = router.instantiate_model(oai_ep) + model = ModelFactory.create_model(oai_ep) assert model.model_name == "o3-mini" assert getattr(model, "reasoning_effort", None) == "medium" diff --git a/tests/unit/test_tools.py b/tests/unit/test_tools.py deleted file mode 100644 index d6c08b5..0000000 --- a/tests/unit/test_tools.py +++ /dev/null @@ -1,70 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""Unit tests for Universal Tool Protocol and Actions.""" - -import pytest -from artemis.drivers.mock.mock_driver import MockDeviceDriver -from artemis.tools.base import ToolRegistry -import artemis.tools.mobile.exec_tools # noqa: F401 - Register action tools - - -@pytest.mark.asyncio -async def test_tool_registry_registration(): - """Verify standard device actions are registered in ToolRegistry.""" - click_tool = ToolRegistry.get("click") - assert click_tool is not None - assert click_tool.name == "click" - assert click_tool.category == "action" - - swipe_tool = ToolRegistry.get("swipe") - assert swipe_tool is not None - assert swipe_tool.name == "swipe" - - -@pytest.mark.asyncio -async def test_tool_execution_with_driver(): - """Verify tool execution correctly dispatches to MockDeviceDriver.""" - driver = MockDeviceDriver(width=1000, height=2000) - - # Click normalized [500, 500] -> pixel (500, 1000) - result = await ToolRegistry.execute("click", {"target": [500, 500]}, driver=driver, ctx=None) - assert "Click" in result or "Clicked" in result - assert driver.action_history[-1]["action"] == "tap" - assert driver.action_history[-1]["x"] == 500 - assert driver.action_history[-1]["y"] == 1000 - - # Swipe cardinal "up" - result = await ToolRegistry.execute("swipe", {"action": "up"}, driver=driver, ctx=None) - assert "Swipe" in result or "Swiped" in result - assert driver.action_history[-1]["action"] == "swipe_direction" - - -def test_tool_to_genai_declaration(): - """Verify tools export to Google GenAI FunctionDeclaration properly.""" - declarations = ToolRegistry.get_genai_declarations(["click", "swipe", "input_text"]) - assert len(declarations) == 3 - - decl_names = [d.name for d in declarations] - assert "click" in decl_names - assert "swipe" in decl_names - assert "input_text" in decl_names - - -def test_tool_to_langchain_tool(): - """Verify tools export to LangChain BaseTool.""" - lc_tools = ToolRegistry.get_langchain_tools(ctx=None, names=["click", "press_key"]) - assert len(lc_tools) == 2 - assert lc_tools[0].name == "click" - assert lc_tools[1].name == "press_key" diff --git a/tests/unit/tools/test_click_tool.py b/tests/unit/tools/test_click_tool.py deleted file mode 100644 index 396fa69..0000000 --- a/tests/unit/tools/test_click_tool.py +++ /dev/null @@ -1,148 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from unittest.mock import AsyncMock, MagicMock, patch - -from artemis.context import ArtemisContext -from artemis.drivers.base import BaseDeviceDriver -from artemis.graph.state import State -from artemis.tools.base import ArtemisTool, ToolRegistry -from artemis.tools.mobile.exec_tools import ( - Click, - ClickArgs, - ClickTool, - click, - click_wrapper, - get_click_tool, -) -from langchain_core.messages import ToolMessage -import pytest - - -@pytest.fixture -def mock_ctx(): - ctx = MagicMock(spec=ArtemisContext) - ctx.device = MagicMock() - ctx.device.device_width = 1080 - ctx.device.device_height = 2400 - return ctx - - -@pytest.fixture -def mock_driver(): - driver = MagicMock(spec=BaseDeviceDriver) - driver.screen_size = (1080, 2400) - driver.tap = AsyncMock(return_value=True) - return driver - - -@pytest.fixture -def mock_state(): - return MagicMock(spec=State) - - -def test_click_tool_subclass_and_registry(): - """Verify ClickTool is a subclass of ArtemisTool and properly registered.""" - assert issubclass(ClickTool, ArtemisTool) - assert issubclass(Click, ArtemisTool) - assert isinstance(click, ArtemisTool) - assert isinstance(click, ClickTool) - - assert click.name == "click" - assert click.category == "action" - assert click.args_schema == ClickArgs - - # Registry lookup - reg_tool = ToolRegistry.get("click") - assert reg_tool is not None - assert isinstance(reg_tool, ClickTool) - - # GenAI FunctionDeclaration export - declaration = click.to_genai_declaration() - assert declaration.name == "click" - assert "target" in declaration.parameters.properties - - # Wrapper check - assert click_wrapper is not None - assert click_wrapper.tool_fn_getter == get_click_tool - - -@pytest.mark.asyncio -async def test_click_direct_execution_with_driver(mock_driver): - """Verify direct execution of ClickTool with BaseDeviceDriver.""" - result = await click.execute(driver=mock_driver, target=[500, 500]) - mock_driver.tap.assert_called_once_with(540, 1200) - assert "Clicked at [500, 500] (normalized) successfully." in result - - -@pytest.mark.asyncio -async def test_click_direct_execution_with_ctx(mock_ctx): - """Verify direct execution of ClickTool with ArtemisContext.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_cls.return_value = mock_controller_inst - - result = await click.execute(ctx=mock_ctx, target=[250, 750]) - mock_controller_inst.tap_at.assert_called_once_with(270, 1800) - assert "Clicked at [250, 750] (normalized) successfully." in result - - -@pytest.mark.asyncio -async def test_click_with_state_command(mock_ctx, mock_state): - """Verify ClickTool returns ToolMessage when state is provided.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_cls.return_value = mock_controller_inst - - cmd = await click.execute( - ctx=mock_ctx, - target=[100, 200], - tool_call_id="call_click_1", - state=mock_state, - ) - assert isinstance(cmd, ToolMessage) - assert cmd.tool_call_id == "call_click_1" - assert cmd.status == "success" - assert "Clicked at [100, 200]" in cmd.content - - -@pytest.mark.asyncio -async def test_click_callable_execution(mock_driver): - """Verify invoking click directly as a callable.""" - result = await click(driver=mock_driver, target=[100, 100]) - assert "Clicked at [100, 100] (normalized) successfully." in result - - -@pytest.mark.asyncio -async def test_click_invalid_target(mock_ctx): - """Verify error handling on invalid target coordinates.""" - result = await click.execute(ctx=mock_ctx, target=[]) - assert "Error during click: Invalid target coordinates" in result - - -@pytest.mark.asyncio -async def test_get_click_tool_langchain_ainvoke(mock_ctx): - """Verify get_click_tool exports a LangChain tool that works with ainvoke.""" - click_tool = get_click_tool(mock_ctx) - assert click_tool.name == "click" - - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_cls.return_value = mock_controller_inst - - result = await click_tool.ainvoke({"target": [500, 500]}) - assert "Clicked at [500, 500] (normalized) successfully." in result diff --git a/tests/unit/tools/test_input_text_tool.py b/tests/unit/tools/test_input_text_tool.py deleted file mode 100644 index be7e648..0000000 --- a/tests/unit/tools/test_input_text_tool.py +++ /dev/null @@ -1,172 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from unittest.mock import AsyncMock, MagicMock, patch - -from artemis.context import ArtemisContext -from artemis.drivers.base import BaseDeviceDriver -from artemis.graph.state import State -from artemis.tools.base import ArtemisTool, ToolRegistry -from artemis.tools.mobile.exec_tools import ( - InputText, - InputTextArgs, - InputTextTool, - get_input_text_tool, - input_text, - input_text_wrapper, -) -from langchain_core.messages import ToolMessage -import pytest - - -@pytest.fixture -def mock_ctx(): - ctx = MagicMock(spec=ArtemisContext) - ctx.device = MagicMock() - ctx.device.device_width = 1080 - ctx.device.device_height = 2400 - return ctx - - -@pytest.fixture -def mock_driver(): - driver = MagicMock(spec=BaseDeviceDriver) - driver.screen_size = (1080, 2400) - driver.tap = AsyncMock(return_value=True) - driver.input_text = AsyncMock(return_value=True) - return driver - - -@pytest.fixture -def mock_state(): - return MagicMock(spec=State) - - -def test_input_text_tool_subclass_and_registry(): - """Verify InputTextTool is a subclass of ArtemisTool and properly registered.""" - assert issubclass(InputTextTool, ArtemisTool) - assert issubclass(InputText, ArtemisTool) - assert isinstance(input_text, ArtemisTool) - assert isinstance(input_text, InputTextTool) - - assert input_text.name == "input_text" - assert input_text.category == "action" - assert input_text.args_schema == InputTextArgs - - # Registry lookup - reg_tool = ToolRegistry.get("input_text") - assert reg_tool is not None - assert isinstance(reg_tool, InputTextTool) - - # GenAI FunctionDeclaration export - declaration = input_text.to_genai_declaration() - assert declaration.name == "input_text" - assert "text" in declaration.parameters.properties - assert "target" in declaration.parameters.properties - assert "clear_exist" in declaration.parameters.properties - - # Wrapper check - assert input_text_wrapper is not None - assert input_text_wrapper.tool_fn_getter == get_input_text_tool - - -@pytest.mark.asyncio -async def test_input_text_direct_execution_with_driver(mock_driver): - """Verify direct execution of InputTextTool with BaseDeviceDriver.""" - result = await input_text.execute( - driver=mock_driver, text="hello world", target=[500, 500], clear_exist=True - ) - mock_driver.tap.assert_called_once_with(540, 1200) - mock_driver.input_text.assert_called_once_with("hello world", clear_existing=True) - assert "Typed 'hello world' successfully." in result - - -@pytest.mark.asyncio -async def test_input_text_direct_execution_with_ctx(mock_ctx): - """Verify direct execution of InputTextTool with ArtemisContext.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_inst.erase_text = AsyncMock(return_value=True) - mock_controller_inst.type_text = AsyncMock(return_value=True) - mock_controller_cls.return_value = mock_controller_inst - - result = await input_text.execute( - ctx=mock_ctx, text="test input", target=[250, 750], clear_exist=True - ) - mock_controller_inst.tap_at.assert_called_once_with(270, 1800) - mock_controller_inst.erase_text.assert_called_once() - mock_controller_inst.type_text.assert_called_once_with("test input", clear_existing=False) - assert "Typed 'test input' successfully." in result - - -@pytest.mark.asyncio -async def test_input_text_with_state_command(mock_ctx, mock_state): - """Verify InputTextTool returns ToolMessage when state is provided.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_inst.erase_text = AsyncMock(return_value=True) - mock_controller_inst.type_text = AsyncMock(return_value=True) - mock_controller_cls.return_value = mock_controller_inst - - cmd = await input_text.execute( - ctx=mock_ctx, - text="query", - target=[100, 200], - tool_call_id="call_it_1", - state=mock_state, - ) - assert isinstance(cmd, ToolMessage) - assert cmd.tool_call_id == "call_it_1" - assert cmd.status == "success" - assert "Typed 'query'" in cmd.content - - -@pytest.mark.asyncio -async def test_input_text_callable_execution(mock_driver): - """Verify invoking input_text directly as a callable.""" - result = await input_text(driver=mock_driver, text="call text", target=[100, 100]) - assert "Typed 'call text' successfully." in result - - -@pytest.mark.asyncio -async def test_input_text_invalid_target(mock_ctx): - """Verify error handling on invalid target coordinates.""" - result = await input_text.execute(ctx=mock_ctx, text="bad", target=[]) - assert "Error during input text: Invalid target coordinates" in result - - -@pytest.mark.asyncio -async def test_get_input_text_tool_langchain_ainvoke(mock_ctx): - """Verify get_input_text_tool exports a LangChain tool that works with ainvoke.""" - it_tool = get_input_text_tool(mock_ctx) - assert it_tool.name == "input_text" - - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_inst.press_key = AsyncMock(return_value=True) - mock_controller_inst.type_text = AsyncMock(return_value=True) - mock_controller_cls.return_value = mock_controller_inst - - result = await it_tool.ainvoke( - { - "text": "abc", - "target": [500, 500], - "clear_exist": False, - } - ) - mock_controller_inst.press_key.assert_called_once_with("123") - assert "Typed 'abc' successfully." in result diff --git a/tests/unit/tools/test_long_press_tool.py b/tests/unit/tools/test_long_press_tool.py deleted file mode 100644 index fa0cde9..0000000 --- a/tests/unit/tools/test_long_press_tool.py +++ /dev/null @@ -1,152 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from unittest.mock import AsyncMock, MagicMock, patch - -from artemis.context import ArtemisContext -from artemis.drivers.base import BaseDeviceDriver -from artemis.graph.state import State -from artemis.tools.base import ArtemisTool, ToolRegistry -from artemis.tools.mobile.exec_tools import ( - LongPress, - LongPressArgs, - LongPressTool, - get_long_press_tool, - long_press, - long_press_wrapper, -) -from langchain_core.messages import ToolMessage -import pytest - - -@pytest.fixture -def mock_ctx(): - ctx = MagicMock(spec=ArtemisContext) - ctx.device = MagicMock() - ctx.device.device_width = 1080 - ctx.device.device_height = 2400 - return ctx - - -@pytest.fixture -def mock_driver(): - driver = MagicMock(spec=BaseDeviceDriver) - driver.screen_size = (1080, 2400) - driver.long_press = AsyncMock(return_value=True) - return driver - - -@pytest.fixture -def mock_state(): - return MagicMock(spec=State) - - -def test_long_press_tool_subclass_and_registry(): - """Verify LongPressTool is a subclass of ArtemisTool and properly registered.""" - assert issubclass(LongPressTool, ArtemisTool) - assert issubclass(LongPress, ArtemisTool) - assert isinstance(long_press, ArtemisTool) - assert isinstance(long_press, LongPressTool) - - assert long_press.name == "long_press" - assert long_press.category == "action" - assert long_press.args_schema == LongPressArgs - - # Registry lookup - reg_tool = ToolRegistry.get("long_press") - assert reg_tool is not None - assert isinstance(reg_tool, LongPressTool) - - # GenAI FunctionDeclaration export - declaration = long_press.to_genai_declaration() - assert declaration.name == "long_press" - assert "target" in declaration.parameters.properties - assert "duration" in declaration.parameters.properties - - # Wrapper check - assert long_press_wrapper is not None - assert long_press_wrapper.tool_fn_getter == get_long_press_tool - - -@pytest.mark.asyncio -async def test_long_press_direct_execution_with_driver(mock_driver): - """Verify direct execution of LongPressTool with BaseDeviceDriver.""" - result = await long_press.execute(driver=mock_driver, target=[500, 500], duration=1500) - mock_driver.long_press.assert_called_once_with(540, 1200, duration_ms=1500) - assert "Long pressed at [500, 500] (normalized) successfully." in result - - -@pytest.mark.asyncio -async def test_long_press_direct_execution_with_ctx(mock_ctx): - """Verify direct execution of LongPressTool with ArtemisContext.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_cls.return_value = mock_controller_inst - - result = await long_press.execute(ctx=mock_ctx, target=[250, 750], duration=2000) - mock_controller_inst.tap_at.assert_called_once_with( - 270, 1800, long_press=True, long_press_duration=2000 - ) - assert "Long pressed at [250, 750] (normalized) successfully." in result - - -@pytest.mark.asyncio -async def test_long_press_with_state_command(mock_ctx, mock_state): - """Verify LongPressTool returns ToolMessage when state is provided.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_cls.return_value = mock_controller_inst - - cmd = await long_press.execute( - ctx=mock_ctx, - target=[100, 200], - duration=1200, - tool_call_id="call_lp_1", - state=mock_state, - ) - assert isinstance(cmd, ToolMessage) - assert cmd.tool_call_id == "call_lp_1" - assert cmd.status == "success" - assert "Long pressed at [100, 200]" in cmd.content - - -@pytest.mark.asyncio -async def test_long_press_callable_execution(mock_driver): - """Verify invoking long_press directly as a callable.""" - result = await long_press(driver=mock_driver, target=[100, 100]) - assert "Long pressed at [100, 100] (normalized) successfully." in result - - -@pytest.mark.asyncio -async def test_long_press_invalid_target(mock_ctx): - """Verify error handling on invalid target coordinates.""" - result = await long_press.execute(ctx=mock_ctx, target=[]) - assert "Error during long press: Invalid target coordinates" in result - - -@pytest.mark.asyncio -async def test_get_long_press_tool_langchain_ainvoke(mock_ctx): - """Verify get_long_press_tool exports a LangChain tool that works with ainvoke.""" - lp_tool = get_long_press_tool(mock_ctx) - assert lp_tool.name == "long_press" - - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_cls.return_value = mock_controller_inst - - result = await lp_tool.ainvoke({"target": [500, 500], "duration": 1500}) - assert "Long pressed at [500, 500] (normalized) successfully." in result diff --git a/tests/unit/tools/test_press_key_tool.py b/tests/unit/tools/test_press_key_tool.py deleted file mode 100644 index 78b3fe0..0000000 --- a/tests/unit/tools/test_press_key_tool.py +++ /dev/null @@ -1,164 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from unittest.mock import AsyncMock, MagicMock, patch - -from artemis.context import ArtemisContext -from artemis.drivers.base import BaseDeviceDriver -from artemis.graph.state import State -from artemis.tools.base import ArtemisTool, ToolRegistry -from artemis.tools.mobile.exec_tools import ( - PressKey, - PressKeyArgs, - PressKeyTool, - get_press_key_tool, - press_key, - press_key_wrapper, -) -from langchain_core.messages import ToolMessage -import pytest - - -@pytest.fixture -def mock_ctx(): - ctx = MagicMock(spec=ArtemisContext) - ctx.device = MagicMock() - return ctx - - -@pytest.fixture -def mock_driver(): - driver = MagicMock(spec=BaseDeviceDriver) - driver.press_key = AsyncMock(return_value=True) - return driver - - -@pytest.fixture -def mock_state(): - return MagicMock(spec=State) - - -def test_press_key_tool_subclass_and_registry(): - """Verify PressKeyTool is a subclass of ArtemisTool and properly registered.""" - assert issubclass(PressKeyTool, ArtemisTool) - assert issubclass(PressKey, ArtemisTool) - assert isinstance(press_key, ArtemisTool) - assert isinstance(press_key, PressKeyTool) - - assert press_key.name == "press_key" - assert press_key.category == "action" - assert press_key.args_schema == PressKeyArgs - - # Registry lookup - reg_tool = ToolRegistry.get("press_key") - assert reg_tool is not None - assert isinstance(reg_tool, PressKeyTool) - - # GenAI FunctionDeclaration export - declaration = press_key.to_genai_declaration() - assert declaration.name == "press_key" - assert "key" in declaration.parameters.properties - - # Wrapper check - assert press_key_wrapper is not None - assert press_key_wrapper.tool_fn_getter == get_press_key_tool - - -@pytest.mark.asyncio -async def test_press_key_direct_execution_with_driver(mock_driver): - """Verify direct execution with BaseDeviceDriver.""" - result = await press_key.execute(driver=mock_driver, key="BACK") - mock_driver.press_key.assert_called_once_with("BACK") - assert "Pressed key 'BACK' successfully." in result - - -@pytest.mark.asyncio -async def test_press_key_direct_execution_with_ctx(mock_ctx): - """Verify direct execution with ArtemisContext for various keys.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.press_enter = AsyncMock(return_value=True) - mock_controller_inst.go_back = AsyncMock(return_value=True) - mock_controller_inst.go_home = AsyncMock(return_value=True) - mock_controller_inst.press_key = AsyncMock(return_value=True) - mock_controller_cls.return_value = mock_controller_inst - - # Test ENTER - res_enter = await press_key.execute(ctx=mock_ctx, key="ENTER") - mock_controller_inst.press_enter.assert_called_once() - assert "Pressed key 'ENTER' successfully." in res_enter - - # Test BACK - res_back = await press_key.execute(ctx=mock_ctx, key="BACK") - mock_controller_inst.go_back.assert_called_once() - assert "Pressed key 'BACK' successfully." in res_back - - # Test HOME - res_home = await press_key.execute(ctx=mock_ctx, key="HOME") - mock_controller_inst.go_home.assert_called_once() - assert "Pressed key 'HOME' successfully." in res_home - - # Test APP_SWITCH - res_app_switch = await press_key.execute(ctx=mock_ctx, key="APP_SWITCH") - mock_controller_inst.press_key.assert_called_with("KEYCODE_APP_SWITCH") - assert "Pressed key 'APP_SWITCH' successfully." in res_app_switch - - -@pytest.mark.asyncio -async def test_press_key_with_state_command(mock_ctx, mock_state): - """Verify PressKeyTool returns ToolMessage when state is provided.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.go_home = AsyncMock(return_value=True) - mock_controller_cls.return_value = mock_controller_inst - - cmd = await press_key.execute( - ctx=mock_ctx, - key="HOME", - tool_call_id="call_pk_1", - state=mock_state, - ) - assert isinstance(cmd, ToolMessage) - assert cmd.tool_call_id == "call_pk_1" - assert cmd.status == "success" - assert "Pressed key 'HOME' successfully." in cmd.content - - -@pytest.mark.asyncio -async def test_press_key_callable_execution(mock_driver): - """Verify invoking press_key directly as a callable.""" - result = await press_key(driver=mock_driver, key="ENTER") - assert "Pressed key 'ENTER' successfully." in result - - -@pytest.mark.asyncio -async def test_press_key_missing_key(mock_ctx): - """Verify error handling when key is empty or missing.""" - result = await press_key.execute(ctx=mock_ctx, key="") - assert "Error during press key: Key parameter is required." in result - - -@pytest.mark.asyncio -async def test_get_press_key_tool_langchain_ainvoke(mock_ctx): - """Verify get_press_key_tool exports a LangChain tool that works with ainvoke.""" - pk_tool = get_press_key_tool(mock_ctx) - assert pk_tool.name == "press_key" - - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.go_back = AsyncMock(return_value=True) - mock_controller_cls.return_value = mock_controller_inst - - result = await pk_tool.ainvoke({"key": "BACK"}) - assert "Pressed key 'BACK' successfully." in result diff --git a/tests/unit/tools/test_swipe_tool.py b/tests/unit/tools/test_swipe_tool.py deleted file mode 100644 index d1208f5..0000000 --- a/tests/unit/tools/test_swipe_tool.py +++ /dev/null @@ -1,316 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from unittest.mock import AsyncMock, MagicMock, patch - -from artemis.context import ArtemisContext -from artemis.drivers.base import BaseDeviceDriver -from artemis.graph.state import State -from artemis.tools.base import ArtemisTool, ToolRegistry -from artemis.tools.mobile.exec_tools import ( - Swipe, - SwipeArgs, - SwipeTool, - get_swipe_tool, - swipe, - swipe_wrapper, -) -from langchain_core.messages import ToolMessage -import pytest - - -@pytest.fixture -def mock_ctx(): - ctx = MagicMock(spec=ArtemisContext) - ctx.device = MagicMock() - ctx.device.device_width = 1080 - ctx.device.device_height = 2400 - return ctx - - -@pytest.fixture -def mock_driver(): - driver = MagicMock(spec=BaseDeviceDriver) - driver.screen_size = (1080, 2400) - driver.swipe_direction = AsyncMock(return_value=True) - driver.swipe = AsyncMock(return_value=True) - return driver - - -@pytest.fixture -def mock_state(): - return MagicMock(spec=State) - - -def test_swipe_tool_subclass_and_registry(): - """Verify SwipeTool is a subclass of ArtemisTool and properly registered.""" - assert issubclass(SwipeTool, ArtemisTool) - assert issubclass(Swipe, ArtemisTool) - assert isinstance(swipe, ArtemisTool) - assert isinstance(swipe, SwipeTool) - - assert swipe.name == "swipe" - assert swipe.category == "action" - assert swipe.args_schema == SwipeArgs - - # Registry lookup - reg_tool = ToolRegistry.get("swipe") - assert reg_tool is not None - assert isinstance(reg_tool, ArtemisTool) - - # GenAI FunctionDeclaration export - declaration = swipe.to_genai_declaration() - assert declaration.name == "swipe" - assert "action" in declaration.parameters.properties - assert "duration" in declaration.parameters.properties - - # Wrapper check - assert swipe_wrapper is not None - assert swipe_wrapper.tool_fn_getter == get_swipe_tool - - -@pytest.mark.asyncio -async def test_swipe_direct_execution_with_driver_direction(mock_driver): - """Verify direct execution with direction on BaseDeviceDriver.""" - result = await swipe.execute(driver=mock_driver, action="up", duration=500) - mock_driver.swipe_direction.assert_called_once_with("up", duration_ms=500) - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_direct_execution_with_driver_coords(mock_driver): - """Verify direct execution with coordinates on BaseDeviceDriver.""" - result = await swipe.execute(driver=mock_driver, action=[100, 200, 300, 400], duration=600) - mock_driver.swipe.assert_called_once_with(108, 480, 324, 960, duration_ms=600) - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_direct_execution_with_ctx_direction(mock_ctx): - """Verify direct execution with direction on ArtemisContext.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.swipe_coords = AsyncMock(return_value=None) - mock_controller_cls.return_value = mock_controller_inst - - result = await swipe.execute(ctx=mock_ctx, action="down", duration=400) - mock_controller_inst.swipe_coords.assert_called_once_with(648, 720, 648, 1680, 400) - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_direct_execution_with_ctx_coords(mock_ctx): - """Verify direct execution with coordinates on ArtemisContext.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.swipe_coords = AsyncMock(return_value=None) - mock_controller_cls.return_value = mock_controller_inst - - result = await swipe.execute(ctx=mock_ctx, action=[200, 300, 800, 900], duration=800) - mock_controller_inst.swipe_coords.assert_called_once_with(216, 720, 864, 2160, 800) - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_with_state_command(mock_ctx, mock_state): - """Verify SwipeTool returns ToolMessage when state is provided.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.swipe_coords = AsyncMock(return_value=None) - mock_controller_cls.return_value = mock_controller_inst - - cmd = await swipe.execute( - ctx=mock_ctx, - action="left", - tool_call_id="call_sw_1", - state=mock_state, - ) - assert isinstance(cmd, ToolMessage) - assert cmd.tool_call_id == "call_sw_1" - assert cmd.status == "success" - assert "Swipe completed successfully." in cmd.content - - -@pytest.mark.asyncio -async def test_swipe_callable_execution(mock_driver): - """Verify invoking swipe directly as a callable.""" - result = await swipe(driver=mock_driver, action="right") - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_invalid_action(mock_ctx): - """Verify error handling on invalid action.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController"): - result = await swipe.execute(ctx=mock_ctx, action="diagonal") - assert "Error during swipe: Invalid" in result - - -@pytest.mark.asyncio -async def test_get_swipe_tool_langchain_ainvoke(mock_ctx): - """Verify get_swipe_tool exports a LangChain tool that works with ainvoke.""" - sw_tool = get_swipe_tool(mock_ctx) - assert sw_tool.name == "swipe" - - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.swipe_coords = AsyncMock(return_value=None) - mock_controller_cls.return_value = mock_controller_inst - - result = await sw_tool.ainvoke({"action": "up"}) - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_with_direction_and_start_end(mock_driver): - """Verify swipe works with explicit direction and start/end parameters.""" - # Direction parameter - result_dir = await swipe.execute(driver=mock_driver, direction="left", duration=500) - mock_driver.swipe_direction.assert_called_with("left", duration_ms=500) - assert "Swipe completed successfully." in result_dir - - # Start and End parameters - result_drag = await swipe.execute( - driver=mock_driver, start=[100, 200], end=[300, 400], duration=1500 - ) - mock_driver.swipe.assert_called_with(108, 480, 324, 960, duration_ms=1500) - assert "Swipe completed successfully." in result_drag - - -@pytest.mark.asyncio -async def test_swipe_with_coordinates_alias(mock_driver): - """Verify swipe works with coordinates alias parameter.""" - result = await swipe.execute(driver=mock_driver, coordinates=[200, 300, 400, 500], duration=800) - mock_driver.swipe.assert_called_with(216, 720, 432, 1200, duration_ms=800) - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_with_string_coordinates(mock_driver): - """Verify swipe works with string-formatted coordinates.""" - # List format in string - result1 = await swipe.execute(driver=mock_driver, action="[920, 290, 920, 180]", duration=1000) - mock_driver.swipe.assert_called_with(993, 696, 993, 432, duration_ms=1000) - assert "Swipe completed successfully." in result1 - - # Space-separated format in string - result2 = await swipe.execute(driver=mock_driver, action="920 290 920 180", duration=1000) - assert "Swipe completed successfully." in result2 - - -def test_parse_swipe_parameters_direct(): - """Verify parse_swipe_parameters with various parameter formats.""" - from artemis.utils.coordinates import parse_swipe_parameters - - # List - k, t, d = parse_swipe_parameters([100, 200, 300, 400]) - assert k == "coords" - assert t == [100, 200, 300, 400] - - # String list - k, t, d = parse_swipe_parameters("[920, 290, 920, 180]") - assert k == "coords" - assert t == [920, 290, 920, 180] - - # String space separated - k, t, d = parse_swipe_parameters("920 290 920 180") - assert k == "coords" - assert t == [920, 290, 920, 180] - - # Dict with action string - k, t, d = parse_swipe_parameters({"action": "[920, 290, 920, 180]", "duration": 1000}) - assert k == "coords" - assert t == [920, 290, 920, 180] - assert d == 1000 - - # Dict with start and end - k, t, d = parse_swipe_parameters({"start": [100, 200], "end": [300, 400]}) - assert k == "coords" - assert t == [100, 200, 300, 400] - - # Direction string - k, t, d = parse_swipe_parameters({"action": "up"}) - assert k == "direction" - assert t == "up" - - -def test_compute_smart_swipe_coordinates_fallback(): - """Verify fallback smart swipe coordinates on full screen.""" - from artemis.utils.coordinates import compute_smart_swipe_coordinates - - sx, sy, ex, ey, dur = compute_smart_swipe_coordinates("up", width=1080, height=2400) - assert sx == 648 - assert ex == 648 - assert sy == 1680 - assert ey == 720 - assert 700 <= dur <= 900 - - -def test_compute_smart_swipe_coordinates_with_target_element(): - """Verify container-aware coordinates when target element index is specified.""" - from artemis.utils.coordinates import compute_smart_swipe_coordinates - - indexed_elements = [ - {"bounds": "[0, 200][1080, 1200]"}, # Element 1 - {"bounds": "[100, 500][900, 1500]"}, # Element 2 (width=800, height=1000) - ] - # Swipe up within element 2 - sx, sy, ex, ey, dur = compute_smart_swipe_coordinates( - direction="up", - target=2, - indexed_elements=indexed_elements, - width=1080, - height=2400, - ) - # c_left=100, c_top=500, c_width=800, c_height=1000 - # sx = ex = 100 + 800 * 0.6 = 580 - # sy = 500 + 1000 * 0.7 = 1200 - # ey = 500 + 1000 * 0.3 = 800 - assert sx == 580 - assert ex == 580 - assert sy == 1200 - assert ey == 800 - - -def test_compute_smart_swipe_coordinates_with_scrollable_hierarchy(): - """Verify auto-detection of scrollable container in UI hierarchy.""" - from artemis.utils.coordinates import compute_smart_swipe_coordinates - - ui_hierarchy = [ - { - "class": "android.widget.FrameLayout", - "bounds": "[0, 0][1080, 2400]", - "children": [ - { - "class": "androidx.recyclerview.widget.RecyclerView", - "scrollable": True, - "bounds": "[50, 300][1030, 2100]", # width=980, height=1800 - } - ], - } - ] - sx, sy, ex, ey, dur = compute_smart_swipe_coordinates( - direction="down", - ui_hierarchy=ui_hierarchy, - width=1080, - height=2400, - ) - # c_left=50, c_top=300, c_width=980, c_height=1800 - # sx = ex = 50 + 980 * 0.6 = 638 - # sy = 300 + 1800 * 0.3 = 840 - # ey = 300 + 1800 * 0.7 = 1560 - assert sx == 638 - assert ex == 638 - assert sy == 840 - assert ey == 1560