chore: prepare ARTEMIS-Eapil 1.3.1 release

.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
@@ -40,6 +50,8 @@ ARTEMIS_CHECKPOINT_TIMEOUT_SECONDS=180
 ARTEMIS_SETTLEMENT_TIMEOUT_SECONDS=120
 ARTEMIS_CREDENTIAL_TIMEOUT_SECONDS=12
 ARTEMIS_MODEL_CACHE_TTL_SECONDS=10
+# 守护服务轻量探测超时；探测超时且端口仍占用时，不重复启动服务。
+ARTEMIS_DAEMON_PROBE_TIMEOUT_SECONDS=2
 # 模型重试耗尽后的暂停等待上限（秒）；0 表示无限等待
 LLM_PAUSE_TIMEOUT_SECONDS=900
 
@@ -56,6 +68,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

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,
     }
 
 

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

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):

apps/admin_console/server.py

@@ -24,6 +24,7 @@ import os
 from pathlib import Path
 import secrets
 import signal
+import socket
 import sys
 from types import FrameType
 
@@ -421,6 +422,14 @@ queue_goals = state.queue_goals
 class ArtemisUvicornServer(uvicorn.Server):
     """Notify application streams before Uvicorn waits for them to close."""
 
+    async def startup(self, sockets: list[socket.socket] | None = None) -> None:
+        # Reload workers start in a fresh interpreter; restore the actual bind
+        # address there before lifespan publishes runtime metadata.
+        state.host = self.config.host
+        state.port = self.config.port
+        app.state.uvicorn_server = self
+        await super().startup(sockets=sockets)
+
     @staticmethod
     def _task_is_active() -> bool:
         """Check task state without doing process I/O from a signal handler."""
@@ -470,31 +479,40 @@ class ArtemisUvicornServer(uvicorn.Server):
 
 def run_ui_server(host: str, port: int, reload: bool = False) -> None:
     """Run the UI server with bounded, signal-aware graceful shutdown."""
+    # Reserve a listening socket before publishing metadata or running lifespan.
+    # Uvicorn normally runs application startup *before* binding its port; a
+    # duplicate daemon would otherwise recover jobs owned by the real server.
+    family = socket.AF_INET6 if ":" in host else socket.AF_INET
+    try:
+        listener = socket.create_server((host, port), family=family)
+    except OSError as exc:
+        print(f"无法启动服务：{host}:{port} 端口不可用（{exc}）。现有服务和任务保持不变。")
+        raise SystemExit(1) from exc
+    listener.set_inheritable(True)
+    port = listener.getsockname()[1]
     state.host = host
     state.port = port
-    write_server_info(port=port, host=host, lifecycle_token=LIFECYCLE_TOKEN)
 
     try:
-        if reload:
-            uvicorn.run(
-                "apps.admin_console.server:app",
-                host=host,
-                port=port,
-                reload=True,
-                timeout_graceful_shutdown=5,
-            )
-            return
-
         config = uvicorn.Config(
-            app,
+            "apps.admin_console.server:app" if reload else app,
             host=host,
             port=port,
+            reload=reload,
             timeout_graceful_shutdown=5,
         )
+        if reload:
+            from uvicorn.supervisors import ChangeReload
+
+            server = ArtemisUvicornServer(config)
+            ChangeReload(config, target=server.run, sockets=[listener]).run()
+            return
+
         server = ArtemisUvicornServer(config)
         app.state.uvicorn_server = server
-        server.run()
+        server.run(sockets=[listener])
     finally:
+        listener.close()
         app.state.uvicorn_server = None
         clear_server_info(port=port, lifecycle_token=LIFECYCLE_TOKEN)
 

apps/admin_console/services/automation_service.py

@@ -8,8 +8,10 @@ import time
 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.automation_store import RECOVERY_ERRORS, 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"}
@@ -40,9 +42,46 @@ def register_submission(item: dict, case: dict | None = None, formats=None, **ex
     )
 
 
+def resolve_recovery_error(job: dict) -> bool:
+    """Archive a superseded recovery warning without erasing execution failures."""
+    if job.get("error") not in RECOVERY_ERRORS:
+        return False
+    job.setdefault("recovery_events", []).append(
+        {"message": job.pop("error"), "resolved_at": time.time()}
+    )
+    return True
+
+
 def recover_batches():
     """Recovery never silently repeats a potentially side-effecting interrupted task."""
+    from apps.admin_console.database.repositories.session_repository import SessionRepository
+    from artemis.runtime import DeviceExecutionLock
+
+    repository = SessionRepository(store.path)
+    live_sessions = {
+        str(owner.session_id)
+        for owner in DeviceExecutionLock.get_active_owners().values()
+        if owner.session_id
+    }
+    live_sessions.update(
+        str(row["session_id"])
+        for row in repository.get_all_sessions()
+        if row.get("status") not in TERMINAL and repository.process_is_alive(row.get("pid"))
+    )
+    live_sessions.update(
+        str(item["session_id"])
+        for item in state.queue_items
+        if isinstance(item, dict) and item.get("session_id") and item.get("status") not in TERMINAL
+    )
     for job in store.jobs():
+        if job["status"] == "completed" and resolve_recovery_error(job):
+            store.save_job(job)
+        if str(job["session_id"]) in live_sessions:
+            if job["status"] not in TERMINAL and str(job["session_id"]) not in state.active_runs:
+                job["recovery_waiting"] = True
+                store.save_job(job)
+            logger.info("Recovery preserved live session %s", job["session_id"])
+            continue
         if job["status"] in {"running", "queued", "finalizing", "task_paused", "resuming"}:
             job.update(
                 status=job.get("terminal_status", "interrupted"),
@@ -50,6 +89,9 @@ def recover_batches():
                 or ("" if job.get("terminal_status") else "服务中断；未完成任务需手动重跑"),
                 ended_at=job.get("ended_at") or time.time(),
             )
+            if job["status"] == "completed":
+                resolve_recovery_error(job)
+            logger.info("Recovery resolved session %s to %s", job["session_id"], job["status"])
             store.save_job(job)
     for batch in store.records("batch"):
         if has_legacy_replay(batch["id"]):
@@ -62,6 +104,8 @@ def recover_batches():
                 store.put("batch", batch["id"], batch)
             continue
         if batch["status"] in {"running", "pausing", "paused"}:
+            if any(str(job["session_id"]) in live_sessions for job in store.jobs(batch["id"])):
+                continue
             batch.update(
                 status="paused", reason="服务已重启；请检查中断任务，点击继续执行未开始任务"
             )
@@ -86,6 +130,9 @@ async def on_terminal(item: dict, status: str):
         raise RuntimeError("任务元数据写入失败")
     if job.get("timed_out"):
         status = "failed"
