Make start_tasks() apply task visibility before handlers run

[jumski]

Summary

pgflow.start_tasks() computes effective per-task visibility delays and calls pgflow.set_vt_batch() in an unreferenced SELECT CTE. PostgreSQL may skip an unreferenced SELECT CTE, so a claimed task can keep only the shorter visibility set by the initial PGMQ read.

The message may become visible while its step_tasks row remains started. Current workers repeatedly read that message and return fewer tasks than messages. Future per-step workers must not mistake this healthy duplicate visibility for unsupported work and stop.

Current path

pkgs/core/schemas/0120_function_start_tasks.sql contains:

timeouts as (
  select
    task.message_id,
    task.flow_slug,
    coalesce(step.opt_timeout, flow.opt_timeout) + 2 as vt_delay
  from tasks task
  ...
),
set_vt_batch as (
  select pgflow.set_vt_batch(...)
  from timeouts
)
select ...
from tasks ...

The final query does not reference set_vt_batch. This does not guarantee that pgflow.set_vt_batch() runs.

Required behavior

For every task that changes from queued to started:

1. compute the same effective timeout used by stalled recovery;
2. extend that message's visibility before returning the task to a handler;
3. increment the attempt exactly once;
4. return no task whose visibility extension failed;
5. preserve atomic behavior for the claimed batch.

A message that becomes visible while its exact task remains started is a benign duplicate. Reading it must consume no new attempt and must not stop the worker.

Implementation boundary

Make the visibility side effect structurally required. Use PL/pgSQL PERFORM, a referenced CTE whose result the final statement consumes, or another statement form whose execution does not depend on demand for an unreferenced SELECT CTE.

Keep timeout semantics aligned across:

initial PGMQ read visibility
start_tasks() effective visibility extension
#621 stalled-task threshold
requeue_stalled_tasks() visibility reset

Do not fold queue-per-step routing into this fix. #650 will later add canonical queue_name and group visibility updates by queue.

Tests

• A claimed task's PGMQ visibility uses coalesce(step.opt_timeout, flow.opt_timeout) + 2 before start_tasks() returns.
• A shorter step timeout overrides the flow timeout.
• A longer step timeout is not shortened to the flow timeout.
• A null step timeout falls back to the flow timeout.
• A visibility-update failure rolls back the task transition and attempt increment.
• A repeatedly visible message for a started task consumes no additional attempt.
• A mixed batch cannot return a task whose visibility update did not execute.

Relationship to other work

• Parent epic: Epic: staged per-step queues, aliases, and shared queue routing #653.
• requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621 fixes the recovery threshold after a worker stalls.
• PR Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #649 fixes terminal state under skip paths.
• Persist physical queue identity and ownership for flow tasks #650 must build queue-aware claiming on top of this corrected visibility behavior.
• Add private per-step queues with typed step workers #651 and Add explicit shared queues with multi-flow dispatch #652 must classify visible started messages as benign duplicates rather than unsupported work.