diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py index 6d10e61f..354b5ccf 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py @@ -356,8 +356,6 @@ def on_operation_start(self, info: OperationStartInfo) -> None: def on_operation_end(self, info: OperationEndInfo) -> None: logger.debug("Durable operation ended: %s", info) - if info.operation_type is OperationType.CONTEXT: - return span = self._get_span(info.operation_id) if span is None: # Cross-invocation stitching: operation started in a prior @@ -477,25 +475,28 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None: raise RuntimeError( "on_user_function_end without matching on_user_function_start" ) - span.set_attributes(self._operation_attributes(info)) - if info.outcome is UserFunctionOutcome.FAILED: - span.set_status(StatusCode.ERROR, info.error.message if info.error else "") - span.record_exception( - Exception( - (info.error.message or info.error.type) - if info.error - else "Unknown error" + if info.operation_type is OperationType.STEP: + span.set_attributes(self._operation_attributes(info)) + if info.outcome is UserFunctionOutcome.FAILED: + span.set_status( + StatusCode.ERROR, info.error.message if info.error else "" ) - ) - else: - span.set_status(StatusCode.OK) + span.record_exception( + Exception( + (info.error.message or info.error.type) + if info.error + else "Unknown error" + ) + ) + else: + span.set_status(StatusCode.OK) - end_time = info.end_time - if end_time is not None and end_time == info.start_time: - end_time += datetime.timedelta(microseconds=1) - popped = self._pop_span(key) - if popped is not None: - popped.end(end_time=_to_otel_timestamp(end_time)) + end_time = info.end_time + if end_time is not None and end_time == info.start_time: + end_time += datetime.timedelta(microseconds=1) + popped = self._pop_span(key) + if popped is not None: + popped.end(end_time=_to_otel_timestamp(end_time)) # Restore the enclosing span as active (parent op, else invocation/workflow). enclosing = ( @@ -519,7 +520,14 @@ def _operation_attributes(self, info: Any) -> dict[str, Any]: attributes["durable.operation.type"] = info.operation_type.value if getattr(info, "sub_type", None) is not None: attributes["durable.operation.subtype"] = info.sub_type.value - if getattr(info, "status", None) is not None: + # STEP user-function spans represent attempts, not durable operations. + if ( + not ( + isinstance(info, (UserFunctionStartInfo, UserFunctionEndInfo)) + and info.operation_type is OperationType.STEP + ) + and getattr(info, "status", None) is not None + ): attributes["durable.operation.status"] = info.status.value if getattr(info, "name", None) is not None: attributes["durable.operation.name"] = info.name diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py index 2ed3da15..4f2744d5 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py @@ -470,9 +470,6 @@ def on_operation_end(self, info: OperationEndInfo) -> None: operation span before being ended. """ logger.debug("Durable operation ended: %s", info) - if info.operation_type is OperationType.CONTEXT: - # Context operations are tracked using on_user_function_end. - return span = self._get_span(info.operation_id) if span is None: # the span was not started in the current invocation, so we need to @@ -548,9 +545,8 @@ def on_user_function_start(self, info: UserFunctionStartInfo) -> None: def on_user_function_end(self, info: UserFunctionEndInfo) -> None: """Called when a context or step operation finishes user code. - This callback records the final attempt status, captures exceptions for - failed attempts, and ends the span that was attached in - ``on_user_function_start``. + STEP attempt spans are finalized here. CONTEXT spans stay open until + ``on_operation_end`` supplies the authoritative durable status. Args: info: Information about the operation attempt. @@ -572,23 +568,26 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None: "on_user_function_end called without matching on_user_function_start" ) - span.set_attributes(self._extract_attributes(info)) - if info.outcome is UserFunctionOutcome.FAILED: - span.set_status(StatusCode.ERROR, info.error.message if info.error else "") - span.record_exception( - Exception( - (info.error.message or info.error.type) - if info.error - else "Unknown error" + if info.operation_type is OperationType.STEP: + span.set_attributes(self._extract_attributes(info)) + if info.outcome is UserFunctionOutcome.FAILED: + span.set_status( + StatusCode.ERROR, info.error.message if info.error else "" ) - ) - else: - span.set_status(StatusCode.OK) + span.record_exception( + Exception( + (info.error.message or info.error.type) + if info.error + else "Unknown error" + ) + ) + else: + span.set_status(StatusCode.OK) - end_timestamp = info.end_time - if end_timestamp is not None and end_timestamp == info.start_time: - end_timestamp += datetime.timedelta(microseconds=1) - self._end_span(span_key, end_timestamp) + end_timestamp = info.end_time + if end_timestamp is not None and end_timestamp == info.start_time: + end_timestamp += datetime.timedelta(microseconds=1) + self._end_span(span_key, end_timestamp) # Restore the enclosing operation span as current so code that runs # after this operation (e.g. between steps in a child context) # correlates to its enclosing operation, not the operation that just @@ -621,7 +620,15 @@ def _extract_attributes(self, info: Any) -> _SpanAttributes: attributes["durable.operation.type"] = info.operation_type.value if hasattr(info, "sub_type") and info.sub_type is not None: attributes["durable.operation.subtype"] = info.sub_type.value - if hasattr(info, "status") and info.status is not None: + # STEP user-function spans represent attempts, not durable operations. + if ( + not ( + isinstance(info, (UserFunctionStartInfo, UserFunctionEndInfo)) + and info.operation_type is OperationType.STEP + ) + and hasattr(info, "status") + and info.status is not None + ): attributes["durable.operation.status"] = info.status.value if hasattr(info, "name") and info.name is not None: attributes["durable.operation.name"] = info.name diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py index 03a56969..8b7aedd1 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py @@ -7,6 +7,7 @@ import opentelemetry.context as otel_context import pytest from aws_durable_execution_sdk_python.lambda_service import ( + ErrorObject, InvocationStatus, OperationStatus, OperationSubType, @@ -17,6 +18,9 @@ InvocationStartInfo, OperationEndInfo, OperationStartInfo, + UserFunctionEndInfo, + UserFunctionOutcome, + UserFunctionStartInfo, ) from opentelemetry import trace from opentelemetry.context import Context @@ -235,6 +239,153 @@ def test_cross_invocation_operation_end_uses_deterministic_span_id(): ) +@pytest.mark.parametrize( + ("outcome", "terminal_status", "error", "expected_span_status"), + [ + ( + UserFunctionOutcome.SUCCEEDED, + OperationStatus.SUCCEEDED, + None, + trace.StatusCode.OK, + ), + ( + UserFunctionOutcome.SUCCEEDED, + OperationStatus.FAILED, + ErrorObject( + message="serialization failed", + type="SerializationError", + data=None, + stack_trace=None, + ), + trace.StatusCode.ERROR, + ), + ], +) +def test_context_span_waits_for_terminal_operation_status( + outcome, + terminal_status, + error, + expected_span_status, +): + plugin, exporter = _create_plugin() + plugin.on_invocation_start(_invocation_start_info()) + operation_id = "context-1" + + plugin.on_user_function_start( + UserFunctionStartInfo( + operation_id=operation_id, + operation_type=OperationType.CONTEXT, + sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT, + name="book-trip", + parent_id=None, + start_time=START_TIME, + is_replayed=False, + status=OperationStatus.STARTED, + is_replay_children=False, + attempt=1, + ) + ) + active_span = plugin._get_span(operation_id) + assert active_span is not None + assert ( + active_span.attributes["durable.operation.status"] + == OperationStatus.STARTED.value + ) + + plugin.on_user_function_end( + UserFunctionEndInfo( + operation_id=operation_id, + operation_type=OperationType.CONTEXT, + sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT, + name="book-trip", + parent_id=None, + start_time=START_TIME, + is_replayed=False, + status=OperationStatus.STARTED, + is_replay_children=False, + attempt=1, + outcome=outcome, + end_time=END_TIME, + error=None, + ) + ) + + assert plugin._get_span(operation_id) is active_span + assert ( + active_span.attributes["durable.operation.status"] + == OperationStatus.STARTED.value + ) + assert not exporter.get_finished_spans() + + plugin.on_operation_end( + OperationEndInfo( + operation_id=operation_id, + operation_type=OperationType.CONTEXT, + sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT, + name="book-trip", + parent_id=None, + start_time=START_TIME, + is_replayed=False, + status=terminal_status, + end_time=END_TIME, + error=error, + ) + ) + + span = exporter.get_finished_spans()[0] + assert span.attributes["durable.operation.status"] == terminal_status.value + assert span.status.status_code is expected_span_status + + +def test_step_attempt_span_omits_operation_status(): + plugin, exporter = _create_plugin() + plugin.on_invocation_start(_invocation_start_info()) + operation_id = "step-1" + + plugin.on_user_function_start( + UserFunctionStartInfo( + operation_id=operation_id, + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="fetch-user", + parent_id=None, + start_time=START_TIME, + is_replayed=False, + status=OperationStatus.STARTED, + is_replay_children=False, + attempt=1, + ) + ) + active_span = plugin._get_span("step-1:attempt:1") + assert active_span is not None + assert "durable.operation.status" not in active_span.attributes + + plugin.on_user_function_end( + UserFunctionEndInfo( + operation_id=operation_id, + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="fetch-user", + parent_id=None, + start_time=START_TIME, + is_replayed=False, + status=OperationStatus.STARTED, + is_replay_children=False, + attempt=1, + outcome=UserFunctionOutcome.SUCCEEDED, + end_time=END_TIME, + error=None, + ) + ) + + span = exporter.get_finished_spans()[0] + assert ( + span.attributes["durable.attempt.outcome"] + == UserFunctionOutcome.SUCCEEDED.value + ) + assert "durable.operation.status" not in span.attributes + + # --------------------------------------------------------------------------- # Default-provider mode: invocation span # --------------------------------------------------------------------------- diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py index f379aa90..c24f091e 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py @@ -9,6 +9,7 @@ import opentelemetry.context as otel_context import pytest from aws_durable_execution_sdk_python.lambda_service import ( + ErrorObject, InvocationStatus, OperationStatus, OperationSubType, @@ -517,6 +518,7 @@ def test_step_operation_span_parents_attempt_span(): == OperationStatus.SUCCEEDED.value ) assert attempt_span.attributes["durable.attempt.number"] == 1 + assert "durable.operation.status" not in attempt_span.attributes def test_user_function_callbacks_emit_attempt_span_attributes(): @@ -540,6 +542,7 @@ def test_user_function_callbacks_emit_attempt_span_attributes(): plugin.on_user_function_start(start_info) active_span = plugin._get_span("step-1:attempt:1") assert active_span is not None + assert "durable.operation.status" not in active_span.attributes active_span.set_attributes( { "durable.operation.name": "stale-name", @@ -575,6 +578,7 @@ def test_user_function_callbacks_emit_attempt_span_attributes(): span.attributes["durable.attempt.outcome"] == UserFunctionOutcome.SUCCEEDED.value ) + assert "durable.operation.status" not in span.attributes def test_step_attempt_span_name_includes_attempt_number(): @@ -661,8 +665,35 @@ def test_step_attempt_span_name_defaults_to_first_attempt(): assert span.name == "fetch-user attempt 1" -def test_context_span_omits_attempt_attributes(): - """CONTEXT operations do not carry per-attempt attributes. +@pytest.mark.parametrize( + ("outcome", "terminal_status", "error", "expected_span_status"), + [ + ( + UserFunctionOutcome.SUCCEEDED, + OperationStatus.SUCCEEDED, + None, + StatusCode.OK, + ), + ( + UserFunctionOutcome.SUCCEEDED, + OperationStatus.FAILED, + ErrorObject( + message="serialization failed", + type="SerializationError", + data=None, + stack_trace=None, + ), + StatusCode.ERROR, + ), + ], +) +def test_context_span_waits_for_terminal_status_and_omits_attempt_attributes( + outcome, + terminal_status, + error, + expected_span_status, +): + """Completed CONTEXT spans carry terminal status and no attempt attributes. durable.attempt.number and durable.attempt.outcome are meaningful for STEP operations (each retry is an attempt) but not for CONTEXT, so the @@ -698,16 +729,41 @@ def test_context_span_omits_attempt_attributes(): status=OperationStatus.STARTED, is_replay_children=False, attempt=1, - outcome=UserFunctionOutcome.SUCCEEDED, + outcome=outcome, end_time=END_TIME, error=None, ) ) + active_span = plugin._get_span(operation_id) + assert active_span is not None + assert ( + active_span.attributes["durable.operation.status"] + == OperationStatus.STARTED.value + ) + assert not exporter.get_finished_spans() + + plugin.on_operation_end( + OperationEndInfo( + operation_id=operation_id, + operation_type=OperationType.CONTEXT, + sub_type=None, + name="book-trip", + parent_id=None, + start_time=START_TIME, + is_replayed=False, + status=terminal_status, + end_time=END_TIME, + error=error, + ) + ) + span = exporter.get_finished_spans()[0] assert span.attributes["durable.operation.type"] == OperationType.CONTEXT.value + assert span.attributes["durable.operation.status"] == terminal_status.value assert "durable.attempt.number" not in span.attributes assert "durable.attempt.outcome" not in span.attributes + assert span.status.status_code is expected_span_status def test_span_registry_helpers_can_be_called_from_multiple_threads(): diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py index f4967338..b1523aa6 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py @@ -14,6 +14,7 @@ from aws_durable_execution_sdk_python.lambda_service import ( ContextOptions, ErrorObject, + OperationStatus, OperationSubType, OperationUpdate, ) @@ -169,6 +170,11 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: self.operation_identifier.operation_id, self.operation_identifier.name, ) + self.state.emit_child_context_end_hook( + self.operation_identifier, + OperationStatus.SUCCEEDED, + is_replayed=checkpointed_result.is_existent(), + ) return raw_result # If in replay_children mode, return without checkpointing @@ -178,6 +184,11 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: self.operation_identifier.operation_id, self.operation_identifier.name, ) + self.state.emit_child_context_end_hook( + self.operation_identifier, + OperationStatus.SUCCEEDED, + is_replayed=True, + ) return raw_result # Serialize result @@ -259,6 +270,13 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: # Must ensure the failure state is persisted before raising the exception. # This guarantees the error is durable and child operations won't be re-executed on replay. self.state.create_checkpoint(operation_update=fail_operation) + else: + self.state.emit_child_context_end_hook( + self.operation_identifier, + OperationStatus.FAILED, + error=error_object, + is_replayed=checkpointed_result.is_existent(), + ) # Reconstruct from the checkpointed error (same path as replay) so # first run and replay surface an identical ChildContextError. 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 c2d24750..197a9cf3 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 @@ -225,9 +225,9 @@ def on_operation_start(self, info: OperationStartInfo) -> None: def on_operation_end(self, info: OperationEndInfo) -> None: """ - Called when an operation checkpoints a terminal status. Terminal - operations are not emitted again during replay. This is called NOT - within the thread that runs operation. + Called when an operation reaches a terminal status. Terminal operations + are not emitted again during replay. Child contexts without a terminal + checkpoint may emit this from the thread that runs the operation. Args: info: Information about the operation. @@ -424,6 +424,32 @@ def on_operation_replay(self, operation: Operation) -> None: ) self.execute_plugins(start_info, sync=True) + def on_child_context_end( + self, + operation_identifier: OperationIdentifier, + status: OperationStatus, + *, + error: ErrorObject | None = None, + is_replayed: bool = False, + ) -> None: + """Execute plugins for a child context that completed without a checkpoint.""" + now = datetime.datetime.now(datetime.UTC) + self.execute_plugins( + OperationEndInfo( + operation_id=operation_identifier.operation_id, + operation_type=operation_identifier.type, + sub_type=operation_identifier.sub_type, + name=operation_identifier.name, + parent_id=operation_identifier.parent_id, + start_time=None, + end_time=now, + status=status, + error=error, + is_replayed=is_replayed, + ), + sync=True, + ) + def on_operation_update( self, operation_or_operations: Operation | Sequence[Operation] | None, diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py index 4fef3353..26aefbe3 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py @@ -485,6 +485,22 @@ def emit_operation_replay_hook(self, operation: Operation) -> None: self._plugin_executor.on_operation_replay(operation) + def emit_child_context_end_hook( + self, + operation_identifier: OperationIdentifier, + status: OperationStatus, + *, + error: ErrorObject | None = None, + is_replayed: bool = False, + ) -> None: + """Fire a terminal hook for a child context completed without a checkpoint.""" + self._plugin_executor.on_child_context_end( + operation_identifier, + status, + error=error, + is_replayed=is_replayed, + ) + def is_operation_updated_since_last_invocation(self, operation_id: str) -> bool: """Return True if an operation changed while this execution was suspended.""" return operation_id in self._updated_operation_ids diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py index 07c7150e..d937a557 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py @@ -21,6 +21,7 @@ from aws_durable_execution_sdk_python.lambda_service import ( ErrorObject, OperationAction, + OperationStatus, OperationSubType, OperationType, ) @@ -104,6 +105,7 @@ def test_child_handler_not_started( assert success_operation.payload == json.dumps("fresh_result") mock_callable.assert_called_once() + mock_state.emit_child_context_end_hook.assert_not_called() def test_child_handler_already_succeeded(): @@ -398,6 +400,7 @@ def test_child_handler_invocation_error_reraised(): for call in mock_state.create_checkpoint.call_args_list ] assert OperationAction.FAIL not in actions + mock_state.emit_child_context_end_hook.assert_not_called() def test_child_handler_execution_error_wrapped(): @@ -784,12 +787,13 @@ def test_child_handler_replay_children_mode() -> None: mock_state.wrap_user_function.return_value = mock_callable child_config: ChildConfig = ChildConfig() + identifier = OperationIdentifier( + "op9", OperationSubType.RUN_IN_CHILD_CONTEXT, None, "test_name" + ) actual_result = child_handler( mock_callable, mock_state, - OperationIdentifier( - "op9", OperationSubType.RUN_IN_CHILD_CONTEXT, None, "test_name" - ), + identifier, child_config, ) @@ -800,6 +804,11 @@ def test_child_handler_replay_children_mode() -> None: mock_state.create_checkpoint.assert_not_called() # Verify get_checkpoint_result called once assert mock_state.get_checkpoint_result.call_count == 1 + mock_state.emit_child_context_end_hook.assert_called_once_with( + identifier, + OperationStatus.SUCCEEDED, + is_replayed=True, + ) def test_small_payload_with_summary_generator(): @@ -958,12 +967,13 @@ def test_child_handler_is_virtual_no_succeed(): config = ChildConfig(is_virtual=True) + identifier = OperationIdentifier( + "op2", OperationSubType.RUN_IN_CHILD_CONTEXT, None, "test_name" + ) result = child_handler( mock_callable, mock_state, - OperationIdentifier( - "op2", OperationSubType.RUN_IN_CHILD_CONTEXT, None, "test_name" - ), + identifier, config, ) @@ -971,6 +981,11 @@ def test_child_handler_is_virtual_no_succeed(): # Verify no checkpoints created mock_state.create_checkpoint.assert_not_called() + mock_state.emit_child_context_end_hook.assert_called_once_with( + identifier, + OperationStatus.SUCCEEDED, + is_replayed=False, + ) mock_callable.assert_called_once() @@ -1017,6 +1032,7 @@ def test_child_handler_not_is_virtual_finish_mode(): assert success_operation.action.value == "SUCCEED" mock_callable.assert_called_once() + mock_state.emit_child_context_end_hook.assert_not_called() def test_child_handler_is_virtual_with_exception(): @@ -1043,13 +1059,14 @@ def test_child_handler_is_virtual_with_exception(): config = ChildConfig(is_virtual=True) + identifier = OperationIdentifier( + "op4", OperationSubType.RUN_IN_CHILD_CONTEXT, None, "test_name" + ) with pytest.raises(ChildContextError): child_handler( mock_callable, mock_state, - OperationIdentifier( - "op4", OperationSubType.RUN_IN_CHILD_CONTEXT, None, "test_name" - ), + identifier, config, ) @@ -1057,6 +1074,12 @@ def test_child_handler_is_virtual_with_exception(): assert mock_state.create_checkpoint.call_count == 0 mock_callable.assert_called_once() + end_call = mock_state.emit_child_context_end_hook.call_args + assert end_call.args[:2] == (identifier, OperationStatus.FAILED) + assert end_call.kwargs["is_replayed"] is False + error = end_call.kwargs["error"] + assert error.type == "ValueError" + assert error.message == "Test error" def test_child_handler_not_is_virtual_with_exception(): @@ -1095,6 +1118,7 @@ def test_child_handler_not_is_virtual_with_exception(): assert fail_operation.action.value == "FAIL" mock_callable.assert_called_once() + mock_state.emit_child_context_end_hook.assert_not_called() def test_child_handler_is_virtual_comparison(): 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 0dff5f5b..97e3d9f5 100644 --- a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py @@ -3,6 +3,7 @@ import unittest from unittest.mock import MagicMock +from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( DurableExecutionInvocationOutput, ErrorObject, @@ -684,6 +685,53 @@ def test_non_terminal_operation_fires_operation_start(self): self.assertEqual(plugin.calls, ["operation_start:op-1"]) +class TestPluginExecutorOnChildContextEnd(unittest.TestCase): + """Tests for locally determined child-context completion.""" + + def test_builds_terminal_operation_info(self): + captured: list[OperationEndInfo] = [] + + class _CapturingPlugin(_TrackingPlugin): + def on_operation_end(self, info: OperationEndInfo) -> None: + super().on_operation_end(info) + captured.append(info) + + plugin = _CapturingPlugin() + executor = PluginExecutor(plugins=[plugin]) + identifier = OperationIdentifier( + operation_id="context-1", + sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT, + parent_id="parent-1", + name="book-trip", + ) + before = datetime.datetime.now(datetime.UTC) + + with executor.run(): + executor.on_child_context_end( + identifier, + OperationStatus.FAILED, + error=ERROR, + is_replayed=True, + ) + + after = datetime.datetime.now(datetime.UTC) + self.assertEqual(plugin.calls, ["operation_end:context-1"]) + self.assertEqual(len(captured), 1) + info = captured[0] + self.assertEqual(info.operation_id, "context-1") + self.assertEqual(info.operation_type, OperationType.CONTEXT) + self.assertEqual(info.sub_type, OperationSubType.RUN_IN_CHILD_CONTEXT) + 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.assertTrue(info.is_replayed) + self.assertIsNone(info.start_time) + assert info.end_time is not None + self.assertLessEqual(before, info.end_time) + self.assertLessEqual(info.end_time, after) + + class TestPluginExecutorOnOperationUpdate(unittest.TestCase): """Tests for PluginExecutor.on_operation_update."""