perf(handoff-report): read selection heads in one snapshot

src/powercontext/builtin/handoff_report/adapters.py

@@ -16,31 +16,67 @@
 
 from __future__ import annotations
 
-from typing import Protocol
+from collections.abc import Sequence
+from typing import cast
+
+from sqlalchemy import select
 
 from powercontext.artifacts import ArtifactRef
 from powercontext.builtin.artifacts.handoff import Handoff
+from powercontext.builtin.handoff_report.errors import HandoffReportInconsistentError
+from powercontext.builtin.persistence.artifacts import ArtifactRepository
+from powercontext.builtin.persistence.database import AsyncDatabase
+from powercontext.builtin.persistence.tables import ARTIFACT_HEADS_TABLE
 
-
-class _ScopedHandoffReader(Protocol):
-    async def latest(self) -> Handoff | None: ...
-
-    async def revision(self, reference: ArtifactRef, /) -> Handoff: ...
-
-
-class _HandoffApplicationReader(Protocol):
-    def for_scope(self, scope_id: str, /) -> _ScopedHandoffReader: ...
+_HEAD_BATCH = 100
 
 
 class RuntimeHandoffReadAdapter:
-    def __init__(self, application: _HandoffApplicationReader, /) -> None:
-        self._application = application
+    """Read the committed Handoff of every selected Scope in one snapshot."""
 
-    async def latest(self, scope_id: str, /) -> Handoff | None:
-        return await self._application.for_scope(scope_id).latest()
+    def __init__(
+        self,
+        database: AsyncDatabase,
+        artifacts: ArtifactRepository,
+        handoff_artifact_id: str,
+        /,
+    ) -> None:
+        self._database = database
+        self._artifacts = artifacts
+        self._handoff_artifact_id = handoff_artifact_id
 
-    async def get(self, scope_id: str, reference: ArtifactRef, /) -> Handoff:
-        return await self._application.for_scope(scope_id).revision(reference)
+    async def latest_many(self, scope_ids: Sequence[str], /) -> dict[str, Handoff | None]:
+        if not scope_ids:
+            return {}
+        heads: dict[str, int] = {}
+        loaded: dict[str, Handoff | None] = {}
+        async with self._database.transaction(consistent_snapshot=True) as connection:
+            for offset in range(0, len(scope_ids), _HEAD_BATCH):
+                rows = (
+                    await connection.execute(
+                        select(ARTIFACT_HEADS_TABLE.c.scope_id, ARTIFACT_HEADS_TABLE.c.revision).where(
+                            ARTIFACT_HEADS_TABLE.c.scope_id.in_(scope_ids[offset : offset + _HEAD_BATCH]),
+                            ARTIFACT_HEADS_TABLE.c.family == Handoff.family,
+                            ARTIFACT_HEADS_TABLE.c.artifact_id == self._handoff_artifact_id,
+                        )
+                    )
+                ).all()
+                heads.update((str(scope_id), int(revision)) for scope_id, revision in rows)
+            for scope_id in scope_ids:
+                revision = heads.get(scope_id)
+                if revision is None:
+                    loaded[scope_id] = None
+                    continue
+                reference = ArtifactRef(
+                    family=Handoff.family,
+                    artifact_id=self._handoff_artifact_id,
+                    revision=revision,
+                )
+                handoff = cast(Handoff, await self._artifacts.get(connection, scope_id, reference))
+                if handoff.as_ref() != reference:
+                    raise HandoffReportInconsistentError(scope_id)
+                loaded[scope_id] = handoff
+        return loaded
 
 
 __all__ = ["RuntimeHandoffReadAdapter"]

src/powercontext/builtin/handoff_report/application.py

@@ -20,7 +20,6 @@
 
 from powercontext.artifacts import ArtifactAddress
 from powercontext.builtin.handoff_report.canonical import finalize_digests
-from powercontext.builtin.handoff_report.errors import HandoffReportInconsistentError
 from powercontext.builtin.handoff_report.protocols import HandoffReadAdapter, ScopeSelectionResolver
 from powercontext.builtin.handoff_report.report import HandoffReport, HandoffReportSummary, ScopeHandoffReport
 from powercontext.builtin.scope.models import ScopeSelection
