diff --git a/.env.example b/.env.example index d2ee912..6a7a36d 100644 --- a/.env.example +++ b/.env.example @@ -30,6 +30,16 @@ ARTEMIS_SCREENSHOT_LIMIT_ENABLED=true ARTEMIS_OPERATOR_MAX_NO_PROGRESS_ATTEMPTS=3 # Flash 连续无工具调用的最大次数；无效回复不会重复送回模型。 ARTEMIS_FLASH_MAX_NO_PROGRESS_ATTEMPTS=3 +# 所有网页/MCP 任务的总执行时限；排队、用户暂停不计入。任务 JSON 的 timeout_seconds 可覆盖。 +ARTEMIS_TASK_TIMEOUT_SECONDS=1800 +# 单次屏幕观测预算、最多恢复次数和备用 ADB 命令时限。 +ARTEMIS_OBSERVATION_TIMEOUT_SECONDS=60 +ARTEMIS_OBSERVATION_MAX_ATTEMPTS=3 +ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS=10 +# Flash 连续设备动作失败后，重新观察并切换兜底模型；兜底仍失败时结束。 +ARTEMIS_ACTION_FAILURE_ATTEMPTS=3 +# 已写出任务结果后，后台线程/录屏清理最多等待多久；超时回收进程释放队列。 +ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS=30 # 视频分段、重试与单次分析超时（秒） ARTEMIS_VIDEO_CHUNK_SECONDS=60 @@ -56,6 +66,5 @@ ARTEMIS_HELPER_AUTO_INSTALL=true ARTEMIS_DEFAULT_PROFILE=flash # 批量任务与 Excel 导入限制 -ARTEMIS_BATCH_TASK_TIMEOUT=1800 ARTEMIS_IMPORT_MAX_ROWS=1000 ARTEMIS_IMPORT_MAX_BYTES=10485760 diff --git a/apps/admin_console/routers/automation.py b/apps/admin_console/routers/automation.py index 4167a09..a9e2006 100644 --- a/apps/admin_console/routers/automation.py +++ b/apps/admin_console/routers/automation.py @@ -12,6 +12,7 @@ from apps.admin_console.services.automation_store import identifier, store from apps.admin_console.services import automation_service as service from apps.admin_console.services import batch_archive as archive_service from apps.admin_console.services import report_export, task_import +from artemis.config.settings import settings router = APIRouter(prefix="/api/automation", tags=["automation"]) @@ -30,7 +31,7 @@ class BatchRequest(ReportPreferences): profile: Literal["flash", "pro"] = "flash" stop_on_failure: bool = False timeout_seconds: int = Field( - default_factory=lambda: int(os.getenv("ARTEMIS_BATCH_TASK_TIMEOUT", "1800")), + default_factory=lambda: settings.ARTEMIS_TASK_TIMEOUT_SECONDS, ge=30, le=86400, ) @@ -50,7 +51,7 @@ def require(kind, key): def preferences(): return { **store.preferences(), - "batch_timeout_seconds": int(os.getenv("ARTEMIS_BATCH_TASK_TIMEOUT", "1800")), + "batch_timeout_seconds": settings.ARTEMIS_TASK_TIMEOUT_SECONDS, } diff --git a/apps/admin_console/routers/tasks.py b/apps/admin_console/routers/tasks.py index 58eea87..3818dfe 100644 --- a/apps/admin_console/routers/tasks.py +++ b/apps/admin_console/routers/tasks.py @@ -202,7 +202,9 @@ async def run_task(request: RunRequest): # With no explicit serial the probe itself resolves a live target (it # prefers the diagnostics target preference, then any unlocked ready # device); the verified serial is bound below. - target_serial = request.device_serial + target_serial = ( + request.runtime_overrides.device_serial if request.runtime_overrides else None + ) or request.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 = ( @@ -224,7 +226,7 @@ async def run_task(request: RunRequest): # Only auto-selected targets may be re-bound to the probed device. An # explicitly requested serial is never silently replaced -- if it is # invalid, enqueue_tasks rejects the submission with a clear error. - if verified_serial and not request.device_serial: + if verified_serial and not target_serial: target_serial = verified_serial response = await task_queue_service.enqueue_tasks( @@ -240,6 +242,9 @@ async def run_task(request: RunRequest): ingress=request.ingress or "frontend", session_id=request.session_id, conversation_id=request.conversation_id, + runtime_overrides=request.runtime_overrides.model_dump(exclude_none=True) + if request.runtime_overrides + else None, ) from apps.admin_console.services.automation_service import register_submission diff --git a/apps/admin_console/schemas/task_schema.py b/apps/admin_console/schemas/task_schema.py index 144a972..9ceeb90 100644 --- a/apps/admin_console/schemas/task_schema.py +++ b/apps/admin_console/schemas/task_schema.py @@ -13,6 +13,7 @@ # limitations under the License. from pydantic import BaseModel +from apps.admin_console.services.task_import import TaskOptions class RunRequest(BaseModel): @@ -32,6 +33,7 @@ class RunRequest(BaseModel): ingress: str | None = "frontend" session_id: str | None = None conversation_id: str | None = None + runtime_overrides: TaskOptions | None = None class ReplayRequest(BaseModel): diff --git a/apps/admin_console/services/automation_service.py b/apps/admin_console/services/automation_service.py index 205de51..6d32584 100644 --- a/apps/admin_console/services/automation_service.py +++ b/apps/admin_console/services/automation_service.py @@ -10,6 +10,8 @@ from uuid import UUID from apps.admin_console.core.state import state from apps.admin_console.services.automation_store import store, new_id from apps.admin_console.services.task_import import case_goal, case_title +from artemis.config.settings import settings +from artemis.runtime.task_limits import task_timeout logger = logging.getLogger(__name__) TERMINAL = {"completed", "failed", "cancelled", "skipped", "interrupted"} @@ -135,10 +137,11 @@ async def tick_batches(): # not consume this case's execution budget. if job["status"] != "running": continue - limit = int( - (job.get("case", {}).get("task_config") or {}).get("timeout_seconds") - or batch.get("timeout_seconds", 1800) - ) + from artemis.runtime.task_result import read_worker_outcome + + if read_worker_outcome(str(job["session_id"])): + continue # Execution has ended; the worker supervisor bounds cleanup. + limit = task_timeout({}, job, batch) task_state = getattr(state, "active_runs", {}).get(str(job["session_id"])) or {} excluded = float(task_state.get("non_execution_seconds") or 0) if task_state.get("non_execution_started_at"): @@ -151,7 +154,17 @@ async def tick_batches(): error=f"任务执行超时（{limit} 秒，不含暂停/排队时间）", timed_out=True ) store.save_job(job) - task_queue_service.stop_tasks(session_id=job["session_id"]) + from artemis.runtime.task_result import save_failure_result + + try: + await asyncio.to_thread( + save_failure_result, + str(job["session_id"]), + job["error"], + code="task_timeout", + ) + finally: + task_queue_service.stop_tasks(session_id=job["session_id"]) # on_terminal records the failed case. The next pending # case then runs normally unless stop_on_failure is set. continue @@ -255,8 +268,7 @@ def create_batch(payload: dict) -> dict: "formats": payload.get("formats") or store.preferences()["formats"], "profile": payload.get("profile", "flash"), "device_serial": payload.get("device_serial"), - "timeout_seconds": payload.get("timeout_seconds") - or int(os.getenv("ARTEMIS_BATCH_TASK_TIMEOUT", "1800")), + "timeout_seconds": payload.get("timeout_seconds") or settings.ARTEMIS_TASK_TIMEOUT_SECONDS, "stop_on_failure": payload.get("stop_on_failure", False), "request_id": payload.get("request_id"), } diff --git a/apps/admin_console/services/model_settings_service.py b/apps/admin_console/services/model_settings_service.py index 0eda476..a94ada3 100644 --- a/apps/admin_console/services/model_settings_service.py +++ b/apps/admin_console/services/model_settings_service.py @@ -172,6 +172,24 @@ class CapabilitySettings(BaseModel): step_summarizer_enabled: bool = True screenshot_limit_enabled: bool = True screenshot_max_long_edge: int = 2048 + task_timeout_seconds: int = Field( + default_factory=lambda: settings.ARTEMIS_TASK_TIMEOUT_SECONDS, ge=30, le=86400 + ) + observation_timeout_seconds: float = Field( + default_factory=lambda: settings.ARTEMIS_OBSERVATION_TIMEOUT_SECONDS, gt=0, le=300 + ) + observation_max_attempts: int = Field( + default_factory=lambda: settings.ARTEMIS_OBSERVATION_MAX_ATTEMPTS, ge=1, le=5 + ) + device_command_timeout_seconds: float = Field( + default_factory=lambda: settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS, gt=0, le=60 + ) + action_failure_attempts: int = Field( + default_factory=lambda: settings.ARTEMIS_ACTION_FAILURE_ATTEMPTS, ge=1, le=10 + ) + worker_cleanup_timeout_seconds: float = Field( + default_factory=lambda: settings.ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS, gt=0, le=300 + ) def get_capability_settings() -> CapabilitySettings: @@ -193,6 +211,12 @@ def save_capability_settings(update: CapabilitySettings) -> CapabilitySettings: "ARTEMIS_OUTPUTTER_ENABLED": update.outputter_enabled, "ARTEMIS_STEP_SUMMARIZER_ENABLED": update.step_summarizer_enabled, "ARTEMIS_SCREENSHOT_LIMIT_ENABLED": update.screenshot_limit_enabled, + "ARTEMIS_TASK_TIMEOUT_SECONDS": update.task_timeout_seconds, + "ARTEMIS_OBSERVATION_TIMEOUT_SECONDS": update.observation_timeout_seconds, + "ARTEMIS_OBSERVATION_MAX_ATTEMPTS": update.observation_max_attempts, + "ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS": update.device_command_timeout_seconds, + "ARTEMIS_ACTION_FAILURE_ATTEMPTS": update.action_failure_attempts, + "ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS": update.worker_cleanup_timeout_seconds, } persist_environment({key: str(value).lower() for key, value in values.items()}) for key, value in values.items(): diff --git a/apps/admin_console/services/report_export.py b/apps/admin_console/services/report_export.py index 3d23d14..50d595f 100644 --- a/apps/admin_console/services/report_export.py +++ b/apps/admin_console/services/report_export.py @@ -95,6 +95,11 @@ def failure_reason(directory: Path, status: str, recorded: str | None) -> str: state = read_json(directory / "status.json", {}) if isinstance(state, dict) and state.get("error"): return redact(state["error"]) + result = read_json(directory / "result.json", {}) + if isinstance(result, dict) and result.get("status") == "failed": + reason = result.get("error") or result.get("explanation") + if reason: + return redact(reason) log_lines: list[str] = [] for name in ("stdout.log", "stderr.log"): path = directory / name diff --git a/apps/admin_console/services/task_import.py b/apps/admin_console/services/task_import.py index 6567d42..eab390c 100644 --- a/apps/admin_console/services/task_import.py +++ b/apps/admin_console/services/task_import.py @@ -42,6 +42,11 @@ class TaskOptions(BaseModel): max_retries: int | None = Field(default=None, ge=0, le=10) retry_interval_seconds: float | None = Field(default=None, ge=0, le=300) timeout_seconds: int | None = Field(default=None, ge=30, le=86400) + observation_timeout_seconds: float | None = Field(default=None, gt=0, le=300) + observation_max_attempts: int | None = Field(default=None, ge=1, le=5) + device_command_timeout_seconds: float | None = Field(default=None, gt=0, le=60) + action_failure_attempts: int | None = Field(default=None, ge=1, le=10) + worker_cleanup_timeout_seconds: float | None = Field(default=None, gt=0, le=300) class TaskCase(BaseModel): diff --git a/apps/admin_console/services/task_queue_service.py b/apps/admin_console/services/task_queue_service.py index b407524..d55349c 100644 --- a/apps/admin_console/services/task_queue_service.py +++ b/apps/admin_console/services/task_queue_service.py @@ -581,6 +581,10 @@ class TaskQueueService: "retry_interval_seconds": "OPENAI_RETRY_INTERVAL_SECONDS", "record_video": "ARTEMIS_RECORDING_ENABLED", "analyze_video": "ARTEMIS_VIDEO_ENABLED", + "observation_timeout_seconds": "ARTEMIS_OBSERVATION_TIMEOUT_SECONDS", + "observation_max_attempts": "ARTEMIS_OBSERVATION_MAX_ATTEMPTS", + "device_command_timeout_seconds": "ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS", + "action_failure_attempts": "ARTEMIS_ACTION_FAILURE_ATTEMPTS", }.items(): if overrides.get(option) is not None: env[variable] = ( @@ -630,6 +634,72 @@ class TaskQueueService: env["RESULTS_OUTPUT_PATH"] = str(TRACES_PATH / str(sess_id) / "result.json") return cmd, env + @classmethod + async def _await_worker_with_budget(cls, proc, task_item: dict, run_key: str) -> int: + from artemis.runtime.task_limits import ExecutionBudget, task_timeout + from artemis.config.settings import settings + from artemis.runtime.task_result import save_failure_result, read_worker_outcome + from artemis.runtime.task_control import read_state + from apps.admin_console.services.automation_store import store + + sid = str(task_item.get("session_id") or run_key) + job = store.job(sid) + batch = store.get("batch", job["batch_id"]) if job and job.get("batch_id") else None + limit = task_timeout(task_item, job, batch) + budget = ExecutionBudget(limit) + cleanup_limit = (task_item.get("runtime_overrides") or {}).get( + "worker_cleanup_timeout_seconds" + ) or settings.ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS + finalized_at = None + waiter = asyncio.create_task(cls._wait_for_worker_process(proc)) + try: + while True: + done, _ = await asyncio.wait({waiter}, timeout=0.25) + if waiter in done: + return waiter.result() + outcome = read_worker_outcome(sid) + if outcome: + finalized_at = finalized_at if finalized_at is not None else time.monotonic() + if time.monotonic() - finalized_at >= cleanup_limit: + logger.warning( + "%s: task finalized but worker cleanup exceeded %s seconds; reclaiming process", + sid, + cleanup_limit, + ) + await cls._terminate_worker_process(proc) + return await waiter + continue # Final result is authoritative; only bound resource cleanup now. + cls._sync_control_states() + run = state.active_runs.get(run_key, {}) + remaining = budget.update( + excluded=bool(run.get("parked")) or read_state(sid).get("status") == "queued" + ) + if remaining > 0: + continue + error = f"任务执行超时（{limit} 秒，不含暂停/排队时间）" + task_item.update(timed_out=True, error=error) + logger.error("%s: %s", sid, error) + try: + latest = store.job(sid) + if latest: + latest.update(timed_out=True, error=error) + store.save_job(latest) + except Exception: + logger.exception( + "Cannot persist timeout for %s; terminating worker anyway", sid + ) + await cls._terminate_worker_process(proc) + returncode = await waiter + try: + await asyncio.to_thread(save_failure_result, sid, error, code="task_timeout") + except OSError: + logger.exception("Cannot save timeout report for %s", sid) + return returncode + finally: + if not waiter.done(): + waiter.cancel() + await asyncio.gather(waiter, return_exceptions=True) + @classmethod def _register_worker_run( cls, @@ -721,6 +791,10 @@ class TaskQueueService: returncode, manual_stop, ) + from apps.admin_console.services.automation_store import store + + if (store.job(str(sess_id)) or {}).get("timed_out"): + new_status, should_persist = "failed", True if ( state.is_shutting_down and not manual_stop @@ -943,7 +1017,7 @@ class TaskQueueService: await cls._terminate_if_cancelled_during_launch(run_key, sess_id, proc) # 3. Await subprocess completion - returncode = await cls._wait_for_worker_process(proc) + returncode = await cls._await_worker_with_budget(proc, task_item, run_key) print(f"[QueueWorker] Task [{sess_id}] exited with returncode {returncode}") manual_stop = run_key in state.manually_stopped_run_ids or bool( @@ -1141,6 +1215,12 @@ class TaskQueueService: forwarded to the worker as ``--verification-level`` / ``--explorer-pro-mode``; they are normalised here so the queue item and the CLI see one spelling. """ + if runtime_overrides: + profile = runtime_overrides.get("profile") or profile + device_serial = runtime_overrides.get("device_serial") or device_serial + locked_app_package = runtime_overrides.get("locked_app_package") or locked_app_package + verification_level = runtime_overrides.get("verification_level") or verification_level + explorer_mode = runtime_overrides.get("explorer_mode") or explorer_mode verification_level = ( str(verification_level).strip().lower() or None if verification_level else None ) diff --git a/apps/showcase_ui/src/app/components/model-settings/capability-settings.component.ts b/apps/showcase_ui/src/app/components/model-settings/capability-settings.component.ts index e7a3c04..652b012 100644 --- a/apps/showcase_ui/src/app/components/model-settings/capability-settings.component.ts +++ b/apps/showcase_ui/src/app/components/model-settings/capability-settings.component.ts @@ -10,6 +10,12 @@ interface Capabilities { step_summarizer_enabled: boolean; screenshot_limit_enabled: boolean; screenshot_max_long_edge: number; + task_timeout_seconds: number; + observation_timeout_seconds: number; + observation_max_attempts: number; + device_command_timeout_seconds: number; + action_failure_attempts: number; + worker_cleanup_timeout_seconds: number; } @Component({ @@ -28,7 +34,19 @@ interface Capabilities { 步骤摘要压缩Flash 模式下在后台生成摘要，减少长任务的上下文占用。 - +
+ 执行保护 +

