fix(sdk): load state with unlisted subtypes

packages/aws-durable-execution-sdk-python-testing/tests/model_test.py

@@ -8,6 +8,7 @@
 import pytest
 from aws_durable_execution_sdk_python.lambda_service import (
     OperationStatus,
+    OperationSubType,
     OperationType,
 )
 
@@ -3552,21 +3553,21 @@ def test_events_to_operations_preserves_sub_type():
     assert operations[0].sub_type.value == "Step"
 
 
-def test_events_to_operations_invalid_sub_type():
-    """Test events_to_operations raises InvalidParameterValueException when sub_type is invalid."""
-    invalid_sub_type: str = "INVALID_SUB_TYPE"
+def test_events_to_operations_keeps_unlisted_sub_type():
+    """An event subtype that the enum does not list loads as its own member."""
+    unlisted_sub_type: str = "PyTestUnlistedHistory"
     event = Event(
         event_type="StepStarted",
         event_timestamp=datetime.datetime(2023, 1, 1, 0, 0, 0, tzinfo=datetime.UTC),
         operation_id="step-1",
-        sub_type=invalid_sub_type,
+        sub_type=unlisted_sub_type,
     )
 
-    with pytest.raises(
-        InvalidParameterValueException,
-        match=f"'{invalid_sub_type}' is not a valid OperationSubType",
-    ):
-        events_to_operations([event])
+    operations = events_to_operations([event])
+
+    assert len(operations) == 1
+    assert operations[0].sub_type is OperationSubType(unlisted_sub_type)
+    assert operations[0].sub_type.value == unlisted_sub_type
 
 
 def test_invocation_completed_details_to_json_dict():

packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py

@@ -105,6 +105,25 @@ class CallbackTimeoutType(Enum):
 
 
 class OperationSubType(Enum):
+    """The subtype that an operation records in its checkpoint.
+
+    The members are the subtypes that this SDK records. The checkpointed
+    state can also hold a subtype that this enum does not list:
+
+    1. Another SDK version, or a library built on the SDK, can record a
+       subtype of its own.
+    2. The SDK loads every operation of the execution on each invocation.
+    3. So a lookup that raised on an unknown subtype would fail the whole
+       execution, even when the current code never reads that operation.
+
+    So ``OperationSubType("SomeSubtype")`` returns a member for any non-empty
+    string. A listed value returns its listed member. Any other string
+    returns a member that this class creates once and then reuses. So
+    ``.value`` returns the recorded string, and ``is`` compares two lookups
+    of the same string as equal. The created members do not appear in
+    iteration or in ``__members__``.
+    """
+
     STEP = "Step"
     WAIT = "Wait"
     CALLBACK = "Callback"
@@ -117,6 +136,26 @@ class OperationSubType(Enum):
     WAIT_FOR_CONDITION = "WaitForCondition"
     CHAINED_INVOKE = "ChainedInvoke"
 
+    @classmethod
+    def _missing_(cls, value: object) -> OperationSubType | None:
+        """Return the member for a subtype string that the enum does not list.
+
+        The Enum lookup checks ``_value2member_map_`` before it calls this
+        method. So the member is stored there, and every later lookup of the
+        same string returns the same object without calling this method.
+        Two threads can create a member for the same string at once.
+        ``setdefault`` keeps the first one, so both lookups return it.
+
+        A value that is not a non-empty string returns ``None``, and the
+        lookup raises ``ValueError`` as before.
+        """
+        if not isinstance(value, str) or not value:
+            return None
+        member = object.__new__(cls)
+        member._name_ = value
+        member._value_ = value
+        return cls._value2member_map_.setdefault(value, member)  # type: ignore[return-value]
+
 
 class InvocationStatus(Enum):
     SUCCEEDED = "SUCCEEDED"

packages/aws-durable-execution-sdk-python/tests/execution_test.py

@@ -2957,6 +2957,102 @@ def test_handler(event: Any, context: DurableContext) -> str:
     assert child_body_calls == []
 
 
