diff --git a/‎.changeset/sparkly-deer-smoke.md‎
Lines changed: 10 additions & 0 deletions b/‎.changeset/sparkly-deer-smoke.md‎
Lines changed: 10 additions & 0 deletions
diff --git a/‎docs/content/worlds/v5/building-a-world.mdx‎
Lines changed: 40 additions & 1 deletion b/‎docs/content/worlds/v5/building-a-world.mdx‎
Lines changed: 40 additions & 1 deletion
diff --git a/‎docs/content/worlds/v5/postgres.mdx‎
Lines changed: 54 additions & 0 deletions b/‎docs/content/worlds/v5/postgres.mdx‎
Lines changed: 54 additions & 0 deletions
diff --git a/‎packages/core/README.md‎
Lines changed: 16 additions & 0 deletions b/‎packages/core/README.md‎
Lines changed: 16 additions & 0 deletions
diff --git a/‎packages/core/src/runtime.ts‎
Lines changed: 41 additions & 5 deletions b/‎packages/core/src/runtime.ts‎
Lines changed: 41 additions & 5 deletions
@@ -0,0 +1,10 @@
+---
+"@workflow/world": minor
+"@workflow/core": minor
+"@workflow/errors": minor
+"@workflow/world-postgres": minor
+"@workflow/world-local": patch
+"@workflow/world-vercel": patch
+---
+
+Add an optional `invoke` method and capability to the World interface. It routes a payload to the runner handling the specified `runId` and returns a promise for its response. In world-postgres, this uses a regular queue roundtrip with a run-scoped queue. The runtime uses `invoke` when available, initially to resume hooks.
@@ -239,13 +239,52 @@ interface Queue {
     messages: readonly { message: QueuePayload; opts?: QueueOptions }[]
   ): Promise<QueueBatchResult[]>;
 
