Fix/workers and schedulers

app/schemas/job.py

@@ -152,7 +152,6 @@ class JobResponse(BaseModel):
     completed_at: datetime | None
     is_dlq: bool
     created_at: datetime
-    updated_at: datetime
 
     # Populated on detail endpoint only
     dependencies: list[UUID] | None = None

app/services/dlq.py

@@ -7,7 +7,7 @@
 from app.core.logger import get_logger
 from app.models.job import Job, JobStatus
 from app.models.job_log import JobLog, LogEvent
-from app.queues.base import BaseQueue
+from app.queues.heapq import HeapQueue
 from app.services.settings import get_dlq_threshold
 
 logger = get_logger(__name__)
@@ -16,12 +16,19 @@
 async def send_to_dlq(
     job_id: uuid.UUID,
     error: str,
-    db: AsyncSession,
+    session: AsyncSession,
 ) -> None:
     """
     Move a job to the dead letter queue after exhausting all retries.
+
+    Steps:
+      1. Set status='failed', is_dlq=True, last_error, worker_id=None
+      2. Write job_failed log entry
+      3. Cascade-cancel downstream DAG dependents
+      4. Check DLQ count vs threshold — alert if breached
+      5. Publish SSE event (handled by caller via redis publish)
     """
-    await db.execute(
+    await session.execute(
         update(Job)
         .where(Job.id == job_id)
         .values(
@@ -39,8 +46,8 @@ async def send_to_dlq(
         message=f"Job moved to DLQ after exhausting all retries. Error: {error}",
         metadata_={"error": error, "reason": "max_retries_exhausted"},
     )
-    db.add(log_entry)
-    await db.flush()
+    session.add(log_entry)
+    await session.flush()
 
     logger.error(
         "job_failed",
@@ -49,16 +56,18 @@ async def send_to_dlq(
         reason="max_retries_exhausted",
     )
 
+    # Cascade cancel downstream jobs in the DAG
     from app.services import dag
 
-    await dag.on_job_failed(job_id, db)
+    await dag.on_job_failed(job_id, session)
 
-    await _check_and_alert(db)
+    await _check_and_alert(session)
 
 
 async def _check_and_alert(session: AsyncSession) -> None:
     """
     Count current DLQ jobs. If count >= threshold, fire alert email.
+    Imports alert_service lazily to avoid circular imports.
     """
     count = await get_dlq_count(session)
     threshold = await get_dlq_threshold(session)
@@ -71,29 +80,31 @@ async def _check_and_alert(session: AsyncSession) -> None:
             count=count,
             threshold=threshold,
         )
-        from app.services import alert
+        from app.services import alert as alert_service
 
-        await alert.send_dlq_alert(count, session)
+        await alert_service.send_dlq_alert(count, session)
 
 
 async def get_dlq_jobs(
     page: int,
     limit: int,
-    db: AsyncSession,
+    session: AsyncSession,
 ) -> tuple[list[Job], int]:
     """
     Return paginated DLQ jobs with total count.
+    Excludes soft-deleted jobs.
+    Ordered by most recently failed first.
     """
     base_filter = [
         Job.is_dlq.is_(True),
         Job.deleted_at.is_(None),
     ]
 
-    total_result = await db.execute(select(func.count(Job.id)).where(*base_filter))
+    total_result = await session.execute(select(func.count(Job.id)).where(*base_filter))
     total = total_result.scalar() or 0
 
     offset = (page - 1) * limit
-    jobs_result = await db.execute(
+    jobs_result = await session.execute(
         select(Job)
         .where(*base_filter)
         .order_by(Job.updated_at.desc())
@@ -105,9 +116,9 @@ async def get_dlq_jobs(
     return jobs, total
 
 
-async def get_dlq_count(db: AsyncSession) -> int:
+async def get_dlq_count(session: AsyncSession) -> int:
     """Return the current number of jobs in the DLQ."""
-    result = await db.execute(
+    result = await session.execute(
         select(func.count(Job.id)).where(
             Job.is_dlq.is_(True),
             Job.deleted_at.is_(None),
@@ -117,13 +128,14 @@ async def get_dlq_count(db: AsyncSession) -> int:
 
 
 async def get_recent_dlq_jobs(
-    db: AsyncSession,
+    session: AsyncSession,
     limit: int = 10,
 ) -> list[dict]:
     """
     Return the most recent DLQ jobs as plain dicts.
+    Used by alert_service to populate the email template.
     """
-    result = await db.execute(
+    result = await session.execute(
         select(Job)
         .where(
             Job.is_dlq.is_(True),
@@ -135,27 +147,35 @@ async def get_recent_dlq_jobs(
     jobs = result.scalars().all()
     return [
         {
-            "id": str(job.id),
-            "short_id": str(job.id)[:8],
-            "type": job.type,
-            "last_error": job.last_error or "Unknown error",
-            "retry_count": job.retry_count,
-            "max_retries": job.max_retries,
-            "updated_at": job.updated_at.strftime("%Y-%m-%d %H:%M:%S UTC"),
+            "id": str(j.id),
+            "short_id": str(j.id)[:8],
+            "type": j.type,
+            "last_error": j.last_error or "Unknown error",
+            "retry_count": j.retry_count,
+            "max_retries": j.max_retries,
+            "updated_at": j.updated_at.strftime("%Y-%m-%d %H:%M:%S UTC"),
         }
-        for job in jobs
+        for j in jobs
     ]
 
 
 async def retry_dlq_job(
     job_id: uuid.UUID,
-    db: AsyncSession,
-    queue: BaseQueue,
+    session: AsyncSession,
+    queue: HeapQueue,
 ) -> Job:
     """
     Manually retry a job from the DLQ.
+
+    Steps:
+      1. Verify job exists and is in DLQ
+      2. Reset: status='pending', is_dlq=False, retry_count=0,
+                last_error=None, worker_id=None
+      3. Call dag_service.on_dag_root_retried — reset downstream cancelled jobs
+      4. Re-evaluate dependencies: if all met, push to queue
+      5. Log and return updated job
     """
-    result = await db.execute(
+    result = await session.execute(
         select(Job).where(
             Job.id == job_id,
             Job.deleted_at.is_(None),
@@ -167,7 +187,7 @@ async def retry_dlq_job(
         raise JobNotInDLQException(str(job_id))
 
     # Reset the job
-    await db.execute(
+    await session.execute(
         update(Job)
         .where(Job.id == job_id)
         .values(
@@ -187,32 +207,28 @@ async def retry_dlq_job(
         message="Job manually retried from DLQ by engineer.",
         metadata_={"reason": "manual_dlq_retry"},
     )
-    db.add(log_entry)
-    await db.flush()
+    session.add(log_entry)
+    await session.flush()
 
     logger.info("dlq_job_retried", job_id=str(job_id))
 
-    from app.services import dag
+    from app.services import dag as dag_service
 
-    await dag.on_dag_root_retried(job_id, db)
+    await dag_service.on_dag_root_retried(job_id, session)
 
-    has_unmet = await dag.has_unmet_dependencies(job_id, db)
+    has_unmet = await dag_service.has_unmet_dependencies(job_id, session)
     if not has_unmet:
         await queue.push(
             job_id=str(job_id),
             effective_priority=float(job.priority),
             scheduled_at=job.scheduled_at.timestamp(),
             created_at=job.created_at.timestamp(),
         )
-        from app.services.job import _sync_job_to_redis
-
-        await _sync_job_to_redis(str(job_id), float(job.priority))
 
-    await db.commit()
+    await session.commit()
 
-    refreshed = await db.get(Job, job_id)
-    assert refreshed is not None
-    return refreshed
+    refreshed = await session.get(Job, job_id)
+    return refreshed  # type: ignore
 
 
 async def remove_from_dlq(