fix(otel): export sampled fallback roots early

.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@a66037abbbfa55fde97f714e30f0bc262edefd63
+    uses: aws/aws-durable-execution-conformance-tests/.github/workflows/opentelemetry-orchestrator.yml@f18bd0b5f28c5c90e288d0fb8bca08a849b51863
     with:
       language: python
       runs_on: codebuild-github-actions-runner-${{ github.run_id }}-${{ github.run_attempt }}

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

@@ -278,8 +278,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:
@@ -308,6 +308,32 @@ 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 exports `DurableExecutionRoot` at invocation
+start and flushes through the normal invocation-end lifecycle, including when
+it returns `PENDING` or `RETRY`. A suspended execution therefore already has an
+exported ancestor if it is later stopped or times out without resuming.
+
+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 uses the same configured tracer, provider resource, processors and
+resolved sampling result as other durable spans. SDK-added IDs, timestamps and
+attributes are stable across recovery exports; provider resources, sampler
+metadata and custom processor enrichment can vary between execution environments.
+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
@@ -401,6 +427,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/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,
@@ -489,6 +490,15 @@ 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.
+        if isinstance(self._tracer, SdkTracer):
+            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,75 @@
+"""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,
+)
+
+
+@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. Resources and sampler metadata remain owned
+    by the configured OpenTelemetry provider; SDK-added fields remain stable.
+    """
+
+    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 the invocation's resolved decision and sampler metadata without
+        # resampling. An empty parent context makes this an actual root, while
+        # 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,
+        }
+        # Sampler attributes normally override span attributes. Give this root
+        # its own intent so SDK-owned identity reaches processors intact, while
+        # retaining all other metadata and the invocation's original result.
+        result = sampling_intent.result
+        root_intent = DurableSamplingIntent(
+            SamplingResult(
+                result.decision,
+                attributes={**dict(result.attributes or {}), **root_attributes},
+                trace_state=result.trace_state,
+            )
+        )
+        root_context = store_sampling_intent(Context(), root_intent)
+        timestamp = int(self.start_time.timestamp() * 1_000_000_000)
+        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,
@@ -585,6 +586,15 @@ 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.
+        if isinstance(self._tracer, SdkTracer):
+            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/tests/e2e/test_fallback_root_lifecycle.py

@@ -0,0 +1,114 @@
+"""Fallback anchor coverage through the decorator and local durable runner."""
+
+import inspect
+from typing import Any
+
+import pytest
+from aws_durable_execution_sdk_python import DurableContext
+from aws_durable_execution_sdk_python.config import Duration
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.plugin import (
+    DurableInstrumentationPlugin,
+    InvocationEndInfo,
+    InvocationStatus,
+)
+from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner
+from opentelemetry import context
+from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+
+from aws_durable_execution_sdk_python_otel import (
+    ExecutionOtelPlugin,
+    InvocationOtelPlugin,
+    OtelPluginConfig,
+)
+
+
+class _CaptureInvocationEnd(DurableInstrumentationPlugin):
+    def __init__(self, exporter: InMemorySpanExporter) -> None:
+        self.exporter = exporter
+        self.snapshots: list[tuple[InvocationStatus, tuple[ReadableSpan, ...]]] = []
+
+    def on_invocation_end(self, info: InvocationEndInfo) -> None:
+        self.snapshots.append((info.status, tuple(self.exporter.get_finished_spans())))
+
+
+@pytest.mark.parametrize("plugin_type", [ExecutionOtelPlugin, InvocationOtelPlugin])
+@pytest.mark.parametrize("outcome", ["success", "failure", "timeout"])
+def test_fallback_anchor_survives_suspension(
+    plugin_type: type[ExecutionOtelPlugin] | type[InvocationOtelPlugin],
+    outcome: str,
+) -> None:
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(
+        BatchSpanProcessor(exporter, schedule_delay_millis=60000)
+    )
+    plugin = plugin_type(
+        OtelPluginConfig(
+            tracer_provider=provider,
+            context_extractor=lambda _: None,
+            enrich_logger=False,
+        )
+    )
+    observer = _CaptureInvocationEnd(exporter)
+    completed_steps: list[str] = []
+    before = context.get_current()
+
+    def step(_step_context: Any) -> str:
+        completed_steps.append("step")
+        return "stored"
+
+    def handler(_event: Any, durable: DurableContext) -> str:
+        value = durable.step(step, name="completed-step")
+        durable.wait(
+            Duration.from_seconds(60 if outcome == "timeout" else 1), name="wait"
+        )
+        if outcome == "failure":
+            raise ValueError("failed after resume")
+        return value
+
+    wrapped = durable_execution(handler, plugins=[plugin, observer])
+    try:
+        # Published runners use real time; newer workspace runners default to
+        # skipping durable waits. Exercise timeout without advancing that wait.
+        runner_options: dict[str, Any] = {"handler": wrapped}
+        if "skip_time" in inspect.signature(DurableFunctionTestRunner).parameters:
+            runner_options["skip_time"] = outcome != "timeout"
+        with DurableFunctionTestRunner(**runner_options) as runner:
+            arn = runner.run_async(
+                input="{}", timeout=3 if outcome == "timeout" else 15
+            )
+            result = runner.wait_for_result(arn, timeout=10)
+        assert completed_steps == ["step"]
+        assert observer.snapshots[0][0] is InvocationStatus.PENDING
+        first_spans = observer.snapshots[0][1]
+        first_roots = [s for s in first_spans if s.name == "DurableExecutionRoot"]
+        assert len(first_roots) == 1
+        assert not any(s.name == "Workflow" for s in first_spans)
+        roots = [
+            s for s in exporter.get_finished_spans() if s.name == "DurableExecutionRoot"
+        ]
+        assert all(s.to_json() == first_roots[0].to_json() for s in roots)
+        if outcome == "timeout":
+            assert result.status.value == "FAILED"
+            assert result.error is not None
+            assert "timed out" in (result.error.message or "")
+            assert len(observer.snapshots) == 1
+            assert len(roots) == 1
+            assert not any(s.name == "Workflow" for s in exporter.get_finished_spans())
+        else:
+            assert result.status.value == (
+                "FAILED" if outcome == "failure" else "SUCCEEDED"
+            )
+            workflows = [
+                s for s in exporter.get_finished_spans() if s.name == "Workflow"
+            ]
+            assert len(workflows) == 1
+            assert workflows[0].parent is not None
+            assert first_roots[0].context is not None
+            assert workflows[0].parent.span_id == first_roots[0].context.span_id
+        assert context.get_current() == before
+    finally:
+        provider.shutdown()