Add plugin-suite conformance handlers for requirements 10-8..10-18

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

@@ -0,0 +1,71 @@
+"""10-9: Plugin observes an externally-updated wait on re-invocation.
+
+Python does NOT surface externally-updated operations on the invocation-start
+info. Instead, when a suspended execution is re-invoked, the SDK routes an
+operation that changed while the execution was suspended (its
+UpdatedOperationIds) through ``on_operation_update`` -> ``on_operation_end``
+during replay (see ExecutionState.emit_operation_update_hook /
+DurableContext replay gating). The requirement explicitly allows emitting the
+``updated-on-invoke`` record from whichever hook carries this semantic, so this
+handler emits it from ``on_operation_end`` for wait-type operations.
+
+A 2-second ``context.wait`` always suspends after checkpointing WaitStarted and
+completes externally (the timer fires between invocations), so the wait's
+terminal end only ever fires on the replay invocation via the external-update
+path — never on the first invocation. The ``first`` flag is captured from the
+invocation-start hook of the same invocation, so it is naturally False on replay.
+"""
+
+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,
+    InvocationStartInfo,
+    OperationEndInfo,
+)
+
+
+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 ExternalUpdatePlugin(DurableInstrumentationPlugin):
+    def __init__(self) -> None:
+        self._execution_arn: str | None = None
+        self._is_first: bool = False
+
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        self._execution_arn = info.execution_arn
+        self._is_first = info.is_first_invocation
+
+    def on_operation_end(self, info: OperationEndInfo) -> None:
+        # For a 2s wait, on_operation_end fires exclusively via the
+        # external-update path on the replay invocation (the wait never
+        # completes in-process), so this record is the externally-updated signal.
+        if info.operation_type.name != "WAIT":
+            return
+        _emit(
+            {
+                "plugin": "CONFPLUGIN",
+                "hook": "updated-on-invoke",
+                "op": info.operation_id,
+                "status": info.status.name,
+                "first": self._is_first,
+            },
+            self._execution_arn,
+        )
+
+
+@durable_execution(plugins=[ExternalUpdatePlugin()])
+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_faulty_and_healthy.py

