A World is the abstraction that allows workflows to run on any infrastructure. It handles workflow storage, step execution queuing, and data streaming. This guide explains the World interface and how to implement your own.

A World connects workflows to the infrastructure that powers them. The World interface abstracts three core responsibilities:

1. Storage — Persisting workflow runs, steps, hooks, and the event log
2. Queue — Enqueuing and processing workflow and step invocations
3. Streamer — Managing real-time data streams between workflows and clients

interface World extends Storage, Queue, Streamer {
  start?(): Promise<void>;
  close?(): Promise<void>;
  getEncryptionKeyForRun?(run: WorkflowRun): Promise<Uint8Array | undefined>;
  getEncryptionKeyForRun?(runId: string, context?: Record<string, unknown>): Promise<Uint8Array | undefined>;
}

The optional start() method initializes background tasks (for example, queue polling). The optional close() method releases resources like connection pools and listeners. The optional getEncryptionKeyForRun() method returns the AES-256 key used to encrypt data for a run; if it is not implemented, encryption is disabled.

Workflow storage is built on an append-only event log. All state changes happen through events — you never modify runs, steps, or hooks directly. Instead, you create events that update the materialized state.

Events fall into three categories: run lifecycle events, step lifecycle events, and hook lifecycle events. See the Event Sourcing documentation for a complete list of event types and their semantics.

The Storage interface provides read access to materialized entities and write access through events:

interface Storage {
  runs: {
    get(id: string, params?: GetWorkflowRunParams): Promise<WorkflowRun>;
    list(params?: ListWorkflowRunsParams): Promise<PaginatedResponse<WorkflowRun>>;
  };

  steps: {
    get(runId: string | undefined, stepId: string, params?: GetStepParams): Promise<Step>;
    list(params: ListWorkflowRunStepsParams): Promise<PaginatedResponse<Step>>;
  };

  events: {
    // Create a new workflow run (runId may be client-provided or null for server generation)
    create(runId: string | null, data: RunCreatedEventRequest, params?: CreateEventParams): Promise<EventResult>;
    
    // Create an event for an existing run
    create(runId: string, data: CreateEventRequest, params?: CreateEventParams): Promise<EventResult>;
    
    list(params: ListEventsParams): Promise<PaginatedResponse<Event>>;
    listByCorrelationId(params: ListEventsByCorrelationIdParams): Promise<PaginatedResponse<Event>>;
  };

  hooks: {
    get(hookId: string, params?: GetHookParams): Promise<Hook>;
    getByToken(token: string, params?: GetHookParams): Promise<Hook>;
    list(params: ListHooksParams): Promise<PaginatedResponse<Hook>>;
  };
}

Key Implementation Details

Event Creation: When events.create() is called, your implementation must:

1. Persist the event to the event log
2. Atomically update the affected entity (run, step, or hook)
3. Return both the created event and the updated entity

Run Creation: For run_created events, the runId parameter may be a client-provided string or null. When null, your World generates and returns a new runId.

Hook Tokens: Hook tokens must be unique. If a hook_created event conflicts with an existing token, return a hook_conflict event instead and include the active hook owner's run ID as eventData.conflictingRunId.

Automatic Hook Disposal: When a workflow reaches a terminal state (completed, failed, or cancelled), automatically dispose of all associated hooks to release tokens for reuse.

The Queue interface handles asynchronous execution of workflows and steps:

interface Queue {
  getDeploymentId(): Promise<string>;

  queue(
    queueName: ValidQueueName,
    message: QueuePayload,
    opts?: QueueOptions
  ): Promise<{ messageId: MessageId }>;

  createQueueHandler(
    queueNamePrefix: QueuePrefix,
    handler: (message: unknown, meta: { attempt: number; queueName: ValidQueueName; messageId: MessageId }) => Promise<void | { timeoutSeconds: number }>
  ): (req: Request) => Promise<Response>;
}

Queue Names

Queue names follow a specific pattern:

• __wkf_workflow_<name> — For workflow invocations
• __wkf_step_<name> — For step invocations

Message Payloads

Two types of messages flow through queues:

Workflow Invocations:

interface WorkflowInvokePayload {
  runId: string;
  traceCarrier?: Record<string, string>;  // OpenTelemetry context
  requestedAt?: Date;
}

Step Invocations:

interface StepInvokePayload {
  workflowName: string;
  workflowRunId: string;
  workflowStartedAt: number;
  stepId: string;
  traceCarrier?: Record<string, string>;
  requestedAt?: Date;
}

Implementation Considerations

• Messages must be delivered at-least-once
• Support configurable retry policies
• Track attempt counts for observability
• Implement idempotency using the idempotencyKey option when provided

The Streamer interface enables real-time data streaming:

interface Streamer {
  streamFlushIntervalMs?: number;

  streams: {
    write(
      runId: string,
      name: string,
      chunk: string | Uint8Array
    ): Promise<void>;

    writeMulti?(
      runId: string,
      name: string,
      chunks: (string | Uint8Array)[]
    ): Promise<void>;

    close(runId: string, name: string): Promise<void>;

    get(
      runId: string,
      name: string,
      startIndex?: number
    ): Promise<ReadableStream<Uint8Array>>;

    list(runId: string): Promise<string[]>;

    /** Paginated snapshot of stream chunks. */
    getChunks(
      runId: string,
      name: string,
      options?: { limit?: number; cursor?: string }
    ): Promise<{
      data: { index: number; data: Uint8Array }[];
      cursor: string | null;
      hasMore: boolean;
      done: boolean;
    }>;

    /** Lightweight metadata: tail index and completion flag. */
    getInfo(
      runId: string,
      name: string
    ): Promise<{ tailIndex: number; done: boolean }>;
  };
}

Streams are identified by a combination of runId and name. Each workflow run can have multiple named streams. writeMulti() is an optional optimization for batching multiple writes.

getChunks returns a paginated snapshot of currently available chunks (unlike get which returns a live ReadableStream that waits for new chunks). getInfo returns the tail index (last chunk index, 0-based, or -1 when empty) and whether the stream is complete — useful for resolving negative startIndex values into absolute positions.

Study these implementations for guidance:

• Local World — Filesystem-based, great for understanding the basics
• Postgres World — Database-backed with graphile-worker for queuing

Workflow SDK includes an E2E test suite that validates World implementations. Once your World is published to npm:

1. Add your world to worlds-manifest.json
2. Open a PR to the Workflow repository
3. CI will automatically run the E2E test suite against your implementation

Your world will then appear on the Worlds Ecosystem page with its compatibility status and performance benchmarks.

1. Package your World — Export a default World instance from your package
2. Publish to npm — Publish your package to npm
3. Add to the manifest — Submit a PR adding your world to worlds-manifest.json
4. Document configuration — Clearly document any required environment variables

// worlds-manifest.json entry
{
  "package": "your-world-package",
  "repository": "https://github.com/you/your-world",
  "docs": "https://github.com/you/your-world#readme"
}