feat: add root map step support

Author: jumski

PLAN.md

@@ -1,48 +1,48 @@
 # Map Infrastructure (SQL Core)
 
+**NOTE: This PLAN.md file should be removed in the final PR once all map infrastructure is complete.**
+
 ## Implementation Status
 
 ### Sequential Child PR Plan
 
 - [x] **PR #207: Add .array() to DSL** - `feature-map-and-array`
   - TypeScript DSL enhancement for array creation
   - Foundation for map step functionality
-- [x] **PR #208: Foundation - Schema & add_step()** - `09-10-feat_add_map_step_type_in_sql` (CURRENT PR)
-
+- [x] **PR #208: Foundation - Schema & add_step()** - `09-10-feat_add_map_step_type_in_sql`
   - Schema changes (initial_tasks, remaining_tasks, constraints)
   - add_step() function with map step validation
   - Basic tests for map step creation
-
-- [ ] **PR #209: Root Map Support** - `09-11-root-map-support`
+- [x] **PR #209: Root Map Support** - `09-11-root-map-support` (COMPLETED)
 
   - Enhanced start_flow() for root map validation and count setting
   - Tests for root map scenarios
 
-- [ ] **PR #210: Task Spawning** - `09-12-task-spawning`
+- [ ] **Task Spawning**
 
   - Enhanced start_ready_steps() for N task generation
   - Empty array auto-completion
   - Tests for batch task creation
 
-- [ ] **PR #211: Array Element Extraction** - `09-13-array-extraction`
+- [ ] **Array Element Extraction**
 
   - Enhanced start_tasks() for map input extraction
   - Support for root and dependent maps
   - Tests for element extraction
 
-- [ ] **PR #212: Dependent Map Support** - `09-14-dependent-map`
+- [ ] **Dependent Map Support**
 
   - Enhanced complete_task() for map dependency handling
   - Array validation and count propagation
   - Tests for dependency scenarios
 
-- [ ] **PR #213: Output Aggregation** - `09-15-output-aggregation`
+- [ ] **Output Aggregation**
 
   - Enhanced maybe_complete_run() for array aggregation
   - Ordered output collection
   - Tests for aggregation
 
-- [ ] **PR #214: Integration Tests** - `09-16-integration-tests`
+- [ ] **Integration Tests**
   - End-to-end test suite
   - Edge case coverage
   - Performance validation
@@ -97,54 +97,9 @@ The implementation is split across multiple PRs as shown in the Sequential Child
 
 ## Database Schema Changes
 
-### Migration Strategy
-
-Using Atlas migrations with the established pkgs/core/scripts workflow:
-
-#### Initial Migration Generation
-
-```bash
-# Navigate to pkgs/core directory
-cd pkgs/core
-
-# First update schema files in pkgs/core/schemas/, then generate migration
-./scripts/atlas-migrate-diff add_map_step_type
-
-# Review generated migration file, then apply and verify
-pnpm nx verify-migrations core
-```
-
-#### Regenerating Migration After Schema Updates
-
-```bash
-# 1. Decide on migration name
-migration_name=add_map_step_type
+For detailed schema development workflow, migration generation, and regeneration instructions, see:
 
-# 1. Remove the previous version of the migration file
-git rm -f supabase/migrations/*_pgflow_${migration_name}.sql
-
-# 2. Reset the Atlas hash to allow regeneration
-./scripts/atlas-migrate-hash --yes
-
-# 3. reset database state to pre-migration
-pnpm nx supabase:reset core
-
-# 4. Update schema files in pkgs/core/schemas/ as needed (or if already updated, skip this step)
-
-# 5. generate the migration with the same name
-./scripts/atlas-migrate-diff ${migration_name}
-
-# 6. verify the migration
-pnpm nx verify-migrations core
-```
-
-**Key Points:**
-
-- START WITH REMOVING PREVIOUS MIGRATION FILE BEFORE REGENERATING TO AVOID CONFLICTS !!!!
-- Always use the same migration name (`add_map_step_type`) for the entire PR
-- do not include `pgflow_` prefix when genrating migration - it is included by atlas-migrate-diff automatically
-- Reset Atlas hash before regeneration to allow the same name to be used
-- This maintains a single, comprehensive migration per PR
+- `.claude/schema_development.md` - Concise workflow guide
 
 ### Schema Updates (DONE)
 
@@ -180,7 +135,7 @@ pnpm nx verify-migrations core
 - Added validation for map steps (max 1 dependency)
 - Function now stores step_type in database
 
-### 2. `start_flow()` - Root Map Count Setting (TODO: PR #209)
+### 2. `start_flow()` - Root Map Count Setting (CURRENT PR)
 
 **File**: `pkgs/core/schemas/0100_function_start_flow.sql`
 
@@ -191,7 +146,7 @@ pnpm nx verify-migrations core
 - Set `initial_tasks = jsonb_array_length(input)` for root maps
 - Fail with clear error if input is not array for root map
 
-### 3. `complete_task()` - Dependent Map Count Setting (TODO: PR #212)
+### 3. `complete_task()` - Dependent Map Count Setting (TODO)
 
 **File**: `pkgs/core/schemas/0100_function_complete_task.sql`
 
@@ -202,7 +157,7 @@ pnpm nx verify-migrations core
 - For map→map: count completed tasks, set `initial_tasks = task_count`
 - Fail with clear error if dependency output is not array when needed
 
-### 4. `start_ready_steps()` - Task Spawning (TODO: PR #210)
+### 4. `start_ready_steps()` - Task Spawning (TODO)
 
 **File**: `pkgs/core/schemas/0100_function_start_ready_steps.sql`
 
@@ -218,7 +173,7 @@ pnpm nx verify-migrations core
 - Transition directly `created` → `completed` for initial_tasks=0
 - Send single `step:completed` event with `output: []`
 
-### 5. `start_tasks()` - Array Element Extraction (TODO: PR #211)
+### 5. `start_tasks()` - Array Element Extraction (TODO)
 
 **File**: `pkgs/core/schemas/0120_function_start_tasks.sql`
 
@@ -229,7 +184,7 @@ pnpm nx verify-migrations core
 - Dependent maps: extract from aggregated dependency output
 - Single steps: unchanged behavior (keep existing logic)
 
-### 6. `maybe_complete_run()` - Output Aggregation (TODO: PR #213)
+### 6. `maybe_complete_run()` - Output Aggregation (TODO)
 
 **File**: `pkgs/core/schemas/0100_function_maybe_complete_run.sql`
 
@@ -250,13 +205,26 @@ pnpm nx verify-migrations core
 - map_dependency_limit.test.sql
 - map_step_with_no_deps.test.sql
 
-**PR #209-214:** Each PR will include its own comprehensive test suite covering:
+**PR #209 (CURRENT):**
+
+- root_map_array_validation.test.sql
+- root_map_initial_tasks.test.sql
+- mixed_step_types.test.sql
+- multiple_root_maps.test.sql
+- null_input_validation.test.sql
+- large_array_handling.test.sql
+- nested_array_handling.test.sql
+- mixed_type_arrays.test.sql
+- invalid_json_types.test.sql
+- flow_only_maps.test.sql
+
+**Subsequent PRs:** Each will include its own comprehensive test suite covering:
 
 - Function-specific tests
 - Edge cases and error handling
 - Integration with existing functionality
 
-**PR #214:** Final integration test suite covering end-to-end workflows
+**Final PR:** Integration test suite covering end-to-end workflows
 
 ## Edge Cases Handled
 

pkgs/core/schemas/0060_tables_runtime.sql

