Add plugin conformance suite handlers (10-1..10-7)

packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_hooks_retry.py

@@ -0,0 +1,69 @@
+"""10-3: Plugin attempt hooks fire per step attempt with attempt number/outcome.
+
+Uses the SDK's real retry strategy (``RetryStrategyConfig`` + ``create_retry_strategy``)
+so the step fails once then succeeds on the second attempt (mirrors handler
+1-11). The plugin emits its lines from the SDK's real ``on_user_function_start`` /
+``on_user_function_end`` hooks, filtering to STEP-type operations.
+"""
+
+from typing import Any
+
+from aws_durable_execution_sdk_python.config import Duration, StepConfig
+from aws_durable_execution_sdk_python.context import (
+    DurableContext,
+    StepContext,
+    durable_step,
+)
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.plugin import (
+    DurableInstrumentationPlugin,
+    UserFunctionEndInfo,
+    UserFunctionStartInfo,
+)
+from aws_durable_execution_sdk_python.retries import (
+    RetryStrategyConfig,
+    create_retry_strategy,
+)
+
+
+def _is_step(info: UserFunctionStartInfo | UserFunctionEndInfo) -> bool:
+    return info.operation_type.name == "STEP"
+
+
+class AttemptPlugin(DurableInstrumentationPlugin):
+    def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
+        if not _is_step(info):
+            return
+        print(f"CONFPLUGIN attempt-start n={info.attempt}", flush=True)
+
+    def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
+        if not _is_step(info):
+            return
+        print(
+            f"CONFPLUGIN attempt-end n={info.attempt} outcome={info.outcome.name}",
+            flush=True,
+        )
+
+
+@durable_step
+def unreliable_operation(step_context: StepContext) -> str:
+    # Fail on the first attempt, succeed on the second, using the SDK's built-in
+    # durable attempt counter (1-based) from the step context.
+    if step_context.attempt < 2:
+        msg = f"Attempt {step_context.attempt} failed"
+        raise RuntimeError(msg)
+    return "Operation succeeded"
+
+
+@durable_execution(plugins=[AttemptPlugin()])
+def handler(_event: Any, context: DurableContext) -> str:
+    retry_config = RetryStrategyConfig(
+        max_attempts=3,
+        initial_delay=Duration.from_seconds(1),
+        retryable_error_types=[RuntimeError],
+    )
+    result: str = context.step(
+        unreliable_operation(),
+        config=StepConfig(create_retry_strategy(retry_config)),
+    )
+    return result

packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_error_isolation.py

