fix(core): terminalize permanently stalled tasks through whenExhausted

.changeset/permanently-stalled-escalation.md

@@ -0,0 +1,5 @@
+---
+"@pgflow/core": minor
+---
+
+Terminalize permanently stalled tasks through the step's `whenExhausted` policy. A new `pgflow.escalate_permanently_stalled()` function, scheduled as its own `pgflow_escalate_permanently_stalled` cron job, hands tasks marked with `permanently_stalled_at` to `fail_task()`, which now treats the marker as exhaustion regardless of remaining attempts: the run fails, the step skips or cascades per policy, dependents start, and siblings are cancelled or archived. `attempts_count` stays untouched as committed task claims, and ordinary failure callbacks on a marked task exhaust it instead of requeueing against its archived message. Includes the SQL migration (which preserves an absent or paused recovery cron job and copies the installed recovery interval), generated types, and upgrade guidance with a stamped-backlog inventory query.

pkgs/core/schemas/0063_function_setup_requeue_stalled_tasks_cron.sql

@@ -9,31 +9,56 @@ set search_path = pgflow, cron, pg_temp
 as $$
 declare
   job_id bigint;
+  escalate_job_id bigint;
 begin
-  -- Remove existing job if any
+  -- Remove existing jobs if any
   begin
     perform cron.unschedule('pgflow_requeue_stalled_tasks');
   exception
     when others then null;
   end;
 
-  -- Schedule the new job
+  begin
+    perform cron.unschedule('pgflow_escalate_permanently_stalled');
+  exception
+    when others then null;
+  end;
+
+  -- Schedule the recovery job
   job_id := cron.schedule(
     job_name => 'pgflow_requeue_stalled_tasks',
     schedule => setup_requeue_stalled_tasks_cron.cron_interval,
     command => 'select pgflow.requeue_stalled_tasks()'
   );
 
