fix(core): terminalize unfinished tasks when runs fail

.changeset/cancel-failed-run-tasks.md

@@ -0,0 +1,5 @@
+---
+"@pgflow/core": patch
+---
+
+Mark unfinished tasks as cancelled when their run fails, prevent late callbacks and stalled recovery from reviving them, and repair active tasks on historical failed runs.

NOMENCLATURE_GUIDE.md

@@ -151,6 +151,7 @@ Slugs are unique text identifiers with specific rules:
 - `completed` - Task completed successfully
 - `failed` - Task failed (may be retried or permanent)
 - `skipped` - Task was cancelled because its parent step was skipped
+- `cancelled` - Task was cancelled because its run failed before the task finished; `runs.failed_at` is the cancellation time
 
 ## Configuration Terms
 

pkgs/core/README.md

@@ -348,9 +348,12 @@ The system handles failures by:
    - Marking the step as 'failed'
    - Marking the run as 'failed'
    - Archiving the message in PGMQ
-   - **Archiving all queued messages for the failed run** (preventing orphaned messages)
+   - **Marking every remaining queued or started task in the run as 'cancelled'** and archiving their messages (the failed-run invariant: a failed run has no task rows left in `queued` or `started`)
 4. Additional failure handling:
-   - **No retries on already-failed runs** - tasks are immediately marked as failed
+   - **No retries on already-failed runs** - late `fail_task()` callbacks keep cancelled tasks cancelled
+   - **Cancellation wins over late callbacks** - late `complete_task()` or `fail_task()` cannot revive or rewrite a cancelled task; a committed completed task stays completed
+   - **`cancelled` differs from `failed` and `skipped`** - `failed` counts real handler failures, `skipped` is step-skip policy, `cancelled` is orchestration state invalidated by the run failure; `runs.failed_at` is the cancellation time
+   - **Database cancellation does not terminate handlers** - JavaScript already executing in a worker is not forcibly stopped and external side effects are not undone
    - **Graceful type constraint violations** - handled without exceptions when single steps feed map steps
    - **Stores invalid output on type violations** - captures the output that caused the violation for debugging
    - **Performance-optimized message archiving** using indexed queries

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', 'skipped')
+    status in ('queued', 'started', 'completed', 'failed', 'skipped', 'cancelled')
   ),
   constraint output_valid_only_for_completed check (
     output is null or status in ('completed', 'failed')

pkgs/core/schemas/0062_function_requeue_stalled_tasks.sql

@@ -13,6 +13,8 @@ begin
   -- Find and requeue stalled tasks (where started_at > timeout + 30s buffer)
   -- Tasks with requeued_count >= max_requeues will have their message archived
   -- but status left as 'started' for easy identification via requeued_count column
+  -- Eligibility requires the parent run AND parent step to still be 'started':
+  -- stale rows on failed runs or terminal steps must not be revived (#645).
   with stalled_tasks as (
     select
       st.run_id,
@@ -24,8 +26,11 @@ begin
       f.opt_timeout
     from pgflow.step_tasks st
     join pgflow.runs r on r.run_id = st.run_id
+    join pgflow.step_states ss on ss.run_id = st.run_id and ss.step_slug = st.step_slug
     join pgflow.flows f on f.flow_slug = r.flow_slug
     where st.status = 'started'
+      and r.status = 'started'
+      and ss.status = 'started'
       and st.permanently_stalled_at is null
       and st.started_at < now() - (f.opt_timeout * interval '1 second') - interval '30 seconds'
     for update of st skip locked
@@ -84,9 +89,9 @@ begin
   _vr as (select count(*) from visibility_reset),
   -- Force execution of mark_permanently_stalled CTE
   _mps as (select count(*) from mark_permanently_stalled),
-  -- Force execution of archived CTE  
+  -- Force execution of archived CTE
   _ar as (select count(*) from archived)
-  select count(*) into result_count 
+  select count(*) into result_count
   from requeued, _vr, _mps, _ar;
 
   return result_count;

pkgs/core/schemas/0100_function_cascade_resolve_conditions.sql

@@ -17,6 +17,9 @@ DECLARE
   v_iteration_count int := 0;
   v_max_iterations int := 50;
   v_processed_count int;
+  v_run_transitioned boolean;
+  v_flow_slug text;
+  v_cancelled_message_ids bigint[];
 BEGIN
   -- ==========================================
   -- GUARD: Early return if run is already terminal
@@ -108,54 +111,72 @@ BEGIN
     -- Note: Cannot use "v_first_fail IS NOT NULL" because records with NULL fields
     -- evaluate to NULL in IS NOT NULL checks. Use FOUND instead.
     IF FOUND THEN
-      UPDATE pgflow.step_states
-      SET status = 'failed',
-          failed_at = now(),
-          error_message = 'Condition not met'
-      WHERE pgflow.step_states.run_id = cascade_resolve_conditions.run_id
-        AND pgflow.step_states.step_slug = v_first_fail.step_slug;
-
+      -- Fail the run only if it is still started. The conditional UPDATE takes
+      -- the run row lock and rechecks status atomically, so replayed or
+      -- concurrent calls cannot duplicate the terminal transition or its events.
       UPDATE pgflow.runs
       SET status = 'failed',
           failed_at = now()
-      WHERE pgflow.runs.run_id = cascade_resolve_conditions.run_id;
+      WHERE pgflow.runs.run_id = cascade_resolve_conditions.run_id
+        AND pgflow.runs.status = 'started'
+      RETURNING true INTO v_run_transitioned;
+
+      IF v_run_transitioned THEN
+        UPDATE pgflow.step_states
+        SET status = 'failed',
+            failed_at = now(),
+            error_message = 'Condition not met'
+        WHERE pgflow.step_states.run_id = cascade_resolve_conditions.run_id
+          AND pgflow.step_states.step_slug = v_first_fail.step_slug;
+
+        PERFORM realtime.send(
+          jsonb_build_object(
+            'event_type', 'step:failed',
+            'run_id', cascade_resolve_conditions.run_id,
+            'step_slug', v_first_fail.step_slug,
+            'status', 'failed',
+            'error_message', 'Condition not met',
+            'failed_at', now()
+          ),
+          concat('step:', v_first_fail.step_slug, ':failed'),
+          concat('pgflow:run:', cascade_resolve_conditions.run_id),
+          false
+        );
 
-      PERFORM realtime.send(
-        jsonb_build_object(
-          'event_type', 'step:failed',
-          'run_id', cascade_resolve_conditions.run_id,
-          'step_slug', v_first_fail.step_slug,
-          'status', 'failed',
-          'error_message', 'Condition not met',
-          'failed_at', now()
-        ),
-        concat('step:', v_first_fail.step_slug, ':failed'),
-        concat('pgflow:run:', cascade_resolve_conditions.run_id),
-        false
-      );
+        PERFORM realtime.send(
+          jsonb_build_object(
+            'event_type', 'run:failed',
+            'run_id', cascade_resolve_conditions.run_id,
+            'flow_slug', v_first_fail.flow_slug,
+            'status', 'failed',
+            'error_message', 'Condition not met',
+            'failed_at', now()
+          ),
+          'run:failed',
+          concat('pgflow:run:', cascade_resolve_conditions.run_id),
+          false
+        );
 
-      PERFORM realtime.send(
-        jsonb_build_object(
-          'event_type', 'run:failed',
-          'run_id', cascade_resolve_conditions.run_id,
-          'flow_slug', v_first_fail.flow_slug,
-          'status', 'failed',
-          'error_message', 'Condition not met',
-          'failed_at', now()
-        ),
-        'run:failed',
-        concat('pgflow:run:', cascade_resolve_conditions.run_id),
-        false
-      );
+        -- Terminalize every unfinished task across all branches as cancelled,
+        -- capturing their message ids for archival below. Lock-order invariant:
+        -- always lock/update step_tasks before PGMQ queue rows.
+        WITH cancelled_tasks AS (
+          UPDATE pgflow.step_tasks AS task
+          SET status = 'cancelled'
+          WHERE task.run_id = cascade_resolve_conditions.run_id
+            AND task.status IN ('queued', 'started')
+          RETURNING task.message_id
+        )
+        SELECT ARRAY_AGG(ct.message_id) INTO v_cancelled_message_ids
+        FROM cancelled_tasks ct
+        WHERE ct.message_id IS NOT NULL;
 
-      PERFORM pgmq.archive(r.flow_slug, ARRAY_AGG(st.message_id))
-      FROM pgflow.step_tasks st
-      JOIN pgflow.runs r ON st.run_id = r.run_id
-      WHERE st.run_id = cascade_resolve_conditions.run_id
-        AND st.status IN ('queued', 'started')
-        AND st.message_id IS NOT NULL
-      GROUP BY r.flow_slug
-      HAVING COUNT(st.message_id) > 0;
+        -- Archive the cancelled task messages captured above (only after their
+        -- task rows are terminalized)
+        IF v_cancelled_message_ids IS NOT NULL THEN
+          PERFORM pgmq.archive(v_first_fail.flow_slug, v_cancelled_message_ids);
+        END IF;
+      END IF;
 
       RETURN false;
     END IF;

pkgs/core/schemas/0100_function_complete_task.sql

@@ -14,6 +14,7 @@ declare
   v_dependent_map_slug text;
   v_run_record pgflow.runs%ROWTYPE;
   v_step_record pgflow.step_states%ROWTYPE;
+  v_violation_archived_ids bigint[];
 begin
 
 -- ==========================================
@@ -40,6 +41,27 @@ WHERE pgflow.step_states.run_id = complete_task.run_id
   AND pgflow.step_states.step_slug = complete_task.step_slug
 FOR UPDATE;
 
+-- ==========================================
+-- GUARD: Run failed while this callback waited for the lock
+-- ==========================================
+-- The failed-run guard above ran before the failure committed. Recheck under
+-- lock so cancellation wins: archived message stays archived, task row keeps
+-- its terminal status, and no events or counters are emitted.
+IF v_run_record.status = 'failed' THEN
+  -- Archive the task message if present (no-op when already archived)
+  PERFORM pgflow._archive_task_message(
+    complete_task.run_id,
+    complete_task.step_slug,
+    complete_task.task_index
+  );
+  -- Return the current task row without any mutations
+  RETURN QUERY SELECT * FROM pgflow.step_tasks
+    WHERE pgflow.step_tasks.run_id = complete_task.run_id
+      AND pgflow.step_tasks.step_slug = complete_task.step_slug
+      AND pgflow.step_tasks.task_index = complete_task.task_index;
+  RETURN;
+END IF;
+
 -- ==========================================
 -- GUARD: Late callback - step not started
 -- ==========================================
@@ -84,6 +106,20 @@ LIMIT 1;
 
 -- Handle type violation if detected
 IF v_dependent_map_slug IS NOT NULL THEN
+  -- Mark current task as failed FIRST and store the output that caused the
+  -- violation, so the task row is terminal before any queue row is touched.
+  UPDATE pgflow.step_tasks
+  SET status = 'failed',
+      failed_at = now(),
+      output = complete_task.output,  -- Store the output that caused the violation
+      error_message = '[TYPE_VIOLATION] Produced ' ||
+                     CASE WHEN complete_task.output IS NULL THEN 'null'
+                          ELSE jsonb_typeof(complete_task.output) END ||
+                     ' instead of array'
+  WHERE pgflow.step_tasks.run_id = complete_task.run_id
+    AND pgflow.step_tasks.step_slug = complete_task.step_slug
+    AND pgflow.step_tasks.task_index = complete_task.task_index;
+
   -- Mark run as failed immediately
   UPDATE pgflow.runs
   SET status = 'failed',
@@ -105,30 +141,6 @@ IF v_dependent_map_slug IS NOT NULL THEN
     false
   );
 
-  -- Archive all active messages (both queued and started) to prevent orphaned messages
-  PERFORM pgmq.archive(
-    v_run_record.flow_slug,
-    array_agg(st.message_id)
-  )
-  FROM pgflow.step_tasks st
-  WHERE st.run_id = complete_task.run_id
-    AND st.status IN ('queued', 'started')
-    AND st.message_id IS NOT NULL
-  HAVING count(*) > 0;  -- Only call archive if there are messages to archive
-
-  -- Mark current task as failed and store the output
-  UPDATE pgflow.step_tasks
-  SET status = 'failed',
-      failed_at = now(),
-      output = complete_task.output,  -- Store the output that caused the violation
-      error_message = '[TYPE_VIOLATION] Produced ' ||
-                     CASE WHEN complete_task.output IS NULL THEN 'null'
-                          ELSE jsonb_typeof(complete_task.output) END ||
-                     ' instead of array'
-  WHERE pgflow.step_tasks.run_id = complete_task.run_id
-    AND pgflow.step_tasks.step_slug = complete_task.step_slug
-    AND pgflow.step_tasks.task_index = complete_task.task_index;
-
   -- Mark step state as failed
   UPDATE pgflow.step_states
   SET status = 'failed',
@@ -159,16 +171,37 @@ IF v_dependent_map_slug IS NOT NULL THEN
     false
   );
 
-  -- Archive the current task's message (it was started, now failed)
-  PERFORM pgmq.archive(
-    v_run_record.flow_slug,
-    st.message_id  -- Single message, use scalar form
+  -- Terminalize every other unfinished task as cancelled, capturing their
+  -- message ids for archival below. Lock-order invariant: always lock/update
+  -- step_tasks before PGMQ queue rows. The culprit task is already terminal
+  -- (failed above), so it is excluded from the cancellation set.
+  WITH cancelled_tasks AS (
+    UPDATE pgflow.step_tasks AS task
+    SET status = 'cancelled'
+    WHERE task.run_id = complete_task.run_id
+      AND task.status IN ('queued', 'started')
+    RETURNING task.message_id
+  ),
+  culprit_task AS (
+    -- Terminal culprit row: safe to read for its message id after terminalization
+    SELECT st.message_id
+    FROM pgflow.step_tasks st
+    WHERE st.run_id = complete_task.run_id
+      AND st.step_slug = complete_task.step_slug
+      AND st.task_index = complete_task.task_index
+      AND st.message_id IS NOT NULL
   )
-  FROM pgflow.step_tasks st
-  WHERE st.run_id = complete_task.run_id
-    AND st.step_slug = complete_task.step_slug
-    AND st.task_index = complete_task.task_index
-    AND st.message_id IS NOT NULL;
+  SELECT ARRAY_AGG(ids.message_id) INTO v_violation_archived_ids
+  FROM (
+    SELECT message_id FROM culprit_task
+    UNION ALL
+    SELECT message_id FROM cancelled_tasks WHERE message_id IS NOT NULL
+  ) ids;
+
+  -- Archive the culprit and cancelled task messages (only after their task rows are terminalized)
+  IF v_violation_archived_ids IS NOT NULL THEN
+    PERFORM pgmq.archive(v_run_record.flow_slug, v_violation_archived_ids);
+  END IF;
 
   -- Return the failed task row (API contract: always return task row)
   RETURN QUERY