csv-wave-pipeline

Name: csv-wave-pipeline
Author: catlog22

github.com/catlog22Agent 工作流2.1k46 次安装19 周前更新

代码仓库

关于 csv-wave-pipeline

csv-wave-pipeline 是一项技能，可将实现需求拆分为依赖有序的任务并分波并发执行。它会在各波次间传递任务发现与共享信息，并支持续接会话。

主要功能

01将需求拆分为 3–10 项任务
02按依赖关系安排波次
03并发执行 Agent 任务
04跨波次传递发现与结果
05续接会话并生成报告

适用场景

01并行推进功能实现
02向后续任务传递前置发现
03恢复中断的批处理

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/catlog22/claude-code-workflow --skill csv-wave-pipeline

安装后在 Claude Code 中输入/csv-wave-pipeline确认已加载。

详细介绍

自动模式

使用 --yes 或 -y 时：自动确认任务拆解，跳过交互式验证，并使用默认值。

CSV 波次流水线

用法

$csv-wave-pipeline "Implement user authentication with OAuth, JWT, and 2FA"
$csv-wave-pipeline -c 4 "Refactor payment module with Stripe and PayPal"
$csv-wave-pipeline -y "Build notification system with email and SMS"
$csv-wave-pipeline --continue "auth-20260228"

标志：

-y, --yes：跳过所有确认（自动模式）
-c, --concurrency N：每个波次中的最大并发 Agent 数量（默认值：4）
--continue：恢复现有会话

概述

使用 spawn_agents_on_csv 进行基于波次的批量执行，并支持跨波次上下文传递。任务按依赖关系分组成多个波次；每个波次并发执行，其结果会传递给下一个波次。

核心工作流：拆解 → 计算波次 → 逐波次执行 → 汇总

Phase 1: Requirement → CSV
   ├─ Parse requirement into subtasks (3-10 tasks)
   ├─ Identify dependencies (deps column)
   ├─ Compute dependency waves (topological sort → depth grouping)
   ├─ Generate tasks.csv with wave column
   └─ User validates task breakdown (skip if -y)

Phase 2: Wave Execution Engine
   ├─ For each wave (1..N):
   │   ├─ Build wave CSV (filter rows for this wave)
   │   ├─ Inject previous wave findings into prev_context column
   │   ├─ spawn_agents_on_csv(wave CSV)
   │   ├─ Collect results, merge into master tasks.csv
   │   └─ Check: any failed? → skip dependents or retry
   └─ discoveries.ndjson shared across all waves (append-only)

Phase 3: Results Aggregation
   ├─ Export final results.csv
   ├─ Generate context.md with all findings
   ├─ Display summary: completed/failed/skipped per wave
   └─ Offer: view results | retry failed | done

上下文传递

两个上下文通道会在各波次之间传递：

CSV 发现结果（结构化）：context_from 列 → 注入 prev_context — 针对特定任务的定向上下文
NDJSON 发现结果（广播）：discoveries.ndjson — 所有 Agent 均可使用的通用探索结果

Wave 1 agents:
  ├─ Execute tasks (no prev_context)
  ├─ Write findings to report_agent_job_result
  └─ Append discoveries to discoveries.ndjson
        ↓ merge results into master CSV
Wave 2 agents:
  ├─ Read discoveries.ndjson (exploration sharing)
  ├─ Read prev_context column (wave 1 findings from context_from)
  ├─ Execute tasks with full upstream context
  ├─ Write findings to report_agent_job_result
  └─ Append new discoveries to discoveries.ndjson
        ↓ merge results into master CSV
Wave 3+ agents: same pattern, accumulated context from all prior waves

会话与输出结构

.workflow/.csv-wave/{session-id}/
├── tasks.csv                  # Master state (updated per wave)
├── results.csv                # Final results export (Phase 3)
├── discoveries.ndjson         # Shared discovery board (all agents, append-only)
├── context.md                 # Human-readable report (Phase 3)
├── wave-{N}.csv               # Temporary per-wave input (cleaned up after merge)
└── wave-{N}-results.csv       # Temporary per-wave output (cleaned up after merge)

文件	用途	生命周期
`tasks.csv`	主状态 — 包含所有任务及其状态/发现结果	每个波次结束后更新
`wave-{N}.csv`	每个波次的输入文件，包含 prev_context 列	波次开始前创建，结束后删除
`wave-{N}-results.csv`	`spawn_agents_on_csv` 的每个波次输出	波次执行期间创建，合并后删除
`results.csv`	所有任务结果的最终导出文件	在第 3 阶段创建
`discoveries.ndjson`	所有 Agent 共享的探索看板	仅追加，贯穿各个波次
`context.md`	便于阅读的执行报告	在第 3 阶段创建

CSV 架构

tasks.csv（主状态）

id,title,description,test,acceptance_criteria,scope,hints,execution_directives,deps,context_from,wave,status,findings,files_modified,tests_passed,acceptance_met,error
"1","Setup auth module","Create auth directory structure and base files","Verify directory exists and base files export expected interfaces","auth/ dir created; index.ts and types.ts export AuthProvider interface","src/auth/**","Follow monorepo module pattern || package.json;src/shared/types.ts","","","","1","","","","","",""
"2","Implement OAuth","Add OAuth provider integration with Google and GitHub","Unit test: mock OAuth callback returns valid token; Integration test: verify redirect URL generation","OAuth login redirects to provider; callback returns JWT; supports Google and GitHub","src/auth/oauth/**","Use passport.js strategy pattern || src/auth/index.ts;docs/oauth-flow.md","Run npm test -- --grep oauth before completion","1","1","2","","","","","",""
"3","Add JWT tokens","Implement JWT generation and validation","Unit test: sign/verify round-trip; Edge test: expired token returns 401","generateToken() returns valid JWT; verifyToken() rejects expired/tampered tokens","src/auth/jwt/**","Use jsonwebtoken library; Set default expiry 1h || src/config/auth.ts","Ensure tsc --noEmit passes","1","1","2","","","","","",""
"4","Setup 2FA","Add TOTP-based 2FA with QR code generation","Unit test: TOTP verify with correct code; Test: QR data URL is valid","QR code generates scannable image; TOTP verification succeeds within time window","src/auth/2fa/**","Use speakeasy + qrcode libraries || src/auth/oauth/strategy.ts;src/auth/jwt/token.ts","Run full test suite: npm test","2;3","1;2;3","3","","","","","",""

列：

列	阶段	描述
`id`	输入	唯一任务标识符（字符串）
`title`	输入	简短任务标题
`description`	输入	详细任务描述 — 要实现的内容
`test`	输入	测试用例：要编写哪些测试，以及如何验证（单元/集成/边界）
`acceptance_criteria`	输入	验收标准：定义“完成”的可衡量条件
`scope`	输入	目标文件/目录 glob — 限定 Agent 的工作范围，防止不同任务之间发生文件冲突
`hints`	输入	实现提示 + 参考文件。格式：`tips text \|\| file1;file2`。`\|\|` 之前是实现方法；之后是开始前要阅读的现有文件。任一部分均可省略
`execution_directives`	输入	执行约束：用于验证的命令、工具限制、环境要求
`deps`	输入	以分号分隔的依赖任务 ID（为空 = 无依赖）
`context_from`	输入	此任务需要其发现结果的任务 ID，以分号分隔
`wave`	计算得出	波次编号（通过拓扑排序计算，从 1 开始）
`status`	输出	`pending` → `completed` / `failed` / `skipped`
`findings`	输出	关键发现或实现说明（最多 500 个字符）
`files_modified`	输出	以分号分隔的文件路径
`tests_passed`	输出	所有已定义的测试用例是否通过（true/false）
`acceptance_met`	输出	已满足/未满足的验收标准摘要
`error`	输出	失败时的错误消息（成功时为空）

每个波次的 CSV（临时）

每个波次都会生成一个临时文件 wave-{N}.csv，其中包含一个额外的 prev_context 列。此列根据 context_from 生成，方法是从主 CSV 中查找已完成任务的 findings：

id,title,description,test,acceptance_criteria,scope,hints,execution_directives,deps,context_from,wave,prev_context
"2","Implement OAuth","Add OAuth integration","Unit test: mock OAuth callback returns valid token","OAuth login redirects to provider; callback returns JWT","src/auth/oauth/**","Use passport.js strategy pattern || src/auth/index.ts;docs/oauth-flow.md","Run npm test -- --grep oauth","1","1","2","[Task 1] Created auth/ with index.ts and types.ts"
"3","Add JWT tokens","Implement JWT","Unit test: sign/verify round-trip; Edge test: expired token returns 401","generateToken() returns valid JWT; verifyToken() rejects expired/tampered tokens","src/auth/jwt/**","Use jsonwebtoken library; Set default expiry 1h || src/config/auth.ts","Ensure tsc --noEmit passes","1","1","2","[Task 1] Created auth/ with index.ts and types.ts"

共享发现板协议

所有波次中的所有 Agent 都共享 discoveries.ndjson。这可避免重复探索代码库。

生命周期：由第一个写入发现信息的 Agent 创建。跨波次保留，绝不清空。Agent 通过 echo '...' >> discoveries.ndjson 追加内容。

格式：NDJSON，每行都是独立的 JSON：

{"ts":"2026-02-28T10:00:00+08:00","worker":"1","type":"code_pattern","data":{"name":"repository-pattern","file":"src/repos/Base.ts","description":"Abstract CRUD repository"}}
{"ts":"2026-02-28T10:01:00+08:00","worker":"2","type":"integration_point","data":{"file":"src/auth/index.ts","description":"Auth module entry","exports":["authenticate","authorize"]}}

发现类型：

类型	去重键	描述
`code_pattern`	`data.name`	发现的可复用代码模式
`integration_point`	`data.file`	模块连接点
`convention`	单例	代码风格约定
`blocker`	`data.issue`	遇到的阻塞问题
`tech_stack`	单例	项目技术栈
`test_command`	单例	发现的测试命令

协议规则：

在自行探索前先读取发现板 → 跳过已覆盖的区域
立即通过 echo >> 写入发现信息 → 不要批量处理
去重——检查现有条目；如果存在类型和去重键都相同的条目，则跳过
仅追加——绝不修改或删除现有行

实现

会话初始化

const getUtc8ISOString = () => new Date(Date.now() + 8 * 60 * 60 * 1000).toISOString()

// Parse flags
const AUTO_YES = $ARGUMENTS.includes('--yes') || $ARGUMENTS.includes('-y')
const continueMode = $ARGUMENTS.includes('--continue')
const concurrencyMatch = $ARGUMENTS.match(/(?:--concurrency|-c)\s+(\d+)/)
const maxConcurrency = concurrencyMatch ? parseInt(concurrencyMatch[1]) : 4

// Clean requirement text (remove flags — word-boundary safe)
const requirement = $ARGUMENTS
  .replace(/--yes|(?:^|\s)-y(?=\s|$)|--continue|--concurrency\s+\d+|-c\s+\d+/g, '')
  .trim()

let sessionId, sessionFolder

const slug = requirement.toLowerCase()
  .replace(/[^a-z0-9\u4e00-\u9fa5]+/g, '-')
  .substring(0, 40)
const dateStr = getUtc8ISOString().substring(0, 10).replace(/-/g, '')
sessionId = `cwp-${dateStr}-${slug}`
sessionFolder = `.workflow/.csv-wave/${sessionId}`

// Continue mode: find existing session
if (continueMode) {
  const existing = Bash(`ls -t .workflow/.csv-wave/ 2>/dev/null | head -1`).trim()
  if (existing) {
    sessionId = existing
    sessionFolder = `.workflow/.csv-wave/${sessionId}`
    // Read existing tasks.csv, find incomplete waves, resume from there
    const existingCsv = Read(`${sessionFolder}/tasks.csv`)
    // → jump to Phase 2 with remaining waves
  }
}

Bash(`mkdir -p ${sessionFolder}`)

上游交接信息接收（可选）

csv-wave-pipeline 可通过 handoff.json 接收来自 analyze-with-file 或 workflow-lite-plan 的结构化上下文：

// Check if requirement references a prior analysis session
const handoffPathMatch = requirement.match(/handoff:(.+\.json)/)
let handoffContext = null

