diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/__init__.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_hooks_retry.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_hooks_retry.py
new file mode 100644
index 00000000..a8cdc04b
--- /dev/null
+++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_hooks_retry.py
@@ -0,0 +1,103 @@
+"""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.
+"""
+
+import json
+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,
+    InvocationStartInfo,
+    UserFunctionEndInfo,
+    UserFunctionStartInfo,
+)
+from aws_durable_execution_sdk_python.retries import (
+    RetryStrategyConfig,
+    create_retry_strategy,
+)
+
+
+def _emit(record: dict[str, Any], execution_arn: str | None) -> None:
+    # Prefix every plugin record with the execution ARN as a top-level field so
+    # the conformance runner's CloudWatch JSON filter can scope logs to a single
+    # execution. Omit the field when the ARN is unset (never invent a value).
+    if execution_arn:
+        record = {"durableExecutionArn": execution_arn, **record}
+    print(json.dumps(record), flush=True)
+
+
+def _is_step(info: UserFunctionStartInfo | UserFunctionEndInfo) -> bool:
+    return info.operation_type.name == "STEP"
+
+
+class AttemptPlugin(DurableInstrumentationPlugin):
+    def __init__(self) -> None:
+        # User-function hooks do not carry the execution ARN, so capture it from
+        # the invocation-start hook and reuse it for later attempt emissions.
+        self._execution_arn: str | None = None
+
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        # Capture only; this test asserts attempt-start/-end lines exclusively.
+        self._execution_arn = info.execution_arn
+
+    def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
+        if not _is_step(info):
+            return
+        _emit(
+            {
+                "plugin": "CONFPLUGIN",
+                "hook": "attempt-start",
+                "n": info.attempt,
+                "op": info.operation_id,
+            },
+            self._execution_arn,
+        )
+
+    def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
+        if not _is_step(info):
+            return
+        _emit(
+            {
+                "plugin": "CONFPLUGIN",
+                "hook": "attempt-end",
+                "n": info.attempt,
+                "outcome": info.outcome.name,
+                "op": info.operation_id,
+            },
+            self._execution_arn,
+        )
+
+
+@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
diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_error_isolation.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_error_isolation.py
new file mode 100644
index 00000000..0a116f88
--- /dev/null
+++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_error_isolation.py
@@ -0,0 +1,93 @@
+"""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.
+"""
+
+import json
+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"
+
+
+def _emit(hook: str, execution_arn: str | None) -> None:
+    # Prefix every plugin record with the execution ARN as a top-level field so
+    # the conformance runner's CloudWatch JSON filter can scope logs to a single
+    # execution. Omit the field when the ARN is unset (never invent a value).
+    record: dict[str, Any] = {"plugin": "CONFPLUGIN-FAULTY", "hook": hook}
+    if execution_arn:
+        record = {"durableExecutionArn": execution_arn, **record}
+    print(json.dumps(record), flush=True)
+
+
+class FaultyPlugin(DurableInstrumentationPlugin):
+    def __init__(self) -> None:
+        # Operation/attempt hooks do not carry the execution ARN, so capture it
+        # from the invocation-start hook and reuse it for later emissions.
+        self._execution_arn: str | None = None
+
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        # Capture BEFORE raising so operation/attempt hooks can still emit the ARN.
+        self._execution_arn = info.execution_arn
+        _emit("invocation-start", info.execution_arn)
+        raise RuntimeError("faulty invocation-start")
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        _emit("invocation-end", info.execution_arn)
+        raise RuntimeError("faulty invocation-end")
+
+    def on_operation_start(self, info: OperationStartInfo) -> None:
+        if not _is_step(info):
+            return
+        _emit("operation-start", self._execution_arn)
+        raise RuntimeError("faulty operation-start")
+
+    def on_operation_end(self, info: OperationEndInfo) -> None:
+        if not _is_step(info):
+            return
+        _emit("operation-end", self._execution_arn)
+        raise RuntimeError("faulty operation-end")
+
+    def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
+        if not _is_step(info):
+            return
+        _emit("attempt-start", self._execution_arn)
+        raise RuntimeError("faulty attempt-start")
+
+    def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
+        if not _is_step(info):
+            return
+        _emit("attempt-end", self._execution_arn)
+        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
diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_first_invocation_flag.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_first_invocation_flag.py
new file mode 100644
index 00000000..8b5ecefd
--- /dev/null
+++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_first_invocation_flag.py
@@ -0,0 +1,53 @@
+"""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.
+"""
+
+import json
+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,
+)
+
+
+def _emit(record: dict[str, Any], execution_arn: str | None) -> None:
+    # Prefix every plugin record with the execution ARN as a top-level field so
+    # the conformance runner's CloudWatch JSON filter can scope logs to a single
+    # execution. Omit the field when the ARN is unset (never invent a value).
+    if execution_arn:
+        record = {"durableExecutionArn": execution_arn, **record}
+    print(json.dumps(record), flush=True)
+
+
+class FirstInvocationPlugin(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        _emit(
+            {
+                "plugin": "CONFPLUGIN",
+                "hook": "invocation-start",
+                "first": info.is_first_invocation,
+            },
+            info.execution_arn,
+        )
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        status = info.status.name if info.status is not None else "NONE"
+        _emit(
+            {"plugin": "CONFPLUGIN", "hook": "invocation-end", "status": status},
+            info.execution_arn,
+        )
+
+
+@durable_execution(plugins=[FirstInvocationPlugin()])
+def handler(_event: Any, context: DurableContext) -> str:
+    context.wait(Duration.from_seconds(2))
+    return "Wait completed"
diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_lifecycle.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_lifecycle.py
new file mode 100644
index 00000000..ddd22673
--- /dev/null
+++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_lifecycle.py
@@ -0,0 +1,62 @@
+"""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).
+"""
+
+import json
+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,
+)
+
+
+def _emit(record: dict[str, Any], execution_arn: str | None) -> None:
+    # Prefix every plugin record with the execution ARN as a top-level field so
+    # the conformance runner's CloudWatch JSON filter can scope logs to a single
+    # execution. Omit the field when the ARN is unset (never invent a value).
+    if execution_arn:
+        record = {"durableExecutionArn": execution_arn, **record}
+    print(json.dumps(record), flush=True)
+
+
+class LifecyclePlugin(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        _emit(
+            {
+                "plugin": "CONFPLUGIN",
+                "hook": "invocation-start",
+                "first": info.is_first_invocation,
+            },
+            info.execution_arn,
+        )
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        status = info.status.name if info.status is not None else "NONE"
+        _emit(
+            {"plugin": "CONFPLUGIN", "hook": "invocation-end", "status": status},
+            info.execution_arn,
+        )
+
+
+@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
diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_multiple_plugins.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_multiple_plugins.py
new file mode 100644
index 00000000..f9669a8e
--- /dev/null
+++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_multiple_plugins.py
@@ -0,0 +1,71 @@
+"""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.
+"""
+
+import json
+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,
+)
+
+
+def _emit(record: dict[str, Any], execution_arn: str | None) -> None:
+    # Prefix every plugin record with the execution ARN as a top-level field so
+    # the conformance runner's CloudWatch JSON filter can scope logs to a single
+    # execution. Omit the field when the ARN is unset (never invent a value).
+    if execution_arn:
+        record = {"durableExecutionArn": execution_arn, **record}
+    print(json.dumps(record), flush=True)
+
+
+class PluginA(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        _emit(
+            {"plugin": "CONFPLUGIN-A", "hook": "invocation-start"},
+            info.execution_arn,
+        )
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        status = info.status.name if info.status is not None else "NONE"
+        _emit(
+            {"plugin": "CONFPLUGIN-A", "hook": "invocation-end", "status": status},
+            info.execution_arn,
+        )
+
+
+class PluginB(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        _emit(
+            {"plugin": "CONFPLUGIN-B", "hook": "invocation-start"},
+            info.execution_arn,
+        )
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        status = info.status.name if info.status is not None else "NONE"
+        _emit(
+            {"plugin": "CONFPLUGIN-B", "hook": "invocation-end", "status": status},
+            info.execution_arn,
+        )
+
+
+@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
diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_lifecycle.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_lifecycle.py
new file mode 100644
index 00000000..48b25cf4
--- /dev/null
+++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_lifecycle.py
@@ -0,0 +1,83 @@
+"""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.
+"""
+
+import json
+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,
+    InvocationStartInfo,
+    OperationEndInfo,
+    OperationStartInfo,
+)
+
+
+def _emit(record: dict[str, Any], execution_arn: str | None) -> None:
+    # Prefix every plugin record with the execution ARN as a top-level field so
+    # the conformance runner's CloudWatch JSON filter can scope logs to a single
+    # execution. Omit the field when the ARN is unset (never invent a value).
+    if execution_arn:
+        record = {"durableExecutionArn": execution_arn, **record}
+    print(json.dumps(record), flush=True)
+
+
+def _is_step(info: OperationStartInfo | OperationEndInfo) -> bool:
+    return info.operation_type.name == "STEP"
+
+
+class OperationLifecyclePlugin(DurableInstrumentationPlugin):
+    def __init__(self) -> None:
+        # Operation hooks do not carry the execution ARN, so capture it from the
+        # invocation-start hook and reuse it for later operation emissions.
+        self._execution_arn: str | None = None
+
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        # Capture only; this test asserts operation-start/-end lines exclusively.
+        self._execution_arn = info.execution_arn
+
+    def on_operation_start(self, info: OperationStartInfo) -> None:
+        if not _is_step(info):
+            return
+        _emit(
+            {
+                "plugin": "CONFPLUGIN",
+                "hook": "operation-start",
+                "op": info.operation_id,
+            },
+            self._execution_arn,
+        )
+
+    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"
+        _emit(
+            {
+                "plugin": "CONFPLUGIN",
+                "hook": "operation-end",
+                "op": info.operation_id,
+                "status": status,
+            },
+            self._execution_arn,
+        )
+
+
+@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
diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_terminal_failure.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_terminal_failure.py
new file mode 100644
index 00000000..2939d09e
--- /dev/null
+++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_terminal_failure.py
@@ -0,0 +1,67 @@
+"""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.
+"""
+
+import json
+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
+
+
+def _emit(record: dict[str, Any], execution_arn: str | None) -> None:
+    # Prefix every plugin record with the execution ARN as a top-level field so
+    # the conformance runner's CloudWatch JSON filter can scope logs to a single
+    # execution. Omit the field when the ARN is unset (never invent a value).
+    if execution_arn:
+        record = {"durableExecutionArn": execution_arn, **record}
+    print(json.dumps(record), flush=True)
+
+
+class TerminalFailurePlugin(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        _emit(
+            {
+                "plugin": "CONFPLUGIN",
+                "hook": "invocation-start",
+                "first": info.is_first_invocation,
+            },
+            info.execution_arn,
+        )
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        status = info.status.name if info.status is not None else "NONE"
+        _emit(
+            {"plugin": "CONFPLUGIN", "hook": "invocation-end", "status": status},
+            info.execution_arn,
+        )
+
+
+@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
diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml b/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml
new file mode 100644
index 00000000..8577409f
--- /dev/null
+++ b/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml
@@ -0,0 +1,135 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: AWS::Serverless-2016-10-31
+Globals:
+  Function:
+    Runtime: python3.13
+    Timeout: 60
+    MemorySize: 128
+Resources:
+  DurableFunctionRole:
+    Type: AWS::IAM::Role
+    Properties:
+      AssumeRolePolicyDocument:
+        Version: '2012-10-17'
+        Statement:
+        - Effect: Allow
+          Principal:
+            Service: lambda.amazonaws.com
+          Action: sts:AssumeRole
+      ManagedPolicyArns:
+      - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
+      Policies:
+      - PolicyName: DurableExecutionPolicy
+        PolicyDocument:
+          Version: '2012-10-17'
+          Statement:
+          - Effect: Allow
+            Action:
+            - lambda:CheckpointDurableExecution
+            - lambda:GetDurableExecutionState
+            Resource: '*'
+  PluginInvocationLifecycle:
+    Type: AWS::Serverless::Function
+    TestingMetadata:
+      TestDescription: ["10-1"]
+    Properties:
+      CodeUri: lambda-build/
+      Handler: plugin.plugin_invocation_lifecycle.handler
+      Description: Plugin invocation lifecycle hooks (start and end)
+      Role:
+        Fn::GetAtt:
+        - DurableFunctionRole
+        - Arn
+      DurableConfig:
+        RetentionPeriodInDays: 7
+        ExecutionTimeout: 300
+  PluginOperationLifecycle:
+    Type: AWS::Serverless::Function
+    TestingMetadata:
+      TestDescription: ["10-2"]
+    Properties:
+      CodeUri: lambda-build/
+      Handler: plugin.plugin_operation_lifecycle.handler
+      Description: Plugin operation lifecycle hooks (step start and terminal end)
+      Role:
+        Fn::GetAtt:
+        - DurableFunctionRole
+        - Arn
+      DurableConfig:
+        RetentionPeriodInDays: 7
+        ExecutionTimeout: 300
+  PluginAttemptHooksRetry:
+    Type: AWS::Serverless::Function
+    TestingMetadata:
+      TestDescription: ["10-3"]
+    Properties:
+      CodeUri: lambda-build/
+      Handler: plugin.plugin_attempt_hooks_retry.handler
+      Description: Plugin attempt hooks fire per step attempt under retry
+      Role:
+        Fn::GetAtt:
+        - DurableFunctionRole
+        - Arn
+      DurableConfig:
+        RetentionPeriodInDays: 7
+        ExecutionTimeout: 300
+  PluginErrorIsolation:
+    Type: AWS::Serverless::Function
+    TestingMetadata:
+      TestDescription: ["10-4"]
+    Properties:
+      CodeUri: lambda-build/
+      Handler: plugin.plugin_error_isolation.handler
+      Description: Plugin exceptions are swallowed and never affect the outcome
+      Role:
+        Fn::GetAtt:
+        - DurableFunctionRole
+        - Arn
+      DurableConfig:
+        RetentionPeriodInDays: 7
+        ExecutionTimeout: 300
+  PluginMultiplePlugins:
+    Type: AWS::Serverless::Function
+    TestingMetadata:
+      TestDescription: ["10-5"]
+    Properties:
+      CodeUri: lambda-build/
+      Handler: plugin.plugin_multiple_plugins.handler
+      Description: Multiple registered plugins all receive lifecycle hooks
+      Role:
+        Fn::GetAtt:
+        - DurableFunctionRole
+        - Arn
+      DurableConfig:
+        RetentionPeriodInDays: 7
+        ExecutionTimeout: 300
+  PluginFirstInvocationFlag:
+    Type: AWS::Serverless::Function
+    TestingMetadata:
+      TestDescription: ["10-6"]
+    Properties:
+      CodeUri: lambda-build/
+      Handler: plugin.plugin_first_invocation_flag.handler
+      Description: Plugin sees is-first-invocation true once, then false on replay
+      Role:
+        Fn::GetAtt:
+        - DurableFunctionRole
+        - Arn
+      DurableConfig:
+        RetentionPeriodInDays: 7
+        ExecutionTimeout: 300
+  PluginTerminalFailure:
+    Type: AWS::Serverless::Function
+    TestingMetadata:
+      TestDescription: ["10-7"]
+    Properties:
+      CodeUri: lambda-build/
+      Handler: plugin.plugin_terminal_failure.handler
+      Description: Plugin invocation-end hook receives FAILED status on failure
+      Role:
+        Fn::GetAtt:
+        - DurableFunctionRole
+        - Arn
+      DurableConfig:
+        RetentionPeriodInDays: 7
+        ExecutionTimeout: 300