websocket add

backend/src/core/task_manager/event_bus.py

@@ -49,11 +49,17 @@ class EventBus:
     LOG_FLUSH_INTERVAL = 2.0
 
     # #region __init__ [TYPE Function]
-    # @BRIEF Initialize empty log buffer, subscriber registry, and start background flusher.
+    # @BRIEF Initialize empty log buffer, subscriber registry, status subscribers, and start background flusher.
     def __init__(self, log_persistence_service: TaskLogPersistenceService, loop: asyncio.AbstractEventLoop):
         self.log_persistence_service = log_persistence_service
         self.loop = loop
         self.subscribers: dict[str, list[asyncio.Queue]] = {}
+        # Per-task status subscribers (same pattern as log subscribers)
+        self._status_subscribers: dict[str, list[asyncio.Queue]] = {}
+        # Global task event subscribers (all tasks)
+        self._task_event_subscribers: list[asyncio.Queue] = []
+        # Maintenance event subscribers
+        self._maintenance_subscribers: list[asyncio.Queue] = []
         self._log_buffer: dict[str, list[LogEntry]] = {}
         self._log_buffer_lock = threading.Lock()
         self._flusher_stop_event = threading.Event()
@@ -178,6 +184,101 @@ class EventBus:
                 del self.subscribers[task_id]
     # #endregion unsubscribe_logs
 
+    # ── Task Status Subscribers ──
+
+    # #region subscribe_status [C:2] [TYPE Function] [SEMANTICS subscribe,status,realtime]
+    # @BRIEF Subscribes to real-time status updates for a specific task.
+    # @POST Returns an asyncio.Queue for Task status dicts.
+    async def subscribe_status(self, task_id: str) -> asyncio.Queue:
+        queue = asyncio.Queue()
+        if task_id not in self._status_subscribers:
+            self._status_subscribers[task_id] = []
+        self._status_subscribers[task_id].append(queue)
+        return queue
+    # #endregion subscribe_status
+
+    # #region unsubscribe_status [C:2] [TYPE Function] [SEMANTICS unsubscribe,status,remove]
+    # @BRIEF Unsubscribes from real-time status updates for a task.
+    # @POST Queue removed from status subscribers.
+    def unsubscribe_status(self, task_id: str, queue: asyncio.Queue):
+        if task_id in self._status_subscribers:
+            if queue in self._status_subscribers[task_id]:
+                self._status_subscribers[task_id].remove(queue)
+            if not self._status_subscribers[task_id]:
+                del self._status_subscribers[task_id]
+    # #endregion unsubscribe_status
+
+    # #region broadcast_status [C:2] [TYPE Function] [SEMANTICS broadcast,status,notify]
+    # @BRIEF Broadcast a task status update to all per-task and global subscribers.
+    # @POST Status event pushed to all matching queues.
+    def broadcast_status(self, task_id: str, task_dict: dict):
+        from ..logger import logger as _lg
+        event = {"type": "task_status", "task_id": task_id, "task": task_dict}
+        # Per-task subscribers
+        if task_id in self._status_subscribers:
+            for queue in self._status_subscribers[task_id]:
+                try:
+                    self.loop.call_soon_threadsafe(queue.put_nowait, event)
+                except Exception as _e:
+                    _lg.warning("Failed to send status to per-task subscriber", extra={"task_id": task_id, "error": str(_e)})
+        # Global task event subscribers
+        for queue in self._task_event_subscribers:
+            try:
+                self.loop.call_soon_threadsafe(queue.put_nowait, event)
+            except Exception as _e:
+                _lg.warning("Failed to send status to global subscriber", extra={"task_id": task_id, "error": str(_e)})
+    # #endregion broadcast_status
+
+    # ── Maintenance Event Subscribers ──
+
+    # #region subscribe_maintenance_events [C:2] [TYPE Function] [SEMANTICS subscribe,maintenance,events]
+    # @BRIEF Subscribes to maintenance event updates (created/ended/banner changes).
+    # @POST Returns an asyncio.Queue for maintenance event dicts.
+    async def subscribe_maintenance_events(self) -> asyncio.Queue:
+        queue = asyncio.Queue()
+        self._maintenance_subscribers.append(queue)
+        return queue
+    # #endregion subscribe_maintenance_events
+
+    # #region unsubscribe_maintenance_events [C:2] [TYPE Function] [SEMANTICS unsubscribe,maintenance,events]
+    # @BRIEF Unsubscribes from maintenance events.
+    # @POST Queue removed from maintenance subscribers.
+    def unsubscribe_maintenance_events(self, queue: asyncio.Queue):
+        if queue in self._maintenance_subscribers:
+            self._maintenance_subscribers.remove(queue)
+    # #endregion unsubscribe_maintenance_events
+
+    # #region broadcast_maintenance_event [C:2] [TYPE Function] [SEMANTICS broadcast,maintenance,notify]
+    # @BRIEF Broadcast a maintenance event to all subscribers.
+    # @POST Event pushed to all maintenance subscriber queues.
+    def broadcast_maintenance_event(self, event: dict):
+        from ..logger import logger as _lg
+        for queue in self._maintenance_subscribers:
+            try:
+                self.loop.call_soon_threadsafe(queue.put_nowait, event)
+            except Exception as _e:
+                _lg.warning("Failed to send maintenance event to subscriber", extra={"event_type": event.get("type"), "error": str(_e)})
+    # #endregion broadcast_maintenance_event
+
+    # ── Global Task Event Subscribers ──
+
+    # #region subscribe_task_events [C:2] [TYPE Function] [SEMANTICS subscribe,global,events]
+    # @BRIEF Subscribes to ALL task events (status changes) globally.
+    # @POST Returns an asyncio.Queue for task event dicts.
+    async def subscribe_task_events(self) -> asyncio.Queue:
+        queue = asyncio.Queue()
+        self._task_event_subscribers.append(queue)
+        return queue
+    # #endregion subscribe_task_events
+
+    # #region unsubscribe_task_events [C:2] [TYPE Function] [SEMANTICS unsubscribe,global,events]
+    # @BRIEF Unsubscribes from global task events.
+    # @POST Queue removed from global subscribers.
+    def unsubscribe_task_events(self, queue: asyncio.Queue):
+        if queue in self._task_event_subscribers:
+            self._task_event_subscribers.remove(queue)
+    # #endregion unsubscribe_task_events
+
     # #region get_task_logs [C:3] [TYPE Function] [SEMANTICS logs,read,persistence,backfill]
     # @BRIEF Retrieves logs for a specific task (from memory for running, persistence for completed).
     # @RELATION CALLS -> [EXT:method:TaskLogPersistenceService.get_logs]

