Resume failed runs by reusing every completed task

[jumski]

Summary

Add pgflow.resume_run to create a new execution from a failed run while reusing every task that already completed successfully. The new run must enqueue only unfinished task indexes, including within partially completed map steps and parallel branches.

A step-level approximation is explicitly not acceptable. If a map step completed 9 of 10 tasks before failure, resuming must reuse those nine task outputs and queue only the missing task.

User report or idea

A failed run should be resumable from its actual task frontier, not merely from its last fully completed step. The operation should:

• create a new run rather than mutate the failed run;
• copy the same flow input;
• set restart_of_run_id to the failed source run;
• reuse every completed task from the source run;
• reuse fully completed step outputs;
• queue every unfinished task index with a new queue message;
• continue normal dependency, condition, aggregation, and completion behavior.

This issue is intentionally separate from a full retry. retry_run creates a clean run and reuses no execution state. resume_run means task-level continuation or should not exist.

The lineage field must remain restart_of_run_id. parent_run_id is reserved for planned subflows, where a parent task links to a child run and completes after that child run completes.

Evidence supplied

No runtime logs or external reproduction were supplied. The design came from inspection of the current schema source:

pkgs/core/schemas/0060_tables_runtime.sql
pkgs/core/schemas/0100_function_start_flow.sql
pkgs/core/schemas/0100_function_start_ready_steps.sql
pkgs/core/schemas/0100_function_complete_task.sql
pkgs/core/schemas/0100_function_fail_task.sql
pkgs/core/schemas/0100_function_cascade_resolve_conditions.sql
pkgs/core/schemas/0100_function_cascade_complete_taskless_steps.sql
pkgs/core/schemas/0100_function_maybe_complete_run.sql

Current task generation always creates every index:

CROSS JOIN LATERAL generate_series(0, started_step.initial_tasks - 1) AS task_idx(task_index)

Current step start bookkeeping sets all tasks as remaining:

remaining_tasks = ready_steps.initial_tasks

Current map completion aggregates all completed task rows by index. This existing behavior is suitable for reused task rows:

jsonb_agg(all_outputs.output ORDER BY all_outputs.task_index)

Investigation and findings

pgflow.fail_task terminalizes the run after an exhausted failure with when_exhausted = 'fail' and archives active PGMQ messages across the run. At that point a map or parallel step may contain any combination of completed, failed, queued, and started tasks. Issue #645 proposes terminalizing unfinished rows as cancelled; resume behavior must not depend on which non-completed terminal or active status the source row has.

The reusable unit is therefore:

step_tasks.status = 'completed'

Every other task index is unfinished for resume purposes and needs a fresh queue message and a reset attempt count.

Fully completed steps can be copied as completed with their stored step_states.output. Partially completed steps cannot be copied as completed. They must start in the new run with their completed task rows already present and with remaining_tasks equal to the number of missing indexes.

The existing complete_task path already supports the final aggregation needed by resumed maps. When the last pending task completes, its map-output query reads all previously completed tasks in the new run, including copied rows, adds the current task, and orders the result by task_index.

The existing start_ready_steps implementation is the one blocking task-level resume. It always sets remaining_tasks = initial_tasks and always queues every index from 0 through initial_tasks - 1. Generalizing it to account for pre-existing completed task rows keeps queue batching in one place and leaves fresh-run behavior unchanged.

Only successful work should be reused. A skipped step is not treated as completed work: reset it to created and let cascade_resolve_conditions evaluate it again against the copied input and reused dependency outputs. Failure, error, and skip metadata from incomplete steps must not leak into the new run.

Proposed solution or design

Preconditions and run creation

Add:

pgflow.resume_run(run_id uuid)
returns setof pgflow.runs

The function should reject a missing source and a source whose status is not failed. It should create a new run with:

flow_slug            = source.flow_slug
input                = source.input
restart_of_run_id    = source.run_id
status               = started
remaining_steps      = total steps - reused completed steps

The source run remains unchanged.

Seed step states

Create one new step_states row for every flow step.

For a source step with status = 'completed':

• copy status = 'completed';
• copy its stored output;
• set remaining_tasks = 0;
• retain a valid initial_tasks value;
• count it as already complete in the new run.

For every other source step status (created, started, failed, or skipped):

• create it with status = 'created';
• clear output, error, skip reason, and terminal timestamps;
• preserve or reconstruct initial_tasks from the source state;
• recompute remaining_deps from dependencies whose steps were copied as completed;
• leave remaining_tasks null until the step starts.

A dependent step whose source dependencies were completed should become ready immediately. A dependency reset from skipped or failed state should resolve through the normal cascade before its child starts.

Copy every completed task

Copy every source step_tasks row whose status is completed, including rows under:

• fully completed steps;
• failed map steps with partial success;
• interrupted parallel steps with partial success.

Each copied row uses the new run_id, the same step_slug and task_index, and the same successful output. Set message_id to null because the source PGMQ message belongs to the old run and may already be archived.

Do not copy failed, cancelled, skipped, queued, or started task rows. Their indexes remain missing and receive new task rows/messages. Their new attempt counts start at zero.

Generalize start_ready_steps

