feat(plugin): add onOperationChange hook

sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java

@@ -11,6 +11,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.jupiter.api.Test;
 import software.amazon.awssdk.services.lambda.model.ErrorObject;
+import software.amazon.awssdk.services.lambda.model.OperationStatus;
 import software.amazon.lambda.durable.config.StepConfig;
 import software.amazon.lambda.durable.model.ExecutionStatus;
 import software.amazon.lambda.durable.model.WaitForConditionResult;
@@ -330,6 +331,63 @@ void plugin_operationEnd_noError_whenOperationSucceeds() {
         assertNull(stepEnd.error(), "error should be null for successful step");
     }
 
+    // ─── Operation change hook ───────────────────────────────────────────
+
+    @Test
+    void plugin_receivesOperationChange_forStep() {
+        var plugin = new RecordingPlugin();
+        var config = DurableConfig.builder().withPlugins(plugin).build();
+
+        var runner = LocalDurableTestRunner.create(
+                String.class, (input, context) -> context.step("my-step", String.class, stepCtx -> "result"), config);
+
+        runner.runUntilComplete("input");
+
+        // A checkpoint response that reports the step at a new status should fire onOperationChange
+        assertFalse(
+                plugin.operationChanges.isEmpty(), "onOperationChange should fire when an operation changes status");
+
+        var changeWithStep = plugin.operationChanges.stream()
+                .filter(info -> info.updatedOperations().values().stream().anyMatch(op -> "my-step".equals(op.name())))
+                .findFirst()
+                .orElse(null);
+        assertNotNull(changeWithStep, "onOperationChange should include the 'my-step' operation that changed status");
+        assertNotNull(changeWithStep.durableExecutionArn());
+        assertFalse(changeWithStep.operations().isEmpty(), "snapshot of all operations should be present");
+    }
+
+    @Test
+    void plugin_operationChange_includesErrorAndStatus_whenStepFails() {
+        var plugin = new RecordingPlugin();
+        var config = DurableConfig.builder().withPlugins(plugin).build();
+
+        var runner = LocalDurableTestRunner.create(
+                String.class,
+                (input, context) -> context.step(
+                        "failing-step",
+                        String.class,
+                        stepCtx -> {
+                            throw new RuntimeException("step exploded");
+                        },
+                        StepConfig.builder()
+                                .retryStrategy(RetryStrategies.Presets.NO_RETRY)
+                                .build()),
+                config);
+
+        var result = runner.run("input");
+        assertEquals(ExecutionStatus.FAILED, result.getStatus());
+
+        // The failing step transitions through several statuses; find the one reporting FAILED
+        var failedItem = plugin.operationChanges.stream()
+                .flatMap(info -> info.updatedOperations().values().stream())
+                .filter(op -> "failing-step".equals(op.name()) && op.status() == OperationStatus.FAILED)
+                .findFirst()
+                .orElse(null);
+        assertNotNull(failedItem, "onOperationChange should report 'failing-step' with FAILED status");
+        assertNotNull(failedItem.error(), "error should be propagated for the failed step");
+        assertTrue(failedItem.error().getMessage().contains("step exploded"));
+    }
+
     // ─── User function hooks ─────────────────────────────────────────────
 
     @Test