backend/src/core/task_manager/lifecycle.py

@@ -34,6 +34,22 @@ from .context import TaskContext
 from .graph import TaskGraph
 from .models import Task, TaskStatus
 
+def _task_to_dict(task: Task) -> dict:
+    """Serialize a Task to a dict for WebSocket broadcast."""
+    return {
+        "id": task.id,
+        "plugin_id": task.plugin_id,
+        "status": task.status.value if isinstance(task.status, TaskStatus) else task.status,
+        "started_at": task.started_at.isoformat() if task.started_at else None,
+        "finished_at": task.finished_at.isoformat() if task.finished_at else None,
+        "user_id": task.user_id,
+        "params": task.params,
+        "result": task.result,
+        "input_required": task.input_required,
+        "input_request": task.input_request,
+        "error": str(task.result) if task.status == TaskStatus.FAILED and task.result else None,
+    }
+
 
 # #region JobLifecycle [C:5] [TYPE Class] [SEMANTICS task,lifecycle,execution,state,machine]
 # @BRIEF Encodes task creation, execution, pause/resume, and completion transitions for
@@ -129,6 +145,7 @@ class JobLifecycle:
             task.status = TaskStatus.RUNNING
             task.started_at = datetime.now(UTC)
             self.persistence_service.persist_task(task)
+            self._broadcast_task_status(task)
             if add_log_callback:
                 add_log_callback(
                     task_id, "INFO",
@@ -164,6 +181,7 @@ class JobLifecycle:
                         )
                 logger.info(f"Task {task_id} completed successfully")
                 task.status = TaskStatus.SUCCESS
