feat(agents): npm 安装失败/超时回退国内镜像 registry.npmmirror.com

官方 registry.npmjs.org 国内常卡住。_run_install 现在：官方源优先（180s
超时护栏），失败或超时自动用 npmmirror 重试一次；NPM_REGISTRY 环境变量
可整体覆盖首选源。含 6 个测试。顺带清理 launcher 既有 lint（import 排序/
嵌套 with/contextlib.suppress）。

myagents/launcher.py

@@ -1,5 +1,6 @@
 """Shared launcher logic for AI coding agent CLIs."""
 
+import contextlib
 import hashlib
 import json
 import os
@@ -140,11 +141,19 @@ _BACKENDS: dict[str, dict] = {
 }
 
 _NPM_OFFICIAL_REGISTRY = "https://registry.npmjs.org/"
+_NPM_MIRROR_REGISTRY = "https://registry.npmmirror.com/"
 _NPM_INSTALL_FLAGS = (
     "--include=optional",
     "--foreground-scripts",
-    f"--registry={_NPM_OFFICIAL_REGISTRY}",
 )
+# Hang guard: npm against a slow/unreachable registry blocks forever
+# without a timeout. One install attempt, then a mirror retry.
+_NPM_INSTALL_TIMEOUT = 180  # seconds per attempt
+
+
+def _npm_registry() -> str:
+    """Registry to use — env ``NPM_REGISTRY`` overrides the official default."""
+    return os.environ.get("NPM_REGISTRY", "").strip() or _NPM_OFFICIAL_REGISTRY
 
 
 def _resolve_chat_cwd(cwd: str | None) -> Path:
@@ -594,20 +603,43 @@ def _ensure_agent_hint() -> str:
     )
 
 
-def _npm_install_argv(install_cmd: list[str]) -> list[str]:
+def _npm_install_argv(
+    install_cmd: list[str], registry: str | None = None
+) -> list[str]:
     """``npm install -g <pkg>`` plus flags so platform optional deps resolve."""
     argv = list(install_cmd)
     if argv and argv[0] == "npm":
         argv.extend(_NPM_INSTALL_FLAGS)
+        argv.append(f"--registry={registry or _npm_registry()}")
     return argv
 
 
+def _run_npm_install(
+    installer: str, args: list[str]
+) -> tuple[subprocess.CompletedProcess, bool]:
+    """Run ``installer args`` with a hang guard; ``(proc, timed_out)``."""
+    try:
+        proc = subprocess.run(
+            [installer, *args], check=False, timeout=_NPM_INSTALL_TIMEOUT
+        )
+        return proc, False
+    except subprocess.TimeoutExpired:
+        return (
+            subprocess.CompletedProcess([installer, *args], returncode=-1),
+            True,
+        )
+
+
 def _run_install(config: dict) -> str | None:
     """Run ``install_cmd`` and return the resolved binary path, or None."""
     install_cmd = config.get("install_cmd")
     if not install_cmd:
         return None
-    argv = _npm_install_argv(list(install_cmd))
+    primary = _npm_registry()
+    mirror = (
+        _NPM_MIRROR_REGISTRY if primary != _NPM_MIRROR_REGISTRY else None
+    )
+    argv = _npm_install_argv(list(install_cmd), registry=primary)
     installer = shutil.which(argv[0])
     if not installer:
         hint = _ensure_agent_hint()
@@ -618,6 +650,7 @@ def _run_install(config: dict) -> str | None:
             f"[cyan]{hint}[/cyan]."
         )
         return None
