fix(plugin): send one end notification per invocation

The success dispatch of on_invocation_end sat inside the try that reports a
handler failure, so an end hook that raised was caught as though the handler
had failed and the hook ran a second time with a RETRY outcome. Later
plugins were told the wrong thing about an invocation that succeeded, and an
exporter exported twice. A hook can raise: _dispatch_plugin re-raises the
three exceptions that instruct the calling thread to stop, and from_dict can
reject an output the handler built.

The output is parsed inside the try and the success hook is dispatched after
it, so each invocation produces exactly one end notification.

Registration also rejects more factory-class shapes. A signature bind
accepts create_plugin(self, info=None) and create_plugin(self, *args),
because the probe binds to self and what remains is satisfied, so
plugins=[MyFactory] still passed and then failed on every invocation where
the error is swallowed. For a class the kind of the member now decides:
a classmethod carries __self__, a staticmethod is identified through its
descriptor, and an attribute holding a callable object takes no implicit
first argument. Anything else read off a class takes self and is rejected.
inspect.getattr_static walks the MRO without running a descriptor, so this
still runs no factory code.

Author: ParidelPooya

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

@@ -1094,13 +1094,18 @@ def decorator(
             @functools.wraps(func)
             def wrapper(event: Any, context: LambdaContext):
                 with self.invocation() as plugin_executor:
+                    # The end hook is dispatched exactly once per invocation, so
+                    # the success dispatch sits outside the try. Inside it, an
+                    # end hook that raised would be caught as though the handler
+                    # had failed, and the hook would run a second time with a
+                    # RETRY outcome -- telling later plugins the wrong thing about
+                    # an invocation that succeeded, and letting an exporter export
+                    # twice. A hook can raise: _dispatch_plugin re-raises the three
+                    # exceptions that instruct the calling thread to stop, and
+                    # from_dict below can reject an output the handler built.
                     try:
                         output = func(event, context, plugin_executor)
-
-                        plugin_executor.on_invocation_end(
-                            output=DurableExecutionInvocationOutput.from_dict(output),
-                        )
-                        return output
+                        completed = DurableExecutionInvocationOutput.from_dict(output)
                     except BaseException as e:
                         # Every exit fires the end hook, not only the ones that
                         # derive from Exception. A handler that surfaces an
@@ -1126,6 +1131,8 @@ def wrapper(event: Any, context: LambdaContext):
                             ),
                         )
                         raise
+                    plugin_executor.on_invocation_end(output=completed)
+                    return output
 
             return wrapper
 

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

@@ -73,9 +73,12 @@ def _is_plugin_factory(value: object) -> bool:
     plain function whose first parameter is ``self``, which is callable. The
     per-invocation call supplies only the info, Python binds it to ``self``, and
     the resulting :exc:`TypeError` is contained like any other factory failure:
-    telemetry is silently absent for the lifetime of the function. Binding one
-    positional argument to the signature rejects that at registration instead.
-    The bind is a signature operation, so no factory code runs.
+    telemetry is silently absent for the lifetime of the function. Two checks
+    reject that at registration instead. For a class, the *kind* of the member
+    decides, because a signature bind cannot tell ``create_plugin(self, info)``
+    from ``create_plugin(info)``: see :func:`_is_unbound_instance_method`. For
+    everything else, binding one positional argument to the signature rejects a
+    member that cannot receive the info. Neither check runs factory code.
 
     A callable with no introspectable signature -- a C-implemented callable, for
     example -- is accepted on the member alone. ``inspect.signature`` raises for
@@ -94,9 +97,45 @@ def _is_plugin_factory(value: object) -> bool:
     create_plugin = getattr(value, "create_plugin", None)
     if not callable(create_plugin):
         return False
+    if isinstance(value, type) and _is_unbound_instance_method(value, create_plugin):
+        return False
     return _accepts_one_positional_argument(create_plugin)
 
 
