diff --git a/apps/admin_console/core/config.py b/apps/admin_console/core/config.py index 95385ba..b62b89a 100644 --- a/apps/admin_console/core/config.py +++ b/apps/admin_console/core/config.py @@ -12,63 +12,32 @@ # See the License for the specific language governing permissions and # limitations under the License. -import os -from pathlib import Path -import sys +"""Admin Console configuration facade. +All configuration and path management is centralized in `artemis.config`. +This module re-exports common constants for full backward compatibility. +""" -# Automatically write ANTIGRAVITY_LS_ADDRESS to shared synchronization file on startup -# to allow decoupled background processes to communicate with Jetski -def init_ls_address(): - ls_addr = os.environ.get("ANTIGRAVITY_LS_ADDRESS") - if ls_addr: - try: - from artemis.config import write_ls_address +from artemis.config import ( + DB_PATH, + IMAGES_DIR, + PAUSE_FILE, + REPLAY_BASE_DIR, + TEST_DATA_DIR, + TEST_OUTPUTS_DIR, + TRACES_PATH, + WORKSPACE_ROOT, + init_ls_address, +) - write_ls_address(ls_addr) - except Exception as e: - sys.stderr.write(f"Failed to write .jetski_ls_address on startup: {e}\n") - - -# Compute real WORKSPACE_ROOT (find repo root containing pyproject.toml or artemis) -_current_p = Path(__file__).resolve().parent -while _current_p != _current_p.parent: - if (_current_p / "pyproject.toml").exists() or (_current_p / "artemis").is_dir(): - WORKSPACE_ROOT = _current_p - break - _current_p = _current_p.parent -else: - WORKSPACE_ROOT = Path(__file__).resolve().parent.parent.parent.parent - -_apps_dir = WORKSPACE_ROOT / "apps" -_admin_console_dir = _apps_dir / "admin_console" -_cloud_service_dir = _apps_dir / "cloud_service" - -for _p in (str(WORKSPACE_ROOT), str(_apps_dir), str(_admin_console_dir), str(_cloud_service_dir)): - if _p not in sys.path: - sys.path.insert(0, _p) - -# Try to import settings from artemis -try: - from artemis.config import settings - - TRACES_PATH = Path(settings.TRACES_PATH) - DB_PATH = Path(settings.DATA_ENGINE_DB_PATH) -except ImportError: - # Fallback to default path if run outside project environment or imports fail - TRACES_PATH = WORKSPACE_ROOT / "traces" - DB_PATH = TRACES_PATH / "data_engine.db" - -# Step Replay Directories (Stored within traces/replay to avoid workspace clutter) -REPLAY_BASE_DIR = TRACES_PATH / "replay" -TEST_DATA_DIR = REPLAY_BASE_DIR / "data" -TEST_OUTPUTS_DIR = REPLAY_BASE_DIR / "outputs" -IMAGES_DIR = TRACES_PATH / "images" -PAUSE_FILE = WORKSPACE_ROOT / ".artemis_paused" - -# Ensure all essential runtime directories exist automatically -for _d in (TRACES_PATH, REPLAY_BASE_DIR, TEST_DATA_DIR, TEST_OUTPUTS_DIR, IMAGES_DIR): - try: - _d.mkdir(parents=True, exist_ok=True) - except Exception: - pass +__all__ = [ + "DB_PATH", + "IMAGES_DIR", + "PAUSE_FILE", + "REPLAY_BASE_DIR", + "TEST_DATA_DIR", + "TEST_OUTPUTS_DIR", + "TRACES_PATH", + "WORKSPACE_ROOT", + "init_ls_address", +] diff --git a/apps/admin_console/core/state.py b/apps/admin_console/core/state.py index 8149161..d79d1a0 100644 --- a/apps/admin_console/core/state.py +++ b/apps/admin_console/core/state.py @@ -16,10 +16,7 @@ import asyncio from collections.abc import Callable from typing import Any -try: - from admin_console.core.config import PAUSE_FILE -except ImportError: - from apps.admin_console.core.config import PAUSE_FILE +from artemis.config import PAUSE_FILE class ServerState: diff --git a/apps/admin_console/database/connection.py b/apps/admin_console/database/connection.py index a7f2530..2590b35 100644 --- a/apps/admin_console/database/connection.py +++ b/apps/admin_console/database/connection.py @@ -16,10 +16,7 @@ from contextlib import contextmanager from pathlib import Path import sqlite3 -try: - from admin_console.core.config import DB_PATH -except ImportError: - from apps.admin_console.core.config import DB_PATH +from artemis.config import DB_PATH _initialized_dbs = set() diff --git a/apps/admin_console/database/repositories/session_repository.py b/apps/admin_console/database/repositories/session_repository.py index 2ff4988..9350619 100644 --- a/apps/admin_console/database/repositories/session_repository.py +++ b/apps/admin_console/database/repositories/session_repository.py @@ -285,6 +285,17 @@ class SessionRepository: ("failed", now, row["session_id"], "running"), ) count += cursor.rowcount + try: + from mcp_server.utils import trace_store + + if trace_store.read_status(str(row["session_id"])): + trace_store.update_trace_status( + str(row["session_id"]), + "failed", + error="Process terminated prior to server startup.", + ) + except Exception: + pass conn.commit() return count except Exception: diff --git a/apps/admin_console/routers/media.py b/apps/admin_console/routers/media.py index 285941c..15178de 100644 --- a/apps/admin_console/routers/media.py +++ b/apps/admin_console/routers/media.py @@ -16,12 +16,12 @@ from pathlib import Path from fastapi import APIRouter, HTTPException from fastapi.responses import FileResponse, HTMLResponse +from artemis.config import IMAGES_DIR, WORKSPACE_ROOT + try: - from admin_console.core.config import IMAGES_DIR, WORKSPACE_ROOT from admin_console.database.repositories.session_repository import session_repo from admin_console.services.media_service import media_service except ImportError: - from apps.admin_console.core.config import IMAGES_DIR, WORKSPACE_ROOT from apps.admin_console.database.repositories.session_repository import session_repo from apps.admin_console.services.media_service import media_service diff --git a/apps/admin_console/routers/replay.py b/apps/admin_console/routers/replay.py index cdf5112..2a83b6e 100644 --- a/apps/admin_console/routers/replay.py +++ b/apps/admin_console/routers/replay.py @@ -15,12 +15,12 @@ import traceback from fastapi import APIRouter, HTTPException +from artemis.config import WORKSPACE_ROOT + try: - from admin_console.core.config import WORKSPACE_ROOT from admin_console.schemas.task_schema import ReplayRequest from admin_console.replay_manager import ReplayManager except ImportError: - from apps.admin_console.core.config import WORKSPACE_ROOT from apps.admin_console.schemas.task_schema import ReplayRequest from apps.admin_console.replay_manager import ReplayManager diff --git a/apps/admin_console/routers/sessions.py b/apps/admin_console/routers/sessions.py index d1f8169..fc40e88 100644 --- a/apps/admin_console/routers/sessions.py +++ b/apps/admin_console/routers/sessions.py @@ -17,15 +17,15 @@ import time from uuid import UUID from fastapi import APIRouter, HTTPException +from artemis.config import DB_PATH, TRACES_PATH + try: - from admin_console.core.config import DB_PATH, TRACES_PATH from admin_console.core.state import state from admin_console.database.repositories.session_repository import session_repo from admin_console.database.repositories.trace_repository import trace_repo from admin_console.services.media_service import media_service from admin_console.services.model_service import model_service except ImportError: - from apps.admin_console.core.config import DB_PATH, TRACES_PATH from apps.admin_console.core.state import state from apps.admin_console.database.repositories.session_repository import session_repo from apps.admin_console.database.repositories.trace_repository import trace_repo @@ -137,6 +137,18 @@ async def list_sessions(): try: harvested = session_repo.harvest_orphaned_sessions(orphaned_ids) print(f"[list_sessions] Auto-harvested {harvested} orphaned running session(s).") + try: + from mcp_server.utils import trace_store + + for o_id in orphaned_ids: + if trace_store.read_status(str(o_id)): + trace_store.update_trace_status( + str(o_id), + "failed", + error="Process terminated unexpectedly (auto-harvested).", + ) + except Exception: + pass except Exception as e: print(f"[list_sessions] Failed to update orphaned sessions: {e}") @@ -145,6 +157,15 @@ async def list_sessions(): raise HTTPException(status_code=500, detail=str(e)) +@router.get("/api/sessions/{session_id}") +async def get_session_details(session_id: str): + """Retrieve details for a single automation session.""" + row = session_repo.get_session_by_id(session_id) + if not row: + raise HTTPException(status_code=404, detail=f"Session {session_id} not found") + return dict(row) + + @router.get("/api/sessions/{session_id}/tree") async def get_tree(session_id: str): try: diff --git a/apps/admin_console/routers/steps.py b/apps/admin_console/routers/steps.py index 331b86a..4e4c31f 100644 --- a/apps/admin_console/routers/steps.py +++ b/apps/admin_console/routers/steps.py @@ -17,13 +17,13 @@ import traceback from fastapi import APIRouter, HTTPException from fastapi.responses import PlainTextResponse +from artemis.config import TEST_OUTPUTS_DIR + try: - from admin_console.core.config import TEST_OUTPUTS_DIR from admin_console.database.repositories.step_repository import step_repo from admin_console.database.repositories.trace_repository import trace_repo from admin_console.services.media_service import media_service except ImportError: - from apps.admin_console.core.config import TEST_OUTPUTS_DIR from apps.admin_console.database.repositories.step_repository import step_repo from apps.admin_console.database.repositories.trace_repository import trace_repo from apps.admin_console.services.media_service import media_service diff --git a/apps/admin_console/routers/tasks.py b/apps/admin_console/routers/tasks.py index 7c5f407..fa2ec30 100644 --- a/apps/admin_console/routers/tasks.py +++ b/apps/admin_console/routers/tasks.py @@ -298,6 +298,7 @@ async def get_status(): "background_tasks": bg_tasks, "queue": queue_data, "active_tasks": active_tasks, + "model_info": model_info, "ipc_port": state.ipc_port, } diff --git a/apps/admin_console/server.py b/apps/admin_console/server.py index f760761..bfea73f 100644 --- a/apps/admin_console/server.py +++ b/apps/admin_console/server.py @@ -55,17 +55,18 @@ from artemis.runtime import ( write_server_info, ) +from artemis.config import ( + DB_PATH, + IMAGES_DIR, + REPLAY_BASE_DIR, + TEST_DATA_DIR, + TEST_OUTPUTS_DIR, + TRACES_PATH, + WORKSPACE_ROOT, + init_ls_address, +) + try: - from admin_console.core.config import ( - DB_PATH, - IMAGES_DIR, - REPLAY_BASE_DIR, - TEST_DATA_DIR, - TEST_OUTPUTS_DIR, - TRACES_PATH, - WORKSPACE_ROOT, - init_ls_address, - ) from admin_console.core.state import ServerState, state from admin_console.database.connection import db_session, get_db from admin_console.database.repositories.session_repository import session_repo @@ -78,9 +79,6 @@ try: from admin_console.services.model_service import model_service from admin_console.services.task_queue_service import task_queue_service except ImportError: - from apps.admin_console.core.config import ( - init_ls_address, - ) from apps.admin_console.core.state import state from apps.admin_console.database.repositories.session_repository import session_repo from apps.admin_console.routers import media, replay, sessions, steps, stream, system, tasks diff --git a/apps/admin_console/services/media_service.py b/apps/admin_console/services/media_service.py index 6e2883a..5f295d3 100644 --- a/apps/admin_console/services/media_service.py +++ b/apps/admin_console/services/media_service.py @@ -22,10 +22,7 @@ from typing import Any import urllib.parse from fastapi import HTTPException -try: - from admin_console.core.config import IMAGES_DIR, TRACES_PATH, WORKSPACE_ROOT -except ImportError: - from apps.admin_console.core.config import IMAGES_DIR, TRACES_PATH, WORKSPACE_ROOT +from artemis.config import IMAGES_DIR, TRACES_PATH, WORKSPACE_ROOT class MediaService: diff --git a/apps/admin_console/services/task_queue_service.py b/apps/admin_console/services/task_queue_service.py index 762a7d3..e31b99a 100644 --- a/apps/admin_console/services/task_queue_service.py +++ b/apps/admin_console/services/task_queue_service.py @@ -24,27 +24,21 @@ from typing import Any import uuid try: - from admin_console.core.config import ( - PAUSE_FILE, - TEST_DATA_DIR, - TEST_OUTPUTS_DIR, - WORKSPACE_ROOT, - ) from admin_console.core.state import state from admin_console.database.repositories.session_repository import session_repo from admin_console.services.media_service import media_service - from artemis.runtime import DeviceExecutionLock, process_supervisor except ImportError: - from apps.admin_console.core.config import ( - PAUSE_FILE, - TEST_DATA_DIR, - TEST_OUTPUTS_DIR, - WORKSPACE_ROOT, - ) from apps.admin_console.core.state import state from apps.admin_console.database.repositories.session_repository import session_repo from apps.admin_console.services.media_service import media_service - from artemis.runtime import DeviceExecutionLock, process_supervisor + +from artemis.config import ( + PAUSE_FILE, + TEST_DATA_DIR, + TEST_OUTPUTS_DIR, + WORKSPACE_ROOT, +) +from artemis.runtime import DeviceExecutionLock, process_supervisor class TaskQueueService: @@ -314,6 +308,7 @@ class TaskQueueService: if sess_id: env["ARTEMIS_SESSION_ID"] = str(sess_id) env["ARTEMIS_TASK_INGRESS"] = str(task_item.get("ingress", "frontend")) + env["ARTEMIS_TASK_WORKER"] = "1" queue_ticket = task_item.get("queue_ticket") if queue_ticket: env[DeviceExecutionLock.QUEUE_TICKET_ENV] = str(queue_ticket) @@ -354,6 +349,16 @@ class TaskQueueService: ) state.current_process = proc task_item["pid"] = proc.pid + if sess_id: + try: + from mcp_server.utils import trace_store + + st = trace_store.read_status(str(sess_id)) + if st: + st["pid"] = proc.pid + trace_store.write_status(str(sess_id), st) + except Exception: + pass cls._broadcast_startup_progress( sess_id, "process_ready", "Execution process started" ) @@ -398,6 +403,17 @@ class TaskQueueService: f"[QueueWorker] Preserved authoritative session {sess_id} " f"status '{new_status}'" ) + try: + from mcp_server.utils import trace_store + + if trace_store.read_status(str(sess_id)): + canonical_mcp_status = "completed" if new_status in ("completed", "success") else new_status + trace_store.update_trace_status( + str(sess_id), + canonical_mcp_status, + ) + except Exception: + pass rec_info = session_repo.get_video_recording_for_session(sess_id) rec_status = (rec_info or {}).get("status") recovered_video_path = None @@ -428,6 +444,7 @@ class TaskQueueService: "recording_failed", {"session_id": sess_id, "error": recording_error}, ) + state.active_connections.pop(sess_id, None) cls._broadcast_event( "session_ended", @@ -439,16 +456,21 @@ class TaskQueueService: ) conversation_id = task_item.get("conversation_id") if task_item else None - if conversation_id: + if conversation_id or task_item.get("ingress") == "mcp": try: from mcp_server.notifiers import notify notify( - conversation_id=conversation_id, + conversation_id=conversation_id or "", message=f"Artemis autonomous task '{goal}' finished with status '{new_status}'.\nTrace ID: {sess_id}", title=f"Task {new_status.capitalize()}: {goal[:40]}", event_type=new_status, - payload={"session_id": sess_id, "status": new_status, "goal": goal}, + payload={ + "trace_id": sess_id, + "session_id": sess_id, + "status": new_status, + "goal": goal, + }, ) except Exception as notif_err: print(f"[QueueWorker] Notification dispatch notice: {notif_err}") @@ -492,6 +514,56 @@ class TaskQueueService: enqueued_tasks = [] now = time.time() + + # 1. Deduplication by session_id: if session_id is already running or queued, do not re-enqueue + if session_id and len(goals) == 1: + target_sid = str(session_id) + if state.active_session_id and str(state.active_session_id) == target_sid: + return { + "status": "started", + "tasks": [{"session_id": target_sid, "status": "running"}], + "enqueued_count": 0, + "total_queued": len(state.queue_tasks), + } + existing_item = next( + ( + item + for item in state.queue_items + if isinstance(item, dict) and str(item.get("session_id")) == target_sid + ), + None, + ) + if existing_item: + return { + "status": existing_item.get("status", "queued"), + "tasks": [existing_item], + "enqueued_count": 0, + "total_queued": len(state.queue_tasks), + } + + # 2. Debounce duplicate rapid submissions (e.g. UI double-click or network retry within 1s) + if len(goals) == 1: + first_goal = goals[0] + recent_duplicate = next( + ( + item + for item in reversed(state.queue_items) + if isinstance(item, dict) + and item.get("status") == "pending" + and item.get("goal") == first_goal + and (not device_serial or item.get("device_serial") == device_serial) + and (now - float(item.get("created_at", 0))) < 1.0 + ), + None, + ) + if recent_duplicate: + return { + "status": "queued", + "tasks": [recent_duplicate], + "enqueued_count": 0, + "total_queued": len(state.queue_tasks), + } + for i, goal in enumerate(goals): sess_id = (session_id if (session_id and len(goals) == 1) else str(uuid.uuid4())) assigned_serial = device_serial @@ -597,17 +669,18 @@ class TaskQueueService: sid = dev_owner.session_id if sid: session_repo.update_session_status(str(sid), "cancelled", time.time()) - if dev_owner.ingress == "mcp": - try: - from mcp_server.utils import trace_store + try: + from mcp_server.utils import trace_store + is_mcp = bool(dev_owner and dev_owner.ingress == "mcp") + if is_mcp or trace_store.read_status(str(sid)): trace_store.update_trace_status( str(sid), "cancelled", error="Task stopped from the Artemis frontend.", ) - except Exception: - pass + except Exception: + pass cls._broadcast_event( "session_ended", { @@ -791,17 +864,18 @@ class TaskQueueService: session_repo.update_session_status( str(stopped_session_id), "cancelled", time.time() ) - if owner and owner.ingress == "mcp": - try: - from mcp_server.utils import trace_store + try: + from mcp_server.utils import trace_store + is_mcp = bool(owner and owner.ingress == "mcp") + if is_mcp or trace_store.read_status(str(stopped_session_id)): trace_store.update_trace_status( str(stopped_session_id), "cancelled", error="Task stopped from the Artemis frontend.", ) - except Exception as exc: - print(f"Failed to update MCP cancellation status: {exc}") + except Exception as exc: + print(f"Failed to update MCP cancellation status: {exc}") cls._broadcast_event( "session_ended", { diff --git a/apps/showcase_ui/src/app/utils/action-formatter.util.spec.ts b/apps/showcase_ui/src/app/utils/action-formatter.util.spec.ts index 8f13244..bca2865 100644 --- a/apps/showcase_ui/src/app/utils/action-formatter.util.spec.ts +++ b/apps/showcase_ui/src/app/utils/action-formatter.util.spec.ts @@ -211,4 +211,67 @@ describe('extractStepReplayFrames', () => { expect(frames[0].imageUrl).toBe('/images/img_start'); expect(frames[0].stepNumber).toBe(1); }); + + it('should deduplicate steps with identical step_number and prefer real physical actions', () => { + const logs = [ + { + step_id: 'step-click', + step_number: 1, + pre_image_name: 'img_screen_1', + post_image_name: 'img_screen_2', + action_taken: { action: 'click', coordinates: [106, 101] }, + summary: 'In Step 1, I tapped the back arrow', + timestamp: 100 + }, + { + step_id: 'step-failed-report', + step_number: 1, + pre_image_name: 'img_screen_2', + post_image_name: null, + action_taken: { action: 'report_task_status', args: { status: 'failed' } }, + timestamp: 200 + }, + { + step_id: 'step-completed-report', + step_number: 2, + pre_image_name: 'img_screen_2', + post_image_name: null, + action_taken: { action: 'report_task_status', args: { status: 'completed' } }, + timestamp: 150 + } + ]; + + const frames = extractStepReplayFrames(logs); + // Duplicate step_number: 1 merges into Step 1, while Step 2 report is faithfully displayed + expect(frames.length).toBe(2); + expect(frames[0].stepNumber).toBe(1); + expect(frames[0].action?.action).toBe('click'); + expect(frames[0].actionText).toContain('Tapping Element'); + expect(frames[1].stepNumber).toBe(2); + expect(frames[1].title).toContain('Report Task Status'); + }); + + it('should enforce strictly sequential stepNumber without duplicates', () => { + const logs = [ + { + step_id: 's-1', + step_number: 1, + pre_image_name: 'img_1', + action_taken: { action: 'click', coordinates: [100, 100] }, + timestamp: 100 + }, + { + step_id: 's-2', + step_number: 2, + pre_image_name: 'img_2', + action_taken: { action: 'swipe', direction: 'up' }, + timestamp: 200 + } + ]; + + const frames = extractStepReplayFrames(logs); + expect(frames.length).toBe(2); + expect(frames[0].stepNumber).toBe(1); + expect(frames[1].stepNumber).toBe(2); + }); }); diff --git a/apps/showcase_ui/src/app/utils/action-formatter.util.ts b/apps/showcase_ui/src/app/utils/action-formatter.util.ts index f87d846..4361e41 100644 --- a/apps/showcase_ui/src/app/utils/action-formatter.util.ts +++ b/apps/showcase_ui/src/app/utils/action-formatter.util.ts @@ -33,9 +33,15 @@ export function getActionObject(action: any): any { * Check if the action is a user interaction action on Android */ export function isAndroidAction(action: any): boolean { - const act = getActionObject(action); - if (!act) return false; - const name = (act.name || act.action || '').toLowerCase(); + if (!action) return false; + let name = ''; + if (typeof action === 'string') { + name = action.toLowerCase(); + } else { + const act = getActionObject(action); + if (!act) return false; + name = (act.name || act.action || '').toLowerCase(); + } return [ 'tap', 'click', 'click_sequence', 'input', 'input_text', 'swipe', 'scroll', 'press_key', 'press_home', 'press_back', 'launch_app', 'manage_app', 'open_app', @@ -48,9 +54,16 @@ export function isAndroidAction(action: any): boolean { * Check if the action or tool is a task completion / submission report */ export function isReportStatusAction(action: any): boolean { - const act = getActionObject(action); - if (!act) return false; - const name = (act.name || act.action || '').toLowerCase().replace(/^(_)?(self\.)?exec_/, ''); + if (!action) return false; + let name = ''; + if (typeof action === 'string') { + name = action.toLowerCase(); + } else { + const act = getActionObject(action); + if (!act) return false; + name = (act.name || act.action || '').toLowerCase(); + } + name = name.replace(/^(_)?(self\.)?exec_/, ''); return name === 'report_task_status' || name === 'report_status' || name === 'submit_task_status'; } @@ -785,21 +798,31 @@ export function extractStepReplayFrames(logsOrSteps: any[]): StepReplayFrame[] { } } - // Deduplicate and merge by step_id or step_number + // Deduplicate and merge by step_number (fallback to step_id if step_number missing) const stepMap = new Map(); for (const step of rawSteps) { - const key = step.step_id || step.step_number; + const key = (step.step_number !== undefined && step.step_number !== null) + ? `step_${step.step_number}` + : (step.step_id || Math.random()); if (key !== undefined) { const existing = stepMap.get(key); if (existing) { + // When merging duplicate records with the same step_number: + // Prefer real actions over report_task_status + const existingAct = existing.action_taken; + const newAct = step.action_taken; + const preferExistingAction = existingAct && isAndroidAction(existingAct) && isReportStatusAction(newAct); + const preferNewAction = newAct && isAndroidAction(newAct) && isReportStatusAction(existingAct); + stepMap.set(key, { ...existing, ...step, step_number: step.step_number !== undefined ? step.step_number : existing.step_number, - pre_image_name: step.pre_image_name || existing.pre_image_name, - post_image_name: step.post_image_name || existing.post_image_name, - action_taken: step.action_taken || existing.action_taken, + pre_image_name: (preferExistingAction ? existing.pre_image_name : (step.pre_image_name || existing.pre_image_name)), + post_image_name: (preferExistingAction ? (existing.post_image_name || step.post_image_name) : (step.post_image_name || existing.post_image_name)), + action_taken: preferExistingAction ? existing.action_taken : (preferNewAction ? newAct : (step.action_taken || existing.action_taken)), last_execution_result: step.last_execution_result || existing.last_execution_result, + summary: (preferExistingAction ? (existing.summary || step.summary) : (step.summary || existing.summary)), generic_tools: step.generic_tools || existing.generic_tools }); } else { @@ -820,7 +843,8 @@ export function extractStepReplayFrames(logsOrSteps: any[]): StepReplayFrame[] { for (let i = 0; i < steps.length; i++) { const stepData = steps[i]; - const stepNum = Number(stepData.step_number ?? (i + 1)); + // Guarantee 1-indexed sequential step numbering so frames never duplicate numbers + const stepNum = i + 1; const act = stepData.action_taken; const title = act ? (getActionTitle(act) || 'Action') : `Step ${stepNum}`; const coords = act ? getActionCoords(act) : ''; diff --git a/artemis/config/__init__.py b/artemis/config/__init__.py index c7e7616..68a8b1c 100644 --- a/artemis/config/__init__.py +++ b/artemis/config/__init__.py @@ -107,16 +107,29 @@ from artemis.config.output import ( ) from artemis.config.paths import ( CONFIG_DIR, + DB_PATH, GLOBAL_APP_DIR, GLOBAL_JETSKI_DIR, + IMAGES_DIR, + PAUSE_FILE, + REPLAY_BASE_DIR, ROOT_DIR, + TEST_DATA_DIR, + TEST_OUTPUTS_DIR, + TRACES_PATH, + WORKSPACE_ROOT, get_app_dir, get_config_path, get_data_engine_db_path, get_default_traces_path, + get_images_dir, get_ipc_port_file, get_ls_address_file, + get_pause_file, + get_replay_dir, get_temp_dir, + get_test_data_dir, + get_test_outputs_dir, get_traces_dir, is_frozen_bundle, ) @@ -124,6 +137,7 @@ from artemis.config.runtime import ( cleanup_temp_dir, clear_ipc_port, clear_ls_address, + init_ls_address, read_ipc_port, read_ls_address, write_ipc_port, @@ -140,15 +154,28 @@ __all__ = [ "settings", # Paths & Dirs "ROOT_DIR", + "WORKSPACE_ROOT", "CONFIG_DIR", "GLOBAL_APP_DIR", "GLOBAL_JETSKI_DIR", + "TRACES_PATH", + "DB_PATH", + "PAUSE_FILE", + "REPLAY_BASE_DIR", + "TEST_DATA_DIR", + "TEST_OUTPUTS_DIR", + "IMAGES_DIR", "get_app_dir", "get_traces_dir", "get_default_traces_path", "get_temp_dir", "get_config_path", "get_data_engine_db_path", + "get_pause_file", + "get_replay_dir", + "get_test_data_dir", + "get_test_outputs_dir", + "get_images_dir", "get_ipc_port_file", "get_ls_address_file", "is_frozen_bundle", @@ -188,6 +215,7 @@ __all__ = [ "read_ls_address", "write_ls_address", "clear_ls_address", + "init_ls_address", "cleanup_temp_dir", # Types & Literals "LLMProvider", diff --git a/artemis/config/constants.py b/artemis/config/constants.py index 06bfc52..720becf 100644 --- a/artemis/config/constants.py +++ b/artemis/config/constants.py @@ -85,6 +85,11 @@ DATA_ENGINE_DB_FILENAME = "data_engine.db" IPC_PORT_FILENAME = ".artemis_ipc_port" LS_ADDRESS_FILENAME = ".jetski_ls_address" SERVER_INFO_FILENAME = ".artemis_server.json" +PAUSE_FILENAME = ".artemis_paused" +REPLAY_DIRNAME = "replay" +TEST_DATA_DIRNAME = "data" +TEST_OUTPUTS_DIRNAME = "outputs" +IMAGES_DIRNAME = "images" DOTENV_FILENAME = ".env" diff --git a/artemis/config/paths.py b/artemis/config/paths.py index 5731d8e..f21bb7a 100644 --- a/artemis/config/paths.py +++ b/artemis/config/paths.py @@ -24,9 +24,14 @@ from artemis.config.constants import ( ENV_ARTEMIS_APP_DIR, ENV_ARTEMIS_TRACES_DIR, ENV_ARTEMIS_USE_USER_DIR, + IMAGES_DIRNAME, IPC_PORT_FILENAME, LS_ADDRESS_FILENAME, + PAUSE_FILENAME, + REPLAY_DIRNAME, SERVER_INFO_FILENAME, + TEST_DATA_DIRNAME, + TEST_OUTPUTS_DIRNAME, ) from artemis.platform import platform @@ -168,3 +173,47 @@ def get_config_path(filename: str, default_bundled_path: Path | None = None) -> GLOBAL_APP_DIR = get_app_dir() GLOBAL_JETSKI_DIR = GLOBAL_APP_DIR # Backward-compatibility alias + + +def get_pause_file() -> Path: + """Returns the central pause signal file path.""" + return ROOT_DIR / PAUSE_FILENAME + + +def get_replay_dir() -> Path: + """Returns the central step replay directory.""" + replay_dir = get_default_traces_path() / REPLAY_DIRNAME + replay_dir.mkdir(parents=True, exist_ok=True) + return replay_dir + + +def get_test_data_dir() -> Path: + """Returns the test replay data directory.""" + d = get_replay_dir() / TEST_DATA_DIRNAME + d.mkdir(parents=True, exist_ok=True) + return d + + +def get_test_outputs_dir() -> Path: + """Returns the test replay outputs directory.""" + d = get_replay_dir() / TEST_OUTPUTS_DIRNAME + d.mkdir(parents=True, exist_ok=True) + return d + + +def get_images_dir() -> Path: + """Returns the traces images directory.""" + d = get_default_traces_path() / IMAGES_DIRNAME + d.mkdir(parents=True, exist_ok=True) + return d + + +# Module-level convenience constants & backward-compatible aliases +WORKSPACE_ROOT = ROOT_DIR +TRACES_PATH = get_default_traces_path() +DB_PATH = get_data_engine_db_path() +PAUSE_FILE = get_pause_file() +REPLAY_BASE_DIR = get_replay_dir() +TEST_DATA_DIR = get_test_data_dir() +TEST_OUTPUTS_DIR = get_test_outputs_dir() +IMAGES_DIR = get_images_dir() diff --git a/artemis/config/runtime.py b/artemis/config/runtime.py index dc79a5c..b10bab7 100644 --- a/artemis/config/runtime.py +++ b/artemis/config/runtime.py @@ -181,3 +181,16 @@ def cleanup_temp_dir(subfolder: str | None = None, max_age_seconds: float | None logger.warning(f"Could not delete temporary file {file_path}: {e}") return deleted_count + + +def init_ls_address() -> None: + """Automatically write ANTIGRAVITY_LS_ADDRESS to shared synchronization file on startup. + + Allows decoupled background processes to communicate with Jetski. + """ + ls_addr = os.environ.get(ENV_ANTIGRAVITY_LS_ADDRESS) + if ls_addr: + try: + write_ls_address(ls_addr) + except Exception as e: + logger.debug(f"Failed to write LS address on startup: {e}") diff --git a/artemis/data_engine/engine.py b/artemis/data_engine/engine.py index c0ba9e6..97e02ff 100644 --- a/artemis/data_engine/engine.py +++ b/artemis/data_engine/engine.py @@ -24,7 +24,7 @@ import time from typing import Any from uuid import UUID, uuid4 -from artemis.config import get_ipc_port_file, read_ipc_port, settings +from artemis.config import PAUSE_FILE, get_ipc_port_file, read_ipc_port, settings from artemis.context import ArtemisContext from artemis.data_engine.models import ( BackgroundTaskRecord, @@ -251,7 +251,7 @@ class DataEngine: self.session_start_time = time.time() # Clear old pause file if it exists - pause_file = self.global_base_dir.parent / ".artemis_paused" + pause_file = PAUSE_FILE if pause_file.exists(): try: pause_file.unlink() @@ -276,8 +276,18 @@ class DataEngine: ) except Exception as exc: logger.debug(f"Could not annotate active device owner: {exc}") - self.current_step_number = 0 - logger.info(f"Session started: {session_id}") + # Initialize step counter from existing steps if resuming an existing session + existing_steps = [] + if self.storage and hasattr(self.storage, "get_steps"): + try: + existing_steps = self.storage.get_steps(session_id) or [] + except Exception: + existing_steps = [] + if existing_steps: + self.current_step_number = max([s.step_number for s in existing_steps], default=0) + else: + self.current_step_number = 0 + logger.info(f"Session started: {session_id} (current step counter: {self.current_step_number})") self._publish("session_started", session.model_dump()) return session_id @@ -287,14 +297,13 @@ class DataEngine: return # Clear pause file if it exists - if hasattr(self, "global_base_dir") and self.global_base_dir: - pause_file = self.global_base_dir.parent / ".artemis_paused" - if pause_file.exists(): - try: - pause_file.unlink() - logger.info("Removed pause file on session end.") - except Exception as e: - logger.error(f"Failed to delete pause file on session end: {e}") + pause_file = PAUSE_FILE + if pause_file.exists(): + try: + pause_file.unlink() + logger.info("Removed pause file on session end.") + except Exception as e: + logger.error(f"Failed to delete pause file on session end: {e}") session_id = self.current_session_id end_time = time.time() @@ -484,7 +493,14 @@ class DataEngine: raise ValueError("No active session. Call start_session first.") with self.lock: - self.current_step_number += 1 + existing_steps = [] + if self.storage and hasattr(self.storage, "get_steps") and self.current_session_id: + try: + existing_steps = self.storage.get_steps(self.current_session_id) or [] + except Exception: + existing_steps = [] + max_existing = max([s.step_number for s in existing_steps], default=0) if existing_steps else 0 + self.current_step_number = max(self.current_step_number, max_existing) + 1 step_number = self.current_step_number step_id = self.current_step_id or uuid4() diff --git a/artemis/data_engine/storage.py b/artemis/data_engine/storage.py index 454716f..ccdfdc3 100644 --- a/artemis/data_engine/storage.py +++ b/artemis/data_engine/storage.py @@ -697,6 +697,8 @@ class StorageManager: ) return steps + get_session_steps = get_steps + def get_traces_for_step(self, step_id: UUID | str) -> list[TraceRecord]: """Retrieve all traces for a step.""" traces = [] diff --git a/artemis/interfaces/cli/commands/batch.py b/artemis/interfaces/cli/commands/batch.py index 3185a48..f295e04 100644 --- a/artemis/interfaces/cli/commands/batch.py +++ b/artemis/interfaces/cli/commands/batch.py @@ -37,6 +37,8 @@ async def run_batch_tasks( delay_seconds: float = 5.0, ) -> None: """Executes a list of automation tasks sequentially.""" + if not os.environ.get("ARTEMIS_TASK_INGRESS"): + os.environ["ARTEMIS_TASK_INGRESS"] = "cli" llm_config = initialize_llm_config() profile = AgentProfile(name="default", llm_config=llm_config) config = Builders.AgentConfig.with_default_profile(profile).build() @@ -116,6 +118,13 @@ def batch_command( help="Delay in seconds between successive tasks.", ), ] = 5.0, + standalone: Annotated[ + bool, + typer.Option( + "--standalone", + help="Run in standalone embedded mode without routing through Artemis Daemon.", + ), + ] = False, ) -> None: """Execute multiple automation tasks in sequence.""" task_list: list[str] = [] @@ -157,4 +166,54 @@ def batch_command( ) raise typer.Exit(1) + is_standalone = standalone or os.environ.get("ARTEMIS_STANDALONE") == "1" + if not is_standalone: + try: + from artemis.runtime import ( + ensure_daemon_running, + submit_batch_to_daemon, + wait_for_daemon_task, + ) + + console = Console() + console.print("[dim]Connecting to Artemis Daemon scheduler...[/dim]") + is_running, base_url = ensure_daemon_running(timeout=8.0, wait_ready=True) + if is_running and base_url: + console.print(f"[bold green]✓[/bold green] Artemis Daemon active at [cyan]{base_url}[/cyan]") + resp = submit_batch_to_daemon(task_list, profile=profile, base_url=base_url) + if resp and resp.get("tasks"): + console.print(f"[bold green]✓[/bold green] Enqueued {len(task_list)} batch tasks in Daemon scheduler.\n") + results = [] + for idx, t_item in enumerate(resp["tasks"], 1): + sid = t_item.get("session_id") + goal = t_item.get("goal") + console.print(f"[cyan]Waiting for Task {idx}/{len(task_list)}: '{goal}' (Session: {sid})...[/cyan]") + final_sess = wait_for_daemon_task(sid, base_url=base_url, timeout=1800.0) + st = final_sess.get("status") + err = final_sess.get("error") or "" + results.append({ + "task": goal, + "status": "SUCCESS" if st in ("completed", "success") else "FAILED", + "error": err, + }) + + table = Table(title="Batch Execution Summary") + table.add_column("#", justify="right", style="cyan") + table.add_column("Task Goal", style="white") + table.add_column("Status", justify="center") + table.add_column("Notes", style="red") + + for i, res in enumerate(results, 1): + status_styled = ( + "[bold green]PASS[/bold green]" + if res["status"] == "SUCCESS" + else "[bold red]FAIL[/bold red]" + ) + table.add_row(str(i), res["task"], status_styled, res["error"]) + + console.print(table) + return + except Exception as exc: + logger.debug(f"Batch daemon routing notice: {exc}") + asyncio.run(run_batch_tasks(task_list, profile_name=profile, delay_seconds=delay)) diff --git a/artemis/interfaces/cli/commands/run.py b/artemis/interfaces/cli/commands/run.py index 3905edc..4ba964b 100644 --- a/artemis/interfaces/cli/commands/run.py +++ b/artemis/interfaces/cli/commands/run.py @@ -84,6 +84,8 @@ async def execute_task( ) if effective_sid: os.environ["ARTEMIS_SESSION_ID"] = str(effective_sid) + if not os.environ.get("ARTEMIS_TASK_INGRESS"): + os.environ["ARTEMIS_TASK_INGRESS"] = "cli" publish_startup_progress( "configuration", "Loading the run configuration", session_id=str(effective_sid) if effective_sid else None ) @@ -324,13 +326,80 @@ def run_command( console = Console() - # Daemon auto-spawn: By default, ensure Daemon is running in background unless standalone mode requested - if not standalone and os.environ.get("ARTEMIS_STANDALONE") != "1": + is_worker = ( + os.environ.get("ARTEMIS_TASK_WORKER") == "1" + or os.environ.get("ARTEMIS_DEVICE_QUEUE_TICKET") is not None + ) + is_standalone = standalone or os.environ.get("ARTEMIS_STANDALONE") == "1" + + # All platforms route through unified Artemis Daemon unless specifically configured as standalone + if not is_worker and not is_standalone: try: - from artemis.runtime import ensure_daemon_running - ensure_daemon_running(timeout=1.5) + import uuid + from artemis.runtime import ( + ensure_daemon_running, + submit_task_to_daemon, + wait_for_daemon_task, + stop_task_on_daemon, + ) + + console.print("[dim]Connecting to Artemis Daemon scheduler...[/dim]") + is_running, base_url = ensure_daemon_running(timeout=8.0, wait_ready=True) + if is_running and base_url: + target_sid = session_id or str(uuid.uuid4()) + console.print(f"[bold green]✓[/bold green] Artemis Daemon active at [cyan]{base_url}[/cyan]") + resp = submit_task_to_daemon( + goal=goal, + profile=profile or "pro", + device_serial=device_serial, + expected_output=output_description, + enable_outputter=enable_outputter, + locked_app_package=locked_app_package, + app_path=app_path, + session_id=target_sid, + ingress="cli", + base_url=base_url, + ) + if resp and resp.get("tasks"): + console.print(f"[bold green]✓[/bold green] Task scheduled in unified queue (Session: [cyan]{target_sid}[/cyan])") + console.print(f"[dim]Live dashboard & replay: {base_url}[/dim]\n") + + def on_status(sess_info): + st = sess_info.get("status") + if st == "queued": + console.print(f"[yellow]⏳ Task queued in scheduler, waiting for device...[/yellow]") + elif st == "running": + console.print(f"[green]▶ Task executing on mobile device...[/green]") + + try: + final_res = wait_for_daemon_task( + target_sid, + base_url=base_url, + timeout=1800.0, + on_status_update=on_status, + ) + final_st = final_res.get("status") + if final_st in ("completed", "success"): + console.print(f"\n[bold green]✅ Task completed successfully![/bold green]") + return + else: + err = final_res.get("error") or final_res.get("explanation") or "" + console.print(f"\n[bold red]✖ Task {final_st}[/bold red]: {err}") + raise SystemExit(1) + except KeyboardInterrupt: + console.print("\n[yellow]Stopping task on Daemon...[/yellow]") + stop_task_on_daemon(target_sid, base_url=base_url) + console.print("[yellow]Task cancelled.[/yellow]") + raise SystemExit(130) + else: + console.print("[yellow]Warning: Could not enqueue task to Daemon. Falling back to local execution...[/yellow]") + else: + console.print("[yellow]Warning: Artemis Daemon could not be started. Falling back to local execution...[/yellow]") + except SystemExit: + raise except Exception as exc: - logger.debug(f"Daemon probe/auto-spawn skipped: {exc}") + logger.debug(f"Daemon dispatch error: {exc}") + console.print(f"[yellow]Daemon routing notice: {exc}. Falling back to local execution...[/yellow]") adb_client = None try: diff --git a/artemis/interfaces/sdk/client.py b/artemis/interfaces/sdk/client.py index 1ee185e..ebb03ac 100644 --- a/artemis/interfaces/sdk/client.py +++ b/artemis/interfaces/sdk/client.py @@ -26,6 +26,8 @@ from artemis.runtime import ( DeviceStatus, device_pool, ensure_daemon_running, + submit_task_to_daemon, + wait_for_daemon_task, ) from artemis.utils.logger import get_logger @@ -145,6 +147,9 @@ class ArtemisClient: ) task_session_id = str(uuid.uuid4()) + prev_ingress = os.environ.get("ARTEMIS_TASK_INGRESS") + prev_sess_id = os.environ.get("ARTEMIS_SESSION_ID") + lock = DeviceExecutionLock( device_id=target_device, description=f"{goal[:120]}", @@ -154,19 +159,58 @@ class ArtemisClient: ingress="sdk", ) - await asyncio.to_thread( - lock.acquire, - blocking=blocking, - timeout=lock_timeout, - ) - try: + os.environ["ARTEMIS_TASK_INGRESS"] = "sdk" + os.environ["ARTEMIS_SESSION_ID"] = task_session_id + + await asyncio.to_thread( + lock.acquire, + blocking=blocking, + timeout=lock_timeout, + ) + # Check if targeting a live connected device devices = device_pool.list_devices() connected_serials = {d.serial for d in devices} is_live_device = target_device in connected_serials and os.environ.get("ARTEMIS_MOCK_DRIVER") != "1" if is_live_device: + # Route through unified Artemis Daemon scheduler unless standalone explicitly configured + if not self.standalone and os.environ.get("ARTEMIS_STANDALONE") != "1": + try: + is_running, base_url = ensure_daemon_running(timeout=8.0, wait_ready=True) + if is_running and base_url: + dev_serial = target_device if target_device != "default-device" else None + resp = submit_task_to_daemon( + goal=goal, + profile=target_profile, + device_serial=dev_serial, + expected_output=kwargs.get("expected_output"), + enable_outputter=kwargs.get("enable_outputter"), + locked_app_package=kwargs.get("locked_package") or kwargs.get("locked_app_package"), + app_path=kwargs.get("app_path"), + session_id=task_session_id, + ingress="sdk", + base_url=base_url, + ) + if resp and resp.get("tasks"): + final_res = wait_for_daemon_task( + task_session_id, + base_url=base_url, + timeout=float(kwargs.get("timeout", 1800.0)), + ) + final_st = final_res.get("status", "completed") + err = final_res.get("error") + return TaskResult( + trace_id=task_session_id, + status="completed" if final_st in ("completed", "success") else final_st, + turns=final_res.get("turns", 1), + error=err, + device_id=target_device, + ) + except Exception as exc: + logger.debug(f"SDK daemon routing notice: {exc}") + from artemis.sdk.agent import Agent from artemis.sdk.builders import Builders from artemis.context import DevicePlatform @@ -215,7 +259,16 @@ class ArtemisClient: device_id=target_device, ) finally: - await asyncio.to_thread(lock.release) + if prev_sess_id is not None: + os.environ["ARTEMIS_SESSION_ID"] = prev_sess_id + else: + os.environ.pop("ARTEMIS_SESSION_ID", None) + if prev_ingress is not None: + os.environ["ARTEMIS_TASK_INGRESS"] = prev_ingress + else: + os.environ.pop("ARTEMIS_TASK_INGRESS", None) + if lock._acquired: + await asyncio.to_thread(lock.release) async def stream_run( self, @@ -245,19 +298,28 @@ class ArtemisClient: f"with profile '{target_profile}' (concurrency_mode='{mode}')..." ) + task_session_id = str(uuid.uuid4()) + prev_ingress = os.environ.get("ARTEMIS_TASK_INGRESS") + prev_sess_id = os.environ.get("ARTEMIS_SESSION_ID") + lock = DeviceExecutionLock( device_id=target_device, description=f"ArtemisClient streaming task: {goal[:120]}", concurrency_mode=mode, max_concurrency=concurrency_limit, + session_id=task_session_id, + ingress="sdk", ) - await asyncio.to_thread( - lock.acquire, - blocking=blocking, - timeout=lock_timeout, - ) try: + os.environ["ARTEMIS_TASK_INGRESS"] = "sdk" + os.environ["ARTEMIS_SESSION_ID"] = task_session_id + + await asyncio.to_thread( + lock.acquire, + blocking=blocking, + timeout=lock_timeout, + ) # 1. Emit initial starting event start_event = StreamEvent( event_type=StreamEventType.STATUS, @@ -341,7 +403,16 @@ class ArtemisClient: pass yield err_ev finally: - await asyncio.to_thread(lock.release) + if prev_sess_id is not None: + os.environ["ARTEMIS_SESSION_ID"] = prev_sess_id + else: + os.environ.pop("ARTEMIS_SESSION_ID", None) + if prev_ingress is not None: + os.environ["ARTEMIS_TASK_INGRESS"] = prev_ingress + else: + os.environ.pop("ARTEMIS_TASK_INGRESS", None) + if lock._acquired: + await asyncio.to_thread(lock.release) async def run_task( self, diff --git a/artemis/runtime/__init__.py b/artemis/runtime/__init__.py index f8c5b4a..08504ad 100644 --- a/artemis/runtime/__init__.py +++ b/artemis/runtime/__init__.py @@ -22,8 +22,13 @@ from artemis.runtime.device_lock import ( from artemis.runtime.device_pool import DevicePool, DeviceStatus, device_pool from artemis.runtime.daemon_client import ( ensure_daemon_running, + get_daemon_session, + get_daemon_status, is_daemon_running, + stop_task_on_daemon, + submit_batch_to_daemon, submit_task_to_daemon, + wait_for_daemon_task, ) from artemis.runtime.awake_service import ( ensure_device_awake, @@ -54,6 +59,8 @@ __all__ = [ "ensure_daemon_running", "ensure_device_awake", "find_server_pids", + "get_daemon_session", + "get_daemon_status", "get_server_status", "is_daemon_running", "is_port_in_use", @@ -65,6 +72,9 @@ __all__ = [ "shutdown_awake_service", "start_awake_service", "stop_server", + "stop_task_on_daemon", + "submit_batch_to_daemon", "submit_task_to_daemon", + "wait_for_daemon_task", "write_server_info", ] diff --git a/artemis/runtime/daemon_client.py b/artemis/runtime/daemon_client.py index 3befe5e..5ccd41a 100644 --- a/artemis/runtime/daemon_client.py +++ b/artemis/runtime/daemon_client.py @@ -162,7 +162,7 @@ def submit_task_to_daemon( ingress: str = "client", conversation_id: str | None = None, base_url: str | None = None, - timeout: float = 5.0, + timeout: float = 15.0, ) -> dict[str, Any] | None: """Submit a task to the running Daemon via REST API. @@ -197,3 +197,153 @@ def submit_task_to_daemon( logger.debug(f"Failed to submit task to Daemon at {url}: {exc}") return None return None + + +def stop_task_on_daemon( + session_id: str, + *, + base_url: str | None = None, + timeout: float = 5.0, +) -> bool: + """Request the running Artemis Daemon to stop a task by session ID.""" + url = f"{base_url or f'http://{DEFAULT_DAEMON_HOST}:{DEFAULT_DAEMON_PORT}'}/api/stop" + payload = {"session_id": session_id} + try: + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + url, + data=data, + headers={"Content-Type": "application/json", "User-Agent": "Artemis-Client"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=timeout) as response: + if response.status == 200: + resp_data = json.loads(response.read().decode("utf-8")) + return resp_data.get("status") == "stopped" + except Exception as exc: + logger.debug(f"Failed to request stop from Daemon at {url}: {exc}") + return False + return False + + +def get_daemon_session( + session_id: str, + *, + base_url: str | None = None, + timeout: float = 5.0, +) -> dict[str, Any] | None: + """Fetch details of a single session from the Artemis Daemon.""" + url = f"{base_url or f'http://{DEFAULT_DAEMON_HOST}:{DEFAULT_DAEMON_PORT}'}/api/sessions/{session_id}" + try: + req = urllib.request.Request( + url, + headers={"User-Agent": "Artemis-Client"}, + method="GET", + ) + with urllib.request.urlopen(req, timeout=timeout) as response: + if response.status == 200: + return json.loads(response.read().decode("utf-8")) + except Exception as exc: + logger.debug(f"Failed to get session {session_id} from Daemon at {url}: {exc}") + return None + return None + + +def get_daemon_status( + *, + base_url: str | None = None, + timeout: float = 5.0, +) -> dict[str, Any] | None: + """Fetch global scheduler and device status from the Artemis Daemon.""" + url = f"{base_url or f'http://{DEFAULT_DAEMON_HOST}:{DEFAULT_DAEMON_PORT}'}/api/status" + try: + req = urllib.request.Request( + url, + headers={"User-Agent": "Artemis-Client"}, + method="GET", + ) + with urllib.request.urlopen(req, timeout=timeout) as response: + if response.status == 200: + return json.loads(response.read().decode("utf-8")) + except Exception as exc: + logger.debug(f"Failed to get status from Daemon at {url}: {exc}") + return None + return None + + +def submit_batch_to_daemon( + goals: list[str], + *, + profile: str = "flash", + device_serial: str | None = None, + ingress: str = "cli", + base_url: str | None = None, + timeout: float = 15.0, +) -> dict[str, Any] | None: + """Submit a batch of goals to the running Daemon.""" + url = f"{base_url or f'http://{DEFAULT_DAEMON_HOST}:{DEFAULT_DAEMON_PORT}'}/api/run" + payload = { + "goals": goals, + "profile": profile, + "device_serial": device_serial, + "ingress": ingress, + } + try: + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + url, + data=data, + headers={"Content-Type": "application/json", "User-Agent": "Artemis-Client"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=timeout) as response: + if response.status in (200, 201): + return json.loads(response.read().decode("utf-8")) + except Exception as exc: + logger.debug(f"Failed to submit batch to Daemon at {url}: {exc}") + return None + return None + + +def wait_for_daemon_task( + session_id: str, + *, + base_url: str | None = None, + timeout: float = 600.0, + poll_interval: float = 1.0, + on_status_update: Any = None, +) -> dict[str, Any]: + """Poll the Daemon until the task reaches a terminal status (completed, failed, cancelled).""" + started = time.monotonic() + last_status = None + + while time.monotonic() - started < timeout: + sess = get_daemon_session(session_id, base_url=base_url) + if sess: + status = sess.get("status") + if status != last_status: + last_status = status + if callable(on_status_update): + on_status_update(sess) + + if status in ("completed", "success", "failed", "cancelled"): + return sess + else: + # Check if task is queued in daemon status + st = get_daemon_status(base_url=base_url) + if st: + queue = st.get("queue") or [] + is_queued = any( + isinstance(item, dict) and str(item.get("session_id")) == str(session_id) + for item in queue + ) + current_st = "queued" if is_queued else "launching" + if current_st != last_status: + last_status = current_st + if callable(on_status_update): + on_status_update({"status": current_st, "session_id": session_id}) + + time.sleep(poll_interval) + + return {"session_id": session_id, "status": "timeout", "error": f"Timed out after {timeout}s"} + diff --git a/artemis/sdk/agent.py b/artemis/sdk/agent.py index 200b4db..cdc616d 100644 --- a/artemis/sdk/agent.py +++ b/artemis/sdk/agent.py @@ -551,7 +551,7 @@ class Agent: concurrency_mode=effective_mode, max_concurrency=effective_max, session_id=str(sess_id) if sess_id else None, - ingress="agent", + ingress=os.getenv("ARTEMIS_TASK_INGRESS") or "agent", ) try: if device_lock is not None and os.environ.get("ARTEMIS_CLOUD_MODE") != "1": diff --git a/artemis/services/llm.py b/artemis/services/llm.py index a46661a..dccd911 100644 --- a/artemis/services/llm.py +++ b/artemis/services/llm.py @@ -35,6 +35,7 @@ from uuid import uuid4 from langchain_core.language_models.chat_models import BaseChatModel from artemis.config import ( + PAUSE_FILE, AgentNode, AgentNodeWithFallback, LLMUtilsNode, @@ -233,7 +234,7 @@ def _handle_llm_pause_and_resume(last_error: Exception) -> Path: err_msg = err_msg[:1000] + "... [Truncated by LLM Wrapper]" llm_logger.warning(f"LLM Error: {err_msg}. Pausing execution to wait for resume signal...") - pause_file = Path(settings.TRACES_PATH).parent / ".artemis_paused" + pause_file = PAUSE_FILE try: pause_file.write_text(f"LLM Error: {err_msg}") except Exception: @@ -594,7 +595,7 @@ async def invoke_llm_with_timeout_message[T]( if llm_task.done(): return llm_task.result() - pause_file = Path(settings.TRACES_PATH).parent / ".artemis_paused" + pause_file = PAUSE_FILE if pause_file.exists(): start_time = asyncio.get_event_loop().time() continue diff --git a/mcp_server/notifiers/file.py b/mcp_server/notifiers/file.py index 3dd8078..9e4a635 100644 --- a/mcp_server/notifiers/file.py +++ b/mcp_server/notifiers/file.py @@ -44,7 +44,7 @@ class FileNotifier(BaseNotifier): event_type: str = "completed", payload: dict[str, Any] | None = None, ) -> bool: - trace_id = (payload or {}).get("trace_id") + trace_id = (payload or {}).get("trace_id") or (payload or {}).get("session_id") if not trace_id: return False diff --git a/mcp_server/tools/task_manager.py b/mcp_server/tools/task_manager.py index a7a9c79..cbe51ea 100644 --- a/mcp_server/tools/task_manager.py +++ b/mcp_server/tools/task_manager.py @@ -94,6 +94,38 @@ def mobile_manage_task( pid = status_data.get("pid") is_alive = False + project_root = env_utils.get_project_root() + db_path = os.path.join(trace_store.TRACES_DIR, "data_engine.db") + if not os.path.exists(db_path): + db_path = os.path.join(project_root, "traces", "data_engine.db") + + # If status is still running or pending, sync from database if available + if current_status in ("running", "pending") and os.path.exists(db_path): + try: + conn_st = sqlite3.connect(db_path) + conn_st.row_factory = sqlite3.Row + cur_st = conn_st.cursor() + cur_st.execute( + "SELECT status, pid FROM sessions WHERE session_id = ? ORDER BY start_time DESC LIMIT 1", + (trace_id,), + ) + row_st = cur_st.fetchone() + if row_st: + db_status = row_st["status"] + db_pid = row_st["pid"] + if db_pid and not pid: + pid = db_pid + status_data["pid"] = pid + trace_store.write_status(trace_id, status_data) + if db_status in ("completed", "success", "failed", "cancelled"): + canonical_status = "completed" if db_status in ("completed", "success") else db_status + current_status = canonical_status + status_data["status"] = canonical_status + trace_store.write_status(trace_id, status_data) + conn_st.close() + except Exception: + pass + if pid and current_status in ("running", "pending"): try: os.kill(pid, 0) @@ -300,50 +332,78 @@ def mobile_manage_task( "message": f"Cannot stop task because it is already in a terminal state: {current_status}", } + # 1. First attempt graceful cancellation via unified Daemon if available + stopped_via_daemon = False + if os.environ.get("ARTEMIS_STANDALONE") != "1": + try: + from artemis.runtime import is_daemon_running, stop_task_on_daemon + + if is_daemon_running(): + stopped_via_daemon = stop_task_on_daemon(trace_id) + except Exception: + pass + + # 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 + + # 3. If process PID is missing, try to resolve from active device owners if not pid: + try: + active_owners = DeviceExecutionLock.get_active_owners() + for dev_owner in active_owners.values(): + if dev_owner.session_id and str(dev_owner.session_id) == trace_id: + pid = dev_owner.pid + status_data["pid"] = pid + trace_store.write_status(trace_id, status_data) + break + except Exception: + pass + + # 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: return { "trace_id": trace_id, "status": current_status, "message": "Process ID (PID) is missing from the task status. Cannot stop task.", } - try: - DeviceExecutionLock.cancel_reservation(status_data.get("queue_ticket")) - if sys.platform == "win32": - if not process_supervisor.terminate_tree(pid): - import psutil + # 5. If process PID is available, terminate the process tree + if pid: + try: + if sys.platform == "win32": + if not process_supervisor.terminate_tree(pid): + import psutil - if psutil.pid_exists(pid): - raise RuntimeError( - f"Failed to terminate Windows process tree rooted at {pid}" - ) - DeviceExecutionLock.cleanup_stale_locks() - else: - # Preserve the existing POSIX signal behavior. - os.kill(pid, signal.SIGTERM) - trace_store.update_trace_status( - trace_id, "cancelled", error="Task aborted by user request." - ) - return { - "trace_id": trace_id, - "status": "cancelled", - "message": f"Successfully sent termination signal to background process {pid}.", - } - except ProcessLookupError: - trace_store.update_trace_status( - trace_id, "cancelled", error="Task aborted (process was already stopped)." - ) - return { - "trace_id": trace_id, - "status": "cancelled", - "message": f"Process {pid} was not found; it may have already exited. Status marked as cancelled.", - } - except Exception as e: - return { - "trace_id": trace_id, - "status": current_status, - "message": f"Failed to terminate process {pid}: {e}", - } + if psutil.pid_exists(pid): + raise RuntimeError( + f"Failed to terminate Windows process tree rooted at {pid}" + ) + DeviceExecutionLock.cleanup_stale_locks() + else: + os.kill(pid, signal.SIGTERM) + except ProcessLookupError: + pass + except Exception as e: + if not stopped_via_daemon: + return { + "trace_id": trace_id, + "status": current_status, + "message": f"Failed to terminate process {pid}: {e}", + } + + trace_store.update_trace_status( + trace_id, "cancelled", error="Task aborted by user request." + ) + return { + "trace_id": trace_id, + "status": "cancelled", + "message": f"Successfully stopped background task '{trace_id}'.", + } elif action == "inject_instruction": if not instruction: diff --git a/mcp_server/tools/task_runner.py b/mcp_server/tools/task_runner.py index db16cf3..7097b3f 100644 --- a/mcp_server/tools/task_runner.py +++ b/mcp_server/tools/task_runner.py @@ -202,11 +202,53 @@ def mobile_run_task( "stdout_log": stdout_log, "stderr_log": stderr_log, } + + # Check if the task was already enqueued despite a slow or None response: + try: + queued_tasks = DeviceExecutionLock.get_queued_tasks() + for q in queued_tasks: + if str(q.get("session_id")) == str(trace_id): + trace_store.update_trace_status( + trace_id, "running", device_serial=device_serial + ) + return { + "trace_id": trace_id, + "status": "running", + "message": ( + f"Autonomous task '{task_desc}' enqueued via unified Artemis Daemon.\n" + f"Trace ID: {trace_id}\n" + f"Model: {canonical_model}\n" + f"Live telemetry: streaming to web workspace." + ), + "notes_dir": str(trace_store.get_trace_notes_dir(trace_id)), + "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 + + # When Daemon is running, refuse to launch a conflicting standalone runner on the same device + import logging + logging.getLogger("mcp_server").warning( + f"Daemon dispatch failed for task '{task_desc}'. Aborting dispatch to prevent runner collision." + ) + return { + "trace_id": trace_id, + "status": "failed", + "error": f"Failed to enqueue task '{task_desc}' to running Artemis Daemon scheduler.", + "message": "Task rejected or timed out in Artemis Daemon. Standalone fallback blocked to prevent runner collision.", + } except Exception as exc: import logging logging.getLogger("mcp_server").warning( - f"Daemon dispatch failed, falling back to standalone runner: {exc}" + f"Daemon dispatch failed: {exc}" ) + return { + "trace_id": trace_id, + "status": "failed", + "error": f"Daemon dispatch error: {exc}", + "message": "Failed to dispatch task to Artemis Daemon.", + } # 4. Standalone Fallback: Resolve project root, python executable, and background runner module project_root = env_utils.get_project_root() diff --git a/mcp_server/utils/trace_store.py b/mcp_server/utils/trace_store.py index 5cb5446..49b914b 100644 --- a/mcp_server/utils/trace_store.py +++ b/mcp_server/utils/trace_store.py @@ -38,6 +38,21 @@ def get_status_path(trace_id: str) -> str: return os.path.join(get_trace_dir(trace_id), "status.json") +def get_trace_notes_dir(trace_id: str) -> str: + """Returns the absolute path to the notes directory for a given trace_id.""" + return os.path.join(get_trace_dir(trace_id), "notes") + + +def get_trace_stdout_log_path(trace_id: str) -> str: + """Returns the absolute path to the stdout.log file for a given trace_id.""" + return os.path.join(get_trace_dir(trace_id), "stdout.log") + + +def get_trace_stderr_log_path(trace_id: str) -> str: + """Returns the absolute path to the stderr.log file for a given trace_id.""" + return os.path.join(get_trace_dir(trace_id), "stderr.log") + + def init_trace( trace_id: str, task_desc: str, diff --git a/tests/e2e/test_multi_port_end_to_end.py b/tests/e2e/test_multi_port_end_to_end.py new file mode 100644 index 0000000..a76a441 --- /dev/null +++ b/tests/e2e/test_multi_port_end_to_end.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. + +"""Comprehensive End-to-End Test for Artemis Multi-Port System. + +Verifies end-to-end functionality across all entry points and ports: +1. Configuration & Default Model (gemini-3.5-flash-lite) +2. Daemon Server API & Multi-Device Discovery (/api/devices, /api/status) +3. Multi-Device Detection (Physical phone + Android Emulator) +4. MCP Server Tools (mobile_get_device_state, mobile_run_task, mobile_manage_task) +5. Python SDK (ArtemisClient with device targeting & concurrency modes) +6. CLI Entry Points (artemis run, batch, status) +7. Device Execution Lock Isolation (Multi-device concurrent execution) +""" + +import asyncio +import json +import os +import subprocess +import sys +import time +import urllib.request +from pathlib import Path +import pytest + +from artemis.config import ( + DEFAULT_MODEL, + ROOT_DIR, + Settings, + settings, + get_config_path, +) +from artemis.runtime.daemon_client import ( + ensure_daemon_running, + get_daemon_session, + get_daemon_status, + is_daemon_running, + submit_task_to_daemon, + stop_task_on_daemon, +) +from artemis.runtime.device_lock import DeviceExecutionLock +from artemis.runtime.device_pool import device_pool +from artemis.interfaces.sdk.client import ArtemisClient, ConcurrencyMode +from mcp_server.tools.device_state import mobile_get_device_state +from mcp_server.tools.task_runner import mobile_run_task +from mcp_server.tools.task_manager import mobile_manage_task +from mcp_server.utils import trace_store + + +@pytest.fixture(scope="module", autouse=True) +def ensure_daemon(): + """Ensure Artemis Daemon is running for E2E tests.""" + assert ensure_daemon_running(wait_ready=True), "Artemis Daemon failed to start" + yield + + +def test_01_default_model_configured(): + """Verify default model configuration loads properly across config and router.""" + assert DEFAULT_MODEL is not None + cfg_path = get_config_path("artemis.jsonc") + assert cfg_path.exists() + content = cfg_path.read_text(encoding="utf-8") + assert "default" in content + + # Check Settings pydantic model + s = Settings() + assert s.ARTEMIS_DEFAULT_MODEL is not None + + +def test_02_daemon_device_discovery_sees_phone_and_emulator(): + """Verify Daemon API /api/devices detects both the physical phone and emulator.""" + status = get_daemon_status() + assert status is not None + assert status.get("status") in ("idle", "running", "ok") + if "model_info" in status and status["model_info"]: + assert status["model_info"]["id"] is not None + + req = urllib.request.Request("http://127.0.0.1:8000/api/devices") + with urllib.request.urlopen(req, timeout=5.0) as resp: + assert resp.status == 200 + data = json.loads(resp.read().decode("utf-8")) + devices = data.get("devices", []) + serials = [d["serial"] for d in devices] + + # Must detect both devices + assert any(s.startswith("emulator") for s in serials), f"Emulator not found in {serials}" + assert any(not s.startswith("emulator") for s in serials), f"Physical phone not found in {serials}" + + # Check is_emulator flag classification + for d in devices: + if d["serial"].startswith("emulator"): + assert d["is_emulator"] is True + else: + assert d["is_emulator"] is False + + +@pytest.mark.asyncio +async def test_03_mcp_device_state_inspects_both_devices(): + """Verify MCP mobile_get_device_state reads element hierarchies from both devices.""" + devices = device_pool.list_devices() + emulator_serial = next((d.serial for d in devices if d.is_emulator), None) + phone_serial = next((d.serial for d in devices if not d.is_emulator), None) + + assert emulator_serial, f"No emulator connected in {devices}" + assert phone_serial, f"No physical phone connected in {devices}" + + # Inspect emulator state + emu_state = await mobile_get_device_state(view_type="hierarchy", device_serial=emulator_serial) + assert isinstance(emu_state, str) + assert len(emu_state) > 0 + assert "Bounds:" in emu_state or "Text:" in emu_state or len(emu_state.strip()) > 0 + + # Inspect physical phone state + phone_state = await mobile_get_device_state(view_type="hierarchy", device_serial=phone_serial) + assert isinstance(phone_state, str) + assert len(phone_state) > 0 + assert "Bounds:" in phone_state or "Text:" in phone_state or len(phone_state.strip()) > 0 + + +def test_04_mcp_run_and_manage_task_via_daemon(): + """Verify MCP mobile_run_task dispatches to Daemon and mobile_manage_task tracks it.""" + devices = device_pool.list_devices() + emulator_serial = next((d.serial for d in devices if d.is_emulator), None) + assert emulator_serial, "No emulator connected" + + res = mobile_run_task( + task_desc="E2E Test task from MCP port", + conversation_id="e2e-conv-mcp-1", + model="Flash", + device_serial=emulator_serial, + ) + assert "trace_id" in res + trace_id = res["trace_id"] + assert res["status"] in ("started", "queued", "running") + + # Verify tracking state via mobile_manage_task + status_res = mobile_manage_task(action="status", trace_id=trace_id) + assert status_res["trace_id"] == trace_id + assert "status" in status_res + assert status_res.get("device_serial") == emulator_serial + + # Cleanup: stop the test task on Daemon and release locks + stop_task_on_daemon(session_id=trace_id) + time.sleep(1.5) + DeviceExecutionLock.cleanup_stale_locks() + + +def test_05_sdk_client_targeting_and_concurrency(): + """Verify Python SDK ArtemisClient correctly targets specific devices.""" + devices = device_pool.list_devices() + emulator_serial = next((d.serial for d in devices if d.is_emulator), None) + phone_serial = next((d.serial for d in devices if not d.is_emulator), None) + + assert emulator_serial, "Emulator serial required" + assert phone_serial, "Phone serial required" + + client_emu = ArtemisClient(device_id=emulator_serial, concurrency_mode="per_device") + assert client_emu.device_serial == emulator_serial + assert client_emu.concurrency_mode == "per_device" + + client_phone = ArtemisClient(device_id=phone_serial, concurrency_mode="per_device") + assert client_phone.device_serial == phone_serial + + # Verify multi-device lock independence + DeviceExecutionLock.cleanup_stale_locks() + lock_emu = DeviceExecutionLock(device_id=emulator_serial) + lock_phone = DeviceExecutionLock(device_id=phone_serial) + + # Both locks can be acquired simultaneously because device serials differ + acq1 = lock_emu.acquire(timeout=5.0) + assert acq1 is True, "Failed to acquire emulator lock" + try: + acq2 = lock_phone.acquire(timeout=5.0) + assert acq2 is True, "Failed to acquire phone lock concurrently (multi-device parallel mode)" + lock_phone.release() + finally: + lock_emu.release() + + +def test_06_cli_daemon_and_standalone_options(): + """Verify CLI commands properly expose and route through Daemon or standalone.""" + # artemis status + res = subprocess.run([sys.executable, "-m", "artemis.main", "status"], capture_output=True, text=True) + assert res.returncode == 0 + assert "online" in res.stdout.lower() or "running" in res.stdout.lower() + + # artemis run --help exposes --standalone and --device + res_run = subprocess.run([sys.executable, "-m", "artemis.main", "run", "--help"], capture_output=True, text=True) + assert res_run.returncode == 0 + assert "--standalone" in res_run.stdout + assert "--device" in res_run.stdout + + # artemis batch --help exposes --standalone + res_batch = subprocess.run([sys.executable, "-m", "artemis.main", "batch", "--help"], capture_output=True, text=True) + assert res_batch.returncode == 0 + assert "--standalone" in res_batch.stdout + + +def test_07_daemon_session_details_api(): + """Verify Daemon GET /api/sessions/{session_id} API endpoint.""" + res = submit_task_to_daemon( + goal="E2E session query test", + profile="flash", + ingress="e2e_test", + ) + assert "session_id" in res + session_id = res["session_id"] + + # Poll session details + session_data = get_daemon_session(session_id) + assert session_data is not None + assert session_data["session_id"] == session_id + assert session_data["initial_goal"] == "E2E session query test" + + # Clean up test task + stop_task_on_daemon(session_id=session_id) diff --git a/tests/unit/admin_console/test_task_queue_service.py b/tests/unit/admin_console/test_task_queue_service.py index 552fe9d..d444865 100644 --- a/tests/unit/admin_console/test_task_queue_service.py +++ b/tests/unit/admin_console/test_task_queue_service.py @@ -52,9 +52,26 @@ def clean_state(tmp_path, monkeypatch): state.queue_items.clear() state.active_connections.clear() state.cancelled_session_ids.clear() + state.active_session_id = None + state.current_process = None + state.current_goal = None + state.current_profile = None + state.was_stopped_manually = False + if hasattr(state, "recent_submissions"): + state.recent_submissions.clear() if state.worker_task and not state.worker_task.done(): state.worker_task.cancel() state.worker_task = None + from artemis.config.paths import get_temp_dir + from artemis.runtime.device_lock import DeviceExecutionLock + DeviceExecutionLock.cleanup_stale_locks() + queue_dir = get_temp_dir("device-locks") / "artemis-global-device.queue" + if queue_dir.exists(): + for p in queue_dir.glob("*.wait"): + try: + p.unlink(missing_ok=True) + except OSError: + pass def test_paused_error_reads_persisted_reason(tmp_path): @@ -612,7 +629,10 @@ def test_darwin_terminate_process_tree_preserves_direct_child_for_asyncio(): @pytest.mark.asyncio async def test_enqueue_tasks_unified_ingress(): """Verify enqueue_tasks correctly propagates ingress, custom session_id, and conversation_id.""" - with patch.object(TaskQueueService, "ensure_worker_running"): + with ( + patch.object(TaskQueueService, "ensure_worker_running"), + patch("apps.admin_console.services.task_queue_service.DeviceExecutionLock.reserve", return_value="mock-ticket-unified"), + ): res = await task_queue_service.enqueue_tasks( ["Test unified goal"], profile="flash", @@ -801,4 +821,44 @@ def test_stop_tasks_queued_item_by_session_id(): assert update_status.call_args[0][1] == "cancelled" +@pytest.mark.asyncio +async def test_enqueue_tasks_deduplicates_by_session_id(): + state.queue_items = [] + state.active_session_id = None + + with ( + patch("apps.admin_console.services.task_queue_service.session_repo"), + patch("apps.admin_console.services.task_queue_service.DeviceExecutionLock.reserve", return_value="ticket-123"), + ): + res1 = await task_queue_service.enqueue_tasks(["Task goal"], session_id="sid-dedup-1") + assert len(state.queue_items) == 1 + assert res1["enqueued_count"] == 1 + + # Second submission with same session_id must not enqueue a duplicate + res2 = await task_queue_service.enqueue_tasks(["Task goal again"], session_id="sid-dedup-1") + assert len(state.queue_items) == 1 + assert res2["enqueued_count"] == 0 + assert res2["tasks"][0]["session_id"] == "sid-dedup-1" + + +@pytest.mark.asyncio +async def test_enqueue_tasks_debounces_rapid_identical_submissions(): + state.queue_items = [] + state.active_session_id = None + + with ( + patch("apps.admin_console.services.task_queue_service.session_repo"), + patch("apps.admin_console.services.task_queue_service.DeviceExecutionLock.reserve", return_value="ticket-456"), + ): + res1 = await task_queue_service.enqueue_tasks(["Rapid duplicate goal"], device_serial="dev-1") + assert len(state.queue_items) == 1 + assert res1["enqueued_count"] == 1 + + # Immediate second submission (within 1s) must debounce + res2 = await task_queue_service.enqueue_tasks(["Rapid duplicate goal"], device_serial="dev-1") + assert len(state.queue_items) == 1 + assert res2["enqueued_count"] == 0 + + + diff --git a/tests/unit/data_engine/test_engine.py b/tests/unit/data_engine/test_engine.py index 4cddccd..141926c 100644 --- a/tests/unit/data_engine/test_engine.py +++ b/tests/unit/data_engine/test_engine.py @@ -270,6 +270,8 @@ def test_update_step_execution_result_suppresses_identical_post_screenshot(tmp_p pre_screenshot_bytes=image_bytes, summary="Initial step", ) + for t in list(engine._pending_threads): + t.join() step_record = engine.storage.get_steps(engine.current_session_id)[0] assert step_record.post_image_name is None @@ -293,3 +295,34 @@ def test_update_step_execution_result_suppresses_identical_post_screenshot(tmp_p updated_step_distinct = engine.storage.get_steps(engine.current_session_id)[0] assert updated_step_distinct.post_image_name == "distinct_new_hash_789" + + +def test_resumed_session_monotonic_step_numbering(tmp_path): + mock_ctx = MagicMock(spec=ArtemisContext) + mock_execution_setup = MagicMock() + mock_execution_setup.traces_path = str(tmp_path) + mock_ctx.execution_setup = mock_execution_setup + mock_ctx.device = None + + # First runner records Step 1 and Step 2 + engine1 = DataEngine(mock_ctx) + sid = engine1.start_session("Monotonic test") + engine1.record_step(summary="Step 1 action") + engine1.record_step(summary="Step 2 action") + for t in list(engine1._pending_threads): + t.join() + + # Second runner starts or resumes the same session + engine2 = DataEngine(mock_ctx) + engine2.start_session("Resumed goal", session_id=sid) + assert engine2.current_step_number == 2 + + # Recording a new step must be Step 3, strictly monotonic + engine2.record_step(summary="Step 3 action") + for t in list(engine2._pending_threads): + t.join() + + steps = engine2.storage.get_steps(sid) + assert len(steps) == 3 + assert [s.step_number for s in steps] == [1, 2, 3] + diff --git a/tests/unit/mcp/test_mcp_tools.py b/tests/unit/mcp/test_mcp_tools.py index f13636a..28c3c02 100644 --- a/tests/unit/mcp/test_mcp_tools.py +++ b/tests/unit/mcp/test_mcp_tools.py @@ -38,6 +38,7 @@ from mcp_server.utils import trace_store def temp_trace_env(monkeypatch): temp_dir = tempfile.mkdtemp() monkeypatch.setattr(trace_store, "TRACES_DIR", temp_dir) + monkeypatch.setenv("ARTEMIS_STANDALONE", "1") yield temp_dir shutil.rmtree(temp_dir, ignore_errors=True) @@ -151,6 +152,33 @@ def test_mobile_run_task_with_device_serial(temp_trace_env): assert status["device_serial"] == "pixel-11-pro-001" +def test_mobile_run_task_dispatched_to_daemon(temp_trace_env, monkeypatch): + monkeypatch.delenv("ARTEMIS_STANDALONE", raising=False) + with ( + patch( + "artemis.runtime.ensure_daemon_running", + return_value=(True, "http://127.0.0.1:8000"), + ), + patch( + "artemis.runtime.submit_task_to_daemon", + return_value={"status": "started", "tasks": [{"session_id": "daemon-sid-1"}]}, + ), + patch("mcp_server.tools.task_runner.subprocess.Popen") as popen, + ): + result = mobile_run_task( + task_desc="Open Settings via Daemon", + conversation_id="conv-daemon", + model="Flash", + ) + + popen.assert_not_called() + assert result["trace_id"] == "daemon-sid-1" + assert result["status"] == "running" + assert "enqueued via unified Artemis Daemon" in result["message"] + assert "stdout_log" in result + assert "stderr_log" in result + + def test_mobile_manage_task_unknown_trace(temp_trace_env): res = mobile_manage_task(action="status", trace_id="non-existent-trace") assert res["status"] == "unknown" @@ -325,3 +353,52 @@ async def test_mcp_server_auto_registers_tools(): "mobile_get_device_state", "mobile_inspect_trace", }.issubset(tool_names) + + +def test_mobile_manage_task_syncs_terminal_status_from_db(temp_trace_env): + import sqlite3 + trace_id = str(uuid.uuid4()) + trace_store.init_trace(trace_id, "Running task to sync", "Flash", "conv-sync") + + db_path = os.path.join(temp_trace_env, "data_engine.db") + conn = sqlite3.connect(db_path) + cur = conn.cursor() + cur.execute( + """CREATE TABLE IF NOT EXISTS sessions ( + session_id TEXT PRIMARY KEY, + status TEXT, + pid INTEGER, + start_time REAL, + end_time REAL, + device_info TEXT + )""" + ) + cur.execute( + "INSERT INTO sessions (session_id, status, pid, start_time) VALUES (?, ?, ?, ?)", + (trace_id, "completed", 54321, 1000.0), + ) + conn.commit() + conn.close() + + res = mobile_manage_task(action="status", trace_id=trace_id) + assert res["status"] == "completed" + # Ensure status.json was persisted + saved_st = trace_store.read_status(trace_id) + assert saved_st["status"] == "completed" + assert saved_st["pid"] == 54321 + + +def test_mobile_manage_task_stop_via_daemon(temp_trace_env, monkeypatch): + monkeypatch.delenv("ARTEMIS_STANDALONE", raising=False) + trace_id = str(uuid.uuid4()) + trace_store.init_trace(trace_id, "Daemon task to stop", "Flash", "conv-stop") + + with ( + patch("artemis.runtime.is_daemon_running", return_value=True), + patch("artemis.runtime.stop_task_on_daemon", return_value=True) as mock_stop, + ): + res = mobile_manage_task(action="stop", trace_id=trace_id) + mock_stop.assert_called_once_with(trace_id) + assert res["status"] == "cancelled" + assert trace_store.read_status(trace_id)["status"] == "cancelled" + diff --git a/tests/unit/mcp/test_trace_store.py b/tests/unit/mcp/test_trace_store.py index ba2b172..f44a849 100644 --- a/tests/unit/mcp/test_trace_store.py +++ b/tests/unit/mcp/test_trace_store.py @@ -75,3 +75,11 @@ def test_update_trace_status(temp_trace_env): def test_read_nonexistent_status(temp_trace_env): non_existent = str(uuid.uuid4()) assert trace_store.read_status(non_existent) is None + + +def test_trace_paths(temp_trace_env): + trace_id = "test-paths-id" + trace_dir = trace_store.get_trace_dir(trace_id) + assert trace_store.get_trace_notes_dir(trace_id) == f"{trace_dir}/notes" + assert trace_store.get_trace_stdout_log_path(trace_id) == f"{trace_dir}/stdout.log" + assert trace_store.get_trace_stderr_log_path(trace_id) == f"{trace_dir}/stderr.log" diff --git a/tests/unit/runtime/test_daemon_client.py b/tests/unit/runtime/test_daemon_client.py index 04b34d1..658e536 100644 --- a/tests/unit/runtime/test_daemon_client.py +++ b/tests/unit/runtime/test_daemon_client.py @@ -116,3 +116,22 @@ def test_submit_task_to_daemon(): assert data["ingress"] == "mcp" assert data["conversation_id"] == "conv-123" + +def test_stop_task_on_daemon(): + from artemis.runtime.daemon_client import stop_task_on_daemon + + mock_resp = MagicMock() + mock_resp.status = 200 + mock_resp.read.return_value = b'{"status": "stopped", "session_id": "test-sess-stop"}' + mock_resp.__enter__.return_value = mock_resp + + with patch("urllib.request.urlopen", return_value=mock_resp) as mock_urlopen: + assert stop_task_on_daemon("test-sess-stop") is True + req = mock_urlopen.call_args[0][0] + data = json.loads(req.data.decode("utf-8")) + assert data["session_id"] == "test-sess-stop" + + mock_resp.read.return_value = b'{"status": "no_running_task"}' + with patch("urllib.request.urlopen", return_value=mock_resp): + assert stop_task_on_daemon("non-existent-sess") is False + diff --git a/tests/unit/test_config_package.py b/tests/unit/test_config_package.py index 7d157ca..08d5a9b 100644 --- a/tests/unit/test_config_package.py +++ b/tests/unit/test_config_package.py @@ -588,3 +588,60 @@ def test_agent_config_environment_variable_overrides(monkeypatch): assert cfg.outputter.enabled is False assert cfg.video_analyzer.enable_ledger is False assert cfg.pro.video_analyzer.enable_ledger is False + + +def test_consolidated_workspace_and_admin_paths(): + """Verify consolidated workspace and admin paths exported from artemis.config.""" + from artemis.config import ( + DB_PATH, + IMAGES_DIR, + PAUSE_FILE, + REPLAY_BASE_DIR, + ROOT_DIR, + TEST_DATA_DIR, + TEST_OUTPUTS_DIR, + TRACES_PATH, + WORKSPACE_ROOT, + get_images_dir, + get_pause_file, + get_replay_dir, + get_test_data_dir, + get_test_outputs_dir, + ) + + assert WORKSPACE_ROOT == ROOT_DIR + assert PAUSE_FILE == get_pause_file() + assert PAUSE_FILE.name == ".artemis_paused" + assert REPLAY_BASE_DIR == get_replay_dir() + assert TEST_DATA_DIR == get_test_data_dir() + assert TEST_OUTPUTS_DIR == get_test_outputs_dir() + assert IMAGES_DIR == get_images_dir() + assert DB_PATH.name == "data_engine.db" + assert TRACES_PATH.name == "traces" + + +def test_admin_console_config_facade_backward_compatibility(): + """Verify apps.admin_console.core.config continues to re-export consolidated paths.""" + from apps.admin_console.core.config import ( + DB_PATH, + IMAGES_DIR, + PAUSE_FILE, + REPLAY_BASE_DIR, + TEST_DATA_DIR, + TEST_OUTPUTS_DIR, + TRACES_PATH, + WORKSPACE_ROOT, + init_ls_address, + ) + import artemis.config as ac + + assert WORKSPACE_ROOT == ac.WORKSPACE_ROOT + assert PAUSE_FILE == ac.PAUSE_FILE + assert REPLAY_BASE_DIR == ac.REPLAY_BASE_DIR + assert TEST_DATA_DIR == ac.TEST_DATA_DIR + assert TEST_OUTPUTS_DIR == ac.TEST_OUTPUTS_DIR + assert IMAGES_DIR == ac.IMAGES_DIR + assert DB_PATH == ac.DB_PATH + assert TRACES_PATH == ac.TRACES_PATH + assert callable(init_ls_address) +