fix: resolve sync race conditions and search errors

- Add IntegrityError handling in entity_service.create_entity_from_markdown for file_path/permalink constraint violations
- Add IntegrityError handling in sync_service.sync_regular_file for concurrent sync race conditions
- Fix FTS "unknown special query" error when searching for wildcard "*" patterns
- Add comprehensive test coverage for race condition edge cases and error handling
- Gracefully handle concurrent sync processes with fallback to update operations

Fixes sync errors from beta testing including:
- "UNIQUE constraint failed: entity.file_path"
- "UNIQUE constraint failed: entity.permalink"
- "unknown special query" FTS errors

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

Author: phernandez

src/basic_memory/repository/search_repository.py

@@ -237,19 +237,24 @@ async def search(
 
         # Handle text search for title and content
         if search_text:
-            # Check for explicit boolean operators - only detect them in proper boolean contexts
-            has_boolean = any(op in f" {search_text} " for op in [" AND ", " OR ", " NOT "])
-
-            if has_boolean:
-                # If boolean operators are present, use the raw query
-                # No need to prepare it, FTS5 will understand the operators
-                params["text"] = search_text
-                conditions.append("(title MATCH :text OR content_stems MATCH :text)")
+            # Skip FTS for wildcard-only queries that would cause "unknown special query" errors
+            if search_text.strip() == "*" or search_text.strip() == "":
+                # For wildcard searches, don't add any text conditions - return all results
+                pass
             else:
-                # Standard search with term preparation
-                processed_text = self._prepare_search_term(search_text.strip())
-                params["text"] = processed_text
-                conditions.append("(title MATCH :text OR content_stems MATCH :text)")
+                # Check for explicit boolean operators - only detect them in proper boolean contexts
+                has_boolean = any(op in f" {search_text} " for op in [" AND ", " OR ", " NOT "])
+
+                if has_boolean:
+                    # If boolean operators are present, use the raw query
+                    # No need to prepare it, FTS5 will understand the operators
+                    params["text"] = search_text
+                    conditions.append("(title MATCH :text OR content_stems MATCH :text)")
+                else:
+                    # Standard search with term preparation
+                    processed_text = self._prepare_search_term(search_text.strip())
+                    params["text"] = processed_text
+                    conditions.append("(title MATCH :text OR content_stems MATCH :text)")
 
         # Handle title match search
         if title:
@@ -453,4 +458,4 @@ async def execute_query(
             end_time = time.perf_counter()
             elapsed_time = end_time - start_time
             logger.debug(f"Query executed successfully in {elapsed_time:.2f}s.")
-            return result
+            return result

src/basic_memory/services/entity_service.py

@@ -299,7 +299,16 @@ async def create_entity_from_markdown(
         # Mark as incomplete because we still need to add relations
         model.checksum = None
         # Repository will set project_id automatically
-        return await self.repository.add(model)
+        try:
+            return await self.repository.add(model)
+        except IntegrityError as e:
+            # Handle race condition where entity was created by another process
+            if "UNIQUE constraint failed: entity.file_path" in str(e) or "UNIQUE constraint failed: entity.permalink" in str(e):
+                logger.info(f"Entity already exists for file_path={file_path} (file_path or permalink conflict), updating instead of creating")
+                return await self.update_entity_and_observations(file_path, markdown)
+            else:
+                # Re-raise if it's a different integrity error
+                raise
 
     async def update_entity_and_observations(
         self, file_path: Path, markdown: EntityMarkdown

src/basic_memory/sync/sync_service.py

@@ -364,18 +364,41 @@ async def sync_regular_file(self, path: str, new: bool = True) -> Tuple[Optional
             content_type = self.file_service.content_type(path)
 
             file_path = Path(path)
-            entity = await self.entity_repository.add(
-                Entity(
-                    entity_type="file",
-                    file_path=path,
-                    checksum=checksum,
-                    title=file_path.name,
-                    created_at=created,
-                    updated_at=modified,
-                    content_type=content_type,
+            try:
+                entity = await self.entity_repository.add(
+                    Entity(
+                        entity_type="file",
+                        file_path=path,
+                        checksum=checksum,
+                        title=file_path.name,
+                        created_at=created,
+                        updated_at=modified,
+                        content_type=content_type,
+                    )
                 )
-            )
-            return entity, checksum
+                return entity, checksum
+            except IntegrityError as e:
+                # Handle race condition where entity was created by another process
+                if "UNIQUE constraint failed: entity.file_path" in str(e):
+                    logger.info(f"Entity already exists for file_path={path}, updating instead of creating")
+                    # Treat as update instead of create
+                    entity = await self.entity_repository.get_by_file_path(path)
+                    if entity is None:  # pragma: no cover
+                        logger.error(f"Entity not found after constraint violation, path={path}")
+                        raise ValueError(f"Entity not found after constraint violation: {path}")
+                    
+                    updated = await self.entity_repository.update(
+                        entity.id, {"file_path": path, "checksum": checksum}
+                    )
+                    
+                    if updated is None:  # pragma: no cover
+                        logger.error(f"Failed to update entity, entity_id={entity.id}, path={path}")
+                        raise ValueError(f"Failed to update entity with ID {entity.id}")
+                    
+                    return updated, checksum
+                else:
+                    # Re-raise if it's a different integrity error
+                    raise
         else:
             entity = await self.entity_repository.get_by_file_path(path)
             if entity is None:  # pragma: no cover

tests/repository/test_search_repository.py

@@ -483,3 +483,37 @@ async def test_version_string_search_integration(self, search_repository, search
         # Test with other problematic patterns
         results3 = await search_repository.search(search_text="node.js version")
         assert isinstance(results3, list)  # Should not crash
+
+    @pytest.mark.asyncio
+    async def test_wildcard_only_search(self, search_repository, search_entity):
+        """Test that wildcard-only search '*' doesn't cause FTS5 errors (line 243 coverage)."""
+        # Index an entity for testing
+        search_row = SearchIndexRow(
+            id=search_entity.id,
+            type=SearchItemType.ENTITY.value,
+            title="Test Entity",
+            content_stems="test entity content",
+            content_snippet="This is a test entity",
+            permalink=search_entity.permalink,
+            file_path=search_entity.file_path,
+            entity_id=search_entity.id,
+            metadata={"entity_type": search_entity.entity_type},
+            created_at=search_entity.created_at,
+            updated_at=search_entity.updated_at,
+            project_id=search_repository.project_id,
+        )
+
+        await search_repository.index_item(search_row)
+
+        # Test wildcard-only search - should not crash and should return results
+        results = await search_repository.search(search_text="*")
+        assert isinstance(results, list)  # Should not crash
+        assert len(results) >= 1  # Should return all results, including our test entity
+
+        # Test empty string search - should also not crash
+        results_empty = await search_repository.search(search_text="")
+        assert isinstance(results_empty, list)  # Should not crash
+
+        # Test whitespace-only search
+        results_whitespace = await search_repository.search(search_text="   ")
+        assert isinstance(results_whitespace, list)  # Should not crash

tests/services/test_entity_service.py

@@ -869,6 +869,109 @@ async def test_edit_entity_with_observations_and_relations(
     assert new_rel.relation_type == "relates to"
 
 
+@pytest.mark.asyncio
+async def test_create_entity_from_markdown_race_condition_handling(
+    entity_service: EntityService, file_service: FileService
+):
+    """Test that create_entity_from_markdown handles race condition with IntegrityError (lines 304-311)."""
+    from unittest.mock import patch, AsyncMock
+    from sqlalchemy.exc import IntegrityError
+
+    file_path = Path("test/race-condition.md")
+    
+    # Create a mock EntityMarkdown object
+    from basic_memory.markdown.schemas import EntityFrontmatter, EntityMarkdown as RealEntityMarkdown
+    from datetime import datetime, timezone
+    
+    frontmatter = EntityFrontmatter(metadata={"title": "Race Condition Test", "type": "test"})
+    markdown = RealEntityMarkdown(
+        frontmatter=frontmatter, 
+        observations=[], 
+        relations=[],
+        created=datetime.now(timezone.utc),
+        modified=datetime.now(timezone.utc)
+    )
+
+    # Mock the repository.add to raise IntegrityError on first call, then succeed on second
+    original_add = entity_service.repository.add
+    original_update = entity_service.update_entity_and_observations
+    
+    call_count = 0
+    
+    async def mock_add(*args, **kwargs):
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            # Simulate race condition - another process created the entity
+            raise IntegrityError("UNIQUE constraint failed: entity.file_path", None, None)
+        else:
+            return await original_add(*args, **kwargs)
+    
+    # Mock update method to return a dummy entity
+    async def mock_update(*args, **kwargs):
+        from basic_memory.models import Entity
+        from datetime import datetime, timezone
+        
+        return Entity(
+            id=1,
+            title="Race Condition Test",
+            entity_type="test",
+            file_path=str(file_path),
+            permalink="test/race-condition-test",
+            content_type="text/markdown",
+            created_at=datetime.now(timezone.utc),
+            updated_at=datetime.now(timezone.utc),
+        )
+
+    with patch.object(entity_service.repository, 'add', side_effect=mock_add), \
+         patch.object(entity_service, 'update_entity_and_observations', side_effect=mock_update) as mock_update_call:
+        
+        # Call the method
+        result = await entity_service.create_entity_from_markdown(file_path, markdown)
+        
+        # Verify it handled the race condition gracefully
+        assert result is not None
+        assert result.title == "Race Condition Test"
+        assert result.file_path == str(file_path)
+        
+        # Verify that update_entity_and_observations was called as fallback
+        mock_update_call.assert_called_once_with(file_path, markdown)
+
+
+@pytest.mark.asyncio
+async def test_create_entity_from_markdown_integrity_error_reraise(
+    entity_service: EntityService, file_service: FileService
+):
+    """Test that create_entity_from_markdown re-raises IntegrityError for non-race-condition cases."""
+    from unittest.mock import patch
+    from sqlalchemy.exc import IntegrityError
+
+    file_path = Path("test/integrity-error.md")
+    
+    # Create a mock EntityMarkdown object
+    from basic_memory.markdown.schemas import EntityFrontmatter, EntityMarkdown as RealEntityMarkdown
+    from datetime import datetime, timezone
+    
+    frontmatter = EntityFrontmatter(metadata={"title": "Integrity Error Test", "type": "test"})
+    markdown = RealEntityMarkdown(
+        frontmatter=frontmatter, 
+        observations=[], 
+        relations=[],
+        created=datetime.now(timezone.utc),
+        modified=datetime.now(timezone.utc)
+    )
+
+    # Mock the repository.add to raise a different IntegrityError (not file_path/permalink constraint)
+    async def mock_add(*args, **kwargs):
+        # Simulate a different constraint violation
+        raise IntegrityError("UNIQUE constraint failed: entity.some_other_field", None, None)
+
+    with patch.object(entity_service.repository, 'add', side_effect=mock_add):
+        # Should re-raise the IntegrityError since it's not a file_path/permalink constraint
+        with pytest.raises(IntegrityError, match="UNIQUE constraint failed: entity.some_other_field"):
+            await entity_service.create_entity_from_markdown(file_path, markdown)
+
+
 # Edge case tests for find_replace operation
 @pytest.mark.asyncio
 async def test_edit_entity_find_replace_not_found(entity_service: EntityService):