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

.github/scripts/build_lambda_layer.py

@@ -71,11 +71,13 @@ def _install_layer_distributions(config: BuildConfig, target_dir: Path) -> None:
         "--only-binary",
         ":all:",
         "--no-compile",
-        "--no-deps",
-        str(config.sdk_distribution),
-        str(config.otel_distribution),
     ]
-    subprocess.run(command, check=True)
+    # The Core wheel's requirements also apply to layer deployments, including
+    # the botocore service model needed by its public Lambda operations.
+    subprocess.run([*command, str(config.sdk_distribution)], check=True)
+    # Telemetry providers belong to the ADOT/community layer. Do not resolve
+    # plugin dependencies here and accidentally shadow that distribution.
+    subprocess.run([*command, "--no-deps", str(config.otel_distribution)], check=True)
 
 
 def _write_zip(output: Path, layer_root: Path) -> None:

.github/scripts/tests/test_build_lambda_layer.py

@@ -19,6 +19,7 @@ def _write_test_wheel(
     distribution: str,
     package_files: tuple[str, ...],
     dependencies: tuple[str, ...] = (),
+    package_sources: dict[str, str] | None = None,
 ) -> Path:
     normalized_distribution = distribution.replace("-", "_")
     wheel = directory / f"{normalized_distribution}-1.0.0-py3-none-any.whl"
@@ -33,7 +34,9 @@ def _write_test_wheel(
 
     with zipfile.ZipFile(wheel, "w") as archive:
         for package_file in package_files:
-            archive.writestr(package_file, "")
+            archive.writestr(
+                package_file, (package_sources or {}).get(package_file, "")
+            )
         archive.writestr(f"{dist_info}/METADATA", "\n".join(metadata))
         archive.writestr(
             f"{dist_info}/WHEEL",
@@ -67,12 +70,13 @@ def test_build_layer_installs_distributions_and_zips_lambda_layout(
     def fake_run(command: list[str], check: bool) -> subprocess.CompletedProcess[str]:
         commands.append(command)
         target = Path(command[command.index("--target") + 1])
-        (target / "aws_durable_execution_sdk_python").mkdir()
-        (target / "aws_durable_execution_sdk_python" / "__init__.py").write_text("")
-        (target / "aws_durable_execution_sdk_python_otel").mkdir()
-        (target / "aws_durable_execution_sdk_python_otel" / "__init__.py").write_text(
-            ""
-        )
+        for wheel, package in (
+            (sdk_wheel, "aws_durable_execution_sdk_python"),
+            (otel_wheel, "aws_durable_execution_sdk_python_otel"),
+        ):
+            if str(wheel) in command:
+                (target / package).mkdir(exist_ok=True)
+                (target / package / "__init__.py").write_text("")
         return subprocess.CompletedProcess(command, 0)
 
     monkeypatch.setattr(subprocess, "run", fake_run)
@@ -86,13 +90,15 @@ def fake_run(command: list[str], check: bool) -> subprocess.CompletedProcess[str
     )
 
     assert output == tmp_path / "layer.zip"
-    assert "--no-compile" in commands[0]
-    assert "--no-deps" in commands[0]
-    assert "--platform" not in commands[0]
-    assert "--python-version" not in commands[0]
-    assert "--abi" not in commands[0]
-    assert str(sdk_wheel) in commands[0]
-    assert str(otel_wheel) in commands[0]
+    sdk_command = next(command for command in commands if str(sdk_wheel) in command)
+    otel_command = next(command for command in commands if str(otel_wheel) in command)
+    assert "--no-deps" not in sdk_command
+    assert "--no-deps" in otel_command
+    for command in commands:
+        assert "--no-compile" in command
+        assert "--platform" not in command
+        assert "--python-version" not in command
+        assert "--abi" not in command
 
     with zipfile.ZipFile(output) as archive:
         assert (
@@ -104,7 +110,7 @@ def fake_run(command: list[str], check: bool) -> subprocess.CompletedProcess[str
         )
 
 
-def test_build_layer_excludes_adot_and_runtime_dependencies(
+def test_build_layer_includes_sdk_dependencies_without_shadowing_adot(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
 ) -> None:
@@ -120,7 +126,16 @@ def test_build_layer_excludes_adot_and_runtime_dependencies(
         ("aws_durable_execution_sdk_python_otel/__init__.py",),
         ("aws-opentelemetry-distro", "opentelemetry-api"),
     )
-    _write_test_wheel(tmp_path, "boto3", ("boto3/__init__.py",))
+    # Minimal offline wheel fixtures exercise dependency resolution and layer
+    # import precedence. They contain no service model or simulated wire API.
+    _write_test_wheel(
+        tmp_path,
+        "boto3",
+        ("boto3/__init__.py",),
+        ("botocore",),
+        package_sources={"boto3/__init__.py": '__version__ = "layer-dependency"\n'},
+    )
+    _write_test_wheel(tmp_path, "botocore", ("botocore/__init__.py",))
     _write_test_wheel(
         tmp_path,
         "aws-opentelemetry-distro",
@@ -148,9 +163,28 @@ def test_build_layer_excludes_adot_and_runtime_dependencies(
     assert "python/aws_durable_execution_sdk_python/__init__.py" in names
     assert "python/aws_durable_execution_sdk_python_otel/__init__.py" in names
     assert not any(name.startswith("python/amazon/") for name in names)
-    assert not any(name.startswith("python/boto3/") for name in names)
+    assert "python/boto3/__init__.py" in names
+    assert "python/botocore/__init__.py" in names
     assert not any(name.startswith("python/opentelemetry/") for name in names)
 
+    layer_root = tmp_path / "unpacked"
+    with zipfile.ZipFile(output) as archive:
+        archive.extractall(layer_root)
+    runtime = tmp_path / "older-runtime"
+    (runtime / "boto3").mkdir(parents=True)
+    (runtime / "boto3" / "__init__.py").write_text('__version__ = "older-runtime"\n')
+    # Lambda searches the layer before its runtime SDK. An omitted dependency
+    # would silently import the older runtime package instead.
+    probe = (
+        "import pathlib,sys; "
+        f"layer=pathlib.Path({str(layer_root / 'python')!r}); "
+        f"sys.path[:0]=[str(layer),{str(runtime)!r}]; "
+        "import boto3; "
+        "assert boto3.__version__ == 'layer-dependency'; "
+        "assert pathlib.Path(boto3.__file__).is_relative_to(layer)"
+    )
+    subprocess.run([sys.executable, "-I", "-c", probe], check=True)
+
 
 def test_build_layer_requires_built_distributions(tmp_path: Path) -> None:
     with pytest.raises(FileNotFoundError):

RELEASING.md

@@ -74,6 +74,10 @@ Releases containing an `otel-v` tag also trigger the
 workflow. It builds the SDK and OTel plugin into Lambda layers for each
 supported Python runtime and architecture, then publishes public versions of the
 `aws-durable-execution-sdk-python-otel-plugin` layer.
+The archive resolves and includes the selected Core wheel's runtime dependencies,
+including boto3/botocore, so it does not depend on the Lambda runtime already
+having the required service model. The plugin wheel is installed without its
+optional provider dependencies; ADOT/community layers continue to own those.
 For OTel-only releases, the workflow downloads the exact SDK version pinned by
 `layer.sdk-version` in `.github/lambda-layer-publish.toml`; that version must
 already be published to PyPI. Combined SDK and OTel releases require the pin to

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

@@ -24,6 +24,12 @@ The base package intentionally does not install OpenTelemetry libraries. The
 Lambda layer supplies a version-aligned API, SDK, exporter, and propagators,
 preventing packages in the function artifact from shadowing parts of the layer.
 
+The generated `aws-durable-execution-sdk-python-otel-plugin` Lambda layer includes
+Core's runtime dependencies, including its required boto3/botocore service model.
+It still leaves OpenTelemetry provider packages to the ADOT/community layer.
+Dependencies bundled in the function artifact take precedence over layer packages;
+those application-owned boto3/botocore versions must also meet Core's minimum.
+
 For an application that configures its own OpenTelemetry provider instead of
 using a Lambda layer:
 
@@ -424,6 +430,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)}"
+        )
+    )