[BUG] K8s executor: block that succeeds on Job retry still fails the pipeline run (backoff_limit > 0)

Describe the bug

When using the Kubernetes executor with backoff_limit > 0, a block that fails on the first pod attempt and then succeeds on a Job retry still causes the whole pipeline run to be marked FAILED, and a failure notification is sent.

The BlockRun correctly ends up COMPLETED, but the PipelineRun stays FAILED and the pipeline stops. It has to be retried by hand, even though Kubernetes already recovered the block on its own.

The net effect is that backoff_limit does not actually buy any resilience for transient block errors: the retry happens and succeeds, but the pipeline has already been failed and the on-call has already been paged.

Root cause

Three code paths race, because BlockRun status is written from inside the job pod while the Job's retry state is only known to the scheduler.

1. mage_ai/data_preparation/executors/block_executor.py (~L686) — when the block raises, the in-pod executor writes BlockRun.BlockRunStatus.FAILED straight to the shared DB. This happens on every pod attempt, with no knowledge of whether the Job has retries left:

self.__update_block_run_status(
    BlockRun.BlockRunStatus.FAILED,
    block_run_id=block_run_id,
    ...
)

1. mage_ai/services/k8s/job_manager.py (L90–L106) — the scheduler side does this correctly. It keeps waiting until the Job either succeeds or exhausts its backoff:

backoff_limit = job.spec.backoff_limit or 0
while not job_completed:
    ...
    if succeeded >= 1:
        job_completed = True
    elif failed >= backoff_limit:
        job_completed = True

1. mage_ai/orchestration/pipeline_scheduler_original.py (~L246) — meanwhile the scheduler loop independently observes the FAILED block run written in (1) and fails the entire pipeline run, before the Job in (2) has finished retrying:

if self.pipeline_run.any_blocks_failed():
    self.pipeline_run.update(
        status=PipelineRun.PipelineRunStatus.FAILED,
        completed_at=datetime.now(tz=pytz.UTC),
    )
    ...
    self.on_pipeline_run_failure(error_msg)

So (1) publishes a non-terminal, per-attempt failure as if it were terminal, and (3) acts on it while (2) is still legitimately retrying.

To Reproduce

1. Use the k8s executor with backoff_limit set above 0 in the executor config (services/k8s/config.py reads executor_config.job.backoff_limit; default is 0).
2. Have a block fail on its first attempt with a transient/retryable error (in our case a Redshift XX000 on a materialized-view refresh that conflicts with a concurrent vacuum/truncate — the DB's own message is "Please try again").
3. Let the Kubernetes Job retry the pod.

Observed run (times UTC, backoffLimit=3, restartPolicy=Never, completions=1):

05:14:13  K8sBlockExecutor creates Job mage-data-prep-block-<id>
05:15:08  attempt 1 pod starts
05:17:05  attempt 1 raises -> in-pod executor writes BlockRun = FAILED
05:17:21  scheduler sees any_blocks_failed() ->
          PipelineRun = FAILED, completed_at stamped, failure notification sent
05:18:14  Kubernetes retries: attempt 2 pod starts   <-- backoff_limit working as intended
05:41:05  Job reaches SuccessCriteriaMet / CompletionsReached
05:41:10  BlockRun = COMPLETED, "Finish executing block with K8sBlockExecutor"

Final DB state — the block succeeded, but the run was already failed and stayed that way until a manual retry:

block_run     status=COMPLETED  started_at=05:14:11  completed_at=05:41:10
pipeline_run  status=FAILED     completed_at=05:17:21

Expected behavior

A block whose Kubernetes Job ultimately succeeds within backoff_limit should not fail the pipeline run, and should not emit a failure notification.

Concretely, one of:

• the in-pod executor should not write a terminal FAILED while the Job still has attempts remaining (e.g. write an ATTEMPT_FAILED/retrying state, or let the Job outcome in job_manager be what sets the terminal status); or
• pipeline_scheduler should not treat a FAILED block run as terminal while its backing Job is still active.

Secondary bug: completed_at is never cleared

Transitions back to RUNNING set the status without clearing completed_at:

• mage_ai/api/resources/PipelineRunResource.py L396: return super().update(dict(status=PipelineRun.PipelineRunStatus.RUNNING))
• mage_ai/orchestration/pipeline_scheduler_original.py L193: updates started_at + status=RUNNING, leaves completed_at

After retrying the run above, we are left with a contradictory row that a RUNNING pipeline run carries a completion timestamp from its earlier failure:

pipeline_run  status=RUNNING  started_at=04:01:11  completed_at=05:17:21

This is misleading in the UI and in any reporting built on completed_at. completed_at should be reset to NULL whenever a run moves back to RUNNING.

Additional context

• mage-ai 0.9.78, Kubernetes executor on EKS. Line numbers are from the installed 0.9.78 package.
• Related but distinct: #4150 is the inverse symptom (Job exhausts backoff_limit and fails, Mage still shows it running). Both come from the same underlying design point — BlockRun status is authored in-pod rather than derived from the Job's terminal state — so they may be worth fixing together.
• backoff_limit support came from #5032.