fix(otel): fix cross language otel plugin differences

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

@@ -46,6 +46,7 @@
     Link,
     Span,
     SpanContext,
+    SpanKind,
     StatusCode,
     Tracer,
 )
@@ -141,7 +142,6 @@ def __init__(self, config: OtelPluginConfig | None = None) -> None:
         # Per-invocation state.
         self._execution_arn = ""
         self._extracted_context: Context | None = None
-        self._saved_invocation_context: Context | None = None
         self._workflow_span: Span | None = None
         self._invocation_span: Span | None = None
         self._operation_spans: dict[str, Span] = {}
@@ -188,12 +188,7 @@ def get_current_span_context(self) -> SpanContext | None:
     # Links
     # ------------------------------------------------------------------
     def _build_invocation_links(self) -> list[Link]:
-        """Link operation/attempt spans to the invocation span."""
-        if self._use_default and self._saved_invocation_context is not None:
-            span = trace.get_current_span(self._saved_invocation_context)
-            ctx = span.get_span_context()
-            if ctx and ctx.is_valid:
-                return [Link(context=ctx)]
+        """Link operation/attempt spans to the durable invocation span."""
         if self._invocation_span is not None:
             ctx = self._invocation_span.get_span_context()
             if ctx and ctx.is_valid:
@@ -217,10 +212,6 @@ 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)
 
-        # Capture the ambient context for link-building in default mode.
-        if self._use_default:
-            self._saved_invocation_context = otel_context.get_current()
-
         self._start_workflow_span(info)
         # Create the Invocation span in both modes. In default-provider mode it
         # is parented to the ambient Lambda invocation span.
@@ -259,10 +250,11 @@ def _start_invocation_span(self, info: InvocationStartInfo) -> None:
         if self._use_default:
             # Default-provider mode: parent the Invocation span to the ambient
             # Lambda invocation span (from the ADOT layer or other
