csv-wave-pipeline

## Auto Mode When `--yes` or `-y`: Auto-confirm task decomposition, skip interactive validation, use defaults. # CSV Wave Pipeline ## Usage ```bash $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: 1. **CSV findings** (structured): `context_from` column → `prev_context` injection — task-specific directed context 2. **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) ```csv id,title,description,test,acceptance_criteria,scope,hints,execution_directives,deps,context_from,wave,status,findings,files_modified,tests_passed,acceptance_met,error "1","Setup auth module","Create auth directory structure and base files","Verify directory exists and base files export expected interfaces","auth/ dir created; index.ts and types.ts export AuthProvider interface","src/auth/**","Follow monorepo module pattern || package.json;src/shared/types.ts","","","","1","","","","","","" "2","Implement OAuth","Add OAuth provider integration with Google and GitHub","Unit test: mock OAuth callback returns valid token; Integration test: verify redirect URL generation","OAuth login redirects to provider; callback returns JWT; supports Google and GitHub","src/auth/oauth/**","Use passport.js strategy pattern || src/auth/index.ts;docs/oauth-flow.md","Run npm test -- --grep oauth before completion","1","1","2","","","","","","" "3","Add JWT tokens","Implement JWT generation and validation","Unit test: sign/verify round-trip; Edge test: expired token returns 401","generateToken() returns valid JWT; verifyToken() rejects expired/tampered tokens","src/auth/jwt/**","Use jsonwebtoken library; Set default expiry 1h || src/config/auth.ts","Ensure tsc --noEmit passes","1","1","2","","","","","","" "4","Setup 2FA","Add TOTP-based 2FA with QR code generation","Unit test: TOTP verify with correct code; Test: QR data URL is valid","QR code generates scannable image; TOTP verification succeeds within time window","src/auth/2fa/**","Use speakeasy + qrcode libraries || src/auth/oauth/strategy.ts;src/auth/jwt/token.ts","Run full test suite: npm test","2;3","1;2;3","3","","","","","","" ``` **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: ```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: ```jsonl {"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**: 1. Read board before own exploration → skip covered areas 2. Write discoveries immediately via `echo >>` → don't batch 3. Deduplicate — check existing entries; skip if same type + dedup key exists 4. Append-only — never modify or delete existing lines --- ## Implementation ### Session Initialization ```javascript 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: ```javascript // 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) =>