feat: T-1.21 + T-1.22 + P1-23 — 打通 dispatch→Worker→merge 全链路

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 <noreply@anthropic.com>

commands/eng.md

@@ -31,11 +31,8 @@ python "$HOME/plugins/airplan/scripts/airplan.py" --mode eng --project . --sub s
 ```
 
 然后：
-- 无活跃波次 → 派发下一波次
+- 无活跃波次 → 执行 dispatch 流程（见下方 dispatch 段）
-- 有活跃 Worker → 监控，不要盲目重新派发
+- 有活跃 Worker → 执行 monitor，不要盲目重新派发
-- 每个 dispatched 任务 spawn 一个 /do 子代理（fork_context=false）
-- 只传递项目路径+任务 handoff 内容，不要 fork 完整父对话
-- 最多 recommendedConcurrency 个 Worker 同时活跃
 - 就绪结果出现时：merge through `--sub merge --result <path>`
 
 ### 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 <path>`
+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
 

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]  # 空列表也是正确答案，不 fallback
-                return ready[:max_count]
         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 变更级联失效处理。
 

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)