From dd4aa6b34583d538322672ca50c777a509522205 Mon Sep 17 00:00:00 2001 From: AirLongDian Date: Thu, 11 Jun 2026 14:58:04 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20T-1.21=20+=20T-1.22=20+=20P1-23=20?= =?UTF-8?q?=E2=80=94=20=E6=89=93=E9=80=9A=20dispatch=E2=86=92Worker?= =?UTF-8?q?=E2=86=92merge=20=E5=85=A8=E9=93=BE=E8=B7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T-1.21 (P1-22): Dispatch → Worker 桥接 - eng_mode: spawn_workers() 为每个 ready 任务生成标准 Skill 调用参数 - commands/eng.md: dispatch 段从意图描述改为可执行5步伪代码 Agent 不再需要猜测工具名、参数格式、task-text 来源 T-1.22 (P1-24): Merge → TaskGraph 状态同步 - merge_worker_result Phase 6.5: 写回 task-graph.json 节点 status - merge 后节点状态 DONE，ready_tasks() 过滤已完成任务 _select_ready_tasks 修复： - DAG ready 为空时不再 fallback 到 todo.md（todo.md 状态陈旧时导致重复派发） P1-23: commands/eng.md dispatch 指令操作化 - 保留 Eng 工具白名单不变（极端接管需要 Write/Edit） - Arc/Eng 约束非对称性是刻意的设计决策 63+4 项测试全通过，含端到端全链路测试 Co-Authored-By: Claude Opus 4.7 --- commands/eng.md | 25 ++- lib/air_runtime/modes/eng_mode.py | 37 +++- test_t_121_122.py | 271 ++++++++++++++++++++++++++++++ 3 files changed, 323 insertions(+), 10 deletions(-) create mode 100644 test_t_121_122.py diff --git a/commands/eng.md b/commands/eng.md index 281325a..87d72eb 100644 --- a/commands/eng.md +++ b/commands/eng.md @@ -31,11 +31,8 @@ python "$HOME/plugins/airplan/scripts/airplan.py" --mode eng --project . --sub s ``` 然后： -- 无活跃波次 → 派发下一波次 -- 有活跃 Worker → 监控，不要盲目重新派发 -- 每个 dispatched 任务 spawn 一个 /do 子代理（fork_context=false） -- 只传递项目路径+任务 handoff 内容，不要 fork 完整父对话 -- 最多 recommendedConcurrency 个 Worker 同时活跃 +- 无活跃波次 → 执行 dispatch 流程（见下方 dispatch 段） +- 有活跃 Worker → 执行 monitor，不要盲目重新派发 - 就绪结果出现时：merge through `--sub merge --result ` ### status @@ -56,7 +53,23 @@ python "$HOME/plugins/airplan/scripts/airplan.py" --mode eng --project . --sub p python "$HOME/plugins/airplan/scripts/airplan.py" --mode eng --project . --sub dispatch ``` -读取 AirPlan/state/aireng/dispatch/ 中的派发清单，为每个任务 spawn 一个隔离的 /do Worker 子代理。 +dispatch 返回 `{waveId, taskIds, dispatchPath}`。**然后按以下步骤操作，不可跳过**： + +1. 从 dispatch 返回值中取 `taskIds` 列表 +2. 对 `taskIds` 中的**每个** `tid`，顺序执行： + a. 读 `AirPlan/state/airarc/reviews/task-graph.json`，从 `nodes[tid].task` 取任务描述文本 + b. 调用 `Skill` 工具启动子代理： + - `skill`: `"airplan"` + - `args`: `"do --sub enter --task-id {tid} --task-text '{task描述}' --project ."` + c. 每个 `Skill` 调用自动以 `fork_context=false` 运行，创建隔离的 Do Worker +3. 所有 Worker spawn 完成后，进入 monitor 状态 +4. Worker 完成后，对其 `result.json` 调用 `--sub merge --result ` +5. merge 后 `task-graph.json` 节点状态自动同步为 DONE，不会被重复派发 + +**注意**： +- 一个 task 对应一次 `Skill("airplan", ...)` 调用，多个 task 可以在同一条消息中并发发起 +- task-text 从 task-graph.json 获取，不是猜的 +- Do Worker 子代理的 allowed-tools 是 `[Read, Glob, Grep, Bash, Write, Edit]`，Agent 框架自动应用 ### monitor diff --git a/lib/air_runtime/modes/eng_mode.py b/lib/air_runtime/modes/eng_mode.py index 506e42d..4c6a799 100644 --- a/lib/air_runtime/modes/eng_mode.py +++ b/lib/air_runtime/modes/eng_mode.py @@ -556,6 +556,16 @@ def merge_worker_result(project_root: Path, result_path: Path) -> dict: state["lastMergeAt"] = now_iso() atomic_json_write(paths["state"], state) + # Phase 6.5: 同步 task-graph.json 节点状态（P1-24） + tg_json = airplan_root(project_root) / "state" / "airarc" / "reviews" / "task-graph.json" + if tg_json.exists(): + from air_runtime.modes.arc_mode import _export_task_graph_json + graph = TaskGraph.load(tg_json) + if task_id in graph.nodes: + new_status = "DONE" if status == "done" else status.upper() + graph.nodes[task_id].status = new_status + _export_task_graph_json(graph, tg_json) + log.emit(MERGE_COMPLETED, { "taskId": task_id, "status": status, @@ -586,17 +596,17 @@ def merge_worker_result(project_root: Path, result_path: Path) -> dict: def _select_ready_tasks(project_root: Path, max_count: int) -> list[str]: - """优先从 task-graph.json 的 DAG 计算 in-degree 为 0 的 TODO task，fallback parse_tasks。""" + """优先从 task-graph.json 的 DAG 计算 in-degree 为 0 的 TODO task。 + DAG 中 ready 为空意味着无任务可派发（全部完成或全部被依赖阻塞），不应 fallback 到 todo.md。""" task_graph_json = airplan_root(project_root) / "state" / "airarc" / "reviews" / "task-graph.json" if task_graph_json.exists(): try: graph = TaskGraph.load(task_graph_json) ready = graph.ready_tasks() - if ready: - return ready[:max_count] + return ready[:max_count] # 空列表也是正确答案，不 fallback except Exception: pass - # fallback + # fallback：仅在 task-graph.json 不存在时使用 todo.md todo = get_todo_path(project_root) if not todo.exists(): return [] @@ -604,6 +614,25 @@ def _select_ready_tasks(project_root: Path, max_count: int) -> list[str]: return [t.task_id for t in tasks if t.status == "TODO"][:max_count] +def spawn_workers(project_root: Path, task_ids: list[str]) -> list[dict]: + """T-1.21: 为每个 ready 任务准备 spawn 指令，返回 Agent 可直接消费的 Skill 调用参数列表。 + 不实际启动子进程——启动由 Agent 框架的 Skill 工具完成。 + """ + tg_json = airplan_root(project_root) / "state" / "airarc" / "reviews" / "task-graph.json" + graph = TaskGraph.load(tg_json) if tg_json.exists() else TaskGraph() + instructions = [] + for tid in task_ids: + node = graph.nodes.get(tid) + task_text = node.task if node else "" + instructions.append({ + "skill": "airplan", + "args": f"do --sub enter --task-id {tid} --task-text '{task_text}' --project .", + "taskId": tid, + "taskText": task_text, + }) + return instructions + + def handle_adr_invalidation(project_root: Path, adr_id: str) -> dict: """P1-21: ADR 变更级联失效处理。 diff --git a/test_t_121_122.py b/test_t_121_122.py new file mode 100644 index 0000000..1db5714 --- /dev/null +++ b/test_t_121_122.py @@ -0,0 +1,271 @@ +#!/usr/bin/env python3 +"""T-1.21 + T-1.22 端到端测试：arc → eng → do → merge → task-graph sync 全链路""" + +import sys +import tempfile +import json +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent / "lib")) + + +def test_spawn_workers(): + """测试 spawn_workers 返回正确的 Skill 调用参数""" + from air_runtime.task_graph import TaskGraph, TaskNode + from air_runtime.modes.arc_mode import _export_task_graph_json + from air_runtime.modes.eng_mode import spawn_workers + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + arc_dir = root / "AirPlan" / "state" / "airarc" / "reviews" + arc_dir.mkdir(parents=True, exist_ok=True) + + graph = TaskGraph() + graph.add_node(TaskNode(id="G-001", status="TODO", task="实现视频解码器", + files_dirs="src/decoder.cpp")) + graph.add_node(TaskNode(id="G-002", status="TODO", task="实现音频解码器", + files_dirs="src/audio.cpp")) + _export_task_graph_json(graph, arc_dir / "task-graph.json") + + instructions = spawn_workers(root, ["G-001", "G-002"]) + assert len(instructions) == 2 + assert instructions[0]["skill"] == "airplan" + assert "do --sub enter --task-id G-001" in instructions[0]["args"] + assert "实现视频解码器" in instructions[0]["args"] + assert instructions[1]["taskId"] == "G-002" + + print("✓ spawn_workers 测试通过") + return True + + +def test_merge_syncs_task_graph(): + """P1-24: 测试 merge 后 task-graph.json 节点状态同步""" + from air_runtime.task_graph import TaskGraph, TaskNode + from air_runtime.modes.arc_mode import _export_task_graph_json, ArcPhaseGate + from air_runtime.modes.eng_mode import enter_engine, merge_worker_result + from air_runtime.modes.do_mode import enter_worker, finish_worker + from air_runtime.contracts import WorkerResult + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + + # Arc 准备 + from air_runtime.modes.arc_mode import enter_mode, parallel_review_mode + enter_mode(root) + gate = ArcPhaseGate(root / "AirPlan" / "state" / "airarc" / "state.json") + gate.advance_to("confirmed") + + todo = root / "AirPlan" / "todo.md" + todo.parent.mkdir(parents=True, exist_ok=True) + todo.write_text("| Task | Status | Files/Dirs | Done When |\n" + "|------|--------|------------|-----------|\n" + "| G-001 解码器 | TODO | src/decoder.cpp | 编译通过 |\n") + + parallel_review_mode(root, todo) + + # Eng 初始化 + enter_engine(root) + + # 验证 merge 前 task-graph 中 G-001 是 TODO + tg_json = root / "AirPlan" / "state" / "airarc" / "reviews" / "task-graph.json" + graph = TaskGraph.load(tg_json) + assert "G-001" in graph.nodes + assert graph.nodes["G-001"].status == "TODO" + print(f" merge 前: G-001 status = {graph.nodes['G-001'].status}") + + # Do Worker 执行 + w = enter_worker(root, "G-001", task_text="G-001 解码器") + wr = WorkerResult(task_id="G-001", status="done", summary="完成", + files_changed=["src/decoder.cpp"], validations="编译通过") + rpath = root / "AirPlan" / "state" / "airdo" / "tasks" / "G-001" / "result.json" + rpath.parent.mkdir(parents=True, exist_ok=True) + rpath.write_text(json.dumps(wr.to_dict())) + + f = finish_worker(root, "G-001", rpath) + print(f" Do finish: status={f['status']}, route={f['routingDecision']['target']}") + + # Eng merge + m = merge_worker_result(root, rpath) + print(f" Eng merge: task={m['taskId']}, status={m['status']}") + + # 验证 merge 后 task-graph 中 G-001 是 DONE + graph_after = TaskGraph.load(tg_json) + assert graph_after.nodes["G-001"].status == "DONE", \ + f"merge 后 G-001 应为 DONE，实际: {graph_after.nodes['G-001'].status}" + print(f" merge 后: G-001 status = {graph_after.nodes['G-001'].status}") + + # 验证不会重复派发（ready_tasks 不包含已完成的 G-001） + ready = graph_after.ready_tasks() + assert "G-001" not in ready, f"G-001 不应在 ready 队列中: {ready}" + print(f" ready queue: {ready}") + + print("✓ merge 同步 task-graph 状态测试通过") + return True + + +def test_full_pipeline_e2e(): + """完整端到端流水线：arc → eng plan → dispatch → do → merge → task-graph sync""" + from air_runtime.task_graph import TaskGraph + from air_runtime.modes.arc_mode import enter_mode, parallel_review_mode, ArcPhaseGate, _export_task_graph_json + from air_runtime.modes.eng_mode import (enter_engine, build_engine_plan, + dispatch_worker_group, spawn_workers, + merge_worker_result) + from air_runtime.modes.do_mode import enter_worker, finish_worker + from air_runtime.contracts import WorkerResult + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + + # ============ ARC ============ + enter_mode(root) + gate = ArcPhaseGate(root / "AirPlan" / "state" / "airarc" / "state.json") + gate.advance_to("confirmed") + + todo = root / "AirPlan" / "todo.md" + todo.parent.mkdir(parents=True, exist_ok=True) + todo.write_text("| Task | Status | Files/Dirs | Done When |\n" + "|------|--------|------------|-----------|\n" + "| H-000 前置 | DONE | | 完成 |\n" + "| G-001 解码器 | TODO | src/decoder.cpp | 编译通过 |\n" + "| G-002 编码器 | TODO | src/encoder.cpp | 编译通过 |\n") + + r = parallel_review_mode(root, todo) + assert not r.get("blocked"), f"Arc blocked: {r.get('reason')}" + print("1. ARC parallel-review ✓") + + # ============ ENG PLAN ============ + enter_engine(root) + plan = build_engine_plan(root, todo) + assert plan["planningSource"] == "airarc-execution-plan" + assert "G-001" in plan["selectedTasks"] + assert "G-002" in plan["selectedTasks"] + print(f"2. ENG plan: tasks={plan['selectedTasks']}") + + # ============ ENG DISPATCH ============ + dispatch = dispatch_worker_group(root) + assert not dispatch.get("blocked"), f"dispatch blocked: {dispatch.get('reason')}" + assert len(dispatch["taskIds"]) >= 2 + print(f"3. ENG dispatch: wave={dispatch['waveId']}, tasks={dispatch['taskIds']}") + + # ============ SPAWN INSTRUCTIONS ============ + instructions = spawn_workers(root, dispatch["taskIds"]) + assert len(instructions) >= 2 + for inst in instructions: + assert inst["skill"] == "airplan" + assert "do --sub enter --task-id" in inst["args"] + assert inst["taskText"] # 不能为空 + print(f"4. spawn_workers: {len(instructions)} 个 Worker 指令准备完成") + for inst in instructions: + print(f" - {inst['taskId']}: {inst['taskText'][:40]}") + + # ============ DO WORKER (模拟 Agent 调 Skill) ============ + for inst in instructions: + tid = inst["taskId"] + task_text = inst["taskText"] + w = enter_worker(root, tid, task_text=task_text) + wr = WorkerResult(task_id=tid, status="done", summary="完成", + files_changed=[f"src/{tid.lower()}.cpp"], + validations="编译通过") + rpath = root / "AirPlan" / "state" / "airdo" / "tasks" / tid / "result.json" + rpath.parent.mkdir(parents=True, exist_ok=True) + rpath.write_text(json.dumps(wr.to_dict())) + f = finish_worker(root, tid, rpath) + assert f["routingDecision"]["target"] == "merge" + print("5. DO Workers 全部完成 (G-001, G-002)") + + # ============ ENG MERGE ============ + for inst in instructions: + tid = inst["taskId"] + rpath = root / "AirPlan" / "state" / "airdo" / "tasks" / tid / "result.json" + m = merge_worker_result(root, rpath) + assert m["status"] == "done" + print("6. ENG merge 全部完成") + + # ============ VERIFY TASK-GRAPH SYNC ============ + tg_json = root / "AirPlan" / "state" / "airarc" / "reviews" / "task-graph.json" + graph_after = TaskGraph.load(tg_json) + assert graph_after.nodes["G-001"].status == "DONE", \ + f"G-001 应为 DONE，实际: {graph_after.nodes['G-001'].status}" + assert graph_after.nodes["G-002"].status == "DONE", \ + f"G-002 应为 DONE，实际: {graph_after.nodes['G-002'].status}" + print(f"7. task-graph sync: G-001={graph_after.nodes['G-001'].status}, " + f"G-002={graph_after.nodes['G-002'].status}") + + # ============ VERIFY NO RE-DISPATCH ============ + ready_after = graph_after.ready_tasks() + assert "G-001" not in ready_after + assert "G-002" not in ready_after + # H-000 是 DONE，也不在 ready + # 只有 T-TEST-* 还在 TODO + print(f"8. 无重复派发: ready={ready_after}") + # 再次 dispatch 不应包含已完成任务 + dispatch2 = dispatch_worker_group(root) + for done_tid in ["G-001", "G-002", "H-000"]: + assert done_tid not in dispatch2.get("taskIds", []), \ + f"已完成任务 {done_tid} 不应再次被派发" + + print(f"9. 二次 dispatch: tasks={dispatch2.get('taskIds', [])} (无已完成任务)") + print("✓ 完整端到端流水线测试通过") + return True + + +def test_dispatch_no_workers_means_no_dispatch(): + """测试所有任务完成后 dispatch 返回空""" + from air_runtime.task_graph import TaskGraph, TaskNode + from air_runtime.modes.arc_mode import _export_task_graph_json + from air_runtime.modes.eng_mode import dispatch_worker_group, enter_engine + from air_runtime.modes.arc_mode import enter_mode + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + enter_mode(root) + enter_engine(root) + + # 所有任务都是 DONE + arc_dir = root / "AirPlan" / "state" / "airarc" / "reviews" + arc_dir.mkdir(parents=True, exist_ok=True) + graph = TaskGraph() + graph.add_node(TaskNode(id="G-001", status="DONE", task="已完成")) + graph.add_node(TaskNode(id="G-002", status="DONE", task="已完成")) + _export_task_graph_json(graph, arc_dir / "task-graph.json") + + result = dispatch_worker_group(root) + assert result["taskIds"] == [], f"全部完成时应为空，实际: {result['taskIds']}" + print("✓ 全部完成时 dispatch 返回空测试通过") + return True + + +def main(): + print("=" * 50) + print("T-1.21 + T-1.22 端到端全链路测试") + print("=" * 50) + + tests = [ + ("spawn_workers", test_spawn_workers), + ("merge 同步 task-graph", test_merge_syncs_task_graph), + ("完整端到端流水线", test_full_pipeline_e2e), + ("全完成时 dispatch 返回空", test_dispatch_no_workers_means_no_dispatch), + ] + + 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)