diff --git a/typescript/google-adk/README.md b/typescript/google-adk/README.md index 767c19a8..1504d75f 100644 --- a/typescript/google-adk/README.md +++ b/typescript/google-adk/README.md @@ -685,6 +685,19 @@ 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, 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. - `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..8251b23b 100644 --- a/typescript/google-adk/src/__tests__/plugin.test.ts +++ b/typescript/google-adk/src/__tests__/plugin.test.ts @@ -7,12 +7,16 @@ import { createRequire } from 'node:module'; 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'; -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 +245,88 @@ 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))); + // 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) + .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..406ca3c5 --- /dev/null +++ b/typescript/google-adk/src/activity-sources.ts @@ -0,0 +1,67 @@ +/** + * 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 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. + 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]) && + 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 068b3d41..10f027f2 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,20 @@ 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, or with different + * `setActivityOptions` metadata. + */ + 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. *