diff --git a/typescript/vercel-ai-sdk/README.md b/typescript/vercel-ai-sdk/README.md index 25334e99..d6626997 100644 --- a/typescript/vercel-ai-sdk/README.md +++ b/typescript/vercel-ai-sdk/README.md @@ -34,4 +34,25 @@ for model, tool, streaming and MCP configuration. That guide currently uses the SDK-owned `@temporalio/ai-sdk` name; this repository owns the new `@temporalio/vercel-ai-sdk` coordinate. +The plugin registers Activities named `invokeModel`, `invokeModelStreaming` and +`invokeEmbeddingModel`, plus `-listTools` and `-callTool` for +each MCP server. A different Activity with one of these names now fails with +`Duplicate registration of Activity ''`; earlier versions kept the last +one without a warning. If it is registered before `AiSdkPlugin` (in the Worker's +`activities` or by an earlier plugin), `Worker.create` fails. If a plugin +applied after `AiSdkPlugin` replaces one, `worker.run()` fails before the Worker +polls. `Worker.create` applies the Worker's `plugins` in order, then the +`NativeConnection`'s. + +To fix a collision with your own Activity or an MCP server name, rename one of +them. `StrandsPlugin` (`@temporalio/strands-agents`) also registers +`invokeModel` and `invokeModelStreaming`, and neither plugin can rename them, so +run the two integrations on separate Workers and task queues. Two `AiSdkPlugin` +instances count as the same only when they share the `modelProvider` object +and, for each MCP server name both register, the same factory function and idle +timeout. A provider or factory created separately for each instance is refused +even when its settings are equal, for example `createOpenAI({ apiKey })` called +once per plugin, or an `async () => ...` factory written inline in each options +object. + Development details are in [typescript/README.md](../README.md). diff --git a/typescript/vercel-ai-sdk/src/__tests__/test-ai-sdk.ts b/typescript/vercel-ai-sdk/src/__tests__/test-ai-sdk.ts index b4c84993..7ef11335 100644 --- a/typescript/vercel-ai-sdk/src/__tests__/test-ai-sdk.ts +++ b/typescript/vercel-ai-sdk/src/__tests__/test-ai-sdk.ts @@ -47,8 +47,9 @@ import { OpenTelemetryWorkflowClientInterceptor, } from '@temporalio/interceptors-opentelemetry'; import { workflowInterceptorModules } from '@temporalio/testing'; -import { bundleWorkflowCode, DefaultLogger, Worker as CoreWorker } from '@temporalio/worker'; -import type { InjectedSinks } from '@temporalio/worker'; +import { SimplePlugin } from '@temporalio/plugin'; +import { bundleWorkflowCode, DefaultLogger, NativeConnection, Worker as CoreWorker } from '@temporalio/worker'; +import type { InjectedSinks, WorkerOptions } from '@temporalio/worker'; import type { BaseContext } from '@temporalio/test-helpers'; import { test as anyTest, @@ -962,6 +963,88 @@ const createTestMcpClient = async () => { return createMCPClient({ transport: clientTransport }); }; +test('configureWorker refuses to replace a different Activity with the same name', (t) => { + const provider = new TestProvider(helloWorkflowGenerator()); + const files = async () => ({}) as any; + const base = { taskQueue: 'tq' } as WorkerOptions; + const duplicate = (name: string) => ({ + instanceOf: TypeError, + message: new RegExp(`^Duplicate registration of Activity '${name}': AiSdkPlugin would replace`), + }); + + // A caller Activity, another instance's provider, or another instance's MCP + // client factory under the same server name is refused. + const callerActivities = { ...base, activities: { invokeModel: async () => 'caller' } }; + t.throws(() => new AiSdkPlugin({ modelProvider: provider }).configureWorker(callerActivities), duplicate('invokeModel')); + const withMcp = new AiSdkPlugin({ modelProvider: provider, mcpClientFactories: { files } }).configureWorker(base); + const otherProvider = new AiSdkPlugin({ modelProvider: new TestProvider(helloWorkflowGenerator()) }); + t.throws(() => otherProvider.configureWorker(withMcp), duplicate('invokeModel')); + const otherFactory = new AiSdkPlugin({ modelProvider: provider, mcpClientFactories: { files: async () => ({}) as any } }); + t.throws(() => otherFactory.configureWorker(withMcp), duplicate('files-listTools')); + // Another operation built from the same provider, registered under this name. + const aliased = { ...base, activities: { invokeModel: (createActivities(provider) as any).invokeEmbeddingModel } }; + t.throws(() => new AiSdkPlugin({ modelProvider: provider }).configureWorker(aliased), duplicate('invokeModel')); + + // configureWorker accepts the same instance twice (NativeConnection and + // Worker), the same options twice, and createActivities with the same + // provider. With workflowsPath, Worker.create still rejects a repeated + // AiSdkPlugin in the bundler: each one prepends its polyfill entry. + const plugin = new AiSdkPlugin({ modelProvider: provider, mcpClientFactories: { files } }); + t.notThrows(() => plugin.configureWorker(plugin.configureWorker(base))); + t.notThrows(() => new AiSdkPlugin({ modelProvider: provider, mcpClientFactories: { files } }).configureWorker(withMcp)); + const callerSet = { ...base, activities: createActivities(provider) }; + t.notThrows(() => new AiSdkPlugin({ modelProvider: provider }).configureWorker(callerSet)); +}); + +test('runWorker refuses an Activity that a later plugin replaced, before polling', async (t) => { + const connection = await NativeConnection.connect({ address: t.context.env.address }); + // StrandsPlugin also registers invokeModel; this plugin stands in for it. + const later = new SimplePlugin({ name: 'later', activities: { invokeModel: async () => 'other' } }); + const worker = await Worker.create({ + connection, + namespace: t.context.env.namespace, + taskQueue: 'test-ai-later-plugin', + plugins: [new AiSdkPlugin({ modelProvider: new TestProvider(helloWorkflowGenerator()) }), later], + }); + await t.throwsAsync(worker.run(), { + instanceOf: TypeError, + message: /^Duplicate registration of Activity 'invokeModel': a plugin applied after AiSdkPlugin replaced/, + }); + t.is(worker.getState(), 'STOPPED'); + // Throws if the Worker still holds the connection. + await connection.close(); +}); + +test('runWorker starts the Worker at once, so runUntil stops it after its callback', async (t) => { + const connection = await NativeConnection.connect({ address: t.context.env.address }); + // runUntil stops the Worker after its callback only if the Worker is already + // running. A callback that throws synchronously finishes before run() yields. + const alone = await Worker.create({ + connection, + namespace: t.context.env.namespace, + taskQueue: 'test-ai-run-until-throws', + plugins: [new AiSdkPlugin({ modelProvider: new TestProvider(helloWorkflowGenerator()) })], + }); + await t.throwsAsync( + alone.runUntil(() => { + throw new Error('boom'); + }), + { message: 'boom' } + ); + t.is(alone.getState(), 'STOPPED'); + // Each plugin's check runs before the Worker starts; none may yield. + const options = { modelProvider: new TestProvider(helloWorkflowGenerator()) }; + const two = await Worker.create({ + connection, + namespace: t.context.env.namespace, + taskQueue: 'test-ai-run-until-two', + plugins: [new AiSdkPlugin(options), new AiSdkPlugin(options)], + }); + t.is(await two.runUntil(() => Promise.resolve('done')), 'done'); + t.is(two.getState(), 'STOPPED'); + await connection.close(); +}); + test('MCP tool schema survives activity serialization', async (t) => { const { createWorker, executeWorkflow } = helpers(t); diff --git a/typescript/vercel-ai-sdk/src/activities.ts b/typescript/vercel-ai-sdk/src/activities.ts index 6d67b6a2..e28ebc57 100644 --- a/typescript/vercel-ai-sdk/src/activities.ts +++ b/typescript/vercel-ai-sdk/src/activities.ts @@ -18,6 +18,7 @@ import { Context } from '@temporalio/activity'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; import { msOptionalToNumber, type Duration } from '@temporalio/common/lib/time'; import type { McpClientFactories, McpClientFactory } from './mcp'; +import { withSources } from './activity-sources'; const encoder = new TextEncoder(); @@ -396,12 +397,13 @@ export function createActivities( }); }, }; + withSources(activities, [provider]); if (mcpClientFactories !== undefined) { const idleMs = msOptionalToNumber(options?.mcpConnectionIdleTimeout) ?? MCP_CONNECTION_IDLE_MS; Object.entries(mcpClientFactories).forEach(([name, func]) => { activities = { ...activities, - ...activitiesForName(name, func, idleMs), + ...withSources(activitiesForName(name, func, idleMs), [func, idleMs]), }; }); } diff --git a/typescript/vercel-ai-sdk/src/activity-sources.ts b/typescript/vercel-ai-sdk/src/activity-sources.ts new file mode 100644 index 00000000..0646356f --- /dev/null +++ b/typescript/vercel-ai-sdk/src/activity-sources.ts @@ -0,0 +1,92 @@ +/** + * 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, or a later plugin silently replaces it. @internal + */ +import type { Worker } from '@temporalio/worker'; + +/** + * 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. ` + + remedy(plugin) + ); + } +} + +/** + * For `runWorker`, which runs after every plugin's `configureWorker`: if a + * plugin applied after this one replaced an Activity in `added` with a + * different one, returns a promise that rejects with a `TypeError`. Before + * rejecting, it stops the Worker so that it never polls, and lets `next` + * release it as `run()` does, so `connection.close()` still works. + * + * Otherwise it returns `undefined` without yielding, so the caller can start + * the Worker in the same tick. `Worker.runUntil` calls its callback right + * after `run()` and stops the Worker afterwards only if it is already running; + * a Worker started later polls until something else stops it. + */ +export function refuseLaterReplacement( + plugin: string, + worker: Worker, + added: object, + next: (worker: Worker) => Promise +): Promise | undefined { + for (const [name, fn] of Object.entries(added)) { + const kept = worker.options.activities.get(name); + if (kept === undefined || kept === fn || sameSources(kept, fn)) continue; + const error = new TypeError( + `Duplicate registration of Activity '${name}': a plugin applied after ${plugin} replaced its Activity with this name. ` + + remedy(plugin) + ); + // There is no public way to stop a Worker that has not run yet. + (worker as unknown as { nativeWorker: { initiateShutdown(): void } }).nativeWorker.initiateShutdown(); + const fail = (): never => { + throw error; + }; + return next(worker).then(fail, fail); + } + return undefined; +} + +function remedy(plugin: string): string { + return ( + 'Rename or remove the other Activity, run the other integration on its own Worker and task queue, ' + + `or use one ${plugin} instance on this Worker.` + ); +} + +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/vercel-ai-sdk/src/plugin.ts b/typescript/vercel-ai-sdk/src/plugin.ts index 2397e233..33ad7b41 100644 --- a/typescript/vercel-ai-sdk/src/plugin.ts +++ b/typescript/vercel-ai-sdk/src/plugin.ts @@ -1,8 +1,9 @@ import type { ProviderV4 } from '@ai-sdk/provider'; import { SimplePlugin } from '@temporalio/plugin'; -import type { BundleOptions } from '@temporalio/worker'; +import type { BundleOptions, Worker, WorkerOptions } from '@temporalio/worker'; import type { Duration } from '@temporalio/common/lib/time'; import { createActivities, _evictMcpConnectionsForServer } from './activities'; +import { refuseLaterReplacement, refuseReplacedActivities } from './activity-sources'; import type { McpClientFactories } from './mcp'; /** @@ -60,6 +61,35 @@ export class AiSdkPlugin extends SimplePlugin { }); } + /** + * 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 an + * `AiSdkPlugin` with a different `modelProvider`, MCP client factory or + * `mcpConnectionIdleTimeout`. + */ + override configureWorker(options: WorkerOptions): WorkerOptions { + refuseReplacedActivities('AiSdkPlugin', options.activities, this.options.activities as object); + return super.configureWorker(options); + } + + /** + * Runs the Worker. + * + * @throws TypeError, before the Worker polls, if a plugin applied after this + * one replaced one of its Activities with a different one. `Worker.create` + * applies the Worker's plugins in order, then the `NativeConnection`'s. + */ + override runWorker(worker: Worker, next: (worker: Worker) => Promise): Promise { + // Not async: when the check passes, the Worker starts in this tick, as + // `Worker.runUntil` needs (see `refuseLaterReplacement`). + return ( + refuseLaterReplacement('AiSdkPlugin', worker, this.options.activities as object, next) ?? + super.runWorker(worker, next) + ); + } + override configureBundler(options: BundleOptions): BundleOptions { // Prepend a polyfill-installer module to the webpack `entry` array so it // evaluates before any other module in the Workflow bundle. Webpack 5