diff --git a/typescript/openai-agents/README.md b/typescript/openai-agents/README.md index 3e727d1f..a0b20542 100644 --- a/typescript/openai-agents/README.md +++ b/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 `-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. + 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..190299bd --- /dev/null +++ b/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; + +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 = { + 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}`, + }); +}); 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..ca32f9ae 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,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(); + }); +}); diff --git a/typescript/openai-agents/src/worker/plugin.ts b/typescript/openai-agents/src/worker/plugin.ts index b1bc66dd..8bc8aa51 100644 --- a/typescript/openai-agents/src/worker/plugin.ts +++ b/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(); + +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; 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 }; @@ -140,6 +158,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( @@ -147,7 +166,15 @@ export class OpenAIAgentsPlugin extends SimplePlugin { ); } 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 | 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): 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)); } @@ -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 }>(options: T): T { // Registered unconditionally; per-Workflow useOtelInstrumentation flag drives suppression // at the bridge layer via the SUPPRESS_OTEL_BRIDGE marker.