refactor(api): route every async dispatch through one off-loop helper

backend/api/services/health_service.py

@@ -14,6 +14,7 @@
 from backend.celery_app import celery_app
 from backend.core.db import sync_engine
 from backend.core.redis import get_redis_url
+from backend.core.task_dispatch import dispatch_task
 from backend.preload_models import check_model_status
 from backend.utils.config_manager import async_get_system_api_keys, config_manager
 from backend.utils.deployment_warnings import get_deployment_warnings
@@ -528,7 +529,7 @@ async def _get_device_component(worker_status: str) -> tuple[dict[str, Any], boo
         )
 
     try:
-        task = celery_app.send_task("backend.worker.tasks.get_worker_device_status")
+        task = await dispatch_task("backend.worker.tasks.get_worker_device_status")
         payload = await asyncio.to_thread(task.get, timeout=5)
     except Exception:  # noqa: BLE001
         return (

backend/api/v1/endpoints/backup.py

@@ -29,6 +29,7 @@
     RESTORE_LOCK_TTL_SECONDS,
 )
 from backend.core.redis import REDIS_URL
+from backend.core.task_dispatch import dispatch_task
 from backend.models.user import User
 from backend.utils.path_manager import PathManager
 from backend.utils.rate_limit import enforce_upload_concurrency
@@ -92,7 +93,7 @@ async def _dispatch_restore(
         )
 
     try:
