feat(sdk): custom subtypes typed str

packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py

@@ -45,7 +45,7 @@ def _attempt_record(hook: str, info: OperationInfo) -> dict[str, Any]:
     if info.name is not None:
         record["name"] = info.name
     if info.sub_type is not None:
-        record["subType"] = info.sub_type.value
+        record["subType"] = info.sub_type
     if info.parent_id is not None:
         record["parentId"] = info.parent_id
     if info.attempt is not None:

packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_context_info_shape.py

@@ -47,7 +47,7 @@ def _context_record(hook: str, info: OperationInfo) -> dict[str, Any]:
     if info.name is not None:
         record["name"] = info.name
     if info.sub_type is not None:
-        record["subType"] = info.sub_type.value
+        record["subType"] = info.sub_type
     if info.parent_id is not None:
         record["parentId"] = info.parent_id
     if info.start_time is not None:

packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py

@@ -39,7 +39,7 @@ def _add_operation_fields(record: dict[str, Any], info: OperationInfo) -> None:
     if info.name is not None:
         record["name"] = info.name
     if info.sub_type is not None:
-        record["subType"] = info.sub_type.value
+        record["subType"] = info.sub_type
     if info.parent_id is not None:
         record["parentId"] = info.parent_id
     if info.start_time is not None:

packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py

@@ -40,7 +40,7 @@ def _operation_record(hook: str, info: OperationInfo) -> dict[str, Any]:
     if info.name is not None:
         record["name"] = info.name
     if info.sub_type is not None:
-        record["subType"] = info.sub_type.value
+        record["subType"] = info.sub_type
     if info.parent_id is not None:
         record["parentId"] = info.parent_id
     if info.start_time is not None:

packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_parallel_branch_hooks.py

