From 2cd8e0fb3b652a728dca8b1d2cd369c720ed306f Mon Sep 17 00:00:00 2001
From: Cowork Supervisor <supervisor@cowork.local>
Date: Wed, 12 Aug 2026 03:04:53 -0400
Subject: [PATCH] feat(#92): detect unpinned npx specs, resolve version,
 one-click pin, drift
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`npx -y ssh-mcp` resolves *latest* on every launch — in one session ssh-mcp went
v1 → v2 and the tool set changed under a running agent, mid-task, with no warning.
This detects that and turns it into a comprehensible pin/upgrade prompt.

Core (pure, no network — reuses parse_version / is_newer_version / the catalog
spec parsers; the resolved-version lookup is fully injectable for tests):
- server_package_spec / server_package_name — the npm spec an npx-style server runs.
- is_unpinned_spec — bare name or dist-tag (@latest/@next) is unpinned; an exact
  numeric version is pinned.
- resolved_npx_version — reads the local ~/.npm/_npx cache (highest version wins),
  degrades to None cleanly. NO network. find/read injectable.
- pin_spec_transform — rewrite the spec to name@version (mirrors pin_command_path's
  (new_data, note) contract). No-op when already pinned / bad version / not npx.
- version_drift_note — "moved X → Y since you pinned" via numeric is_newer_version.
- version_status — the badge's high-level dict (unpinned / pinned_version /
  resolved_version / can_pin / drift). A _UNSET sentinel lets callers force an
  explicit resolved=None ("unknown") vs. omitting it to do the local lookup.

GUI: a version badge row under the dependency status (mirrors that surface) with a
one-click "Pin to <version>" button, shown only for npx servers; a drift note when
a pinned version has been overtaken. Smoke-tested headlessly.

pytest green (520 passed), ruff + format clean. Closes #92. Part of epic #94.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 bcc.py             |  66 +++++++++++++++++
 bcc_core.py        | 178 +++++++++++++++++++++++++++++++++++++++++++++
 tests/test_core.py | 106 +++++++++++++++++++++++++++
 3 files changed, 350 insertions(+)

diff --git a/bcc.py b/bcc.py
index afbf20b..97688ac 100644
--- a/bcc.py
+++ b/bcc.py
@@ -819,6 +819,27 @@ class ServerEditor(QFrame):
         dep.addWidget(recheck)
         outer.addLayout(dep)
 
+        # Version status row (#92): for an npx-style server, show the currently
+        # resolved version and, when the spec is unpinned, a one-click "pin".
+        # Mirrors the dependency-status surface above. Hidden for everything else.
+        ver = QHBoxLayout()
+        self.ver_dot = QLabel("○")
+        self.ver_label = QLabel("—")
+        self.ver_label.setObjectName("muted")
+        self.ver_label.setWordWrap(True)
+        self.pin_btn = QPushButton("Pin")
+        self.pin_btn.setToolTip(
+            "Pin the package spec to the currently-resolved version so it can't "
+            "change under you on the next launch"
+        )
+        self.pin_btn.clicked.connect(self._pin_version)
+        self.pin_btn.setVisible(False)
+        ver.addWidget(self.ver_dot)
+        ver.addWidget(self.ver_label, 1)
+        ver.addWidget(self.pin_btn)
+        self.ver_row_widgets = (self.ver_dot, self.ver_label, self.pin_btn)
+        outer.addLayout(ver)
+
         # Collapsible diagnostics panel
         self.diag_card = QFrame()
         self.diag_card.setObjectName("diagCard")
@@ -1146,6 +1167,8 @@ class ServerEditor(QFrame):
         if not self.isEnabled():
             self._set_dep({"status": "unknown", "label": "—"})
             self.diag_text.clear()
+            for wdg in self.ver_row_widgets:
+                wdg.hide()
             return
         data = self.dump_data()
         res = core.check_dependency(data)
@@ -1159,6 +1182,49 @@ class ServerEditor(QFrame):
             self.details_btn.setChecked(True)  # opens panel (fills text via _toggle_diag)
         if self.diag_card.isVisible():
             self.diag_text.setPlainText(self._full_diag_text())
