85: Refactor pipeline service - simplified

Author: MatejFrnka

service/src/ai_document_plugin_service/api/routes.py

@@ -2,7 +2,6 @@
 
 import fastapi
 
-from ai_document_plugin_service.ai.common.config import LLMConfig
 from ai_document_plugin_service.api.auth import verify_authenticated
 from ai_document_plugin_service.api.jwt import extract_identity_from_token
 from ai_document_plugin_service.api.types import (
@@ -100,19 +99,11 @@ async def start_pipeline(
     run_id = str(uuid4())
     pipeline.enqueue_pipeline_job(
         run_id,
-        payload.questionnaire_uuid,
-        payload.template_uuid,
+        payload,
         template['title'],
         user_uuid,
         tenant_uuid,
-        auth.token,
-        auth.api_url,
-        LLMConfig(
-            model=payload.llm_model,
-            api_key=payload.llm_api_key,
-            api_url=payload.llm_api_url,
-            parallel_workers=payload.llm_max_workers,
-        ),
+        auth,
         config,
     )
     return _model_from_fields(
@@ -139,5 +130,4 @@ def get_pipeline_status(run_id: str, pipeline: PipelineServiceDI) -> PipelineSta
 async def save_pipeline_result(
     run_id: str, save_request: PipelineSaveRequest, pipeline: PipelineServiceDI
 ) -> PipelineStatusResponse:
-
     return await pipeline.update_pipeline_result(run_id, save_request)

service/src/ai_document_plugin_service/service/pipeline_queue_manager.py

@@ -21,7 +21,8 @@ def format_queue_progress(jobs_ahead: int) -> str:
 class PipelineQueueManager:
     """FIFO pipeline job queue running coroutines on a dedicated event loop.
 
-    Jobs are coroutines scheduled onto a single background event loop and gated by an
+    Jobs are coroutines scheduled onto a single background event loop and gated by a
+    semaphore so at most ``max_concurrent_jobs`` run at once.
     """
 
     def __init__(self, max_concurrent_jobs: int) -> None:
@@ -56,10 +57,8 @@ def progress_message(self, run_id: str) -> str | None:
 
     def remove(self, run_id: str) -> None:
         with self._order_lock:
-            try:
+            if run_id in self._order:
                 self._order.remove(run_id)
-            except ValueError:
-                return
 
     def _jobs_waiting_ahead(self, run_id: str) -> int | None:
         with self._order_lock: