feat(plugin): surface execution input and result on invocation hooks

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)

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,13 +336,15 @@ 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(
             execution_arn=execution_arn,
             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)
 

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")