feat: make stage independent from each other (#908)

Author: dieppa

CLAUDE.md

@@ -204,6 +204,28 @@ This is a multi-module Gradle project using Kotlin DSL.
 - Test resources in `src/test/resources/flamingock/pipeline.yaml`
 - Each module has isolated test suite
 
+### Validation Policy (when changing code)
+
+Two tiers — use them in order:
+
+**Tier 1 — Incremental validation (during iteration):**
+- Run targeted module tests for fast feedback: `./gradlew :core:flamingock-core:test`, `./gradlew :legacy:mongock-importer-mongodb:test`, etc.
+- Use `--tests "fully.qualified.ClassName"` to narrow further when iterating on a specific test.
+- Cheap, quick, good for confirming a focused change compiles and the obvious cases pass.
+
+**Tier 2 — Final validation (before declaring a task done):**
+- Run `./gradlew clean build` from the repo root. This is the **authoritative, definitive check**.
+- It surfaces failures that targeted/incremental runs miss:
+  - Cross-module integration regressions.
+  - Stale Gradle / annotation-processor caches masking compile errors.
+  - License-header drift (`spotlessCheck` runs as part of `build`).
+  - Test-resource generation issues (annotation processors regenerating pipeline metadata).
+  - Module-dependency ordering bugs.
+- A passing per-module test run is **not sufficient** to claim a task complete. Per-module runs use cached artifacts and may not exercise the full graph.
+- Do not skip this step on the grounds that "the affected module passed in isolation." If you reach the end of a task without a clean full build, say so explicitly rather than implying success.
+
+If `clean build` is too slow to run on every minor iteration, that's expected — Tier 1 is for iteration. But the **final hand-off** must include a clean build.
+
 ### Java Version
 - Target Java 8 compatibility
 - Kotlin stdlib used in build scripts only

cloud/flamingock-cloud-api/src/main/java/io/flamingock/cloud/api/request/StageRequest.java

@@ -15,21 +15,34 @@
  */
 package io.flamingock.cloud.api.request;
 
+import io.flamingock.cloud.api.vo.CloudStageStatus;
+
 import java.util.List;
 
 public class StageRequest {
     private String name;
 
     private int order;
 
+    /**
+     * Per-stage status reported by the client. Nullable for back-compat with older clients;
+     * servers must treat {@code null} as {@link CloudStageStatus#NOT_STARTED}.
+     */
+    private CloudStageStatus status;
+
     private List<ChangeRequest> changes;
 
     public StageRequest() {
     }
 
     public StageRequest(String name, int order, List<ChangeRequest> changes) {
+        this(name, order, null, changes);
+    }
+
+    public StageRequest(String name, int order, CloudStageStatus status, List<ChangeRequest> changes) {
         this.name = name;
         this.order = order;
+        this.status = status;
         this.changes = changes;
     }
 
@@ -41,6 +54,10 @@ public int getOrder() {
         return order;
     }
 
+    public CloudStageStatus getStatus() {
+        return status;
+    }
+
     public List<ChangeRequest> getChanges() {
         return changes;
     }
@@ -53,6 +70,10 @@ public void setOrder(int order) {
         this.order = order;
     }
 
+    public void setStatus(CloudStageStatus status) {
+        this.status = status;
+    }
+
     public void setChanges(List<ChangeRequest> changes) {
         this.changes = changes;
     }
@@ -64,11 +85,12 @@ public boolean equals(Object o) {
         StageRequest that = (StageRequest) o;
         return order == that.order
                 && java.util.Objects.equals(name, that.name)
+                && status == that.status
                 && java.util.Objects.equals(changes, that.changes);
     }
 
     @Override
     public int hashCode() {
-        return java.util.Objects.hash(name, order, changes);
+        return java.util.Objects.hash(name, order, status, changes);
     }
 }

cloud/flamingock-cloud-api/src/main/java/io/flamingock/cloud/api/vo/CloudStageStatus.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2026 Flamingock (https://www.flamingock.io)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.flamingock.cloud.api.vo;
+
+/**
+ * Wire-level per-stage status sent from the client to the cloud planner.
+ *
+ * <p>Mirrors the internal {@code StageState} hierarchy in shape; the cloud server uses this to
+ * decide what to do with each stage on the next iteration (e.g., skip stages already failed
+ * or blocked, route MI cases). Client-side mapping lives in {@code CloudApiMapper.toCloud(StageState)}.
+ */
+public enum CloudStageStatus {
+    NOT_STARTED,
+    STARTED,
+    COMPLETED,
+    FAILED,
+    BLOCKED_MANUAL_INTERVENTION
+}

cloud/flamingock-cloud/src/main/java/io/flamingock/cloud/CloudApiMapper.java

@@ -17,10 +17,12 @@
 
 import io.flamingock.cloud.api.vo.CloudAuditStatus;
 import io.flamingock.cloud.api.vo.CloudChangeType;
+import io.flamingock.cloud.api.vo.CloudStageStatus;
 import io.flamingock.cloud.api.vo.CloudTargetSystemAuditMarkType;
 import io.flamingock.cloud.api.vo.CloudTxStrategy;
 import io.flamingock.internal.common.core.audit.AuditEntry;
 import io.flamingock.internal.common.core.audit.AuditTxType;
+import io.flamingock.internal.common.core.response.data.StageState;
 import io.flamingock.internal.common.core.targets.TargetSystemAuditMarkType;
 
 public final class CloudApiMapper {
@@ -44,4 +46,23 @@ public static CloudChangeType toCloud(AuditEntry.ChangeType changeType) {
         return CloudChangeType.valueOf(changeType.name());
     }
 
+    /**
+     * Maps the internal {@link StageState} hierarchy to the wire enum {@link CloudStageStatus}.
+     *
+     * <p>Returns {@code null} for {@code NOT_STARTED} (or a null state) — the canonical wire
+     * shape for "not started" is field absence/null, matching back-compat semantics with older
+     * clients that don't populate the field. The server treats {@code null} as {@code NOT_STARTED}.
+     *
+     * <p>Order is important: {@code BlockedForMI} extends {@code Failed}, so the
+     * blocked-for-MI check must come before the generic failed check.
+     */
+    public static CloudStageStatus toCloud(StageState state) {
+        if (state == null || state.isNotStarted()) return null;
+        if (state.isBlockedForManualIntervention()) return CloudStageStatus.BLOCKED_MANUAL_INTERVENTION;
+        if (state.isFailed()) return CloudStageStatus.FAILED;
+        if (state.isCompleted()) return CloudStageStatus.COMPLETED;
+        if (state.isStarted()) return CloudStageStatus.STARTED;
+        throw new IllegalStateException("Unknown StageState: " + state);
+    }
+
 }

cloud/flamingock-cloud/src/main/java/io/flamingock/cloud/planner/CloudExecutionPlanMapper.java

@@ -24,8 +24,11 @@
 import io.flamingock.cloud.api.response.StageResponse;
 import io.flamingock.cloud.api.response.ChangeResponse;
 import io.flamingock.cloud.api.vo.CloudChangeAction;
+import io.flamingock.cloud.api.vo.CloudStageStatus;
 import io.flamingock.cloud.api.vo.CloudTargetSystemAuditMarkType;
 import io.flamingock.cloud.CloudApiMapper;
+import io.flamingock.internal.core.pipeline.run.PipelineRun;
+import io.flamingock.internal.core.pipeline.run.StageRun;
 import io.flamingock.internal.common.core.targets.TargetSystemAuditMarkType;
 import io.flamingock.cloud.lock.CloudLockService;
 import io.flamingock.internal.core.configuration.core.CoreConfigurable;
@@ -51,19 +54,22 @@
 
 public final class CloudExecutionPlanMapper {
 
-    public static ExecutionPlanRequest toRequest(List<AbstractLoadedStage> loadedStages,
+    public static ExecutionPlanRequest toRequest(PipelineRun pipelineRun,
                                                  long lockAcquiredForMillis,
                                                  Map<String, TargetSystemAuditMarkType> ongoingStatusesMap) {
 
-        List<StageRequest> requestStages = new ArrayList<>(loadedStages.size());
-        for (int i = 0; i < loadedStages.size(); i++) {
-            AbstractLoadedStage currentStage = loadedStages.get(i);
+        List<StageRun> stageRuns = pipelineRun.getStageRuns();
+        List<StageRequest> requestStages = new ArrayList<>(stageRuns.size());
+        for (int i = 0; i < stageRuns.size(); i++) {
+            StageRun stageRun = stageRuns.get(i);
+            AbstractLoadedStage currentStage = stageRun.getLoadedStage();
             List<ChangeRequest> stageChanges = currentStage
                     .getChanges()
                     .stream()
                     .map(descriptor -> CloudExecutionPlanMapper.mapToChangeRequest(descriptor, ongoingStatusesMap))
                     .collect(Collectors.toList());
-            requestStages.add(new StageRequest(currentStage.getName(), i, stageChanges));
+            CloudStageStatus status = CloudApiMapper.toCloud(stageRun.getState());
+            requestStages.add(new StageRequest(currentStage.getName(), i, status, stageChanges));
         }
 
         return new ExecutionPlanRequest(lockAcquiredForMillis, requestStages);

cloud/flamingock-cloud/src/main/java/io/flamingock/cloud/planner/CloudExecutionPlanner.java

@@ -33,7 +33,6 @@
 import io.flamingock.internal.core.plan.ExecutionPlan;
 import io.flamingock.internal.core.plan.ExecutionPlanner;
 import io.flamingock.internal.core.external.store.lock.LockException;
-import io.flamingock.internal.core.pipeline.execution.ExecutableStage;
 import io.flamingock.internal.core.pipeline.loaded.stage.AbstractLoadedStage;
 import io.flamingock.internal.core.pipeline.run.PipelineRun;
 import io.flamingock.internal.util.log.FlamingockLoggerFactory;
@@ -94,7 +93,7 @@ public ExecutionPlan getNextExecution(PipelineRun pipelineRun) throws LockExcept
         do {
             try {
                 logger.info("Requesting cloud execution plan - elapsed[{}ms]", counterPerGuid.getElapsed());
-                ExecutionPlanResponse response = createExecution(loadedStages, snapshot.getMarks(), lastOwnerGuid, counterPerGuid.getElapsed());
+                ExecutionPlanResponse response = createExecution(pipelineRun, snapshot.getMarks(), lastOwnerGuid, counterPerGuid.getElapsed());
                 logger.info("Obtained cloud execution plan: {}", response.getAction());
 
                 //TODO should check if it has the lock?
@@ -103,8 +102,7 @@ public ExecutionPlan getNextExecution(PipelineRun pipelineRun) throws LockExcept
                 }
 
                 if (response.isContinue()) {
-                    List<ExecutableStage> executableStages = CloudExecutionPlanMapper.getExecutableStages(response, loadedStages);
-                    return ExecutionPlan.CONTINUE(executableStages);
+                    return ExecutionPlan.CONTINUE();
 
                 } else if (response.isExecute()) {
                     Lock lock = CloudLock.initialiseLocal(response.getLock(), coreConfiguration, runnerId, lockService, timeService);
@@ -132,8 +130,7 @@ public ExecutionPlan getNextExecution(PipelineRun pipelineRun) throws LockExcept
                     );
 
                 } else if (response.isAbort()) {
-                    List<ExecutableStage> stages = CloudExecutionPlanMapper.getExecutableStages(response, loadedStages);
-                    return ExecutionPlan.ABORT(stages);
+                    return ExecutionPlan.ABORT();
 
                 } else {
                     throw new RuntimeException("Unrecognized action from response. Not within(CONTINUE, EXECUTE, AWAIT, ABORT)");
@@ -148,7 +145,7 @@ public ExecutionPlan getNextExecution(PipelineRun pipelineRun) throws LockExcept
         } while (true);
     }
 
-    private ExecutionPlanResponse createExecution(List<AbstractLoadedStage> loadedStages,
+    private ExecutionPlanResponse createExecution(PipelineRun pipelineRun,
                                                   Collection<TargetSystemAuditMark> auditMarks,
                                                   String lastAcquisitionId,
                                                   long elapsedMillis) {
@@ -158,7 +155,7 @@ private ExecutionPlanResponse createExecution(List<AbstractLoadedStage> loadedSt
                 .collect(Collectors.toMap(TargetSystemAuditMark::getChangeId, TargetSystemAuditMark::getOperation));
 
         ExecutionPlanRequest requestBody = CloudExecutionPlanMapper.toRequest(
-                loadedStages,
+                pipelineRun,
                 coreConfiguration.getLockAcquiredForMillis(),
                 auditMarksMap);
 

cloud/flamingock-cloud/src/test/java/io/flamingock/cloud/CloudApiMapperTest.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2026 Flamingock (https://www.flamingock.io)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.flamingock.cloud;
+
+import io.flamingock.cloud.api.vo.CloudStageStatus;
+import io.flamingock.internal.common.core.recovery.RecoveryIssue;
+import io.flamingock.internal.common.core.response.data.ErrorInfo;
+import io.flamingock.internal.common.core.response.data.StageState;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class CloudApiMapperTest {
+
+    @Test
+    @DisplayName("toCloud(null) returns null (wire shape for NOT_STARTED is field absent)")
+    void nullReturnsNull() {
+        assertNull(CloudApiMapper.toCloud((StageState) null));
+    }
+
+    @Test
+    @DisplayName("toCloud(NOT_STARTED) returns null (wire shape for NOT_STARTED is field absent)")
+    void notStartedReturnsNull() {
+        assertNull(CloudApiMapper.toCloud(StageState.NOT_STARTED));
+    }
+
+    @Test
+    @DisplayName("toCloud(STARTED) maps to STARTED")
+    void startedMaps() {
+        assertEquals(CloudStageStatus.STARTED, CloudApiMapper.toCloud(StageState.STARTED));
+    }
+
+    @Test
+    @DisplayName("toCloud(COMPLETED) maps to COMPLETED")
+    void completedMaps() {
+        assertEquals(CloudStageStatus.COMPLETED, CloudApiMapper.toCloud(StageState.COMPLETED));
+    }
+
+    @Test
+    @DisplayName("toCloud(Failed) maps to FAILED")
+    void failedMaps() {
+        StageState failed = StageState.failed(new ErrorInfo("RuntimeException", "boom", Collections.emptyList(), "stage-1"));
+        assertEquals(CloudStageStatus.FAILED, CloudApiMapper.toCloud(failed));
+    }
+
+    @Test
+    @DisplayName("toCloud(BlockedForMI) maps to BLOCKED_MANUAL_INTERVENTION — not FAILED — even though BlockedForMI extends Failed")
+    void blockedForMIMapsBeforeFailed() {
+        StageState blocked = StageState.blockedManualIntervention(
+                "stage-1", Collections.singletonList(new RecoveryIssue("change-1")));
+        assertEquals(CloudStageStatus.BLOCKED_MANUAL_INTERVENTION, CloudApiMapper.toCloud(blocked));
+    }
+}