#!/usr/bin/env python3 """ AirArc mode — V2 架构规划器。 L1 代码级保障：allowed-tools 限制为只读。 产出 execution-plan.json（完整 DAG）+ plan-delta.json（增量重规划）。 """ from __future__ import annotations import json import sys from pathlib import Path from air_runtime.contracts import now_iso from air_runtime.io import atomic_json_write, safe_json_load from air_runtime.paths import airplan_root, todo_path as get_todo_path from air_runtime.review import build_parallel_review, render_review_markdown from air_runtime.task_graph import TaskGraph, TaskNode, Edge, PlanDelta from air_runtime.events import EventLog from air_runtime.paths import event_log_path from air_runtime.utils import ordered_unique # INV-16: 弱模型安全 — 危险词列表 DANGEROUS_TASK_WORDS = [ "清理", "清除", "删除所有", "重构整个", "重写全部", "clear", "delete all", "remove all", "rewrite entire", "refactor whole", ] DIR_LEVEL_FILE_SCOPE_PATTERNS = [ "src/", "lib/", "include/", "tests/", "modules/", ] def _audit_task_safety(task_id: str, task_text: str, files_dirs: str, done_when: str) -> list[str]: """INV-16: 扫描单个任务的危险词和宽泛文件范围，返回警告列表。""" warnings = [] text_lower = task_text.lower() for word in DANGEROUS_TASK_WORDS: if word.lower() in text_lower: warnings.append(f"[{task_id}] 含危险词 '{word}' — 建议改为精确描述（如'修改 CMakeLists.txt 去掉 sipclient 依赖'而非'清理{word}'）") for pattern in DIR_LEVEL_FILE_SCOPE_PATTERNS: if pattern in files_dirs and not any(f.endswith(ext) for ext in [".cpp", ".hpp", ".h", ".py", ".ts", ".md", ".json", ".txt", ".cmake"] for f in files_dirs.split(",")): if not done_when or "不" not in done_when: warnings.append(f"[{task_id}] 文件范围含目录级 '{pattern}' 但 done_when 无否定约束 — Worker 可能误解文件范围") break return warnings class ArcPhaseGate: """三阶段门控：discussing → proposing → confirmed。 execution-plan.json 仅在 phase=confirmed 时允许写入。 """ PHASES = ["discussing", "proposing", "confirmed"] def __init__(self, state_path: Path): self._state_path = state_path @property def current_phase(self) -> str: data = safe_json_load(self._state_path) or {} return data.get("arcPhase", "discussing") def advance_to(self, phase: str) -> None: if phase not in self.PHASES: raise ValueError(f"invalid phase: {phase!r}") idx_current = self.PHASES.index(self.current_phase) idx_target = self.PHASES.index(phase) if idx_target <= idx_current: return data = safe_json_load(self._state_path) or {} data["arcPhase"] = phase data[f"arcPhase_{phase}At"] = now_iso() atomic_json_write(self._state_path, data) def can_write_plan(self) -> bool: return self.current_phase == "confirmed" def confirm_architecture(self, user_confirmation: str) -> bool: """检查用户确认文本中的关键词，确认后推进到 confirmed。""" confirm_keywords = ["确认", "可以", "同意", "confirm", "yes", "ok", "好的", "没问题"] if any(kw in user_confirmation.lower() for kw in confirm_keywords): self.advance_to("confirmed") return True return False def _paths(project_root: Path) -> dict[str, Path]: root = airplan_root(project_root) / "state" / "airarc" return { "root": root, "state": root / "state.json", "reviews_dir": root / "reviews", "execution_plan_json": root / "reviews" / "execution-plan.json", "execution_plan_md": root / "reviews" / "execution-plan.md", "plan_delta_json": root / "reviews" / "plan-delta.json", "task_graph_json": root / "reviews" / "task-graph.json", } def _ensure_dirs(paths: dict[str, Path]) -> None: paths["reviews_dir"].mkdir(parents=True, exist_ok=True) def _export_task_graph_json(graph: TaskGraph, path: Path) -> None: data = { "dispatchFrozen": graph.dispatch_frozen, # P1-21 "nodes": {nid: {"id": n.id, "status": n.status, "task": n.task, "filesDirs": n.files_dirs, "doneWhen": n.done_when, "inDegree": n.in_degree, "outEdges": n.out_edges, "writeSet": n.write_set, "testRequired": n.test_required, "adrRefs": n.adr_refs} # P1-21 for nid, n in graph.nodes.items()}, "edges": [{"source": e.source, "target": e.target, "kind": e.kind} for e in graph.edges], } atomic_json_write(path, data) def _build_graph_from_todo(todo_path: Path) -> tuple[TaskGraph, list[str]]: """从 todo.md 构建初始 DAG，返回图和未满足 Done When 条件的任务列表。""" from air_runtime.todo_parser import parse_tasks tasks = parse_tasks(todo_path) graph = TaskGraph() violations = [] # P1-19.1: 记录 Done When 不含"测试通过"的任务 for t in tasks: # P1-21: 从 todo.md ADR 列提取 adr_refs adr_refs = [] if hasattr(t, "adr") and t.adr: adr_refs = [a.strip() for a in t.adr.split(",") if a.strip()] node = TaskNode( id=t.task_id, status=t.status, task=t.task, files_dirs=t.files_dirs, done_when=t.done_when, adr_refs=adr_refs, ) graph.add_node(node) # P1-19.1: Done When 必须包含"测试通过" if t.done_when and "测试通过" not in t.done_when: violations.append(t.task_id) # P1-19.1: 注入边界测试任务 _inject_boundary_tests(graph, tasks) return graph, violations def _inject_boundary_tests(graph: TaskGraph, tasks: list) -> None: """P1-19.1: 为每个模块边界注入测试任务。 规则： - 每个模块的公共接口必须有对应的接口测试任务 - 每个模块的核心逻辑必须有对应的单元测试任务 - 测试任务标记 test_required=True - Done When 必须包含"测试通过" """ # 从 tasks 提取模块信息（通过 files_dirs 推断模块） module_files: dict[str, set[str]] = {} for t in tasks: if t.files_dirs: for fd in t.files_dirs.split(","): fd = fd.strip() if fd: # 取第一级目录作为模块名 parts = fd.split("/") if len(parts) > 1: module = parts[0] else: module = fd.split(".")[0] if "." in fd else fd module_files.setdefault(module, set()).add(fd) # 为每个模块创建测试任务 test_counter = 0 for module, files in module_files.items(): if not files: continue # 接口测试任务（模块边界） test_counter += 1 interface_test_id = f"T-TEST-{test_counter:03d}" interface_test_node = TaskNode( id=interface_test_id, status="TODO", task=f"[边界测试] {module} 模块接口测试", files_dirs=",".join(sorted(files)), done_when="测试通过", test_required=True, ) graph.add_node(interface_test_node) # 单元测试任务（核心逻辑） test_counter += 1 unit_test_id = f"T-TEST-{test_counter:03d}" unit_test_node = TaskNode( id=unit_test_id, status="TODO", task=f"[单元测试] {module} 模块核心逻辑测试", files_dirs=",".join(sorted(files)), done_when="测试通过", test_required=True, ) graph.add_node(unit_test_node) # 添加依赖边：实现任务 → 测试任务 for tid, node in graph.nodes.items(): if tid.startswith("T-TEST-"): continue # 检查是否属于同一模块 node_files = set(node.files_dirs.split(",")) if node.files_dirs else set() if node_files & files: # 有交集，说明是同一模块的任务 graph.add_edge(Edge(source=tid, target=interface_test_id, kind="dependency")) graph.add_edge(Edge(source=tid, target=unit_test_id, kind="dependency")) def enter_mode(project_root: Path) -> dict: paths = _paths(project_root) _ensure_dirs(paths) payload = {"enabled": True, "updatedAt": now_iso(), "projectRoot": str(project_root)} atomic_json_write(paths["state"], payload) return {"state_path": str(paths["state"])} def parallel_review_mode(project_root: Path, todo_path: Path) -> dict: paths = _paths(project_root) _ensure_dirs(paths) # 三阶段门控：仅在 confirmed 阶段允许写入 execution-plan.json gate = ArcPhaseGate(paths["state"]) if not gate.can_write_plan(): return { "blocked": True, "reason": f"arc phase is '{gate.current_phase}', must be 'confirmed' before generating plan", "currentPhase": gate.current_phase, } review = build_parallel_review(todo_path) review_json_path = paths["reviews_dir"] / "parallel-review.json" review_md_path = paths["reviews_dir"] / "parallel-review.md" atomic_json_write(review_json_path, review.to_dict()) review_md_path.write_text(render_review_markdown(review), encoding="utf-8") # 构建 DAG（包含边界测试任务注入） graph, done_when_violations = _build_graph_from_todo(todo_path) for edge_info in review.to_dict().get("edges", []): graph.add_edge(Edge(source=edge_info["source"], target=edge_info["target"], kind=edge_info.get("kind", "dependency"))) # INV-16: 弱模型安全审计 — 扫描所有 TODO 任务的危险词和宽泛文件范围 safety_warnings = [] for nid, node in graph.nodes.items(): if node.status == "TODO": safety_warnings.extend( _audit_task_safety(nid, node.task, node.files_dirs, node.done_when) ) _export_task_graph_json(graph, paths["task_graph_json"]) # 生成执行计划 selected_tasks = review.parallel_groups[0].task_ids if review.parallel_groups else [] execution_plan = { "generatedAt": now_iso(), "projectRoot": str(project_root), "todoPath": str(todo_path), "planSource": "airarc-post-plan-review", "parallelReview": review.to_dict(), "selectedTasks": selected_tasks, "parallelGroups": [g.to_dict() for g in review.parallel_groups], "conflicts": [c.to_dict() for c in review.conflicts], "serializationPoints": review.serialization_points, "doneWhenViolations": done_when_violations, # P1-19.1: Done When 不含"测试通过"的任务 "boundaryTestTasks": [n.id for n in graph.nodes.values() if n.test_required], "safetyWarnings": safety_warnings, # INV-16: 弱模型安全警告 } atomic_json_write(paths["execution_plan_json"], execution_plan) markdown_lines = [ "# AirArc Execution Plan", "", f"- Generated: `{execution_plan['generatedAt']}`", f"- Plan Source: `{execution_plan['planSource']}`", "", "## Selected Tasks", ] for tid in selected_tasks: markdown_lines.append(f"- `{tid}`") markdown_lines.extend(["", "## Parallel Groups"]) for g in review.parallel_groups: markdown_lines.append(f"- `{g.name}`: {', '.join(g.task_ids)} — {g.reason}") paths["execution_plan_md"].write_text("\n".join(markdown_lines) + "\n", encoding="utf-8") # 保留 arcPhase 字段，避免 phase 被 reset 为 discussing existing_state = safe_json_load(paths["state"]) or {} state_payload = {**existing_state, "enabled": True, "updatedAt": now_iso(), "projectRoot": str(project_root)} atomic_json_write(paths["state"], state_payload) return {"json_path": str(review_json_path), "markdown_path": str(review_md_path), "execution_plan_json_path": str(paths["execution_plan_json"]), "parallel_group_count": len(review.parallel_groups), "conflict_count": len(review.conflicts), "done_when_violations": done_when_violations, "boundary_test_task_count": len([n for n in graph.nodes.values() if n.test_required])} def incremental_replan_mode(project_root: Path, todo_path: Path, previous_graph_path: Path | None = None) -> dict: """增量重规划：产出 PlanDelta 并喂回 prev graph，再写 plan-delta.json。""" paths = _paths(project_root) _ensure_dirs(paths) review = build_parallel_review(todo_path) new_graph, _ = _build_graph_from_todo(todo_path) for edge_info in review.to_dict().get("edges", []): new_graph.add_edge(Edge(source=edge_info["source"], target=edge_info["target"], kind=edge_info.get("kind", "dependency"))) delta = PlanDelta() if previous_graph_path and previous_graph_path.exists(): prev_graph = TaskGraph.load(previous_graph_path) delta = new_graph.diff(prev_graph) # 关键：把 delta 喂回去给 prev graph，保留已调度状态 if previous_graph_path.exists(): prev_graph.apply_delta(delta) _export_task_graph_json(prev_graph, previous_graph_path) else: delta.added_tasks = list(new_graph.nodes.values()) delta.edge_changes.added = list(new_graph.edges) atomic_json_write(paths["plan_delta_json"], { "generatedAt": now_iso(), "removedTasks": delta.removed_tasks, "addedTasks": [{"id": n.id, "task": n.task, "filesDirs": n.files_dirs, "doneWhen": n.done_when, "writeSet": n.write_set} for n in delta.added_tasks], "modifiedTasks": [{"id": n.id, "task": n.task, "filesDirs": n.files_dirs, "doneWhen": n.done_when, "writeSet": n.write_set} for n in delta.modified_tasks], "edgeChanges": { "added": [{"source": e.source, "target": e.target, "kind": e.kind} for e in delta.edge_changes.added], "removed": [{"source": e.source, "target": e.target, "kind": e.kind} for e in delta.edge_changes.removed], }, }) _export_task_graph_json(new_graph, paths["task_graph_json"]) log = EventLog(event_log_path(project_root)) log.emit("arc.replanned", {"delta_added": len(delta.added_tasks), "delta_removed": len(delta.removed_tasks), "delta_modified": len(delta.modified_tasks), "edges_added": len(delta.edge_changes.added), "edges_removed": len(delta.edge_changes.removed)}) return {"plan_delta_json_path": str(paths["plan_delta_json"]), "task_graph_json_path": str(paths["task_graph_json"]), "added_count": len(delta.added_tasks), "removed_count": len(delta.removed_tasks), "modified_count": len(delta.modified_tasks), "edges_added": len(delta.edge_changes.added), "edges_removed": len(delta.edge_changes.removed)} def main(args) -> None: project_root = Path(args.project).expanduser().resolve() sub = args.sub or "status" if sub == "enter": result = enter_mode(project_root) print("airplan_mode=arc") print(f"state_path={result['state_path']}") elif sub == "parallel-review": tpath = Path(args.todo).expanduser().resolve() if args.todo else get_todo_path(project_root) result = parallel_review_mode(project_root, tpath) print("airplan_mode=arc") print(f"json_path={result['json_path']}") print(f"parallel_group_count={result['parallel_group_count']}") print(f"conflict_count={result['conflict_count']}") elif sub == "incremental-replan": tpath = Path(args.todo).expanduser().resolve() if args.todo else get_todo_path(project_root) prev = _paths(project_root)["task_graph_json"] result = incremental_replan_mode(project_root, tpath, prev) print("airplan_mode=arc") print(f"plan_delta_path={result['plan_delta_json_path']}") print(f"added={result['added_count']} removed={result['removed_count']}") else: paths = _paths(project_root) state = safe_json_load(paths["state"]) or {} print(f"airplan_mode=arc") print(f"enabled={state.get('enabled', False)}")