Breaking: fail the Worker when an OpenAI Agents plugin Activity would replace another Activity or be replaced by one

typescript/openai-agents/README.md

@@ -266,6 +266,10 @@ The Worker plugin does five things:
 - Adds Workflow, Activity, and Nexus interceptors for trace propagation.
 - Installs Workflow bundle polyfills required by the OpenAI Agents SDK.
 
+The model Activities are named `invokeModelActivity` and `invokeModelStreamActivity`. Provider Activity names start with the provider's name, such as `<name>-stateless-call-tool`. 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 `OpenAIAgentsPlugin` (in the Worker's `activities` or by an earlier plugin), `Worker.create` fails. If a plugin applied after `OpenAIAgentsPlugin` replaces one, `worker.run()` fails before the Worker polls. `Worker.create` applies the Worker's `plugins` in order, then the `NativeConnection`'s. A plugin applied after `OpenAIAgentsPlugin` that wraps these Activities counts as replacing them; wrap Activity execution with an Activity interceptor (`interceptors.activity`) instead.
+
+Another `OpenAIAgentsPlugin`'s Activity counts as the same only when it was built from the same objects: the same `modelProvider` and `hostedToolCredentials` for the model Activities, and the same provider for a provider's Activities. So one instance on both the `NativeConnection` and the Worker, or two instances built from one options object, still start, while two instances with equal but separate providers fail.
+
 The Client plugin injects the OpenAI Agents config header into Workflow-starting calls. Attach one `OpenAIAgentsPlugin` instance per Client or Connection configuration. If multiple instances are chained, the last config header wins.
 
 `modelParams` controls scheduling for the model Activity:

typescript/openai-agents/src/__tests__/plugin.test.ts

@@ -0,0 +1,73 @@
+import test from 'ava';
+import type { NativeConnection, WorkerOptions } from '@temporalio/worker';
+import { StatelessMCPServerProvider } from '../worker/mcp-provider';
+import { StatefulMCPServerProvider } from '../worker/stateful-mcp-provider';
+import { OpenAIAgentsPlugin } from '../worker/plugin';
+import { FakeModelProvider } from './stubs/openai-agents';
+
+type Activities = Record<string, unknown>;
+
+const callerActivity = async () => 'caller';
+
+const tail =
+  'or remove the other Activity, run the other integration on its own Worker and task queue, or use one ' +
+  'OpenAIAgentsPlugin instance on this Worker.';
+
+test('configureWorker refuses to replace a different Activity with the same name', (t) => {
+  const options = {
+    modelProvider: new FakeModelProvider([]),
+    mcpServerProviders: [
+      new StatelessMCPServerProvider('docs', () => {
+        throw new Error('not called');
+      }),
+      new StatefulMCPServerProvider(
+        'notes',
+        () => {
+          throw new Error('not called');
+        },
+        {} as NativeConnection
+      ),
+    ],
+  };
+  const plugin = new OpenAIAgentsPlugin(options);
+
+  const refusals: Record<string, string> = {
+    invokeModelActivity:
+      "Duplicate registration of Activity 'invokeModelActivity': OpenAIAgentsPlugin would replace a different " +
+      `Activity with this name. Rename ${tail}`,
+    'docs-stateless-call-tool':
+      "Duplicate registration of Activity 'docs-stateless-call-tool': OpenAIAgentsPlugin provider 'docs' would " +
+      `replace a different Activity with this name. Rename the provider, rename ${tail}`,
+    // The name the caller passed, without the stateful provider's internal suffix.
+    'notes-stateful-server-session':
+      "Duplicate registration of Activity 'notes-stateful-server-session': OpenAIAgentsPlugin provider 'notes' " +
+      `would replace a different Activity with this name. Rename the provider, rename ${tail}`,
+  };
+  for (const [name, message] of Object.entries(refusals)) {
+    t.throws(
+      () => plugin.configureWorker({ taskQueue: 'test', activities: { [name]: callerActivity } } as WorkerOptions),
+      { instanceOf: TypeError, message }
+    );
+  }
+
+  // Caller Activities with other names stay beside the plugin's.
+  const once = plugin.configureWorker({ taskQueue: 'test', activities: { lookUp: callerActivity } } as WorkerOptions);
+  t.is((once.activities as Activities).lookUp, callerActivity);
+
+  // Worker.create runs configureWorker twice when one instance is on both the NativeConnection
+  // and the Worker.
+  const twice = plugin.configureWorker(once);
+  t.is((twice.activities as Activities).invokeModelActivity, (once.activities as Activities).invokeModelActivity);
+
+  // A second instance built from the same options is the same plugin.
+  const same = new OpenAIAgentsPlugin(options).configureWorker(once);
+  t.is((same.activities as Activities).lookUp, callerActivity);
+
+  // A second instance with another model provider would replace the first one's Activities.
+  t.throws(() => new OpenAIAgentsPlugin({ modelProvider: new FakeModelProvider([]) }).configureWorker(once), {
+    instanceOf: TypeError,
+    message:
+      "Duplicate registration of Activity 'invokeModelActivity': OpenAIAgentsPlugin would replace a different " +
+      `Activity with this name. Rename ${tail}`,
+  });
+});

