From 98b108c22937b7be099bac760288cc1c077e7f47 Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Fri, 23 Oct 2026 00:07:56 -0700 Subject: [PATCH 1/2] Fail Worker creation when GoogleAdkPlugin would replace another Activity (breaking) The SDK's plugin merge keeps the last Activity of each name. A caller Activity named adk-invokeModel, adk-invokeModelStreaming or -listTools/-callTool/-listResources/-readResource, a second GoogleAdkPlugin with a different modelProvider or toolset factory, or another integration's MCP Activity with the same server name was replaced without a warning, and the Worker ran the wrong handler. GoogleAdkPlugin.configureWorker now throws a TypeError ("Duplicate registration of Activity ''") when one of its Activities would replace a different function. Each plugin Activity records the operation it was built for and its modelProvider or toolset factory, so another operation registered under the same name is refused, while the same plugin instance registered twice (for example on the NativeConnection and the Worker), or two instances built from the same options object, still start. The model Activities now read modelProvider once, when the plugin is constructed, so the provider they call is the one recorded for the check. Changing the options object after construction no longer changes the provider. They still call it as a method of the options object, as before, so a provider that reads `this` keeps working; that object is recorded with the provider, so two instances with one provider in two different options objects are refused. Breaking change: Workers with such a collision used to start and now fail in Worker.create. Rename or remove the other Activity or MCP server, or put every model provider and toolset in one GoogleAdkPlugin. Activity names, inputs and results do not change, so recorded histories replay unchanged. --- typescript/google-adk/README.md | 12 ++++ .../google-adk/src/__tests__/plugin.test.ts | 71 ++++++++++++++++++- typescript/google-adk/src/activities.ts | 13 +++- typescript/google-adk/src/activity-sources.ts | 49 +++++++++++++ typescript/google-adk/src/plugin.ts | 21 ++++-- 5 files changed, 157 insertions(+), 9 deletions(-) create mode 100644 typescript/google-adk/src/activity-sources.ts diff --git a/typescript/google-adk/README.md b/typescript/google-adk/README.md index 767c19a8..a4ed51ac 100644 --- a/typescript/google-adk/README.md +++ b/typescript/google-adk/README.md @@ -685,6 +685,18 @@ Cautions: - Register `GoogleAdkPlugin` on the Worker. Passing it directly to `Client` does not register the model/MCP Activities. If composing plugins, place observability and governance plugins before this one. +- The plugin registers Activities named `adk-invokeModel` and + `adk-invokeModelStreaming`, plus `-listTools`, `-callTool`, + `-listResources` and `-readResource` for each MCP toolset. If the + Worker already has a different Activity with one of these names (your own + Activity, another integration's MCP Activity, or a second `GoogleAdkPlugin` + with a different `modelProvider` or toolset factory), `Worker.create` fails + with `Duplicate registration of Activity ''`. Earlier versions kept the + last one without a warning. Rename the other Activity or MCP server, or put + every model provider and toolset in one `GoogleAdkPlugin`. The same instance + on the `NativeConnection` and the Worker, or two instances built from the + same options object, still work. `modelProvider` is read once, when the + plugin is constructed, and is called as a method of the options object. - Model calls use Temporal retries. The plugin disables nested GenAI SDK retries for model requests and honors `retry-after` where available. - `heartbeatTimeout` detects a dead worker, not a stalled call: when set, the model diff --git a/typescript/google-adk/src/__tests__/plugin.test.ts b/typescript/google-adk/src/__tests__/plugin.test.ts index 38e88f07..5ce35c7d 100644 --- a/typescript/google-adk/src/__tests__/plugin.test.ts +++ b/typescript/google-adk/src/__tests__/plugin.test.ts @@ -7,12 +7,15 @@ import { createRequire } from 'node:module'; import path from 'node:path'; import test from 'ava'; +import type { LlmResponse } from '@google/adk'; +import { MockActivityEnvironment } from '@temporalio/testing'; import type { BundleOptions, WorkerOptions } from '@temporalio/worker'; -import { GoogleAdkPlugin } from '../index'; +import { GoogleAdkPlugin, type GoogleAdkPluginOptions } from '../index'; import { interceptors as autoRouteInterceptors } from '../auto-route'; import { interceptors as polyfillInterceptors } from '../load-polyfills'; -import { mockMCPToolset } from '../testing'; +import type { InvokeModelArgs, ModelActivities } from '../model'; +import { fakeModelProvider, mockMCPToolset } from '../testing'; import { sandboxCompatResolveHook } from './helpers'; interface NamedPlugin { @@ -241,3 +244,67 @@ test('configureWorker registers model activities, plus the four MCP activities p t.is(typeof activities['weather-listResources'], 'function'); t.is(typeof activities['weather-readResource'], 'function'); }); + +test('configureWorker refuses to replace a different Activity with the same name', async (t) => { + const provider = fakeModelProvider(); + const weather = mockMCPToolset([]); + const base = { taskQueue: 'tq' } as WorkerOptions; + const duplicate = (name: string) => ({ + instanceOf: TypeError, + message: new RegExp(`^Duplicate registration of Activity '${name}': GoogleAdkPlugin would replace`), + }); + + // A caller Activity, another instance's model provider, or another instance's + // toolset factory under the same name is refused. + const callerActivities = { ...base, activities: { 'adk-invokeModel': async () => 'caller' } }; + t.throws(() => new GoogleAdkPlugin().configureWorker(callerActivities), duplicate('adk-invokeModel')); + const withProvider = new GoogleAdkPlugin({ modelProvider: provider }).configureWorker(base); + t.throws(() => new GoogleAdkPlugin().configureWorker(withProvider), duplicate('adk-invokeModel')); + const withToolset = new GoogleAdkPlugin({ mcpToolsets: { weather } }).configureWorker(base); + const otherToolset = new GoogleAdkPlugin({ mcpToolsets: { weather: mockMCPToolset([]) } }); + t.throws(() => otherToolset.configureWorker(withToolset), duplicate('weather-listTools')); + // Another operation built from the same provider, registered under this name. + const streaming = (withProvider.activities as Record)['adk-invokeModelStreaming']; + const aliased = { ...base, activities: { 'adk-invokeModel': streaming } }; + t.throws(() => new GoogleAdkPlugin({ modelProvider: provider }).configureWorker(aliased), duplicate('adk-invokeModel')); + // The provider is read once, at construction: changing the options object + // afterwards does not change what the first plugin registered. + const opts: GoogleAdkPluginOptions = { modelProvider: provider }; + const first = new GoogleAdkPlugin(opts); + opts.modelProvider = fakeModelProvider(); + t.throws(() => new GoogleAdkPlugin(opts).configureWorker(first.configureWorker(base)), duplicate('adk-invokeModel')); + // The provider is still called as a method of its options object, so the + // object is part of what the model Activities depend on: one provider method + // on two options objects is refused, as is one provider in two objects. + function viaDelegate(this: { delegate: GoogleAdkPluginOptions['modelProvider'] }, model: string) { + return this.delegate!(model); + } + const reply = (text: string) => fakeModelProvider([{ content: { role: 'model', parts: [{ text }] } }]); + const methodOpts = { delegate: reply('from delegate'), modelProvider: viaDelegate }; + const otherMethodOpts = { delegate: reply('from another delegate'), modelProvider: viaDelegate }; + t.throws( + () => new GoogleAdkPlugin(otherMethodOpts).configureWorker(new GoogleAdkPlugin(methodOpts).configureWorker(base)), + duplicate('adk-invokeModel') + ); + t.throws(() => new GoogleAdkPlugin({ modelProvider: provider }).configureWorker(withProvider), duplicate('adk-invokeModel')); + const invokeArgs = { model: 'm', request: { model: 'm', contents: [{ role: 'user', parts: [{ text: 'hi' }] }], config: {} } } as InvokeModelArgs; + const methodActivities = new GoogleAdkPlugin(methodOpts).configureWorker(base).activities as ModelActivities; + const responses = (await new MockActivityEnvironment().run(methodActivities['adk-invokeModel'], invokeArgs)) as LlmResponse[]; + t.is(responses[0]?.content?.parts?.[0]?.text, 'from delegate'); + + // configureWorker accepts the same instance twice (NativeConnection and + // Worker) and the same options object twice. + const plugin = new GoogleAdkPlugin({ modelProvider: provider, mcpToolsets: { weather } }); + t.notThrows(() => plugin.configureWorker(plugin.configureWorker(base))); + t.notThrows(() => new GoogleAdkPlugin().configureWorker(new GoogleAdkPlugin().configureWorker(base))); + t.notThrows(() => new GoogleAdkPlugin({ mcpToolsets: { weather } }).configureWorker(withToolset)); + const shared: GoogleAdkPluginOptions = { modelProvider: provider, mcpToolsets: { weather } }; + t.notThrows(() => new GoogleAdkPlugin(shared).configureWorker(new GoogleAdkPlugin(shared).configureWorker(base))); + t.notThrows(() => new GoogleAdkPlugin(methodOpts).configureWorker(new GoogleAdkPlugin(methodOpts).configureWorker(base))); + // Different toolset names from two instances are both kept. + const both = new GoogleAdkPlugin({ mcpToolsets: { news: mockMCPToolset([]) } }) + .configureWorker(withToolset) + .activities as Record; + t.is(typeof both['weather-callTool'], 'function'); + t.is(typeof both['news-callTool'], 'function'); +}); diff --git a/typescript/google-adk/src/activities.ts b/typescript/google-adk/src/activities.ts index d3afb22a..69a2be19 100644 --- a/typescript/google-adk/src/activities.ts +++ b/typescript/google-adk/src/activities.ts @@ -37,6 +37,7 @@ import { MCP_TOOL_NOT_FOUND_FAILURE_TYPE, MODEL_ERROR_FAILURE_TYPE, } from './error-types'; +import { withSources } from './activity-sources'; import type { InvokeModelArgs, InvokeModelStreamingArgs, ModelActivities, WireLlmRequest } from './model'; import type { MCPCallToolArgs, MCPReadResourceArgs, MCPResourceContents, MCPToolsetFactory } from './mcp'; @@ -89,9 +90,13 @@ export interface ModelActivitiesOptions { /** @internal */ export function createModelActivities(options: ModelActivitiesOptions = {}): ModelActivities { - const resolveModel = (model: string): BaseLlm => options.modelProvider?.(model) ?? LLMRegistry.newLlm(model); + // Read once, so the Activities use the provider recorded with them below. + // It is still called as a method of `options`, as before, so a provider + // that reads `this` keeps working. + const { modelProvider } = options; + const resolveModel = (model: string): BaseLlm => modelProvider?.call(options, model) ?? LLMRegistry.newLlm(model); - return { + const activities: ModelActivities = { async 'adk-invokeModel'(args: InvokeModelArgs): Promise { const abortSignal = ActivityContext.current().cancellationSignal; const stopHeartbeat = startAdaptiveHeartbeat(); @@ -144,6 +149,8 @@ export function createModelActivities(options: ModelActivitiesOptions = {}): Mod } }, }; + // The provider may read `this`, so its receiver is part of what they depend on. + return withSources(activities, modelProvider === undefined ? [] : [modelProvider, options]); } /** @@ -155,7 +162,7 @@ export function createMCPActivities( ): Record Promise> { const activities: Record Promise> = {}; for (const [name, factory] of Object.entries(toolsets)) { - Object.assign(activities, mcpActivitiesForName(name, factory)); + Object.assign(activities, withSources(mcpActivitiesForName(name, factory), [factory])); } return activities; } diff --git a/typescript/google-adk/src/activity-sources.ts b/typescript/google-adk/src/activity-sources.ts new file mode 100644 index 00000000..2c4faaa9 --- /dev/null +++ b/typescript/google-adk/src/activity-sources.ts @@ -0,0 +1,49 @@ +/** + * Lets the plugin refuse an Activity name that is already taken. The SDK's + * plugin merge (`{...existing, ...param}`) keeps the last Activity of each + * name, so without this check a plugin Activity silently replaces a different + * Activity with the same name. @internal + */ + +/** + * What each plugin Activity is: the name it was built under (its operation) + * followed by what it depends on. Two Activities with the same operation and + * the same dependencies do the same thing, so one may replace the other. + */ +const activitySources = new WeakMap(); + +/** + * Records, for every Activity in `activities`, its own key and `sources`, and + * returns `activities`. The key keeps one operation, registered under another + * operation's name, from passing as that operation. + */ +export function withSources(activities: T, sources: readonly unknown[]): T { + for (const [operation, fn] of Object.entries(activities)) activitySources.set(fn, [operation, ...sources]); + return activities; +} + +/** + * Throws if an Activity in `added` would replace a different Activity with the + * same name in `existing`. The same function, or one built for the same + * operation from the same sources, may replace itself: that is the same plugin + * instance, or one with the same options, registered twice (e.g. on the + * NativeConnection and the Worker). + */ +export function refuseReplacedActivities(plugin: string, existing: object | undefined, added: object): void { + // What the Worker registers: own enumerable keys with function values. + const registered = new Map(Object.entries(existing ?? {}).filter(([, fn]) => typeof fn === 'function')); + for (const [name, fn] of Object.entries(added)) { + const other = registered.get(name); + if (other === undefined || other === fn || sameSources(other, fn)) continue; + throw new TypeError( + `Duplicate registration of Activity '${name}': ${plugin} would replace a different Activity with this name. ` + + `Rename or remove the other registration, or give each ${plugin} on this Worker the same options.` + ); + } +} + +function sameSources(a: object, b: object): boolean { + const x = activitySources.get(a); + const y = activitySources.get(b); + return x !== undefined && y !== undefined && x.length === y.length && x.every((v, i) => v === y[i]); +} diff --git a/typescript/google-adk/src/plugin.ts b/typescript/google-adk/src/plugin.ts index 068b3d41..27ef7219 100644 --- a/typescript/google-adk/src/plugin.ts +++ b/typescript/google-adk/src/plugin.ts @@ -2,9 +2,10 @@ import { builtinModules, createRequire } from 'node:module'; import type { BaseLlm } from '@google/adk'; import { SimplePlugin } from '@temporalio/plugin'; -import type { BundleOptions } from '@temporalio/worker'; +import type { BundleOptions, WorkerOptions } from '@temporalio/worker'; import { createModelActivities, createMCPActivities } from './activities'; +import { refuseReplacedActivities } from './activity-sources'; import { type MCPToolsetFactory } from './mcp'; /** The webpack `Configuration` object the bundler hands to `webpackConfigHook`. */ @@ -600,9 +601,8 @@ export class GoogleAdkPlugin extends SimplePlugin { constructor(options: GoogleAdkPluginOptions = {}) { super({ name: 'google.AdkPlugin', - // Object-keyed activities dedupe by name in the TS SDK's plugin merge - // (`{...existing, ...param}`), so double-registration (e.g. the plugin - // passed to both Client and Worker) is tolerated rather than a crash. + // `configureWorker` refuses to let these replace a different Activity + // with the same name (see `refuseReplacedActivities`). activities: { ...createModelActivities(options), ...createMCPActivities(options.mcpToolsets), @@ -611,6 +611,19 @@ export class GoogleAdkPlugin extends SimplePlugin { this.autoRouteModels = options.autoRouteModels ?? true; } + /** + * Adds the model and MCP Activities to the Worker. + * + * @throws TypeError if the Worker already has a different Activity with one + * of these names: a caller Activity, another integration's Activity, or a + * `GoogleAdkPlugin` with a different `modelProvider` (or the same one in a + * different options object) or MCP toolset factory. + */ + override configureWorker(options: WorkerOptions): WorkerOptions { + refuseReplacedActivities('GoogleAdkPlugin', options.activities, this.options.activities as object); + return super.configureWorker(options); + } + /** * Makes the `@google/adk` agent loop bundle into the Workflow sandbox. * From 56615d981bf2de80ea333062b02548514fe7f901 Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Wed, 28 Oct 2026 17:20:35 -0700 Subject: [PATCH 2/2] Address review findings --- typescript/google-adk/README.md | 3 ++- .../google-adk/src/__tests__/plugin.test.ts | 22 ++++++++++++++++ typescript/google-adk/src/activity-sources.ts | 26 ++++++++++++++++--- typescript/google-adk/src/plugin.ts | 3 ++- 4 files changed, 48 insertions(+), 6 deletions(-) diff --git a/typescript/google-adk/README.md b/typescript/google-adk/README.md index a4ed51ac..1504d75f 100644 --- a/typescript/google-adk/README.md +++ b/typescript/google-adk/README.md @@ -695,7 +695,8 @@ Cautions: last one without a warning. Rename the other Activity or MCP server, or put every model provider and toolset in one `GoogleAdkPlugin`. The same instance on the `NativeConnection` and the Worker, or two instances built from the - same options object, still work. `modelProvider` is read once, when the + same options object, still work, unless `setActivityOptions` gave their + Activities different metadata objects. `modelProvider` is read once, when the plugin is constructed, and is called as a method of the options object. - Model calls use Temporal retries. The plugin disables nested GenAI SDK retries for model requests and honors `retry-after` where available. diff --git a/typescript/google-adk/src/__tests__/plugin.test.ts b/typescript/google-adk/src/__tests__/plugin.test.ts index 5ce35c7d..8251b23b 100644 --- a/typescript/google-adk/src/__tests__/plugin.test.ts +++ b/typescript/google-adk/src/__tests__/plugin.test.ts @@ -8,6 +8,7 @@ import path from 'node:path'; import test from 'ava'; import type { LlmResponse } from '@google/adk'; +import { setActivityOptions } from '@temporalio/activity'; import { MockActivityEnvironment } from '@temporalio/testing'; import type { BundleOptions, WorkerOptions } from '@temporalio/worker'; @@ -301,6 +302,27 @@ test('configureWorker refuses to replace a different Activity with the same name const shared: GoogleAdkPluginOptions = { modelProvider: provider, mcpToolsets: { weather } }; t.notThrows(() => new GoogleAdkPlugin(shared).configureWorker(new GoogleAdkPlugin(shared).configureWorker(base))); t.notThrows(() => new GoogleAdkPlugin(methodOpts).configureWorker(new GoogleAdkPlugin(methodOpts).configureWorker(base))); + // The Worker decodes arguments and encodes results with the metadata + // `setActivityOptions` attaches, so same-source Activities must also carry the + // same metadata object, in either order, for the model and MCP Activities. + const annotate = (plugin: GoogleAdkPlugin, name: string, metadata = { typeInfo: { inputTypes: [] } }) => { + // configureWorker returns the plugin's own Activity functions. + const activities = plugin.configureWorker(base).activities as Record Promise>; + setActivityOptions(metadata, activities[name]!); + return plugin; + }; + const sharedPlugin = () => new GoogleAdkPlugin(shared); + const annotatedFirst = (name: string) => annotate(sharedPlugin(), name).configureWorker(base); + t.throws(() => sharedPlugin().configureWorker(annotatedFirst('adk-invokeModel')), duplicate('adk-invokeModel')); + t.throws(() => sharedPlugin().configureWorker(annotatedFirst('weather-callTool')), duplicate('weather-callTool')); + const annotatedLast = annotate(sharedPlugin(), 'adk-invokeModel'); + t.throws(() => annotatedLast.configureWorker(sharedPlugin().configureWorker(base)), duplicate('adk-invokeModel')); + const otherMetadata = annotate(sharedPlugin(), 'adk-invokeModel'); + t.throws(() => otherMetadata.configureWorker(annotatedFirst('adk-invokeModel')), duplicate('adk-invokeModel')); + const metadata = { typeInfo: { inputTypes: [] } }; + const sameMetadata = annotate(sharedPlugin(), 'adk-invokeModel', metadata); + const firstWithMetadata = annotate(sharedPlugin(), 'adk-invokeModel', metadata).configureWorker(base); + t.notThrows(() => sameMetadata.configureWorker(firstWithMetadata)); // Different toolset names from two instances are both kept. const both = new GoogleAdkPlugin({ mcpToolsets: { news: mockMCPToolset([]) } }) .configureWorker(withToolset) diff --git a/typescript/google-adk/src/activity-sources.ts b/typescript/google-adk/src/activity-sources.ts index 2c4faaa9..406ca3c5 100644 --- a/typescript/google-adk/src/activity-sources.ts +++ b/typescript/google-adk/src/activity-sources.ts @@ -25,9 +25,9 @@ export function withSources(activities: T, sources: readonly u /** * Throws if an Activity in `added` would replace a different Activity with the * same name in `existing`. The same function, or one built for the same - * operation from the same sources, may replace itself: that is the same plugin - * instance, or one with the same options, registered twice (e.g. on the - * NativeConnection and the Worker). + * operation from the same sources with the same Activity metadata, may replace + * itself: that is the same plugin instance, or one with the same options, + * registered twice (e.g. on the NativeConnection and the Worker). */ export function refuseReplacedActivities(plugin: string, existing: object | undefined, added: object): void { // What the Worker registers: own enumerable keys with function values. @@ -45,5 +45,23 @@ export function refuseReplacedActivities(plugin: string, existing: object | unde function sameSources(a: object, b: object): boolean { const x = activitySources.get(a); const y = activitySources.get(b); - return x !== undefined && y !== undefined && x.length === y.length && x.every((v, i) => v === y[i]); + return ( + x !== undefined && + y !== undefined && + x.length === y.length && + x.every((v, i) => v === y[i]) && + activityOptions(a) === activityOptions(b) + ); +} + +/** + * The metadata `setActivityOptions` attached to `fn`, read the way the Worker + * reads it (an own object-valued `activityDefinitionOptions`). Its `typeInfo` + * decodes the Activity's arguments and encodes its result, so a function with + * other metadata is a different Activity even when its sources match. + */ +function activityOptions(fn: object): object | undefined { + if (!Object.hasOwn(fn, 'activityDefinitionOptions')) return undefined; + const options: unknown = (fn as { activityDefinitionOptions?: unknown }).activityDefinitionOptions; + return typeof options === 'object' && options !== null ? options : undefined; } diff --git a/typescript/google-adk/src/plugin.ts b/typescript/google-adk/src/plugin.ts index 27ef7219..10f027f2 100644 --- a/typescript/google-adk/src/plugin.ts +++ b/typescript/google-adk/src/plugin.ts @@ -617,7 +617,8 @@ export class GoogleAdkPlugin extends SimplePlugin { * @throws TypeError if the Worker already has a different Activity with one * of these names: a caller Activity, another integration's Activity, or a * `GoogleAdkPlugin` with a different `modelProvider` (or the same one in a - * different options object) or MCP toolset factory. + * different options object) or MCP toolset factory, or with different + * `setActivityOptions` metadata. */ override configureWorker(options: WorkerOptions): WorkerOptions { refuseReplacedActivities('GoogleAdkPlugin', options.activities, this.options.activities as object);