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..1e9a73bb 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,7 @@ def wrapper(event: Any, context: LambdaContext) -> MutableMapping[str, Any]: else None ), is_first_invocation=not has_prior_operations, + 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..3f70143c 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 @@ -163,6 +163,11 @@ class InvocationInfo: execution_arn: str | None is_first_invocation: bool execution_start_time: datetime.datetime | None = None + # The deserialized execution input, surfaced to instrumentation plugins that + # need to record it (e.g. Workflow Insight). Mirrors the JS SDK's + # InvocationInfo.executionInput. kw_only so it never reorders the positional + # fields above. None when the input is empty or unavailable. + execution_input: Any = field(default=None, kw_only=True) @dataclass(frozen=True) @@ -174,6 +179,11 @@ class InvocationStartInfo(InvocationInfo): class InvocationEndInfo(InvocationInfo): status: InvocationStatus = field(kw_only=True) error: ErrorObject | None = None + # The serialized execution result (a JSON string, or "" when the result was + # checkpointed out-of-band for a large payload). Surfaced to instrumentation + # plugins that record execution output (e.g. Workflow Insight); mirrors the + # JS SDK's InvocationEndInfo.executionResult. None on failure/suspend. + execution_result: str | None = field(default=None, kw_only=True) @classmethod def from_durable_execution_invocation_output( @@ -186,8 +196,10 @@ def from_durable_execution_invocation_output( 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, + execution_input=invocation_start_info.execution_input, status=output.status, error=output.error, + execution_result=output.result, ) @@ -324,6 +336,7 @@ def on_invocation_start( is_first_invocation: bool, execution_start_time: datetime.datetime | None, lambda_context: LambdaContext | None, + execution_input: Any = None, ) -> None: aws_request_id = lambda_context.aws_request_id if lambda_context else None self._invocation_status = InvocationStartInfo( @@ -331,6 +344,7 @@ def on_invocation_start( request_id=aws_request_id, is_first_invocation=is_first_invocation, execution_start_time=execution_start_time, + execution_input=execution_input, ) self.execute_plugins(self._invocation_status, 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..f5896993 100644 --- a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py @@ -71,6 +71,7 @@ execution_arn="arn:aws:lambda:us-east-1:123:durable:abc", execution_start_time=START_TS, is_first_invocation=True, + execution_input={"name": "World"}, ) INVOCATION_END_INFO = InvocationEndInfo( request_id="req-1", @@ -79,6 +80,8 @@ status=InvocationStatus.FAILED, error=ERROR, is_first_invocation=False, + execution_input={"name": "World"}, + execution_result='"Hello, World!"', ) USER_FUNCTION_START_INFO = UserFunctionStartInfo( @@ -139,6 +142,16 @@ def test_invocation_start_info(self): ) self.assertEqual(INVOCATION_START_INFO.execution_start_time, START_TS) self.assertTrue(INVOCATION_START_INFO.is_first_invocation) + self.assertEqual(INVOCATION_START_INFO.execution_input, {"name": "World"}) + + def test_invocation_info_execution_input_defaults_to_none(self): + info = InvocationStartInfo( + request_id="req-1", + execution_arn="arn:test", + execution_start_time=START_TS, + is_first_invocation=True, + ) + self.assertIsNone(info.execution_input) def test_invocation_end_info(self): self.assertEqual(INVOCATION_END_INFO.request_id, "req-1") @@ -147,6 +160,23 @@ def test_invocation_end_info(self): self.assertFalse(INVOCATION_END_INFO.is_first_invocation) self.assertEqual(INVOCATION_END_INFO.status, InvocationStatus.FAILED) self.assertEqual(INVOCATION_END_INFO.error.message, "boom") + self.assertEqual(INVOCATION_END_INFO.execution_input, {"name": "World"}) + self.assertEqual(INVOCATION_END_INFO.execution_result, '"Hello, World!"') + + def test_invocation_end_info_from_invocation_output_carries_input_and_result(self): + output = DurableExecutionInvocationOutput( + status=InvocationStatus.SUCCEEDED, + result='"Hello, World!"', + ) + end_info = InvocationEndInfo.from_durable_execution_invocation_output( + INVOCATION_START_INFO, output + ) + self.assertEqual(end_info.request_id, INVOCATION_START_INFO.request_id) + self.assertEqual(end_info.execution_arn, INVOCATION_START_INFO.execution_arn) + self.assertEqual(end_info.execution_input, {"name": "World"}) + self.assertEqual(end_info.execution_result, '"Hello, World!"') + self.assertEqual(end_info.status, InvocationStatus.SUCCEEDED) + self.assertIsNone(end_info.error) def test_user_function_start_info(self): self.assertEqual(USER_FUNCTION_START_INFO.operation_id, "op-1")