feat: Automatically update Homebrew

Signed-off-by: Drew Cain <groksrc@users.noreply.github.com>

.github/workflows/release.yml

@@ -51,4 +51,32 @@ jobs:
       - name: Publish to PyPI
         uses: pypa/gh-action-pypi-publish@release/v1
         with:
-          password: ${{ secrets.PYPI_TOKEN }}
+          password: ${{ secrets.PYPI_TOKEN }}
+
+  homebrew:
+    name: Update Homebrew Formula
+    needs: release
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      actions: read
+    steps:
+      - name: Update Homebrew formula
+        uses: mislav/bump-homebrew-formula-action@v3
+        with:
+          # Formula name in homebrew-basic-memory repo
+          formula-name: basic-memory
+          # The tap repository
+          homebrew-tap: basicmachines-co/homebrew-basic-memory
+          # Base branch of the tap repository
+          base-branch: main
+          # Download URL will be automatically constructed from the tag
+          download-url: https://github.com/basicmachines-co/basic-memory/archive/refs/tags/${{ github.ref_name }}.tar.gz
+          # Commit message for the formula update
+          commit-message: |
+            {{formulaName}} {{version}}
+
+            Created by https://github.com/basicmachines-co/basic-memory/actions/runs/${{ github.run_id }}
+        env:
+          # Personal Access Token with repo scope for homebrew-basic-memory repo
+          COMMITTER_TOKEN: ${{ secrets.HOMEBREW_TOKEN }}

CHANGELOG.md

@@ -1,5 +1,79 @@
 # CHANGELOG
 
