From baf069854bc146fc074b4cbe388c8a6e0f65a3b7 Mon Sep 17 00:00:00 2001 From: chenmingxuan Date: Wed, 28 Oct 2026 11:50:05 +0800 Subject: [PATCH] fix(runtime): recover awake ownership and batch model failures --- DESIGN.md | 4 + .../services/automation_service.py | 46 +++++++- .../services/task_queue_service.py | 6 +- artemis/agents/flash/runner.py | 25 +++- artemis/agents/operator/operator.py | 16 ++- artemis/interfaces/cli/commands/run.py | 8 ++ artemis/runtime/awake_service.py | 19 ++- artemis/sdk/agent.py | 8 +- artemis/services/llm.py | 36 ++++-- docs/plans/2026-10-28-runtime-recovery.md | 33 ++++++ tests/unit/admin_console/test_automation.py | 110 +++++++++++++++++- tests/unit/agents/test_flash_runner_ledger.py | 38 ++++++ tests/unit/agents/test_operator_recovery.py | 25 ++++ tests/unit/runtime/test_awake_service.py | 60 ++++++++++ .../unit/sdk/test_agent_trace_finalization.py | 12 +- tests/unit/test_cli.py | 29 +++++ tests/unit/test_llm_pause.py | 34 ++++++ 17 files changed, 477 insertions(+), 32 deletions(-) create mode 100644 docs/plans/2026-10-28-runtime-recovery.md diff --git a/DESIGN.md b/DESIGN.md index b2142d8..34dbbc5 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -66,6 +66,10 @@ ARTEMIS 面向使用自然语言执行、探索和复跑测试的人。当前产 请求重试、业务执行与最终验收分别处理。提交使用稳定标识；创建批次或探索草稿不自动执行，查询不产生设备动作。执行准入和恢复遵守现有资源锁，任务固定所用插件版本，不能通过另建队列绕开互斥。 +设备保活是辅助服务。保活所属锁或旧保活清理目录不可访问时记录警告，不使设备连接、恢复或监控退出；未取得保活所属锁时不发送保活动作，也不创建替代锁绕过所属权。执行资源锁、主服务注册锁和队列写锁仍须成功取得。 + +运行时准备阶段失败记录 `initialization_failed`，批次保留详细原因并暂停未开始项，修复环境后由用户继续，已失败项不自动重跑。模型请求导致的批次暂停单独标记；模型暂停结束后恢复既定调度及失败即停止策略。用户暂停、停止全部任务和服务重启后的恢复暂停均需用户继续，不被模型恢复覆盖。Flash 与 Pro/探索 Operator 的生成失败复用统一备用模型与重试预算；不完整输出不会成为动作，已有动作不因模型切换而重放。 + 最终验收有独立总预算，按请求超时、重试和剩余任务时间协调，保留取证结果；完整有效的结构化结论才更新正式裁定。`completed`、工具成功、探索通过、用例提炼和复跑通过各自表达自己的状态。 Responses 消息交给工具、下游解析器或正式裁定前，必须具有服务端 `completed` 状态；工具参数的原始字符串必须是完整 JSON 对象，不能使用库自动补全的半截参数。流结束或库补出的 `last` 标记不能替代完成状态。缺少终止事件、明确未完成或失败的流丢弃预览，进入现有无效输出恢复策略，最多重新请求一次，配置为零次重试时不重试；恢复保留已有工具证据，不重新提交业务任务。非流式、结构化输出和串接解析器遵守同一检查。 diff --git a/apps/admin_console/services/automation_service.py b/apps/admin_console/services/automation_service.py index f40bece..0c6be11 100644 --- a/apps/admin_console/services/automation_service.py +++ b/apps/admin_console/services/automation_service.py @@ -15,6 +15,7 @@ from artemis.runtime.task_limits import task_timeout logger = logging.getLogger(__name__) TERMINAL = {"completed", "failed", "cancelled", "skipped", "interrupted"} +MODEL_PAUSE_REASON = "模型请求已暂停，请先恢复模型连接再继续批次" background_tasks: set[asyncio.Task] = set() @@ -123,7 +124,9 @@ def recover_batches(): if any(str(job["session_id"]) in live_sessions for job in store.jobs(batch["id"])): continue batch.update( - status="paused", reason="服务已重启；请检查中断任务，点击继续执行未开始任务" + status="paused", + reason="服务已重启；请检查中断任务，点击继续执行未开始任务", + pause_source="recovery", ) for job in store.jobs(batch["id"]): if job["status"] in {"running", "queued", "finalizing", "task_paused", "resuming"}: @@ -164,9 +167,29 @@ async def on_terminal(item: dict, status: str): } ): job["status"] = "failed" - job["error"] = job.get("error") or "任务执行结束，但业务检查未通过" + job["error"] = ( + job.get("error") + or report.get("error") + or (report.get("test_result") or {}).get("reason") + or "任务未完成或验证未通过，请查看报告中的具体原因" + ) elif job["status"] == "failed" and report.get("error") and not job.get("error"): job["error"] = report["error"] + if job["status"] == "failed" and report["outcome"].get("code") == "initialization_failed": + job["failure_code"] = "initialization_failed" + batch = store.get("batch", job.get("batch_id")) if job.get("batch_id") else None + if ( + batch + and (batch["status"] == "running" or batch.get("pause_source") == "model") + and any(j["status"] == "pending" for j in store.jobs(batch["id"])) + ): + batch.update( + status="paused", + pause_source="initialization", + reason="任务初始化失败；请处理环境问题后继续剩余用例：" + + (job.get("error") or "请查看任务报告"), + ) + store.put("batch", batch["id"], batch) # Persist terminal classification before export, but the coordinator waits for finalization. store.save_job({**job, "status": "finalizing", "terminal_status": job["status"]}) await asyncio.to_thread(export_session, sid, job["formats"]) @@ -217,10 +240,18 @@ async def tick_batches(): if batch["status"] not in {"running", "paused"}: continue jobs = store.jobs(batch["id"]) + if ( + batch["status"] == "paused" + and batch.get("pause_source") == "model" + and not state.is_paused + ): + batch.update(status="running", reason="") + batch.pop("pause_source", None) + store.put("batch", batch["id"], batch) active = [j for j in jobs if j["status"] in {"running", "queued", "finalizing"}] if active: - if state.is_paused: - batch.update(status="paused", reason="模型请求已暂停，请先恢复模型连接再继续批次") + if state.is_paused and batch["status"] == "running": + batch.update(status="paused", reason=MODEL_PAUSE_REASON, pause_source="model") store.put("batch", batch["id"], batch) for job in active: # The queue records started_at at submission, then replaces it @@ -284,7 +315,7 @@ async def tick_batches(): store.put("batch", batch["id"], batch) continue if state.is_paused: - batch.update(status="paused", reason="模型请求已暂停，请先恢复模型连接再继续批次") + batch.update(status="paused", reason=MODEL_PAUSE_REASON, pause_source="model") store.put("batch", batch["id"], batch) continue job = pending[0] @@ -438,11 +469,14 @@ def batch_action(batch_id: str, action: str) -> dict: if action == "pause": if batch["status"] not in {"running", "paused"}: raise ValueError("只能暂停正在执行的批次") - batch.update(status="paused", reason="用户暂停；当前任务继续，后续任务等待") + batch.update( + status="paused", reason="用户暂停；当前任务继续，后续任务等待", pause_source="user" + ) elif action == "resume": if batch["status"] not in {"draft", "paused"}: raise ValueError("只能开始草稿或继续已暂停批次") batch.update(status="running", reason="") + batch.pop("pause_source", None) elif action == "retry_failed": if batch["status"] == "running": raise ValueError("请先暂停批次") diff --git a/apps/admin_console/services/task_queue_service.py b/apps/admin_console/services/task_queue_service.py index 5b4f720..a080ef4 100644 --- a/apps/admin_console/services/task_queue_service.py +++ b/apps/admin_console/services/task_queue_service.py @@ -1522,8 +1522,10 @@ class TaskQueueService: from artemis.plugins.store import PluginStore for batch in store.records("batch"): - if batch["status"] == "running": - batch.update(status="paused", reason="已停止全部任务，等待手动继续") + if batch["status"] in {"running", "paused"}: + batch.update( + status="paused", reason="已停止全部任务，等待手动继续", pause_source="user" + ) store.put("batch", batch["id"], batch) # 1. Cancel local queue reservations for item in state.queue_items: diff --git a/artemis/agents/flash/runner.py b/artemis/agents/flash/runner.py index 85d9587..6b67d65 100644 --- a/artemis/agents/flash/runner.py +++ b/artemis/agents/flash/runner.py @@ -96,6 +96,7 @@ from artemis.services.llm import ( get_openai_llm, get_llm, invoke_llm_with_timeout_message, + with_fallback, ) from artemis.tools.history import history_tool_declarations from artemis.tools.android_files import android_file @@ -1506,20 +1507,32 @@ class FlashRunner: turn_base = len(messages) - 1 current_tools = report_only_tools if is_final else self._get_tools() + async def invoke_turn(): + if using_tool_fallback: + return await self._invoke_model(fallback_llm, current_tools, messages) + + async def invoke_fallback(): + nonlocal fallback_llm + if fallback_llm is None: + fallback_llm = get_llm(self.ctx, name="operator", use_fallback=True) + return await self._invoke_model(fallback_llm, current_tools, messages) + + return await with_fallback( + lambda: self._invoke_model(llm, current_tools, messages), + invoke_fallback, + none_should_fallback=False, + ) + if isinstance(control, TaskControl): interrupted, response = await control.run_interruptibly( - self._invoke_model( - fallback_llm if using_tool_fallback else llm, current_tools, messages - ), + invoke_turn(), on_pause=getattr(self.ctx, "on_task_pause", None), on_resume=getattr(self.ctx, "on_task_resume", None), ) else: interrupted, response = ( False, - await self._invoke_model( - fallback_llm if using_tool_fallback else llm, current_tools, messages - ), + await invoke_turn(), ) if interrupted or ( diff --git a/artemis/agents/operator/operator.py b/artemis/agents/operator/operator.py index 4afb7a3..e21b63d 100644 --- a/artemis/agents/operator/operator.py +++ b/artemis/agents/operator/operator.py @@ -35,7 +35,7 @@ from artemis.mcp.action_specs import OPERATOR_SHELL_ORDER, operator_shell_tool from artemis.llm.usage import usage_from_message from artemis.llm.structured import content_to_text from artemis.runtime.task_control import TaskControl, has_pending_instruction -from artemis.services.llm import acomplete, get_llm, invoke_llm_with_timeout_message +from artemis.services.llm import acomplete, get_llm, invoke_llm_with_timeout_message, with_fallback from artemis.tools.command_tool import ( analyze_task_output_wrapper, get_adb_task_registry, @@ -638,9 +638,19 @@ class OperatorNode: and t.name not in {"plugin_load", "plugin_read_file", "plugin_read_result"} ] traced_tools.extend(plugins.langchain_tools()) - bound_llm = active_llm.bind_tools(tools=traced_tools) + + async def invoke(model): + bound = model.bind_tools(tools=traced_tools) + return await invoke_llm_with_timeout_message(acomplete(bound, current_messages)) + interrupted, response = await self._await_or_pause( - invoke_llm_with_timeout_message(acomplete(bound_llm, current_messages)) + invoke(active_llm) + if using_tool_fallback + else with_fallback( + lambda: invoke(base_llm), + lambda: invoke(get_llm(ctx=self.ctx, name="operator", use_fallback=True)), + none_should_fallback=False, + ) ) if interrupted: return None, None, None, False diff --git a/artemis/interfaces/cli/commands/run.py b/artemis/interfaces/cli/commands/run.py index 5f53c4c..71fc26d 100644 --- a/artemis/interfaces/cli/commands/run.py +++ b/artemis/interfaces/cli/commands/run.py @@ -164,6 +164,7 @@ async def execute_task( config.with_graph_config_callbacks(graph_config_callbacks) agent: Agent | None = None + task_started = False try: agent = Agent(config=config.build(), session_id=effective_sid) await agent.init( @@ -196,7 +197,14 @@ async def execute_task( request.plugins = plugins elif plugin_choices is not None: request.plugins = TypeAdapter(list[PluginSelection]).validate_json(plugin_choices) + task_started = True await agent.run_task(request=request) + except Exception as exc: + if not task_started and effective_sid: + from artemis.runtime.task_result import save_failure_result + + save_failure_result(str(effective_sid), str(exc), code="initialization_failed") + raise finally: if agent is not None: await agent.clean() diff --git a/artemis/runtime/awake_service.py b/artemis/runtime/awake_service.py index db4eac3..411778e 100644 --- a/artemis/runtime/awake_service.py +++ b/artemis/runtime/awake_service.py @@ -101,7 +101,10 @@ def _configure_usb_stay_awake(device_id: str) -> str | None: # Migration cleanup belongs to the long-lived host. A queued task worker # must not wait on another process's optional legacy wake-lock cleanup. if os.environ.get("ARTEMIS_TASK_WORKER") != "1": - ScreenAwakeLease(device_id).cleanup_unowned_references() + try: + ScreenAwakeLease(device_id).cleanup_unowned_references() + except OSError as exc: + logger.warning(f"Could not prepare optional legacy awake cleanup on {device_id}: {exc}") commands = ( ( @@ -209,6 +212,7 @@ class ScreenAwakeService: self._monitor_thread: threading.Thread | None = None self._shutdown_requested = False self._owners: dict[str, FileLock] = {} + self._ownership_warnings: set[str] = set() def start(self, device_ids: Iterable[str] | None = None) -> dict[str, str | None]: if self._shutdown_requested: @@ -241,13 +245,24 @@ class ScreenAwakeService: existing = self._strategies.get(device_id) if existing is not None: return existing - owner = _owner_lock(device_id) try: + owner = _owner_lock(device_id) owner.acquire(timeout=0) except Timeout: # Do not cache a follower as owner: discovery can take over if # the process holding the OS lock exits or releases the device. return None + except OSError as exc: + # Awake is optional. Never bypass ownership or let an ACL / I/O + # failure abort device connection, resume, or the host monitor. + if device_id not in self._ownership_warnings: + logger.warning( + f"Could not acquire awake ownership on {device_id}; skipping optional " + f"stay-awake policy: {exc}" + ) + self._ownership_warnings.add(device_id) + return None + self._ownership_warnings.discard(device_id) try: strategy = _configure_usb_stay_awake(device_id) except BaseException: diff --git a/artemis/sdk/agent.py b/artemis/sdk/agent.py index 31f3892..2be394b 100644 --- a/artemis/sdk/agent.py +++ b/artemis/sdk/agent.py @@ -584,6 +584,7 @@ class Agent: or getattr(getattr(task, "request", None), "task_name", None) ) device_lock = None + execution_started = False try: # Construct the runtime only inside the task's cleanup boundary. A # cancelled queue wait or setup failure must not leave version pins. @@ -761,6 +762,7 @@ class Agent: publish_startup_progress( "environment_ready", "Device environment is ready", session_id=str(sess_id) ) + execution_started = True try: if request.profile and request.profile.lower() == "flash": logger.info(f"[{task_name}] Invoking FlashRunner reactive loop...") @@ -972,7 +974,11 @@ class Agent: from artemis.runtime.task_result import save_failure_result if sess_id: - save_failure_result(str(sess_id), err) + save_failure_result( + str(sess_id), + err, + code="execution_failed" if execution_started else "initialization_failed", + ) if context.data_engine: context.data_engine.end_session("failed") diff --git a/artemis/services/llm.py b/artemis/services/llm.py index 73444ad..8dcd0b9 100644 --- a/artemis/services/llm.py +++ b/artemis/services/llm.py @@ -317,12 +317,32 @@ async def _wait_for_resume(pause_file: Path) -> bool: """Wait for the pause file to be cleared. Returns False on deadline.""" deadline = float(getattr(settings, "LLM_PAUSE_TIMEOUT_SECONDS", 0.0) or 0.0) waited = 0.0 - while pause_file.exists(): - if deadline > 0 and waited >= deadline: - return False - await asyncio.sleep(1) - waited += 1 - return True + try: + owned = pause_file.stat() + except FileNotFoundError: + return True + try: + while pause_file.exists(): + if deadline > 0 and waited >= deadline: + return False + await asyncio.sleep(1) + waited += 1 + return True + finally: + # Cancellation (stop, guidance, outer deadline) also ends this wait. + # Do not leave a global pause behind or remove a newer request's pause. + try: + current = pause_file.stat() + if (current.st_ino, current.st_mtime_ns, current.st_size) == ( + owned.st_ino, + owned.st_mtime_ns, + owned.st_size, + ): + pause_file.unlink(missing_ok=True) + except FileNotFoundError: + pass + except OSError as exc: + llm_logger.warning("Could not clear finished LLM pause: %s", exc) async def _wait_for_breaker(key: str) -> None: @@ -510,10 +530,6 @@ async def _run_with_recovery[T]( "LLM pause deadline" f" ({settings.LLM_PAUSE_TIMEOUT_SECONDS:.0f}s) exceeded; giving up." ) - try: - pause_file.unlink(missing_ok=True) - except OSError: - pass raise LLMExhaustedError( str(last_error), failure=last_failure, cause=last_error ) from last_error diff --git a/docs/plans/2026-10-28-runtime-recovery.md b/docs/plans/2026-10-28-runtime-recovery.md new file mode 100644 index 0000000..cb15920 --- /dev/null +++ b/docs/plans/2026-10-28-runtime-recovery.md @@ -0,0 +1,33 @@ +# 2026-10-28 运行时故障与批次恢复修复 + +## 已确认问题 + +- awake 所属锁的目录创建和获取只处理正常争用，`PermissionError` 可从 UIAutomator、无障碍连接、设备恢复和主程序监控的共享入口传播。 +- 旧唤醒租约清理虽然处理了清理过程异常，构造时创建临时目录仍可能失败。 +- Flash、Pro/探索 Operator 的工具无进展恢复有备用模型，但模型生成异常未进入该入口。 +- 批次模型暂停与用户暂停未区分；模型等待超时后文件已清除，批次仍维持暂停。 +- 模型等待被取消时可能残留暂停文件；终态报告的通用失败说明可能遮蔽真实初始化/模型错误。 + +## 修复范围 + +- 保活权限/I/O 异常降级为警告。未取得所属锁时不执行保活动作；后续可重新获取，不缓存为成功。同一设备持续获取失败只警告一次，成功后重置。 +- 工作进程早期 Agent 准备及 SDK 运行时准备失败记录 `initialization_failed`。批次在报告导出前保存失败原因并暂停剩余项；导出失败不改变这一分类。 +- Flash 和 Operator 接入现有 `with_fallback`。取消、永久请求错误和同端点重试预算仍遵守统一网关规则，不修改用户模型配置，不放宽工具参数/完整响应保护。 +- 区分模型、初始化、用户、服务恢复暂停。模型恢复后按既定失败策略继续；停止全部任务覆盖自动恢复标记。重启后没有存活任务时仍要求人工继续。 +- 模型等待结束或被取消时清理本次暂停；文件已被新请求改写时保留。 + +## 边界 + +- 不删除或改 ACL，不更换锁目录绕过互斥。设备执行锁、主程序注册锁和队列 I/O 锁保留失败即阻止执行的行为。 +- 未重跑原批次、修改历史裁定或调用真实设备/付费模型；未构建 Docker。 +- 原 Windows 锁文件的 ACL、占用者及异常发生时依赖版本尚无本机证据；本修复处理应用传播和恢复缺口，不宣称证明了操作系统拒绝访问的具体来源。 +- 旧批次不自动迁移暂停原因，需用户显式继续；不会从历史零步骤数量推断并改写失败分类。 + +## 验证 + +针对性测试覆盖锁目录/文件权限拒绝后恢复、监控继续处理其他设备、旧保活清理、Flash/Operator 生成失败切换、暂停取消/超时、初始化错误保留与批次暂停、手动继续不重跑失败项、用户暂停与停止全部、服务重启恢复。 + +- 默认测试：3225 通过、6 跳过、8 按默认标记排除。102 条既有 Pillow `getdata()` 弃用警告。 +- Ruff 格式检查与静态检查通过，质量基线通过，核心 Pyright 检查 0 错误、0 警告。 +- 第一轮完整测试暴露新增 CLI 测试的环境变量隔离遗漏，影响两项录像测试；修正测试隔离后第二轮完整测试全部通过，未为此改变产品行为或放宽断言。 +- 本次仅修改 Python 与文档，未运行前端构建、真实设备或付费模型测试。 diff --git a/tests/unit/admin_console/test_automation.py b/tests/unit/admin_console/test_automation.py index 7bad91e..7c53abd 100644 --- a/tests/unit/admin_console/test_automation.py +++ b/tests/unit/admin_console/test_automation.py @@ -417,13 +417,21 @@ def test_batch_deletion_preserves_shared_screenshot_until_last_task(client, isol assert not image_file.exists() -def test_restart_never_repeats_started_work(isolated): +@pytest.mark.parametrize("model_paused", [False, True]) +def test_restart_never_repeats_started_work(isolated, model_paused): batch = service.create_batch(payload()) service.batch_action(batch["id"], "resume") + if model_paused: + isolated.put( + "batch", + batch["id"], + {**isolated.get("batch", batch["id"]), "status": "paused", "pause_source": "model"}, + ) job = isolated.jobs(batch["id"])[0] isolated.save_job({**job, "status": "queued"}) service.recover_batches() assert isolated.get("batch", batch["id"])["status"] == "paused" + assert isolated.get("batch", batch["id"])["pause_source"] == "recovery" assert [j["status"] for j in isolated.jobs(batch["id"])] == ["interrupted", "pending"] retry = service.batch_action(batch["id"], "retry_failed") assert retry["status"] == "draft" @@ -603,6 +611,106 @@ async def test_device_outage_pauses_without_consuming_pending_case(isolated, mon assert all(job["status"] == "pending" for job in isolated.jobs(batch["id"])) +def test_stop_all_overrides_model_auto_resume(isolated, monkeypatch): + from apps.admin_console.services import task_queue_service as queue + + batch = service.create_batch(payload()) + isolated.put("batch", batch["id"], {**batch, "status": "paused", "pause_source": "model"}) + monkeypatch.setattr("apps.admin_console.services.automation_store.store", isolated) + monkeypatch.setattr(queue, "state", Mock(queue_items=[], active_connections={})) + for method in ( + "_terminate_all_device_owners", + "_kill_all_local_runs", + "_clear_pause_file", + "ensure_worker_running", + ): + monkeypatch.setattr(queue.TaskQueueService, method, Mock()) + assert queue.TaskQueueService._stop_all_tasks() + saved = isolated.get("batch", batch["id"]) + assert saved["pause_source"] == "user" + assert "等待手动继续" in saved["reason"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("manual_pause", [False, True]) +@pytest.mark.parametrize("stop_on_failure", [False, True]) +async def test_model_pause_recovery_preserves_user_pause_and_failure_policy( + isolated, monkeypatch, manual_pause, stop_on_failure +): + from apps.admin_console import server + from artemis.runtime.device_pool import DeviceStatus + + batch = service.create_batch(payload(stop_on_failure=stop_on_failure)) + service.batch_action(batch["id"], "resume") + first, second = isolated.jobs(batch["id"]) + isolated.save_job({**first, "status": "running", "started_at": time.time()}) + pause_state = type("State", (), {"is_paused": True, "active_runs": {}})() + monkeypatch.setattr(service, "state", pause_state) + enqueue = AsyncMock(return_value={"tasks": [{"device_serial": "test-device"}]}) + monkeypatch.setattr(server.task_queue_service, "enqueue_tasks", enqueue) + monkeypatch.setattr( + "artemis.runtime.device_pool.try_list_devices_async", + AsyncMock(return_value=[DeviceStatus(serial="test-device", state="device")]), + ) + monkeypatch.setattr(reports, "export_batch", Mock()) + await service.tick_batches() + assert isolated.get("batch", batch["id"])["pause_source"] == "model" + if manual_pause: + service.batch_action(batch["id"], "pause") + # Another paused-model tick must not overwrite the explicit user pause. + await service.tick_batches() + assert isolated.get("batch", batch["id"])["pause_source"] == "user" + isolated.save_job({**first, "status": "failed"}) + pause_state.is_paused = False + await service.tick_batches() + assert isolated.job(first["session_id"])["status"] == "failed" + expected = "pending" if manual_pause else "skipped" if stop_on_failure else "queued" + assert isolated.job(second["session_id"])["status"] == expected + assert enqueue.await_count == (0 if manual_pause or stop_on_failure else 1) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("code", ["initialization_failed", "execution_failed"]) +async def test_failure_reason_retained_and_only_initialization_blocks_remaining_batch( + isolated, monkeypatch, code +): + from apps.admin_console import server + from artemis.runtime import task_result + from artemis.runtime.device_pool import DeviceStatus + + batch = service.create_batch(payload()) + service.batch_action(batch["id"], "resume") + first, second = isolated.jobs(batch["id"]) + isolated.save_job({**first, "status": "running"}) + monkeypatch.setattr(task_result, "TRACES_PATH", reports.TRACES_PATH) + task_result.save_failure_result( + first["session_id"], "PermissionError: awake lock denied", code=code + ) + monkeypatch.setattr(reports, "export_session", Mock(side_effect=OSError("export unavailable"))) + monkeypatch.setattr(service, "state", type("State", (), {"is_paused": False})()) + enqueue = AsyncMock(return_value={"tasks": [{"device_serial": "test-device"}]}) + monkeypatch.setattr(server.task_queue_service, "enqueue_tasks", enqueue) + monkeypatch.setattr( + "artemis.runtime.device_pool.try_list_devices_async", + AsyncMock(return_value=[DeviceStatus(serial="test-device", state="device")]), + ) + + await service.on_terminal({"session_id": first["session_id"]}, "failed") + await service.tick_batches() + saved = isolated.job(first["session_id"]) + assert saved["error"] == "PermissionError: awake lock denied" + assert saved["report_error"] == "export unavailable" + if code == "initialization_failed": + assert isolated.get("batch", batch["id"])["pause_source"] == "initialization" + assert isolated.job(second["session_id"])["status"] == "pending" + enqueue.assert_not_awaited() + service.batch_action(batch["id"], "resume") + await service.tick_batches() + assert enqueue.await_count == 1 + assert enqueue.call_args.kwargs["session_id"] == second["session_id"] + assert isolated.job(first["session_id"])["status"] == "failed" + + @pytest.mark.asyncio async def test_fail_fast_includes_skipped_reports(isolated, monkeypatch): batch = service.create_batch(payload(stop_on_failure=True)) diff --git a/tests/unit/agents/test_flash_runner_ledger.py b/tests/unit/agents/test_flash_runner_ledger.py index 31d17dd..f3c5959 100644 --- a/tests/unit/agents/test_flash_runner_ledger.py +++ b/tests/unit/agents/test_flash_runner_ledger.py @@ -429,6 +429,44 @@ def _report(status="completed", tc_id="tc-report", **extra): _INITIAL_OBSERVATION = ("shot0.png", b"IMG0", "[1] Settings") +@pytest.mark.asyncio +async def test_generation_failure_uses_fallback_without_repeating_device_actions(mock_context): + from artemis.llm.response_guard import ResponseIntegrityError + from artemis.services import llm as gateway + + click = AIMessage( + content="tap", tool_calls=[{"name": "click", "args": {"index": 1}, "id": "click"}] + ) + with patch("artemis.controllers.unified_controller.get_driver"): + runner = _make_runner(mock_context, []) + primary = runner._init_llm.return_value + fallback = Mock() + calls = [] + + async def invoke(model, tools, messages): + calls.append(model) + if len(calls) == 1: + assert gateway._FALLBACK_AVAILABLE.get() + raise ResponseIntegrityError("missing completed response") + if model is fallback: + assert not gateway._FALLBACK_AVAILABLE.get() + return click + return _report() + + runner._invoke_model = AsyncMock(side_effect=invoke) + with ( + patch( + "artemis.agents.flash.runner.capture_screenshot_and_parse_ui", + AsyncMock(return_value=_INITIAL_OBSERVATION), + ), + patch("artemis.agents.flash.runner.get_llm", return_value=fallback), + ): + result = await runner.run(State(initial_goal="Open Settings")) + assert result["status"] == "completed" + assert calls == [primary, fallback, primary] + assert runner.executor.execute.await_count == 1 + + @pytest.mark.asyncio @pytest.mark.parametrize("fallback_succeeds", [True, False]) async def test_failed_actions_have_finite_fallback_and_restore_primary( diff --git a/tests/unit/agents/test_operator_recovery.py b/tests/unit/agents/test_operator_recovery.py index f42fb5a..b66fda2 100644 --- a/tests/unit/agents/test_operator_recovery.py +++ b/tests/unit/agents/test_operator_recovery.py @@ -82,6 +82,31 @@ async def run_loop(operator, monkeypatch, responses): return result, messages, invoke +@pytest.mark.asyncio +async def test_generation_failure_hands_over_before_pausing(operator, monkeypatch): + from artemis.llm.response_guard import ResponseIntegrityError + from artemis.services import llm as gateway + + node, state = operator + primary, fallback = Mock(), Mock() + contexts = [] + + async def invoke(model, messages): + contexts.append(gateway._FALLBACK_AVAILABLE.get()) + if model is primary.bind_tools.return_value: + raise ResponseIntegrityError("missing completed response") + assert model is fallback.bind_tools.return_value + return click() + + monkeypatch.setattr("artemis.agents.operator.operator.acomplete", invoke) + monkeypatch.setattr("artemis.agents.operator.operator.get_llm", Mock(return_value=fallback)) + messages = [] + result = await node._invoke_llm_loop(primary, messages, [], [], state) + assert contexts == [True, False] + assert len(result[0]) == 1 + assert len([m for m in messages if isinstance(m, AIMessage)]) == 1 + + @pytest.mark.asyncio async def test_prose_retry_keeps_context_and_recovers_once(operator, monkeypatch): prose = AIMessage(content="I will click the model menu.") diff --git a/tests/unit/runtime/test_awake_service.py b/tests/unit/runtime/test_awake_service.py index 759bfea..0cefe1c 100644 --- a/tests/unit/runtime/test_awake_service.py +++ b/tests/unit/runtime/test_awake_service.py @@ -24,6 +24,66 @@ def completed(stdout="", returncode=0, stderr=""): return MagicMock(returncode=returncode, stdout=stdout, stderr=stderr) +@pytest.mark.parametrize("stage", ["directory", "acquire"]) +def test_inaccessible_awake_lock_skips_actions_and_can_recover(stage, monkeypatch, tmp_path): + owner = FileLock(tmp_path / "recovered.lock", thread_local=False) + denied_owner = MagicMock() + denied_owner.acquire.side_effect = PermissionError("awake lock access denied") + factory = MagicMock( + side_effect=[ + PermissionError("awake directory denied") if stage == "directory" else denied_owner, + owner, + ] + ) + configure = MagicMock(return_value=AWAKE_STRATEGY_USB) + monkeypatch.setattr("artemis.runtime.awake_service._owner_lock", factory) + monkeypatch.setattr("artemis.runtime.awake_service._configure_usb_stay_awake", configure) + service = ScreenAwakeService() + try: + assert service.ensure_device("device") is None + configure.assert_not_called() + assert service.device_ids == () + assert service.ensure_device("device") == AWAKE_STRATEGY_USB + configure.assert_called_once_with("device") + finally: + service.shutdown() + + +def test_awake_monitor_continues_to_other_devices_after_access_denied(monkeypatch, tmp_path): + def owner(device): + if device == "denied": + raise PermissionError("denied") + return FileLock(tmp_path / f"{device}.lock", thread_local=False) + + monkeypatch.setattr("artemis.runtime.awake_service._owner_lock", owner) + configure = MagicMock(return_value=AWAKE_STRATEGY_USB) + monkeypatch.setattr("artemis.runtime.awake_service._configure_usb_stay_awake", configure) + service = ScreenAwakeService() + try: + service._reconcile_connected_devices({"denied", "healthy"}) + service._reconcile_connected_devices({"denied", "healthy"}) + assert service.device_ids == ("healthy",) + configure.assert_called_once_with("healthy") + finally: + service.shutdown() + + +@patch( + "artemis.runtime.awake_service.ScreenAwakeLease", + side_effect=PermissionError("legacy directory denied"), +) +@patch("artemis.runtime.awake_service.subprocess.run") +def test_optional_legacy_directory_failure_does_not_abort_usb_policy(mock_run, _lease): + mock_run.side_effect = [ + completed(), + completed(), + completed(), + completed("2"), + completed("mStayOn=true"), + ] + assert _configure_usb_stay_awake("device") == AWAKE_STRATEGY_USB + + @patch( "artemis.runtime.awake_service._configure_usb_stay_awake", return_value=AWAKE_STRATEGY_HEARTBEAT ) diff --git a/tests/unit/sdk/test_agent_trace_finalization.py b/tests/unit/sdk/test_agent_trace_finalization.py index e3b2b7e..ee12c09 100644 --- a/tests/unit/sdk/test_agent_trace_finalization.py +++ b/tests/unit/sdk/test_agent_trace_finalization.py @@ -64,7 +64,10 @@ async def test_device_context_uses_adb_size_without_starting_ui_client(): @pytest.mark.asyncio -async def test_task_that_never_acquires_queue_does_not_create_trace_session(): +async def test_task_that_never_acquires_queue_does_not_create_trace_session(tmp_path, monkeypatch): + from artemis.runtime import task_result + + monkeypatch.setattr(task_result, "TRACES_PATH", tmp_path) agent = Agent() agent._initialized = True agent._device_context = DeviceContext( @@ -86,6 +89,13 @@ async def test_task_that_never_acquires_queue_does_not_create_trace_session(): await agent.run_task(goal="queued task", profile="flash") agent._prepare_tracing.assert_not_called() + outcomes = list(tmp_path.glob("*/run_outcome.json")) + assert len(outcomes) == 1 + import json + + outcome = json.loads(outcomes[0].read_text(encoding="utf-8")) + assert outcome["code"] == "initialization_failed" + assert "queue cancelled" in outcome["error"] @pytest.mark.asyncio diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index c4a7350..013b6fe 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -163,6 +163,35 @@ def test_cli_trace_help(): assert "view" in result.output +@pytest.mark.asyncio +async def test_worker_initialization_failure_is_persisted_and_cleaned(monkeypatch): + from unittest.mock import AsyncMock, MagicMock + import artemis.interfaces.cli.commands.run as run_module + from artemis.runtime import task_result + + monkeypatch.setenv("ARTEMIS_SESSION_ID", "test-session") + monkeypatch.setenv("ARTEMIS_TASK_INGRESS", "test") + fake_agent = MagicMock() + fake_agent.init = AsyncMock(side_effect=PermissionError("device initialization denied")) + fake_agent.clean = AsyncMock() + fake_agent.run_task = AsyncMock() + monkeypatch.setattr(run_module, "initialize_llm_config", MagicMock()) + monkeypatch.setattr(run_module, "AgentProfile", MagicMock()) + monkeypatch.setattr(run_module, "Builders", MagicMock()) + monkeypatch.setattr(run_module, "Agent", MagicMock(return_value=fake_agent)) + save = MagicMock() + monkeypatch.setattr(task_result, "save_failure_result", save) + with pytest.raises(PermissionError, match="initialization denied"): + await run_module.execute_task( + "inspect", execution_target="computer", session_id="test-session" + ) + save.assert_called_once_with( + "test-session", "device initialization denied", code="initialization_failed" + ) + fake_agent.clean.assert_awaited_once() + fake_agent.run_task.assert_not_awaited() + + def test_cli_mcp_help(): """Verify 'artemis mcp --help' lists server options.""" result = runner.invoke(app, ["mcp", "--help"]) diff --git a/tests/unit/test_llm_pause.py b/tests/unit/test_llm_pause.py index cabc998..968dfbb 100644 --- a/tests/unit/test_llm_pause.py +++ b/tests/unit/test_llm_pause.py @@ -6,6 +6,40 @@ import pytest from artemis.services import llm +@pytest.mark.asyncio +@pytest.mark.parametrize("replace_pause", [False, True]) +async def test_cancelled_wait_clears_only_its_own_pause(tmp_path, monkeypatch, replace_pause): + pause = tmp_path / "pause" + pause.write_text("old request") + entered = asyncio.Event() + + async def wait(_seconds): + entered.set() + await asyncio.Event().wait() + + monkeypatch.setattr(llm.asyncio, "sleep", wait) + task = asyncio.create_task(llm._wait_for_resume(pause)) + await entered.wait() + if replace_pause: + pause.write_text("a newer request owns this pause") + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert pause.exists() is replace_pause + + +@pytest.mark.asyncio +async def test_pause_deadline_clears_file(tmp_path, monkeypatch): + from unittest.mock import AsyncMock + + pause = tmp_path / "pause" + pause.write_text("failed request") + monkeypatch.setattr(llm.settings, "LLM_PAUSE_TIMEOUT_SECONDS", 1) + monkeypatch.setattr(llm.asyncio, "sleep", AsyncMock()) + assert await llm._wait_for_resume(pause) is False + assert not pause.exists() + + @pytest.mark.asyncio async def test_timeout_wrapper_cancels_llm_call_when_caller_is_cancelled(): started = asyncio.Event()