diff --git a/src/plugin_runtime/host/workflow_executor.py b/src/plugin_runtime/host/workflow_executor.py
index e191eebb..ffc4e995 100644
--- a/src/plugin_runtime/host/workflow_executor.py
+++ b/src/plugin_runtime/host/workflow_executor.py
@@ -345,7 +345,8 @@ class WorkflowExecutor:
     ) -> None:
         """Non-blocking hook 调用，只读，忽略结果。"""
         timeout_ms = step.metadata.get("timeout_ms", 0)
-        timeout_sec = timeout_ms / 1000 if timeout_ms > 0 else None
+        # 使用 hook 声明的超时，但无声明时回退到全局安全阀，防止 task 泄漏
+        timeout_sec = timeout_ms / 1000 if timeout_ms > 0 else _get_blocking_timeout()
 
         try:
             coro = invoke_fn(step.plugin_id, step.name, {
@@ -354,10 +355,9 @@ class WorkflowExecutor:
                 "message": message,
                 "stage_outputs": ctx.stage_outputs,
             })
-            if timeout_sec:
-                await asyncio.wait_for(coro, timeout=timeout_sec)
-            else:
-                await coro
+            await asyncio.wait_for(coro, timeout=timeout_sec)
+        except asyncio.TimeoutError:
+            logger.warning(f"[{ctx.trace_id}] non-blocking hook {step.full_name} 超时 ({timeout_sec}s)")
         except Exception as e:
             logger.debug(f"[{ctx.trace_id}] non-blocking hook {step.full_name}: {e}")
 
diff --git a/src/plugin_runtime/integration.py b/src/plugin_runtime/integration.py
index dcd04629..b306acb7 100644
--- a/src/plugin_runtime/integration.py
+++ b/src/plugin_runtime/integration.py
@@ -82,20 +82,24 @@ class PluginRuntimeManager:
             return
 
         # 从配置读取自定义 IPC socket 路径（留空则自动生成）
-        socket_path = _cfg.ipc_socket_path or None
+        socket_path_base = _cfg.ipc_socket_path or None
+
+        # 当用户指定了自定义路径时，为两个 Supervisor 添加后缀以避免 UDS 冲突
+        builtin_socket = f"{socket_path_base}-builtin" if socket_path_base else None
+        thirdparty_socket = f"{socket_path_base}-thirdparty" if socket_path_base else None
 
         # 创建两个 Supervisor，各自拥有独立的 socket / Runner 子进程
         if builtin_dirs:
             self._builtin_supervisor = PluginSupervisor(
                 plugin_dirs=builtin_dirs,
-                socket_path=socket_path,
+                socket_path=builtin_socket,
             )
             self._register_capability_impls(self._builtin_supervisor)
 
         if thirdparty_dirs:
             self._thirdparty_supervisor = PluginSupervisor(
                 plugin_dirs=thirdparty_dirs,
-                socket_path=socket_path,
+                socket_path=thirdparty_socket,
             )
             self._register_capability_impls(self._thirdparty_supervisor)