@@ -43,23 +42,21 @@ async def get_report(
         """Resolve one common selection and freeze each Scope at an exact Handoff revision."""
 
         scopes = await self._scopes.resolve_selection(selection)
+        latest = await self._handoffs.latest_many(tuple(scope.scope_id for scope in scopes))
         projected: list[ScopeHandoffReport] = []
         for scope in scopes:
-            latest = await self._handoffs.latest(scope.scope_id)
-            if latest is None:
+            handoff = latest.get(scope.scope_id)
+            if handoff is None:
                 projected.append(ScopeHandoffReport(scope=scope, status="no_handoff"))
                 continue
 
-            reference = latest.as_ref()
-            frozen = await self._handoffs.get(scope.scope_id, reference)
-            if frozen.as_ref() != reference:
-                raise HandoffReportInconsistentError(scope.scope_id)
+            reference = handoff.as_ref()
             projected.append(
                 ScopeHandoffReport(
                     scope=scope,
-                    status=frozen.content.disposition,
+                    status=handoff.content.disposition,
                     handoff=ArtifactAddress(scope_id=scope.scope_id, artifact=reference),
-                    content=frozen.content,
+                    content=handoff.content,
                 )
             )
 

src/powercontext/builtin/handoff_report/protocols.py

@@ -16,17 +16,15 @@
 
 from __future__ import annotations
 
+from collections.abc import Sequence
 from typing import Protocol
 
-from powercontext.artifacts import ArtifactRef
 from powercontext.builtin.artifacts.handoff import Handoff
 from powercontext.builtin.scope.models import ScopeDescriptor, ScopeSelection
 
 
 class HandoffReadAdapter(Protocol):
-    async def latest(self, scope_id: str, /) -> Handoff | None: ...
-
-    async def get(self, scope_id: str, reference: ArtifactRef, /) -> Handoff: ...
+    async def latest_many(self, scope_ids: Sequence[str], /) -> dict[str, Handoff | None]: ...
 
 
 class ScopeSelectionResolver(Protocol):

src/powercontext/builtin/runtime/composition.py

@@ -893,7 +893,11 @@ async def run_profile(scope_id, high):
         if config.handoff_report.enabled:
             runtime.handoff_report = HandoffReportApplication(
                 contexts.scopes,
-                RuntimeHandoffReadAdapter(runtime.handoff),
+                RuntimeHandoffReadAdapter(
+                    contexts.database,
+                    contexts.repositories.artifacts,
+                    contexts.handoff_artifact_id,
+                ),
             )
         if config.runtime.memory_rerank_enabled and configured_reranker is None:
             raise BuiltinConfigurationError("memory-reranker")

src/powercontext/builtin/runtime/relational.py

@@ -742,6 +742,10 @@ def __init__(
         self._tracing = tracing
         self._scope_states: dict[str, _ScopeState] = {}
 
+    @property
+    def handoff_artifact_id(self) -> str:
+        return self._handoff_artifact_id
+
     @property
     def token_estimator(self) -> TokenEstimator:
         """Return the deployment-fixed estimator used by internal processors."""

tests/test_handoff_report.py

@@ -15,6 +15,8 @@
 from __future__ import annotations
 
 from fastapi.testclient import TestClient
+from sqlalchemy import event
+from sqlalchemy.engine import Engine
 
 from powercontext.builtin.persistence.sqlite import SQLiteConfig
 from powercontext.builtin.runtime.config import HandoffReportConfig
@@ -150,3 +152,39 @@ def _commit_handoff(client: TestClient, scope_id: str, *, disposition: str) -> d
     )
     assert committed.status_code == 200
     return committed.json()["reference"]
+
+
+def test_handoff_report_statement_count_does_not_grow_with_scope_count(tmp_path) -> None:
+    """A selection must not read each Scope's Handoff separately."""
+
+    def counted(scope_count: int) -> tuple[int, int]:
+        app = create_server_app(
+            settings=ServerSettings(
+                database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / f'report-{scope_count}.db'}"),
+                handoff_report=HandoffReportConfig(enabled=True),
+                mcp=McpConfig(enabled=False),
+            )
+        )
+        statements: list[str] = []
+
+        def record(_conn, _cursor, statement, *_args) -> None:
+            statements.append(statement)
+
+        with TestClient(app) as client:
+            for index in range(scope_count):
+                scope_id = _create_scope(client, title=f"Scope {index}", key=f"batch-{scope_count}-{index}")
+                _commit_handoff(client, scope_id, disposition="continuable")
+            event.listen(Engine, "before_cursor_execute", record)
+            try:
+                response = client.post(
+                    "/v1/handoff-reports/get",
+                    json={"selection": {"mode": "all"}, "format": "json"},
+                )
+            finally:
+                event.remove(Engine, "before_cursor_execute", record)
+        assert response.status_code == 200
+        return len(response.json()["report"]["scope_ids"]), len(statements)
+
+    selected, statements = counted(16)
+
+    assert statements <= 2 * selected + 3