feat(generation): standardize source access

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,

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,

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,

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"]

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"]

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."""