fix(otel): export sampled fallback roots early

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

@@ -312,8 +312,8 @@ Both bundled plugins use the same execution ancestor:
 
 - a propagated backend parent when `_X_AMZN_TRACE_ID` contains a valid `Root`
   and `Parent`
-- otherwise a deterministic, non-recording synthetic root derived from the
-  durable execution ARN
+- otherwise a deterministic `DurableExecutionRoot` span derived from the
+  durable execution ARN, with `durable.execution.synthetic_root=true`
 
 `InvocationOtelPlugin` keeps durable operation spans under the Invocation span
 and links operations to Workflow:
@@ -342,10 +342,46 @@ Invocation span uses that ambient span as its parent. Ambient spans on a
 different trace are ignored for durable parenting so Invocation remains on the
 execution trace.
 
+### Synthetic fallback anchor
+
+Each sampled fallback invocation creates and ends `DurableExecutionRoot` at
+invocation start and attempts to flush it through the normal invocation-end
+lifecycle, including when it returns `PENDING` or `RETRY`. A successfully
+delivered copy supplies ancestry if the execution is later stopped or times out
+without resuming. Export and flush failures, or termination before flushing,
+do not provide a delivery guarantee.
+
+The anchor uses the existing execution-scoped span ID and the stable execution
+start timestamp for both its start and end. It carries no terminal outcome;
+`Workflow` remains the span for execution duration and success/failure. Retries
+of the first invocation and later invocations may re-export the same anchor to
+recover lost telemetry; export is not exactly once. Two executions sharing a
+propagated trace ID retain distinct ARN-scoped anchor IDs.
+The anchor is intentionally zero-duration, so its descendants end after it;
+trace viewers may flag that timing relationship even though the ancestry is valid.
+
+The anchor uses the same configured tracer, provider resource and processors as
+other durable spans, but carries only the resolved sampling decision. Its SDK
+attributes are `durable.execution.arn` and `durable.execution.synthetic_root`;
+sampler-added attributes and tracestate are omitted from this repeated identity.
+Ordinary durable spans retain the complete resolved sampler metadata. SDK-added
+IDs, timestamps and anchor attributes stay stable across recovery exports;
+provider resources and custom processor enrichment may vary. The SDK does not
+replace configured service identity or freeze detector fields.
+
+Complete remote parents are externally owned and are never materialized by the
+SDK. Unsampled invocations emit no anchor. Sampling remains resolved per
+invocation: a later sampled fallback invocation also emits the anchor even if
+an earlier invocation was unsampled. Missing execution start time continues to
+disable telemetry; no wall-clock timestamp is invented for the anchor.
+
 ### Sampling
 
 Sampling is resolved once per invocation and carried to every durable span in
-that invocation. Precedence is:
+that invocation. Each span consumes its private sampling intent before processors
+run. Processor-created spans use their own configured sampling policy and normal
+parent-context semantics, including when a processor forwards the supplied parent
+context. Precedence is:
 
 1. `Sampled=1` or `Sampled=0` from `_X_AMZN_TRACE_ID`
 2. a same-trace ambient span's recording/sampled state
@@ -446,6 +482,11 @@ them to Invocation.
 
 A custom OpenTelemetry `IdGenerator` that produces reproducible trace and span IDs from execution metadata. Exported for advanced use cases.
 
+`use_ids` retains its existing scope-wide trace override by default. The
+synthetic-root implementation uses a separate internal one-span override,
+preventing processor-created roots from inheriting that internal ID. Existing
+`use_ids` signatures and default behavior remain unchanged.
+
 ### `xray_context_extractor`
 
 Default context extractor. Reads the `_X_AMZN_TRACE_ID` environment variable and

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

@@ -21,6 +21,8 @@
 class _IdOverride:
     trace_id: int | None
     span_id: int | None
+    trace_id_consumed: bool = False
+    consume_trace_id: bool = False
 
 
 def _to_otel_trace_id(execution_arn: str, start_timestamp: datetime) -> int:
