diff --git a/apps/admin_console/database/repositories/step_repository.py b/apps/admin_console/database/repositories/step_repository.py index 9d7f73a..9b689c3 100644 --- a/apps/admin_console/database/repositories/step_repository.py +++ b/apps/admin_console/database/repositories/step_repository.py @@ -177,11 +177,22 @@ class StepRepository: return None if trace_type == "llm_call": - return ( - {"error": self._clean_value(payload_obj.get("error"))} - if "error" in payload_obj - else None + allowed_keys = ( + "error", + "delay", + "attempt", + "max_retries", + "provider", + "source", + "recoverable", + "pause", ) + cleaned = { + key: self._clean_value(payload_obj.get(key)) + for key in allowed_keys + if key in payload_obj + } + return cleaned or None args = payload_obj.get("args", payload_obj) cleaned_args = ( @@ -291,7 +302,7 @@ class StepRepository: " t1.payload, t2.name as agent_name FROM traces t1 LEFT" " JOIN traces t2 ON t1.parent_trace_id = t2.trace_id WHERE" " t1.step_id = ? AND (t1.type = 'tool' OR t1.type = 'action' OR (t1.type =" - " 'llm_call' AND t1.status = 'failed') OR (t1.type = 'log' AND" + " 'llm_call' AND t1.status IN ('failed', 'retrying')) OR (t1.type = 'log' AND" " t1.payload LIKE '%LLM Error:%Pausing execution%' AND NOT EXISTS" " (SELECT 1 FROM traces t3 WHERE t3.session_id = t1.session_id AND" " t3.type = 'llm_call' AND t3.status = 'failed' AND" @@ -363,7 +374,7 @@ class StepRepository: " t2.name as agent_name FROM traces t1 LEFT JOIN traces t2 ON" " t1.parent_trace_id = t2.trace_id WHERE t1.session_id = ? AND" " (t1.step_id IS NULL OR t1.step_id = '') AND (t1.type = 'tool' OR" - " (t1.type = 'llm_call' AND t1.status = 'failed') OR (t1.type = 'log'" + " (t1.type = 'llm_call' AND t1.status IN ('failed', 'retrying')) OR (t1.type = 'log'" " AND t1.payload LIKE '%LLM Error:%Pausing execution%' AND NOT EXISTS" " (SELECT 1 FROM traces t3 WHERE t3.session_id = t1.session_id AND" " t3.type = 'llm_call' AND t3.status = 'failed' AND" diff --git a/apps/admin_console/services/task_queue_service.py b/apps/admin_console/services/task_queue_service.py index 8fe0bc6..73ec96b 100644 --- a/apps/admin_console/services/task_queue_service.py +++ b/apps/admin_console/services/task_queue_service.py @@ -13,6 +13,7 @@ # limitations under the License. import asyncio +import codecs from datetime import datetime import os import shutil @@ -78,17 +79,66 @@ class TaskQueueService: ] @staticmethod - def _subprocess_creation_kwargs() -> dict[str, int]: - """Keep task workers out of the UI server's Windows console group. + def _subprocess_creation_kwargs() -> dict[str, Any]: + """Isolate task workers from the UI server's Windows console. - Without a new process group, a console control event intended for the - uvicorn server is also delivered to a task that is still importing - Artemis, causing STATUS_CONTROL_C_EXIT (0xC000013A). + A new process group alone is insufficient on Windows: the worker still + shares the parent's console, so a CTRL_C_EVENT generated anywhere in + that console can reach the UI server. CREATE_NO_WINDOW removes that + shared console boundary. Output is captured and forwarded explicitly + so detached workers remain visible in the server terminal. """ if sys.platform == "win32": - return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} + return { + "creationflags": ( + subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW + ), + "stdout": asyncio.subprocess.PIPE, + "stderr": asyncio.subprocess.STDOUT, + } return {} + @staticmethod + async def _forward_worker_output(stream: asyncio.StreamReader | None) -> None: + """Forward a detached Windows worker's combined output without corrupting UTF-8.""" + if stream is None: + return + + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") + while True: + chunk = await stream.read(4096) + if not chunk: + break + text = decoder.decode(chunk) + if text: + sys.stdout.write(text) + sys.stdout.flush() + + tail = decoder.decode(b"", final=True) + if tail: + sys.stdout.write(tail) + sys.stdout.flush() + + @staticmethod + async def _finish_output_forwarder(output_task: asyncio.Task[None] | None) -> None: + """Drain final worker output without allowing inherited handles to stall the queue.""" + if output_task is None: + return + try: + await asyncio.wait_for(asyncio.shield(output_task), timeout=2.0) + except asyncio.CancelledError: + if output_task.cancelled(): + return + raise + except TimeoutError: + output_task.cancel() + try: + await output_task + except asyncio.CancelledError: + pass + except Exception as exc: + print(f"[QueueWorker] Failed to forward detached worker output: {exc}") + @staticmethod async def _terminate_worker_process(proc: asyncio.subprocess.Process | None) -> None: """Stop a worker without changing the established POSIX behavior.""" @@ -125,6 +175,7 @@ class TaskQueueService: sess_id = None goal = None profile = "flash" + output_task: asyncio.Task[None] | None = None try: if state.is_running: await asyncio.sleep(0.2) @@ -212,6 +263,8 @@ class TaskQueueService: ) state.current_process = proc task_item["pid"] = proc.pid + if sys.platform == "win32" and isinstance(proc.stdout, asyncio.StreamReader): + output_task = asyncio.create_task(cls._forward_worker_output(proc.stdout)) if sess_id and ( str(sess_id) in getattr(state, "cancelled_session_ids", set()) @@ -255,6 +308,7 @@ class TaskQueueService: print(f"[QueueWorker] Unexpected error executing task: {e}") await asyncio.sleep(0.5) finally: + await cls._finish_output_forwarder(output_task) # 5. Clean up completed task from queue and reset execution state if sess_id: cls._remove_task(sess_id) 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 1f3dbb4..c8263d3 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 @@ -285,18 +285,33 @@
+ +
+
+ pause_circle +
+ Task paused + The AI request failed. Resume the task to try again. +
+
+ +
+
sensors

No Stream Activity Selected

Please select a session from the Task List on the right to view its real-time log payloads.

- -
- sync - {{ agentService.retryMessage() || 'Reconnecting to AI model...' }} -
-
@@ -513,6 +528,48 @@ + +
+
+ cloud_sync + AI service retry + + {{ getLLMRetryCount(item.data) }}× retried + + + {{ isActionExpanded('retry-' + block.id + '-' + itemIdx) ? 'expand_less' : 'expand_more' }} + +
+
+
+
+ Attempts + {{ getLLMRetryCount(item.data) }} +
+
+ Extra wait + {{ totalDelay }} +
+
+ Provider + {{ retryProvider }} +
+
+
+
+
+ Retry {{ retryIndex + 1 }} + +{{ retryDelay }} +
+
{{ getLLMRetryEntryError(retry) }}
+
+
+
+
+
@@ -530,14 +587,14 @@
diff --git a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.scss b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.scss index 3da10c9..c22b0ba 100644 --- a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.scss +++ b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.scss @@ -1053,24 +1053,96 @@ $uniform-gap: 8px; } } - .retrying-banner { + .paused-task-banner { display: flex; align-items: center; - gap: 8px; - padding: 10px 16px; - background-color: #fef9c3; - border: 1px solid #fde047; - border-radius: 8px; - color: #854d0e; - font-size: 13px; - font-weight: 500; + justify-content: space-between; + gap: 16px; + padding: 12px 14px; margin-bottom: 12px; - box-shadow: 0 1px 3px rgba(0, 0, 0, 0.05); + background: #fff7ed; + border: 1px solid #fdba74; + border-radius: 10px; + color: #9a3412; + box-shadow: 0 1px 3px rgba(154, 52, 18, 0.08); - .spinning-icon { - font-size: 18px; - color: #d97706; - animation: spin 1.5s linear infinite; + .paused-task-message { + min-width: 0; + display: flex; + align-items: center; + gap: 10px; + + .paused-task-icon { + flex: 0 0 auto; + font-size: 21px; + color: #ea580c; + } + + div { + min-width: 0; + display: flex; + flex-direction: column; + gap: 2px; + } + + strong { + font-size: 12px; + line-height: 1.2; + } + + span:not(.paused-task-icon) { + font-size: 11px; + line-height: 1.35; + color: #c2410c; + } + } + + .btn-resume-task { + flex: 0 0 auto; + min-height: 32px; + padding: 6px 12px; + display: inline-flex; + align-items: center; + justify-content: center; + gap: 5px; + border: 1px solid #2563eb; + border-radius: 8px; + background: #2563eb; + color: #ffffff; + font-family: inherit; + font-size: 12px; + font-weight: 700; + line-height: 1; + cursor: pointer; + box-shadow: 0 2px 5px rgba(37, 99, 235, 0.2); + transition: background 0.18s ease, box-shadow 0.18s ease, transform 0.18s ease; + + .material-symbols-outlined { + font-size: 17px; + } + + &:hover { + background: #1d4ed8; + box-shadow: 0 3px 8px rgba(37, 99, 235, 0.28); + } + + &:active { + transform: scale(0.97); + } + + &:focus-visible { + outline: 2px solid rgba(37, 99, 235, 0.35); + outline-offset: 2px; + } + } + + @media (max-width: 560px) { + align-items: stretch; + flex-direction: column; + + .btn-resume-task { + width: 100%; + } } } @@ -2183,6 +2255,19 @@ $uniform-gap: 8px; } } + &.is-retrying { + background: #fffbeb; + border-color: #fcd34d; + + &:hover { + border-color: #f59e0b; + } + + .action-card-body .detail-value { + color: #78350f; + } + } + &.task-report-card { background: #f0fdf4; border: 1px solid #bbf7d0; @@ -2287,6 +2372,10 @@ $uniform-gap: 8px; &.is-failed { color: #ef4444; } + + &.is-retrying { + color: #d97706; + } } .action-title { @@ -2352,12 +2441,34 @@ $uniform-gap: 8px; color: #f97316; } } + + &.is-retrying { + background: #fef3c7; + color: #92400e; + border: 1px solid #fde68a; + + .status-icon { + color: #d97706; + + &.is-active { + animation: spin 1.5s linear infinite; + } + } + } + } + + &.llm-retry-header { + border-bottom-color: #fde68a; + + .action-title { + color: #78350f; + } } &.llm-failure-header { flex-wrap: wrap; - .btn-retry-llm { + .btn-resume-llm { flex: 0 0 auto; min-height: 26px; padding: 3px 9px; @@ -2417,6 +2528,83 @@ $uniform-gap: 8px; flex-direction: column; gap: 6px; + .retry-summary-grid { + display: grid; + grid-template-columns: repeat(3, minmax(0, 1fr)); + gap: 6px; + } + + .retry-summary-item { + min-width: 0; + padding: 7px 8px; + border: 1px solid rgba(245, 158, 11, 0.18); + border-radius: 7px; + background: rgba(255, 255, 255, 0.68); + display: flex; + flex-direction: column; + gap: 2px; + + .retry-summary-label { + color: #a16207; + font-size: 9.5px; + font-weight: 600; + letter-spacing: 0.02em; + text-transform: uppercase; + } + + strong { + color: #78350f; + font-size: 11.5px; + font-weight: 700; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + } + } + + .retry-attempt-list { + border-top: 1px solid rgba(245, 158, 11, 0.22); + display: flex; + flex-direction: column; + gap: 6px; + margin-top: 2px; + padding-top: 8px; + animation: fadeInDetails 0.2s ease-out; + + .retry-attempt-item { + padding: 8px 9px; + border: 1px solid rgba(245, 158, 11, 0.2); + border-radius: 7px; + background: rgba(255, 255, 255, 0.76); + } + + .retry-attempt-heading { + display: flex; + align-items: center; + justify-content: space-between; + gap: 8px; + margin-bottom: 4px; + color: #92400e; + font-size: 10px; + font-weight: 700; + } + + .retry-wait-pill { + padding: 1px 5px; + border-radius: 999px; + background: #fef3c7; + color: #b45309; + } + + .retry-attempt-error { + color: #78350f; + font-size: 10.5px; + line-height: 1.45; + overflow-wrap: anywhere; + user-select: text; + } + } + .action-detail-row { display: flex; align-items: baseline; diff --git a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts index 83e71fd..78014fb 100644 --- a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts +++ b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts @@ -230,6 +230,14 @@ export class AgentStreamComponent implements AfterViewInit { return checkPlanningLoader(this.filteredLogs(), isRunning, isCurrentRunningSession); }); + public isViewingPausedTask = computed(() => { + if (!this.agentService.isPaused()) return false; + + const currentSessionId = this.agentService.currentSessionId(); + const pausedSessionId = this.agentService.runningSessionId(); + return !currentSessionId || currentSessionId === pausedSessionId; + }); + public sessionDuration = computed(() => { const currentSession = this.agentService.sessions().find(s => s.session_id === this.agentService.currentSessionId()); return computeSessionDuration(currentSession?.start_time || 0, this.filteredLogs()); @@ -819,7 +827,7 @@ export class AgentStreamComponent implements AfterViewInit { if (!block) return false; const hasNative = Boolean(this.getNativeThinking(block)); const hasRaw = Boolean(this.getRawThinking(block)); - const hasVisibleTools = Boolean(block.data?.generic_tools) && block.data.generic_tools.some((t: any) => this.shouldShowTool(t, block.data) || (t.type === 'llm_call' && t.status === 'failed') || this.isReportStatusAction(t)); + const hasVisibleTools = Boolean(block.data?.generic_tools) && block.data.generic_tools.some((t: any) => this.shouldShowTool(t, block.data) || (t.type === 'llm_call' && (t.status === 'failed' || t.status === 'retrying')) || this.isReportStatusAction(t)); const hasAndroidActions = Boolean(block.data?.action_taken) && (this.isAndroidAction(block.data.action_taken) || this.isReportStatusAction(block.data.action_taken)); return hasNative || hasRaw || hasVisibleTools || hasAndroidActions; } @@ -844,7 +852,7 @@ export class AgentStreamComponent implements AfterViewInit { return !!tool && this.retryablePauseTrace() === tool; } - public retryPausedTask(event: Event): void { + public resumePausedTask(event: Event): void { event.stopPropagation(); this.agentService.resumeTask(); } @@ -856,6 +864,52 @@ export class AgentStreamComponent implements AfterViewInit { return this.cleanErrorMessage(rawError); } + public isLLMRetry(tool: any): boolean { + return tool?.type === 'llm_call' && tool?.status === 'retrying'; + } + + public getLLMRetryEntries(tool: any): any[] { + const retries = tool?.payload?.retries; + return Array.isArray(retries) && retries.length > 0 ? retries : [tool?.payload || tool]; + } + + public getLLMRetryCount(tool: any): number { + const count = Number(tool?.payload?.retry_count); + return Number.isFinite(count) && count > 0 ? count : this.getLLMRetryEntries(tool).length; + } + + public getLLMRetryTotalDelay(tool: any): string | null { + const configuredTotal = Number(tool?.payload?.total_delay); + const total = Number.isFinite(configuredTotal) + ? configuredTotal + : this.getLLMRetryEntries(tool).reduce( + (sum: number, retry: any) => sum + (Number(retry?.delay) || 0), + 0 + ); + if (total <= 0) return null; + return `${total.toFixed(2).replace(/\.00$/, '')}s`; + } + + public getLLMRetryEntryError(entry: any): string { + return this.cleanErrorMessage(entry?.error || 'Unknown error'); + } + + public getLLMRetryEntryDelay(entry: any): string | null { + const delay = Number(entry?.delay); + if (!Number.isFinite(delay) || delay <= 0) return null; + return `${delay.toFixed(2).replace(/\.00$/, '')}s`; + } + + public getLLMRetryProvider(tool: any): string | null { + const providers = Array.isArray(tool?.payload?.providers) + ? tool.payload.providers + : [tool?.payload?.provider]; + const labels = providers + .filter((provider: any) => typeof provider === 'string' && provider.trim()) + .map((provider: string) => provider.trim().replace(/^./, (value: string) => value.toUpperCase())); + return labels.length > 0 ? Array.from(new Set(labels)).join(', ') : null; + } + public cleanErrorMessage(rawError: any): string { return cleanErrorMessage(rawError); } diff --git a/apps/showcase_ui/src/app/services/agent.service.ts b/apps/showcase_ui/src/app/services/agent.service.ts index 3025956..8df14f0 100644 --- a/apps/showcase_ui/src/app/services/agent.service.ts +++ b/apps/showcase_ui/src/app/services/agent.service.ts @@ -518,9 +518,12 @@ export class AgentService { if (eventType === 'llm_retrying') { this.isRetrying.set(true); - const attempt = parsedData.attempt || 1; - const max = parsedData.max_retries || 3; - this.retryMessage.set(`Reconnecting to AI model (Attempt ${attempt}/${max})...`); + const attempt = Number(parsedData.attempt || 0); + const max = Number(parsedData.max_retries || 0); + const delay = Number(parsedData.delay || 0); + 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}...`); return; } diff --git a/apps/showcase_ui/src/app/utils/stream-aggregator.util.ts b/apps/showcase_ui/src/app/utils/stream-aggregator.util.ts index 2003b42..cbed23a 100644 --- a/apps/showcase_ui/src/app/utils/stream-aggregator.util.ts +++ b/apps/showcase_ui/src/app/utils/stream-aggregator.util.ts @@ -162,8 +162,9 @@ export function consolidateLogsToBlocks(rawLogs: any[]): StepBlock[] { } else if (log.type === 'trace_recorded') { const isAction = log.data.type === 'action'; const isTool = log.data.type === 'tool' || isAction; - const isFailedLLM = log.data.type === 'llm_call' && log.data.status === 'failed'; - if (!isTool && !isFailedLLM) return; + const isVisibleLLMEvent = log.data.type === 'llm_call' + && (log.data.status === 'failed' || log.data.status === 'retrying'); + if (!isTool && !isVisibleLLMEvent) return; let stepId = log.data.step_id; let existingIndex = -1; diff --git a/apps/showcase_ui/src/app/utils/tool-formatter.util.spec.ts b/apps/showcase_ui/src/app/utils/tool-formatter.util.spec.ts new file mode 100644 index 0000000..a22ac4a --- /dev/null +++ b/apps/showcase_ui/src/app/utils/tool-formatter.util.spec.ts @@ -0,0 +1,57 @@ +import { getUniqueGenericTools } from './tool-formatter.util'; + +describe('getUniqueGenericTools retry aggregation', () => { + it('groups recoverable LLM retries while preserving every attempt', () => { + const tools = [ + { + trace_id: 'retry-1', + type: 'llm_call', + name: 'llm_retry', + status: 'retrying', + timestamp: 100, + payload: { error: '503 first', delay: 1.18, provider: 'google' } + }, + { + trace_id: 'retry-2', + type: 'llm_call', + name: 'llm_retry', + status: 'retrying', + timestamp: 101, + payload: { error: '503 second', delay: 2.96, provider: 'google' } + } + ]; + + const result = getUniqueGenericTools(tools); + + expect(result.length).toBe(1); + expect(result[0].name).toBe('llm_retry_group'); + expect(result[0].payload.retry_count).toBe(2); + expect(result[0].payload.total_delay).toBeCloseTo(4.14); + expect(result[0].payload.providers).toEqual(['google']); + expect(result[0].payload.retries.map((retry: any) => retry.error)).toEqual([ + '503 first', + '503 second' + ]); + }); + + it('keeps a terminal LLM failure separate from its preceding retries', () => { + const result = getUniqueGenericTools([ + { + trace_id: 'retry-1', + type: 'llm_call', + name: 'llm_retry', + status: 'retrying', + payload: { error: '503', delay: 1 } + }, + { + trace_id: 'failure-1', + type: 'llm_call', + name: 'llm_pause', + status: 'failed', + payload: { error: '503 exhausted', pause: true } + } + ]); + + expect(result.map(tool => tool.name)).toEqual(['llm_retry_group', 'llm_pause']); + }); +}); diff --git a/apps/showcase_ui/src/app/utils/tool-formatter.util.ts b/apps/showcase_ui/src/app/utils/tool-formatter.util.ts index 4fd55ac..73b9845 100644 --- a/apps/showcase_ui/src/app/utils/tool-formatter.util.ts +++ b/apps/showcase_ui/src/app/utils/tool-formatter.util.ts @@ -185,6 +185,42 @@ export function getToolAgentName(tool: any): string | null { export function getUniqueGenericTools(tools: any[] | undefined): any[] { if (!tools) return []; + const retryTools = tools.filter(tool => + tool?.type === 'llm_call' + && tool?.status === 'retrying' + && (tool?.name === 'llm_retry' || tool?.name === 'llm_retry_group') + ); + const firstRetry = retryTools[0]; + const latestRetry = retryTools[retryTools.length - 1]; + const retryAggregate = firstRetry ? { + ...firstRetry, + trace_id: `llm-retry-group-${firstRetry.trace_id || firstRetry.timestamp || 'unknown'}`, + name: 'llm_retry_group', + payload: { + ...(latestRetry?.payload || {}), + retry_count: retryTools.length, + total_delay: retryTools.reduce( + (total, retry) => total + (Number(retry?.payload?.delay) || 0), + 0 + ), + providers: Array.from(new Set( + retryTools + .map(retry => retry?.payload?.provider) + .filter((provider): provider is string => typeof provider === 'string' && !!provider) + )), + retries: retryTools.map(retry => ({ + trace_id: retry.trace_id, + timestamp: retry.timestamp, + error: retry?.payload?.error || retry.error, + delay: Number(retry?.payload?.delay) || 0, + provider: retry?.payload?.provider, + source: retry?.payload?.source, + attempt: retry?.payload?.attempt, + max_retries: retry?.payload?.max_retries + })) + } + } : null; + const toolMap = new Map(); for (const t of tools) { if (t && t.trace_id) { @@ -196,11 +232,20 @@ export function getUniqueGenericTools(tools: any[] | undefined): any[] { const result: any[] = []; const seenTraces = new Set(); + let retryAggregateAdded = false; for (const tool of tools) { if (!tool || !tool.name) continue; if (tool.type === 'agent') continue; + if (tool.type === 'llm_call' && tool.status === 'retrying') { + if (!retryAggregateAdded && retryAggregate) { + result.push(retryAggregate); + retryAggregateAdded = true; + } + continue; + } + if (tool.type === 'llm_call' && tool.status === 'failed') { result.push(tool); continue; diff --git a/artemis/drivers/android/adb_driver.py b/artemis/drivers/android/adb_driver.py index bf09671..5b11dd0 100644 --- a/artemis/drivers/android/adb_driver.py +++ b/artemis/drivers/android/adb_driver.py @@ -21,10 +21,14 @@ from pathlib import Path from typing import Any, Literal from adbutils import AdbClient, AdbDevice -from artemis.clients.ui_automator_client import UIAutomatorClient +from artemis.clients.ui_automator_client import ( + UIAutomatorClient, + _parse_hierarchy_xml_to_elements, +) from artemis.config.paths import get_temp_dir from artemis.drivers.base import BaseDeviceDriver, KeyCode, ScreenData, SwipeDirection from artemis.toolchain import find_ffmpeg, find_scrcpy +from artemis.utils.ui_filter import filter_ui_hierarchy from artemis.utils.logger import get_logger logger = get_logger(__name__) @@ -119,6 +123,8 @@ class AndroidAdbDriver(BaseDeviceDriver): # 1. Capture raw screenshot screenshot_bytes: bytes = b"" screenshot_base64: str = "" + ui_hierarchy_xml: str | None = None + ui_elements: list[dict[str, Any]] = [] # Check if ui_adb_client has direct screen data support if self._ui_adb_client and hasattr(self._ui_adb_client, "get_screen_data"): @@ -135,6 +141,16 @@ class AndroidAdbDriver(BaseDeviceDriver): self._width = raw_ui_data.width if hasattr(raw_ui_data, "height") and isinstance(raw_ui_data.height, int): self._height = raw_ui_data.height + if hasattr(raw_ui_data, "hierarchy_xml") and isinstance( + raw_ui_data.hierarchy_xml, str + ): + ui_hierarchy_xml = raw_ui_data.hierarchy_xml + if hasattr(raw_ui_data, "elements"): + raw_elements = raw_ui_data.elements + if isinstance(raw_elements, dict): + ui_elements = [raw_elements] + elif isinstance(raw_elements, list): + ui_elements = raw_elements except Exception as e: logger.debug(f"Direct ui_adb_client.get_screen_data error: {e}") @@ -156,13 +172,16 @@ class AndroidAdbDriver(BaseDeviceDriver): screenshot_bytes = base64.b64decode(screenshot_base64) # 2. Capture UI Hierarchy via UIAutomator - ui_hierarchy_xml: str | None = None - ui_elements: list[dict[str, Any]] = [] - if self._ui_adb_client: + if self._ui_adb_client and not ui_elements: try: if hasattr(self._ui_adb_client, "get_hierarchy"): res = self._ui_adb_client.get_hierarchy() - ui_elements = await res if asyncio.iscoroutine(res) else res + hierarchy = await res if asyncio.iscoroutine(res) else res + if isinstance(hierarchy, str): + ui_hierarchy_xml = hierarchy + ui_elements = _parse_hierarchy_xml_to_elements(hierarchy) + else: + ui_elements = hierarchy elif hasattr(self._ui_adb_client, "get_ui_elements"): res = self._ui_adb_client.get_ui_elements() ui_elements = await res if asyncio.iscoroutine(res) else res @@ -178,6 +197,11 @@ class AndroidAdbDriver(BaseDeviceDriver): w = self._width if isinstance(self._width, int) else 1080 h = self._height if isinstance(self._height, int) else 2400 + clean_ui_elements = filter_ui_hierarchy( + clean_ui_elements, + screen_width=w, + screen_height=h, + ) clean_b64 = ( screenshot_base64 if isinstance(screenshot_base64, str) and screenshot_base64 diff --git a/artemis/services/llm.py b/artemis/services/llm.py index a76797c..c9b00ff 100644 --- a/artemis/services/llm.py +++ b/artemis/services/llm.py @@ -25,6 +25,8 @@ import functools import logging from pathlib import Path import random +import re +import sys import time from typing import Any, Literal, TypeVar, overload from uuid import uuid4 @@ -41,7 +43,6 @@ from artemis.config import ( settings, ) from artemis.context import ArtemisContext -from artemis.data_engine.engine import _CURRENT_DATA_ENGINE from artemis.data_engine.trace import CURRENT_TRACE_ID, DataEngineCallbackHandler from artemis.llm.router import ModelEndpoint, ModelFactory, ModelProvider from artemis.utils.logger import get_logger @@ -53,6 +54,131 @@ user_messages_logger = get_logger(__name__) T = TypeVar("T") +# Retained as an override seam for unit tests. At runtime the active engine must +# be resolved dynamically because DataEngine.start_session() updates the module +# global after this module has already been imported. +_CURRENT_DATA_ENGINE = None + + +def _get_current_data_engine(): + if _CURRENT_DATA_ENGINE is not None: + return _CURRENT_DATA_ENGINE + engine_module = sys.modules.get("artemis.data_engine.engine") + return getattr(engine_module, "_CURRENT_DATA_ENGINE", None) if engine_module else None + + +def _record_llm_retry( + error: str, + delay: float, + *, + attempt: int | None = None, + max_retries: int | None = None, + provider: str | None = None, + source: str = "artemis", +) -> None: + """Persist and publish one recoverable LLM retry for UI transparency.""" + engine = _get_current_data_engine() + if not engine or not getattr(engine, "current_session_id", None): + return + + error_text = str(error) + if len(error_text) > 1000: + error_text = error_text[:1000] + "... [Truncated by LLM Retry Telemetry]" + + payload = { + "error": error_text, + "delay": float(delay), + "attempt": attempt, + "max_retries": max_retries, + "provider": provider, + "source": source, + "recoverable": True, + } + payload = {key: value for key, value in payload.items() if value is not None} + step_id = getattr(engine, "current_step_id", None) + parent_trace_id = CURRENT_TRACE_ID.get() + timestamp = time.time() + + try: + trace_id = engine.record_trace( + type="llm_call", + name="llm_retry", + payload=payload, + step_id=step_id, + parent_trace_id=parent_trace_id, + status="retrying", + ) + except Exception as trace_error: + llm_logger.warning("Failed to persist LLM retry trace: %s", trace_error) + trace_id = None + + try: + engine._publish( + "llm_retrying", + { + **payload, + "step_id": str(step_id) if step_id else None, + "trace_id": str(trace_id) if trace_id else None, + "timestamp": timestamp, + }, + ) + except Exception as publish_error: + # Retry visibility is best-effort and must never break the LLM retry itself. + llm_logger.warning("Failed to publish LLM retry event: %s", publish_error) + + +class _ProviderRetryTelemetryHandler(logging.Handler): + """Turn provider SDK retry log records into structured Artemis telemetry.""" + + _DELAY_PATTERN = re.compile(r"\bin\s+(?P\d+(?:\.\d+)?)\s+seconds?\b", re.IGNORECASE) + _ERROR_PATTERN = re.compile(r"\bas it raised\s+(?P.+)$", re.IGNORECASE) + _RETRYABLE_MARKERS = ( + "503", + "429", + "unavailable", + "resource_exhausted", + "high demand", + "quota", + "throttled", + "overloaded", + ) + + def emit(self, record: logging.LogRecord) -> None: + try: + message = record.getMessage() + lowered = message.lower() + if "retrying" not in lowered or not any( + marker in lowered for marker in self._RETRYABLE_MARKERS + ): + return + + delay_match = self._DELAY_PATTERN.search(message) + error_match = self._ERROR_PATTERN.search(message) + delay = float(delay_match.group("delay")) if delay_match else 0.0 + error = error_match.group("error").strip() if error_match else message + _record_llm_retry( + error, + delay, + provider="google", + source="provider_sdk", + ) + except Exception: + # Telemetry must never interfere with the provider's own retry. + self.handleError(record) + + +def _install_provider_retry_telemetry() -> None: + provider_logger = logging.getLogger("google_genai._api_client") + if any( + isinstance(handler, _ProviderRetryTelemetryHandler) + for handler in provider_logger.handlers + ): + return + provider_logger.addHandler(_ProviderRetryTelemetryHandler(level=logging.INFO)) + + +_install_provider_retry_telemetry() + def _handle_llm_pause_and_resume(last_error: Exception) -> Path: """Handles pausing task execution upon persistent failure until resumed.""" @@ -67,13 +193,14 @@ def _handle_llm_pause_and_resume(last_error: Exception) -> Path: except Exception: pass - if _CURRENT_DATA_ENGINE: - step_id = getattr(_CURRENT_DATA_ENGINE, "current_step_id", None) + current_engine = _get_current_data_engine() + if current_engine: + step_id = getattr(current_engine, "current_step_id", None) timestamp = time.time() try: # Persist exhausted LLM retries as a failed call so the same error # card is available both in the live stream and after a refresh. - _CURRENT_DATA_ENGINE.record_trace( + current_engine.record_trace( type="llm_call", name="llm_pause", payload={"error": err_msg, "pause": True}, @@ -83,7 +210,7 @@ def _handle_llm_pause_and_resume(last_error: Exception) -> Path: except Exception as trace_error: llm_logger.warning("Failed to persist paused LLM error trace: %s", trace_error) - _CURRENT_DATA_ENGINE._publish( + current_engine._publish( "task_paused", { "error": err_msg, @@ -129,16 +256,13 @@ def robust_retry_async(func: Callable) -> Callable: if retry_delay > 0: llm_logger.warning(f"LLM call failed, retrying in {retry_delay:.2f}s...") - if _CURRENT_DATA_ENGINE: - _CURRENT_DATA_ENGINE._publish( - "llm_retrying", - { - "attempt": attempt + 1, - "max_retries": max_retries, - "delay": retry_delay, - "error": str(e), - }, - ) + _record_llm_retry( + str(e), + retry_delay, + attempt=attempt + 1, + max_retries=max_retries, + source="artemis_wrapper", + ) await asyncio.sleep(retry_delay) pause_file = _handle_llm_pause_and_resume(last_error) @@ -147,8 +271,9 @@ def robust_retry_async(func: Callable) -> Callable: llm_logger.info("Resume signal received, retrying LLM call...") - if _CURRENT_DATA_ENGINE: - _CURRENT_DATA_ENGINE._publish("task_resumed", {}) + current_engine = _get_current_data_engine() + if current_engine: + current_engine._publish("task_resumed", {}) return wrapper @@ -194,16 +319,13 @@ def robust_retry_astream(func: Callable) -> Callable: f"LLM stream handshake throttled, retrying in {retry_delay:.2f}s..." ) - if _CURRENT_DATA_ENGINE: - _CURRENT_DATA_ENGINE._publish( - "llm_retrying", - { - "attempt": attempt + 1, - "max_retries": max_retries, - "delay": retry_delay, - "error": str(e), - }, - ) + _record_llm_retry( + str(e), + retry_delay, + attempt=attempt + 1, + max_retries=max_retries, + source="artemis_wrapper", + ) await asyncio.sleep(retry_delay) pause_file = _handle_llm_pause_and_resume(last_error) @@ -212,8 +334,9 @@ def robust_retry_astream(func: Callable) -> Callable: llm_logger.info("Resume signal received, retrying LLM stream...") - if _CURRENT_DATA_ENGINE: - _CURRENT_DATA_ENGINE._publish("task_resumed", {}) + current_engine = _get_current_data_engine() + if current_engine: + current_engine._publish("task_resumed", {}) return wrapper diff --git a/mcp_server/tools/device_state.py b/mcp_server/tools/device_state.py index d8366f2..3cc631f 100644 --- a/mcp_server/tools/device_state.py +++ b/mcp_server/tools/device_state.py @@ -26,7 +26,7 @@ from artemis.utils.ocr_xml_fusion import ( _crop_image_remove_status_bar, _map_coordinates_back, ) -from artemis.utils.ocr_api import perform_ocr +from artemis.utils.ocr_api import is_ocr_configured, perform_ocr from artemis.utils.visualization import format_minimal_list_with_elements @@ -77,19 +77,25 @@ async def mobile_get_device_state(view_type: str) -> str: return f"file://{screenshot_path}" elif view_type == "hierarchy": - try: - screen_height = device_data.height - status_bar_height = _detect_status_bar_height(xml_hierarchy, screen_height) - if status_bar_height > 0: - cropped_b64, _, _ = _crop_image_remove_status_bar( - latest_screenshot_b64, status_bar_height + ocr_results = [] + if is_ocr_configured(): + try: + screen_height = device_data.height + status_bar_height = _detect_status_bar_height( + xml_hierarchy, screen_height ) - raw_ocr_results = await perform_ocr(cropped_b64) - ocr_results = _map_coordinates_back(raw_ocr_results, status_bar_height) - else: - ocr_results = await perform_ocr(latest_screenshot_b64) - except Exception: - ocr_results = [] + if status_bar_height > 0: + cropped_b64, _, _ = _crop_image_remove_status_bar( + latest_screenshot_b64, status_bar_height + ) + raw_ocr_results = await perform_ocr(cropped_b64) + ocr_results = _map_coordinates_back( + raw_ocr_results, status_bar_height + ) + else: + ocr_results = await perform_ocr(latest_screenshot_b64) + except Exception: + ocr_results = [] fused_xml = fuse_ocr_with_xml(xml_hierarchy, ocr_results) hierarchy_text, elements, labels = format_minimal_list_with_elements( diff --git a/tests/unit/admin_console/test_step_repository.py b/tests/unit/admin_console/test_step_repository.py index dbef13e..6fb872b 100644 --- a/tests/unit/admin_console/test_step_repository.py +++ b/tests/unit/admin_console/test_step_repository.py @@ -43,3 +43,33 @@ def test_unrelated_log_is_not_presented_as_llm_failure(): assert normalized["type"] == "log" assert normalized["status"] == "success" + + +def test_retrying_llm_payload_keeps_user_visible_retry_details(): + repository = StepRepository() + trace = { + "trace_id": "retry-1", + "type": "llm_call", + "name": "llm_retry", + "status": "retrying", + "payload": json.dumps( + { + "error": "503 UNAVAILABLE: high demand", + "delay": 1.18, + "provider": "google", + "source": "provider_sdk", + "recoverable": True, + "internal_noise": "do not expose", + } + ), + } + + normalized = repository._normalize_display_trace(trace) + + assert normalized["payload"] == { + "error": "503 UNAVAILABLE: high demand", + "delay": 1.18, + "provider": "google", + "source": "provider_sdk", + "recoverable": True, + } diff --git a/tests/unit/admin_console/test_task_queue_service.py b/tests/unit/admin_console/test_task_queue_service.py index 54d7c69..1df7542 100644 --- a/tests/unit/admin_console/test_task_queue_service.py +++ b/tests/unit/admin_console/test_task_queue_service.py @@ -14,6 +14,9 @@ import asyncio import importlib +import json +import os +import subprocess import sys from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -203,9 +206,11 @@ async def test_queue_worker_cmd_construction(): assert "--app-path" in cmd assert "/path/to/app.apk" in cmd if sys.platform == "win32": - import subprocess - - assert executed_kwargs[0]["creationflags"] == subprocess.CREATE_NEW_PROCESS_GROUP + assert executed_kwargs[0]["creationflags"] == ( + subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW + ) + assert executed_kwargs[0]["stdout"] == asyncio.subprocess.PIPE + assert executed_kwargs[0]["stderr"] == asyncio.subprocess.STDOUT else: assert "creationflags" not in executed_kwargs[0] @@ -218,6 +223,40 @@ async def test_queue_worker_cmd_construction(): pass +@pytest.mark.asyncio +async def test_forward_worker_output_preserves_split_utf8(capsys): + stream = asyncio.StreamReader() + encoded = "worker 输出正常\n".encode() + stream.feed_data(encoded[:9]) + stream.feed_data(encoded[9:]) + stream.feed_eof() + + await TaskQueueService._forward_worker_output(stream) + + assert capsys.readouterr().out == "worker 输出正常\n" + + +@pytest.mark.skipif(sys.platform != "win32", reason="Windows console isolation only") +def test_windows_worker_has_no_inherited_console(): + probe = r""" +import ctypes +import json + +buffer = (ctypes.c_uint32 * 32)() +count = ctypes.windll.kernel32.GetConsoleProcessList(buffer, len(buffer)) +print(json.dumps(list(buffer[:min(count, len(buffer))]))) +""" + result = subprocess.run( + [sys.executable, "-c", probe], + check=True, + text=True, + **TaskQueueService._subprocess_creation_kwargs(), + ) + + attached_processes = json.loads(result.stdout) + assert os.getpid() not in attached_processes + + @pytest.mark.asyncio async def test_stop_tasks_dead_process_resets_state(): state.queue_items = [ diff --git a/tests/unit/mcp/test_mcp_tools.py b/tests/unit/mcp/test_mcp_tools.py index a73ee10..acaf3fd 100644 --- a/tests/unit/mcp/test_mcp_tools.py +++ b/tests/unit/mcp/test_mcp_tools.py @@ -17,7 +17,10 @@ import inspect import shutil import tempfile +from types import SimpleNamespace import uuid +from unittest.mock import AsyncMock, MagicMock, patch + import pytest from mcp_server.tools import ( @@ -106,6 +109,40 @@ def test_mobile_manage_task_inject_instruction(temp_trace_env): assert "Successfully injected" in ok_res["message"] +@pytest.mark.asyncio +async def test_mobile_get_device_state_hierarchy_without_ocr(): + """XML hierarchy remains available when OCR is not configured.""" + controller = MagicMock() + controller.ctx.device.device_width = 1080 + controller.ctx.device.device_height = 2400 + controller.get_screen_data = AsyncMock( + return_value=SimpleNamespace( + base64="ZHVtbXk=", + elements=[ + { + "class": "android.widget.Button", + "text": "Start", + "bounds": "[100,500][400,600]", + } + ], + width=1080, + height=2400, + ) + ) + + with ( + patch("mcp_server.tools.device_state._get_controller", return_value=controller), + patch("mcp_server.tools.device_state.is_ocr_configured", return_value=False), + patch( + "mcp_server.tools.device_state.perform_ocr", new_callable=AsyncMock + ) as perform_ocr_mock, + ): + result = await mobile_get_device_state(view_type="hierarchy") + + assert "Start" in result + perform_ocr_mock.assert_not_awaited() + + @pytest.mark.asyncio async def test_mobile_inspect_trace_invalid_action(): res = await mobile_inspect_trace(action="invalid_action", trace_id="trace-123") diff --git a/tests/unit/test_drivers.py b/tests/unit/test_drivers.py index 9c38a10..aa35bcb 100644 --- a/tests/unit/test_drivers.py +++ b/tests/unit/test_drivers.py @@ -14,14 +14,24 @@ """Unit tests for Unified Base Device Driver and implementations.""" -import pytest +import base64 +from types import SimpleNamespace from unittest.mock import MagicMock +import pytest + from artemis.drivers.base import KeyCode, ScreenData, SwipeDirection from artemis.drivers.mock.mock_driver import MockDeviceDriver from artemis.drivers.android.adb_driver import AndroidAdbDriver +_ONE_PIXEL_PNG = base64.b64encode( + base64.b64decode( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==" + ) +).decode("ascii") + + @pytest.mark.asyncio async def test_mock_driver_actions(): """Verify MockDeviceDriver records and performs all core actions.""" @@ -89,3 +99,70 @@ async def test_android_driver_with_mock_adb(): # Test press key await driver.press_key(KeyCode.BACK) mock_adb_device.shell.assert_called_with("input keyevent 4") + + +@pytest.mark.asyncio +async def test_android_driver_preserves_elements_from_combined_screen_data(): + """The combined uiautomator2 response is the authoritative live snapshot.""" + mock_adb_client = MagicMock() + mock_ui_client = MagicMock() + expected_elements = [ + { + "text": "", + "class": "android.widget.FrameLayout", + "bounds": "[0,0][1080,2400]", + "clickable": "false", + "focusable": "false", + }, + { + "text": "Start", + "class": "android.widget.Button", + "bounds": "[10,20][100,80]", + } + ] + mock_ui_client.get_screen_data.return_value = SimpleNamespace( + base64=_ONE_PIXEL_PNG, + hierarchy_xml="", + elements=expected_elements, + width=1080, + height=2400, + ) + + driver = AndroidAdbDriver( + device_id="emulator-5554", + adb_client=mock_adb_client, + ui_adb_client=mock_ui_client, + ) + + screen_data = await driver.get_screen_data(skip_settling=True) + + assert screen_data.ui_elements == [expected_elements[1]] + assert screen_data.ui_hierarchy_xml == "" + mock_ui_client.get_hierarchy.assert_not_called() + + +@pytest.mark.asyncio +async def test_android_driver_parses_xml_hierarchy_fallback(): + """XML-only clients remain supported without discarding their hierarchy.""" + mock_adb_client = MagicMock() + mock_ui_client = MagicMock() + mock_ui_client.get_screen_data.return_value = SimpleNamespace( + base64=_ONE_PIXEL_PNG, + width=1080, + height=2400, + ) + mock_ui_client.get_hierarchy.return_value = ( + '' + ) + + driver = AndroidAdbDriver( + device_id="emulator-5554", + adb_client=mock_adb_client, + ui_adb_client=mock_ui_client, + ) + + screen_data = await driver.get_screen_data(skip_settling=True) + + assert screen_data.ui_hierarchy_xml is not None + assert any(element.get("text") == "Settings" for element in screen_data.ui_elements) diff --git a/tests/unit/test_llm_pause.py b/tests/unit/test_llm_pause.py index e9b8769..e995181 100644 --- a/tests/unit/test_llm_pause.py +++ b/tests/unit/test_llm_pause.py @@ -1,8 +1,11 @@ +import logging + from artemis.services import llm class _RecordingEngine: current_step_id = "step-7" + current_session_id = "session-1" def __init__(self): self.trace_kwargs = None @@ -10,6 +13,7 @@ class _RecordingEngine: def record_trace(self, **kwargs): self.trace_kwargs = kwargs + return "trace-1" def _publish(self, event_type, payload): self.published = (event_type, payload) @@ -35,3 +39,42 @@ def test_pause_error_is_persisted_and_published(monkeypatch, tmp_path): assert payload["error"] == "503 high demand" assert payload["step_id"] == "step-7" assert isinstance(payload["timestamp"], float) + + +def test_provider_sdk_retry_is_persisted_and_published(monkeypatch): + engine = _RecordingEngine() + monkeypatch.setattr(llm, "_CURRENT_DATA_ENGINE", engine) + record = logging.LogRecord( + name="google_genai._api_client", + level=logging.INFO, + pathname=__file__, + lineno=1, + msg=( + "Retrying google.genai request in 1.18 seconds as it raised " + "ServerError: 503 UNAVAILABLE: high demand" + ), + args=(), + exc_info=None, + ) + + llm._ProviderRetryTelemetryHandler().emit(record) + + assert engine.trace_kwargs == { + "type": "llm_call", + "name": "llm_retry", + "payload": { + "error": "ServerError: 503 UNAVAILABLE: high demand", + "delay": 1.18, + "provider": "google", + "source": "provider_sdk", + "recoverable": True, + }, + "step_id": "step-7", + "parent_trace_id": None, + "status": "retrying", + } + event_type, payload = engine.published + assert event_type == "llm_retrying" + assert payload["trace_id"] == "trace-1" + assert payload["delay"] == 1.18 + assert payload["error"].startswith("ServerError: 503")