+  // Optional request/response delivery, enabled with capabilities.invoke.
+  invoke?(runId: string, payload: unknown, options?: InvokeOptions): Promise<unknown>;
+
   createQueueHandler(
     queueNamePrefix: QueuePrefix,
-    handler: (message: unknown, meta: { attempt: number; queueName: ValidQueueName; messageId: MessageId }) => Promise<void | { timeoutSeconds: number }>
+    handler: (message: unknown, meta: { attempt: number; queueName: ValidQueueName; messageId: MessageId }) => Promise<unknown>
   ): (req: Request) => Promise<Response>;
 }
 ```
 
+### Invocation delivery
+
+Implement the optional `invoke` method to send an input to a workflow runner and
+return the runner's response. Enable `capabilities.invoke` only when your World
+supports this operation.
+
+Your World must guarantee at most one active workflow runner per `runId` across
+all worker processes. Different runs may execute concurrently. A replacement
+runner can take over after the previous runner stops.
+
+Route inputs to the active runner, or start or resume the runner when none is
+active. Call the existing `createQueueHandler` callback with
+`{ runId, invoke: true, requestId, input }`. The callback validates the input,
+waits for required event writes, and returns a result. Deliver that result to the
+original caller. Support input processing while the runner awaits step work,
+without starting a second runner for the same run.
+
+`requestId` identifies the logical input across retries and is separate from a
+queue delivery ID. `InvokeOptions` accepts an `idempotencyKey` and a `timeoutMs`
+response-wait limit.
+
+Treat invocation return values as response data, including any `timeoutSeconds`
+property. Ordinary workflow wake results use `timeoutSeconds` to schedule another
+execution.
+
+You can transport results with `InvocationOutcome`, using `{ ok: true, value }`
+for success or `{ ok: false, error: SerializedWorkflowError }` for an error. The
+helpers in `@workflow/errors/invocation` capture handler results and restore
+errors with their known Workflow classes and diagnostic fields. Return the value
+or throw the restored error from `invoke()`.
+
+Keep failures to store or deliver a response distinct from handler errors. A
+transport failure leaves processing unknown, and a handler error may follow
+committed event writes. If your World advertises `hookResumeDedup`, enforce that
+capability when retrying a hook event whose response was lost.
+
 ### Batched publishing
 
 `queueBatch` is optional. Implement it when your transport can accept several messages in one round trip, and the runtime will use it to dispatch a wide `Promise.all` fan-out. A fan-out otherwise costs one round trip per branch, paid before the first branch's step body runs, so it lands directly on time-to-first-step.
 
@@ -80,6 +80,60 @@ The migration is idempotent and can safely be run as a post-deployment lifecycle
 
 ## Starting the World
 
+### Experimental invocation delivery
+
+Set `WORKFLOW_POSTGRES_INVOKE=1`, or pass `enableInvoke: true` to the Postgres
+World's factory, to send hook inputs to the run's executor and wait for its
+response. Invocation is off by default. Apply the migrations and upgrade
+producers and workers together before enabling it.
+
+The executor validates the hook input and writes its event before responding.
+Workflow user code can consume the event later. The World stores inputs and
+responses in `workflow.workflow_invocations`. Inserting an input and enqueueing
+a workflow execution job share one transaction. Retries also enqueue an execution
+job, including when a retained response already exists, so committed events can
+be replayed after a runner failure.
+
+With the default job prefix, Graphile uses `workflow_flows_executor` for workflow
+execution and `workflow_flows` for step jobs and health checks. Execution jobs
+use a named queue per run, `workflow_flows:<runId>:executor`. Graphile permits
+one active job on each named queue across workers. Other runs, steps, and health
+checks can run concurrently. `queueConcurrency` limits total active jobs per
+worker process and defaults to 50. A custom job prefix replaces `workflow_` in
+these names.
+
+The HTTP receiver checks each execution request against its active Graphile job
+before processing pending inputs. Updated workers and receivers move legacy
+orchestration requests to the run's named queue. Before acknowledging an execution
+job, the World finishes any input already being processed and replays newly
+committed events. These entry checks do not stop an old handler from writing
+after its job is reclaimed.
+
+The caller receives a stored return value or a restored Workflow error. Terminal
+errors are retained as outcomes; transient or unrecognized failures leave inputs
+pending for Graphile to retry. Migration 0022 preserves the meaning of responses
+stored by earlier previews. Failure to store or read a response leaves the
+processing outcome unknown. Durable resume identities deduplicate the hook event
+when response storage must be retried.
+
+A shared `LISTEN/NOTIFY` connection signals changes to pending inputs and
+responses. Notifications contain identifiers, and readers fetch data from the
+table. Use a session-capable connection for `LISTEN`. Readers fall back to polling
+at 1s intervals if notifications fail, and connection retries use a 1s backoff.
+Notification failure and recovery are logged once per transition without
+payloads or connection details.
+
+The response-wait timeout defaults to 30s and can be set with `timeoutMs`. Encoded
+inputs and outcomes are limited to 1 MiB each. A timeout does not cancel
+processing. `$retention: 0` purges invocation data and prevents late writes from
+restoring it. Callers whose result has expired receive a 410 error with code
+`INVOCATION_DATA_EXPIRED`, even if the hook event committed earlier.
+
+Automatic cleanup of other results, stopping writes from reclaimed handlers, and
+routing requests across incompatible worker versions remain unimplemented.
+
+### Server startup
+
 To subscribe to the graphile-worker queue, your workflow app needs to start the world on server start. Here are examples for a few frameworks:
 
 <Callout type="warn">
 
@@ -8,3 +8,19 @@ Queued step messages carry immutable run identity in `runContext`, so step
 execution skips the initial `runs.get` and fetches the run row only when
 continuing into workflow replay. Messages from older producers without
 `runContext` retain the initial fetch.
+
+When a World advertises `capabilities.invoke`, `resumeHook()` sends serialized
+hook inputs through `world.invoke()` and waits for the executor's decision. The
+World calls the existing handler with `invoke: true`, `requestId`, and the hook
+input. Core validates the input, waits for its event write, and returns a
+decision. Core shares a run's activity state between workflow execution and these
+input calls.
+
+The event write completes before the response is stored. On Worlds with
+`hookResumeDedup`, a stable request identity makes retries reuse the same hook
+event, including after hook disposal or run completion. Core can process inputs
+while inline steps wait. Workflow code observes committed inputs at replay
+boundaries, reusing a retained Node virtual machine when available.
+
+Errors returned by the executor propagate through `world.invoke()` to the caller
+of `resumeHook()`.
@@ -91,6 +91,7 @@ import {
   stepDispatchIdempotencyKey,
   withHealthCheck,
 } from './runtime/helpers.js';
+import { withRunInputs } from './runtime/invocations.js';
 import {
   handleReplayBudgetExhausted,
   ReplayBudget,
@@ -120,7 +121,7 @@ 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 } from './runtime/world.js';
 import { dehydrateRunError } from './serialization.js';
 import { remapErrorStack } from './source-map.js';
 import * as Attribute from './telemetry/semantic-conventions.js';
@@ -584,6 +585,7 @@ function getRetentionDecision({
   suspension,
   serializationBlockerCount,
   hookContinuation = false,
+  invocationContinuation = false,
 }: {
   suspension: WorkflowSuspension;
   serializationBlockerCount: number;
@@ -594,6 +596,7 @@ function getRetentionDecision({
    * write of its own. See the policy above.
    */
   hookContinuation?: boolean;
+  invocationContinuation?: boolean;
 }): RetentionDecision {
   if (!isVmRetentionEnabled()) {
     return { retain: false, reason: 'disabled' };
@@ -607,7 +610,11 @@ function getRetentionDecision({
   if (hookContinuation) {
     return { retain: true };
   }
-  if (suspension.stepCount === 0 && suspension.attributeCount === 0) {
+  if (
+    !invocationContinuation &&
+    suspension.stepCount === 0 &&
+    suspension.attributeCount === 0
+  ) {
     return { retain: false, reason: 'no_replay_driver' };
   }
   return { retain: true };
@@ -715,10 +722,10 @@ export function workflowEntrypoint(
   const namespace = resolveQueueNamespace(options?.namespace);
   const workflowPrefix = getQueueTopicPrefix('workflow', namespace);
 
-  const handler = (worldHandlers: WorldHandlers) =>
+  const handler = (worldHandlers: World) =>
     worldHandlers.createQueueHandler(
       workflowPrefix,
-      async (message_, metadata) => {
+      withRunInputs(worldHandlers)(async (message_, metadata, activity) => {
         // T2 of the hook-resume TTR window (see runtime/resume-latency.ts):
         // the instant this consumer began, before message parsing. Only used
         // when the message turns out to carry resume timing; taking it
@@ -1965,6 +1972,24 @@ export function workflowEntrypoint(
                         return { timeoutSeconds: stepResult.timeoutSeconds };
                       }
 
+                      // Invoke-capable worlds serialize orchestrator deliveries,
+                      // not step bodies. Return the replay to that executor lane.
+                      if (
+                        world.capabilities?.invoke &&
+                        stepResult.type !== 'gone'
+                      ) {
+                        await queueMessage(
+                          world,
+                          getWorkflowQueueName(workflowName, namespace),
+                          {
+                            runId,
+                            traceCarrier: await nextTraceCarrier(),
+                            requestedAt: new Date(),
+                          }
+                        );
+                        return;
+                      }
+
                       // If step had pending ops (stream writes), break and let
                       // waitUntil flush them, so can't continue inline.
                       if (
@@ -2897,6 +2922,7 @@ export function workflowEntrypoint(
 
                   // Main replay loop
                   while (true) {
+                    const invocationRevision = activity?.revision ?? 0;
                     loopIteration++;
 
                     // Replay-budget check: bail out (retry or fail) if
@@ -3654,6 +3680,7 @@ export function workflowEntrypoint(
                               hookContinuation:
                                 suspensionResult.hasAwaitedHookCreation ||
                                 suspensionResult.hasHookConflict,
+                              invocationContinuation: activity !== undefined,
                             })
                           : undefined;
                         if (retentionDecision?.retain === false) {
@@ -4334,6 +4361,15 @@ export function workflowEntrypoint(
                             // committed hook_created.
                             return await reinvoke(0);
                           }
+                          if (
+                            activity &&
+                            Date.now() - invocationStartTime <
+                              noInlineReplayAfterMs &&
+                            (await activity.waitForActivity(invocationRevision))
+                          ) {
+                            eventLog = nextEventLogLoad(eventLog);
+                            continue;
+                          }
                           return;
                         }
 
@@ -5229,7 +5265,7 @@ export function workflowEntrypoint(
             }
           );
         });
-      }
+      })
     );
 
   let cachedHandler: ((req: Request) => Promise<Response>) | undefined;