feat: hold tasks in a per-worker in-flight list so none can be lost in transit

[adhikjoshi]

Stacks on #16. Based on fix/redis-blocking-read-hang because it rewrites the same queue-read line. GitHub will retarget this to main automatically once #16 merges. Review #16 first.

The hole

BLPOP removes the task from the queue and then writes it to the client. If that write is lost, the task exists nowhere:

• not in ml_tasks — it was popped
• not in processing_tasks — that is only populated after the reply arrives
• therefore invisible to requeue_stuck_processing_tasks and every other recovery path

This is not theoretical — I reproduced it in production

On 2026-08-15, with a worker's connection blackholed at the Redis host (verified biting: 23 packets dropped Redis→worker, 92 worker→Redis):

Marked probe pushed to ml_tasks	09:53:48
Redis served it to the blocked client	client idle reset at that exact moment — pop and write both happened
In the queue afterwards	0 copies, llen=0
Worker log	nothing — never received it

Both BLPOP clients were on the blackholed IP, so no other replica could have taken it. The task was destroyed.

This very likely explains the generation records seen stuck in processing with nothing queued behind them — 15 on text_to_audio and 20 on flux_2_dev that same morning.

The fix

BLMOVE makes taking the task and recording who took it one atomic step.

• The worker moves the task into inflight:{server_id}:{worker_id} and holds custody until done — whatever the outcome — via a finally.
• Every in-flight list is registered in a set, so recovery enumerates them rather than SCANning (this codebase has been bitten by SCAN storms before).
• At startup a server drains its own lists: those can only be debris from a previous run of the same id.
• The pruning loop drains lists whose owning server is no longer registered, running after prune_inactive_servers so the dead are already deregistered on the same pass.

The one flag that must not be wrong

include_self. At startup our lists are debris and must be drained. From the periodic sweep they belong to our own running workers, and draining those would hand a live task to a second worker while the first is still executing it. Both directions have a test.

drain_inflight is bounded by the list length read up front rather than looping until empty — an unbounded loop would spin the day the no-appender assumption breaks.

Compatibility

Requires Redis 6.2+ for BLMOVE/LMOVE. The fleet runs 7.4.2.

Testing

10 new tests in tests/test_inflight_custody.py; suite 62 → 72. The bounded-read test from #16 now asserts the timeout on BLMOVE rather than BLPOP — same property, new command.

Mutation-tested, and worth being explicit about: reverting to BLPOP and dropping the custody release initially survived. My first assertions only inspected list lengths after the fact, which is true either way — a test that passes both ways is worse than no test. The custody test now observes a genuinely in-flight task mid-run, holding the worker inside the task body with an event:

mutation	result
revert BLMOVE → BLPOP	red
drop the lrem custody release	red
let the sweep drain live workers' lists	red

What this does not fix

Only loss in transit to a worker. A worker that receives a task and then dies mid-execution is covered by the existing processing_tasks sweep, unchanged here. And this is orthogonal to #16: #16 stops workers stalling, this stops tasks vanishing during whatever stall remains.

---

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