if (handoffPathMatch) {
  const handoffPath = handoffPathMatch[1]
  if (file_exists(handoffPath)) {
    handoffContext = JSON.parse(Read(handoffPath))
    // handoffContext: { source, session_id, summary, implementation_scope[], code_anchors[], key_findings[], exploration_artifacts{} }

    // Enrich requirement with handoff context for CLI decomposition in Phase 1
    requirement = `${handoffContext.summary}\n\n` +
      `## Implementation Scope\n${handoffContext.implementation_scope.map((s, i) =>
        `${i+1}. **${s.objective}** [${s.priority}]\n   Files: ${s.target_files?.join(', ') || 'TBD'}\n   Done when: ${s.acceptance_criteria?.join('; ') || 'TBD'}`
      ).join('\n')}\n\n` +
      (handoffContext.key_findings?.length > 0
        ? `## Key Findings\n${handoffContext.key_findings.map(f => `- ${f.point || f}`).join('\n')}\n\n`
        : '') +
      (handoffContext.code_anchors?.length > 0
        ? `## Code Anchors\n${handoffContext.code_anchors.slice(0, 8).map(a => `- \`${a.file}:${a.lines}\`: ${a.significance}`).join('\n')}\n\n`
        : '')

    // Load exploration artifacts into discoveries.ndjson seed (if available)
    if (handoffContext.exploration_artifacts?.exploration_codebase && file_exists(handoffContext.exploration_artifacts.exploration_codebase)) {
      const codebaseData = JSON.parse(Read(handoffContext.exploration_artifacts.exploration_codebase))
      const seedDiscoveries = [
        ...(codebaseData.patterns || []).map(p => JSON.stringify({
          ts: getUtc8ISOString(), worker: 'handoff', type: 'code_pattern',
          data: { name: p.pattern || p, file: p.files?.[0] || '', description: p.description || '' }
        })),
        ...(codebaseData.relevant_files || []).slice(0, 5).map(f => JSON.stringify({
          ts: getUtc8ISOString(), worker: 'handoff', type: 'integration_point',
          data: { file: f.path, description: f.annotation || f.summary || '' }
        }))
      ]
      if (seedDiscoveries.length > 0) {
        Write(`${sessionFolder}/discoveries.ndjson`, seedDiscoveries.join('\n') + '\n')
      }
    }

    console.log(`[Handoff] Loaded from ${handoffContext.source} session ${handoffContext.session_id}: ${handoffContext.implementation_scope?.length || 0} scope items`)
  }
}

交接信息用法：

$csv-wave-pipeline "handoff:.workflow/.analysis/ANL-2026-04-29-auth/handoff.json"
$csv-wave-pipeline -y "handoff:.workflow/.lite-plan/auth-plan/handoff.json"

提供交接信息时：

implementation_scope[] 会将预先分析的目标、目标文件和验收标准补充到第 1 阶段的 CLI 分解提示词中
code_anchors[] 为 Agent 提供具体的 file:line 入口点
exploration_artifacts.exploration_codebase 会使用已知模式和集成点为 discoveries.ndjson 提供初始内容，因此第 1 波的 Agent 可跳过重复的代码库探索

CSV 实用函数

// Escape a value for CSV (wrap in quotes, double internal quotes)
function csvEscape(value) {
  const str = String(value ?? '')
  return str.replace(/"/g, '""')
}

// Parse CSV string into array of objects (header row → keys)
function parseCsv(csvString) {
  const lines = csvString.trim().split('\n')
  if (lines.length < 2) return []
  const headers = parseCsvLine(lines[0]).map(h => h.replace(/^"|"$/g, ''))
  return lines.slice(1).map(line => {
    const cells = parseCsvLine(line).map(c => c.replace(/^"|"$/g, '').replace(/""/g, '"'))
    const obj = {}
    headers.forEach((h, i) => { obj[h] = cells[i] ?? '' })
    return obj
  })
}

// Parse a single CSV line respecting quoted fields with commas/newlines
function parseCsvLine(line) {
  const cells = []
  let current = ''
  let inQuotes = false
  for (let i = 0; i < line.length; i++) {
    const ch = line[i]
    if (inQuotes) {
      if (ch === '"' && line[i + 1] === '"') {
        current += '"'
        i++ // skip escaped quote
      } else if (ch === '"') {
        inQuotes = false
      } else {
        current += ch
      }
    } else {
      if (ch === '"') {
        inQuotes = true
      } else if (ch === ',') {
        cells.push(current)
        current = ''
      } else {
        current += ch
      }
    }
  }
  cells.push(current)
  return cells
}

第 1 阶段：需求 → CSV

目标：将需求分解为任务，计算依赖波次，并生成 tasks.csv。

步骤：

分解需求

// Use ccw cli to decompose requirement into subtasks
Bash({
  command: `ccw cli -p "PURPOSE: Decompose requirement into 3-10 atomic tasks for batch agent execution. Each task must include implementation description, test cases, and acceptance criteria.

TASK: • Parse requirement into independent subtasks • Identify dependencies between tasks (which must complete before others) • Identify context flow (which tasks need previous tasks' findings) • For each task, define concrete test cases (unit/integration/edge) • For each task, define measurable acceptance criteria (what defines 'done') • Each task must be executable by a single agent with file read/write access MODE: analysis CONTEXT: @**/* EXPECTED: JSON object with tasks array. Each task: {id: string, title: string, description: string, test: string, acceptance_criteria: string, scope: string, hints: string, execution_directives: string, deps: string[], context_from: string[]}.

description: what to implement (specific enough for an agent to execute independently)
test: what tests to write and how to verify (e.g. 'Unit test: X returns Y; Edge test: handles Z')
acceptance_criteria: measurable conditions that define done (e.g. 'API returns 200; token expires after 1h')
scope: target file/directory glob (e.g. 'src/auth/**') — tasks in same wave MUST have non-overlapping scopes
hints: implementation tips + reference files, format ' || <ref_file1>;<ref_file2>' (e.g. 'Use strategy pattern || src/base/Strategy.ts;docs/design.md')
execution_directives: commands to run for verification or tool constraints (e.g. 'Run npm test --bail; Ensure tsc passes')
deps: task IDs that must complete first
context_from: task IDs whose findings are needed CONSTRAINTS: 3-10 tasks | Each task is atomic | No circular deps | test and acceptance_criteria must be concrete and verifiable | Same-wave tasks must have non-overlapping scopes

REQUIREMENT: ${requirement}" --tool gemini --mode analysis --rule planning-breakdown-task-steps`, run_in_background: true }) // Wait for CLI completion via hook callback // Parse JSON from CLI output → decomposedTasks[]


2. **计算波次**（带深度跟踪的 Kahn BFS 拓扑排序）

```javascript
// Algorithm:
// 1. Build in-degree map and adjacency list from deps
// 2. Enqueue all tasks with in-degree 0 at wave 1
// 3. BFS: for each dequeued task at wave W, for each dependent D:
//    - Decrement D's in-degree
//    - D.wave = max(D.wave, W + 1)
//    - If D's in-degree reaches 0, enqueue D
// 4. Any task without wave assignment → circular dependency error
//
// Wave properties:
//   Wave 1: no dependencies — fully independent
//   Wave N: all deps in waves 1..(N-1) — guaranteed completed before start
//   Within a wave: tasks are independent → safe for concurrent execution
//
// Example:
//   A(no deps)→W1, B(no deps)→W1, C(deps:A)→W2, D(deps:A,B)→W2, E(deps:C,D)→W3
//   Wave 1: [A,B] concurrent → Wave 2: [C,D] concurrent → Wave 3: [E]

function computeWaves(tasks) {
  const taskMap = new Map(tasks.map(t => [t.id, t]))
  const inDegree = new Map(tasks.map(t => [t.id, 0]))
  const adjList = new Map(tasks.map(t => [t.id, []]))

  for (const task of tasks) {
    for (const dep of task.deps) {
      if (taskMap.has(dep)) {
        adjList.get(dep).push(task.id)
        inDegree.set(task.id, inDegree.get(task.id) + 1)
      }
    }
  }

  // BFS-based topological sort with depth tracking
  const queue = []  // [taskId, depth]
  const waveAssignment = new Map()

  for (const [id, deg] of inDegree) {
    if (deg === 0) {
      queue.push([id, 1])
      waveAssignment.set(id, 1)
    }
  }

  let maxWave = 1
  let idx = 0
  while (idx < queue.length) {
    const [current, depth] = queue[idx++]
    for (const next of adjList.get(current)) {
      const newDeg = inDegree.get(next) - 1
      inDegree.set(next, newDeg)
      const nextDepth = Math.max(waveAssignment.get(next) || 0, depth + 1)
      waveAssignment.set(next, nextDepth)
      if (newDeg === 0) {
        queue.push([next, nextDepth])
        maxWave = Math.max(maxWave, nextDepth)
      }
    }
  }

  // Detect cycles
  for (const task of tasks) {
    if (!waveAssignment.has(task.id)) {
      throw new Error(`Circular dependency detected involving task ${task.id}`)
    }
  }

  return { waveAssignment, maxWave }
}

const { waveAssignment, maxWave } = computeWaves(decomposedTasks)

生成 tasks.csv

const header = 'id,title,description,test,acceptance_criteria,scope,hints,execution_directives,deps,context_from,wave,status,findings,files_modified,tests_passed,acceptance_met,error'
const rows = decomposedTasks.map(task => {
  const wave = waveAssignment.get(task.id)
  return [
    task.id,
    csvEscape(task.title),
    csvEscape(task.description),
    csvEscape(task.test),
    csvEscape(task.acceptance_criteria),
    csvEscape(task.scope),
    csvEscape(task.hints),
    csvEscape(task.execution_directives),
    task.deps.join(';'),
    task.context_from.join(';'),
    wave,
    'pending',  // status
    '',         // findings
    '',         // files_modified
    '',         // tests_passed
    '',         // acceptance_met
    ''          // error
  ].map(cell => `"${String(cell).replace(/"/g, '""')}"`).join(',')
})

