From 415f6bb49602e9014170f9be0e12a347fef804ed Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Mon, 10 Aug 2026 22:36:04 +0000 Subject: [PATCH 1/4] feat(plugin): expose operation maps on invocation hooks Invocation-level plugin infos carried no view of the execution's operation state, so Python plugins could not see what the JS SDK exposes on every invocation hook. Add to the invocation infos: - InvocationInfo.operations, the checkpointed operation map converted to OperationInfo, on both invocation-start and invocation-end - InvocationStartInfo.updated_operations, the subset named by the invocation input's UpdatedOperationIds, i.e. operations completed externally while the execution was suspended Both are kw-only with empty-map defaults, so existing constructor calls and plugins are unaffected. The end hook re-reads the map so it reports end-of-invocation state rather than the start snapshot. The map is snapshotted when the hook fires but converted to OperationInfo only on first access, so an operation-heavy execution does not pay per invocation for a view no plugin reads, while a plugin that stashes the info still sees the state as of its hook. Refs #617 --- .../execution.py | 6 + .../plugin.py | 167 +++++++++++- .../tests/plugin_test.py | 255 +++++++++++++++++- 3 files changed, 420 insertions(+), 8 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/execution.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/execution.py index 7b280d46..bad5b399 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/execution.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/execution.py @@ -304,6 +304,12 @@ def wrapper(event: Any, context: LambdaContext) -> MutableMapping[str, Any]: else None ), is_first_invocation=not has_prior_operations, + # Read the map through a callable rather than snapshotting it + # here: the invocation-end hook needs the state as of the end of + # the invocation, and neither hook pays for the conversion until + # a plugin actually reads it. + operations_provider=lambda: execution_state.operations, + updated_operation_ids=invocation_input.updated_operation_ids, ) # Thread 1: Run background checkpoint processing executor.submit(execution_state.checkpoint_batches_forever) 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 197a9cf3..f1c1c2d9 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 @@ -8,7 +8,8 @@ from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from enum import Enum -from typing import Any, Callable, MutableMapping, cast +from threading import Lock +from typing import Any, Callable, Iterator, MutableMapping, cast from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( @@ -94,6 +95,82 @@ def from_operation( ) +def _to_operation_info_map( + operations: Mapping[str, Operation], +) -> dict[str, OperationInfo]: + """Convert a map of checkpointed operations to the plugin ``OperationInfo`` view. + + ``is_replayed`` is left at its default ``False``: these entries describe the + stored state of an operation, not a replay event for it. Replay is signalled + through the dedicated operation hooks. + """ + return { + operation_id: OperationInfo.from_operation(operation) + for operation_id, operation in operations.items() + } + + +class _LazyOperationInfoMap(Mapping[str, OperationInfo]): + """Read-only ``operation id -> OperationInfo`` map built on first access. + + Invocation hook infos carry the execution's whole operation map. Converting + it eagerly would charge every invocation of an operation-heavy execution for + a view most plugins never read, so the conversion is deferred to the first + mapping operation and then cached. The underlying operations are snapshotted + before this map is handed out, so deferring the conversion does not move the + point in time the map describes. Behaves like a plain read-only ``dict`` + (iteration, ``len``, ``in``, ``get``, ``==`` against any mapping). + """ + + __slots__ = ("_provider", "_lock", "_resolved") + + def __init__(self, provider: Callable[[], dict[str, OperationInfo]] | None) -> None: + self._provider = provider + self._lock = Lock() + self._resolved: dict[str, OperationInfo] | None = None + + def _resolve(self) -> dict[str, OperationInfo]: + with self._lock: + if self._resolved is None: + if self._provider is None: + self._resolved = {} + else: + try: + self._resolved = self._provider() + except Exception: + # A plugin-facing view must never break the execution. + logger.exception( + "Failed to build plugin operations map; using empty map" + ) + self._resolved = {} + return self._resolved + + def __getitem__(self, key: str) -> OperationInfo: + return self._resolve()[key] + + def __iter__(self) -> Iterator[str]: + return iter(self._resolve()) + + def __len__(self) -> int: + return len(self._resolve()) + + def __eq__(self, other: object) -> bool: + if isinstance(other, Mapping): + return self._resolve() == dict(other) + return NotImplemented + + def __ne__(self, other: object) -> bool: + result = self.__eq__(other) + if result is NotImplemented: + return result + return not result + + __hash__ = None # type: ignore[assignment] # mutable-by-materialization view + + def __repr__(self) -> str: + return f"{type(self).__name__}({self._resolve()!r})" + + @dataclass(frozen=True) class OperationStartInfo(OperationInfo): pass @@ -163,11 +240,29 @@ class InvocationInfo: execution_arn: str | None is_first_invocation: bool execution_start_time: datetime.datetime | None = None + operations: Mapping[str, OperationInfo] = field(default_factory=dict, kw_only=True) + """Checkpointed operations for this execution, keyed by operation id. + + A point-in-time view of the execution's operation map: as observed at the + start of the invocation on ``on_invocation_start``, and as observed at the + end of the invocation on ``on_invocation_end``. Empty on the very first + invocation-start, before any operation has been checkpointed. + """ @dataclass(frozen=True) class InvocationStartInfo(InvocationInfo): - pass + updated_operations: Mapping[str, OperationInfo] = field( + default_factory=dict, kw_only=True + ) + """Operations updated externally while this execution was suspended. + + A wait timer that expired, a callback that was delivered, or a chained + invoke that completed between the previous invocation and this one. This is + the subset of :attr:`InvocationInfo.operations` named by the durable + invocation input's ``UpdatedOperationIds``, so it is empty on the first + invocation. + """ @dataclass(frozen=True) @@ -180,12 +275,20 @@ def from_durable_execution_invocation_output( cls, invocation_start_info: InvocationStartInfo, output: "DurableExecutionInvocationOutput", + operations: Mapping[str, OperationInfo] | None = None, ): return InvocationEndInfo( request_id=invocation_start_info.request_id, execution_arn=invocation_start_info.execution_arn, is_first_invocation=invocation_start_info.is_first_invocation, execution_start_time=invocation_start_info.execution_start_time, + # Default to the start-of-invocation view when the caller has no + # fresher snapshot to offer. + operations=( + operations + if operations is not None + else invocation_start_info.operations + ), status=output.status, error=output.error, ) @@ -266,6 +369,7 @@ def __init__(self, plugins: list[DurableInstrumentationPlugin] | None): self._plugins = plugins or [] self._executor: ThreadPoolExecutor | None = None self._invocation_status: InvocationStartInfo | None = None + self._operations_provider: Callable[[], Mapping[str, Operation]] | None = None @contextlib.contextmanager def run(self): @@ -278,6 +382,7 @@ def run(self): yield finally: self._invocation_status = None + self._operations_provider = None # Shut down the thread pool, waiting for pending tasks to complete. if self._executor: self._executor.shutdown(wait=True) @@ -318,19 +423,66 @@ def execute_plugins(self, info, sync): # this is called asynchronously, so plugins cannot manipulate thread local objects self._executor.submit(self._dispatch_plugin, plugin, info) + @staticmethod + def _snapshot_operation_infos( + operations_provider: Callable[[], Mapping[str, Operation]] | None, + ) -> Mapping[str, OperationInfo]: + """Capture the operation map now; defer the ``OperationInfo`` conversion. + + Copying the map is cheap, building an ``OperationInfo`` per operation is + not. Snapshotting eagerly pins the point in time the hook reports, so a + plugin that stashes the info and reads it later still sees the state as + of its hook; deferring the conversion keeps operation-heavy executions + from paying for a view no plugin reads. + """ + if operations_provider is None: + return _LazyOperationInfoMap(None) + try: + snapshot = dict(operations_provider()) + except Exception: + # A plugin-facing view must never break the execution. + logger.exception("Failed to snapshot operations for plugin hook") + return _LazyOperationInfoMap(None) + return _LazyOperationInfoMap(lambda: _to_operation_info_map(snapshot)) + def on_invocation_start( self, execution_arn: str, is_first_invocation: bool, execution_start_time: datetime.datetime | None, lambda_context: LambdaContext | None, + operations_provider: Callable[[], Mapping[str, Operation]] | None = None, + updated_operation_ids: Sequence[str] | None = None, ) -> None: + """Fire the invocation-start hook. + + Args: + execution_arn: ARN of the durable execution. + is_first_invocation: False when prior operations exist (a replay). + execution_start_time: Start timestamp of the execution operation. + lambda_context: Lambda context, for the request id. + operations_provider: Returns the current checkpointed operation map. + Called once here to snapshot it; the conversion to the plugin's + ``OperationInfo`` view is deferred to first access. + updated_operation_ids: Operation ids from the invocation input's + ``UpdatedOperationIds`` -- those updated while suspended. + """ aws_request_id = lambda_context.aws_request_id if lambda_context else None + self._operations_provider = operations_provider + operations = self._snapshot_operation_infos(operations_provider) self._invocation_status = InvocationStartInfo( execution_arn=execution_arn, request_id=aws_request_id, is_first_invocation=is_first_invocation, execution_start_time=execution_start_time, + operations=operations, + updated_operations=_LazyOperationInfoMap( + lambda: { + operation_id: operations[operation_id] + for operation_id in (updated_operation_ids or []) + if operation_id in operations + } + ), ) self.execute_plugins(self._invocation_status, sync=True) @@ -342,9 +494,13 @@ def on_invocation_end( # on_invocation_start not called, skip return + # Re-read the operation map so the end hook sees the state as of the end + # of this invocation, not the snapshot taken at its start. invocation_end_info = ( InvocationEndInfo.from_durable_execution_invocation_output( - self._invocation_status, output + self._invocation_status, + output, + operations=self._snapshot_operation_infos(self._operations_provider), ) ) self.execute_plugins(invocation_end_info, sync=True) @@ -523,10 +679,7 @@ def on_operation_update( operation.operation_id: OperationInfo.from_operation(operation) for operation in changed_operations }, - operations={ - operation_id: OperationInfo.from_operation(operation) - for operation_id, operation in operations.items() - }, + operations=_to_operation_info_map(operations), ), sync=True, ) 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 97e3d9f5..ab6789da 100644 --- a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py @@ -1,7 +1,7 @@ import datetime import logging import unittest -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( @@ -148,6 +148,69 @@ def test_invocation_end_info(self): self.assertEqual(INVOCATION_END_INFO.status, InvocationStatus.FAILED) self.assertEqual(INVOCATION_END_INFO.error.message, "boom") + def test_invocation_info_operation_maps_default_to_empty(self): + """Both maps default to empty so existing constructor calls keep working.""" + self.assertEqual({}, INVOCATION_START_INFO.operations) + self.assertEqual({}, INVOCATION_START_INFO.updated_operations) + self.assertEqual({}, INVOCATION_END_INFO.operations) + + def test_invocation_start_info_carries_operation_maps(self): + info = InvocationStartInfo( + request_id="req-1", + execution_arn="arn:test", + is_first_invocation=False, + operations={"op-1": OPERATION_END_INFO, "op-2": OPERATION_START_INFO}, + updated_operations={"op-1": OPERATION_END_INFO}, + ) + + self.assertEqual( + {"op-1": OPERATION_END_INFO, "op-2": OPERATION_START_INFO}, + info.operations, + ) + self.assertEqual({"op-1": OPERATION_END_INFO}, info.updated_operations) + + def test_invocation_end_info_inherits_start_operations(self): + """The end factory reuses the start snapshot when given no override.""" + start = InvocationStartInfo( + request_id="req-1", + execution_arn="arn:test", + is_first_invocation=True, + operations={"op-1": OPERATION_END_INFO}, + updated_operations={"op-1": OPERATION_END_INFO}, + ) + output = DurableExecutionInvocationOutput( + status=InvocationStatus.SUCCEEDED, result=None, error=None + ) + + end = InvocationEndInfo.from_durable_execution_invocation_output(start, output) + + self.assertEqual({"op-1": OPERATION_END_INFO}, end.operations) + # updated_operations is a start-hook-only surface. + self.assertFalse(hasattr(end, "updated_operations")) + + def test_invocation_end_info_accepts_fresher_operations(self): + """An explicit map wins, letting the end hook see end-of-invocation state.""" + start = InvocationStartInfo( + request_id="req-1", + execution_arn="arn:test", + is_first_invocation=True, + operations={"op-1": OPERATION_START_INFO}, + ) + output = DurableExecutionInvocationOutput( + status=InvocationStatus.SUCCEEDED, result=None, error=None + ) + + end = InvocationEndInfo.from_durable_execution_invocation_output( + start, + output, + operations={"op-1": OPERATION_END_INFO, "op-2": OPERATION_START_INFO}, + ) + + self.assertEqual( + {"op-1": OPERATION_END_INFO, "op-2": OPERATION_START_INFO}, + end.operations, + ) + def test_user_function_start_info(self): self.assertEqual(USER_FUNCTION_START_INFO.operation_id, "op-1") self.assertEqual(USER_FUNCTION_START_INFO.operation_type, OperationType.STEP) @@ -524,6 +587,196 @@ def test_pending_fires_invocation_end(self): self.assertIn("invocation_end:req-1", self.plugin.calls) +class TestInvocationHookOperationMaps(unittest.TestCase): + """Tests for the operations / updated_operations maps on invocation hooks.""" + + def setUp(self): + self.captured: list[object] = [] + + class _CapturingPlugin(DurableInstrumentationPlugin): + def on_invocation_start(_self, info): # noqa: N805 + self.captured.append(info) + + def on_invocation_end(_self, info): # noqa: N805 + self.captured.append(info) + + self.executor = PluginExecutor(plugins=[_CapturingPlugin()]) + + @staticmethod + def _operation(operation_id, status=OperationStatus.SUCCEEDED): + return Operation( + operation_id=operation_id, + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name=f"name-{operation_id}", + parent_id="root", + status=status, + start_timestamp=START_TS, + end_timestamp=END_TS, + step_details=StepDetails(attempt=1), + ) + + def _start(self, *, operations=None, updated_operation_ids=None): + self.executor.on_invocation_start( + execution_arn="arn:exec", + lambda_context=LAMBDA_CTX, + execution_start_time=START_TS, + is_first_invocation=False, + operations_provider=(lambda: operations) + if operations is not None + else None, + updated_operation_ids=updated_operation_ids, + ) + + def test_start_info_converts_operations_to_operation_info(self): + operations = {"op-1": self._operation("op-1")} + + with self.executor.run(): + self._start(operations=operations) + + (start_info,) = self.captured + self.assertEqual(["op-1"], list(start_info.operations)) + converted = start_info.operations["op-1"] + self.assertIsInstance(converted, OperationInfo) + self.assertEqual("op-1", converted.operation_id) + self.assertEqual(OperationStatus.SUCCEEDED, converted.status) + self.assertEqual(START_TS, converted.start_time) + self.assertEqual(END_TS, converted.end_time) + self.assertEqual(1, converted.attempt) + # Stored state, not a replay event for the operation. + self.assertFalse(converted.is_replayed) + + def test_updated_operations_is_the_subset_named_by_updated_ids(self): + operations = { + "op-1": self._operation("op-1"), + "op-2": self._operation("op-2"), + } + + with self.executor.run(): + self._start(operations=operations, updated_operation_ids=["op-2"]) + + (start_info,) = self.captured + self.assertEqual(2, len(start_info.operations)) + self.assertEqual(["op-2"], list(start_info.updated_operations)) + # Same converted instance as in the full map, not a second conversion. + self.assertIs( + start_info.operations["op-2"], start_info.updated_operations["op-2"] + ) + + def test_updated_ids_absent_from_the_map_are_skipped(self): + operations = {"op-1": self._operation("op-1")} + + with self.executor.run(): + self._start( + operations=operations, updated_operation_ids=["op-1", "op-missing"] + ) + + (start_info,) = self.captured + self.assertEqual(["op-1"], list(start_info.updated_operations)) + + def test_first_invocation_has_empty_maps(self): + with self.executor.run(): + self._start(operations={}, updated_operation_ids=[]) + + (start_info,) = self.captured + self.assertEqual({}, start_info.operations) + self.assertEqual({}, start_info.updated_operations) + + def test_missing_provider_yields_empty_maps(self): + """Callers that supply no provider still get a usable, empty map.""" + with self.executor.run(): + self._start() + self.executor.on_invocation_end( + output=DurableExecutionInvocationOutput( + status=InvocationStatus.SUCCEEDED, result=None, error=None + ) + ) + + start_info, end_info = self.captured + self.assertEqual({}, start_info.operations) + self.assertEqual({}, start_info.updated_operations) + self.assertEqual({}, end_info.operations) + + def test_operation_info_conversion_is_deferred_and_cached(self): + conversions: list[str] = [] + real_from_operation = OperationInfo.from_operation + + def counting(operation, **kwargs): + conversions.append(operation.operation_id) + return real_from_operation(operation, **kwargs) + + with ( + patch.object(OperationInfo, "from_operation", staticmethod(counting)), + self.executor.run(), + ): + self._start(operations={"op-1": self._operation("op-1")}) + # Nothing read the map inside the hook, so nothing was converted. + self.assertEqual([], conversions) + + (start_info,) = self.captured + self.assertEqual(1, len(start_info.operations)) + self.assertEqual(["op-1"], conversions) + # Repeated reads reuse the cached conversion. + self.assertEqual(["op-1"], list(start_info.operations)) + self.assertIn("op-1", start_info.operations) + self.assertEqual(["op-1"], conversions) + + def test_end_info_reflects_operations_added_during_the_invocation(self): + operations = {"op-1": self._operation("op-1")} + + with self.executor.run(): + self.executor.on_invocation_start( + execution_arn="arn:exec", + lambda_context=LAMBDA_CTX, + execution_start_time=START_TS, + is_first_invocation=True, + operations_provider=lambda: operations, + ) + operations["op-2"] = self._operation("op-2") + self.executor.on_invocation_end( + output=DurableExecutionInvocationOutput( + status=InvocationStatus.SUCCEEDED, result=None, error=None + ) + ) + + start_info, end_info = self.captured + self.assertEqual(["op-1", "op-2"], sorted(end_info.operations)) + # The start map is snapshotted at its hook, so a later checkpoint does + # not retroactively change what the start hook reported. + self.assertEqual(["op-1"], sorted(start_info.operations)) + + def test_failing_provider_degrades_to_an_empty_map(self): + def provider(): + raise RuntimeError("boom") + + with ( + self.executor.run(), + self.assertLogs( + "aws_durable_execution_sdk_python.plugin", level="ERROR" + ) as logs, + ): + self.executor.on_invocation_start( + execution_arn="arn:exec", + lambda_context=LAMBDA_CTX, + execution_start_time=START_TS, + is_first_invocation=True, + operations_provider=provider, + updated_operation_ids=["op-1"], + ) + + self.assertTrue(any("Failed to snapshot operations" in m for m in logs.output)) + (start_info,) = self.captured + self.assertEqual(0, len(start_info.operations)) + self.assertEqual(0, len(start_info.updated_operations)) + + def test_provider_is_released_when_the_run_scope_exits(self): + with self.executor.run(): + self._start(operations={"op-1": self._operation("op-1")}) + self.assertIsNotNone(self.executor._operations_provider) # noqa: SLF001 + + self.assertIsNone(self.executor._operations_provider) # noqa: SLF001 + + class TestPluginExecutorOnOperationAction(unittest.TestCase): """Tests for PluginExecutor.on_operation_action.""" From a215cdaae6360c1e7f7192e0f0011c0505047b0f Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Mon, 10 Aug 2026 22:36:10 +0000 Subject: [PATCH 2/4] test(conformance): dump operation counts for 10-19 The 10-19 handler is a canonical dump of each invocation hook's own info. Now that the Python infos expose the operation maps, emit the canonical operationsCount on both hooks and updatedOperationsCount on invocation-start instead of omitting them. Refs #617 --- .../plugin/plugin_invocation_info_shape.py | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py index 5df8ed21..a75871bb 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py @@ -10,11 +10,11 @@ ``terminal`` := status in (SUCCEEDED, FAILED). No cross-hook reconstruction — ``isFirstInvocation`` on the end record comes from the invocation-end info. -Python surface note: ``InvocationInfo`` carries only ``request_id``, -``execution_arn``, ``is_first_invocation`` and ``execution_start_time``; the end -info adds ``status`` + ``error``. It has NO execution-input, operations-map, -externally-updated-operations, or execution-result field, so the canonical -``executionInput`` / ``operationsCount`` / ``updatedOperationsCount`` / +Python surface note: ``InvocationInfo`` carries ``request_id``, +``execution_arn``, ``is_first_invocation``, ``execution_start_time`` and the +``operations`` map; ``InvocationStartInfo`` adds ``updated_operations`` and the +end info adds ``status`` + ``error``. It has NO execution-input or +execution-result field, so the canonical ``executionInput`` / ``executionResult`` keys are omitted — the honest red for those probes. """ @@ -42,13 +42,15 @@ def _emit(record: dict[str, Any], execution_arn: str | None) -> None: class InvocationInfoShapePlugin(DurableInstrumentationPlugin): def on_invocation_start(self, info: InvocationStartInfo) -> None: - # Canonical dump of InvocationStartInfo. executionInput / operationsCount - # / updatedOperationsCount are absent from the Python type and therefore - # omitted — that omission is the parity signal under test. + # Canonical dump of InvocationStartInfo. executionInput is absent from + # the Python type and therefore omitted — that omission is the parity + # signal under test. record: dict[str, Any] = { "plugin": "CONFPLUGIN", "hook": "invocation-start", "isFirstInvocation": info.is_first_invocation, + "operationsCount": len(info.operations), + "updatedOperationsCount": len(info.updated_operations), } if info.request_id is not None: record["requestId"] = info.request_id @@ -58,13 +60,13 @@ def on_invocation_start(self, info: InvocationStartInfo) -> None: def on_invocation_end(self, info: InvocationEndInfo) -> None: # isFirstInvocation MUST come from the END info itself. executionInput / - # operationsCount / executionResult are absent from the Python type and - # therefore omitted. + # executionResult are absent from the Python type and therefore omitted. status = info.status.name record: dict[str, Any] = { "plugin": "CONFPLUGIN", "hook": "invocation-end", "isFirstInvocation": info.is_first_invocation, + "operationsCount": len(info.operations), "status": status, "terminal": status in ("SUCCEEDED", "FAILED"), } From 2eb6071f9bd4191f9fffb2e952b07b4038a60043 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Mon, 10 Aug 2026 22:59:18 +0000 Subject: [PATCH 3/4] feat(plugin): surface execution input and result Port the remaining JS invocation-hook payload surfaces so plugins can record what an execution was given and what it produced. The Workflow Insight plugin needs both to emit execution records. - InvocationInfo.execution_input: the deserialized input event, the same object the durable handler receives, on both hooks - InvocationEndInfo.execution_result: the serialized result from the invocation output, None when the invocation suspended or failed Both are kw-only with None defaults, so this is purely additive. These are experimental and out of GA conformance scope, so they are marked EXPERIMENTAL in their docstrings and no conformance requirement asserts them. Reaching them already requires the plugins= parameter, which emits a FutureWarning. Refs #616 --- .../execution.py | 1 + .../plugin.py | 19 ++++ .../tests/plugin_test.py | 91 ++++++++++++++++++- 3 files changed, 109 insertions(+), 2 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/execution.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/execution.py index bad5b399..881d5074 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/execution.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/execution.py @@ -310,6 +310,7 @@ def wrapper(event: Any, context: LambdaContext) -> MutableMapping[str, Any]: # a plugin actually reads it. operations_provider=lambda: execution_state.operations, updated_operation_ids=invocation_input.updated_operation_ids, + execution_input=input_event, ) # Thread 1: Run background checkpoint processing executor.submit(execution_state.checkpoint_batches_forever) 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 f1c1c2d9..c191cce8 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 @@ -248,6 +248,13 @@ class InvocationInfo: end of the invocation on ``on_invocation_end``. Empty on the very first invocation-start, before any operation has been checkpointed. """ + execution_input: Any = field(default=None, kw_only=True) + """EXPERIMENTAL: the deserialized input the execution was started with. + + The same value the durable handler receives as its ``event`` argument, + parsed from the execution's stored input payload. ``None`` when the + execution has no input payload, or when the info was built without one. + """ @dataclass(frozen=True) @@ -269,6 +276,13 @@ class InvocationStartInfo(InvocationInfo): class InvocationEndInfo(InvocationInfo): status: InvocationStatus = field(kw_only=True) error: ErrorObject | None = None + execution_result: str | None = field(default=None, kw_only=True) + """EXPERIMENTAL: the serialized result of the execution, when it produced one. + + Taken from the invocation output's ``Result``. ``None`` when the invocation + suspended or failed instead of completing, and an empty string when the + result was too large to inline and was checkpointed separately. + """ @classmethod def from_durable_execution_invocation_output( @@ -289,6 +303,8 @@ def from_durable_execution_invocation_output( if operations is not None else invocation_start_info.operations ), + execution_input=invocation_start_info.execution_input, + execution_result=output.result, status=output.status, error=output.error, ) @@ -453,6 +469,7 @@ def on_invocation_start( lambda_context: LambdaContext | None, operations_provider: Callable[[], Mapping[str, Operation]] | None = None, updated_operation_ids: Sequence[str] | None = None, + execution_input: Any = None, ) -> None: """Fire the invocation-start hook. @@ -466,6 +483,7 @@ def on_invocation_start( ``OperationInfo`` view is deferred to first access. updated_operation_ids: Operation ids from the invocation input's ``UpdatedOperationIds`` -- those updated while suspended. + execution_input: The deserialized execution input event. """ aws_request_id = lambda_context.aws_request_id if lambda_context else None self._operations_provider = operations_provider @@ -476,6 +494,7 @@ def on_invocation_start( is_first_invocation=is_first_invocation, execution_start_time=execution_start_time, operations=operations, + execution_input=execution_input, updated_operations=_LazyOperationInfoMap( lambda: { operation_id: operations[operation_id] 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 ab6789da..165382be 100644 --- a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py @@ -148,6 +148,53 @@ def test_invocation_end_info(self): self.assertEqual(INVOCATION_END_INFO.status, InvocationStatus.FAILED) self.assertEqual(INVOCATION_END_INFO.error.message, "boom") + def test_invocation_info_payload_fields_default_to_none(self): + """Both payload surfaces default to None for backward compatibility.""" + self.assertIsNone(INVOCATION_START_INFO.execution_input) + self.assertIsNone(INVOCATION_END_INFO.execution_input) + self.assertIsNone(INVOCATION_END_INFO.execution_result) + + def test_invocation_start_info_carries_execution_input(self): + info = InvocationStartInfo( + request_id="req-1", + execution_arn="arn:test", + is_first_invocation=True, + execution_input={"order": 7}, + ) + + self.assertEqual({"order": 7}, info.execution_input) + + def test_invocation_end_info_carries_input_and_result(self): + """The end factory forwards the input and picks up the output result.""" + start = InvocationStartInfo( + request_id="req-1", + execution_arn="arn:test", + is_first_invocation=False, + execution_input={"order": 7}, + ) + output = DurableExecutionInvocationOutput( + status=InvocationStatus.SUCCEEDED, result='"done"', error=None + ) + + end = InvocationEndInfo.from_durable_execution_invocation_output(start, output) + + self.assertEqual({"order": 7}, end.execution_input) + self.assertEqual('"done"', end.execution_result) + + def test_invocation_end_info_has_no_result_when_suspended(self): + start = InvocationStartInfo( + request_id="req-1", + execution_arn="arn:test", + is_first_invocation=True, + ) + output = DurableExecutionInvocationOutput( + status=InvocationStatus.PENDING, result=None, error=None + ) + + end = InvocationEndInfo.from_durable_execution_invocation_output(start, output) + + self.assertIsNone(end.execution_result) + def test_invocation_info_operation_maps_default_to_empty(self): """Both maps default to empty so existing constructor calls keep working.""" self.assertEqual({}, INVOCATION_START_INFO.operations) @@ -587,8 +634,8 @@ def test_pending_fires_invocation_end(self): self.assertIn("invocation_end:req-1", self.plugin.calls) -class TestInvocationHookOperationMaps(unittest.TestCase): - """Tests for the operations / updated_operations maps on invocation hooks.""" +class TestInvocationHookInfoFields(unittest.TestCase): + """Tests for the operation maps and payload fields on invocation hooks.""" def setUp(self): self.captured: list[object] = [] @@ -776,6 +823,46 @@ def test_provider_is_released_when_the_run_scope_exits(self): self.assertIsNone(self.executor._operations_provider) # noqa: SLF001 + def test_execution_input_reaches_both_hooks(self): + with self.executor.run(): + self.executor.on_invocation_start( + execution_arn="arn:exec", + lambda_context=LAMBDA_CTX, + execution_start_time=START_TS, + is_first_invocation=True, + execution_input={"order": 7}, + ) + self.executor.on_invocation_end( + output=DurableExecutionInvocationOutput( + status=InvocationStatus.SUCCEEDED, result='"done"', error=None + ) + ) + + start_info, end_info = self.captured + self.assertEqual({"order": 7}, start_info.execution_input) + # The end info inherits the input and adds the result. + self.assertEqual({"order": 7}, end_info.execution_input) + self.assertEqual('"done"', end_info.execution_result) + + def test_suspending_invocation_end_has_no_result(self): + with self.executor.run(): + self.executor.on_invocation_start( + execution_arn="arn:exec", + lambda_context=LAMBDA_CTX, + execution_start_time=START_TS, + is_first_invocation=True, + execution_input="abc", + ) + self.executor.on_invocation_end( + output=DurableExecutionInvocationOutput( + status=InvocationStatus.PENDING, result=None, error=None + ) + ) + + _, end_info = self.captured + self.assertEqual("abc", end_info.execution_input) + self.assertIsNone(end_info.execution_result) + class TestPluginExecutorOnOperationAction(unittest.TestCase): """Tests for PluginExecutor.on_operation_action.""" From a93d2989653f3266aa927f1dba2c7840abf9948f Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Mon, 10 Aug 2026 22:59:18 +0000 Subject: [PATCH 4/4] docs(conformance): correct 10-19 payload field note The handler docstring claimed the Python invocation infos have no execution-input or execution-result field. They do now; the canonical dump still omits them because the requirement puts the payload surfaces out of GA scope and asserts nothing about them. Refs #616 --- .../plugin/plugin_invocation_info_shape.py | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py index a75871bb..11e2a82d 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py @@ -11,11 +11,13 @@ ``isFirstInvocation`` on the end record comes from the invocation-end info. Python surface note: ``InvocationInfo`` carries ``request_id``, -``execution_arn``, ``is_first_invocation``, ``execution_start_time`` and the -``operations`` map; ``InvocationStartInfo`` adds ``updated_operations`` and the -end info adds ``status`` + ``error``. It has NO execution-input or -execution-result field, so the canonical ``executionInput`` / -``executionResult`` keys are omitted — the honest red for those probes. +``execution_arn``, ``is_first_invocation``, ``execution_start_time``, the +``operations`` map and ``execution_input``; ``InvocationStartInfo`` adds +``updated_operations`` and the end info adds ``status``, ``error`` and +``execution_result``. The payload surfaces (``execution_input`` / +``execution_result``) are deliberately NOT dumped here: they are out of GA +conformance scope, so this requirement's canonical schema omits +``executionInput`` / ``executionResult`` and asserts nothing about them. """ import json @@ -42,9 +44,9 @@ def _emit(record: dict[str, Any], execution_arn: str | None) -> None: class InvocationInfoShapePlugin(DurableInstrumentationPlugin): def on_invocation_start(self, info: InvocationStartInfo) -> None: - # Canonical dump of InvocationStartInfo. executionInput is absent from - # the Python type and therefore omitted — that omission is the parity - # signal under test. + # Canonical dump of InvocationStartInfo, minus execution_input: the + # payload surfaces are out of GA scope and this requirement asserts + # nothing about them. record: dict[str, Any] = { "plugin": "CONFPLUGIN", "hook": "invocation-start", @@ -59,8 +61,8 @@ def on_invocation_start(self, info: InvocationStartInfo) -> None: _emit(record, info.execution_arn) def on_invocation_end(self, info: InvocationEndInfo) -> None: - # isFirstInvocation MUST come from the END info itself. executionInput / - # executionResult are absent from the Python type and therefore omitted. + # isFirstInvocation MUST come from the END info itself. execution_input + # and execution_result are omitted for the same out-of-scope reason. status = info.status.name record: dict[str, Any] = { "plugin": "CONFPLUGIN",