fix(scheduler): treat only busy-thread 409s as retryable (#7196)

* fix(scheduler): treat only busy-thread 409s as retryable

start_run exposes the retryable run-manager admission conflict as BusyThreadConflict (an HTTP 409), but also uses plain 409s for permanent idempotency-key and thread/assistant mismatches. The scheduler treated every HTTP 409 as an overlap conflict and requeued the occurrence, so a permanent launch error was retried until scheduler.queue_timeout_seconds elapsed. Other launch errors are documented to fail immediately, and the MCP notification launcher already distinguishes these types.

Match ConflictError and BusyThreadConflict only; a plain 409 now follows the normal failure path. The type is imported lazily so the scheduler module does not take a load-time dependency on app.gateway.services.

Tests: a plain idempotency 409 terminalizes the occurrence as failed, while BusyThreadConflict remains queued; existing ConflictError coverage is unchanged.

* fix(scheduler): document the busy-thread import boundary and pin the failure detail

Keep the BusyThreadConflict import local to the classifier and document why it is guaranteed to resolve: the scheduler is wired by the Gateway, where app.gateway.services is already loaded, and this must not become an optional dependency. Import the same type locally in the test so collection does not initialize gateway services for the rest of the scheduler suite. Also pin the persisted failure detail so the 409 reason stays operator-visible.

---------

Co-authored-by: Tong-bit-art <Tong-bit-art@users.noreply.github.com>

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:

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