test: add plugin info-shape handlers

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

@@ -0,0 +1,101 @@
+"""10-21: Attempt hook info field shape.
+
+The named step fails on its first built-in durable attempt and succeeds on its
+second under the SDK's real retry strategy. Each user-function hook dumps only
+fields carried by its own info object.
+"""
+
+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,
+    OperationInfo,
+    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:
+    if execution_arn is not None:
+        record = {"durableExecutionArn": execution_arn, **record}
+    print(json.dumps(record), flush=True)
+
+
+def _attempt_record(hook: str, info: OperationInfo) -> dict[str, Any]:
+    record: dict[str, Any] = {
+        "plugin": "CONFPLUGIN",
+        "hook": hook,
+        "id": info.operation_id,
+        "type": info.operation_type.name,
+        "isReplay": info.is_replayed,
+    }
+    if info.name is not None:
+        record["name"] = info.name
+    if info.sub_type is not None:
+        record["subType"] = info.sub_type.value
+    if info.parent_id is not None:
+        record["parentId"] = info.parent_id
+    if info.attempt is not None:
+        record["attempt"] = info.attempt
+    if info.start_time is not None:
+        record["startTimestamp"] = info.start_time.isoformat()
+    if info.end_time is not None:
+        record["endTimestamp"] = info.end_time.isoformat()
+    if info.error is not None and info.error.message is not None:
+        record["error"] = info.error.message
+    return record
+
+
+class AttemptInfoShapePlugin(DurableInstrumentationPlugin):
+    def __init__(self) -> None:
+        self._execution_arn: str | None = None
+
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        self._execution_arn = info.execution_arn
+
+    def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
+        if info.operation_type.name != "STEP":
+            return
+        _emit(_attempt_record("attempt-start", info), self._execution_arn)
+
+    def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
+        if info.operation_type.name != "STEP":
+            return
+        record = _attempt_record("attempt-end", info)
+        record["outcome"] = info.outcome.name
+        _emit(record, self._execution_arn)
+
+
+@durable_step
+def flaky(step_context: StepContext) -> str:
+    if step_context.attempt < 2:
+        raise RuntimeError(f"Attempt {step_context.attempt} failed")
+    return "ok"
+
+
+@durable_execution(plugins=[AttemptInfoShapePlugin()])
+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(
+        flaky(),
+        name="flaky",
+        config=StepConfig(create_retry_strategy(retry_config)),
+    )
+    return result

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

@@ -0,0 +1,107 @@
+"""10-23: Context-typed hook info field shape.
+
+A serialised two-branch parallel operation suspends inside branch-a, causing its
+function to run again and replay its children. The plugin dumps context operation
+and user-function start info directly, including the SDK's children-replay flag.
+"""
+
+import json
+from typing import Any
+
+from aws_durable_execution_sdk_python import BatchResult
+from aws_durable_execution_sdk_python.config import (
+    Duration,
+    ParallelBranch,
+    ParallelConfig,
+)
+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,
+    OperationInfo,
+    OperationStartInfo,
+    UserFunctionStartInfo,
+)
+
+
+def _emit(record: dict[str, Any], execution_arn: str | None) -> None:
+    if execution_arn is not None:
+        record = {"durableExecutionArn": execution_arn, **record}
+    print(json.dumps(record), flush=True)
+
+
+def _context_record(hook: str, info: OperationInfo) -> dict[str, Any]:
+    record: dict[str, Any] = {
+        "plugin": "CONFPLUGIN",
+        "hook": hook,
+        "id": info.operation_id,
+        "type": info.operation_type.name,
+        "status": info.status.name,
+        "isReplay": info.is_replayed,
+    }
+    if info.name is not None:
+        record["name"] = info.name
+    if info.sub_type is not None:
+        record["subType"] = info.sub_type.value
+    if info.parent_id is not None:
+        record["parentId"] = info.parent_id
+    if info.start_time is not None:
+        record["startTimestamp"] = info.start_time.isoformat()
+    if info.end_time is not None:
+        record["endTimestamp"] = info.end_time.isoformat()
+    if info.attempt is not None:
+        record["attempt"] = info.attempt
+    return record
+
+
+class ContextInfoShapePlugin(DurableInstrumentationPlugin):
+    def __init__(self) -> None:
+        self._execution_arn: str | None = None
+
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        self._execution_arn = info.execution_arn
+
+    def on_operation_start(self, info: OperationStartInfo) -> None:
+        if info.operation_type.name != "CONTEXT":
+            return
+        _emit(_context_record("operation-start", info), self._execution_arn)
+
+    def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
+        if info.operation_type.name != "CONTEXT":
+            return
+        record = _context_record("fn-start", info)
+        record["isReplayingChildren"] = info.is_replay_children
+        _emit(record, self._execution_arn)
+
+
+@durable_step
+def inner(_step_context: StepContext) -> str:
+    return "x"
+
+
+def branch_a(context: DurableContext) -> str:
+    context.step(inner(), name="inner")
+    context.wait(Duration.from_seconds(2))
+    return "a-done"
+
+
+def branch_b(_context: DurableContext) -> str:
+    return "b-done"
+
+
+@durable_execution(plugins=[ContextInfoShapePlugin()])
+def handler(_event: Any, context: DurableContext) -> list[str]:
+    result: BatchResult[str] = context.parallel(
+        [
+            ParallelBranch(func=branch_a, name="branch-a"),
+            ParallelBranch(func=branch_b, name="branch-b"),
+        ],
+        name="ctx",
+        config=ParallelConfig(max_concurrency=1),
+    )
+    return result.get_results()

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

