fix: keep started_at through completion and stop re-queuing live tasks

[adhikjoshi]

Two coupled defects in in-flight task tracking, found while timing the production generation pipeline end to end.

1. started_at is erased on every completion

The worker writes the pickup time into task_dict and saves it to Redis, but never sets it on the Task object:

task = Task.from_dict(task_dict)        # task.started_at is None
...
task_dict["started_at"] = time.time()   # the dict only

_store_final_task_state() then re-serialises from the object, overwriting the real value with null:

task_dict = task.to_dict()              # started_at: None
task_dict["finished_at"] = time.time()
self.redis_client.set(f"task:{task.task_id}", json.dumps(task_dict), ex=86400)

Measured on production Redis: of 6,953 completed tasks sampled across 24 queues, started_at survived on 127 (1.8%) — and on zero tasks in 18 of the 24 queues.

That is the only signal separating queue wait from actual run time, which is the number you need to tell "this endpoint needs more replicas" from "this model is slow". Reconstructing it required polling processing_tasks at 1 Hz for 22 minutes.

2. Healthy long jobs are re-queued and re-run

requeue_stuck_processing_tasks() asked "did this start more than threshold seconds ago?", which is not the same question as "is anyone still working on it?". A 500s video generation crosses the 180s default every time, gets pushed back onto ml_tasks, and a second GPU renders the same output again.

queue	GPU p50	GPU p90	tasks with a rewritten queued_at
video_server_ultra_wan22	499s	949s	7.8%
flux_klein	15.6s	21.4s	1.5%
ltx_2.3_server	196s	205s	1.0%

Why these ship together

Fixing (1) alone is a regression. Today a finished task left behind in processing_tasks has started_at: null, so the age rule skips it. Once started_at survives, that same stray looks exactly like an old stuck task — and gets re-run. Re-running a completed generation is worse than leaking a set member.

What changed

Liveness is now a heartbeat the owning worker republishes on the existing heartbeat thread (task_heartbeats hash, refreshed every HEARTBEAT_INTERVAL):

• a job that runs for 20 minutes is left alone as long as its worker is alive;
• a task whose worker actually died is re-queued once its heartbeat goes stale — sooner than the old age rule managed, not later;
• tasks in a terminal state are drained from processing_tasks rather than re-queued.

That last part also clears a backlog the erased started_at had been hiding. flux_klein's processing_tasks currently holds 61 ids of which 10 are real — 47 have already completed, 50 are over an hour old, the oldest is 23.7h. Anything reading that set for a processing count (including VerdaAutoscaler's queued + processing) is reading a number 6× too large.

Mixed-version safety: a task with no heartbeat falls back to the age rule, so a fleet running old and new workers side by side behaves exactly as it does today and converges as workers roll. Ordering is also tightened — _store_final_task_state() leaves processing_tasks before publishing the terminal blob, so an old sweeper can never see "finished + old started_at" together.

Side effect worth having: the sweep no longer GETs every in-flight payload once a minute on every worker just to decide to do nothing. A fresh heartbeat short-circuits before the fetch. On flux_klein that set is 61 entries of multi-hundred-KB payloads, read by 10 workers every 60s.

Tests

17 new tests in tests/test_inflight_tracking.py, including two that run a real worker through start_workers(). Full suite: 83 passed.

Mutation-tested — each fix removed, confirmed red, restored:

mutation	result
drop task.started_at = started_at	test_started_at_survives_a_real_completion → assert None is not None
ignore heartbeat freshness	test_long_running_task_with_fresh_heartbeat_is_not_requeued + 1 more fail
drop the terminal-status drain	test_finished_stray_is_drained_not_rerun + 3 more fail
drop _mark_task_inflight at pickup	test_a_task_is_marked_inflight_while_it_runs fails

Note on conflicts

Touches worker_loop and process_task, so it will likely conflict with open PR #17 (per-worker in-flight list). Happy to rebase whichever lands second.

---

^{Need help on this PR? Tag @codesmith-bot with what you need. Autofix is disabled.}