fix(plugin): exclude payloads from info equality

Addresses the medium-severity Codex review comment on #616.

The payload fields joined the generated __eq__ and __hash__, so the
widening was not additive. execution_input holds arbitrary deserialized
JSON, and a dict or list value made a previously hashable
InvocationStartInfo raise TypeError on hash(); both fields also made
infos built from the earlier field set compare unequal to infos
carrying a payload. Both effects were reproduced first.

Set compare=False, hash=False on execution_input and execution_result.
Identity fields still drive equality, so payload-only differences now
compare equal -- payloads are incidental data on what is otherwise an
event record.

Adds tests for hashability across dict, list, nested and scalar
payloads, for equality against the prior field set, and for the field
declarations themselves.

Note OperationInfo.result / OperationInfo.error from #625 remain in
compare. They are hashable types so they do not break hash(), but the
equality asymmetry with these fields is worth a maintainer decision.

Refs #616

Author: Alex Wang

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

@@ -180,6 +180,8 @@ class InvocationInfo:
         default=None,
         kw_only=True,
         repr=False,
+        compare=False,
+        hash=False,
         metadata={"experimental": True},
     )
     """EXPERIMENTAL: The deserialized execution input, when available.
@@ -193,6 +195,11 @@ class InvocationInfo:
     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.
@@ -216,6 +223,8 @@ class InvocationEndInfo(InvocationInfo):
         default=None,
         kw_only=True,
         repr=False,
+        compare=False,
+        hash=False,
         metadata={"experimental": True},
     )
     """EXPERIMENTAL: The serialized execution result, when available.
@@ -224,9 +233,10 @@ class InvocationEndInfo(InvocationInfo):
     large payload. Mirrors the JS SDK's ``InvocationEndInfo.executionResult``.
     ``None`` on failure or suspend.
 
-    Excluded from ``repr`` for the same reason as
+    Excluded from ``repr``, ``__eq__`` and ``__hash__`` for the same reasons as
     :attr:`InvocationInfo.execution_input`: hook infos are logged wholesale by
-    instrumentation, and the result is customer data.
+    instrumentation, and adding the field should not change how existing infos
+    compare.
     """
 
     @classmethod

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

@@ -161,6 +161,87 @@ def test_payload_fields_are_declared_non_repr(self):
         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,