-  return format('Scheduled pgflow_requeue_stalled_tasks (every %s, job_id=%s)', 
-    setup_requeue_stalled_tasks_cron.cron_interval, job_id);
+  -- Schedule escalation as its own job so it runs in a separate transaction
+  -- after recovery commits: recovery locks step_tasks first, while
+  -- fail_task() (driven by escalation) locks the run and step first (#689).
+  -- Holding one job's locks across both calls would invert that order
+  -- against live workers.
+  escalate_job_id := cron.schedule(
+    job_name => 'pgflow_escalate_permanently_stalled',
+    schedule => setup_requeue_stalled_tasks_cron.cron_interval,
+    command => 'select pgflow.escalate_permanently_stalled()'
+  );
+
+  return format(
+    'Scheduled pgflow_requeue_stalled_tasks (every %s, job_id=%s) and pgflow_escalate_permanently_stalled (job_id=%s)',
+    setup_requeue_stalled_tasks_cron.cron_interval,
+    job_id,
+    escalate_job_id
+  );
 end;
 $$;
 
 comment on function pgflow.setup_requeue_stalled_tasks_cron(text) is
-'Sets up cron job to automatically requeue stalled tasks.
-Schedules pgflow_requeue_stalled_tasks at the specified cron_interval (default: 15 seconds).
-Replaces existing job if it exists (idempotent).
-Returns a confirmation message with job ID.';
+'Sets up cron jobs to automatically recover and terminalize stalled tasks.
+Schedules pgflow_requeue_stalled_tasks and pgflow_escalate_permanently_stalled
+at the specified cron_interval (default: 15 seconds).
+Escalation runs as a separate job so recovery commits before fail_task()
+locks are taken (#689).
+Replaces existing jobs if they exist (idempotent).
+Returns a confirmation message with job IDs.';
 
 -- Automatically set up the cron job when migration runs
 select pgflow.setup_requeue_stalled_tasks_cron();

pkgs/core/schemas/0064_function_escalate_permanently_stalled.sql

@@ -0,0 +1,80 @@
+-- Terminalize permanently stalled tasks through the step's when_exhausted
+-- policy (#689). requeue_stalled_tasks() archives the message and stamps
+-- permanently_stalled_at but leaves the task nonterminal; this function hands
+-- each marked task to fail_task(), which treats the marker as exhaustion and
+-- decides whether the run fails, the step skips, dependents start, and
+-- siblings are cancelled or archived.
+--
+-- Lock order matches fail_task(): lock the run and step first, then the task,
+-- so recovery (which locks step_tasks first) and live workers cannot deadlock.
+-- The candidate loop is a deliberate exception to the declarative-first SQL
+-- style: each candidate needs its own lock-and-recheck sequence before
+-- fail_task() runs, which a set operation cannot interleave.
+-- Runs as its own cron job, in a separate transaction from recovery, so
+-- recovery's locks are released before escalation begins. Safe to repeat:
+-- candidates are rechecked under lock, and rows already terminalized by
+-- fail_task() lose their nonterminal status and stop matching.
+create or replace function pgflow.escalate_permanently_stalled()
+returns int
+language plpgsql
+security definer
+set search_path = ''
+as $$
+declare
+  v_candidate record;
+  v_run_status text;
+  v_step_status text;
+  v_task_status text;
+  v_stalled_at timestamptz;
+  v_escalated int := 0;
+begin
+  -- Pick up every marked nonterminal task, including marked queued leftovers
+  -- requeued by older callbacks and rows stamped by earlier deployments
+  for v_candidate in
+    select
+      st.run_id,
+      st.step_slug,
+      st.task_index
+    from pgflow.step_tasks st
+    where st.status in ('started', 'queued')
+      and st.permanently_stalled_at is not null
+    order by st.run_id, st.step_slug, st.task_index
+  loop
+    -- Lock and recheck the run and step: if either left 'started', that
+    -- transition owns the task's terminal state and escalation must not
+    -- fight it
+    select r.status, step_state.status into v_run_status, v_step_status
+    from pgflow.runs r
+    join pgflow.step_states step_state on step_state.run_id = r.run_id
+    where r.run_id = v_candidate.run_id
+      and step_state.step_slug = v_candidate.step_slug
+    for update of r, step_state;
+
+    if v_run_status = 'started' and v_step_status = 'started' then
+      -- Lock and recheck the task: recovery or a live callback may have
+      -- transitioned it while this pass waited for the run and step locks
+      select st.status, st.permanently_stalled_at
+      into v_task_status, v_stalled_at
+      from pgflow.step_tasks st
+      where st.run_id = v_candidate.run_id
+        and st.step_slug = v_candidate.step_slug
+        and st.task_index = v_candidate.task_index
+      for update;
+
+      if v_task_status in ('started', 'queued') and v_stalled_at is not null then
+        -- fail_task() treats the marker as exhaustion and applies
+        -- when_exhausted; attempts_count stays untouched (#689)
+        perform pgflow.fail_task(
+          run_id => v_candidate.run_id,
+          step_slug => v_candidate.step_slug,
+          task_index => v_candidate.task_index,
+          error_message => 'permanently stalled: worker did not report after 3 requeues'
+        );
+        v_escalated := v_escalated + 1;
+      end if;
+    end if;
+  end loop;
+
+  return v_escalated;
+end;
+$$;

pkgs/core/schemas/0100_function_fail_task.sql

@@ -87,23 +87,37 @@ WITH flow_info AS (
 fail_or_retry_task as (
   UPDATE pgflow.step_tasks as task
   SET
+    -- Exhaustion: attempts_count reached max_attempts, or stalled-task
+    -- recovery gave up on the task (permanently_stalled_at set). A marked
+    -- task exhausts regardless of its remaining attempts: its message was
+    -- archived, so a retry could never be delivered. attempts_count stays
+    -- untouched; it records committed claims (#689).
     status = CASE
-      WHEN task.attempts_count < (SELECT opt_max_attempts FROM config) THEN 'queued'
+      WHEN task.permanently_stalled_at IS NULL
+        AND task.attempts_count < (SELECT opt_max_attempts FROM config) THEN 'queued'
       ELSE 'failed'
     END,
     failed_at = CASE
-      WHEN task.attempts_count >= (SELECT opt_max_attempts FROM config) THEN now()
-      ELSE NULL
+      WHEN task.permanently_stalled_at IS NULL
+        AND task.attempts_count < (SELECT opt_max_attempts FROM config) THEN NULL
+      ELSE now()
     END,
     started_at = CASE
-      WHEN task.attempts_count < (SELECT opt_max_attempts FROM config) THEN NULL
+      WHEN task.permanently_stalled_at IS NULL
+        AND task.attempts_count < (SELECT opt_max_attempts FROM config) THEN NULL
       ELSE task.started_at
     END,
     error_message = fail_task.error_message
   WHERE task.run_id = fail_task.run_id
     AND task.step_slug = fail_task.step_slug
     AND task.task_index = fail_task.task_index
-    AND task.status = 'started'
+    AND (
+      task.status = 'started'
+      -- Marked queued leftovers from older callbacks: terminalize them too.
+      -- Ordinary unmarked queued tasks stay unchanged (a recovery requeue
+      -- owns their next attempt).
+      OR (task.status = 'queued' AND task.permanently_stalled_at IS NOT NULL)
+    )
   RETURNING *
 ),
  -- Determine if task exhausted retries and get when_exhausted mode

pkgs/core/scripts/run-upgrade-fixture

@@ -218,3 +218,62 @@ SQL
 done
 
 echo "upgrade fixture: PASS (0.16.0 slug conflict rollbacks)"
+
+# =====================================================================
+# Fixture 5: escalation cron scheduling across upgrade states (#689)
+# =====================================================================
+# The escalation migration schedules pgflow_escalate_permanently_stalled
+# only where scheduled recovery already exists, copies the installed
+# recovery interval, and mirrors the recovery job's paused state. Proves:
+#   1. active recovery at a custom interval gains an active escalation job
+#      at that interval, while the recovery job itself is untouched,
+#   2. paused (active = false) recovery gains a paused escalation job,
+#   3. an intentionally absent recovery job gains no escalation job.
+ESCALATION_BASELINE='20261102104928_pgflow_worker_startup_handshake.sql'
+escalation_migration=$(ls supabase/migrations/*_pgflow_escalate_permanently_stalled.sql)
+
+for cron_case in active paused absent; do
+  start_container "pgflow-upgrade-fixture-escalation-${cron_case}"
+
+  echo "upgrade fixture escalation ${cron_case}: applying supabase baseline schema"
+  psql_in < atlas/supabase-baseline-schema.sql
+
+  echo "upgrade fixture escalation ${cron_case}: applying pgflow migrations up to ($ESCALATION_BASELINE)"
+  apply_migrations_up_to "$ESCALATION_BASELINE"
+
+  case "$cron_case" in
+    active)
+      psql_in <<'SQL'
+SELECT cron.alter_job(
+  job_id => job.jobid,
+  schedule => '45 seconds'
+)
+FROM cron.job job
+WHERE job.jobname = 'pgflow_requeue_stalled_tasks';
+SQL
+      ;;
+    paused)
+      psql_in <<'SQL'
+SELECT cron.alter_job(
+  job_id => job.jobid,
+  schedule => '45 seconds',
+  active => false
+)
+FROM cron.job job
+WHERE job.jobname = 'pgflow_requeue_stalled_tasks';
+SQL
+      ;;
+    absent)
+      psql_in <<'SQL'
+SELECT cron.unschedule('pgflow_requeue_stalled_tasks');
+SQL
+      ;;
+  esac
+
+  echo "upgrade fixture escalation ${cron_case}: applying migration $(basename "$escalation_migration")"
+  psql_in < "$escalation_migration"
+
+  sed "s/__CASE__/${cron_case}/g" supabase/upgrade_fixture/assertions_escalation_cron.sql | psql_in
+done
+
+echo "upgrade fixture: PASS (escalation cron scheduling states)"

pkgs/core/src/database-types.ts

@@ -582,6 +582,7 @@ export type Database = {
           request_id: number
         }[]
       }
+      escalate_permanently_stalled: { Args: never; Returns: number }
       fail_task: {
         Args: {
           error_message: string