fix(worker): periodic zombie cleanup via beat instead of worker_ready

worker_ready signal handler was NOT firing in our setup (verified via
DB breadcrumb after 3 deploys — zero rows of stage='worker_ready' in
nspd_geo_log). Root cause of unreliability unknown — possibly Celery
internals, possibly compose recreate timing. Either way, after every
redeploy users had to manually cancel/resume jobs to keep them moving.

Replace signal-based resume with periodic beat task:
- cleanup_zombies runs every minute (* * * * *)
- Finds nspd_geo_jobs in status running/paused with heartbeat >2 min stale
- Sets status='queued' + apply_async with queue=geo
- Idempotent — if no zombies, no-op

worker_ready handler kept (with FK-fix breadcrumb on NULL job_id) for
diagnostic purposes — if signal ever does fire, we'll have evidence.

backend/app/workers/celery_app.py

@@ -203,6 +203,15 @@ def _build_beat_schedule() -> dict:
         except Exception as e:
             logger.warning("beat_schedule: refresh_analytics failed: %s", e)
 
+    # Zombie cleanup: каждую минуту проверяет nspd_geo_jobs с heartbeat > 2 мин
+    # и re-enqueue'ит их. Замена worker_ready signal handler'а — он не fires
+    # стабильно в нашем setup, beat-task надёжнее.
+    schedule["nspd-geo-zombie-cleanup"] = {
+        "task": "tasks.nspd_geo.cleanup_zombies",
+        "schedule": _parse_cron("* * * * *"),
+        "options": {"queue": "celery"},
+    }
+
     return schedule
 
 
@@ -249,6 +258,8 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None:
 
     # Persistent breadcrumb: write to nspd_geo_log so we can confirm via DB
     # query whether worker_ready actually fired (independent of container logs).
+    # NULL job_id to avoid FK violation (FK is to nspd_geo_jobs.job_id which
+    # has no 0 row).
     try:
         from app.core.db import SessionLocal
 
@@ -257,7 +268,7 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None:
             _db.execute(
                 text(
                     "INSERT INTO nspd_geo_log (job_id, level, stage, message) "
-                    "VALUES (0, 'info', 'worker_ready', 'worker_ready signal fired')"
+                    "VALUES (NULL, 'info', 'worker_ready', 'worker_ready signal fired')"
                 )
             )
             _db.commit()

backend/app/workers/tasks/nspd_geo.py

@@ -496,3 +496,54 @@ def process_nspd_geo_job(self: Any, job_id: int) -> dict[str, Any]:
         raise
     finally:
         db.close()
+
+
+@celery_app.task(name="tasks.nspd_geo.cleanup_zombies")
+def cleanup_zombies() -> dict[str, Any]:
+    """Periodic zombie cleanup — runs every minute via beat schedule.
+
+    Catches nspd_geo_jobs in status='running' / 'paused' with stale heartbeat
+    (>2 min) and re-enqueues them. Replaces the unreliable worker_ready signal
+    handler — beat is independent of worker lifecycle and fires every minute.
+
+    Idempotent: if no zombies, does nothing. If a job is genuinely active, its
+    heartbeat will be fresh and the WHERE clause won't match.
+    """
+    db = SessionLocal()
+    resumed: list[int] = []
+    try:
+        rows = (
+            db.execute(
+                text(
+                    """
+                    UPDATE nspd_geo_jobs
+                       SET status = 'queued',
+                           error = COALESCE(error, 'auto-resume by beat cleanup')
+                     WHERE status IN ('running', 'paused')
+                       AND heartbeat_at < NOW() - INTERVAL '2 minutes'
+                    RETURNING job_id
+                    """
+                )
+            )
+            .mappings()
+            .all()
+        )
+        db.commit()
+        resumed = [int(r["job_id"]) for r in rows]
+        for jid in resumed:
+            try:
+                process_nspd_geo_job.apply_async(args=[jid], queue="geo")
+                logger.info("cleanup_zombies: re-enqueued job=%s", jid)
+            except Exception as e:
+                logger.warning("cleanup_zombies: enqueue job=%s failed: %s", jid, e)
+        if resumed:
+            logger.info("cleanup_zombies: resumed %d zombie job(s): %s", len(resumed), resumed)
+    except Exception as e:
+        logger.exception("cleanup_zombies failed: %s", e)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+    finally:
+        db.close()
+    return {"resumed_count": len(resumed), "resumed_job_ids": resumed}