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..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 @@ -10,12 +10,14 @@ ``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`` / -``executionResult`` keys are omitted — the honest red for those probes. +Python surface note: ``InvocationInfo`` carries ``request_id``, +``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,13 +44,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, 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", "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 @@ -57,14 +61,14 @@ 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 / - # operationsCount / 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", "hook": "invocation-end", "isFirstInvocation": info.is_first_invocation, + "operationsCount": len(info.operations), "status": status, "terminal": status in ("SUCCEEDED", "FAILED"), } 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..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 @@ -304,6 +304,13 @@ 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, + 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 197a9cf3..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 @@ -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,29 +240,71 @@ 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. + """ + 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) 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) 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( 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 + ), + execution_input=invocation_start_info.execution_input, + execution_result=output.result, status=output.status, error=output.error, ) @@ -266,6 +385,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 +398,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 +439,69 @@ 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, + execution_input: Any = 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. + 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 + 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, + execution_input=execution_input, + 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 +513,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 +698,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..165382be 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,116 @@ 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) + 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 +634,236 @@ def test_pending_fires_invocation_end(self): self.assertIn("invocation_end:req-1", self.plugin.calls) +class TestInvocationHookInfoFields(unittest.TestCase): + """Tests for the operation maps and payload fields 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 + + 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."""