Circuit breaker flag for consecutively failing automations

Author: dylan-openhands

openhands/automation/config.py

@@ -451,6 +451,12 @@ class ServiceSettings(BaseSettings):
         AUTOMATION_WATCHDOG_INTERVAL_SECONDS: Watchdog poll interval (default: 60)
         AUTOMATION_FAILURE_DISABLE_THRESHOLD: Consecutive permanent failures before
             auto-disabling an automation (default: 3, <=0 disables auto-disable)
+        AUTOMATION_CONSECUTIVE_FAILURE_DISABLE_THRESHOLD: Consecutive failures
+            before auto-disabling. Unset (the default) turns the rule off;
+            setting a number turns it on and uses that number.
+        AUTOMATION_CONSECUTIVE_FAILURE_DISABLE_WINDOW_HOURS: The rule only trips
+            if nothing has succeeded in this many hours, which is what makes it
+            ignore provider outages shorter than the window (default: 24)
 
         # API pagination
         AUTOMATION_API_DEFAULT_PAGE_SIZE: Default page size (default: 50)
@@ -534,8 +540,20 @@ class ServiceSettings(BaseSettings):
     dispatcher_interval_seconds: int = 10
     dispatcher_batch_size: int = 10
     watchdog_interval_seconds: int = 60
+
+    # Circuit breaker. `failure_disable_threshold` is the fast path for
+    # unambiguous config faults (bad key, revoked token) and needs no time
+    # guard. The two rules below catch automations that merely fail forever;
+    # their span/window guards are what keep a provider outage from pausing
+    # healthy automations en masse, so both must exceed any tolerable outage.
     failure_disable_threshold: int = 3
 
+    # Setting a threshold at all is what turns the consecutive rule on; it is
+    # off by default. The window only applies once the rule is on, and must
+    # exceed any outage you would rather ride out than pause for.
+    consecutive_failure_disable_threshold: int | None = None
+    consecutive_failure_disable_window_hours: float = 24.0
+
     # How long an accepted event stays in `integration_events`. It bounds two
     # things: the dedupe window (a redelivery older than this is indistinguishable
     # from a new event) and the table, which otherwise grows with every delivery.

openhands/automation/dispatcher.py

@@ -236,7 +236,9 @@ async def _fail(
                 else None,
                 run_id=run.id,
             )
-        elif status_detail is not None:
+        else:
+            # Evaluate whether the automation has failed consecutively and may never
+            # succeed
             automation_disabled = await maybe_disable_unhealthy_automation_after_run(
                 session_factory,
                 automation.id,

openhands/automation/utils/unhealthy.py

@@ -1,13 +1,30 @@
-"""Classify unhealthy automations and auto-disable chronic permanent failures."""
+"""Circuit breaker that pauses automations which keep failing.
+
+Two rules, both evaluated from a single fetch of the automation's most recent
+terminal runs:
+
+1. Consecutive *permanent* failures (bad key, revoked token). Unambiguous, so
+   it trips fast and needs no time guard.
+2. The last N runs all failed and nothing has succeeded in the configured
+   window.
+
+Rule 2's window is what keeps a provider outage from pausing healthy
+automations: a run count alone cannot tell "broken forever" from "the provider
+was down", because a five-minute cron racks up hundreds of failures during any
+outage. Asking whether anything succeeded recently answers that directly, and
+is indifferent to how often the automation runs.
+"""
 
 from __future__ import annotations
 
 import logging
 import uuid
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import datetime, timedelta
 from typing import Any
 
-from sqlalchemy import CursorResult, select, update
+from sqlalchemy import CursorResult, func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from openhands.automation.config import get_config
@@ -19,7 +36,7 @@
     AutomationRunStatus,
 )
 from openhands.automation.utils.run import skip_pending_runs_for_disabled_automation
-from openhands.automation.utils.time import utcnow
+from openhands.automation.utils.time import ensure_utc, utcnow
 
 
 logger = logging.getLogger(__name__)
@@ -30,6 +47,9 @@
     AutomationRunStatus.FAILED,
 )
 
+SOURCE_PERMANENT = "consecutive_permanent_failures"
+SOURCE_CONSECUTIVE = "consecutive_failures"
+
 
 def _string_value(value: Any) -> str | None:
     if isinstance(value, str) and value:
@@ -66,71 +86,153 @@ def is_permanent_failure_detail(detail: Mapping[str, Any] | None) -> bool:
     return detail.get("user_action") == "settings"
 
 
