""" AirDo mode — V2 任务执行器。 V2 改进：强制 AirDbg 路由（L1 代码级），task_id 注入防护，UI 任务 frontend-design Skill 路由（P1-20）。 """ from __future__ import annotations import subprocess import sys from pathlib import Path from air_runtime.io import atomic_json_write, safe_json_load from air_runtime.paths import airplan_root, event_log_path from air_runtime.events import EventLog, TASK_COMPLETED, TASK_BLOCKED from air_runtime.contracts import WorkerResult, now_iso from air_runtime.utils import sanitize_task_id, session_stamp # P1-20: UI 任务检测关键词 UI_TASK_INDICATORS = ( "gui", "ui", "render", "layout", "dialog", "osd", "overlay", "visual", "screenshot", "display", "widget", "pane", "toolbar", "settings_dialog", "canvas", "button", "window", "popup", "menu", "drm", "kms", "opengl", "vulkan", "frontend", "react", "vue", "angular", "web", "css", "html", ) def is_ui_task(task_text: str) -> bool: """P1-20: 检测任务是否涉及 UI/前端界面设计。""" text = task_text.lower() return any(kw in text for kw in UI_TASK_INDICATORS) def ensure_frontend_design_skill() -> bool: """P1-20: 检测 frontend-design Skill 是否存在，不存在则尝试自动安装。""" # 检查 skill 是否已安装（检查 ~/.claude/skills/frontend-design 或类似路径） import os home = Path.home() skill_path = home / ".claude" / "skills" / "frontend-design" if skill_path.exists(): return True # 尝试自动安装 import logging logging.info("frontend-design skill not found, attempting auto-install...") try: result = subprocess.run( ["claude", "plugin", "install", "frontend-design"], capture_output=True, text=True, timeout=60, ) if result.returncode == 0: logging.info("frontend-design skill installed successfully") return True logging.warning("frontend-design skill install failed: %s", result.stderr) except Exception as e: logging.warning("frontend-design skill install error: %s", e) return False def route_ui_task(task_text: str, task_id: str) -> dict: """P1-20: UI 任务路由决策。检测 UI 任务并确保 frontend-design Skill 可用。""" if not is_ui_task(task_text): return {"target": "execute", "skill": None, "is_ui_task": False} # 是 UI 任务，检查 skill 可用性 if ensure_frontend_design_skill(): return {"target": "execute", "skill": "frontend-design", "is_ui_task": True} # Skill 不可用，阻止任务 return { "target": "blocked", "reason": "UI task requires frontend-design skill but installation failed", "skill": "frontend-design", "is_ui_task": True, } def _paths(project_root: Path, task_id: str) -> dict[str, Path]: root = airplan_root(project_root) / "state" / "airdo" task_dir = root / "tasks" / task_id return { "root": root, "state": root / "state.json", "task_dir": task_dir, "brief": task_dir / "brief.md", "handoff": task_dir / "subagent-handoff.md", "result": task_dir / "result.json", "worker_state": task_dir / "worker-state.json", } def _ensure_dirs(paths: dict[str, Path]) -> None: paths["task_dir"].mkdir(parents=True, exist_ok=True) def enter_worker(project_root: Path, task_id: str, task_text: str = "") -> dict: """P1-20: 新增 task_text 参数用于 UI 任务检测。""" tid = sanitize_task_id(task_id) paths = _paths(project_root, tid) _ensure_dirs(paths) # P1-20: UI 任务检测和路由 ui_routing = {"target": "execute", "skill": None, "is_ui_task": False} if task_text: ui_routing = route_ui_task(task_text, tid) if ui_routing.get("target") == "blocked": # UI 任务但 skill 不可用，阻止执行 worker_state = { "taskId": tid, "status": "blocked", "enteredAt": now_iso(), "resultPath": str(paths["result"]), "blockReason": ui_routing.get("reason", "frontend-design skill unavailable"), "uiRouting": ui_routing, } atomic_json_write(paths["worker_state"], worker_state) atomic_json_write(paths["state"], {"enabled": True, "activeTaskId": tid, "updatedAt": now_iso(), "blocked": True}) log = EventLog(event_log_path(project_root)) log.emit("task.blocked", {"taskId": tid, "reason": ui_routing.get("reason")}) return { "taskId": tid, "status": "blocked", "blockReason": ui_routing.get("reason"), "uiRouting": ui_routing, } worker_state = { "taskId": tid, "status": "implementing", "enteredAt": now_iso(), "resultPath": str(paths["result"]), "uiRouting": ui_routing, } atomic_json_write(paths["worker_state"], worker_state) atomic_json_write(paths["state"], {"enabled": True, "activeTaskId": tid, "updatedAt": now_iso()}) log = EventLog(event_log_path(project_root)) log.emit("task.entered", {"taskId": tid, "uiRouting": ui_routing}) return { "taskId": tid, "briefPath": str(paths["brief"]), "handoffPath": str(paths["handoff"]), "resultPath": str(paths["result"]), "workerStatePath": str(paths["worker_state"]), "uiRouting": ui_routing, } def finish_worker(project_root: Path, task_id: str, result_path: Path | None = None) -> dict: """V2 核心改进：强制 AirDbg 路由。""" tid = sanitize_task_id(task_id) paths = _paths(project_root, tid) # 加载 result if result_path and result_path.exists(): result_data = safe_json_load(result_path) elif paths["result"].exists(): result_data = safe_json_load(paths["result"]) else: result_data = {"taskId": tid, "status": "blocked", "summary": "no result found"} if not isinstance(result_data, dict): result_data = {"taskId": tid, "status": "blocked"} result = WorkerResult.from_dict(result_data) status = result.status # V2 L1 代码级：done 但无证据 → 强制 AirDbg if status == "done": if not result.validations and not result.files_changed: routing_decision = { "target": "airdbg", "reason": "done without evidence — mandatory debug review", "forced": True, } else: routing_decision = {"target": "merge", "forced": False} # V2 L1 代码级：blocked/failed → 强制 AirDbg elif status in ("blocked", "failed"): routing_decision = { "target": "airdbg", "reason": f"status={status} — AirDbg mandatory before return", "forced": True, } else: routing_decision = {"target": "merge", "forced": False} # 持久化 finalized = result.to_dict() finalized["routingDecision"] = routing_decision finalized["finalizedAt"] = now_iso() atomic_json_write(paths["result"], finalized) atomic_json_write(paths["worker_state"], {"taskId": tid, "status": "finished", "resultPath": str(paths["result"]), "routingDecision": routing_decision}) log = EventLog(event_log_path(project_root)) log.emit("task.finished", {"taskId": tid, "status": status, "routingTarget": routing_decision["target"]}) # emit task.completed / task.blocked based on final status if status == "done": log.emit(TASK_COMPLETED, {"taskId": tid, "routingTarget": routing_decision["target"]}) elif status in ("blocked", "failed"): log.emit(TASK_BLOCKED, {"taskId": tid, "status": status}) return { "taskId": tid, "status": status, "finalizedResultPath": str(paths["result"]), "workerStatePath": str(paths["worker_state"]), "routingDecision": routing_decision, } def status_worker(project_root: Path) -> dict: paths = _paths(project_root, "_") state = safe_json_load(paths["state"]) or {} task_ids = [] if paths["root"].joinpath("tasks").exists(): task_ids = [d.name for d in paths["root"].joinpath("tasks").iterdir() if d.is_dir()] return { "enabled": state.get("enabled", False), "activeTaskId": state.get("activeTaskId", ""), "taskIds": task_ids, } def main(args) -> None: project_root = Path(args.project).expanduser().resolve() sub = args.sub or "status" tid = args.task_id if sub == "status": s = status_worker(project_root) print("airplan_mode=do") print(f"enabled={s['enabled']}") print(f"active_task_id={s['activeTaskId']}") print(f"known_tasks={','.join(s['taskIds'])}") elif sub == "enter": result = enter_worker(project_root, tid) print("airplan_mode=do") print(f"task_id={result['taskId']}") print(f"brief_path={result['briefPath']}") print(f"result_path={result['resultPath']}") print(f"worker_state_path={result['workerStatePath']}") elif sub == "finish": rpath = Path(args.result).expanduser().resolve() if args.result else None finalized = finish_worker(project_root, tid, rpath) print("airplan_mode=do") print(f"task_id={finalized['taskId']}") print(f"status={finalized['status']}") print(f"routing_target={finalized['routingDecision']['target']}") print(f"routing_forced={finalized['routingDecision']['forced']}")