fix: correctly apply change profiles filter at pipeline

build.gradle.kts

@@ -19,7 +19,7 @@ plugins {
 
 allprojects {
     group = "io.flamingock"
-    val declaredVersion = "1.4.3-SNAPSHOT"
+    val declaredVersion = "1.4.4-SNAPSHOT"
     version = VersionManager.resolveVersion(declaredVersion, project.hasProperty("release"))
 
     extra["generalUtilVersion"] = "1.5.3"

core/flamingock-core/src/main/java/io/flamingock/internal/core/pipeline/loaded/LoadedPipeline.java

@@ -108,6 +108,16 @@ public List<AbstractLoadedStage> getStages() {
         return loadedStages;
     }
 
+    /**
+     * Returns the change filters contributed by plugins and applied at runtime construction
+     * time. A change is included in the runtime pipeline only if every filter returns
+     * {@code true} for it; any filter returning {@code false} excludes the change. May be
+     * empty when no plugin contributed a filter.
+     */
+    public Collection<ChangeFilter> getChangeFilters() {
+        return changeFilters == null ? Collections.emptyList() : changeFilters;
+    }
+
     @Override
     public Optional<AbstractLoadedChange> getLoadedChange(String changeId) {
         return loadedStages.stream()

core/flamingock-core/src/main/java/io/flamingock/internal/core/pipeline/loaded/stage/AbstractLoadedStage.java

@@ -103,6 +103,31 @@ public Collection<AbstractLoadedChange> getChanges() {
         return changes;
     }
 
+    /**
+     * Returns a new instance of the same concrete stage type carrying the provided changes
+     * instead of this stage's current changes. Name, type, and the per-subclass validation
+     * context are preserved.
+     *
+     * <p>Used at runtime construction time to materialize a stage with a filtered subset
+     * of changes without mutating the original {@code AbstractLoadedStage} (whose
+     * {@code changes} collection is final and immutable post-construction). The copy is
+     * produced by reflecting on the concrete subclass to find its public
+     * {@code (String, StageType, Collection)} constructor — every concrete subclass
+     * ({@link DefaultLoadedStage}, {@link LegacyLoadedStage}, {@link SystemLoadedStage})
+     * exposes one.
+     */
+    public AbstractLoadedStage withChanges(Collection<AbstractLoadedChange> newChanges) {
+        try {
+            return getClass()
+                    .getConstructor(String.class, StageType.class, Collection.class)
+                    .newInstance(getName(), getType(), newChanges);
+        } catch (ReflectiveOperationException e) {
+            throw new IllegalStateException(
+                    "Cannot copy stage [" + getName() + "] of type " + getClass().getName()
+                            + " with filtered changes", e);
+        }
+    }
+
 
 
     /**

core/flamingock-core/src/main/java/io/flamingock/internal/core/pipeline/run/PipelineRun.java

@@ -25,6 +25,8 @@
 import io.flamingock.internal.common.core.response.data.PlannerVerdict;
 import io.flamingock.internal.common.core.response.data.StageResult;
 import io.flamingock.internal.common.core.response.data.StageState;
+import io.flamingock.internal.core.change.filter.ChangeFilter;
+import io.flamingock.internal.core.change.loaded.AbstractLoadedChange;
 import io.flamingock.internal.core.pipeline.execution.StageExecutionException;
 import io.flamingock.internal.core.pipeline.loaded.LoadedPipeline;
 import io.flamingock.internal.core.pipeline.loaded.stage.AbstractLoadedStage;
@@ -33,6 +35,7 @@
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
@@ -52,14 +55,70 @@ public class PipelineRun {
     public static PipelineRun of(LoadedPipeline pipeline) {
         // Static-structure validation runs as part of construction. Builder-time validation in
         // AbstractChangeRunnerBuilder still fails fast on malformed pipelines; this call covers
-        // the runtime entry point so callers don't have to validate separately.
+        // the runtime entry point so callers don't have to validate separately. Validation
+        // operates on the unfiltered pipeline so structural checks (duplicate IDs, empty
+        // stages, etc.) catch issues regardless of which changes any plugin would exclude.
         pipeline.validate();
+        Collection<ChangeFilter> filters = pipeline.getChangeFilters();
         List<AbstractLoadedStage> stages = new ArrayList<>();
-        pipeline.getSystemStage().ifPresent(stages::add);
-        stages.addAll(pipeline.getStages());
+        // Apply the same drop-empty rule to the system stage as to user stages. A stage is
+        // dropped only when filtering actually removed every change from a stage that had
+        // some to begin with. A stage that was already empty before filtering is preserved
+        // unchanged; this matches the pre-existing "empty stages survive" contract used by
+        // builder-mocked tests and by incremental compile-time flows where a stage may be
+        // temporarily empty before any @Change has been added to its package.
+        pipeline.getSystemStage()
+                .map(stage -> applyFilters(stage, filters))
+                .filter(result -> !result.filteredToEmpty)
+                .map(result -> result.stage)
+                .ifPresent(stages::add);
+        for (AbstractLoadedStage stage : pipeline.getStages()) {
+            FilterResult result = applyFilters(stage, filters);
+            if (!result.filteredToEmpty) {
+                stages.add(result.stage);
+            }
+        }
         return of(stages);
     }
 
+    /**
+     * Result of running change filters against a single stage. {@code stage} is either the
+     * original stage (no filter removed any change, OR the original was empty, OR no filters
+     * were provided) or a new instance with the surviving changes. {@code filteredToEmpty}
+     * is {@code true} only when the filter actually removed every change from a non-empty
+     * stage — that's the signal the caller uses to drop the stage from the runtime list.
+     */
+    private static final class FilterResult {
+        final AbstractLoadedStage stage;
+        final boolean filteredToEmpty;
+
+        FilterResult(AbstractLoadedStage stage, boolean filteredToEmpty) {
+            this.stage = stage;
+            this.filteredToEmpty = filteredToEmpty;
+        }
+    }
+
+    /**
+     * Applies the given change filters to a stage's changes, returning a {@link FilterResult}
+     * that preserves the original stage reference when no filter removed any change. Filters
+     * are AND-composed: a change is kept only if every filter returns {@code true} for it.
+     */
+    private static FilterResult applyFilters(AbstractLoadedStage stage, Collection<ChangeFilter> filters) {
+        Collection<AbstractLoadedChange> originalChanges = stage.getChanges();
+        if (filters == null || filters.isEmpty() || originalChanges == null || originalChanges.isEmpty()) {
+            // No filtering meaningfully applied. Preserve the original stage unchanged — this
+            // includes the case of a pre-existing empty stage (filteredToEmpty=false).
+            return new FilterResult(stage, false);
+        }
+        Collection<AbstractLoadedChange> surviving = originalChanges.stream()
+                .filter(change -> filters.stream().allMatch(f -> f.filter(change)))
+                .collect(Collectors.toList());
+        if (surviving.size() == originalChanges.size()) {
+            return new FilterResult(stage, false);
+        }
+        return new FilterResult(stage.withChanges(surviving), surviving.isEmpty());
+    }
+
     public static PipelineRun of(List<AbstractLoadedStage> stages) {
         // Partition by StageType in dependency order, skipping empty types (sparse). The resulting
         // flat list is the concatenation of blocks in canonical order — a single source of truth