+        self._refresh_version_badge(data)
+
+    def _refresh_version_badge(self, data: dict):
+        """Show the resolved version / pin state for an npx-style server (#92)."""
+        st = core.version_status(data) if self.type.currentIndex() == 0 else None
+        if st is None:
+            for wdg in self.ver_row_widgets:
+                wdg.hide()
+            return
+        for wdg in self.ver_row_widgets:
+            wdg.show()
+        resolved = st["resolved_version"] or "unknown"
+        if st["drift"]:
+            glyph, color, text = "●", WARN, f"{st['package']} {st['drift']}"
+        elif st["unpinned"]:
+            glyph, color = "●", WARN
+            text = f"{st['package']} · resolved {resolved} · unpinned (resolves latest each launch)"
+        else:
+            glyph, color = "●", GOOD
+            text = f"{st['package']} · pinned {st['pinned_version']}"
+        self.ver_dot.setText(glyph)
+        self.ver_dot.setStyleSheet(f"color: {color}; font-size: 14px;")
+        self.ver_label.setText(text)
+        self.ver_label.setStyleSheet(f"color: {color};")
+        # Offer the pin only when unpinned AND we know what to pin to.
+        self.pin_btn.setVisible(st["can_pin"])
+        if st["can_pin"]:
+            self.pin_btn.setText(f"Pin to {resolved}")
+        self._version_resolved = st["resolved_version"]
+
+    def _pin_version(self):
+        if self._before_change:
+            self._before_change()
+        resolved = getattr(self, "_version_resolved", None)
+        if not resolved:
+            return
+        new_data, note = core.pin_spec_transform(self.dump_data(), resolved)
+        if not note:
+            return
+        self._loading = True
+        self.args.setPlainText("\n".join(str(a) for a in (new_data.get("args") or [])))
+        self._loading = False
+        self._emit()  # writes back to the model and re-checks (badge now "pinned")
 
     def _set_dep(self, res: dict):
         status = res.get("status", "unknown")
diff --git a/bcc_core.py b/bcc_core.py
index 23804ca..6df83ba 100644
--- a/bcc_core.py
+++ b/bcc_core.py
@@ -2535,6 +2535,184 @@ def drift_warnings(data: dict) -> list[str]:
     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
 # --------------------------------------------------------------------------- #
diff --git a/tests/test_core.py b/tests/test_core.py
index 31d9b5f..f0ec06e 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -3297,6 +3297,112 @@ def test_drift_warning_quiet_for_other_values_and_unknown_pkg():
     assert c.drift_warnings({"command": "npx", "args": ["other", "--maxChars=none"]}) == []
 
 
