fix(connectors): fall back to user-level npm prefix when global dir not writable

src/octop/infra/connectors/gateway/cli_install.py

@@ -2,14 +2,20 @@
 
 from __future__ import annotations
 
+import contextlib
+import os
 import re
 import shutil
 import subprocess
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any
 
 _INSTALL_TIMEOUT_S = 300.0
 _VERSION_RE = re.compile(r"(\d+\.\d+\.\d+(?:[-+][\w.]+)?)")
+# fnOS / 容器里 Octop 常以非 root 用户运行，npm 全局目录（/usr/local）不可写，
+# 此时降级到用户级目录安装，目录名沿用 npm 官方推荐的 ~/.npm-global。
+_NPM_USER_PREFIX_NAME = ".npm-global"
 
 
 @dataclass(frozen=True)
@@ -50,7 +56,62 @@ def get_cli_install_spec(kind: str) -> CliInstallSpec | None:
     return _SPECS.get(kind)
 
 
+def _prefix_bin_dir(prefix: str) -> str:
+    # npm 在 POSIX 下把全局 bin 放在 <prefix>/bin，Windows 下放在 <prefix> 根目录。
+    return prefix if os.name == "nt" else str(Path(prefix) / "bin")
+
+
+def _npm_prefix_info(npm: str) -> tuple[str, str]:
+    """Return ``(prefix, bin_dir)`` reported by ``npm config get prefix``."""
+    try:
+        completed = subprocess.run(
+            [npm, "config", "get", "prefix"],
+            capture_output=True,
+            text=True,
+            timeout=15.0,
+            check=False,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return "", ""
+    prefix = (completed.stdout or "").strip()
+    if not prefix:
+        return "", ""
+    return prefix, _prefix_bin_dir(prefix)
+
+
+def _user_npm_prefix() -> tuple[str, str]:
+    """Return ``(prefix, bin_dir)`` for the user-level npm global directory."""
+    prefix = os.path.join(os.path.expanduser("~"), _NPM_USER_PREFIX_NAME)
+    return prefix, _prefix_bin_dir(prefix)
+
+
+def _prefix_writable(prefix: str) -> bool:
+    if not prefix:
+        return False
+    try:
+        return os.access(prefix, os.W_OK)
+    except OSError:
+        return False
+
+
+def ensure_cli_path() -> str:
+    """Prepend the user-level npm global bin dir to the in-process PATH.
+
+    Octop 在 fnOS 上常以非 root 用户运行，``/usr/local`` 下的 npm 全局目录
+    不可写，安装会降级到用户级目录（~/.npm-global）。这里确保该 bin 目录
+    进入进程 PATH，使 ``shutil.which`` 与后续 CLI 子进程调用都能找到命令。
+    目录不存在时不做任何修改，返回 bin 目录（可能为空串）。
+    """
+    _, bin_dir = _user_npm_prefix()
+    if bin_dir and os.path.isdir(bin_dir):
+        current = os.environ.get("PATH", "")
+        if bin_dir not in [part for part in current.split(os.pathsep) if part]:
+            os.environ["PATH"] = bin_dir + os.pathsep + current
+    return bin_dir
+
+
 def cli_install_status(kind: str) -> dict[str, Any]:
+    ensure_cli_path()
     spec = get_cli_install_spec(kind)
     if spec is None:
         raise ValueError(f"kind {kind!r} does not support CLI install")
@@ -86,9 +147,19 @@ def install_connector_cli(kind: str) -> dict[str, Any]:
             f"未找到 npm，请先在 Octop 主机安装 Node.js，然后执行：{status['install_command']}",
         )
 
