fix: bound task failures and release stalled workers

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/services/automation_service.py

@@ -10,6 +10,8 @@ from uuid import UUID
 from apps.admin_console.core.state import state
 from apps.admin_console.services.automation_store import store, new_id
 from apps.admin_console.services.task_import import case_goal, case_title
+from artemis.config.settings import settings
+from artemis.runtime.task_limits import task_timeout
 
 logger = logging.getLogger(__name__)
 TERMINAL = {"completed", "failed", "cancelled", "skipped", "interrupted"}
@@ -135,10 +137,11 @@ async def tick_batches():
                 # not consume this case's execution budget.
                 if job["status"] != "running":
                     continue
-                limit = int(
-                    (job.get("case", {}).get("task_config") or {}).get("timeout_seconds")
-                    or batch.get("timeout_seconds", 1800)
-                )
+                from artemis.runtime.task_result import read_worker_outcome
+
+                if read_worker_outcome(str(job["session_id"])):
+                    continue  # Execution has ended; the worker supervisor bounds cleanup.
+                limit = task_timeout({}, job, batch)
                 task_state = getattr(state, "active_runs", {}).get(str(job["session_id"])) or {}
                 excluded = float(task_state.get("non_execution_seconds") or 0)
                 if task_state.get("non_execution_started_at"):
@@ -151,7 +154,17 @@ async def tick_batches():
                         error=f"任务执行超时（{limit} 秒，不含暂停/排队时间）", timed_out=True
                     )
                     store.save_job(job)
-                    task_queue_service.stop_tasks(session_id=job["session_id"])
+                    from artemis.runtime.task_result import save_failure_result
+
+                    try:
+                        await asyncio.to_thread(
+                            save_failure_result,
+                            str(job["session_id"]),
+                            job["error"],
+                            code="task_timeout",
+                        )
+                    finally:
+                        task_queue_service.stop_tasks(session_id=job["session_id"])
                     # on_terminal records the failed case. The next pending
                     # case then runs normally unless stop_on_failure is set.
             continue
@@ -255,8 +268,7 @@ def create_batch(payload: dict) -> dict:
         "formats": payload.get("formats") or store.preferences()["formats"],
         "profile": payload.get("profile", "flash"),
         "device_serial": payload.get("device_serial"),
-        "timeout_seconds": payload.get("timeout_seconds")
-        or int(os.getenv("ARTEMIS_BATCH_TASK_TIMEOUT", "1800")),
+        "timeout_seconds": payload.get("timeout_seconds") or settings.ARTEMIS_TASK_TIMEOUT_SECONDS,
         "stop_on_failure": payload.get("stop_on_failure", False),
         "request_id": payload.get("request_id"),
     }

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

@@ -95,6 +95,11 @@ def failure_reason(directory: Path, status: str, recorded: str | None) -> str:
     state = read_json(directory / "status.json", {})
     if isinstance(state, dict) and state.get("error"):
         return redact(state["error"])
+    result = read_json(directory / "result.json", {})
+    if isinstance(result, dict) and result.get("status") == "failed":
+        reason = result.get("error") or result.get("explanation")
+        if reason:
+            return redact(reason)
     log_lines: list[str] = []
     for name in ("stdout.log", "stderr.log"):
         path = directory / name

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
         )