Write(`${sessionFolder}/tasks.csv`, [header, ...rows].join('\n'))

用户验证（AUTO_YES 时跳过）

if (!AUTO_YES) {
  // Display task breakdown with wave assignment
  console.log(`\n## Task Breakdown (${decomposedTasks.length} tasks, ${maxWave} waves)\n`)
  for (let w = 1; w <= maxWave; w++) {
    const waveTasks = decomposedTasks.filter(t => waveAssignment.get(t.id) === w)
    console.log(`### Wave ${w} (${waveTasks.length} tasks, concurrent)`)
    waveTasks.forEach(t => console.log(`  - [${t.id}] ${t.title}`))
  }

  const answer = functions.request_user_input({
    questions: [{
      header: "验证",
      id: "validation",
      question: "Approve task breakdown?",
      options: [
        { label: "Approve(Recommended)", description: "Proceed with wave execution" },
        { label: "Modify", description: `Edit ${sessionFolder}/tasks.csv manually, then --continue` },
        { label: "Cancel", description: "Abort" }
      ]
    }]
  })  // BLOCKS

  if (answer.answers.validation.answers[0] === "Modify") {
    console.log(`Edit: ${sessionFolder}/tasks.csv\nResume: $csv-wave-pipeline --continue`)
    return
  } else if (answer.answers.validation.answers[0] === "Cancel") {
    return
  }
}

成功标准：已创建 schema 有效且包含波次分配的 tasks.csv，没有循环依赖，并且已获用户批准（或设置了 AUTO_YES）。

第 2 阶段：波次执行引擎

目标：通过 spawn_agents_on_csv 按波次执行任务。每个波次都能看到之前波次的结果。

步骤：

波次循环

const failedIds = new Set()
const skippedIds = new Set()

