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
Output Directory: .workflow/.csv-wave/{session-id}/
Core Output: tasks.csv (master state) + results.csv (final) + discoveries.ndjson (shared exploration) + context.md (human-readable report)
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
┌─────────────────────────────────────────────────────────────────────────┐
│ CSV BATCH EXECUTION WORKFLOW │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ Phase 1: Requirement → CSV │
│ ├─ Parse requirement into subtasks (3-10 tasks) │
│ ├─ Identify dependencies (deps column) │
│ ├─ Compute dependency waves (topological sort → depth grouping) │
│ ├─ Generate tasks.csv with wave column │
│ └─ User validates task breakdown (skip if -y) │
│ │
│ Phase 2: Wave Execution Engine │
│ ├─ For each wave (1..N): │
│ │ ├─ Build wave CSV (filter rows for this wave) │
│ │ ├─ Inject previous wave findings into prev_context column │
│ │ ├─ spawn_agents_on_csv(wave CSV) │
│ │ ├─ Collect results, merge into master tasks.csv │
│ │ └─ Check: any failed? → skip dependents or retry │
│ └─ discoveries.ndjson shared across all waves (append-only) │
│ │
│ Phase 3: Results Aggregation │
│ ├─ Export final results.csv │
│ ├─ Generate context.md with all findings │
│ ├─ Display summary: completed/failed/skipped per wave │
│ └─ Offer: view results | retry failed | done │
│ │
└─────────────────────────────────────────────────────────────────────────┘
CSV 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:
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"
The prev_context column is built from context_from by looking up completed tasks' findings in the master CSV.
Output Artifacts
| File |
Purpose |
Lifecycle |
tasks.csv |
Master state — all tasks with status/findings |
Updated after each wave |
wave-{N}.csv |
Per-wave input (temporary) |
Created before wave, deleted after |
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 |
Session Structure
.workflow/.csv-wave/{session-id}/
├── tasks.csv # Master state (updated per wave)
├── results.csv # Final results export
├── discoveries.ndjson # Shared discovery board (all agents)
├── context.md # Human-readable report
└── wave-{N}.csv # Temporary per-wave input (cleaned up)
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)
const requirement = $ARGUMENTS
.replace(/--yes|-y|--continue|--concurrency\s+\d+|-c\s+\d+/g, '')
.trim()
const slug = requirement.toLowerCase()
.replace(/[^a-z0-9\u4e00-\u9fa5]+/g, '-')
.substring(0, 40)
const dateStr = getUtc8ISOString().substring(0, 10).replace(/-/g, '')
const sessionId = `cwp-${slug}-${dateStr}`
const 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}`)
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** (Topological Sort → Depth Grouping)
```javascript
function computeWaves(tasks) {
// Build adjacency: task.deps → predecessors
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: any task without wave assignment
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 = AskUserQuestion({
questions: [{
question: "Approve task breakdown?",
header: "Validation",
multiSelect: false,
options: [
{ label: "Approve", description: "Proceed with wave execution" },
{ label: "Modify", description: `Edit ${sessionFolder}/tasks.csv manually, then --continue` },
{ label: "Cancel", description: "Abort" }
]
}]
}) // BLOCKS
if (answer.Validation === "Modify") {
console.log(`Edit: ${sessionFolder}/tasks.csv\nResume: $csv-wave-pipeline --continue`)
return
} else if (answer.Validation === "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)
// Update master CSV: mark as skipped
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
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 CSV
Bash(`rm -f "${sessionFolder}/wave-${wave}.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
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 = AskUserQuestion({
questions: [{
question: `${failed.length} tasks failed. Next action?`,
header: "Next Step",
multiSelect: false,
options: [
{ label: "Retry Failed", 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['Next Step'] === "Retry Failed") {
// 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['Next Step'] === "View Report") {
console.log(Read(`${sessionFolder}/context.md`))
}
}
Success Criteria:
- results.csv exported
- context.md generated
- Summary displayed to user
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
Wave Computation Details
Algorithm
Kahn's BFS topological sort with depth tracking:
Input: tasks[] with deps[]
Output: waveAssignment (taskId → wave number)
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 task 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 — all tasks in wave 1 are fully independent
- Wave N: All dependencies are in waves 1..(N-1) — guaranteed completed before wave N starts
- Within a wave: Tasks are independent of each other → safe for concurrent execution
Example
Task A (no deps) → Wave 1
Task B (no deps) → Wave 1
Task C (deps: A) → Wave 2
Task D (deps: A, B) → Wave 2
Task E (deps: C, D) → Wave 3
Execution:
Wave 1: [A, B] ← concurrent
Wave 2: [C, D] ← concurrent, sees A+B findings
Wave 3: [E] ← sees A+B+C+D findings
Context Propagation Flow
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:
├─ Read discoveries.ndjson (accumulated from waves 1+2)
├─ Read prev_context column (wave 1+2 findings from context_from)
├─ Execute tasks
└─ ...
Two context channels:
- CSV findings (structured):
context_from column → prev_context injection — task-specific directed context
- NDJSON discoveries (broadcast):
discoveries.ndjson — general exploration findings available to all
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 |
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 after results are merged
- DO NOT STOP: Continuous execution until all waves complete or all remaining tasks are skipped
Best Practices
- Task 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
Usage 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---5
6## Auto Mode
7
8When `--yes` or `-y`: Auto-confirm task decomposition, skip interactive validation, use defaults.
9
10# CSV Wave Pipeline
11
12## Usage
13
14```bash
15$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```
20
21**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 session
25
26**Output Directory**: `.workflow/.csv-wave/{session-id}/`
27**Core Output**: `tasks.csv` (master state) + `results.csv` (final) + `discoveries.ndjson` (shared exploration) + `context.md` (human-readable report)
28
29---
30
31## Overview
32
33Wave-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.
34
35**Core workflow**: Decompose → Compute Waves → Execute Wave-by-Wave → Aggregate
36
37```
38┌─────────────────────────────────────────────────────────────────────────┐
39│ CSV BATCH EXECUTION WORKFLOW │
40├─────────────────────────────────────────────────────────────────────────┤
41│ │
42│ Phase 1: Requirement → CSV │
43│ ├─ Parse requirement into subtasks (3-10 tasks) │
44│ ├─ Identify dependencies (deps column) │
45│ ├─ Compute dependency waves (topological sort → depth grouping) │
46│ ├─ Generate tasks.csv with wave column │
47│ └─ User validates task breakdown (skip if -y) │
48│ │
49│ Phase 2: Wave Execution Engine │
50│ ├─ For each wave (1..N): │
51│ │ ├─ Build wave CSV (filter rows for this wave) │
52│ │ ├─ Inject previous wave findings into prev_context column │
53│ │ ├─ spawn_agents_on_csv(wave CSV) │
54│ │ ├─ Collect results, merge into master tasks.csv │
55│ │ └─ Check: any failed? → skip dependents or retry │
56│ └─ discoveries.ndjson shared across all waves (append-only) │
57│ │
58│ Phase 3: Results Aggregation │
59│ ├─ Export final results.csv │
60│ ├─ Generate context.md with all findings │
61│ ├─ Display summary: completed/failed/skipped per wave │
62│ └─ Offer: view results | retry failed | done │
63│ │
64└─────────────────────────────────────────────────────────────────────────┘
65```
66
67---
68
69## CSV Schema
70
71### tasks.csv (Master State)
72
73```csv
74id,title,description,test,acceptance_criteria,scope,hints,execution_directives,deps,context_from,wave,status,findings,files_modified,tests_passed,acceptance_met,error
75"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","","","","","",""
76"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","","","","","",""
77"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","","","","","",""
78"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","","","","","",""
79```
80
81**Columns**:
82
83| Column | Phase | Description |
84|--------|-------|-------------|
85| `id` | Input | Unique task identifier (string) |
86| `title` | Input | Short task title |
87| `description` | Input | Detailed task description — what to implement |
88| `test` | Input | Test cases: what tests to write and how to verify (unit/integration/edge) |
89| `acceptance_criteria` | Input | Acceptance criteria: measurable conditions that define "done" |
90| `scope` | Input | Target file/directory glob — constrains agent work area, prevents cross-task file conflicts |
91| `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 |
92| `execution_directives` | Input | Execution constraints: commands to run for verification, tool restrictions, environment requirements |
93| `deps` | Input | Semicolon-separated dependency task IDs (empty = no deps) |
94| `context_from` | Input | Semicolon-separated task IDs whose findings this task needs |
95| `wave` | Computed | Wave number (computed by topological sort, 1-based) |
96| `status` | Output | `pending` → `completed` / `failed` / `skipped` |
97| `findings` | Output | Key discoveries or implementation notes (max 500 chars) |
98| `files_modified` | Output | Semicolon-separated file paths |
99| `tests_passed` | Output | Whether all defined test cases passed (true/false) |
100| `acceptance_met` | Output | Summary of which acceptance criteria were met/unmet |
101| `error` | Output | Error message if failed (empty if success) |
102
103### Per-Wave CSV (Temporary)
104
105Each wave generates a temporary `wave-{N}.csv` with an extra `prev_context` column:
106
107```csv
108id,title,description,test,acceptance_criteria,scope,hints,execution_directives,deps,context_from,wave,prev_context
109"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"
110"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"
111```
112
113The `prev_context` column is built from `context_from` by looking up completed tasks' `findings` in the master CSV.
114
115---
116
117## Output Artifacts
118
119| File | Purpose | Lifecycle |
120|------|---------|-----------|
121| `tasks.csv` | Master state — all tasks with status/findings | Updated after each wave |
122| `wave-{N}.csv` | Per-wave input (temporary) | Created before wave, deleted after |
123| `results.csv` | Final export of all task results | Created in Phase 3 |
124| `discoveries.ndjson` | Shared exploration board across all agents | Append-only, carries across waves |
125| `context.md` | Human-readable execution report | Created in Phase 3 |
126
127---
128
129## Session Structure
130
131```
132.workflow/.csv-wave/{session-id}/
133├── tasks.csv # Master state (updated per wave)
134├── results.csv # Final results export
135├── discoveries.ndjson # Shared discovery board (all agents)
136├── context.md # Human-readable report
137└── wave-{N}.csv # Temporary per-wave input (cleaned up)
138```
139
140---
141
142## Implementation
143
144### Session Initialization
145
146```javascript
147const getUtc8ISOString = () => new Date(Date.now() + 8 * 60 * 60 * 1000).toISOString()
148
149// Parse flags
150const AUTO_YES = $ARGUMENTS.includes('--yes') || $ARGUMENTS.includes('-y')
151const continueMode = $ARGUMENTS.includes('--continue')
152const concurrencyMatch = $ARGUMENTS.match(/(?:--concurrency|-c)\s+(\d+)/)
153const maxConcurrency = concurrencyMatch ? parseInt(concurrencyMatch[1]) : 4
154
155// Clean requirement text (remove flags)
156const requirement = $ARGUMENTS
157 .replace(/--yes|-y|--continue|--concurrency\s+\d+|-c\s+\d+/g, '')
158 .trim()
159
160const slug = requirement.toLowerCase()
161 .replace(/[^a-z0-9\u4e00-\u9fa5]+/g, '-')
162 .substring(0, 40)
163const dateStr = getUtc8ISOString().substring(0, 10).replace(/-/g, '')
164const sessionId = `cwp-${slug}-${dateStr}`
165const sessionFolder = `.workflow/.csv-wave/${sessionId}`
166
167// Continue mode: find existing session
168if (continueMode) {
169 const existing = Bash(`ls -t .workflow/.csv-wave/ 2>/dev/null | head -1`).trim()
170 if (existing) {
171 sessionId = existing
172 sessionFolder = `.workflow/.csv-wave/${sessionId}`
173 // Read existing tasks.csv, find incomplete waves, resume from there
174 const existingCsv = Read(`${sessionFolder}/tasks.csv`)
175 // → jump to Phase 2 with remaining waves
176 }
177}
178
179Bash(`mkdir -p ${sessionFolder}`)
180```
181
182---
183
184### Phase 1: Requirement → CSV
185
186**Objective**: Decompose requirement into tasks, compute dependency waves, generate tasks.csv.
187
188**Steps**:
189
1901. **Decompose Requirement**
191
192 ```javascript
193 // Use ccw cli to decompose requirement into subtasks
194 Bash({
195 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.
196TASK:
197 • Parse requirement into independent subtasks
198 • Identify dependencies between tasks (which must complete before others)
199 • Identify context flow (which tasks need previous tasks' findings)
200 • For each task, define concrete test cases (unit/integration/edge)
201 • For each task, define measurable acceptance criteria (what defines 'done')
202 • Each task must be executable by a single agent with file read/write access
203MODE: analysis
204CONTEXT: @**/*
205EXPECTED: 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[]}.
206 - description: what to implement (specific enough for an agent to execute independently)
207 - test: what tests to write and how to verify (e.g. 'Unit test: X returns Y; Edge test: handles Z')
208 - acceptance_criteria: measurable conditions that define done (e.g. 'API returns 200; token expires after 1h')
209 - scope: target file/directory glob (e.g. 'src/auth/**') — tasks in same wave MUST have non-overlapping scopes
210 - hints: implementation tips + reference files, format '<tips> || <ref_file1>;<ref_file2>' (e.g. 'Use strategy pattern || src/base/Strategy.ts;docs/design.md')
211 - execution_directives: commands to run for verification or tool constraints (e.g. 'Run npm test --bail; Ensure tsc passes')
212 - deps: task IDs that must complete first
213 - context_from: task IDs whose findings are needed
214CONSTRAINTS: 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
215
216REQUIREMENT: ${requirement}" --tool gemini --mode analysis --rule planning-breakdown-task-steps`,
217 run_in_background: true
218 })
219 // Wait for CLI completion via hook callback
220 // Parse JSON from CLI output → decomposedTasks[]
221 ```
222
2232. **Compute Waves** (Topological Sort → Depth Grouping)
224
225 ```javascript
226 function computeWaves(tasks) {
227 // Build adjacency: task.deps → predecessors
228 const taskMap = new Map(tasks.map(t => [t.id, t]))
229 const inDegree = new Map(tasks.map(t => [t.id, 0]))
230 const adjList = new Map(tasks.map(t => [t.id, []]))
231
232 for (const task of tasks) {
233 for (const dep of task.deps) {
234 if (taskMap.has(dep)) {
235 adjList.get(dep).push(task.id)
236 inDegree.set(task.id, inDegree.get(task.id) + 1)
237 }
238 }
239 }
240
241 // BFS-based topological sort with depth tracking
242 const queue = [] // [taskId, depth]
243 const waveAssignment = new Map()
244
245 for (const [id, deg] of inDegree) {
246 if (deg === 0) {
247 queue.push([id, 1])
248 waveAssignment.set(id, 1)
249 }
250 }
251
252 let maxWave = 1
253 let idx = 0
254 while (idx < queue.length) {
255 const [current, depth] = queue[idx++]
256 for (const next of adjList.get(current)) {
257 const newDeg = inDegree.get(next) - 1
258 inDegree.set(next, newDeg)
259 const nextDepth = Math.max(waveAssignment.get(next) || 0, depth + 1)
260 waveAssignment.set(next, nextDepth)
261 if (newDeg === 0) {
262 queue.push([next, nextDepth])
263 maxWave = Math.max(maxWave, nextDepth)
264 }
265 }
266 }
267
268 // Detect cycles: any task without wave assignment
269 for (const task of tasks) {
270 if (!waveAssignment.has(task.id)) {
271 throw new Error(`Circular dependency detected involving task ${task.id}`)
272 }
273 }
274
275 return { waveAssignment, maxWave }
276 }
277
278 const { waveAssignment, maxWave } = computeWaves(decomposedTasks)
279 ```
280
2813. **Generate tasks.csv**
282
283 ```javascript
284 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'
285 const rows = decomposedTasks.map(task => {
286 const wave = waveAssignment.get(task.id)
287 return [
288 task.id,
289 csvEscape(task.title),
290 csvEscape(task.description),
291 csvEscape(task.test),
292 csvEscape(task.acceptance_criteria),
293 csvEscape(task.scope),
294 csvEscape(task.hints),
295 csvEscape(task.execution_directives),
296 task.deps.join(';'),
297 task.context_from.join(';'),
298 wave,
299 'pending', // status
300 '', // findings
301 '', // files_modified
302 '', // tests_passed
303 '', // acceptance_met
304 '' // error
305 ].map(cell => `"${String(cell).replace(/"/g, '""')}"`).join(',')
306 })
307
308 Write(`${sessionFolder}/tasks.csv`, [header, ...rows].join('\n'))
309 ```
310
3114. **User Validation** (skip if AUTO_YES)
312
313 ```javascript
314 if (!AUTO_YES) {
315 // Display task breakdown with wave assignment
316 console.log(`\n## Task Breakdown (${decomposedTasks.length} tasks, ${maxWave} waves)\n`)
317 for (let w = 1; w <= maxWave; w++) {
318 const waveTasks = decomposedTasks.filter(t => waveAssignment.get(t.id) === w)
319 console.log(`### Wave ${w} (${waveTasks.length} tasks, concurrent)`)
320 waveTasks.forEach(t => console.log(` - [${t.id}] ${t.title}`))
321 }
322
323 const answer = AskUserQuestion({
324 questions: [{
325 question: "Approve task breakdown?",
326 header: "Validation",
327 multiSelect: false,
328 options: [
329 { label: "Approve", description: "Proceed with wave execution" },
330 { label: "Modify", description: `Edit ${sessionFolder}/tasks.csv manually, then --continue` },
331 { label: "Cancel", description: "Abort" }
332 ]
333 }]
334 }) // BLOCKS
335
336 if (answer.Validation === "Modify") {
337 console.log(`Edit: ${sessionFolder}/tasks.csv\nResume: $csv-wave-pipeline --continue`)
338 return
339 } else if (answer.Validation === "Cancel") {
340 return
341 }
342 }
343 ```
344
345**Success Criteria**:
346- tasks.csv created with valid schema and wave assignments
347- No circular dependencies
348- User approved (or AUTO_YES)
349
350---
351
352### Phase 2: Wave Execution Engine
353
354**Objective**: Execute tasks wave-by-wave via `spawn_agents_on_csv`. Each wave sees previous waves' results.
355
356**Steps**:
357
3581. **Wave Loop**
359
360 ```javascript
361 const failedIds = new Set()
362 const skippedIds = new Set()
363
364 for (let wave = 1; wave <= maxWave; wave++) {
365 console.log(`\n## Wave ${wave}/${maxWave}\n`)
366
367 // 1. Read current master CSV
368 const masterCsv = parseCsv(Read(`${sessionFolder}/tasks.csv`))
369
370 // 2. Filter tasks for this wave
371 const waveTasks = masterCsv.filter(row => parseInt(row.wave) === wave)
372
373 // 3. Skip tasks whose deps failed
374 const executableTasks = []
375 for (const task of waveTasks) {
376 const deps = task.deps.split(';').filter(Boolean)
377 if (deps.some(d => failedIds.has(d) || skippedIds.has(d))) {
378 skippedIds.add(task.id)
379 // Update master CSV: mark as skipped
380 updateMasterCsvRow(sessionFolder, task.id, {
381 status: 'skipped',
382 error: 'Dependency failed or skipped'
383 })
384 console.log(` [${task.id}] ${task.title} → SKIPPED (dependency failed)`)
385 continue
386 }
387 executableTasks.push(task)
388 }
389
390 if (executableTasks.length === 0) {
391 console.log(` No executable tasks in wave ${wave}`)
392 continue
393 }
394
395 // 4. Build prev_context for each task
396 for (const task of executableTasks) {
397 const contextIds = task.context_from.split(';').filter(Boolean)
398 const prevFindings = contextIds
399 .map(id => {
400 const prevRow = masterCsv.find(r => r.id === id)
401 if (prevRow && prevRow.status === 'completed' && prevRow.findings) {
402 return `[Task ${id}: ${prevRow.title}] ${prevRow.findings}`
403 }
404 return null
405 })
406 .filter(Boolean)
407 .join('\n')
408 task.prev_context = prevFindings || 'No previous context available'
409 }
410
411 // 5. Write wave CSV
412 const waveHeader = 'id,title,description,test,acceptance_criteria,scope,hints,execution_directives,deps,context_from,wave,prev_context'
413 const waveRows = executableTasks.map(t =>
414 [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]
415 .map(cell => `"${String(cell).replace(/"/g, '""')}"`)
416 .join(',')
417 )
418 Write(`${sessionFolder}/wave-${wave}.csv`, [waveHeader, ...waveRows].join('\n'))
419
420 // 6. Execute wave
421 console.log(` Executing ${executableTasks.length} tasks (concurrency: ${maxConcurrency})...`)
422
423 const waveResult = spawn_agents_on_csv({
424 csv_path: `${sessionFolder}/wave-${wave}.csv`,
425 id_column: "id",
426 instruction: buildInstructionTemplate(sessionFolder, wave),
427 max_concurrency: maxConcurrency,
428 max_runtime_seconds: 600,
429 output_csv_path: `${sessionFolder}/wave-${wave}-results.csv`,
430 output_schema: {
431 type: "object",
432 properties: {
433 id: { type: "string" },
434 status: { type: "string", enum: ["completed", "failed"] },
435 findings: { type: "string" },
436 files_modified: { type: "array", items: { type: "string" } },
437 tests_passed: { type: "boolean" },
438 acceptance_met: { type: "string" },
439 error: { type: "string" }
440 },
441 required: ["id", "status", "findings", "tests_passed"]
442 }
443 })
444 // ↑ Blocks until all agents in this wave complete
445
446 // 7. Merge results into master CSV
447 const waveResults = parseCsv(Read(`${sessionFolder}/wave-${wave}-results.csv`))
448 for (const result of waveResults) {
449 updateMasterCsvRow(sessionFolder, result.id, {
450 status: result.status,
451 findings: result.findings || '',
452 files_modified: (result.files_modified || []).join(';'),
453 tests_passed: String(result.tests_passed ?? ''),
454 acceptance_met: result.acceptance_met || '',
455 error: result.error || ''
456 })
457
458 if (result.status === 'failed') {
459 failedIds.add(result.id)
460 console.log(` [${result.id}] ${result.title} → FAILED: ${result.error}`)
461 } else {
462 console.log(` [${result.id}] ${result.title} → COMPLETED`)
463 }
464 }
465
466 // 8. Cleanup temporary wave CSV
467 Bash(`rm -f "${sessionFolder}/wave-${wave}.csv"`)
468
469 console.log(` Wave ${wave} done: ${waveResults.filter(r => r.status === 'completed').length} completed, ${waveResults.filter(r => r.status === 'failed').length} failed`)
470 }
471 ```
472
4732. **Instruction Template Builder**
474
475 ```javascript
476 function buildInstructionTemplate(sessionFolder, wave) {
477 return `
478## TASK ASSIGNMENT
479
480### MANDATORY FIRST STEPS
4811. Read shared discoveries: ${sessionFolder}/discoveries.ndjson (if exists, skip if not)
4822. Read project context: .workflow/project-tech.json (if exists)
483
484---
485
486## Your Task
487
488**Task ID**: {id}
489**Title**: {title}
490**Description**: {description}
491**Scope**: {scope}
492
493### Implementation Hints & Reference Files
494{hints}
495
496> Format: \`<tips> || <ref_file1>;<ref_file2>\`. Read ALL reference files (after ||) before starting implementation. Apply tips (before ||) as implementation guidance.
497
498### Execution Directives
499{execution_directives}
500
501> Commands to run for verification, tool restrictions, or environment requirements. Follow these constraints during and after implementation.
502
503### Test Cases
504{test}
505
506### Acceptance Criteria
507{acceptance_criteria}
508
509### Previous Tasks' Findings (Context)
510{prev_context}
511
512---
513
514## Execution Protocol
515
5161. **Read references**: Parse {hints} — read all files listed after \`||\` to understand existing patterns
5172. **Read discoveries**: Load ${sessionFolder}/discoveries.ndjson for shared exploration findings
5183. **Use context**: Apply previous tasks' findings from prev_context above
5194. **Stay in scope**: ONLY create/modify files within {scope} — do NOT touch files outside this boundary
5205. **Apply hints**: Follow implementation tips from {hints} (before \`||\`)
5216. **Execute**: Implement the task as described
5227. **Write tests**: Implement the test cases defined above
5238. **Run directives**: Execute commands from {execution_directives} to verify your work
5249. **Verify acceptance**: Ensure all acceptance criteria are met before reporting completion
52510. **Share discoveries**: Append exploration findings to shared board:
526 \`\`\`bash
527 echo '{"ts":"<ISO8601>","worker":"{id}","type":"<type>","data":{...}}' >> ${sessionFolder}/discoveries.ndjson
528 \`\`\`
52911. **Report result**: Return JSON via report_agent_job_result
530
531### Discovery Types to Share
532- \`code_pattern\`: {name, file, description} — reusable patterns found
533- \`integration_point\`: {file, description, exports[]} — module connection points
534- \`convention\`: {naming, imports, formatting} — code style conventions
535- \`blocker\`: {issue, severity, impact} — blocking issues encountered
536
537---
538
539## Output (report_agent_job_result)
540
541Return JSON:
542{
543 "id": "{id}",
544 "status": "completed" | "failed",
545 "findings": "Key discoveries and implementation notes (max 500 chars)",
546 "files_modified": ["path1", "path2"],
547 "tests_passed": true | false,
548 "acceptance_met": "Summary of which acceptance criteria were met/unmet",
549 "error": ""
550}
551
552**IMPORTANT**: Set status to "completed" ONLY if:
553- All test cases pass
554- All acceptance criteria are met
555Otherwise set status to "failed" with details in error field.
556`
557 }
558 ```
559
5603. **Master CSV Update Helper**
561
562 ```javascript
563 function updateMasterCsvRow(sessionFolder, taskId, updates) {
564 const csvPath = `${sessionFolder}/tasks.csv`
565 const content = Read(csvPath)
566 const lines = content.split('\n')
567 const header = lines[0].split(',')
568
569 for (let i = 1; i < lines.length; i++) {
570 const cells = parseCsvLine(lines[i])
571 if (cells[0] === taskId || cells[0] === `"${taskId}"`) {
572 // Update specified columns
573 for (const [col, val] of Object.entries(updates)) {
574 const colIdx = header.indexOf(col)
575 if (colIdx >= 0) {
576 cells[colIdx] = `"${String(val).replace(/"/g, '""')}"`
577 }
578 }
579 lines[i] = cells.join(',')
580 break
581 }
582 }
583
584 Write(csvPath, lines.join('\n'))
585 }
586 ```
587
588**Success Criteria**:
589- All waves executed in order
590- Each wave's results merged into master CSV before next wave starts
591- Dependent tasks skipped when predecessor failed
592- discoveries.ndjson accumulated across all waves
593
594---
595
596### Phase 3: Results Aggregation
597
598**Objective**: Generate final results and human-readable report.
599
600**Steps**:
601
6021. **Export results.csv**
603
604 ```javascript
605 const masterCsv = Read(`${sessionFolder}/tasks.csv`)
606 // results.csv = master CSV (already has all results populated)
607 Write(`${sessionFolder}/results.csv`, masterCsv)
608 ```
609
6102. **Generate context.md**
611
612 ```javascript
613 const tasks = parseCsv(masterCsv)
614 const completed = tasks.filter(t => t.status === 'completed')
615 const failed = tasks.filter(t => t.status === 'failed')
616 const skipped = tasks.filter(t => t.status === 'skipped')
617
618 const contextContent = `# CSV Batch Execution Report
619
620**Session**: ${sessionId}
621**Requirement**: ${requirement}
622**Completed**: ${getUtc8ISOString()}
623**Waves**: ${maxWave} | **Concurrency**: ${maxConcurrency}
624
625---
626
627## Summary
628
629| Metric | Count |
630|--------|-------|
631| Total Tasks | ${tasks.length} |
632| Completed | ${completed.length} |
633| Failed | ${failed.length} |
634| Skipped | ${skipped.length} |
635| Waves | ${maxWave} |
636
637---
638
639## Wave Execution
640
641${Array.from({ length: maxWave }, (_, i) => i + 1).map(w => {
642 const waveTasks = tasks.filter(t => parseInt(t.wave) === w)
643 return `### Wave ${w}
644${waveTasks.map(t => `- **[${t.id}] ${t.title}**: ${t.status}${t.tests_passed ? ' ✓tests' : ''}${t.error ? ' — ' + t.error : ''}
645 ${t.findings ? 'Findings: ' + t.findings : ''}`).join('\n')}`
646}).join('\n\n')}
647
648---
649
650## Task Details
651
652${tasks.map(t => `### ${t.id}: ${t.title}
653
654| Field | Value |
655|-------|-------|
656| Status | ${t.status} |
657| Wave | ${t.wave} |
658| Scope | ${t.scope || 'none'} |
659| Dependencies | ${t.deps || 'none'} |
660| Context From | ${t.context_from || 'none'} |
661| Tests Passed | ${t.tests_passed || 'N/A'} |
662| Acceptance Met | ${t.acceptance_met || 'N/A'} |
663| Error | ${t.error || 'none'} |
664
665**Description**: ${t.description}
666
667**Test Cases**: ${t.test || 'N/A'}
668
669**Acceptance Criteria**: ${t.acceptance_criteria || 'N/A'}
670
671**Hints**: ${t.hints || 'N/A'}
672
673**Execution Directives**: ${t.execution_directives || 'N/A'}
674
675**Findings**: ${t.findings || 'N/A'}
676
677**Files Modified**: ${t.files_modified || 'none'}
678`).join('\n---\n')}
679
680---
681
682## All Modified Files
683
684${[...new Set(tasks.flatMap(t => (t.files_modified || '').split(';')).filter(Boolean))].map(f => '- ' + f).join('\n') || 'None'}
685`
686
687 Write(`${sessionFolder}/context.md`, contextContent)
688 ```
689
6903. **Display Summary**
691
692 ```javascript
693 console.log(`
694## Execution Complete
695
696- **Session**: ${sessionId}
697- **Waves**: ${maxWave}
698- **Completed**: ${completed.length}/${tasks.length}
699- **Failed**: ${failed.length}
700- **Skipped**: ${skipped.length}
701
702**Results**: ${sessionFolder}/results.csv
703**Report**: ${sessionFolder}/context.md
704**Discoveries**: ${sessionFolder}/discoveries.ndjson
705`)
706 ```
707
7084. **Offer Next Steps** (skip if AUTO_YES)
709
710 ```javascript
711 if (!AUTO_YES && failed.length > 0) {
712 const answer = AskUserQuestion({
713 questions: [{
714 question: `${failed.length} tasks failed. Next action?`,
715 header: "Next Step",
716 multiSelect: false,
717 options: [
718 { label: "Retry Failed", description: `Re-execute ${failed.length} failed tasks with updated context` },
719 { label: "View Report", description: "Display context.md" },
720 { label: "Done", description: "Complete session" }
721 ]
722 }]
723 }) // BLOCKS
724
725 if (answer['Next Step'] === "Retry Failed") {
726 // Reset failed tasks to pending, re-run Phase 2 for their waves
727 for (const task of failed) {
728 updateMasterCsvRow(sessionFolder, task.id, { status: 'pending', error: '' })
729 }
730 // Also reset skipped tasks whose deps are now retrying
731 for (const task of skipped) {
732 updateMasterCsvRow(sessionFolder, task.id, { status: 'pending', error: '' })
733 }
734 // Re-execute Phase 2 (loop will skip already-completed tasks)
735 // → goto Phase 2
736 } else if (answer['Next Step'] === "View Report") {
737 console.log(Read(`${sessionFolder}/context.md`))
738 }
739 }
740 ```
741
742**Success Criteria**:
743- results.csv exported
744- context.md generated
745- Summary displayed to user
746
747---
748
749## Shared Discovery Board Protocol
750
751All agents across all waves share `discoveries.ndjson`. This eliminates redundant codebase exploration.
752
753**Lifecycle**:
754- Created by the first agent to write a discovery
755- Carries over across waves — never cleared
756- Agents append via `echo '...' >> discoveries.ndjson`
757
758**Format**: NDJSON, each line is a self-contained JSON:
759
760```jsonl
761{"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"}}
762{"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"]}}
763```
764
765**Discovery Types**:
766
767| type | Dedup Key | Description |
768|------|-----------|-------------|
769| `code_pattern` | `data.name` | Reusable code pattern found |
770| `integration_point` | `data.file` | Module connection point |
771| `convention` | singleton | Code style conventions |
772| `blocker` | `data.issue` | Blocking issue encountered |
773| `tech_stack` | singleton | Project technology stack |
774| `test_command` | singleton | Test commands discovered |
775
776**Protocol Rules**:
7771. Read board before own exploration → skip covered areas
7782. Write discoveries immediately via `echo >>` → don't batch
7793. Deduplicate — check existing entries; skip if same type + dedup key exists
7804. Append-only — never modify or delete existing lines
781
782---
783
784## Wave Computation Details
785
786### Algorithm
787
788Kahn's BFS topological sort with depth tracking:
789
790```
791Input: tasks[] with deps[]
792Output: waveAssignment (taskId → wave number)
793
7941. Build in-degree map and adjacency list from deps
7952. Enqueue all tasks with in-degree 0 at wave 1
7963. BFS: for each dequeued task at wave W:
797 - For each dependent task D:
798 - Decrement D's in-degree
799 - D.wave = max(D.wave, W + 1)
800 - If D's in-degree reaches 0, enqueue D
8014. Any task without wave assignment → circular dependency error
802```
803
804### Wave Properties
805
806- **Wave 1**: No dependencies — all tasks in wave 1 are fully independent
807- **Wave N**: All dependencies are in waves 1..(N-1) — guaranteed completed before wave N starts
808- **Within a wave**: Tasks are independent of each other → safe for concurrent execution
809
810### Example
811
812```
813Task A (no deps) → Wave 1
814Task B (no deps) → Wave 1
815Task C (deps: A) → Wave 2
816Task D (deps: A, B) → Wave 2
817Task E (deps: C, D) → Wave 3
818
819Execution:
820 Wave 1: [A, B] ← concurrent
821 Wave 2: [C, D] ← concurrent, sees A+B findings
822 Wave 3: [E] ← sees A+B+C+D findings
823```
824
825---
826
827## Context Propagation Flow
828
829```
830Wave 1 agents:
831 ├─ Execute tasks (no prev_context)
832 ├─ Write findings to report_agent_job_result
833 └─ Append discoveries to discoveries.ndjson
834
835 ↓ merge results into master CSV
836
837Wave 2 agents:
838 ├─ Read discoveries.ndjson (exploration sharing)
839 ├─ Read prev_context column (wave 1 findings from context_from)
840 ├─ Execute tasks with full upstream context
841 ├─ Write findings to report_agent_job_result
842 └─ Append new discoveries to discoveries.ndjson
843
844 ↓ merge results into master CSV
845
846Wave 3 agents:
847 ├─ Read discoveries.ndjson (accumulated from waves 1+2)
848 ├─ Read prev_context column (wave 1+2 findings from context_from)
849 ├─ Execute tasks
850 └─ ...
851```
852
853**Two context channels**:
8541. **CSV findings** (structured): `context_from` column → `prev_context` injection — task-specific directed context
8552. **NDJSON discoveries** (broadcast): `discoveries.ndjson` — general exploration findings available to all
856
857---
858
859## Error Handling
860
861| Error | Resolution |
862|-------|------------|
863| Circular dependency | Detect in wave computation, abort with error message |
864| Agent timeout | Mark as failed in results, continue with wave |
865| Agent failed | Mark as failed, skip dependent tasks in later waves |
866| All agents in wave failed | Log error, offer retry or abort |
867| CSV parse error | Validate CSV format before execution, show line number |
868| discoveries.ndjson corrupt | Ignore malformed lines, continue with valid entries |
869| Continue mode: no session found | List available sessions, prompt user to select |
870
871---
872
873## Core Rules
874
8751. **Start Immediately**: First action is session initialization, then Phase 1
8762. **Wave Order is Sacred**: Never execute wave N before wave N-1 completes and results are merged
8773. **CSV is Source of Truth**: Master tasks.csv holds all state — always read before wave, always write after
8784. **Context Propagation**: prev_context built from master CSV, not from memory
8795. **Discovery Board is Append-Only**: Never clear, modify, or recreate discoveries.ndjson
8806. **Skip on Failure**: If a dependency failed, skip the dependent task (don't attempt)
8817. **Cleanup Temp Files**: Remove wave-{N}.csv after results are merged
8828. **DO NOT STOP**: Continuous execution until all waves complete or all remaining tasks are skipped
883
884---
885
886## Best Practices
887
8881. **Task Granularity**: 3-10 tasks optimal; too many = overhead, too few = no parallelism benefit
8892. **Minimize Cross-Wave Deps**: More tasks in wave 1 = more parallelism
8903. **Specific Descriptions**: Agent sees only its CSV row + prev_context — make description self-contained
8914. **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)
8925. **Concurrency Tuning**: `-c 1` for serial execution (maximum context sharing); `-c 8` for I/O-bound tasks
893
894---
895
896## Usage Recommendations
897
898| Scenario | Recommended Approach |
899|----------|---------------------|
900| Independent parallel tasks (no deps) | `$csv-wave-pipeline -c 8` — single wave, max parallelism |
901| Linear pipeline (A→B→C) | `$csv-wave-pipeline -c 1` — 3 waves, serial, full context |
902| Diamond dependency (A→B,C→D) | `$csv-wave-pipeline` — 3 waves, B+C concurrent in wave 2 |
903| Complex requirement, unclear tasks | Use `$roadmap-with-file` first for planning, then feed issues here |
904| Single complex task | Use `$workflow-lite-plan` instead |