fix(testing): match replay state to the service

.github/scripts/install_otel_test_wheels.py

@@ -16,6 +16,7 @@
 ROOT = Path(__file__).resolve().parents[2]
 CORE = "aws-durable-execution-sdk-python"
 OTEL = CORE + "-otel"
+TESTING = CORE + "-testing"
 
 
 def built_wheel(package: str) -> Path:
@@ -72,7 +73,9 @@ def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--legacy-plugin", action="store_true")
     args = parser.parse_args()
-    packages = [CORE] if args.legacy_plugin else [CORE, OTEL]
+    # Both lanes use the built local runner, including its real callback-update
+    # producer. The legacy lane keeps the actual published OTel plugin unchanged.
+    packages = [CORE, TESTING] if args.legacy_plugin else [CORE, OTEL, TESTING]
     wheels = [built_wheel(package) for package in packages]
     subprocess.run(
         [

.github/workflows/ci.yml

@@ -10,7 +10,9 @@ on:
     branches: [ main ]
 
   pull_request:
-    branches: [ main ]
+    branches:
+      - main
+      - fix/python-testing-replay-state-20261021
 
 jobs:
   lint-commits:

.github/workflows/cloud-tests.yml

@@ -2,7 +2,9 @@ name: Cloud tests
 
 on:
   pull_request:
-    branches: ["main"]
+    branches:
+      - main
+      - fix/python-testing-replay-state-20261021
   workflow_dispatch:
     inputs:
       lambda_endpoint:

.github/workflows/conformance-tests.yml

@@ -8,7 +8,9 @@ on:
   push:
     branches: ["main"]
   pull_request:
-    branches: ["main"]
+    branches:
+      - main
+      - fix/python-testing-replay-state-20261021
     paths:
       - "packages/aws-durable-execution-sdk-python/**"
       - "packages/aws-durable-execution-sdk-python-conformance-tests/**"

.github/workflows/js-examples.yml

@@ -20,7 +20,9 @@ on:
       - ".github/workflows/js-examples.yml"
       - "pyproject.toml"
   pull_request:
-    branches: ["main"]
+    branches:
+      - main
+      - fix/python-testing-replay-state-20261021
     paths:
       - "packages/aws-durable-execution-sdk-python-testing/**"
       - ".github/scripts/js_examples/**"

.github/workflows/opentelemetry-conformance-tests.yml

@@ -8,7 +8,9 @@ name: OpenTelemetry Conformance Tests
 
 on:
   pull_request:
-    branches: ["main"]
+    branches:
+      - main
+      - fix/python-testing-replay-state-20261021
     paths:
       - "packages/aws-durable-execution-sdk-python/**"
       - "packages/aws-durable-execution-sdk-python-otel/**"

CONTRIBUTING.md

@@ -72,7 +72,7 @@ hatch run dev-examples:test    # run examples tests only
 
 ### Installed package compatibility testing
 
-Build the core and OTel distributions with `hatch build` in each package, then run
+Build the core, OTel and testing-library distributions with `hatch build` in each package, then run
 these commands from the repository root:
 
 ```bash

packages/aws-durable-execution-sdk-python-testing/README.md

@@ -170,6 +170,31 @@ flow, pagination, and a map of the code.
 
 ## Documentation
 
+### Operation state on replay
+
+Invocation input and `GetDurableExecutionState` follow the service's retained
+operation view. An operation is omitted when any ancestor is a `CONTEXT` that
+has `SUCCEEDED` or `FAILED` without `ReplayChildren: true`. The walk includes
+running intermediate contexts; a nested context's own retention flag cannot
+override a completed ancestor that prunes it. Roots and operations whose parent
+is absent are kept, and ordering does not change the rule.
+
+`ContextOptions.ReplayChildren` on START is ignored. A closing update records
+true only when it explicitly supplies true; otherwise it records false. Testing
+service responses include that closing false, while Core's public model and
+generic serialization contract remain unchanged.
+
+Checkpoint responses still contain all updated operations. `UpdatedOperationIds`
+can include an ID whose operation record is hidden from invocation input and
+GetState. Raw stores, history, results and nested operation inspection retain the
+complete execution. An updated ID alone does not restore missing metadata for a
+plugin. A real large child result retains its children through ReplayChildren
+when a test needs that public replay behavior.
+
+This corrects local service parity; it does not fix duplicate execution-view
+spans for virtual FLAT map iterations while their map is still running.
+
+
 ### Error Handling
 
 The testing framework implements AWS-compliant error responses that match the exact format expected by boto3 and AWS services. For detailed information about error response formats, exception types, and troubleshooting, see:

packages/aws-durable-execution-sdk-python-testing/docs/architecture.md

@@ -234,10 +234,18 @@ list in a checkpoint response. The delta for the next response is the
 operations whose `operation_last_touched_seq` is greater than
 `handler_seen_seq`.
 
-The watermark advances only when a checkpoint response is built. It never
-advances when invocation input is built. That is what makes retries
-correct. A failed invocation never moves the watermark. So the retry
-delivers the same state again.
+The watermark advances when a checkpoint response is built or an invocation
+response passes validation, including a `PENDING` response. Building invocation
+input does not acknowledge it: input updates remain pending until delivery is
+confirmed. A failed call or rejected response does not consume that input, so a
+retry can receive it again.
+
+The invocation acknowledgment uses the sequence captured when its input was
+built, on the execution's worker. It never moves a higher checkpoint watermark
+backward or consumes operations that changed after the captured snapshot. This
+also prevents an input-only callback resume from receiving the same terminal
+update again at a later checkpoint. The acknowledgment concerns runner state
+delivery, not telemetry export.
 
 ```
       seq_counter    handler_seen_seq    delta next checkpoint

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py

@@ -21,6 +21,9 @@
     CheckpointRequestDispatcher,
 )
 from aws_durable_execution_sdk_python_testing.clock import RealClock
+from aws_durable_execution_sdk_python_testing.operation_state import (
+    prune_replay_children,
+)
 from aws_durable_execution_sdk_python_testing.exceptions import (
     InvalidParameterValueException,
 )
@@ -153,7 +156,7 @@ def get_execution_state(
         token = CheckpointToken.from_str(checkpoint_token)
         execution = self._store.load(token.execution_arn)
         return StateOutput(
-            operations=execution.get_navigable_operations(),
+            operations=prune_replay_children(execution.get_navigable_operations()),
             next_marker=None,
         )
 

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py

@@ -100,9 +100,11 @@ def _create_context_details(self, update: OperationUpdate) -> ContextDetails | N
             ContextDetails(
                 result=update.payload,
                 error=update.error,
-                replay_children=update.context_options.replay_children
-                if update.context_options
-                else False,
+                replay_children=(
+                    update.action in (OperationAction.SUCCEED, OperationAction.FAIL)
+                    and update.context_options is not None
+                    and update.context_options.replay_children is True
+                ),
             )
             if update.operation_type == OperationType.CONTEXT
             else None

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py

@@ -24,6 +24,9 @@
 )
 
 from aws_durable_execution_sdk_python_testing.clock import real_now
+from aws_durable_execution_sdk_python_testing.operation_state import (
+    prune_replay_children,
+)
 from aws_durable_execution_sdk_python_testing.exceptions import (
     IllegalStateException,
     InvalidParameterValueException,
@@ -479,15 +482,11 @@ def record_invocation_completion(
         )
 
     def mark_state_delivered(self) -> None:
-        """Reset the list of changed operations once an invocation's input
-        carries them.
-
-        The service reports, on each invocation, the operations that
-        changed after the last state the handler observed. The runner
-        delivers state in the invocation input. So the list is reset when
-        that input is built, not when the invocation completes: an
-        operation that completes while the handler is still running is
-        reported on the next invocation.
+        """Explicitly clear all pending operation IDs after their delivery.
+
+        The live executor uses versioned ``advance_handler_seen`` acknowledgments
+        so failed invocations retain their input updates and concurrent changes
+        newer than the acknowledged input snapshot remain pending.
         """
         self.updated_operation_ids = []
 
@@ -692,6 +691,7 @@ def complete_callback_success(
                 end_timestamp=now if now is not None else real_now(),
                 callback_details=updated_callback_details,
             )
+            self._record_updated_operation(operation.operation_id)
             return self.operations[index]
 
     def complete_callback_failure(
@@ -708,8 +708,11 @@ def complete_callback_failure(
             self.touch_operation(operation.operation_id)
             updated_callback_details = None
             if operation.callback_details:
+                # Match CallbackDetails.from_dict without depending on a store
+                # serialization round trip: an empty wire Error has no details.
                 updated_callback_details = replace(
-                    operation.callback_details, error=error
+                    operation.callback_details,
+                    error=error if error is not None and error.to_dict() else None,
                 )
 
             self.operations[index] = replace(
@@ -718,6 +721,7 @@ def complete_callback_failure(
                 end_timestamp=now if now is not None else real_now(),
                 callback_details=updated_callback_details,
             )
+            self._record_updated_operation(operation.operation_id)
             return self.operations[index]
 
     def complete_callback_timeout(
@@ -744,6 +748,7 @@ def complete_callback_timeout(
                 end_timestamp=now if now is not None else real_now(),
                 callback_details=updated_callback_details,
             )
+            self._record_updated_operation(operation.operation_id)
             return self.operations[index]
 
     def complete_chained_invoke(
@@ -897,10 +902,9 @@ def page(
         ``max_size_bytes`` and, when given, ``max_items``. Second element
         of the tuple is a marker for the next page, or ``None`` when the
         page fits everything."""
-        start_idx = self._resolve_marker(marker)
-        return self._walk_page(
-            self.snapshot_operations, start_idx, max_size_bytes, max_items
-        )
+        operations = prune_replay_children(self.snapshot_operations)
+        start_idx = self._resolve_marker(marker, len(operations))
+        return self._walk_page(operations, start_idx, max_size_bytes, max_items)
 
     def unseen_operations(self) -> list[Operation]:
         """Operations whose ``operation_last_touched_seq`` is strictly
@@ -920,11 +924,20 @@ def unseen_operations(self) -> list[Operation]:
         ]
 
     def advance_handler_seen(self, seq: int) -> None:
-        """Advance the wrapped execution's ``handler_seen_seq`` to
-        ``seq`` if that represents forward progress. Monotonic:
-        smaller or equal values are ignored."""
+        """Acknowledge delivered updates without retreating the watermark.
+
+        Equal acknowledgments also consume old-format, unversioned update IDs.
+        Only versions covered by this delivery may be removed.
+        """
         if seq > self.execution.handler_seen_seq:
             self.execution.handler_seen_seq = seq
+        # A checkpoint response or completed invocation delivered this state.
+        # Retain changes newer than that acknowledged snapshot.
+        self.execution.updated_operation_ids = [
+            operation_id
+            for operation_id in self.execution.updated_operation_ids
+            if self.execution.operation_last_touched_seq.get(operation_id, 0) > seq
+        ]
 
     # --- internals -------------------------------------------------
 
@@ -953,14 +966,21 @@ def _size_for(self, op: Operation) -> int:
         # floor at 1 to guarantee pagination always advances.
         return max(self.execution.operation_size_bytes.get(op.operation_id, 0), 1)
 
-    def _resolve_marker(self, marker: str | None) -> int:
+    def _resolve_marker(
+        self, marker: str | None, operation_count: int | None = None
+    ) -> int:
         if marker is None:
             return 0
         seq, idx = self._decode_marker(marker)
         if seq != self.pinned_token_sequence:
             msg = "Invalid marker"
             raise InvalidParameterValueException(msg)
-        if idx < 0 or idx > len(self.snapshot_operations):
+        count = (
+            len(self.snapshot_operations)
+            if operation_count is None
+            else operation_count
+        )
+        if idx < 0 or idx > count:
             msg = "Invalid marker"
             raise InvalidParameterValueException(msg)
         return idx

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py

@@ -1551,7 +1551,8 @@ def _begin_invocation(
         self._set_invocation_gate(execution_arn, InvocationState.INVOKING)
         execution.begin_new_invocation()
         invocation_input = self._invoker.create_invocation_input(execution=execution)
-        execution.mark_state_delivered()
+        # Keep input updates until a valid response acknowledges this snapshot.
+        # A failed invocation must not consume their delivery before a retry.
         self._store.save(execution)
         return execution, invocation_input, execution.seq_counter
 
@@ -1618,6 +1619,13 @@ def _finish_invocation(
         # A clean invocation resets the retry counter.
         reset_target = self._store.load(execution_arn)
         reset_target.consecutive_failed_invocation_attempts = 0
+        if invocation_seq is not None:
+            # A valid response acknowledges the input snapshot, even when this
+            # invocation suspended without another checkpoint. Newer changes
+            # must remain available for the next invocation or checkpoint.
+            OperationPaginatorState.pin(reset_target).advance_handler_seen(
+                invocation_seq
+            )
         self._store.save(reset_target)
 
         # Release the gate: terminal -> COMPLETED, otherwise PRE_INVOKE

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py

@@ -27,6 +27,9 @@
     ResourceNotFoundException,
 )
 from aws_durable_execution_sdk_python_testing.execution import OperationPaginatorState
+from aws_durable_execution_sdk_python_testing.operation_state import (
+    operation_state_to_dict,
+)
 from aws_durable_execution_sdk_python_testing.model import (
     LambdaContext,
     executed_version,
@@ -511,10 +514,15 @@ def invoke(
             input.durable_execution_arn, endpoint_url, region_name
         )
 
+        invocation_payload = input.to_json_dict()
+        invocation_payload["InitialExecutionState"]["Operations"] = [
+            operation_state_to_dict(operation, json_timestamps=True)
+            for operation in input.initial_execution_state.operations
+        ]
         invoke_kwargs: dict[str, Any] = {
             "FunctionName": function_name,
             "InvocationType": "RequestResponse",  # Synchronous invocation
-            "Payload": json.dumps(input.to_json_dict()),
+            "Payload": json.dumps(invocation_payload),
         }
         if tenant_id is not None:
             invoke_kwargs["TenantId"] = tenant_id