diff --git a/src/powercontext/builtin/handoff_report/adapters.py b/src/powercontext/builtin/handoff_report/adapters.py index 6f617edeb8..e0504922d1 100644 --- a/src/powercontext/builtin/handoff_report/adapters.py +++ b/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"] diff --git a/src/powercontext/builtin/handoff_report/application.py b/src/powercontext/builtin/handoff_report/application.py index 2f5ad4039d..11223d8f4b 100644 --- a/src/powercontext/builtin/handoff_report/application.py +++ b/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, ) ) diff --git a/src/powercontext/builtin/handoff_report/protocols.py b/src/powercontext/builtin/handoff_report/protocols.py index 8ffc98f2c4..cc66e3f72c 100644 --- a/src/powercontext/builtin/handoff_report/protocols.py +++ b/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): diff --git a/src/powercontext/builtin/runtime/composition.py b/src/powercontext/builtin/runtime/composition.py index 6862718bd6..33cac25fd6 100644 --- a/src/powercontext/builtin/runtime/composition.py +++ b/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") diff --git a/src/powercontext/builtin/runtime/relational.py b/src/powercontext/builtin/runtime/relational.py index 710afeffb5..03aebe12aa 100644 --- a/src/powercontext/builtin/runtime/relational.py +++ b/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.""" diff --git a/tests/test_handoff_report.py b/tests/test_handoff_report.py index 88fd35a02e..eeb20521e8 100644 --- a/tests/test_handoff_report.py +++ b/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