perf(sources): bound the receipt migration scan and detect shared plugin drift

integrations/workbuddy/plugins/powercontext/hooks/prepared_context.py

@@ -12,14 +12,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Strict parsing for final context prepared by the PowerContext Runtime."""
+"""The prepare contract: the recall query a host sends and the context it accepts."""
 
 from __future__ import annotations
 
 from collections.abc import Mapping
 
 PREPARED_CONTEXT_SCHEMA = "powercontext.prepared-context.v1"
 MAX_CONTEXT_BYTES = 8_000
+# ``POST /v1/context/prepare`` bounds a query in characters. The byte bound is the Hook's own
+# margin, so a recall also stays acceptable to a server version that measures the storage layer's
+# limit. Both hosts sharing this module recall under the same bounds.
+MAX_QUERY_CHARACTERS = 8_192
+MAX_QUERY_BYTES = 8_192
 
 _PREPARED_CONTEXT_FIELDS = frozenset({"schema", "status", "content", "content_bytes"})
 
@@ -61,3 +66,24 @@ def validate_prepared_context(response: Mapping[str, object]) -> dict[str, objec
         "content": content,
         "content_bytes": content_bytes,
     }
+
+
+def bounded_recall_query(prompt: str) -> str:
+    """Reduce a submitted prompt to the deterministic, bounded query an automatic recall sends.
+
+    The Runtime bounds a prepare query in characters, so a host that forwards a longer prompt
+    unchanged loses the recall to a request-validation error. A prompt within the bounds is used
+    as submitted. A longer prompt keeps its most recent text: that tail holds the turn in hand,
+    while the text before it is what the surrounding transcript already carries. The tail is cut
+    by character and then by UTF-8 byte, so the result splits no character, fits both bounds, and
+    keeps at least the last non-blank character of a prompt that was not blank.
+    """
+
+    query = prompt.strip()
+    if len(query) <= MAX_QUERY_CHARACTERS and len(query.encode("utf-8")) <= MAX_QUERY_BYTES:
+        return query
+    retained = query[-MAX_QUERY_CHARACTERS:]
+    encoded = retained.encode("utf-8")
+    if len(encoded) > MAX_QUERY_BYTES:
+        retained = encoded[-MAX_QUERY_BYTES:].decode("utf-8", errors="ignore")
+    return retained.strip()

src/powercontext/builtin/persistence/receipt_migration.py

@@ -45,88 +45,122 @@ async def migrate_handoff_receipts(  # noqa: C901 - Keep receipt classification
     """
     if batch_size < 1:
         raise ValueError("batch_size must be positive")  # noqa: TRY003
-    after = ("", "")
     attested = unresolved = 0
