diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py index 0bb85e962..49cabf821 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py +++ b/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: diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_context_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_context_info_shape.py index 9e40bb39e..f727cf241 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_context_info_shape.py +++ b/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: diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py index c1982a333..99c05271f 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py +++ b/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: diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py index 11de1570b..b8a83e5b5 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py +++ b/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: diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_parallel_branch_hooks.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_parallel_branch_hooks.py index f8ff4c016..3d00405e5 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_parallel_branch_hooks.py +++ b/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): diff --git a/packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_timeout.py b/packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_timeout.py index d2cbd3309..cfe1ae536 100644 --- a/packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_timeout.py +++ b/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 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 64f55caf3..c01ef6a7d 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 @@ -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 ( 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 a861c76b6..1b0c39604 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 @@ -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 ( diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_invoke_checkpoint_propagation.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_invoke_checkpoint_propagation.py index 4271eb18e..ed9136cb9 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_invoke_checkpoint_propagation.py +++ b/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", diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_propagation_metadata.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_propagation_metadata.py index 491fbeb64..e34ea80bc 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_propagation_metadata.py +++ b/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, diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_replaced_sampler_invoke.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_replaced_sampler_invoke.py index bd1fb7fa7..584af1e77 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_replaced_sampler_invoke.py +++ b/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, diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py index ffb9ee1c8..6481d36f7 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py index 324d31f06..27547c76c 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py +++ b/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() diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_propagation_sampling.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_propagation_sampling.py index 50c3c0506..343f118f5 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_propagation_sampling.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_propagation_sampling.py @@ -4,6 +4,7 @@ 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, @@ -11,7 +12,6 @@ OperationEndInfo, OperationStartInfo, OperationStatus, - OperationSubType, OperationType, PropagationInput, ) diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py index 41cb5f8cc..b6f4e2660 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py +++ b/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" diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_plugin.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_plugin.py index add2614c1..f8d7b2db1 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_plugin.py +++ b/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( diff --git a/packages/aws-durable-execution-sdk-python-testing/pyproject.toml b/packages/aws-durable-execution-sdk-python-testing/pyproject.toml index b3417262e..78d1ff814 100644 --- a/packages/aws-durable-execution-sdk-python-testing/pyproject.toml +++ b/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] 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 5c82e4d75..3735c7d09 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 @@ -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,14 +2646,6 @@ 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, @@ -2666,7 +2653,7 @@ def events_to_operations(events: list[Event]) -> list[Operation]: 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, 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 10279a593..8c6550747 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 @@ -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) 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..c4194a93c 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 @@ -8,6 +8,7 @@ import pytest from aws_durable_execution_sdk_python.lambda_service import ( OperationStatus, + OperationSubType, OperationType, ) @@ -949,6 +950,28 @@ def test_checkpoint_durable_execution_request_serialization(): assert round_trip == request_obj +def test_checkpoint_durable_execution_request_keeps_custom_sub_type(): + """The local checkpoint endpoint accepts a caller-defined SubType as a string.""" + execution_arn = ( + "arn:aws:lambda:us-east-1:123456789012:function:my-function:execution:test" + ) + data = { + "CheckpointToken": "checkpoint-123", + "Updates": [ + {"Id": "op-1", "Type": "STEP", "Action": "START", "SubType": "OrderCharge"}, + {"Id": "op-2", "Type": "CONTEXT", "Action": "START", "SubType": "Map"}, + ], + } + + request_obj = CheckpointDurableExecutionRequest.from_dict(data, execution_arn) + + assert request_obj.updates is not None + assert request_obj.updates[0].sub_type == "OrderCharge" + assert request_obj.updates[1].sub_type == OperationSubType.MAP + assert type(request_obj.updates[1].sub_type) is str + assert request_obj.to_dict() == {"DurableExecutionArn": execution_arn, **data} + + def test_checkpoint_durable_execution_request_minimal(): """Test CheckpointDurableExecutionRequest with only required fields.""" execution_arn = ( @@ -3548,25 +3571,34 @@ def test_events_to_operations_preserves_sub_type(): operations = events_to_operations([event]) assert len(operations) == 1 - assert operations[0].sub_type is not None - assert operations[0].sub_type.value == "Step" + assert operations[0].sub_type == OperationSubType.STEP + assert type(operations[0].sub_type) is str -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_custom_sub_type(): + """A caller-defined subtype is kept as the string the event carries.""" 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="OrderCharge", ) - 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 == "OrderCharge" + assert type(operations[0].sub_type) is str + + +def test_events_to_operations_treats_empty_sub_type_as_none(): + event = Event( + event_type="StepStarted", + event_timestamp=datetime.datetime(2023, 1, 1, 0, 0, 0, tzinfo=datetime.UTC), + operation_id="step-1", + sub_type="", + ) + assert events_to_operations([event])[0].sub_type is None def test_invocation_completed_details_to_json_dict(): diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/executor.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/executor.py index fd049d3bf..e62c3e1f9 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/executor.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/executor.py @@ -53,7 +53,6 @@ from collections.abc import Callable from aws_durable_execution_sdk_python.config import CompletionConfig from aws_durable_execution_sdk_python.context import DurableContext - from aws_durable_execution_sdk_python.lambda_service import OperationSubType from aws_durable_execution_sdk_python.serdes import SerDes from aws_durable_execution_sdk_python.state import ( CheckpointedResult, @@ -129,8 +128,8 @@ def __init__( executables: list[Executable[CallableType]], max_concurrency: int | None, completion_config: CompletionConfig, - sub_type_top: OperationSubType, - sub_type_iteration: OperationSubType, + sub_type_top: str, + sub_type_iteration: str, name_prefix: str, serdes: SerDes | None, operation_id_namespace: OperationIdNamespace, @@ -702,7 +701,6 @@ def run_in_child_handler() -> ResultType: operation_identifier=operation_identifier, config=ChildConfig( serdes=self.item_serdes or self.serdes, - sub_type=self.sub_type_iteration, is_virtual=is_virtual, ), ) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/config.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/config.py index aa5a2b320..199d4e89f 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/config.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/config.py @@ -4,6 +4,7 @@ import math import random +import string from dataclasses import dataclass, field from enum import Enum, StrEnum from typing import TYPE_CHECKING, Generic, TypeVar @@ -18,7 +19,6 @@ if TYPE_CHECKING: from collections.abc import Callable - from aws_durable_execution_sdk_python.lambda_service import OperationSubType from aws_durable_execution_sdk_python.retries import RetryDecision from aws_durable_execution_sdk_python.serdes import SerDes from aws_durable_execution_sdk_python.types import SummaryGenerator @@ -481,13 +481,69 @@ class StepSemantics(Enum): AT_LEAST_ONCE_PER_RETRY = "AT_LEAST_ONCE_PER_RETRY" +_SUB_TYPE_CHARACTERS = frozenset(string.ascii_letters + string.digits + "-_") +_SUB_TYPE_MAX_LENGTH = 32 + + +def _validate_sub_type(sub_type: str | None) -> None: + """Raise ValidationError when sub_type is set and the service would reject it. + + The service accepts 1 to 32 characters from A-Z, a-z, 0-9, hyphen, and + underscore. None and "" select the SDK default, so both are accepted. + Checking here raises before any checkpoint is sent, so a step body never + runs under a subtype the service will reject. + """ + if not sub_type: + return + if ( + len(sub_type) > _SUB_TYPE_MAX_LENGTH + or not set(sub_type) <= _SUB_TYPE_CHARACTERS + ): + msg = ( + f"sub_type must be 1 to {_SUB_TYPE_MAX_LENGTH} characters from " + f"A-Z, a-z, 0-9, hyphen, and underscore, got: {sub_type!r}" + ) + raise ValidationError(msg) + + @dataclass(frozen=True) class StepConfig: - """Configuration for a step.""" + """Configuration for a step. + + Args: + retry_strategy: Decides whether and when a failed attempt is retried. + If None, the SDK uses its default retry strategy. + + step_semantics: How often one attempt may run the function. + AT_LEAST_ONCE_PER_RETRY (the default) checkpoints START without + waiting, so an interrupted attempt runs again on the next + invocation. AT_MOST_ONCE_PER_RETRY waits for the START checkpoint + first, and an interrupted attempt fails with StepInterruptedError + and goes to the retry strategy instead of running again. + + serdes: Custom serialization/deserialization for the step result. + If None, uses the default JSON serializer. + + sub_type: Subtype recorded in the checkpoint, in replay validation, and + in plugin events. The default is "Step". + A custom operation built from steps uses this to label its steps, + so history and plugins can filter them without parsing names. + The value must be 1 to 32 characters from A-Z, a-z, 0-9, hyphen, + and underscore; the config raises ValidationError otherwise. None + or an empty string selects the default. Do not reuse an + ``OperationSubType`` value: those label the SDK's own operations. + Replay compares the subtype with the checkpoint. So changing the + subtype of an existing step is a breaking change for executions + in flight. + """ retry_strategy: Callable[[Exception, int], RetryDecision] | None = None step_semantics: StepSemantics = StepSemantics.AT_LEAST_ONCE_PER_RETRY serdes: SerDes | None = None + sub_type: str | None = field(default=None, kw_only=True) + + def __post_init__(self) -> None: + _validate_sub_type(self.sub_type) @dataclass(frozen=True) @@ -502,9 +558,18 @@ class ChildConfig(Generic[T]): Applied at the handler level to serialize the entire BatchResult object. If None, uses the default JSON serializer for BatchResult. - sub_type: Operation subtype identifier used for tracking and debugging. - Examples: OperationSubType.MAP_ITERATION, OperationSubType.PARALLEL_BRANCH. - Used internally by the execution engine for operation classification. + sub_type: Subtype recorded in the checkpoint, in replay validation, and + in plugin events. The default is "RunInChildContext". + A custom operation built from several durable operations uses this + to label its child context, so history and plugins can filter it + without parsing names. The value must be 1 to 32 characters from + A-Z, a-z, 0-9, hyphen, and underscore; the config raises + ValidationError otherwise. None or an empty string selects the + default. Do not reuse an ``OperationSubType`` value: those label + the SDK's own operations. + Replay compares the subtype with the checkpoint. So changing the + subtype of an existing child context is a breaking change for + executions in flight. summary_generator: Function generating the checkpoint payload for large results (>256KB). When the serialized result exceeds CHECKPOINT_SIZE_LIMIT, @@ -528,10 +593,13 @@ class ChildConfig(Generic[T]): """ serdes: SerDes | None = None - sub_type: OperationSubType | None = None + sub_type: str | None = None summary_generator: SummaryGenerator | None = None is_virtual: bool = False + def __post_init__(self) -> None: + _validate_sub_type(self.sub_type) + @dataclass(frozen=True) class MapConfig(Generic[T]): @@ -649,11 +717,45 @@ class InvokeConfig(Generic[P, R]): @dataclass(frozen=True) class CallbackConfig: - """Configuration for callbacks.""" + """Configuration for callbacks. + + Args: + timeout: Maximum time to wait for the callback result. The default, + a zero Duration, means no timeout. + + heartbeat_timeout: Maximum time between heartbeats before the callback + fails. The default, a zero Duration, means no heartbeat timeout. + + serdes: Custom deserialization for the callback result. + If None, the payload is returned as is, without JSON parsing. + + sub_type: Subtype recorded in the checkpoint, in replay validation, and + in plugin events. The default is "Callback". + A custom operation built from callbacks uses this to label its + callbacks, so history and plugins can filter them without parsing + names. The value must be 1 to 32 characters from A-Z, a-z, 0-9, + hyphen, and underscore; the config raises ValidationError + otherwise. None or an empty string selects the default. Do not + reuse an ``OperationSubType`` value: those label the SDK's own + operations. Under + ``wait_for_callback`` the subtype labels the callback the + operation creates; the surrounding context keeps the subtype + "WaitForCallback". + Replay compares the subtype with the checkpoint. So changing the + subtype of an existing callback is a breaking change for + executions in flight. + """ timeout: Duration = field(default_factory=Duration) heartbeat_timeout: Duration = field(default_factory=Duration) serdes: SerDes | None = None + # Keyword-only, so the positional order of every config, including + # WaitForCallbackConfig(timeout, heartbeat_timeout, serdes, retry_strategy), + # is unchanged from 2.x. + sub_type: str | None = field(default=None, kw_only=True) + + def __post_init__(self) -> None: + _validate_sub_type(self.sub_type) @property def timeout_seconds(self) -> int: diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py index 9ac766eb5..e1957932e 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py @@ -622,7 +622,7 @@ def _replay_aware( @contextmanager def _operation_replay_aware( self, - sub_type: OperationSubType, + sub_type: str, name: str | None = None, *, operation_type: OperationType, @@ -656,14 +656,16 @@ def create_callback( Args: name (str): Optional name for the operation. config (CallbackConfig): Configuration for the callback. + ``config.sub_type`` labels the operation; see CallbackConfig. Return: - Callback future. Use result() on this future to wait for the callback resuilt. + Callback future. Use result() on this future to wait for the callback result. """ if not config: config = CallbackConfig() + sub_type = config.sub_type or OperationSubType.CALLBACK.value with self._operation_replay_aware( - OperationSubType.CALLBACK, name, operation_type=OperationType.CALLBACK + sub_type, name, operation_type=OperationType.CALLBACK ) as operation_identifier: executor: CallbackOperationExecutor = CallbackOperationExecutor( state=self.state, @@ -699,7 +701,7 @@ def invoke( if not config: config = InvokeConfig[P, R]() with self._operation_replay_aware( - OperationSubType.CHAINED_INVOKE, + OperationSubType.CHAINED_INVOKE.value, name, operation_type=OperationType.CHAINED_INVOKE, ) as operation_identifier: @@ -729,7 +731,7 @@ def map( config.completion_config._validate_for_total(len(inputs)) with self._operation_replay_aware( - OperationSubType.MAP, map_name, operation_type=OperationType.CONTEXT + OperationSubType.MAP.value, map_name, operation_type=OperationType.CONTEXT ) as operation_identifier: operation_id = operation_identifier.operation_id map_context = self.create_child_context(operation_id=operation_id) @@ -754,7 +756,6 @@ def map_in_child_context() -> BatchResult[R]: state=self.state, operation_identifier=operation_identifier, config=ChildConfig( - sub_type=OperationSubType.MAP, serdes=config.serdes if config else None, # The SDK-owned envelope records the completion decision for # replay. A configured summary_generator only contributes the @@ -780,7 +781,7 @@ def parallel( config.completion_config._validate_for_total(len(functions)) with self._operation_replay_aware( - OperationSubType.PARALLEL, name, operation_type=OperationType.CONTEXT + OperationSubType.PARALLEL.value, name, operation_type=OperationType.CONTEXT ) as operation_identifier: operation_id = operation_identifier.operation_id parallel_context = self.create_child_context(operation_id=operation_id) @@ -804,7 +805,6 @@ def parallel_in_child_context() -> BatchResult[T]: state=self.state, operation_identifier=operation_identifier, config=ChildConfig( - sub_type=OperationSubType.PARALLEL, serdes=config.serdes if config else None, # The SDK-owned envelope records the completion decision for # replay. A configured summary_generator only contributes the @@ -827,19 +827,33 @@ def run_in_child_context( Use this to nest and group operations. Args: - callable (Callable[[DurableContext], T]): Run this callable and pass the child context as the argument to it. + func (Callable[[DurableContext], T]): Run this callable and pass the child context as the argument to it. name (str | None): name for the operation. - config (ChildConfig | None = None): c + config (ChildConfig | None = None): Configuration for the child + context. ``config.sub_type`` labels the operation; see + ChildConfig. Returns: T: The result of the callable. """ step_name: str | None = self._resolve_step_name(name, func) sub_type = ( - config.sub_type - if config and config.sub_type - else OperationSubType.RUN_IN_CHILD_CONTEXT - ) + config.sub_type if config else None + ) or OperationSubType.RUN_IN_CHILD_CONTEXT.value + return self._run_in_child_context(func, step_name, config, sub_type) + + def _run_in_child_context( + self, + func: Callable[[DurableContext], T], + step_name: str | None, + config: ChildConfig | None, + sub_type: str, + ) -> T: + """Run ``func`` in a child context that records ``sub_type``. + + ``run_in_child_context`` resolves the caller's ``config.sub_type`` + into ``sub_type``; ``wait_for_callback`` passes its own SDK subtype. + """ is_virtual: bool = config.is_virtual if config else False with self._operation_replay_aware( @@ -870,12 +884,24 @@ def step( name: str | None = None, config: StepConfig | None = None, ) -> T: + """Run a function once and checkpoint its result. + + Args: + func (Callable[[StepContext], T]): The function to run. + name (str | None): Optional name for the operation. + config (StepConfig | None): Configuration for the step. + ``config.sub_type`` labels the operation; see StepConfig. + + Returns: + T: The function's result, from the checkpoint on replay. + """ step_name = self._resolve_step_name(name, func) logger.debug("Step name: %s", step_name) if not config: config = StepConfig() + sub_type = config.sub_type or OperationSubType.STEP.value with self._operation_replay_aware( - OperationSubType.STEP, step_name, operation_type=OperationType.STEP + sub_type, step_name, operation_type=OperationType.STEP ) as operation_identifier: executor: StepOperationExecutor[T] = StepOperationExecutor( func=func, @@ -898,7 +924,7 @@ def wait(self, duration: Duration, name: str | None = None) -> None: msg = "duration must be at least 1 second" raise ValidationError(msg) with self._operation_replay_aware( - OperationSubType.WAIT, name, operation_type=OperationType.WAIT + OperationSubType.WAIT.value, name, operation_type=OperationType.WAIT ) as operation_identifier: wait_seconds = duration.seconds executor: WaitOperationExecutor = WaitOperationExecutor( @@ -924,10 +950,11 @@ def wait_in_child_context(context: DurableContext): # inner failure on __cause__; translate to callback-typed errors here, at # the call site, so it re-fires identically on replay. try: - return self.run_in_child_context( + return self._run_in_child_context( wait_in_child_context, step_name, - ChildConfig(sub_type=OperationSubType.WAIT_FOR_CALLBACK), + None, + OperationSubType.WAIT_FOR_CALLBACK.value, ) except ChildContextError as e: inner: BaseException | None = e.__cause__ @@ -968,7 +995,9 @@ def wait_for_condition( raise ValidationError(msg) with self._operation_replay_aware( - OperationSubType.WAIT_FOR_CONDITION, name, operation_type=OperationType.STEP + OperationSubType.WAIT_FOR_CONDITION.value, + name, + operation_type=OperationType.STEP, ) as operation_identifier: executor: WaitForConditionOperationExecutor[T] = ( WaitForConditionOperationExecutor( diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/identifier.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/identifier.py index 1c58ab179..209d51670 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/identifier.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/identifier.py @@ -11,7 +11,6 @@ from aws_durable_execution_sdk_python.lambda_service import ( Operation, OperationType, - OperationSubType, ) @@ -34,10 +33,15 @@ def create_id_for_step(self, step: int) -> str: @dataclass(frozen=True) class OperationIdentifier: - """Operation identity with an explicit type independent of its subtype.""" + """Operation identity with an explicit type independent of its subtype. + + ``sub_type`` is the subtype the operation records: the value of an + ``OperationSubType`` member for an SDK operation, or the caller-defined + string a step, a callback, or a child context was configured with. + """ operation_id: str - sub_type: OperationSubType + sub_type: str parent_id: str | None = None name: str | None = None operation_type: OperationType = field(kw_only=True) @@ -61,12 +65,9 @@ def validate_checkpoint(self, checkpoint: Operation | None) -> None: mismatches.append( f"type checkpoint={checkpoint.operation_type.value!r} current={self.type.value!r}" ) - if checkpoint.sub_type is not self.sub_type: - checkpoint_sub_type = ( - checkpoint.sub_type.value if checkpoint.sub_type is not None else None - ) + if checkpoint.sub_type != self.sub_type: mismatches.append( - f"subtype checkpoint={checkpoint_sub_type!r} current={self.sub_type.value!r}" + f"subtype checkpoint={checkpoint.sub_type!r} current={self.sub_type!r}" ) if checkpoint_name != expected_name: mismatches.append( diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py index c7c567fa3..233812312 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py @@ -6,7 +6,7 @@ import logging from collections.abc import MutableMapping from dataclasses import dataclass, field -from enum import Enum +from enum import Enum, StrEnum from typing import TYPE_CHECKING, Any, NoReturn, Protocol, TypeAlias, cast import boto3 @@ -93,7 +93,19 @@ class CallbackTimeoutType(Enum): HEARTBEAT = "Callback.Heartbeat" -class OperationSubType(Enum): +class OperationSubType(StrEnum): + """The subtypes the SDK records for its own operations. + + A subtype is a plain ``str``, and every subtype field is typed + ``str | None``. These members catalog the values the SDK writes for its + own operations; the SDK assigns ``.value`` into the field. A step, a + callback, or a child context records a caller-defined string instead when + ``StepConfig.sub_type``, ``CallbackConfig.sub_type``, or + ``ChildConfig.sub_type`` is set. Compare with ``==``: + ``operation.sub_type == OperationSubType.MAP`` is true for the operation + that ``context.map()`` records. + """ + STEP = "Step" WAIT = "Wait" CALLBACK = "Callback" @@ -497,7 +509,7 @@ class OperationUpdate: action: OperationAction parent_id: str | None = None name: str | None = None - sub_type: OperationSubType | None = None + sub_type: str | None = None payload: str | None = None error: ErrorObject | None = None context_options: ContextOptions | None = None @@ -518,7 +530,7 @@ def to_dict(self) -> MutableMapping[str, Any]: if self.name: result["Name"] = self.name if self.sub_type: - result["SubType"] = self.sub_type.value + result["SubType"] = self.sub_type if self.payload is not None: result["Payload"] = self.payload if self.error: @@ -567,7 +579,7 @@ def from_dict(cls, data: MutableMapping[str, Any]) -> OperationUpdate: action=OperationAction(data["Action"]), parent_id=data.get("ParentId"), name=data.get("Name"), - sub_type=OperationSubType(data["SubType"]) if data.get("SubType") else None, + sub_type=data.get("SubType") or None, payload=data.get("Payload"), error=error, context_options=context_options, @@ -586,7 +598,7 @@ def create_callback( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.CALLBACK, - sub_type=OperationSubType.CALLBACK, + sub_type=identifier.sub_type, action=OperationAction.START, name=identifier.name, callback_options=callback_options, @@ -594,15 +606,13 @@ def create_callback( # region context @classmethod - def create_context_start( - cls, identifier: OperationIdentifier, sub_type: OperationSubType - ) -> OperationUpdate: + def create_context_start(cls, identifier: OperationIdentifier) -> OperationUpdate: """Create an instance of OperationUpdate for type: CONTEXT, action: START.""" return cls( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.CONTEXT, - sub_type=sub_type, + sub_type=identifier.sub_type, action=OperationAction.START, name=identifier.name, ) @@ -612,7 +622,7 @@ def create_context_succeed( cls, identifier: OperationIdentifier, payload: str | None, - sub_type: OperationSubType, + *, context_options: ContextOptions | None = None, ) -> OperationUpdate: """Create an instance of OperationUpdate for type: CONTEXT, action: SUCCEED.""" @@ -620,7 +630,7 @@ def create_context_succeed( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.CONTEXT, - sub_type=sub_type, + sub_type=identifier.sub_type, action=OperationAction.SUCCEED, name=identifier.name, payload=payload, @@ -632,14 +642,13 @@ def create_context_fail( cls, identifier: OperationIdentifier, error: ErrorObject, - sub_type: OperationSubType, ) -> OperationUpdate: """Create an instance of OperationUpdate for type: CONTEXT, action: FAIL.""" return cls( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.CONTEXT, - sub_type=sub_type, + sub_type=identifier.sub_type, action=OperationAction.FAIL, name=identifier.name, error=error, @@ -680,7 +689,7 @@ def create_step_succeed( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.STEP, - sub_type=OperationSubType.STEP, + sub_type=identifier.sub_type, action=OperationAction.SUCCEED, name=identifier.name, payload=payload, @@ -695,7 +704,7 @@ def create_step_fail( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.STEP, - sub_type=OperationSubType.STEP, + sub_type=identifier.sub_type, action=OperationAction.FAIL, name=identifier.name, error=error, @@ -708,7 +717,7 @@ def create_step_start(cls, identifier: OperationIdentifier) -> OperationUpdate: operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.STEP, - sub_type=OperationSubType.STEP, + sub_type=identifier.sub_type, action=OperationAction.START, name=identifier.name, ) @@ -725,7 +734,7 @@ def create_step_retry( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.STEP, - sub_type=OperationSubType.STEP, + sub_type=identifier.sub_type, action=OperationAction.RETRY, name=identifier.name, error=error, @@ -749,7 +758,7 @@ def create_invoke_start( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.CHAINED_INVOKE, - sub_type=OperationSubType.CHAINED_INVOKE, + sub_type=identifier.sub_type, action=OperationAction.START, name=identifier.name, payload=payload, @@ -768,7 +777,7 @@ def create_wait_for_condition_start( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.STEP, - sub_type=OperationSubType.WAIT_FOR_CONDITION, + sub_type=identifier.sub_type, action=OperationAction.START, name=identifier.name, ) @@ -782,7 +791,7 @@ def create_wait_for_condition_succeed( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.STEP, - sub_type=OperationSubType.WAIT_FOR_CONDITION, + sub_type=identifier.sub_type, action=OperationAction.SUCCEED, name=identifier.name, payload=payload, @@ -800,7 +809,7 @@ def create_wait_for_condition_retry( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.STEP, - sub_type=OperationSubType.WAIT_FOR_CONDITION, + sub_type=identifier.sub_type, action=OperationAction.RETRY, name=identifier.name, payload=payload, @@ -818,7 +827,7 @@ def create_wait_for_condition_fail( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.STEP, - sub_type=OperationSubType.WAIT_FOR_CONDITION, + sub_type=identifier.sub_type, action=OperationAction.FAIL, name=identifier.name, error=error, @@ -836,7 +845,7 @@ def create_wait_start( operation_id=identifier.operation_id, parent_id=identifier.parent_id, operation_type=OperationType.WAIT, - sub_type=OperationSubType.WAIT, + sub_type=identifier.sub_type, action=OperationAction.START, name=identifier.name, wait_options=wait_options, @@ -874,7 +883,7 @@ class Operation: name: str | None = None start_timestamp: datetime.datetime | None = None end_timestamp: datetime.datetime | None = None - sub_type: OperationSubType | None = None + sub_type: str | None = None execution_details: ExecutionDetails | None = None context_details: ContextDetails | None = None step_details: StepDetails | None = None @@ -895,9 +904,7 @@ def from_dict(cls, data: MutableMapping[str, Any]) -> Operation: operation_type = OperationType(data.get("Type")) operation_status = OperationStatus(data.get("Status")) - sub_type = None - if sub_type_input := data.get("SubType"): - sub_type = OperationSubType(sub_type_input) + sub_type = data.get("SubType") or None execution_details = None if execution_details_input := data.get("ExecutionDetails"): @@ -957,7 +964,7 @@ def to_dict(self) -> MutableMapping[str, Any]: if self.end_timestamp: result["EndTimestamp"] = self.end_timestamp if self.sub_type: - result["SubType"] = self.sub_type.value + result["SubType"] = self.sub_type if self.execution_details: result["ExecutionDetails"] = { "InputPayload": self.execution_details.input_payload diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py index f29262673..03a0d3fc8 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py @@ -18,7 +18,6 @@ ErrorObject, Operation, OperationStatus, - OperationSubType, OperationUpdate, ) from aws_durable_execution_sdk_python.constants import CHECKPOINT_SIZE_LIMIT_BYTES @@ -71,7 +70,6 @@ def __init__( self.operation_identifier = operation_identifier self.config = config self.is_virtual: bool = config.is_virtual - self.sub_type = config.sub_type or OperationSubType.RUN_IN_CHILD_CONTEXT @staticmethod def validate_virtual_checkpoint( @@ -151,7 +149,6 @@ def check_result_status(self) -> CheckResult[T]: if not checkpointed_result.is_existent() and not self.is_virtual: start_operation: OperationUpdate = OperationUpdate.create_context_start( identifier=self.operation_identifier, - sub_type=self.sub_type, ) # Checkpoint child context START with non-blocking (is_sync=False). # This is a fire-and-forget operation for performance - we don't need to wait for @@ -282,7 +279,6 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: success_operation: OperationUpdate = OperationUpdate.create_context_succeed( identifier=self.operation_identifier, payload=payload_to_checkpoint, - sub_type=self.sub_type, context_options=ContextOptions(replay_children=replay_children), ) # Checkpoint child context SUCCEED with blocking (is_sync=True, default). @@ -319,7 +315,6 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: fail_operation: OperationUpdate = OperationUpdate.create_context_fail( identifier=self.operation_identifier, error=error_object, - sub_type=self.sub_type, ) # Checkpoint child context FAIL with blocking (is_sync=True, default). # Must ensure the failure state is persisted before raising the exception. diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/map.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/map.py index ad99d3b5b..29722c7af 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/map.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/map.py @@ -39,8 +39,8 @@ def __init__( items: Sequence[T], max_concurrency: int | None, completion_config, - top_level_sub_type: OperationSubType, - iteration_sub_type: OperationSubType, + top_level_sub_type: str, + iteration_sub_type: str, name_prefix: str, serdes: SerDes | None, operation_id_namespace: OperationIdNamespace, @@ -92,8 +92,8 @@ def run(child_context: DurableContext) -> R: items=items, max_concurrency=config.max_concurrency, completion_config=config.completion_config, - top_level_sub_type=OperationSubType.MAP, - iteration_sub_type=OperationSubType.MAP_ITERATION, + top_level_sub_type=OperationSubType.MAP.value, + iteration_sub_type=OperationSubType.MAP_ITERATION.value, name_prefix="map-item-", serdes=config.serdes, operation_id_namespace=operation_id_namespace, diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/parallel.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/parallel.py index 5aad4da04..31e676dbb 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/parallel.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/parallel.py @@ -38,8 +38,8 @@ def __init__( executables: list[Executable[Callable]], max_concurrency: int | None, completion_config, - top_level_sub_type: OperationSubType, - iteration_sub_type: OperationSubType, + top_level_sub_type: str, + iteration_sub_type: str, name_prefix: str, serdes: SerDes | None, operation_id_namespace: OperationIdNamespace, @@ -84,8 +84,8 @@ def from_callables( executables=executables, max_concurrency=config.max_concurrency, completion_config=config.completion_config, - top_level_sub_type=OperationSubType.PARALLEL, - iteration_sub_type=OperationSubType.PARALLEL_BRANCH, + top_level_sub_type=OperationSubType.PARALLEL.value, + iteration_sub_type=OperationSubType.PARALLEL_BRANCH.value, name_prefix="parallel-branch-", serdes=config.serdes, operation_id_namespace=operation_id_namespace, diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py index fd72dae91..43b578f36 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py @@ -20,7 +20,6 @@ Operation, OperationAction, OperationStatus, - OperationSubType, OperationType as ServiceOperationType, OperationUpdate, ) @@ -89,7 +88,11 @@ def _extract_error(operation: Operation) -> ErrorObject | None: class OperationInfo: operation_id: str operation_type: OperationType - sub_type: OperationSubType | None + sub_type: str | None + """The recorded subtype as a plain ``str``, or None for the execution root. + Compare it with ``==`` against an ``OperationSubType`` member, which + equals its value, or against a caller-defined string. Do not read + ``.value`` or compare with ``is``.""" name: str | None parent_id: str | None start_time: datetime.datetime | None diff --git a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py index bf0c96ebc..1aaae311c 100644 --- a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py @@ -2233,7 +2233,7 @@ def patched_child_handler( ): """Patched child handler that captures operation_id -> result mapping.""" assert config.is_virtual - assert config.sub_type == "TEST_ITER" + assert operation_identifier.sub_type == "TEST_ITER" result = func() # Execute the function captured_associations.append((operation_identifier.operation_id, result)) return result diff --git a/packages/aws-durable-execution-sdk-python/tests/config_test.py b/packages/aws-durable-execution-sdk-python/tests/config_test.py index 5f230974a..c2fbc78b2 100644 --- a/packages/aws-durable-execution-sdk-python/tests/config_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/config_test.py @@ -16,6 +16,7 @@ ParallelConfig, StepConfig, StepSemantics, + WaitForCallbackConfig, ) from aws_durable_execution_sdk_python.exceptions import ValidationError from aws_durable_execution_sdk_python.waits import ( @@ -145,10 +146,39 @@ def test_child_config_with_serdes(): def test_child_config_with_sub_type(): """Test ChildConfig with sub_type.""" - sub_type = Mock() - config = ChildConfig(sub_type=sub_type) + config = ChildConfig(sub_type="OrderSaga") assert config.serdes is None - assert config.sub_type is sub_type + assert config.sub_type == "OrderSaga" + + +def test_step_config_sub_type_defaults_to_none(): + assert StepConfig().sub_type is None + assert StepConfig(sub_type="OrderCharge").sub_type == "OrderCharge" + + +def test_callback_config_sub_type_defaults_to_none(): + assert CallbackConfig().sub_type is None + assert CallbackConfig(sub_type="Approval").sub_type == "Approval" + # WaitForCallbackConfig inherits the field and labels the inner callback. + assert WaitForCallbackConfig(sub_type="Approval").sub_type == "Approval" + + +def test_sub_type_is_keyword_only_so_positional_order_is_unchanged(): + """2.x positional construction keeps its meaning; sub_type never takes a slot.""" + retry = Mock() + serdes = Mock() + config = WaitForCallbackConfig( + Duration.from_seconds(1), Duration.from_seconds(2), serdes, retry + ) + assert config.retry_strategy is retry + assert config.sub_type is None + assert ( + StepConfig(retry, StepSemantics.AT_MOST_ONCE_PER_RETRY, serdes).serdes is serdes + ) + # ChildConfig.sub_type keeps its 2.x positional slot; only its type changed. + assert ChildConfig(serdes, "OrderSaga").sub_type == "OrderSaga" + with pytest.raises(TypeError): + StepConfig(retry, StepSemantics.AT_LEAST_ONCE_PER_RETRY, serdes, "OrderCharge") # type: ignore[misc] def test_child_config_with_summary_generator(): @@ -279,3 +309,35 @@ def test_completion_decision_rejects_outcome_when_not_complete(): # endregion Config validation + + +# --- sub_type validation --------------------------------------------------- + + +@pytest.mark.parametrize( + "sub_type", + ["OrderCharge", "a", "A" * 32, "with-hyphen_and_underscore9", None, ""], + ids=["custom", "one-char", "max-length", "all-classes", "none", "empty"], +) +def test_configs_accept_a_valid_or_default_sub_type(sub_type: str | None) -> None: + assert StepConfig(sub_type=sub_type).sub_type == sub_type + assert CallbackConfig(sub_type=sub_type).sub_type == sub_type + assert ChildConfig(sub_type=sub_type).sub_type == sub_type + + +@pytest.mark.parametrize( + "sub_type", + ["A" * 33, "Order Charge", "order.charge", "order/charge", "ordér"], + ids=["too-long", "space", "dot", "slash", "non-ascii"], +) +@pytest.mark.parametrize("config_class", [StepConfig, CallbackConfig, ChildConfig]) +def test_configs_reject_a_sub_type_the_service_rejects( + config_class: type, sub_type: str +) -> None: + with pytest.raises(ValidationError, match="1 to 32 characters"): + config_class(sub_type=sub_type) + + +def test_sub_type_rejection_names_the_value() -> None: + with pytest.raises(ValidationError, match=r"got: 'Order Charge'"): + StepConfig(sub_type="Order Charge") diff --git a/packages/aws-durable-execution-sdk-python/tests/context_test.py b/packages/aws-durable-execution-sdk-python/tests/context_test.py index 623fee84e..beb9df4ad 100644 --- a/packages/aws-durable-execution-sdk-python/tests/context_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/context_test.py @@ -1238,7 +1238,7 @@ def test_run_in_child_context_with_name_and_config(mock_handler): mock_callable = Mock() mock_callable._original_name = "original_function" # noqa: SLF001 - config = ChildConfig(sub_type=OperationSubType.STEP) + config = ChildConfig(sub_type="OrderSaga") context = create_test_context(state=mock_state) [context._create_step_id() for _ in range(3)] # Set counter to 3 # noqa: SLF001 @@ -1253,7 +1253,7 @@ def test_run_in_child_context_with_name_and_config(mock_handler): call_args = mock_handler.call_args assert call_args[1]["operation_identifier"] == OperationIdentifier( expected_id, - OperationSubType.STEP, + "OrderSaga", None, "original_function", operation_type=OperationType.CONTEXT, @@ -1261,6 +1261,93 @@ def test_run_in_child_context_with_name_and_config(mock_handler): assert call_args[1]["config"] is config +# region custom subtypes + + +@pytest.mark.parametrize("sub_type", [None, ""], ids=["none", "empty"]) +@patch("aws_durable_execution_sdk_python.context.child_handler") +def test_run_in_child_context_default_subtype(mock_handler, sub_type): + """None and the empty string select the default, as in the JS SDK.""" + mock_handler.return_value = "child" + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + context = create_test_context(state=mock_state) + + context.run_in_child_context( + Mock(), name="child", config=ChildConfig(sub_type=sub_type) + ) + + identifier = mock_handler.call_args[1]["operation_identifier"] + assert identifier.sub_type == OperationSubType.RUN_IN_CHILD_CONTEXT + + +@patch("aws_durable_execution_sdk_python.context.StepOperationExecutor") +def test_step_custom_subtype_reaches_identifier(mock_executor_class): + mock_executor_class.return_value.process.return_value = "stepped" + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + context = create_test_context(state=mock_state) + + result = context.step( + Mock(), name="charge", config=StepConfig(sub_type="OrderCharge") + ) + + assert result == "stepped" + identifier = mock_executor_class.call_args[1]["operation_identifier"] + assert identifier.sub_type == "OrderCharge" + assert identifier.type is OperationType.STEP + + +@pytest.mark.parametrize("sub_type", [None, ""], ids=["none", "empty"]) +@patch("aws_durable_execution_sdk_python.context.StepOperationExecutor") +def test_step_default_subtype(mock_executor_class, sub_type): + mock_executor_class.return_value.process.return_value = "stepped" + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + context = create_test_context(state=mock_state) + + context.step(Mock(), name="charge", config=StepConfig(sub_type=sub_type)) + + identifier = mock_executor_class.call_args[1]["operation_identifier"] + assert identifier.sub_type == OperationSubType.STEP + + +@patch("aws_durable_execution_sdk_python.context.CallbackOperationExecutor") +def test_create_callback_custom_subtype_reaches_identifier(mock_executor_class): + mock_executor_class.return_value.process.return_value = "callback-id" + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + context = create_test_context(state=mock_state) + + callback = context.create_callback( + name="approval", config=CallbackConfig(sub_type="Approval") + ) + + assert callback.callback_id == "callback-id" + identifier = mock_executor_class.call_args[1]["operation_identifier"] + assert identifier.sub_type == "Approval" + assert identifier.type is OperationType.CALLBACK + + +@patch("aws_durable_execution_sdk_python.context.StepOperationExecutor") +def test_step_with_rejected_subtype_never_runs_the_body(mock_executor_class): + """The config raises before the step is built, so no START checkpoint is + sent and the step body never runs under a subtype the service rejects.""" + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + context = create_test_context(state=mock_state) + body = Mock() + + with pytest.raises(ValidationError, match="1 to 32 characters"): + context.step(body, name="charge", config=StepConfig(sub_type="Order Charge")) + + body.assert_not_called() + mock_executor_class.assert_not_called() + + +# endregion custom subtypes + + @patch("aws_durable_execution_sdk_python.context.child_handler") def test_run_in_child_context_with_parent_id(mock_executor_class): """Test run_in_child_context with parent_id.""" @@ -1419,7 +1506,7 @@ def test_wait_for_callback_basic(mock_executor_class): mock_submitter._original_name # noqa: SLF001 ) # Ensure _original_name doesn't exist - with patch.object(DurableContext, "run_in_child_context") as mock_run_in_child: + with patch.object(DurableContext, "_run_in_child_context") as mock_run_in_child: mock_run_in_child.return_value = "callback_result" context = create_test_context(state=mock_state) @@ -1431,7 +1518,8 @@ def test_wait_for_callback_basic(mock_executor_class): # Verify the child context callable call_args = mock_run_in_child.call_args assert call_args[0][1] is None # name should be None - assert call_args[0][2].sub_type is OperationSubType.WAIT_FOR_CALLBACK + assert call_args[0][2] is None # no ChildConfig: the SDK sets the subtype + assert call_args[0][3] == OperationSubType.WAIT_FOR_CALLBACK @patch("aws_durable_execution_sdk_python.context.wait_for_callback_handler") @@ -1450,7 +1538,7 @@ def test_wait_for_callback_with_name_and_config(mock_executor_class): mock_submitter._original_name = "submit_function" # noqa: SLF001 config = CallbackConfig() - with patch.object(DurableContext, "run_in_child_context") as mock_run_in_child: + with patch.object(DurableContext, "_run_in_child_context") as mock_run_in_child: mock_run_in_child.return_value = "configured_callback_result" context = create_test_context(state=mock_state) @@ -1461,7 +1549,8 @@ def test_wait_for_callback_with_name_and_config(mock_executor_class): assert ( call_args[0][1] == "submit_function" ) # name should be from _original_name - assert call_args[0][2].sub_type is OperationSubType.WAIT_FOR_CALLBACK + assert call_args[0][2] is None # no ChildConfig: the SDK sets the subtype + assert call_args[0][3] == OperationSubType.WAIT_FOR_CALLBACK @patch("aws_durable_execution_sdk_python.context.wait_for_callback_handler") @@ -1479,7 +1568,7 @@ def test_wait_for_callback_resolves_name_from_submitter(mock_executor_class): mock_submitter = Mock() mock_submitter._original_name = "submit_task" # noqa: SLF001 - with patch.object(DurableContext, "run_in_child_context") as mock_run_in_child: + with patch.object(DurableContext, "_run_in_child_context") as mock_run_in_child: mock_run_in_child.return_value = "named_callback_result" context = create_test_context(state=mock_state) @@ -1487,7 +1576,8 @@ def test_wait_for_callback_resolves_name_from_submitter(mock_executor_class): call_args = mock_run_in_child.call_args assert call_args[0][1] == "submit_task" - assert call_args[0][2].sub_type is OperationSubType.WAIT_FOR_CALLBACK + assert call_args[0][2] is None # no ChildConfig: the SDK sets the subtype + assert call_args[0][3] == OperationSubType.WAIT_FOR_CALLBACK @patch("aws_durable_execution_sdk_python.context.wait_for_callback_handler") @@ -1506,11 +1596,12 @@ def capture_handler_call(context, submitter, name, config): mock_executor_class.side_effect = capture_handler_call - with patch.object(DurableContext, "run_in_child_context") as mock_run_in_child: + with patch.object(DurableContext, "_run_in_child_context") as mock_run_in_child: - def run_child_context(callable_func, name, config): + def run_child_context(callable_func, name, config, sub_type): # Execute the child context callable - assert config.sub_type is OperationSubType.WAIT_FOR_CALLBACK + assert config is None + assert sub_type == OperationSubType.WAIT_FOR_CALLBACK child_context = create_test_context(state=mock_state, parent_id="test") return callable_func(child_context) @@ -1552,7 +1643,7 @@ def test_wait_for_callback_passes_callback_errors_through(inner): with patch.object( DurableContext, - "run_in_child_context", + "_run_in_child_context", side_effect=_child_context_error_with_cause(inner), ): with pytest.raises(CallbackError) as exc_info: @@ -1569,7 +1660,7 @@ def test_wait_for_callback_translates_submitter_step_failure(): inner = StepError("submitter blew up", data="payload", stack_trace=["frame"]) with patch.object( DurableContext, - "run_in_child_context", + "_run_in_child_context", side_effect=_child_context_error_with_cause(inner), ): with pytest.raises(CallbackSubmitterError) as exc_info: @@ -1592,7 +1683,7 @@ def test_wait_for_callback_reraises_unrelated_child_context_error(): child_error = _child_context_error_with_cause(inner) with patch.object( DurableContext, - "run_in_child_context", + "_run_in_child_context", side_effect=child_error, ): with pytest.raises(ChildContextError) as exc_info: @@ -1630,7 +1721,8 @@ def test_function(context, item, index, items): # Verify the child handler was called with correct parameters call_args = mock_handler.call_args - assert call_args[1]["config"].sub_type.value == "Map" + assert call_args[1]["operation_identifier"].sub_type == OperationSubType.MAP + assert call_args[1]["config"].sub_type is None @patch("aws_durable_execution_sdk_python.context.child_handler") @@ -1762,7 +1854,8 @@ def task2(context): # Verify the child handler was called with correct parameters call_args = mock_handler.call_args - assert call_args[1]["config"].sub_type.value == "Parallel" + assert call_args[1]["operation_identifier"].sub_type == OperationSubType.PARALLEL + assert call_args[1]["config"].sub_type is None @patch("aws_durable_execution_sdk_python.context.child_handler") diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/custom_subtype_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/custom_subtype_int_test.py new file mode 100644 index 000000000..b93570ce0 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/custom_subtype_int_test.py @@ -0,0 +1,165 @@ +"""Custom subtypes on steps, callbacks, and child contexts. + +A function that composes several durable operations can label each part with +its own subtype. The subtype reaches the checkpoint, the execution history, +and plugin events, and replay compares it with the checkpoint. +""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python.config import ( + CallbackConfig, + ChildConfig, + Duration, + StepConfig, + WaitForCallbackConfig, +) +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 ( + OperationStatus, + OperationSubType, +) +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationEndInfo, + InvocationStartInfo, + InvocationStatus as PluginInvocationStatus, + OperationEndInfo, + OperationStartInfo, + OperationType, +) +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from tests.test_helpers import plugin_factory + + +class _RecordingPlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + self.invocation_starts: list[InvocationStartInfo] = [] + self.invocation_ends: list[InvocationEndInfo] = [] + self.operation_starts: list[OperationStartInfo] = [] + self.operation_ends: list[OperationEndInfo] = [] + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self.invocation_starts.append(info) + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + self.invocation_ends.append(info) + + def on_operation_start(self, info: OperationStartInfo) -> None: + self.operation_starts.append(info) + + def on_operation_end(self, info: OperationEndInfo) -> None: + self.operation_ends.append(info) + + +def test_custom_subtypes_survive_suspend_and_replay() -> None: + """Each part of a composed operation records its own subtype. + + The wait inside the child context suspends the execution, so the second + invocation replays every operation against its checkpoint. + """ + plugin = _RecordingPlugin() + step_calls: list[str] = [] + + def charge(_step_context: Any) -> str: + step_calls.append("charged") + return "receipt" + + def order_saga(child: DurableContext) -> str: + receipt = child.step( + charge, name="charge", config=StepConfig(sub_type="OrderCharge") + ) + callback = child.create_callback( + name="approval", config=CallbackConfig(sub_type="OrderApproval") + ) + approval = callback.result() + child.wait(Duration.from_seconds(60), name="settle") + return f"{receipt}:{approval}" + + def handler(_event: Any, context: DurableContext) -> str: + return context.run_in_child_context( + order_saga, name="saga", config=ChildConfig(sub_type="OrderSaga") + ) + + wrapped = durable_execution(handler, plugins=[plugin_factory(plugin)]) + with DurableFunctionTestRunner(handler=wrapped) as runner: + arn = runner.run_async(input="{}") + callback_id = runner.wait_for_callback(arn, name="approval", timeout=60) + runner.send_callback_success(callback_id, b"approved") + result = runner.wait_for_result(execution_arn=arn, timeout=120) + history = runner.get_execution_history(arn) + + assert result.result == '"receipt:approved"' + assert step_calls == ["charged"] + assert len(plugin.invocation_starts) >= 2, "the callback must suspend the execution" + assert plugin.invocation_ends[-1].status is PluginInvocationStatus.SUCCEEDED + + # The testing runner reads the recorded subtype as a plain string. + saga = result.get_context("saga") + assert saga.status is OperationStatus.SUCCEEDED + assert saga.sub_type == "OrderSaga" + by_name = {operation.name: operation for operation in saga.child_operations} + assert by_name["charge"].sub_type == "OrderCharge" + assert by_name["approval"].sub_type == "OrderApproval" + assert by_name["settle"].sub_type == OperationSubType.WAIT + + # The execution history carries the subtype on every event of the operation. + def history_sub_types(name: str) -> set[str | None]: + return {event.sub_type for event in history.events if event.name == name} + + assert history_sub_types("saga") == {"OrderSaga"} + assert history_sub_types("charge") == {"OrderCharge"} + assert history_sub_types("approval") == {"OrderApproval"} + + # Plugin events report the subtype, with the operation type unchanged. + expected = { + "saga": ("OrderSaga", OperationType.CONTEXT), + "charge": ("OrderCharge", OperationType.STEP), + "approval": ("OrderApproval", OperationType.CALLBACK), + "settle": (OperationSubType.WAIT, OperationType.WAIT), + } + for info in [*plugin.operation_starts, *plugin.operation_ends]: + assert info.name is not None + sub_type, operation_type = expected[info.name] + assert info.sub_type == sub_type, info + assert info.operation_type is operation_type, info + + # The replaying invocation receives every recorded subtype as a plain string. + replayed = plugin.invocation_starts[-1].operations.values() + replayed_by_name = {info.name: info for info in replayed if info.name} + assert replayed_by_name["saga"].sub_type == "OrderSaga" + assert replayed_by_name["charge"].sub_type == "OrderCharge" + assert replayed_by_name["approval"].sub_type == "OrderApproval" + assert replayed_by_name["settle"].sub_type == OperationSubType.WAIT + assert type(replayed_by_name["saga"].sub_type) is str + + +def test_wait_for_callback_subtype_labels_the_inner_callback() -> None: + """The callback inside wait_for_callback takes the configured subtype. + + The surrounding context keeps the subtype "WaitForCallback", and the + submitter step keeps "Step". + """ + + def handler(_event: Any, context: DurableContext) -> str: + return context.wait_for_callback( + lambda callback_id, _ctx: runner.send_callback_success( + callback_id, b"approved" + ), + name="approval", + config=WaitForCallbackConfig(sub_type="OrderApproval"), + ) + + wrapped = durable_execution(handler) + with DurableFunctionTestRunner(handler=wrapped) as runner: + result = runner.run(input="{}", execution_timeout=120) + + assert result.result == '"approved"' + context_operation = result.get_context("approval") + assert context_operation.sub_type == OperationSubType.WAIT_FOR_CALLBACK + by_name = {op.name: op for op in context_operation.child_operations} + assert by_name["approval create callback id"].sub_type == "OrderApproval" + assert by_name["approval submitter"].sub_type == OperationSubType.STEP diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_operation_type_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_operation_type_int_test.py index ffabc11bf..2762e2eb3 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_operation_type_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_operation_type_int_test.py @@ -107,7 +107,7 @@ def handler(_event: Any, context: DurableContext) -> str: ) assert plugin.invocation_ends[-1].status is InvocationStatus.SUCCEEDED - expected = { + expected: dict[str, OperationType] = { OperationSubType.STEP: OperationType.STEP, OperationSubType.WAIT_FOR_CONDITION: OperationType.STEP, OperationSubType.WAIT: OperationType.WAIT, @@ -156,7 +156,7 @@ def handler(_event: Any, context: DurableContext) -> str: return context.run_in_child_context( child, name="custom-subtype-child", - config=ChildConfig(sub_type=OperationSubType.STEP, is_virtual=is_virtual), + config=ChildConfig(sub_type="OrderSaga", is_virtual=is_virtual), ) wrapped = durable_execution(handler, plugins=[plugin_factory(plugin)]) @@ -187,7 +187,7 @@ def handler(_event: Any, context: DurableContext) -> str: assert completed for info in [*starts, *ends, *completed]: assert info.operation_type is OperationType.CONTEXT - assert info.sub_type is OperationSubType.STEP + assert info.sub_type == "OrderSaga" assert info.operation_id == starts[0].operation_id snapshots = [ @@ -202,4 +202,4 @@ def handler(_event: Any, context: DurableContext) -> str: assert snapshots for info in snapshots: assert info.operation_type is OperationType.CONTEXT - assert info.sub_type is OperationSubType.STEP + assert info.sub_type == "OrderSaga" diff --git a/packages/aws-durable-execution-sdk-python/tests/execution_test.py b/packages/aws-durable-execution-sdk-python/tests/execution_test.py index 0cf360f34..10258bad6 100644 --- a/packages/aws-durable-execution-sdk-python/tests/execution_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/execution_test.py @@ -14,6 +14,7 @@ import pytest +from aws_durable_execution_sdk_python.concurrency.models import BatchResult from aws_durable_execution_sdk_python.config import ( ChildConfig, MapConfig, @@ -3264,8 +3265,8 @@ def test_handler(event: Any, context: DurableContext) -> str: mock_client.checkpoint.assert_not_called() -def test_durable_execution_replays_child_context_with_step_subtype(): - """A CONTEXT/Step checkpoint remains valid for an unchanged child replay.""" +def test_durable_execution_replays_child_context_with_custom_subtype(): + """A CONTEXT checkpoint with a custom subtype replays an unchanged child.""" mock_client = Mock(spec=DurableServiceClient) child_id = OperationIdNamespace().create_id_for_step(1) execution_operation = Operation( @@ -3278,7 +3279,7 @@ def test_durable_execution_replays_child_context_with_step_subtype(): operation_id=child_id, operation_type=OperationType.CONTEXT, status=OperationStatus.SUCCEEDED, - sub_type=OperationSubType.STEP, + sub_type="OrderSaga", name="child", context_details=ContextDetails(result=json.dumps("cached")), ) @@ -3302,7 +3303,7 @@ def test_handler(event: Any, context: DurableContext) -> str: return context.run_in_child_context( child_body, name="child", - config=ChildConfig(sub_type=OperationSubType.STEP), + config=ChildConfig(sub_type="OrderSaga"), ) result = test_handler(invocation_input, _make_lambda_context()) @@ -3311,6 +3312,105 @@ def test_handler(event: Any, context: DurableContext) -> str: assert child_body_calls == [] +def test_durable_execution_replays_state_recorded_by_core_2x(): + """State loaded from the service holds plain-string SDK subtypes. + + Core 2.x recorded only SDK subtypes. Replay compares them with the member + the handler puts on each identifier, so a 2.x execution resumes under 3.0 + without re-running any completed body. + """ + mock_client = Mock(spec=DurableServiceClient) + root = OperationIdNamespace() + child_id = root.create_id_for_step(1) + map_id = root.create_id_for_step(2) + iteration_id = OperationIdNamespace(map_id).create_id_for_step(0) + step_id = root.create_id_for_step(3) + + def op(**kwargs: Any) -> Operation: + return Operation.from_dict(kwargs) + + initial = [ + op( + Id="exec1", + Type="EXECUTION", + Status="STARTED", + ExecutionDetails={"InputPayload": "{}"}, + ), + op( + Id=child_id, + Type="CONTEXT", + Status="SUCCEEDED", + SubType="RunInChildContext", + Name="child", + ContextDetails={"Result": json.dumps("child")}, + ), + op( + Id=map_id, + Type="CONTEXT", + Status="SUCCEEDED", + SubType="Map", + Name="map", + ContextDetails={ + "Result": json.dumps({"MapResult": {"Items": [], "Summary": None}}), + "ReplayChildren": True, + }, + ), + op( + Id=iteration_id, + Type="CONTEXT", + Status="SUCCEEDED", + SubType="MapIteration", + Name="map-item-0", + ParentId=map_id, + ContextDetails={"Result": json.dumps(2)}, + ), + op( + Id=step_id, + Type="STEP", + Status="SUCCEEDED", + SubType="Step", + Name="step", + StepDetails={"Result": json.dumps("stepped")}, + ), + ] + for operation in initial: + assert operation.sub_type is None or type(operation.sub_type) is str + + invocation_input = DurableExecutionInvocationInputWithClient( + durable_execution_arn="arn:test:execution/exec1", + checkpoint_token="token123", # noqa: S106 + initial_execution_state=InitialExecutionState( + operations=initial, next_marker="" + ), + service_client=mock_client, + ) + body_calls: list[str] = [] + + def child_body(_child: DurableContext) -> str: + body_calls.append("child") + return "child" + + def map_item(_child: DurableContext, item: int, _index: int, _items: Any) -> int: + body_calls.append("item") + return item * 2 + + def step_body(_step: Any) -> str: + body_calls.append("step") + return "stepped" + + @durable_execution + def test_handler(event: Any, context: DurableContext) -> str: + child = context.run_in_child_context(child_body, name="child") + items: BatchResult[int] = context.map([1], map_item, name="map") + step = context.step(step_body, name="step") + return f"{child}:{items.get_results()}:{step}" + + result = test_handler(invocation_input, _make_lambda_context()) + + assert result["Status"] == InvocationStatus.SUCCEEDED.value, result + assert body_calls == [] + + @pytest.mark.parametrize( "child_status", [OperationStatus.SUCCEEDED, OperationStatus.FAILED, OperationStatus.STARTED], diff --git a/packages/aws-durable-execution-sdk-python/tests/identifier_test.py b/packages/aws-durable-execution-sdk-python/tests/identifier_test.py index 8d24787ff..3d742801f 100644 --- a/packages/aws-durable-execution-sdk-python/tests/identifier_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/identifier_test.py @@ -54,3 +54,81 @@ def test_explicit_operation_type_controls_replay_validation( def test_operation_identifier_requires_explicit_operation_type() -> None: with pytest.raises(TypeError, match="operation_type"): OperationIdentifier("operation", OperationSubType.STEP) # type: ignore[call-arg] + + +def _checkpoint(sub_type: str | None) -> Operation: + """A checkpoint as the SDK loads it: the subtype is the stored string.""" + return Operation( + operation_id="operation", + operation_type=OperationType.STEP, + sub_type=sub_type, + status=OperationStatus.SUCCEEDED, + ) + + +@pytest.mark.parametrize( + "sub_type", + [OperationSubType.STEP, "OrderCharge"], + ids=["sdk-subtype", "custom-subtype"], +) +def test_replay_accepts_checkpoint_with_same_subtype(sub_type: str) -> None: + identifier = OperationIdentifier( + "operation", sub_type, operation_type=OperationType.STEP + ) + identifier.validate_checkpoint(_checkpoint(sub_type)) + + +def test_replay_accepts_loaded_sdk_subtype_against_member() -> None: + """A loaded "Step" string equals the member the SDK puts on the identifier.""" + identifier = OperationIdentifier( + "operation", OperationSubType.STEP, operation_type=OperationType.STEP + ) + identifier.validate_checkpoint(_checkpoint("Step")) + + +def test_replay_mismatch_message_prints_sdk_subtype_as_its_value() -> None: + """The SDK assigns .value, so the message shows 'Step' like any other string.""" + identifier = OperationIdentifier( + "operation", OperationSubType.STEP.value, operation_type=OperationType.STEP + ) + with pytest.raises(NonDeterministicExecutionError) as exc_info: + identifier.validate_checkpoint(_checkpoint("OrderCharge")) + assert "subtype checkpoint='OrderCharge' current='Step'" in str(exc_info.value) + + +@pytest.mark.parametrize( + ("current", "recorded", "expected_message"), + [ + ( + "OrderCharge", + "Step", + "subtype checkpoint='Step' current='OrderCharge'", + ), + ( + "Step", + "OrderCharge", + "subtype checkpoint='OrderCharge' current='Step'", + ), + ( + "OrderCharge", + "OrderNotify", + "subtype checkpoint='OrderNotify' current='OrderCharge'", + ), + ( + "OrderCharge", + None, + "subtype checkpoint=None current='OrderCharge'", + ), + ], + ids=["custom-vs-sdk", "sdk-vs-custom", "custom-vs-custom", "custom-vs-none"], +) +def test_replay_rejects_changed_subtype( + current: str, recorded: str | None, expected_message: str +) -> None: + """Changing a subtype is non-deterministic; the message shows plain values.""" + identifier = OperationIdentifier( + "operation", current, operation_type=OperationType.STEP + ) + with pytest.raises(NonDeterministicExecutionError) as exc_info: + identifier.validate_checkpoint(_checkpoint(recorded)) + assert expected_message in str(exc_info.value) diff --git a/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py b/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py index 8ae3d6111..757036d75 100644 --- a/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py @@ -1,6 +1,7 @@ """Tests for the service module.""" import datetime +import json from datetime import UTC from unittest.mock import Mock, patch @@ -796,7 +797,7 @@ def test_operation_update_create_callback(): assert update.action is OperationAction.START assert update.name == "test_callback" assert update.callback_options == callback_options - assert update.sub_type is OperationSubType.CALLBACK + assert update.sub_type == OperationSubType.CALLBACK def test_operation_update_create_wait_start(): @@ -818,7 +819,7 @@ def test_operation_update_create_wait_start(): assert update.action is OperationAction.START assert update.name == "test_wait" assert update.wait_options == wait_options - assert update.sub_type is OperationSubType.WAIT + assert update.sub_type == OperationSubType.WAIT @patch("aws_durable_execution_sdk_python.lambda_service.datetime") @@ -852,7 +853,7 @@ def test_operation_update_create_step_succeed(): assert update.action is OperationAction.SUCCEED assert update.name == "test_step" assert update.payload == "step_payload" - assert update.sub_type is OperationSubType.STEP + assert update.sub_type == OperationSubType.STEP def test_operation_update_to_dict_includes_empty_payload(): @@ -889,11 +890,10 @@ def test_operation_update_factory_methods(): "test_context", operation_type=OperationType.CONTEXT, ), - OperationSubType.RUN_IN_CHILD_CONTEXT, ) assert update.operation_type is OperationType.CONTEXT assert update.action is OperationAction.START - assert update.sub_type is OperationSubType.RUN_IN_CHILD_CONTEXT + assert update.sub_type == OperationSubType.RUN_IN_CHILD_CONTEXT # Test create_context_succeed update = OperationUpdate.create_context_succeed( @@ -905,11 +905,10 @@ def test_operation_update_factory_methods(): operation_type=OperationType.CONTEXT, ), "payload", - OperationSubType.RUN_IN_CHILD_CONTEXT, ) assert update.action is OperationAction.SUCCEED assert update.payload == "payload" - assert update.sub_type is OperationSubType.RUN_IN_CHILD_CONTEXT + assert update.sub_type == OperationSubType.RUN_IN_CHILD_CONTEXT # Test create_context_fail update = OperationUpdate.create_context_fail( @@ -921,11 +920,10 @@ def test_operation_update_factory_methods(): operation_type=OperationType.CONTEXT, ), error, - OperationSubType.RUN_IN_CHILD_CONTEXT, ) assert update.action is OperationAction.FAIL assert update.error == error - assert update.sub_type is OperationSubType.RUN_IN_CHILD_CONTEXT + assert update.sub_type == OperationSubType.RUN_IN_CHILD_CONTEXT # Test create_execution_fail update = OperationUpdate.create_execution_fail(error) @@ -945,7 +943,7 @@ def test_operation_update_factory_methods(): ) assert update.operation_type is OperationType.STEP assert update.action is OperationAction.FAIL - assert update.sub_type is OperationSubType.STEP + assert update.sub_type == OperationSubType.STEP # Test create_step_start update = OperationUpdate.create_step_start( @@ -958,7 +956,7 @@ def test_operation_update_factory_methods(): ) ) assert update.action is OperationAction.START - assert update.sub_type is OperationSubType.STEP + assert update.sub_type == OperationSubType.STEP # Test create_step_retry update = OperationUpdate.create_step_retry( @@ -974,7 +972,7 @@ def test_operation_update_factory_methods(): ) assert update.action is OperationAction.RETRY assert update.step_options.next_attempt_delay_seconds == 30 - assert update.sub_type is OperationSubType.STEP + assert update.sub_type == OperationSubType.STEP def test_operation_update_with_parent_id(): @@ -1676,7 +1674,8 @@ def test_operation_from_dict_with_subtype(): assert operation.operation_id == "op1" assert operation.operation_type is OperationType.STEP assert operation.status is OperationStatus.SUCCEEDED - assert operation.sub_type is OperationSubType.STEP + assert operation.sub_type == OperationSubType.STEP + assert type(operation.sub_type) is str def test_operation_from_dict_complete(): @@ -1710,7 +1709,8 @@ def test_operation_from_dict_complete(): assert operation.name == "test_step" assert operation.start_timestamp == start_time assert operation.end_timestamp == end_time - assert operation.sub_type is OperationSubType.STEP + assert operation.sub_type == OperationSubType.STEP + assert type(operation.sub_type) is str assert operation.execution_details.input_payload == "exec_payload" assert operation.context_details.result == "context_result" assert operation.step_details.result == "step_result" @@ -3084,3 +3084,127 @@ def test_operation_to_dict_omits_absent_context_error_and_replay_children(): context = op.to_dict()["ContextDetails"] assert context == {"Result": '"hello"'} + + +# region subtypes + + +def test_operation_sub_type_members_are_strings(): + """Members compare equal to their service value and serialize as it.""" + assert OperationSubType.MAP == "Map" + assert isinstance(OperationSubType.MAP, str) + assert json.dumps({"SubType": OperationSubType.MAP}) == '{"SubType": "Map"}' + + +@pytest.mark.parametrize( + ("raw", "expected"), + [("Step", "Step"), ("OrderCharge", "OrderCharge"), ("", None), (None, None)], + ids=["sdk-value", "custom-value", "empty", "missing"], +) +def test_operation_from_dict_keeps_sub_type_as_plain_string(raw, expected): + """A loaded subtype is the string the service stores; it is never an enum.""" + data = {"Id": "op1", "Type": "STEP", "Status": "SUCCEEDED"} + if raw is not None: + data["SubType"] = raw + operation = Operation.from_dict(data) + assert operation.sub_type == expected + if expected is not None: + assert type(operation.sub_type) is str + # An SDK value still compares equal to its member. + if raw == "Step": + assert operation.sub_type == OperationSubType.STEP + + +def test_operation_from_dict_keeps_custom_sub_type(): + """An execution that recorded a custom subtype loads its own state.""" + operation = Operation.from_dict( + { + "Id": "op1", + "Type": "STEP", + "Status": "SUCCEEDED", + "SubType": "OrderCharge", + "Name": "charge", + } + ) + assert operation.sub_type == "OrderCharge" + assert operation.to_dict()["SubType"] == "OrderCharge" + + +def test_operation_to_dict_writes_sdk_sub_type_as_its_value(): + """A member is a str, so it serializes as the value the service stores.""" + operation = Operation( + operation_id="op1", + operation_type=OperationType.STEP, + status=OperationStatus.SUCCEEDED, + sub_type=OperationSubType.STEP, + ) + sub_type = operation.to_dict()["SubType"] + assert sub_type == "Step" + assert json.dumps(sub_type) == '"Step"' + + +def test_operation_update_round_trips_custom_sub_type(): + identifier = OperationIdentifier( + "step1", "OrderCharge", None, "charge", operation_type=OperationType.STEP + ) + update = OperationUpdate.create_step_start(identifier) + assert update.sub_type == "OrderCharge" + serialized = update.to_dict() + assert serialized["SubType"] == "OrderCharge" + assert OperationUpdate.from_dict(serialized).sub_type == "OrderCharge" + + +@pytest.mark.parametrize( + ("factory", "args"), + [ + (OperationUpdate.create_step_start, ()), + (OperationUpdate.create_step_succeed, ("payload",)), + ( + OperationUpdate.create_step_fail, + (ErrorObject(message="m", type="T", data=None, stack_trace=None),), + ), + ( + OperationUpdate.create_step_retry, + (ErrorObject(message="m", type="T", data=None, stack_trace=None), 1), + ), + (OperationUpdate.create_callback, (CallbackOptions(),)), + (OperationUpdate.create_context_start, ()), + (OperationUpdate.create_context_succeed, ("payload",)), + ( + OperationUpdate.create_context_fail, + (ErrorObject(message="m", type="T", data=None, stack_trace=None),), + ), + (OperationUpdate.create_wait_start, (WaitOptions(wait_seconds=1),)), + ( + OperationUpdate.create_invoke_start, + ("payload", ChainedInvokeOptions(function_name="fn")), + ), + (OperationUpdate.create_wait_for_condition_start, ()), + (OperationUpdate.create_wait_for_condition_succeed, ("payload",)), + (OperationUpdate.create_wait_for_condition_retry, ("payload", 1)), + ( + OperationUpdate.create_wait_for_condition_fail, + (ErrorObject(message="m", type="T", data=None, stack_trace=None),), + ), + ], +) +def test_operation_update_factories_record_identifier_sub_type(factory, args): + """Every factory records the identifier's subtype, custom or SDK.""" + identifier = OperationIdentifier( + "op1", "Custom", None, "name", operation_type=OperationType.STEP + ) + assert factory(identifier, *args).sub_type == "Custom" + + +def test_create_context_succeed_rejects_the_removed_positional_sub_type(): + """A 2.x call that passed sub_type positionally fails at the call site.""" + identifier = OperationIdentifier( + "ctx1", "RunInChildContext", None, "ctx", operation_type=OperationType.CONTEXT + ) + with pytest.raises(TypeError): + OperationUpdate.create_context_succeed( + identifier, "payload", OperationSubType.RUN_IN_CHILD_CONTEXT + ) # type: ignore[misc] + + +# endregion subtypes diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py index a99e7f19e..5cbe57b39 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py @@ -59,20 +59,19 @@ def deserialize(self, data: str, _: SerDesContext) -> dict[str, Any]: # region child_handler +# The executor records the identifier's subtype. run_in_child_context resolves +# config.sub_type into the identifier before the executor runs, so the config +# value is never read here: a config carrying a different value proves that. @pytest.mark.parametrize( - ("config", "expected_sub_type"), + ("config", "sub_type"), [ - ( - ChildConfig(sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT), - OperationSubType.RUN_IN_CHILD_CONTEXT, - ), - (ChildConfig(sub_type=OperationSubType.STEP), OperationSubType.STEP), + (ChildConfig(), OperationSubType.RUN_IN_CHILD_CONTEXT), + (ChildConfig(sub_type="Ignored"), "OrderSaga"), (None, OperationSubType.RUN_IN_CHILD_CONTEXT), ], + ids=["sdk-subtype", "custom-subtype", "no-config"], ) -def test_child_handler_not_started( - config: ChildConfig | None, expected_sub_type: OperationSubType -): +def test_child_handler_not_started(config: ChildConfig | None, sub_type: str): """Test child_handler when operation not started. Verifies: @@ -97,7 +96,7 @@ def test_child_handler_not_started( mock_state, OperationIdentifier( "op1", - OperationSubType.RUN_IN_CHILD_CONTEXT, + sub_type, None, "test_name", operation_type=OperationType.CONTEXT, @@ -120,7 +119,7 @@ def test_child_handler_not_started( assert start_operation.operation_id == "op1" assert start_operation.name == "test_name" assert start_operation.operation_type is OperationType.CONTEXT - assert start_operation.sub_type is expected_sub_type + assert start_operation.sub_type == sub_type assert start_operation.action is OperationAction.START # CRITICAL: Verify is_sync=False for START checkpoint (async, no immediate response) assert start_call[1]["is_sync"] is False @@ -131,7 +130,7 @@ def test_child_handler_not_started( assert success_operation.operation_id == "op1" assert success_operation.name == "test_name" assert success_operation.operation_type is OperationType.CONTEXT - assert success_operation.sub_type is expected_sub_type + assert success_operation.sub_type == sub_type assert success_operation.action is OperationAction.SUCCEED assert success_operation.payload == json.dumps("fresh_result") @@ -244,20 +243,19 @@ def test_child_handler_already_failed(): assert mock_state.get_checkpoint_result.call_count == 1 +# The executor records the identifier's subtype. run_in_child_context resolves +# config.sub_type into the identifier before the executor runs, so the config +# value is never read here: a config carrying a different value proves that. @pytest.mark.parametrize( - ("config", "expected_sub_type"), + ("config", "sub_type"), [ - ( - ChildConfig(sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT), - OperationSubType.RUN_IN_CHILD_CONTEXT, - ), - (ChildConfig(sub_type=OperationSubType.STEP), OperationSubType.STEP), + (ChildConfig(), OperationSubType.RUN_IN_CHILD_CONTEXT), + (ChildConfig(sub_type="Ignored"), "OrderSaga"), (None, OperationSubType.RUN_IN_CHILD_CONTEXT), ], + ids=["sdk-subtype", "custom-subtype", "no-config"], ) -def test_child_handler_already_started( - config: ChildConfig | None, expected_sub_type: OperationSubType -): +def test_child_handler_already_started(config: ChildConfig | None, sub_type: str): """Test child_handler when operation already started. Verifies: @@ -281,7 +279,7 @@ def test_child_handler_already_started( mock_state, OperationIdentifier( "op5", - OperationSubType.RUN_IN_CHILD_CONTEXT, + sub_type, None, "test_name", operation_type=OperationType.CONTEXT, @@ -301,27 +299,26 @@ def test_child_handler_already_started( assert success_operation.operation_id == "op5" assert success_operation.name == "test_name" assert success_operation.operation_type is OperationType.CONTEXT - assert success_operation.sub_type == expected_sub_type + assert success_operation.sub_type == sub_type assert success_operation.action is OperationAction.SUCCEED assert success_operation.payload == json.dumps("started_result") mock_callable.assert_called_once() +# The executor records the identifier's subtype. run_in_child_context resolves +# config.sub_type into the identifier before the executor runs, so the config +# value is never read here: a config carrying a different value proves that. @pytest.mark.parametrize( - ("config", "expected_sub_type"), + ("config", "sub_type"), [ - ( - ChildConfig(sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT), - OperationSubType.RUN_IN_CHILD_CONTEXT, - ), - (ChildConfig(sub_type=OperationSubType.STEP), OperationSubType.STEP), + (ChildConfig(), OperationSubType.RUN_IN_CHILD_CONTEXT), + (ChildConfig(sub_type="Ignored"), "OrderSaga"), (None, OperationSubType.RUN_IN_CHILD_CONTEXT), ], + ids=["sdk-subtype", "custom-subtype", "no-config"], ) -def test_child_handler_callable_exception( - config: ChildConfig | None, expected_sub_type: OperationSubType -): +def test_child_handler_callable_exception(config: ChildConfig | None, sub_type: str): """Test child_handler when callable raises exception. Verifies: @@ -346,7 +343,7 @@ def test_child_handler_callable_exception( mock_state, OperationIdentifier( "op6", - OperationSubType.RUN_IN_CHILD_CONTEXT, + sub_type, None, "test_name", operation_type=OperationType.CONTEXT, @@ -367,7 +364,7 @@ def test_child_handler_callable_exception( assert start_operation.operation_id == "op6" assert start_operation.name == "test_name" assert start_operation.operation_type is OperationType.CONTEXT - assert start_operation.sub_type is expected_sub_type + assert start_operation.sub_type == sub_type assert start_operation.action is OperationAction.START assert start_call[1]["is_sync"] is False @@ -377,7 +374,7 @@ def test_child_handler_callable_exception( assert fail_operation.operation_id == "op6" assert fail_operation.name == "test_name" assert fail_operation.operation_type is OperationType.CONTEXT - assert fail_operation.sub_type is expected_sub_type + assert fail_operation.sub_type == sub_type assert fail_operation.action is OperationAction.FAIL # The checkpoint records the escaping error as-is (its own type); the # ChildContextError wrapper is raised and recorded one level up. diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py index 507dc4200..08ffae984 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py @@ -284,7 +284,7 @@ def test_step_handler_success_at_least_once(): start_operation = start_call[1]["operation_update"] assert start_operation.operation_id == "step6" assert start_operation.operation_type is OperationType.STEP - assert start_operation.sub_type is OperationSubType.STEP + assert start_operation.sub_type == OperationSubType.STEP assert start_operation.action is OperationAction.START # Verify success checkpoint @@ -293,7 +293,7 @@ def test_step_handler_success_at_least_once(): assert success_operation.operation_id == "step6" assert success_operation.payload == json.dumps("success_result") assert success_operation.operation_type is OperationType.STEP - assert success_operation.sub_type is OperationSubType.STEP + assert success_operation.sub_type == OperationSubType.STEP assert success_operation.action is OperationAction.SUCCEED @@ -399,7 +399,7 @@ def test_step_handler_success_at_most_once(): assert start_operation.operation_id == "step7" assert start_operation.name == "test_step" assert start_operation.operation_type is OperationType.STEP - assert start_operation.sub_type is OperationSubType.STEP + assert start_operation.sub_type == OperationSubType.STEP assert start_operation.action is OperationAction.START # Verify success checkpoint @@ -407,7 +407,7 @@ def test_step_handler_success_at_most_once(): success_operation = success_call[1]["operation_update"] assert success_operation.payload == json.dumps("success_result") assert success_operation.operation_type is OperationType.STEP - assert success_operation.sub_type is OperationSubType.STEP + assert success_operation.sub_type == OperationSubType.STEP assert success_operation.action is OperationAction.SUCCEED @@ -477,7 +477,7 @@ def test_step_handler_retry_success(): start_operation = start_call[1]["operation_update"] assert start_operation.operation_id == "step9" assert start_operation.operation_type is OperationType.STEP - assert start_operation.sub_type is OperationSubType.STEP + assert start_operation.sub_type == OperationSubType.STEP assert start_operation.action is OperationAction.START # Verify retry checkpoint @@ -485,7 +485,7 @@ def test_step_handler_retry_success(): retry_operation = retry_call[1]["operation_update"] assert retry_operation.operation_id == "step9" assert retry_operation.operation_type is OperationType.STEP - assert retry_operation.sub_type is OperationSubType.STEP + assert retry_operation.sub_type == OperationSubType.STEP assert retry_operation.action is OperationAction.RETRY @@ -527,7 +527,7 @@ def test_step_handler_retry_exhausted(): start_operation = start_call[1]["operation_update"] assert start_operation.operation_id == "step10" assert start_operation.operation_type is OperationType.STEP - assert start_operation.sub_type is OperationSubType.STEP + assert start_operation.sub_type == OperationSubType.STEP assert start_operation.action is OperationAction.START # Verify fail checkpoint @@ -535,7 +535,7 @@ def test_step_handler_retry_exhausted(): fail_operation = fail_call[1]["operation_update"] assert fail_operation.operation_id == "step10" assert fail_operation.operation_type is OperationType.STEP - assert fail_operation.sub_type is OperationSubType.STEP + assert fail_operation.sub_type == OperationSubType.STEP assert fail_operation.action is OperationAction.FAIL # The checkpoint records the raw escaping error's type; the StepError wrapper # is surfaced to the caller and recorded one level up. @@ -1445,7 +1445,7 @@ def test_step_creates_start_checkpoint_when_status_is_ready(): start_operation = start_call[1]["operation_update"] assert start_operation.operation_id == "step_ready_1" assert start_operation.operation_type is OperationType.STEP - assert start_operation.sub_type is OperationSubType.STEP + assert start_operation.sub_type == OperationSubType.STEP assert start_operation.action is OperationAction.START # Verify SUCCEED checkpoint was also created after execution diff --git a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py index acad8b9db..cb277e4a5 100644 --- a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py @@ -391,6 +391,35 @@ def test_operation_info_converts_service_operation_type(self): self.assertIs(info.operation_type, OperationType.WAIT) + def test_operation_info_carries_custom_subtype_as_plain_string(self): + """A caller-defined subtype reaches plugins as the recorded string.""" + operation = Operation( + operation_id="step-1", + operation_type=ServiceOperationType.STEP, + status=OperationStatus.SUCCEEDED, + sub_type="OrderCharge", + ) + + info = OperationInfo.from_operation(operation) + + self.assertEqual(info.sub_type, "OrderCharge") + self.assertIs(type(info.sub_type), str) + self.assertIs(info.operation_type, OperationType.STEP) + + def test_operation_info_carries_sdk_subtype_member(self): + """An SDK subtype compares with its member and with its string value.""" + operation = Operation( + operation_id="map-1", + operation_type=ServiceOperationType.CONTEXT, + status=OperationStatus.SUCCEEDED, + sub_type=OperationSubType.MAP, + ) + + info = OperationInfo.from_operation(operation) + + self.assertEqual(info.sub_type, OperationSubType.MAP) + self.assertEqual(info.sub_type, "Map") + def test_invocation_end_info_converts_service_invocation_status(self): output = DurableExecutionInvocationOutput( status=ServiceInvocationStatus.PENDING,