diff --git a/migrations/versions/020_add_automation_disabled_reason.py b/migrations/versions/020_add_automation_disabled_reason.py
new file mode 100644
index 0000000..2c007c7
--- /dev/null
+++ b/migrations/versions/020_add_automation_disabled_reason.py
@@ -0,0 +1,102 @@
+"""Add automation disabled reason metadata.
+
+Revision ID: 020
+Revises: 019
+Create Date: 2026-08-20
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision: str = "020"
+down_revision: str = "019"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _is_sqlite() -> bool:
+    return op.get_bind().dialect.name == "sqlite"
+
+
+def upgrade() -> None:
+    op.add_column("automations", sa.Column("disabled_reason", sa.Text(), nullable=True))
+    op.add_column("automations", sa.Column("disabled_detail", sa.JSON(), nullable=True))
+    op.add_column(
+        "automations",
+        sa.Column("disabled_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.create_table(
+        "automation_disable_events",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("automation_id", sa.Uuid(), nullable=False),
+        sa.Column("run_id", sa.Uuid(), nullable=True),
+        sa.Column("reason", sa.Text(), nullable=False),
+        sa.Column("detail", sa.JSON(), nullable=True),
+        sa.Column("source", sa.String(length=64), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+            nullable=False,
+        ),
+        sa.ForeignKeyConstraint(
+            ["automation_id"], ["automations.id"], ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(
+            ["run_id"], ["automation_runs.id"], ondelete="SET NULL"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        "ix_automation_disable_events_automation_id",
+        "automation_disable_events",
+        ["automation_id"],
+    )
+    op.create_index(
+        "ix_automation_disable_events_run_id",
+        "automation_disable_events",
+        ["run_id"],
+    )
+    op.create_index(
+        "ix_automation_disable_events_created_at",
+        "automation_disable_events",
+        ["created_at"],
+    )
+
+    if _is_sqlite():
+        return
+
+    op.execute(
+        "COMMENT ON COLUMN automations.disabled_reason IS "
+        "'Human-readable reason an automation is currently disabled.'"
+    )
+    op.execute(
+        "COMMENT ON COLUMN automations.disabled_detail IS "
+        "'Structured metadata for current automation disabled state.'"
+    )
+    op.execute(
+        "COMMENT ON TABLE automation_disable_events IS "
+        "'Historical records for automation auto-disable decisions.'"
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_automation_disable_events_created_at",
+        table_name="automation_disable_events",
+    )
+    op.drop_index(
+        "ix_automation_disable_events_run_id",
+        table_name="automation_disable_events",
+    )
+    op.drop_index(
+        "ix_automation_disable_events_automation_id",
+        table_name="automation_disable_events",
+    )
+    op.drop_table("automation_disable_events")
+    op.drop_column("automations", "disabled_at")
+    op.drop_column("automations", "disabled_detail")
+    op.drop_column("automations", "disabled_reason")
diff --git a/openhands/automation/config.py b/openhands/automation/config.py
index a306ea0..c589a54 100644
--- a/openhands/automation/config.py
+++ b/openhands/automation/config.py
@@ -449,6 +449,8 @@ class ServiceSettings(BaseSettings):
         AUTOMATION_DISPATCHER_INTERVAL_SECONDS: Dispatcher poll interval (default: 10)
         AUTOMATION_DISPATCHER_BATCH_SIZE: Dispatcher batch size (default: 10)
         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)
 
         # API pagination
         AUTOMATION_API_DEFAULT_PAGE_SIZE: Default page size (default: 50)
@@ -532,6 +534,7 @@ class ServiceSettings(BaseSettings):
     dispatcher_interval_seconds: int = 10
     dispatcher_batch_size: int = 10
     watchdog_interval_seconds: int = 60
+    failure_disable_threshold: int = 3
 
     # How long an accepted event stays in `integration_events`. It bounds two
     # things: the dedupe window (a redelivery older than this is indistinguishable
diff --git a/openhands/automation/dispatcher.py b/openhands/automation/dispatcher.py
index bf800a1..52d8f35 100644
--- a/openhands/automation/dispatcher.py
+++ b/openhands/automation/dispatcher.py
@@ -65,6 +65,9 @@
 )
 from openhands.automation.utils.time import utcnow
 from openhands.automation.utils.timeout import resolve_automation_timeout_seconds
+from openhands.automation.utils.unhealthy import (
+    maybe_disable_unhealthy_automation_after_run,
+)
 
 
 logger = logging.getLogger("automation.dispatcher")
@@ -127,8 +130,13 @@ async def _poll_pending_runs(
     """
     select_query = (
         select(AutomationRun)
+        .join(AutomationRun.automation)
         .options(selectinload(AutomationRun.automation))
-        .where(AutomationRun.status == AutomationRunStatus.PENDING)
+        .where(
+            AutomationRun.status == AutomationRunStatus.PENDING,
+            Automation.enabled.is_(True),
+            Automation.deleted_at.is_(None),
+        )
         .order_by(AutomationRun.created_at.asc())
         .limit(batch_size)
     )
@@ -216,6 +224,22 @@ async def _fail(
             error,
             status_detail=status_detail,
         )
