Merge branch 'feat/17-stale-size-hardening' into integration/v1.2.0

This commit is contained in:
Cowork Supervisor
2026-07-07 22:10:59 -04:00
3 changed files with 74 additions and 9 deletions
+11 -8
View File
@@ -1104,7 +1104,7 @@ class MainWindow(QMainWindow):
self.full_config: dict = {} self.full_config: dict = {}
self.servers: list[core.ServerEntry] = [] self.servers: list[core.ServerEntry] = []
self.current_profile: core.Profile | None = None self.current_profile: core.Profile | None = None
self._loaded_mtime: float | None = None self._loaded_stat: core.ConfigStat | None = None
self.dirty = False self.dirty = False
self._suppress_table = False self._suppress_table = False
self._suppress_sel = False self._suppress_sel = False
@@ -1380,7 +1380,7 @@ class MainWindow(QMainWindow):
return return
self.full_config = cfg self.full_config = cfg
repaired = True repaired = True
self._loaded_mtime = core.config_mtime(profile.path) self._loaded_stat = core.config_fingerprint(profile.path)
self.current_profile = profile self.current_profile = profile
self.servers = core.extract_servers(self.full_config) self.servers = core.extract_servers(self.full_config)
self.dirty = False self.dirty = False
@@ -1723,11 +1723,14 @@ class MainWindow(QMainWindow):
return return
# Stale-file check: if the file changed on disk since we loaded it, prompt. # 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 ( if (
disk_mtime is not None disk_stat is not None
and self._loaded_mtime is not None and self._loaded_stat is not None
and disk_mtime != self._loaded_mtime and disk_stat != self._loaded_stat
): ):
changed_keys, server_diff = core.external_change_summary( changed_keys, server_diff = core.external_change_summary(
self.full_config, self.current_profile.path self.full_config, self.current_profile.path
@@ -1748,7 +1751,7 @@ class MainWindow(QMainWindow):
QMessageBox.critical(self, "Save failed", str(e)) QMessageBox.critical(self, "Save failed", str(e))
return return
self.full_config = fresh 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.current_profile.config_exists = True
self.dirty = False self.dirty = False
self.save_btn.setEnabled(False) self.save_btn.setEnabled(False)
@@ -1766,7 +1769,7 @@ class MainWindow(QMainWindow):
except Exception as e: except Exception as e:
QMessageBox.critical(self, "Save failed", str(e)) QMessageBox.critical(self, "Save failed", str(e))
return 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.current_profile.config_exists = True
self.dirty = False self.dirty = False
self.save_btn.setEnabled(False) self.save_btn.setEnabled(False)
+26 -1
View File
@@ -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"
@@ -399,6 +400,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.
@@ -493,7 +518,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(chr(0xA0), " ") # non-breaking space (defensive: avoid a literal char here)
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:
+37
View File
@@ -2,6 +2,7 @@
proper test functions with tmp_path/monkeypatch fixtures).""" proper test functions with tmp_path/monkeypatch fixtures)."""
import json import json
import os
import sys import sys
import pytest import pytest
@@ -520,6 +521,42 @@ def test_config_mtime_returns_none_for_missing_file(tmp_path):
assert c.config_mtime(tmp_path / "nonexistent.json") is None 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): def test_external_change_summary_detects_non_server_key_change(tmp_path):
cfgpath = tmp_path / "cfg.json" cfgpath = tmp_path / "cfg.json"
original = {"numStartups": 1, "mcpServers": {"s": {"command": "node"}}} original = {"numStartups": 1, "mcpServers": {"s": {"command": "node"}}}