@@ -0,0 +1,64 @@
+"""10-19: Invocation hook info field shape.
+
+A two-second durable wait forces one suspension and replay. Each hook logs a
+canonical camelCase dump built only from that hook's own info object; optional
+fields are omitted rather than reconstructed.
+"""
+
+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:
+    if execution_arn is not None:
+        record = {"durableExecutionArn": execution_arn, **record}
+    print(json.dumps(record), flush=True)
+
+
+class InvocationInfoShapePlugin(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        record: dict[str, Any] = {
+            "plugin": "CONFPLUGIN",
+            "hook": "invocation-start",
+            "isFirstInvocation": info.is_first_invocation,
+            "operationsCount": len(info.operations),
+            "updatedOperationsCount": len(info.updated_operations),
+        }
+        if info.request_id is not None:
+            record["requestId"] = info.request_id
+        if info.execution_start_time is not None:
+            record["executionStartTimestamp"] = info.execution_start_time.isoformat()
+        _emit(record, info.execution_arn)
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        status = info.status.name
+        record: dict[str, Any] = {
+            "plugin": "CONFPLUGIN",
+            "hook": "invocation-end",
+            "isFirstInvocation": info.is_first_invocation,
+            "operationsCount": len(info.operations),
+            "status": status,
+            "terminal": status in ("SUCCEEDED", "FAILED"),
+        }
+        if info.request_id is not None:
+            record["requestId"] = info.request_id
+        if info.execution_start_time is not None:
+            record["executionStartTimestamp"] = info.execution_start_time.isoformat()
+        if info.error is not None and info.error.message is not None:
+            record["executionError"] = info.error.message
+        _emit(record, info.execution_arn)
+
+
+@durable_execution(plugins=[InvocationInfoShapePlugin()])
+def handler(event: Any, context: DurableContext) -> str:
+    context.wait(Duration.from_seconds(2))
+    return f"done-{event}"

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

@@ -0,0 +1,89 @@
+"""10-22: Operation-change hook info field shape.
+
+A named step succeeds once. For every step in the hook's updated-operation map,
+the plugin emits hook-level counts and a canonical dump of that delta item.
+"""
+
+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,
+    OperationChangeInfo,
+    OperationInfo,
+)
+
+
+def _emit(record: dict[str, Any], execution_arn: str | None) -> None:
+    if execution_arn is not None:
+        record = {"durableExecutionArn": execution_arn, **record}
+    print(json.dumps(record), flush=True)
+
+
+def _add_operation_fields(record: dict[str, Any], info: OperationInfo) -> None:
+    record.update(
+        {
+            "id": info.operation_id,
+            "type": info.operation_type.name,
+            "status": info.status.name,
+            "isReplay": info.is_replayed,
+        }
+    )
+    if info.name is not None:
+        record["name"] = info.name
+    if info.sub_type is not None:
+        record["subType"] = info.sub_type.value
+    if info.parent_id is not None:
+        record["parentId"] = info.parent_id
+    if info.start_time is not None:
+        record["startTimestamp"] = info.start_time.isoformat()
+    if info.end_time is not None:
+        record["endTimestamp"] = info.end_time.isoformat()
+    if info.result is not None:
+        record["result"] = info.result
+    if info.error is not None and info.error.message is not None:
+        record["error"] = info.error.message
+    if info.attempt is not None:
+        record["attempt"] = info.attempt
+
+
+class OperationChangeShapePlugin(DurableInstrumentationPlugin):
+    def __init__(self) -> None:
+        self._execution_arn: str | None = None
+
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        self._execution_arn = info.execution_arn
+
+    def on_operation_change(self, info: OperationChangeInfo) -> None:
+        for operation_id, operation in info.updated_operations.items():
+            if operation.operation_type.name != "STEP":
+                continue
+            record: dict[str, Any] = {
+                "plugin": "CONFPLUGIN",
+                "hook": "operation-change",
+                "updatedOperationsCount": len(info.updated_operations),
+                "operationsCount": len(info.operations),
+                "inFullMap": operation_id in info.operations,
+            }
+            if info.execution_arn is not None:
+                record["executionArn"] = info.execution_arn
+            _add_operation_fields(record, operation)
+            _emit(record, self._execution_arn)
+
+
+@durable_step
+def greet(_step_context: StepContext) -> str:
+    return "task-a"
+
+
+@durable_execution(plugins=[OperationChangeShapePlugin()])
+def handler(_event: Any, context: DurableContext) -> str:
+    result: str = context.step(greet(), name="greet")
+    return result