chore: update PLAN.md with current feature and task statuses

Reflects ongoing work on map step output aggregation, related migration, testing,
and documentation efforts, along with current implementation status and pending
tasks for map step support in the project.

Author: jumski

.claude/commands/help-review.md

@@ -0,0 +1,9 @@
+Your job is to help me understand changes made to $ARGUMENTS by line/section changed.
+**Current branch:** !`git branch --show-current`
+If there is `PLAN.md` or `PLAN_<somethign_related_to_current_branch>.md`, read it before starting.
+
+Here is the diff of the changes:
+
+<changes>
+!`git show -p --no-ext-diff -- $ARGUMENTS`
+</changes>

.claude/settings.json

@@ -5,6 +5,7 @@
       "Bash(./scripts/atlas-migrate-hash:*)",
       "Bash(./scripts/run-test-with-colors:*)",
       "Bash(PGPASSWORD=postgres psql -h 127.0.0.1 -p 50422 -U postgres -d postgres -c:*)",
+      "Bash(PGPASSWORD=postgres psql -h 127.0.0.1 -p 50422 -U postgres -d postgres -f:*)",
       "Bash(bin/run-test-with-colors:*)",
       "Bash(cat:*)",
       "Bash(cd:*)",
@@ -17,6 +18,8 @@
       "Bash(gh run list:*)",
       "Bash(gh run view:*)",
       "Bash(git rm:*)",
+      "Bash(git show -p --no-ext-diff --:*)",
+      "Bash(git whatchanged:*)",
       "Bash(grep:*)",
       "Bash(ls:*)",
       "Bash(mkdir:*)",

.claude/sql_style.md

@@ -18,4 +18,5 @@ Always qualify columns and arguments:
 - `start_flow.run_id` not just `run_id` in functions
 
 ## Keyword Arguments
-Use `param => "value"` NOT `param := "value"`
+Use `param => "value"` NOT `param := "value"`
+- Note on aliasing tables: when writing SQL functions and working with dependencies/dependents and steps and states, I want you to build your aliases such that you use parent/child prefixes and _step (for pgflow.steps) or _state (for pgflow.step_states) suffixes accordingly. dep should mean a row in pgflow.deps, not a parent dependency. do not use dep to indicate a row from steps or step_states.

PLAN.md

@@ -2,13 +2,21 @@
 
 **NOTE: This PLAN.md file should be removed in the final PR once all map infrastructure is complete.**
 
-### Current State
+### Features
 
-- ✅ **WORKING**: Empty array maps (taskless) cascade and complete correctly
-- ✅ **WORKING**: Task spawning creates N tasks with correct indices
-- ✅ **WORKING**: Dependency count propagation for map steps
-- ✅ **WORKING**: Array element extraction - tasks get full array instead of individual items
-- ❌ **MISSING**: Output aggregation - no way to combine map task outputs for dependents
+- ✅ **DONE**: Empty array maps (taskless) cascade and complete correctly
+- ✅ **DONE**: Task spawning creates N tasks with correct indices
+- ✅ **DONE**: Dependency count propagation for map steps
+- ✅ **DONE**: Array element extraction - tasks receive individual array elements
+- ✅ **DONE**: Output aggregation - inline implementation aggregates map task outputs for dependents
+- ⏳ **NEXT**: DSL support for `.map()` for defining map steps
+
+### Chores
+
+- ⏳ **WAITING**: Integration tests for map steps
+- ⏳ **WAITING**: Consolidated migration for map steps
+- ⏳ **WAITING**: Documentation for map steps
+- ⏳ **WAITING**: Graphite stack merge for map steps
 
 ## Implementation Status
 
@@ -67,16 +75,15 @@
   - Handles both root maps (from run input) and dependent maps (from step outputs)
   - Tests with actual array data processing
 
-#### ❌ Remaining Work
+- [x] **PR #217: Output Aggregation** - `09-17-add-map-step-output-aggregation` (THIS PR)
 
-- [ ] **Output Aggregation** (CRITICAL - BLOCKS MAP OUTPUT CONSUMPTION)
+  - Inline aggregation implementation in complete_task, start_tasks, maybe_complete_run
+  - Full test coverage (17 tests) for all aggregation scenarios
+  - Handles NULL preservation, empty arrays, order preservation
+  - Validates non-array outputs to map steps fail correctly
+  - Fixed broadcast aggregation to send full array not individual task output
 