-        task = celery_app.send_task(
+        task = await dispatch_task(
             "backend.worker.tasks.restore_backup_task",
             kwargs={
                 "zip_path": str(archive_path),
@@ -138,7 +139,7 @@ async def export_backup(
     try:
         # Trigger Celery task
         # Uses send_task to avoid importing the task function directly (bypasses heavy imports).
-        task = celery_app.send_task(
+        task = await dispatch_task(
             "backend.worker.tasks.create_backup_task",
             kwargs={
                 "include_audio": include_audio,

backend/api/v1/endpoints/cli_oauth.py

@@ -23,7 +23,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.api.deps import get_current_admin_user, get_current_user, get_db
-from backend.celery_app import celery_app
+from backend.core.task_dispatch import dispatch_task
 from backend.models.cli_oauth import (
     CliOAuthCredential,
     CliOAuthCredentialStatus,
@@ -363,7 +363,7 @@ async def get_codex_models(
             models=[CliCodexModel(**model) for model in cached], source="live"
         )
     # Warm the cache for next time (runs in worker-io); serve the fallback now.
-    celery_app.send_task("backend.worker.tasks.refresh_codex_models_task")
+    await dispatch_task("backend.worker.tasks.refresh_codex_models_task")
     return CliCodexModelsRead(models=_CODEX_FALLBACK_MODELS, source="fallback")
 
 
@@ -400,7 +400,7 @@ async def refresh_codex_models(
     stale list it was pressed to replace.
     """
     await codex_oauth.clear_model_catalog()
-    celery_app.send_task("backend.worker.tasks.refresh_codex_models_task")
+    await dispatch_task("backend.worker.tasks.refresh_codex_models_task")
     return CliCodexModelsRead(models=_CODEX_FALLBACK_MODELS, source="fallback")
 
 
@@ -476,7 +476,7 @@ async def start_cli_oauth(
         # verification URL + code. Never block the request — a slow login would
         # otherwise trip a proxy timeout (Cloudflare 520).
         await codex_oauth.clear_login_state(current_user.id)
-        celery_app.send_task(
+        await dispatch_task(
             "backend.worker.tasks.codex_device_login_task",
             args=[current_user.id],
         )

backend/api/v1/endpoints/documents.py

@@ -9,7 +9,7 @@
 
 from backend.api.deps import get_current_user, get_db
 from backend.api.error_handling import sanitized_http_exception
-from backend.celery_app import celery_app
+from backend.core.task_dispatch import dispatch_task
 from backend.models.document import Document, DocumentStatus
 from backend.models.recording import Recording
 from backend.models.recording_public import DocumentPublicRead, serialize_document
@@ -119,7 +119,7 @@ async def upload_document(
     await db.commit()
     await db.refresh(document)
 
-    task = celery_app.send_task(
+    task = await dispatch_task(
         "backend.worker.tasks.process_document_task", args=[document.id]
     )
     from backend.models.task import register_task_ownership

backend/api/v1/endpoints/notes_templates.py

@@ -15,7 +15,7 @@
 from sqlmodel import select
 
 from backend.api.deps import get_current_user, get_db
-from backend.celery_app import celery_app
+from backend.core.task_dispatch import dispatch_task
 from backend.models.notes_template import (
     NotesTemplate,
     NotesTemplateCreate,
@@ -414,7 +414,7 @@ async def generate_notes_structure(
 
     job_id = new_job_id()
     await publish_job_async(job_id, {"status": STATUS_PENDING})
-    task = celery_app.send_task(
+    task = await dispatch_task(
         "backend.worker.tasks.generate_notes_structure_task",
         args=[job_id, current_user.id, brief],
     )

backend/api/v1/endpoints/recordings/helpers.py

@@ -12,6 +12,7 @@
 from sqlmodel import select
 
 import backend.api.v1.endpoints.recordings as recordings_module
+from backend.core.task_dispatch import dispatch_task
 from backend.models.calendar import CalendarConnection, CalendarEvent, CalendarSource
 from backend.models.chat import ChatMessage
 from backend.models.context_chunk import ContextChunk
@@ -700,7 +701,7 @@ async def _requeue_for_processing(
     await db.commit()
     await db.refresh(recording)
 
-    task = recordings_module.celery_app.send_task(
+    task = await dispatch_task(
         "backend.worker.tasks.process_recording_task",
         args=[recording.id, True, engine_override],
     )

backend/api/v1/endpoints/recordings/routes_actions.py

@@ -6,6 +6,7 @@
 
 import backend.api.v1.endpoints.recordings as recordings_module
 from backend.api.deps import get_current_user, get_db
+from backend.core.task_dispatch import dispatch_task
 from backend.models.calendar import CalendarEvent
 from backend.models.recording import RecordingStatus, RecordingUpdate
 from backend.models.recording_public import RecordingPublicRead, serialize_recording
@@ -298,7 +299,7 @@ async def infer_speakers_for_recording(
     await db.commit()
     await db.refresh(recording)
 
-    task = recordings_module.celery_app.send_task(
+    task = await dispatch_task(
         "backend.worker.tasks.infer_speakers_task", args=[recording.id]
     )
     recording.celery_task_id = task.id

backend/api/v1/endpoints/recordings/routes_capture.py

@@ -10,6 +10,7 @@
 import backend.api.v1.endpoints.recordings as recordings_module
 from backend.api.deps import get_current_recording_client_user, get_db
 from backend.api.error_handling import sanitized_http_exception
+from backend.core.task_dispatch import dispatch_task
 from backend.models.pipeline import RecordingAudioChunk, RecordingAudioWindowManifest
 from backend.models.recording import (
     CaptureSourceReportCreate,
@@ -221,7 +222,7 @@ async def upload_segment(
         "enable_live_transcription"
     ):
         try:
-            recordings_module.celery_app.send_task(
+            await dispatch_task(
                 "backend.processing.live_transcribe.transcribe_segment_live_task",
                 args=[recording.id, sequence],
             )
@@ -234,7 +235,7 @@ async def upload_segment(
             )
     elif segment_suffix in BROWSER_AUDIO_SEGMENT_SUFFIXES:
         try:
-            recordings_module.celery_app.send_task(
+            await dispatch_task(
                 "backend.processing.segment_transcode.transcode_segment_task",
                 args=[recording.id, sequence],
             )
@@ -466,7 +467,7 @@ async def finalize_upload(
     await db.commit()
     await db.refresh(recording)
 
-    task = recordings_module.celery_app.send_task(
+    task = await dispatch_task(
         "backend.worker.tasks.process_recording_task", args=[recording.id]
     )
     recording.celery_task_id = task.id
@@ -476,7 +477,7 @@ async def finalize_upload(
 
     await register_task_ownership(db, task.id, recording.user_id)
 
-    proxy_task = recordings_module.celery_app.send_task(
+    proxy_task = await dispatch_task(
         "backend.worker.tasks.generate_proxy_task", args=[recording.id]
     )
     if proxy_task:

backend/api/v1/endpoints/recordings/routes_import_upload.py

@@ -14,6 +14,7 @@
 import backend.api.v1.endpoints.recordings as recordings_module
 from backend.api.deps import get_current_user, get_db
 from backend.api.error_handling import sanitized_http_exception
+from backend.core.task_dispatch import dispatch_task
 from backend.models.pipeline import RecordingAudioChunk, RecordingAudioWindowManifest
 from backend.models.recording import ClientStatus, Recording, RecordingStatus
 from backend.models.recording_public import RecordingPublicRead, serialize_recording
@@ -172,7 +173,7 @@ async def import_audio(
     await db.commit()
 
     # Trigger processing task
-    task = recordings_module.celery_app.send_task(
+    task = await dispatch_task(
         "backend.worker.tasks.process_recording_task", args=[recording.id]
     )
     recording.celery_task_id = task.id
@@ -184,7 +185,7 @@ async def import_audio(
 
     # Trigger proxy generation task
     if not recording.proxy_path:
-        proxy_task = recordings_module.celery_app.send_task(
+        proxy_task = await dispatch_task(
             "backend.worker.tasks.generate_proxy_task", args=[recording.id]
         )
         if proxy_task:
@@ -438,7 +439,7 @@ async def finalize_chunked_import(
     await db.commit()
     await db.refresh(recording)
 
-    task = recordings_module.celery_app.send_task(
+    task = await dispatch_task(
         "backend.worker.tasks.process_recording_task", args=[recording.id]
     )
     recording.celery_task_id = task.id
@@ -449,7 +450,7 @@ async def finalize_chunked_import(
     await register_task_ownership(db, task.id, recording.user_id)
 
     if not recording.proxy_path:
-        proxy_task = recordings_module.celery_app.send_task(
+        proxy_task = await dispatch_task(
             "backend.worker.tasks.generate_proxy_task", args=[recording.id]
         )
         if proxy_task:
@@ -538,7 +539,7 @@ async def upload_recording(
     await db.commit()
     await db.refresh(recording)
 
-    task = recordings_module.celery_app.send_task(
+    task = await dispatch_task(
         "backend.worker.tasks.process_recording_task", args=[recording.id]
     )
     recording.celery_task_id = task.id
@@ -549,7 +550,7 @@ async def upload_recording(
     await register_task_ownership(db, task.id, recording.user_id)
 
     if not recording.proxy_path:
-        proxy_task = recordings_module.celery_app.send_task(
+        proxy_task = await dispatch_task(
             "backend.worker.tasks.generate_proxy_task", args=[recording.id]
         )
         if proxy_task:

backend/api/v1/endpoints/speakers/routes_global.py

@@ -13,6 +13,7 @@
 
 import backend.api.v1.endpoints.speakers as speakers_module
 from backend.api.deps import get_current_user, get_db
+from backend.core.task_dispatch import dispatch_task
 from backend.models.people_tag_schemas import PeopleTagRead
 from backend.models.recording import (
     Recording,
@@ -545,7 +546,7 @@ async def recalibrate_voiceprint(
             )
             continue
 
-        task = speakers_module.celery_app.send_task(
+        task = await dispatch_task(
             "backend.worker.tasks.extract_embedding_task",
             args=[target_audio, segs, device_str, hf_token],
         )
@@ -729,7 +730,7 @@ async def split_speaker(
             seg_tuples = [(s.start, s.end) for s in segments]
             target_audio = select_recording_audio_for_embedding(rec)
 
-            task = speakers_module.celery_app.send_task(
+            task = await dispatch_task(
                 "backend.worker.tasks.extract_embedding_task",
                 args=[target_audio, seg_tuples, device_str, hf_token],
             )
@@ -801,7 +802,7 @@ async def split_speaker(
 
         if remaining_seg_tuples:
             target_audio = select_recording_audio_for_embedding(rec)
-            task = speakers_module.celery_app.send_task(
+            task = await dispatch_task(
                 "backend.worker.tasks.extract_embedding_task",
                 args=[target_audio, remaining_seg_tuples, device_str, hf_token],
             )

backend/api/v1/endpoints/speakers/routes_voiceprint.py

@@ -8,6 +8,7 @@
 
 import backend.api.v1.endpoints.speakers as speakers_module
 from backend.api.deps import get_current_user, get_db
+from backend.core.task_dispatch import dispatch_task
 from backend.models.speaker import GlobalSpeaker, RecordingSpeaker
 from backend.models.transcript import Transcript
 from backend.models.user import User
@@ -91,7 +92,7 @@ async def extract_voiceprint(
     )
 
     target_audio = select_recording_audio_for_embedding(recording)
-    task = speakers_module.celery_app.send_task(
+    task = await dispatch_task(
         "backend.worker.tasks.extract_embedding_task",
         args=[target_audio, speaker_segments, device_str, hf_token],
     )
@@ -375,7 +376,7 @@ async def extract_all_voiceprints(
         )
 
         target_audio = select_recording_audio_for_embedding(recording)
-        task = speakers_module.celery_app.send_task(
+        task = await dispatch_task(
             "backend.worker.tasks.extract_embedding_task",
             args=[target_audio, speaker_segments, device_str, hf_token],
         )

backend/api/v1/endpoints/system.py

@@ -36,12 +36,12 @@
     is_system_initialized,
     require_first_run_password,
 )
-from backend.celery_app import celery_app
 from backend.core.security import (
     MIN_PASSWORD_LENGTH,
     hash_user_password,
     validate_password_policy,
 )
+from backend.core.task_dispatch import dispatch_task
 from backend.models.user import User
 from backend.preload_models import check_model_status
 from backend.seed_demo import seed_demo_data
@@ -484,7 +484,7 @@ async def setup_system(
 
     model_preparation_task_id = None
     try:
-        model_preparation_task_id = enqueue_model_preparation(
+        model_preparation_task_id = await enqueue_model_preparation(
             whisper_model_size=setup_in.whisper_model_size or "turbo",
             transcription_backend="whisper",
             include_core=True,
@@ -647,7 +647,7 @@ def effective(key: str, default: str) -> str:
         include_core = False
 
     try:
-        task_id = enqueue_model_preparation(
+        task_id = await enqueue_model_preparation(
             whisper_model_size=effective("whisper_model_size", "turbo"),
             transcription_backend=transcription_backend,
             parakeet_model=effective("parakeet_model", "parakeet-tdt-0.6b-v3"),
@@ -683,7 +683,7 @@ async def delete_model_endpoint(
         raise HTTPException(status_code=400, detail="Invalid model name")
 
     try:
-        task = celery_app.send_task(
+        task = await dispatch_task(
             MODEL_DELETION_TASK,
             kwargs={"model_name": model_name, "variant": variant},
         )

backend/api/v1/endpoints/transcripts/helpers.py

@@ -13,9 +13,8 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm.attributes import flag_modified
 from sqlmodel import select
-from starlette.concurrency import run_in_threadpool
 
-from backend.celery_app import celery_app
+from backend.core.task_dispatch import dispatch_task_best_effort
 from backend.models.pipeline import SpeakerCorrectionScope
 from backend.models.recording import (
     LEGACY_RECORDING_REPROCESS_REQUIRED_DETAIL,
@@ -70,18 +69,11 @@ async def _dispatch_meeting_edge_refresh(
     if not enabled:
         return
 
-    try:
-        await run_in_threadpool(
-            celery_app.send_task,
-            "backend.worker.tasks.refresh_meeting_edge_task",
-            args=[recording_id],
-        )
-    except Exception as exc:  # noqa: BLE001
-        logger.warning(
-            "Failed to dispatch Meeting Edge refresh for recording %s: %s",
-            recording_id,
-            exc,
-        )
+    await dispatch_task_best_effort(
+        "backend.worker.tasks.refresh_meeting_edge_task",
+        args=[recording_id],
+        context=f"recording {recording_id}",
+    )
 
 
 async def _get_recording_transcript(