+def _is_unbound_instance_method(cls: type, create_plugin: object) -> bool:
+    """Report whether a class's ``create_plugin`` is an instance method.
+
+    Read off the class, an instance method is a plain function whose first
+    parameter is ``self``, so the per-invocation call binds the info to ``self``
+    and the factory never sees it. The signature bind below cannot catch every
+    such shape: ``create_plugin(self, info=None)`` and
+    ``create_plugin(self, *args)`` both bind one argument to ``self`` and leave
+    the rest satisfied. The kind of the member decides it instead.
+
+    Three shapes read off a class are usable and none of them is a plain
+    function. A ``@classmethod`` is already bound to the class, so it carries
+    ``__self__``. A ``@staticmethod`` is a plain function, but its descriptor says
+    it takes no implicit first argument. And an attribute holding a callable
+    object -- ``create_plugin = SomeCallable()`` -- is not a function at all and
+    takes no implicit first argument either. Anything else read off a class takes
+    ``self`` and cannot serve.
+
+    :func:`inspect.getattr_static` is what distinguishes the ``@staticmethod``,
+    because it returns the descriptor rather than what reading the attribute
+    produces. It walks the MRO without running any descriptor, so no factory code
+    runs here.
+    """
+    if getattr(create_plugin, "__self__", None) is not None:
+        return False
+    if not inspect.isfunction(create_plugin):
+        return False
+    try:
+        declared = inspect.getattr_static(cls, "create_plugin")
+    except AttributeError:
+        return False
+    return not isinstance(declared, staticmethod)
+
+
 def _accepts_one_positional_argument(create_plugin: object) -> bool:
     """Report whether one positional argument can be bound to a callable.
 

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

@@ -45,6 +45,20 @@ def create_plugin(self, info: InvocationStartInfo) -> _PluginB:
         return _PluginB()
 
 
+class _DefaultedArgumentFactory:
+    """Instance method whose info parameter has a default, so one argument binds."""
+
+    def create_plugin(self, info: InvocationStartInfo | None = None) -> _PluginA:
+        return _PluginA()
+
+
+class _VariadicFactory:
+    """Instance method taking ``*args``, so any argument count binds."""
+
+    def create_plugin(self, *args: object) -> _PluginA:
+        return _PluginA()
+
+
 _plugin_a_factory = _PluginAFactory()
 _plugin_b_factory = _PluginBFactory()
 
@@ -579,26 +593,49 @@ def create_plugin(cls, info: InvocationStartInfo) -> _ClassFactoryPlugin:
     assert plugin.info is INVOCATION_START_INFO
 
 
-def test_explicit_factory_class_with_an_instance_method_is_rejected() -> None:
-    """The factory class is not the factory, and callability does not reveal it.
-
-    ``MyFactory.create_plugin`` read off the class is a plain function whose
-    first parameter is ``self``, so it is callable and used to pass. The
-    per-invocation call supplies only the info, Python binds it to ``self``, and
-    the resulting ``TypeError`` is contained like any other factory failure:
-    instrumentation is silently absent for the lifetime of the function. The
-    signature is bound at registration so the mistake fails here instead.
+@pytest.mark.parametrize(
+    "factory_class",
+    [
+        _PluginAFactory,
+        _DefaultedArgumentFactory,
+        _VariadicFactory,
+    ],
+    ids=["plain", "defaulted", "variadic"],
+)
+def test_explicit_factory_class_with_an_instance_method_is_rejected(
+    factory_class: type,
+) -> None:
+    """The factory class is not the factory, and no signature shape rescues it.
+
+    ``MyFactory.create_plugin`` read off the class is a plain function whose first
+    parameter is ``self``, so the per-invocation call binds the info to ``self``
+    and the factory never sees it. A signature bind alone does not catch every
+    such shape: ``create_plugin(self, info=None)`` and
+    ``create_plugin(self, *args)`` both bind one argument to ``self`` and leave
+    the rest satisfied, so they used to pass and then fail on every invocation
+    where the error is swallowed. The kind of the member decides instead.
     """
     with pytest.raises(PluginLoadError) as error:
-        load_configured_plugins([_PluginAFactory], environment={})  # type: ignore[list-item]
+        load_configured_plugins([factory_class], environment={})  # type: ignore[list-item]
 
     assert "plugins[0]" in str(error.value)
     assert "a factory instance rather than the factory class" in str(error.value)
 
 
