feat(core): add note_content tenant schema primitive (#719)

Signed-off-by: phernandez <paul@basicmachines.co>

src/basic_memory/alembic/env.py

@@ -66,7 +66,7 @@ target_metadata = Base.metadata
 
 
 # Add this function to tell Alembic what to include/exclude
-def include_object(object, name, type_, reflected, compare_to):
+def include_object(obj, name, type_, reflected, compare_to):
     # Ignore SQLite FTS tables
     if type_ == "table" and name.startswith("search_index"):
         return False
@@ -118,6 +118,54 @@ async def run_async_migrations(connectable):
     await connectable.dispose()
 
 
+def _run_async_migrations_with_asyncio_run(connectable) -> None:
+    """Run async migrations with asyncio.run while closing failed coroutines.
+
+    Trigger: asyncio.run() may reject execution when another event loop is already active.
+    Why: Python raises before awaiting the coroutine, which otherwise leaks a
+    RuntimeWarning about an un-awaited coroutine.
+    Outcome: close the pending coroutine before bubbling the RuntimeError to the
+    fallback path.
+    """
+    migration_coro = run_async_migrations(connectable)
+    try:
+        asyncio.run(migration_coro)
+    except RuntimeError:
+        migration_coro.close()
+        raise
+
+
+def _run_async_migrations_in_thread(connectable) -> None:
+    """Run async migrations in a dedicated thread with its own event loop."""
+    import concurrent.futures
+
+    def run_in_thread():
+        """Run async migrations in a new event loop in a separate thread."""
+        new_loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(new_loop)
+        try:
+            new_loop.run_until_complete(run_async_migrations(connectable))
+        finally:
+            new_loop.close()
+
+    with concurrent.futures.ThreadPoolExecutor() as executor:
+        future = executor.submit(run_in_thread)
+        future.result()  # Wait for completion and re-raise any exceptions
+
+
+def _run_async_engine_migrations(connectable) -> None:
+    """Run async-engine migrations with a running-loop fallback."""
+    try:
+        _run_async_migrations_with_asyncio_run(connectable)
+    except RuntimeError as e:
+        if "cannot be called from a running event loop" in str(e):
+            # We're in a running event loop (likely uvloop or Python 3.14+ tests).
+            # Switch to a dedicated thread so Alembic can finish without nesting loops.
+            _run_async_migrations_in_thread(connectable)
+        else:
+            raise
+
+
 def run_migrations_online() -> None:
     """Run migrations in 'online' mode.
 
@@ -148,30 +196,10 @@ def run_migrations_online() -> None:
 
     # Handle async engines (PostgreSQL with asyncpg)
     if isinstance(connectable, AsyncEngine):
-        # Try to run async migrations
-        # nest_asyncio allows asyncio.run() from within event loops, but doesn't work with uvloop
-        try:
-            asyncio.run(run_async_migrations(connectable))
-        except RuntimeError as e:
-            if "cannot be called from a running event loop" in str(e):
-                # We're in a running event loop (likely uvloop) - need to use a different approach
-                # Create a new thread to run the async migrations
-                import concurrent.futures
-
-                def run_in_thread():
-                    """Run async migrations in a new event loop in a separate thread."""
-                    new_loop = asyncio.new_event_loop()
-                    asyncio.set_event_loop(new_loop)
-                    try:
-                        new_loop.run_until_complete(run_async_migrations(connectable))
-                    finally:
-                        new_loop.close()
-
-                with concurrent.futures.ThreadPoolExecutor() as executor:
-                    future = executor.submit(run_in_thread)
-                    future.result()  # Wait for completion and re-raise any exceptions
-            else:
-                raise
+        # Trigger: async engines need Alembic work to cross the sync/async boundary.
+        # Why: most callers can use asyncio.run(), but running-loop contexts need a thread fallback.
+        # Outcome: migrations complete without leaking un-awaited coroutines.
+        _run_async_engine_migrations(connectable)
     else:
         # Handle sync engines (SQLite) or sync connections
         if hasattr(connectable, "connect"):

src/basic_memory/alembic/versions/l5g6h7i8j9k0_add_note_content_table.py

@@ -0,0 +1,65 @@
+"""Add note_content table
+
+Revision ID: l5g6h7i8j9k0
+Revises: k4e5f6g7h8i9
+Create Date: 2026-04-04 12:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = "l5g6h7i8j9k0"
+down_revision: Union[str, None] = "k4e5f6g7h8i9"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create note_content for materialized note content and sync state."""
+    op.create_table(
+        "note_content",
+        sa.Column("entity_id", sa.Integer(), nullable=False),
+        sa.Column("project_id", sa.Integer(), nullable=False),
+        sa.Column("external_id", sa.String(), nullable=False),
+        sa.Column("file_path", sa.String(), nullable=False),
+        sa.Column("markdown_content", sa.Text(), nullable=False),
+        sa.Column("db_version", sa.BigInteger(), nullable=False),
+        sa.Column("db_checksum", sa.String(), nullable=False),
+        sa.Column("file_version", sa.BigInteger(), nullable=True),
+        sa.Column("file_checksum", sa.String(), nullable=True),
+        sa.Column("file_write_status", sa.String(), nullable=False),
+        sa.Column("last_source", sa.String(), nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("file_updated_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("last_materialization_error", sa.Text(), nullable=True),
+        sa.Column("last_materialization_attempt_at", sa.DateTime(timezone=True), nullable=True),
+        sa.CheckConstraint(
+            "file_write_status IN ("
+            "'pending', "
+            "'writing', "
+            "'synced', "
+            "'failed', "
+            "'external_change_detected'"
+            ")",
+            name="ck_note_content_file_write_status",
+        ),
+        sa.ForeignKeyConstraint(["entity_id"], ["entity.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["project_id"], ["project.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("entity_id"),
+    )
+    op.create_index("ix_note_content_project_id", "note_content", ["project_id"], unique=False)
+    op.create_index("ix_note_content_file_path", "note_content", ["file_path"], unique=False)
+    op.create_index("ix_note_content_external_id", "note_content", ["external_id"], unique=True)
+
+
+def downgrade() -> None:
+    """Drop note_content and its supporting indexes."""
+    op.drop_index("ix_note_content_external_id", table_name="note_content")
+    op.drop_index("ix_note_content_file_path", table_name="note_content")
+    op.drop_index("ix_note_content_project_id", table_name="note_content")
+    op.drop_table("note_content")

src/basic_memory/models/__init__.py

@@ -2,12 +2,13 @@
 
 import basic_memory
 from basic_memory.models.base import Base
-from basic_memory.models.knowledge import Entity, Observation, Relation
+from basic_memory.models.knowledge import Entity, NoteContent, Observation, Relation
 from basic_memory.models.project import Project
 
 __all__ = [
     "Base",
     "Entity",
+    "NoteContent",
     "Observation",
     "Relation",
     "Project",

src/basic_memory/models/knowledge.py

@@ -6,6 +6,8 @@ from basic_memory.utils import ensure_timezone_aware
 from typing import Optional
 
 from sqlalchemy import (
+    BigInteger,
+    CheckConstraint,
     Integer,
     String,
     Text,
@@ -116,6 +118,12 @@ class Entity(Base):
         foreign_keys="[Relation.to_id]",
         cascade="all, delete-orphan",
     )
+    note_content = relationship(
+        "NoteContent",
+        back_populates="entity",
+        cascade="all, delete-orphan",
+        uselist=False,
+    )
 
     @property
     def relations(self):
@@ -141,6 +149,74 @@ class Entity(Base):
         return f"Entity(id={self.id}, external_id='{self.external_id}', name='{self.title}', type='{self.note_type}', checksum='{self.checksum}')"
 
 
+class NoteContent(Base):
+    """Materialized markdown content and sync state for a note entity."""
+
+    __tablename__ = "note_content"
+    __table_args__ = (
+        CheckConstraint(
+            "file_write_status IN ("
+            "'pending', "
+            "'writing', "
+            "'synced', "
+            "'failed', "
+            "'external_change_detected'"
+            ")",
+            name="ck_note_content_file_write_status",
+        ),
+        Index("ix_note_content_project_id", "project_id"),
+        Index("ix_note_content_file_path", "file_path"),
+        Index("ix_note_content_external_id", "external_id", unique=True),
+    )
+
+    # Core identity mirrored from entity for hot note reads
+    entity_id: Mapped[int] = mapped_column(
+        Integer,
+        ForeignKey("entity.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    project_id: Mapped[int] = mapped_column(
+        Integer,
+        ForeignKey("project.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    external_id: Mapped[str] = mapped_column(String, nullable=False)
+    file_path: Mapped[str] = mapped_column(String, nullable=False)
+
+    # Materialized content version tracked in the tenant database
+    markdown_content: Mapped[str] = mapped_column(Text, nullable=False)
+    db_version: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    db_checksum: Mapped[str] = mapped_column(String, nullable=False)
+
+    # File materialization state tracked against the latest write attempts
+    file_version: Mapped[Optional[int]] = mapped_column(BigInteger, nullable=True)
+    file_checksum: Mapped[Optional[str]] = mapped_column(String, nullable=True)
+    file_write_status: Mapped[str] = mapped_column(String, nullable=False, default="pending")
+    last_source: Mapped[Optional[str]] = mapped_column(String, nullable=True)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now().astimezone(),
+        onupdate=lambda: datetime.now().astimezone(),
+    )
+    file_updated_at: Mapped[Optional[datetime]] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+    )
+    last_materialization_error: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    last_materialization_attempt_at: Mapped[Optional[datetime]] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+    )
+
+    entity = relationship("Entity", back_populates="note_content")
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"NoteContent(entity_id={self.entity_id}, external_id='{self.external_id}', "
+            f"file_path='{self.file_path}', file_write_status='{self.file_write_status}')"
+        )
+
+
 class Observation(Base):
     """An observation about an entity.
 

src/basic_memory/repository/__init__.py

@@ -1,10 +1,12 @@
 from .entity_repository import EntityRepository
+from .note_content_repository import NoteContentRepository
 from .observation_repository import ObservationRepository
 from .project_repository import ProjectRepository
 from .relation_repository import RelationRepository
 
 __all__ = [
     "EntityRepository",
+    "NoteContentRepository",
     "ObservationRepository",
     "ProjectRepository",
     "RelationRepository",

src/basic_memory/repository/note_content_repository.py

@@ -0,0 +1,191 @@
+"""Repository for managing note materialization state."""
+
+from pathlib import Path
+from typing import Any, Mapping, Optional
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from basic_memory import db
+from basic_memory.models import Entity, NoteContent
+from basic_memory.repository.repository import Repository
+
+NOTE_CONTENT_MUTABLE_FIELDS = frozenset(
+    {
+        "markdown_content",
+        "db_version",
+        "db_checksum",
+        "file_version",
+        "file_checksum",
+        "file_write_status",
+        "last_source",
+        "updated_at",
+        "file_updated_at",
+        "last_materialization_error",
+        "last_materialization_attempt_at",
+    }
+)
+
+
+class NoteContentRepository(Repository[NoteContent]):
+    """Repository for project-scoped note materialization state."""
+
+    def __init__(self, session_maker: async_sessionmaker[AsyncSession], project_id: int):
+        """Initialize with session maker and project-scoped filtering."""
+        super().__init__(session_maker, NoteContent, project_id=project_id)
+
+    def _coerce_note_content(
+        self, data: Mapping[str, Any] | NoteContent
+    ) -> tuple[NoteContent, set[str]]:
+        """Convert input data to a NoteContent model and track explicit fields."""
+        if isinstance(data, NoteContent):
+            model_data = {
+                key: value for key, value in data.__dict__.items() if key in self.valid_columns
+            }
+        else:
+            model_data = {key: value for key, value in data.items() if key in self.valid_columns}
+
+        entity_id = model_data.get("entity_id")
+        if entity_id is None:
+            raise ValueError("entity_id is required for note_content writes")
+
+        return NoteContent(**model_data), set(model_data)
+
+    async def _load_entity_identity(self, session: AsyncSession, entity_id: int) -> Entity:
+        """Load the owning entity so duplicated identity fields stay aligned."""
+        result = await session.execute(select(Entity).where(Entity.id == entity_id))
+        entity = result.scalar_one_or_none()
+        if entity is None:
+            raise ValueError(f"Entity {entity_id} does not exist")
+
+        if self.project_id is not None and entity.project_id != self.project_id:
+            raise ValueError(
+                f"Entity {entity_id} belongs to project {entity.project_id}, "
+                f"not repository project {self.project_id}"
+            )
+
+        return entity
+
+    async def _align_identity_fields(
+        self, session: AsyncSession, note_content: NoteContent
+    ) -> None:
+        """Mirror project identity from entity before persisting note content."""
+        entity = await self._load_entity_identity(session, note_content.entity_id)
+        note_content.project_id = entity.project_id
+        note_content.external_id = entity.external_id
+        note_content.file_path = Path(entity.file_path).as_posix()
+
+    async def get_by_entity_id(self, entity_id: int) -> Optional[NoteContent]:
+        """Get note content by the owning entity identifier."""
+        return await self.find_by_id(entity_id)
+
+    async def get_by_external_id(self, external_id: str) -> Optional[NoteContent]:
+        """Get note content by the mirrored entity external identifier."""
+        query = self.select().where(NoteContent.external_id == external_id)
+        return await self.find_one(query)
+
+    async def get_by_file_path(self, file_path: Path | str) -> Optional[NoteContent]:
+        """Get note content by file path, preferring rows whose entity still owns that path."""
+        normalized_path = Path(file_path).as_posix()
+
+        # Trigger: note_content mirrors entity.file_path but does not enforce project-level uniqueness.
+        # Why: entity renames can leave stale mirrored paths behind until note_content realigns.
+        # Outcome: prefer the row whose current entity path still matches, then the newest mirror.
+        query = (
+            self.select()
+            .join(Entity, Entity.id == NoteContent.entity_id)
+            .where(NoteContent.file_path == normalized_path)
+            .order_by(
+                (Entity.file_path == normalized_path).desc(),
+                NoteContent.updated_at.desc(),
+                NoteContent.entity_id.desc(),
+            )
+            .limit(1)
+            .options(*self.get_load_options())
+        )
+
+        async with db.scoped_session(self.session_maker) as session:
+            result = await session.execute(query)
+            return result.scalars().first()
+
+    async def create(self, data: Mapping[str, Any] | NoteContent) -> NoteContent:
+        """Create a note_content row aligned to its owning entity."""
+        note_content, _ = self._coerce_note_content(data)
+
+        async with db.scoped_session(self.session_maker) as session:
+            await self._align_identity_fields(session, note_content)
+            session.add(note_content)
+            await session.flush()
+
+            created = await self.select_by_id(session, note_content.entity_id)
+            if created is None:  # pragma: no cover
+                raise ValueError(
+                    f"Can't find NoteContent for entity {note_content.entity_id} after add"
+                )
+            return created
+
+    async def upsert(self, data: Mapping[str, Any] | NoteContent) -> NoteContent:
+        """Insert or update note_content while keeping mirrored identity fields in sync."""
+        note_content, provided_fields = self._coerce_note_content(data)
+
+        async with db.scoped_session(self.session_maker) as session:
+            await self._align_identity_fields(session, note_content)
+            existing = await self.select_by_id(session, note_content.entity_id)
+
+            if existing is None:
+                session.add(note_content)
+                await session.flush()
+                created = await self.select_by_id(session, note_content.entity_id)
+                if created is None:  # pragma: no cover
+                    raise ValueError(
+                        f"Can't find NoteContent for entity {note_content.entity_id} after upsert"
+                    )
+                return created
+
+            fields_to_update = (provided_fields - {"entity_id"}) | {
+                "project_id",
+                "external_id",
+                "file_path",
+            }
+            for column_name in fields_to_update:
+                setattr(existing, column_name, getattr(note_content, column_name))
+
+            await session.flush()
+            updated = await self.select_by_id(session, existing.entity_id)
+            if updated is None:  # pragma: no cover
+                raise ValueError(
+                    f"Can't find NoteContent for entity {existing.entity_id} after upsert"
+                )
+            return updated
+
+    async def update_state_fields(self, entity_id: int, **updates: Any) -> Optional[NoteContent]:
+        """Update sync fields and re-align project_id, external_id, and file_path from entity."""
+        invalid_fields = set(updates) - NOTE_CONTENT_MUTABLE_FIELDS
+        if invalid_fields:
+            invalid_list = ", ".join(sorted(invalid_fields))
+            raise ValueError(f"Unsupported note_content update fields: {invalid_list}")
+
+        async with db.scoped_session(self.session_maker) as session:
+            note_content = await self.select_by_id(session, entity_id)
+            if note_content is None:
+                return None
+
+            await self._align_identity_fields(session, note_content)
+            for field_name, value in updates.items():
+                setattr(note_content, field_name, value)
+
+            await session.flush()
+            updated = await self.select_by_id(session, entity_id)
+            if updated is None:  # pragma: no cover
+                raise ValueError(f"Can't find NoteContent for entity {entity_id} after update")
+            return updated
+
+    async def delete_by_entity_id(self, entity_id: int) -> bool:
+        """Delete note_content by entity identifier."""
+        async with db.scoped_session(self.session_maker) as session:
+            note_content = await self.select_by_id(session, entity_id)
+            if note_content is None:
+                return False
+
+            await session.delete(note_content)
+            return True