's'

src/web_process_backend.py

@@ -308,6 +308,42 @@ class WebProcessBackend:
         if self._use_pm2 is None:
             await self.refresh_mode()
 
+    async def online_process_names(self) -> list[str]:
+        """当前在线的 PM2 / 本地注册进程名（用于 TikTok·OBS 等 HDMI 链路互斥）。"""
+        await self.ensure_mode()
+        if not self._use_pm2:
+            data = self._local._read_state()
+            names: list[str] = []
+            for name, row in (data.get("processes") or {}).items():
+                try:
+                    pid = int((row or {}).get("pid") or 0)
+                except (TypeError, ValueError):
+                    continue
+                if self._local._pid_alive(pid):
+                    names.append(str(name))
+            return sorted(names)
+        code, text = await _shell_out("pm2 jlist", timeout=20.0)
+        if code != 0 or not text.strip():
+            return []
+        try:
+            items = json.loads(text)
+        except json.JSONDecodeError:
+            return []
+        if not isinstance(items, list):
+            return []
+        out: list[str] = []
+        for it in items:
+            if not isinstance(it, dict):
+                continue
+            name = it.get("name")
+            if not name:
+                continue
+            env = it.get("pm2_env") if isinstance(it.get("pm2_env"), dict) else {}
+            status = str(env.get("status") or "").lower()
+            if status in ("online", "launching"):
+                out.append(str(name))
+        return out
+
     async def start(self, process: str, script: str) -> str:
         await self.ensure_mode()
         if self._use_pm2: