From 468307624d43ced0be2736e3241a95a217c528ea Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Wed, 14 Oct 2026 00:07:15 +0000 Subject: [PATCH 1/3] fix: accept string subtypes in companion readers --- .../README.md | 3 + .../execution_plugin.py | 4 +- .../invocation_plugin.py | 4 +- .../tests/test_subtype_compatibility.py | 96 ++++++++++++++++ .../README.md | 8 ++ .../model.py | 40 +++++-- .../runner.py | 2 +- .../tests/e2e/subtype_history_int_test.py | 63 +++++++++++ .../tests/model_test.py | 8 +- .../tests/subtype_compatibility_test.py | 107 ++++++++++++++++++ 10 files changed, 316 insertions(+), 19 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python-otel/tests/test_subtype_compatibility.py create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/e2e/subtype_history_int_test.py create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/subtype_compatibility_test.py diff --git a/packages/aws-durable-execution-sdk-python-otel/README.md b/packages/aws-durable-execution-sdk-python-otel/README.md index 187971ba5..4fdb0c1da 100644 --- a/packages/aws-durable-execution-sdk-python-otel/README.md +++ b/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: diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py index 2008615e1..f6e7e6bff 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py +++ b/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 ( diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py index 4fc2bf02b..294ef2119 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py +++ b/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 ( diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_subtype_compatibility.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_subtype_compatibility.py new file mode 100644 index 000000000..af8e3b92a --- /dev/null +++ b/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 diff --git a/packages/aws-durable-execution-sdk-python-testing/README.md b/packages/aws-durable-execution-sdk-python-testing/README.md index 58b21e793..b4735a2e3 100644 --- a/packages/aws-durable-execution-sdk-python-testing/README.md +++ b/packages/aws-durable-execution-sdk-python-testing/README.md @@ -170,6 +170,14 @@ flow, pagination, and a map of the code. ## Documentation +History projection and checkpoint-request parsing preserve string subtype labels, +including labels not listed in the current core enum. Known labels still parse to +their existing enum members. Public history results expose unknown labels as plain +strings; code inspecting `operation.sub_type` should handle both representations. +Creating, serializing, persisting and replaying operations remains governed by the +installed core SDK's subtype support. This reader compatibility does not add custom +subtype configuration or change the core enum and replay contract. + ### 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: diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py index 43026ec60..e208665e4 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py +++ b/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 @@ -80,6 +80,17 @@ class EventType(Enum): } +def _parse_operation_sub_type(value: str | None) -> OperationSubType | str | None: + if not value: + return None + try: + return OperationSubType(value) + except ValueError: + if not isinstance(value, str): + raise + return value + + @dataclass(frozen=True) class LambdaContext(LambdaContextProtocol): """Lambda context for testing.""" @@ -1330,7 +1341,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: @@ -2652,12 +2667,10 @@ def events_to_operations(events: list[Event]) -> list[Operation]: ) # 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 + try: + sub_type = _parse_operation_sub_type(event.sub_type) + except ValueError as e: + raise InvalidParameterValueException(str(e)) from e # Create base operation operation = Operation( @@ -2666,7 +2679,8 @@ def events_to_operations(events: list[Event]) -> list[Operation]: status=status, name=event.name, parent_id=event.parent_id, - sub_type=sub_type, + # Core models still annotate enums; history readers also retain strings. + sub_type=cast(OperationSubType | None, sub_type), start_timestamp=event.event_timestamp, ) @@ -3170,9 +3184,11 @@ def from_dict( 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, + # The current core constructor annotation is enum-only. + sub_type=cast( + OperationSubType | None, + _parse_operation_sub_type(update_data.get("SubType")), + ), payload=update_data.get("Payload"), error=ErrorObject.from_dict(update_data["Error"]) if update_data.get("Error") diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py index a099652da..544a5fd78 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py @@ -152,7 +152,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: OperationSubType | 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) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/subtype_history_int_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/subtype_history_int_test.py new file mode 100644 index 000000000..895ee880f --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/subtype_history_int_test.py @@ -0,0 +1,63 @@ +"""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", "Library_CustomStep", None]) +def test_history_result_preserves_subtype_and_unrelated_operations(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 + else: + assert type(target.sub_type) is str + assert target.sub_type == label diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py index 746f083d6..9a745b355 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py @@ -3552,9 +3552,9 @@ 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" +@pytest.mark.parametrize("invalid_sub_type", [42, True]) +def test_events_to_operations_invalid_sub_type(invalid_sub_type): + """Non-string subtypes are still rejected rather than retained as labels.""" event = Event( event_type="StepStarted", event_timestamp=datetime.datetime(2023, 1, 1, 0, 0, 0, tzinfo=datetime.UTC), @@ -3564,7 +3564,7 @@ def test_events_to_operations_invalid_sub_type(): with pytest.raises( InvalidParameterValueException, - match=f"'{invalid_sub_type}' is not a valid OperationSubType", + match=f"{invalid_sub_type} is not a valid OperationSubType", ): events_to_operations([event]) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/subtype_compatibility_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/subtype_compatibility_test.py new file mode 100644 index 000000000..49aab97cb --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/subtype_compatibility_test.py @@ -0,0 +1,107 @@ +"""Preserve subtype labels at the testing library's public wire boundaries.""" + +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, +) + + +@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 + (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", "Library_CustomStep", None, ""]) +def test_checkpoint_request_parses_recorded_or_unknown_string_subtype(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 + else: + assert isinstance(update.sub_type, str) + assert update.sub_type == sub_type + + +@pytest.mark.parametrize("sub_type", [42, True]) +def test_checkpoint_request_rejects_non_string_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", + ) From c9c9b60a55afee3d599040f46e38ade14ebd01e1 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Wed, 14 Oct 2026 00:15:11 +0000 Subject: [PATCH 2/3] fix: preserve public subtype type contract --- .../README.md | 12 +++++++----- .../runner.py | 2 +- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-testing/README.md b/packages/aws-durable-execution-sdk-python-testing/README.md index b4735a2e3..1d45259a7 100644 --- a/packages/aws-durable-execution-sdk-python-testing/README.md +++ b/packages/aws-durable-execution-sdk-python-testing/README.md @@ -172,11 +172,13 @@ flow, pagination, and a map of the code. History projection and checkpoint-request parsing preserve string subtype labels, including labels not listed in the current core enum. Known labels still parse to -their existing enum members. Public history results expose unknown labels as plain -strings; code inspecting `operation.sub_type` should handle both representations. -Creating, serializing, persisting and replaying operations remains governed by the -installed core SDK's subtype support. This reader compatibility does not add custom -subtype configuration or change the core enum and replay contract. +their existing enum members; unknown labels are retained as plain strings at these +runtime reading boundaries. Declared public core and testing subtype types remain +unchanged. This prepares readers for future string metadata; it does not promise +typed custom-subtype support. Creating, serializing, persisting and replaying +operations remains governed by the installed core SDK's subtype support. This +reader compatibility does not add custom subtype configuration or change the core +enum and replay contract. ### Error Handling diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py index 544a5fd78..a099652da 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py @@ -152,7 +152,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 | str | None = field(default=None, kw_only=True) + sub_type: OperationSubType | 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) From 1c2158da49050f12546bd5a580df52844103c746 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Wed, 14 Oct 2026 01:20:28 +0000 Subject: [PATCH 3/3] fix: keep subtype parsing within core contracts --- .../README.md | 15 +-- .../model.py | 32 ++--- .../tests/e2e/subtype_history_int_test.py | 7 +- .../tests/model_test.py | 8 +- .../tests/subtype_compatibility_test.py | 21 ++-- .../web/e2e/checkpoint_subtype_int_test.py | 110 ++++++++++++++++++ 6 files changed, 145 insertions(+), 48 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/web/e2e/checkpoint_subtype_int_test.py diff --git a/packages/aws-durable-execution-sdk-python-testing/README.md b/packages/aws-durable-execution-sdk-python-testing/README.md index 1d45259a7..3df614344 100644 --- a/packages/aws-durable-execution-sdk-python-testing/README.md +++ b/packages/aws-durable-execution-sdk-python-testing/README.md @@ -170,15 +170,12 @@ flow, pagination, and a map of the code. ## Documentation -History projection and checkpoint-request parsing preserve string subtype labels, -including labels not listed in the current core enum. Known labels still parse to -their existing enum members; unknown labels are retained as plain strings at these -runtime reading boundaries. Declared public core and testing subtype types remain -unchanged. This prepares readers for future string metadata; it does not promise -typed custom-subtype support. Creating, serializing, persisting and replaying -operations remains governed by the installed core SDK's subtype support. This -reader compatibility does not add custom subtype configuration or change the core -enum and replay contract. +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 diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py index e208665e4..b7e8bd257 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py @@ -80,17 +80,6 @@ class EventType(Enum): } -def _parse_operation_sub_type(value: str | None) -> OperationSubType | str | None: - if not value: - return None - try: - return OperationSubType(value) - except ValueError: - if not isinstance(value, str): - raise - return value - - @dataclass(frozen=True) class LambdaContext(LambdaContextProtocol): """Lambda context for testing.""" @@ -2667,10 +2656,12 @@ def events_to_operations(events: list[Event]) -> list[Operation]: ) # Parse sub_type - try: - sub_type = _parse_operation_sub_type(event.sub_type) - except ValueError as e: - raise InvalidParameterValueException(str(e)) from e + 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( @@ -2679,8 +2670,7 @@ def events_to_operations(events: list[Event]) -> list[Operation]: status=status, name=event.name, parent_id=event.parent_id, - # Core models still annotate enums; history readers also retain strings. - sub_type=cast(OperationSubType | None, sub_type), + sub_type=sub_type, start_timestamp=event.event_timestamp, ) @@ -3184,11 +3174,9 @@ def from_dict( action=OperationAction(update_data["Action"]), parent_id=update_data.get("ParentId"), name=update_data.get("Name"), - # The current core constructor annotation is enum-only. - sub_type=cast( - OperationSubType | None, - _parse_operation_sub_type(update_data.get("SubType")), - ), + 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") diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/subtype_history_int_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/subtype_history_int_test.py index 895ee880f..63dca4d64 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/subtype_history_int_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/subtype_history_int_test.py @@ -16,8 +16,8 @@ from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestResult -@pytest.mark.parametrize("label", ["Step", "Library_CustomStep", None]) -def test_history_result_preserves_subtype_and_unrelated_operations(label): +@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 @@ -58,6 +58,3 @@ def test_history_result_preserves_subtype_and_unrelated_operations(label): assert target.sub_type is OperationSubType.STEP elif label is None: assert target.sub_type is None - else: - assert type(target.sub_type) is str - assert target.sub_type == label diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py index 9a745b355..746f083d6 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py @@ -3552,9 +3552,9 @@ def test_events_to_operations_preserves_sub_type(): assert operations[0].sub_type.value == "Step" -@pytest.mark.parametrize("invalid_sub_type", [42, True]) -def test_events_to_operations_invalid_sub_type(invalid_sub_type): - """Non-string subtypes are still rejected rather than retained as labels.""" +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" event = Event( event_type="StepStarted", event_timestamp=datetime.datetime(2023, 1, 1, 0, 0, 0, tzinfo=datetime.UTC), @@ -3564,7 +3564,7 @@ def test_events_to_operations_invalid_sub_type(invalid_sub_type): with pytest.raises( InvalidParameterValueException, - match=f"{invalid_sub_type} is not a valid OperationSubType", + match=f"'{invalid_sub_type}' is not a valid OperationSubType", ): events_to_operations([event]) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/subtype_compatibility_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/subtype_compatibility_test.py index 49aab97cb..14cbcfe95 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/subtype_compatibility_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/subtype_compatibility_test.py @@ -1,4 +1,4 @@ -"""Preserve subtype labels at the testing library's public wire boundaries.""" +"""Read string metadata without widening current checkpoint/result contracts.""" import datetime @@ -17,6 +17,9 @@ StartDurableExecutionInput, events_to_operations, ) +from aws_durable_execution_sdk_python_testing.exceptions import ( + InvalidParameterValueException, +) @pytest.mark.parametrize( @@ -55,13 +58,18 @@ def test_history_projection_keeps_enum_or_plain_string_subtype(sub_type, expecte 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", "Library_CustomStep", None, ""]) -def test_checkpoint_request_parses_recorded_or_unknown_string_subtype(sub_type): +@pytest.mark.parametrize("sub_type", ["Step", None, ""]) +def test_checkpoint_request_preserves_existing_enum_contract(sub_type): request = CheckpointDurableExecutionRequest.from_dict( { "CheckpointToken": "token", @@ -83,13 +91,10 @@ def test_checkpoint_request_parses_recorded_or_unknown_string_subtype(sub_type): assert update.sub_type is OperationSubType.STEP elif not sub_type: assert update.sub_type is None - else: - assert isinstance(update.sub_type, str) - assert update.sub_type == sub_type -@pytest.mark.parametrize("sub_type", [42, True]) -def test_checkpoint_request_rejects_non_string_subtype(sub_type): +@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( { diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/web/e2e/checkpoint_subtype_int_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/web/e2e/checkpoint_subtype_int_test.py new file mode 100644 index 000000000..d55a6ef46 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/web/e2e/checkpoint_subtype_int_test.py @@ -0,0 +1,110 @@ +"""Unsupported wire subtypes must not consume checkpoint state or tokens.""" + +from threading import Event +from urllib.parse import quote + +import pytest + +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.sqlite import SQLiteExecutionStore +from aws_durable_execution_sdk_python_testing.web.handlers import ( + CheckpointDurableExecutionHandler, +) +from aws_durable_execution_sdk_python_testing.web.models import HTTPRequest +from aws_durable_execution_sdk_python_testing.web.routes import ( + CheckpointDurableExecutionRoute, + Route, +) + + +@pytest.mark.parametrize("store_kind", ["memory", "filesystem", "sqlite"]) +@pytest.mark.parametrize("label", ["Step", "Library_CustomStep"]) +def test_checkpoint_subtype_validation_precedes_mutation(store_kind, label, tmp_path): + if store_kind == "memory": + store = InMemoryExecutionStore() + elif store_kind == "filesystem": + store = FileSystemExecutionStore.create(tmp_path / "store") + else: + store = SQLiteExecutionStore.create_and_initialize(str(tmp_path / "store.db")) + + ready = Event() + release = Event() + inputs = [] + + def handler(event, context): + inputs.append(event) + ready.set() + release.wait(10) + return {"Status": "SUCCEEDED", "Result": '"done"'} + + runner = DurableFunctionTestRunner( + handler=handler, store=store, skip_time=False, execution_timeout=20 + ) + try: + arn = runner.run_async(input="{}") + assert ready.wait(3), "The real invocation must start before checkpointing" + token = inputs[0].checkpoint_token + before = store.load(arn) + before_sequence = before.token_sequence + before_ids = [operation.operation_id for operation in before.operations] + route = CheckpointDurableExecutionRoute.from_route( + Route.from_string( + "/2025-12-01/durable-executions/" + quote(arn, safe="") + "/checkpoint" + ) + ) + # Wire the actual endpoint to the runner's actual executor; the runner + # establishes the invocation gate, worker and checkpoint token itself. + endpoint = CheckpointDurableExecutionHandler(runner._executor) # noqa: SLF001 + response = endpoint.handle( + route, + HTTPRequest( + method="POST", + path=route, + headers={}, + query_params={}, + body={ + "CheckpointToken": token, + "Updates": [ + { + "Id": "target", + "Type": "STEP", + "Action": "START", + "Name": "target", + "SubType": label, + } + ], + }, + ), + ) + after = store.load(arn) + after_ids = [operation.operation_id for operation in after.operations] + if label == "Step": + assert response.status_code == 200 + assert after.token_sequence == before_sequence + 1 + assert after_ids == [*before_ids, "target"] + else: + assert response.status_code == 400 + assert "is not a valid OperationSubType" in str(response.body) + assert after.token_sequence == before_sequence + assert after_ids == before_ids + + retry = endpoint.handle( + route, + HTTPRequest( + method="POST", + path=route, + headers={}, + query_params={}, + body={"CheckpointToken": token, "Updates": []}, + ), + ) + assert retry.status_code == (400 if label == "Step" else 200) + finally: + runner.close() + release.set()