-def test_discovery_rejects_a_factory_class_at_the_entry_point() -> None:
-    """The entry-point path applies the same signature check."""
-    entry_point = _FakeEntryPoint("a", _PluginAFactory)
+@pytest.mark.parametrize(
+    "factory_class",
+    [
+        _PluginAFactory,
+        _DefaultedArgumentFactory,
+        _VariadicFactory,
+    ],
+    ids=["plain", "defaulted", "variadic"],
+)
+def test_discovery_rejects_a_factory_class_at_the_entry_point(
+    factory_class: type,
+) -> None:
+    """The entry-point path applies the same rule."""
+    entry_point = _FakeEntryPoint("a", factory_class)
 
     with (
         patch(
@@ -616,6 +653,28 @@ def test_discovery_rejects_a_factory_class_at_the_entry_point() -> None:
     assert "not the factory class" in str(error.value)
 
 
+def test_explicit_class_holding_a_callable_create_plugin_is_accepted() -> None:
+    """A class attribute holding a callable takes no implicit first argument.
+
+    Reading it off the class produces the callable itself, so the info reaches it.
+    """
+
+    class _CallableMember:
+        def __call__(self, info: InvocationStartInfo) -> _PluginA:
+            return _PluginA()
+
+    class _MemberFactory:
+        create_plugin = _CallableMember()
+
+    result = load_configured_plugins([_MemberFactory], environment={})  # type: ignore[list-item]
+
+    assert result == [_MemberFactory]
+    assert isinstance(
+        _MemberFactory.create_plugin(INVOCATION_START_INFO),
+        _PluginA,
+    )
+
+
 def test_explicit_class_declaring_a_static_create_plugin_is_accepted() -> None:
     """A ``@staticmethod`` presents the signature the SDK calls, so it binds."""
 

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

@@ -647,6 +647,36 @@ def create_plugin(self, info: InvocationStartInfo) -> _TrackingPlugin:
         self.assertEqual(built[0].calls, ["invocation_start:req-1"])
         self.assertEqual(built[1].calls, ["invocation_start:req-2"])
 
+    def test_an_end_hook_that_stops_the_thread_is_not_reported_twice(self):
+        """Exactly one end notification per invocation, whatever the hook does.
+
+        ``_dispatch_plugin`` re-raises the three exceptions that instruct the
+        calling thread to stop, so a hook can raise. With the success dispatch
+        inside the try, that raise was caught as a handler failure and the hook
+        ran a second time with a RETRY outcome -- the wrong outcome for an
+        invocation that succeeded, and a second export for an exporter.
+        """
+        plugin = _ControlOnEndPlugin()
+        host = PluginHost(plugins=[plugin_factory(plugin)])
+
+        @host.handle_durable_output
+        def handler(event, context, plugin_executor):
+            plugin_executor.on_invocation_start(
+                execution_arn="arn:exec",
+                lambda_context=LAMBDA_CTX,
+                execution_start_time=START_TS,
+                is_first_invocation=True,
+            )
+            return {
+                "Status": ServiceInvocationStatus.SUCCEEDED.value,
+                "Result": None,
+            }
+
+        with self.assertRaises(KeyboardInterrupt):
+            handler({}, LAMBDA_CTX)
+
+        self.assertEqual(plugin.end_statuses, [InvocationStatus.SUCCEEDED])
+
     def test_a_body_raising_cancellation_still_fires_the_end_hook(self):
         """Every exit fires the end hook, not only the ones deriving from Exception.
 
@@ -2294,6 +2324,17 @@ def on_operation_start(self, info):
         raise asyncio.CancelledError("hook cancelled")
 
 
+class _ControlOnEndPlugin(DurableInstrumentationPlugin):
+    """Plugin whose end hook records the outcome and then stops the thread."""
+
+    def __init__(self) -> None:
+        self.end_statuses: list[InvocationStatus] = []
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        self.end_statuses.append(info.status)
+        raise KeyboardInterrupt
+
+
 class _FailingPlugin(DurableInstrumentationPlugin):
     """Plugin that raises on every hook call."""