fix: normalize underscores in memory:// URLs for build_context

src/basic_memory/repository/entity_repository.py

@@ -101,62 +101,23 @@ async def find_by_permalinks(self, permalinks: List[str]) -> Sequence[Entity]:
         return list(result.scalars().all())
 
     async def upsert_entity(self, entity: Entity) -> Entity:
-        """Insert or update entity using a hybrid approach.
+        """Insert or update entity using simple try/catch with database-level conflict resolution.
 
-        This method provides a cleaner alternative to the try/catch approach
-        for handling permalink and file_path conflicts. It first tries direct
-        insertion, then handles conflicts intelligently.
+        Handles file_path race conditions by checking for existing entity on IntegrityError.
+        For permalink conflicts, generates a unique permalink with numeric suffix.
 
         Args:
             entity: The entity to insert or update
 
         Returns:
             The inserted or updated entity
         """
-
         async with db.scoped_session(self.session_maker) as session:
             # Set project_id if applicable and not already set
             self._set_project_id_if_needed(entity)
 
-            # Check for existing entity with same file_path first
-            existing_by_path = await session.execute(
-                select(Entity).where(
-                    Entity.file_path == entity.file_path, Entity.project_id == entity.project_id
-                )
-            )
-            existing_path_entity = existing_by_path.scalar_one_or_none()
-
-            if existing_path_entity:
-                # Update existing entity with same file path
-                for key, value in {
-                    "title": entity.title,
-                    "entity_type": entity.entity_type,
-                    "entity_metadata": entity.entity_metadata,
-                    "content_type": entity.content_type,
-                    "permalink": entity.permalink,
-                    "checksum": entity.checksum,
-                    "updated_at": entity.updated_at,
-                }.items():
-                    setattr(existing_path_entity, key, value)
-
-                await session.flush()
-                # Return with relationships loaded
-                query = (
-                    self.select()
-                    .where(Entity.file_path == entity.file_path)
-                    .options(*self.get_load_options())
-                )
-                result = await session.execute(query)
-                found = result.scalar_one_or_none()
-                if not found:  # pragma: no cover
-                    raise RuntimeError(
-                        f"Failed to retrieve entity after update: {entity.file_path}"
-                    )
-                return found
-
-            # No existing entity with same file_path, try insert
+            # Try simple insert first
             try:
-                # Simple insert for new entity
                 session.add(entity)
                 await session.flush()
 
@@ -175,20 +136,20 @@ async def upsert_entity(self, entity: Entity) -> Entity:
                 return found
 
             except IntegrityError:
-                # Could be either file_path or permalink conflict
                 await session.rollback()
 
-                # Check if it's a file_path conflict (race condition)
-                existing_by_path_check = await session.execute(
-                    select(Entity).where(
+                # Re-query after rollback to get a fresh, attached entity
+                existing_result = await session.execute(
+                    select(Entity)
+                    .where(
                         Entity.file_path == entity.file_path, Entity.project_id == entity.project_id
                     )
+                    .options(*self.get_load_options())
                 )
-                race_condition_entity = existing_by_path_check.scalar_one_or_none()
+                existing_entity = existing_result.scalar_one_or_none()
 
-                if race_condition_entity:
-                    # Race condition: file_path conflict detected after our initial check
-                    # Update the existing entity instead
+                if existing_entity:
+                    # File path conflict - update the existing entity
                     for key, value in {
                         "title": entity.title,
                         "entity_type": entity.entity_type,
@@ -198,25 +159,22 @@ async def upsert_entity(self, entity: Entity) -> Entity:
                         "checksum": entity.checksum,
                         "updated_at": entity.updated_at,
                     }.items():
-                        setattr(race_condition_entity, key, value)
-
-                    await session.flush()
-                    # Return the updated entity with relationships loaded
-                    query = (
-                        self.select()
-                        .where(Entity.file_path == entity.file_path)
-                        .options(*self.get_load_options())
-                    )
-                    result = await session.execute(query)
-                    found = result.scalar_one_or_none()
-                    if not found:  # pragma: no cover
-                        raise RuntimeError(
-                            f"Failed to retrieve entity after race condition update: {entity.file_path}"
-                        )
-                    return found
+                        setattr(existing_entity, key, value)
+
+                    # Clear and re-add observations
+                    existing_entity.observations.clear()
+                    for obs in entity.observations:
+                        obs.entity_id = existing_entity.id
+                        existing_entity.observations.append(obs)
+
+                    await session.commit()
+                    return existing_entity
+
                 else:
-                    # Must be permalink conflict - generate unique permalink
-                    return await self._handle_permalink_conflict(entity, session)
+                    # No file_path conflict - must be permalink conflict
+                    # Generate unique permalink and retry
+                    entity = await self._handle_permalink_conflict(entity, session)
+                    return entity
 
     async def _handle_permalink_conflict(self, entity: Entity, session: AsyncSession) -> Entity:
         """Handle permalink conflicts by generating a unique permalink."""
@@ -237,18 +195,7 @@ async def _handle_permalink_conflict(self, entity: Entity, session: AsyncSession
                 break
             suffix += 1
 
-        # Insert with unique permalink (no conflict possible now)
+        # Insert with unique permalink
         session.add(entity)
         await session.flush()
-
-        # Return the inserted entity with relationships loaded
-        query = (
-            self.select()
-            .where(Entity.file_path == entity.file_path)
-            .options(*self.get_load_options())
-        )
-        result = await session.execute(query)
-        found = result.scalar_one_or_none()
-        if not found:  # pragma: no cover
-            raise RuntimeError(f"Failed to retrieve entity after insert: {entity.file_path}")
-        return found
+        return entity

src/basic_memory/services/context_service.py

@@ -100,20 +100,30 @@ async def build_context(
             f"Building context for URI: '{memory_url}' depth: '{depth}' since: '{since}' limit: '{limit}' offset: '{offset}'  max_related: '{max_related}'"
         )
 
+        normalized_path: Optional[str] = None
         if memory_url:
             path = memory_url_path(memory_url)
-            # Pattern matching - use search
-            if "*" in path:
-                logger.debug(f"Pattern search for '{path}'")
+            # Check for wildcards before normalization
+            has_wildcard = "*" in path
+
+            if has_wildcard:
+                # For wildcard patterns, normalize each segment separately to preserve the *
+                parts = path.split("*")
+                normalized_parts = [
+                    generate_permalink(part, split_extension=False) if part else ""
+                    for part in parts
+                ]
+                normalized_path = "*".join(normalized_parts)
+                logger.debug(f"Pattern search for '{normalized_path}'")
                 primary = await self.search_repository.search(
-                    permalink_match=path, limit=limit, offset=offset
+                    permalink_match=normalized_path, limit=limit, offset=offset
                 )
-
-            # Direct lookup for exact path
             else:
-                logger.debug(f"Direct lookup for '{path}'")
+                # For exact paths, normalize the whole thing
+                normalized_path = generate_permalink(path, split_extension=False)
+                logger.debug(f"Direct lookup for '{normalized_path}'")
                 primary = await self.search_repository.search(
-                    permalink=path, limit=limit, offset=offset
+                    permalink=normalized_path, limit=limit, offset=offset
                 )
         else:
             logger.debug(f"Build context for '{types}'")
@@ -151,7 +161,7 @@ async def build_context(
 
         # Create metadata dataclass
         metadata = ContextMetadata(
-            uri=memory_url_path(memory_url) if memory_url else None,
+            uri=normalized_path if memory_url else None,
             types=types,
             depth=depth,
             timeframe=since.isoformat() if since else None,

test-int/mcp/test_build_context_underscore.py

@@ -0,0 +1,171 @@
+"""Integration test for build_context with underscore in memory:// URLs."""
+
+import pytest
+from fastmcp import Client
+
+
+@pytest.mark.asyncio
+async def test_build_context_underscore_normalization(mcp_server, app, test_project):
+    """Test that build_context normalizes underscores in relation types."""
+
+    async with Client(mcp_server) as client:
+        # Create parent note
+        await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Parent Entity",
+                "folder": "testing",
+                "content": "# Parent Entity\n\nMain entity for testing underscore relations.",
+                "tags": "test,parent",
+            },
+        )
+
+        # Create child notes with different relation formats
+        await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Child with Underscore",
+                "folder": "testing",
+                "content": """# Child with Underscore
+
+- part_of [[Parent Entity]]
+- related_to [[Parent Entity]]
+                """,
+                "tags": "test,child",
+            },
+        )
+
+        await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Child with Hyphen",
+                "folder": "testing",
+                "content": """# Child with Hyphen
+
+- part-of [[Parent Entity]]
+- related-to [[Parent Entity]]
+                """,
+                "tags": "test,child",
+            },
+        )
+
+        # Test 1: Search with underscore format should return results
+        # Relation permalinks are: source/relation_type/target
+        # So child-with-underscore/part-of/parent-entity
+        result_underscore = await client.call_tool(
+            "build_context",
+            {
+                "project": test_project.name,
+                "url": "memory://testing/*/part_of/*parent*",  # Using underscore
+            },
+        )
+
+        # Parse response
+        assert len(result_underscore.content) == 1
+        response_text = result_underscore.content[0].text  # pyright: ignore
+        assert '"results"' in response_text
+
+        # Both relations should be found since they both connect to parent-entity
+        # The system should normalize the underscore to hyphen internally
+        assert "part-of" in response_text.lower()
+
+        # Test 2: Search with hyphen format should also return results
+        result_hyphen = await client.call_tool(
+            "build_context",
+            {
+                "project": test_project.name,
+                "url": "memory://testing/*/part-of/*parent*",  # Using hyphen
+            },
+        )
+
+        response_text_hyphen = result_hyphen.content[0].text  # pyright: ignore
+        assert '"results"' in response_text_hyphen
+        assert "part-of" in response_text_hyphen.lower()
+
+        # Test 3: Test with related_to/related-to as well
+        result_related = await client.call_tool(
+            "build_context",
+            {
+                "project": test_project.name,
+                "url": "memory://testing/*/related_to/*parent*",  # Using underscore
+            },
+        )
+
+        response_text_related = result_related.content[0].text  # pyright: ignore
+        assert '"results"' in response_text_related
+        assert "related-to" in response_text_related.lower()
+
+        # Test 4: Test exact path (non-wildcard) with underscore
+        # Exact relation permalink would be child/relation/target
+        result_exact = await client.call_tool(
+            "build_context",
+            {
+                "project": test_project.name,
+                "url": "memory://testing/child-with-underscore/part_of/testing/parent-entity",
+            },
+        )
+
+        response_text_exact = result_exact.content[0].text  # pyright: ignore
+        assert '"results"' in response_text_exact
+        assert "part-of" in response_text_exact.lower()
+
+
+@pytest.mark.asyncio
+async def test_build_context_complex_underscore_paths(mcp_server, app, test_project):
+    """Test build_context with complex paths containing underscores."""
+
+    async with Client(mcp_server) as client:
+        # Create notes with underscores in titles and relations
+        await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "workflow_manager_agent",
+                "folder": "specs",
+                "content": """# Workflow Manager Agent
+
+Specification for the workflow manager agent.
+                """,
+                "tags": "spec,workflow",
+            },
+        )
+
+        await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "task_parser",
+                "folder": "components",
+                "content": """# Task Parser
+
+- part_of [[workflow_manager_agent]]
+- implements_for [[workflow_manager_agent]]
+                """,
+                "tags": "component,parser",
+            },
+        )
+
+        # Test with underscores in all parts of the path
+        # Relations are created as: task-parser/part-of/workflow-manager-agent
+        # So search for */part_of/* or */part-of/* to find them
+        test_cases = [
+            "memory://components/*/part_of/*workflow*",
+            "memory://components/*/part-of/*workflow*",
+            "memory://*/task*/part_of/*",
+            "memory://*/task*/part-of/*",
+        ]
+
+        for url in test_cases:
+            result = await client.call_tool(
+                "build_context", {"project": test_project.name, "url": url}
+            )
+
+            # All variations should work and find the related content
+            assert len(result.content) == 1
+            response = result.content[0].text  # pyright: ignore
+            assert '"results"' in response
+            # The relation should be found showing part-of connection
+            assert "part-of" in response.lower(), f"Failed for URL: {url}"