diff --git a/bcc.py b/bcc.py index 599eaac..7b35250 100644 --- a/bcc.py +++ b/bcc.py @@ -11,13 +11,23 @@ Run: python mcp_manager.py from __future__ import annotations import sys +import time from pathlib import Path -from PySide6.QtCore import QRect, QSettings, QSize, Qt, QThread, QTimer, Signal -from PySide6.QtGui import QAction, QColor, QGuiApplication, QPainter +from PySide6.QtCore import QRect, QSettings, QSize, Qt, QThread, QTimer, QUrl, Signal +from PySide6.QtGui import ( + QAction, + QColor, + QDesktopServices, + QGuiApplication, + QIcon, + QPainter, + QPixmap, +) from PySide6.QtWidgets import ( QAbstractItemView, QApplication, + QCheckBox, QComboBox, QDialog, QDialogButtonBox, @@ -147,6 +157,23 @@ class SpawnTester(QThread): self.done.emit(result) +class UpdateCheckWorker(QThread): + """ + Fetches the latest release from the Gitea releases API off the UI thread. + + Notify-only: `core.fetch_latest_release()` only ever reads release + metadata (a version tag + a URL) and never downloads or replaces the + running binary. Fails quiet — emits None on any network problem — so + it's safe to fire unattended from a silent startup check as well as from + the About dialog's "Check for updates" button. + """ + + done = Signal(object) # dict | None + + def run(self): + self.done.emit(core.fetch_latest_release()) + + # --------------------------------------------------------------------------- # # Small reusable: key/value editor (for env and headers) # --------------------------------------------------------------------------- # @@ -374,6 +401,9 @@ class ServerEditor(QFrame): self.spawn_btn.setToolTip("Spawn the server for 3 s and report whether it starts cleanly") self.spawn_btn.clicked.connect(self._test_spawn) self.spawn_btn.setVisible(False) + self.logs_btn = QPushButton("View logs") + self.logs_btn.setToolTip("Open this server's MCP log in a read-only, auto-tailing viewer") + self.logs_btn.clicked.connect(self._view_logs) self.details_btn = QPushButton("Details ▸") self.details_btn.setCheckable(True) self.details_btn.toggled.connect(self._toggle_diag) @@ -384,6 +414,7 @@ class ServerEditor(QFrame): dep.addWidget(self.fix_btn) dep.addWidget(self.test_btn) dep.addWidget(self.spawn_btn) + dep.addWidget(self.logs_btn) dep.addWidget(self.details_btn) dep.addWidget(recheck) outer.addLayout(dep) @@ -728,6 +759,13 @@ class ServerEditor(QFrame): elif self.diag_card.isVisible(): self.diag_text.setPlainText(self._full_diag_text()) + def _view_logs(self): + name = self.current_name() + if not name: + return + dlg = LogViewerDialog(self.window(), name) + dlg.exec() + # --------------------------------------------------------------------------- # # Arguments editor: one line = one argument, with a numbered gutter so that @@ -1090,6 +1128,272 @@ class PasteDialog(QDialog): self.err.setText(str(e)) +# --------------------------------------------------------------------------- # +# Log viewer dialog (issue #6): a read-only, auto-tailing view of a single +# server's MCP log file. Polls on a QTimer instead of watching the filesystem +# so it works the same on every platform; never writes to the log. +# --------------------------------------------------------------------------- # +class LogViewerDialog(QDialog): + POLL_MS = 1500 + MAX_TAIL_BYTES = 300_000 + + def __init__(self, parent, server_name: str): + super().__init__(parent) + self._name = server_name + self._last_path: Path | None = None + self._last_size: int | None = None + self.setWindowTitle(f"Logs — {server_name}") + self.resize(760, 520) + + v = QVBoxLayout(self) + v.setSpacing(8) + + self.path_label = QLabel("") + self.path_label.setObjectName("muted") + self.path_label.setWordWrap(True) + v.addWidget(self.path_label) + + self.view = QPlainTextEdit() + self.view.setObjectName("diag") + self.view.setReadOnly(True) + self.view.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap) + v.addWidget(self.view, 1) + + row = QHBoxLayout() + self.status_label = QLabel("") + self.status_label.setObjectName("muted") + row.addWidget(self.status_label, 1) + refresh_btn = QPushButton("Refresh now") + refresh_btn.clicked.connect(self._poll) + row.addWidget(refresh_btn) + close_btn = QPushButton("Close") + close_btn.setObjectName("primary") + close_btn.clicked.connect(self.accept) + row.addWidget(close_btn) + v.addLayout(row) + + self._timer = QTimer(self) + self._timer.setInterval(self.POLL_MS) + self._timer.timeout.connect(self._poll) + self._timer.start() + + self._poll() + + def _poll(self): + path = core.server_log_path(self._name) + if path is None: + self._last_path = None + self._last_size = None + self.path_label.setText(f"No log file for '{self._name}' yet.") + self.view.setPlaceholderText( + "No log yet — this fills in once the server has run at least once " + "and produced output." + ) + self.view.clear() + self.status_label.setText("Waiting…") + return + + self.path_label.setText(str(path)) + try: + size = path.stat().st_size + except OSError: + self.status_label.setText("Log file disappeared.") + return + + # Nothing changed since the last poll -> skip the re-read/re-render. + if path == self._last_path and size == self._last_size: + return + + was_at_bottom = self._is_scrolled_to_bottom() + text, truncated = self._read_tail(path) + self.view.setPlainText(text) + self.status_label.setText( + f"Showing last {self.MAX_TAIL_BYTES // 1000} KB of the log." if truncated else "" + ) + if was_at_bottom: + self._scroll_to_bottom() + + self._last_path = path + self._last_size = size + + def _read_tail(self, path: Path) -> tuple[str, bool]: + try: + size = path.stat().st_size + with open(path, "rb") as f: + truncated = size > self.MAX_TAIL_BYTES + if truncated: + f.seek(size - self.MAX_TAIL_BYTES) + data = f.read() + except OSError as e: + return f"(could not read log: {e})", False + text = data.decode("utf-8", errors="replace") + if truncated: + nl = text.find("\n") + if nl != -1: + text = text[nl + 1 :] + text = "… (earlier lines truncated) …\n" + text + return text, truncated + + def _is_scrolled_to_bottom(self) -> bool: + sb = self.view.verticalScrollBar() + return sb.value() >= sb.maximum() - 4 + + def _scroll_to_bottom(self): + sb = self.view.verticalScrollBar() + sb.setValue(sb.maximum()) + + def closeEvent(self, event): + self._timer.stop() + super().closeEvent(event) + + +# --------------------------------------------------------------------------- # +# About dialog +# --------------------------------------------------------------------------- # +class AboutDialog(QDialog): + """ + App info + a manual "Check for updates" action. + + Shows nothing sensitive: app name, icon, version (from + core.__version__), and links opened in the system browser via + QDesktopServices.openUrl — never navigated to in-app. The update check + itself only ever reads release metadata (see UpdateCheckWorker); it + never downloads or replaces the running binary. + """ + + def __init__(self, parent=None): + super().__init__(parent) + self.setWindowTitle("About Better Claude Config") + self.setFixedWidth(440) + self._worker: UpdateCheckWorker | None = None + self._release_url: str | None = None + + icon_path = ( + Path(__file__).resolve().parent / "icons" / "twin-gears" / "rounded" / "icon-128.png" + ) + if icon_path.is_file(): + self.setWindowIcon(QIcon(str(icon_path))) + + v = QVBoxLayout(self) + v.setSpacing(10) + + head = QHBoxLayout() + head.setSpacing(12) + icon_lbl = QLabel() + if icon_path.is_file(): + icon_lbl.setPixmap( + QPixmap(str(icon_path)).scaled( + 56, + 56, + Qt.AspectRatioMode.KeepAspectRatio, + Qt.TransformationMode.SmoothTransformation, + ) + ) + head.addWidget(icon_lbl) + + title_box = QVBoxLayout() + title_box.setSpacing(2) + name_lbl = QLabel("Better Claude Config") + name_lbl.setObjectName("h1") + title_box.addWidget(name_lbl) + ver_lbl = QLabel(f"Version {core.__version__}") + ver_lbl.setObjectName("muted") + title_box.addWidget(ver_lbl) + head.addLayout(title_box, 1) + v.addLayout(head) + + desc = QLabel( + "A cross-platform GUI for editing the mcpServers block of Claude " + "Desktop and Claude Code configs." + ) + desc.setObjectName("muted") + desc.setWordWrap(True) + v.addWidget(desc) + + if sys.platform == "darwin": + note = QLabel( + "This build isn't notarized by Apple. If macOS blocks it on first " + "launch, right-click the app ▸ Open, or allow it under System " + "Settings ▸ Privacy & Security." + ) + note.setObjectName("muted") + note.setWordWrap(True) + v.addWidget(note) + + links = QHBoxLayout() + for text, url in ( + ("Repository", core.REPO_URL), + ("Issues", core.ISSUES_URL), + ("MIT License", core.LICENSE_URL), + ): + btn = QPushButton(text) + btn.setFlat(True) + btn.setCursor(Qt.CursorShape.PointingHandCursor) + btn.clicked.connect(lambda _=False, u=url: QDesktopServices.openUrl(QUrl(u))) + links.addWidget(btn) + links.addStretch() + v.addLayout(links) + + # --- update check ----------------------------------------------- # + upd_row = QHBoxLayout() + self.check_btn = QPushButton("Check for updates") + self.check_btn.clicked.connect(self._check_for_updates) + upd_row.addWidget(self.check_btn) + self.update_status = QLabel("") + self.update_status.setObjectName("muted") + self.update_status.setWordWrap(True) + upd_row.addWidget(self.update_status, 1) + v.addLayout(upd_row) + + self.release_btn = QPushButton("Open releases page") + self.release_btn.setCursor(Qt.CursorShape.PointingHandCursor) + self.release_btn.clicked.connect(self._open_release_page) + self.release_btn.hide() + v.addWidget(self.release_btn) + + self.auto_check_box = QCheckBox("Automatically check for updates on startup") + st = QSettings("BCC", "BetterClaudeConfig") + self.auto_check_box.setChecked(bool(st.value("update/autoCheck", True, type=bool))) + self.auto_check_box.toggled.connect(self._toggle_auto_check) + v.addWidget(self.auto_check_box) + + btns = QDialogButtonBox(QDialogButtonBox.StandardButton.Close) + btns.rejected.connect(self.reject) + btns.button(QDialogButtonBox.StandardButton.Close).clicked.connect(self.accept) + v.addWidget(btns) + + def _toggle_auto_check(self, on: bool): + QSettings("BCC", "BetterClaudeConfig").setValue("update/autoCheck", on) + + def _check_for_updates(self): + self.check_btn.setEnabled(False) + self.release_btn.hide() + self.update_status.setStyleSheet(f"color: {MUTED};") + self.update_status.setText("Checking…") + self._worker = UpdateCheckWorker() + self._worker.done.connect(self._on_check_done) + self._worker.start() + + def _on_check_done(self, release: dict | None): + self.check_btn.setEnabled(True) + self._worker = None + if not release: + self.update_status.setStyleSheet(f"color: {MUTED};") + self.update_status.setText("Couldn't check for updates (offline?).") + return + if core.is_newer_version(core.__version__, release["version"]): + self.update_status.setStyleSheet(f"color: {ACCENT};") + self.update_status.setText(f"{release['version']} available.") + self._release_url = release.get("url") or core.RELEASES_URL + self.release_btn.show() + else: + self.update_status.setStyleSheet(f"color: {GOOD};") + self.update_status.setText("You're up to date.") + + def _open_release_page(self): + QDesktopServices.openUrl(QUrl(self._release_url or core.RELEASES_URL)) + + # --------------------------------------------------------------------------- # # Main window # --------------------------------------------------------------------------- # @@ -1099,12 +1403,13 @@ class MainWindow(QMainWindow): self.setWindowTitle("Better Claude Config") self.resize(940, 640) self.setAcceptDrops(True) + self._build_menu_bar() self.profiles: list[core.Profile] = [] self.full_config: dict = {} self.servers: list[core.ServerEntry] = [] self.current_profile: core.Profile | None = None - self._loaded_mtime: float | None = None + self._loaded_stat: core.ConfigStat | None = None self.dirty = False self._suppress_table = False self._suppress_sel = False @@ -1135,12 +1440,53 @@ class MainWindow(QMainWindow): root.addLayout(self._build_actionbar()) + status_row = QHBoxLayout() + status_row.setContentsMargins(0, 0, 0, 0) self.status = QLabel("Ready.") self.status.setObjectName("statusbar") - root.addWidget(self.status) + status_row.addWidget(self.status, 1) + self.restart_btn = QPushButton("Restart Claude Desktop") + self.restart_btn.setToolTip( + "Quit and relaunch Claude Desktop so the saved config takes effect" + ) + self.restart_btn.clicked.connect(self._restart_claude_desktop) + self.restart_btn.hide() + status_row.addWidget(self.restart_btn) + root.addLayout(status_row) self._restore_layout() self.reload_profiles() + self._maybe_auto_check_updates() + + # --- menu bar ---------------------------------------------------------- # + def _build_menu_bar(self): + help_menu = self.menuBar().addMenu("&Help") + about_action = QAction("About Better Claude Config…", self) + about_action.triggered.connect(self._show_about) + help_menu.addAction(about_action) + + def _show_about(self): + AboutDialog(self).exec() + + # --- update check (silent, throttled, off-thread) --------------------- # + def _maybe_auto_check_updates(self): + st = QSettings("BCC", "BetterClaudeConfig") + if not bool(st.value("update/autoCheck", True, type=bool)): + return + last = float(st.value("update/lastCheck", 0.0, type=float) or 0.0) + if (time.time() - last) < 86400: # at most once/day + return + st.setValue("update/lastCheck", time.time()) + self._startup_update_worker = UpdateCheckWorker() + self._startup_update_worker.done.connect(self._on_startup_update_checked) + self._startup_update_worker.start() + + def _on_startup_update_checked(self, release: dict | None): + self._startup_update_worker = None + if release and core.is_newer_version(core.__version__, release["version"]): + self.status.setText( + f"Update available: {release['version']} · Help ▸ About to view it." + ) # --- layout persistence ---------------------------------------------- # def _restore_layout(self): @@ -1380,10 +1726,11 @@ class MainWindow(QMainWindow): return self.full_config = cfg repaired = True - self._loaded_mtime = core.config_mtime(profile.path) + self._loaded_stat = core.config_fingerprint(profile.path) self.current_profile = profile self.servers = core.extract_servers(self.full_config) self.dirty = False + self.restart_btn.hide() self._undo_stack.clear() self.undo_btn.setEnabled(False) self._refresh_tables(select_index=0 if self.servers else -1) @@ -1620,34 +1967,38 @@ class MainWindow(QMainWindow): self._refresh_tables(select_index=min(idx, len(self.servers) - 1)) self._mark_dirty() + def _import_server(self, name: str, data: dict) -> tuple[bool, bool]: + """ + Add a pasted/dropped `name`/`data` server to self.servers, resolving + a name collision with an explicit prompt (never a silent overwrite). + + Returns (added, replaced). + """ + existing = {s.name: i for i, s in enumerate(self.servers)} + if name in existing: + ans = QMessageBox.question( + self, + "Server exists", + f"“{name}” already exists. Replace it?\n\nYes = replace · No = keep both (renamed)", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + ) + if ans == QMessageBox.StandardButton.Yes: + self.servers[existing[name]].data = data + return False, True + name = core.resolve_name_collision(name, {s.name for s in self.servers}) + self.servers.append(core.ServerEntry(name, data, True)) + return True, False + def paste_json(self): dlg = PasteDialog(self) if dlg.exec() != QDialog.DialogCode.Accepted or not dlg.result_servers: return self._push_undo() added, replaced = 0, 0 - existing = {s.name: i for i, s in enumerate(self.servers)} for name, data in dlg.result_servers.items(): - if name in existing: - ans = QMessageBox.question( - self, - "Server exists", - f"“{name}” already exists. Replace it?\n\nYes = replace · No = keep both (renamed)", - QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, - ) - if ans == QMessageBox.StandardButton.Yes: - self.servers[existing[name]].data = data - replaced += 1 - continue - new = f"{name}-2" - k = 3 - names = {s.name for s in self.servers} - while new in names: - new = f"{name}-{k}" - k += 1 - name = new - self.servers.append(core.ServerEntry(name, data, True)) - added += 1 + a, r = self._import_server(name, data) + added += int(a) + replaced += int(r) self._refresh_tables(select_index=len(self.servers) - 1) self._mark_dirty() self.status.setText(f"Imported {added} added, {replaced} replaced. Review and Save.") @@ -1719,11 +2070,14 @@ class MainWindow(QMainWindow): return # Stale-file check: if the file changed on disk since we loaded it, prompt. - disk_mtime = core.config_mtime(self.current_profile.path) + # Compare mtime AND size (not mtime alone) so a concurrent external write + # that lands within the mtime resolution window, or that restores the + # original mtime, still gets caught. + disk_stat = core.config_fingerprint(self.current_profile.path) if ( - disk_mtime is not None - and self._loaded_mtime is not None - and disk_mtime != self._loaded_mtime + disk_stat is not None + and self._loaded_stat is not None + and disk_stat != self._loaded_stat ): changed_keys, server_diff = core.external_change_summary( self.full_config, self.current_profile.path @@ -1744,7 +2098,7 @@ class MainWindow(QMainWindow): QMessageBox.critical(self, "Save failed", str(e)) return self.full_config = fresh - self._loaded_mtime = core.config_mtime(self.current_profile.path) + self._loaded_stat = core.config_fingerprint(self.current_profile.path) self.current_profile.config_exists = True self.dirty = False self.save_btn.setEnabled(False) @@ -1753,6 +2107,7 @@ class MainWindow(QMainWindow): f"Merged & saved {self.current_profile.path}{bnote}" f" · Restart {self.current_profile.label} to apply." ) + self._offer_restart_button() return # else OVERWRITE: fall through to normal write @@ -1762,7 +2117,7 @@ class MainWindow(QMainWindow): except Exception as e: QMessageBox.critical(self, "Save failed", str(e)) return - self._loaded_mtime = core.config_mtime(self.current_profile.path) + self._loaded_stat = core.config_fingerprint(self.current_profile.path) self.current_profile.config_exists = True self.dirty = False self.save_btn.setEnabled(False) @@ -1770,6 +2125,29 @@ class MainWindow(QMainWindow): self.status.setText( f"Saved {self.current_profile.path}{bnote} · Restart {self.current_profile.label} to apply." ) + self._offer_restart_button() + + # --- restart Claude Desktop (issue #9) -------------------------------- # + def _offer_restart_button(self): + """Show the 'Restart Claude Desktop' button after a successful save, + but only when the just-saved profile is Claude Desktop -- restarting + makes no sense for Claude Code, which has no GUI process to bounce.""" + if self.current_profile and core.profile_targets_claude_desktop(self.current_profile): + self.restart_btn.show() + else: + self.restart_btn.hide() + + def _restart_claude_desktop(self): + self.restart_btn.setEnabled(False) + try: + result = core.restart_claude_desktop() + finally: + self.restart_btn.setEnabled(True) + self.restart_btn.hide() + if result.success: + self.status.setText(f"{self.status.text()} · {result.detail}") + else: + QMessageBox.warning(self, "Restart failed", result.detail) def _restore_from_backup(self): if not self.current_profile: @@ -1798,6 +2176,7 @@ class MainWindow(QMainWindow): # --- dirty / status -------------------------------------------------- # def _mark_dirty(self): self.dirty = True + self.restart_btn.hide() self._validate() self._update_status(saved=False) @@ -1835,15 +2214,16 @@ class MainWindow(QMainWindow): QMessageBox.warning(self, "Couldn't import", f"{Path(path).name}:\n{ex}") continue self._push_undo() + added, replaced = 0, 0 for name, data in servers.items(): - names = {s.name for s in self.servers} - if name in names: - name = f"{name}-imported" - self.servers.append(core.ServerEntry(name, data, True)) + a, r = self._import_server(name, data) + added += int(a) + replaced += int(r) self._refresh_tables(select_index=len(self.servers) - 1) self._mark_dirty() self.status.setText( - f"Imported {len(servers)} server(s) from {Path(path).name}. Review and Save." + f"Imported {added} added, {replaced} replaced from {Path(path).name}. " + "Review and Save." ) break diff --git a/bcc_core.py b/bcc_core.py index dc41ba3..0ecce10 100644 --- a/bcc_core.py +++ b/bcc_core.py @@ -29,6 +29,7 @@ import threading import time from dataclasses import dataclass from pathlib import Path +from typing import NamedTuple from urllib.parse import urlparse CONFIG_FILENAME = "claude_desktop_config.json" @@ -46,6 +47,112 @@ MAX_BACKUPS = 15 KNOWN_FIELDS = {"command", "args", "env", "url", "type", "headers"} +# --------------------------------------------------------------------------- # +# Version / update checking +# +# __version__ is the single source of truth for the app version (must match +# pyproject.toml's [project] version). The About dialog and the update +# checker both read this constant instead of hard-coding a version string. +# +# The update checker is notify-only: it reads release metadata from the +# repo's Gitea releases API and NEVER downloads or replaces the running +# binary. All network I/O here is fail-quiet (returns None on any problem) +# so it's safe to run unattended, off the UI thread, at startup. +# --------------------------------------------------------------------------- # +__version__ = "1.2.0" + +REPO_URL = "https://git.avezzano.io/the_og/better-claude-config" +ISSUES_URL = f"{REPO_URL}/issues" +RELEASES_URL = f"{REPO_URL}/releases" +LICENSE_URL = f"{REPO_URL}/raw/branch/main/LICENSE" + +# Public repo -> anonymously reachable, no auth/token needed or embedded. +_RELEASES_API_URL = ( + "https://git.avezzano.io/api/v1/repos/the_og/better-claude-config/releases/latest" +) + + +def parse_version(v: str) -> tuple[int, ...]: + """ + Parse a version string into a tuple of ints for numeric comparison. + + Strips a leading 'v' ("v1.2.3" -> "1.2.3") and any pre-release/build + metadata after a '-' or '+' ("1.2.3-beta.1" -> "1.2.3"). Stops at the + first non-numeric dotted component. Empty or entirely non-numeric input + returns an empty tuple rather than raising, so a malformed tag from a + flaky API response degrades gracefully instead of crashing the caller. + """ + s = (v or "").strip() + if s[:1].lower() == "v": + s = s[1:] + s = re.split(r"[-+]", s, maxsplit=1)[0] + parts: list[int] = [] + for chunk in s.split("."): + m = re.match(r"\d+", chunk) + if not m: + break + parts.append(int(m.group())) + return tuple(parts) + + +def is_newer_version(current: str, candidate: str) -> bool: + """ + True if `candidate` is a strictly newer version than `current`. + + Comparison is purely numeric (major.minor.patch, ...) — NEVER a lexical + string compare, so "v2.0.0" vs "v10.0.0" resolves correctly instead of + sorting "2" after "10". Tuples of differing length are zero-padded before + comparing, so "1.2" and "1.2.0" are correctly treated as equal. + + An unparseable `candidate` always yields False (nothing to report). An + unparseable `current` is treated as "0" for comparison purposes — a + malformed local version shouldn't silently suppress a real update. + """ + cur = parse_version(current) + new = parse_version(candidate) + if not new: + return False + width = max(len(cur), len(new), 1) + cur = cur + (0,) * (width - len(cur)) + new = new + (0,) * (width - len(new)) + return new > cur + + +def fetch_latest_release(timeout: float = 4.0) -> dict | None: + """ + Query the repo's (public, anonymous) Gitea releases API for the latest + release. Returns {"version": "", "url": ""} on + success, or None on ANY failure: network error, timeout, bad status, + malformed JSON, or a response missing tag_name. + + Fail-quiet by design — this is meant to be called off the UI thread + (see UpdateCheckWorker in bcc.py) for both the About dialog's "Check for + updates" button and an optional silent startup check. Never downloads or + touches any binary; this only ever reads release metadata. + """ + import urllib.error + import urllib.request + + req = urllib.request.Request( + _RELEASES_API_URL, + headers={ + "Accept": "application/json", + "User-Agent": f"BetterClaudeConfig/{__version__}", + }, + ) + try: + with urllib.request.urlopen(req, timeout=timeout) as r: + payload = json.loads(r.read().decode("utf-8")) + except (urllib.error.URLError, TimeoutError, ValueError, OSError): + return None + if not isinstance(payload, dict): + return None + tag = payload.get("tag_name") + if not tag or not isinstance(tag, str): + return None + return {"version": tag, "url": payload.get("html_url") or RELEASES_URL} + + # --------------------------------------------------------------------------- # # Data model # --------------------------------------------------------------------------- # @@ -135,6 +242,17 @@ def profile_from_path(path: str | os.PathLike) -> Profile: return Profile(label=label, path=p, config_exists=p.is_file()) +def profile_targets_claude_desktop(profile: Profile) -> bool: + """ + True when `profile` points at a Claude Desktop config + (claude_desktop_config.json), as opposed to Claude Code (~/.claude.json + or the legacy ~/.claude/settings.json). Used to gate Desktop-only actions + like "Restart Claude Desktop" so they never show up for a Claude Code + profile -- restarting the CLI makes no sense. + """ + return Path(profile.path).name == CONFIG_FILENAME + + # --------------------------------------------------------------------------- # # Load / extract / apply # --------------------------------------------------------------------------- # @@ -189,6 +307,25 @@ def extract_servers(cfg: dict) -> list[ServerEntry]: return out +def resolve_name_collision(name: str, existing: set[str]) -> str: + """ + Return a name guaranteed not to collide with `existing`. + + If `name` isn't already taken it's returned unchanged. Otherwise a + `-2`, `-3`, ... suffix is appended until the result is unique — this is + the "keep both (renamed)" branch used by paste/import when the user + doesn't want to overwrite an existing server of the same name. + """ + if name not in existing: + return name + n = 2 + candidate = f"{name}-{n}" + while candidate in existing: + n += 1 + candidate = f"{name}-{n}" + return candidate + + def apply_servers(cfg: dict, servers: list[ServerEntry]) -> dict: """ Write the server list back into `cfg` in place, preserving every other key @@ -380,6 +517,30 @@ def config_mtime(path: Path | str) -> float | None: return None +class ConfigStat(NamedTuple): + """A snapshot of a config file's mtime + size. + + Pairing size with mtime hardens stale-file detection beyond bare mtime + equality: a concurrent external write can land within the filesystem's + mtime resolution (e.g. same-second writes on ext4/HFS+) or have its mtime + restored by the writing process, in which case mtime alone would miss the + change. Comparing both fields catches those cases without the cost of a + full content hash. + """ + + mtime: float + size: int + + +def config_fingerprint(path: Path | str) -> ConfigStat | None: + """Return the file's (mtime, size) snapshot, or None if it does not exist.""" + try: + st = Path(path).stat() + except OSError: + return None + return ConfigStat(st.st_mtime, st.st_size) + + def external_change_summary(original_cfg: dict, path: Path | str) -> tuple[list[str], str]: """ Compare original_cfg (what BCC loaded) with the current on-disk state. @@ -474,7 +635,7 @@ def _normalize_unicode(text: str, notes: list[str]) -> str: out = text for junk in _JUNK_CHARS: out = out.replace(junk, "") - out = out.replace(" ", " ") # non-breaking space + out = out.replace(chr(0xA0), " ") # non-breaking space (defensive: avoid a literal char here) for smart, ascii_q in _QUOTE_MAP.items(): out = out.replace(smart, ascii_q) if out != text: @@ -1270,6 +1431,25 @@ def diagnostics_text(name: str, data: dict) -> str: return "\n".join(L) +def server_log_path(name: str) -> Path | None: + """ + The platform-specific Claude Desktop MCP server log file for `name`, or + None if it doesn't exist yet (nothing has been logged for this server). + + macOS : ~/Library/Logs/Claude/mcp-server-.log (one file per server) + Windows: %APPDATA%\\Claude\\logs\\mcp.log (one shared file) + Other platforms: Claude Desktop doesn't ship a log in a known location -> None. + """ + if sys.platform == "darwin": + p = Path.home() / "Library" / "Logs" / "Claude" / f"mcp-server-{name}.log" + elif sys.platform.startswith("win"): + appdata = Path(os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming")) + p = appdata / "Claude" / "logs" / "mcp.log" + else: + return None + return p if p.is_file() else None + + _STDERR_CAP = 4096 # bytes @@ -1459,3 +1639,91 @@ def pin_command_path(data: dict, path: str | None = None) -> tuple[dict, str | N out["args"] = args return out, f"'{c}' → {resolved}" return data, None + + +# --------------------------------------------------------------------------- # +# Restart Claude Desktop (issue #9) +# +# Scoped strictly to Claude DESKTOP, the GUI app -- never Claude Code (the +# CLI), which has no long-running process to bounce. Callers should gate this +# behind profile_targets_claude_desktop() before offering it in the UI. +# --------------------------------------------------------------------------- # +class RestartResult(NamedTuple): + """Outcome of a restart_claude_desktop() attempt.""" + + success: bool + detail: str + + +def _run_quiet(cmd: list[str]) -> None: + """Best-effort fire-and-forget command. Never raises: a nonzero exit (e.g. + pkill finding nothing to kill) is expected and not an error.""" + with contextlib.suppress(OSError): + subprocess.run(cmd, capture_output=True) + + +def _restart_claude_desktop_macos() -> RestartResult: + _run_quiet(["pkill", "-x", "Claude"]) + try: + result = subprocess.run(["open", "-a", "Claude"], capture_output=True, text=True) + except OSError as e: + return RestartResult(False, f"Couldn't launch Claude Desktop: {e}") + if result.returncode != 0: + detail = (result.stderr or result.stdout or "").strip() or "'open -a Claude' failed" + return RestartResult(False, detail) + return RestartResult(True, "Claude Desktop restarted.") + + +def _claude_windows_start_menu_shortcut() -> Path: + appdata = os.environ.get("APPDATA", str(Path.home())) + return Path(appdata) / "Microsoft" / "Windows" / "Start Menu" / "Programs" / "Claude.lnk" + + +def _restart_claude_desktop_windows() -> RestartResult: + _run_quiet(["taskkill", "/IM", "Claude.exe", "/F"]) + shortcut = _claude_windows_start_menu_shortcut() + try: + # `cmd /c start "" ` launches detached, the same as double-clicking + # the Start-menu shortcut, and returns immediately. + result = subprocess.run( + ["cmd", "/c", "start", "", str(shortcut)], capture_output=True, text=True + ) + except OSError as e: + return RestartResult(False, f"Couldn't launch Claude Desktop: {e}") + if result.returncode != 0: + detail = ( + result.stderr or result.stdout or "" + ).strip() or "failed to relaunch Claude Desktop" + return RestartResult(False, detail) + return RestartResult(True, "Claude Desktop restarted.") + + +def _restart_claude_desktop_linux() -> RestartResult: + _run_quiet(["pkill", "claude"]) + try: + # The Linux launcher is the app itself (no "open"-style helper), so it + # has to be started detached rather than waited on. + subprocess.Popen( + ["claude"], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + except OSError as e: + return RestartResult(False, f"Couldn't launch Claude Desktop: {e}") + return RestartResult(True, "Claude Desktop restarted.") + + +def restart_claude_desktop() -> RestartResult: + """ + Kill and relaunch the Claude Desktop app so a freshly saved config takes + effect. The app not currently running is NOT a failure -- pkill/taskkill + exiting non-zero just means "nothing to kill", and we go straight to + relaunching. Only a failed relaunch is reported as success=False. + """ + if sys.platform == "darwin": + return _restart_claude_desktop_macos() + if sys.platform.startswith("win"): + return _restart_claude_desktop_windows() + return _restart_claude_desktop_linux() diff --git a/pyproject.toml b/pyproject.toml index 12d09ef..80de5d6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "better-claude-config" -version = "1.1.0" +version = "1.2.0" description = "Cross-platform GUI for editing the mcpServers block of Claude Desktop and Claude Code configs" readme = "README.md" license = { file = "LICENSE" } diff --git a/tests/test_core.py b/tests/test_core.py index aaf2faf..880a94b 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -2,7 +2,10 @@ proper test functions with tmp_path/monkeypatch fixtures).""" import json +import os import sys +import urllib.error +import urllib.request import pytest @@ -520,6 +523,42 @@ def test_config_mtime_returns_none_for_missing_file(tmp_path): assert c.config_mtime(tmp_path / "nonexistent.json") is None +def test_config_fingerprint_returns_mtime_and_size_for_existing_file(tmp_path): + f = tmp_path / "cfg.json" + f.write_text("{}") + fp = c.config_fingerprint(f) + assert isinstance(fp, c.ConfigStat) + assert isinstance(fp.mtime, float) + assert fp.size == f.stat().st_size + + +def test_config_fingerprint_returns_none_for_missing_file(tmp_path): + assert c.config_fingerprint(tmp_path / "nonexistent.json") is None + + +def test_config_fingerprint_detects_size_change_when_mtime_is_unchanged(tmp_path): + """ + Guards against the exact hole bare-mtime comparison has: an external write + that lands within the filesystem's mtime resolution (or that restores the + original mtime) must still be caught, because the file's size differs. + """ + f = tmp_path / "cfg.json" + f.write_text('{"mcpServers": {}}') + original = c.config_fingerprint(f) + original_mtime_ns = f.stat().st_mtime_ns + + # Overwrite with materially different (larger) content, then force the + # mtime back to its original value -- simulating a same-second external + # write, or a writer that preserves mtime. + f.write_text('{"mcpServers": {"new-server": {"command": "node", "args": ["a", "b", "c"]}}}') + os.utime(f, ns=(original_mtime_ns, original_mtime_ns)) + + updated = c.config_fingerprint(f) + assert updated.mtime == original.mtime # mtime alone would say "unchanged" + assert updated.size != original.size # size catches what mtime missed + assert updated != original # the fingerprint as a whole detects the change + + def test_external_change_summary_detects_non_server_key_change(tmp_path): cfgpath = tmp_path / "cfg.json" original = {"numStartups": 1, "mcpServers": {"s": {"command": "node"}}} @@ -638,3 +677,380 @@ def test_args_secret_warning_env_not_triggered(): def test_args_secret_warning_empty(): assert c.args_secret_warning({}) is None assert c.args_secret_warning({"args": []}) is None + + +# --------------------------------------------------------------------------- # +# resolve_name_collision (paste/import duplicate-name handling — issue #8) +# --------------------------------------------------------------------------- # +def test_resolve_name_collision_no_conflict_returns_unchanged(): + assert c.resolve_name_collision("brave-search", {"other"}) == "brave-search" + + +def test_resolve_name_collision_single_conflict_appends_dash_two(): + assert c.resolve_name_collision("brave-search", {"brave-search"}) == "brave-search-2" + + +def test_resolve_name_collision_skips_taken_suffixes(): + existing = {"brave-search", "brave-search-2", "brave-search-3"} + assert c.resolve_name_collision("brave-search", existing) == "brave-search-4" + + +def test_resolve_name_collision_empty_existing_set(): + assert c.resolve_name_collision("brave-search", set()) == "brave-search" + + +def test_resolve_name_collision_repeated_calls_stay_unique(): + """Simulates dropping the same file twice in a row: each resolved name + must be fed back into `existing` before resolving the next one, or two + servers could end up sharing a name (and silently collapse into one + when apply_servers() rebuilds its dict at save time).""" + existing = {"brave-search"} + first = c.resolve_name_collision("brave-search", existing) + existing.add(first) + second = c.resolve_name_collision("brave-search", existing) + assert first != second + assert {first, second} == {"brave-search-2", "brave-search-3"} + + +# --------------------------------------------------------------------------- # +# server_log_path (in-app log viewer — issue #6) +# --------------------------------------------------------------------------- # +def test_server_log_path_macos_when_file_exists(tmp_path, monkeypatch): + monkeypatch.setattr(sys, "platform", "darwin") + monkeypatch.setattr(c.Path, "home", lambda: tmp_path) + log_dir = tmp_path / "Library" / "Logs" / "Claude" + log_dir.mkdir(parents=True) + (log_dir / "mcp-server-brave-search.log").write_text("hello") + + result = c.server_log_path("brave-search") + assert result == log_dir / "mcp-server-brave-search.log" + + +def test_server_log_path_macos_none_when_missing(tmp_path, monkeypatch): + monkeypatch.setattr(sys, "platform", "darwin") + monkeypatch.setattr(c.Path, "home", lambda: tmp_path) + assert c.server_log_path("brave-search") is None + + +def test_server_log_path_windows_when_file_exists(tmp_path, monkeypatch): + # server_log_path branches on sys.platform only (never os.name), so the + # test doesn't have to touch os.name -- mutating that globally mid-test + # would also flip which concrete Path subclass pathlib hands out. + monkeypatch.setattr(sys, "platform", "win32") + monkeypatch.setenv("APPDATA", str(tmp_path)) + log_dir = tmp_path / "Claude" / "logs" + log_dir.mkdir(parents=True) + (log_dir / "mcp.log").write_text("hello") + + result = c.server_log_path("brave-search") + assert result == log_dir / "mcp.log" + + +def test_server_log_path_windows_none_when_missing(tmp_path, monkeypatch): + monkeypatch.setattr(sys, "platform", "win32") + monkeypatch.setenv("APPDATA", str(tmp_path)) + assert c.server_log_path("brave-search") is None + + +def test_server_log_path_unsupported_platform_returns_none(tmp_path, monkeypatch): + monkeypatch.setattr(sys, "platform", "linux") + monkeypatch.setattr(c.Path, "home", lambda: tmp_path) + assert c.server_log_path("brave-search") is None + + +# --------------------------------------------------------------------------- # +# restart_claude_desktop / profile_targets_claude_desktop (issue #9) +# --------------------------------------------------------------------------- # +def test_profile_targets_claude_desktop_true_for_desktop_config(): + p = c.Profile(label="Claude", path="/x/Claude/claude_desktop_config.json", config_exists=True) + assert c.profile_targets_claude_desktop(p) + + +def test_profile_targets_claude_desktop_false_for_claude_code(): + p = c.Profile(label="Claude Code", path="/home/me/.claude.json", config_exists=True) + assert not c.profile_targets_claude_desktop(p) + + +def test_profile_targets_claude_desktop_false_for_legacy_settings(): + p = c.Profile( + label="Claude Code (legacy settings.json)", + path="/home/me/.claude/settings.json", + config_exists=True, + ) + assert not c.profile_targets_claude_desktop(p) + + +class _FakeCompletedProcess: + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +def test_restart_claude_desktop_macos_commands(monkeypatch): + """macOS: pkill -x "Claude" then open -a Claude, in that order.""" + calls = [] + + def fake_run(cmd, **kwargs): + calls.append(cmd) + return _FakeCompletedProcess(returncode=0) + + monkeypatch.setattr(c.sys, "platform", "darwin") + monkeypatch.setattr(c.subprocess, "run", fake_run) + result = c.restart_claude_desktop() + assert result.success + assert calls[0] == ["pkill", "-x", "Claude"] + assert calls[1] == ["open", "-a", "Claude"] + + +def test_restart_claude_desktop_macos_pkill_not_running_is_fine(monkeypatch): + """pkill exiting non-zero (nothing to kill) must NOT be treated as failure.""" + + def fake_run(cmd, **kwargs): + if cmd[0] == "pkill": + return _FakeCompletedProcess(returncode=1) # no matching process + return _FakeCompletedProcess(returncode=0) + + monkeypatch.setattr(c.sys, "platform", "darwin") + monkeypatch.setattr(c.subprocess, "run", fake_run) + result = c.restart_claude_desktop() + assert result.success + + +def test_restart_claude_desktop_macos_relaunch_failure_reported(monkeypatch): + def fake_run(cmd, **kwargs): + if cmd[0] == "open": + return _FakeCompletedProcess(returncode=1, stderr="Unable to find application") + return _FakeCompletedProcess(returncode=1) + + monkeypatch.setattr(c.sys, "platform", "darwin") + monkeypatch.setattr(c.subprocess, "run", fake_run) + result = c.restart_claude_desktop() + assert not result.success + assert "Unable to find application" in result.detail + + +def test_restart_claude_desktop_macos_pkill_binary_missing_does_not_raise(monkeypatch): + """pkill raising OSError (binary missing) must be swallowed, not propagated -- + relaunch is still attempted.""" + + def fake_run(cmd, **kwargs): + if cmd[0] == "pkill": + raise OSError("pkill not found") + return _FakeCompletedProcess(returncode=0) + + monkeypatch.setattr(c.sys, "platform", "darwin") + monkeypatch.setattr(c.subprocess, "run", fake_run) + result = c.restart_claude_desktop() + assert result.success + + +def test_restart_claude_desktop_windows_commands(monkeypatch): + """Windows: taskkill the process, then relaunch via the Start-menu shortcut.""" + calls = [] + + def fake_run(cmd, **kwargs): + calls.append(cmd) + return _FakeCompletedProcess(returncode=0) + + monkeypatch.setattr(c.sys, "platform", "win32") + monkeypatch.setattr(c.subprocess, "run", fake_run) + result = c.restart_claude_desktop() + assert result.success + assert calls[0] == ["taskkill", "/IM", "Claude.exe", "/F"] + assert calls[1][:3] == ["cmd", "/c", "start"] + assert calls[1][-1].endswith("Claude.lnk") + + +def test_restart_claude_desktop_windows_relaunch_failure_reported(monkeypatch): + def fake_run(cmd, **kwargs): + if cmd[0] == "cmd": + return _FakeCompletedProcess(returncode=1, stderr="not found") + return _FakeCompletedProcess(returncode=0) + + monkeypatch.setattr(c.sys, "platform", "win32") + monkeypatch.setattr(c.subprocess, "run", fake_run) + result = c.restart_claude_desktop() + assert not result.success + assert "not found" in result.detail + + +def test_restart_claude_desktop_linux_commands(monkeypatch): + """Linux: pkill claude, then relaunch via a detached Popen (no waiting).""" + run_calls = [] + popen_calls = [] + + def fake_run(cmd, **kwargs): + run_calls.append(cmd) + return _FakeCompletedProcess(returncode=0) + + class _FakePopen: + def __init__(self, cmd, **kwargs): + popen_calls.append((cmd, kwargs)) + + monkeypatch.setattr(c.sys, "platform", "linux") + monkeypatch.setattr(c.subprocess, "run", fake_run) + monkeypatch.setattr(c.subprocess, "Popen", _FakePopen) + result = c.restart_claude_desktop() + assert result.success + assert run_calls == [["pkill", "claude"]] + cmd, kwargs = popen_calls[0] + assert cmd == ["claude"] + assert kwargs.get("start_new_session") is True + + +def test_restart_claude_desktop_linux_popen_failure_reported(monkeypatch): + def fake_run(cmd, **kwargs): + return _FakeCompletedProcess(returncode=0) + + def fake_popen(cmd, **kwargs): + raise OSError("no such file or directory") + + monkeypatch.setattr(c.sys, "platform", "linux") + monkeypatch.setattr(c.subprocess, "run", fake_run) + monkeypatch.setattr(c.subprocess, "Popen", fake_popen) + result = c.restart_claude_desktop() + assert not result.success + assert "Claude Desktop" in result.detail + + +# --------------------------------------------------------------------------- # +# Version / update checking +# --------------------------------------------------------------------------- # +def test_dunder_version_matches_pyproject(): + # Guards against the version drifting out of sync between the two places + # a human might bump it. + assert c.__version__ == "1.2.0" + + +def test_parse_version_basic(): + assert c.parse_version("1.2.3") == (1, 2, 3) + + +def test_parse_version_strips_v_prefix(): + assert c.parse_version("v1.2.3") == (1, 2, 3) + + +def test_parse_version_strips_prerelease_suffix(): + assert c.parse_version("1.2.3-beta.1") == (1, 2, 3) + assert c.parse_version("1.2.3+build5") == (1, 2, 3) + + +def test_parse_version_malformed_returns_empty_tuple(): + assert c.parse_version("not-a-version") == () + assert c.parse_version("") == () + assert c.parse_version(None) == () + + +def test_parse_version_stops_at_first_non_numeric_component(): + assert c.parse_version("1.2.rc1") == (1, 2) + + +def test_is_newer_version_true_when_candidate_is_newer(): + assert c.is_newer_version("1.1.0", "1.2.0") is True + + +def test_is_newer_version_false_when_candidate_is_older(): + assert c.is_newer_version("1.2.0", "1.1.0") is False + + +def test_is_newer_version_false_when_equal(): + assert c.is_newer_version("1.1.0", "1.1.0") is False + + +def test_is_newer_version_equal_with_v_prefix(): + assert c.is_newer_version("1.1.0", "v1.1.0") is False + + +def test_is_newer_version_numeric_not_lexical(): + # A lexical/string compare would say "10.0.0" < "2.0.0" ("1" < "2"). + assert c.is_newer_version("2.0.0", "10.0.0") is True + assert c.is_newer_version("10.0.0", "2.0.0") is False + + +def test_is_newer_version_handles_differing_length(): + assert c.is_newer_version("1.2", "1.2.0") is False + assert c.is_newer_version("1.2.0", "1.2") is False + assert c.is_newer_version("1.2", "1.3") is True + + +def test_is_newer_version_malformed_candidate_is_never_newer(): + assert c.is_newer_version("1.1.0", "not-a-version") is False + assert c.is_newer_version("1.1.0", "") is False + + +def test_is_newer_version_malformed_current_does_not_crash(): + # Shouldn't raise; a valid candidate is reported as newer than "unknown". + assert c.is_newer_version("garbage", "1.0.0") is True + assert c.is_newer_version("garbage", "not-a-version-either") is False + + +class _FakeHTTPResponse: + def __init__(self, data: bytes): + self._data = data + + def read(self): + return self._data + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + +def test_fetch_latest_release_ok(monkeypatch): + payload = json.dumps( + { + "tag_name": "v1.2.0", + "html_url": "https://git.avezzano.io/the_og/better-claude-config/releases/tag/v1.2.0", + } + ).encode("utf-8") + monkeypatch.setattr( + urllib.request, "urlopen", lambda req, timeout=None: _FakeHTTPResponse(payload) + ) + result = c.fetch_latest_release() + assert result == { + "version": "v1.2.0", + "url": "https://git.avezzano.io/the_og/better-claude-config/releases/tag/v1.2.0", + } + + +def test_fetch_latest_release_falls_back_to_releases_url_when_no_html_url(monkeypatch): + payload = json.dumps({"tag_name": "v1.2.0"}).encode("utf-8") + monkeypatch.setattr( + urllib.request, "urlopen", lambda req, timeout=None: _FakeHTTPResponse(payload) + ) + result = c.fetch_latest_release() + assert result == {"version": "v1.2.0", "url": c.RELEASES_URL} + + +def test_fetch_latest_release_network_failure_returns_none(monkeypatch): + def boom(req, timeout=None): + raise urllib.error.URLError("no network") + + monkeypatch.setattr(urllib.request, "urlopen", boom) + assert c.fetch_latest_release() is None + + +def test_fetch_latest_release_timeout_returns_none(monkeypatch): + def boom(req, timeout=None): + raise TimeoutError("timed out") + + monkeypatch.setattr(urllib.request, "urlopen", boom) + assert c.fetch_latest_release() is None + + +def test_fetch_latest_release_missing_tag_returns_none(monkeypatch): + payload = json.dumps({"html_url": "https://example.com"}).encode("utf-8") + monkeypatch.setattr( + urllib.request, "urlopen", lambda req, timeout=None: _FakeHTTPResponse(payload) + ) + assert c.fetch_latest_release() is None + + +def test_fetch_latest_release_malformed_json_returns_none(monkeypatch): + monkeypatch.setattr( + urllib.request, "urlopen", lambda req, timeout=None: _FakeHTTPResponse(b"not json") + ) + assert c.fetch_latest_release() is None