perf(experience): read the incubation scope-head snapshot once per window

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,14 +128,16 @@ 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))):
             return ()
         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)

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