feat(telemetry): emit entrypoint invoke_workflow span under schema v2

When ADK_TELEMETRY_SCHEMA_VERSION_OPT_IN resolves to 2, replace the legacy
top-level 'invocation' span with an entrypoint 'invoke_workflow {entrypoint}'
span (entrypoint = root agent or root node name), set is_entrypoint, and emit
the gen_ai.invoke_workflow.duration metric alongside it. Entrypoint status is
tracked via a contextvar so nested/agent-as-tool workflows report
is_entrypoint=false and the span is not double-emitted when the ADK entrypoint
is itself a workflow. Schema v1 is unchanged.

Co-authored-by: Max Ind <maxind@google.com>
PiperOrigin-RevId: 939850650

Author: RKest

src/google/adk/runners.py

@@ -56,16 +56,22 @@
 from .sessions.base_session_service import BaseSessionService
 from .sessions.base_session_service import GetSessionConfig
 from .sessions.session import Session
+from .telemetry import _instrumentation
 from .telemetry.tracing import tracer
 from .tools.base_toolset import BaseToolset
 from .utils._debug_output import print_event
 
 if TYPE_CHECKING:
   from .apps.app import App
   from .apps.app import ResumabilityConfig
+  from .workflow._base_node import BaseNode
 
 logger = logging.getLogger('google_adk.' + __name__)
 
+# Silence unused warning.
+# tracer is imported for backwards compatibility, to avoid breaking change in the API.
+_ = tracer
+
 
 def _find_active_task_isolation_scope(session) -> Optional[str]:
   """Walk session backwards; find the active paused task agent's scope.
@@ -454,7 +460,9 @@ async def _run_node_async(
     Events flow through ic._event_queue via NodeRunner.
     """
 