-def _disabled_reason(detail: Mapping[str, Any], count: int) -> str:
-    formatted = _string_value(detail.get("formatted_detail"))
-    message = formatted or _string_value(detail.get("detail")) or "Permanent failure"
-    kind = _string_value(detail.get("kind")) or "permanent_failure"
-    return f"{kind}: {message} (seen in {count} consecutive runs)"
+def _humanize(hours: float) -> str:
+    if hours < 48:
+        return f"{round(hours)} hours"
+    return f"{round(hours / 24)} days"
+
+
+@dataclass(frozen=True)
+class _Trip:
+    """A breaker rule that fired, and everything needed to record it."""
 
+    reason: str
+    source: str
+    detail: dict[str, Any]
+    run_id: uuid.UUID | None
 
-async def get_consecutive_permanent_failure_count(
+
+async def _history_boundary(
+    session: AsyncSession, automation_id: uuid.UUID
+) -> datetime | None:
+    """Timestamp before which run history is ignored.
+
+    Re-enabling an automation leaves its old failures in scope, so without
+    this the very next failure trips the breaker again immediately. A
+    re-enable always follows a disable event, so the latest event is the
+    boundary.
+    """
+    return await session.scalar(
+        select(func.max(AutomationDisableEvent.created_at)).where(
+            AutomationDisableEvent.automation_id == automation_id
+        )
+    )
+
+
+async def _recent_terminal_runs(
     session: AsyncSession,
     automation_id: uuid.UUID,
-    *,
+    boundary: datetime | None,
     limit: int,
-) -> tuple[int, dict[str, Any] | None, uuid.UUID | None]:
-    """Count latest consecutive terminal runs with permanent failure details."""
+) -> Sequence[AutomationRun]:
+    """The automation's most recent terminal runs, newest first."""
+    stmt = select(AutomationRun).where(
+        AutomationRun.automation_id == automation_id,
+        AutomationRun.status.in_(TERMINAL_OUTCOME_STATUSES),
+    )
+    if boundary is not None:
+        stmt = stmt.where(AutomationRun.created_at > boundary)
     result = await session.execute(
-        select(AutomationRun)
+        stmt.order_by(AutomationRun.created_at.desc()).limit(limit)
+    )
+    return result.scalars().all()
+
+
+async def _succeeded_since(
+    session: AsyncSession,
+    automation_id: uuid.UUID,
+    since: datetime,
+) -> bool:
+    """Whether any run has completed since `since`. Stops at the first hit."""
+    found = await session.scalar(
+        select(AutomationRun.id)
         .where(
             AutomationRun.automation_id == automation_id,
-            AutomationRun.status.in_(TERMINAL_OUTCOME_STATUSES),
+            AutomationRun.status == AutomationRunStatus.COMPLETED,
+            AutomationRun.created_at > since,
         )
-        .order_by(AutomationRun.created_at.desc())
-        .limit(limit)
+        .limit(1)
     )
+    return found is not None
 
+
+def _leading_failures(runs: Sequence[AutomationRun]) -> int:
     count = 0
-    latest_detail: dict[str, Any] | None = None
-    latest_run_id: uuid.UUID | None = None
-    for run in result.scalars().all():
-        detail = run.status_detail
-        if not is_permanent_failure_detail(detail):
+    for run in runs:
+        if run.status != AutomationRunStatus.FAILED:
             break
         count += 1
-        if latest_detail is None:
-            latest_detail = detail
-            latest_run_id = run.id
-    return count, latest_detail, latest_run_id
+    return count
 
 
-async def maybe_disable_unhealthy_automation(
-    session: AsyncSession,
-    automation_id: uuid.UUID,
-    *,
-    threshold: int | None = None,
-) -> bool:
-    """Disable an automation once permanent failures reach the threshold."""
-    if threshold is None:
-        threshold = get_config().service.failure_disable_threshold
-    if threshold <= 0:
-        return False
+def _check_permanent(
+    runs: Sequence[AutomationRun],
+    threshold: int,
+) -> _Trip | None:
+    """Trip on consecutive terminal runs that all carry permanent-fault detail."""
+    if threshold <= 0 or len(runs) < threshold:
+        return None
 
