fix(plugin): make payload fields additive on hook infos

packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py

@@ -179,13 +179,27 @@ class InvocationInfo:
     execution_input: Any = field(
         default=None,
         kw_only=True,
+        repr=False,
+        compare=False,
+        hash=False,
         metadata={"experimental": True},
     )
     """EXPERIMENTAL: The deserialized execution input, when available.
 
     Surfaced to instrumentation plugins that need to record it (e.g. Workflow
     Insight). Mirrors the JS SDK's ``InvocationInfo.executionInput``.
 
+    Excluded from ``repr`` on purpose: instrumentation logs hook infos wholesale
+    (the bundled OTel plugins at debug level, the plugin example at info), so
+    including the payload here would implicitly write customer input -- possibly
+    secrets, possibly megabytes -- into logs. Read the attribute explicitly to
+    record it.
+
+    Excluded from ``__eq__`` and ``__hash__`` so adding it stays additive. The
+    value is arbitrary deserialized JSON, so a dict or list payload would make a
+    previously hashable info unhashable, and comparisons against infos built
+    from the earlier field set would start returning False.
+
     Defaults to ``None`` only when the field is not populated (a hook info built
     without it); ``durable_execution()`` always populates it with the
     deserialized input payload, which is ``{}`` when the payload is empty.
@@ -208,13 +222,21 @@ class InvocationEndInfo(InvocationInfo):
     execution_result: str | None = field(
         default=None,
         kw_only=True,
+        repr=False,
+        compare=False,
+        hash=False,
         metadata={"experimental": True},
     )
     """EXPERIMENTAL: The serialized execution result, when available.
 
     A JSON string, or ``""`` when the result was checkpointed out-of-band for a
     large payload. Mirrors the JS SDK's ``InvocationEndInfo.executionResult``.
     ``None`` on failure or suspend.
+
+    Excluded from ``repr``, ``__eq__`` and ``__hash__`` for the same reasons as
+    :attr:`InvocationInfo.execution_input`: hook infos are logged wholesale by
+    instrumentation, and adding the field should not change how existing infos
+    compare.
     """
 
     @classmethod

packages/aws-durable-execution-sdk-python/tests/plugin_test.py

@@ -115,6 +115,133 @@
 
 
 class TestDataClasses(unittest.TestCase):
+    def test_payload_fields_are_excluded_from_repr(self):
+        """Payload values must not leak into repr.
+
+        Instrumentation logs hook infos wholesale -- the bundled OTel plugins at
+        debug level, the plugin example at info -- so a payload in repr would
+        implicitly write customer input and results into logs.
+        """
+        secret_input = {"password": "hunter2", "ssn": "123-45-6789"}
+        secret_result = '{"token": "sk-live-do-not-log"}'
+
+        start_info = InvocationStartInfo(
+            request_id="req-1",
+            execution_arn="arn:test",
+            is_first_invocation=True,
+            execution_input=secret_input,
+        )
+        end_info = InvocationEndInfo(
+            request_id="req-1",
+            execution_arn="arn:test",
+            is_first_invocation=False,
+            status=InvocationStatus.SUCCEEDED,
+            execution_input=secret_input,
+            execution_result=secret_result,
+        )
+
+        for info in (start_info, end_info):
+            rendered = repr(info)
+            for leaked in ("hunter2", "123-45-6789", "sk-live-do-not-log"):
+                self.assertNotIn(leaked, rendered, f"{type(info).__name__}: {leaked}")
+            # Identity fields are still rendered, so the repr stays useful.
+            self.assertIn("req-1", rendered)
+            self.assertIn("arn:test", rendered)
+
+        # The values remain readable via the attributes themselves.
+        self.assertEqual(secret_input, start_info.execution_input)
+        self.assertEqual(secret_result, end_info.execution_result)
+
+    def test_payload_fields_are_declared_non_repr(self):
+        """Pin the field declaration, not just the rendered string."""
+        start_fields = {f.name: f for f in fields(InvocationStartInfo)}
+        end_fields = {f.name: f for f in fields(InvocationEndInfo)}
+
+        self.assertFalse(start_fields["execution_input"].repr)
+        self.assertFalse(end_fields["execution_input"].repr)
+        self.assertFalse(end_fields["execution_result"].repr)
+
+    def test_payload_fields_do_not_break_hashability(self):
+        """A payload must not make a previously hashable info unhashable.
+
+        ``execution_input`` holds arbitrary deserialized JSON, so a dict or list
+        value would otherwise propagate into the generated ``__hash__`` and
+        raise ``TypeError``.
+        """
+        base = {
+            "request_id": "req-1",
+            "execution_arn": "arn:test",
+            "is_first_invocation": True,
+        }
+
+        for payload in ({"k": "v"}, ["a", "b"], {"nested": {"deep": [1, 2]}}, "plain"):
+            info = InvocationStartInfo(**base, execution_input=payload)
+            # Must not raise, and must match the payload-free hash.
+            self.assertEqual(hash(InvocationStartInfo(**base)), hash(info))
+
+        end_info = InvocationEndInfo(
+            **base,
+            status=InvocationStatus.SUCCEEDED,
+            execution_input={"k": "v"},
+            execution_result='{"big": "payload"}',
+        )
+        self.assertEqual(
+            hash(InvocationEndInfo(**base, status=InvocationStatus.SUCCEEDED)),
+            hash(end_info),
+        )
+
+    def test_payload_fields_are_excluded_from_equality(self):
+        """Adding the payload fields must not change how infos compare.
+
+        Infos built from the earlier field set still compare equal to infos
+        carrying a payload, so this widening stays additive for callers.
+        """
+        base = {
+            "request_id": "req-1",
+            "execution_arn": "arn:test",
+            "is_first_invocation": True,
+        }
+
+        self.assertEqual(
+            InvocationStartInfo(**base),
+            InvocationStartInfo(**base, execution_input={"k": "v"}),
+        )
+        # Two different payloads also compare equal -- payloads are incidental
+        # data, not part of the info's identity.
+        self.assertEqual(
+            InvocationStartInfo(**base, execution_input={"a": 1}),
+            InvocationStartInfo(**base, execution_input={"b": 2}),
+        )
+        self.assertEqual(
+            InvocationEndInfo(**base, status=InvocationStatus.SUCCEEDED),
+            InvocationEndInfo(
+                **base,
+                status=InvocationStatus.SUCCEEDED,
+                execution_input={"k": "v"},
+                execution_result='"result"',
+            ),
+        )
+        # Identity fields still drive inequality.
+        self.assertNotEqual(
+            InvocationStartInfo(**base, execution_input={"k": "v"}),
+            InvocationStartInfo(
+                **{**base, "request_id": "req-2"}, execution_input={"k": "v"}
+            ),
+        )
+
+    def test_payload_fields_are_declared_non_compare(self):
+        """Pin the declarations, not just the observed behaviour."""
+        start_fields = {f.name: f for f in fields(InvocationStartInfo)}
+        end_fields = {f.name: f for f in fields(InvocationEndInfo)}
+
+        for holder, name in (
+            (start_fields, "execution_input"),
+            (end_fields, "execution_input"),
+            (end_fields, "execution_result"),
+        ):
+            self.assertFalse(holder[name].compare, name)
+            self.assertIs(holder[name].hash, False, name)
+
     def test_payload_fields_are_marked_experimental(self):
         plugin_info_types = (
             OperationInfo,