diff --git a/apps/admin_console/routers/automation.py b/apps/admin_console/routers/automation.py index d574c87..945d5a6 100644 --- a/apps/admin_console/routers/automation.py +++ b/apps/admin_console/routers/automation.py @@ -182,11 +182,7 @@ async def batch_archive(batch_id: str): if archive_service.status(batch_id, batch)["status"] == "preparing": raise HTTPException(409, "报告正在打包,请等待准备完成") path = archive_service.archive_path(batch_id) - if ( - not path.is_file() - or not batch.get("report_ready") - and archive_service.status(batch_id, batch)["status"] != "ready" - ): + if not path.is_file() or archive_service.status(batch_id, batch)["status"] != "ready": path = await asyncio.to_thread(report_export.export_batch, batch_id) return FileResponse(path, filename=f"批次报告-{batch_id[:8]}.zip") @@ -205,8 +201,7 @@ async def download_report( ): try: if fmt in {"screenshots", "video", "all"}: - path = await asyncio.to_thread(report_export.export_session_media, session_id, fmt) - label = report_export.archive_case_name(report_export.session_report(session_id)) + path, label = await asyncio.to_thread(report_export.session_download, session_id, fmt) 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}") @@ -224,7 +219,9 @@ def review_case(session_id: str, value: CaseReviewRequest): and archive_service.status(batch_id, require("batch", batch_id))["status"] == "preparing" ): raise HTTPException(409, "批次报告正在打包,请完成后再保存人工复验") - with report_export.EXPORT_LOCK: + with report_export.export_lock( + *(["batch:" + batch_id] if batch_id else []), "session:" + session_id + ): current = report(session_id) if current["status"] not in { "completed", diff --git a/apps/admin_console/routers/favorites.py b/apps/admin_console/routers/favorites.py new file mode 100644 index 0000000..02954ed --- /dev/null +++ b/apps/admin_console/routers/favorites.py @@ -0,0 +1,50 @@ +"""Favorite templates shared by the home page and MCP.""" + +from fastapi import APIRouter, HTTPException + +from apps.admin_console.services.favorites import ( + FavoriteCreate, + FavoriteRun, + FavoriteUpdate, + favorites, +) + +router = APIRouter(prefix="/api/automation/favorites", tags=["automation"]) + + +@router.get("") +def list_favorites(): + return favorites.store.records("favorite", 10000) + + +@router.post("") +def create_favorite(value: FavoriteCreate): + try: + return favorites.create(value) + except (KeyError, ValueError) as exc: + raise HTTPException(409, str(exc)) from exc + + +@router.patch("/{key}") +def update_favorite(key: str, value: FavoriteUpdate): + try: + return favorites.update(key, value) + except (KeyError, ValueError) as exc: + raise HTTPException(409, str(exc)) from exc + + +@router.delete("/{key}") +def delete_favorite(key: str): + try: + favorites.delete(key) + except ValueError as exc: + raise HTTPException(400, str(exc)) from exc + return {"status": "deleted"} + + +@router.post("/{key}/run") +async def run_favorite(key: str, value: FavoriteRun): + try: + return await favorites.run(key, value) + except (KeyError, ValueError) as exc: + raise HTTPException(409, str(exc)) from exc diff --git a/apps/admin_console/server.py b/apps/admin_console/server.py index 0923c08..862ba6c 100644 --- a/apps/admin_console/server.py +++ b/apps/admin_console/server.py @@ -277,6 +277,9 @@ app.include_router(system.router) from apps.admin_console.routers.automation import router as automation_router app.include_router(automation_router) +from apps.admin_console.routers.favorites import router as favorites_router + +app.include_router(favorites_router) from apps.admin_console.routers.plugins import router as plugins_router diff --git a/apps/admin_console/services/automation_service.py b/apps/admin_console/services/automation_service.py index dc5d84c..f328e72 100644 --- a/apps/admin_console/services/automation_service.py +++ b/apps/admin_console/services/automation_service.py @@ -36,6 +36,22 @@ def register_submission(item: dict, case: dict | None = None, formats=None, **ex "status": "queued", "created_at": time.time(), "device_serial": item.get("device_serial"), + "execution_config": item.get("execution_config"), + "submission": { + key: item.get(key) + for key in ( + "goal", + "profile", + "device_serial", + "runtime_overrides", + "expected_output", + "enable_outputter", + "verification_level", + "explorer_mode", + "locked_app_package", + "app_path", + ) + }, "formats": formats if formats is not None else store.preferences()["formats"], **extra, } @@ -391,7 +407,9 @@ def delete_batch(batch_id: str) -> bool: return False storage = StorageManager(store.path, report_export.TRACES_PATH) mcp_exports = report_export.EXPORT_ROOT.parent / "mcp_exports" - with report_export.EXPORT_LOCK: + with report_export.export_lock( + "batch:" + batch_id, *("session:" + j["session_id"] for j in jobs) + ): for job in jobs: session_id = str(UUID(job["session_id"])) storage.delete_session(UUID(session_id)) diff --git a/apps/admin_console/services/batch_archive.py b/apps/admin_console/services/batch_archive.py index 3a0e1f2..1aaa9f2 100644 --- a/apps/admin_console/services/batch_archive.py +++ b/apps/admin_console/services/batch_archive.py @@ -13,6 +13,12 @@ def archive_path(batch_id: str): return report_export.EXPORT_ROOT / "batches" / batch_id / "批次报告.zip" +def cached(batch_id: str, batch: dict) -> bool: + return bool(batch.get("report_ready")) and report_export.reusable( + archive_path(batch_id), report_export.BATCH_ARCHIVE_SIGNATURE, "completed" + ) + + def _snapshot(state: dict) -> dict: elapsed = max(0, (state.get("finished_at") or time.monotonic()) - state["started_at"]) completed, total = state["completed"], state["total"] @@ -37,7 +43,7 @@ def status(batch_id: str, batch: dict) -> dict: state = _states.get(batch_id) if state: return _snapshot(state) - if batch.get("report_ready") and archive_path(batch_id).is_file(): + if cached(batch_id, batch): return { "status": "ready", "stage": "报告包已就绪", @@ -63,7 +69,7 @@ def prepare(batch_id: str, batch: dict) -> dict: current = _states.get(batch_id) if current and current["status"] == "preparing": return _snapshot(current) - if batch.get("report_ready") and archive_path(batch_id).is_file(): + if cached(batch_id, batch): _states.pop(batch_id, None) return status_from_cache() state = { @@ -81,9 +87,9 @@ def prepare(batch_id: str, batch: dict) -> dict: def run() -> None: try: - with report_export.EXPORT_LOCK: + with report_export.export_lock("batch:" + batch_id): latest = report_export.store.get("batch", batch_id) - if not (latest and latest.get("report_ready") and archive_path(batch_id).is_file()): + if not (latest and cached(batch_id, latest)): report_export.export_batch(batch_id, progress=update) with _lock: state.update(status="ready", stage="报告包已就绪", finished_at=time.monotonic()) diff --git a/apps/admin_console/services/evidence_service.py b/apps/admin_console/services/evidence_service.py index 47d8830..a8f8230 100644 --- a/apps/admin_console/services/evidence_service.py +++ b/apps/admin_console/services/evidence_service.py @@ -105,14 +105,16 @@ def plugin_files(session_id: str) -> list[tuple[Path, str]]: return list(files.items()) -def export_index(session_id: str, report: dict, android_videos: list) -> dict: +def export_index( + session_id: str, report: dict, android_videos: list, *, images=None, artifacts=None, extras=None +) -> dict: from apps.admin_console.services.report_export import session_artifacts, session_screenshots from apps.admin_console.services.exploration_evidence import ( exploration_index, demonstration_sources, ) - extras = plugin_files(session_id) + extras = plugin_files(session_id) if extras is None else extras recordings = plugin_sources(session_id) + demonstration_sources(session_id) root = task_root(session_id) paths = {path.resolve(): name for path, name in extras} @@ -120,10 +122,10 @@ def export_index(session_id: str, report: dict, android_videos: list) -> dict: for segment in source["segments"]: segment["file"] = paths.get((root / segment["file"]).resolve()) segment.pop("url", None) - images = session_screenshots(report) - artifacts = session_artifacts(report) + images = session_screenshots(report) if images is None else images + artifacts = session_artifacts(report) if artifacts is None else artifacts files = [(path, "步骤截图/" + name) for path, name in images] - files += [(path, "任务录屏/" + name) for path, name in android_videos] + files += [(path, "任务录屏/安卓/" + name) for path, name in android_videos] files += [(path, "任务文件/安卓/" + name) for path, name in artifacts] files += extras calls = [] @@ -151,7 +153,7 @@ def export_index(session_id: str, report: dict, android_videos: list) -> dict: "steps": report.get("steps", []), "calls": calls, "recordings": recordings, - "android_videos": ["任务录屏/" + name for _, name in android_videos], + "android_videos": ["任务录屏/安卓/" + name for _, name in android_videos], "image_references": {path.stem: "步骤截图/" + name for path, name in images}, "files": [{"path": name, "bytes": path.stat().st_size} for path, name in files], "note": "工具执行完成不等于用例通过。录像保留真实时间与缺口;各来源截图不能相互替代。", diff --git a/apps/admin_console/services/execution_config.py b/apps/admin_console/services/execution_config.py new file mode 100644 index 0000000..daca01f --- /dev/null +++ b/apps/admin_console/services/execution_config.py @@ -0,0 +1,53 @@ +"""Credential-free execution settings persisted for repeatable favorites.""" + +from apps.admin_console.services.task_import import TaskOptions +from artemis.config import load_agent_config, parse_llm_config, verification_level_for_checker +from artemis.config.settings import settings + + +def execution_config(item: dict) -> dict: + agent = load_agent_config() + model = parse_llm_config().operator + options = { + "profile": item.get("profile") or "flash", + "model": model.model, + "reasoning_effort": model.reasoning_effort or "default", + "request_timeout_seconds": model.timeout, + "max_retries": model.max_retries, + "retry_interval_seconds": settings.OPENAI_RETRY_INTERVAL_SECONDS, + "record_video": settings.ARTEMIS_RECORDING_ENABLED, + "recording": agent.recording.model_dump(), + "analyze_video": agent.video_analyzer.enable_ledger, + "timeout_seconds": settings.ARTEMIS_TASK_TIMEOUT_SECONDS, + "verification_level": verification_level_for_checker(agent.checker), + "explorer_mode": agent.explorer.resolve(profile=item.get("profile") or "flash"), + "observation_timeout_seconds": settings.ARTEMIS_OBSERVATION_TIMEOUT_SECONDS, + "observation_max_attempts": settings.ARTEMIS_OBSERVATION_MAX_ATTEMPTS, + "device_command_timeout_seconds": settings.ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS, + "action_failure_attempts": settings.ARTEMIS_ACTION_FAILURE_ATTEMPTS, + "worker_cleanup_timeout_seconds": settings.ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS, + } + for key in ("device_serial", "verification_level", "explorer_mode", "locked_app_package"): + if item.get(key) is not None: + options[key] = item[key] + overrides = item.get("runtime_overrides") or {} + options["recording"] = {**options["recording"], **(overrides.get("recording") or {})} + options.update({k: v for k, v in overrides.items() if k != "recording" and v is not None}) + return { + "runtime_overrides": TaskOptions.model_validate(options).model_dump(exclude_none=True), + "expected_output": item.get("expected_output"), + "enable_outputter": item.get("enable_outputter") + if item.get("enable_outputter") is not None + else agent.outputter.enabled, + "app_path": item.get("app_path"), + } + + +def persist_execution_config(item: dict) -> None: + from apps.admin_console.services.automation_store import store + + snapshot = execution_config(item) + item["execution_config"] = snapshot + if job := store.job(str(item["session_id"])): + job["execution_config"] = snapshot + store.save_job(job) diff --git a/apps/admin_console/services/exploration_deletion.py b/apps/admin_console/services/exploration_deletion.py index e8c74fc..859da5b 100644 --- a/apps/admin_console/services/exploration_deletion.py +++ b/apps/admin_console/services/exploration_deletion.py @@ -32,7 +32,7 @@ def delete_exploration(repository: ExplorationStore, exploration_id: str) -> lis # Validation sessions and batch copies remain ordinary Flash/Pro history. deleted = [run["id"] for run in runs if run["kind"] == "exploration"] storage = StorageManager(repository.database.path, report_export.TRACES_PATH) - with report_export.EXPORT_LOCK: + with report_export.export_lock(*("session:" + run_id for run_id in deleted)): for run_id in deleted: safe_id = str(UUID(run_id)) storage.delete_session(UUID(safe_id)) diff --git a/apps/admin_console/services/export_cache.py b/apps/admin_console/services/export_cache.py new file mode 100644 index 0000000..51743d7 --- /dev/null +++ b/apps/admin_console/services/export_cache.py @@ -0,0 +1,66 @@ +"""Per-object export coordination and content-aware artifact reuse.""" + +from contextlib import contextmanager, ExitStack +import hashlib +import json +from pathlib import Path +import threading +from weakref import WeakValueDictionary + +_guard = threading.Lock() +_locks: WeakValueDictionary = WeakValueDictionary() +TERMINAL = {"completed", "success", "failed", "cancelled", "interrupted", "skipped"} + + +@contextmanager +def export_lock(*keys: str): + # Retain strong references until every lock is released. Unrelated exports + # run concurrently; nested session exports can reacquire their own RLock. + with _guard: + locks = [] + for key in sorted(set(keys)): + lock = _locks.get(key) + if lock is None: + lock = threading.RLock() + _locks[key] = lock + locks.append(lock) + with ExitStack() as stack: + for lock in locks: + stack.enter_context(lock) + yield + + +def fingerprint(report: dict, files=(), extra=None) -> str: + metadata = [] + for path, name in files: + stat = path.stat() + metadata.append((name, str(path), stat.st_size, stat.st_mtime_ns)) + payload = json.dumps([2, report, metadata, extra], sort_keys=True, default=str) + return hashlib.sha256(payload.encode()).hexdigest() + + +def reusable(path: Path, signature: str, status: str) -> bool: + if status not in TERMINAL or not path.is_file(): + return False + try: + return path.with_suffix(path.suffix + ".fingerprint").read_text() == signature + except OSError: + return False + + +def remember(path: Path, signature: str) -> None: + marker = path.with_suffix(path.suffix + ".fingerprint") + temporary = marker.with_suffix(".tmp") + temporary.write_text(signature, encoding="ascii") + temporary.replace(marker) + + +def compression(path: Path) -> int: + import zipfile + + return ( + zipfile.ZIP_STORED + if path.suffix.lower() + in {".mp4", ".mkv", ".webm", ".docx", ".xlsx", ".png", ".jpg", ".jpeg", ".zip"} + else zipfile.ZIP_DEFLATED + ) diff --git a/apps/admin_console/services/favorites.py b/apps/admin_console/services/favorites.py new file mode 100644 index 0000000..a284b26 --- /dev/null +++ b/apps/admin_console/services/favorites.py @@ -0,0 +1,406 @@ +"""Independent favorite snapshots dispatched through the existing task services.""" + +from copy import deepcopy +import asyncio +import hashlib +import json +import time +from typing import Literal +from uuid import NAMESPACE_URL, uuid5 +from weakref import WeakValueDictionary + +from pydantic import BaseModel, ConfigDict, Field + +from apps.admin_console.database.repositories.session_repository import SessionRepository +from apps.admin_console.services.automation_store import AutomationStore, identifier, store +from apps.admin_console.services.execution_config import execution_config +from apps.admin_console.services.model_service import model_service +from apps.admin_console.services.task_import import TaskCase, TaskOptions, execution_title +from artemis.config.exploration import ExplorationConfig +from artemis.exploration.case_prompt import require_resolved_case +from artemis.exploration.store import ExplorationStore + + +class FavoriteCreate(BaseModel): + model_config = ConfigDict(extra="forbid") + source_kind: Literal["task", "batch", "exploration", "exploration_case"] = "task" + source_id: str = Field(pattern=r"^[A-Za-z0-9_-]{1,100}$") + version: int | None = Field(default=None, ge=1) + name: str | None = Field(default=None, min_length=1, max_length=200) + + +class FavoriteUpdate(BaseModel): + model_config = ConfigDict(extra="forbid") + name: str | None = Field(default=None, min_length=1, max_length=200) + task_config: TaskOptions | None = None + case_configs: dict[int, TaskOptions] = Field(default_factory=dict) + + +class FavoriteRun(FavoriteUpdate): + request_id: str = Field(pattern=r"^[A-Za-z0-9_-]{1,100}$") + + +def _merge_options(target: dict, updates: dict) -> None: + if isinstance(updates.get("recording"), dict): + updates = { + **updates, + "recording": {**(target.get("recording") or {}), **updates["recording"]}, + } + target.update(updates) + + +class FavoriteService: + def __init__(self, storage: AutomationStore = store): + self.store = storage + self._run_locks: WeakValueDictionary = WeakValueDictionary() + + def get(self, key: str) -> dict: + value = self.store.get("favorite", identifier(key)) + if value is None: + raise KeyError("收藏不存在") + return value + + def _task(self, session_id: str, *, fallback: dict | None = None) -> tuple[dict, list[str]]: + job = self.store.job(session_id) or {} + if snapshot := job.get("execution_config"): + return deepcopy(snapshot), [] + row = SessionRepository(self.store.path).get_session_by_id(session_id) or {} + submission = job.get("submission") or fallback or {} + options = { + **(job.get("case") or {}).get("task_config", {}), + **(submission.get("runtime_overrides") or {}), + } + options = {key: value for key, value in options.items() if value is not None} + try: + device = row.get("device_info") or {} + device = json.loads(device) if isinstance(device, str) else device + except (ValueError, TypeError): + device = {} + device = device if isinstance(device, dict) else {} + profile = ( + options.get("profile") + or submission.get("profile") + or model_service.resolve_session_profile(row) + or "flash" + ) + options["profile"] = profile + serial = ( + options.get("device_serial") + or submission.get("device_serial") + or job.get("device_serial") + or device.get("serial") + ) + if serial: + options["device_serial"] = serial + if serial == "computer:local" or device.get("execution_target") == "computer": + options["execution_target"] = "computer" + # Historical diagnostics describe the actual model, unlike display metadata + # which may reflect today's defaults. Read only execution model roles. + with self.store.connection() as conn: + rows = conn.execute( + "SELECT payload FROM traces WHERE session_id=? AND type='llm_diagnostic' ORDER BY timestamp DESC", + (session_id,), + ).fetchall() + for trace in rows: + try: + data = json.loads(trace[0]) + except (ValueError, TypeError): + continue + if str(data.get("role", "")).lower() in { + "operator", + "flashrunner", + "reactiverunner", + "planner", + }: + for key in ("model", "reasoning_effort"): + if data.get(key): + options.setdefault(key, data[key]) + if options.get("model"): + break + missing = [ + key + for key in ( + "model", + "reasoning_effort", + "record_video", + "recording", + "analyze_video", + "request_timeout_seconds", + "max_retries", + "retry_interval_seconds", + "timeout_seconds", + ) + if key not in options + ] + snapshot = execution_config( + {**submission, "profile": profile, "runtime_overrides": options} + ) + return snapshot, missing + + def create(self, value: FavoriteCreate) -> dict: + token = f"{value.source_kind}:{value.source_id}:{value.version or ''}" + key = hashlib.sha256(token.encode()).hexdigest()[:32] + if existing := self.store.get("favorite", key): + return existing + snapshot, missing, kind, version = {}, [], value.source_kind, None + source_id = value.source_id + repo = ExplorationStore(self.store) + if kind == "task": + run = repo.run(source_id) + if run and run["kind"] == "exploration": + source_id = run["exploration_id"] + kind = "exploration_case" if repo.case(source_id) else "exploration" + if kind in {"exploration", "exploration_case"}: + session = repo.get(source_id) + options = deepcopy(session["task_config"]) + if kind == "exploration": + snapshot = { + "goal": session["goal"], + "config": session["config"], + "task_config": options, + } + name = session["goal"][:120] + else: + case = repo.case(source_id, value.version) + if not case: + raise KeyError("确认用例不存在") + version = case["version"] + validations = [ + r + for r in repo.runs(source_id) + if r["kind"] == "validation" and r.get("case_version") == version + ] + if validations: + config, missing = self._task( + validations[-1]["id"], + fallback={"runtime_overrides": validations[-1]["task_config"]}, + ) + else: + options = { + k: v + for k, v in options.items() + if k not in {"profile", "model", "reasoning_effort", "timeout_seconds"} + } + config = execution_config({"profile": "flash", "runtime_overrides": options}) + missing = ["首次用例复跑,默认 Flash"] + snapshot = {"goal": require_resolved_case(case), **config} + name = case["name"] + elif kind == "batch": + batch = self.store.get("batch", source_id) + if not batch or batch.get("status") == "deleting": + raise KeyError("批次不存在") + cases, seen = [], set() + for job in self.store.jobs(source_id): + position = job.get("case_position", job["position"]) + if position in seen: + continue + seen.add(position) + case = deepcopy(job["case"]) + config, inherited = self._task( + job["session_id"], + fallback={ + "profile": batch["profile"], + "device_serial": batch.get("device_serial"), + "runtime_overrides": { + "timeout_seconds": batch.get("timeout_seconds"), + **case.get("task_config", {}), + }, + }, + ) + case["task_config"] = config["runtime_overrides"] + cases.append(TaskCase.model_validate(case).model_dump(exclude_none=True)) + missing.extend(inherited) + if not cases: + raise ValueError("批次没有可收藏的用例") + snapshot = { + key: batch[key] + for key in ( + "name", + "formats", + "profile", + "device_serial", + "timeout_seconds", + "stop_on_failure", + "repeat_count", + ) + if key in batch + } + snapshot["cases"] = cases + name = batch["name"] + else: + row = SessionRepository(self.store.path).get_session_by_id(source_id) or {} + job = self.store.job(source_id) or {} + goal = ( + (job.get("case") or {}).get("goal") + or row.get("initial_goal") + or (job.get("submission") or {}).get("goal") + ) + if not goal: + raise KeyError("任务不存在或没有可复用的提示词") + config, missing = self._task(source_id) + snapshot = {"goal": goal, **config} + name = execution_title(job) or goal[:120] + favorite = { + "id": key, + "name": (value.name or name).strip(), + "kind": kind, + "source_kind": value.source_kind, + "source_id": value.source_id, + "case_version": version, + "snapshot": snapshot, + "inherited_fields": sorted(set(missing)), + "created_at": time.time(), + "updated_at": time.time(), + } + if not favorite["name"]: + raise ValueError("收藏名称不能为空") + return self.store.put("favorite", key, favorite) + + def updated(self, favorite: dict, value: FavoriteUpdate) -> dict: + result = deepcopy(favorite) + if value.name is not None: + if not value.name.strip(): + raise ValueError("收藏名称不能为空") + result["name"] = value.name.strip() + snapshot = result["snapshot"] + options = value.task_config.model_dump(exclude_none=True) if value.task_config else {} + if result["kind"] == "batch": + for index, case in enumerate(snapshot["cases"]): + if override := value.case_configs.get(index): + _merge_options(case["task_config"], override.model_dump(exclude_none=True)) + _merge_options(case["task_config"], options) + if set(value.case_configs) - set(range(len(snapshot["cases"]))): + raise ValueError("批次用例下标不存在") + if options.get("device_serial"): + snapshot["device_serial"] = options["device_serial"] + elif result["kind"] == "exploration": + config = snapshot["config"] + config.update( + { + k: options[k] + for k in ("model", "reasoning_effort", "timeout_seconds") + if k in options + } + ) + snapshot["config"] = ExplorationConfig.model_validate(config).model_dump() + _merge_options(snapshot["task_config"], options) + else: + _merge_options(snapshot["runtime_overrides"], options) + return result + + def update(self, key: str, value: FavoriteUpdate) -> dict: + favorite = self.updated(self.get(key), value) + favorite["updated_at"] = time.time() + return self.store.put("favorite", key, favorite) + + def delete(self, key: str) -> None: + with self.store.connection() as conn: + conn.execute( + "DELETE FROM automation_records WHERE kind='favorite' AND id=?", (identifier(key),) + ) + conn.commit() + + async def run(self, key: str, value: FavoriteRun) -> dict: + lock = self._run_locks.setdefault(value.request_id, asyncio.Lock()) + async with lock: + return await self._run(key, value) + + async def _run(self, key: str, value: FavoriteRun) -> dict: + key = identifier(key) + request = {"favorite_id": key, **value.model_dump(exclude_none=True)} + signature = hashlib.sha256(json.dumps(request, sort_keys=True).encode()).hexdigest() + # Reserve a durable snapshot before admission. Retries survive restarts, + # edits/deletions of the favorite, and lost HTTP responses. + with self.store.connection() as conn: + conn.execute("BEGIN IMMEDIATE") + row = conn.execute( + "SELECT payload FROM automation_records WHERE kind='favorite_run' AND id=?", + (value.request_id,), + ).fetchone() + record = json.loads(row[0]) if row else None + if record and record["signature"] != signature: + raise ValueError("request_id 已用于其他内容") + if record and record.get("result"): + return record["result"] + if record is None: + row = conn.execute( + "SELECT payload FROM automation_records WHERE kind='favorite' AND id=?", (key,) + ).fetchone() + if row is None: + raise KeyError("收藏不存在") + record = { + "signature": signature, + "favorite": self.updated(json.loads(row[0]), value), + } + conn.execute( + "INSERT INTO automation_records VALUES (?,?,?,?)", + ( + "favorite_run", + value.request_id, + json.dumps(record, ensure_ascii=False), + time.time(), + ), + ) + conn.commit() + favorite = record["favorite"] + snapshot = favorite["snapshot"] + run_id = str(uuid5(NAMESPACE_URL, "artemis-favorite:" + value.request_id)) + if favorite["kind"] == "batch": + from apps.admin_console.services.automation_service import create_batch, batch_action + + batch = create_batch( + {**snapshot, "name": favorite["name"], "request_id": "favorite-" + run_id} + ) + if batch["status"] == "draft": + batch = batch_action(batch["id"], "resume") + result = {"kind": "batch", "batch_id": batch["id"], "status": batch["status"]} + elif favorite["kind"] == "exploration": + from apps.admin_console.services.exploration_service import exploration_service + from artemis.exploration.models import CreateExploration + + session = exploration_service.create(CreateExploration(request_id=run_id, **snapshot)) + admission = await exploration_service.start(session["id"], run_id) + run = admission.get("run", admission) + rejection = admission.get("admission", {}) + outcome = run.get("outcome") or {} + if ( + rejection.get("status") in {"error", "rejected"} + or outcome.get("status") == "rejected" + ): + raise ValueError( + rejection.get("error") + or rejection.get("message") + or outcome.get("error") + or "探索未入队" + ) + result = { + "kind": "exploration", + "session_id": run["id"], + "exploration_id": session["id"], + "admission": admission, + } + else: + from apps.admin_console.routers.tasks import run_task + from apps.admin_console.schemas.task_schema import RunRequest + + admission = await run_task( + RunRequest( + goal=snapshot["goal"], + profile=snapshot["runtime_overrides"].get("profile", "flash"), + runtime_overrides=TaskOptions.model_validate(snapshot["runtime_overrides"]), + session_id=run_id, + ingress="favorite", + expected_output=snapshot.get("expected_output"), + enable_outputter=snapshot.get("enable_outputter"), + app_path=snapshot.get("app_path"), + ) + ) + if not admission.get("tasks"): + raise ValueError(admission.get("error") or admission.get("message") or "任务未入队") + result = {"kind": "task", "session_id": run_id, "admission": admission} + record["result"] = result + self.store.put("favorite_run", value.request_id, record) + return result + + +favorites = FavoriteService() diff --git a/apps/admin_console/services/report_export.py b/apps/admin_console/services/report_export.py index 552eee0..95ec61f 100644 --- a/apps/admin_console/services/report_export.py +++ b/apps/admin_console/services/report_export.py @@ -5,8 +5,14 @@ from typing import Callable, cast import json from pathlib import Path import re -import threading import zipfile +from apps.admin_console.services.export_cache import ( + compression, + export_lock, + fingerprint, + remember, + reusable, +) from apps.admin_console.services.request_diagnostics import diagnostics_json from docx import Document @@ -28,7 +34,7 @@ from artemis.runtime.case_result import RESULT_LABELS, resolve_case_result from apps.admin_console.services.case_review import reviewed_result EXPORT_ROOT = WORKSPACE_ROOT / "artifacts" / "reports" -EXPORT_LOCK = threading.RLock() +BATCH_ARCHIVE_SIGNATURE = "artemis-batch-layout-2" STATUS = { "completed": "执行完成", "success": "执行完成", @@ -741,9 +747,12 @@ def excel_report(report: dict, path: Path): wb.save(path) -def export_session(session_id: str, formats: list[str] | None = None) -> list[Path]: - with EXPORT_LOCK: - report = session_report(session_id) +def export_session( + session_id: str, formats: list[str] | None = None, *, report: dict | None = None +) -> list[Path]: + with export_lock("session:" + identifier(session_id)): + report = report if report is not None else session_report(session_id) + signature = fingerprint(report, session_screenshots(report)) directory = EXPORT_ROOT / "sessions" / identifier(session_id) directory.mkdir(parents=True, exist_ok=True) outputs = [] @@ -751,9 +760,13 @@ def export_session(session_id: str, formats: list[str] | None = None) -> list[Pa if fmt not in {"docx", "xlsx"}: raise ValueError("不支持的报告格式") path = directory / f"任务报告.{fmt}" + if reusable(path, signature, report["status"]): + outputs.append(path) + continue temporary = directory / f"生成中.{fmt}" (word_report if fmt == "docx" else excel_report)(report, temporary) temporary.replace(path) + remember(path, signature) outputs.append(path) return outputs @@ -845,13 +858,19 @@ def session_artifacts(report: dict) -> list[tuple[Path, str]]: ] -def export_session_media(session_id: str, kind: str) -> Path: +def session_download(session_id: str, kind: str) -> tuple[Path, str]: + with export_lock("session:" + identifier(session_id)): + report = session_report(session_id) + return export_session_media(session_id, kind, report=report), archive_case_name(report) + + +def export_session_media(session_id: str, kind: str, *, report: dict | None = None) -> Path: from apps.admin_console.services.evidence_service import export_index, plugin_files if kind not in {"screenshots", "video", "all"}: raise ValueError("不支持的下载类型") - with EXPORT_LOCK: - report = session_report(session_id) + with export_lock("session:" + identifier(session_id)): + report = report if report is not None else session_report(session_id) target_dir = EXPORT_ROOT / "sessions" / identifier(session_id) target_dir.mkdir(parents=True, exist_ok=True) images = session_screenshots(report) if kind in {"screenshots", "all"} else [] @@ -870,42 +889,61 @@ def export_session_media(session_id: str, kind: str) -> Path: raise ValueError("此任务没有可下载的步骤截图") target = target_dir / f"{kind}.zip" temporary = target_dir / f".{kind}.tmp" + files = images + videos + artifacts + extras + diagnostics = diagnostics_json(session_id) if kind == "all" else None + diagnostic_state = json.loads(diagnostics) if diagnostics else None + if isinstance(diagnostic_state, dict): + diagnostic_state.pop("collected_at", None) + signature = fingerprint(report, files, diagnostic_state) + if reusable(target, signature, report["status"]): + return target manifest = { "session_id": session_id, "name": report["name"], "status": report["status"], "verdict": report["verdict"], "screenshots": [name for _, name in images], - "videos": ["任务录屏/" + name for _, name in videos] + "videos": ["任务录屏/安卓/" + name for _, name in videos] + [name for path, name in extras if path.suffix.lower() in {".mp4", ".mkv", ".webm"}], "artifacts": [name for _, name in artifacts], "note": "录屏按时间顺序分段;暂停期间未占用设备,也未录制其他任务。", + "snapshot": report["status"] + not in {"completed", "success", "failed", "cancelled", "interrupted", "skipped"}, } 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, prefix + path.name) + for path in export_session(session_id, ["docx", "xlsx"], report=report): + archive.write(path, prefix + path.name, compress_type=compression(path)) for source, name in images: - archive.write(source, f"{prefix}步骤截图/{name}") + archive.write(source, f"{prefix}步骤截图/{name}", compress_type=compression(source)) for source, name in videos: - archive.write(source, f"{prefix}任务录屏/{name}", compress_type=zipfile.ZIP_STORED) + archive.write( + source, f"{prefix}任务录屏/安卓/{name}", compress_type=zipfile.ZIP_STORED + ) for source, name in artifacts: - archive.write(source, f"{prefix}任务文件/安卓/{name}") + archive.write( + source, f"{prefix}任务文件/安卓/{name}", compress_type=compression(source) + ) for source, name in extras: archive.write( source, prefix + name, - compress_type=zipfile.ZIP_STORED - if source.suffix.lower() in {".mp4", ".mkv", ".webm"} - else zipfile.ZIP_DEFLATED, + compress_type=compression(source), ) if kind == "all": - archive.writestr(prefix + "模型请求诊断.json", diagnostics_json(session_id)) + archive.writestr(prefix + "模型请求诊断.json", diagnostics or "{}") archive.writestr( prefix + "证据索引.json", json.dumps( - export_index(session_id, report, videos), + export_index( + session_id, + report, + videos, + images=images, + artifacts=artifacts, + extras=extras, + ), ensure_ascii=False, indent=2, default=str, @@ -915,13 +953,18 @@ def export_session_media(session_id: str, kind: str) -> Path: prefix + "文件清单.json", json.dumps(manifest, ensure_ascii=False, indent=2) ) temporary.replace(target) + if fingerprint(report, files, diagnostic_state) == signature: + remember(target, signature) return target def export_batch(batch_id: str, progress: Callable[[int, int, str], None] | None = None) -> Path: from apps.admin_console.services.evidence_service import export_index, plugin_files - with EXPORT_LOCK: + with export_lock( + "batch:" + identifier(batch_id), + *("session:" + job["session_id"] for job in store.jobs(batch_id)), + ): batch = store.get("batch", identifier(batch_id)) if not batch: raise ValueError("批次不存在") @@ -959,34 +1002,47 @@ def export_batch(batch_id: str, progress: Callable[[int, int, str], None] | None temporary = directory / "批次报告.tmp" with zipfile.ZipFile(temporary, "w", zipfile.ZIP_DEFLATED) as archive: for path in outputs: - archive.write(path, path.name) + archive.write(path, path.name, compress_type=compression(path)) for number, report in enumerate(records, 1): if progress: progress(number - 1, len(records), f"正在打包任务 {number}/{len(records)}") prefix = f"任务报告/{number:04d}_{archive_case_name(report)}" - for path in export_session(report["id"], formats): - archive.write(path, f"{prefix}/{path.name}") - for source, name in session_screenshots(report): - archive.write(source, f"{prefix}/步骤截图/{name}") - for source, name in session_videos(report): + images = session_screenshots(report) + videos = session_videos(report) + artifacts = session_artifacts(report) + extras = plugin_files(report["id"]) + for path in export_session(report["id"], formats, report=report): + archive.write(path, f"{prefix}/{path.name}", compress_type=compression(path)) + for source, name in images: archive.write( - source, f"{prefix}/任务录屏/{name}", compress_type=zipfile.ZIP_STORED + source, f"{prefix}/步骤截图/{name}", compress_type=compression(source) ) - for source, name in session_artifacts(report): - archive.write(source, f"{prefix}/任务文件/安卓/{name}") - for source, name in plugin_files(report["id"]): + for source, name in videos: + archive.write( + source, f"{prefix}/任务录屏/安卓/{name}", compress_type=zipfile.ZIP_STORED + ) + for source, name in artifacts: + archive.write( + source, f"{prefix}/任务文件/安卓/{name}", compress_type=compression(source) + ) + for source, name in extras: archive.write( source, f"{prefix}/{name}", - compress_type=zipfile.ZIP_STORED - if source.suffix.lower() in {".mp4", ".mkv", ".webm"} - else zipfile.ZIP_DEFLATED, + compress_type=compression(source), ) archive.writestr(f"{prefix}/模型请求诊断.json", diagnostics_json(report["id"])) archive.writestr( f"{prefix}/证据索引.json", json.dumps( - export_index(report["id"], report, session_videos(report)), + export_index( + report["id"], + report, + videos, + images=images, + artifacts=artifacts, + extras=extras, + ), ensure_ascii=False, indent=2, default=str, @@ -997,4 +1053,5 @@ def export_batch(batch_id: str, progress: Callable[[int, int, str], None] | None if progress: progress(len(records), len(records), "正在完成压缩包") temporary.replace(target) + remember(target, BATCH_ARCHIVE_SIGNATURE) return target diff --git a/apps/admin_console/services/task_queue_service.py b/apps/admin_console/services/task_queue_service.py index 49e34a7..f1214ef 100644 --- a/apps/admin_console/services/task_queue_service.py +++ b/apps/admin_console/services/task_queue_service.py @@ -1033,6 +1033,9 @@ class TaskQueueService: if not isinstance(goal, str) or not goal.strip(): raise ValueError("Queued task must contain a non-empty string goal.") cls._begin_task_run(task_item, run_key, sess_id, goal, profile) + from apps.admin_console.services.execution_config import persist_execution_config + + await asyncio.to_thread(persist_execution_config, task_item) target = cls._task_target(task_item) cmd, env = cls._build_worker_invocation( @@ -1342,6 +1345,21 @@ class TaskQueueService: device_serial = await device_pool.select_device_async() except Exception: device_serial = None + from apps.admin_console.services.execution_config import execution_config + + config_snapshot = execution_config( + { + "profile": profile, + "device_serial": device_serial, + "expected_output": expected_output, + "enable_outputter": enable_outputter, + "verification_level": verification_level, + "explorer_mode": explorer_mode, + "locked_app_package": locked_app_package, + "app_path": app_path, + "runtime_overrides": runtime_overrides, + } + ) for i, goal in enumerate(goals): task_item = cls._create_queue_item( goal, @@ -1362,6 +1380,7 @@ class TaskQueueService: ) if runtime_overrides: task_item["runtime_overrides"] = runtime_overrides + task_item["execution_config"] = config_snapshot if pinned: try: task_item["plugin_resources"] = PluginStore().reserve( 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 ecd93b5..4deadb0 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 @@ -14,7 +14,7 @@

当前任务报告

@if (report(); as r) {
- {{r.name}} +
{{r.name}}
{{label(r.status)}}
@if(r.test_result; as result){} @if (r.error) {
异常原因

{{r.error}}

} @@ -76,6 +76,7 @@

批次记录

@for(b of batches(); track b.id){
+
@@ -106,6 +107,6 @@
} @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.scss b/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.scss index f04235a..9e5e7eb 100644 --- a/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.scss +++ b/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.scss @@ -26,7 +26,7 @@ input[type=checkbox]{width:15px;height:15px;accent-color:#2563eb;vertical-align: .case-goal{display:-webkit-box;-webkit-box-orient:vertical;-webkit-line-clamp:2;overflow:hidden;white-space:pre-wrap;margin:5px 0;font-size:12px;} .execution-summary{padding:9px 12px;border-radius:8px;background:#eff6ff;color:#1d4ed8;font-weight:500;} .record{display:flex;width:100%;text-align:left;align-items:flex-start;flex-direction:column;gap:4px;margin:8px 0 0;padding:12px;overflow-wrap:anywhere;}.record span{color:#64748b;font-size:12px;font-weight:400;}.record.selected{border-color:#93c5fd;background:#eff6ff;}.record.selected strong{color:#1d4ed8;} -.batch-record{display:flex;gap:8px;align-items:stretch;margin-top:8px;}.batch-record .record{flex:1;min-width:0;width:auto;margin:0;}.batch-record .delete-batch{flex:0 0 40px;min-width:40px;padding:0;color:#b91c1c;}.batch-record .delete-batch:hover:not(:disabled){background:#fff1f2;border-color:#fda4af;}.batch-record .delete-batch .material-symbols-outlined{font-size:19px;} +.batch-record{position:relative;display:flex;gap:8px;align-items:stretch;margin-top:8px;}.batch-record>app-favorite-button{position:absolute;left:9px;top:9px;z-index:1}.batch-record .record{padding-left:44px;flex:1;min-width:0;width:auto;margin:0;}.batch-record .delete-batch{flex:0 0 40px;min-width:40px;padding:0;color:#b91c1c;}.batch-record .delete-batch:hover:not(:disabled){background:#fff1f2;border-color:#fda4af;}.batch-record .delete-batch .material-symbols-outlined{font-size:19px;} .archive-progress{margin:12px 0;padding:11px 13px;border:1px solid #bfdbfe;border-radius:10px;background:#f8fbff;color:#1e40af;} .archive-progress-heading{display:flex;align-items:center;gap:8px;} .archive-progress p{margin:7px 0 0;font-size:12px;color:#475569;} @@ -41,6 +41,7 @@ input[type=checkbox]{width:15px;height:15px;accent-color:#2563eb;vertical-align: .task-guidance{margin-top:16px;padding:14px;border:1px solid #bfdbfe;border-radius:10px;background:#f8fbff;} .task-guidance label{margin:8px 0 4px;} .report-overview{padding:14px;border:1px solid #dbeafe;border-radius:10px;background:#f8fbff;} +.report-heading{display:flex;align-items:flex-start;gap:8px;}.report-heading .report-title{flex:1;min-width:0;} .report-title{display:block;font-size:14px;line-height:1.5;color:#0f172a;overflow-wrap:anywhere;} .report-badges{display:flex;gap:7px;flex-wrap:wrap;margin-top:9px;span{padding:3px 8px;border:1px solid #bfdbfe;border-radius:20px;background:#eff6ff;color:#1e40af;font-size:12px;&.failed{background:#fff1f2;border-color:#fecdd3;color:#be123c;}}} .report-error{margin-top:11px;padding:10px 12px;background:#fff1f2;border-left:3px solid #e11d48;border-radius:6px;color:#9f1239;strong{font-size:12px;}p{margin:4px 0 0;color:#9f1239;white-space:pre-wrap;}} diff --git a/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.spec.ts b/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.spec.ts index 11e6ad2..73152b9 100644 --- a/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.spec.ts +++ b/apps/showcase_ui/src/app/components/automation-panel/automation-panel.component.spec.ts @@ -129,7 +129,7 @@ describe('AutomationPanelComponent batch deletion', () => { {session_id:'task-b',status:'completed',case:{name:'用例 B',goal:'观察主页'}} ]}); fixture.detectChanges(); - const jobs: HTMLButtonElement[] = [...fixture.nativeElement.querySelectorAll('.job button')]; + const jobs: HTMLButtonElement[] = [...fixture.nativeElement.querySelectorAll('.job > button')]; jobs[0].click(); jobs[1].click(); expect(agent.selectSession.calls.allArgs()).toEqual([['task-a',true],['task-b',true]]); expect(component.tab()).toBe('batches'); 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 b02c0dc..a548b54 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 @@ -1,4 +1,5 @@ import { Component, DestroyRef, computed, effect, inject, output, signal } from '@angular/core'; +import { FavoriteButtonComponent } from '../favorite-button/favorite-button.component'; import { ReportMarkdownComponent } from '../report-markdown/report-markdown.component'; import { CommonModule } from '@angular/common'; import { FormsModule } from '@angular/forms'; @@ -22,7 +23,7 @@ interface ArchiveProgress { } @Component({ - selector: 'app-automation-panel', standalone: true, imports: [ReportMarkdownComponent, CommonModule, FormsModule, TestResultBadgeComponent, TestResultCardComponent, TaskPluginsComponent], + selector: 'app-automation-panel', standalone: true, imports: [FavoriteButtonComponent, ReportMarkdownComponent, CommonModule, FormsModule, TestResultBadgeComponent, TestResultCardComponent, TaskPluginsComponent], templateUrl: './automation-panel.component.html', styleUrl: './automation-panel.component.scss' }) export class AutomationPanelComponent { diff --git a/apps/showcase_ui/src/app/components/chat-interface/chat-interface.component.html b/apps/showcase_ui/src/app/components/chat-interface/chat-interface.component.html index 623bed6..c44fd8c 100644 --- a/apps/showcase_ui/src/app/components/chat-interface/chat-interface.component.html +++ b/apps/showcase_ui/src/app/components/chat-interface/chat-interface.component.html @@ -125,8 +125,11 @@ limitations under the License. } -
- {{ session.display_name || session.initial_goal || '未命名任务' }} +
+ +
+ {{ session.display_name || session.initial_goal || '未命名任务' }} +
} - -
- -
-
- - - -
- - - - - - - - - -
-
- -
- -
-
- - -
- @for (item of filteredSuggestions(); track item.id) { -
- -
-
- @for (app of item.apps; track app.name) { -
- {{ app.icon }} -
- } - {{ getAppNamesDisplay(item.apps) }} -
- - - {{ item.profile === 'pro' ? 'Pro' : 'Flash' }} - -
- - -
-

{{ item.title }}

-

{{ item.description }}

-
- - - -
- } -
-
+ diff --git a/apps/showcase_ui/src/app/pages/home/home.component.ts b/apps/showcase_ui/src/app/pages/home/home.component.ts index a17e683..36a74bd 100644 --- a/apps/showcase_ui/src/app/pages/home/home.component.ts +++ b/apps/showcase_ui/src/app/pages/home/home.component.ts @@ -1,6 +1,7 @@ import { CapabilitySettingsComponent } from '../../components/model-settings/capability-settings.component'; import { SettingSliderComponent, REASONING_VALUES, REASONING_LABELS } from '../../components/setting-slider/setting-slider.component'; import { ModelSettingsComponent } from '../../components/model-settings/model-settings.component'; +import { FeaturedTasksComponent } from '../../components/featured-tasks/featured-tasks.component'; /** * Copyright 2026 Google LLC * @@ -108,7 +109,7 @@ type AdbGuideTab = 'emulator' | 'usb' | 'wifi' | 'remote'; @Component({ selector: 'app-home', standalone: true, - imports: [FormsModule, ModelSettingsComponent, CapabilitySettingsComponent, TaskPluginsComponent, SettingSliderComponent], + imports: [FormsModule, ModelSettingsComponent, CapabilitySettingsComponent, TaskPluginsComponent, SettingSliderComponent, FeaturedTasksComponent], templateUrl: './home.component.html', changeDetection: ChangeDetectionStrategy.Eager, styleUrl: './home.component.scss' diff --git a/apps/showcase_ui/src/app/services/favorites.service.spec.ts b/apps/showcase_ui/src/app/services/favorites.service.spec.ts new file mode 100644 index 0000000..d6e4b7e --- /dev/null +++ b/apps/showcase_ui/src/app/services/favorites.service.spec.ts @@ -0,0 +1,30 @@ +import { TestBed, fakeAsync, flushMicrotasks } from '@angular/core/testing'; +import { provideHttpClient } from '@angular/common/http'; +import { HttpTestingController, provideHttpClientTesting } from '@angular/common/http/testing'; +import { FavoritesService } from './favorites.service'; + +describe('Favorites shared service', () => { + let api: FavoritesService, http: HttpTestingController; + beforeEach(() => { TestBed.configureTestingModule({providers:[provideHttpClient(),provideHttpClientTesting()]});api=TestBed.inject(FavoritesService);http=TestBed.inject(HttpTestingController); }); + afterEach(() => http.verify()); + it('shares one initial load across all favorite buttons and exposes errors for retry', fakeAsync(() => { + void api.load();void api.load(); + http.expectOne('/api/automation/favorites').flush('offline',{status:503,statusText:'Unavailable'});flushMicrotasks(); + expect(api.loaded()).toBeFalse();expect(api.error()).toContain('重试'); + void api.load(true);http.expectOne('/api/automation/favorites').flush([]);flushMicrotasks(); + void api.load();expect(api.loaded()).toBeTrue();http.expectNone('/api/automation/favorites'); + })); + it('reuses the submission ID after a lost reply but uses a new ID for the next intentional run',fakeAsync(() => { + void api.run('id',{}).catch(()=>{});const first=http.expectOne('/api/automation/favorites/id/run');const id=first.request.body.request_id; + first.error(new ProgressEvent('network'));flushMicrotasks(); + void api.run('id',{});const retry=http.expectOne('/api/automation/favorites/id/run');expect(retry.request.body.request_id).toBe(id);retry.flush({kind:'task',session_id:'new'});flushMicrotasks(); + void api.run('id',{});const next=http.expectOne('/api/automation/favorites/id/run');expect(next.request.body.request_id).not.toBe(id);next.flush({kind:'task'});flushMicrotasks(); + })); + it('allows a fresh user submission after a definitive admission rejection',fakeAsync(() => { + void api.run('id',{}).catch(()=>{}); + const rejected=http.expectOne('/api/automation/favorites/id/run');const id=rejected.request.body.request_id; + rejected.flush({detail:'设备离线'},{status:409,statusText:'Conflict'});flushMicrotasks(); + void api.run('id',{});const next=http.expectOne('/api/automation/favorites/id/run'); + expect(next.request.body.request_id).not.toBe(id);next.flush({kind:'task'});flushMicrotasks(); + })); +}); diff --git a/apps/showcase_ui/src/app/services/favorites.service.ts b/apps/showcase_ui/src/app/services/favorites.service.ts new file mode 100644 index 0000000..96975e0 --- /dev/null +++ b/apps/showcase_ui/src/app/services/favorites.service.ts @@ -0,0 +1,67 @@ +import { Injectable, inject, signal } from '@angular/core'; +import { HttpClient } from '@angular/common/http'; +import { firstValueFrom, timeout } from 'rxjs'; +import { UI_READ_TIMEOUT_MS } from '../core/http-timeouts'; + +export type FavoriteKind = 'task' | 'batch' | 'exploration' | 'exploration_case'; +export interface Favorite { + id: string; name: string; kind: FavoriteKind; source_kind: FavoriteKind; source_id: string; + case_version?: number; inherited_fields: string[]; + snapshot: { goal?: string; config?: Record; task_config?: Record; + runtime_overrides?: Record; cases?: {name: string; goal: string; task_config: Record}[]; + repeat_count?: number; device_serial?: string }; +} +export interface FavoriteRunResult {kind: string; session_id?: string; batch_id?: string; exploration_id?: string; admission?: any;} + +@Injectable({providedIn: 'root'}) +export class FavoritesService { + private http = inject(HttpClient); + private base = '/api/automation/favorites'; + items = signal([]); + loaded = signal(false); + loading = signal(false); + error = signal(''); + private pending?: Promise; + private requests = new Map(); + + load(force = false): Promise { + if (this.pending) return this.pending; + if (this.loaded() && !force) return Promise.resolve(); + this.loading.set(true); this.error.set(''); + this.pending = firstValueFrom(this.http.get(this.base).pipe(timeout(UI_READ_TIMEOUT_MS))) + .then(items => { this.items.set(items); this.loaded.set(true); }) + .catch(() => { this.error.set('精选任务读取失败,请重试'); }) + .finally(() => { this.loading.set(false); this.pending = undefined; }); + return this.pending; + } + async create(source_kind: FavoriteKind, source_id: string, version?: number) { + const item = await firstValueFrom(this.http.post(this.base, {source_kind, source_id, version})); + this.items.update(items => [item, ...items.filter(i => i.id !== item.id)]); + return item; + } + async remove(id: string) { + await firstValueFrom(this.http.delete(`${this.base}/${id}`)); + this.items.update(items => items.filter(i => i.id !== id)); + } + async update(id: string, value: Record) { + const item = await firstValueFrom(this.http.patch(`${this.base}/${id}`, value)); + this.items.update(items => items.map(i => i.id === id ? item : i)); + return item; + } + async run(id: string, value: Record): Promise { + const key = JSON.stringify([id, value]); + const request_id = this.requests.get(key) || crypto.randomUUID(); + this.requests.set(key, request_id); + try { + const result = await firstValueFrom(this.http.post(`${this.base}/${id}/run`, {...value, request_id})); + this.requests.delete(key); + return result; + } catch(e) { + // A definitive rejection permits a new user-initiated submission after + // fixing the cause. Network/5xx errors retain the original request ID. + const status = (e as {status?: number}).status || 0; + if(status >= 400 && status < 500)this.requests.delete(key); + throw e; + } + } +} diff --git a/artemis/data_engine/storage.py b/artemis/data_engine/storage.py index 825eebc..b234a2b 100644 --- a/artemis/data_engine/storage.py +++ b/artemis/data_engine/storage.py @@ -199,6 +199,18 @@ class StorageManager: ) """) + # Applied to existing databases as well. Report and timeline readers + # look up traces per step/session; never scan the entire history. + conn.execute( + "CREATE INDEX IF NOT EXISTS traces_step_type_time ON traces(step_id,type,timestamp)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS traces_session_type_time ON traces(session_id,type,timestamp)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS steps_session_number ON steps(session_id,step_number)" + ) + conn.execute(""" CREATE TABLE IF NOT EXISTS failed_outputs ( id INTEGER PRIMARY KEY AUTOINCREMENT, diff --git a/artemis/interfaces/cli/commands/mcp.py b/artemis/interfaces/cli/commands/mcp.py index 138ed94..0a59eb5 100644 --- a/artemis/interfaces/cli/commands/mcp.py +++ b/artemis/interfaces/cli/commands/mcp.py @@ -43,6 +43,7 @@ ARTEMIS_MCP_TOOLS = ( "mobile_config", "mobile_task_file", "mobile_batch", + "mobile_favorites", "mobile_exploration", "mobile_search_history", "mobile_export_report", diff --git a/docs/ai-automation.md b/docs/ai-automation.md index 27333bf..f8343d6 100644 --- a/docs/ai-automation.md +++ b/docs/ai-automation.md @@ -35,6 +35,7 @@ Flash 和 Pro 都支持任务内的安卓文件工具 `android_file`:保存原 | `mobile_manage_task` | 单任务状态、提交反查、暂停/继续、指导和停止;删除预览及删除、代录用户人工复验 | | `mobile_get_device_state` | 设备截图、界面层级 | | `mobile_task_file` | XLSX 模板与导入校验 | +| `mobile_favorites` | 收藏独立任务/用例/批次模板、调整默认配置、按原配置复跑;run 使用稳定 request_id | | `mobile_batch` | JSON 用例建批次、开始、暂停、继续、复制失败项、删除非活跃批次 | | `mobile_search_history` | 全库搜索、日期/状态/批次筛选和分页 | | `mobile_inspect_trace` | 只读步骤、截图和执行证据 | diff --git a/docs/automation-workspace.md b/docs/automation-workspace.md index bd55f5b..0c36ec3 100644 --- a/docs/automation-workspace.md +++ b/docs/automation-workspace.md @@ -79,3 +79,14 @@ Flash 跨端任务由模型通过内部 `update_task_progress` 记录短阶段 批次、任务元数据和报告偏好保存在现有 SQLite 数据库的 `automation_records`、`automation_jobs` 表。全局“清空任务历史”会清除批次与执行元数据;它与上文的单批次删除不同,已导出的文件会保留。 任务已经产生最终结果,但 SDK 线程或资源清理仍卡住时,默认最多再等待 30 秒后回收工作进程,避免继续占用队列。全局可用 `ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS` 或功能配置的 `worker_cleanup_timeout_seconds` 调整,任务 JSON 同名字段可覆盖。原有结果与截图保留,未完成保存的录屏尽力恢复。 + + +## 精选任务 + +首页的精选任务来自用户收藏。任务队列、历史、报告及批次旁的星标均可收藏;批次中的单项也可单独收藏。收藏保存独立模板,删除原记录不影响复跑。 + +卡片直接点击“运行”会创建新任务;展开“配置”可调整模型、思考强度、模式或执行设备。不修改时沿用原配置,修改默认仅影响本次,点击“保存为默认配置”才更新收藏。旧任务没有完整记录的配置会明确标出,收藏时采用当前默认值。原设备离线时需要选择在线设备,纯电脑任务无需安卓设备。 + +批次保留原顺序、循环次数和失败停止策略,默认保留各项配置,也可明确选择统一模型和思考强度。探索用例固定收藏版本,首次无复跑配置时默认 Flash;“收藏为重新探索”会创建新的探索会话并沿用原探索配置。 + +单任务下载全部资料时,未变化的终态报告和资料包会复用;人工复验、结果和资料变化会重新生成。批次沿用已生成的报告包,人工复验后重新打包;旧目录格式的包首次下载时更新。运行中的下载是当前快照。录像分为“安卓”“电脑”“人工示范”目录,索引路径与实际文件一致。打包优先速度,对已压缩的图片、视频和 Office 文件直接存储,保持原画质。 diff --git a/mcp_server/README.md b/mcp_server/README.md index 0ebf77e..9fcc3dd 100644 --- a/mcp_server/README.md +++ b/mcp_server/README.md @@ -100,3 +100,5 @@ Run `uv run artemis mcp --generate-config codex` to output a configuration templ ## 完整 AI 自动化 工作区工具与网页共用后台服务,覆盖普通与探索任务、批次、历史、诊断、人工复验及报告导出。参见 [接入说明](../docs/ai-automation.md) 和 [Skill](../skills/artemis-use/SKILL.md)。网页启动时同步官方 Skill 与 MCP 接入;已有审批策略及工具白名单保留。更新后重新加载 MCP 才能使用新版工具定义。 + +* **`mobile_favorites`**: Save independent task/case/batch templates, list or rename favorites, adjust their defaults, and start a new run with original settings. Per-run overrides are temporary; run retries use the same request_id. Source deletion does not delete favorites. See the [batch and favorite workflow](../skills/artemis-use/references/batches.md). diff --git a/mcp_server/catalog.json b/mcp_server/catalog.json index f4eaaee..5b3702b 100644 --- a/mcp_server/catalog.json +++ b/mcp_server/catalog.json @@ -501,6 +501,102 @@ "type": "object" } }, + { + "name": "mobile_favorites", + "description": "Save independent favorite templates and run them on the existing queue.\n\nlist is read-only. create snapshots source_id (task session/batch/exploration),\nincluding original execution settings; exploration_case pins a case version.\nAn exploration task automatically favorites its confirmed case when available.\nchanges accepts name, task_config (TaskOptions), case_configs (zero-based case\nindices to TaskOptions). No credentials or arbitrary snapshot input is accepted.\nupdate saves defaults; run applies changes to this execution only and requires\na stable request_id. After an uncertain reply retry the same payload and ID.\nrun creates a new execution, never continues the source or copies its verdict.\nBatch order/repetitions and per-case settings are preserved. Missing historical\nsettings are explicitly marked inherited_fields. Offline devices are not replaced.\ndelete removes only the favorite, never the source history or previous runs.\n", + "inputSchema": { + "properties": { + "action": { + "enum": [ + "list", + "create", + "update", + "delete", + "run" + ], + "title": "Action", + "type": "string" + }, + "favorite_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Favorite Id" + }, + "source_kind": { + "default": "task", + "enum": [ + "task", + "batch", + "exploration", + "exploration_case" + ], + "title": "Source Kind", + "type": "string" + }, + "source_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Source Id" + }, + "version": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Version" + }, + "request_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Request Id" + }, + "changes": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Changes" + } + }, + "required": [ + "action" + ], + "title": "mobile_favoritesArguments", + "type": "object" + } + }, { "name": "mobile_batch", "description": "Create, execute and manage durable batches on the unified device queue.\n\ncreate requires a unique request_id and batch={name,device_serial,cases:[{name,goal,\ncase_id?,category?,subcategory?,preconditions?,task_config?:{profile?,model?,\nreasoning_effort?,device_serial?,verification_level?,explorer_mode?,\nlocked_app_package?,record_video?,analyze_video?,request_timeout_seconds?,\nmax_retries?,retry_interval_seconds?,timeout_seconds?}}],profile?:flash|pro,\nformats?:[docx,xlsx],stop_on_failure?:false,timeout_seconds?:1800,repeat_count?:1}.\nrepeat_count is a positive integer: execute all selected cases in order for each\nround, with independent session IDs, statuses and reports. The default total limit\n(cases * repeat_count) is 10000, configurable via ARTEMIS_BATCH_MAX_EXECUTIONS.\nOnly include user-selected cases; get returns iteration/repeat_count per job.\nRequired device\nmust be user-selected. Creation validates and saves a DRAFT; start executes it.\nReuse the same request_id and exact payload after a connection failure; never invent\na new ID for the same submission. lookup(request_id) resolves an uncertain outcome.\nget returns per-case status. pause stops future dispatch, not the active task;\nuse mobile_manage_task(stop) to stop that task. resume never repeats finished cases.\nretry_failed copies each failed execution once into a new draft (repeat_count=1)\nonly after an explicit rerun request; it is not\ntransport-idempotent, so after an uncertain response use list instead of retrying.\ndelete removes a non-active batch and its task sessions, reports and files after\nan explicit user request. Confirm the batch ID with get first; deletion is irreversible.\nRestarted batches remain paused. No automatic failed-case retries or action replay.\n", diff --git a/mcp_server/tools/__init__.py b/mcp_server/tools/__init__.py index d15f98a..c962812 100644 --- a/mcp_server/tools/__init__.py +++ b/mcp_server/tools/__init__.py @@ -33,6 +33,7 @@ __all__ = [ "mobile_batch", "mobile_search_history", "mobile_export_report", + "mobile_favorites", "mobile_exploration", ] @@ -43,4 +44,5 @@ from mcp_server.tools.automation import ( mobile_batch, mobile_search_history, mobile_export_report, + mobile_favorites, ) diff --git a/mcp_server/tools/automation.py b/mcp_server/tools/automation.py index dd9f3fd..963c977 100644 --- a/mcp_server/tools/automation.py +++ b/mcp_server/tools/automation.py @@ -268,6 +268,73 @@ async def mobile_task_file( ) +@mcp.tool() +async def mobile_favorites( + action: Literal["list", "create", "update", "delete", "run"], + favorite_id: str | None = None, + source_kind: Literal["task", "batch", "exploration", "exploration_case"] = "task", + source_id: str | None = None, + version: int | None = None, + request_id: str | None = None, + changes: dict[str, Any] | None = None, +) -> Any: + """Save independent favorite templates and run them on the existing queue. + + list is read-only. create snapshots source_id (task session/batch/exploration), + including original execution settings; exploration_case pins a case version. + An exploration task automatically favorites its confirmed case when available. + changes accepts name, task_config (TaskOptions), case_configs (zero-based case + indices to TaskOptions). No credentials or arbitrary snapshot input is accepted. + update saves defaults; run applies changes to this execution only and requires + a stable request_id. After an uncertain reply retry the same payload and ID. + run creates a new execution, never continues the source or copies its verdict. + Batch order/repetitions and per-case settings are preserved. Missing historical + settings are explicitly marked inherited_fields. Offline devices are not replaced. + delete removes only the favorite, never the source history or previous runs. + """ + if action == "list": + return await _request("GET", "/api/automation/favorites") + payload = changes or {} + if set(payload) - {"name", "task_config", "case_configs"}: + raise ValueError("收藏只接受名称和非敏感执行配置") + + # The HTTP schema applies the complete TaskOptions allowlist. Refuse secret + # fields before making a network request, including nested case overrides. + def check(value): + if isinstance(value, dict): + for key, child in value.items(): + if any( + word in str(key).lower() + for word in ("secret", "password", "api_key", "token", "authorization") + ): + raise ValueError("收藏不接受密钥或授权配置") + check(child) + elif isinstance(value, list): + for child in value: + check(child) + + check(payload) + if action == "create": + if set(payload) - {"name"}: + raise ValueError("创建收藏从源记录读取配置,修改默认配置请使用 update") + return await _request( + "POST", + "/api/automation/favorites", + json={ + "source_kind": source_kind, + "source_id": _id(source_id), + "version": version, + **payload, + }, + ) + route = f"/api/automation/favorites/{_id(favorite_id)}" + if action == "update": + return await _request("PATCH", route, json=payload) + if action == "delete": + return await _request("DELETE", route) + return await _request("POST", route + "/run", json={**payload, "request_id": _id(request_id)}) + + @mcp.tool() async def mobile_batch( action: Literal[ diff --git a/skills/artemis-use/references/batches.md b/skills/artemis-use/references/batches.md index af42341..ef1391b 100644 --- a/skills/artemis-use/references/batches.md +++ b/skills/artemis-use/references/batches.md @@ -66,3 +66,19 @@ `mobile_export_report(batch_id=...)` 下载批次 ZIP,包含汇总、各项报告和有序步骤截图;任务保存了录屏时,还包含对应录屏片段。报告格式由创建批次时的设置决定。单任务传 `session_id`(单任务 MCP 的 `trace_id`)以及可选 `formats=["docx"]`、`["screenshots"]`、`["video"]` 或 `["all"]`。默认格式通过 `mobile_config(section="reports")` 查看。录屏与 AI 视频分析独立,只有保存了录屏才可下载片段。 结果文件在 MCP 主机的 `artifacts/mcp_exports/`。不要把 Office 文件内容当作文本塞入上下文;先检查工具返回的结构化结论,需要进一步审阅时再用宿主具备的文档工具打开文件。失败和未验证项也必须交付,保持原始证据。 + + +## 精选任务与收藏复跑 + +`mobile_favorites(action="list")` 读取与首页相同的收藏。`create` 指定 `source_kind`(task、batch、exploration、exploration_case)和 `source_id`,从服务端原记录保存独立模板及执行配置;不传任意 snapshot,也不接受密钥。删除原任务或清理历史不影响收藏。 + +- task 收藏普通 Flash/Pro 任务;探索执行记录默认收藏已确认用例,无用例时保存为重新探索。 +- exploration 明确保存为新的探索对话,保留原探索模型与配置;不继续旧会话。 +- exploration_case 保存固定用例版本,可传 version;默认使用该版本之前的复跑配置,首次复跑默认 Flash 并明确标示。后来新版本不会覆盖已有收藏。 +- batch 保存一轮用例、原顺序、循环次数及执行策略;复跑不会把已展开的执行列表再循环。 + +`run` 必须提供稳定的 request_id,返回新 session_id、batch_id 或 exploration_id。连接异常时沿用相同 ID 和内容重试,不能新建 ID 自动重提。changes.task_config 仅覆盖本次;批次 changes.case_configs 使用从 0 开始的用例下标,changes.task_config 统一覆盖全批。`update` 才修改收藏默认配置,可改名称。`delete` 只取消收藏,不删除历史或执行结果。 + +不修改时使用收藏的原配置。旧任务未记录的字段会在 inherited_fields 标明采用收藏时的默认值,不能声称全部还原。原设备离线时先让用户选择设备,不自动换机。插件仍须通过当前安装、授权和资源校验。 + +单任务全部资料下载按任务合并并发请求,已结束且内容未变化时复用报告和 ZIP;人工复验、结果或附件变化会重新生成。批次复用成品,人工复验后重建,旧目录格式首次下载时更新。执行中导出为当前快照。录像位于 `任务录屏/安卓/`、`任务录屏/电脑/`、`任务录屏/人工示范/`,以文件清单和证据索引为准。 diff --git a/tests/unit/admin_console/test_automation.py b/tests/unit/admin_console/test_automation.py index 59fe93e..4a3a508 100644 --- a/tests/unit/admin_console/test_automation.py +++ b/tests/unit/admin_console/test_automation.py @@ -158,6 +158,7 @@ def test_completed_batch_reuses_existing_archive(client, isolated, monkeypatch): path = reports.EXPORT_ROOT / "batches" / batch_id / "批次报告.zip" path.parent.mkdir(parents=True) path.write_bytes(b"existing archive") + reports.remember(path, reports.BATCH_ARCHIVE_SIGNATURE) exporter = Mock(side_effect=AssertionError("archive must not be rebuilt")) monkeypatch.setattr(reports, "export_batch", exporter) base = f"/api/automation/batches/{batch_id}/archive" @@ -166,6 +167,21 @@ def test_completed_batch_reuses_existing_archive(client, isolated, monkeypatch): exporter.assert_not_called() +def test_legacy_batch_archive_is_rebuilt_with_current_recording_layout(client, isolated): + batch = service.create_batch(payload()) + batch_id = batch["id"] + isolated.put("batch", batch_id, {**batch, "status": "completed", "report_ready": True}) + path = reports.EXPORT_ROOT / "batches" / batch_id / "批次报告.zip" + path.parent.mkdir(parents=True) + path.write_bytes(b"legacy archive") + base = f"/api/automation/batches/{batch_id}/archive" + assert client.get(base + "/status").json()["status"] == "idle" + response = client.get(base) + assert response.status_code == 200 + assert zipfile.is_zipfile(BytesIO(response.content)) + assert client.get(base + "/status").json()["status"] == "ready" + + def test_preview_does_not_enqueue(client, isolated): response = client.post( "/api/automation/import/preview", @@ -1021,7 +1037,7 @@ def test_task_material_zip_contains_ordered_evidence_and_reports( assert files.index("1.1/步骤截图/0001_步骤0001_操作前.png") < files.index( "1.1/步骤截图/0002_步骤0002_操作后.png" ) - assert archive.read("1.1/任务录屏/0001_recording.mp4") == b"recording.mp4" + 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 manifest["artifacts"] == ["安卓截图.png"] @@ -1042,6 +1058,63 @@ def test_batch_zip_contains_each_tasks_android_artifacts(isolated): assert {archive.read(name).decode() for name in names} == {"任务0", "任务1"} +def test_all_material_cache_reuses_reports_and_invalidates_on_review_or_new_files( + isolated, monkeypatch +): + job = isolated.jobs(service.create_batch(payload())["id"])[0] + job["status"] = "completed" + isolated.save_job(job) + word = Mock(wraps=reports.word_report) + monkeypatch.setattr(reports, "word_report", word) + monkeypatch.setattr( + reports, + "diagnostics_json", + lambda sid: json.dumps({"requests": [], "collected_at": time.time()}), + ) + first = reports.export_session_media(job["session_id"], "all") + first_bytes = first.read_bytes() + reports.export_session_media(job["session_id"], "all") + assert word.call_count == 1 + assert first.read_bytes() == first_bytes + isolated.save_case_review(job["session_id"], "passed", "人工核对") + reports.export_session_media(job["session_id"], "all") + assert word.call_count == 2 + folder = reports.TRACES_PATH / job["session_id"] / "artifacts" / "android" + folder.mkdir(parents=True) + (folder / "新资料.txt").write_text("new evidence", encoding="utf-8") + reports.export_session_media(job["session_id"], "all") + with zipfile.ZipFile(first) as archive: + assert any(name.endswith("任务文件/安卓/新资料.txt") for name in archive.namelist()) + assert not json.loads( + archive.read(next(n for n in archive.namelist() if n.endswith("文件清单.json"))) + )["snapshot"] + # File-only changes rebuild the ZIP while keeping unchanged documents. + assert word.call_count == 2 + + +def test_different_tasks_export_concurrently_and_running_exports_are_snapshots( + isolated, monkeypatch +): + from concurrent.futures import ThreadPoolExecutor + + jobs = isolated.jobs(service.create_batch(payload())["id"]) + barrier = threading.Barrier(2) + + def generate(report, path): + barrier.wait(timeout=5) + path.write_bytes(b"generated report") + + monkeypatch.setattr(reports, "word_report", generate) + with ThreadPoolExecutor(max_workers=2) as pool: + exports = [pool.submit(reports.export_session, j["session_id"], ["docx"]) for j in jobs] + assert all(result.result(timeout=10)[0].is_file() for result in exports) + count = Mock(side_effect=lambda report, path: path.write_bytes(b"snapshot")) + monkeypatch.setattr(reports, "word_report", count) + for _ in range(2): + reports.export_session(jobs[0]["session_id"], ["docx"]) + assert count.call_count == 2 + + def test_android_artifact_export_does_not_follow_other_session_links(isolated): root = reports.TRACES_PATH / "session" / "artifacts" root.mkdir(parents=True) @@ -1078,8 +1151,8 @@ def test_video_export_accepts_finalized_worker_trace_and_rejects_external_path( with zipfile.ZipFile(archive_path) as archive: names = archive.namelist() assert names == [ - "任务录屏/0001_web_123_" + session_id[:8] + "_PASS_2026-09-28_recording.mp4", - "任务录屏/0002_resume_0001_recording.mp4", + "任务录屏/安卓/0001_web_123_" + session_id[:8] + "_PASS_2026-09-28_recording.mp4", + "任务录屏/安卓/0002_resume_0001_recording.mp4", "文件清单.json", ] assert archive.read(names[0]) == b"first segment" diff --git a/tests/unit/admin_console/test_favorites.py b/tests/unit/admin_console/test_favorites.py new file mode 100644 index 0000000..81c5285 --- /dev/null +++ b/tests/unit/admin_console/test_favorites.py @@ -0,0 +1,325 @@ +"""Favorite snapshots and transport retries never replay an old result or task.""" + +import asyncio +from copy import deepcopy +import json +from unittest.mock import AsyncMock + +from fastapi import FastAPI +from fastapi.testclient import TestClient +from pydantic import ValidationError +import pytest + +from apps.admin_console.services import automation_service +from apps.admin_console.services.automation_store import AutomationStore +from apps.admin_console.services.favorites import ( + FavoriteCreate, + FavoriteRun, + FavoriteService, + FavoriteUpdate, +) +from artemis.config.exploration import ExplorationConfig +from artemis.data_engine.storage import StorageManager +from artemis.exploration.models import CaseDraft +from artemis.exploration.store import ExplorationStore + + +@pytest.fixture +def favorites(tmp_path, monkeypatch): + storage = AutomationStore(tmp_path / "history.db") + StorageManager(storage.path, tmp_path / "traces") + monkeypatch.setattr(automation_service, "store", storage) + return FavoriteService(storage) + + +def task(service, session_id="task", **options): + config = { + "profile": "pro", + "model": "original-model", + "reasoning_effort": "high", + "device_serial": "original-device", + "record_video": True, + **options, + } + job = { + "session_id": session_id, + "status": "completed", + "case": {"name": "原用例", "goal": "检查真实页面"}, + "execution_config": { + "runtime_overrides": config, + "expected_output": "保留证据", + "enable_outputter": True, + }, + } + service.store.save_job(job) + return service.create(FavoriteCreate(source_id=session_id)) + + +def test_snapshot_survives_source_deletion_and_edits_do_not_modify_source(favorites): + item = task(favorites) + updated = favorites.update( + item["id"], FavoriteUpdate(name="常用检查", task_config={"model": "new-model"}) + ) + assert updated["snapshot"]["runtime_overrides"]["model"] == "new-model" + assert ( + favorites.store.job("task")["execution_config"]["runtime_overrides"]["model"] + == "original-model" + ) + favorites.store.delete_history() + assert favorites.get(item["id"])["snapshot"]["goal"] == "检查真实页面" + assert favorites.store.job("task") is None + + +@pytest.mark.asyncio +async def test_rerun_preserves_settings_and_is_durably_idempotent(favorites, monkeypatch): + item = task(favorites) + enqueue = AsyncMock(return_value={"tasks": [{"session_id": "new-task"}], "status": "queued"}) + monkeypatch.setattr("apps.admin_console.routers.tasks.run_task", enqueue) + value = FavoriteRun(request_id="same-submission", task_config={"reasoning_effort": "max"}) + first, second = await asyncio.gather( + favorites.run(item["id"], value), favorites.run(item["id"], value) + ) + assert first == second + enqueue.assert_awaited_once() + request = enqueue.call_args.args[0] + assert request.session_id != "task" + assert request.runtime_overrides.model == "original-model" + assert request.runtime_overrides.reasoning_effort == "max" + assert request.runtime_overrides.record_video is True + assert request.runtime_overrides.device_serial == "original-device" + assert request.expected_output == "保留证据" + assert favorites.get(item["id"])["snapshot"]["runtime_overrides"]["reasoning_effort"] == "high" + favorites.delete(item["id"]) + assert await FavoriteService(favorites.store).run(item["id"], value) == first + enqueue.assert_awaited_once() + with pytest.raises(ValueError, match="request_id"): + await favorites.run( + item["id"], + FavoriteRun(request_id="same-submission", task_config={"model": "different"}), + ) + + +@pytest.mark.asyncio +async def test_lost_admission_reply_reuses_snapshot_and_session(favorites, monkeypatch): + item = task(favorites) + enqueue = AsyncMock(side_effect=[ConnectionError("reply lost"), {"tasks": [{}]}]) + monkeypatch.setattr("apps.admin_console.routers.tasks.run_task", enqueue) + value = FavoriteRun(request_id="retry") + with pytest.raises(ConnectionError): + await favorites.run(item["id"], value) + favorites.update(item["id"], FavoriteUpdate(task_config={"model": "changed"})) + await FavoriteService(favorites.store).run(item["id"], value) + requests = [call.args[0] for call in enqueue.call_args_list] + assert requests[0].session_id == requests[1].session_id + assert requests[1].runtime_overrides.model == "original-model" + + +@pytest.mark.asyncio +async def test_batch_favorite_keeps_one_round_and_original_repetitions(favorites): + batch = automation_service.create_batch( + { + "name": "压测", + "repeat_count": 3, + "profile": "flash", + "device_serial": "device", + "cases": [ + {"name": "甲", "goal": "检查甲", "task_config": {"model": "model-a"}}, + { + "name": "乙", + "goal": "检查乙", + "task_config": {"profile": "pro", "model": "model-b"}, + }, + ], + } + ) + item = favorites.create(FavoriteCreate(source_kind="batch", source_id=batch["id"])) + assert len(item["snapshot"]["cases"]) == 2 + assert item["snapshot"]["repeat_count"] == 3 + result = await favorites.run( + item["id"], + FavoriteRun( + request_id="batch-rerun", + task_config={"reasoning_effort": "max"}, + case_configs={1: {"model": "changed-b"}}, + ), + ) + jobs = favorites.store.jobs(result["batch_id"]) + assert len(jobs) == 6 + assert [j["case"]["name"] for j in jobs] == ["甲", "乙"] * 3 + assert [j["case"]["task_config"]["model"] for j in jobs] == ["model-a", "changed-b"] * 3 + assert all(j["case"]["task_config"]["reasoning_effort"] == "max" for j in jobs) + assert favorites.store.get("batch", result["batch_id"])["status"] == "running" + again = await favorites.run( + item["id"], + FavoriteRun( + request_id="batch-rerun", + task_config={"reasoning_effort": "max"}, + case_configs={1: {"model": "changed-b"}}, + ), + ) + assert again == result + assert len(favorites.store.records("batch")) == 2 + + +def test_exploration_case_pins_version_without_inheriting_expensive_model(favorites): + repo = ExplorationStore(favorites.store) + session = repo.create( + "探索目标", + ExplorationConfig(model="exploration-model", reasoning_effort="max"), + {"model": "exploration-model", "device_serial": "device"}, + "create", + ) + repo.save_case( + session["id"], + CaseDraft(name="完成用例", goal="验证主页", preconditions="已登录"), + 0, + "case-v1", + ) + item = favorites.create(FavoriteCreate(source_kind="exploration_case", source_id=session["id"])) + assert item["case_version"] == 1 + assert item["snapshot"]["goal"] == "前置条件:\n已登录\n\n验证主页" + assert item["snapshot"]["runtime_overrides"]["profile"] == "flash" + assert item["snapshot"]["runtime_overrides"]["model"] != "exploration-model" + repo.save_case(session["id"], CaseDraft(name="新用例", goal="新目标"), 1, "case-v2") + assert favorites.get(item["id"])["snapshot"]["goal"].endswith("验证主页") + exploration = favorites.create( + FavoriteCreate(source_kind="exploration", source_id=session["id"]) + ) + assert exploration["snapshot"]["config"]["model"] == "exploration-model" + + +def test_old_task_recovers_real_model_and_marks_missing_settings(favorites): + with favorites.store.connection() as conn: + conn.execute( + "INSERT INTO sessions(session_id,initial_goal,status,device_info) VALUES (?,?,?,?)", + ( + "legacy", + "检查浏览器", + "completed", + json.dumps({"profile": "flash", "serial": "computer:local"}), + ), + ) + conn.execute( + "INSERT INTO traces(trace_id,session_id,type,timestamp,payload) VALUES (?,?,?,?,?)", + ( + "request", + "legacy", + "llm_diagnostic", + 1, + json.dumps( + {"role": "FlashRunner", "model": "actual-model", "reasoning_effort": "max"} + ), + ), + ) + conn.commit() + item = favorites.create(FavoriteCreate(source_id="legacy")) + options = item["snapshot"]["runtime_overrides"] + assert options["model"] == "actual-model" + assert options["reasoning_effort"] == "max" + assert options["execution_target"] == "computer" + assert "model" not in item["inherited_fields"] + assert "recording" in item["inherited_fields"] + + +def test_api_rejects_credentials_and_arbitrary_snapshots(favorites, monkeypatch): + from apps.admin_console.routers import favorites as api + + monkeypatch.setattr(api, "favorites", favorites) + app = FastAPI() + app.include_router(api.router) + client = TestClient(app) + item = task(favorites) + assert len(client.get("/api/automation/favorites").json()) == 1 + assert ( + client.patch( + f"/api/automation/favorites/{item['id']}", json={"task_config": {"api_key": "never"}} + ).status_code + == 422 + ) + assert ( + client.post( + "/api/automation/favorites", json={"source_id": "task", "snapshot": {}} + ).status_code + == 422 + ) + assert ( + client.patch(f"/api/automation/favorites/{item['id']}", json={"name": " "}).status_code + == 409 + ) + assert client.delete(f"/api/automation/favorites/{item['id']}").status_code == 200 + assert favorites.store.job("task") + with pytest.raises(ValidationError): + FavoriteRun(request_id="run", task_config={"password": "never"}) + + +def test_collection_defaults_cannot_modify_original_snapshot(favorites): + item = task(favorites) + original = deepcopy(item) + favorites.updated(item, FavoriteUpdate(task_config={"model": "temporary"})) + assert item == original + + +def test_partial_recording_update_keeps_other_favorite_settings(favorites): + item = task(favorites, recording={"codec": "h264", "fps": 10, "max_long_edge": 720}) + updated = favorites.updated(item, FavoriteUpdate(task_config={"recording": {"codec": "h265"}})) + assert updated["snapshot"]["runtime_overrides"]["recording"] == { + "codec": "h265", + "fps": 10, + "max_long_edge": 720, + } + assert item["snapshot"]["runtime_overrides"]["recording"]["codec"] == "h264" + + +def test_legacy_empty_recording_configuration_uses_current_defaults(): + from apps.admin_console.services.execution_config import execution_config + from artemis.config.agent import load_agent_config + + snapshot = execution_config({"profile": "flash", "runtime_overrides": {"recording": None}}) + assert snapshot["runtime_overrides"]["recording"] == load_agent_config().recording.model_dump() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("rejected", [False, True]) +async def test_restart_exploration_creates_new_conversation_and_reports_rejection( + favorites, monkeypatch, rejected +): + from apps.admin_console.services.exploration_service import ExplorationService + + repo = ExplorationStore(favorites.store) + original = repo.create( + "探索目标", + ExplorationConfig(model="explore-model", reasoning_effort="max"), + {"device_serial": "device", "record_video": True}, + "source", + ) + item = favorites.create(FavoriteCreate(source_kind="exploration", source_id=original["id"])) + monkeypatch.setattr( + "apps.admin_console.services.exploration_service.exploration_service", + ExplorationService(repo), + ) + enqueue = AsyncMock( + return_value=( + {"status": "rejected", "error": "设备离线"} + if rejected + else {"status": "queued", "tasks": [{}]} + ) + ) + monkeypatch.setattr("apps.admin_console.routers.tasks.run_task", enqueue) + value = FavoriteRun(request_id="new-exploration", task_config={"model": "selected-model"}) + if rejected: + with pytest.raises(ValueError, match="设备离线"): + await favorites.run(item["id"], value) + with pytest.raises(ValueError, match="设备离线"): + await favorites.run(item["id"], value) + else: + result = await favorites.run(item["id"], value) + assert result["exploration_id"] != original["id"] + assert repo.run(result["session_id"])["exploration_id"] == result["exploration_id"] + assert await favorites.run(item["id"], value) == result + request = enqueue.call_args.args[0] + assert request.conversation_id != original["id"] + assert request.runtime_overrides.model == "selected-model" + assert request.runtime_overrides.reasoning_effort == "max" + assert request.runtime_overrides.record_video is True + enqueue.assert_awaited_once() + assert repo.get(original["id"]) == original diff --git a/tests/unit/mcp/test_automation_tools.py b/tests/unit/mcp/test_automation_tools.py index 56a75af..aa30a0c 100644 --- a/tests/unit/mcp/test_automation_tools.py +++ b/tests/unit/mcp/test_automation_tools.py @@ -7,6 +7,39 @@ import pytest from mcp_server.tools import automation as tools +@pytest.mark.asyncio +async def test_favorites_share_backend_preserve_request_id_and_refuse_secrets(monkeypatch): + request = AsyncMock(return_value={"id": "favorite"}) + monkeypatch.setattr(tools, "_request", request) + await tools.mobile_favorites("create", source_kind="batch", source_id="source") + request.assert_awaited_once_with( + "POST", + "/api/automation/favorites", + json={"source_kind": "batch", "source_id": "source", "version": None}, + ) + request.reset_mock() + with pytest.raises(ValueError): + await tools.mobile_favorites("run", favorite_id="favorite") + with pytest.raises(ValueError, match="密钥"): + await tools.mobile_favorites( + "update", + favorite_id="favorite", + changes={"case_configs": {"0": {"api_key": "never-send"}}}, + ) + request.assert_not_awaited() + await tools.mobile_favorites( + "run", + favorite_id="favorite", + request_id="stable", + changes={"task_config": {"model": "original"}}, + ) + request.assert_awaited_once_with( + "POST", + "/api/automation/favorites/favorite/run", + json={"task_config": {"model": "original"}, "request_id": "stable"}, + ) + + @pytest.mark.asyncio async def test_config_requires_explicit_persistence_and_excludes_secrets(monkeypatch): request = AsyncMock( diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 80e29a7..c4a7350 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -233,6 +233,7 @@ def test_cli_mcp_generate_config_codex(): "mobile_config", "mobile_task_file", "mobile_batch", + "mobile_favorites", "mobile_exploration", "mobile_search_history", "mobile_export_report", @@ -278,6 +279,7 @@ def test_cli_mcp_install_codex_preserves_config_and_is_idempotent(tmp_path, monk "mobile_config", "mobile_task_file", "mobile_batch", + "mobile_favorites", "mobile_exploration", "mobile_search_history", "mobile_export_report",