@@ -522,6 +580,7 @@ private static class RecordingPlugin implements DurableExecutionPlugin {
         final List<OperationEndInfo> operationEnds = Collections.synchronizedList(new ArrayList<>());
         final List<UserFunctionStartInfo> userFunctionStarts = Collections.synchronizedList(new ArrayList<>());
         final List<UserFunctionEndInfo> userFunctionEnds = Collections.synchronizedList(new ArrayList<>());
+        final List<OperationChangeInfo> operationChanges = Collections.synchronizedList(new ArrayList<>());
 
         @Override
         public void onInvocationStart(InvocationInfo info) {
@@ -552,6 +611,11 @@ public void onUserFunctionStart(UserFunctionStartInfo info) {
         public void onUserFunctionEnd(UserFunctionEndInfo info) {
             userFunctionEnds.add(info);
         }
+
+        @Override
+        public void onOperationChange(OperationChangeInfo info) {
+            operationChanges.add(info);
+        }
     }
 
     /** Plugin that throws on every hook to verify error isolation. */
@@ -585,5 +649,10 @@ public void onUserFunctionStart(UserFunctionStartInfo info) {
         public void onUserFunctionEnd(UserFunctionEndInfo info) {
             throw new RuntimeException("plugin error");
         }
+
+        @Override
+        public void onOperationChange(OperationChangeInfo info) {
+            throw new RuntimeException("plugin error");
+        }
     }
 }

sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java

@@ -54,7 +54,7 @@ public static <I, O> DurableExecutionOutput execute(
             BiFunction<I, DurableContext, O> handler,
             DurableConfig config) {
         var pluginRunner = config.getPluginRunner();
-        try (var executionManager = new ExecutionManager(input, config)) {
+        try (var executionManager = new ExecutionManager(input, config, lambdaContext)) {
             var isFirstInvocation = !executionManager.isReplaying();
             var requestId = lambdaContext != null ? lambdaContext.getAwsRequestId() : null;
             var executionArn = input.durableExecutionArn();

sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 package software.amazon.lambda.durable.execution;
 
+import com.amazonaws.services.lambda.runtime.Context;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -27,6 +28,7 @@
 import software.amazon.lambda.durable.model.DurableExecutionInput;
 import software.amazon.lambda.durable.model.SafeCloseable;
 import software.amazon.lambda.durable.operation.BaseDurableOperation;
+import software.amazon.lambda.durable.plugin.PluginInfoConverter;
 
 /**
  * Central manager for durable execution coordination.
@@ -57,6 +59,7 @@ public class ExecutionManager implements SafeCloseable {
     private final Map<String, Operation> operationStorage;
     private final Operation executionOp;
     private final String durableExecutionArn;
+    private final Context lambdaContext;
     private final AtomicReference<ExecutionMode> executionMode;
     private final DurableConfig durableConfig;
     private final Set<String> updatedOperationIdsSinceLastInvocation;
@@ -70,9 +73,10 @@ public class ExecutionManager implements SafeCloseable {
     // ===== Checkpoint Batching =====
     private final CheckpointManager checkpointManager;
 
-    public ExecutionManager(DurableExecutionInput input, DurableConfig config) {
+    public ExecutionManager(DurableExecutionInput input, DurableConfig config, Context lambdaContext) {
         durableConfig = config;
         this.durableExecutionArn = input.durableExecutionArn();
+        this.lambdaContext = lambdaContext;
 
         // Store the set of operation IDs updated since the last successful invocation
         this.updatedOperationIdsSinceLastInvocation =
@@ -136,7 +140,13 @@ public void registerOperation(BaseDurableOperation operation) {
     // ===== Checkpoint Completion Handler =====
     /** Called by CheckpointManager when a checkpoint completes. Updates operationStorage and notify operations . */
     private void onCheckpointComplete(List<Operation> newOperations) {
+        var updatedOperations = new ArrayList<Operation>();
         newOperations.forEach(op -> {
+            // Detect a status change against the previously stored operation
+            var previous = operationStorage.get(op.id());
+            if (previous == null || previous.status() != op.status()) {
+                updatedOperations.add(op);
+            }
             // Update operation storage
             operationStorage.put(op.id(), op);
             // call registered operation's onCheckpointComplete method for completed operations
@@ -145,6 +155,15 @@ private void onCheckpointComplete(List<Operation> newOperations) {
                 return operation;
             });
         });
+
+        // Fire onOperationChange when a checkpoint response changed one or more operations
+        if (!updatedOperations.isEmpty()) {
+            var requestId = lambdaContext != null ? lambdaContext.getAwsRequestId() : null;
+            durableConfig
+                    .getPluginRunner()
+                    .onOperationChange(PluginInfoConverter.toOperationChangeInfo(
+                            requestId, durableExecutionArn, updatedOperations, operationStorage.values()));
+        }
     }
 
     /**

sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java

@@ -535,7 +535,7 @@ protected void fireOnOperationEnd(Operation operation, Throwable error, boolean
      * Extracts the error from a terminal operation as a Throwable. Returns null if the operation succeeded or has no
      * error details.
      */
-    private Throwable extractErrorFromOperation(Operation operation) {
+    public static Throwable extractErrorFromOperation(Operation operation) {
         if (operation.status() != OperationStatus.FAILED
                 && operation.status() != OperationStatus.TIMED_OUT
                 && operation.status() != OperationStatus.STOPPED) {
@@ -549,7 +549,7 @@ private Throwable extractErrorFromOperation(Operation operation) {
     }
 
     /** Extracts the ErrorObject from an operation based on its type. */
-    private static ErrorObject getErrorObject(Operation operation) {
+    public static ErrorObject getErrorObject(Operation operation) {
         if (operation.type() == null) {
             return null;
         }

sdk/src/main/java/software/amazon/lambda/durable/plugin/DurableExecutionPlugin.java

@@ -51,6 +51,13 @@ default void onOperationStart(OperationInfo info) {}
      */
     default void onOperationEnd(OperationEndInfo info) {}
 
+    /**
+     * Called when a checkpoint response changes the status of one or more operations.
+     *
+     * <p>Receives the operations that changed and a snapshot of all operations.
+     */
+    default void onOperationChange(OperationChangeInfo info) {}
+
     // ─── User function hooks ─────────────────────────────────────────────
 
     /**

sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeInfo.java

@@ -0,0 +1,23 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.plugin;
+
+import java.util.Map;
+
+/**
+ * Information provided when a checkpoint response changes one or more operations.
+ *
+ * <p>Holds a map with an {@link OperationChangeItemInfo} for every operation that changed in the checkpoint response.
+ *
+ * @param requestId the Lambda request ID for the invocation that observed the change
+ * @param durableExecutionArn the durable execution ARN
+ * @param updatedOperations operations whose status changed in this checkpoint response, keyed by operation ID
+ * @param operations a snapshot of all known operations after the update, keyed by operation ID
+ * @deprecated This is a preview API that is experimental and may be changed or removed in future releases.
+ */
+@Deprecated
+public record OperationChangeInfo(
+        String requestId,
+        String durableExecutionArn,
+        Map<String, OperationChangeItemInfo> updatedOperations,
+        Map<String, OperationChangeItemInfo> operations) {}

sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeItemInfo.java

@@ -0,0 +1,32 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.plugin;
+
+import java.time.Instant;
+import software.amazon.awssdk.services.lambda.model.OperationStatus;
+
+/**
+ * Operation-level information for a single operation within an {@link OperationChangeInfo}.
+ *
+ * @param id operation ID
+ * @param name human-readable operation name (may be null)
+ * @param type operation type
+ * @param subType operation sub-type (may be null)
+ * @param parentId parent operation ID (null for root-level operations)
+ * @param startTimestamp when the operation started
+ * @param endTimestamp when the operation ended
+ * @param error non-null if the operation failed
+ * @param status operation status
+ * @deprecated This is a preview API that is experimental and may be changed or removed in future releases.
+ */
+@Deprecated
+public record OperationChangeItemInfo(
+        String id,
+        String name,
+        String type,
+        String subType,
+        String parentId,
+        Instant startTimestamp,
+        Instant endTimestamp,
+        Throwable error,
+        OperationStatus status) {}

sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java

@@ -3,8 +3,11 @@
 package software.amazon.lambda.durable.plugin;
 
 import java.time.Instant;
+import java.util.Collection;
+import java.util.stream.Collectors;
 import software.amazon.awssdk.services.lambda.model.Operation;
 import software.amazon.lambda.durable.model.OperationIdentifier;
+import software.amazon.lambda.durable.operation.BaseDurableOperation;
 
 /**
  * Utility methods for converting SDK internal types to plugin info records.
@@ -108,4 +111,43 @@ public static UserFunctionEndInfo toUserFunctionEndInfo(
                 succeeded,
                 error);
     }
+
+    /**
+     * Creates an {@link OperationChangeInfo} from the durable operations whose status changed in a checkpoint response
+     * and a snapshot of all operations tracked for the execution.
+     *
+     * @param requestId the Lambda request ID for the invocation
+     * @param durableExecutionArn the durable execution ARN
+     * @param updatedOperations the durable operations whose status changed in this checkpoint response
+     * @param allOperations all durable operations tracked for the execution after this response
+     * @return an OperationChangeInfo record
+     */
+    public static OperationChangeInfo toOperationChangeInfo(
+            String requestId,
+            String durableExecutionArn,
+            Collection<Operation> updatedOperations,
+            Collection<Operation> allOperations) {
+        return new OperationChangeInfo(
+                requestId,
+                durableExecutionArn,
+                updatedOperations.stream()
+                        .collect(Collectors.toUnmodifiableMap(
+                                Operation::id, PluginInfoConverter::toOperationChangeItemInfo)),
+                allOperations.stream()
+                        .collect(Collectors.toUnmodifiableMap(
+                                Operation::id, PluginInfoConverter::toOperationChangeItemInfo)));
+    }
+
+    private static OperationChangeItemInfo toOperationChangeItemInfo(Operation operation) {
+        return new OperationChangeItemInfo(
+                operation.id(),
+                operation.name(),
+                operation.typeAsString(),
+                operation.subType(),
+                operation.parentId(),
+                operation.startTimestamp(),
+                operation.endTimestamp(),
+                BaseDurableOperation.extractErrorFromOperation(operation),
+                operation.status());
+    }
 }

sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginRunner.java

@@ -78,6 +78,10 @@ public void onOperationEnd(OperationEndInfo info) {
         run(p -> p.onOperationEnd(info));
     }
 
+    public void onOperationChange(OperationChangeInfo info) {
+        run(p -> p.onOperationChange(info));
+    }
+
     public void onUserFunctionStart(UserFunctionStartInfo info) {
         run(p -> p.onUserFunctionStart(info));
     }

sdk/src/test/java/software/amazon/lambda/durable/DurableContextTest.java

@@ -46,7 +46,8 @@ private DurableContext createTestContext(List<Operation> initialOperations) {
                 CheckpointUpdatedExecutionState.builder().operations(operations).build();
         var executionManager = new ExecutionManager(
                 new DurableExecutionInput(EXECUTION_ARN, "test-token", initialExecutionState),
-                DurableConfig.builder().withDurableExecutionClient(client).build());
+                DurableConfig.builder().withDurableExecutionClient(client).build(),
+                null);
         var root = DurableContextImpl.createRootContext(
                 executionManager,
                 software.amazon.lambda.durable.DurableConfig.builder().build(),

sdk/src/test/java/software/amazon/lambda/durable/ReplayValidationTest.java

@@ -44,7 +44,8 @@ private DurableContext createTestContext(List<Operation> initialOperations) {
                 CheckpointUpdatedExecutionState.builder().operations(operations).build();
         var executionManager = new ExecutionManager(
                 new DurableExecutionInput(EXECUTION_ARN, "test-token", initialExecutionState),
-                DurableConfig.builder().withDurableExecutionClient(client).build());
+                DurableConfig.builder().withDurableExecutionClient(client).build(),
+                null);
         var context = DurableContextImpl.createRootContext(
                 executionManager, DurableConfig.builder().build(), null);
         executionManager.setCurrentThreadContext(new ThreadContext(EXECUTION_OP_ID + "-execution", ThreadType.CONTEXT));

sdk/src/test/java/software/amazon/lambda/durable/context/BaseContextImplTest.java

@@ -45,7 +45,8 @@ private ExecutionManager createExecutionManager() {
                                 + EXECUTION_NAME + "/" + INVOCATION_ID,
                         "test-token",
                         initialState),
-                DurableConfig.builder().withDurableExecutionClient(client).build());
+                DurableConfig.builder().withDurableExecutionClient(client).build(),
+                null);
     }
 
     @Test

sdk/src/test/java/software/amazon/lambda/durable/context/DurableContextImplTest.java

@@ -43,7 +43,8 @@ void setUp() {
                                 + EXECUTION_NAME + "/" + INVOCATION_ID,
                         "test-token",
                         initialState),
-                DurableConfig.builder().withDurableExecutionClient(client).build());
+                DurableConfig.builder().withDurableExecutionClient(client).build(),
+                null);
         // Simulate the root thread context as the executor would set it
         executionManager.setCurrentThreadContext(new ThreadContext(null, ThreadType.CONTEXT));
         rootContext = DurableContextImpl.createRootContext(