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

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

@@ -343,6 +343,34 @@ 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.
 
+## Draft 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. 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.
+
+This PR stays draft for [#751](https://github.com/aws/aws-durable-execution-sdk-python/issues/751)
+until the public generated model and backend support are available. The normal
+botocore request tests intentionally expose the missing field; they do not bypass
+the serializer or hide the dependency failure. 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. Dependency floors and provider API
+version are unchanged; deployed downstream topology still needs validation after
+model/backend publication.
+
 ## 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
 
 from aws_durable_execution_sdk_python.plugin import (
     DurableInstrumentationPlugin,
@@ -96,6 +96,19 @@
     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
+
+
+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__)
@@ -428,6 +441,33 @@ 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)
+        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
 
 from aws_durable_execution_sdk_python.plugin import (
     DurableInstrumentationPlugin,
@@ -65,6 +65,19 @@
 )
 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
+
+
+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__)
@@ -523,6 +536,33 @@ 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)
+        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)}"
+        )
+    )