diff --git a/src/basic_memory/index/watch_service.py b/src/basic_memory/index/watch_service.py index 12af870e8..04bd4af41 100644 --- a/src/basic_memory/index/watch_service.py +++ b/src/basic_memory/index/watch_service.py @@ -20,6 +20,11 @@ from basic_memory import db from basic_memory.config import BasicMemoryConfig, ConfigManager, ProjectMode, WATCH_STATUS_JSON from basic_memory.ignore_utils import load_gitignore_patterns +from basic_memory.index.local_project import ( + LocalProjectIndexRuntimeFactory, + LocalProjectIndexRuntimeProvider, + run_local_project_index_for_project, +) from basic_memory.index.local_runtime import LocalWatchEventIndexRuntimeFactory from basic_memory.index.local_watch import ( LocalWatchEventIndexRequest, @@ -104,6 +109,8 @@ def __init__( quiet: bool = False, event_index_runtime_factory: WatchEventIndexRuntimeFactory | None = None, constrained_project: str | None = None, + project_index_runtime_factory: LocalProjectIndexRuntimeProvider | None = None, + initial_projects: Sequence[Project] = (), ) -> None: self.app_config = app_config self.project_repository = project_repository @@ -116,13 +123,36 @@ def __init__( self._event_index_runtime_factory = ( event_index_runtime_factory or LocalWatchEventIndexRuntimeFactory() ) + self._project_index_runtime_factory = ( + project_index_runtime_factory or LocalProjectIndexRuntimeFactory() + ) + # Startup already schedules these projects' initial scans. Only projects + # added later need a catch-up pass after their filesystem watch is attached. + # Track by project identity, not root: a path change rewrites config before + # the files move, so rescanning the new root would index every old path as + # deleted (`bm project move`). + self._indexed_project_ids = {project.external_id for project in initial_projects} + self._catch_up_in_progress = False self.constrained_project = constrained_project self.console = Console(quiet=quiet) async def _schedule_restart(self, stop_event: asyncio.Event) -> None: """Schedule a watch cycle restart so project config changes are observed.""" - await asyncio.sleep(self.app_config.watch_project_reload_interval) - stop_event.set() + waited_for_catch_up = False + while True: + await asyncio.sleep(self.app_config.watch_project_reload_interval) + # Trigger: a newly watched project is still being indexed or draining + # the batches collected while it was scanned. + # Why: watchfiles discards its pending batches when stop is set, and + # the catch-up scan is the only consumer step long enough to + # collect a meaningful backlog. + # Outcome: give the catch-up one full interval to drain. Then reload + # anyway, so a project with continuous writes cannot defer + # config reloads forever. + if not self._catch_up_in_progress or waited_for_catch_up: + stop_event.set() + return + waited_for_catch_up = True def _get_ignore_patterns(self, project_path: Path) -> set[str]: """Return cached ignore patterns for one project root.""" @@ -137,17 +167,70 @@ async def _watch_projects_cycle( ) -> None: """Run one watchfiles cycle and route batches into project-local indexing.""" project_paths = [project.path for project in projects] + projects_by_id = {project.external_id: project for project in projects} + self._indexed_project_ids.intersection_update(projects_by_id) + new_projects = { + external_id: project + for external_id, project in projects_by_id.items() + if external_id not in self._indexed_project_ids + } + catch_up_needs_drain = bool(new_projects) + self._catch_up_in_progress = catch_up_needs_drain previous_filter_roots = self._sorted_watch_filter_roots self._sorted_watch_filter_roots = local_watch_filter_roots(projects) + watcher = awatch( + *project_paths, + debounce=self.app_config.index_delay, + watch_filter=self.filter_changes, + recursive=True, + stop_event=stop_event, + yield_on_timeout=bool(new_projects), + rust_timeout=( + min(1000, max(1, self.app_config.watch_project_reload_interval * 500)) + if new_projects + else None + ), + ) try: - async for changes in awatch( - *project_paths, - debounce=self.app_config.index_delay, - watch_filter=self.filter_changes, - recursive=True, - stop_event=stop_event, - ): + async for changes in watcher: + # awatch installs its watches before yielding changes or a timeout. + # Scanning earlier leaves the same gap between a scan and attach. + # Stay in this consumer so catch-up cannot overlap a watch batch. + caught_up_now = bool(new_projects) + for external_id, project in new_projects.items(): + if not ConfigManager().config.is_locally_syncable(project.name, project.path): + continue + try: + result = await run_local_project_index_for_project( + project, runtime_factory=self._project_index_runtime_factory + ) + if not stop_event.is_set(): + self._indexed_project_ids.add(external_id) + self.state.last_scan = datetime.now() + logger.info( + "Indexed a newly watched project", + project=project.name, + enqueued_files=result.enqueued_files, + ) + except Exception as exc: + logger.exception(f"New project index failed for {project.name}: {exc}") + self.state.record_error(str(exc)) + await self.write_status() + new_projects.clear() + if not changes: + # Trigger: the timeout that released the catch-up arrived before + # the scan started, so it proves nothing about files queued + # during the scan. + # Why: the reload must not stop the watcher until those batches + # have been handed to indexing. + # Outcome: clear the catch-up guard only after a later quiet batch. + if caught_up_now: + continue + if catch_up_needs_drain: + catch_up_needs_drain = False + self._catch_up_in_progress = False + continue ignore_patterns_by_project_root = { local_project_root(project): self._get_ignore_patterns( local_project_root(project) @@ -165,7 +248,11 @@ async def _watch_projects_cycle( ] await asyncio.gather(*change_handlers) finally: - self._sorted_watch_filter_roots = previous_filter_roots + self._catch_up_in_progress = False + try: + await watcher.aclose() + finally: + self._sorted_watch_filter_roots = previous_filter_roots async def _select_projects_to_watch(self) -> list[Project]: """Return locally syncable projects that this watcher instance owns.""" @@ -178,10 +265,11 @@ async def _select_projects_to_watch(self) -> list[Project]: project for project in projects if project.permalink == constrained_permalink ] + current_config = ConfigManager().config skipped = [ project.name for project in projects - if not self.app_config.is_locally_syncable(project.name, project.path) + if not current_config.is_locally_syncable(project.name, project.path) ] if skipped: projects = [project for project in projects if project.name not in skipped] diff --git a/src/basic_memory/man/man3/bm-project(3).md b/src/basic_memory/man/man3/bm-project(3).md index c953f7fe2..7acc71389 100644 --- a/src/basic_memory/man/man3/bm-project(3).md +++ b/src/basic_memory/man/man3/bm-project(3).md @@ -38,6 +38,10 @@ bm_project(command, name=None, path=None, set_default=False, workspace=None, bm_project never deletes; bm_project_delete does, and only when the server enables it. +While the local server is running, its watcher picks up a new project's folder +at the next project reload. Once the watch is attached, a one-time index pass +includes files copied in before attachment; later changes use normal file events. + The shell counterpart for `recent` is tail(1). ## PARAMETERS diff --git a/src/basic_memory/services/initialization.py b/src/basic_memory/services/initialization.py index cde1c39d6..195686a2e 100644 --- a/src/basic_memory/services/initialization.py +++ b/src/basic_memory/services/initialization.py @@ -270,16 +270,6 @@ async def initialize_file_indexing( read_cache=read_cache, ) - # Initialize watch service - watch_service = WatchService( - app_config=app_config, - project_repository=project_repository, - session_maker=session_maker, - quiet=quiet, - event_index_runtime_factory=event_index_runtime_factory, - constrained_project=constrained_project, - ) - # Get active projects async with db.scoped_session(session_maker) as session: active_projects = await project_repository.get_active_projects(session) @@ -298,6 +288,17 @@ async def initialize_file_indexing( active_projects = [p for p in active_projects if p.name not in skip] logger.info(f"Skipping projects that are not locally indexable: {skip}") + watch_service = WatchService( + app_config=app_config, + project_repository=project_repository, + session_maker=session_maker, + quiet=quiet, + event_index_runtime_factory=event_index_runtime_factory, + constrained_project=constrained_project, + project_index_runtime_factory=project_index_runtime_factory, + initial_projects=active_projects, + ) + # Recover materializations left stuck (writing/pending/failed) and source # cleanup whose in-process enqueue was lost. Runs synchronously so the files # converge before the initial project index scans them. diff --git a/tests/index/test_watch_new_projects.py b/tests/index/test_watch_new_projects.py new file mode 100644 index 000000000..6f4d629de --- /dev/null +++ b/tests/index/test_watch_new_projects.py @@ -0,0 +1,474 @@ +"""Files written before a new project's watch is attached must be indexed (#1817).""" + +from __future__ import annotations + +import asyncio +from collections import Counter +from pathlib import Path + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from basic_memory import db +from basic_memory.config import ConfigManager, ProjectEntry +from basic_memory.index.local_project import LocalProjectIndexRuntimeFactory +from basic_memory.index import watch_service as watch_module +from basic_memory.index.watch_service import WatchService +from basic_memory.models import Project +from basic_memory.repository import ProjectRepository +from basic_memory.repository.entity_repository import EntityRepository + + +async def _register_project( + name: str, + root: Path, + config_manager: ConfigManager, + project_repository: ProjectRepository, + session_maker: async_sessionmaker[AsyncSession], +) -> Project: + root.mkdir(parents=True, exist_ok=True) + async with db.scoped_session(session_maker) as session: + project = await project_repository.create( + session, + {"name": name, "path": str(root), "is_active": True, "is_default": False}, + ) + config = config_manager.config.model_copy(deep=True) + config.projects[name] = ProjectEntry(path=str(root)) + config_manager.save_config(config) + return project + + +@pytest.mark.asyncio +async def test_watch_selection_rereads_added_and_removed_project_config( + app_config, config_manager, project_repository, session_maker, test_project, tmp_path_factory +): + watcher = WatchService(app_config.model_copy(deep=True), project_repository, session_maker) + assert [project.id for project in await watcher._select_projects_to_watch()] == [ + test_project.id + ] + + added = await _register_project( + "added-project", + tmp_path_factory.mktemp("added-watch-root"), + config_manager, + project_repository, + session_maker, + ) + selected = await watcher._select_projects_to_watch() + assert {project.id for project in selected} == {test_project.id, added.id} + + config = config_manager.config.model_copy(deep=True) + del config.projects[test_project.name] + config.default_project = added.name + config_manager.save_config(config) + assert [project.id for project in await watcher._select_projects_to_watch()] == [added.id] + + +@pytest.mark.asyncio +async def test_new_project_backfill_runs_after_attach_and_only_once( + app_config, + config_manager, + project_repository, + session_maker, + test_project, + tmp_path_factory, + monkeypatch, +): + watcher = WatchService(app_config.model_copy(deep=True), project_repository, session_maker) + root = tmp_path_factory.mktemp("new-watch-project") + attached = False + timeout_yields: list[bool] = [] + + async def ready_watch(*paths, stop_event, yield_on_timeout=False, **_kwargs): + nonlocal attached + attached = True + if str(root) in paths: + # A pre-attach scan would miss this file; no file-change event follows. + (root / "during-attach.md").write_text("# During Attach\n", encoding="utf-8") + timeout_yields.append(yield_on_timeout) + yield set() + stop_event.set() + + monkeypatch.setattr(watch_module, "awatch", ready_watch) + await watcher._watch_projects_cycle([test_project], asyncio.Event()) + added = await _register_project( + "new-project", root, config_manager, project_repository, session_maker + ) + (root / "before-attach.md").write_text("# Before Attach\n- [goal] find me\n", encoding="utf-8") + (root / ".gitignore").write_text("ignored/\n", encoding="utf-8") + (root / "ignored").mkdir() + (root / "ignored" / "skip.md").write_text("# Ignore Me\n", encoding="utf-8") + (root / ".hidden.md").write_text("# Hidden\n", encoding="utf-8") + indexed: list[int] = [] + original_runtime = LocalProjectIndexRuntimeFactory.runtime_for_project + + async def record_runtime(factory, project): + assert attached, "the project must be watched before its one-off scan" + indexed.append(project.id) + return await original_runtime(factory, project) + + monkeypatch.setattr(LocalProjectIndexRuntimeFactory, "runtime_for_project", record_runtime) + attached = False + await watcher._watch_projects_cycle([test_project, added], asyncio.Event()) + + repository = EntityRepository(added.id) + async with db.scoped_session(session_maker) as session: + assert await repository.get_by_file_path(session, "before-attach.md") is not None + assert await repository.get_by_file_path(session, "during-attach.md") is not None + assert await repository.get_by_file_path(session, "ignored/skip.md") is None + assert await repository.get_by_file_path(session, ".hidden.md") is None + await watcher._watch_projects_cycle([test_project, added], asyncio.Event()) + assert indexed == [added.id] + assert timeout_yields[0] is True + + +@pytest.mark.asyncio +async def test_startup_projects_do_not_get_a_second_scan( + app_config, config_manager, project_repository, session_maker, test_project, monkeypatch +): + watcher = WatchService( + app_config, project_repository, session_maker, initial_projects=[test_project] + ) + + async def ready_watch(*_paths, stop_event, **_kwargs): + yield set() + stop_event.set() + + async def unexpected_scan(*_args, **_kwargs): + raise AssertionError("startup already owns the initial project scan") + + monkeypatch.setattr(watch_module, "awatch", ready_watch) + monkeypatch.setattr(watch_module, "run_local_project_index_for_project", unexpected_scan) + await watcher._watch_projects_cycle([test_project], asyncio.Event()) + assert watcher.state.error_count == 0 + + +@pytest.mark.asyncio +async def test_failed_backfill_is_isolated_and_retried_on_the_next_cycle( + app_config, + config_manager, + project_repository, + session_maker, + test_project, + tmp_path_factory, + monkeypatch, +): + watcher = WatchService( + app_config, project_repository, session_maker, initial_projects=[test_project] + ) + broken = await _register_project( + "broken", + tmp_path_factory.mktemp("broken-root"), + config_manager, + project_repository, + session_maker, + ) + healthy = await _register_project( + "healthy", + tmp_path_factory.mktemp("healthy-root"), + config_manager, + project_repository, + session_maker, + ) + for project in (broken, healthy): + (Path(project.path) / "note.md").write_text("# Note\n", encoding="utf-8") + attempts: list[int] = [] + fail = True + original_index = watch_module.run_local_project_index_for_project + + async def index(project, **kwargs): + attempts.append(project.id) + if fail and project.id == broken.id: + raise RuntimeError("scan failed") + return await original_index(project, **kwargs) + + async def ready_watch(*_paths, stop_event, **_kwargs): + yield set() + stop_event.set() + + monkeypatch.setattr(watch_module, "awatch", ready_watch) + monkeypatch.setattr(watch_module, "run_local_project_index_for_project", index) + projects = [test_project, broken, healthy] + await watcher._watch_projects_cycle(projects, asyncio.Event()) + assert watcher.state.error_count == 1 + async with db.scoped_session(session_maker) as session: + assert await EntityRepository(healthy.id).get_by_file_path(session, "note.md") is not None + fail = False + await watcher._watch_projects_cycle(projects, asyncio.Event()) + assert Counter(attempts) == {broken.id: 2, healthy.id: 1} + async with db.scoped_session(session_maker) as session: + assert await EntityRepository(broken.id).get_by_file_path(session, "note.md") is not None + + +@pytest.mark.asyncio +async def test_project_removed_during_attach_is_not_backfilled( + app_config, + config_manager, + project_repository, + session_maker, + test_project, + tmp_path_factory, + monkeypatch, +): + added = await _register_project( + "removed", + tmp_path_factory.mktemp("removed-root"), + config_manager, + project_repository, + session_maker, + ) + watcher = WatchService( + app_config, project_repository, session_maker, initial_projects=[test_project] + ) + (Path(added.path) / "note.md").write_text("# Must Not Index\n", encoding="utf-8") + + async def ready_watch(*_paths, stop_event, **_kwargs): + config = config_manager.config.model_copy(deep=True) + del config.projects[added.name] + config_manager.save_config(config) + yield set() + stop_event.set() + + monkeypatch.setattr(watch_module, "awatch", ready_watch) + await watcher._watch_projects_cycle([test_project, added], asyncio.Event()) + async with db.scoped_session(session_maker) as session: + assert await EntityRepository(added.id).get_by_file_path(session, "note.md") is None + assert watcher.state.error_count == 0 + + +@pytest.mark.asyncio +async def test_a_restart_during_backfill_keeps_the_project_pending( + app_config, + config_manager, + project_repository, + session_maker, + test_project, + tmp_path_factory, + monkeypatch, +): + added = await _register_project( + "late", + tmp_path_factory.mktemp("late-root"), + config_manager, + project_repository, + session_maker, + ) + watcher = WatchService( + app_config, project_repository, session_maker, initial_projects=[test_project] + ) + stop_event = asyncio.Event() + original_index = watch_module.run_local_project_index_for_project + attempts = 0 + + async def index(project, **kwargs): + nonlocal attempts + attempts += 1 + result = await original_index(project, **kwargs) + if attempts == 1: + (Path(project.path) / "late.md").write_text("# Late Arrival\n", encoding="utf-8") + stop_event.set() + return result + + async def ready_watch(*_paths, **_kwargs): + yield set() + + monkeypatch.setattr(watch_module, "awatch", ready_watch) + monkeypatch.setattr(watch_module, "run_local_project_index_for_project", index) + await watcher._watch_projects_cycle([test_project, added], stop_event) + await watcher._watch_projects_cycle([test_project, added], asyncio.Event()) + async with db.scoped_session(session_maker) as session: + assert await EntityRepository(added.id).get_by_file_path(session, "late.md") is not None + assert attempts == 2 + + +@pytest.mark.asyncio +async def test_real_quiet_watcher_backfills_then_handles_live_writes( + app_config, + config_manager, + project_repository, + session_maker, + test_project, + tmp_path_factory, + monkeypatch, +): + root = tmp_path_factory.mktemp("native-watch-root") + added = await _register_project( + "native", root, config_manager, project_repository, session_maker + ) + (root / "before.md").write_text("# Before\n", encoding="utf-8") + watcher = WatchService( + app_config, project_repository, session_maker, initial_projects=[test_project] + ) + scan_done = asyncio.Event() + live_done = asyncio.Event() + stop_event = asyncio.Event() + original_index = watch_module.run_local_project_index_for_project + original_handle = watcher.handle_changes + + async def index(project, **kwargs): + result = await original_index(project, **kwargs) + scan_done.set() + return result + + async def handle(project, changes): + await original_handle(project, changes) + if any(Path(path).name == "live.md" for _change, path in changes): + live_done.set() + + monkeypatch.setattr(watch_module, "run_local_project_index_for_project", index) + monkeypatch.setattr(watcher, "handle_changes", handle) + task = asyncio.create_task(watcher._watch_projects_cycle([added], stop_event)) + try: + await asyncio.wait_for(scan_done.wait(), timeout=15) + async with db.scoped_session(session_maker) as session: + assert ( + await EntityRepository(added.id).get_by_file_path(session, "before.md") is not None + ) + (root / "live.md").write_text("# Live\n", encoding="utf-8") + await asyncio.wait_for(live_done.wait(), timeout=15) + async with db.scoped_session(session_maker) as session: + assert await EntityRepository(added.id).get_by_file_path(session, "live.md") is not None + finally: + stop_event.set() + await asyncio.wait_for(task, timeout=5) + + +@pytest.mark.asyncio +async def test_cancelling_backfill_closes_the_watcher_and_allows_retry( + app_config, + config_manager, + project_repository, + session_maker, + test_project, + tmp_path_factory, + monkeypatch, +): + added = await _register_project( + "cancelled", + tmp_path_factory.mktemp("cancelled-root"), + config_manager, + project_repository, + session_maker, + ) + watcher = WatchService( + app_config, project_repository, session_maker, initial_projects=[test_project] + ) + scan_started = asyncio.Event() + closed = asyncio.Event() + + async def blocked_index(*_args, **_kwargs): + scan_started.set() + await asyncio.Event().wait() + + async def ready_watch(*_paths, **_kwargs): + try: + yield set() + finally: + closed.set() + + monkeypatch.setattr(watch_module, "awatch", ready_watch) + with monkeypatch.context() as blocked: + blocked.setattr(watch_module, "run_local_project_index_for_project", blocked_index) + task = asyncio.create_task(watcher._watch_projects_cycle([added], asyncio.Event())) + try: + await asyncio.wait_for(scan_started.wait(), timeout=5) + finally: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert closed.is_set() + assert watcher._sorted_watch_filter_roots is None + assert watcher.state.error_count == 0 + + (Path(added.path) / "retry.md").write_text("# Retry\n", encoding="utf-8") + await watcher._watch_projects_cycle([added], asyncio.Event()) + async with db.scoped_session(session_maker) as session: + assert await EntityRepository(added.id).get_by_file_path(session, "retry.md") is not None + + +@pytest.mark.asyncio +async def test_changing_a_project_root_does_not_run_a_full_scan( + app_config, + config_manager, + project_repository, + session_maker, + test_project, + tmp_path_factory, + monkeypatch, +): + """`bm project move` rewrites the path before the user moves the files. + + Rescanning the new, still-empty root would plan every old path as deleted and + remove the project's index. Only a project identity the watcher has never seen + needs the catch-up pass. + """ + watcher = WatchService( + app_config, project_repository, session_maker, initial_projects=[test_project] + ) + root = tmp_path_factory.mktemp("changed-project-root") + (root / "relocated.md").write_text("# Relocated\n", encoding="utf-8") + async with db.scoped_session(session_maker) as session: + moved = await project_repository.update(session, test_project.id, {"path": str(root)}) + assert moved is not None + config = config_manager.config.model_copy(deep=True) + config.projects[test_project.name] = ProjectEntry(path=str(root)) + config_manager.save_config(config) + + async def unexpected_scan(*_args, **_kwargs): + raise AssertionError("a path change must not rescan the new root") + + async def ready_watch(*_paths, **_kwargs): + yield set() + + monkeypatch.setattr(watch_module, "run_local_project_index_for_project", unexpected_scan) + monkeypatch.setattr(watch_module, "awatch", ready_watch) + await watcher._watch_projects_cycle([moved], asyncio.Event()) + assert watcher.state.error_count == 0 + + +@pytest.mark.asyncio +async def test_reload_waits_for_catch_up_and_buffered_changes( + app_config, project_repository, session_maker +): + """A reload must not stop the watcher while the catch-up scan blocks the consumer. + + watchfiles clears its pending batches on stop, so the timer waits for the + catch-up drain instead of closing the cycle underneath it. + """ + config = app_config.model_copy(update={"watch_project_reload_interval": 1}) + watcher = WatchService(config, project_repository, session_maker) + stop_event = asyncio.Event() + watcher._catch_up_in_progress = True + + timer = asyncio.create_task(watcher._schedule_restart(stop_event)) + try: + with pytest.raises(TimeoutError): + await asyncio.wait_for(stop_event.wait(), timeout=1.2) + watcher._catch_up_in_progress = False + await asyncio.wait_for(stop_event.wait(), timeout=2) + finally: + timer.cancel() + try: + await timer + except asyncio.CancelledError: + pass + + +@pytest.mark.asyncio +async def test_current_config_selection_preserves_project_constraint( + app_config, config_manager, project_repository, session_maker, test_project, tmp_path_factory +): + watcher = WatchService( + app_config.model_copy(deep=True), + project_repository, + session_maker, + constrained_project="Added Project", + ) + added = await _register_project( + "added-project", + tmp_path_factory.mktemp("constrained-root"), + config_manager, + project_repository, + session_maker, + ) + assert [project.id for project in await watcher._select_projects_to_watch()] == [added.id] diff --git a/tests/mcp/test_new_project_watching.py b/tests/mcp/test_new_project_watching.py new file mode 100644 index 000000000..6a86eb67a --- /dev/null +++ b/tests/mcp/test_new_project_watching.py @@ -0,0 +1,48 @@ +"""A project created through MCP accepts files copied before watcher reload (#1817).""" + +import asyncio + +import pytest + +from basic_memory.index import watch_service as watch_module +from basic_memory.index.watch_service import WatchService +from basic_memory.mcp.tools import bm_project, bm_search + + +@pytest.mark.asyncio +async def test_files_copied_after_project_creation_are_searchable_after_watch_attach( + client, + app_config, + project_repository, + session_maker, + test_project, + tmp_path_factory, + monkeypatch, +): + watcher = WatchService( + app_config.model_copy(deep=True), + project_repository, + session_maker, + initial_projects=[test_project], + ) + root = tmp_path_factory.mktemp("mcp-created-project") + await bm_project(command="create", name="copied-project", path=str(root)) + (root / "inbox").mkdir() + (root / "inbox" / "copied.md").write_text( + "# Copied Note\n\n- [fact] unicorn\n", encoding="utf-8" + ) + + async def quiet_watch(*_paths, stop_event, **_kwargs): + yield set() + stop_event.set() + + monkeypatch.setattr(watch_module, "awatch", quiet_watch) + projects = await watcher._select_projects_to_watch() + assert "copied-project" in {project.name for project in projects} + await watcher._watch_projects_cycle(projects, asyncio.Event()) + + results = await bm_search( + query="unicorn", mode="text", project="copied-project", output_format="json" + ) + assert isinstance(results, dict) + assert "inbox/copied.md" in {row["file_path"] for row in results["results"]} diff --git a/tests/services/test_initialization.py b/tests/services/test_initialization.py index 89f2f1d23..c4d0532cc 100644 --- a/tests/services/test_initialization.py +++ b/tests/services/test_initialization.py @@ -14,8 +14,10 @@ from basic_memory.config import BasicMemoryConfig, DatabaseBackend from basic_memory.indexing.project_index_coordinator import ProjectIndexCoordinatorResult from basic_memory.index.local_runtime import LocalWatchEventIndexRuntimeFactory +from basic_memory.models import Project from basic_memory.repository.project_repository import ProjectRepository from basic_memory.services.initialization import ( + _initial_index_tasks, ensure_initialization, initialize_database, initialize_file_indexing, @@ -315,6 +317,15 @@ async def run_project_index_for_project( # noqa: ANN001 await initialize_file_indexing(updated, quiet=True) + watch_kwargs = _FakeWatchService.last_kwargs + assert isinstance( + watch_kwargs["project_index_runtime_factory"], RecordingProjectIndexRuntimeFactory + ) + initial_projects = watch_kwargs["initial_projects"] + assert isinstance(initial_projects, list) + assert len(initial_projects) == 1 + assert isinstance(initial_projects[0], Project) + assert initial_projects[0].name == "event-startup" assert len(created_coroutines) == 1 await created_coroutines[0] @@ -333,6 +344,7 @@ async def test_initialize_file_indexing_skips_project_with_non_absolute_path( recognized as cloud, yet Path("") resolves to the process cwd. Indexing it would inject frontmatter into unrelated files, so it must be skipped. """ + initial_index_tasks = _initial_index_tasks.copy() await db.shutdown_db() try: from basic_memory.config import ProjectEntry @@ -371,6 +383,9 @@ async def test_initialize_file_indexing_skips_project_with_non_absolute_path( assert "empty-path" in skip_logs[0] assert "good" not in skip_logs[0] finally: + # The fake watcher returns immediately; let this test's startup index + # finish before closing its database and event loop. + await asyncio.gather(*(_initial_index_tasks - initial_index_tasks)) await db.shutdown_db()