chore: make initial_tasks NULL by default to indicate 'unknown yet'

Author: jumski

PLAN.md

@@ -89,6 +89,16 @@
 
 #### ❌ Remaining Work
 
+- [ ] **Semantic Improvement: NULL for Unknown initial_tasks**
+
+  - Change initial_tasks from "1 as placeholder" to NULL for dependent map steps
+  - Benefits: Semantic correctness (NULL = unknown, not "1 task")
+  - Scope: Schema change to allow NULL, update 5+ SQL functions
+  - See detailed plan in `pkgs/core/PLAN_use_null_for_map_initial_tasks.md`
+  - **Note**: This is a semantic improvement only - current approach works functionally
+    But it is important to have only valid states and being explicit, that's why
+    we are doing this change first.
+
 - [ ] **Array Element Distribution** (CRITICAL - BLOCKS REAL MAP USAGE)
 
   - Enhanced start_tasks() to distribute array elements to map tasks
@@ -102,6 +112,8 @@
   - 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)
 
 - [ ] **DSL Support for .map() Step Type**
 
@@ -124,23 +136,41 @@
   - Basic happy path coverage
   - This should be minimal and added to the Edge Worker integration test suite for now
 
-- [ ] **Semantic Improvement: NULL for Unknown initial_tasks** (OPTIONAL - Can be deferred)
-
-  - Change initial_tasks from "1 as placeholder" to NULL for dependent map steps
-  - Benefits: Semantic correctness (NULL = unknown, not "1 task")
-  - Scope: Schema change to allow NULL, update 5+ SQL functions
-  - See detailed plan in `pkgs/core/PLAN_use_null_for_map_initial_tasks.md`
-  - **Note**: This is a semantic improvement only - current approach works functionally
-  - **Warning**: If deferred, new tests for Array Distribution and Output Aggregation will
-    assume initial_tasks = 1 for dependent maps, making this change harder later
-
 - [ ] **Migration Consolidation**
 
   - Remove all temporary/incremental migrations from feature branches
   - Generate a single consolidated migration for the entire map infrastructure
   - Ensure clean migration path from current production schema
   - If NULL improvement is done, include it in the consolidated migration
 
+- [ ] **Update README's** and **Docs**
+
+  - `pkgs/core/README.md`
+
+    - Add new section describing the step types
+    - Describe single step briefly, focus on describing map step type and how it differs
+    - Make sure to mention that maps are constrained to have exactly one dependency
+    - Show multiple cases of inputs -> task creation
+    - Explain edge cases (empty array propagation, invalid array input)
+    - Explain root map vs dependent map and how it gets handled and what restrictions those apply on the Flow input
+    - Explain cascade completion of taskless steps and its limitations
+
+  - `pkgs/dsl/README.md`
+
+    - Briefly describe the new `.array()` and `.map()` methods
+    - Mention `.array` is mostly sugar, and `.map` is the new step type
+    - Link to `pkgs/core/README.md` sections for more explanation about map steps
+    - Make sure to mention that maps are constrained to have exactly one dependency
+
+  - **Add basic docs page**
+
+    - put it into `pkgs/website/src/content/docs/concepts/array-and-map-steps.mdx`
+    - describe the DSL and how the map works and why we need it
+    - show example usage of root map
+    - show example usage of dependent map
+    - focus mostly on how to use it, instead of how it works under the hood
+    - link to the README's for more details
+
 - [ ] **Graphite Stack Merge**
 
   - Configure Graphite merge queue for the complete PR stack

pkgs/core/PLAN_use_null_for_map_initial_tasks.md renamed to PLAN_use_null_for_map_initial_tasks.md

@@ -1,39 +1,31 @@
 # Plan: Use NULL for Unknown initial_tasks in Dependent Map Steps
 
 ## Motivation
+
 Currently, dependent map steps have `initial_tasks = 1` as a placeholder until their dependencies complete. This is semantically incorrect and confusing:
+
 - `1` implies "will spawn exactly 1 task" but that's false
 - `NULL` correctly means "unknown until dependencies complete"
 - Reduces cognitive load - see NULL, know it's unknown
 
 ## Critical Considerations
+
 Before implementing, these issues must be addressed:
 
-### 1. ~~The "Last Dependency" Problem~~ RESOLVED
-**Map steps can have at most 1 dependency** (enforced in add_step.sql:24):
-```sql
--- This constraint simplifies everything:
-IF step_type = 'map' AND array_length(deps_slugs, 1) > 1 THEN
-  RAISE EXCEPTION 'Map step can have at most one dependency'
-```
-This means we always know exactly when to resolve `initial_tasks` - when the single dependency completes!
+### 1. Non-Array Output Handling
 
-### 2. Non-Array Output Handling
 What if a dependency doesn't produce an array?
-```sql
--- Must handle both cases:
-CASE
-  WHEN jsonb_typeof(output) = 'array' THEN jsonb_array_length(output)
-  ELSE 1  -- Treat non-array as single item to map
-END
-```
 
