Refuse a different Activity under an AiSdkPlugin Activity name (breaking)

typescript/vercel-ai-sdk/README.md

@@ -34,4 +34,23 @@ 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 `<server>-listTools` and `<server>-callTool` for
+each MCP server. A different Activity with one of these names now fails with
+`Duplicate registration of Activity '<name>'`; 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, the same factory function and idle timeout, as
+when both are built from one options object. Two equal object literals are
+refused.
+
 Development details are in [typescript/README.md](../README.md).

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);
 

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]),
       };
     });
   }

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<object, readonly unknown[]>();
+
+/**
+ * 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<T extends object>(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<void>
+): Promise<never> | 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]);
+}

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<void>): Promise<void> {
+    // 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