From 0f5fb1d32c5aafe948fe4029a918c0416297a8c6 Mon Sep 17 00:00:00 2001 From: somew1nd Date: Thu, 20 Aug 2026 00:37:53 -0700 Subject: [PATCH] Improve task resilience and LLM retry visibility Persist and publish recoverable provider retries so temporary 503 and throttling events remain visible in live and historical task streams. Aggregate repeated retries into an expandable action card, preserve terminal pause recovery, harden Windows worker isolation and output forwarding, and retain reliable Android hierarchy fallbacks with regression coverage. --- .../database/repositories/step_repository.py | 23 +- .../services/task_queue_service.py | 66 +++++- .../agent-stream/agent-stream.component.html | 81 ++++++- .../agent-stream/agent-stream.component.scss | 218 ++++++++++++++++-- .../agent-stream/agent-stream.component.ts | 58 ++++- .../src/app/services/agent.service.ts | 9 +- .../src/app/utils/stream-aggregator.util.ts | 5 +- .../src/app/utils/tool-formatter.util.spec.ts | 57 +++++ .../src/app/utils/tool-formatter.util.ts | 45 ++++ artemis/drivers/android/adb_driver.py | 34 ++- artemis/services/llm.py | 181 ++++++++++++--- mcp_server/tools/device_state.py | 32 +-- .../admin_console/test_step_repository.py | 30 +++ .../admin_console/test_task_queue_service.py | 45 +++- tests/unit/mcp/test_mcp_tools.py | 37 +++ tests/unit/test_drivers.py | 79 ++++++- tests/unit/test_llm_pause.py | 43 ++++ 17 files changed, 946 insertions(+), 97 deletions(-) create mode 100644 apps/showcase_ui/src/app/utils/tool-formatter.util.spec.ts 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")