feat(workflow): trace workflow and node execution with OpenTelemetry

core/package.json

@@ -72,6 +72,7 @@
   },
   "devDependencies": {
     "@mikro-orm/sqlite": "^6.6.6",
+    "@opentelemetry/context-async-hooks": "^2.1.0",
     "@types/adm-zip": "^0.5.8",
     "@types/express": "^4.17.25",
     "@types/lodash-es": "^4.17.12",

core/src/telemetry/tracing.ts

@@ -35,6 +35,13 @@ const GEN_AI_TOOL_DESCRIPTION = 'gen_ai.tool.description';
 const GEN_AI_TOOL_NAME = 'gen_ai.tool.name';
 const GEN_AI_TOOL_TYPE = 'gen_ai.tool.type';
 
+const ADK_WORKFLOW_NAME = 'adk.workflow.name';
+const ADK_NODE_PATH = 'adk.node.path';
+const ADK_NODE_RUN_ID = 'adk.node.run_id';
+const ADK_NODE_ATTEMPT = 'adk.node.attempt';
+const ADK_NODE_STATUS = 'adk.node.status';
+const ADK_NODE_INTERRUPT_COUNT = 'adk.node.interrupt_count';
+
 export const tracer = trace.getTracer('gcp.vertex.agent', version);
 
 /**
@@ -90,6 +97,55 @@ export function traceAgentInvocation({
   });
 }
 
+export interface TraceWorkflowInvocationParams {
+  workflowName: string;
+  nodePath: string;
+}
+
+export function traceWorkflowInvocation({
+  workflowName,
+  nodePath,
+}: TraceWorkflowInvocationParams): void {
+  const span = trace.getActiveSpan();
+  if (!span) return;
+
+  span.setAttributes({
+    [GEN_AI_OPERATION_NAME]: 'invoke_workflow',
+    [ADK_WORKFLOW_NAME]: workflowName,
+    [ADK_NODE_PATH]: nodePath,
+  });
+}
+
+export type NodeExecutionStatus = 'completed' | 'waiting' | 'failed';
+
+export interface TraceNodeExecutionParams {
+  nodePath: string;
+  runId: string;
+  attempt: number;
+  status: NodeExecutionStatus;
+  interruptCount: number;
+}
+
+export function traceNodeExecution({
+  nodePath,
+  runId,
+  attempt,
+  status,
+  interruptCount,
+}: TraceNodeExecutionParams): void {
+  const span = trace.getActiveSpan();
+  if (!span) return;
+
+  span.setAttributes({
+    [GEN_AI_OPERATION_NAME]: 'execute_node',
+    [ADK_NODE_PATH]: nodePath,
+    [ADK_NODE_RUN_ID]: runId,
+    [ADK_NODE_ATTEMPT]: attempt,
+    [ADK_NODE_STATUS]: status,
+    [ADK_NODE_INTERRUPT_COUNT]: interruptCount,
+  });
+}
+
 export interface TraceToolCallParams {
   tool: BaseTool;
   args: Record<string, unknown>;

core/src/workflow/node_runner.ts

@@ -4,8 +4,11 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+import {context, type Span, SpanStatusCode, trace} from '@opentelemetry/api';
 import {InvocationContext} from '../agents/invocation_context.js';
 import {Event} from '../events/event.js';
+import {traceNodeExecution, tracer} from '../telemetry/tracing.js';
+import {formatError} from '../utils/error_utils.js';
 import {BaseNode} from './base_node.js';
 import {createSubBranch} from './branch_path.js';
 import {
@@ -84,20 +87,56 @@ export interface ExecuteChildNodeParams {
  * duplicate observable events across retries should emit only after their
  * fallible work has succeeded.
  */
-export async function executeChildNode({
-  parent,
-  node,
-  input,
-  options = {},
-  abortSignal,
-  nodeState: callerNodeState,
-}: ExecuteChildNodeParams): Promise<NodeContext> {
+export function executeChildNode(
+  params: ExecuteChildNodeParams,
+): Promise<NodeContext> {
+  const {parent, node, options = {}} = params;
   const nodeName = options.nodeName ?? node.name;
-  const runId = options.runId ?? nodeName;
   const nodePath =
     options.overrideNodePath ??
     (parent.nodePath ? `${parent.nodePath}.${nodeName}` : nodeName);
 
+  // The span is started here, synchronously, rather than inside `runChildNode`:
+  // its parent must be whatever span was active when the workflow SCHEDULED
+  // this node. Nodes are raced concurrently in `Workflow.runLoop`, so a parent
+  // captured any later would nest concurrent siblings inside whichever task
+  // happened to resolve first.
+  const span = tracer.startSpan(`execute_node ${nodeName}`);
+
+  // Deliberately not `async`, and the callback is deliberately not `async`
+  // either: `context.with` hands the inner promise straight back, so the child
+  // settles on exactly the same microtask it did before tracing existed. Async
+  // wrappers here would each add promise-adoption ticks, which is enough to
+  // change how concurrently scheduled nodes interleave — an observability
+  // change must not move execution around.
+  return context.with(trace.setSpan(context.active(), span), () =>
+    runChildNode({params, nodeName, nodePath, span}),
+  );
+}
+
+interface RunChildNodeParams {
+  params: ExecuteChildNodeParams;
+  nodeName: string;
+  nodePath: string;
+  span: Span;
+}
+
+/** The body of {@link executeChildNode}, running under its `execute_node` span. */
+async function runChildNode({
+  params: {
+    parent,
+    node,
+    input,
+    options = {},
+    abortSignal,
+    nodeState: callerNodeState,
+  },
+  nodeName,
+  nodePath,
+  span,
+}: RunChildNodeParams): Promise<NodeContext> {
+  const runId = options.runId ?? nodeName;
+
   let branch = parent.branch;
   if (options.overrideBranch !== undefined) {
     branch = options.overrideBranch;
@@ -150,67 +189,158 @@ export async function executeChildNode({
     });
 
   const pluginManager = child.invocationContext.pluginManager;
-  if (pluginManager?.hasPlugins) {
-    const skipOutput = await pluginManager.runBeforeNodeCallback({
-      node,
-      nodeContext: child,
-      input,
-    });
-    if (skipOutput !== undefined) {
-      child.output = skipOutput;
-      if (options.useAsOutput) {
-        parent.output = child.output;
-        parent.route = child.route;
+
+  try {
+    if (pluginManager?.hasPlugins) {
+      const skipOutput = await pluginManager.runBeforeNodeCallback({
+        node,
+        nodeContext: child,
+        input,
+      });
+      if (skipOutput !== undefined) {
+        child.output = skipOutput;
+        // A skipped node still fills its slot in the trace, so record it as
+        // completed rather than leaving an attribute-less span behind.
+        traceNodeExecution({
+          nodePath,
+          runId,
+          attempt: nodeState.attemptCount,
+          status: 'completed',
+          interruptCount: child.interruptIds.length,
+        });
+        if (options.useAsOutput) {
+          parent.output = child.output;
+          parent.route = child.route;
+        }
+        return child;
       }
-      return child;
     }
-  }
 
-  let succeeded = false;
-  while (!succeeded) {
-    resetState(child);
-    child.attemptCount = nodeState.attemptCount;
-    try {
-      await runOnce({node, child, input, nodeName, branch, isolationScope});
-      succeeded = true;
-    } catch (err) {
-      // Cancellation is terminal: an aborted invocation (or a sibling failure
-      // that cancelled this node) is never retried.
-      if (isInvocationAbortedError(err)) {
-        throw err;
-      }
-      // Check retry eligibility with the attempt that just failed, compute its
-      // backoff delay, THEN advance the counter (matches Python semantics).
-      const retryConfig = node.preparedRetryConfig;
-      if (
-        !retryConfig ||
-        !shouldRetryNode({error: err, retryConfig, nodeState})
-      ) {
-        throw err;
+    let succeeded = false;
+    while (!succeeded) {
+      resetState(child);
+      child.attemptCount = nodeState.attemptCount;
+      try {
+        await runAttempt({
+          node,
+          child,
+          input,
+          nodeName,
+          branch,
+          isolationScope,
+          nodePath,
+          runId,
+          attempt: nodeState.attemptCount,
+        });
+        succeeded = true;
+      } catch (err) {
+        // Cancellation is terminal: an aborted invocation (or a sibling
+        // failure that cancelled this node) is never retried.
+        if (isInvocationAbortedError(err)) {
+          throw err;
+        }
+        // Check retry eligibility with the attempt that just failed, compute
+        // its backoff delay, THEN advance the counter (matches Python
+        // semantics).
+        const retryConfig = node.preparedRetryConfig;
+        if (
+          !retryConfig ||
+          !shouldRetryNode({error: err, retryConfig, nodeState})
+        ) {
+          throw err;
+        }
+        const delaySeconds = getRetryDelaySeconds({retryConfig, nodeState});
+        nodeState.attemptCount += 1;
+        await delay(delaySeconds * 1000, effectiveAbortSignal);
       }
-      const delaySeconds = getRetryDelaySeconds({retryConfig, nodeState});
-      nodeState.attemptCount += 1;
-      await delay(delaySeconds * 1000, effectiveAbortSignal);
     }
-  }
 
-  if (pluginManager?.hasPlugins) {
-    const replacedOutput = await pluginManager.runAfterNodeCallback({
-      node,
-      nodeContext: child,
-      output: child.output,
+    traceNodeExecution({
+      nodePath,
+      runId,
+      attempt: nodeState.attemptCount,
+      status: child.interruptIds.length > 0 ? 'waiting' : 'completed',
+      interruptCount: child.interruptIds.length,
     });
-    if (replacedOutput !== undefined) {
-      child.output = replacedOutput;
+
+    if (pluginManager?.hasPlugins) {
+      const replacedOutput = await pluginManager.runAfterNodeCallback({
+        node,
+        nodeContext: child,
+        output: child.output,
+      });
+      if (replacedOutput !== undefined) {
+        child.output = replacedOutput;
+      }
+    }
+
+    if (options.useAsOutput) {
+      parent.output = child.output;
+      parent.route = child.route;
     }
-  }
 
-  if (options.useAsOutput) {
-    parent.output = child.output;
-    parent.route = child.route;
+    return child;
+  } catch (err) {
+    traceNodeExecution({
+      nodePath,
+      runId,
+      attempt: nodeState.attemptCount,
+      status: 'failed',
+      interruptCount: child.interruptIds.length,
+    });
+    span.setStatus({code: SpanStatusCode.ERROR, message: formatError(err)});
+    throw err;
+  } finally {
+    span.end();
   }
+}
+
+interface RunAttemptParams extends RunOnceParams {
+  nodePath: string;
+  runId: string;
+  attempt: number;
+}
 
-  return child;
+/**
+ * Not `async`: a node without a retry config must reach `runOnce` and settle on
+ * exactly the microtask it would have without tracing (see `executeChildNode`).
+ */
+function runAttempt(params: RunAttemptParams): Promise<void> {
+  const {node, nodePath, runId, attempt} = params;
+  if (!node.preparedRetryConfig) {
+    return runOnce(params);
+  }
+  return tracer.startActiveSpan(
+    `execute_node_attempt ${params.nodeName}`,
+    async (span) => {
+      try {
+        await runOnce(params);
+        traceNodeExecution({
+          nodePath,
+          runId,
+          attempt,
+          status:
+            params.child.interruptIds.length > 0 ? 'waiting' : 'completed',
+          interruptCount: params.child.interruptIds.length,
+        });
+      } catch (err) {
+        traceNodeExecution({
+          nodePath,
+          runId,
+          attempt,
+          status: 'failed',
+          interruptCount: params.child.interruptIds.length,
+        });
+        span.setStatus({
+          code: SpanStatusCode.ERROR,
+          message: formatError(err),
+        });
+        throw err;
+      } finally {
+        span.end();
+      }
+    },
+  );
 }
 
 /**

core/src/workflow/workflow.ts

@@ -4,7 +4,9 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+import {context, trace} from '@opentelemetry/api';
 import {Event} from '../events/event.js';
+import {tracer, traceWorkflowInvocation} from '../telemetry/tracing.js';
 import {experimental} from '../utils/experimental.js';
 import {BaseNode, BaseNodeConfig} from './base_node.js';
 import {commonPrefixOf} from './branch_path.js';
@@ -173,10 +175,21 @@ export class Workflow extends BaseNode {
       abort.controller.signal,
     );
 
+    const span = tracer.startSpan(`invoke_workflow ${this.name}`);
     try {
-      await this.orchestrate(ctx, nodeInput, dynamicState, abort.controller);
+      // Sync callback returning the promise, not `async () => await …`: the
+      // wrapper must not insert microtask hops around the orchestration loop
+      // (see the note in `executeChildNode`).
+      await context.with(trace.setSpan(context.active(), span), () => {
+        traceWorkflowInvocation({
+          workflowName: this.name,
+          nodePath: ctx.nodePath,
+        });
+        return this.orchestrate(ctx, nodeInput, dynamicState, abort.controller);
+      });
     } finally {
       abort.dispose();
+      span.end();
     }
   }