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

ADR-0007 bounded how long an unreachable Redis can block a dispatch and
moved one call site off the event loop, leaving the other 33 bounded but
still inline. This finishes that follow-up and makes the property
enforceable rather than remembered.

Add backend/core/task_dispatch.py with dispatch_task and
dispatch_task_best_effort, and route every dispatch reachable from a
request handler through it. That covers the 33 remaining sites plus two
helpers that were synchronous but only ever called from async code.
enqueue_model_preparation is one, and it blocked twice, since it also wrote
download progress to Redis inline. _enqueue_push_channel_refresh is the
other, and its own docstring said it existed to keep API paths fast, which
the blocking publish undermined.

The helper uses asyncio.to_thread rather than Starlette's
run_in_threadpool, for two reasons beyond taste. calendar_service is
imported by the worker, whose image ships no ASGI stack, so a Starlette
import there would not resolve. And the loop's default executor is separate
from the anyio limiter that serves sync route handlers, so a Redis outage
can no longer consume the threads those handlers need. That closes the
threadpool residual ADR-0007 recorded rather than merely shrinking it.

Best-effort dispatch is now a named function rather than a bare except at
each site, so the choice to swallow is visible where it is made instead of
being inferred from a try block. Worker-side code still dispatches inline,
deliberately, having no event loop to protect.

A test walks the tree and fails on any send_task, apply_async or delay
inside an async def, naming file and line. Verified by reintroducing one
inline dispatch, which it caught at documents.py:123.

Align the tests on one seam while here. Four files reached for a module's
celery_app re-export, which is the same import-time binding mistake that
made the suite slow, and they broke as soon as the import moved. They now
patch the shared app object. Two more stubbed enqueue_model_preparation
with sync lambdas that are now awaited.

Behaviour is unchanged and re-measured against an unreachable broker: 6.03s
to fail a dispatch, with 60 concurrent no-op requests at a 0.9ms median and
a 0.00s worst case, matching the run before the mechanism swap. 1151 tests
pass.

Refs: docs/adr/0007-bounded-fail-fast-task-dispatch.md, docs/ARCHITECTURE.md

Author: Valtora

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],
             )