-### 3. ~~Race Condition Prevention~~ RESOLVED
-Since map steps have at most 1 dependency, no race conditions possible!
-The single dependency completes once, updates initial_tasks atomically.
+We should chat about how to handle this case, as it is an invalid state
+and we need to handle it somehow.
+
+We should probably raise an explicit, well versed and detailed error,
+so user's flow will fail early.
+
+### 2. The Start Ready Steps Problem
 
-### 4. The Start Ready Steps Problem
 **CRITICAL**: We cannot use COALESCE - steps must NOT start with NULL initial_tasks
+
 ```sql
 -- WRONG: COALESCE(initial_tasks, 1)
 -- RIGHT: Add assertion before starting
@@ -45,6 +37,7 @@ END IF;
 ## Implementation Plan
 
 ### 1. Schema Change
+
 ```sql
 -- In 0060_tables_runtime.sql
 ALTER TABLE pgflow.step_states
@@ -59,6 +52,7 @@ ADD CONSTRAINT step_states_initial_tasks_check
 ```
 
 ### 2. Update start_flow Function
+
 ```sql
 -- In 0100_function_start_flow.sql
 -- Change initial_tasks assignment logic:
@@ -81,6 +75,7 @@ END
 ```
 
 ### 3. Update start_ready_steps Function
+
 ```sql
 -- In 0100_function_start_ready_steps.sql
 
@@ -101,6 +96,7 @@ CROSS JOIN LATERAL generate_series(0, started_step.initial_tasks - 1)
 ```
 
 ### 4. Update complete_task Function
+
 ```sql
 -- In 0100_function_complete_task.sql
 
@@ -121,6 +117,7 @@ END
 ```
 
 ### 5. Update cascade_complete_taskless_steps Function
+
 ```sql
 -- In 0100_function_cascade_complete_taskless_steps.sql
 
@@ -136,6 +133,7 @@ END
 ```
 
 ### 6. Add Safety Assertions
+
 ```sql
 -- Add check constraint or trigger to ensure:
 -- When status changes from 'created' to 'started',
@@ -150,46 +148,31 @@ ADD CONSTRAINT initial_tasks_known_when_started
 ```
 
 ### 7. Update Tests
+
 - Update test expectations to check for NULL instead of 1
 - Add specific tests for NULL -> actual value transitions
 - Test that steps can't start with NULL initial_tasks
 
-### 8. Migration Strategy
-```sql
--- Create migration to update existing data:
-UPDATE pgflow.step_states
-SET initial_tasks = NULL
-WHERE step_slug IN (
-  SELECT s.step_slug
-  FROM pgflow.steps s
-  WHERE s.step_type = 'map'
-    AND EXISTS (
-      SELECT 1 FROM pgflow.deps d
-      WHERE d.flow_slug = s.flow_slug
-        AND d.step_slug = s.step_slug
-    )
-)
-AND status = 'created';
-```
-
 ## Benefits
+
 1. **Semantic correctness**: NULL = unknown, not "1 task" placeholder
 2. **Clearer mental model**: No translation needed when reading state
 3. **Easier debugging**: Can immediately see which values are unresolved
-4. **Type safety**: TypeScript `number | null` enforces proper handling
-5. **Simpler than expected**: Map steps having max 1 dependency eliminates complexity
 
 ## Simplified Implementation Path
+
 Since map steps can only have 0 or 1 dependency:
+
 1. Root maps (0 deps): Get initial_tasks from flow input immediately
 2. Dependent maps (1 dep): Start with NULL, resolve when dependency completes
 3. No multi-dependency complexity or race conditions!
 
 ## Testing Checklist
+
 - [ ] Root map steps get correct initial_tasks from input array
 - [ ] Dependent map steps start with NULL initial_tasks
 - [ ] Single -> Map updates NULL to array length
 - [ ] Map -> Map updates NULL to aggregated array length (future)
 - [ ] Empty array propagation sets 0, not NULL
-- [ ] Steps cannot start with NULL initial_tasks
-- [ ] All arithmetic operations handle NULL safely
+- [ ] 'single' steps must start with `initial_tasks = 1`
+- [ ] If any arithmetic operations on initial_tasks, verify if they handle NULL safely

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 not null default 1 check (initial_tasks >= 0),  -- Planned task count: 1 for singles, N for maps
+  initial_tasks int null check (initial_tasks is null or initial_tasks >= 0),
   remaining_deps int not null default 0 check (remaining_deps >= 0),
   error_message text,
   created_at timestamptz not null default now(),
@@ -43,6 +43,9 @@ create table pgflow.step_states (
   constraint remaining_tasks_state_consistency check (
     remaining_tasks is null or status != 'created'
   ),
+  constraint initial_tasks_known_when_started check (
+    status != 'started' or initial_tasks is not null
+  ),
   constraint completed_at_or_failed_at check (not (completed_at is not null and failed_at is not null)),
   constraint started_at_is_after_created_at check (started_at is null or started_at >= created_at),
   constraint completed_at_is_after_started_at check (completed_at is null or completed_at >= started_at),

pkgs/core/schemas/0100_function_cascade_complete_taskless_steps.sql

@@ -40,8 +40,8 @@ BEGIN
           -- set its initial_tasks to 0 as well
           initial_tasks = CASE
             WHEN s.step_type = 'map' AND dep_count.has_zero_tasks
-            THEN 0
-            ELSE ss.initial_tasks
+            THEN 0  -- Empty array propagation
+            ELSE ss.initial_tasks  -- Keep existing value (including NULL)
           END
       FROM (
         -- Count how many completed steps are dependencies of each dependent

pkgs/core/schemas/0100_function_complete_task.sql

@@ -11,8 +11,32 @@ set search_path to ''
 as $$
 declare
   v_step_state pgflow.step_states%ROWTYPE;
+  v_dependent_map_slug text;
 begin
 
+-- Check if output is non-array for dependent map steps
+-- This validation must happen BEFORE acquiring locks and making updates
+-- to fail fast without holding resources or starting changes
+SELECT ds.step_slug INTO v_dependent_map_slug
+FROM pgflow.deps d
+JOIN pgflow.steps ds ON ds.flow_slug = d.flow_slug AND ds.step_slug = d.step_slug
+JOIN pgflow.step_states ss ON ss.flow_slug = ds.flow_slug AND ss.step_slug = ds.step_slug
+WHERE d.dep_slug = complete_task.step_slug
+  AND d.flow_slug = (SELECT r.flow_slug FROM pgflow.runs r WHERE r.run_id = complete_task.run_id)
+  AND ds.step_type = 'map'
+  AND ss.run_id = complete_task.run_id
+  AND ss.initial_tasks IS NULL
+  AND (complete_task.output IS NULL OR jsonb_typeof(complete_task.output) != 'array')
+LIMIT 1;
+
+IF v_dependent_map_slug IS NOT NULL THEN
+  RAISE EXCEPTION 'Map step % expects array input but dependency % produced % (output: %)',
+    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;
+END IF;
+
 WITH run_lock AS (
   SELECT * FROM pgflow.runs
   WHERE pgflow.runs.run_id = complete_task.run_id
@@ -72,11 +96,15 @@ dependent_steps_lock AS (
 dependent_steps_update AS (
   UPDATE pgflow.step_states ss
   SET remaining_deps = ss.remaining_deps - 1,
-      -- For map dependents of single steps producing arrays, set initial_tasks
+      -- NEW: Resolve NULL initial_tasks for dependent map steps
       initial_tasks = CASE
-        WHEN s.step_type = 'map' AND jsonb_typeof(complete_task.output) = 'array'
-        THEN jsonb_array_length(complete_task.output)
-        ELSE ss.initial_tasks
+        WHEN s.step_type = 'map' AND ss.initial_tasks IS NULL
+             AND complete_task.output IS NOT NULL
+             AND jsonb_typeof(complete_task.output) = 'array' THEN
+          -- Resolve NULL to actual value based on output
+          -- Non-array case is already handled by validation above
+          jsonb_array_length(complete_task.output)
+        ELSE ss.initial_tasks  -- Keep existing value
       END
   FROM dependent_steps ds, pgflow.steps s
   WHERE ss.run_id = complete_task.run_id

pkgs/core/schemas/0100_function_start_flow.sql

@@ -60,17 +60,21 @@ WITH
       (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 
+      -- Updated logic for initial_tasks:
+      CASE
+        WHEN fs.step_type = 'map' AND fs.deps_count = 0 THEN
+          -- Root map: get array length from input
+          CASE
+            WHEN jsonb_typeof(start_flow.input) = 'array' THEN
               jsonb_array_length(start_flow.input)
-            ELSE 
+            ELSE
               1
           END
-        ELSE 
+        WHEN fs.step_type = 'map' AND fs.deps_count > 0 THEN
+          -- Dependent map: unknown until dependencies complete
+          NULL
+        ELSE
+          -- Single steps: always 1 task
           1
       END
     FROM flow_steps fs

pkgs/core/schemas/0100_function_start_ready_steps.sql

@@ -58,10 +58,12 @@ ready_steps AS (
   WHERE step_state.run_id = start_ready_steps.run_id
     AND step_state.status = 'created'
     AND step_state.remaining_deps = 0
+    AND step_state.initial_tasks IS NOT NULL  -- NEW: Cannot start with unknown count
+    AND step_state.initial_tasks > 0  -- Don't start taskless steps
     -- Exclude empty map steps already handled
     AND NOT EXISTS (
-      SELECT 1 FROM empty_map_steps 
-      WHERE empty_map_steps.run_id = step_state.run_id 
+      SELECT 1 FROM empty_map_steps
+      WHERE empty_map_steps.run_id = step_state.run_id
         AND empty_map_steps.step_slug = step_state.step_slug
     )
   ORDER BY step_state.step_slug

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
+          initial_tasks: number | null
           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
+          initial_tasks?: number | null
           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
+          initial_tasks?: number | null
           remaining_deps?: number
           remaining_tasks?: number | null
           run_id?: string