diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py
index 76a21df9b4..5ede6915a9 100644
--- a/backend/app/scheduler/service.py
+++ b/backend/app/scheduler/service.py
@@ -11,7 +11,6 @@ from datetime import UTC, datetime, timedelta
 from typing import Any, Literal
 from zoneinfo import ZoneInfo
 
-from fastapi import HTTPException
 from sqlalchemy import select
 
 from app.channels.capabilities import supports_proactive_group_notifications, supports_proactive_notifications
@@ -473,7 +472,19 @@ class ScheduledTaskService:
     def _is_overlap_conflict(exc: Exception) -> bool:
         if isinstance(exc, ConflictError):
             return True
-        return isinstance(exc, HTTPException) and exc.status_code == 409
+        # ``start_run`` converts a run-manager admission conflict into this
+        # typed HTTP 409. Other 409s (idempotency-key mismatch, thread or
+        # assistant drift) are permanent launch failures; retrying them would
+        # keep the occurrence queued until the queue timeout.
+        #
+        # Keep the import local to avoid changing the scheduler module's
+        # load-time import graph. The scheduler is wired by the Gateway, where
+        # ``app.gateway.services`` is already loaded, so this lookup is
+        # guaranteed; do not make it optional without moving the
+        # classification to a dependency that is always importable here.
+        from app.gateway.services import BusyThreadConflict
+
+        return isinstance(exc, BusyThreadConflict)
 
     @staticmethod
     def _task_status_for_failure(task: dict[str, Any], *, trigger: str) -> str:
diff --git a/backend/tests/test_scheduled_task_service.py b/backend/tests/test_scheduled_task_service.py
index 0bc2766c5a..34ea9b3403 100644
--- a/backend/tests/test_scheduled_task_service.py
+++ b/backend/tests/test_scheduled_task_service.py
@@ -3,6 +3,7 @@ from datetime import UTC, datetime, timedelta
 
 import pytest
 import pytest_asyncio
+from fastapi import HTTPException
 
 from app.scheduler.service import ScheduledTaskService
 from deerflow.runtime import ConflictError, RunStatus
@@ -351,6 +352,85 @@ async def test_manual_overlap_conflict_is_kept_in_queue():
     assert task_repo.release_calls == []
 
 
+@pytest.mark.asyncio
+async def test_plain_409_from_launch_is_a_permanent_failure():
+    async def fake_launch(**_kwargs):
+        raise HTTPException(status_code=409, detail="Idempotency-Key already used with a different request")
+
+    task_repo = DummyTaskRepo(
+        [
+            {
+                "id": "task-permanent-409",
+                "user_id": "user-1",
+                "thread_id": "thread-1",
+                "context_mode": "reuse_thread",
+                "assistant_id": "lead_agent",
+                "prompt": "Summarize thread",
+                "schedule_type": "cron",
+                "schedule_spec": {"cron": "0 9 * * *"},
+                "timezone": "UTC",
+                "status": "enabled",
+                "overlap_policy": "enqueue",
+            }
+        ]
+    )
+    run_repo = DummyRunRepo()
+    service = ScheduledTaskService(
+        task_repo=task_repo,
+        task_run_repo=run_repo,
+        launch_run=fake_launch,
+        poll_interval_seconds=5,
+        lease_seconds=120,
+        max_concurrent_runs=3,
+    )
+
+    result = await service.dispatch_task(task_repo.rows[0], now=datetime.now(UTC), trigger="scheduled")
+
+    assert result["outcome"] == "failed"
+    assert run_repo.updated[-1][1]["status"] == "failed"
+    assert "Idempotency-Key already used" in run_repo.updated[-1][1]["error"]
+
+
+@pytest.mark.asyncio
+async def test_busy_thread_http_conflict_is_kept_in_queue():
+    from app.gateway.services import BusyThreadConflict
+
+    async def fake_launch(**_kwargs):
+        raise BusyThreadConflict("Thread thread-1 already has an active run")
+
+    task_repo = DummyTaskRepo(
+        [
+            {
+                "id": "task-busy-thread",
+                "user_id": "user-1",
+                "thread_id": "thread-1",
+                "context_mode": "reuse_thread",
+                "assistant_id": "lead_agent",
+                "prompt": "Summarize thread",
+                "schedule_type": "cron",
+                "schedule_spec": {"cron": "0 9 * * *"},
+                "timezone": "UTC",
+                "status": "enabled",
+                "overlap_policy": "enqueue",
+            }
+        ]
+    )
+    run_repo = DummyRunRepo()
+    service = ScheduledTaskService(
+        task_repo=task_repo,
+        task_run_repo=run_repo,
+        launch_run=fake_launch,
+        poll_interval_seconds=5,
+        lease_seconds=120,
+        max_concurrent_runs=3,
+    )
+
+    result = await service.dispatch_task(task_repo.rows[0], now=datetime.now(UTC), trigger="scheduled")
+
+    assert result["outcome"] == "queued"
+    assert run_repo.updated[-1][1]["status"] == "queued"
+
+
 @pytest.mark.asyncio
 async def test_dispatch_task_records_failure_for_legacy_invalid_thread_id():
     """Rows persisted before the thread-id contract was centralized may store