@@ -0,0 +1,75 @@
+"""10-4: Plugin exceptions are swallowed and never affect the execution outcome.
+
+Every plugin hook first logs its line and then raises. The SDK is expected to
+catch and ignore every plugin exception, so the execution result and history are
+identical to running without the plugin. Operation/attempt hooks filter to
+STEP-type operations. No mocking: the isolation guarantee under test is the
+SDK's own hook-dispatch try/except, which we exercise by genuinely raising.
+"""
+
+from typing import Any
+
+from aws_durable_execution_sdk_python.context import (
+    DurableContext,
+    StepContext,
+    durable_step,
+)
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.plugin import (
+    DurableInstrumentationPlugin,
+    InvocationEndInfo,
+    InvocationStartInfo,
+    OperationEndInfo,
+    OperationStartInfo,
+    UserFunctionEndInfo,
+    UserFunctionStartInfo,
+)
+
+
+def _is_step(info: Any) -> bool:
+    return info.operation_type.name == "STEP"
+
+
+class FaultyPlugin(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        print("CONFPLUGIN faulty invocation-start", flush=True)
+        raise RuntimeError("faulty invocation-start")
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        print("CONFPLUGIN faulty invocation-end", flush=True)
+        raise RuntimeError("faulty invocation-end")
+
+    def on_operation_start(self, info: OperationStartInfo) -> None:
+        if not _is_step(info):
+            return
+        print("CONFPLUGIN faulty operation-start", flush=True)
+        raise RuntimeError("faulty operation-start")
+
+    def on_operation_end(self, info: OperationEndInfo) -> None:
+        if not _is_step(info):
+            return
+        print("CONFPLUGIN faulty operation-end", flush=True)
+        raise RuntimeError("faulty operation-end")
+
+    def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
+        if not _is_step(info):
+            return
+        print("CONFPLUGIN faulty attempt-start", flush=True)
+        raise RuntimeError("faulty attempt-start")
+
+    def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
+        if not _is_step(info):
+            return
+        print("CONFPLUGIN faulty attempt-end", flush=True)
+        raise RuntimeError("faulty attempt-end")
+
+
+@durable_step
+def greet(_step_context: StepContext, name: str) -> str:
+    return f"Hello, {name}!"
+
+
+@durable_execution(plugins=[FaultyPlugin()])
+def handler(event: Any, context: DurableContext) -> str:
+    result: str = context.step(greet(event))
+    return result

packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_first_invocation_flag.py

@@ -0,0 +1,34 @@
+"""10-6: Plugin sees is-first-invocation true once, then false on replay.
+
+Uses the SDK's real ``context.wait`` (mirrors handler 2-1) so the execution
+suspends on the first invocation and replays after the timer completes. The
+plugin emits its lines from the invocation-start / invocation-end hooks; the
+terminal invocation-end (SUCCEEDED) fires on the replay invocation.
+"""
+
+from typing import Any
+
+from aws_durable_execution_sdk_python.config import Duration
+from aws_durable_execution_sdk_python.context import DurableContext
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.plugin import (
+    DurableInstrumentationPlugin,
+    InvocationEndInfo,
+    InvocationStartInfo,
+)
+
+
+class FirstInvocationPlugin(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        first = str(info.is_first_invocation).lower()
+        print(f"CONFPLUGIN invocation-start first={first}", flush=True)
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        status = info.status.name if info.status is not None else "NONE"
+        print(f"CONFPLUGIN invocation-end status={status}", flush=True)
+
+
+@durable_execution(plugins=[FirstInvocationPlugin()])
+def handler(_event: Any, context: DurableContext) -> str:
+    context.wait(Duration.from_seconds(2))
+    return "Wait completed"

packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_lifecycle.py

@@ -0,0 +1,43 @@
+"""10-1: Plugin invocation lifecycle hooks (start and end on a single invocation).
+
+Registers an instrumentation plugin through the SDK's real ``plugins=[...]``
+parameter on ``durable_execution``. The plugin emits its lines from the SDK's
+``on_invocation_start`` / ``on_invocation_end`` hooks; the step body logs its
+running line via the SDK-provided step context logger (mirrors handler 1-7).
+"""
+
+from typing import Any
+
+from aws_durable_execution_sdk_python.context import (
+    DurableContext,
+    StepContext,
+    durable_step,
+)
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.plugin import (
+    DurableInstrumentationPlugin,
+    InvocationEndInfo,
+    InvocationStartInfo,
+)
+
+
+class LifecyclePlugin(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        first = str(info.is_first_invocation).lower()
+        print(f"CONFPLUGIN invocation-start first={first}", flush=True)
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        status = info.status.name if info.status is not None else "NONE"
+        print(f"CONFPLUGIN invocation-end status={status}", flush=True)
+
+
+@durable_step
+def greet(step_context: StepContext, name: str) -> str:
+    step_context.logger.info(f"Greeting step running for: {name}")
+    return f"Hello, {name}!"
+
+
+@durable_execution(plugins=[LifecyclePlugin()])
+def handler(event: Any, context: DurableContext) -> str:
+    result: str = context.step(greet(event))
+    return result

packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_multiple_plugins.py

@@ -0,0 +1,49 @@
+"""10-5: Multiple registered plugins all receive lifecycle hooks.
+
+Two instrumentation plugins are registered together, in order A then B, through
+the SDK's real ``plugins=[...]`` parameter. Each emits its own prefixed lines
+from the invocation-start / invocation-end hooks.
+"""
+
+from typing import Any
+
+from aws_durable_execution_sdk_python.context import (
+    DurableContext,
+    StepContext,
+    durable_step,
+)
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.plugin import (
+    DurableInstrumentationPlugin,
+    InvocationEndInfo,
+    InvocationStartInfo,
+)
+
+
+class PluginA(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        print("CONFPLUGIN-A invocation-start", flush=True)
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        status = info.status.name if info.status is not None else "NONE"
+        print(f"CONFPLUGIN-A invocation-end status={status}", flush=True)
+
+
+class PluginB(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        print("CONFPLUGIN-B invocation-start", flush=True)
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        status = info.status.name if info.status is not None else "NONE"
+        print(f"CONFPLUGIN-B invocation-end status={status}", flush=True)
+
+
+@durable_step
+def greet(_step_context: StepContext, name: str) -> str:
+    return f"Hello, {name}!"
+
+
+@durable_execution(plugins=[PluginA(), PluginB()])
+def handler(event: Any, context: DurableContext) -> str:
+    result: str = context.step(greet(event))
+    return result

packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_lifecycle.py

@@ -0,0 +1,48 @@
+"""10-2: Plugin operation lifecycle hooks (step start and terminal end).
+
+The plugin emits its lines from the SDK's real ``on_operation_start`` /
+``on_operation_end`` hooks, filtering to STEP-type operations only via the
+``OperationType`` enum reported on the hook info.
+"""
+
+from typing import Any
+
+from aws_durable_execution_sdk_python.context import (
+    DurableContext,
+    StepContext,
+    durable_step,
+)
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.plugin import (
+    DurableInstrumentationPlugin,
+    OperationEndInfo,
+    OperationStartInfo,
+)
+
+
+def _is_step(info: OperationStartInfo | OperationEndInfo) -> bool:
+    return info.operation_type.name == "STEP"
+
+
+class OperationLifecyclePlugin(DurableInstrumentationPlugin):
+    def on_operation_start(self, info: OperationStartInfo) -> None:
+        if not _is_step(info):
+            return
+        print("CONFPLUGIN operation-start", flush=True)
+
+    def on_operation_end(self, info: OperationEndInfo) -> None:
+        if not _is_step(info):
+            return
+        status = info.status.name if info.status is not None else "NONE"
+        print(f"CONFPLUGIN operation-end status={status}", flush=True)
+
+
+@durable_step
+def greet(_step_context: StepContext, name: str) -> str:
+    return f"Hello, {name}!"
+
+
+@durable_execution(plugins=[OperationLifecyclePlugin()])
+def handler(event: Any, context: DurableContext) -> str:
+    result: str = context.step(greet(event))
+    return result

packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_terminal_failure.py

@@ -0,0 +1,48 @@
+"""10-7: Plugin invocation-end hook receives FAILED status when execution fails.
+
+Uses the SDK's real no-retry strategy (``RetryPresets.none()``, mirrors handler
+1-19) on a step that always throws, so the execution terminates as FAILED. The
+plugin emits its lines from the invocation-start / invocation-end hooks; the
+terminal invocation-end reports the FAILED status.
+"""
+
+from typing import Any
+
+from aws_durable_execution_sdk_python.config import StepConfig
+from aws_durable_execution_sdk_python.context import (
+    DurableContext,
+    StepContext,
+    durable_step,
+)
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.plugin import (
+    DurableInstrumentationPlugin,
+    InvocationEndInfo,
+    InvocationStartInfo,
+)
+from aws_durable_execution_sdk_python.retries import RetryPresets
+
+
+class TerminalFailurePlugin(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        first = str(info.is_first_invocation).lower()
+        print(f"CONFPLUGIN invocation-start first={first}", flush=True)
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        status = info.status.name if info.status is not None else "NONE"
+        print(f"CONFPLUGIN invocation-end status={status}", flush=True)
+
+
+@durable_step
+def failing_step(_step_context: StepContext) -> str:
+    msg = "Something went wrong"
+    raise RuntimeError(msg)
+
+
+@durable_execution(plugins=[TerminalFailurePlugin()])
+def handler(_event: Any, context: DurableContext) -> str:
+    result: str = context.step(
+        failing_step(),
+        config=StepConfig(retry_strategy=RetryPresets.none()),
+    )
+    return result