-    count, latest_detail, latest_run_id = await get_consecutive_permanent_failure_count(
-        session,
-        automation_id,
-        limit=threshold,
+    count = 0
+    for run in runs:
+        if not is_permanent_failure_detail(run.status_detail):
+            break
+        count += 1
+    if count < threshold:
+        return None
+
+    detail = runs[0].status_detail or {}
+    formatted = _string_value(detail.get("formatted_detail"))
+    message = formatted or _string_value(detail.get("detail")) or "unknown error"
+    kind = _string_value(detail.get("kind")) or "permanent_failure"
+    return _Trip(
+        reason=(
+            f"Paused automatically: {kind} — {message}. "
+            f"This failed the last {count} runs and needs a configuration fix."
+        ),
+        source=SOURCE_PERMANENT,
+        detail={
+            "rule": SOURCE_PERMANENT,
+            "threshold": threshold,
+            "consecutive_permanent_failures": count,
+            "status_detail": detail,
+        },
+        run_id=runs[0].id,
     )
-    if count < threshold or latest_detail is None:
-        return False
 
-    disabled_reason = _disabled_reason(latest_detail, count)
+
+def _consecutive_trip(
+    runs: Sequence[AutomationRun],
+    threshold: int,
+    window_hours: float,
+) -> _Trip:
+    count = _leading_failures(runs)
+    return _Trip(
+        reason=(
+            f"Paused automatically: the last {count} runs all failed and nothing "
+            f"has succeeded in {_humanize(window_hours)}."
+        ),
+        source=SOURCE_CONSECUTIVE,
+        detail={
+            "rule": SOURCE_CONSECUTIVE,
+            "threshold": threshold,
+            "consecutive_failures": count,
+            "window_hours": window_hours,
+        },
+        run_id=runs[0].id,
+    )
+
+
+async def _apply_disable(
+    session: AsyncSession,
+    automation_id: uuid.UUID,
+    trip: _Trip,
+) -> bool:
     disabled_detail = {
-        "reason": disabled_reason,
-        "threshold": threshold,
-        "consecutive_permanent_failures": count,
-        "run_id": str(latest_run_id) if latest_run_id else None,
-        "status_detail": latest_detail,
+        "reason": trip.reason,
+        "source": trip.source,
+        "run_id": str(trip.run_id) if trip.run_id else None,
+        **trip.detail,
     }
     disabled_at = utcnow()
     result: CursorResult = await session.execute(  # type: ignore[assignment]
@@ -141,7 +243,7 @@ async def maybe_disable_unhealthy_automation(
         )
         .values(
             enabled=False,
-            disabled_reason=disabled_reason,
+            disabled_reason=trip.reason,
             disabled_detail=disabled_detail,
             disabled_at=disabled_at,
         )
@@ -152,7 +254,7 @@ async def maybe_disable_unhealthy_automation(
     await skip_pending_runs_for_disabled_automation(
         session,
         automation_id,
-        reason=disabled_reason,
+        reason=trip.reason,
         disabled_detail=disabled_detail,
         completed_at=disabled_at,
     )
@@ -164,21 +266,67 @@ async def maybe_disable_unhealthy_automation(
     session.add(
         AutomationDisableEvent(
             automation_id=automation_id,
-            run_id=latest_run_id,
-            reason=disabled_reason,
+            run_id=trip.run_id,
+            reason=trip.reason,
             detail=disabled_detail,
-            source="consecutive_permanent_failures",
+            source=trip.source,
         )
     )
 
     logger.warning(
-        "Automation disabled after %s consecutive permanent failures",
-        count,
-        extra={"automation_id": str(automation_id), "run_id": str(latest_run_id)},
+        "Automation auto-disabled (%s): %s",
+        trip.source,
+        trip.reason,
+        extra={
+            "automation_id": str(automation_id),
+            "run_id": str(trip.run_id) if trip.run_id else None,
+        },
     )
     return True
 
 
+async def maybe_disable_unhealthy_automation(
+    session: AsyncSession,
+    automation_id: uuid.UUID,
+    *,
+    threshold: int | None = None,
+) -> bool:
+    """Disable an automation if either circuit breaker rule has tripped.
+
+    Args:
+        threshold: Override for the permanent-failure rule only; the
+            consecutive rule always reads its config values.
+    """
+    service = get_config().service
+    if threshold is None:
+        threshold = service.failure_disable_threshold
+    threshold = max(threshold, 0)
+    consecutive = max(service.consecutive_failure_disable_threshold or 0, 0)
+    window_hours = service.consecutive_failure_disable_window_hours
+
+    limit = max(threshold, consecutive)
+    if limit <= 0:
+        return False
+
+    boundary = await _history_boundary(session, automation_id)
+    runs = await _recent_terminal_runs(session, automation_id, boundary, limit)
+
+    trip = _check_permanent(runs, threshold)
+
+    # The all-failed check is in memory, so the success lookup only runs for
+    # automations that are already failing straight through.
+    if trip is None and consecutive > 0 and _leading_failures(runs) >= consecutive:
+        since = utcnow() - timedelta(hours=window_hours)
+        if boundary is not None:
+            since = max(since, ensure_utc(boundary))
+        if not await _succeeded_since(session, automation_id, since):
+            trip = _consecutive_trip(runs, consecutive, window_hours)
+
+    if trip is None:
+        return False
+    return await _apply_disable(session, automation_id, trip)
+
+
 async def maybe_disable_unhealthy_automation_after_run(
     session_factory: async_sessionmaker[AsyncSession],
     automation_id: uuid.UUID,