diff --git a/src/powercontext/builtin/artifacts/handoff/service.py b/src/powercontext/builtin/artifacts/handoff/service.py index 1fc78e9f1b..3fbf90457a 100644 --- a/src/powercontext/builtin/artifacts/handoff/service.py +++ b/src/powercontext/builtin/artifacts/handoff/service.py @@ -16,7 +16,8 @@ from __future__ import annotations -from collections.abc import Awaitable, Callable, Iterable +from collections.abc import Awaitable, Callable, Iterable, Sequence +from typing import Protocol, runtime_checkable from powercontext.artifacts import ArtifactRef from powercontext.builtin.artifacts.handoff.errors import ( @@ -37,6 +38,7 @@ HandoffContent, HandoffDraft, HandoffEvidenceCheck, + HandoffGenerationEvidence, HandoffGenerationRequest, HandoffMemoryCitation, HandoffResolution, @@ -58,6 +60,15 @@ HandoffEvidenceAuthorizer = Callable[[HandoffCitation], Awaitable[bool]] +@runtime_checkable +class _BatchHandoffEvidenceResolver(Protocol): + async def resolve_many( + self, + citations: Sequence[HandoffCitation], + /, + ) -> tuple[HandoffGenerationEvidence, ...]: ... + + class HandoffService: """Generate, finalize, commit, inspect, and resolve one scope's Handoff lifecycle.""" @@ -89,7 +100,10 @@ async def prepare(self, action: PrepareHandoff, /) -> HandoffDraft: if self._generation_pipeline is None: raise HandoffGenerationUnavailableError - evidence = tuple([await self._evidence_resolver.resolve(citation) for citation in action.evidence]) + if isinstance(self._evidence_resolver, _BatchHandoffEvidenceResolver): + evidence = await self._evidence_resolver.resolve_many(action.evidence) + else: + evidence = tuple([await self._evidence_resolver.resolve(citation) for citation in action.evidence]) draft = await self._generation_pipeline.generate( HandoffGenerationRequest( objective=action.objective, diff --git a/src/powercontext/builtin/persistence/artifacts.py b/src/powercontext/builtin/persistence/artifacts.py index 1cc24fdcbd..ecafb2bed7 100644 --- a/src/powercontext/builtin/persistence/artifacts.py +++ b/src/powercontext/builtin/persistence/artifacts.py @@ -240,8 +240,11 @@ async def get_many( if not requested: return () keys = tuple(dict.fromkeys((ref.family, ref.artifact_id, ref.revision) for ref in requested)) - for family, _, _ in keys: - self._artifact_type(family) + for ref in requested: + try: + self._artifact_type(ref.family) + except RepositoryNotFoundError: + raise RepositoryNotFoundError("artifact", (scope_id, ref)) from None identity = tuple_( ARTIFACTS_TABLE.c.family, ARTIFACTS_TABLE.c.artifact_id, diff --git a/src/powercontext/builtin/persistence/family_management.py b/src/powercontext/builtin/persistence/family_management.py index 4185f095b4..e1283b8437 100644 --- a/src/powercontext/builtin/persistence/family_management.py +++ b/src/powercontext/builtin/persistence/family_management.py @@ -47,6 +47,7 @@ from powercontext.builtin.persistence.database import AsyncDatabase from powercontext.builtin.persistence.errors import RepositoryNotFoundError from powercontext.builtin.persistence.experience_index import ExperienceIndex +from powercontext.builtin.persistence.generation_sources import GenerationSourceAccess from powercontext.builtin.persistence.handoff import RelationalHandoffBackend, RelationalHandoffEvidenceResolver from powercontext.builtin.persistence.memory import RelationalMemoryBackend from powercontext.builtin.persistence.memory_index import MemoryIndex @@ -512,7 +513,7 @@ def _service(self, scope_id: str, connection: AsyncConnection) -> HandoffService evidence_resolver=RelationalHandoffEvidenceResolver( database=self._database, scope_id=scope_id, - sources=self._sources, + sources=GenerationSourceAccess(self._sources), artifacts=self._artifacts, memory=memory, connection=connection, diff --git a/src/powercontext/builtin/persistence/generation_sources.py b/src/powercontext/builtin/persistence/generation_sources.py new file mode 100644 index 0000000000..11548a56a5 --- /dev/null +++ b/src/powercontext/builtin/persistence/generation_sources.py @@ -0,0 +1,63 @@ +# 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. + +"""Source reads whose purpose is Artifact generation.""" + +from __future__ import annotations + +from collections.abc import Sequence + +from sqlalchemy.ext.asyncio import AsyncConnection + +from powercontext.builtin.persistence.sources import SourceRepository, StoredSource +from powercontext.builtin.source_eligibility import is_generation_eligible, require_source_eligible +from powercontext.sources import SourceRef + + +class GenerationSourceAccess: + """Keep generation reads and Source eligibility inseparable.""" + + def __init__(self, repository: SourceRepository, /) -> None: + self._repository = repository + + async def require_for_generation( + self, + connection: AsyncConnection, + scope_id: str, + refs: Sequence[SourceRef], + /, + ) -> tuple[StoredSource, ...]: + """Resolve explicit references, rejecting the whole request if any is ineligible.""" + + rows = await self._repository.get_many(connection, scope_id, refs) + for row in rows: + require_source_eligible(row.ref, row.value) + return rows + + async def list_window_for_generation( + self, + connection: AsyncConnection, + scope_id: str, + /, + *, + after: int, + through: int, + ) -> tuple[StoredSource, ...]: + """Resolve one bounded journal window and omit valid lineage-only Sources.""" + + rows = await self._repository.list_window(connection, scope_id, after=after, through=through) + return tuple(row for row in rows if is_generation_eligible(row.value)) + + +__all__ = ["GenerationSourceAccess"] diff --git a/src/powercontext/builtin/persistence/handoff.py b/src/powercontext/builtin/persistence/handoff.py index 488f2f3b99..f6bfe61026 100644 --- a/src/powercontext/builtin/persistence/handoff.py +++ b/src/powercontext/builtin/persistence/handoff.py @@ -16,6 +16,7 @@ from __future__ import annotations +from collections.abc import Sequence from typing import cast from sqlalchemy.ext.asyncio import AsyncConnection @@ -42,8 +43,7 @@ from powercontext.builtin.persistence.artifacts import ArtifactRepository from powercontext.builtin.persistence.database import AsyncDatabase from powercontext.builtin.persistence.errors import RepositoryNotFoundError -from powercontext.builtin.persistence.sources import SourceRepository -from powercontext.builtin.source_eligibility import require_source_eligible +from powercontext.builtin.persistence.generation_sources import GenerationSourceAccess from powercontext.errors import ArtifactNotFoundError @@ -123,7 +123,7 @@ def __init__( *, database: AsyncDatabase, scope_id: str, - sources: SourceRepository, + sources: GenerationSourceAccess, artifacts: ArtifactRepository, memory: MemoryService, connection: AsyncConnection | None = None, @@ -136,32 +136,76 @@ def __init__( self._bound_connection = connection async def resolve(self, citation: HandoffCitation, /) -> HandoffGenerationEvidence: + return (await self.resolve_many((citation,)))[0] + + async def resolve_many( + self, + citations: Sequence[HandoffCitation], + /, + ) -> tuple[HandoffGenerationEvidence, ...]: + requested = tuple(citations) + for citation in requested: + if isinstance(citation, HandoffArtifactCitation) and citation.artifact_ref.family == "prompt": + raise HandoffEvidenceUnavailableError(citation) + source_refs = tuple( + citation.source_ref for citation in requested if isinstance(citation, HandoffSourceCitation) + ) + artifact_refs = tuple( + citation.artifact_ref for citation in requested if isinstance(citation, HandoffArtifactCitation) + ) try: + async with self._database.connection(self._bound_connection) as connection: + sources = await self._sources.require_for_generation(connection, self._scope_id, source_refs) + artifacts = await self._artifacts.get_many(connection, self._scope_id, artifact_refs) + except RepositoryNotFoundError as error: + unavailable = _missing_citation(requested, error) + raise HandoffEvidenceUnavailableError(unavailable) from error + + source_values = {(row.ref.source_type, row.ref.source_id): row.value for row in sources} + artifact_values = { + (artifact.family, artifact.artifact_id, artifact.revision): artifact for artifact in artifacts + } + evidence: list[HandoffGenerationEvidence] = [] + for citation in requested: if isinstance(citation, HandoffSourceCitation): - async with self._database.connection(self._bound_connection) as connection: - source = await self._sources.get(connection, self._scope_id, citation.source_ref) - require_source_eligible(citation.source_ref, source.value) - return HandoffSourceEvidence(citation=citation, source=source.value) - if isinstance(citation, HandoffArtifactCitation): - if citation.artifact_ref.family == "prompt": - raise HandoffEvidenceUnavailableError(citation) - async with self._database.connection(self._bound_connection) as connection: - artifact = await self._artifacts.get(connection, self._scope_id, citation.artifact_ref) - return HandoffArtifactEvidence(citation=citation, artifact=artifact) - if isinstance(citation, HandoffMemoryCitation): - entry = await self._memory.validate_citation(citation.memory_citation) - return HandoffMemoryEvidence(citation=citation, entry=entry) - except ( - ArtifactNotFoundError, - InvalidMemoryCitationError, - MemoryEntryNotFoundError, - RepositoryNotFoundError, - ) as error: - raise HandoffEvidenceUnavailableError(citation) from error - raise TypeError(f"unsupported Handoff citation: {type(citation).__name__}") # noqa: TRY003 + source = source_values[(citation.source_ref.source_type, citation.source_ref.source_id)] + evidence.append(HandoffSourceEvidence(citation=citation, source=source)) + elif isinstance(citation, HandoffArtifactCitation): + reference = citation.artifact_ref + evidence.append( + HandoffArtifactEvidence( + citation=citation, + artifact=artifact_values[(reference.family, reference.artifact_id, reference.revision)], + ) + ) + elif isinstance(citation, HandoffMemoryCitation): + try: + entry = await self._memory.validate_citation(citation.memory_citation) + except (ArtifactNotFoundError, InvalidMemoryCitationError, MemoryEntryNotFoundError) as error: + raise HandoffEvidenceUnavailableError(citation) from error + evidence.append(HandoffMemoryEvidence(citation=citation, entry=entry)) + else: + raise TypeError(f"unsupported Handoff citation: {type(citation).__name__}") # noqa: TRY003 + return tuple(evidence) async def validate(self, citation: HandoffCitation, /) -> None: await self.resolve(citation) +def _missing_citation( + citations: tuple[HandoffCitation, ...], + error: RepositoryNotFoundError, +) -> HandoffCitation: + identity = error.identity + missing = identity[-1] if isinstance(identity, tuple) and identity else None + for citation in citations: + if isinstance(citation, HandoffSourceCitation) and citation.source_ref == missing: + return citation + if isinstance(citation, HandoffArtifactCitation) and citation.artifact_ref == missing: + return citation + if citations: + return citations[0] + raise error + + __all__ = ["RelationalHandoffBackend", "RelationalHandoffEvidenceResolver"] diff --git a/src/powercontext/builtin/persistence/sources.py b/src/powercontext/builtin/persistence/sources.py index 4ebb50c1e6..9e64bd5342 100644 --- a/src/powercontext/builtin/persistence/sources.py +++ b/src/powercontext/builtin/persistence/sources.py @@ -16,13 +16,13 @@ from __future__ import annotations -from collections.abc import Iterable, Mapping +from collections.abc import Iterable, Mapping, Sequence from contextlib import suppress from typing import Any, Literal, cast import rfc8785 from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter -from sqlalchemy import func, insert, select, update +from sqlalchemy import func, insert, select, tuple_, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncConnection @@ -139,6 +139,51 @@ async def get( raise RepositoryNotFoundError("source", (scope_id, ref)) return self._decode_row(row) + async def get_many( + self, + connection: AsyncConnection, + scope_id: str, + refs: Sequence[SourceRef], + /, + ) -> tuple[StoredSource, ...]: + """Load exact Sources in first-requested order using one query.""" + + _require_identity("scope_id", scope_id, MAX_SCOPE_ID_LENGTH) + ordered: list[SourceRef] = [] + identities: set[tuple[str, str]] = set() + for ref in refs: + identity = (ref.source_type, ref.source_id) + if identity not in identities: + ordered.append(ref) + identities.add(identity) + if not ordered: + 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_(tuple(identities)), + ) + ) + ) + .mappings() + .all() + ) + decoded: dict[tuple[str, str], StoredSource] = {} + for row in rows: + try: + stored = self._decode_row(row) + except RepositoryNotFoundError: + ref = SourceRef(source_type=str(row["source_type"]), source_id=str(row["source_id"])) + raise RepositoryNotFoundError("source", (scope_id, ref)) from None + decoded[(stored.ref.source_type, stored.ref.source_id)] = stored + for ref in ordered: + if (ref.source_type, ref.source_id) not in decoded: + raise RepositoryNotFoundError("source", (scope_id, ref)) + return tuple(decoded[(ref.source_type, ref.source_id)] for ref in ordered) + async def list( self, connection: AsyncConnection, @@ -168,6 +213,39 @@ async def list( rows = (await connection.execute(statement)).mappings() return tuple(self._decode_row(row) for row in rows) + async def list_window( + self, + connection: AsyncConnection, + scope_id: str, + /, + *, + after: int, + through: int, + ) -> tuple[StoredSource, ...]: + """Return the exact journal interval ``(after, through]``.""" + + _require_identity("scope_id", scope_id, MAX_SCOPE_ID_LENGTH) + if after < 0: + raise InvalidRepositoryArgumentError("after", "must be non-negative") + if through < after: + raise InvalidRepositoryArgumentError("through", "must not precede after") + rows = ( + ( + await connection.execute( + select(SOURCES_TABLE) + .where( + SOURCES_TABLE.c.scope_id == scope_id, + SOURCES_TABLE.c.journal_position > after, + SOURCES_TABLE.c.journal_position <= through, + ) + .order_by(SOURCES_TABLE.c.journal_position) + ) + ) + .mappings() + .all() + ) + return tuple(self._decode_row(row) for row in rows) + async def journal_position(self, connection: AsyncConnection, scope_id: str, /) -> int: """Return the current high watermark for one scope.""" diff --git a/src/powercontext/builtin/review/generation.py b/src/powercontext/builtin/review/generation.py index 2d8e751493..fa73ed83bd 100644 --- a/src/powercontext/builtin/review/generation.py +++ b/src/powercontext/builtin/review/generation.py @@ -34,11 +34,10 @@ from powercontext.builtin.persistence.artifacts import ArtifactRepository from powercontext.builtin.persistence.database import AsyncDatabase from powercontext.builtin.persistence.errors import RepositoryNotFoundError -from powercontext.builtin.persistence.sources import SourceRepository +from powercontext.builtin.persistence.generation_sources import GenerationSourceAccess from powercontext.builtin.review.errors import InvalidCandidateError from powercontext.builtin.review.models import ArtifactCandidate from powercontext.builtin.review.service import ReviewService -from powercontext.builtin.source_eligibility import require_source_eligible from powercontext.errors import PowerContextError from powercontext.sources import Source, SourceRef @@ -80,7 +79,7 @@ def __init__( *, database: AsyncDatabase, scope_id: str, - sources: SourceRepository, + sources: GenerationSourceAccess, artifacts: ArtifactRepository, review: ReviewService, experience_generator: ExperienceGenerator | None, @@ -156,10 +155,8 @@ async def _evidence( evidence: list[GenerationEvidence] = [] try: async with self._database.transaction() as connection: - for ref in sources: - row = await self._sources.get(connection, self._scope_id, ref) - require_source_eligible(ref, row.value) - evidence.append(_source_evidence(ref, row.value)) + source_rows = await self._sources.require_for_generation(connection, self._scope_id, sources) + evidence.extend(_source_evidence(row.ref, row.value) for row in source_rows) for ref in artifacts: if ref.family == "prompt": raise InvalidCandidateError("evidence", "Prompt configuration is not factual evidence") diff --git a/src/powercontext/builtin/review/service.py b/src/powercontext/builtin/review/service.py index c074452278..57b0ee080e 100644 --- a/src/powercontext/builtin/review/service.py +++ b/src/powercontext/builtin/review/service.py @@ -29,8 +29,8 @@ from powercontext.builtin.persistence.database import AsyncDatabase from powercontext.builtin.persistence.errors import RepositoryNotFoundError from powercontext.builtin.persistence.experience_index import ExperienceIndex +from powercontext.builtin.persistence.generation_sources import GenerationSourceAccess from powercontext.builtin.persistence.skill_packages import SkillPackageRepository -from powercontext.builtin.persistence.sources import SourceRepository from powercontext.builtin.review.errors import ArtifactTargetConflictError, InvalidCandidateError from powercontext.builtin.review.models import ( MAX_CANDIDATE_EVIDENCE, @@ -39,7 +39,6 @@ ArtifactCandidatePage, CandidateStatus, ) -from powercontext.builtin.source_eligibility import require_source_eligible from powercontext.errors import ArtifactNotFoundError, RevisionConflictError from powercontext.sources import SourceRef @@ -62,7 +61,7 @@ def __init__( artifacts: ArtifactRepository, experience_index: ExperienceIndex, skill_packages: SkillPackageRepository, - sources: SourceRepository, + sources: GenerationSourceAccess, id_factory: IdFactory, connection: AsyncConnection | None = None, ) -> None: @@ -286,6 +285,7 @@ async def approve( ) ) _validate_approval_lineage(candidate) + await self._validate_evidence(connection, candidate.sources, candidate.artifacts) if isinstance(candidate.proposal, SkillContent) and candidate.proposal.package is not None: await self._canonical_skill_proposal(connection, candidate.proposal) draft = _candidate_draft(candidate) @@ -353,9 +353,7 @@ async def _validate_evidence( if len(sources) + len(artifacts) > MAX_CANDIDATE_EVIDENCE: raise InvalidCandidateError("evidence", f"must not exceed {MAX_CANDIDATE_EVIDENCE} exact references") try: - for source in sources: - stored = await self._sources.get(connection, self._scope_id, source) - require_source_eligible(source, stored.value) + await self._sources.require_for_generation(connection, self._scope_id, sources) for artifact in artifacts: await self._artifacts.get(connection, self._scope_id, artifact) except RepositoryNotFoundError as error: diff --git a/src/powercontext/builtin/runtime/relational.py b/src/powercontext/builtin/runtime/relational.py index 424e69e349..74f253daba 100644 --- a/src/powercontext/builtin/runtime/relational.py +++ b/src/powercontext/builtin/runtime/relational.py @@ -108,6 +108,7 @@ PromptManagementWriter, SkillManagementWriter, ) +from powercontext.builtin.persistence.generation_sources import GenerationSourceAccess from powercontext.builtin.persistence.handoff import ( RelationalHandoffBackend, RelationalHandoffEvidenceResolver, @@ -241,6 +242,9 @@ class _ScopedServices: prompts: PromptService generation_receipts: HandoffGenerationReceipts + def generation_sources(self) -> GenerationSourceAccess: + return GenerationSourceAccess(self.repositories.sources) + def sources( self, connection: AsyncConnection | None = None, @@ -273,7 +277,13 @@ def memory( embedding_model=self.embedding_model, reranker=self.memory_reranker, rerank_candidate_limit=self.memory_rerank_candidate_limit, - source_resolver=source_resolver, + source_resolver=_RelationalMemorySourceResolver( + database=self.database, + scope_id=self.scope_id, + catalog=source_resolver, + access=self.generation_sources(), + connection=connection, + ), artifact_resolver=_RelationalArtifactResolver( database=self.database, scope_id=self.scope_id, @@ -291,7 +301,7 @@ def review(self, connection: AsyncConnection | None = None) -> ReviewService: artifacts=self.repositories.artifacts, experience_index=self.experience_index, skill_packages=self.repositories.skill_packages, - sources=self.repositories.sources, + sources=self.generation_sources(), id_factory=self.id_factory, connection=connection, ) @@ -301,7 +311,7 @@ def generation(self) -> ReviewedGenerationService: prompt_context=ScopedPrompts(self.prompts, self.scope_id), database=self.database, scope_id=self.scope_id, - sources=self.repositories.sources, + sources=self.generation_sources(), artifacts=self.repositories.artifacts, review=self.review(), experience_generator=self.experience_generator, @@ -320,7 +330,7 @@ def evidence_resolver(scope_id: str) -> RelationalHandoffEvidenceResolver: return RelationalHandoffEvidenceResolver( database=services.database, scope_id=scope_id, - sources=services.repositories.sources, + sources=services.generation_sources(), artifacts=services.repositories.artifacts, memory=services.memory(catalog), ) @@ -338,7 +348,7 @@ def evidence_resolver(scope_id: str) -> RelationalHandoffEvidenceResolver: evidence_resolver=RelationalHandoffEvidenceResolver( database=self.database, scope_id=self.scope_id, - sources=self.repositories.sources, + sources=self.generation_sources(), artifacts=self.repositories.artifacts, memory=memory, ), @@ -1056,6 +1066,40 @@ def _services_for(self, scope_id: str) -> _ScopedServices: ) +class _RelationalMemorySourceResolver: + """Admit Memory evidence without changing the ordinary Source catalog.""" + + def __init__( + self, + *, + database: AsyncDatabase, + scope_id: str, + catalog: SourceCatalog, + access: GenerationSourceAccess, + connection: AsyncConnection | None = None, + ) -> None: + self._database = database + self._scope_id = scope_id + self._catalog = catalog + self._access = access + self._connection = connection + + def as_ref(self, source: Source, /) -> SourceRef: + return self._catalog.as_ref(source) + + async def get(self, source: Source, /) -> Source: + try: + async with self._database.connection(self._connection) as connection: + (stored,) = await self._access.require_for_generation( + connection, self._scope_id, (self.as_ref(source),) + ) + except RepositoryNotFoundError: + raise SourceNotFoundError(source) from None + if type(stored.value) is not type(source) or stored.value != source: + raise SourceNotFoundError(source) + return stored.value + + class _RelationalSources: def __init__( self, @@ -1086,9 +1130,7 @@ async def get(self, source: Source, /) -> Source: ref = self._as_ref(source) try: async with self._database.connection(self._bound_connection) as connection: - value = (await self._repository.get(connection, self._scope_id, ref)).value - require_source_eligible(ref, value) - return value + return (await self._repository.get(connection, self._scope_id, ref)).value except RepositoryNotFoundError: raise SourceNotFoundError(source) from None @@ -1166,10 +1208,10 @@ async def activate_handoff(self, request: ActivateHandoff, /) -> HandoffActivati async with self._lock: async with self._services.database.transaction() as connection: try: - source = await self._services.repositories.sources.get( + (source,) = await self._services.generation_sources().require_for_generation( connection, self._services.scope_id, - request.boundary_source, + (request.boundary_source,), ) except RepositoryNotFoundError: raise HandoffEvidenceUnavailableError( @@ -1292,10 +1334,11 @@ async def _sources( connection: AsyncConnection, action: ProcessSourceWindow, ) -> tuple[Source, ...]: - rows = await self._services.repositories.sources.list( + rows = await self._services.generation_sources().list_window_for_generation( connection, self._services.scope_id, after=action.after, + through=action.through, ) return tuple( row.value for row in rows if row.journal_position <= action.through and is_generation_eligible(row.value) @@ -1411,11 +1454,11 @@ async def _sources( connection: AsyncConnection, action: ProcessSourceWindow, ) -> tuple[StoredSource, ...]: - return await self._services.repositories.sources.list( + return await self._services.generation_sources().list_window_for_generation( connection, self._services.scope_id, after=action.after, - limit=action.through - action.after, + through=action.through, ) diff --git a/src/powercontext/builtin/sources/content.py b/src/powercontext/builtin/sources/content.py index 0029cdb886..0f49917f3d 100644 --- a/src/powercontext/builtin/sources/content.py +++ b/src/powercontext/builtin/sources/content.py @@ -20,6 +20,7 @@ from pydantic import BaseModel, Field, JsonValue, field_validator +from powercontext.limits import MAX_ARTIFACT_FAMILY_LENGTH from powercontext.sources import TEXT_EVIDENCE_PROJECTION_KEY, AdapterSourceDefinition, TextEvidence from powercontext.sources.models import Source, SourceMaterialization @@ -46,10 +47,17 @@ class ContentSourceTarget(BaseModel): """Exact Artifact revision to which one system Source is bound.""" scope_id: str - family: Literal["memory", "experience", "skill", "handoff", "prompt"] + family: Annotated[str, Field(min_length=1, max_length=MAX_ARTIFACT_FAMILY_LENGTH)] artifact_id: str revision: Annotated[int, Field(ge=1)] = 1 + @field_validator("family") + @classmethod + def require_canonical_family(cls, value: str) -> str: + if value != value.strip(): + raise ValueError("family must not contain leading or trailing whitespace") # noqa: TRY003 + return value + class ContentSourceInternal(BaseModel): """Server-owned Source purpose data never exposed by the base REST API.""" diff --git a/src/powercontext/server/app.py b/src/powercontext/server/app.py index 3d3da98b4f..9e5ecc1cf5 100644 --- a/src/powercontext/server/app.py +++ b/src/powercontext/server/app.py @@ -806,7 +806,6 @@ async def get_artifact_revision( ) -> RuntimeArtifactRecord: ... async def logical_artifacts(self) -> tuple[LogicalArtifactRecord, ...]: ... - async def query_artifacts( self, family: str, @@ -4682,8 +4681,8 @@ def _map_base_access_error(error: Exception) -> tuple[int, str, str, dict[str, A return ( status.HTTP_422_UNPROCESSABLE_CONTENT, "source_not_eligible", - "The Source is reserved for its bound Artifact creation lineage.", - {"source": error.source.model_dump(mode="json")}, + "The Source cannot be used as Artifact generation evidence.", + {"source_ref": error.source.model_dump(mode="json")}, ) if isinstance(error, _PreconditionRequiredError): return ( diff --git a/tests/builtin/persistence/test_generation_sources.py b/tests/builtin/persistence/test_generation_sources.py new file mode 100644 index 0000000000..32809b9834 --- /dev/null +++ b/tests/builtin/persistence/test_generation_sources.py @@ -0,0 +1,115 @@ +# 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. + +from __future__ import annotations + +import asyncio + +import pytest +from sqlalchemy import update + +from powercontext.builtin.persistence import InvalidStoredPayloadError +from powercontext.builtin.persistence.generation_sources import GenerationSourceAccess +from powercontext.builtin.persistence.sources import SourceRepository +from powercontext.builtin.persistence.tables import SOURCES_TABLE +from powercontext.builtin.source_eligibility import SourceNotEligibleError +from powercontext.builtin.sources import ( + CONTENT_SOURCE_ADAPTER, + ContentSource, + ContentSourceInternal, + ContentSourceTarget, +) +from powercontext.sources import SourceMaterialization +from tests.builtin.persistence.contract import repository_profile + + +def _content_source(name: str, *, lineage_only: bool = False) -> ContentSource: + internal = ( + ContentSourceInternal( + role="lineage_only", + operation="artifact_create", + target=ContentSourceTarget( + scope_id="scope-a", + family="future-family", + artifact_id="artifact-1", + revision=1, + ), + ) + if lineage_only + else None + ) + return ContentSource( + name=name, + materialization=SourceMaterialization.CAPTURED, + content=name, + internal=internal, + ) + + +def test_generation_source_access_rejects_explicit_refs_but_skips_window_entries() -> None: + async def scenario() -> None: + repository = SourceRepository((CONTENT_SOURCE_ADAPTER,)) + access = GenerationSourceAccess(repository) + async with repository_profile() as (profile, _repositories): + async with profile.database.transaction() as connection: + ordinary = await repository.add(connection, "scope-a", _content_source("ordinary")) + reserved = await repository.add( + connection, + "scope-a", + _content_source("reserved", lineage_only=True), + ) + + with pytest.raises(SourceNotEligibleError) as error: + await access.require_for_generation(connection, "scope-a", (ordinary.ref, reserved.ref)) + window = await access.list_window_for_generation( + connection, + "scope-a", + after=0, + through=reserved.journal_position, + ) + + assert error.value.source == reserved.ref + assert window == (ordinary,) + + asyncio.run(scenario()) + + +def test_generation_window_fails_closed_on_a_malformed_source_payload() -> None: + async def scenario() -> None: + repository = SourceRepository((CONTENT_SOURCE_ADAPTER,)) + access = GenerationSourceAccess(repository) + async with ( + repository_profile() as (profile, _repositories), + profile.database.transaction() as connection, + ): + stored = await repository.add(connection, "scope-a", _content_source("broken")) + await connection.execute( + update(SOURCES_TABLE) + .where( + SOURCES_TABLE.c.scope_id == "scope-a", + SOURCES_TABLE.c.source_type == stored.ref.source_type, + SOURCES_TABLE.c.source_id == stored.ref.source_id, + ) + .values(payload=b"not-json") + ) + + with pytest.raises(InvalidStoredPayloadError): + await access.list_window_for_generation( + connection, + "scope-a", + after=0, + through=stored.journal_position, + ) + + asyncio.run(scenario()) diff --git a/tests/builtin/persistence/test_provider.py b/tests/builtin/persistence/test_provider.py index 01de310d62..310fa8f74d 100644 --- a/tests/builtin/persistence/test_provider.py +++ b/tests/builtin/persistence/test_provider.py @@ -29,7 +29,9 @@ ) from powercontext.builtin.artifacts.memory import MemoryCandidateRequest, MemoryEntryInput from powercontext.builtin.persistence.sqlite import SQLiteConfig +from powercontext.builtin.records import ArtifactWrite from powercontext.builtin.runtime import BuiltinConfig, open_builtin_contexts +from powercontext.builtin.source_eligibility import SourceNotEligibleError from powercontext.builtin.sources import BUILTIN_SOURCE_REGISTRY, ContentCapture, ContentSource, SourceCursor @@ -87,6 +89,39 @@ class StateSaveFailure(RuntimeError): pass +@pytest.mark.parametrize("mode", ["extract", "auto"]) +def test_explicit_memory_extraction_rejects_lineage_only_before_pipeline(mode) -> None: + class EmptyPipeline: + called = False + + async def extract(self, request): + self.called = True + return () + + async def scenario() -> None: + pipeline = EmptyPipeline() + async with open_builtin_contexts( + BuiltinConfig(database=SQLiteConfig()), + candidate_pipeline=pipeline, + ) as contexts: + context = await contexts.get("project") + created = await contexts.records.create_artifact( + "project", + "memory", + ArtifactWrite(content={"entries": [{"kind": "fact", "text": "Managed input."}]}), + ) + async with contexts.database.transaction() as connection: + stored = await contexts.repositories.sources.get(connection, "project", created.sources[0]) + assert await context.sources.get(stored.value) == stored.value + with pytest.raises(SourceNotEligibleError): + await context.artifacts.memory.remember(memory=None, sources=(stored.value,), mode=mode) + assert pipeline.called is False + head = await contexts.records.get_artifact("project", "memory", created.artifact_id) + assert head.revision == 1 + + asyncio.run(scenario()) + + def test_provider_uses_one_injected_source_registry_for_routing_and_persistence() -> None: async def scenario() -> None: registry = SourceDefinitionRegistry(( @@ -108,6 +143,30 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_lineage_only_source_remains_readable_but_is_skipped_by_memory_flush() -> None: + async def scenario() -> None: + async with open_builtin_contexts( + BuiltinConfig(database=SQLiteConfig()), + candidate_pipeline=EchoCandidatePipeline(), + ) as contexts: + context = await contexts.get("project") + created = await contexts.records.create_artifact( + "project", + "memory", + ArtifactWrite(content={"entries": [{"kind": "fact", "text": "Directly managed."}]}), + ) + async with contexts.database.transaction() as connection: + stored = await contexts.repositories.sources.get(connection, "project", created.sources[0]) + + assert await context.sources.get(stored.value) == stored.value + flushed = await context.triggers.flush(limit=10) + assert flushed.source_count == 0 + assert flushed.current_cursor == stored.journal_position + assert flushed.memory_ref is None + + asyncio.run(scenario()) + + def test_provider_translates_repository_source_identity_conflicts() -> None: async def scenario() -> None: async with open_builtin_contexts(BuiltinConfig(database=SQLiteConfig())) as contexts: diff --git a/tests/builtin/persistence/test_sources.py b/tests/builtin/persistence/test_sources.py index 19289926f3..22e080499f 100644 --- a/tests/builtin/persistence/test_sources.py +++ b/tests/builtin/persistence/test_sources.py @@ -22,6 +22,7 @@ from powercontext import InvalidSourceDefinitionError from powercontext.builtin.persistence import ( InvalidRepositoryArgumentError, + RepositoryNotFoundError, StoredPayloadConflictError, ) from powercontext.builtin.persistence.codec import dump_model @@ -98,6 +99,56 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_repository_batch_reads_deduplicate_refs_and_bound_the_journal_window() -> None: + async def scenario() -> None: + async with repository_profile() as (profile, repositories): + first_source = NoteSource( + name="note-1", + materialization=SourceMaterialization.CAPTURED, + body="first", + ) + second_source = NoteSource( + name="note-2", + materialization=SourceMaterialization.CAPTURED, + body="second", + ) + third_source = NoteSource( + name="note-3", + materialization=SourceMaterialization.CAPTURED, + body="third", + ) + async with profile.database.transaction() as connection: + first = await repositories.sources.add(connection, "scope-a", first_source) + second = await repositories.sources.add(connection, "scope-a", second_source) + third = await repositories.sources.add(connection, "scope-a", third_source) + + loaded = await repositories.sources.get_many( + connection, + "scope-a", + (second.ref, first.ref, second.ref), + ) + window = await repositories.sources.list_window( + connection, + "scope-a", + after=first.journal_position, + through=second.journal_position, + ) + + assert loaded == (second, first) + assert window == (second,) + assert third.journal_position > second.journal_position + with pytest.raises(RepositoryNotFoundError): + await repositories.sources.get_many( + connection, + "scope-a", + (SourceRef(source_type="note", source_id="missing"),), + ) + with pytest.raises(InvalidRepositoryArgumentError): + await repositories.sources.list_window(connection, "scope-a", after=2, through=1) + + asyncio.run(scenario()) + + def test_repository_rejects_a_scope_outside_the_relational_identity_baseline() -> None: async def scenario() -> None: async with repository_profile() as (profile, repositories): diff --git a/tests/builtin/review/test_generation.py b/tests/builtin/review/test_generation.py index 03a0defc07..93a30e6fa3 100644 --- a/tests/builtin/review/test_generation.py +++ b/tests/builtin/review/test_generation.py @@ -22,6 +22,7 @@ from powercontext.builtin.artifacts.generation import ArtifactGenerationInput from powercontext.builtin.artifacts.skill import SkillContent from powercontext.builtin.persistence.sqlite import SQLiteConfig +from powercontext.builtin.records import ArtifactWrite from powercontext.builtin.review import InvalidCandidateError from powercontext.builtin.review.generation import GenerationCapabilityUnavailableError from powercontext.builtin.runtime import ( @@ -36,6 +37,7 @@ open_builtin_runtime, ) from powercontext.builtin.scope import ScopeDraft +from powercontext.builtin.source_eligibility import SourceNotEligibleError async def _create_scope(runtime: BuiltinRuntime, idempotency_key: str = "project") -> str: @@ -150,6 +152,30 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_explicit_generation_rejects_a_lineage_only_source_before_model_use() -> None: + async def scenario() -> None: + generator = _ExperienceGenerator(_experience()) + async with open_builtin_runtime( + BuiltinConfig(database=SQLiteConfig()), + experience_generator=generator, + ) as runtime: + scope_id = await _create_scope(runtime) + created = await runtime.records.for_scope(scope_id).create_artifact( + "memory", + ArtifactWrite(content={"entries": [{"kind": "fact", "text": "Direct input."}]}), + ) + + with pytest.raises(SourceNotEligibleError) as error: + await runtime.experience.for_scope(scope_id).generate( + GenerateExperienceRequest(sources=(created.sources[0],)) + ) + + assert error.value.source == created.sources[0] + assert generator.inputs == [] + + asyncio.run(scenario()) + + def test_experience_generation_targets_an_exact_approved_revision() -> None: async def scenario() -> None: generator = _ExperienceGenerator(_experience()) diff --git a/tests/builtin/review/test_service.py b/tests/builtin/review/test_service.py index 1b4734f7bd..5d38a2ff89 100644 --- a/tests/builtin/review/test_service.py +++ b/tests/builtin/review/test_service.py @@ -27,6 +27,7 @@ from powercontext.builtin.persistence.errors import RepositoryNotFoundError from powercontext.builtin.persistence.sqlite import SQLiteConfig, SQLiteProfile from powercontext.builtin.persistence.tables import ARTIFACTS_TABLE, BUILTIN_TABLES +from powercontext.builtin.records import ArtifactWrite from powercontext.builtin.review import ( ArtifactTargetConflictError, CandidateConflictError, @@ -53,6 +54,7 @@ ) from powercontext.builtin.runtime.relational import RelationalContexts from powercontext.builtin.scope import ScopeDraft +from powercontext.builtin.source_eligibility import SourceNotEligibleError from powercontext.builtin.sources import ContentCapture @@ -191,6 +193,48 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_approval_rechecks_sources_saved_by_an_older_candidate_path() -> None: + async def scenario() -> None: + async with SQLiteProfile.open(SQLiteConfig(), tables=BUILTIN_TABLES) as profile: + contexts = RelationalContexts(database=profile.database) + await contexts.get("project") + created = await contexts.records.create_artifact( + "project", + "memory", + ArtifactWrite(content={"entries": [{"kind": "fact", "text": "Direct input."}]}), + ) + async with profile.database.transaction() as connection: + candidate = await contexts.repositories.candidates.create( + connection, + "project", + "candidate-legacy", + "experience", + _proposal(), + sources=created.sources, + artifacts=(), + target=None, + reason=None, + ) + + with pytest.raises(SourceNotEligibleError): + await contexts.review("project").approve(candidate.candidate_id, candidate.version) + + current = await contexts.review("project").get_candidate(candidate.candidate_id) + async with profile.database.transaction() as connection: + experiences = await connection.scalar( + select(func.count()) + .select_from(ARTIFACTS_TABLE) + .where( + ARTIFACTS_TABLE.c.scope_id == "project", + ARTIFACTS_TABLE.c.family == Experience.family, + ) + ) + assert current.status == "pending" + assert experiences == 0 + + asyncio.run(scenario()) + + def test_experience_candidate_revise_approve_and_retrieval_gate() -> None: async def scenario() -> None: async with open_builtin_runtime(BuiltinConfig(database=SQLiteConfig())) as runtime: diff --git a/tests/e2e/test_handoff_runtime.py b/tests/e2e/test_handoff_runtime.py index 8d311cce2d..fa1a42a793 100644 --- a/tests/e2e/test_handoff_runtime.py +++ b/tests/e2e/test_handoff_runtime.py @@ -21,9 +21,15 @@ import pytest from powercontext.artifacts import ArtifactRef -from powercontext.builtin.artifacts.handoff import HandoffScopeMismatchError +from powercontext.builtin.artifacts.handoff import ( + HandoffEvidenceUnavailableError, + HandoffScopeMismatchError, + PrepareHandoff, +) from powercontext.builtin.artifacts.memory import MemoryEntryInput -from powercontext.builtin.persistence.sqlite import SQLiteConfig +from powercontext.builtin.persistence.sqlite import SQLiteConfig, SQLiteProfile +from powercontext.builtin.persistence.tables import BUILTIN_TABLES +from powercontext.builtin.records import ArtifactWrite from powercontext.builtin.runtime import ( ActivateHandoff, BuiltinConfig, @@ -34,12 +40,17 @@ HandoffOmission, HandoffSourceCitation, HandoffStatement, + InferenceConfig, PreparedHandoff, RememberMemoryRequest, open_builtin_runtime, ) +from powercontext.builtin.runtime.relational import RelationalContexts from powercontext.builtin.scope import ScopeDraft +from powercontext.builtin.source_eligibility import SourceNotEligibleError from powercontext.errors import RevisionConflictError +from powercontext.sources import SourceDefinitionRegistry, SourceRef +from tests.builtin.persistence.contract import SOURCE_ADAPTERS, CommitAdapter, CommitInput, NoteInput class _EchoHandoffPipeline: @@ -61,6 +72,111 @@ async def generate(self, request, /) -> HandoffDraft: ) +@pytest.mark.parametrize("family", ["unregistered-family", "experience"]) +def test_handoff_batch_error_identifies_missing_artifact_after_valid_source(family) -> None: + async def scenario() -> None: + async with open_builtin_runtime( + BuiltinConfig(database=SQLiteConfig()), + handoff_pipeline=_EchoHandoffPipeline(), + ) as runtime: + assert runtime.scopes is not None + scope = await runtime.scopes.create( + ScopeDraft(title="Batch evidence", summary="Error identity", idempotency_key="batch-error") + ) + source = await runtime.sources.for_scope(scope.scope_id).capture( + CaptureSource(source_id="valid", content="Valid evidence.", metadata={}) + ) + missing = HandoffArtifactCitation( + artifact_ref=ArtifactRef(family=family, artifact_id="missing", revision=1) + ) + with pytest.raises(HandoffEvidenceUnavailableError) as error: + await runtime.handoff.for_scope(scope.scope_id).prepare( + PrepareHandoff( + objective="Report missing evidence.", + evidence=(HandoffSourceCitation(source_ref=source.source_ref), missing), + ) + ) + assert error.value.citation == missing + assert await runtime.handoff.for_scope(scope.scope_id).latest() is None + + asyncio.run(scenario()) + + +def test_handoff_batch_rejects_existing_prompt_as_evidence() -> None: + async def scenario() -> None: + async with open_builtin_runtime( + BuiltinConfig(database=SQLiteConfig(), inference=InferenceConfig(generation_model="test")), + handoff_pipeline=_EchoHandoffPipeline(), + ) as runtime: + assert runtime.scopes is not None + scope = await runtime.scopes.create( + ScopeDraft(title="Prompt boundary", summary="Evidence isolation", idempotency_key="prompt-evidence") + ) + source = await runtime.sources.for_scope(scope.scope_id).capture( + CaptureSource(source_id="valid", content="Valid evidence.", metadata={}) + ) + prompt = await runtime.records.for_scope(scope.scope_id).create_artifact( + "prompt", + ArtifactWrite( + prompt_key="memory.extract", + content={ + "schema_version": "powercontext.prompt.v1", + "mode": "custom", + "instructions": "Keep personal preferences.", + "demonstrations": [], + }, + ), + ) + citation = HandoffArtifactCitation( + artifact_ref=ArtifactRef(family="prompt", artifact_id=prompt.artifact_id, revision=prompt.revision) + ) + handoffs = runtime.handoff.for_scope(scope.scope_id) + with pytest.raises(HandoffEvidenceUnavailableError) as error: + await handoffs.prepare( + PrepareHandoff( + objective="Do not use instructions as evidence.", + evidence=(HandoffSourceCitation(source_ref=source.source_ref), citation), + ) + ) + assert error.value.citation == citation + assert await handoffs.latest() is None + + asyncio.run(scenario()) + + +def test_handoff_batch_error_identifies_source_with_unregistered_definition() -> None: + async def scenario() -> None: + async with SQLiteProfile.open(SQLiteConfig(), tables=BUILTIN_TABLES) as profile: + original = RelationalContexts( + database=profile.database, + source_registry=SourceDefinitionRegistry.from_adapters(SOURCE_ADAPTERS), + ) + context = await original.get("project") + valid = await context.sources.add(await context.sources.resolve(CommitInput(revision="valid"))) + unavailable = await context.sources.add( + await context.sources.resolve(NoteInput(note_id="unavailable", body="Historical note.")) + ) + citations = ( + HandoffSourceCitation(source_ref=SourceRef(source_type="commit", source_id=valid.name)), + HandoffSourceCitation(source_ref=SourceRef(source_type="note", source_id=unavailable.name)), + ) + + # Reopen the same data without the historical Note adapter. + reopened = RelationalContexts( + database=profile.database, + source_registry=SourceDefinitionRegistry.from_adapters((CommitAdapter(),)), + handoff_pipeline=_EchoHandoffPipeline(), + ) + handoffs = (await reopened.get("project")).artifacts.handoff + with pytest.raises(HandoffEvidenceUnavailableError) as error: + await handoffs.prepare(PrepareHandoff(objective="Report unavailable evidence.", evidence=citations)) + + assert error.value.citation == citations[1] + assert await handoffs.latest() is None + + asyncio.run(scenario()) + + def test_runtime_owns_handoff_trigger_activation_and_deduplication() -> None: async def scenario() -> None: async with open_builtin_runtime( @@ -106,6 +222,33 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_handoff_activation_rejects_a_lineage_only_boundary_source() -> None: + async def scenario() -> None: + pipeline = _EchoHandoffPipeline() + async with open_builtin_runtime( + BuiltinConfig(database=SQLiteConfig()), + handoff_pipeline=pipeline, + ) as runtime: + assert runtime.scopes is not None + scope = await runtime.scopes.create( + ScopeDraft(title="Project", summary="Reserved boundary", idempotency_key="reserved-boundary") + ) + created = await runtime.records.for_scope(scope.scope_id).create_artifact( + "memory", + ArtifactWrite(content={"entries": [{"kind": "fact", "text": "Direct input."}]}), + ) + + with pytest.raises(SourceNotEligibleError): + await runtime.handoff.for_scope(scope.scope_id).activate( + ActivateHandoff( + boundary_source=created.sources[0], + objective="Do not regenerate from management provenance.", + ) + ) + + asyncio.run(scenario()) + + def test_handoff_trigger_state_survives_runtime_restart(tmp_path: Path) -> None: async def scenario() -> None: config = BuiltinConfig(