@@ -0,0 +1,193 @@
+"""10-17: Faulty plugin does not affect a healthy plugin.
+
+Two plugins are registered together, in order: a faulty plugin whose every
+exercised hook logs a line and then raises, and a healthy plugin that logs
+normally. The exercised hooks span the full lifecycle: invocation-start,
+operation-start, attempt-start, attempt-end, operation-end, and invocation-end.
+In the Python SDK the per-attempt hooks are the real ``on_user_function_start`` /
+``on_user_function_end`` callbacks (the latter carries the attempt ``outcome``),
+and the operation hooks are ``on_operation_start`` / ``on_operation_end``.
+
+The SDK must isolate each plugin (the faulty plugin's exceptions are swallowed)
+so the healthy plugin still receives every hook and the execution
+result/history are identical to running without the faulty plugin. No mocking:
+the isolation guarantee under test is the SDK's own per-plugin hook-dispatch
+try/except, which we exercise by genuinely raising from every exercised hook.
+"""
+
+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 _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 FaultyPlugin(DurableInstrumentationPlugin):
+    def __init__(self) -> None:
+        # Operation/attempt hooks do not carry the execution ARN; capture it
+        # from invocation-start (before raising) and reuse it so every faulty
+        # record is still execution-scoped for CloudWatch retrieval.
+        self._execution_arn: str | None = None
+
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        self._execution_arn = info.execution_arn
+        _emit(
+            {"plugin": "CONFPLUGIN-FAULTY", "hook": "invocation-start"},
+            info.execution_arn,
+        )
+        raise RuntimeError("faulty invocation-start")
+
+    def on_operation_start(self, info: OperationStartInfo) -> None:
+        if info.operation_type.name != "STEP":
+            return
+        _emit(
+            {"plugin": "CONFPLUGIN-FAULTY", "hook": "operation-start"},
+            self._execution_arn,
+        )
+        raise RuntimeError("faulty operation-start")
+
+    def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
+        if info.operation_type.name != "STEP":
+            return
+        _emit(
+            {"plugin": "CONFPLUGIN-FAULTY", "hook": "attempt-start"},
+            self._execution_arn,
+        )
+        raise RuntimeError("faulty attempt-start")
+
+    def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
+        if info.operation_type.name != "STEP":
+            return
+        _emit(
+            {"plugin": "CONFPLUGIN-FAULTY", "hook": "attempt-end"},
+            self._execution_arn,
+        )
+        raise RuntimeError("faulty attempt-end")
+
+    def on_operation_end(self, info: OperationEndInfo) -> None:
+        if info.operation_type.name != "STEP":
+            return
+        _emit(
+            {"plugin": "CONFPLUGIN-FAULTY", "hook": "operation-end"},
+            self._execution_arn,
+        )
+        raise RuntimeError("faulty operation-end")
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        _emit(
+            {"plugin": "CONFPLUGIN-FAULTY", "hook": "invocation-end"},
+            info.execution_arn,
+        )
+        raise RuntimeError("faulty invocation-end")
+
+
+class HealthyPlugin(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:
+        self._execution_arn = info.execution_arn
+        _emit(
+            {
+                "plugin": "CONFPLUGIN-HEALTHY",
+                "hook": "invocation-start",
+                "first": info.is_first_invocation,
+            },
+            info.execution_arn,
+        )
+
+    def on_operation_start(self, info: OperationStartInfo) -> None:
+        if info.operation_type.name != "STEP":
+            return
+        _emit(
+            {
+                "plugin": "CONFPLUGIN-HEALTHY",
+                "hook": "operation-start",
+                "op": info.operation_id,
+            },
+            self._execution_arn,
+        )
+
+    def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
+        if info.operation_type.name != "STEP":
+            return
+        _emit(
+            {
+                "plugin": "CONFPLUGIN-HEALTHY",
+                "hook": "attempt-start",
+                "op": info.operation_id,
+            },
+            self._execution_arn,
+        )
+
+    def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
+        if info.operation_type.name != "STEP":
+            return
+        _emit(
+            {
+                "plugin": "CONFPLUGIN-HEALTHY",
+                "hook": "attempt-end",
+                "op": info.operation_id,
+                "outcome": info.outcome.name,
+            },
+            self._execution_arn,
+        )
+
+    def on_operation_end(self, info: OperationEndInfo) -> None:
+        if info.operation_type.name != "STEP":
+            return
+        _emit(
+            {
+                "plugin": "CONFPLUGIN-HEALTHY",
+                "hook": "operation-end",
+                "op": info.operation_id,
+                "status": info.status.name,
+            },
+            self._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-HEALTHY",
+                "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=[FaultyPlugin(), HealthyPlugin()])
+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_nested_parent_linkage.py

@@ -0,0 +1,72 @@
+"""10-11: Plugin parent linkage for nested operations.
+
+A child context (``run_in_child_context``) wraps a single step. The plugin emits
+from the SDK's real ``on_operation_end`` hook for every operation that reaches a
+terminal status, reporting the operation id and its parent id (the literal
+string NONE when the info carries no parent). The inner step's parent is the
+child context's operation id; the child context itself has no parent.
+"""
+
+import json
+from typing import Any
+
+from aws_durable_execution_sdk_python.context import (
+    DurableContext,
+    StepContext,
+    durable_step,
+    durable_with_child_context,
+)
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.plugin import (
+    DurableInstrumentationPlugin,
+    InvocationStartInfo,
+    OperationEndInfo,
+)
+
+
+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 ParentLinkagePlugin(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:
+        self._execution_arn = info.execution_arn
+
+    def on_operation_end(self, info: OperationEndInfo) -> None:
+        parent = info.parent_id if info.parent_id else "NONE"
+        _emit(
+            {
+                "plugin": "CONFPLUGIN",
+                "hook": "operation-end",
+                "op": info.operation_id,
+                "parent": parent,
+                "status": info.status.name,
+            },
+            self._execution_arn,
+        )
+
+
+@durable_step
+def greet(_step_context: StepContext, name: str) -> str:
+    return f"Hello, {name}!"
+
+
+@durable_with_child_context
+def child_operation(ctx: DurableContext, name: str) -> str:
+    return ctx.step(greet(name))
+
+
+@durable_execution(plugins=[ParentLinkagePlugin()])
+def handler(event: Any, context: DurableContext) -> str:
+    result: str = context.run_in_child_context(child_operation(str(event)))
+    return result