feat(plugin): propagate chained invoke trace context in START checkpoints

packages/aws-durable-execution-sdk-python-otel/README.md

@@ -424,6 +424,40 @@ context onto every emitted log record using these attributes:
 These attributes are only set when a valid span context is active, so any log
 formatter or schema must treat the fields as optional.
 
+## Chained-invoke propagation
+
+Both views supply the synchronous `provide_propagation_metadata` hook with
+SDK-owned types. It encodes canonical X-Ray `Root`, the calling operation's
+`Parent`, and resolved `Sampled=1` or `Sampled=0`, without creating an extra span.
+An existing operation's actual span ID is used; before span creation, its stable
+ID is derived from the execution ARN and operation ID. Unrelated ambient spans
+cannot replace execution ownership. An operation that has already started uses
+its actual span's sampled flag. A deferred operation contributes metadata only
+while the instrumentation tracer still uses the installed `DurableSampler`:
+replacing that sampler after invocation Start may give the operation a different
+sampling decision, so the producer abstains instead of guessing or querying the
+sampler early. Inactive/mismatched executions contribute nothing. Tracer,
+provider and resource ownership remain unchanged.
+
+The core now calls the collector only for a new invoke START and persists the
+contribution in flat `ChainedInvokeOptions.XAmznTraceId`. Pending and terminal
+replay do not recollect metadata; an uncommitted START can be retried. Separate
+invokes carry separate operation parents. Public invoke tests cover both views,
+sampled/unsampled context, parallel branches, preserved tenant/payload and replay.
+
+For [#751](https://github.com/aws/aws-durable-execution-sdk-python/issues/751),
+deployed backend delivery and topology still require validation. The core requires
+boto3 1.43.114 or newer, which depends on a botocore model containing the optional
+field. The normal request tests use the published serializer and parameter
+validation. Python has no distributed-map
+model/START path; existing map/parallel APIs are CONTEXT operations.
+
+On older supported cores without the propagation contract, plugin loading and
+existing tracing continue, and the optional hook contributes no metadata. The
+new capability requires the coordinated core and its boto3 minimum. The core/OTel
+package floor and provider API version are unchanged; the published model alone
+does not establish deployed downstream topology.
+
 ## Verification
 
 After deploying your function with the plugin configured:

packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py

@@ -34,7 +34,7 @@
 import datetime
 import logging
 import threading
-from typing import Any, ClassVar
+from typing import Any, ClassVar, TYPE_CHECKING, cast
 
 from aws_durable_execution_sdk_python.plugin import (
     DurableInstrumentationPlugin,
@@ -97,9 +97,22 @@
     uninstall_log_filter,
 )
 from aws_durable_execution_sdk_python_otel.provider import create_tracer_provider
+from aws_durable_execution_sdk_python_otel.propagation import propagation_metadata
 from aws_durable_execution_sdk_python_otel.timestamps import to_otel_timestamp
 
 
+if TYPE_CHECKING:
+    from aws_durable_execution_sdk_python.plugin import (
+        PropagationInput,
+        PropagationMetadata,
+    )
+else:
+    from aws_durable_execution_sdk_python import plugin as _core_plugin
+
+    PropagationInput = getattr(_core_plugin, "PropagationInput", Any)
+    PropagationMetadata = getattr(_core_plugin, "PropagationMetadata", Any)
+
+
 logger = logging.getLogger(__name__)
 
 _TERMINAL_INVOCATION_STATUSES = frozenset(
@@ -430,6 +443,40 @@ def _with_sampling(self, parent_context: Context) -> Context:
     # ------------------------------------------------------------------
     # Invocation lifecycle
     # ------------------------------------------------------------------
+    def provide_propagation_metadata(
+        self,
+        info: PropagationInput,
+    ) -> PropagationMetadata | None:
+        """Describe this operation as the downstream parent without starting a span."""
+        execution_context = self._execution_trace_context
+        if (
+            not self._tracing_enabled
+            or execution_context is None
+            or info.execution_arn != self._execution_arn
+            or not info.operation_id
+        ):
+            return None
+        operation_span = self._get_span(info.operation_id)
+        if (
+            operation_span is None or isinstance(operation_span, DurableParentSpan)
+        ) and type(cast(SdkTracer, self._tracer).sampler) is not DurableSampler:
+            # A deferred identity has no operation sampling decision yet. A
+            # replaced sampler may decide differently for Invocation and invoke;
+            # do not guess its result or invoke it early to produce a carrier.
+            return None
+        span_context = (
+            operation_span.get_span_context()
+            if operation_span is not None
+            else self._operation_span_context(info.operation_id)
+        )
+        if (
+            span_context is None
+            or not span_context.is_valid
+            or span_context.trace_id != execution_context.trace_id
+        ):
+            return None
+        return propagation_metadata(span_context)
+
     def on_invocation_start(self, info: InvocationStartInfo) -> None:
         logger.debug("Durable invocation started: %s", info)
         self._registration_accepted = True

packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py

@@ -5,7 +5,7 @@
 import datetime
 import logging
 import threading
-from typing import Any, ClassVar
+from typing import Any, ClassVar, TYPE_CHECKING, cast
 
 from aws_durable_execution_sdk_python.plugin import (
     DurableInstrumentationPlugin,
@@ -66,9 +66,22 @@
 )
 from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig
 from aws_durable_execution_sdk_python_otel.provider import create_tracer_provider
+from aws_durable_execution_sdk_python_otel.propagation import propagation_metadata
 from aws_durable_execution_sdk_python_otel.timestamps import to_otel_timestamp
 
 
+if TYPE_CHECKING:
+    from aws_durable_execution_sdk_python.plugin import (
+        PropagationInput,
+        PropagationMetadata,
+    )
+else:
+    from aws_durable_execution_sdk_python import plugin as _core_plugin
+
+    PropagationInput = getattr(_core_plugin, "PropagationInput", Any)
+    PropagationMetadata = getattr(_core_plugin, "PropagationMetadata", Any)
+
+
 logger = logging.getLogger(__name__)
 
 _TERMINAL_INVOCATION_STATUSES = frozenset(
@@ -522,6 +535,40 @@ def _end_span(
     # ------------------------------------------------------------------
     # Plugin lifecycle callbacks
     # ------------------------------------------------------------------
+    def provide_propagation_metadata(
+        self,
+        info: PropagationInput,
+    ) -> PropagationMetadata | None:
+        """Describe this operation as the downstream parent without starting a span."""
+        execution_context = self._execution_trace_context
+        if (
+            not self._tracing_enabled
+            or execution_context is None
+            or info.execution_arn != self._execution_arn
+            or not info.operation_id
+        ):
+            return None
+        operation_span = self._get_span(info.operation_id)
+        if (
+            operation_span is None or isinstance(operation_span, DurableParentSpan)
+        ) and type(cast(SdkTracer, self._tracer).sampler) is not DurableSampler:
+            # A deferred identity has no operation sampling decision yet. A
+            # replaced sampler may decide differently for Invocation and invoke;
+            # do not guess its result or invoke it early to produce a carrier.
+            return None
+        span_context = (
+            operation_span.get_span_context()
+            if operation_span is not None
+            else self._operation_link_context(info.operation_id)
+        )
+        if (
+            span_context is None
+            or not span_context.is_valid
+            or span_context.trace_id != execution_context.trace_id
+        ):
+            return None
+        return propagation_metadata(span_context)
+
     def on_invocation_start(self, info: InvocationStartInfo) -> None:
         """Called at the start of each invocation. Creates the invocation span."""
         logger.debug("Durable invocation started: %s", info)

packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/propagation.py

@@ -0,0 +1,31 @@
+"""Pure X-Ray propagation encoding for the SDK-owned plugin contract."""
+
+from __future__ import annotations
+
+from typing import Any, TYPE_CHECKING
+
+from aws_durable_execution_sdk_python import plugin as core_plugin
+from opentelemetry.trace import SpanContext
+
+
+if TYPE_CHECKING:
+    from aws_durable_execution_sdk_python.plugin import PropagationMetadata
+else:
+    PropagationMetadata = getattr(core_plugin, "PropagationMetadata", Any)
+
+
+def propagation_metadata(span_context: SpanContext) -> PropagationMetadata | None:
+    """Encode an operation's context without creating a span or changing state."""
+    # Old supported cores do not expose the additive contract. Existing tracing
+    # still works; only this new optional contribution is unavailable there.
+    metadata_type = getattr(core_plugin, "PropagationMetadata", None)
+    if metadata_type is None:
+        return None
+    trace_id = f"{span_context.trace_id:032x}"
+    return metadata_type(
+        x_amzn_trace_id=(
+            f"Root=1-{trace_id[:8]}-{trace_id[8:]};"
+            f"Parent={span_context.span_id:016x};"
+            f"Sampled={int(span_context.trace_flags.sampled)}"
+        )
+    )