for (let wave = 1; wave <= maxWave; wave++) {
  console.log(`\n## Wave ${wave}/${maxWave}\n`)

  // 1. Read current master CSV
  const masterCsv = parseCsv(Read(`${sessionFolder}/tasks.csv`))

  // 2. Filter tasks for this wave
  const waveTasks = masterCsv.filter(row => parseInt(row.wave) === wave)

  // 3. Skip tasks whose deps failed
  const executableTasks = []
  for (const task of waveTasks) {
    const deps = task.deps.split(';').filter(Boolean)
    if (deps.some(d => failedIds.has(d) || skippedIds.has(d))) {
      skippedIds.add(task.id)
      updateMasterCsvRow(sessionFolder, task.id, {
        status: 'skipped',
        error: 'Dependency failed or skipped'
      })
      console.log(`  [${task.id}] ${task.title} → SKIPPED (dependency failed)`)
      continue
    }
    executableTasks.push(task)
  }

  if (executableTasks.length === 0) {
    console.log(`  No executable tasks in wave ${wave}`)
    continue
  }

  // 4. Build prev_context for each task (from context_from → master CSV findings)
  for (const task of executableTasks) {
    const contextIds = task.context_from.split(';').filter(Boolean)
    const prevFindings = contextIds
      .map(id => {
        const prevRow = masterCsv.find(r => r.id === id)
        if (prevRow && prevRow.status === 'completed' && prevRow.findings) {
          return `[Task ${id}: ${prevRow.title}] ${prevRow.findings}`
        }
        return null
      })
      .filter(Boolean)
      .join('\n')
    task.prev_context = prevFindings || 'No previous context available'
  }

  // 5. Write wave CSV
  const waveHeader = 'id,title,description,test,acceptance_criteria,scope,hints,execution_directives,deps,context_from,wave,prev_context'
  const waveRows = executableTasks.map(t =>
    [t.id, t.title, t.description, t.test, t.acceptance_criteria, t.scope, t.hints, t.execution_directives, t.deps, t.context_from, t.wave, t.prev_context]
      .map(cell => `"${String(cell).replace(/"/g, '""')}"`)
      .join(',')
  )
  Write(`${sessionFolder}/wave-${wave}.csv`, [waveHeader, ...waveRows].join('\n'))

  // 6. Execute wave
  console.log(`  Executing ${executableTasks.length} tasks (concurrency: ${maxConcurrency})...`)

  const waveResult = spawn_agents_on_csv({
    csv_path: `${sessionFolder}/wave-${wave}.csv`,
    id_column: "id",
    instruction: buildInstructionTemplate(sessionFolder, wave),
    max_concurrency: maxConcurrency,
    max_runtime_seconds: 600,
    output_csv_path: `${sessionFolder}/wave-${wave}-results.csv`,
    output_schema: {
      type: "object",
      properties: {
        id: { type: "string" },
        status: { type: "string", enum: ["completed", "failed"] },
        findings: { type: "string" },
        files_modified: { type: "array", items: { type: "string" } },
        tests_passed: { type: "boolean" },
        acceptance_met: { type: "string" },
        error: { type: "string" }
      },
      required: ["id", "status", "findings", "tests_passed"]
    }
  })
  // ↑ Blocks until all agents in this wave complete

  // 7. Merge results into master CSV
  const waveResults = parseCsv(Read(`${sessionFolder}/wave-${wave}-results.csv`))
  for (const result of waveResults) {
    updateMasterCsvRow(sessionFolder, result.id, {
      status: result.status,
      findings: result.findings || '',
      files_modified: (result.files_modified || []).join(';'),
      tests_passed: String(result.tests_passed ?? ''),
      acceptance_met: result.acceptance_met || '',
      error: result.error || ''
    })

    if (result.status === 'failed') {
      failedIds.add(result.id)
      console.log(`  [${result.id}] ${result.title} → FAILED: ${result.error}`)
    } else {
      console.log(`  [${result.id}] ${result.title} → COMPLETED`)
    }
  }

  // 8. Cleanup temporary wave CSVs
  Bash(`rm -f "${sessionFolder}/wave-${wave}.csv" "${sessionFolder}/wave-${wave}-results.csv"`)

  console.log(`  Wave ${wave} done: ${waveResults.filter(r => r.status === 'completed').length} completed, ${waveResults.filter(r => r.status === 'failed').length} failed`)
}

指令模板构建器

