Claude Code 并发批处理 Worker 池实战：用 Hooks 护栏驱动 -p 批量扇出

flowchart LR
    Q[任务队列 jobs.jsonl] --> SCHED[调度器 xargs -P N]
    SCHED --> W1[Worker 1 claude -p]
    SCHED --> W2[Worker 2 claude -p]
    SCHED --> WN[Worker N claude -p]
    W1 -.策略.-> H[(Hooks settings.json)]
    W2 -.策略.-> H
    WN -.策略.-> H
    H -->|PreToolUse 拦截| GUARD{放行/阻断/提醒}
    W1 --> R1[stream-json 结果]
    W2 --> R2[stream-json 结果]
    WN --> RN[stream-json 结果]
    R1 --> AGG[结果汇聚 report.md]
    R2 --> AGG
    RN --> AGG

# text：纯文本，最简单
claude -p "把 README 里所有的 TODO 收集成一个列表" 

# json：单条 JSON 对象，便于 jq 解析
claude -p "..." --output-format json | jq -r '.result'

# stream-json：换行分隔的事件流，批处理的黄金格式
claude -p "..." --output-format stream-json

# 从 stream-json 抓出 session_id 和最终结果
claude -p "修复 utils.py 里的类型错误" \
  --output-format stream-json \
  --allowedTools "Edit,Read,Bash(pytest *)" \
  > /tmp/w1.jsonl

# 取 session_id（用于 --resume）
grep '"type":"system"' /tmp/w1.jsonl | head -1 | jq -r '.session_id'

# 取最终结果与成本
grep '"type":"result"' /tmp/w1.jsonl | jq -r '"cost=\(.cost_usd) turns=\(.num_turns)\n\(.result)"'

claude -p "..." \
  --allowedTools "Read,Edit,Write,Bash(git *),Bash(npm test)"

claude -p "重构这个模块" --max-turns 25 --output-format stream-json

# 第一次跑，记住 session_id
SID=$(claude -p "..." --output-format stream-json | \
      grep '"type":"system"' | jq -r '.session_id' | head -1)

# 之后基于这个会话继续（保留全部上下文）
claude -p "继续刚才的重构，跑一下测试" --resume "$SID" --output-format stream-json

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [{ "type": "command", "command": ".claude/hooks/guard-bash.sh" }]
      }
    ],
    "PostToolUse": [
      {
        "matcher": "Edit|Write",
        "hooks": [{ "type": "command", "command": ".claude/hooks/auto-fmt.sh" }]
      }
    ],
    "Stop": [
      {
        "hooks": [{ "type": "command", "command": ".claude/hooks/budget-guard.sh" }]
      }
    ]
  }
}

{
  "session_id": "abc123",
  "transcript_path": "/path/to/transcript.jsonl",
  "cwd": "/repo/worktree-3",
  "hook_event_name": "PreToolUse",
  "tool_name": "Bash",
  "tool_input": { "command": "git push --force origin main", "description": "..." }
}

#!/usr/bin/env bash
# 读取 hook 输入
input=$(cat)
cmd=$(echo "$input" | jq -r '.tool_input.command // ""')

# 危险模式黑名单（确定性拦截，不依赖模型判断）
deny_patterns='rm[[:space:]]+-rf|git[[:space:]]+push.*--force|git[[:space:]]+push.*-f|>[[:space:]]*/dev/sd|chmod[[:space:]]+-R[[:space:]]+777'

if echo "$cmd" | grep -Eq "$deny_patterns"; then
  echo "⛔ 被护栏拦截：检测到危险命令模式 → $cmd" >&2
  exit 2   # 阻断，stderr 回显给模型
fi

# 放行
exit 0

#!/usr/bin/env bash
input=$(cat)
file=$(echo "$input" | jq -r '.tool_input.file_path // ""')

[ -z "$file" ] || [ ! -f "$file" ] && exit 0

case "$file" in
  *.py)   ruff format "$file" 2>/dev/null; ruff check --fix "$file" 2>/dev/null ;;
  *.ts|*.tsx|*.js) npx --no-install prettier --write "$file" 2>/dev/null ;;
  *.go)   gofmt -w "$file" 2>/dev/null ;;
esac
exit 0   # PostToolUse 用 0，格式化失败也不阻断

#!/usr/bin/env bash
input=$(cat)
transcript=$(echo "$input" | jq -r '.transcript_path')

# 从 transcript 估算 token 用量，超阈值就输出一个醒目提醒
# stdout 会被注入为额外上下文（exit 0 时）
echo "⏱️ Worker 即将停止，请确认任务已完成且测试通过。" 
exit 0

# 强制继续（让模型别急着停）
printf '{"continue": true, "stopReason": "测试尚未全部通过，继续工作"}'
exit 0

{
  "continue": false,
  "stopReason": "预算耗尽",
  "suppressOutput": false,
  "additionalContext": "（注入到对话的额外上下文）"
}

