fix: validate checkpoint SubType constraints

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

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import re
 from typing import TYPE_CHECKING, ClassVar
 
 from aws_durable_execution_sdk_python.lambda_service import (
@@ -43,6 +44,7 @@
 )
 from aws_durable_execution_sdk_python_testing.exceptions import (
     InvalidParameterValueException,
+    ValidationException,
 )
 
 
@@ -52,6 +54,9 @@
     from aws_durable_execution_sdk_python_testing.execution import Execution
 
 MAX_ERROR_PAYLOAD_SIZE_BYTES = 1048576
+SUBTYPE_PATTERN = "[a-zA-Z0-9-_]+"
+_SUBTYPE_RE = re.compile(SUBTYPE_PATTERN)
+_MAX_VALIDATION_ERRORS = 10
 
 
 class CheckpointValidator:
@@ -74,12 +79,51 @@ def validate_input(updates: list[OperationUpdate], execution: Execution) -> None
         if not updates:
             return
 
+        # Reject service-model violations before any stateful validation or writes.
+        CheckpointValidator._validate_subtypes(updates)
         CheckpointValidator._validate_conflicting_execution_update(updates)
         CheckpointValidator._validate_parent_id_and_duplicate_id(updates, execution)
 
         for update in updates:
             CheckpointValidator._validate_operation_update(update, execution)
 
+    @staticmethod
+    def _validate_subtypes(updates: list[OperationUpdate]) -> None:
+        """Validate SubType constraints before checkpoint state is mutated.
+
+        The service lists up to ten errors in update order. The empty string
+        violates both the minimum length and the required pattern.
+        """
+        errors: list[str] = []
+        for index, update in enumerate(updates, start=1):
+            sub_type = update.sub_type
+            if sub_type is None:
+                continue
+            prefix = (
+                f"Value '{sub_type}' at 'updates.{index}.member.subType' "
+                "failed to satisfy constraint:"
+            )
+            if len(sub_type) < 1:
+                errors.append(
+                    f"{prefix} Member must have length greater than or equal to 1"
+                )
+            if len(sub_type) > 32:
+                errors.append(
+                    f"{prefix} Member must have length less than or equal to 32"
+                )
+            if _SUBTYPE_RE.fullmatch(sub_type) is None:
+                errors.append(
+                    f"{prefix} Member must satisfy regular expression pattern: {SUBTYPE_PATTERN}"
+                )
+
+        if errors:
+            reported = errors[:_MAX_VALIDATION_ERRORS]
+            count = len(reported)
+            noun = "error" if count == 1 else "errors"
+            raise ValidationException(
+                f"{count} validation {noun} detected: {'; '.join(reported)}"
+            )
+
     @staticmethod
     def _validate_conflicting_execution_update(updates: list[OperationUpdate]) -> None:
         """Validate that there are no conflicting execution updates."""

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

@@ -2,6 +2,10 @@
 
 import datetime
 
+from aws_durable_execution_sdk_python.exceptions import (
+    CheckpointError,
+    DurableApiErrorCategory,
+)
 from aws_durable_execution_sdk_python.lambda_service import (
     CheckpointOutput,
     DurableServiceClient,
@@ -10,6 +14,7 @@
 )
 
 from aws_durable_execution_sdk_python_testing.clock import real_now
+from aws_durable_execution_sdk_python_testing.exceptions import ValidationException
 from aws_durable_execution_sdk_python_testing.checkpoint.processor import (
     CheckpointProcessor,
 )
