feat: add created_by and last_updated_by user tracking to Entity (#602)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

src/basic_memory/alembic/versions/k4e5f6g7h8i9_add_created_by_and_last_updated_by_to_entity.py

@@ -0,0 +1,74 @@
+"""Add created_by and last_updated_by columns to entity table.
+
+Revision ID: k4e5f6g7h8i9
+Revises: j3d4e5f6g7h8
+Create Date: 2026-02-23 00:00:00.000000
+
+These columns track which cloud user created and last modified each entity.
+Both are nullable — NULL for local/CLI usage and existing entities.
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import text
+
+# revision identifiers, used by Alembic.
+revision: str = "k4e5f6g7h8i9"
+down_revision: Union[str, None] = "j3d4e5f6g7h8"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def column_exists(connection, table: str, column: str) -> bool:
+    """Check if a column exists in a table (idempotent migration support)."""
+    if connection.dialect.name == "postgresql":
+        result = connection.execute(
+            text(
+                "SELECT 1 FROM information_schema.columns "
+                "WHERE table_name = :table AND column_name = :column"
+            ),
+            {"table": table, "column": column},
+        )
+        return result.fetchone() is not None
+    else:
+        # SQLite
+        result = connection.execute(text(f"PRAGMA table_info({table})"))
+        columns = [row[1] for row in result]
+        return column in columns
+
+
+def upgrade() -> None:
+    """Add created_by and last_updated_by columns to entity table.
+
+    Both columns are nullable strings that store cloud user_profile_id UUIDs.
+    No data backfill — existing rows get NULL.
+    """
+    connection = op.get_bind()
+
+    if not column_exists(connection, "entity", "created_by"):
+        op.add_column("entity", sa.Column("created_by", sa.String(), nullable=True))
+
+    if not column_exists(connection, "entity", "last_updated_by"):
+        op.add_column("entity", sa.Column("last_updated_by", sa.String(), nullable=True))
+
+
+def downgrade() -> None:
+    """Remove created_by and last_updated_by columns from entity table."""
+    connection = op.get_bind()
+    dialect = connection.dialect.name
+
+    if column_exists(connection, "entity", "last_updated_by"):
+        if dialect == "postgresql":
+            op.drop_column("entity", "last_updated_by")
+        else:
+            with op.batch_alter_table("entity") as batch_op:
+                batch_op.drop_column("last_updated_by")
+
+    if column_exists(connection, "entity", "created_by"):
+        if dialect == "postgresql":
+            op.drop_column("entity", "created_by")
+        else:
+            with op.batch_alter_table("entity") as batch_op:
+                batch_op.drop_column("created_by")

src/basic_memory/models/knowledge.py

@@ -94,6 +94,11 @@ class Entity(Base):
         onupdate=lambda: datetime.now().astimezone(),
     )
 
+    # Who created this entity (cloud user_profile_id UUID, null for local/CLI usage)
+    created_by: Mapped[Optional[str]] = mapped_column(String, nullable=True, default=None)
+    # Who last modified this entity (cloud user_profile_id UUID, null for local/CLI usage)
+    last_updated_by: Mapped[Optional[str]] = mapped_column(String, nullable=True, default=None)
+
     # Relationships
     project = relationship("Project", back_populates="entities")
     observations = relationship(

src/basic_memory/schemas/v2/entity.py

@@ -139,6 +139,10 @@ class EntityResponseV2(BaseModel):
     created_at: datetime = Field(..., description="Creation timestamp")
     updated_at: datetime = Field(..., description="Last update timestamp")
 
+    # User tracking (cloud only, null for local/CLI usage)
+    created_by: Optional[str] = Field(None, description="User profile ID of creator")
+    last_updated_by: Optional[str] = Field(None, description="User profile ID of last editor")
+
     # V2-specific metadata
     api_version: Literal["v2"] = Field(
         default="v2", description="API version (always 'v2' for this response)"

src/basic_memory/services/entity_service.py

@@ -1,5 +1,6 @@
 """Service for managing entities in the database."""
 
+from collections.abc import Callable
 from datetime import datetime
 from pathlib import Path
 from typing import List, Optional, Sequence, Tuple, Union
@@ -68,6 +69,9 @@ class EntityService(BaseService[EntityModel]):
         self.search_service = search_service
         self.app_config = app_config
         self._project_permalink: Optional[str] = None
+        # Callable that returns the current user ID (cloud user_profile_id UUID as string).
+        # Default returns None for local/CLI usage. Cloud overrides this to read from UserContext.
+        self.get_user_id: Callable[[], Optional[str]] = lambda: None
 
     async def detect_file_path_conflicts(
         self, file_path: str, skip_check: bool = False
@@ -445,7 +449,12 @@ class EntityService(BaseService[EntityModel]):
             "updated_at": datetime.now().astimezone(),
         }
 
+        user_id = self.get_user_id()
+
         if existing:
+            # Preserve existing created_by; only update last_updated_by
+            if user_id is not None:
+                update_data["last_updated_by"] = user_id
             updated = await self.repository.update(existing.id, update_data)
             if not updated:
                 raise ValueError(f"Failed to update entity in database: {existing.id}")
@@ -454,6 +463,9 @@ class EntityService(BaseService[EntityModel]):
         create_data = dict(update_data)
         if external_id is not None:
             create_data["external_id"] = external_id
+        if user_id is not None:
+            create_data["created_by"] = user_id
+            create_data["last_updated_by"] = user_id
         return await self.repository.create(create_data)
 
     async def fast_edit_entity(
@@ -481,6 +493,10 @@ class EntityService(BaseService[EntityModel]):
             "checksum": checksum,
             "updated_at": datetime.now().astimezone(),
         }
+        user_id = self.get_user_id()
+        if user_id is not None:
+            update_data["last_updated_by"] = user_id
+
         content_markdown = None
         if has_frontmatter(new_content):
             content_frontmatter = parse_frontmatter(new_content)
@@ -611,6 +627,12 @@ class EntityService(BaseService[EntityModel]):
         # Mark as incomplete because we still need to add relations
         model.checksum = None
 
+        # Set user tracking fields for cloud usage
+        user_id = self.get_user_id()
+        if user_id is not None:
+            model.created_by = user_id
+            model.last_updated_by = user_id
+
         # Use UPSERT to handle conflicts cleanly
         try:
             return await self.repository.upsert_entity(model)
@@ -653,6 +675,11 @@ class EntityService(BaseService[EntityModel]):
         # checksum value is None == not finished with sync
         db_entity.checksum = None
 
+        # Set last_updated_by for cloud usage (preserve existing created_by)
+        user_id = self.get_user_id()
+        if user_id is not None:
+            db_entity.last_updated_by = user_id
+
         # update entity
         return await self.repository.update(
             db_entity.id,

tests/api/v2/test_knowledge_router.py

@@ -833,3 +833,24 @@ async def test_delete_directory_v2_nested_structure(client: AsyncClient, v2_proj
     assert result.total_files == 2
     assert result.successful_deletes == 2
     assert result.failed_deletes == 0
+
+
+@pytest.mark.asyncio
+async def test_entity_response_includes_user_tracking_fields(
+    client: AsyncClient, v2_project_url
+):
+    """EntityResponseV2 includes created_by and last_updated_by fields (null for local)."""
+    entity_data = {
+        "title": "UserTrackingTest",
+        "directory": "test",
+        "content": "Test content",
+    }
+    response = await client.post(f"{v2_project_url}/knowledge/entities", json=entity_data)
+    assert response.status_code == 200
+
+    body = response.json()
+    # Fields should be present in the response (null for local/CLI usage)
+    assert "created_by" in body
+    assert "last_updated_by" in body
+    assert body["created_by"] is None
+    assert body["last_updated_by"] is None

tests/services/test_entity_service.py

@@ -2017,3 +2017,151 @@ async def test_create_or_update_entity_fuzzy_search_bug(
     assert "Original content for Node A" not in content_c, (
         "Node C.md should not contain Node A content"
     )
+
+
+# --- User Tracking (created_by / last_updated_by) ---
+
+
+@pytest.mark.asyncio
+async def test_created_by_null_by_default(entity_service: EntityService):
+    """created_by and last_updated_by are NULL when get_user_id returns None (local/CLI usage)."""
+    schema = EntitySchema(
+        title="Local Entity",
+        directory="test",
+        entity_type="note",
+    )
+    entity = await entity_service.create_entity(schema)
+    assert entity.created_by is None
+    assert entity.last_updated_by is None
+
+
+@pytest.mark.asyncio
+async def test_created_by_set_when_get_user_id_returns_value(entity_service: EntityService):
+    """created_by and last_updated_by are set when get_user_id returns a user ID."""
+    user_id = str(uuid.uuid4())
+    entity_service.get_user_id = lambda: user_id
+
+    schema = EntitySchema(
+        title="Cloud Entity",
+        directory="test",
+        entity_type="note",
+    )
+    entity = await entity_service.create_entity(schema)
+    assert entity.created_by == user_id
+    assert entity.last_updated_by == user_id
+
+
+@pytest.mark.asyncio
+async def test_update_preserves_created_by(entity_service: EntityService):
+    """Updating an entity preserves created_by and updates last_updated_by."""
+    creator_id = str(uuid.uuid4())
+    editor_id = str(uuid.uuid4())
+
+    # Create as creator
+    entity_service.get_user_id = lambda: creator_id
+    schema = EntitySchema(
+        title="Owned Entity",
+        directory="test",
+        entity_type="note",
+        content="Original content",
+    )
+    entity = await entity_service.create_entity(schema)
+    assert entity.created_by == creator_id
+
+    # Update as editor
+    entity_service.get_user_id = lambda: editor_id
+    update_schema = EntitySchema(
+        title="Owned Entity",
+        directory="test",
+        entity_type="note",
+        content="Updated content",
+    )
+    updated = await entity_service.update_entity(entity, update_schema)
+    assert updated.created_by == creator_id  # preserved
+    assert updated.last_updated_by == editor_id  # updated
+
+
+@pytest.mark.asyncio
+async def test_fast_write_entity_sets_user_tracking(entity_service: EntityService):
+    """fast_write_entity sets created_by and last_updated_by on create."""
+    user_id = str(uuid.uuid4())
+    entity_service.get_user_id = lambda: user_id
+
+    schema = EntitySchema(
+        title="Fast Write Tracked",
+        directory="test",
+        entity_type="note",
+    )
+    entity = await entity_service.fast_write_entity(schema, external_id=str(uuid.uuid4()))
+    assert entity.created_by == user_id
+    assert entity.last_updated_by == user_id
+
+
+@pytest.mark.asyncio
+async def test_fast_write_entity_update_preserves_created_by(entity_service: EntityService):
+    """fast_write_entity update path preserves created_by, sets last_updated_by."""
+    creator_id = str(uuid.uuid4())
+    editor_id = str(uuid.uuid4())
+    external_id = str(uuid.uuid4())
+
+    # Create
+    entity_service.get_user_id = lambda: creator_id
+    schema = EntitySchema(
+        title="Fast Write Update",
+        directory="test",
+        entity_type="note",
+    )
+    entity = await entity_service.fast_write_entity(schema, external_id=external_id)
+    assert entity.created_by == creator_id
+
+    # Update (same external_id triggers update path)
+    entity_service.get_user_id = lambda: editor_id
+    update_schema = EntitySchema(
+        title="Fast Write Update",
+        directory="test",
+        entity_type="note",
+        content="Updated",
+    )
+    updated = await entity_service.fast_write_entity(update_schema, external_id=external_id)
+    assert updated.created_by == creator_id  # preserved
+    assert updated.last_updated_by == editor_id  # updated
+
+
+@pytest.mark.asyncio
+async def test_fast_edit_entity_sets_last_updated_by(entity_service: EntityService):
+    """fast_edit_entity sets last_updated_by on edit."""
+    creator_id = str(uuid.uuid4())
+    editor_id = str(uuid.uuid4())
+
+    # Create entity first
+    entity_service.get_user_id = lambda: creator_id
+    schema = EntitySchema(
+        title="Fast Edit Tracked",
+        directory="test",
+        entity_type="note",
+        content="Original content",
+    )
+    entity = await entity_service.fast_write_entity(schema, external_id=str(uuid.uuid4()))
+
+    # Edit as different user
+    entity_service.get_user_id = lambda: editor_id
+    edited = await entity_service.fast_edit_entity(
+        entity=entity,
+        operation="append",
+        content="\nAppended content",
+    )
+    assert edited.created_by == creator_id  # preserved
+    assert edited.last_updated_by == editor_id  # updated
+
+
+@pytest.mark.asyncio
+async def test_fast_write_entity_null_user_id(entity_service: EntityService):
+    """fast_write_entity with default get_user_id (None) leaves tracking fields null."""
+    schema = EntitySchema(
+        title="No User Tracking",
+        directory="test",
+        entity_type="note",
+    )
+    entity = await entity_service.fast_write_entity(schema, external_id=str(uuid.uuid4()))
+    assert entity.created_by is None
+    assert entity.last_updated_by is None