From 5de1bd680e05e0e01662efddafb8e1c3f48da484 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Wed, 21 Oct 2026 12:57:46 +0800 Subject: [PATCH 1/2] perf(sources): page the receipt migration inside one Scope The migration cursor spanned Scopes, so SQLite could not steer the (scope_id, source_type, source_id) primary key: every page used a temporary B-tree for the ORDER BY and rescanned the rows already visited, costing O(N^2) index reads per Scope at startup. Paging inside one Scope turns both statements into bounded seeks. --- .../builtin/persistence/receipt_migration.py | 178 +++++++++++------- 1 file changed, 106 insertions(+), 72 deletions(-) diff --git a/src/powercontext/builtin/persistence/receipt_migration.py b/src/powercontext/builtin/persistence/receipt_migration.py index 256c244267..e1d21d71cb 100644 --- a/src/powercontext/builtin/persistence/receipt_migration.py +++ b/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: From c31534b94d6b17da233400c9d91367bb99bf1a17 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Wed, 21 Oct 2026 12:57:46 +0800 Subject: [PATCH 2/2] test(integrations): fail on drift between shared plugin sources Four hosts ship the same client configuration and three ship the same prepared-context helper, with nothing detecting divergence. The WorkBuddy copy had already fallen behind on the bounded recall query, so this aligns it and adds a guard. --- .../powercontext/hooks/prepared_context.py | 28 +++++++++++- tests/test_integration_shared_files.py | 45 +++++++++++++++++++ 2 files changed, 72 insertions(+), 1 deletion(-) create mode 100644 tests/test_integration_shared_files.py diff --git a/integrations/workbuddy/plugins/powercontext/hooks/prepared_context.py b/integrations/workbuddy/plugins/powercontext/hooks/prepared_context.py index 31fd0852ac..3fbdf5d86f 100644 --- a/integrations/workbuddy/plugins/powercontext/hooks/prepared_context.py +++ b/integrations/workbuddy/plugins/powercontext/hooks/prepared_context.py @@ -12,7 +12,7 @@ # 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 @@ -20,6 +20,11 @@ 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() diff --git a/tests/test_integration_shared_files.py b/tests/test_integration_shared_files.py new file mode 100644 index 0000000000..b67c1f4b63 --- /dev/null +++ b/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)}"