fix(celery): reschedule in-flight tasks on worker shutdown

[yasha-dev1]

Problem

When a worker is reclaimed mid-step (SIGTERM / AWS spot-node drain), its in-flight task message was only redelivered after the Redis broker visibility_timeout (~1h, time-based). The parent workflow stayed SUSPENDED that whole time, so the user-facing flow silently hung with no output.

Observed in production: a chatbot DeepAgent run sat cancelled/suspended for ~2h with recovery_attempts: 0, because the dispatched step's worker was reclaimed and nothing re-drove the run until the visibility window elapsed.

Fix

Re-enqueue a worker's in-flight tasks the moment its broker consumer is torn down, so a live worker's blocking BRPOP resumes them within seconds (event-based) instead of waiting out visibility_timeout.

• Children record what they're executing in a per-pod Redis hash pyworkflow:inflight:<pod> on task_prerun, and clear it on task_postrun.
• A consumer bootstep (RescheduleConsumerStep, requires Tasks) fires as the consumer is cancelled during shutdown. There is no Celery signal for that instant:
  • worker_shutting_down fires before the consumer is cancelled — an immediate re-publish gets self-re-grabbed back into Redis unacked.
  • worker_shutdown fires only after the pool drains the long task — too late.
• The bootstep cancels this worker's task consumer itself (guaranteeing it stops fetching), then re-enqueues: it releases the singleton lock and re-publishes to the same queue, so the message lands in the ready list for another worker instead of this dying one.
• Guarded by celery.worker.state.should_stop/should_terminate (identity checks — the flag is the exit code and 0 == False) so it only runs on real shutdown, not a transient consumer restart (broker reconnect).

Safe against double execution: only SingletonWorkflowTask tasks with unique_on are rescheduled, and execute_step_task short-circuits on an existing STEP_COMPLETED event. Gated by PYWORKFLOW_RESCHEDULE_ON_SIGTERM (default on).

Tests

• Unit (tests/unit/test_reschedule.py, + test_singleton.py): tracking rules, re-enqueue/lock-release, queue resolution, env gate, the consumer-stop hook + the shutdown-state guard, and release_lock key reproduction.
• Integration, real Redis (tests/integration/test_reschedule.py): registry hash round-trip, real singleton lock release + re-acquire, real broker enqueue lands in the ready list, and the dedup-bypass proof.
• End-to-end (tests/integration/test_reschedule_e2e.py): boots real prefork worker subprocesses, runs a blocking step, sends a real SIGTERM, hard-kills that worker so only a fresh worker can finish, and asserts the workflow completed within seconds. The broker visibility_timeout is 3600s, so completion proves the reschedule did it — not redelivery. (@integration/@slow, skipped without Redis.)

All green; ruff clean.

🤖 Generated with Claude Code