refactor(regeneration): inject selectors so each owns its kind-specific logic

backend/infrahub/core/branch/tasks.py

@@ -33,7 +33,10 @@
 )
 from infrahub.core.merge.regeneration_dispatcher import PostMergeRegenerationDispatcher, submit_full_regeneration
 from infrahub.core.merge.schema_analyzer import MergeSchemaAnalyzer
-from infrahub.core.merge.selective_regen.generator_diff_capturer import GeneratorTrackingGroupDiffCapturer
+from infrahub.core.merge.selective_regen.generator_output import (
+    GeneratorCascadeOutput,
+    GeneratorTrackingGroupDiffCapturer,
+)
 from infrahub.core.merge.selective_regen.orchestrator import build_merge_selective_regeneration
 from infrahub.core.merge.write_blocker import MergeWriteBlocker
 from infrahub.core.migrations.exceptions import MigrationFailureError
@@ -494,19 +497,20 @@ async def _build_post_merge_regeneration_dispatcher(
     component_registry = get_component_registry()
     diff_coordinator = await component_registry.get_component(DiffCoordinator, db=db, branch=branch)
     diff_repository = await component_registry.get_component(DiffRepository, db=db, branch=branch)
+    output_capturer = GeneratorTrackingGroupDiffCapturer(
+        diff_coordinator=diff_coordinator,
+        diff_repository=diff_repository,
+        serializer=DiffSummarySerializer(),
+        client=get_client(),
+        branch=branch,
+    )
+    generator_output = GeneratorCascadeOutput(capturer=output_capturer)
     return PostMergeRegenerationDispatcher(
         workflow=get_workflow(),
-        selector=build_merge_selective_regeneration(client=get_client(), log=log),
+        planner=build_merge_selective_regeneration(client=get_client(), log=log, generator_output=generator_output),
         summary_cache=DiffSummaryCache(
             cache=await get_cache(), serializer=DiffSummarySerializer(), key_namespace="branch_merge"
         ),
-        generator_diff_capturer=GeneratorTrackingGroupDiffCapturer(
-            diff_coordinator=diff_coordinator,
-            diff_repository=diff_repository,
-            serializer=DiffSummarySerializer(),
-            client=get_client(),
-            branch=branch,
-        ),
         log=log,
     )
 

backend/infrahub/core/merge/regeneration_dispatcher.py

@@ -5,27 +5,26 @@
 from typing import TYPE_CHECKING
 
 from infrahub import config
+from infrahub.core.merge.selective_regen.models import CascadeRole
 from infrahub.core.timestamp import Timestamp
 from infrahub.exceptions import ResourceNotFoundError
 from infrahub.generators.constants import GeneratorDefinitionRunSource
 from infrahub.workflows.catalogue import (
-    REQUEST_ARTIFACT_DEFINITION_GENERATE,
-    REQUEST_GENERATOR_DEFINITION_RUN,
     TRIGGER_ARTIFACT_DEFINITION_GENERATE,
     TRIGGER_GENERATOR_DEFINITION_RUN,
 )
 
 if TYPE_CHECKING:
     from logging import Logger, LoggerAdapter
 
+    from infrahub_sdk.diff import NodeDiff
+
     from infrahub.context import InfrahubContext
     from infrahub.core.diff.summary_cache import DiffSummaryCache
-    from infrahub.git.models import RequestArtifactDefinitionGenerate
     from infrahub.services.adapters.workflow import InfrahubWorkflow
 
-    from .selective_regen.generator_diff_capturer import GeneratorMutationDiffCapturer
-    from .selective_regen.models import SelectiveRegenerationPlan
-    from .selective_regen.orchestrator import RegenerationSelector
+    from .selective_regen.models import PlannedRegeneration, SelectiveRegenerationPlan
+    from .selective_regen.orchestrator import RegenerationPlanner
 
 
 class FullRegenerationReason(StrEnum):
@@ -49,26 +48,6 @@ async def submit_full_regeneration(*, workflow: InfrahubWorkflow, context: Infra
     )
 
 
-def _consolidate_artifact_generates(
-    artifact_generates: list[RequestArtifactDefinitionGenerate],
-) -> list[RequestArtifactDefinitionGenerate]:
-    """Merge requests for the same artifact definition into one, unioning their member/limit filters.
-
-    An artifact selected from both the merge diff and a generator's output would otherwise be dispatched
-    twice; an empty filter means "all members", so it subsumes any specific filter.
-    """
-    consolidated: dict[str, RequestArtifactDefinitionGenerate] = {}
-    for request in artifact_generates:
-        merged = consolidated.get(request.artifact_definition_id)
-        if merged is None:
-            consolidated[request.artifact_definition_id] = request
-            continue
-        members = [] if not merged.members or not request.members else sorted({*merged.members, *request.members})
-        limit = [] if not merged.limit or not request.limit else sorted({*merged.limit, *request.limit})
-        consolidated[request.artifact_definition_id] = merged.model_copy(update={"members": members, "limit": limit})
-    return list(consolidated.values())
-
-
 class PostMergeRegenerationDispatcher:
     """Decide and submit which generators and artifacts a committed merge should regenerate.
 
@@ -81,15 +60,13 @@ class PostMergeRegenerationDispatcher:
     def __init__(
         self,
         workflow: InfrahubWorkflow,
-        selector: RegenerationSelector,
+        planner: RegenerationPlanner,
         summary_cache: DiffSummaryCache,
-        generator_diff_capturer: GeneratorMutationDiffCapturer,
         log: Logger | LoggerAdapter[Logger],
     ) -> None:
         self.workflow = workflow
-        self.selector = selector
+        self.planner = planner
         self.summary_cache = summary_cache
-        self.generator_diff_capturer = generator_diff_capturer
         self.log = log
 
     async def dispatch(
@@ -126,7 +103,7 @@ async def dispatch(
         # leaving the merge under-regenerated. A single generator run failing is handled granularly in
         # _dispatch_plan and does not reach here.
         try:
-            plan = await self.selector.build_plan(diff_summary=diff_summary, target_branch=target_branch)
+            plan = await self.planner.build_plan(diff_summary=diff_summary, target_branch=target_branch)
             await self._dispatch_plan(context=context, target_branch=target_branch, plan=plan)
         except Exception:
             self.log.exception("Selective post-merge regeneration failed; falling back to full regeneration")
@@ -141,80 +118,85 @@ async def _dispatch_plan(
         target_branch: str,
         plan: SelectiveRegenerationPlan,
     ) -> None:
-        generator_cascade = bool(plan.generator_runs)
+        sources = plan.for_role(CascadeRole.SOURCE)
+        terminals = plan.for_role(CascadeRole.TERMINAL)
+        source_runs = [request for entry in sources for request in entry.requests]
+        generator_cascade = bool(source_runs)
         cascade_started_at = Timestamp() if generator_cascade else None
 
         self.log.debug(
-            f"Selective post-merge execution: {len(plan.generator_runs)} generator run(s), "
-            f"{len(plan.artifact_generates)} artifact generation(s)"
+            f"Selective post-merge execution: {len(source_runs)} cascade-source run(s), "
+            f"{sum(len(entry.requests) for entry in terminals)} terminal generation(s)"
             + ("; generator cascade engaged" if generator_cascade else "")
         )
 
         if cascade_started_at is None:
-            await self._submit_artifacts(context=context, artifact_generates=plan.artifact_generates)
+            await self._submit(context=context, entries=terminals)
             return
 
         generator_failed = False
-        for generator_run in plan.generator_runs:
-            try:
-                # Await each generator so its writes have landed before they are captured.
-                await self.workflow.execute_workflow(
-                    workflow=REQUEST_GENERATOR_DEFINITION_RUN, context=context, parameters={"model": generator_run}
-                )
-            except Exception:
-                generator_failed = True
-                self.log.exception("Post-merge generator run failed")
+        for entry in sources:
+            for run in entry.requests:
+                try:
+                    # Await each generator so its writes have landed before they are captured.
+                    await self.workflow.execute_workflow(
+                        workflow=entry.workflow, context=context, parameters={"model": run}
+                    )
+                except Exception:
+                    generator_failed = True
+                    self.log.exception("Post-merge generator run failed")
 
         if generator_failed:
-            # A failed generator's consuming artifacts cannot be selected from its output, so regenerate
-            # every artifact -- but never re-run the generators, which would fail the same way again.
-            await self._submit_full_artifact_regeneration(context=context, target_branch=target_branch)
+            # A failed source's consuming terminals cannot be selected from its output, so regenerate
+            # every terminal -- but never re-run the sources, which would fail the same way again.
+            await self._submit_full_terminal_regeneration(context=context, target_branch=target_branch)
             return
 
-        targeted = await self._artifacts_from_generator_output(
-            context=context,
-            target_branch=target_branch,
-            since=cascade_started_at,
-            generator_definition_names=[run.generator_definition.definition_name for run in plan.generator_runs],
+        targeted = await self._reselect_from_cascade_output(
+            context=context, target_branch=target_branch, sources=sources, since=cascade_started_at
         )
         if targeted is None:
-            # Every artifact was already regenerated wholesale, which covers the merge-diff selection too.
+            # Every terminal was already regenerated wholesale, which covers the merge-diff selection too.
             return
-        # Dispatched only after the capture, so the capture window never sees these artifact generations'
-        # own writes.
-        await self._submit_artifacts(context=context, artifact_generates=[*plan.artifact_generates, *targeted])
-
-    async def _submit_artifacts(
-        self, *, context: InfrahubContext, artifact_generates: list[RequestArtifactDefinitionGenerate]
-    ) -> None:
-        for artifact_generate in _consolidate_artifact_generates(artifact_generates):
-            await self.workflow.submit_workflow(
-                workflow=REQUEST_ARTIFACT_DEFINITION_GENERATE, context=context, parameters={"model": artifact_generate}
-            )
+        # Dispatched only after the capture, so the capture window never sees these generations' own writes.
+        await self._submit(context=context, entries=[*terminals, *targeted])
+
+    async def _submit(self, *, context: InfrahubContext, entries: list[PlannedRegeneration]) -> None:
+        """Submit each fire-and-forget request, letting the owning selector consolidate its own kind."""
+        for entry in self.planner.consolidate_submissions(entries):
+            for request in entry.requests:
+                await self.workflow.submit_workflow(
+                    workflow=entry.workflow, context=context, parameters={"model": request}
+                )
 
-    async def _artifacts_from_generator_output(
+    async def _reselect_from_cascade_output(
         self,
         *,
         context: InfrahubContext,
         target_branch: str,
+        sources: list[PlannedRegeneration],
         since: Timestamp,
-        generator_definition_names: list[str],
-    ) -> list[RequestArtifactDefinitionGenerate] | None:
-        """Select the artifacts the just-run generators' writes require be regenerated.
+    ) -> list[PlannedRegeneration] | None:
+        """Reselect the fire-and-forget generations the just-run sources' own output requires.
 
-        Returns ``None`` after regenerating every artifact wholesale when the generator output cannot be
-        captured or selected, so a generator's writes can never leave a consuming artifact stale.
+        Each source captures its own output; the terminals that read it are then reselected from the
+        combined diff. Returns ``None`` after regenerating every terminal wholesale when that output
+        cannot be captured or selected, so a source's writes can never leave a consuming terminal stale.
         """
         try:
-            generator_diff = await self.generator_diff_capturer.capture(
-                since=since, generator_definition_names=generator_definition_names
+            captured: list[NodeDiff] = []
+            for entry in sources:
+                if entry.output is not None:
+                    captured.extend(await entry.output.capture(since=since, requests=entry.requests))
+            targeted = await self.planner.reselect_from_cascade_output(
+                diff_summary=captured, target_branch=target_branch
             )
-            targeted = await self.selector.select_artifacts(diff_summary=generator_diff, target_branch=target_branch)
         except Exception:
-            self.log.exception("Failed to target artifacts from generator output; regenerating all artifacts instead")
-            await self._submit_full_artifact_regeneration(context=context, target_branch=target_branch)
+            self.log.exception("Failed to target terminals from cascade output; regenerating all terminals instead")
+            await self._submit_full_terminal_regeneration(context=context, target_branch=target_branch)
             return None
-        self.log.debug(f"Targeted {len(targeted)} artifact definition(s) from generator output")
+        targeted_count = sum(len(entry.requests) for entry in targeted)
+        self.log.debug(f"Targeted {targeted_count} terminal definition(s) from cascade output")
         return targeted
 
     async def _full_regeneration(
@@ -223,7 +205,8 @@ async def _full_regeneration(
         self.log.debug(f"{reason}; regenerating all definitions")
         await submit_full_regeneration(workflow=self.workflow, context=context, target_branch=target_branch)
 
-    async def _submit_full_artifact_regeneration(self, context: InfrahubContext, target_branch: str) -> None:
-        await self.workflow.submit_workflow(
-            workflow=TRIGGER_ARTIFACT_DEFINITION_GENERATE, context=context, parameters={"branch": target_branch}
-        )
+    async def _submit_full_terminal_regeneration(self, context: InfrahubContext, target_branch: str) -> None:
+        for regeneration in self.planner.terminal_full_regenerations(target_branch):
+            await self.workflow.submit_workflow(
+                workflow=regeneration.workflow, context=context, parameters=regeneration.parameters
+            )

backend/infrahub/core/merge/selective_regen/definition_selector/artifact_selector.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from infrahub_sdk.protocols import CoreArtifactDefinition
 
 from infrahub.core.constants import InfrahubKind
@@ -8,15 +10,42 @@
 from infrahub.git.models import RequestArtifactDefinitionGenerate
 from infrahub.git.utils import fetch_artifact_definition_targets
 from infrahub.message_bus.types import ProposedChangeArtifactDefinition
+from infrahub.workflows.catalogue import REQUEST_ARTIFACT_DEFINITION_GENERATE, TRIGGER_ARTIFACT_DEFINITION_GENERATE
 
 from ..models import LoadedDefinition
 from .base import DefinitionSelectorBase
 
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
 
 class ArtifactSelector(DefinitionSelectorBase[ProposedChangeArtifactDefinition, RequestArtifactDefinitionGenerate]):
     """Selects the artifact definitions a merge changed, narrowed to the members it affects."""
 
     subscriber_kind = InfrahubKind.ARTIFACT
+    workflow = REQUEST_ARTIFACT_DEFINITION_GENERATE
+    full_regeneration_workflow = TRIGGER_ARTIFACT_DEFINITION_GENERATE
+
+    def consolidate(
+        self, requests: Sequence[RequestArtifactDefinitionGenerate]
+    ) -> list[RequestArtifactDefinitionGenerate]:
+        """Merge requests for the same artifact definition, unioning their member/limit filters.
+
+        An artifact selected from both the merge diff and a generator's output would otherwise be
+        dispatched twice; an empty filter means "all members", so it subsumes any specific filter.
+        """
+        consolidated: dict[str, RequestArtifactDefinitionGenerate] = {}
+        for request in requests:
+            merged = consolidated.get(request.artifact_definition_id)
+            if merged is None:
+                consolidated[request.artifact_definition_id] = request
+                continue
+            members = [] if not merged.members or not request.members else sorted({*merged.members, *request.members})
+            limit = [] if not merged.limit or not request.limit else sorted({*merged.limit, *request.limit})
+            consolidated[request.artifact_definition_id] = merged.model_copy(
+                update={"members": members, "limit": limit}
+            )
+        return list(consolidated.values())
 
     async def load_definitions(self, *, target_branch: str) -> list[LoadedDefinition[ProposedChangeArtifactDefinition]]:
         definition_information = await self.client.execute_graphql(

backend/infrahub/core/merge/selective_regen/definition_selector/base.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from infrahub.core.regeneration.members import map_subscriber_ids_by_member
 
@@ -10,11 +10,13 @@
 
 if TYPE_CHECKING:
     import logging
+    from collections.abc import Sequence
 
     from infrahub_sdk.client import InfrahubClient
     from infrahub_sdk.diff import NodeDiff
 
     from infrahub.core.regeneration.models import RegenerationTrigger
+    from infrahub.workflows.models import WorkflowDefinition
 
     from ..gate import DefinitionGate
     from ..impacted import ImpactedSubscriberResolver
@@ -42,6 +44,10 @@ class DefinitionSelectorBase[DefinitionT: DefinitionModel, RequestT](ABC):
     """
 
     subscriber_kind: str
+    workflow: WorkflowDefinition
+    """The workflow that runs this selector's requests."""
+    full_regeneration_workflow: WorkflowDefinition
+    """The workflow that regenerates every definition of this kind, ignoring selective narrowing."""
 
     def __init__(
         self,
@@ -132,6 +138,19 @@ async def select(
             requests.append(self._build_request(definition=definition, target_branch=target_branch, members=members))
         return requests
 
+    def full_regeneration_parameters(self, *, target_branch: str) -> dict[str, Any]:
+        """Parameters for the blanket regeneration of this kind; extended by kinds that need more."""
+        return {"branch": target_branch}
+
+    def consolidate(self, requests: Sequence[RequestT]) -> Sequence[RequestT]:
+        """Combine this selector's requests before dispatch; by default each is dispatched as-is.
+
+        Overridden where several requests can target the same definition (an artifact selected by both
+        the merge diff and a generator's output), so the follow-up submits one request per definition
+        without knowing how to merge them.
+        """
+        return requests
+
     async def _map_subscribers_by_member(self, *, definition: DefinitionT, target_branch: str) -> dict[str, str]:
         """Map each member to the id of its existing subscriber for this definition on the branch.