feat: add basic-memory watch CLI command (#559)

Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

src/basic_memory/cli/app.py

@@ -50,7 +50,17 @@ def app_callback(
     # Skip for 'mcp' command - it has its own lifespan that handles initialization
     # Skip for API-using commands (status, sync, etc.) - they handle initialization via deps.py
     # Skip for 'reset' command - it manages its own database lifecycle
-    skip_init_commands = {"doctor", "mcp", "status", "sync", "project", "tool", "reset", "reindex"}
+    skip_init_commands = {
+        "doctor",
+        "mcp",
+        "status",
+        "sync",
+        "project",
+        "tool",
+        "reset",
+        "reindex",
+        "watch",
+    }
     if (
         not version
         and ctx.invoked_subcommand is not None

src/basic_memory/cli/commands/__init__.py

@@ -1,7 +1,7 @@
 """CLI commands for basic-memory."""
 
 from . import status, db, doctor, import_memory_json, mcp, import_claude_conversations
-from . import import_claude_projects, import_chatgpt, tool, project, format, schema
+from . import import_claude_projects, import_chatgpt, tool, project, format, schema, watch
 
 __all__ = [
     "status",
@@ -16,4 +16,5 @@ __all__ = [
     "project",
     "format",
     "schema",
+    "watch",
 ]

src/basic_memory/cli/commands/db.py

@@ -111,9 +111,7 @@ def reindex(
     embeddings: bool = typer.Option(
         False, "--embeddings", "-e", help="Rebuild vector embeddings (requires semantic search)"
     ),
-    search: bool = typer.Option(
-        False, "--search", "-s", help="Rebuild full-text search index"
-    ),
+    search: bool = typer.Option(False, "--search", "-s", help="Rebuild full-text search index"),
     project: str = typer.Option(
         None, "--project", "-p", help="Reindex a specific project (default: all)"
     ),
@@ -193,12 +191,8 @@ async def _reindex(app_config, search: bool, embeddings: bool, project: str | No
                 project_path = Path(proj.path)
                 entity_parser = EntityParser(project_path)
                 markdown_processor = MarkdownProcessor(entity_parser, app_config=app_config)
-                file_service = FileService(
-                    project_path, markdown_processor, app_config=app_config
-                )
-                search_service = SearchService(
-                    search_repository, entity_repository, file_service
-                )
+                file_service = FileService(project_path, markdown_processor, app_config=app_config)
+                search_service = SearchService(search_repository, entity_repository, file_service)
 
                 with Progress(
                     SpinnerColumn(),
@@ -212,9 +206,7 @@ async def _reindex(app_config, search: bool, embeddings: bool, project: str | No
                     def on_progress(entity_id, index, total):
                         progress.update(task, total=total, completed=index)
 
-                    stats = await search_service.reindex_vectors(
-                        progress_callback=on_progress
-                    )
+                    stats = await search_service.reindex_vectors(progress_callback=on_progress)
                     progress.update(task, completed=stats["total_entities"])
 
                 console.print(

src/basic_memory/cli/commands/watch.py

@@ -0,0 +1,97 @@
+"""Watch command - run file watcher as a standalone long-running process."""
+
+import asyncio
+import os
+import signal
+import sys
+from typing import Optional
+
+import typer
+from loguru import logger
+
+from basic_memory import db
+from basic_memory.cli.app import app
+from basic_memory.cli.container import get_container
+from basic_memory.config import ConfigManager
+from basic_memory.services.initialization import initialize_app
+from basic_memory.sync.coordinator import SyncCoordinator
+
+
+async def run_watch(project: Optional[str] = None) -> None:
+    """Run the file watcher as a long-running process.
+
+    This is the async core of the watch command. It:
+    1. Initializes the app (DB migrations + project reconciliation)
+    2. Validates and sets project constraint if --project given
+    3. Creates a SyncCoordinator with quiet=False for Rich console output
+    4. Blocks until SIGINT/SIGTERM, then shuts down cleanly
+    """
+    container = get_container()
+    config = container.config
+
+    # --- Initialization ---
+    # Wrapped in try/finally so DB resources are cleaned up on all exit paths,
+    # including early exits from invalid --project names.
+    await initialize_app(config)
+    sync_coordinator = None
+
+    try:
+        # --- Project constraint ---
+        if project:
+            config_manager = ConfigManager()
+            project_name, _ = config_manager.get_project(project)
+            if not project_name:
+                typer.echo(f"No project found named: {project}", err=True)
+                raise typer.Exit(1)
+
+            os.environ["BASIC_MEMORY_MCP_PROJECT"] = project_name
+            logger.info(f"Watch constrained to project: {project_name}")
+
+        # --- Sync coordinator ---
+        # quiet=False so file change events are printed to the terminal
+        sync_coordinator = SyncCoordinator(config=config, should_sync=True, quiet=False)
+
+        # --- Signal handling ---
+        shutdown_event = asyncio.Event()
+
+        def _signal_handler() -> None:
+            logger.info("Shutdown signal received")
+            shutdown_event.set()
+
+        loop = asyncio.get_running_loop()
+
+        # Windows ProactorEventLoop does not support add_signal_handler;
+        # fall back to the stdlib signal module which works cross-platform.
+        try:
+            for sig in (signal.SIGINT, signal.SIGTERM):
+                loop.add_signal_handler(sig, _signal_handler)
+        except NotImplementedError:
+            for sig in (signal.SIGINT, signal.SIGTERM):
+                signal.signal(sig, lambda _signum, _frame: _signal_handler())
+
+        # --- Run ---
+        await sync_coordinator.start()
+        logger.info("Watch service running, press Ctrl+C to stop")
+        await shutdown_event.wait()
+    finally:
+        if sync_coordinator is not None:
+            await sync_coordinator.stop()
+        await db.shutdown_db()
+        logger.info("Watch service stopped")
+
+
+@app.command()
+def watch(
+    project: Optional[str] = typer.Option(None, help="Restrict watcher to a single project"),
+) -> None:
+    """Run file watcher as a long-running process (no MCP server).
+
+    Watches for file changes in project directories and syncs them to the
+    database. Useful for running Basic Memory sync alongside external tools
+    that don't use the MCP server.
+    """
+    # On Windows, use SelectorEventLoop to avoid ProactorEventLoop cleanup issues
+    if sys.platform == "win32":  # pragma: no cover
+        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
+    asyncio.run(run_watch(project=project))

src/basic_memory/repository/openai_provider.py

@@ -44,8 +44,7 @@ class OpenAIEmbeddingProvider(EmbeddingProvider):
                 from openai import AsyncOpenAI
             except ImportError as exc:  # pragma: no cover - covered via monkeypatch tests
                 raise SemanticDependenciesMissingError(
-                    "OpenAI dependency is missing. "
-                    "Reinstall basic-memory: pip install basic-memory"
+                    "OpenAI dependency is missing. Reinstall basic-memory: pip install basic-memory"
                 ) from exc
 
             api_key = self._api_key or os.getenv("OPENAI_API_KEY")

src/basic_memory/repository/postgres_search_repository.py

@@ -311,9 +311,7 @@ class PostgresSearchRepository(SearchRepositoryBase):
                         f"provider expects {self._vector_dimensions}. "
                         "Dropping and recreating search_vector_embeddings."
                     )
-                    await session.execute(
-                        text("DROP TABLE IF EXISTS search_vector_embeddings")
-                    )
+                    await session.execute(text("DROP TABLE IF EXISTS search_vector_embeddings"))
 
                 await session.execute(
                     text(

src/basic_memory/repository/search_repository_base.py

@@ -904,13 +904,9 @@ class SearchRepositoryBase(ABC):
             )
             # Use (id, type) tuples to avoid collisions between different
             # search_index row types that share the same auto-increment id.
-            allowed_keys = {
-                (row.id, row.type) for row in filtered_rows if row.id is not None
-            }
+            allowed_keys = {(row.id, row.type) for row in filtered_rows if row.id is not None}
             search_index_rows = {
-                k: v
-                for k, v in search_index_rows.items()
-                if (v.id, v.type) in allowed_keys
+                k: v for k, v in search_index_rows.items() if (v.id, v.type) in allowed_keys
             }
 
         ranked_rows: list[SearchIndexRow] = []
@@ -1077,17 +1073,13 @@ class SearchRepositoryBase(ABC):
         for rank, row in enumerate(fts_results, start=1):
             if row.id is None:
                 continue
-            fused_scores[row.id] = fused_scores.get(row.id, 0.0) + (
-                1.0 / (RRF_K + rank)
-            )
+            fused_scores[row.id] = fused_scores.get(row.id, 0.0) + (1.0 / (RRF_K + rank))
             rows_by_id[row.id] = row
 
         for rank, row in enumerate(vector_results, start=1):
             if row.id is None:
                 continue
-            fused_scores[row.id] = fused_scores.get(row.id, 0.0) + (
-                1.0 / (RRF_K + rank)
-            )
+            fused_scores[row.id] = fused_scores.get(row.id, 0.0) + (1.0 / (RRF_K + rank))
             rows_by_id[row.id] = row
 
         ranked = sorted(fused_scores.items(), key=lambda item: item[1], reverse=True)

src/basic_memory/repository/sqlite_search_repository.py

@@ -339,8 +339,7 @@ class SQLiteSearchRepository(SearchRepositoryBase):
             import sqlite_vec
         except ImportError as exc:
             raise SemanticDependenciesMissingError(
-                "sqlite-vec package is missing. "
-                "Reinstall basic-memory: pip install basic-memory"
+                "sqlite-vec package is missing. Reinstall basic-memory: pip install basic-memory"
             ) from exc
 
         async with self._sqlite_vec_lock:

src/basic_memory/services/initialization.py

@@ -71,11 +71,13 @@ async def reconcile_projects_with_config(app_config: BasicMemoryConfig):
 
 async def initialize_file_sync(
     app_config: BasicMemoryConfig,
+    quiet: bool = True,
 ) -> None:
     """Initialize file synchronization services. This function starts the watch service and does not return
 
     Args:
         app_config: The Basic Memory project configuration
+        quiet: Whether to suppress Rich console output (True for MCP, False for CLI watch)
 
     Returns:
         The watch service task that's monitoring file changes
@@ -101,7 +103,7 @@ async def initialize_file_sync(
     watch_service = WatchService(
         app_config=app_config,
         project_repository=project_repository,
-        quiet=True,
+        quiet=quiet,
     )
 
     # Get active projects

src/basic_memory/sync/coordinator.py

@@ -55,6 +55,7 @@ class SyncCoordinator:
     config: BasicMemoryConfig
     should_sync: bool = True
     skip_reason: Optional[str] = None
+    quiet: bool = True
 
     # Internal state (not constructor args)
     _status: SyncStatus = field(default=SyncStatus.NOT_STARTED, init=False)
@@ -96,7 +97,7 @@ class SyncCoordinator:
             async def _file_sync_runner() -> None:  # pragma: no cover
                 """Run the file sync service."""
                 try:
-                    await initialize_file_sync(self.config)
+                    await initialize_file_sync(self.config, quiet=self.quiet)
                 except asyncio.CancelledError:
                     logger.debug("File sync cancelled")
                     raise

tests/cli/test_watch.py

@@ -0,0 +1,278 @@
+"""Tests for the watch CLI command."""
+
+import asyncio
+import os
+import signal
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+import typer
+
+from basic_memory.cli.commands.watch import run_watch
+from basic_memory.config import BasicMemoryConfig
+
+
+@pytest.fixture
+def mock_config():
+    """Create a mock config for testing."""
+    return BasicMemoryConfig()
+
+
+@pytest.fixture
+def mock_container(mock_config):
+    """Create a mock CLI container."""
+    container = MagicMock()
+    container.config = mock_config
+    return container
+
+
+class TestRunWatch:
+    """Tests for run_watch async function."""
+
+    @pytest.mark.asyncio
+    async def test_initializes_app(self, mock_container):
+        """run_watch calls initialize_app with the container's config."""
+        mock_init = AsyncMock()
+
+        with (
+            patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
+            patch("basic_memory.cli.commands.watch.initialize_app", mock_init),
+            patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
+            patch("basic_memory.cli.commands.watch.db") as mock_db,
+        ):
+            # Make coordinator.start() set the shutdown event so we don't block
+            mock_coordinator = AsyncMock()
+            mock_coordinator_cls.return_value = mock_coordinator
+
+            async def start_then_shutdown():
+                # Simulate immediate shutdown after start
+                pass
+
+            mock_coordinator.start = start_then_shutdown
+            mock_coordinator.stop = AsyncMock()
+            mock_db.shutdown_db = AsyncMock()
+
+            # Patch signal handlers and make shutdown_event fire immediately
+            with patch("asyncio.get_running_loop") as mock_loop:
+                mock_loop_instance = MagicMock()
+                mock_loop.return_value = mock_loop_instance
+
+                # Capture the signal handler so we can trigger it
+                signal_handlers = {}
+
+                def capture_handler(sig, handler):
+                    signal_handlers[sig] = handler
+
+                mock_loop_instance.add_signal_handler.side_effect = capture_handler
+
+                # Run in a task so we can trigger shutdown
+                async def run_and_shutdown():
+                    task = asyncio.create_task(run_watch())
+                    # Give it a moment to start
+                    await asyncio.sleep(0.01)
+                    # Trigger shutdown via captured signal handler
+                    import signal
+
+                    if signal.SIGINT in signal_handlers:
+                        signal_handlers[signal.SIGINT]()
+                    await task
+
+                await run_and_shutdown()
+
+            mock_init.assert_called_once_with(mock_container.config)
+
+    @pytest.mark.asyncio
+    async def test_creates_coordinator_with_quiet_false(self, mock_container):
+        """SyncCoordinator is created with should_sync=True and quiet=False."""
+        with (
+            patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
+            patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
+            patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
+            patch("basic_memory.cli.commands.watch.db") as mock_db,
+        ):
+            mock_coordinator = AsyncMock()
+            mock_coordinator_cls.return_value = mock_coordinator
+            mock_db.shutdown_db = AsyncMock()
+
+            with patch("asyncio.get_running_loop") as mock_loop:
+                mock_loop_instance = MagicMock()
+                mock_loop.return_value = mock_loop_instance
+
+                signal_handlers = {}
+
+                def capture_handler(sig, handler):
+                    signal_handlers[sig] = handler
+
+                mock_loop_instance.add_signal_handler.side_effect = capture_handler
+
+                async def run_and_shutdown():
+                    task = asyncio.create_task(run_watch())
+                    await asyncio.sleep(0.01)
+                    import signal
+
+                    if signal.SIGINT in signal_handlers:
+                        signal_handlers[signal.SIGINT]()
+                    await task
+
+                await run_and_shutdown()
+
+            mock_coordinator_cls.assert_called_once_with(
+                config=mock_container.config,
+                should_sync=True,
+                quiet=False,
+            )
+
+    @pytest.mark.asyncio
+    async def test_project_sets_env_var(self, mock_container):
+        """--project validates and sets BASIC_MEMORY_MCP_PROJECT env var."""
+        mock_config_manager = MagicMock()
+        mock_config_manager.get_project.return_value = ("my-project", "/some/path")
+
+        with (
+            patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
+            patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
+            patch(
+                "basic_memory.cli.commands.watch.ConfigManager",
+                return_value=mock_config_manager,
+            ),
+            patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
+            patch("basic_memory.cli.commands.watch.db") as mock_db,
+            patch.dict(os.environ, {}, clear=False),
+        ):
+            mock_coordinator = AsyncMock()
+            mock_coordinator_cls.return_value = mock_coordinator
+            mock_db.shutdown_db = AsyncMock()
+
+            with patch("asyncio.get_running_loop") as mock_loop:
+                mock_loop_instance = MagicMock()
+                mock_loop.return_value = mock_loop_instance
+
+                signal_handlers = {}
+
+                def capture_handler(sig, handler):
+                    signal_handlers[sig] = handler
+
+                mock_loop_instance.add_signal_handler.side_effect = capture_handler
+
+                async def run_and_shutdown():
+                    task = asyncio.create_task(run_watch(project="my-project"))
+                    await asyncio.sleep(0.01)
+                    import signal
+
+                    if signal.SIGINT in signal_handlers:
+                        signal_handlers[signal.SIGINT]()
+                    await task
+
+                await run_and_shutdown()
+
+            assert os.environ.get("BASIC_MEMORY_MCP_PROJECT") == "my-project"
+
+        # Clean up env var
+        os.environ.pop("BASIC_MEMORY_MCP_PROJECT", None)
+
+    @pytest.mark.asyncio
+    async def test_invalid_project_exits_with_error(self, mock_container):
+        """--project with unknown name exits with error and still cleans up DB."""
+        mock_config_manager = MagicMock()
+        mock_config_manager.get_project.return_value = (None, None)
+
+        with (
+            patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
+            patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
+            patch(
+                "basic_memory.cli.commands.watch.ConfigManager",
+                return_value=mock_config_manager,
+            ),
+            patch("basic_memory.cli.commands.watch.db") as mock_db,
+        ):
+            mock_db.shutdown_db = AsyncMock()
+
+            with pytest.raises(typer.Exit) as exc_info:
+                await run_watch(project="nonexistent")
+
+            assert exc_info.value.exit_code == 1
+            # DB should still be cleaned up even on early exit
+            mock_db.shutdown_db.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_shutdown_stops_coordinator_and_db(self, mock_container):
+        """On shutdown, coordinator.stop() and db.shutdown_db() are called."""
+        with (
+            patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
+            patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
+            patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
+            patch("basic_memory.cli.commands.watch.db") as mock_db,
+        ):
+            mock_coordinator = AsyncMock()
+            mock_coordinator_cls.return_value = mock_coordinator
+            mock_db.shutdown_db = AsyncMock()
+
+            with patch("asyncio.get_running_loop") as mock_loop:
+                mock_loop_instance = MagicMock()
+                mock_loop.return_value = mock_loop_instance
+
+                signal_handlers = {}
+
+                def capture_handler(sig, handler):
+                    signal_handlers[sig] = handler
+
+                mock_loop_instance.add_signal_handler.side_effect = capture_handler
+
+                async def run_and_shutdown():
+                    task = asyncio.create_task(run_watch())
+                    await asyncio.sleep(0.01)
+                    import signal
+
+                    if signal.SIGINT in signal_handlers:
+                        signal_handlers[signal.SIGINT]()
+                    await task
+
+                await run_and_shutdown()
+
+            mock_coordinator.stop.assert_called_once()
+            mock_db.shutdown_db.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_falls_back_to_signal_module_on_windows(self, mock_container):
+        """When add_signal_handler raises NotImplementedError, falls back to signal.signal()."""
+        with (
+            patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
+            patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
+            patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
+            patch("basic_memory.cli.commands.watch.db") as mock_db,
+        ):
+            mock_coordinator = AsyncMock()
+            mock_coordinator_cls.return_value = mock_coordinator
+            mock_db.shutdown_db = AsyncMock()
+
+            with patch("asyncio.get_running_loop") as mock_loop:
+                mock_loop_instance = MagicMock()
+                mock_loop.return_value = mock_loop_instance
+
+                # Simulate Windows: add_signal_handler raises NotImplementedError
+                mock_loop_instance.add_signal_handler.side_effect = NotImplementedError
+
+                with patch("basic_memory.cli.commands.watch.signal.signal") as mock_signal:
+                    # Track calls to signal.signal for the fallback path
+                    registered_handlers = {}
+
+                    def capture_signal(sig, handler):
+                        registered_handlers[sig] = handler
+
+                    mock_signal.side_effect = capture_signal
+
+                    async def run_and_shutdown():
+                        task = asyncio.create_task(run_watch())
+                        await asyncio.sleep(0.01)
+                        # Trigger shutdown via the fallback handler
+                        if signal.SIGINT in registered_handlers:
+                            registered_handlers[signal.SIGINT](signal.SIGINT, None)
+                        await task
+
+                    await run_and_shutdown()
+
+                # Verify fallback signal.signal was called for both signals
+                assert mock_signal.call_count == 2
+                called_signals = {call.args[0] for call in mock_signal.call_args_list}
+                assert signal.SIGINT in called_signals
+                assert signal.SIGTERM in called_signals

tests/repository/test_postgres_search_repository.py

@@ -432,9 +432,7 @@ class StubEmbeddingProvider8d:
 
 
 @pytest.mark.asyncio
-async def test_postgres_dimension_mismatch_triggers_table_recreation(
-    session_maker, test_project
-):
+async def test_postgres_dimension_mismatch_triggers_table_recreation(session_maker, test_project):
     """Changing embedding dimensions should drop and recreate the embeddings table."""
     await _skip_if_pgvector_unavailable(session_maker)
 

tests/services/test_search_service.py

@@ -980,17 +980,19 @@ async def test_reindex_vectors(search_service, session_maker, test_project):
 
     # Create some entities
     for i in range(3):
-        entity = await entity_repo.create({
-            "title": f"Vector Test Entity {i}",
-            "entity_type": "note",
-            "entity_metadata": {},
-            "content_type": "text/markdown",
-            "file_path": f"test/vector-test-{i}.md",
-            "permalink": f"test/vector-test-{i}",
-            "project_id": test_project.id,
-            "created_at": datetime.now(),
-            "updated_at": datetime.now(),
-        })
+        entity = await entity_repo.create(
+            {
+                "title": f"Vector Test Entity {i}",
+                "entity_type": "note",
+                "entity_metadata": {},
+                "content_type": "text/markdown",
+                "file_path": f"test/vector-test-{i}.md",
+                "permalink": f"test/vector-test-{i}",
+                "project_id": test_project.id,
+                "created_at": datetime.now(),
+                "updated_at": datetime.now(),
+            }
+        )
         await search_service.index_entity(entity, content=f"Content for entity {i}")
 
     # Track progress calls
@@ -1020,17 +1022,19 @@ async def test_reindex_vectors_no_callback(search_service, session_maker, test_p
     from datetime import datetime
 
     entity_repo = EntityRepository(session_maker, project_id=test_project.id)
-    entity = await entity_repo.create({
-        "title": "No Callback Entity",
-        "entity_type": "note",
-        "entity_metadata": {},
-        "content_type": "text/markdown",
-        "file_path": "test/no-callback.md",
-        "permalink": "test/no-callback",
-        "project_id": test_project.id,
-        "created_at": datetime.now(),
-        "updated_at": datetime.now(),
-    })
+    entity = await entity_repo.create(
+        {
+            "title": "No Callback Entity",
+            "entity_type": "note",
+            "entity_metadata": {},
+            "content_type": "text/markdown",
+            "file_path": "test/no-callback.md",
+            "permalink": "test/no-callback",
+            "project_id": test_project.id,
+            "created_at": datetime.now(),
+            "updated_at": datetime.now(),
+        }
+    )
     await search_service.index_entity(entity, content="Test content")
 
     stats = await search_service.reindex_vectors()

tests/sync/test_coordinator.py

@@ -1,5 +1,7 @@
 """Tests for SyncCoordinator - centralized sync/watch lifecycle."""
 
+import asyncio
+
 import pytest
 from unittest.mock import AsyncMock, patch
 
@@ -21,6 +23,16 @@ class TestSyncCoordinator:
         assert coordinator.status == SyncStatus.NOT_STARTED
         assert coordinator.is_running is False
 
+    def test_quiet_defaults_to_true(self, mock_config):
+        """quiet field defaults to True (for MCP/background use)."""
+        coordinator = SyncCoordinator(config=mock_config)
+        assert coordinator.quiet is True
+
+    def test_quiet_can_be_set_false(self, mock_config):
+        """quiet field can be set to False (for CLI watch command)."""
+        coordinator = SyncCoordinator(config=mock_config, quiet=False)
+        assert coordinator.quiet is False
+
     @pytest.mark.asyncio
     async def test_start_when_sync_disabled(self, mock_config):
         """When should_sync is False, start() sets status to STOPPED."""
@@ -113,6 +125,27 @@ class TestSyncCoordinator:
             assert coordinator.status == SyncStatus.STOPPED
             assert coordinator._sync_task is None
 
+    @pytest.mark.asyncio
+    async def test_start_passes_quiet_to_file_sync(self, mock_config):
+        """quiet value is threaded through to initialize_file_sync."""
+        coordinator = SyncCoordinator(
+            config=mock_config,
+            should_sync=True,
+            quiet=False,
+        )
+
+        with patch(
+            "basic_memory.services.initialization.initialize_file_sync",
+            new_callable=AsyncMock,
+        ) as mock_init:
+            await coordinator.start()
+            # Let the task run so initialize_file_sync gets called
+            await asyncio.sleep(0.01)
+
+            mock_init.assert_called_once_with(mock_config, quiet=False)
+
+            await coordinator.stop()
+
     @pytest.mark.asyncio
     async def test_start_already_running(self, mock_config):
         """Starting when already running is a no-op."""