UN-3497 [FEAT] Introduce queue_backend seam and migrate Celery call sites

workers/api-deployment/tasks.py

@@ -7,6 +7,7 @@
 import time
 from typing import Any
 
+from queue_backend import worker_task
 from shared.api import InternalAPIClient
 from shared.enums.status_enums import PipelineStatus
 from shared.enums.task_enums import TaskName
@@ -289,7 +290,7 @@ def _unified_api_execution(
             logger.warning(f"Failed to cleanup StateStore context: {cleanup_error}")
 
 
-@app.task(
+@worker_task(
     bind=True,
     name=TaskName.ASYNC_EXECUTE_BIN_API,
     autoretry_for=(Exception,),
@@ -348,7 +349,7 @@ def async_execute_bin_api(
     )
 
 
-@app.task(
+@worker_task(
     bind=True,
     name=TaskName.ASYNC_EXECUTE_BIN,
     autoretry_for=(Exception,),
@@ -993,7 +994,7 @@ def _calculate_manual_review_decisions_for_batch_api(
         return [False] * len(batch)
 
 
-@app.task(bind=True)
+@worker_task(bind=True)
 @monitor_performance
 @retry(max_attempts=3, base_delay=2.0)
 @with_execution_context
@@ -1064,7 +1065,7 @@ def api_deployment_status_check(
         raise
 
 
-@app.task(bind=True)
+@worker_task(bind=True)
 @monitor_performance
 @with_execution_context
 def api_deployment_cleanup(

workers/api-deployment/worker.py

@@ -4,6 +4,7 @@
 Handles API deployment, cleanup, and status checking tasks.
 """
 
+from queue_backend import worker_task
 from shared.enums.worker_enums import WorkerType
 from shared.infrastructure.config.builder import WorkerBuilder
 from shared.infrastructure.config.registry import WorkerRegistry
@@ -59,7 +60,7 @@ def check_api_deployment_health():
 )
 
 
-@app.task(bind=True)
+@worker_task(bind=True)
 def healthcheck(self):
     """Health check task for monitoring systems."""
     return {

workers/callback/tasks.py

@@ -7,8 +7,7 @@
 import time
 from typing import Any
 
-# Use Celery current_app to avoid circular imports
-from celery import current_app as app
+from queue_backend import worker_task
 
 # Import shared worker infrastructure
 from shared.api import InternalAPIClient
@@ -1512,7 +1511,7 @@ def _process_batch_callback_core(
             logger.warning("api_client.close() failed during callback cleanup: %s", e)
 
 
-@app.task(
+@worker_task(
     bind=True,
     name=TaskName.PROCESS_BATCH_CALLBACK,
     max_retries=0,  # Match Django backend pattern
@@ -1536,7 +1535,7 @@ def process_batch_callback(self, results, *args, **kwargs) -> dict[str, Any]:
     return _process_batch_callback_core(self, results, *args, **kwargs)
 
 
-@app.task(
+@worker_task(
     bind=True,
     name="process_batch_callback_api",
     autoretry_for=(Exception,),
@@ -1932,7 +1931,7 @@ def _publish_final_workflow_ui_logs_api(
         )
 
 
-@app.task(
+@worker_task(
     bind=True,
     name="workflow_manager.workflow_v2.file_execution_tasks.process_batch_callback",
     max_retries=0,

workers/callback/worker.py

@@ -3,6 +3,7 @@
 Celery worker for processing file processing callbacks and status updates.
 """
 
+from queue_backend import worker_task
 from shared.enums.worker_enums import WorkerType
 from shared.infrastructure.config.builder import WorkerBuilder
 from shared.infrastructure.config.registry import WorkerRegistry
@@ -61,7 +62,7 @@ def check_callback_health():
 )
 
 
-@app.task(bind=True)
+@worker_task(bind=True)
 def healthcheck(self):
     """Health check task for monitoring systems."""
     return {

workers/executor/tasks.py

@@ -5,7 +5,7 @@
 ExecutionOrchestrator, and returns an ExecutionResult dict.
 """
 
-from celery import shared_task
+from queue_backend import worker_task
 from shared.clients import UsageAPIClient
 from shared.enums.task_enums import TaskName
 from shared.infrastructure.config import WorkerConfig
@@ -27,7 +27,7 @@
 )
 
 
-@shared_task(
+@worker_task(
     bind=True,
     name=TaskName.EXECUTE_EXTRACTION,
     autoretry_for=(ConnectionError, TimeoutError, OSError),

workers/executor/worker.py

@@ -7,6 +7,7 @@
 import logging
 import os
 
+from queue_backend import worker_task
 from shared.enums.worker_enums import WorkerType
 from shared.infrastructure.config.builder import WorkerBuilder
 from shared.infrastructure.config.registry import WorkerRegistry
@@ -67,7 +68,7 @@ def check_executor_health():
 )
 
 
-@app.task(bind=True)
+@worker_task(bind=True)
 def healthcheck(self):
     """Health check task for monitoring systems."""
     return {

workers/file_processing/structure_tool_task.py

@@ -23,6 +23,7 @@
 from typing import Any
 
 from file_processing.worker import app
+from queue_backend import worker_task
 from shared.enums.task_enums import TaskName
 from shared.infrastructure.context import StateStore
 
@@ -190,7 +191,7 @@ def _should_skip_extraction_for_smart_table(
 # -----------------------------------------------------------------------
 
 
-@app.task(bind=True, name=str(TaskName.EXECUTE_STRUCTURE_TOOL))
+@worker_task(bind=True, name=str(TaskName.EXECUTE_STRUCTURE_TOOL))
 def execute_structure_tool(self, params: dict) -> dict:
     """Execute structure tool as a Celery task.
 

workers/file_processing/tasks.py

@@ -10,6 +10,8 @@
 import time
 from typing import Any
 
+from queue_backend import worker_task
+
 # Import shared worker infrastructure
 from shared.api import InternalAPIClient
 
@@ -40,9 +42,6 @@
 )
 from shared.processing.files.processor import FileProcessor
 
-# Import manual review service with WorkflowUtil access
-from worker import app
-
 from unstract.core.data_models import (
     ExecutionStatus,
     FileBatchData,
@@ -255,7 +254,7 @@ def _process_file_batch_core(
     return _compile_batch_result(context)
 
 
-@app.task(
+@worker_task(
     bind=True,
     name=TaskName.PROCESS_FILE_BATCH,
     max_retries=0,  # Match Django backend pattern
@@ -1507,7 +1506,7 @@ def _process_file(
     )
 
 
-@app.task(
+@worker_task(
     bind=True,
     name=TaskName.PROCESS_FILE_BATCH_API,
     max_retries=0,  # Match Django backend
@@ -1993,7 +1992,7 @@ def resilient_executor(func):
 
 
 # Resilient file processor
-@app.task(bind=True)
+@worker_task(bind=True)
 @resilient_executor
 @with_execution_context
 def process_file_batch_resilient(
@@ -2030,7 +2029,7 @@ def process_file_batch_resilient(
 # Register the same task function with the old Django task names for compatibility
 
 
-@app.task(
+@worker_task(
     bind=True,
     name="workflow_manager.workflow_v2.file_execution_tasks.process_file_batch",
     max_retries=0,

workers/file_processing/worker.py

@@ -4,6 +4,7 @@
 Handles file uploads, text extraction, and processing workflows.
 """
 
+from queue_backend import worker_task
 from shared.enums.worker_enums import WorkerType
 from shared.infrastructure.config.builder import WorkerBuilder
 from shared.infrastructure.config.registry import WorkerRegistry
@@ -59,7 +60,7 @@ def check_file_processing_health():
 )
 
 
-@app.task(bind=True)
+@worker_task(bind=True)
 def healthcheck(self):
     """Health check task for monitoring systems."""
     return {

workers/general/tasks.py

@@ -7,7 +7,7 @@
 import time
 from typing import Any
 
-from celery import shared_task
+from queue_backend import worker_task
 from scheduler.tasks import execute_pipeline_task_v2
 
 # Import shared worker infrastructure using new structure
@@ -133,7 +133,7 @@ def _log_batch_creation_statistics(
         )
 
 
-@app.task(
+@worker_task(
     bind=True,
     name=TaskName.ASYNC_EXECUTE_BIN_GENERAL,
     autoretry_for=(Exception,),
@@ -1440,7 +1440,7 @@ def _validate_batch_data_integrity_dataclass(
         )
 
 
-@app.task(
+@worker_task(
     bind=True,
     name="async_execute_bin",
     autoretry_for=(Exception,),
@@ -1536,7 +1536,7 @@ def async_execute_bin(
 logger.info("✅ Registered scheduler tasks in general worker for backward compatibility")
 
 
-@shared_task(name="scheduler.tasks.execute_pipeline_task", bind=True)
+@worker_task(name="scheduler.tasks.execute_pipeline_task", bind=True)
 def execute_pipeline_task(
     self,
     workflow_id: Any,

workers/general/worker.py

@@ -4,6 +4,7 @@
 processing, and workflow orchestration.
 """
 
+from queue_backend import worker_task
 from shared.enums.worker_enums import WorkerType
 from shared.infrastructure.config.builder import WorkerBuilder
 from shared.infrastructure.config.registry import WorkerRegistry
@@ -59,7 +60,7 @@ def check_general_worker_health():
 )
 
 
-@app.task(bind=True)
+@worker_task(bind=True)
 def healthcheck(self):
     """Health check task for monitoring systems."""
     return {

workers/ide_callback/tasks.py

@@ -15,6 +15,7 @@
 from typing import Any
 
 from celery import current_app as app
+from queue_backend import worker_task
 from shared.clients.prompt_studio_client import PromptStudioAPIClient
 
 logger = logging.getLogger(__name__)
@@ -140,7 +141,7 @@ def _track_subscription_usage(org_id: str, run_id: str) -> None:
 # ------------------------------------------------------------------
 
 
-@app.task(name="ide_index_complete")
+@worker_task(name="ide_index_complete")
 def ide_index_complete(
     result_dict: dict[str, Any],
     callback_kwargs: dict[str, Any] | None = None,
@@ -299,7 +300,7 @@ def ide_index_complete(
         raise
 
 
-@app.task(name="ide_index_error")
+@worker_task(name="ide_index_error")
 def ide_index_error(
     failed_task_id: str,
     callback_kwargs: dict[str, Any] | None = None,
@@ -345,7 +346,7 @@ def ide_index_error(
         logger.exception("ide_index_error callback failed")
 
 
-@app.task(name="ide_prompt_complete")
+@worker_task(name="ide_prompt_complete")
 def ide_prompt_complete(
     result_dict: dict[str, Any],
     callback_kwargs: dict[str, Any] | None = None,
@@ -480,7 +481,7 @@ def ide_prompt_complete(
         raise
 
 
-@app.task(name="ide_prompt_error")
+@worker_task(name="ide_prompt_error")
 def ide_prompt_error(
     failed_task_id: str,
     callback_kwargs: dict[str, Any] | None = None,
@@ -533,7 +534,7 @@ def _get_extraction_client():
     return ExtractionAPIClient()
 
 
-@app.task(name="extraction_complete")
+@worker_task(name="extraction_complete")
 def extraction_complete(
     result_dict: dict[str, Any],
     callback_kwargs: dict[str, Any] | None = None,
@@ -625,7 +626,7 @@ def extraction_complete(
             raise
 
 
-@app.task(name="extraction_error")
+@worker_task(name="extraction_error")
 def extraction_error(
     failed_task_id: str,
     callback_kwargs: dict[str, Any] | None = None,