typescript/openai-agents/src/__tests__/test-openai-agents-comprehensive.ts

@@ -11,7 +11,8 @@ import { withTrace, withCustomSpan } from '@openai/agents-core';
 import { Client, WorkflowFailedError } from '@temporalio/client';
 import { ApplicationFailure } from '@temporalio/common';
 import type { WorkflowBundleWithSourceMap } from '@temporalio/worker';
-import { bundleWorkflowCode } from '@temporalio/worker';
+import { bundleWorkflowCode, NativeConnection } from '@temporalio/worker';
+import { SimplePlugin } from '@temporalio/plugin';
 import type { TestWorkflowEnvironment } from '@temporalio/test-helpers';
 import { Worker, createTestWorkflowEnvironment } from '@temporalio/test-helpers';
 import {
@@ -929,3 +930,67 @@ test('Stateful MCP dedicated-worker failure produces DedicatedWorkerFailure', as
     });
   });
 });
+
+test('runWorker refuses an Activity a later plugin replaced, and starts the Worker at once', async (t) => {
+  await withTestEnv(async (env) => {
+    const replaced =
+      /^Duplicate registration of Activity 'invokeModelActivity': a plugin applied after OpenAIAgentsPlugin replaced/;
+    const connection = await NativeConnection.connect({ address: env.address });
+    const later = new SimplePlugin({ name: 'later', activities: { invokeModelActivity: async () => 'other' } });
+    const worker = await Worker.create({
+      connection,
+      namespace: env.namespace,
+      plugins: [new OpenAIAgentsPlugin({ modelProvider: new FakeModelProvider([]) }), later],
+      workflowBundle: sharedWorkflowBundle,
+      taskQueue: `later-plugin-${randomUUID()}`,
+    });
+    await t.throwsAsync(worker.run(), { instanceOf: TypeError, message: replaced });
+    t.is(worker.getState(), 'STOPPED');
+
+    // One instance on the connection and the Worker, with the later plugin on the connection: both of
+    // the instance's checks run, and the Worker must still stop cleanly.
+    const plugin = new OpenAIAgentsPlugin({ modelProvider: new FakeModelProvider([]) });
+    const shared = await NativeConnection.connect({ address: env.address, plugins: [plugin, later] });
+    const twice = await Worker.create({
+      connection: shared,
+      namespace: env.namespace,
+      plugins: [plugin],
+      workflowBundle: sharedWorkflowBundle,
+      taskQueue: `later-plugin-twice-${randomUUID()}`,
+    });
+    await t.throwsAsync(twice.run(), { instanceOf: TypeError, message: replaced });
+    t.is(twice.getState(), 'STOPPED');
+    await shared.close();
+
+    // runUntil stops the Worker after its callback only if the Worker is already running, so the
+    // check must not yield before it starts: not for a callback that throws at once, and not for
+    // two instances that both check.
+    const options = { modelProvider: new FakeModelProvider([]) };
+    const alone = await Worker.create({
+      connection,
+      namespace: env.namespace,
+      plugins: [new OpenAIAgentsPlugin(options)],
+      workflowBundle: sharedWorkflowBundle,
+      taskQueue: `run-until-throws-${randomUUID()}`,
+    });
+    await t.throwsAsync(
+      alone.runUntil(() => {
+        throw new Error('boom');
+      }),
+      { message: 'boom' }
+    );
+    t.is(alone.getState(), 'STOPPED');
+    const two = await Worker.create({
+      connection,
+      namespace: env.namespace,
+      plugins: [new OpenAIAgentsPlugin(options), new OpenAIAgentsPlugin(options)],
+      workflowBundle: sharedWorkflowBundle,
+      taskQueue: `run-until-two-${randomUUID()}`,
+    });
+    t.is(await two.runUntil(() => Promise.resolve('done')), 'done');
+    t.is(two.getState(), 'STOPPED');
+
+    // Throws if a Worker still holds the connection.
+    await connection.close();
+  });
+});

