From 8439875a32a7de9b54b5a5dc4cfdc0fb85f98a33 Mon Sep 17 00:00:00 2001 From: Mao Meng <1707647656@qq.com> Date: Wed, 14 Oct 2026 14:35:27 +0800 Subject: [PATCH 1/8] feat(memory): persist versioned source evidence Implement issue01 of the Source evidence declarations specification. Bind trusted native and remote declarations to exact Source material and retain per-revision Atomic Memory snapshots in the existing transaction. Preserve neutral fallback, exact identities, scope authorization and public retrieval contracts. Add versioned p0004 migrations and SQLite persistence regressions. Refs #1975. Full suite: 5255 passed, 431 skipped, 7 failed; introduced observability adapter failures fixed and its file passes 18 tests. Topic Memory vector failure also reproduces at the unchanged baseline. --- .../powercontext_connector_opendal/source.py | 2 + src/powercontext/__init__.py | 4 + .../artifacts/atomic_memory/service.py | 8 + .../persistence/atomic_memory_evidence.py | 161 ++++++++++++++++++ .../atomic_memory_evidence_schema.py | 59 +++++++ .../persistence/atomic_memory_index.py | 52 +++++- .../migrations/atomic_memory_evidence.py | 100 +++++++++++ .../versions/p0004_atomic_memory_evidence.py | 25 +++ .../persistence/migrations/mysql_schema.py | 6 +- .../migrations/resources/manifest.json | 17 +- .../schemas/atomic_memory_evidence_v1.json | 14 ++ .../versions/p0004_atomic_memory_evidence.py | 30 ++++ .../builtin/persistence/schema.py | 10 +- .../builtin/persistence/source_definitions.py | 35 +++- .../builtin/persistence/sources.py | 158 ++++++++++++++++- .../builtin/runtime/atomic_memory.py | 55 ++++++ .../builtin/runtime/composition.py | 14 +- src/powercontext/sources/__init__.py | 3 + src/powercontext/sources/definitions.py | 63 ++++++- src/powercontext/sources/evidence.py | 60 +++++++ 20 files changed, 860 insertions(+), 16 deletions(-) create mode 100644 src/powercontext/builtin/persistence/atomic_memory_evidence.py create mode 100644 src/powercontext/builtin/persistence/atomic_memory_evidence_schema.py create mode 100644 src/powercontext/builtin/persistence/migrations/atomic_memory_evidence.py create mode 100644 src/powercontext/builtin/persistence/migrations/mysql_resources/versions/p0004_atomic_memory_evidence.py create mode 100644 src/powercontext/builtin/persistence/migrations/resources/schemas/atomic_memory_evidence_v1.json create mode 100644 src/powercontext/builtin/persistence/migrations/resources/versions/p0004_atomic_memory_evidence.py create mode 100644 src/powercontext/sources/evidence.py 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..cb054ff704 --- /dev/null +++ b/src/powercontext/builtin/persistence/atomic_memory_evidence.py @@ -0,0 +1,161 @@ +# 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 _require_identity +from powercontext.limits import MAX_SCOPE_ID_LENGTH +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) + + @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..b4b446a20a --- /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 = "p0004" +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/migrations/atomic_memory_evidence.py b/src/powercontext/builtin/persistence/migrations/atomic_memory_evidence.py new file mode 100644 index 0000000000..cafc2cdfe9 --- /dev/null +++ b/src/powercontext/builtin/persistence/migrations/atomic_memory_evidence.py @@ -0,0 +1,100 @@ +# 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 p0004 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 = { + "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 p0004 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] + for table, statement in zip(ATOMIC_MEMORY_EVIDENCE_TABLES, statements[:-1], strict=True): + exists = await connection.run_sync(lambda sync, name=table.name: inspect(sync).has_table(name)) + if not exists: + await connection.exec_driver_sql(statement) + 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 not versions: + await connection.exec_driver_sql(statements[-1]) + elif 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.mysql_schema import _matches, _parse_table + + actual = (await connection.exec_driver_sql(f"SHOW CREATE TABLE `{name}`")).one()[1] + actual_name, actual_table = _parse_table(actual) + expected_name, expected_table = _parse_table(statement) + 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/p0004_atomic_memory_evidence.py b/src/powercontext/builtin/persistence/migrations/mysql_resources/versions/p0004_atomic_memory_evidence.py new file mode 100644 index 0000000000..e83df8afbf --- /dev/null +++ b/src/powercontext/builtin/persistence/migrations/mysql_resources/versions/p0004_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 = "p0004" +down_revision = "p0003" + + +def upgrade() -> None: + context.config.attributes["execute_revision"](revision) + op.execute("INSERT INTO pc_atomic_memory_evidence_schema (schema_version) VALUES ('p0004')") diff --git a/src/powercontext/builtin/persistence/migrations/mysql_schema.py b/src/powercontext/builtin/persistence/migrations/mysql_schema.py index b3502db1bc..01cf62dc69 100644 --- a/src/powercontext/builtin/persistence/migrations/mysql_schema.py +++ b/src/powercontext/builtin/persistence/migrations/mysql_schema.py @@ -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") +_REVISIONS = ("p0001", "p0002", "p0003", "p0004") _CITATION_TABLES = ("pc_artifacts", "pc_artifact_candidate_versions") _FAMILY_CHECK = "ck_pc_artifact_tags_family" _CONTROL_TABLE = "pc_schema_revision" @@ -141,10 +141,14 @@ 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": 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 65c798ea8b..73dbd2a358 100644 --- a/src/powercontext/builtin/persistence/migrations/resources/manifest.json +++ b/src/powercontext/builtin/persistence/migrations/resources/manifest.json @@ -4,11 +4,15 @@ "schemas/pre_dream.json" ], "p0002": [], - "p0003": [] + "p0003": [], + "p0004": [ + "schemas/atomic_memory_evidence_v1.json" + ] }, "sqlite_baselines": { "e1cae089a8a565291a7a1f782cea1f6c6e1a6cdd9eab75ef0c4ca61dfda453cb": "p0001", - "fcee685ed307ee6d02fe5e950ed4e80be75f38d19de902ccacf443f21ee8c73d": "p0003" + "fcee685ed307ee6d02fe5e950ed4e80be75f38d19de902ccacf443f21ee8c73d": "p0003", + "93c1c20ae9bba2114158658e7852324af2bbaef12d28fa298cdbe3cae4847a84": "p0004" }, "sqlite_fingerprints": { "p0001": [ @@ -20,13 +24,20 @@ "p0003": [ "9958fa0413ed3c16442df27dc48057d438e2cf3f7e4312c5a914cccd9d3acaab", "fcee685ed307ee6d02fe5e950ed4e80be75f38d19de902ccacf443f21ee8c73d" + ], + "p0004": [ + "4f21a735d9ff2e53b70599f658d762ee30a14604e4fe96c97826215433dcd509", + "93c1c20ae9bba2114158658e7852324af2bbaef12d28fa298cdbe3cae4847a84" ] }, "affected_objects": [ "pc_artifacts", "pc_artifact_heads", "pc_artifact_candidate_versions", - "pc_artifact_tags" + "pc_artifact_tags", + "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..fdd79d655e --- /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 ('p0004')" + ], + "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 ('p0004')" + ] +} diff --git a/src/powercontext/builtin/persistence/migrations/resources/versions/p0004_atomic_memory_evidence.py b/src/powercontext/builtin/persistence/migrations/resources/versions/p0004_atomic_memory_evidence.py new file mode 100644 index 0000000000..32648cfd55 --- /dev/null +++ b/src/powercontext/builtin/persistence/migrations/resources/versions/p0004_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 = "p0004" +down_revision = "p0003" + + +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..5b38249f1f 100644 --- a/src/powercontext/builtin/persistence/sources.py +++ b/src/powercontext/builtin/persistence/sources.py @@ -18,6 +18,7 @@ from collections.abc import AsyncGenerator, Iterable, Mapping, Sequence from contextlib import suppress +from hashlib import sha256 from typing import Any, Literal, cast import rfc8785 @@ -29,6 +30,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 +41,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.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 +177,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 +246,123 @@ 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. + """ + + stored_sources = await self.get_many(connection, scope_id, refs) + # These server-owned records describe a management write, not evidence + # supplied for that write. Other lineage-only Sources remain represented. + sources = tuple(stored for stored in stored_sources if not _management_write_source(stored.value)) + if not sources: + 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((stored.ref.source_type, stored.ref.source_id) for stored in sources) + ), + ) + ) + ) + .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 stored in sources: + source = stored.value + digest = _source_material_digest(source) + value = SourceEvidence( + source_ref=stored.ref, + materialization=source.materialization, + material_digest=digest, + definition_name=stored.ref.source_type, + definition_version=source.definition_version, + ) + binding = bindings.get((stored.ref.source_type, stored.ref.source_id)) + if binding is None or ( + binding["material_digest"] != 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 + identity = (value.definition_name, value.definition_version) + if identity not in manifests: + manifests[identity] = await _find_evidence_manifest(connection, *identity) + manifest = manifests[identity] + if ( + manifest is None + or manifest.fingerprint != binding["manifest_fingerprint"] + or (isinstance(source, SourceObservation) and source.definition_fingerprint != 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 +572,32 @@ 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 _dump_source( source: Source, *, diff --git a/src/powercontext/builtin/runtime/atomic_memory.py b/src/powercontext/builtin/runtime/atomic_memory.py index f79cb78fcb..f2a2a14ed9 100644 --- a/src/powercontext/builtin/runtime/atomic_memory.py +++ b/src/powercontext/builtin/runtime/atomic_memory.py @@ -58,6 +58,7 @@ ) 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 from powercontext.builtin.persistence.atomic_memory_index import ( ATOMIC_MEMORY_RRF_CONSTANT, AtomicMemoryIndex, @@ -79,6 +80,7 @@ from powercontext.builtin.records import BaseOperationNotSupportedError, InvalidBaseAccessRequestError from powercontext.builtin.runtime.artifact_merge import ArtifactMergeAccess, merge_artifact_tags from powercontext.builtin.tags import TagFilter +from powercontext.sources.evidence import MemoryEvidenceDeclaration class AtomicMemoryAccess(ArtifactMergeAccess): @@ -186,6 +188,7 @@ def __init__( self.publisher = AtomicMemoryProjectionPublisher( index, embedding_model=embedding_model, + sources=artifacts.source_repository, ) self.service = AtomicMemoryService( artifacts=artifacts, @@ -222,6 +225,58 @@ 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) + sources = application.artifacts.source_repository + materials = ( + () + if sources is None + else await sources.evidence(connection, self.scope_id, record.artifact.lineage.sources) + ) + annotation = await application.index.read_evidence(connection, self.scope_id, record.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 AtomicMemoryEvidenceAnnotation( + scope_id=self.scope_id, + artifact_ref=record.ref, + sources=tuple( + material.model_copy(update={"declaration": MemoryEvidenceDeclaration()}) + for material in 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}) + async def list( self, *, diff --git a/src/powercontext/builtin/runtime/composition.py b/src/powercontext/builtin/runtime/composition.py index b8015b896c..cd2e4b9e40 100644 --- a/src/powercontext/builtin/runtime/composition.py +++ b/src/powercontext/builtin/runtime/composition.py @@ -196,7 +196,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 @@ -796,7 +796,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"] From 5dfa2ad62bc8ba1515de43c3181a5c1e3013e54f Mon Sep 17 00:00:00 2001 From: Mao Meng <1707647656@qq.com> Date: Wed, 14 Oct 2026 15:41:42 +0800 Subject: [PATCH 2/8] feat(memory): verify historical evidence during projection rebuild --- .../persistence/atomic_memory_evidence.py | 52 ++++- .../builtin/persistence/sources.py | 196 ++++++++++++++++-- .../builtin/runtime/atomic_memory.py | 50 +---- .../builtin/runtime/atomic_memory_rebuild.py | 58 +++++- 4 files changed, 291 insertions(+), 65 deletions(-) diff --git a/src/powercontext/builtin/persistence/atomic_memory_evidence.py b/src/powercontext/builtin/persistence/atomic_memory_evidence.py index cb054ff704..99d5c38311 100644 --- a/src/powercontext/builtin/persistence/atomic_memory_evidence.py +++ b/src/powercontext/builtin/persistence/atomic_memory_evidence.py @@ -29,8 +29,9 @@ 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 _require_identity +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 @@ -104,6 +105,55 @@ async def read( 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) diff --git a/src/powercontext/builtin/persistence/sources.py b/src/powercontext/builtin/persistence/sources.py index 5b38249f1f..597c89a842 100644 --- a/src/powercontext/builtin/persistence/sources.py +++ b/src/powercontext/builtin/persistence/sources.py @@ -22,7 +22,12 @@ 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 @@ -43,7 +48,7 @@ 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 @@ -267,6 +272,167 @@ async def evidence( sources = tuple(stored for stored in stored_sources if not _management_write_source(stored.value)) if not sources: return () + materials = tuple( + SourceEvidence( + source_ref=stored.ref, + materialization=stored.value.materialization, + material_digest=_source_material_digest(stored.value), + definition_name=stored.ref.source_type, + definition_version=stored.value.definition_version, + ) + for stored in sources + ) + fingerprints = { + (stored.ref.source_type, stored.ref.source_id): stored.value.definition_fingerprint + for stored in sources + if isinstance(stored.value, SourceObservation) + } + return await self._bound_evidence(connection, scope_id, materials, fingerprints) + + 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=f"sha256:{sha256(rfc8785.dumps(material)).hexdigest()}", + 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 + return source, source.model_dump(mode="json", by_alias=True) + 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 = ( ( @@ -274,7 +440,7 @@ async def evidence( select(table).where( table.c.scope_id == scope_id, tuple_(table.c.source_type, table.c.source_id).in_( - tuple((stored.ref.source_type, stored.ref.source_id) for stored in sources) + tuple((value.source_ref.source_type, value.source_ref.source_id) for value in materials) ), ) ) @@ -285,33 +451,25 @@ async def evidence( bindings = {(row["source_type"], row["source_id"]): row for row in rows} manifests: dict[tuple[str, str], SourceDefinitionManifest | None] = {} evidence: list[SourceEvidence] = [] - for stored in sources: - source = stored.value - digest = _source_material_digest(source) - value = SourceEvidence( - source_ref=stored.ref, - materialization=source.materialization, - material_digest=digest, - definition_name=stored.ref.source_type, - definition_version=source.definition_version, - ) - binding = bindings.get((stored.ref.source_type, stored.ref.source_id)) + 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"] != digest + 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 - identity = (value.definition_name, value.definition_version) - if identity not in manifests: - manifests[identity] = await _find_evidence_manifest(connection, *identity) - manifest = manifests[identity] + 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 (isinstance(source, SourceObservation) and source.definition_fingerprint != manifest.fingerprint) + or (identity in fingerprints and fingerprints[identity] != manifest.fingerprint) ): evidence.append(value) continue diff --git a/src/powercontext/builtin/runtime/atomic_memory.py b/src/powercontext/builtin/runtime/atomic_memory.py index f2a2a14ed9..70688c79e7 100644 --- a/src/powercontext/builtin/runtime/atomic_memory.py +++ b/src/powercontext/builtin/runtime/atomic_memory.py @@ -58,7 +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 +from powercontext.builtin.persistence.atomic_memory_evidence import ( + AtomicMemoryEvidenceAnnotation, + AtomicMemoryEvidenceRepository, +) from powercontext.builtin.persistence.atomic_memory_index import ( ATOMIC_MEMORY_RRF_CONSTANT, AtomicMemoryIndex, @@ -80,7 +83,6 @@ from powercontext.builtin.records import BaseOperationNotSupportedError, InvalidBaseAccessRequestError from powercontext.builtin.runtime.artifact_merge import ArtifactMergeAccess, merge_artifact_tags from powercontext.builtin.tags import TagFilter -from powercontext.sources.evidence import MemoryEvidenceDeclaration class AtomicMemoryAccess(ArtifactMergeAccess): @@ -237,45 +239,13 @@ async def evidence(self, artifact_id: str, *, revision: int, context=None) -> At application.database.transaction(consistent_snapshot=True) as connection, ): record = await application.service.get(connection, self.scope_id, artifact_id, context, revision=revision) - sources = application.artifacts.source_repository - materials = ( - () - if sources is None - else await sources.evidence(connection, self.scope_id, record.artifact.lineage.sources) - ) - annotation = await application.index.read_evidence(connection, self.scope_id, record.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 AtomicMemoryEvidenceAnnotation( - scope_id=self.scope_id, - artifact_ref=record.ref, - sources=tuple( - material.model_copy(update={"declaration": MemoryEvidenceDeclaration()}) - for material in 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 await AtomicMemoryEvidenceRepository().verify( + connection, + self.scope_id, + record.ref, + record.artifact.lineage.sources, + sources=application.artifacts.source_repository, ) - return annotation.model_copy(update={"sources": verified}) 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": { From 034e0cc5a8d661f928485fb243000763c612bb0c Mon Sep 17 00:00:00 2001 From: Mao Meng <1707647656@qq.com> Date: Wed, 14 Oct 2026 16:34:18 +0800 Subject: [PATCH 3/8] fix(memory): enforce evidence schema safety and backend conformance --- .../persistence/atomic_memory_readiness.py | 10 +++- .../migrations/atomic_memory_evidence.py | 46 +++++++++++++++---- 2 files changed, 47 insertions(+), 9 deletions(-) diff --git a/src/powercontext/builtin/persistence/atomic_memory_readiness.py b/src/powercontext/builtin/persistence/atomic_memory_readiness.py index fdd46b5b74..455af19b99 100644 --- a/src/powercontext/builtin/persistence/atomic_memory_readiness.py +++ b/src/powercontext/builtin/persistence/atomic_memory_readiness.py @@ -26,6 +26,10 @@ from sqlalchemy.ext.asyncio import AsyncConnection 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 +94,15 @@ 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] = [] + 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 index cafc2cdfe9..72ef0935db 100644 --- a/src/powercontext/builtin/persistence/migrations/atomic_memory_evidence.py +++ b/src/powercontext/builtin/persistence/migrations/atomic_memory_evidence.py @@ -43,6 +43,7 @@ class AtomicMemoryEvidenceSchemaError(PersistenceError): def __init__(self, detail: str) -> None: message = { + "missing": "Atomic Memory evidence schema is incomplete; apply the registered p0004 migration", "version": "Atomic Memory evidence schema version is unsupported", "shape": "Atomic Memory evidence table has an incompatible shape", }[detail] @@ -63,16 +64,41 @@ async def apply_atomic_memory_evidence_migration(connection: AsyncConnection) -> 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): - exists = await connection.run_sync(lambda sync, name=table.name: inspect(sync).has_table(name)) - if not exists: - await connection.exec_driver_sql(statement) - await _require_shape(connection, table.name, statement, dialect) + 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()) + 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]) - elif versions != (ATOMIC_MEMORY_EVIDENCE_SCHEMA_VERSION,): + + +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") @@ -88,11 +114,15 @@ async def _require_shape(connection: AsyncConnection, name: str, statement: str, 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] - actual_name, actual_table = _parse_table(actual) - expected_name, expected_table = _parse_table(statement) + 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": []}, From 4d1bf17abba42774d189d28eb46d768100ffa497 Mon Sep 17 00:00:00 2001 From: Mao Meng <1707647656@qq.com> Date: Wed, 14 Oct 2026 16:38:37 +0800 Subject: [PATCH 4/8] fix(memory): preserve unrelated database inspection --- .../builtin/persistence/atomic_memory_readiness.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/powercontext/builtin/persistence/atomic_memory_readiness.py b/src/powercontext/builtin/persistence/atomic_memory_readiness.py index 455af19b99..52df67b4a5 100644 --- a/src/powercontext/builtin/persistence/atomic_memory_readiness.py +++ b/src/powercontext/builtin/persistence/atomic_memory_readiness.py @@ -25,6 +25,7 @@ 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, @@ -99,10 +100,11 @@ async def atomic_memory_readiness_issues( if tables is None: tables = set(await connection.run_sync(lambda value: inspect(value).get_table_names())) issues: list[str] = [] - try: - await require_atomic_memory_evidence_schema(connection, tables) - except AtomicMemoryEvidenceSchemaError as error: - issues.append(str(error)) + 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 From cf8068a0f883e73ca78f2b7e4d5d794612b74584 Mon Sep 17 00:00:00 2001 From: Mao Meng <1707647656@qq.com> Date: Thu, 15 Oct 2026 01:02:50 +0800 Subject: [PATCH 5/8] test(memory): consolidate source evidence E2E acceptance Cover persisted declarations, scope isolation, historical neutrality, synthetic model trust claims, and unchanged recall consumers on SQLite and OceanBase. --- tests/e2e/conftest.py | 13 + ...t_atomic_memory_evidence_neutral_recall.py | 131 ++++++++++ .../test_atomic_memory_projection_rebuild.py | 17 +- .../e2e/test_atomic_memory_source_evidence.py | 228 ++++++++++++++++++ tests/e2e/test_observability.py | 9 +- 5 files changed, 396 insertions(+), 2 deletions(-) create mode 100644 tests/e2e/test_atomic_memory_evidence_neutral_recall.py create mode 100644 tests/e2e/test_atomic_memory_source_evidence.py 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 From 0d70cbaf2798a3cb1f453527f67670a0bd86d002 Mon Sep 17 00:00:00 2001 From: Mao Meng <1707647656@qq.com> Date: Thu, 15 Oct 2026 18:20:15 +0800 Subject: [PATCH 6/8] fix(memory): verify evidence against retained source material --- .../builtin/persistence/sources.py | 35 ++++++------------- src/powercontext/sources/observations.py | 2 +- 2 files changed, 12 insertions(+), 25 deletions(-) diff --git a/src/powercontext/builtin/persistence/sources.py b/src/powercontext/builtin/persistence/sources.py index 597c89a842..0b58316aa7 100644 --- a/src/powercontext/builtin/persistence/sources.py +++ b/src/powercontext/builtin/persistence/sources.py @@ -266,28 +266,7 @@ async def evidence( value is substituted for historical registration facts. """ - stored_sources = await self.get_many(connection, scope_id, refs) - # These server-owned records describe a management write, not evidence - # supplied for that write. Other lineage-only Sources remain represented. - sources = tuple(stored for stored in stored_sources if not _management_write_source(stored.value)) - if not sources: - return () - materials = tuple( - SourceEvidence( - source_ref=stored.ref, - materialization=stored.value.materialization, - material_digest=_source_material_digest(stored.value), - definition_name=stored.ref.source_type, - definition_version=stored.value.definition_version, - ) - for stored in sources - ) - fingerprints = { - (stored.ref.source_type, stored.ref.source_id): stored.value.definition_fingerprint - for stored in sources - if isinstance(stored.value, SourceObservation) - } - return await self._bound_evidence(connection, scope_id, materials, fingerprints) + return await self.historical_evidence(connection, scope_id, refs) async def historical_evidence( self, connection: AsyncConnection, scope_id: str, refs: Sequence[SourceRef], / @@ -341,7 +320,7 @@ async def historical_evidence( SourceEvidence( source_ref=reference, materialization=source.materialization, - material_digest=f"sha256:{sha256(rfc8785.dumps(material)).hexdigest()}", + material_digest=_source_material_digest_material(material), definition_name=ref[0], definition_version=source.definition_version, ) @@ -372,7 +351,11 @@ async def _historical_material( raise InvalidStoredPayloadError("source", reference.source_type, "payload is not JSON") from error return source, material source = self._decode_row(row).value - return source, source.model_dump(mode="json", by_alias=True) + 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) @@ -756,6 +739,10 @@ async def _find_evidence_manifest( 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/sources/observations.py b/src/powercontext/sources/observations.py index 576e29fb79..a7b1858c03 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), From 3305b6bb1edb9b1c0b86b3ec4955239dad9d36d8 Mon Sep 17 00:00:00 2001 From: Mao Meng <1707647656@qq.com> Date: Thu, 15 Oct 2026 21:17:41 +0800 Subject: [PATCH 7/8] fix(sources): align remote payload aliases with manifests --- src/powercontext/sources/observations.py | 2 +- tests/builtin/persistence/test_sources.py | 53 +++++++++++++++++++++++ 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/src/powercontext/sources/observations.py b/src/powercontext/sources/observations.py index a7b1858c03..c9bb8a802f 100644 --- a/src/powercontext/sources/observations.py +++ b/src/powercontext/sources/observations.py @@ -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_sources.py b/tests/builtin/persistence/test_sources.py index 22e080499f..1e88d9bb96 100644 --- a/tests/builtin/persistence/test_sources.py +++ b/tests/builtin/persistence/test_sources.py @@ -17,6 +17,7 @@ import asyncio import pytest +from pydantic import BaseModel, Field from sqlalchemy import insert from powercontext import InvalidSourceDefinitionError @@ -29,8 +30,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 +50,31 @@ ) +class SerializationAliasInput(BaseModel): + source_id: str + body: str + + +class SerializationAliasSource(Source): + body: str = Field(serialization_alias="wire_text") + + +class SerializationAliasAdapter: + name = "serialization-alias" + input_class = SerializationAliasInput + source_class = SerializationAliasSource + + async def resolve(self, value: SerializationAliasInput, /) -> SerializationAliasSource: + return SerializationAliasSource( + name=value.source_id, + materialization=SourceMaterialization.CAPTURED, + 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 +178,30 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_remote_observation_ingestion_accepts_serialization_aliases() -> None: + async def scenario() -> None: + worker = SourceDefinitionRegistry((AdapterSourceDefinition(SerializationAliasAdapter()),)) + 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["wire_text"] == "Remote note text." + assert "body" not in stored.payload + + 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): From c9ee82d5dc9cf75f2e7c8800959a611c507a7d03 Mon Sep 17 00:00:00 2001 From: Mao Meng <1707647656@qq.com> Date: Fri, 16 Oct 2026 17:16:32 +0800 Subject: [PATCH 8/8] fix(sources): preserve envelope keys with alias generators --- src/powercontext/sources/models.py | 11 ++++--- tests/builtin/persistence/test_sources.py | 37 ++++++++++++++++++----- 2 files changed, 36 insertions(+), 12 deletions(-) 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/tests/builtin/persistence/test_sources.py b/tests/builtin/persistence/test_sources.py index 1e88d9bb96..b74a798a7e 100644 --- a/tests/builtin/persistence/test_sources.py +++ b/tests/builtin/persistence/test_sources.py @@ -17,7 +17,8 @@ import asyncio import pytest -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field +from pydantic.alias_generators import to_camel from sqlalchemy import insert from powercontext import InvalidSourceDefinitionError @@ -59,15 +60,27 @@ 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 - source_class = SerializationAliasSource + + def __init__(self, source_class: type[SerializationAliasSource]) -> None: + self.source_class = source_class async def resolve(self, value: SerializationAliasInput, /) -> SerializationAliasSource: - return SerializationAliasSource( + return self.source_class( name=value.source_id, + definition_version="2", materialization=SourceMaterialization.CAPTURED, + description="Remote alias description.", body=value.body, ) @@ -178,9 +191,14 @@ async def scenario() -> None: asyncio.run(scenario()) -def test_remote_observation_ingestion_accepts_serialization_aliases() -> None: +@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()),)) + 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) @@ -196,8 +214,13 @@ async def scenario() -> None: assert receipt.source_ref == SourceRef(source_type="serialization-alias", source_id="remote-1") assert isinstance(stored, SourceObservation) - assert stored.payload["wire_text"] == "Remote note text." - assert "body" not in stored.payload + assert stored.payload == { + "name": "remote-1", + "definition_version": "2", + "materialization": "captured", + "description": "Remote alias description.", + "wire_text": "Remote note text.", + } asyncio.run(scenario())