+        automation_disabled = disable
+        if disable:
+            automation_disabled = await disable_automation(
+                session_factory,
+                automation.id,
+                error,
+                disabled_detail={"status_detail": status_detail}
+                if status_detail is not None
+                else None,
+                run_id=run.id,
+            )
+        elif status_detail is not None:
+            automation_disabled = await maybe_disable_unhealthy_automation_after_run(
+                session_factory,
+                automation.id,
+            )
         await capture_automation_event(
             "automation_run_failed",
             automation=automation,
@@ -224,11 +248,9 @@ async def _fail(
             properties={
                 "trigger_source": "dispatcher",
                 "failure_kind": "dispatch_error",
-                "automation_disabled": disable,
+                "automation_disabled": automation_disabled,
             },
         )
-        if disable:
-            await disable_automation(session_factory, automation.id, error)
 
     # 1. Calculate effective timeout (doesn't depend on ctx). This same value
     # drives both the bash command timeout and the watchdog cleanup deadline.
diff --git a/openhands/automation/models.py b/openhands/automation/models.py
index 816b12b..d11f722 100644
--- a/openhands/automation/models.py
+++ b/openhands/automation/models.py
@@ -97,6 +97,13 @@ class Automation(Base):
     # Whether the automation is enabled (can be triggered)
     enabled: Mapped[bool] = mapped_column(default=True, nullable=False, index=True)
 
+    # Current disabled-state metadata. AutomationDisableEvent keeps history.
+    disabled_reason: Mapped[str | None] = mapped_column(Text, nullable=True)
+    disabled_detail: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
+    disabled_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+
     # Soft delete timestamp (NULL = not deleted)
     deleted_at: Mapped[datetime | None] = mapped_column(
         DateTime(timezone=True), nullable=True, index=True
@@ -128,6 +135,11 @@ class Automation(Base):
     runs: Mapped[list["AutomationRun"]] = relationship(
         "AutomationRun", back_populates="automation", cascade="all, delete-orphan"
     )
+    disable_events: Mapped[list["AutomationDisableEvent"]] = relationship(
+        "AutomationDisableEvent",
+        back_populates="automation",
+        cascade="all, delete-orphan",
+    )
 
 
 class AutomationRun(Base):
@@ -206,6 +218,7 @@ class AutomationRun(Base):
         server_default=text("CURRENT_TIMESTAMP"),
         nullable=False,
     )
+
     started_at: Mapped[datetime | None] = mapped_column(
         DateTime(timezone=True), nullable=True
     )
@@ -230,6 +243,41 @@ class AutomationRun(Base):
     )
 
 
+class AutomationDisableEvent(Base):
+    """Historical record of an automation being disabled."""
+
+    __tablename__ = "automation_disable_events"
+
+    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
+    automation_id: Mapped[uuid.UUID] = mapped_column(
+        Uuid,
+        ForeignKey("automations.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    run_id: Mapped[uuid.UUID | None] = mapped_column(
+        Uuid,
+        ForeignKey("automation_runs.id", ondelete="SET NULL"),
+        nullable=True,
+        index=True,
+    )
+    reason: Mapped[str] = mapped_column(Text, nullable=False)
+    detail: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
+    source: Mapped[str] = mapped_column(String(64), nullable=False)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=text("CURRENT_TIMESTAMP"),
+        nullable=False,
+        index=True,
+    )
+
+    automation: Mapped["Automation"] = relationship(
+        "Automation",
+        back_populates="disable_events",
+    )
+    run: Mapped["AutomationRun | None"] = relationship("AutomationRun")
+
+
 class TarballUpload(Base):
     """A tarball upload for automation code.
 
diff --git a/openhands/automation/router.py b/openhands/automation/router.py
index 3a8b49b..044fea7 100644
--- a/openhands/automation/router.py
+++ b/openhands/automation/router.py
@@ -27,6 +27,7 @@
 from openhands.automation.git_sync import mark_git_sync_dirty
 from openhands.automation.models import (
     Automation,
+    AutomationDisableEvent,
     AutomationRun,
     AutomationRunStatus,
     TarballUpload,
@@ -56,7 +57,11 @@
     fetch_latest_finish_tool_response_for_run,
 )
 from openhands.automation.utils.model_profiles import resolve_model_profile_for_user
-from openhands.automation.utils.run import create_pending_run, record_first_run_outcome
+from openhands.automation.utils.run import (
+    create_pending_run,
+    record_first_run_outcome,
+    skip_pending_runs_for_disabled_automation,
+)
 from openhands.automation.utils.run_status_detail import (
     run_status_detail_from_callback_error,
 )
@@ -71,6 +76,7 @@
     find_existing_template_automation,
 )
 from openhands.automation.utils.timeout import default_automation_timeout
+from openhands.automation.utils.unhealthy import maybe_disable_unhealthy_automation
 
 
 logger = logging.getLogger(__name__)
@@ -223,6 +229,27 @@ async def update_automation(
     if body.trigger is not None:
         update_data["trigger"] = body.trigger.model_dump()
 
+    disable_event: AutomationDisableEvent | None = None
+    skip_pending_reason: str | None = None
+    if update_data.get("enabled") is True:
+        update_data["disabled_reason"] = None
+        update_data["disabled_detail"] = None
+        update_data["disabled_at"] = None
+    elif update_data.get("enabled") is False:
+        if auto.enabled:
+            skip_pending_reason = "Automation disabled by user"
+            disabled_at = utcnow()
+            disabled_detail = {"reason": "manual", "source": "user"}
+            update_data["disabled_reason"] = "manual"
+            update_data["disabled_detail"] = disabled_detail
+            update_data["disabled_at"] = disabled_at
+            disable_event = AutomationDisableEvent(
+                automation_id=auto.id,
+                reason="manual",
+                detail=disabled_detail,
+                source="manual",
+            )
+
     if "model" in update_data:
         update_data["model"] = resolve_model_profile_for_user(body.model, user)
 
@@ -250,6 +277,17 @@ async def update_automation(
         if auto.preset_metadata is not None:
             auto.preset_metadata = {**auto.preset_metadata, "prompt": auto.prompt}
 
+    if skip_pending_reason is not None:
+        await skip_pending_runs_for_disabled_automation(
+            session,
+            auto.id,
+            reason=skip_pending_reason,
+            disabled_detail=auto.disabled_detail,
+            completed_at=auto.disabled_at,
+        )
+    if disable_event is not None:
+        session.add(disable_event)
+
     # Note: updated_at is handled automatically by the model's onupdate=utcnow
     await session.flush()
     await session.refresh(auto)
@@ -273,8 +311,30 @@ async def delete_automation(
 ) -> None:
     """Soft delete an automation."""
     auto = await _get_user_automation(session, automation_id, user.user_id, user.org_id)
+    was_enabled = auto.enabled
     auto.enabled = False
-    auto.deleted_at = utcnow()
+    deleted_at = utcnow()
+    auto.deleted_at = deleted_at
+    if was_enabled:
+        disabled_detail = {"reason": "manual_delete", "source": "user"}
+        auto.disabled_reason = "manual_delete"
+        auto.disabled_detail = disabled_detail
+        auto.disabled_at = deleted_at
+        session.add(
+            AutomationDisableEvent(
+                automation_id=auto.id,
+                reason="manual_delete",
+                detail=disabled_detail,
+                source="manual_delete",
+            )
+        )
+    await skip_pending_runs_for_disabled_automation(
+        session,
+        auto.id,
+        reason="Automation deleted by user",
+        disabled_detail=auto.disabled_detail,
+        completed_at=deleted_at,
+    )
     await session.flush()
     await mark_git_sync_dirty(session, auto)
     await capture_automation_event(
@@ -368,6 +428,16 @@ async def dispatch_automation(
     picked up by the dispatcher and executed.
     """
     auto = await _get_user_automation(session, automation_id, user.user_id, user.org_id)
