fix: P1-20 UI 任务检测添加中文关键词

中文关键词: 界面, UI, 界面设计, 前端, 界面开发, 按钮, 对话框, 窗口, 菜单, 控件, 渲染, 布局, 登录界面, 界面组件

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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']}")

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)