任务委派：Subagent 与并行执行

主 Agent (Build / Sisyphus)
    │
    ├─ task → explore    "找叫 lock  的所有文件"
    ├─ task → explore    "find all TODO 与 FIXME"
    └─ task → librarian  "查 anthropic Messages API 当前最大 token"
       │
    并行 (.await asyncio.gather)
       │
    合并 sub-agent 的最终输出 → 给主 Agent

1. 探索任务 (explore)：多个独立方向可同时找
2. 文档查询 (librarian)：外部知识 / API / OSS 库
3. 复杂推理 (oracle)：把"想清楚"的任务外包给一个强推理 sub
4. 多步骤执行：拆成 5 个，每个 sub 独立写自己报告
   合并出 "5 个并列 PR-able 子任务" 后再合并

async def parallel_explore(query: str) -> list[str]:
    tasks = [
        explore("find files matching: " + query, scope="src"),
        explore("find files matching: " + query, scope="tests"),
        explore("find files matching: " + query, scope="scripts"),
    ]
    return await asyncio.gather(*tasks)

async def sequential_refine(goal: str) -> str:
    plan    = await metis(f"plan: {goal}")
    critique = await momus(f"review plan: {plan}")
    revised = await metis(f"adjust plan with: {critique}")
    return revised

async def multi_draft(prompt: str, k: int = 3) -> str:
    drafts = await asyncio.gather(*[
        build_subagent(f"draft {i+1}：{prompt}")
        for i in range(k)
    ])
    return await oracle(f"挑 3 稿中最优的一部分，合并出 final: {drafts}")

[CONTEXT]
   project: ai_learning/ai-course, hugo.laozhang.work repo
[GOAL]
   找出所有 .md 中文 front matter 不规范的文章（少了 slug）
[DOWNSTREAM]
   我要把结果汇报给主 Agent, 它要修这些文章
[REQUEST]
   列表 → 每行一个文件路径 + 标题 + 缺的 field
[REQUIRED TOOLS]
   rg / Glob / Read
[MUST DO]
   - 不修改任何文件（read-only）
   - 不输出冗长 markdown, 仅列表
[MUST NOT DO]
   - 不要解读为何缺
   - 不要并行修改文件

───────┬─────────┬─────────────────────────────────────
 role  │ cost    │ typical use
───────┼─────────┼─────────────────────────────────────
explore │ free    │ 找文件、grep 模式、survey 某区域
librarian│ cheap   │ 文档 / OSS / 库内幕
oracle  │ expensv │ 当局迷呼救：根基架构 / bug
metis   │ expensv │ plan 前把"意图不清晰"的部分祛
momus   │ expensv │ 挑 plan 毛病
build   │ default │ 主执行
───────┴─────────┴─────────────────────────────────────

async def collect(sources: list) -> str:
    results = await asyncio.gather(*sources, return_exceptions=True)
    return "\n\n".join(
        f"### {src.__name__}\n{r if not isinstance(r, Exception) else f'[ERROR]{r}'}"
        for src, r in zip(sources, results)
    )

[NOTE] 让一个 sub-agent 写 skill 体系时,它倾向于:
        - 若只读文案符合工作流,写得很好
        - 若需要"实际经验感"(知道哪些 prompt 真起效),就写些听起来扎实但实战无效

async def double_check(commit_msg: str) -> tuple[bool, str]:
    # 并行派两个 sub：一个 grep 找 secret 风险，一个跑 ruff/basedpyright
    secret_check, lint_check = await asyncio.gather(
        explore_sub("[CONTEXT] 当前 staged diff\n"
                    f"[GOAL] 找任何可能 secret 泄漏痕迹\n"
                    f"[MUST DO] 只看 git diff\n"
                    f"[OUTPUT] 单文字 `clean` / `risk: <理由>`"),
        build_sub("[CONTEXT] 当前 staged diff\n"
                   "[GOAL] 跑 ruff check + basedpyright\n"
                   "[OUTPUT] two-line summary")
    )
    if "clean" not in secret_check or "0 errors" not in lint_check:
        return False, secret_check + "\n" + lint_check
    return True, commit_msg