fix(plugin): surface missing state fields on hook info records

examples/src/main/java/software/amazon/lambda/durable/examples/general/PluginExample.java

@@ -84,8 +84,8 @@ public void onOperationEnd(OperationEndInfo info) {
         @Override
         public void onUserFunctionStart(UserFunctionStartInfo info) {
             System.out.printf(
-                    "[PLUGIN] onUserFunctionStart: name=%s, type=%s, attempt=%s, isReplayingChildren=%s%n",
-                    info.name(), info.type(), info.attempt(), info.isReplayingChildren());
+                    "[PLUGIN] onUserFunctionStart: name=%s, type=%s, attempt=%s, isReplay=%s%n",
+                    info.name(), info.type(), info.attempt(), info.isReplay());
         }
 
         @Override

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

@@ -103,6 +103,68 @@ void plugin_receivesInvocationEnd_withPendingStatus_onSuspension() {
         assertEquals(InvocationStatus.PENDING, plugin.invocationEnds.get(0).invocationStatus());
     }
 
+    @Test
+    void plugin_invocationSnapshots_trackReplayAcrossSuspension() {
+        var plugin = new RecordingPlugin();
+        var config = DurableConfig.builder().withPlugins(plugin).build();
+
+        var runner = LocalDurableTestRunner.create(
+                String.class,
+                (input, context) -> context.runInChildContext("child", String.class, child -> {
+                    child.wait("pause", Duration.ofMinutes(1));
+                    return "done";
+                }),
+                config);
+
+        var firstResult = runner.run("input");
+        assertEquals(ExecutionStatus.PENDING, firstResult.getStatus());
+
+        assertEquals(1, plugin.invocationStarts.size());
+        assertEquals(1, plugin.invocationEnds.size());
+        var firstStart = plugin.invocationStarts.get(0);
+        assertTrue(firstStart.isFirstInvocation());
+        assertEquals(1, firstStart.operations().size(), "first invocation starts with only the execution operation");
+        assertTrue(firstStart.updatedOperations().isEmpty());
+
+        var firstEnd = plugin.invocationEnds.get(0);
+        assertEquals(InvocationStatus.PENDING, firstEnd.invocationStatus());
+        assertTrue(firstEnd.operations().values().stream().anyMatch(op -> "child".equals(op.name())));
+        assertTrue(firstEnd.operations().values().stream().anyMatch(op -> "pause".equals(op.name())));
+
+        runner.advanceTime();
+        var secondResult = runner.run("input");
+        assertEquals(ExecutionStatus.SUCCEEDED, secondResult.getStatus());
+
+        assertEquals(2, plugin.invocationStarts.size());
+        assertEquals(2, plugin.invocationEnds.size());
+        var secondStart = plugin.invocationStarts.get(1);
+        assertFalse(secondStart.isFirstInvocation());
+        var replayedChild = secondStart.operations().values().stream()
+                .filter(op -> "child".equals(op.name()))
+                .findFirst()
+                .orElseThrow();
+        assertTrue(replayedChild.isReplay());
+        assertEquals(1, secondStart.updatedOperations().size());
+        var updatedPause = secondStart.updatedOperations().values().iterator().next();
+        assertEquals("pause", updatedPause.name());
+        assertEquals(OperationStatus.SUCCEEDED, updatedPause.status());
+        assertTrue(updatedPause.isReplay());
+
+        var secondEnd = plugin.invocationEnds.get(1);
+        assertEquals(InvocationStatus.SUCCEEDED, secondEnd.invocationStatus());
+        assertTrue(secondEnd.operations().values().stream()
+                .anyMatch(op -> "child".equals(op.name()) && op.status() == OperationStatus.SUCCEEDED));
+        assertTrue(secondEnd.operations().values().stream()
+                .anyMatch(op -> "pause".equals(op.name()) && op.status() == OperationStatus.SUCCEEDED));
+
+        var childStarts = plugin.userFunctionStarts.stream()
+                .filter(info -> "child".equals(info.name()))
+                .toList();
+        assertEquals(2, childStarts.size());
+        assertFalse(childStarts.get(0).isReplay());
+        assertTrue(childStarts.get(1).isReplay());
+    }
+
     @Test
     void plugin_receivesInvocationEnd_withFailedStatus_onError() {
         var plugin = new RecordingPlugin();

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

@@ -29,6 +29,7 @@
 import software.amazon.lambda.durable.plugin.InvocationEndInfo;
 import software.amazon.lambda.durable.plugin.InvocationInfo;
 import software.amazon.lambda.durable.plugin.InvocationStatus;
+import software.amazon.lambda.durable.plugin.PluginInfoConverter;
 import software.amazon.lambda.durable.plugin.PluginRunner;
 import software.amazon.lambda.durable.serde.SerDes;
 import software.amazon.lambda.durable.util.ExceptionHelper;
@@ -86,12 +87,20 @@ public static <I, O> DurableExecutionOutput execute(
                         // onInvocationStart runs on the user thread so plugins can
                         // inject ThreadLocal objects, update MDC, etc.
                         // executionStartTime comes from the initial EXECUTION operation in the first backend event.
-                        pluginRunner.onInvocationStart(new InvocationInfo(
-                                requestId,
-                                executionArn,
-                                isFirstInvocation,
-                                executionManager.getExecutionOperation().startTimestamp(),
-                                userInput));
+                        if (!pluginRunner.isEmpty()) {
+                            pluginRunner.onInvocationStart(new InvocationInfo(
+                                    requestId,
+                                    executionArn,
+                                    isFirstInvocation,
+                                    executionManager.getExecutionOperation().startTimestamp(),
+                                    userInput,
+                                    PluginInfoConverter.toOperationItemMap(
+                                            executionManager.getOperationsSnapshot(),
+                                            executionManager.getInitialOperationIds()),
+                                    PluginInfoConverter.toOperationItemMap(
+                                            executionManager.getUpdatedOperationsSnapshot(),
+                                            executionManager.getInitialOperationIds())));
+                        }
                         if (inputFailure != null) {
                             ExceptionHelper.sneakyThrow(inputFailure);
                         }
@@ -120,6 +129,7 @@ public static <I, O> DurableExecutionOutput execute(
                                 if (cause instanceof SuspendExecutionException) {
                                     fireOnInvocationEnd(
                                             pluginRunner,
+                                            executionManager,
                                             requestId,
                                             executionArn,
                                             isFirstInvocation,
@@ -138,6 +148,7 @@ public static <I, O> DurableExecutionOutput execute(
                                         && unrecoverableDurableExecutionException.isRetryable()) {
                                     fireOnInvocationEnd(
                                             pluginRunner,
+                                            executionManager,
                                             requestId,
                                             executionArn,
                                             isFirstInvocation,
@@ -152,6 +163,7 @@ public static <I, O> DurableExecutionOutput execute(
                                 logger.debug("Execution failed: {}", cause.getMessage());
                                 fireOnInvocationEnd(
                                         pluginRunner,
+                                        executionManager,
                                         requestId,
                                         executionArn,
                                         isFirstInvocation,
@@ -168,6 +180,7 @@ public static <I, O> DurableExecutionOutput execute(
                                     DurableExecutionOutput.success(handleLargePayload(executionManager, outputPayload));
                             fireOnInvocationEnd(
                                     pluginRunner,
+                                    executionManager,
                                     requestId,
                                     executionArn,
                                     isFirstInvocation,
@@ -188,15 +201,28 @@ public static <I, O> DurableExecutionOutput execute(
 
     private static void fireOnInvocationEnd(
             PluginRunner pluginRunner,
+            ExecutionManager executionManager,
             String requestId,
             String executionArn,
             boolean isFirstInvocation,
             InvocationStatus status,
             Throwable error,
             Object executionInput,
             Object executionResult) {
+        if (pluginRunner.isEmpty()) {
+            return;
+        }
         pluginRunner.onInvocationEnd(new InvocationEndInfo(
-                requestId, executionArn, isFirstInvocation, status, error, executionInput, executionResult));
+                requestId,
+                executionArn,
+                isFirstInvocation,
+                executionManager.getExecutionOperation().startTimestamp(),
+                PluginInfoConverter.toOperationItemMap(
+                        executionManager.getOperationsSnapshot(), executionManager.getInitialOperationIds()),
+                status,
+                error,
+                executionInput,
+                executionResult));
     }
 
     private static String handleLargePayload(ExecutionManager executionManager, String outputPayload) {

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

@@ -5,6 +5,7 @@
 import com.amazonaws.services.lambda.runtime.Context;
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
@@ -63,6 +64,7 @@ public class ExecutionManager implements SafeCloseable {
     private final AtomicReference<ExecutionMode> executionMode;
     private final DurableConfig durableConfig;
     private final Set<String> updatedOperationIdsSinceLastInvocation;
+    private final Set<String> initialOperationIds;
 
     // ===== Thread Coordination =====
     private final Map<String, BaseDurableOperation> registeredOperations = new ConcurrentHashMap<>();
@@ -89,6 +91,11 @@ public ExecutionManager(DurableExecutionInput input, DurableConfig config, Conte
         this.operationStorage = checkpointManager.fetchAllPages(input.initialExecutionState()).stream()
                 .collect(Collectors.toConcurrentMap(Operation::id, op -> op));
 
+        // The ids delivered in this invocation's initial state. Everything else in operationStorage is created during
+        // this invocation, so this set is what distinguishes replayed operations from freshly-started ones for the
+        // plugin hooks' isReplay indicators.
+        this.initialOperationIds = Set.copyOf(operationStorage.keySet());
+
         // Start in REPLAY mode if we have more than just the initial EXECUTION operation
         this.executionMode =
                 new AtomicReference<>(operationStorage.size() > 1 ? ExecutionMode.REPLAY : ExecutionMode.EXECUTION);
@@ -132,6 +139,47 @@ public boolean isOperationUpdatedSinceLastInvocation(String operationId) {
         return updatedOperationIdsSinceLastInvocation.contains(operationId);
     }
 
+    /**
+     * Returns {@code true} if the given operation was present in the checkpointed state delivered at the start of this
+     * invocation, i.e. it predates this invocation and is being replayed rather than started fresh. Unlike
+     * {@link #getOperationAndUpdateReplayState(String)} this does not mutate the execution's replay mode, so it is safe
+     * to call from plugin-hook firing sites.
+     *
+     * @param operationId the operation ID to check
+     * @return true if the operation was delivered in this invocation's initial state
+     */
+    public boolean wasObservedAtInvocationStart(String operationId) {
+        return initialOperationIds.contains(operationId);
+    }
+
+    /** Returns the ids of the operations delivered in this invocation's initial state. */
+    public Set<String> getInitialOperationIds() {
+        return initialOperationIds;
+    }
+
+    /**
+     * Returns an immutable snapshot of the operations currently tracked for this execution, including the initial
+     * EXECUTION operation. Non-mutating; intended for the invocation-level plugin hooks.
+     *
+     * @return a snapshot of the tracked operations
+     */
+    public Collection<Operation> getOperationsSnapshot() {
+        return List.copyOf(operationStorage.values());
+    }
+
+    /**
+     * Returns the subset of {@link #getOperationsSnapshot()} whose ids the backend reported as updated since the last
+     * successful invocation. Empty on the first invocation. Ids without a corresponding tracked operation are skipped.
+     *
+     * @return a snapshot of the externally-updated operations
+     */
+    public Collection<Operation> getUpdatedOperationsSnapshot() {
+        return updatedOperationIdsSinceLastInvocation.stream()
+                .map(operationStorage::get)
+                .filter(Objects::nonNull)
+                .toList();
+    }
+
     /** Registers an operation so it can receive checkpoint completion notifications. */
     public void registerOperation(BaseDurableOperation operation) {
         registeredOperations.put(operation.getOperationId(), operation);
@@ -162,7 +210,11 @@ private void onCheckpointComplete(List<Operation> newOperations) {
             durableConfig
                     .getPluginRunner()
                     .onOperationChange(PluginInfoConverter.toOperationChangeInfo(
-                            requestId, durableExecutionArn, updatedOperations, operationStorage.values()));
+                            requestId,
+                            durableExecutionArn,
+                            updatedOperations,
+                            operationStorage.values(),
+                            initialOperationIds));
         }
     }
 

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

@@ -351,7 +351,10 @@ protected void runUserHandler(Runnable runnable, ThreadType threadType) {
     protected <T> T runUserFunction(Integer attempt, Supplier<T> userFunction) {
         var pluginRunner = getPluginRunner();
         var startInfo = PluginInfoConverter.toUserFunctionStartInfo(
-                operationIdentifier, durableContext.getParentId(), durableContext.isReplaying(), attempt);
+                operationIdentifier,
+                durableContext.getParentId(),
+                executionManager.wasObservedAtInvocationStart(getOperationId()),
+                attempt);
         pluginRunner.onUserFunctionStart(startInfo);
         try {
             T result = userFunction.get();

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

@@ -2,58 +2,78 @@
 // SPDX-License-Identifier: Apache-2.0
 package software.amazon.lambda.durable.plugin;
 
+import java.time.Instant;
+import java.util.Map;
 import software.amazon.lambda.durable.annotations.Experimental;
 
 /**
  * Information provided at the end of a Lambda invocation.
  *
+ * <p>Carries the same invocation-identity surface as {@link InvocationInfo} so an invocation-end hook does not have to
+ * correlate back to the start hook to learn the execution start time or operation state.
+ *
  * @param requestId the Lambda request ID for this invocation
  * @param durableExecutionArn the durable execution ARN
  * @param isFirstInvocation true if this is the first invocation of the execution
- * @param invocationStatus the invocation outcome (SUCCEEDED, FAILED, or PENDING)
+ * @param executionStartTime the stable start timestamp of the durable execution
+ * @param operations a snapshot of operations known when the invocation ended, keyed by operation ID
+ * @param invocationStatus the invocation outcome
  * @param executionError non-null if the execution failed; this component is experimental
- * @param executionInput the deserialized execution input passed to the user handler, or null when no plugins are
- *     registered or the input could not be deserialized; this component is experimental
- * @param executionResult the value the user handler returned, or null unless the invocation completed the execution
- *     successfully; this component is experimental
+ * @param executionInput the deserialized execution input; this component is experimental
+ * @param executionResult the value returned by a successful handler; this component is experimental
  */
 public record InvocationEndInfo(
         String requestId,
         String durableExecutionArn,
         boolean isFirstInvocation,
+        Instant executionStartTime,
+        Map<String, OperationChangeItemInfo> operations,
         InvocationStatus invocationStatus,
         @Experimental Throwable executionError,
         @Experimental Object executionInput,
         @Experimental Object executionResult) {
 
-    /**
-     * Creates invocation-end information without the execution input or result.
-     *
-     * <p>Retained so callers written before {@code executionInput} and {@code executionResult} were added keep
-     * compiling; both resolve to null.
-     *
-     * @param requestId the Lambda request ID for this invocation
-     * @param durableExecutionArn the durable execution ARN
-     * @param isFirstInvocation true if this is the first invocation of the execution
-     * @param invocationStatus the invocation outcome
-     * @param executionError non-null if the execution failed
-     */
+    /** Creates invocation-end information without execution payloads or operation state. */
     public InvocationEndInfo(
             String requestId,
             String durableExecutionArn,
             boolean isFirstInvocation,
             InvocationStatus invocationStatus,
             Throwable executionError) {
-        this(requestId, durableExecutionArn, isFirstInvocation, invocationStatus, executionError, null, null);
+        this(
+                requestId,
+                durableExecutionArn,
+                isFirstInvocation,
+                null,
+                Map.of(),
+                invocationStatus,
+                executionError,
+                null,
+                null);
+    }
+
+    /** Creates invocation-end information without execution start time or operation state. */
+    public InvocationEndInfo(
+            String requestId,
+            String durableExecutionArn,
+            boolean isFirstInvocation,
+            InvocationStatus invocationStatus,
+            Throwable executionError,
+            Object executionInput,
+            Object executionResult) {
+        this(
+                requestId,
+                durableExecutionArn,
+                isFirstInvocation,
+                null,
+                Map.of(),
+                invocationStatus,
+                executionError,
+                executionInput,
+                executionResult);
     }
 
-    /**
-     * Returns a representation that omits {@code executionInput} and {@code executionResult}.
-     *
-     * <p>The generated representation would render both payloads, so plugins that log this object whole would start
-     * emitting customer inputs and results, potentially including secrets or personal data. Read the components
-     * explicitly to record them.
-     */
+    /** Returns a representation that omits execution payloads and operation snapshots. */
     @Override
     public String toString() {
         return "InvocationEndInfo[requestId=" + requestId + ", durableExecutionArn=" + durableExecutionArn