feat: worker hot-swap with health check, staggered startup and live dashboard

- Staggered worker startup with 3s intervals to prevent resource exhaustion
- Health check loop (30s interval, 3 strikes) with auto-restart for crashed workers
- Dynamic add/remove/start/stop individual workers at runtime
- WebSocket push for worker_snapshot and worker_status events
- Dashboard real-time worker status cards (running/stopped/crashed/rate_limited)
- Graceful stop waits for active requests before killing worker
- Startup loop respects stop signal for immediate cancellation

Author: Mag1cFall

README.md

@@ -21,7 +21,7 @@
   🎤 Gemini 2.5 TTS 语音生成
 </p>
 
-<img src="docs/img/demo.gif" alt="Demo GIF" width="100%" />
+<!-- <img src="docs/img/demo.gif" alt="Demo GIF" width="100%" /> -->
 
 <!-- <p align="center">
   <img src="docs/img/多worker并发和媒体模型支援.png" alt="多Worker并发与媒体模型支援" width="80%" />

README_en.md

@@ -21,7 +21,7 @@
   🎤 Gemini 2.5 TTS Speech Synthesis
 </p>
 
-<img src="docs/img/demo.gif" alt="Demo GIF" width="100%" />
+<!-- <img src="docs/img/demo.gif" alt="Demo GIF" width="100%" /> -->
 
 <!-- <p align="center">
   <img src="docs/img/多worker并发和媒体模型支援.png" alt="Multi-Worker Concurrency & Media Model Support" width="80%" />

src/manager/app.py

@@ -31,11 +31,24 @@ async def lifespan(app: FastAPI):
     manager.loop = asyncio.get_running_loop()
     config = manager.load_config()
     manager._log_enabled = config.get("log_enabled", True)
+    health_task = None
     if WORKER_POOL_AVAILABLE and worker_pool is not None:
         worker_pool.init_from_config()
+        worker_pool.configure_runtime(config)
+        worker_pool.register_status_listener(manager.handle_worker_status_event)
+        worker_pool.register_process_listener(manager.handle_worker_process_started)
+        health_task = asyncio.create_task(worker_pool.health_check_loop())
     yield
+    if health_task is not None:
+        health_task.cancel()
+        try:
+            await health_task
+        except asyncio.CancelledError:
+            pass
     if manager.process or manager.worker_processes:
         manager.stop_service()
+    if WORKER_POOL_AVAILABLE and worker_pool is not None:
+        await worker_pool.close()
     manager.loop = None
 
 

src/manager/routes/control.py

@@ -10,8 +10,9 @@
 
 @router.post("/start")
 async def start_service(config: Dict[str, Any] = Body(...)):
-    success, message = manager.start_service(config)
+    success, message = await manager.start_service(config)
     await manager.broadcast_status()
+    await manager.broadcast_worker_snapshot()
     if not success:
         raise HTTPException(status_code=400, detail=message)
     return {"success": True, "message": message}
@@ -21,6 +22,7 @@ async def start_service(config: Dict[str, Any] = Body(...)):
 async def stop_service():
     success, message = manager.stop_service()
     await manager.broadcast_status()
+    await manager.broadcast_worker_snapshot()
     if not success:
         raise HTTPException(status_code=500, detail=message)
     return {"success": True, "message": message}

src/manager/routes/websocket.py

@@ -22,6 +22,7 @@ async def websocket_logs(websocket: WebSocket):
                 }
             )
         )
+        await manager.broadcast_worker_snapshot()
         while True:
             await websocket.receive_text()
     except WebSocketDisconnect:

src/manager/routes/workers.py

@@ -7,7 +7,7 @@
 except ImportError:
     from config.settings import SAVED_AUTH_DIR
 
-from ..service import WORKER_POOL_AVAILABLE, worker_pool
+from ..service import WORKER_POOL_AVAILABLE, manager, worker_pool
 
 
 router = APIRouter(prefix="/api/workers", tags=["Workers"])
