From a8c369c5f444d27ae181a7596d7481ea9105d754 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:33:40 -0700 Subject: [PATCH 1/4] test: expose terminal error snapshot aliasing --- .../e2e/plugin_error_snapshot_int_test.py | 113 +++++++++++++ .../tests/plugin_test.py | 158 ++++++++++++++++++ 2 files changed, 271 insertions(+) create mode 100644 packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py new file mode 100644 index 000000000..1042a96b2 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py @@ -0,0 +1,113 @@ +"""A terminal plugin cannot change the failed checkpoint replay raises.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import Mock, patch + +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.plugin import ( + DurableInstrumentationPlugin, + InvocationEndInfo, + InvocationStartInfo, + OperationEndInfo, +) +from aws_durable_execution_sdk_python.types import StepContext +from tests.e2e.checkpoint_response_int_test import ( + create_mock_checkpoint_with_operations, +) +from tests.test_helpers import operation_id_sequence + + +def test_terminal_error_mutation_preserves_failure_on_handler_replay() -> None: + step_id = next(operation_id_sequence()) + terminal_errors: list[list[str]] = [] + invocation_errors: list[list[str]] = [] + executed: list[str] = [] + + class MutatingPlugin(DurableInstrumentationPlugin): + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.operation_id == step_id: + assert info.error is not None + assert info.error.stack_trace is not None + terminal_errors.append(list(info.error.stack_trace)) + info.error.stack_trace.clear() + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + error = info.operations[step_id].error + assert error is not None + assert error.stack_trace is not None + invocation_errors.append(list(error.stack_trace)) + + class Factory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return MutatingPlugin() + + def charge(_context: StepContext) -> str: + executed.append("charged") + return "unexpected execution" + + @durable_execution(plugins=[Factory()]) + def handler(_event: Any, context: DurableContext) -> str: + return context.step(charge, name="charge") + + stored_error = { + "ErrorType": "ValueError", + "ErrorMessage": "original failure", + "StackTrace": ["original frame"], + } + event: dict[str, Any] = { + "DurableExecutionArn": "test-arn/execution-1", + "CheckpointToken": "test-token", + "InitialExecutionState": { + "Operations": [ + { + "Id": "execution-1", + "Type": "EXECUTION", + "Status": "STARTED", + "ExecutionDetails": {"InputPayload": "{}"}, + }, + { + "Id": step_id, + "Type": "STEP", + "Status": "FAILED", + "SubType": "Step", + "Name": "charge", + "StepDetails": {"Error": stored_error}, + }, + ], + "NextMarker": "", + }, + "UpdatedOperationIds": [step_id], + "LocalRunner": True, + } + lambda_context = Mock() + lambda_context.aws_request_id = "request-1" + lambda_context.client_context = None + lambda_context.identity = None + lambda_context._epoch_deadline_time_in_ms = 0 # noqa: SLF001 + lambda_context.invoked_function_arn = "test-arn" + lambda_context.tenant_id = None + + with patch("aws_durable_execution_sdk_python.execution.LambdaClient") as client: + checkpoint, _calls = create_mock_checkpoint_with_operations() + client.initialize_client.return_value.checkpoint = checkpoint + first = handler(event, lambda_context) + event["UpdatedOperationIds"] = [] + lambda_context.aws_request_id = "request-2" + replay = handler(event, lambda_context) + + for result in (first, replay): + assert result["Status"] == "FAILED" + assert result["Error"]["ErrorType"] == ( + "aws_durable_execution_sdk_python.exceptions.StepError" + ) + assert result["Error"]["ErrorMessage"] == "original failure" + assert result["Error"]["StackTrace"] == ["original frame"] + assert terminal_errors == [["original frame"]] + assert invocation_errors == [["original frame"], ["original frame"]] + assert stored_error["StackTrace"] == ["original frame"] + assert executed == [] 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 cb277e4a5..6ca23d54c 100644 --- a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py @@ -9,10 +9,14 @@ from collections.abc import Iterator from copy import deepcopy from dataclasses import asdict, fields +from itertools import product from unittest.mock import MagicMock, patch from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( + CallbackDetails, + ChainedInvokeDetails, + ContextDetails, DurableExecutionInvocationOutput, ErrorObject, InvocationStatus as ServiceInvocationStatus, @@ -582,6 +586,59 @@ def test_user_function_end_info(self): self.assertEqual(USER_FUNCTION_END_INFO.end_time, END_TS) self.assertEqual(USER_FUNCTION_END_INFO.error.message, "boom") + def test_operation_info_factory_preserves_subclass_and_replay_flag(self) -> None: + operation = Operation( + operation_id="step-1", + operation_type=ServiceOperationType.STEP, + status=OperationStatus.SUCCEEDED, + sub_type="CustomStep", + name="saved-step", + parent_id="parent-1", + start_timestamp=START_TS, + end_timestamp=END_TS, + step_details=StepDetails(attempt=2, result='"saved"', error=ERROR), + ) + + for info_type in (OperationInfo, OperationStartInfo, OperationEndInfo): + with self.subTest(info_type=info_type): + info = info_type.from_operation(operation, is_replayed=True) + + self.assertIs(type(info), info_type) + self.assertIs(info.operation_type, OperationType.STEP) + self.assertEqual(info.operation_id, "step-1") + self.assertEqual(info.sub_type, "CustomStep") + self.assertEqual(info.name, "saved-step") + self.assertEqual(info.parent_id, "parent-1") + self.assertEqual(info.status, OperationStatus.SUCCEEDED) + self.assertEqual(info.result, '"saved"') + self.assertEqual(info.attempt, 2) + self.assertEqual(info.error, ERROR) + self.assertIsNot(info.error, ERROR) + self.assertEqual(info.start_time, START_TS) + self.assertEqual(info.end_time, END_TS) + self.assertTrue(info.is_replayed) + + def test_terminal_snapshot_preserves_absent_and_empty_errors(self) -> None: + for error in ( + None, + ERROR, + ErrorObject(message="boom", type="Error", data=None, stack_trace=[]), + ): + with self.subTest(error=error): + operation = Operation( + operation_id="step-1", + operation_type=ServiceOperationType.STEP, + status=OperationStatus.FAILED, + step_details=StepDetails(error=error), + ) + info = PluginExecutor(plugins=[])._end_info(operation) + self.assertEqual(info.error, error) + if error is not None: + self.assertIsNot(info.error, error) + assert info.error is not None + if error.stack_trace is not None: + self.assertIsNot(info.error.stack_trace, error.stack_trace) + # endregion Dataclass Tests @@ -1928,6 +1985,107 @@ def test_provider_is_released_when_the_run_scope_exits(self): self.assertIsNone(self.executor._operations_provider) # noqa: SLF001 + def test_plugin_error_mutations_do_not_change_checkpoint_or_later_hooks( + self, + ) -> None: + """Snapshots and terminal hooks must own their mutable error stack traces.""" + for operation_type, checkpoint_delivery in product( + ( + ServiceOperationType.STEP, + ServiceOperationType.CALLBACK, + ServiceOperationType.CHAINED_INVOKE, + ServiceOperationType.CONTEXT, + ), + (False, True), + ): + with self.subTest( + operation_type=operation_type, checkpoint_delivery=checkpoint_delivery + ): + error = ErrorObject( + message="saved error", + type="ExternalError", + data="details", + stack_trace=["original frame"], + ) + operation = Operation( + operation_id="op-1", + operation_type=operation_type, + status=OperationStatus.FAILED, + step_details=( + StepDetails(attempt=2, error=error) + if operation_type is ServiceOperationType.STEP + else None + ), + callback_details=( + CallbackDetails(callback_id="callback-1", error=error) + if operation_type is ServiceOperationType.CALLBACK + else None + ), + chained_invoke_details=( + ChainedInvokeDetails(error=error) + if operation_type is ServiceOperationType.CHAINED_INVOKE + else None + ), + context_details=( + ContextDetails(error=error) + if operation_type is ServiceOperationType.CONTEXT + else None + ), + ) + seen: list[list[str]] = [] + + class MutatingPlugin(DurableInstrumentationPlugin): + def on_invocation_start(self, info: InvocationStartInfo) -> None: + snapshot_error = info.updated_operations["op-1"].error + assert snapshot_error is not None + assert snapshot_error.stack_trace is not None + snapshot_error.stack_trace.clear() + + def on_operation_end(self, info: OperationEndInfo) -> None: + assert info.error is not None + assert info.error.stack_trace is not None + seen.append(list(info.error.stack_trace)) + info.error.stack_trace.clear() + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + snapshot_error = info.operations["op-1"].error + assert snapshot_error is not None + assert snapshot_error.stack_trace is not None + seen.append(list(snapshot_error.stack_trace)) + + executor = PluginExecutor([plugin_factory(MutatingPlugin())]) + with executor.run(): + executor.on_invocation_start( + execution_arn="arn:exec", + lambda_context=LAMBDA_CTX, + execution_start_time=START_TS, + is_first_invocation=False, + operations_provider=lambda: {"op-1": operation}, + updated_operation_ids=["op-1"], + ) + self.assertEqual(error.stack_trace, ["original frame"]) + self.assertEqual(seen, []) + # Both delivery paths must snapshot errors before hooks run. + if checkpoint_delivery: + executor.record_end_hooks([operation]) + executor.drain_end_hooks() + else: + executor.on_operation_update(operation) + # A later observation must not repeat the terminal hook. + executor.on_operation_update( + [operation], + operations={"op-1": operation}, + previous_operations={"op-1": operation}, + ) + executor.on_invocation_end( + DurableExecutionInvocationOutput( + status=ServiceInvocationStatus.SUCCEEDED, + ) + ) + + self.assertEqual(seen, [["original frame"]] * 2) + self.assertEqual(error.stack_trace, ["original frame"]) + class TestInvocationInfoCopySafety(unittest.TestCase): """Invocation infos carrying the lazy operation map must stay copyable. From 5f73f24b535da1c02a2dffbf219ecb8e8b204fe8 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:37:00 -0700 Subject: [PATCH 2/4] fix: isolate terminal plugin error snapshots --- .../README.md | 5 ++++ .../plugin.py | 26 +++++-------------- .../e2e/plugin_error_snapshot_int_test.py | 6 ++--- 3 files changed, 14 insertions(+), 23 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python/README.md b/packages/aws-durable-execution-sdk-python/README.md index 0589e9a4e..bb0d6efbc 100644 --- a/packages/aws-durable-execution-sdk-python/README.md +++ b/packages/aws-durable-execution-sdk-python/README.md @@ -63,6 +63,11 @@ earlier invocation), or put it on the factory, which outlives every invocation keyed by execution ARN, and pruned by the owner, because the SDK will not tell the factory when an execution ends for good. +Operation snapshots, including terminal `OperationEndInfo`, own a copy of the +error's mutable stack-trace list. A plugin may modify that list without changing +the checkpointed error or a later snapshot. Plugins receiving the same hook +object still share that snapshot; it is not copied separately for each plugin. + A factory is an object with a method rather than a plain callable so the registration type can grow a second, optional member later -- a process-level flush on execution-environment shutdown, for example -- without a second breaking 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 43b578f36..19fb99ac1 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 @@ -10,7 +10,7 @@ from collections.abc import Collection, Iterator, Mapping, Sequence from dataclasses import dataclass, field from enum import Enum -from typing import Any, Callable, MutableMapping, Protocol, cast +from typing import Any, Callable, MutableMapping, Protocol, Self, cast from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( @@ -111,13 +111,14 @@ class OperationInfo: """The operation error, when available.""" attempt: int | None = field(default=None, kw_only=True) - @staticmethod + @classmethod def from_operation( + cls, operation: Operation, *, is_replayed: bool = False, - ) -> OperationInfo: - return OperationInfo( + ) -> Self: + return cls( operation_id=operation.operation_id, operation_type=_to_operation_type(operation.operation_type), sub_type=operation.sub_type, @@ -1856,22 +1857,7 @@ def on_operation_update( self._submit_change(change_info) def _end_info(self, operation: Operation) -> OperationEndInfo: - return OperationEndInfo( - operation_id=operation.operation_id, - operation_type=_to_operation_type(operation.operation_type), - sub_type=operation.sub_type, - name=operation.name, - parent_id=operation.parent_id, - start_time=operation.start_timestamp, - end_time=operation.end_timestamp, - result=_extract_result(operation), - status=operation.status, - error=self._extract_error(operation), - attempt=( - operation.step_details.attempt if operation.step_details else None - ), - is_replayed=False, - ) + return OperationEndInfo.from_operation(operation) def _change_info( self, diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py index 1042a96b2..e368c1e62 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py @@ -50,10 +50,10 @@ def charge(_context: StepContext) -> str: executed.append("charged") return "unexpected execution" - @durable_execution(plugins=[Factory()]) def handler(_event: Any, context: DurableContext) -> str: return context.step(charge, name="charge") + wrapped = durable_execution(handler, plugins=[Factory()]) stored_error = { "ErrorType": "ValueError", "ErrorMessage": "original failure", @@ -95,10 +95,10 @@ def handler(_event: Any, context: DurableContext) -> str: with patch("aws_durable_execution_sdk_python.execution.LambdaClient") as client: checkpoint, _calls = create_mock_checkpoint_with_operations() client.initialize_client.return_value.checkpoint = checkpoint - first = handler(event, lambda_context) + first = wrapped(event, lambda_context) event["UpdatedOperationIds"] = [] lambda_context.aws_request_id = "request-2" - replay = handler(event, lambda_context) + replay = wrapped(event, lambda_context) for result in (first, replay): assert result["Status"] == "FAILED" From 62e636131af7dc04aaa2a169e709eedaa0756ab4 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Tue, 27 Oct 2026 16:44:14 -0700 Subject: [PATCH 3/4] test: cover virtual child error snapshot isolation --- .../e2e/plugin_error_snapshot_int_test.py | 59 +++++++++++++++++++ .../tests/plugin_test.py | 45 +++++++------- 2 files changed, 83 insertions(+), 21 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py index e368c1e62..85370634f 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py @@ -5,7 +5,12 @@ from typing import Any from unittest.mock import Mock, patch +from aws_durable_execution_sdk_python.config import ChildConfig from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.exceptions import ( + ChildContextError, + DurableOperationError, +) from aws_durable_execution_sdk_python.execution import durable_execution from aws_durable_execution_sdk_python.plugin import ( DurableInstrumentationPlugin, @@ -14,6 +19,7 @@ OperationEndInfo, ) from aws_durable_execution_sdk_python.types import StepContext +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner from tests.e2e.checkpoint_response_int_test import ( create_mock_checkpoint_with_operations, ) @@ -111,3 +117,56 @@ def handler(_event: Any, context: DurableContext) -> str: assert invocation_errors == [["original frame"], ["original frame"]] assert stored_error["StackTrace"] == ["original frame"] assert executed == [] + + +def test_virtual_child_terminal_snapshot_preserves_raised_error() -> None: + original = DurableOperationError( + message="original failure", data="details", stack_trace=["original frame"] + ) + terminal_infos: list[OperationEndInfo] = [] + caught: list[ChildContextError] = [] + + class MutatingPlugin(DurableInstrumentationPlugin): + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.name == "virtual-child": + assert info.error is not None + assert info.error.stack_trace is not None + terminal_infos.append(info) + info.error.stack_trace.clear() + + class Factory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return MutatingPlugin() + + def fail(_context: DurableContext) -> None: + raise original + + def handler(_event: Any, context: DurableContext) -> str: + try: + context.run_in_child_context( + fail, name="virtual-child", config=ChildConfig(is_virtual=True) + ) + except ChildContextError as error: + caught.append(error) + return "caught" + raise AssertionError("Expected the virtual child's original failure") + + with DurableFunctionTestRunner( + handler=durable_execution(handler, plugins=[Factory()]) + ) as runner: + result = runner.run(input="{}", execution_timeout=15) + + assert result.status.value == "SUCCEEDED" + assert result.result == '"caught"' + (terminal_info,) = terminal_infos + assert terminal_info.start_time is None + assert terminal_info.error is not None + assert terminal_info.error.stack_trace == [] + (raised,) = caught + assert raised.message == "original failure" + assert raised.error_type == original.error_type + assert raised.data == "details" + assert raised.stack_trace == ["original frame"] + assert original.stack_trace == ["original frame"] 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 6ca23d54c..c6399a7bb 100644 --- a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py @@ -586,7 +586,7 @@ def test_user_function_end_info(self): self.assertEqual(USER_FUNCTION_END_INFO.end_time, END_TS) self.assertEqual(USER_FUNCTION_END_INFO.error.message, "boom") - def test_operation_info_factory_preserves_subclass_and_replay_flag(self) -> None: + def test_terminal_snapshot_preserves_operation_fields(self) -> None: operation = Operation( operation_id="step-1", operation_type=ServiceOperationType.STEP, @@ -599,24 +599,22 @@ def test_operation_info_factory_preserves_subclass_and_replay_flag(self) -> None step_details=StepDetails(attempt=2, result='"saved"', error=ERROR), ) - for info_type in (OperationInfo, OperationStartInfo, OperationEndInfo): - with self.subTest(info_type=info_type): - info = info_type.from_operation(operation, is_replayed=True) - - self.assertIs(type(info), info_type) - self.assertIs(info.operation_type, OperationType.STEP) - self.assertEqual(info.operation_id, "step-1") - self.assertEqual(info.sub_type, "CustomStep") - self.assertEqual(info.name, "saved-step") - self.assertEqual(info.parent_id, "parent-1") - self.assertEqual(info.status, OperationStatus.SUCCEEDED) - self.assertEqual(info.result, '"saved"') - self.assertEqual(info.attempt, 2) - self.assertEqual(info.error, ERROR) - self.assertIsNot(info.error, ERROR) - self.assertEqual(info.start_time, START_TS) - self.assertEqual(info.end_time, END_TS) - self.assertTrue(info.is_replayed) + info = PluginExecutor(plugins=[])._end_info(operation) + + self.assertIs(type(info), OperationEndInfo) + self.assertIs(info.operation_type, OperationType.STEP) + self.assertEqual(info.operation_id, "step-1") + self.assertEqual(info.sub_type, "CustomStep") + self.assertEqual(info.name, "saved-step") + self.assertEqual(info.parent_id, "parent-1") + self.assertEqual(info.status, OperationStatus.SUCCEEDED) + self.assertEqual(info.result, '"saved"') + self.assertEqual(info.attempt, 2) + self.assertEqual(info.error, ERROR) + self.assertIsNot(info.error, ERROR) + self.assertEqual(info.start_time, START_TS) + self.assertEqual(info.end_time, END_TS) + self.assertFalse(info.is_replayed) def test_terminal_snapshot_preserves_absent_and_empty_errors(self) -> None: for error in ( @@ -2351,6 +2349,7 @@ class TestPluginExecutorOnChildContextEnd(unittest.TestCase): def test_builds_terminal_operation_info(self): captured: list[OperationEndInfo] = [] + error = ErrorObject("boom", "Error", None, ["original frame"]) class _CapturingPlugin(_TrackingPlugin): def on_operation_end(self, info: OperationEndInfo) -> None: @@ -2372,7 +2371,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None: executor.on_child_context_end( identifier, OperationStatus.FAILED, - error=ERROR, + error=error, is_replayed=True, ) @@ -2386,7 +2385,11 @@ def on_operation_end(self, info: OperationEndInfo) -> None: self.assertEqual(info.parent_id, "parent-1") self.assertEqual(info.name, "book-trip") self.assertEqual(info.status, OperationStatus.FAILED) - self.assertEqual(info.error, ERROR) + self.assertEqual(info.error, error) + assert info.error is not None + assert info.error.stack_trace is not None + info.error.stack_trace.clear() + self.assertEqual(error.stack_trace, ["original frame"]) self.assertTrue(info.is_replayed) self.assertIsNone(info.start_time) assert info.end_time is not None From dfce1bd8f1605ea03e948eba7ec94e3f09187269 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Tue, 27 Oct 2026 16:45:44 -0700 Subject: [PATCH 4/4] fix: limit terminal isolation to error copies --- .../README.md | 9 +++--- .../plugin.py | 28 ++++++++++++++----- 2 files changed, 26 insertions(+), 11 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python/README.md b/packages/aws-durable-execution-sdk-python/README.md index bb0d6efbc..dc84924b9 100644 --- a/packages/aws-durable-execution-sdk-python/README.md +++ b/packages/aws-durable-execution-sdk-python/README.md @@ -63,10 +63,11 @@ earlier invocation), or put it on the factory, which outlives every invocation keyed by execution ARN, and pruned by the owner, because the SDK will not tell the factory when an execution ends for good. -Operation snapshots, including terminal `OperationEndInfo`, own a copy of the -error's mutable stack-trace list. A plugin may modify that list without changing -the checkpointed error or a later snapshot. Plugins receiving the same hook -object still share that snapshot; it is not copied separately for each plugin. +Invocation operation-map snapshots and terminal `OperationEndInfo` own a copy of +the error's mutable stack-trace list, including virtual child completions without +a checkpoint. A plugin may modify that list without changing the checkpointed +error, the child error raised to the caller, or a later snapshot. Plugins receiving +the same hook object still share that snapshot; it is not copied for each plugin. A factory is an object with a method rather than a plain callable so the registration type can grow a second, optional member later -- a process-level 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 19fb99ac1..6cd688e07 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 @@ -10,7 +10,7 @@ from collections.abc import Collection, Iterator, Mapping, Sequence from dataclasses import dataclass, field from enum import Enum -from typing import Any, Callable, MutableMapping, Protocol, Self, cast +from typing import Any, Callable, MutableMapping, Protocol, cast from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( @@ -111,14 +111,13 @@ class OperationInfo: """The operation error, when available.""" attempt: int | None = field(default=None, kw_only=True) - @classmethod + @staticmethod def from_operation( - cls, operation: Operation, *, is_replayed: bool = False, - ) -> Self: - return cls( + ) -> OperationInfo: + return OperationInfo( operation_id=operation.operation_id, operation_type=_to_operation_type(operation.operation_type), sub_type=operation.sub_type, @@ -1806,7 +1805,7 @@ def on_child_context_end( start_time=None, end_time=now, status=status, - error=error, + error=_copy_error(error), is_replayed=is_replayed, ), ) @@ -1857,7 +1856,22 @@ def on_operation_update( self._submit_change(change_info) def _end_info(self, operation: Operation) -> OperationEndInfo: - return OperationEndInfo.from_operation(operation) + return OperationEndInfo( + operation_id=operation.operation_id, + operation_type=_to_operation_type(operation.operation_type), + sub_type=operation.sub_type, + name=operation.name, + parent_id=operation.parent_id, + start_time=operation.start_timestamp, + end_time=operation.end_timestamp, + result=_extract_result(operation), + status=operation.status, + error=_copy_error(self._extract_error(operation)), + attempt=( + operation.step_details.attempt if operation.step_details else None + ), + is_replayed=False, + ) def _change_info( self,