+## v0.13.6 (2025-06-18)
+
+### Bug Fixes
+
+- **Custom Entity Types** - Support for custom entity types in write_note
+  ([`7789864`](https://github.com/basicmachines-co/basic-memory/commit/77898644933589c2da9bdd60571d54137a5309ed))
+  - Fixed `entity_type` parameter for `write_note` MCP tool to respect value passed in
+  - Frontmatter `type` field automatically respected when no explicit parameter provided
+  - Maintains backward compatibility with default "note" type
+
+- **#139**: Fix "UNIQUE constraint failed: entity.permalink" database error
+  ([`c6215fd`](https://github.com/basicmachines-co/basic-memory/commit/c6215fd819f9564ead91cf3a950f855241446096))
+  - Implement SQLAlchemy UPSERT strategy to handle permalink conflicts gracefully
+  - Eliminates crashes when creating notes with existing titles in same folders
+  - Seamlessly updates existing entities instead of failing with constraint errors
+
+- **Database Migration Performance** - Eliminate redundant migration initialization
+  ([`84d2aaf`](https://github.com/basicmachines-co/basic-memory/commit/84d2aaf6414dd083af4b0df73f6c8139b63468f6))
+  - Fix duplicate migration calls that slowed system startup
+  - Improve performance with multiple projects (tested with 28+ projects)
+  - Add migration deduplication safeguards with comprehensive test coverage
+
+- **User Experience** - Correct spelling error in continue_conversation prompt
+  ([`b4c26a6`](https://github.com/basicmachines-co/basic-memory/commit/b4c26a613379e6f2ba655efe3d7d8d40c27999e5))
+  - Fix "Chose a folder" → "Choose a folder" in MCP prompt instructions
+  - Improve grammar and clarity in user-facing prompt text
+
+### Documentation
+
+- **Website Updates** - Add new website and community links to README
+  ([`3fdce68`](https://github.com/basicmachines-co/basic-memory/commit/3fdce683d7ad8b6f4855d7138d5ff2136d4c07bc))
+
+- **Project Documentation** - Update README.md and CLAUDE.md with latest project information
+  ([`782cb2d`](https://github.com/basicmachines-co/basic-memory/commit/782cb2df28803482d209135a054e67cc32d7363e))
+
+### Technical Improvements
+
+- **Comprehensive Test Coverage** - Add extensive test suites for new features
+  - Custom entity type validation with 8 new test scenarios
+  - UPSERT behavior testing with edge case coverage
+  - Migration deduplication testing with 6 test scenarios
+  - Database constraint handling validation
+
+- **Code Quality** - Enhanced error handling and validation
+  - Improved SQLAlchemy patterns with modern UPSERT operations
+  - Better conflict resolution strategies for entity management
+  - Strengthened database consistency guarantees
+
+### Performance
+
+- **Database Operations** - Faster startup and improved scalability
+  - Reduced migration overhead for multi-project setups
+  - Optimized conflict resolution for entity creation
+  - Enhanced performance with growing knowledge bases
+
+### Migration Guide
+
+This release includes automatic database improvements. No manual migration required:
+
+- Existing notes and entity types continue working unchanged
+- New `entity_type` parameter is optional and backward compatible
+- Database performance improvements apply automatically
+- All existing MCP tool behavior preserved
+
+### Installation
+
+```bash
+# Latest stable release
+uv tool install basic-memory
+
+# Update existing installation
+uv tool upgrade basic-memory
+```
+
 ## v0.13.5 (2025-06-11)
 
 ### Bug Fixes

CLAUDE.md

@@ -97,15 +97,26 @@ See the [README.md](README.md) file for a project overview.
 
   **Content Management:**
     - `write_note(title, content, folder, tags)` - Create/update markdown notes with semantic observations and relations
-    - `read_note(identifier, page, page_size)` - Read notes by title, permalink, or memory:// URL with knowledge graph
-      awareness
-    - `read_file(path)` - Read raw file content (text, images, binaries) without knowledge graph processing
+    - `read_note(identifier, page, page_size)` - Read notes by title, permalink, or memory:// URL with knowledge graph awareness
+    - `edit_note(identifier, operation, content)` - Edit notes incrementally (append, prepend, find/replace, section replace)
+    - `move_note(identifier, destination_path)` - Move notes with database consistency and search reindexing
+    - `view_note(identifier)` - Display notes as formatted artifacts for better readability in Claude Desktop
+    - `read_content(path)` - Read raw file content (text, images, binaries) without knowledge graph processing
+    - `delete_note(identifier)` - Delete notes from knowledge base
+
+  **Project Management:**
+    - `list_memory_projects()` - List all available projects with status indicators
+    - `switch_project(project_name)` - Switch to different project context during conversations
+    - `get_current_project()` - Show currently active project with statistics
+    - `create_memory_project(name, path, set_default)` - Create new Basic Memory projects
+    - `delete_project(name)` - Delete projects from configuration and database
+    - `set_default_project(name)` - Set default project in config
+    - `sync_status()` - Check file synchronization status and background operations
 
   **Knowledge Graph Navigation:**
-    - `build_context(url, depth, timeframe)` - Navigate the knowledge graph via memory:// URLs for conversation
-      continuity
-    - `recent_activity(type, depth, timeframe)` - Get recently updated information with specified timeframe (e.g., "
-      1d", "1 week")
+    - `build_context(url, depth, timeframe)` - Navigate the knowledge graph via memory:// URLs for conversation continuity
+    - `recent_activity(type, depth, timeframe)` - Get recently updated information with specified timeframe (e.g., "1d", "1 week")
+    - `list_directory(dir_name, depth, file_name_glob)` - List directory contents with filtering and depth control
 
   **Search & Discovery:**
     - `search_notes(query, page, page_size)` - Full-text search across all content with filtering options
@@ -212,13 +223,33 @@ Basic Memory uses `uv-dynamic-versioning` for automatic version management based
 - Users install with: `pip install basic-memory --pre`
 - Use for milestone testing before stable release
 
-#### Stable Releases (Manual)
+#### Stable Releases (Automated)
+- Use the automated release system: `just release v0.13.0`
+- Includes comprehensive quality checks (lint, format, type-check, tests)
+- Automatically updates version in `__init__.py`
+- Creates git tag and pushes to GitHub
+- Triggers GitHub Actions workflow for:
+  - PyPI publication
+  - Homebrew formula update (requires HOMEBREW_TOKEN secret)
+
+**Manual method (legacy):**
 - Create version tag: `git tag v0.13.0 && git push origin v0.13.0`
-- Automatically builds, creates GitHub release, and publishes to PyPI
-- Users install with: `pip install basic-memory`
+
+#### Homebrew Formula Updates
+- Automatically triggered after successful PyPI release
+- Updates formula in `basicmachines-co/homebrew-basic-memory` repo
+- Requires `HOMEBREW_TOKEN` secret in GitHub repository settings:
+  - Create a fine-grained Personal Access Token with `Contents: Read and Write` and `Actions: Read` scopes on `basicmachines-co/homebrew-basic-memory`
+  - Add as repository secret named `HOMEBREW_TOKEN` in `basicmachines-co/basic-memory`
+- Formula updates include new version URL and SHA256 checksum
 
 ### For Development
-- No manual version bumping required
-- Versions automatically derived from git tags
-- `pyproject.toml` uses `dynamic = ["version"]`
-- `__init__.py` dynamically reads version from package metadata
+- **Automated releases**: Use `just release v0.13.x` for stable releases and `just beta v0.13.0b1` for beta releases
+- **Quality gates**: All releases require passing lint, format, type-check, and test suites
+- **Version management**: Versions automatically derived from git tags via `uv-dynamic-versioning`
+- **Configuration**: `pyproject.toml` uses `dynamic = ["version"]`
+- **Release automation**: `__init__.py` updated automatically during release process
+- **CI/CD**: GitHub Actions handles building and PyPI publication
+
+## Development Notes
+- make sure you sign off on commits

README.md

@@ -13,8 +13,11 @@ Basic Memory lets you build persistent knowledge through natural conversations w
 Claude, while keeping everything in simple Markdown files on your computer. It uses the Model Context Protocol (MCP) to
 enable any compatible LLM to read and write to your local knowledge base.
 
-- Website: https://basicmachines.co
+- Website: https://basicmemory.com
+- Company: https://basicmachines.co
 - Documentation: https://memory.basicmachines.co
+- Discord: https://discord.gg/tyvKNccgqN
+- YouTube: https://www.youtube.com/@basicmachines-co
 
 ## Pick up your conversation right where you left off
 
@@ -61,8 +64,7 @@ Memory for Claude Desktop:
 npx -y @smithery/cli install @basicmachines-co/basic-memory --client claude
 ```
 
-This installs and configures Basic Memory without requiring manual edits to the Claude Desktop configuration file. The
-Smithery server hosts the MCP server component, while your data remains stored locally as Markdown files.
+This installs and configures Basic Memory without requiring manual edits to the Claude Desktop configuration file. Note: The Smithery installation uses their hosted MCP server, while your data remains stored locally as Markdown files.
 
 ### Glama.ai
 
@@ -153,7 +155,8 @@ The note embeds semantic content and links to other topics via simple Markdown f
 
 3. You see this file on your computer in real time in the current project directory (default `~/$HOME/basic-memory`).
 
-- Realtime sync is enabled by default with the v0.12.0 version
+- Realtime sync is enabled by default starting with v0.12.0
+- Project switching during conversations is supported starting with v0.13.0
 
 4. In a chat with the LLM, you can reference a topic:
 
@@ -351,10 +354,20 @@ Basic Memory will sync the files in your project in real time if you make manual
 ```
 write_note(title, content, folder, tags) - Create or update notes
 read_note(identifier, page, page_size) - Read notes by title or permalink
+edit_note(identifier, operation, content) - Edit notes incrementally (append, prepend, find/replace)
+move_note(identifier, destination_path) - Move notes with database consistency
+view_note(identifier) - Display notes as formatted artifacts for better readability
 build_context(url, depth, timeframe) - Navigate knowledge graph via memory:// URLs
 search_notes(query, page, page_size) - Search across your knowledge base
 recent_activity(type, depth, timeframe) - Find recently updated information
 canvas(nodes, edges, title, folder) - Generate knowledge visualizations
+list_memory_projects() - List all available projects with status
+switch_project(project_name) - Switch to different project context
+get_current_project() - Show current project and statistics
+create_memory_project(name, path, set_default) - Create new projects
+delete_project(name) - Delete projects from configuration
+set_default_project(name) - Set default project
+sync_status() - Check file synchronization status
 ```
 
 5. Example prompts to try:
@@ -365,6 +378,10 @@ canvas(nodes, edges, title, folder) - Generate knowledge visualizations
 "Create a canvas visualization of my project components"
 "Read my notes on the authentication system"
 "What have I been working on in the past week?"
+"Switch to my work-notes project"
+"List all my available projects"
+"Edit my coffee brewing note to add a new technique"
+"Move my old meeting notes to the archive folder"
 ```
 
 ## Futher info

src/basic_memory/__init__.py

@@ -1,7 +1,7 @@
 """basic-memory - Local-first knowledge management combining Zettelkasten with knowledge graphs"""
 
 # Package version - updated by release automation
-__version__ = "0.13.5"
+__version__ = "0.13.6"
 
 # API version for FastAPI - independent of package version
 __api_version__ = "v0"

src/basic_memory/alembic/env.py

@@ -8,12 +8,12 @@ from sqlalchemy import pool
 
 from alembic import context
 
-from basic_memory.models import Base
-
 # set config.env to "test" for pytest to prevent logging to file in utils.setup_logging()
 os.environ["BASIC_MEMORY_ENV"] = "test"
 
-from basic_memory.config import app_config
+# Import after setting environment variable  # noqa: E402
+from basic_memory.config import app_config  # noqa: E402
+from basic_memory.models import Base  # noqa: E402
 
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.

src/basic_memory/db.py

@@ -23,6 +23,7 @@ from basic_memory.repository.search_repository import SearchRepository
 # Module level state
 _engine: Optional[AsyncEngine] = None
 _session_maker: Optional[async_sessionmaker[AsyncSession]] = None
+_migrations_completed: bool = False
 
 
 class DatabaseType(Enum):
@@ -72,18 +73,35 @@ async def scoped_session(
         await factory.remove()
 
 
+def _create_engine_and_session(
+    db_path: Path, db_type: DatabaseType = DatabaseType.FILESYSTEM
+) -> tuple[AsyncEngine, async_sessionmaker[AsyncSession]]:
+    """Internal helper to create engine and session maker."""
+    db_url = DatabaseType.get_db_url(db_path, db_type)
+    logger.debug(f"Creating engine for db_url: {db_url}")
+    engine = create_async_engine(db_url, connect_args={"check_same_thread": False})
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+    return engine, session_maker
+
+
 async def get_or_create_db(
     db_path: Path,
     db_type: DatabaseType = DatabaseType.FILESYSTEM,
+    ensure_migrations: bool = True,
+    app_config: Optional["BasicMemoryConfig"] = None,
 ) -> tuple[AsyncEngine, async_sessionmaker[AsyncSession]]:  # pragma: no cover
     """Get or create database engine and session maker."""
     global _engine, _session_maker
 
     if _engine is None:
-        db_url = DatabaseType.get_db_url(db_path, db_type)
-        logger.debug(f"Creating engine for db_url: {db_url}")
-        _engine = create_async_engine(db_url, connect_args={"check_same_thread": False})
-        _session_maker = async_sessionmaker(_engine, expire_on_commit=False)
+        _engine, _session_maker = _create_engine_and_session(db_path, db_type)
+        
+        # Run migrations automatically unless explicitly disabled
+        if ensure_migrations:
+            if app_config is None:
+                from basic_memory.config import app_config as global_app_config
+                app_config = global_app_config
+            await run_migrations(app_config, db_type)
 
     # These checks should never fail since we just created the engine and session maker
     # if they were None, but we'll check anyway for the type checker
@@ -100,12 +118,13 @@ async def get_or_create_db(
 
 async def shutdown_db() -> None:  # pragma: no cover
     """Clean up database connections."""
-    global _engine, _session_maker
+    global _engine, _session_maker, _migrations_completed
 
     if _engine:
         await _engine.dispose()
         _engine = None
         _session_maker = None
+        _migrations_completed = False
 
 
 @asynccontextmanager
@@ -119,7 +138,7 @@ async def engine_session_factory(
     for each test. For production use, use get_or_create_db() instead.
     """
 
-    global _engine, _session_maker
+    global _engine, _session_maker, _migrations_completed
 
     db_url = DatabaseType.get_db_url(db_path, db_type)
     logger.debug(f"Creating engine for db_url: {db_url}")
@@ -143,12 +162,20 @@ async def engine_session_factory(
             await _engine.dispose()
             _engine = None
             _session_maker = None
+            _migrations_completed = False
 
 
 async def run_migrations(
-    app_config: BasicMemoryConfig, database_type=DatabaseType.FILESYSTEM
+    app_config: BasicMemoryConfig, database_type=DatabaseType.FILESYSTEM, force: bool = False
 ):  # pragma: no cover
     """Run any pending alembic migrations."""
+    global _migrations_completed
+    
+    # Skip if migrations already completed unless forced
+    if _migrations_completed and not force:
+        logger.debug("Migrations already completed in this session, skipping")
+        return
+        
     logger.info("Running database migrations...")
     try:
         # Get the absolute path to the alembic directory relative to this file
@@ -170,11 +197,18 @@ async def run_migrations(
         command.upgrade(config, "head")
         logger.info("Migrations completed successfully")
 
-        _, session_maker = await get_or_create_db(app_config.database_path, database_type)
+        # Get session maker - ensure we don't trigger recursive migration calls
+        if _session_maker is None:
+            _, session_maker = _create_engine_and_session(app_config.database_path, database_type)
+        else:
+            session_maker = _session_maker
 
         # initialize the search Index schema
         # the project_id is not used for init_search_index, so we pass a dummy value
         await SearchRepository(session_maker, 1).init_search_index()
+        
+        # Mark migrations as completed
+        _migrations_completed = True
     except Exception as e:  # pragma: no cover
         logger.error(f"Error running migrations: {e}")
         raise

src/basic_memory/mcp/tools/write_note.py

@@ -27,6 +27,7 @@ async def write_note(
     content: str,
     folder: str,
     tags=None,  # Remove type hint completely to avoid schema issues
+    entity_type: str = "note",
     project: Optional[str] = None,
 ) -> str:
     """Write a markdown note to the knowledge base.
@@ -58,6 +59,7 @@ async def write_note(
                 Use forward slashes (/) as separators. Examples: "notes", "projects/2025", "research/ml"
         tags: Tags to categorize the note. Can be a list of strings, a comma-separated string, or None.
               Note: If passing from external MCP clients, use a string format (e.g. "tag1,tag2,tag3")
+        entity_type: Type of entity to create. Defaults to "note". Can be "guide", "report", "config", etc.
         project: Optional project name to write to. If not provided, uses current active project.
 
     Returns:
@@ -84,7 +86,7 @@ async def write_note(
     entity = Entity(
         title=title,
         folder=folder,
-        entity_type="note",
+        entity_type=entity_type,
         content_type="text/markdown",
         content=content,
         entity_metadata=metadata,

src/basic_memory/repository/entity_repository.py

@@ -3,10 +3,13 @@
 from pathlib import Path
 from typing import List, Optional, Sequence, Union
 
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 from sqlalchemy.orm import selectinload
 from sqlalchemy.orm.interfaces import LoaderOption
 
+from basic_memory import db
 from basic_memory.models.knowledge import Entity, Observation, Relation
 from basic_memory.repository.repository import Repository
 
@@ -96,3 +99,153 @@ class EntityRepository(Repository[Entity]):
 
         result = await self.execute_query(query)
         return list(result.scalars().all())
+
+    async def upsert_entity(self, entity: Entity) -> Entity:
+        """Insert or update entity using a hybrid approach.
+        
+        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.
+        
+        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 = (
+                    select(Entity)
+                    .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 for new entity
+                session.add(entity)
+                await session.flush()
+                
+                # Return with relationships loaded
+                query = (
+                    select(Entity)
+                    .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
+                
+            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(
+                        Entity.file_path == entity.file_path,
+                        Entity.project_id == entity.project_id
+                    )
+                )
+                race_condition_entity = existing_by_path_check.scalar_one_or_none()
+                
+                if race_condition_entity:
+                    # Race condition: file_path conflict detected after our initial check
+                    # Update the existing entity instead
+                    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(race_condition_entity, key, value)
+                    
+                    await session.flush()
+                    # Return the updated entity with relationships loaded
+                    query = (
+                        select(Entity)
+                        .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
+                else:
+                    # Must be permalink conflict - generate unique permalink
+                    return await self._handle_permalink_conflict(entity, session)
+
+    async def _handle_permalink_conflict(self, entity: Entity, session: AsyncSession) -> Entity:
+        """Handle permalink conflicts by generating a unique permalink."""
+        base_permalink = entity.permalink
+        suffix = 1
+        
+        # Find a unique permalink
+        while True:
+            test_permalink = f"{base_permalink}-{suffix}"
+            existing = await session.execute(
+                select(Entity).where(
+                    Entity.permalink == test_permalink,
+                    Entity.project_id == entity.project_id
+                )
+            )
+            if existing.scalar_one_or_none() is None:
+                # Found unique permalink
+                entity.permalink = test_permalink
+                break
+            suffix += 1
+            
+        # Insert with unique permalink (no conflict possible now)
+        session.add(entity)
+        await session.flush()
+        
+        # Return the inserted entity with relationships loaded
+        query = (
+            select(Entity)
+            .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

src/basic_memory/services/entity_service.py

@@ -117,10 +117,15 @@ class EntityService(BaseService[EntityModel]):
                 f"file for entity {schema.folder}/{schema.title} already exists: {file_path}"
             )
 
-        # Parse content frontmatter to check for user-specified permalink
+        # Parse content frontmatter to check for user-specified permalink and entity_type
         content_markdown = None
         if schema.content and has_frontmatter(schema.content):
             content_frontmatter = parse_frontmatter(schema.content)
+
+            # If content has entity_type/type, use it to override the schema entity_type
+            if "type" in content_frontmatter:
+                schema.entity_type = content_frontmatter["type"]
+
             if "permalink" in content_frontmatter:
                 # Create a minimal EntityMarkdown object for permalink resolution
                 from basic_memory.markdown.schemas import EntityFrontmatter
@@ -172,10 +177,15 @@ class EntityService(BaseService[EntityModel]):
         # Read existing frontmatter from the file if it exists
         existing_markdown = await self.entity_parser.parse_file(file_path)
 
-        # Parse content frontmatter to check for user-specified permalink
+        # Parse content frontmatter to check for user-specified permalink and entity_type
         content_markdown = None
         if schema.content and has_frontmatter(schema.content):
             content_frontmatter = parse_frontmatter(schema.content)
+
+            # If content has entity_type/type, use it to override the schema entity_type
+            if "type" in content_frontmatter:
+                schema.entity_type = content_frontmatter["type"]
+
             if "permalink" in content_frontmatter:
                 # Create a minimal EntityMarkdown object for permalink resolution
                 from basic_memory.markdown.schemas import EntityFrontmatter
@@ -292,27 +302,21 @@ class EntityService(BaseService[EntityModel]):
 
         Creates the entity with null checksum to indicate sync not complete.
         Relations will be added in second pass.
+        
+        Uses UPSERT approach to handle permalink/file_path conflicts cleanly.
         """
         logger.debug(f"Creating entity: {markdown.frontmatter.title} file_path: {file_path}")
         model = entity_model_from_markdown(file_path, markdown)
 
         # Mark as incomplete because we still need to add relations
         model.checksum = None
-        # Repository will set project_id automatically
+        
+        # Use UPSERT to handle conflicts cleanly
         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
+            return await self.repository.upsert_entity(model)
+        except Exception as e:
+            logger.error(f"Failed to upsert entity for {file_path}: {e}")
+            raise EntityCreationError(f"Failed to create entity: {str(e)}") from e
 
     async def update_entity_and_observations(
         self, file_path: Path, markdown: EntityMarkdown

src/basic_memory/services/initialization.py

@@ -17,17 +17,21 @@ from basic_memory.repository import ProjectRepository
 
 
 async def initialize_database(app_config: BasicMemoryConfig) -> None:
-    """Run database migrations to ensure schema is up to date.
+    """Initialize database with migrations handled automatically by get_or_create_db.
 
     Args:
         app_config: The Basic Memory project configuration
+        
+    Note:
+        Database migrations are now handled automatically when the database 
+        connection is first established via get_or_create_db().
     """
+    # Trigger database initialization and migrations by getting the database connection
     try:
-        logger.info("Running database migrations...")
-        await db.run_migrations(app_config)
-        logger.info("Migrations completed successfully")
+        await db.get_or_create_db(app_config.database_path)
+        logger.info("Database initialization completed")
     except Exception as e:
-        logger.error(f"Error running migrations: {e}")
+        logger.error(f"Error initializing database: {e}")
         # Allow application to continue - it might still work
         # depending on what the error was, and will fail with a
         # more specific error if the database is actually unusable
@@ -44,9 +48,9 @@ async def reconcile_projects_with_config(app_config: BasicMemoryConfig):
     """
     logger.info("Reconciling projects from config with database...")
 
-    # Get database session
+    # Get database session - migrations handled centrally
     _, session_maker = await db.get_or_create_db(
-        db_path=app_config.database_path, db_type=db.DatabaseType.FILESYSTEM
+        db_path=app_config.database_path, db_type=db.DatabaseType.FILESYSTEM, ensure_migrations=False
     )
     project_repository = ProjectRepository(session_maker)
 
@@ -65,9 +69,9 @@ async def reconcile_projects_with_config(app_config: BasicMemoryConfig):
 
 
 async def migrate_legacy_projects(app_config: BasicMemoryConfig):
-    # Get database session
+    # Get database session - migrations handled centrally
     _, session_maker = await db.get_or_create_db(
-        db_path=app_config.database_path, db_type=db.DatabaseType.FILESYSTEM
+        db_path=app_config.database_path, db_type=db.DatabaseType.FILESYSTEM, ensure_migrations=False
     )
     logger.info("Migrating legacy projects...")
     project_repository = ProjectRepository(session_maker)
@@ -134,9 +138,9 @@ async def initialize_file_sync(
     # delay import
     from basic_memory.sync import WatchService
 
-    # Load app configuration
+    # Load app configuration - migrations handled centrally
     _, session_maker = await db.get_or_create_db(
-        db_path=app_config.database_path, db_type=db.DatabaseType.FILESYSTEM
+        db_path=app_config.database_path, db_type=db.DatabaseType.FILESYSTEM, ensure_migrations=False
     )
     project_repository = ProjectRepository(session_maker)
 

src/basic_memory/templates/prompts/continue_conversation.hbs

@@ -96,7 +96,7 @@ You can also:
 You can:
 - Explore more with: `search_notes("{{ topic }}")`
 - See what's changed: `recent_activity(timeframe="{{default timeframe "7d"}}")`
-- **Record new learnings or decisions from this conversation:** `write_note(folder="[Chose a folder]" title="[Create a meaningful title]", content="[Content with observations and relations]")`
+- **Record new learnings or decisions from this conversation:** `write_note(folder="[Choose a folder]" title="[Create a meaningful title]", content="[Content with observations and relations]")`
 
 ## Knowledge Capture Recommendation
 

tests/mcp/test_tool_write_note.py

@@ -63,7 +63,7 @@ async def test_write_note_no_tags(app):
     content = await read_note.fn("test/simple-note")
     assert (
         dedent("""
-        --
+        ---
         title: Simple Note
         type: note
         permalink: test/simple-note
@@ -413,3 +413,248 @@ async def test_write_note_preserves_content_frontmatter(app):
         ).strip()
         in content
     )
+
+
+@pytest.mark.asyncio
+async def test_write_note_permalink_collision_fix_issue_139(app):
+    """Test fix for GitHub Issue #139: UNIQUE constraint failed: entity.permalink.
+    
+    This reproduces the exact scenario described in the issue:
+    1. Create a note with title "Note 1" 
+    2. Create another note with title "Note 2"
+    3. Try to create/replace first note again with same title "Note 1"
+    
+    Before the fix, step 3 would fail with UNIQUE constraint error.
+    After the fix, it should either update the existing note or create with unique permalink.
+    """
+    # Step 1: Create first note
+    result1 = await write_note.fn(
+        title="Note 1",
+        folder="test",
+        content="Original content for note 1"
+    )
+    assert "# Created note" in result1
+    assert "permalink: test/note-1" in result1
+    
+    # Step 2: Create second note with different title
+    result2 = await write_note.fn(
+        title="Note 2", 
+        folder="test",
+        content="Content for note 2"
+    )
+    assert "# Created note" in result2
+    assert "permalink: test/note-2" in result2
+    
+    # Step 3: Try to create/replace first note again
+    # This scenario would trigger the UNIQUE constraint failure before the fix
+    result3 = await write_note.fn(
+        title="Note 1",  # Same title as first note
+        folder="test",   # Same folder as first note
+        content="Replacement content for note 1"  # Different content
+    )
+    
+    # This should not raise a UNIQUE constraint failure error
+    # It should succeed and either:
+    # 1. Update the existing note (preferred behavior)
+    # 2. Create a new note with unique permalink (fallback behavior)
+    
+    assert result3 is not None
+    assert ("Updated note" in result3 or "Created note" in result3)
+    
+    # The result should contain either the original permalink or a unique one
+    assert ("permalink: test/note-1" in result3 or "permalink: test/note-1-1" in result3)
+    
+    # Verify we can read back the content
+    if "permalink: test/note-1" in result3:
+        # Updated existing note case
+        content = await read_note.fn("test/note-1")
+        assert "Replacement content for note 1" in content
+    else:
+        # Created new note with unique permalink case  
+        content = await read_note.fn("test/note-1-1")
+        assert "Replacement content for note 1" in content
+        # Original note should still exist
+        original_content = await read_note.fn("test/note-1")
+        assert "Original content for note 1" in original_content
+
+
+@pytest.mark.asyncio
+async def test_write_note_with_custom_entity_type(app):
+    """Test creating a note with custom entity_type parameter.
+
+    This test verifies the fix for Issue #144 where entity_type parameter
+    was hardcoded to "note" instead of allowing custom types.
+    """
+    result = await write_note.fn(
+        title="Test Guide",
+        folder="guides",
+        content="# Guide Content\nThis is a guide",
+        tags=["guide", "documentation"],
+        entity_type="guide",
+    )
+
+    assert result
+    assert "# Created note" in result
+    assert "file_path: guides/Test Guide.md" in result
+    assert "permalink: guides/test-guide" in result
+    assert "## Tags" in result
+    assert "- guide, documentation" in result
+
+    # Verify the entity type is correctly set in the frontmatter
+    content = await read_note.fn("guides/test-guide")
+    assert (
+        dedent("""
+        ---
+        title: Test Guide
+        type: guide
+        permalink: guides/test-guide
+        tags:
+        - guide
+        - documentation
+        ---
+        
+        # Guide Content
+        This is a guide
+        """).strip()
+        in content
+    )
+
+
+@pytest.mark.asyncio
+async def test_write_note_with_report_entity_type(app):
+    """Test creating a note with entity_type="report"."""
+    result = await write_note.fn(
+        title="Monthly Report",
+        folder="reports",
+        content="# Monthly Report\nThis is a monthly report",
+        tags=["report", "monthly"],
+        entity_type="report",
+    )
+
+    assert result
+    assert "# Created note" in result
+    assert "file_path: reports/Monthly Report.md" in result
+    assert "permalink: reports/monthly-report" in result
+
+    # Verify the entity type is correctly set in the frontmatter
+    content = await read_note.fn("reports/monthly-report")
+    assert "type: report" in content
+    assert "# Monthly Report" in content
+
+
+@pytest.mark.asyncio
+async def test_write_note_with_config_entity_type(app):
+    """Test creating a note with entity_type="config"."""
+    result = await write_note.fn(
+        title="System Config",
+        folder="config",
+        content="# System Configuration\nThis is a config file",
+        entity_type="config",
+    )
+
+    assert result
+    assert "# Created note" in result
+    assert "file_path: config/System Config.md" in result
+    assert "permalink: config/system-config" in result
+
+    # Verify the entity type is correctly set in the frontmatter
+    content = await read_note.fn("config/system-config")
+    assert "type: config" in content
+    assert "# System Configuration" in content
+
+
+@pytest.mark.asyncio
+async def test_write_note_entity_type_default_behavior(app):
+    """Test that the entity_type parameter defaults to "note" when not specified.
+
+    This ensures backward compatibility - existing code that doesn't specify
+    entity_type should continue to work as before.
+    """
+    result = await write_note.fn(
+        title="Default Type Test",
+        folder="test",
+        content="# Default Type Test\nThis should be type 'note'",
+        tags=["test"],
+    )
+
+    assert result
+    assert "# Created note" in result
+    assert "file_path: test/Default Type Test.md" in result
+    assert "permalink: test/default-type-test" in result
+
+    # Verify the entity type defaults to "note"
+    content = await read_note.fn("test/default-type-test")
+    assert "type: note" in content
+    assert "# Default Type Test" in content
+
+
+@pytest.mark.asyncio
+async def test_write_note_update_existing_with_different_entity_type(app):
+    """Test updating an existing note with a different entity_type."""
+    # Create initial note as "note" type
+    result1 = await write_note.fn(
+        title="Changeable Type",
+        folder="test",
+        content="# Initial Content\nThis starts as a note",
+        tags=["test"],
+        entity_type="note",
+    )
+
+    assert result1
+    assert "# Created note" in result1
+
+    # Update the same note with a different entity_type
+    result2 = await write_note.fn(
+        title="Changeable Type",
+        folder="test",
+        content="# Updated Content\nThis is now a guide",
+        tags=["guide"],
+        entity_type="guide",
+    )
+
+    assert result2
+    assert "# Updated note" in result2
+
+    # Verify the entity type was updated
+    content = await read_note.fn("test/changeable-type")
+    assert "type: guide" in content
+    assert "# Updated Content" in content
+    assert "- guide" in content
+
+
+@pytest.mark.asyncio
+async def test_write_note_respects_frontmatter_entity_type(app):
+    """Test that entity_type in frontmatter is respected when parameter is not provided.
+
+    This verifies that when write_note is called without entity_type parameter,
+    but the content includes frontmatter with a 'type' field, that type is respected
+    instead of defaulting to 'note'.
+    """
+    note = dedent("""
+        ---
+        title: Test Guide
+        type: guide
+        permalink: guides/test-guide
+        tags:
+        - guide
+        - documentation
+        ---
+        
+        # Guide Content
+        This is a guide
+        """).strip()
+
+    # Call write_note without entity_type parameter - it should respect frontmatter type
+    result = await write_note.fn(title="Test Guide", folder="guides", content=note)
+
+    assert result
+    assert "# Created note" in result
+    assert "file_path: guides/Test Guide.md" in result
+    assert "permalink: guides/test-guide" in result
+
+    # Verify the entity type from frontmatter is respected (should be "guide", not "note")
+    content = await read_note.fn("guides/test-guide")
+    assert "type: guide" in content
+    assert "# Guide Content" in content
+    assert "- guide" in content
+    assert "- documentation" in content

tests/repository/test_entity_repository_upsert.py

@@ -0,0 +1,248 @@
+"""Tests for the entity repository UPSERT functionality."""
+
+import pytest
+from datetime import datetime, timezone
+
+from basic_memory.models.knowledge import Entity
+from basic_memory.repository.entity_repository import EntityRepository
+
+
+@pytest.mark.asyncio
+async def test_upsert_entity_new_entity(entity_repository: EntityRepository):
+    """Test upserting a completely new entity."""
+    entity = Entity(
+        project_id=entity_repository.project_id,
+        title="Test Entity",
+        entity_type="note",
+        permalink="test/test-entity",
+        file_path="test/test-entity.md",
+        content_type="text/markdown",
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+    result = await entity_repository.upsert_entity(entity)
+    
+    assert result.id is not None
+    assert result.title == "Test Entity"
+    assert result.permalink == "test/test-entity"
+    assert result.file_path == "test/test-entity.md"
+
+
+@pytest.mark.asyncio
+async def test_upsert_entity_same_file_update(entity_repository: EntityRepository):
+    """Test upserting an entity that already exists with same file_path."""
+    # Create initial entity
+    entity1 = Entity(
+        project_id=entity_repository.project_id,
+        title="Original Title",
+        entity_type="note",
+        permalink="test/test-entity",
+        file_path="test/test-entity.md",
+        content_type="text/markdown",
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+    result1 = await entity_repository.upsert_entity(entity1)
+    original_id = result1.id
+
+    # Update with same file_path and permalink
+    entity2 = Entity(
+        project_id=entity_repository.project_id,
+        title="Updated Title",
+        entity_type="note",
+        permalink="test/test-entity",  # Same permalink
+        file_path="test/test-entity.md",  # Same file_path
+        content_type="text/markdown",
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+    result2 = await entity_repository.upsert_entity(entity2)
+    
+    # Should update existing entity (same ID)
+    assert result2.id == original_id
+    assert result2.title == "Updated Title"
+    assert result2.permalink == "test/test-entity"
+    assert result2.file_path == "test/test-entity.md"
+
+
+@pytest.mark.asyncio
+async def test_upsert_entity_permalink_conflict_different_file(entity_repository: EntityRepository):
+    """Test upserting an entity with permalink conflict but different file_path."""
+    # Create initial entity
+    entity1 = Entity(
+        project_id=entity_repository.project_id,
+        title="First Entity",
+        entity_type="note",
+        permalink="test/shared-permalink",
+        file_path="test/first-file.md",
+        content_type="text/markdown",
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+    result1 = await entity_repository.upsert_entity(entity1)
+    first_id = result1.id
+
+    # Try to create entity with same permalink but different file_path
+    entity2 = Entity(
+        project_id=entity_repository.project_id,
+        title="Second Entity",
+        entity_type="note",
+        permalink="test/shared-permalink",  # Same permalink
+        file_path="test/second-file.md",   # Different file_path
+        content_type="text/markdown",
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+    result2 = await entity_repository.upsert_entity(entity2)
+    
+    # Should create new entity with unique permalink
+    assert result2.id != first_id
+    assert result2.title == "Second Entity"
+    assert result2.permalink == "test/shared-permalink-1"  # Should get suffix
+    assert result2.file_path == "test/second-file.md"
+    
+    # Original entity should be unchanged
+    original = await entity_repository.get_by_permalink("test/shared-permalink")
+    assert original is not None
+    assert original.id == first_id
+    assert original.title == "First Entity"
+
+
+@pytest.mark.asyncio
+async def test_upsert_entity_multiple_permalink_conflicts(entity_repository: EntityRepository):
+    """Test upserting multiple entities with permalink conflicts."""
+    base_permalink = "test/conflict"
+    
+    # Create entities with conflicting permalinks
+    entities = []
+    for i in range(3):
+        entity = Entity(
+            project_id=entity_repository.project_id,
+            title=f"Entity {i+1}",
+            entity_type="note",
+            permalink=base_permalink,  # All try to use same permalink
+            file_path=f"test/file-{i+1}.md",  # Different file paths
+            content_type="text/markdown",
+            created_at=datetime.now(timezone.utc),
+            updated_at=datetime.now(timezone.utc),
+        )
+        
+        result = await entity_repository.upsert_entity(entity)
+        entities.append(result)
+    
+    # Verify permalinks are unique
+    expected_permalinks = ["test/conflict", "test/conflict-1", "test/conflict-2"]
+    actual_permalinks = [entity.permalink for entity in entities]
+    
+    assert set(actual_permalinks) == set(expected_permalinks)
+    
+    # Verify all entities were created (different IDs)
+    entity_ids = [entity.id for entity in entities]
+    assert len(set(entity_ids)) == 3
+
+
+@pytest.mark.asyncio
+async def test_upsert_entity_race_condition_file_path(entity_repository: EntityRepository):
+    """Test that upsert handles race condition where file_path conflict occurs after initial check."""
+    from unittest.mock import patch
+    from sqlalchemy.exc import IntegrityError
+    
+    # Create an entity first
+    entity1 = Entity(
+        project_id=entity_repository.project_id,
+        title="Original Entity",
+        entity_type="note",
+        permalink="test/original",
+        file_path="test/race-file.md",
+        content_type="text/markdown",
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+    
+    result1 = await entity_repository.upsert_entity(entity1)
+    original_id = result1.id
+    
+    # Create another entity with different file_path and permalink
+    entity2 = Entity(
+        project_id=entity_repository.project_id,
+        title="Race Condition Test",
+        entity_type="note", 
+        permalink="test/race-entity",
+        file_path="test/different-file.md",  # Different initially
+        content_type="text/markdown",
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+    
+    # Now simulate race condition: change file_path to conflict after the initial check
+    original_add = entity_repository.session_maker().add
+    call_count = 0
+    
+    def mock_add(obj):
+        nonlocal call_count
+        if isinstance(obj, Entity) and call_count == 0:
+            call_count += 1
+            # Simulate race condition by changing file_path to conflict
+            obj.file_path = "test/race-file.md"  # Same as entity1
+            # This should trigger IntegrityError for file_path constraint
+            raise IntegrityError("UNIQUE constraint failed: entity.file_path", None, None)
+        return original_add(obj)
+    
+    # Mock session.add to simulate the race condition
+    with patch.object(entity_repository.session_maker().__class__, 'add', side_effect=mock_add):
+        # This should handle the race condition gracefully by updating the existing entity
+        result2 = await entity_repository.upsert_entity(entity2)
+        
+        # Should return the updated original entity (same ID)
+        assert result2.id == original_id
+        assert result2.title == "Race Condition Test"  # Updated title
+        assert result2.file_path == "test/race-file.md"  # Same file path
+        assert result2.permalink == "test/race-entity"  # Updated permalink
+
+
+@pytest.mark.asyncio
+async def test_upsert_entity_gap_in_suffixes(entity_repository: EntityRepository):
+    """Test that upsert finds the next available suffix even with gaps."""
+    # Manually create entities with non-sequential suffixes
+    base_permalink = "test/gap"
+    
+    # Create entities with permalinks: "test/gap", "test/gap-1", "test/gap-3"
+    # (skipping "test/gap-2")
+    permalinks = [base_permalink, f"{base_permalink}-1", f"{base_permalink}-3"]
+    
+    for i, permalink in enumerate(permalinks):
+        entity = Entity(
+            project_id=entity_repository.project_id,
+            title=f"Entity {i+1}",
+            entity_type="note",
+            permalink=permalink,
+            file_path=f"test/gap-file-{i+1}.md",
+            content_type="text/markdown",
+            created_at=datetime.now(timezone.utc),
+            updated_at=datetime.now(timezone.utc),
+        )
+        await entity_repository.add(entity)  # Use direct add to set specific permalinks
+    
+    # Now try to upsert a new entity that should get "test/gap-2"
+    new_entity = Entity(
+        project_id=entity_repository.project_id,
+        title="Gap Filler",
+        entity_type="note",
+        permalink=base_permalink,  # Will conflict
+        file_path="test/gap-new-file.md",
+        content_type="text/markdown",
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+    
+    result = await entity_repository.upsert_entity(new_entity)
+    
+    # Should get the next available suffix - our implementation finds gaps
+    # so it should be "test/gap-2" (filling the gap)
+    assert result.permalink == "test/gap-2"
+    assert result.title == "Gap Filler"

tests/services/test_entity_service.py

@@ -870,14 +870,11 @@ async def test_edit_entity_with_observations_and_relations(
 
 
 @pytest.mark.asyncio
-async def test_create_entity_from_markdown_race_condition_handling(
+async def test_create_entity_from_markdown_with_upsert(
     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
-    from sqlalchemy.exc import IntegrityError
-
-    file_path = Path("test/race-condition.md")
+    """Test that create_entity_from_markdown uses UPSERT approach for conflict resolution."""
+    file_path = Path("test/upsert-test.md")
 
     # Create a mock EntityMarkdown object
     from basic_memory.markdown.schemas import (
@@ -886,7 +883,7 @@ async def test_create_entity_from_markdown_race_condition_handling(
     )
     from datetime import datetime, timezone
 
-    frontmatter = EntityFrontmatter(metadata={"title": "Race Condition Test", "type": "test"})
+    frontmatter = EntityFrontmatter(metadata={"title": "UPSERT Test", "type": "test"})
     markdown = RealEntityMarkdown(
         frontmatter=frontmatter,
         observations=[],
@@ -895,63 +892,26 @@ async def test_create_entity_from_markdown_race_condition_handling(
         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
+    # Call the method - should succeed without complex exception handling
+    result = await entity_service.create_entity_from_markdown(file_path, markdown)
 
-    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)
+    # Verify it created the entity successfully using the UPSERT approach
+    assert result is not None
+    assert result.title == "UPSERT Test"
+    assert result.file_path == str(file_path)
+    # create_entity_from_markdown sets checksum to None (incomplete sync)
+    assert result.checksum is None
 
 
 @pytest.mark.asyncio
-async def test_create_entity_from_markdown_integrity_error_reraise(
+async def test_create_entity_from_markdown_error_handling(
     entity_service: EntityService, file_service: FileService
 ):
-    """Test that create_entity_from_markdown re-raises IntegrityError for non-race-condition cases."""
+    """Test that create_entity_from_markdown handles repository errors gracefully."""
     from unittest.mock import patch
-    from sqlalchemy.exc import IntegrityError
+    from basic_memory.services.exceptions import EntityCreationError
 
-    file_path = Path("test/integrity-error.md")
+    file_path = Path("test/error-test.md")
 
     # Create a mock EntityMarkdown object
     from basic_memory.markdown.schemas import (
@@ -960,7 +920,7 @@ async def test_create_entity_from_markdown_integrity_error_reraise(
     )
     from datetime import datetime, timezone
 
-    frontmatter = EntityFrontmatter(metadata={"title": "Integrity Error Test", "type": "test"})
+    frontmatter = EntityFrontmatter(metadata={"title": "Error Test", "type": "test"})
     markdown = RealEntityMarkdown(
         frontmatter=frontmatter,
         observations=[],
@@ -969,16 +929,14 @@ async def test_create_entity_from_markdown_integrity_error_reraise(
         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)
+    # Mock the repository.upsert_entity to raise a general error
+    async def mock_upsert(*args, **kwargs):
+        # Simulate a general database error
+        raise Exception("Database connection failed")
 
-    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"
-        ):
+    with patch.object(entity_service.repository, "upsert_entity", side_effect=mock_upsert):
+        # Should wrap the error in EntityCreationError
+        with pytest.raises(EntityCreationError, match="Failed to create entity"):
             await entity_service.create_entity_from_markdown(file_path, markdown)
 
 

tests/services/test_initialization.py

@@ -17,20 +17,21 @@ from basic_memory.services.initialization import (
 
 
 @pytest.mark.asyncio
-@patch("basic_memory.services.initialization.db.run_migrations")
-async def test_initialize_database(mock_run_migrations, project_config):
+@patch("basic_memory.services.initialization.db.get_or_create_db")
+async def test_initialize_database(mock_get_or_create_db, app_config):
     """Test initializing the database."""
-    await initialize_database(project_config)
-    mock_run_migrations.assert_called_once_with(project_config)
+    mock_get_or_create_db.return_value = (MagicMock(), MagicMock())
+    await initialize_database(app_config)
+    mock_get_or_create_db.assert_called_once_with(app_config.database_path)
 
 
 @pytest.mark.asyncio
-@patch("basic_memory.services.initialization.db.run_migrations")
-async def test_initialize_database_error(mock_run_migrations, project_config):
+@patch("basic_memory.services.initialization.db.get_or_create_db")
+async def test_initialize_database_error(mock_get_or_create_db, app_config):
     """Test handling errors during database initialization."""
-    mock_run_migrations.side_effect = Exception("Test error")
-    await initialize_database(project_config)
-    mock_run_migrations.assert_called_once_with(project_config)
+    mock_get_or_create_db.side_effect = Exception("Test error")
+    await initialize_database(app_config)
+    mock_get_or_create_db.assert_called_once_with(app_config.database_path)
 
 
 @pytest.mark.asyncio

tests/test_db_migration_deduplication.py

@@ -0,0 +1,187 @@
+"""Tests for database migration deduplication functionality."""
+
+import pytest
+from unittest.mock import patch, AsyncMock, MagicMock
+
+from basic_memory import db
+
+
+@pytest.fixture
+def mock_alembic_config():
+    """Mock Alembic config to avoid actual migration runs."""
+    with patch("basic_memory.db.Config") as mock_config_class:
+        mock_config = MagicMock()
+        mock_config_class.return_value = mock_config
+        yield mock_config
+
+
+@pytest.fixture
+def mock_alembic_command():
+    """Mock Alembic command to avoid actual migration runs."""
+    with patch("basic_memory.db.command") as mock_command:
+        yield mock_command
+
+
+@pytest.fixture
+def mock_search_repository():
+    """Mock SearchRepository to avoid database dependencies."""
+    with patch("basic_memory.db.SearchRepository") as mock_repo_class:
+        mock_repo = AsyncMock()
+        mock_repo_class.return_value = mock_repo
+        yield mock_repo
+
+
+# Use the app_config fixture from conftest.py
+
+
+@pytest.mark.asyncio
+async def test_migration_deduplication_single_call(
+    app_config, mock_alembic_config, mock_alembic_command, mock_search_repository
+):
+    """Test that migrations are only run once when called multiple times."""
+    # Reset module state
+    db._migrations_completed = False
+    db._engine = None
+    db._session_maker = None
+
+    # First call should run migrations
+    await db.run_migrations(app_config)
+    
+    # Verify migrations were called
+    mock_alembic_command.upgrade.assert_called_once_with(mock_alembic_config, "head")
+    mock_search_repository.init_search_index.assert_called_once()
+    
+    # Reset mocks for second call
+    mock_alembic_command.reset_mock()
+    mock_search_repository.reset_mock()
+    
+    # Second call should skip migrations
+    await db.run_migrations(app_config)
+    
+    # Verify migrations were NOT called again
+    mock_alembic_command.upgrade.assert_not_called()
+    mock_search_repository.init_search_index.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_migration_force_parameter(
+    app_config, mock_alembic_config, mock_alembic_command, mock_search_repository
+):
+    """Test that migrations can be forced to run even if already completed."""
+    # Reset module state
+    db._migrations_completed = False
+    db._engine = None
+    db._session_maker = None
+
+    # First call should run migrations
+    await db.run_migrations(app_config)
+    
+    # Verify migrations were called
+    mock_alembic_command.upgrade.assert_called_once_with(mock_alembic_config, "head")
+    mock_search_repository.init_search_index.assert_called_once()
+    
+    # Reset mocks for forced call
+    mock_alembic_command.reset_mock()
+    mock_search_repository.reset_mock()
+    
+    # Forced call should run migrations again
+    await db.run_migrations(app_config, force=True)
+    
+    # Verify migrations were called again
+    mock_alembic_command.upgrade.assert_called_once_with(mock_alembic_config, "head")
+    mock_search_repository.init_search_index.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_migration_state_reset_on_shutdown():
+    """Test that migration state is reset when database is shut down."""
+    # Set up completed state
+    db._migrations_completed = True
+    db._engine = AsyncMock()
+    db._session_maker = AsyncMock()
+    
+    # Shutdown should reset state
+    await db.shutdown_db()
+    
+    # Verify state was reset
+    assert db._migrations_completed is False
+    assert db._engine is None
+    assert db._session_maker is None
+
+
+@pytest.mark.asyncio
+async def test_get_or_create_db_runs_migrations_automatically(
+    app_config, mock_alembic_config, mock_alembic_command, mock_search_repository
+):
+    """Test that get_or_create_db runs migrations automatically."""
+    # Reset module state
+    db._migrations_completed = False
+    db._engine = None
+    db._session_maker = None
+
+    # First call should create engine and run migrations
+    engine, session_maker = await db.get_or_create_db(
+        app_config.database_path, app_config=app_config
+    )
+    
+    # Verify we got valid objects
+    assert engine is not None
+    assert session_maker is not None
+    
+    # Verify migrations were called
+    mock_alembic_command.upgrade.assert_called_once_with(mock_alembic_config, "head")
+    mock_search_repository.init_search_index.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_get_or_create_db_skips_migrations_when_disabled(
+    app_config, mock_alembic_config, mock_alembic_command, mock_search_repository
+):
+    """Test that get_or_create_db can skip migrations when ensure_migrations=False."""
+    # Reset module state
+    db._migrations_completed = False
+    db._engine = None
+    db._session_maker = None
+
+    # Call with ensure_migrations=False should skip migrations
+    engine, session_maker = await db.get_or_create_db(
+        app_config.database_path, ensure_migrations=False
+    )
+    
+    # Verify we got valid objects
+    assert engine is not None
+    assert session_maker is not None
+    
+    # Verify migrations were NOT called
+    mock_alembic_command.upgrade.assert_not_called()
+    mock_search_repository.init_search_index.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_multiple_get_or_create_db_calls_deduplicated(
+    app_config, mock_alembic_config, mock_alembic_command, mock_search_repository
+):
+    """Test that multiple get_or_create_db calls only run migrations once."""
+    # Reset module state
+    db._migrations_completed = False
+    db._engine = None
+    db._session_maker = None
+
+    # First call should create engine and run migrations
+    await db.get_or_create_db(app_config.database_path, app_config=app_config)
+    
+    # Verify migrations were called
+    mock_alembic_command.upgrade.assert_called_once_with(mock_alembic_config, "head")
+    mock_search_repository.init_search_index.assert_called_once()
+    
+    # Reset mocks for subsequent calls
+    mock_alembic_command.reset_mock()
+    mock_search_repository.reset_mock()
+    
+    # Subsequent calls should not run migrations again
+    await db.get_or_create_db(app_config.database_path, app_config=app_config)
+    await db.get_or_create_db(app_config.database_path, app_config=app_config)
+    
+    # Verify migrations were NOT called again
+    mock_alembic_command.upgrade.assert_not_called()
+    mock_search_repository.init_search_index.assert_not_called()