feat(task-queue): implement task queue management with settings and UI integration

services/task_queue.py

@@ -32,11 +32,16 @@ class Task(BaseModel):
     meta: Dict[str, Any] | None = None
 
 
+_SENTINEL = object()
+
+
 class TaskQueueService:
     def __init__(self):
-        self._queue = asyncio.Queue()
+        self._queue: asyncio.Queue[Task | object] = asyncio.Queue()
         self._tasks: Dict[str, Task] = {}
-        self._worker_task: asyncio.Task | None = None
+        self._worker_tasks: list[asyncio.Task] = []
+        self._concurrency: int = 1
+        self._worker_seq: int = 0
 
     async def add_task(self, name: str, task_info: Dict[str, Any]) -> Task:
         task = Task(name=name, task_info=task_info)
@@ -83,7 +88,7 @@ class TaskQueueService:
                     overwrite=params.get("overwrite", False),
                 )
                 task.result = result
-            elif task.name == "automation_task":
+            elif task.name == "automation_task" or self._is_processor_task(task.name):
                 from models.database import AutomationTask
                 from services.processors.registry import get as get_processor
                 from services.virtual_fs import read_file, write_file
@@ -92,9 +97,21 @@ class TaskQueueService:
                 auto_task = await AutomationTask.get(id=params["task_id"])
                 path = params["path"]
 
-                processor = get_processor(auto_task.processor_type)
+                processor_type = auto_task.processor_type if task.name == "automation_task" else task.name
+                processor = get_processor(processor_type)
                 if not processor:
-                    raise ValueError(f"Processor {auto_task.processor_type} not found for task {auto_task.id}")
+                    raise ValueError(f"Processor {processor_type} not found for task {auto_task.id}")
+
+                if processor_type != auto_task.processor_type:
+                    await LogService.warning(
+                        "task_queue",
+                        "Processor type mismatch; falling back to stored type",
+                        {"task_id": auto_task.id, "expected": auto_task.processor_type, "got": processor_type},
+                    )
+                    processor_type = auto_task.processor_type
+                    processor = get_processor(processor_type)
+                    if not processor:
+                        raise ValueError(f"Processor {processor_type} not found for task {auto_task.id}")
 
                 file_content = await read_file(path)
                 result = await processor.process(file_content, path, auto_task.processor_config)
@@ -124,35 +141,88 @@ class TaskQueueService:
             task.error = str(e)
             await LogService.error("task_queue", f"Task {task.name} ({task.id}) failed: {e}", {"task_id": task.id, "name": task.name})
 
-    async def worker(self):
-        await LogService.info("task_queue", "Task worker started")
-        while True:
-            try:
-                task = await self._queue.get()
-                await self._execute_task(task)
-            except asyncio.CancelledError:
-                await LogService.info("task_queue", "Task worker stopped")
-                break
-            except Exception as e:
-                await LogService.error("task_queue", f"Error in task worker: {e}", exc_info=True)
-            finally:
-                self._queue.task_done()
+    def _cleanup_workers(self):
+        self._worker_tasks = [task for task in self._worker_tasks if not task.done()]
 
-    async def start_worker(self):
-        if self._worker_task is None or self._worker_task.done():
-            self._worker_task = asyncio.create_task(self.worker())
-            await LogService.info("task_queue", "Task worker created.")
+    def _is_processor_task(self, task_name: str) -> bool:
+        try:
+            from services.processors.registry import get as get_processor
+
+            return get_processor(task_name) is not None
+        except Exception:
+            return False
+
+    async def _ensure_worker_count(self):
+        self._cleanup_workers()
+        current = len(self._worker_tasks)
+        if current < self._concurrency:
+            for _ in range(self._concurrency - current):
+                self._worker_seq += 1
+                worker_id = self._worker_seq
+                worker_task = asyncio.create_task(self._worker_loop(worker_id))
+                self._worker_tasks.append(worker_task)
+            await LogService.info("task_queue", "Task workers adjusted", {"active_workers": len(self._worker_tasks), "target": self._concurrency})
+        elif current > self._concurrency:
+            for _ in range(current - self._concurrency):
+                await self._queue.put(_SENTINEL)
+            await LogService.info("task_queue", "Task workers scaling down", {"active_workers": len(self._worker_tasks), "target": self._concurrency})
+
+    async def _worker_loop(self, worker_id: int):
+        current_task = asyncio.current_task()
+        await LogService.info("task_queue", f"Worker {worker_id} started")
+        try:
+            while True:
+                job = await self._queue.get()
+                if job is _SENTINEL:
+                    self._queue.task_done()
+                    break
+                try:
+                    await self._execute_task(job)
+                except Exception as e:
+                    await LogService.error(
+                        "task_queue",
+                        f"Error executing task {job.id}: {e}",
+                        {"task_id": job.id, "name": job.name},
+                    )
+                finally:
+                    self._queue.task_done()
+        finally:
+            if current_task in self._worker_tasks:
+                self._worker_tasks.remove(current_task)  # type: ignore[arg-type]
+            await LogService.info("task_queue", f"Worker {worker_id} stopped")
+
+    async def start_worker(self, concurrency: int | None = None):
+        if concurrency is None:
+            from services.config import ConfigCenter
+
+            stored_value = await ConfigCenter.get("TASK_QUEUE_CONCURRENCY", self._concurrency)
+            try:
+                concurrency = int(stored_value)
+            except (TypeError, ValueError):
+                concurrency = self._concurrency
+        await self.set_concurrency(concurrency)
+
+    async def set_concurrency(self, value: int):
+        value = max(1, int(value))
+        if value != self._concurrency:
+            self._concurrency = value
+        await self._ensure_worker_count()
 
     async def stop_worker(self):
-        if self._worker_task and not self._worker_task.done():
-            self._worker_task.cancel()
-            try:
-                await self._worker_task
-            except asyncio.CancelledError:
-                pass
-            finally:
-                self._worker_task = None
-                await LogService.info("task_queue", "Task worker has been stopped.")
+        self._cleanup_workers()
+        for _ in range(len(self._worker_tasks)):
+            await self._queue.put(_SENTINEL)
+        if self._worker_tasks:
+            await asyncio.gather(*self._worker_tasks, return_exceptions=True)
+        self._worker_tasks.clear()
+        await LogService.info("task_queue", "Task workers have been stopped.")
+
+    def get_concurrency(self) -> int:
+        return self._concurrency
+
+    def get_active_worker_count(self) -> int:
+        self._cleanup_workers()
+        return len(self._worker_tasks)
 
 
 task_queue_service = TaskQueueService()

services/tasks.py

@@ -25,11 +25,11 @@ class TaskService:
 
     async def execute(self, task: AutomationTask, path: str):
         await task_queue_service.add_task(
-            "automation_task",
+            task.processor_type,
             {
                 "task_id": task.id,
                 "path": path,
             },
         )
 
-task_service = TaskService()
+task_service = TaskService()