-    with tracer.start_as_current_span('invocation'):
+    with _instrumentation.record_invocation(
+        entrypoint_node=node or self.agent, conversation_id=session_id
+    ):
       # 1. Setup
       session = await self._get_or_create_session(
           user_id=user_id, session_id=session_id
@@ -1040,7 +1048,9 @@ async def _run_with_trace(
         new_message: Optional[types.Content] = None,
         invocation_id: Optional[str] = None,
     ) -> AsyncGenerator[Event, None]:
-      with tracer.start_as_current_span('invocation'):
+      with _instrumentation.record_invocation(
+          entrypoint_node=self.agent, conversation_id=session_id
+      ):
         session = await self._get_or_create_session(
             user_id=user_id,
             session_id=session_id,

src/google/adk/telemetry/_instrumentation.py

@@ -20,6 +20,7 @@
 import sys
 import time
 from typing import AsyncIterator
+from typing import Iterator
 from typing import TYPE_CHECKING
 
 from opentelemetry import trace
@@ -28,13 +29,16 @@
 from . import _metrics
 from . import tracing
 from ..events import event as event_lib
+from ._schema_version import resolve_schema_version
+from ._schema_version import SCHEMA_VERSION_SEMCONV_ALIGNED
 
 if TYPE_CHECKING:
   from ..agents.base_agent import BaseAgent
   from ..agents.invocation_context import InvocationContext
   from ..models.llm_request import LlmRequest
   from ..models.llm_response import LlmResponse
   from ..tools.base_tool import BaseTool
+  from ..workflow._base_node import BaseNode
 
 logger = logging.getLogger("google_adk." + __name__)
 
@@ -69,6 +73,46 @@ def _get_elapsed_s(
   return time.monotonic() - fallback_start
 
 
+@contextlib.contextmanager
+def record_invocation(
+    entrypoint_node: BaseNode | None,
+    conversation_id: str,
+) -> Iterator[None]:
+  """Top-level invocation span for a runner invocation.
+
+  Schema v1 emits the legacy ``invocation`` span. Schema v2 replaces it with an
+  entrypoint ``invoke_workflow {entrypoint}`` span (entrypoint = root agent or
+  root node name), with ``is_entrypoint`` set and a
+  ``gen_ai.invoke_workflow.duration`` metric -- unless the entrypoint is itself
+  a workflow, in which case its own node span is the entrypoint
+  ``invoke_workflow`` span and we avoid double-emitting it here.
+
+  Args:
+    entrypoint_node: The runner's root agent/node.
+    conversation_id: Session/conversation id (stamped on the v2 span).
+
+  Yields:
+    Nothing; the span (if any) is active for the duration of the block.
+  """
+  if resolve_schema_version() < SCHEMA_VERSION_SEMCONV_ALIGNED:
+    with tracing.tracer.start_as_current_span("invocation"):
+      yield
+    return
+
+  from . import node_tracing
+  from ..workflow._workflow import Workflow
+
+  if isinstance(entrypoint_node, Workflow):
+    # The workflow's own node span is the entrypoint `invoke_workflow` span.
+    yield
+    return
+
+  with node_tracing._use_invoke_workflow_span(
+      entrypoint_node.name, conversation_id
+  ):
+    yield
+
+
 @dataclasses.dataclass
 class TelemetryContext:
   """Stores all telemetry related state."""

src/google/adk/telemetry/_metrics.py

@@ -61,6 +61,11 @@
         409.6,
     ],
 )
+_workflow_invocation_duration = meter.create_histogram(
+    "gen_ai.invoke_workflow.duration",
+    unit="s",
+    description="Duration of workflow invocations.",
+)
 _tool_execution_duration = meter.create_histogram(
     "gen_ai.tool.execution.duration",
     unit="s",
@@ -160,6 +165,23 @@ def record_agent_invocation_duration(
   _agent_invocation_duration.record(elapsed_s, attributes=attrs)
 
 
+def record_workflow_invocation_duration(
+    workflow_name: str,
+    elapsed_s: float,
+    is_entrypoint: bool,
+    error: Exception | None = None,
+):
+  """Records the duration of a workflow (entrypoint) invocation."""
+  attrs = {
+      gen_ai_attributes.GEN_AI_OPERATION_NAME: "invoke_workflow",
+      "gen_ai.workflow.name": workflow_name,
+      "gen_ai.workflow.is_entrypoint": is_entrypoint,
+  }
+  if error is not None:
+    attrs[error_attributes.ERROR_TYPE] = type(error).__name__
+  _workflow_invocation_duration.record(elapsed_s, attributes=attrs)
+
+
 def record_agent_request_size(
     agent_name: str, user_content: types.Content | None
 ):

src/google/adk/telemetry/node_tracing.py

@@ -15,24 +15,41 @@
 from __future__ import annotations
 
 from collections.abc import AsyncIterator
+from collections.abc import Iterator
 from contextlib import asynccontextmanager
+from contextlib import contextmanager
 from dataclasses import dataclass
 from dataclasses import field
+import time
 from typing import TYPE_CHECKING
 
 from opentelemetry import context as context_api
 from opentelemetry.semconv._incubating.attributes.gen_ai_attributes import GEN_AI_CONVERSATION_ID
 from opentelemetry.semconv._incubating.attributes.gen_ai_attributes import GEN_AI_OPERATION_NAME
-from opentelemetry.util.types import Attributes
+from opentelemetry.trace import Span
 
 from ..agents.context import Context
 from ..workflow._base_node import BaseNode
 from .tracing import tracer
 
 if TYPE_CHECKING:
+  from ..agents.base_agent import BaseAgent
   from ..events.event import Event
   from ..workflow._workflow import Workflow
 
+# Span/metric attribute flagging whether an `invoke_workflow` span is the
+# entrypoint (first workflow) of its invocation.
+GEN_AI_WORKFLOW_IS_ENTRYPOINT = "gen_ai.workflow.is_entrypoint"
+
+# OTel-context key recording that an entrypoint workflow is already active. It
+# rides along the otel_context propagated to child nodes, so only the first
+# workflow invoked within an invocation is marked as the entrypoint -- nested
+# workflows (incl. agents-as-tool that spin up their own runner) see the key
+# already set and report is_entrypoint=false.
+_ENTRYPOINT_WORKFLOW_KEY = context_api.create_key(
+    "adk-entrypoint-workflow-active"
+)
+
 
 @dataclass(frozen=True)
 class TelemetryContext:
@@ -49,12 +66,6 @@ def add_event(self, event: Event) -> None:
     self._associated_event_ids.append(event.id)
 
 
-@dataclass
-class _SpanMetadata:
-  name: str
-  attributes: Attributes
-
-
 @asynccontextmanager
 async def start_as_current_node_span(
     context: Context, node: BaseNode
@@ -83,64 +94,117 @@ async def start_as_current_node_span(
     Context with the started span.
   """
 
-  span_metadata = _span_metadata(context, node)
-  if span_metadata is None:
-    token = context_api.attach(context.telemetry_context.otel_context)
-    try:
-      yield TelemetryContext(
-          otel_context=context.telemetry_context.otel_context
-      )
-    finally:
-      context_api.detach(token)
-    return
-
-  with tracer.start_as_current_span(
-      span_metadata.name,
-      attributes=span_metadata.attributes,
-      context=context.telemetry_context.otel_context,
-  ) as span:
-    telemetry_context = TelemetryContext(otel_context=context_api.get_current())
-    yield telemetry_context
-
-    if span.is_recording() and len(telemetry_context._associated_event_ids) > 0:
-      span.set_attribute(
-          "gcp.vertex.agent.associated_event_ids",
-          telemetry_context._associated_event_ids,
-      )
-
-
-def _span_metadata(context: Context, node: BaseNode) -> _SpanMetadata | None:
   from ..agents.base_agent import BaseAgent
   from ..workflow._workflow import Workflow
 
   if isinstance(node, BaseAgent):
-    return None
+    with _invoke_agent_span(context, node) as tel_ctx:
+      yield tel_ctx
   elif isinstance(node, Workflow):
-    return _workflow_span_metadata(context, node)
+    with _invoke_workflow_span(context, node) as tel_ctx:
+      yield tel_ctx
   else:
-    return _default_node_span_metadata(context, node)
+    with _invoke_node_span(context, node) as tel_ctx:
+      yield tel_ctx
+
+
+@contextmanager
+def _invoke_agent_span(
+    context: Context, agent: BaseAgent
+) -> Iterator[TelemetryContext]:
+  """Passes through an agent node; agents emit their own `invoke_agent` span."""
+  del agent
+  token = context_api.attach(context.telemetry_context.otel_context)
+  try:
+    yield TelemetryContext(otel_context=context.telemetry_context.otel_context)
+  finally:
+    context_api.detach(token)
 
 
-def _workflow_span_metadata(
+@contextmanager
+def _invoke_workflow_span(
     context: Context, workflow: Workflow
-) -> _SpanMetadata:
-  return _SpanMetadata(
-      name=f"invoke_workflow {workflow.name}",
-      attributes={
-          GEN_AI_OPERATION_NAME: "invoke_workflow",
-          "gen_ai.workflow.name": workflow.name,
-          GEN_AI_CONVERSATION_ID: context.session.id,
-      },
-  )
+) -> Iterator[TelemetryContext]:
+  """Opens an `invoke_workflow` span plus its duration metric for ``node``."""
+  with _use_invoke_workflow_span(
+      workflow.name,
+      context.session.id,
+      otel_context=context.telemetry_context.otel_context,
+  ) as span:
+    tel_ctx = TelemetryContext(otel_context=context_api.get_current())
+    yield tel_ctx
+    _maybe_set_associated_events(span, tel_ctx)
 
 
-def _default_node_span_metadata(
+@contextmanager
+def _invoke_node_span(
     context: Context, node: BaseNode
-) -> _SpanMetadata:
-  return _SpanMetadata(
-      name=f"invoke_node {node.name}",
+) -> Iterator[TelemetryContext]:
+  """Opens an `invoke_node` span for a plain node."""
+  with tracer.start_as_current_span(
+      f"invoke_node {node.name}",
       attributes={
           GEN_AI_OPERATION_NAME: "invoke_node",
           GEN_AI_CONVERSATION_ID: context.session.id,
       },
+      context=context.telemetry_context.otel_context,
+  ) as span:
+    tel_ctx = TelemetryContext(otel_context=context_api.get_current())
+    yield tel_ctx
+    _maybe_set_associated_events(span, tel_ctx)
+
+
+def _maybe_set_associated_events(
+    span: Span, telemetry_context: TelemetryContext
+) -> None:
+  """Stamps the node's associated event IDs onto its span, if any."""
+  if span.is_recording() and len(telemetry_context._associated_event_ids) > 0:
+    span.set_attribute(
+        "gcp.vertex.agent.associated_event_ids",
+        telemetry_context._associated_event_ids,
+    )
+
+
+@contextmanager
+def _use_invoke_workflow_span(
+    workflow_name: str,
+    conversation_id: str,
+    *,
+    otel_context: context_api.Context | None = None,
+) -> Iterator[Span]:
+  """Opens an `invoke_workflow {workflow_name}` span."""
+  from . import _metrics
+
+  if otel_context is None:
+    otel_context = context_api.get_current()
+  # First workflow in the invocation is the entrypoint. The flag rides along the
+  # otel_context propagated to child nodes, so nested workflows see it set.
+  is_entrypoint = not context_api.get_value(
+      _ENTRYPOINT_WORKFLOW_KEY, otel_context
   )
+  if is_entrypoint:
+    otel_context = context_api.set_value(
+        _ENTRYPOINT_WORKFLOW_KEY, True, otel_context
+    )
+  attributes = {
+      GEN_AI_OPERATION_NAME: "invoke_workflow",
+      "gen_ai.workflow.name": workflow_name,
+      GEN_AI_CONVERSATION_ID: conversation_id,
+      GEN_AI_WORKFLOW_IS_ENTRYPOINT: is_entrypoint,
+  }
+  start_s = time.monotonic()
+  error: Exception | None = None
+  try:
+    with tracer.start_as_current_span(
+        f"invoke_workflow {workflow_name}",
+        attributes=attributes,
+        context=otel_context,
+    ) as span:
+      yield span
+  except Exception as e:  # pylint: disable=broad-exception-caught
+    error = e
+    raise
+  finally:
+    _metrics.record_workflow_invocation_duration(
+        workflow_name, time.monotonic() - start_s, is_entrypoint, error
+    )