+def _unlisted_subtype_event(operation: dict[str, Any]) -> dict[str, Any]:
+    """An invocation event whose state holds one more operation."""
+    return {
+        "DurableExecutionArn": "arn:test:execution/exec1",
+        "CheckpointToken": "token123",
+        "InitialExecutionState": {
+            "Operations": [
+                {
+                    "Id": "exec1",
+                    "Type": "EXECUTION",
+                    "Status": "STARTED",
+                    "ExecutionDetails": {"InputPayload": "{}"},
+                },
+                operation,
+            ],
+            "NextMarker": "",
+        },
+    }
+
+
+def test_durable_execution_loads_state_with_unlisted_subtype():
+    """An operation with an unlisted subtype does not fail the invocation.
+
+    Another SDK version, or a library on top of the SDK, can record a subtype
+    that this SDK does not list. The SDK parses every operation in the state
+    on each invocation, including operations that the code never reads.
+    """
+    with patch(
+        "aws_durable_execution_sdk_python.execution.LambdaClient"
+    ) as mock_lambda_client:
+        mock_client = Mock(spec=DurableServiceClient)
+        mock_lambda_client.initialize_client.return_value = mock_client
+        event = _unlisted_subtype_event(
+            {
+                "Id": "other-operation",
+                "Type": "STEP",
+                "Status": "SUCCEEDED",
+                "Name": "launch",
+                "SubType": "PyTestUnlistedLaunch",
+                "StepDetails": {"Result": json.dumps("launched")},
+            }
+        )
+
+        @durable_execution
+        def test_handler(event: Any, context: DurableContext) -> str:
+            return "done"
+
+        result = test_handler(event, _make_lambda_context())
+
+    assert result["Status"] == InvocationStatus.SUCCEEDED.value
+    assert result["Result"] == json.dumps("done")
+
+
+def test_durable_execution_reports_unlisted_subtype_replay_mismatch():
+    """A replayed operation whose checkpoint has another subtype fails as nondeterministic.
+
+    The checkpoint's unlisted subtype appears in the error message, instead of
+    a ValueError from parsing the state.
+    """
+    with patch(
+        "aws_durable_execution_sdk_python.execution.LambdaClient"
+    ) as mock_lambda_client:
+        mock_client = Mock(spec=DurableServiceClient)
+        mock_lambda_client.initialize_client.return_value = mock_client
+        event = _unlisted_subtype_event(
+            {
+                "Id": OperationIdNamespace().create_id_for_step(1),
+                "Type": "CONTEXT",
+                "Status": "SUCCEEDED",
+                "Name": "child",
+                "SubType": "PyTestUnlistedChild",
+                "ContextDetails": {"Result": json.dumps("cached")},
+            }
+        )
+        child_body_calls: list[bool] = []
+
+        def child_body(_child: DurableContext) -> str:
+            child_body_calls.append(True)
+            return "executed"
+
+        @durable_execution
+        def test_handler(event: Any, context: DurableContext) -> str:
+            return context.run_in_child_context(child_body, name="child")
+
+        result = test_handler(event, _make_lambda_context())
+
+    assert result["Status"] == InvocationStatus.FAILED.value
+    assert (
+        result["Error"]["ErrorType"]
+        == "aws_durable_execution_sdk_python.exceptions.NonDeterministicExecutionError"
+    )
+    assert "subtype checkpoint='PyTestUnlistedChild'" in result["Error"]["ErrorMessage"]
+    assert "current='RunInChildContext'" in result["Error"]["ErrorMessage"]
+    assert child_body_calls == []
+
+
 @pytest.mark.parametrize("operation_kind", ["map", "parallel"])
 @pytest.mark.parametrize("parent_replay_children", [False, True])
 @pytest.mark.parametrize(

packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py

@@ -1604,6 +1604,58 @@ def test_operation_from_dict_with_subtype():
     assert operation.sub_type is OperationSubType.STEP
 
 
+def test_operation_sub_type_lookup_returns_member_for_unlisted_string():
+    """A subtype that the enum does not list resolves to a reused member."""
+    sub_type = OperationSubType("PyTestUnlistedLookup")
+
+    assert isinstance(sub_type, OperationSubType)
+    assert sub_type.value == "PyTestUnlistedLookup"
+    assert OperationSubType("PyTestUnlistedLookup") is sub_type
+    assert sub_type is not OperationSubType.STEP
+    # A listed value still resolves to its listed member.
+    assert OperationSubType("Step") is OperationSubType.STEP
+    # The created member does not join the listed members.
+    assert "PyTestUnlistedLookup" not in OperationSubType.__members__
+    assert sub_type not in list(OperationSubType)
+
+
+@pytest.mark.parametrize("value", ["", None, 3])
+def test_operation_sub_type_lookup_rejects_non_string_or_empty(value):
+    """Only a non-empty string resolves to a created member."""
+    with pytest.raises(ValueError, match="is not a valid OperationSubType"):
+        OperationSubType(value)
+
+
+def test_operation_from_dict_keeps_unlisted_subtype():
+    """State recorded with an unlisted subtype loads and round-trips."""
+    data = {
+        "Id": "op1",
+        "Type": "STEP",
+        "Status": "SUCCEEDED",
+        "SubType": "PyTestUnlistedOperation",
+    }
+
+    operation = Operation.from_dict(data)
+
+    assert operation.sub_type is OperationSubType("PyTestUnlistedOperation")
+    assert operation.to_dict()["SubType"] == "PyTestUnlistedOperation"
+
+
+def test_operation_update_from_dict_keeps_unlisted_subtype():
+    """An update with an unlisted subtype loads and round-trips."""
+    data = {
+        "Id": "op1",
+        "Type": "STEP",
+        "Action": "START",
+        "SubType": "PyTestUnlistedUpdate",
+    }
+
+    update = OperationUpdate.from_dict(data)
+
+    assert update.sub_type is OperationSubType("PyTestUnlistedUpdate")
+    assert update.to_dict()["SubType"] == "PyTestUnlistedUpdate"
+
+
 def test_operation_from_dict_complete():
     """Test Operation.from_dict with all fields."""
     start_time = datetime.datetime(2023, 1, 1, 10, 0, 0, tzinfo=datetime.UTC)