+    hint = _ensure_agent_hint()
     stderr_console.print(
         f"[dim]Installing {config['binary']} via[/dim] "
         f"[cyan]{shlex.join(argv)}[/cyan]"
@@ -625,9 +658,16 @@ def _run_install(config: dict) -> str | None:
     stderr_console.print(
         "[dim]  (npm global install — may take a minute)[/dim]"
     )
-    proc = subprocess.run([installer, *argv[1:]], check=False)
-    if proc.returncode != 0:
-        hint = _ensure_agent_hint()
+    proc, timed_out = _run_npm_install(installer, argv[1:])
+    if (proc.returncode != 0 or timed_out) and mirror:
+        stderr_console.print(
+            f"[yellow]{config['binary']} install via {primary} "
+            f"{'timed out' if timed_out else f'exited {proc.returncode}'} — "
+            f"retrying with mirror {mirror}[/yellow]"
+        )
+        argv = _npm_install_argv(list(install_cmd), registry=mirror)
+        proc, timed_out = _run_npm_install(installer, argv[1:])
+    if proc.returncode != 0 or timed_out:
         stderr_console.print(
             f"[red]Install failed (exit {proc.returncode}).[/red]\n"
             f"  Run: [cyan]{hint}[/cyan]."
@@ -643,7 +683,7 @@ def _run_install(config: dict) -> str | None:
         "[red]Installed CLI is not a valid native binary for this OS[/red] "
         "(npm mirrors often ship a 1KB stub or the wrong platform).\n"
         f"  [cyan]npm uninstall -g {pkg}[/cyan]\n"
-        f"  [cyan]npm install -g {pkg} --registry={_NPM_OFFICIAL_REGISTRY} "
+        f"  [cyan]npm install -g {pkg} --registry={_npm_registry()} "
         "--include=optional --foreground-scripts[/cyan]"
     )
     return None
@@ -1411,25 +1451,28 @@ def _dsh_session_cwd(session_file: Path) -> str | None:
     """Read the ``cwd`` from a dsh ``session.jsonl.zstd`` header line."""
     try:
         import io
+
         import zstandard as zstd
     except ImportError:
         return None
     try:
-        with session_file.open("rb") as fh:
-            with zstd.ZstdDecompressor().stream_reader(fh) as reader:
-                text = io.TextIOWrapper(reader, encoding="utf-8", errors="ignore")
-                for line in text:
-                    line = line.strip()
-                    if not line:
-                        continue
-                    try:
-                        entry = json.loads(line)
-                    except json.JSONDecodeError:
-                        continue
-                    if not isinstance(entry, dict) or entry.get("type") != "session":
-                        return None
-                    cwd = entry.get("cwd")
-                    return cwd if isinstance(cwd, str) else None
+        with (
+            session_file.open("rb") as fh,
+            zstd.ZstdDecompressor().stream_reader(fh) as reader,
+        ):
+            text = io.TextIOWrapper(reader, encoding="utf-8", errors="ignore")
+            for line in text:
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    entry = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+                if not isinstance(entry, dict) or entry.get("type") != "session":
+                    return None
+                cwd = entry.get("cwd")
+                return cwd if isinstance(cwd, str) else None
     except (OSError, zstd.ZstdError):
         return None
     return None
@@ -1442,40 +1485,43 @@ def _first_prompt_dsh(session_file: Path) -> str:
     """
     try:
         import io
+
         import zstandard as zstd
     except ImportError:
         return ""
     try:
-        with session_file.open("rb") as fh:
-            with zstd.ZstdDecompressor().stream_reader(fh) as reader:
-                text = io.TextIOWrapper(reader, encoding="utf-8", errors="ignore")
-                for line in text:
-                    line = line.strip()
-                    if not line:
-                        continue
-                    try:
-                        entry = json.loads(line)
-                    except json.JSONDecodeError:
-                        continue
-                    if not isinstance(entry, dict):
-                        continue
-                    etype = entry.get("type")
-                    data = entry.get("data")
-                    if etype == "session/title" and isinstance(data, dict):
-                        title = data.get("title")
-                        if isinstance(title, str) and title.strip():
-                            return " ".join(title.split())[:80]
-                    if etype == "user/message" and isinstance(data, dict):
-                        content = data.get("content")
-                        if isinstance(content, list):
-                            for block in content:
-                                if (
-                                    isinstance(block, dict)
-                                    and block.get("type") == "text"
-                                ):
-                                    t = block.get("text")
-                                    if isinstance(t, str) and t.strip():
-                                        return " ".join(t.split())[:80]
+        with (
+            session_file.open("rb") as fh,
+            zstd.ZstdDecompressor().stream_reader(fh) as reader,
+        ):
+            text = io.TextIOWrapper(reader, encoding="utf-8", errors="ignore")
+            for line in text:
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    entry = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+                if not isinstance(entry, dict):
+                    continue
+                etype = entry.get("type")
+                data = entry.get("data")
+                if etype == "session/title" and isinstance(data, dict):
+                    title = data.get("title")
+                    if isinstance(title, str) and title.strip():
+                        return " ".join(title.split())[:80]
+                if etype == "user/message" and isinstance(data, dict):
+                    content = data.get("content")
+                    if isinstance(content, list):
+                        for block in content:
+                            if (
+                                isinstance(block, dict)
+                                and block.get("type") == "text"
+                            ):
+                                t = block.get("text")
+                                if isinstance(t, str) and t.strip():
+                                    return " ".join(t.split())[:80]
     except (OSError, zstd.ZstdError):
         return ""
     return ""
@@ -1490,10 +1536,8 @@ def _decode_cursor_meta_value(value: str) -> dict | None:
     """Decode Cursor store.db meta values (hex-encoded JSON or plain JSON)."""
     candidates = [value]
     if len(value) % 2 == 0 and all(c in "0123456789abcdefABCDEF" for c in value):
-        try:
+        with contextlib.suppress(ValueError, UnicodeDecodeError):
             candidates.insert(0, bytes.fromhex(value).decode("utf-8"))
-        except (ValueError, UnicodeDecodeError):
-            pass
     for raw in candidates:
         try:
             obj = json.loads(raw)

tests/test_ensure_backend.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import os
+import subprocess
 from pathlib import Path
 from unittest.mock import patch
 
@@ -114,3 +115,86 @@ class TestOfferInstallGuided:
             prompt.assert_called_once()
             assert prompt.call_args.kwargs.get("default") == "y"
             run_install.assert_called_once()
+
+
+class TestNpmRegistryFallback:
+    """npm install registry selection + mirror fallback on failure/timeout."""
+
+    DSH = {
+        "binary": "dsh",
+        "install_cmd": ["npm", "install", "-g", "@deepseek-ai/dsh"],
+    }
+
+    def test_npm_install_argv_default_official(self) -> None:
+        argv = launcher_mod._npm_install_argv(
+            ["npm", "install", "-g", "@deepseek-ai/dsh"]
+        )
+        assert "--registry=https://registry.npmjs.org/" in argv
+        assert "--include=optional" in argv
+        assert "--foreground-scripts" in argv
+
+    def test_npm_registry_env_override(self, monkeypatch) -> None:
+        monkeypatch.setenv("NPM_REGISTRY", "https://registry.npmmirror.com/")
+        argv = launcher_mod._npm_install_argv(
+            ["npm", "install", "-g", "@deepseek-ai/dsh"]
+        )
+        assert "--registry=https://registry.npmmirror.com/" in argv
+
+    def _patch_run(self, side_effect):
+        return patch.object(launcher_mod, "_run_npm_install", side_effect=side_effect)
+
+    def test_run_install_success_first_try(self) -> None:
+        ok = subprocess.CompletedProcess(["npm"], returncode=0)
+        with (
+            self._patch_run([(ok, False)]),
+            patch.object(
+                launcher_mod, "_resolve_backend_binary", return_value="/usr/bin/dsh"
+            ),
+            patch.object(launcher_mod, "ensure_npm_bin_on_path"),
+            patch.object(launcher_mod.shutil, "which", return_value="/usr/bin/npm"),
+        ):
+            assert launcher_mod._run_install(self.DSH) == "/usr/bin/dsh"
+
+    def test_run_install_falls_back_to_mirror_on_failure(self) -> None:
+        fail = subprocess.CompletedProcess(["npm"], returncode=1)
+        ok = subprocess.CompletedProcess(["npm"], returncode=0)
+        with (
+            self._patch_run([(fail, False), (ok, False)]) as run,
+            patch.object(
+                launcher_mod, "_resolve_backend_binary", return_value="/usr/bin/dsh"
+            ),
+            patch.object(launcher_mod, "ensure_npm_bin_on_path"),
+            patch.object(launcher_mod.shutil, "which", return_value="/usr/bin/npm"),
+        ):
+            assert launcher_mod._run_install(self.DSH) == "/usr/bin/dsh"
+            assert run.call_count == 2
+            second_args = run.call_args_list[1][0][1]
+            assert "--registry=https://registry.npmmirror.com/" in second_args
+
+    def test_run_install_falls_back_to_mirror_on_timeout(self) -> None:
+        timed = subprocess.CompletedProcess(["npm"], returncode=-1)
+        ok = subprocess.CompletedProcess(["npm"], returncode=0)
+        with (
+            self._patch_run([(timed, True), (ok, False)]) as run,
+            patch.object(
+                launcher_mod, "_resolve_backend_binary", return_value="/usr/bin/dsh"
+            ),
+            patch.object(launcher_mod, "ensure_npm_bin_on_path"),
+            patch.object(launcher_mod.shutil, "which", return_value="/usr/bin/npm"),
+        ):
+            assert launcher_mod._run_install(self.DSH) == "/usr/bin/dsh"
+            assert run.call_count == 2
+            second_args = run.call_args_list[1][0][1]
+            assert "--registry=https://registry.npmmirror.com/" in second_args
+
+    def test_run_install_fails_when_both_registries_fail(self) -> None:
+        fail1 = subprocess.CompletedProcess(["npm"], returncode=1)
+        fail2 = subprocess.CompletedProcess(["npm"], returncode=2)
+        with (
+            self._patch_run([(fail1, False), (fail2, False)]),
+            patch.object(
+                launcher_mod, "_resolve_backend_binary", return_value=None
+            ),
+            patch.object(launcher_mod.shutil, "which", return_value="/usr/bin/npm"),
+        ):
+            assert launcher_mod._run_install(self.DSH) is None