@@ -15,6 +15,7 @@
 from aws_durable_execution_sdk_python.config import ParallelConfig
 from aws_durable_execution_sdk_python.context import DurableContext
 from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.lambda_service import OperationSubType
 from aws_durable_execution_sdk_python.plugin import (
     DurableInstrumentationPlugin,
     InvocationStartInfo,
@@ -33,7 +34,7 @@ def _emit(record: dict[str, Any], execution_arn: str | None) -> None:
 
 
 def _is_branch(info: UserFunctionStartInfo | UserFunctionEndInfo) -> bool:
-    return info.sub_type is not None and info.sub_type.name == "PARALLEL_BRANCH"
+    return info.sub_type == OperationSubType.PARALLEL_BRANCH
 
 
 class ParallelBranchPlugin(DurableInstrumentationPlugin):

packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_timeout.py

@@ -53,7 +53,7 @@ def test_handle_wait_for_callback_timeout_scenarios(durable_runner):
     wait_for_callback_ops = [
         operation
         for operation in result.get_all_operations()
-        if operation.sub_type is OperationSubType.WAIT_FOR_CALLBACK
+        if operation.sub_type == OperationSubType.WAIT_FOR_CALLBACK
     ]
     assert len(wait_for_callback_ops) == 1
     wait_for_callback_error = wait_for_callback_ops[0].error

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

@@ -1058,7 +1058,7 @@ 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"] = 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

@@ -1129,7 +1129,7 @@ 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"] = info.sub_type
         # STEP user-function spans represent attempts, not durable operations.
         if (
             not (

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

@@ -121,7 +121,7 @@ def checkpoint(
                 "Id": update.operation_id,
                 "Type": update.operation_type.value,
                 "Name": update.name,
-                "SubType": update.sub_type.value,
+                "SubType": update.sub_type,
                 "ParentId": update.parent_id,
                 "StartTimestamp": timestamp,
                 "Status": "STARTED" if update.action.value == "START" else "SUCCEEDED",

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

@@ -5,6 +5,7 @@
 
 import pytest
 from aws_durable_execution_sdk_python import plugin as core_plugin
+from aws_durable_execution_sdk_python.lambda_service import OperationSubType
 from aws_durable_execution_sdk_python.plugin import (
     DurableInstrumentationPlugin,
     InvocationStartInfo,
@@ -13,7 +14,6 @@
     OperationStartInfo,
     OperationEndInfo,
     OperationStatus,
-    OperationSubType,
     OperationType,
     PluginExecutor,
     PropagationInput,

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

@@ -79,7 +79,7 @@ def checkpoint(
                 dict(
                     Id=update.operation_id,
                     Type=update.operation_type.value,
-                    SubType=update.sub_type.value,
+                    SubType=update.sub_type,
                     Name=update.name,
                     Status="SUCCEEDED",
                     StartTimestamp=timestamp,

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

@@ -145,6 +145,34 @@ def test_operation_attributes_use_structural_user_function_marker():
     assert "durable.operation.status" not in user_function_attributes
 
 
+@pytest.mark.parametrize(
+    ("sub_type", "expected"),
+    [
+        (OperationSubType.STEP, "Step"),
+        ("OrderCharge", "OrderCharge"),
+        (None, None),
+    ],
+    ids=["sdk-subtype", "custom-subtype", "no-subtype"],
+)
+def test_operation_attributes_record_subtype(sub_type, expected):
+    """A subtype is a str: an SDK member or a caller-defined string."""
+    plugin, _ = _create_plugin()
+
+    attributes = plugin._operation_attributes(
+        SimpleNamespace(
+            operation_id="op-1",
+            operation_type=OperationType.STEP,
+            sub_type=sub_type,
+            status=OperationStatus.STARTED,
+        )
+    )
+
+    if expected is None:
+        assert "durable.operation.subtype" not in attributes
+    else:
+        assert attributes["durable.operation.subtype"] == expected
+
+
 # ---------------------------------------------------------------------------
 # derive_workflow_span_id
 # ---------------------------------------------------------------------------

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

@@ -778,6 +778,45 @@ def test_invocation_end_closes_callback_child_before_parent_context():
     assert callback_span.end_time <= context_span.end_time <= invocation_span.end_time
 
 
+def test_operation_span_records_custom_subtype_as_plain_string():
+    """A caller-defined subtype reaches the span attribute unchanged."""
+    plugin, exporter = _create_plugin()
+    plugin.on_invocation_start(_invocation_start_info())
+
+    plugin.on_operation_start(
+        OperationStartInfo(
+            operation_id="step-1",
+            operation_type=OperationType.STEP,
+            sub_type="OrderCharge",
+            name="charge",
+            parent_id=None,
+            start_time=START_TIME,
+            is_replayed=False,
+            status=OperationStatus.STARTED,
+        )
+    )
+    plugin.on_operation_end(
+        OperationEndInfo(
+            operation_id="step-1",
+            operation_type=OperationType.STEP,
+            sub_type="OrderCharge",
+            name="charge",
+            parent_id=None,
+            start_time=START_TIME,
+            is_replayed=False,
+            status=OperationStatus.SUCCEEDED,
+            end_time=END_TIME,
+            error=None,
+        )
+    )
+    plugin.on_invocation_end(_invocation_end_info())
+
+    spans_by_name = {span.name: span for span in exporter.get_finished_spans()}
+    step_span = spans_by_name["charge"]
+    assert step_span.attributes["durable.operation.subtype"] == "OrderCharge"
+    assert step_span.attributes["durable.operation.type"] == OperationType.STEP.value
+
+
 def test_operation_callbacks_emit_child_span_with_deterministic_span_id():
     """Verify non-user-function operations are traced beneath the invocation."""
     plugin, exporter = _create_plugin()

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

@@ -4,14 +4,14 @@
 from typing import Any
 
 import pytest
+from aws_durable_execution_sdk_python.lambda_service import OperationSubType
 from aws_durable_execution_sdk_python.plugin import (
     InvocationEndInfo,
     InvocationStartInfo,
     InvocationStatus,
     OperationEndInfo,
     OperationStartInfo,
     OperationStatus,
-    OperationSubType,
     OperationType,
     PropagationInput,
 )

packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py

@@ -761,7 +761,7 @@ def _build_operations(
             entry: dict[str, Any] = {"id": op.operation_id, "name": op.name}
             entry["type"] = op.operation_type.value
             if op.sub_type is not None:
-                entry["subType"] = op.sub_type.value
+                entry["subType"] = op.sub_type
             if op.parent_id is not None:
                 entry["parentId"] = op.parent_id
             entry["status"] = op.status.value if op.status is not None else "UNKNOWN"

packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_plugin.py

@@ -266,6 +266,25 @@ def test_basic_success_record():
     assert "result" not in op  # results omitted by default
 
 
+def test_record_writes_subtype():
+    """A subtype lands in the record as the string the service stores."""
+    exporter = CaptureExporter()
+    factory = durable_execution_telemetry(
+        DurableExecutionTelemetryConfig(exporters=[exporter])
+    )
+    _run(
+        factory,
+        ops=[
+            _step("charge", sub_type="OrderCharge"),
+            _step("settle", op_type=OperationType.WAIT, sub_type=OperationSubType.WAIT),
+        ],
+    )
+    ops = {op["name"]: op for op in exporter.records[0]["operations"]}
+    assert ops["charge"]["subType"] == "OrderCharge"
+    assert ops["settle"]["subType"] == "Wait"
+    assert json.dumps(ops["settle"]["subType"]) == '"Wait"'
+
+
 def test_on_failure_success_emits_nothing():
     exporter = CaptureExporter()
     factory = durable_execution_telemetry(

packages/aws-durable-execution-sdk-python-testing/pyproject.toml

@@ -23,7 +23,7 @@ classifiers = [
 ]
 dependencies = [
   "boto3>=1.42.1",
-  "aws-durable-execution-sdk-python>=2.1.0",
+  "aws-durable-execution-sdk-python>=3.0.0",
 ]
 
 [project.urls]
@@ -66,7 +66,7 @@ dependencies = [
   "pytest",
   "pytest-cov",
   "ruff",
-  "aws-durable-execution-sdk-python>=2.1.0",
+  "aws-durable-execution-sdk-python>=3.0.0",
 ]
 
 [tool.hatch.envs.test.scripts]

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

@@ -17,22 +17,17 @@
     CallbackDetails,
     CallbackOptions,
     ChainedInvokeDetails,
-    ChainedInvokeOptions,
     ContextDetails,
-    ContextOptions,
     ErrorObject,
     ExecutionDetails,
     Operation,
     OperationAction,
     OperationStatus,
-    OperationSubType,
     OperationType,
     OperationUpdate,
     StepDetails,
-    StepOptions,
     TimestampConverter,
     WaitDetails,
-    WaitOptions,
 )
 from aws_durable_execution_sdk_python.types import (
     LambdaContext as LambdaContextProtocol,
@@ -1330,7 +1325,7 @@ def create(
 
     @property
     def sub_type(self) -> str | None:
-        return self.operation.sub_type.value if self.operation.sub_type else None
+        return self.operation.sub_type
 
     def get_retry_details(self) -> RetryDetails | None:
         if not self.operation.step_details or not self.operation_update:
@@ -2651,22 +2646,14 @@ def events_to_operations(events: list[Event]) -> list[Operation]:
             event_config.operation_status or OperationStatus.PENDING
         )
 
-        # Parse sub_type
-        sub_type: OperationSubType | None = None
-        if event.sub_type:
-            try:
-                sub_type = OperationSubType(event.sub_type)
-            except ValueError as e:
-                raise InvalidParameterValueException(str(e)) from e
-
         # Create base operation
         operation = Operation(
             operation_id=event.operation_id,
             operation_type=operation_type,
             status=status,
             name=event.name,
             parent_id=event.parent_id,
-            sub_type=sub_type,
+            sub_type=event.sub_type or None,
             start_timestamp=event.event_timestamp,
         )
 
@@ -3161,45 +3148,7 @@ def from_dict(
     ) -> CheckpointDurableExecutionRequest:
         updates = None
         if updates_data := data.get("Updates"):
-            updates = []
-            for update_data in updates_data:
-                # Map dictionary fields to OperationUpdate constructor parameters
-                operation_update = OperationUpdate(
-                    operation_id=update_data["Id"],
-                    operation_type=OperationType(update_data["Type"]),
-                    action=OperationAction(update_data["Action"]),
-                    parent_id=update_data.get("ParentId"),
-                    name=update_data.get("Name"),
-                    sub_type=OperationSubType(update_data["SubType"])
-                    if update_data.get("SubType")
-                    else None,
-                    payload=update_data.get("Payload"),
-                    error=ErrorObject.from_dict(update_data["Error"])
-                    if update_data.get("Error")
-                    else None,
-                    context_options=ContextOptions.from_dict(
-                        update_data["ContextOptions"]
-                    )
-                    if update_data.get("ContextOptions")
-                    else None,
-                    step_options=StepOptions.from_dict(update_data["StepOptions"])
-                    if update_data.get("StepOptions")
-                    else None,
-                    wait_options=WaitOptions.from_dict(update_data["WaitOptions"])
-                    if update_data.get("WaitOptions")
-                    else None,
-                    callback_options=CallbackOptions.from_dict(
-                        update_data["CallbackOptions"]
-                    )
-                    if update_data.get("CallbackOptions")
-                    else None,
-                    chained_invoke_options=ChainedInvokeOptions.from_dict(
-                        update_data["ChainedInvokeOptions"]
-                    )
-                    if update_data.get("ChainedInvokeOptions")
-                    else None,
-                )
-                updates.append(operation_update)
+            updates = [OperationUpdate.from_dict(update) for update in updates_data]
 
         return cls(
             durable_execution_arn=durable_execution_arn,

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

@@ -26,7 +26,6 @@
     ErrorObject,
     OperationPayload,
     OperationStatus,
-    OperationSubType,
     OperationType,
 )
 from aws_durable_execution_sdk_python.lambda_service import Operation as SvcOperation
@@ -152,7 +151,7 @@ class Operation:
     status: OperationStatus
     parent_id: str | None = field(default=None, kw_only=True)
     name: str | None = field(default=None, kw_only=True)
-    sub_type: OperationSubType | None = field(default=None, kw_only=True)
+    sub_type: str | None = field(default=None, kw_only=True)
     start_timestamp: datetime.datetime | None = field(default=None, kw_only=True)
     end_timestamp: datetime.datetime | None = field(default=None, kw_only=True)