From e0e1d31cb1e1576ff1352ff5941584a0c4db6234 Mon Sep 17 00:00:00 2001 From: AirLongDian Date: Wed, 10 Jun 2026 17:16:15 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20P1-20=20UI=20=E4=BB=BB=E5=8A=A1=E6=A3=80?= =?UTF-8?q?=E6=B5=8B=E6=B7=BB=E5=8A=A0=E4=B8=AD=E6=96=87=E5=85=B3=E9=94=AE?= =?UTF-8?q?=E8=AF=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 中文关键词: 界面, UI, 界面设计, 前端, 界面开发, 按钮, 对话框, 窗口, 菜单, 控件, 渲染, 布局, 登录界面, 界面组件 Co-Authored-By: Claude Sonnet 4.6 --- do_mode.py | 260 +++++++++++++++++++++++++++++++++++++++++++++++ test_p1_19_20.py | 152 +++++++++++++++++++++++++++ 2 files changed, 412 insertions(+) create mode 100644 do_mode.py create mode 100644 test_p1_19_20.py diff --git a/do_mode.py b/do_mode.py new file mode 100644 index 0000000..9bc7d7d --- /dev/null +++ b/do_mode.py @@ -0,0 +1,260 @@ +""" +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", + # 中文关键词 + "界面", "UI", "界面设计", "前端", "界面开发", + "按钮", "对话框", "窗口", "菜单", "控件", + "渲染", "布局", "登录界面", "界面组件", +) + + +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']}") diff --git a/test_p1_19_20.py b/test_p1_19_20.py new file mode 100644 index 0000000..9c17eae --- /dev/null +++ b/test_p1_19_20.py @@ -0,0 +1,152 @@ +#!/usr/bin/env python3 +"""P1-19 P1-20 功能测试""" + +import sys +import os +import tempfile +from pathlib import Path + +# 添加 lib 路径 +sys.path.insert(0, str(Path(__file__).parent / "lib")) + +def test_task_graph_test_required(): + """P1-19.1: 测试 TaskNode 的 test_required 字段""" + from air_runtime.task_graph import TaskNode + + node = TaskNode(id="T-001", task="test", test_required=True) + assert node.test_required == True, "test_required should be True" + + node2 = TaskNode(id="T-002", task="test2") + assert node2.test_required == False, "test_required should default to False" + + # 测试序列化/反序列化 + data = {"id": "T-003", "testRequired": True, "status": "TODO", "task": "test", "filesDirs": "", "doneWhen": "", "inDegree": 0, "outEdges": [], "writeSet": []} + node3 = TaskNode( + id=data["id"], status=data["status"], task=data["task"], + files_dirs=data["filesDirs"], done_when=data["doneWhen"], + test_required=data["testRequired"] + ) + assert node3.test_required == True, "test_required should deserialize correctly" + + print("✓ P1-19.1 test_required 字段测试通过") + return True + + +def test_arc_boundary_test_injection(): + """P1-19.1: 测试边界测试任务注入""" + from air_runtime.modes.arc_mode import _inject_boundary_tests, _build_graph_from_todo + from air_runtime.task_graph import TaskGraph, TaskNode + from air_runtime.todo_parser import TodoTask + + # 创建模拟 tasks + tasks = [ + TodoTask(task_id="T-001", task="实现模块A", files_dirs="src/moduleA/", status="TODO"), + TodoTask(task_id="T-002", task="实现模块B", files_dirs="src/moduleB/", status="TODO"), + ] + + graph = TaskGraph() + for t in tasks: + graph.add_node(TaskNode(id=t.task_id, task=t.task, files_dirs=t.files_dirs)) + + # 注入边界测试 + _inject_boundary_tests(graph, tasks) + + # 检查是否生成了测试任务 + test_nodes = [n for n in graph.nodes.values() if n.test_required] + assert len(test_nodes) >= 2, f"应该生成至少2个测试任务，实际: {len(test_nodes)}" + + # 检查测试任务的 done_when + for node in test_nodes: + assert "测试通过" in node.done_when, f"测试任务的 done_when 应包含'测试通过': {node.done_when}" + + print(f"✓ P1-19.1 边界测试注入测试通过，生成了 {len(test_nodes)} 个测试任务") + return True + + +def test_review_runtime_high_risk_audit(): + """P1-19.2: 测试高风险审计数据结构""" + from air_runtime.review_runtime import ReviewReport, HighRiskAudit, HighRiskFinding + + # 创建带 highRiskAudit 的报告 + finding = HighRiskFinding(file="src/test.cpp", line=100, severity="high", issue="空指针风险") + hra = HighRiskAudit( + lifecycle=[finding], + nullPointer=[finding], + overallRisk="high", + deliveryVerdict="needs-fix" + ) + + report = ReviewReport( + task_id="T-001", + verdict="pass", + high_risk_audit=hra + ) + + assert report.high_risk_audit.overallRisk == "high" + assert report.high_risk_audit.deliveryVerdict == "needs-fix" + assert len(report.high_risk_audit.lifecycle) == 1 + + # 测试序列化 + from air_runtime.review_runtime import ReviewRuntime + d = ReviewRuntime._report_to_dict(report) + assert d["highRiskAudit"]["overallRisk"] == "high" + assert d["highRiskAudit"]["deliveryVerdict"] == "needs-fix" + + # 测试反序列化 + report2 = ReviewRuntime._dict_to_report(d) + assert report2.high_risk_audit.deliveryVerdict == "needs-fix" + + print("✓ P1-19.2 highRiskAudit 测试通过") + return True + + +def test_do_ui_task_detection(): + """P1-20: 测试 UI 任务检测""" + from air_runtime.modes.do_mode import is_ui_task, route_ui_task + + # UI 任务检测 + assert is_ui_task("实现登录界面UI") == True, "应该检测为 UI 任务" + assert is_ui_task("实现按钮组件") == True, "应该检测为 UI 任务" + assert is_ui_task("React 前端开发") == True, "应该检测为 UI 任务" + assert is_ui_task("实现后端API") == False, "不应该检测为 UI 任务" + assert is_ui_task("数据库优化") == False, "不应该检测为 UI 任务" + + print("✓ P1-20 is_ui_task 测试通过") + return True + + +def main(): + print("=" * 50) + print("P1-19 P1-20 功能测试") + print("=" * 50) + + tests = [ + ("P1-19.1 test_required 字段", test_task_graph_test_required), + ("P1-19.1 边界测试注入", test_arc_boundary_test_injection), + ("P1-19.2 highRiskAudit", test_review_runtime_high_risk_audit), + ("P1-20 UI 任务检测", test_do_ui_task_detection), + ] + + passed = 0 + failed = 0 + + for name, test_fn in tests: + try: + test_fn() + passed += 1 + except Exception as e: + print(f"✗ {name} 失败: {e}") + import traceback + traceback.print_exc() + failed += 1 + + print("=" * 50) + print(f"测试结果: {passed} 通过, {failed} 失败") + print("=" * 50) + + return failed == 0 + + +if __name__ == "__main__": + success = main() + sys.exit(0 if success else 1) \ No newline at end of file