-    while True:
-        async with database.transaction() as connection:
-            rows = (
-                await connection.execute(
-                    select(SOURCES_TABLE.c.scope_id, SOURCES_TABLE.c.source_id)
+    # Page one Scope at a time. A cursor that spans Scopes cannot steer the
+    # (scope_id, source_type, source_id) primary key, so each page used to rescan every
+    # row already visited and one Scope cost O(N²) index reads. Inside a Scope the
+    # equality on the first two key columns turns the cursor into a bounded seek.
+    after_scope = ""
+    while (current_scope := await _next_content_scope(database, after_scope)) is not None:
+        after_source_id = ""
+        while True:
+            rows = await _page_content_sources(database, current_scope, after_source_id, batch_size)
+            if not rows:
+                break
+            cleanup = []
+            candidates = []
+            seen = set()
+            async with database.transaction() as connection:
+                statement = (
+                    select(SOURCES_TABLE)
                     .where(
+                        SOURCES_TABLE.c.scope_id == current_scope,
                         SOURCES_TABLE.c.source_type == "content",
-                        tuple_(SOURCES_TABLE.c.scope_id, SOURCES_TABLE.c.source_id) > after,
+                        SOURCES_TABLE.c.source_id.in_(rows),
                     )
-                    .order_by(SOURCES_TABLE.c.scope_id, SOURCES_TABLE.c.source_id)
-                    .limit(batch_size)
-                )
-            ).all()
-        if not rows:
-            return attested, unresolved
-        cleanup = []
-        candidates = []
-        seen = set()
-        async with database.transaction() as connection:
-            statement = (
-                select(SOURCES_TABLE)
-                .where(
-                    SOURCES_TABLE.c.source_type == "content",
-                    tuple_(SOURCES_TABLE.c.scope_id, SOURCES_TABLE.c.source_id).in_(rows),
+                    .order_by(SOURCES_TABLE.c.source_id)
                 )
-                .order_by(SOURCES_TABLE.c.scope_id, SOURCES_TABLE.c.source_id)
-            )
-            async with connection.stream(statement.execution_options(yield_per=1)) as result:
-                async for row in result.mappings():
-                    identity = (row["scope_id"], row["source_id"])
-                    seen.add(identity)
-                    source = sources._decode_row(row).value
-                    if isinstance(source, ContentSource):
-                        if source.handoff_receipt:
-                            cleanup.append(identity)
-                        elif _is_receipt_candidate(source):
-                            candidates.append(identity)
-        missing = set(rows) - seen
-        if missing:
-            scope_id, source_id = min(missing)
-            raise RepositoryNotFoundError("source", (scope_id, SourceRef(source_type="content", source_id=source_id)))
-        if cleanup:
-            async with database.transaction() as connection:
-                await connection.execute(
-                    delete(RECEIPT_MIGRATION_REVIEW_TABLE).where(
-                        tuple_(
-                            RECEIPT_MIGRATION_REVIEW_TABLE.c.scope_id, RECEIPT_MIGRATION_REVIEW_TABLE.c.source_id
-                        ).in_(cleanup)
-                    )
+                async with connection.stream(statement.execution_options(yield_per=1)) as result:
+                    async for row in result.mappings():
+                        seen.add(row["source_id"])
+                        source = sources._decode_row(row).value
+                        if isinstance(source, ContentSource):
+                            identity = (row["scope_id"], row["source_id"])
+                            if source.handoff_receipt:
+                                cleanup.append(identity)
+                            elif _is_receipt_candidate(source):
+                                candidates.append(identity)
+            missing = set(rows) - seen
+            if missing:
+                raise RepositoryNotFoundError(
+                    "source",
+                    (current_scope, SourceRef(source_type="content", source_id=min(missing))),
                 )
-        after = tuple(rows[-1])
-        for scope_id, source_id in candidates:
-            ref = SourceRef(source_type="content", source_id=source_id)
-            identity = await committed_identity_lookup(scope_id, source_id)
-            async with database.transaction() as connection:
-                # Serialize concurrent startup migrations using the existing
-                # journal lock, without allocating a position or queueing work.
-                await _lock_journal_head(connection, scope_id)
-                current = await sources.get(connection, scope_id, ref)
-                target = current.value
-                if identity is not None:
-                    target = target.model_copy(update={"handoff_receipt": True})
-                await sources.add_with_status(connection, scope_id, target)
-                await connection.execute(
-                    delete(RECEIPT_MIGRATION_REVIEW_TABLE).where(
-                        RECEIPT_MIGRATION_REVIEW_TABLE.c.scope_id == scope_id,
-                        RECEIPT_MIGRATION_REVIEW_TABLE.c.source_id == source_id,
+            if cleanup:
+                async with database.transaction() as connection:
+                    await connection.execute(
+                        delete(RECEIPT_MIGRATION_REVIEW_TABLE).where(
+                            tuple_(
+                                RECEIPT_MIGRATION_REVIEW_TABLE.c.scope_id, RECEIPT_MIGRATION_REVIEW_TABLE.c.source_id
+                            ).in_(cleanup)
+                        )
                     )
-                )
-                if identity is None and not getattr(target, "handoff_receipt", False):
+            after_source_id = rows[-1]
+            for scope_id, source_id in candidates:
+                ref = SourceRef(source_type="content", source_id=source_id)
+                identity = await committed_identity_lookup(scope_id, source_id)
+                async with database.transaction() as connection:
+                    # Serialize concurrent startup migrations using the existing
+                    # journal lock, without allocating a position or queueing work.
+                    await _lock_journal_head(connection, scope_id)
+                    current = await sources.get(connection, scope_id, ref)
+                    target = current.value
+                    if identity is not None:
+                        target = target.model_copy(update={"handoff_receipt": True})
+                    await sources.add_with_status(connection, scope_id, target)
                     await connection.execute(
-                        insert(RECEIPT_MIGRATION_REVIEW_TABLE).values(
-                            scope_id=scope_id,
-                            source_id=source_id,
-                            reason="missing_committed_receipt",
+                        delete(RECEIPT_MIGRATION_REVIEW_TABLE).where(
+                            RECEIPT_MIGRATION_REVIEW_TABLE.c.scope_id == scope_id,
+                            RECEIPT_MIGRATION_REVIEW_TABLE.c.source_id == source_id,
                         )
                     )
-                    unresolved += 1
-                else:
-                    attested += 1
+                    if identity is None and not getattr(target, "handoff_receipt", False):
+                        await connection.execute(
+                            insert(RECEIPT_MIGRATION_REVIEW_TABLE).values(
+                                scope_id=scope_id,
+                                source_id=source_id,
+                                reason="missing_committed_receipt",
+                            )
+                        )
+                        unresolved += 1
+                    else:
+                        attested += 1
+        after_scope = current_scope
+    return attested, unresolved
+
+
+async def _next_content_scope(database: AsyncDatabase, after: str, /) -> str | None:
+    """Return the next Scope holding a content Source, in primary-key order."""
+
+    async with database.transaction() as connection:
+        return await connection.scalar(
+            select(SOURCES_TABLE.c.scope_id)
+            .where(SOURCES_TABLE.c.scope_id > after, SOURCES_TABLE.c.source_type == "content")
+            .order_by(SOURCES_TABLE.c.scope_id)
+            .limit(1)
+        )
+
+
+async def _page_content_sources(
+    database: AsyncDatabase, scope_id: str, after: str, batch_size: int, /
+) -> tuple[str, ...]:
+    """Return the next page of content Source identities inside one Scope."""
+
+    async with database.transaction() as connection:
+        return tuple(
+            (
+                await connection.execute(
+                    select(SOURCES_TABLE.c.source_id)
+                    .where(
+                        SOURCES_TABLE.c.scope_id == scope_id,
+                        SOURCES_TABLE.c.source_type == "content",
+                        SOURCES_TABLE.c.source_id > after,
+                    )
+                    .order_by(SOURCES_TABLE.c.source_id)
+                    .limit(batch_size)
+                )
+            ).scalars()
+        )
 
 
 def _is_receipt_candidate(source: ContentSource) -> bool:

tests/test_integration_shared_files.py

@@ -0,0 +1,45 @@
+# 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.
+
+"""Shared plugin sources must stay identical across hosts until they move to one package.
+
+Several hosts ship byte-identical copies of the client configuration and the prepared-context
+helper. Nothing detected drift, and one copy had already diverged.
+"""
+
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).parents[1]
+PLUGINS = "plugins/powercontext"
+SHARED_SOURCES = ("powercontext_client_config.py", "hooks/prepared_context.py")
+
+
+def _copies(name: str) -> list[Path]:
+    return sorted(ROOT.glob(f"integrations/*/{PLUGINS}/{name}")) + sorted(
+        ROOT.glob(f"integrations/*/{PLUGINS}/**/{Path(name).name}")
+    )
+
+
+@pytest.mark.parametrize("name", SHARED_SOURCES)
+def test_shared_plugin_source_is_identical_in_every_host(name: str) -> None:
+    copies = sorted(set(_copies(name)))
+    assert len(copies) > 1, f"{name} is no longer shared; drop it from SHARED_SOURCES"
+
+    expected = copies[0].read_bytes().replace(b"\r\n", b"\n")
+    drifted = [
+        str(copy.relative_to(ROOT)) for copy in copies[1:] if copy.read_bytes().replace(b"\r\n", b"\n") != expected
+    ]
+    assert not drifted, f"{name} differs from {copies[0].relative_to(ROOT)} in: {', '.join(drifted)}"