+    # npm 全局目录（默认 /usr/local）不可写时（fnOS/容器内非 root 用户），
+    # 自动降级到用户级目录 ~/.npm-global 安装，避免 EACCES 导致安装失败。
+    prefix, _ = _npm_prefix_info(npm)
+    install_args = [npm, "install", "-g"]
+    if not _prefix_writable(prefix):
+        user_prefix, _user_bin = _user_npm_prefix()
+        with contextlib.suppress(OSError):
+            os.makedirs(user_prefix, exist_ok=True)
+        install_args += ["--prefix", user_prefix]
+
     try:
         completed = subprocess.run(
-            [npm, "install", "-g", status["npm_package"]],
+            install_args + [status["npm_package"]],
             capture_output=True,
             text=True,
             timeout=_INSTALL_TIMEOUT_S,
@@ -109,9 +180,16 @@ def install_connector_cli(kind: str) -> dict[str, Any]:
         msg = f"npm install 失败（exit {completed.returncode}）"
         if detail:
             msg = f"{msg}：{detail}"
-        msg = f"{msg}。请在主机手动执行：{status['install_command']}"
+        if "--prefix" in install_args:
+            msg = f"{msg}。已尝试写入用户级目录（~/.npm-global）仍失败，请在主机手动执行：{status['install_command']}"
+        else:
+            msg = f"{msg}。请在主机手动执行：{status['install_command']}"
         return _fail(status, msg)
 
+    # 降级安装到用户级目录后，把该 bin 目录加入进程 PATH，使状态检测与后续 CLI 调用可见。
+    if "--prefix" in install_args:
+        ensure_cli_path()
+
     refreshed = cli_install_status(kind)
     if not refreshed["installed"]:
         return _fail(

src/octop/infra/server.py

@@ -145,6 +145,11 @@ async def start(self) -> None:
         if self._started:
             return
         self.paths.ensure_root()
+        # fnOS/容器内非 root 用户场景：把用户级 npm 全局 bin 目录（~/.npm-global）
+        # 纳入进程 PATH，保证连接器 CLI（wecom-cli / lark-cli）可被检测与调用。
+        from octop.infra.connectors.gateway.cli_install import ensure_cli_path  # noqa: PLC0415
+
+        ensure_cli_path()
         from octop.infra.utils.env_file import apply_env_file, env_file_path  # noqa: PLC0415
 
         apply_env_file(env_file_path(self.paths.root))

tests/unit/connectors/test_cli_install.py

@@ -44,7 +44,7 @@ def _which(name: str) -> str | None:
     assert out["doc_url"]
 
 
-def test_install_runs_npm(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_install_runs_npm(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None:
     calls: list[list[str]] = []
     state = {"installed": False}
 
@@ -58,6 +58,14 @@ def _which(name: str) -> str | None:
     def _run(argv: list[str], **kwargs: Any) -> Any:
         del kwargs
         calls.append(list(argv))
+        if argv[1:3] == ["config", "get"]:
+
+            class _Cfg:
+                returncode = 0
+                stdout = str(tmp_path)
+                stderr = ""
+
+            return _Cfg()
         state["installed"] = True
 
         class _Completed:
@@ -74,4 +82,78 @@ class _Completed:
     assert out["ok"] is True
     assert out["already_installed"] is False
     assert out["version"] == "9.9.9"
-    assert calls and calls[0][:3] == [fake_bin_path("npm"), "install", "-g"]
+    install_call = [c for c in calls if c[1:3] == ["install", "-g"]][0]
+    assert install_call[:3] == [fake_bin_path("npm"), "install", "-g"]
+    # 全局目录可写时保持原行为：不加 --prefix 降级参数
+    assert "--prefix" not in install_call
+
+
+def test_install_degrades_to_user_prefix_when_global_not_writable(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    """npm 全局目录不可写（fnOS/容器内非 root 用户）时降级到 ~/.npm-global。"""
+    calls: list[list[str]] = []
+    state = {"installed": False}
+    home = tmp_path / "home"
+    monkeypatch.setenv("HOME", str(home))
+    # Windows 上 os.path.expanduser("~") 读 USERPROFILE，需一并覆盖以跨平台
+    monkeypatch.setenv("USERPROFILE", str(home))
+
+    def _which(name: str) -> str | None:
+        if name == "npm":
+            return fake_bin_path("npm")
+        if name in ("lark-cli", "wecom-cli"):
+            return fake_bin_path(name) if state["installed"] else None
+        return None
+
+    def _run(argv: list[str], **kwargs: Any) -> Any:
+        del kwargs
+        calls.append(list(argv))
+        if argv[1:3] == ["config", "get"]:
+
+            class _Cfg:
+                returncode = 0
+                stdout = "/usr/local"
+                stderr = ""
+
+            return _Cfg()
+        state["installed"] = True
+
+        class _Completed:
+            returncode = 0
+            stdout = "added 1 package"
+            stderr = ""
+
+        return _Completed()
+
+    monkeypatch.setattr(cli_install.shutil, "which", _which)
+    monkeypatch.setattr(cli_install.subprocess, "run", _run)
+    # 模拟 /usr/local 不可写（非 root 用户）
+    monkeypatch.setattr(cli_install.os, "access", lambda _p, _m: False)
+    monkeypatch.setattr(cli_install, "_read_version", lambda _path: "9.9.9")
+    out = cli_install.install_connector_cli("wecom-cli")
+    assert out["ok"] is True
+    assert out["already_installed"] is False
+    install_call = [c for c in calls if c[1:3] == ["install", "-g"]][0]
+    assert "--prefix" in install_call
+    assert str(home / ".npm-global") in install_call
+    assert cli_install._user_npm_prefix()[0] == str(home / ".npm-global")
+
+
+def test_ensure_cli_path_injects_user_bin(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None:
+    home = tmp_path / "home"
+    user_prefix = str(home / ".npm-global")
+    bin_dir = cli_install._prefix_bin_dir(user_prefix)
+    import os as _os
+
+    _os.makedirs(bin_dir, exist_ok=True)
+    monkeypatch.setenv("HOME", str(home))
+    # Windows 上 os.path.expanduser("~") 读 USERPROFILE，需一并覆盖以跨平台
+    monkeypatch.setenv("USERPROFILE", str(home))
+    monkeypatch.setitem(cli_install.os.environ, "PATH", "/usr/bin")
+    out = cli_install.ensure_cli_path()
+    assert out == bin_dir
+    assert cli_install.os.environ["PATH"].startswith(bin_dir + _os.pathsep)
+    # 幂等：重复调用不重复追加
+    cli_install.ensure_cli_path()
+    assert cli_install.os.environ["PATH"].count(bin_dir) == 1