限制故障等待与反复失败；保存后对新任务生效，任务 JSON 可单独覆盖。

+
+ + + + + + +
+
+ @if (message()) {

{{ message() }}

} @if (error()) { } @@ -38,12 +56,18 @@ interface Capabilities { label { display: flex; gap: 12px; align-items: flex-start; padding: 14px 0; cursor: pointer; } input { width: 20px; height: 20px; accent-color: #0284c7; flex-shrink: 0; } small { display: block; margin-top: 4px; } button { padding: 10px 18px; border-radius: 10px; border: 1px solid #38bdf877; background: #0284c7; color: white; font: inherit; cursor: pointer; } - button:disabled { opacity: .5; cursor: default; } .error { color: #be123c; }`] + button:disabled { opacity: .5; cursor: default; } .error { color: #be123c; } + .execution-settings { margin: 12px 0 20px; padding: 12px 14px; border: 1px solid #cbd5e1; border-radius: 10px; background: #f8fafc; } + summary { cursor: pointer; font-weight: 600; color: #334155; } + .number-fields { display: grid; grid-template-columns: 1fr 1fr; gap: 12px 16px; } + .number-fields label { display: block; padding: 0; color: #334155; } + .number-fields input { box-sizing: border-box; display: block; width: 100%; height: auto; margin-top: 8px; padding: 9px; border: 1px solid #cbd5e1; border-radius: 8px; font: inherit; } + @media(max-width: 480px) { .number-fields { grid-template-columns: 1fr; } }`] }) export class CapabilitySettingsComponent implements OnInit { changed = output(); private http = inject(HttpClient); - config: Capabilities = { video_enabled: false, recording_enabled: false, outputter_enabled: true, step_summarizer_enabled: true, screenshot_limit_enabled: true, screenshot_max_long_edge: 2048 }; + config: Capabilities = { video_enabled: false, recording_enabled: false, outputter_enabled: true, step_summarizer_enabled: true, screenshot_limit_enabled: true, screenshot_max_long_edge: 2048, task_timeout_seconds: 1800, observation_timeout_seconds: 60, observation_max_attempts: 3, device_command_timeout_seconds: 10, action_failure_attempts: 3, worker_cleanup_timeout_seconds: 30 }; busy = signal(false); loading = signal(true); loaded = signal(false); message = signal(''); error = signal(''); ngOnInit(): void { this.http.get('/api/system/capability-settings').pipe(finalize(() => this.loading.set(false))).subscribe({ @@ -51,11 +75,20 @@ export class CapabilitySettingsComponent implements OnInit { }); } save(): void { - if (this.busy() || this.loading() || !this.loaded()) return; + if (this.busy() || this.loading() || !this.loaded() || !this.valid()) return; this.busy.set(true); this.message.set(''); this.error.set(''); this.http.post('/api/system/capability-settings', this.config).pipe(finalize(() => this.busy.set(false))).subscribe({ next: config => { this.config = config; this.changed.emit(config); this.message.set('功能设置已保存，新任务将使用此配置。'); }, error: () => this.error.set('保存失败，请检查服务器连接和 .env 文件权限。') }); } + valid(): boolean { + const c = this.config; + return Number.isInteger(c.task_timeout_seconds) && c.task_timeout_seconds >= 30 && c.task_timeout_seconds <= 86400 + && c.observation_timeout_seconds > 0 && c.observation_timeout_seconds <= 300 + && Number.isInteger(c.observation_max_attempts) && c.observation_max_attempts >= 1 && c.observation_max_attempts <= 5 + && c.device_command_timeout_seconds > 0 && c.device_command_timeout_seconds <= 60 + && Number.isInteger(c.action_failure_attempts) && c.action_failure_attempts >= 1 && c.action_failure_attempts <= 10 + && c.worker_cleanup_timeout_seconds > 0 && c.worker_cleanup_timeout_seconds <= 300; + } } diff --git a/apps/showcase_ui/src/app/components/model-settings/model-settings.component.spec.ts b/apps/showcase_ui/src/app/components/model-settings/model-settings.component.spec.ts index 0f62306..b743769 100644 --- a/apps/showcase_ui/src/app/components/model-settings/model-settings.component.spec.ts +++ b/apps/showcase_ui/src/app/components/model-settings/model-settings.component.spec.ts @@ -138,7 +138,7 @@ describe('持久化模型与功能设置', () => { }); it('录屏与视频分析独立保存，重新打开后保留各自状态', () => { - const capabilities = { video_enabled: true, recording_enabled: false, outputter_enabled: true, step_summarizer_enabled: false, screenshot_limit_enabled: true, screenshot_max_long_edge: 2048 }; + const capabilities = { video_enabled: true, recording_enabled: false, outputter_enabled: true, step_summarizer_enabled: false, screenshot_limit_enabled: true, screenshot_max_long_edge: 2048, task_timeout_seconds: 1800, observation_timeout_seconds: 60, observation_max_attempts: 3, device_command_timeout_seconds: 10, action_failure_attempts: 3, worker_cleanup_timeout_seconds: 30 }; const fixture = TestBed.createComponent(CapabilitySettingsComponent); fixture.detectChanges(); http.expectOne('/api/system/capability-settings').flush(capabilities); @@ -160,4 +160,34 @@ describe('持久化模型与功能设置', () => { expect(reopened.componentInstance.config.recording_enabled).toBeTrue(); expect(reopened.componentInstance.config.screenshot_limit_enabled).toBeFalse(); }); + + it('执行保护默认折叠，修改时限可保存且拒绝无效数值', async () => { + const capabilities = { video_enabled: false, recording_enabled: false, outputter_enabled: true, step_summarizer_enabled: true, screenshot_limit_enabled: true, screenshot_max_long_edge: 2048, task_timeout_seconds: 1800, observation_timeout_seconds: 60, observation_max_attempts: 3, device_command_timeout_seconds: 10, action_failure_attempts: 3, worker_cleanup_timeout_seconds: 30 }; + const fixture = TestBed.createComponent(CapabilitySettingsComponent); + fixture.detectChanges(); + http.expectOne('/api/system/capability-settings').flush(capabilities); + fixture.detectChanges(); + await fixture.whenStable(); + const root: HTMLElement = fixture.nativeElement; + expect(root.querySelector('.execution-settings')!.open).toBeFalse(); + root.querySelector('summary')!.click(); + const timeout = root.querySelector('input[type="number"]')!; + timeout.value = '3600'; + timeout.dispatchEvent(new Event('input')); + fixture.detectChanges(); + await fixture.whenStable(); + root.querySelector('button')!.click(); + const request = http.expectOne('/api/system/capability-settings'); + expect(request.request.body.task_timeout_seconds).toBe(3600); + expect(request.request.body.observation_timeout_seconds).toBe(60); + request.flush({ ...capabilities, task_timeout_seconds: 3600 }); + expect(fixture.componentInstance.message()).toContain('已保存'); + timeout.value = '0'; + timeout.dispatchEvent(new Event('input')); + fixture.detectChanges(); + await fixture.whenStable(); + expect(root.querySelector('button')!.disabled).toBeTrue(); + fixture.componentInstance.save(); + http.expectNone('/api/system/capability-settings'); + }); }); diff --git a/artemis/agents/flash/runner.py b/artemis/agents/flash/runner.py index 712514c..3ce2f69 100644 --- a/artemis/agents/flash/runner.py +++ b/artemis/agents/flash/runner.py @@ -81,7 +81,7 @@ from artemis.graph.state import State from artemis.llm.router import ModelEndpoint from artemis.llm.structured import ParseFailure, parse_structured from artemis.mcp.action_executor import McpActionExecutor -from artemis.mcp.observation import observe +from artemis.mcp.observation import DeviceObservationError, require_observation from artemis.memory.transcript import PRO_UI_LIST_MARKER, TranscriptLedger, mark_ephemeral from artemis.services.llm import ( RobustChatModelWrapper, @@ -380,8 +380,16 @@ class FlashRunner: blocks.append( {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"}} ) - if xml_list: - blocks.append({"type": "text", "text": f"{PRO_UI_LIST_MARKER}\n{xml_list}"}) + blocks.append( + { + "type": "text", + "text": f"{PRO_UI_LIST_MARKER}\n" + + ( + xml_list + or "当前没有可用的元素索引。仅根据当前截图定位坐标，禁止使用历史画面的元素编号。" + ), + } + ) ephemeral: list[int] = [] # (text, ephemeral) in tail order. per_turn: list[tuple[str, bool]] = [(text, True) for text in (notices or [])] @@ -604,19 +612,14 @@ class FlashRunner: post_img_bytes = exec_result.screenshot_bytes if post_img_bytes or name not in action_names: return post_img_bytes - try: - obs, screenshot_bytes = await observe(self.ctx, self.controller, settle_ms=0) - except Exception as shot_err: - logger.warning(f"Failed to capture fallback screenshot in FlashRunner: {shot_err}") - return post_img_bytes - if not obs.ok: - logger.warning(f"Failed to capture fallback screenshot in FlashRunner: {obs.message}") - return post_img_bytes + if state is not None: + state.indexed_points = [] + state.indexed_elements = [] + state.latest_screenshot = None + obs, screenshot_bytes = await require_observation(self.ctx, self.controller, settle_ms=0) post_img_bytes = screenshot_bytes - if not exec_result.ui_elements_text: - exec_result.ui_elements_text = obs.elements_text - if not exec_result.screenshot_path and obs.screenshot_path: - exec_result.screenshot_path = obs.screenshot_path + exec_result.ui_elements_text = obs.elements_text + exec_result.screenshot_path = obs.screenshot_path if state is not None: if obs.hierarchy_ok: state.indexed_points = [el["center"] for el in obs.elements] @@ -831,7 +834,7 @@ class FlashRunner: turn.actions.append((name, exec_result.status, exec_result.text_summary)) # ⚡ Non-blocking dispatch of objective visual transition summarizer - if self.summarizer and name in action_names: + if self.summarizer and name in action_names and post_img_bytes: action_sequence += 1 self.summarizer.dispatch( step_number=action_sequence, @@ -857,7 +860,7 @@ class FlashRunner: if post_img_bytes: pre_screenshot_bytes = post_img_bytes - if exec_result.ui_elements_text: + if exec_result.ui_elements_text is not None: xml_list = exec_result.ui_elements_text # Helper tools may return multimodal blocks (a step screenshot); @@ -876,6 +879,8 @@ class FlashRunner: ) ) + except DeviceObservationError: + raise except Exception as e: logger.error(f"Error executing tool {name}: {e}") if name in action_names: @@ -1047,6 +1052,21 @@ class FlashRunner: @trace(type="agent", name="FlashRunner") async def run(self, state: State) -> dict: + try: + return await self._run(state) + except DeviceObservationError as exc: + state.indexed_points = [] + state.indexed_elements = [] + state.latest_screenshot = None + logger.error(f"Flash 因设备观测失败停止：{exc}") + return { + "status": "failed", + "explanation": str(exc), + "error": str(exc), + "code": "device_observation_failed", + } + + async def _run(self, state: State) -> dict: limit = self.turn_limit logger.info( f"Starting Artemis Flash reactive loop for goal: {self.goal}" @@ -1073,6 +1093,7 @@ class FlashRunner: no_tool_limit = settings.ARTEMIS_FLASH_MAX_NO_PROGRESS_ATTEMPTS fallback_llm = None using_tool_fallback = False + action_failure_attempts = 0 while limit is None or turns < limit: control = getattr(self.ctx, "task_control", None) @@ -1084,6 +1105,7 @@ class FlashRunner: if resumed: using_tool_fallback = False no_tool_attempts = 0 + action_failure_attempts = 0 self._llm = llm ( shot_path, @@ -1157,6 +1179,7 @@ class FlashRunner: # assembled before the instruction arrived. using_tool_fallback = False no_tool_attempts = 0 + action_failure_attempts = 0 self._llm = llm ( shot_path, @@ -1241,6 +1264,7 @@ class FlashRunner: fallback_llm = get_llm(self.ctx, name="operator", use_fallback=True) using_tool_fallback = True no_tool_attempts = 0 + action_failure_attempts = 0 logger.warning("Flash 首选模型未调用工具，当前步骤切换到兜底模型") pending_notices.append("首选模型连续未调用工具；本步骤改用兜底模型。") pending_notices.append(_NO_TOOL_CALL_NOTICE) @@ -1279,6 +1303,37 @@ class FlashRunner: using_tool_fallback = False self._llm = llm pending_notices.append("兜底模型已完成本步骤；后续步骤恢复首选模型。") + for _, status, _ in turn.actions: + action_failure_attempts = 0 if status == "success" else action_failure_attempts + 1 + failure_limit = ( + settings.ARTEMIS_TOOL_FALLBACK_MAX_NO_PROGRESS_ATTEMPTS + if using_tool_fallback + else settings.ARTEMIS_ACTION_FAILURE_ATTEMPTS + ) + if action_failure_attempts >= failure_limit: + if using_tool_fallback: + final_report = { + "status": "failed", + "explanation": f"兜底模型连续 {action_failure_attempts} 次设备动作失败，已停止任务。最后原因：{turn.result()}", + } + break + # A fresh, valid observation is required before spending any + # fallback model calls. Hardware recovery is not model retry. + ( + shot_path, + current_pre_screenshot_bytes, + current_xml_list, + ) = await capture_screenshot_and_parse_ui(self.ctx, state, self.controller) + state.latest_screenshot = shot_path + fallback_llm = get_llm(self.ctx, name="operator", use_fallback=True) + using_tool_fallback = True + self._llm = fallback_llm + action_failure_attempts = 0 + no_tool_attempts = 0 + pending_notices.append( + f"首选模型连续 {settings.ARTEMIS_ACTION_FAILURE_ATTEMPTS} 次设备动作失败。" + f"最后原因：{turn.result()}。已重新观察；本步骤由兜底模型重新定位，勿重复旧动作。" + ) ledger.stage_turn(messages[turn_base:]) previous_turn = turn if final_report_from_calls is not None: diff --git a/artemis/agents/validator/tool_declarations.py b/artemis/agents/validator/tool_declarations.py index 2d74a78..8bf5ddd 100644 --- a/artemis/agents/validator/tool_declarations.py +++ b/artemis/agents/validator/tool_declarations.py @@ -136,9 +136,14 @@ async def capture_screenshot_and_parse_ui( Thin adapter over :func:`artemis.mcp.observation.observe` that adds the LangGraph ``State`` write-back (indexed points/elements) the agents rely on. """ - obs, screenshot_bytes = await observe(ctx, controller, settle_ms=0 if skip_settling else 400) - if not obs.ok: - return None, None, None + from artemis.mcp.observation import require_observation + + state.indexed_points = [] + state.indexed_elements = [] + state.latest_screenshot = None + obs, screenshot_bytes = await require_observation( + ctx, controller, settle_ms=0 if skip_settling else 400 + ) if obs.hierarchy_ok: state.indexed_points = [el["center"] for el in obs.elements] diff --git a/artemis/config/settings.py b/artemis/config/settings.py index 037060e..ea49c13 100644 --- a/artemis/config/settings.py +++ b/artemis/config/settings.py @@ -122,6 +122,12 @@ class Settings(BaseSettings): ARTEMIS_OPERATOR_MAX_NO_PROGRESS_ATTEMPTS: int = Field(default=3, ge=1, le=20) # Consecutive Flash responses without any tool call before failing the task. ARTEMIS_FLASH_MAX_NO_PROGRESS_ATTEMPTS: int = Field(default=3, ge=1, le=20) + ARTEMIS_TASK_TIMEOUT_SECONDS: int = Field(default=1800, ge=30, le=86400) + ARTEMIS_OBSERVATION_TIMEOUT_SECONDS: float = Field(default=60, gt=0, le=300) + ARTEMIS_OBSERVATION_MAX_ATTEMPTS: int = Field(default=3, ge=1, le=5) + ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS: float = Field(default=10, gt=0, le=60) + ARTEMIS_ACTION_FAILURE_ATTEMPTS: int = Field(default=3, ge=1, le=10) + ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS: float = Field(default=30, gt=0, le=300) # Android ADB Connectivity ADB_HOST: str | None = Field(default=DEFAULT_ADB_HOST) diff --git a/artemis/drivers/android/adb_driver.py b/artemis/drivers/android/adb_driver.py index 8934081..1abbece 100644 --- a/artemis/drivers/android/adb_driver.py +++ b/artemis/drivers/android/adb_driver.py @@ -19,6 +19,7 @@ import base64 from io import BytesIO from pathlib import Path from typing import Any, Literal +from PIL import Image from adbutils import AdbClient, AdbDevice from artemis.clients.ui_automator_client import ( @@ -26,6 +27,7 @@ from artemis.clients.ui_automator_client import ( _parse_hierarchy_xml_to_elements, ) from artemis.config.agent import load_agent_config +from artemis.config.settings import settings 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 @@ -91,6 +93,7 @@ class AndroidAdbDriver(BaseDeviceDriver): self._recording_process: asyncio.subprocess.Process | None = None self._recording_output_path: Path | None = None self._screenshot_config = load_agent_config().screenshot + self._screen_capture_task: asyncio.Task[ScreenData] | None = None @property def device_id(self) -> str: @@ -122,6 +125,24 @@ class AndroidAdbDriver(BaseDeviceDriver): await asyncio.to_thread(self._ui_adb_client.disconnect) async def get_screen_data(self, skip_settling: bool = False) -> ScreenData: + """Bound observation latency and share unfinished capture work across retries.""" + task = self._screen_capture_task + if task is None or task.done(): + task = asyncio.create_task(self._get_screen_data(skip_settling)) + # A timed-out caller does not cancel a blocking SDK thread or start + # another concurrent read on the same device. Retrieve late errors. + task.add_done_callback(lambda done: None if done.cancelled() else done.exception()) + self._screen_capture_task = task + try: + return await asyncio.wait_for( + asyncio.shield(task), timeout=settings.ARTEMIS_OBSERVATION_TIMEOUT_SECONDS + ) + except TimeoutError as exc: + raise TimeoutError( + f"设备屏幕观测超时（{settings.ARTEMIS_OBSERVATION_TIMEOUT_SECONDS:g} 秒）" + ) from exc + + async def _get_screen_data(self, skip_settling: bool = False) -> ScreenData: """Captures screenshot and XML hierarchy concurrently.""" if not skip_settling: await asyncio.sleep(0.3) @@ -131,11 +152,14 @@ class AndroidAdbDriver(BaseDeviceDriver): screenshot_base64: str = "" ui_hierarchy_xml: str | None = None ui_elements: list[dict[str, Any]] = [] + capture_errors: list[str] = [] # Check if ui_adb_client has direct screen data support if self._ui_adb_client and hasattr(self._ui_adb_client, "get_screen_data"): try: - raw_ui_data = self._ui_adb_client.get_screen_data() + raw_ui_data = await asyncio.to_thread(self._ui_adb_client.get_screen_data) + if asyncio.iscoroutine(raw_ui_data): + raw_ui_data = await raw_ui_data if ( hasattr(raw_ui_data, "base64") and isinstance(raw_ui_data.base64, str) @@ -158,11 +182,18 @@ class AndroidAdbDriver(BaseDeviceDriver): elif isinstance(raw_elements, list): ui_elements = raw_elements except Exception as e: + capture_errors.append(str(e)) logger.debug(f"Direct ui_adb_client.get_screen_data error: {e}") if not screenshot_base64 or not isinstance(screenshot_base64, str): try: - pil_img = await asyncio.to_thread(self.device.screenshot) + raw_image = await asyncio.to_thread( + self.device.shell, + ["screencap", "-p"], + encoding=None, + timeout=settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS, + ) + pil_img = Image.open(BytesIO(raw_image)) if hasattr(pil_img, "save"): buf = BytesIO() pil_img.save(buf, format="PNG") @@ -170,18 +201,24 @@ class AndroidAdbDriver(BaseDeviceDriver): self._width, self._height = pil_img.size screenshot_base64 = base64.b64encode(screenshot_bytes).decode("utf-8") except Exception as e: + capture_errors.append(str(e)) logger.warning(f"Device screenshot capture failed on {self._device_id}: {e}") if not screenshot_base64 or not isinstance(screenshot_base64, str): - # Fallback for headless testing environments - screenshot_base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==" - screenshot_bytes = base64.b64decode(screenshot_base64) + raise RuntimeError( + "设备截图失败：ADB 与屏幕辅助服务均未返回有效截图；" + "; ".join(capture_errors) + ) + + with Image.open(BytesIO(screenshot_bytes)) as image: + if min(image.size) <= 1: + raise RuntimeError("设备截图无效：收到占位图，禁止基于旧画面继续操作") + image.verify() # 2. Capture UI Hierarchy via UIAutomator 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() + res = await asyncio.to_thread(self._ui_adb_client.get_hierarchy) hierarchy = await res if asyncio.iscoroutine(res) else res if isinstance(hierarchy, str): ui_hierarchy_xml = hierarchy @@ -189,7 +226,7 @@ class AndroidAdbDriver(BaseDeviceDriver): else: ui_elements = hierarchy elif hasattr(self._ui_adb_client, "get_ui_elements"): - res = self._ui_adb_client.get_ui_elements() + res = await asyncio.to_thread(self._ui_adb_client.get_ui_elements) ui_elements = await res if asyncio.iscoroutine(res) else res except Exception as e: logger.debug(f"UI hierarchy extraction failed: {e}") @@ -208,16 +245,8 @@ class AndroidAdbDriver(BaseDeviceDriver): screen_width=w, screen_height=h, ) - clean_b64 = ( - screenshot_base64 - if isinstance(screenshot_base64, str) and screenshot_base64 - else "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==" - ) - clean_bytes = ( - screenshot_bytes - if isinstance(screenshot_bytes, bytes) and screenshot_bytes - else base64.b64decode(clean_b64) - ) + clean_b64 = screenshot_base64 + clean_bytes = screenshot_bytes if self._screenshot_config.enabled: limited_bytes = limit_screenshot_long_edge( clean_bytes, self._screenshot_config.max_long_edge @@ -255,7 +284,9 @@ class AndroidAdbDriver(BaseDeviceDriver): cmd = f" && sleep {delay_ms / 1000.0:.3f} && ".join(taps) logger.info(f"[ADB] {cmd}") - await asyncio.to_thread(self.device.shell, cmd) + await asyncio.to_thread( + self.device.shell, cmd, timeout=settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS + ) return True except Exception as e: logger.error(f"Tap failed at ({x}, {y}): {e}") @@ -275,7 +306,9 @@ class AndroidAdbDriver(BaseDeviceDriver): try: cmd = f"input swipe {start_x} {start_y} {end_x} {end_y} {duration_ms}" logger.info(f"[ADB] {cmd}") - await asyncio.to_thread(self.device.shell, cmd) + await asyncio.to_thread( + self.device.shell, cmd, timeout=settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS + ) return True except Exception as e: logger.error(f"Swipe failed from ({start_x},{start_y}) to ({end_x},{end_y}): {e}") @@ -324,7 +357,11 @@ class AndroidAdbDriver(BaseDeviceDriver): "input keyevent 67 && " "input keyevent 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67" ) - await asyncio.to_thread(self.device.shell, clear_cmd) + await asyncio.to_thread( + self.device.shell, + clear_cmd, + timeout=settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS, + ) else: # Append mode: move cursor to the very end of existing text await asyncio.to_thread( @@ -346,7 +383,11 @@ class AndroidAdbDriver(BaseDeviceDriver): set_clip_ok = True if set_clip_ok: - await asyncio.to_thread(self.device.shell, "input keyevent 279") + await asyncio.to_thread( + self.device.shell, + "input keyevent 279", + timeout=settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS, + ) return True except Exception as e: logger.debug(f"Clipboard paste fallback to ADB input: {e}") @@ -359,7 +400,11 @@ class AndroidAdbDriver(BaseDeviceDriver): if "adbkeyboard" in str(default_ime).lower(): b64_text = base64.b64encode(norm_text.encode("utf-8")).decode("utf-8") broadcast_cmd = f"am broadcast -a ADB_INPUT_B64 --es msg '{b64_text}'" - await asyncio.to_thread(self.device.shell, broadcast_cmd) + await asyncio.to_thread( + self.device.shell, + broadcast_cmd, + timeout=settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS, + ) return True except Exception as e: # ADBKeyboard probe/broadcast failed; fall through to native input. @@ -370,10 +415,18 @@ class AndroidAdbDriver(BaseDeviceDriver): for i, line in enumerate(lines): if i > 0: # Send Enter key between lines - await asyncio.to_thread(self.device.shell, "input keyevent 66") + await asyncio.to_thread( + self.device.shell, + "input keyevent 66", + timeout=settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS, + ) if line: escaped = _escape_for_adb_text(line) - await asyncio.to_thread(self.device.shell, f"input text {escaped}") + await asyncio.to_thread( + self.device.shell, + f"input text {escaped}", + timeout=settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS, + ) return True except Exception as e: logger.error(f"Input text failed for '{text}': {e}") @@ -386,7 +439,11 @@ class AndroidAdbDriver(BaseDeviceDriver): key_name = str(key).lower().replace("keycode.", "") keycode_val = ANDROID_KEYCODE_MAP.get(key_name, key) - await asyncio.to_thread(self.device.shell, f"input keyevent {keycode_val}") + await asyncio.to_thread( + self.device.shell, + f"input keyevent {keycode_val}", + timeout=settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS, + ) return True except Exception as e: logger.error(f"Press key failed for '{key}': {e}") @@ -395,7 +452,9 @@ class AndroidAdbDriver(BaseDeviceDriver): async def launch_app(self, package_name: str) -> bool: try: cmd = f"monkey -p {package_name} -c android.intent.category.LAUNCHER 1" - await asyncio.to_thread(self.device.shell, cmd) + await asyncio.to_thread( + self.device.shell, cmd, timeout=settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS + ) return True except Exception as e: logger.error(f"Launch app failed for '{package_name}': {e}") @@ -403,7 +462,11 @@ class AndroidAdbDriver(BaseDeviceDriver): async def stop_app(self, package_name: str) -> bool: try: - await asyncio.to_thread(self.device.shell, f"am force-stop {package_name}") + await asyncio.to_thread( + self.device.shell, + f"am force-stop {package_name}", + timeout=settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS, + ) return True except Exception as e: logger.error(f"Stop app failed for '{package_name}': {e}") @@ -439,7 +502,11 @@ class AndroidAdbDriver(BaseDeviceDriver): async def execute_shell(self, command: str, timeout_seconds: float = 15.0) -> str: try: return await asyncio.wait_for( - asyncio.to_thread(self.device.shell, command), + asyncio.to_thread( + self.device.shell, + command, + timeout=settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS, + ), timeout=timeout_seconds, ) except Exception as e: diff --git a/artemis/mcp/observation.py b/artemis/mcp/observation.py index 42fffa6..cbccee9 100644 --- a/artemis/mcp/observation.py +++ b/artemis/mcp/observation.py @@ -22,9 +22,12 @@ Validator -- which runs its own UI-change polling -- simply never calls it. import asyncio import base64 +from io import BytesIO import time +from PIL import Image from artemis.config import get_temp_dir +from artemis.config.settings import settings from artemis.context import ArtemisContext from artemis.controllers.unified_controller import UnifiedMobileController from artemis.mcp.action_types import ActionCode, ObserveResult @@ -34,7 +37,29 @@ from artemis.utils.visualization import format_minimal_list_with_elements logger = get_logger(__name__) -__all__ = ["observe"] +__all__ = ["observe", "require_observation", "DeviceObservationError"] + + +class DeviceObservationError(RuntimeError): + """A live device could not be observed after bounded recovery.""" + + +async def require_observation( + ctx: ArtemisContext, controller: UnifiedMobileController, settle_ms: int = 400 +) -> tuple[ObserveResult, bytes | None]: + """Recover observation without spending model calls on unavailable screens.""" + last = "设备未返回有效画面" + for attempt in range(settings.ARTEMIS_OBSERVATION_MAX_ATTEMPTS): + observation, image = await observe(ctx, controller, settle_ms=settle_ms) + if observation.ok: + return observation, image + last = observation.message + logger.warning( + f"设备观测恢复 {attempt + 1}/{settings.ARTEMIS_OBSERVATION_MAX_ATTEMPTS}：{last}" + ) + raise DeviceObservationError( + f"设备观测失败（已尝试 {settings.ARTEMIS_OBSERVATION_MAX_ATTEMPTS} 次）：{last}" + ) async def observe( @@ -60,9 +85,19 @@ async def observe( await asyncio.sleep(settle_ms / 1000.0) else: logger.info("Retrieving screen data directly (skip settling)...") - device_data = await controller.get_screen_data() + device_data = await asyncio.wait_for( + controller.get_screen_data(), timeout=settings.ARTEMIS_OBSERVATION_TIMEOUT_SECONDS + ) screenshot_bytes = base64.b64decode(device_data.base64) + with Image.open(BytesIO(screenshot_bytes)) as image: + if min(image.size) <= 1: + raise ValueError("设备截图为占位图，当前画面不可用") + image.verify() except Exception as e: + if isinstance(e, TimeoutError): + e = TimeoutError( + f"设备屏幕观测超时（{settings.ARTEMIS_OBSERVATION_TIMEOUT_SECONDS:g} 秒）" + ) logger.error(f"Failed to capture screen state: {e}") return ( ObserveResult( diff --git a/artemis/runtime/daemon_client.py b/artemis/runtime/daemon_client.py index 56c70e0..4f8d289 100644 --- a/artemis/runtime/daemon_client.py +++ b/artemis/runtime/daemon_client.py @@ -221,6 +221,7 @@ def submit_task_to_daemon( conversation_id: str | None = None, verification_level: str | None = None, explorer_mode: str | None = None, + runtime_overrides: dict[str, Any] | None = None, base_url: str | None = None, timeout: float = 15.0, ) -> dict[str, Any] | None: @@ -246,6 +247,7 @@ def submit_task_to_daemon( "session_id": session_id, "ingress": ingress, "conversation_id": conversation_id, + "runtime_overrides": runtime_overrides, } try: diff --git a/artemis/runtime/task_limits.py b/artemis/runtime/task_limits.py new file mode 100644 index 0000000..78035e3 --- /dev/null +++ b/artemis/runtime/task_limits.py @@ -0,0 +1,30 @@ +"""Execution budgets shared by supervised web and MCP tasks.""" + +from dataclasses import dataclass, field +import time + +from artemis.config.settings import settings + + +def task_timeout(item: dict, job: dict | None = None, batch: dict | None = None) -> int: + options = item.get("runtime_overrides") or (job or {}).get("case", {}).get("task_config") or {} + return int( + options.get("timeout_seconds") + or (batch or {}).get("timeout_seconds") + or settings.ARTEMIS_TASK_TIMEOUT_SECONDS + ) + + +@dataclass +class ExecutionBudget: + limit: float + elapsed: float = 0 + _last: float = field(default_factory=time.monotonic) + _excluded: bool = False + + def update(self, *, excluded: bool = False) -> float: + now = time.monotonic() + if not self._excluded: + self.elapsed += max(0, now - self._last) + self._last, self._excluded = now, excluded + return max(0, self.limit - self.elapsed) diff --git a/artemis/runtime/task_result.py b/artemis/runtime/task_result.py new file mode 100644 index 0000000..ed39d43 --- /dev/null +++ b/artemis/runtime/task_result.py @@ -0,0 +1,64 @@ +"""Persist useful failure evidence even when a worker must be terminated.""" + +import json +import os +from pathlib import Path +import uuid + +from artemis.config import TRACES_PATH + + +def read_worker_outcome(session_id: str) -> dict: + """Read a worker's terminal signal without waiting for hung SDK threads.""" + try: + data = json.loads( + (TRACES_PATH / session_id / "run_outcome.json").read_text(encoding="utf-8") + ) + if ( + isinstance(data, dict) + and data.get("worker_finalized") is True + and data.get("task_status") + in { + "failed", + "completed", + "cancelled", + "interrupted", + } + ): + return data + except (OSError, ValueError): + pass # Atomic outcome is absent until the task finalizes. + return {} + + +def save_failure_result(session_id: str, error: str, *, code: str = "execution_failed") -> None: + if not session_id or not all(c.isalnum() or c in "-_" for c in session_id): + raise ValueError("无效的任务编号") + directory = TRACES_PATH / session_id + directory.mkdir(parents=True, exist_ok=True) + payload: dict = {} + try: + previous = json.loads((directory / "result.json").read_text(encoding="utf-8")) + if isinstance(previous, dict): + payload.update(previous) + except (OSError, ValueError): + pass # Missing/truncated results are expected after a killed worker. + payload.update( + { + "status": "failed", + "task_status": "failed", + "explanation": error, + "error": error, + "code": code, + "worker_finalized": True, + "tests": payload.get("tests") or {}, + } + ) + for name in ("result.json", "run_outcome.json"): + target = directory / name + temporary = target.with_name(f".{name}.{uuid.uuid4().hex}.tmp") + try: + temporary.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") + os.replace(temporary, target) + finally: + temporary.unlink(missing_ok=True) diff --git a/artemis/sdk/agent.py b/artemis/sdk/agent.py index 284b90c..e457b01 100644 --- a/artemis/sdk/agent.py +++ b/artemis/sdk/agent.py @@ -32,6 +32,7 @@ import shutil from shutil import which import sys import threading +import time from types import NoneType from typing import Any, TypeVar, overload import uuid @@ -573,6 +574,13 @@ class Agent: ) try: if device_lock is not None and os.environ.get("ARTEMIS_CLOUD_MODE") != "1": + from artemis.runtime.task_control import control_dir, _write_atomic + + if sess_id and os.environ.get("ARTEMIS_TASK_WORKER") == "1": + _write_atomic( + control_dir(str(sess_id)) / "task_control_state.json", + {"status": "queued", "updated_at": time.time(), "pid": os.getpid()}, + ) queue_cancel_event = threading.Event() acquire_task = asyncio.create_task( asyncio.to_thread( @@ -605,6 +613,7 @@ class Agent: task_control = TaskControl(str(sess_id), device_lock) context.task_control = task_control + task_control._state("running") self._prepare_output_files(task=task) if os.environ.get("ARTEMIS_CLOUD_MODE") != "1": if self._ui_adb_client is not None: @@ -712,6 +721,14 @@ class Agent: state=last_state_snapshot, error=err, ) + from artemis.runtime.task_result import save_failure_result + + if sess_id: + save_failure_result( + str(sess_id), + err, + code=flash_result.get("code", "execution_failed"), + ) if context.data_engine: context.data_engine.end_session("failed") return output @@ -862,6 +879,10 @@ class Agent: state=last_state_snapshot, error=err, ) + from artemis.runtime.task_result import save_failure_result + + if sess_id: + save_failure_result(str(sess_id), err) if context.data_engine: context.data_engine.end_session("failed") diff --git a/docs/ai-automation.md b/docs/ai-automation.md index 86881c5..03c0e69 100644 --- a/docs/ai-automation.md +++ b/docs/ai-automation.md @@ -49,3 +49,9 @@ MCP 同步会识别已安装客户端的配置目录，为缺少 ARTEMIS 条目 批次任务在队列中使用“用例编号 + 用例名称”显示，提示词仍是模型实际收到的内容。任务历史展示失败原因、完成时间和总耗时；导出全部资料时以用例编号命名目录。删除非活跃批次会同步清理其任务记录、步骤、截图、录屏及生成的报告，删除后不能通过旧 `request_id` 重复执行。 保存全局配置要求用户明确指令；只读配置不包含密钥，MCP 不接收密钥。密钥仍在网页或 `.env` 中填写。设置对新任务生效，模型请求重试不会触发测试用例自动重跑。历史步骤证据保留，设备动作回放已移除。 + +执行保护默认总时限 1800 秒，单任务、批次和 MCP 都适用，排队及用户暂停不计时。设备观测默认 60 秒、最多 3 次，ADB 命令默认 10 秒；观测失败不会继续向模型发送占位图或旧元素索引。Flash 连续 3 次动作失败先重新观察再切换兜底，兜底仍失败则结束任务并保存原因。不能以提高步数或反复重提任务代替设备诊断。 + +`mobile_run_task(runtime_overrides={"timeout_seconds":1800,"observation_timeout_seconds":60,"observation_max_attempts":3,"device_command_timeout_seconds":10,"action_failure_attempts":3})` 可为单任务覆盖执行保护；其中 `model` 是 API 模型名，外层 `model` 仍是 Flash/Pro 模式。相同字段也支持批次的 `task_config`。全局默认通过网页功能配置或明确授权的 `mobile_config(section="capabilities", changes={"task_timeout_seconds":1800}, persist_defaults=true)` 保存，对后续任务生效，不接受密钥。 + +任务已经产生最终结果，但 SDK 线程或资源清理仍卡住时，默认最多再等待 30 秒后回收工作进程，避免继续占用队列。全局可用 `ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS` 或功能配置的 `worker_cleanup_timeout_seconds` 调整，任务 JSON 同名字段可覆盖。原有结果与截图保留，未完成保存的录屏尽力恢复。 diff --git a/docs/automation-workspace.md b/docs/automation-workspace.md index e4fcaff..3def2bc 100644 --- a/docs/automation-workspace.md +++ b/docs/automation-workspace.md @@ -20,7 +20,7 @@ 4. 上传文件，查看逐行校验结果。所有错误修正后才能提交。 5. 明确选择设备、执行模式和超时时间，然后点击“确认导入并开始执行”。 -同一批次按顺序派发。默认失败后继续且不自动重试，也可以选择失败即停止。设备不可用、模型暂停或任务超时会暂停后续派发。暂停批次只停止后续派发。单个任务可在安全边界暂停并释放设备；批次下一项可继续，暂停的任务待提交指导或点击继续后在设备队列末尾恢复。运行中提交指导会中断当前模型思考或只读诊断，重新观察画面；已经开始的设备动作会安全结束。网页流程会显示指导的提交与领取状态，仍应检查后续动作是否符合指导。暂停等待和重新排队时间不计入该任务执行时限。 +同一批次按顺序派发。默认失败后继续且不自动重试，也可以选择失败即停止。当前任务因设备异常或超时失败后，默认继续后续用例；设置失败即停止时才停止后续派发。暂停批次只停止后续派发。单个任务可在安全边界暂停并释放设备；批次下一项可继续，暂停的任务待提交指导或点击继续后在设备队列末尾恢复。运行中提交指导会中断当前模型思考或只读诊断，重新观察画面；已经开始的设备动作会安全结束。网页流程会显示指导的提交与领取状态，仍应检查后续动作是否符合指导。暂停等待和重新排队时间不计入该任务执行时限。 服务重启后保留批次，等待用户继续未开始的用例。正在执行或暂停的任务标记中断，不自动重做；“复制失败项为新批次”保留原记录并产生新任务编号。 @@ -34,6 +34,10 @@ ## 环境配置 -`.env` 可配置 `ARTEMIS_BATCH_TASK_TIMEOUT`（默认 1800 秒）、`ARTEMIS_IMPORT_MAX_ROWS`（默认 1000 行）和 `ARTEMIS_IMPORT_MAX_BYTES`（默认 10 MiB）。修改后重启服务。网页的单批次超时设置优先于环境默认值。 +`.env` 可配置 `ARTEMIS_TASK_TIMEOUT_SECONDS`（单任务总执行时限，默认 1800 秒）、`ARTEMIS_IMPORT_MAX_ROWS`（默认 1000 行）和 `ARTEMIS_IMPORT_MAX_BYTES`（默认 10 MiB）。修改后重启服务。网页的单批次超时设置优先于环境默认值，每行 JSON 的 `timeout_seconds` 优先于批次。单任务、批次和 MCP 均受总执行时限保护，不计排队和用户暂停时间；即使执行进程卡住，也由父进程终止并保存失败原因。 + +网页“功能配置 → 执行保护”及 MCP `mobile_config(section="capabilities")` 可配置总时限、设备观测时限（默认 60 秒）、设备观测次数（默认 3 次）、ADB 命令超时（默认 10 秒）、连续动作失败次数（默认 3 次）。任务 JSON 可用 `observation_timeout_seconds`、`observation_max_attempts`、`device_command_timeout_seconds`、`action_failure_attempts` 覆盖。截图失效或收到占位图时先有限重取，耗尽后结束任务，不把旧画面发给模型；有效截图即使没有 XML 也可按坐标定位。连续动作失败时先刷新画面再尝试兜底，兜底耗尽后结束，不循环重提用例。 批次、任务元数据和报告偏好保存在现有 SQLite 数据库的 `automation_records`、`automation_jobs` 表。全局“清空任务历史”会清除批次与执行元数据；它与上文的单批次删除不同，已导出的文件会保留。 + +任务已经产生最终结果，但 SDK 线程或资源清理仍卡住时，默认最多再等待 30 秒后回收工作进程，避免继续占用队列。全局可用 `ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS` 或功能配置的 `worker_cleanup_timeout_seconds` 调整，任务 JSON 同名字段可覆盖。原有结果与截图保留，未完成保存的录屏尽力恢复。 diff --git a/mcp_server/tools/automation.py b/mcp_server/tools/automation.py index 73cb51b..e81950b 100644 --- a/mcp_server/tools/automation.py +++ b/mcp_server/tools/automation.py @@ -26,7 +26,19 @@ MODEL_FIELDS = { "tool_fallback_no_progress_attempts", "reasoning_effort", } -CAPABILITY_FIELDS = {"video_enabled", "outputter_enabled", "step_summarizer_enabled"} +CAPABILITY_FIELDS = { + "video_enabled", + "recording_enabled", + "outputter_enabled", + "step_summarizer_enabled", + "screenshot_limit_enabled", + "task_timeout_seconds", + "observation_timeout_seconds", + "observation_max_attempts", + "device_command_timeout_seconds", + "action_failure_attempts", + "worker_cleanup_timeout_seconds", +} def _id(value: str | None) -> str: @@ -131,8 +143,12 @@ async def mobile_config( tool_fallback_reasoning_effort, tool_fallback_max_retries, tool_fallback_no_progress_attempts. 'default' omits reasoning; unsupported values fail visibly. - capabilities: video_enabled (recording + analysis), outputter_enabled, - step_summarizer_enabled. reports: formats=['docx','xlsx'] (at least one). + capabilities: video_enabled and recording_enabled are independent; outputter_enabled, + step_summarizer_enabled, screenshot_limit_enabled, task_timeout_seconds (default 1800), + observation_timeout_seconds (60), observation_max_attempts (3), + device_command_timeout_seconds (10), action_failure_attempts (3), + worker_cleanup_timeout_seconds (30). + reports: formats=['docx','xlsx'] (at least one). API keys are never accepted or returned: use the Web UI or .env for credentials. Saving settings does not change an already running task. Flash/Pro are execution profiles, not model names. Model retries do not repeat device test cases. diff --git a/mcp_server/tools/task_runner.py b/mcp_server/tools/task_runner.py index d30c42e..bea75af 100644 --- a/mcp_server/tools/task_runner.py +++ b/mcp_server/tools/task_runner.py @@ -35,6 +35,10 @@ from artemis.runtime import ( submit_task_to_daemon, trace_store, ) +from artemis.runtime.task_limits import ExecutionBudget, task_timeout +from artemis.runtime.task_control import read_state +from artemis.runtime.task_result import save_failure_result, read_worker_outcome +from artemis.config.settings import settings # Seconds the spawned runner gets to finish its imports and open its log files # before the spawn is declared dead. Normal startup creates stdout.log within a @@ -44,6 +48,54 @@ from artemis.runtime import ( SPAWN_WATCHDOG_SECONDS = 60 +def _watch_execution(proc, trace_id: str, limit: int, cleanup_limit: float | None = None) -> None: + """Supervise standalone MCP workers without charging device queue or pause time.""" + budget = ExecutionBudget(limit) + finalized_at = None + cleanup_limit = cleanup_limit or settings.ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS + while proc.poll() is None: + status = trace_store.read_status(trace_id) or {} + outcome = read_worker_outcome(trace_id) + if outcome and status.get("status") not in { + "completed", + "success", + "failed", + "cancelled", + "interrupted", + }: + try: + trace_store.update_trace_status( + trace_id, outcome["task_status"], error=outcome.get("error") + ) + except OSError: + logging.exception("Cannot persist terminal status for %s", trace_id) + if outcome or status.get("status") in { + "completed", + "success", + "failed", + "cancelled", + "interrupted", + }: + finalized_at = finalized_at if finalized_at is not None else time.monotonic() + if time.monotonic() - finalized_at >= cleanup_limit: + _kill_process_tree(proc.pid) + return + time.sleep(0.5) + continue + control_status = read_state(trace_id).get("status") + remaining = budget.update(excluded=control_status in {"queued", "task_paused", "resuming"}) + if remaining <= 0: + error = f"任务执行超时（{limit} 秒，不含暂停/排队时间）" + _kill_process_tree(proc.pid) + try: + save_failure_result(trace_id, error, code="task_timeout") + trace_store.update_trace_status(trace_id, "failed", error=error) + except OSError: + logging.exception("Cannot persist timeout report for %s", trace_id) + return + time.sleep(0.5) + + def _kill_process_tree(pid: int) -> None: """Terminate a spawned runner and its children (the venv shim re-execs python).""" if sys.platform == "win32": @@ -212,6 +264,7 @@ def mobile_run_task( device_serial: str | None = None, verification_level: str | None = None, explorer_mode: str | None = None, + runtime_overrides: dict[str, Any] | None = None, ) -> dict[str, Any]: """Starts an autonomous mobile UI automation subagent on a connected Android device. @@ -277,7 +330,25 @@ def mobile_run_task( by the Operator: `"flash"` (1-shot detection, the default), `"pro"` (3-turn ReAct), `"ultra"` (deep pixel reasoning; slowest). Ignored for Flash. + runtime_overrides: Optional TaskOptions JSON for this task only. Supports + timeout_seconds (total execution, default 1800; excludes queue/pause), + observation_timeout_seconds (60), observation_max_attempts (3), + device_command_timeout_seconds (10), action_failure_attempts (3), + worker_cleanup_timeout_seconds (30), + model (API model name), reasoning_effort, request timeout/retries, + recording and video analysis. Unknown fields and credentials are rejected. """ + if runtime_overrides: + from apps.admin_console.services.task_import import TaskOptions + + runtime_overrides = TaskOptions.model_validate(runtime_overrides).model_dump( + exclude_none=True + ) + model = runtime_overrides.get("profile") or model + device_serial = runtime_overrides.get("device_serial") or device_serial + locked_app_package = runtime_overrides.get("locked_app_package") or locked_app_package + verification_level = runtime_overrides.get("verification_level") or verification_level + explorer_mode = runtime_overrides.get("explorer_mode") or explorer_mode # 0. Validate and normalize model if model.lower() not in ("flash", "pro"): raise ValueError(f"Invalid model '{model}'. Must be either 'Flash' or 'Pro'.") @@ -325,6 +396,7 @@ def mobile_run_task( conversation_id=conversation_id, verification_level=verification_level, explorer_mode=explorer_mode, + runtime_overrides=runtime_overrides, base_url=base_url, ) if resp and resp.get("status") == "rejected": @@ -475,6 +547,26 @@ def mobile_run_task( env = os.environ.copy() env["ARTEMIS_SESSION_ID"] = trace_id env["ARTEMIS_TASK_INGRESS"] = "mcp" + env["ARTEMIS_TASK_WORKER"] = "1" + for option, variable in { + "model": "OPENAI_MODEL", + "reasoning_effort": "OPENAI_REASONING_EFFORT", + "request_timeout_seconds": "OPENAI_TIMEOUT_SECONDS", + "max_retries": "OPENAI_MAX_RETRIES", + "retry_interval_seconds": "OPENAI_RETRY_INTERVAL_SECONDS", + "record_video": "ARTEMIS_RECORDING_ENABLED", + "analyze_video": "ARTEMIS_VIDEO_ENABLED", + "observation_timeout_seconds": "ARTEMIS_OBSERVATION_TIMEOUT_SECONDS", + "observation_max_attempts": "ARTEMIS_OBSERVATION_MAX_ATTEMPTS", + "device_command_timeout_seconds": "ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS", + "action_failure_attempts": "ARTEMIS_ACTION_FAILURE_ATTEMPTS", + }.items(): + if (runtime_overrides or {}).get(option) is not None: + env[variable] = ( + str(runtime_overrides[option]).lower() + if isinstance(runtime_overrides[option], bool) + else str(runtime_overrides[option]) + ) if device_serial: env["ADB_DEVICE_SERIAL"] = device_serial env["ARTEMIS_DEVICE_ID"] = device_serial @@ -519,6 +611,17 @@ def mobile_run_task( # interpreter startup and must be killed instead of squatting on the # queue as 'running' forever. _start_spawn_watchdog(trace_id, proc.pid, queue_ticket, conversation_id) + threading.Thread( + target=_watch_execution, + args=( + proc, + trace_id, + task_timeout({"runtime_overrides": runtime_overrides}), + (runtime_overrides or {}).get("worker_cleanup_timeout_seconds"), + ), + daemon=True, + name=f"artemis-task-budget-{trace_id[:8]}", + ).start() trace_dir = trace_store.get_trace_dir(trace_id) stdout_log_path = os.path.join(trace_dir, "stdout.log") diff --git a/skills/artemis-automation/SKILL.md b/skills/artemis-automation/SKILL.md index 3b0030d..53ff98b 100644 --- a/skills/artemis-automation/SKILL.md +++ b/skills/artemis-automation/SKILL.md @@ -35,3 +35,9 @@ description: 使用 ARTEMIS MCP 在 Android 真机或模拟器上完成自然语 - 批次暂停只停止后续派发；当前任务等待指导时用 `mobile_manage_task(action="pause", trace_id=...)`，真正取消用 `stop`。用户明确要求删除批次记录时先 `mobile_batch(action="get")` 核对编号，再用 `delete`；执行中/排队中的批次不能删，删除会清除该批次的任务历史、报告和文件。服务重启后运行/暂停的任务标记中断，不自动重跑。停止可能保留已经产生的业务效果，不声称自动回滚。 需要编写外部可执行测试代码时，先通过 ARTEMIS 实际探索目标应用，确认页面、定位器和转换。优先动态定位，框架支持时才以已验证坐标兜底；在最终脚本中明确等待条件，不将 AI 思考耗时当作确定性等待。 + +执行保护默认总时限 1800 秒，单任务、批次和 MCP 都适用，排队及用户暂停不计时。设备观测默认 60 秒、最多 3 次，ADB 命令默认 10 秒；观测失败不会继续向模型发送占位图或旧元素索引。Flash 连续 3 次动作失败先重新观察再切换兜底，兜底仍失败则结束任务并保存原因。不能以提高步数或反复重提任务代替设备诊断。 + +`mobile_run_task(runtime_overrides={"timeout_seconds":1800,"observation_timeout_seconds":60,"observation_max_attempts":3,"device_command_timeout_seconds":10,"action_failure_attempts":3})` 可为单任务覆盖执行保护；其中 `model` 是 API 模型名，外层 `model` 仍是 Flash/Pro 模式。相同字段也支持批次的 `task_config`。全局默认通过网页功能配置或明确授权的 `mobile_config(section="capabilities", changes={"task_timeout_seconds":1800}, persist_defaults=true)` 保存，对后续任务生效，不接受密钥。 + +任务已经产生最终结果，但 SDK 线程或资源清理仍卡住时，默认最多再等待 30 秒后回收工作进程，避免继续占用队列。全局可用 `ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS` 或功能配置的 `worker_cleanup_timeout_seconds` 调整，任务 JSON 同名字段可覆盖。原有结果与截图保留，未完成保存的录屏尽力恢复。 diff --git a/tests/unit/admin_console/test_model_settings.py b/tests/unit/admin_console/test_model_settings.py index 1807562..a0623da 100644 --- a/tests/unit/admin_console/test_model_settings.py +++ b/tests/unit/admin_console/test_model_settings.py @@ -33,6 +33,12 @@ def isolated_settings(monkeypatch, tmp_path): "ARTEMIS_OUTPUTTER_ENABLED", "ARTEMIS_STEP_SUMMARIZER_ENABLED", "ARTEMIS_SCREENSHOT_LIMIT_ENABLED", + "ARTEMIS_TASK_TIMEOUT_SECONDS", + "ARTEMIS_OBSERVATION_TIMEOUT_SECONDS", + "ARTEMIS_OBSERVATION_MAX_ATTEMPTS", + "ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS", + "ARTEMIS_ACTION_FAILURE_ATTEMPTS", + "ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS", ] for key in keys: monkeypatch.setattr(settings, key, getattr(settings, key)) @@ -202,3 +208,26 @@ def test_responses_payload_has_reasoning_and_no_sdk_retry(monkeypatch, effort): assert factory.call_args.kwargs["use_responses_api"] is True ModelFactory.create_model(ModelEndpoint(api_key="test-only", reasoning_effort="default")) assert "reasoning_effort" not in factory.call_args.kwargs + + +def test_execution_protection_saves_reload_without_touching_secret(isolated_settings): + result = service.save_capability_settings( + service.CapabilitySettings( + task_timeout_seconds=3600, + observation_timeout_seconds=45, + observation_max_attempts=2, + device_command_timeout_seconds=8, + action_failure_attempts=4, + worker_cleanup_timeout_seconds=15, + ) + ) + reloaded = Settings(_env_file=isolated_settings) + assert reloaded.ARTEMIS_TASK_TIMEOUT_SECONDS == result.task_timeout_seconds == 3600 + assert reloaded.ARTEMIS_OBSERVATION_TIMEOUT_SECONDS == 45 + assert reloaded.ARTEMIS_OBSERVATION_MAX_ATTEMPTS == 2 + assert reloaded.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS == 8 + assert reloaded.ARTEMIS_ACTION_FAILURE_ATTEMPTS == 4 + assert reloaded.ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS == 15 + assert reloaded.OPENAI_API_KEY is None + assert settings.OPENAI_API_KEY.get_secret_value() == "existing-test-secret" + assert dotenv_values(isolated_settings)["UNRELATED"] == "keep" diff --git a/tests/unit/agents/test_flash_runner.py b/tests/unit/agents/test_flash_runner.py index f188b5e..ef03f25 100644 --- a/tests/unit/agents/test_flash_runner.py +++ b/tests/unit/agents/test_flash_runner.py @@ -15,6 +15,9 @@ """Unit tests for Universal FlashRunner.""" import base64 +from io import BytesIO +from PIL import Image +from artemis.mcp.observation import DeviceObservationError from types import SimpleNamespace from unittest.mock import AsyncMock, Mock, patch import pytest @@ -72,9 +75,18 @@ def test_flash_runner_tools_initialization(mock_context): # --- Post-action screenshot fallback (failed action / unknown tool) ---------------- +def _post_image(): + buffer = BytesIO() + Image.new("RGB", (64, 64), "white").save(buffer, format="PNG") + return buffer.getvalue() + + +_POST_IMG = _post_image() + + def _screen(elements): return SimpleNamespace( - base64=base64.b64encode(b"POST-IMG").decode("ascii"), + base64=base64.b64encode(_POST_IMG).decode("ascii"), elements=elements, width=1080, height=2400, @@ -120,7 +132,7 @@ async def test_fallback_observation_renders_the_indexed_ui_list(mock_context, tm exec_result, "click", frozenset({"click"}), state ) - assert post == b"POST-IMG" + assert post == _POST_IMG assert isinstance(exec_result.ui_elements_text, str) assert exec_result.ui_elements_text == "[1] Text: 'Settings' | Bounds: [37,50][962,112]" assert "{'text'" not in exec_result.ui_elements_text @@ -154,8 +166,8 @@ async def test_fallback_observation_is_skipped_for_non_action_tools(mock_context @pytest.mark.asyncio -async def test_fallback_observation_keeps_the_executors_ui_list(mock_context, tmp_path): - """An action that reported a UI list but no screenshot keeps its own list.""" +async def test_fallback_observation_replaces_stale_executor_ui_list(mock_context, tmp_path): + """Fresh screenshots and element lists must describe the same observation.""" with patch("artemis.controllers.unified_controller.get_driver"): runner = FlashRunner(mock_context, goal="g") runner.controller = Mock() @@ -175,12 +187,13 @@ async def test_fallback_observation_keeps_the_executors_ui_list(mock_context, tm exec_result, "click", frozenset({"click"}), None ) - assert post == b"POST-IMG" - assert exec_result.ui_elements_text == "[1] Text: 'Mine' | Bounds: [0,0][10,10]" + assert post == _POST_IMG + assert "Other" in exec_result.ui_elements_text + assert "Mine" not in exec_result.ui_elements_text @pytest.mark.asyncio -async def test_fallback_observation_failure_is_contained(mock_context): +async def test_fallback_observation_failure_stops_after_bounded_recovery(mock_context): with patch("artemis.controllers.unified_controller.get_driver"): runner = FlashRunner(mock_context, goal="g") runner.controller = Mock() @@ -189,11 +202,13 @@ async def test_fallback_observation_failure_is_contained(mock_context): state.latest_screenshot = "old.jpg" exec_result = _failed_click() - post = await runner._capture_post_screenshot(exec_result, "click", frozenset({"click"}), state) - - assert post is None - assert exec_result.ui_elements_text is None - assert state.latest_screenshot == "old.jpg" + with patch("artemis.mcp.observation.settings.ARTEMIS_OBSERVATION_MAX_ATTEMPTS", 3): + with pytest.raises(DeviceObservationError, match="adb gone"): + await runner._capture_post_screenshot(exec_result, "click", frozenset({"click"}), state) + assert runner.controller.get_screen_data.await_count == 3 + assert state.latest_screenshot is None + assert state.indexed_elements == [] + assert state.indexed_points == [] def test_flash_runner_executor_follows_the_flash_explorer_tier(mock_context): diff --git a/tests/unit/agents/test_flash_runner_ledger.py b/tests/unit/agents/test_flash_runner_ledger.py index 59ebb2f..29c6f8f 100644 --- a/tests/unit/agents/test_flash_runner_ledger.py +++ b/tests/unit/agents/test_flash_runner_ledger.py @@ -213,7 +213,8 @@ def test_observation_tail_has_pro_shape(mock_context): texts = [b["text"] for b in later.content] assert OBSERVATION_HEADER_RE.match(texts[0]) assert "objective" not in "".join(texts).lower() - assert texts[1:] == [ + assert "当前没有可用的元素索引" in texts[1] + assert texts[2:] == [ "nudge", "[INJECTED]", 'User instruction: "stop"', @@ -408,6 +409,98 @@ def _report(status="completed", tc_id="tc-report", **extra): _INITIAL_OBSERVATION = ("shot0.png", b"IMG0", "[1] Settings") +@pytest.mark.asyncio +@pytest.mark.parametrize("fallback_succeeds", [True, False]) +async def test_failed_actions_have_finite_fallback_and_restore_primary( + mock_context, monkeypatch, fallback_succeeds +): + monkeypatch.setattr("artemis.agents.flash.runner.settings.ARTEMIS_ACTION_FAILURE_ATTEMPTS", 3) + monkeypatch.setattr( + "artemis.agents.flash.runner.settings.ARTEMIS_TOOL_FALLBACK_MAX_NO_PROGRESS_ATTEMPTS", 3 + ) + action_count = 4 if fallback_succeeds else 6 + responses = [ + AIMessage( + content="try", tool_calls=[{"name": "click", "args": {"index": 1}, "id": f"tc-{i}"}] + ) + for i in range(action_count) + ] + if fallback_succeeds: + responses.append(_report()) + with patch("artemis.controllers.unified_controller.get_driver"): + runner = _make_runner(mock_context, responses) + fallback = Mock() + count = 0 + + def execute(name, args, tc_id, state, **kwargs): + nonlocal count + count += 1 + result = _exec_result(tc_id, name, b"current-screen") + if count <= 3 or not fallback_succeeds: + result.status = "error" + result.text_summary = "目标不存在" + return result + + runner.executor.execute = AsyncMock(side_effect=execute) + with ( + patch( + "artemis.agents.flash.runner.capture_screenshot_and_parse_ui", + AsyncMock(return_value=_INITIAL_OBSERVATION), + ) as observe, + patch("artemis.agents.flash.runner.get_llm", return_value=fallback), + ): + outcome = await runner.run(State(initial_goal="Open Settings")) + primary = runner._init_llm.return_value + assert all(call["llm"] is primary for call in runner.calls[:3]) + assert runner.calls[3]["llm"] is fallback + assert observe.await_count == 2 # initial + fresh observation before fallback + if fallback_succeeds: + assert outcome["status"] == "completed" + assert runner.calls[-1]["llm"] is primary + else: + assert outcome["status"] == "failed" + assert "兜底模型连续 3" in outcome["explanation"] + assert "目标不存在" in outcome["explanation"] + assert len(runner.calls) == 6 + + +@pytest.mark.asyncio +async def test_empty_current_ui_clears_previous_turn_element_list(mock_context): + response = AIMessage( + content="tap", tool_calls=[{"name": "click", "args": {"index": 1}, "id": "tc"}] + ) + with patch("artemis.controllers.unified_controller.get_driver"): + runner = _make_runner(mock_context, [response, _report()]) + runner.executor.execute = AsyncMock( + return_value=_exec_result("tc", "click", b"fresh-screen", ui="") + ) + with patch( + "artemis.agents.flash.runner.capture_screenshot_and_parse_ui", + AsyncMock(return_value=_INITIAL_OBSERVATION), + ): + await runner.run(State(initial_goal="Open Settings")) + tail = runner.calls[-1]["messages"][-1] + text = "\n".join(block.get("text", "") for block in tail.content) + assert "当前没有可用的元素索引" in text + assert "[1] Settings" not in text + + +@pytest.mark.asyncio +async def test_unavailable_screen_ends_before_any_model_call(mock_context): + from artemis.mcp.observation import DeviceObservationError + + with patch("artemis.controllers.unified_controller.get_driver"): + runner = _make_runner(mock_context, []) + with patch( + "artemis.agents.flash.runner.capture_screenshot_and_parse_ui", + AsyncMock(side_effect=DeviceObservationError("设备断连")), + ): + outcome = await runner.run(State(initial_goal="Open Settings")) + assert outcome["code"] == "device_observation_failed" + assert outcome["error"] == "设备断连" + runner._invoke_model.assert_not_awaited() + + @pytest.mark.asyncio async def test_run_builds_prompt_from_ledger_with_session_offsets(mock_context): """Turn 2 must see: the static system prefix, the committed turn 1 (its diff --git a/tests/unit/agents/test_planner_node.py b/tests/unit/agents/test_planner_node.py index 9fabb11..116b331 100644 --- a/tests/unit/agents/test_planner_node.py +++ b/tests/unit/agents/test_planner_node.py @@ -33,6 +33,17 @@ class DummyState: self.operator_replan_reason = operator_replan_reason +import base64 +from io import BytesIO +from PIL import Image + + +def _valid_screen_base64(): + buffer = BytesIO() + Image.new("RGB", (64, 64), "white").save(buffer, format="PNG") + return base64.b64encode(buffer.getvalue()).decode("ascii") + + @pytest.fixture def mock_context(): ctx = Mock(spec=ArtemisContext) @@ -41,7 +52,7 @@ def mock_context(): ctx.adb_client = Mock() ctx.ui_adb_client = Mock() mock_screen_data = Mock() - mock_screen_data.base64 = "YmFzZTY0" + mock_screen_data.base64 = _valid_screen_base64() mock_screen_data.elements = [] mock_screen_data.width = 1080 mock_screen_data.height = 2400 diff --git a/tests/unit/test_drivers.py b/tests/unit/test_drivers.py index 592ebdc..7012e63 100644 --- a/tests/unit/test_drivers.py +++ b/tests/unit/test_drivers.py @@ -14,7 +14,9 @@ """Unit tests for Unified Base Device Driver and implementations.""" +import asyncio import base64 +import threading from io import BytesIO from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock @@ -36,6 +38,48 @@ _ONE_PIXEL_PNG = base64.b64encode( ).decode("ascii") +def _valid_png(): + buffer = BytesIO() + Image.new("RGB", (64, 64), "white").save(buffer, format="PNG") + return base64.b64encode(buffer.getvalue()).decode("ascii") + + +_VALID_PNG = _valid_png() + + +@pytest.mark.asyncio +async def test_android_driver_rejects_placeholder_without_returning_stale_screen(): + client = MagicMock() + client.get_screen_data.return_value = SimpleNamespace(base64=_ONE_PIXEL_PNG) + driver = AndroidAdbDriver("test-device", MagicMock(), ui_adb_client=client) + with pytest.raises(RuntimeError, match="占位图"): + await driver.get_screen_data(skip_settling=True) + + +@pytest.mark.asyncio +async def test_screen_capture_timeout_does_not_start_overlapping_sdk_calls(monkeypatch): + released = threading.Event() + client = MagicMock() + + def blocked_capture(): + released.wait(2) + return SimpleNamespace(base64=_VALID_PNG, width=64, height=64, elements=[]) + + client.get_screen_data.side_effect = blocked_capture + driver = AndroidAdbDriver("test-device", MagicMock(), ui_adb_client=client) + monkeypatch.setattr( + "artemis.drivers.android.adb_driver.settings.ARTEMIS_OBSERVATION_TIMEOUT_SECONDS", 0.02 + ) + try: + for _ in range(3): + with pytest.raises(TimeoutError, match="观测超时"): + await driver.get_screen_data(skip_settling=True) + assert client.get_screen_data.call_count == 1 + finally: + released.set() + await asyncio.wait_for(driver._screen_capture_task, 2) + + def test_screenshot_limit_keeps_small_and_invalid_images_unchanged(): small_image = base64.b64decode(_ONE_PIXEL_PNG) assert limit_screenshot_long_edge(small_image, 2048) is small_image @@ -100,16 +144,16 @@ async def test_android_driver_with_mock_adb(): # Test tap await driver.tap(540, 1200) - mock_adb_device.shell.assert_called_with("input tap 540 1200") + mock_adb_device.shell.assert_called_with("input tap 540 1200", timeout=10.0) # Test multi-tap await driver.tap(540, 1200, times=3, delay_ms=100) expected_multi = "input tap 540 1200 && sleep 0.100 && input tap 540 1200 && sleep 0.100 && input tap 540 1200" - mock_adb_device.shell.assert_called_with(expected_multi) + mock_adb_device.shell.assert_called_with(expected_multi, timeout=10.0) # Test press key await driver.press_key(KeyCode.BACK) - mock_adb_device.shell.assert_called_with("input keyevent 4") + mock_adb_device.shell.assert_called_with("input keyevent 4", timeout=10.0) @pytest.mark.asyncio @@ -132,7 +176,7 @@ async def test_android_driver_preserves_elements_from_combined_screen_data(): }, ] mock_ui_client.get_screen_data.return_value = SimpleNamespace( - base64=_ONE_PIXEL_PNG, + base64=_VALID_PNG, hierarchy_xml="", elements=expected_elements, width=1080, @@ -188,7 +232,7 @@ async def test_android_driver_parses_xml_hierarchy_fallback(): mock_adb_client = MagicMock() mock_ui_client = MagicMock() mock_ui_client.get_screen_data.return_value = SimpleNamespace( - base64=_ONE_PIXEL_PNG, + base64=_VALID_PNG, width=1080, height=2400, ) @@ -216,7 +260,7 @@ async def test_find_element_prefers_fresh_bounds_over_stale_center(): driver.get_screen_data = AsyncMock( return_value=ScreenData( screenshot_bytes=base64.b64decode(_ONE_PIXEL_PNG), - screenshot_base64=_ONE_PIXEL_PNG, + screenshot_base64=_VALID_PNG, ui_elements=[ { "resource_id": "app:id/digit_4", diff --git a/tests/unit/test_task_limits.py b/tests/unit/test_task_limits.py new file mode 100644 index 0000000..0716dc6 --- /dev/null +++ b/tests/unit/test_task_limits.py @@ -0,0 +1,194 @@ +"""Task deadlines must apply to every ingress and exclude waits.""" + +import json +from unittest.mock import AsyncMock, Mock + +import pytest + +from artemis.runtime.task_limits import ExecutionBudget, task_timeout +from artemis.runtime.task_result import save_failure_result + + +def test_deadline_precedence_and_single_task_default(monkeypatch): + monkeypatch.setattr("artemis.runtime.task_limits.settings.ARTEMIS_TASK_TIMEOUT_SECONDS", 1800) + assert task_timeout({}) == 1800 + job = {"case": {"task_config": {"timeout_seconds": 120}}} + assert task_timeout({}, job, {"timeout_seconds": 300}) == 120 + assert task_timeout({"runtime_overrides": {"timeout_seconds": 60}}, job) == 60 + assert task_timeout({}, {}, {"timeout_seconds": 300}) == 300 + + +def test_budget_excludes_device_queue_and_user_pause(monkeypatch): + clock = Mock(return_value=0.0) + monkeypatch.setattr("artemis.runtime.task_limits.time.monotonic", clock) + budget = ExecutionBudget(30, _last=0) + budget.update(excluded=True) + clock.return_value = 3600 + assert budget.update(excluded=False) == 30 + clock.return_value = 3610 + assert budget.update(excluded=True) == 20 + clock.return_value = 7200 + assert budget.update(excluded=False) == 20 + clock.return_value = 7221 + assert budget.update() == 0 + + +def test_failure_report_replaces_empty_result_atomically(tmp_path, monkeypatch): + monkeypatch.setattr("artemis.runtime.task_result.TRACES_PATH", tmp_path) + directory = tmp_path / "test-task" + directory.mkdir() + (directory / "result.json").touch() + save_failure_result("test-task", "设备截图失败：ADB gone", code="device_observation_failed") + for name in ("result.json", "run_outcome.json"): + payload = json.loads((directory / name).read_text(encoding="utf-8")) + assert payload["task_status"] == "failed" + assert "ADB gone" in payload["error"] + assert not list(directory.glob("*.tmp")) + (directory / "result.json").write_text( + '{"tests":{"checkpoint":{"passed":false}}}', encoding="utf-8" + ) + save_failure_result("test-task", "任务超时", code="task_timeout") + payload = json.loads((directory / "result.json").read_text(encoding="utf-8")) + assert payload["tests"]["checkpoint"]["passed"] is False + + +def test_checker_outcome_cannot_trigger_worker_cleanup(tmp_path, monkeypatch): + from artemis.runtime.task_result import read_worker_outcome + + monkeypatch.setattr("artemis.runtime.task_result.TRACES_PATH", tmp_path) + directory = tmp_path / "checking" + directory.mkdir() + for status in ("failed", "completed"): + (directory / "run_outcome.json").write_text( + json.dumps({"task_status": status}), encoding="utf-8" + ) + assert read_worker_outcome("checking") == {} + save_failure_result("checking", "设备故障") + assert read_worker_outcome("checking")["worker_finalized"] is True + + +@pytest.mark.asyncio +async def test_single_task_hung_worker_is_killed_and_records_timeout(tmp_path, monkeypatch): + import asyncio + from apps.admin_console.services.task_queue_service import TaskQueueService + + monkeypatch.setattr("artemis.runtime.task_result.TRACES_PATH", tmp_path) + monkeypatch.setattr( + "apps.admin_console.services.automation_store.store", Mock(job=Mock(return_value=None)) + ) + monkeypatch.setattr("artemis.runtime.task_limits.task_timeout", lambda *args: 0) + monkeypatch.setattr("artemis.runtime.task_control.read_state", lambda *args: {}) + monkeypatch.setattr(TaskQueueService, "_sync_control_states", lambda: None) + stopped = asyncio.Event() + + async def wait(proc): + await stopped.wait() + return -9 + + kill = AsyncMock(side_effect=lambda proc: stopped.set()) + monkeypatch.setattr(TaskQueueService, "_wait_for_worker_process", wait) + monkeypatch.setattr(TaskQueueService, "_terminate_worker_process", kill) + item = {"session_id": "hung-single"} + assert ( + await asyncio.wait_for( + TaskQueueService._await_worker_with_budget(Mock(), item, "hung-single"), 2 + ) + == -9 + ) + kill.assert_awaited_once() + assert item["timed_out"] + result = json.loads((tmp_path / "hung-single" / "result.json").read_text(encoding="utf-8")) + assert result["code"] == "task_timeout" + assert "不含暂停/排队" in result["error"] + + +def test_standalone_mcp_worker_deadline_kills_before_report_write(monkeypatch): + from mcp_server.tools import task_runner + + proc = Mock(pid=999999, poll=Mock(return_value=None)) + monkeypatch.setattr(task_runner.trace_store, "read_status", lambda *_: {"status": "running"}) + monkeypatch.setattr(task_runner, "read_state", lambda *_: {"status": "running"}) + monkeypatch.setattr( + task_runner.DeviceExecutionLock, + "get_active_owners", + lambda: {"device": Mock(session_id="mcp-hung")}, + ) + events = [] + monkeypatch.setattr(task_runner, "_kill_process_tree", lambda *_: events.append("kill")) + + def fail_report(*args, **kwargs): + events.append("report") + raise OSError("disk full") + + monkeypatch.setattr(task_runner, "save_failure_result", fail_report) + task_runner._watch_execution(proc, "mcp-hung", 0) + assert events == ["kill", "report"] + + +def test_mcp_finalized_worker_is_reclaimed_without_overwriting_device_failure(monkeypatch): + from mcp_server.tools import task_runner + + outcome = {"task_status": "failed", "error": "设备断连", "code": "device_observation_failed"} + monkeypatch.setattr(task_runner, "read_worker_outcome", lambda _: outcome) + monkeypatch.setattr(task_runner.trace_store, "read_status", lambda _: {"status": "running"}) + updated = Mock() + monkeypatch.setattr(task_runner.trace_store, "update_trace_status", updated) + kill = Mock() + monkeypatch.setattr(task_runner, "_kill_process_tree", kill) + clock = Mock(side_effect=[0.0, 0.1]) + monkeypatch.setattr(task_runner.time, "monotonic", clock) + task_runner._watch_execution( + Mock(pid=123, poll=Mock(return_value=None)), "test-mcp", 1800, cleanup_limit=0.01 + ) + kill.assert_called_once_with(123) + updated.assert_called_once_with("test-mcp", "failed", error="设备断连") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("finalized", [False, True]) +async def test_real_hung_process_stops_and_next_worker_can_complete( + tmp_path, monkeypatch, finalized +): + import asyncio + import sys + from apps.admin_console.services.task_queue_service import TaskQueueService + + monkeypatch.setattr("artemis.runtime.task_result.TRACES_PATH", tmp_path) + monkeypatch.setattr( + "apps.admin_console.services.automation_store.store", Mock(job=Mock(return_value=None)) + ) + monkeypatch.setattr( + "artemis.runtime.task_limits.task_timeout", + lambda item, *args: 0 if item["session_id"] == "hung" and not finalized else 30, + ) + monkeypatch.setattr("artemis.runtime.task_control.read_state", lambda *args: {}) + monkeypatch.setattr(TaskQueueService, "_sync_control_states", lambda: None) + item = {"session_id": "hung", "runtime_overrides": {"worker_cleanup_timeout_seconds": 0.01}} + if finalized: + save_failure_result("hung", "设备截图失效", code="device_observation_failed") + hung = await asyncio.create_subprocess_exec( + sys.executable, "-c", "import time; time.sleep(3600)" + ) + try: + code = await asyncio.wait_for( + TaskQueueService._await_worker_with_budget(hung, item, "hung"), 10 + ) + assert code != 0 + assert hung.returncode is not None + result = json.loads((tmp_path / "hung" / "result.json").read_text(encoding="utf-8")) + assert result["code"] == ("device_observation_failed" if finalized else "task_timeout") + assert bool(item.get("timed_out")) is not finalized + next_worker = await asyncio.create_subprocess_exec(sys.executable, "-c", "pass") + assert ( + await asyncio.wait_for( + TaskQueueService._await_worker_with_budget( + next_worker, {"session_id": "next"}, "next" + ), + 10, + ) + == 0 + ) + finally: + if hung.returncode is None: + hung.kill() + await hung.wait() diff --git a/tests/unit/tools/test_utils.py b/tests/unit/tools/test_utils.py index 63b9fd9..42b42d4 100644 --- a/tests/unit/tools/test_utils.py +++ b/tests/unit/tools/test_utils.py @@ -47,6 +47,17 @@ else: del sys.modules["artemis.graph.state"] +import base64 +from io import BytesIO +from PIL import Image + + +def _valid_screen_base64(): + buffer = BytesIO() + Image.new("RGB", (64, 64), "white").save(buffer, format="PNG") + return base64.b64encode(buffer.getvalue()).decode("ascii") + + @pytest.fixture def mock_context(): """Create a mock ArtemisContext for testing.""" @@ -70,6 +81,7 @@ def mock_context(): # Mock the ADB client for Android mock_response = Mock() + mock_response.base64 = _valid_screen_base64() mock_response.json.return_value = {"elements": []} ctx.ui_adb_client.get_screen_data = Mock(return_value=mock_response) ctx.ui_adb_client.get_hierarchy = Mock( @@ -288,6 +300,7 @@ class TestFocusElementIfNeeded: focused_element["attributes"]["focused"] = "true" mock_response = Mock() + mock_response.base64 = _valid_screen_base64() mock_response.json.return_value = {"elements": [focused_element]} mock_context.ui_adb_client.get_screen_data = Mock(return_value=mock_response) mock_find_element.return_value = focused_element["attributes"] @@ -366,6 +379,7 @@ class TestFocusElementIfNeeded: } mock_response = Mock() + mock_response.base64 = _valid_screen_base64() mock_response.json.return_value = {"elements": [element_from_text]} mock_context.ui_adb_client.get_screen_data = Mock(return_value=mock_response) mock_find_id.return_value = element_from_id @@ -401,6 +415,7 @@ class TestFocusElementIfNeeded: } mock_response = Mock() + mock_response.base64 = _valid_screen_base64() mock_response.json.return_value = {"elements": [element_with_bounds]} mock_context.ui_adb_client.get_screen_data = Mock(return_value=mock_response) mock_find_text.return_value = element_with_bounds["attributes"] @@ -428,6 +443,7 @@ class TestFocusElementIfNeeded: """Test failure when no locator can find an element.""" mock_response = Mock() + mock_response.base64 = _valid_screen_base64() mock_response.json.return_value = {"elements": []} mock_context.ui_adb_client.get_screen_data = Mock(return_value=mock_response) with (