From 0381aa1b34716138b88eb2e1d0bd9381fcb69842 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Thu, 6 Aug 2026 21:16:21 +0000 Subject: [PATCH 1/3] test: add hook-info field-shape handlers 10-19..22 --- .../plugin/plugin_attempt_info_shape.py | 118 ++++++++++++++++++ .../plugin/plugin_invocation_info_shape.py | 83 ++++++++++++ .../plugin/plugin_operation_change_shape.py | 87 +++++++++++++ .../plugin/plugin_operation_info_shape.py | 102 +++++++++++++++ .../template_plugin.yaml | 60 +++++++++ 5 files changed, 450 insertions(+) create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py new file mode 100644 index 00000000..e6609820 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py @@ -0,0 +1,118 @@ +"""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. Every logged field +is read from the CURRENT hook's own info parameter — never reconstructed from +another hook or from plugin state. When the Python info type does not expose a +field, the plugin logs the corresponding ``has_*`` flag as false; that omission +is the honest signal of a missing API surface. + +Python surface note: ``UserFunctionStartInfo`` / ``UserFunctionEndInfo`` expose +operation_id, name, operation_type, attempt and start_time; the end info adds +``outcome`` (a ``UserFunctionOutcome`` enum) and ``error``. The failure is a +genuine thrown exception surfaced through the SDK's retry path — never a +hand-rolled outcome. +""" + +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) + + +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( + { + "plugin": "CONFPLUGIN", + "hook": "attempt-start", + "op": info.operation_id, + "name": info.name, + "type": info.operation_type.name.upper(), + "attempt": info.attempt, + "has_start_time": info.start_time is not None, + }, + self._execution_arn, + ) + + def on_user_function_end(self, info: UserFunctionEndInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "attempt-end", + "op": info.operation_id, + "name": info.name, + "type": info.operation_type.name.upper(), + "attempt": info.attempt, + # outcome as reported by the info's own outcome enum — a + # presentation of the API's data, not a reconstruction. + "outcome": info.outcome.name, + "has_error": info.error is not None, + }, + 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 diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py new file mode 100644 index 00000000..8011ed92 --- /dev/null +++ b/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-". The plugin emits from the SDK's +real ``on_invocation_start`` / ``on_invocation_end`` hooks. Every logged field +is read from the CURRENT hook's own info parameter — never reconstructed from +another hook or from plugin state. When the Python ``InvocationInfo`` type does +not expose a field, the plugin logs the corresponding ``has_*`` flag as false +and omits the value key; that omission is the honest signal of a missing API +surface (the reference field set is the union across SDKs). + +Python surface note: ``InvocationInfo`` carries only ``request_id``, +``execution_arn``, ``is_first_invocation`` and ``execution_start_time``. It has +NO execution-input, operations-map, or externally-updated-operations field, and +``InvocationEndInfo`` adds only ``status`` + ``error`` (no execution-result +field). Those absences are surfaced faithfully as has_*: false. +""" + +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: + # Probe the invocation-start info surface. Python's InvocationInfo has + # no execution-input, operations-map, or updated-operations field, so + # those has_* flags are honestly false and the "input" key is omitted. + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "invocation-start", + "first": info.is_first_invocation, + "has_request_id": info.request_id is not None, + "has_input": False, + "has_operations": False, + "updated_nonempty": False, + "has_start_time": info.execution_start_time is not None, + }, + info.execution_arn, + ) + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + # "first" MUST come from the END info itself (its presence there is what + # is under test) — never captured at invocation-start. + status = info.status.name if info.status is not None else "NONE" + terminal = status in ("SUCCEEDED", "FAILED") + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "invocation-end", + "first": info.is_first_invocation, + "terminal": terminal, + "status": status, + # InvocationEndInfo exposes no execution-result field. + "has_result": False, + "has_error": info.error is not None, + }, + info.execution_arn, + ) + + +@durable_execution(plugins=[InvocationInfoShapePlugin()]) +def handler(event: Any, context: DurableContext) -> str: + context.wait(Duration.from_seconds(2)) + return f"done-{event}" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py new file mode 100644 index 00000000..056dca88 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py @@ -0,0 +1,87 @@ +"""10-22: Operation-change hook info field shape (interface-shape probe). + +A single step named "greet" returns the constant "task-a" and succeeds on the +first attempt. The plugin implements the SDK's real ``on_operation_change`` +hook and, for each step-type operation in the change info's updated-operations +delta, probes the DELTA ITEM's own field surface. Every logged field is read +from the CURRENT hook's own info parameter (the change info and its delta +items) — never reconstructed from another hook or from plugin state. When the +Python type does not expose a field, the plugin logs the corresponding +``has_*`` flag as false; that omission is the honest signal of a missing API +surface. + +Python surface note: change-delta items are full ``OperationInfo`` objects +(identity + status + payloads), so the item exposes result, end_time, attempt +and an ``is_replayed`` replay indicator; ``OperationChangeInfo`` itself carries +the execution ARN. +""" + +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, +) + + +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 OperationChangeShapePlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # durableExecutionArn stamping is captured at invocation-start; has_arn + # below is probed from the change info's OWN execution_arn field. + 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 op_id, op in info.updated_operations.items(): + if op.operation_type.name != "STEP": + continue + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "operation-change", + "op": op_id, + "status": op.status.name, + "in_full_map": op_id in info.operations, + # has_arn probes whether the change info itself carries the + # execution ARN (its own field, read directly here). + "has_arn": info.execution_arn is not None, + "item_name": op.name, + "item_type": op.operation_type.name.upper(), + "item_has_result": op.result is not None, + "item_has_end_time": op.end_time is not None, + "item_has_attempt": op.attempt is not None, + # A replay indicator field exists on the item type itself. + "item_has_replay": hasattr(op, "is_replayed"), + }, + 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 diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py new file mode 100644 index 00000000..39a9bf1b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py @@ -0,0 +1,102 @@ +"""10-20: Operation hook info field shape (interface-shape probe). + +A single step named "greet" returns the constant "task-a" and succeeds on the +first attempt. The plugin emits from the SDK's real ``on_operation_start`` / +``on_operation_end`` hooks, filtering to step-type operations. Every logged +field is read from the CURRENT hook's own info parameter — never reconstructed +from another hook or from plugin state. When the Python ``OperationInfo`` type +does not expose a field, the plugin logs the corresponding ``has_*`` flag as +false; that omission is the honest signal of a missing API surface. + +Python surface note: ``OperationInfo`` exposes operation_id, operation_type, +name, start_time, is_replayed, status, end_time, result, error and attempt, so +the full operation-end field set is available. ``has_status`` is emitted at +operation-start for observability but not asserted (status population on a live +first start varies by SDK). +""" + +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) + + +class OperationInfoShapePlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # Operation hooks do not carry the execution ARN, so capture it from the + # invocation-start hook and stamp it on 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_start(self, info: OperationStartInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "operation-start", + "op": info.operation_id, + "name": info.name, + "type": info.operation_type.name.upper(), + "replay": info.is_replayed, + "has_start_time": info.start_time is not None, + "has_status": info.status is not None, + }, + self._execution_arn, + ) + + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.operation_type.name != "STEP": + return + status = info.status.name if info.status is not None else "NONE" + record: dict[str, Any] = { + "plugin": "CONFPLUGIN", + "hook": "operation-end", + "op": info.operation_id, + "name": info.name, + "type": info.operation_type.name.upper(), + "replay": info.is_replayed, + "status": status, + "has_result": info.result is not None, + "has_error": info.error is not None, + "attempt": info.attempt, + "has_end_time": info.end_time is not None, + } + # Include the checkpointed serialized result exactly as exposed on the + # info; omit the key entirely when the info carries no result value. + if info.result is not None: + record["result"] = info.result + _emit(record, self._execution_arn) + + +@durable_step +def greet(_step_context: StepContext) -> str: + return "task-a" + + +@durable_execution(plugins=[OperationInfoShapePlugin()]) +def handler(_event: Any, context: DurableContext) -> str: + result: str = context.step(greet(), name="greet") + 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 index 9101b343..72481800 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml @@ -298,3 +298,63 @@ Resources: DurableConfig: RetentionPeriodInDays: 7 ExecutionTimeout: 300 + PluginInvocationInfoShape: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-19"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_invocation_info_shape.handler + Description: Invocation-start and invocation-end hook info field shape + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginOperationInfoShape: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-20"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_operation_info_shape.handler + Description: Operation-start and operation-end hook info field shape + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginAttemptInfoShape: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-21"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_attempt_info_shape.handler + Description: Attempt-start and attempt-end hook info field shape + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginOperationChangeShape: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-22"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_operation_change_shape.handler + Description: Operation-change hook info field shape (delta items and full map) + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 From 628d6c28b056a631ca63ae989f9db5a8943bfa48 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Thu, 6 Aug 2026 22:49:55 +0000 Subject: [PATCH 2/3] test: convert shape probes to canonical dumps --- .../plugin/plugin_attempt_info_shape.py | 83 ++++++++++-------- .../plugin/plugin_invocation_info_shape.py | 86 +++++++++---------- .../plugin/plugin_operation_change_shape.py | 72 +++++++++------- .../plugin/plugin_operation_info_shape.py | 86 ++++++++++--------- 4 files changed, 178 insertions(+), 149 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py index e6609820..ef2ec431 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py @@ -4,17 +4,19 @@ 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. Every logged field -is read from the CURRENT hook's own info parameter — never reconstructed from -another hook or from plugin state. When the Python info type does not expose a -field, the plugin logs the corresponding ``has_*`` flag as false; that omission -is the honest signal of a missing API surface. +``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, name, operation_type, attempt and start_time; the end info adds -``outcome`` (a ``UserFunctionOutcome`` enum) and ``error``. The failure is a -genuine thrown exception surfaced through the SDK's retry path — never a -hand-rolled outcome. +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 @@ -48,6 +50,33 @@ def _emit(record: dict[str, Any], execution_arn: str | None) -> None: 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 @@ -60,37 +89,19 @@ def on_invocation_start(self, info: InvocationStartInfo) -> None: def on_user_function_start(self, info: UserFunctionStartInfo) -> None: if info.operation_type.name != "STEP": return - _emit( - { - "plugin": "CONFPLUGIN", - "hook": "attempt-start", - "op": info.operation_id, - "name": info.name, - "type": info.operation_type.name.upper(), - "attempt": info.attempt, - "has_start_time": info.start_time is not None, - }, - self._execution_arn, - ) + _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 - _emit( - { - "plugin": "CONFPLUGIN", - "hook": "attempt-end", - "op": info.operation_id, - "name": info.name, - "type": info.operation_type.name.upper(), - "attempt": info.attempt, - # outcome as reported by the info's own outcome enum — a - # presentation of the API's data, not a reconstruction. - "outcome": info.outcome.name, - "has_error": info.error is not None, - }, - self._execution_arn, - ) + 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 diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py index 8011ed92..5df8ed21 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py @@ -2,18 +2,20 @@ A single 2-second wait suspends on the first invocation and completes on replay; the handler returns "done-". The plugin emits from the SDK's -real ``on_invocation_start`` / ``on_invocation_end`` hooks. Every logged field -is read from the CURRENT hook's own info parameter — never reconstructed from -another hook or from plugin state. When the Python ``InvocationInfo`` type does -not expose a field, the plugin logs the corresponding ``has_*`` flag as false -and omits the value key; that omission is the honest signal of a missing API -surface (the reference field set is the union across SDKs). +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``. It has -NO execution-input, operations-map, or externally-updated-operations field, and -``InvocationEndInfo`` adds only ``status`` + ``error`` (no execution-result -field). Those absences are surfaced faithfully as has_*: false. +``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 @@ -40,41 +42,39 @@ def _emit(record: dict[str, Any], execution_arn: str | None) -> None: class InvocationInfoShapePlugin(DurableInstrumentationPlugin): def on_invocation_start(self, info: InvocationStartInfo) -> None: - # Probe the invocation-start info surface. Python's InvocationInfo has - # no execution-input, operations-map, or updated-operations field, so - # those has_* flags are honestly false and the "input" key is omitted. - _emit( - { - "plugin": "CONFPLUGIN", - "hook": "invocation-start", - "first": info.is_first_invocation, - "has_request_id": info.request_id is not None, - "has_input": False, - "has_operations": False, - "updated_nonempty": False, - "has_start_time": info.execution_start_time is not None, - }, - info.execution_arn, - ) + # 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: - # "first" MUST come from the END info itself (its presence there is what - # is under test) — never captured at invocation-start. - status = info.status.name if info.status is not None else "NONE" - terminal = status in ("SUCCEEDED", "FAILED") - _emit( - { - "plugin": "CONFPLUGIN", - "hook": "invocation-end", - "first": info.is_first_invocation, - "terminal": terminal, - "status": status, - # InvocationEndInfo exposes no execution-result field. - "has_result": False, - "has_error": info.error is not None, - }, - info.execution_arn, - ) + # 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()]) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py index 056dca88..bef02db3 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py @@ -3,17 +3,19 @@ A single step named "greet" returns the constant "task-a" and succeeds on the first attempt. The plugin implements the SDK's real ``on_operation_change`` hook and, for each step-type operation in the change info's updated-operations -delta, probes the DELTA ITEM's own field surface. Every logged field is read -from the CURRENT hook's own info parameter (the change info and its delta -items) — never reconstructed from another hook or from plugin state. When the -Python type does not expose a field, the plugin logs the corresponding -``has_*`` flag as false; that omission is the honest signal of a missing API -surface. +delta, logs ONE single-line JSON record: a CANONICAL DUMP of that DELTA ITEM's +own field surface (a full ``OperationInfo``) plus the hook-level fields. Every +field the item 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). Hook-level fields: ``executionArn`` (the change +info's own ARN), ``updatedOperationsCount`` / ``operationsCount`` (the two map +sizes), and the derived scalar ``inFullMap`` := the item id also appears in the +info's full operations map. Python surface note: change-delta items are full ``OperationInfo`` objects (identity + status + payloads), so the item exposes result, end_time, attempt and an ``is_replayed`` replay indicator; ``OperationChangeInfo`` itself carries -the execution ARN. +the execution ARN and both operation maps. """ import json @@ -43,8 +45,8 @@ def _emit(record: dict[str, Any], execution_arn: str | None) -> None: class OperationChangeShapePlugin(DurableInstrumentationPlugin): def __init__(self) -> None: - # durableExecutionArn stamping is captured at invocation-start; has_arn - # below is probed from the change info's OWN execution_arn field. + # Operation-change hooks carry their own execution ARN, but the + # top-level durableExecutionArn stamp is captured at invocation-start. self._execution_arn: str | None = None def on_invocation_start(self, info: InvocationStartInfo) -> None: @@ -54,26 +56,38 @@ def on_operation_change(self, info: OperationChangeInfo) -> None: for op_id, op in info.updated_operations.items(): if op.operation_type.name != "STEP": continue - _emit( - { - "plugin": "CONFPLUGIN", - "hook": "operation-change", - "op": op_id, - "status": op.status.name, - "in_full_map": op_id in info.operations, - # has_arn probes whether the change info itself carries the - # execution ARN (its own field, read directly here). - "has_arn": info.execution_arn is not None, - "item_name": op.name, - "item_type": op.operation_type.name.upper(), - "item_has_result": op.result is not None, - "item_has_end_time": op.end_time is not None, - "item_has_attempt": op.attempt is not None, - # A replay indicator field exists on the item type itself. - "item_has_replay": hasattr(op, "is_replayed"), - }, - self._execution_arn, - ) + # Hook-level fields dumped from the change info's own surface. + record: dict[str, Any] = { + "plugin": "CONFPLUGIN", + "hook": "operation-change", + "updatedOperationsCount": len(info.updated_operations), + "operationsCount": len(info.operations), + "inFullMap": op_id in info.operations, + } + if info.execution_arn is not None: + record["executionArn"] = info.execution_arn + # Canonical dump of the delta ITEM's own OperationInfo surface. + record["id"] = op.operation_id + record["type"] = op.operation_type.name.upper() + record["status"] = op.status.name + record["isReplay"] = op.is_replayed + if op.name is not None: + record["name"] = op.name + if op.sub_type is not None: + record["subType"] = op.sub_type.name + if op.parent_id is not None: + record["parentId"] = op.parent_id + if op.start_time is not None: + record["startTimestamp"] = op.start_time.isoformat() + if op.end_time is not None: + record["endTimestamp"] = op.end_time.isoformat() + if op.result is not None: + record["result"] = op.result + if op.error is not None and op.error.message is not None: + record["error"] = op.error.message + if op.attempt is not None: + record["attempt"] = op.attempt + _emit(record, self._execution_arn) @durable_step diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py index 39a9bf1b..198239a8 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py @@ -2,17 +2,19 @@ A single step named "greet" returns the constant "task-a" and succeeds on the first attempt. The plugin emits from the SDK's real ``on_operation_start`` / -``on_operation_end`` hooks, filtering to step-type operations. Every logged -field is read from the CURRENT hook's own info parameter — never reconstructed -from another hook or from plugin state. When the Python ``OperationInfo`` type -does not expose a field, the plugin logs the corresponding ``has_*`` flag as -false; that omission is the honest signal of a missing API surface. +``on_operation_end`` hooks (filtering to step-type operations) a CANONICAL DUMP +of the CURRENT hook's own info parameter: every field the Python +``OperationInfo`` 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); type tokens are upper-cased, timestamps ISO-8601, +the serialized result the raw serialized string, errors their message string. Python surface note: ``OperationInfo`` exposes operation_id, operation_type, -name, start_time, is_replayed, status, end_time, result, error and attempt, so -the full operation-end field set is available. ``has_status`` is emitted at -operation-start for observability but not asserted (status population on a live -first start varies by SDK). +sub_type, name, parent_id, start_time, is_replayed, status, end_time, result, +error and attempt, so the full operation-end field set is available. +``status`` / ``startTimestamp`` / ``attempt`` are dumped at operation-start when +populated (STARTED) but NOT asserted — live-first-start population is +legitimately SDK-divergent. """ import json @@ -28,6 +30,7 @@ DurableInstrumentationPlugin, InvocationStartInfo, OperationEndInfo, + OperationInfo, OperationStartInfo, ) @@ -41,6 +44,37 @@ def _emit(record: dict[str, Any], execution_arn: str | None) -> None: print(json.dumps(record), flush=True) +def _dump_operation(hook: str, info: OperationInfo) -> dict[str, Any]: + # Canonical dump of an OperationInfo's own field surface. Identity + replay + # flag + status are always present; optional fields are emitted only when + # the info populates them (None -> omitted key = honest missing-field red). + record: dict[str, Any] = { + "plugin": "CONFPLUGIN", + "hook": hook, + "id": info.operation_id, + "type": info.operation_type.name.upper(), + "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.name + 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 + return record + + class OperationInfoShapePlugin(DurableInstrumentationPlugin): def __init__(self) -> None: # Operation hooks do not carry the execution ARN, so capture it from the @@ -53,42 +87,12 @@ def on_invocation_start(self, info: InvocationStartInfo) -> None: def on_operation_start(self, info: OperationStartInfo) -> None: if info.operation_type.name != "STEP": return - _emit( - { - "plugin": "CONFPLUGIN", - "hook": "operation-start", - "op": info.operation_id, - "name": info.name, - "type": info.operation_type.name.upper(), - "replay": info.is_replayed, - "has_start_time": info.start_time is not None, - "has_status": info.status is not None, - }, - self._execution_arn, - ) + _emit(_dump_operation("operation-start", info), self._execution_arn) def on_operation_end(self, info: OperationEndInfo) -> None: if info.operation_type.name != "STEP": return - status = info.status.name if info.status is not None else "NONE" - record: dict[str, Any] = { - "plugin": "CONFPLUGIN", - "hook": "operation-end", - "op": info.operation_id, - "name": info.name, - "type": info.operation_type.name.upper(), - "replay": info.is_replayed, - "status": status, - "has_result": info.result is not None, - "has_error": info.error is not None, - "attempt": info.attempt, - "has_end_time": info.end_time is not None, - } - # Include the checkpointed serialized result exactly as exposed on the - # info; omit the key entirely when the info carries no result value. - if info.result is not None: - record["result"] = info.result - _emit(record, self._execution_arn) + _emit(_dump_operation("operation-end", info), self._execution_arn) @durable_step From 8e22510102879ea37475a8afcc45c13407e4afa7 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Fri, 7 Aug 2026 21:04:30 +0000 Subject: [PATCH 3/3] test: add context-typed shape handler for 10-23 --- .../plugin/plugin_context_info_shape.py | 139 ++++++++++++++++++ .../template_plugin.yaml | 15 ++ 2 files changed, 154 insertions(+) create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_context_info_shape.py diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_context_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_context_info_shape.py new file mode 100644 index 00000000..bbefe05b --- /dev/null +++ b/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() 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 index 72481800..0231a292 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml @@ -358,3 +358,18 @@ Resources: DurableConfig: RetentionPeriodInDays: 7 ExecutionTimeout: 300 + PluginContextInfoShape: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-23"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_context_info_shape.handler + Description: Context-typed hook info field shape (subType tokens and children-replay indicator) + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 \ No newline at end of file