function buildInstructionTemplate(sessionFolder, wave) {
  return `

TASK ASSIGNMENT

MANDATORY FIRST STEPS

Read shared discoveries: ${sessionFolder}/discoveries.ndjson (if exists, skip if not)
Read project context: .workflow/project-tech.json (if exists)

Your Task

Task ID: {id} Title: {title} Description: {description} Scope: {scope}

Implementation Hints & Reference Files

{hints}

Format: ` || <ref_file1>;<ref_file2>`. Read ALL reference files (after ||) before starting implementation. Apply tips (before ||) as implementation guidance.

Execution Directives

{execution_directives}

Commands to run for verification, tool restrictions, or environment requirements. Follow these constraints during and after implementation.

Test Cases

{test}

Acceptance Criteria

{acceptance_criteria}

Previous Tasks' Findings (Context)

{prev_context}

Execution Protocol

Read references: Parse {hints} — read all files listed after `||` to understand existing patterns
Read discoveries: Load ${sessionFolder}/discoveries.ndjson for shared exploration findings
Use context: Apply previous tasks' findings from prev_context above
Stay in scope: ONLY create/modify files within {scope} — do NOT touch files outside this boundary
Apply hints: Follow implementation tips from {hints} (before `||`)
Execute: Implement the task as described
Write tests: Implement the test cases defined above
Run directives: Execute commands from {execution_directives} to verify your work
Verify acceptance: Ensure all acceptance criteria are met before reporting completion
Share discoveries: Append exploration findings to shared board: ```bash echo '{"ts":"","worker":"{id}","type":"","data":{...}}' >> ${sessionFolder}/discoveries.ndjson ```
Report result: Return JSON via report_agent_job_result

Discovery Types to Share

`code_pattern`: {name, file, description} — reusable patterns found
`integration_point`: {file, description, exports[]} — module connection points
`convention`: {naming, imports, formatting} — code style conventions
`blocker`: {issue, severity, impact} — blocking issues encountered
`tech_stack`: {runtime, framework, language} — project technology stack
`test_command`: {command, scope, description} — test commands discovered

Output (report_agent_job_result)

Return JSON: { "id": "{id}", "status": "completed" | "failed", "findings": "Key discoveries and implementation notes (max 500 chars)", "files_modified": ["path1", "path2"], "tests_passed": true | false, "acceptance_met": "Summary of which acceptance criteria were met/unmet", "error": "" }

IMPORTANT: Set status to "completed" ONLY if:

All test cases pass
All acceptance criteria are met Otherwise set status to "failed" with details in error field. ` }

主 CSV 更新助手

function updateMasterCsvRow(sessionFolder, taskId, updates) {
  const csvPath = `${sessionFolder}/tasks.csv`
  const content = Read(csvPath)
  const lines = content.split('\n')
  const header = lines[0].split(',')

  for (let i = 1; i < lines.length; i++) {
    const cells = parseCsvLine(lines[i])
    if (cells[0] === taskId || cells[0] === `"${taskId}"`) {
      // Update specified columns
      for (const [col, val] of Object.entries(updates)) {
        const colIdx = header.indexOf(col)
        if (colIdx >= 0) {
          cells[colIdx] = `"${String(val).replace(/"/g, '""')}"`
        }
      }
      lines[i] = cells.join(',')
      break
    }
  }

  Write(csvPath, lines.join('\n'))
}

成功标准：所有波次均按顺序执行；每个波次的结果都会在下一波开始前合并到主 CSV 中；前置任务失败时会跳过依赖任务；discoveries.ndjson 会累积所有波次的发现信息。

第 3 阶段：结果汇总

目标：生成最终结果和面向用户的报告。

步骤：

导出 results.csv

const masterCsv = Read(`${sessionFolder}/tasks.csv`)
// results.csv = master CSV (already has all results populated)
Write(`${sessionFolder}/results.csv`, masterCsv)

生成 context.md

const tasks = parseCsv(masterCsv)
const completed = tasks.filter(t => t.status === 'completed')
const failed = tasks.filter(t => t.status === 'failed')
const skipped = tasks.filter(t => t.status === 'skipped')

const contextContent = `# CSV Batch Execution Report

Session: ${sessionId} Requirement: ${requirement} Completed: ${getUtc8ISOString()} Waves: ${maxWave} | Concurrency: ${maxConcurrency}

Summary

Metric	Count
Total Tasks	${tasks.length}
Completed	${completed.length}
Failed	${failed.length}
Skipped	${skipped.length}
Waves	${maxWave}

Wave Execution

${Array.from({ length: maxWave }, (_, i) => i + 1).map(w => { const waveTasks = tasks.filter(t => parseInt(t.wave) === w) return ### Wave ${w} ${waveTasks.map(t => - [${t.id}] ${t.title}: ${t.status}${t.tests_passed ? ' ✓tests' : ''}${t.error ? ' — ' + t.error : ''} ${t.findings ? 'Findings: ' + t.findings : ''}).join('\n')} }).join('\n\n')}

Task Details

${tasks.map(t => `### ${t.id}: ${t.title}