-            # auto-instrumentation) captured at invocation start.
-            # Lambda semantic attributes belong to that ambient span, so carry
-            # only durable correlation attributes here.
-            parent_ctx = self._saved_invocation_context or otel_context.get_current()
+            # auto-instrumentation), which is still the active context here (the
+            # Workflow span is created with an empty context and not yet
+            # attached). Lambda semantic attributes belong to that ambient span,
+            # so carry only durable correlation attributes here.
+            parent_ctx = otel_context.get_current()
             attributes = {
                 "durable.execution.arn": self._execution_arn,
                 "durable.invocation.first": info.is_first_invocation,
@@ -275,6 +267,7 @@ def _start_invocation_span(self, info: InvocationStartInfo) -> None:
             )
             attributes = {
                 "durable.execution.arn": self._execution_arn,
+                "durable.invocation.first": info.is_first_invocation,
                 "faas.coldstart": self._is_cold_start,
                 "cloud.provider": "aws",
                 "cloud.platform": "aws_lambda",
@@ -283,6 +276,7 @@ def _start_invocation_span(self, info: InvocationStartInfo) -> None:
                 attributes["faas.invocation_id"] = info.request_id
         self._invocation_span = self._tracer.start_span(
             name="invocation",
+            kind=SpanKind.INTERNAL,
             attributes=attributes,
             context=parent_ctx,
         )
@@ -296,23 +290,38 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None:
         # so they are not exported as if completed. _reset_state
         # clears the span map below.
 
-        # End the invocation span regardless of terminal status.
+        # End the invocation span regardless of terminal status. Record the
+        # invocation status and map it to a span status: SUCCEEDED/PENDING are OK
+        # (this invocation did its work, whether it completed or suspended),
+        # RETRY/FAILED are ERROR (this invocation failed).
         if self._invocation_span is not None:
+            self._invocation_span.set_attribute(
+                "durable.invocation.status",
+                info.status.value if info.status else "",
+            )
+            if info.status in (InvocationStatus.SUCCEEDED, InvocationStatus.PENDING):
+                self._invocation_span.set_status(StatusCode.OK)
+            elif info.status in (InvocationStatus.RETRY, InvocationStatus.FAILED):
+                self._invocation_span.set_status(
+                    StatusCode.ERROR, info.error.message if info.error else ""
+                )
             self._invocation_span.end()
 
-        # The Workflow span is exported only on a terminal status; otherwise its
-        # reference is dropped without ending it.
+        # The Workflow span (execution view) is exported only on a terminal
+        # status; otherwise its reference is dropped without ending it. Its span
+        # status reflects the execution outcome: SUCCEEDED -> OK, FAILED -> ERROR
+        # (RETRY/PENDING are non-terminal and never reach here -> UNSET).
         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.error:
+                if info.status is InvocationStatus.FAILED:
                     self._workflow_span.set_status(
-                        StatusCode.ERROR, info.error.message or ""
+                        StatusCode.ERROR, info.error.message if info.error else ""
                     )
-                else:
+                elif info.status is InvocationStatus.SUCCEEDED:
                     self._workflow_span.set_status(StatusCode.OK)
                 self._workflow_span.end()
 
@@ -327,7 +336,6 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None:
     def _reset_state(self) -> None:
         self._execution_arn = ""
         self._extracted_context = None
-        self._saved_invocation_context = None
         self._workflow_span = None
         self._invocation_span = None
         with self._lock:

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

@@ -350,12 +350,12 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None:
             invocation_span.set_attribute(
                 "durable.invocation.status", info.status.value
             )
-            if info.status is InvocationStatus.FAILED:
+            if info.status in (InvocationStatus.SUCCEEDED, InvocationStatus.PENDING):
+                invocation_span.set_status(StatusCode.OK)
+            elif info.status in (InvocationStatus.RETRY, InvocationStatus.FAILED):
                 invocation_span.set_status(
                     StatusCode.ERROR, info.error.message if info.error else ""
                 )
-            elif info.status is InvocationStatus.SUCCEEDED:
-                invocation_span.set_status(StatusCode.OK)
 
         # end the invocation span
         self._end_span(None)

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

@@ -317,3 +317,27 @@ def test_open_operation_span_not_exported_at_invocation_end():
     exported = {s.name for s in exporter.get_finished_spans()}
     # The open operation span is NOT exported (never ended).
     assert "wait-for-signal" not in exported
+
+
+@pytest.mark.parametrize(
+    ("status", "expected_code"),
+    [
+        (InvocationStatus.PENDING, trace.StatusCode.OK),
+        (InvocationStatus.SUCCEEDED, trace.StatusCode.OK),
+        (InvocationStatus.RETRY, trace.StatusCode.ERROR),
+        (InvocationStatus.FAILED, trace.StatusCode.ERROR),
+    ],
+)
+def test_invocation_span_status_kind_and_attributes(status, expected_code):
+    """Invocation span is INTERNAL, carries status/first; SUCCEEDED/PENDING are
+    OK and RETRY/FAILED are ERROR."""
+    plugin, exporter = _create_plugin()
+    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"]
+    assert invocation.kind is trace.SpanKind.INTERNAL
+    assert invocation.attributes is not None
+    assert invocation.attributes["durable.invocation.status"] == status.value
+    assert invocation.attributes["durable.invocation.first"] is True
+    assert invocation.status.status_code is expected_code

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

@@ -237,9 +237,10 @@ def test_adot_layer_full_lifecycle_parents_to_ambient_span():
     assert invocation.parent.span_id == ambient.get_span_context().span_id
     assert invocation.attributes["durable.invocation.first"] is True
 
-    # Operation span still uses the deterministic id and links to the ambient span.
+    # Operation span still uses the deterministic id and links to the durable
+    # invocation span (which is itself parented to the ambient ADOT span).
     assert operation.context.span_id == operation_id_to_span_id(EXECUTION_ARN, OP_ID)
-    assert ambient.get_span_context().span_id in {
+    assert invocation.context.span_id in {
         link.context.span_id for link in operation.links
     }
 

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

@@ -174,8 +174,8 @@ def test_invocation_span_records_subsequent_invocation():
 @pytest.mark.parametrize(
     ("invocation_status", "expected_span_status"),
     [
-        (InvocationStatus.PENDING, StatusCode.UNSET),
-        (InvocationStatus.RETRY, StatusCode.UNSET),
+        (InvocationStatus.PENDING, StatusCode.OK),
+        (InvocationStatus.RETRY, StatusCode.ERROR),
         (InvocationStatus.SUCCEEDED, StatusCode.OK),
         (InvocationStatus.FAILED, StatusCode.ERROR),
     ],