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