+# --------------------------------------------------------------------------- #
+# Version resolve + pin + drift (issue #92)
+# --------------------------------------------------------------------------- #
+def test_server_package_spec_and_name():
+    assert c.server_package_spec({"command": "npx", "args": ["-y", "ssh-mcp"]}) == "ssh-mcp"
+    assert c.server_package_spec({"command": "npx", "args": ["ssh-mcp@2.1.0"]}) == "ssh-mcp@2.1.0"
+    assert c.server_package_name({"command": "npx", "args": ["ssh-mcp@2.1.0"]}) == "ssh-mcp"
+    # A direct binary launch has no run-time spec to pin.
+    assert c.server_package_spec({"command": "ssh-mcp", "args": []}) is None
+    assert c.server_package_spec({"url": "https://x"}) is None
+
+
+def test_is_unpinned_spec():
+    assert c.is_unpinned_spec({"command": "npx", "args": ["-y", "ssh-mcp"]}) is True
+    assert c.is_unpinned_spec({"command": "npx", "args": ["ssh-mcp@latest"]}) is True
+    assert c.is_unpinned_spec({"command": "npx", "args": ["ssh-mcp@next"]}) is True
+    assert c.is_unpinned_spec({"command": "npx", "args": ["ssh-mcp@2.1.0"]}) is False
+    assert c.is_unpinned_spec({"command": "ssh-mcp", "args": []}) is False
+
+
+def test_parse_package_json_version():
+    assert c.parse_package_json_version('{"name":"ssh-mcp","version":"2.1.0"}') == "2.1.0"
+    assert c.parse_package_json_version('{"name":"x"}') is None
+    assert c.parse_package_json_version("not json") is None
+    assert c.parse_package_json_version('{"version":""}') is None
+
+
+def test_resolved_npx_version_reads_cache_highest_wins():
+    # Two cache entries for ssh-mcp; the highest version wins. No real filesystem.
+    files = {
+        "/h/.npm/_npx/aaa/node_modules/ssh-mcp/package.json": '{"version":"1.9.0"}',
+        "/h/.npm/_npx/bbb/node_modules/ssh-mcp/package.json": '{"version":"2.1.0"}',
+    }
+    got = c.resolved_npx_version(
+        "ssh-mcp",
+        home="/h",
+        find=lambda pat: list(files),
+        read=lambda p: files[p],
+    )
+    assert got == "2.1.0"
+
+
+def test_resolved_npx_version_unknown_degrades_to_none():
+    assert (
+        c.resolved_npx_version("ssh-mcp", home="/h", find=lambda pat: [], read=lambda p: "") is None
+    )
+    assert c.resolved_npx_version("", home="/h") is None
+
+
+def test_pin_spec_transform():
+    data = {"command": "npx", "args": ["-y", "ssh-mcp", "--host=h"]}
+    new, note = c.pin_spec_transform(data, "2.1.0")
+    assert new["args"] == ["-y", "ssh-mcp@2.1.0", "--host=h"]
+    assert note and "2.1.0" in note
+    # Already pinned to that exact version -> no-op.
+    again, note2 = c.pin_spec_transform(new, "2.1.0")
+    assert again == new
+    assert note2 is None
+    # Bad / empty version -> no-op.
+    assert c.pin_spec_transform(data, "")[1] is None
+    assert c.pin_spec_transform(data, "latest")[1] is None
+    # Not an npx server -> no-op.
+    assert c.pin_spec_transform({"command": "ssh-mcp", "args": []}, "2.1.0")[1] is None
+
+
+def test_version_drift_note():
+    assert c.version_drift_note("2.1.0", "3.0.0") == "moved 2.1.0 → 3.0.0 since you pinned"
+    assert c.version_drift_note("3.0.0", "3.0.0") is None
+    assert c.version_drift_note("3.0.0", "2.1.0") is None  # never a backwards "drift"
+    assert c.version_drift_note(None, "3.0.0") is None
+    assert c.version_drift_note("2.1.0", None) is None
+    # Numeric, not lexical: 2 < 10.
+    assert c.version_drift_note("2.0.0", "10.0.0") is not None
+
+
+def test_version_status_unpinned_offers_pin():
+    st = c.version_status({"command": "npx", "args": ["-y", "ssh-mcp"]}, resolved="2.1.0")
+    assert st["package"] == "ssh-mcp"
+    assert st["unpinned"] is True
+    assert st["pinned_version"] is None
+    assert st["resolved_version"] == "2.1.0"
+    assert st["can_pin"] is True
+    assert st["drift"] is None
+
+
+def test_version_status_pinned_reports_drift():
+    st = c.version_status({"command": "npx", "args": ["ssh-mcp@2.1.0"]}, resolved="3.0.0")
+    assert st["unpinned"] is False
+    assert st["pinned_version"] == "2.1.0"
+    assert st["can_pin"] is False  # already pinned
+    assert st["drift"] == "moved 2.1.0 → 3.0.0 since you pinned"
+
+
+def test_version_status_unknown_resolved_cannot_pin():
+    st = c.version_status({"command": "npx", "args": ["-y", "ssh-mcp"]}, resolved=None)
+    assert st["unpinned"] is True
+    assert st["resolved_version"] is None
+    assert st["can_pin"] is False  # nothing to pin TO
+    assert st["drift"] is None
+
+
+def test_version_status_none_for_non_npx():
+    assert c.version_status({"command": "ssh-mcp", "args": []}) is None
+    assert c.version_status({"url": "https://x"}) is None
+
+
 # --------------------------------------------------------------------------- #
 # Move to environment variable (issue #83)
 # --------------------------------------------------------------------------- #