+    job.pop("recovery_waiting", None)
+    if status == "completed":
+        resolve_recovery_error(job)
     job.update(status="finalizing", terminal_status=status, ended_at=time.time())
     store.save_job(job)
     try:
@@ -115,11 +162,37 @@ async def on_terminal(item: dict, status: str):
     store.save_job(job)
 
 
+async def reconcile_recovered_jobs():
+    """Finalize surviving workers once they exit; never resubmit their goals."""
+    from apps.admin_console.database.repositories.session_repository import SessionRepository
+    from artemis.runtime import DeviceExecutionLock
+    from artemis.runtime.task_result import read_worker_outcome
+
+    waiting = [job for job in store.jobs() if job.get("recovery_waiting")]
+    if not waiting:
+        return
+    repository = SessionRepository(store.path)
+    owners = {str(owner.session_id) for owner in DeviceExecutionLock.get_active_owners().values()}
+    for job in waiting:
+        sid = str(job["session_id"])
+        row = repository.get_session_by_id(sid) or {}
+        if sid in owners or repository.process_is_alive(row.get("pid")):
+            continue
+        outcome = read_worker_outcome(sid)
+        status = job.get("terminal_status") or outcome.get("task_status") or row.get("status")
+        if status not in TERMINAL:
+            status = "interrupted"
+            job["error"] = job.get("error") or "服务中断；未完成任务需手动重跑"
+            store.save_job(job)
+        await on_terminal({"session_id": sid}, status)
+
+
 async def tick_batches():
     from apps.admin_console.server import task_queue_service
     from artemis.runtime import device_pool
     from apps.admin_console.services.report_export import export_batch
 
+    await reconcile_recovered_jobs()
     for batch in store.records("batch"):
         if batch["status"] not in {"running", "paused"}:
             continue
@@ -135,10 +208,11 @@ async def tick_batches():
                 # not consume this case's execution budget.
                 if job["status"] != "running":
                     continue
-                limit = int(
+                from artemis.runtime.task_result import read_worker_outcome
-                    (job.get("case", {}).get("task_config") or {}).get("timeout_seconds")
+
-                    or batch.get("timeout_seconds", 1800)
+                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 +225,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 +339,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")
+        "timeout_seconds": payload.get("timeout_seconds") or settings.ARTEMIS_TASK_TIMEOUT_SECONDS,
-        or int(os.getenv("ARTEMIS_BATCH_TASK_TIMEOUT", "1800")),
         "stop_on_failure": payload.get("stop_on_failure", False),
         "request_id": payload.get("request_id"),
     }

apps/admin_console/services/automation_store.py

@@ -12,6 +12,12 @@ import uuid
 from artemis.config import DB_PATH
 
 
+RECOVERY_ERRORS = {
+    "服务中断；未完成任务需手动重跑",
+    "服务中断，需手动重跑",
+}
+
+
 def identifier(value: str) -> str:
     if not re.fullmatch(r"[A-Za-z0-9_-]{1,100}", value):
         raise ValueError("无效的记录编号")

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():

apps/admin_console/services/report_export.py

@@ -88,6 +88,10 @@ def failure_reason(directory: Path, status: str, recorded: str | None) -> str:
         )
         return safe[:500]
 
+    from apps.admin_console.services.automation_store import RECOVERY_ERRORS
+
+    if status == "completed" and recorded in RECOVERY_ERRORS:
+        return ""
     if recorded:
         return redact(recorded)
     if status not in {"failed", "interrupted"}:
@@ -95,6 +99,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

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):

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
         )

