fix: improve WSS connectivity and host action visibility

tests/test_host_business_actions.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+from types import ModuleType, SimpleNamespace
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_host_status_exposes_json_business_actions(monkeypatch) -> None:
+    fake_host = SimpleNamespace(
+        devices_names={},
+        _online_devices=set(),
+        device_machine_names={},
+        _subscribed_topics=set(),
+        _action_clients={},
+        _action_value_mappings={
+            "CONDUCTIVITY_STATION": {
+                "station_status": {
+                    "type": "UniLabJsonCommand",
+                    "schema": {"description": "查询工站状态"},
+                },
+                "auto-close": {"type": "UniLabJsonCommand", "schema": {}},
+                "legacy_action": {"type": "EmptyIn", "schema": {}},
+            }
+        },
+        device_status={},
+        device_status_timestamps={},
+    )
+
+    config_module = ModuleType("unilabos.config.config")
+    config_module.BasicConfig = SimpleNamespace(is_host_mode=True)
+    host_node_module = ModuleType("unilabos.ros.nodes.presets.host_node")
+    host_node_module.HostNode = SimpleNamespace(get_instance=lambda _timeout: fake_host)
+    action_utils_module = ModuleType("unilabos.app.web.utils.action_utils")
+    action_utils_module.get_action_info = lambda client, full_name: {
+        "client": client,
+        "full_name": full_name,
+    }
+
+    monkeypatch.setitem(sys.modules, "unilabos.config.config", config_module)
+    monkeypatch.setitem(sys.modules, "unilabos.ros.nodes.presets.host_node", host_node_module)
+    monkeypatch.setitem(sys.modules, "unilabos.app.web.utils.action_utils", action_utils_module)
+
+    module_path = ROOT / "unilabos" / "app" / "web" / "utils" / "host_utils.py"
+    spec = importlib.util.spec_from_file_location("test_host_utils", module_path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+
+    info = module.get_host_node_info()
+
+    assert info["business_actions"] == {
+        "CONDUCTIVITY_STATION/station_status": {
+            "device_id": "CONDUCTIVITY_STATION",
+            "action_name": "station_status",
+            "action_type": "UniLabJsonCommand",
+            "description": "查询工站状态",
+        }
+    }

tests/test_ws_client_ssl.py

@@ -0,0 +1,19 @@
+from unittest.mock import patch
+
+from unilabos.app.ssl_utils import create_wss_ssl_context
+
+
+def test_wss_ssl_context_uses_certifi_ca_bundle() -> None:
+    expected_context = object()
+
+    with (
+        patch("unilabos.app.ssl_utils.certifi.where", return_value="certifi-ca.pem"),
+        patch(
+            "unilabos.app.ssl_utils.ssl.create_default_context",
+            return_value=expected_context,
+        ) as create_default_context,
+    ):
+        actual_context = create_wss_ssl_context()
+
+    assert actual_context is expected_context
+    create_default_context.assert_called_once_with(cafile="certifi-ca.pem")

unilabos/app/ssl_utils.py

@@ -0,0 +1,12 @@
+"""WebSocket TLS 工具。"""
+
+from __future__ import annotations
+
+import ssl
+
+import certifi
+
+
+def create_wss_ssl_context() -> ssl.SSLContext:
+    """使用 certifi CA 构建 WSS 上下文，避免 Windows 证书库异常。"""
+    return ssl.create_default_context(cafile=certifi.where())

unilabos/app/web/templates/status.html

@@ -201,6 +201,34 @@ <h4>已接纳动作:</h4>
     </table>
   </div>
 
+  <!-- UniLabJsonCommand 复用 command ROS action，业务注册单独展示。 -->
+  <div class="host-section">
+    <h3>
+      业务动作
+      <span class="count-badge">{{ host_node_info.business_actions|length }}</span>
+    </h3>
+    <table class="responsive-table">
+      <tr>
+        <th>设备 ID</th>
+        <th>动作名</th>
+        <th>类型</th>
+        <th>说明</th>
+      </tr>
+      {% for action_key, action_info in host_node_info.business_actions.items() %}
+      <tr>
+        <td>{{ action_info.device_id }}</td>
+        <td><code>{{ action_info.action_name }}</code></td>
+        <td>{{ action_info.action_type }}</td>
+        <td>{{ action_info.description }}</td>
+      </tr>
+      {% else %}
+      <tr>
+        <td colspan="4" class="empty-state">没有发现已注册的业务动作</td>
+      </tr>
+      {% endfor %}
+    </table>
+  </div>
+
   <!-- 主机已订阅的主题 -->
   <div class="host-section">
     <h3>

unilabos/app/web/utils/host_utils.py

@@ -21,7 +21,13 @@ def get_host_node_info() -> Dict[str, Any]:
     Returns:
         Dict: 包含主机节点信息的字典
     """
-    host_info = {"available": False, "devices": {}, "subscribed_topics": [], "action_clients": {}}
+    host_info = {
+        "available": False,
+        "devices": {},
+        "subscribed_topics": [],
+        "action_clients": {},
+        "business_actions": {},
+    }
     if not BasicConfig.is_host_mode:
         return host_info
     # 尝试获取HostNode实例，设置超时为0秒
@@ -44,6 +50,22 @@ def get_host_node_info() -> Dict[str, Any]:
     for action_id, client in host_node._action_clients.items():
         host_info["action_clients"][action_id] = get_action_info(client, full_name=action_id)
 
+    # @action 生成的 UniLabJsonCommand 复用底层 command ROS action，不会出现在
+    # _action_clients 中；单独展示业务映射，避免主机页误认为动作未注册。
+    for device_id, mappings in host_node._action_value_mappings.items():
+        for action_name, mapping in (mappings or {}).items():
+            action_type = str(mapping.get("type", ""))
+            if not action_type.startswith("UniLabJsonCommand") or action_name == "auto-close":
+                continue
+            schema = mapping.get("schema") or {}
+            key = f"{device_id}/{action_name}"
+            host_info["business_actions"][key] = {
+                "device_id": device_id,
+                "action_name": action_name,
+                "action_type": action_type,
+                "description": schema.get("description", ""),
+            }
+
     # 获取设备状态
     host_info["device_status"] = host_node.device_status
 

unilabos/app/ws_client.py

@@ -16,7 +16,7 @@
 import asyncio
 import traceback
 import websockets
-import ssl as ssl_module
+from websockets.exceptions import ConnectionClosed, InvalidStatus
 import copy
 from queue import Queue, Empty
 from dataclasses import dataclass, field
@@ -33,6 +33,7 @@
 from unilabos.app.communication import BaseCommunicationClient
 from unilabos.config.config import WSConfig, HTTPConfig, BasicConfig
 from unilabos.utils.log import get_comm_logger
+from unilabos.app.ssl_utils import create_wss_ssl_context
 
 # 服务端通信专用 logger：独立成文件(unilabos_data/logs/ws_comm_*.log)，
 # 全量 TRACE 落本地、微秒级时间戳 + 线程名，便于排查通信/queue 时序问题。
@@ -456,7 +457,7 @@ async def _connection_handler(self):
                 # 构建SSL上下文
                 ssl_context = None
                 if self.websocket_url.startswith("wss://"):
-                    ssl_context = ssl_module.create_default_context()
+                    ssl_context = create_wss_ssl_context()
 
                 ws_logger = logging.getLogger("websockets.client")
                 ws_logger.setLevel(logging.INFO)
@@ -504,13 +505,13 @@ async def _connection_handler(self):
                         except asyncio.CancelledError:
                             pass
 
-            except websockets.exceptions.ConnectionClosed:
+            except ConnectionClosed:
                 logger.warning("[MessageProcessor] 与服务端连接中断")
             except TimeoutError:
                 logger.warning(
                     f"[MessageProcessor] 与服务端连接通信超时 (已尝试 {self.reconnect_count + 1} 次)，请检查您的网络状况"
                 )
-            except websockets.exceptions.InvalidStatus as e:
+            except InvalidStatus as e:
                 logger.warning(
                     f"[MessageProcessor] 收到服务端注册码 {e.response.status_code}, 上一进程可能还未退出"
                 )

unilabos/utils/requirements.txt

@@ -1,6 +1,7 @@
 networkx
 typing_extensions
 websockets
+certifi
 msgcenterpy>=0.1.8
 orjson>=3.11
 opentrons_shared_data
@@ -17,4 +18,4 @@ pandas
 crcmod-plus
 pymodbus
 matplotlib
-pylibftdi
+pylibftdi