Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs

.changeset/terminalize-skipped-tasks.md

@@ -0,0 +1,5 @@
+---
+"@pgflow/core": patch
+---
+
+Terminalize queued and started task rows when their parent step is skipped: sibling tasks of a step skipped via `whenExhausted: 'skip'`/`'skip-cascade'` (and cascade-skipped steps) now end as `skipped` instead of staying `queued`/`started` forever, and a migration repairs existing rows.

NOMENCLATURE_GUIDE.md

@@ -142,13 +142,15 @@ Slugs are unique text identifiers with specific rules:
 - `started` - Step is executing
 - `completed` - Step completed successfully
 - `failed` - Step failed permanently
+- `skipped` - Step was skipped due to failed dependency, unmet condition, or exhausted retries
 
 ### Task Statuses
 
 - `queued` - Task queued in PGMQ
 - `started` - Task is executing
 - `completed` - Task completed successfully
 - `failed` - Task failed (may be retried or permanent)
+- `skipped` - Task was cancelled because its parent step was skipped
 
 ## Configuration Terms
 

pkgs/core/schemas/0060_tables_runtime.sql

@@ -104,7 +104,7 @@ create table pgflow.step_tasks (
   foreign key (run_id, step_slug)
   references pgflow.step_states(run_id, step_slug),
   constraint valid_status check (
-    status in ('queued', 'started', 'completed', 'failed')
+    status in ('queued', 'started', 'completed', 'failed', 'skipped')
   ),
   constraint output_valid_only_for_completed check (
     output is null or status in ('completed', 'failed')

pkgs/core/schemas/0100_function__cascade_force_skip_steps.sql

@@ -90,15 +90,24 @@ BEGIN
         false
       ) as _broadcast_result
   ),
+  -- ---------- Terminalize active tasks of newly skipped steps ----------
+  skipped_tasks AS (
+    UPDATE pgflow.step_tasks AS task
+    SET status = 'skipped'
+    WHERE task.run_id = _cascade_force_skip_steps.run_id
+      AND task.step_slug IN (
+        SELECT skipped_step.step_slug
+        FROM skipped AS skipped_step
+      )
+      AND task.status IN ('queued', 'started')
+    RETURNING task.message_id
+  ),
   -- ---------- Archive queued/started task messages for skipped steps ----------
   archived_messages AS (
-    SELECT pgmq.archive(v_flow_slug, ARRAY_AGG(st.message_id)) as result
-    FROM pgflow.step_tasks st
-    WHERE st.run_id = _cascade_force_skip_steps.run_id
-      AND st.step_slug IN (SELECT sk.step_slug FROM skipped sk)
-      AND st.status IN ('queued', 'started')
-      AND st.message_id IS NOT NULL
-    HAVING COUNT(st.message_id) > 0
+    SELECT pgmq.archive(v_flow_slug, ARRAY_AGG(task.message_id)) as result
+    FROM skipped_tasks AS task
+    WHERE task.message_id IS NOT NULL
+    HAVING COUNT(task.message_id) > 0
   ),
   -- ---------- Update run counters ----------
   run_updates AS (

pkgs/core/schemas/0100_function_fail_task.sql

@@ -225,6 +225,13 @@ END IF;
   GROUP BY r.flow_slug
   HAVING COUNT(st.message_id) > 0;
 
+  -- Terminalize all still-active sibling task rows for the skipped step
+  UPDATE pgflow.step_tasks AS task
+  SET status = 'skipped'
+  WHERE task.run_id = fail_task.run_id
+    AND task.step_slug = fail_task.step_slug
+    AND task.status IN ('queued', 'started');
+
   -- Send broadcast event for step skipped
   PERFORM realtime.send(
     jsonb_build_object(