fix(core): terminalize unfinished tasks when runs fail (#663)

## Summary

Failed runs archive their PGMQ messages but leave sibling `step_tasks` rows `queued` or `started`, so stale recovery can requeue undispatchable work and late callbacks can misrepresent terminal state (#645).

## Invariant

```text
runs.status = 'failed'
  implies
no step_tasks row for that run has status IN ('queued', 'started')
```

Task outcomes stay truthful:

- task that caused the failure → `failed`
- task completed before the failure → `completed`
- unfinished queued or started task → `cancelled`

## Why `cancelled` and not `failed`/`skipped`

- `failed` counts real handler failures; reusing it corrupts failure metrics.
- `skipped` is step-skip policy (`whenUnmet`/`whenExhausted`); reusing it conflates run failure with explicit skip decisions.
- `cancelled` is orchestration state invalidated by the run failure.
- `runs.failed_at` is the cancellation time; no `cancelled_at` column and no reason column were added.

Database cancellation does not terminate JavaScript already executing in a worker and does not undo external side effects the handler already performed. Worker-side abort behavior remains out of scope (#646).

## Changed failure paths

1. **`fail_task()` with exhausted `when_exhausted = 'fail'`** — the culprit stays `failed`; every remaining queued/started task in the run becomes `cancelled` with `UPDATE ... RETURNING` message capture, then archived. Retries before exhaustion are preserved.
2. **`complete_task()` type violation** — the directly invalid task is marked `failed` first (output and `[TYPE_VIOLATION]` error preserved), then every other queued/started task is cancelled and archived afterward.
3. **`cascade_resolve_conditions()` with `when_unmet = 'fail'`** — the condition step and run still fail; unfinished tasks across all independent branches become cancelled, then archived. The run transition is now conditional (`WHERE status = 'started'`) so replayed or concurrent calls cannot duplicate events.
4. **Already-failed-run `fail_task()` path** — no longer rewrites a `started`/`cancelled` task to `failed`; it only archives any still-active message and returns the row.

All three paths reuse the #638/#649 lock-order pattern: update task rows first, capture message IDs with `RETURNING`, archive queue rows only afterward.

## Late-callback and concurrency guards

- `complete_task()` rechecks `runs.status` under the `FOR UPDATE` lock and returns unchanged when the run is failed — closes the race where the pre-lock guard saw `started` but the run failed while the callback waited.
- `fail_task()` captures run status under the existing run/step lock and rechecks it after lock acquisition for the same race.
- Cancellation wins over late `complete_task()` and `fail_task()`; repeated callbacks are idempotent and do not revive cancelled rows, rewrite history, clear `runs.failed_at`, duplicate events, or drift counters.
- A committed completed task remains `completed`; a genuine failure remains `failed`.

## Stalled recovery guards

`requeue_stalled_tasks()` now requires `run.status = 'started'` AND `step_state.status = 'started'` AND `task.status = 'started'`. Genuine recovery, requeue counts, max-requeue and permanent-stall behavior, and the current timeout calculation are unchanged (#621 owns the timeout bug).

## Migration and backfill

`20260901183450_pgflow_temp_failed_run_terminalization.sql` (Atlas-generated, temp-prefixed for the stacked #645 → #621 → #656 order):

1. allows `cancelled` in `step_tasks.valid_status`;
2. recreates the four changed functions;
3. repairs historical rows — only `queued`/`started` tasks on `failed` runs flip to `cancelled`; completed/failed outcomes and every history field (attempts, requeue history, worker identity, timestamps, outputs, errors) are preserved.

Verified with an upgrade fixture: DB reset through `20260827180017`, a failed run holding completed/failed/started/queued tasks, migration applied → completed/failed/cancelled/cancelled with all history fields byte-identical.

## Tests

- `fail_task/archive_sibling_map_tasks.test.sql` — exhausted map: task 0 failed, task 1 started→cancelled, task 2 queued→cancelled; run/step failed; zero active tasks/messages; all archived.
- `type_violations/cancels_unfinished_tasks.test.sql` — invalid task failed with preserved output/error; completed work stays completed; unrelated queued/started tasks cancelled.
- `condition_evaluation/dependent_unmet_fail_archives_active_messages.test.sql` — completed trigger task stays completed; condition step fails; independent unfinished work cancelled; replay emits no duplicate `run:failed`.
- `complete_task/no_mutations_on_failed_run.test.sql` — late `complete_task()` after run failure leaves the task cancelled with history intact.
- `fail_task_when_exhausted/cancelled_task_late_callbacks_are_idempotent.test.sql` — repeated late `complete_task()`/`fail_task()` leave cancelled rows, history, parent state, counters, and events unchanged.
- `fail_task_when_exhausted/late_callbacks_post_lock_race.test.sql` — two-session dblink test for the post-lock race: a ctrl session holds `FOR UPDATE` on the run row; late `complete_task()` and `fail_task()` — both aimed at the task the winning failure cancels — block after their pre-lock checks (proven blocked by ctrl via `pg_blocking_pids()`); ctrl then fails the run and commits. Both callbacks return the cancelled row unchanged: no output, no error, no `failed_at`, no moved `runs.failed_at`, no duplicate events, counters untouched.
- `requeue_stalled_tasks/ignores_terminal_parent_state.test.sql` — started run/step recovery still requeues; failed-run and terminal-step stale tasks are ignored.
- Each failure-path regression installs the #638 queue-table trigger that fails deterministically if archival ever precedes task terminalization.

## Checks run

- `pnpm nx test:pgtap core --skip-nx-cache` — 284 files, 1316 tests, all pass
- `pnpm nx verify-migrations core --skip-nx-cache` — pass
- `pnpm nx gen-types core --skip-nx-cache` / `pnpm nx verify-gen-types core --skip-nx-cache` — pass (types unchanged; status columns are `string`)
- `pnpm nx test core --skip-nx-cache` — pass
- `pnpm nx build core --skip-nx-cache` — pass
- `pnpm nx lint core --skip-nx-cache` — pass (`fix-sql` hit the known sqruff `--force` incompatibility; equivalent direct `sqruff fix schemas/` ran clean)
- `pnpm changeset status`, `git diff --check` — clean

## Out of scope

#621 (effective-timeout calculation), #656 (queue identity / visibility extension), #646 (worker-side handler cancellation), migration consolidation, manual tasks.

Closes #645.

Author: jumski

.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