@@ -41,7 +46,13 @@ def checkpoint(
         task = CheckpointTask(
             self._checkpoint_processor, checkpoint_token, updates, client_token
         )
-        return self._registry.submit(durable_execution_arn, task).result()
+        try:
+            return self._registry.submit(durable_execution_arn, task).result()
+        except ValidationException as exc:
+            # Match the boto client: service-model 400s terminate the execution.
+            raise CheckpointError(
+                str(exc), error_category=DurableApiErrorCategory.EXECUTION
+            ) from exc
 
     def get_execution_state(
         self,

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

@@ -153,6 +153,20 @@ def to_dict(self) -> dict[str, Any]:
         return {"Type": "InvalidParameterValueException", "message": self.message}
 
 
+class ValidationException(AwsApiException):
+    """Service-model constraint error returned by CheckpointDurableExecution."""
+
+    http_status_code = 400
+
+    def __init__(self, message: str) -> None:
+        self.message = message
+        super().__init__(message)
+
+    def to_dict(self) -> dict[str, Any]:
+        # Botocore recognizes the unmodeled ValidationException via __type.
+        return {"__type": "ValidationException", "message": self.message}
+
+
 class ResourceNotFoundException(AwsApiException):
     """Exception for resource not found errors."""
 

packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/validators/subtype_validation_test.py

@@ -0,0 +1,162 @@
+"""Regression coverage for CheckpointDurableExecution SubType validation (#836)."""
+
+from datetime import UTC, datetime
+from unittest.mock import Mock
+
+import pytest
+from aws_durable_execution_sdk_python.lambda_service import (
+    OperationAction,
+    OperationType,
+    OperationUpdate,
+)
+
+from aws_durable_execution_sdk_python_testing.checkpoint.core import CheckpointCore
+from aws_durable_execution_sdk_python_testing.checkpoint.validators.checkpoint import (
+    CheckpointValidator,
+)
+from aws_durable_execution_sdk_python_testing.exceptions import AwsApiException
+from aws_durable_execution_sdk_python_testing.execution import Execution
+from aws_durable_execution_sdk_python_testing.model import StartDurableExecutionInput
+from aws_durable_execution_sdk_python_testing.web.models import HTTPResponse
+
+
+PATTERN = "[a-zA-Z0-9-_]+"
+
+
+def _execution() -> Execution:
+    execution = Execution.new(
+        StartDurableExecutionInput(
+            account_id="123456789012",
+            function_name="test-function",
+            function_qualifier="$LATEST",
+            execution_name="subtype-validation-test",
+            execution_timeout_seconds=900,
+            execution_retention_period_days=7,
+            input="{}",
+            invocation_id="test-invocation-id",
+        )
+    )
+    execution.start()
+    return execution
+
+
+def _update(sub_type: str | None, operation_id: str = "step-1") -> OperationUpdate:
+    return OperationUpdate(
+        operation_id=operation_id,
+        operation_type=OperationType.STEP,
+        action=OperationAction.START,
+        sub_type=sub_type,
+    )
+
+
+@pytest.mark.parametrize("sub_type", ["a", "x" * 32, "Order_Charge-21"])
+def test_accepted_subtypes_at_boundaries(sub_type: str) -> None:
+    CheckpointValidator.validate_input([_update(sub_type)], _execution())
+
+
+def test_missing_optional_subtype_is_allowed() -> None:
+    CheckpointValidator.validate_input([_update(None)], _execution())
+
+
+@pytest.mark.parametrize(
+    ("sub_type", "constraints"),
+    [
+        pytest.param(
+            "",
+            (
+                "Member must have length greater than or equal to 1",
+                f"Member must satisfy regular expression pattern: {PATTERN}",
+            ),
+            id="empty-has-two-errors",
+        ),
+        pytest.param(
+            "x" * 33,
+            ("Member must have length less than or equal to 32",),
+            id="over-max-length",
+        ),
+        pytest.param(
+            "order charge",
+            (f"Member must satisfy regular expression pattern: {PATTERN}",),
+            id="spaces",
+        ),
+        pytest.param(
+            "ücret",
+            (f"Member must satisfy regular expression pattern: {PATTERN}",),
+            id="non-ascii",
+        ),
+    ],
+)
+def test_invalid_subtype_is_rejected_with_service_errors(
+    sub_type: str, constraints: tuple[str, ...]
+) -> None:
+    with pytest.raises(AwsApiException) as exc_info:
+        CheckpointValidator.validate_input([_update(sub_type)], _execution())
+    message = str(exc_info.value)
+    expected_errors = [
+        (
+            f"Value '{sub_type}' at 'updates.1.member.subType' "
+            f"failed to satisfy constraint: {constraint}"
+        )
+        for constraint in constraints
+    ]
+    noun = "error" if len(expected_errors) == 1 else "errors"
+    expected = (
+        f"{len(expected_errors)} validation {noun} detected: "
+        + "; ".join(expected_errors)
+    )
+    assert message == expected
+    response = HTTPResponse.create_error_from_exception(exc_info.value)
+    assert response.status_code == 400
+    assert response.body == {"__type": "ValidationException", "message": message}
+
+
+def test_multiple_invalid_updates_report_in_input_order() -> None:
+    oversized = "x" * 33
+    updates = [
+        _update("order charge", operation_id="step-1"),
+        _update(oversized, operation_id="step-2"),
+    ]
+    with pytest.raises(AwsApiException) as exc_info:
+        CheckpointValidator.validate_input(updates, _execution())
+    message = str(exc_info.value)
+    assert message == (
+        "2 validation errors detected: "
+        "Value 'order charge' at 'updates.1.member.subType' "
+        "failed to satisfy constraint: "
+        f"Member must satisfy regular expression pattern: {PATTERN}; "
+        f"Value '{oversized}' at 'updates.2.member.subType' "
+        "failed to satisfy constraint: "
+        "Member must have length less than or equal to 32"
+    )
+
+
+def test_rejects_whole_checkpoint_without_mutating_execution() -> None:
+    execution = _execution()
+    dispatcher = Mock()
+    operations_before = list(execution.operations)
+    checkpoint_before = execution.last_checkpoint
+    token_before = execution.token_sequence
+    updates = [_update("valid"), _update("invalid subtype", "step-2")]
+    with pytest.raises(AwsApiException):
+        CheckpointCore.apply(
+            execution=execution,
+            checkpoint_token="original-token",
+            updates=updates,
+            client_token="client-1",
+            dispatcher=dispatcher,
+            now=datetime(2026, 1, 1, tzinfo=UTC),
+        )
+    dispatcher.apply_updates.assert_not_called()
+    assert list(execution.operations) == operations_before
+    assert execution.last_checkpoint == checkpoint_before
+    assert execution.token_sequence == token_before
+
+
+def test_model_validation_caps_reported_errors_at_ten() -> None:
+    updates = [_update("x" * 33, operation_id=f"step-{index}") for index in range(12)]
+    with pytest.raises(AwsApiException) as exc_info:
+        CheckpointValidator.validate_input(updates, _execution())
+    message = str(exc_info.value)
+    assert message.startswith("10 validation errors detected:")
+    assert "updates.10.member.subType" in message
+    assert "updates.11.member.subType" not in message

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

@@ -3,6 +3,11 @@
 import datetime
 from unittest.mock import Mock
 
+import pytest
+from aws_durable_execution_sdk_python.exceptions import (
+    CheckpointError,
+    DurableApiErrorCategory,
+)
 from aws_durable_execution_sdk_python.lambda_service import (
     CheckpointOutput,
     OperationAction,
@@ -12,6 +17,7 @@
 )
 
 from aws_durable_execution_sdk_python_testing.client import InMemoryServiceClient
+from aws_durable_execution_sdk_python_testing.exceptions import ValidationException
 from aws_durable_execution_sdk_python_testing.worker.registry import ExecutionRegistry
 
 
@@ -52,6 +58,22 @@ def test_checkpoint():
     )
 
 
+def test_checkpoint_validation_error_is_non_retryable():
+    """Local model validation has the same terminal behavior as HTTP 400."""
+    processor = Mock()
+    processor.process_checkpoint.side_effect = ValidationException(
+        "Value 'bad subtype' at 'updates.1.member.subType' failed validation"
+    )
+    client = _client(processor)
+
+    with pytest.raises(CheckpointError, match="updates.1.member.subType") as exc_info:
+        client.checkpoint("execution-arn", "token", [], None)
+
+    assert not exc_info.value.is_retryable()
+    assert exc_info.value.error_category is DurableApiErrorCategory.EXECUTION
+    assert isinstance(exc_info.value.__cause__, ValidationException)
+
+
 def test_get_execution_state():
     """Test get_execution_state method delegates to processor."""
     processor = Mock()

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

@@ -0,0 +1,52 @@
+"""End-to-end parity: invalid SubType fails the durable execution (#836)."""
+
+from typing import Any
+
+import pytest
+from aws_durable_execution_sdk_python.config import StepConfig
+from aws_durable_execution_sdk_python.context import DurableContext, durable_step
+from aws_durable_execution_sdk_python.execution import (
+    InvocationStatus,
+    durable_execution,
+)
+from aws_durable_execution_sdk_python.types import StepContext
+
+from aws_durable_execution_sdk_python_testing.execution import ExecutionStatus
+from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner
+
+
+@pytest.mark.parametrize(
+    ("sub_type", "constraint"),
+    [
+        ("order charge", "Member must satisfy regular expression pattern"),
+        ("x" * 33, "Member must have length less than or equal to 32"),
+    ],
+)
+def test_invalid_step_subtype_fails_execution(sub_type: str, constraint: str) -> None:
+    reached_after_step: list[bool] = []
+
+    @durable_step
+    def charge(step_context: StepContext) -> str:  # noqa: ARG001
+        return "ok"
+
+    # The SDK validates StepConfig at construction. Test-only mutation bypasses
+    # that client guard to exercise the independent local service constraint.
+    config = StepConfig(sub_type="ValidSubtype")
+    object.__setattr__(config, "sub_type", sub_type)
+
+    @durable_execution
+    def handler(event: Any, context: DurableContext) -> str:  # noqa: ARG001
+        context.step(charge(), config=config)
+        reached_after_step.append(True)
+        return "done"
+
+    with DurableFunctionTestRunner(handler=handler, execution_timeout=10) as runner:
+        result = runner.run(input="{}")
+
+    assert result.status is InvocationStatus.FAILED
+    assert result.execution_status is ExecutionStatus.FAILED
+    assert result.error is not None
+    assert result.error.message is not None
+    assert ".member.subType" in result.error.message
+    assert constraint in result.error.message
+    assert not reached_after_step