@@ -54,6 +54,12 @@ async def add_worker(profile: str = Body(..., embed=True)):
     )
     pool.workers[worker_id] = worker
     pool.save_config()
+    if manager.is_worker_mode and manager.service_status == "running":
+        pool.configure_runtime(manager.load_config())
+        success, message = pool.start_worker(worker_id)
+        if not success:
+            raise HTTPException(status_code=500, detail=message)
+    await manager.broadcast_worker_snapshot()
     return {"success": True, "worker": worker.to_dict()}
 
 
@@ -65,11 +71,16 @@ async def remove_worker(worker_id: str):
         raise HTTPException(status_code=404, detail="Worker not found")
 
     worker = pool.workers[worker_id]
+    process = worker.process
     if worker.status == "running":
-        pool.stop_worker(worker_id)
+        success, message = await pool.stop_worker(worker_id)
+        if not success:
+            raise HTTPException(status_code=500, detail=message)
+        manager.unregister_worker_process(process)
 
     del pool.workers[worker_id]
     pool.save_config()
+    await manager.broadcast_worker_snapshot()
     return {"success": True}
 
 
@@ -86,6 +97,7 @@ async def list_workers():
 async def init_workers():
     pool = _require_worker_pool()
     pool.init_from_config()
+    await manager.broadcast_worker_snapshot()
     return {"success": True, "count": len(pool.workers)}
 
 
@@ -94,6 +106,7 @@ async def save_workers_config():
     pool = _require_worker_pool()
     try:
         pool.save_config()
+        await manager.broadcast_worker_snapshot()
         return {"success": True, "count": len(pool.workers)}
     except Exception as exc:
         return {"success": False, "error": str(exc)}
@@ -124,44 +137,57 @@ async def get_next_available_worker(model: str = ""):
 async def mark_worker_rate_limited(worker_id: str, model: str = Body(..., embed=True)):
     pool = _require_worker_pool()
     pool.mark_rate_limited(worker_id, model)
+    await manager.broadcast_worker_snapshot()
     return {"success": True}
 
 
 @router.post("/{worker_id}/start")
 async def start_worker_api(worker_id: str):
     pool = _require_worker_pool()
+    pool.configure_runtime(manager.load_config())
     success, message = pool.start_worker(worker_id)
     if not success:
         raise HTTPException(status_code=400, detail=message)
+    await manager.broadcast_worker_snapshot()
     return {"success": True, "message": message}
 
 
 @router.post("/{worker_id}/stop")
 async def stop_worker_api(worker_id: str):
     pool = _require_worker_pool()
-    success, message = pool.stop_worker(worker_id)
+    process = pool.workers.get(worker_id).process if worker_id in pool.workers else None
+    success, message = await pool.stop_worker(worker_id)
     if not success:
         raise HTTPException(status_code=400, detail=message)
+    manager.unregister_worker_process(process)
+    await manager.broadcast_worker_snapshot()
     return {"success": True, "message": message}
 
 
 @router.post("/{worker_id}/clear-limits")
 async def clear_worker_limits(worker_id: str):
     pool = _require_worker_pool()
     if pool.clear_rate_limits(worker_id):
+        await manager.broadcast_worker_snapshot()
         return {"success": True}
     raise HTTPException(status_code=404, detail="Worker not found")
 
 
 @router.post("/start-all")
 async def start_all_workers():
     pool = _require_worker_pool()
-    pool.start_all()
+    pool.configure_runtime(manager.load_config())
+    await pool.start_all()
+    await manager.broadcast_worker_snapshot()
     return {"success": True}
 
 
 @router.post("/stop-all")
 async def stop_all_workers():
     pool = _require_worker_pool()
-    pool.stop_all()
+    processes = [worker.process for worker in pool.workers.values() if worker.process]
+    await pool.stop_all()
+    for process in processes:
+        manager.unregister_worker_process(process)
+    await manager.broadcast_worker_snapshot()
     return {"success": True}