admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(sync): ignore hidden paths relative to watched project (#815)

Browse Source 
Signed-off-by: phernandez <paul@basicmachines.co>
This commit is contained in:
Paul Hernandez
2026-05-11 11:14:03 -05:00
committed by GitHub
parent 55f314237d
commit 4aa0cbdd62
3 changed files with 143 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/sync/watch_service.py
+75 -33
View File
@@ -93,6 +93,7 @@ class WatchService:
self.status_path = app_config.data_dir_path / WATCH_STATUS_JSON
self.status_path.parent.mkdir(parents=True, exist_ok=True)
self._ignore_patterns_cache: dict[Path, Set[str]] = {}
self._sorted_watch_filter_roots: tuple[Path, ...] | None = None
self._sync_service_factory = sync_service_factory
# When set (typically from BASIC_MEMORY_MCP_PROJECT), the watch cycle
# only observes this project. Without it, each `basic-memory mcp --project X`
@@ -126,41 +127,54 @@ class WatchService:
async def _watch_projects_cycle(self, projects: Sequence[Project], stop_event: asyncio.Event):
"""Run one cycle of watching the given projects until stop_event is set."""
project_paths = [project.path for project in projects]
previous_filter_roots = self._sorted_watch_filter_roots
self._sorted_watch_filter_roots = tuple(
sorted(
(Path(project.path).expanduser().resolve() for project in projects),
# Trigger: configured project roots can overlap.
# Why: an enclosing project's hidden directory should still hide descendants.
# Outcome: choose the outermost matching root when checking hidden path parts.
key=lambda project_path: len(project_path.parts),
)
)
async for changes in awatch(
*project_paths,
debounce=self.app_config.sync_delay,
watch_filter=self.filter_changes,
recursive=True,
stop_event=stop_event,
):
# group changes by project and filter using ignore patterns
project_changes = defaultdict(list)
for change, path in changes:
for project in projects:
if self.is_project_path(project, path):
# Check if the file should be ignored based on gitignore patterns
project_path = Path(project.path)
file_path = Path(path)
ignore_patterns = self._get_ignore_patterns(project_path)
try:
async for changes in awatch(
*project_paths,
debounce=self.app_config.sync_delay,
watch_filter=self.filter_changes,
recursive=True,
stop_event=stop_event,
):
# group changes by project and filter using ignore patterns
project_changes = defaultdict(list)
for change, path in changes:
for project in projects:
if self.is_project_path(project, path):
# Check if the file should be ignored based on gitignore patterns
project_path = Path(project.path)
file_path = Path(path)
ignore_patterns = self._get_ignore_patterns(project_path)
if should_ignore_path(file_path, project_path, ignore_patterns):
logger.trace(
f"Ignoring watched file change: {file_path.relative_to(project_path)}"
)
continue
if should_ignore_path(file_path, project_path, ignore_patterns):
logger.trace(
f"Ignoring watched file change: {file_path.relative_to(project_path)}"
)
continue
project_changes[project].append((change, path))
break
project_changes[project].append((change, path))
break
# create coroutines to handle changes
change_handlers = [
self.handle_changes(project, set(changes))
for project, changes in project_changes.items()
]
# create coroutines to handle changes
change_handlers = [
self.handle_changes(project, set(changes))
for project, changes in project_changes.items()
]
# process changes
await asyncio.gather(*change_handlers)
# process changes
await asyncio.gather(*change_handlers)
finally:
self._sorted_watch_filter_roots = previous_filter_roots
async def _select_projects_to_watch(self) -> list[Project]:
"""Return the set of projects this watch cycle should observe.
@@ -267,15 +281,43 @@ class WatchService:
self.state.running = False
await self.write_status()
def filter_changes(self, change: Change, path: str) -> bool: # pragma: no cover
def filter_changes(self, change: Change, path: str) -> bool:
"""Filter to only watch non-hidden files and directories.
Returns:
True if the file should be watched, False if it should be ignored
"""
# Skip hidden directories and files
path_parts = Path(path).parts
path_obj = Path(path).expanduser().resolve()
project_paths = self._sorted_watch_filter_roots
if project_paths is None:
project_paths = tuple(
sorted(
(
Path(entry.path).expanduser().resolve()
for entry in self.app_config.projects.values()
if entry.path
),
# Trigger: direct callers may not run inside a watch cycle.
# Why: tests and one-off calls still need the same hidden-path semantics.
# Outcome: compute the stable outermost-first order only for fallback calls.
key=lambda project_path: len(project_path.parts),
)
)
relative_path = None
for project_path in project_paths:
try:
relative_path = path_obj.relative_to(project_path)
break
except ValueError:
continue
# Trigger: a project may live under a hidden parent such as ~/.claude.
# Why: only dotfiles and dot-directories inside the watched project should be ignored.
# Outcome: hidden parents outside the project root do not mute legitimate project changes.
path_parts = relative_path.parts if relative_path is not None else path_obj.parts
for part in path_parts:
if part.startswith("."):
return False