@@ -27,7 +27,7 @@ create table pgflow.step_states (
   step_slug text not null,
   status text not null default 'created',
   remaining_tasks int null,  -- NULL = not started, >0 = active countdown
-  initial_tasks int default 1 check (initial_tasks >= 0),  -- Planned task count: 1 for singles, N for maps
+  initial_tasks int not null default 1 check (initial_tasks >= 0),  -- Planned task count: 1 for singles, N for maps
   remaining_deps int not null default 0 check (remaining_deps >= 0),
   error_message text,
   created_at timestamptz not null default now(),

pkgs/core/schemas/0100_function_start_flow.sql

@@ -10,11 +10,36 @@ volatile
 as $$
 declare
   v_created_run pgflow.runs%ROWTYPE;
+  v_root_map_count int;
 begin
 
+-- Check for root map steps and validate input
+WITH root_maps AS (
+  SELECT step_slug
+  FROM pgflow.steps
+  WHERE steps.flow_slug = start_flow.flow_slug
+    AND steps.step_type = 'map'
+    AND steps.deps_count = 0
+)
+SELECT COUNT(*) INTO v_root_map_count FROM root_maps;
+
+-- If we have root map steps, validate that input is an array
+IF v_root_map_count > 0 THEN
+  -- First check for NULL (should be caught by NOT NULL constraint, but be defensive)
+  IF start_flow.input IS NULL THEN
+    RAISE EXCEPTION 'Flow % has root map steps but input is NULL', start_flow.flow_slug;
+  END IF;
+  
+  -- Then check if it's not an array
+  IF jsonb_typeof(start_flow.input) != 'array' THEN
+    RAISE EXCEPTION 'Flow % has root map steps but input is not an array (got %)', 
+      start_flow.flow_slug, jsonb_typeof(start_flow.input);
+  END IF;
+END IF;
+
 WITH
   flow_steps AS (
-    SELECT steps.flow_slug, steps.step_slug, steps.deps_count
+    SELECT steps.flow_slug, steps.step_slug, steps.step_type, steps.deps_count
     FROM pgflow.steps
     WHERE steps.flow_slug = start_flow.flow_slug
   ),
@@ -35,7 +60,19 @@ WITH
       (SELECT created_run.run_id FROM created_run),
       fs.step_slug,
       fs.deps_count,
-      1  -- For now, all steps get initial_tasks = 1 (single steps)
+      -- For root map steps (map with no deps), set initial_tasks to array length
+      -- For all other steps, set initial_tasks to 1
+      CASE 
+        WHEN fs.step_type = 'map' AND fs.deps_count = 0 THEN 
+          CASE 
+            WHEN jsonb_typeof(start_flow.input) = 'array' THEN 
+              jsonb_array_length(start_flow.input)
+            ELSE 
+              1
+          END
+        ELSE 
+          1
+      END
     FROM flow_steps fs
   )
 SELECT * FROM created_run INTO v_created_run;

pkgs/core/src/database-types.ts

@@ -127,7 +127,7 @@ export type Database = {
           error_message: string | null
           failed_at: string | null
           flow_slug: string
-          initial_tasks: number | null
+          initial_tasks: number
           remaining_deps: number
           remaining_tasks: number | null
           run_id: string
@@ -141,7 +141,7 @@ export type Database = {
           error_message?: string | null
           failed_at?: string | null
           flow_slug: string
-          initial_tasks?: number | null
+          initial_tasks?: number
           remaining_deps?: number
           remaining_tasks?: number | null
           run_id: string
@@ -155,7 +155,7 @@ export type Database = {
           error_message?: string | null
           failed_at?: string | null
           flow_slug?: string
-          initial_tasks?: number | null
+          initial_tasks?: number
           remaining_deps?: number
           remaining_tasks?: number | null
           run_id?: string

pkgs/core/supabase/migrations/20250912080800_pgflow_temp_pr2_root_maps.sql

@@ -0,0 +1,95 @@
+-- Modify "step_states" table
+ALTER TABLE "pgflow"."step_states" ALTER COLUMN "initial_tasks" SET NOT NULL;
+-- Modify "start_flow" function
+CREATE OR REPLACE FUNCTION "pgflow"."start_flow" ("flow_slug" text, "input" jsonb, "run_id" uuid DEFAULT NULL::uuid) RETURNS SETOF "pgflow"."runs" LANGUAGE plpgsql SET "search_path" = '' AS $$
+declare
+  v_created_run pgflow.runs%ROWTYPE;
+  v_root_map_count int;
+begin
+
+-- Check for root map steps and validate input
+WITH root_maps AS (
+  SELECT step_slug
+  FROM pgflow.steps
+  WHERE steps.flow_slug = start_flow.flow_slug
+    AND steps.step_type = 'map'
+    AND steps.deps_count = 0
+)
+SELECT COUNT(*) INTO v_root_map_count FROM root_maps;
+
+-- If we have root map steps, validate that input is an array
+IF v_root_map_count > 0 THEN
+  -- First check for NULL (should be caught by NOT NULL constraint, but be defensive)
+  IF start_flow.input IS NULL THEN
+    RAISE EXCEPTION 'Flow % has root map steps but input is NULL', start_flow.flow_slug;
+  END IF;
+  
+  -- Then check if it's not an array
+  IF jsonb_typeof(start_flow.input) != 'array' THEN
+    RAISE EXCEPTION 'Flow % has root map steps but input is not an array (got %)', 
+      start_flow.flow_slug, jsonb_typeof(start_flow.input);
+  END IF;
+END IF;
+
+WITH
+  flow_steps AS (
+    SELECT steps.flow_slug, steps.step_slug, steps.step_type, steps.deps_count
+    FROM pgflow.steps
+    WHERE steps.flow_slug = start_flow.flow_slug
+  ),
+  created_run AS (
+    INSERT INTO pgflow.runs (run_id, flow_slug, input, remaining_steps)
+    VALUES (
+      COALESCE(start_flow.run_id, gen_random_uuid()),
+      start_flow.flow_slug,
+      start_flow.input,
+      (SELECT count(*) FROM flow_steps)
+    )
+    RETURNING *
+  ),
+  created_step_states AS (
+    INSERT INTO pgflow.step_states (flow_slug, run_id, step_slug, remaining_deps, initial_tasks)
+    SELECT
+      fs.flow_slug,
+      (SELECT created_run.run_id FROM created_run),
+      fs.step_slug,
+      fs.deps_count,
+      -- For root map steps (map with no deps), set initial_tasks to array length
+      -- For all other steps, set initial_tasks to 1
+      CASE 
+        WHEN fs.step_type = 'map' AND fs.deps_count = 0 THEN 
+          CASE 
+            WHEN jsonb_typeof(start_flow.input) = 'array' THEN 
+              jsonb_array_length(start_flow.input)
+            ELSE 
+              1
+          END
+        ELSE 
+          1
+      END
+    FROM flow_steps fs
+  )
+SELECT * FROM created_run INTO v_created_run;
+
+-- Send broadcast event for run started
+PERFORM realtime.send(
+  jsonb_build_object(
+    'event_type', 'run:started',
+    'run_id', v_created_run.run_id,
+    'flow_slug', v_created_run.flow_slug,
+    'input', v_created_run.input,
+    'status', 'started',
+    'remaining_steps', v_created_run.remaining_steps,
+    'started_at', v_created_run.started_at
+  ),
+  'run:started',
+  concat('pgflow:run:', v_created_run.run_id),
+  false
+);
+
+PERFORM pgflow.start_ready_steps(v_created_run.run_id);
+
+RETURN QUERY SELECT * FROM pgflow.runs where pgflow.runs.run_id = v_created_run.run_id;
+
+end;
+$$;

pkgs/core/supabase/migrations/atlas.sum

@@ -1,4 +1,4 @@
-h1:p+C5rFGJ1qtyZhk/6hw+wKNa8qxySt48uRW7nURXhb4=
+h1:4npyXHLdBlxXDvZBwovtYW/ZqGaaZprPdyxEsmO4Iz0=
 20250429164909_pgflow_initial.sql h1:5K7OqB/vj73TWJTQquUzn+i6H2wWduaW+Ir1an3QYmQ=
 20250517072017_pgflow_fix_poll_for_tasks_to_use_separate_statement_for_polling.sql h1:gnT6hYn43p5oIfr0HqoGlqX/4Si+uxMsCBtBa0/Z2Cg=
 20250609105135_pgflow_add_start_tasks_and_started_status.sql h1:9Yv/elMz9Nht9rCJOybx62eNrUyghsEMbMKeOJPUMVc=
@@ -9,3 +9,4 @@ h1:p+C5rFGJ1qtyZhk/6hw+wKNa8qxySt48uRW7nURXhb4=
 20250707210212_pgflow_add_opt_start_delay.sql h1:11J7SDgS6EVFUwxSi0bRZnNQgVGTV0EJGj9yuC0vczY=
 20250719205006_pgflow_worker_deprecation.sql h1:L3LDsVrUeABlRBXhHsu60bilfgDKEJHci5xWknH9XIg=
 20250912075001_pgflow_temp_pr1_schema.sql h1:zVvGuRX/m8uPFCuJ7iAqOQ71onkCtze6P9d9ZsOgs98=
+20250912080800_pgflow_temp_pr2_root_maps.sql h1:v2KdChKBPBOIq3nCVVtKWy1OVcIROV+tPtaTUPQujSo=