csv-wave-pipeline について
csv-wave-pipelineは、実装要件を依存関係に沿ったタスクに分解し、複数のウェーブで並行実行するスキルです。ウェーブ間で調査結果を共有し、セッションの再開にも対応します。
主な機能
- 01要件を3〜10件のタスクに分解
- 02依存関係に基づくウェーブ編成
- 03エージェントによる並行実行
- 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:各ウェーブ内で同時に実行するエージェントの最大数(デフォルト: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
コンテキストの伝播
2 つのコンテキストチャネルがウェーブ間で引き継がれます。
- CSV の調査結果(構造化):
context_from列 →prev_contextの注入。タスク固有の、指定されたコンテキストです - NDJSON の発見事項(ブロードキャスト):
discoveries.ndjson。全員が参照できる一般的な調査結果です
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 | 全エージェントが共有する調査ボード | 追記専用で、ウェーブ間で引き継がれる |
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。エージェントの作業範囲を制限し、タスク間のファイル競合を防止します |
hints | 入力 | 実装のヒントと参照ファイル。形式:ヒントのテキスト || 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 列があり、マスター CSV から完了済みタスクの findings を参照して context_from を基に作成されます。
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"
共有ディスカバリーボードプロトコル
すべてのウェーブのすべてのエージェントが discoveries.ndjson を共有します。これにより、コードベースの探索の重複をなくせます。
ライフサイクル:最初のエージェントがディスカバリーを書き込むと作成されます。ウェーブをまたいで引き継がれ、消去されることはありません。エージェントは 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"]}}
ディスカバリーの種類:
| type | 重複判定キー | 説明 |
|---|---|---|
code_pattern | data.name | 見つかった再利用可能なコードパターン |
integration_point | data.file | モジュールの接続ポイント |
convention | singleton | コードスタイルの規約 |
blocker | data.issue | 発生したブロッキング問題 |
tech_stack | singleton | プロジェクトの技術スタック |
test_command | singleton | 見つかったテストコマンド |
プロトコルのルール:
- 自分で探索する前にボードを読む → 対応済みの領域はスキップする
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 は、analyze-with-file または workflow-lite-plan から handoff.json 経由で構造化コンテキストを受け取れます。
// 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`)
}
}
handoff を使った使用方法:
$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"
handoff が指定されている場合:
implementation_scope[]は、事前に分析された目的、対象ファイル、受け入れ基準を使って、Phase 1 の CLI 分解プロンプトを補強しますcode_anchors[]は、エージェントに特定の file:line エントリーポイントを示しますexploration_artifacts.exploration_codebaseは、既知のパターンと統合ポイントを使ってdiscoveries.ndjsonを初期化するため、Wave 1 のエージェントはコードベースの重複探索をスキップできます
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
}
Phase 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 } }
成功基準:有効なスキーマとウェーブ割り当てを含む tasks.csv が作成され、循環依存がなく、ユーザーの承認を得ている(または AUTO_YES)。
Phase 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 にディスカバリーが蓄積されていること。
Phase 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 が生成され、概要がユーザーに表示されていること。
エラー処理
| エラー | 解決方法 |
|---|---|
| 循環依存 | ウェーブの計算中に検出し、エラーメッセージを表示して中止する |
| エージェントのタイムアウト | 結果で失敗としてマークし、ウェーブを続行する |
| エージェントの失敗 | 失敗としてマークし、後続のウェーブでは依存タスクをスキップする |
| ウェーブ内のすべてのエージェントが失敗 | エラーを記録し、再試行または中止を提示する |
| 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 のタスクが多いほど、並列性が高まります
- 具体的な説明:エージェントが参照できるのは、自分の CSV 行と prev_context のみです。説明だけで内容が完結するようにしてください
- コンテキストと依存関係は別物:
depsは実行順序の制約、context_fromは情報の流れを表します。depsがなくてもcontext_fromを指定できます(前の調査結果を読み取るだけで、同じウェーブ内で先に完了する必要はありません) - 並行数の調整:逐次実行(コンテキスト共有を最大化)には
-c 1、I/O 負荷の高いタスクには-c 8を指定します
シナリオ別の推奨事項
| シナリオ | 推奨するアプローチ |
|---|---|
| 依存関係のない独立した並列タスク | $csv-wave-pipeline -c 8 — 1 つのウェーブで、並列性を最大化 |
| 直線的なパイプライン(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 を使います |