{"id":"auth","repo":"https://github.com/acme/svc-auth.git","task":"把所有 print 换成 logging，并补类型注解"}
{"id":"billing","repo":"https://github.com/acme/svc-billing.git","task":"为 /invoice 接口补 pytest 覆盖到 80%"}
{"id":"gateway","repo":"https://github.com/acme/svc-gateway.git","task":"把 requests 替换为 httpx，保持行为不变"}

#!/usr/bin/env bash
# worker.sh —— 单个 Worker 的生命周期
set -u
LINE="$1"
OUT_DIR="${2:-./out}"

id=$(echo "$LINE"   | jq -r '.id')
repo=$(echo "$LINE"  | jq -r '.repo')
task=$(echo "$LINE"  | jq -r '.task')

WORK=".work/$id"
LOG="$OUT_DIR/$id.jsonl"
mkdir -p "$WORK" "$OUT_DIR"

# 1) 获取代码（已存在则 fetch）
if [ -d "$WORK/.git" ]; then
  git -C "$WORK" fetch --quiet && git -C "$WORK" reset --hard origin/HEAD -q
else
  git clone --quiet --depth 1 "$repo" "$WORK"
fi

# 把共享 hooks 策略软链进去（策略唯一来源）
mkdir -p "$WORK/.claude/hooks"
ln -sf "$(pwd)/.claude/settings.json" "$WORK/.claude/settings.json"
ln -sf "$(pwd)/.claude/hooks/"* "$WORK/.claude/hooks/" 2>/dev/null || true

# 2) 跑 Worker（带工具白名单 + 回合预算 + stream-json）
cd "$WORK"
claude -p "$task" \
  --output-format stream-json \
  --allowedTools "Read,Edit,Write,Bash(git *),Bash(pytest*),Bash(npm test)" \
  --max-turns 30 \
  > "$OUT_DIR/$id.jsonl" 2> "$OUT_DIR/$id.err"

cd - >/dev/null

# 3) 抽取结果摘要
sid=$(grep '"type":"system"' "$LOG" | head -1 | jq -r '.session_id')
result=$(grep '"type":"result"' "$LOG" | head -1)
cost=$(echo "$result" | jq -r '.cost_usd // "?"')
turns=$(echo "$result" | jq -r '.num_turns // "?"')
ok=$(echo "$result" | jq -r '.is_error // "true"')

# 4) 汇报行（供后续聚合）
printf '%s\t%s\t%s\t%s\t%s\n' "$id" "$ok" "$cost" "$turns" "$sid" >> "$OUT_DIR/_summary.tsv"
echo "[$id] done ok=$ok cost=\$$cost turns=$turns session=$sid"

# 姿势 A：xargs -P（POSIX 通用）
mkdir -p out
cat jobs.jsonl | xargs -P 4 -I{} ./worker.sh {} ./out

# 姿势 B：GNU parallel（输出更整齐，自带进度）
mkdir -p out
parallel --bar -j 4 ./worker.sh {} ./out :::: jobs.jsonl

# 把 _summary.tsv 渲染成 Markdown 总表
{
  echo "# 批处理报告"
  echo
  echo "| 任务 | 成功 | 成本(\$) | 回合数 | session_id |"
  echo "|---|---|---|---|---|"
  sort out/_summary.tsv | awk -F'\t' '{printf "| %s | %s | %s | %s | `%s` |\n",$1,$2,$3,$4,$5}'
  echo
  total=$(awk -F'\t' '{s+=$3} END{print s}' out/_summary.tsv)
  echo "**总成本：\$${total}**"
} > report.md

#!/usr/bin/env bash
set -euo pipefail

JOBS="${1:-jobs.jsonl}"
PARALLEL="${2:-4}"
rm -rf out .work && mkdir -p out .work

echo "→ 扇出 $PARALLEL 个 Worker 处理 $(wc -l < "$JOBS") 个任务"
parallel -j "$PARALLEL" ./worker.sh {} ./out :::: "$JOBS"

echo "→ 汇聚报告"
./report.sh        # 即上一节的 awk 片段
echo "完成，见 report.md"

claude -p "上一次中断了，继续完成并把测试跑绿" --resume <session_id>

# .claude/agents/refactor-worker.yaml
---
name: refactor-worker
description: 独立重构一个模块并跑测试，用于批量子任务并行
tools: Read, Edit, Write, Bash(git *), Bash(pytest*)
model: sonnet
---
你是模块重构专员。只改分配给你的模块目录，完成后必须运行测试并报告结果。
不要触碰其他模块。

flowchart TD
    A[批量任务需求] --> B{任务间共享代码库?}
    B -- 多个独立仓库 --> C[外层 Worker 池<br/>xargs -P + claude -p]
    B -- 同一仓库多模块 --> D[Subagent 内层扇出]
    A --> E{需要确定性安全约束?}
    E -- 是 --> F[PreToolUse/PostToolUse Hooks]
    E -- 仅需预算控制 --> G[Stop hook + --max-turns]
    C --> F
    D --> F
    F --> Z[汇聚 stream-json → report]
    G --> Z