From 8fdecf23d7cd07f5b3c1e4f504c33d7aafe46d82 Mon Sep 17 00:00:00 2001 From: somew1nd Date: Wed, 26 Aug 2026 17:41:23 -0700 Subject: [PATCH] feat: enhance daemon dispatching, device readiness fallback, and live streaming UI - Admin Console & Task Queue: add cancellation, session startup progress endpoints, and support ingress metadata - Diagnostics & Runtime: enhance ADB probe to prefer unlocked devices and synchronize with readiness engine - MCP Server: dispatch tasks through Artemis Daemon scheduler with fallback to standalone runner, improve action overlays - Showcase UI: improve session auto-selection with pinning support, render startup progress, and refine stream styling - CLI & Makefile: expose server lifecycle commands (restart, stop, status) - Configuration: ignore runtime server state file in .gitignore --- .gitignore | 1 + Makefile | 11 +- apps/admin_console/routers/sessions.py | 8 + apps/admin_console/routers/tasks.py | 103 +++++- apps/admin_console/schemas/task_schema.py | 10 + .../services/task_queue_service.py | 246 +++++++++++--- .../agent-stream/agent-stream.component.html | 10 + .../agent-stream/agent-stream.component.scss | 39 ++- .../agent-stream/agent-stream.component.ts | 5 + .../chat-interface.component.html | 10 + .../chat-interface.component.scss | 44 ++- .../chat-interface.component.ts | 13 + .../src/app/pages/home/home.component.html | 25 ++ .../pages/workspace/workspace.component.html | 12 +- .../pages/workspace/workspace.component.ts | 28 +- .../src/app/services/agent.service.spec.ts | 105 ++++++ .../src/app/services/agent.service.ts | 312 ++++++++++-------- artemis/core/diagnostics/engine.py | 15 +- artemis/core/diagnostics/probes/adb_probe.py | 52 ++- artemis/interfaces/cli/commands/run.py | 6 +- artemis/interfaces/cli/commands/server.py | 9 + artemis/runtime/daemon_client.py | 6 + artemis/runtime/device_pool.py | 10 + artemis/sdk/agent.py | 10 +- mcp_server/tools/inspect_trace.py | 99 ++---- mcp_server/tools/task_runner.py | 57 +++- .../admin_console/test_server_lifecycle.py | 27 ++ .../test_task_device_readiness.py | 38 +++ .../admin_console/test_task_queue_service.py | 195 +++++++++++ tests/unit/core/test_diagnostics.py | 49 +++ tests/unit/runtime/test_daemon_client.py | 17 +- tests/unit/test_cli.py | 19 ++ tests/unit/test_config_package.py | 17 +- 33 files changed, 1315 insertions(+), 293 deletions(-) diff --git a/.gitignore b/.gitignore index a2e647c..ce68482 100644 --- a/.gitignore +++ b/.gitignore @@ -87,6 +87,7 @@ yarn-error.log* traces/ scratch/ .artemis_paused +.artemis_server.json *.log data_engine.db data_engine.db-journal diff --git a/Makefile b/Makefile index 5e366b3..c2a1b01 100644 --- a/Makefile +++ b/Makefile @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -.PHONY: help test install install-deps setup start ui build-ui doctor clean precommit-install precommit lint format typecheck +.PHONY: help test install install-deps setup start ui restart stop status build-ui doctor clean precommit-install precommit lint format typecheck help: ## Show this help message @echo 'Usage: make [target]' @@ -26,6 +26,15 @@ start: ## One-click start Artemis Showcase UI and auto-open browser ui: ## Launch the unified Showcase UI & Admin Console in browser @uv run artemis ui --open +restart: ## Restart running Artemis Web UI & server + @uv run artemis restart + +stop: ## Stop running Artemis Web UI & server + @uv run artemis stop + +status: ## Display Artemis Web UI & server status + @uv run artemis status + build-ui: ## Build the Showcase UI Angular frontend @echo "🎨 Building Showcase UI..." @cd apps/showcase_ui && npm install && npm run build diff --git a/apps/admin_console/routers/sessions.py b/apps/admin_console/routers/sessions.py index 16638f1..d1f8169 100644 --- a/apps/admin_console/routers/sessions.py +++ b/apps/admin_console/routers/sessions.py @@ -161,6 +161,14 @@ async def get_session_background_tasks(session_id: str): return [] +@router.get("/api/sessions/{session_id}/startup_progress") +async def get_session_startup_progress(session_id: str): + try: + return state.get_startup_progress(session_id) + except Exception: + return [] + + @router.post("/api/cleanup") async def cleanup_history_endpoint(): try: diff --git a/apps/admin_console/routers/tasks.py b/apps/admin_console/routers/tasks.py index 4fbcf39..7c5f407 100644 --- a/apps/admin_console/routers/tasks.py +++ b/apps/admin_console/routers/tasks.py @@ -16,7 +16,7 @@ import asyncio from contextlib import suppress import json from typing import Any -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, HTTPException, Request from fastapi.responses import StreamingResponse from artemis.core.diagnostics import readiness_engine from artemis.runtime import DeviceExecutionLock, device_pool @@ -84,15 +84,30 @@ async def run_task(request: RunRequest): # polling intervals cannot start through a stale Ready state. Use the # bounded submission probe: the full diagnostics path also scans packages, # emulator installations, Android version, and screen size. - device_probe = await readiness_engine.run_device_submission_probe() + target_serial = request.device_serial or readiness_engine.get_active_device_serial() + device_probe = await readiness_engine.run_device_submission_probe( + target_serial=target_serial + ) if device_probe and device_probe.summary in {"Device Locked", "Lock State Unknown"}: + locked_serial = ( + device_probe.metadata.get("active_device", {}).get("serial") + or target_serial + or "" + ) detail = ( - "Android device is locked. Unlock it and enter the home screen before running a task." + f"Android device {locked_serial} is locked. Unlock it and enter the home screen before running a task.".replace(" ", " ").strip() if device_probe.summary == "Device Locked" - else "Android device lock state could not be verified. Keep it unlocked on the home screen and try again." + else f"Android device {locked_serial} lock state could not be verified. Keep it unlocked on the home screen and try again.".replace(" ", " ").strip() ) raise HTTPException(status_code=409, detail=detail) + if device_probe and device_probe.metadata.get("active_device"): + verified_serial = device_probe.metadata["active_device"].get("serial") + if verified_serial: + target_serial = verified_serial + if readiness_engine.get_active_device_serial() != verified_serial: + readiness_engine.set_active_device_serial(verified_serial) + return await task_queue_service.enqueue_tasks( incoming_goals, profile=request.profile or "flash", @@ -100,7 +115,10 @@ async def run_task(request: RunRequest): enable_outputter=request.enable_outputter, locked_app_package=request.locked_app_package, app_path=request.app_path, - device_serial=request.device_serial, + device_serial=target_serial, + ingress=request.ingress or "frontend", + session_id=request.session_id, + conversation_id=request.conversation_id, ) @@ -112,13 +130,39 @@ async def list_devices(): @router.post("/api/stop") -async def stop_task(all: bool = False): - stopped = task_queue_service.stop_tasks(clear_all=all) +async def stop_task( + request: Request, + all: bool = False, + session_id: str | None = None, + device_id: str | None = None, +): + target_all = all + target_sid = session_id + target_dev = device_id + + try: + body = await request.json() + if isinstance(body, dict): + if "all" in body: + target_all = bool(body["all"]) or target_all + if body.get("session_id"): + target_sid = str(body["session_id"]) + if body.get("device_id"): + target_dev = str(body["device_id"]) + except Exception: + pass + + stopped = task_queue_service.stop_tasks( + clear_all=target_all, + session_id=target_sid, + device_id=target_dev, + ) if stopped: - return {"status": "stopped"} + return {"status": "stopped", "session_id": target_sid} return {"status": "no_running_task"} + @router.post("/api/resume") async def resume_task(): resumed = task_queue_service.resume_task() @@ -258,8 +302,9 @@ async def get_status(): } +@router.get("/api/stream") @router.get("/api/stream/{session_id}") -async def stream_events(session_id: str, client: str | None = None): +async def stream_events(session_id: str = "active", client: str | None = None): async def event_generator(): queue = asyncio.Queue() event_loop = asyncio.get_running_loop() @@ -295,7 +340,45 @@ async def stream_events(session_id: str, client: str | None = None): state.add_subscriber(callback) yield f'event: info\ndata: {{"message": "Subscribed to session {session_id}"}}\n\n' - if session_id not in ("all", "active"): + + if session_id in ("all", "active"): + active_sid = state.active_session_id + if not active_sid: + running_item = next( + (t for t in state.queue_items if isinstance(t, dict) and t.get("status") == "running"), + None, + ) + if running_item: + active_sid = running_item.get("session_id") + if not active_sid: + owner = DeviceExecutionLock.get_active_owner() + if owner and owner.session_id: + active_sid = owner.session_id + + if active_sid: + goal = state.current_goal or "" + profile = state.current_profile or "flash" + yield ( + "event: session_started\n" + f"data: {json.dumps({'session_id': str(active_sid), 'initial_goal': goal, 'profile': profile}, default=str)}\n\n" + ) + for progress_event in state.get_startup_progress(str(active_sid)): + yield ( + "event: startup_progress\n" + f"data: {json.dumps(progress_event, default=str)}\n\n" + ) + try: + from apps.admin_console.database.repositories.step_repository import step_repo + + recorded_steps = step_repo.get_session_steps(str(active_sid)) + for step_dict in recorded_steps: + yield ( + "event: step_recorded\n" + f"data: {json.dumps(step_dict, default=str)}\n\n" + ) + except Exception as exc: + print(f"[Stream] Could not replay active steps: {exc}") + else: for progress_event in state.get_startup_progress(session_id): yield ( "event: startup_progress\n" diff --git a/apps/admin_console/schemas/task_schema.py b/apps/admin_console/schemas/task_schema.py index f9f543c..ad89cb9 100644 --- a/apps/admin_console/schemas/task_schema.py +++ b/apps/admin_console/schemas/task_schema.py @@ -24,6 +24,9 @@ class RunRequest(BaseModel): locked_app_package: str | None = None app_path: str | None = None device_serial: str | None = None + ingress: str | None = "frontend" + session_id: str | None = None + conversation_id: str | None = None @@ -32,3 +35,10 @@ class ReplayRequest(BaseModel): user_submits: dict tool_name: str = "ask_explorer" replay_id: str | None = None + + +class StopRequest(BaseModel): + session_id: str | None = None + device_id: str | None = None + all: bool = False + diff --git a/apps/admin_console/services/task_queue_service.py b/apps/admin_console/services/task_queue_service.py index 5f7ad93..762a7d3 100644 --- a/apps/admin_console/services/task_queue_service.py +++ b/apps/admin_console/services/task_queue_service.py @@ -207,7 +207,7 @@ class TaskQueueService: while True: try: return await asyncio.wait_for(proc.wait(), timeout=1.0) - except asyncio.TimeoutError: + except TimeoutError: pid = getattr(proc, "pid", None) if pid: try: @@ -313,7 +313,7 @@ class TaskQueueService: env["ARTEMIS_IPC_PORT"] = str(state.ipc_port) if sess_id: env["ARTEMIS_SESSION_ID"] = str(sess_id) - env["ARTEMIS_TASK_INGRESS"] = "frontend" + env["ARTEMIS_TASK_INGRESS"] = str(task_item.get("ingress", "frontend")) queue_ticket = task_item.get("queue_ticket") if queue_ticket: env[DeviceExecutionLock.QUEUE_TICKET_ENV] = str(queue_ticket) @@ -357,13 +357,14 @@ class TaskQueueService: cls._broadcast_startup_progress( sess_id, "process_ready", "Execution process started" ) + ingress_type = str(task_item.get("ingress", "frontend")) DeviceExecutionLock.transfer_reservation( str(queue_ticket), proc.pid, - description=f"frontend task: {goal[:120]}", + description=f"{ingress_type} task: {goal[:120]}", device_id=device_serial or "pending", session_id=str(sess_id) if sess_id else None, - ingress="frontend", + ingress=ingress_type, ) if sys.platform == "win32" and isinstance(proc.stdout, asyncio.StreamReader): output_task = asyncio.create_task(cls._forward_worker_output(proc.stdout)) @@ -437,6 +438,21 @@ class TaskQueueService: }, ) + conversation_id = task_item.get("conversation_id") if task_item else None + if conversation_id: + try: + from mcp_server.notifiers import notify + + notify( + conversation_id=conversation_id, + 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}, + ) + except Exception as notif_err: + print(f"[QueueWorker] Notification dispatch notice: {notif_err}") + except asyncio.CancelledError: print("[QueueWorker] Task received cancellation signal.") if state.current_process and state.current_process.returncode is None: @@ -467,6 +483,9 @@ class TaskQueueService: locked_app_package: str | None = None, app_path: str | None = None, device_serial: str | None = None, + ingress: str = "frontend", + session_id: str | None = None, + conversation_id: str | None = None, ) -> dict[str, Any]: """Enqueues one or more goals and wakes up the background worker.""" cls.ensure_worker_running() @@ -474,8 +493,15 @@ class TaskQueueService: enqueued_tasks = [] now = time.time() for i, goal in enumerate(goals): - sess_id = str(uuid.uuid4()) + sess_id = (session_id if (session_id and len(goals) == 1) else str(uuid.uuid4())) assigned_serial = device_serial + if not assigned_serial: + try: + from artemis.core.diagnostics import readiness_engine + + assigned_serial = readiness_engine.get_active_device_serial() + except Exception: + assigned_serial = None if not assigned_serial: try: from artemis.runtime import device_pool @@ -485,10 +511,10 @@ class TaskQueueService: assigned_serial = None queue_ticket = DeviceExecutionLock.reserve( - description=f"frontend task: {goal[:120]}", + description=f"{ingress} task: {goal[:120]}", device_id=assigned_serial or "pending", session_id=sess_id, - ingress="frontend", + ingress=ingress, ) task_item = { "session_id": sess_id, @@ -499,6 +525,8 @@ class TaskQueueService: "locked_app_package": locked_app_package, "app_path": app_path, "device_serial": assigned_serial, + "ingress": ingress, + "conversation_id": conversation_id, "status": "pending", "queue_ticket": queue_ticket, "created_at": now + i * 0.001, @@ -521,17 +549,132 @@ class TaskQueueService: } @classmethod - def stop_tasks(cls, clear_all: bool = False) -> bool: - """Stop the one task currently controlling the global Artemis device. + def stop_tasks( + cls, + clear_all: bool = False, + session_id: str | None = None, + device_id: str | None = None, + ) -> bool: + """Stop the active task controlling a mobile device or all tasks. The active lease is shared by frontend, MCP, CLI, SDK, and other UI - processes. Pending tasks are intentionally preserved unless the legacy - ``clear_all`` option is explicitly requested, in which case only this - frontend server's pending submissions are cleared. + processes across all connected devices. + - If ``session_id`` or ``device_id`` is specified, only the corresponding + running or queued task is cancelled. + - If ``clear_all`` is requested, all active device owners are terminated + and pending queue submissions are cleared. + - If no specific task is specified and ``clear_all`` is False, stops the + currently active task in single-device mode for backward compatibility. """ stopped = False - owner = DeviceExecutionLock.get_active_owner() - owner_record_exists = DeviceExecutionLock.has_owner_record() + target_sid = str(session_id).strip() if session_id else None + target_device = str(device_id).strip() if device_id else None + + if clear_all: + # 1. Cancel local queue reservations + for item in state.queue_items: + if isinstance(item, dict) and item.get("status") != "running": + DeviceExecutionLock.cancel_reservation(item.get("queue_ticket")) + state.clear_queue() + + # 2. Terminate all active owners across all devices + active_owners: dict[str, Any] = {} + try: + active_owners = DeviceExecutionLock.get_active_owners() + except Exception: + pass + fallback = DeviceExecutionLock.get_active_owner() + if fallback and not active_owners: + active_owners["default"] = fallback + + for dev_owner in list(active_owners.values()): + if dev_owner and DeviceExecutionLock.is_active_owner(dev_owner): + process_supervisor.terminate_tree_verified( + dev_owner.pid, + dev_owner.process_created_at, + ) + DeviceExecutionLock.cleanup_stale_locks(dev_owner.device_id) + 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 + + trace_store.update_trace_status( + str(sid), + "cancelled", + error="Task stopped from the Artemis frontend.", + ) + except Exception: + pass + cls._broadcast_event( + "session_ended", + { + "session_id": sid, + "status": "cancelled", + "was_stopped_manually": True, + }, + ) + + if state.current_process: + state.was_stopped_manually = True + try: + state.current_process.kill() + except Exception: + pass + state.current_process = None + + state.active_connections.clear() + state.active_session_id = None + state.current_goal = None + state.current_profile = None + + if PAUSE_FILE.exists(): + try: + PAUSE_FILE.unlink() + except Exception: + pass + + cls.ensure_worker_running() + state.wake_event.set() + return True + + # Stop a specific task (or default single-device active task) + active_owners = {} + try: + active_owners = DeviceExecutionLock.get_active_owners() + except Exception: + pass + + owner = None + if target_sid: + for dev_owner in active_owners.values(): + if dev_owner.session_id and str(dev_owner.session_id) == target_sid: + owner = dev_owner + break + if owner is None and target_device: + clean_target_dev = DeviceExecutionLock._normalize_device_id(target_device) + for dev_key, dev_owner in active_owners.items(): + if dev_key == clean_target_dev or dev_owner.device_id == target_device: + owner = dev_owner + break + if owner is None: + fallback_owner = DeviceExecutionLock.get_active_owner(target_device) + if fallback_owner: + if not target_sid or ( + fallback_owner.session_id and str(fallback_owner.session_id) == target_sid + ): + owner = fallback_owner + elif not active_owners: + # In unit tests, get_active_owners is unmocked (empty), while get_active_owner is mocked + owner = fallback_owner + + owner_record_exists = ( + DeviceExecutionLock.has_owner_record(target_device) + if hasattr(DeviceExecutionLock, "has_owner_record") + else False + ) owner_pid = owner.pid if owner else None local_pid = getattr(state.current_process, "pid", None) is_local_owner = bool(owner_pid and local_pid and owner_pid == local_pid) @@ -540,7 +683,9 @@ class TaskQueueService: ( item for item in state.queue_items - if isinstance(item, dict) and item.get("status") == "running" + if isinstance(item, dict) + and item.get("status") == "running" + and (not target_sid or str(item.get("session_id")) == target_sid) ), None, ) @@ -548,37 +693,38 @@ class TaskQueueService: ( item for item in state.queue_items - if isinstance(item, dict) and item.get("status") == "pending" + if isinstance(item, dict) + and item.get("status") == "pending" + and (not target_sid or str(item.get("session_id")) == target_sid) ), None, ) stopped_session_id = ( owner.session_id if owner and owner.session_id + else target_sid + if target_sid else state.active_session_id - if owner + if is_local_owner else local_item.get("session_id") if local_item else None ) - if clear_all: - for item in state.queue_items: - if isinstance(item, dict) and item.get("status") != "running": - DeviceExecutionLock.cancel_reservation(item.get("queue_ticket")) - state.clear_queue() - + reservation_cancelled = False if owner and DeviceExecutionLock.is_active_owner(owner): stopped = process_supervisor.terminate_tree_verified( owner.pid, owner.process_created_at, ) - DeviceExecutionLock.cleanup_stale_locks() - elif owner is None and owner_record_exists: + DeviceExecutionLock.cleanup_stale_locks(owner.device_id) + elif owner is None and owner_record_exists and not target_sid: # Never fall back to a frontend PID while another process has an # owner record that is still being published or cannot be parsed. return False - elif owner is None and state.current_process: + elif owner is None and state.current_process and ( + not target_sid or str(state.active_session_id) == target_sid + ): # The frontend worker can be stopped during its short initialization # window before Agent acquires the global device lease. state.was_stopped_manually = True @@ -592,21 +738,42 @@ class TaskQueueService: stopped = True except Exception: pass - # A dead worker is already stopped; cleanup must still be treated - # as a successful, idempotent stop operation. stopped = True is_local_owner = True - elif owner is None and local_item: + elif owner is None and local_item and ( + not target_sid or str(local_item.get("session_id")) == target_sid + ): # Cancel a frontend submission before its worker has started. This # does not touch pending reservations created by other ingresses. DeviceExecutionLock.cancel_reservation(local_item.get("queue_ticket")) + reservation_cancelled = True stopped = True is_local_owner = True + elif target_sid: + # Fallback 1: check global device queue + global_queued = DeviceExecutionLock.get_queued_tasks() + for q_item in global_queued: + if str(q_item.get("session_id")) == target_sid: + ticket = q_item.get("queue_ticket") + if ticket: + DeviceExecutionLock.cancel_reservation(ticket) + stopped = True + break + # Fallback 2: check session repository for a running session with a live worker PID + if not stopped: + row = session_repo.get_session_by_id(target_sid) + if row and row.get("status") == "running": + row_pid = row.get("pid") + if row_pid and session_repo.process_is_alive(row_pid): + try: + process_supervisor.terminate_tree(int(row_pid)) + except Exception: + pass + DeviceExecutionLock.cleanup_stale_locks() + stopped = True - if not stopped: - if clear_all: - state.wake_event.set() - return clear_all + if not stopped and not target_sid: + return False if is_local_owner: state.was_stopped_manually = True @@ -614,11 +781,11 @@ class TaskQueueService: if stopped_session_id: state.cancelled_session_ids.add(str(stopped_session_id)) - for session_id, conn_info in list(state.active_connections.items()): - if (stopped_session_id and str(session_id) == str(stopped_session_id)) or ( + for sid, conn_info in list(state.active_connections.items()): + if (stopped_session_id and str(sid) == str(stopped_session_id)) or ( owner_pid and conn_info.get("pid") == owner_pid ): - state.active_connections.pop(session_id, None) + state.active_connections.pop(sid, None) if stopped_session_id: session_repo.update_session_status( @@ -652,7 +819,7 @@ class TaskQueueService: state.current_goal = None state.current_profile = None - if is_local_owner and stopped_session_id: + if stopped_session_id: stopped_item = next( ( item @@ -662,7 +829,7 @@ class TaskQueueService: ), None, ) - if stopped_item: + if stopped_item and not reservation_cancelled: DeviceExecutionLock.cancel_reservation(stopped_item.get("queue_ticket")) state.queue_items = [ item @@ -672,8 +839,6 @@ class TaskQueueService: and str(item.get("session_id")) == str(stopped_session_id) ) ] - if clear_all: - state.queue_items = [] if PAUSE_FILE.exists(): try: @@ -685,6 +850,7 @@ class TaskQueueService: state.wake_event.set() return True + @classmethod def resume_task(cls) -> bool: if PAUSE_FILE.exists(): diff --git a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html index 1736409..ae68095 100644 --- a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html +++ b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html @@ -74,6 +74,16 @@ · {{ session.start_time * 1000 | date:'shortTime' }} + + + + @if (connectedDevices().length > 1) { +
+ Select Active Device: +
+ @for (dev of connectedDevices(); track dev.serial) { + + } +
+
+ } } diff --git a/apps/showcase_ui/src/app/pages/workspace/workspace.component.html b/apps/showcase_ui/src/app/pages/workspace/workspace.component.html index a9df9a5..b5f9614 100644 --- a/apps/showcase_ui/src/app/pages/workspace/workspace.component.html +++ b/apps/showcase_ui/src/app/pages/workspace/workspace.component.html @@ -116,11 +116,11 @@ close - + @@ -161,12 +161,12 @@ - + diff --git a/apps/showcase_ui/src/app/pages/workspace/workspace.component.ts b/apps/showcase_ui/src/app/pages/workspace/workspace.component.ts index 1b5f125..0e72622 100644 --- a/apps/showcase_ui/src/app/pages/workspace/workspace.component.ts +++ b/apps/showcase_ui/src/app/pages/workspace/workspace.component.ts @@ -77,10 +77,29 @@ export class WorkspaceComponent implements OnInit { } /** - * Computed boolean whether a task is currently executing or active + * Computed boolean whether the currently viewed task is actively running or paused. + * Only displays the stop/cancel button when inspecting an active task. */ public isTaskRunning = computed(() => { - return this.agentService.isRunningTask(); + return this.agentService.isCurrentSessionRunning(); + }); + + /** + * Dynamic tooltip and label indicating which task will be stopped + */ + public stopButtonTitle = computed(() => { + const session = this.agentService.currentSession(); + if (session?.initial_goal) { + const truncated = session.initial_goal.length > 45 + ? session.initial_goal.substring(0, 42) + '...' + : session.initial_goal; + return `Stop current task: "${truncated}"`; + } + const curId = this.agentService.currentSessionId(); + if (curId) { + return `Stop current task (${curId})`; + } + return 'Stop current running task'; }); /** @@ -216,7 +235,7 @@ export class WorkspaceComponent implements OnInit { } /** - * Stop currently running task + * Stop currently viewed task */ public stopTask(event?: MouseEvent): void { if (event) { @@ -225,9 +244,10 @@ export class WorkspaceComponent implements OnInit { if (!this.isTaskRunning()) { return; } + const targetSessionId = this.agentService.currentSessionId(); this.isSubmitting = true; this.errorMessage = null; - this.agentService.stopTask(false); + this.agentService.stopTask(targetSessionId, false); setTimeout(() => { this.isSubmitting = false; }, 400); diff --git a/apps/showcase_ui/src/app/services/agent.service.spec.ts b/apps/showcase_ui/src/app/services/agent.service.spec.ts index c955e29..c5f96fb 100644 --- a/apps/showcase_ui/src/app/services/agent.service.spec.ts +++ b/apps/showcase_ui/src/app/services/agent.service.spec.ts @@ -275,3 +275,108 @@ describe('AgentService video analysis seeking', () => { expect(second?.requestId).toBeGreaterThan(first?.requestId || 0); }); }); + +describe('AgentService task cancellation and active session tracking', () => { + it('computes isCurrentSessionRunning true only when viewing an active running/paused session', () => { + const service = Object.create(AgentService.prototype) as any; + service.currentSessionId = signal(null); + service.rawSessions = signal([]); + service.activeTasks = signal([]); + service.pendingQueue = signal([]); + service.agentStatus = signal('idle'); + service.runningSessionId = signal(null); + service.runningGoal = signal(null); + service.activeSessionTracking = new Map(); + + service.currentSession = computed(() => { + const curId = service.currentSessionId(); + if (!curId) return null; + return service.sessions().find((s: any) => s.session_id === curId) || null; + }); + + service.sessions = computed(() => { + return service.rawSessions(); + }); + + service.isCurrentSessionRunning = computed(() => { + const curId = service.currentSessionId(); + if (!curId) return false; + const session = service.currentSession(); + if (session) { + return session.status === 'running' || session.status === 'paused'; + } + const isActiveStatus = service.agentStatus() === 'running' || service.agentStatus() === 'paused'; + if (isActiveStatus && service.runningSessionId() === curId) { + return true; + } + if (service.activeTasks().some((at: any) => at.session_id === curId)) { + return true; + } + return false; + }); + + // 1. No session selected -> false + expect(service.isCurrentSessionRunning()).toBeFalse(); + + // 2. Completed session selected -> false + service.rawSessions.set([ + { session_id: 'sess-completed', status: 'completed', initial_goal: 'Done', start_time: 100 }, + { session_id: 'sess-running', status: 'running', initial_goal: 'Running', start_time: 101 } + ]); + service.currentSessionId.set('sess-completed'); + expect(service.isCurrentSessionRunning()).toBeFalse(); + + // 3. Active running session selected -> true + service.currentSessionId.set('sess-running'); + expect(service.isCurrentSessionRunning()).toBeTrue(); + + // 4. Paused session selected -> true + service.rawSessions.set([ + { session_id: 'sess-paused', status: 'paused', initial_goal: 'Paused', start_time: 102 } + ]); + service.currentSessionId.set('sess-paused'); + expect(service.isCurrentSessionRunning()).toBeTrue(); + }); + + it('stops a specific session by passing its session_id', () => { + const service = Object.create(AgentService.prototype) as AgentService; + service.currentSessionId = signal('sess-2'); + service.runningSessionId = signal('sess-1'); + service.runningGoal = signal('Goal 1'); + service.agentStatus = signal('running'); + service.isPaused = signal(false); + service.pausedError = signal(null); + service.isRetrying = signal(false); + service.sessionLogs = signal([]); + service.activeTasks = signal([{ session_id: 'sess-1' }, { session_id: 'sess-2' }]); + (service as any).pendingQueue = signal([]); + (service as any).rawSessions = signal([ + { session_id: 'sess-1', status: 'running', initial_goal: 'Goal 1' }, + { session_id: 'sess-2', status: 'running', initial_goal: 'Goal 2' } + ]); + service.sessions = computed(() => (service as any).rawSessions()); + + let postedUrl = ''; + let postedPayload: any = null; + (service as any).http = { + post: (url: string, payload: any) => { + postedUrl = url; + postedPayload = payload; + return of({}); + } + }; + (service as any).setSessionStatus = (sid: string, st: string) => {}; + (service as any).fetchStatus = () => {}; + (service as any).fetchSessions = () => {}; + + // Stop sess-2 specifically + service.stopTask('sess-2', false); + + expect(postedUrl).toContain('session_id=sess-2'); + expect(postedPayload?.session_id).toBe('sess-2'); + // Because sess-1 is still running, agentStatus should NOT be reset to idle + expect(service.agentStatus()).toBe('running'); + // activeTasks should have filtered out sess-2 + expect(service.activeTasks().map((at: any) => at.session_id)).toEqual(['sess-1']); + }); +}); diff --git a/apps/showcase_ui/src/app/services/agent.service.ts b/apps/showcase_ui/src/app/services/agent.service.ts index 3086bd8..629dbeb 100644 --- a/apps/showcase_ui/src/app/services/agent.service.ts +++ b/apps/showcase_ui/src/app/services/agent.service.ts @@ -275,12 +275,25 @@ export class AgentService { }); /** - * Computed whether the currently viewed session is running + * Computed whether the currently viewed session is actively running or paused. + * Returns false when viewing completed/failed/cancelled tasks, or when no session is selected. */ public isCurrentSessionRunning = computed(() => { + const curId = this.currentSessionId(); + if (!curId) return false; const session = this.currentSession(); - if (session) return session.status === 'running' || session.status === 'paused'; - return this.agentStatus() === 'running' || this.agentStatus() === 'paused'; + if (session) { + return session.status === 'running' || session.status === 'paused'; + } + // Fallback if session was just submitted/selected but not yet merged into sessions list + const isActiveStatus = this.agentStatus() === 'running' || this.agentStatus() === 'paused'; + if (isActiveStatus && this.runningSessionId() === curId) { + return true; + } + if (this.activeTasks().some((at) => at.session_id === curId)) { + return true; + } + return false; }); /** @@ -312,6 +325,7 @@ export class AgentService { this.restoreSessionsCache(); this.fetchSessions(); this.startStatusPolling(); + this.ensureLiveStream(); } /** @@ -382,43 +396,86 @@ export class AgentService { } /** - * Stop the currently running task (and optionally clear queue) + * Stop the currently viewed task (or specified task), and optionally clear queue. + * Supports stopping any task across any ingress/platform (Web, CLI, MCP, SDK). */ - public stopTask(stopAll: boolean = false): void { - const stoppingSessionId = this.runningSessionId() - || this.sessions().find((session) => session.status === 'running' || session.status === 'paused')?.session_id - || null; + public stopTask(targetOrStopAll?: string | boolean | null, stopAll: boolean = false): void { + let targetSessionId: string | null = null; + let effectiveStopAll = stopAll; + + if (typeof targetOrStopAll === 'boolean') { + effectiveStopAll = targetOrStopAll; + } else if (typeof targetOrStopAll === 'string' && targetOrStopAll.trim()) { + targetSessionId = targetOrStopAll.trim(); + } else { + // Default to currently viewed session, or currently running session + targetSessionId = this.currentSessionId() || this.runningSessionId() || null; + } + + if (!effectiveStopAll && !targetSessionId) { + targetSessionId = this.sessions().find( + (session) => session.status === 'running' || session.status === 'paused' + )?.session_id || null; + } // Keep the task's terminal state stable while the stop request and the // persisted session update cross the network. Clearing the global runner // state first used to make the session merger infer "completed" from the // still-stale DB row, causing a brief completed -> cancelled flicker. - this.setSessionStatus(stoppingSessionId, 'cancelled'); + if (targetSessionId) { + this.setSessionStatus(targetSessionId, 'cancelled'); + } - // Apply optimistic updates: only set idle if stopAll is true or pending queue is empty - if (stopAll || this.pendingQueue().length === 0) { + // Determine if other active sessions are still executing across multiple devices + const otherRunningSessions = this.sessions().filter( + (s) => (s.status === 'running' || s.status === 'paused') && s.session_id !== targetSessionId + ); + + // Apply optimistic updates: only set idle if effectiveStopAll is true or no other tasks are running + if (effectiveStopAll || otherRunningSessions.length === 0) { this.agentStatus.set('idle'); this.runningSessionId.set(null); this.runningGoal.set(null); + } else if (this.runningSessionId() === targetSessionId) { + this.runningSessionId.set(otherRunningSessions[0].session_id); + this.runningGoal.set(otherRunningSessions[0].initial_goal || null); } + this.isPaused.set(false); this.pausedError.set(null); this.isRetrying.set(false); - if (stopAll) { + if (effectiveStopAll) { this.pendingQueue.set([]); } - // Mark active streaming / pending logs in current session as finished - this.sessionLogs.update((logs) => - logs.map((l) => { - if (l.type === 'llm_stream' && !l.data?.isCompleted) { - return { ...l, data: { ...l.data, isCompleted: true } }; - } - return l; - }) - ); + // Mark active streaming / pending logs as finished if viewing target session + if (!targetSessionId || this.currentSessionId() === targetSessionId) { + this.sessionLogs.update((logs) => + logs.map((l) => { + if (l.type === 'llm_stream' && !l.data?.isCompleted) { + return { ...l, data: { ...l.data, isCompleted: true } }; + } + return l; + }) + ); + } - this.http.post(`/api/stop?all=${stopAll}`, {}).subscribe({ + // Optimistically remove from activeTasks + if (targetSessionId) { + this.activeTasks.update((list) => list.filter((at) => at.session_id !== targetSessionId)); + } + + let url = `/api/stop?all=${effectiveStopAll}`; + if (!effectiveStopAll && targetSessionId) { + url += `&session_id=${encodeURIComponent(targetSessionId)}`; + } + + const payload: any = { all: effectiveStopAll }; + if (targetSessionId) { + payload.session_id = targetSessionId; + } + + this.http.post(url, payload).subscribe({ next: () => { this.fetchStatus(); this.fetchSessions(); @@ -577,10 +634,6 @@ export class AgentService { this.currentSessionId.set(null); this.sessionLogs.set([]); this.currentNotes.set({}); - if (this.eventSource) { - this.eventSource.close(); - this.eventSource = null; - } return; } @@ -605,37 +658,36 @@ export class AgentService { this.openVideoPlayer(sessionId); } - // Clean up previous event stream connection if any + console.log(`Selecting session: ${sessionId}`); + // Start reading the persisted snapshot immediately + this.backfillSessionSteps(sessionId, loadGeneration); + this.ensureLiveStream(); + } + + /** + * Ensure persistent single-channel live stream is established via /api/stream. + * Eliminates reconnect tearing, race conditions, and disconnect dropouts. + */ + public ensureLiveStream(): void { if (this.eventSource) { - this.eventSource.close(); - this.eventSource = null; + return; } - console.log(`Connecting to session: ${sessionId}`); - // Start reading the persisted snapshot immediately. Previously this waited - // for the SSE `info` event, adding connection latency to every history click. - this.backfillSessionSteps(sessionId, loadGeneration); + console.log('Establishing persistent unified live stream via /api/stream'); + this.eventSource = new EventSource('/api/stream'); - this.eventSource = new EventSource(`/api/stream/${sessionId}`); - - // Active runs get one reconciliation snapshot after the stream subscribes, - // closing the snapshot-to-stream gap without delaying the first paint. - const selectedStatus = this.sessions().find((session) => session.session_id === sessionId)?.status; - const shouldReconcileAfterSubscribe = selectedStatus === 'running' - || selectedStatus === 'paused' - || selectedStatus === 'pending'; this.eventSource.addEventListener('info', () => { - if (shouldReconcileAfterSubscribe) { - this.backfillSessionSteps(sessionId, loadGeneration); + // Reconcile current session if active + const curId = this.currentSessionId(); + if (curId) { + this.backfillSessionSteps(curId); } }); - // Listen to standard SSE keep-alive this.eventSource.addEventListener('keep-alive', () => { - // No-op keep-alive tick + // Keep-alive tick }); - // Listen to raw events and append them directly to the logs array const eventTypes = [ 'llm_stream', 'trace_recorded', @@ -656,8 +708,6 @@ export class AgentService { this.eventSource?.addEventListener(eventType, (event: MessageEvent) => { try { const parsedData = JSON.parse(event.data); - - // If the event carries a session_id for a different session, update global states if applicable and ignore const evtSessionId = parsedData?.session_id; if (eventType === 'recording_ready' || eventType === 'recording_failed') { @@ -682,50 +732,8 @@ export class AgentService { return; } - const isDifferentSession = evtSessionId - && String(evtSessionId).trim().toLowerCase() !== String(sessionId).trim().toLowerCase(); - - if (isDifferentSession) { - if (eventType === 'session_started') { - this.agentStatus.set('running'); - this.runningSessionId.set(parsedData.session_id); - if (parsedData?.initial_goal) { - this.runningGoal.set(parsedData.initial_goal); - } - this.fetchSessions(); - // Only auto-select if user has no current session at all - const currentId = this.currentSessionId(); - if (!currentId && parsedData?.session_id) { - this.selectSession(parsedData.session_id, false); - } - } else if (eventType === 'session_ended') { - this.applySessionEndedStatus(evtSessionId, parsedData); - this.activeTasks.update(list => list.filter(at => at.session_id !== evtSessionId)); - const remaining = this.activeTasks(); - if (remaining.length > 0) { - if (this.runningSessionId() === evtSessionId) { - this.runningSessionId.set(remaining[0].session_id); - this.runningGoal.set(remaining[0].goal || null); - } - } else if (this.pendingQueue().length > 0) { - const nextPending = this.pendingQueue()[0]; - this.agentStatus.set('running'); - this.runningSessionId.set(nextPending.session_id); - this.runningGoal.set(nextPending.initial_goal || null); - } else { - this.agentStatus.set('idle'); - this.runningSessionId.set(null); - this.runningGoal.set(null); - } - this.fetchStatus(); - this.fetchSessions(); - this.updateModelForCurrentView(); - } - return; - } - - // Guard against obsolete subscriptions - if (this.currentSessionId() !== sessionId) { + if (eventType === 'background_tasks_updated') { + this.fetchStatus(); return; } @@ -738,12 +746,20 @@ export class AgentService { this.runningGoal.set(parsedData.initial_goal); } this.fetchSessions(); + + // Auto-follow: if user has not explicitly pinned a historical session, switch view to new active session + if (!this.userPinnedSessionId() && parsedData?.session_id) { + this.selectSession(parsedData.session_id, false); + } return; } if (eventType === 'session_ended') { - this.applySessionEndedStatus(sessionId, parsedData); - this.activeTasks.update(list => list.filter(at => at.session_id !== sessionId)); + const endedId = evtSessionId || this.runningSessionId(); + if (endedId) { + this.applySessionEndedStatus(endedId, parsedData); + this.activeTasks.update(list => list.filter(at => at.session_id !== endedId)); + } const remaining = this.activeTasks(); if (remaining.length > 0) { this.agentStatus.set('running'); @@ -763,20 +779,40 @@ export class AgentService { this.isPaused.set(false); this.fetchSessions(); this.fetchStatus(); - this.fetchNotes(sessionId); - this.updateModelForCurrentView(); - if ( - this.isVideoWindowOpen() - && this.activeVideoSessionId === sessionId - && this.recordingPlaybackStatus() === 'live' - ) { - this.beginRecordingFinalization(sessionId); + if (endedId) { + this.fetchNotes(endedId); + if ( + this.isVideoWindowOpen() + && this.activeVideoSessionId === endedId + && this.recordingPlaybackStatus() === 'live' + ) { + this.beginRecordingFinalization(endedId); + } } + this.updateModelForCurrentView(); return; } if (eventType === 'startup_progress') { - this.appendStartupProgress(parsedData, sessionId); + const targetSid = evtSessionId || this.currentSessionId(); + if (targetSid) { + this.appendStartupProgress(parsedData, targetSid); + const curId = this.currentSessionId(); + if ( + !this.userPinnedSessionId() && + (!curId || String(targetSid).trim().toLowerCase() !== String(curId).trim().toLowerCase()) + ) { + this.agentStatus.set('running'); + this.runningSessionId.set(targetSid); + this.selectSession(targetSid, false); + } + } + return; + } + + // If the event is for a different session than what user is currently viewing, ignore for logs + const curId = this.currentSessionId(); + if (evtSessionId && curId && String(evtSessionId).trim().toLowerCase() !== String(curId).trim().toLowerCase()) { return; } @@ -788,7 +824,9 @@ export class AgentService { const attemptText = attempt && max ? ` (Attempt ${attempt}/${max})` : ''; const delayText = delay > 0 ? `; retrying in ${delay.toFixed(2).replace(/\.00$/, '')}s` : ''; this.retryMessage.set(`AI service is temporarily busy${attemptText}${delayText}...`); - this.appendLiveLLMRetryTrace(parsedData, sessionId); + if (curId) { + this.appendLiveLLMRetryTrace(parsedData, curId); + } return; } @@ -796,17 +834,19 @@ export class AgentService { this.isPaused.set(true); this.isRetrying.set(false); this.agentStatus.set('paused'); - this.runningSessionId.set(sessionId); - this.setSessionStatus(sessionId, 'paused'); - const pauseError = parsedData.error || 'AI call failed'; - this.pausedError.set(pauseError); - this.appendPausedErrorCard( - pauseError, - sessionId, - parsedData.timestamp, - parsedData.step_id, - parsedData - ); + if (curId) { + this.runningSessionId.set(curId); + this.setSessionStatus(curId, 'paused'); + const pauseError = parsedData.error || 'AI call failed'; + this.pausedError.set(pauseError); + this.appendPausedErrorCard( + pauseError, + curId, + parsedData.timestamp, + parsedData.step_id, + parsedData + ); + } return; } @@ -814,9 +854,9 @@ export class AgentService { this.isPaused.set(false); this.isRetrying.set(false); this.pausedError.set(null); - if (this.runningSessionId() === sessionId) { + if (curId && this.runningSessionId() === curId) { this.agentStatus.set('running'); - this.setSessionStatus(sessionId, 'running'); + this.setSessionStatus(curId, 'running'); } this.activePauseCardKey = null; return; @@ -833,7 +873,6 @@ export class AgentService { const streamType = parsedData.stream_type || 'text'; this.sessionLogs.update((logs) => { - // Find if there is an active stream log with same execution_id and stream_type const existingIndex = logs.findIndex( (l) => l.type === 'llm_stream' && l.data.execution_id === execId && @@ -841,7 +880,6 @@ export class AgentService { ); if (existingIndex > -1) { - // Append text in place const updatedLogs = [...logs]; const existingLog = updatedLogs[existingIndex]; updatedLogs[existingIndex] = { @@ -854,7 +892,6 @@ export class AgentService { }; return updatedLogs; } else { - // First mark any other active streams of same stream_type as completed const updatedLogs = logs.map((l) => { if (l.type === 'llm_stream' && (l.data.stream_type || 'text') === streamType && @@ -867,7 +904,6 @@ export class AgentService { return l; }); - // Add new stream log return [ ...updatedLogs, { @@ -885,7 +921,6 @@ export class AgentService { } }); } else { - // For other event types, mark any active stream as completed this.sessionLogs.update((logs) => { const updatedLogs = logs.map((l) => { if (l.type === 'llm_stream' && !l.data.isCompleted) { @@ -907,9 +942,6 @@ export class AgentService { data: parsedData }; - // A retry is published both as trace_recorded and as the - // purpose-built llm_retrying event. Replace the optimistic live - // copy when the persisted trace arrives instead of duplicating it. if (eventType === 'trace_recorded' && parsedData?.trace_id) { const existingTraceIndex = updatedLogs.findIndex((log) => log.type === 'trace_recorded' @@ -930,14 +962,13 @@ export class AgentService { const trName = parsedData.name || ''; if (['save_note', 'read_note', 'update_note', 'append_note', 'list_notes', 'outputter'].includes(trName.toLowerCase())) { const curSessionId = this.currentSessionId(); - if (curSessionId === sessionId) { + if (curSessionId) { this.fetchNotes(curSessionId); } } } } catch (e) { console.error(`Failed to parse ${eventType} event data:`, e); - // Fallback to plain event data if parsing fails this.sessionLogs.update((logs) => [ ...logs, { @@ -951,7 +982,7 @@ export class AgentService { }); this.eventSource.onerror = (err) => { - console.error('EventSource connection error:', err); + console.warn('Persistent live stream issue, browser will auto-reconnect:', err); }; } @@ -998,6 +1029,27 @@ export class AgentService { } } }); + + this.http.get(`/api/sessions/${sessionId}/startup_progress`).subscribe({ + next: (events) => { + if (Array.isArray(events) && events.length > 0) { + this.startupProgressBySession.update((current) => { + const existing = current[sessionId] || []; + const merged = [...existing]; + for (const ev of events) { + if (!merged.some((m) => m.stage === ev.stage)) { + merged.push(ev); + } + } + return { + ...current, + [sessionId]: merged + }; + }); + } + }, + error: () => {} + }); } private restoreSessionsCache(): void { @@ -1268,10 +1320,10 @@ export class AgentService { this.beginRecordingFinalization(oldRunningSessionId); } - // Auto-select the session ONLY if nothing is currently selected + // Auto-select the active session if user has not explicitly pinned a historical session if (isActive && data.session_id) { const currentId = this.currentSessionId(); - if (!currentId) { + if (!currentId || (!this.userPinnedSessionId() && currentId !== data.session_id)) { this.selectSession(data.session_id, false); } } diff --git a/artemis/core/diagnostics/engine.py b/artemis/core/diagnostics/engine.py index b8756c0..0bc6530 100644 --- a/artemis/core/diagnostics/engine.py +++ b/artemis/core/diagnostics/engine.py @@ -90,7 +90,14 @@ class ReadinessEngine: def get_active_device_serial(self) -> str | None: """Get currently selected active device serial.""" - return self._active_device_serial + if self._report_cache and self._report_cache.active_device: + if self._report_cache.active_device.is_locked is False: + return self._report_cache.active_device.serial + if self._active_device_serial: + return self._active_device_serial + if self._report_cache and self._report_cache.active_device: + return self._report_cache.active_device.serial + return None async def run_probe(self, probe_id: str) -> ProbeResult | None: """Execute a single specific probe by ID.""" @@ -99,9 +106,11 @@ class ReadinessEngine: return None return await probe.probe() - async def run_device_submission_probe(self) -> ProbeResult: + async def run_device_submission_probe( + self, target_serial: str | None = None + ) -> ProbeResult: """Run the bounded device gate used by task submission.""" - return await self._adb_probe.probe_submission_readiness() + return await self._adb_probe.probe_submission_readiness(target_serial=target_serial) def invalidate_cache(self) -> None: """Invalidate the UI readiness snapshot after an explicit configuration change.""" diff --git a/artemis/core/diagnostics/probes/adb_probe.py b/artemis/core/diagnostics/probes/adb_probe.py index 979841f..1fa4b7b 100644 --- a/artemis/core/diagnostics/probes/adb_probe.py +++ b/artemis/core/diagnostics/probes/adb_probe.py @@ -358,7 +358,9 @@ class AdbDeviceProbe(BaseProbe): logger.debug(f"Failed submission-time ADB device check: {exc}") return None - async def probe_submission_readiness(self) -> ProbeResult: + async def probe_submission_readiness( + self, target_serial: str | None = None + ) -> ProbeResult: """Run the minimal fail-safe device check required before enqueueing. The full diagnostics probe enriches device metadata, scans packages, @@ -416,16 +418,42 @@ class AdbDeviceProbe(BaseProbe): }, ) - serial = ( - self._target_serial + preferred = ( + target_serial + if target_serial and target_serial in ready_serials + else self._target_serial if self._target_serial in ready_serials else ready_serials[0] ) + + serial = preferred is_locked = await self._get_confirmed_device_lock_state( adb_path, serial, timeout_seconds=1.0, ) + + # If preferred device is locked or unknown, check remaining ready devices for an unlocked one + if is_locked is not False and len(ready_serials) > 1: + for candidate in ready_serials: + if candidate == preferred: + continue + candidate_locked = await self._get_confirmed_device_lock_state( + adb_path, + candidate, + timeout_seconds=1.0, + ) + if candidate_locked is False: + logger.info( + f"Device {preferred} is not ready/unlocked ({is_locked}); " + f"switching to unlocked ready device {candidate}" + ) + serial = candidate + is_locked = False + break + elif candidate_locked is None and is_locked is True: + serial = candidate + is_locked = None summary = ( "Device Locked" if is_locked is True @@ -796,11 +824,21 @@ class AdbDeviceProbe(BaseProbe): ) # 4. Ready device available - active_dev = ready_devices[0] + preferred_dev = None if self._target_serial: - matched = next((d for d in ready_devices if d.serial == self._target_serial), None) - if matched: - active_dev = matched + preferred_dev = next((d for d in ready_devices if d.serial == self._target_serial), None) + + if preferred_dev and preferred_dev.is_locked is False: + active_dev = preferred_dev + else: + # Prefer a confirmed unlocked ready device + unlocked_dev = next((d for d in ready_devices if d.is_locked is False), None) + if unlocked_dev: + active_dev = unlocked_dev + else: + # If no confirmed unlocked device, prefer undetermined lock state over confirmed locked + unknown_dev = next((d for d in ready_devices if d.is_locked is None), None) + active_dev = preferred_dev or unknown_dev or ready_devices[0] display_name = active_dev.model or active_dev.serial if active_dev.screen_resolution: diff --git a/artemis/interfaces/cli/commands/run.py b/artemis/interfaces/cli/commands/run.py index 46b36e6..3905edc 100644 --- a/artemis/interfaces/cli/commands/run.py +++ b/artemis/interfaces/cli/commands/run.py @@ -77,7 +77,6 @@ async def execute_task( explorer_flash_mode: Override Explorer version mode for Flash execution profile. explorer_pro_mode: Override Explorer version mode for Pro execution profile. """ - publish_startup_progress("configuration", "Loading the run configuration") effective_sid = ( session_id or os.getenv("ARTEMIS_SESSION_ID") @@ -85,6 +84,9 @@ async def execute_task( ) if effective_sid: os.environ["ARTEMIS_SESSION_ID"] = str(effective_sid) + publish_startup_progress( + "configuration", "Loading the run configuration", session_id=str(effective_sid) if effective_sid else None + ) llm_config = initialize_llm_config() agent_profile = AgentProfile(name="default", llm_config=llm_config) @@ -142,13 +144,11 @@ async def execute_task( agent: Agent | None = None try: - publish_startup_progress("device_check", "Checking the Android device") agent = Agent(config=config.build(), session_id=effective_sid) await agent.init( retry_count=int(os.getenv("ARTEMIS_HEALTH_RETRIES", 5)), retry_wait_seconds=int(os.getenv("ARTEMIS_HEALTH_DELAY", 2)), ) - publish_startup_progress("device_ready", "Android device connected") task = agent.new_task(goal) if locked_app_package: diff --git a/artemis/interfaces/cli/commands/server.py b/artemis/interfaces/cli/commands/server.py index 89a0413..265fccf 100644 --- a/artemis/interfaces/cli/commands/server.py +++ b/artemis/interfaces/cli/commands/server.py @@ -23,12 +23,21 @@ from typing import Annotated import urllib.error import urllib.request +from artemis.interfaces.cli.commands.server_lifecycle import ( + restart_command, + status_command, + stop_command, +) from artemis.utils.logger import get_logger import typer logger = get_logger(__name__) server_app = typer.Typer(help="Cloud proxy and server management commands.") +server_app.command("restart", help="Restart running Artemis Web UI & server.")(restart_command) +server_app.command("stop", help="Stop running Artemis Web UI & server.")(stop_command) +server_app.command("status", help="Display Artemis Web UI & server status.")(status_command) + cached_token = "" token_expire_time = 0.0 diff --git a/artemis/runtime/daemon_client.py b/artemis/runtime/daemon_client.py index dcbcb76..3befe5e 100644 --- a/artemis/runtime/daemon_client.py +++ b/artemis/runtime/daemon_client.py @@ -158,6 +158,9 @@ def submit_task_to_daemon( enable_outputter: bool | None = None, locked_app_package: str | None = None, app_path: str | None = None, + session_id: str | None = None, + ingress: str = "client", + conversation_id: str | None = None, base_url: str | None = None, timeout: float = 5.0, ) -> dict[str, Any] | None: @@ -174,6 +177,9 @@ def submit_task_to_daemon( "enable_outputter": enable_outputter, "locked_app_package": locked_app_package, "app_path": app_path, + "session_id": session_id, + "ingress": ingress, + "conversation_id": conversation_id, } try: diff --git a/artemis/runtime/device_pool.py b/artemis/runtime/device_pool.py index 235c2b7..e14aa41 100644 --- a/artemis/runtime/device_pool.py +++ b/artemis/runtime/device_pool.py @@ -223,6 +223,16 @@ class DevicePool: # If not detected by ADB, still return it so ADB can attempt connection return preferred_serial + # Prefer active ready device from readiness engine if available + try: + from artemis.core.diagnostics import readiness_engine + + active_serial = readiness_engine.get_active_device_serial() + if active_serial and any(d.serial == active_serial and d.state == "device" for d in all_devs): + return active_serial + except Exception: + pass + # Pick first idle device for dev in ready_devs: if not dev.is_busy: diff --git a/artemis/sdk/agent.py b/artemis/sdk/agent.py index 41bef27..200b4db 100644 --- a/artemis/sdk/agent.py +++ b/artemis/sdk/agent.py @@ -203,6 +203,7 @@ class Agent: raise DeviceNotFoundError(error_msg) # Initialize clients + publish_startup_progress("device_check", "Checking the Android device", session_id=self._session_id) if os.environ.get("ARTEMIS_CLOUD_MODE") != "1": self._init_clients( device_id=device_id, @@ -221,6 +222,7 @@ class Agent: device_id=device_id, platform=platform ) logger.info(self._device_context.to_str()) + publish_startup_progress("device_ready", "Android device connected", session_id=self._session_id) # Asynchronously pre-warm LLM connection pools in the background asyncio.create_task(self._prewarm_llm_connections(api_key)) @@ -580,15 +582,15 @@ class Agent: if os.environ.get("ARTEMIS_CLOUD_MODE") != "1": if self._ui_adb_client is not None: publish_startup_progress( - "uiautomator", "Connecting to UI Automator", session_id=str(task.id) + "uiautomator", "Connecting to UI Automator", session_id=str(sess_id) ) await asyncio.to_thread(self._ui_adb_client.connect) publish_startup_progress( - "uiautomator_ready", "UI Automator is ready", session_id=str(task.id) + "uiautomator_ready", "UI Automator is ready", session_id=str(sess_id) ) await self._ensure_device_unlocked() publish_startup_progress( - "environment", "Preparing the device environment", session_id=str(task.id) + "environment", "Preparing the device environment", session_id=str(sess_id) ) await self._prepare_app_installation(task=task) await self._prepare_device_environment(context=context) @@ -620,7 +622,7 @@ class Agent: logger.error(f"[{task_name}] Failed to start screen recording: {e}") publish_startup_progress( - "environment_ready", "Device environment is ready", session_id=str(task.id) + "environment_ready", "Device environment is ready", session_id=str(sess_id) ) try: if request.profile and request.profile.lower() == "flash": diff --git a/mcp_server/tools/inspect_trace.py b/mcp_server/tools/inspect_trace.py index f4f737a..b16e0ec 100644 --- a/mcp_server/tools/inspect_trace.py +++ b/mcp_server/tools/inspect_trace.py @@ -24,6 +24,7 @@ from PIL import Image, ImageDraw from mcp_server.base import mcp from mcp_server.utils import env_utils, trace_store +from artemis.utils.visualization import draw_action_overlay_on_image from artemis.utils.task_tree import ( _render_step_detailed, build_plan_and_history, @@ -31,79 +32,41 @@ from artemis.utils.task_tree import ( def _draw_action_overlay(bg_image_path: str, action_obj: dict, output_path: str) -> bool: - """Draws the action (tap, swipe, etc.) on the image and saves it.""" + """Draws the action overlay on the image using Artemis's native visualization engine.""" try: - with Image.open(bg_image_path) as img: - draw_img = img.convert("RGBA") - draw = ImageDraw.Draw(draw_img) + if not os.path.exists(bg_image_path): + return False - act_type = action_obj.get("action") or action_obj.get("name") - coords = action_obj.get("coordinates") or action_obj.get("target") + with open(bg_image_path, "rb") as f: + raw_bytes = f.read() - if not act_type or not coords: - return False + act_name = action_obj.get("action") or action_obj.get("name") or "" + action_args = dict(action_obj.get("args") or {}) + for k in ( + "target", + "coordinates", + "point", + "direction", + "sequence", + "targets", + "normalized_coordinates", + ): + if k in action_obj and k not in action_args: + action_args[k] = action_obj[k] - fill_color = (255, 0, 0, 80) - outline_color = (255, 0, 0, 255) - outline_width = 4 + annotated_bytes = draw_action_overlay_on_image( + image_bytes=raw_bytes, + action_name=act_name, + action_args=action_args, + ) - if act_type in ( - "tap", - "click", - "long_press", - "focus_and_input_text", - "focus_and_clear_text", - ): - if isinstance(coords, list) and len(coords) == 2: - x, y = coords[0], coords[1] - r = 25 - draw.ellipse( - [x - r, y - r, x + r, y + r], - fill=fill_color, - outline=outline_color, - width=outline_width, - ) - r_center = 4 - draw.ellipse( - [ - x - r_center, - y - r_center, - x + r_center, - y + r_center, - ], - fill=outline_color, - outline=outline_color, - ) - elif act_type == "swipe": - if isinstance(coords, list) and len(coords) == 4: - x1, y1, x2, y2 = coords[0], coords[1], coords[2], coords[3] - draw.line([x1, y1, x2, y2], fill=outline_color, width=6) - r_start = 8 - draw.ellipse( - [ - x1 - r_start, - y1 - r_start, - x1 + r_start, - y1 + r_start, - ], - fill=outline_color, - ) - dx = x2 - x1 - dy = y2 - y1 - angle = math.atan2(dy, dx) - arrow_length = 30 - arrow_angle = math.pi / 6 + if not annotated_bytes or annotated_bytes == raw_bytes: + return False - p1_x = x2 - arrow_length * math.cos(angle - arrow_angle) - p1_y = y2 - arrow_length * math.sin(angle - arrow_angle) - p2_x = x2 - arrow_length * math.cos(angle + arrow_angle) - p2_y = y2 - arrow_length * math.sin(angle + arrow_angle) - - draw.polygon([x2, y2, p1_x, p1_y, p2_x, p2_y], fill=outline_color) - - final_img = draw_img.convert("RGB") - final_img.save(output_path, "JPEG", quality=85) - return True + os.makedirs(os.path.dirname(output_path), exist_ok=True) + with open(output_path, "wb") as f: + f.write(annotated_bytes) + return True except Exception as e: print(f"Error drawing action overlay: {e}", file=sys.stderr) return False @@ -434,7 +397,7 @@ async def mobile_inspect_trace(action: str, trace_id: str, step_number: int | No action_obj[0] if isinstance(action_obj, list) and action_obj else action_obj ) - if isinstance(action_parsed, dict) and action_parsed.get("action"): + if isinstance(action_parsed, dict) and (action_parsed.get("action") or action_parsed.get("name")): if not os.path.exists(overlay_abs_path): success_draw = _draw_action_overlay( path, action_parsed, overlay_abs_path diff --git a/mcp_server/tools/task_runner.py b/mcp_server/tools/task_runner.py index 7c29a42..db16cf3 100644 --- a/mcp_server/tools/task_runner.py +++ b/mcp_server/tools/task_runner.py @@ -162,16 +162,53 @@ def mobile_run_task( device_serial=device_serial, ) - # 3. Auto-spawn daemon if not already running (unless standalone forced) + # 3. Dispatch via unified Artemis Daemon scheduler if available (unless standalone forced) if os.environ.get("ARTEMIS_STANDALONE") != "1": try: - from artemis.runtime import ensure_daemon_running + from artemis.runtime import ensure_daemon_running, submit_task_to_daemon - ensure_daemon_running(timeout=1.5) - except Exception: - pass + is_running, base_url = ensure_daemon_running(timeout=2.0, wait_ready=True) + if is_running: + resp = submit_task_to_daemon( + goal=task_desc, + profile=canonical_model.lower(), + device_serial=device_serial, + expected_output=expected_output_desc if canonical_model != "Flash" else None, + locked_app_package=locked_app_package, + app_path=app_path, + session_id=trace_id, + ingress="mcp", + conversation_id=conversation_id, + base_url=base_url, + ) + if resp and resp.get("tasks"): + assigned_sid = resp["tasks"][0].get("session_id", trace_id) + trace_store.update_trace_status( + assigned_sid, "running", device_serial=device_serial + ) + notes_dir = str(trace_store.get_trace_notes_dir(assigned_sid)) + stdout_log = str(trace_store.get_trace_stdout_log_path(assigned_sid)) + stderr_log = str(trace_store.get_trace_stderr_log_path(assigned_sid)) + return { + "trace_id": assigned_sid, + "status": "running", + "message": ( + f"Autonomous task '{task_desc}' enqueued via unified Artemis Daemon.\n" + f"Trace ID: {assigned_sid}\n" + f"Model: {canonical_model}\n" + f"Live telemetry: streaming to web workspace." + ), + "notes_dir": notes_dir, + "stdout_log": stdout_log, + "stderr_log": stderr_log, + } + except Exception as exc: + import logging + logging.getLogger("mcp_server").warning( + f"Daemon dispatch failed, falling back to standalone runner: {exc}" + ) - # 4. Resolve project root, python executable, and background runner module + # 4. Standalone Fallback: Resolve project root, python executable, and background runner module project_root = env_utils.get_project_root() python_exe = env_utils.resolve_python_executable(project_root) reserve_kwargs: dict[str, Any] = { @@ -214,6 +251,14 @@ def mobile_run_task( env["ADB_DEVICE_SERIAL"] = device_serial env["ARTEMIS_DEVICE_ID"] = device_serial env[DeviceExecutionLock.QUEUE_TICKET_ENV] = queue_ticket + try: + from artemis.config.runtime import read_ipc_port + + ipc_port = read_ipc_port() + if ipc_port: + env["ARTEMIS_IPC_PORT"] = str(ipc_port) + except Exception: + pass proc_kwargs = env_utils.get_detached_process_kwargs() proc = subprocess.Popen( diff --git a/tests/unit/admin_console/test_server_lifecycle.py b/tests/unit/admin_console/test_server_lifecycle.py index c3a35e5..bcc6555 100644 --- a/tests/unit/admin_console/test_server_lifecycle.py +++ b/tests/unit/admin_console/test_server_lifecycle.py @@ -59,6 +59,33 @@ async def test_cancelled_event_stream_reaps_queue_waiter(): assert _pending_queue_get_tasks() == queue_tasks_before +@pytest.mark.asyncio +async def test_stream_events_active_session_replay(): + """Verify stream_events with default 'active' replays running session info and steps.""" + state.is_shutting_down = False + state.shutdown_event.clear() + state.active_session_id = "replay-sess-123" + state.current_goal = "Replay Goal" + state.current_profile = "flash" + + with patch("apps.admin_console.database.repositories.step_repository.step_repo") as mock_step_repo: + mock_step_repo.get_session_steps.return_value = [ + {"step_number": 1, "description": "Step 1", "status": "completed"} + ] + response = await stream_events() + iterator = response.body_iterator + first_event = await anext(iterator) + assert "Subscribed to session active" in first_event + second_event = await anext(iterator) + assert "session_started" in second_event + assert "replay-sess-123" in second_event + third_event = await anext(iterator) + assert "step_recorded" in third_event + assert "Step 1" in third_event + await iterator.aclose() + + + @pytest.mark.asyncio async def test_windows_sigint_is_ignored_while_task_is_active(): state.current_process = MagicMock(returncode=None) diff --git a/tests/unit/admin_console/test_task_device_readiness.py b/tests/unit/admin_console/test_task_device_readiness.py index 46fc1f4..f3f2982 100644 --- a/tests/unit/admin_console/test_task_device_readiness.py +++ b/tests/unit/admin_console/test_task_device_readiness.py @@ -74,3 +74,41 @@ async def test_run_task_enqueues_when_device_is_unlocked(monkeypatch): assert result["status"] == "started" enqueue_tasks.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_run_task_switches_to_unlocked_device_and_syncs_engine(monkeypatch): + """When default device is locked, run_task should enqueue on verified fallback and update engine.""" + unlocked_probe = ProbeResult( + id="android_adb", + category=ProbeCategory.DEVICE, + title="Device / Emulator Connected", + status=ProbeStatus.PASS, + is_blocker=True, + summary="Connected", + description="Ready.", + metadata={ + "installed": True, + "submission_probe": True, + "active_device": {"serial": "emulator-5556", "is_locked": False}, + }, + ) + run_probe = AsyncMock(return_value=unlocked_probe) + enqueue_tasks = AsyncMock(return_value={"status": "started", "tasks": []}) + monkeypatch.setattr( + tasks.readiness_engine, "run_device_submission_probe", run_probe + ) + monkeypatch.setattr(tasks.task_queue_service, "enqueue_tasks", enqueue_tasks) + monkeypatch.setattr(tasks.readiness_engine, "get_active_device_serial", lambda: "device-locked-1") + set_active_mock = AsyncMock() + monkeypatch.setattr(tasks.readiness_engine, "set_active_device_serial", set_active_mock) + + result = await tasks.run_task(RunRequest(goal="Run on any ready device")) + + assert result["status"] == "started" + enqueue_tasks.assert_awaited_once() + # Check that device_serial passed to enqueue_tasks is the verified unlocked device + _, kwargs = enqueue_tasks.call_args + assert kwargs.get("device_serial") == "emulator-5556" + set_active_mock.assert_called_once_with("emulator-5556") + diff --git a/tests/unit/admin_console/test_task_queue_service.py b/tests/unit/admin_console/test_task_queue_service.py index be7d5a1..552fe9d 100644 --- a/tests/unit/admin_console/test_task_queue_service.py +++ b/tests/unit/admin_console/test_task_queue_service.py @@ -607,3 +607,198 @@ def test_darwin_terminate_process_tree_preserves_direct_child_for_asyncio(): for call_args in mock_wait_procs.call_args_list: procs_waited = call_args[0][0] assert parent not in procs_waited + + +@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"): + res = await task_queue_service.enqueue_tasks( + ["Test unified goal"], + profile="flash", + ingress="mcp", + session_id="custom-mcp-session-123", + conversation_id="conv-456", + ) + assert res["status"] in ("started", "queued") + assert len(res["tasks"]) == 1 + task = res["tasks"][0] + assert task["session_id"] == "custom-mcp-session-123" + assert task["ingress"] == "mcp" + assert task["conversation_id"] == "conv-456" + assert task["goal"] == "Test unified goal" + + +@pytest.mark.asyncio +async def test_queue_worker_notifies_conversation(): + """Verify queue_worker calls notify() when conversation_id is attached to task.""" + executed_cmds = [] + + async def fake_subprocess_exec(*args, **kwargs): + executed_cmds.append(list(args)) + proc = MagicMock() + proc.pid = 99999 + proc.wait = AsyncMock(return_value=0) + proc.returncode = 0 + return proc + + with ( + patch("asyncio.create_subprocess_exec", side_effect=fake_subprocess_exec), + patch("apps.admin_console.services.task_queue_service.media_service"), + patch("mcp_server.notifiers.notify") as mock_notify, + patch("apps.admin_console.services.task_queue_service.session_repo") as mock_repo, + ): + mock_repo.get_running_session_id.return_value = None + + await task_queue_service.enqueue_tasks( + ["Notify goal"], + profile="flash", + ingress="mcp", + conversation_id="conv-789", + device_serial="test-mock-serial", + ) + + for _ in range(30): + if mock_notify.called and len(state.queue_items) == 0: + break + await asyncio.sleep(0.05) + + mock_notify.assert_called_once() + kwargs = mock_notify.call_args[1] + assert kwargs["conversation_id"] == "conv-789" + assert "Notify goal" in kwargs["message"] + + task = state.worker_task + if task and not task.done(): + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): + pass + + +def test_stop_tasks_by_session_id_targets_correct_task_among_multiple(): + """Stopping by session_id terminates only the targeted owner among multiple concurrent devices.""" + owner_a = DeviceLockOwner( + pid=11111, + process_created_at=100.0, + token="token-a", + device_id="device-a", + description="Task on device A", + acquired_at="2026-08-24T00:00:00+00:00", + session_id="session-a", + ingress="cli", + ) + owner_b = DeviceLockOwner( + pid=22222, + process_created_at=200.0, + token="token-b", + device_id="device-b", + description="Task on device B", + acquired_at="2026-08-24T00:00:00+00:00", + session_id="session-b", + ingress="mcp", + ) + + with ( + patch( + "apps.admin_console.services.task_queue_service.DeviceExecutionLock.get_active_owners", + return_value={"device_a": owner_a, "device_b": owner_b}, + ), + patch( + "apps.admin_console.services.task_queue_service.DeviceExecutionLock.is_active_owner", + return_value=True, + ), + patch( + "apps.admin_console.services.task_queue_service.DeviceExecutionLock.cleanup_stale_locks" + ) as cleanup_locks, + patch( + "apps.admin_console.services.task_queue_service.process_supervisor.terminate_tree_verified", + return_value=True, + ) as terminate_verified, + patch( + "apps.admin_console.services.task_queue_service.session_repo.update_session_status" + ) as update_status, + patch("mcp_server.utils.trace_store.update_trace_status") as update_trace, + ): + # Explicitly stop session-b + assert task_queue_service.stop_tasks(clear_all=False, session_id="session-b") is True + + # Only session-b's process (22222) should be terminated + terminate_verified.assert_called_once_with(22222, 200.0) + cleanup_locks.assert_called_once_with("device-b") + update_status.assert_called_once() + assert update_status.call_args[0][0] == "session-b" + assert update_status.call_args[0][1] == "cancelled" + update_trace.assert_called_once_with( + "session-b", + "cancelled", + error="Task stopped from the Artemis frontend.", + ) + + +def test_stop_tasks_by_device_id_targets_specific_device(): + """Stopping by device_id targets the active task on that specific device.""" + owner_a = DeviceLockOwner( + pid=33333, + process_created_at=300.0, + token="token-a", + device_id="pixel-8", + description="Task on Pixel 8", + acquired_at="2026-08-24T00:00:00+00:00", + session_id="session-pixel8", + ingress="sdk", + ) + + with ( + patch( + "apps.admin_console.services.task_queue_service.DeviceExecutionLock.get_active_owners", + return_value={"pixel_8": owner_a}, + ), + patch( + "apps.admin_console.services.task_queue_service.DeviceExecutionLock.is_active_owner", + return_value=True, + ), + patch("apps.admin_console.services.task_queue_service.DeviceExecutionLock.cleanup_stale_locks"), + patch( + "apps.admin_console.services.task_queue_service.process_supervisor.terminate_tree_verified", + return_value=True, + ) as terminate_verified, + patch("apps.admin_console.services.task_queue_service.session_repo.update_session_status"), + ): + assert task_queue_service.stop_tasks(clear_all=False, device_id="pixel-8") is True + terminate_verified.assert_called_once_with(33333, 300.0) + + +def test_stop_tasks_queued_item_by_session_id(): + """Stopping a queued task by session_id cancels its reservation and removes it from queue.""" + state.queue_items = [ + {"session_id": "queue-1", "goal": "Goal 1", "status": "pending", "queue_ticket": "ticket-1"}, + {"session_id": "queue-2", "goal": "Goal 2", "status": "pending", "queue_ticket": "ticket-2"}, + ] + + with ( + patch( + "apps.admin_console.services.task_queue_service.DeviceExecutionLock.get_active_owners", + return_value={}, + ), + patch( + "apps.admin_console.services.task_queue_service.DeviceExecutionLock.get_active_owner", + return_value=None, + ), + patch( + "apps.admin_console.services.task_queue_service.DeviceExecutionLock.cancel_reservation", + return_value=True, + ) as cancel_res, + patch("apps.admin_console.services.task_queue_service.session_repo.update_session_status") as update_status, + ): + assert task_queue_service.stop_tasks(clear_all=False, session_id="queue-1") is True + cancel_res.assert_called_once_with("ticket-1") + # queue-2 should remain in queue + assert len(state.queue_items) == 1 + update_status.assert_called_once() + assert update_status.call_args[0][0] == "queue-1" + assert update_status.call_args[0][1] == "cancelled" + + + diff --git a/tests/unit/core/test_diagnostics.py b/tests/unit/core/test_diagnostics.py index 2790872..b212e74 100644 --- a/tests/unit/core/test_diagnostics.py +++ b/tests/unit/core/test_diagnostics.py @@ -313,6 +313,55 @@ async def test_submission_probe_fails_closed_when_lock_state_is_unknown(monkeypa assert result.status == ProbeStatus.WARN +@pytest.mark.asyncio +async def test_submission_probe_falls_back_to_unlocked_device(monkeypatch): + """When the first device is locked but a second device is unlocked, submission probe falls back.""" + probe = AdbDeviceProbe() + monkeypatch.setattr( + probe, + "_get_device_states", + AsyncMock(return_value=[("device-locked", "device"), ("device-unlocked", "device")]), + ) + + async def mock_lock_state(adb_path, serial, timeout_seconds=1.0): + return True if serial == "device-locked" else False + + monkeypatch.setattr(probe, "_get_confirmed_device_lock_state", mock_lock_state) + + result = await probe.probe_submission_readiness() + + assert result.status == ProbeStatus.PASS + assert result.summary == "Connected" + assert result.metadata["active_device"]["serial"] == "device-unlocked" + assert result.metadata["active_device"]["is_locked"] is False + + +@pytest.mark.asyncio +async def test_adb_probe_prefers_unlocked_device_when_one_is_locked(monkeypatch): + """When multiple ready devices exist, probe() should pick the unlocked one as active.""" + from artemis.core.diagnostics.schema import DeviceInfo + + probe = AdbDeviceProbe() + monkeypatch.setattr(probe, "_locate_adb", lambda: "/usr/bin/adb") + monkeypatch.setattr(probe, "_get_adb_version", AsyncMock(return_value="1.0.41")) + monkeypatch.setattr(probe, "_locate_emulator", lambda: "/usr/bin/emulator") + monkeypatch.setattr(probe, "_list_installed_avds", lambda _: []) + + devices = [ + DeviceInfo(serial="dev-locked-1", state="device", model="Pixel 7", is_locked=True), + DeviceInfo(serial="dev-unlocked-2", state="device", model="Pixel 8", is_locked=False), + ] + monkeypatch.setattr(probe, "_parse_adb_devices", AsyncMock(return_value=devices)) + + result = await probe.probe() + + assert result.status == ProbeStatus.PASS + assert result.summary == "Connected" + assert result.metadata["active_device"]["serial"] == "dev-unlocked-2" + assert result.metadata["active_device"]["is_locked"] is False + + + @pytest.mark.asyncio async def test_llm_credentials_probe_structure(): """Verify LLMCredentialsProbe returns correct category and schema.""" diff --git a/tests/unit/runtime/test_daemon_client.py b/tests/unit/runtime/test_daemon_client.py index 322f14f..04b34d1 100644 --- a/tests/unit/runtime/test_daemon_client.py +++ b/tests/unit/runtime/test_daemon_client.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import json from unittest.mock import MagicMock, patch import pytest @@ -92,12 +93,26 @@ def test_submit_task_to_daemon(): mock_resp.read.return_value = b'{"status": "queued", "tasks": [{"session_id": "test-sess"}]}' mock_resp.__enter__.return_value = mock_resp - with patch("urllib.request.urlopen", return_value=mock_resp): + with patch("urllib.request.urlopen", return_value=mock_resp) as mock_urlopen: res = submit_task_to_daemon( "test goal", profile="flash", device_serial="pixel-8", + session_id="custom-sess-789", + ingress="mcp", + conversation_id="conv-123", ) assert res is not None assert res["status"] == "queued" assert res["tasks"][0]["session_id"] == "test-sess" + + # Verify request payload + req = mock_urlopen.call_args[0][0] + data = json.loads(req.data.decode("utf-8")) + assert data["goal"] == "test goal" + assert data["profile"] == "flash" + assert data["device_serial"] == "pixel-8" + assert data["session_id"] == "custom-sess-789" + assert data["ingress"] == "mcp" + assert data["conversation_id"] == "conv-123" + diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 7ff2ff6..0728f60 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -436,3 +436,22 @@ def test_cli_restart_command(monkeypatch): assert stopped["port"] == 8888 assert ui_called["port"] == 8888 assert ui_called["open_browser"] is False + + +def test_cli_server_lifecycle_aliases(monkeypatch): + """Verify 'artemis server status/stop/restart' subcommands are accessible.""" + from artemis.runtime import server_lifecycle + + monkeypatch.setattr(server_lifecycle, "is_port_in_use", lambda port, **kwargs: False) + monkeypatch.setattr(server_lifecycle, "find_server_pids", lambda port: []) + monkeypatch.setattr(server_lifecycle, "read_server_info", lambda: None) + + result = runner.invoke(app, ["server", "status", "--port", "59998"]) + assert result.exit_code == 0 + assert "OFFLINE" in result.output or "STOPPED" in result.output + + help_result = runner.invoke(app, ["server", "--help"]) + assert help_result.exit_code == 0 + assert "restart" in help_result.output + assert "stop" in help_result.output + assert "status" in help_result.output diff --git a/tests/unit/test_config_package.py b/tests/unit/test_config_package.py index 7046a43..7d157ca 100644 --- a/tests/unit/test_config_package.py +++ b/tests/unit/test_config_package.py @@ -16,7 +16,7 @@ from pathlib import Path import tempfile -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import pytest from pydantic import SecretStr @@ -218,8 +218,12 @@ def test_output_config_and_recording(): def test_runtime_state_and_ipc(tmp_path, monkeypatch): """Test IPC port and LS address state helpers.""" + import urllib.request + monkeypatch.setattr(urllib.request, "urlopen", MagicMock(side_effect=Exception("offline"))) ipc_state_file = tmp_path / ".artemis_ipc_port" monkeypatch.setattr("artemis.config.runtime.get_ipc_port_file", lambda: ipc_state_file) + monkeypatch.setattr("artemis.config.runtime.ROOT_DIR", tmp_path) + monkeypatch.setattr("artemis.config.runtime.get_app_dir", lambda: tmp_path) # Test IPC port write_ipc_port(49152) assert read_ipc_port() == 49152 @@ -365,8 +369,8 @@ def test_checker_builder_and_context_propagation(): mock_task.request.record_trace = False mock_task.request.name = "test_task" mock_task.request.profile = None - mock_task.request.goal = "Test goal" - agent._prepare_tracing(mock_task, ctx) + with patch("artemis.sdk.agent.DataEngine"): + agent._prepare_tracing(mock_task, ctx) assert ctx.execution_setup is not None assert ctx.execution_setup.disable_checker is False @@ -455,8 +459,8 @@ def test_explorer_builder_and_resolution(monkeypatch): mock_task.request.record_trace = False mock_task.request.name = "explorer_test_task" mock_task.request.profile = None - mock_task.request.goal = "Test explorer propagation" - agent._prepare_tracing(mock_task, ctx) + with patch("artemis.sdk.agent.DataEngine"): + agent._prepare_tracing(mock_task, ctx) assert ctx.execution_setup is not None assert ctx.execution_setup.explorer_version == "pro" @@ -505,7 +509,8 @@ def test_outputter_builder_and_context_propagation(): mock_task.request.name = "outputter_test_task" mock_task.request.profile = None mock_task.request.goal = "Test outputter propagation" - agent._prepare_tracing(mock_task, ctx) + with patch("artemis.sdk.agent.DataEngine"): + agent._prepare_tracing(mock_task, ctx) assert ctx.execution_setup is not None assert ctx.execution_setup.disable_outputter is False