fix(scheduler): skip cron runs while an automation is active

openhands/automation/scheduler.py

@@ -11,14 +11,19 @@
 import asyncio
 import logging
 from datetime import datetime, timedelta
+from uuid import UUID
 from zoneinfo import ZoneInfoNotFoundError
 
 from croniter import CroniterBadDateError, CroniterBadTypeRangeError, CroniterError
 from sqlalchemy import select, update
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from openhands.automation.db import using_sqlite
-from openhands.automation.models import Automation, AutomationRun
+from openhands.automation.models import (
+    Automation,
+    AutomationRun,
+    AutomationRunStatus,
+)
 from openhands.automation.telemetry import capture_automation_event
 from openhands.automation.utils import get_next_fire_time, is_automation_due, utcnow
 from openhands.automation.utils.run import create_pending_run
@@ -38,6 +43,11 @@
     ZoneInfoNotFoundError,
 )
 
+_IN_FLIGHT_RUN_STATUSES = (
+    AutomationRunStatus.PENDING,
+    AutomationRunStatus.RUNNING,
+)
+
 
 def _schedule_log_extra(automation: Automation) -> dict[str, str | None]:
     trigger = automation.trigger or {}
@@ -142,6 +152,24 @@ async def _fetch_enabled_automations(
     return list(result.scalars().all())
 
 
+async def _get_in_flight_run_statuses(
+    session: AsyncSession,
+    automations: list[Automation],
+) -> dict[UUID, AutomationRunStatus]:
+    """Return one active run status for each automation that has one."""
+    automation_ids = [automation.id for automation in automations]
+    if not automation_ids:
+        return {}
+
+    result = await session.execute(
+        select(AutomationRun.automation_id, AutomationRun.status).where(
+            AutomationRun.automation_id.in_(automation_ids),
+            AutomationRun.status.in_(_IN_FLIGHT_RUN_STATUSES),
+        )
+    )
+    return {automation_id: status for automation_id, status in result}
+
+
 async def poll_and_schedule(
     session_factory: async_sessionmaker[AsyncSession],
     batch_size: int = DEFAULT_BATCH_SIZE,
@@ -152,8 +180,9 @@ async def poll_and_schedule(
     Fetches enabled automations (using FOR UPDATE SKIP LOCKED on PostgreSQL for
     multi-worker safety), updates last_polled_at for ALL fetched automations
     (to ensure fair batch rotation), filters to those that are due, and creates
-    PENDING runs. All within a single transaction so row locks are held throughout
-    and no schedules can be lost or duplicated.
+    PENDING runs only when the automation has no in-flight run. All within a single
+    transaction so row locks are held throughout and no schedules can be lost or
+    duplicated by concurrent scheduler workers.
 
     Note: SQLite deployments skip row locking (single-process mode assumed).
 
@@ -189,9 +218,23 @@ async def poll_and_schedule(
                 automation.last_polled_at = now
 
         due_automations = [a for a in automations if _is_automation_due_safely(a, now)]
+        in_flight_statuses = await _get_in_flight_run_statuses(session, due_automations)
 
         for automation in due_automations:
             try:
+                in_flight_status = in_flight_statuses.get(automation.id)
+                if in_flight_status is not None:
+                    logger.info(
+                        "Skipping cron run for active automation",
+                        extra={
+                            "automation_id": str(automation.id),
+                            "existing_run_status": in_flight_status.value,
+                            "trigger_source": "cron",
+                            "skip_reason": "automation_run_in_flight",
+                        },
+                    )
+                    continue
+
                 run = await create_pending_run(session, automation)
                 created_runs.append(run)
                 schedule_properties = {
@@ -243,8 +286,9 @@ async def scheduler_loop(
 ) -> None:
     """Main scheduler loop that polls for due automations.
 
-    For each due automation, creates a PENDING run in the automation_runs table.
-    The dispatcher (separate process) picks up PENDING runs and executes them.
+    For each due automation without a PENDING or RUNNING run, creates a PENDING run
+    in the automation_runs table. The dispatcher picks up PENDING runs and executes
+    them.
 
     Args:
         session_factory: SQLAlchemy async session factory