fix(memory): enforce authoritative revision integrity

src/powercontext/builtin/artifacts/memory/service.py

@@ -930,7 +930,10 @@ async def _validated_entries(self, memory: Memory) -> tuple[MemoryEntryVersion,
             if version.memory_artifact_id != memory.artifact_id or version.entry_id != item.entry_id:
                 raise InvalidMemoryCitationError("cross-identity")
             material = self._material_from_version(version)
-            if material.content_hash != item.entry_content_hash:
+            if (
+                material.content_hash != item.entry_content_hash
+                or version.entry_content_hash != item.entry_content_hash
+            ):
                 raise InvalidMemoryCitationError("hash-mismatch")
             ordered.append(version)
         return tuple(ordered)

src/powercontext/builtin/persistence/memory_schema.py

@@ -0,0 +1,92 @@
+# Copyright (c) 2026 OceanBase.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Schema upgrades for authoritative Memory entry history."""
+
+from __future__ import annotations
+
+from sqlalchemy import func, select, text
+from sqlalchemy.exc import DBAPIError, IntegrityError
+from sqlalchemy.ext.asyncio import AsyncConnection
+
+from powercontext.builtin.artifacts.memory.errors import MemoryBackendConfigurationError
+from powercontext.builtin.persistence.tables import (
+    MEMORY_ENTRY_VERSION_SCOPE_INDEX_NAME,
+    MEMORY_ENTRY_VERSIONS_TABLE,
+)
+
+_SQLITE_INDEX_EXISTS = text(
+    "SELECT COUNT(*) FROM pragma_index_list('pc_memory_entry_versions') WHERE name = :index_name"
+).bindparams(index_name=MEMORY_ENTRY_VERSION_SCOPE_INDEX_NAME)
+_MYSQL_INDEX_EXISTS = text(
+    "SELECT COUNT(*) FROM information_schema.statistics "
+    "WHERE table_schema = DATABASE() "
+    "AND table_name = 'pc_memory_entry_versions' "
+    "AND index_name = :index_name"
+).bindparams(index_name=MEMORY_ENTRY_VERSION_SCOPE_INDEX_NAME)
+_CREATE_SCOPE_VERSION_INDEX = (
+    f"CREATE UNIQUE INDEX {MEMORY_ENTRY_VERSION_SCOPE_INDEX_NAME} "
+    "ON pc_memory_entry_versions (scope_id, entry_version_id)"
+)
+
+
+async def ensure_memory_entry_version_scope_identity(connection: AsyncConnection, /) -> None:
+    """Make entry-version identities scope-global on new and existing databases."""
+
+    if await _scope_version_index_exists(connection):
+        return
+    duplicate = (
+        await connection.execute(
+            select(
+                MEMORY_ENTRY_VERSIONS_TABLE.c.scope_id,
+                MEMORY_ENTRY_VERSIONS_TABLE.c.entry_version_id,
+            )
+            .group_by(
+                MEMORY_ENTRY_VERSIONS_TABLE.c.scope_id,
+                MEMORY_ENTRY_VERSIONS_TABLE.c.entry_version_id,
+            )
+            .having(func.count() > 1)
+            .limit(1)
+        )
+    ).first()
+    if duplicate is not None:
+        raise _duplicate_identity_error()
+    try:
+        await connection.exec_driver_sql(_CREATE_SCOPE_VERSION_INDEX)
+    except DBAPIError as error:
+        if await _scope_version_index_exists(connection):
+            return
+        if isinstance(error, IntegrityError):
+            raise _duplicate_identity_error() from error
+        raise
+
+
+async def _scope_version_index_exists(connection: AsyncConnection) -> bool:
+    dialect = connection.dialect.name
+    if dialect == "sqlite":
+        statement = _SQLITE_INDEX_EXISTS
+    elif dialect == "mysql":
+        statement = _MYSQL_INDEX_EXISTS
+    else:
+        raise ValueError(f"unsupported Memory schema migration dialect: {dialect}")  # noqa: TRY003
+    return int(await connection.scalar(statement) or 0) > 0
+
+
+def _duplicate_identity_error() -> MemoryBackendConfigurationError:
+    return MemoryBackendConfigurationError(
+        "pc_memory_entry_versions contains duplicate scope-global entry_version_id values"
+    )
+
+
+__all__ = ["ensure_memory_entry_version_scope_identity"]

src/powercontext/builtin/persistence/oceanbase/profile.py

@@ -29,7 +29,9 @@
 
 from powercontext.builtin.persistence.database import AsyncDatabase
 from powercontext.builtin.persistence.errors import PersistenceError
+from powercontext.builtin.persistence.memory_schema import ensure_memory_entry_version_scope_identity
 from powercontext.builtin.persistence.schema import create_tables
+from powercontext.builtin.persistence.tables import MEMORY_ENTRY_VERSIONS_TABLE
 
 _DIALECT_DRIVER = "mysql+aoceanbase"
 _DIALECT_REGISTRY_NAME = "mysql.aoceanbase"
@@ -114,6 +116,8 @@ async def _initialized_profile(profile: OceanBaseProfile) -> AsyncIterator[Ocean
         async with profile.database.transaction() as connection:
             await _require_mysql_tenant(connection)
             await create_tables(connection, profile.tables)
+            if any(table is MEMORY_ENTRY_VERSIONS_TABLE for table in profile.tables):
+                await ensure_memory_entry_version_scope_identity(connection)
         yield profile
     finally:
         await profile.database.close()

src/powercontext/builtin/persistence/sqlite/profile.py

@@ -33,7 +33,9 @@
 from sqlalchemy.pool import StaticPool
 
 from powercontext.builtin.persistence.database import AsyncDatabase
+from powercontext.builtin.persistence.memory_schema import ensure_memory_entry_version_scope_identity
 from powercontext.builtin.persistence.schema import create_tables
+from powercontext.builtin.persistence.tables import MEMORY_ENTRY_VERSIONS_TABLE
 
 _WARMUP_RETRY_SECONDS = 0.05
 _WARMUP_LOCKS: WeakKeyDictionary[asyncio.AbstractEventLoop, dict[str, asyncio.Lock]] = WeakKeyDictionary()
@@ -98,6 +100,8 @@ async def open(
             await _warm_sqlite(engine, config)
             async with database.transaction() as connection:
                 await create_tables(connection, tables)
+                if any(table is MEMORY_ENTRY_VERSIONS_TABLE for table in tables):
+                    await ensure_memory_entry_version_scope_identity(connection)
             yield profile
         finally:
             await database.close()

src/powercontext/builtin/persistence/tables.py

@@ -21,6 +21,7 @@
     Column,
     Date,
     ForeignKeyConstraint,
+    Index,
     Integer,
     LargeBinary,
     MetaData,
@@ -364,6 +365,7 @@ def _entry_text_type():
 MAX_MEMORY_ENTRY_ID_LENGTH = 128
 MAX_MEMORY_ENTRY_KIND_LENGTH = 128
 MAX_MEMORY_HASH_LENGTH = 64
+MEMORY_ENTRY_VERSION_SCOPE_INDEX_NAME = "uq_pc_memory_entry_versions_scope_version"
 
 
 MEMORY_ENTRY_VERSIONS_TABLE = Table(
@@ -413,6 +415,13 @@ def _entry_text_type():
     ),
 )
 
+MEMORY_ENTRY_VERSION_SCOPE_INDEX = Index(
+    MEMORY_ENTRY_VERSION_SCOPE_INDEX_NAME,
+    MEMORY_ENTRY_VERSIONS_TABLE.c.scope_id,
+    MEMORY_ENTRY_VERSIONS_TABLE.c.entry_version_id,
+    unique=True,
+)
+
 MEMORY_ENTRY_HEADS_TABLE = Table(
     "pc_memory_entry_heads",
     SHARED_METADATA,