Merge remote-tracking branch 'origin/feat/92' into integ

# Conflicts:
#	bcc_core.py
#	tests/test_core.py

bcc_core.py

@@ -2777,6 +2777,183 @@ def sidecar_warnings(
     return out
 
 
+# Version resolve + pin + drift (issue #92, epic #94)
+#
+# `npx -y ssh-mcp` resolves *latest* on every launch. In one working session the
+# package went v1 → v2 and the exposed tool set changed under a running agent,
+# mid-task, with no warning. This surfaces that: detect an unpinned spec, show
+# the currently-resolved version (read locally — NO network calls, degrade to
+# "unknown" cleanly), offer a one-click pin, and re-check drift with
+# is_newer_version. Reuses parse_version / is_newer_version / the catalog spec
+# parsers — additive plumbing on existing primitives.
+# --------------------------------------------------------------------------- #
+# npx-style launchers that resolve a package spec at run time (the unpinned case).
+_NPX_LAUNCHERS = {"npx", "bunx", "pnpx"}
+
+# Sentinel so version_status can tell "argument omitted" (do the local lookup)
+# from an explicit resolved=None ("caller knows the version is unknown").
+_UNSET = object()
+
+
+def server_package_spec(data: dict) -> str | None:
+    """The npm package spec token an ``npx``-style stdio server launches, or None.
+
+    ``{"command": "npx", "args": ["-y", "ssh-mcp"]}`` → ``"ssh-mcp"``;
+    ``["-y", "ssh-mcp@2.1.0"]`` → ``"ssh-mcp@2.1.0"``. Only npx-style launchers
+    are considered — that is where "resolve latest each launch" bites. A direct
+    binary command (``{"command": "ssh-mcp"}``) has no run-time spec to pin.
+    """
+    if not isinstance(data, dict):
+        return None
+    cmd = str(data.get("command", "")).strip()
+    base = cmd.rsplit("/", 1)[-1] if "/" in cmd else cmd
+    if base not in _NPX_LAUNCHERS:
+        return None
+    return _first_catalog_package_spec([str(a) for a in (data.get("args") or [])])
+
+
+def server_package_name(data: dict) -> str | None:
+    """The bare package name for an npx-style server (spec minus any @version)."""
+    spec = server_package_spec(data)
+    return _normalize_pkg_token(spec) if spec else None
+
+
+def is_unpinned_spec(data: dict) -> bool:
+    """True when an npx-style server resolves 'latest' each launch.
+
+    Unpinned == a bare name (``ssh-mcp``) or a dist-tag (``ssh-mcp@latest``,
+    ``@next``): anything that is not an exact numeric version. An exact pin
+    (``ssh-mcp@2.1.0``) is stable and returns False.
+    """
+    spec = server_package_spec(data)
+    if spec is None:
+        return False
+    ver = _catalog_package_spec_version(spec)
+    # parse_version("latest") -> () (falsy); parse_version("2.1.0") -> (2,1,0).
+    return not (ver and parse_version(ver))
+
+
+def parse_package_json_version(text: str) -> str | None:
+    """Extract the ``version`` string from a package.json blob, or None."""
+    try:
+        d = json.loads(text)
+    except (ValueError, TypeError):
+        return None
+    v = d.get("version") if isinstance(d, dict) else None
+    return v if isinstance(v, str) and v.strip() else None
+
+
+def resolved_npx_version(
+    package: str,
+    *,
+    home: str | os.PathLike | None = None,
+    find=None,
+    read=None,
+) -> str | None:
+    """Best-effort *resolved* version of an npx-cached package. NO network.
+
+    npx unpacks each package under ``~/.npm/_npx/<hash>/node_modules/<pkg>/``; this
+    reads the ``version`` from that package.json. When several cache entries exist
+    (different launches), the highest version wins. ``find`` (a glob callable) and
+    ``read`` (path → text) are injectable so tests drive it with fixtures instead
+    of a real cache; both default to safe real implementations. Returns None when
+    nothing is found or the package name is empty — the caller shows "unknown".
+    """
+    if not package:
+        return None
+    home = Path(home) if home is not None else Path.home()
+    pattern = str(home / ".npm" / "_npx" / "*" / "node_modules" / package / "package.json")
+    if find is None:
+        find = glob.glob
+    if read is None:
+
+        def read(p):
+            try:
+                return Path(p).read_text(encoding="utf-8", errors="replace")
+            except OSError:
+                return ""
+
+    best: str | None = None
+    for path in find(pattern):
+        ver = parse_package_json_version(read(path))
+        if ver and (best is None or is_newer_version(best, ver)):
+            best = ver
+    return best
+
+
+def pin_spec_transform(data: dict, version: str) -> tuple[dict, str | None]:
+    """Rewrite an npx server's package spec to an exact ``name@version`` pin.
+
+    Returns ``(new_data, note)``; ``note`` is None when there is nothing to pin
+    (not an npx server, no resolvable spec, empty version, or already pinned to
+    that exact version). Mirrors ``pin_command_path``'s contract so the GUI wiring
+    is identical. Only the package token in ``args`` is touched.
+    """
+    if not version or not parse_version(version):
+        return data, None
+    spec = server_package_spec(data)
+    if spec is None:
+        return data, None
+    package = _normalize_pkg_token(spec)
+    if not package:
+        return data, None
+    pinned = f"{package}@{version}"
+    if spec == pinned:
+        return data, None
+    args = [str(a) for a in (data.get("args") or [])]
+    new_args = [pinned if a == spec else a for a in args]
+    if new_args == args:
+        return data, None
+    new_data = dict(data)
+    new_data["args"] = new_args
+    return new_data, f"pinned {package} to {version}"
+
+
+def version_drift_note(pinned: str | None, resolved: str | None) -> str | None:
+    """A 'moved X → Y' note when the resolved version is newer than the pin.
+
+    Only fires on a forward move (a package the user pinned that the cache has
+    since advanced past). Equal or older resolved versions, or unknown inputs,
+    return None. Uses is_newer_version so the compare is numeric, never lexical.
+    """
+    if not pinned or not resolved:
+        return None
+    if is_newer_version(pinned, resolved):
+        return f"moved {pinned} → {resolved} since you pinned"
+    return None
+
+
+def version_status(data: dict, *, resolved: str | None = _UNSET, **lookup) -> dict | None:
+    """High-level version state for a server, for the GUI badge. None if N/A.
+
+    Returns ``{package, spec, unpinned, pinned_version, resolved_version,
+    can_pin, drift}``. Pass ``resolved`` to supply the resolved version directly
+    (including an explicit ``None`` for "unknown"); omit it to have
+    ``resolved_npx_version`` read the local npx cache (injectable via ``**lookup``
+    → home/find/read). A "pin" action makes sense when the spec is unpinned AND a
+    resolved version is known (``can_pin``).
+    """
+    spec = server_package_spec(data)
+    if spec is None:
+        return None
+    package = _normalize_pkg_token(spec)
+    pinned_version = _catalog_package_spec_version(spec)
+    if not (pinned_version and parse_version(pinned_version)):
+        pinned_version = None  # a dist-tag ("latest") is not a real pin
+    if resolved is _UNSET:
+        resolved = resolved_npx_version(package, **lookup)
+    unpinned = pinned_version is None
+    return {
+        "package": package,
+        "spec": spec,
+        "unpinned": unpinned,
+        "pinned_version": pinned_version,
+        "resolved_version": resolved,
+        "can_pin": unpinned and bool(resolved),
+        "drift": version_drift_note(pinned_version, resolved),
+    }
+
+
 # --------------------------------------------------------------------------- #
 # Validation
 # --------------------------------------------------------------------------- #