From 8f11996738ca28b7d99ac91f7b260207cc651e6a Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Thu, 13 Aug 2026 23:54:03 -0700 Subject: [PATCH 1/9] [workflow] Load per-source replay bundles --- .changeset/lazy-workflow-source-bundles.md | 7 + packages/builders/src/base-builder.ts | 450 +++++++++--------- .../src/workflow-bundle-boundary.test.ts | 269 ++++++++++- .../builders/src/workflow-bundle-module.ts | 52 ++ .../builders/src/workflows-extractor.test.ts | 40 +- packages/builders/src/workflows-extractor.ts | 61 ++- packages/core/src/runtime.test.ts | 71 ++- packages/core/src/runtime.ts | 46 +- packages/core/src/vm/script-cache.test.ts | 11 +- packages/core/src/vm/script-cache.ts | 41 +- packages/next/src/builder-eager.ts | 14 +- 11 files changed, 756 insertions(+), 306 deletions(-) create mode 100644 .changeset/lazy-workflow-source-bundles.md create mode 100644 packages/builders/src/workflow-bundle-module.ts diff --git a/.changeset/lazy-workflow-source-bundles.md b/.changeset/lazy-workflow-source-bundles.md new file mode 100644 index 0000000000..295e0a18b6 --- /dev/null +++ b/.changeset/lazy-workflow-source-bundles.md @@ -0,0 +1,7 @@ +--- +'@workflow/builders': patch +'@workflow/core': patch +'@workflow/next': patch +--- + +Load only the selected workflow source bundle when replay starts. diff --git a/packages/builders/src/base-builder.ts b/packages/builders/src/base-builder.ts index 607df9c109..41e14b3f2b 100644 --- a/packages/builders/src/base-builder.ts +++ b/packages/builders/src/base-builder.ts @@ -1,6 +1,7 @@ import { randomUUID } from 'node:crypto'; import { mkdir, + readdir, readFile, realpath, rename, @@ -42,6 +43,13 @@ import { createPseudoPackagePlugin } from './pseudo-package-esbuild-plugin.js'; import { createSwcPlugin } from './swc-esbuild-plugin.js'; import { detectWorkflowPatterns } from './transform-utils.js'; import type { SourcemapMode, WorkflowConfig } from './types.js'; +import { + encodeWorkflowBundle, + isWorkflowBundleFileName, + serializeWorkflowBundle, + WORKFLOW_BUNDLE_DIRECTORY, + workflowBundleFileName, +} from './workflow-bundle-module.js'; import { extractWorkflowGraphs } from './workflows-extractor.js'; const enhancedResolve = promisify(enhancedResolveOriginal); @@ -64,6 +72,10 @@ const VALID_SOURCEMAP_STRINGS = new Set([ 'external', 'both', ]); +const WORKFLOW_ROUTE_EXTERNALS = [ + '@aws-sdk/credential-provider-web-identity', + `./${WORKFLOW_BUNDLE_DIRECTORY}/*`, +]; /** * Parse the value of the `WORKFLOW_SOURCEMAP` environment variable into a @@ -168,6 +180,46 @@ type CachedManifestTransform = { manifest: WorkflowManifest; }; +type WorkflowBundle = { + code: string; + fileName: string; + workflowIds: string[]; +}; + +function getWorkflowIds(manifest: WorkflowManifest): string[] { + return Object.values(manifest.workflows ?? {}).flatMap((workflows) => + Object.values(workflows).map(({ workflowId }) => workflowId) + ); +} + +function createWorkflowBundleLoaders( + bundles: WorkflowBundle[], + source: 'module' | 'inline' +): string { + const loaders = bundles + .map(({ code, fileName }, index) => { + const modulePath = `./${WORKFLOW_BUNDLE_DIRECTORY}/${fileName}`; + if (source === 'module') { + return `let workflowBundlePromise${index}; +const loadWorkflowBundle${index} = () => workflowBundlePromise${index} ??= import('${modulePath}').then((module) => Buffer.from(module.default, 'base64').toString('utf8'));`; + } + return `let workflowBundle${index}; +const loadWorkflowBundle${index} = () => Promise.resolve(workflowBundle${index} ??= Buffer.from(${JSON.stringify(encodeWorkflowBundle(code))}, 'base64').toString('utf8')); +loadWorkflowBundle${index}.bundleFile = ${JSON.stringify(modulePath)};`; + }) + .join('\n'); + const entries = bundles + .flatMap(({ workflowIds }, index) => + workflowIds.map( + (workflowId) => + ` ${JSON.stringify(workflowId)}: loadWorkflowBundle${index},` + ) + ) + .join('\n'); + + return `${loaders}\nconst workflowCode = {\n${entries}\n};`; +} + function formatIdLocation(location: ManifestEntryLocation): string { return `${location.filePath}#${location.name}`; } @@ -1223,16 +1275,12 @@ export const __steps_registered = true; } /** - * Creates a bundle for workflow orchestration functions. + * Creates one VM bundle per workflow source. * Workflows run in a sandboxed VM and coordinate step execution. - * - * @param bundleFinalOutput - If false, skips the final bundling step (used by Next.js) */ protected async createWorkflowsBundle({ inputFiles, - format = 'esm', outfile, - bundleFinalOutput = true, keepInterimBundleContext = this.config.watch, tsconfigPath, discoveredEntries, @@ -1240,111 +1288,92 @@ export const __steps_registered = true; tsconfigPath?: string; inputFiles: string[]; outfile: string; - format?: 'cjs' | 'esm'; - bundleFinalOutput?: boolean; keepInterimBundleContext?: boolean; discoveredEntries?: DiscoveredEntries; }): Promise<{ manifest: WorkflowManifest; interimBundleCtx?: esbuild.BuildContext; - bundleFinal?: (interimBundleResult: string) => Promise; - /** The raw workflow VM code (before wrapping with entrypoint) */ - interimBundleText?: string; + bundleFinal?: ( + interimBundleResult: esbuild.BuildResult + ) => Promise; + workflowBundles: WorkflowBundle[]; }> { const discovered = discoveredEntries ?? (await this.discoverEntries(inputFiles, dirname(outfile), tsconfigPath)); - const workflowFiles = [...discovered.discoveredWorkflows].sort(); - const serdeFiles = [...discovered.discoveredSerdeFiles].sort(); - - // Include serde files that aren't already workflow files for cross-context class registration. - // Classes need to be registered in the workflow bundle so they can be deserialized - // when receiving data from steps or when serializing data to send to steps. - const workflowFilesSet = new Set(workflowFiles); - const serdeOnlyFiles = serdeFiles.filter((f) => !workflowFilesSet.has(f)); - - // log the workflow files for debugging - await this.writeDebugFile(outfile, { workflowFiles, serdeOnlyFiles }); - - // Helper to create import statement from file path - // For packages, uses the package name so esbuild will resolve through - // package.json exports with conditions: ['workflow'] - const createImport = (file: string) => { - const { importPath, isPackage } = getImportPath( - file, - this.config.workingDir - ); - - if (isPackage) { - // Use package name - esbuild will resolve via package.json exports - // and apply the 'workflow' condition - return `import '${importPath}';`; - } - - // Local app file - use relative path - // Normalize both paths to forward slashes before calling relative() - // This is critical on Windows where relative() can produce unexpected results with mixed path formats - const normalizedWorkingDir = this.config.workingDir.replace(/\\/g, '/'); - const normalizedFile = file.replace(/\\/g, '/'); - // Calculate relative path from working directory to the file - let relativePath = relative(normalizedWorkingDir, normalizedFile).replace( - /\\/g, - '/' - ); - // Ensure relative paths start with ./ so esbuild resolves them correctly. - // Paths like ".output/..." are not valid relative specifiers and must - // become "./.output/...". - if (!relativePath.startsWith('./') && !relativePath.startsWith('../')) { - relativePath = `./${relativePath}`; - } - return `import '${relativePath}';`; + const uniqueFiles = (files: string[], excluded = new Set()) => { + const identities = new Set(excluded); + return files.filter((file) => { + const identity = moduleIdentityKey(file, this.moduleSpecifierRoot); + if (identities.has(identity)) return false; + identities.add(identity); + return true; + }); }; + const workflowFiles = uniqueFiles( + [...discovered.discoveredWorkflows].sort() + ); + const workflowIndexByIdentity = new Map(); + await Promise.all( + workflowFiles.map(async (file, index) => { + for (const path of await withRealpaths([file])) { + workflowIndexByIdentity.set( + moduleIdentityKey(path, this.moduleSpecifierRoot), + index + ); + } + }) + ); + const serdeFiles = uniqueFiles([...discovered.discoveredSerdeFiles].sort()); - // Create a virtual entry that imports all workflow files. Dedupe by - // canonical module identity so source/dist copies of the same workspace - // package export don't both get imported (which would make the swc - // plugin generate duplicate workflow IDs). - const emittedImportIdentities = new Set(); - const buildImports = (files: string[]): string => - files - .filter((file) => { - const identity = moduleIdentityKey(file, this.moduleSpecifierRoot); - if (emittedImportIdentities.has(identity)) return false; - emittedImportIdentities.add(identity); - return true; - }) + // log the workflow files for debugging + await this.writeDebugFile(outfile, { workflowFiles, serdeFiles }); + + const createImport = (file: string) => + `import '${this.createRouteImportSpecifier(file, this.config.workingDir)}';`; + + const bundleFiles = workflowFiles.length > 0 ? workflowFiles : [undefined]; + const bundleEntries = bundleFiles.map((workflowFile) => { + const workflowImport = workflowFile ? createImport(workflowFile) : ''; + const workflowIdentity = workflowFile + ? moduleIdentityKey(workflowFile, this.moduleSpecifierRoot) + : undefined; + const serdeImports = serdeFiles + .filter( + (file) => + moduleIdentityKey(file, this.moduleSpecifierRoot) !== + workflowIdentity + ) .map(createImport) .join('\n'); - - // The SWC plugin in workflow mode emits `globalThis.__private_workflows.set(workflowId, fn)` - // calls directly, so we just need to import the files (Map is initialized via banner) - const workflowImports = buildImports(workflowFiles); - - // Include serde-only files for class registration side effects - const serdeImports = buildImports(serdeOnlyFiles); - - const imports = serdeImports - ? `${workflowImports}\n// Serde files for cross-context class registration\n${serdeImports}` - : workflowImports; - + return serdeImports + ? `${workflowImport}\n// Serde files for cross-context class registration\n${serdeImports}` + : workflowImport; + }); const bundleStartTime = Date.now(); const workflowManifest: WorkflowManifest = {}; + const workflowIdsByBundleIndex = new Map(); const esbuildTsconfigOptions = await getEsbuildTsconfigOptions(tsconfigPath); const normalizedWorkflowSideEffectEntries = await withRealpaths([ ...workflowFiles, - ...serdeOnlyFiles, + ...serdeFiles, ]); - // Bundle with esbuild and our custom SWC plugin in workflow mode. - // this bundle will be run inside a vm isolate + const entryPoints = Object.fromEntries( + bundleEntries.map((_, index) => [ + `workflow-${index}`, + `workflow-entry:${index}`, + ]) + ); + const workflowResolveDir = this.config.workingDir; + + // Bundle each workflow source independently. A source may register several + // workflow functions, which all share one lazy VM bundle. const interimBundleCtx = await esbuild.context({ - stdin: { - contents: imports, - resolveDir: this.config.workingDir, - sourcefile: 'virtual-entry.js', - loader: 'js', - }, + entryPoints, + entryNames: '[name]', + outdir: join(dirname(outfile), '.workflow-vm'), bundle: true, absWorkingDir: this.config.workingDir, format: 'cjs', // Runs inside the VM which expects cjs @@ -1385,6 +1414,24 @@ export const __steps_registered = true; '.cjs', ], plugins: [ + { + name: 'workflow-entries', + setup(build) { + build.onResolve({ filter: /^workflow-entry:/ }, ({ path }) => ({ + path, + namespace: 'workflow-entry', + })); + build.onLoad( + { filter: /.*/, namespace: 'workflow-entry' }, + ({ path }) => ({ + contents: + bundleEntries[Number(path.slice(path.indexOf(':') + 1))], + loader: 'js', + resolveDir: workflowResolveDir, + }) + ); + }, + }, // Handle pseudo-packages like 'server-only' and 'client-only' by providing // empty modules. Must run first to intercept these before other resolution. createPseudoPackagePlugin(), @@ -1393,7 +1440,23 @@ export const __steps_registered = true; projectRoot: this.transformProjectRoot, moduleSpecifierRoot: this.moduleSpecifierRoot, workflowManifest, - onAfterTransform: this.config.onAfterTransform, + onAfterTransform: async (result) => { + const sourceIdentity = moduleIdentityKey( + result.absolutePath, + this.moduleSpecifierRoot + ); + const bundleIndex = workflowIndexByIdentity.get(sourceIdentity); + if (bundleIndex !== undefined) { + // Keep the loader IDs tied to the exact transform that esbuild + // used for this bundle. Re-reading through the manifest cache + // can pair new bundle code with stale IDs during watch rebuilds. + workflowIdsByBundleIndex.set( + bundleIndex, + getWorkflowIds(result.workflowManifest) + ); + } + await this.config.onAfterTransform?.(result); + }, sideEffectEntries: normalizedWorkflowSideEffectEntries, }), // This plugin must run after the swc plugin to ensure dead code elimination @@ -1409,6 +1472,36 @@ export const __steps_registered = true; // - createPseudoPackagePlugin() to handle server-only/client-only with empty modules // - createNodeModuleErrorPlugin() to catch Node.js builtin imports at build time }); + const readWorkflowBundles = ( + result: esbuild.BuildResult + ): WorkflowBundle[] => { + return bundleEntries.map((_, index) => { + const output = result.outputFiles?.find( + ({ path }) => basename(path) === `workflow-${index}.js` + ); + if (!output) { + throw new WorkflowBuildError( + `No output generated for workflow bundle ${index}` + ); + } + const workflowFile = workflowFiles[index]; + const workflowIds = workflowFile + ? workflowIdsByBundleIndex.get(index) + : []; + if (!workflowIds) { + throw new WorkflowBuildError( + `No workflow manifest generated for workflow bundle ${index}` + ); + } + return { + code: output.text, + fileName: workflowBundleFileName(index, output.text), + workflowIds, + }; + }); + }; + const workflowBundleDir = join(dirname(outfile), WORKFLOW_BUNDLE_DIRECTORY); + let shouldResetWorkflowBundleDir = true; let shouldDisposeInterimBundleCtx = !keepInterimBundleContext; try { const interimBundle = await interimBundleCtx.rebuild(); @@ -1451,14 +1544,7 @@ export const __steps_registered = true; await this.ensureSwcIgnored(); - if ( - !interimBundle.outputFiles || - interimBundle.outputFiles.length === 0 - ) { - throw new WorkflowBuildError('No output files generated from esbuild', { - hint: 'This usually indicates a misconfigured entry point or an empty workflow directory. Check that your workflow files contain a `"use workflow"` or `"use step"` directive.', - }); - } + const workflowBundles = readWorkflowBundles(interimBundle); // Serde compliance warnings: check if workflow bundle has Node.js imports // alongside serde-registered classes (these will fail at runtime in the sandbox) @@ -1467,10 +1553,9 @@ export const __steps_registered = true; Object.keys(workflowManifest.classes).length > 0 ) { const { analyzeSerdeCompliance } = await import('./serde-checker.js'); - const bundleText = interimBundle.outputFiles[0].text; const serdeResult = analyzeSerdeCompliance({ sourceCode: '', - workflowCode: bundleText, + workflowCode: workflowBundles.map(({ code }) => code).join('\n'), manifest: workflowManifest, }); // De-dupe warnings: group identical issues across classes @@ -1499,83 +1584,35 @@ export const __steps_registered = true; } } - const workflowEntrypointOptionsCode = createWorkflowEntrypointOptionsCode( - { - basePath: this.config.basePath, - routeModuleBodyStartedAt: 'workflowRouteModuleBodyStartedAt', - } - ); - - const bundleFinal = async (interimBundle: string) => { - const workflowBundleCode = interimBundle; - - const workflowFunctionCode = `// biome-ignore-all lint: generated file -/* eslint-disable */ -import { workflowEntrypoint } from 'workflow/runtime'; - -const workflowRouteModuleBodyStartedAt = Date.now(); -const workflowCode = \`${workflowBundleCode.replace(/[\\`$]/g, '\\$&')}\`; - -${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntrypointOptionsCode})`)}`; - - // we skip the final bundling step for Next.js so it can bundle itself - if (!bundleFinalOutput) { - if (!outfile) { - throw new Error(`Invariant: missing outfile for workflow bundle`); - } - // Ensure the output directory exists - const outputDir = dirname(outfile); - await mkdir(outputDir, { recursive: true }); - - await this.writeGeneratedFile(outfile, workflowFunctionCode); - return; + const writeWorkflowBundles = async (bundles: WorkflowBundle[]) => { + await mkdir(dirname(outfile), { recursive: true }); + await mkdir(workflowBundleDir, { recursive: true }); + if (shouldResetWorkflowBundleDir || this.config.watch) { + const generatedFiles = (await readdir(workflowBundleDir)).filter( + isWorkflowBundleFileName + ); + await Promise.all( + generatedFiles.map((file) => + rm(join(workflowBundleDir, file), { force: true }) + ) + ); + shouldResetWorkflowBundleDir = false; } - - const bundleStartTime = Date.now(); - - // Now bundle this so we can resolve the @workflow/core dependency - // we could remove this if we do nft tracing or similar instead - const finalEsmRequireBanner = this.getEsmRequireBanner(format); - const finalWorkflowResult = await esbuild.build({ - banner: { - js: `// biome-ignore-all lint: generated file\n/* eslint-disable */\n${finalEsmRequireBanner}`, - }, - stdin: { - contents: workflowFunctionCode, - resolveDir: this.config.workingDir, - sourcefile: 'virtual-entry.js', - loader: 'js', - }, - outfile, - // Source maps for the final workflow bundle wrapper (not important since this code - // doesn't run in the VM - only the intermediate bundle sourcemap is relevant) - sourcemap: this.resolveSourcemap(EMIT_SOURCEMAPS_FOR_DEBUGGING), - absWorkingDir: this.config.workingDir, - bundle: true, - format, - platform: 'node', - target: 'es2022', - write: true, - keepNames: true, - minify: false, - external: ['@aws-sdk/credential-provider-web-identity'], - }); - - this.logEsbuildMessages( - finalWorkflowResult, - 'final workflow bundle', - true, - { - suppressWarnings: this.config.suppressCreateWorkflowsBundleWarnings, - } - ); - this.logCreateWorkflowsBundleInfo( - 'Created final workflow bundle', - `${Date.now() - bundleStartTime}ms` + await Promise.all( + bundles.map(({ code, fileName }) => + this.writeGeneratedFile( + join(workflowBundleDir, fileName), + serializeWorkflowBundle(code) + ) + ) ); }; - const interimBundleText = interimBundle.outputFiles[0].text; - await bundleFinal(interimBundleText); + const bundleFinal = async (result: esbuild.BuildResult) => { + const bundles = readWorkflowBundles(result); + await writeWorkflowBundles(bundles); + return bundles; + }; + await writeWorkflowBundles(workflowBundles); if (keepInterimBundleContext) { shouldDisposeInterimBundleCtx = false; @@ -1583,10 +1620,10 @@ ${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntr manifest: workflowManifest, interimBundleCtx, bundleFinal, - interimBundleText, + workflowBundles, }; } - return { manifest: workflowManifest, interimBundleText }; + return { manifest: workflowManifest, workflowBundles }; } catch (error) { shouldDisposeInterimBundleCtx = true; throw error; @@ -1641,7 +1678,7 @@ ${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntr manifest: WorkflowManifest; stepsContext?: esbuild.BuildContext; interimBundleCtx?: esbuild.BuildContext; - bundleFinal?: (interimBundleResult: string) => Promise; + bundleFinal?: (interimBundleResult: esbuild.BuildResult) => Promise; discoveredEntries: DiscoveredEntries; stepsManifest: WorkflowManifest; workflowsManifest: WorkflowManifest; @@ -1680,38 +1717,23 @@ ${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntr }); // 2. Build workflow VM code - const tempWorkflowOutfile = `${flowOutfile}.__wf_tmp.js`; const workflowsResult = await this.createWorkflowsBundle({ inputFiles, - outfile: tempWorkflowOutfile, - format, - bundleFinalOutput: false, + outfile: flowOutfile, tsconfigPath, discoveredEntries: effectiveDiscoveredEntries, }); - const workflowVMCode = workflowsResult.interimBundleText; - if (!workflowVMCode) { - throw new Error('createWorkflowsBundle did not return interimBundleText'); - } - - // Clean up the wrapper file - try { - const { unlink } = await import('node:fs/promises'); - await unlink(tempWorkflowOutfile); - } catch { - // Ignore cleanup errors - } - // 3. Generate combined route file const stepsRelativePath = `./${basename(stepsOutfile).replace(/\\/g, '/')}`; - const escapedVMCode = workflowVMCode.replace(/[\\`$]/g, '\\$&'); const workflowEntrypointOptionsCode = createWorkflowEntrypointOptionsCode({ basePath: this.config.basePath, routeModuleBodyStartedAt: 'workflowRouteModuleBodyStartedAt', }); - const combinedFunctionCode = `// biome-ignore-all lint: generated file + const createCombinedFunctionCode = ( + bundles: WorkflowBundle[] + ) => `// biome-ignore-all lint: generated file /* eslint-disable */ import { __steps_registered } from '${stepsRelativePath}'; import { workflowEntrypoint } from 'workflow/runtime'; @@ -1721,9 +1743,12 @@ const workflowRouteModuleBodyStartedAt = Date.now(); // Prevent rollup from tree-shaking the steps side-effect import void __steps_registered; -const workflowCode = \`${escapedVMCode}\`; +${createWorkflowBundleLoaders(bundles, this.config.watch ? 'inline' : 'module')} ${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntrypointOptionsCode})`)}`; + const combinedFunctionCode = createCombinedFunctionCode( + workflowsResult.workflowBundles + ); if (!bundleFinalOutput) { await this.writeGeneratedFile(flowOutfile, combinedFunctionCode); @@ -1756,7 +1781,7 @@ ${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntr keepNames: true, minify: false, define: importMetaDefine, - external: ['@aws-sdk/credential-provider-web-identity'], + external: WORKFLOW_ROUTE_EXTERNALS, }); this.logEsbuildMessages(finalResult, 'combined bundle', true); this.logBaseBuilderInfo( @@ -1779,30 +1804,15 @@ ${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntr }; // Create a custom bundleFinal for watch mode that uses workflowEntrypoint - const combinedBundleFinal = async (interimBundleText: string) => { - const escaped = interimBundleText.replace(/[\\`$]/g, '\\$&'); - const workflowEntrypointOptionsCode = createWorkflowEntrypointOptionsCode( - { - basePath: this.config.basePath, - routeModuleBodyStartedAt: 'workflowRouteModuleBodyStartedAt', - } + const combinedBundleFinal = async (interimBundle: esbuild.BuildResult) => { + if (!workflowsResult.bundleFinal) { + throw new Error('Invariant: missing workflow bundle finalizer'); + } + const bundles = await workflowsResult.bundleFinal(interimBundle); + await this.writeGeneratedFile( + flowOutfile, + createCombinedFunctionCode(bundles) ); - const code = `// biome-ignore-all lint: generated file -/* eslint-disable */ -import { __steps_registered } from '${stepsRelativePath}'; -import { workflowEntrypoint } from 'workflow/runtime'; - -const workflowRouteModuleBodyStartedAt = Date.now(); - -void __steps_registered; - -const workflowCode = \`${escaped}\`; - -${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntrypointOptionsCode})`)}`; - - const outputDir = dirname(flowOutfile); - await mkdir(outputDir, { recursive: true }); - await this.writeGeneratedFile(flowOutfile, code); }; if (this.config.watch) { diff --git a/packages/builders/src/workflow-bundle-boundary.test.ts b/packages/builders/src/workflow-bundle-boundary.test.ts index f8795dbb68..72f0fa3173 100644 --- a/packages/builders/src/workflow-bundle-boundary.test.ts +++ b/packages/builders/src/workflow-bundle-boundary.test.ts @@ -1,9 +1,25 @@ -import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import assert from 'node:assert/strict'; +import { + mkdirSync, + mkdtempSync, + readdirSync, + readFileSync, + rmSync, + statSync, + utimesSync, + writeFileSync, +} from 'node:fs'; import { tmpdir } from 'node:os'; import { join, resolve } from 'node:path'; +import { pathToFileURL } from 'node:url'; import { afterEach, describe, expect, it } from 'vitest'; import { BaseBuilder, type DiscoveredEntries } from './base-builder.js'; import type { StandaloneConfig } from './types.js'; +import { + deserializeWorkflowBundle, + serializeWorkflowBundle, +} from './workflow-bundle-module.js'; +import { extractWorkflowGraphs } from './workflows-extractor.js'; class TestBuilder extends BaseBuilder { async build(): Promise {} @@ -16,12 +32,83 @@ class TestBuilder extends BaseBuilder { return this.createWorkflowsBundle({ inputFiles: [inputFile], outfile, - bundleFinalOutput: false, + discoveredEntries, + }); + } + + createCombinedWorkflowBundle( + inputFiles: string[], + stepsOutfile: string, + flowOutfile: string, + discoveredEntries: DiscoveredEntries + ) { + return this.createCombinedBundle({ + inputFiles, + stepsOutfile, + flowOutfile, + bundleFinalOutput: true, discoveredEntries, }); } } +function createConfig( + repoRoot: string, + workingDir: string, + outputDir: string, + watch: boolean +): StandaloneConfig { + return { + buildTarget: 'standalone', + workingDir, + projectRoot: repoRoot, + moduleSpecifierRoot: repoRoot, + dirs: ['.'], + stepsBundlePath: join(outputDir, 'steps.js'), + workflowsBundlePath: join(outputDir, 'flow.js'), + webhookBundlePath: join(outputDir, 'webhook.js'), + sourcemap: false, + watch, + }; +} + +function writeWorkflowBuiltinsFixture(root: string): void { + const packageDir = join(root, 'node_modules/workflow'); + const serdePackageDir = join(root, 'node_modules/@workflow/serde'); + mkdirSync(join(packageDir, 'internal'), { recursive: true }); + mkdirSync(serdePackageDir, { recursive: true }); + writeFileSync( + join(packageDir, 'package.json'), + JSON.stringify({ + name: 'workflow', + version: '0.0.0-test', + exports: { + './internal/builtins': './internal/builtins.js', + './runtime': './runtime.js', + }, + }) + ); + writeFileSync(join(packageDir, 'internal/builtins.js'), 'export {};\n'); + writeFileSync( + join(packageDir, 'runtime.js'), + 'export const workflowEntrypoint = () => async () => {};\n' + ); + writeFileSync( + join(serdePackageDir, 'package.json'), + JSON.stringify({ + name: '@workflow/serde', + version: '0.0.0-test', + type: 'module', + exports: './index.js', + }) + ); + writeFileSync( + join(serdePackageDir, 'index.js'), + `export const WORKFLOW_SERIALIZE = Symbol.for('workflow.serialize'); +export const WORKFLOW_DESERIALIZE = Symbol.for('workflow.deserialize');\n` + ); +} + describe('workflow bundle boundary', () => { const outputDirs: string[] = []; @@ -31,6 +118,18 @@ describe('workflow bundle boundary', () => { } }); + it('round-trips VM source without exposing nested template syntax', () => { + const code = + 'const value = `hello $' + '{name}`;\u2028const done = true;\u2029'; + const moduleCode = serializeWorkflowBundle(code); + + expect(moduleCode).not.toContain('`'); + expect(moduleCode).not.toContain('${'); + expect(moduleCode).not.toContain('\u2028'); + expect(moduleCode).not.toContain('\u2029'); + expect(deserializeWorkflowBundle(moduleCode)).toBe(code); + }); + it('does not bundle world schemas into a workflow without schemas', async () => { const repoRoot = resolve(import.meta.dirname, '../../..'); const outputDir = mkdtempSync(join(tmpdir(), 'workflow-pruning-')); @@ -41,24 +140,14 @@ describe('workflow bundle boundary', () => { `export async function minimal() { "use workflow"; return 1; }` ); - const config: StandaloneConfig = { - buildTarget: 'standalone', - workingDir: outputDir, - projectRoot: repoRoot, - moduleSpecifierRoot: repoRoot, - dirs: ['.'], - stepsBundlePath: join(outputDir, 'steps.js'), - workflowsBundlePath: join(outputDir, 'workflow.js'), - webhookBundlePath: join(outputDir, 'webhook.js'), - sourcemap: false, - }; + const config = createConfig(repoRoot, outputDir, outputDir, false); const discoveredEntries: DiscoveredEntries = { discoveredSteps: new Set(), discoveredWorkflows: new Set([inputFile]), discoveredSerdeFiles: new Set(), }; - const { interimBundleText } = await new TestBuilder( + const { workflowBundles } = await new TestBuilder( config ).createWorkflowBundle( inputFile, @@ -66,7 +155,155 @@ describe('workflow bundle boundary', () => { discoveredEntries ); - expect(interimBundleText).toBeDefined(); - expect(interimBundleText).not.toContain('/node_modules/zod'); + expect(workflowBundles).toHaveLength(1); + expect(workflowBundles[0].code).not.toContain('/node_modules/zod'); + }); + + it('emits one lazy VM bundle per workflow source', async () => { + const repoRoot = resolve(import.meta.dirname, '../../..'); + const workingDir = join(repoRoot, 'workbench/nextjs-turbopack'); + const outputDir = mkdtempSync(join(workingDir, '.workflow-sources-')); + outputDirs.push(outputDir); + const first = join(outputDir, 'first.ts'); + const second = join(outputDir, 'second.ts'); + writeFileSync( + first, + `import { WORKFLOW_DESERIALIZE, WORKFLOW_SERIALIZE } from '@workflow/serde'; +export class HybridSerde { + static [WORKFLOW_SERIALIZE](value) { return { value: value.value }; } + static [WORKFLOW_DESERIALIZE](data) { return new HybridSerde(data.value); } + constructor(value) { this.value = value; } +} +export async function first() { "use workflow"; return "lazy-first-marker"; } +export async function alsoFirst() { "use workflow"; return 2; }` + ); + writeFileSync( + second, + `export async function second() { "use workflow"; return "lazy-second-marker"; }` + ); + + const config = createConfig(repoRoot, outputDir, outputDir, false); + const discoveredEntries: DiscoveredEntries = { + discoveredSteps: new Set(), + discoveredWorkflows: new Set([second, first]), + discoveredSerdeFiles: new Set([first]), + }; + const workflowBundleDir = join(outputDir, 'workflow-bundles'); + writeWorkflowBuiltinsFixture(outputDir); + mkdirSync(workflowBundleDir); + writeFileSync(join(workflowBundleDir, 'keep.txt'), 'user-owned'); + + await new TestBuilder(config).createCombinedWorkflowBundle( + [first, second], + config.stepsBundlePath, + config.workflowsBundlePath, + discoveredEntries + ); + + const bundleFiles = readdirSync(workflowBundleDir) + .filter((file) => file.endsWith('.mjs')) + .sort(); + expect(bundleFiles).toHaveLength(2); + expect( + bundleFiles.every((file) => /^\d+-[a-f0-9]{16}\.mjs$/.test(file)) + ).toBe(true); + expect(readFileSync(join(workflowBundleDir, 'keep.txt'), 'utf8')).toBe( + 'user-owned' + ); + const route = readFileSync(config.workflowsBundlePath, 'utf8'); + expect(route).toContain(`workflow-bundles/${bundleFiles[0]}`); + expect(route).toContain(`workflow-bundles/${bundleFiles[1]}`); + expect(route.match(/: loadWorkflowBundle0,/g)).toHaveLength(2); + expect(route.match(/: loadWorkflowBundle1/g)).toHaveLength(1); + expect(route).not.toContain('lazy-first-marker'); + const firstBundle = await import( + `${pathToFileURL(join(workflowBundleDir, bundleFiles[0])).href}?test` + ); + const secondBundle = await import( + `${pathToFileURL(join(workflowBundleDir, bundleFiles[1])).href}?test` + ); + const firstCode = Buffer.from(firstBundle.default, 'base64').toString(); + const secondCode = Buffer.from(secondBundle.default, 'base64').toString(); + expect(firstCode).toContain('lazy-first-marker'); + expect(firstCode).not.toContain('lazy-second-marker'); + expect(secondCode).toContain('HybridSerde'); + }); + + it('changes the lazy module URL after a watch rebuild', async () => { + const repoRoot = resolve(import.meta.dirname, '../../..'); + const workingDir = join(repoRoot, 'workbench/nextjs-turbopack'); + const outputDir = mkdtempSync(join(workingDir, '.workflow-watch-')); + outputDirs.push(outputDir); + const workflowFile = join(outputDir, 'watched.ts'); + writeFileSync( + workflowFile, + `export async function watched() { "use workflow"; return "before-watch"; } +export async function removedAfterWatch() { "use workflow"; return "remove-me"; }` + ); + const config = createConfig(repoRoot, outputDir, outputDir, true); + writeWorkflowBuiltinsFixture(outputDir); + const discoveredEntries: DiscoveredEntries = { + discoveredSteps: new Set(), + discoveredWorkflows: new Set([workflowFile]), + discoveredSerdeFiles: new Set(), + }; + const result = await new TestBuilder(config).createCombinedWorkflowBundle( + [workflowFile], + config.stepsBundlePath, + config.workflowsBundlePath, + discoveredEntries + ); + assert(result.interimBundleCtx); + assert(result.stepsContext); + assert(result.bundleFinal); + + try { + const workflowBundleDir = join(outputDir, 'workflow-bundles'); + const oldFile = readdirSync(workflowBundleDir).find((file) => + file.endsWith('.mjs') + ); + assert(oldFile); + const oldStats = statSync(workflowFile); + writeFileSync( + workflowFile, + `export async function watched() { "use workflow"; return "after--watch"; } +export async function renamedAfterWatch() { "use workflow"; return "rename-me"; }` + ); + // Reproduce a coalesced watcher update that is indistinguishable to the + // legacy size/mtime manifest cache while esbuild rebuilds new code. + utimesSync(workflowFile, oldStats.atime, oldStats.mtime); + + const rebuild = await result.interimBundleCtx.rebuild(); + await result.bundleFinal(rebuild); + + const route = readFileSync(config.workflowsBundlePath, 'utf8'); + const currentFiles = readdirSync(workflowBundleDir).filter((file) => + file.endsWith('.mjs') + ); + expect(currentFiles).toHaveLength(1); + const currentFile = currentFiles[0]; + assert(currentFile); + expect(currentFile).not.toBe(oldFile); + expect(route).toContain(`workflow-bundles/${currentFile}`); + expect(route).toContain('Promise.resolve'); + expect(route).toContain('renamedAfterWatch'); + expect(route).not.toContain('removedAfterWatch'); + expect(route).not.toContain('after--watch'); + const currentBundle = await import( + pathToFileURL(join(workflowBundleDir, currentFile)).href + ); + expect(Buffer.from(currentBundle.default, 'base64').toString()).toContain( + 'after--watch' + ); + const graphs = await extractWorkflowGraphs(config.workflowsBundlePath); + expect(JSON.stringify(graphs)).toContain('watched'); + expect(JSON.stringify(graphs)).toContain('renamedAfterWatch'); + expect(JSON.stringify(graphs)).not.toContain('removedAfterWatch'); + } finally { + await Promise.all([ + result.interimBundleCtx.dispose(), + result.stepsContext.dispose(), + ]); + } }); }); diff --git a/packages/builders/src/workflow-bundle-module.ts b/packages/builders/src/workflow-bundle-module.ts new file mode 100644 index 0000000000..ebadccf54c --- /dev/null +++ b/packages/builders/src/workflow-bundle-module.ts @@ -0,0 +1,52 @@ +import assert from 'node:assert/strict'; +import { createHash } from 'node:crypto'; + +export const WORKFLOW_BUNDLE_DIRECTORY = 'workflow-bundles'; + +const WORKFLOW_BUNDLE_FILE = /^\d+(?:-[a-f0-9]{16})?\.mjs$/; + +export function isWorkflowBundleFileName(fileName: string): boolean { + return WORKFLOW_BUNDLE_FILE.test(fileName); +} + +export function findWorkflowBundleFileNames(routeCode: string): string[] { + const prefix = `./${WORKFLOW_BUNDLE_DIRECTORY}/`; + const fileNames = new Set(); + const loaderPatterns = [ + /import\(\s*(['"])([^'"]+)\1\s*\)/g, + /\.bundleFile\s*=\s*(['"])([^'"]+)\1/g, + ]; + for (const pattern of loaderPatterns) { + for (const match of routeCode.matchAll(pattern)) { + const specifier = match[2]; + if (!specifier.startsWith(prefix)) continue; + const fileName = specifier.slice(prefix.length); + if (isWorkflowBundleFileName(fileName)) fileNames.add(fileName); + } + } + return [...fileNames]; +} + +export function workflowBundleFileName(index: number, code: string): string { + const hash = createHash('sha256').update(code).digest('hex').slice(0, 16); + return `${index}-${hash}.mjs`; +} + +export function encodeWorkflowBundle(code: string): string { + return Buffer.from(code, 'utf8').toString('base64'); +} + +export function serializeWorkflowBundle(code: string): string { + // Keep inert VM source opaque to framework plugins. Nitro, for example, + // runs textual global/template transforms over every .mjs file and can + // otherwise rewrite JavaScript that only exists inside the exported string. + return `export default ${JSON.stringify(encodeWorkflowBundle(code))};\n`; +} + +export function deserializeWorkflowBundle(moduleCode: string): string { + const prefix = 'export default '; + assert(moduleCode.startsWith(prefix)); + assert(moduleCode.endsWith(';\n')); + const encoded = JSON.parse(moduleCode.slice(prefix.length, -2)) as string; + return Buffer.from(encoded, 'base64').toString('utf8'); +} diff --git a/packages/builders/src/workflows-extractor.test.ts b/packages/builders/src/workflows-extractor.test.ts index 95fcc73437..448907c467 100644 --- a/packages/builders/src/workflows-extractor.test.ts +++ b/packages/builders/src/workflows-extractor.test.ts @@ -1,7 +1,8 @@ -import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterEach, describe, expect, it, vi } from 'vitest'; +import { serializeWorkflowBundle } from './workflow-bundle-module.js'; import { extractWorkflowGraphs } from './workflows-extractor.js'; describe('extractWorkflowGraphs', () => { @@ -81,4 +82,41 @@ describe('extractWorkflowGraphs', () => { }, }); }); + + it('extracts each lazy workflow source independently', async () => { + tempDir = await mkdtemp(join(tmpdir(), 'workflow-builders-')); + const bundlePath = join(tempDir, 'workflow-bundle.js'); + const bundlesDir = join(tempDir, 'workflow-bundles'); + await mkdir(bundlesDir); + await writeFile( + bundlePath, + `const first = import('./workflow-bundles/0.mjs'); +const second = import('./workflow-bundles/1.mjs'); +const unrelated = './workflow-bundles/9.mjs';` + ); + + const bundle = (file: string, name: string) => + `function ${name}() { return ${JSON.stringify(file)}; }\n${name}.workflowId = "workflow//${file}//${name}";`; + await Promise.all( + ['./first.ts', './second.ts'].map((file, index) => + writeFile( + join(bundlesDir, `${index}.mjs`), + serializeWorkflowBundle(bundle(file, `workflow${index}`)) + ) + ) + ); + + await expect(extractWorkflowGraphs(bundlePath)).resolves.toEqual({ + './first.ts': { + workflow0: expect.objectContaining({ + workflowId: 'workflow//./first.ts//workflow0', + }), + }, + './second.ts': { + workflow1: expect.objectContaining({ + workflowId: 'workflow//./second.ts//workflow1', + }), + }, + }); + }); }); diff --git a/packages/builders/src/workflows-extractor.ts b/packages/builders/src/workflows-extractor.ts index 419421cf38..802dc10840 100644 --- a/packages/builders/src/workflows-extractor.ts +++ b/packages/builders/src/workflows-extractor.ts @@ -1,4 +1,5 @@ import { readFile } from 'node:fs/promises'; +import { dirname, join } from 'node:path'; import type { ArrowFunctionExpression, BlockStatement, @@ -13,6 +14,11 @@ import type { VariableDeclaration, } from '@swc/core'; import { parseSync } from '@swc/core'; +import { + deserializeWorkflowBundle, + findWorkflowBundleFileNames, + WORKFLOW_BUNDLE_DIRECTORY, +} from './workflow-bundle-module.js'; // ============================================================================ // Constants @@ -233,31 +239,44 @@ export async function extractWorkflowGraphs(bundlePath: string): Promise<{ [workflowName: string]: ManifestWorkflowEntry; }; }> { - const bundleCode = await readFile(bundlePath, 'utf-8'); - try { - let actualWorkflowCode = bundleCode; - - const bundleAst = parseSync(bundleCode, { - syntax: 'ecmascript', - target: 'es2022', - }); + const bundleCode = await readFile(bundlePath, 'utf8'); + const lazyBundleDir = join(dirname(bundlePath), WORKFLOW_BUNDLE_DIRECTORY); + const lazyBundleFiles = findWorkflowBundleFileNames(bundleCode).sort( + (left, right) => Number.parseInt(left, 10) - Number.parseInt(right, 10) + ); + const graphs: Record> = {}; + const mergeWorkflowCode = (workflowCode: string) => { + const ast = parseSync(workflowCode, { + syntax: 'ecmascript', + target: 'es2022', + }); + const stepDeclarations = extractStepDeclarations(workflowCode); + const bundleGraphs = extractWorkflows( + ast, + stepDeclarations, + buildFunctionMap(ast, stepDeclarations), + buildVariableMap(ast) + ); + for (const [filePath, workflows] of Object.entries(bundleGraphs)) { + graphs[filePath] = { ...graphs[filePath], ...workflows }; + } + }; - const workflowCodeValue = extractWorkflowCodeFromBundle(bundleAst); - if (workflowCodeValue) { - actualWorkflowCode = workflowCodeValue; + if (lazyBundleFiles.length === 0) { + const bundleAst = parseSync(bundleCode, { + syntax: 'ecmascript', + target: 'es2022', + }); + mergeWorkflowCode(extractWorkflowCodeFromBundle(bundleAst) ?? bundleCode); + } else { + for (const file of lazyBundleFiles) { + const moduleCode = await readFile(join(lazyBundleDir, file), 'utf8'); + mergeWorkflowCode(deserializeWorkflowBundle(moduleCode)); + } } - const ast = parseSync(actualWorkflowCode, { - syntax: 'ecmascript', - target: 'es2022', - }); - - const stepDeclarations = extractStepDeclarations(actualWorkflowCode); - const functionMap = buildFunctionMap(ast, stepDeclarations); - const variableMap = buildVariableMap(ast); - - return extractWorkflows(ast, stepDeclarations, functionMap, variableMap); + return graphs; } catch (error) { console.error('Failed to extract workflow graphs from bundle:', error); return {}; diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 0d478e30dc..44293bbac8 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -20,7 +20,7 @@ import { REPLAY_DIVERGENCE_MAX_RETRIES, } from './runtime/constants.js'; import { setWorld } from './runtime/world.js'; -import { workflowEntrypoint } from './runtime.js'; +import { type WorkflowCode, workflowEntrypoint } from './runtime.js'; import { dehydrateStepArguments, dehydrateStepReturnValue, @@ -62,7 +62,7 @@ type QueueCall = { }; async function runWorkflowHandlerWithEvents( - workflowCode: string, + workflowCode: WorkflowCode, workflowRun: WorkflowRun, events: Event[], options: { @@ -217,6 +217,46 @@ describe('workflowEntrypoint replay guards', () => { throw new Error('workflow code must not execute'); }${getWorkflowTransformCode('workflow')}`; + it('loads only the bundle for the delivered workflow', async () => { + const workflowRun = await misroutedRun(); + const loadWorkflow = vi.fn( + async () => `async function workflow() { + return 'done'; + }${getWorkflowTransformCode('workflow')}` + ); + const loadOtherWorkflow = vi.fn(async () => ''); + + const createdEvents = await runWorkflowHandlerWithEvents( + { + workflow: loadWorkflow, + otherWorkflow: loadOtherWorkflow, + }, + workflowRun, + [] + ); + + expect(loadWorkflow).toHaveBeenCalledOnce(); + expect(loadOtherWorkflow).not.toHaveBeenCalled(); + expect(createdEvents).toContainEqual( + expect.objectContaining({ eventType: 'run_completed' }) + ); + }); + + it('records a lazy bundle load failure on the run', async () => { + const workflowRun = await misroutedRun(); + const loadError = new Error('missing workflow chunk'); + + const createdEvents = await runWorkflowHandlerWithEvents( + { workflow: async () => Promise.reject(loadError) }, + workflowRun, + [] + ); + + expect(createdEvents).toContainEqual( + expect.objectContaining({ eventType: 'run_failed' }) + ); + }); + it('re-routes a flow replay delivered to a different deployment', async () => { const workflowRun = await misroutedRun(); const queueCalls: QueueCall[] = []; @@ -1973,21 +2013,34 @@ describe('workflowEntrypoint resilient step consumption (stepInput re-ensure)', getEncryptionKeyForRun: vi.fn(async () => undefined), } as any); - const handler = workflowEntrypoint(resilientWorkflow); + const loadWorkflow = vi.fn(async () => resilientWorkflow); + const handler = workflowEntrypoint({ workflow: loadWorkflow }); const response = (await handler( new Request('https://example.test') )) as Response; - return { response, createdEvents, createdEventParams, dehydratedInput }; + return { + response, + createdEvents, + createdEventParams, + dehydratedInput, + loadWorkflow, + }; } it('materializes step_created from stepInput on a redelivery before executing', async () => { - const { response, createdEvents, createdEventParams, dehydratedInput } = - await driveStepMessage({ - runId: 'wrun_resilient_step_materialize', - attempt: 2, - }); + const { + response, + createdEvents, + createdEventParams, + dehydratedInput, + loadWorkflow, + } = await driveStepMessage({ + runId: 'wrun_resilient_step_materialize', + attempt: 2, + }); expect(response.status).toBe(204); + expect(loadWorkflow).not.toHaveBeenCalled(); // The re-ensure wrote the step_created with the message's payload… expect(createdEvents).toContainEqual( expect.objectContaining({ diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 0a3dcb8634..80d96d24e0 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -11,6 +11,7 @@ import { RUN_ERROR_CODES, type RunErrorCode, RunExpiredError, + WorkflowNotRegisteredError, WorkflowRuntimeError, WorkflowWorldError, } from '@workflow/errors'; @@ -610,11 +611,25 @@ async function getMaxInlineDurationMs( * The handler loops: replay workflow → execute step inline → replay → ... * until the workflow completes, times out, or encounters non-step suspensions. * - * @param workflowCode - The workflow bundle code containing all workflow functions + * @param workflowCode - A legacy workflow bundle or lazy loaders keyed by workflow ID * @returns A function that can be used as a Vercel API route */ +export type WorkflowCode = + | string + | Readonly Promise>>; + +async function loadWorkflowCode( + workflowCode: WorkflowCode, + workflowName: string +): Promise { + if (typeof workflowCode === 'string') return workflowCode; + const load = workflowCode[workflowName]; + if (!load) throw new WorkflowNotRegisteredError(workflowName); + return load(); +} + export function workflowEntrypoint( - workflowCode: string, + workflowCode: WorkflowCode, options?: { namespace?: string; routeModuleBodyStartedAt?: number; @@ -808,6 +823,20 @@ export function workflowEntrypoint( return await withWorkflowBaggage( { workflowRunId: runId, workflowName }, async () => { + let loadedWorkflowCode: string | undefined; + let workflowCodeLoad: Promise | undefined; + const startWorkflowCodeLoad = (): Promise => { + workflowCodeLoad ??= trace('workflow.bundle.load', () => + loadWorkflowCode(workflowCode, workflowName) + ).then((code) => { + loadedWorkflowCode = code; + return code; + }); + void workflowCodeLoad.catch(() => {}); + return workflowCodeLoad; + }; + if (incomingStepId === undefined) startWorkflowCodeLoad(); + const world = await trace('workflow.route.get_world', async () => getWorld() ); @@ -1921,6 +1950,11 @@ export function workflowEntrypoint( } } + // Queue-only step deliveries usually return above and never + // replay. Start loading only once this invocation is known to + // need the workflow VM, while run/event setup is still ahead. + const workflowCodePromise = startWorkflowCodeLoad(); + // Deployment-affinity pre-check for the lazy hook fast // path below. New lazy-resume messages carry the run's // pinned deployment (`hookInput.deploymentId`), so a @@ -2726,7 +2760,7 @@ export function workflowEntrypoint( './runtime/quickjs-entrypoint.js' ); const quickjsResult = await runWorkflowWithQuickJS({ - workflowCode, + workflowCode: await workflowCodePromise, workflowName, workflowRun, preloadedEvents: @@ -3020,7 +3054,7 @@ export function workflowEntrypoint( if (workflowResult.type === 'replay') { retainedSession = null; workflowResult = await replayWorkflow({ - workflowCode, + workflowCode: await workflowCodePromise, workflowRun, events: eventLog.events, encryptionKey, @@ -4400,14 +4434,14 @@ export function workflowEntrypoint( let errorStack = normalizedError.stack || getErrorStack(terminalError); - if (errorStack) { + if (errorStack && loadedWorkflowCode) { const parsedName = parseWorkflowName(workflowName); const filename = parsedName?.moduleSpecifier || workflowName; errorStack = remapErrorStack( errorStack, filename, - workflowCode + loadedWorkflowCode ); } diff --git a/packages/core/src/vm/script-cache.test.ts b/packages/core/src/vm/script-cache.test.ts index 1dc2b34db5..7e1868a79e 100644 --- a/packages/core/src/vm/script-cache.test.ts +++ b/packages/core/src/vm/script-cache.test.ts @@ -1,5 +1,5 @@ import { type Context, runInContext } from 'node:vm'; -import { afterEach, describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import { createContext } from './index.js'; import { clearWorkflowScriptCache, @@ -47,6 +47,7 @@ function runScript(code: string, filename: string, context: Context) { describe('script-cache', () => { afterEach(() => { clearWorkflowScriptCache(); + vi.unstubAllEnvs(); }); it('returns the same compiled Script for identical (code, filename)', () => { @@ -147,6 +148,14 @@ describe('script-cache', () => { expect(getScript(latest, filename)).toBe(getScript(latest, filename)); }); + it('retains every immutable source bundle in production', () => { + vi.stubEnv('NODE_ENV', 'production'); + for (let i = 0; i < 12; i++) { + getScript(buildBundle(`source-${i}`), `source-${i}.ts`); + } + expect(workflowScriptCacheSize()).toBe(12); + }); + it('keeps the most-recently-used bundle and evicts the stale one', () => { const filename = 'workflows/a.ts'; // Seed an "old" bundle, then keep it hot by re-touching it while many diff --git a/packages/core/src/vm/script-cache.ts b/packages/core/src/vm/script-cache.ts index 85f4ad3014..cf315458f7 100644 --- a/packages/core/src/vm/script-cache.ts +++ b/packages/core/src/vm/script-cache.ts @@ -7,9 +7,8 @@ import { Script } from 'node:vm'; * --------------- * Replaying a workflow re-evaluates the workflow bundle against a fresh VM * context on every iteration of the inline replay loop (see - * `runWorkflow` in `../workflow.ts`). The bundle is a single string that - * contains every workflow function in the app and registers them on - * `globalThis.__private_workflows`. Previously each replay called + * `runWorkflow` in `../workflow.ts`). Each source bundle registers its + * workflow functions on `globalThis.__private_workflows`. Previously each replay called * `vm.runInContext(workflowCode, context, { filename })`, which RE-PARSES and * RE-COMPILES the entire bundle every time — O(N) full re-parses for a * sequential workflow of N steps, plus the same parse cost repeated across @@ -44,32 +43,23 @@ import { Script } from 'node:vm'; * * Bounding * -------- - * The top-level (`code`-keyed) map is an insertion-ordered LRU capped at - * `MAX_BUNDLES` entries. In production this bound is never reached: a - * deployment is its own process serving exactly one build-time bundle literal - * (skew protection runs old versions as separate processes), so there is a - * single `code` key for the process lifetime. The bound exists for dev/watch - * mode, where the dev route re-reads `workflowCode` from disk and re-invokes - * the entrypoint on every edit — each edit produces a NEW bundle string, which + * In production, a deployment's immutable set of source bundles naturally + * bounds this cache. In dev/watch mode, the top-level (`code`-keyed) map is an + * insertion-ordered LRU capped at `MAX_DEV_BUNDLES`: the dev route re-reads + * `workflowCode` from disk and re-invokes the entrypoint on every edit. Each + * edit produces a NEW bundle string, which * without a bound would pin every historical version forever (~0.8MB per edit, - * growing monotonically with edit count). The dev path only ever needs the - * latest bundle, so an LRU that keeps the few most-recent bundles and evicts - * the rest preserves the pre-cache GC behaviour while still serving the - * steady-state single-bundle case for free. The per-`filename` inner map is not - * separately bounded: it is naturally bounded by the (small) number of workflow - * source files in a bundle and is dropped wholesale when its parent `code` - * entry is evicted. + * growing monotonically with edit count). A small LRU preserves the pre-cache + * GC behaviour while keeping recently exercised sources hot. The per-filename + * inner map is dropped wholesale when its parent `code` entry is evicted. */ const scriptCache = new Map>(); /** - * Max number of distinct bundle (`code`) versions to retain. One is enough for - * production; a handful covers pathological dev hot-reload / repeated-rebuild - * churn within a single long-lived process (e.g. a watch session or a test - * file) without unbounded growth. Kept deliberately small — there is no value - * in retaining stale bundles, only a memory cost. + * Max number of distinct bundle versions retained outside production. There is + * no value in pinning every stale dev build. */ -const MAX_BUNDLES = 8; +const MAX_DEV_BUNDLES = 8; /** * Looks up the per-filename map for `code`, marking it most-recently-used. @@ -108,7 +98,10 @@ export function getCachedWorkflowScript( scriptCache.set(code, byFilename); // Evict the least-recently-used bundle(s) when over the cap. New bundles // are appended at the end, so the oldest live at the front. - while (scriptCache.size > MAX_BUNDLES) { + while ( + process.env.NODE_ENV !== 'production' && + scriptCache.size > MAX_DEV_BUNDLES + ) { const oldest = scriptCache.keys().next().value; if (oldest === undefined) { break; diff --git a/packages/next/src/builder-eager.ts b/packages/next/src/builder-eager.ts index 15e1ace9bb..62467b9e3a 100644 --- a/packages/next/src/builder-eager.ts +++ b/packages/next/src/builder-eager.ts @@ -240,14 +240,7 @@ export async function getNextBuilderEager( } const workflowResult = await workflowsCtx.interimBundleCtx.rebuild(); - const workflowOutput = workflowResult.outputFiles?.[0]?.text; - if (!workflowOutput) { - throw new Error( - 'Invariant: expected workflow output from hot rebuild' - ); - } - - await workflowsCtx.bundleFinal(workflowOutput); + await workflowsCtx.bundleFinal(workflowResult); await writeManifest(mergeCombinedManifest(stepsManifest)); }; @@ -267,6 +260,11 @@ export async function getNextBuilderEager( sourceSnapshots, rebuild: async () => { this.clearDiscoveredEntriesCache(); + // A definition-level change can preserve both file size and an + // effectively identical mtime on fast/coalesced dev writes. A + // full rediscovery must never reuse manifests from the previous + // graph. + this.clearManifestTransformCache(); const newInputFiles = await this.getInputFiles(); options.inputFiles = newInputFiles; From fcb732a9acfc592738aedb6f05d9603f5eab320e Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Fri, 14 Aug 2026 02:16:51 -0700 Subject: [PATCH 2/9] fix(world-testing): publish lazy workflow bundles --- .changeset/lazy-workflow-source-bundles.md | 1 + packages/world-testing/package.json | 2 +- .../test/bundle-artifacts.test.ts | 19 +++++++++++++++++++ 3 files changed, 21 insertions(+), 1 deletion(-) create mode 100644 packages/world-testing/test/bundle-artifacts.test.ts diff --git a/.changeset/lazy-workflow-source-bundles.md b/.changeset/lazy-workflow-source-bundles.md index 295e0a18b6..1f42c719bc 100644 --- a/.changeset/lazy-workflow-source-bundles.md +++ b/.changeset/lazy-workflow-source-bundles.md @@ -2,6 +2,7 @@ '@workflow/builders': patch '@workflow/core': patch '@workflow/next': patch +'@workflow/world-testing': patch --- Load only the selected workflow source bundle when replay starts. diff --git a/packages/world-testing/package.json b/packages/world-testing/package.json index f72378544d..4fd8dd19e5 100644 --- a/packages/world-testing/package.json +++ b/packages/world-testing/package.json @@ -16,7 +16,7 @@ "directory": "packages/world-testing" }, "scripts": { - "build": "wf build && node scripts/generate-well-known-dts.mjs && tsc && cp .well-known/workflow/v1/*.mjs dist/.well-known/workflow/v1/", + "build": "wf build && node scripts/generate-well-known-dts.mjs && tsc && cp -R .well-known/workflow/v1/. dist/.well-known/workflow/v1/", "clean": "tsc --build --clean && rm -rf dist .well-known* .workflow-data", "start": "node --watch src/server.mts", "test": "vitest run" diff --git a/packages/world-testing/test/bundle-artifacts.test.ts b/packages/world-testing/test/bundle-artifacts.test.ts new file mode 100644 index 0000000000..9b5aa04224 --- /dev/null +++ b/packages/world-testing/test/bundle-artifacts.test.ts @@ -0,0 +1,19 @@ +import { readdir } from 'node:fs/promises'; +import { expect, test } from 'vitest'; + +test('publishes every lazy workflow bundle with the flow route', async () => { + const bundlesDirectory = new URL( + '../dist/.well-known/workflow/v1/workflow-bundles/', + import.meta.url + ); + const files = (await readdir(bundlesDirectory)).filter((file) => + file.endsWith('.mjs') + ); + + expect(files.length).toBeGreaterThan(0); + + for (const file of files) { + const bundle = await import(new URL(file, bundlesDirectory).href); + expect(bundle.default).toEqual(expect.any(String)); + } +}); From 5f7d9e07b394be4911d37cfa76cfa91444bfd866 Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Fri, 14 Aug 2026 03:03:58 -0700 Subject: [PATCH 3/9] test(core): settle HMR before workflow execution --- packages/core/e2e/dev.test.ts | 30 +++++++++++++----------------- 1 file changed, 13 insertions(+), 17 deletions(-) diff --git a/packages/core/e2e/dev.test.ts b/packages/core/e2e/dev.test.ts index bd5cf7841d..141238417d 100644 --- a/packages/core/e2e/dev.test.ts +++ b/packages/core/e2e/dev.test.ts @@ -1223,10 +1223,22 @@ export function hmrFuzzWorkflowHelper(value: HmrFuzzBox) { const logCursor = await readDevServerLogCursor(); await fs.writeFile(testCase.file, testCase.source(iteration)); + await expectHmrLogCounts(logCursor, testCase.expectedLogCounts); + snapshot = await waitForGeneratedArtifactStability(); + if (testCase.kind === 'workflow') { + expect(snapshot.stepMtimeMs).toBe(previousSnapshot.stepMtimeMs); + } else if (testCase.kind !== 'none') { + expect(snapshot.stepMtimeMs).toBeGreaterThanOrEqual( + previousSnapshot.stepMtimeMs + ); + } + // Next canary can keep executing a stale workflow bundle after the // workflow hot-rebuild completed. Stable still covers execution // correctness; canary keeps covering classification/log/artifact - // behavior for these changes. + // behavior for these changes. Wait for the rebuild above before + // starting runs so a slow dev compiler does not turn polling into a + // burst of concurrent workflow executions. if (!(finalConfig.canary && testCase.kind === 'workflow')) { await expectWorkflowResult({ description: `${testCase.kind} HMR update to affect workflow execution`, @@ -1240,22 +1252,6 @@ export function hmrFuzzWorkflowHelper(value: HmrFuzzBox) { : undefined, }); } - - if (testCase.kind === 'none') { - await expectHmrLogCounts(logCursor, testCase.expectedLogCounts); - snapshot = await waitForGeneratedArtifactStability(); - continue; - } - - snapshot = await waitForGeneratedArtifactStability(); - if (testCase.kind === 'workflow') { - expect(snapshot.stepMtimeMs).toBe(previousSnapshot.stepMtimeMs); - } else { - expect(snapshot.stepMtimeMs).toBeGreaterThanOrEqual( - previousSnapshot.stepMtimeMs - ); - } - await expectHmrLogCounts(logCursor, testCase.expectedLogCounts); } const fullCases = [ From d9136d8e69b9bf09649319c74def226830a137bc Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Fri, 14 Aug 2026 03:10:15 -0700 Subject: [PATCH 4/9] test(core): preserve HMR log coverage --- packages/core/e2e/dev.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/core/e2e/dev.test.ts b/packages/core/e2e/dev.test.ts index 141238417d..7695a92db2 100644 --- a/packages/core/e2e/dev.test.ts +++ b/packages/core/e2e/dev.test.ts @@ -1252,6 +1252,7 @@ export function hmrFuzzWorkflowHelper(value: HmrFuzzBox) { : undefined, }); } + await expectHmrLogCounts(logCursor, testCase.expectedLogCounts); } const fullCases = [ From 59540c918ae9fc2ef9a27b01b0bd87027e0c3aab Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Fri, 14 Aug 2026 03:25:18 -0700 Subject: [PATCH 5/9] test(core): tolerate duplicate HMR skip events --- packages/core/e2e/dev.test.ts | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/packages/core/e2e/dev.test.ts b/packages/core/e2e/dev.test.ts index 7695a92db2..9dd1eb5b8d 100644 --- a/packages/core/e2e/dev.test.ts +++ b/packages/core/e2e/dev.test.ts @@ -1111,7 +1111,11 @@ ${apiFileContent}` { file: files.step, kind: 'none', - expectedLogCounts: { skip: 1 }, + expectedLogCounts: { + skip: { min: 1 }, + hot: { max: 0 }, + full: { max: 0 }, + }, expectedStepValue: (iteration: number) => `step-only-${iteration}`, source: ( iteration: number @@ -1127,7 +1131,11 @@ export async function hmrFuzzStep() { { file: files.stepHelper, kind: 'none', - expectedLogCounts: { skip: 1 }, + expectedLogCounts: { + skip: { min: 1 }, + hot: { max: 0 }, + full: { max: 0 }, + }, expectedStepValue: (iteration: number) => `step-helper-only-${iteration}`, source: ( From c8ea4226fddb8187b7ffc5838af6e24c7f4d6542 Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Fri, 14 Aug 2026 03:26:45 -0700 Subject: [PATCH 6/9] test(core): await full HMR rebuild completion --- packages/core/e2e/dev.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/core/e2e/dev.test.ts b/packages/core/e2e/dev.test.ts index 9dd1eb5b8d..37c094d1c8 100644 --- a/packages/core/e2e/dev.test.ts +++ b/packages/core/e2e/dev.test.ts @@ -1437,7 +1437,6 @@ ${apiFileContent}` const fullCase = fullCases[index]; const logCursor = await readDevServerLogCursor(); await fullCase.write(index + 1); - await fullCase.assert(index + 1); await expectHmrLogCounts( logCursor, 'expectedLogCounts' in fullCase @@ -1445,6 +1444,7 @@ ${apiFileContent}` : { full: 1 } ); snapshot = await waitForGeneratedArtifactStability(); + await fullCase.assert(index + 1); } const unrelatedLogCursor = await readDevServerLogCursor(); From 90be320f055d254206eda29e6c78784022dadb02 Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Fri, 14 Aug 2026 03:36:22 -0700 Subject: [PATCH 7/9] fix(next): log HMR after rebuild completion --- packages/next/src/builder-eager.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/packages/next/src/builder-eager.ts b/packages/next/src/builder-eager.ts index 62467b9e3a..de601e5fd5 100644 --- a/packages/next/src/builder-eager.ts +++ b/packages/next/src/builder-eager.ts @@ -467,10 +467,10 @@ export async function getNextBuilderEager( return; } if (decision.kind === 'full') { - logDevHmr('workflow dev hmr: full rediscovery'); try { await fullRebuild(); await refreshKnownFiles(); + logDevHmr('workflow dev hmr: full rediscovery'); } finally { // Lets a log reader tell "quiet" from "rebuild in flight". // The e2e HMR tests drain-to-quiet before counting lines. @@ -479,14 +479,14 @@ export async function getNextBuilderEager( return; } - logDevHmr( - `workflow dev hmr: hot rebuild${decision.refreshStepRegistrations ? ' with step registration refresh' : ''}` - ); try { await hotRebuild(decision.refreshStepRegistrations); for (const [file, snapshot] of decision.snapshots) { sourceSnapshots.set(file, snapshot); } + logDevHmr( + `workflow dev hmr: hot rebuild${decision.refreshStepRegistrations ? ' with step registration refresh' : ''}` + ); } finally { // See the matching line on the full path above. logDevHmr('workflow dev hmr: rebuild complete'); From b69760b8e8d1da089f273173ca01c498f0e8aa5e Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Fri, 14 Aug 2026 03:46:33 -0700 Subject: [PATCH 8/9] fix(next): publish HMR manifests after watcher state --- packages/next/src/builder-eager.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/next/src/builder-eager.ts b/packages/next/src/builder-eager.ts index de601e5fd5..074df9eb16 100644 --- a/packages/next/src/builder-eager.ts +++ b/packages/next/src/builder-eager.ts @@ -287,8 +287,9 @@ export async function getNextBuilderEager( bundleFinal: newCombined.bundleFinal, }; - await writeManifest(newCombined.manifest); await refreshSourceSnapshots(); + await refreshKnownFiles(); + await writeManifest(newCombined.manifest); }, }); @@ -469,7 +470,6 @@ export async function getNextBuilderEager( if (decision.kind === 'full') { try { await fullRebuild(); - await refreshKnownFiles(); logDevHmr('workflow dev hmr: full rediscovery'); } finally { // Lets a log reader tell "quiet" from "rebuild in flight". From 2bbe211985de380b630692a40ab6143e4ef111ce Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Fri, 14 Aug 2026 23:04:22 -0700 Subject: [PATCH 9/9] refactor(world-vercel): use v4 event listing directly --- packages/world-vercel/src/events.test.ts | 18 +++++++++--------- packages/world-vercel/src/events.ts | 12 +----------- packages/world-vercel/src/storage.ts | 4 ++-- 3 files changed, 12 insertions(+), 22 deletions(-) diff --git a/packages/world-vercel/src/events.test.ts b/packages/world-vercel/src/events.test.ts index 87ba906319..901ca559c6 100644 --- a/packages/world-vercel/src/events.test.ts +++ b/packages/world-vercel/src/events.test.ts @@ -8,10 +8,10 @@ import { MockAgent } from 'undici'; import { describe, expect, it, vi } from 'vitest'; import { createWorkflowRunEvent, - getWorkflowRunEvents, getWorkflowRunEventsByCorrelationId, splitEventDataForV4, } from './events.js'; +import { getWorkflowRunEventsV4 } from './events-v4.js'; import { encodeFrame, V4_FRAME_CONTENT_TYPE } from './frames.js'; import { encode as encodeRunId, REGION_IDS } from './run-id/index.js'; import { WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js'; @@ -1241,7 +1241,7 @@ describe('getWorkflowRunEvents remoteRefBehavior mapping', () => { headers: { 'content-type': V4_FRAME_CONTENT_TYPE }, }); - const result = await getWorkflowRunEvents( + const result = await getWorkflowRunEventsV4( { runId: 'wrun_1', resolveData: 'none' }, { token: 'test-token', dispatcher: agent } ); @@ -1276,7 +1276,7 @@ describe('getWorkflowRunEvents remoteRefBehavior mapping', () => { }); // No resolveData → defaults to 'all' → resolve. - const result = await getWorkflowRunEvents( + const result = await getWorkflowRunEventsV4( { runId: 'wrun_1' }, { token: 'test-token', dispatcher: agent } ); @@ -1301,7 +1301,7 @@ describe('getWorkflowRunEvents remoteRefBehavior mapping', () => { headers: { 'content-type': V4_FRAME_CONTENT_TYPE }, }); - await getWorkflowRunEvents( + await getWorkflowRunEventsV4( { runId: 'wrun_1' }, { token: 'test-token', dispatcher: agent } ); @@ -1337,7 +1337,7 @@ describe('getWorkflowRunEvents remoteRefBehavior mapping', () => { }); await expect( - getWorkflowRunEvents( + getWorkflowRunEventsV4( { runId: 'wrun_1' }, { token: 'test-token', dispatcher: agent } ) @@ -1389,7 +1389,7 @@ describe('getWorkflowRunEvents legacy structured-error compatibility', () => { headers: { 'content-type': V4_FRAME_CONTENT_TYPE }, }); - const result = await getWorkflowRunEvents( + const result = await getWorkflowRunEventsV4( { runId: 'wrun_1', resolveData: 'all' }, { token: 'test-token', dispatcher: agent } ); @@ -1470,7 +1470,7 @@ describe('getWorkflowRunEvents hasMore mapping', () => { const agent = mockAgent(); mockListResponse(agent, { _end: 1, next: 'eid:last', hasMore: false }); - const result = await getWorkflowRunEvents( + const result = await getWorkflowRunEventsV4( { runId: 'wrun_1' }, { token: 'test-token', dispatcher: agent } ); @@ -1490,7 +1490,7 @@ describe('getWorkflowRunEvents hasMore mapping', () => { { limit: '500', remoteRefBehavior: 'resolve' } ); - const result = await getWorkflowRunEvents( + const result = await getWorkflowRunEventsV4( { runId: 'wrun_1', pagination: { limit: 500 } }, { token: 'test-token', dispatcher: agent } ); @@ -1504,7 +1504,7 @@ describe('getWorkflowRunEvents hasMore mapping', () => { mockListResponse(agent, { _end: 1, next: 'cursor-2' }); await expect( - getWorkflowRunEvents( + getWorkflowRunEventsV4( { runId: 'wrun_1' }, { token: 'test-token', dispatcher: agent } ) diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index e4f10b6ae0..beb9617eed 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -49,7 +49,6 @@ import { getEventDataPayloadField, isHookEventRequiringExistence, type ListEventsByCorrelationIdParams, - type ListEventsParams, type PaginatedResponse, validateUlidTimestamp, type WorkflowRun, @@ -62,7 +61,6 @@ import { createWorkflowRunStartedEventV4, getEventsByCorrelationIdV4, getEventV4, - getWorkflowRunEventsV4, } from './events-v4.js'; import { decode as decodeRunId } from './run-id/index.js'; import { cancelWorkflowRunV1, createWorkflowRunV1 } from './runs.js'; @@ -230,8 +228,7 @@ assertEventDataWireContractExhaustive<[Unhandled, Stale]>(); * CBOR-encoded meta block of the same frame. * * Exported for unit tests (the meta allowlist is the eventData wire - * contract — see the warning on EVENT_DATA_PAYLOAD_FIELD_BY_EVENT_TYPE in - * @workflow/world). + * contract — see getEventDataPayloadField in @workflow/world). */ export function splitEventDataForV4(data: AnyEventRequest): SplitEventData { // Some event types in the AnyEventRequest discriminated union (e.g. @@ -431,13 +428,6 @@ export async function getEvent( ); } -export async function getWorkflowRunEvents( - params: ListEventsParams, - config?: APIConfig -): Promise> { - return getWorkflowRunEventsV4(params, config); -} - export async function getWorkflowRunEventsByCorrelationId( params: ListEventsByCorrelationIdParams, config?: APIConfig diff --git a/packages/world-vercel/src/storage.ts b/packages/world-vercel/src/storage.ts index 72abe6c57b..e6e2c4d403 100644 --- a/packages/world-vercel/src/storage.ts +++ b/packages/world-vercel/src/storage.ts @@ -7,9 +7,9 @@ import { createWorkflowRunEvent, createWorkflowRunEventBatch, getEvent, - getWorkflowRunEvents, getWorkflowRunEventsByCorrelationId, } from './events.js'; +import { getWorkflowRunEventsV4 } from './events-v4.js'; import { getHook, getHookByToken, listHooks } from './hooks.js'; import { instrumentObject } from './instrumentObject.js'; import { @@ -53,7 +53,7 @@ export function createStorage(config?: APIConfig): Storage { createBatch: (runId, events, params) => createWorkflowRunEventBatch(runId, events, params, config), get: (runId, eventId, params) => getEvent(runId, eventId, params, config), - list: (params) => getWorkflowRunEvents(params, config), + list: (params) => getWorkflowRunEventsV4(params, config), listByCorrelationId: (params) => getWorkflowRunEventsByCorrelationId(params, config), },