Harness实战：并行调度——拓扑排序让Subagent飞起来

task("分析前端技术栈")     → 30s
task("分析后端技术栈")     → 25s
task("分析数据库技术栈")   → 20s
task("汇总写报告")         → 15s（需要前三个结果）
───────────────────────────────
总耗时: 90s

Layer 0 (并行):  [分析前端]  [分析后端]  [分析数据库]
                      │           │            │
                      └───────────┼────────────┘
                                  ▼
Layer 1 (串行):          [汇总写报告]

PARALLEL_TASKS_TOOL = {
    "name": "parallel_tasks",
    "description": (
        "Run multiple subtasks with dependency-aware parallel scheduling. "
        "Each task has an id, prompt, and optional depends_on list. "
        "Tasks with no dependencies run in parallel. "
        "Tasks with dependencies wait until all dependencies complete, "
        "then receive their results as context. "
        "Returns a map of {task_id: result}."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "tasks": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "id": {"type": "string"},
                        "prompt": {"type": "string"},
                        "depends_on": {
                            "type": "array",
                            "items": {"type": "string"},
                        },
                    },
                    "required": ["id", "prompt"],
                },
            }
        },
        "required": ["tasks"],
    },
}

from collections import defaultdict

def topological_layers(tasks: list[dict]) -> list[list[dict]]:
    """将任务按依赖关系分层。同层任务无互相依赖，可以并行。"""
    task_map = {t["id"]: t for t in tasks}
    in_degree = {t["id"]: 0 for t in tasks}
    dependents = defaultdict(list)

    for t in tasks:
        for dep in t.get("depends_on", []):
            if dep not in task_map:
                raise ValueError(
                    f"Task '{t['id']}' depends on unknown task '{dep}'"
                )
            in_degree[t["id"]] += 1
            dependents[dep].append(t["id"])

    layers = []
    remaining = dict(in_degree)

    while remaining:
        layer = [tid for tid, deg in remaining.items() if deg == 0]
        if not layer:
            raise ValueError(
                f"Circular dependency: {list(remaining.keys())}"
            )
        layers.append([task_map[tid] for tid in layer])
        for tid in layer:
            del remaining[tid]
            for dependent in dependents[tid]:
                if dependent in remaining:
                    remaining[dependent] -= 1

    return layers

from concurrent.futures import ThreadPoolExecutor, as_completed

def run_parallel_tasks(tasks: list[dict]) -> str:
    layers = topological_layers(tasks)
    results = {}  # task_id → result string

    for i, layer in enumerate(layers):
        with ThreadPoolExecutor(max_workers=len(layer)) as pool:
            futures = {}
            for task in layer:
                # ★ 注入依赖任务的结果作为上下文
                prompt = task["prompt"]
                deps = task.get("depends_on", [])
                if deps:
                    dep_context = "\n".join(
                        f"[Result of '{d}']: {results[d]}"
                        for d in deps
                    )
                    prompt = (
                        f"{prompt}\n\n"
                        f"Context from prerequisite tasks:\n"
                        f"{dep_context}"
                    )

                futures[pool.submit(run_subagent, prompt)] = task["id"]

            for future in as_completed(futures):
                task_id = futures[future]
                results[task_id] = future.result()

    output_parts = [
        f"[{tid}]: {result}" for tid, result in results.items()
    ]
    return "\n\n".join(output_parts)

TOOLS = BASE_TOOLS + [TASK_TOOL, PARALLEL_TASKS_TOOL]

TOOL_HANDLERS = {
    **BASE_HANDLERS,
    "task": lambda **kw: run_subagent(kw["prompt"]),
    "parallel_tasks": lambda **kw: run_parallel_tasks(kw["tasks"]),
}

❯ /plan 分析这个项目的前端、后端和数据库技术栈，汇总写一份技术报告

⏳ Generating plan...

📋 Plan:
1. 分析前端技术栈
2. 分析后端技术栈
3. 分析数据库技术栈
4. 汇总三个分析结果，写技术报告

 Execute? (y/n) y

⚡ Executing...

> parallel_tasks:

⚡ Parallel scheduler: 4 tasks, 2 layers
  Layer 0: ['frontend', 'backend', 'database'] (parallel)

  🔀 Subagent started: 分析前端技术栈...
  🔀 Subagent started: 分析后端技术栈...
  🔀 Subagent started: 分析数据库技术栈...
    > bash: cat frontend/package.json | head -30
    > bash: cat backend/pyproject.toml
    > bash: cat docker-compose.yml | grep -A5 postgres
  🔀 Subagent done (6 messages)
  ✓ database (1/4)
  🔀 Subagent done (8 messages)
  ✓ backend (2/4)
  🔀 Subagent done (10 messages)
  ✓ frontend (3/4)
  Layer 0 done in 12.3s

  Layer 1: ['report'] (parallel)

  🔀 Subagent started: 汇总写技术报告...
  🔀 Subagent done (4 messages)
  ✓ report (4/4)
  Layer 1 done in 8.1s

if not layer:
    raise ValueError(f"Circular dependency: {list(remaining.keys())}")

if deps:
    dep_context = "\n".join(
        f"[Result of '{d}']: {results[d]}" for d in deps
    )
    prompt = f"{prompt}\n\nContext from prerequisite tasks:\n{dep_context}"

SYSTEM = "You are a coding agent at {WORKDIR}. Use tools to solve tasks. Act, don't explain."

SYSTEM = f"""You are a coding agent at {WORKDIR}. Use tools to solve tasks. Act, don't explain.

When a task can be broken into subtasks, use `parallel_tasks` to run them concurrently.
Put ALL tasks (including dependent ones) in a SINGLE `parallel_tasks` call — the scheduler handles ordering.
Each task needs an id and prompt. Use depends_on to declare dependencies between tasks.
Do NOT split tasks across multiple `parallel_tasks` calls — dependent tasks would lose access to earlier results.
Only use sequential `task` calls when the next step depends on seeing the previous result to decide what to do."""

"Analyze the task and outline a step-by-step plan. Do NOT execute anything yet."

"Analyze the task and outline a step-by-step plan. Do NOT execute anything yet. "
"For each step, note which other steps it depends on (if any). "
"Mark steps that can run in parallel."

1. 分析前端技术栈（无依赖）
2. 分析后端技术栈（无依赖）
3. 分析数据库技术栈（无依赖）
4. 汇总写报告（依赖 1, 2, 3）

步骤 1-3 可并行执行。

Step 1: 统计文件信息（无依赖）
Step 2: 提取 import 列表（无依赖）
Step 3a: 分析复杂度（依赖 Step 1）
Step 3b: 分析依赖用途（依赖 Step 2）
Step 4: 交叉对比（依赖 Step 3a + 3b）
Step 5: 生成报告（依赖 Step 4）

> parallel_tasks: [step1_file_info, step2_imports]     ← 第一次调用，只提交了 Layer 0

✓ step1_file_info
✓ step2_imports

> parallel_tasks: [step3a_complexity, step3b_deps]     ← 第二次调用，提交 Layer 1
  step3a_complexity depends_on: [step1_file_info]

ValueError: Task 'step3a_complexity' depends on unknown task 'step1_file_info'

Put ALL tasks (including dependent ones) in a SINGLE `parallel_tasks` call
— the scheduler handles ordering.
Do NOT split tasks across multiple `parallel_tasks` calls
— dependent tasks would lose access to earlier results.