@@ -83,9 +85,10 @@ def derive_workflow_span_id(durable_execution_arn: str) -> int:
 def derive_execution_root_span_id(durable_execution_arn: str) -> int:
     """Derive the deterministic synthetic execution-root span ID.
 
-    The synthetic root is a non-recording parent context used when the backend
-    does not provide a complete remote parent. Its ID is stable across
-    reinvocations and uses a namespace distinct from Workflow and operation
+    The synthetic root anchors the trace when the backend does not provide a
+    complete remote parent. Its zero-duration span is exported on each sampled
+    fallback invocation. Its ID is stable across reinvocations and uses a
+    namespace distinct from Workflow and operation
     span IDs.
     """
     if not durable_execution_arn:
@@ -143,28 +146,61 @@ def use_ids(self, *, trace_id: int | None, span_id: int | None) -> Iterator[None
         finally:
             self._id_override.reset(token)
 
+    @contextmanager
+    def _use_ids_for_span(
+        self,
+        *,
+        trace_id: int,
+        span_id: int,
+    ) -> Iterator[None]:
+        """Reserve IDs for one SDK-created span without changing public scope semantics."""
+        token = self._id_override.set(
+            _IdOverride(trace_id, span_id, consume_trace_id=True)
+        )
+        try:
+            yield
+        finally:
+            self._id_override.reset(token)
+
     def generate_trace_id(self) -> int:
         """Generate a 128-bit trace ID."""
         override = self._id_override.get()
         if override is not None and override.trace_id is not None:
+            # Consume the identity once, but keep its non-random classification
+            # until the SDK asks for the span ID below.
+            if override.consume_trace_id:
+                self._id_override.set(_IdOverride(None, override.span_id, True, True))
             return override.trace_id
+        if override is not None and override.trace_id_consumed:
+            self._id_override.set(
+                _IdOverride(
+                    None, override.span_id, consume_trace_id=override.consume_trace_id
+                )
+            )
         return self._fallback_id_generator.generate_trace_id()
 
     def generate_span_id(self) -> int:
         """Generate a 64-bit span ID."""
         override = self._id_override.get()
-        if override is not None and override.span_id is not None:
-            span_id = override.span_id
-            # Consume before returning so a re-entrant call in the same span
-            # creation falls back instead of reusing the deterministic ID.
-            self._id_override.set(_IdOverride(override.trace_id, None))
-            return span_id
+        if override is not None:
+            # Span ID generation completes the ID phase of SDK start_span, before
+            # processors run. Release the consumed trace classification too, so
+            # a processor creating another root sees only its own generator.
+            self._id_override.set(
+                _IdOverride(
+                    override.trace_id, None, consume_trace_id=override.consume_trace_id
+                )
+            )
+            if override.span_id is not None:
+                return override.span_id
         return self._fallback_id_generator.generate_span_id()
 
     def is_trace_id_random(self) -> bool:
         """Report whether the current trace ID is randomly generated."""
         override = self._id_override.get()
-        if override is not None and override.trace_id is not None:
+        if override is not None and (
+            override.trace_id is not None or override.trace_id_consumed
+        ):
             return False
         fallback_method = getattr(
             self._fallback_id_generator, "is_trace_id_random", None

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

@@ -5,6 +5,7 @@
 import functools
 import inspect
 from dataclasses import dataclass
+from threading import Lock
 from typing import Any
 
 from opentelemetry import context as otel_context
@@ -31,6 +32,20 @@ class DurableSamplingIntent:
     result: SamplingResult
 
 
+class _SamplingIntentDelivery:
+    """Give one sampler call the intent, even if its context is forwarded."""
+
+    def __init__(self, intent: DurableSamplingIntent) -> None:
+        self._intent: DurableSamplingIntent | None = intent
+        self._lock = Lock()
+
+    def consume(self) -> DurableSamplingIntent | None:
+        with self._lock:
+            intent = self._intent
+            self._intent = None
+            return intent
+
+
 class DurableSampler(Sampler):
     """Sampler that honors a durable sampling intent carried on parent context."""
 
@@ -56,8 +71,13 @@ def should_sample(
         links: Any = None,
         trace_state: Any = None,
     ) -> SamplingResult:
-        intent = otel_context.get_value(_DURABLE_SAMPLING_INTENT_KEY, parent_context)
-        if isinstance(intent, DurableSamplingIntent):
+        delivery = otel_context.get_value(_DURABLE_SAMPLING_INTENT_KEY, parent_context)
+        intent = (
+            delivery.consume()
+            if isinstance(delivery, _SamplingIntentDelivery)
+            else None
+        )
+        if intent is not None:
             merged_attributes = dict(attributes or {})
             merged_attributes.update(dict(intent.result.attributes or {}))
             return SamplingResult(
@@ -84,10 +104,17 @@ def store_sampling_intent(
     parent_context: Context,
     intent: DurableSamplingIntent | None,
 ) -> Context:
-    """Attach a durable sampling intent to a span parent context."""
+    """Attach a fresh single-use delivery of the invocation's sampling result.
+
+    The sampler consumes it before processors run. A processor forwarding this
+    context therefore uses its own sampling policy; another durable span gets a
+    new delivery without changing the reusable invocation-level intent.
+    """
     if intent is None:
         return parent_context
-    return otel_context.set_value(_DURABLE_SAMPLING_INTENT_KEY, intent, parent_context)
+    return otel_context.set_value(
+        _DURABLE_SAMPLING_INTENT_KEY, _SamplingIntentDelivery(intent), parent_context
+    )
 
 
 def resolve_sampling_result(

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

@@ -86,6 +86,7 @@
     DurableParentSpan,
     ensure_end_after_start,
 )
+from aws_durable_execution_sdk_python_otel.execution_root import ExecutionRoot
 from aws_durable_execution_sdk_python_otel.execution_trace_context import (
     ExecutionTraceContext,
     canonical_trace_id,
@@ -96,6 +97,7 @@
     uninstall_log_filter,
 )
 from aws_durable_execution_sdk_python_otel.provider import create_tracer_provider
+from aws_durable_execution_sdk_python_otel.timestamps import to_otel_timestamp
 
 
 logger = logging.getLogger(__name__)
@@ -115,7 +117,7 @@ def _to_otel_timestamp(dt: datetime.datetime | None) -> int | None:
     """Convert a datetime to an OTel timestamp (ns since epoch), or None."""
     if dt is None:
         return None
-    return int(dt.timestamp() * 1_000_000_000)
+    return to_otel_timestamp(dt)
 
 
 class ExecutionOtelPlugin(DurableInstrumentationPlugin):
@@ -489,6 +491,14 @@ def on_invocation_start(self, info: InvocationStartInfo) -> None:
             root_sampled=lambda: is_sampled(sampling_result),
         )
 
+        # Materialize the fallback anchor before user code can suspend or fail.
+        # Repeated invocations retain the same SDK-owned identity and timing.
+        ExecutionRoot(
+            execution_arn=self._execution_arn,
+            ancestor=self._execution_trace_context.execution_ancestor,
+            start_time=info.execution_start_time,
+        ).export(self._tracer, self._id_generator, self._sampling_intent)
+
         self._start_workflow_span(info)
         # Keep the invocation on the shared execution trace.
         self._start_invocation_span(info)

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

@@ -0,0 +1,68 @@
+"""Materialize the stable SDK-owned ancestor of a fallback execution trace."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+
+from opentelemetry.context import Context
+from opentelemetry.sdk.trace.sampling import SamplingResult
+from opentelemetry.trace import SpanContext, SpanKind, Tracer
+
+from aws_durable_execution_sdk_python_otel.deterministic_id_generator import (
+    DeterministicIdGenerator,
+)
+from aws_durable_execution_sdk_python_otel.durable_sampling import (
+    DurableSamplingIntent,
+    store_sampling_intent,
+)
+from aws_durable_execution_sdk_python_otel.timestamps import to_otel_timestamp
+
+
+@dataclass(frozen=True)
+class ExecutionRoot:
+    """A zero-duration anchor, reproducible in any invocation of an execution.
+
+    Re-exporting the same anchor permits recovery after a failed flush without
+    checkpointing telemetry state. The anchor carries stable SDK attributes;
+    configured provider resources and processor enrichment remain unchanged.
+    """
+
+    execution_arn: str
+    ancestor: SpanContext
+    start_time: datetime
+
+    def export(
+        self,
+        tracer: Tracer,
+        id_generator: DeterministicIdGenerator,
+        sampling_intent: DurableSamplingIntent,
+    ) -> None:
+        """Export a sampled local ancestor; never replace an external parent."""
+        if self.ancestor.is_remote or not self.ancestor.trace_flags.sampled:
+            return
+
+        # Reuse only the resolved decision without resampling. This same-ID
+        # anchor must not acquire variable sampler attributes or tracestate.
+        # The regular tracer preserves configured resources and processors.
+        root_attributes: dict[str, str | bool] = {
+            "durable.execution.arn": self.execution_arn,
+            "durable.execution.synthetic_root": True,
+        }
+        root_intent = DurableSamplingIntent(
+            SamplingResult(sampling_intent.result.decision)
+        )
+        root_context = store_sampling_intent(Context(), root_intent)
+        timestamp = to_otel_timestamp(self.start_time)
+        with id_generator._use_ids_for_span(
+            trace_id=self.ancestor.trace_id,
+            span_id=self.ancestor.span_id,
+        ):
+            span = tracer.start_span(
+                "DurableExecutionRoot",
+                context=root_context,
+                kind=SpanKind.INTERNAL,
+                attributes=root_attributes,
+                start_time=timestamp,
+            )
+        span.end(end_time=timestamp)

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

@@ -55,6 +55,7 @@
 from aws_durable_execution_sdk_python_otel.durable_parent_span import (
     DurableParentSpan,
 )
+from aws_durable_execution_sdk_python_otel.execution_root import ExecutionRoot
 from aws_durable_execution_sdk_python_otel.execution_trace_context import (
     ExecutionTraceContext,
     canonical_trace_id,
@@ -65,6 +66,7 @@
 )
 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.timestamps import to_otel_timestamp
 
 
 logger = logging.getLogger(__name__)
@@ -82,7 +84,7 @@ def _to_otel_timestamp(dt: datetime.datetime | None) -> int:
     """Convert a datetime to OTel timestamp (nanoseconds since epoch)."""
     if dt is None:
         dt = datetime.datetime.now(datetime.UTC)
-    return int(dt.timestamp() * 1_000_000_000)
+    return to_otel_timestamp(dt)
 
 
 class InvocationOtelPlugin(DurableInstrumentationPlugin):
@@ -582,6 +584,14 @@ def on_invocation_start(self, info: InvocationStartInfo) -> None:
             root_sampled=lambda: is_sampled(sampling_result),
         )
 
+        # Materialize the fallback anchor before user code can suspend or fail.
+        # Repeated invocations retain the same SDK-owned identity and timing.
+        ExecutionRoot(
+            execution_arn=self._execution_arn,
+            ancestor=self._execution_trace_context.execution_ancestor,
+            start_time=info.execution_start_time,
+        ).export(self._tracer, self._id_generator, self._sampling_intent)
+
         self._start_workflow_span(info)
 
         self._start_span(

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

@@ -0,0 +1,8 @@
+"""OpenTelemetry timestamp conversion shared by span producers."""
+
+from datetime import datetime
+
+
+def to_otel_timestamp(value: datetime) -> int:
+    """Convert a datetime to nanoseconds using the SDK's existing precision."""
+    return int(value.timestamp() * 1_000_000_000)