From 8d87321edd346d5dd83297e9fbc04330d44602a6 Mon Sep 17 00:00:00 2001 From: Inference1 <68734681+Inference1@users.noreply.github.com> Date: Sat, 24 Oct 2026 19:50:03 +0800 Subject: [PATCH 1/2] perf(experience): read the incubation scope-head snapshot once per window An outcome without a handoff receipt matches against every active Experience, and every such row reloaded that snapshot with one get() per Experience. Load it once per window and batch the revision reads with get_many; the window writes no artifacts, so each row sees the same heads. Measured with 40 Experiences and four outcome rows: 355 -> 34 statements. --- .../builtin/runtime/recurrence.py | 32 ++++++++---- tests/e2e/test_recurring_failure_repair.py | 50 ++++++++++++++++++- 2 files changed, 70 insertions(+), 12 deletions(-) diff --git a/src/powercontext/builtin/runtime/recurrence.py b/src/powercontext/builtin/runtime/recurrence.py index 97339533b6..75850fd373 100644 --- a/src/powercontext/builtin/runtime/recurrence.py +++ b/src/powercontext/builtin/runtime/recurrence.py @@ -112,12 +112,14 @@ async def record_window( ) -> tuple[RecurrenceRevisionProposal, ...]: """Record every event one incubation window proves, in evidence order.""" + outcomes = tuple((row, outcome) for row in rows if (outcome := _task_outcome(row)) is not None) + # Every unlinked outcome matches against the same scope-head snapshot, and + # this method writes no artifacts, so read the snapshot once per window. + needs_heads = any(outcome.handoff_receipt_ref is None for _, outcome in outcomes) + heads: _Contents = await self._experience_heads(connection) if needs_heads else () proposals: list[RecurrenceRevisionProposal] = [] - for row in rows: - outcome = _task_outcome(row) - if outcome is None: - continue - proposals.extend(await self._record_outcome(connection, row, outcome)) + for row, outcome in outcomes: + proposals.extend(await self._record_outcome(connection, row, outcome, heads=heads)) return tuple(proposals) async def _record_outcome( @@ -126,6 +128,8 @@ async def _record_outcome( row: StoredSource, outcome: TaskOutcome, /, + *, + heads: _Contents, ) -> tuple[RecurrenceRevisionProposal, ...]: link = await self._resolve_link(connection, outcome) if not isinstance(link, (_LinkedHandoff, type(None))): @@ -133,7 +137,7 @@ async def _record_outcome( contents: _Contents = () if link is None else link.contents if link is not None: await self._record_selected(connection, row=row, link=link, contents=contents) - candidates: _Contents = contents if link is not None else await self._experience_heads(connection) + candidates: _Contents = contents if link is not None else heads matches = await self._freeze_matches( connection, row=row, @@ -477,12 +481,18 @@ async def _experience_heads(self, connection: AsyncConnection, /) -> _Contents: async def _contents(self, connection: AsyncConnection, refs: tuple[ArtifactRef, ...], /) -> _Contents: if not refs: return () + try: + artifacts = await self.artifacts.get_many(connection, self.scope_id, refs) + except RepositoryNotFoundError: + # get_many is all-or-nothing; skip missing revisions one by one as before. + artifacts = [] + for ref in refs: + try: + artifacts.append(await self.artifacts.get(connection, self.scope_id, ref)) + except RepositoryNotFoundError: + continue contents: list[tuple[ArtifactRef, ExperienceContent]] = [] - for ref in refs: - try: - artifact = await self.artifacts.get(connection, self.scope_id, ref) - except RepositoryNotFoundError: - continue + for artifact in artifacts: if isinstance(artifact.content, ExperienceContent): contents.append((artifact.as_ref(), artifact.content)) return tuple(contents) diff --git a/tests/e2e/test_recurring_failure_repair.py b/tests/e2e/test_recurring_failure_repair.py index 644054ef1e..f354344b13 100644 --- a/tests/e2e/test_recurring_failure_repair.py +++ b/tests/e2e/test_recurring_failure_repair.py @@ -23,7 +23,8 @@ import asyncio -from sqlalchemy import insert +from sqlalchemy import event, insert +from sqlalchemy.engine import Engine from sqlalchemy.ext.asyncio import AsyncConnection from powercontext.artifacts import ArtifactRef @@ -330,3 +331,50 @@ async def scenario() -> None: assert "avoided" in events asyncio.run(scenario()) + + +# Measured with 40 active Experiences and one unlinked outcome; slack of one statement. +_INCUBATION_STATEMENT_CEILING = 14 + + +def test_incubation_statement_count_does_not_grow_with_scope_experiences() -> None: + """An unlinked outcome must not read the scope Experience heads one by one.""" + + async def counted(experiences: int) -> int: + statements: list[str] = [] + + def record(_conn, _cursor, statement, *_args) -> None: + statements.append(statement) + + async with SQLiteProfile.open(SQLiteConfig(), tables=TABLES) as profile: + sources = SourceRepository((CONTENT_SOURCE_ADAPTER,)) + artifacts = ArtifactRepository((Handoff, Experience), sources=sources) + ledger = _ledger(profile.database, sources, artifacts, RecurrenceRepository()) + async with profile.database.transaction() as connection: + await _seed(connection, sources, artifacts, "experience_content") + for index in range(2, experiences + 1): + await artifacts.create( + connection, + SCOPE, + f"experience-{index}", + artifacts.draft( + Experience.family, + _content(repair_surface="experience_content").model_dump(mode="json"), + ), + ) + await _add_failure(sources, connection, "failure-1") + event.listen(Engine, "before_cursor_execute", record) + try: + async with profile.database.transaction() as connection: + through = await sources.journal_position(connection, SCOPE) + rows = await sources.list_window(connection, SCOPE, after=0, through=through) + await ledger.record_window(connection, rows) + finally: + event.remove(Engine, "before_cursor_execute", record) + return len(statements) + + medium = asyncio.run(counted(10)) + large = asyncio.run(counted(40)) + + assert medium == large + assert large <= _INCUBATION_STATEMENT_CEILING From f430a5098c3627a8279c64dc987c3c1d2624bd05 Mon Sep 17 00:00:00 2001 From: Inference1 <68734681+Inference1@users.noreply.github.com> Date: Sun, 25 Oct 2026 13:32:21 +0800 Subject: [PATCH 2/2] chore: retrigger CI The code-seekdb (3.14) job timed out in the embedded SeekDB migration step; the same step also timed out on unrelated branches that day.