diff --git a/.github/scripts/tests/test_opentelemetry_conformance_workflow.py b/.github/scripts/tests/test_opentelemetry_conformance_workflow.py index 034be671e..37841a888 100644 --- a/.github/scripts/tests/test_opentelemetry_conformance_workflow.py +++ b/.github/scripts/tests/test_opentelemetry_conformance_workflow.py @@ -1,8 +1,12 @@ +import re from pathlib import Path import yaml +SHARED_WORKFLOW_REF = "main" +CONFORMANCE_TEST_REF = "main" + WORKFLOW_PATH = ( Path(__file__).parents[2] / "workflows" / "opentelemetry-conformance-tests.yml" ) @@ -18,7 +22,10 @@ def test_opentelemetry_conformance_caller_uses_current_workflow_contract() -> No "uses: aws/aws-durable-execution-conformance-tests/.github/workflows/" "opentelemetry-orchestrator.yml@" ) - assert orchestrator in workflow + shared_ref = re.search(re.escape(orchestrator) + r"(\S+)", workflow) + assert shared_ref is not None + assert shared_ref.group(1) == SHARED_WORKFLOW_REF + assert f"default: {CONFORMANCE_TEST_REF}" in workflow assert "python-opentelemetry.yml@" not in workflow assert "\n otlp_endpoint:" not in workflow @@ -28,7 +35,7 @@ def test_opentelemetry_conformance_caller_uses_current_workflow_contract() -> No "resource_prefix: p", "sdk_repository: aws/aws-durable-execution-sdk-python", "sdk_ref: ${{ github.event.pull_request.head.sha || github.sha }}", - "conformance_test_ref: ${{ inputs.conformance_test_ref || 'main' }}", + f"conformance_test_ref: ${{{{ inputs.conformance_test_ref || '{CONFORMANCE_TEST_REF}' }}}}", "checkout_sdk: true", f"examples_dir: {EXAMPLES_DIR}", "adot_release_repository: aws-observability/aws-otel-python-instrumentation", diff --git a/.github/workflows/opentelemetry-conformance-tests.yml b/.github/workflows/opentelemetry-conformance-tests.yml index 9d7835610..fca1b1f40 100644 --- a/.github/workflows/opentelemetry-conformance-tests.yml +++ b/.github/workflows/opentelemetry-conformance-tests.yml @@ -66,7 +66,7 @@ jobs: actions: write contents: read id-token: write - uses: aws/aws-durable-execution-conformance-tests/.github/workflows/opentelemetry-orchestrator.yml@d4b7df927ea7eb7022bcca339d2cddd387d3d650 + uses: aws/aws-durable-execution-conformance-tests/.github/workflows/opentelemetry-orchestrator.yml@main with: language: python runs_on: codebuild-github-actions-runner-${{ github.run_id }}-${{ github.run_attempt }} diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/README.md b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/README.md index 9c23dd9b6..5d184293a 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/README.md +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/README.md @@ -12,7 +12,7 @@ language-agnostic OTel conformance runner in which invokes each function, collects its spans from the configured backend, and asserts they match the shared requirement specification. That repository owns the runner, the requirement specifications, and the orchestration (backend matrix, -ADOT layer resolution, collector build, long-running cycle); this package owns +ADOT layer resolution, collector build); this package owns the Python handlers and SAM templates. This mirrors the non-OTel @@ -33,7 +33,7 @@ template-long-running.yaml # otel-long-running suite tests/ # contract tests for the templates and handlers ``` -The 20 invocation and 20 execution requirements reuse the same scenario +The invocation and execution requirements reuse the same scenario handlers; the view is selected per function through the `OTEL_PLUGIN_MODE` environment variable, which `common.otel_plugin_factory()` reads to pick `InvocationOtelPluginFactory` or `ExecutionOtelPluginFactory`. `template.yaml` @@ -63,6 +63,13 @@ deploys only the view named by its `OtelSuite` parameter. | `otel-invocation-18` | `otel_18_chained_invoke_failure.handler` | Verifies failed chained-invoke telemetry. | | `otel-invocation-19` | `otel_19_execution_failure.handler` | Verifies telemetry for a direct handler failure. | | `otel-invocation-20` | `otel_20_virtual_context.handler` | Verifies a virtual child-context span without context checkpoints. | +| `otel-invocation-21` | `otel_21_completed_step_replay.handler` | Local coverage verifies completed steps are skipped after wait/resume; cloud coverage is deferred. | +| `otel-invocation-22` | `otel_22_user_function_context.handler` | Creates ordinary user spans under the active handler, attempt, child, branch, and iteration contexts. | +| `otel-invocation-23` | `otel_23_callback_function_context.handler` | Verifies retry/check attempts, callback submitter, wrapped retry helper, and virtual-child callback parents. | +| `otel-invocation-24` | `otel_24_invocation_retry_status.handler` | Raises a retryable invocation error after a completed step, then resumes with its saved result. | +| `otel-invocation-25` | `otel_25_caught_callback_failure_replay.handler` | Catches a callback failure without diagnostic details, then replays it through two callback barriers. | +| `otel-invocation-26` | `otel_26_external_callback_completion_replay.handler` | Exports an external callback completion before the observed step and preserves its export count across two later barriers. | +| `otel-invocation-27` | `otel_27_chained_invoke_suspend.handler` | Verifies caller and child invocation spans when the child suspends for a durable wait and resumes. | | `otel-execution-1` | `otel_1_success.handler` | Verifies the execution-view workflow, step, and attempt hierarchy. | | `otel-execution-2` | `otel_2_wait_resume.handler` | Verifies the execution view across a resumed invocation. | | `otel-execution-3` | `otel_3_retry.handler` | Verifies the execution view across retry attempts. | @@ -83,13 +90,91 @@ deploys only the view named by its `OtelSuite` parameter. | `otel-execution-18` | `otel_18_chained_invoke_failure.handler` | Verifies source and target failed workflow roots. | | `otel-execution-19` | `otel_19_execution_failure.handler` | Verifies a failed invocation without a completed workflow. | | `otel-execution-20` | `otel_20_virtual_context.handler` | Verifies a virtual child-context span under the workflow root. | +| `otel-execution-21` | `otel_21_completed_step_replay.handler` | Local coverage verifies completed-operation spans are exported once across replay; cloud coverage is deferred. | +| `otel-execution-22` | `otel_22_user_function_context.handler` | Observes active execution-view callback contexts without supplying or repairing parents. | +| `otel-execution-23` | `otel_23_callback_function_context.handler` | Verifies the same SDK-owned callback lifecycle parents in the execution view. | +| `otel-execution-24` | `otel_24_invocation_retry_status.handler` | Verifies invocation retry status independently of step retry and recovery re-exports. | +| `otel-execution-25` | `otel_25_caught_callback_failure_replay.handler` | Verifies one failed callback export with ERROR and no exception details while the caught execution succeeds. | +| `otel-execution-26` | `otel_26_external_callback_completion_replay.handler` | Verifies first-completion export timing and no duplicate export across later callback resumes. | +| `otel-execution-27` | `otel_27_chained_invoke_suspend.handler` | Verifies caller/child workflow parentage and continuation links across child suspension. | | `otel-long-running-1` | `otel_long_running_1_wait.handler` | Verifies wait and resume telemetry across a long durable suspension. | | `otel-long-running-2` | `otel_long_running_2_retry.handler` | Verifies retry telemetry across a long durable backoff. | | `otel-long-running-3` | `otel_long_running_3_callback.handler` | Verifies callback telemetry when completion arrives after a long delay. | | `otel-long-running-4` | `otel_long_running_4_chained_invoke.handler` | Verifies chained-invoke telemetry while the target stays suspended. | -The runner discovers each mapping from `TestingMetadata.TestDescription` on the -functions in the templates. +The runner discovers each active cloud mapping from +`TestingMetadata.TestDescription` on the functions in the templates. + +## Deferred cloud coverage + +Case 21 is declared `NOT_IMPLEMENTED` in both views until the shared history +requirement accepts `WaitSucceeded` either before or after +`InvocationCompleted`. Both catalogs currently require invocation completion +first, but the wait timer runs independently of invocation cleanup. The +[execution-view cloud run](https://github.com/aws/aws-durable-execution-sdk-python/actions/runs/40588786563/job/122487525577) +observed the reverse order while the execution succeeded with `before-after`. +See the [review finding](https://github.com/aws/aws-durable-execution-sdk-python/pull/796#discussion_r4394770596). +The handler and local tests remain: they verify one execution of each step +body, `PENDING`/`SUCCEEDED` invocations, and one export per completed operation +in the execution view. Re-enable the SAM functions and build targets after the +shared requirement accepts both history orderings without relaxing those replay +assertions. + +## Callback regressions awaiting fixes + +Cases 25 and 26 are active in both views, including their SAM functions, build +targets, and local integration tests. These tests intentionally keep CI failing +until the required SDK and testing-library fixes merge. They are ordinary +failing tests, without skips, expected-failure markers, or failure suppression. + +- Case 25 needs terminal failure status without diagnostic details + ([#853](https://github.com/aws/aws-durable-execution-sdk-python/issues/853)) + and consistent empty callback error history + ([#862](https://github.com/aws/aws-durable-execution-sdk-python/issues/862)). +- Case 26 needs completion discovery and export + ([#854](https://github.com/aws/aws-durable-execution-sdk-python/issues/854)), + callback update production + ([#860](https://github.com/aws/aws-durable-execution-sdk-python/issues/860)), + and invocation-input acknowledgment + ([#861](https://github.com/aws/aws-durable-execution-sdk-python/issues/861)). +- The additional case 17 callback-error regression also needs snapshot isolation + ([#852](https://github.com/aws/aws-durable-execution-sdk-python/issues/852)). + The existing case 17 handler and supplied-error coverage remain active. + +Case 25 asserts the local-runner contract proposed in #862: detailed failure +history includes an empty `Payload` with `Truncated: false`, while metadata-only +history omits the payload and sets `Truncated: true`. The exact empty history +`Payload` shape remains **unverified** against deployed AWS Lambda and requires +service evidence before claiming service parity. The test inspects raw exported +spans and ordinary pytest diagnostics; it has no configurable proof-file output. + +Case 26 asserts inside the observed step that the terminal callback has already +reached the exporter, then checks that its raw export count remains one at both +later barriers and after completion. Recorded span timestamps alone are not +evidence of export order. + +The additional case 17 test remains available for follow-up in +[commit `6484e577`](https://github.com/aws/aws-durable-execution-sdk-python/commit/6484e577b478aeb6345e346515b40bd516cacb1f). +Before restoring it, correct its data-inclusion expectations: detailed errors +are untruncated; metadata-only history omits payloads and cannot reconstruct +diagnostic details. + +Case 24 remains active and covers invocation `RETRY` becoming `RETRYING`/`UNSET`. + +## Callback coverage boundary + +Cases 22 and 23 create normal user spans from the active context, without an +explicit parent or a copied execution ARN. They cover handler context, step and +condition attempts, child and branch bodies, callback submission, wrapped +`with_retry` body/strategy callbacks, and a virtual child. Case 23 places the +retry helper and virtual child after callback completion so ordinary successful +replay does not repeat their probes. + +This does not promise an operation/attempt parent for every arbitrary callback. +General retry/wait policies, serializers, summary generators, and item naming +have phase-specific caller scopes outside the wrapped user-function lifecycle; +these cases do not impose a new ownership policy on them. Instrumentation +extensions and user-created threads are outside this business-callback contract. ## How a handler maps to a requirement @@ -197,6 +282,16 @@ write access; the runner identity needs list, read, and cleanup access. `.github/workflows/opentelemetry-conformance-tests.yml` calls the shared orchestrator in the conformance repository and points it at this package with `examples_dir`, so orchestration stays centralized while the handlers run from -the commit under test. Pull requests run the invocation and execution suites plus -a short (60-second) long-running cycle; the full multi-hour long-running cycle is -driven by `workflow_dispatch` with `phase: launch` and `phase: check`. +the commit under test. The shared workflow tracks the conformance repository's +`main` branch, and the test catalog defaults to `main`. Manual runs can override +the catalog with the `conformance_test_ref` input. Pull requests run the +invocation and execution suites, with only case 21 reported as +`NOT_IMPLEMENTED`. Cases 25 and 26 remain enabled while their prerequisite fixes +are pending. The orchestrator no longer runs the legacy long-running +suite; its templates and local contract tests remain in this package. + +Case 27 (`chained-invoke-suspend`) invokes a child that performs a one-second +durable wait and returns the original payload after resuming. Both views +check two caller invocations, two child invocations and the child wait's +parentage and continuation links. It replaces the suspended-child scenario +formerly exercised by the short mode of `otel-long-running-4`. diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/Makefile b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/Makefile index 320577e7f..b89926ca4 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/Makefile +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/Makefile @@ -1,22 +1,27 @@ .PHONY: build-Otel1Success build-Otel2WaitResume build-Otel3Retry build-Otel4TerminalFailure \ build-Otel5ChildContext build-Otel6Parallel build-Otel7Map build-Otel8HandledFailure \ build-Otel9WaitForCondition build-Otel10WaitForCallback build-Otel11ChainedInvoke \ - build-Otel11InvokeTarget build-Otel12ChildContextFailure build-Otel13ParallelFailure \ + build-Otel27ChainedInvokeSuspend build-Otel27InvokeTarget build-Otel11InvokeTarget build-Otel12ChildContextFailure build-Otel13ParallelFailure \ build-Otel14MapFailure build-Otel15WaitInterrupted build-Otel16WaitForConditionFailure \ build-Otel17WaitForCallbackFailure build-Otel18ChainedInvokeFailure \ build-Otel18InvokeTarget build-Otel19ExecutionFailure build-Otel20VirtualContext \ + build-Otel22UserFunctionContext build-Otel23CallbackFunctionContext build-Otel24InvocationRetryStatus \ + build-Otel25CallbackFailureWithoutError build-Otel26ExternalCallbackReplay \ build-OtelExecution1Success \ build-OtelExecution2WaitResume build-OtelExecution3Retry \ build-OtelExecution4TerminalFailure build-OtelExecution5ChildContext \ build-OtelExecution6Parallel build-OtelExecution7Map build-OtelExecution8HandledFailure \ build-OtelExecution9WaitForCondition build-OtelExecution10WaitForCallback \ - build-OtelExecution11ChainedInvoke build-OtelExecution11InvokeTarget \ + build-OtelExecution11ChainedInvoke build-OtelExecution27ChainedInvokeSuspend build-OtelExecution27InvokeTarget build-OtelExecution11InvokeTarget \ build-OtelExecution12ChildContextFailure build-OtelExecution13ParallelFailure \ build-OtelExecution14MapFailure build-OtelExecution15WaitInterrupted \ build-OtelExecution16WaitForConditionFailure \ build-OtelExecution17WaitForCallbackFailure build-OtelExecution18ChainedInvokeFailure \ build-OtelExecution18InvokeTarget build-OtelExecution19ExecutionFailure \ build-OtelExecution20VirtualContext \ + build-OtelExecution22UserFunctionContext build-OtelExecution23CallbackFunctionContext \ + build-OtelExecution24InvocationRetryStatus \ + build-OtelExecution25CallbackFailureWithoutError build-OtelExecution26ExternalCallbackReplay \ build-OtelLongRunning1Wait \ build-OtelLongRunning2Retry build-OtelLongRunning3Callback \ build-OtelLongRunning4ChainedInvoke build-OtelLongRunning4InvokeTarget @@ -24,22 +29,27 @@ build-Otel1Success build-Otel2WaitResume build-Otel3Retry build-Otel4TerminalFailure \ build-Otel5ChildContext build-Otel6Parallel build-Otel7Map build-Otel8HandledFailure \ build-Otel9WaitForCondition build-Otel10WaitForCallback build-Otel11ChainedInvoke \ -build-Otel11InvokeTarget build-Otel12ChildContextFailure build-Otel13ParallelFailure \ +build-Otel27ChainedInvokeSuspend build-Otel27InvokeTarget build-Otel11InvokeTarget build-Otel12ChildContextFailure build-Otel13ParallelFailure \ build-Otel14MapFailure build-Otel15WaitInterrupted build-Otel16WaitForConditionFailure \ build-Otel17WaitForCallbackFailure build-Otel18ChainedInvokeFailure \ build-Otel18InvokeTarget build-Otel19ExecutionFailure build-Otel20VirtualContext \ +build-Otel22UserFunctionContext build-Otel23CallbackFunctionContext build-Otel24InvocationRetryStatus \ +build-Otel25CallbackFailureWithoutError build-Otel26ExternalCallbackReplay \ build-OtelExecution1Success \ build-OtelExecution2WaitResume build-OtelExecution3Retry \ build-OtelExecution4TerminalFailure build-OtelExecution5ChildContext \ build-OtelExecution6Parallel build-OtelExecution7Map build-OtelExecution8HandledFailure \ build-OtelExecution9WaitForCondition build-OtelExecution10WaitForCallback \ -build-OtelExecution11ChainedInvoke build-OtelExecution11InvokeTarget \ +build-OtelExecution11ChainedInvoke build-OtelExecution27ChainedInvokeSuspend build-OtelExecution27InvokeTarget build-OtelExecution11InvokeTarget \ build-OtelExecution12ChildContextFailure build-OtelExecution13ParallelFailure \ build-OtelExecution14MapFailure build-OtelExecution15WaitInterrupted \ build-OtelExecution16WaitForConditionFailure \ build-OtelExecution17WaitForCallbackFailure build-OtelExecution18ChainedInvokeFailure \ build-OtelExecution18InvokeTarget build-OtelExecution19ExecutionFailure \ build-OtelExecution20VirtualContext \ +build-OtelExecution22UserFunctionContext build-OtelExecution23CallbackFunctionContext \ +build-OtelExecution24InvocationRetryStatus \ +build-OtelExecution25CallbackFailureWithoutError build-OtelExecution26ExternalCallbackReplay \ build-OtelLongRunning1Wait \ build-OtelLongRunning2Retry build-OtelLongRunning3Callback \ build-OtelLongRunning4ChainedInvoke build-OtelLongRunning4InvokeTarget: diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_21_completed_step_replay.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_21_completed_step_replay.py new file mode 100644 index 000000000..2204a82d5 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_21_completed_step_replay.py @@ -0,0 +1,36 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Normal successful replay of a completed step for OTel case 21.""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python import ( + DurableContext, + StepContext, + durable_execution, + durable_step, +) +from aws_durable_execution_sdk_python.config import Duration +from common import otel_plugin_factory, require_scenario + + +@durable_step +def before_wait(_step_context: StepContext) -> str: + return "before" + + +@durable_step +def after_wait(_step_context: StepContext) -> str: + return "after" + + +@durable_execution(plugins=[otel_plugin_factory()]) +def handler(event: dict[str, Any], context: DurableContext) -> str: + require_scenario(event, "completed-step-replay") + before = context.step(before_wait(), name="otel-before-wait") + context.wait(Duration.from_seconds(1), name="otel-replay-wait") + after = context.step(after_wait(), name="otel-after-wait") + return f"{before}-{after}" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_22_user_function_context.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_22_user_function_context.py new file mode 100644 index 000000000..7e983dfc3 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_22_user_function_context.py @@ -0,0 +1,113 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Observe the real active SDK context inside public user callbacks.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + +from aws_durable_execution_sdk_python import ( + DurableContext, + StepContext, + durable_execution, +) +from aws_durable_execution_sdk_python.config import ( + Duration, + MapConfig, + ParallelBranch, + ParallelConfig, +) +from common import otel_plugin_factory, require_scenario +from opentelemetry import trace + + +def probe(label: str) -> None: + if not trace.get_current_span().get_span_context().is_valid: + raise RuntimeError(f"No active span for conformance.{label}") + span = trace.get_tracer("aws-durable-execution-conformance").start_span( + f"conformance.{label}", attributes={"conformance.callback": label} + ) + span.end() + + +def step_body(_context: StepContext) -> str: + probe("step") + return "step" + + +def child_step(_context: StepContext) -> str: + probe("child-step") + return "child-step" + + +def child_body(context: DurableContext) -> str: + probe("child") + context.step(child_step, name="otel-context-child-step") + probe("child-restored") + return "child" + + +def parallel_step_a(_context: StepContext) -> str: + probe("parallel-step-a") + return "a" + + +def parallel_step_b(_context: StepContext) -> str: + probe("parallel-step-b") + return "b" + + +def parallel_a(context: DurableContext) -> str: + probe("parallel-a") + return context.step(parallel_step_a, name="otel-context-branch-step-a") + + +def parallel_b(context: DurableContext) -> str: + probe("parallel-b") + return context.step(parallel_step_b, name="otel-context-branch-step-b") + + +def iteration_name(_item: int, index: int) -> str: + return ("otel-context-iteration-0", "otel-context-iteration-1")[index] + + +def mapper( + context: DurableContext, item: int, index: int, _items: Sequence[int] +) -> int: + probe(("map-0", "map-1")[index]) + + def map_step(_step_context: StepContext) -> int: + probe(("map-step-0", "map-step-1")[index]) + return item + + return context.step( + map_step, name=("otel-context-map-step-0", "otel-context-map-step-1")[index] + ) + + +@durable_execution(plugins=[otel_plugin_factory()]) +def handler(event: dict[str, Any], context: DurableContext) -> str: + require_scenario(event, "user-function-context") + probe("handler") + context.step(step_body, name="otel-context-step") + context.run_in_child_context(child_body, name="otel-context-child") + context.parallel( + [ + ParallelBranch(parallel_a, name="otel-context-branch-a"), + ParallelBranch(parallel_b, name="otel-context-branch-b"), + ], + name="otel-context-parallel", + config=ParallelConfig(max_concurrency=2), + ) + context.map( + [0, 1], + mapper, + name="otel-context-map", + config=MapConfig(max_concurrency=2, item_namer=iteration_name), + ) + probe("handler-restored") + context.wait(Duration.from_seconds(1), name="otel-context-resume") + probe("handler-after-resume") + return "context-complete" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_23_callback_function_context.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_23_callback_function_context.py new file mode 100644 index 000000000..d6b1bbf06 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_23_callback_function_context.py @@ -0,0 +1,136 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Observe callback contexts at SDK-owned user-function lifecycle boundaries.""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python import ( + DurableContext, + StepContext, + durable_execution, +) +from aws_durable_execution_sdk_python.config import ( + ChildConfig, + Duration, + JitterStrategy, + StepConfig, +) +from aws_durable_execution_sdk_python.exceptions import ChildContextError +from aws_durable_execution_sdk_python.retries import ( + RetryDecision, + RetryStrategyConfig, + WithRetryConfig, + create_retry_strategy, + with_retry, +) +from aws_durable_execution_sdk_python.types import ( + WaitForCallbackContext, + WaitForConditionCheckContext, +) +from aws_durable_execution_sdk_python.waits import ( + WaitForConditionConfig, + WaitForConditionDecision, +) +from common import otel_plugin_factory, require_scenario +from opentelemetry import trace + + +HELPER_FAILURE = "intentional-helper-failure" + + +def probe(label: str) -> None: + if not trace.get_current_span().get_span_context().is_valid: + raise RuntimeError(f"No active span for conformance.{label}") + span = trace.get_tracer("aws-durable-execution-conformance").start_span( + f"conformance.{label}", attributes={"conformance.callback": label} + ) + span.end() + + +def retry_step(context: StepContext) -> str: + probe(f"retry-attempt-{context.attempt}") + if context.attempt == 1: + raise RuntimeError("intentional-step-retry") + return "retried" + + +def check_condition(state: int, _context: WaitForConditionCheckContext) -> int: + next_state = state + 1 + probe(f"condition-check-{next_state}") + return next_state + + +def wait_strategy(state: int, _attempt: int) -> WaitForConditionDecision: + if state >= 2: + return WaitForConditionDecision.stop_polling() + return WaitForConditionDecision.continue_waiting(Duration.from_seconds(1)) + + +def submit_callback(_callback_id: str, _context: WaitForCallbackContext) -> None: + probe("callback-submitter") + + +def helper_body(_context: DurableContext, _attempt: int) -> str: + probe("with-retry-body") + raise RuntimeError(HELPER_FAILURE) + + +def helper_retry_strategy(_error: Exception, _attempt: int) -> RetryDecision: + probe("with-retry-strategy") + return RetryDecision.no_retry() + + +def virtual_child(_context: DurableContext) -> str: + probe("virtual-child") + return "virtual" + + +@durable_execution(plugins=[otel_plugin_factory()]) +def handler(event: dict[str, Any], context: DurableContext) -> str: + require_scenario(event, "callback-function-context") + context.step( + retry_step, + name="otel-context-retry-step", + config=StepConfig( + retry_strategy=create_retry_strategy( + RetryStrategyConfig( + max_attempts=2, + initial_delay=Duration.from_seconds(1), + max_delay=Duration.from_seconds(1), + backoff_rate=1.0, + jitter_strategy=JitterStrategy.NONE, + retryable_error_types=[RuntimeError], + ) + ) + ), + ) + context.wait_for_condition( + check_condition, + name="otel-context-condition", + config=WaitForConditionConfig(initial_state=0, wait_strategy=wait_strategy), + ) + context.wait_for_callback(submit_callback, name="otel-context-callback") + + # These callbacks run after the last asynchronous wait, so normal replay + # does not repeat the helper body or the checkpointless virtual child. + try: + with_retry( + context, + helper_body, + WithRetryConfig(retry_strategy=helper_retry_strategy), + name="otel-context-with-retry", + ) + except ChildContextError as error: + if error.message != HELPER_FAILURE: + raise + else: + raise RuntimeError("Expected the intentional helper failure") + context.run_in_child_context( + virtual_child, + name="otel-context-virtual", + config=ChildConfig(is_virtual=True), + ) + return "callback-context-complete" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_24_invocation_retry_status.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_24_invocation_retry_status.py new file mode 100644 index 000000000..382b7bb4a --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_24_invocation_retry_status.py @@ -0,0 +1,33 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""A real invocation retry for OTel status-mapping case 24.""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python import ( + DurableContext, + StepContext, + durable_execution, + durable_step, +) +from aws_durable_execution_sdk_python.exceptions import InvocationError +from common import otel_plugin_factory, require_scenario + + +@durable_step +def before_invocation_retry(_step_context: StepContext) -> str: + return "saved" + + +@durable_execution(plugins=[otel_plugin_factory()]) +def handler(event: dict[str, Any], context: DurableContext) -> str: + require_scenario(event, "invocation-retry-status") + # Capture this at entry: consuming the completed step can end replay mode. + entered_replay = context.is_replaying() + context.step(before_invocation_retry(), name="otel-before-invocation-retry") + if not entered_replay: + raise InvocationError("Conformance invocation retry after a completed step") + return "retry-complete" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_25_caught_callback_failure_replay.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_25_caught_callback_failure_replay.py new file mode 100644 index 000000000..c77b30bb0 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_25_caught_callback_failure_replay.py @@ -0,0 +1,60 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Catch a real callback failure, then replay it before a successful return.""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python import ( + DurableContext, + StepContext, + durable_execution, + durable_step, +) +from aws_durable_execution_sdk_python.config import WaitForCallbackConfig +from aws_durable_execution_sdk_python.exceptions import CallbackExternalError +from aws_durable_execution_sdk_python.serdes import JsonSerDes +from aws_durable_execution_sdk_python.types import LambdaContext, WaitForCallbackContext +from common import otel_plugin_factory, require_scenario + + +def submit_callback(_callback_id: str, _context: WaitForCallbackContext) -> None: + return None + + +@durable_step +def observe_failure(_context: StepContext) -> str: + return "caught" + + +def _run(event: dict[str, Any], context: DurableContext) -> str: + require_scenario(event, "caught-callback-failure-replay") + config = WaitForCallbackConfig(serdes=JsonSerDes()) + try: + context.wait_for_callback( + submit_callback, name="otel-failed-callback", config=config + ) + except CallbackExternalError: + # wait_for_callback restores the callback-typed cause on both first + # consumption and replay of its persisted failed child context. + pass + else: + raise AssertionError("Expected the callback to fail") + + observed = context.step(observe_failure(), name="otel-failed-callback-observed") + one = context.wait_for_callback( + submit_callback, name="otel-failed-callback-barrier-one", config=config + ) + two = context.wait_for_callback( + submit_callback, name="otel-failed-callback-barrier-two", config=config + ) + return "/".join((observed, one, two)) + + +_durable_handler = durable_execution(_run, plugins=[otel_plugin_factory()]) + + +def handler(event: Any, context: LambdaContext) -> Any: + return _durable_handler(event, context) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_26_external_callback_completion_replay.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_26_external_callback_completion_replay.py new file mode 100644 index 000000000..0a3905927 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_26_external_callback_completion_replay.py @@ -0,0 +1,53 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Public callback completion followed by two controlled replay barriers.""" + +from __future__ import annotations + +from typing import Any, cast + +from aws_durable_execution_sdk_python import ( + DurableContext, + StepContext, + durable_execution, + durable_step, +) +from aws_durable_execution_sdk_python.config import ( + CallbackConfig, + WaitForCallbackConfig, +) +from aws_durable_execution_sdk_python.serdes import JsonSerDes +from aws_durable_execution_sdk_python.types import WaitForCallbackContext +from common import otel_plugin_factory, require_scenario + + +def submit_callback(_callback_id: str, _context: WaitForCallbackContext) -> None: + return None + + +@durable_step +def observe_target(_context: StepContext, result: str) -> str: + return result + + +@durable_execution(plugins=[otel_plugin_factory()]) +def handler(event: dict[str, Any], context: DurableContext) -> str: + require_scenario(event, "external-callback-completion-replay") + config = WaitForCallbackConfig(serdes=JsonSerDes()) + target = cast( + str, + context.create_callback( + name="otel-external-target", config=CallbackConfig(serdes=JsonSerDes()) + ).result(), + ) + observed = context.step( + observe_target(target), name="otel-external-target-observed" + ) + one = context.wait_for_callback( + submit_callback, name="otel-external-barrier-one", config=config + ) + two = context.wait_for_callback( + submit_callback, name="otel-external-barrier-two", config=config + ) + return "/".join((observed, one, two)) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_27_chained_invoke_suspend.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_27_chained_invoke_suspend.py new file mode 100644 index 000000000..d3137cf29 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_27_chained_invoke_suspend.py @@ -0,0 +1,39 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Chained invoke with child suspension for both OTel view case 27 requirements.""" + +from __future__ import annotations + +import os +from typing import Any + +from aws_durable_execution_sdk_python import DurableContext, durable_execution +from aws_durable_execution_sdk_python.config import Duration +from common import otel_plugin_factory, require_scenario + + +@durable_execution(plugins=[otel_plugin_factory()]) +def handler( + event: dict[str, Any], + context: DurableContext, +) -> dict[str, Any]: + require_scenario(event, "chained-invoke-suspend") + return context.invoke( + function_name=os.environ["OTEL_INVOKE_TARGET_FUNCTION_NAME"], + payload=event, + name="otel-suspended-invoke", + ) + + +@durable_execution(plugins=[otel_plugin_factory()]) +def target_handler( + event: dict[str, Any], + context: DurableContext, +) -> dict[str, Any]: + require_scenario(event, "chained-invoke-suspend") + context.wait( + Duration.from_seconds(1), + name="otel-suspended-target-wait", + ) + return event diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/template.yaml b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/template.yaml index a99f19a73..68d2203d3 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/template.yaml +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/template.yaml @@ -109,35 +109,9 @@ Resources: NotImplemented: - id: otel-invocation-21 reason: >- - Conformance handler coverage is pending in - https://github.com/aws/aws-durable-execution-sdk-python/pull/796. - - id: otel-invocation-22 - reason: >- - Conformance handler coverage is pending in - https://github.com/aws/aws-durable-execution-sdk-python/pull/796. - - id: otel-invocation-23 - reason: >- - Conformance handler coverage is pending in - https://github.com/aws/aws-durable-execution-sdk-python/pull/796. - - id: otel-invocation-24 - reason: >- - Conformance handler coverage is pending in - https://github.com/aws/aws-durable-execution-sdk-python/pull/796. - - id: otel-invocation-25 - reason: >- - Deferred until terminal failure status and empty callback error history - are implemented: https://github.com/aws/aws-durable-execution-sdk-python/issues/853 - and https://github.com/aws/aws-durable-execution-sdk-python/issues/862. - - id: otel-invocation-26 - reason: >- - Deferred until callback completion delivery and replay acknowledgment - are implemented: https://github.com/aws/aws-durable-execution-sdk-python/issues/854, - https://github.com/aws/aws-durable-execution-sdk-python/issues/860, - and https://github.com/aws/aws-durable-execution-sdk-python/issues/861. - - id: otel-invocation-27 - reason: >- - Conformance handler coverage is pending in - https://github.com/aws/aws-durable-execution-sdk-python/pull/796. + Deferred until the shared history requirement accepts WaitSucceeded + before or after InvocationCompleted while preserving replay assertions: + https://github.com/aws/aws-durable-execution-sdk-python/pull/796#discussion_r4394770596. Properties: CodeUri: src/ Handler: otel_1_success.handler @@ -441,6 +415,104 @@ Resources: FunctionName: !Sub "${AWS::StackName}-otel-invocation-20" Role: !Ref LambdaExecutionRoleArn + Otel22UserFunctionContext: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-invocation-22 + Properties: + CodeUri: src/ + Handler: otel_22_user_function_context.handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-22" + Role: !Ref LambdaExecutionRoleArn + + Otel23CallbackFunctionContext: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-invocation-23 + Properties: + CodeUri: src/ + Handler: otel_23_callback_function_context.handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-23" + Role: !Ref LambdaExecutionRoleArn + + Otel24InvocationRetryStatus: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-invocation-24 + Properties: + CodeUri: src/ + Handler: otel_24_invocation_retry_status.handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-24" + Role: !Ref LambdaExecutionRoleArn + + Otel25CallbackFailureWithoutError: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-invocation-25 + Properties: + CodeUri: src/ + Handler: otel_25_caught_callback_failure_replay.handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-25" + Role: !Ref LambdaExecutionRoleArn + + Otel26ExternalCallbackReplay: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-invocation-26 + Properties: + CodeUri: src/ + Handler: otel_26_external_callback_completion_replay.handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-26" + Role: !Ref LambdaExecutionRoleArn + + Otel27ChainedInvokeSuspend: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-invocation-27 + Properties: + CodeUri: src/ + Handler: otel_27_chained_invoke_suspend.handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-27" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_INVOKE_TARGET_FUNCTION_NAME: !Sub "${Otel27InvokeTarget.Arn}:$LATEST" + + Otel27InvokeTarget: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + Properties: + CodeUri: src/ + Handler: otel_27_chained_invoke_suspend.target_handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-27-target" + Role: !Ref LambdaExecutionRoleArn + OtelExecution1Success: Type: AWS::Serverless::Function Condition: DeployExecutionView @@ -452,35 +524,9 @@ Resources: NotImplemented: - id: otel-execution-21 reason: >- - Conformance handler coverage is pending in - https://github.com/aws/aws-durable-execution-sdk-python/pull/796. - - id: otel-execution-22 - reason: >- - Conformance handler coverage is pending in - https://github.com/aws/aws-durable-execution-sdk-python/pull/796. - - id: otel-execution-23 - reason: >- - Conformance handler coverage is pending in - https://github.com/aws/aws-durable-execution-sdk-python/pull/796. - - id: otel-execution-24 - reason: >- - Conformance handler coverage is pending in - https://github.com/aws/aws-durable-execution-sdk-python/pull/796. - - id: otel-execution-25 - reason: >- - Deferred until terminal failure status and empty callback error history - are implemented: https://github.com/aws/aws-durable-execution-sdk-python/issues/853 - and https://github.com/aws/aws-durable-execution-sdk-python/issues/862. - - id: otel-execution-26 - reason: >- - Deferred until callback completion delivery and replay acknowledgment - are implemented: https://github.com/aws/aws-durable-execution-sdk-python/issues/854, - https://github.com/aws/aws-durable-execution-sdk-python/issues/860, - and https://github.com/aws/aws-durable-execution-sdk-python/issues/861. - - id: otel-execution-27 - reason: >- - Conformance handler coverage is pending in - https://github.com/aws/aws-durable-execution-sdk-python/pull/796. + Deferred until the shared history requirement accepts WaitSucceeded + before or after InvocationCompleted while preserving replay assertions: + https://github.com/aws/aws-durable-execution-sdk-python/pull/796#discussion_r4394770596. Properties: CodeUri: src/ Handler: otel_1_success.handler @@ -845,3 +891,120 @@ Resources: Environment: Variables: OTEL_PLUGIN_MODE: execution + + OtelExecution22UserFunctionContext: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-execution-22 + Properties: + CodeUri: src/ + Handler: otel_22_user_function_context.handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-22" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution + + OtelExecution23CallbackFunctionContext: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-execution-23 + Properties: + CodeUri: src/ + Handler: otel_23_callback_function_context.handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-23" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution + + OtelExecution24InvocationRetryStatus: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-execution-24 + Properties: + CodeUri: src/ + Handler: otel_24_invocation_retry_status.handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-24" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution + + OtelExecution25CallbackFailureWithoutError: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-execution-25 + Properties: + CodeUri: src/ + Handler: otel_25_caught_callback_failure_replay.handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-25" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution + + OtelExecution26ExternalCallbackReplay: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-execution-26 + Properties: + CodeUri: src/ + Handler: otel_26_external_callback_completion_replay.handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-26" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution + + OtelExecution27ChainedInvokeSuspend: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-execution-27 + Properties: + CodeUri: src/ + Handler: otel_27_chained_invoke_suspend.handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-27" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution + OTEL_INVOKE_TARGET_FUNCTION_NAME: !Sub "${OtelExecution27InvokeTarget.Arn}:$LATEST" + + OtelExecution27InvokeTarget: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + Properties: + CodeUri: src/ + Handler: otel_27_chained_invoke_suspend.target_handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-27-target" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_callback_function_context.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_callback_function_context.py new file mode 100644 index 000000000..99ce7a78b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_callback_function_context.py @@ -0,0 +1,153 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Verify actual SDK callback parents across retries and external completion.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +import time +from collections import Counter +from pathlib import Path + +import pytest +from aws_durable_execution_sdk_python import DurableContext +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry import context, trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from aws_durable_execution_sdk_python_otel import ( + ExecutionOtelPluginFactory, + InvocationOtelPluginFactory, + OtelPluginConfig, +) + + +SRC_DIR = Path(__file__).resolve().parents[1] / "src" +EXPECTED_PARENTS = { + "retry-attempt-1": "otel-context-retry-step attempt 1", + "retry-attempt-2": "otel-context-retry-step attempt 2", + "condition-check-1": "otel-context-condition attempt 1", + "condition-check-2": "otel-context-condition attempt 2", + "callback-submitter": "otel-context-callback submitter attempt 1", + "with-retry-body": "otel-context-with-retry", + "with-retry-strategy": "otel-context-with-retry", + "virtual-child": "otel-context-virtual", +} + + +@pytest.mark.parametrize( + "plugin_type", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +@pytest.mark.parametrize( + ("helper_outcome", "expected_error"), + [ + ("intentional-failure", None), + ("unexpected-failure", "unexpected-helper-failure"), + ("unexpected-success", "Expected the intentional helper failure"), + ], +) +def test_callback_probes_keep_their_actual_sdk_parent( + monkeypatch: pytest.MonkeyPatch, + plugin_type: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], + helper_outcome: str, + expected_error: str | None, +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) + plugin = plugin_type( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + before_context = context.get_current() + common_spec = importlib.util.spec_from_file_location( + "common", SRC_DIR / "common.py" + ) + assert common_spec is not None and common_spec.loader is not None + common = importlib.util.module_from_spec(common_spec) + monkeypatch.setitem(sys.modules, "common", common) + common_spec.loader.exec_module(common) + monkeypatch.setattr(common, "otel_plugin_factory", lambda: plugin) + spec = importlib.util.spec_from_file_location( + "otel_23_callback_function_context", + SRC_DIR / "otel_23_callback_function_context.py", + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + if helper_outcome != "intentional-failure": + + def changed_helper_body(_context: DurableContext, _attempt: int) -> str: + module.probe("with-retry-body") + if helper_outcome == "unexpected-failure": + raise RuntimeError("unexpected-helper-failure") + return "unexpected-success" + + monkeypatch.setattr(module, "helper_body", changed_helper_body) + try: + with DurableFunctionTestRunner(handler=module.handler) as runner: + arn = runner.run_async( + input=json.dumps({"scenario": "callback-function-context"}), timeout=30 + ) + callback_id = runner.wait_for_callback( + arn, name="otel-context-callback create callback id", timeout=10 + ) + # External delivery delay, outside the durable handler. + time.sleep(1) + runner.send_callback_success(callback_id, result=b"callback-complete") + result = runner.wait_for_result(arn, timeout=15) + if expected_error is not None: + assert result.status.value == "FAILED" + assert result.error is not None + assert result.error.message == expected_error + assert "conformance.virtual-child" not in { + span.name for span in exporter.get_finished_spans() + } + assert context.get_current() == before_context + return + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == "callback-context-complete" + spans = exporter.get_finished_spans() + probes = [span for span in spans if span.name.startswith("conformance.")] + assert Counter(span.name for span in probes) == Counter( + f"conformance.{label}" for label in EXPECTED_PARENTS + ) + spans_by_id = { + (span.context.trace_id, span.context.span_id): span + for span in spans + if span.context is not None + } + sdk_traces = { + span.context.trace_id + for span in spans + if span.context is not None + and span.attributes is not None + and span.attributes.get("durable.execution.arn") == arn + } + for probe_span in probes: + assert probe_span.context is not None + assert probe_span.parent is not None + assert probe_span.attributes is not None + label = str(probe_span.attributes["conformance.callback"]) + assert "durable.execution.arn" not in probe_span.attributes + assert probe_span.context.trace_id in sdk_traces + parent = spans_by_id[ + (probe_span.context.trace_id, probe_span.parent.span_id) + ] + assert parent.name == EXPECTED_PARENTS[label] + assert parent.attributes is not None + assert parent.attributes["durable.execution.arn"] == arn + assert context.get_current() == before_context + finally: + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_caught_callback_failure_replay.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_caught_callback_failure_replay.py new file mode 100644 index 000000000..c9ee0c789 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_caught_callback_failure_replay.py @@ -0,0 +1,328 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""The real case 25 handler catches only its intended failure across replay.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +import time +from collections.abc import Callable, Sequence +from pathlib import Path +from types import ModuleType +from typing import Any + +import pytest +from aws_durable_execution_sdk_python import DurableContext, StepContext +from aws_durable_execution_sdk_python.exceptions import ( + CallbackExternalError, + ChildContextError, +) +from aws_durable_execution_sdk_python.lambda_service import ErrorObject +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationEndInfo, + InvocationStartInfo, + UserFunctionStartInfo, +) +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) +from opentelemetry.sdk.trace import ReadableSpan, TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExportResult +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from aws_durable_execution_sdk_python_otel import ( + ExecutionOtelPluginFactory, + InvocationOtelPluginFactory, + OtelPluginConfig, +) + + +SRC = Path(__file__).resolve().parents[1] / "src" +FAILED = "otel-failed-callback" +OBSERVED = FAILED + "-observed" +BARRIERS = [FAILED + "-barrier-one", FAILED + "-barrier-two"] +EXPECTED_HISTORY = [ + "ExecutionStarted", + "ContextStarted", + "CallbackStarted", + "StepStarted", + "StepSucceeded", + "InvocationCompleted", + "CallbackFailed", + "ContextFailed", + "StepStarted", + "StepSucceeded", + "ContextStarted", + "CallbackStarted", + "StepStarted", + "StepSucceeded", + "InvocationCompleted", + "CallbackSucceeded", + "ContextSucceeded", + "ContextStarted", + "CallbackStarted", + "StepStarted", + "StepSucceeded", + "InvocationCompleted", + "CallbackSucceeded", + "ContextSucceeded", + "InvocationCompleted", + "ExecutionSucceeded", +] + + +def _callback_after_suspend( + runner: DurableFunctionTestRunner, arn: str, name: str +) -> tuple[str, tuple[int, int]]: + deadline = time.monotonic() + 8 + while time.monotonic() < deadline: + events = runner.get_execution_history(arn, include_execution_data=True).events + starts = [ + event + for event in events + if event.event_type == "CallbackStarted" + and event.name == name + " create callback id" + ] + ends = [ + event + for event in events + if starts + and event.event_type == "InvocationCompleted" + and event.event_id > starts[0].event_id + ] + if starts and ends: + details = starts[0].callback_started_details + assert details is not None and details.callback_id is not None + return details.callback_id, (starts[0].event_id, ends[0].event_id) + time.sleep(0.01) + raise AssertionError(f"No creating invocation completion for {name}") + + +def _load( + monkeypatch: pytest.MonkeyPatch, + plugin: ExecutionOtelPluginFactory | InvocationOtelPluginFactory, +) -> ModuleType: + spec = importlib.util.spec_from_file_location("common", SRC / "common.py") + assert spec is not None and spec.loader is not None + common = importlib.util.module_from_spec(spec) + monkeypatch.setitem(sys.modules, "common", common) + spec.loader.exec_module(common) + monkeypatch.setattr(common, "otel_plugin_factory", lambda: plugin) + spec = importlib.util.spec_from_file_location( + "otel_25_caught_callback_failure_replay", + SRC / "otel_25_caught_callback_failure_replay.py", + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.mark.parametrize( + ("view", "mode"), + [ + (ExecutionOtelPluginFactory, "normal"), + (InvocationOtelPluginFactory, "normal"), + (ExecutionOtelPluginFactory, "unexpected-success"), + (ExecutionOtelPluginFactory, "barrier-failure"), + ], +) +def test_case25_public_caught_failure_and_replay( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + view: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], + mode: str, +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + starts: list[InvocationStartInfo] = [] + invocation_ends: list[InvocationEndInfo] = [] + exported: list[tuple[ReadableSpan, int]] = [] + user_starts: list[tuple[int, UserFunctionStartInfo]] = [] + failures: list[tuple[int, CallbackExternalError | ChildContextError]] = [] + marker_calls: list[int] = [] + + class Exporter(InMemorySpanExporter): + def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: + exported.extend((span, len(starts)) for span in spans) + return super().export(spans) + + exporter = Exporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = view( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + original_create = plugin.create_plugin + + def create_plugin(info: InvocationStartInfo) -> DurableInstrumentationPlugin: + instance = original_create(info) + original_start = instance.on_invocation_start + original_end = instance.on_invocation_end + original_user_start = instance.on_user_function_start + + def on_start(info: InvocationStartInfo) -> None: + starts.append(info) + original_start(info) + + def on_end(info: InvocationEndInfo) -> None: + invocation_ends.append(info) + original_end(info) + + def on_user_start(info: UserFunctionStartInfo) -> None: + user_starts.append((len(starts), info)) + original_user_start(info) + + monkeypatch.setattr(instance, "on_invocation_start", on_start) + monkeypatch.setattr(instance, "on_invocation_end", on_end) + monkeypatch.setattr(instance, "on_user_function_start", on_user_start) + return instance + + monkeypatch.setattr(plugin, "create_plugin", create_plugin) + original_wait = DurableContext.wait_for_callback + + def wait(context: DurableContext, *args: Any, **kwargs: Any) -> Any: + try: + return original_wait(context, *args, **kwargs) + except (CallbackExternalError, ChildContextError) as error: + if kwargs.get("name") == FAILED: + failures.append((len(starts), error)) + raise + + monkeypatch.setattr(DurableContext, "wait_for_callback", wait) + module = _load(monkeypatch, plugin) + original_marker = module.observe_failure + + def counted_marker() -> Callable[[StepContext], str]: + step = original_marker() + + def run(context: StepContext) -> str: + marker_calls.append(len(starts)) + return step(context) + + return run + + monkeypatch.setattr(module, "observe_failure", counted_marker) + gates: list[tuple[int, int]] = [] + try: + with DurableFunctionTestRunner( + handler=module.handler, + store=FileSystemExecutionStore.create(tmp_path / "store"), + skip_time=False, + poll_interval=0.01, + execution_timeout=30, + ) as runner: + arn = runner.run_async( + input=json.dumps({"scenario": "caught-callback-failure-replay"}) + ) + callback, gate = _callback_after_suspend(runner, arn, FAILED) + gates.append(gate) + if mode == "unexpected-success": + runner.send_callback_success(callback, result=b'"unexpected"') + else: + runner.send_callback_failure(callback) + for name, payload in zip(BARRIERS, ("one", "two")): + callback, gate = _callback_after_suspend(runner, arn, name) + gates.append(gate) + if mode == "barrier-failure": + runner.send_callback_failure( + callback, error=ErrorObject.from_message("barrier failure") + ) + break + runner.send_callback_success( + callback, result=json.dumps(payload).encode() + ) + result = runner.wait_for_result(arn, timeout=10) + history = runner.get_execution_history( + arn, include_execution_data=True + ).to_dict() + metadata_history = runner.get_execution_history( + arn, include_execution_data=False + ).to_dict() + if mode != "normal": + assert result.status.value == "FAILED" + assert marker_calls == ([] if mode == "unexpected-success" else [2]) + assert not any( + event.get("Name") == BARRIERS[1] for event in history["Events"] + ) + return + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == "caught/one/two" + assert [info.status.value for info in invocation_ends] == [ + "PENDING", + "PENDING", + "PENDING", + "SUCCEEDED", + ] + assert [event["EventType"] for event in history["Events"]] == EXPECTED_HISTORY + assert [event["EventId"] for event in history["Events"]] == list(range(1, 27)) + assert gates == [(3, 6), (12, 15), (19, 22)] + assert marker_calls == [2] + assert [invocation for invocation, _ in failures] == [2, 3, 4] + assert {type(error) for _, error in failures} == {CallbackExternalError} + assert len({error.error_type for _, error in failures}) == 1 + failed = history["Events"][6] + # Intended local history contract from #862. The exact empty Payload + # service shape remains unverified against deployed AWS Lambda. + assert failed["CallbackFailedDetails"]["Error"] == { + "Payload": {}, + "Truncated": False, + } + assert metadata_history["Events"][6]["CallbackFailedDetails"]["Error"] == { + "Truncated": True, + } + target_id = failed["Id"] + leaves = [ + (span, invocation) + for span, invocation in exported + if (span.attributes or {}).get("durable.operation.id") == target_id + and (span.attributes or {}).get("durable.operation.status") == "FAILED" + ] + assert len(leaves) == 1 + leaf, invocation = leaves[0] + assert invocation == 2 + assert leaf.status.status_code.name == "ERROR" + assert not leaf.events + marker_spans = [ + (span, invocation) + for span, invocation in exported + if (span.attributes or {}).get("durable.operation.name") == OBSERVED + ] + assert marker_spans and all(invocation == 2 for _, invocation in marker_spans) + assert leaf.end_time is not None + for span, _ in marker_spans: + assert span.start_time is not None + assert leaf.end_time <= span.start_time + workflows = [span for span, _ in exported if span.name == "Workflow"] + assert len(workflows) == 1 and workflows[0].status.status_code.name == "OK" + assert leaf.parent is not None + parent = next( + span + for span, _ in exported + if span.context is not None and span.context.span_id == leaf.parent.span_id + ) + assert (parent.attributes or {}).get("durable.operation.id") == ( + history["Events"][1]["Id"] + ) + if view is InvocationOtelPluginFactory: + assert parent.parent is not None + invocation_span, invocation = next( + (span, number) + for span, number in exported + if span.context is not None + and span.context.span_id == parent.parent.span_id + ) + assert invocation_span.name == "Invocation" and invocation == 2 + submitter_id = history["Events"][3]["Id"] + assert sum(info.operation_id == submitter_id for _, info in user_starts) == 1 + finally: + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_chained_invoke_suspend.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_chained_invoke_suspend.py new file mode 100644 index 000000000..9e39b54b8 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_chained_invoke_suspend.py @@ -0,0 +1,135 @@ +"""The real case 27 child suspends and resumes before its caller completes.""" + +import importlib.util +import json +from pathlib import Path +import sys + +import pytest +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from aws_durable_execution_sdk_python_otel import ( + ExecutionOtelPluginFactory, + InvocationOtelPluginFactory, + OtelPluginConfig, +) + + +SRC = Path(__file__).resolve().parents[1] / "src" + + +@pytest.mark.parametrize( + "view", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +def test_case27_resumes_the_real_child_and_preserves_its_result(view, monkeypatch): + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + monkeypatch.delenv("_X_AMZN_TRACE_ID", raising=False) + monkeypatch.setenv("OTEL_INVOKE_TARGET_FUNCTION_NAME", "case27-child") + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + spec = importlib.util.spec_from_file_location("common", SRC / "common.py") + common = importlib.util.module_from_spec(spec) + monkeypatch.setitem(sys.modules, "common", common) + spec.loader.exec_module(common) + monkeypatch.setattr( + common, + "otel_plugin_factory", + lambda: view( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ), + ) + spec = importlib.util.spec_from_file_location( + "otel_27_chained_invoke_suspend", SRC / "otel_27_chained_invoke_suspend.py" + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + payload = {"scenario": "chained-invoke-suspend", "nested": {"value": 42}} + try: + with DurableFunctionTestRunner( + handler=module.handler, + skip_time=False, + poll_interval=0.01, + execution_timeout=30, + ) as runner: + runner.register_durable_function("case27-child", module.target_handler) + arn = runner.run_async(input=json.dumps(payload)) + result = runner.wait_for_result(arn, timeout=15) + history = runner.get_execution_history(arn, include_execution_data=True) + starts = [ + e for e in history.events if e.event_type == "ChainedInvokeStarted" + ] + ends = [ + e for e in history.events if e.event_type == "ChainedInvokeSucceeded" + ] + assert len(starts) == len(ends) == 1 + assert starts[0].operation_id == ends[0].operation_id + child_arn = starts[0].chained_invoke_started_details.durable_execution_arn + child_history = runner.get_execution_history( + child_arn, include_execution_data=True + ) + + assert result.status.value == "SUCCEEDED" + assert json.loads(result.result) == payload + for execution_history in (history, child_history): + assert ( + sum( + e.event_type == "InvocationCompleted" + for e in execution_history.events + ) + == 2 + ) + assert ( + sum( + e.event_type == "ExecutionStarted" for e in execution_history.events + ) + == 1 + ) + assert sum(e.event_type == "WaitStarted" for e in child_history.events) == 1 + assert sum(e.event_type == "WaitSucceeded" for e in child_history.events) == 1 + spans = exporter.get_finished_spans() + for execution_arn in (arn, child_arn): + invocations = sorted( + [ + s + for s in spans + if s.name == "Invocation" + and s.attributes.get("durable.execution.arn") == execution_arn + ], + key=lambda s: s.start_time, + ) + assert [s.attributes["durable.invocation.status"] for s in invocations] == [ + "PENDING", + "SUCCEEDED", + ] + assert [s.attributes["durable.invocation.first"] for s in invocations] == [ + True, + False, + ] + waits = sorted( + [s for s in spans if s.name == "otel-suspended-target-wait"], + key=lambda s: s.start_time, + ) + assert len(waits) == (2 if view is InvocationOtelPluginFactory else 1) + assert len({s.attributes["durable.operation.id"] for s in waits}) == 1 + assert all(s.attributes["durable.execution.arn"] == child_arn for s in waits) + if view is InvocationOtelPluginFactory: + assert waits[-1].links[0].context.span_id == waits[0].context.span_id + else: + resumed = next( + s + for s in spans + if s.name == "Invocation" + and s.attributes.get("durable.execution.arn") == child_arn + and not s.attributes["durable.invocation.first"] + ) + assert waits[0].links[0].context.span_id == resumed.context.span_id + finally: + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_completed_step_replay.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_completed_step_replay.py new file mode 100644 index 000000000..7cf5c7353 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_completed_step_replay.py @@ -0,0 +1,102 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Exercise case 21 through the public decorator and local durable runner.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any + +import pytest +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry import context +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from aws_durable_execution_sdk_python_otel import ( + ExecutionOtelPluginFactory, + InvocationOtelPluginFactory, + OtelPluginConfig, +) + + +SRC_DIR = Path(__file__).resolve().parents[1] / "src" + + +@pytest.mark.parametrize( + "plugin_type", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +def test_completed_step_replay_uses_saved_result( + monkeypatch: pytest.MonkeyPatch, + plugin_type: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = plugin_type( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + before_context = context.get_current() + common_spec = importlib.util.spec_from_file_location( + "common", SRC_DIR / "common.py" + ) + assert common_spec is not None and common_spec.loader is not None + common = importlib.util.module_from_spec(common_spec) + monkeypatch.setitem(sys.modules, "common", common) + common_spec.loader.exec_module(common) + monkeypatch.setattr(common, "otel_plugin_factory", lambda: plugin) + spec = importlib.util.spec_from_file_location( + "otel_21_completed_step_replay", SRC_DIR / "otel_21_completed_step_replay.py" + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + calls: list[str] = [] + + def track(factory: Any, label: str) -> Any: + def make_step() -> Any: + step = factory() + + def counted(step_context: Any) -> str: + calls.append(label) + return step(step_context) + + return counted + + return make_step + + monkeypatch.setattr(module, "before_wait", track(module.before_wait, "before")) + monkeypatch.setattr(module, "after_wait", track(module.after_wait, "after")) + try: + with DurableFunctionTestRunner(handler=module.handler) as runner: + result = runner.run( + input=json.dumps({"scenario": "completed-step-replay"}), timeout=15 + ) + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == "before-after" + assert calls == ["before", "after"] + spans = exporter.get_finished_spans() + invocations = [span for span in spans if span.name == "Invocation"] + assert [ + span.attributes["durable.invocation.status"] + for span in invocations + if span.attributes is not None + ] == ["PENDING", "SUCCEEDED"] + if plugin_type is ExecutionOtelPluginFactory: + # Count raw exports, including duplicates that reuse a span ID. + for name in ("otel-before-wait", "otel-replay-wait", "otel-after-wait"): + assert len([span for span in spans if span.name == name]) == 1 + assert context.get_current() == before_context + finally: + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_external_callback_completion_replay.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_external_callback_completion_replay.py new file mode 100644 index 000000000..cda354cd9 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_external_callback_completion_replay.py @@ -0,0 +1,215 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""A real callback completion is exported before two later public replays.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +import time +from pathlib import Path +from typing import Any + +import pytest +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) +from opentelemetry import context +from opentelemetry.sdk.trace import ReadableSpan, TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from aws_durable_execution_sdk_python_otel import ( + ExecutionOtelPluginFactory, + InvocationOtelPluginFactory, + OtelPluginConfig, +) + + +SRC_DIR = Path(__file__).resolve().parents[1] / "src" + + +def _suspended_callback( + runner: DurableFunctionTestRunner, arn: str, name: str +) -> tuple[str, tuple[int, int]]: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + events = runner.get_execution_history(arn, include_execution_data=True).events + starts = [ + event + for event in events + if event.event_type == "CallbackStarted" and event.name == name + ] + completions = [ + event + for event in events + if event.event_type == "InvocationCompleted" + and starts + and event.event_id > starts[0].event_id + ] + if starts and completions: + details = starts[0].callback_started_details + assert details is not None and details.callback_id is not None + return details.callback_id, (starts[0].event_id, completions[0].event_id) + time.sleep(0.01) + raise AssertionError(f"Callback {name} did not suspend") + + +@pytest.mark.parametrize( + "plugin_type", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +def test_external_callback_completion_precedes_later_replays( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + plugin_type: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = plugin_type( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + before_context = context.get_current() + common_spec = importlib.util.spec_from_file_location( + "common", SRC_DIR / "common.py" + ) + assert common_spec is not None and common_spec.loader is not None + common = importlib.util.module_from_spec(common_spec) + monkeypatch.setitem(sys.modules, "common", common) + common_spec.loader.exec_module(common) + monkeypatch.setattr(common, "otel_plugin_factory", lambda: plugin) + spec = importlib.util.spec_from_file_location( + "otel_26_external_callback_completion_replay", + SRC_DIR / "otel_26_external_callback_completion_replay.py", + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + calls: list[str] = [] + original_observe = module.observe_target + + def terminal_target_exports() -> tuple[ReadableSpan, ...]: + return tuple( + span + for span in exporter.get_finished_spans() + if span.name == "otel-external-target" + and (span.attributes or {}).get("durable.operation.status") == "SUCCEEDED" + ) + + def counted_observe(value: str) -> Any: + step = original_observe(value) + + def counted(step_context: Any) -> str: + targets = terminal_target_exports() + assert len(targets) == 1, ( + "target completion must reach the exporter before the observed step" + ) + calls.append(value) + return step(step_context) + + return counted + + monkeypatch.setattr(module, "observe_target", counted_observe) + gates: list[tuple[int, int]] = [] + try: + with DurableFunctionTestRunner( + handler=module.handler, + store=FileSystemExecutionStore.create(tmp_path), + poll_interval=0.01, + execution_timeout=25, + skip_time=False, + ) as runner: + arn = runner.run_async( + input=json.dumps({"scenario": "external-callback-completion-replay"}) + ) + for name, payload in ( + ("otel-external-target", "target"), + ("otel-external-barrier-one create callback id", "one"), + ("otel-external-barrier-two create callback id", "two"), + ): + callback_id, gate = _suspended_callback(runner, arn, name) + gates.append(gate) + if name == "otel-external-target": + assert terminal_target_exports() == () + else: + # Each barrier is suspended before its success is sent. + # The raw export count must stay at one across later + # invocations, even if duplicate records reuse a span ID. + assert len(terminal_target_exports()) == 1, ( + f"target export count changed before completing {name}" + ) + runner.send_callback_success( + callback_id, result=json.dumps(payload).encode("utf-8") + ) + result = runner.wait_for_result(arn, timeout=10) + history = runner.get_execution_history(arn, include_execution_data=True) + + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == "target/one/two" + assert calls == ["target"] + assert gates == [(2, 3), (8, 11), (15, 18)] + assert [ + event.event_id + for event in history.events + if event.event_type == "InvocationCompleted" + ] == [3, 11, 18, 21] + assert history.events[-1].event_type == "ExecutionSucceeded" + assert history.events[-1].event_id == 22 + + spans = exporter.get_finished_spans() + assert all(span.context is not None for span in spans) + assert len({span.context.trace_id for span in spans if span.context}) == 1 + invocations = sorted( + [span for span in spans if span.name == "Invocation"], + key=lambda span: span.start_time or 0, + ) + assert len(invocations) == 4 + assert [ + (span.attributes or {}).get("durable.invocation.status") + for span in invocations + ] == ["PENDING", "PENDING", "PENDING", "SUCCEEDED"] + targets = terminal_target_exports() + # Count raw exports, including duplicate records with identical span IDs. + assert len(targets) == 1 + target = targets[0] + assert target.status.status_code.name == "OK" + observed = [ + span + for span in spans + if span.name == "otel-external-target-observed attempt 1" + ] + assert len(observed) == 1 + assert target.end_time is not None and observed[0].start_time is not None + assert target.end_time <= observed[0].start_time + assert invocations[2].start_time is not None + assert invocations[3].start_time is not None + assert target.end_time < invocations[2].start_time < invocations[3].start_time + parent = ( + invocations[1] + if plugin_type is InvocationOtelPluginFactory + else next(span for span in spans if span.name == "Workflow") + ) + assert target.parent is not None and parent.context is not None + assert target.parent.span_id == parent.context.span_id + # Invocation view also exports the legitimate first-invocation segment. + pending_target = [ + span + for span in spans + if span.name == "otel-external-target" + and (span.attributes or {}).get("durable.operation.status") == "STARTED" + ] + assert len(pending_target) == ( + 1 if plugin_type is InvocationOtelPluginFactory else 0 + ) + assert context.get_current() == before_context + finally: + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_invocation_retry_status.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_invocation_retry_status.py new file mode 100644 index 000000000..b9e0d2793 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_invocation_retry_status.py @@ -0,0 +1,123 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Exercise a real invocation retry through the public decorator and local runner.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any + +import pytest +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, +) +from aws_durable_execution_sdk_python.plugin import InvocationEndInfo, InvocationStatus +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry import context +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from aws_durable_execution_sdk_python_otel import ( + ExecutionOtelPluginFactory, + InvocationOtelPluginFactory, + OtelPluginConfig, +) + + +SRC_DIR = Path(__file__).resolve().parents[1] / "src" + + +@pytest.mark.parametrize( + "plugin_type", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +def test_invocation_retry_preserves_completed_step( + monkeypatch: pytest.MonkeyPatch, + plugin_type: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = plugin_type( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + statuses: list[InvocationStatus] = [] + original_create = plugin.create_plugin + + def create_plugin(info: InvocationStartInfo) -> DurableInstrumentationPlugin: + instance = original_create(info) + original_end = instance.on_invocation_end + + def observe_end(info: InvocationEndInfo) -> None: + statuses.append(info.status) + original_end(info) + + monkeypatch.setattr(instance, "on_invocation_end", observe_end) + return instance + + monkeypatch.setattr(plugin, "create_plugin", create_plugin) + before_context = context.get_current() + common_spec = importlib.util.spec_from_file_location( + "common", SRC_DIR / "common.py" + ) + assert common_spec is not None and common_spec.loader is not None + common = importlib.util.module_from_spec(common_spec) + monkeypatch.setitem(sys.modules, "common", common) + common_spec.loader.exec_module(common) + monkeypatch.setattr(common, "otel_plugin_factory", lambda: plugin) + spec = importlib.util.spec_from_file_location( + "otel_24_invocation_retry_status", + SRC_DIR / "otel_24_invocation_retry_status.py", + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + calls: list[str] = [] + + def track(factory: Any, label: str) -> Any: + def make_step() -> Any: + step = factory() + + def counted(step_context: Any) -> str: + calls.append(label) + return step(step_context) + + return counted + + return make_step + + monkeypatch.setattr( + module, + "before_invocation_retry", + track(module.before_invocation_retry, "saved"), + ) + try: + with DurableFunctionTestRunner(handler=module.handler) as runner: + result = runner.run( + input=json.dumps({"scenario": "invocation-retry-status"}), timeout=15 + ) + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == "retry-complete" + assert calls == ["saved"] + assert statuses == [InvocationStatus.RETRY, InvocationStatus.SUCCEEDED] + spans = exporter.get_finished_spans() + invocations = [span for span in spans if span.name == "Invocation"] + assert len(invocations) == 2 + assert [span.status.status_code.name for span in invocations] == ["UNSET", "OK"] + workflows = [span for span in spans if span.name == "Workflow"] + assert len(workflows) == 1 + assert workflows[0].status.status_code.name == "OK" + assert context.get_current() == before_context + finally: + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_otel_examples.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_otel_examples.py index 3ccd0d72d..66ec0edf4 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_otel_examples.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_otel_examples.py @@ -50,6 +50,12 @@ ("Otel18ChainedInvokeFailure", "otel-invocation-18"), ("Otel19ExecutionFailure", "otel-invocation-19"), ("Otel20VirtualContext", "otel-invocation-20"), + ("Otel22UserFunctionContext", "otel-invocation-22"), + ("Otel23CallbackFunctionContext", "otel-invocation-23"), + ("Otel24InvocationRetryStatus", "otel-invocation-24"), + ("Otel25CallbackFailureWithoutError", "otel-invocation-25"), + ("Otel26ExternalCallbackReplay", "otel-invocation-26"), + ("Otel27ChainedInvokeSuspend", "otel-invocation-27"), ("OtelExecution1Success", "otel-execution-1"), ("OtelExecution2WaitResume", "otel-execution-2"), ("OtelExecution3Retry", "otel-execution-3"), @@ -70,6 +76,12 @@ ("OtelExecution18ChainedInvokeFailure", "otel-execution-18"), ("OtelExecution19ExecutionFailure", "otel-execution-19"), ("OtelExecution20VirtualContext", "otel-execution-20"), + ("OtelExecution22UserFunctionContext", "otel-execution-22"), + ("OtelExecution23CallbackFunctionContext", "otel-execution-23"), + ("OtelExecution24InvocationRetryStatus", "otel-execution-24"), + ("OtelExecution25CallbackFailureWithoutError", "otel-execution-25"), + ("OtelExecution26ExternalCallbackReplay", "otel-execution-26"), + ("OtelExecution27ChainedInvokeSuspend", "otel-execution-27"), ] EXPECTED_LONG_RUNNING_MAPPINGS: list[tuple[str, str]] = [ ("OtelLongRunning1Wait", "otel-long-running-1"), @@ -80,6 +92,8 @@ # Invoke targets carry no requirement of their own; the chained-invoke cases # assert on the telemetry the target produces. EXPECTED_TARGETS: dict[str, str] = { + "Otel27InvokeTarget": "otel-invocation-27-target", + "OtelExecution27InvokeTarget": "otel-execution-27-target", "Otel11InvokeTarget": "otel-invocation-11-target", "Otel18InvokeTarget": "otel-invocation-18-target", "OtelExecution11InvokeTarget": "otel-execution-11-target", @@ -89,6 +103,12 @@ "OtelLongRunning4InvokeTarget": "otel-long-running-4-target", } CHAINED_INVOKE_PAIRS: list[tuple[Path, str, str]] = [ + (TEMPLATE_PATH, "Otel27ChainedInvokeSuspend", "Otel27InvokeTarget"), + ( + TEMPLATE_PATH, + "OtelExecution27ChainedInvokeSuspend", + "OtelExecution27InvokeTarget", + ), (TEMPLATE_PATH, "Otel11ChainedInvoke", "Otel11InvokeTarget"), (TEMPLATE_PATH, "Otel18ChainedInvokeFailure", "Otel18InvokeTarget"), (TEMPLATE_PATH, "OtelExecution11ChainedInvoke", "OtelExecution11InvokeTarget"), @@ -143,6 +163,7 @@ "otel_9_wait_for_condition", "otel_10_wait_for_callback", "otel_11_chained_invoke", + "otel_27_chained_invoke_suspend", "otel_12_child_context_failure", "otel_13_parallel_failure", "otel_14_map_failure", @@ -152,6 +173,12 @@ "otel_18_chained_invoke_failure", "otel_19_execution_failure", "otel_20_virtual_context", + "otel_21_completed_step_replay", + "otel_22_user_function_context", + "otel_23_callback_function_context", + "otel_24_invocation_retry_status", + "otel_25_caught_callback_failure_replay", + "otel_26_external_callback_completion_replay", "otel_long_running_1_wait", "otel_long_running_2_retry", "otel_long_running_3_callback", @@ -271,14 +298,10 @@ def test_deferred_requirements_are_explicit(logical_id: str, suite: str) -> None "NotImplemented" ] } - assert set(gaps) == {f"{suite}-{case}" for case in range(21, 28)} - for case in (21, 22, 23, 24, 27): - assert "pull/796" in gaps[f"{suite}-{case}"] - assert "issues/853" in gaps[f"{suite}-25"] - assert "issues/862" in gaps[f"{suite}-25"] - assert "issues/854" in gaps[f"{suite}-26"] - assert "issues/860" in gaps[f"{suite}-26"] - assert "issues/861" in gaps[f"{suite}-26"] + assert set(gaps) == {f"{suite}-21"} + assert "WaitSucceeded" in gaps[f"{suite}-21"] + assert "InvocationCompleted" in gaps[f"{suite}-21"] + assert "discussion_r4394770596" in gaps[f"{suite}-21"] covered = { requirement for _, requirement in requirement_mappings(template) @@ -420,6 +443,25 @@ def test_every_registered_handler_exists( ) +def test_active_callback_and_retry_handlers_keep_their_mapping( + template: dict[str, Any], +) -> None: + resources = template["Resources"] + for prefix in ("Otel", "OtelExecution"): + assert resources[prefix + "17WaitForCallbackFailure"]["Properties"][ + "Handler" + ] == ("otel_17_wait_for_callback_failure.handler") + assert resources[prefix + "24InvocationRetryStatus"]["Properties"][ + "Handler" + ] == ("otel_24_invocation_retry_status.handler") + assert resources[prefix + "25CallbackFailureWithoutError"]["Properties"][ + "Handler" + ] == ("otel_25_caught_callback_failure_replay.handler") + assert resources[prefix + "26ExternalCallbackReplay"]["Properties"][ + "Handler" + ] == ("otel_26_external_callback_completion_replay.handler") + + def test_template_deploys_only_the_selected_view(template: dict[str, Any]) -> None: assert template["Parameters"]["OtelSuite"]["AllowedValues"] == [ "all", diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_user_function_context.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_user_function_context.py new file mode 100644 index 000000000..a54f3fa87 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_user_function_context.py @@ -0,0 +1,262 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Exercise case 22's callback contexts through real concurrent work and resume.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from collections import Counter +from contextlib import nullcontext +from pathlib import Path +from threading import Barrier +from typing import Any + +import pytest +from aws_durable_execution_sdk_python.execution import DurableExecutionInvocationInput +from aws_durable_execution_sdk_python.types import LambdaContext +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry import context, trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from aws_durable_execution_sdk_python_otel import ( + ExecutionOtelPluginFactory, + InvocationOtelPluginFactory, + OtelPluginConfig, +) + + +SRC_DIR = Path(__file__).resolve().parents[1] / "src" +CALLBACK_PARENTS = { + "step": "otel-context-step attempt 1", + "child": "otel-context-child", + "child-step": "otel-context-child-step attempt 1", + "child-restored": "otel-context-child", + "parallel-a": "otel-context-branch-a", + "parallel-step-a": "otel-context-branch-step-a attempt 1", + "parallel-b": "otel-context-branch-b", + "parallel-step-b": "otel-context-branch-step-b attempt 1", + "map-0": "otel-context-iteration-0", + "map-step-0": "otel-context-map-step-0 attempt 1", + "map-1": "otel-context-iteration-1", + "map-step-1": "otel-context-map-step-1 attempt 1", +} +HANDLER_COUNTS = {"handler": 2, "handler-restored": 2, "handler-after-resume": 1} + + +@pytest.mark.parametrize( + "plugin_type", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +@pytest.mark.parametrize( + "ambient", [False, True], ids=["no-ambient", "unrelated-ambient"] +) +def test_user_function_probes_keep_sdk_context_across_resume( + monkeypatch: pytest.MonkeyPatch, + plugin_type: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], + ambient: bool, +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) + plugin = plugin_type( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + before_context = context.get_current() + common_spec = importlib.util.spec_from_file_location( + "common", SRC_DIR / "common.py" + ) + assert common_spec is not None and common_spec.loader is not None + common = importlib.util.module_from_spec(common_spec) + monkeypatch.setitem(sys.modules, "common", common) + common_spec.loader.exec_module(common) + monkeypatch.setattr(common, "otel_plugin_factory", lambda: plugin) + spec = importlib.util.spec_from_file_location( + "otel_22_user_function_context", SRC_DIR / "otel_22_user_function_context.py" + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + barriers: dict[str, Barrier] = {} + for pair in ( + ("parallel-a", "parallel-b"), + ("parallel-step-a", "parallel-step-b"), + ("map-0", "map-1"), + ("map-step-0", "map-step-1"), + ): + barrier = Barrier(2, timeout=5) + barriers.update(dict.fromkeys(pair, barrier)) + original_probe = module.probe + + def synchronized_probe(label: str) -> None: + callback_context = context.get_current() + # Both sibling callbacks must be active before either observes its + # parent. Synchronize only in the test; never supply a tracing context. + if label in barriers: + barriers[label].wait() + original_probe(label) + if label in barriers: + barriers[label].wait() + assert context.get_current() is callback_context + + monkeypatch.setattr(module, "probe", synchronized_probe) + invocation_contexts: list[tuple[context.Context, context.Context]] = [] + + def entry( + event: DurableExecutionInvocationInput, lambda_context: LambdaContext + ) -> dict[str, Any]: + # Simulate host instrumentation outside the unmodified durable handler. + # With no extracted parent, its unrelated trace must not become the + # parent of any SDK operation or handler probe. + host_context = context.get_current() + scope = ( + provider.get_tracer("host-instrumentation").start_as_current_span( + "ambient-invocation" + ) + if ambient + else nullcontext() + ) + try: + with scope: + invocation_context = context.get_current() + try: + return module.handler(event, lambda_context) + finally: + invocation_contexts.append( + (invocation_context, context.get_current()) + ) + finally: + assert context.get_current() is host_context + + try: + with DurableFunctionTestRunner(handler=entry, skip_time=False) as runner: + arn = runner.run_async( + input=json.dumps({"scenario": "user-function-context"}), + execution_timeout=30, + ) + result = runner.wait_for_result(arn, timeout=30) + history = runner.get_execution_history(arn) + assert result.status.value == "SUCCEEDED", result.error + assert result.result is not None + assert json.loads(result.result) == "context-complete" + assert [ + event.event_type + for event in history.events + if event.event_type.startswith("Wait") + ] == ["WaitStarted", "WaitSucceeded"] + + spans = exporter.get_finished_spans() + invocations = [span for span in spans if span.name == "Invocation"] + assert len(invocations) == 2 + assert [ + span.attributes["durable.invocation.status"] + for span in invocations + if span.attributes is not None + ] == ["PENDING", "SUCCEEDED"] + assert [ + span.attributes["durable.invocation.first"] + for span in invocations + if span.attributes is not None + ] == [True, False] + workflows = [span for span in spans if span.name == "Workflow"] + assert len(workflows) == 1 + assert workflows[0].context is not None + canonical_trace_id = workflows[0].context.trace_id + assert { + span.context.trace_id + for span in spans + if span.context is not None + and span.attributes is not None + and span.attributes.get("durable.execution.arn") == arn + } == {canonical_trace_id} + + probes = [span for span in spans if span.name.startswith("conformance.")] + # Count raw exports so replayed callback bodies and duplicate exports + # fail even if they reuse a span ID. + assert Counter(span.name for span in probes) == Counter( + { + f"conformance.{label}": count + for label, count in ( + {**dict.fromkeys(CALLBACK_PARENTS, 1), **HANDLER_COUNTS} + ).items() + } + ) + spans_by_id = { + (span.context.trace_id, span.context.span_id): span + for span in spans + if span.context is not None + } + handler_parent = ( + "Workflow" if plugin_type is ExecutionOtelPluginFactory else "Invocation" + ) + for probe_span in probes: + assert probe_span.context is not None + assert probe_span.context.trace_id == canonical_trace_id + assert probe_span.parent is not None + assert probe_span.parent.trace_id == canonical_trace_id + assert probe_span.attributes is not None + assert "durable.execution.arn" not in probe_span.attributes + label = str(probe_span.attributes["conformance.callback"]) + assert probe_span.name == f"conformance.{label}" + parent = spans_by_id[(canonical_trace_id, probe_span.parent.span_id)] + assert parent.name == CALLBACK_PARENTS.get(label, handler_parent) + assert parent.attributes is not None + assert parent.attributes["durable.execution.arn"] == arn + + for index, invocation in enumerate(invocations): + assert invocation.start_time is not None and invocation.end_time is not None + invocation_probes = [ + span + for span in probes + if span.start_time is not None + and invocation.start_time <= span.start_time <= invocation.end_time + ] + expected = ( + { + **dict.fromkeys(CALLBACK_PARENTS, 1), + "handler": 1, + "handler-restored": 1, + } + if index == 0 + else dict.fromkeys(HANDLER_COUNTS, 1) + ) + assert Counter(span.name for span in invocation_probes) == Counter( + {f"conformance.{label}": count for label, count in expected.items()} + ) + handler_probes = [ + span + for span in invocation_probes + if span.attributes is not None + and span.attributes["conformance.callback"] in HANDLER_COUNTS + ] + expected_parent = ( + workflows[0] + if plugin_type is ExecutionOtelPluginFactory + else invocation + ) + assert all( + span.parent == expected_parent.context for span in handler_probes + ) + + ambient_spans = [span for span in spans if span.name == "ambient-invocation"] + assert len(ambient_spans) == (2 if ambient else 0) + for span in ambient_spans: + assert span.context is not None + assert span.context.trace_id != canonical_trace_id + # Observe restoration on the runner's invocation threads, including + # PENDING, rather than checking only the pytest caller's context. + assert len(invocation_contexts) == 2 + assert all(before is after for before, after in invocation_contexts) + assert context.get_current() is before_context + finally: + provider.shutdown()