diff --git a/packages/core/src/__tests__/integration-step-executor-fallback.test.ts b/packages/core/src/__tests__/integration-step-executor-fallback.test.ts new file mode 100644 index 0000000..d683857 --- /dev/null +++ b/packages/core/src/__tests__/integration-step-executor-fallback.test.ts @@ -0,0 +1,147 @@ +import { describe, it, expect } from 'vitest'; +import { WorkflowRunner } from '../runner.js'; +import type { RelayYamlConfig, RunnerStepExecutor, WorkflowStep } from '../types.js'; + +/** + * Integration steps used to be fatal unless the caller supplied an executor + * implementing `executeIntegrationStep`. Nothing did: the local default has no + * executor at all, and the cloud runtime's executor implements only + * `executeAgentStep` and `executeDeterministicStep`. They now resolve a + * built-in executor per step, the way deterministic steps already did. + */ +function configWithIntegrationStep(integration: string): RelayYamlConfig { + const step: WorkflowStep = { + name: 'integration-step', + type: 'integration', + integration, + action: 'listIssues', + params: { repo: 'owner/repo' }, + } as WorkflowStep; + + return { + version: '1.0', + name: 'integration-fallback', + swarm: { pattern: 'dag' }, + agents: [{ name: 'unused', cli: 'claude' }], + workflows: [{ name: 'wf', steps: [step] }], + } as RelayYamlConfig; +} + +describe('integration step executor resolution', () => { + it('does not reject a known integration when no executor is supplied', async () => { + const runner = new WorkflowRunner(); + const step = configWithIntegrationStep('github').workflows![0].steps[0]; + const resolved = await ( + runner as unknown as { + resolveBuiltinIntegrationExecutor(s: WorkflowStep): Promise; + } + ).resolveBuiltinIntegrationExecutor(step); + + expect(resolved?.executeIntegrationStep).toBeTypeOf('function'); + }); + + it('does NOT resolve a built-in for stateful integrations', async () => { + // BrowserStepExecutor holds live BrowserClient sessions and needs + // closeAll() on teardown. Caching it on the runner would leak browser + // processes across runs, so it is excluded until that lifecycle exists. + const runner = new WorkflowRunner(); + for (const integration of ['browser', 'slack']) { + const step = configWithIntegrationStep(integration).workflows![0].steps[0]; + const resolved = await ( + runner as unknown as { + resolveBuiltinIntegrationExecutor(s: WorkflowStep): Promise; + } + ).resolveBuiltinIntegrationExecutor(step); + expect(resolved, integration).toBeUndefined(); + } + }); + + it('gives concurrent resolvers the SAME instance', async () => { + // Caching only completed executors lets two parallel first calls both miss + // and construct their own. + const runner = new WorkflowRunner(); + const step = configWithIntegrationStep('github').workflows![0].steps[0]; + const resolve = ( + runner as unknown as { + resolveBuiltinIntegrationExecutor(s: WorkflowStep): Promise; + } + ).resolveBuiltinIntegrationExecutor.bind(runner); + + const [a, b, c] = await Promise.all([resolve(step), resolve(step), resolve(step)]); + expect(a).toBe(b); + expect(b).toBe(c); + }); + + it('memoises the built-in executor per integration', async () => { + const runner = new WorkflowRunner(); + const step = configWithIntegrationStep('github').workflows![0].steps[0]; + const resolve = ( + runner as unknown as { + resolveBuiltinIntegrationExecutor(s: WorkflowStep): Promise; + } + ).resolveBuiltinIntegrationExecutor.bind(runner); + + expect(await resolve(step)).toBe(await resolve(step)); + }); + + it('returns undefined for an unknown integration so the caller can name the built-ins', async () => { + const runner = new WorkflowRunner(); + const step = configWithIntegrationStep('not-a-real-integration').workflows![0].steps[0]; + const resolved = await ( + runner as unknown as { + resolveBuiltinIntegrationExecutor(s: WorkflowStep): Promise; + } + ).resolveBuiltinIntegrationExecutor(step); + + expect(resolved).toBeUndefined(); + }); + + it('routes the step to an injected executor, not the built-in, through execute()', async () => { + // Drive the real dispatch path. Asserting on resolveBuiltinIntegrationExecutor + // alone would still pass if the runner picked the built-in at the call site. + let injectedCalls = 0; + const injected: RunnerStepExecutor = { + async executeAgentStep() { + throw new Error('no agent steps in this workflow'); + }, + async executeDeterministicStep() { + return { output: 'stub', exitCode: 0 }; + }, + async executeIntegrationStep() { + injectedCalls += 1; + return { output: 'from-injected', success: true }; + }, + }; + + const runner = new WorkflowRunner({ cwd: process.cwd(), executor: injected }); + const row = await runner.execute(configWithIntegrationStep('github')); + + expect(injectedCalls).toBe(1); + expect(row.status).toBe('completed'); + }); + + it('falls back to the built-in when the executor cannot run integration steps', async () => { + // The exact shape cloud passes: agent + deterministic only. Before this + // change the runner threw "Integration steps require a cloud executor". + const cloudShaped: RunnerStepExecutor = { + async executeAgentStep() { + throw new Error('no agent steps in this workflow'); + }, + async executeDeterministicStep() { + return { output: 'stub', exitCode: 0 }; + }, + }; + + const runner = new WorkflowRunner({ cwd: process.cwd(), executor: cloudShaped }); + const resolved = await ( + runner as unknown as { + resolveBuiltinIntegrationExecutor(s: WorkflowStep): Promise; + } + ).resolveBuiltinIntegrationExecutor( + configWithIntegrationStep('github').workflows![0].steps[0], + ); + + expect(cloudShaped.executeIntegrationStep).toBeUndefined(); + expect(resolved?.executeIntegrationStep).toBeTypeOf('function'); + }); +}); diff --git a/packages/core/src/runner.ts b/packages/core/src/runner.ts index 7344a0d..80092d9 100644 --- a/packages/core/src/runner.ts +++ b/packages/core/src/runner.ts @@ -5463,6 +5463,62 @@ export class WorkflowRunner { /** * Execute an integration step (external service interaction via executor). */ + /** + * Built-in integration executors. + * + * Only STATELESS executors belong here. The runner caches what it resolves + * for the lifetime of the instance, so a stateful executor would leak across + * runs — `BrowserStepExecutor` holds a Map of live `BrowserClient` sessions + * and needs `closeAll()` on run teardown, so it is deliberately excluded + * until that lifecycle exists. `GitHubStepExecutor` and `SlackStepExecutor` + * hold only a readonly options object. + * + * Scoped to `github` for now: that is the integration with a reported + * failure. Adding `slack` is a one-line change once something needs it. + */ + private static readonly BUILTIN_INTEGRATION_LOADERS: Record< + string, + () => Promise + > = { + github: async () => new (await import('./integrations/github.js')).GitHubStepExecutor(), + }; + + /** + * In-flight loads, not finished executors: two steps resolving the same + * integration concurrently must get the SAME instance. Caching only on + * completion lets both observe a miss and construct their own. + */ + private readonly builtinIntegrationExecutors = new Map< + string, + Promise + >(); + + /** + * Resolve the built-in executor for a step's integration. Returns undefined + * for an unknown integration so the caller can name the ones that exist. + */ + private async resolveBuiltinIntegrationExecutor( + step: WorkflowStep + ): Promise { + const integration = step.integration; + if (!integration) return undefined; + + const inFlight = this.builtinIntegrationExecutors.get(integration); + if (inFlight) return inFlight; + + const load = WorkflowRunner.BUILTIN_INTEGRATION_LOADERS[integration]; + if (!load) return undefined; + + // Store the promise before awaiting so concurrent callers share it. Evict + // on failure so a transient import error is not cached forever. + const pending = load().catch((error: unknown) => { + this.builtinIntegrationExecutors.delete(integration); + throw error; + }); + this.builtinIntegrationExecutors.set(integration, pending); + return pending; + } + private async executeIntegrationStep( step: WorkflowStep, state: StepState, @@ -5479,14 +5535,26 @@ export class WorkflowRunner { resolvedParams[key] = this.interpolateStepTask(value, stepOutputContext); } - if (!this.executor?.executeIntegrationStep) { + // Integration steps resolve their executor per-step, exactly as + // deterministic steps do: an injected executor wins, otherwise the + // built-in for that integration runs. Before this, an executor without + // `executeIntegrationStep` was fatal — which meant integration steps + // could not run locally (no executor) OR in cloud (whose executor + // implements only agent + deterministic steps). + const integrationExecutor = this.executor?.executeIntegrationStep + ? this.executor + : await this.resolveBuiltinIntegrationExecutor(step); + + if (!integrationExecutor?.executeIntegrationStep) { + const builtins = Object.keys(WorkflowRunner.BUILTIN_INTEGRATION_LOADERS).join(', '); throw new Error( - `Integration steps require a cloud executor. Step "${step.name}" cannot run locally. ` + - `Use "cloud run" to execute workflows with integration steps.` + `Step "${step.name}" uses integration "${step.integration}", which has no executor. ` + + `Built-in executors exist for: ${builtins}. ` + + `Pass a RunnerStepExecutor implementing executeIntegrationStep to run others.` ); } - const integrationResult = await this.executor.executeIntegrationStep(step, resolvedParams, { + const integrationResult = await integrationExecutor.executeIntegrationStep(step, resolvedParams, { workspaceId: this.workspaceId, injectAnswerToAgent: (input) => this.injectAnswerToAgent(input), });