Field	Value
Status	${t.status}
Wave	${t.wave}
Scope	${t.scope
Dependencies	${t.deps
Context From	${t.context_from
Tests Passed	${t.tests_passed
Acceptance Met	${t.acceptance_met
Error	${t.error

Description: ${t.description}

Test Cases: ${t.test || 'N/A'}

Acceptance Criteria: ${t.acceptance_criteria || 'N/A'}

Hints: ${t.hints || 'N/A'}

Execution Directives: ${t.execution_directives || 'N/A'}

Findings: ${t.findings || 'N/A'}

Files Modified: ${t.files_modified || 'none'} `).join('\n---\n')}

All Modified Files

${[...new Set(tasks.flatMap(t => (t.files_modified || '').split(';')).filter(Boolean))].map(f => '- ' + f).join('\n') || 'None'} `

Write(${sessionFolder}/context.md, contextContent)


3. **显示摘要**

```javascript
console.log(`
## Execution Complete

- **Session**: ${sessionId}
- **Waves**: ${maxWave}
- **Completed**: ${completed.length}/${tasks.length}
- **Failed**: ${failed.length}
- **Skipped**: ${skipped.length}

**Results**: ${sessionFolder}/results.csv
**Report**: ${sessionFolder}/context.md
**Discoveries**: ${sessionFolder}/discoveries.ndjson
`)

提供后续步骤（AUTO_YES 时跳过）

if (!AUTO_YES && failed.length > 0) {
  const answer = functions.request_user_input({
    questions: [{
      header: "下一步",
      id: "next_step",
      question: `${failed.length} tasks failed. Next action?`,
      options: [
        { label: "Retry Failed(Recommended)", description: `Re-execute ${failed.length} failed tasks with updated context` },
        { label: "View Report", description: "Display context.md" },
        { label: "Done", description: "Complete session" }
      ]
    }]
  })  // BLOCKS

  if (answer.answers.next_step.answers[0] === "Retry Failed(Recommended)") {
    // Reset failed tasks to pending, re-run Phase 2 for their waves
    for (const task of failed) {
      updateMasterCsvRow(sessionFolder, task.id, { status: 'pending', error: '' })
    }
    // Also reset skipped tasks whose deps are now retrying
    for (const task of skipped) {
      updateMasterCsvRow(sessionFolder, task.id, { status: 'pending', error: '' })
    }
    // Re-execute Phase 2 (loop will skip already-completed tasks)
    // → goto Phase 2
  } else if (answer.answers.next_step.answers[0] === "View Report") {
    console.log(Read(`${sessionFolder}/context.md`))
  }
}

成功标准：已导出 results.csv、已生成 context.md，并向用户显示摘要。

错误处理

错误	解决方法
循环依赖	在计算波次时检测，并显示错误信息后中止
Agent 超时	在结果中标记为失败，并继续执行当前波次
Agent 失败	标记为失败，并在后续波次中跳过依赖任务
波次中的所有 Agent 均失败	记录错误，并提供重试或中止选项
CSV 解析错误	执行前验证 CSV 格式，并显示行号
discoveries.ndjson 损坏	忽略格式错误的行，继续处理有效条目
继续模式：未找到会话	列出可用会话，并提示用户选择

规则与最佳实践

核心规则

立即开始：首先初始化会话，然后进入第 1 阶段
严格遵守波次顺序：第 N-1 波次完成并合并结果前，绝不执行第 N 波次
CSV 是唯一事实来源：主 tasks.csv 保存所有状态——每次执行波次前都要读取，之后都要写入
上下文传递：prev_context 根据主 CSV 构建，而非依赖记忆
发现板只追加：绝不清空、修改或重新创建 discoveries.ndjson
失败时跳过：如果依赖任务失败，则跳过依赖该任务的任务（不要尝试执行）
清理临时文件：合并结果后，删除 wave-{N}.csv 和 wave-{N}-results.csv
不要停止：持续执行，直到所有波次完成或所有剩余任务都被跳过

任务设计

粒度：3–10 个任务为宜；过多会增加开销，过少则无法发挥并行优势
尽量减少跨波次依赖：第 1 波次的任务越多，并行度越高
描述要具体：Agent 只能看到自己的 CSV 行和 prev_context，因此描述必须自成一体
上下文来源 ≠ 依赖：deps = 执行顺序约束；context_from = 信息流。任务可以设置 context_from 而不设置 deps（它只需读取之前的发现，但不要求这些发现必须在同一波次中先完成）
并发调优：-c 1 用于串行执行（最大限度共享上下文）；-c 8 用于 I/O 密集型任务

场景建议

场景	建议方法
相互独立的并行任务（无依赖）	`$csv-wave-pipeline -c 8` — 单个波次，最大并行度
线性工作流（A→B→C）	`$csv-wave-pipeline -c 1` — 3 个波次，串行执行，完整上下文
菱形依赖（A→B,C→D）	`$csv-wave-pipeline` — 3 个波次，第 2 波次中 B 和 C 并发执行
需求复杂、任务不明确	先使用 `$roadmap-with-file` 进行规划，然后将问题交给此工具
单个复杂任务	改用 `$workflow-lite-plan`

支持 Markdown

登录后评论

基本信息

作者: catlog22
仓库: catlog22/claude-code-workflow
分类: Agent 工作流
安装次数: 46
GitHub Stars: 2,119

报告问题