"""Web and MCP use the same exploration admission and conversation service.""" from artemis.config.agent import load_agent_config from artemis.config.exploration import ExplorationConfig from artemis.exploration.case_prompt import require_resolved_case from artemis.exploration.models import CreateExploration, SaveCaseRequest, ValidateCaseRequest from artemis.exploration.demonstrations import Demonstrations from artemis.exploration.store import ExplorationConflict, ExplorationStore, explorations from artemis.runtime.task_control import add_instruction, read_state, request_pause, request_resume from apps.admin_console.services.task_import import TaskOptions def exploration_case_prompt(case: dict) -> str: return require_resolved_case(case) class ExplorationService: def __init__(self, repository: ExplorationStore | None = None): self.repository = repository if repository is not None else explorations def settings(self) -> dict: saved = self.repository.database.get("settings", "exploration") or {} return ExplorationConfig.model_validate( { **load_agent_config().exploration.model_dump(), **saved, } ).model_dump() def save_settings(self, value: ExplorationConfig) -> dict: return self.repository.database.put("settings", "exploration", value.model_dump()) def create(self, value: CreateExploration, *, source: str = "web") -> dict: options = TaskOptions.model_validate(value.task_config).model_dump(exclude_none=True) return self.repository.create( value.goal, value.config or ExplorationConfig.model_validate(self.settings()), options, value.request_id, source=source, ) def detail(self, session_id: str, *, after: int = 0) -> dict: session = self.repository.get(session_id) runs = self.repository.runs(session_id) version = self.repository.case(session_id) with self.repository.database.connection() as connection: import json generations = [ json.loads(row[0]) for row in connection.execute( "SELECT payload FROM automation_records WHERE kind='exploration_generation' " "AND json_extract(payload,'$.exploration_id')=? ORDER BY updated_at DESC LIMIT 20", (session_id,), ).fetchall() ] from artemis.runtime.case_result import resolve_case_result for run in runs: outcome = run.get("outcome") verdict = resolve_case_result(outcome, run["status"]) if outcome else None run["test_result"] = verdict.model_dump() if verdict else None return { **session, "messages": self.repository.messages(session_id, after=after), "runs": runs, "case": version, "generations": [ {k: v for k, v in g.items() if k not in {"owner", "owner_pid", "lease_until"}} for g in generations ], "demonstrations": Demonstrations(self.repository).list(session_id), "validation_runs": [ r for r in runs if r["kind"] == "validation" and r["case_version"] == session["case_version"] ], "control": read_state(session["active_run_id"]) if session["active_run_id"] else {}, } async def _admit(self, run: dict, *, source: str) -> dict: # RunRequest performs the same readiness and idempotency checks as /api/run. from apps.admin_console.routers.tasks import run_task from apps.admin_console.schemas.task_schema import RunRequest from fastapi import HTTPException current = self.repository.run(run["id"]) if current and current.get("outcome") is not None: return current options = TaskOptions.model_validate(run["task_config"]) try: result = await run_task( RunRequest( goal=run["goal"], profile=options.profile or "pro", session_id=run["id"], ingress=source, conversation_id=run["exploration_id"], runtime_overrides=options, ) ) except HTTPException as exc: if exc.status_code < 500: self.repository.finish_run( run["id"], {"status": "rejected", "error": str(exc.detail)} ) raise if result.get("status") in {"error", "rejected"}: self.repository.finish_run( run["id"], { "status": "rejected", "error": result.get("error") or result.get("message") or "任务入队失败", }, ) else: self.repository.set_run_state(run["id"], "queued", expected_status="reserved") return {"run": self.repository.run(run["id"]), "admission": result} async def start(self, session_id: str, request_id: str, *, source: str = "web") -> dict: session = self.repository.get(session_id) config = ExplorationConfig.model_validate(session["config"]) options = { **session["task_config"], "profile": "pro", "model": config.model, "reasoning_effort": config.reasoning_effort, "timeout_seconds": config.timeout_seconds, "verification_level": "final", } run = self.repository.reserve_run( session_id, request_id, kind="exploration", goal=session["goal"], task_config=options, ) return await self._admit(run, source=source) async def validate( self, session_id: str, value: ValidateCaseRequest, *, source: str = "web" ) -> dict: case = self.repository.case(session_id, value.version) if case is None: raise KeyError("用例版本不存在") session = self.repository.get(session_id) # Device/plugin binding may be reused; exploration's expensive model is not. options = { k: v for k, v in session["task_config"].items() if k not in {"model", "reasoning_effort", "profile", "timeout_seconds"} } options.update(TaskOptions.model_validate(value.task_config).model_dump(exclude_none=True)) options["profile"] = value.profile run = self.repository.reserve_run( session_id, value.request_id, kind="validation", goal=exploration_case_prompt(case), task_config=options, case_version=value.version, ) return await self._admit(run, source=source) def message( self, session_id: str, text: str, request_id: str, *, source: str = "web", expected_run_id: str | None = None, ) -> dict: session = self.repository.get(session_id) if expected_run_id and session["active_run_id"] != expected_run_id: raise ExplorationConflict("所选执行已结束，指导未发送，请刷新任务状态") question = session.get("pending_question") message = self.repository.add_message( session_id, text, request_id, source=source, metadata={"question_id": question, "run_id": session["active_run_id"]}, ) question = message.get("question_id") run_id = message.get("run_id") if run_id and run_id == session["active_run_id"]: run = self.repository.run(run_id) if run and run.get("outcome") is None: add_instruction( run_id, message["text"], source=source, instruction_id=message["id"] ) self.repository.answer_question(session_id, question) if read_state(run_id).get("status") == "task_paused": request_resume(run_id) return message def pause(self, session_id: str) -> dict: session = self.repository.get(session_id) if session["status"] == "finalizing": raise ExplorationConflict("设备操作已结束，正在提炼用例，无需暂停。") if not session["active_run_id"]: raise ExplorationConflict("当前没有正在执行的探索") request_pause(session["active_run_id"], reason="user") return {"status": "pausing", "id": session_id} def stop(self, session_id: str) -> dict: from apps.admin_console.services.task_queue_service import TaskQueueService session = self.repository.get(session_id) if run_id := session["active_run_id"]: requested = TaskQueueService.stop_tasks(session_id=run_id) return {"status": "stopping" if requested else "not_running", "id": session_id} return {"status": session["status"], "id": session_id} async def resume(self, session_id: str, request_id: str, *, source: str = "web") -> dict: repo = self.repository with repo._transaction() as connection: session = repo._require(connection, session_id) def prepare(): if session.get("pending_question"): raise ExplorationConflict("请先回答探索助手的问题，再继续执行。") if run_id := session["active_run_id"]: if read_state(run_id).get("status") != "task_paused": raise ExplorationConflict("请等待探索安全暂停后再继续") request_resume(run_id) return {"status": "resuming", "id": session_id, "run_id": run_id} return {"status": "start", "id": session_id} result = repo._operation(connection, f"{session_id}:continue", request_id, {}, prepare) if result["status"] == "resuming": return result return await self.start(session_id, request_id, source=source) def save_case(self, session_id: str, value: SaveCaseRequest) -> dict: return self.repository.save_case( session_id, value.case, value.expected_version, value.request_id ) def demonstrate(self, session_id: str, request_id: str, source: str | None = None) -> dict: session = self.repository.get(session_id) if session["status"] == "finalizing": raise ExplorationConflict("正在提炼用例，请完成后继续探索再示范。") source = source or ( "computer" if session["task_config"].get("execution_target") == "computer" else "android" ) if source not in {"android", "computer", "both"}: raise ValueError("无效的示范来源") demo = Demonstrations(self.repository).request(session_id, source, request_id) if demo["status"] == "requested": request_pause(demo["run_id"], reason="demonstration") if read_state(demo["run_id"]).get("status") == "task_paused": request_resume(demo["run_id"]) return demo def demonstration_command( self, session_id: str, demo_id: str, command: str, request_id: str, *, seconds: int = 0 ) -> dict: return Demonstrations(self.repository).command( session_id, demo_id, command, request_id, seconds=seconds ) async def generate_case(self, session_id: str, value) -> dict: from artemis.exploration.synthesis import generate_case, _generation from apps.admin_console.services.automation_service import spawn generation = _generation( self.repository, session_id, value.request_id, value.source_run_id, value.expected_version, ) if generation["status"] == "completed": return {"status": "completed", "id": generation["id"], "case": generation["case"]} async def work(): try: await generate_case( self.repository, session_id, value.request_id, value.source_run_id, value.expected_version, generation=generation, ) except Exception: import logging logging.getLogger(__name__).exception("Exploration synthesis request failed") spawn(work()) return {"status": "running", "id": generation["id"]} def create_case_batch(self, session_id: str, value) -> dict: from apps.admin_console.routers.automation import BatchRequest from apps.admin_console.services.automation_service import create_batch from artemis.exploration.store import _digest case = self.repository.case(session_id, value.version) if not case: raise KeyError("用例版本不存在") session = self.repository.get(session_id) options = { k: v for k, v in session["task_config"].items() if k not in {"model", "reasoning_effort", "profile", "timeout_seconds"} } options.update(TaskOptions.model_validate(value.task_config).model_dump(exclude_none=True)) options["profile"] = value.profile payload = BatchRequest( name=value.name, profile=value.profile, request_id="explore-" + _digest([session_id, value.request_id]), cases=[ { "name": case["name"], "goal": exploration_case_prompt(case), "preconditions": case["preconditions"], "case_id": f"{session_id}-v{value.version}", "task_config": options, "category": "探索用例", "subcategory": f"版本{value.version}", } ], ).model_dump() return create_batch(payload) exploration_service = ExplorationService()