From 84e183d28de0d759acbd1e61f78acec767739a4c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E9=93=AD=E8=BD=A9?= Date: Fri, 25 Sep 2026 02:02:48 +0800 Subject: [PATCH] feat: improve batch task controls and release 1.2.0 --- apps/admin_console/routers/automation.py | 20 ++- apps/admin_console/routers/sessions.py | 14 ++ .../services/automation_service.py | 45 ++++++- .../services/automation_store.py | 80 ++++++++++-- apps/admin_console/services/report_export.py | 31 ++++- apps/admin_console/services/task_import.py | 7 + .../services/task_queue_service.py | 3 + apps/showcase_ui/package-lock.json | 4 +- apps/showcase_ui/package.json | 2 +- .../agent-stream/agent-stream.component.html | 14 +- .../agent-stream/agent-stream.component.scss | 41 ++++-- .../agent-stream/agent-stream.component.ts | 28 ++-- .../automation-panel.component.html | 8 +- .../automation-panel.component.ts | 7 +- .../task-history.component.spec.ts | 15 +++ .../task-history/task-history.component.ts | 7 +- .../src/app/core/models/session.model.ts | 3 + .../pages/workspace/workspace.component.html | 2 +- .../src/app/services/agent.service.ts | 6 +- .../src/app/utils/ui-label.pipe.ts | 5 +- artemis/agents/flash/runner.py | 48 ++++++- artemis/agents/operator/operator.py | 14 +- artemis/data_engine/storage.py | 78 +++++------ artemis/runtime/task_control.py | 11 +- docs/ai-automation.md | 6 +- docs/automation-workspace.md | 6 +- docs/releases/1.2.0.md | 14 ++ mcp_server/tools/automation.py | 31 ++++- mcp_server/tools/task_manager.py | 8 +- pyproject.toml | 2 +- skills/artemis-automation/SKILL.md | 4 +- .../artemis-automation/references/batches.md | 2 +- tests/unit/admin_console/test_automation.py | 123 +++++++++++++++++- .../test_session_list_batching.py | 47 +++++++ tests/unit/mcp/test_automation_tools.py | 10 +- tests/unit/mcp/test_mcp_tools.py | 2 +- tests/unit/runtime/test_task_control.py | 29 +++++ uv.lock | 2 +- 38 files changed, 632 insertions(+), 147 deletions(-) create mode 100644 docs/releases/1.2.0.md diff --git a/apps/admin_console/routers/automation.py b/apps/admin_console/routers/automation.py index ddadd12..bf9511f 100644 --- a/apps/admin_console/routers/automation.py +++ b/apps/admin_console/routers/automation.py @@ -128,7 +128,7 @@ def delete_batch(batch_id: str): except ValueError as exc: raise HTTPException(400, str(exc)) from exc try: - if not store.delete_batch(key): + if not service.delete_batch(key): raise HTTPException(404, "批次不存在") except ValueError as exc: raise HTTPException(409, str(exc)) from exc @@ -166,7 +166,8 @@ async def download_report( try: if fmt in {"screenshots", "video", "all"}: path = await asyncio.to_thread(report_export.export_session_media, session_id, fmt) - return FileResponse(path, filename=f"任务{session_id[:8]}-{fmt}.zip") + label = report_export.archive_case_name(report_export.session_report(session_id)) + return FileResponse(path, filename=f"{label}-{fmt}.zip") paths = await asyncio.to_thread(report_export.export_session, session_id, [fmt]) return FileResponse(paths[0], filename=f"任务报告-{session_id[:8]}.{fmt}") except ValueError as exc: @@ -184,13 +185,15 @@ def history( ): # SQL filtering/pagination searches the entire persisted history, not a UI cache. query = """WITH history AS ( - SELECT s.session_id, s.initial_goal, s.start_time, s.end_time, + SELECT s.session_id, s.initial_goal, s.start_time, + COALESCE(json_extract(j.payload,'$.ended_at'),s.end_time) end_time, COALESCE(j.status,s.status) status, COALESCE(json_extract(j.payload,'$.case.name'),s.initial_goal) name, COALESCE(json_extract(j.payload,'$.case.tags'),'') tags, COALESCE(json_extract(j.payload,'$.case.case_id'),'') case_id, COALESCE(json_extract(j.payload,'$.case.category'),'') category, COALESCE(json_extract(j.payload,'$.case.subcategory'),'') subcategory, - COALESCE(json_extract(j.payload,'$.case.preconditions'),'') preconditions + COALESCE(json_extract(j.payload,'$.case.preconditions'),'') preconditions, + COALESCE(json_extract(j.payload,'$.error'),'') error FROM sessions s LEFT JOIN automation_jobs j ON s.session_id=j.session_id UNION ALL SELECT j.session_id,json_extract(j.payload,'$.case.goal'), @@ -200,7 +203,8 @@ def history( COALESCE(json_extract(j.payload,'$.case.case_id'),''), COALESCE(json_extract(j.payload,'$.case.category'),''), COALESCE(json_extract(j.payload,'$.case.subcategory'),''), - COALESCE(json_extract(j.payload,'$.case.preconditions'),'') + COALESCE(json_extract(j.payload,'$.case.preconditions'),''), + COALESCE(json_extract(j.payload,'$.error'),'') FROM automation_jobs j WHERE NOT EXISTS (SELECT 1 FROM sessions s WHERE s.session_id=j.session_id) ) """ where = " WHERE status IN ('completed','success','failed','cancelled','interrupted','skipped')" @@ -226,4 +230,8 @@ def history( + " ORDER BY start_time DESC,session_id LIMIT ? OFFSET ?", [*args, size, (page - 1) * size], ).fetchall() - return {"items": [dict(r) for r in rows], "total": total, "page": page, "size": size} + items = [dict(row) for row in rows] + for item in items: + if item["case_id"]: + item["name"] = task_import.case_title(item) + return {"items": items, "total": total, "page": page, "size": size} diff --git a/apps/admin_console/routers/sessions.py b/apps/admin_console/routers/sessions.py index 4dde2e6..675103b 100644 --- a/apps/admin_console/routers/sessions.py +++ b/apps/admin_console/routers/sessions.py @@ -27,12 +27,16 @@ try: from admin_console.database.repositories.trace_repository import trace_repo from admin_console.services.media_service import media_service from admin_console.services.model_service import model_service + from admin_console.services.automation_store import store + from admin_console.services.task_import import case_title except ImportError: from apps.admin_console.core.state import state from apps.admin_console.database.repositories.session_repository import session_repo from apps.admin_console.database.repositories.trace_repository import trace_repo from apps.admin_console.services.media_service import media_service from apps.admin_console.services.model_service import model_service + from apps.admin_console.services.automation_store import store + from apps.admin_console.services.task_import import case_title router = APIRouter(tags=["sessions"]) @@ -52,6 +56,7 @@ async def list_sessions(): def _list_sessions_sync(): rows = session_repo.get_all_sessions() + jobs_by_id = store.jobs_for_sessions([str(row.get("session_id")) for row in rows]) video_rec_map = session_repo.get_video_recordings_map() latest_recordings = session_repo.get_latest_video_recordings_map() agent_names_by_session = session_repo.get_agent_trace_names_map() @@ -74,6 +79,15 @@ def _list_sessions_sync(): for row_dict in rows: s_id = str(row_dict.get("session_id")) + job = jobs_by_id.get(s_id) + if job: + case = job.get("case") or {} + row_dict["case_id"] = case.get("case_id") or "" + row_dict["display_name"] = case_title(case) or row_dict.get("initial_goal") + row_dict["error"] = job.get("error") or "" + if job.get("status") in {"completed", "failed", "cancelled", "interrupted", "skipped"}: + row_dict["status"] = job["status"] + row_dict["end_time"] = job.get("ended_at") or row_dict.get("end_time") recording = latest_recordings.get(s_id) d_info_raw = row_dict.get("device_info") device_id = None diff --git a/apps/admin_console/services/automation_service.py b/apps/admin_console/services/automation_service.py index ddaa69e..5b61b8e 100644 --- a/apps/admin_console/services/automation_service.py +++ b/apps/admin_console/services/automation_service.py @@ -3,11 +3,13 @@ import asyncio import logging import os +import shutil 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.task_import import case_goal +from apps.admin_console.services.task_import import case_goal, case_title logger = logging.getLogger(__name__) TERMINAL = {"completed", "failed", "cancelled", "skipped", "interrupted"} @@ -82,9 +84,9 @@ async def on_terminal(item: dict, status: str): job = store.job(sid) if job is None: raise RuntimeError("任务元数据写入失败") - job.update(status="finalizing", ended_at=time.time()) if job.get("timed_out"): status = "failed" + job.update(status="finalizing", terminal_status=status, ended_at=time.time()) store.save_job(job) try: job["status"] = status @@ -99,7 +101,9 @@ async def on_terminal(item: dict, status: str): } ): job["status"] = "failed" - job["error"] = "任务执行结束，但业务检查未通过" + job["error"] = job.get("error") or "任务执行结束，但业务检查未通过" + elif job["status"] == "failed" and report.get("error") and not job.get("error"): + job["error"] = report["error"] # Persist terminal classification before export, but the coordinator waits for finalization. store.save_job({**job, "status": "finalizing", "terminal_status": job["status"]}) await asyncio.to_thread(export_session, sid, job["formats"]) @@ -107,6 +111,7 @@ async def on_terminal(item: dict, status: str): except Exception as exc: logger.exception("Report finalization failed for %s", sid) job["report_error"] = str(exc)[:500] + job.pop("terminal_status", None) store.save_job(job) @@ -138,7 +143,9 @@ async def tick_batches(): and not job.get("timed_out") and time.time() - job.get("started_at", time.time()) - excluded > limit ): - job.update(error="任务达到批次配置的执行时限", timed_out=True) + job.update( + error=f"任务执行超时（{limit} 秒，不含暂停/排队时间）", timed_out=True + ) store.save_job(job) task_queue_service.stop_tasks(session_id=job["session_id"]) batch.update(status="paused", reason="任务超时，已暂停剩余任务；请确认设备状态") @@ -202,6 +209,7 @@ async def tick_batches(): runtime_overrides=options, session_id=job["session_id"], ingress="batch", + display_name=case_title(job["case"]), ) if response.get("status") == "rejected" or not response.get("tasks"): raise ValueError(response.get("error") or "队列拒绝接收任务") @@ -265,6 +273,35 @@ def create_batch(payload: dict) -> dict: return store.create_batch(batch, jobs, payload) +def delete_batch(batch_id: str) -> bool: + """Delete an inactive batch, its sessions and its generated reports.""" + from artemis.data_engine.storage import StorageManager + from apps.admin_console.services import report_export + + jobs = store.begin_delete_batch(batch_id) + if jobs is None: + return False + storage = StorageManager(store.path, report_export.TRACES_PATH) + mcp_exports = report_export.EXPORT_ROOT.parent / "mcp_exports" + with report_export.EXPORT_LOCK: + for job in jobs: + session_id = str(UUID(job["session_id"])) + storage.delete_session(UUID(session_id)) + report_dir = report_export.EXPORT_ROOT / "sessions" / session_id + if report_dir.is_dir(): + shutil.rmtree(report_dir) + mcp_session_dir = mcp_exports / "sessions" / session_id + if mcp_session_dir.is_dir(): + shutil.rmtree(mcp_session_dir) + batch_dir = report_export.EXPORT_ROOT / "batches" / batch_id + if batch_dir.is_dir(): + shutil.rmtree(batch_dir) + mcp_batch_dir = mcp_exports / "batches" / batch_id + if mcp_batch_dir.is_dir(): + shutil.rmtree(mcp_batch_dir) + return store.delete_batch(batch_id) + + def batch_action(batch_id: str, action: str) -> dict: batch = store.get("batch", batch_id) if not batch: diff --git a/apps/admin_console/services/automation_store.py b/apps/admin_console/services/automation_store.py index bc70463..40b0318 100644 --- a/apps/admin_console/services/automation_store.py +++ b/apps/admin_console/services/automation_store.py @@ -54,6 +54,13 @@ class AutomationStore: def put(self, kind: str, key: str, payload: dict) -> dict: with self.connection() as conn: + if kind == "batch": + conn.execute("BEGIN IMMEDIATE") + row = conn.execute( + "SELECT payload FROM automation_records WHERE kind='batch' AND id=?", (key,) + ).fetchone() + if row is None or json.loads(row[0]).get("status") == "deleting": + raise ValueError("批次正在删除或已删除，不能继续更新") conn.execute( "INSERT INTO automation_records VALUES (?,?,?,?) ON CONFLICT(kind,id) DO UPDATE SET payload=excluded.payload,updated_at=excluded.updated_at", (kind, key, json.dumps(payload, ensure_ascii=False), time.time()), @@ -78,6 +85,14 @@ class AutomationStore: def save_job(self, payload: dict) -> dict: with self.connection() as conn: + conn.execute("BEGIN IMMEDIATE") + if batch_id := payload.get("batch_id"): + row = conn.execute( + "SELECT payload FROM automation_records WHERE kind='batch' AND id=?", + (batch_id,), + ).fetchone() + if row is None or json.loads(row[0]).get("status") == "deleting": + raise ValueError("批次正在删除，不能继续派发任务") conn.execute( "INSERT INTO automation_jobs VALUES (?,?,?,?,?,?) ON CONFLICT(session_id) DO UPDATE SET payload=excluded.payload,status=excluded.status,updated_at=excluded.updated_at", ( @@ -106,6 +121,25 @@ class AutomationStore: ) return [{**json.loads(row[0]), "status": row[1]} for row in rows] + def jobs_for_sessions(self, session_ids: list[str]) -> dict[str, dict]: + """Load only metadata belonging to the sessions shown in history.""" + result = {} + with self.connection() as conn: + for offset in range(0, len(session_ids), 400): + chunk = session_ids[offset : offset + 400] + markers = ",".join("?" for _ in chunk) + rows = conn.execute( + f"SELECT session_id,payload,status FROM automation_jobs WHERE session_id IN ({markers})", + chunk, + ).fetchall() + result.update( + { + row["session_id"]: {**json.loads(row["payload"]), "status": row["status"]} + for row in rows + } + ) + return result + def preferences(self) -> dict: return self.get("settings", "reports") or {"formats": ["docx", "xlsx"]} @@ -164,8 +198,38 @@ class AutomationStore: conn.commit() return batch + def begin_delete_batch(self, batch_id: str) -> list[dict] | None: + """Reserve an inactive batch for cleanup before any more jobs can start.""" + with self.connection() as conn: + conn.execute("BEGIN IMMEDIATE") + row = conn.execute( + "SELECT payload FROM automation_records WHERE kind='batch' AND id=?", + (batch_id,), + ).fetchone() + if row is None: + return None + batch = json.loads(row[0]) + active = conn.execute( + "SELECT 1 FROM automation_jobs WHERE batch_id=? AND status IN " + "('queued','running','task_paused','resuming','finalizing') LIMIT 1", + (batch_id,), + ).fetchone() + if batch["status"] == "running" or active: + raise ValueError("批次仍有任务执行中或排队中，请等待完成后删除") + batch["status"] = "deleting" + conn.execute( + "UPDATE automation_records SET payload=?,updated_at=? WHERE kind='batch' AND id=?", + (json.dumps(batch, ensure_ascii=False), time.time(), batch_id), + ) + jobs = conn.execute( + "SELECT payload,status FROM automation_jobs WHERE batch_id=? ORDER BY position", + (batch_id,), + ).fetchall() + conn.commit() + return [{**json.loads(job[0]), "status": job[1]} for job in jobs] + def delete_batch(self, batch_id: str) -> bool: - """Remove a batch record; retain executed job metadata and its reports.""" + """Remove a batch record and all of its jobs after cleanup.""" with self.connection() as conn: conn.execute("BEGIN IMMEDIATE") row = conn.execute( @@ -175,17 +239,9 @@ class AutomationStore: if row is None: return False batch = json.loads(row[0]) - active = conn.execute( - "SELECT 1 FROM automation_jobs WHERE batch_id=? AND status IN " - "('queued','running','task_paused','resuming','finalizing') LIMIT 1", - (batch_id,), - ).fetchone() - if batch["status"] == "running" or active: - raise ValueError("批次仍有任务执行中或排队中，请等待完成后删除") - conn.execute( - "DELETE FROM automation_jobs WHERE batch_id=? AND status='pending'", - (batch_id,), - ) + if batch.get("status") != "deleting": + raise ValueError("批次尚未准备删除") + conn.execute("DELETE FROM automation_jobs WHERE batch_id=?", (batch_id,)) conn.execute( "DELETE FROM automation_records WHERE kind='batch' AND id=?", (batch_id,), diff --git a/apps/admin_console/services/report_export.py b/apps/admin_console/services/report_export.py index 0c7496e..cb8aef4 100644 --- a/apps/admin_console/services/report_export.py +++ b/apps/admin_console/services/report_export.py @@ -22,6 +22,7 @@ from artemis.config import IMAGES_DIR, TRACES_PATH, WORKSPACE_ROOT from apps.admin_console.database.repositories.session_repository import SessionRepository from apps.admin_console.database.repositories.step_repository import StepRepository from apps.admin_console.services.automation_store import store, identifier +from apps.admin_console.services.task_import import case_title EXPORT_ROOT = WORKSPACE_ROOT / "artifacts" / "reports" EXPORT_LOCK = threading.RLock() @@ -40,6 +41,14 @@ STATUS = { } +def archive_case_name(report: dict) -> str: + """Keep case identifiers readable without allowing ZIP paths or Windows-invalid names.""" + fallback = str(report.get("id") or "task")[:8] + candidate = str(report.get("case_id") or fallback).strip() + candidate = re.sub(r'[\\/:*?"<>|\x00-\x1f]', "_", candidate).strip(" .") + return candidate[:80] or fallback + + def read_json(path: Path, fallback=None): try: return json.loads(path.read_text(encoding="utf-8")) @@ -83,6 +92,9 @@ def failure_reason(directory: Path, status: str, recorded: str | None) -> str: return redact(recorded) if status not in {"failed", "interrupted"}: return "" + state = read_json(directory / "status.json", {}) + if isinstance(state, dict) and state.get("error"): + return redact(state["error"]) for name in ("stdout.log", "stderr.log"): path = directory / name try: @@ -96,6 +108,8 @@ def failure_reason(directory: Path, status: str, recorded: str | None) -> str: match = re.search(r"Error running automation:\s*(.+)", line) if match: return redact(match.group(1)) + if re.search(r"(?i)(?:timeout|timed out|超时|deadline exceeded)", line): + return redact("模型请求或任务执行超时：" + line.strip()) return "任务未完成，运行日志中没有可提取的具体异常；请查看执行步骤。" @@ -142,7 +156,7 @@ def session_report(session_id: str, db_path=None) -> dict: ) return { "id": session_id, - "name": case.get("name") or session.get("initial_goal", session_id)[:120], + "name": case_title(case) or session.get("initial_goal", session_id)[:120], "goal": case.get("goal") or session.get("initial_goal", ""), "expected": case.get("expected", ""), "status": status, @@ -638,15 +652,18 @@ def export_session_media(session_id: str, kind: str) -> Path: "videos": [name for _, name in videos], "note": "录屏按时间顺序分段；暂停期间未占用设备，也未录制其他任务。", } + prefix = archive_case_name(report) + "/" if kind == "all" else "" with zipfile.ZipFile(temporary, "w", zipfile.ZIP_DEFLATED) as archive: if kind == "all": for path in export_session(session_id, ["docx", "xlsx"]): - archive.write(path, path.name) + archive.write(path, prefix + path.name) for source, name in images: - archive.write(source, f"步骤截图/{name}") + archive.write(source, f"{prefix}步骤截图/{name}") for source, name in videos: - archive.write(source, f"任务录屏/{name}") - archive.writestr("文件清单.json", json.dumps(manifest, ensure_ascii=False, indent=2)) + archive.write(source, f"{prefix}任务录屏/{name}") + archive.writestr( + prefix + "文件清单.json", json.dumps(manifest, ensure_ascii=False, indent=2) + ) temporary.replace(target) return target @@ -691,6 +708,8 @@ def export_batch(batch_id: str) -> Path: archive.write(path, path.name) for number, report in enumerate(records, 1): for path in export_session(report["id"], formats): - archive.write(path, f"任务报告/{number:04d}_{report['id']}/{path.name}") + archive.write( + path, f"任务报告/{number:04d}_{archive_case_name(report)}/{path.name}" + ) temporary.replace(target) return target diff --git a/apps/admin_console/services/task_import.py b/apps/admin_console/services/task_import.py index f612044..6567d42 100644 --- a/apps/admin_console/services/task_import.py +++ b/apps/admin_console/services/task_import.py @@ -177,3 +177,10 @@ def parse_workbook(data: bytes) -> dict: def case_goal(case: dict) -> str: """Execution gets the prompt verbatim; metadata stays outside model input.""" return case["goal"] + + +def case_title(case: dict) -> str: + """Human-facing title; never prepend metadata to the model prompt.""" + case_id = str(case.get("case_id") or "").strip() + name = str(case.get("name") or "").strip() + return " ".join(part for part in (case_id, name) if part) diff --git a/apps/admin_console/services/task_queue_service.py b/apps/admin_console/services/task_queue_service.py index 2389fff..b407524 100644 --- a/apps/admin_console/services/task_queue_service.py +++ b/apps/admin_console/services/task_queue_service.py @@ -1133,6 +1133,7 @@ class TaskQueueService: verification_level: str | None = None, explorer_mode: str | None = None, runtime_overrides: dict[str, Any] | None = None, + display_name: str | None = None, ) -> dict[str, Any]: """Enqueues one or more goals and wakes up the background worker. @@ -1189,6 +1190,8 @@ class TaskQueueService: ) if runtime_overrides: task_item["runtime_overrides"] = runtime_overrides + if display_name: + task_item["display_name"] = display_name state.queue_items.append(task_item) enqueued_tasks.append(task_item) cls._broadcast_startup_progress( diff --git a/apps/showcase_ui/package-lock.json b/apps/showcase_ui/package-lock.json index 1858911..a73614d 100644 --- a/apps/showcase_ui/package-lock.json +++ b/apps/showcase_ui/package-lock.json @@ -1,12 +1,12 @@ { "name": "artemis-showcase-ui", - "version": "1.1.2", + "version": "1.2.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "artemis-showcase-ui", - "version": "1.1.2", + "version": "1.2.0", "dependencies": { "@angular/cdk": "^22.1.5", "@angular/common": "^22.1.4", diff --git a/apps/showcase_ui/package.json b/apps/showcase_ui/package.json index bb3a62f..932baa9 100644 --- a/apps/showcase_ui/package.json +++ b/apps/showcase_ui/package.json @@ -1,6 +1,6 @@ { "name": "artemis-showcase-ui", - "version": "1.1.2", + "version": "1.2.0", "scripts": { "ng": "ng", "start": "ng serve --proxy-config proxy.conf.json", diff --git a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html index 05648f5..4676d4d 100644 --- a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html +++ b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html @@ -76,6 +76,7 @@ limitations under the License. class="dropdown-item-card" [class.selected]="agentService.currentSessionId() === session.session_id" [class.running]="getTaskStatus(session) === 'running'" + [ngClass]="getTaskStatus(session)" (click)="selectTask(session.session_id, $event)" >
@@ -96,7 +97,7 @@ limitations under the License. · } {{ session.start_time * 1000 | date:'shortTime' }} - @if (getTaskStatus(session) === 'running' || getTaskStatus(session) === 'paused') { + @if (['running', 'paused', 'task_paused', 'resuming'].includes(getTaskStatus(session))) { } - @if (getTaskStatus(session) !== 'running' && getTaskStatus(session) !== 'paused') { + @if (!['running', 'paused', 'task_paused', 'resuming'].includes(getTaskStatus(session))) {
-
- {{ session.initial_goal || '未命名任务' }} +
+ {{ session.display_name || session.initial_goal || '未命名任务' }}
+ @if (session.error && getTaskStatus(session) === 'failed') { +
{{session.error}}
+ }
- {{ session.start_time * 1000 | date:'MM/dd/yy' }} + {{ session.start_time * 1000 | date:'MM/dd/yy' }} · 已用 {{ formatTaskDuration(session.start_time, session.end_time) }}
} diff --git a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.scss b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.scss index 6e3c149..38ad366 100644 --- a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.scss +++ b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.scss @@ -3684,6 +3684,12 @@ $stream-row-height: 25px; &.running { border-left: 3.5px solid #2563eb; } + &.paused, &.task_paused { border-left: 3.5px solid #d97706; background: #fffbeb; } + &.resuming, &.queued { border-left: 3.5px solid #7c3aed; } + &.failed { border-left: 3.5px solid #dc2626; background: #fff7f7; } + &.completed { border-left: 3.5px solid #16a34a; } + &.cancelled, &.interrupted { border-left: 3.5px solid #ea580c; } + .item-error { color: #b91c1c; font-size: 11px; line-height: 1.4; overflow-wrap: anywhere; } .item-header { display: flex; @@ -3729,12 +3735,24 @@ $stream-row-height: 25px; border-color: #fdba74; &::before { background-color: #f97316; } } + &.task_paused { + background: #ffedd5; + color: #9a3412; + border-color: #fb923c; + &::before { background-color: #ea580c; } + } + &.resuming, &.queued { + background: #f3e8ff; + color: #6b21a8; + border-color: #c4b5fd; + &::before { background-color: #7c3aed; } + } &.completed { - background: #fafafa; - color: #1f1f1f; - border-color: #e2e8f0; + background: #dcfce7; + color: #166534; + border-color: #86efac; &::before { - background-color: #1f1f1f; + background-color: #16a34a; } } &.pending { @@ -3746,21 +3764,22 @@ $stream-row-height: 25px; } } &.failed { - background: #fafafa; - color: #ef4444; - border-color: #fee2e2; + background: #fee2e2; + color: #991b1b; + border-color: #fca5a5; &::before { background-color: #ef4444; } } &.cancelled { - background: #fafafa; - color: #d97706; - border-color: #fef3c7; + background: #ffedd5; + color: #9a3412; + border-color: #fdba74; &::before { - background-color: #f59e0b; + background-color: #ea580c; } } + &.interrupted { background: #fef3c7; color: #92400e; border-color: #fbbf24; &::before { background-color: #d97706; } } .status-dot { display: none; diff --git a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts index 79da82b..5fbd2af 100644 --- a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts +++ b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts @@ -388,13 +388,13 @@ export class AgentStreamComponent implements AfterViewInit { public activeQueue = computed(() => { const list = this.agentService.sessions().filter((s) => { const status = this.getTaskStatus(s); - return status === 'running' || status === 'paused' || status === 'pending'; + return ['running', 'paused', 'task_paused', 'resuming', 'pending', 'queued'].includes(status); }); return list.sort((a, b) => { const statusA = this.getTaskStatus(a); const statusB = this.getTaskStatus(b); - const isRunA = statusA === 'running' || statusA === 'paused'; - const isRunB = statusB === 'running' || statusB === 'paused'; + const isRunA = ['running', 'paused', 'task_paused', 'resuming'].includes(statusA); + const isRunB = ['running', 'paused', 'task_paused', 'resuming'].includes(statusB); if (isRunA && !isRunB) return -1; if (!isRunA && isRunB) return 1; return (a.start_time || 0) - (b.start_time || 0); @@ -404,7 +404,7 @@ export class AgentStreamComponent implements AfterViewInit { public historyTasks = computed(() => { return this.agentService.sessions().filter((s) => { const status = this.getTaskStatus(s); - return status === 'completed' || status === 'failed' || status === 'cancelled'; + return ['completed', 'failed', 'cancelled', 'interrupted', 'skipped'].includes(status); }); }); @@ -491,7 +491,7 @@ export class AgentStreamComponent implements AfterViewInit { const terminal = ['completed', 'failed', 'cancelled', 'interrupted'] .includes(this.agentService.currentSession()?.status || ''); if (event.status === 'pending') { - return terminal ? '任务结束前未进入模型' : '已接收，等待下一步'; + return terminal ? '任务结束前未进入模型' : '已接收，正在打断当前模型步骤'; } return terminal ? '模型已读取，未留下后续动作记录' : '模型已读取，等待动作记录'; } @@ -912,14 +912,14 @@ export class AgentStreamComponent implements AfterViewInit { return tuningLabel(kind, id); } - public getTaskStatus(session: Session): 'running' | 'paused' | 'completed' | 'pending' | 'failed' | 'cancelled' { + public getTaskStatus(session: Session): string { if (session.status) { const s = session.status.toLowerCase(); - if (s === 'completed' || s === 'success' || s === 'failed' || s === 'cancelled') { - return (s === 'success' ? 'completed' : s) as any; + if (['completed', 'success', 'failed', 'cancelled', 'interrupted', 'skipped'].includes(s)) { + return s === 'success' ? 'completed' : s; } - if (s === 'running' || s === 'paused' || s === 'pending') { - return s as any; + if (['running', 'paused', 'task_paused', 'resuming', 'pending', 'queued'].includes(s)) { + return s; } } if (session.session_id === this.agentService.runningSessionId() && (this.agentService.agentStatus() === 'running' || this.agentService.agentStatus() === 'paused')) { @@ -1360,6 +1360,14 @@ export class AgentStreamComponent implements AfterViewInit { return isAndroidAction(action); } + public formatTaskDuration(start?: number, end?: number): string { + if (!start || !Number.isFinite(start)) return '—'; + const seconds = Math.max(0, Math.floor((end || Date.now() / 1000) - start)); + const hours = Math.floor(seconds / 3600); + const minutes = Math.floor((seconds % 3600) / 60); + return hours ? `${hours}时${minutes}分` : minutes ? `${minutes}分${seconds % 60}秒` : `${seconds}秒`; + } + public hasRecordedDeviceAction(action: any): boolean { const actions = Array.isArray(action) ? action : action ? [action] : []; return actions.some(item => this.isAndroidAction(item) && !this.isReportStatusAction(item)); diff --git a/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.html b/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.html index 7028cec..e98894e 100644 --- a/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.html +++ b/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.html @@ -28,7 +28,7 @@ @if (liveTask() && agent.currentSessionId()) {

任务指导 · {{label(taskStatus())}}

-

{{ taskStatus() === 'paused' ? '模型请求已暂停；提交指导后请在执行日志中恢复模型调用。' : '运行中提交将在下一步执行前生效；暂停后会释放设备，提交指导即可重新排队恢复。' }}

+

{{ taskStatus() === 'paused' ? '模型请求已暂停；提交指导后请在执行日志中恢复模型调用。' : '运行中提交会立即打断模型思考并重新观察设备；正在执行的设备动作会先安全结束。' }}

@@ -61,7 +61,7 @@ @for(b of batches(); track b.id){
- +
} @if(!batches().length){

暂无批次。

} @@ -70,10 +70,10 @@
@if(b.status === 'running'){} @if(b.status === 'draft' || b.status === 'paused'){} - @if(b.status !== 'running'){} + @if(b.status !== 'running' && b.status !== 'deleting'){} 下载批次报告 ZIP

服务重启后等待手动继续。失败项不会自动重试；暂停不打断当前任务。

@if(b.report_error){

报告生成失败：{{b.report_error}}

} - @for(j of b.jobs; track j.session_id){
{{label(j.status)}}@if(j.error){{{j.error}}}
} + @for(j of b.jobs; track j.session_id){
{{label(j.status)}}@if(j.error){{{j.error}}}
} } } diff --git a/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.ts b/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.ts index a4f852f..95d42ff 100644 --- a/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.ts +++ b/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.ts @@ -38,7 +38,7 @@ export class AutomationPanelComponent { }); } formats() { return [ ...(this.word ? ['docx'] : []), ...(this.excel ? ['xlsx'] : []) ]; } - label(s: string) { return ({draft:'待开始',running:'执行中',paused:'已暂停',task_paused:'等待指导',resuming:'等待设备恢复',completed:'执行完成',failed:'失败',cancelled:'已取消',pending:'待执行',queued:'排队中',interrupted:'中断',skipped:'跳过',finalizing:'正在生成报告'} as Record)[s] || s; } + label(s: string) { return ({draft:'待开始',running:'执行中',paused:'已暂停',task_paused:'等待指导',resuming:'等待设备恢复',completed:'执行完成',failed:'失败',cancelled:'已取消',pending:'待执行',queued:'排队中',interrupted:'中断',skipped:'跳过',finalizing:'正在生成报告',deleting:'删除中，可重试清理'} as Record)[s] || s; } renderReport(content: string) { return renderMarkdownToHtml(content || ''); } async action(fn: () => Promise) { if (this.busy()) return; @@ -73,12 +73,13 @@ export class AutomationPanelComponent { }); } selectBatch(id: string) { void this.action(async () => { this.batch.set(await firstValueFrom(this.http.get(`/api/automation/batches/${id}`))); }); } deleteBatch(item: { id: string; name: string }) { - if (!window.confirm(`删除批次“${item.name}”？未执行任务会移除，已执行任务与报告仍保留。`)) return; + if (!window.confirm(`删除批次“${item.name}”及其全部任务记录、报告和文件？此操作无法撤销。`)) return; void this.action(async () => { await firstValueFrom(this.http.delete(`/api/automation/batches/${encodeURIComponent(item.id)}`)); if (this.batch()?.id === item.id) this.batch.set(null); await this.refresh(); - this.notice.set('批次记录已删除；已执行任务与报告仍可在历史中查看'); + this.agent.fetchSessions(); + this.notice.set('批次及关联任务、报告和文件已删除'); }); } batchAction(action: string) { void this.action(async () => { this.batch.set(await firstValueFrom(this.http.post(`/api/automation/batches/${this.batch().id}/${action}`,{}))); await this.refresh(); }); } diff --git a/apps/showcase_ui/src/app/components/task-history/task-history.component.spec.ts b/apps/showcase_ui/src/app/components/task-history/task-history.component.spec.ts index 9e43f84..b62b0fd 100644 --- a/apps/showcase_ui/src/app/components/task-history/task-history.component.spec.ts +++ b/apps/showcase_ui/src/app/components/task-history/task-history.component.spec.ts @@ -49,6 +49,21 @@ describe('历史任务检索', () => { fixture.destroy(); }); + it('显示明确的结果、完成时间与总耗时，标题悬停显示完整提示词', () => { + const fixture = TestBed.createComponent(TaskHistoryComponent); + http.expectOne(r => r.url === '/api/automation/history').flush({total: 1, items: [ + {session_id:'case-1', name:'1.1 打开并核对设置', initial_goal:'请打开设置首页并核对两个真实分类', status:'failed', start_time:100, end_time:165, error:'任务执行超时（60 秒）'} + ]}); + fixture.detectChanges(); + const row: HTMLButtonElement = fixture.nativeElement.querySelector('button.row'); + expect(row.getAttribute('title')).toBe('请打开设置首页并核对两个真实分类'); + expect(row.textContent).toContain('1.1 打开并核对设置'); + expect(row.textContent).toContain('总耗时 1分5秒'); + expect(row.textContent).toContain('任务执行超时'); + expect(fixture.nativeElement.querySelector('.status-failed .history-status').textContent).toContain('失败'); + fixture.destroy(); + }); + it('日期筛选用本地零点到次日零点，后端分页参数独立传递', () => { const fixture = TestBed.createComponent(TaskHistoryComponent); http.expectOne(r => r.url === '/api/automation/history').flush({total:0,items:[]}); diff --git a/apps/showcase_ui/src/app/components/task-history/task-history.component.ts b/apps/showcase_ui/src/app/components/task-history/task-history.component.ts index d837f52..baa4804 100644 --- a/apps/showcase_ui/src/app/components/task-history/task-history.component.ts +++ b/apps/showcase_ui/src/app/components/task-history/task-history.component.ts @@ -13,12 +13,12 @@ import { AgentService } from '../../services/agent.service'; @if(error()){

{{error()}}

} @if(loading()){正在查询…} @for(group of groups(); track group.key){
{{group.title}} 本页 {{group.rows.length}} 条 - @for(row of group.rows; track row.session_id){
} + @for(row of group.rows; track row.session_id){
}
} @if(!loading() && !total()){

没有符合条件的历史任务。

}
{{page}} / {{pages()}}
`, - styles:[`:host{display:block;font-size:12px;color:inherit}h3{font-size:14px;margin:14px 0 8px}small{opacity:.6}input,select,button{font:inherit;color:inherit;border:1px solid #64748b55;border-radius:6px;padding:8px;background:transparent;box-sizing:border-box}input{width:100%}select option{background:white;color:#1e293b}.filters{display:flex;gap:6px;flex-wrap:wrap;margin:8px 0}.filters label{flex:1;min-width:110px}.filters input,.filters select{display:block;width:100%}summary{padding:10px 0;cursor:pointer}.row{display:flex;flex-direction:column;gap:4px;width:100%;text-align:left;cursor:pointer;margin:5px 0;overflow-wrap:anywhere}.row strong{font-weight:500}.row span{opacity:.75}.row.selected{border-color:#60a5fa;background:#3b82f615}.history-item{position:relative;border:1px solid #64748b55;border-radius:6px;margin:5px 0;padding:8px}.history-item:has(.row.selected){border-color:#60a5fa;background:#3b82f615}.history-item .row{border:0;padding:0;margin:0 0 4px}.history-item .row.selected{background:transparent}.history-item .row{padding-right:35px}.delete-history{position:absolute;right:5px;top:8px;border:0;padding:3px;color:#64748b;cursor:pointer;font-size:12px}.delete-history .material-symbols-outlined{font-size:16px}.delete-history:hover{color:#b91c1c}.pager{display:flex;align-items:center;justify-content:space-between;gap:8px;margin:12px 0}button:disabled{opacity:.4}p{opacity:.7}`] + styles:[`:host{display:block;font-size:12px;color:inherit}h3{font-size:14px;margin:14px 0 8px}small{opacity:.6}input,select,button{font:inherit;color:inherit;border:1px solid #64748b55;border-radius:6px;padding:8px;background:transparent;box-sizing:border-box}input{width:100%}select option{background:white;color:#1e293b}.filters{display:flex;gap:6px;flex-wrap:wrap;margin:8px 0}.filters label{flex:1;min-width:110px}.filters input,.filters select{display:block;width:100%}summary{padding:10px 0;cursor:pointer}.row{display:flex;flex-direction:column;gap:6px;width:100%;text-align:left;cursor:pointer;margin:5px 0;overflow-wrap:anywhere}.row strong{font-weight:600}.row.selected{border-color:#60a5fa;background:#3b82f615}.history-item{position:relative;border:1px solid #cbd5e1;border-left:3px solid #94a3b8;border-radius:8px;margin:6px 0;padding:8px;background:#fff}.history-item.status-completed,.history-item.status-success{border-left-color:#16a34a}.history-item.status-failed{border-left-color:#dc2626;background:#fffafa}.history-item.status-cancelled,.history-item.status-interrupted{border-left-color:#d97706;background:#fffdf8}.history-item.status-skipped{border-left-color:#64748b}.history-item:has(.row.selected){border-color:#60a5fa;background:#eff6ff}.history-item .row{border:0;padding:0;margin:0 0 4px}.history-item .row.selected{background:transparent}.history-item .row{padding-right:35px}.history-meta{display:flex;gap:5px;align-items:center;flex-wrap:wrap;color:#475569;font-size:11px}.history-status{display:inline-block;border-radius:5px;padding:2px 6px;background:#f1f5f9;color:#334155}.status-completed .history-status,.status-success .history-status{background:#dcfce7;color:#166534}.status-failed .history-status{background:#fee2e2;color:#991b1b}.status-cancelled .history-status,.status-interrupted .history-status{background:#fef3c7;color:#92400e}.history-error{font-size:11px;color:#b91c1c;line-height:1.4}.delete-history{position:absolute;right:5px;top:8px;border:0;padding:3px;color:#64748b;cursor:pointer;font-size:12px}.delete-history .material-symbols-outlined{font-size:16px}.delete-history:hover{color:#b91c1c}.pager{display:flex;align-items:center;justify-content:space-between;gap:8px;margin:12px 0}button:disabled{opacity:.4}p{opacity:.7}`] }) export class TaskHistoryComponent { agent=inject(AgentService); private http=inject(HttpClient); query='';start='';end='';status='';page=1; @@ -27,7 +27,8 @@ export class TaskHistoryComponent { constructor(){this.load();const timer=setInterval(()=>this.load(),15000);inject(DestroyRef).onDestroy(()=>{this.revision++;clearInterval(timer);clearTimeout(this.debounce);});} remove(id:string){if(!confirm("确定删除此历史任务？此操作无法撤销。"))return;this.agent.deleteSession(id).subscribe({next:()=>this.load(),error:()=>this.error.set("删除失败，请稍后重试")});} pages(){return Math.max(1,Math.ceil(this.total()/30));} - label(s:string){return ({completed:'执行完成',success:'执行完成',failed:'失败',cancelled:'已取消',interrupted:'中断',skipped:'跳过'} as Record)[s] || s;} + label(s:string){return ({completed:'执行完成',success:'执行完成',failed:'失败',cancelled:'已终止',interrupted:'中断',skipped:'跳过'} as Record)[s] || s;} + duration(start:number,end?:number){if(!start||!end)return '—';const seconds=Math.max(0,Math.floor(end-start));const hours=Math.floor(seconds/3600);const minutes=Math.floor((seconds%3600)/60);return hours?`${hours}时${minutes}分`:minutes?`${minutes}分${seconds%60}秒`:`${seconds}秒`;} toggle(key:string,event:Event){if((event.target as HTMLDetailsElement).open)this.collapsed.delete(key);else this.collapsed.add(key);} search(){this.page=1;this.revision++;clearTimeout(this.debounce);this.debounce=setTimeout(()=>this.load(),250);} load(){ diff --git a/apps/showcase_ui/src/app/core/models/session.model.ts b/apps/showcase_ui/src/app/core/models/session.model.ts index ef322d1..c2f15d7 100644 --- a/apps/showcase_ui/src/app/core/models/session.model.ts +++ b/apps/showcase_ui/src/app/core/models/session.model.ts @@ -38,6 +38,9 @@ export interface Session { start_time: number; end_time?: number; status?: string; + display_name?: string; + case_id?: string; + error?: string; video_url?: string; recording_status?: 'recording' | 'finalizing' | 'processing' | 'ready' | 'failed' | 'unavailable'; model_info?: ModelInfo; diff --git a/apps/showcase_ui/src/app/pages/workspace/workspace.component.html b/apps/showcase_ui/src/app/pages/workspace/workspace.component.html index 37ed9c4..4cd4265 100644 --- a/apps/showcase_ui/src/app/pages/workspace/workspace.component.html +++ b/apps/showcase_ui/src/app/pages/workspace/workspace.component.html @@ -34,7 +34,7 @@ limitations under the License.
+[disabled]="guidanceBusy()" placeholder="输入指导；将中断当前模型思考并立即重新规划">
{{ status === 'paused' ? '提交指导后请在执行日志中恢复模型调用' : '下方输入框用于开始新任务' }} @if (status === 'running') { diff --git a/apps/showcase_ui/src/app/services/agent.service.ts b/apps/showcase_ui/src/app/services/agent.service.ts index 2f78f9c..2a7bc6c 100644 --- a/apps/showcase_ui/src/app/services/agent.service.ts +++ b/apps/showcase_ui/src/app/services/agent.service.ts @@ -108,7 +108,7 @@ export class AgentService { const pendingMatch = pendingById.get(s.session_id); const isCurrentActive = ((status === 'running' || status === 'paused') && runId === s.session_id) || !!activeMatch; - const isTerminal = s.status === 'completed' || s.status === 'success' || s.status === 'failed' || s.status === 'cancelled'; + const isTerminal = ['completed', 'success', 'failed', 'cancelled', 'interrupted', 'skipped'].includes(s.status || ''); let sStatus = s.status; if (!isTerminal) { const isPending = !!pendingMatch && !isCurrentActive; @@ -1515,12 +1515,12 @@ export class AgentService { return; // Paused while the tab is in the background; resumed by visibilitychange. } this.fetchStatus(); + const selected = this.currentSessionId(); + if (selected) this.fetchGuidance(selected); this.pollCounter++; // Periodically refresh sessions every 6 seconds (every 3 polling intervals of 2s) to stay in sync with external DB modifications if (this.pollCounter % 3 === 0) { this.fetchSessions(); - const selected = this.currentSessionId(); - if (selected) this.fetchGuidance(selected); } }, 2000); }); diff --git a/apps/showcase_ui/src/app/utils/ui-label.pipe.ts b/apps/showcase_ui/src/app/utils/ui-label.pipe.ts index cc084f0..688846e 100644 --- a/apps/showcase_ui/src/app/utils/ui-label.pipe.ts +++ b/apps/showcase_ui/src/app/utils/ui-label.pipe.ts @@ -1,9 +1,10 @@ import { Pipe, PipeTransform } from '@angular/core'; const LABELS: Record = { - running: '执行中', queued: '排队中', pending: '等待中', completed: '已完成', + running: '执行中', queued: '排队中', pending: '等待中', completed: '执行完成', success: '成功', succeeded: '成功', failed: '失败', error: '错误', paused: '已暂停', - cancelled: '已取消', canceled: '已取消', stopped: '已停止', unknown: '未知', + cancelled: '已终止', canceled: '已终止', stopped: '已终止', interrupted: '中断', + task_paused: '等待指导', resuming: '恢复排队', skipped: '已跳过', unknown: '未知', idle: '空闲', starting: '启动中', launching: '启动中', ready: '就绪', pass: '通过', fail: '失败', warn: '警告', warning: '警告', blocked: '受阻', done: '完成', active: '当前', waiting: '等待中', processing: '处理中', diff --git a/artemis/agents/flash/runner.py b/artemis/agents/flash/runner.py index 5b17acb..c247fae 100644 --- a/artemis/agents/flash/runner.py +++ b/artemis/agents/flash/runner.py @@ -45,7 +45,7 @@ import re import uuid from artemis.config.settings import settings -from artemis.runtime.task_control import TaskControl +from artemis.runtime.task_control import TaskControl, has_pending_instruction from jinja2 import Template from langchain_core.messages import ( @@ -937,6 +937,9 @@ class FlashRunner: action_names = self.executor.action_tool_names index_elements = self._turn_index_snapshot(state) for tc in tool_calls: + control = getattr(self.ctx, "task_control", None) + if isinstance(control, TaskControl) and has_pending_instruction(control.directory): + return None, pre_screenshot_bytes, xml_list, action_sequence name = tc["name"].split(":")[-1] if ":" in tc["name"] else tc["name"] args = tc.get("args") or {} tc_id = tc.get("id") or str(uuid.uuid4()) @@ -1118,7 +1121,36 @@ class FlashRunner: turn_base = len(messages) - 1 current_tools = report_only_tools if is_final else tools_declaration - response = await self._invoke_model(llm, current_tools, messages) + if isinstance(control, TaskControl): + interrupted, response = await control.run_interruptibly( + self._invoke_model(llm, current_tools, messages), + on_pause=getattr(self.ctx, "on_task_pause", None), + on_resume=getattr(self.ctx, "on_task_resume", None), + ) + else: + interrupted, response = ( + False, + await self._invoke_model(llm, current_tools, messages), + ) + + if interrupted or ( + isinstance(control, TaskControl) and has_pending_instruction(control.directory) + ): + # Do not dispatch a stale action or final answer from a prompt + # assembled before the instruction arrived. + ( + shot_path, + current_pre_screenshot_bytes, + current_xml_list, + ) = await capture_screenshot_and_parse_ui( + self.ctx, state, self.controller, skip_settling=False + ) + state.latest_screenshot = shot_path + pending_notices.append( + "新指导已打断旧模型步骤。请重新观察画面并按最新指导决定下一步。" + ) + turns -= 1 + continue if response is None: break @@ -1145,6 +1177,18 @@ class FlashRunner: self._record_llm_trace(step_token_usage, raw_text) tool_calls = self._resolve_tool_calls(response, raw_text) + if isinstance(control, TaskControl) and has_pending_instruction(control.directory): + ( + shot_path, + current_pre_screenshot_bytes, + current_xml_list, + ) = await capture_screenshot_and_parse_ui( + self.ctx, state, self.controller, skip_settling=False + ) + state.latest_screenshot = shot_path + pending_notices.append("新指导已收到；忽略旧模型动作，重新观察画面。") + turns -= 1 + continue if not tool_calls: logger.info( diff --git a/artemis/agents/operator/operator.py b/artemis/agents/operator/operator.py index b947a61..63b2471 100644 --- a/artemis/agents/operator/operator.py +++ b/artemis/agents/operator/operator.py @@ -33,7 +33,7 @@ from artemis.graph.state import State from artemis.graph.visibility import strict_state from artemis.mcp.action_specs import OPERATOR_SHELL_ORDER, operator_shell_tool from artemis.llm.usage import usage_from_message -from artemis.runtime.task_control import TaskControl +from artemis.runtime.task_control import TaskControl, has_pending_instruction from artemis.services.llm import acomplete, get_llm, invoke_llm_with_timeout_message from artemis.tools.command_tool import ( analyze_task_output_wrapper, @@ -539,6 +539,9 @@ class OperatorNode: if interrupted: return None, None, None, False assert response is not None + if isinstance(control, TaskControl) and has_pending_instruction(control.directory): + # The model answered a prompt that predates the user's new direction. + return None, None, None, False usage = usage_from_message(response) # The ledger's compaction thresholds read the operator's own @@ -690,10 +693,10 @@ class OperatorNode: if normalize_name(tool_name) in INTERRUPTIBLE_HELPERS: interrupted, result_obj = await self._await_or_pause(tool_call) if interrupted: - span.result = "任务暂停；诊断已中断，恢复后重新观察屏幕" + span.result = "任务被暂停或指导打断；重新观察屏幕后继续" results.append((span.result, "error")) for remaining in other_calls[len(results) :]: - results.append(("任务暂停，未执行该工具", "error")) + results.append(("任务已打断，未执行该工具", "error")) for completed, (content, status) in zip( other_calls, results ): @@ -710,7 +713,7 @@ class OperatorNode: current_messages.append( ToolMessage( tool_call_id=action["id"], - content="任务暂停，未执行屏幕操作", + content="任务已打断，未执行屏幕操作", status="error", ) ) @@ -827,6 +830,9 @@ class OperatorNode: action_calls = [] + if isinstance(control, TaskControl) and has_pending_instruction(control.directory): + return None, None, None, False + if action_calls: logger.info( f"Operator requested {len(action_calls)} action(s)." diff --git a/artemis/data_engine/storage.py b/artemis/data_engine/storage.py index 76446c5..a2a4184 100644 --- a/artemis/data_engine/storage.py +++ b/artemis/data_engine/storage.py @@ -1158,11 +1158,21 @@ class StorageManager: """Delete all data associated with a session, including files on disk.""" session_id_str = str(session_id) - # 1. Get video paths before deleting from DB + # 1. Get files referenced by this session before deleting from DB. video_paths = [] video_ids: list[str] = [] + image_names: set[str] = set() try: with self._get_connection() as conn: + image_names = { + str(name) + for row in conn.execute( + "SELECT pre_image_name, post_image_name FROM steps WHERE session_id = ?", + (session_id_str,), + ) + for name in row + if name + } cursor = conn.execute( "SELECT video_id, local_video_path FROM video_recordings WHERE session_id = ?", (session_id_str,), @@ -1173,9 +1183,11 @@ class StorageManager: if row["local_video_path"]: video_paths.append(Path(row["local_video_path"])) except sqlite3.OperationalError as e: - logger.warning(f"Could not query video_recordings: {e}") + logger.warning(f"Could not query session media: {e}") + raise # 2. Delete from DB tables + orphaned_images: list[str] = [] with self._get_connection() as conn: conn.execute("PRAGMA foreign_keys = OFF") board_keys = [f"session:{session_id_str}"] + [ @@ -1207,13 +1219,26 @@ class StorageManager: (session_id_str,), ) except sqlite3.OperationalError as e: - logger.warning(f"Failed to delete from table {table}: {e}") - conn.commit() - conn.execute("VACUUM") + logger.error(f"Failed to delete from table {table}: {e}") + raise + for name in image_names: + if not conn.execute( + "SELECT 1 FROM steps WHERE pre_image_name=? OR post_image_name=? LIMIT 1", + (name, name), + ).fetchone(): + conn.execute("DELETE FROM images WHERE image_name=?", (name,)) + orphaned_images.append(name) conn.commit() logger.info(f"Database records for session {session_id} cleared.") + images_root = (self.base_trace_dir / "images").resolve() + for name in orphaned_images: + if len(name) == 64 and all(c in "0123456789abcdef" for c in name): + image_path = (images_root / f"{name}.jpg").resolve() + if image_path.is_relative_to(images_root): + image_path.unlink(missing_ok=True) + # 3. Delete session directory (notes, etc.) session_dir = self.base_trace_dir / session_id_str if session_dir.exists() and session_dir.is_dir(): @@ -1222,38 +1247,15 @@ class StorageManager: logger.info(f"Deleted session directory: {session_dir}") except Exception as e: logger.error(f"Failed to delete session directory {session_dir}: {e}") + raise - # 4. Delete video files and their containing folders (which might have been renamed) + # 4. Never let a persisted recording path remove a sibling session. + trace_root = self.base_trace_dir.resolve() for video_path in video_paths: - # video_path is likely: /path/to/traces/task_name/recording.mp4 - # If it was renamed, it might be: /path/to/traces/task_name_PASS_timestamp/recording.mp4 - - task_name = video_path.parent.name - - if task_name and task_name != "traces" and task_name != "..": - # Search for directories starting with task_name in base_trace_dir - try: - for path in self.base_trace_dir.iterdir(): - if path.is_dir() and path.name.startswith(task_name): - try: - shutil.rmtree(path) - logger.info(f"Deleted trace/video directory: {path}") - except Exception as e: - logger.error(f"Failed to delete directory {path}: {e}") - elif path.is_file() and path.name.startswith(task_name): - try: - path.unlink() - logger.info(f"Deleted trace/video file: {path}") - except Exception as e: - logger.error(f"Failed to delete file {path}: {e}") - except Exception as e: - logger.error(f"Error iterating base_trace_dir for cleanup: {e}") - - if video_path.exists(): - try: - video_path.unlink() - logger.info(f"Deleted video file directly: {video_path}") - if video_path.parent.exists() and not any(video_path.parent.iterdir()): - video_path.parent.rmdir() - except Exception as e: - logger.error(f"Failed to delete video file directly {video_path}: {e}") + resolved = video_path.resolve() + if not resolved.is_relative_to(trace_root): + continue + relative = resolved.relative_to(trace_root) + if not relative.parts or relative.parts[0] != session_id_str: + continue + resolved.unlink(missing_ok=True) diff --git a/artemis/runtime/task_control.py b/artemis/runtime/task_control.py index f7a3ee9..f67389d 100644 --- a/artemis/runtime/task_control.py +++ b/artemis/runtime/task_control.py @@ -84,6 +84,11 @@ def list_instructions(session_id: str) -> list[dict]: return sorted(records, key=lambda record: (record.get("timestamp", 0), record["id"])) +def has_pending_instruction(base_dir: str | Path) -> bool: + directory = Path(base_dir) / "instructions" + return any(directory.glob("*.json")) or (Path(base_dir) / "injected_instruction.json").exists() + + def consume_instructions(base_dir: str | Path) -> dict | None: directory = Path(base_dir) / "instructions" found = [] @@ -175,7 +180,7 @@ class TaskControl: async def run_interruptibly[T]( self, operation: Awaitable[T], *, on_pause=None, on_resume=None ) -> tuple[bool, T | None]: - """Park a task during a long model/helper call, then discard its stale result. + """Interrupt a stale model/helper call for pause or fresh user guidance. Only use for operations whose cancellation cannot leave a device action half executed. The caller must return to fresh perception after resume. @@ -189,6 +194,10 @@ class TaskControl: await asyncio.gather(running, return_exceptions=True) await self.checkpoint(on_pause=on_pause, on_resume=on_resume) return True, None + if has_pending_instruction(self.directory): + running.cancel() + await asyncio.gather(running, return_exceptions=True) + return True, None if running.done(): return False, running.result() await asyncio.wait({running}, timeout=0.2) diff --git a/docs/ai-automation.md b/docs/ai-automation.md index 33856b4..700fee3 100644 --- a/docs/ai-automation.md +++ b/docs/ai-automation.md @@ -40,8 +40,10 @@ MCP 同步会识别已安装客户端的配置目录，为缺少 ARTEMIS 条目 创建批次时必须提供唯一 `request_id`；请求超时后可以按编号查询或用相同内容重发，避免重复用例。创建只保存草稿，`start` 才执行。AI 无需先制造 Excel，也可直接提交 JSON；字段与示例见 [Skill 批量任务参考](../skills/artemis-automation/references/batches.md)。 -报告导出到 `artifacts/mcp_exports/`，包含本地路径及服务下载链接。运行中导出是快照；执行完成不等于验证通过，报告会保留未验证、失败、跳过与中断状态。 +报告按任务或批次归档到 `artifacts/mcp_exports/`，包含本地路径及服务下载链接；删除批次时会清理对应导出副本。运行中导出是快照；执行完成不等于验证通过，报告会保留未验证、失败、跳过与中断状态。 -模型默认使用 `mimo-v2.6-flash`，也预设 `gpt-5.6-luna` 并支持自定义。思考强度最高支持 `max`，默认 `high`，选“由模型默认”不发送该参数；接口不支持时明确报错。单次请求 300 秒，首次失败后最多再试 5 次，每次等待 10 秒，SDK 不叠加重试。耗尽后沿用任务暂停/恢复机制，鉴权或非法参数立即失败。视频分析与录屏是独立开关，默认均关闭。每项批量用例可覆盖模式、模型、设备、录屏和超时等配置，不影响其他任务。暂停先返回 `pausing`，确认状态为 `task_paused` 才表示设备已释放；Pro 等待模型回复或只读诊断时可及时中断，恢复后重新观察设备。AI 可用 `mobile_manage_task(action="inject_instruction")` 提交指导并恢复，用 `guidance_history` 核对是否领取，或用 `resume` 单独恢复。 +模型默认使用 `mimo-v2.6-flash`，也预设 `gpt-5.6-luna` 并支持自定义。思考强度最高支持 `max`，默认 `high`，选“由模型默认”不发送该参数；接口不支持时明确报错。单次请求 300 秒，首次失败后最多再试 5 次，每次等待 10 秒，SDK 不叠加重试。耗尽后沿用任务暂停/恢复机制，鉴权或非法参数立即失败。视频分析与录屏是独立开关，默认均关闭。每项批量用例可覆盖模式、模型、设备、录屏和超时等配置，不影响其他任务。暂停先返回 `pausing`，确认状态为 `task_paused` 才表示设备已释放；Pro 等待模型回复或只读诊断时可及时中断，恢复后重新观察设备。AI 可用 `mobile_manage_task(action="inject_instruction")` 提交指导并恢复；运行中的模型思考会中断并重新观察设备，已经开始的设备动作会安全结束。用 `guidance_history` 核对指导是否被领取、在任务流程中核对实际执行情况，或用 `resume` 单独恢复。 + +批次任务在队列中使用“用例编号 + 用例名称”显示，提示词仍是模型实际收到的内容。任务历史展示失败原因、完成时间和总耗时；导出全部资料时以用例编号命名目录。删除非活跃批次会同步清理其任务记录、步骤、截图、录屏及生成的报告，删除后不能通过旧 `request_id` 重复执行。 保存全局配置要求用户明确指令；只读配置不包含密钥，MCP 不接收密钥。密钥仍在网页或 `.env` 中填写。设置对新任务生效，模型请求重试不会触发测试用例自动重跑。历史步骤证据保留，设备动作回放已移除。 diff --git a/docs/automation-workspace.md b/docs/automation-workspace.md index 36836ac..7fa0d5e 100644 --- a/docs/automation-workspace.md +++ b/docs/automation-workspace.md @@ -20,12 +20,14 @@ 4. 上传文件，查看逐行校验结果。所有错误修正后才能提交。 5. 明确选择设备、执行模式和超时时间，然后点击“确认导入并开始执行”。 -同一批次按顺序派发。默认失败后继续且不自动重试，也可以选择失败即停止。设备不可用、模型暂停或任务超时会暂停后续派发。暂停批次只停止后续派发。单个任务可在步骤间安全暂停，释放设备；批次下一项可继续，暂停的任务待提交指导或点击继续后在设备队列末尾恢复。运行中提交指导在下一步执行前生效，不打断当前步骤。暂停等待和重新排队时间不计入该任务执行时限。 +同一批次按顺序派发。默认失败后继续且不自动重试，也可以选择失败即停止。设备不可用、模型暂停或任务超时会暂停后续派发。暂停批次只停止后续派发。单个任务可在安全边界暂停并释放设备；批次下一项可继续，暂停的任务待提交指导或点击继续后在设备队列末尾恢复。运行中提交指导会中断当前模型思考或只读诊断，重新观察画面；已经开始的设备动作会安全结束。网页流程会显示指导的提交与领取状态，仍应检查后续动作是否符合指导。暂停等待和重新排队时间不计入该任务执行时限。 服务重启后保留批次，等待用户继续未开始的用例。正在执行或暂停的任务标记中断，不自动重做；“复制失败项为新批次”保留原记录并产生新任务编号。 每个任务结束后生成所选格式报告。批次结束后生成汇总及每项任务报告的 ZIP；失败、跳过和中断任务也会列入。暂停时下载的是当前快照。录屏与 AI 视频分析分别配置，录屏默认关闭；仅开启分析时，录制内容用于本次分析，结束后不保留下载文件。暂停期间不录制其他任务的设备画面。 +队列中批量任务按“用例编号 + 测试用例”显示，悬停可看完整提示词；失败原因、完成时间及总耗时在历史中可查看。单任务“下载全部资料”以用例编号命名压缩包内目录。删除已结束的批次会同步删除关联任务记录、步骤、截图、录屏、报告和 MCP 导出副本；正在执行或排队的批次不能删除。 + ## AI 使用 通过 MCP 可直接提交结构化批次、检索历史和导出报告；见 [AI 自动化与 Skill](ai-automation.md)。操作回放已移除，旧记录仅供历史查看，不能继续或重新执行。 @@ -34,4 +36,4 @@ `.env` 可配置 `ARTEMIS_BATCH_TASK_TIMEOUT`（默认 1800 秒）、`ARTEMIS_IMPORT_MAX_ROWS`（默认 1000 行）和 `ARTEMIS_IMPORT_MAX_BYTES`（默认 10 MiB）。修改后重启服务。网页的单批次超时设置优先于环境默认值。 -批次、任务元数据和报告偏好保存在现有 SQLite 数据库的 `automation_records`、`automation_jobs` 表。清空任务历史同时清除批次与执行元数据；已导出的文件保留。 +批次、任务元数据和报告偏好保存在现有 SQLite 数据库的 `automation_records`、`automation_jobs` 表。全局“清空任务历史”会清除批次与执行元数据；它与上文的单批次删除不同，已导出的文件会保留。 diff --git a/docs/releases/1.2.0.md b/docs/releases/1.2.0.md new file mode 100644 index 0000000..0c6c8da --- /dev/null +++ b/docs/releases/1.2.0.md @@ -0,0 +1,14 @@ +# ARTEMIS-Eapil 1.2.0 + +## 工作区改进 + +- 批量任务在队列和历史中显示“用例编号 + 测试用例”；悬停标题可查看完整提示词。任务状态、完成时间、总耗时和失败原因更易辨认。 +- 超时失败会写明原因和批次时限。单任务资料 ZIP 以用例编号命名目录。 +- 删除已结束的批次会清理关联任务、步骤、截图、录屏和报告，包括 MCP 导出副本；仍被其他任务引用的截图会保留。正在执行或排队的批次不能删除。 +- 网页或 MCP 提交指导时，中断当前模型思考并重新观察设备。已开始的设备动作安全结束；流程显示指导提交、领取和后续动作。 + +## 安装 + +下载附件 `artemis-eapil-1.2.0-source.zip`，解压后运行 `start.bat`；国内网络可运行 `start-cn.bat`。附件包含精简 Git 仓库，可供后续启动脚本检查更新。Gitea 自动生成的源码 ZIP 不含 `.git`，请使用发布附件。 + +升级后刷新网页，并在 AI 客户端重新加载 MCP。验证：Python 默认测试 2177 项通过、6 项跳过；前端 149 项通过且生产构建成功；Ruff、质量基线和核心类型检查通过。 diff --git a/mcp_server/tools/automation.py b/mcp_server/tools/automation.py index 5e5fdcb..e887916 100644 --- a/mcp_server/tools/automation.py +++ b/mcp_server/tools/automation.py @@ -67,10 +67,16 @@ async def _request(method: str, route: str, **kwargs) -> Any: return response.json() -async def _download(route: str, filename: str) -> dict: +async def _download( + route: str, + filename: str, + *, + owner_kind: Literal["sessions", "batches"] | None = None, + owner_id: str | None = None, +) -> dict: url = await _base_url() - EXPORT_DIR.mkdir(parents=True, exist_ok=True) - path = EXPORT_DIR / f"{uuid4().hex[:12]}-{filename}" + directory = EXPORT_DIR / owner_kind / _id(owner_id) if owner_kind else EXPORT_DIR + path = directory / f"{uuid4().hex[:12]}-{filename}" partial = path.with_suffix(path.suffix + ".part") try: async with httpx.AsyncClient(base_url=url, timeout=120, trust_env=False) as client: @@ -78,6 +84,7 @@ async def _download(route: str, filename: str) -> dict: if response.is_error: await response.aread() _check(response) + directory.mkdir(parents=True, exist_ok=True) with partial.open("wb") as stream: async for chunk in response.aiter_bytes(): stream.write(chunk) @@ -198,8 +205,8 @@ async def mobile_batch( use mobile_manage_task(stop) to stop that task. resume never repeats finished cases. retry_failed creates a new draft only after an explicit rerun request; it is not transport-idempotent, so after an uncertain response use list instead of retrying. - delete removes a non-active batch record after an explicit user request; it keeps - already executed task sessions and reports. Confirm the batch ID with get first. + delete removes a non-active batch and its task sessions, reports and files after + an explicit user request. Confirm the batch ID with get first; deletion is irreversible. Restarted batches remain paused. No automatic failed-case retries or action replay. """ if action == "create": @@ -270,7 +277,12 @@ async def mobile_export_report( return { "batch_id": bid, "files": [ - await _download(f"/api/automation/batches/{bid}/archive", f"batch-{bid}.zip") + await _download( + f"/api/automation/batches/{bid}/archive", + f"batch-{bid}.zip", + owner_kind="batches", + owner_id=bid, + ) ], } sid = _id(session_id) @@ -286,10 +298,15 @@ async def mobile_export_report( ): raise ValueError("formats 需选择 docx、xlsx、screenshots、video 或 all") report = await _request("GET", f"/api/automation/reports/{sid}") + from apps.admin_console.services.report_export import archive_case_name + + label = archive_case_name(report) files = [ await _download( f"/api/automation/reports/{sid}/{fmt}", - f"task-{sid}.{fmt}" if fmt in {"docx", "xlsx"} else f"task-{sid}-{fmt}.zip", + f"task-{sid}.{fmt}" if fmt in {"docx", "xlsx"} else f"{label}-{fmt}.zip", + owner_kind="sessions", + owner_id=sid, ) for fmt in dict.fromkeys(chosen) ] diff --git a/mcp_server/tools/task_manager.py b/mcp_server/tools/task_manager.py index 1c8284c..faa08fc 100644 --- a/mcp_server/tools/task_manager.py +++ b/mcp_server/tools/task_manager.py @@ -282,8 +282,10 @@ def mobile_manage_task( - **'resume'**: Queues a paused task for its device again; optional `instruction` provides guidance before the next action. - **'inject_instruction'**: Injects real-time guidance mid-flight when the - subagent errs, stalls, or loops. Applied at the start of the next - planning turn (Pro) or the next reactive loop (Flash). Requires + subagent errs, stalls, or loops. An in-flight model response or read-only + helper is cancelled immediately, then the runner observes the device again + and consumes the new guidance. An already-started device action finishes + safely before the runner changes direction. Requires `instruction` (unless `release_loop=True`). To end a continuous monitoring task ([Loop:continuous] milestone) gracefully, you MUST pass `release_loop=True` — this is the only signal that unlocks the @@ -697,7 +699,7 @@ def mobile_manage_task( return { "trace_id": trace_id, "status": current_status, - "message": f"Successfully injected instruction{suffix}: '{instruction}'", + "message": f"Guidance accepted and current model step interrupted{suffix}; check guidance_history for consumption.", } except Exception as e: return { diff --git a/pyproject.toml b/pyproject.toml index 7b5d99a..1f375ba 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,7 @@ [project] name = "artemis" -version = "1.1.2" +version = "1.2.0" description = "Autonomous AI Assistant for Mobile Automation" authors = [ diff --git a/skills/artemis-automation/SKILL.md b/skills/artemis-automation/SKILL.md index 2e698cf..392fcba 100644 --- a/skills/artemis-automation/SKILL.md +++ b/skills/artemis-automation/SKILL.md @@ -14,7 +14,7 @@ description: 使用 ARTEMIS MCP 在 Android 真机或模拟器上完成自然语 3. 首次使用、工具出错、任务拒绝或设备异常，先 `mobile_diagnose`。按 `next_steps` 的依赖顺序处理；`attempt_fix=true` 可尝试工具自带修复。只有凭据相关故障才使用 `verify_credentials=true`，观察异常才使用 `probe_device=true`。不要盲目重发任务。MCP 不可用时在项目目录运行 `uv run artemis doctor` 并检查宿主 MCP 日志。 4. 选择执行模式：Flash 适合路径明确的界面操作；Pro 适合动态分支、持续计划、ADB/日志诊断或需要 Checker 明确校验的任务。**`mobile_run_task.model` 的 `flash`/`pro` 是执行模式，不是 API 模型名称**。步数长不自动等于需要 Pro；视频能力只有显式启用后才可用。 5. 单任务用 `mobile_run_task`；批量任务用 `mobile_batch`，读 [批量任务与交付](references/batches.md)。AI 可直接提交结构化用例，不必先创建 Excel。用户提供 Excel 时先 `mobile_task_file(action="preview", path=绝对路径)`，修正所有行错误后提交。提示词列原样作为目标，前置约束列只作元数据。 -6. 单任务返回 `trace_id` 后，用 `mobile_manage_task(action="status", trace_id=...)` 跟踪；网页任务也可用其任务编号。批次用 `mobile_batch(action="get", batch_id=...)`。需要人工补充时用 `pause`，其返回的 `pausing` 只表示请求已接收；轮询到 `task_paused` 才表示设备已释放。Pro 等待模型回复或只读诊断时可中断，已开始的设备动作会先安全结束。`inject_instruction` 可在运行中指导下一步，暂停中提交则恢复；`guidance_history` 的 `pending/consumed` 区分提交和执行端领取，无新指导用 `resume`。恢复进入设备 FIFO 队尾并重新观察手机状态。尊重工具建议的等待周期，避免连续高频轮询；停滞或失败先诊断，不把同一业务任务自动重提。 +6. 单任务返回 `trace_id` 后，用 `mobile_manage_task(action="status", trace_id=...)` 跟踪；网页任务也可用其任务编号。批次用 `mobile_batch(action="get", batch_id=...)`。需要人工补充时用 `pause`，其返回的 `pausing` 只表示请求已接收；轮询到 `task_paused` 才表示设备已释放。`inject_instruction` 会立即取消进行中的模型回复或只读诊断并重新观察手机，已开始的设备动作会先安全结束；暂停中提交则恢复。`guidance_history` 的 `pending/consumed` 区分提交和执行端领取，无新指导用 `resume`。恢复进入设备 FIFO 队尾并重新观察手机状态。尊重工具建议的等待周期，避免连续高频轮询；停滞或失败先诊断，不把同一业务任务自动重提。 ## 结果、证据与交付 @@ -32,6 +32,6 @@ description: 使用 ARTEMIS MCP 在 Android 真机或模拟器上完成自然语 - API Key 只通过网页或 `.env` 配置。不要请求用户把 key 发到聊天中，不读取/回显已有 key。保存后新任务生效；直接编辑环境文件后需重启相应服务，运行中的任务保留自己的配置。 - 录屏与 AI 视频分析分别配置。批量任务的 `task_config` 可覆盖单项，不修改全局设置；只开视频分析时录制内容不作为可下载文件保存。暂停时停止录屏，恢复后按片段继续，不能把其他任务画面混入该任务资料。 - 请求重试是模型调用恢复，不是测试用例重跑。批次默认失败继续、不自动重跑；服务重启后保留状态并暂停，用户明确要求后才继续或复制失败项。 -- 批次暂停只停止后续派发；当前任务等待指导时用 `mobile_manage_task(action="pause", trace_id=...)`，真正取消用 `stop`。用户明确要求删除批次记录时先 `mobile_batch(action="get")` 核对编号，再用 `delete`；执行中/排队中的批次不能删，已执行的任务历史和报告会保留。服务重启后运行/暂停的任务标记中断，不自动重跑。停止可能保留已经产生的业务效果，不声称自动回滚。 +- 批次暂停只停止后续派发；当前任务等待指导时用 `mobile_manage_task(action="pause", trace_id=...)`，真正取消用 `stop`。用户明确要求删除批次记录时先 `mobile_batch(action="get")` 核对编号，再用 `delete`；执行中/排队中的批次不能删，删除会清除该批次的任务历史、报告和文件。服务重启后运行/暂停的任务标记中断，不自动重跑。停止可能保留已经产生的业务效果，不声称自动回滚。 需要编写外部可执行测试代码时，先通过 ARTEMIS 实际探索目标应用，确认页面、定位器和转换。优先动态定位，框架支持时才以已验证坐标兜底；在最终脚本中明确等待条件，不将 AI 思考耗时当作确定性等待。 diff --git a/skills/artemis-automation/references/batches.md b/skills/artemis-automation/references/batches.md index 52973f4..c91d420 100644 --- a/skills/artemis-automation/references/batches.md +++ b/skills/artemis-automation/references/batches.md @@ -56,7 +56,7 @@ - 单项 `pending/queued/running/pausing/task_paused/resuming/finalizing` 尚未结束，`completed/failed/cancelled/interrupted/skipped` 是终态。`pausing` 是请求已发出而设备仍被占用，`task_paused` 才表示设备已释放；`finalizing` 在生成报告，不能提前宣称报告已完成。 - 批次 `pause` 保留当前执行，只暂停后续派发；单项 `mobile_manage_task(pause)` 会在安全边界释放设备，Pro 的模型等待与只读诊断可中断。恢复后重新观察画面，并排到设备队尾，不阻塞后续项。`mobile_manage_task(guidance_history)` 可核对指导是否从 `pending` 变为 `consumed`，但仍需检查后续操作是否符合指导。设备断开、模型持续失败或任务达到批次时限会暂停后续派发；用诊断确认恢复后再响应用户的继续指令。 - `resume` 只继续未开始项；`retry_failed` 复制失败、取消和中断项为新草稿，不修改原结果。必须来自用户的明确重跑请求。该动作不提供创建去重，响应不确定时先 `list` 查找新批次，不自动重复调用。 -- 用户明确要求删除批次记录时，先 `get` 核对批次编号与状态，再 `delete`；正在执行、排队或等待指导的批次不能删除，已经运行过的任务历史及报告不会随批次记录一起消失。 +- 用户明确要求删除批次记录时，先 `get` 核对批次编号与状态，再 `delete`；正在执行、排队或等待指导的批次不能删除。删除会同时移除该批次任务历史、报告和文件，无法撤销。 ## 报告 diff --git a/tests/unit/admin_console/test_automation.py b/tests/unit/admin_console/test_automation.py index a45a752..3e7e415 100644 --- a/tests/unit/admin_console/test_automation.py +++ b/tests/unit/admin_console/test_automation.py @@ -2,8 +2,10 @@ from io import BytesIO import json +import sqlite3 import time from unittest.mock import AsyncMock +from uuid import UUID import zipfile from docx import Document @@ -125,19 +127,44 @@ def test_preferences_and_required_device(client): assert client.post("/api/automation/batches", json=payload(device_serial="")).status_code == 422 -def test_batch_delete_retains_executed_task_and_prevents_request_replay(client, isolated): +def test_batch_delete_removes_tasks_and_files_and_prevents_request_replay(client, isolated): created = client.post( "/api/automation/batches", json=payload(request_id="delete-example") ).json() first, second = isolated.jobs(created["id"]) isolated.save_job({**first, "status": "completed"}) + trace_dir = reports.TRACES_PATH / first["session_id"] + trace_dir.mkdir(parents=True) + (trace_dir / "result.json").write_text("{}", encoding="utf-8") + report_dir = reports.EXPORT_ROOT / "sessions" / first["session_id"] + report_dir.mkdir(parents=True) + (report_dir / "任务报告.docx").write_bytes(b"report") + batch_dir = reports.EXPORT_ROOT / "batches" / created["id"] + batch_dir.mkdir(parents=True) + (batch_dir / "批次报告.zip").write_bytes(b"zip") + mcp_task_dir = reports.EXPORT_ROOT.parent / "mcp_exports" / "sessions" / first["session_id"] + mcp_task_dir.mkdir(parents=True) + (mcp_task_dir / "报告.docx").write_bytes(b"report") + mcp_batch_dir = reports.EXPORT_ROOT.parent / "mcp_exports" / "batches" / created["id"] + mcp_batch_dir.mkdir(parents=True) + (mcp_batch_dir / "批次.zip").write_bytes(b"zip") + unrelated = reports.TRACES_PATH / "unrelated-task" + unrelated.mkdir() + (unrelated / "keep.txt").write_text("keep", encoding="utf-8") response = client.delete(f"/api/automation/batches/{created['id']}") assert response.status_code == 200 assert client.get(f"/api/automation/batches/{created['id']}").status_code == 404 - assert isolated.job(first["session_id"])["case"]["name"] == "用例甲" + assert isolated.job(first["session_id"]) is None assert isolated.job(second["session_id"]) is None - assert reports.session_report(first["session_id"])["name"] == "用例甲" + assert not trace_dir.exists() + assert not report_dir.exists() + assert not batch_dir.exists() + assert not mcp_task_dir.exists() + assert not mcp_batch_dir.exists() + assert (unrelated / "keep.txt").read_text(encoding="utf-8") == "keep" + with pytest.raises(ValueError, match="不存在"): + reports.session_report(first["session_id"]) assert ( client.post( "/api/automation/batches", json=payload(request_id="delete-example") @@ -158,6 +185,49 @@ def test_batch_delete_rejects_active_jobs(client, isolated, status): assert len(isolated.jobs(batch["id"])) == 2 +def test_batch_deletion_reservation_blocks_new_dispatch(isolated): + batch = service.create_batch(payload()) + job = isolated.jobs(batch["id"])[0] + + isolated.begin_delete_batch(batch["id"]) + + with pytest.raises(ValueError, match="正在删除"): + isolated.save_job({**job, "status": "queued"}) + with pytest.raises(ValueError, match="正在删除"): + isolated.put("batch", batch["id"], {**batch, "status": "running"}) + assert isolated.jobs_for_sessions([job["session_id"]])[job["session_id"]]["status"] == "pending" + + +def test_batch_deletion_preserves_shared_screenshot_until_last_task(client, isolated): + first_batch = service.create_batch(payload()) + second_batch = service.create_batch(payload(name="另一批次")) + first = isolated.jobs(first_batch["id"])[0] + other = isolated.jobs(second_batch["id"])[0] + shared_hash = "a" * 64 + storage = StorageManager(isolated.path, reports.TRACES_PATH) + image_file = reports.TRACES_PATH / "images" / f"{shared_hash}.jpg" + image_file.parent.mkdir(parents=True) + image_file.write_bytes(b"shared image") + with sqlite3.connect(isolated.path) as conn: + conn.execute("INSERT INTO images(image_name) VALUES (?)", (shared_hash,)) + for job in (first, other): + conn.execute( + "INSERT INTO sessions(session_id,initial_goal,start_time,status,device_info) " + "VALUES (?,?,?,?,?)", + (job["session_id"], "观察设置", 1.0, "completed", "{}"), + ) + conn.execute( + "INSERT INTO steps(step_id,session_id,pre_image_name) VALUES (?,?,?)", + (job["session_id"], job["session_id"], shared_hash), + ) + + assert client.delete(f"/api/automation/batches/{first_batch['id']}").status_code == 200 + assert image_file.read_bytes() == b"shared image" + assert storage.get_session(UUID(other["session_id"])) is not None + assert client.delete(f"/api/automation/batches/{second_batch['id']}").status_code == 200 + assert not image_file.exists() + + def test_restart_never_repeats_started_work(isolated): batch = service.create_batch(payload()) service.batch_action(batch["id"], "resume") @@ -260,6 +330,41 @@ async def test_report_failure_does_not_strand_job(isolated, monkeypatch): assert "disk full" in saved["report_error"] +@pytest.mark.asyncio +async def test_batch_timeout_stays_visible_in_job_report_and_history(client, isolated): + job = isolated.jobs(service.create_batch(payload())["id"])[0] + isolated.save_job( + {**job, "status": "running", "timed_out": True, "error": "任务执行超时（30 秒）"} + ) + await service.on_terminal({"session_id": job["session_id"]}, "cancelled") + saved = isolated.job(job["session_id"]) + assert saved["status"] == "failed" + assert "超时" in saved["error"] + assert "超时" in reports.session_report(job["session_id"])["error"] + history = client.get("/api/automation/history").json()["items"] + assert any( + item["session_id"] == job["session_id"] and "超时" in item["error"] for item in history + ) + + +@pytest.mark.asyncio +async def test_model_timeout_from_runner_log_reaches_failed_task_and_report(isolated): + job = isolated.jobs(service.create_batch(payload())["id"])[0] + folder = reports.TRACES_PATH / job["session_id"] + folder.mkdir(parents=True) + (folder / "stderr.log").write_text( + "TimeoutError: model request timed out after 300 seconds\n", encoding="utf-8" + ) + + await service.on_terminal({"session_id": job["session_id"]}, "failed") + + saved = isolated.job(job["session_id"]) + assert saved["status"] == "failed" + assert "超时" in saved["error"] + assert reports.session_report(job["session_id"])["status"] == "failed" + assert "超时" in reports.session_report(job["session_id"])["error"] + + def test_reports_do_not_execute_formulas_or_invent_pass(isolated): batch = service.create_batch(payload()) job = isolated.jobs(batch["id"])[0] @@ -296,6 +401,7 @@ def test_failed_report_exposes_logged_reason_and_formats_chinese_markdown(isolat assert "abc123secret" not in reports.failure_reason( folder, "failed", "认证失败 OPENAI_API_KEY=abc123secret Authorization: Bearer abc123secret" ) + assert "超时" not in reports.failure_reason(folder, "failed", "LLMExhaustedError: 额度不足") paths = reports.export_session(job["session_id"]) doc = Document(next(path for path in paths if path.suffix == ".docx")) assert doc.styles["Normal"].font.name == "Microsoft YaHei" @@ -325,6 +431,7 @@ def test_task_material_zip_contains_ordered_evidence_and_reports( client, isolated, tmp_path, monkeypatch ): job = isolated.jobs(service.create_batch(payload())["id"])[0] + isolated.save_job({**job, "case": {**job["case"], "case_id": "1.1"}}) evidence = tmp_path / "evidence" evidence.mkdir() first = evidence / "first.png" @@ -343,14 +450,16 @@ def test_task_material_zip_contains_ordered_evidence_and_reports( assert response.status_code == 200 with zipfile.ZipFile(BytesIO(response.content)) as archive: files = archive.namelist() + assert all(name.startswith("1.1/") for name in files) assert any(name.endswith(".docx") for name in files) assert any(name.endswith(".xlsx") for name in files) - assert files.index("步骤截图/0001_步骤0001_操作前.png") < files.index( - "步骤截图/0002_步骤0002_操作后.png" + assert files.index("1.1/步骤截图/0001_步骤0001_操作前.png") < files.index( + "1.1/步骤截图/0002_步骤0002_操作后.png" ) - assert archive.read("任务录屏/0001_recording.mp4") == b"recording.mp4" - manifest = json.loads(archive.read("文件清单.json")) + assert archive.read("1.1/任务录屏/0001_recording.mp4") == b"recording.mp4" + manifest = json.loads(archive.read("1.1/文件清单.json")) assert manifest["session_id"] == job["session_id"] + assert "1.1-all.zip" in response.headers["content-disposition"] def test_history_filters_all_persisted_rows_and_escapes_like(client, isolated): diff --git a/tests/unit/admin_console/test_session_list_batching.py b/tests/unit/admin_console/test_session_list_batching.py index 2126af6..b23503a 100644 --- a/tests/unit/admin_console/test_session_list_batching.py +++ b/tests/unit/admin_console/test_session_list_batching.py @@ -5,6 +5,14 @@ import pytest from apps.admin_console.routers import sessions as sessions_router +@pytest.fixture(autouse=True) +def isolate_jobs(monkeypatch): + mocked = MagicMock() + mocked.jobs_for_sessions.return_value = {} + monkeypatch.setattr(sessions_router, "store", mocked) + return mocked + + @pytest.mark.asyncio async def test_list_sessions_uses_batched_history_metadata(monkeypatch): repo = MagicMock() @@ -96,3 +104,42 @@ async def test_list_sessions_extracts_device_serial(monkeypatch): assert result[0]["device_id"] == "63191FDKX00062" assert result[1]["device_serial"] == "emulator-5554" assert result[1]["device_id"] == "emulator-5554" + + +@pytest.mark.asyncio +async def test_list_sessions_uses_batch_title_and_timeout_error(monkeypatch, isolate_jobs): + repo = MagicMock() + repo.get_all_sessions.return_value = [ + { + "session_id": "session-case", + "status": "completed", + "start_time": 1.0, + "initial_goal": "打开设置并检查页面", + "device_info": '{"profile":"flash"}', + } + ] + repo.get_video_recordings_map.return_value = {} + repo.get_latest_video_recordings_map.return_value = {} + repo.get_agent_trace_names_map.return_value = {} + monkeypatch.setattr(sessions_router, "session_repo", repo) + monkeypatch.setattr( + sessions_router.media_service, "build_video_index", MagicMock(return_value={}) + ) + isolate_jobs.jobs_for_sessions.return_value = { + "session-case": { + "session_id": "session-case", + "status": "failed", + "ended_at": 65.0, + "error": "任务执行超时（60 秒）", + "case": {"case_id": "1.1", "name": "打开并核对设置", "goal": "打开设置并检查页面"}, + } + } + + result = await sessions_router.list_sessions() + + isolate_jobs.jobs_for_sessions.assert_called_once_with(["session-case"]) + assert result[0]["display_name"] == "1.1 打开并核对设置" + assert result[0]["initial_goal"] == "打开设置并检查页面" + assert result[0]["status"] == "failed" + assert result[0]["end_time"] == 65.0 + assert "超时" in result[0]["error"] diff --git a/tests/unit/mcp/test_automation_tools.py b/tests/unit/mcp/test_automation_tools.py index 1ba7609..bcda3de 100644 --- a/tests/unit/mcp/test_automation_tools.py +++ b/tests/unit/mcp/test_automation_tools.py @@ -80,13 +80,19 @@ async def test_download_writes_actual_binary_atomically(monkeypatch, tmp_path): monkeypatch.setattr( tools.httpx, "AsyncClient", lambda **kw: client_type(transport=transport, **kw) ) - result = await tools._download("/api/automation/reports/test/docx", "test.docx") + result = await tools._download( + "/api/automation/reports/test/docx", + "test.docx", + owner_kind="sessions", + owner_id="test", + ) from pathlib import Path assert Path(result["path"]).read_bytes() == b"PK\x03\x04report-bytes" assert Path(result["path"]).is_absolute() + assert Path(result["path"]).parent == tmp_path / "sessions" / "test" assert result["download_url"] == "http://test/api/automation/reports/test/docx" - assert not list(tmp_path.glob("*.part")) + assert not list(tmp_path.rglob("*.part")) @pytest.mark.asyncio diff --git a/tests/unit/mcp/test_mcp_tools.py b/tests/unit/mcp/test_mcp_tools.py index 4e2cbdf..c3c3085 100644 --- a/tests/unit/mcp/test_mcp_tools.py +++ b/tests/unit/mcp/test_mcp_tools.py @@ -373,7 +373,7 @@ def test_mobile_manage_task_inject_instruction(temp_trace_env): ok_res = mobile_manage_task( action="inject_instruction", trace_id=trace_id, instruction="Scroll down slowly" ) - assert "Successfully injected" in ok_res["message"] + assert "current model step interrupted" in ok_res["message"] @pytest.mark.asyncio diff --git a/tests/unit/runtime/test_task_control.py b/tests/unit/runtime/test_task_control.py index 1c22afe..107c6c9 100644 --- a/tests/unit/runtime/test_task_control.py +++ b/tests/unit/runtime/test_task_control.py @@ -116,6 +116,35 @@ async def test_interruptible_call_returns_normal_result_without_pause(tmp_path, assert await control.run_interruptibly(model_call()) == (False, "current result") +@pytest.mark.asyncio +async def test_new_guidance_cancels_stale_model_without_releasing_device(tmp_path, monkeypatch): + monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path) + released = [] + + class Lock: + def release(self): + released.append(True) + + control = task_control.TaskControl("task-123", Lock()) + started = asyncio.Event() + cancelled = asyncio.Event() + + async def stale_model(): + started.set() + try: + await asyncio.Event().wait() + finally: + cancelled.set() + + call = asyncio.create_task(control.run_interruptibly(stale_model())) + await asyncio.wait_for(started.wait(), 1) + task_control.add_instruction("task-123", "改为检查当前页面", source="mcp") + assert await asyncio.wait_for(call, 1) == (True, None) + assert cancelled.is_set() + assert released == [] + assert task_control.consume_instructions(control.directory)["instruction"] == "改为检查当前页面" + + def test_guidance_history_distinguishes_submission_from_runner_consumption(tmp_path, monkeypatch): monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path) first = task_control.add_instruction("task-123", "从快捷设置进入", source="web") diff --git a/uv.lock b/uv.lock index f379fa3..c36ae77 100644 --- a/uv.lock +++ b/uv.lock @@ -206,7 +206,7 @@ wheels = [ [[package]] name = "artemis" -version = "1.1.2" +version = "1.2.0" source = { editable = "." } dependencies = [ { name = "adbutils" },