From 6cbe37a40b6ad0328b0b918abf2f877d63f4951a Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Fri, 23 Oct 2026 00:33:37 -0700 Subject: [PATCH 1/3] Breaking: fail Worker creation when an OpenAI Agents plugin Activity would replace another Activity OpenAIAgentsPlugin merged its Activities over the Worker's `activities` option. A caller Activity named `invokeModelActivity`, `invokeModelStreamActivity`, or like a provider Activity (`-stateless-...`, `-stateful-...`, `-sandbox-...`) was dropped with no error, and Workflows calling it ran the plugin's handler. configureWorker now throws a TypeError for such a name. Activities that an OpenAIAgentsPlugin built are still replaced, so one instance on both the NativeConnection and the Worker still starts, and chained instances keep the last one's Activities, as the README describes. Breaking change: Worker.create now fails when another Activity on the Worker has the name of an OpenAI Agents plugin Activity. Rename or remove that Activity. Activity names, inputs and History do not change. --- typescript/openai-agents/README.md | 2 + .../src/__tests__/plugin.test.ts | 46 +++++++++++++++++++ typescript/openai-agents/src/worker/plugin.ts | 21 +++++++++ 3 files changed, 69 insertions(+) create mode 100644 typescript/openai-agents/src/__tests__/plugin.test.ts diff --git a/typescript/openai-agents/README.md b/typescript/openai-agents/README.md index 5a1f0d2e..bf4b212a 100644 --- a/typescript/openai-agents/README.md +++ b/typescript/openai-agents/README.md @@ -266,6 +266,8 @@ 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 `-stateless-call-tool`. If another Activity on the Worker already has one of these names, `Worker.create` fails with a `TypeError` instead of replacing it. Activities from another `OpenAIAgentsPlugin` instance are still replaced. + 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: diff --git a/typescript/openai-agents/src/__tests__/plugin.test.ts b/typescript/openai-agents/src/__tests__/plugin.test.ts new file mode 100644 index 00000000..2901d690 --- /dev/null +++ b/typescript/openai-agents/src/__tests__/plugin.test.ts @@ -0,0 +1,46 @@ +import test from 'ava'; +import type { WorkerOptions } from '@temporalio/worker'; +import { StatelessMCPServerProvider } from '../worker/mcp-provider'; +import { OpenAIAgentsPlugin } from '../worker/plugin'; +import { FakeModelProvider } from './stubs/openai-agents'; + +type Activities = Record; + +const callerActivity = async () => 'caller'; + +test('configureWorker refuses to replace a caller Activity, but not its own', (t) => { + const plugin = new OpenAIAgentsPlugin({ + modelProvider: new FakeModelProvider([]), + mcpServerProviders: [ + new StatelessMCPServerProvider('docs', () => { + throw new Error('not called'); + }), + ], + }); + + for (const name of ['invokeModelActivity', 'docs-stateless-call-tool']) { + t.throws( + () => plugin.configureWorker({ taskQueue: 'test', activities: { [name]: callerActivity } } as WorkerOptions), + { + instanceOf: TypeError, + message: + `Duplicate registration of Activity '${name}': OpenAIAgentsPlugin would replace a different Activity ` + + 'with this name. Rename or remove the other registration.', + } + ); + } + + // 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); + + // Chained instances keep the last instance's Activities, as before. + const chained = new OpenAIAgentsPlugin({ modelProvider: new FakeModelProvider([]) }).configureWorker(once); + t.not((chained.activities as Activities).invokeModelActivity, (once.activities as Activities).invokeModelActivity); + t.is((chained.activities as Activities).lookUp, callerActivity); +}); diff --git a/typescript/openai-agents/src/worker/plugin.ts b/typescript/openai-agents/src/worker/plugin.ts index 05a89618..703d7c77 100644 --- a/typescript/openai-agents/src/worker/plugin.ts +++ b/typescript/openai-agents/src/worker/plugin.ts @@ -48,6 +48,11 @@ function resolveAgentsCoreAsCommonJs(compiler: WebpackCompiler): void { }); } +// Every Activity function an OpenAIAgentsPlugin builds. configureWorker lets these be replaced, so +// one instance can be on both the NativeConnection and the Worker, and chained instances keep the +// last one's Activities. +const pluginActivityFunctions = new WeakSet(); + /** * Options controlling trace interceptor behavior on both the Activity side * (directly) and the Workflow side (via header propagation). @@ -115,6 +120,8 @@ export interface OpenAIAgentsPluginOptions { * @experimental The OpenAI Agents plugin is an experimental feature; APIs may change without notice. */ export class OpenAIAgentsPlugin extends SimplePlugin { + private readonly activityNames: readonly string[]; + constructor(options: OpenAIAgentsPluginOptions) { const modelActivities = createModelActivity(options.modelProvider, options.hostedToolCredentials); @@ -140,6 +147,9 @@ export class OpenAIAgentsPlugin extends SimplePlugin { seenNames.add(provider.name); allActivities = { ...allActivities, ...provider._getActivities() }; } + for (const fn of Object.values(allActivities)) { + pluginActivityFunctions.add(fn); + } const activityInterceptorOptions: OpenAIAgentsTraceInterceptorOptions = { addTemporalSpans: interceptorOpts?.addTemporalSpans, @@ -180,9 +190,20 @@ export class OpenAIAgentsPlugin extends SimplePlugin { nexus: [() => ({ inbound: new OpenAIAgentsTraceNexusInboundInterceptor() })], }, }); + this.activityNames = Object.keys(allActivities); } override configureWorker(options: WorkerOptions): WorkerOptions { + const existing = options.activities as Record | undefined; + for (const name of this.activityNames) { + const fn = existing && Object.prototype.propertyIsEnumerable.call(existing, name) ? existing[name] : undefined; + if (typeof fn === 'function' && !pluginActivityFunctions.has(fn)) { + throw new TypeError( + `Duplicate registration of Activity '${name}': OpenAIAgentsPlugin would replace a different Activity ` + + 'with this name. Rename or remove the other registration.' + ); + } + } return super.configureWorker(this.injectSinks(options)); } From 64c6e891254cd8339c9edb8dce6657ba3ec42ed0 Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Mon, 2 Nov 2026 09:44:54 -0800 Subject: [PATCH 2/3] Refuse a second OpenAIAgentsPlugin with different sources, and later replacements configureWorker let any Activity an OpenAIAgentsPlugin built be replaced, so a second instance with another model provider, credentials resolver or MCP server or sandbox client provider replaced the first one's Activities silently. Since Worker.create applies the NativeConnection's plugins after the Worker's, the connection's instance won. Each Activity now records its name and what it was built from: the modelProvider and hostedToolCredentials for the model Activities, the provider for a provider's Activities. An Activity may replace another only when both entries match, so the same instance on the NativeConnection and the Worker, and two instances built from one options object, still start. A plugin applied after OpenAIAgentsPlugin could also replace its Activities, which configureWorker cannot see. runWorker now refuses that before the Worker polls: it shuts the native Worker down, lets run() release the connection, and throws. The error names the provider for a provider Activity and uses the wording shared with the Google ADK, Vercel AI SDK and Strands plugins. --- typescript/openai-agents/README.md | 4 +- .../src/__tests__/plugin.test.ts | 43 ++++++--- .../test-openai-agents-comprehensive.ts | 54 +++++++++++- typescript/openai-agents/src/worker/plugin.ts | 87 +++++++++++++++---- 4 files changed, 156 insertions(+), 32 deletions(-) diff --git a/typescript/openai-agents/README.md b/typescript/openai-agents/README.md index bc1a7117..ce638811 100644 --- a/typescript/openai-agents/README.md +++ b/typescript/openai-agents/README.md @@ -266,7 +266,9 @@ 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 `-stateless-call-tool`. If another Activity on the Worker already has one of these names, `Worker.create` fails with a `TypeError` instead of replacing it. Activities from another `OpenAIAgentsPlugin` instance are still replaced. +The model Activities are named `invokeModelActivity` and `invokeModelStreamActivity`. Provider Activity names start with the provider's name, such as `-stateless-call-tool`. 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 `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. + +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. diff --git a/typescript/openai-agents/src/__tests__/plugin.test.ts b/typescript/openai-agents/src/__tests__/plugin.test.ts index 2901d690..a26574f4 100644 --- a/typescript/openai-agents/src/__tests__/plugin.test.ts +++ b/typescript/openai-agents/src/__tests__/plugin.test.ts @@ -8,25 +8,33 @@ type Activities = Record; const callerActivity = async () => 'caller'; -test('configureWorker refuses to replace a caller Activity, but not its own', (t) => { - const plugin = new OpenAIAgentsPlugin({ +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'); }), ], - }); + }; + const plugin = new OpenAIAgentsPlugin(options); - for (const name of ['invokeModelActivity', 'docs-stateless-call-tool']) { + const refusals: Record = { + 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}`, + }; + for (const [name, message] of Object.entries(refusals)) { t.throws( () => plugin.configureWorker({ taskQueue: 'test', activities: { [name]: callerActivity } } as WorkerOptions), - { - instanceOf: TypeError, - message: - `Duplicate registration of Activity '${name}': OpenAIAgentsPlugin would replace a different Activity ` + - 'with this name. Rename or remove the other registration.', - } + { instanceOf: TypeError, message } ); } @@ -39,8 +47,15 @@ test('configureWorker refuses to replace a caller Activity, but not its own', (t const twice = plugin.configureWorker(once); t.is((twice.activities as Activities).invokeModelActivity, (once.activities as Activities).invokeModelActivity); - // Chained instances keep the last instance's Activities, as before. - const chained = new OpenAIAgentsPlugin({ modelProvider: new FakeModelProvider([]) }).configureWorker(once); - t.not((chained.activities as Activities).invokeModelActivity, (once.activities as Activities).invokeModelActivity); - t.is((chained.activities as Activities).lookUp, callerActivity); + // 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}`, + }); }); diff --git a/typescript/openai-agents/src/__tests__/test-openai-agents-comprehensive.ts b/typescript/openai-agents/src/__tests__/test-openai-agents-comprehensive.ts index a5944641..37b1772f 100644 --- a/typescript/openai-agents/src/__tests__/test-openai-agents-comprehensive.ts +++ b/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,54 @@ 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 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: + /^Duplicate registration of Activity 'invokeModelActivity': a plugin applied after OpenAIAgentsPlugin replaced/, + }); + t.is(worker.getState(), 'STOPPED'); + + // 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(); + }); +}); diff --git a/typescript/openai-agents/src/worker/plugin.ts b/typescript/openai-agents/src/worker/plugin.ts index 0aa30521..d70ac7cf 100644 --- a/typescript/openai-agents/src/worker/plugin.ts +++ b/typescript/openai-agents/src/worker/plugin.ts @@ -5,7 +5,7 @@ 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 type { SerializableModelActivityOptions } from '../common/model-activity-options'; @@ -49,10 +49,17 @@ function resolveAgentsCoreAsCommonJs(compiler: WebpackCompiler): void { }); } -// Every Activity function an OpenAIAgentsPlugin builds. configureWorker lets these be replaced, so -// one instance can be on both the NativeConnection and the Worker, and chained instances keep the -// last one's Activities. -const pluginActivityFunctions = new WeakSet(); +// 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(); + +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 @@ -127,11 +134,15 @@ export interface OpenAIAgentsPluginOptions { * @experimental The OpenAI Agents plugin is an experimental feature; APIs may change without notice. */ export class OpenAIAgentsPlugin extends SimplePlugin { - private readonly activityNames: readonly string[]; + // The provider each provider Activity comes from, by Activity name, for the error message. + private readonly activityProviders: ReadonlyMap; 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 Promise> = { ...modelActivities }; @@ -146,6 +157,7 @@ export class OpenAIAgentsPlugin extends SimplePlugin { ...(options.sandboxClientProviders ?? []), ]; const seenNames = new Set(); + const activityProviders = new Map(); for (const provider of providers) { if (seenNames.has(provider.name)) { throw new Error( @@ -153,10 +165,12 @@ export class OpenAIAgentsPlugin extends SimplePlugin { ); } seenNames.add(provider.name); - allActivities = { ...allActivities, ...provider._getActivities() }; - } - for (const fn of Object.values(allActivities)) { - pluginActivityFunctions.add(fn); + const providerActivities = provider._getActivities(); + for (const [name, fn] of Object.entries(providerActivities)) { + activitySources.set(fn, [name, provider]); + activityProviders.set(name, provider.name); + } + allActivities = { ...allActivities, ...providerActivities }; } const activityInterceptorOptions: OpenAIAgentsTraceInterceptorOptions = { @@ -201,24 +215,57 @@ export class OpenAIAgentsPlugin extends SimplePlugin { ], }, }); - this.activityNames = Object.keys(allActivities); + 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 | undefined; - for (const name of this.activityNames) { - const fn = existing && Object.prototype.propertyIsEnumerable.call(existing, name) ? existing[name] : undefined; - if (typeof fn === 'function' && !pluginActivityFunctions.has(fn)) { + 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}': OpenAIAgentsPlugin would replace a different Activity ` + - 'with this name. Rename or remove the other registration.' + `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): Promise { + 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)); } @@ -249,6 +296,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 }>(options: T): T { // Registered unconditionally; per-Workflow useOtelInstrumentation flag drives suppression // at the bridge layer via the SUPPRESS_OTEL_BRIDGE marker. From 7d6993818fef73a9c71cdcafe21c0d8b40c60c70 Mon Sep 17 00:00:00 2001 From: Francesc Campoy Date: Mon, 2 Nov 2026 12:43:20 -0800 Subject: [PATCH 3/3] Name a stateful provider as passed, document wrapping plugins, test the connection layout The duplicate-Activity error named a stateful MCP server provider by its internal name ('docs-stateful'). It now shows the name the caller passed. The README now says a plugin applied after OpenAIAgentsPlugin that wraps its Activities counts as replacing them, and points to Activity interceptors instead. The runWorker test now also covers one instance on the NativeConnection and the Worker with the later plugin on the connection, where both of the instance's checks run. --- typescript/openai-agents/README.md | 2 +- .../src/__tests__/plugin.test.ts | 14 ++++++++++- .../test-openai-agents-comprehensive.ts | 23 +++++++++++++++---- typescript/openai-agents/src/worker/plugin.ts | 8 +++++-- 4 files changed, 38 insertions(+), 9 deletions(-) diff --git a/typescript/openai-agents/README.md b/typescript/openai-agents/README.md index ce638811..a0b20542 100644 --- a/typescript/openai-agents/README.md +++ b/typescript/openai-agents/README.md @@ -266,7 +266,7 @@ 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 `-stateless-call-tool`. 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 `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. +The model Activities are named `invokeModelActivity` and `invokeModelStreamActivity`. Provider Activity names start with the provider's name, such as `-stateless-call-tool`. 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 `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. diff --git a/typescript/openai-agents/src/__tests__/plugin.test.ts b/typescript/openai-agents/src/__tests__/plugin.test.ts index a26574f4..190299bd 100644 --- a/typescript/openai-agents/src/__tests__/plugin.test.ts +++ b/typescript/openai-agents/src/__tests__/plugin.test.ts @@ -1,6 +1,7 @@ import test from 'ava'; -import type { WorkerOptions } from '@temporalio/worker'; +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'; @@ -19,6 +20,13 @@ test('configureWorker refuses to replace a different Activity with the same name new StatelessMCPServerProvider('docs', () => { throw new Error('not called'); }), + new StatefulMCPServerProvider( + 'notes', + () => { + throw new Error('not called'); + }, + {} as NativeConnection + ), ], }; const plugin = new OpenAIAgentsPlugin(options); @@ -30,6 +38,10 @@ test('configureWorker refuses to replace a different Activity with the same name '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( diff --git a/typescript/openai-agents/src/__tests__/test-openai-agents-comprehensive.ts b/typescript/openai-agents/src/__tests__/test-openai-agents-comprehensive.ts index 37b1772f..ca32f9ae 100644 --- a/typescript/openai-agents/src/__tests__/test-openai-agents-comprehensive.ts +++ b/typescript/openai-agents/src/__tests__/test-openai-agents-comprehensive.ts @@ -933,6 +933,8 @@ 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({ @@ -942,13 +944,24 @@ test('runWorker refuses an Activity a later plugin replaced, and starts the Work workflowBundle: sharedWorkflowBundle, taskQueue: `later-plugin-${randomUUID()}`, }); - await t.throwsAsync(worker.run(), { - instanceOf: TypeError, - message: - /^Duplicate registration of Activity 'invokeModelActivity': a plugin applied after OpenAIAgentsPlugin replaced/, - }); + 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. diff --git a/typescript/openai-agents/src/worker/plugin.ts b/typescript/openai-agents/src/worker/plugin.ts index d70ac7cf..8bc8aa51 100644 --- a/typescript/openai-agents/src/worker/plugin.ts +++ b/typescript/openai-agents/src/worker/plugin.ts @@ -8,6 +8,7 @@ import { SimplePlugin } from '@temporalio/plugin'; 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, @@ -165,10 +166,13 @@ export class OpenAIAgentsPlugin extends SimplePlugin { ); } seenNames.add(provider.name); + // 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, provider.name); + activityProviders.set(name, shownName); } allActivities = { ...allActivities, ...providerActivities }; }