For each ready created step, calculate pending work from its total task count and its existing completed task rows:

remaining_tasks = initial_tasks - count(existing completed tasks)

Generate queue messages only for missing completed indexes:

generate_series(0, initial_tasks - 1)
where not exists (
  select 1
  from pgflow.step_tasks as existing_task
  where existing_task.run_id = started_step.run_id
    and existing_task.step_slug = started_step.step_slug
    and existing_task.task_index = task_idx.task_index
    and existing_task.status = 'completed'
)

Insert fresh queued rows and PGMQ messages only for those indexes. Fresh runs contain no pre-existing task rows, so they continue to queue every index exactly as today.

Keep queue-message batching in start_ready_steps; do not duplicate it inside resume_run.

Continue through the normal cascade

After state seeding, invoke the same post-creation sequence as start_flow:

cascade_resolve_conditions
cascade_complete_taskless_steps
start_ready_steps
maybe_complete_run

Copied completed steps and tasks should not emit new completion events. Newly resumed steps should emit the normal start and eventual completion events. The new run should emit run:started with restart_of_run_id through the lineage work in #657.

Required examples

Partially completed map:

source initial_tasks: 10
source task 0..8:      completed
source task 9:         failed

new copied tasks:      0..8 completed
new remaining_tasks:   1
new queued tasks:      9 only

When task 9 completes, existing map aggregation must produce an ordered ten-item output from copied tasks 0..8 plus new task 9.

Parallel work:

branch A: causes run failure
branch B: has completed and unfinished map tasks

The resumed run must reuse branch B's completed tasks and queue only branch B's unfinished indexes, even though branch B did not cause the original failure.

Development workflow

Develop test-first from pkgs/core/schemas/. Add focused pgTAP tests before changing schema functions, apply schema files with psql, run focused and full pgTAP suites, then generate the migration through Atlas.

Likely test location:

pkgs/core/supabase/tests/resume_run/

Acceptance criteria

• pgflow.resume_run creates a new run only from a failed source run.
• The new run copies the source flow_slug and input and sets restart_of_run_id to the source.
• The source run and all source state remain unchanged.
• Every completed source task is represented as completed under the same step slug and task index in the new run.
• Completed tasks inside partially failed map steps and interrupted parallel steps are reused.
• No source task whose status is not completed is treated as successful or reused.
• Old PGMQ message IDs are not reused.
• Every unfinished task index receives exactly one fresh task row and one fresh queue message.
• Attempt counts for newly queued task indexes start at zero.
• A resumed step sets remaining_tasks to the number of unfinished indexes, not initial_tasks.
• A resumed map's final output contains reused and newly completed task outputs in task_index order.
• Fully completed steps do not run again and retain their stored outputs for dependent steps.
• Skipped and failed steps are reset and processed through normal condition and dependency behavior.
• Fresh start_flow behavior remains unchanged when no completed task rows already exist.
• The normal condition, taskless-step, ready-step, event, and run-completion paths remain authoritative.
• Incompatible flow or task shape cannot silently reuse outputs under the wrong step or task index.
• Focused pgTAP tests cover partial maps, parallel partial work, completed dependencies, zero-task maps, and queue-message uniqueness.
• Full pgTAP, migration, concurrency, and generated-type checks pass.

Related work

• Retry failed runs as fresh runs with explicit restart lineage #657 — adds fresh-run retry behavior and the restart_of_run_id lineage used by this issue. This is a plain related-work link, not a native dependency relationship.
• Failed runs leave unfinished sibling tasks queued or started instead of cancelled #645 — proposes cancelled for unfinished tasks on failed runs. Resume must reuse only completed rows and treat cancelled like every other unfinished status.
• Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #638 — terminalizes active task rows under skipped steps. Resume does not reuse skipped task outcomes; conditions are evaluated again in the new run.
• Use terminal events to cooperatively abort in-flight worker tasks #646 — explores cooperative cancellation of handlers after a run becomes terminal. This issue concerns persisted state and new queue messages, not stopping old JavaScript execution.

Open questions and risks

Reused-task timestamps and provenance

Copied task rows need a non-null queued_at. Decide whether to preserve source timestamps or timestamp the reuse in the new run. Preserving source timestamps accurately records when the work ran but places task timestamps before the new run's started_at. New timestamps avoid that ordering but can look like zero-duration execution.

restart_of_run_id lets callers find the immediate source task by (step_slug, task_index). If consumers must query reuse directly without comparing runs, consider explicit task provenance such as reused_from_run_id. Do not add it without a concrete observability requirement.

Flow definition drift

Task indexes and dependency outputs are safe to reuse only under a compatible flow shape. Current production compilation rejects shape mismatches, while local recompilation deletes flow data, but runs does not store an explicit flow-version snapshot.

The implementation must either prove the current lifecycle prevents a surviving incompatible source run or add a compatibility check. If compatibility cannot be established, fail with guidance to use retry_run instead of silently reusing data.

Locking and callbacks

A failed source run is terminal, but late worker callbacks and task-state cleanup can race with failure. Review the existing run/step lock order and make sure the resume snapshot sees one committed source state. Only tasks committed as completed before the snapshot may be reused.