Auto Mode
When --yes or -y: Auto-confirm task decomposition, skip interactive validation, use defaults.
CSV Wave Pipeline
Usage
$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"
Flags:
-y, --yes: Skip all confirmations (auto mode)
-c, --concurrency N: Max concurrent agents within each wave (default: 4)
--continue: Resume existing session
Overview
Wave-based batch execution using spawn_agents_on_csv with cross-wave context propagation. Tasks are grouped into dependency waves; each wave executes concurrently, and its results feed into the next wave.
Core workflow: Decompose → Compute Waves → Execute Wave-by-Wave → Aggregate
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
Context Propagation
Two context channels flow across waves:
- CSV findings (structured):
context_from column → prev_context injection — task-specific directed context
- NDJSON discoveries (broadcast):
discoveries.ndjson — general exploration findings available to all
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
Session & Output Structure
.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)
| File |
Purpose |
Lifecycle |
tasks.csv |
Master state — all tasks with status/findings |
Updated after each wave |
wave-{N}.csv |
Per-wave input with prev_context column |
Created before wave, deleted after |
wave-{N}-results.csv |
Per-wave output from spawn_agents_on_csv |
Created during wave, deleted after merge |
results.csv |
Final export of all task results |
Created in Phase 3 |
discoveries.ndjson |
Shared exploration board across all agents |
Append-only, carries across waves |
context.md |
Human-readable execution report |
Created in Phase 3 |
CSV Schema
tasks.csv (Master State)
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","","","","","",""
Columns:
| Column |
Phase |
Description |
id |
Input |
Unique task identifier (string) |
title |
Input |
Short task title |
description |
Input |
Detailed task description — what to implement |
test |
Input |
Test cases: what tests to write and how to verify (unit/integration/edge) |
acceptance_criteria |
Input |
Acceptance criteria: measurable conditions that define "done" |
scope |
Input |
Target file/directory glob — constrains agent work area, prevents cross-task file conflicts |
hints |
Input |
Implementation tips + reference files. Format: tips text || file1;file2. Before || = how to implement; after || = existing files to read before starting. Either part is optional |
execution_directives |
Input |
Execution constraints: commands to run for verification, tool restrictions, environment requirements |
deps |
Input |
Semicolon-separated dependency task IDs (empty = no deps) |
context_from |
Input |
Semicolon-separated task IDs whose findings this task needs |
wave |
Computed |
Wave number (computed by topological sort, 1-based) |
status |
Output |
pending → completed / failed / skipped |
findings |
Output |
Key discoveries or implementation notes (max 500 chars) |
files_modified |
Output |
Semicolon-separated file paths |
tests_passed |
Output |
Whether all defined test cases passed (true/false) |
acceptance_met |
Output |
Summary of which acceptance criteria were met/unmet |
error |
Output |
Error message if failed (empty if success) |
Per-Wave CSV (Temporary)
Each wave generates a temporary wave-{N}.csv with an extra prev_context column built from context_from by looking up completed tasks' findings in the master CSV:
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"
Shared Discovery Board Protocol
All agents across all waves share discoveries.ndjson. This eliminates redundant codebase exploration.
Lifecycle: Created by the first agent to write a discovery. Carries over across waves — never cleared. Agents append via echo '...' >> discoveries.ndjson.
Format: NDJSON, each line is a self-contained 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"]}}
Discovery Types:
| type |
Dedup Key |
Description |
code_pattern |
data.name |
Reusable code pattern found |
integration_point |
data.file |
Module connection point |
convention |
singleton |
Code style conventions |
blocker |
data.issue |
Blocking issue encountered |
tech_stack |
singleton |
Project technology stack |
test_command |
singleton |
Test commands discovered |
Protocol Rules:
- Read board before own exploration → skip covered areas
- Write discoveries immediately via
echo >> → don't batch
- Deduplicate — check existing entries; skip if same type + dedup key exists
- Append-only — never modify or delete existing lines
Implementation
Session Initialization
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}`)
Upstream Handoff Intake (Optional)
csv-wave-pipeline can receive structured context from analyze-with-file or workflow-lite-plan via 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`)
}
}
Usage with 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"
When handoff is provided:
implementation_scope[] enriches the CLI decomposition prompt in Phase 1 with pre-analyzed objectives, target files, and acceptance criteria
code_anchors[] give agents specific file:line entry points
exploration_artifacts.exploration_codebase seeds discoveries.ndjson with known patterns and integration points, so Wave 1 agents skip redundant codebase exploration
CSV Utility Functions
// 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: Requirement → CSV
Objective: Decompose requirement into tasks, compute dependency waves, generate tasks.csv.
Steps:
Decompose Requirement
// 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 ' || ;' (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. **Compute Waves** (Kahn's BFS topological sort with depth tracking)
```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)
Generate 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'))
User Validation (skip if 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
}
}
Success Criteria: tasks.csv created with valid schema and wave assignments, no circular dependencies, user approved (or AUTO_YES).
Phase 2: Wave Execution Engine
Objective: Execute tasks wave-by-wave via spawn_agents_on_csv. Each wave sees previous waves' results.
Steps:
Wave Loop
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`)
}
Instruction Template Builder
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: ` || ;`. 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.
`
}
Master CSV Update Helper
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'))
}
Success Criteria: All waves executed in order, each wave's results merged into master CSV before next wave starts, dependent tasks skipped when predecessor failed, discoveries.ndjson accumulated across all waves.
Phase 3: Results Aggregation
Objective: Generate final results and human-readable report.
Steps:
Export results.csv
const masterCsv = Read(`${sessionFolder}/tasks.csv`)
// results.csv = master CSV (already has all results populated)
Write(`${sessionFolder}/results.csv`, masterCsv)
Generate 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. **Display Summary**
```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
`)
Offer Next Steps (skip if 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`))
}
}
Success Criteria: results.csv exported, context.md generated, summary displayed to user.
Error Handling
| Error |
Resolution |
| Circular dependency |
Detect in wave computation, abort with error message |
| Agent timeout |
Mark as failed in results, continue with wave |
| Agent failed |
Mark as failed, skip dependent tasks in later waves |
| All agents in wave failed |
Log error, offer retry or abort |
| CSV parse error |
Validate CSV format before execution, show line number |
| discoveries.ndjson corrupt |
Ignore malformed lines, continue with valid entries |
| Continue mode: no session found |
List available sessions, prompt user to select |
Rules & Best Practices
Core Rules
- Start Immediately: First action is session initialization, then Phase 1
- Wave Order is Sacred: Never execute wave N before wave N-1 completes and results are merged
- CSV is Source of Truth: Master tasks.csv holds all state — always read before wave, always write after
- Context Propagation: prev_context built from master CSV, not from memory
- Discovery Board is Append-Only: Never clear, modify, or recreate discoveries.ndjson
- Skip on Failure: If a dependency failed, skip the dependent task (don't attempt)
- Cleanup Temp Files: Remove wave-{N}.csv and wave-{N}-results.csv after results are merged
- DO NOT STOP: Continuous execution until all waves complete or all remaining tasks are skipped
Task Design
- Granularity: 3-10 tasks optimal; too many = overhead, too few = no parallelism benefit
- Minimize Cross-Wave Deps: More tasks in wave 1 = more parallelism
- Specific Descriptions: Agent sees only its CSV row + prev_context — make description self-contained
- Context From ≠ Deps:
deps = execution order constraint; context_from = information flow. A task can have context_from without deps (it just reads previous findings but doesn't require them to be done first in its wave)
- Concurrency Tuning:
-c 1 for serial execution (maximum context sharing); -c 8 for I/O-bound tasks
Scenario Recommendations
| Scenario |
Recommended Approach |
| Independent parallel tasks (no deps) |
$csv-wave-pipeline -c 8 — single wave, max parallelism |
| Linear pipeline (A→B→C) |
$csv-wave-pipeline -c 1 — 3 waves, serial, full context |
| Diamond dependency (A→B,C→D) |
$csv-wave-pipeline — 3 waves, B+C concurrent in wave 2 |
| Complex requirement, unclear tasks |
Use $roadmap-with-file first for planning, then feed issues here |
| Single complex task |
Use $workflow-lite-plan instead |
1---2name: csv-wave-pipeline3description: Requirement planning to wave-based CSV execution pipeline. Decomposes requirement into dependency-sorted CSV tasks, computes execution waves, runs wave-by-wave via spawn_agents_on_csv with cross-wave context propagation.4---56## Auto Mode78When `--yes` or `-y`: Auto-confirm task decomposition, skip interactive validation, use defaults.910# CSV Wave Pipeline1112## Usage1314```bash15$csv-wave-pipeline "Implement user authentication with OAuth, JWT, and 2FA"16$csv-wave-pipeline -c 4 "Refactor payment module with Stripe and PayPal"17$csv-wave-pipeline -y "Build notification system with email and SMS"18$csv-wave-pipeline --continue "auth-20260228"19```2021**Flags**:22- `-y, --yes`: Skip all confirmations (auto mode)23- `-c, --concurrency N`: Max concurrent agents within each wave (default: 4)24- `--continue`: Resume existing session2526---2728## Overview2930Wave-based batch execution using `spawn_agents_on_csv` with **cross-wave context propagation**. Tasks are grouped into dependency waves; each wave executes concurrently, and its results feed into the next wave.3132**Core workflow**: Decompose → Compute Waves → Execute Wave-by-Wave → Aggregate3334```35Phase 1: Requirement → CSV36 ├─ Parse requirement into subtasks (3-10 tasks)37 ├─ Identify dependencies (deps column)38 ├─ Compute dependency waves (topological sort → depth grouping)39 ├─ Generate tasks.csv with wave column40 └─ User validates task breakdown (skip if -y)4142Phase 2: Wave Execution Engine43 ├─ For each wave (1..N):44 │ ├─ Build wave CSV (filter rows for this wave)45 │ ├─ Inject previous wave findings into prev_context column46 │ ├─ spawn_agents_on_csv(wave CSV)47 │ ├─ Collect results, merge into master tasks.csv48 │ └─ Check: any failed? → skip dependents or retry49 └─ discoveries.ndjson shared across all waves (append-only)5051Phase 3: Results Aggregation52 ├─ Export final results.csv53 ├─ Generate context.md with all findings54 ├─ Display summary: completed/failed/skipped per wave55 └─ Offer: view results | retry failed | done56```5758### Context Propagation5960Two context channels flow across waves:61621. **CSV findings** (structured): `context_from` column → `prev_context` injection — task-specific directed context632. **NDJSON discoveries** (broadcast): `discoveries.ndjson` — general exploration findings available to all6465```66Wave 1 agents:67 ├─ Execute tasks (no prev_context)68 ├─ Write findings to report_agent_job_result69 └─ Append discoveries to discoveries.ndjson70 ↓ merge results into master CSV71Wave 2 agents:72 ├─ Read discoveries.ndjson (exploration sharing)73 ├─ Read prev_context column (wave 1 findings from context_from)74 ├─ Execute tasks with full upstream context75 ├─ Write findings to report_agent_job_result76 └─ Append new discoveries to discoveries.ndjson77 ↓ merge results into master CSV78Wave 3+ agents: same pattern, accumulated context from all prior waves79```8081---8283## Session & Output Structure8485```86.workflow/.csv-wave/{session-id}/87├── tasks.csv # Master state (updated per wave)88├── results.csv # Final results export (Phase 3)89├── discoveries.ndjson # Shared discovery board (all agents, append-only)90├── context.md # Human-readable report (Phase 3)91├── wave-{N}.csv # Temporary per-wave input (cleaned up after merge)92└── wave-{N}-results.csv # Temporary per-wave output (cleaned up after merge)93```9495| File | Purpose | Lifecycle |96|------|---------|-----------|97| `tasks.csv` | Master state — all tasks with status/findings | Updated after each wave |98| `wave-{N}.csv` | Per-wave input with prev_context column | Created before wave, deleted after |99| `wave-{N}-results.csv` | Per-wave output from spawn_agents_on_csv | Created during wave, deleted after merge |100| `results.csv` | Final export of all task results | Created in Phase 3 |101| `discoveries.ndjson` | Shared exploration board across all agents | Append-only, carries across waves |102| `context.md` | Human-readable execution report | Created in Phase 3 |103104---105106## CSV Schema107108### tasks.csv (Master State)109110```csv111id,title,description,test,acceptance_criteria,scope,hints,execution_directives,deps,context_from,wave,status,findings,files_modified,tests_passed,acceptance_met,error112"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","","","","","",""113"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","","","","","",""114"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","","","","","",""115"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","","","","","",""116```117118**Columns**:119120| Column | Phase | Description |121|--------|-------|-------------|122| `id` | Input | Unique task identifier (string) |123| `title` | Input | Short task title |124| `description` | Input | Detailed task description — what to implement |125| `test` | Input | Test cases: what tests to write and how to verify (unit/integration/edge) |126| `acceptance_criteria` | Input | Acceptance criteria: measurable conditions that define "done" |127| `scope` | Input | Target file/directory glob — constrains agent work area, prevents cross-task file conflicts |128| `hints` | Input | Implementation tips + reference files. Format: `tips text \|\| file1;file2`. Before `\|\|` = how to implement; after `\|\|` = existing files to read before starting. Either part is optional |129| `execution_directives` | Input | Execution constraints: commands to run for verification, tool restrictions, environment requirements |130| `deps` | Input | Semicolon-separated dependency task IDs (empty = no deps) |131| `context_from` | Input | Semicolon-separated task IDs whose findings this task needs |132| `wave` | Computed | Wave number (computed by topological sort, 1-based) |133| `status` | Output | `pending` → `completed` / `failed` / `skipped` |134| `findings` | Output | Key discoveries or implementation notes (max 500 chars) |135| `files_modified` | Output | Semicolon-separated file paths |136| `tests_passed` | Output | Whether all defined test cases passed (true/false) |137| `acceptance_met` | Output | Summary of which acceptance criteria were met/unmet |138| `error` | Output | Error message if failed (empty if success) |139140### Per-Wave CSV (Temporary)141142Each wave generates a temporary `wave-{N}.csv` with an extra `prev_context` column built from `context_from` by looking up completed tasks' `findings` in the master CSV:143144```csv145id,title,description,test,acceptance_criteria,scope,hints,execution_directives,deps,context_from,wave,prev_context146"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"147"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"148```149150---151152## Shared Discovery Board Protocol153154All agents across all waves share `discoveries.ndjson`. This eliminates redundant codebase exploration.155156**Lifecycle**: Created by the first agent to write a discovery. Carries over across waves — never cleared. Agents append via `echo '...' >> discoveries.ndjson`.157158**Format**: NDJSON, each line is a self-contained JSON:159160```jsonl161{"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"}}162{"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"]}}163```164165**Discovery Types**:166167| type | Dedup Key | Description |168|------|-----------|-------------|169| `code_pattern` | `data.name` | Reusable code pattern found |170| `integration_point` | `data.file` | Module connection point |171| `convention` | singleton | Code style conventions |172| `blocker` | `data.issue` | Blocking issue encountered |173| `tech_stack` | singleton | Project technology stack |174| `test_command` | singleton | Test commands discovered |175176**Protocol Rules**:1771. Read board before own exploration → skip covered areas1782. Write discoveries immediately via `echo >>` → don't batch1793. Deduplicate — check existing entries; skip if same type + dedup key exists1804. Append-only — never modify or delete existing lines181182---183184## Implementation185186### Session Initialization187188```javascript189const getUtc8ISOString = () => new Date(Date.now() + 8 * 60 * 60 * 1000).toISOString()190191// Parse flags192const AUTO_YES = $ARGUMENTS.includes('--yes') || $ARGUMENTS.includes('-y')193const continueMode = $ARGUMENTS.includes('--continue')194const concurrencyMatch = $ARGUMENTS.match(/(?:--concurrency|-c)\s+(\d+)/)195const maxConcurrency = concurrencyMatch ? parseInt(concurrencyMatch[1]) : 4196197// Clean requirement text (remove flags — word-boundary safe)198const requirement = $ARGUMENTS199 .replace(/--yes|(?:^|\s)-y(?=\s|$)|--continue|--concurrency\s+\d+|-c\s+\d+/g, '')200 .trim()201202let sessionId, sessionFolder203204const slug = requirement.toLowerCase()205 .replace(/[^a-z0-9\u4e00-\u9fa5]+/g, '-')206 .substring(0, 40)207const dateStr = getUtc8ISOString().substring(0, 10).replace(/-/g, '')208sessionId = `cwp-${dateStr}-${slug}`209sessionFolder = `.workflow/.csv-wave/${sessionId}`210211// Continue mode: find existing session212if (continueMode) {213 const existing = Bash(`ls -t .workflow/.csv-wave/ 2>/dev/null | head -1`).trim()214 if (existing) {215 sessionId = existing216 sessionFolder = `.workflow/.csv-wave/${sessionId}`217 // Read existing tasks.csv, find incomplete waves, resume from there218 const existingCsv = Read(`${sessionFolder}/tasks.csv`)219 // → jump to Phase 2 with remaining waves220 }221}222223Bash(`mkdir -p ${sessionFolder}`)224```225226### Upstream Handoff Intake (Optional)227228csv-wave-pipeline can receive structured context from analyze-with-file or workflow-lite-plan via handoff.json:229230```javascript231// Check if requirement references a prior analysis session232const handoffPathMatch = requirement.match(/handoff:(.+\.json)/)233let handoffContext = null234235if (handoffPathMatch) {236 const handoffPath = handoffPathMatch[1]237 if (file_exists(handoffPath)) {238 handoffContext = JSON.parse(Read(handoffPath))239 // handoffContext: { source, session_id, summary, implementation_scope[], code_anchors[], key_findings[], exploration_artifacts{} }240241 // Enrich requirement with handoff context for CLI decomposition in Phase 1242 requirement = `${handoffContext.summary}\n\n` +243 `## Implementation Scope\n${handoffContext.implementation_scope.map((s, i) =>244 `${i+1}. **${s.objective}** [${s.priority}]\n Files: ${s.target_files?.join(', ') || 'TBD'}\n Done when: ${s.acceptance_criteria?.join('; ') || 'TBD'}`245 ).join('\n')}\n\n` +246 (handoffContext.key_findings?.length > 0247 ? `## Key Findings\n${handoffContext.key_findings.map(f => `- ${f.point || f}`).join('\n')}\n\n`248 : '') +249 (handoffContext.code_anchors?.length > 0250 ? `## Code Anchors\n${handoffContext.code_anchors.slice(0, 8).map(a => `- \`${a.file}:${a.lines}\`: ${a.significance}`).join('\n')}\n\n`251 : '')252253 // Load exploration artifacts into discoveries.ndjson seed (if available)254 if (handoffContext.exploration_artifacts?.exploration_codebase && file_exists(handoffContext.exploration_artifacts.exploration_codebase)) {255 const codebaseData = JSON.parse(Read(handoffContext.exploration_artifacts.exploration_codebase))256 const seedDiscoveries = [257 ...(codebaseData.patterns || []).map(p => JSON.stringify({258 ts: getUtc8ISOString(), worker: 'handoff', type: 'code_pattern',259 data: { name: p.pattern || p, file: p.files?.[0] || '', description: p.description || '' }260 })),261 ...(codebaseData.relevant_files || []).slice(0, 5).map(f => JSON.stringify({262 ts: getUtc8ISOString(), worker: 'handoff', type: 'integration_point',263 data: { file: f.path, description: f.annotation || f.summary || '' }264 }))265 ]266 if (seedDiscoveries.length > 0) {267 Write(`${sessionFolder}/discoveries.ndjson`, seedDiscoveries.join('\n') + '\n')268 }269 }270271 console.log(`[Handoff] Loaded from ${handoffContext.source} session ${handoffContext.session_id}: ${handoffContext.implementation_scope?.length || 0} scope items`)272 }273}274```275276**Usage with handoff**:277```bash278$csv-wave-pipeline "handoff:.workflow/.analysis/ANL-2026-04-29-auth/handoff.json"279$csv-wave-pipeline -y "handoff:.workflow/.lite-plan/auth-plan/handoff.json"280```281282When handoff is provided:283- `implementation_scope[]` enriches the CLI decomposition prompt in Phase 1 with pre-analyzed objectives, target files, and acceptance criteria284- `code_anchors[]` give agents specific file:line entry points285- `exploration_artifacts.exploration_codebase` seeds `discoveries.ndjson` with known patterns and integration points, so Wave 1 agents skip redundant codebase exploration286287### CSV Utility Functions288289```javascript290// Escape a value for CSV (wrap in quotes, double internal quotes)291function csvEscape(value) {292 const str = String(value ?? '')293 return str.replace(/"/g, '""')294}295296// Parse CSV string into array of objects (header row → keys)297function parseCsv(csvString) {298 const lines = csvString.trim().split('\n')299 if (lines.length < 2) return []300 const headers = parseCsvLine(lines[0]).map(h => h.replace(/^"|"$/g, ''))301 return lines.slice(1).map(line => {302 const cells = parseCsvLine(line).map(c => c.replace(/^"|"$/g, '').replace(/""/g, '"'))303 const obj = {}304 headers.forEach((h, i) => { obj[h] = cells[i] ?? '' })305 return obj306 })307}308309// Parse a single CSV line respecting quoted fields with commas/newlines310function parseCsvLine(line) {311 const cells = []312 let current = ''313 let inQuotes = false314 for (let i = 0; i < line.length; i++) {315 const ch = line[i]316 if (inQuotes) {317 if (ch === '"' && line[i + 1] === '"') {318 current += '"'319 i++ // skip escaped quote320 } else if (ch === '"') {321 inQuotes = false322 } else {323 current += ch324 }325 } else {326 if (ch === '"') {327 inQuotes = true328 } else if (ch === ',') {329 cells.push(current)330 current = ''331 } else {332 current += ch333 }334 }335 }336 cells.push(current)337 return cells338}339```340341---342343### Phase 1: Requirement → CSV344345**Objective**: Decompose requirement into tasks, compute dependency waves, generate tasks.csv.346347**Steps**:3483491. **Decompose Requirement**350351 ```javascript352 // Use ccw cli to decompose requirement into subtasks353 Bash({354 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.355TASK:356 • Parse requirement into independent subtasks357 • Identify dependencies between tasks (which must complete before others)358 • Identify context flow (which tasks need previous tasks' findings)359 • For each task, define concrete test cases (unit/integration/edge)360 • For each task, define measurable acceptance criteria (what defines 'done')361 • Each task must be executable by a single agent with file read/write access362MODE: analysis363CONTEXT: @**/*364EXPECTED: 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[]}.365 - description: what to implement (specific enough for an agent to execute independently)366 - test: what tests to write and how to verify (e.g. 'Unit test: X returns Y; Edge test: handles Z')367 - acceptance_criteria: measurable conditions that define done (e.g. 'API returns 200; token expires after 1h')368 - scope: target file/directory glob (e.g. 'src/auth/**') — tasks in same wave MUST have non-overlapping scopes369 - hints: implementation tips + reference files, format '<tips> || <ref_file1>;<ref_file2>' (e.g. 'Use strategy pattern || src/base/Strategy.ts;docs/design.md')370 - execution_directives: commands to run for verification or tool constraints (e.g. 'Run npm test --bail; Ensure tsc passes')371 - deps: task IDs that must complete first372 - context_from: task IDs whose findings are needed373CONSTRAINTS: 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 scopes374375REQUIREMENT: ${requirement}" --tool gemini --mode analysis --rule planning-breakdown-task-steps`,376 run_in_background: true377 })378 // Wait for CLI completion via hook callback379 // Parse JSON from CLI output → decomposedTasks[]380 ```3813822. **Compute Waves** (Kahn's BFS topological sort with depth tracking)383384 ```javascript385 // Algorithm:386 // 1. Build in-degree map and adjacency list from deps387 // 2. Enqueue all tasks with in-degree 0 at wave 1388 // 3. BFS: for each dequeued task at wave W, for each dependent D:389 // - Decrement D's in-degree390 // - D.wave = max(D.wave, W + 1)391 // - If D's in-degree reaches 0, enqueue D392 // 4. Any task without wave assignment → circular dependency error393 //394 // Wave properties:395 // Wave 1: no dependencies — fully independent396 // Wave N: all deps in waves 1..(N-1) — guaranteed completed before start397 // Within a wave: tasks are independent → safe for concurrent execution398 //399 // Example:400 // A(no deps)→W1, B(no deps)→W1, C(deps:A)→W2, D(deps:A,B)→W2, E(deps:C,D)→W3401 // Wave 1: [A,B] concurrent → Wave 2: [C,D] concurrent → Wave 3: [E]402403 function computeWaves(tasks) {404 const taskMap = new Map(tasks.map(t => [t.id, t]))405 const inDegree = new Map(tasks.map(t => [t.id, 0]))406 const adjList = new Map(tasks.map(t => [t.id, []]))407408 for (const task of tasks) {409 for (const dep of task.deps) {410 if (taskMap.has(dep)) {411 adjList.get(dep).push(task.id)412 inDegree.set(task.id, inDegree.get(task.id) + 1)413 }414 }415 }416417 // BFS-based topological sort with depth tracking418 const queue = [] // [taskId, depth]419 const waveAssignment = new Map()420421 for (const [id, deg] of inDegree) {422 if (deg === 0) {423 queue.push([id, 1])424 waveAssignment.set(id, 1)425 }426 }427428 let maxWave = 1429 let idx = 0430 while (idx < queue.length) {431 const [current, depth] = queue[idx++]432 for (const next of adjList.get(current)) {433 const newDeg = inDegree.get(next) - 1434 inDegree.set(next, newDeg)435 const nextDepth = Math.max(waveAssignment.get(next) || 0, depth + 1)436 waveAssignment.set(next, nextDepth)437 if (newDeg === 0) {438 queue.push([next, nextDepth])439 maxWave = Math.max(maxWave, nextDepth)440 }441 }442 }443444 // Detect cycles445 for (const task of tasks) {446 if (!waveAssignment.has(task.id)) {447 throw new Error(`Circular dependency detected involving task ${task.id}`)448 }449 }450451 return { waveAssignment, maxWave }452 }453454 const { waveAssignment, maxWave } = computeWaves(decomposedTasks)455 ```4564573. **Generate tasks.csv**458459 ```javascript460 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'461 const rows = decomposedTasks.map(task => {462 const wave = waveAssignment.get(task.id)463 return [464 task.id,465 csvEscape(task.title),466 csvEscape(task.description),467 csvEscape(task.test),468 csvEscape(task.acceptance_criteria),469 csvEscape(task.scope),470 csvEscape(task.hints),471 csvEscape(task.execution_directives),472 task.deps.join(';'),473 task.context_from.join(';'),474 wave,475 'pending', // status476 '', // findings477 '', // files_modified478 '', // tests_passed479 '', // acceptance_met480 '' // error481 ].map(cell => `"${String(cell).replace(/"/g, '""')}"`).join(',')482 })483484 Write(`${sessionFolder}/tasks.csv`, [header, ...rows].join('\n'))485 ```4864874. **User Validation** (skip if AUTO_YES)488489 ```javascript490 if (!AUTO_YES) {491 // Display task breakdown with wave assignment492 console.log(`\n## Task Breakdown (${decomposedTasks.length} tasks, ${maxWave} waves)\n`)493 for (let w = 1; w <= maxWave; w++) {494 const waveTasks = decomposedTasks.filter(t => waveAssignment.get(t.id) === w)495 console.log(`### Wave ${w} (${waveTasks.length} tasks, concurrent)`)496 waveTasks.forEach(t => console.log(` - [${t.id}] ${t.title}`))497 }498499 const answer = functions.request_user_input({500 questions: [{501 header: "验证",502 id: "validation",503 question: "Approve task breakdown?",504 options: [505 { label: "Approve(Recommended)", description: "Proceed with wave execution" },506 { label: "Modify", description: `Edit ${sessionFolder}/tasks.csv manually, then --continue` },507 { label: "Cancel", description: "Abort" }508 ]509 }]510 }) // BLOCKS511512 if (answer.answers.validation.answers[0] === "Modify") {513 console.log(`Edit: ${sessionFolder}/tasks.csv\nResume: $csv-wave-pipeline --continue`)514 return515 } else if (answer.answers.validation.answers[0] === "Cancel") {516 return517 }518 }519 ```520521**Success Criteria**: tasks.csv created with valid schema and wave assignments, no circular dependencies, user approved (or AUTO_YES).522523---524525### Phase 2: Wave Execution Engine526527**Objective**: Execute tasks wave-by-wave via `spawn_agents_on_csv`. Each wave sees previous waves' results.528529**Steps**:5305311. **Wave Loop**532533 ```javascript534 const failedIds = new Set()535 const skippedIds = new Set()536537 for (let wave = 1; wave <= maxWave; wave++) {538 console.log(`\n## Wave ${wave}/${maxWave}\n`)539540 // 1. Read current master CSV541 const masterCsv = parseCsv(Read(`${sessionFolder}/tasks.csv`))542543 // 2. Filter tasks for this wave544 const waveTasks = masterCsv.filter(row => parseInt(row.wave) === wave)545546 // 3. Skip tasks whose deps failed547 const executableTasks = []548 for (const task of waveTasks) {549 const deps = task.deps.split(';').filter(Boolean)550 if (deps.some(d => failedIds.has(d) || skippedIds.has(d))) {551 skippedIds.add(task.id)552 updateMasterCsvRow(sessionFolder, task.id, {553 status: 'skipped',554 error: 'Dependency failed or skipped'555 })556 console.log(` [${task.id}] ${task.title} → SKIPPED (dependency failed)`)557 continue558 }559 executableTasks.push(task)560 }561562 if (executableTasks.length === 0) {563 console.log(` No executable tasks in wave ${wave}`)564 continue565 }566567 // 4. Build prev_context for each task (from context_from → master CSV findings)568 for (const task of executableTasks) {569 const contextIds = task.context_from.split(';').filter(Boolean)570 const prevFindings = contextIds571 .map(id => {572 const prevRow = masterCsv.find(r => r.id === id)573 if (prevRow && prevRow.status === 'completed' && prevRow.findings) {574 return `[Task ${id}: ${prevRow.title}] ${prevRow.findings}`575 }576 return null577 })578 .filter(Boolean)579 .join('\n')580 task.prev_context = prevFindings || 'No previous context available'581 }582583 // 5. Write wave CSV584 const waveHeader = 'id,title,description,test,acceptance_criteria,scope,hints,execution_directives,deps,context_from,wave,prev_context'585 const waveRows = executableTasks.map(t =>586 [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]587 .map(cell => `"${String(cell).replace(/"/g, '""')}"`)588 .join(',')589 )590 Write(`${sessionFolder}/wave-${wave}.csv`, [waveHeader, ...waveRows].join('\n'))591592 // 6. Execute wave593 console.log(` Executing ${executableTasks.length} tasks (concurrency: ${maxConcurrency})...`)594595 const waveResult = spawn_agents_on_csv({596 csv_path: `${sessionFolder}/wave-${wave}.csv`,597 id_column: "id",598 instruction: buildInstructionTemplate(sessionFolder, wave),599 max_concurrency: maxConcurrency,600 max_runtime_seconds: 600,601 output_csv_path: `${sessionFolder}/wave-${wave}-results.csv`,602 output_schema: {603 type: "object",604 properties: {605 id: { type: "string" },606 status: { type: "string", enum: ["completed", "failed"] },607 findings: { type: "string" },608 files_modified: { type: "array", items: { type: "string" } },609 tests_passed: { type: "boolean" },610 acceptance_met: { type: "string" },611 error: { type: "string" }612 },613 required: ["id", "status", "findings", "tests_passed"]614 }615 })616 // ↑ Blocks until all agents in this wave complete617618 // 7. Merge results into master CSV619 const waveResults = parseCsv(Read(`${sessionFolder}/wave-${wave}-results.csv`))620 for (const result of waveResults) {621 updateMasterCsvRow(sessionFolder, result.id, {622 status: result.status,623 findings: result.findings || '',624 files_modified: (result.files_modified || []).join(';'),625 tests_passed: String(result.tests_passed ?? ''),626 acceptance_met: result.acceptance_met || '',627 error: result.error || ''628 })629630 if (result.status === 'failed') {631 failedIds.add(result.id)632 console.log(` [${result.id}] ${result.title} → FAILED: ${result.error}`)633 } else {634 console.log(` [${result.id}] ${result.title} → COMPLETED`)635 }636 }637638 // 8. Cleanup temporary wave CSVs639 Bash(`rm -f "${sessionFolder}/wave-${wave}.csv" "${sessionFolder}/wave-${wave}-results.csv"`)640641 console.log(` Wave ${wave} done: ${waveResults.filter(r => r.status === 'completed').length} completed, ${waveResults.filter(r => r.status === 'failed').length} failed`)642 }643 ```6446452. **Instruction Template Builder**646647 ```javascript648 function buildInstructionTemplate(sessionFolder, wave) {649 return `650## TASK ASSIGNMENT651652### MANDATORY FIRST STEPS6531. Read shared discoveries: ${sessionFolder}/discoveries.ndjson (if exists, skip if not)6542. Read project context: .workflow/project-tech.json (if exists)655656---657658## Your Task659660**Task ID**: {id}661**Title**: {title}662**Description**: {description}663**Scope**: {scope}664665### Implementation Hints & Reference Files666{hints}667668> Format: \`<tips> || <ref_file1>;<ref_file2>\`. Read ALL reference files (after ||) before starting implementation. Apply tips (before ||) as implementation guidance.669670### Execution Directives671{execution_directives}672673> Commands to run for verification, tool restrictions, or environment requirements. Follow these constraints during and after implementation.674675### Test Cases676{test}677678### Acceptance Criteria679{acceptance_criteria}680681### Previous Tasks' Findings (Context)682{prev_context}683684---685686## Execution Protocol6876881. **Read references**: Parse {hints} — read all files listed after \`||\` to understand existing patterns6892. **Read discoveries**: Load ${sessionFolder}/discoveries.ndjson for shared exploration findings6903. **Use context**: Apply previous tasks' findings from prev_context above6914. **Stay in scope**: ONLY create/modify files within {scope} — do NOT touch files outside this boundary6925. **Apply hints**: Follow implementation tips from {hints} (before \`||\`)6936. **Execute**: Implement the task as described6947. **Write tests**: Implement the test cases defined above6958. **Run directives**: Execute commands from {execution_directives} to verify your work6969. **Verify acceptance**: Ensure all acceptance criteria are met before reporting completion69710. **Share discoveries**: Append exploration findings to shared board:698 \`\`\`bash699 echo '{"ts":"<ISO8601>","worker":"{id}","type":"<type>","data":{...}}' >> ${sessionFolder}/discoveries.ndjson700 \`\`\`70111. **Report result**: Return JSON via report_agent_job_result702703### Discovery Types to Share704- \`code_pattern\`: {name, file, description} — reusable patterns found705- \`integration_point\`: {file, description, exports[]} — module connection points706- \`convention\`: {naming, imports, formatting} — code style conventions707- \`blocker\`: {issue, severity, impact} — blocking issues encountered708- \`tech_stack\`: {runtime, framework, language} — project technology stack709- \`test_command\`: {command, scope, description} — test commands discovered710711---712713## Output (report_agent_job_result)714715Return JSON:716{717 "id": "{id}",718 "status": "completed" | "failed",719 "findings": "Key discoveries and implementation notes (max 500 chars)",720 "files_modified": ["path1", "path2"],721 "tests_passed": true | false,722 "acceptance_met": "Summary of which acceptance criteria were met/unmet",723 "error": ""724}725726**IMPORTANT**: Set status to "completed" ONLY if:727- All test cases pass728- All acceptance criteria are met729Otherwise set status to "failed" with details in error field.730`731 }732 ```7337343. **Master CSV Update Helper**735736 ```javascript737 function updateMasterCsvRow(sessionFolder, taskId, updates) {738 const csvPath = `${sessionFolder}/tasks.csv`739 const content = Read(csvPath)740 const lines = content.split('\n')741 const header = lines[0].split(',')742743 for (let i = 1; i < lines.length; i++) {744 const cells = parseCsvLine(lines[i])745 if (cells[0] === taskId || cells[0] === `"${taskId}"`) {746 // Update specified columns747 for (const [col, val] of Object.entries(updates)) {748 const colIdx = header.indexOf(col)749 if (colIdx >= 0) {750 cells[colIdx] = `"${String(val).replace(/"/g, '""')}"`751 }752 }753 lines[i] = cells.join(',')754 break755 }756 }757758 Write(csvPath, lines.join('\n'))759 }760 ```761762**Success Criteria**: All waves executed in order, each wave's results merged into master CSV before next wave starts, dependent tasks skipped when predecessor failed, discoveries.ndjson accumulated across all waves.763764---765766### Phase 3: Results Aggregation767768**Objective**: Generate final results and human-readable report.769770**Steps**:7717721. **Export results.csv**773774 ```javascript775 const masterCsv = Read(`${sessionFolder}/tasks.csv`)776 // results.csv = master CSV (already has all results populated)777 Write(`${sessionFolder}/results.csv`, masterCsv)778 ```7797802. **Generate context.md**781782 ```javascript783 const tasks = parseCsv(masterCsv)784 const completed = tasks.filter(t => t.status === 'completed')785 const failed = tasks.filter(t => t.status === 'failed')786 const skipped = tasks.filter(t => t.status === 'skipped')787788 const contextContent = `# CSV Batch Execution Report789790**Session**: ${sessionId}791**Requirement**: ${requirement}792**Completed**: ${getUtc8ISOString()}793**Waves**: ${maxWave} | **Concurrency**: ${maxConcurrency}794795---796797## Summary798799| Metric | Count |800|--------|-------|801| Total Tasks | ${tasks.length} |802| Completed | ${completed.length} |803| Failed | ${failed.length} |804| Skipped | ${skipped.length} |805| Waves | ${maxWave} |806807---808809## Wave Execution810811${Array.from({ length: maxWave }, (_, i) => i + 1).map(w => {812 const waveTasks = tasks.filter(t => parseInt(t.wave) === w)813 return `### Wave ${w}814${waveTasks.map(t => `- **[${t.id}] ${t.title}**: ${t.status}${t.tests_passed ? ' ✓tests' : ''}${t.error ? ' — ' + t.error : ''}815 ${t.findings ? 'Findings: ' + t.findings : ''}`).join('\n')}`816}).join('\n\n')}817818---819820## Task Details821822${tasks.map(t => `### ${t.id}: ${t.title}823824| Field | Value |825|-------|-------|826| Status | ${t.status} |827| Wave | ${t.wave} |828| Scope | ${t.scope || 'none'} |829| Dependencies | ${t.deps || 'none'} |830| Context From | ${t.context_from || 'none'} |831| Tests Passed | ${t.tests_passed || 'N/A'} |832| Acceptance Met | ${t.acceptance_met || 'N/A'} |833| Error | ${t.error || 'none'} |834835**Description**: ${t.description}836837**Test Cases**: ${t.test || 'N/A'}838839**Acceptance Criteria**: ${t.acceptance_criteria || 'N/A'}840841**Hints**: ${t.hints || 'N/A'}842843**Execution Directives**: ${t.execution_directives || 'N/A'}844845**Findings**: ${t.findings || 'N/A'}846847**Files Modified**: ${t.files_modified || 'none'}848`).join('\n---\n')}849850---851852## All Modified Files853854${[...new Set(tasks.flatMap(t => (t.files_modified || '').split(';')).filter(Boolean))].map(f => '- ' + f).join('\n') || 'None'}855`856857 Write(`${sessionFolder}/context.md`, contextContent)858 ```8598603. **Display Summary**861862 ```javascript863 console.log(`864## Execution Complete865866- **Session**: ${sessionId}867- **Waves**: ${maxWave}868- **Completed**: ${completed.length}/${tasks.length}869- **Failed**: ${failed.length}870- **Skipped**: ${skipped.length}871872**Results**: ${sessionFolder}/results.csv873**Report**: ${sessionFolder}/context.md874**Discoveries**: ${sessionFolder}/discoveries.ndjson875`)876 ```8778784. **Offer Next Steps** (skip if AUTO_YES)879880 ```javascript881 if (!AUTO_YES && failed.length > 0) {882 const answer = functions.request_user_input({883 questions: [{884 header: "下一步",885 id: "next_step",886 question: `${failed.length} tasks failed. Next action?`,887 options: [888 { label: "Retry Failed(Recommended)", description: `Re-execute ${failed.length} failed tasks with updated context` },889 { label: "View Report", description: "Display context.md" },890 { label: "Done", description: "Complete session" }891 ]892 }]893 }) // BLOCKS894895 if (answer.answers.next_step.answers[0] === "Retry Failed(Recommended)") {896 // Reset failed tasks to pending, re-run Phase 2 for their waves897 for (const task of failed) {898 updateMasterCsvRow(sessionFolder, task.id, { status: 'pending', error: '' })899 }900 // Also reset skipped tasks whose deps are now retrying901 for (const task of skipped) {902 updateMasterCsvRow(sessionFolder, task.id, { status: 'pending', error: '' })903 }904 // Re-execute Phase 2 (loop will skip already-completed tasks)905 // → goto Phase 2906 } else if (answer.answers.next_step.answers[0] === "View Report") {907 console.log(Read(`${sessionFolder}/context.md`))908 }909 }910 ```911912**Success Criteria**: results.csv exported, context.md generated, summary displayed to user.913914---915916## Error Handling917918| Error | Resolution |919|-------|------------|920| Circular dependency | Detect in wave computation, abort with error message |921| Agent timeout | Mark as failed in results, continue with wave |922| Agent failed | Mark as failed, skip dependent tasks in later waves |923| All agents in wave failed | Log error, offer retry or abort |924| CSV parse error | Validate CSV format before execution, show line number |925| discoveries.ndjson corrupt | Ignore malformed lines, continue with valid entries |926| Continue mode: no session found | List available sessions, prompt user to select |927928---929930## Rules & Best Practices931932### Core Rules9339341. **Start Immediately**: First action is session initialization, then Phase 19352. **Wave Order is Sacred**: Never execute wave N before wave N-1 completes and results are merged9363. **CSV is Source of Truth**: Master tasks.csv holds all state — always read before wave, always write after9374. **Context Propagation**: prev_context built from master CSV, not from memory9385. **Discovery Board is Append-Only**: Never clear, modify, or recreate discoveries.ndjson9396. **Skip on Failure**: If a dependency failed, skip the dependent task (don't attempt)9407. **Cleanup Temp Files**: Remove wave-{N}.csv and wave-{N}-results.csv after results are merged9418. **DO NOT STOP**: Continuous execution until all waves complete or all remaining tasks are skipped942943### Task Design944945- **Granularity**: 3-10 tasks optimal; too many = overhead, too few = no parallelism benefit946- **Minimize Cross-Wave Deps**: More tasks in wave 1 = more parallelism947- **Specific Descriptions**: Agent sees only its CSV row + prev_context — make description self-contained948- **Context From ≠ Deps**: `deps` = execution order constraint; `context_from` = information flow. A task can have `context_from` without `deps` (it just reads previous findings but doesn't require them to be done first in its wave)949- **Concurrency Tuning**: `-c 1` for serial execution (maximum context sharing); `-c 8` for I/O-bound tasks950951### Scenario Recommendations952953| Scenario | Recommended Approach |954|----------|---------------------|955| Independent parallel tasks (no deps) | `$csv-wave-pipeline -c 8` — single wave, max parallelism |956| Linear pipeline (A→B→C) | `$csv-wave-pipeline -c 1` — 3 waves, serial, full context |957| Diamond dependency (A→B,C→D) | `$csv-wave-pipeline` — 3 waves, B+C concurrent in wave 2 |958| Complex requirement, unclear tasks | Use `$roadmap-with-file` first for planning, then feed issues here |959| Single complex task | Use `$workflow-lite-plan` instead |