fix: tolerate string subtype metadata

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

@@ -12,6 +12,9 @@ OpenTelemetry instrumentation plugin for the [AWS Durable Execution SDK for Pyth
 - **Provider Integration**: Use the global ADOT provider or supply an explicit SDK `TracerProvider`
 - **Execution Sampling**: Resolve sampling once per invocation and apply it consistently to Workflow, Invocation, operation, and attempt spans
 
+Operation subtype attributes accept enum members or plain string labels. The
+plugin preserves the label and uses the operation's explicit type independently.
+
 ## Installation
 
 When using an ADOT or community OpenTelemetry Lambda layer:

packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py

@@ -921,7 +921,9 @@ def _operation_attributes(self, info: Any) -> dict[str, Any]:
         if getattr(info, "operation_type", None) is not None:
             attributes["durable.operation.type"] = info.operation_type.value
         if getattr(info, "sub_type", None) is not None:
-            attributes["durable.operation.subtype"] = info.sub_type.value
+            attributes["durable.operation.subtype"] = getattr(
+                info.sub_type, "value", info.sub_type
+            )
         # STEP user-function spans represent attempts, not durable operations.
         if (
             not (

packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py

@@ -984,7 +984,9 @@ def _extract_attributes(self, info: Any) -> _SpanAttributes:
         if hasattr(info, "operation_type") and info.operation_type is not None:
             attributes["durable.operation.type"] = info.operation_type.value
         if hasattr(info, "sub_type") and info.sub_type is not None:
-            attributes["durable.operation.subtype"] = info.sub_type.value
+            attributes["durable.operation.subtype"] = getattr(
+                info.sub_type, "value", info.sub_type
+            )
         # STEP user-function spans represent attempts, not durable operations.
         if (
             not (

packages/aws-durable-execution-sdk-python-otel/tests/test_subtype_compatibility.py

@@ -0,0 +1,96 @@
+"""Plugin hook metadata may come from enum- or string-based SDK subtypes."""
+
+from datetime import UTC, datetime, timedelta
+
+import opentelemetry.context as otel_context
+import pytest
+from aws_durable_execution_sdk_python.lambda_service import (
+    OperationStatus,
+    OperationSubType,
+)
+from aws_durable_execution_sdk_python.plugin import (
+    InvocationEndInfo,
+    InvocationStartInfo,
+    InvocationStatus,
+    OperationEndInfo,
+    OperationStartInfo,
+    OperationType,
+)
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+
+from aws_durable_execution_sdk_python_otel import (
+    ExecutionOtelPlugin,
+    InvocationOtelPlugin,
+    OtelPluginConfig,
+)
+
+
+@pytest.mark.parametrize("plugin_type", [ExecutionOtelPlugin, InvocationOtelPlugin])
+@pytest.mark.parametrize(
+    "sub_type,expected",
+    [
+        (OperationSubType.WAIT, "Wait"),
+        ("Wait", "Wait"),
+        ("Library_CustomWait", "Library_CustomWait"),
+        (None, None),
+    ],
+)
+def test_hook_exports_enum_or_string_subtype_without_changing_type(
+    plugin_type, sub_type, expected
+):
+    provider = TracerProvider()
+    exporter = InMemorySpanExporter()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    plugin = plugin_type(
+        OtelPluginConfig(
+            tracer_provider=provider,
+            context_extractor=lambda _: None,
+            enrich_logger=False,
+        )
+    )
+    before = otel_context.get_current()
+    start = datetime.now(UTC)
+    invocation = {
+        "execution_arn": "arn:aws:lambda:us-west-2:123456789012:function:subtype:$LATEST",
+        "request_id": "request",
+        "execution_start_time": start,
+        "is_first_invocation": True,
+    }
+    operation = {
+        "operation_id": "target",
+        "operation_type": OperationType.WAIT,
+        "sub_type": sub_type,
+        "name": "target",
+        "parent_id": None,
+        "start_time": start,
+        "is_replayed": False,
+    }
+    try:
+        plugin.on_invocation_start(InvocationStartInfo(**invocation))
+        plugin.on_operation_start(
+            OperationStartInfo(**operation, status=OperationStatus.STARTED)
+        )
+        plugin.on_operation_end(
+            OperationEndInfo(
+                **operation,
+                status=OperationStatus.SUCCEEDED,
+                end_time=start + timedelta(seconds=1),
+            )
+        )
+        (span,) = [
+            span for span in exporter.get_finished_spans() if span.name == "target"
+        ]
+        assert span.attributes["durable.operation.type"] == "WAIT"
+        assert span.attributes["durable.operation.status"] == "SUCCEEDED"
+        if expected is None:
+            assert "durable.operation.subtype" not in span.attributes
+        else:
+            assert span.attributes["durable.operation.subtype"] == expected
+    finally:
+        plugin.on_invocation_end(
+            InvocationEndInfo(**invocation, status=InvocationStatus.SUCCEEDED)
+        )
+        provider.shutdown()
+        assert otel_context.get_current() == before

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

@@ -170,6 +170,13 @@ flow, pagination, and a map of the code.
 
 ## Documentation
 
+History event projection can read enum members or plain string subtype metadata.
+Parsing checkpoints and reconstructing typed operation results still requires
+known core enum labels; unknown labels are rejected before checkpoint mutation.
+Declared public core and testing subtype types remain unchanged. Support for
+custom-subtype configuration, persistence and replay requires a coordinated future
+core implementation; this preparation does not change those contracts.
+
 ### 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/src/aws_durable_execution_sdk_python_testing/model.py

@@ -8,7 +8,7 @@
 import json
 from dataclasses import dataclass, replace
 from enum import Enum
-from typing import Any
+from typing import Any, cast
 
 from aws_durable_execution_sdk_python.execution import DurableExecutionInvocationOutput
 
@@ -1330,7 +1330,11 @@ def create(
 
     @property
     def sub_type(self) -> str | None:
-        return self.operation.sub_type.value if self.operation.sub_type else None
+        if not self.operation.sub_type:
+            return None
+        return cast(
+            str, getattr(self.operation.sub_type, "value", self.operation.sub_type)
+        )
 
     def get_retry_details(self) -> RetryDetails | None:
         if not self.operation.step_details or not self.operation_update:

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

@@ -0,0 +1,60 @@
+"""Read subtype labels through the public cloud-history result API."""
+
+from datetime import UTC, datetime
+
+import pytest
+from aws_durable_execution_sdk_python.lambda_service import (
+    OperationStatus,
+    OperationSubType,
+    OperationType,
+)
+
+from aws_durable_execution_sdk_python_testing.model import (
+    GetDurableExecutionHistoryResponse,
+    GetDurableExecutionResponse,
+)
+from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestResult
+
+
+@pytest.mark.parametrize("label", ["Step", None])
+def test_history_result_preserves_declared_enum_contract(label):
+    events = []
+    for index, (name, sub_type) in enumerate(
+        [("known", "Step"), ("target", label)], start=1
+    ):
+        event = {
+            "EventType": "StepSucceeded",
+            "EventId": index,
+            "EventTimestamp": datetime(2024, 1, 1, tzinfo=UTC),
+            "Id": name,
+            "Name": name,
+            "StepSucceededDetails": {
+                "Result": {"Payload": '"saved"', "Truncated": False}
+            },
+        }
+        if sub_type is not None:
+            event["SubType"] = sub_type
+        events.append(event)
+    history = GetDurableExecutionHistoryResponse.from_dict({"Events": events})
+    # Exercise the history DTO round trip as well as the public result factory.
+    history = GetDurableExecutionHistoryResponse.from_dict(history.to_dict())
+    result = DurableFunctionTestResult.from_execution_history(
+        GetDurableExecutionResponse(
+            durable_execution_arn="execution",
+            durable_execution_name="execution",
+            function_arn="function",
+            status="SUCCEEDED",
+            start_timestamp=datetime(2024, 1, 1, tzinfo=UTC),
+        ),
+        history,
+    )
+    assert len(result.operations) == 2
+    assert result.get_step("known").sub_type is OperationSubType.STEP
+    target = result.get_step("target")
+    assert target.operation_type is OperationType.STEP
+    assert target.status is OperationStatus.SUCCEEDED
+    assert target.result == '"saved"'
+    if label == "Step":
+        assert target.sub_type is OperationSubType.STEP
+    elif label is None:
+        assert target.sub_type is None

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

@@ -0,0 +1,112 @@
+"""Read string metadata without widening current checkpoint/result contracts."""
+
+import datetime
+
+import pytest
+from aws_durable_execution_sdk_python.lambda_service import (
+    Operation,
+    OperationStatus,
+    OperationSubType,
+    OperationType,
+)
+
+from aws_durable_execution_sdk_python_testing.model import (
+    CheckpointDurableExecutionRequest,
+    Event,
+    EventCreationContext,
+    StartDurableExecutionInput,
+    events_to_operations,
+)
+from aws_durable_execution_sdk_python_testing.exceptions import (
+    InvalidParameterValueException,
+)
+
+
+@pytest.mark.parametrize(
+    "sub_type,expected",
+    [
+        (OperationSubType.STEP, "Step"),
+        ("Step", "Step"),
+        ("Library_CustomStep", "Library_CustomStep"),
+        (None, None),
+        ("", None),
+    ],
+)
+def test_history_projection_keeps_enum_or_plain_string_subtype(sub_type, expected):
+    operation = Operation(
+        operation_id="operation",
+        operation_type=OperationType.STEP,
+        status=OperationStatus.STARTED,
+        sub_type=sub_type,
+        start_timestamp=datetime.datetime(2024, 1, 1, tzinfo=datetime.UTC),
+    )
+    context = EventCreationContext(
+        operation=operation,
+        event_id=1,
+        durable_execution_arn="execution",
+        start_durable_execution_input=StartDurableExecutionInput(
+            account_id="123456789012",
+            function_name="function",
+            function_qualifier="$LATEST",
+            execution_name="execution",
+            execution_timeout_seconds=30,
+            execution_retention_period_days=1,
+        ),
+    )
+    event = Event.create_event_started(context)
+    if expected is None:
+        assert "SubType" not in event.to_dict()
+    else:
+        assert event.to_dict()["SubType"] == expected
+    assert Event.from_dict(event.to_dict()).sub_type == expected
+    if expected == "Library_CustomStep":
+        with pytest.raises(InvalidParameterValueException):
+            events_to_operations([event])
+        return
+    (restored,) = events_to_operations([event])
+    assert restored.operation_type is OperationType.STEP
+    assert getattr(restored.sub_type, "value", restored.sub_type) == expected
+
+
+@pytest.mark.parametrize("sub_type", ["Step", None, ""])
+def test_checkpoint_request_preserves_existing_enum_contract(sub_type):
+    request = CheckpointDurableExecutionRequest.from_dict(
+        {
+            "CheckpointToken": "token",
+            "Updates": [
+                {
+                    "Id": "operation",
+                    "Type": "STEP",
+                    "Action": "START",
+                    "SubType": sub_type,
+                }
+            ],
+        },
+        durable_execution_arn="execution",
+    )
+    assert request.updates is not None and len(request.updates) == 1
+    update = request.updates[0]
+    assert update.operation_type is OperationType.STEP
+    if sub_type == "Step":
+        assert update.sub_type is OperationSubType.STEP
+    elif not sub_type:
+        assert update.sub_type is None
+
+
+@pytest.mark.parametrize("sub_type", ["Library_CustomStep", 42, True])
+def test_checkpoint_request_rejects_unsupported_subtype(sub_type):
+    with pytest.raises(ValueError, match="not a valid OperationSubType"):
+        CheckpointDurableExecutionRequest.from_dict(
+            {
+                "CheckpointToken": "token",
+                "Updates": [
+                    {
+                        "Id": "operation",
+                        "Type": "STEP",
+                        "Action": "START",
+                        "SubType": sub_type,
+                    }
+                ],
+            },
+            durable_execution_arn="execution",
+        )