+                self._broadcast_task_status(task)
                 if add_log_callback:
                     add_log_callback(
                         task_id, "INFO",
@@ -173,6 +191,7 @@ class JobLifecycle:
             except Exception as e:
                 logger.error(f"Task {task_id} failed: {e}")
                 task.status = TaskStatus.FAILED
+                self._broadcast_task_status(task)
                 if add_log_callback:
                     add_log_callback(
                         task_id, "ERROR",
@@ -214,6 +233,7 @@ class JobLifecycle:
             task.params.update(resolution_params)
             task.status = TaskStatus.RUNNING
             self.persistence_service.persist_task(task)
+            self._broadcast_task_status(task)
             self.graph.resolve_future(task_id, True)
     # #endregion resolve_task
 
@@ -228,6 +248,7 @@ class JobLifecycle:
                 return
             task.status = TaskStatus.AWAITING_MAPPING
             self.persistence_service.persist_task(task)
+            self._broadcast_task_status(task)
             future = self.loop.create_future()
             self.graph.create_future(task_id, future)
             try:
@@ -274,6 +295,7 @@ class JobLifecycle:
             task.input_required = True
             task.input_request = input_request
             self.persistence_service.persist_task(task)
+            self._broadcast_task_status(task)
             if add_log_callback:
                 add_log_callback(
                     task_id, "INFO",
@@ -308,6 +330,7 @@ class JobLifecycle:
             task.input_request = None
             task.status = TaskStatus.RUNNING
             self.persistence_service.persist_task(task)
+            self._broadcast_task_status(task)
             if add_log_callback:
                 add_log_callback(
                     task_id, "INFO",
@@ -337,6 +360,13 @@ class JobLifecycle:
                 del self._dataset_subscribers[env_id]
     # #endregion unsubscribe_dataset_events
 
+    # #region _broadcast_task_status [C:2] [TYPE Function] [SEMANTICS broadcast,status,websocket]
+    # @BRIEF Broadcast task status update to all subscribers via EventBus.
+    # @POST EventBus.broadcast_status called with serialized task dict.
+    def _broadcast_task_status(self, task: Task):
+        self.event_bus.broadcast_status(task.id, _task_to_dict(task))
+    # #endregion _broadcast_task_status
+
     # #region _broadcast_dataset_updated [C:3] [TYPE Function] [SEMANTICS dataset,broadcast,event,websocket]
     # @BRIEF Broadcast dataset.updated event to all subscribers for a given environment.
     def _broadcast_dataset_updated(self, env_id: str, dataset_ids: list[int]):

backend/src/core/task_manager/manager.py

@@ -261,6 +261,32 @@ class TaskManager:
         self.event_bus.unsubscribe_logs(task_id, queue)
     # #endregion unsubscribe_logs
 
+    # ── Status subscribers ──
+
+    # #region subscribe_status [TYPE Function] [C:2]
+    # @BRIEF Subscribes to real-time status updates for a task.
+    async def subscribe_status(self, task_id: str) -> asyncio.Queue:
+        return await self.event_bus.subscribe_status(task_id)
+    # #endregion subscribe_status
+
+    # #region unsubscribe_status [TYPE Function] [C:2]
+    # @BRIEF Unsubscribes from status updates for a task.
+    def unsubscribe_status(self, task_id: str, queue: asyncio.Queue):
+        self.event_bus.unsubscribe_status(task_id, queue)
+    # #endregion unsubscribe_status
+
+    # #region subscribe_task_events [TYPE Function] [C:2]
+    # @BRIEF Subscribes to global task events (all task status changes).
+    async def subscribe_task_events(self) -> asyncio.Queue:
+        return await self.event_bus.subscribe_task_events()
+    # #endregion subscribe_task_events
+
+    # #region unsubscribe_task_events [TYPE Function] [C:2]
+    # @BRIEF Unsubscribes from global task events.
+    def unsubscribe_task_events(self, queue: asyncio.Queue):
+        self.event_bus.unsubscribe_task_events(queue)
+    # #endregion unsubscribe_task_events
+
     # ── Lifecycle delegates to JobLifecycle ──
 
     # #region create_task [TYPE Function] [C:4]
@@ -318,6 +344,26 @@ class TaskManager:
         )
     # #endregion resume_task_with_password
 
+    # ── Maintenance event delegates to EventBus ──
+
+    # #region subscribe_maintenance_events [TYPE Function] [C:2]
+    # @BRIEF Subscribes to global maintenance events.
+    async def subscribe_maintenance_events(self) -> asyncio.Queue:
+        return await self.event_bus.subscribe_maintenance_events()
+    # #endregion subscribe_maintenance_events
+
+    # #region unsubscribe_maintenance_events [TYPE Function] [C:2]
+    # @BRIEF Unsubscribes from global maintenance events.
+    def unsubscribe_maintenance_events(self, queue: asyncio.Queue):
+        self.event_bus.unsubscribe_maintenance_events(queue)
+    # #endregion unsubscribe_maintenance_events
+
+    # #region broadcast_maintenance_event [TYPE Function] [C:2]
+    # @BRIEF Broadcast a maintenance event to all subscribers.
+    def broadcast_maintenance_event(self, event: dict):
+        self.event_bus.broadcast_maintenance_event(event)
+    # #endregion broadcast_maintenance_event
+
     # ── Dataset event delegates to JobLifecycle ──
 
     # #region subscribe_dataset_events [TYPE Function] [C:2]