typescript/openai-agents/src/worker/plugin.ts

@@ -5,9 +5,10 @@
 import { trace } from '@opentelemetry/api';
 import type { ModelProvider } from '@openai/agents-core';
 import { SimplePlugin } from '@temporalio/plugin';
-import type { BundleOptions, InjectedSinks, ReplayWorkerOptions, WorkerOptions } from '@temporalio/worker';
+import type { BundleOptions, InjectedSinks, ReplayWorkerOptions, Worker, WorkerOptions } from '@temporalio/worker';
 import { OpenAIAgentsTraceClientInterceptor } from '../client/trace-interceptor';
 import { AGENT_TRACING_SINK_NAME, type AgentTracingSinks } from '../common/agent-sink-types';
+import { MCP_STATEFUL_SUFFIX } from '../common/mcp-types';
 import type { SerializableModelActivityOptions } from '../common/model-activity-options';
 import { isReplaySafeTracerProvider } from '../common/tracing-bridge';
 import { createModelActivity } from './activities';
@@ -16,7 +17,7 @@ import { makeAgentTracingSink } from './agent-sink-bridge';
 import type { HostedToolCredentialsResolver } from './hosted-tool-credentials';
 import type { StatelessMCPServerProvider } from './mcp-provider';
 import type { SandboxClientProvider } from './sandbox-provider';
-import type { StatefulMCPServerProvider } from './stateful-mcp-provider';
+import { StatefulMCPServerProvider } from './stateful-mcp-provider';
 import {
   OpenAIAgentsTraceActivityInboundInterceptor,
   OpenAIAgentsTraceNexusInboundInterceptor,
@@ -49,6 +50,18 @@ function resolveAgentsCoreAsCommonJs(compiler: WebpackCompiler): void {
   });
 }
 
+// What each Activity an OpenAIAgentsPlugin builds was built from: its name, then what it depends on.
+// configureWorker lets one of these replace another with the same entries: that is the same
+// instance, or one built from the same options, registered twice (e.g. on the NativeConnection
+// and the Worker).
+const activitySources = new WeakMap<object, readonly unknown[]>();
+
+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]);
+}
+
 /**
  * Options controlling trace interceptor behavior on both the Activity side
  * (directly) and the Workflow side (via header propagation).
@@ -122,10 +135,15 @@ export interface OpenAIAgentsPluginOptions {
  * @experimental The OpenAI Agents plugin is an experimental feature; APIs may change without notice.
  */
 export class OpenAIAgentsPlugin extends SimplePlugin {
+  // The provider each provider Activity comes from, by Activity name, for the error message.
+  private readonly activityProviders: ReadonlyMap<string, string>;
   private readonly tracingApiKey?: TracingApiKeyResolver;
 
   constructor(options: OpenAIAgentsPluginOptions) {
     const modelActivities = createModelActivity(options.modelProvider, options.hostedToolCredentials);
+    for (const [name, fn] of Object.entries(modelActivities)) {
+      activitySources.set(fn, [name, options.modelProvider, options.hostedToolCredentials]);
+    }
 
     let allActivities: Record<string, (...args: any[]) => Promise<any>> = { ...modelActivities };
 
@@ -140,14 +158,23 @@ export class OpenAIAgentsPlugin extends SimplePlugin {
       ...(options.sandboxClientProviders ?? []),
     ];
     const seenNames = new Set<string>();
+    const activityProviders = new Map<string, string>();
     for (const provider of providers) {
       if (seenNames.has(provider.name)) {
         throw new Error(
           `Duplicate provider name: '${provider.name}'. Each MCP server and sandbox client provider must have a unique name.`
         );
       }
       seenNames.add(provider.name);
-      allActivities = { ...allActivities, ...provider._getActivities() };
+      // A stateful MCP server provider adds a suffix to the name the caller passed.
+      const shownName =
+        provider instanceof StatefulMCPServerProvider ? provider.name.slice(0, -MCP_STATEFUL_SUFFIX.length) : provider.name;
+      const providerActivities = provider._getActivities();
+      for (const [name, fn] of Object.entries(providerActivities)) {
+        activitySources.set(fn, [name, provider]);
+        activityProviders.set(name, shownName);
+      }
+      allActivities = { ...allActivities, ...providerActivities };
     }
 
     const activityInterceptorOptions: OpenAIAgentsTraceInterceptorOptions = {
@@ -192,13 +219,57 @@ export class OpenAIAgentsPlugin extends SimplePlugin {
         ],
       },
     });
+    this.activityProviders = activityProviders;
     this.tracingApiKey = options.tracingApiKey;
   }
 
+  /**
+   * Adds the plugin's Activities, interceptors and sinks 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 one from an `OpenAIAgentsPlugin` built from
+   * a different model provider, credentials resolver, or MCP server or sandbox client provider.
+   * Activities that a plugin applied later replaces are refused by {@link runWorker}.
+   */
   override configureWorker(options: WorkerOptions): WorkerOptions {
+    const existing = options.activities as Record<string, unknown> | undefined;
+    for (const [name, fn] of Object.entries(this.options.activities as object)) {
+      const other = existing && Object.prototype.propertyIsEnumerable.call(existing, name) ? existing[name] : undefined;
+      if (typeof other === 'function' && other !== fn && !sameSources(other, fn)) {
+        const provider = this.activityProviders.get(name);
+        const source = provider === undefined ? 'OpenAIAgentsPlugin' : `OpenAIAgentsPlugin provider '${provider}'`;
+        throw new TypeError(
+          `Duplicate registration of Activity '${name}': ${source} would replace a different Activity with this ` +
+            `name. ${this.remedy(name)}`
+        );
+      }
+    }
     return super.configureWorker(this.injectSinks(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 async runWorker(worker: Worker, next: (worker: Worker) => Promise<void>): Promise<void> {
+    for (const [name, fn] of Object.entries(this.options.activities as object)) {
+      const kept = worker.options.activities.get(name);
+      if (kept === undefined || kept === fn || sameSources(kept, fn)) continue;
+      // There is no public way to stop a Worker that has not run yet. With Core shut down first, the
+      // pollers end at once and run()'s own teardown releases the connection.
+      (worker as unknown as { nativeWorker: { initiateShutdown(): void } }).nativeWorker.initiateShutdown();
+      await next(worker).catch(() => undefined);
+      throw new TypeError(
+        `Duplicate registration of Activity '${name}': a plugin applied after OpenAIAgentsPlugin replaced its ` +
+          `Activity with this name. ${this.remedy(name)}`
+      );
+    }
+    return super.runWorker(worker, next);
+  }
+
   override configureReplayWorker(options: ReplayWorkerOptions): ReplayWorkerOptions {
     return super.configureReplayWorker(this.injectSinks(options));
   }
@@ -229,6 +300,14 @@ export class OpenAIAgentsPlugin extends SimplePlugin {
     };
   }
 
+  private remedy(name: string): string {
+    const rename = this.activityProviders.has(name) ? 'Rename the provider, rename' : 'Rename';
+    return (
+      `${rename} or remove the other Activity, run the other integration on its own Worker and task queue, ` +
+      'or use one OpenAIAgentsPlugin instance on this Worker.'
+    );
+  }
+
   private injectSinks<T extends { sinks?: InjectedSinks<any> }>(options: T): T {
     // Registered unconditionally; per-Workflow useOtelInstrumentation flag drives suppression
     // at the bridge layer via the SUPPRESS_OTEL_BRIDGE marker.