feat: HYBIM-518 Mark root GenAI span with gen_ai.conversation_root + promote root to AgentInvocation (#236)

Change:
1. gen_ai.conversation_root attribute
Adds a new gen_ai.conversation_root = true boolean span attribute that explicitly marks the invocation-level root GenAI span in a trace. This gives the backend a reliable way to find the "one span" that represents the entire AI invocation — carrying the user's input and the final assistant output — without relying on topology traversal.

The attribute is independent of gen_ai.conversation.id: it marks the per-trace entry point for both multi-turn conversations (where conversation.id groups turns) and single invocations (where no conversation exists).

2. Root span promoted to AgentInvocation (was Workflow)
Root spans now default to AgentInvocation (invoke_agent) instead of Workflow (invoke_workflow). This better reflects that the top-level entity in a LangGraph/CrewAI trace is an agent invocation.

Workflow root spans are still available via two opt-in mechanisms:

Environment variable: OTEL_INSTRUMENTATION_GENAI_ROOT_SPAN_AS_WORKFLOW=true
LangGraph config metadata: config = {"metadata": {"workflow_name": "My Workflow"}, ...}
When workflow_name is provided in metadata, it also becomes the span name (e.g. workflow My Workflow instead of workflow LangGraph).

The root agent span also includes improved output capture: when the LangGraph state has no AIMessages (common pattern — nodes update structured state fields), the agent's output falls back to serializing non-message state fields (triage_result, confidence_score, etc.) as a JSON summary.

3. Resume detection fix
Fixed a timing bug where gen_ai.command="resume" was set on the Python entity AFTER start_agent() had already created the OTel span (so the attribute never appeared). Now set in the attrs dict BEFORE _start_agent_invocation().

4. Renamed gen_ai.workflow.command → gen_ai.command
The command attribute (used for resume detection with value "resume") was renamed to be entity-agnostic since it now applies to both Workflow and AgentInvocation root spans.

5. Interrupt status bubbles up to root span
When a child entity (step, tool, LLM invocation) encounters a GraphInterrupt or NodeInterrupt, the root span now receives gen_ai.finish_reason="interrupted" attribute. This ensures the root span clearly reflects that the conversation was interrupted, even if the interrupt occurred deeper in the call hierarchy.

Author: zhirafovod

instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/instrumentation.py

@@ -6,6 +6,7 @@
 
 import json
 import logging
+from contextvars import ContextVar
 from typing import Any, Collection, Optional
 
 from wrapt import wrap_function_wrapper
@@ -16,7 +17,6 @@
     get_telemetry_handler,
 )
 from opentelemetry.util.genai.types import (
-    Workflow,
     AgentInvocation,
     Step,
     ToolCall,
@@ -30,6 +30,11 @@
 
 # Global handler instance (singleton)
 _handler: Optional[TelemetryHandler] = None
+# Active root entity (set during kickoff, used by child wrappers).
+# ContextVar ensures thread safety for concurrent kickoffs.
+_active_root_entity: ContextVar[Optional[Any]] = ContextVar(
+    "_active_root_entity", default=None
+)
 
 _logger = logging.getLogger(__name__)
 
@@ -182,44 +187,45 @@ def _safe_unwrap(module: str, name: str):
 
 def _wrap_crew_kickoff(wrapped, instance, args, kwargs):
     """
-    Wrap Crew.kickoff to create a Workflow span.
+    Wrap Crew.kickoff to create a root span.
+
+    By default creates an AgentInvocation (agent type) at root.
+    When OTEL_INSTRUMENTATION_GENAI_ROOT_SPAN_AS_WORKFLOW is truthy,
+    creates a Workflow instead.
 
-    Maps to: Workflow type from splunk-otel-util-genai
+    Uses handler.create_and_start_root() to centralize the Workflow vs
+    AgentInvocation decision.
     """
     try:
         handler = _handler
-
-        # Create workflow invocation
-        workflow = Workflow(
-            name=getattr(instance, "name", None) or "CrewAI Workflow",
-            workflow_type="crewai.crew",
-            framework="crewai",
-            system="crewai",
-        )
+        crew_name = getattr(instance, "name", None) or "CrewAI Workflow"
 
         inputs = kwargs.get("inputs")
         if inputs is None and args:
             inputs = args[0]
-        if inputs is not None:
-            workflow.input_messages = _make_input_message(inputs)
 
-        # Start the workflow
-        handler.start_workflow(workflow)
+        root_entity = handler.create_and_start_root(
+            crew_name,
+            workflow_type="crewai.crew",
+            framework="crewai",
+            system="crewai",
+            input_messages=_make_input_message(inputs) if inputs else None,
+        )
     except Exception:
         # If instrumentation setup fails, just run the original function
         return wrapped(*args, **kwargs)
 
+    token = _active_root_entity.set(root_entity)
     try:
         result = wrapped(*args, **kwargs)
 
         # Capture result information
         try:
             if result:
                 if hasattr(result, "raw"):
-                    workflow.output_messages = _make_output_message(result.raw)
+                    root_entity.output_messages = _make_output_message(result.raw)
 
-            # Stop the workflow successfully
-            handler.stop_workflow(workflow)
+            handler.finish(root_entity)
         except Exception:
             # Ignore instrumentation errors on success path
             pass
@@ -228,10 +234,12 @@ def _wrap_crew_kickoff(wrapped, instance, args, kwargs):
     except Exception as exc:
         # Wrapped function failed - record error and end span
         try:
-            handler.fail(workflow, Error(message=str(exc), type=type(exc)))
+            handler.fail(root_entity, Error(message=str(exc), type=type(exc)))
         except Exception:
             pass
         raise
+    finally:
+        _active_root_entity.reset(token)
 
 
 def _wrap_agent_execute_task(wrapped, instance, args, kwargs):
@@ -264,6 +272,11 @@ def _wrap_agent_execute_task(wrapped, instance, args, kwargs):
                 messages["expected_output"] = expected_output
             agent_invocation.input_messages = _make_input_message(messages)
 
+        # Communicate parent hierarchy for conversation root detection.
+        root = _active_root_entity.get()
+        if root and getattr(root, "span", None):
+            agent_invocation.parent_span = root.span
+
         # Start the agent invocation
         handler.start_agent(agent_invocation)
     except Exception:

instrumentation-genai/opentelemetry-instrumentation-crewai/tests/conftest.py

@@ -98,6 +98,70 @@ def __init__(self):
         self.stopped_tool_calls = []
         self.failed_entities = []
 
+    def _should_use_workflow_root(self, force_workflow=False, workflow_name=None):
+        """Check if root should be workflow based on env var."""
+        if force_workflow or workflow_name:
+            return True
+        val = os.environ.get("OTEL_INSTRUMENTATION_GENAI_ROOT_SPAN_AS_WORKFLOW", "")
+        return val.strip().lower() in {"1", "true", "yes", "on"}
+
+    def create_and_start_root(
+        self,
+        name,
+        *,
+        force_workflow=False,
+        workflow_name=None,
+        workflow_type=None,
+        framework=None,
+        system=None,
+        input_messages=None,
+        conversation_id=None,
+        attributes=None,
+    ):
+        """Create and start a root entity (Workflow or AgentInvocation)."""
+        from opentelemetry.util.genai.types import Workflow, AgentInvocation
+
+        use_workflow = self._should_use_workflow_root(force_workflow, workflow_name)
+        attrs = attributes or {}
+
+        if use_workflow:
+            wf_name = workflow_name or name
+            entity = Workflow(
+                name=wf_name,
+                workflow_type=workflow_type,
+                framework=framework,
+                system=system,
+                attributes=attrs,
+            )
+            if input_messages:
+                entity.input_messages = input_messages
+            if conversation_id:
+                entity.conversation_id = conversation_id
+            self.start_workflow(entity)
+        else:
+            entity = AgentInvocation(
+                name=name,
+                framework=framework,
+                system=system,
+                attributes=attrs,
+            )
+            entity.agent_name = name
+            if input_messages:
+                entity.input_messages = input_messages
+            if conversation_id:
+                entity.conversation_id = conversation_id
+            self.start_agent(entity)
+
+        return entity
+
+    def finish(self, entity):
+        """Finish any entity (generic stop dispatcher)."""
+        from opentelemetry.util.genai.types import Workflow
+
+        if isinstance(entity, Workflow):
+            return self.stop_workflow(entity)
+        return self.stop_agent(entity)
+
     def start_workflow(self, workflow):
         self.started_workflows.append(workflow)
         return workflow