diff --git a/src/basic_memory/alembic/versions/s2p3e4c5w6k7_add_project_partition_position.py b/src/basic_memory/alembic/versions/s2p3e4c5w6k7_add_project_partition_position.py new file mode 100644 index 000000000..1e8c47452 --- /dev/null +++ b/src/basic_memory/alembic/versions/s2p3e4c5w6k7_add_project_partition_position.py @@ -0,0 +1,87 @@ +"""Add the strict accepted-change partition head to projects. + +Revision ID: s2p3e4c5w6k7 +Revises: bcdbd5a942ca +Create Date: 2026-08-29 20:15:00.000000 + +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "s2p3e4c5w6k7" +down_revision: Union[str, None] = "bcdbd5a942ca" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Add the project partition head and its durable accepted evidence.""" + with op.batch_alter_table("project", schema=None) as batch_op: + batch_op.add_column( + sa.Column( + "partition_position", + sa.Integer(), + server_default=sa.text("0"), + nullable=False, + ) + ) + + op.create_table( + "accepted_project_note_change", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("project_id", sa.Integer(), nullable=False), + sa.Column("project_external_id", sa.String(), nullable=False), + sa.Column("partition_position", sa.Integer(), nullable=False), + sa.Column("entity_id", sa.Integer(), nullable=False), + sa.Column("note_external_id", sa.String(), nullable=False), + sa.Column("title", sa.Text(), nullable=False), + sa.Column("operation", sa.String(), nullable=False), + sa.Column("file_path", sa.Text(), nullable=False), + sa.Column("previous_file_path", sa.Text(), nullable=True), + sa.Column("accepted_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("source", sa.String(), nullable=False), + sa.Column("db_version", sa.Integer(), nullable=True), + sa.Column("db_checksum", sa.String(), nullable=True), + sa.Column("actor_user_profile_id", sa.String(), nullable=True), + sa.Column("actor_kind", sa.String(), nullable=True), + sa.Column("actor_name", sa.String(), nullable=True), + sa.Column("materialized_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["project_id"], ["project.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "project_id", + "partition_position", + name="uq_accepted_project_note_change_project_position", + ), + ) + op.create_index( + "ix_accepted_project_note_change_project_materialized", + "accepted_project_note_change", + ["project_id", "materialized_at"], + unique=False, + ) + op.create_index( + "ix_accepted_project_note_change_note_external_id", + "accepted_project_note_change", + ["note_external_id"], + unique=False, + ) + + +def downgrade() -> None: + """Remove accepted evidence and the project partition head.""" + op.drop_index( + "ix_accepted_project_note_change_note_external_id", + table_name="accepted_project_note_change", + ) + op.drop_index( + "ix_accepted_project_note_change_project_materialized", + table_name="accepted_project_note_change", + ) + op.drop_table("accepted_project_note_change") + with op.batch_alter_table("project", schema=None) as batch_op: + batch_op.drop_column("partition_position") diff --git a/src/basic_memory/alembic/versions/t3q4r5s6x7y8_reconcile_accepted_project_note_change.py b/src/basic_memory/alembic/versions/t3q4r5s6x7y8_reconcile_accepted_project_note_change.py new file mode 100644 index 000000000..526d0100f --- /dev/null +++ b/src/basic_memory/alembic/versions/t3q4r5s6x7y8_reconcile_accepted_project_note_change.py @@ -0,0 +1,103 @@ +"""Reconcile accepted project change storage for pre-release tenants. + +Revision ID: t3q4r5s6x7y8 +Revises: s2p3e4c5w6k7 +Create Date: 2026-08-30 02:15:00.000000 + +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy import inspect + + +revision: str = "t3q4r5s6x7y8" +down_revision: Union[str, None] = "s2p3e4c5w6k7" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def _create_accepted_project_note_change() -> None: + op.create_table( + "accepted_project_note_change", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("project_id", sa.Integer(), nullable=False), + sa.Column("project_external_id", sa.String(), nullable=False), + sa.Column("partition_position", sa.Integer(), nullable=False), + sa.Column("entity_id", sa.Integer(), nullable=False), + sa.Column("note_external_id", sa.String(), nullable=False), + sa.Column("title", sa.Text(), nullable=False), + sa.Column("operation", sa.String(), nullable=False), + sa.Column("file_path", sa.Text(), nullable=False), + sa.Column("previous_file_path", sa.Text(), nullable=True), + sa.Column("accepted_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("source", sa.String(), nullable=False), + sa.Column("db_version", sa.Integer(), nullable=True), + sa.Column("db_checksum", sa.String(), nullable=True), + sa.Column("actor_user_profile_id", sa.String(), nullable=True), + sa.Column("actor_kind", sa.String(), nullable=True), + sa.Column("actor_name", sa.String(), nullable=True), + sa.Column("materialized_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["project_id"], ["project.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "project_id", + "partition_position", + name="uq_accepted_project_note_change_project_position", + ), + ) + + +def _create_missing_indexes(existing_indexes: set[str]) -> None: + indexes = { + "ix_accepted_project_note_change_project_materialized": [ + "project_id", + "materialized_at", + ], + "ix_accepted_project_note_change_note_external_id": ["note_external_id"], + } + for index_name, columns in indexes.items(): + if index_name not in existing_indexes: + op.create_index( + index_name, + "accepted_project_note_change", + columns, + unique=False, + ) + + +def upgrade() -> None: + """Repair tenants stamped while the preceding revision was still pre-release.""" + connection = op.get_bind() + inspector = inspect(connection) + + project_columns = {column["name"] for column in inspector.get_columns("project")} + if "partition_position" not in project_columns: + with op.batch_alter_table("project", schema=None) as batch_op: + batch_op.add_column( + sa.Column( + "partition_position", + sa.Integer(), + server_default=sa.text("0"), + nullable=False, + ) + ) + + table_names = set(inspector.get_table_names()) + if "accepted_project_note_change" not in table_names: + _create_accepted_project_note_change() + _create_missing_indexes(set()) + return + + existing_indexes = { + index["name"] + for index in inspector.get_indexes("accepted_project_note_change") + if index["name"] is not None + } + _create_missing_indexes(existing_indexes) + + +def downgrade() -> None: + """Keep the schema promised by the preceding revision.""" diff --git a/src/basic_memory/indexing/accepted_note_mutation_runner.py b/src/basic_memory/indexing/accepted_note_mutation_runner.py index ffa3b9b25..618770efa 100644 --- a/src/basic_memory/indexing/accepted_note_mutation_runner.py +++ b/src/basic_memory/indexing/accepted_note_mutation_runner.py @@ -2,7 +2,7 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import UTC, datetime from enum import StrEnum from typing import NoReturn, Protocol @@ -46,6 +46,10 @@ ) from basic_memory.runtime.note_move import normalize_note_move_destination_path from basic_memory.runtime.note_object_metadata import NOTE_SOURCE_COLLABORATION_RELAY +from basic_memory.runtime.project_partition import ( + RuntimeAcceptedProjectNoteChange, + RuntimeProjectNoteOperation, +) from basic_memory.runtime.storage import ( NoteExternalId, ProjectExternalId, @@ -64,6 +68,7 @@ type AcceptedNoteMutationChange = RuntimeAcceptedNoteChange[RuntimeNoteContentResponsePayload] type AcceptedNoteMutationUserProfileId = UUID +ACCEPTED_NOTE_DELETE_SOURCE: RuntimeNoteChangeSource = "delete_note" class AcceptedNoteMutationRejectKind(StrEnum): @@ -142,6 +147,7 @@ class AcceptedNoteCreateMutation: data: EntitySchema actor: AcceptedNoteMutationActor source: RuntimeNoteChangeSource + publish_graph_facts: bool = True @dataclass(frozen=True, slots=True) @@ -155,6 +161,7 @@ class AcceptedNoteUpdateMutation: source: RuntimeNoteChangeSource # db_checksum the caller last synced; None means no precondition (issue #1445). base_checksum: str | None = None + publish_graph_facts: bool = True @dataclass(frozen=True, slots=True) @@ -214,6 +221,18 @@ async def get_by_external_id( external_id: ProjectExternalId, ) -> Project | None: ... + async def advance_partition_position( + self, + session: AsyncSession, + project_id: ProjectId, + ) -> int: ... + + async def record_accepted_note_change( + self, + session: AsyncSession, + change: RuntimeAcceptedProjectNoteChange, + ) -> None: ... + class AcceptedNoteMutationEntityRepository(Protocol): """Entity lookup capability for accepted-note mutations.""" @@ -321,6 +340,87 @@ class AcceptedNoteMutationResult: relation_publication: RelationGenerationPublication | None = None +async def record_accepted_project_note_change( + session: AsyncSession, + *, + project: Project, + entity: Entity, + operation: RuntimeProjectNoteOperation, + accepted_at: datetime, + source: RuntimeNoteChangeSource, + previous_file_path: RuntimeFilePath | None, + note_content: NoteContent | None, + actor: AcceptedNoteMutationActor | None, + dependencies: AcceptedNoteMutationDependencies, +) -> RuntimeAcceptedProjectNoteChange: + """Claim and describe one accepted change in the project's strict partition.""" + position = await dependencies.project_repository.advance_partition_position( + session, + project.id, + ) + change = RuntimeAcceptedProjectNoteChange( + project_id=project.id, + project_external_id=project.external_id, + partition_position=position, + entity_id=entity.id, + note_external_id=entity.external_id, + title=entity.title, + operation=operation, + file_path=entity.file_path, + previous_file_path=previous_file_path, + accepted_at=accepted_at, + source=source, + db_version=note_content.db_version if note_content is not None else None, + db_checksum=note_content.db_checksum if note_content is not None else None, + actor_user_profile_id=actor.user_profile_id if actor is not None else None, + actor_kind=actor.kind if actor is not None else None, + actor_name=actor.name if actor is not None else None, + ) + await dependencies.project_repository.record_accepted_note_change(session, change) + return change + + +def attach_accepted_project_note_change( + change: AcceptedNoteMutationChange, + project_change: RuntimeAcceptedProjectNoteChange, +) -> AcceptedNoteMutationChange: + """Carry accepted partition evidence through existing runtime follow-up work.""" + materialization = ( + replace(change.materialization, project_change=project_change) + if change.materialization is not None + else None + ) + file_delete = ( + replace(change.file_delete, project_change=project_change) + if change.file_delete is not None + else None + ) + return replace( + change, + project_change=project_change, + materialization=materialization, + file_delete=file_delete, + ) + + +def apply_accepted_note_graph_policy( + prepared_write: AcceptedPreparedNoteWrite, + *, + publish_graph_facts: bool, +) -> AcceptedPreparedNoteWrite: + """Keep canonical Markdown while suppressing graph facts for derived documents.""" + if publish_graph_facts: + return prepared_write + prepared = prepared_write.prepared + graph_silent_markdown = prepared.entity_markdown.model_copy( + update={"observations": [], "relations": []} + ) + return replace( + prepared_write, + prepared=replace(prepared, entity_markdown=graph_silent_markdown), + ) + + def accepted_note_integrity_rejection(error: IntegrityError) -> AcceptedNoteMutationRejection: """Map repository integrity errors into portable accepted-note rejections.""" conflict_kind = classify_accepted_note_write_conflict(str(error.orig or error)) @@ -530,14 +630,27 @@ async def run_accepted_note_delete( dependencies=dependencies, missing_kind=None, ) + change = await delete_accepted_note( + session, + project_id=project.id, + entity=entity, + note_content=note_content, + repositories=dependencies.write_repositories, + ) + project_change = await record_accepted_project_note_change( + session, + project=project, + entity=entity, + operation=RuntimeProjectNoteOperation.deleted, + accepted_at=accepted_note_mutation_utc_now(), + source=ACCEPTED_NOTE_DELETE_SOURCE, + previous_file_path=None, + note_content=note_content, + actor=None, + dependencies=dependencies, + ) return AcceptedNoteMutationResult( - change=await delete_accepted_note( - session, - project_id=project.id, - entity=entity, - note_content=note_content, - repositories=dependencies.write_repositories, - ) + change=attach_accepted_project_note_change(change, project_change) ) @@ -583,7 +696,10 @@ async def _run_accepted_note_create( check_storage_exists=dependencies.verify_storage_absent_on_create, session=session, ) - + prepared_write = apply_accepted_note_graph_policy( + prepared_write, + publish_graph_facts=request.publish_graph_facts, + ) prepared = prepared_write.prepared entity = await create_accepted_pending_entity( session, @@ -602,15 +718,30 @@ async def _run_accepted_note_create( self_relation_resolver=preparer, repositories=dependencies.write_repositories, ) + project_change = await record_accepted_project_note_change( + session, + project=project, + entity=entity, + operation=RuntimeProjectNoteOperation.created, + accepted_at=now, + source=request.source, + previous_file_path=None, + note_content=persisted.note_content, + actor=request.actor, + dependencies=dependencies, + ) return AcceptedNoteMutationResult( - change=plan_accepted_note_write_change( - status_code=201, - entity=entity, - note_content=persisted.note_content, - actor_user_profile_id=request.actor.user_profile_id, - actor_kind=request.actor.kind, - actor_name=request.actor.name, - fallback_source=request.source, + change=attach_accepted_project_note_change( + plan_accepted_note_write_change( + status_code=201, + entity=entity, + note_content=persisted.note_content, + actor_user_profile_id=request.actor.user_profile_id, + actor_kind=request.actor.kind, + actor_name=request.actor.name, + fallback_source=request.source, + ), + project_change, ), relation_publication=persisted.relation_publication, ) @@ -780,6 +911,10 @@ async def _run_accepted_note_update( except (ParseError, ValueError) as error: reject_accepted_note_mutation(AcceptedNoteMutationRejectKind.bad_request, str(error)) + prepared_write = apply_accepted_note_graph_policy( + prepared_write, + publish_graph_facts=request.publish_graph_facts, + ) prepared = prepared_write.prepared persisted = await persist_accepted_note_snapshot( session, @@ -806,16 +941,43 @@ async def _run_accepted_note_update( file_path=vacated_source[0], file_checksum=vacated_source[1], ) + operation = ( + RuntimeProjectNoteOperation.created + if created + else ( + RuntimeProjectNoteOperation.moved + if existing_file_path != entity.file_path + else RuntimeProjectNoteOperation.updated + ) + ) + previous_file_path = ( + existing_file_path if operation == RuntimeProjectNoteOperation.moved else None + ) + project_change = await record_accepted_project_note_change( + session, + project=project, + entity=entity, + operation=operation, + accepted_at=now, + source=request.source, + previous_file_path=previous_file_path, + note_content=persisted.note_content, + actor=request.actor, + dependencies=dependencies, + ) return AcceptedNoteMutationResult( - change=plan_accepted_note_write_change( - status_code=201 if created else 200, - entity=entity, - note_content=persisted.note_content, - actor_user_profile_id=request.actor.user_profile_id, - actor_kind=request.actor.kind, - actor_name=request.actor.name, - cleanup_after_write=persisted.previous_file_delete, - fallback_source=request.source, + change=attach_accepted_project_note_change( + plan_accepted_note_write_change( + status_code=201 if created else 200, + entity=entity, + note_content=persisted.note_content, + actor_user_profile_id=request.actor.user_profile_id, + actor_kind=request.actor.kind, + actor_name=request.actor.name, + cleanup_after_write=persisted.previous_file_delete, + fallback_source=request.source, + ), + project_change, ), relation_publication=persisted.relation_publication, ) @@ -869,15 +1031,30 @@ async def _run_accepted_note_edit( self_relation_resolver=preparer, repositories=dependencies.write_repositories, ) + project_change = await record_accepted_project_note_change( + session, + project=project, + entity=entity, + operation=RuntimeProjectNoteOperation.updated, + accepted_at=now, + source=request.source, + previous_file_path=None, + note_content=persisted.note_content, + actor=request.actor, + dependencies=dependencies, + ) return AcceptedNoteMutationResult( - change=plan_accepted_note_write_change( - status_code=200, - entity=entity, - note_content=persisted.note_content, - actor_user_profile_id=request.actor.user_profile_id, - actor_kind=request.actor.kind, - actor_name=request.actor.name, - fallback_source=request.source, + change=attach_accepted_project_note_change( + plan_accepted_note_write_change( + status_code=200, + entity=entity, + note_content=persisted.note_content, + actor_user_profile_id=request.actor.user_profile_id, + actor_kind=request.actor.kind, + actor_name=request.actor.name, + fallback_source=request.source, + ), + project_change, ), relation_publication=persisted.relation_publication, ) @@ -994,17 +1171,32 @@ async def _run_accepted_note_move( file_path=existing_file_path, file_checksum=vacated_source_checksum, ) + project_change = await record_accepted_project_note_change( + session, + project=project, + entity=entity, + operation=RuntimeProjectNoteOperation.moved, + accepted_at=now, + source=request.source, + previous_file_path=existing_file_path, + note_content=persisted.note_content, + actor=request.actor, + dependencies=dependencies, + ) return AcceptedNoteMutationResult( - change=plan_accepted_note_write_change( - status_code=200, - entity=entity, - note_content=persisted.note_content, - actor_user_profile_id=request.actor.user_profile_id, - actor_kind=request.actor.kind, - actor_name=request.actor.name, - previous_file_path=existing_file_path, - cleanup_after_write=persisted.previous_file_delete, - fallback_source=request.source, + change=attach_accepted_project_note_change( + plan_accepted_note_write_change( + status_code=200, + entity=entity, + note_content=persisted.note_content, + actor_user_profile_id=request.actor.user_profile_id, + actor_kind=request.actor.kind, + actor_name=request.actor.name, + previous_file_path=existing_file_path, + cleanup_after_write=persisted.previous_file_delete, + fallback_source=request.source, + ), + project_change, ), relation_publication=persisted.relation_publication, ) diff --git a/src/basic_memory/models/__init__.py b/src/basic_memory/models/__init__.py index 186d45b5e..cc5285b78 100644 --- a/src/basic_memory/models/__init__.py +++ b/src/basic_memory/models/__init__.py @@ -9,11 +9,12 @@ Observation, Relation, ) -from basic_memory.models.project import Project +from basic_memory.models.project import AcceptedProjectNoteChange, Project from basic_memory.models.relation_search_refresh import RelationSearchRefresh __all__ = [ "Base", + "AcceptedProjectNoteChange", "Entity", "NoteContent", "NoteFileVacate", diff --git a/src/basic_memory/models/project.py b/src/basic_memory/models/project.py index a39ddaacf..f271c980c 100644 --- a/src/basic_memory/models/project.py +++ b/src/basic_memory/models/project.py @@ -11,7 +11,10 @@ Boolean, DateTime, Float, + ForeignKey, Index, + UniqueConstraint, + text, event, ) from sqlalchemy.orm import Mapped, mapped_column, relationship @@ -70,9 +73,25 @@ class Project(Base): last_scan_timestamp: Mapped[Optional[float]] = mapped_column(Float, nullable=True) last_file_count: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) + # Strict, project-local ordering for accepted durable changes. This is the + # generic partition head that future event-journal projectors can reuse; it + # is deliberately not derived from timestamps, webhooks, or scan activity. + partition_position: Mapped[int] = mapped_column( + Integer, + nullable=False, + default=0, + server_default=text("0"), + ) + # Define relationships to entities, observations, and relations # These relationships will be established once we add project_id to those models entities = relationship("Entity", back_populates="project", cascade="all, delete-orphan") + accepted_note_changes = relationship( + "AcceptedProjectNoteChange", + back_populates="project", + cascade="all, delete-orphan", + order_by="AcceptedProjectNoteChange.partition_position", + ) @override def __repr__(self) -> str: # pragma: no cover @@ -90,3 +109,56 @@ def set_project_permalink(mapper, connection, project): # If the name changed or permalink is empty, regenerate permalink if not project.permalink or project.permalink != generate_permalink(project.name): project.permalink = generate_permalink(project.name) + + +class AcceptedProjectNoteChange(Base): + """Durable evidence for one accepted note mutation in project order. + + The row intentionally retains note identity and path values instead of an + entity foreign key: delete evidence must survive after the entity is gone. + """ + + __tablename__ = "accepted_project_note_change" + __table_args__ = ( + UniqueConstraint( + "project_id", + "partition_position", + name="uq_accepted_project_note_change_project_position", + ), + Index( + "ix_accepted_project_note_change_project_materialized", + "project_id", + "materialized_at", + ), + Index( + "ix_accepted_project_note_change_note_external_id", + "note_external_id", + ), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + project_id: Mapped[int] = mapped_column( + ForeignKey("project.id", ondelete="CASCADE"), + nullable=False, + ) + project_external_id: Mapped[str] = mapped_column(String, nullable=False) + partition_position: Mapped[int] = mapped_column(Integer, nullable=False) + entity_id: Mapped[int] = mapped_column(Integer, nullable=False) + note_external_id: Mapped[str] = mapped_column(String, nullable=False) + title: Mapped[str] = mapped_column(Text, nullable=False) + operation: Mapped[str] = mapped_column(String, nullable=False) + file_path: Mapped[str] = mapped_column(Text, nullable=False) + previous_file_path: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + accepted_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + source: Mapped[str] = mapped_column(String, nullable=False) + db_version: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) + db_checksum: Mapped[Optional[str]] = mapped_column(String, nullable=True) + actor_user_profile_id: Mapped[Optional[str]] = mapped_column(String, nullable=True) + actor_kind: Mapped[Optional[str]] = mapped_column(String, nullable=True) + actor_name: Mapped[Optional[str]] = mapped_column(String, nullable=True) + materialized_at: Mapped[Optional[datetime]] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + + project = relationship("Project", back_populates="accepted_note_changes") diff --git a/src/basic_memory/repository/project_repository.py b/src/basic_memory/repository/project_repository.py index e5d64dbb3..ef2360629 100644 --- a/src/basic_memory/repository/project_repository.py +++ b/src/basic_memory/repository/project_repository.py @@ -1,16 +1,18 @@ """Repository for managing projects in Basic Memory.""" +from datetime import datetime from pathlib import Path from typing import Any, override, Optional, Sequence, Union from loguru import logger -from sqlalchemy import Executable, inspect as sa_inspect, select, text +from sqlalchemy import Executable, inspect as sa_inspect, select, text, update from sqlalchemy.exc import NoResultFound, OperationalError from sqlalchemy.ext.asyncio import AsyncSession -from basic_memory.models.project import Project +from basic_memory.models.project import AcceptedProjectNoteChange, Project from basic_memory.repository.repository import Repository +from basic_memory.runtime.project_partition import RuntimeAcceptedProjectNoteChange async def _load_sqlite_vec_on_session(session) -> bool: @@ -153,6 +155,99 @@ async def get_default_project(self, session: AsyncSession) -> Optional[Project]: query = self.select().where(Project.is_default.is_(True)) return await self.find_one(session, query) + async def advance_partition_position( + self, + session: AsyncSession, + project_id: int, + ) -> int: + """Atomically claim the next strict position in one project partition.""" + statement = ( + update(Project) + .where(Project.id == project_id) + .values(partition_position=Project.partition_position + 1) + .returning(Project.partition_position) + .execution_options(synchronize_session=False) + ) + position = (await session.execute(statement)).scalar_one_or_none() + if position is None: + raise RuntimeError(f"Project partition is missing for project_id={project_id}") + return int(position) + + async def record_accepted_note_change( + self, + session: AsyncSession, + change: RuntimeAcceptedProjectNoteChange, + ) -> None: + """Persist replayable accepted evidence in the mutation transaction.""" + persisted = AcceptedProjectNoteChange( + project_id=change.project_id, + project_external_id=change.project_external_id, + partition_position=change.partition_position, + entity_id=change.entity_id, + note_external_id=change.note_external_id, + title=change.title, + operation=change.operation.value, + file_path=change.file_path, + previous_file_path=change.previous_file_path, + accepted_at=change.accepted_at, + source=change.source, + db_version=change.db_version, + db_checksum=change.db_checksum, + actor_user_profile_id=( + str(change.actor_user_profile_id) + if change.actor_user_profile_id is not None + else None + ), + actor_kind=change.actor_kind, + actor_name=change.actor_name, + ) + session.add(persisted) + await session.flush() + + async def list_accepted_note_changes( + self, + session: AsyncSession, + project_id: int, + *, + after_position: int = 0, + through_position: int | None = None, + ) -> Sequence[AcceptedProjectNoteChange]: + """List one contiguous project evidence range in strict order.""" + statement = select(AcceptedProjectNoteChange).where( + AcceptedProjectNoteChange.project_id == project_id, + AcceptedProjectNoteChange.partition_position > after_position, + ) + if through_position is not None: + statement = statement.where( + AcceptedProjectNoteChange.partition_position <= through_position + ) + result = await session.execute( + statement.order_by(AcceptedProjectNoteChange.partition_position) + ) + return tuple(result.scalars().all()) + + async def mark_accepted_note_change_materialized( + self, + session: AsyncSession, + project_id: int, + partition_position: int, + *, + materialized_at: datetime, + ) -> bool: + """Record when accepted evidence reached canonical file storage.""" + statement = ( + update(AcceptedProjectNoteChange) + .where( + AcceptedProjectNoteChange.project_id == project_id, + AcceptedProjectNoteChange.partition_position == partition_position, + AcceptedProjectNoteChange.materialized_at.is_(None), + ) + .values(materialized_at=materialized_at) + .returning(AcceptedProjectNoteChange.id) + .execution_options(synchronize_session=False) + ) + return (await session.execute(statement)).scalar_one_or_none() is not None + async def get_active_projects(self, session: AsyncSession) -> Sequence[Project]: """Get all active projects.""" query = self.select().where(Project.is_active == True) # noqa: E712 diff --git a/src/basic_memory/runtime/accepted_note_changes.py b/src/basic_memory/runtime/accepted_note_changes.py index 34b12ba3f..ca7fd4053 100644 --- a/src/basic_memory/runtime/accepted_note_changes.py +++ b/src/basic_memory/runtime/accepted_note_changes.py @@ -29,6 +29,7 @@ RuntimePendingNoteMaterializationSource, plan_pending_note_materialization, ) +from basic_memory.runtime.project_partition import RuntimeAcceptedProjectNoteChange from basic_memory.runtime.storage import ( NoteExternalId, ProjectId, @@ -213,6 +214,7 @@ class RuntimeAcceptedNoteChange(Generic[_PayloadT_co]): status_code: int payload: _PayloadT_co + project_change: RuntimeAcceptedProjectNoteChange | None = None materialization: RuntimePendingNoteMaterialization | None = None file_delete: RuntimePendingNoteFileDelete | None = None # Surviving notes whose relations pointed at a deleted target. Their search diff --git a/src/basic_memory/runtime/cleanup.py b/src/basic_memory/runtime/cleanup.py index 8a195460a..03b9d1d20 100644 --- a/src/basic_memory/runtime/cleanup.py +++ b/src/basic_memory/runtime/cleanup.py @@ -21,6 +21,7 @@ RuntimeFileChecksum, RuntimeFilePath, ) +from basic_memory.runtime.project_partition import RuntimeAcceptedProjectNoteChange RUNTIME_FILE_SNAPSHOT_TIMESTAMP_MATCH_EPSILON_SECONDS = 0.001 @@ -196,6 +197,7 @@ class RuntimeNoteFileDeleteJobRequest: entity_id: RuntimeEntityId file_path: RuntimeFilePath file_checksum: RuntimeFileChecksum | None = None + project_change: RuntimeAcceptedProjectNoteChange | None = None # Live note path after the move that scheduled this cleanup; a local adapter # skips the delete when it shares a physical file with file_path. Not part of # dedupe_key: it does not change the logical identity of the delete. @@ -230,6 +232,7 @@ def plan_note_file_delete_job_request( entity_id=file_delete.entity_id, file_path=file_delete.file_path, file_checksum=file_delete.file_checksum, + project_change=file_delete.project_change, live_file_path=file_delete.live_file_path, ) diff --git a/src/basic_memory/runtime/job_payloads.py b/src/basic_memory/runtime/job_payloads.py index 73519f56d..7b54e65bd 100644 --- a/src/basic_memory/runtime/job_payloads.py +++ b/src/basic_memory/runtime/job_payloads.py @@ -18,6 +18,7 @@ VALID_NOTE_OBJECT_SOURCES, normalize_actor_name, ) +from basic_memory.runtime.project_partition import RuntimeAcceptedProjectNoteChange DELETE_NOTE_FILE_ENTRYPOINT: JobEntrypoint = "delete_note_file" @@ -31,6 +32,7 @@ class RuntimeNoteFileDeleteJobPayload(BaseModel): entity_id: int file_path: str file_checksum: str | None = None + project_change: RuntimeAcceptedProjectNoteChange | None = None @classmethod def from_runtime_request(cls, request: RuntimeNoteFileDeleteJobRequest) -> Self: @@ -40,6 +42,7 @@ def from_runtime_request(cls, request: RuntimeNoteFileDeleteJobRequest) -> Self: entity_id=request.entity_id, file_path=request.file_path, file_checksum=request.file_checksum, + project_change=request.project_change, ) def to_runtime_request(self) -> RuntimeNoteFileDeleteJobRequest: @@ -49,6 +52,7 @@ def to_runtime_request(self) -> RuntimeNoteFileDeleteJobRequest: entity_id=self.entity_id, file_path=self.file_path, file_checksum=self.file_checksum, + project_change=self.project_change, ) def runtime_job_request( @@ -72,6 +76,7 @@ class RuntimeNoteMaterializationJobPayload(BaseModel): entity_id: int db_version: int db_checksum: str + project_change: RuntimeAcceptedProjectNoteChange | None = None actor_user_profile_id: UUID | None = None actor_kind: str | None = None actor_name: str | None = None @@ -122,6 +127,7 @@ def from_runtime_request(cls, request: RuntimeNoteMaterializationJobRequest) -> entity_id=request.entity_id, db_version=request.db_version, db_checksum=request.db_checksum, + project_change=request.project_change, actor_user_profile_id=request.actor_user_profile_id, actor_kind=request.actor_kind, actor_name=request.actor_name, @@ -138,6 +144,7 @@ def to_runtime_request(self) -> RuntimeNoteMaterializationJobRequest: entity_id=self.entity_id, db_version=self.db_version, db_checksum=self.db_checksum, + project_change=self.project_change, actor_user_profile_id=self.actor_user_profile_id, actor_kind=self.actor_kind, actor_name=self.actor_name, diff --git a/src/basic_memory/runtime/note_content_deletes.py b/src/basic_memory/runtime/note_content_deletes.py index 2f4337156..5d8bd6126 100644 --- a/src/basic_memory/runtime/note_content_deletes.py +++ b/src/basic_memory/runtime/note_content_deletes.py @@ -14,6 +14,7 @@ RuntimeFilePath, runtime_content_type_is_markdown, ) +from basic_memory.runtime.project_partition import RuntimeAcceptedProjectNoteChange class RuntimeDeletedNoteEntitySource(RuntimeContentTypeSource, Protocol): @@ -222,6 +223,7 @@ class RuntimePendingNoteFileDelete: entity_id: RuntimeEntityId file_path: RuntimeFilePath file_checksum: RuntimeFileChecksum | None = None + project_change: RuntimeAcceptedProjectNoteChange | None = None # The note's live path after the move that scheduled this cleanup. Object # storage treats case-different keys as distinct; a local adapter re-checks # it against the physical filesystem before deleting because a case-only diff --git a/src/basic_memory/runtime/note_materialization_planning.py b/src/basic_memory/runtime/note_materialization_planning.py index efe768fc6..5935f14b1 100644 --- a/src/basic_memory/runtime/note_materialization_planning.py +++ b/src/basic_memory/runtime/note_materialization_planning.py @@ -9,6 +9,7 @@ from uuid import UUID from basic_memory.runtime.note_content_deletes import RuntimePendingNoteFileDelete +from basic_memory.runtime.project_partition import RuntimeAcceptedProjectNoteChange from basic_memory.runtime.storage import ( ProjectId, RuntimeEntityId, @@ -93,6 +94,7 @@ class RuntimePendingNoteMaterialization: entity_id: RuntimeEntityId db_version: RuntimeNoteContentVersion db_checksum: RuntimeNoteContentChecksum + project_change: RuntimeAcceptedProjectNoteChange | None = None actor_user_profile_id: UUID | None = None actor_kind: RuntimeNoteActorKind | None = None actor_name: RuntimeNoteActorName | None = None @@ -137,6 +139,7 @@ class RuntimeNoteMaterializationJobRequest: entity_id: RuntimeEntityId db_version: RuntimeNoteContentVersion db_checksum: RuntimeNoteContentChecksum + project_change: RuntimeAcceptedProjectNoteChange | None = None actor_user_profile_id: UUID | None = None actor_kind: RuntimeNoteActorKind | None = None actor_name: RuntimeNoteActorName | None = None @@ -194,6 +197,7 @@ def plan_note_materialization_job_request( entity_id=materialization.entity_id, db_version=materialization.db_version, db_checksum=materialization.db_checksum, + project_change=materialization.project_change, actor_user_profile_id=materialization.actor_user_profile_id, actor_kind=materialization.actor_kind, actor_name=materialization.actor_name, diff --git a/src/basic_memory/runtime/note_object_metadata.py b/src/basic_memory/runtime/note_object_metadata.py index 350cd7947..ca77324ef 100644 --- a/src/basic_memory/runtime/note_object_metadata.py +++ b/src/basic_memory/runtime/note_object_metadata.py @@ -23,6 +23,7 @@ NOTE_OBJECT_ACTOR_KIND_METADATA = "bm-actor-kind" NOTE_OBJECT_ACTOR_KIND_MCP_CLIENT = "mcp_client" +NOTE_OBJECT_ACTOR_KIND_SYSTEM = "system" NOTE_OBJECT_ACTOR_NAME_METADATA = "bm-actor-name" NOTE_OBJECT_ACTOR_USER_PROFILE_ID_METADATA = "bm-actor-user-profile-id" NOTE_OBJECT_DB_CHECKSUM_METADATA = "bm-db-checksum" @@ -32,7 +33,7 @@ NOTE_OBJECT_FILE_VERSION_METADATA = "bm-file-version" NOTE_OBJECT_SOURCE_METADATA = "bm-note-source" VALID_NOTE_OBJECT_ACTOR_KINDS: frozenset[RuntimeNoteActorKind] = frozenset( - {NOTE_OBJECT_ACTOR_KIND_MCP_CLIENT} + {NOTE_OBJECT_ACTOR_KIND_MCP_CLIENT, NOTE_OBJECT_ACTOR_KIND_SYSTEM} ) # web_v2 = a note write originating from the web-v2 UI. Distinguishing it from # `api` lets clients tell a genuine web-UI edit apart from api/materialization @@ -42,8 +43,18 @@ # note.updated event echoes this source as the write's actor origin. # document_ingestion = a hosted worker accepting a deterministic document # extraction or ingestion-run note through the canonical note mutation path. +# wiki_projector = the deterministic OKF projector accepting generated index +# and log notes through that same path before materializing them as Markdown. VALID_NOTE_OBJECT_SOURCES: frozenset[RuntimeNoteChangeSource] = frozenset( - {"api", "collaboration_relay", "document_ingestion", "mcp", "s3_webhook", "web_v2"} + { + "api", + "collaboration_relay", + "document_ingestion", + "mcp", + "s3_webhook", + "web_v2", + "wiki_projector", + } ) # Named because the accepted-note write path special-cases relay writes: the # relay superseding its own prior write is never a real conflict (#1589). diff --git a/src/basic_memory/runtime/project_partition.py b/src/basic_memory/runtime/project_partition.py new file mode 100644 index 000000000..050ab9622 --- /dev/null +++ b/src/basic_memory/runtime/project_partition.py @@ -0,0 +1,80 @@ +"""Portable evidence for one accepted change in a strict project partition.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from enum import StrEnum +from uuid import UUID + +from basic_memory.runtime.storage import ( + NoteExternalId, + ProjectExternalId, + ProjectId, + RuntimeEntityId, + RuntimeFilePath, + RuntimeNoteActorKind, + RuntimeNoteActorName, + RuntimeNoteChangeSource, + RuntimeNoteContentChecksum, + RuntimeNoteContentVersion, +) + +type ProjectPartitionPosition = int + + +class RuntimeProjectNoteOperation(StrEnum): + """Accepted note operation recorded for project-wide consumers.""" + + created = "created" + updated = "updated" + moved = "moved" + deleted = "deleted" + + +@dataclass(frozen=True, slots=True) +class RuntimeAcceptedProjectNoteChange: + """Replay-complete accepted note evidence carried to runtime follow-ups.""" + + project_id: ProjectId + project_external_id: ProjectExternalId + partition_position: ProjectPartitionPosition + entity_id: RuntimeEntityId + note_external_id: NoteExternalId + title: str + operation: RuntimeProjectNoteOperation + file_path: RuntimeFilePath + accepted_at: datetime + source: RuntimeNoteChangeSource + previous_file_path: RuntimeFilePath | None = None + db_version: RuntimeNoteContentVersion | None = None + db_checksum: RuntimeNoteContentChecksum | None = None + actor_user_profile_id: UUID | None = None + actor_kind: RuntimeNoteActorKind | None = None + actor_name: RuntimeNoteActorName | None = None + + def __post_init__(self) -> None: + if self.partition_position <= 0: + raise ValueError("Accepted project change position must be positive") + if not self.project_external_id.strip(): + raise ValueError("Accepted project change requires project_external_id") + if not self.note_external_id.strip(): + raise ValueError("Accepted project change requires note_external_id") + if not self.title.strip(): + raise ValueError("Accepted project change requires title") + if not self.file_path.strip(): + raise ValueError("Accepted project change requires file_path") + if self.accepted_at.tzinfo is None: + raise ValueError("Accepted project change accepted_at must be timezone-aware") + if not self.source.strip(): + raise ValueError("Accepted project change requires source") + if (self.db_version is None) != (self.db_checksum is None): + raise ValueError( + "Accepted project change revision requires both db_version and db_checksum" + ) + if self.db_version is not None and self.db_version <= 0: + raise ValueError("Accepted project change db_version must be positive") + if self.db_checksum is not None and not self.db_checksum.strip(): + raise ValueError("Accepted project change db_checksum must not be empty") + if self.operation == RuntimeProjectNoteOperation.moved and not self.previous_file_path: + raise ValueError("Moved accepted project change requires previous_file_path") diff --git a/tests/api/v2/test_accepted_note_atomicity.py b/tests/api/v2/test_accepted_note_atomicity.py index 339f3a409..b57f47d75 100644 --- a/tests/api/v2/test_accepted_note_atomicity.py +++ b/tests/api/v2/test_accepted_note_atomicity.py @@ -11,7 +11,14 @@ from basic_memory import db from basic_memory.deps.services import get_note_content_materialization_provider -from basic_memory.models import Entity, NoteContent, Observation, Project, Relation +from basic_memory.models import ( + AcceptedProjectNoteChange, + Entity, + NoteContent, + Observation, + Project, + Relation, +) from basic_memory.runtime.note_content import ( RuntimeAcceptedNoteChange, RuntimeNoteContentResponsePayload, @@ -24,6 +31,8 @@ class PersistedAcceptedSnapshot: entity: Entity note_content: NoteContent + project_partition_position: int + accepted_project_changes: tuple[AcceptedProjectNoteChange, ...] observations: tuple[Observation, ...] relations: tuple[Relation, ...] search_content: str @@ -36,6 +45,7 @@ async def _load_persisted_snapshot( entity_id: int, ) -> PersistedAcceptedSnapshot: async with db.scoped_session(session_maker) as session: + project = await session.get(Project, project_id) entity = await session.get(Entity, entity_id) note_content = await session.get(NoteContent, entity_id) observations = tuple( @@ -62,6 +72,15 @@ async def _load_persisted_snapshot( ) ).all() ) + accepted_project_changes = tuple( + ( + await session.scalars( + select(AcceptedProjectNoteChange) + .where(AcceptedProjectNoteChange.project_id == project_id) + .order_by(AcceptedProjectNoteChange.partition_position) + ) + ).all() + ) search_content = ( await session.execute( text(""" @@ -77,9 +96,12 @@ async def _load_persisted_snapshot( assert entity is not None assert note_content is not None + assert project is not None return PersistedAcceptedSnapshot( entity=entity, note_content=note_content, + project_partition_position=project.partition_position, + accepted_project_changes=accepted_project_changes, observations=observations, relations=relations, search_content=str(search_content), @@ -164,6 +186,11 @@ async def test_create_and_update_persist_complete_snapshot_at_materialization_bo assert created_snapshot.note_content.markdown_content == created.content assert created_snapshot.note_content.db_version == 1 assert created_snapshot.note_content.file_write_status == "pending" + assert created_snapshot.project_partition_position == 1 + assert [change.partition_position for change in created_snapshot.accepted_project_changes] == [ + 1 + ] + assert created_snapshot.accepted_project_changes[0].materialized_at is None assert [observation.content for observation in created_snapshot.observations] == [ "Create snapshot observation" ] @@ -199,6 +226,11 @@ async def test_create_and_update_persist_complete_snapshot_at_materialization_bo assert updated_snapshot.note_content.markdown_content == updated.content assert updated_snapshot.note_content.db_version == 2 assert updated_snapshot.note_content.file_write_status == "pending" + assert updated_snapshot.project_partition_position == 2 + assert [change.partition_position for change in updated_snapshot.accepted_project_changes] == [ + 1, + 2, + ] assert [observation.content for observation in updated_snapshot.observations] == [ "Replacing update observation" ] @@ -208,3 +240,12 @@ async def test_create_and_update_persist_complete_snapshot_at_materialization_bo assert "Replacing update observation" in updated_snapshot.search_content assert "Create snapshot observation" not in updated_snapshot.search_content assert len(materializer.accepted_changes) == 2 + created_change, updated_change = materializer.accepted_changes + assert created_change.project_change is not None + assert created_change.project_change.partition_position == 1 + assert created_change.materialization is not None + assert created_change.materialization.project_change is created_change.project_change + assert updated_change.project_change is not None + assert updated_change.project_change.partition_position == 2 + assert updated_change.materialization is not None + assert updated_change.materialization.project_change is updated_change.project_change diff --git a/tests/indexing/test_accepted_note_mutation_runner.py b/tests/indexing/test_accepted_note_mutation_runner.py index ce1f1683c..2a622a37d 100644 --- a/tests/indexing/test_accepted_note_mutation_runner.py +++ b/tests/indexing/test_accepted_note_mutation_runner.py @@ -50,6 +50,10 @@ from basic_memory.repository.observation_repository import ObservationGenerationWriteResult from basic_memory.repository.relation_repository import RelationGenerationWriteResult from basic_memory.runtime.note_content import RuntimeAcceptedNoteResponse +from basic_memory.runtime.project_partition import ( + RuntimeAcceptedProjectNoteChange, + RuntimeProjectNoteOperation, +) from basic_memory.schemas.base import Entity as EntitySchema from basic_memory.schemas.request import EditEntityRequest from basic_memory.services.exceptions import EntityAlreadyExistsError @@ -331,9 +335,12 @@ async def load_current_file_checksum(self, project: Project, file_path: str) -> class _ProjectRepository: - def __init__(self, project: Project | None) -> None: + def __init__(self, project: Project | None, *, next_partition_position: int = 1) -> None: self.project = project self.calls: list[tuple[AsyncSession, str]] = [] + self.partition_calls: list[tuple[AsyncSession, int]] = [] + self.recorded_changes: list[RuntimeAcceptedProjectNoteChange] = [] + self.next_partition_position = next_partition_position async def get_by_external_id( self, @@ -343,6 +350,24 @@ async def get_by_external_id( self.calls.append((session, external_id)) return self.project + async def advance_partition_position( + self, + session: AsyncSession, + project_id: int, + ) -> int: + self.partition_calls.append((session, project_id)) + position = self.next_partition_position + self.next_partition_position += 1 + return position + + async def record_accepted_note_change( + self, + session: AsyncSession, + change: RuntimeAcceptedProjectNoteChange, + ) -> None: + _ = session + self.recorded_changes.append(change) + class _EntityLookupRepository: def __init__( @@ -772,6 +797,24 @@ async def test_run_accepted_note_create_persists_prepared_markdown( assert change.materialization.actor_kind == "user" assert change.materialization.actor_name == "Ada" assert change.materialization.previous_file_path is None + assert project_repository.partition_calls == [(session, project.id)] + assert change.project_change is not None + project_change = change.project_change + assert project_change.partition_position == 1 + assert project_change.operation is RuntimeProjectNoteOperation.created + assert project_change.project_external_id == "project-123" + assert project_change.note_external_id == "note-123" + assert project_change.file_path == "notes/accepted.md" + assert project_change.previous_file_path is None + assert project_change.accepted_at == _NOW + assert project_change.source == "api" + assert project_change.db_version == 1 + assert project_change.db_checksum == note_content.db_checksum + assert project_change.actor_user_profile_id == _ACTOR_ID + assert project_change.actor_kind == "user" + assert project_change.actor_name == "Ada" + assert project_repository.recorded_changes == [project_change] + assert change.materialization.project_change is project_change assert result.relation_publication is not None assert result.relation_publication.generation == 1 assert persistence_calls[0].await_count == 1 @@ -910,6 +953,13 @@ async def test_run_accepted_note_update_replaces_existing_note_content( assert change.materialization is not None assert change.materialization.db_version == 2 assert change.materialization.previous_file_path is None + assert project_repository.partition_calls == [(cast(AsyncSession, session), project.id)] + assert change.project_change is not None + assert change.project_change.operation is RuntimeProjectNoteOperation.moved + assert change.project_change.previous_file_path == "notes/accepted.md" + assert change.project_change.file_path == "notes/replacement.md" + assert change.project_change.db_version == 2 + assert change.materialization.project_change is change.project_change assert result.relation_publication is not None assert result.relation_publication.generation == 2 assert persistence_calls[0].await_count == 1 @@ -1486,6 +1536,12 @@ async def test_run_accepted_note_edit_applies_patch_against_db_content( assert change.status_code == 200 assert change.materialization is not None assert change.materialization.source == "mcp" + assert project_repository.partition_calls == [(cast(AsyncSession, session), project.id)] + assert change.project_change is not None + assert change.project_change.operation is RuntimeProjectNoteOperation.updated + assert change.project_change.previous_file_path is None + assert change.project_change.actor_user_profile_id is None + assert change.materialization.project_change is change.project_change assert persistence_calls[0].await_count == 1 assert persistence_calls[1].await_count == 0 @@ -1646,6 +1702,15 @@ async def test_run_accepted_note_move_carries_previous_path_and_materialized_cle assert change.status_code == 200 assert change.materialization is not None assert change.materialization.previous_file_path == "notes/accepted.md" + assert project_repository.partition_calls == [(cast(AsyncSession, session), project.id)] + assert change.project_change is not None + assert change.project_change.operation is RuntimeProjectNoteOperation.moved + assert change.project_change.previous_file_path == "notes/accepted.md" + assert change.project_change.file_path == "archive/accepted.md" + assert change.project_change.actor_user_profile_id == _ACTOR_ID + assert change.project_change.actor_kind == "mcp" + assert change.project_change.actor_name == "Claude" + assert change.materialization.project_change is change.project_change cleanup = change.materialization.cleanup_after_write if expected_source_checksum is None: assert cleanup is None @@ -1992,6 +2057,15 @@ async def test_run_accepted_note_delete_removes_entity_and_returns_cleanup() -> assert change.file_delete is not None assert change.file_delete.file_path == "notes/accepted.md" assert change.file_delete.file_checksum == "file-checksum" + assert project_repository.partition_calls == [(cast(AsyncSession, session), project.id)] + assert change.project_change is not None + assert change.project_change.operation is RuntimeProjectNoteOperation.deleted + assert change.project_change.file_path == "notes/accepted.md" + assert change.project_change.source == "delete_note" + assert change.project_change.db_version == note_content.db_version + assert change.project_change.db_checksum == note_content.db_checksum + assert change.project_change.actor_user_profile_id is None + assert change.file_delete.project_change is change.project_change assert change.relation_cleanup_entity_ids == frozenset() assert result.relation_publication is None @@ -2073,6 +2147,57 @@ async def test_run_accepted_note_create_returns_graph_publication() -> None: assert result.relation_publication.relations[0].target_name == "XSYS Target" +@pytest.mark.asyncio +async def test_run_accepted_note_create_can_suppress_derived_graph_facts() -> None: + """Derived documents keep their Markdown without recursively expanding the graph.""" + session = cast(AsyncSession, object()) + prepared = _prepared_with_graph( + observations=[ + AcceptedObservationWrite( + content="Generated list item", + category="note", + context=None, + tags=None, + ) + ], + relations=[ + AcceptedRelationWrite( + relation_type="links_to", + target_name="Source Note", + context=None, + ) + ], + ) + entity = _entity() + note_content = _note_content(entity) + + result = await run_accepted_note_create( + session, + request=AcceptedNoteCreateMutation( + project_external_id="project-123", + data=_schema(), + actor=AcceptedNoteMutationActor(user_profile_id=None, kind="system"), + source="wiki_projector", + publish_graph_facts=False, + ), + dependencies=_dependencies( + project_repository=_ProjectRepository(_project()), + entity_lookup_repository=_EntityLookupRepository(), + note_content_lookup_repository=_NoteContentLookupRepository(), + preparer_factory=_PreparerFactory(_CreatePreparer(prepared)), + pending_entity_repository=_PendingEntityRepository(entity), + note_content_accept_repository=_NoteContentAcceptRepository(note_content), + search_repository=_SearchRepository(), + ), + ) + + assert isinstance(result.change.payload, RuntimeAcceptedNoteResponse) + assert result.change.payload.markdown_content == "# Accepted\n" + assert result.relation_publication is not None + assert result.relation_publication.observations == () + assert result.relation_publication.relations == () + + @pytest.mark.asyncio async def test_run_accepted_note_create_pre_resolves_only_unambiguous_self_links() -> None: """Safe self aliases resolve inline while ambiguous title aliases stay deferred.""" @@ -2177,6 +2302,58 @@ async def test_run_accepted_note_update_returns_replacement_graph() -> None: assert result.relation_publication.relations[0].target_name == "Other" +@pytest.mark.asyncio +async def test_run_accepted_note_update_can_clear_derived_graph_facts() -> None: + """A graph-silent replacement publishes empty sets so earlier facts are removed.""" + session = _MutationSession() + prepared = _prepared_with_graph( + observations=[ + AcceptedObservationWrite( + content="Generated list item", + category="note", + context=None, + tags=None, + ) + ], + relations=[ + AcceptedRelationWrite( + relation_type="links_to", + target_name="Source Note", + context=None, + ) + ], + ) + entity = _entity(file_path="notes/accepted.md") + note_content = _note_content(entity) + + result = await run_accepted_note_update( + cast(AsyncSession, session), + request=AcceptedNoteUpdateMutation( + project_external_id="project-123", + entity_external_id="note-123", + data=_schema(), + actor=AcceptedNoteMutationActor(user_profile_id=None, kind="system"), + source="wiki_projector", + publish_graph_facts=False, + ), + dependencies=_dependencies( + project_repository=_ProjectRepository(_project()), + entity_lookup_repository=_EntityLookupRepository(by_external_id=entity), + note_content_lookup_repository=_NoteContentLookupRepository(note_content), + preparer_factory=_PreparerFactory(_CreatePreparer(prepared)), + pending_entity_repository=_PendingEntityRepository(entity), + note_content_accept_repository=_NoteContentAcceptRepository(note_content), + search_repository=_SearchRepository(), + ), + ) + + assert isinstance(result.change.payload, RuntimeAcceptedNoteResponse) + assert result.change.payload.markdown_content == "# Accepted\n" + assert result.relation_publication is not None + assert result.relation_publication.observations == () + assert result.relation_publication.relations == () + + @pytest.mark.asyncio async def test_run_accepted_note_edit_returns_empty_replacement_graph() -> None: """An edit that drops the graph returns empty sets for fenced cleanup.""" diff --git a/tests/repository/test_project_partition_repository.py b/tests/repository/test_project_partition_repository.py new file mode 100644 index 000000000..9bc7e9ad4 --- /dev/null +++ b/tests/repository/test_project_partition_repository.py @@ -0,0 +1,147 @@ +"""Database regressions for strict project partition positions.""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from basic_memory import db +from basic_memory.models import AcceptedProjectNoteChange, Project +from basic_memory.repository.project_repository import ProjectRepository +from basic_memory.runtime.project_partition import ( + RuntimeAcceptedProjectNoteChange, + RuntimeProjectNoteOperation, +) + + +_ACCEPTED_AT = datetime(2026, 8, 29, 20, 15, tzinfo=UTC) + + +def _accepted_change( + project: Project, + *, + partition_position: int, +) -> RuntimeAcceptedProjectNoteChange: + return RuntimeAcceptedProjectNoteChange( + project_id=project.id, + project_external_id=project.external_id, + partition_position=partition_position, + entity_id=42, + note_external_id="note-42", + title="Accepted evidence", + operation=RuntimeProjectNoteOperation.updated, + file_path="notes/accepted-evidence.md", + accepted_at=_ACCEPTED_AT, + source="api", + db_version=3, + db_checksum="accepted-checksum", + ) + + +@pytest.mark.asyncio +async def test_project_partition_positions_advance_in_transaction_order( + session_maker: async_sessionmaker[AsyncSession], + test_project: Project, +) -> None: + repository = ProjectRepository() + + async with db.scoped_session(session_maker) as session: + first = await repository.advance_partition_position(session, test_project.id) + second = await repository.advance_partition_position(session, test_project.id) + + assert (first, second) == (1, 2) + async with db.scoped_session(session_maker) as session: + persisted = await session.get(Project, test_project.id) + assert persisted is not None + assert persisted.partition_position == 2 + + +@pytest.mark.asyncio +async def test_accepted_project_note_change_is_replayable_and_materialization_aware( + session_maker: async_sessionmaker[AsyncSession], + test_project: Project, +) -> None: + repository = ProjectRepository() + + async with db.scoped_session(session_maker) as session: + position = await repository.advance_partition_position(session, test_project.id) + await repository.record_accepted_note_change( + session, + _accepted_change(test_project, partition_position=position), + ) + + async with db.scoped_session(session_maker) as session: + changes = await repository.list_accepted_note_changes( + session, + test_project.id, + after_position=0, + through_position=1, + ) + assert len(changes) == 1 + assert changes[0].operation == RuntimeProjectNoteOperation.updated.value + assert changes[0].db_checksum == "accepted-checksum" + assert changes[0].materialized_at is None + assert await repository.mark_accepted_note_change_materialized( + session, + test_project.id, + 1, + materialized_at=_ACCEPTED_AT, + ) + + async with db.scoped_session(session_maker) as session: + [materialized] = await repository.list_accepted_note_changes( + session, + test_project.id, + ) + assert materialized.materialized_at is not None + assert not await repository.mark_accepted_note_change_materialized( + session, + test_project.id, + 1, + materialized_at=_ACCEPTED_AT, + ) + + +@pytest.mark.asyncio +async def test_project_partition_position_rolls_back_with_rejected_transaction( + session_maker: async_sessionmaker[AsyncSession], + test_project: Project, +) -> None: + repository = ProjectRepository() + + with pytest.raises(RuntimeError, match="reject accepted mutation"): + async with db.scoped_session(session_maker) as session: + position = await repository.advance_partition_position(session, test_project.id) + await repository.record_accepted_note_change( + session, + _accepted_change(test_project, partition_position=position), + ) + raise RuntimeError("reject accepted mutation") + + async with db.scoped_session(session_maker) as session: + persisted = await session.get(Project, test_project.id) + assert persisted is not None + assert persisted.partition_position == 0 + rolled_back_changes = ( + await session.scalars( + select(AcceptedProjectNoteChange).where( + AcceptedProjectNoteChange.project_id == test_project.id + ) + ) + ).all() + assert rolled_back_changes == [] + assert await repository.advance_partition_position(session, test_project.id) == 1 + + +@pytest.mark.asyncio +async def test_project_partition_advance_rejects_missing_project( + session_maker: async_sessionmaker[AsyncSession], +) -> None: + repository = ProjectRepository() + + with pytest.raises(RuntimeError, match="project_id=999999"): + async with db.scoped_session(session_maker) as session: + await repository.advance_partition_position(session, 999999) diff --git a/tests/runtime/test_project_partition.py b/tests/runtime/test_project_partition.py new file mode 100644 index 000000000..fe9f94c31 --- /dev/null +++ b/tests/runtime/test_project_partition.py @@ -0,0 +1,108 @@ +"""Portable project partition evidence and propagation tests.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from uuid import UUID + +import pytest + +from basic_memory.runtime.cleanup import plan_note_file_delete_job_request +from basic_memory.runtime.note_content_deletes import RuntimePendingNoteFileDelete +from basic_memory.runtime.note_materialization_planning import ( + RuntimePendingNoteMaterialization, + plan_note_materialization_job_request, +) +from basic_memory.runtime.project_partition import ( + RuntimeAcceptedProjectNoteChange, + RuntimeProjectNoteOperation, +) + + +def _project_change( + operation: RuntimeProjectNoteOperation = RuntimeProjectNoteOperation.updated, +) -> RuntimeAcceptedProjectNoteChange: + return RuntimeAcceptedProjectNoteChange( + project_id=7, + project_external_id="project-123", + partition_position=4, + entity_id=42, + note_external_id="note-123", + title="Accepted", + operation=operation, + file_path="notes/accepted.md", + accepted_at=datetime(2026, 8, 29, 12, tzinfo=UTC), + source="api", + previous_file_path=( + "notes/old.md" if operation is RuntimeProjectNoteOperation.moved else None + ), + db_version=3, + db_checksum="db-checksum", + actor_user_profile_id=UUID("11111111-1111-4111-8111-111111111111"), + actor_kind="user", + actor_name="Ada", + ) + + +def test_project_change_requires_complete_revision_identity() -> None: + with pytest.raises(ValueError, match="both db_version and db_checksum"): + RuntimeAcceptedProjectNoteChange( + project_id=7, + project_external_id="project-123", + partition_position=1, + entity_id=42, + note_external_id="note-123", + title="Accepted", + operation=RuntimeProjectNoteOperation.updated, + file_path="notes/accepted.md", + accepted_at=datetime(2026, 8, 29, 12, tzinfo=UTC), + source="api", + db_version=3, + ) + + +def test_moved_project_change_requires_previous_path() -> None: + with pytest.raises(ValueError, match="requires previous_file_path"): + RuntimeAcceptedProjectNoteChange( + project_id=7, + project_external_id="project-123", + partition_position=1, + entity_id=42, + note_external_id="note-123", + title="Accepted", + operation=RuntimeProjectNoteOperation.moved, + file_path="notes/accepted.md", + accepted_at=datetime(2026, 8, 29, 12, tzinfo=UTC), + source="api", + ) + + +def test_project_change_survives_materialization_job_flattening() -> None: + project_change = _project_change() + request = plan_note_materialization_job_request( + RuntimePendingNoteMaterialization( + project_id=7, + entity_id=42, + db_version=3, + db_checksum="db-checksum", + project_change=project_change, + source="api", + ) + ) + + assert request.project_change is project_change + + +def test_project_change_survives_file_delete_job_flattening() -> None: + project_change = _project_change(RuntimeProjectNoteOperation.deleted) + request = plan_note_file_delete_job_request( + RuntimePendingNoteFileDelete( + project_id=7, + entity_id=42, + file_path="notes/accepted.md", + file_checksum="file-checksum", + project_change=project_change, + ) + ) + + assert request.project_change is project_change diff --git a/tests/runtime/test_runtime_job_payloads.py b/tests/runtime/test_runtime_job_payloads.py index 0b5918b5c..6f6897a77 100644 --- a/tests/runtime/test_runtime_job_payloads.py +++ b/tests/runtime/test_runtime_job_payloads.py @@ -1,9 +1,11 @@ """Tests for portable runtime worker payload boundaries.""" +from datetime import UTC, datetime from uuid import UUID import pytest +from basic_memory.indexing.wiki_projector import WIKI_PROJECTOR_SOURCE from basic_memory.runtime.cleanup import RuntimeNoteFileDeleteJobRequest from basic_memory.runtime.job_payloads import ( DELETE_NOTE_FILE_ENTRYPOINT, @@ -13,7 +15,34 @@ ) from basic_memory.runtime.jobs import RuntimeJobRequest from basic_memory.runtime.note_content import RuntimeNoteMaterializationJobRequest -from basic_memory.runtime.note_object_metadata import NOTE_OBJECT_ACTOR_KIND_MCP_CLIENT +from basic_memory.runtime.note_object_metadata import ( + NOTE_OBJECT_ACTOR_KIND_MCP_CLIENT, + NOTE_OBJECT_ACTOR_KIND_SYSTEM, +) +from basic_memory.runtime.project_partition import ( + RuntimeAcceptedProjectNoteChange, + RuntimeProjectNoteOperation, +) + + +def _project_change() -> RuntimeAcceptedProjectNoteChange: + return RuntimeAcceptedProjectNoteChange( + project_id=101, + project_external_id="project-123", + partition_position=7, + entity_id=42, + note_external_id="note-123", + title="A", + operation=RuntimeProjectNoteOperation.updated, + file_path="notes/a.md", + accepted_at=datetime(2026, 8, 29, 12, tzinfo=UTC), + source="mcp", + db_version=4, + db_checksum="db-sum", + actor_user_profile_id=UUID("33333333-3333-4333-8333-333333333333"), + actor_kind=NOTE_OBJECT_ACTOR_KIND_MCP_CLIENT, + actor_name="Claude Code", + ) def test_runtime_note_file_delete_job_payload_round_trips_runtime_request() -> None: @@ -23,6 +52,7 @@ def test_runtime_note_file_delete_job_payload_round_trips_runtime_request() -> N entity_id=42, file_path="notes/a.md", file_checksum="file-sum", + project_change=_project_change(), ) payload = RuntimeNoteFileDeleteJobPayload.from_runtime_request(runtime_request) @@ -62,6 +92,7 @@ def test_runtime_note_materialization_job_payload_round_trips_runtime_request() entity_id=42, db_version=4, db_checksum="db-sum", + project_change=_project_change(), actor_user_profile_id=UUID("33333333-3333-3333-3333-333333333333"), actor_kind=NOTE_OBJECT_ACTOR_KIND_MCP_CLIENT, actor_name="Claude Code", @@ -119,6 +150,23 @@ def test_runtime_note_materialization_job_payload_normalizes_origin_fields() -> assert payload.source == "mcp" +def test_runtime_note_materialization_job_payload_accepts_wiki_projector_source() -> None: + """Generated OKF notes preserve their projector source through materialization.""" + payload = RuntimeNoteMaterializationJobPayload( + project_id=101, + entity_id=42, + db_version=4, + db_checksum="db-sum", + actor_kind=NOTE_OBJECT_ACTOR_KIND_SYSTEM, + actor_name="Basic Memory Wiki Projector", + source=WIKI_PROJECTOR_SOURCE, + ) + + assert payload.actor_kind == NOTE_OBJECT_ACTOR_KIND_SYSTEM + assert payload.actor_name == "Basic Memory Wiki Projector" + assert payload.source == WIKI_PROJECTOR_SOURCE + + def test_runtime_note_materialization_job_payload_rejects_unknown_origin_fields() -> None: """Bad queued origins should fail before they become materialized file metadata.""" with pytest.raises(ValueError, match="unsupported note materialization actor kind"): diff --git a/tests/test_project_partition_migration.py b/tests/test_project_partition_migration.py new file mode 100644 index 000000000..c71e2a37e --- /dev/null +++ b/tests/test_project_partition_migration.py @@ -0,0 +1,89 @@ +"""Migration regressions for accepted project change storage.""" + +import sqlite3 +from pathlib import Path + +from alembic import command +from alembic.config import Config +from alembic.script import ScriptDirectory + +from basic_memory import db + + +def _sqlite_alembic_config(database_path: Path) -> Config: + alembic_dir = Path(db.__file__).parent / "alembic" + config = Config() + config.set_main_option("script_location", str(alembic_dir)) + config.set_main_option("revision_environment", "false") + config.set_main_option("sqlalchemy.url", f"sqlite:///{database_path}") + return config + + +def test_upgrade_repairs_stamped_project_partition_without_change_table( + tmp_path: Path, + monkeypatch, +) -> None: + """A tenant stamped at the pre-release revision receives its missing journal.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("BASIC_MEMORY_HOME", str(tmp_path / "basic-memory")) + database_path = tmp_path / "project-partition-repair.db" + config = _sqlite_alembic_config(database_path) + + command.upgrade(config, "bcdbd5a942ca") + connection = sqlite3.connect(database_path) + try: + connection.execute( + "ALTER TABLE project ADD COLUMN partition_position INTEGER NOT NULL DEFAULT 0" + ) + connection.commit() + finally: + connection.close() + command.stamp(config, "s2p3e4c5w6k7") + + command.upgrade(config, "head") + + connection = sqlite3.connect(database_path) + try: + columns = { + row[1] + for row in connection.execute( + "PRAGMA table_info(accepted_project_note_change)" + ).fetchall() + } + indexes = { + row[1] + for row in connection.execute( + "PRAGMA index_list(accepted_project_note_change)" + ).fetchall() + } + project_columns = { + row[1] for row in connection.execute("PRAGMA table_info(project)").fetchall() + } + version = connection.execute("SELECT version_num FROM alembic_version").fetchone() + finally: + connection.close() + + assert columns == { + "id", + "project_id", + "project_external_id", + "partition_position", + "entity_id", + "note_external_id", + "title", + "operation", + "file_path", + "previous_file_path", + "accepted_at", + "source", + "db_version", + "db_checksum", + "actor_user_profile_id", + "actor_kind", + "actor_name", + "materialized_at", + } + assert "partition_position" in project_columns + assert "ix_accepted_project_note_change_project_materialized" in indexes + assert "ix_accepted_project_note_change_note_external_id" in indexes + assert version == (ScriptDirectory.from_config(config).get_current_head(),)