-  - Aggregate map task outputs when step completes
-  - Store aggregated output for dependent steps to consume
-  - Maintain task_index ordering in aggregated arrays
-  - Tests for aggregation with actual map task outputs
-  - **IMPORTANT**: Must add test for map->map NULL propagation when this is implemented
-  - **IMPORTANT**: Must handle non-array outputs to map steps (should fail the run)
+#### ❌ Remaining Work
 
 - [ ] **DSL Support for .map() Step Type**
 
@@ -93,6 +100,30 @@
     - Type safety for input/output types
     - Compile-time enforcement of single dependency rule
 
+- [ ] **Fix Orphaned Messages on Run Failure**
+
+  - Archive all pending messages when run fails
+  - Handle map sibling tasks specially
+  - Fix type constraint violations to fail immediately without retries
+  - See detailed plan: [PLAN_orphaned_messages.md](./PLAN_orphaned_messages.md)
+  - Critical for production: prevents queue performance degradation
+  - Tests already written (stashed) that document the problem
+
+- [ ] **Performance Optimization: step_states.output Column**
+
+  - Migrate from inline aggregation to storing outputs in step_states
+  - See detailed plan: [PLAN_step_output.md](./PLAN_step_output.md)
+  - Benefits:
+    - Eliminate redundant aggregation queries
+    - 30-70% performance improvement for map chains
+    - Cleaner architecture with single source of truth
+  - Implementation:
+    - Add output column to step_states table
+    - Update complete_task to populate output on completion
+    - Simplify consumers (start_tasks, maybe_complete_run, broadcasts)
+    - Update all aggregation tests (~17 files)
+  - **Note**: This is an optimization that should be done after core functionality is stable
+
 - [ ] **Integration Tests**
 
   - End-to-end workflows with real array data

PLAN_orphaned_messages.md

