feat(#101): live hot-reload of external sidecar/config changes #103

Merged
the_og merged 1 commits from feat/101 into main 2026-08-13 00:55:04 -04:00
3 changed files with 352 additions and 1 deletions
+160 -1
View File
@@ -16,7 +16,18 @@ import time
from pathlib import Path from pathlib import Path
from typing import ClassVar from typing import ClassVar
from PySide6.QtCore import QRect, QSettings, QSize, Qt, QThread, QTimer, QUrl, Signal from PySide6.QtCore import (
QEvent,
QFileSystemWatcher,
QRect,
QSettings,
QSize,
Qt,
QThread,
QTimer,
QUrl,
Signal,
)
from PySide6.QtGui import ( from PySide6.QtGui import (
QAction, QAction,
QActionGroup, QActionGroup,
@@ -1125,6 +1136,17 @@ class ServerEditor(QFrame):
self.refresh_dependency(auto_open=False) self.refresh_dependency(auto_open=False)
self._check_args() self._check_args()
def recheck_advisories(self):
"""Re-run the read-only sidecar / permission advisories (#101 hot-reload).
Public entry point for the file-watcher and focus-in fallback: it only
recomputes the warning labels from the current form + the filesystem — it
never touches the form's field values, so it is safe to call regardless of
unsaved edits (it can't clobber them). Delegates to the same recompute the
editor runs on every field change.
"""
self._check_args()
# --- args sanity check ------------------------------------------------ # # --- args sanity check ------------------------------------------------ #
def _current_arg_lines(self) -> list[str]: def _current_arg_lines(self) -> list[str]:
return [ln for ln in self.args.toPlainText().splitlines() if ln.strip() != ""] return [ln for ln in self.args.toPlainText().splitlines() if ln.strip() != ""]
@@ -2180,6 +2202,18 @@ class MainWindow(QMainWindow):
self._test_all_done = 0 self._test_all_done = 0
self._health_tester: SpawnTester | None = None self._health_tester: SpawnTester | None = None
# Hot-reload (#101): watch the selected server's external sidecar path(s)
# and BCC's own loaded config, so changes made outside BCC surface without
# a restart. The watcher fires the debounce timer; the timer re-checks.
self._fs_watcher = QFileSystemWatcher(self)
self._fs_watcher.fileChanged.connect(self._on_fs_signal)
self._fs_watcher.directoryChanged.connect(self._on_fs_signal)
self._fs_debounce = QTimer(self)
self._fs_debounce.setSingleShot(True)
self._fs_debounce.setInterval(300) # coalesce a burst of writes
self._fs_debounce.timeout.connect(self._recheck_external_state)
self._sidecar_fp: tuple | None = None # last observed sidecar state
central = QWidget() central = QWidget()
self.setCentralWidget(central) self.setCentralWidget(central)
root = QVBoxLayout(central) root = QVBoxLayout(central)
@@ -2202,6 +2236,12 @@ class MainWindow(QMainWindow):
self.update_banner = NoticeBanner(self) self.update_banner = NoticeBanner(self)
root.addWidget(self.update_banner) root.addWidget(self.update_banner)
# External-change notice (#101): when BCC's own loaded config is edited
# outside BCC, surface it here with a one-click Reload rather than
# silently overwriting — a live reload must never clobber unsaved edits.
self.reload_banner = NoticeBanner(self)
root.addWidget(self.reload_banner)
# User-draggable divider between the server list and the editor. # User-draggable divider between the server list and the editor.
split = QSplitter(Qt.Orientation.Horizontal) split = QSplitter(Qt.Orientation.Horizontal)
split.setChildrenCollapsible(False) split.setChildrenCollapsible(False)
@@ -2703,6 +2743,8 @@ class MainWindow(QMainWindow):
self._health.clear() # health results are per-profile; a fresh load invalidates them self._health.clear() # health results are per-profile; a fresh load invalidates them
self._refresh_sets_combo() # sets are per-config; repopulate from the loaded file self._refresh_sets_combo() # sets are per-config; repopulate from the loaded file
self._refresh_tables(select_index=0 if self.servers else -1) self._refresh_tables(select_index=0 if self.servers else -1)
self.reload_banner.hide() # loaded fresh: any external-change notice is now stale
self._rewatch_paths() # watch this profile's config + the selection's sidecar (#101)
self._update_status(saved=False) self._update_status(saved=False)
if repaired: if repaired:
self._mark_dirty() self._mark_dirty()
@@ -2993,6 +3035,120 @@ class MainWindow(QMainWindow):
self.copy_btn.setEnabled(entry is not None and len(self.profiles) > 1) self.copy_btn.setEnabled(entry is not None and len(self.profiles) > 1)
self.dup_btn.setEnabled(entry is not None) self.dup_btn.setEnabled(entry is not None)
self.del_btn.setEnabled(entry is not None) self.del_btn.setEnabled(entry is not None)
self._rewatch_paths()
# --- hot-reload (#101) ------------------------------------------------- #
def _watch_targets(self) -> list[Path]:
"""Every external path worth watching for the current view.
BCC's own loaded config (+ its directory), plus the selected server's
sidecar path(s) resolved from its ServerSpec. Directories are watched too
so a file appearing/disappearing — which a watch on a not-yet-existent
file would miss, and which an atomic-rename replace looks like — is still
observed. All entries must exist for QFileSystemWatcher to accept them.
"""
targets: list[Path] = []
if self.current_profile is not None:
cfg = Path(self.current_profile.path)
targets.append(cfg)
targets.append(cfg.parent)
if self.editor.isEnabled():
data = self.editor.dump_data()
targets.extend(core.sidecar_watch_paths(data))
# De-dup, keep only paths that currently exist (the watcher rejects the
# rest; the parent dir covers a not-yet-created file).
seen: set[str] = set()
out: list[Path] = []
for p in targets:
key = str(p)
if key in seen:
continue
seen.add(key)
if p.exists():
out.append(p)
return out
def _rewatch_paths(self):
"""Point the watcher at the current targets and snapshot sidecar state.
Re-applied whenever the selection or the edited command/args change (which
server is selected decides which sidecar to watch) and after a reload.
"""
watcher = self._fs_watcher
existing = watcher.files() + watcher.directories()
if existing:
watcher.removePaths(existing)
wanted = [str(p) for p in self._watch_targets()]
if wanted:
watcher.addPaths(wanted)
self._sidecar_fp = self._current_sidecar_fp()
def _current_sidecar_fp(self) -> tuple | None:
if not self.editor.isEnabled():
return None
return core.sidecar_state_fingerprint(self.editor.dump_data())
def _on_fs_signal(self, _path=None):
"""A watched path changed — coalesce a burst of writes via the debounce."""
self._fs_debounce.start()
def _recheck_external_state(self):
"""Debounced re-check: re-arm the watcher, refresh advisories if state
actually changed, and surface an external edit to BCC's own config.
Runs from both the file-watcher and the focus-in fallback. Never mutates
the editor form or the loaded config — the only content reload is the
user pressing Reload on the banner, so unsaved edits are safe.
"""
# Re-arm: an atomic-rename replace drops the old inode from the watcher,
# so paths must be re-added or the next change goes unseen.
self._rewatch_after_event()
# Sidecar/permission advisories for the selected server (#91/#93). Only
# refresh when the observable state changed, so an unrelated write in the
# watched directory doesn't thrash the panel.
new_fp = self._current_sidecar_fp()
if core.sidecar_state_changed(self._sidecar_fp, new_fp):
self._sidecar_fp = new_fp
self.editor.recheck_advisories()
# BCC's own loaded config edited outside BCC: surface it (non-destructive).
self._check_config_changed_on_disk()
def _rewatch_after_event(self):
"""Re-add any wanted paths the watcher dropped, without disturbing the
sidecar fingerprint (which _recheck_external_state compares itself)."""
watcher = self._fs_watcher
current = set(watcher.files()) | set(watcher.directories())
readd = [str(p) for p in self._watch_targets() if str(p) not in current]
if readd:
watcher.addPaths(readd)
def _check_config_changed_on_disk(self):
if self.current_profile is None:
return
disk = core.config_fingerprint(self.current_profile.path)
if disk is not None and self._loaded_stat is not None and disk != self._loaded_stat:
self.reload_banner.show_notice(
f"{self.current_profile.path} changed on disk (edited outside BCC).",
action_label="Reload from disk",
on_action=lambda: self.load_profile(self.current_profile, confirm=True),
)
else:
# Back in sync (e.g. the user reloaded, or the change was reverted).
self.reload_banner.hide()
def changeEvent(self, event):
"""Re-check external state when the window regains focus (#101).
A cheap, always-works fallback: QFileSystemWatcher can miss changes
(notably atomic-rename replaces, and files that didn't exist when the
watch was set), so a re-check on activation covers the gap.
"""
if event.type() == QEvent.Type.ActivationChange and self.isActiveWindow():
# Go through the debounce so activation + a watcher signal coalesce.
self._fs_debounce.start()
super().changeEvent(event)
def _table_item_changed(self, item: QTableWidgetItem): def _table_item_changed(self, item: QTableWidgetItem):
if self._suppress_table or item.column() != 0: if self._suppress_table or item.column() != 0:
@@ -3047,6 +3203,9 @@ class MainWindow(QMainWindow):
health_item.setToolTip("Not tested since last edit.") health_item.setToolTip("Not tested since last edit.")
self._suppress_table = False self._suppress_table = False
self._refresh_badges() self._refresh_badges()
# The edited command/args may change which package (and thus which
# sidecar) this server resolves to — re-point the watcher (#101).
self._rewatch_paths()
self._mark_dirty() self._mark_dirty()
# --- server actions -------------------------------------------------- # # --- server actions -------------------------------------------------- #
+92
View File
@@ -3100,6 +3100,98 @@ def sidecar_permission_fix_target(
return sidecar_path(spec, platform=platform, environ=environ, home=home) return sidecar_path(spec, platform=platform, environ=environ, home=home)
# --------------------------------------------------------------------------- #
# Hot-reload — live external-change detection (issue #101, epic #94)
#
# #91/#93 compute the sidecar and permission advisories only at profile load /
# server selection, so a config.toml created (or chmod-ed) WHILE BCC runs stays
# invisible until a restart. The OS file-watcher, the focus-in re-check and the
# debounce are GUI wiring (bcc.py). The testable core is two pure functions:
# * which external paths to WATCH for a given server, and
# * a comparable SNAPSHOT of the externally-observable state, so the GUI can
# cheaply decide "did anything the user can see actually change?" and skip a
# redundant (flicker-y) refresh on an unrelated write in the watched dir.
# --------------------------------------------------------------------------- #
def sidecar_watch_paths(
data: dict,
*,
platform: str | None = None,
environ: dict | None = None,
home: str | os.PathLike | None = None,
) -> list[Path]:
"""External filesystem paths whose changes affect a server's advisories.
Returns the resolved sidecar path, its containing directory (so the file
appearing or being deleted — which a file-only watch on a not-yet-existent
path would miss — is still observed), and, when distinct, the README/doc
path (the #91 wrong-path trap: a file a user placed where the docs say but
the server never reads). Empty when the server has no sidecar. Order is
stable and de-duplicated so the GUI can hand it straight to a watcher.
"""
spec = resolve_server_spec(data)
if spec is None:
return []
out: list[Path] = []
p = sidecar_path(spec, platform=platform, environ=environ, home=home)
if p is not None:
out.append(p)
out.append(p.parent)
doc = sidecar_doc_path(spec, home=home)
if doc is not None and doc != p:
out.append(doc)
out.append(doc.parent)
# De-dupe while preserving order (the sidecar file's own dir may equal the
# doc dir, or on Linux the doc path coincides with the real one).
seen: set[Path] = set()
uniq: list[Path] = []
for path in out:
if path not in seen:
seen.add(path)
uniq.append(path)
return uniq
def sidecar_state_fingerprint(
data: dict,
*,
platform: str | None = None,
environ: dict | None = None,
home: str | os.PathLike | None = None,
exists=None,
stat_mode=None,
) -> tuple | None:
"""A comparable snapshot of a server's externally-observable sidecar state.
Folds the #91 sidecar status (existence, precedence, wrong-path) and the #93
permission status (mode + ok) into one hashable tuple. The GUI takes a
fingerprint before and after a filesystem event and refreshes the advisories
only when it changed — so an unrelated write in the watched directory doesn't
thrash the UI. Returns None when the server has no sidecar (nothing to
watch). All inputs are injectable so tests drive a virtual filesystem.
"""
status = sidecar_status(data, platform=platform, environ=environ, home=home, exists=exists)
if status is None:
return None
perm = permission_status(status["path"], platform=platform, stat_mode=stat_mode)
return (
status["exists"],
status["doc_exists"],
status["args_inert"],
status["wrong_path"],
None if perm is None else perm["ok"],
None if perm is None else perm["mode"],
)
def sidecar_state_changed(before: tuple | None, after: tuple | None) -> bool:
"""True when two ``sidecar_state_fingerprint`` snapshots differ.
A thin, explicitly-named equality so the GUI's watcher/focus-in handlers read
intentionally and the "should I refresh?" decision stays covered by tests.
"""
return before != after
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# Validation # Validation
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
+100
View File
@@ -3645,6 +3645,106 @@ def test_sidecar_permission_fix_target():
assert c.sidecar_permission_fix_target({"command": "npx", "args": ["other"]}) is None assert c.sidecar_permission_fix_target({"command": "npx", "args": ["other"]}) is None
# --------------------------------------------------------------------------- #
# Hot-reload — live external-change detection (issue #101)
# --------------------------------------------------------------------------- #
def test_sidecar_watch_paths_covers_file_dir_and_doc():
data = dict(_SSH)
paths = c.sidecar_watch_paths(data, platform="darwin", environ={}, home=_HOME)
posix = [p.as_posix() for p in paths]
real = c.sidecar_path(c.SERVER_SPECS["ssh-mcp"], platform="darwin", environ={}, home=_HOME)
# The resolved sidecar file AND its directory (so create/delete registers).
assert real.as_posix() in posix
assert real.parent.as_posix() in posix
# The README/doc path is distinct on macOS -> also watched.
doc = c.sidecar_doc_path(c.SERVER_SPECS["ssh-mcp"], home=_HOME)
assert doc.as_posix() in posix
# No duplicates.
assert len(posix) == len(set(posix))
def test_sidecar_watch_paths_dedupes_on_linux():
# On Linux the doc path coincides with the real path, so the file + dir are
# each listed once, not twice.
data = dict(_SSH)
paths = c.sidecar_watch_paths(
data, platform="linux", environ={"XDG_CONFIG_HOME": "/cfg"}, home=_HOME
)
posix = [p.as_posix() for p in paths]
assert posix == list(dict.fromkeys(posix)) # order-preserving de-dup is a no-op
assert "/cfg/ssh-mcp/config.toml" in posix
assert "/cfg/ssh-mcp" in posix
def test_sidecar_watch_paths_empty_for_non_sidecar():
assert c.sidecar_watch_paths({"command": "npx", "args": ["other"]}) == []
def test_sidecar_state_fingerprint_reflects_appearance_and_perms():
data = dict(_SSH)
real = c.sidecar_path(c.SERVER_SPECS["ssh-mcp"], platform="darwin", environ={}, home=_HOME)
# File absent -> a fingerprint that encodes "not there, no perms".
absent = c.sidecar_state_fingerprint(
data, platform="darwin", environ={}, home=_HOME, exists=lambda p: False
)
# File present but world-readable.
bad = c.sidecar_state_fingerprint(
data,
platform="darwin",
environ={},
home=_HOME,
exists=lambda p: Path(p) == real,
stat_mode=lambda p: 0o644 if Path(p) == real else 0o700,
)
# File present and tight.
good = c.sidecar_state_fingerprint(
data,
platform="darwin",
environ={},
home=_HOME,
exists=lambda p: Path(p) == real,
stat_mode=lambda p: 0o600 if Path(p) == real else 0o700,
)
assert absent is not None
# Each observable transition changes the fingerprint.
assert c.sidecar_state_changed(absent, bad)
assert c.sidecar_state_changed(bad, good)
assert c.sidecar_state_changed(absent, good)
# Stable when nothing changed.
assert not c.sidecar_state_changed(good, good)
def test_sidecar_state_fingerprint_none_for_non_sidecar():
assert c.sidecar_state_fingerprint({"command": "npx", "args": ["other"]}) is None
def test_sidecar_state_fingerprint_tracks_args_inert_flip():
# args_inert only becomes true once the sidecar file exists AND managed args
# are present; the fingerprint must capture that precedence flip.
data = dict(_SSH) # carries --host/--user (managed args)
real = c.sidecar_path(
c.SERVER_SPECS["ssh-mcp"], platform="linux", environ={"XDG_CONFIG_HOME": "/cfg"}, home=_HOME
)
before = c.sidecar_state_fingerprint(
data,
platform="linux",
environ={"XDG_CONFIG_HOME": "/cfg"},
home=_HOME,
exists=lambda p: False,
stat_mode=lambda p: None,
)
after = c.sidecar_state_fingerprint(
data,
platform="linux",
environ={"XDG_CONFIG_HOME": "/cfg"},
home=_HOME,
exists=lambda p: Path(p) == real,
stat_mode=lambda p: 0o600,
)
assert c.sidecar_state_changed(before, after)
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# Move to environment variable (issue #83) # Move to environment variable (issue #83)
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #