feat(otel): add Workflow span to InvocationOtelPlugin

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

@@ -268,7 +268,7 @@ def _start_invocation_span(self, info: InvocationStartInfo) -> None:
                 "durable.invocation.first": info.is_first_invocation,
             }
         self._invocation_span = self._tracer.start_span(
-            name="invocation",
+            name="Invocation",
             kind=SpanKind.INTERNAL,
             attributes=attributes,
             context=parent_ctx,

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

@@ -40,13 +40,21 @@
 )
 from aws_durable_execution_sdk_python_otel.deterministic_id_generator import (
     DeterministicIdGenerator,
+    derive_workflow_span_id,
     operation_id_to_span_id,
 )
 from aws_durable_execution_sdk_python_otel.log_filter import install_log_filter
+from aws_durable_execution_sdk_python_otel.otel_plugin_config import (
+    DEFAULT_WORKFLOW_SPAN_NAME,
+)
 
 
 logger = logging.getLogger(__name__)
 
+_TERMINAL_INVOCATION_STATUSES = frozenset(
+    {InvocationStatus.SUCCEEDED, InvocationStatus.FAILED}
+)
+
 _SpanAttributes = dict[str, str | bool | int]
 
 
@@ -88,6 +96,7 @@ def __init__(
         context_extractor: ContextExtractor | None = None,
         instrument_name: str = DEFAULT_INSTRUMENT_NAME,
         enrich_logger: bool = True,
+        workflow_span_name: str = DEFAULT_WORKFLOW_SPAN_NAME,
     ) -> None:
         """Initialize the plugin with an OpenTelemetry tracer provider.
 
@@ -101,6 +110,7 @@ def __init__(
         OTel trace context onto every emitted log record.
         """
         self._enrich_logger = enrich_logger
+        self._workflow_span_name = workflow_span_name
         self._context_extractor: ContextExtractor = (
             context_extractor or xray_context_extractor
         )