@@ -0,0 +1,184 @@
+# Plan: Fix Orphaned Messages on Run Failure
+
+## Problem Statement
+
+When a run fails, messages for pending tasks remain in the queue indefinitely, causing:
+1. **Resource waste**: Workers continuously poll orphaned messages
+2. **Performance degradation**: Queue operations slow down over time
+3. **Map step issues**: Failing one map task leaves N-1 sibling messages orphaned
+4. **Type violations**: Deterministic errors retry unnecessarily
+
+## Current Behavior
+
+### When fail_task is called
+```sql
+-- Only archives the single failing task's message
+SELECT pgmq.archive('pgflow_tasks_queue', fail_task.msg_id);
+-- Leaves all other queued messages orphaned
+```
+
+### When type constraint violation occurs
+```sql
+-- Raises exception, causes retries
+RAISE EXCEPTION 'Map step % expects array input...';
+-- Transaction rolls back, but retries will hit same error
+```
+
+## Implementation Plan
+
+### 1. Update fail_task Function
+**File**: `pkgs/core/schemas/0100_function_fail_task.sql`
+
+Add after marking run as failed (around line 47):
+```sql
+-- Archive all pending messages for this run
+WITH tasks_to_archive AS (
+  SELECT t.msg_id
+  FROM pgflow.step_tasks t
+  WHERE t.run_id = fail_task.run_id
+    AND t.status = 'pending'
+    AND t.msg_id IS NOT NULL
+)
+SELECT pgmq.archive('pgflow_tasks_queue', msg_id)
+FROM tasks_to_archive;
+```
+
+### 2. Update complete_task for Type Violations
+**File**: `pkgs/core/schemas/0100_function_complete_task.sql`
+
+Replace the current RAISE EXCEPTION block (lines 115-120) with:
+```sql
+IF v_dependent_map_slug IS NOT NULL THEN
+  -- Mark run as failed immediately (no retries for type violations)
+  UPDATE pgflow.runs
+  SET status = 'failed',
+      failed_at = now(),
+      error = format('Type contract violation: Map step %s expects array input but dependency %s produced %s (output: %s)',
+                     v_dependent_map_slug,
+                     complete_task.step_slug,
+                     CASE WHEN complete_task.output IS NULL THEN 'null'
+                          ELSE jsonb_typeof(complete_task.output) END,
+                     complete_task.output)
+  WHERE run_id = complete_task.run_id;
+
+  -- Archive ALL pending messages for this run
+  PERFORM pgmq.archive('pgflow_tasks_queue', t.msg_id)
+  FROM pgflow.step_tasks t
+  WHERE t.run_id = complete_task.run_id
+    AND t.status = 'pending'
+    AND t.msg_id IS NOT NULL;
+
+  -- Mark the current task as failed (not completed)
+  UPDATE pgflow.step_tasks
+  SET status = 'failed',
+      failed_at = now(),
+      error_message = format('Type contract violation: produced %s instead of array',
+                            CASE WHEN complete_task.output IS NULL THEN 'null'
+                                 ELSE jsonb_typeof(complete_task.output) END)
+  WHERE run_id = complete_task.run_id
+    AND step_slug = complete_task.step_slug
+    AND task_index = complete_task.task_index;
+
+  -- Return empty result set (task not completed)
+  RETURN QUERY SELECT * FROM pgflow.step_tasks WHERE false;
+  RETURN;
+END IF;
+```
+
+### 3. Add Supporting Index
+**File**: New migration or add to existing
+
+```sql
+-- Speed up the archiving query
+CREATE INDEX IF NOT EXISTS idx_step_tasks_pending_with_msg
+ON pgflow.step_tasks(run_id, status)
+WHERE status = 'pending' AND msg_id IS NOT NULL;
+```
+
+## Testing
+
+### Tests Already Written (Stashed)
+
+1. **`supabase/tests/fail_task/archive_sibling_map_tasks.test.sql`**
+   - Verifies all map task messages are archived when one fails
+   - Tests: 8 assertions about message archiving and status
+
+2. **`supabase/tests/initial_tasks_null/archive_messages_on_type_constraint_failure.test.sql`**
+   - Verifies type violations archive all pending messages
+   - Tests: 8 assertions about queue cleanup and run status
+
+### How to Run Tests
+```bash
+# After unstashing and implementing the fixes:
+pnpm nx test:pgtap core -- supabase/tests/fail_task/archive_sibling_map_tasks.test.sql
+pnpm nx test:pgtap core -- supabase/tests/initial_tasks_null/archive_messages_on_type_constraint_failure.test.sql
+```
+
+## Migration Considerations
+
+### Backward Compatibility
+- New behavior only affects failed runs (safe)
+- Archiving preserves messages (can be recovered if needed)
+- No schema changes to existing tables
+
+### Performance Impact
+- One-time cost during failure (acceptable)
+- Prevents ongoing performance degradation (improvement)
+- Index ensures archiving query is efficient
+
+### Rollback Plan
+If issues arise:
+1. Remove the archiving logic
+2. Messages remain in queue (old behavior)
+3. No data loss since we archive, not delete
+
+## Edge Cases to Consider
+
+### 1. Concurrent Task Completion
+If multiple tasks complete/fail simultaneously:
+- PostgreSQL row locks ensure consistency
+- Each failure archives all pending messages
+- Idempotent: archiving already-archived messages is safe
+
+### 2. Very Large Map Steps
+For maps with 1000+ tasks:
+- Archiving might take several seconds
+- Consider batching if performance issues arise
+- Current approach should handle up to ~10k tasks reasonably
+
+### 3. Mixed Step Types
+When run has both map and single steps:
+- Archive logic handles all pending tasks regardless of type
+- Correctly archives both map siblings and unrelated pending tasks
+
+## Future Enhancements (Not for this PR)
+
+1. **Selective Archiving**: Only archive tasks that can't proceed
+2. **Batch Operations**: Archive in chunks for very large runs
+3. **Recovery Mechanism**: Function to unarchive and retry
+4. **Monitoring**: Track archived message counts for alerting
+
+## Success Criteria
+
+- [ ] All tests pass (both new test files)
+- [ ] No orphaned messages after run failure
+- [ ] Type violations don't retry
+- [ ] Performance acceptable for maps with 100+ tasks
+- [ ] No impact on successful run performance
+
+## Implementation Checklist
+
+- [ ] Update `fail_task` function
+- [ ] Update `complete_task` function
+- [ ] Add database index
+- [ ] Unstash and run tests
+- [ ] Test with large map steps (100+ tasks)
+- [ ] Update migration file
+- [ ] Document behavior change in function comments
+
+## Notes
+
+- This fix is **critical for production** - without it, queue performance will degrade over time
+- Type violations are **deterministic** - retrying them is always wasteful
+- Archiving (vs deleting) preserves debugging capability
+- The fix is relatively simple (~30 lines of SQL) but high impact