+    if not auto.enabled:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            detail={
+                "message": "Automation is disabled",
+                "disabled_reason": auto.disabled_reason,
+                "disabled_detail": auto.disabled_detail,
+            },
+        )
+
     run = await create_pending_run(
         session,
         auto,
@@ -498,6 +568,8 @@ async def complete_run(
             previous=run.status_detail,
         )
     elif body.status == "COMPLETED":
+        # Task outcomes and blocking factors are agent/user-level result metadata;
+        # only SDK callback errors and system dispatch errors feed auto-disablement.
         values["status_detail"] = None
     if body.conversation_id:
         finish_tool_response = await fetch_latest_finish_tool_response_for_run(
@@ -552,12 +624,22 @@ async def complete_run(
                 status.HTTP_409_CONFLICT,
                 detail=f"Run is {run.status.value}, expected RUNNING",
             )
+    if new_status == AutomationRunStatus.FAILED:
+        automation_disabled = await maybe_disable_unhealthy_automation(
+            session,
+            automation.id,
+        )
+    else:
+        automation_disabled = False
 
     await session.refresh(run)
     logger.info("Run %s → %s", run_id, new_status.value)
     telemetry_properties: dict = {"trigger_source": "callback"}
     if reconciled:
         telemetry_properties["reconciled_watchdog_timeout"] = True
+    if automation_disabled:
+        telemetry_properties["automation_disabled"] = True
+
     await capture_automation_event(
         "automation_run_completed"
         if new_status == AutomationRunStatus.COMPLETED
diff --git a/openhands/automation/schemas.py b/openhands/automation/schemas.py
index b7b9811..fadaa6b 100644
--- a/openhands/automation/schemas.py
+++ b/openhands/automation/schemas.py
@@ -735,6 +735,9 @@ class AutomationResponse(BaseModel):
     timeout: int | None
     keep_alive: bool | None
     enabled: bool
+    disabled_reason: str | None = None
+    disabled_detail: dict[str, Any] | None = None
+    disabled_at: UtcDatetime | None = None
     last_triggered_at: UtcDatetime | None
     created_at: UtcDatetime
     updated_at: UtcDatetime
@@ -760,6 +763,8 @@ class RunCompleteRequest(BaseModel):
     conversation_id: str | None = None
     error: str | ConversationErrorEvent | dict[str, Any] | None = None
     cost: float | None = None
+    blocking_factor: dict[str, Any] | None = None
+    task_outcome: dict[str, Any] | None = None
 
     @field_validator("error", mode="before")
     @classmethod
diff --git a/openhands/automation/utils/run.py b/openhands/automation/utils/run.py
index dd35a3c..f8ef5fe 100644
--- a/openhands/automation/utils/run.py
+++ b/openhands/automation/utils/run.py
@@ -8,7 +8,13 @@
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from openhands.automation.db import using_sqlite
-from openhands.automation.models import Automation, AutomationRun, AutomationRunStatus
+from openhands.automation.git_sync import mark_git_sync_dirty
+from openhands.automation.models import (
+    Automation,
+    AutomationDisableEvent,
+    AutomationRun,
+    AutomationRunStatus,
+)
 from openhands.automation.telemetry import capture_automation_event
 from openhands.automation.utils.time import utcnow
 from openhands.automation.utils.timeout import resolve_automation_timeout_seconds
@@ -28,6 +34,10 @@ async def disable_automation(
     session_factory: async_sessionmaker[AsyncSession],
     automation_id: uuid.UUID,
     reason: str,
+    *,
+    disabled_detail: dict | None = None,
+    run_id: uuid.UUID | None = None,
+    source: str = "permanent_dispatch_failure",
 ) -> bool:
     """Disable an automation due to a permanent configuration error.
 
@@ -50,6 +60,7 @@ async def disable_automation(
 
     try:
         async with session_factory() as session:
+            disabled_at = utcnow()
             # Use optimistic locking: only update if currently enabled
             result: CursorResult = await session.execute(  # type: ignore[assignment]
                 update(Automation)
@@ -57,7 +68,12 @@ async def disable_automation(
                     Automation.id == automation_id,
                     Automation.enabled == True,  # noqa: E712
                 )
-                .values(enabled=False)
+                .values(
+                    enabled=False,
+                    disabled_reason=reason,
+                    disabled_detail=disabled_detail,
+                    disabled_at=disabled_at,
+                )
             )
 
             if result.rowcount == 0:
@@ -71,6 +87,27 @@ async def disable_automation(
                     logger.info("Automation already disabled", extra=extra)
                 return False
 
+            await skip_pending_runs_for_disabled_automation(
+                session,
+                automation_id,
+                reason=reason,
+                disabled_detail=disabled_detail,
+                completed_at=disabled_at,
+            )
+
+            automation = await session.get(Automation, automation_id)
+            if automation is not None:
+                await mark_git_sync_dirty(session, automation)
+
+            session.add(
+                AutomationDisableEvent(
+                    automation_id=automation_id,
+                    run_id=run_id,
+                    reason=reason,
+                    detail=disabled_detail,
+                    source=source,
+                )
+            )
             await session.commit()
 
             logger.warning(
@@ -85,6 +122,44 @@ async def disable_automation(
         return False
 
 
+async def skip_pending_runs_for_disabled_automation(
+    session: AsyncSession,
+    automation_id: uuid.UUID,
+    *,
+    reason: str,
+    disabled_detail: dict | None = None,
+    completed_at: datetime | None = None,
+) -> int:
+    """Mark accepted-but-not-dispatched runs terminal when automation is disabled."""
+    completed_at = completed_at or utcnow()
+    status_detail: dict = {
+        "phase": "dispatch",
+        "kind": "blocked",
+        "detail": reason,
+        "transient": False,
+        "source": "automation_service",
+        "operation": "automation_disabled",
+        "user_action": "settings",
+    }
+    if disabled_detail is not None:
+        status_detail["disabled_detail"] = disabled_detail
+
+    result: CursorResult = await session.execute(  # type: ignore[assignment]
+        update(AutomationRun)
+        .where(
+            AutomationRun.automation_id == automation_id,
+            AutomationRun.status == AutomationRunStatus.PENDING,
+        )
+        .values(
+            status=AutomationRunStatus.SKIPPED,
+            completed_at=completed_at,
+            error_detail="Automation disabled",
+            status_detail=status_detail,
+        )
+    )
+    return result.rowcount or 0
+
+
 async def create_pending_run(
     session: AsyncSession,
     automation: Automation,
diff --git a/openhands/automation/utils/run_status_detail.py b/openhands/automation/utils/run_status_detail.py
index 1b0a8a9..8dc2caf 100644
--- a/openhands/automation/utils/run_status_detail.py
+++ b/openhands/automation/utils/run_status_detail.py
@@ -31,6 +31,7 @@ class RunStatusDetailKind(StrEnum):
     ENVIRONMENT_UNAVAILABLE = "environment_unavailable"
     EXECUTION_ERROR = "execution_error"
     CONCURRENCY_LIMIT = "concurrency_limit"
+    BLOCKED = "blocked"
     UNKNOWN = "unknown"
 
 
diff --git a/openhands/automation/utils/unhealthy.py b/openhands/automation/utils/unhealthy.py
new file mode 100644
index 0000000..5332448
--- /dev/null
+++ b/openhands/automation/utils/unhealthy.py
@@ -0,0 +1,204 @@
+"""Classify unhealthy automations and auto-disable chronic permanent failures."""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from collections.abc import Mapping
+from typing import Any
+
+from sqlalchemy import CursorResult, select, update
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from openhands.automation.config import get_config
+from openhands.automation.git_sync import mark_git_sync_dirty
+from openhands.automation.models import (
+    Automation,
+    AutomationDisableEvent,
+    AutomationRun,
+    AutomationRunStatus,
+)
+from openhands.automation.utils.run import skip_pending_runs_for_disabled_automation
+from openhands.automation.utils.time import utcnow
+
+
+logger = logging.getLogger(__name__)
+
+PERMANENT_FAILURE_KINDS = frozenset({"auth", "config", "quota", "blocked"})
+TERMINAL_OUTCOME_STATUSES = (
+    AutomationRunStatus.COMPLETED,
+    AutomationRunStatus.FAILED,
+)
+
+
+def _string_value(value: Any) -> str | None:
+    if isinstance(value, str) and value:
+        return value
+    return None
+
+
+def is_permanent_failure_detail(detail: Mapping[str, Any] | None) -> bool:
+    """Return whether status_detail describes a non-transient config fault."""
+    if not detail:
+        return False
+    if detail.get("transient") is True:
+        return False
+    if detail.get("permanent") is True:
+        return True
+
+    kind = _string_value(detail.get("kind"))
+    if kind and kind.casefold() in PERMANENT_FAILURE_KINDS:
+        return True
+
+    classification = detail.get("classification")
+    if isinstance(classification, Mapping):
+        if classification.get("retryable") is True:
+            return False
+        classification_kind = _string_value(classification.get("kind"))
+        if (
+            classification_kind
+            and classification_kind.casefold() in PERMANENT_FAILURE_KINDS
+        ):
+            return True
+        if classification.get("user_action") == "settings":
+            return True
+
+    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)"
+
+
+async def get_consecutive_permanent_failure_count(
+    session: AsyncSession,
+    automation_id: uuid.UUID,
+    *,
+    limit: int,
+) -> tuple[int, dict[str, Any] | None, uuid.UUID | None]:
+    """Count latest consecutive terminal runs with permanent failure details."""
+    result = await session.execute(
+        select(AutomationRun)
+        .where(
+            AutomationRun.automation_id == automation_id,
+            AutomationRun.status.in_(TERMINAL_OUTCOME_STATUSES),
+        )
+        .order_by(AutomationRun.created_at.desc())
+        .limit(limit)
+    )
+
+    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):
+            break
+        count += 1
+        if latest_detail is None:
+            latest_detail = detail
+            latest_run_id = run.id
+    return count, latest_detail, latest_run_id
+
+
+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
+
+    count, latest_detail, latest_run_id = await get_consecutive_permanent_failure_count(
+        session,
+        automation_id,
+        limit=threshold,
+    )
+    if count < threshold or latest_detail is None:
+        return False
+
+    disabled_reason = _disabled_reason(latest_detail, count)
+    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,
+    }
+    disabled_at = utcnow()
+    result: CursorResult = await session.execute(  # type: ignore[assignment]
+        update(Automation)
+        .where(
+            Automation.id == automation_id,
+            Automation.enabled == True,  # noqa: E712
+        )
+        .values(
+            enabled=False,
+            disabled_reason=disabled_reason,
+            disabled_detail=disabled_detail,
+            disabled_at=disabled_at,
+        )
+    )
+    if result.rowcount == 0:
+        return False
+
+    await skip_pending_runs_for_disabled_automation(
+        session,
+        automation_id,
+        reason=disabled_reason,
+        disabled_detail=disabled_detail,
+        completed_at=disabled_at,
+    )
+
+    automation = await session.get(Automation, automation_id)
+    if automation is not None:
+        await mark_git_sync_dirty(session, automation)
+
+    session.add(
+        AutomationDisableEvent(
+            automation_id=automation_id,
+            run_id=latest_run_id,
+            reason=disabled_reason,
+            detail=disabled_detail,
+            source="consecutive_permanent_failures",
+        )
+    )
+
+    logger.warning(
+        "Automation disabled after %s consecutive permanent failures",
+        count,
+        extra={"automation_id": str(automation_id), "run_id": str(latest_run_id)},
+    )
+    return True
+
+
+async def maybe_disable_unhealthy_automation_after_run(
+    session_factory: async_sessionmaker[AsyncSession],
+    automation_id: uuid.UUID,
+    *,
+    threshold: int | None = None,
+) -> bool:
+    """Open a short transaction and maybe auto-disable an automation."""
+    try:
+        async with session_factory() as session:
+            disabled = await maybe_disable_unhealthy_automation(
+                session,
+                automation_id,
+                threshold=threshold,
+            )
+            if disabled:
+                await session.commit()
+            return disabled
+    except Exception:
+        logger.exception(
+            "Failed to evaluate automation unhealthy state",
+            extra={"automation_id": str(automation_id)},
+        )
+        return False
diff --git a/tests/test_config.py b/tests/test_config.py
index 9735c4a..5ac5fac 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -6,6 +6,7 @@
     HttpSettings,
     LogSettings,
     SandboxSettings,
+    ServiceSettings,
     Settings,
     clear_config_cache,
     get_config,
@@ -57,6 +58,18 @@ def test_resolve_caps_stored_timeout_to_configured_max(self):
         assert resolve_automation_timeout_seconds(max_duration + 600) == max_duration
 
 
+class TestServiceSettings:
+    """Tests for service-level configuration."""
+
+    def test_failure_disable_threshold_uses_documented_env_var(self, monkeypatch):
+        monkeypatch.setenv("AUTOMATION_FAILURE_DISABLE_THRESHOLD", "0")
+        monkeypatch.setenv("AUTOMATION_AUTOMATION_FAILURE_DISABLE_THRESHOLD", "7")
+
+        settings = ServiceSettings()
+
+        assert settings.failure_disable_threshold == 0
+
+
 class TestBasePath:
     """Verify base_path is derived from base_url path + /api/automation."""
 
diff --git a/tests/test_dispatcher.py b/tests/test_dispatcher.py
index 6734591..11b15e5 100644
--- a/tests/test_dispatcher.py
+++ b/tests/test_dispatcher.py
@@ -465,6 +465,39 @@ async def test_ignores_completed_runs(
 
         assert len(dispatched) == 0
 
+    @patch("openhands.automation.dispatcher._execute_run_safe", new_callable=AsyncMock)
+    async def test_ignores_pending_runs_for_disabled_automations(
+        self, mock_execute, async_session_factory, mock_settings, mock_client
+    ):
+        """Pending runs are not dispatched once their automation is disabled."""
+        async with async_session_factory() as session:
+            automation = Automation(
+                user_id=TEST_USER_ID,
+                org_id=TEST_ORG_ID,
+                name="Test",
+                trigger={"type": "cron", "schedule": "* * * * *", "timezone": "UTC"},
+                tarball_path="s3://bucket/code.tar.gz",
+                entrypoint="uv run main.py",
+                enabled=False,
+                disabled_reason="auth: Invalid API key",
+            )
+            session.add(automation)
+            await session.commit()
+
+            run = AutomationRun(
+                automation_id=automation.id,
+                status=AutomationRunStatus.PENDING,
+            )
+            session.add(run)
+            await session.commit()
+
+        dispatched = await dispatch_pending_runs(
+            async_session_factory, mock_settings, mock_client
+        )
+
+        assert dispatched == []
+        mock_execute.assert_not_awaited()
+
     @patch("openhands.automation.dispatcher._execute_run_safe", new_callable=AsyncMock)
     async def test_respects_batch_size(
         self, mock_execute, async_session_factory, mock_settings, mock_client
diff --git a/tests/test_router.py b/tests/test_router.py
index 9501c41..c9b9969 100644
--- a/tests/test_router.py
+++ b/tests/test_router.py
@@ -6,9 +6,11 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from sqlalchemy import select
 
 from openhands.automation.models import (
     Automation,
+    AutomationDisableEvent,
     AutomationRun,
     TarballUpload,
     UploadStatus,
@@ -918,6 +920,8 @@ class TestDeleteAutomation:
 
     async def test_delete_automation_soft_deletes(self, async_client, async_session):
         """DELETE sets enabled=False and deleted_at."""
+        from openhands.automation.models import AutomationRunStatus
+
         automation = Automation(
             user_id=TEST_USER_ID,
             org_id=TEST_ORG_ID,
@@ -928,6 +932,12 @@ async def test_delete_automation_soft_deletes(self, async_client, async_session)
             enabled=True,
         )
         async_session.add(automation)
+        await async_session.flush()
+        pending_run = AutomationRun(
+            automation_id=automation.id,
+            status=AutomationRunStatus.PENDING,
+        )
+        async_session.add(pending_run)
         await async_session.commit()
         automation_id = automation.id
 
@@ -937,8 +947,39 @@ async def test_delete_automation_soft_deletes(self, async_client, async_session)
 
         # Refresh from DB
         await async_session.refresh(automation)
+        await async_session.refresh(pending_run)
         assert automation.enabled is False
         assert automation.deleted_at is not None
+        assert automation.disabled_reason == "manual_delete"
+        assert automation.disabled_detail == {
+            "reason": "manual_delete",
+            "source": "user",
+        }
+        assert automation.disabled_at == automation.deleted_at
+        assert pending_run.status == AutomationRunStatus.SKIPPED
+        assert pending_run.completed_at == automation.deleted_at
+        assert pending_run.status_detail is not None
+        assert pending_run.status_detail["detail"] == "Automation deleted by user"
+        assert pending_run.status_detail["disabled_detail"] == {
+            "reason": "manual_delete",
+            "source": "user",
+        }
+
+        events = (
+            (
+                await async_session.execute(
+                    select(AutomationDisableEvent).where(
+                        AutomationDisableEvent.automation_id == automation.id
+                    )
+                )
+            )
+            .scalars()
+            .all()
+        )
+        assert len(events) == 1
+        assert events[0].reason == "manual_delete"
+        assert events[0].detail == {"reason": "manual_delete", "source": "user"}
+        assert events[0].source == "manual_delete"
 
     async def test_delete_automation_not_found(self, async_client):
         """DELETE on non-existent ID returns 404."""
@@ -1035,6 +1076,8 @@ async def test_update_automation_schedule(self, async_client, async_session):
 
     async def test_update_automation_disable(self, async_client, async_session):
         """PATCH can disable an automation."""
+        from openhands.automation.models import AutomationRunStatus
+
         automation = Automation(
             user_id=TEST_USER_ID,
             org_id=TEST_ORG_ID,
@@ -1045,6 +1088,12 @@ async def test_update_automation_disable(self, async_client, async_session):
             enabled=True,
         )
         async_session.add(automation)
+        await async_session.flush()
+        pending_run = AutomationRun(
+            automation_id=automation.id,
+            status=AutomationRunStatus.PENDING,
+        )
+        async_session.add(pending_run)
         await async_session.commit()
 
         response = await async_client.patch(
@@ -1053,7 +1102,41 @@ async def test_update_automation_disable(self, async_client, async_session):
         )
 
         assert response.status_code == 200
-        assert response.json()["enabled"] is False
+        data = response.json()
+        assert data["enabled"] is False
+        assert data["disabled_reason"] == "manual"
+        assert data["disabled_detail"] == {"reason": "manual", "source": "user"}
+        assert data["disabled_at"] is not None
+
+        await async_session.refresh(automation)
+        await async_session.refresh(pending_run)
+        assert automation.disabled_reason == "manual"
+        assert automation.disabled_detail == {"reason": "manual", "source": "user"}
+        assert automation.disabled_at is not None
+        assert pending_run.status == AutomationRunStatus.SKIPPED
+        assert pending_run.completed_at == automation.disabled_at
+        assert pending_run.status_detail is not None
+        assert pending_run.status_detail["detail"] == "Automation disabled by user"
+        assert pending_run.status_detail["disabled_detail"] == {
+            "reason": "manual",
+            "source": "user",
+        }
+
+        events = (
+            (
+                await async_session.execute(
+                    select(AutomationDisableEvent).where(
+                        AutomationDisableEvent.automation_id == automation.id
+                    )
+                )
+            )
+            .scalars()
+            .all()
+        )
+        assert len(events) == 1
+        assert events[0].reason == "manual"
+        assert events[0].detail == {"reason": "manual", "source": "user"}
+        assert events[0].source == "manual"
 
     async def test_update_automation_model_profile(self, async_client, async_session):
         """PATCH can update the selected model profile."""
@@ -1521,6 +1604,34 @@ async def test_dispatch_automation_not_found(self, async_client):
         assert response.status_code == 404
         assert "Automation not found" in response.json()["detail"]
 
+    async def test_dispatch_disabled_automation_returns_reason(
+        self, async_client, async_session
+    ):
+        """Dispatching a disabled automation returns its blocking reason."""
+        automation = Automation(
+            user_id=TEST_USER_ID,
+            org_id=TEST_ORG_ID,
+            name="Disabled Automation",
+            trigger={"type": "cron", "schedule": "0 9 * * *", "timezone": "UTC"},
+            tarball_path="s3://bucket/code.tar.gz",
+            entrypoint="uv run script.py",
+            enabled=False,
+            disabled_reason="auth: Invalid API key",
+            disabled_detail={"kind": "auth", "threshold": 3},
+        )
+        async_session.add(automation)
+        await async_session.commit()
+
+        response = await async_client.post(
+            f"/api/automation/v1/{automation.id}/dispatch"
+        )
+
+        assert response.status_code == 409
+        detail = response.json()["detail"]
+        assert detail["message"] == "Automation is disabled"
+        assert detail["disabled_reason"] == "auth: Invalid API key"
+        assert detail["disabled_detail"] == {"kind": "auth", "threshold": 3}
+
     async def test_dispatch_automation_deleted(self, async_client, async_session):
         """Dispatching a soft-deleted automation returns 404."""
         automation = Automation(
@@ -2031,6 +2142,165 @@ async def test_complete_run_saves_conversation_id_for_completed_runs(
         assert run.conversation_id == "conv-completed-123"
         assert run.status == AutomationRunStatus.COMPLETED
 
+    async def test_complete_run_ignores_task_result_metadata_for_status_detail(
+        self, async_client, async_session
+    ):
+        """Task result metadata is not trusted for automation disablement."""
+        from openhands.automation.models import AutomationRun, AutomationRunStatus
+
+        automation = Automation(
+            user_id=TEST_USER_ID,
+            org_id=TEST_ORG_ID,
+            name="Task Outcome Automation",
+            trigger={"type": "cron", "schedule": "0 9 * * *", "timezone": "UTC"},
+            tarball_path="s3://bucket/code.tar.gz",
+            entrypoint="uv run script.py",
+        )
+        async_session.add(automation)
+        await async_session.commit()
+
+        run = AutomationRun(
+            automation_id=automation.id,
+            status=AutomationRunStatus.RUNNING,
+        )
+        async_session.add(run)
+        await async_session.commit()
+
+        response = await async_client.post(
+            f"/api/automation/v1/runs/{run.id}/complete",
+            json={
+                "status": "COMPLETED",
+                "blocking_factor": {
+                    "kind": "config",
+                    "reason": "User-defined task outcome",
+                    "source": "task",
+                },
+                "task_outcome": {
+                    "success": False,
+                    "message": "User-defined incomplete state",
+                    "classification": {"kind": "auth", "user_action": "settings"},
+                },
+            },
+        )
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "COMPLETED"
+        assert data["status_detail"] is None
+
+        await async_session.refresh(run)
+        assert run.status == AutomationRunStatus.COMPLETED
+        assert run.status_detail is None
+
+    async def test_failed_complete_run_uses_sdk_callback_error_classification(
+        self, async_client, async_session
+    ):
+        """Only SDK callback errors classify failed callbacks for disablement."""
+        from openhands.automation.models import AutomationRun, AutomationRunStatus
+
+        automation = Automation(
+            user_id=TEST_USER_ID,
+            org_id=TEST_ORG_ID,
+            name="Callback Error Automation",
+            trigger={"type": "cron", "schedule": "0 9 * * *", "timezone": "UTC"},
+            tarball_path="s3://bucket/code.tar.gz",
+            entrypoint="uv run script.py",
+        )
+        async_session.add(automation)
+        await async_session.commit()
+
+        run = AutomationRun(
+            automation_id=automation.id,
+            status=AutomationRunStatus.RUNNING,
+        )
+        async_session.add(run)
+        await async_session.commit()
+
+        response = await async_client.post(
+            f"/api/automation/v1/runs/{run.id}/complete",
+            json={
+                "status": "FAILED",
+                "error": {
+                    "source": "environment",
+                    "code": "MissingSecret",
+                    "detail": "Missing GitHub token",
+                    "classification": {
+                        "kind": "auth",
+                        "retryable": False,
+                        "user_action": "settings",
+                    },
+                },
+                "task_outcome": {
+                    "success": False,
+                    "classification": {"kind": "quota", "user_action": "settings"},
+                },
+            },
+        )
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "FAILED"
+        assert data["status_detail"]["kind"] == "auth"
+        assert data["status_detail"]["source"] == "environment"
+        assert data["status_detail"]["code"] == "MissingSecret"
+        assert data["status_detail"]["user_action"] == "settings"
+        assert "blocking_factor" not in data["status_detail"]
+
+        await async_session.refresh(run)
+        assert run.status == AutomationRunStatus.FAILED
+        assert run.status_detail is not None
+        assert run.status_detail["kind"] == "auth"
+        assert run.status_detail["source"] == "environment"
+
+    async def test_failed_complete_run_ignores_task_outcome_without_sdk_error(
+        self, async_client, async_session
+    ):
+        """User-defined task outcomes alone remain generic execution failures."""
+        from openhands.automation.models import AutomationRun, AutomationRunStatus
+
+        automation = Automation(
+            user_id=TEST_USER_ID,
+            org_id=TEST_ORG_ID,
+            name="Generic Failure Automation",
+            trigger={"type": "cron", "schedule": "0 9 * * *", "timezone": "UTC"},
+            tarball_path="s3://bucket/code.tar.gz",
+            entrypoint="uv run script.py",
+        )
+        async_session.add(automation)
+        await async_session.commit()
+
+        run = AutomationRun(
+            automation_id=automation.id,
+            status=AutomationRunStatus.RUNNING,
+        )
+        async_session.add(run)
+        await async_session.commit()
+
+        response = await async_client.post(
+            f"/api/automation/v1/runs/{run.id}/complete",
+            json={
+                "status": "FAILED",
+                "task_outcome": {
+                    "success": False,
+                    "message": "User-defined failed task",
+                    "classification": {"kind": "auth", "user_action": "settings"},
+                },
+            },
+        )
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "FAILED"
+        assert data["error_detail"] == "Completion callback reported failure"
+        assert data["status_detail"]["kind"] == "execution_error"
+        assert data["status_detail"]["source"] == "sdk_callback"
+        assert "user_action" not in data["status_detail"]
+
+        await async_session.refresh(run)
+        assert run.status == AutomationRunStatus.FAILED
+        assert run.status_detail is not None
+        assert run.status_detail["kind"] == "execution_error"
+
     async def test_complete_run_stores_finish_tool_response_metadata(
         self, async_client, async_session, monkeypatch
     ):
diff --git a/tests/test_schemas.py b/tests/test_schemas.py
index f221f4a..80d3f07 100644
--- a/tests/test_schemas.py
+++ b/tests/test_schemas.py
@@ -98,6 +98,16 @@ def test_preserves_legacy_structured_error(self):
 
         assert request.error == error
 
+    def test_accepts_blocking_factor_metadata(self):
+        blocking_factor = {"kind": "config", "reason": "Missing MCP token"}
+
+        request = RunCompleteRequest(
+            status="COMPLETED",
+            blocking_factor=blocking_factor,
+        )
+
+        assert request.blocking_factor == blocking_factor
+
 
 class TestAutomationRunResponseUtcSerialisation:
     """AutomationRunResponse must include a UTC offset in all datetime fields."""
@@ -196,6 +206,22 @@ def test_naive_created_at_serialises_with_utc_offset(self):
         data = automation.model_dump(mode="json")
         assert data["created_at"].endswith("+00:00") or data["created_at"].endswith("Z")
 
+    def test_disabled_metadata_serialises_for_api_consumers(self):
+        automation = self._make_automation(
+            enabled=False,
+            disabled_reason="auth: Invalid API key",
+            disabled_detail={"kind": "auth", "threshold": 3},
+            disabled_at=_NAIVE,
+        )
+        data = automation.model_dump(mode="json")
+
+        assert data["enabled"] is False
+        assert data["disabled_reason"] == "auth: Invalid API key"
+        assert data["disabled_detail"] == {"kind": "auth", "threshold": 3}
+        assert data["disabled_at"].endswith("+00:00") or data["disabled_at"].endswith(
+            "Z"
+        )
+
     def test_naive_last_triggered_at_serialises_with_utc_offset(self):
         automation = self._make_automation()
         data = automation.model_dump(mode="json")
diff --git a/tests/test_unhealthy_automations.py b/tests/test_unhealthy_automations.py
new file mode 100644
index 0000000..ea04a86
--- /dev/null
+++ b/tests/test_unhealthy_automations.py
@@ -0,0 +1,302 @@
+"""Tests for unhealthy automation classification and auto-disable behavior."""
+
+import uuid
+from datetime import timedelta
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from openhands.automation.models import (
+    Automation,
+    AutomationDisableEvent,
+    AutomationRun,
+    AutomationRunStatus,
+    Base,
+)
+from openhands.automation.utils.time import utcnow
+from openhands.automation.utils.unhealthy import (
+    is_permanent_failure_detail,
+    maybe_disable_unhealthy_automation,
+)
+
+
+TEST_USER_ID = uuid.UUID("12345678-1234-5678-1234-567812345678")
+TEST_ORG_ID = uuid.UUID("87654321-4321-8765-4321-876543218765")
+
+
+@pytest.fixture
+async def sqlite_session_factory():
+    engine = create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+    )
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+    factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    yield factory
+    await engine.dispose()
+
+
+def _permanent_detail(kind: str = "auth") -> dict:
+    return {
+        "phase": "callback",
+        "kind": kind,
+        "detail": "Invalid API key",
+        "transient": False,
+        "user_action": "settings",
+        "fingerprint": f"callback:sdk_callback:{kind}",
+    }
+
+
+def _transient_detail() -> dict:
+    return {
+        "phase": "callback",
+        "kind": "rate_limit",
+        "detail": "Provider returned 429",
+        "transient": True,
+        "user_action": "retry",
+        "fingerprint": "callback:sdk_callback:rate_limit",
+    }
+
+
+async def _create_automation(session: AsyncSession) -> Automation:
+    automation = Automation(
+        user_id=TEST_USER_ID,
+        org_id=TEST_ORG_ID,
+        name="Unhealthy automation",
+        trigger={"type": "cron", "schedule": "* * * * *", "timezone": "UTC"},
+        tarball_path="https://example.com/automation.tar",
+        entrypoint="python main.py",
+        enabled=True,
+    )
+    session.add(automation)
+    await session.flush()
+    return automation
+
+
+async def _add_terminal_run(
+    session: AsyncSession,
+    automation: Automation,
+    *,
+    status_detail: dict | None,
+    index: int,
+    status: AutomationRunStatus = AutomationRunStatus.FAILED,
+) -> AutomationRun:
+    now = utcnow() + timedelta(seconds=index)
+    run = AutomationRun(
+        automation_id=automation.id,
+        status=status,
+        status_detail=status_detail,
+        created_at=now,
+        completed_at=now,
+    )
+    session.add(run)
+    await session.flush()
+    return run
+
+
+def test_permanent_failure_classifier_uses_sdk_semantics():
+    assert is_permanent_failure_detail(_permanent_detail("auth")) is True
+    assert is_permanent_failure_detail(_permanent_detail("config")) is True
+    assert is_permanent_failure_detail(_permanent_detail("quota")) is True
+    assert is_permanent_failure_detail(_transient_detail()) is False
+    assert (
+        is_permanent_failure_detail(
+            {"kind": "config", "detail": "bad model", "transient": True}
+        )
+        is False
+    )
+    assert (
+        is_permanent_failure_detail(
+            {"kind": "internal", "detail": "service bug", "transient": False}
+        )
+        is False
+    )
+
+
+async def test_auto_disables_after_consecutive_permanent_failures(
+    sqlite_session_factory,
+):
+    async with sqlite_session_factory() as session:
+        automation = await _create_automation(session)
+        await _add_terminal_run(
+            session, automation, status_detail=_permanent_detail(), index=1
+        )
+        await _add_terminal_run(
+            session, automation, status_detail=_permanent_detail(), index=2
+        )
+        await _add_terminal_run(
+            session, automation, status_detail=_permanent_detail(), index=3
+        )
+        pending_run = AutomationRun(
+            automation_id=automation.id,
+            status=AutomationRunStatus.PENDING,
+        )
+        session.add(pending_run)
+        await session.flush()
+
+        disabled = await maybe_disable_unhealthy_automation(
+            session,
+            automation.id,
+            threshold=3,
+        )
+        await session.refresh(automation)
+
+        assert disabled is True
+        assert automation.enabled is False
+        assert automation.disabled_reason is not None
+        assert "auth" in automation.disabled_reason
+        assert automation.disabled_detail is not None
+        assert automation.disabled_detail["consecutive_permanent_failures"] == 3
+        await session.refresh(pending_run)
+        assert pending_run.status == AutomationRunStatus.SKIPPED
+        assert pending_run.completed_at is not None
+        assert pending_run.status_detail is not None
+        assert pending_run.status_detail["operation"] == "automation_disabled"
+
+        events = (
+            (
+                await session.execute(
+                    select(AutomationDisableEvent).where(
+                        AutomationDisableEvent.automation_id == automation.id
+                    )
+                )
+            )
+            .scalars()
+            .all()
+        )
+        assert len(events) == 1
+        assert events[0].source == "consecutive_permanent_failures"
+        assert events[0].run_id is not None
+        assert str(events[0].run_id) == automation.disabled_detail["run_id"]
+
+
+async def test_direct_disable_records_event_history(sqlite_session_factory):
+    from openhands.automation.utils.run import disable_automation
+
+    async with sqlite_session_factory() as session:
+        automation = await _create_automation(session)
+        run = await _add_terminal_run(
+            session,
+            automation,
+            status_detail=_permanent_detail(),
+            index=1,
+        )
+        pending_run = AutomationRun(
+            automation_id=automation.id,
+            status=AutomationRunStatus.PENDING,
+        )
+        session.add(pending_run)
+        await session.flush()
+        automation_id = automation.id
+        run_id = run.id
+        pending_run_id = pending_run.id
+        await session.commit()
+
+    disabled = await disable_automation(
+        sqlite_session_factory,
+        automation_id,
+        "Tarball not found",
+        disabled_detail={"kind": "config"},
+        run_id=run_id,
+        source="permanent_dispatch_failure",
+    )
+
+    assert disabled is True
+
+    async with sqlite_session_factory() as session:
+        events = (
+            (
+                await session.execute(
+                    select(AutomationDisableEvent).where(
+                        AutomationDisableEvent.automation_id == automation_id
+                    )
+                )
+            )
+            .scalars()
+            .all()
+        )
+        assert len(events) == 1
+        assert events[0].run_id == run_id
+        assert events[0].reason == "Tarball not found"
+        assert events[0].detail == {"kind": "config"}
+        assert events[0].source == "permanent_dispatch_failure"
+
+        pending_run = await session.get(AutomationRun, pending_run_id)
+        assert pending_run is not None
+        assert pending_run.status == AutomationRunStatus.SKIPPED
+        assert pending_run.completed_at is not None
+        assert pending_run.status_detail is not None
+        assert pending_run.status_detail["operation"] == "automation_disabled"
+
+
+async def test_transient_failures_do_not_count_toward_disable(
+    sqlite_session_factory,
+):
+    async with sqlite_session_factory() as session:
+        automation = await _create_automation(session)
+        await _add_terminal_run(
+            session, automation, status_detail=_permanent_detail(), index=1
+        )
+        await _add_terminal_run(
+            session, automation, status_detail=_permanent_detail(), index=2
+        )
+        await _add_terminal_run(
+            session, automation, status_detail=_transient_detail(), index=3
+        )
+
+        disabled = await maybe_disable_unhealthy_automation(
+            session,
+            automation.id,
+            threshold=3,
+        )
+        await session.refresh(automation)
+
+        assert disabled is False
+        assert automation.enabled is True
+        assert automation.disabled_reason is None
+
+
+async def test_sdk_callback_error_runs_count_as_permanent_failures(
+    sqlite_session_factory,
+):
+    async with sqlite_session_factory() as session:
+        automation = await _create_automation(session)
+        callback_error_detail = {
+            "phase": "callback",
+            "kind": "auth",
+            "detail": "Missing MCP token",
+            "transient": False,
+            "source": "environment",
+            "code": "MissingSecret",
+            "user_action": "settings",
+        }
+        await _add_terminal_run(
+            session,
+            automation,
+            status=AutomationRunStatus.FAILED,
+            status_detail=callback_error_detail,
+            index=1,
+        )
+        await _add_terminal_run(
+            session,
+            automation,
+            status=AutomationRunStatus.FAILED,
+            status_detail=callback_error_detail,
+            index=2,
+        )
+
+        disabled = await maybe_disable_unhealthy_automation(
+            session,
+            automation.id,
+            threshold=2,
+        )
+        await session.refresh(automation)
+
+        assert disabled is True
+        assert automation.enabled is False
+        assert automation.disabled_detail is not None
+        assert automation.disabled_detail["status_detail"]["source"] == "environment"
+        assert automation.disabled_detail["status_detail"]["code"] == "MissingSecret"