diff --git a/pyproject.toml b/pyproject.toml
index a6be69aa..f9715b42 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -23,6 +23,7 @@ dependencies = [
     "rich>=13.9.4",
     "unidecode>=1.3.8",
     "dateparser>=1.2.0",
+    "watchfiles>=1.0.4",
 ]
 
 [project.optional-dependencies]
diff --git a/src/basic_memory/config.py b/src/basic_memory/config.py
index b1337a66..ad02c498 100644
--- a/src/basic_memory/config.py
+++ b/src/basic_memory/config.py
@@ -22,6 +22,10 @@ class ProjectConfig(BaseSettings):
     # Name of the project
     project: str = Field(default="default", description="Project name")
 
+    # Watch service configuration
+    sync_delay: int = Field(
+        default=500, description="Milliseconds to wait after changes before syncing", gt=0
+    )
 
     model_config = SettingsConfigDict(
         env_prefix="BASIC_MEMORY_",
diff --git a/src/basic_memory/services/file_service.py b/src/basic_memory/services/file_service.py
index f8b89974..5ca830be 100644
--- a/src/basic_memory/services/file_service.py
+++ b/src/basic_memory/services/file_service.py
@@ -208,3 +208,6 @@ class FileService:
         except Exception as e:
             logger.error(f"Failed to delete file {path}: {e}")
             raise FileOperationError(f"Failed to delete file: {e}")
+
+    def path(self, path_string: str, absolute: bool = False):
+        return Path( self.base_path / path_string ) if absolute else Path(path_string).relative_to(self.base_path)
\ No newline at end of file
diff --git a/src/basic_memory/sync/sync_service.py b/src/basic_memory/sync/sync_service.py
index 273d5187..94225288 100644
--- a/src/basic_memory/sync/sync_service.py
+++ b/src/basic_memory/sync/sync_service.py
@@ -1,7 +1,7 @@
 """Service for syncing files between filesystem and database."""
 
 from pathlib import Path
-from typing import Dict
+from typing import Dict, Optional
 
 from loguru import logger
 
@@ -10,7 +10,8 @@ from basic_memory.repository import EntityRepository, RelationRepository
 from basic_memory.services import EntityService
 from basic_memory.services.search_service import SearchService
 from basic_memory.sync import FileChangeScanner
-from basic_memory.sync.utils import SyncReport
+from basic_memory.sync.utils import SyncReport, FileChange
+from watchfiles import Change
 
 
 class SyncService:
@@ -60,10 +61,28 @@ class SyncService:
         else:
             logger.debug(f"No entity found to delete: {file_path}")
 
-    async def sync(self, directory: Path) -> SyncReport:
+    async def sync(self, directory: Optional[Path] = None, file_changes: Optional[dict[str, FileChange]] = None) -> SyncReport:
         """Sync knowledge files with database."""
-        changes = await self.scanner.find_knowledge_changes(directory)
-        logger.info(f"Found {changes.total_changes} knowledge changes")
+        if file_changes is not None:
+            changes = SyncReport()
+            for path, file_change in file_changes.items():
+                logger.debug(f"path {path} file_change {file_change}")
+                match file_change.change_type:
+                    case Change.added:
+                        changes.new.add(path)
+                        changes.checksums[path] = file_change.checksum
+                    case Change.modified:
+                        changes.modified.add(path)
+                        changes.checksums[path] = file_change.checksum
+                    case Change.deleted:
+                        changes.deleted.add(path)
+        else:
+            # Traditional directory scan mode
+            if directory is None:
+                raise ValueError("Must provide either directory or file_changes")
+            
+            changes = await self.scanner.find_knowledge_changes(directory)
+            logger.info(f"Found {changes.total_changes} knowledge changes")
 
         # Handle moves first
         for old_path, new_path in changes.moves.items():
@@ -86,7 +105,7 @@ class SyncService:
         parsed_entities: Dict[str, EntityMarkdown] = {}
 
         for file_path in [*changes.new, *changes.modified]:
-            entity_markdown = await self.entity_parser.parse_file(directory / file_path)
+            entity_markdown = await self.entity_parser.parse_file(Path(file_path))
             parsed_entities[file_path] = entity_markdown
 
         # First pass: Create/update entities
diff --git a/src/basic_memory/sync/utils.py b/src/basic_memory/sync/utils.py
index 8276e226..fcaf7dbd 100644
--- a/src/basic_memory/sync/utils.py
+++ b/src/basic_memory/sync/utils.py
@@ -2,6 +2,38 @@
 
 from dataclasses import dataclass, field
 from typing import Set, Dict, Optional
+from watchfiles import Change
+from basic_memory.services.file_service import FileService
+
+
+@dataclass
+class FileChange:
+    """A change to a file detected by the watch service.
+    
+    Attributes:
+        change_type: Type of change (added, modified, deleted)
+        path: Path to the file
+        checksum: File checksum (None for deleted files)
+    """
+    change_type: Change
+    path: str
+    checksum: Optional[str] = None
+
+    @classmethod
+    async def from_path(cls, path: str, change_type: Change, file_service: FileService) -> "FileChange":
+        """Create FileChange from a path, computing checksum if file exists.
+        
+        Args:
+            path: Path to the file
+            change_type: Type of change detected
+            file_service: Service to read file and compute checksum
+            
+        Returns:
+            FileChange with computed checksum for non-deleted files
+        """
+        file_path = file_service.path(path)
+        content, checksum = await file_service.read_file(file_path) if change_type != Change.deleted else (None, None)
+        return cls(path=file_path, change_type=change_type, checksum=checksum)
 
 
 @dataclass
@@ -18,7 +50,7 @@ class SyncReport:
     new: Set[str] = field(default_factory=set)
     modified: Set[str] = field(default_factory=set)
     deleted: Set[str] = field(default_factory=set)
-    moves: Dict[str, str] = field(default_factory=dict)  # old_path -> new_path
+    moves: Dict[str, str] = field(default_factory=dict)  # old_path -> new_path  
     checksums: Dict[str, str] = field(default_factory=dict)
 
     @property
diff --git a/src/basic_memory/sync/watch_service.py b/src/basic_memory/sync/watch_service.py
new file mode 100644
index 00000000..6047085d
--- /dev/null
+++ b/src/basic_memory/sync/watch_service.py
@@ -0,0 +1,131 @@
+"""Watch service for Basic Memory."""
+
+import json
+import dataclasses
+
+from pydantic import BaseModel
+from pydantic.dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import List, Optional
+from watchfiles import awatch, Change
+import os
+
+from basic_memory.config import ProjectConfig
+from basic_memory.sync.sync_service import SyncService
+from basic_memory.services.file_service import FileService
+from basic_memory.sync.utils import FileChange
+
+
+class WatchEvent(BaseModel):
+    timestamp: datetime
+    path: str
+    action: str  # sync, delete, etc
+    status: str  # success, error
+    error: Optional[str] = None
+
+
+class WatchServiceState(BaseModel):
+    # Service status
+    running: bool = False
+    start_time: datetime = dataclasses.field(default_factory=datetime.now)
+    pid: int = dataclasses.field(default_factory=os.getpid)
+
+    # Stats
+    files_synced: int = 0
+    bytes_processed: int = 0
+    error_count: int = 0
+    last_error: Optional[datetime] = None
+    last_scan: Optional[datetime] = None
+
+    # File counts
+    total_files: int = 0
+    markdown_files: int = 0
+
+    # Recent activity
+    recent_events: List[WatchEvent] = dataclasses.field(default_factory=list)
+
+    def add_event(self, path: str, action: str, status: str, error: Optional[str] = None):
+        event = WatchEvent(
+            timestamp=datetime.now(), path=path, action=action, status=status, error=error
+        )
+        self.recent_events.insert(0, event)
+        self.recent_events = self.recent_events[:100]  # Keep last 100
+
+    def record_error(self, error: str):
+        self.error_count += 1
+        self.last_error = datetime.now()
+
+
+class WatchService:
+    def __init__(self, sync_service: SyncService, file_service: FileService, config: ProjectConfig):
+        self.sync_service = sync_service
+        self.file_service = file_service
+        self.config = config
+        self.state = WatchServiceState()
+        self.status_path = config.home / ".basic-memory" / "watch-status.json"
+        self.status_path.parent.mkdir(parents=True, exist_ok=True)
+
+    async def run(self):
+        """Watch for file changes and sync them"""
+        self.state.running = True
+        self.state.start_time = datetime.now()
+        await self.write_status()
+
+        try:
+            async for changes in awatch(
+                self.config.home,
+                watch_filter=self.filter_changes,
+                debounce=self.config.sync_delay,
+                recursive=True,
+            ):
+                await self.handle_changes(changes)
+
+        except Exception as e:
+            self.state.record_error(str(e))
+            await self.write_status()
+            raise
+        finally:
+            self.state.running = False
+            await self.write_status()
+
+    async def write_status(self):
+        """Write current state to status file"""
+        self.status_path.write_text(WatchServiceState.model_dump_json(self.state, indent=2))
+
+    def filter_changes(self, change: Change, path: str) -> bool:
+        """Filter to only watch markdown files"""
+        return path.endswith(".md") and not Path(path).name.startswith(".")
+
+    async def handle_changes(self, changes: set[tuple[Change, str]]):
+        """Process a batch of file changes"""
+
+        # Group changes by file path
+        changes_by_file = {}
+        try:
+            for change_type, path in changes:
+                file_change = await FileChange.from_path(path, change_type, self.file_service)
+                
+                # store changes by relative path
+                changes_by_file[str(file_change.path)] = file_change
+
+            # Process changes with timeout
+            await self.sync_service.sync(file_changes=changes_by_file)
+
+            # Update stats
+            self.state.files_synced += len(changes_by_file)
+            for path, change in changes_by_file.items():
+                if change.change_type != Change.deleted:
+                    size = self.file_service.path(path,absolute=True).stat().st_size
+                    self.state.bytes_processed += size
+
+                self.state.add_event(path=path, action="sync", status="success")
+
+            await self.write_status()
+
+        except Exception as e:
+            self.state.record_error(str(e))
+            for path in changes_by_file:
+                self.state.add_event(path=path, action="sync", status="error", error=str(e))
+            await self.write_status()
+            raise
diff --git a/tests/sync/test_watch_service.py b/tests/sync/test_watch_service.py
new file mode 100644
index 00000000..6026a905
--- /dev/null
+++ b/tests/sync/test_watch_service.py
@@ -0,0 +1,166 @@
+"""Tests for the watch service."""
+
+from dataclasses import asdict
+import pytest
+import pytest_asyncio
+from datetime import datetime
+from pathlib import Path
+from watchfiles import Change
+
+from basic_memory.sync.utils import FileChange
+from basic_memory.sync.watch_service import WatchService, WatchEvent, WatchServiceState
+
+
+@pytest_asyncio.fixture
+def watch_service(sync_service, file_service, test_config):
+    return WatchService(
+        sync_service=sync_service,
+        file_service=file_service,
+        config=test_config
+    )
+
+
+@pytest_asyncio.fixture
+async def sample_markdown(tmp_path):
+    content = """---
+title: Test Note
+type: note
+---
+# Test Note
+This is a test note.
+"""
+    file_path = tmp_path / "test.md"
+    file_path.write_text(content)
+    return file_path
+
+
+@pytest.mark.asyncio
+async def test_handle_file_added(watch_service, sync_service, sample_markdown):
+    """Test handling a new file event"""
+    changes = {(Change.added, str(sample_markdown))}
+    
+    await watch_service.handle_changes(changes)
+    
+    # Check stats updated
+    assert watch_service.state.files_synced == 1
+    assert watch_service.state.bytes_processed > 0
+    
+    # Check event recorded
+    assert len(watch_service.state.recent_events) == 1
+    event = watch_service.state.recent_events[0]
+    assert event.path == "test.md"
+    assert event.action == "sync"
+    assert event.status == "success"
+
+
+@pytest.mark.asyncio
+async def test_handle_file_modified(watch_service, sync_service, sample_markdown):
+    """Test handling a modified file event"""
+    # First add the file
+    await watch_service.handle_changes({(Change.added, str(sample_markdown))})
+    
+    # Modify the file
+    sample_markdown.write_text(sample_markdown.read_text() + "\nModified content")
+    await watch_service.handle_changes({(Change.modified, str(sample_markdown))})
+    
+    # Should have two events
+    assert len(watch_service.state.recent_events) == 2
+    assert watch_service.state.files_synced == 2
+
+
+@pytest.mark.asyncio
+async def test_handle_file_deleted(watch_service, sync_service, sample_markdown):
+    """Test handling a deleted file event"""
+    # First add the file
+    await watch_service.handle_changes({(Change.added, str(sample_markdown))})
+    
+    # Delete the file
+    sample_markdown.unlink()
+    await watch_service.handle_changes({(Change.deleted, str(sample_markdown))})
+    
+    # Should have two events
+    assert len(watch_service.state.recent_events) == 2
+    delete_event = watch_service.state.recent_events[0]
+    assert delete_event.action == "sync"
+    assert delete_event.path == "test.md"
+
+
+@pytest.mark.asyncio
+async def test_filter_changes(watch_service):
+    """Test change filtering"""
+    assert watch_service.filter_changes(Change.added, "test.md") is True
+    assert watch_service.filter_changes(Change.added, "test.txt") is False
+    assert watch_service.filter_changes(Change.added, ".test.md") is False
+
+
+@pytest.mark.asyncio
+async def test_handle_error(watch_service, sync_service, ):
+    """Test error handling during sync"""
+    
+    changes = {(Change.added, "test.md")}
+    
+    # ValueError raised because "test.md" path does not exist
+    with pytest.raises(ValueError):
+        await watch_service.handle_changes(changes)
+    
+    assert watch_service.state.error_count == 1
+    assert watch_service.state.last_error is not None
+    
+
+@pytest.mark.asyncio
+async def test_recent_events_limit(watch_service):
+    """Test that recent events are limited to 100"""
+    for i in range(150):
+        watch_service.state.add_event(
+            path=f"test{i}.md",
+            action="sync",
+            status="success"
+        )
+    
+    assert len(watch_service.state.recent_events) == 100
+    # Most recent should be at the start
+    assert watch_service.state.recent_events[0].path == "test149.md"
+
+
+@pytest.mark.asyncio
+async def test_state_serialization(watch_service):
+    """Test state serializes to dict correctly"""
+    # Add some test state
+    watch_service.state.running = True
+    watch_service.state.files_synced = 42
+    watch_service.state.add_event(
+        path="test.md",
+        action="sync",
+        status="success"
+    )
+    
+    data = WatchServiceState.model_dump(watch_service.state)
+    
+    # Check basic fields
+    assert data['running'] is True
+    assert data['files_synced'] == 42
+    assert isinstance(data['start_time'], datetime)
+    
+    # Check events serialized
+    assert len(data['recent_events']) == 1
+    event = data['recent_events'][0]
+    assert event['path'] == "test.md"
+    assert event['action'] == "sync"
+    assert isinstance(event['timestamp'], datetime)
+
+
+@pytest.mark.asyncio
+async def test_status_file(watch_service, tmp_path):
+    """Test status file writing"""
+    watch_service.state.running = True
+    await watch_service.write_status()
+    
+    status_file = watch_service.status_path
+    assert status_file.exists()
+    
+    # Should be valid JSON with our fields
+    import json
+    data = json.loads(status_file.read_text())
+    assert data['running'] is True
+    assert isinstance(data['start_time'], str)
+    assert data['pid'] > 0
diff --git a/tests/sync/test_watch_sync.py b/tests/sync/test_watch_sync.py
new file mode 100644
index 00000000..6ba74a53
--- /dev/null
+++ b/tests/sync/test_watch_sync.py
@@ -0,0 +1,129 @@
+"""Tests for watch service integration with sync service."""
+
+import pytest
+import pytest_asyncio
+from pathlib import Path
+from watchfiles import Change
+
+from basic_memory.sync.utils import FileChange
+
+
+@pytest.mark.asyncio
+async def test_sync_with_file_changes(sync_service, tmp_path):
+    """Test sync with file changes from watch service"""
+    # Create a test file
+    content = """---
+title: Test Note
+type: note
+---
+# Test Note
+This is a test."""
+    
+    file_path = tmp_path / "test.md"
+    file_path.write_text(content)
+    
+    # Create FileChange for a new file
+    changes = {
+        str(file_path): FileChange(
+            change_type=Change.added,
+            path=str(file_path),
+            checksum="abc123"
+        )
+    }
+    
+    report = await sync_service.sync(file_changes=changes)
+    assert str(file_path) in report.new
+    assert report.checksums[str(file_path)] == "abc123"
+
+
+@pytest.mark.asyncio
+async def test_sync_mixed_changes(sync_service, tmp_path):
+    """Test sync with multiple types of changes"""
+    # Setup initial files
+    new_file = tmp_path / "new.md"
+    new_file.write_text("New file")
+    
+    mod_file = tmp_path / "modified.md"
+    mod_file.write_text("Original content")
+    
+    del_file = tmp_path / "deleted.md"
+    
+    changes = {
+        str(new_file): FileChange(
+            change_type=Change.added,
+            path=str(new_file),
+            checksum="new123"
+        ),
+        str(mod_file): FileChange(
+            change_type=Change.modified,
+            path=str(mod_file),
+            checksum="mod123"
+        ),
+        str(del_file): FileChange(
+            change_type=Change.deleted,
+            path=str(del_file)
+        )
+    }
+    
+    report = await sync_service.sync(file_changes=changes)
+    
+    # Verify report contains all changes
+    assert str(new_file) in report.new
+    assert str(mod_file) in report.modified
+    assert str(del_file) in report.deleted
+    
+    # Check checksums
+    assert report.checksums[str(new_file)] == "new123"
+    assert report.checksums[str(mod_file)] == "mod123"
+    assert str(del_file) not in report.checksums
+
+
+@pytest.mark.asyncio
+async def test_sync_requires_params(sync_service):
+    """Test sync requires either directory or file_changes"""
+    with pytest.raises(ValueError):
+        await sync_service.sync()
+
+
+@pytest.mark.asyncio
+async def test_sync_with_invalid_change_type(sync_service, tmp_path):
+    """Test handling of invalid change types"""
+    file_path = tmp_path / "test.md"
+    file_path.touch()
+    
+    # Type checker won't let us create this directly, so ignore the type
+    changes = {
+        str(file_path): FileChange(
+            change_type="invalid",  # type: ignore
+            path=str(file_path),
+            checksum="test123"
+        )
+    }
+    
+    with pytest.raises(Exception):
+        await sync_service.sync(file_changes=changes)
+
+
+@pytest.mark.asyncio
+async def test_end_to_end_watch_sync(watch_service, sync_service, tmp_path):
+    """Test complete watch -> sync flow"""
+    # Create a test file that will trigger a watch event
+    test_file = tmp_path / "test.md"
+    test_file.write_text("""---
+title: Test Note
+type: note
+---
+# Test
+Testing watch -> sync flow
+""")
+    
+    # Simulate watchfiles event
+    changes = {(Change.added, str(test_file))}
+    await watch_service.handle_changes(changes)
+    
+    # Check watch service state
+    assert watch_service.state.files_synced == 1
+    assert len(watch_service.state.recent_events) == 1
+    event = watch_service.state.recent_events[0]
+    assert event.path == str(test_file)
+    assert event.status == "success"
diff --git a/uv.lock b/uv.lock
index 37379d30..af1369d5 100644
--- a/uv.lock
+++ b/uv.lock
@@ -198,6 +198,7 @@ dependencies = [
     { name = "sqlalchemy" },
     { name = "typer" },
     { name = "unidecode" },
+    { name = "watchfiles" },
 ]
 
 [package.optional-dependencies]
@@ -238,6 +239,7 @@ requires-dist = [
     { name = "sqlalchemy", specifier = ">=2.0.0" },
     { name = "typer", specifier = ">=0.9.0" },
     { name = "unidecode", specifier = ">=1.3.8" },
+    { name = "watchfiles", specifier = ">=1.0.4" },
 ]
 
 [package.metadata.requires-dev]