@@ -132,6 +142,7 @@ def __init__(
         # per invocation status:
         self._execution_arn = ""
         self._extracted_context: Context | None = None
+        self._workflow_span: Span | None = None
         # Maps operation ID (None for root) to the active span.
         self._operation_spans: dict[str | None, Span] = {}
         self._operation_spans_lock = threading.RLock()
@@ -279,6 +290,12 @@ def _start_span(
                     if operation_id
                     else None
                 )
+            # Operation and attempt spans link to the execution-scoped Workflow
+            # span (the invocation span itself, operation_id=None, does not).
+            if self._workflow_span is not None and operation_id is not None:
+                workflow_ctx = self._workflow_span.get_span_context()
+                if workflow_ctx and workflow_ctx.is_valid:
+                    links = [*links, Link(context=workflow_ctx)]
             if parent_span is None:
                 # root span
                 parent_context = self._extracted_context
@@ -329,12 +346,42 @@ def on_invocation_start(self, info: InvocationStartInfo) -> None:
         self._extracted_context = self._context_extractor(info)
         self._id_generator.set_trace_id(self._execution_arn, info.execution_start_time)
 
+        self._start_workflow_span(info)
+
         self._start_span(
             operation_id=None,
-            name="invocation",
+            name="Invocation",
             attributes=self._extract_attributes(info),
         )
 
+    def _start_workflow_span(self, info: InvocationStartInfo) -> None:
+        """Create the deterministic, execution-scoped Workflow root span.
+
+        The Workflow span is a parentless root keyed to a deterministic span ID
+        derived from the execution ARN, so every invocation of the same durable
+        execution contributes to one Workflow span. It is exported once, on a
+        terminal invocation. Operation and attempt spans link to it while
+        remaining parented to the invocation span. It is created unconditionally
+        -- InvocationOtelPlugin has no default/owned tracer-provider distinction,
+        so the span is emitted whether the provider is the ambient (ADOT/global)
+        one or an explicitly supplied one.
+        """
+        if not self._execution_arn:
+            logger.warning("No execution ARN; skipping Workflow span creation")
+            return
+        self._id_generator.set_next_span_id(
+            derive_workflow_span_id(self._execution_arn)
+        )
+        # Empty context => root span with no parent. _to_otel_timestamp falls
+        # back to now() when execution_start_time is None.
+        self._workflow_span = self._tracer.start_span(
+            name=self._workflow_span_name,
+            kind=SpanKind.INTERNAL,
+            attributes={"durable.execution.arn": self._execution_arn},
+            start_time=_to_otel_timestamp(info.execution_start_time),
+            context=Context(),
+        )
+
     def on_invocation_end(self, info: InvocationEndInfo) -> None:
         """Called at the end of each invocation. Ends the invocation span and flushes."""
         logger.debug("Durable invocation ended: %s", info)
@@ -365,9 +412,28 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None:
         # end the invocation span
         self._end_span(None)
 
+        # The Workflow span (execution view) is exported only on a terminal
+        # status; on non-terminal statuses its reference is dropped without
+        # ending it (so it is not exported yet). SUCCEEDED -> OK, FAILED -> ERROR;
+        # RETRY/PENDING are non-terminal and leave it unexported.
+        if self._workflow_span is not None:
+            if info.status in _TERMINAL_INVOCATION_STATUSES:
+                self._workflow_span.set_attribute(
+                    "durable.execution.status",
+                    info.status.value if info.status else "",
+                )
+                if info.status is InvocationStatus.FAILED:
+                    self._workflow_span.set_status(
+                        StatusCode.ERROR, info.error.message if info.error else ""
+                    )
+                elif info.status is InvocationStatus.SUCCEEDED:
+                    self._workflow_span.set_status(StatusCode.OK)
+                self._workflow_span.end()
+
         # Clear all per-invocation state to prevent leaks across warm Lambda reuses
         self._execution_arn = ""
         self._extracted_context = None
+        self._workflow_span = None
         with self._operation_spans_lock:
             self._operation_spans = {}
 

packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py

@@ -126,10 +126,10 @@ def test_workflow_span_is_root_and_invocation_is_its_child():
 
     spans = {s.name: s for s in exporter.get_finished_spans()}
     assert "Workflow" in spans
-    assert "invocation" in spans
+    assert "Invocation" in spans
 
     workflow = spans["Workflow"]
-    invocation = spans["invocation"]
+    invocation = spans["Invocation"]
 
     # Workflow is a root span (no parent) with the deterministic span ID.
     assert workflow.parent is None
@@ -154,7 +154,7 @@ def test_workflow_span_dropped_on_non_terminal_status():
     names = [s.name for s in exporter.get_finished_spans()]
     # Invocation span is always ended/exported; the Workflow span is dropped
     # (not ended) on a non-terminal status, so it must not be exported.
-    assert "invocation" in names
+    assert "Invocation" in names
     assert "Workflow" not in names
 
 
@@ -193,7 +193,7 @@ def test_operation_parented_under_workflow_and_linked_to_invocation():
 
     spans = {s.name: s for s in exporter.get_finished_spans()}
     workflow = spans["Workflow"]
-    invocation = spans["invocation"]
+    invocation = spans["Invocation"]
     operation = spans["wait-for-signal"]
 
     # Parented under the Workflow span (no parentId => Workflow fallback).
@@ -266,8 +266,8 @@ def test_default_mode_creates_invocation_span():
 
     spans = {s.name: s for s in exporter.get_finished_spans()}
     # The invocation span is now created even in default-provider mode.
-    assert "invocation" in spans
-    invocation = spans["invocation"]
+    assert "Invocation" in spans
+    invocation = spans["Invocation"]
     assert invocation.attributes["durable.execution.arn"] == EXECUTION_ARN
     assert invocation.attributes["durable.invocation.first"] is True
 
@@ -285,7 +285,7 @@ def test_default_mode_invocation_span_parented_to_ambient_span():
         otel_context.detach(token)
         ambient.end()
 
-    invocation = {s.name: s for s in exporter.get_finished_spans()}["invocation"]
+    invocation = {s.name: s for s in exporter.get_finished_spans()}["Invocation"]
     assert invocation.parent is not None
     assert invocation.parent.span_id == ambient.get_span_context().span_id
 
@@ -335,7 +335,7 @@ def test_invocation_span_status_kind_and_attributes(status, expected_code):
     plugin.on_invocation_start(_invocation_start_info())
     plugin.on_invocation_end(_invocation_end_info(status=status))
 
-    invocation = {s.name: s for s in exporter.get_finished_spans()}["invocation"]
+    invocation = {s.name: s for s in exporter.get_finished_spans()}["Invocation"]
     assert invocation.kind is trace.SpanKind.INTERNAL
     assert invocation.attributes is not None
     assert invocation.attributes["durable.invocation.status"] == status.value

packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py

@@ -172,7 +172,7 @@ def test_community_layer_full_lifecycle_is_workflow_rooted():
     finished = exporter.get_finished_spans()
     spans = {s.name: s for s in finished}
     workflow = spans["Workflow"]
-    invocation = spans["invocation"]
+    invocation = spans["Invocation"]
     operation = spans[OP_NAME]
     attempt = spans[f"{OP_NAME} attempt 1"]
 
@@ -229,7 +229,7 @@ def test_adot_layer_full_lifecycle_parents_to_ambient_span():
 
     finished = exporter.get_finished_spans()
     spans = {s.name: s for s in finished}
-    invocation = spans["invocation"]
+    invocation = spans["Invocation"]
     operation = spans[OP_NAME]
 
     # Invocation span parents to the ambient ADOT span and carries the first flag.

packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py

@@ -31,6 +31,7 @@
 from opentelemetry.trace import SpanKind, StatusCode
 
 from aws_durable_execution_sdk_python_otel.deterministic_id_generator import (
+    derive_workflow_span_id,
     operation_id_to_span_id,
 )
 from aws_durable_execution_sdk_python_otel.invocation_plugin import InvocationOtelPlugin
@@ -149,12 +150,15 @@ def test_invocation_start_and_end_emit_invocation_span():
     plugin.on_invocation_end(_invocation_end_info())
 
     spans = exporter.get_finished_spans()
-    assert [span.name for span in spans] == ["invocation"]
-    assert spans[0].kind is SpanKind.INTERNAL
-    assert spans[0].attributes["durable.execution.arn"] == EXECUTION_ARN
-    assert spans[0].attributes["durable.invocation.first"] is True
+    spans_by_name = {span.name: span for span in spans}
+    # Terminal invocation also exports the Workflow span.
+    assert set(spans_by_name) == {"Invocation", "Workflow"}
+    invocation = spans_by_name["Invocation"]
+    assert invocation.kind is SpanKind.INTERNAL
+    assert invocation.attributes["durable.execution.arn"] == EXECUTION_ARN
+    assert invocation.attributes["durable.invocation.first"] is True
     assert (
-        spans[0].attributes["durable.invocation.status"]
+        invocation.attributes["durable.invocation.status"]
         == InvocationStatus.SUCCEEDED.value
     )
     assert plugin._get_span(None) is None
@@ -168,8 +172,8 @@ def test_invocation_span_records_subsequent_invocation():
     plugin.on_invocation_end(_invocation_end_info())
 
     spans = exporter.get_finished_spans()
-    assert len(spans) == 1
-    assert spans[0].attributes["durable.invocation.first"] is False
+    invocation = next(s for s in spans if s.name == "Invocation")
+    assert invocation.attributes["durable.invocation.first"] is False
 
 
 @pytest.mark.parametrize(
@@ -192,11 +196,11 @@ def test_invocation_span_status_reflects_execution_status(
     plugin.on_invocation_end(_invocation_end_info(invocation_status))
 
     spans = exporter.get_finished_spans()
-    assert len(spans) == 1
-    attributes = spans[0].attributes
+    invocation = next(s for s in spans if s.name == "Invocation")
+    attributes = invocation.attributes
     assert attributes is not None
     assert attributes["durable.invocation.status"] == invocation_status.value
-    assert spans[0].status.status_code is expected_span_status
+    assert invocation.status.status_code is expected_span_status
 
 
 def test_operation_callbacks_emit_child_span_with_deterministic_span_id():
@@ -249,7 +253,7 @@ def test_operation_callbacks_emit_child_span_with_deterministic_span_id():
     spans_by_name = {span.name: span for span in exporter.get_finished_spans()}
     assert all(span.kind is SpanKind.INTERNAL for span in spans_by_name.values())
     wait_span = spans_by_name["wait-for-signal"]
-    invocation_span = spans_by_name["invocation"]
+    invocation_span = spans_by_name["Invocation"]
     assert wait_span.context.span_id == operation_id_to_span_id(
         EXECUTION_ARN, operation_id
     )
@@ -889,3 +893,102 @@ def test_nested_steps_restore_context_span_across_multiple_iterations():
         )
         # Between each inner step, the child-context span is current.
         assert trace.get_current_span().get_span_context().span_id == context_span_id
+
+
+@pytest.mark.parametrize(
+    ("status", "expected_code"),
+    [
+        (InvocationStatus.SUCCEEDED, StatusCode.OK),
+        (InvocationStatus.FAILED, StatusCode.ERROR),
+    ],
+)
+def test_workflow_span_exported_on_terminal(status, expected_code):
+    """A terminal invocation exports a deterministic Workflow root span."""
+    plugin, exporter = _create_plugin()
+    plugin.on_invocation_start(_invocation_start_info())
+    plugin.on_invocation_end(_invocation_end_info(status))
+
+    workflow = next(s for s in exporter.get_finished_spans() if s.name == "Workflow")
+    # Root span: no parent.
+    assert workflow.parent is None
+    assert workflow.kind is SpanKind.INTERNAL
+    # Deterministic span id derived from the execution ARN.
+    assert workflow.context.span_id == derive_workflow_span_id(EXECUTION_ARN)
+    assert workflow.attributes["durable.execution.arn"] == EXECUTION_ARN
+    assert workflow.attributes["durable.execution.status"] == status.value
+    assert workflow.status.status_code is expected_code
+    # Anchored to the execution start time.
+    assert workflow.start_time == int(START_TIME.timestamp() * 1_000_000_000)
+
+
+@pytest.mark.parametrize("status", [InvocationStatus.PENDING, InvocationStatus.RETRY])
+def test_workflow_span_not_exported_on_non_terminal(status):
+    """Non-terminal invocations do not export (end) the Workflow span."""
+    plugin, exporter = _create_plugin()
+    plugin.on_invocation_start(_invocation_start_info())
+    plugin.on_invocation_end(_invocation_end_info(status))
+
+    names = [s.name for s in exporter.get_finished_spans()]
+    assert "Workflow" not in names
+    assert "Invocation" in names
+
+
+def test_operation_span_links_to_workflow_span():
+    """Operation spans link to the Workflow span while parented to invocation."""
+    plugin, exporter = _create_plugin()
+    plugin.on_invocation_start(_invocation_start_info())
+    operation_id = "wait-1"
+    plugin.on_operation_start(
+        OperationStartInfo(
+            operation_id=operation_id,
+            operation_type=OperationType.WAIT,
+            sub_type=OperationSubType.WAIT,
+            name="wait-for-signal",
+            parent_id=None,
+            start_time=START_TIME,
+            is_replayed=False,
+            status=OperationStatus.STARTED,
+        )
+    )
+    plugin.on_operation_end(
+        OperationEndInfo(
+            operation_id=operation_id,
+            operation_type=OperationType.WAIT,
+            sub_type=OperationSubType.WAIT,
+            name="wait-for-signal",
+            parent_id=None,
+            start_time=START_TIME,
+            is_replayed=False,
+            status=OperationStatus.SUCCEEDED,
+            end_time=END_TIME,
+            error=None,
+        )
+    )
+    plugin.on_invocation_end(_invocation_end_info())
+
+    spans_by_name = {s.name: s for s in exporter.get_finished_spans()}
+    op_span = spans_by_name["wait-for-signal"]
+    workflow_span_id = derive_workflow_span_id(EXECUTION_ARN)
+    linked_span_ids = {link.context.span_id for link in op_span.links}
+    assert workflow_span_id in linked_span_ids
+    # Still parented to the invocation span (not the Workflow span).
+    assert op_span.parent is not None
+    assert op_span.parent.span_id == spans_by_name["Invocation"].context.span_id
+
+
+def test_workflow_span_name_is_configurable():
+    """The Workflow span name can be overridden via constructor kwarg."""
+    exporter = InMemorySpanExporter()
+    trace_provider = TracerProvider()
+    trace_provider.add_span_processor(SimpleSpanProcessor(exporter))
+    plugin = InvocationOtelPlugin(
+        trace_provider=trace_provider,
+        context_extractor=lambda _: Context(),
+        workflow_span_name="MyExecution",
+    )
+    plugin.on_invocation_start(_invocation_start_info())
+    plugin.on_invocation_end(_invocation_end_info())
+
+    names = [s.name for s in exporter.get_finished_spans()]
+    assert "MyExecution" in names
+    assert "Workflow" not in names

packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py

@@ -155,7 +155,7 @@ def _run_step_lifecycle(plugin: InvocationOtelPlugin) -> None:
 
 def _assert_hierarchy(exporter: InMemorySpanExporter) -> None:
     spans = {s.name: s for s in exporter.get_finished_spans()}
-    invocation = spans["invocation"]
+    invocation = spans["Invocation"]
     operation = spans[OP_NAME]
     attempt = spans[f"{OP_NAME} attempt 1"]