diff --git a/integrations/opendal/src/powercontext_connector_opendal/source.py b/integrations/opendal/src/powercontext_connector_opendal/source.py
index 5ab48cf7d6..c089367cba 100644
--- a/integrations/opendal/src/powercontext_connector_opendal/source.py
+++ b/integrations/opendal/src/powercontext_connector_opendal/source.py
@@ -23,6 +23,7 @@
 
 from powercontext.sources import (
     TEXT_EVIDENCE_PROJECTION_KEY,
+    MemoryEvidenceDeclaration,
     Source,
     SourceMaterialization,
     SourceProjection,
@@ -92,6 +93,7 @@ class TextFileSnapshotSourceDefinition:
     name = TEXT_FILE_SNAPSHOT_SOURCE_NAME
     version = "1"
     source_class = TextFileSnapshotSource
+    memory_evidence: MemoryEvidenceDeclaration | None = None
 
     def __init__(self) -> None:
         projection = cast(SourceProjection[TextFileSnapshotSource], TextFileEvidenceProjection())
diff --git a/src/powercontext/__init__.py b/src/powercontext/__init__.py
index 7448a04133..f7589beabc 100644
--- a/src/powercontext/__init__.py
+++ b/src/powercontext/__init__.py
@@ -51,12 +51,14 @@
 )
 from powercontext.sources import (
     AdapterSourceDefinition,
+    MemoryEvidenceDeclaration,
     Source,
     SourceAdapter,
     SourceCatalog,
     SourceCatalogBackend,
     SourceDefinition,
     SourceDefinitionRegistry,
+    SourceEvidence,
     SourceMaterialization,
     SourceProjection,
     SourceProjectionKey,
@@ -90,6 +92,7 @@
     "InvalidSourceProjectionError",
     "InvalidSourceReferenceError",
     "InvalidSourceResultError",
+    "MemoryEvidenceDeclaration",
     "PolicyTransition",
     "PowerContext",
     "PowerContextError",
@@ -104,6 +107,7 @@
     "SourceDefinitionNotFoundError",
     "SourceDefinitionRegistry",
     "SourceError",
+    "SourceEvidence",
     "SourceMaterialization",
     "SourceNotFoundError",
     "SourceProjection",
diff --git a/src/powercontext/builtin/artifacts/atomic_memory/service.py b/src/powercontext/builtin/artifacts/atomic_memory/service.py
index 0947035cd5..90fb984ef4 100644
--- a/src/powercontext/builtin/artifacts/atomic_memory/service.py
+++ b/src/powercontext/builtin/artifacts/atomic_memory/service.py
@@ -80,6 +80,9 @@ async def publish(
         prepared: PreparedAtomicMemoryProjection,
     ) -> None: ...
     async def remove(self, connection: AsyncConnection, scope_id: str, artifact_id: str) -> None: ...
+    async def capture_evidence(
+        self, connection: AsyncConnection, scope_id: str, record: AtomicMemoryRecord
+    ) -> None: ...
 
 
 ATOMIC_MERGE_ERRORS = ArtifactMergeErrors(
@@ -478,6 +481,11 @@ async def commit_window(
             read_set=tuple(_common_read(read) for read in read_set),
             direct_source=direct_source,
         )
+        for item, result in zip(prepared, results, strict=True):
+            revisions = {write.artifact_id for write in item.plan.writes if write.draft is not None}
+            for record in result.records:
+                if record.artifact.artifact_id in revisions:
+                    await self.projections.capture_evidence(connection, scope_id, _atomic_record(record))
         return tuple(
             AtomicMemoryMutationResult(
                 changed=result.changed,
diff --git a/src/powercontext/builtin/persistence/atomic_memory_evidence.py b/src/powercontext/builtin/persistence/atomic_memory_evidence.py
new file mode 100644
index 0000000000..99d5c38311
--- /dev/null
+++ b/src/powercontext/builtin/persistence/atomic_memory_evidence.py
@@ -0,0 +1,211 @@
+# 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.
+
+"""Immutable per-revision evidence inputs and their neutral internal annotation."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Sequence
+from typing import Literal
+
+import rfc8785
+from aiosqlite import Connection as SQLiteConnection
+from pydantic import BaseModel, ConfigDict, ValidationError
+from sqlalchemy import insert, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncConnection
+
+from powercontext.artifacts import ArtifactRef
+from powercontext.builtin.persistence.atomic_memory_evidence_schema import ATOMIC_MEMORY_EVIDENCE_HISTORY_TABLE
+from powercontext.builtin.persistence.sources import SourceRepository, _require_identity
+from powercontext.limits import MAX_SCOPE_ID_LENGTH
+from powercontext.sources import SourceRef
+from powercontext.sources.evidence import MemoryEvidenceDeclaration, SourceEvidence
+
+
+class AtomicMemoryEvidenceAnnotation(BaseModel):
+    """Exact revision facts, not a trust score or recursively inferred evidence."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    scope_id: str
+    artifact_ref: ArtifactRef
+    sources: tuple[SourceEvidence, ...]
+    format: Literal["powercontext.atomic-memory.evidence.v1"] = "powercontext.atomic-memory.evidence.v1"
+
+
+class AtomicMemoryEvidenceRepository:
+    """Join caller transactions; retained inputs are never deleted with the index.
+
+    The annotation is derived from retained inputs at read time. There is no
+    separate cache to confuse with accepted history or join into search.
+    """
+
+    async def save(
+        self,
+        connection: AsyncConnection,
+        scope_id: str,
+        artifact_ref: ArtifactRef,
+        sources: Sequence[SourceEvidence],
+    ) -> AtomicMemoryEvidenceAnnotation:
+        artifact_ref = self._reference(scope_id, artifact_ref)
+        payload = await self._payload(connection, scope_id, artifact_ref)
+        if payload is not None:
+            return self._decode(scope_id, artifact_ref, payload) or self._neutral(scope_id, artifact_ref, sources)
+        accepted = []
+        for source in sources:
+            value = source.model_dump(mode="json")
+            try:
+                declaration = MemoryEvidenceDeclaration.model_validate(value.get("declaration"))
+            except ValidationError:
+                declaration = MemoryEvidenceDeclaration()
+            accepted.append(SourceEvidence.model_validate({**value, "declaration": declaration}))
+        annotation = AtomicMemoryEvidenceAnnotation(
+            scope_id=scope_id, artifact_ref=artifact_ref, sources=tuple(accepted)
+        )
+        if connection.dialect.name == "sqlite":
+            driver = (await connection.get_raw_connection()).driver_connection
+            if isinstance(driver, SQLiteConnection) and not driver.in_transaction:
+                # SQLite legacy transaction mode does not BEGIN on SELECT;
+                # SAVEPOINT must not accidentally own and commit the write.
+                await connection.exec_driver_sql("BEGIN")
+        try:
+            async with connection.begin_nested():
+                await connection.execute(
+                    insert(ATOMIC_MEMORY_EVIDENCE_HISTORY_TABLE).values(
+                        scope_id=scope_id,
+                        artifact_id=artifact_ref.artifact_id,
+                        revision=artifact_ref.revision,
+                        sources=rfc8785.dumps([source.model_dump(mode="json") for source in annotation.sources]),
+                    )
+                )
+        except IntegrityError:
+            payload = await self._payload(connection, scope_id, artifact_ref)
+            if payload is None:
+                raise
+            return self._decode(scope_id, artifact_ref, payload) or self._neutral(scope_id, artifact_ref, sources)
+        return annotation
+
+    async def read(
+        self, connection: AsyncConnection, scope_id: str, artifact_ref: ArtifactRef
+    ) -> AtomicMemoryEvidenceAnnotation | None:
+        artifact_ref = self._reference(scope_id, artifact_ref)
+        payload = await self._payload(connection, scope_id, artifact_ref)
+        return None if payload is None else self._decode(scope_id, artifact_ref, payload)
+
+    async def verify(
+        self,
+        connection: AsyncConnection,
+        scope_id: str,
+        artifact_ref: ArtifactRef,
+        source_refs: Sequence[SourceRef],
+        *,
+        sources: SourceRepository | None = None,
+    ) -> AtomicMemoryEvidenceAnnotation:
+        """Derive exact facts from retained inputs, never today's declarations.
+
+        Missing or damaged declaration records are neutral. Missing Sources,
+        corrupt authoritative payloads and unequal exact materials remain errors.
+        No accepted snapshot is saved or repaired by this operation.
+        """
+
+        from powercontext.builtin.persistence.atomic_memory_index import AtomicMemoryIndexError
+        from powercontext.builtin.sources import BUILTIN_SOURCE_REGISTRY
+
+        artifact_ref = self._reference(scope_id, artifact_ref)
+        repository = SourceRepository(BUILTIN_SOURCE_REGISTRY) if sources is None else sources
+        materials = await repository.historical_evidence(connection, scope_id, source_refs)
+        annotation = await self.read(connection, scope_id, artifact_ref)
+        if annotation is not None:
+            by_ref = {(item.source_ref.source_type, item.source_ref.source_id): item for item in materials}
+            for accepted in annotation.sources:
+                material = by_ref.get((accepted.source_ref.source_type, accepted.source_ref.source_id))
+                if material is not None and accepted.material_digest != material.material_digest:
+                    raise AtomicMemoryIndexError(
+                        "evidence-material", "Historical evidence does not match the exact Source material"
+                    )
+        if annotation is None or tuple(item.source_ref for item in annotation.sources) != tuple(
+            item.source_ref for item in materials
+        ):
+            return self._neutral(scope_id, artifact_ref, materials)
+        verified = tuple(
+            accepted
+            if (
+                accepted.materialization == material.materialization
+                and accepted.definition_name == material.definition_name
+                and accepted.definition_version == material.definition_version
+                and accepted.manifest_fingerprint == material.manifest_fingerprint
+                and accepted.declaration == material.declaration
+            )
+            else accepted.model_copy(update={"declaration": MemoryEvidenceDeclaration()})
+            for accepted, material in zip(annotation.sources, materials, strict=True)
+        )
+        return annotation.model_copy(update={"sources": verified})
+
+    @staticmethod
+    def _reference(scope_id: str, artifact_ref: ArtifactRef) -> ArtifactRef:
+        _require_identity("scope_id", scope_id, MAX_SCOPE_ID_LENGTH)
+        reference = ArtifactRef.model_validate(artifact_ref.model_dump())
+        if reference.family != "atomic-memory":
+            from powercontext.builtin.persistence.errors import InvalidRepositoryArgumentError
+
+            raise InvalidRepositoryArgumentError("artifact_ref", "must reference an Atomic Memory revision")
+        return reference
+
+    @staticmethod
+    async def _payload(connection: AsyncConnection, scope_id: str, artifact_ref: ArtifactRef) -> object | None:
+        table = ATOMIC_MEMORY_EVIDENCE_HISTORY_TABLE
+        return await connection.scalar(
+            select(table.c.sources).where(
+                table.c.scope_id == scope_id,
+                table.c.artifact_id == artifact_ref.artifact_id,
+                table.c.revision == artifact_ref.revision,
+            )
+        )
+
+    @staticmethod
+    def _decode(scope_id: str, artifact_ref: ArtifactRef, payload: object) -> AtomicMemoryEvidenceAnnotation | None:
+        try:
+            if not isinstance(payload, (bytes, bytearray, memoryview)):
+                return None
+            values = json.loads(bytes(payload))
+            if not isinstance(values, list):
+                return None
+            sources = []
+            for value in values:
+                if not isinstance(value, dict):
+                    return None
+                try:
+                    declaration = MemoryEvidenceDeclaration.model_validate(value.get("declaration"))
+                except ValidationError:
+                    declaration = MemoryEvidenceDeclaration()
+                sources.append(SourceEvidence.model_validate({**value, "declaration": declaration}))
+            return AtomicMemoryEvidenceAnnotation(scope_id=scope_id, artifact_ref=artifact_ref, sources=tuple(sources))
+        except (ValueError, TypeError):
+            # An unreadable declaration snapshot conveys no historical claim.
+            # This does not catch SQL errors or actual Source resolution errors.
+            return None
+
+    @staticmethod
+    def _neutral(
+        scope_id: str, artifact_ref: ArtifactRef, sources: Sequence[SourceEvidence]
+    ) -> AtomicMemoryEvidenceAnnotation:
+        # Preserve a damaged row exactly, without replacing it with today's
+        # stronger configuration. A caller can only receive neutral facts.
+        return AtomicMemoryEvidenceAnnotation(
+            scope_id=scope_id,
+            artifact_ref=artifact_ref,
+            sources=tuple(source.model_copy(update={"declaration": MemoryEvidenceDeclaration()}) for source in sources),
+        )
diff --git a/src/powercontext/builtin/persistence/atomic_memory_evidence_schema.py b/src/powercontext/builtin/persistence/atomic_memory_evidence_schema.py
new file mode 100644
index 0000000000..22846565c1
--- /dev/null
+++ b/src/powercontext/builtin/persistence/atomic_memory_evidence_schema.py
@@ -0,0 +1,59 @@
+# 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.
+
+"""Retained exact-version evidence, independent of disposable search projections."""
+
+from sqlalchemy import BigInteger, CheckConstraint, Column, LargeBinary, Table, Text
+from sqlalchemy.dialects.mysql import LONGBLOB, MEDIUMTEXT
+
+from powercontext.builtin.persistence.tables import SHARED_METADATA, identity_string
+from powercontext.limits import (
+    MAX_ARTIFACT_ID_LENGTH,
+    MAX_SCOPE_ID_LENGTH,
+    MAX_SOURCE_ID_LENGTH,
+    MAX_SOURCE_TYPE_LENGTH,
+)
+
+ATOMIC_MEMORY_EVIDENCE_SCHEMA_VERSION = "p0005"
+ATOMIC_MEMORY_EVIDENCE_SCHEMA_TABLE = Table(
+    "pc_atomic_memory_evidence_schema",
+    SHARED_METADATA,
+    Column("schema_version", identity_string(32), primary_key=True),
+)
+ATOMIC_MEMORY_EVIDENCE_HISTORY_TABLE = Table(
+    "pc_atomic_memory_evidence_history",
+    SHARED_METADATA,
+    Column("scope_id", identity_string(MAX_SCOPE_ID_LENGTH), primary_key=True),
+    Column("artifact_id", identity_string(MAX_ARTIFACT_ID_LENGTH), primary_key=True),
+    Column("revision", BigInteger, primary_key=True),
+    Column("sources", LargeBinary().with_variant(LONGBLOB(), "mysql"), nullable=False),
+    CheckConstraint("revision > 0", name="ck_pc_atomic_memory_evidence_history_revision"),
+)
+SOURCE_EVIDENCE_BINDINGS_TABLE = Table(
+    "pc_source_evidence_bindings",
+    SHARED_METADATA,
+    Column("scope_id", identity_string(MAX_SCOPE_ID_LENGTH), primary_key=True),
+    Column("source_type", identity_string(MAX_SOURCE_TYPE_LENGTH), primary_key=True),
+    Column("source_id", identity_string(MAX_SOURCE_ID_LENGTH), primary_key=True),
+    Column("material_digest", identity_string(71), nullable=False),
+    Column("definition_name", Text().with_variant(MEDIUMTEXT(), "mysql"), nullable=False),
+    Column("definition_version", Text().with_variant(MEDIUMTEXT(), "mysql"), nullable=False),
+    Column("manifest_fingerprint", identity_string(71)),
+    Column("declaration", LargeBinary().with_variant(LONGBLOB(), "mysql"), nullable=False),
+)
+ATOMIC_MEMORY_EVIDENCE_TABLES = (
+    ATOMIC_MEMORY_EVIDENCE_SCHEMA_TABLE,
+    ATOMIC_MEMORY_EVIDENCE_HISTORY_TABLE,
+    SOURCE_EVIDENCE_BINDINGS_TABLE,
+)
diff --git a/src/powercontext/builtin/persistence/atomic_memory_index.py b/src/powercontext/builtin/persistence/atomic_memory_index.py
index 4cb2cbb00a..d125a46861 100644
--- a/src/powercontext/builtin/persistence/atomic_memory_index.py
+++ b/src/powercontext/builtin/persistence/atomic_memory_index.py
@@ -18,6 +18,7 @@
 
 import json
 import math
+from collections.abc import Sequence
 from dataclasses import dataclass, replace
 from hashlib import sha256
 from typing import Any, Literal, Protocol
@@ -37,9 +38,16 @@
     fts_query_requirements,
 )
 from powercontext.builtin.inference import EmbeddingModel
+from powercontext.builtin.persistence.atomic_memory_evidence import (
+    AtomicMemoryEvidenceAnnotation,
+    AtomicMemoryEvidenceRepository,
+)
+from powercontext.builtin.persistence.atomic_memory_evidence_schema import ATOMIC_MEMORY_EVIDENCE_TABLES
 from powercontext.builtin.persistence.atomic_memory_index_schema import ATOMIC_MEMORY_PROJECTION_FORMAT
+from powercontext.builtin.persistence.sources import SourceRepository
 from powercontext.builtin.persistence.tables import ARTIFACT_TAGS_TABLE
 from powercontext.builtin.tags import TagFilter
+from powercontext.sources.evidence import SourceEvidence
 
 AtomicMemorySearchMode = Literal["fts", "vector", "hybrid"]
 ATOMIC_MEMORY_RRF_CONSTANT = 60
@@ -166,6 +174,19 @@ async def replace(
 
     async def delete(self, connection: AsyncConnection, scope_id: str, artifact_id: str, /) -> None: ...
 
+    async def save_evidence(
+        self,
+        connection: AsyncConnection,
+        scope_id: str,
+        artifact_ref: ArtifactRef,
+        sources: Sequence[SourceEvidence],
+        /,
+    ) -> AtomicMemoryEvidenceAnnotation: ...
+
+    async def read_evidence(
+        self, connection: AsyncConnection, scope_id: str, artifact_ref: ArtifactRef, /
+    ) -> AtomicMemoryEvidenceAnnotation | None: ...
+
     async def search(
         self, connection: AsyncConnection, scope_id: str, request: AtomicMemorySearchRequest, /
     ) -> AtomicMemorySearchChannels: ...
@@ -191,9 +212,11 @@ def __init__(
         index: AtomicMemoryIndex,
         *,
         embedding_model: EmbeddingModel | None = None,
+        sources: SourceRepository | None = None,
     ) -> None:
         self.index = index
         self._embedding_model = embedding_model
+        self._sources = sources
 
     @property
     def profile_fingerprint(self) -> str | None:
@@ -243,6 +266,17 @@ async def prepare(self, content: BaseModel) -> PreparedAtomicMemoryProjection:
             embedding_input_hash=None if vector is None else atomic_memory_embedding_input_hash(kind, body),
         )
 
+    async def capture_evidence(self, connection: AsyncConnection, scope_id: str, record: Any) -> None:
+        """Accept history only for a committed body write, never state-only republishing."""
+
+        artifact = record.artifact
+        if (
+            self._sources is not None
+            and await self.index.read_evidence(connection, scope_id, artifact.as_ref()) is None
+        ):
+            evidence = await self._sources.evidence(connection, scope_id, artifact.lineage.sources)
+            await self.index.save_evidence(connection, scope_id, artifact.as_ref(), evidence)
+
     async def publish(
         self,
         connection: AsyncConnection,
@@ -450,12 +484,28 @@ def __init__(self, table: Table, profile: EmbeddingProfile | None) -> None:
         if profile is not None and (profile.dimension < 1 or profile.distance != "l2"):
             raise AtomicMemoryIndexError("embedding-profile", "Atomic Memory requires a positive L2 embedding profile")
         self.table = table
-        self.tables: tuple[Table, ...] = (table,)
+        self.tables: tuple[Table, ...] = (table, *ATOMIC_MEMORY_EVIDENCE_TABLES)
+        self._evidence = AtomicMemoryEvidenceRepository()
         self.profile = profile
         self.capabilities = AtomicMemoryIndexCapabilities(
             vector=profile is not None, hybrid=profile is not None, embedding_profile=profile
         )
 
+    async def save_evidence(
+        self,
+        connection: AsyncConnection,
+        scope_id: str,
+        artifact_ref: ArtifactRef,
+        sources: Sequence[SourceEvidence],
+        /,
+    ) -> AtomicMemoryEvidenceAnnotation:
+        return await self._evidence.save(connection, scope_id, artifact_ref, sources)
+
+    async def read_evidence(
+        self, connection: AsyncConnection, scope_id: str, artifact_ref: ArtifactRef, /
+    ) -> AtomicMemoryEvidenceAnnotation | None:
+        return await self._evidence.read(connection, scope_id, artifact_ref)
+
     def _encode_embedding(self, vector: tuple[float, ...]) -> object:
         return vector
 
diff --git a/src/powercontext/builtin/persistence/atomic_memory_readiness.py b/src/powercontext/builtin/persistence/atomic_memory_readiness.py
index fdd46b5b74..52df67b4a5 100644
--- a/src/powercontext/builtin/persistence/atomic_memory_readiness.py
+++ b/src/powercontext/builtin/persistence/atomic_memory_readiness.py
@@ -25,7 +25,12 @@
 from sqlalchemy import inspect, text
 from sqlalchemy.ext.asyncio import AsyncConnection
 
+from powercontext.builtin.persistence.atomic_memory_evidence_schema import ATOMIC_MEMORY_EVIDENCE_TABLES
 from powercontext.builtin.persistence.errors import PersistenceError
+from powercontext.builtin.persistence.migrations.atomic_memory_evidence import (
+    AtomicMemoryEvidenceSchemaError,
+    require_atomic_memory_evidence_schema,
+)
 
 LEGACY_CITATION_COLUMN = "memory_citations"
 _LEGACY_CITATION_TABLES = ("pc_artifacts", "pc_artifact_candidate_versions")
@@ -90,11 +95,16 @@ async def legacy_foreign_keys(connection: AsyncConnection, tables: set[str]) ->
 async def atomic_memory_readiness_issues(
     connection: AsyncConnection, tables: set[str] | None = None, *, collections_removed: bool = True
 ) -> list[str]:
-    """Check public migration completion, allowing collections only during offline pre-removal acceptance."""
+    """Check versioned schema and public migration completion without reading retained history."""
 
     if tables is None:
         tables = set(await connection.run_sync(lambda value: inspect(value).get_table_names()))
     issues: list[str] = []
+    if "pc_artifacts" in tables or any(table.name in tables for table in ATOMIC_MEMORY_EVIDENCE_TABLES):
+        try:
+            await require_atomic_memory_evidence_schema(connection, tables)
+        except AtomicMemoryEvidenceSchemaError as error:
+            issues.append(str(error))
     if (
         collections_removed
         and "pc_artifacts" in tables
diff --git a/src/powercontext/builtin/persistence/migrations/atomic_memory_evidence.py b/src/powercontext/builtin/persistence/migrations/atomic_memory_evidence.py
new file mode 100644
index 0000000000..14e2e2809f
--- /dev/null
+++ b/src/powercontext/builtin/persistence/migrations/atomic_memory_evidence.py
@@ -0,0 +1,130 @@
+# 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.
+
+"""Apply the frozen additive p0005 revision to runtime-owned evidence tables.
+
+The production bundle covers a historical subset of business tables. Runtime
+profiles therefore apply this same frozen additive revision independently of
+that bundle's whole-database inventory, without stamping unrelated schema or
+backfilling declarations for old Source/Artifact rows. SQLite DDL joins the
+caller's transaction; MySQL DDL may commit and is resumed by exact postconditions.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from aiosqlite import Connection as SQLiteConnection
+from sqlalchemy import inspect, select
+from sqlalchemy.ext.asyncio import AsyncConnection
+
+from powercontext.builtin.persistence.atomic_memory_evidence_schema import (
+    ATOMIC_MEMORY_EVIDENCE_SCHEMA_TABLE,
+    ATOMIC_MEMORY_EVIDENCE_SCHEMA_VERSION,
+    ATOMIC_MEMORY_EVIDENCE_TABLES,
+)
+from powercontext.builtin.persistence.errors import PersistenceError
+
+
+class AtomicMemoryEvidenceSchemaError(PersistenceError):
+    """An evidence schema is not the registered versioned layout."""
+
+    def __init__(self, detail: str) -> None:
+        message = {
+            "missing": "Atomic Memory evidence schema is incomplete; apply the registered p0005 migration",
+            "version": "Atomic Memory evidence schema version is unsupported",
+            "shape": "Atomic Memory evidence table has an incompatible shape",
+        }[detail]
+        super().__init__(message)
+
+
+async def apply_atomic_memory_evidence_migration(connection: AsyncConnection) -> None:
+    """Add only missing p0005 tables, validate exact shapes, then mark ready.
+
+    No reads or writes of Source/Artifact history are needed, and no current
+    declaration is used to invent accepted historical inputs.
+    """
+
+    if connection.dialect.name == "sqlite":
+        driver = (await connection.get_raw_connection()).driver_connection
+        if isinstance(driver, SQLiteConnection) and not driver.in_transaction:
+            await connection.exec_driver_sql("BEGIN IMMEDIATE")
+    dialect = "sqlite" if connection.dialect.name == "sqlite" else "mysql"
+    resources = Path(__file__).parent / "resources/schemas/atomic_memory_evidence_v1.json"
+    statements = json.loads(resources.read_text())[dialect]
+    tables = set(await connection.run_sync(lambda sync: inspect(sync).get_table_names()))
+    for table, statement in zip(ATOMIC_MEMORY_EVIDENCE_TABLES, statements[:-1], strict=True):
+        if table.name in tables:
+            await _require_shape(connection, table.name, statement, dialect)
+    marker = ATOMIC_MEMORY_EVIDENCE_SCHEMA_TABLE
+    versions = (
+        tuple((await connection.execute(select(marker.c.schema_version))).scalars()) if marker.name in tables else ()
+    )
+    if versions and versions != (ATOMIC_MEMORY_EVIDENCE_SCHEMA_VERSION,):
+        raise AtomicMemoryEvidenceSchemaError("version")
+    if versions and any(table.name not in tables for table in ATOMIC_MEMORY_EVIDENCE_TABLES):
+        # A ready marker cannot authorize recreating missing retained history.
+        raise AtomicMemoryEvidenceSchemaError("missing")
+    for table, statement in zip(ATOMIC_MEMORY_EVIDENCE_TABLES, statements[:-1], strict=True):
+        if table.name not in tables:
+            await connection.exec_driver_sql(statement)
+            await _require_shape(connection, table.name, statement, dialect)
+    if not versions:
+        await connection.exec_driver_sql(statements[-1])
+
+
+async def require_atomic_memory_evidence_schema(connection: AsyncConnection, tables: set[str] | None = None) -> None:
+    """Check retained evidence readiness without DDL or historical payload reads."""
+
+    if tables is None:
+        tables = set(await connection.run_sync(lambda sync: inspect(sync).get_table_names()))
+    if any(table.name not in tables for table in ATOMIC_MEMORY_EVIDENCE_TABLES):
+        raise AtomicMemoryEvidenceSchemaError("missing")
+    dialect = "sqlite" if connection.dialect.name == "sqlite" else "mysql"
+    resources = Path(__file__).parent / "resources/schemas/atomic_memory_evidence_v1.json"
+    statements = json.loads(resources.read_text())[dialect]
+    for table, statement in zip(ATOMIC_MEMORY_EVIDENCE_TABLES, statements[:-1], strict=True):
+        await _require_shape(connection, table.name, statement, dialect)
+    versions = tuple((await connection.execute(select(ATOMIC_MEMORY_EVIDENCE_SCHEMA_TABLE.c.schema_version))).scalars())
+    if versions != (ATOMIC_MEMORY_EVIDENCE_SCHEMA_VERSION,):
+        raise AtomicMemoryEvidenceSchemaError("version")
+
+
+async def _require_shape(connection: AsyncConnection, name: str, statement: str, dialect: str) -> None:
+    if dialect == "sqlite":
+        actual = await connection.exec_driver_sql(
+            "SELECT type, name, sql FROM sqlite_schema WHERE (name = ? OR tbl_name = ?) "
+            "AND substr(name, 1, 7) != 'sqlite_' ORDER BY type, name",
+            (name, name),
+        )
+        if actual.all() != [("table", name, statement)]:
+            raise AtomicMemoryEvidenceSchemaError("shape")
+        return
+    # Reuse the frozen MySQL inventory parser so physical formatting accepted
+    # by the shared backend does not become a second schema convention.
+    from powercontext.builtin.persistence.migrations.models import MigrationError
+    from powercontext.builtin.persistence.migrations.mysql_schema import _matches, _parse_table
+
+    actual = (await connection.exec_driver_sql(f"SHOW CREATE TABLE `{name}`")).one()[1]
+    try:
+        actual_name, actual_table = _parse_table(actual)
+        expected_name, expected_table = _parse_table(statement)
+    except MigrationError as error:
+        raise AtomicMemoryEvidenceSchemaError("shape") from error
+    if not _matches(
+        {"tables": {actual_name: actual_table}, "views": [], "triggers": []},
+        {"tables": {expected_name: expected_table}, "views": [], "triggers": []},
+    ):
+        raise AtomicMemoryEvidenceSchemaError("shape")
diff --git a/src/powercontext/builtin/persistence/migrations/mysql_resources/versions/p0005_atomic_memory_evidence.py b/src/powercontext/builtin/persistence/migrations/mysql_resources/versions/p0005_atomic_memory_evidence.py
new file mode 100644
index 0000000000..2550f917ed
--- /dev/null
+++ b/src/powercontext/builtin/persistence/migrations/mysql_resources/versions/p0005_atomic_memory_evidence.py
@@ -0,0 +1,25 @@
+# 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.
+
+"""Create the frozen additive evidence layout without backfilling trust."""
+
+from alembic import context, op
+
+revision = "p0005"
+down_revision = "p0004"
+
+
+def upgrade() -> None:
+    context.config.attributes["execute_revision"](revision)
+    op.execute("INSERT INTO pc_atomic_memory_evidence_schema (schema_version) VALUES ('p0005')")
diff --git a/src/powercontext/builtin/persistence/migrations/mysql_schema.py b/src/powercontext/builtin/persistence/migrations/mysql_schema.py
index e2128c09e3..3fbb8a2200 100644
--- a/src/powercontext/builtin/persistence/migrations/mysql_schema.py
+++ b/src/powercontext/builtin/persistence/migrations/mysql_schema.py
@@ -14,7 +14,7 @@
 
 """Recognize the frozen MySQL layouts without importing application metadata.
 
-Only the reviewed linear p0001--p0004 history is supported. DDL recovery accepts
+Only the reviewed linear p0001--p0005 history is supported. DDL recovery accepts
 an exact prefix of the next revision, never an arbitrary collection of objects.
 The caller owns maintenance evidence, backup, the pinned database lock and the
 version row; this module never writes or stamps a revision.
@@ -35,7 +35,7 @@
 from .models import MigrationError, digest
 
 _LEXEME = re.compile(r"\s+|`(?:``|[^`])*`|'(?:''|\\.|[^'\\])*'|[A-Za-z_][A-Za-z_0-9$]*|[0-9]+|>=|<=|<>|!=|[(),.=<>+-]")
-_REVISIONS = ("p0001", "p0002", "p0003", "p0004")
+_REVISIONS = ("p0001", "p0002", "p0003", "p0004", "p0005")
 _CITATION_TABLES = ("pc_artifacts", "pc_artifact_candidate_versions")
 _FAMILY_CHECK = "ck_pc_artifact_tags_family"
 _CONTROL_TABLE = "pc_schema_revision"
@@ -151,11 +151,15 @@ def __init__(self, bundle: MigrationBundle) -> None:
         self.bundle = bundle
         baseline = json.loads((bundle.snapshot_directory / "schemas/pre_dream.json").read_text())["mysql"]
         current = json.loads((bundle.snapshot_directory / "schemas/v1_1_0.json").read_text())["mysql"]
+        evidence = json.loads((bundle.snapshot_directory / "schemas/atomic_memory_evidence_v1.json").read_text())[
+            "mysql"
+        ]
         self._statements = {
             "p0001": tuple(baseline),
             "p0002": tuple(f"ALTER TABLE {table} ADD COLUMN memory_citations MEDIUMBLOB" for table in _CITATION_TABLES),
             "p0003": (f"ALTER TABLE pc_artifact_tags DROP CHECK {_FAMILY_CHECK}",),
             "p0004": (_ARTIFACT_HANDLES_DDL,),
+            "p0005": tuple(evidence[:-1]),
         }
         inventory = _empty_inventory()
         self._prefixes: dict[str, list[dict[str, Any]]] = {}
diff --git a/src/powercontext/builtin/persistence/migrations/resources/manifest.json b/src/powercontext/builtin/persistence/migrations/resources/manifest.json
index 16364daa4f..dc2b373564 100644
--- a/src/powercontext/builtin/persistence/migrations/resources/manifest.json
+++ b/src/powercontext/builtin/persistence/migrations/resources/manifest.json
@@ -5,12 +5,16 @@
     ],
     "p0002": [],
     "p0003": [],
-    "p0004": []
+    "p0004": [],
+    "p0005": [
+      "schemas/atomic_memory_evidence_v1.json"
+    ]
   },
   "sqlite_baselines": {
     "e1cae089a8a565291a7a1f782cea1f6c6e1a6cdd9eab75ef0c4ca61dfda453cb": "p0001",
     "fcee685ed307ee6d02fe5e950ed4e80be75f38d19de902ccacf443f21ee8c73d": "p0003",
-    "8cb2d41e53cde751c8d0ef4023c344ec5ec78d9e579366c1c93d17de30aa5ea4": "p0004"
+    "8cb2d41e53cde751c8d0ef4023c344ec5ec78d9e579366c1c93d17de30aa5ea4": "p0004",
+    "5d2a458f6cb278e4fbff5aa7cef9429bcb940be5fb0a29649be50c30482e00e1": "p0005"
   },
   "sqlite_fingerprints": {
     "p0001": [
@@ -26,6 +30,10 @@
     "p0004": [
       "f65244be963a382a62565118ff874cd775ab1ea33cf51a49881b40391e47c051",
       "8cb2d41e53cde751c8d0ef4023c344ec5ec78d9e579366c1c93d17de30aa5ea4"
+    ],
+    "p0005": [
+      "10b8f62310f90e4e267e795efef0aa6e6716c1ecad515ab745e783507f6496f3",
+      "5d2a458f6cb278e4fbff5aa7cef9429bcb940be5fb0a29649be50c30482e00e1"
     ]
   },
   "affected_objects": [
@@ -33,7 +41,10 @@
     "pc_artifact_heads",
     "pc_artifact_candidate_versions",
     "pc_artifact_tags",
-    "pc_artifact_handles"
+    "pc_artifact_handles",
+    "pc_atomic_memory_evidence_schema",
+    "pc_atomic_memory_evidence_history",
+    "pc_source_evidence_bindings"
   ],
   "retained_objects": [
     "pc_retained_p0002_artifact_tags"
diff --git a/src/powercontext/builtin/persistence/migrations/resources/schemas/atomic_memory_evidence_v1.json b/src/powercontext/builtin/persistence/migrations/resources/schemas/atomic_memory_evidence_v1.json
new file mode 100644
index 0000000000..e4af1dc164
--- /dev/null
+++ b/src/powercontext/builtin/persistence/migrations/resources/schemas/atomic_memory_evidence_v1.json
@@ -0,0 +1,14 @@
+{
+  "sqlite": [
+    "CREATE TABLE pc_atomic_memory_evidence_schema (\n\tschema_version VARCHAR(32) NOT NULL, \n\tPRIMARY KEY (schema_version)\n)",
+    "CREATE TABLE pc_atomic_memory_evidence_history (\n\tscope_id VARCHAR(256) NOT NULL, \n\tartifact_id VARCHAR(128) NOT NULL, \n\trevision BIGINT NOT NULL, \n\tsources BLOB NOT NULL, \n\tPRIMARY KEY (scope_id, artifact_id, revision), \n\tCONSTRAINT ck_pc_atomic_memory_evidence_history_revision CHECK (revision > 0)\n)",
+    "CREATE TABLE pc_source_evidence_bindings (\n\tscope_id VARCHAR(256) NOT NULL, \n\tsource_type VARCHAR(128) NOT NULL, \n\tsource_id VARCHAR(256) NOT NULL, \n\tmaterial_digest VARCHAR(71) NOT NULL, \n\tdefinition_name TEXT NOT NULL, \n\tdefinition_version TEXT NOT NULL, \n\tmanifest_fingerprint VARCHAR(71), \n\tdeclaration BLOB NOT NULL, \n\tPRIMARY KEY (scope_id, source_type, source_id)\n)",
+    "INSERT INTO pc_atomic_memory_evidence_schema (schema_version) VALUES ('p0005')"
+  ],
+  "mysql": [
+    "CREATE TABLE pc_atomic_memory_evidence_schema (\n\tschema_version VARCHAR(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\tPRIMARY KEY (schema_version)\n)",
+    "CREATE TABLE pc_atomic_memory_evidence_history (\n\tscope_id VARCHAR(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\tartifact_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\trevision BIGINT NOT NULL, \n\tsources LONGBLOB NOT NULL, \n\tPRIMARY KEY (scope_id, artifact_id, revision), \n\tCONSTRAINT ck_pc_atomic_memory_evidence_history_revision CHECK (revision > 0)\n)",
+    "CREATE TABLE pc_source_evidence_bindings (\n\tscope_id VARCHAR(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\tsource_type VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\tsource_id VARCHAR(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\tmaterial_digest VARCHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\tdefinition_name MEDIUMTEXT NOT NULL, \n\tdefinition_version MEDIUMTEXT NOT NULL, \n\tmanifest_fingerprint VARCHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin, \n\tdeclaration LONGBLOB NOT NULL, \n\tPRIMARY KEY (scope_id, source_type, source_id)\n)",
+    "INSERT INTO pc_atomic_memory_evidence_schema (schema_version) VALUES ('p0005')"
+  ]
+}
diff --git a/src/powercontext/builtin/persistence/migrations/resources/versions/p0005_atomic_memory_evidence.py b/src/powercontext/builtin/persistence/migrations/resources/versions/p0005_atomic_memory_evidence.py
new file mode 100644
index 0000000000..9190834ba0
--- /dev/null
+++ b/src/powercontext/builtin/persistence/migrations/resources/versions/p0005_atomic_memory_evidence.py
@@ -0,0 +1,30 @@
+# 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.
+
+"""Frozen additive evidence revision; never synthesize historical declarations."""
+
+import json
+from pathlib import Path
+
+from alembic import op
+
+revision = "p0005"
+down_revision = "p0004"
+
+
+def upgrade() -> None:
+    statements = json.loads((Path(__file__).parents[1] / "schemas/atomic_memory_evidence_v1.json").read_text())
+    connection = op.get_bind()
+    for statement in statements["sqlite" if connection.dialect.name == "sqlite" else "mysql"]:
+        connection.exec_driver_sql(statement)
diff --git a/src/powercontext/builtin/persistence/schema.py b/src/powercontext/builtin/persistence/schema.py
index d1e4170bc3..3fff568777 100644
--- a/src/powercontext/builtin/persistence/schema.py
+++ b/src/powercontext/builtin/persistence/schema.py
@@ -21,12 +21,20 @@
 from sqlalchemy import MetaData, Table
 from sqlalchemy.ext.asyncio import AsyncConnection
 
+from powercontext.builtin.persistence.atomic_memory_evidence_schema import ATOMIC_MEMORY_EVIDENCE_TABLES
+from powercontext.builtin.persistence.migrations.atomic_memory_evidence import apply_atomic_memory_evidence_migration
+
 
 async def create_tables(connection: AsyncConnection, tables: Sequence[Table], /) -> None:
-    """Create missing tables and indexes using SQLAlchemy's native metadata support."""
+    """Create selected tables, applying retained evidence through its frozen revision."""
 
     groups: dict[MetaData, list[Table]] = {}
+    evidence_names = {table.name for table in ATOMIC_MEMORY_EVIDENCE_TABLES}
+    if any(table.name in evidence_names or table.name == "pc_sources" for table in tables):
+        await apply_atomic_memory_evidence_migration(connection)
     for table in tables:
+        if table.name in evidence_names:
+            continue
         groups.setdefault(table.metadata, []).append(table)
 
     for metadata, selected in groups.items():
diff --git a/src/powercontext/builtin/persistence/source_definitions.py b/src/powercontext/builtin/persistence/source_definitions.py
index 9e5d4a8640..3705e9b566 100644
--- a/src/powercontext/builtin/persistence/source_definitions.py
+++ b/src/powercontext/builtin/persistence/source_definitions.py
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Persistence for worker-owned Source Definition manifests."""
+"""Immutable historical manifests owned by the Source registration boundary."""
 
 from __future__ import annotations
 
@@ -26,15 +26,17 @@
 from powercontext.builtin.persistence.codec import dump_model, load_model, stored_bytes
 from powercontext.builtin.persistence.errors import (
     IdentityMismatchError,
+    InvalidStoredColumnError,
+    InvalidStoredPayloadError,
     RepositoryNotFoundError,
     StoredPayloadConflictError,
 )
 from powercontext.builtin.persistence.tables import SOURCE_DEFINITION_MANIFESTS_TABLE
-from powercontext.sources import SourceDefinitionManifest
+from powercontext.sources import SourceDefinition, SourceDefinitionManifest, manifest_for_definition
 
 
 class SourceDefinitionManifestRepository:
-    """Register immutable worker-owned Definition manifests by name and version."""
+    """Retain native and worker manifests without changing their identities."""
 
     async def register(
         self,
@@ -59,7 +61,9 @@ async def register(
             )
         except IntegrityError:
             existing = await self.find(connection, manifest.name, manifest.version)
-            if existing is None or existing != manifest:
+            if existing is None:
+                raise
+            if existing != manifest:
                 raise StoredPayloadConflictError(
                     "source-definition-manifest",
                     (manifest.name, manifest.version),
@@ -67,6 +71,29 @@ async def register(
             return existing
         return manifest
 
+    async def register_native(
+        self,
+        connection: AsyncConnection,
+        definition: SourceDefinition[Any, Any, Any],
+        /,
+    ) -> SourceDefinitionManifest | None:
+        """Retain the existing manifest representation for trusted native material.
+
+        This internal registration cannot replace an older manifest. A missing,
+        invalid or conflicting ancillary manifest leaves the Source declaration
+        neutral; database failures still abort the caller's transaction. External
+        registration keeps its existing active-Definition ownership guard.
+        """
+
+        try:
+            manifest = manifest_for_definition(definition)
+        except Exception:  # Native schema providers may not expose a valid manifest.
+            return None
+        try:
+            return await self.register(connection, manifest)
+        except (IdentityMismatchError, InvalidStoredColumnError, InvalidStoredPayloadError, StoredPayloadConflictError):
+            return None
+
     async def get(
         self,
         connection: AsyncConnection,
diff --git a/src/powercontext/builtin/persistence/sources.py b/src/powercontext/builtin/persistence/sources.py
index ba3cd56c56..0b58316aa7 100644
--- a/src/powercontext/builtin/persistence/sources.py
+++ b/src/powercontext/builtin/persistence/sources.py
@@ -18,10 +18,16 @@
 
 from collections.abc import AsyncGenerator, Iterable, Mapping, Sequence
 from contextlib import suppress
+from hashlib import sha256
 from typing import Any, Literal, cast
 
 import rfc8785
+from jsonschema import Draft202012Validator
+from jsonschema.exceptions import SchemaError
+from jsonschema.exceptions import ValidationError as JsonSchemaValidationError
 from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
+from referencing import Registry
+from referencing.exceptions import Unresolvable
 from sqlalchemy import func, insert, select, tuple_, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncConnection
@@ -29,6 +35,7 @@
 from powercontext.builtin.artifacts.experience import EXPERIENCE_INCUBATION_CURSOR_NAME
 from powercontext.builtin.artifacts.profile.models import PROFILE_SOURCE_WINDOW_BINDING
 from powercontext.builtin.artifacts.topic_memory import TOPIC_MEMORY_SOURCE_WINDOW_BINDING
+from powercontext.builtin.persistence.atomic_memory_evidence_schema import SOURCE_EVIDENCE_BINDINGS_TABLE
 from powercontext.builtin.persistence.codec import load_model, stored_bytes
 from powercontext.builtin.persistence.errors import (
     IdentityMismatchError,
@@ -39,12 +46,22 @@
     StoredPayloadConflictError,
 )
 from powercontext.builtin.persistence.processing_intents import ArtifactProcessingIntentRepository
+from powercontext.builtin.persistence.source_definitions import SourceDefinitionManifestRepository
 from powercontext.builtin.persistence.tables import SOURCE_JOURNAL_HEADS_TABLE, SOURCES_TABLE
-from powercontext.builtin.sources.content import ContentSource
+from powercontext.builtin.sources.content import CONTENT_SOURCE_NAME, ContentSource
 from powercontext.builtin.triggers import SOURCE_WINDOW_TRIGGER_NAME
 from powercontext.errors import SourceDefinitionNotFoundError
 from powercontext.limits import MAX_SCOPE_ID_LENGTH
-from powercontext.sources import Source, SourceAdapter, SourceDefinitionRegistry, SourceObservation, SourceRef
+from powercontext.sources import (
+    MemoryEvidenceDeclaration,
+    Source,
+    SourceAdapter,
+    SourceDefinitionManifest,
+    SourceDefinitionRegistry,
+    SourceEvidence,
+    SourceObservation,
+    SourceRef,
+)
 
 _AnySourceAdapter = SourceAdapter[Any, Any, Any]
 _JSON_OBJECT = TypeAdapter(dict[str, JsonValue])
@@ -165,6 +182,7 @@ async def add_with_status(
             if stored.value != source:
                 raise StoredPayloadConflictError("source", (scope_id, ref)) from None
             return stored, False
+        await self._bind_memory_evidence(connection, scope_id, ref, source)
         # Centralized here so capture, record projection and import all publish
         # discoverable input atomically. Replayed Source identities do not dirty
         # a binding again. Disabled processors simply retain ordinary dirty.
@@ -233,6 +251,259 @@ async def get_many(
                 raise RepositoryNotFoundError("source", (scope_id, ref))
         return tuple(decoded[(ref.source_type, ref.source_id)] for ref in ordered)
 
+    async def evidence(
+        self,
+        connection: AsyncConnection,
+        scope_id: str,
+        refs: Sequence[SourceRef],
+        /,
+    ) -> tuple[SourceEvidence, ...]:
+        """Read accepted declarations against exact stored Source material.
+
+        Source reads retain their existing errors and never become generation
+        eligibility checks. Missing or invalid declaration bindings are neutral;
+        database failures propagate. No current registry declaration or provider
+        value is substituted for historical registration facts.
+        """
+
+        return await self.historical_evidence(connection, scope_id, refs)
+
+    async def historical_evidence(
+        self, connection: AsyncConnection, scope_id: str, refs: Sequence[SourceRef], /
+    ) -> tuple[SourceEvidence, ...]:
+        """Check retained material without loading today's native Definition.
+
+        Only the stored representation selects decoding. Canonical JSON retains
+        all native fields, rather than round-tripping through a newer model.
+        This does not supply a decoder or provider value for ordinary reads.
+        """
+
+        _require_identity("scope_id", scope_id, MAX_SCOPE_ID_LENGTH)
+        identities = tuple(dict.fromkeys((ref.source_type, ref.source_id) for ref in refs))
+        if not identities:
+            return ()
+        rows = (
+            (
+                await connection.execute(
+                    select(SOURCES_TABLE).where(
+                        SOURCES_TABLE.c.scope_id == scope_id,
+                        tuple_(SOURCES_TABLE.c.source_type, SOURCES_TABLE.c.source_id).in_(identities),
+                    )
+                )
+            )
+            .mappings()
+            .all()
+        )
+        by_ref = {(row["source_type"], row["source_id"]): row for row in rows}
+        materials: list[SourceEvidence] = []
+        fingerprints: dict[tuple[str, str], str] = {}
+        for ref in identities:
+            reference = SourceRef(source_type=ref[0], source_id=ref[1])
+            row = by_ref.get(ref)
+            if row is None:
+                raise RepositoryNotFoundError("source", (scope_id, reference))
+            source, material = await self._historical_material(connection, reference, row)
+            if isinstance(source, SourceObservation):
+                fingerprints[ref] = source.definition_fingerprint
+            elif ref[0] == CONTENT_SOURCE_NAME:
+                # The server-owned Content witness is the sole permitted material
+                # mutation; this exclusion is the same one used at acceptance.
+                internal = material.get("internal")
+                if (
+                    isinstance(internal, dict)
+                    and internal.get("role") == "lineage_only"
+                    and internal.get("operation") in {"artifact_create", "artifact_replace"}
+                ):
+                    continue
+                material = {key: value for key, value in material.items() if key != "handoff_receipt"}
+            materials.append(
+                SourceEvidence(
+                    source_ref=reference,
+                    materialization=source.materialization,
+                    material_digest=_source_material_digest_material(material),
+                    definition_name=ref[0],
+                    definition_version=source.definition_version,
+                )
+            )
+        return await self._bound_evidence(connection, scope_id, tuple(materials), fingerprints)
+
+    async def _historical_material(
+        self, connection: AsyncConnection, reference: SourceRef, row: Mapping[Any, Any]
+    ) -> tuple[Source, dict[str, JsonValue]]:
+        """Validate the stored representation and its retained schema, not a live model."""
+
+        payload = stored_bytes(row["payload"], column="payload")
+        try:
+            envelope = load_model(_StoredSourcePayload, payload, kind="source-envelope", name=reference.source_type)
+        except InvalidStoredPayloadError:
+            # Bare legacy material has no representation discriminator. Use
+            # its registered decoder when available; otherwise retain opaque
+            # canonical fields for neutral maintenance, not a guessed schema.
+            try:
+                self._registry.definition_for_name(reference.source_type)
+            except SourceDefinitionNotFoundError:
+                source = load_model(Source, payload, kind="source", name=reference.source_type)
+                if source.name != reference.source_id:
+                    raise IdentityMismatchError("source", reference.source_id, source.name) from None
+                try:
+                    material = _JSON_OBJECT.validate_json(payload, strict=True)
+                except ValueError as error:
+                    raise InvalidStoredPayloadError("source", reference.source_type, "payload is not JSON") from error
+                return source, material
+            source = self._decode_row(row).value
+            try:
+                material = _JSON_OBJECT.validate_json(payload, strict=True)
+            except ValueError as error:
+                raise InvalidStoredPayloadError("source", reference.source_type, "payload is not JSON") from error
+            return source, material
+        material = envelope.value
+        model = SourceObservation if envelope.representation == "observation" else Source
+        source = load_model(model, rfc8785.dumps(material), kind="source", name=reference.source_type)
+        decoded = SourceRef(
+            source_type=source.source_type if isinstance(source, SourceObservation) else reference.source_type,
+            source_id=source.name,
+        )
+        if reference != decoded:
+            raise IdentityMismatchError("source", reference, decoded)
+        table = SOURCE_EVIDENCE_BINDINGS_TABLE
+        fingerprint = await connection.scalar(
+            select(table.c.manifest_fingerprint).where(
+                table.c.scope_id == row["scope_id"],
+                table.c.source_type == reference.source_type,
+                table.c.source_id == reference.source_id,
+                table.c.definition_name == reference.source_type,
+                table.c.definition_version == source.definition_version,
+            )
+        )
+        manifest = await _find_evidence_manifest(connection, reference.source_type, source.definition_version)
+        if manifest is not None and manifest.fingerprint == fingerprint:
+            try:
+                Draft202012Validator.check_schema(manifest.source_schema)
+                value = source.payload if isinstance(source, SourceObservation) else material
+                Draft202012Validator(manifest.source_schema, registry=Registry()).validate(value)
+            except JsonSchemaValidationError as error:
+                raise InvalidStoredPayloadError(
+                    "source", reference.source_type, "payload does not match the historical schema"
+                ) from error
+            except (SchemaError, Unresolvable):
+                self._validate_unbound_material(reference, envelope, source)
+        else:
+            self._validate_unbound_material(reference, envelope, source)
+        return source, material
+
+    def _validate_unbound_material(self, reference: SourceRef, envelope: _StoredSourcePayload, source: Source) -> None:
+        """Keep supported exact-read errors when no historical schema is available.
+
+        A live decoder validates only material of its own version; it never
+        contributes a declaration or reinterprets a valid historical binding.
+        """
+
+        if envelope.representation == "observation":
+            return  # SourceObservation already validated its envelope and payload.
+        try:
+            definition = self._registry.definition_for_name(reference.source_type)
+        except SourceDefinitionNotFoundError:
+            return
+        if definition.version == source.definition_version:
+            self._decode_envelope(reference.source_type, envelope)
+
+    @staticmethod
+    async def _bound_evidence(
+        connection: AsyncConnection,
+        scope_id: str,
+        materials: tuple[SourceEvidence, ...],
+        fingerprints: Mapping[tuple[str, str], str],
+    ) -> tuple[SourceEvidence, ...]:
+        if not materials:
+            return ()
+        table = SOURCE_EVIDENCE_BINDINGS_TABLE
+        rows = (
+            (
+                await connection.execute(
+                    select(table).where(
+                        table.c.scope_id == scope_id,
+                        tuple_(table.c.source_type, table.c.source_id).in_(
+                            tuple((value.source_ref.source_type, value.source_ref.source_id) for value in materials)
+                        ),
+                    )
+                )
+            )
+            .mappings()
+            .all()
+        )
+        bindings = {(row["source_type"], row["source_id"]): row for row in rows}
+        manifests: dict[tuple[str, str], SourceDefinitionManifest | None] = {}
+        evidence: list[SourceEvidence] = []
+        for value in materials:
+            identity = (value.source_ref.source_type, value.source_ref.source_id)
+            binding = bindings.get(identity)
+            if binding is None or (
+                binding["material_digest"] != value.material_digest
+                or binding["definition_name"] != value.definition_name
+                or binding["definition_version"] != value.definition_version
+                or not isinstance(binding["manifest_fingerprint"], str)
+            ):
+                evidence.append(value)
+                continue
+            definition_identity = (value.definition_name, value.definition_version)
+            if definition_identity not in manifests:
+                manifests[definition_identity] = await _find_evidence_manifest(connection, *definition_identity)
+            manifest = manifests[definition_identity]
+            if (
+                manifest is None
+                or manifest.fingerprint != binding["manifest_fingerprint"]
+                or (identity in fingerprints and fingerprints[identity] != manifest.fingerprint)
+            ):
+                evidence.append(value)
+                continue
+            try:
+                declaration = MemoryEvidenceDeclaration.model_validate_json(
+                    stored_bytes(binding["declaration"], column="declaration")
+                )
+            except (InvalidStoredColumnError, TypeError, ValueError):
+                declaration = MemoryEvidenceDeclaration()
+            evidence.append(
+                value.model_copy(
+                    update={
+                        "manifest_fingerprint": manifest.fingerprint,
+                        "declaration": declaration,
+                    }
+                )
+            )
+        return tuple(evidence)
+
+    async def _bind_memory_evidence(
+        self,
+        connection: AsyncConnection,
+        scope_id: str,
+        ref: SourceRef,
+        source: Source,
+    ) -> None:
+        """Freeze registration evidence only for a newly accepted Source row."""
+
+        if isinstance(source, SourceObservation):
+            manifest = await _find_evidence_manifest(connection, ref.source_type, source.definition_version)
+            if manifest is not None and manifest.fingerprint != source.definition_fingerprint:
+                manifest = None
+        else:
+            definition = self._registry.definition_for_source(source)
+            manifest = await SourceDefinitionManifestRepository().register_native(connection, definition)
+        declaration = (
+            MemoryEvidenceDeclaration() if manifest is None else self._registry.memory_evidence_for_source(source)
+        )
+        await connection.execute(
+            insert(SOURCE_EVIDENCE_BINDINGS_TABLE).values(
+                scope_id=scope_id,
+                source_type=ref.source_type,
+                source_id=ref.source_id,
+                material_digest=_source_material_digest(source),
+                definition_name=ref.source_type,
+                definition_version=source.definition_version,
+                manifest_fingerprint=None if manifest is None else manifest.fingerprint,
+                declaration=declaration.model_dump_json().encode("utf-8"),
+            )
+        )
+
     async def list(
         self,
         connection: AsyncConnection,
@@ -442,6 +713,36 @@ def _decode_legacy_payload(self, source_type: str, payload: bytes) -> tuple[Sour
         return cast(Source, source), decoded
 
 
+def _source_material_digest(source: Source) -> str:
+    # Preserve every material field except the narrow server witness upgrade.
+    material = source.model_dump(mode="json", by_alias=True)
+    if isinstance(source, ContentSource):
+        material.pop("handoff_receipt", None)
+    return f"sha256:{sha256(rfc8785.dumps(material)).hexdigest()}"
+
+
+def _management_write_source(source: Source) -> bool:
+    return (
+        isinstance(source, ContentSource)
+        and source.internal is not None
+        and source.internal.role == "lineage_only"
+        and source.internal.operation in {"artifact_create", "artifact_replace"}
+    )
+
+
+async def _find_evidence_manifest(
+    connection: AsyncConnection, name: str, version: str
+) -> SourceDefinitionManifest | None:
+    try:
+        return await SourceDefinitionManifestRepository().find(connection, name, version)
+    except (IdentityMismatchError, InvalidStoredColumnError, InvalidStoredPayloadError):
+        return None
+
+
+def _source_material_digest_material(material: Mapping[str, JsonValue]) -> str:
+    return f"sha256:{sha256(rfc8785.dumps(material)).hexdigest()}"
+
+
 def _dump_source(
     source: Source,
     *,
diff --git a/src/powercontext/builtin/runtime/atomic_memory.py b/src/powercontext/builtin/runtime/atomic_memory.py
index f79cb78fcb..70688c79e7 100644
--- a/src/powercontext/builtin/runtime/atomic_memory.py
+++ b/src/powercontext/builtin/runtime/atomic_memory.py
@@ -58,6 +58,10 @@
 )
 from powercontext.builtin.persistence.artifacts import ArtifactRepository
 from powercontext.builtin.persistence.atomic_memory import AtomicMemoryStateRepository, atomic_memory_state_expression
+from powercontext.builtin.persistence.atomic_memory_evidence import (
+    AtomicMemoryEvidenceAnnotation,
+    AtomicMemoryEvidenceRepository,
+)
 from powercontext.builtin.persistence.atomic_memory_index import (
     ATOMIC_MEMORY_RRF_CONSTANT,
     AtomicMemoryIndex,
@@ -186,6 +190,7 @@ def __init__(
         self.publisher = AtomicMemoryProjectionPublisher(
             index,
             embedding_model=embedding_model,
+            sources=artifacts.source_repository,
         )
         self.service = AtomicMemoryService(
             artifacts=artifacts,
@@ -222,6 +227,26 @@ async def get(self, artifact_id: str, *, revision: int | None = None, context=No
                 connection, self.scope_id, artifact_id, context, revision=revision
             )
 
+    async def evidence(self, artifact_id: str, *, revision: int, context=None) -> AtomicMemoryEvidenceAnnotation:
+        """Inspect neutral evidence facts for one authorized exact body revision.
+
+        A missing historical snapshot remains unknown; current registration
+        cannot attest a revision that did not accept a declaration.
+        """
+        application = self.application
+        async with (
+            deferred_decision_audit(context),
+            application.database.transaction(consistent_snapshot=True) as connection,
+        ):
+            record = await application.service.get(connection, self.scope_id, artifact_id, context, revision=revision)
+            return await AtomicMemoryEvidenceRepository().verify(
+                connection,
+                self.scope_id,
+                record.ref,
+                record.artifact.lineage.sources,
+                sources=application.artifacts.source_repository,
+            )
+
     async def list(
         self,
         *,
diff --git a/src/powercontext/builtin/runtime/atomic_memory_rebuild.py b/src/powercontext/builtin/runtime/atomic_memory_rebuild.py
index 90757129b0..67f429a9ca 100644
--- a/src/powercontext/builtin/runtime/atomic_memory_rebuild.py
+++ b/src/powercontext/builtin/runtime/atomic_memory_rebuild.py
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Explicit offline repair of only the current Atomic Memory projection."""
+"""Offline current-projection repair and verification of retained Atomic history."""
 
 from __future__ import annotations
 
@@ -25,12 +25,14 @@
 from sqlalchemy.engine import RowMapping
 from sqlalchemy.ext.asyncio import AsyncConnection
 
+from powercontext.artifacts import ArtifactRef
 from powercontext.builtin.artifacts.atomic_memory.models import AtomicMemory, AtomicMemoryRecord, AtomicMemoryStateValue
 from powercontext.builtin.artifacts.memory.canonical import canonical_json
 from powercontext.builtin.artifacts.search import analyze_text
 from powercontext.builtin.inference import EmbeddingModel
 from powercontext.builtin.persistence.artifacts import ArtifactRepository
 from powercontext.builtin.persistence.atomic_memory import AtomicMemoryStateRepository
+from powercontext.builtin.persistence.atomic_memory_evidence import AtomicMemoryEvidenceRepository
 from powercontext.builtin.persistence.atomic_memory_index import (
     AtomicMemoryIndex,
     AtomicMemoryIndexError,
@@ -48,6 +50,7 @@
 )
 from powercontext.builtin.persistence.database import AsyncDatabase
 from powercontext.builtin.persistence.oceanbase.atomic_memory_index import OceanBaseAtomicMemoryIndex
+from powercontext.builtin.persistence.sources import SourceRepository
 from powercontext.builtin.persistence.tables import ARTIFACT_HEADS_TABLE, ARTIFACTS_TABLE
 
 
@@ -114,6 +117,42 @@ async def _identity_batch(
     return tuple((str(row.scope_id), str(row.artifact_id)) for row in rows)
 
 
+async def _verify_history(
+    connection: AsyncConnection, artifacts: ArtifactRepository, batch_size: int, sources: SourceRepository | None
+) -> dict[str, int]:
+    table = ARTIFACTS_TABLE
+    evidence = AtomicMemoryEvidenceRepository()
+    after: tuple[str, str, int] | None = None
+    revisions = neutral_sources = 0
+    while True:
+        statement = select(table.c.scope_id, table.c.artifact_id, table.c.revision).where(
+            table.c.family == AtomicMemory.family
+        )
+        if after is not None:
+            statement = statement.where(tuple_(table.c.scope_id, table.c.artifact_id, table.c.revision) > after)
+        rows = (
+            await connection.execute(
+                statement.order_by(table.c.scope_id, table.c.artifact_id, table.c.revision).limit(batch_size)
+            )
+        ).all()
+        if not rows:
+            break
+        for row in rows:
+            scope_id, artifact_id, revision = str(row.scope_id), str(row.artifact_id), int(row.revision)
+            reference = ArtifactRef(family=AtomicMemory.family, artifact_id=artifact_id, revision=revision)
+            artifact = await artifacts.get(connection, scope_id, reference)
+            annotation = await evidence.verify(
+                connection, scope_id, reference, artifact.lineage.sources, sources=sources
+            )
+            revisions += 1
+            neutral_sources += sum(
+                source.declaration.authority == "untrusted" and source.declaration.verification == "unknown"
+                for source in annotation.sources
+            )
+            after = scope_id, artifact_id, revision
+    return {"historical_revisions": revisions, "neutral_sources": neutral_sources}
+
+
 async def _record(
     connection: AsyncConnection, artifacts: ArtifactRepository, scope_id: str, artifact_id: str
 ) -> AtomicMemoryRecord:
@@ -175,12 +214,20 @@ async def _projection_differences(
 
 
 async def verify_atomic_memory_current(
-    connection: AsyncConnection, index: AtomicMemoryIndex, /, *, batch_size: int = 100
+    connection: AsyncConnection,
+    index: AtomicMemoryIndex,
+    /,
+    *,
+    batch_size: int = 100,
+    sources: SourceRepository | None = None,
 ) -> AtomicMemoryProjectionRebuildReport:
-    """Check every current identity against authority, including post-migration creations."""
+    """Check current projections and every exact revision's retained evidence."""
 
+    if isinstance(batch_size, bool) or not 1 <= batch_size <= 1000:
+        raise AtomicMemoryIndexError("batch-size", "Projection rebuild batch size must be between 1 and 1000")
     await _require_authority(connection)
     artifacts = ArtifactRepository((AtomicMemory,))
+    history_counts = await _verify_history(connection, artifacts, batch_size, sources)
     profile = index.capabilities.embedding_profile
     fingerprint = None if profile is None else atomic_memory_profile_fingerprint(profile)
     errors: list[str] = []
@@ -218,7 +265,7 @@ async def verify_atomic_memory_current(
     return AtomicMemoryProjectionRebuildReport(
         ready=not errors,
         profile_fingerprint=fingerprint,
-        counts={"active_rows": active, "obsolete_rows": obsolete},
+        counts={"active_rows": active, "obsolete_rows": obsolete, **history_counts},
         errors=tuple(errors),
     )
 
@@ -241,6 +288,7 @@ async def rebuild_atomic_memory_projection(
     maintenance_confirmed: bool,
     embedding_model: EmbeddingModel | None = None,
     batch_size: int = 100,
+    sources: SourceRepository | None = None,
 ) -> AtomicMemoryProjectionRebuildReport:
     """Repair current without changing revisions, Family state or retained history.
 
@@ -309,7 +357,7 @@ async def rebuild_atomic_memory_projection(
         )
         await connection.execute(index.table.delete().where(_obsolete_current(index)))
         removed += obsolete
-        report = await verify_atomic_memory_current(connection, index, batch_size=batch_size)
+        report = await verify_atomic_memory_current(connection, index, batch_size=batch_size, sources=sources)
     return report.model_copy(
         update={
             "counts": {
diff --git a/src/powercontext/builtin/runtime/composition.py b/src/powercontext/builtin/runtime/composition.py
index b2c0cd1b99..c2231c4e63 100644
--- a/src/powercontext/builtin/runtime/composition.py
+++ b/src/powercontext/builtin/runtime/composition.py
@@ -197,7 +197,7 @@
 )
 from powercontext.builtin.statistics import ModelUsagePurpose
 from powercontext.errors import InvalidSourceProjectionError, SourceProjectionNotFoundError
-from powercontext.sources import Source, SourceDefinitionRegistry, SourceProjectionKey
+from powercontext.sources import AdapterSourceDefinition, Source, SourceDefinitionRegistry, SourceProjectionKey
 
 if TYPE_CHECKING:
     from pydantic_ai.models import Model
@@ -798,7 +798,17 @@ def _validate_processing_source_registry(
         type(source_registry) is not SourceDefinitionRegistry
         or len(definitions) != len(builtin_definitions)
         or any(
-            definition is not expected for definition, expected in zip(definitions, builtin_definitions, strict=True)
+            not (
+                definition is expected
+                or (
+                    type(definition) is AdapterSourceDefinition
+                    and type(expected) is AdapterSourceDefinition
+                    and definition.adapter is expected.adapter
+                    and definition.version == expected.version
+                    and definition.projections is expected.projections
+                )
+            )
+            for definition, expected in zip(definitions, builtin_definitions, strict=True)
         )
     ):
         raise BuiltinConfigurationError("artifact-processing-source-registry")
diff --git a/src/powercontext/sources/__init__.py b/src/powercontext/sources/__init__.py
index 91e1a27b08..5469244670 100644
--- a/src/powercontext/sources/__init__.py
+++ b/src/powercontext/sources/__init__.py
@@ -30,6 +30,7 @@
     SourceDefinitionRegistry,
     SourceProjection,
 )
+from powercontext.sources.evidence import MemoryEvidenceDeclaration, SourceEvidence
 from powercontext.sources.models import Source, SourceMaterialization, SourceProjectionKey, SourceRef
 from powercontext.sources.observations import (
     SourceDefinitionManifest,
@@ -53,6 +54,7 @@
     "ConnectorRunSession",
     "ConnectorRunStatus",
     "ConnectorSubmissionStatus",
+    "MemoryEvidenceDeclaration",
     "Source",
     "SourceAdapter",
     "SourceCatalog",
@@ -60,6 +62,7 @@
     "SourceDefinition",
     "SourceDefinitionManifest",
     "SourceDefinitionRegistry",
+    "SourceEvidence",
     "SourceMaterialization",
     "SourceObservation",
     "SourceProjection",
diff --git a/src/powercontext/sources/definitions.py b/src/powercontext/sources/definitions.py
index d9c54572dc..18f6bd5162 100644
--- a/src/powercontext/sources/definitions.py
+++ b/src/powercontext/sources/definitions.py
@@ -18,6 +18,7 @@
 
 from collections.abc import Iterable, Mapping
 from dataclasses import dataclass
+from types import MappingProxyType
 from typing import Any, Generic, Protocol, TypeVar, cast
 
 from pydantic import BaseModel, JsonValue, TypeAdapter
@@ -34,6 +35,7 @@
     SourceProjectionNotFoundError,
 )
 from powercontext.sources.adapters import SourceAdapter
+from powercontext.sources.evidence import MemoryEvidenceDeclaration
 from powercontext.sources.models import Source, SourceProjectionKey
 
 InputT = TypeVar("InputT")
@@ -42,6 +44,7 @@
 
 _JSON_VALUE = TypeAdapter(JsonValue)
 _AnySourceAdapter = SourceAdapter[Any, Any, Any]
+_NEUTRAL_MEMORY_EVIDENCE = MemoryEvidenceDeclaration()
 
 
 class SourceProjection(Protocol[SourceT]):
@@ -60,6 +63,7 @@ class SourceDefinition(SourceAdapter[InputT, SourceT, ValueT_co], Protocol[Input
 
     version: str
     projections: tuple[SourceProjection[SourceT], ...]
+    memory_evidence: MemoryEvidenceDeclaration | None
 
 
 @dataclass(frozen=True, slots=True)
@@ -69,6 +73,7 @@ class AdapterSourceDefinition(Generic[InputT, SourceT, ValueT_co]):
     adapter: SourceAdapter[InputT, SourceT, ValueT_co]
     version: str = "1"
     projections: tuple[SourceProjection[SourceT], ...] = ()
+    memory_evidence: MemoryEvidenceDeclaration | None = None
 
     @property
     def input_class(self) -> type[InputT]:
@@ -93,14 +98,28 @@ async def read(self, source: SourceT, /) -> ValueT_co:
 
 
 class SourceDefinitionRegistry:
-    """Provide one immutable routing view for Source persistence and consumers."""
-
-    def __init__(self, definitions: Iterable[_AnySourceDefinition], /) -> None:
+    """Provide one immutable routing and trusted declaration view.
+
+    Native Definitions optionally supply ``memory_evidence``. Deployment-owned
+    ``remote_declarations`` keys are exact ``(name, version, fingerprint)``
+    manifest identities, never worker payload or projection values. Declarations
+    are validated and copied at construction; invalid or unavailable values are
+    neutral and never prevent registration of an otherwise valid Definition.
+    """
+
+    def __init__(
+        self,
+        definitions: Iterable[_AnySourceDefinition],
+        /,
+        *,
+        remote_declarations: Mapping[tuple[str, str, str], MemoryEvidenceDeclaration] | None = None,
+    ) -> None:
         by_input: dict[type[object], _AnySourceDefinition] = {}
         by_source: dict[type[Source], _AnySourceDefinition] = {}
         by_name: dict[str, _AnySourceDefinition] = {}
         projections: dict[type[Source], Mapping[SourceProjectionKey, SourceProjection[Any]]] = {}
         registered: list[_AnySourceDefinition] = []
+        memory_evidence: dict[type[Source], MemoryEvidenceDeclaration] = {}
 
         for definition in definitions:
             input_class, source_class = _validate_definition(definition)
@@ -122,6 +141,11 @@ def __init__(self, definitions: Iterable[_AnySourceDefinition], /) -> None:
             by_source[source_class] = definition
             by_name[definition.name] = definition
             projections[source_class] = projection_routes
+            try:
+                declaration = getattr(definition, "memory_evidence", None)
+            except Exception:  # Declaration providers may be unavailable; routing remains valid.
+                declaration = None
+            memory_evidence[source_class] = _snapshot_memory_evidence(declaration)
             registered.append(definition)
 
         self._definitions = tuple(registered)
@@ -129,6 +153,10 @@ def __init__(self, definitions: Iterable[_AnySourceDefinition], /) -> None:
         self._by_source = by_source
         self._by_name = by_name
         self._projections = projections
+        self._memory_evidence = MappingProxyType(memory_evidence)
+        self._remote_declarations = MappingProxyType({
+            key: _snapshot_memory_evidence(value) for key, value in (remote_declarations or {}).items()
+        })
 
     @classmethod
     def from_adapters(cls, adapters: Iterable[_AnySourceAdapter], /) -> SourceDefinitionRegistry:
@@ -161,6 +189,23 @@ def definition_for_source(self, source: object, /) -> _AnySourceDefinition:
             )
         return definition
 
+    def memory_evidence_for_source(self, source: Source, /) -> MemoryEvidenceDeclaration:
+        """Read the frozen trusted declaration using the exact registered route."""
+
+        from powercontext.sources.observations import SourceObservation
+
+        if isinstance(source, SourceObservation):
+            return self.memory_evidence_for_manifest(
+                source.source_type, source.definition_version, source.definition_fingerprint
+            )
+        self.definition_for_source(source)
+        return self._memory_evidence[type(source)]
+
+    def memory_evidence_for_manifest(self, name: str, version: str, fingerprint: str, /) -> MemoryEvidenceDeclaration:
+        """Return only deployment-configured evidence for an exact manifest."""
+
+        return self._remote_declarations.get((name, version, fingerprint), _NEUTRAL_MEMORY_EVIDENCE)
+
     async def resolve(self, value: object, /) -> Source:
         input_class = type(value)
         try:
@@ -203,6 +248,18 @@ def project(self, source: Source, key: SourceProjectionKey, /) -> JsonValue:
             raise InvalidSourceProjectionError(key.name, "result", "must match the declared output schema") from error
 
 
+def _snapshot_memory_evidence(value: object) -> MemoryEvidenceDeclaration:
+    if value is None:
+        return _NEUTRAL_MEMORY_EVIDENCE
+    try:
+        # Revalidate even constructed models, and never retain caller-owned data.
+        if isinstance(value, MemoryEvidenceDeclaration):
+            value = value.model_dump(mode="python")
+        return MemoryEvidenceDeclaration.model_validate(value)
+    except (TypeError, ValueError):
+        return _NEUTRAL_MEMORY_EVIDENCE
+
+
 def _validate_definition(definition: object) -> tuple[type[object], type[Source]]:
     definition_type = type(definition)
     input_class = getattr(definition, "input_class", None)
diff --git a/src/powercontext/sources/evidence.py b/src/powercontext/sources/evidence.py
new file mode 100644
index 0000000000..33eebfed9f
--- /dev/null
+++ b/src/powercontext/sources/evidence.py
@@ -0,0 +1,60 @@
+# 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.
+
+"""Typed declarations and exact material facts accepted at Source registration."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from powercontext.sources.models import SourceMaterialization, SourceRef
+
+
+class MemoryEvidenceDeclaration(BaseModel):
+    """A trusted adapter's claim, not a score, permission or verification engine.
+
+    ``declaration_version`` identifies this contract, independently of the
+    Definition version and Artifact content revision. Missing or unsupported
+    declarations resolve to ``untrusted`` / ``unknown``.
+    """
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    authority: Literal["untrusted", "user_asserted", "repository_attested", "system_attested"] = "untrusted"
+    verification: Literal["unknown", "verified", "not_verified"] = "unknown"
+    declaration_version: Literal["1"] = "1"
+
+
+class SourceEvidence(BaseModel):
+    """Retain one exact Source material binding without duplicating its body.
+
+    The digest checks equality of canonical Source JSON, excluding only the
+    server-owned ContentSource handoff receipt witness. It authenticates neither
+    the Source nor its declaration. Each Source retains its own declaration.
+    """
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    source_ref: SourceRef
+    materialization: SourceMaterialization
+    material_digest: str
+    definition_name: str
+    definition_version: str
+    manifest_fingerprint: str | None = None
+    declaration: MemoryEvidenceDeclaration = Field(default_factory=MemoryEvidenceDeclaration)
+
+
+__all__ = ["MemoryEvidenceDeclaration", "SourceEvidence"]
diff --git a/src/powercontext/sources/models.py b/src/powercontext/sources/models.py
index 6e2cd48df6..6235d78f9e 100644
--- a/src/powercontext/sources/models.py
+++ b/src/powercontext/sources/models.py
@@ -16,7 +16,7 @@
 
 from enum import StrEnum
 
-from pydantic import BaseModel, ConfigDict, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 from powercontext.errors import InvalidSourceReferenceError
 from powercontext.limits import MAX_SOURCE_ID_LENGTH, MAX_SOURCE_TYPE_LENGTH
@@ -60,10 +60,11 @@ def validate_key_part(cls, value: str, info) -> str:
 class Source(BaseModel):
     """Base value for an adapter-owned Source description."""
 
-    name: str
-    definition_version: str = "1"
-    materialization: SourceMaterialization
-    description: str | None = None
+    # Envelope keys remain canonical when a Definition generates aliases for its own fields.
+    name: str = Field(serialization_alias="name")
+    definition_version: str = Field(default="1", serialization_alias="definition_version")
+    materialization: SourceMaterialization = Field(serialization_alias="materialization")
+    description: str | None = Field(default=None, serialization_alias="description")
 
     @field_validator("definition_version")
     @classmethod
diff --git a/src/powercontext/sources/observations.py b/src/powercontext/sources/observations.py
index 576e29fb79..c9bb8a802f 100644
--- a/src/powercontext/sources/observations.py
+++ b/src/powercontext/sources/observations.py
@@ -163,7 +163,7 @@ def projection(self, key: SourceProjectionKey, /) -> JsonValue:
 def manifest_for_definition(definition: SourceDefinition[Any, Any, Any], /) -> SourceDefinitionManifest:
     """Build the immutable declaration transported by a remote worker."""
 
-    source_schema = _json_object(definition.source_class.model_json_schema())
+    source_schema = _json_object(definition.source_class.model_json_schema(mode="serialization"))
     projections = tuple(
         SourceProjectionManifest(
             key=SourceProjectionKey(name=projection.name, version=projection.version),
@@ -199,7 +199,7 @@ def project_source_for_transport(
             "remote Source observations must retain their canonical value",
         )
     manifest = manifest_for_definition(definition)
-    payload = _json_object(source.model_dump(mode="json"))
+    payload = _json_object(source.model_dump(mode="json", by_alias=True))
     projections = tuple(
         SourceProjectionValue(key=key, value=registry.project(source, key)) for key in registry.projection_keys(source)
     )
diff --git a/tests/builtin/persistence/test_mysql_migration_runner.py b/tests/builtin/persistence/test_mysql_migration_runner.py
index 17365988d1..aa083f4786 100644
--- a/tests/builtin/persistence/test_mysql_migration_runner.py
+++ b/tests/builtin/persistence/test_mysql_migration_runner.py
@@ -298,10 +298,10 @@ def test_mysql_initialization_and_shared_evidence_noop(backend: _BackendTarget)
     assert result.state == "ready"
     assert result.changed
     assert result.backup_state == "not_required"
-    assert result.revision == "p0004"
+    assert result.revision == "p0005"
     replacement = backend.runner()
     assert replacement.plan().state == "ready"
-    assert replacement.verify().revision == "p0004"
+    assert replacement.verify().revision == "p0005"
     assert not replacement.apply().changed
 
 
@@ -321,7 +321,7 @@ def test_mysql_operator_backup_choices_preserve_baseline_data_without_fork(
     result = _apply(runner, policy=policy)
     assert result.backup_state == ("user_confirmed" if policy == "manual" else "skipped")
     state = _read_state(backend)
-    assert state["revision"] == "p0004"
+    assert state["revision"] == "p0005"
     assert state["content"] == [(b'{"content":"preserved"}',)]
     assert state["tags"] == [("Keep",)]
     assert "memory_citations" in state["columns"]["pc_artifacts"]
@@ -389,7 +389,7 @@ def test_mysql_committed_partial_ddl_resumes_from_shared_original_evidence(backe
     replacement = _acknowledge_known_completed_interruption(backend, recovery)
     assert _apply(replacement, policy="manual").state == "ready"
     state = _read_state(backend)
-    assert state["revision"] == "p0004"
+    assert state["revision"] == "p0005"
     assert state["content"] == [(b'{"content":"preserved"}',)]
     assert state["tags"] == [("Keep",)]
     assert replacement.verify().state == "ready"
@@ -435,7 +435,7 @@ def test_mysql_empty_source_recovers_baseline_and_citation_ddl_without_probing_b
     )
     result = _apply(replacement)
     assert result.backup_state == "not_required"
-    assert replacement.verify().revision == "p0004"
+    assert replacement.verify().revision == "p0005"
     assert _read_state(backend)["content"] == []
 
 
@@ -452,7 +452,7 @@ def test_mysql_baseline_adoption_recovers_an_empty_committed_version_table(backe
     assert recovered.adopt_baseline
     replacement = _acknowledge_known_completed_interruption(backend, recovered)
     assert _apply(replacement, policy="manual").state == "ready"
-    assert replacement.verify().revision == "p0004"
+    assert replacement.verify().revision == "p0005"
     assert _read_state(backend)["tags"] == [("Keep",)]
 
 
@@ -478,7 +478,7 @@ def interrupt(journal, evidence) -> None:
     assert recovery.source_revision == "p0002"
     replacement = _acknowledge_known_completed_interruption(backend, recovery)
     assert _apply(replacement, policy="manual").state == "ready"
-    assert replacement.verify().revision == "p0004"
+    assert replacement.verify().revision == "p0005"
     assert _read_state(backend)["content"] == [(b'{"content":"preserved"}',)]
 
 
@@ -487,7 +487,7 @@ def test_mysql_completed_head_with_an_interrupted_run_requires_exact_acknowledge
         pytest.skip("embedded seekdb closes its engine and has no remote in-flight run receipt")
     _load_baseline(backend)
     assert _apply(backend.runner(), policy="manual").state == "ready"
-    assert backend.runner().verify().revision == "p0004"
+    assert backend.runner().verify().revision == "p0005"
 
     def committed_write_then_interrupt(
         connection: Connection | None, _identity: BackendIdentity, verify: Callable[[], None]
@@ -501,12 +501,12 @@ def committed_write_then_interrupt(
     with pytest.raises(_InterruptedDDL):
         backend.connections().run(committed_write_then_interrupt, writable=True)
     before = _read_state(backend)
-    assert before["revision"] == "p0004"
+    assert before["revision"] == "p0005"
     assert before["tags"] == [("Confirmed",)]
     replacement = backend.runner(backup_factory=_forbidden_automatic_backup)
     recovery = replacement.plan()
     assert recovery.state == "recovery_required"
-    assert recovery.source_revision == "p0004"
+    assert recovery.source_revision == "p0005"
     assert recovery.revisions == ()
     pending = (recovery.coordination or {}).get("pending_run_id")
     assert pending
@@ -525,6 +525,6 @@ def committed_write_then_interrupt(
     assert result.state == "ready"
     assert not result.changed
     assert _read_state(backend) == before
-    assert backend.runner().verify().revision == "p0004"
+    assert backend.runner().verify().revision == "p0005"
     assert not backend.runner().apply().changed
     assert "pending_run_id" not in (backend.runner().plan().coordination or {})
diff --git a/tests/builtin/persistence/test_sources.py b/tests/builtin/persistence/test_sources.py
index 22e080499f..b74a798a7e 100644
--- a/tests/builtin/persistence/test_sources.py
+++ b/tests/builtin/persistence/test_sources.py
@@ -17,6 +17,8 @@
 import asyncio
 
 import pytest
+from pydantic import BaseModel, ConfigDict, Field
+from pydantic.alias_generators import to_camel
 from sqlalchemy import insert
 
 from powercontext import InvalidSourceDefinitionError
@@ -29,8 +31,11 @@
 from powercontext.builtin.persistence.sources import SourceRepository
 from powercontext.builtin.persistence.sqlite import SQLiteConfig, SQLiteProfile
 from powercontext.builtin.persistence.tables import SHARED_TABLES, SOURCES_TABLE
+from powercontext.builtin.runtime import SubmitSourceObservation
 from powercontext.builtin.runtime.relational import RelationalContexts
 from powercontext.sources import (
+    AdapterSourceDefinition,
+    Source,
     SourceDefinitionRegistry,
     SourceMaterialization,
     SourceObservation,
@@ -46,6 +51,43 @@
 )
 
 
+class SerializationAliasInput(BaseModel):
+    source_id: str
+    body: str
+
+
+class SerializationAliasSource(Source):
+    body: str = Field(serialization_alias="wire_text")
+
+
+class CamelCaseSource(SerializationAliasSource):
+    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
+
+
+class UpperCaseSource(SerializationAliasSource):
+    model_config = ConfigDict(alias_generator=str.upper, populate_by_name=True)
+
+
+class SerializationAliasAdapter:
+    name = "serialization-alias"
+    input_class = SerializationAliasInput
+
+    def __init__(self, source_class: type[SerializationAliasSource]) -> None:
+        self.source_class = source_class
+
+    async def resolve(self, value: SerializationAliasInput, /) -> SerializationAliasSource:
+        return self.source_class(
+            name=value.source_id,
+            definition_version="2",
+            materialization=SourceMaterialization.CAPTURED,
+            description="Remote alias description.",
+            body=value.body,
+        )
+
+    async def read(self, source: SerializationAliasSource, /) -> str:
+        return source.body
+
+
 def test_two_source_adapters_share_one_repository_and_journal() -> None:
     async def scenario() -> None:
         async with repository_profile() as (profile, repositories):
@@ -149,6 +191,40 @@ async def scenario() -> None:
     asyncio.run(scenario())
 
 
+@pytest.mark.parametrize("source_class", (SerializationAliasSource, CamelCaseSource, UpperCaseSource))
+def test_remote_observation_ingestion_accepts_serialization_aliases(
+    source_class: type[SerializationAliasSource],
+) -> None:
+    async def scenario() -> None:
+        worker = SourceDefinitionRegistry((
+            AdapterSourceDefinition(SerializationAliasAdapter(source_class), version="2"),
+        ))
+        source = await worker.resolve(SerializationAliasInput(source_id="remote-1", body="Remote note text."))
+        manifest = manifest_for_definition(worker.definition_for_name("serialization-alias"))
+        observation = project_source_for_transport(worker, source)
+
+        async with repository_profile() as (profile, _repositories):
+            contexts = RelationalContexts(database=profile.database, source_registry=SourceDefinitionRegistry(()))
+            await contexts.register_source_definition(manifest)
+            receipt = await contexts.submit_source_observation(
+                SubmitSourceObservation(scope_id="scope-a", observation=observation)
+            )
+            context = await contexts.get("scope-a")
+            stored = await context.sources.get(observation)
+
+        assert receipt.source_ref == SourceRef(source_type="serialization-alias", source_id="remote-1")
+        assert isinstance(stored, SourceObservation)
+        assert stored.payload == {
+            "name": "remote-1",
+            "definition_version": "2",
+            "materialization": "captured",
+            "description": "Remote alias description.",
+            "wire_text": "Remote note text.",
+        }
+
+    asyncio.run(scenario())
+
+
 def test_repository_rejects_a_scope_outside_the_relational_identity_baseline() -> None:
     async def scenario() -> None:
         async with repository_profile() as (profile, repositories):
diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py
index bdb820e071..a1f2778fa8 100644
--- a/tests/e2e/conftest.py
+++ b/tests/e2e/conftest.py
@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import os
 from collections.abc import Iterator
 from pathlib import Path
@@ -26,6 +27,7 @@
 from pydantic import SecretStr
 from sqlalchemy.engine import make_url
 
+from powercontext.builtin.persistence.migrations.backup import identify_fork_server
 from powercontext.builtin.persistence.oceanbase import OceanBaseConfig
 from powercontext.builtin.persistence.seekdb import SeekDBConfig
 from powercontext.builtin.persistence.sqlite import SQLiteConfig
@@ -63,6 +65,17 @@ async def manage(*, create: bool) -> None:
         nonlocal owned
         async with migration_profile(config) as profile, profile.database.transaction() as connection:
             assert (await connection.exec_driver_sql("SELECT DATABASE()")).scalar_one() == url.database
+            version = str((await connection.exec_driver_sql("SELECT VERSION()")).scalar_one())
+            comment = str((await connection.exec_driver_sql("SELECT @@version_comment")).scalar_one())
+            product, engine_version = identify_fork_server(version, comment)
+            if product != "oceanbase_ai" or engine_version is None:
+                pytest.fail(
+                    "Acceptance requires an identified real OceanBase version, not MySQL or SeekDB", pytrace=False
+                )
+            if create:
+                evidence = {"backend": "oceanbase", "version": version, "engine_version": engine_version}
+                request.node.user_properties.append(("backend_identity", evidence))
+                print(json.dumps(evidence, ensure_ascii=False))
             await connection.exec_driver_sql(
                 f"CREATE DATABASE `{database}`" if create else f"DROP DATABASE `{database}`"
             )
diff --git a/tests/e2e/test_atomic_memory_evidence_neutral_recall.py b/tests/e2e/test_atomic_memory_evidence_neutral_recall.py
new file mode 100644
index 0000000000..c3ad6ee50b
--- /dev/null
+++ b/tests/e2e/test_atomic_memory_evidence_neutral_recall.py
@@ -0,0 +1,131 @@
+# 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.
+
+"""Accepted Source declarations do not alter recall or consumer revision citations."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from dataclasses import replace
+from typing import cast
+
+import httpx
+import pytest
+
+from powercontext.builtin.artifacts.handoff import PrepareHandoff
+from powercontext.builtin.runtime import (
+    BuiltinConfig,
+    HandoffArtifactCitation,
+    PrepareContextRequest,
+    RuntimeConfig,
+    open_builtin_contexts,
+    open_builtin_runtime,
+)
+from powercontext.builtin.scope import ScopeDraft
+from powercontext.builtin.sources import CONTENT_SOURCE_DEFINITION
+from powercontext.server.app import ServerApplication, create_app
+from powercontext.sources import AdapterSourceDefinition, SourceDefinitionRegistry, SourceRef
+from powercontext.sources.evidence import MemoryEvidenceDeclaration
+from tests.builtin.persistence.contract import NoteAdapter, NoteInput
+from tests.e2e.atomic_memory_migration_backend import MigrationBackend
+from tests.e2e.test_atomic_memory_source_evidence import _commit
+from tests.e2e.test_handoff_runtime import _EchoHandoffPipeline
+
+pytestmark = pytest.mark.parametrize("migration_backend", ["sqlite", "oceanbase"], indirect=True)
+
+
+def test_source_declarations_preserve_recall_and_consumers(migration_backend: MigrationBackend) -> None:
+    async def scenario() -> None:
+        config = BuiltinConfig(
+            database=migration_backend.config,
+            runtime=RuntimeConfig(artifact_processing_families=(), atomic_memory_related_mode="fts"),
+        )
+        arms = []
+        neutral = MemoryEvidenceDeclaration()
+        candidates = (
+            (SourceRef(source_type="note", source_id="note"), "memory-a", "alpha acceptance"),
+            (SourceRef(source_type="content", source_id="source"), "memory-b", "alpha guidance"),
+        )
+        for declaration in (
+            neutral,
+            MemoryEvidenceDeclaration(authority="system_attested", verification="verified"),
+        ):
+            registry = SourceDefinitionRegistry((
+                replace(CONTENT_SOURCE_DEFINITION, memory_evidence=declaration),
+                AdapterSourceDefinition(NoteAdapter()),
+            ))
+            async with open_builtin_runtime(
+                config,
+                scheduler_path=migration_backend.directory / "consumer-scheduler.db",
+                source_registry=registry,
+                handoff_pipeline=_EchoHandoffPipeline(),
+            ) as runtime:
+                assert runtime.scopes is not None and runtime.atomic_memory is not None
+                scope = await runtime.scopes.create(
+                    ScopeDraft(
+                        title="Recall",
+                        summary="Evidence neutrality.",
+                        idempotency_key=declaration.authority,
+                    )
+                )
+                scope_id = scope.scope_id
+                async with open_builtin_contexts(config, source_registry=registry) as contexts:
+                    await contexts.records.capture_source(scope_id, "content", "source", "alpha guidance", {})
+                    context = await contexts.get(scope_id)
+                    note = await context.sources.resolve(NoteInput(note_id="note", body="alpha acceptance"))
+                    await context.sources.add(note)
+                    memories = []
+                    for source, artifact_id, text in candidates:
+                        memory = await _commit(contexts, scope_id, artifact_id, text, sources=(source,))
+                        annotation = await contexts.atomic_memory.for_scope(scope_id).evidence(artifact_id, revision=1)
+                        assert annotation.artifact_ref == memory.ref
+                        expected = declaration if source.source_type == "content" else neutral
+                        assert tuple((item.source_ref, item.declaration) for item in annotation.sources) == (
+                            (source, expected),
+                        )
+                        memories.append(memory)
+                page = await runtime.atomic_memory.for_scope(scope_id).search("alpha", mode="text", limit=2)
+                assert tuple((hit.hit.artifact_ref, hit.text, hit.matched_by) for hit in page.hits) == tuple(
+                    (memory.ref, text, ("text",)) for memory, (_, _, text) in zip(memories, candidates, strict=True)
+                )
+                recalled = tuple((hit.hit.artifact_ref.model_dump(mode="json"), hit.text) for hit in page.hits)
+                app = create_app(application=cast(ServerApplication, runtime))
+                async with httpx.AsyncClient(
+                    transport=httpx.ASGITransport(app=app), base_url="http://testserver"
+                ) as client:
+                    response = await client.post("/v1/memory/search", json={"scope_id": scope_id, "query": "alpha"})
+                assert response.status_code == 200, response.text
+                public = response.json()
+                assert (
+                    tuple((item["memory"]["artifact"], item["memory"]["text"]) for item in public["hits"]) == recalled
+                )
+                prepared = await runtime.context.for_scope(scope_id).prepare(PrepareContextRequest(query="alpha"))
+                assert prepared.status == "ready" and prepared.content is not None
+                items = json.loads(prepared.content.splitlines()[-2])["items"]
+                assert tuple((item["citation"]["artifact_ref"], item["content"]) for item in items) == recalled
+                citation = HandoffArtifactCitation(artifact_ref=page.hits[-1].hit.artifact_ref)
+                handoffs = runtime.handoff.for_scope(scope_id)
+                finalized = await handoffs.finalize(
+                    await handoffs.prepare(
+                        PrepareHandoff(objective="Keep the recalled revision.", evidence=(citation,))
+                    )
+                )
+                assert finalized.content.state[0].citations == (citation,)
+                assert finalized.content.next_action is not None
+                assert finalized.content.next_action.citations == (citation,)
+                arms.append((page.hits, public, prepared.content.replace(scope_id, "scope"), finalized.content))
+        assert arms[0] == arms[1]
+
+    asyncio.run(scenario())
diff --git a/tests/e2e/test_atomic_memory_projection_rebuild.py b/tests/e2e/test_atomic_memory_projection_rebuild.py
index 3075699659..5263ff1a0c 100644
--- a/tests/e2e/test_atomic_memory_projection_rebuild.py
+++ b/tests/e2e/test_atomic_memory_projection_rebuild.py
@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import asyncio
+from dataclasses import replace
 from pathlib import Path
 
 import pytest
@@ -29,6 +30,9 @@
 from powercontext.builtin.persistence.tables import ARTIFACT_HEADS_TABLE, ARTIFACTS_TABLE
 from powercontext.builtin.runtime import BuiltinConfig, open_builtin_contexts
 from powercontext.builtin.runtime.atomic_memory_rebuild import rebuild_atomic_memory_projection
+from powercontext.builtin.sources import CONTENT_SOURCE_DEFINITION
+from powercontext.sources import SourceDefinitionRegistry
+from powercontext.sources.evidence import MemoryEvidenceDeclaration
 from tests.e2e.atomic_memory_migration_fixture import SCOPE, _atomic, _seed_and_migrate
 from tests.legacy_memory import add_legacy_citation_columns
 
@@ -36,6 +40,12 @@
 @pytest.mark.parametrize("retention", ["complete", "archive_only", "entries_only", "removed", "unreadable"])
 def test_rebuild_uses_current_authority_independently_of_legacy_retention(tmp_path: Path, retention: str) -> None:
     config = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'projection.db'}")
+    strong_registry = SourceDefinitionRegistry((
+        replace(
+            CONTENT_SOURCE_DEFINITION,
+            memory_evidence=MemoryEvidenceDeclaration(authority="system_attested", verification="verified"),
+        ),
+    ))
 
     async def scenario() -> None:
         await _seed_and_migrate(config)
@@ -43,6 +53,8 @@ async def scenario() -> None:
             memory = contexts.atomic_memory.for_scope(SCOPE)
             current = await memory.get(_atomic("alpha", 2).artifact_id)
             historical = await memory.get(_atomic("alpha", 1).artifact_id, revision=1)
+            annotation = await memory.evidence(historical.ref.artifact_id, revision=1)
+            assert all(source.declaration == MemoryEvidenceDeclaration() for source in annotation.sources)
         async with SQLiteProfile.open(config, tables=()) as profile:
             async with profile.database.transaction() as connection:
                 if retention in {"entries_only", "removed"}:
@@ -63,10 +75,13 @@ async def scenario() -> None:
             )
             assert report.ready, report.errors
             assert report.counts["rebuilt_rows"] == 2
-        async with open_builtin_contexts(BuiltinConfig(database=config)) as contexts:
+            assert report.counts["historical_revisions"] == 3
+            assert report.counts["neutral_sources"] == 4
+        async with open_builtin_contexts(BuiltinConfig(database=config), source_registry=strong_registry) as contexts:
             memory = contexts.atomic_memory.for_scope(SCOPE)
             assert await memory.get(current.ref.artifact_id) == current
             assert await memory.get(historical.ref.artifact_id, revision=1) == historical
+            assert await memory.evidence(historical.ref.artifact_id, revision=1) == annotation
             result = await memory.search("Tasks A and C", mode="text")
             assert current.ref in tuple(hit.hit.artifact_ref for hit in result.hits)
 
diff --git a/tests/e2e/test_atomic_memory_source_evidence.py b/tests/e2e/test_atomic_memory_source_evidence.py
new file mode 100644
index 0000000000..5fe44a12e3
--- /dev/null
+++ b/tests/e2e/test_atomic_memory_source_evidence.py
@@ -0,0 +1,228 @@
+# 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.
+
+"""Versioned source evidence acceptance on disposable SQLite and OceanBase."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from dataclasses import replace
+
+import pytest
+from sqlalchemy import event, text
+from sqlalchemy.exc import SQLAlchemyError
+
+from powercontext.artifacts import ArtifactLineage
+from powercontext.builtin.artifacts.atomic_memory import AtomicMemoryContent
+from powercontext.builtin.artifacts.atomic_memory.extraction import (
+    AtomicMemoryCandidate,
+    AtomicMemoryExtractionOutput,
+)
+from powercontext.builtin.inference import GenerationResult
+from powercontext.builtin.runtime import BuiltinConfig, SubmitSourceObservation, open_builtin_contexts
+from powercontext.builtin.runtime.atomic_memory_rebuild import rebuild_atomic_memory_projection
+from powercontext.builtin.sources import CONTENT_SOURCE_DEFINITION
+from powercontext.sources import AdapterSourceDefinition, SourceDefinitionRegistry, SourceRef
+from powercontext.sources.evidence import MemoryEvidenceDeclaration
+from powercontext.sources.observations import manifest_for_definition, project_source_for_transport
+from tests.builtin.persistence.contract import NoteAdapter, NoteInput
+from tests.e2e.atomic_memory_migration_backend import MigrationBackend
+from tests.e2e.dream_support import atomic_memory_pipeline
+
+pytestmark = pytest.mark.parametrize("migration_backend", ["sqlite", "oceanbase"], indirect=True)
+
+
+async def _commit(contexts, scope_id, artifact_id, text_value, *, sources=(), artifacts=(), expected_revision=None):
+    application = contexts.atomic_memory
+    async with contexts.database.transaction() as connection:
+        plan = await application.service.inspect_change(
+            connection,
+            scope_id,
+            artifact_id,
+            AtomicMemoryContent(kind="fact", text=text_value),
+            None,
+            expected_revision=expected_revision,
+            lineage=ArtifactLineage(sources=sources, artifacts=artifacts),
+        )
+    prepared = await application.service.prepare_change(plan)
+    async with contexts.database.transaction() as connection:
+        result = await application.service.commit(connection, prepared, None)
+    return result.primary
+
+
+def test_mixed_evidence_survives_failed_revision_restart_and_rebuild(migration_backend: MigrationBackend) -> None:
+    strong = MemoryEvidenceDeclaration(authority="repository_attested", verification="verified")
+    asserted = MemoryEvidenceDeclaration(authority="user_asserted", verification="not_verified")
+    remote = MemoryEvidenceDeclaration(authority="system_attested", verification="verified")
+
+    class RemoteNoteAdapter(NoteAdapter):
+        name = "remote-note"
+
+    worker = SourceDefinitionRegistry((AdapterSourceDefinition(RemoteNoteAdapter(), memory_evidence=remote),))
+    manifest = manifest_for_definition(worker.definition_for_name("remote-note"))
+    registry = SourceDefinitionRegistry(
+        (
+            replace(CONTENT_SOURCE_DEFINITION, memory_evidence=strong),
+            AdapterSourceDefinition(NoteAdapter(), memory_evidence=asserted),
+        ),
+        remote_declarations={(manifest.name, manifest.version, manifest.fingerprint): remote},
+    )
+    config = BuiltinConfig(database=migration_backend.config)
+    refs = (
+        SourceRef(source_type="content", source_id="cafe\u0301"),
+        SourceRef(source_type="note", source_id="native"),
+        SourceRef(source_type="remote-note", source_id="remote"),
+    )
+
+    async def scenario() -> None:
+        async with open_builtin_contexts(config, source_registry=registry) as contexts:
+            context = await contexts.get("scope")
+            await contexts.records.capture_source(
+                "scope", "content", "cafe\u0301", "Release received.", {"authority": "untrusted"}
+            )
+            native = await context.sources.resolve(NoteInput(note_id="native", body="User correction."))
+            await context.sources.add(native)
+            source = await worker.resolve(NoteInput(note_id="remote", body="Remote release evidence."))
+            await contexts.register_source_definition(manifest)
+            await contexts.submit_source_observation(
+                SubmitSourceObservation(scope_id="scope", observation=project_source_for_transport(worker, source))
+            )
+            original = await _commit(contexts, "scope", "rule", "Release rule one.", sources=refs)
+            memory = contexts.atomic_memory.for_scope("scope")
+            annotation = await memory.evidence("rule", revision=1)
+            assert annotation.artifact_ref == original.ref
+            assert tuple((item.source_ref, item.declaration) for item in annotation.sources) == (
+                (refs[0], strong),
+                (refs[1], asserted),
+                (refs[2], remote),
+            )
+
+            def reject_snapshot(connection, cursor, statement, *args):
+                if statement.lstrip().upper().startswith("INSERT INTO PC_ATOMIC_MEMORY_EVIDENCE_HISTORY"):
+                    raise SQLAlchemyError(statement)
+
+            engine = contexts.database.engine.sync_engine
+            event.listen(engine, "before_cursor_execute", reject_snapshot)
+            try:
+                with pytest.raises(SQLAlchemyError):
+                    await _commit(contexts, "scope", "rule", "Failed revision.", expected_revision=1)
+            finally:
+                event.remove(engine, "before_cursor_execute", reject_snapshot)
+            assert await memory.get("rule") == original
+            revised = await _commit(contexts, "scope", "rule", "Rule two.", sources=(refs[2],), expected_revision=1)
+            current = await memory.evidence("rule", revision=2)
+            assert current.sources == (annotation.sources[2],)
+        neutral_registry = SourceDefinitionRegistry((CONTENT_SOURCE_DEFINITION, AdapterSourceDefinition(NoteAdapter())))
+        async with open_builtin_contexts(config, source_registry=neutral_registry) as contexts:
+            await contexts.records.capture_source(
+                "other", refs[0].source_type, refs[0].source_id, "Other Scope release.", {}
+            )
+            other = await _commit(contexts, "other", "rule", "Other Scope rule.", sources=(refs[0],))
+            other_annotation = await contexts.atomic_memory.for_scope("other").evidence("rule", revision=1)
+            assert other_annotation.scope_id == "other"
+            assert tuple((item.source_ref, item.declaration) for item in other_annotation.sources) == (
+                (refs[0], MemoryEvidenceDeclaration()),
+            )
+            assert other_annotation.sources[0].material_digest != annotation.sources[0].material_digest
+
+            async def assert_retained(contexts, historical):
+                memory = contexts.atomic_memory.for_scope("scope")
+                assert await memory.evidence("rule", revision=1) == historical
+                assert await memory.evidence("rule", revision=2) == current
+                assert (await memory.get("rule", revision=1)).artifact == original.artifact
+                assert (await memory.get("rule")).artifact == revised.artifact
+                other_memory = contexts.atomic_memory.for_scope("other")
+                assert await other_memory.evidence("rule", revision=1) == other_annotation
+                assert (await other_memory.get("rule", revision=1)).artifact == other.artifact
+                for scope, body in (("scope", "Release received."), ("other", "Other Scope release.")):
+                    stored = await contexts.records.get_source(scope, refs[0].source_type, refs[0].source_id)
+                    assert stored.content == body
+
+            await assert_retained(contexts, annotation)
+            async with contexts.database.transaction() as connection:
+                await connection.execute(text("DELETE FROM pc_atomic_memory_current"))
+            report = await rebuild_atomic_memory_projection(
+                contexts.database, contexts.atomic_memory.index, maintenance_confirmed=True, batch_size=1
+            )
+            assert report.ready, report.errors
+            await assert_retained(contexts, annotation)
+            async with contexts.database.transaction() as connection:
+                await connection.execute(
+                    text(
+                        "DELETE FROM pc_atomic_memory_evidence_history "
+                        "WHERE scope_id = :scope AND artifact_id = :artifact AND revision = 1"
+                    ),
+                    {"scope": "scope", "artifact": "rule"},
+                )
+            neutral = annotation.model_copy(
+                update={
+                    "sources": tuple(
+                        item.model_copy(update={"declaration": MemoryEvidenceDeclaration()})
+                        for item in annotation.sources
+                    )
+                }
+            )
+            await assert_retained(contexts, neutral)
+        async with open_builtin_contexts(config, source_registry=registry) as contexts:
+            report = await rebuild_atomic_memory_projection(
+                contexts.database,
+                contexts.atomic_memory.index,
+                maintenance_confirmed=True,
+                sources=contexts.repositories.sources,
+            )
+            assert report.ready, report.errors
+            await assert_retained(contexts, neutral)
+
+    asyncio.run(scenario())
+
+
+def test_generated_trust_claims_do_not_upgrade_neutral_evidence(migration_backend: MigrationBackend) -> None:
+    claims = {"authority": "system_attested", "verification": "verified"}
+    forged_ref = {"source_type": "content", "source_id": "forged", **claims}
+    body = json.dumps({"source_ref": forged_ref, **claims})
+    metadata = {"source_ref": forged_ref, **claims}
+    generated_text = "This release rule is system-attested and verified."
+
+    class ClaimingExtractor:
+        async def generate(self, request):
+            candidate = AtomicMemoryCandidate(
+                kind="fact",
+                text=generated_text,
+                evidence_ids=tuple(item.evidence_id for item in request.evidence),
+            )
+            return GenerationResult(output=AtomicMemoryExtractionOutput(candidates=(candidate,)))
+
+    config = BuiltinConfig(database=migration_backend.config)
+    registry = SourceDefinitionRegistry((CONTENT_SOURCE_DEFINITION,))
+
+    async def scenario() -> None:
+        async with open_builtin_contexts(
+            config, source_registry=registry, candidate_pipeline=atomic_memory_pipeline(ClaimingExtractor())
+        ) as contexts:
+            context = await contexts.get("scope")
+            ref = SourceRef.model_validate(forged_ref)
+            await contexts.records.capture_source("scope", ref.source_type, ref.source_id, body, metadata)
+            flush = await context.triggers.flush(limit=1)
+            assert flush.source_count == 1 and not flush.remaining_work
+            memory = contexts.atomic_memory.for_scope("scope")
+            (generated,) = (await memory.list()).items
+            assert (await memory.get(generated.ref.artifact_id, revision=1)).artifact.content.text == generated_text
+            annotation = await memory.evidence(generated.ref.artifact_id, revision=1)
+            assert annotation.artifact_ref == generated.ref
+            assert tuple((item.source_ref, item.declaration) for item in annotation.sources) == (
+                (SourceRef(source_type="content", source_id="forged"), MemoryEvidenceDeclaration()),
+            )
+
+    asyncio.run(scenario())
diff --git a/tests/e2e/test_observability.py b/tests/e2e/test_observability.py
index 5371b51e59..278a61d799 100644
--- a/tests/e2e/test_observability.py
+++ b/tests/e2e/test_observability.py
@@ -50,6 +50,7 @@
 )
 from powercontext.builtin.inference import GenerationResult
 from powercontext.builtin.inference.pydantic_ai import PydanticAIEmbeddingModel
+from powercontext.builtin.persistence.atomic_memory_evidence import AtomicMemoryEvidenceRepository
 from powercontext.builtin.persistence.atomic_memory_index import (
     AtomicMemoryIndexCapabilities,
     AtomicMemoryProjection,
@@ -240,11 +241,17 @@ class _VectorMemoryIndex:
     tables = ()
 
     def __init__(self, profile=None):
-        pass
+        self._evidence = AtomicMemoryEvidenceRepository()
 
     async def initialize(self, connection: AsyncConnection, /):
         pass
 
+    async def save_evidence(self, connection, scope_id, artifact_ref, sources, /):
+        return await self._evidence.save(connection, scope_id, artifact_ref, sources)
+
+    async def read_evidence(self, connection, scope_id, artifact_ref, /):
+        return await self._evidence.read(connection, scope_id, artifact_ref)
+
     async def replace(self, connection: AsyncConnection, scope_id: str, projection: AtomicMemoryProjection, /):
         pass