bcc_core.py: add resolve_name_collision() for paste/import name-collision handling
This commit is contained in:
+162
-1
@@ -29,6 +29,7 @@ import threading
|
|||||||
import time
|
import time
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from typing import NamedTuple
|
||||||
from urllib.parse import urlparse
|
from urllib.parse import urlparse
|
||||||
|
|
||||||
CONFIG_FILENAME = "claude_desktop_config.json"
|
CONFIG_FILENAME = "claude_desktop_config.json"
|
||||||
@@ -134,6 +135,16 @@ def profile_from_path(path: str | os.PathLike) -> Profile:
|
|||||||
label = p.parent.name or p.name
|
label = p.parent.name or p.name
|
||||||
return Profile(label=label, path=p, config_exists=p.is_file())
|
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
|
# Load / extract / apply
|
||||||
@@ -189,6 +200,25 @@ def extract_servers(cfg: dict) -> list[ServerEntry]:
|
|||||||
return out
|
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:
|
def apply_servers(cfg: dict, servers: list[ServerEntry]) -> dict:
|
||||||
"""
|
"""
|
||||||
Write the server list back into `cfg` in place, preserving every other key
|
Write the server list back into `cfg` in place, preserving every other key
|
||||||
@@ -380,6 +410,30 @@ def config_mtime(path: Path | str) -> float | None:
|
|||||||
return 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]:
|
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.
|
Compare original_cfg (what BCC loaded) with the current on-disk state.
|
||||||
@@ -474,7 +528,7 @@ def _normalize_unicode(text: str, notes: list[str]) -> str:
|
|||||||
out = text
|
out = text
|
||||||
for junk in _JUNK_CHARS:
|
for junk in _JUNK_CHARS:
|
||||||
out = out.replace(junk, "")
|
out = out.replace(junk, "")
|
||||||
out = out.replace(" ", " ") # non-breaking space
|
out = out.replace(" ", " ") # non-breaking space
|
||||||
for smart, ascii_q in _QUOTE_MAP.items():
|
for smart, ascii_q in _QUOTE_MAP.items():
|
||||||
out = out.replace(smart, ascii_q)
|
out = out.replace(smart, ascii_q)
|
||||||
if out != text:
|
if out != text:
|
||||||
@@ -1270,6 +1324,25 @@ def diagnostics_text(name: str, data: dict) -> str:
|
|||||||
return "\n".join(L)
|
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-<name>.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 os.name == "nt":
|
||||||
|
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
|
_STDERR_CAP = 4096 # bytes
|
||||||
|
|
||||||
|
|
||||||
@@ -1459,3 +1532,91 @@ def pin_command_path(data: dict, path: str | None = None) -> tuple[dict, str | N
|
|||||||
out["args"] = args
|
out["args"] = args
|
||||||
return out, f"'{c}' → {resolved}"
|
return out, f"'{c}' → {resolved}"
|
||||||
return data, None
|
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 "" <target>` 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()
|
||||||
|
|||||||
Reference in New Issue
Block a user