Add explicit workflow handler initialization

.changeset/eager-direct-queue-registration.md

@@ -0,0 +1,7 @@
+---
+'@workflow/core': patch
+'@workflow/world': patch
+'workflow': patch
+---
+
+Allow generated flow routes to register direct queue handlers before workers start.

packages/core/e2e/e2e.test.ts

@@ -2930,7 +2930,7 @@ describe.concurrent('e2e', () => {
   // queue-based `healthCheck(world, options)` function instead, which
   // bypasses protection by sending messages through the Queue infrastructure.
   test.skipIf(!isLocalDeployment())(
-    'health check endpoint (HTTP) - workflow endpoint responds to __health query parameter',
+    'health check endpoint (HTTP) - follows the World delivery capability',
     { timeout: 30_000 },
     async () => {
       // NOTE: This tests the HTTP-based health check using the `?__health` query parameter.
@@ -2940,14 +2940,19 @@ describe.concurrent('e2e', () => {
       // queue-based `healthCheck(world, options)` function instead, which
       // bypasses protection by sending messages through the Queue infrastructure.
 
-      // Test the flow endpoint health check (V2: combined handler for both workflow + step)
+      const world = await getWorld();
       const flowRes = await fetch(
         createWorkflowUrl(deploymentUrl, { type: 'health' }),
         {
           method: 'POST',
           headers: await getTrustedSourcesHeaders(),
         }
       );
+      if (world.capabilities?.directQueueDelivery === true) {
+        expect(flowRes.status).toBe(404);
+        return;
+      }
+
       expect(flowRes.status).toBe(200);
       expect(flowRes.headers.get('Content-Type')).toBe('application/json');
       const { workflowCoreVersion, ...flowBody } = await flowRes.json();

packages/core/src/runtime-world-singleton.test.ts

@@ -1,3 +1,4 @@
+import { withResolvers } from '@workflow/utils';
 import { SPEC_VERSION_CURRENT, type World } from '@workflow/world';
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { getWorld, setWorld } from './runtime/world.js';
@@ -37,4 +38,64 @@ describe('workflowEntrypoint world initialization', () => {
     await expect(getWorld()).resolves.toBe(world);
     expect(createLocalWorld).toHaveBeenCalledTimes(1);
   });
+
+  it('registers a handler before the first request when initialized', async () => {
+    const entrypoint = workflowEntrypoint('');
+
+    await entrypoint.initialize();
+
+    expect(world.createQueueHandler).toHaveBeenCalledOnce();
+  });
+
+  it('does not expose the public health response for direct delivery', async () => {
+    const directWorld = {
+      ...world,
+      capabilities: { directQueueDelivery: true },
+    } as unknown as World;
+    createLocalWorld.mockResolvedValue(directWorld);
+
+    const response = await workflowEntrypoint('')(
+      new Request('https://example.test/?__health')
+    );
+
+    expect(response.status).toBe(204);
+  });
+
+  it('rebinds an initialized entrypoint after setWorld()', async () => {
+    const entrypoint = workflowEntrypoint('');
+    await entrypoint.initialize();
+    const nextWorld = {
+      ...world,
+      createQueueHandler: vi.fn(
+        () => async () => new Response(null, { status: 204 })
+      ),
+    } as unknown as World;
+
+    setWorld(nextWorld);
+    await entrypoint.initialize();
+
+    expect(nextWorld.createQueueHandler).toHaveBeenCalledOnce();
+  });
+
+  it('uses a direct World that replaces in-flight initialization', async () => {
+    const pendingWorld = withResolvers<World>();
+    createLocalWorld.mockReturnValueOnce(pendingWorld.promise);
+    const directWorld = {
+      ...world,
+      capabilities: { directQueueDelivery: true },
+      createQueueHandler: vi.fn(
+        () => async () => new Response(null, { status: 404 })
+      ),
+    } as unknown as World;
+    const response = workflowEntrypoint('')(
+      new Request('https://example.test/?__health')
+    );
+    await vi.waitFor(() => expect(createLocalWorld).toHaveBeenCalledOnce());
+
+    setWorld(directWorld);
+    pendingWorld.resolve(world);
+
+    await expect(response).resolves.toMatchObject({ status: 404 });
+    expect(directWorld.createQueueHandler).toHaveBeenCalledOnce();
+  });
 });

packages/core/src/runtime.ts

@@ -118,7 +118,11 @@ import {
 } from './runtime/suspension-handler.js';
 import { useQuickJSVm } from './runtime/vm-mode.js';
 import { getWaitContinuationDispatch } from './runtime/wait-continuation.js';
-import { getWorld, type WorldHandlers } from './runtime/world.js';
+import {
+  getWorld,
+  getWorldGeneration,
+  type WorldHandlers,
+} from './runtime/world.js';
 import { dehydrateRunError } from './serialization.js';
 import { remapErrorStack } from './source-map.js';
 import * as Attribute from './telemetry/semantic-conventions.js';
@@ -681,14 +685,19 @@ async function getMaxInlineDurationMs(
  * @param workflowCode - The workflow bundle code containing all workflow functions
  * @returns A function that can be used as a Vercel API route
  */
+export interface WorkflowEntrypoint {
+  (req: Request): Promise<Response>;
+  initialize(): Promise<void>;
+}
+
 export function workflowEntrypoint(
   workflowCode: string,
   options?: {
     namespace?: string;
     routeModuleBodyStartedAt?: number;
     basePath?: string;
   }
-): (req: Request) => Promise<Response> {
+): WorkflowEntrypoint {
   setWorkflowBasePath(options?.basePath);
 
   const namespace = resolveQueueNamespace(options?.namespace);
@@ -4856,17 +4865,55 @@ export function workflowEntrypoint(
       }
     );
 
-  let cachedHandler: ((req: Request) => Promise<Response>) | undefined;
+  type InitializedHandler = (req: Request) => Promise<Response>;
+  type HandlerCache = {
+    generation: number;
+    promise: Promise<InitializedHandler>;
+  };
+
+  let handlerCache: HandlerCache | undefined;
   let invocationCount = 0;
   const entrypointCreatedAt = Date.now();
   const routeModuleBodyInitMs =
     typeof options?.routeModuleBodyStartedAt === 'number'
       ? entrypointCreatedAt - options.routeModuleBodyStartedAt
       : undefined;
 
-  return withHealthCheck(async (req) => {
+  async function initializeHandler(): Promise<InitializedHandler> {
+    const generation = getWorldGeneration();
+    let cache = handlerCache;
+    if (!cache || cache.generation !== generation) {
+      const promise = trace('workflow.route.init', async () => {
+        const worldHandlers = await trace(
+          'workflow.route.get_world_handlers',
+          async () => getWorld()
+        );
+        if (generation !== getWorldGeneration()) return initializeHandler();
+        const queueHandler = handler(worldHandlers);
+        return worldHandlers.capabilities?.directQueueDelivery === true
+          ? queueHandler
+          : withHealthCheck(queueHandler);
+      });
+      cache = { generation, promise };
+      handlerCache = cache;
+    }
+
+    try {
+      const initialized = await cache.promise;
+      if (cache.generation !== getWorldGeneration()) {
+        if (handlerCache === cache) handlerCache = undefined;
+        return initializeHandler();
+      }
+      return initialized;
+    } catch (error) {
+      if (handlerCache === cache) handlerCache = undefined;
+      throw error;
+    }
+  }
+
+  const invokeQueueHandler = async (req: Request): Promise<Response> => {
     invocationCount += 1;
-    const handlerCached = cachedHandler !== undefined;
+    const handlerCached = handlerCache?.generation === getWorldGeneration();
     const spanKind = await getSpanKind('SERVER');
 
     return trace(
@@ -4889,27 +4936,8 @@ export function workflowEntrypoint(
         },
       },
       async (span) => {
-        if (!cachedHandler) {
-          cachedHandler = await trace('workflow.route.init', async () => {
-            // The full runtime World, not `getWorldHandlers()`. That accessor
-            // owns a second, build-time-safe cache, so calling it here built a
-            // second World in the same process: duplicate connection pools and
-            // queue workers for a stateful World, plus a second copy of that
-            // world package's modules once it is bundled, which is what
-            // silently demoted the events WebSocket transport to HTTP. #3665.
-            //
-            // The span keeps its original name. It is a distinct span from the
-            // per-request `workflow.route.get_world` at the top of the flow
-            // route, and renaming it would collide with that one.
-            const worldHandlers = await trace(
-              'workflow.route.get_world_handlers',
-              async () => getWorld()
-            );
-            return handler(worldHandlers);
-          });
-        }
-
-        const response = await cachedHandler(req);
+        const initializedHandler = await initializeHandler();
+        const response = await initializedHandler(req);
         if (response instanceof Response) {
           span?.setAttributes(
             Attribute.HttpResponseStatusCode(response.status)
@@ -4918,5 +4946,11 @@ export function workflowEntrypoint(
         return response;
       }
     );
+  };
+  const entrypoint: WorkflowEntrypoint = Object.assign(invokeQueueHandler, {
+    initialize: async () => {
+      await initializeHandler();
+    },
   });
+  return entrypoint;
 }

packages/core/src/runtime/world.ts

@@ -45,14 +45,20 @@ const WorldCachePromise = Symbol.for('@workflow/world//cachePromise');
 const StubbedWorldCachePromise = Symbol.for(
   '@workflow/world//stubbedCachePromise'
 );
+const WorldCacheGeneration = Symbol.for('@workflow/world//cacheGeneration');
 
 const globalSymbols: typeof globalThis & {
   [WorldCache]?: World;
   [StubbedWorldCache]?: World;
   [WorldCachePromise]?: Promise<World>;
   [StubbedWorldCachePromise]?: Promise<World>;
+  [WorldCacheGeneration]?: number;
 } = globalThis;
 
+export function getWorldGeneration(): number {
+  return globalSymbols[WorldCacheGeneration] ?? 0;
+}
+
 export type WorldFactoryModule = {
   createWorld?: () => World | Promise<World>;
   default?: (() => World | Promise<World>) | World;
@@ -243,21 +249,29 @@ export const getWorld = async (): Promise<World> => {
   // Store the promise immediately to prevent race conditions with concurrent calls.
   // Clear on rejection so subsequent calls can retry instead of caching the failure.
   if (!globalSymbols[WorldCachePromise]) {
-    globalSymbols[WorldCachePromise] = createWorld().catch((err) => {
-      globalSymbols[WorldCachePromise] = undefined;
+    let pendingWorld!: Promise<World>;
+    pendingWorld = createWorld().catch((err) => {
+      if (globalSymbols[WorldCachePromise] === pendingWorld) {
+        globalSymbols[WorldCachePromise] = undefined;
+      }
       throw err;
     });
+    globalSymbols[WorldCachePromise] = pendingWorld;
   }
-  globalSymbols[WorldCache] = await globalSymbols[WorldCachePromise];
-  assertWorldSupportsRuntimeProtocol(globalSymbols[WorldCache]);
-  return globalSymbols[WorldCache];
+  const worldPromise = globalSymbols[WorldCachePromise];
+  const world = await worldPromise;
+  if (globalSymbols[WorldCachePromise] !== worldPromise) return getWorld();
+  globalSymbols[WorldCache] = world;
+  assertWorldSupportsRuntimeProtocol(world);
+  return world;
 };
 
 /**
  * Reset the cached world instance. This should be called when environment
  * variables change and you need to reinitialize the world with new config.
  */
 export const setWorld = (world: World | undefined): void => {
+  globalSymbols[WorldCacheGeneration] = getWorldGeneration() + 1;
   globalSymbols[WorldCache] = world;
   globalSymbols[StubbedWorldCache] = world;
   globalSymbols[WorldCachePromise] = undefined;

packages/workflow/src/runtime.ts

@@ -10,5 +10,6 @@ export {
   type HealthCheckResult,
   healthCheck,
   setWorld,
+  type WorkflowEntrypoint,
   workflowEntrypoint,
 } from '@workflow/core/runtime';

packages/world/src/interfaces.ts

@@ -466,6 +466,12 @@ export interface Storage {
  * explicitly declares it.
  */
 export interface WorldCapabilities {
+  /**
+   * Queue messages are delivered to an in-process handler. Core must not add
+   * its unauthenticated HTTP health response to the generated flow route.
+   */
+  directQueueDelivery?: boolean;
+
   /**
    * Supports `experimental_minRetention` for Hooks. Missing or inactive means
    * the runtime rejects retained Hooks before registration.