mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
fix(sync): constrain watch service to --project scope (#759)
Signed-off-by: phernandez <paul@basicmachines.co>
This commit is contained in:
@@ -99,18 +99,23 @@ async def initialize_file_sync(
|
||||
)
|
||||
project_repository = ProjectRepository(session_maker)
|
||||
|
||||
# Filter to constrained project if MCP server was started with --project.
|
||||
# Applied to both the initial background sync and the watch service so that
|
||||
# running multiple `basic-memory mcp --project X` processes does not produce
|
||||
# duplicate watchers fighting over the same files.
|
||||
constrained_project = os.environ.get("BASIC_MEMORY_MCP_PROJECT")
|
||||
|
||||
# Initialize watch service
|
||||
watch_service = WatchService(
|
||||
app_config=app_config,
|
||||
project_repository=project_repository,
|
||||
quiet=quiet,
|
||||
constrained_project=constrained_project,
|
||||
)
|
||||
|
||||
# Get active projects
|
||||
active_projects = await project_repository.get_active_projects()
|
||||
|
||||
# Filter to constrained project if MCP server was started with --project
|
||||
constrained_project = os.environ.get("BASIC_MEMORY_MCP_PROJECT")
|
||||
if constrained_project:
|
||||
active_projects = [p for p in active_projects if p.name == constrained_project]
|
||||
logger.info(f"Background sync constrained to project: {constrained_project}")
|
||||
@@ -154,7 +159,10 @@ async def initialize_file_sync(
|
||||
# Don't await the tasks - let them run in background while we continue
|
||||
|
||||
# Then start the watch service in the background
|
||||
logger.info("Starting watch service for all projects")
|
||||
if constrained_project:
|
||||
logger.info(f"Starting watch service constrained to project: {constrained_project}")
|
||||
else:
|
||||
logger.info("Starting watch service for all projects")
|
||||
|
||||
# run the watch service
|
||||
await watch_service.run()
|
||||
|
||||
@@ -85,6 +85,7 @@ class WatchService:
|
||||
project_repository: ProjectRepository,
|
||||
quiet: bool = False,
|
||||
sync_service_factory: Optional[SyncServiceFactory] = None,
|
||||
constrained_project: Optional[str] = None,
|
||||
):
|
||||
self.app_config = app_config
|
||||
self.project_repository = project_repository
|
||||
@@ -93,6 +94,11 @@ class WatchService:
|
||||
self.status_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
self._ignore_patterns_cache: dict[Path, Set[str]] = {}
|
||||
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`
|
||||
# process spawns a watcher over every project and racing writers collide
|
||||
# on the same files.
|
||||
self.constrained_project = constrained_project
|
||||
|
||||
# quiet mode for mcp so it doesn't mess up stdout
|
||||
self.console = Console(quiet=quiet)
|
||||
@@ -156,6 +162,35 @@ class WatchService:
|
||||
# process changes
|
||||
await asyncio.gather(*change_handlers)
|
||||
|
||||
async def _select_projects_to_watch(self) -> list[Project]:
|
||||
"""Return the set of projects this watch cycle should observe.
|
||||
|
||||
Applies two filters in order:
|
||||
1. ``constrained_project`` — if the MCP server was started with
|
||||
``--project``, only that project is watched. This keeps concurrent
|
||||
MCP processes from producing duplicate watchers that race on the
|
||||
same files.
|
||||
2. Cloud-only projects without a local bisync copy are skipped so we
|
||||
don't watch a path that does not exist on disk.
|
||||
"""
|
||||
projects = await self.project_repository.get_active_projects()
|
||||
|
||||
if self.constrained_project:
|
||||
projects = [p for p in projects if p.name == self.constrained_project]
|
||||
|
||||
cloud_skip: list[str] = []
|
||||
for p in projects:
|
||||
if self.app_config.get_project_mode(p.name) == ProjectMode.CLOUD:
|
||||
entry = self.app_config.projects.get(p.name)
|
||||
if entry and Path(entry.path).is_absolute():
|
||||
continue # Cloud project with local bisync copy — keep watching
|
||||
cloud_skip.append(p.name)
|
||||
if cloud_skip:
|
||||
projects = [p for p in projects if p.name not in cloud_skip]
|
||||
logger.debug(f"Skipping cloud-mode projects in watch cycle: {cloud_skip}")
|
||||
|
||||
return list(projects)
|
||||
|
||||
async def run(self): # pragma: no cover
|
||||
"""Watch for file changes and sync them"""
|
||||
|
||||
@@ -174,23 +209,22 @@ class WatchService:
|
||||
# Clear ignore patterns cache to pick up any .gitignore changes
|
||||
self._ignore_patterns_cache.clear()
|
||||
|
||||
# Reload projects to catch any new/removed projects
|
||||
projects = await self.project_repository.get_active_projects()
|
||||
projects = await self._select_projects_to_watch()
|
||||
|
||||
# Trigger: project is configured for cloud routing
|
||||
# Why: cloud-only projects (no local directory) should not be watched;
|
||||
# cloud projects with a local bisync copy (absolute path) need watching
|
||||
# Outcome: watch cycle skips cloud projects without a local directory
|
||||
cloud_skip = []
|
||||
for p in projects:
|
||||
if self.app_config.get_project_mode(p.name) == ProjectMode.CLOUD:
|
||||
entry = self.app_config.projects.get(p.name)
|
||||
if entry and Path(entry.path).is_absolute():
|
||||
continue # Cloud project with local bisync copy — keep watching
|
||||
cloud_skip.append(p.name)
|
||||
if cloud_skip:
|
||||
projects = [p for p in projects if p.name not in cloud_skip]
|
||||
logger.debug(f"Skipping cloud-mode projects in watch cycle: {cloud_skip}")
|
||||
# Trigger: no projects selected (e.g. constrained_project names a
|
||||
# project not in the DB, or every project was filtered out)
|
||||
# Why: watchfiles.awatch() requires at least one path. Calling it
|
||||
# with an empty list raises ValueError, which the outer handler
|
||||
# catches with a 5s sleep — producing a tight error-log loop.
|
||||
# Outcome: sleep the configured reload interval before retrying, so
|
||||
# newly added projects get picked up on the next cycle.
|
||||
if not projects:
|
||||
logger.warning(
|
||||
"No projects to watch; sleeping before retry "
|
||||
f"(constrained_project={self.constrained_project!r})"
|
||||
)
|
||||
await asyncio.sleep(self.app_config.watch_project_reload_interval)
|
||||
continue
|
||||
|
||||
project_paths = [project.path for project in projects]
|
||||
logger.debug(f"Starting watch cycle for directories: {project_paths}")
|
||||
|
||||
@@ -17,6 +17,7 @@ from basic_memory.services.initialization import (
|
||||
ensure_initialization,
|
||||
initialize_app,
|
||||
initialize_database,
|
||||
initialize_file_sync,
|
||||
reconcile_projects_with_config,
|
||||
)
|
||||
|
||||
@@ -167,6 +168,60 @@ async def test_initialize_app_warns_on_frontmatter_permalink_precedence(
|
||||
)
|
||||
|
||||
|
||||
class _FakeWatchService:
|
||||
"""Captures init kwargs so tests can assert what the real service receives."""
|
||||
|
||||
last_kwargs: dict[str, object] = {}
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
_FakeWatchService.last_kwargs = kwargs
|
||||
|
||||
async def run(self):
|
||||
return None
|
||||
|
||||
|
||||
def _disable_test_env_short_circuit(monkeypatch) -> None:
|
||||
"""Bypass ``is_test_env`` so ``initialize_file_sync`` actually runs.
|
||||
|
||||
``is_test_env`` returns True whenever pytest is running, which would cause
|
||||
``initialize_file_sync`` to return before constructing a WatchService.
|
||||
"""
|
||||
monkeypatch.setattr(BasicMemoryConfig, "is_test_env", property(lambda self: False))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_initialize_file_sync_passes_constrained_project_to_watch_service(
|
||||
app_config: BasicMemoryConfig, monkeypatch
|
||||
):
|
||||
"""``BASIC_MEMORY_MCP_PROJECT`` must reach the watch service, not just the
|
||||
one-shot background sync. Otherwise multiple ``basic-memory mcp --project X``
|
||||
processes each spawn a watcher over every project and race on file writes.
|
||||
"""
|
||||
_disable_test_env_short_circuit(monkeypatch)
|
||||
monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", "target-project")
|
||||
monkeypatch.setattr("basic_memory.sync.WatchService", _FakeWatchService)
|
||||
_FakeWatchService.last_kwargs = {}
|
||||
|
||||
await initialize_file_sync(app_config, quiet=True)
|
||||
|
||||
assert _FakeWatchService.last_kwargs.get("constrained_project") == "target-project"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_initialize_file_sync_no_constraint_when_env_unset(
|
||||
app_config: BasicMemoryConfig, monkeypatch
|
||||
):
|
||||
"""With no env var set, the watch service is unconstrained."""
|
||||
_disable_test_env_short_circuit(monkeypatch)
|
||||
monkeypatch.delenv("BASIC_MEMORY_MCP_PROJECT", raising=False)
|
||||
monkeypatch.setattr("basic_memory.sync.WatchService", _FakeWatchService)
|
||||
_FakeWatchService.last_kwargs = {}
|
||||
|
||||
await initialize_file_sync(app_config, quiet=True)
|
||||
|
||||
assert _FakeWatchService.last_kwargs.get("constrained_project") is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_initialize_app_no_precedence_warning_when_not_conflicting(
|
||||
app_config: BasicMemoryConfig, monkeypatch
|
||||
|
||||
@@ -8,7 +8,7 @@ from unittest.mock import MagicMock
|
||||
import pytest
|
||||
from watchfiles import Change
|
||||
|
||||
from basic_memory.config import BasicMemoryConfig, WATCH_STATUS_JSON
|
||||
from basic_memory.config import BasicMemoryConfig, ProjectMode, WATCH_STATUS_JSON
|
||||
from basic_memory.models.project import Project
|
||||
from basic_memory.sync.watch_service import WatchService, WatchServiceState
|
||||
|
||||
@@ -44,6 +44,106 @@ def test_watch_service_status_path_honors_basic_memory_config_dir(tmp_path, monk
|
||||
assert service.status_path.parent.exists()
|
||||
|
||||
|
||||
async def _register_local_projects(
|
||||
app_config: BasicMemoryConfig, project_repository, specs
|
||||
) -> None:
|
||||
"""Register projects as local in both the DB and app_config.
|
||||
|
||||
Projects that aren't present in ``app_config.projects`` are treated as
|
||||
cloud-only by ``get_project_mode`` and get filtered out of the watch
|
||||
cycle, so tests that exercise ``_select_projects_to_watch`` need them
|
||||
added to both sides.
|
||||
"""
|
||||
from basic_memory.config import ProjectEntry
|
||||
|
||||
for spec in specs:
|
||||
await project_repository.create(
|
||||
{
|
||||
"name": spec["name"],
|
||||
"description": spec["name"],
|
||||
"path": spec["path"],
|
||||
"is_active": True,
|
||||
"is_default": False,
|
||||
}
|
||||
)
|
||||
app_config.projects[spec["name"]] = ProjectEntry(path=spec["path"], mode=ProjectMode.LOCAL)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_select_projects_to_watch_returns_all_when_unconstrained(
|
||||
app_config: BasicMemoryConfig, project_repository
|
||||
):
|
||||
"""Without a --project constraint, every active project is watched."""
|
||||
await _register_local_projects(
|
||||
app_config,
|
||||
project_repository,
|
||||
[
|
||||
{"name": "project-alpha", "path": "/tmp/alpha"},
|
||||
{"name": "project-beta", "path": "/tmp/beta"},
|
||||
],
|
||||
)
|
||||
|
||||
service = WatchService(app_config=app_config, project_repository=project_repository)
|
||||
|
||||
projects = await service._select_projects_to_watch()
|
||||
names = {p.name for p in projects}
|
||||
|
||||
assert "project-alpha" in names
|
||||
assert "project-beta" in names
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_select_projects_to_watch_filters_to_constrained_project(
|
||||
app_config: BasicMemoryConfig, project_repository
|
||||
):
|
||||
"""With ``constrained_project`` set, only that project is returned.
|
||||
|
||||
Regression: multiple ``basic-memory mcp --project X`` processes each spawned
|
||||
a watch service over every project, producing duplicate change handlers
|
||||
that raced on file writes and cascaded deletes.
|
||||
"""
|
||||
await _register_local_projects(
|
||||
app_config,
|
||||
project_repository,
|
||||
[
|
||||
{"name": "project-alpha", "path": "/tmp/alpha"},
|
||||
{"name": "project-beta", "path": "/tmp/beta"},
|
||||
],
|
||||
)
|
||||
|
||||
service = WatchService(
|
||||
app_config=app_config,
|
||||
project_repository=project_repository,
|
||||
constrained_project="project-beta",
|
||||
)
|
||||
|
||||
projects = await service._select_projects_to_watch()
|
||||
|
||||
assert [p.name for p in projects] == ["project-beta"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_select_projects_to_watch_empty_when_constrained_project_missing(
|
||||
app_config: BasicMemoryConfig, project_repository
|
||||
):
|
||||
"""An unknown constraint yields an empty watch set rather than watching everything."""
|
||||
await _register_local_projects(
|
||||
app_config,
|
||||
project_repository,
|
||||
[{"name": "project-alpha", "path": "/tmp/alpha"}],
|
||||
)
|
||||
|
||||
service = WatchService(
|
||||
app_config=app_config,
|
||||
project_repository=project_repository,
|
||||
constrained_project="does-not-exist",
|
||||
)
|
||||
|
||||
projects = await service._select_projects_to_watch()
|
||||
|
||||
assert projects == []
|
||||
|
||||
|
||||
def test_state_add_event():
|
||||
"""Test adding events to state."""
|
||||
state = WatchServiceState()
|
||||
|
||||
@@ -117,7 +117,16 @@ async def test_run_handles_no_projects(monkeypatch):
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_reloads_projects_each_cycle(monkeypatch, tmp_path):
|
||||
config = BasicMemoryConfig(watch_project_reload_interval=1)
|
||||
# Projects must be registered in app_config.projects as local-mode, otherwise
|
||||
# _select_projects_to_watch() treats unknown names as CLOUD and filters them
|
||||
# out, which would short-circuit through the empty-projects guard in run().
|
||||
config = BasicMemoryConfig(
|
||||
watch_project_reload_interval=1,
|
||||
projects={
|
||||
"project1": {"path": str(tmp_path / "p1"), "mode": "local"},
|
||||
"project2": {"path": str(tmp_path / "p2"), "mode": "local"},
|
||||
},
|
||||
)
|
||||
repo = _Repo(
|
||||
projects_side_effect=[
|
||||
[Project(id=1, name="project1", path=str(tmp_path / "p1"), permalink="project1")],
|
||||
@@ -229,7 +238,9 @@ async def test_run_keeps_cloud_projects_with_local_bisync(monkeypatch, tmp_path)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_continues_after_cycle_error(monkeypatch, tmp_path):
|
||||
config = BasicMemoryConfig()
|
||||
config = BasicMemoryConfig(
|
||||
projects={"test": {"path": str(tmp_path / "test"), "mode": "local"}},
|
||||
)
|
||||
repo = _Repo(
|
||||
projects_return=[Project(id=1, name="test", path=str(tmp_path / "test"), permalink="test")]
|
||||
)
|
||||
@@ -264,7 +275,9 @@ async def test_run_continues_after_cycle_error(monkeypatch, tmp_path):
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_timer_task_cancelled_properly(monkeypatch, tmp_path):
|
||||
config = BasicMemoryConfig()
|
||||
config = BasicMemoryConfig(
|
||||
projects={"test": {"path": str(tmp_path / "test"), "mode": "local"}},
|
||||
)
|
||||
repo = _Repo(
|
||||
projects_return=[Project(id=1, name="test", path=str(tmp_path / "test"), permalink="test")]
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user