apps/showcase_ui/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "artemis-showcase-ui",
-  "version": "1.3.0",
+  "version": "1.3.1",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "artemis-showcase-ui",
-      "version": "1.3.0",
+      "version": "1.3.1",
       "dependencies": {
         "@angular/cdk": "^22.1.5",
         "@angular/common": "^22.1.4",

apps/showcase_ui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "artemis-showcase-ui",
-  "version": "1.3.0",
+  "version": "1.3.1",
   "scripts": {
     "ng": "ng",
     "start": "ng serve --proxy-config proxy.conf.json",

apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.html

@@ -338,6 +338,8 @@ limitations under the License.
                         playsinline
                         (loadedmetadata)="onLoadedMetadata()"
                         (timeupdate)="onTimeUpdate()"
+                        (seeked)="onSeeked()"
+                        (emptied)="onPause()"
                         (play)="onPlay()"
                         (pause)="onPause()"
                         (ended)="onEnded()"

apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.spec.ts

@@ -1,17 +1,26 @@
-import { TestBed } from '@angular/core/testing';
+import { ComponentFixture, TestBed } from '@angular/core/testing';
 import { signal } from '@angular/core';
 import { AgentService } from '../../services/agent.service';
 import { StepReplayFrame } from '../../core/models/stream.model';
 import { FloatingVideoPlayerComponent } from './floating-video-player.component';
 
 describe('FloatingVideoPlayerComponent video markers', () => {
-  it('shows a recorded step on the video timeline and seeks when clicked', () => {
+  let fixture: ComponentFixture<FloatingVideoPlayerComponent>;
+  let component: FloatingVideoPlayerComponent;
+  let video: HTMLVideoElement;
+  let paused: boolean;
+  let seeking: boolean;
+  let time: number;
+  let play: jasmine.Spy;
+  let agentService: ReturnType<typeof createAgentService>;
+
+  function createAgentService() {
     const start = 1_800_000_000;
     const frame: StepReplayFrame = {
       index: 0, stepNumber: 1, title: '步骤 1: 点击设置', imageUrl: 'step.png',
       actionText: '点击设置', timestamp: start + 5
     };
-    const agentService = {
+    return {
       isVideoWindowOpen: signal(true),
       isVideoMinimized: signal(false),
       recordingPlaybackStatus: signal('ready'),
@@ -25,15 +34,57 @@ describe('FloatingVideoPlayerComponent video markers', () => {
       currentSession: signal({ session_id: 'test', start_time: start }),
       currentSessionId: signal('test'),
       videoSeekRequest: signal(null),
-      stepSeekRequest: signal(null)
+      stepSeekRequest: signal(null),
+      consumeVideoAutoplay: jasmine.createSpy('consumeVideoAutoplay').and.returnValue(false)
     };
+  }
+
+  beforeEach(() => {
+    agentService = createAgentService();
     TestBed.configureTestingModule({
       imports: [FloatingVideoPlayerComponent],
       providers: [{ provide: AgentService, useValue: agentService }]
     });
 
-    const fixture = TestBed.createComponent(FloatingVideoPlayerComponent);
+    fixture = TestBed.createComponent(FloatingVideoPlayerComponent);
     fixture.detectChanges();
+    component = fixture.componentInstance;
+    video = component.videoRef!.nativeElement;
+    paused = true;
+    seeking = false;
+    time = 0;
+    Object.defineProperties(video, {
+      paused: { configurable: true, get: () => paused },
+      seeking: { configurable: true, get: () => seeking },
+      ended: { configurable: true, get: () => false },
+      duration: { configurable: true, get: () => 20 },
+      currentTime: {
+        configurable: true, get: () => time,
+        set: (value: number) => { time = value; seeking = true; }
+      }
+    });
+    spyOn(video, 'pause').and.callFake(() => {
+      paused = true;
+      video.dispatchEvent(new Event('pause'));
+    });
+    play = spyOn(video, 'play').and.callFake(() => {
+      paused = false;
+      video.dispatchEvent(new Event('play'));
+      return Promise.resolve();
+    });
+  });
+
+  function finishSeek(): void {
+    seeking = false;
+    video.dispatchEvent(new Event('seeked'));
+    fixture.detectChanges();
+  }
+
+  function playButton(): HTMLButtonElement {
+    return fixture.nativeElement.querySelector('.play-btn');
+  }
+
+  it('shows a recorded step on the video timeline and seeks when clicked', () => {
     const marker = fixture.nativeElement.querySelector('.video-marker') as HTMLButtonElement;
     expect(marker).not.toBeNull();
     expect(marker.getAttribute('aria-label')).toContain('步骤 1');
@@ -43,4 +94,150 @@ describe('FloatingVideoPlayerComponent video markers', () => {
     marker.click();
     expect(seek).toHaveBeenCalledOnceWith(5);
   });
+
+  it('resumes a playing video after a marker seek finishes and synchronizes the button', async () => {
+    component.togglePlay();
+    await Promise.resolve();
+    play.calls.reset();
+    fixture.nativeElement.querySelector('.video-marker').click();
+    expect(time).toBe(5);
+    expect(play).not.toHaveBeenCalled();
+    finishSeek();
+    await Promise.resolve();
+    fixture.detectChanges();
+    expect(paused).toBeFalse();
+    expect(play).toHaveBeenCalledTimes(1);
+    expect(playButton().title).toBe('暂停');
+  });
+
+  it('keeps a paused video paused after a marker seek and resumes with one click', async () => {
+    fixture.nativeElement.querySelector('.video-marker').click();
+    finishSeek();
+    expect(paused).toBeTrue();
+    expect(play).not.toHaveBeenCalled();
+    expect(playButton().title).toBe('播放');
+    playButton().click();
+    await Promise.resolve();
+    fixture.detectChanges();
+    expect(paused).toBeFalse();
+    expect(playButton().title).toBe('暂停');
+  });
+
+  it('preserves playing intent across rapid consecutive marker seeks', () => {
+    component.togglePlay();
+    component.seek(5);
+    component.seek(8);
+    play.calls.reset();
+    finishSeek();
+    expect(time).toBe(8);
+    expect(play).toHaveBeenCalledTimes(1);
+    expect(paused).toBeFalse();
+  });
+
+  it('honors play and pause clicks while the destination is still seeking', () => {
+    component.togglePlay();
+    component.seek(5);
+    play.calls.reset();
+    fixture.detectChanges();
+    expect(playButton().title).toBe('播放');
+    playButton().click();
+    fixture.detectChanges();
+    expect(playButton().title).toBe('暂停');
+    expect(play).not.toHaveBeenCalled();
+    playButton().click();
+    finishSeek();
+    expect(paused).toBeTrue();
+    expect(play).not.toHaveBeenCalled();
+    expect(playButton().title).toBe('播放');
+  });
+
+  it('waits for the destination segment to seek before resuming playback', () => {
+    agentService.activeVideoSegments.set([
+      { url: '/test.mp4', start: 0, duration: 20, width: 400, height: 800 },
+      { url: '/second.mp4', start: 20, duration: 20, width: 400, height: 800 }
+    ]);
+    fixture.detectChanges();
+    component.togglePlay();
+    component.seek(25);
+    play.calls.reset();
+    fixture.detectChanges();
+    video.dispatchEvent(new Event('loadedmetadata'));
+    expect(component.activeSegmentIndex()).toBe(1);
+    expect(time).toBe(5);
+    expect(play).not.toHaveBeenCalled();
+    finishSeek();
+    expect(play).toHaveBeenCalledTimes(1);
+    expect(paused).toBeFalse();
+    expect(component.currentTime()).toBe(25);
+  });
+
+  it('keeps a paused video paused when jumping to another segment', () => {
+    agentService.activeVideoSegments.set([
+      { url: '/test.mp4', start: 0, duration: 20, width: 400, height: 800 },
+      { url: '/second.mp4', start: 20, duration: 20, width: 400, height: 800 }
+    ]);
+    fixture.detectChanges();
+    component.seek(25);
+    fixture.detectChanges();
+    video.dispatchEvent(new Event('loadedmetadata'));
+    finishSeek();
+    expect(time).toBe(5);
+    expect(paused).toBeTrue();
+    expect(play).not.toHaveBeenCalled();
+    expect(playButton().title).toBe('播放');
+  });
+
+  it('resumes after seeking a recording without a segment manifest', () => {
+    agentService.activeVideoSegments.set([]);
+    fixture.detectChanges();
+    video.dispatchEvent(new Event('loadedmetadata'));
+    component.togglePlay();
+    component.seek(5);
+    play.calls.reset();
+    finishSeek();
+    expect(time).toBe(5);
+    expect(component.currentTime()).toBe(5);
+    expect(paused).toBeFalse();
+    expect(play).toHaveBeenCalledTimes(1);
+  });
+
+  it('resets a rejected play request so a single subsequent click can resume', async () => {
+    play.and.callFake(() => {
+      paused = false;
+      video.dispatchEvent(new Event('play'));
+      return Promise.reject(new DOMException('Interrupted', 'AbortError'));
+    });
+    component.seek(5, true);
+    finishSeek();
+    await Promise.resolve();
+    await Promise.resolve();
+    fixture.detectChanges();
+    expect(paused).toBeTrue();
+    expect(playButton().title).toBe('播放');
+    play.and.callFake(() => {
+      paused = false;
+      video.dispatchEvent(new Event('play'));
+      return Promise.resolve();
+    });
+    playButton().click();
+    await Promise.resolve();
+    fixture.detectChanges();
+    expect(paused).toBeFalse();
+    expect(playButton().title).toBe('暂停');
+  });
+
+  it('ignores an older aborted play request after a newer seek resumes', async () => {
+    let rejectOldPlay!: (reason: unknown) => void;
+    play.and.returnValue(new Promise<void>((_, reject) => { rejectOldPlay = reject; }));
+    component.togglePlay();
+    paused = false;
+    component.seek(5);
+    play.and.callFake(() => { paused = false; return Promise.resolve(); });
+    finishSeek();
+    rejectOldPlay(new DOMException('Interrupted', 'AbortError'));
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(paused).toBeFalse();
+    expect(component.isPlaying()).toBeTrue();
+  });
 });

apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.ts

@@ -71,6 +71,8 @@ export class FloatingVideoPlayerComponent implements OnDestroy {
   private pendingLocalTime: number | null = null;
   private pendingAutoplay = false;
   private pendingAbsoluteSeek: number | null = null;
+  private seekPlaybackIntent: boolean | null = null;
+  private playRequestId = 0;
 
   // Step Replay state signals
   public activeStepIndex = signal<number>(0);
@@ -138,6 +140,10 @@ export class FloatingVideoPlayerComponent implements OnDestroy {
         this.currentTime.set(0);
         this.duration.set(segments.reduce((sum, segment) => sum + segment.duration, 0));
         this.isPlaying.set(false);
+        this.pendingLocalTime = null;
+        this.pendingAutoplay = false;
+        this.seekPlaybackIntent = null;
+        this.playRequestId++;
       },
       { allowSignalWrites: true }
     );
@@ -315,20 +321,19 @@ export class FloatingVideoPlayerComponent implements OnDestroy {
         // that segment's metadata event apply pendingLocalTime.
         if (this.pendingLocalTime !== null) return;
       }
-      if (this.pendingLocalTime !== null) {
+      const autoplay = this.agentService.consumeVideoAutoplay();
-        v.currentTime = Math.max(0, Math.min(v.duration || this.pendingLocalTime, this.pendingLocalTime));
+      if (autoplay) {
-        this.pendingLocalTime = null;
-      }
-      if (this.pendingAutoplay) {
-        this.pendingAutoplay = false;
-        v.play().catch(() => {});
-      }
-      if (this.agentService.consumeVideoAutoplay()) {
         this.isMuted.set(true);
         v.muted = true;
-        v.play().catch(() => {
+      }
-          this.isPlaying.set(false);
+      const shouldPlay = this.pendingAutoplay || autoplay;
-        });
+      this.pendingAutoplay = false;
+      if (this.pendingLocalTime !== null) {
+        const localTime = Math.max(0, Math.min(v.duration || this.pendingLocalTime, this.pendingLocalTime));
+        this.pendingLocalTime = null;
+        this.seekWithinVideo(localTime, shouldPlay);
+      } else if (shouldPlay) {
+        this.playVideo();
       }
     }
   }
@@ -346,14 +351,28 @@ export class FloatingVideoPlayerComponent implements OnDestroy {
   }
 
   public onPlay(): void {
-    this.isPlaying.set(true);
+    this.isPlaying.set(!this.videoRef?.nativeElement.paused);
   }
 
   public onPause(): void {
     this.isPlaying.set(false);
   }
 
+  public onSeeked(): void {
+    const v = this.videoRef?.nativeElement;
+    if (!v || v.seeking || this.pendingLocalTime !== null) return;
+    const shouldPlay = this.seekPlaybackIntent;
+    this.seekPlaybackIntent = null;
+    this.onTimeUpdate();
+    if (shouldPlay) {
+      this.playVideo();
+    } else {
+      this.isPlaying.set(!v.paused && !v.ended);
+    }
+  }
+
   public onEnded(): void {
+    this.isPlaying.set(false);
     const segments = this.agentService.activeVideoSegments();
     if (segments.length && this.activeSegmentIndex() < segments.length - 1) {
       this.pendingLocalTime = 0;
@@ -369,6 +388,10 @@ export class FloatingVideoPlayerComponent implements OnDestroy {
   }
 
   public onError(): void {
+    this.playRequestId++;
+    this.pendingLocalTime = null;
+    this.pendingAutoplay = false;
+    this.seekPlaybackIntent = null;
     this.videoLoadError.set(true);
     this.isPlaying.set(false);
   }
@@ -379,11 +402,52 @@ export class FloatingVideoPlayerComponent implements OnDestroy {
   public togglePlay(): void {
     const v = this.videoRef?.nativeElement;
     if (!v) return;
-    if (v.paused) {
+    if (this.pendingLocalTime !== null) {
-      v.play().catch(() => {});
+      this.pendingAutoplay = !this.isPlaying();
-    } else {
+      this.isPlaying.set(this.pendingAutoplay);
-      v.pause();
+      return;
     }
+    if (this.seekPlaybackIntent !== null) {
+      this.seekPlaybackIntent = !this.isPlaying();
+      this.isPlaying.set(this.seekPlaybackIntent);
+      return;
+    }
+    if (v.paused) {
+      this.playVideo();
+    } else {
+      this.playRequestId++;
+      v.pause();
+      this.isPlaying.set(false);
+    }
+  }
+
+  private playVideo(): void {
+    const v = this.videoRef?.nativeElement;
+    if (!v) return;
+    const requestId = ++this.playRequestId;
+    v.play().then(() => {
+      if (requestId === this.playRequestId) this.isPlaying.set(!v.paused && !v.ended);
+    }).catch(() => {
+      // A rejected play request can leave paused=false without advancing frames.
+      // Reset the native state too, so the next click really starts playback.
+      if (requestId === this.playRequestId) {
+        v.pause();
+        this.isPlaying.set(false);
+      }
+    });
+  }
+
+  private seekWithinVideo(seconds: number, shouldPlay: boolean): void {
+    const v = this.videoRef?.nativeElement;
+    if (!v) return;
+    this.playRequestId++;
+    this.seekPlaybackIntent = shouldPlay;
+    // Cancel any unfinished play request before seeking; resume only once the
+    // browser has decoded the destination, including after a source switch.
+    v.pause();
+    this.isPlaying.set(false);
+    v.currentTime = seconds;
+    if (!v.seeking) this.onSeeked();
   }
 
   public restart(): void {
@@ -421,19 +485,19 @@ export class FloatingVideoPlayerComponent implements OnDestroy {
     const target = Math.max(0, Math.min(this.duration(), seconds));
     const segments = this.agentService.activeVideoSegments();
     const location = locateTimelineTime(segments, target);
+    const shouldPlay = autoplay || this.pendingAutoplay || (this.seekPlaybackIntent ?? !v.paused);
     if (!location) {
-      v.currentTime = target;
+      this.seekWithinVideo(target, shouldPlay);
-      if (autoplay) v.play().catch(() => {});
+    } else if (location.index === this.activeSegmentIndex() && this.pendingLocalTime === null) {
-      return;
+      this.seekWithinVideo(location.localTime, shouldPlay);
-    }
-    const { index, localTime } = location;
-    if (index === this.activeSegmentIndex()) {
-      v.currentTime = localTime;
-      if (autoplay) v.play().catch(() => {});
     } else {
-      this.pendingLocalTime = localTime;
+      this.playRequestId++;
-      this.pendingAutoplay = autoplay || !v.paused;
+      this.seekPlaybackIntent = null;
-      this.activeSegmentIndex.set(index);
+      v.pause();
+      this.isPlaying.set(false);
+      this.pendingLocalTime = location.localTime;
+      this.pendingAutoplay = shouldPlay;
+      this.activeSegmentIndex.set(location.index);
     }
     this.currentTime.set(target);
   }
@@ -641,6 +705,7 @@ export class FloatingVideoPlayerComponent implements OnDestroy {
   }
 
   public ngOnDestroy(): void {
+    this.playRequestId++;
     if (this.stepTimer) {
       clearTimeout(this.stepTimer);
       this.stepTimer = null;

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 {
         <span>步骤摘要压缩<small>Flash 模式下在后台生成摘要，减少长任务的上下文占用。</small></span></label>
       <label><input type="checkbox" role="switch" [(ngModel)]="config.screenshot_limit_enabled" [disabled]="busy() || loading() || !loaded()" />
         <span>限制截图分辨率<small>开启时长边最多 {{ config.screenshot_max_long_edge }} 像素；关闭后使用设备原始分辨率。JSON 配置可覆盖开关和上限。</small></span></label>
-      <button type="button" (click)="save()" [disabled]="busy() || loading() || !loaded()">{{ busy() ? '正在保存…' : '保存功能设置' }}</button>
+      <details class="execution-settings">
+        <summary>执行保护</summary>
+        <p>限制故障等待与反复失败；保存后对新任务生效，任务 JSON 可单独覆盖。</p>
+        <div class="number-fields">
+          <label>任务总超时（秒）<input type="number" min="30" max="86400" step="1" [(ngModel)]="config.task_timeout_seconds" [disabled]="busy() || !loaded()" /><small>不包含排队和用户暂停时间。</small></label>
+          <label>屏幕观测超时（秒）<input type="number" min="1" max="300" [(ngModel)]="config.observation_timeout_seconds" [disabled]="busy() || !loaded()" /></label>
+          <label>观测恢复最多尝试次数<input type="number" min="1" max="5" step="1" [(ngModel)]="config.observation_max_attempts" [disabled]="busy() || !loaded()" /></label>
+          <label>ADB 命令超时（秒）<input type="number" min="1" max="60" [(ngModel)]="config.device_command_timeout_seconds" [disabled]="busy() || !loaded()" /></label>
+          <label>连续动作失败次数<input type="number" min="1" max="10" step="1" [(ngModel)]="config.action_failure_attempts" [disabled]="busy() || !loaded()" /><small>先重新观察，再让兜底模型处理当前步骤。</small></label>
+          <label>任务收尾超时（秒）<input type="number" min="1" max="300" [(ngModel)]="config.worker_cleanup_timeout_seconds" [disabled]="busy() || !loaded()" /><small>已有最终结果但进程未退出时，回收进程以释放队列。</small></label>
+        </div>
+      </details>
+      <button type="button" (click)="save()" [disabled]="busy() || loading() || !loaded() || !valid()">{{ busy() ? '正在保存…' : '保存功能设置' }}</button>
       @if (message()) { <p role="status">{{ message() }}</p> }
       @if (error()) { <p role="alert" class="error">{{ error() }}</p> }
     </section>
@@ -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<Capabilities>();
   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<Capabilities>('/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<Capabilities>('/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;
+  }
 }

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<HTMLDetailsElement>('.execution-settings')!.open).toBeFalse();
+    root.querySelector<HTMLElement>('summary')!.click();
+    const timeout = root.querySelector<HTMLInputElement>('input[type="number"]')!;
+    timeout.value = '3600';
+    timeout.dispatchEvent(new Event('input'));
+    fixture.detectChanges();
+    await fixture.whenStable();
+    root.querySelector<HTMLButtonElement>('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<HTMLButtonElement>('button')!.disabled).toBeTrue();
+    fixture.componentInstance.save();
+    http.expectNone('/api/system/capability-settings');
+  });
 });

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(
-            blocks.append({"type": "text", "text": f"{PRO_UI_LIST_MARKER}\n{xml_list}"})
+            {
+                "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:
+        if state is not None:
-            obs, screenshot_bytes = await observe(self.ctx, self.controller, settle_ms=0)
+            state.indexed_points = []
-        except Exception as shot_err:
+            state.indexed_elements = []
-            logger.warning(f"Failed to capture fallback screenshot in FlashRunner: {shot_err}")
+            state.latest_screenshot = None
-            return post_img_bytes
+        obs, screenshot_bytes = await require_observation(self.ctx, self.controller, settle_ms=0)
-        if not obs.ok:
-            logger.warning(f"Failed to capture fallback screenshot in FlashRunner: {obs.message}")
-            return post_img_bytes
         post_img_bytes = screenshot_bytes
-        if not exec_result.ui_elements_text:
+        exec_result.ui_elements_text = obs.elements_text
-            exec_result.ui_elements_text = obs.elements_text
+        exec_result.screenshot_path = obs.screenshot_path
-        if not exec_result.screenshot_path and obs.screenshot_path:
-            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:

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)
+    from artemis.mcp.observation import require_observation
-    if not obs.ok:
+
-        return None, None, None
+    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]

artemis/config/settings.py

@@ -108,6 +108,7 @@ class Settings(BaseSettings):
     OPENAI_REASONING_EFFORT: str = "high"
     ARTEMIS_CREDENTIAL_TIMEOUT_SECONDS: float = Field(default=12.0, gt=0)
     ARTEMIS_MODEL_CACHE_TTL_SECONDS: float = Field(default=10.0, ge=0)
+    ARTEMIS_DAEMON_PROBE_TIMEOUT_SECONDS: float = Field(default=2.0, gt=0)
     ARTEMIS_VIDEO_TIMEOUT_SECONDS: float = Field(default=120.0, gt=0)
     ARTEMIS_VIDEO_MAX_RETRIES: int = Field(default=1, ge=0)
     ARTEMIS_VIDEO_CHUNK_SECONDS: float = Field(default=60.0, gt=0)
@@ -122,6 +123,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)

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
+            raise RuntimeError(
-            screenshot_base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
+                "设备截图失败：ADB 与屏幕辅助服务均未返回有效截图；" + "; ".join(capture_errors)
-            screenshot_bytes = base64.b64decode(screenshot_base64)
+            )
+
+        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 = (
+        clean_b64 = screenshot_base64
-            screenshot_base64
+        clean_bytes = screenshot_bytes
-            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)
-        )
         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:

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(

artemis/runtime/daemon_client.py

@@ -28,6 +28,9 @@ import urllib.error
 import urllib.request
 
 from artemis.config.paths import ROOT_DIR
+from artemis.config.settings import settings
+from artemis.runtime.process_probe import pid_is_alive
+from artemis.runtime.server_lifecycle import is_port_in_use, read_server_info
 from artemis.utils.logger import get_logger
 
 logger = get_logger(__name__)
@@ -44,10 +47,11 @@ def is_standalone_forced() -> bool:
 def is_daemon_running(
     host: str = DEFAULT_DAEMON_HOST,
     port: int = DEFAULT_DAEMON_PORT,
-    timeout: float = 0.3,
+    timeout: float | None = None,
 ) -> bool:
     """Fast probe to check if the Artemis Daemon HTTP server is responding."""
-    url = f"http://{host}:{port}/api/system/readiness"
+    timeout = timeout if timeout is not None else settings.ARTEMIS_DAEMON_PROBE_TIMEOUT_SECONDS
+    url = f"http://{host}:{port}/api/system/server-status"
     try:
         req = urllib.request.Request(url, headers={"User-Agent": "Artemis-Daemon-Probe"})
         with urllib.request.urlopen(req, timeout=timeout) as response:
@@ -190,6 +194,15 @@ def ensure_daemon_running(
     if is_daemon_running(host, port):
         return True, f"http://{host}:{port}"
 
+    if is_port_in_use(port, host):
+        # A slow/busy HTTP loop is not a stopped daemon. Only trust persisted
+        # metadata belonging to a live process on the requested endpoint.
+        info = read_server_info() or {}
+        if info.get("port") == port and info.get("host") == host and pid_is_alive(info.get("pid")):
+            return True, f"http://{host}:{port}"
+        logger.warning(f"Port {host}:{port} is occupied; refusing to spawn another daemon")
+        return False, None
+
     # Auto-spawn Daemon in background
     spawn_daemon(host, port)
 
@@ -221,6 +234,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 +260,7 @@ def submit_task_to_daemon(
         "session_id": session_id,
         "ingress": ingress,
         "conversation_id": conversation_id,
+        "runtime_overrides": runtime_overrides,
     }
 
     try:

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)

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)

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")
 

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 同名字段可覆盖。原有结果与截图保留，未完成保存的录屏尽力恢复。

docs/automation-workspace.md

@@ -20,9 +20,9 @@
 4. 上传文件，查看逐行校验结果。所有错误修正后才能提交。
 5. 明确选择设备、执行模式和超时时间，然后点击“确认导入并开始执行”。
 
-同一批次按顺序派发。默认失败后继续且不自动重试，也可以选择失败即停止。设备不可用、模型暂停或任务超时会暂停后续派发。暂停批次只停止后续派发。单个任务可在安全边界暂停并释放设备；批次下一项可继续，暂停的任务待提交指导或点击继续后在设备队列末尾恢复。运行中提交指导会中断当前模型思考或只读诊断，重新观察画面；已经开始的设备动作会安全结束。网页流程会显示指导的提交与领取状态，仍应检查后续动作是否符合指导。暂停等待和重新排队时间不计入该任务执行时限。
+同一批次按顺序派发。默认失败后继续且不自动重试，也可以选择失败即停止。当前任务因设备异常或超时失败后，默认继续后续用例；设置失败即停止时才停止后续派发。暂停批次只停止后续派发。单个任务可在安全边界暂停并释放设备；批次下一项可继续，暂停的任务待提交指导或点击继续后在设备队列末尾恢复。运行中提交指导会中断当前模型思考或只读诊断，重新观察画面；已经开始的设备动作会安全结束。网页流程会显示指导的提交与领取状态，仍应检查后续动作是否符合指导。暂停等待和重新排队时间不计入该任务执行时限。
 
-服务重启后保留批次，等待用户继续未开始的用例。正在执行或暂停的任务标记中断，不自动重做；“复制失败项为新批次”保留原记录并产生新任务编号。
+服务启动前先取得监听端口，重复启动不会修改现有任务或服务信息。恢复时核对原任务进程和设备锁：仍存活的任务继续等待其真实结果，退出后补齐报告；已经退出且没有终态的任务标记中断，不自动重做，批次等待用户继续未开始的用例。“复制失败项为新批次”保留原记录并产生新任务编号。任务正常完成后，已失效的“服务中断”提示归档为恢复事件，不再显示为当前失败原因；启动时也会纠正已有完成记录中的此类旧提示。
 
 每个任务结束后生成所选格式报告。批次 ZIP 包含汇总、每项任务的报告和已保存的有序步骤截图；已开启录屏且文件存在时，还会按任务收录录屏片段。失败、跳过和中断任务也会列入。暂停时下载的是当前快照。录屏与 AI 视频分析分别配置，录屏默认关闭；仅开启分析时，录制内容用于本次分析，结束后不保留下载文件。暂停期间不录制其他任务的设备画面。
 
@@ -34,6 +34,12 @@
 
 ## 环境配置
 
-`.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 均受总执行时限保护，不计排队和用户暂停时间；即使执行进程卡住，也由父进程终止并保存失败原因。
+
+守护服务通过轻量状态接口探测，`ARTEMIS_DAEMON_PROBE_TIMEOUT_SECONDS` 默认 2 秒。短暂超时且端口仍被存活的原服务占用时复用原服务，不重复启动；未知进程占用端口时明确拒绝覆盖。
+
+网页“功能配置 → 执行保护”及 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 同名字段可覆盖。原有结果与截图保留，未完成保存的录屏尽力恢复。

docs/releases/1.3.1.md

@@ -0,0 +1,21 @@
+# ARTEMIS-Eapil 1.3.1
+
+## 修复
+
+- 限制任务执行、设备观察、连续动作失败和资源清理的等待与重试。超时或失败后释放执行资源，让队列继续处理后续任务，避免单个任务长时间反复重试。
+- 修复点击录像时间轴关键步骤后，视频暂停但播放按钮状态未同步的问题。播放中跳转后继续播放，暂停时跳转保持暂停，恢复播放只需点击一次。
+- 修复服务探测超时后重复启动后台服务，导致仍在执行的任务被误标为“服务中断”的问题。启动时先占用监听端口，再初始化服务和恢复任务；已有服务占用端口时安全退出。
+- 恢复任务时保留仍存活的执行进程，并跟踪其最终结果。完成任务会清除已失效的“服务中断”提示并保留恢复记录，真实执行失败和超时原因继续显示。
+
+## 安装与升级
+
+下载附件 `artemis-eapil-1.3.1-source.zip`，解压后运行 `start.bat`；国内网络使用 `start-cn.bat`。源码包包含可继续更新的 Git 仓库，不包含虚拟环境、构建产物、密钥和执行记录。
+
+已有安装可通过启动脚本更新。升级后重启服务、刷新网页，并重新加载 MCP，让各入口使用新版实现。
+
+## 验证
+
+- Python 默认测试：2246 项通过，6 项跳过，8 项未选中；最终恢复逻辑回归测试：65 项通过。
+- 前端测试：169 项通过；前端构建、Ruff、质量基线、核心类型检查及 Git 差异检查通过。
+- 使用独立进程验证重复启动不会修改现有任务或服务元数据；在浏览器验证录像关键步骤跳转及播放状态。
+- 本次验证未运行真实 Android 任务；跨录像片段跳转通过单元测试验证。

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,
+    capabilities: video_enabled and recording_enabled are independent; outputter_enabled,
-    step_summarizer_enabled. reports: formats=['docx','xlsx'] (at least one).
+    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.

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")

pyproject.toml

@@ -14,7 +14,7 @@
 
 [project]
 name = "artemis"
-version = "1.3.0"
+version = "1.3.1"
 description = "Autonomous AI Assistant for Mobile Automation"
 
 authors = [

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 同名字段可覆盖。原有结果与截图保留，未完成保存的录屏尽力恢复。

tests/unit/admin_console/test_automation.py

@@ -300,6 +300,116 @@ def test_restart_never_repeats_started_work(isolated):
     assert isolated.jobs(retry["id"])[0]["session_id"] != job["session_id"]
 
 
+@pytest.mark.parametrize("liveness_source", ["pid", "device_lock"])
+def test_recovery_preserves_live_runner_and_batch(isolated, monkeypatch, liveness_source):
+    import os
+    from types import SimpleNamespace
+    from artemis.runtime import DeviceExecutionLock
+
+    batch = service.create_batch(payload())
+    service.batch_action(batch["id"], "resume")
+    job = isolated.jobs(batch["id"])[0]
+    isolated.save_job({**job, "status": "running"})
+    if liveness_source == "pid":
+        with sqlite3.connect(isolated.path) as conn:
+            conn.execute(
+                "INSERT INTO sessions(session_id,initial_goal,start_time,status,device_info,pid) "
+                "VALUES (?,?,?,?,?,?)",
+                (job["session_id"], "goal", 1, "running", "{}", os.getpid()),
+            )
+    monkeypatch.setattr(
+        DeviceExecutionLock,
+        "get_active_owners",
+        lambda: (
+            {"device": SimpleNamespace(session_id=job["session_id"])}
+            if liveness_source == "device_lock"
+            else {}
+        ),
+    )
+    service.recover_batches()
+    assert isolated.job(job["session_id"])["status"] == "running"
+    assert not isolated.job(job["session_id"]).get("error")
+    assert isolated.get("batch", batch["id"])["status"] == "running"
+
+
+@pytest.mark.asyncio
+async def test_natural_completion_resolves_old_recovery_error_in_job_report_and_history(
+    client, isolated
+):
+    job = isolated.jobs(service.create_batch(payload())["id"])[0]
+    message = "服务中断；未完成任务需手动重跑"
+    isolated.save_job({**job, "status": "interrupted", "error": message})
+    await service.on_terminal({"session_id": job["session_id"]}, "completed")
+    saved = isolated.job(job["session_id"])
+    assert saved["status"] == "completed"
+    assert not saved.get("error")
+    assert saved["recovery_events"][0]["message"] == message
+    assert reports.session_report(job["session_id"])["error"] == ""
+    assert not client.get("/api/automation/history").json()["items"][0].get("error")
+
+
+def test_recovery_repairs_completed_legacy_warning_but_preserves_real_failure(isolated):
+    batch = service.create_batch(payload())
+    first, second = isolated.jobs(batch["id"])
+    isolated.save_job({**first, "status": "completed", "error": "服务中断，需手动重跑"})
+    isolated.save_job({**second, "status": "failed", "error": "任务执行超时（30 秒）"})
+    assert reports.session_report(first["session_id"])["error"] == ""
+    service.recover_batches()
+    assert not isolated.job(first["session_id"]).get("error")
+    assert "服务中断" in isolated.job(first["session_id"])["recovery_events"][0]["message"]
+    assert "超时" in isolated.job(second["session_id"])["error"]
+
+
+@pytest.mark.asyncio
+async def test_surviving_worker_is_finalized_when_it_exits_without_resubmission(
+    isolated, monkeypatch
+):
+    import os
+    from artemis.runtime import DeviceExecutionLock
+
+    batch = service.create_batch(payload())
+    service.batch_action(batch["id"], "resume")
+    job = isolated.jobs(batch["id"])[0]
+    isolated.save_job({**job, "status": "running"})
+    monkeypatch.setattr(DeviceExecutionLock, "get_active_owners", lambda: {})
+    with sqlite3.connect(isolated.path) as conn:
+        conn.execute(
+            "INSERT INTO sessions(session_id,initial_goal,start_time,status,device_info,pid) "
+            "VALUES (?,?,?,?,?,?)",
+            (job["session_id"], "goal", time.time(), "running", "{}", os.getpid()),
+        )
+    service.recover_batches()
+    await service.reconcile_recovered_jobs()
+    assert isolated.job(job["session_id"])["status"] == "running"
+    with sqlite3.connect(isolated.path) as conn:
+        conn.execute(
+            "UPDATE sessions SET pid=NULL,status='completed' WHERE session_id=?",
+            (job["session_id"],),
+        )
+    await service.reconcile_recovered_jobs()
+    saved = isolated.job(job["session_id"])
+    assert saved["status"] == "completed"
+    assert saved["report_ready"]
+    assert not saved.get("recovery_waiting")
+    assert not saved.get("error")
+    assert len(isolated.jobs(batch["id"])) == 2
+    assert isolated.jobs(batch["id"])[1]["status"] == "pending"
+
+
+@pytest.mark.asyncio
+async def test_surviving_worker_that_dies_without_result_is_interrupted(isolated, monkeypatch):
+    from artemis.runtime import DeviceExecutionLock
+
+    job = isolated.jobs(service.create_batch(payload())["id"])[0]
+    isolated.save_job({**job, "status": "running", "recovery_waiting": True})
+    monkeypatch.setattr(DeviceExecutionLock, "get_active_owners", lambda: {})
+    await service.reconcile_recovered_jobs()
+    saved = isolated.job(job["session_id"])
+    assert saved["status"] == "interrupted"
+    assert "服务中断" in saved["error"]
+    assert not saved.get("recovery_waiting")
+
+
 @pytest.mark.asyncio
 async def test_batch_admission_idempotent_and_bound_device(isolated, monkeypatch):
     from apps.admin_console import server

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"

tests/unit/admin_console/test_server_lifecycle.py

@@ -15,6 +15,7 @@
 import asyncio
 from contextlib import suppress
 import signal
+import socket
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -23,10 +24,55 @@ import uvicorn
 from apps.admin_console.core.state import state
 from apps.admin_console.routers.tasks import stream_events
 from apps.admin_console.server import ArtemisUvicornServer, app, on_shutdown
+from apps.admin_console import server as server_module
 
 WINDOWS_FORCE_SIGNAL = getattr(signal, "SIGBREAK", signal.SIGTERM)
 
 
+@pytest.mark.parametrize("reload", [False, True])
+def test_occupied_port_never_initializes_or_overwrites_server_metadata(
+    tmp_path, monkeypatch, reload
+):
+    from artemis.runtime import server_lifecycle
+
+    metadata = tmp_path / "server.json"
+    metadata.write_text('{"pid":123,"lifecycle_token":"original"}', encoding="utf-8")
+    monkeypatch.setattr(server_lifecycle, "get_server_info_file", lambda: metadata)
+    before = metadata.read_bytes()
+    with (
+        socket.create_server(("127.0.0.1", 0)) as original,
+        patch.object(server_module.ArtemisUvicornServer, "run") as run,
+        patch.object(server_module, "write_server_info") as publish,
+    ):
+        with pytest.raises(SystemExit, match="1"):
+            server_module.run_ui_server("127.0.0.1", original.getsockname()[1], reload=reload)
+        run.assert_not_called()
+        publish.assert_not_called()
+        assert metadata.read_bytes() == before
+
+
+def test_server_holds_port_before_running_lifespan(monkeypatch):
+    monkeypatch.setattr(state, "port", 0, raising=False)
+    monkeypatch.setattr(state, "host", "127.0.0.1", raising=False)
+    observed = []
+
+    def run(_server, sockets):
+        listener = sockets[0]
+        port = listener.getsockname()[1]
+        observed.append(listener)
+        assert state.port == port
+        with pytest.raises(OSError), socket.create_server(("127.0.0.1", port)):
+            pass
+
+    monkeypatch.setattr(server_module.ArtemisUvicornServer, "run", run)
+    monkeypatch.setattr(server_module, "clear_server_info", lambda **kwargs: None)
+    with patch.object(server_module, "write_server_info") as publish:
+        server_module.run_ui_server("127.0.0.1", 0)
+        publish.assert_not_called()  # Publication belongs to the later lifespan.
+    assert len(observed) == 1
+    assert observed[0].fileno() == -1
+
+
 def _pending_queue_get_tasks():
     return {
         task

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):
+async def test_fallback_observation_replaces_stale_executor_ui_list(mock_context, tmp_path):
-    """An action that reported a UI list but no screenshot keeps its own list."""
+    """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 post == _POST_IMG
-    assert exec_result.ui_elements_text == "[1] Text: 'Mine' | Bounds: [0,0][10,10]"
+    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)
+    with patch("artemis.mcp.observation.settings.ARTEMIS_OBSERVATION_MAX_ATTEMPTS", 3):
-
+        with pytest.raises(DeviceObservationError, match="adb gone"):
-    assert post is None
+            await runner._capture_post_screenshot(exec_result, "click", frozenset({"click"}), state)
-    assert exec_result.ui_elements_text is None
+    assert runner.controller.get_screen_data.await_count == 3
-    assert state.latest_screenshot == "old.jpg"
+    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):

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

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

tests/unit/runtime/test_daemon_client.py

@@ -93,6 +93,7 @@ def test_ensure_daemon_running_standalone_override(monkeypatch):
 def test_ensure_daemon_running_auto_spawns():
     with (
         patch("artemis.runtime.daemon_client.is_daemon_running", side_effect=[False, True]),
+        patch("artemis.runtime.daemon_client.is_port_in_use", return_value=False),
         patch("artemis.runtime.daemon_client.spawn_daemon") as mock_spawn,
     ):
         ok, base_url = ensure_daemon_running(timeout=1.0, wait_ready=True)
@@ -101,6 +102,42 @@ def test_ensure_daemon_running_auto_spawns():
         mock_spawn.assert_called_once()
 
 
+def test_busy_live_daemon_is_reused_without_spawning():
+    with (
+        patch("artemis.runtime.daemon_client.is_daemon_running", return_value=False),
+        patch("artemis.runtime.daemon_client.is_port_in_use", return_value=True),
+        patch(
+            "artemis.runtime.daemon_client.read_server_info",
+            return_value={"host": "127.0.0.1", "port": 8000, "pid": 12345},
+        ),
+        patch("artemis.runtime.daemon_client.pid_is_alive", return_value=True),
+        patch("artemis.runtime.daemon_client.spawn_daemon") as spawn,
+    ):
+        assert ensure_daemon_running() == (True, "http://127.0.0.1:8000")
+        spawn.assert_not_called()
+
+
+def test_unknown_port_owner_is_not_overwritten():
+    with (
+        patch("artemis.runtime.daemon_client.is_daemon_running", return_value=False),
+        patch("artemis.runtime.daemon_client.is_port_in_use", return_value=True),
+        patch("artemis.runtime.daemon_client.read_server_info", return_value=None),
+        patch("artemis.runtime.daemon_client.spawn_daemon") as spawn,
+    ):
+        assert ensure_daemon_running() == (False, None)
+        spawn.assert_not_called()
+
+
+def test_liveness_probe_uses_lightweight_endpoint_and_configured_timeout(monkeypatch):
+    from artemis.config.settings import settings
+
+    monkeypatch.setattr(settings, "ARTEMIS_DAEMON_PROBE_TIMEOUT_SECONDS", 7.0)
+    with patch("urllib.request.urlopen", return_value=_daemon_ok_response(b"{}")) as probe:
+        assert is_daemon_running()
+        assert probe.call_args.args[0].full_url.endswith("/api/system/server-status")
+        assert probe.call_args.kwargs["timeout"] == 7.0
+
+
 def test_submit_task_to_daemon():
     mock_resp = MagicMock()
     mock_resp.status = 200

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="<hierarchy />",
         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",

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()

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 (

uv.lock

@@ -206,7 +206,7 @@ wheels = [
 
 [[package]]
 name = "artemis"
-version = "1.3.0"
+version = "1.3.1"
 source = { editable = "." }
 dependencies = [
     { name = "adbutils" },