diff --git a/CHANGELOG.md b/CHANGELOG.md index 536e8b1c8..7f4de0894 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,8 @@ to docs, or any other relevant information. customizable via the `cancelWorkflowUpdate` handler option; the default rejects with a `NOT_IMPLEMENTED` handler error. - **Experimental**: `@temporalio/google-adk-agents` package for running Google ADK agents as durable Temporal Workflows. + ADK's OpenTelemetry agent-loop spans can be exported replay-safely from the Workflow sandbox by composing with + `OpenTelemetryPlugin` from `@temporalio/interceptors-opentelemetry`; see the package README's telemetry section. ### Fixed diff --git a/contrib/google-adk-agents/README.md b/contrib/google-adk-agents/README.md index 58db2c4a9..b5ef41f71 100644 --- a/contrib/google-adk-agents/README.md +++ b/contrib/google-adk-agents/README.md @@ -190,6 +190,69 @@ const plugin = new GoogleAdkPlugin({ }); ``` +## Telemetry and observability + +ADK instruments its agent loop with OpenTelemetry: the `gcp.vertex.agent` +tracer creates `invocation`, `invoke_agent `, `call_llm` (with +`gen_ai.usage.input_tokens` / `gen_ai.usage.output_tokens` attributes), and +`execute_tool` spans. Under this plugin the agent loop runs inside the Workflow +sandbox, so those spans are created there too — and **by default they are +silently dropped**: nothing registers a tracer provider inside the sandbox, so +ADK's tracer yields non-recording no-ops. A tracer provider configured in the +worker process (e.g. `NodeSDK`) does not see them either; the sandbox is +isolated from process globals. + +To export them, compose with the SDK's OpenTelemetry integration +([`@temporalio/interceptors-opentelemetry`](https://github.com/temporalio/sdk-typescript/tree/main/contrib/interceptors-opentelemetry)), +placed before this plugin: + +```typescript +import { OpenTelemetryPlugin } from '@temporalio/interceptors-opentelemetry'; + +const worker = await Worker.create({ + // ... + plugins: [new OpenTelemetryPlugin({ resource, spanProcessor }), new GoogleAdkPlugin()], +}); +``` + +Its Workflow interceptor registers a tracer provider inside the sandbox that +ADK's tracer binds to, and exports the spans through a Worker sink that is +**replay-gated**: spans are recorded when Workflow code first executes; +history replays re-run the agent-loop code but re-export nothing. You get +exactly one span per real model call or tool call, plus the interceptor's own +`RunWorkflow` / `StartActivity` spans, nested under the same trace. The plugin +pins `@opentelemetry/api` to a single copy in the Workflow bundle (the one +`@google/adk` resolves — ADK pins an exact api version, so a bundle could +otherwise contain two copies), so ADK's tracer binds to that provider no +matter which module evaluates ADK first. + +Cautions: + +- Do **not** register a custom telemetry sink with `callDuringReplay: true` — + every replayed workflow task would then re-emit the agent-loop spans, + over-counting each operation once per replay. +- The replay gate makes span export at-least-once, not exactly-once: a workflow + task **retry** (a task that failed or timed out and re-executes) is not a + replay, so its spans are re-emitted. Retries are rare in normal operation, but + don't build alerting that assumes exact span counts. +- `call_llm` spans carry the full request/response payloads as + `gcp.vertex.agent.llm_request` / `gcp.vertex.agent.llm_response` attributes. + Point the span processor somewhere approved for prompt content, or strip + those attributes in the processor. +- Custom payload/failure converter modules (`payloadConverterPath` / + `failureConverterPath`) evaluate **before** the plugin's polyfill loader. If + such a module imports `@google/adk` / `@google/genai`, import + `@temporalio/google-adk-agents/workflow` first: it installs the sandbox + polyfills (`Headers`, `structuredClone`, the WHATWG streams globals, and the + deterministic `performance` shim ADK's telemetry chain dereferences at module + load) before ADK evaluates. + +ADK (as of 1.4.0) defines no OpenTelemetry metric instruments, so there is no +workflow-side metric telemetry to configure. Activity-side telemetry (the real +Gemini/MCP calls) runs in the worker process where normal Node OpenTelemetry +setup applies, and is naturally replay-immune — completed Activities are read +from history rather than re-executed. + ## Operational notes - Register `GoogleAdkPlugin` on the Worker. Passing it directly to `Client` does diff --git a/contrib/google-adk-agents/package.json b/contrib/google-adk-agents/package.json index 2ca7bcb92..1580317f9 100644 --- a/contrib/google-adk-agents/package.json +++ b/contrib/google-adk-agents/package.json @@ -75,7 +75,11 @@ "devDependencies": { "@google/adk": "^1.4.0", "@google/genai": "^2.8.0", + "@opentelemetry/api": "^1.9.0", + "@opentelemetry/resources": "^1.25.1", + "@opentelemetry/sdk-trace-base": "^1.25.1", "@temporalio/client": "workspace:*", + "@temporalio/interceptors-opentelemetry": "workspace:*", "@temporalio/testing": "workspace:*", "@temporalio/worker": "workspace:*", "@types/node": "^20.0.0", diff --git a/contrib/google-adk-agents/src/__tests__/adk-converter.ts b/contrib/google-adk-agents/src/__tests__/adk-converter.ts new file mode 100644 index 000000000..b21468cc0 --- /dev/null +++ b/contrib/google-adk-agents/src/__tests__/adk-converter.ts @@ -0,0 +1,25 @@ +/** + * @license + * Copyright 2025 Temporal Technologies Inc. + * SPDX-License-Identifier: MIT + * + * A custom payload-converter module that imports `@google/adk`, following the + * documented workaround: converter modules evaluate before interceptor modules + * (so before the plugin's polyfill loader), and must therefore import + * `@temporalio/google-adk-agents/workflow` first to install the sandbox + * polyfills themselves. The E2E test bundles the tsc-compiled CommonJS form of + * this file (`lib/__tests__/adk-converter.js`), where every `require` is eager + * — the module layout that crashed on the `performance` global before the + * polyfill loader shimmed it. + */ + +// The documented workaround import — must come first. +// eslint-disable-next-line import/no-unassigned-import +import '../workflow'; +import { LlmAgent } from '@google/adk'; +import { defaultPayloadConverter } from '@temporalio/common'; + +/** Referenced so the `@google/adk` barrel import is not elided. */ +export const adkEvaluated = typeof LlmAgent === 'function'; + +export const payloadConverter = defaultPayloadConverter; diff --git a/contrib/google-adk-agents/src/__tests__/adk-first-interceptor.ts b/contrib/google-adk-agents/src/__tests__/adk-first-interceptor.ts new file mode 100644 index 000000000..8884dfe27 --- /dev/null +++ b/contrib/google-adk-agents/src/__tests__/adk-first-interceptor.ts @@ -0,0 +1,17 @@ +/** + * @license + * Copyright 2025 Temporal Technologies Inc. + * SPDX-License-Identifier: MIT + * + * A user `interceptors.workflowModules` entry that imports `@google/adk`. + * Interceptor modules all evaluate before any interceptor factory runs, so + * this makes ADK's `telemetry/tracing.js` cache its tracer at module load — + * before `OpenTelemetryPlugin` registers the sandbox tracer provider. The + * telemetry test uses it to pin that ADK spans still export in this ordering, + * which requires the bundle to hold a single `@opentelemetry/api` copy. + */ + +import { LlmAgent } from '@google/adk'; + +/** Referenced so the `@google/adk` barrel import is not elided. */ +export const adkEvaluated = typeof LlmAgent === 'function'; diff --git a/contrib/google-adk-agents/src/__tests__/compiled-cjs.test.ts b/contrib/google-adk-agents/src/__tests__/compiled-cjs.test.ts new file mode 100644 index 000000000..513c31b59 --- /dev/null +++ b/contrib/google-adk-agents/src/__tests__/compiled-cjs.test.ts @@ -0,0 +1,75 @@ +/** + * @license + * Copyright 2025 Temporal Technologies Inc. + * SPDX-License-Identifier: MIT + * + * Regression coverage for workflow-bundle module layouts that evaluate ADK's + * telemetry chain (`@opentelemetry/sdk-trace-base` → `@opentelemetry/core`, + * whose browser build dereferences the `performance` global at module load) + * eagerly. The ESM test workflows dodge that chain through harmony-import + * pruning; the tsc-compiled CommonJS shape — what a compiled-TS production + * worker and the published `lib/` artifacts actually run — `require`s it + * unconditionally, so these tests pin that such bundles load and run with + * `GoogleAdkPlugin` alone (no `OpenTelemetryPlugin` composed). + */ + +import path from 'node:path'; + +import test from 'ava'; +import { Worker } from '@temporalio/worker'; + +import { GoogleAdkPlugin } from '../index'; +import { defaultTestProvider, setupTestEnv, uid, workflowsPath } from './helpers'; +import { singleModelCall } from './workflows'; + +const getEnv = setupTestEnv(test); + +// `__dirname` is `lib/__tests__` at runtime, so these resolve to the +// tsc-compiled CommonJS artifacts, not the TypeScript sources. +const compiledWorkflowsPath = path.resolve(__dirname, './workflows.js'); +const compiledConverterPath = path.resolve(__dirname, './adk-converter.js'); + +// Compiled-CJS workflow files load and run without OpenTelemetryPlugin (E2E) +test.serial('compiledCjsWorkflowBundleRuns', async (t) => { + const env = getEnv(); + const taskQueue = uid('adk-cjs'); + const worker = await Worker.create({ + connection: env.nativeConnection, + taskQueue, + workflowsPath: compiledWorkflowsPath, + plugins: [new GoogleAdkPlugin({ modelProvider: defaultTestProvider() })], + }); + const result = await worker.runUntil( + env.client.workflow.execute(singleModelCall, { + taskQueue, + workflowId: uid('wf-cjs'), + args: ['hello'], + }) + ); + t.is(result, 'fake-response:fake-model'); +}); + +// A converter module importing @google/adk works via the documented workaround (E2E) +test.serial('converterModuleImportingAdkRunsWithWorkaround', async (t) => { + const env = getEnv(); + const taskQueue = uid('adk-conv'); + // Converter modules evaluate before interceptor modules — before the + // plugin's polyfill loader — so `adk-converter.ts` imports the workflow + // barrel first, per the documented workaround. Its compiled-CJS form + // evaluates the whole ADK barrel (telemetry chain included) eagerly. + const worker = await Worker.create({ + connection: env.nativeConnection, + taskQueue, + workflowsPath, + dataConverter: { payloadConverterPath: compiledConverterPath }, + plugins: [new GoogleAdkPlugin({ modelProvider: defaultTestProvider() })], + }); + const result = await worker.runUntil( + env.client.workflow.execute(singleModelCall, { + taskQueue, + workflowId: uid('wf-conv'), + args: ['hello'], + }) + ); + t.is(result, 'fake-response:fake-model'); +}); diff --git a/contrib/google-adk-agents/src/__tests__/helpers.ts b/contrib/google-adk-agents/src/__tests__/helpers.ts index 85bdb80ea..0ecc9c0e6 100644 --- a/contrib/google-adk-agents/src/__tests__/helpers.ts +++ b/contrib/google-adk-agents/src/__tests__/helpers.ts @@ -127,6 +127,11 @@ export interface WithWorkerOptions { plugins: unknown[]; activities?: Record Promise>; maxCachedWorkflows?: number; + /** + * User `interceptors.workflowModules` entries; the plugin's polyfill loader + * still evaluates first, these follow. + */ + workflowInterceptorModules?: string[]; } /** @@ -144,6 +149,9 @@ export async function withWorker( plugins: options.plugins as any, activities: options.activities as any, maxCachedWorkflows: options.maxCachedWorkflows, + interceptors: options.workflowInterceptorModules + ? { workflowModules: options.workflowInterceptorModules } + : undefined, }); return worker.runUntil(fn()); } diff --git a/contrib/google-adk-agents/src/__tests__/plugin.test.ts b/contrib/google-adk-agents/src/__tests__/plugin.test.ts index 1475ac5ed..817579dcd 100644 --- a/contrib/google-adk-agents/src/__tests__/plugin.test.ts +++ b/contrib/google-adk-agents/src/__tests__/plugin.test.ts @@ -7,6 +7,8 @@ * `configureWorker` output without bundling or executing a Workflow. */ +import { createRequire } from 'node:module'; + import test from 'ava'; import type { BundleOptions, WorkerOptions } from '@temporalio/worker'; @@ -41,6 +43,25 @@ test('configureBundler stubs ADK node-only packages and disallowed builtins', (t for (const polyfilled of ['assert', 'url', 'util']) { t.false(ignored.has(polyfilled), `expected polyfilled builtin ${polyfilled} to remain`); } + // The pure-JS OpenTelemetry tracing packages must stay resolvable: + // `@temporalio/interceptors-opentelemetry` constructs a `BasicTracerProvider` + // from them inside the sandbox — the SDK's only replay-safe workflow span + // path. Re-stubbing any of these breaks composing with `OpenTelemetryPlugin`. + for (const otel of ['@opentelemetry/api', '@opentelemetry/sdk-trace-base', '@opentelemetry/resources']) { + t.false(ignored.has(otel), `expected OpenTelemetry package ${otel} to remain resolvable`); + } +}); + +test('configureBundler prepends the polyfill loader to workflowInterceptorModules', (t) => { + const plugin = new GoogleAdkPlugin(); + const { workflowInterceptorModules } = plugin.configureBundler({ + workflowsPath: 'wf', + workflowInterceptorModules: ['user-interceptors'], + } as BundleOptions); + // Must be first so the web-global polyfills install before any other + // per-workflow module (interceptors, then the user's workflows) evaluates. + t.is(workflowInterceptorModules?.[0], require.resolve('../load-polyfills')); + t.deepEqual(workflowInterceptorModules?.slice(1), ['user-interceptors']); }); test('configureBundler appends the sandbox-compat plugin, preserving a user hook', (t) => { @@ -60,6 +81,54 @@ test('configureBundler appends the sandbox-compat plugin, preserving a user hook t.deepEqual(names, ['user-plugin', 'google-adk-sandbox-compat']); }); +test('sandbox-compat plugin rewrites @opentelemetry/api to the copy @google/adk resolves', (t) => { + const plugin = new GoogleAdkPlugin(); + const { webpackConfigHook } = plugin.configureBundler({ workflowsPath: 'wf' } as BundleOptions); + const cfg = webpackConfigHook!({ plugins: [] } as never) as { + plugins: Array<{ name?: string; apply(compiler: unknown): void }>; + }; + const compat = cfg.plugins.find((p) => p.name === 'google-adk-sandbox-compat')!; + + // Drive the plugin against a minimal fake compiler to capture its + // `beforeResolve` tap. + let beforeResolve: ((data: { request?: string }) => void) | undefined; + compat.apply({ + webpack: { + ProvidePlugin: class { + apply(): void {} + }, + }, + hooks: { + normalModuleFactory: { + tap: (_name: string, fn: (nmf: unknown) => void) => + fn({ + hooks: { + beforeResolve: { + tap: (_tapName: string, f: (data: { request?: string }) => void) => { + beforeResolve = f; + }, + }, + }, + }), + }, + }, + }); + t.truthy(beforeResolve); + + // Every `@opentelemetry/api` request must land on ADK's own resolution, so + // ADK's module-load tracer and the OTel interceptor's provider registration + // share one api instance regardless of module evaluation order. + const expected = createRequire(require.resolve('@google/adk')).resolve('@opentelemetry/api'); + const api = { request: '@opentelemetry/api' }; + beforeResolve!(api); + t.is(api.request, expected); + + // Packages that merely share the name prefix are untouched. + const apiLogs = { request: '@opentelemetry/api-logs' }; + beforeResolve!(apiLogs); + t.is(apiLogs.request, '@opentelemetry/api-logs'); +}); + test('configureWorker registers model activities, plus an MCP pair per toolset', (t) => { const modelOnly = new GoogleAdkPlugin().configureWorker({ taskQueue: 'tq' } as WorkerOptions).activities as Record< string, diff --git a/contrib/google-adk-agents/src/__tests__/telemetry.test.ts b/contrib/google-adk-agents/src/__tests__/telemetry.test.ts new file mode 100644 index 000000000..5f3ea3cef --- /dev/null +++ b/contrib/google-adk-agents/src/__tests__/telemetry.test.ts @@ -0,0 +1,179 @@ +/** + * @license + * Copyright 2025 Temporal Technologies Inc. + * SPDX-License-Identifier: MIT + * + * Telemetry replay-safety. ADK creates OpenTelemetry spans (tracer + * `gcp.vertex.agent`: `invocation`, `invoke_agent `, `call_llm`, …) + * inside the Workflow sandbox. Composing this plugin with the SDK's + * `OpenTelemetryPlugin` exports them through the Worker's replay-gated span + * sink — so with the Workflow cache disabled (every workflow task replays the + * full history), each real operation must still export exactly one span. + * Also pins the isolation property the gate rests on: without the + * OpenTelemetry plugin, sandbox spans never reach a process-global provider. + */ + +import path from 'node:path'; + +import test, { type ExecutionContext } from 'ava'; +import { trace } from '@opentelemetry/api'; +import { Resource } from '@opentelemetry/resources'; +import { BasicTracerProvider, InMemorySpanExporter, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base'; +import { OpenTelemetryPlugin } from '@temporalio/interceptors-opentelemetry'; + +import { GoogleAdkPlugin } from '../index'; +import { defaultTestProvider, setupTestEnv, uid, withWorker } from './helpers'; +import { agentRunnerTwoTurnsWorkflow } from './workflows'; + +const getEnv = setupTestEnv(test); + +function makeAdkPlugin(): GoogleAdkPlugin { + return new GoogleAdkPlugin({ modelProvider: defaultTestProvider() }); +} + +/** Span-name → count for spans recorded by ADK's `gcp.vertex.agent` tracer. */ +function adkSpanCounts(exporter: InMemorySpanExporter): Record { + const counts: Record = {}; + for (const span of exporter.getFinishedSpans()) { + if (span.instrumentationLibrary.name !== 'gcp.vertex.agent') continue; + counts[span.name] = (counts[span.name] ?? 0) + 1; + } + return counts; +} + +/** `WorkflowTaskTimedOut`/`WorkflowTaskFailed` events — retried (not replayed) workflow tasks. */ +function countWorkflowTaskRetries( + events: Array<{ workflowTaskTimedOutEventAttributes?: unknown; workflowTaskFailedEventAttributes?: unknown }> +): number { + return events.filter( + (e) => e.workflowTaskTimedOutEventAttributes != null || e.workflowTaskFailedEventAttributes != null + ).length; +} + +/** + * Asserts exactly `expected[name]` exported ADK spans per name on a clean + * history. A workflow task retry (timeout/failure) is not a replay — the + * retried segment's spans legitimately re-emit (see README) — so when a slow + * runner caused retries, degrade to a lower bound, which still fails on + * silently dropped spans. + */ +function assertAdkSpanCounts( + t: ExecutionContext, + exporter: InMemorySpanExporter, + workflowTaskRetries: number, + expected: Record +): void { + const counts = adkSpanCounts(exporter); + for (const [name, n] of Object.entries(expected)) { + if (workflowTaskRetries === 0) { + t.is(counts[name], n, name); + } else { + t.true( + (counts[name] ?? 0) >= n, + `${name}: expected >= ${n} spans, got ${ + counts[name] ?? 0 + } (history has ${workflowTaskRetries} workflow task retries)` + ); + } + } +} + +// Composing with OpenTelemetryPlugin exports ADK spans; replays add none (E2E) +test.serial('adkSpansExportOncePerOperationUnderReplay', async (t) => { + const env = getEnv(); + const taskQueue = uid('adk-otel'); + const workflowId = uid('wf-otel'); + + const exporter = new InMemorySpanExporter(); + const otelPlugin = new OpenTelemetryPlugin({ + resource: new Resource({ 'service.name': 'adk-telemetry-test' }), + spanProcessor: new SimpleSpanProcessor(exporter), + }); + + // Observability plugins compose before this one (see README). The workflow + // cache is disabled so every workflow task after the first replays the whole + // history — re-running ADK's span-creating agent-loop code in the sandbox. + const result = await withWorker( + env, + { taskQueue, plugins: [otelPlugin, makeAdkPlugin()], maxCachedWorkflows: 0 }, + () => env.client.workflow.execute(agentRunnerTwoTurnsWorkflow, { taskQueue, workflowId, args: ['hi'] }) + ); + t.is(result, 'fake-response:fake-model|fake-response:fake-model'); + + const history = await env.client.workflow.getHandle(workflowId).fetchHistory(); + const events = history.events ?? []; + const workflowTasks = events.filter((e) => e.workflowTaskStartedEventAttributes).length; + const modelActivities = events.filter((e) => e.activityTaskScheduledEventAttributes).length; + t.is(modelActivities, 2); + t.true(workflowTasks >= 3, `expected >= 3 workflow tasks so replays occurred, got ${workflowTasks}`); + + // Exactly one exported span per real operation. A regression that lets + // replayed sandbox code re-emit (e.g. a callDuringReplay sink) would show up + // here as workflowTasks-proportional counts (3+ per name). + assertAdkSpanCounts(t, exporter, countWorkflowTaskRetries(events), { + call_llm: 2, + invocation: 2, + 'invoke_agent assistant': 2, + }); +}); + +// ADK evaluated before the interceptor factories still exports spans (E2E) +test.serial('adkSpansExportWhenAdkEvaluatesBeforeInterceptorFactories', async (t) => { + const env = getEnv(); + const taskQueue = uid('adk-otel-early'); + const workflowId = uid('wf-otel-early'); + + const exporter = new InMemorySpanExporter(); + const otelPlugin = new OpenTelemetryPlugin({ + resource: new Resource({ 'service.name': 'adk-telemetry-test' }), + spanProcessor: new SimpleSpanProcessor(exporter), + }); + + // A user workflow-interceptors module importing `@google/adk` evaluates ADK + // — and its module-load `trace.getTracer(...)` — before any interceptor + // factory registers the sandbox tracer provider. ADK's tracer must still + // bind to that provider, which requires the bundle to hold a single + // `@opentelemetry/api` copy: with two copies (ADK pins one exact version), + // every ADK span is silently dropped while the interceptor's own spans keep + // exporting. + const result = await withWorker( + env, + { + taskQueue, + plugins: [otelPlugin, makeAdkPlugin()], + maxCachedWorkflows: 0, + workflowInterceptorModules: [path.resolve(__dirname, '../../src/__tests__/adk-first-interceptor.ts')], + }, + () => env.client.workflow.execute(agentRunnerTwoTurnsWorkflow, { taskQueue, workflowId, args: ['hi'] }) + ); + t.is(result, 'fake-response:fake-model|fake-response:fake-model'); + + const { events } = await env.client.workflow.getHandle(workflowId).fetchHistory(); + assertAdkSpanCounts(t, exporter, countWorkflowTaskRetries(events ?? []), { + call_llm: 2, + invocation: 2, + 'invoke_agent assistant': 2, + }); +}); + +// Without the OpenTelemetry plugin, sandbox spans never reach a process-global provider (E2E) +test.serial('adkSpansDoNotLeakToProcessGlobalProvider', async (t) => { + const env = getEnv(); + const taskQueue = uid('adk-otel-isolate'); + const workflowId = uid('wf-otel-isolate'); + + const exporter = new InMemorySpanExporter(); + const provider = new BasicTracerProvider(); + provider.addSpanProcessor(new SimpleSpanProcessor(exporter)); + t.true(trace.setGlobalTracerProvider(provider)); + try { + const result = await withWorker(env, { taskQueue, plugins: [makeAdkPlugin()], maxCachedWorkflows: 0 }, () => + env.client.workflow.execute(agentRunnerTwoTurnsWorkflow, { taskQueue, workflowId, args: ['hi'] }) + ); + t.is(result, 'fake-response:fake-model|fake-response:fake-model'); + t.deepEqual(adkSpanCounts(exporter), {}); + } finally { + trace.disable(); + await provider.shutdown(); + } +}); diff --git a/contrib/google-adk-agents/src/__tests__/workflows.ts b/contrib/google-adk-agents/src/__tests__/workflows.ts index 93cae3527..db104c06b 100644 --- a/contrib/google-adk-agents/src/__tests__/workflows.ts +++ b/contrib/google-adk-agents/src/__tests__/workflows.ts @@ -339,3 +339,32 @@ export async function agentRunnerWorkflow(prompt: string): Promise { } return finalText; } + +/** + * Two sequential agent turns through the native runner loop, for the telemetry + * test. Produces a history with two `adk-invokeModel` Activities across three + * (or more) workflow tasks — so a cache-disabled worker replays the first + * turn's code on later tasks, while ADK's spans must be exported exactly once + * per real turn. + */ +export async function agentRunnerTwoTurnsWorkflow(prompt: string): Promise { + const agent = new LlmAgent({ + name: 'assistant', + model: new TemporalModel('fake-model'), + instruction: 'You are a helpful assistant.', + }); + const runner = new InMemoryRunner({ agent }); + + const texts: string[] = []; + for (let turn = 0; turn < 2; turn++) { + for await (const event of runner.runEphemeral({ + userId: 'test-user', + newMessage: { role: 'user', parts: [{ text: `${prompt}-${turn}` }] }, + })) { + if (isFinalResponse(event)) { + texts.push(stringifyContent(event)); + } + } + } + return texts.join('|'); +} diff --git a/contrib/google-adk-agents/src/load-polyfills.ts b/contrib/google-adk-agents/src/load-polyfills.ts index 445843bc8..2931ed141 100644 --- a/contrib/google-adk-agents/src/load-polyfills.ts +++ b/contrib/google-adk-agents/src/load-polyfills.ts @@ -6,11 +6,13 @@ * Web/Node global polyfills for the Temporal Workflow sandbox. * * `@google/adk` and `@google/genai` reference `Headers`, `structuredClone`, - * and the WHATWG streams globals while building requests in the agent loop. - * The Workflow sandbox does not expose all of them, so we install minimal - * polyfills — but ONLY inside Workflow context (gated on `inWorkflowContext()`), - * so a normal Node import (the worker / Activity side, tests, direct ADK use) - * is left untouched. The plugin barrel imports this module for its side effect. + * and the WHATWG streams globals while building requests in the agent loop, + * and ADK's telemetry modules reach `@opentelemetry/core`'s browser build, + * which dereferences the `performance` global at module load. The Workflow + * sandbox does not expose all of them, so we install minimal polyfills — but + * ONLY inside Workflow context (gated on `inWorkflowContext()`), so a normal + * Node import (the worker / Activity side, tests, direct ADK use) is left + * untouched. The plugin barrel imports this module for its side effect. */ import { inWorkflowContext } from '@temporalio/workflow'; @@ -33,4 +35,24 @@ if (inWorkflowContext()) { // eslint-disable-next-line @typescript-eslint/no-require-imports,import/no-unassigned-import require('web-streams-polyfill/polyfill'); } + + if (typeof globals.performance === 'undefined') { + // `@opentelemetry/core`'s browser build (reached from ADK's telemetry + // modules through `@opentelemetry/sdk-trace-base`) evaluates + // `export const otperformance = performance` at module load, and span + // timestamps flow through `performance.timeOrigin` / `performance.now()`. + // Both map onto the sandbox-patched `Date.now()` (workflow time), so the + // values are deterministic under replay. Mirrors the shim + // `@temporalio/interceptors-opentelemetry` installs from its own workflow + // runtime module; ours yields to an existing shim, while that package's + // unconditional `Object.assign` may later replace ours — harmless either + // way, since both compute identical values from the same sandbox clock. + const timeOrigin = Date.now(); + globals.performance = { + timeOrigin, + now() { + return Date.now() - timeOrigin; + }, + }; + } } diff --git a/contrib/google-adk-agents/src/plugin.ts b/contrib/google-adk-agents/src/plugin.ts index 0d11ebef0..29304ff44 100644 --- a/contrib/google-adk-agents/src/plugin.ts +++ b/contrib/google-adk-agents/src/plugin.ts @@ -4,7 +4,7 @@ * SPDX-License-Identifier: MIT */ -import { builtinModules } from 'node:module'; +import { builtinModules, createRequire } from 'node:module'; import type { BaseLlm } from '@google/adk'; import { SimplePlugin } from '@temporalio/plugin'; @@ -18,6 +18,24 @@ type WebpackConfig = Parameters> const NODE_SCHEME = 'node:'; +const OTEL_API_PACKAGE = '@opentelemetry/api'; + +/** + * Resolves the `@opentelemetry/api` copy that `@google/adk` itself resolves. + * ADK pins an exact api version while other packages in the Workflow bundle + * (notably `@temporalio/interceptors-opentelemetry`) may resolve a different + * one, so without intervention the bundle can contain two api copies. ADK's + * `telemetry/tracing.js` caches `trace.getTracer(...)` at module load; if that + * binds a different api copy than the one the OTel interceptor registers its + * tracer provider on, every ADK span is silently lost (the workflow still + * succeeds and interceptor spans still export). Rewriting every bundle request + * for `@opentelemetry/api` to ADK's copy keeps the tracer and the provider on + * the same registration regardless of module evaluation order. + */ +function adkOtelApiEntry(): string { + return createRequire(require.resolve('@google/adk')).resolve(OTEL_API_PACKAGE); +} + /** * ESM source for the `module` builtin shim. Every `@google/adk` compiled ESM * file carries an esbuild interop banner that *calls* `createRequire` at module @@ -125,6 +143,19 @@ const MIKRO_ORM_SHIM_SOURCE = 'export default {JsonType:JsonType,MikroORM:MikroORM,Entity:Entity,' + 'PrimaryKey:PrimaryKey,Property:Property,LockMode:LockMode};\n'; +/** + * ESM source for the `async_hooks` builtin shim. ADK's `utils/client_labels.js` + * executes `new AsyncLocalStorage()` at **module load** on the workflow-reached + * path (`models/base_llm.js` imports it), and later uses only `run(store, fn)` / + * `getStore()`. The Workflow sandbox injects a real, workflow-scoped + * `AsyncLocalStorage` onto its `globalThis` (the SDK's own `CancellationScope` + * is built on it), so re-exporting that global — the same contract as the + * langsmith contrib's `async-hooks-shim` — gives ADK full async-context + * tracking with sandbox-managed lifetime. + */ +const ASYNC_HOOKS_SHIM_SOURCE = + 'export const AsyncLocalStorage=globalThis.AsyncLocalStorage;export default {AsyncLocalStorage};\n'; + /** * Requests redirected (in `beforeResolve`) to an inline `data:` URI shim. These * are the packages/builtins ADK *dereferences at module load* (subclasses, @@ -139,6 +170,8 @@ const REQUEST_SHIM_SOURCES: ReadonlyArray = [ ['winston', WINSTON_SHIM_SOURCE], ['os', OS_SHIM_SOURCE], ['node:os', OS_SHIM_SOURCE], + ['async_hooks', ASYNC_HOOKS_SHIM_SOURCE], + ['node:async_hooks', ASYNC_HOOKS_SHIM_SOURCE], ['@mikro-orm/core', MIKRO_ORM_SHIM_SOURCE], ]; @@ -206,7 +239,8 @@ function disallowedBuiltins(): readonly string[] { * {@link REQUEST_SHIM_SOURCES}. `@opentelemetry/api` and `@opentelemetry/api-logs` * are deliberately *absent* (kept real): they are pure-JS API packages with no * node builtins, and `telemetry/tracing.js` calls `trace.getTracer(...)` from - * them at module load on the in-Workflow path. + * them at module load on the in-Workflow path. (`@opentelemetry/api` is + * additionally pinned to a single bundle copy — see {@link adkOtelApiEntry}.) */ const ADK_NODE_ONLY_SERVICE_PACKAGES: readonly string[] = [ 'google-auth-library', @@ -233,14 +267,19 @@ const ADK_NODE_ONLY_SERVICE_PACKAGES: readonly string[] = [ // setup functions that never run in a Workflow, so aliasing to an empty module // is load-safe. (`@opentelemetry/api` + `api-logs` are intentionally kept // real — see the doc comment above.) + // + // `@opentelemetry/sdk-trace-base` and `@opentelemetry/resources` must NOT be + // listed: they are pure-JS, and `@temporalio/interceptors-opentelemetry(-v2)` + // constructs a `BasicTracerProvider` from them *inside the Workflow sandbox* — + // the SDK's only replay-safe workflow span path (spans leave the isolate via a + // replay-gated sink). Stubbing them is bundle-wide and would break composing + // this plugin with `OpenTelemetryPlugin` for every workflow on the worker. '@opentelemetry/exporter-logs-otlp-http', '@opentelemetry/exporter-metrics-otlp-http', '@opentelemetry/exporter-trace-otlp-http', - '@opentelemetry/resources', '@opentelemetry/resource-detector-gcp', '@opentelemetry/sdk-logs', '@opentelemetry/sdk-metrics', - '@opentelemetry/sdk-trace-base', '@opentelemetry/sdk-trace-node', // The A2A (agent-to-agent) protocol subtree: ADK's `a2a/*` reach `@a2a-js/sdk` // and `express`, whose server/buffer code touches the web `Event` global / @@ -253,17 +292,21 @@ const ADK_NODE_ONLY_SERVICE_PACKAGES: readonly string[] = [ /** * The webpack plugin that makes the `@google/adk` barrel load inside the - * Workflow sandbox. It does three things: + * Workflow sandbox. It does four things: * * 1. **Shim redirects** ({@link REQUEST_SHIM_SOURCES}): in `beforeResolve`, * redirect the load-dereferenced requests to their inline `data:` URI shims. - * 2. **`node:` scheme strip**: every other `node:` → bare ``. The + * 2. **Single `@opentelemetry/api` copy**: rewrite every request for + * `@opentelemetry/api` to the copy `@google/adk` resolves + * ({@link adkOtelApiEntry}), so ADK's module-load tracer and the OTel + * interceptor's tracer-provider registration share one api instance. + * 3. **`node:` scheme strip**: every other `node:` → bare ``. The * Worker bundler aliases each disallowed builtin to `false` by its **bare** * name; a `node:`-prefixed request never reaches `resolve.alias` — webpack's * scheme handler intercepts it first and throws `UnhandledSchemeError` (a * hard *build* failure). Stripping the scheme lets the bundler's bare-name * policy take over. - * 3. **`process` provide**: a `ProvidePlugin` injects the deterministic + * 4. **`process` provide**: a `ProvidePlugin` injects the deterministic * `process` shim ({@link PROCESS_SHIM_SOURCE}) wherever `process` is a free * variable. */ @@ -273,6 +316,7 @@ function googleAdkSandboxCompatPlugin(): unknown { const shimUris = REQUEST_SHIM_SOURCES.map(([request, source]) => [request, toDataUri(source)] as const); const shimByRequest = new Map(shimUris); const processShimUri = toDataUri(PROCESS_SHIM_SOURCE); + const otelApiEntry = adkOtelApiEntry(); return { name: 'google-adk-sandbox-compat', apply(compiler: WebpackCompilerLike): void { @@ -289,6 +333,10 @@ function googleAdkSandboxCompatPlugin(): unknown { data.request = shim; return; } + if (request === OTEL_API_PACKAGE) { + data.request = otelApiEntry; + return; + } if (request.startsWith(NODE_SCHEME)) { data.request = request.slice(NODE_SCHEME.length); } @@ -375,7 +423,7 @@ export class GoogleAdkPlugin extends SimplePlugin { * recipe applies identically on both paths and there is no separate * `configureWorker`/`configureReplayWorker` bundler override. * - * The recipe has two parts, both required: + * The recipe has three parts, all required: * * 1. **`webpackConfigHook`** adds {@link googleAdkSandboxCompatPlugin} (the * `node:` strip, shim redirects, and `process` provide). @@ -386,6 +434,22 @@ export class GoogleAdkPlugin extends SimplePlugin { * `false` by the bundler — listing them additionally tells its determinism * guard "expected, don't fail" for the few ADK *core* reaches on paths that * never run in a Workflow. + * 3. **`workflowInterceptorModules`** gets the `load-polyfills` module + * prepended. Interceptor modules are evaluated per workflow — with the + * activator installed — *before* the user's workflow module, so the web + * globals `@google/adk`/`@google/genai` and ADK's OpenTelemetry chain + * dereference at module load (`ReadableStream`, `performance`, …) exist + * no matter what order the user's own imports evaluate in. The module + * exports no `interceptors`, so it registers nothing. (A webpack entry preload would not work: entry code evaluates + * at bundle load, before any activator, where the polyfill's + * `inWorkflowContext()` gate is false — and in the reusable-V8-context + * mode the no-op evaluation would be cached and never re-run.) + * Known gap: custom payload/failure converter modules + * (`payloadConverterPath` / `failureConverterPath`) evaluate *before* + * interceptor modules, so a converter module that itself imports + * `@google/adk`/`@google/genai` must import + * `@temporalio/google-adk-agents/workflow` (or `./load-polyfills`) first + * to install the polyfills. * * Tradeoff: putting **all** disallowed builtins in `ignoreModules` suppresses * the bundler's friendly "you imported a Node builtin in your Workflow" @@ -399,6 +463,7 @@ export class GoogleAdkPlugin extends SimplePlugin { return { ...base, ignoreModules, + workflowInterceptorModules: [require.resolve('./load-polyfills'), ...(base.workflowInterceptorModules ?? [])], webpackConfigHook: addSandboxCompat(base.webpackConfigHook), }; } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index bf5af7090..5e5d2eb68 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -252,9 +252,21 @@ importers: '@google/genai': specifier: ^2.8.0 version: 2.10.0(@modelcontextprotocol/sdk@1.26.0(zod@4.4.3)) + '@opentelemetry/api': + specifier: ^1.9.0 + version: 1.9.1 + '@opentelemetry/resources': + specifier: ^1.25.1 + version: 1.25.1(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-trace-base': + specifier: ^1.25.1 + version: 1.25.1(@opentelemetry/api@1.9.1) '@temporalio/client': specifier: workspace:* version: link:../../packages/client + '@temporalio/interceptors-opentelemetry': + specifier: workspace:* + version: link:../interceptors-opentelemetry '@temporalio/testing': specifier: workspace:* version: link:../../packages/testing @@ -9174,7 +9186,7 @@ snapshots: '@opentelemetry/resources@2.7.1(@opentelemetry/api@1.9.1)': dependencies: '@opentelemetry/api': 1.9.1 - '@opentelemetry/core': 2.7.1(@opentelemetry/api@1.9.0) + '@opentelemetry/core': 2.7.1(@opentelemetry/api@1.9.1) '@opentelemetry/semantic-conventions': 1.39.0 '@opentelemetry/sdk-logs@0.205.0(@opentelemetry/api@1.9.0)': @@ -9305,8 +9317,8 @@ snapshots: '@opentelemetry/sdk-trace-base@2.7.1(@opentelemetry/api@1.9.1)': dependencies: '@opentelemetry/api': 1.9.1 - '@opentelemetry/core': 2.7.1(@opentelemetry/api@1.9.0) - '@opentelemetry/resources': 2.7.1(@opentelemetry/api@1.9.0) + '@opentelemetry/core': 2.7.1(@opentelemetry/api@1.9.1) + '@opentelemetry/resources': 2.7.1(@opentelemetry/api@1.9.1) '@opentelemetry/semantic-conventions': 1.39.0 '@opentelemetry/sdk-trace-node@1.25.1(@opentelemetry/api@1.9.1)':