Add hook-info field-shape conformance handlers

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

@@ -0,0 +1,129 @@
+"""10-21: Attempt hook info field shape (interface-shape probe).
+
+A single step named "flaky" throws on its first attempt and succeeds on the
+second, using the SDK's real retry strategy (``RetryStrategyConfig`` +
+``create_retry_strategy``, max_attempts >= 2, ~1s delay); it returns "ok". The
+plugin emits from the SDK's real per-attempt hooks (``on_user_function_start`` /
+``on_user_function_end``, filtering to step-type operations) a CANONICAL DUMP of
+the CURRENT hook's own info parameter: every field the Python attempt-info type
+exposes is mapped one-to-one to its canonical camelCase name; unset fields
+(value None) are OMITTED (a missing key fails its assertion — the honest parity
+signal); timestamps ISO-8601, errors their message string, ``outcome`` the
+info's own ``UserFunctionOutcome`` token.
+
+Python surface note: ``UserFunctionStartInfo`` / ``UserFunctionEndInfo`` expose
+operation_id, operation_type, sub_type, name, parent_id, start_time, attempt,
+is_replayed and is_replay_children; the end info adds ``end_time``, ``outcome``
+and ``error``. The failure is a genuine thrown exception surfaced through the
+SDK's retry path — never a hand-rolled outcome. (``status`` is an inherited
+constant STARTED on attempt infos and is not part of the attempt schema.)
+"""
+
+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 _dump_attempt(
+    hook: str, info: UserFunctionStartInfo | UserFunctionEndInfo
+) -> dict[str, Any]:
+    # Canonical dump of an attempt info's own field surface. Identity + attempt
+    # number + replay indicators are always present; optional fields are emitted
+    # only when the info populates them (None -> omitted key = honest red).
+    record: dict[str, Any] = {
+        "plugin": "CONFPLUGIN",
+        "hook": hook,
+        "id": info.operation_id,
+        "type": info.operation_type.name.upper(),
+        "isReplay": info.is_replayed,
+        "isReplayingChildren": info.is_replay_children,
+    }
+    if info.name is not None:
+        record["name"] = info.name
+    if info.sub_type is not None:
+        record["subType"] = info.sub_type.name
+    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()
+    return record
+
+
+class AttemptInfoShapePlugin(DurableInstrumentationPlugin):
+    def __init__(self) -> None:
+        # User-function hooks do not carry the execution ARN, so capture it from
+        # the invocation-start hook and stamp it on later attempt emissions.
+        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(_dump_attempt("attempt-start", info), self._execution_arn)
+
+    def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
+        if info.operation_type.name != "STEP":
+            return
+        record = _dump_attempt("attempt-end", info)
+        # The end info adds its own outcome enum and (on failure) an error.
+        record["outcome"] = info.outcome.name
+        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
+        _emit(record, self._execution_arn)
+
+
+@durable_step
+def flaky(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 "ok"
+
+
+@durable_execution(plugins=[AttemptInfoShapePlugin()])
+def handler(_event: Any, context: DurableContext) -> str:
+    retry_config = RetryStrategyConfig(
+        max_attempts=2,
+        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,139 @@
+"""10-23: Context-typed hook info field shape (interface-shape probe).
+
+A parallel operation named "ctx" with max-concurrency 1 and two branches:
+branch A runs a step named "inner" returning "x", then a 2-second wait, and
+returns "a-done"; branch B returns "b-done" directly. The plugin filters to
+CONTEXT-type operations and, from the SDK's real ``on_operation_start`` /
+``on_user_function_start`` hooks, logs ONE single-line JSON record per event: a
+CANONICAL camelCase DUMP of that hook's OWN info parameter. Unset / unexposed
+fields (value None) are OMITTED (a missing key fails its assertion — the honest
+parity signal); sub-type tokens are dumped as the SDK reports them (the
+``OperationSubType`` enum ``.value``: "Parallel" for the parent, "ParallelBranch"
+for each branch).
+
+The operation-start record carries the context operation's ``isReplay`` flag;
+the fn-start record carries ``isReplayingChildren`` from the info's own
+``is_replay_children`` indicator — true when the context function re-runs so its
+checkpointed child operations replay. Attempt-end hooks are NOT probed (end-hook
+semantics for a suspending context run are SDK-divergent and out of scope).
+Every record also carries a ``durableExecutionArn`` field (ARN captured at
+invocation-start, unasserted) so the runner's CloudWatch filter can scope logs
+to the execution.
+"""
+
+import json
+from typing import Any
+
+from aws_durable_execution_sdk_python.config import Duration, 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,
+    OperationStartInfo,
+    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 ContextInfoShapePlugin(DurableInstrumentationPlugin):
+    def __init__(self) -> None:
+        # Operation / user-function 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
+
+    def on_operation_start(self, info: OperationStartInfo) -> None:
+        # Filter to CONTEXT-type operations (the parent parallel + its branches).
+        if info.operation_type.name != "CONTEXT":
+            return
+        # Canonical dump of the OperationStartInfo's own field surface. Identity +
+        # replay flag are always present; optional fields are emitted only when
+        # populated (None -> omitted key = honest missing-field red).
+        record: dict[str, Any] = {
+            "plugin": "CONFPLUGIN",
+            "hook": "operation-start",
+            "id": info.operation_id,
+            "type": info.operation_type.name.upper(),
+            "isReplay": info.is_replayed,
+        }
+        if info.name is not None:
+            record["name"] = info.name
+        if info.sub_type is not None:
+            # Raw SDK token: OperationSubType enum .value ("Parallel"/"ParallelBranch").
+            record["subType"] = info.sub_type.value
+        if info.parent_id is not None:
+            record["parentId"] = info.parent_id
+        if info.status is not None:
+            record["status"] = info.status.name
+        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()
+        _emit(record, self._execution_arn)
+
+    def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
+        # Filter to CONTEXT-type operations (the branch functions).
+        if info.operation_type.name != "CONTEXT":
+            return
+        # Canonical dump of the UserFunctionStartInfo's own field surface. The
+        # children-replay indicator is the probe under test.
+        record: dict[str, Any] = {
+            "plugin": "CONFPLUGIN",
+            "hook": "fn-start",
+            "id": info.operation_id,
+            "type": info.operation_type.name.upper(),
+            "isReplayingChildren": info.is_replay_children,
+        }
+        if info.name is not None:
+            record["name"] = info.name
+        if info.sub_type is not None:
+            # Raw SDK token: OperationSubType enum .value ("Parallel"/"ParallelBranch").
+            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()
+        _emit(record, self._execution_arn)
+
+
+@durable_step
+def inner(_step_context: StepContext) -> str:
+    return "x"
+
+
+def branch_a(ctx: DurableContext) -> str:
+    ctx.step(inner(), name="inner")
+    ctx.wait(Duration.from_seconds(2))
+    return "a-done"
+
+
+def branch_b(_ctx: DurableContext) -> str:
+    return "b-done"
+
+
+@durable_execution(plugins=[ContextInfoShapePlugin()])
+def handler(_event: Any, context: DurableContext) -> list:
+    result = context.parallel(
+        [branch_a, 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,83 @@
+"""10-19: Invocation hook info field shape (interface-shape probe).
+
+A single 2-second wait suspends on the first invocation and completes on
+replay; the handler returns "done-<input>". The plugin emits from the SDK's
+real ``on_invocation_start`` / ``on_invocation_end`` hooks a CANONICAL DUMP of
+the CURRENT hook's own info parameter: every field the Python ``InvocationInfo``
+type exposes is mapped one-to-one to its canonical camelCase name; a field the
+type does not expose is simply OMITTED (a missing key fails its assertion — the
+honest parity signal). One derived scalar is added on the end record:
+``terminal`` := status in (SUCCEEDED, FAILED). No cross-hook reconstruction —
+``isFirstInvocation`` on the end record comes from the invocation-end info.
+
+Python surface note: ``InvocationInfo`` carries only ``request_id``,
+``execution_arn``, ``is_first_invocation`` and ``execution_start_time``; the end
+info adds ``status`` + ``error``. It has NO execution-input, operations-map,
+externally-updated-operations, or execution-result field, so the canonical
+``executionInput`` / ``operationsCount`` / ``updatedOperationsCount`` /
+``executionResult`` keys are omitted — the honest red for those probes.
+"""
+
+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 InvocationInfoShapePlugin(DurableInstrumentationPlugin):
+    def on_invocation_start(self, info: InvocationStartInfo) -> None:
+        # Canonical dump of InvocationStartInfo. executionInput / operationsCount
+        # / updatedOperationsCount are absent from the Python type and therefore
+        # omitted — that omission is the parity signal under test.
+        record: dict[str, Any] = {
+            "plugin": "CONFPLUGIN",
+            "hook": "invocation-start",
+            "isFirstInvocation": info.is_first_invocation,
+        }
+        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:
+        # isFirstInvocation MUST come from the END info itself. executionInput /
+        # operationsCount / executionResult are absent from the Python type and
+        # therefore omitted.
+        status = info.status.name
+        record: dict[str, Any] = {
+            "plugin": "CONFPLUGIN",
+            "hook": "invocation-end",
+            "isFirstInvocation": info.is_first_invocation,
+            "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}"